add socks5 udp relay

cmd/gost/gost.yml

@@ -3,6 +3,10 @@ log:
   level: debug # debug, info, warn, error, fatal
   format: json # text, json
 
+profiling:
+  addr: ":6060"
+  enabled: true
+
 services:
 - name: http+tcp
   url: "http://gost:gost@:8000"
@@ -50,12 +54,29 @@ services:
       readTimeout: 5s
       retry: 3
       notls: true
+      # udpBufferSize: 4096 # range [512, 66560]
   listener:
     type: tcp
     metadata:
       keepAlive: 15s
   # chain: chain-socks5
   # bypass: bypass01
+- name: socks5+tcp
+  url: "socks5://gost:gost@:1080"
+  addr: ":11080"
+  handler:
+    type: socks5
+    metadata:
+      auths:
+       - gost:gost
+      readTimeout: 5s
+      retry: 3
+      notls: true
+      # udpBufferSize: 4096 # range [512, 66560]
+  listener:
+    type: tcp
+    metadata:
+      keepAlive: 15s
 
 chains:
 - name: chain01
@@ -134,13 +155,13 @@ chains:
   - name: hop01
     nodes:
     - name: node01
-      addr: ":1081"
+      addr: ":11080"
       url: "http://gost:gost@:8081"
       # bypass: bypass01
       connector:
         type: socks5
         metadata:
-          # notls: true
+          notls: true
           auth: gost:gost
       dialer:
         type: tcp

cmd/gost/main.go

