ACL for TUN

README.md

@@ -247,7 +247,7 @@ hysteria_traffic_uplink_bytes_total{auth="aGFja2VyISE="} 37452
   },
   "tun": {
     "name": "tun-hy", // TUN interface name
-    "timeout": 300, // UDP timeout in seconds
+    "timeout": 300, // Timeout in seconds
     "address": "192.0.2.2", // TUN interface address, not applicable for Linux
     "gateway": "192.0.2.1", // TUN interface gateway, not applicable for Linux
     "mask": "255.255.255.252", // TUN interface mask, not applicable for Linux

README.zh.md

@@ -234,7 +234,7 @@ hysteria_traffic_uplink_bytes_total{auth="aGFja2VyISE="} 37452
   },
   "tun": {
     "name": "tun-hy", // TUN 接口名称
-    "timeout": 300, // UDP 超时秒数
+    "timeout": 300, // 超时秒数
     "address": "192.0.2.2", // TUN 接口地址（不适用于 Linux）
     "gateway": "192.0.2.1", // TUN 接口网关（不适用于 Linux）
     "mask": "255.255.255.252", // TUN 接口子网掩码（不适用于 Linux）

cmd/client.go

@@ -20,6 +20,7 @@ import (
 	"io/ioutil"
 	"net"
 	"net/http"
+	"strings"
 	"time"
 )
 
@@ -191,11 +192,40 @@ func client(config *clientConfig) {
 
 	if len(config.TUN.Name) != 0 {
 		go func() {
-			tunServer, err := tun.NewServer(client, time.Duration(config.TUN.Timeout)*time.Second,
+			tunServer, err := tun.NewServer(client, transport.DefaultTransport,
+				time.Duration(config.TUN.Timeout)*time.Second,
 				config.TUN.Name, config.TUN.Address, config.TUN.Gateway, config.TUN.Mask, config.TUN.DNS, config.TUN.Persist)
 			if err != nil {
 				logrus.WithField("error", err).Fatal("Failed to initialize TUN server")
 			}
+			tunServer.RequestFunc = func(addr net.Addr, reqAddr string, action acl.Action, arg string) {
+				logrus.WithFields(logrus.Fields{
+					"action": actionToString(action, arg),
+					"src":    addr.String(),
+					"dst":    reqAddr,
+				}).Debugf("TUN %s request", strings.ToUpper(addr.Network()))
+			}
+			tunServer.ErrorFunc = func(addr net.Addr, reqAddr string, err error) {
+				if err != nil {
+					if err == io.EOF {
+						logrus.WithFields(logrus.Fields{
+							"src": addr.String(),
+							"dst": reqAddr,
+						}).Debugf("TUN %s EOF", strings.ToUpper(addr.Network()))
+					} else if err == core.ErrClosed && strings.HasPrefix(addr.Network(), "udp") {
+						logrus.WithFields(logrus.Fields{
+							"src": addr.String(),
+							"dst": reqAddr,
+						}).Debugf("TUN %s closed for timeout", strings.ToUpper(addr.Network()))
+					} else {
+						logrus.WithFields(logrus.Fields{
+							"error": err,
+							"src":   addr.String(),
+							"dst":   reqAddr,
+						}).Infof("TUN %s error", strings.ToUpper(addr.Network()))
+					}
+				}
+			}
 			errChan <- tunServer.ListenAndServe()
 		}()
 	}

pkg/tun/server.go

@@ -3,8 +3,11 @@ package tun
 import (
 	tun2socks "github.com/eycorsican/go-tun2socks/core"
 	"github.com/eycorsican/go-tun2socks/tun"
+	"github.com/tobyxdd/hysteria/pkg/acl"
 	"github.com/tobyxdd/hysteria/pkg/core"
+	"github.com/tobyxdd/hysteria/pkg/transport"
 	"io"
+	"net"
 	"sync"
 	"time"
 )
@@ -13,8 +16,13 @@ type Server struct {
 	HyClient  *core.Client
 	Timeout   time.Duration
 	TunDev    io.ReadWriteCloser
+	Transport transport.Transport
+	ACLEngine *acl.Engine
 
-	udpConnMap     map[tun2socks.UDPConn]*UDPConnInfo
+	RequestFunc func(addr net.Addr, reqAddr string, action acl.Action, arg string)
+	ErrorFunc   func(addr net.Addr, reqAddr string, err error)
+
+	udpConnMap     map[tun2socks.UDPConn]*udpConnInfo
 	udpConnMapLock sync.RWMutex
 }
 
@@ -22,24 +30,27 @@ const (
 	MTU = 1500
 )
 
-func NewServerWithTunDev(hyClient *core.Client, timeout time.Duration,
+func NewServerWithTunDev(hyClient *core.Client, transport transport.Transport,
+	timeout time.Duration,
 	tunDev io.ReadWriteCloser) (*Server, error) {
 	s := &Server{
 		HyClient:   hyClient,
+		Transport:  transport,
 		Timeout:    timeout,
 		TunDev:     tunDev,
-		udpConnMap: make(map[tun2socks.UDPConn]*UDPConnInfo),
+		udpConnMap: make(map[tun2socks.UDPConn]*udpConnInfo),
 	}
 	return s, nil
 }
 
-func NewServer(hyClient *core.Client, timeout time.Duration,
+func NewServer(hyClient *core.Client, transport transport.Transport,
+	timeout time.Duration,
 	name, address, gateway, mask string, dnsServers []string, persist bool) (*Server, error) {
 	tunDev, err := tun.OpenTunDevice(name, address, gateway, mask, dnsServers, persist)
 	if err != nil {
 		return nil, err
 	}
-	return NewServerWithTunDev(hyClient, timeout, tunDev)
+	return NewServerWithTunDev(hyClient, transport, timeout, tunDev)
 }
 
 func (s *Server) ListenAndServe() error {

pkg/tun/tcp.go

@@ -1,73 +1,74 @@
 package tun
 
 import (
-	"io"
+	"errors"
+	"fmt"
+	"github.com/tobyxdd/hysteria/pkg/acl"
+	"github.com/tobyxdd/hysteria/pkg/utils"
 	"net"
+	"strconv"
 )
 
 func (s *Server) Handle(conn net.Conn, target *net.TCPAddr) error {
-	hyConn, err := s.HyClient.DialTCP(target.String())
+	action, arg := acl.ActionProxy, ""
+	var resErr error
+	if s.ACLEngine != nil {
+		action, arg, _, resErr = s.ACLEngine.ResolveAndMatch(target.IP.String())
+	}
+	if s.RequestFunc != nil {
+		s.RequestFunc(conn.LocalAddr(), target.String(), action, arg)
+	}
+	var closeErr error
+	defer func() {
+		if s.ErrorFunc != nil && closeErr != nil {
+			s.ErrorFunc(conn.LocalAddr(), target.String(), closeErr)
+		}
+	}()
+	switch action {
+	case acl.ActionDirect:
+		if resErr != nil {
+			closeErr = resErr
+			return resErr
+		}
+		rc, err := s.Transport.LocalDialTCP(nil, target)
 		if err != nil {
+			closeErr = err
 			return err
 		}
-	go s.relay(conn, hyConn)
+		go s.relayTCP(conn, rc)
 		return nil
+	case acl.ActionProxy:
+		rc, err := s.HyClient.DialTCP(target.String())
+		if err != nil {
+			closeErr = err
+			return err
+		}
+		go s.relayTCP(conn, rc)
+		return nil
+	case acl.ActionBlock:
+		closeErr = errors.New("blocked in ACL")
+		// caller will abort the connection when err != nil
+		return closeErr
+	case acl.ActionHijack:
+		rc, err := s.Transport.LocalDial("tcp", net.JoinHostPort(arg, strconv.Itoa(target.Port)))
+		if err != nil {
+			closeErr = err
+			return err
+		}
+		go s.relayTCP(conn, rc)
+		return nil
+	default:
+		closeErr = fmt.Errorf("unknown action %d", action)
+		// caller will abort the connection when err != nil
+		return closeErr
+	}
 }
 
-type direction byte
+func (s *Server) relayTCP(clientConn, relayConn net.Conn) {
-
+	closeErr := utils.PipePairWithTimeout(clientConn, relayConn, s.Timeout)
-const (
+	if s.ErrorFunc != nil {
-	directionUplink direction = iota
+		s.ErrorFunc(clientConn.LocalAddr(), relayConn.RemoteAddr().String(), closeErr)
-	directionDownlink
-)
-
-type duplexConn interface {
-	net.Conn
-	CloseRead() error
-	CloseWrite() error
 	}
-
-func (s *Server) relay(clientConn, relayConn net.Conn) {
-	uplinkDone := make(chan struct{})
-
-	halfCloseConn := func(dir direction, interrupt bool) {
-		clientDuplexConn, ok1 := clientConn.(duplexConn)
-		relayDuplexConn, ok2 := relayConn.(duplexConn)
-		if !interrupt && ok1 && ok2 {
-			switch dir {
-			case directionUplink:
-				clientDuplexConn.CloseRead()
-				relayDuplexConn.CloseWrite()
-			case directionDownlink:
-				clientDuplexConn.CloseWrite()
-				relayDuplexConn.CloseRead()
-			}
-		} else {
-			clientConn.Close()
 	relayConn.Close()
-		}
+	clientConn.Close()
-	}
-
-	// Uplink
-	go func() {
-		var err error
-		_, err = io.Copy(relayConn, clientConn)
-		if err != nil {
-			halfCloseConn(directionUplink, true)
-		} else {
-			halfCloseConn(directionUplink, false)
-		}
-		uplinkDone <- struct{}{}
-	}()
-
-	// Downlink
-	var err error
-	_, err = io.Copy(clientConn, relayConn)
-	if err != nil {
-		halfCloseConn(directionDownlink, true)
-	} else {
-		halfCloseConn(directionDownlink, false)
-	}
-
-	<-uplinkDone
 }

pkg/tun/udp.go

@@ -1,21 +1,28 @@
 package tun
 
 import (
+	"bytes"
+	"errors"
 	"fmt"
 	tun2socks "github.com/eycorsican/go-tun2socks/core"
+	"github.com/tobyxdd/hysteria/pkg/acl"
 	"github.com/tobyxdd/hysteria/pkg/core"
-	"log"
+	"io"
 	"net"
+	"strconv"
 	"sync/atomic"
 	"time"
 )
 
-type UDPConnInfo struct {
+const udpBufferSize = 65535
+
+type udpConnInfo struct {
 	hyConn core.UDPConn
+	target string
 	expire atomic.Value
 }
 
-func (s *Server) fetchUDPInput(conn tun2socks.UDPConn, ci *UDPConnInfo) {
+func (s *Server) fetchUDPInput(conn tun2socks.UDPConn, ci *udpConnInfo) {
 	defer func() {
 		s.closeUDPConn(conn)
 	}()
@@ -34,24 +41,85 @@ func (s *Server) fetchUDPInput(conn tun2socks.UDPConn, ci *UDPConnInfo) {
 		}()
 	}
 
+	var err error
+
 	for {
-		bs, from, err := ci.hyConn.ReadFrom()
+		var bs []byte
+		var from string
+		bs, from, err = ci.hyConn.ReadFrom()
 		if err != nil {
 			break
 		}
 		ci.expire.Store(time.Now().Add(s.Timeout))
 		udpAddr, _ := net.ResolveUDPAddr("udp", from)
-		_, _ = conn.WriteFrom(bs, udpAddr)
+		_, err = conn.WriteFrom(bs, udpAddr)
+		if err != nil {
+			break
+		}
+	}
+
+	if s.ErrorFunc != nil {
+		s.ErrorFunc(conn.LocalAddr(), ci.target, err)
 	}
 }
 
 func (s *Server) Connect(conn tun2socks.UDPConn, target *net.UDPAddr) error {
-	c, err := s.HyClient.DialUDP()
+	action, arg := acl.ActionProxy, ""
-	if err != nil {
+	var resErr error
-		return err
+	if s.ACLEngine != nil {
+		action, arg, _, resErr = s.ACLEngine.ResolveAndMatch(target.IP.String())
 	}
-	ci := UDPConnInfo{
+	if s.RequestFunc != nil {
-		hyConn: c,
+		s.RequestFunc(conn.LocalAddr(), target.String(), action, arg)
+	}
+	var hyConn core.UDPConn
+	var closeErr error
+	defer func() {
+		if s.ErrorFunc != nil && closeErr != nil {
+			s.ErrorFunc(conn.LocalAddr(), target.String(), closeErr)
+		}
+	}()
+	switch action {
+	case acl.ActionDirect:
+		if resErr != nil {
+			closeErr = resErr
+			return resErr
+		}
+		var relayConn net.Conn
+		relayConn, closeErr = s.Transport.LocalDial("udp", target.String())
+		if closeErr != nil {
+			return closeErr
+		}
+		hyConn = &delegatedUDPConn{
+			underlayConn:        relayConn,
+			delegatedRemoteAddr: target.String(),
+		}
+	case acl.ActionProxy:
+		hyConn, closeErr = s.HyClient.DialUDP()
+		if closeErr != nil {
+			return closeErr
+		}
+	case acl.ActionBlock:
+		closeErr = errors.New("blocked in ACL")
+		return closeErr
+	case acl.ActionHijack:
+		hijackAddr := net.JoinHostPort(arg, strconv.Itoa(target.Port))
+		var relayConn net.Conn
+		relayConn, closeErr = s.Transport.LocalDial("udp", hijackAddr)
+		if closeErr != nil {
+			return closeErr
+		}
+		hyConn = &delegatedUDPConn{
+			underlayConn:        relayConn,
+			delegatedRemoteAddr: target.String(),
+		}
+	default:
+		closeErr = fmt.Errorf("unknown action %d", action)
+		return nil
+	}
+	ci := udpConnInfo{
+		hyConn: hyConn,
+		target: net.JoinHostPort(target.IP.String(), strconv.Itoa(target.Port)),
 	}
 	ci.expire.Store(time.Now().Add(s.Timeout))
 	s.udpConnMapLock.Lock()
@@ -66,8 +134,9 @@ func (s *Server) ReceiveTo(conn tun2socks.UDPConn, data []byte, addr *net.UDPAdd
 	ci, ok := s.udpConnMap[conn]
 	s.udpConnMapLock.RUnlock()
 	if !ok {
-		log.Printf("not connected: %s <-> %s\n", conn.LocalAddr().String(), addr.String())
+		err := errors.New("previous connection closed for timeout")
-		return fmt.Errorf("not connected: %s <-> %s", conn.LocalAddr().String(), addr.String())
+		s.ErrorFunc(conn.LocalAddr(), addr.String(), err)
+		return err
 	}
 	ci.expire.Store(time.Now().Add(s.Timeout))
 	_ = ci.hyConn.WriteTo(data, addr.String())
@@ -83,3 +152,29 @@ func (s *Server) closeUDPConn(conn tun2socks.UDPConn) {
 		delete(s.udpConnMap, conn)
 	}
 }
+
+type delegatedUDPConn struct {
+	underlayConn        net.Conn
+	delegatedRemoteAddr string
+}
+
+func (c *delegatedUDPConn) ReadFrom() (bs []byte, addr string, err error) {
+	buf := make([]byte, udpBufferSize)
+	n, err := c.underlayConn.Read(buf)
+	if n > 0 {
+		bs = append(bs, buf[0:n]...)
+	}
+	if err != nil || err == io.EOF {
+		addr = c.delegatedRemoteAddr
+	}
+	return
+}
+
+func (c *delegatedUDPConn) WriteTo(bs []byte, addr string) error {
+	_, err := io.Copy(c.underlayConn, bytes.NewReader(bs))
+	return err
+}
+
+func (c *delegatedUDPConn) Close() error {
+	return c.underlayConn.Close()
+}