diff --git a/go-backend/internal/http/handler/control_plane.go b/go-backend/internal/http/handler/control_plane.go index 9cc5479..8fd2c33 100644 --- a/go-backend/internal/http/handler/control_plane.go +++ b/go-backend/internal/http/handler/control_plane.go @@ -1706,10 +1706,11 @@ func buildForwardServiceConfigs(baseName string, forward *forwardRecord, tunnel service["climiter"] = cLimiterName } if forward.ProxyProtocol > 0 { - if service["metadata"] == nil { - service["metadata"] = map[string]interface{}{} + handlerConfig := service["handler"].(map[string]interface{}) + if handlerConfig["metadata"] == nil { + handlerConfig["metadata"] = map[string]interface{}{} } - service["metadata"].(map[string]interface{})["proxyProtocol"] = forward.ProxyProtocol + handlerConfig["metadata"].(map[string]interface{})["proxyProtocol"] = forward.ProxyProtocol } if protocol == "udp" { listenerMetadata := map[string]interface{}{ @@ -1831,12 +1832,12 @@ func (h *Handler) ensureLimiterOnNode(nodeID int64, limiterID int64, speed int) func (h *Handler) ensureConnLimiterOnNode(nodeID int64, limiterName string, maxConn int) error { limitStr := fmt.Sprintf("$ %d", maxConn) - + payload := map[string]interface{}{ "name": limiterName, "limits": []string{limitStr}, } - + if _, err := h.sendNodeCommand(nodeID, "AddCLimiters", payload, false, false); err != nil { if !isAlreadyExistsMessage(err.Error()) { return fmt.Errorf("连接限制器下发失败: %w", err) @@ -1852,7 +1853,6 @@ func (h *Handler) ensureConnLimiterOnNode(nodeID int64, limiterName string, maxC return nil } - func buildLimiterAddPayload(limiterID int64, speed int) (string, map[string]interface{}) { rate := float64(speed) / 8.0 limitStr := fmt.Sprintf("$ %.1fMB %.1fMB", rate, rate) diff --git a/go-backend/internal/http/handler/forward_proxy_protocol_test.go b/go-backend/internal/http/handler/forward_proxy_protocol_test.go index dacaa17..e5db61c 100644 --- a/go-backend/internal/http/handler/forward_proxy_protocol_test.go +++ b/go-backend/internal/http/handler/forward_proxy_protocol_test.go @@ -8,7 +8,7 @@ import ( "go-backend/internal/store/repo" ) -func TestBuildForwardServiceConfigsPreservesProxyProtocolWithInterfaceMetadata(t *testing.T) { +func TestBuildForwardServiceConfigsSendsProxyProtocolToForwardHandler(t *testing.T) { forward := &forwardRecord{ ID: 1, UserID: 2, @@ -30,15 +30,27 @@ func TestBuildForwardServiceConfigsPreservesProxyProtocolWithInterfaceMetadata(t } for _, service := range services { - metadata, ok := service["metadata"].(map[string]interface{}) + serviceMetadata, ok := service["metadata"].(map[string]interface{}) if !ok { t.Fatalf("expected metadata map, got %T", service["metadata"]) } - if metadata["interface"] != "eth0" { - t.Fatalf("expected interface metadata eth0, got %v", metadata["interface"]) + if serviceMetadata["interface"] != "eth0" { + t.Fatalf("expected interface metadata eth0, got %v", serviceMetadata["interface"]) } - if metadata["proxyProtocol"] != 2 { - t.Fatalf("expected proxyProtocol 2, got %v", metadata["proxyProtocol"]) + if _, ok := serviceMetadata["proxyProtocol"]; ok { + t.Fatalf("proxyProtocol should not be listener metadata: %v", serviceMetadata) + } + + handlerConfig, ok := service["handler"].(map[string]interface{}) + if !ok { + t.Fatalf("expected handler config map, got %T", service["handler"]) + } + handlerMetadata, ok := handlerConfig["metadata"].(map[string]interface{}) + if !ok { + t.Fatalf("expected handler metadata map, got %T", handlerConfig["metadata"]) + } + if handlerMetadata["proxyProtocol"] != 2 { + t.Fatalf("expected handler proxyProtocol 2, got %v", handlerMetadata["proxyProtocol"]) } } } diff --git a/go-backend/internal/http/handler/mutations.go b/go-backend/internal/http/handler/mutations.go index 6d041a1..5541c4b 100644 --- a/go-backend/internal/http/handler/mutations.go +++ b/go-backend/internal/http/handler/mutations.go @@ -138,6 +138,15 @@ func (h *Handler) userUpdate(w http.ResponseWriter, r *http.Request) { response.WriteJSON(w, response.ErrDefault("请不要作死")) return } + oldUser, err := h.repo.GetUserByID(id) + if err != nil { + response.WriteJSON(w, response.Err(-2, err.Error())) + return + } + if oldUser == nil { + response.WriteJSON(w, response.ErrDefault("用户不存在")) + return + } dup, err := h.repo.UserExistsExcluding(username, id) if err != nil { @@ -210,6 +219,17 @@ func (h *Handler) userUpdate(w http.ResponseWriter, r *http.Request) { } } } + if oldUser.MaxConn != maxConn { + warnings, syncErr := h.syncUserMaxConnForwards(id) + if syncErr != nil { + response.WriteJSON(w, response.ErrDefault(fmt.Sprintf("最大连接数下发失败: %v", syncErr))) + return + } + if len(warnings) > 0 { + response.WriteJSON(w, response.OK(map[string]interface{}{"warnings": warnings})) + return + } + } response.WriteJSON(w, response.OKEmpty()) } @@ -4254,6 +4274,26 @@ func (h *Handler) syncUserTunnelForwards(userID, tunnelID int64) error { return nil } +func (h *Handler) syncUserMaxConnForwards(userID int64) ([]string, error) { + forwards, err := h.listActiveForwardsByUser(userID) + if err != nil { + return nil, err + } + warnings := make([]string, 0) + for i := range forwards { + f := &forwards[i] + if f.MaxConn > 0 { + continue + } + syncWarnings, syncErr := h.syncForwardServicesWithWarnings(f, "UpdateService", true) + warnings = append(warnings, syncWarnings...) + if syncErr != nil { + return warnings, syncErr + } + } + return warnings, nil +} + // cleanupForwardsForUserTunnel deletes all forwarding rules belonging to a // specific user+tunnel pair. It notifies nodes to remove the runtime services // first, then deletes the DB records. This is best-effort: individual failures diff --git a/go-backend/internal/store/repo/repository_control.go b/go-backend/internal/store/repo/repository_control.go index bff4fbb..038bf9d 100644 --- a/go-backend/internal/store/repo/repository_control.go +++ b/go-backend/internal/store/repo/repository_control.go @@ -45,15 +45,17 @@ func (r *Repository) ListForwardsByTunnelTx(tx *gorm.DB, tunnelID int64) ([]mode rows := make([]model.ForwardRecord, 0, len(forwards)) for _, f := range forwards { rows = append(rows, model.ForwardRecord{ - ID: f.ID, - UserID: f.UserID, - UserName: f.UserName, - Name: f.Name, - TunnelID: f.TunnelID, - RemoteAddr: f.RemoteAddr, - Strategy: f.Strategy, - Status: f.Status, - SpeedID: f.SpeedID, + ID: f.ID, + UserID: f.UserID, + UserName: f.UserName, + Name: f.Name, + TunnelID: f.TunnelID, + RemoteAddr: f.RemoteAddr, + Strategy: f.Strategy, + Status: f.Status, + SpeedID: f.SpeedID, + MaxConn: f.MaxConn, + ProxyProtocol: f.ProxyProtocol, }) } for i := range rows { @@ -64,7 +66,6 @@ func (r *Repository) ListForwardsByTunnelTx(tx *gorm.DB, tunnelID int64) ([]mode return rows, nil } - func (r *Repository) ListActiveTunnelIDsByNode(nodeID int64) ([]int64, error) { if r == nil || r.db == nil { return nil, errors.New("repository not initialized") @@ -142,7 +143,6 @@ func (r *Repository) ListForwardPortsTx(tx *gorm.DB, forwardID int64) ([]model.F return rows, nil } - func (r *Repository) HasOtherForwardOnNodePort(nodeID int64, port int, currentForwardID int64) (bool, error) { if r == nil || r.db == nil { return false, errors.New("repository not initialized") @@ -169,7 +169,6 @@ func (r *Repository) HasOtherForwardOnNodePortTx(tx *gorm.DB, nodeID int64, port return count > 0, nil } - func (r *Repository) GetTunnelOutProtocol(tunnelID int64) (string, error) { if r == nil || r.db == nil { return "", errors.New("repository not initialized") diff --git a/go-backend/internal/store/repo/repository_flow.go b/go-backend/internal/store/repo/repository_flow.go index 21d53cd..4396d14 100644 --- a/go-backend/internal/store/repo/repository_flow.go +++ b/go-backend/internal/store/repo/repository_flow.go @@ -114,15 +114,17 @@ func (r *Repository) ListActiveForwardsByUser(userID int64) ([]model.ForwardReco rows := make([]model.ForwardRecord, 0, len(forwards)) for _, f := range forwards { rows = append(rows, model.ForwardRecord{ - ID: f.ID, - UserID: f.UserID, - UserName: f.UserName, - Name: f.Name, - TunnelID: f.TunnelID, - RemoteAddr: f.RemoteAddr, - Strategy: f.Strategy, - Status: f.Status, - SpeedID: f.SpeedID, + ID: f.ID, + UserID: f.UserID, + UserName: f.UserName, + Name: f.Name, + TunnelID: f.TunnelID, + RemoteAddr: f.RemoteAddr, + Strategy: f.Strategy, + Status: f.Status, + SpeedID: f.SpeedID, + MaxConn: f.MaxConn, + ProxyProtocol: f.ProxyProtocol, }) } for i := range rows { @@ -145,15 +147,17 @@ func (r *Repository) ListActiveForwardsByUserTunnel(userID, tunnelID int64) ([]m rows := make([]model.ForwardRecord, 0, len(forwards)) for _, f := range forwards { rows = append(rows, model.ForwardRecord{ - ID: f.ID, - UserID: f.UserID, - UserName: f.UserName, - Name: f.Name, - TunnelID: f.TunnelID, - RemoteAddr: f.RemoteAddr, - Strategy: f.Strategy, - Status: f.Status, - SpeedID: f.SpeedID, + ID: f.ID, + UserID: f.UserID, + UserName: f.UserName, + Name: f.Name, + TunnelID: f.TunnelID, + RemoteAddr: f.RemoteAddr, + Strategy: f.Strategy, + Status: f.Status, + SpeedID: f.SpeedID, + MaxConn: f.MaxConn, + ProxyProtocol: f.ProxyProtocol, }) } for i := range rows { @@ -176,15 +180,17 @@ func (r *Repository) ListForwardsByUserAndTunnel(userID, tunnelID int64) ([]mode rows := make([]model.ForwardRecord, 0, len(forwards)) for _, f := range forwards { rows = append(rows, model.ForwardRecord{ - ID: f.ID, - UserID: f.UserID, - UserName: f.UserName, - Name: f.Name, - TunnelID: f.TunnelID, - RemoteAddr: f.RemoteAddr, - Strategy: f.Strategy, - Status: f.Status, - SpeedID: f.SpeedID, + ID: f.ID, + UserID: f.UserID, + UserName: f.UserName, + Name: f.Name, + TunnelID: f.TunnelID, + RemoteAddr: f.RemoteAddr, + Strategy: f.Strategy, + Status: f.Status, + SpeedID: f.SpeedID, + MaxConn: f.MaxConn, + ProxyProtocol: f.ProxyProtocol, }) } for i := range rows { diff --git a/go-backend/internal/store/repo/repository_forward_proxy_protocol_test.go b/go-backend/internal/store/repo/repository_forward_proxy_protocol_test.go index 42c27db..2918d4f 100644 --- a/go-backend/internal/store/repo/repository_forward_proxy_protocol_test.go +++ b/go-backend/internal/store/repo/repository_forward_proxy_protocol_test.go @@ -46,6 +46,111 @@ func TestGetForwardRecordIncludesProxyProtocol(t *testing.T) { } } +func TestListForwardsByTunnelIncludesProxyProtocol(t *testing.T) { + r, err := Open(":memory:") + if err != nil { + t.Fatalf("open repo: %v", err) + } + defer r.Close() + + now := time.Now().UnixMilli() + if err := r.DB().Create(&model.Forward{ + UserID: 1, + UserName: "admin", + Name: "proxy-forward", + TunnelID: 7, + RemoteAddr: "1.1.1.1:443", + Strategy: "fifo", + CreatedTime: now, + UpdatedTime: now, + Status: 1, + ProxyProtocol: 2, + }).Error; err != nil { + t.Fatalf("create forward: %v", err) + } + + records, err := r.ListForwardsByTunnel(7) + if err != nil { + t.Fatalf("ListForwardsByTunnel: %v", err) + } + if len(records) != 1 { + t.Fatalf("expected 1 forward record, got %d", len(records)) + } + if records[0].ProxyProtocol != 2 { + t.Fatalf("expected proxyProtocol 2, got %d", records[0].ProxyProtocol) + } +} + +func TestListForwardsByTunnelIncludesMaxConn(t *testing.T) { + r, err := Open(":memory:") + if err != nil { + t.Fatalf("open repo: %v", err) + } + defer r.Close() + + now := time.Now().UnixMilli() + if err := r.DB().Create(&model.Forward{ + UserID: 1, + UserName: "admin", + Name: "max-conn-forward", + TunnelID: 8, + RemoteAddr: "1.1.1.1:443", + Strategy: "fifo", + CreatedTime: now, + UpdatedTime: now, + Status: 1, + MaxConn: 42, + }).Error; err != nil { + t.Fatalf("create forward: %v", err) + } + + records, err := r.ListForwardsByTunnel(8) + if err != nil { + t.Fatalf("ListForwardsByTunnel: %v", err) + } + if len(records) != 1 { + t.Fatalf("expected 1 forward record, got %d", len(records)) + } + if records[0].MaxConn != 42 { + t.Fatalf("expected maxConn 42, got %d", records[0].MaxConn) + } +} + +func TestListActiveForwardsByUserTunnelIncludesMaxConn(t *testing.T) { + r, err := Open(":memory:") + if err != nil { + t.Fatalf("open repo: %v", err) + } + defer r.Close() + + now := time.Now().UnixMilli() + if err := r.DB().Create(&model.Forward{ + UserID: 2, + UserName: "user", + Name: "active-max-conn-forward", + TunnelID: 9, + RemoteAddr: "1.1.1.1:443", + Strategy: "fifo", + CreatedTime: now, + UpdatedTime: now, + Status: 1, + MaxConn: 55, + }).Error; err != nil { + t.Fatalf("create forward: %v", err) + } + + records, err := r.ListActiveForwardsByUserTunnel(2, 9) + if err != nil { + t.Fatalf("ListActiveForwardsByUserTunnel: %v", err) + } + if len(records) != 1 { + t.Fatalf("expected 1 forward record, got %d", len(records)) + } + if records[0].MaxConn != 55 { + t.Fatalf("expected maxConn 55, got %d", records[0].MaxConn) + } +} + func mustRepoLastInsertID(t *testing.T, r *Repository) int64 { t.Helper() var id int64 diff --git a/go-backend/tests/contract/max_conn_limit_contract_test.go b/go-backend/tests/contract/max_conn_limit_contract_test.go index 28a9fe2..c55e7c5 100644 --- a/go-backend/tests/contract/max_conn_limit_contract_test.go +++ b/go-backend/tests/contract/max_conn_limit_contract_test.go @@ -57,13 +57,14 @@ func TestMaxConnLimit(t *testing.T) { INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol) VALUES(?, 1, ?, 32001, 'round', 1, 'tls') `, tunnelID, nodeID).Error; err != nil { - t.Fatalf("insert chain_tunnel: %v", err) + t.Fatalf("insert chain_tunnel: %v", err) } if err := r.DB().Exec(` INSERT INTO user_tunnel(user_id, tunnel_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status) VALUES(1, ?, 10, 99999, 0, 0, 1, ?, 1) - `, tunnelID, now + 365*24*3600*1000).Error; err != nil { t.Fatalf("insert user_tunnel: %v", err) + `, tunnelID, now+365*24*3600*1000).Error; err != nil { + t.Fatalf("insert user_tunnel: %v", err) } var commandMu sync.Mutex @@ -91,11 +92,11 @@ func TestMaxConnLimit(t *testing.T) { waitNodeStatus(t, r, nodeID, 1) payload := map[string]interface{}{ - "name": "max-conn-forward", - "tunnelId": tunnelID, - "remoteAddr": "1.1.1.1:443", - "strategy": "fifo", - "maxConn": 42, + "name": "max-conn-forward", + "tunnelId": tunnelID, + "remoteAddr": "1.1.1.1:443", + "strategy": "fifo", + "maxConn": 42, "proxyProtocol": 2, } body, err := json.Marshal(payload) @@ -225,6 +226,145 @@ func TestMaxConnLimit(t *testing.T) { } } +func TestUserMaxConnUpdateResyncsExistingForwards(t *testing.T) { + secret := "contract-jwt-secret" + router, r := setupContractRouter(t, secret) + server := httptest.NewServer(router) + defer server.Close() + + adminToken, err := auth.GenerateToken(1, "admin_user", 0, secret) + if err != nil { + t.Fatalf("generate admin token: %v", err) + } + + now := time.Now().UnixMilli() + if err := r.DB().Exec(` + INSERT INTO user(id, user, pwd, role_id, exp_time, flow, in_flow, out_flow, flow_reset_time, num, max_conn, created_time, updated_time, status) + VALUES(2, 'limited_user', 'pwd', 1, ?, 99999, 0, 0, 1, 10, 0, ?, ?, 1) + `, now+365*24*3600*1000, now, now).Error; err != nil { + t.Fatalf("insert user: %v", err) + } + if err := r.DB().Exec(` + INSERT INTO tunnel(id, name, traffic_ratio, type, protocol, flow, created_time, updated_time, status, in_ip, inx) + VALUES(10, 'user-max-conn-tunnel', 1.0, 1, 'tls', 99999, ?, ?, 1, NULL, 0) + `, now, now).Error; err != nil { + t.Fatalf("insert tunnel: %v", err) + } + if err := r.DB().Exec(` + INSERT INTO node(id, name, secret, server_ip, server_ip_v4, server_ip_v6, port, interface_name, version, http, tls, socks, created_time, updated_time, status, tcp_listen_addr, udp_listen_addr, inx) + VALUES(20, 'user-max-conn-node', 'user-max-conn-secret', '10.21.0.1', '10.21.0.1', '', '32100-32110', '', 'v1', 1, 1, 1, ?, ?, 1, '[::]', '[::]', 0) + `, now, now).Error; err != nil { + t.Fatalf("insert node: %v", err) + } + if err := r.DB().Exec(` + INSERT INTO chain_tunnel(tunnel_id, chain_type, node_id, port, strategy, inx, protocol) + VALUES(10, 1, 20, 32101, 'round', 1, 'tls') + `).Error; err != nil { + t.Fatalf("insert chain_tunnel: %v", err) + } + if err := r.DB().Exec(` + INSERT INTO user_tunnel(id, user_id, tunnel_id, num, flow, in_flow, out_flow, flow_reset_time, exp_time, status) + VALUES(30, 2, 10, 10, 99999, 0, 0, 1, ?, 1) + `, now+365*24*3600*1000).Error; err != nil { + t.Fatalf("insert user_tunnel: %v", err) + } + if err := r.DB().Exec(` + INSERT INTO forward(id, user_id, user_name, name, tunnel_id, remote_addr, strategy, in_flow, out_flow, created_time, updated_time, status, inx, max_conn) + VALUES(40, 2, 'limited_user', 'user-max-conn-forward', 10, '1.1.1.1:443', 'fifo', 0, 0, ?, ?, 1, 0, 0) + `, now, now).Error; err != nil { + t.Fatalf("insert forward: %v", err) + } + if err := r.DB().Exec(` + INSERT INTO forward_port(forward_id, node_id, port, in_ip) + VALUES(40, 20, 32105, '') + `).Error; err != nil { + t.Fatalf("insert forward_port: %v", err) + } + + var commandMu sync.Mutex + receivedCommands := make([]string, 0) + var addCLimitersData json.RawMessage + var updateServiceData json.RawMessage + + stopNode := startMockSessionForMaxConn(t, server.URL, "user-max-conn-secret", func(cmdType string, data json.RawMessage) (bool, string) { + commandMu.Lock() + defer commandMu.Unlock() + receivedCommands = append(receivedCommands, cmdType) + if cmdType == "AddCLimiters" { + addCLimitersData = append([]byte(nil), data...) + } + if cmdType == "UpdateService" { + updateServiceData = append([]byte(nil), data...) + } + return false, "" + }) + defer stopNode() + + waitNodeStatus(t, r, 20, 1) + + payload := map[string]interface{}{ + "id": 2, + "user": "limited_user", + "flow": 99999, + "num": 10, + "expTime": now + 365*24*3600*1000, + "flowResetTime": 1, + "status": 1, + "maxConn": 37, + } + body, err := json.Marshal(payload) + if err != nil { + t.Fatalf("marshal payload: %v", err) + } + req := httptest.NewRequest(http.MethodPost, "/api/v1/user/update", bytes.NewReader(body)) + req.Header.Set("Authorization", adminToken) + req.Header.Set("Content-Type", "application/json") + res := httptest.NewRecorder() + router.ServeHTTP(res, req) + + var out response.R + if err := json.NewDecoder(res.Body).Decode(&out); err != nil { + t.Fatalf("decode response: %v", err) + } + if out.Code != 0 { + t.Fatalf("expected user update success, got code=%d msg=%s", out.Code, out.Msg) + } + + commandMu.Lock() + defer commandMu.Unlock() + if addCLimitersData == nil { + t.Fatalf("expected AddCLimiters after user maxConn update. Received: %v", receivedCommands) + } + if updateServiceData == nil { + t.Fatalf("expected UpdateService after user maxConn update. Received: %v", receivedCommands) + } + + var addData map[string]interface{} + if err := json.Unmarshal(addCLimitersData, &addData); err != nil { + t.Fatalf("unmarshal AddCLimiters data: %v", err) + } + if addData["name"] != "user_conn_limit_2" { + t.Fatalf("expected limiter name user_conn_limit_2, got %v", addData["name"]) + } + limits, ok := addData["limits"].([]interface{}) + if !ok || len(limits) != 1 || limits[0] != "$ 37" { + t.Fatalf("expected limits to contain '$ 37', got %v", addData["limits"]) + } + + var services []map[string]interface{} + if err := json.Unmarshal(updateServiceData, &services); err != nil { + t.Fatalf("unmarshal UpdateService data: %v", err) + } + if len(services) != 2 { + t.Fatalf("expected 2 services, got %d", len(services)) + } + for _, service := range services { + if service["climiter"] != "user_conn_limit_2" { + t.Fatalf("expected service climiter user_conn_limit_2, got %v", service["climiter"]) + } + } +} + func startMockSessionForMaxConn(t *testing.T, baseURL string, nodeSecret string, onCommand func(cmdType string, data json.RawMessage) (bool, string)) func() { t.Helper() diff --git a/go-gost/x/handler/forward/local/handler.go b/go-gost/x/handler/forward/local/handler.go index db49609..dfbb173 100644 --- a/go-gost/x/handler/forward/local/handler.go +++ b/go-gost/x/handler/forward/local/handler.go @@ -16,6 +16,7 @@ import ( "github.com/go-gost/core/recorder" ctxvalue "github.com/go-gost/x/ctx" xnet "github.com/go-gost/x/internal/net" + "github.com/go-gost/x/internal/net/proxyproto" "github.com/go-gost/x/internal/util/forwarder" "github.com/go-gost/x/internal/util/sniffing" tls_util "github.com/go-gost/x/internal/util/tls" @@ -252,6 +253,8 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand } defer cc.Close() + cc = proxyproto.WrapClientConn(h.md.proxyProtocol, conn.RemoteAddr(), conn.LocalAddr(), cc) + if err := xnet.Transport(conn, cc); err != nil { if marker := target.Marker(); marker != nil { marker.Mark() diff --git a/go-gost/x/handler/forward/local/metadata.go b/go-gost/x/handler/forward/local/metadata.go index 1307eea..530946a 100644 --- a/go-gost/x/handler/forward/local/metadata.go +++ b/go-gost/x/handler/forward/local/metadata.go @@ -14,6 +14,7 @@ import ( type metadata struct { readTimeout time.Duration + proxyProtocol int httpKeepalive bool sniffing bool @@ -38,6 +39,7 @@ func (h *forwardHandler) parseMetadata(md mdata.Metadata) (err error) { if h.md.readTimeout <= 0 { h.md.readTimeout = 15 * time.Second } + h.md.proxyProtocol = mdutil.GetInt(md, "proxyProtocol") h.md.httpKeepalive = mdutil.GetBool(md, "http.keepalive") diff --git a/go-gost/x/handler/forward/local/proxy_protocol_test.go b/go-gost/x/handler/forward/local/proxy_protocol_test.go new file mode 100644 index 0000000..4fe25c2 --- /dev/null +++ b/go-gost/x/handler/forward/local/proxy_protocol_test.go @@ -0,0 +1,111 @@ +package local + +import ( + "bufio" + "context" + "net" + "testing" + "time" + + "github.com/go-gost/core/chain" + "github.com/go-gost/core/handler" + "github.com/go-gost/core/hop" + xlogger "github.com/go-gost/x/logger" + xmd "github.com/go-gost/x/metadata" + proxyproto "github.com/pires/go-proxyproto" +) + +type proxyProtocolTestHop struct { + node *chain.Node +} + +func (h proxyProtocolTestHop) Select(context.Context, ...hop.SelectOption) *chain.Node { + return h.node +} + +func (h proxyProtocolTestHop) Nodes() []*chain.Node { + return []*chain.Node{h.node} +} + +type proxyProtocolTestRouter struct{} + +func (r proxyProtocolTestRouter) Options() *chain.RouterOptions { + return &chain.RouterOptions{} +} + +func (r proxyProtocolTestRouter) Dial(ctx context.Context, network, address string) (net.Conn, error) { + var d net.Dialer + return d.DialContext(ctx, network, address) +} + +func (r proxyProtocolTestRouter) Bind(context.Context, string, string, ...chain.BindOption) (net.Listener, error) { + return nil, net.ErrClosed +} + +func TestLocalForwardHandlerSendsProxyProtocolToTarget(t *testing.T) { + targetListener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen target: %v", err) + } + defer targetListener.Close() + + entryListener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen entry: %v", err) + } + defer entryListener.Close() + + h := NewHandler( + handler.RouterOption(proxyProtocolTestRouter{}), + handler.LoggerOption(xlogger.Nop()), + ) + forwarder := h.(handler.Forwarder) + forwarder.Forward(proxyProtocolTestHop{node: chain.NewNode("target", targetListener.Addr().String())}) + if err := h.Init(xmd.NewMetadata(map[string]any{"proxyProtocol": 2})); err != nil { + t.Fatalf("init handler: %v", err) + } + + handleErr := make(chan error, 1) + acceptErr := make(chan error, 1) + go func() { + serverConn, err := entryListener.Accept() + if err != nil { + acceptErr <- err + return + } + handleErr <- h.Handle(context.Background(), serverConn) + }() + + clientConn, err := net.Dial("tcp", entryListener.Addr().String()) + if err != nil { + t.Fatalf("dial entry: %v", err) + } + defer clientConn.Close() + + targetConn, err := targetListener.Accept() + if err != nil { + t.Fatalf("accept target: %v", err) + } + defer targetConn.Close() + if err := targetConn.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil { + t.Fatalf("set target deadline: %v", err) + } + + header, err := proxyproto.Read(bufio.NewReader(targetConn)) + if err != nil { + t.Fatalf("read proxy protocol header: %v", err) + } + if header.Version != 2 { + t.Fatalf("expected proxy protocol v2, got v%d", header.Version) + } + + _ = clientConn.Close() + _ = targetConn.Close() + select { + case err := <-acceptErr: + t.Fatalf("accept entry: %v", err) + case <-handleErr: + case <-time.After(2 * time.Second): + t.Fatal("handler did not return after closing connections") + } +} diff --git a/vite-frontend/src/pages/forward.tsx b/vite-frontend/src/pages/forward.tsx index 06b6961..2e86b78 100644 --- a/vite-frontend/src/pages/forward.tsx +++ b/vite-frontend/src/pages/forward.tsx @@ -4893,10 +4893,10 @@ export default function ForwardPage() { >