重构代理解析流程，引入端口混合协议转发

README.md

@@ -63,4 +63,8 @@ ERR: 除非有必要，否则全部 error 都使用 `errors.Wrap()` 包裹（如
 1. 构建项目
 2. 使用测试配置 `.env.test` 远程启动 docker
 
-## 转发服务
+### 转发服务结束时资源清理
+
+1. 关闭接听端口，防止新连接接入（user, data, ctrl）
+2. 通知并等待所有正在运行的 conn 处理协程全部关闭（user, data, ctrl）
+3. 结束所有保存且未使用的 conn 连接（user, ctrl）

go.mod

@@ -4,11 +4,11 @@ go 1.24
 
 require (
 	github.com/gin-gonic/gin v1.10.0
-	github.com/google/gopacket v1.1.19
 	github.com/joho/godotenv v1.5.1
 	github.com/lmittmann/tint v1.0.7
 	github.com/mattn/go-colorable v0.1.14
 	github.com/pkg/errors v0.9.1
+	github.com/soheilhy/cmux v0.1.5
 	gorm.io/driver/postgres v1.5.11
 	gorm.io/gen v0.3.26
 	gorm.io/gorm v1.25.12

go.sum

@@ -36,8 +36,6 @@ github.com/golang-sql/sqlexp v0.1.0/go.mod h1:J4ad9Vo8ZCWQ2GMrC4UCQy1JpCbwU9m3EO
 github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
-github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF8=
-github.com/google/gopacket v1.1.19/go.mod h1:iJ8V8n6KS+z2U1A8pUwu8bW5SyEMkXJB8Yo/Vo+TKTo=
 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
 github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
@@ -90,6 +88,8 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII=
 github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o=
+github.com/soheilhy/cmux v0.1.5 h1:jjzc5WVemNEDTLwv9tlmemhC73tI08BNOIGwBOo10Js=
+github.com/soheilhy/cmux v0.1.5/go.mod h1:T7TcVDs9LWfQgPlPsdngu6I6QIoyIFZDDC6sNE1GqG0=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
@@ -109,32 +109,30 @@ github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZ
 golang.org/x/arch v0.14.0 h1:z9JUEZWr8x4rR0OU6c4/4t6E6jOZ8/QBS2bBYBm4tx4=
 golang.org/x/arch v0.14.0/go.mod h1:FEVrYAQjsQXMVJ1nsMoVVXPZg6p2JE2mx8psSWTDQys=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
-golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.33.0 h1:IOBPskki6Lysi0lo9qQvbxiQ+FvsCC/YWOecCHAixus=
 golang.org/x/crypto v0.33.0/go.mod h1:bVdXmD7IV/4GdElGPozy6U7lWdRXA4qyRVGJV57uQ5M=
-golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
-golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
 golang.org/x/mod v0.22.0 h1:D4nJWe9zXqHOmWqj4VMOJhvzj7bEZg4wEYa759z1pH4=
 golang.org/x/mod v0.22.0/go.mod h1:6SkKJ3Xj0I0BrPOZoBy3bdMptDDU9oJrpohJ3eWZ1fY=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
-golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.35.0 h1:T5GQRQb2y08kTAByq9L4/bz8cipCdA8FbRTXewonqY8=
 golang.org/x/net v0.35.0/go.mod h1:EglIi67kWsHKlRzzVMUD93VMSWGFOMSZgxFjparz1Qk=
-golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w=
 golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
 golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM=
 golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY=
-golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.28.0 h1:WuB6qZ4RPCQo5aP3WdKZS7i595EdWqWR8vqJTlwTVK8=
 golang.org/x/tools v0.28.0/go.mod h1:dcIOrVd3mfQKTgrDVQHqCPMWy6lnhfhtX3hLXYVLfRw=
-golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/protobuf v1.36.5 h1:tPhr+woSbjfYvY6/GPufUoYizxw1cF/yFoxJ2fmpwlM=
 google.golang.org/protobuf v1.36.5/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

pkg/utils/chan.go

@@ -9,21 +9,22 @@ import (
 )
 
 func ChanConnAccept(ctx context.Context, ls net.Listener) chan net.Conn {
-	connCh := make(chan net.Conn)
+	ch := make(chan net.Conn)
 	go func() {
+		defer close(ch)
 		for {
 			conn, err := ls.Accept()
 			if err != nil {
 				if errors.Is(err, net.ErrClosed) {
 					return
 				}
-				slog.Error("接受连接失败", err)
 				// 临时错误重试连接
 				var ne net.Error
 				if errors.As(err, &ne) && ne.Temporary() {
 					slog.Debug("临时错误重试")
 					continue
 				}
+				slog.Error("接受连接失败", err)
 				return
 			}
 			// ctx 取消后退出
@@ -31,16 +32,17 @@ func ChanConnAccept(ctx context.Context, ls net.Listener) chan net.Conn {
 			case <-ctx.Done():
 				Close(conn)
 				return
-			case connCh <- conn:
+			case ch <- conn:
 			}
 		}
 	}()
-	return connCh
+	return ch
 }
 
 func ChanWgWait[T WaitGroup](ctx context.Context, wg T) chan struct{} {
 	ch := make(chan struct{})
 	go func() {
+		defer close(ch)
 		wg.Wait()
 		select {
 		case <-ctx.Done():

server/fwd/analysis.go

@@ -6,13 +6,13 @@ import (
 	"io"
 	"log/slog"
 	"proxy-server/pkg/utils"
-	"proxy-server/server/fwd/socks"
+	"proxy-server/server/fwd/core"
 	"strings"
 
 	"github.com/pkg/errors"
 )
 
-func analysisAndLog(conn socks.ProxyConn, reader io.Reader) error {
+func analysisAndLog(conn *core.Conn, reader io.Reader) error {
 	buf := bufio.NewReader(reader)
 
 	domain, proto, err := sniffing(buf)
@@ -21,12 +21,12 @@ func analysisAndLog(conn socks.ProxyConn, reader io.Reader) error {
 	} else {
 		slog.Info(
 			"用户访问记录",
-			slog.Uint64("uid", uint64(conn.Uid)),
+			slog.Uint64("uid", uint64(conn.Auth.Payload.ID)),
-			slog.String("user", conn.Conn.RemoteAddr().String()),
+			slog.String("user", conn.RemoteAddr().String()),
-			slog.String("proxy", "socks"),
+			slog.String("proxy", conn.Protocol),
-			slog.String("node", conn.Conn.LocalAddr().String()),
+			slog.String("node", conn.LocalAddr().String()),
 			slog.String("proto", proto),
-			slog.String("dest", conn.Dest),
+			slog.String("dest", conn.DestAddr().String()),
 			slog.String("domain", domain),
 		)
 	}

server/fwd/auth.go

@@ -1,206 +0,0 @@
-package fwd
-
-import (
-	"context"
-	"io"
-	"log/slog"
-	"net"
-	"proxy-server/pkg/utils"
-	"proxy-server/server/fwd/socks"
-	"proxy-server/server/pkg/orm"
-	"proxy-server/server/web/models"
-	"time"
-
-	"github.com/pkg/errors"
-)
-
-type NoAuthAuthenticator struct {
-}
-
-func (a *NoAuthAuthenticator) Method() socks.AuthMethod {
-	return socks.NoAuth
-}
-
-func (a *NoAuthAuthenticator) Authenticate(ctx context.Context, reader io.Reader, writer io.Writer) (*socks.Authentication, error) {
-
-	// 获取用户地址
-	conn, ok := writer.(net.Conn)
-	if !ok {
-		return nil, errors.New("noAuth 认证失败，无法获取连接信息")
-	}
-	addr := conn.RemoteAddr().String()
-	client, _, err := net.SplitHostPort(addr)
-	if err != nil {
-		return nil, errors.Wrap(err, "noAuth 认证失败")
-	}
-	slog.Debug("用户的地址为 " + client)
-
-	// 获取服务
-	server, ok := ctx.Value("service").(*socks.Server)
-	if !ok {
-		return nil, errors.New("noAuth 认证失败，无法获取服务信息")
-	}
-	node := server.Name
-	slog.Debug("服务的名称为 " + server.Name)
-
-	// 查询权限记录
-	slog.Info("用户 " + client + " 请求连接到 " + node)
-	var channels []models.Channel
-	err = orm.DB.
-		Joins("INNER JOIN public.nodes n ON channels.node_id = n.id AND n.name = ?", node).
-		Joins("INNER JOIN public.users u ON channels.user_id = u.id").
-		Joins("INNER JOIN public.user_ips ip ON u.id = ip.user_id AND ip.ip_address = ?", client).
-		Where(&models.Channel{
-			AuthIp: true,
-		}).
-		Find(&channels).Error
-	if err != nil {
-		return nil, errors.New("noAuth 查询用户权限失败")
-	}
-
-	// 记录应该只有一条
-	channel, err := orm.MaySingle(channels)
-	if err != nil {
-		return nil, errors.Wrap(err, "noAuth 没有权限")
-	}
-
-	// 检查是否需要密码认证
-	if channel.AuthPass {
-		return nil, errors.New("noAuth 没有权限，需要密码认证")
-	}
-
-	// 检查权限是否过期
-	timeout := channel.Expiration.Sub(time.Now()).Seconds()
-	slog.Info("用户剩余时间", "timeout", timeout)
-	if timeout <= 0 {
-		return nil, errors.New("noAuth 权限已过期")
-	}
-	slog.Debug("权限剩余时间", slog.Uint64("timeout", uint64(timeout)))
-
-	return &socks.Authentication{
-		Method:  socks.NoAuth,
-		Timeout: uint(timeout),
-		Payload: socks.Payload{
-			ID: channel.UserId,
-		},
-	}, nil
-}
-
-type UserPassAuthenticator struct {
-}
-
-func (a *UserPassAuthenticator) Method() socks.AuthMethod {
-	return socks.UserPassAuth
-}
-
-func (a *UserPassAuthenticator) Authenticate(ctx context.Context, reader io.Reader, writer io.Writer) (*socks.Authentication, error) {
-
-	// 检查认证版本
-	slog.Debug("验证认证版本")
-	v, err := utils.ReadByte(reader)
-	if err != nil {
-		return nil, errors.Wrap(err, "读取版本号失败")
-	}
-	if v != socks.AuthVersion {
-		_, err := writer.Write([]byte{socks.Version, socks.AuthFailure})
-		if err != nil {
-			return nil, errors.Wrap(err, "响应认证失败")
-		}
-		return nil, errors.New("认证版本参数不正确")
-	}
-
-	// 读取账号
-	slog.Debug("验证用户账号")
-	uLen, err := utils.ReadByte(reader)
-	if err != nil {
-		return nil, errors.Wrap(err, "读取用户名长度失败")
-	}
-	usernameBuf, err := utils.ReadBuffer(reader, int(uLen))
-	if err != nil {
-		return nil, errors.Wrap(err, "读取用户名失败")
-	}
-	username := string(usernameBuf)
-
-	// 读取密码
-	pLen, err := utils.ReadByte(reader)
-	if err != nil {
-		return nil, errors.Wrap(err, "读取密码长度失败")
-	}
-	passwordBuf, err := utils.ReadBuffer(reader, int(pLen))
-	if err != nil {
-		return nil, errors.Wrap(err, "读取密码失败")
-	}
-	password := string(passwordBuf)
-
-	// 查询通道配置
-	var channel models.Channel
-	err = orm.DB.
-		Where(&models.Channel{
-			Username: username,
-			AuthPass: true,
-		}).
-		First(&channel).Error
-	if err != nil {
-		return nil, errors.Wrap(err, "查询用户失败")
-	}
-
-	// 检查密码 todo 哈希
-	if channel.Password != password {
-		return nil, errors.New("密码错误")
-	}
-
-	// 检查权限是否过期
-	timeout := channel.Expiration.Sub(time.Now()).Seconds()
-	slog.Info("用户剩余时间", "timeout", timeout)
-	if timeout <= 0 {
-		return nil, errors.New("权限已过期")
-	}
-
-	// 如果用户设置了双验证则检查 ip 是否在白名单中
-	if channel.AuthIp {
-		slog.Debug("验证用户 ip")
-
-		// 获取用户地址
-		conn, ok := writer.(net.Conn)
-		if !ok {
-			return nil, errors.New("无法获取连接信息")
-		}
-		addr := conn.RemoteAddr().String()
-		client, _, err := net.SplitHostPort(addr)
-		if err != nil {
-			return nil, errors.Wrap(err, "无法获取连接信息")
-		}
-
-		// 查询通道配置
-		var ips []models.UserIp
-		err = orm.DB.
-			Where(&models.UserIp{
-				UserId:    channel.UserId,
-				IpAddress: client,
-			}).
-			Find(&ips).Error
-		if err != nil {
-			return nil, errors.Wrap(err, "查询用户 ip 失败")
-		}
-
-		// 检查是否在白名单中
-		if len(ips) == 0 {
-			return nil, errors.New("没有权限")
-		}
-	}
-
-	// 响应认证成功
-	_, err = writer.Write([]byte{socks.AuthVersion, socks.AuthSuccess})
-	if err != nil {
-		slog.Error("响应认证失败", "err", err)
-		return nil, err
-	}
-
-	return &socks.Authentication{
-		Method:  socks.UserPassAuth,
-		Timeout: uint(timeout),
-		Payload: socks.Payload{
-			ID: channel.UserId,
-		},
-	}, nil
-}

server/fwd/core/auth.go

@@ -0,0 +1,136 @@
+package core
+
+import (
+	"log/slog"
+	"net"
+	"proxy-server/server/models"
+	"proxy-server/server/pkg/orm"
+	"time"
+
+	"github.com/pkg/errors"
+)
+
+type Payload struct {
+	ID uint
+}
+
+type AuthContext struct {
+	Timeout float64
+	Payload Payload
+	Meta    map[string]any
+}
+
+func CheckIp(conn net.Conn) (*AuthContext, error) {
+
+	// 获取用户地址
+	remoteAddr := conn.RemoteAddr().String()
+	remoteHost, _, err := net.SplitHostPort(remoteAddr)
+	if err != nil {
+		return nil, errors.Wrap(err, "noAuth 认证失败")
+	}
+
+	// 获取服务端口
+	localAddr := conn.LocalAddr().String()
+	_, localPort, err := net.SplitHostPort(localAddr)
+
+	// 查询权限记录
+	slog.Info("用户 " + remoteHost + " 请求连接到 " + localPort)
+	var channels []models.Channel
+	err = orm.DB.
+		Joins("INNER JOIN public.nodes n ON channels.node_id = n.id AND n.name = ?", localPort).
+		Joins("INNER JOIN public.users u ON channels.user_id = u.id").
+		Joins("INNER JOIN public.user_ips ip ON u.id = ip.user_id AND ip.ip_address = ?", remoteHost).
+		Where(&models.Channel{
+			AuthIp: true,
+		}).
+		Find(&channels).Error
+	if err != nil {
+		return nil, errors.New("查询用户权限失败")
+	}
+
+	// 记录应该只有一条
+	channel, err := orm.MaySingle(channels)
+	if err != nil {
+		return nil, errors.Wrap(err, "不在白名单内")
+	}
+
+	// 检查是否需要密码认证
+	if channel.AuthPass {
+		return nil, errors.New("需要密码认证")
+	}
+
+	// 检查权限是否过期
+	timeout := channel.Expiration.Sub(time.Now()).Seconds()
+	if timeout <= 0 {
+		return nil, errors.New("权限已过期")
+	}
+
+	return &AuthContext{
+		Timeout: timeout,
+		Payload: Payload{
+			channel.UserId,
+		},
+	}, nil
+}
+
+func CheckPass(conn net.Conn, username, password string) (*AuthContext, error) {
+
+	// 查询通道配置
+	var channel models.Channel
+	err := orm.DB.
+		Where(&models.Channel{
+			Username: username,
+			AuthPass: true,
+		}).
+		First(&channel).Error
+	if err != nil {
+		return nil, errors.Wrap(err, "用户不存在")
+	}
+
+	// 检查密码 todo 哈希
+	if channel.Password != password {
+		return nil, errors.New("密码错误")
+	}
+
+	// 检查权限是否过期
+	timeout := channel.Expiration.Sub(time.Now()).Seconds()
+	if timeout <= 0 {
+		return nil, errors.New("权限已过期")
+	}
+
+	// 如果用户设置了双验证则检查 ip 是否在白名单中
+	if channel.AuthIp {
+		slog.Debug("验证用户 ip")
+
+		// 获取用户地址
+		remoteAddr := conn.RemoteAddr().String()
+		remoteHost, _, err := net.SplitHostPort(remoteAddr)
+		if err != nil {
+			return nil, errors.Wrap(err, "无法获取连接信息")
+		}
+
+		// 查询通道配置
+
+		var ips int64
+		err = orm.DB.
+			Where(&models.UserIp{
+				UserId:    channel.UserId,
+				IpAddress: remoteHost,
+			}).
+			Count(&ips).Error
+		if err != nil {
+			return nil, errors.Wrap(err, "查询白名单失败")
+		}
+
+		if ips == 0 {
+			return nil, errors.New("不在白名单内")
+		}
+	}
+
+	return &AuthContext{
+		Timeout: timeout,
+		Payload: Payload{
+			channel.UserId,
+		},
+	}, nil
+}

server/fwd/core/conn.go

@@ -0,0 +1,67 @@
+package core
+
+import (
+	"bufio"
+	"fmt"
+	"net"
+	"time"
+)
+
+type Conn struct {
+	Conn     net.Conn
+	Reader   *bufio.Reader
+	Tag      string
+	Protocol string
+	Dest     *FwdAddr
+	Auth     *AuthContext
+}
+
+func (c Conn) Read(b []byte) (n int, err error) {
+	return c.Reader.Read(b)
+}
+
+func (c Conn) Write(b []byte) (n int, err error) {
+	return c.Conn.Write(b)
+}
+
+func (c Conn) Close() error {
+	return c.Conn.Close()
+}
+
+func (c Conn) LocalAddr() net.Addr {
+	return c.Conn.LocalAddr()
+}
+
+func (c Conn) RemoteAddr() net.Addr {
+	return c.Conn.RemoteAddr()
+}
+
+func (c Conn) SetDeadline(t time.Time) error {
+	return c.Conn.SetDeadline(t)
+}
+
+func (c Conn) SetReadDeadline(t time.Time) error {
+	return c.Conn.SetReadDeadline(t)
+}
+
+func (c Conn) SetWriteDeadline(t time.Time) error {
+	return c.Conn.SetWriteDeadline(t)
+}
+
+func (c Conn) DestAddr() net.Addr {
+	return c.Dest
+}
+
+type FwdAddr struct {
+	IP     net.IP
+	Port   int
+	Domain string
+}
+
+func (a FwdAddr) Network() string {
+	return "tcp"
+}
+
+func (a FwdAddr) String() string {
+	return fmt.Sprintf("%s:%d", a.IP, a.Port)
+}

server/fwd/dispatcher/dispatch.go

@@ -0,0 +1,142 @@
+package dispatcher
+
+import (
+	"context"
+	"log/slog"
+	"net"
+	"proxy-server/pkg/utils"
+	"proxy-server/server/fwd/core"
+	"proxy-server/server/fwd/socks"
+	"strconv"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/soheilhy/cmux"
+)
+
+type Server struct {
+	ctx    context.Context
+	cancel context.CancelFunc
+	Port   uint16
+	Conn   chan *core.Conn
+}
+
+func New(port uint16) (*Server, error) {
+
+	if port == 0 {
+		return nil, errors.New("port is required")
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	return &Server{
+		ctx,
+		cancel,
+		port,
+		make(chan *core.Conn),
+	}, nil
+}
+
+func (s *Server) Close() {
+	s.cancel()
+}
+
+func (s *Server) Run() error {
+	port := strconv.Itoa(int(s.Port))
+
+	ls, err := net.Listen("tcp", ":"+port)
+	if err != nil {
+		return errors.Wrap(err, "dispatcher 监听失败")
+	}
+
+	m := cmux.New(ls)
+	m.SetReadTimeout(5 * time.Second)
+
+	go func() {
+		<-s.ctx.Done()
+		close(s.Conn)
+		m.Close()
+	}()
+
+	socksLs := m.Match(cmux.PrefixMatcher(string([]byte{0x05})))
+	defer utils.Close(socksLs)
+	go func() {
+		err = s.acceptSocks(socksLs)
+		if err != nil {
+			slog.Error("dispatcher socks accept error", "err", err)
+		}
+	}()
+
+	httpLs := m.Match(cmux.HTTP1Fast("PATCH"))
+	defer utils.Close(httpLs)
+	go func() {
+		err = s.acceptHttp(httpLs)
+		if err != nil {
+			slog.Error("dispatcher http accept error", "err", err)
+		}
+	}()
+
+	err = m.Serve()
+	if err != nil {
+		return errors.Wrap(err, "dispatcher serve error")
+	}
+
+	return nil
+}
+
+func (s *Server) acceptHttp(ls net.Listener) error {
+	for {
+		conn, err := ls.Accept()
+		if err != nil {
+			if errors.Is(err, net.ErrClosed) {
+				return nil
+			}
+			var ne net.Error
+			if errors.As(err, &ne) && ne.Temporary() {
+				continue
+			}
+			return errors.Wrap(err, "dispatcher http accept error")
+		}
+
+		go func() {
+			err := s.processHttp(conn)
+			if err != nil {
+				slog.Error("dispatcher http process error", "err", err)
+			}
+		}()
+	}
+}
+
+func (s *Server) acceptSocks(ls net.Listener) error {
+	for {
+		conn, err := ls.Accept()
+		if err != nil {
+			if errors.Is(err, net.ErrClosed) {
+				return nil
+			}
+			var ne net.Error
+			if errors.As(err, &ne) && ne.Temporary() {
+				continue
+			}
+			return errors.Wrap(err, "dispatcher socks accept error")
+		}
+
+		go func() {
+			conn, err := socks.Process(s.ctx, conn)
+			if err != nil {
+				slog.Error("处理 socks 连接失败", "err", err)
+			}
+			select {
+			case <-s.ctx.Done():
+				utils.Close(conn)
+			case s.Conn <- conn:
+			}
+		}()
+	}
+}
+
+func (s *Server) processHttp(conn net.Conn) error {
+	return nil
+}
+
+type Conn struct {
+}

server/fwd/fwd.go

@@ -8,11 +8,11 @@ import (
 	"log/slog"
 	"net"
 	"proxy-server/pkg/utils"
-	"proxy-server/server/fwd/socks"
+	"proxy-server/server/fwd/core"
+	"proxy-server/server/fwd/dispatcher"
 	"proxy-server/server/pkg/env"
 	"strconv"
 	"sync"
-	"time"
 
 	"github.com/pkg/errors"
 )
@@ -21,14 +21,17 @@ type Config struct {
 }
 
 type Service struct {
-	Config      *Config
+	Config *Config
-	ctx         context.Context
+	ctx    context.Context
-	cancel      context.CancelFunc
+	cancel context.CancelFunc
-	userConnMap sync.Map
 
+	userConnMap sync.Map
+	ctrlConnMap sync.Map
+
+	fwdLesWg   utils.CountWaitGroup
 	ctrlConnWg utils.CountWaitGroup
 	dataConnWg utils.CountWaitGroup
-	fwdLesWg   utils.CountWaitGroup
+	userConnWg utils.CountWaitGroup
 }
 
 func New(config *Config) *Service {
@@ -42,16 +45,17 @@ func New(config *Config) *Service {
 		ctx:         ctx,
 		cancel:      cancel,
 		userConnMap: sync.Map{},
-		ctrlConnWg:  utils.CountWaitGroup{},
+		ctrlConnMap: sync.Map{},
-		dataConnWg:  utils.CountWaitGroup{},
+
-		fwdLesWg:    utils.CountWaitGroup{},
+		fwdLesWg:   utils.CountWaitGroup{},
+		ctrlConnWg: utils.CountWaitGroup{},
+		dataConnWg: utils.CountWaitGroup{},
+		userConnWg: utils.CountWaitGroup{},
 	}
 }
 
 func (s *Service) Close() {
-	start := time.Now()
 	s.cancel()
-	slog.Debug("退出服务", "duration", time.Since(start))
 }
 
 func (s *Service) Run() {
@@ -95,15 +99,28 @@ func (s *Service) Run() {
 		s.Close()
 	}
 
+	wg.Wait()
+	// 协程建立有先后顺序，不能乱，否则会泄露
+	s.dataConnWg.Wait()
+	s.ctrlConnWg.Wait()
+	s.fwdLesWg.Wait()
+	s.userConnWg.Wait()
+
 	// 清理资源
 	s.userConnMap.Range(func(key, value any) bool {
-		conn := value.(socks.ProxyConn)
+		conn := value.(core.Conn)
 		utils.Close(conn)
-		s.userConnMap.Delete(key)
 		return true
 	})
 	s.userConnMap.Clear()
 
+	s.ctrlConnMap.Range(func(key, value any) bool {
+		conn := value.(net.Conn)
+		utils.Close(conn)
+		return true
+	})
+	s.ctrlConnMap.Clear()
+
 	s.ctrlConnWg.Wait()
 	slog.Debug("控制通道连接已关闭")
 	s.dataConnWg.Wait()
@@ -125,38 +142,33 @@ func (s *Service) startCtrlTun() error {
 	}
 	defer utils.Close(ls)
 
-	// 等待连接
-	connCh := utils.ChanConnAccept(s.ctx, ls)
-	defer close(connCh)
-
 	// 处理连接
+	connCh := utils.ChanConnAccept(s.ctx, ls)
 	for {
 		select {
 		case <-s.ctx.Done():
-			slog.Debug("服务关闭 startCtrlTun")
 			return nil
 		case conn, ok := <-connCh:
 			if !ok {
-				slog.Debug("结束处理连接，由于获取连接失败")
 				return errors.New("获取连接失败")
 			}
 			s.ctrlConnWg.Add(1)
 			go func() {
 				defer s.ctrlConnWg.Done()
-				defer utils.Close(conn)
 				err := s.processCtrlConn(conn)
 				if err != nil {
 					slog.Error("处理控制通道连接失败", "err", err)
+					utils.Close(conn)
 				}
 			}()
 		}
 	}
 }
 
-func (s *Service) processCtrlConn(controller net.Conn) error {
+func (s *Service) processCtrlConn(conn net.Conn) error {
-	slog.Debug("客户端连入", "addr", controller.RemoteAddr().String())
+	slog.Debug("客户端连入", "addr", conn.RemoteAddr().String())
 
-	reader := bufio.NewReader(controller)
+	reader := bufio.NewReader(conn)
 
 	// 获取转发端口
 	portBuf, err := utils.ReadBuffer(reader, 2)
@@ -165,63 +177,19 @@ func (s *Service) processCtrlConn(controller net.Conn) error {
 	}
 	port := binary.BigEndian.Uint16(portBuf)
 
-	// 开放转发端口 todo 混合转发
+	// 开放转发端口
-	slog.Debug("开放转发端口", "port", port)
-	proxy, err := socks.New(&socks.Config{
-		Name: strconv.Itoa(int(port)),
-		Port: port,
-		AuthMethods: []socks.Authenticator{
-			&UserPassAuthenticator{},
-			&NoAuthAuthenticator{},
-		},
-	})
-	if err != nil {
-		return errors.Wrap(err, "创建 socks 转发服务失败")
-	}
-	defer proxy.Close()
-
 	s.fwdLesWg.Add(1)
 	go func() {
 		defer s.fwdLesWg.Done()
-		err := proxy.Run()
+		err := s.startFwdTun(port)
 		if err != nil {
 			slog.Error("代理服务启动失败", "err", err)
 			return
 		}
 	}()
 
-	// 等待用户连接
+	// 记录控制连接
-	wg := sync.WaitGroup{}
+	s.ctrlConnMap.Store(port, conn)
-	for loop := true; loop; {
-		select {
-		case <-s.ctx.Done():
-			loop = false
-		case user, ok := <-proxy.Conn:
-			if !ok {
-				loop = false
-				err = errors.New("无法获取连接")
-			}
-			wg.Add(1)
-			go func() {
-				defer wg.Done()
-
-				tag := user.Tag()
-				tagLen := len(tag)
-				tagBuf := make([]byte, 1+tagLen)
-				tagBuf[0] = byte(tagLen)
-				copy(tagBuf[1:], tag)
-				_, err := controller.Write(tagBuf)
-				if err != nil {
-					utils.Close(user)
-					slog.Error("向客户端发送 tag 失败", "err", err)
-					return
-				}
-				s.userConnMap.Store(tag, user)
-			}()
-		}
-	}
-
-	wg.Wait()
 	return nil
 }
 
@@ -236,19 +204,14 @@ func (s *Service) startDataTun() error {
 	}
 	defer utils.Close(ls)
 
-	// 等待连接
-	connCh := utils.ChanConnAccept(s.ctx, ls)
-	defer close(connCh)
-
 	// 处理连接
+	connCh := utils.ChanConnAccept(s.ctx, ls)
 	for {
 		select {
 		case <-s.ctx.Done():
-			slog.Debug("服务关闭 startDataTun")
 			return nil
 		case conn, ok := <-connCh:
 			if !ok {
-				slog.Debug("结束处理连接，由于获取连接失败")
 				return errors.New("获取连接失败")
 			}
 			s.dataConnWg.Add(1)
@@ -268,67 +231,60 @@ func (s *Service) processDataConn(client net.Conn) error {
 	slog.Info("客户端准备接收数据 " + client.RemoteAddr().String())
 
 	// 读取 tag
-	tagLen, err := utils.ReadByte(client)
+	var tag string
-	if err != nil {
-		return errors.Wrap(err, "从客户端获取 tag 失败")
-	}
-	tagBuf, err := utils.ReadBuffer(client, int(tagLen))
-	if err != nil {
-		return errors.Wrap(err, "从客户端获取 tag 失败")
-	}
-	tag := string(tagBuf)
-
-	// 找到用户连接
-	var data socks.ProxyConn
 	select {
 	case <-s.ctx.Done():
 		return nil
 	default:
-		dataAny, ok := s.userConnMap.Load(tag)
+		tagLen, err := utils.ReadByte(client)
-		if !ok {
+		if err != nil {
-			return errors.New("查找用户连接失败")
+			return errors.Wrap(err, "从客户端获取 tag 失败")
 		}
-		data = dataAny.(socks.ProxyConn)
+		tagBuf, err := utils.ReadBuffer(client, int(tagLen))
-		defer func() {
+		if err != nil {
-			s.userConnMap.Delete(tag)
+			return errors.Wrap(err, "从客户端获取 tag 失败")
-			utils.Close(data)
+		}
-		}()
+		tag = string(tagBuf)
 	}
 
-	// 响应用户
+	// 找到用户连接
-	user := data.Conn
+	userAny, ok := s.userConnMap.Load(tag)
-	err = socks.SendSuccess(user, client)
+	if !ok {
-	if err != nil {
+		return errors.New("查找用户连接失败")
-		// todo 考虑是否需要处理服务关闭后导致用户连接被关闭的情况
-		return errors.Wrap(err, "向用户发送成功消息失败")
 	}
+	user := userAny.(*core.Conn)
+	defer utils.Close(user)
+	defer s.userConnMap.Delete(tag)
 
 	// 发送目标地址
-	dest := data.Dest
+	select {
-	destLen := len(dest)
+	case <-s.ctx.Done():
-	destBuf := make([]byte, 1+destLen)
+		return nil
-	destBuf[0] = byte(destLen)
+	default:
-	copy(destBuf[1:], dest)
+		dest := user.Dest.String()
-	_, err = client.Write(destBuf)
+		destLen := len(dest)
-	if err != nil {
+		destBuf := make([]byte, 1+destLen)
-		return errors.Wrap(err, "向客户端发送目标地址失败")
+		destBuf[0] = byte(destLen)
+		copy(destBuf[1:], dest)
+		_, err := client.Write(destBuf)
+		if err != nil {
+			return errors.Wrap(err, "向客户端发送目标地址失败")
+		}
 	}
 
 	// 数据转发
-	slog.Info("开始数据转发 " + client.RemoteAddr().String() + " <-> " + data.Dest)
-
 	userPipeReader, userPipeWriter := io.Pipe()
 	defer utils.Close(userPipeWriter)
 	teeUser := io.TeeReader(user, userPipeWriter)
 	go func() {
-		err := analysisAndLog(data, userPipeReader)
+		err := analysisAndLog(user, userPipeReader)
 		if err != nil {
 			slog.Error("数据解析失败", "err", err)
 		}
 	}()
 
 	wg := sync.WaitGroup{}
-	wg.Add(1)
+	wg.Add(2)
 	go func() {
 		defer wg.Done()
 		_, err := io.Copy(client, teeUser)
@@ -336,7 +292,6 @@ func (s *Service) processDataConn(client net.Conn) error {
 			slog.Error("数据转发失败 user->client", "err", err)
 		}
 	}()
-	wg.Add(1)
 	go func() {
 		defer wg.Done()
 		_, err := io.Copy(user, client)
@@ -348,8 +303,76 @@ func (s *Service) processDataConn(client net.Conn) error {
 			}
 		}
 	}()
-	wg.Wait()
 
-	slog.Info("数据转发结束 " + client.RemoteAddr().String() + " <-> " + data.Dest)
+	select {
+	case <-s.ctx.Done():
+	case <-utils.ChanWgWait(s.ctx, &wg):
+	}
+
 	return nil
 }
+
+func (s *Service) startFwdTun(port uint16) error {
+	slog.Debug("监听转发通道", "port", port)
+
+	proxy, err := dispatcher.New(port)
+	if err != nil {
+		return errors.Wrap(err, "创建 socks 转发服务失败")
+	}
+	defer proxy.Close()
+
+	go func() {
+		err := proxy.Run()
+		if err != nil {
+			slog.Error("代理服务异常退出", "err", err)
+		}
+	}()
+
+	for {
+		select {
+		case <-s.ctx.Done():
+			return nil
+		case conn := <-proxy.Conn:
+			s.userConnWg.Add(1)
+			go func() {
+				defer s.userConnWg.Done()
+				err := s.processUserConn(conn, port)
+				if err != nil {
+					slog.Error("处理用户连接失败", "err", err)
+				}
+			}()
+		}
+	}
+}
+
+func (s *Service) processUserConn(conn *core.Conn, port uint16) error {
+
+	// 记录用户连接
+	s.userConnMap.Store(conn.Tag, conn)
+
+	// 通知客户端建立数据通道
+	ctrlConnAny, ok := s.ctrlConnMap.Load(port)
+	if !ok {
+		return errors.New("查找控制连接失败")
+	}
+	ctrlConn := ctrlConnAny.(net.Conn)
+
+	// 发送 tag
+	select {
+	case <-s.ctx.Done():
+		return nil
+	default:
+		tag := conn.Tag
+		tagLen := len(tag)
+		tagBuf := make([]byte, 1+tagLen)
+		tagBuf[0] = byte(tagLen)
+		copy(tagBuf[1:], tag)
+		_, err := ctrlConn.Write(tagBuf)
+		if err != nil {
+			return errors.Wrap(err, "向控制通道发送 tag 失败")
+		}
+	}
+
+	return nil
+
+}

server/fwd/http/http.go

@@ -1,4 +1 @@
 package http
-
-func Start() {
-}

server/fwd/socks/auth.go

@@ -1,87 +0,0 @@
-package socks
-
-import (
-	"context"
-	"io"
-	"log/slog"
-	"proxy-server/pkg/utils"
-	"slices"
-
-	"github.com/pkg/errors"
-)
-
-type AuthMethod byte
-
-const (
-	AuthVersion  = byte(1)
-	AuthSuccess  = byte(0)
-	AuthFailure  = byte(1)
-	NoAuth       = AuthMethod(0)
-	UserPassAuth = AuthMethod(2)
-	NoAcceptable = byte(0xFF)
-)
-
-type Authenticator interface {
-	Method() AuthMethod
-	Authenticate(ctx context.Context, reader io.Reader, writer io.Writer) (*Authentication, error)
-}
-
-// authenticate 执行认证流程
-func (s *Server) authenticate(reader io.Reader, writer io.Writer) (*Authentication, error) {
-
-	// 版本检查
-	err := checkVersion(reader)
-	if err != nil {
-		return nil, err
-	}
-
-	// 获取客户端认证方式
-	nAuth, err := utils.ReadByte(reader)
-	if err != nil {
-		return nil, err
-	}
-	methods, err := utils.ReadBuffer(reader, int(nAuth))
-	if err != nil {
-		return nil, err
-	}
-
-	// 认证客户端连接
-	for _, authenticator := range s.config.AuthMethods {
-		method := authenticator.Method()
-		if slices.Contains(methods, byte(method)) {
-			slog.Debug("使用的认证方式", method)
-
-			_, err := writer.Write([]byte{Version, byte(method)})
-			if err != nil {
-				slog.Error("响应认证方式失败", err)
-				return nil, err
-			}
-
-			ctx := context.WithValue(context.Background(), "service", s)
-			authContext, err := authenticator.Authenticate(ctx, reader, writer)
-			if err != nil {
-				return nil, err
-			}
-			return authContext, nil
-		}
-	}
-
-	// 无适用的认证方式
-	_, err = writer.Write([]byte{Version, NoAcceptable})
-	if err != nil {
-		return nil, err
-	}
-
-	return nil, errors.New("没有适用的认证方式")
-}
-
-type Authentication struct {
-	Method  AuthMethod
-	Timeout uint
-	Payload Payload
-	Data    map[string]any
-}
-
-type Payload struct {
-	ID uint
-}

server/fwd/socks/error.go

@@ -1,7 +0,0 @@
-package socks
-
-type ConfigError string
-
-func (c ConfigError) Error() string {
-	return string(c)
-}

server/fwd/socks/request.go

@@ -1,353 +0,0 @@
-package socks
-
-import (
-	"context"
-	"fmt"
-	"io"
-	"log/slog"
-	"net"
-	"proxy-server/pkg/utils"
-	"strconv"
-
-	"github.com/pkg/errors"
-)
-
-const (
-	ConnectCommand   = byte(1)
-	BindCommand      = byte(2)
-	AssociateCommand = byte(3)
-	ipv4Address      = byte(1)
-	fqdnAddress      = byte(3)
-	ipv6Address      = byte(4)
-)
-
-const (
-	successReply byte = iota
-	serverFailure
-	ruleFailure
-	networkUnreachable
-	hostUnreachable
-	connectionRefused
-	ttlExpired
-	commandNotSupported
-	addrTypeNotSupported
-)
-
-var (
-	unrecognizedAddrType = fmt.Errorf("unrecognized address type")
-)
-
-// AddressRewriter is used to rewrite a destination transparently
-type AddressRewriter interface {
-	Rewrite(ctx context.Context, request *Request) (context.Context, *AddrSpec)
-}
-
-// AddrSpec 地址
-type AddrSpec struct {
-	FQDN string
-	IP   net.IP
-	Port int
-}
-
-func (a AddrSpec) String() string {
-	if a.FQDN != "" {
-		return fmt.Sprintf("%s (%s):%d", a.FQDN, a.IP, a.Port)
-	}
-	return fmt.Sprintf("%s:%d", a.IP, a.Port)
-}
-
-// Address returns a string suitable to dial; prefer returning IP-based
-// address, fallback to FQDN
-func (a AddrSpec) Address() string {
-	if 0 != len(a.IP) {
-		return net.JoinHostPort(a.IP.String(), strconv.Itoa(a.Port))
-	}
-	return net.JoinHostPort(a.FQDN, strconv.Itoa(a.Port))
-}
-
-func (s *Server) request(reader io.Reader, writer io.Writer) (*Request, error) {
-
-	// 检查版本
-	err := checkVersion(reader)
-	if err != nil {
-		return nil, err
-	}
-
-	// 检查连接命令
-	command, err := utils.ReadByte(reader)
-	if err != nil {
-		return nil, err
-	}
-
-	slog.Debug("客户端使用的连接指令：%v", command)
-	if command != ConnectCommand && command != BindCommand && command != AssociateCommand {
-		err = sendReply(writer, commandNotSupported, nil)
-		if err != nil {
-			return nil, err
-		}
-		return nil, errors.New("不支持该连接指令")
-	}
-
-	// 跳过保留字段 rsv
-	_, err = utils.ReadByte(reader)
-	if err != nil {
-		return nil, err
-	}
-
-	// 获取目标地址
-	dest, err := s.parseTarget(reader, writer)
-	if err != nil {
-		return nil, err
-	}
-
-	request := &Request{
-		Version:  Version,
-		Command:  command,
-		DestAddr: dest,
-		bufConn:  reader,
-	}
-
-	return request, nil
-}
-
-func (s *Server) parseTarget(reader io.Reader, writer io.Writer) (*AddrSpec, error) {
-	dest := &AddrSpec{}
-
-	aTypeBuf := make([]byte, 1)
-	_, err := reader.Read(aTypeBuf)
-	if err != nil {
-		return nil, err
-	}
-
-	switch aTypeBuf[0] {
-
-	case ipv4Address:
-		addr := make([]byte, 4)
-		_, err := io.ReadFull(reader, addr)
-		if err != nil {
-			return nil, err
-		}
-		dest.IP = addr
-
-	case ipv6Address:
-		addr := make([]byte, 16)
-		_, err := io.ReadFull(reader, addr)
-		if err != nil {
-			return nil, err
-		}
-		dest.IP = addr
-
-	case fqdnAddress:
-		aLenBuf := make([]byte, 1)
-		_, err := reader.Read(aLenBuf)
-		if err != nil {
-			return nil, err
-		}
-
-		fqdnBuff := make([]byte, int(aLenBuf[0]))
-		_, err = io.ReadFull(reader, fqdnBuff)
-		if err != nil {
-			return nil, err
-		}
-		dest.FQDN = string(fqdnBuff)
-
-		// 域名解析
-		addr, err := s.config.Resolver.Resolve(dest.FQDN)
-		if err != nil {
-			err := sendReply(writer, hostUnreachable, nil)
-			if err != nil {
-				return nil, fmt.Errorf("failed to send reply: %v", err)
-			}
-			return nil, fmt.Errorf("failed to resolve destination '%v': %v", dest.FQDN, err)
-		}
-		dest.IP = addr
-
-	default:
-		err := sendReply(writer, addrTypeNotSupported, nil)
-		if err != nil {
-			return nil, err
-		}
-		return nil, unrecognizedAddrType
-	}
-
-	portBuf := make([]byte, 2)
-	_, err = io.ReadFull(reader, portBuf)
-	if err != nil {
-		return nil, err
-	}
-	dest.Port = (int(portBuf[0]) << 8) | int(portBuf[1])
-
-	return dest, nil
-}
-
-// A Request represents request received by a server
-type Request struct {
-	// Protocol version
-	Version uint8
-	// Requested command
-	Command uint8
-	// Authentication provided during negotiation
-	Authentication *Authentication
-	// AddrSpec of the  network that sent the request
-	RemoteAddr *AddrSpec
-	// AddrSpec of the desired destination
-	DestAddr *AddrSpec
-	// AddrSpec of the actual destination (might be affected by rewrite)
-	realDestAddr *AddrSpec
-	bufConn      io.Reader
-}
-
-func (s *Server) handle(req *Request, conn net.Conn) error {
-	ctx := context.Background()
-
-	// 目标地址重写
-	req.realDestAddr = req.DestAddr
-	if s.config.Rewriter != nil {
-		ctx, req.realDestAddr = s.config.Rewriter.Rewrite(ctx, req)
-	}
-
-	// 根据协商方法建立连接
-	switch req.Command {
-	case ConnectCommand:
-		return s.handleConnect(ctx, conn, req)
-	case BindCommand:
-		return s.handleBind(ctx, conn, req)
-	case AssociateCommand:
-		return s.handleAssociate(ctx, conn, req)
-	default:
-		return fmt.Errorf("unsupported command: %v", req.Command)
-	}
-}
-
-func (s *Server) handleConnect(ctx context.Context, conn net.Conn, req *Request) error {
-	// 检查规则集约束
-	s.config.Logger.Printf("检查约束规则\n")
-	if ctx_, ok := s.config.Rules.Allow(ctx, req); !ok {
-		if err := sendReply(conn, ruleFailure, nil); err != nil {
-			return fmt.Errorf("failed to send reply: %v", err)
-		}
-		return fmt.Errorf("request to %v blocked by rules", req.DestAddr)
-	} else {
-		ctx = ctx_
-	}
-
-	slog.Info("需要向 " + req.DestAddr.Address() + " 建立连接")
-	select {
-	case <-s.ctx.Done():
-		if conn != nil {
-			utils.Close(conn)
-		}
-	case s.Conn <- ProxyConn{
-		req.Authentication.Payload.ID,
-		conn,
-		req.realDestAddr.Address(),
-	}:
-	}
-	return nil
-}
-
-func (s *Server) handleBind(ctx context.Context, conn net.Conn, req *Request) error {
-	// Check if this is allowed
-	if ctx_, ok := s.config.Rules.Allow(ctx, req); !ok {
-		if err := sendReply(conn, ruleFailure, nil); err != nil {
-			return fmt.Errorf("failed to send reply: %v", err)
-		}
-		return fmt.Errorf("bind to %v blocked by rules", req.DestAddr)
-	} else {
-		ctx = ctx_
-	}
-
-	// TODO: Support bind
-	if err := sendReply(conn, commandNotSupported, nil); err != nil {
-		return fmt.Errorf("failed to send reply: %v", err)
-	}
-	return nil
-}
-
-func (s *Server) handleAssociate(ctx context.Context, conn net.Conn, req *Request) error {
-	// Check if this is allowed
-	if ctx_, ok := s.config.Rules.Allow(ctx, req); !ok {
-		if err := sendReply(conn, ruleFailure, nil); err != nil {
-			return fmt.Errorf("failed to send reply: %v", err)
-		}
-		return fmt.Errorf("associate to %v blocked by rules", req.DestAddr)
-	} else {
-		ctx = ctx_
-	}
-
-	// TODO: Support associate
-	if err := sendReply(conn, commandNotSupported, nil); err != nil {
-		return fmt.Errorf("failed to send reply: %v", err)
-	}
-	return nil
-}
-
-func sendReply(w io.Writer, resp uint8, addr *AddrSpec) error {
-	var addrType uint8
-	var addrBody []byte
-	var addrPort uint16
-	switch {
-	case addr == nil:
-		addrType = ipv4Address
-		addrBody = []byte{0, 0, 0, 0}
-		addrPort = 0
-
-	case addr.FQDN != "":
-		addrType = fqdnAddress
-		addrBody = append([]byte{byte(len(addr.FQDN))}, addr.FQDN...)
-		addrPort = uint16(addr.Port)
-
-	case addr.IP.To4() != nil:
-		addrType = ipv4Address
-		addrBody = addr.IP.To4()
-		addrPort = uint16(addr.Port)
-
-	case addr.IP.To16() != nil:
-		addrType = ipv6Address
-		addrBody = addr.IP.To16()
-		addrPort = uint16(addr.Port)
-
-	default:
-		return fmt.Errorf("failed to format address: %v", addr)
-	}
-
-	msg := make([]byte, 6+len(addrBody))
-	msg[0] = Version
-	msg[1] = resp
-	msg[2] = 0 // Reserved
-	msg[3] = addrType
-	copy(msg[4:], addrBody)
-	msg[4+len(addrBody)] = byte(addrPort >> 8)
-	msg[4+len(addrBody)+1] = byte(addrPort & 0xff)
-
-	_, err := w.Write(msg)
-	return err
-}
-
-func SendSuccess(user net.Conn, target net.Conn) error {
-	local := target.LocalAddr().(*net.TCPAddr)
-	bind := AddrSpec{IP: local.IP, Port: local.Port}
-	err := sendReply(user, successReply, &bind)
-	if err != nil {
-		return err
-	}
-	return nil
-}
-
-type ProxyConn struct {
-	Uid uint
-	// 用户连入的连接
-	Conn net.Conn
-	// 用户目标地址
-	Dest string
-}
-
-func (d ProxyConn) Tag() string {
-	local := d.Conn.LocalAddr()
-	remote := d.Conn.RemoteAddr()
-	return fmt.Sprintf("%s-%s", remote, local)
-}
-
-func (d ProxyConn) Close() error {
-	return d.Conn.Close()
-}

server/fwd/socks/resolver.go

@@ -1,21 +0,0 @@
-package socks
-
-import (
-	"net"
-)
-
-// NameResolver 域名解析器
-type NameResolver interface {
-	Resolve(name string) (net.IP, error)
-}
-
-// DNSResolver 使用系统 dns 服务解析域名
-type DNSResolver struct{}
-
-func (d DNSResolver) Resolve(name string) (net.IP, error) {
-	addr, err := net.ResolveIPAddr("ip", name)
-	if err != nil {
-		return nil, err
-	}
-	return addr.IP, err
-}

server/fwd/socks/ruleset.go

@@ -1,41 +0,0 @@
-package socks
-
-import (
-	"context"
-)
-
-// RuleSet is used to provide custom rules to allow or prohibit actions
-type RuleSet interface {
-	Allow(ctx context.Context, req *Request) (context.Context, bool)
-}
-
-// PermitAll returns a RuleSet which allows all types of connections
-func PermitAll() RuleSet {
-	return &PermitCommand{true, true, true}
-}
-
-// PermitNone returns a RuleSet which disallows all types of connections
-func PermitNone() RuleSet {
-	return &PermitCommand{false, false, false}
-}
-
-// PermitCommand is an implementation of the RuleSet which
-// enables filtering supported commands
-type PermitCommand struct {
-	EnableConnect   bool
-	EnableBind      bool
-	EnableAssociate bool
-}
-
-func (p *PermitCommand) Allow(ctx context.Context, req *Request) (context.Context, bool) {
-	switch req.Command {
-	case ConnectCommand:
-		return ctx, p.EnableConnect
-	case BindCommand:
-		return ctx, p.EnableBind
-	case AssociateCommand:
-		return ctx, p.EnableAssociate
-	}
-
-	return ctx, false
-}

server/fwd/socks/socks.go

@@ -3,212 +3,90 @@ package socks
 import (
 	"bufio"
 	"context"
+	"encoding/binary"
 	"fmt"
 	"io"
-	"log"
 	"log/slog"
 	"net"
-	"os"
 	"proxy-server/pkg/utils"
-	"strconv"
+	"proxy-server/server/fwd/core"
-	"time"
+	"slices"
 
 	"github.com/pkg/errors"
 )
 
 const (
-	Version = byte(5)
+	Version     = byte(5)
+	AuthVersion = byte(1)
 )
 
-type Config struct {
+const (
-	Name string
+	NoAuth       = byte(0)
+	UserPassAuth = byte(2)
+	NoAcceptable = byte(0xFF)
+)
 
-	Host string
+const (
-	Port uint16
+	AuthSuccess = byte(0)
+	AuthFailure = byte(1)
+)
 
-	// 认证方法
+const (
-	AuthMethods []Authenticator
+	ConnectCommand   = byte(1)
+	BindCommand      = byte(2)
+	AssociateCommand = byte(3)
+)
 
-	// 域名解析
+const (
-	Resolver NameResolver
+	ipv4Address = byte(1)
+	fqdnAddress = byte(3)
+	ipv6Address = byte(4)
+)
 
-	// 自定义认证规则
+const (
-	Rules RuleSet
+	successReply byte = iota
-
+	serverFailure
-	// 地址重写
+	ruleFailure
-	Rewriter AddressRewriter
+	networkUnreachable
-
+	hostUnreachable
-	// 用于 bind 和 associate
+	connectionRefused
-	BindIP net.IP
+	ttlExpired
-
+	commandNotSupported
-	// Logger
+	addrTypeNotSupported
-	Logger *log.Logger
+)
-
-	// 自定义连接流程
-	Dial func(network, addr string) (net.Conn, error)
-}
-
-type Server struct {
-	config *Config
-	ctx    context.Context
-	cancel context.CancelFunc
-	wg     utils.CountWaitGroup
-	Name   string
-	Port   uint16
-	Conn   chan ProxyConn
-}
-
-// New 创建服务器
-func New(conf *Config) (*Server, error) {
-	if conf == nil {
-		conf = &Config{}
-	}
-
-	if len(conf.AuthMethods) == 0 {
-		return nil, ConfigError("认证方法不能为空")
-	}
-
-	if conf.Resolver == nil {
-		conf.Resolver = DNSResolver{}
-	}
-
-	if conf.Rules == nil {
-		conf.Rules = PermitAll()
-	}
-
-	if conf.Logger == nil {
-		conf.Logger = log.New(os.Stdout, "", log.LstdFlags)
-	}
-
-	if conf.Dial == nil {
-		conf.Dial = func(network, addr string) (net.Conn, error) {
-			return net.Dial(network, addr)
-		}
-	}
-
-	ctx, cancel := context.WithCancel(context.Background())
-	return &Server{
-		config: conf,
-		ctx:    ctx,
-		cancel: cancel,
-		wg:     utils.CountWaitGroup{},
-		Name:   conf.Name,
-		Port:   conf.Port,
-		Conn:   make(chan ProxyConn),
-	}, nil
-}
-
-// Run 监听端口
-func (s *Server) Run() error {
-	slog.Debug("启动 socks5 代理服务")
-
-	// 监听端口
-	host := s.config.Host
-	port := s.config.Port
-	addr := net.JoinHostPort(host, strconv.Itoa(int(port)))
-	ls, err := net.Listen("tcp", addr)
-	if err != nil {
-		return errors.Wrap(err, "监听端口失败")
-	}
-	defer utils.Close(ls)
-	slog.Debug("正在监听端口", slog.Uint64("port", uint64(port)))
-
-	// 处理连接
-	connCh := utils.ChanConnAccept(s.ctx, ls)
-	defer close(connCh)
-
-	err = nil
-	for loop := true; loop; {
-		select {
-		case <-s.ctx.Done():
-			slog.Debug("socks 服务主动停止")
-			loop = false
-		case conn, ok := <-connCh:
-			if !ok {
-				err = errors.New("意外错误，无法获取连接")
-				loop = false
-				s.Close()
-				break
-			}
-			s.wg.Add(1)
-			go func() {
-				defer s.wg.Done()
-				// 连接要传出，不能在这里关闭连接
-				err := s.process(conn)
-				if err != nil {
-					slog.Error("处理连接失败", err)
-				}
-			}()
-		}
-	}
-
-	// 关闭服务
-	timeout, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-	wgCh := utils.ChanWgWait(timeout, &s.wg)
-
-	err = nil
-	select {
-	case <-timeout.Done():
-		err = errors.New("关闭超时（强制关闭）")
-	case <-wgCh:
-	}
-
-	close(s.Conn)
-	return err
-}
-
-// Close 关闭服务
-func (s *Server) Close() {
-	s.cancel()
-}
-
-// process 建立连接
-func (s *Server) process(conn net.Conn) error {
-	slog.Info("收到来自" + conn.RemoteAddr().String() + "的连接")
 
+// Process 处理连接
+func Process(ctx context.Context, conn net.Conn) (*core.Conn, error) {
 	reader := bufio.NewReader(conn)
 
 	// 认证
-	slog.Debug("开始认证流程")
+	auth, err := authenticate(ctx, reader, conn)
-	authContext, err := s.authenticate(reader, conn)
 	if err != nil {
-		utils.Close(conn)
+		return nil, errors.Wrap(err, "认证失败")
-		slog.Error("认证失败", err)
-		return err
-	} else {
-		slog.Debug("认证完成")
 	}
 
 	// 处理连接请求
-	slog.Debug("处理连接请求")
+	request, err := request(ctx, reader, conn)
-	request, err := s.request(reader, conn)
 	if err != nil {
-		slog.Error("连接请求处理失败", err)
+		return nil, errors.Wrap(err, "处理连接请求失败")
-		return err
-	} else {
-		slog.Debug("连接请求处理完成")
 	}
 
-	request.Authentication = authContext
+	// 代理连接
-	user, ok := conn.RemoteAddr().(*net.TCPAddr)
+	if request.Command != ConnectCommand {
-	if !ok {
+		return nil, errors.New("不支持的连接指令")
-		return fmt.Errorf("获取用户地址失败")
 	}
 
-	request.RemoteAddr = &AddrSpec{
+	// 响应成功
-		IP:   user.IP,
+	err = sendReply(conn, successReply, request.DestAddr)
-		Port: user.Port,
-	}
 
-	// 处理请求
+	return &core.Conn{
-	slog.Debug("开始代理流量")
+		Conn:     conn,
-	err = s.handle(request, conn)
+		Reader:   reader,
-	if err != nil {
+		Protocol: "socks5",
-		return err
+		Tag:      conn.RemoteAddr().String() + "_" + conn.LocalAddr().String(),
-	}
+		Dest:     request.DestAddr,
-
+		Auth:     auth,
-	return nil
+	}, nil
 }
 
 // checkVersion 检查客户端版本
@@ -218,11 +96,280 @@ func checkVersion(reader io.Reader) error {
 		return err
 	}
 
-	slog.Debug("客户端请求版本", "version", version)
-
 	if version != Version {
 		return errors.New("客户端版本不兼容")
 	}
 
 	return nil
 }
+
+// authenticate 执行认证流程
+func authenticate(ctx context.Context, reader *bufio.Reader, conn net.Conn) (*core.AuthContext, error) {
+
+	// 版本检查
+	err := checkVersion(reader)
+	if err != nil {
+		return nil, err
+	}
+
+	// 获取客户端认证方式
+	nAuth, err := utils.ReadByte(reader)
+	if err != nil {
+		return nil, err
+	}
+	methods, err := utils.ReadBuffer(reader, int(nAuth))
+	if err != nil {
+		return nil, err
+	}
+
+	// 密码模式
+	if slices.Contains(methods, UserPassAuth) {
+		_, err := conn.Write([]byte{Version, byte(UserPassAuth)})
+		if err != nil {
+			return nil, errors.Wrap(err, "响应认证方式失败")
+		}
+
+		// 检查认证版本
+		slog.Debug("验证认证版本")
+		v, err := utils.ReadByte(reader)
+		if err != nil {
+			return nil, errors.Wrap(err, "读取版本号失败")
+		}
+		if v != AuthVersion {
+			_, err := conn.Write([]byte{Version, AuthFailure})
+			if err != nil {
+				return nil, errors.Wrap(err, "响应认证失败")
+			}
+			return nil, errors.New("认证版本参数不正确")
+		}
+
+		// 读取账号
+		slog.Debug("验证用户账号")
+		uLen, err := utils.ReadByte(reader)
+		if err != nil {
+			return nil, errors.Wrap(err, "读取用户名长度失败")
+		}
+		usernameBuf, err := utils.ReadBuffer(reader, int(uLen))
+		if err != nil {
+			return nil, errors.Wrap(err, "读取用户名失败")
+		}
+		username := string(usernameBuf)
+
+		// 读取密码
+		pLen, err := utils.ReadByte(reader)
+		if err != nil {
+			return nil, errors.Wrap(err, "读取密码长度失败")
+		}
+		passwordBuf, err := utils.ReadBuffer(reader, int(pLen))
+		if err != nil {
+			return nil, errors.Wrap(err, "读取密码失败")
+		}
+		password := string(passwordBuf)
+
+		// 检查权限
+		authContext, err := core.CheckPass(conn, username, password)
+		if err != nil {
+			return nil, errors.Wrap(err, "权限检查失败")
+		}
+
+		// 响应认证成功
+		_, err = conn.Write([]byte{AuthVersion, AuthSuccess})
+		if err != nil {
+			return nil, errors.Wrap(err, "响应认证成功失败")
+		}
+
+		return authContext, nil
+	}
+
+	// 无认证
+	if slices.Contains(methods, NoAuth) {
+		_, err = conn.Write([]byte{Version, NoAuth})
+		if err != nil {
+			return nil, errors.Wrap(err, "响应认证方式失败")
+		}
+
+		authContext, err := core.CheckIp(conn)
+		if err != nil {
+			return nil, errors.Wrap(err, "权限检查失败")
+		}
+
+		return authContext, nil
+	}
+
+	// 无适用的认证方式
+	_, err = conn.Write([]byte{Version, NoAcceptable})
+	if err != nil {
+		return nil, err
+	}
+
+	return nil, errors.New("没有适用的认证方式")
+}
+
+type Request struct {
+	Command  uint8
+	DestAddr *core.FwdAddr
+}
+
+// request 处理连接请求
+func request(ctx context.Context, reader io.Reader, writer io.Writer) (*Request, error) {
+
+	// 检查版本
+	err := checkVersion(reader)
+	if err != nil {
+		return nil, err
+	}
+
+	// 检查连接命令
+	command, err := utils.ReadByte(reader)
+	if err != nil {
+		return nil, err
+	}
+
+	if command != ConnectCommand && command != BindCommand && command != AssociateCommand {
+		err = sendReply(writer, commandNotSupported, nil)
+		if err != nil {
+			return nil, err
+		}
+		return nil, errors.New("不支持该连接指令")
+	}
+
+	// 跳过保留字段 rsv
+	_, err = utils.ReadByte(reader)
+	if err != nil {
+		return nil, err
+	}
+
+	// 获取目标地址
+	dest, err := parseTarget(reader, writer)
+	if err != nil {
+		return nil, err
+	}
+
+	request := &Request{
+		Command:  command,
+		DestAddr: dest,
+	}
+
+	return request, nil
+}
+
+func parseTarget(reader io.Reader, writer io.Writer) (*core.FwdAddr, error) {
+	dest := &core.FwdAddr{}
+
+	aTypeBuf, err := utils.ReadByte(reader)
+	if err != nil {
+		return nil, err
+	}
+
+	switch aTypeBuf {
+
+	case ipv4Address:
+		addr := make([]byte, 4)
+		_, err := io.ReadFull(reader, addr)
+		if err != nil {
+			return nil, err
+		}
+		dest.IP = addr
+
+	case ipv6Address:
+		addr := make([]byte, 16)
+		_, err := io.ReadFull(reader, addr)
+		if err != nil {
+			return nil, err
+		}
+		dest.IP = addr
+
+	case fqdnAddress:
+		aLenBuf := make([]byte, 1)
+		_, err := reader.Read(aLenBuf)
+		if err != nil {
+			return nil, err
+		}
+
+		fqdnBuff := make([]byte, int(aLenBuf[0]))
+		_, err = io.ReadFull(reader, fqdnBuff)
+		if err != nil {
+			return nil, err
+		}
+		dest.Domain = string(fqdnBuff)
+
+		// 域名解析
+		addr, err := net.ResolveIPAddr("ip", dest.Domain)
+		if err != nil {
+			err := sendReply(writer, hostUnreachable, nil)
+			if err != nil {
+				return nil, fmt.Errorf("failed to send reply: %v", err)
+			}
+			return nil, fmt.Errorf("failed to resolve destination '%v': %v", dest.Domain, err)
+		}
+		dest.IP = addr.IP
+
+	default:
+		err := sendReply(writer, addrTypeNotSupported, nil)
+		if err != nil {
+			return nil, err
+		}
+		return nil, fmt.Errorf("unrecognized address type")
+	}
+
+	portBuf := make([]byte, 2)
+	_, err = io.ReadFull(reader, portBuf)
+	if err != nil {
+		return nil, err
+	}
+	dest.Port = int(binary.BigEndian.Uint16(portBuf))
+
+	return dest, nil
+}
+
+func sendReply(w io.Writer, resp uint8, addr *core.FwdAddr) error {
+	var addrType uint8
+	var addrBody []byte
+	var addrPort uint16
+	switch {
+	case addr == nil:
+		addrType = ipv4Address
+		addrBody = []byte{0, 0, 0, 0}
+		addrPort = 0
+
+	case addr.Domain != "":
+		addrType = fqdnAddress
+		addrBody = append([]byte{byte(len(addr.Domain))}, addr.Domain...)
+		addrPort = uint16(addr.Port)
+
+	case addr.IP.To4() != nil:
+		addrType = ipv4Address
+		addrBody = addr.IP.To4()
+		addrPort = uint16(addr.Port)
+
+	case addr.IP.To16() != nil:
+		addrType = ipv6Address
+		addrBody = addr.IP.To16()
+		addrPort = uint16(addr.Port)
+
+	default:
+		return fmt.Errorf("failed to format address: %v", addr)
+	}
+
+	msg := make([]byte, 6+len(addrBody))
+	msg[0] = Version
+	msg[1] = resp
+	msg[2] = 0 // Reserved
+	msg[3] = addrType
+	copy(msg[4:], addrBody)
+	msg[4+len(addrBody)] = byte(addrPort >> 8)
+	msg[4+len(addrBody)+1] = byte(addrPort & 0xff)
+
+	_, err := w.Write(msg)
+	return err
+}
+
+func SendSuccess(user net.Conn, target net.Conn) error {
+	local := target.LocalAddr().(*net.TCPAddr)
+	bind := core.FwdAddr{IP: local.IP, Port: local.Port}
+	err := sendReply(user, successReply, &bind)
+	if err != nil {
+		return err
+	}
+	return nil
+}

server/server.go

@@ -66,11 +66,9 @@ func Start() {
 
 	timeout, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
-	wgCh := utils.ChanWgWait(timeout, &wg)
-	close(wgCh)
 
 	select {
-	case <-wgCh:
+	case <-utils.ChanWgWait(timeout, &wg):
 		slog.Info("服务已退出")
 	case <-timeout.Done():
 		slog.Warn("退出超时，强制退出")

server/web/handlers/channel.go

@@ -2,9 +2,9 @@ package handlers
 
 import (
 	"log/slog"
+	"proxy-server/server/models"
 	"proxy-server/server/pkg/orm"
 	"proxy-server/server/pkg/resp"
-	"proxy-server/server/web/models"
 	"strings"
 	"time"
 

server/web/handlers/node.go

@@ -2,8 +2,8 @@ package handlers
 
 import (
 	"os"
+	"proxy-server/server/models"
 	"proxy-server/server/pkg/orm"
-	"proxy-server/server/web/models"
 
 	"github.com/gin-gonic/gin"
 	"github.com/pkg/errors"