From 81db46b725abce54739ecde4070850aa40fe2f98 Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Tue, 2 Jun 2026 23:51:55 +0800 Subject: [PATCH] refactor(handler/tunnel): code review fixes and entrypoint subpackage extraction - Fix dead-code branch in bind.go host assignment (always use endpoint hash) - Return descriptive error on bypass match in connect.go (was masking as success) - Update bypass test in connect_test.go for new error behavior - Extract entrypoint subpackage from monolithic entrypoint.go (6 files) - Fix observeStats event-loss bug (break -> fallthrough on retry success) - Add 47 unit tests across handler, connect, bind, metadata packages - Add architecture doc comments to all key files - Build and vet clean, 173 tests pass with -race --- handler/tunnel/bind.go | 10 +- handler/tunnel/bind_test.go | 441 +++++++++ handler/tunnel/connect.go | 16 +- handler/tunnel/connect_test.go | 418 +++++++++ handler/tunnel/dialer.go | 34 +- handler/tunnel/dialer_test.go | 100 ++- handler/tunnel/entrypoint.go | 204 ----- handler/tunnel/entrypoint/config.go | 87 ++ handler/tunnel/entrypoint/entrypoint.go | 316 +++++++ handler/tunnel/entrypoint/entrypoint_test.go | 846 ++++++++++++++++++ handler/tunnel/{ => entrypoint}/ephttp.go | 32 +- handler/tunnel/entrypoint/eplistener.go | 105 +++ handler/tunnel/{ => entrypoint}/eprelay.go | 21 +- handler/tunnel/{ => entrypoint}/eptls.go | 4 +- .../tunnel/{ => entrypoint}/epwebsocket.go | 25 +- handler/tunnel/entrypointsvc.go | 90 +- handler/tunnel/eplistener.go | 71 -- handler/tunnel/handler.go | 69 +- handler/tunnel/handler_test.go | 135 ++- handler/tunnel/helpers_test.go | 9 + handler/tunnel/id.go | 4 +- handler/tunnel/id_test.go | 24 +- handler/tunnel/metadata.go | 36 +- handler/tunnel/metadata_test.go | 499 +++++++++++ handler/tunnel/tunnel.go | 14 +- handler/tunnel/tunnel_test.go | 2 +- 26 files changed, 3145 insertions(+), 467 deletions(-) create mode 100644 handler/tunnel/bind_test.go create mode 100644 handler/tunnel/connect_test.go delete mode 100644 handler/tunnel/entrypoint.go create mode 100644 handler/tunnel/entrypoint/config.go create mode 100644 handler/tunnel/entrypoint/entrypoint.go create mode 100644 handler/tunnel/entrypoint/entrypoint_test.go rename handler/tunnel/{ => entrypoint}/ephttp.go (85%) create mode 100644 handler/tunnel/entrypoint/eplistener.go rename handler/tunnel/{ => entrypoint}/eprelay.go (89%) rename handler/tunnel/{ => entrypoint}/eptls.go (96%) rename handler/tunnel/{ => entrypoint}/epwebsocket.go (70%) delete mode 100644 handler/tunnel/eplistener.go create mode 100644 handler/tunnel/metadata_test.go diff --git a/handler/tunnel/bind.go b/handler/tunnel/bind.go index 9449c0d2..78340415 100644 --- a/handler/tunnel/bind.go +++ b/handler/tunnel/bind.go @@ -55,11 +55,11 @@ func (h *tunnelHandler) handleBind(ctx context.Context, conn net.Conn, network, endpoint := hex.EncodeToString(v[:8]) host, port, _ := net.SplitHostPort(address) - if host == "" || h.md.ingress == nil { - host = endpoint - } else if host != endpoint { - host = endpoint - } + // Always use the endpoint hash as the host — this provides a stable, + // deterministic ingress key regardless of what the internal client sends. + // The original host is ignored; the endpoint hash routes consistently + // across reconnects and multi-node deployments. + host = endpoint addr := net.JoinHostPort(host, port) af := &relay.AddrFeature{} diff --git a/handler/tunnel/bind_test.go b/handler/tunnel/bind_test.go new file mode 100644 index 00000000..f6c8c2fa --- /dev/null +++ b/handler/tunnel/bind_test.go @@ -0,0 +1,441 @@ +package tunnel + +import ( + "context" + "net" + "testing" + "time" + + "github.com/go-gost/core/handler" + "github.com/go-gost/core/ingress" + "github.com/go-gost/core/sd" + "github.com/go-gost/relay" + "github.com/go-gost/x/internal/util/mux" +) + +// TestHandleBind_ResponseWritten tests that handleBind writes a correct relay +// response frame to the connection and adds the connector to the pool. +func TestHandleBind_ResponseWritten(t *testing.T) { + tid := newTestTunnelID(t) + + h := &tunnelHandler{ + options: handler.Options{ + Logger: testLogger(), + }, + md: metadata{ + muxCfg: &mux.Config{Version: 2}, + }, + id: "node1", + pool: NewConnectorPool("node1"), + log: testLogger(), + } + defer h.pool.Close() + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + errCh := make(chan error, 1) + go func() { + errCh <- h.handleBind(context.Background(), server, "tcp", "0.0.0.0:0", tid, testLogger()) + }() + + // Read the relay response from the client side of the pipe. + resp := &relay.Response{} + _, err := resp.ReadFrom(client) + if err != nil { + t.Fatalf("read response: %v", err) + } + if resp.Status != relay.StatusOK { + t.Errorf("expected StatusOK (%d), got %d", relay.StatusOK, resp.Status) + } + + // The response should contain AddrFeature and TunnelFeature. + var addrFound bool + var tunnelFound bool + for _, f := range resp.Features { + switch f.Type() { + case relay.FeatureAddr: + addrFound = true + case relay.FeatureTunnel: + tunnelFound = true + tf := f.(*relay.TunnelFeature) + if tf.ID == (relay.TunnelID{}) { + t.Error("expected non-zero connector ID") + } + } + } + if !addrFound { + t.Error("expected AddrFeature in response") + } + if !tunnelFound { + t.Error("expected TunnelFeature in response") + } + + // handleBind should complete (with or without error depending on mux handshake timing). + select { + case err := <-errCh: + if err != nil { + t.Logf("handleBind returned (expected): %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("handleBind did not complete") + } +} + +// TestHandleBind_WithIngress tests that handleBind sets ingress rules. +func TestHandleBind_WithIngress(t *testing.T) { + tid := newTestTunnelID(t) + + ing := &fakeIngress{} + h := &tunnelHandler{ + options: handler.Options{ + Logger: testLogger(), + }, + md: metadata{ + muxCfg: &mux.Config{Version: 2}, + ingress: ing, + }, + id: "node1", + pool: NewConnectorPool("node1"), + log: testLogger(), + } + defer h.pool.Close() + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + errCh := make(chan error, 1) + go func() { + errCh <- h.handleBind(context.Background(), server, "tcp", "0.0.0.0:0", tid, testLogger()) + }() + + _, _ = (&relay.Response{}).ReadFrom(client) + client.Close() + + select { + case err := <-errCh: + t.Logf("handleBind returned: %v", err) + case <-time.After(2 * time.Second): + t.Fatal("handleBind did not complete") + } + + // Ingress rules should have been set. + if ing.rule == nil { + t.Error("expected ingress rule to be set") + } else if ing.rule.Endpoint != tid.String() { + t.Errorf("expected endpoint %s, got %s", tid.String(), ing.rule.Endpoint) + } +} + +// TestHandleBind_WithSD tests that handleBind registers with SD. +func TestHandleBind_WithSD(t *testing.T) { + tid := newTestTunnelID(t) + + sdCalled := make(chan *sd.Service, 1) + fakeSD := &fakeSD{ + registerFunc: func(ctx context.Context, service *sd.Service) error { + sdCalled <- service + return nil + }, + } + + h := &tunnelHandler{ + options: handler.Options{ + Logger: testLogger(), + }, + md: metadata{ + muxCfg: &mux.Config{Version: 2}, + sd: fakeSD, + entryPoint: "example.com:8080", + }, + id: "node1", + pool: NewConnectorPool("node1"), + log: testLogger(), + } + defer h.pool.Close() + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + errCh := make(chan error, 1) + go func() { + errCh <- h.handleBind(context.Background(), server, "tcp", "0.0.0.0:0", tid, testLogger()) + }() + + // Read the response from client side. + _, _ = (&relay.Response{}).ReadFrom(client) + + // Wait for SD registration. + select { + case svc := <-sdCalled: + if svc.Name != tid.String() { + t.Errorf("expected service name %s, got %s", tid.String(), svc.Name) + } + if svc.Node != "node1" { + t.Errorf("expected node node1, got %s", svc.Node) + } + if svc.Network != "tcp" { + t.Errorf("expected network tcp, got %s", svc.Network) + } + if svc.Address != "example.com:8080" { + t.Errorf("expected address example.com:8080, got %s", svc.Address) + } + case <-time.After(2 * time.Second): + t.Fatal("SD.Register was not called") + } + + client.Close() + + select { + case err := <-errCh: + t.Logf("handleBind returned: %v", err) + case <-time.After(2 * time.Second): + t.Fatal("handleBind did not complete") + } +} + +// TestHandleBind_UDPConnector tests that UDP tunnel requests create UDP connector IDs. +func TestHandleBind_UDPConnector(t *testing.T) { + tid := newTestTunnelID(t) + tid = tid.SetWeight(10) // set weight to verify it's copied + + h := &tunnelHandler{ + options: handler.Options{ + Logger: testLogger(), + }, + md: metadata{ + muxCfg: &mux.Config{Version: 2}, + }, + id: "node1", + pool: NewConnectorPool("node1"), + log: testLogger(), + } + defer h.pool.Close() + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + errCh := make(chan error, 1) + go func() { + errCh <- h.handleBind(context.Background(), server, "udp", "0.0.0.0:0", tid, testLogger()) + }() + + resp := &relay.Response{} + _, err := resp.ReadFrom(client) + if err != nil { + t.Fatalf("read response: %v", err) + } + + // Verify connector ID is UDP and weight is copied. + var connectorID relay.ConnectorID + for _, f := range resp.Features { + if f.Type() == relay.FeatureTunnel { + connectorID = f.(*relay.TunnelFeature).ID + } + } + if connectorID == (relay.ConnectorID{}) { + t.Fatal("expected non-zero connector ID") + } + if !connectorID.IsUDP() { + t.Error("expected UDP connector for UDP tunnel request") + } + if connectorID.Weight() != 10 { + t.Errorf("expected weight 10 (from tunnelID), got %d", connectorID.Weight()) + } + + client.Close() + select { + case err := <-errCh: + t.Logf("handleBind returned: %v", err) + case <-time.After(2 * time.Second): + t.Fatal("handleBind did not complete") + } +} + +// TestConnector_waitClose tests that waitClose properly handles session +// termination and cleans up. +func TestConnector_waitClose(t *testing.T) { + cid := newTestConnectorID(t, false, 1) + tid := newTestTunnelID(t) + + // Create a real mux session for testing waitClose behavior. + s, clientEnd := newTestSession(t) + c := newTestConnector(cid, tid, "node1", s, &ConnectorOptions{}) + + // waitClose is started in NewConnector which we bypassed via newTestConnector. + // Start it manually. + done := make(chan struct{}) + go func() { + c.waitClose() + close(done) + }() + + // Close the client end of the pipe — the mux session should detect this + // and Accept in waitClose should fail, causing waitClose to exit. + clientEnd.Close() + + select { + case <-done: + // waitClose exited as expected + case <-time.After(2 * time.Second): + t.Fatal("waitClose did not exit after session close") + } +} + +// TestConnector_waitClose_DeregisterSD tests that waitClose deregisters from +// SD when configured. +func TestConnector_waitClose_DeregisterSD(t *testing.T) { + deregisterCalled := make(chan struct{}, 1) + fakeSD := &fakeSD{ + deregisterFunc: func(ctx context.Context, service *sd.Service) error { + deregisterCalled <- struct{}{} + return nil + }, + } + + cid := newTestConnectorID(t, false, 1) + tid := newTestTunnelID(t) + s, clientEnd := newTestSession(t) + c := newTestConnector(cid, tid, "node1", s, &ConnectorOptions{ + sd: fakeSD, + }) + + done := make(chan struct{}) + go func() { + c.waitClose() + close(done) + }() + + // Close client end to trigger session failure. + clientEnd.Close() + + select { + case <-deregisterCalled: + // Deregister was called + case <-time.After(2 * time.Second): + t.Fatal("SD.Deregister was not called") + } + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("waitClose did not exit") + } +} + +// TestConnectWithBindEndToEnd tests the full round-trip: handleBind registers +// a connector, then handleConnect dials through it. This verifies the connector +// was properly added to the pool. +func TestConnectWithBindEndToEnd(t *testing.T) { + tid := newTestTunnelID(t) + + h := &tunnelHandler{ + options: handler.Options{ + Logger: testLogger(), + }, + md: metadata{ + muxCfg: &mux.Config{Version: 2}, + ingress: &fakeIngress{ + rule: &ingress.Rule{ + Hostname: "example.com", + Endpoint: tid.String(), + }, + }, + entryPointID: tid, + }, + id: "node1", + pool: NewConnectorPool("node1"), + log: testLogger(), + } + defer h.pool.Close() + + // Simulate a CmdBind: create a pipe, run handleBind on server end. + bindClient, bindServer := net.Pipe() + defer bindClient.Close() + defer bindServer.Close() + + bindResult := make(chan error, 1) + go func() { + bindResult <- h.handleBind(context.Background(), bindServer, "tcp", "0.0.0.0:0", tid, testLogger()) + }() + + // Read the bind response from client side. + _, err := (&relay.Response{}).ReadFrom(bindClient) + if err != nil { + t.Fatalf("read bind response: %v", err) + } + + // Now the connector should be registered in the pool. + // Try to connect through it. + req := &relay.Request{ + Version: relay.Version1, + Cmd: relay.CmdConnect, + } + req.Features = append(req.Features, &relay.TunnelFeature{ID: tid}) + req.Features = append(req.Features, &relay.AddrFeature{Host: "10.0.0.1", Port: 12345}) + req.Features = append(req.Features, &relay.AddrFeature{Host: "example.com", Port: 80}) + + conn := &fakeConn{} + err = h.handleConnect(context.Background(), req, conn, "tcp", "10.0.0.1:12345", "example.com:80", tid, testLogger()) + + if err != nil { + t.Logf("handleConnect returned (expected with pipe): %v", err) + } + + // Clean up bind. + bindClient.Close() + select { + case <-bindResult: + case <-time.After(2 * time.Second): + t.Fatal("handleBind did not complete") + } +} + +// TestHandleBind_WithHostEndpoint tests handleBind when the bind address has +// a non-empty host — the endpoint-based ingress rule should be set. +func TestHandleBind_WithHostEndpoint(t *testing.T) { + tid := newTestTunnelID(t) + + ing := &fakeIngress{} + h := &tunnelHandler{ + options: handler.Options{ + Logger: testLogger(), + }, + md: metadata{ + muxCfg: &mux.Config{Version: 2}, + ingress: ing, + }, + id: "node1", + pool: NewConnectorPool("node1"), + log: testLogger(), + } + defer h.pool.Close() + + client, server := net.Pipe() + defer client.Close() + defer server.Close() + + errCh := make(chan error, 1) + go func() { + errCh <- h.handleBind(context.Background(), server, "tcp", "myapp.example.com:8080", tid, testLogger()) + }() + + _, _ = (&relay.Response{}).ReadFrom(client) + client.Close() + + select { + case err := <-errCh: + t.Logf("handleBind returned: %v", err) + case <-time.After(2 * time.Second): + t.Fatal("handleBind did not complete") + } + + // Should have set at least the endpoint-based ingress rule. + if ing.rule == nil { + t.Error("expected ingress rule to be set") + } +} \ No newline at end of file diff --git a/handler/tunnel/connect.go b/handler/tunnel/connect.go index 4c9a6dd8..7aee55c7 100644 --- a/handler/tunnel/connect.go +++ b/handler/tunnel/connect.go @@ -56,7 +56,7 @@ func (h *tunnelHandler) handleConnect(ctx context.Context, req *relay.Request, c log.Debug("bypass: ", dstAddr) resp.Status = relay.StatusForbidden _, err := resp.WriteTo(conn) - return err + return fmt.Errorf("bypass blocked %s: %w", dstAddr, err) } host, _, _ := net.SplitHostPort(dstAddr) @@ -64,7 +64,7 @@ func (h *tunnelHandler) handleConnect(ctx context.Context, req *relay.Request, c var tid relay.TunnelID if ing := h.md.ingress; ing != nil && host != "" { if rule := ing.GetRule(ctx, host, ingress.WithService(h.options.Service)); rule != nil { - tid = parseTunnelID(rule.Endpoint) + tid = ParseTunnelID(rule.Endpoint) } } @@ -100,12 +100,12 @@ func (h *tunnelHandler) handleConnect(ctx context.Context, req *relay.Request, c } d := Dialer{ - node: h.id, - pool: h.pool, - sd: h.md.sd, - retry: 3, - timeout: 15 * time.Second, - log: log, + Node: h.id, + Pool: h.pool, + SD: h.md.sd, + Retry: 3, + Timeout: 15 * time.Second, + Log: log, } cc, node, cid, err := d.Dial(ctx, network, tid.String()) if err != nil { diff --git a/handler/tunnel/connect_test.go b/handler/tunnel/connect_test.go new file mode 100644 index 00000000..2f6f073a --- /dev/null +++ b/handler/tunnel/connect_test.go @@ -0,0 +1,418 @@ +package tunnel + +import ( + "bytes" + "context" + "testing" + + "github.com/go-gost/core/auth" + "github.com/go-gost/core/bypass" + "github.com/go-gost/core/handler" + "github.com/go-gost/core/ingress" + "github.com/go-gost/relay" +) + +// fakeBypass implements bypass.Bypass for testing. +type fakeBypass struct { + matched bool +} + +func (b *fakeBypass) IsWhitelist() bool { return false } +func (b *fakeBypass) Contains(ctx context.Context, network, addr string, opts ...bypass.Option) bool { + return b.matched +} + +// fakeIngress implements ingress.Ingress for testing. +type fakeIngress struct { + rule *ingress.Rule +} + +func (i *fakeIngress) GetRule(ctx context.Context, host string, opts ...ingress.Option) *ingress.Rule { + return i.rule +} + +func (i *fakeIngress) SetRule(ctx context.Context, rule *ingress.Rule, opts ...ingress.Option) bool { + i.rule = rule + return true +} + +// newTestTunnelID is defined in tunnel_test.go + +func TestHandleConnect_Bypass(t *testing.T) { + req := &relay.Request{ + Version: relay.Version1, + Cmd: relay.CmdConnect, + } + tid := newTestTunnelID(t) + req.Features = append(req.Features, &relay.TunnelFeature{ID: tid}) + req.Features = append(req.Features, &relay.AddrFeature{Host: "example.com", Port: 80}) + + h := &tunnelHandler{ + options: handler.Options{ + Bypass: &fakeBypass{matched: true}, + Logger: testLogger(), + }, + md: metadata{ + entryPointID: relay.TunnelID{}, // zero — not an entrypoint visitor + }, + id: "node1", + pool: NewConnectorPool("node1"), + } + + conn := &fakeConn{} + err := h.handleConnect(context.Background(), req, conn, "tcp", "10.0.0.1:12345", "example.com:80", tid, testLogger()) + + if err == nil { + t.Fatal("expected error for bypass") + } + // Verify relay response was written with StatusForbidden + resp := relay.Response{} + _, err = resp.ReadFrom(bytes.NewReader(conn.writeBuf)) + if err != nil { + t.Fatalf("read relay response: %v", err) + } + if resp.Status != relay.StatusForbidden { + t.Errorf("expected StatusForbidden (%d), got %d", relay.StatusForbidden, resp.Status) + } +} + +func TestHandleConnect_EntrypointNoRoute(t *testing.T) { + req := &relay.Request{ + Version: relay.Version1, + Cmd: relay.CmdConnect, + } + entrypointID := newTestTunnelID(t) + req.Features = append(req.Features, &relay.TunnelFeature{ID: entrypointID}) + req.Features = append(req.Features, &relay.AddrFeature{Host: "example.com", Port: 80}) + + h := &tunnelHandler{ + options: handler.Options{ + Logger: testLogger(), + }, + md: metadata{ + entryPointID: entrypointID, + ingress: &fakeIngress{rule: nil}, // no rule + }, + id: "node1", + pool: NewConnectorPool("node1"), + } + + conn := &fakeConn{} + err := h.handleConnect(context.Background(), req, conn, "tcp", "10.0.0.1:12345", "example.com:80", entrypointID, testLogger()) + + if err == nil { + t.Fatal("expected error for no route") + } + resp := relay.Response{} + _, err = resp.ReadFrom(bytes.NewReader(conn.writeBuf)) + if err != nil { + t.Fatalf("read relay response: %v", err) + } + if resp.Status != relay.StatusNetworkUnreachable { + t.Errorf("expected StatusNetworkUnreachable (%d), got %d", relay.StatusNetworkUnreachable, resp.Status) + } +} + +func TestHandleConnect_EntrypointPrivateTunnel(t *testing.T) { + req := &relay.Request{ + Version: relay.Version1, + Cmd: relay.CmdConnect, + } + entrypointID := newTestTunnelID(t) + req.Features = append(req.Features, &relay.TunnelFeature{ID: entrypointID}) + req.Features = append(req.Features, &relay.AddrFeature{Host: "example.com", Port: 80}) + + // Private tunnel endpoint — ingress returns a $-prefixed tunnel ID + + h := &tunnelHandler{ + options: handler.Options{ + Logger: testLogger(), + }, + md: metadata{ + entryPointID: entrypointID, + ingress: &fakeIngress{ + rule: &ingress.Rule{ + Hostname: "example.com", + Endpoint: "$4e1b0d1a-8e3f-4c7a-9b2c-5d6e7f8a9b0c", + }, + }, + }, + id: "node1", + pool: NewConnectorPool("node1"), + } + + conn := &fakeConn{} + err := h.handleConnect(context.Background(), req, conn, "tcp", "10.0.0.1:12345", "example.com:80", entrypointID, testLogger()) + + if err == nil { + t.Fatal("expected error for private tunnel") + } + resp := relay.Response{} + _, err = resp.ReadFrom(bytes.NewReader(conn.writeBuf)) + if err != nil { + t.Fatalf("read relay response: %v", err) + } + if resp.Status != relay.StatusHostUnreachable { + t.Errorf("expected StatusHostUnreachable (%d), got %d", relay.StatusHostUnreachable, resp.Status) + } +} + +func TestHandleConnect_DirectTunnelMismatch(t *testing.T) { + tid := newTestTunnelID(t) + + req := &relay.Request{ + Version: relay.Version1, + Cmd: relay.CmdConnect, + } + req.Features = append(req.Features, &relay.TunnelFeature{ID: tid}) + req.Features = append(req.Features, &relay.AddrFeature{Host: "example.com", Port: 80}) + + h := &tunnelHandler{ + options: handler.Options{ + Logger: testLogger(), + }, + md: metadata{ + directTunnel: false, // not direct tunnel + }, + id: "node1", + pool: NewConnectorPool("node1"), + } + + conn := &fakeConn{} + // Not entrypoint, not direct tunnel, and tid doesn't match (tid != tid — same value) + // We need a different TID scenario: since there's no ingress and no direct tunnel, + // tid.Equal(tunnelID) is true (both are `tid`), so this path goes to Dialer. + // Instead test the case where the tunnelID is NOT the entrypointID and NOT direct + // and ingress returns a different tid. + err := h.handleConnect(context.Background(), req, conn, "tcp", "10.0.0.1:12345", "example.com:80", tid, testLogger()) + + // With no ingress, tid.Equal(tunnelID) is true, so this tries to dial + // and fails with ErrTunnelNotAvailable + if err == nil { + t.Fatal("expected error") + } +} + +func TestHandleConnect_DirectTunnelEnabled(t *testing.T) { + tid := newTestTunnelID(t) + + req := &relay.Request{ + Version: relay.Version1, + Cmd: relay.CmdConnect, + } + req.Features = append(req.Features, &relay.TunnelFeature{ID: tid}) + req.Features = append(req.Features, &relay.AddrFeature{Host: "example.com", Port: 80}) + + h := &tunnelHandler{ + options: handler.Options{ + Logger: testLogger(), + }, + md: metadata{ + directTunnel: true, // direct tunnel enabled — uses tunnelID directly + }, + id: "node1", + pool: NewConnectorPool("node1"), + } + + conn := &fakeConn{} + err := h.handleConnect(context.Background(), req, conn, "tcp", "10.0.0.1:12345", "example.com:80", tid, testLogger()) + + // Direct tunnel, tid matches — tries to dial, fails with ErrTunnelNotAvailable + if err == nil { + t.Fatal("expected error") + } +} + +func TestHandleConnect_DialerError(t *testing.T) { + tid := newTestTunnelID(t) + + req := &relay.Request{ + Version: relay.Version1, + Cmd: relay.CmdConnect, + } + req.Features = append(req.Features, &relay.TunnelFeature{ID: tid}) + req.Features = append(req.Features, &relay.AddrFeature{Host: "example.com", Port: 80}) + + h := &tunnelHandler{ + options: handler.Options{ + Logger: testLogger(), + }, + md: metadata{ + entryPointID: tid, + ingress: &fakeIngress{ + rule: &ingress.Rule{ + Hostname: "example.com", + Endpoint: tid.String(), + }, + }, + }, + id: "node1", + pool: NewConnectorPool("node1"), + log: testLogger(), + } + + conn := &fakeConn{} + err := h.handleConnect(context.Background(), req, conn, "tcp", "10.0.0.1:12345", "example.com:80", tid, testLogger()) + + if err == nil { + t.Fatal("expected error for dialer failure") + } + // Should write StatusServiceUnavailable + resp := relay.Response{} + _, err = resp.ReadFrom(bytes.NewReader(conn.writeBuf)) + if err != nil { + t.Fatalf("read relay response: %v", err) + } + if resp.Status != relay.StatusServiceUnavailable { + t.Errorf("expected StatusServiceUnavailable (%d), got %d", relay.StatusServiceUnavailable, resp.Status) + } +} + +func TestHandleConnect_AuthFailure(t *testing.T) { + tid := newTestTunnelID(t) + + req := &relay.Request{ + Version: relay.Version1, + Cmd: relay.CmdConnect, + } + req.Features = append(req.Features, &relay.TunnelFeature{ID: tid}) + req.Features = append(req.Features, &relay.AddrFeature{Host: "10.0.0.1", Port: 12345}) + req.Features = append(req.Features, &relay.AddrFeature{Host: "example.com", Port: 80}) + req.Features = append(req.Features, &relay.UserAuthFeature{Username: "user", Password: "wrong"}) + + auther := &fakeAuther{ok: false} + h := &tunnelHandler{ + id: "node1", + pool: NewConnectorPool("node1"), + options: handler.Options{ + Auther: auther, + Logger: testLogger(), + }, + md: metadata{ + directTunnel: true, + }, + log: testLogger(), + } + + conn := &fakeConn{buf: func() []byte { + var buf bytes.Buffer + _, _ = req.WriteTo(&buf) + return buf.Bytes() + }()} + t.Logf("request bytes: %d", len(conn.buf)) + err := h.Handle(context.Background(), conn) + if err == nil { + t.Fatal("expected auth error") + } +} + +func TestHandleConnect_EntrypointWithIngress_Success(t *testing.T) { + tid := newTestTunnelID(t) + + req := &relay.Request{ + Version: relay.Version1, + Cmd: relay.CmdConnect, + } + req.Features = append(req.Features, &relay.TunnelFeature{ID: tid}) + req.Features = append(req.Features, &relay.AddrFeature{Host: "10.0.0.1", Port: 12345}) + req.Features = append(req.Features, &relay.AddrFeature{Host: "example.com", Port: 80}) + + h := &tunnelHandler{ + options: handler.Options{ + Logger: testLogger(), + }, + md: metadata{ + entryPointID: tid, + ingress: &fakeIngress{ + rule: &ingress.Rule{ + Hostname: "example.com", + Endpoint: tid.String(), + }, + }, + }, + id: "node1", + pool: NewConnectorPool("node1"), + log: testLogger(), + } + + // Call handleConnect directly to avoid relay request parsing in Handle(). + conn := &fakeConn{} + err := h.handleConnect(context.Background(), req, conn, "tcp", "10.0.0.1:12345", "example.com:80", tid, testLogger()) + + if err == nil { + t.Fatal("expected error (no connector in pool)") + } + // Should have written a response before the error + if len(conn.writeBuf) == 0 { + t.Error("expected relay response to be written") + } else { + resp := relay.Response{} + _, rerr := resp.ReadFrom(bytes.NewReader(conn.writeBuf)) + if rerr != nil { + t.Fatalf("read response: %v", rerr) + } + if resp.Status != relay.StatusServiceUnavailable { + t.Errorf("expected StatusServiceUnavailable (%d), got %d", relay.StatusServiceUnavailable, resp.Status) + } + } +} + +func TestHandleConnect_NotEntrypointNotDirect_IngressRule(t *testing.T) { + tid := newTestTunnelID(t) + + req := &relay.Request{ + Version: relay.Version1, + Cmd: relay.CmdConnect, + } + req.Features = append(req.Features, &relay.TunnelFeature{ID: tid}) + req.Features = append(req.Features, &relay.AddrFeature{Host: "10.0.0.1", Port: 12345}) + req.Features = append(req.Features, &relay.AddrFeature{Host: "example.com", Port: 80}) + + // ingress returns a DIFFERENT tid than the request's tunnelID + otherTID := ParseTunnelID("7f2c3d4e-5a6b-7c8d-9e0f-1a2b3c4d5e6f") + + h := &tunnelHandler{ + options: handler.Options{ + Logger: testLogger(), + }, + md: metadata{ + entryPointID: relay.TunnelID{}, // not entrypoint + directTunnel: false, + ingress: &fakeIngress{ + rule: &ingress.Rule{ + Hostname: "example.com", + Endpoint: otherTID.String(), + }, + }, + }, + id: "node1", + pool: NewConnectorPool("node1"), + } + + conn := &fakeConn{} + err := h.handleConnect(context.Background(), req, conn, "tcp", "10.0.0.1:12345", "example.com:80", tid, testLogger()) + + if err == nil { + t.Fatal("expected error (ingress tid != request tid)") + } + resp := relay.Response{} + _, err = resp.ReadFrom(bytes.NewReader(conn.writeBuf)) + if err != nil { + t.Fatalf("read relay response: %v", err) + } + if resp.Status != relay.StatusHostUnreachable { + t.Errorf("expected StatusHostUnreachable (%d), got %d", relay.StatusHostUnreachable, resp.Status) + } +} + +// fakeAuther implements auth.Authenticator for testing. +type fakeAuther struct { + ok bool +} + +func (a *fakeAuther) Authenticate(ctx context.Context, user, pass string, opts ...auth.Option) (string, bool) { + if a.ok { + return "client1", true + } + return "", false +} \ No newline at end of file diff --git a/handler/tunnel/dialer.go b/handler/tunnel/dialer.go index b0a4f8e4..ba561e16 100644 --- a/handler/tunnel/dialer.go +++ b/handler/tunnel/dialer.go @@ -17,29 +17,29 @@ import ( // connector is dead, all retries fail until Tunnel.clean() removes it // (up to TTL, default 15s). // 2. SD fallback: if pool returns nil AND sd is configured, query service -// discovery for remote nodes. Filter out self (d.node) and mismatched +// discovery for remote nodes. Filter out self (d.Node) and mismatched // networks. Establish a raw TCP connection to the remote address, // bypassing the mux layer entirely. // // The returned node and connector ID identify which node/hop the stream // is connected to — used by callers to decide the relay protocol framing. type Dialer struct { - node string - pool *ConnectorPool - sd sd.SD - retry int - timeout time.Duration - log logger.Logger + Node string + Pool *ConnectorPool + SD sd.SD + Retry int + Timeout time.Duration + Log logger.Logger } func (d *Dialer) Dial(ctx context.Context, network string, tid string) (conn net.Conn, node string, cid string, err error) { - retry := d.retry + retry := d.Retry if retry <= 0 { retry = 1 } for i := 0; i < retry; i++ { - c := d.pool.Get(network, tid) + c := d.Pool.Get(network, tid) if c == nil { err = nil // clear stale err so SD fallback is not masked break @@ -47,10 +47,10 @@ func (d *Dialer) Dial(ctx context.Context, network string, tid string) (conn net conn, err = c.GetConn() if err != nil { - d.log.Error(err) + d.Log.Error(err) continue } - node = d.node + node = d.Node cid = c.id.String() break @@ -59,20 +59,20 @@ func (d *Dialer) Dial(ctx context.Context, network string, tid string) (conn net return } - if d.sd == nil { + if d.SD == nil { err = ErrTunnelNotAvailable return } - ss, err := d.sd.Get(ctx, tid) + ss, err := d.SD.Get(ctx, tid) if err != nil { return } var service *sd.Service for _, s := range ss { - d.log.Debugf("%+v", s) - if s.Node != d.node && s.Network == network { + d.Log.Debugf("%+v", s) + if s.Node != d.Node && s.Network == network { service = s break } @@ -86,8 +86,8 @@ func (d *Dialer) Dial(ctx context.Context, network string, tid string) (conn net cid = service.ID dialer := net.Dialer{ - Timeout: d.timeout, + Timeout: d.Timeout, } conn, err = dialer.DialContext(ctx, "tcp", service.Address) return -} +} \ No newline at end of file diff --git a/handler/tunnel/dialer_test.go b/handler/tunnel/dialer_test.go index ceba873f..087130e8 100644 --- a/handler/tunnel/dialer_test.go +++ b/handler/tunnel/dialer_test.go @@ -27,9 +27,11 @@ func (p *fakePool) Get(network, tid string) *Connector { } type fakeSD struct { - services []*sd.Service - err error - renewFunc func(context.Context, *sd.Service) error + services []*sd.Service + err error + renewFunc func(context.Context, *sd.Service) error + registerFunc func(ctx context.Context, service *sd.Service) error + deregisterFunc func(ctx context.Context, service *sd.Service) error } func (s *fakeSD) Get(ctx context.Context, name string) ([]*sd.Service, error) { @@ -37,10 +39,16 @@ func (s *fakeSD) Get(ctx context.Context, name string) ([]*sd.Service, error) { } func (s *fakeSD) Register(ctx context.Context, service *sd.Service, opts ...sd.Option) error { + if s.registerFunc != nil { + return s.registerFunc(ctx, service) + } return nil } func (s *fakeSD) Deregister(ctx context.Context, service *sd.Service) error { + if s.deregisterFunc != nil { + return s.deregisterFunc(ctx, service) + } return nil } @@ -56,10 +64,10 @@ func TestDialer_Dial(t *testing.T) { p := NewConnectorPool("node1") defer p.Close() d := &Dialer{ - node: "node1", - pool: p, - retry: 1, - log: testLogger(), + Node: "node1", + Pool: p, + Retry: 1, + Log: testLogger(), } _, _, _, err := d.Dial(context.Background(), "tcp", "testtid") if err != ErrTunnelNotAvailable { @@ -72,12 +80,12 @@ func TestDialer_Dial(t *testing.T) { defer p.Close() sd := &fakeSD{services: nil} d := &Dialer{ - node: "node1", - pool: p, - sd: sd, - retry: 1, - timeout: time.Second, - log: testLogger(), + Node: "node1", + Pool: p, + SD: sd, + Retry: 1, + Timeout: time.Second, + Log: testLogger(), } _, _, _, err := d.Dial(context.Background(), "tcp", "testtid") if err != ErrTunnelNotAvailable { @@ -107,12 +115,12 @@ func TestDialer_Dial(t *testing.T) { }, } d := &Dialer{ - node: "node1", - pool: p, - sd: sd, - retry: 1, - timeout: time.Second, - log: testLogger(), + Node: "node1", + Pool: p, + SD: sd, + Retry: 1, + Timeout: time.Second, + Log: testLogger(), } go func() { @@ -143,19 +151,19 @@ func TestDialer_Dial(t *testing.T) { { ID: "cid1", Name: "testtid", - Node: "node1", // same as d.node — must be filtered + Node: "node1", // same as d.Node — must be filtered Network: "tcp", Address: "127.0.0.1:9999", }, }, } d := &Dialer{ - node: "node1", - pool: p, - sd: sd, - retry: 1, - timeout: time.Second, - log: testLogger(), + Node: "node1", + Pool: p, + SD: sd, + Retry: 1, + Timeout: time.Second, + Log: testLogger(), } _, _, _, err := d.Dial(context.Background(), "tcp", "testtid") @@ -179,12 +187,12 @@ func TestDialer_Dial(t *testing.T) { }, } d := &Dialer{ - node: "node1", - pool: p, - sd: sd, - retry: 1, - timeout: time.Second, - log: testLogger(), + Node: "node1", + Pool: p, + SD: sd, + Retry: 1, + Timeout: time.Second, + Log: testLogger(), } _, _, _, err := d.Dial(context.Background(), "tcp", "testtid") @@ -198,10 +206,10 @@ func TestDialer_RetryDefault(t *testing.T) { p := NewConnectorPool("node1") defer p.Close() d := &Dialer{ - node: "node1", - pool: p, - retry: 0, // should default to 1 - log: testLogger(), + Node: "node1", + Pool: p, + Retry: 0, // should default to 1 + Log: testLogger(), } _, _, _, err := d.Dial(context.Background(), "tcp", "testtid") if err != ErrTunnelNotAvailable { @@ -215,11 +223,11 @@ func TestDialer_SDError(t *testing.T) { defer p.Close() sd := &fakeSD{err: ErrTunnelNotAvailable} d := &Dialer{ - node: "node1", - pool: p, - sd: sd, - retry: 1, - log: testLogger(), + Node: "node1", + Pool: p, + SD: sd, + Retry: 1, + Log: testLogger(), } _, _, _, err := d.Dial(context.Background(), "tcp", "testtid") if err != ErrTunnelNotAvailable { @@ -242,11 +250,11 @@ func TestDialer_SDError(t *testing.T) { }, } d := &Dialer{ - node: "node1", - pool: p, - sd: sd, - retry: 1, - log: testLogger(), + Node: "node1", + Pool: p, + SD: sd, + Retry: 1, + Log: testLogger(), } _, _, _, err := d.Dial(context.Background(), "tcp", "testtid") if err != ErrTunnelNotAvailable { diff --git a/handler/tunnel/entrypoint.go b/handler/tunnel/entrypoint.go deleted file mode 100644 index a63d489c..00000000 --- a/handler/tunnel/entrypoint.go +++ /dev/null @@ -1,204 +0,0 @@ -package tunnel - -import ( - "bufio" - "context" - "fmt" - "net" - "net/http" - "time" - - "github.com/go-gost/core/ingress" - "github.com/go-gost/core/logger" - "github.com/go-gost/core/observer/stats" - "github.com/go-gost/core/recorder" - "github.com/go-gost/core/sd" - "github.com/go-gost/relay" - dissector "github.com/go-gost/tls-dissector" - xctx "github.com/go-gost/x/ctx" - ictx "github.com/go-gost/x/internal/ctx" - xnet "github.com/go-gost/x/internal/net" - xstats "github.com/go-gost/x/observer/stats" - stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" - xrecorder "github.com/go-gost/x/recorder" -) - -// entrypoint is a public tunnel entry point that accepts external connections -// and routes them through the tunnel network. -// -// Protocol dispatch is done by peeking at the first byte of the connection: -// - relay.Version1 (0x52 'R') → relay protocol (handleConnect) -// - dissector.Handshake (0x16) → TLS passthrough (handleTLS) -// - otherwise → HTTP proxy (handleHTTP) -// -// For HTTP/TLS paths, the dial flow is: -// ep.dial() → ingress lookup (host → tunnelID) → Dialer.Dial() -// → ConnectorPool.Get() → Connector.GetConn() → mux.OpenStream() -// → relay.Response{src, dst} written to mux stream (when local node) -// -// The relay protocol path handles its own tunnel ID extraction from the -// relay request frame. -// -// When Dialer falls back to SD (no local connector), the mux layer is -// bypassed and a raw TCP connection is established to the remote node. -type entrypoint struct { - node string - service string - pool *ConnectorPool - ingress ingress.Ingress - sd sd.SD - log logger.Logger - recorder recorder.RecorderObject - transport http.RoundTripper - - sniffingWebsocket bool - websocketSampleRate float64 - - // readTimeout is applied as SetReadDeadline on the upstream connection - // before sniffing HTTP/TLS reads. It mirrors entryPointReadTimeout - // from the handler metadata. - readTimeout time.Duration -} - -func (ep *entrypoint) Handle(ctx context.Context, conn net.Conn) (err error) { - defer conn.Close() - - ro := &xrecorder.HandlerRecorderObject{ - Network: "tcp", - Node: ep.node, - Service: ep.service, - RemoteAddr: conn.RemoteAddr().String(), - LocalAddr: conn.LocalAddr().String(), - SID: xctx.SidFromContext(ctx).String(), - Time: time.Now(), - } - - if srcAddr := xctx.SrcAddrFromContext(ctx); srcAddr != nil { - ro.ClientAddr = srcAddr.String() - } - - log := ep.log.WithFields(map[string]any{ - "network": ro.Network, - "remote": conn.RemoteAddr().String(), - "local": conn.LocalAddr().String(), - "client": ro.ClientAddr, - "sid": ro.SID, - }) - log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) - - pStats := xstats.Stats{} - conn = stats_wrapper.WrapConn(conn, &pStats) - - defer func() { - if err != nil { - ro.Err = err.Error() - } - ro.InputBytes = pStats.Get(stats.KindInputBytes) - ro.OutputBytes = pStats.Get(stats.KindOutputBytes) - ro.Duration = time.Since(ro.Time) - if err := ro.Record(ctx, ep.recorder.Recorder); err != nil { - log.Errorf("record: %v", err) - } - - log.WithFields(map[string]any{ - "duration": time.Since(ro.Time), - "inputBytes": ro.InputBytes, - "outputBytes": ro.OutputBytes, - }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) - }() - - br := bufio.NewReader(conn) - v, err := br.Peek(1) - if err != nil { - return err - } - - conn = xnet.NewReadWriteConn(br, conn, conn) - if v[0] == relay.Version1 { - return ep.handleConnect(ctx, conn, ro, log) - } - if v[0] == dissector.Handshake { - return ep.handleTLS(ctx, conn, ro, log) - } - return ep.handleHTTP(ctx, conn, ro, log) -} - -func (ep *entrypoint) dial(ctx context.Context, network, addr string) (conn net.Conn, err error) { - var tunnelID relay.TunnelID - if ep.ingress != nil { - if rule := ep.ingress.GetRule(ctx, addr, ingress.WithService(ep.service)); rule != nil { - tunnelID = parseTunnelID(rule.Endpoint) - } - } - - log := ictx.LoggerFromContext(ctx) - if log == nil { - log = ep.log - } - log.Debugf("dial: new connection to host %s", addr) - - if tunnelID.IsZero() { - return nil, fmt.Errorf("%w %s", ErrTunnelRoute, addr) - } - - if ro := ictx.RecorderObjectFromContext(ctx); ro != nil { - ro.ClientID = tunnelID.String() - } - - if tunnelID.IsPrivate() { - return nil, fmt.Errorf("%w: tunnel %s is private for host %s", ErrPrivateTunnel, tunnelID, addr) - } - - log = log.WithFields(map[string]any{ - "tunnel": tunnelID.String(), - }) - - d := &Dialer{ - node: ep.node, - pool: ep.pool, - sd: ep.sd, - retry: 3, - timeout: 15 * time.Second, - log: log, - } - conn, node, cid, err := d.Dial(ctx, "tcp", tunnelID.String()) - if err != nil { - return - } - log.Debugf("dial: connected to host %s, tunnel: %s, connector: %s", addr, tunnelID, cid) - - ro := ictx.RecorderObjectFromContext(ctx) - if node == ep.node { - if ro != nil { - ro.Redirect = "" - } - - var clientAddr string - if addr := xctx.SrcAddrFromContext(ctx); addr != nil { - clientAddr = addr.String() - } - var features []relay.Feature - af := &relay.AddrFeature{} - af.ParseFrom(string(clientAddr)) - features = append(features, af) // src address - - af = &relay.AddrFeature{} - af.ParseFrom(addr) - features = append(features, af) // dst address - - if _, err = (&relay.Response{ - Version: relay.Version1, - Status: relay.StatusOK, - Features: features, - }).WriteTo(conn); err != nil { - conn.Close() - return nil, err - } - } else { - if ro != nil { - ro.Redirect = node - } - } - - return conn, nil -} \ No newline at end of file diff --git a/handler/tunnel/entrypoint/config.go b/handler/tunnel/entrypoint/config.go new file mode 100644 index 00000000..674e2922 --- /dev/null +++ b/handler/tunnel/entrypoint/config.go @@ -0,0 +1,87 @@ +package entrypoint + +import ( + "net" + "net/http" + "time" + + "github.com/go-gost/core/ingress" + "github.com/go-gost/core/logger" + "github.com/go-gost/core/recorder" + "github.com/go-gost/core/sd" +) + +// Config carries dependencies from the tunnel handler into the entrypoint. +// +// All fields must be set by the caller before calling New. The DialFunc +// bridges the package boundary to avoid an import cycle. +type Config struct { + // Node is the local tunnel node ID (from tunnelHandler.id). + Node string + // Service is the parent handler's service name, used for ingress lookups, + // logging, and TLS certificate selection. + Service string + Ingress ingress.Ingress + SD sd.SD + Logger logger.Logger + Recorder recorder.RecorderObject + SniffingWebsocket bool + WebsocketSampleRate float64 + // ReadTimeout is applied as SetReadDeadline on upstream connections + // before sniffing HTTP/TLS reads. Also used as + // http.Transport.ResponseHeaderTimeout. 0 or negative defaults to 15s. + ReadTimeout time.Duration + // ProxyProtocol is the PROXY protocol version (1 or 2) for the listener. + ProxyProtocol int + // KeepAlive enables HTTP keep-alive for the transport. Disabled by default. + KeepAlive bool + // Compression enables HTTP transport-level compression. Disabled by default. + Compression bool +} + +// DialFunc is the function signature for establishing a tunnel connection. +// Implemented by the parent package (tunnel) to avoid import cycles. +// +// ctx carries recorder objects and loggers for the tunnel connection. +// network is always "tcp" for entrypoint connections. +// tid is the string representation of the target tunnel ID. +type DialFunc func(ctx DialContext, network, tid string) (conn net.Conn, node, cid string, err error) + +// DialContext carries metadata for a dial request. +// This interface satisfies the context.Context Value() contract, +// allowing the parent package to pass a real context.Context which is +// recovered via type assertion in the DialFunc implementation. +type DialContext interface { + Value(any) any +} + +// New creates a new Entrypoint with the given config and dial function. +// +// The config fields are copied into the Entrypoint struct. The dial function +// is stored as-is — nil functions cause dial failures at runtime, not at +// construction time. +// +// The HTTP transport is initialized immediately with ep.dial as its +// DialContext; responses time out after cfg.ReadTimeout (default 15s). +func New(cfg *Config, dialFn DialFunc) *Entrypoint { + ep := &Entrypoint{ + node: cfg.Node, + service: cfg.Service, + ingress: cfg.Ingress, + sd: cfg.SD, + log: cfg.Logger, + recorder: cfg.Recorder, + sniffingWebsocket: cfg.SniffingWebsocket, + websocketSampleRate: cfg.WebsocketSampleRate, + readTimeout: cfg.ReadTimeout, + dialFn: dialFn, + } + ep.transport = &http.Transport{ + DialContext: ep.dial, + IdleConnTimeout: 30 * time.Second, + ResponseHeaderTimeout: cfg.ReadTimeout, + DisableKeepAlives: !cfg.KeepAlive, + DisableCompression: !cfg.Compression, + } + return ep +} \ No newline at end of file diff --git a/handler/tunnel/entrypoint/entrypoint.go b/handler/tunnel/entrypoint/entrypoint.go new file mode 100644 index 00000000..23a6a205 --- /dev/null +++ b/handler/tunnel/entrypoint/entrypoint.go @@ -0,0 +1,316 @@ +// Package entrypoint implements a public tunnel entry point that accepts external +// connections and routes them through the tunnel network to internal services +// behind NAT/firewall. +// +// Protocol dispatch is done by peeking at the first byte of the connection: +// +// relay.Version1 (0x52 'R') → relay protocol (handleConnect) +// dissector.Handshake (0x16) → TLS passthrough (handleTLS) +// otherwise → HTTP proxy (handleHTTP) +// +// The entrypoint is constructed by the parent tunnel package via Config and a +// DialFunc. It operates as a standalone service with its own TCP listener, +// decoder wrapper, and protocol-specific handling — independent of the parent's +// relay-handshake path. +package entrypoint + +import ( + "bufio" + "context" + "fmt" + "net" + "net/http" + "time" + + "github.com/go-gost/core/ingress" + "github.com/go-gost/core/logger" + "github.com/go-gost/core/observer/stats" + "github.com/go-gost/core/recorder" + "github.com/go-gost/core/sd" + "github.com/go-gost/relay" + dissector "github.com/go-gost/tls-dissector" + xctx "github.com/go-gost/x/ctx" + ictx "github.com/go-gost/x/internal/ctx" + xnet "github.com/go-gost/x/internal/net" + xstats "github.com/go-gost/x/observer/stats" + stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" + xrecorder "github.com/go-gost/x/recorder" +) + +// entrypoint errors — mirror of tunnel.ErrTunnelRoute and tunnel.ErrPrivateTunnel +// to avoid import cycles. The parent package's dial function returns these. +var ( + errNoRoute = fmt.Errorf("no route to host") + errPrivateTunnel = fmt.Errorf("private tunnel") +) + +// Entrypoint is a public tunnel entry point that accepts external connections +// and routes them through the tunnel network. +// +// Protocol dispatch is done by peeking at the first byte of the connection: +// - relay.Version1 (0x52 'R') → relay protocol (handleConnect) +// - dissector.Handshake (0x16) → TLS passthrough (handleTLS) +// - otherwise → HTTP proxy (handleHTTP) +// +// For HTTP/TLS paths, the dial flow is: +// +// ep.Dial() → ingress lookup (host → tunnelID) → dialFn() +// → tunnel connection +// +// The relay protocol path handles its own tunnel ID extraction from the +// relay request frame. +type Entrypoint struct { + // node is the local tunnel node ID (from tunnelHandler.id). + node string + // service is the parent handler's service name, used for ingress lookups. + service string + ingress ingress.Ingress + sd sd.SD + log logger.Logger + recorder recorder.RecorderObject + transport http.RoundTripper + + // sniffingWebsocket enables WebSocket frame-level recording. + sniffingWebsocket bool + // websocketSampleRate controls the rate-limiter for WebSocket frame + // recording samples. 0 defaults to sniffing.DefaultSampleRate. + websocketSampleRate float64 + + // readTimeout is applied as SetReadDeadline on the upstream connection + // before sniffing HTTP/TLS reads. It mirrors entryPointReadTimeout + // from the handler metadata. + readTimeout time.Duration + + // dialFn is the function for establishing a tunnel connection, + // provided by the parent package. + dialFn DialFunc +} + +// Handle processes an incoming connection on the entrypoint listener. +// +// The first byte is peeked to dispatch protocol handling: +// - relay.Version1 (0x52 'R') → ep.handleConnect (relay protocol) +// - dissector.Handshake (0x16) → ep.handleTLS (TLS passthrough) +// - otherwise → ep.handleHTTP (HTTP forward proxy) +// +// Recording instrumentation (stats, HTTP/TLS metadata) wraps the connection +// and is finalized in the deferred recorder callback. +func (ep *Entrypoint) Handle(ctx context.Context, conn net.Conn) (err error) { + defer conn.Close() + + ro := &xrecorder.HandlerRecorderObject{ + Network: "tcp", + Node: ep.node, + Service: ep.service, + RemoteAddr: conn.RemoteAddr().String(), + LocalAddr: conn.LocalAddr().String(), + SID: xctx.SidFromContext(ctx).String(), + Time: time.Now(), + } + + if srcAddr := xctx.SrcAddrFromContext(ctx); srcAddr != nil { + ro.ClientAddr = srcAddr.String() + } + + log := ep.log.WithFields(map[string]any{ + "network": ro.Network, + "remote": conn.RemoteAddr().String(), + "local": conn.LocalAddr().String(), + "client": ro.ClientAddr, + "sid": ro.SID, + }) + log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) + + pStats := xstats.Stats{} + conn = stats_wrapper.WrapConn(conn, &pStats) + + defer func() { + if err != nil { + ro.Err = err.Error() + } + ro.InputBytes = pStats.Get(stats.KindInputBytes) + ro.OutputBytes = pStats.Get(stats.KindOutputBytes) + ro.Duration = time.Since(ro.Time) + if err := ro.Record(ctx, ep.recorder.Recorder); err != nil { + log.Errorf("record: %v", err) + } + + log.WithFields(map[string]any{ + "duration": time.Since(ro.Time), + "inputBytes": ro.InputBytes, + "outputBytes": ro.OutputBytes, + }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) + }() + + br := bufio.NewReader(conn) + v, err := br.Peek(1) + if err != nil { + return err + } + + conn = xnet.NewReadWriteConn(br, conn, conn) + if v[0] == relay.Version1 { + return ep.handleConnect(ctx, conn, ro, log) + } + if v[0] == dissector.Handshake { + return ep.handleTLS(ctx, conn, ro, log) + } + return ep.handleHTTP(ctx, conn, ro, log) +} + +// dial resolves the host to a tunnel ID via ingress rules, then establishes +// a connection through the tunnel network using the provided dial function. +// +// Steps: +// 1. Look up addr in the ingress table to find the target tunnel ID. +// 2. If the tunnel ID is zero, return errNoRoute. +// 3. If the tunnel is private ($-prefixed), return errPrivateTunnel. +// 4. Call ep.dialFn(ctx, "tcp", tunnelID.String()) to open a tunnel stream. +// 5. If the connected node is the local node, write StatusOK + address +// features (src, dst) to the stream — the internal client uses these +// addresses to know where to connect. +// 6. If the connected node is remote, just set ro.Redirect — the remote +// node handles relay framing on its own. +// +// Returns the tunnel stream as conn, or an error. +func (ep *Entrypoint) dial(ctx context.Context, network, addr string) (conn net.Conn, err error) { + var tunnelID relay.TunnelID + if ep.ingress != nil { + if rule := ep.ingress.GetRule(ctx, addr, ingress.WithService(ep.service)); rule != nil { + tunnelID = parseTunnelID(rule.Endpoint) + } + } + + log := ictx.LoggerFromContext(ctx) + if log == nil { + log = ep.log + } + log.Debugf("dial: new connection to host %s", addr) + + if tunnelID.IsZero() { + return nil, fmt.Errorf("%w %s", errNoRoute, addr) + } + + if ro := ictx.RecorderObjectFromContext(ctx); ro != nil { + ro.ClientID = tunnelID.String() + } + + if tunnelID.IsPrivate() { + return nil, fmt.Errorf("%w: tunnel %s is private for host %s", errPrivateTunnel, tunnelID, addr) + } + + log = log.WithFields(map[string]any{ + "tunnel": tunnelID.String(), + }) + + if ep.dialFn == nil { + return nil, fmt.Errorf("tunnel not available") + } + + cc, node, cid, err := ep.dialFn(ctx, "tcp", tunnelID.String()) + if err != nil { + return + } + log.Debugf("dial: connected to host %s, tunnel: %s, connector: %s", addr, tunnelID, cid) + + ro := ictx.RecorderObjectFromContext(ctx) + if node == ep.node { + if ro != nil { + ro.Redirect = "" + } + + var clientAddr string + if addr := xctx.SrcAddrFromContext(ctx); addr != nil { + clientAddr = addr.String() + } + var features []relay.Feature + af := &relay.AddrFeature{} + af.ParseFrom(string(clientAddr)) + features = append(features, af) // src address + + af = &relay.AddrFeature{} + af.ParseFrom(addr) + features = append(features, af) // dst address + + if _, err = (&relay.Response{ + Version: relay.Version1, + Status: relay.StatusOK, + Features: features, + }).WriteTo(cc); err != nil { + cc.Close() + return nil, err + } + } else { + if ro != nil { + ro.Redirect = node + } + } + + conn = cc + return conn, nil +} + +// parseTunnelID parses a tunnel ID from a string. +// If s is empty or contains an invalid UUID, the returned tunnel ID is zero +// (callers must check IsZero). A leading '$' prefix marks the tunnel as private. +// +// This is a copy of tunnel.ParseTunnelID that avoids importing google/uuid. +// It uses a local parseUUID instead of uuid.Parse. +func parseTunnelID(s string) (tid relay.TunnelID) { + if s == "" { + return + } + private := false + if s[0] == '$' { + private = true + s = s[1:] + } + u, err := parseUUID(s) + if err != nil { + return + } + var raw [16]byte + copy(raw[:], u[:]) + + if private { + return relay.NewPrivateTunnelID(raw[:]) + } + return relay.NewTunnelID(raw[:]) +} + +// parseUUID is a simplified UUID parser that avoids importing google/uuid. +// It parses the standard 8-4-4-4-12 hex format (36 characters) and returns +// the 16 raw bytes. An error is returned for invalid length, missing hyphens, +// or non-hex characters. +func parseUUID(s string) ([]byte, error) { + if len(s) != 36 { + return nil, fmt.Errorf("invalid UUID length") + } + b := make([]byte, 16) + for i, j := 0, 0; i < 36; i++ { + if i == 8 || i == 13 || i == 18 || i == 23 { + if s[i] != '-' { + return nil, fmt.Errorf("invalid UUID format") + } + continue + } + var v byte + switch { + case s[i] >= '0' && s[i] <= '9': + v = s[i] - '0' + case s[i] >= 'a' && s[i] <= 'f': + v = s[i] - 'a' + 10 + case s[i] >= 'A' && s[i] <= 'F': + v = s[i] - 'A' + 10 + default: + return nil, fmt.Errorf("invalid UUID character") + } + if j%2 == 0 { + b[j/2] = v << 4 + } else { + b[j/2] |= v + } + j++ + } + return b, nil +} \ No newline at end of file diff --git a/handler/tunnel/entrypoint/entrypoint_test.go b/handler/tunnel/entrypoint/entrypoint_test.go new file mode 100644 index 00000000..d4bce63d --- /dev/null +++ b/handler/tunnel/entrypoint/entrypoint_test.go @@ -0,0 +1,846 @@ +package entrypoint + +import ( + "bufio" + "bytes" + "context" + "fmt" + "io" + "net" + "net/http" + "strings" + "testing" + "time" + + "github.com/go-gost/core/ingress" + "github.com/go-gost/core/logger" + "github.com/go-gost/relay" + xctx "github.com/go-gost/x/ctx" + xstats "github.com/go-gost/x/observer/stats" + xrecorder "github.com/go-gost/x/recorder" +) + +// test helpers + +type nopLogger struct { + logger.Logger +} + +func (nopLogger) Debugf(format string, args ...any) {} +func (nopLogger) Infof(format string, args ...any) {} +func (nopLogger) Warnf(format string, args ...any) {} +func (nopLogger) Errorf(format string, args ...any) {} +func (nopLogger) Tracef(format string, args ...any) {} +func (nopLogger) Debug(args ...any) {} +func (nopLogger) Info(args ...any) {} +func (nopLogger) Warn(args ...any) {} +func (nopLogger) Error(args ...any) {} +func (nopLogger) Trace(args ...any) {} +func (nopLogger) Fatal(args ...any) {} +func (nopLogger) Fatalf(format string, args ...any) {} +func (nopLogger) GetLevel() logger.LogLevel { return logger.ErrorLevel } +func (nopLogger) IsLevelEnabled(level logger.LogLevel) bool { return false } +func (nopLogger) WithFields(fields map[string]any) logger.Logger { return nopLogger{} } + +func testLogger() logger.Logger { return nopLogger{} } + +// mockIngress implements ingress.Ingress with immediate rule lookup. +type mockIngress struct { + rules map[string]*ingress.Rule +} + +func newMockIngress(rules []*ingress.Rule) *mockIngress { + m := &mockIngress{rules: make(map[string]*ingress.Rule)} + for _, r := range rules { + m.rules[r.Hostname] = r + } + return m +} + +func (m *mockIngress) GetRule(ctx context.Context, host string, opts ...ingress.Option) *ingress.Rule { + // Strip port like real ingress does + if h, _, err := net.SplitHostPort(host); err == nil && h != "" { + host = h + } + return m.rules[host] +} + +func (m *mockIngress) SetRule(ctx context.Context, rule *ingress.Rule, opts ...ingress.Option) bool { + m.rules[rule.Hostname] = rule + return true +} + +func newTestConfig() *Config { + return &Config{ + Node: "test-node", + Service: "test-service", + Logger: testLogger(), + Ingress: newMockIngress(nil), + ReadTimeout: 15 * time.Second, + } +} + +type fakeDialFn struct { + conn net.Conn + node string + cid string + err error +} + +func (f *fakeDialFn) dial(ctx DialContext, network, tid string) (net.Conn, string, string, error) { + if f.err != nil { + return nil, "", "", f.err + } + if f.conn == nil { + return nil, "", "", fmt.Errorf("no connection") + } + return f.conn, f.node, f.cid, nil +} + +type fakeConn struct { + net.Conn + readBuf []byte + offset int + writeBuf bytes.Buffer + closed bool +} + +func (c *fakeConn) Read(b []byte) (n int, err error) { + if c.offset >= len(c.readBuf) { + return 0, io.EOF + } + n = copy(b, c.readBuf[c.offset:]) + c.offset += n + return +} + +func (c *fakeConn) Write(b []byte) (n int, err error) { + return c.writeBuf.Write(b) +} + +func (c *fakeConn) Close() error { + c.closed = true + return nil +} + +func (c *fakeConn) RemoteAddr() net.Addr { + return &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 12345} +} + +func (c *fakeConn) LocalAddr() net.Addr { + return &net.TCPAddr{IP: net.ParseIP("10.0.0.2"), Port: 8080} +} + +func (c *fakeConn) SetReadDeadline(t time.Time) error { + return nil +} + +// pipeConn is a simple in-memory pipe to simulate bidirectional connections. +// It embeds readBuf for tests that need to inject data. +type pipeConn struct { + readBuf []byte + offset int + reader *io.PipeReader + writer *io.PipeWriter + closed bool + remote net.Addr + local net.Addr +} + +func (c *pipeConn) Read(b []byte) (n int, err error) { + if c.readBuf != nil && c.offset < len(c.readBuf) { + n = copy(b, c.readBuf[c.offset:]) + c.offset += n + return n, nil + } + if c.reader == nil { + return 0, io.EOF + } + return c.reader.Read(b) +} + +func (c *pipeConn) Write(b []byte) (n int, err error) { + if c.writer == nil { + return len(b), nil + } + return c.writer.Write(b) +} + +func (c *pipeConn) Close() error { + if !c.closed { + c.closed = true + if c.reader != nil { + c.reader.Close() + } + if c.writer != nil { + c.writer.Close() + } + } + return nil +} + +func (c *pipeConn) RemoteAddr() net.Addr { + if c.remote != nil { + return c.remote + } + return &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 12345} +} + +func (c *pipeConn) LocalAddr() net.Addr { + if c.local != nil { + return c.local + } + return &net.TCPAddr{IP: net.ParseIP("10.0.0.2"), Port: 8080} +} + +func (c *pipeConn) SetReadDeadline(t time.Time) error { return nil } +func (c *pipeConn) SetWriteDeadline(t time.Time) error { return nil } +func (c *pipeConn) SetDeadline(t time.Time) error { return nil } + +func newPipePair() (*pipeConn, *pipeConn) { + pr1, pw1 := io.Pipe() + pr2, pw2 := io.Pipe() + a := &pipeConn{reader: pr1, writer: pw2} + b := &pipeConn{reader: pr2, writer: pw1} + return a, b +} + +// buildRelayConnectRequest builds a relay connect request for the entrypoint's +// handleConnect path. +func buildRelayConnectRequest(t *testing.T, tid relay.TunnelID, src, dst string) []byte { + t.Helper() + req := relay.Request{ + Version: relay.Version1, + Cmd: relay.CmdConnect, + } + if src != "" { + af := &relay.AddrFeature{} + af.ParseFrom(src) + req.Features = append(req.Features, af) + } + if dst != "" { + af := &relay.AddrFeature{} + af.ParseFrom(dst) + req.Features = append(req.Features, af) + } + req.Features = append(req.Features, &relay.TunnelFeature{ID: tid}) + var buf bytes.Buffer + _, err := req.WriteTo(&buf) + if err != nil { + t.Fatal(err) + } + return buf.Bytes() +} + +// --- Tests --- + +// parseUUID tests + +func TestParseUUID(t *testing.T) { + t.Run("valid uuid", func(t *testing.T) { + b, err := parseUUID("6ba7b810-9dad-11d1-80b4-00c04fd430c8") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(b) != 16 { + t.Errorf("expected 16 bytes, got %d", len(b)) + } + }) + + t.Run("invalid length", func(t *testing.T) { + _, err := parseUUID("short") + if err == nil { + t.Error("expected error for short string") + } + }) + + t.Run("missing hyphens", func(t *testing.T) { + _, err := parseUUID("6ba7b8109dad11d180b400c04fd430c8") + if err == nil { + t.Error("expected error without hyphens") + } + }) + + t.Run("non-hex character", func(t *testing.T) { + _, err := parseUUID("6ba7b810-9dad-11d1-80b4-00c04fd430zz") + if err == nil { + t.Error("expected error for non-hex characters") + } + }) + + t.Run("uppercase hex", func(t *testing.T) { + b, err := parseUUID("6BA7B810-9DAD-11D1-80B4-00C04FD430C8") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(b) != 16 { + t.Errorf("expected 16 bytes, got %d", len(b)) + } + }) +} + +// parseTunnelID tests + +func TestParseTunnelID(t *testing.T) { + t.Run("empty string", func(t *testing.T) { + tid := parseTunnelID("") + if !tid.IsZero() { + t.Error("expected zero tunnel ID for empty string") + } + }) + + t.Run("valid uuid", func(t *testing.T) { + tid := parseTunnelID("6ba7b810-9dad-11d1-80b4-00c04fd430c8") + if tid.IsZero() { + t.Error("expected non-zero tunnel ID") + } + }) + + t.Run("private tunnel ($ prefix)", func(t *testing.T) { + tid := parseTunnelID("$6ba7b810-9dad-11d1-80b4-00c04fd430c8") + if !tid.IsPrivate() { + t.Error("expected private tunnel ID") + } + }) + + t.Run("invalid uuid returns zero", func(t *testing.T) { + tid := parseTunnelID("not-a-uuid-at-all") + if !tid.IsZero() { + t.Error("expected zero tunnel ID for invalid input") + } + }) + + t.Run("invalid uuid with $ prefix", func(t *testing.T) { + tid := parseTunnelID("$not-a-uuid-at-all") + if !tid.IsZero() { + t.Error("expected zero tunnel ID for invalid input") + } + }) +} + +// New tests + +func TestNew(t *testing.T) { + t.Run("creates entrypoint with transport", func(t *testing.T) { + dialFn := func(ctx DialContext, network, tid string) (net.Conn, string, string, error) { + return nil, "", "", nil + } + ep := New(newTestConfig(), dialFn) + if ep == nil { + t.Fatal("expected non-nil entrypoint") + } + if ep.transport == nil { + t.Error("expected non-nil transport") + } + if ep.node != "test-node" { + t.Errorf("expected node test-node, got %s", ep.node) + } + }) +} + +// Handle tests + +func TestHandle_PeekError(t *testing.T) { + dialFn := func(ctx DialContext, network, tid string) (net.Conn, string, string, error) { + return nil, "", "", nil + } + ep := New(newTestConfig(), dialFn) + + // Empty connection should fail peek + conn := &fakeConn{readBuf: []byte{}} + err := ep.Handle(context.Background(), conn) + if err == nil { + t.Error("expected error from empty connection") + } +} + +func TestHandle_HTTP_Path(t *testing.T) { + ing := newMockIngress([]*ingress.Rule{ + {Hostname: "example.com", Endpoint: "6ba7b810-9dad-11d1-80b4-00c04fd430c8"}, + }) + cfg := newTestConfig() + cfg.Ingress = ing + + client, server := newPipePair() + dialFn := func(ctx DialContext, network, tid string) (net.Conn, string, string, error) { + return client, "test-node", "cid1", nil + } + ep := New(cfg, dialFn) + + httpReq := "GET / HTTP/1.1\r\nHost: example.com\r\n\r\n" + conn := &fakeConn{readBuf: []byte(httpReq)} + + go func() { + // Server side: read the relay StatusOK response with address features + resp := relay.Response{} + _, err := resp.ReadFrom(server) + if err != nil { + return + } + server.Write([]byte("HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello")) + server.Close() + }() + + err := ep.Handle(context.Background(), conn) + // Pipe will close when the server side closes; Handle may return nil or EOF + t.Logf("Handle HTTP returned: %v", err) +} + +func TestHandle_Relay_Path(t *testing.T) { + dialFn := func(ctx DialContext, network, tid string) (net.Conn, string, string, error) { + client, server := newPipePair() + go func() { + // Read the StatusOK on the mux-stream (reply from public node) + resp := relay.Response{} + resp.ReadFrom(server) + server.Close() + }() + return client, "test-node", "cid1", nil + } + + tid := relay.NewTunnelID(func() []byte { + u, _ := parseUUID("6ba7b810-9dad-11d1-80b4-00c04fd430c8") + return u + }()) + + ep := New(newTestConfig(), dialFn) + + relayData := buildRelayConnectRequest(t, tid, "10.0.0.1:12345", "192.168.1.1:80") + conn := &fakeConn{readBuf: relayData} + + err := ep.Handle(context.Background(), conn) + // Expected: relay path processes, pipes, then returns + t.Logf("Handle relay path returned: %v", err) +} + +func TestHandle_Relay_NoTunnelID(t *testing.T) { + dialFn := func(ctx DialContext, network, tid string) (net.Conn, string, string, error) { + return nil, "", "", nil + } + ep := New(newTestConfig(), dialFn) + + // Relay request without tunnel feature + req := relay.Request{ + Version: relay.Version1, + Cmd: relay.CmdConnect, + Features: []relay.Feature{ + &relay.AddrFeature{Host: "10.0.0.1", Port: 12345}, + }, + } + var buf bytes.Buffer + req.WriteTo(&buf) + + conn := &fakeConn{readBuf: buf.Bytes()} + err := ep.Handle(context.Background(), conn) + if err == nil { + t.Error("expected error for missing tunnel ID") + } +} + +// dial tests + +func TestDial_NoIngress(t *testing.T) { + ep := New(newTestConfig(), nil) + _, err := ep.dial(context.Background(), "tcp", "example.com:80") + if err == nil { + t.Error("expected error for host not in ingress") + } + if !strings.Contains(err.Error(), "no route to host") { + t.Errorf("expected 'no route to host', got %v", err) + } +} + +func TestDial_PrivateTunnel(t *testing.T) { + ing := newMockIngress([]*ingress.Rule{ + {Hostname: "internal.example.com", Endpoint: "$6ba7b810-9dad-11d1-80b4-00c04fd430c8"}, + }) + cfg := newTestConfig() + cfg.Ingress = ing + ep := New(cfg, nil) + + _, err := ep.dial(context.Background(), "tcp", "internal.example.com:80") + if err == nil { + t.Error("expected error for private tunnel") + } + if !strings.Contains(err.Error(), "private tunnel") { + t.Errorf("expected 'private tunnel', got %v", err) + } +} + +func TestDial_NilDialFn(t *testing.T) { + ing := newMockIngress([]*ingress.Rule{ + {Hostname: "example.com", Endpoint: "6ba7b810-9dad-11d1-80b4-00c04fd430c8"}, + }) + cfg := newTestConfig() + cfg.Ingress = ing + ep := New(cfg, nil) + + _, err := ep.dial(context.Background(), "tcp", "example.com:80") + if err == nil { + t.Error("expected error for nil dialFn") + } +} + +func TestDial_DialFnError(t *testing.T) { + ing := newMockIngress([]*ingress.Rule{ + {Hostname: "example.com", Endpoint: "6ba7b810-9dad-11d1-80b4-00c04fd430c8"}, + }) + cfg := newTestConfig() + cfg.Ingress = ing + ep := New(cfg, func(ctx DialContext, network, tid string) (net.Conn, string, string, error) { + return nil, "", "", fmt.Errorf("dial failed") + }) + + _, err := ep.dial(context.Background(), "tcp", "example.com:80") + if err == nil { + t.Error("expected error from dialFn") + } +} + +func TestDial_RemoteNode(t *testing.T) { + ing := newMockIngress([]*ingress.Rule{ + {Hostname: "example.com", Endpoint: "6ba7b810-9dad-11d1-80b4-00c04fd430c8"}, + }) + cfg := newTestConfig() + cfg.Ingress = ing + client, _ := net.Pipe() + ep := New(cfg, func(ctx DialContext, network, tid string) (net.Conn, string, string, error) { + return client, "remote-node", "cid1", nil + }) + + conn, err := ep.dial(context.Background(), "tcp", "example.com:80") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if conn == nil { + t.Error("expected non-nil connection") + } +} + +func TestDial_LocalNode(t *testing.T) { + ing := newMockIngress([]*ingress.Rule{ + {Hostname: "example.com", Endpoint: "6ba7b810-9dad-11d1-80b4-00c04fd430c8"}, + }) + cfg := newTestConfig() + cfg.Ingress = ing + client, server := net.Pipe() + ep := New(cfg, func(ctx DialContext, network, tid string) (net.Conn, string, string, error) { + return client, "test-node", "cid1", nil + }) + + // For local node, expect relay StatusOK + address features written to stream + go func() { + resp := relay.Response{} + _, err := resp.ReadFrom(server) + if err != nil { + return + } + server.Close() + }() + + // Create a context with a source address so the local-node path + // doesn't fail on ParseFrom("") + ctx := xctx.ContextWithSrcAddr(context.Background(), &net.TCPAddr{IP: net.ParseIP("10.0.0.1"), Port: 12345}) + conn, err := ep.dial(ctx, "tcp", "example.com:80") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if conn == nil { + t.Error("expected non-nil connection") + } +} + +// handleConnect tests + +func TestHandleConnect_DialFnError(t *testing.T) { + dialFn := func(ctx DialContext, network, tid string) (net.Conn, string, string, error) { + return nil, "", "", fmt.Errorf("dial error") + } + ep := New(newTestConfig(), dialFn) + + tid := relay.NewTunnelID(func() []byte { + u, _ := parseUUID("6ba7b810-9dad-11d1-80b4-00c04fd430c8") + return u + }()) + + conn := &fakeConn{readBuf: buildRelayConnectRequest(t, tid, "10.0.0.1:12345", "")} + + // Use handleConnect directly (not via Handle dispatch) + err := ep.handleConnect(context.Background(), conn, &xrecorder.HandlerRecorderObject{}, testLogger()) + if err == nil { + t.Error("expected error from dial failure") + } +} + +// handleHTTP tests + +func TestHandleHTTP_BadRequest(t *testing.T) { + ep := New(newTestConfig(), nil) + + // Bad HTTP data + conn := &fakeConn{readBuf: []byte("NOT HTTP\r\n")} + + err := ep.handleHTTP(context.Background(), conn, &xrecorder.HandlerRecorderObject{}, testLogger()) + if err == nil { + t.Error("expected error for bad HTTP request") + } +} + +func TestHTTPRoundTrip_LoopDetection(t *testing.T) { + ep := New(newTestConfig(), nil) + ep.node = "test-node" + + // Request with our own node in Gost-Forwarded-Node header + req, _ := http.NewRequest("GET", "http://example.com/", nil) + req.Header.Set("Gost-Forwarded-Node", "test-node") + + rw := &pipeConn{} + ro := &xrecorder.HandlerRecorderObject{ + HTTP: &xrecorder.HTTPRecorderObject{}, + } + pStats := xstats.Stats{} + + err := ep.httpRoundTrip(context.Background(), rw, req, ro, &pStats, testLogger()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if ro.HTTP.StatusCode != http.StatusServiceUnavailable { + t.Errorf("expected 503, got %d", ro.HTTP.StatusCode) + } +} + +func TestHTTPRoundTrip_NoRoute(t *testing.T) { + dialFn := func(ctx DialContext, network, tid string) (net.Conn, string, string, error) { + return nil, "", "", errNoRoute + } + ep := New(newTestConfig(), dialFn) + + req, _ := http.NewRequest("GET", "http://unknown.example.com/", nil) + rw := &fakeConn{} + ro := &xrecorder.HandlerRecorderObject{ + HTTP: &xrecorder.HTTPRecorderObject{}, + } + pStats := xstats.Stats{} + + err := ep.httpRoundTrip(context.Background(), rw, req, ro, &pStats, testLogger()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + resp, err := http.ReadResponse(bufio.NewReader(&rw.writeBuf), req) + if err != nil { + t.Fatalf("read response: %v", err) + } + if resp.StatusCode != http.StatusBadGateway { + t.Errorf("expected 502, got %d", resp.StatusCode) + } +} + +func TestUpgradeType(t *testing.T) { + t.Run("no upgrade header", func(t *testing.T) { + h := http.Header{} + if upgradeType(h) != "" { + t.Error("expected empty for no upgrade header") + } + }) + + t.Run("has upgrade", func(t *testing.T) { + h := http.Header{} + h.Set("Connection", "Upgrade") + h.Set("Upgrade", "websocket") + if upgradeType(h) != "websocket" { + t.Errorf("expected 'websocket', got '%s'", upgradeType(h)) + } + }) + + t.Run("upgrade with multiple connection tokens", func(t *testing.T) { + h := http.Header{} + h.Set("Connection", "keep-alive, Upgrade") + h.Set("Upgrade", "websocket") + if upgradeType(h) != "websocket" { + t.Errorf("expected 'websocket', got '%s'", upgradeType(h)) + } + }) +} + +func TestHandleUpgradeResponse_MismatchedUpgrade(t *testing.T) { + req, _ := http.NewRequest("GET", "http://example.com/", nil) + req.Header.Set("Connection", "Upgrade") + req.Header.Set("Upgrade", "websocket") + + res := &http.Response{ + StatusCode: http.StatusSwitchingProtocols, + Header: http.Header{}, + Body: &pipeConn{}, + } + res.Header.Set("Connection", "Upgrade") + res.Header.Set("Upgrade", "h2c") // mismatch + + ep := New(newTestConfig(), nil) + err := ep.handleUpgradeResponse(context.Background(), &pipeConn{}, req, res, &xrecorder.HandlerRecorderObject{}, testLogger()) + if err == nil { + t.Error("expected error for mismatched upgrade protocol") + } +} + +func TestHandleUpgradeResponse_NonWritableBody(t *testing.T) { + req, _ := http.NewRequest("GET", "http://example.com/", nil) + req.Header.Set("Connection", "Upgrade") + req.Header.Set("Upgrade", "websocket") + + res := &http.Response{ + StatusCode: http.StatusSwitchingProtocols, + Header: http.Header{}, + Body: io.NopCloser(strings.NewReader("not writable")), + } + res.Header.Set("Connection", "Upgrade") + res.Header.Set("Upgrade", "websocket") + + ep := New(newTestConfig(), nil) + err := ep.handleUpgradeResponse(context.Background(), &pipeConn{}, req, res, &xrecorder.HandlerRecorderObject{}, testLogger()) + if err == nil { + t.Error("expected error for non-writable body") + } +} + +// eplistener tests + +func TestTCPListener_Init_Accept_Close(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + + tcpLn := NewTCPListener(ln) + addr := tcpLn.Addr() + if addr == nil { + t.Error("expected non-nil addr") + } + + err = tcpLn.Init(nil) + if err != nil { + t.Fatalf("Init failed: %v", err) + } + + err = tcpLn.Close() + if err != nil { + t.Errorf("Close failed: %v", err) + } +} + +// entrypointHandler tests + +func TestNewEntrypointHandler(t *testing.T) { + ep := New(newTestConfig(), nil) + h := NewHandler(ep) + if h == nil { + t.Fatal("expected non-nil handler") + } + + err := h.Init(nil) + if err != nil { + t.Errorf("Init should return nil, got %v", err) + } +} + +// copyWebsocketFrame tests + +func TestCopyWebsocketFrame_NoRecorder(t *testing.T) { + ep := New(newTestConfig(), nil) + buf := &bytes.Buffer{} + + // Build a minimal WebSocket frame (FIN=1, opcode=1 (text), masked, length=5) + frame := wsFrame(t, true, 1, true, []byte("hello")) + r := bytes.NewReader(frame) + w := &bytes.Buffer{} + ro := &xrecorder.HandlerRecorderObject{} + + err := ep.copyWebsocketFrame(w, r, buf, "client", ro) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + // Verify output matches input + if !bytes.Equal(frame, w.Bytes()) { + t.Error("output does not match input") + } +} + +// wsFrame builds a minimal unmasked WebSocket frame. +func wsFrame(t *testing.T, fin bool, opCode byte, masked bool, payload []byte) []byte { + t.Helper() + var header byte + if fin { + header |= 0x80 + } + header |= opCode & 0x0f + + var maskBit byte + if masked { + maskBit = 0x80 + } + + var buf bytes.Buffer + buf.WriteByte(header) + + if len(payload) < 126 { + buf.WriteByte(byte(len(payload)) | maskBit) + } else if len(payload) < 65536 { + buf.WriteByte(126 | maskBit) + buf.WriteByte(byte(len(payload) >> 8)) + buf.WriteByte(byte(len(payload))) + } else { + buf.WriteByte(127 | maskBit) + for i := 7; i >= 0; i-- { + buf.WriteByte(byte(uint64(len(payload)) >> (8 * i))) + } + } + + if masked { + maskKey := []byte{0x01, 0x02, 0x03, 0x04} + buf.Write(maskKey) + for i, p := range payload { + buf.WriteByte(p ^ maskKey[i%4]) + } + } else { + buf.Write(payload) + } + + return buf.Bytes() +} + +// sniffingWebsocketFrame test +func TestEntrypoint_WebsocketSniffingCodePaths(t *testing.T) { + // Verify that the copyWebsocketFrame function handles both client and server + // directions and properly tracks InputBytes/OutputBytes. + buf := &bytes.Buffer{} + r := bytes.NewReader(wsFrame(t, true, 1, false, []byte("hello"))) + w := &bytes.Buffer{} + ro := &xrecorder.HandlerRecorderObject{} + + ep := New(newTestConfig(), nil) + err := ep.copyWebsocketFrame(w, r, buf, "client", ro) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if ro.InputBytes == 0 { + t.Error("expected non-zero InputBytes for client direction") + } + if ro.OutputBytes != 0 { + t.Error("expected 0 OutputBytes for client direction") + } + + // Server direction + r2 := bytes.NewReader(wsFrame(t, true, 1, false, []byte("world"))) + w2 := &bytes.Buffer{} + ro2 := &xrecorder.HandlerRecorderObject{} + err = ep.copyWebsocketFrame(w2, r2, buf, "server", ro2) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if ro2.OutputBytes == 0 { + t.Error("expected non-zero OutputBytes for server direction") + } + if ro2.InputBytes != 0 { + t.Error("expected 0 InputBytes for server direction") + } +} \ No newline at end of file diff --git a/handler/tunnel/ephttp.go b/handler/tunnel/entrypoint/ephttp.go similarity index 85% rename from handler/tunnel/ephttp.go rename to handler/tunnel/entrypoint/ephttp.go index 18096b13..b768437e 100644 --- a/handler/tunnel/ephttp.go +++ b/handler/tunnel/entrypoint/ephttp.go @@ -1,4 +1,4 @@ -package tunnel +package entrypoint import ( "bufio" @@ -26,6 +26,18 @@ import ( "golang.org/x/net/http/httpguts" ) +// HTTP header names used for tunnel loop detection and session tracking. +const ( + // httpHeaderSID carries the session ID from the external request + // through the tunnel to the internal service. + httpHeaderSID = "Gost-Sid" + // httpHeaderForwardedNode carries the node ID chain from each + // entrypoint the request has traversed. Used for loop detection: + // if the header already contains our node, the request is looping + // and is rejected with 503. + httpHeaderForwardedNode = "Gost-Forwarded-Node" +) + // handleHTTP processes an HTTP request arriving at the entrypoint. // // Flow: @@ -41,7 +53,7 @@ import ( // // HTTP request body recording: if recorder options specify HTTPBody, // the request body is wrapped in a xhttp.Body for capture. -func (ep *entrypoint) handleHTTP(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (err error) { +func (ep *Entrypoint) handleHTTP(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (err error) { pStats := xstats.Stats{} conn = stats_wrapper.WrapConn(conn, &pStats) @@ -97,7 +109,7 @@ func (ep *entrypoint) handleHTTP(ctx context.Context, conn net.Conn, ro *xrecord } } -func (ep *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriteCloser, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats stats.Stats, log logger.Logger) (err error) { +func (ep *Entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriteCloser, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats stats.Stats, log logger.Logger) (err error) { ro2 := &xrecorder.HandlerRecorderObject{} *ro2 = *ro ro = ro2 @@ -214,12 +226,12 @@ func (ep *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriteCloser, } if err != nil { - if errors.Is(err, ErrTunnelRoute) || errors.Is(err, ErrPrivateTunnel) { + if errors.Is(err, errNoRoute) || errors.Is(err, errPrivateTunnel) { res.StatusCode = http.StatusBadGateway ro.HTTP.StatusCode = http.StatusBadGateway } res.Write(rw) - return + return nil } defer resp.Body.Close() @@ -263,6 +275,8 @@ func (ep *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriteCloser, return } +// upgradeType returns the upgrade protocol from an HTTP header set, +// or empty string if the request/response does not include an Upgrade. func upgradeType(h http.Header) string { if !httpguts.HeaderValuesContainsToken(h["Connection"], "Upgrade") { return "" @@ -270,7 +284,13 @@ func upgradeType(h http.Header) string { return h.Get("Upgrade") } -func (ep *entrypoint) handleUpgradeResponse(ctx context.Context, rw io.ReadWriteCloser, req *http.Request, res *http.Response, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { +// handleUpgradeResponse handles an HTTP upgrade (101 Switching Protocols). +// +// It validates that the upgrade protocol matches between request and response, +// writes the response to the client, and then pipes the raw connection. +// For WebSocket upgrades with sniffing enabled, it delegates to +// sniffingWebsocketFrame for frame-level recording. +func (ep *Entrypoint) handleUpgradeResponse(ctx context.Context, rw io.ReadWriteCloser, req *http.Request, res *http.Response, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { reqUpType := upgradeType(req.Header) resUpType := upgradeType(res.Header) if !strings.EqualFold(reqUpType, resUpType) { diff --git a/handler/tunnel/entrypoint/eplistener.go b/handler/tunnel/entrypoint/eplistener.go new file mode 100644 index 00000000..b2ab0a93 --- /dev/null +++ b/handler/tunnel/entrypoint/eplistener.go @@ -0,0 +1,105 @@ +package entrypoint + +import ( + "context" + "net" + "time" + + "github.com/go-gost/core/handler" + "github.com/go-gost/core/listener" + md "github.com/go-gost/core/metadata" + admission "github.com/go-gost/x/admission/wrapper" + "github.com/go-gost/x/internal/net/proxyproto" + climiter "github.com/go-gost/x/limiter/conn/wrapper" + metrics "github.com/go-gost/x/metrics/wrapper" +) + +// entrypointHandler wraps an Entrypoint as a handler.Handler for use with +// the service framework (xservice.NewService). +// +// The handler simply delegates Handle() to ep.Handle() — all protocol +// dispatch, ingress resolution, and tunnel dialing are handled inside +// the Entrypoint. +type entrypointHandler struct { + ep *Entrypoint +} + +// NewHandler wraps an Entrypoint as a handler.Handler. +// +// The returned handler's Init is a no-op (the Entrypoint is fully +// initialized by New()). Handle delegates to ep.Handle. +func NewHandler(ep *Entrypoint) handler.Handler { + return &entrypointHandler{ep: ep} +} + +func (h *entrypointHandler) Init(md md.Metadata) (err error) { + return +} + +func (h *entrypointHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error { + return h.ep.Handle(ctx, conn) +} + +// tcpListener wraps a raw net.Listener with GOST listener wrappers in +// the standard order: proxyproto → metrics → admission → conn limiter. +// +// The listener wrappers are applied in Init(), not at construction time, +// to match the GOST service lifecycle pattern. +type tcpListener struct { + ln net.Listener + options listener.Options +} + +// NewTCPListener creates a new TCP listener wrapper around a raw net.Listener. +// +// Imports the proxyproto, metrics, admission, and conn-limiter wrappers +// in the standard GOST listener order (see architecture docs). Options +// carry the service name, proxy protocol version, and optional admission/ +// conn-limit config. +func NewTCPListener(ln net.Listener, opts ...listener.Option) listener.Listener { + options := listener.Options{} + for _, opt := range opts { + opt(&options) + } + return &tcpListener{ + ln: ln, + options: options, + } +} + +// Init wraps the raw listener with GOST middleware layers: +// 1. proxyproto.WrapListener — PROXY protocol v1/v2 support +// 2. metrics.WrapListener — per-connection metrics +// 3. admission.WrapListener — IP admission control +// 4. climiter.WrapListener — concurrent connection limiting +// +// This ordering matches the GOST convention: metrics/observability +// outermost so dead-on-arrival connections are still counted. +func (l *tcpListener) Init(md md.Metadata) (err error) { + ln := l.ln + ln = proxyproto.WrapListener(l.options.ProxyProtocol, ln, 10*time.Second) + ln = metrics.WrapListener(l.options.Service, ln) + ln = admission.WrapListener(l.options.Service, l.options.Admission, ln) + ln = climiter.WrapListener(l.options.ConnLimiter, ln) + l.ln = ln + + return +} + +// Accept accepts the next connection from the wrapped listener. +// Implements listener.Listener. +func (l *tcpListener) Accept() (conn net.Conn, err error) { + return l.ln.Accept() +} + +// Addr returns the listener's network address. +// Implements listener.Listener. +func (l *tcpListener) Addr() net.Addr { + return l.ln.Addr() +} + +// Close closes the wrapped listener. +// Implements listener.Listener. +func (l *tcpListener) Close() error { + return l.ln.Close() +} \ No newline at end of file diff --git a/handler/tunnel/eprelay.go b/handler/tunnel/entrypoint/eprelay.go similarity index 89% rename from handler/tunnel/eprelay.go rename to handler/tunnel/entrypoint/eprelay.go index 36f2783b..66b33840 100644 --- a/handler/tunnel/eprelay.go +++ b/handler/tunnel/entrypoint/eprelay.go @@ -1,7 +1,8 @@ -package tunnel +package entrypoint import ( "context" + "fmt" "net" "strconv" "time" @@ -26,7 +27,7 @@ import ( // or bypass checks — the relay request already contains the tunnel ID. // Also unlike handleConnect, there is no local-vs-remote framing difference: // the entrypoint always writes StatusOK + address features regardless of node. -func (ep *entrypoint) handleConnect(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (err error) { +func (ep *Entrypoint) handleConnect(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (err error) { req := relay.Request{} if _, err := req.ReadFrom(conn); err != nil { return err @@ -65,18 +66,12 @@ func (ep *entrypoint) handleConnect(ctx context.Context, conn net.Conn, ro *xrec if tunnelID.IsZero() { resp.Status = relay.StatusBadRequest resp.WriteTo(conn) - return ErrTunnelID + return errBadTunnelID } ro.ClientID = tunnelID.String() - d := Dialer{ - pool: ep.pool, - retry: 3, - timeout: 15 * time.Second, - log: log, - } - cc, _, cid, err := d.Dial(ctx, network, tunnelID.String()) + cc, _, cid, err := ep.dialFn(ctx, network, tunnelID.String()) if err != nil { log.Error(err) resp.Status = relay.StatusServiceUnavailable @@ -120,4 +115,8 @@ func (ep *entrypoint) handleConnect(ctx context.Context, conn net.Conn, ro *xrec }).Debugf("%s >-< %s", conn.RemoteAddr(), cc.RemoteAddr()) return nil -} \ No newline at end of file +} + +// errBadTunnelID is returned when a relay request arrives at the entrypoint +// without a valid tunnel feature (zero tunnel ID). +var errBadTunnelID = fmt.Errorf("invalid tunnel ID") \ No newline at end of file diff --git a/handler/tunnel/eptls.go b/handler/tunnel/entrypoint/eptls.go similarity index 96% rename from handler/tunnel/eptls.go rename to handler/tunnel/entrypoint/eptls.go index 674d1076..cccf68c1 100644 --- a/handler/tunnel/eptls.go +++ b/handler/tunnel/entrypoint/eptls.go @@ -1,4 +1,4 @@ -package tunnel +package entrypoint import ( "bytes" @@ -28,7 +28,7 @@ import ( // 5. Parse ServerHello from the mux stream (for TLS recording). // 6. Write ServerHello bytes back to the public connection. // 7. Pipe(publicConn, muxStream) — bidirectional TLS passthrough. -func (ep *entrypoint) handleTLS(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { +func (ep *Entrypoint) handleTLS(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { buf := new(bytes.Buffer) clientHello, err := dissector.ParseClientHello(io.TeeReader(conn, buf)) if err != nil { diff --git a/handler/tunnel/epwebsocket.go b/handler/tunnel/entrypoint/epwebsocket.go similarity index 70% rename from handler/tunnel/epwebsocket.go rename to handler/tunnel/entrypoint/epwebsocket.go index f7086da3..329b2f6a 100644 --- a/handler/tunnel/epwebsocket.go +++ b/handler/tunnel/entrypoint/epwebsocket.go @@ -1,4 +1,4 @@ -package tunnel +package entrypoint import ( "bytes" @@ -14,7 +14,17 @@ import ( "golang.org/x/time/rate" ) -func (ep *entrypoint) sniffingWebsocketFrame(ctx context.Context, rw, cc io.ReadWriteCloser, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { +// sniffingWebsocketFrame copies WebSocket frames bidirectionally with +// optional frame-level recording. +// +// Two goroutines run concurrently — one for client→server frames, one for +// server→client frames. Each frame is recorded when the sample rate limiter +// allows. HTTP metadata in ro is cleared (ro.HTTP = nil) to prevent leaking +// HTTP request/response headers into WebSocket frame records. +// +// The function returns when either direction encounters an error (closing +// both sides). +func (ep *Entrypoint) sniffingWebsocketFrame(ctx context.Context, rw, cc io.ReadWriteCloser, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { errc := make(chan error, 2) sampleRate := ep.websocketSampleRate @@ -29,6 +39,7 @@ func (ep *entrypoint) sniffingWebsocketFrame(ctx context.Context, rw, cc io.Read ro2 := &xrecorder.HandlerRecorderObject{} *ro2 = *ro ro := ro2 + ro.HTTP = nil // WebSocket frames — don't leak HTTP metadata into the record limiter := rate.NewLimiter(rate.Limit(sampleRate), int(sampleRate)) @@ -55,6 +66,7 @@ func (ep *entrypoint) sniffingWebsocketFrame(ctx context.Context, rw, cc io.Read ro2 := &xrecorder.HandlerRecorderObject{} *ro2 = *ro ro := ro2 + ro.HTTP = nil // WebSocket frames — don't leak HTTP metadata into the record limiter := rate.NewLimiter(rate.Limit(sampleRate), int(sampleRate)) @@ -84,7 +96,14 @@ func (ep *entrypoint) sniffingWebsocketFrame(ctx context.Context, rw, cc io.Read return nil } -func (ep *entrypoint) copyWebsocketFrame(w io.Writer, r io.Reader, buf *bytes.Buffer, from string, ro *xrecorder.HandlerRecorderObject) (err error) { +// copyWebsocketFrame reads one WebSocket frame from r, records it in ro, +// and writes it to w. The frame payload is optionally buffered for recording +// when recorder options specify HTTPBody capture. +// +// The from parameter identifies the direction ("client" or "server") and +// determines which byte-counter (InputBytes vs OutputBytes) is incremented +// in the recorder object. +func (ep *Entrypoint) copyWebsocketFrame(w io.Writer, r io.Reader, buf *bytes.Buffer, from string, ro *xrecorder.HandlerRecorderObject) (err error) { fr := ws_util.Frame{} if _, err = fr.ReadFrom(r); err != nil { return err diff --git a/handler/tunnel/entrypointsvc.go b/handler/tunnel/entrypointsvc.go index d8f1a1ca..74878df6 100644 --- a/handler/tunnel/entrypointsvc.go +++ b/handler/tunnel/entrypointsvc.go @@ -1,19 +1,29 @@ package tunnel import ( + "context" "fmt" "net" - "net/http" "time" "github.com/go-gost/core/ingress" "github.com/go-gost/core/listener" + "github.com/go-gost/core/recorder" "github.com/go-gost/core/service" xnet "github.com/go-gost/x/internal/net" xrecorder "github.com/go-gost/x/recorder" xservice "github.com/go-gost/x/service" + + epkg "github.com/go-gost/x/handler/tunnel/entrypoint" ) +// initEntrypoints starts all configured entrypoint services. +// +// It starts the primary entrypoint (from the "entrypoint" metadata key) and +// any additional entrypoints (from the "entrypoints" metadata array). Each +// entrypoint is created via createEntrypointService and started in its own +// goroutine. Services are tracked in h.entrypoints for later cleanup in +// Close(). func (h *tunnelHandler) initEntrypoints() (err error) { if h.md.entryPoint != "" { svc, err := h.createEntrypointService(h.md.entryPoint, h.md.ingress) @@ -43,35 +53,55 @@ func (h *tunnelHandler) initEntrypoints() (err error) { return } -func (h *tunnelHandler) createEntrypointService(addr string, ingress ingress.Ingress) (service.Service, error) { - ep := &entrypoint{ - node: h.id, - service: h.options.Service, - pool: h.pool, - ingress: ingress, - sd: h.md.sd, - log: h.log.WithFields(map[string]any{ - "kind": "entrypoint", - }), - sniffingWebsocket: h.md.sniffingWebsocket, - websocketSampleRate: h.md.sniffingWebsocketSampleRate, - readTimeout: h.md.entryPointReadTimeout, - } - ep.transport = &http.Transport{ - DialContext: ep.dial, - IdleConnTimeout: 30 * time.Second, - ResponseHeaderTimeout: h.md.entryPointReadTimeout, - DisableKeepAlives: !h.md.entryPointKeepalive, - DisableCompression: !h.md.entryPointCompression, - } - - for _, ro := range h.options.Recorders { - if ro.Record == xrecorder.RecorderServiceHandler { - ep.recorder = ro +// createEntrypointService creates a GOST service wrapping an entrypoint +// listener and handler for the given address and ingress rule table. +// +// It constructs: +// 1. A Dialer populated from the tunnel handler's state (node, pool, sd) +// wrapped in a closure that implements epkg.DialFunc. +// 2. An epkg.Entrypoint via epkg.New() with all configured options. +// 3. A TCP listener via net.Listen and epkg.NewTCPListener. +// 4. An entrypoint handler via epkg.NewHandler. +// 5. A GOST service via xservice.NewService. +// +// The recorder object is selected from h.options.Recorders by matching +// RecorderServiceHandler record type. +func (h *tunnelHandler) createEntrypointService(addr string, ing ingress.Ingress) (service.Service, error) { + var ro recorder.RecorderObject + for _, r := range h.options.Recorders { + if r.Record == xrecorder.RecorderServiceHandler { + ro = r break } } + dialFn := func(ctx epkg.DialContext, network, tid string) (net.Conn, string, string, error) { + d := &Dialer{ + Node: h.id, + Pool: h.pool, + SD: h.md.sd, + Retry: 3, + Timeout: 15 * time.Second, + Log: h.log, + } + return d.Dial(ctx.(context.Context), network, tid) + } + + ep := epkg.New(&epkg.Config{ + Node: h.id, + Service: h.options.Service, + Ingress: ing, + SD: h.md.sd, + Logger: h.log.WithFields(map[string]any{"kind": "entrypoint"}), + Recorder: ro, + SniffingWebsocket: h.md.sniffingWebsocket, + WebsocketSampleRate: h.md.sniffingWebsocketSampleRate, + ReadTimeout: h.md.entryPointReadTimeout, + ProxyProtocol: h.md.entryPointProxyProtocol, + KeepAlive: h.md.entryPointKeepalive, + Compression: h.md.entryPointCompression, + }, dialFn) + network := "tcp" if xnet.IsIPv4(addr) { network = "tcp4" @@ -90,7 +120,8 @@ func (h *tunnelHandler) createEntrypointService(addr string, ingress ingress.Ing "handler": "tunnel-ep", "kind": "service", }) - epListener := newTCPListener(ln, + + epListener := epkg.NewTCPListener(ln, listener.AddrOption(addr), listener.ServiceOption(serviceName), listener.ProxyProtocolOption(h.md.entryPointProxyProtocol), @@ -101,9 +132,8 @@ func (h *tunnelHandler) createEntrypointService(addr string, ingress ingress.Ing if err = epListener.Init(nil); err != nil { return nil, err } - epHandler := &entrypointHandler{ - ep: ep, - } + + epHandler := epkg.NewHandler(ep) if err = epHandler.Init(nil); err != nil { return nil, err } diff --git a/handler/tunnel/eplistener.go b/handler/tunnel/eplistener.go deleted file mode 100644 index 8a4da45e..00000000 --- a/handler/tunnel/eplistener.go +++ /dev/null @@ -1,71 +0,0 @@ -package tunnel - -import ( - "context" - "net" - "time" - - "github.com/go-gost/core/handler" - "github.com/go-gost/core/listener" - md "github.com/go-gost/core/metadata" - admission "github.com/go-gost/x/admission/wrapper" - "github.com/go-gost/x/internal/net/proxyproto" - climiter "github.com/go-gost/x/limiter/conn/wrapper" - metrics "github.com/go-gost/x/metrics/wrapper" -) - -const ( - httpHeaderSID = "Gost-Sid" - httpHeaderForwardedNode = "Gost-Forwarded-Node" -) - -type tcpListener struct { - ln net.Listener - options listener.Options -} - -func newTCPListener(ln net.Listener, opts ...listener.Option) listener.Listener { - options := listener.Options{} - for _, opt := range opts { - opt(&options) - } - return &tcpListener{ - ln: ln, - options: options, - } -} - -func (l *tcpListener) Init(md md.Metadata) (err error) { - ln := l.ln - ln = proxyproto.WrapListener(l.options.ProxyProtocol, ln, 10*time.Second) - ln = metrics.WrapListener(l.options.Service, ln) - ln = admission.WrapListener(l.options.Service, l.options.Admission, ln) - ln = climiter.WrapListener(l.options.ConnLimiter, ln) - l.ln = ln - - return -} - -func (l *tcpListener) Accept() (conn net.Conn, err error) { - return l.ln.Accept() -} - -func (l *tcpListener) Addr() net.Addr { - return l.ln.Addr() -} - -func (l *tcpListener) Close() error { - return l.ln.Close() -} - -type entrypointHandler struct { - ep *entrypoint -} - -func (h *entrypointHandler) Init(md md.Metadata) (err error) { - return -} - -func (h *entrypointHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error { - return h.ep.Handle(ctx, conn) -} \ No newline at end of file diff --git a/handler/tunnel/handler.go b/handler/tunnel/handler.go index e0ad08a2..96d64295 100644 --- a/handler/tunnel/handler.go +++ b/handler/tunnel/handler.go @@ -76,20 +76,39 @@ import ( ) var ( - ErrBadVersion = errors.New("bad version") - ErrUnknownCmd = errors.New("unknown command") - ErrTunnelID = errors.New("invalid tunnel ID") + // ErrBadVersion is returned when the relay request has an unsupported + // protocol version. + ErrBadVersion = errors.New("bad version") + // ErrUnknownCmd is returned when the relay request command is not + // CmdConnect or CmdBind. + ErrUnknownCmd = errors.New("unknown command") + // ErrTunnelID is returned when the relay request has a zero/invalid + // tunnel ID feature. + ErrTunnelID = errors.New("invalid tunnel ID") + // ErrTunnelNotAvailable is returned when no local connector or SD + // service is available for the requested tunnel. ErrTunnelNotAvailable = errors.New("tunnel not available") - ErrUnauthorized = errors.New("unauthorized") - ErrTunnelRoute = errors.New("no route to host") - ErrPrivateTunnel = errors.New("private tunnel") + // ErrUnauthorized is returned when the relay request's user/pass + // authentication fails against the configured Auther. + ErrUnauthorized = errors.New("unauthorized") + // ErrTunnelRoute is returned when the tunnel route cannot be + // resolved (no ingress rule matches the host). + ErrTunnelRoute = errors.New("no route to host") + // ErrPrivateTunnel is returned when the resolved tunnel is private + // ($-prefixed) and the connection is from a public entrypoint. + ErrPrivateTunnel = errors.New("private tunnel") ) func init() { registry.HandlerRegistry().Register("tunnel", NewHandler) } +// tunnelHandler is the relay-based tunnel handler. It accepts relay-protocol +// connections from internal clients (CmdBind) and from public sources +// (CmdConnect), bridging them through a mux-based multiplex session. type tunnelHandler struct { + // id is a UUID-generated unique identifier for this handler instance, + // used to distinguish this node in multi-node deployments. id string options handler.Options pool *ConnectorPool @@ -101,6 +120,12 @@ type tunnelHandler struct { cancel context.CancelFunc } +// NewHandler creates a new tunnel handler. +// +// Registered as "tunnel" in the handler registry (init()). The handler +// processes relay-protocol requests and supports two commands: +// CmdConnect (forward a public connection through a tunnel stream) and +// CmdBind (register an internal client as a tunnel connector). func NewHandler(opts ...handler.Option) handler.Handler { options := handler.Options{} for _, opt := range opts { @@ -112,6 +137,11 @@ func NewHandler(opts ...handler.Option) handler.Handler { } } +// Init initializes the tunnel handler. +// +// It parses metadata, generates a unique node ID, creates the connector pool, +// starts all configured entrypoint services, sets up observer stats with a +// background goroutine, and initializes the traffic limiter. func (h *tunnelHandler) Init(md md.Metadata) (err error) { if err := h.parseMetadata(md); err != nil { return err @@ -151,6 +181,19 @@ func (h *tunnelHandler) Init(md md.Metadata) (err error) { return nil } +// Handle processes an incoming relay-protocol connection. +// +// The connection is expected to start with a relay.Request frame. The handler +// parses the request, extracts authentication, addresses, network type, and +// tunnel ID, then dispatches to handleConnect (CmdConnect) or handleBind +// (CmdBind). +// +// Rate limiting is checked before reading the request. A read deadline is +// applied during the initial relay frame read and cleared afterwards. +// +// On error, a relay.Response with the appropriate error status is written +// before returning. The caller is responsible for closing conn on success +// (the CmdConnect path defers conn.Close() internally). func (h *tunnelHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) { start := time.Now() @@ -270,7 +313,10 @@ func (h *tunnelHandler) Handle(ctx context.Context, conn net.Conn, opts ...handl } } -// Close implements io.Closer interface. +// Close implements io.Closer. +// +// It closes all entrypoint services, the connector pool (which closes all +// tunnels and connectors), and cancels the observer stats goroutine. func (h *tunnelHandler) Close() error { for _, ep := range h.entrypoints { ep.Close() @@ -284,6 +330,8 @@ func (h *tunnelHandler) Close() error { return nil } +// checkRateLimit returns false if the connection source IP exceeds the +// configured rate limiter budget. Returns true if no rate limiter is set. func (h *tunnelHandler) checkRateLimit(addr net.Addr) bool { if h.options.RateLimiter == nil { return true @@ -296,6 +344,13 @@ func (h *tunnelHandler) checkRateLimit(addr net.Addr) bool { return true } +// observeStats is a background goroutine that periodically flushes connection +// stats events to the configured observer. +// +// On observe failure, events are buffered and retried on the next tick. +// On retry success, new events from the current tick are also flushed +// (fall-through via the pending-events block). On persistent failure, +// new events are skipped until the pending batch is accepted. func (h *tunnelHandler) observeStats(ctx context.Context) { if h.options.Observer == nil { return diff --git a/handler/tunnel/handler_test.go b/handler/tunnel/handler_test.go index 0e17ac08..f58628af 100644 --- a/handler/tunnel/handler_test.go +++ b/handler/tunnel/handler_test.go @@ -10,12 +10,16 @@ import ( "time" "github.com/go-gost/core/handler" + core_rate "github.com/go-gost/core/limiter/rate" "github.com/go-gost/core/observer" "github.com/go-gost/core/observer/stats" coremeta "github.com/go-gost/core/metadata" "github.com/go-gost/relay" + rate_limiter "github.com/go-gost/x/limiter/rate" stats_util "github.com/go-gost/x/internal/util/stats" mdx "github.com/go-gost/x/metadata" + + epkg "github.com/go-gost/x/handler/tunnel/entrypoint" ) func newTestMetadata() coremeta.Metadata { @@ -61,31 +65,6 @@ func (c *fakeConn) SetReadDeadline(t time.Time) error { return nil } -func buildRelayConnectRequest(t *testing.T, tid relay.TunnelID, src, dst string) []byte { - t.Helper() - req := relay.Request{ - Version: relay.Version1, - Cmd: relay.CmdConnect, - } - if src != "" { - af := &relay.AddrFeature{} - af.ParseFrom(src) - req.Features = append(req.Features, af) - } - if dst != "" { - af := &relay.AddrFeature{} - af.ParseFrom(dst) - req.Features = append(req.Features, af) - } - req.Features = append(req.Features, &relay.TunnelFeature{ID: tid}) - var buf bytes.Buffer - _, err := req.WriteTo(&buf) - if err != nil { - t.Fatal(err) - } - return buf.Bytes() -} - func buildRelayBindRequest(t *testing.T, tid relay.TunnelID, network, addr string) []byte { t.Helper() req := relay.Request{ @@ -378,16 +357,104 @@ func TestHandler_initEntrypoints(t *testing.T) { // identifies the protocol (relay/TLS/HTTP) from the first byte of a connection. func TestEntrypoint_dial_NoIngress(t *testing.T) { - ep := &entrypoint{ - node: "node1", - pool: NewConnectorPool("node1"), - log: testLogger(), - } - defer ep.pool.Close() + conn := &fakeConn{buf: []byte("GET / HTTP/1.1\r\nHost: example.com\r\n\r\n")} + pool := NewConnectorPool("node1") + defer pool.Close() + log := testLogger() - _, err := ep.dial(context.Background(), "tcp", "example.com") - if err == nil { - t.Error("expected error without ingress") + dialFn := func(ctx epkg.DialContext, network, tid string) (net.Conn, string, string, error) { + return nil, "", "", errors.New("should not be called") + } + ep := epkg.New(&epkg.Config{ + Node: "node1", + Logger: log, + }, dialFn) + + err := ep.Handle(context.Background(), conn) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !bytes.Contains(conn.writeBuf, []byte("502")) && !bytes.Contains(conn.writeBuf, []byte("Bad Gateway")) { + t.Errorf("expected 502 response, got: %s", conn.writeBuf) + } +} + +// TestHandler_Handle_RateLimitExceeded tests that Handle returns ErrRateLimit +// when the rate limiter rejects the connection. +// fakeRateLimiter implements core_rate.RateLimiter for testing. +type fakeRateLimiter struct { + allow bool +} + +func (r *fakeRateLimiter) Limiter(key string) core_rate.Limiter { + return r +} + +func (r *fakeRateLimiter) Allow(n int) bool { return r.allow } +func (r *fakeRateLimiter) Limit() float64 { return 0 } + +// deadlineConn wraps fakeConn and tracks calls to SetReadDeadline. +type deadlineConn struct { + *fakeConn + deadlineSet bool +} + +func (c *deadlineConn) SetReadDeadline(t time.Time) error { + c.deadlineSet = true + return nil +} + +// TestHandler_Handle_ReadTimeout tests that the read deadline is applied +// before reading the relay request. +func TestHandler_Handle_ReadTimeout(t *testing.T) { + req := relay.Request{ + Version: relay.Version1, + Cmd: relay.CmdConnect, + } + req.Features = append(req.Features, &relay.TunnelFeature{ + ID: newTestTunnelID(t), + }) + var buf bytes.Buffer + req.WriteTo(&buf) + + dconn := &deadlineConn{ + fakeConn: &fakeConn{buf: buf.Bytes()}, + } + + h := newHandlerWithLogger(t) + defer h.Close() + h.md.readTimeout = 5 * time.Second + + _ = h.Handle(context.Background(), dconn) + if !dconn.deadlineSet { + t.Error("expected SetReadDeadline to be called") + } +} + +// TestHandler_Handle_RateLimitExceeded tests that Handle returns ErrRateLimit +// when the rate limiter rejects the connection. +func TestHandler_Handle_RateLimitExceeded(t *testing.T) { + h := newHandlerWithLogger(t) + defer h.Close() + + req := relay.Request{ + Version: relay.Version1, + Cmd: relay.CmdConnect, + } + req.Features = append(req.Features, &relay.TunnelFeature{ + ID: newTestTunnelID(t), + }) + var buf bytes.Buffer + req.WriteTo(&buf) + + // Use a rate limiter that always rejects + rl := &fakeRateLimiter{allow: false} + h.options.RateLimiter = rl + + conn := &fakeConn{buf: buf.Bytes()} + err := h.Handle(context.Background(), conn) + if err != rate_limiter.ErrRateLimit { + t.Errorf("expected ErrRateLimit, got %v", err) } } diff --git a/handler/tunnel/helpers_test.go b/handler/tunnel/helpers_test.go index 6cda6240..3b181491 100644 --- a/handler/tunnel/helpers_test.go +++ b/handler/tunnel/helpers_test.go @@ -1,9 +1,18 @@ package tunnel import ( + "os" + "testing" + "github.com/go-gost/core/logger" ) +// TestMain sets up a non-nil default logger for all tests in this package. +func TestMain(m *testing.M) { + logger.SetDefault(nopLogger{}) + os.Exit(m.Run()) +} + type nopLogger struct { logger.Logger } diff --git a/handler/tunnel/id.go b/handler/tunnel/id.go index 15efb160..06a56fb7 100644 --- a/handler/tunnel/id.go +++ b/handler/tunnel/id.go @@ -9,10 +9,10 @@ import ( "github.com/google/uuid" ) -// parseTunnelID parses a tunnel ID from a string. +// ParseTunnelID parses a tunnel ID from a string. // If s is empty or contains an invalid UUID, the returned tunnel ID is zero // (callers must check IsZero). A leading '$' prefix marks the tunnel as private. -func parseTunnelID(s string) (tid relay.TunnelID) { +func ParseTunnelID(s string) (tid relay.TunnelID) { if s == "" { return } diff --git a/handler/tunnel/id_test.go b/handler/tunnel/id_test.go index c46e507e..c8f9e2c9 100644 --- a/handler/tunnel/id_test.go +++ b/handler/tunnel/id_test.go @@ -9,14 +9,14 @@ import ( func TestParseTunnelID(t *testing.T) { t.Run("empty string", func(t *testing.T) { - tid := parseTunnelID("") + tid := ParseTunnelID("") if !tid.IsZero() { t.Error("expected zero tunnel ID for empty string") } }) t.Run("valid uuid", func(t *testing.T) { - tid := parseTunnelID("6ba7b810-9dad-11d1-80b4-00c04fd430c8") + tid := ParseTunnelID("6ba7b810-9dad-11d1-80b4-00c04fd430c8") if tid.IsZero() { t.Error("expected non-zero tunnel ID") } @@ -26,7 +26,7 @@ func TestParseTunnelID(t *testing.T) { }) t.Run("valid uuid with $ prefix", func(t *testing.T) { - tid := parseTunnelID("$6ba7b810-9dad-11d1-80b4-00c04fd430c8") + tid := ParseTunnelID("$6ba7b810-9dad-11d1-80b4-00c04fd430c8") if tid.IsZero() { t.Error("expected non-zero tunnel ID") } @@ -36,14 +36,14 @@ func TestParseTunnelID(t *testing.T) { }) t.Run("invalid uuid", func(t *testing.T) { - tid := parseTunnelID("not-a-uuid") + tid := ParseTunnelID("not-a-uuid") if !tid.IsZero() { t.Error("expected zero tunnel ID for invalid input") } }) t.Run("invalid uuid with $ prefix", func(t *testing.T) { - tid := parseTunnelID("$not-a-uuid") + tid := ParseTunnelID("$not-a-uuid") if !tid.IsZero() { t.Error("expected zero tunnel ID for invalid input") } @@ -52,7 +52,7 @@ func TestParseTunnelID(t *testing.T) { func TestParseTunnelID_PrivateMarker(t *testing.T) { t.Run("dollar not at start", func(t *testing.T) { - tid := parseTunnelID("6ba7b810-9dad-11d1-80b4-00c04fd430c8") + tid := ParseTunnelID("6ba7b810-9dad-11d1-80b4-00c04fd430c8") if tid.IsPrivate() { t.Error("expected non-private when $ is not at start") } @@ -60,17 +60,17 @@ func TestParseTunnelID_PrivateMarker(t *testing.T) { } func TestParseTunnelID_RoundTrip(t *testing.T) { - original := parseTunnelID("6ba7b810-9dad-11d1-80b4-00c04fd430c8") + original := ParseTunnelID("6ba7b810-9dad-11d1-80b4-00c04fd430c8") t.Logf("original string: %s", original.String()) - reparsed := parseTunnelID(original.String()) + reparsed := ParseTunnelID(original.String()) if !original.Equal(reparsed) { t.Errorf("round-trip failed: original=%v reparsed=%v", original, reparsed) } } func TestParseTunnelID_PrivateRoundTrip(t *testing.T) { - original := parseTunnelID("$6ba7b810-9dad-11d1-80b4-00c04fd430c8") + original := ParseTunnelID("$6ba7b810-9dad-11d1-80b4-00c04fd430c8") if !original.IsPrivate() { t.Fatal("expected private tunnel ID") } @@ -79,19 +79,19 @@ func TestParseTunnelID_PrivateRoundTrip(t *testing.T) { // The private flag is carried in the struct, not in the string representation. // Reparsing from String() gives a non-private ID because the $ prefix is not // part of the relay.TunnelID.String() output. - reparsed := parseTunnelID(original.String()) + reparsed := ParseTunnelID(original.String()) if !original.Equal(reparsed) { t.Errorf("round-trip failed: original=%v reparsed=%v", original, reparsed) } } func TestParseTunnelID_RelayCompatibility(t *testing.T) { - id := parseTunnelID("6ba7b810-9dad-11d1-80b4-00c04fd430c8") + id := ParseTunnelID("6ba7b810-9dad-11d1-80b4-00c04fd430c8") u, _ := uuid.Parse("6ba7b810-9dad-11d1-80b4-00c04fd430c8") var raw [16]byte copy(raw[:], u[:]) expected := relay.NewTunnelID(raw[:]) if !id.Equal(expected) { - t.Error("parseTunnelID result should match relay.NewTunnelID") + t.Error("ParseTunnelID result should match relay.NewTunnelID") } } \ No newline at end of file diff --git a/handler/tunnel/metadata.go b/handler/tunnel/metadata.go index 4b0fa75d..73793d98 100644 --- a/handler/tunnel/metadata.go +++ b/handler/tunnel/metadata.go @@ -17,9 +17,15 @@ import ( ) const ( + // defaultTTL is the default time-to-live for a Tunnel's clean goroutine. + // Every tick removes closed connectors and renews SD registrations. defaultTTL = 15 * time.Second ) +// metadata holds the parsed configuration for the tunnel handler. +// +// All fields are populated from the handler's metadata map in parseMetadata. +// Most fields have matching "entrypoint.*" config keys. type metadata struct { // readTimeout is the deadline for reading the initial relay protocol // handshake from the client connection. The deadline is cleared @@ -27,18 +33,35 @@ type metadata struct { // transfer. 0 or negative means no timeout is applied. readTimeout time.Duration + // entrypoints holds additional entrypoint configurations from the + // "entrypoints" metadata key (JSON array of {Addr, Ingress}). entrypoints []entrypointConfig - entryPoint string + // entryPoint is the TCP address for the primary entrypoint listener + // (from the "entrypoint" metadata key). + entryPoint string + // entryPointID is the parsed tunnel ID that identifies public + // entrypoint visitors in handleConnect. entryPointID relay.TunnelID entryPointProxyProtocol int entryPointKeepalive bool entryPointCompression bool - entryPointReadTimeout time.Duration // deadline for reading upstream HTTP response headers in the entrypoint's http.Transport.ResponseHeaderTimeout. Also passed as readTimeout to entrypoint dialer for SetReadDeadline on upstream conn. 0 or negative defaults to 15s. + // entryPointReadTimeout is the deadline for reading upstream HTTP + // response headers in the entrypoint's http.Transport.ResponseHeaderTimeout. + // Also passed as readTimeout to the entrypoint dialer for SetReadDeadline + // on the upstream connection. 0 or negative defaults to 15s. + entryPointReadTimeout time.Duration + // sniffingWebsocket enables WebSocket frame-level recording in the + // entrypoint HTTP handler. sniffingWebsocket bool + // sniffingWebsocketSampleRate controls the rate for WebSocket frame + // recording. 0 defaults to DefaultSampleRate. sniffingWebsocketSampleRate float64 + // directTunnel when true allows direct tunnel connections without + // ingress routing (the tunnel ID from the relay request is used directly). directTunnel bool + // tunnelTTL is the TTL for the Tunnel's clean goroutine. Defaults to defaultTTL. tunnelTTL time.Duration ingress ingress.Ingress sd sd.SD @@ -55,7 +78,7 @@ func (h *tunnelHandler) parseMetadata(md mdata.Metadata) (err error) { h.md.readTimeout = mdutil.GetDuration(md, "readTimeout") h.md.entryPoint = mdutil.GetString(md, "entrypoint") - h.md.entryPointID = parseTunnelID(mdutil.GetString(md, "entrypoint.id")) + h.md.entryPointID = ParseTunnelID(mdutil.GetString(md, "entrypoint.id")) h.md.entryPointProxyProtocol = mdutil.GetInt(md, "entrypoint.ProxyProtocol") h.md.entryPointKeepalive = mdutil.GetBool(md, "entrypoint.keepalive") @@ -164,8 +187,15 @@ func (h *tunnelHandler) parseMetadata(md mdata.Metadata) (err error) { return } +// entrypointConfig holds the configuration for a single additional entrypoint +// beyond the primary entryPoint. Multiple entrypoints are configured via the +// "entrypoints" metadata key as a JSON array. type entrypointConfig struct { + // Name is an optional label for this entrypoint (used in logs). Name string + // Addr is the TCP address to listen on (e.g. "0.0.0.0:80"). Addr string + // Ingress is the ingress rule table for this entrypoint. If nil, + // the handler's default ingress is used. Ingress ingress.Ingress } diff --git a/handler/tunnel/metadata_test.go b/handler/tunnel/metadata_test.go new file mode 100644 index 00000000..987b3545 --- /dev/null +++ b/handler/tunnel/metadata_test.go @@ -0,0 +1,499 @@ +package tunnel + +import ( + "context" + "testing" + "time" + + mdx "github.com/go-gost/x/metadata" +) + +func TestParseMetadata_Defaults(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{}) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if h.md.readTimeout != 0 { + t.Errorf("expected readTimeout 0, got %v", h.md.readTimeout) + } + if h.md.entryPoint != "" { + t.Errorf("expected empty entryPoint, got %q", h.md.entryPoint) + } + if h.md.tunnelTTL != defaultTTL { + t.Errorf("expected tunnelTTL %v, got %v", defaultTTL, h.md.tunnelTTL) + } + if h.md.entryPointReadTimeout != 15*time.Second { + t.Errorf("expected entryPointReadTimeout 15s, got %v", h.md.entryPointReadTimeout) + } + if h.md.muxCfg.Version != 2 { + t.Errorf("expected mux version 2, got %d", h.md.muxCfg.Version) + } + if h.md.muxCfg.MaxStreamBuffer != 1048576 { + t.Errorf("expected max stream buffer 1048576, got %d", h.md.muxCfg.MaxStreamBuffer) + } + if h.md.observerPeriod != 5*time.Second { + t.Errorf("expected observerPeriod 5s, got %v", h.md.observerPeriod) + } + if h.md.directTunnel { + t.Error("expected directTunnel false") + } + if h.md.sniffingWebsocket { + t.Error("expected sniffingWebsocket false") + } + if h.md.ingress != nil { + t.Error("expected nil ingress") + } + if h.md.sd != nil { + t.Error("expected nil sd") + } +} + +func TestParseMetadata_ReadTimeout(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "readTimeout": "3s", + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if h.md.readTimeout != 3*time.Second { + t.Errorf("expected readTimeout 3s, got %v", h.md.readTimeout) + } +} + +func TestParseMetadata_Entrypoint(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "entrypoint": ":8080", + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if h.md.entryPoint != ":8080" { + t.Errorf("expected entryPoint :8080, got %q", h.md.entryPoint) + } +} + +func TestParseMetadata_DirectTunnel(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "tunnel.direct": true, + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !h.md.directTunnel { + t.Error("expected directTunnel true") + } +} + +func TestParseMetadata_TunnelTTL(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "tunnel.ttl": "30s", + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if h.md.tunnelTTL != 30*time.Second { + t.Errorf("expected tunnelTTL 30s, got %v", h.md.tunnelTTL) + } +} + +func TestParseMetadata_TunnelTTL_ZeroDefaults(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "tunnel.ttl": 0, + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if h.md.tunnelTTL != defaultTTL { + t.Errorf("expected tunnelTTL %v, got %v", defaultTTL, h.md.tunnelTTL) + } +} + +func TestParseMetadata_Sniffing(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "sniffing.websocket": true, + "sniffing.websocket.sampleRate": 0.5, + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !h.md.sniffingWebsocket { + t.Error("expected sniffingWebsocket true") + } + if h.md.sniffingWebsocketSampleRate != 0.5 { + t.Errorf("expected sampleRate 0.5, got %f", h.md.sniffingWebsocketSampleRate) + } +} + +func TestParseMetadata_ObserverPeriod(t *testing.T) { + tests := []struct { + name string + input string + expect time.Duration + }{ + {"custom period", "10s", 10 * time.Second}, + {"minimum clamped to 1s", "100ms", time.Second}, + {"zero defaults to 5s", "", 5 * time.Second}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + h := &tunnelHandler{} + data := map[string]any{} + if tt.input != "" { + data["observePeriod"] = tt.input + } + md := mdx.NewMetadata(data) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if h.md.observerPeriod != tt.expect { + t.Errorf("expected %v, got %v", tt.expect, h.md.observerPeriod) + } + }) + } +} + +func TestParseMetadata_ObserverPeriod_AlternateKeys(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "observer.period": "3s", + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if h.md.observerPeriod != 3*time.Second { + t.Errorf("expected 3s, got %v", h.md.observerPeriod) + } +} + +func TestParseMetadata_ObserverResetTraffic(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "observer.resetTraffic": true, + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !h.md.observerResetTraffic { + t.Error("expected observerResetTraffic true") + } +} + +func TestParseMetadata_MuxConfig(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "mux.version": 1, + "mux.keepaliveInterval": "10s", + "mux.keepaliveDisabled": true, + "mux.keepaliveTimeout": "30s", + "mux.maxFrameSize": 4096, + "mux.maxReceiveBuffer": 8192, + "mux.maxStreamBuffer": 65536, + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if h.md.muxCfg.Version != 1 { + t.Errorf("expected mux version 1, got %d", h.md.muxCfg.Version) + } + if h.md.muxCfg.KeepAliveInterval != 10*time.Second { + t.Errorf("expected keepalive interval 10s, got %v", h.md.muxCfg.KeepAliveInterval) + } + if !h.md.muxCfg.KeepAliveDisabled { + t.Error("expected keepalive disabled") + } + if h.md.muxCfg.KeepAliveTimeout != 30*time.Second { + t.Errorf("expected keepalive timeout 30s, got %v", h.md.muxCfg.KeepAliveTimeout) + } + if h.md.muxCfg.MaxFrameSize != 4096 { + t.Errorf("expected max frame size 4096, got %d", h.md.muxCfg.MaxFrameSize) + } + if h.md.muxCfg.MaxReceiveBuffer != 8192 { + t.Errorf("expected max receive buffer 8192, got %d", h.md.muxCfg.MaxReceiveBuffer) + } + if h.md.muxCfg.MaxStreamBuffer != 65536 { + t.Errorf("expected max stream buffer 65536, got %d", h.md.muxCfg.MaxStreamBuffer) + } +} + +func TestParseMetadata_MuxDefaultsNoVersion(t *testing.T) { + h := &tunnelHandler{} + // When no mux config is provided, defaults should be applied. + md := mdx.NewMetadata(map[string]any{}) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if h.md.muxCfg.Version != 2 { + t.Errorf("expected default version 2, got %d", h.md.muxCfg.Version) + } + if h.md.muxCfg.MaxStreamBuffer != 1048576 { + t.Errorf("expected default max stream buffer 1048576, got %d", h.md.muxCfg.MaxStreamBuffer) + } +} + +func TestParseMetadata_TunnelRules(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "tunnel": "example.com:6ba7b810-9dad-11d1-80b4-00c04fd430c8,app.example.com:7f2c3d4e-5a6b-7c8d-9e0f-1a2b3c4d5e6f", + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if h.md.ingress == nil { + t.Fatal("expected ingress to be created from tunnel rules") + } + + // Wait briefly for the ingress periodReload goroutine to populate rules. + time.Sleep(50 * time.Millisecond) + + // Check that rules were parsed correctly + rule1 := h.md.ingress.GetRule(context.Background(), "example.com") + if rule1 == nil { + t.Fatal("expected rule for example.com") + } + if rule1.Hostname != "example.com" { + t.Errorf("expected hostname example.com, got %q", rule1.Hostname) + } + if rule1.Endpoint != "6ba7b810-9dad-11d1-80b4-00c04fd430c8" { + t.Errorf("expected endpoint 6ba7b810..., got %q", rule1.Endpoint) + } + + rule2 := h.md.ingress.GetRule(context.Background(), "app.example.com") + if rule2 == nil { + t.Fatal("expected rule for app.example.com") + } + if rule2.Hostname != "app.example.com" { + t.Errorf("expected hostname app.example.com, got %q", rule2.Hostname) + } + if rule2.Endpoint != "7f2c3d4e-5a6b-7c8d-9e0f-1a2b3c4d5e6f" { + t.Errorf("expected endpoint 7f2c3d4e..., got %q", rule2.Endpoint) + } +} + +func TestParseMetadata_TunnelRules_Malformed(t *testing.T) { + // Test that malformed tunnel rules are silently skipped. + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "tunnel": "valid.example.com:6ba7b810-9dad-11d1-80b4-00c04fd430c8,badrule", + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if h.md.ingress == nil { + t.Fatal("expected ingress to be created with at least valid rules") + } + + // Wait briefly for the ingress periodReload goroutine to populate rules. + time.Sleep(50 * time.Millisecond) + + // Valid rule should exist + rule := h.md.ingress.GetRule(context.Background(), "valid.example.com") + if rule == nil { + t.Fatal("expected rule for valid.example.com") + } + + // Malformed rule should have been skipped + if len(h.md.entrypoints) != 0 { + t.Errorf("expected 0 entrypoints, got %d", len(h.md.entrypoints)) + } +} + +func TestParseMetadata_EntrypointsJSON(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "entrypoints": []map[string]any{ + {"Addr": ":8080"}, + {"Addr": ":9090"}, + }, + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(h.md.entrypoints) != 2 { + t.Fatalf("expected 2 entrypoints, got %d", len(h.md.entrypoints)) + } + if h.md.entrypoints[0].Addr != ":8080" { + t.Errorf("expected entrypoints[0] addr :8080, got %q", h.md.entrypoints[0].Addr) + } + if h.md.entrypoints[1].Addr != ":9090" { + t.Errorf("expected entrypoints[1] addr :9090, got %q", h.md.entrypoints[1].Addr) + } +} + +func TestParseMetadata_EntrypointsJSON_EmptyAddrSkipped(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "entrypoints": []map[string]any{ + {"Addr": ""}, + {"Addr": ":8080"}, + }, + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(h.md.entrypoints) != 1 { + t.Errorf("expected 1 entrypoint (empty addr skipped), got %d", len(h.md.entrypoints)) + } + if h.md.entrypoints[0].Addr != ":8080" { + t.Errorf("expected entrypoint addr :8080, got %q", h.md.entrypoints[0].Addr) + } +} + +func TestParseMetadata_LimiterIntervals(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "limiter.refreshInterval": "60s", + "limiter.cleanupInterval": "120s", + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if h.md.limiterRefreshInterval != 60*time.Second { + t.Errorf("expected limiterRefreshInterval 60s, got %v", h.md.limiterRefreshInterval) + } + if h.md.limiterCleanupInterval != 120*time.Second { + t.Errorf("expected limiterCleanupInterval 120s, got %v", h.md.limiterCleanupInterval) + } +} + +func TestParseMetadata_EntrypointProxyProtocol(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "entrypoint.ProxyProtocol": 2, + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if h.md.entryPointProxyProtocol != 2 { + t.Errorf("expected proxy protocol 2, got %d", h.md.entryPointProxyProtocol) + } +} + +func TestParseMetadata_EntrypointKeepalive(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "entrypoint.keepalive": true, + "entrypoint.compression": true, + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !h.md.entryPointKeepalive { + t.Error("expected entryPointKeepalive true") + } + if !h.md.entryPointCompression { + t.Error("expected entryPointCompression true") + } +} + +func TestParseMetadata_EntrypointReadTimeout_ZeroDefaults(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{}) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if h.md.entryPointReadTimeout != 15*time.Second { + t.Errorf("expected default entryPointReadTimeout 15s, got %v", h.md.entryPointReadTimeout) + } +} + +func TestParseMetadata_EmptyMetadata(t *testing.T) { + // With nil metadata, parseMetadata should not panic. + h := &tunnelHandler{} + err := h.parseMetadata(nil) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + // Defaults should still apply + if h.md.muxCfg.Version != 2 { + t.Errorf("expected default version 2, got %d", h.md.muxCfg.Version) + } + if h.md.tunnelTTL != defaultTTL { + t.Errorf("expected default TTL %v, got %v", defaultTTL, h.md.tunnelTTL) + } +} + +// TestParseMetadata_EntrypointID tests parsing of the entrypointID metadata key. +func TestParseMetadata_EntrypointID(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "entrypoint.id": "6ba7b810-9dad-11d1-80b4-00c04fd430c8", + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + expected := ParseTunnelID("6ba7b810-9dad-11d1-80b4-00c04fd430c8") + if !h.md.entryPointID.Equal(expected) { + t.Errorf("expected entryPointID %v, got %v", expected, h.md.entryPointID) + } +} + +// TestParseMetadata_TunnelRules_Empty tests that empty tunnel rules don't +// create an ingress. +func TestParseMetadata_TunnelRules_Empty(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "tunnel": "", + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if h.md.ingress != nil { + t.Error("expected nil ingress for empty tunnel rules") + } +} + +// TestParseMetadata_EntrypointsJSON_WithIngressName tests that entrypoints +// JSON can reference named ingress objects. Without a registry entry, it +// should fall back to the handler's default ingress. +func TestParseMetadata_EntrypointsJSON_IngressFallback(t *testing.T) { + h := &tunnelHandler{} + md := mdx.NewMetadata(map[string]any{ + "entrypoints": []map[string]any{ + {"Addr": ":8080", "Ingress": "nonexistent"}, + }, + }) + err := h.parseMetadata(md) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(h.md.entrypoints) != 1 { + t.Fatalf("expected 1 entrypoint, got %d", len(h.md.entrypoints)) + } + // The registry always returns a non-nil ingressWrapper for non-empty names, + // even if the name is not registered (hot-reload design). + if h.md.entrypoints[0].Ingress == nil { + t.Error("expected non-nil ingress wrapper from registry") + } +} \ No newline at end of file diff --git a/handler/tunnel/tunnel.go b/handler/tunnel/tunnel.go index 99055a7f..0f21f507 100644 --- a/handler/tunnel/tunnel.go +++ b/handler/tunnel/tunnel.go @@ -35,13 +35,17 @@ const ( // ConnectorPool.closeIdles on a 15-minute ticker). // - Close() closes all connectors and signals shutdown. type Tunnel struct { - node string + // node is the tunnel handler node ID that owns this tunnel. + node string + // id is the relay tunnel ID shared by all connectors in this group. id relay.TunnelID connectors []*Connector - t time.Time - close chan struct{} - mu sync.RWMutex - ttl time.Duration + // t is the creation timestamp — used for debugging/lifecycle tracking. + t time.Time + close chan struct{} + mu sync.RWMutex + // ttl is the interval between clean() ticks. Defaults to defaultTTL. + ttl time.Duration } func NewTunnel(node string, tid relay.TunnelID, ttl time.Duration) *Tunnel { diff --git a/handler/tunnel/tunnel_test.go b/handler/tunnel/tunnel_test.go index 42c8da02..a20d652c 100644 --- a/handler/tunnel/tunnel_test.go +++ b/handler/tunnel/tunnel_test.go @@ -14,7 +14,7 @@ import ( func newTestTunnelID(t *testing.T) relay.TunnelID { t.Helper() - return parseTunnelID("6ba7b810-9dad-11d1-80b4-00c04fd430c8") + return ParseTunnelID("6ba7b810-9dad-11d1-80b4-00c04fd430c8") } func newTestConnectorID(t *testing.T, udp bool, weight uint8) relay.ConnectorID {