@@ -2,6 +2,8 @@ package main
 
 import (
 	stdlog "log"
+	"net/http"
+	_ "net/http/pprof"
 
 	"github.com/go-gost/gost/pkg/config"
 	"github.com/go-gost/gost/pkg/logger"
@@ -19,6 +21,16 @@ func main() {
 	}
 	log = logFromConfig(cfg.Log)
 
+	if cfg.Profiling != nil && cfg.Profiling.Enabled {
+		go func() {
+			addr := cfg.Profiling.Addr
+			if addr == "" {
+				addr = ":6060"
+			}
+			log.Info("profiling serve on: ", addr)
+			log.Fatal(http.ListenAndServe(addr, nil))
+		}()
+	}
 	services := buildService(cfg)
 	for _, svc := range services {
 		go svc.Run()

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/coreos/go-iptables v0.5.0 // indirect
 	github.com/ginuerzh/tls-dissector v0.0.2-0.20201202075250-98fa925912da
 	github.com/go-gost/gosocks4 v0.0.1
-	github.com/go-gost/gosocks5 v0.3.1-0.20211107153135-23b5baedc2aa
+	github.com/go-gost/gosocks5 v0.3.1-0.20211108032632-bbfd2de9a32d
 	github.com/gobwas/glob v0.2.3
 	github.com/golang/snappy v0.0.3
 	github.com/google/gopacket v1.1.19 // indirect

go.sum

@@ -121,6 +121,8 @@ github.com/go-gost/gosocks5 v0.3.1-0.20211107150557-ff084b955b6a h1:LQ189f7tRprI
 github.com/go-gost/gosocks5 v0.3.1-0.20211107150557-ff084b955b6a/go.mod h1:1G6I7HP7VFVxveGkoK8mnprnJqSqJjdcASKsdUn4Pp4=
 github.com/go-gost/gosocks5 v0.3.1-0.20211107153135-23b5baedc2aa h1:4yBKO6CPj5LokDeVJy3jbvQTcclG6lMk7zQMQ1/MAYo=
 github.com/go-gost/gosocks5 v0.3.1-0.20211107153135-23b5baedc2aa/go.mod h1:1G6I7HP7VFVxveGkoK8mnprnJqSqJjdcASKsdUn4Pp4=
+github.com/go-gost/gosocks5 v0.3.1-0.20211108032632-bbfd2de9a32d h1:mjoFToMUWNN06IwOyXOk9bEsev3T5RUoC9n4Xt7ZDkg=
+github.com/go-gost/gosocks5 v0.3.1-0.20211108032632-bbfd2de9a32d/go.mod h1:1G6I7HP7VFVxveGkoK8mnprnJqSqJjdcASKsdUn4Pp4=
 github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0 h1:p104kn46Q8WdvHunIJ9dAyjPVtrBPhSr3KT2yUst43I=
 github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0/go.mod h1:fyg7847qk6SyHyPtNmDHnmrv/HOrqktSC+C9fM+CJOE=
 github.com/gobwas/glob v0.2.3 h1:A4xDbljILXROh+kObIiy5kIaPYD8e96x1tgBhUI5J+Y=

pkg/config/config.go

@@ -24,6 +24,11 @@ type LogConfig struct {
 	Format string
 }
 
+type ProfilingConfig struct {
+	Addr    string
+	Enabled bool
+}
+
 type LoadbalancingConfig struct {
 	Strategy    string
 	MaxFails    int
@@ -88,6 +93,7 @@ type NodeConfig struct {
 
 type Config struct {
 	Log       *LogConfig
+	Profiling *ProfilingConfig
 	Services  []ServiceConfig
 	Chains    []ChainConfig
 	Bypasses  []BypassConfig

pkg/handler/socks/v5/bind.go

@@ -81,6 +81,7 @@ func (h *socks5Handler) bindLocal(ctx context.Context, conn net.Conn, addr strin
 
 	// Issue: may not reachable when host has multi-interface
 	socksAddr.Host, _, _ = net.SplitHostPort(conn.LocalAddr().String())
+	socksAddr.Type = 0
 	reply := gosocks5.NewReply(gosocks5.Succeeded, socksAddr)
 	if err := reply.Write(conn); err != nil {
 		h.logger.Error(err)

pkg/handler/socks/v5/handler.go

@@ -99,6 +99,7 @@ func (h *socks5Handler) Handle(ctx context.Context, conn net.Conn) {
 	case gosocks5.CmdUdp:
 		h.handleUDP(ctx, conn, req)
 	case socks.CmdUDPTun:
+		h.handleUDPTun(ctx, conn, req)
 	default:
 		h.logger.Errorf("unknown cmd: %d", req.Cmd)
 		resp := gosocks5.NewReply(gosocks5.CmdUnsupported, nil)

pkg/handler/socks/v5/mbind.go

@@ -87,6 +87,7 @@ func (h *socks5Handler) muxBindLocal(ctx context.Context, conn net.Conn, addr st
 
 	// Issue: may not reachable when host has multi-interface
 	socksAddr.Host, _, _ = net.SplitHostPort(conn.LocalAddr().String())
+	socksAddr.Type = 0
 	reply := gosocks5.NewReply(gosocks5.Succeeded, socksAddr)
 	if err := reply.Write(conn); err != nil {
 		h.logger.Error(err)

pkg/handler/socks/v5/metadata.go

@@ -16,16 +16,20 @@ const (
 	caFile        = "caFile"
 	authsKey      = "auths"
 	readTimeout   = "readTimeout"
+	timeout       = "timeout"
 	retryCount    = "retry"
 	noTLS         = "notls"
+	udpBufferSize = "udpBufferSize"
 )
 
 type metadata struct {
 	tlsConfig     *tls.Config
 	authenticator auth.Authenticator
+	timeout       time.Duration
 	readTimeout   time.Duration
 	retryCount    int
 	noTLS         bool
+	udpBufferSize int
 }
 
 func (h *socks5Handler) parseMetadata(md md.Metadata) error {
@@ -55,8 +59,21 @@ func (h *socks5Handler) parseMetadata(md md.Metadata) error {
 	}
 
 	h.md.readTimeout = md.GetDuration(readTimeout)
+	h.md.timeout = md.GetDuration(timeout)
 	h.md.retryCount = md.GetInt(retryCount)
 	h.md.noTLS = md.GetBool(noTLS)
 
+	h.md.udpBufferSize = md.GetInt(udpBufferSize)
+	if h.md.udpBufferSize > 0 {
+		if h.md.udpBufferSize < 512 {
+			h.md.udpBufferSize = 512 // min buffer size
+		}
+		if h.md.udpBufferSize > 65*1024 {
+			h.md.udpBufferSize = 65 * 1024 // max buffer size
+		}
+	} else {
+		h.md.udpBufferSize = 4096 // default buffer size
+	}
+
 	return nil
 }

pkg/handler/socks/v5/udp.go

@@ -3,13 +3,16 @@ package v5
 import (
 	"bytes"
 	"context"
+	"errors"
 	"io"
 	"io/ioutil"
 	"net"
 	"time"
 
 	"github.com/go-gost/gosocks5"
+	"github.com/go-gost/gost/pkg/handler"
 	"github.com/go-gost/gost/pkg/internal/bufpool"
+	"github.com/go-gost/gost/pkg/internal/utils/socks"
 	"github.com/go-gost/gost/pkg/logger"
 )
 
@@ -50,10 +53,8 @@ func (h *socks5Handler) handleUDP(ctx context.Context, conn net.Conn, req *gosoc
 	})
 	h.logger.Infof("bind on %s OK", saddr.String())
 
-	if !h.chain.IsEmpty() {
+	if h.chain.IsEmpty() {
-
+		// serve as standard socks5 udp relay.
-	}
-
 		peer, err := net.ListenUDP("udp", nil)
 		if err != nil {
 			h.logger.Error(err)
@@ -61,7 +62,17 @@ func (h *socks5Handler) handleUDP(ctx context.Context, conn net.Conn, req *gosoc
 		}
 		defer peer.Close()
 
-	go h.transportUDP(relay, peer)
+		go h.relayUDP(relay, peer)
+	} else {
+		tun, err := h.getUDPTun(ctx)
+		if err != nil {
+			h.logger.Error(err)
+			return
+		}
+		defer tun.Close()
+
+		go h.tunnelClientUDP(relay, tun)
+	}
 
 	t := time.Now()
 	h.logger.Infof("%s <-> %s", conn.RemoteAddr(), saddr)
@@ -71,8 +82,54 @@ func (h *socks5Handler) handleUDP(ctx context.Context, conn net.Conn, req *gosoc
 		Infof("%s >-< %s", conn.RemoteAddr(), saddr)
 }
 
-func (h *socks5Handler) transportUDP(relay, peer net.PacketConn) (err error) {
+func (h *socks5Handler) getUDPTun(ctx context.Context) (conn net.Conn, err error) {
-	const bufSize = 65 * 1024
+	r := (&handler.Router{}).
+		WithChain(h.chain).
+		WithRetry(h.md.retryCount).
+		WithLogger(h.logger)
+	conn, err = r.Connect(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	defer func() {
+		if err != nil {
+			conn.Close()
+			conn = nil
+		}
+	}()
+
+	if h.md.timeout > 0 {
+		conn.SetDeadline(time.Now().Add(h.md.timeout))
+		defer conn.SetDeadline(time.Time{})
+	}
+
+	req := gosocks5.NewRequest(socks.CmdUDPTun, nil)
+	if err = req.Write(conn); err != nil {
+		return
+	}
+	if h.logger.IsLevelEnabled(logger.DebugLevel) {
+		h.logger.Debug(req)
+	}
+
+	reply, err := gosocks5.ReadReply(conn)
+	if err != nil {
+		return
+	}
+	if h.logger.IsLevelEnabled(logger.DebugLevel) {
+		h.logger.Debug(reply)
+	}
+
+	if reply.Rep != gosocks5.Succeeded {
+		err = errors.New("UDP associate failed")
+		return
+	}
+
+	return
+}
+
+func (h *socks5Handler) tunnelClientUDP(c net.PacketConn, tunnel net.Conn) (err error) {
+	bufSize := h.md.udpBufferSize
 	errc := make(chan error, 2)
 
 	var clientAddr net.Addr
@@ -82,7 +139,126 @@ func (h *socks5Handler) transportUDP(relay, peer net.PacketConn) (err error) {
 		defer bufpool.Put(b)
 
 		for {
-			n, laddr, err := relay.ReadFrom(b)
+			n, laddr, err := c.ReadFrom(b)
+			if err != nil {
+				errc <- err
+				return
+			}
+
+			if clientAddr == nil {
+				clientAddr = laddr
+			}
+
+			var addr gosocks5.Addr
+			header := gosocks5.UDPHeader{
+				Addr: &addr,
+			}
+			hlen, err := header.ReadFrom(bytes.NewReader(b[:n]))
+			if err != nil {
+				errc <- err
+				return
+			}
+
+			raddr, err := net.ResolveUDPAddr("udp", addr.String())
+			if err != nil {
+				continue // drop silently
+			}
+
+			if h.bypass != nil && h.bypass.Contains(raddr.String()) {
+				h.logger.Warn("bypass: ", raddr)
+				continue // bypass
+			}
+
+			dgram := gosocks5.UDPDatagram{
+				Header: &header,
+				Data:   b[hlen:n],
+			}
+			dgram.Header.Rsv = uint16(len(dgram.Data))
+
+			if _, err := dgram.WriteTo(tunnel); err != nil {
+				errc <- err
+				return
+			}
+
+			if h.logger.IsLevelEnabled(logger.DebugLevel) {
+				h.logger.Debugf("%s >>> %s: %v data: %d",
+					clientAddr, raddr, b[:hlen], len(dgram.Data))
+			}
+		}
+	}()
+
+	go func() {
+		b := bufpool.Get(bufSize)
+		defer bufpool.Put(b)
+
+		const dataPos = 262
+
+		for {
+			addr := gosocks5.Addr{}
+			header := gosocks5.UDPHeader{
+				Addr: &addr,
+			}
+
+			data := b[dataPos:]
+			dgram := gosocks5.UDPDatagram{
+				Header: &header,
+				Data:   data,
+			}
+			_, err := dgram.ReadFrom(tunnel)
+			if err != nil {
+				errc <- err
+				return
+			}
+			// NOTE: the dgram.Data may be reallocated if the provided buffer is too short,
+			// we drop it for simplicity. As this occurs, you should enlarge the buffer size.
+			if len(dgram.Data) > len(data) {
+				h.logger.Warnf("buffer too short, dropped")
+				continue
+			}
+
+			// pipe from tunnel to relay
+			if clientAddr == nil {
+				h.logger.Warnf("ignore unexpected peer from %s", addr)
+				continue
+			}
+
+			raddr := addr.String()
+			if h.bypass != nil && h.bypass.Contains(raddr) {
+				h.logger.Warn("bypass: ", raddr)
+				continue // bypass
+			}
+
+			addrLen := addr.Length()
+			addr.Encode(b[dataPos-addrLen : dataPos])
+
+			hlen := addrLen + 3
+			if _, err := c.WriteTo(b[dataPos-hlen:dataPos+len(dgram.Data)], clientAddr); err != nil {
+				errc <- err
+				return
+			}
+
+			if h.logger.IsLevelEnabled(logger.DebugLevel) {
+				h.logger.Debugf("%s <<< %s: %v data: %d",
+					clientAddr, addr.String(), b[dataPos-hlen:dataPos], len(dgram.Data))
+			}
+		}
+	}()
+
+	return <-errc
+}
+
+func (h *socks5Handler) relayUDP(c, peer net.PacketConn) (err error) {
+	bufSize := h.md.udpBufferSize
+	errc := make(chan error, 2)
+
+	var clientAddr net.Addr
+
+	go func() {
+		b := bufpool.Get(bufSize)
+		defer bufpool.Put(b)
+
+		for {
+			n, laddr, err := c.ReadFrom(b)
 			if err != nil {
 				errc <- err
 				return
@@ -127,7 +303,7 @@ func (h *socks5Handler) transportUDP(relay, peer net.PacketConn) (err error) {
 		b := bufpool.Get(bufSize)
 		defer bufpool.Put(b)
 
-		const dataPos = 1024
+		const dataPos = 262
 
 		for {
 			n, raddr, err := peer.ReadFrom(b[dataPos:])
@@ -152,7 +328,7 @@ func (h *socks5Handler) transportUDP(relay, peer net.PacketConn) (err error) {
 			socksAddr.Encode(b[dataPos-addrLen : dataPos])
 
 			hlen := addrLen + 3
-			if _, err := relay.WriteTo(b[dataPos-hlen:dataPos+n], clientAddr); err != nil {
+			if _, err := c.WriteTo(b[dataPos-hlen:dataPos+n], clientAddr); err != nil {
 				errc <- err
 				return
 			}

pkg/handler/socks/v5/udp_tun.go

@@ -0,0 +1,190 @@
+package v5
+
+import (
+	"context"
+	"net"
+	"time"
+
+	"github.com/go-gost/gosocks5"
+	"github.com/go-gost/gost/pkg/handler"
+	"github.com/go-gost/gost/pkg/internal/bufpool"
+	"github.com/go-gost/gost/pkg/logger"
+)
+
+func (h *socks5Handler) handleUDPTun(ctx context.Context, conn net.Conn, req *gosocks5.Request) {
+	h.logger = h.logger.WithFields(map[string]interface{}{
+		"cmd": "udp-tun",
+	})
+
+	if h.chain.IsEmpty() {
+		addr := req.Addr.String()
+
+		bindAddr, _ := net.ResolveUDPAddr("udp", addr)
+		relay, err := net.ListenUDP("udp", bindAddr)
+		if err != nil {
+			h.logger.Error(err)
+			return
+		}
+		defer relay.Close()
+
+		saddr, _ := gosocks5.NewAddr(relay.LocalAddr().String())
+		saddr.Host, _, _ = net.SplitHostPort(conn.LocalAddr().String())
+		saddr.Type = 0
+		reply := gosocks5.NewReply(gosocks5.Succeeded, saddr)
+		if err := reply.Write(conn); err != nil {
+			h.logger.Error(err)
+			return
+		}
+		if h.logger.IsLevelEnabled(logger.DebugLevel) {
+			h.logger.Debug(reply)
+		}
+
+		h.logger = h.logger.WithFields(map[string]interface{}{
+			"bind": saddr.String(),
+		})
+
+		t := time.Now()
+		h.logger.Infof("%s <-> %s", conn.RemoteAddr(), saddr)
+		h.tunnelServerUDP(conn, relay)
+		h.logger.
+			WithFields(map[string]interface{}{
+				"duration": time.Since(t),
+			}).
+			Infof("%s >-< %s", conn.RemoteAddr(), saddr)
+
+		return
+	}
+
+	r := (&handler.Router{}).
+		WithChain(h.chain).
+		WithRetry(h.md.retryCount).
+		WithLogger(h.logger)
+	cc, err := r.Connect(ctx)
+	if err != nil {
+		h.logger.Error(err)
+		reply := gosocks5.NewReply(gosocks5.Failure, nil)
+		reply.Write(conn)
+		if h.logger.IsLevelEnabled(logger.DebugLevel) {
+			h.logger.Debug(reply)
+		}
+		return
+	}
+	defer cc.Close()
+
+	// forward request
+	if err := req.Write(cc); err != nil {
+		h.logger.Error(err)
+		reply := gosocks5.NewReply(gosocks5.Failure, nil)
+		reply.Write(conn)
+		if h.logger.IsLevelEnabled(logger.DebugLevel) {
+			h.logger.Debug(reply)
+		}
+	}
+
+	t := time.Now()
+	h.logger.Infof("%s <-> %s", conn.RemoteAddr(), cc.RemoteAddr())
+	handler.Transport(conn, cc)
+	h.logger.
+		WithFields(map[string]interface{}{
+			"duration": time.Since(t),
+		}).
+		Infof("%s >-< %s", conn.RemoteAddr(), cc.RemoteAddr())
+}
+
+func (h *socks5Handler) tunnelServerUDP(tunnel net.Conn, c net.PacketConn) (err error) {
+	bufSize := h.md.udpBufferSize
+	errc := make(chan error, 2)
+
+	go func() {
+		b := bufpool.Get(bufSize)
+		defer bufpool.Put(b)
+
+		const dataPos = 262
+
+		for {
+			addr := gosocks5.Addr{}
+			header := gosocks5.UDPHeader{
+				Addr: &addr,
+			}
+
+			data := b[dataPos:]
+			dgram := gosocks5.UDPDatagram{
+				Header: &header,
+				Data:   data,
+			}
+			_, err := dgram.ReadFrom(tunnel)
+			if err != nil {
+				errc <- err
+				return
+			}
+			// NOTE: the dgram.Data may be reallocated if the provided buffer is too short,
+			// we drop it for simplicity. As this occurs, you should enlarge the buffer size.
+			if len(dgram.Data) > len(data) {
+				h.logger.Warnf("buffer too short, dropped")
+				continue
+			}
+
+			raddr, err := net.ResolveUDPAddr("udp", addr.String())
+			if err != nil {
+				continue // drop silently
+			}
+			if h.bypass != nil && h.bypass.Contains(raddr.String()) {
+				h.logger.Warn("bypass: ", raddr.String())
+				continue // bypass
+			}
+
+			if _, err := c.WriteTo(dgram.Data, raddr); err != nil {
+				errc <- err
+				return
+			}
+
+			if h.logger.IsLevelEnabled(logger.DebugLevel) {
+				h.logger.Debugf("%s >>> %s: %v data: %d",
+					tunnel.RemoteAddr(), raddr, header.String(), len(dgram.Data))
+			}
+		}
+	}()
+
+	go func() {
+		b := bufpool.Get(bufSize)
+		defer bufpool.Put(b)
+
+		for {
+			n, raddr, err := c.ReadFrom(b)
+			if err != nil {
+				errc <- err
+				return
+			}
+
+			if h.bypass != nil && h.bypass.Contains(raddr.String()) {
+				h.logger.Warn("bypass: ", raddr.String())
+				continue // bypass
+			}
+
+			addr, _ := gosocks5.NewAddr(raddr.String())
+			if addr == nil {
+				addr = &gosocks5.Addr{}
+			}
+			addr.Type = 0
+			header := gosocks5.UDPHeader{
+				Rsv:  uint16(n),
+				Addr: addr,
+			}
+			dgram := gosocks5.UDPDatagram{
+				Header: &header,
+				Data:   b[:n],
+			}
+
+			if _, err := dgram.WriteTo(tunnel); err != nil {
+				errc <- err
+				return
+			}
+			if h.logger.IsLevelEnabled(logger.DebugLevel) {
+				h.logger.Debugf("%s <<< %s: %v data: %d",
+					tunnel.RemoteAddr(), raddr, header.String(), len(dgram.Data))
+			}
+		}
+	}()
+
+	return <-errc
+}

pkg/handler/transport.go

@@ -2,19 +2,8 @@ package handler
 
 import (
 	"io"
-	"sync"
-)
 
-const (
+	"github.com/go-gost/gost/pkg/internal/bufpool"
-	poolBufferSize = 32 * 1024
-)
-
-var (
-	pool = sync.Pool{
-		New: func() interface{} {
-			return make([]byte, poolBufferSize)
-		},
-	}
 )
 
 func Transport(rw1, rw2 io.ReadWriter) error {
@@ -35,8 +24,8 @@ func Transport(rw1, rw2 io.ReadWriter) error {
 }
 
 func copyBuffer(dst io.Writer, src io.Reader) error {
-	buf := pool.Get().([]byte)
+	buf := bufpool.Get(16 * 1024)
-	defer pool.Put(buf)
+	defer bufpool.Put(buf)
 
 	_, err := io.CopyBuffer(dst, src, buf)
 	return err