diff --git a/internal/web/runtime/tls_client.go b/internal/web/runtime/tls_client.go index 0e3276979..77073856c 100644 --- a/internal/web/runtime/tls_client.go +++ b/internal/web/runtime/tls_client.go @@ -6,6 +6,7 @@ import ( "crypto/tls" "encoding/base64" "encoding/hex" + "fmt" "net/http" "strings" "sync" @@ -132,22 +133,104 @@ func (t *credentialRotatingTransport) CloseIdleConnections() { current.CloseIdleConnections() } -// defaultNodeHTTPClient reaches nodes trusting the system CA store ("verify" -// mode or plain http); shared so connections pool across nodes. -var defaultNodeHTTPClient = &http.Client{ - Transport: &http.Transport{ - MaxIdleConns: 64, - MaxIdleConnsPerHost: 4, +// The global cap must exceed the fleet size: below it Go closes a node's +// connection before its next heartbeat, costing a handshake every tick. +const ( + maxIdleNodeConns = 512 + maxIdleNodeConnsPerHost = 8 +) + +func newNodeTransport(tlsCfg *tls.Config) *http.Transport { + return &http.Transport{ + MaxIdleConns: maxIdleNodeConns, + MaxIdleConnsPerHost: maxIdleNodeConnsPerHost, IdleConnTimeout: 60 * time.Second, DialContext: netsafe.SSRFGuardedDialContext, - }, + TLSClientConfig: tlsCfg, + } } +// defaultNodeHTTPClient reaches nodes trusting the system CA store ("verify" +// mode or plain http); shared so connections pool across nodes. +var defaultNodeHTTPClient = &http.Client{Transport: newNodeTransport(nil)} + +// nodeClients caches one client per node: heartbeat and traffic sync reach it +// every few seconds, and a rebuilt client would open its own empty pool. +type nodeClientEntry struct { + nodeID int + client *http.Client +} + +var ( + nodeClientsMu sync.Mutex + nodeClientsCache = map[string]nodeClientEntry{} +) + +// nodeClientIdentity covers everything that decides how the node is trusted; the +// proxy URL is a variant of it, so it stays out of the identity itself. +func nodeClientIdentity(n *model.Node, mode string) string { + return fmt.Sprintf("%d|%s|%s|%s|%d|%s", n.Id, mode, n.Scheme, n.Address, n.Port, n.PinnedCertSha256) +} + +// dropNodeClients discards every cached client of one node except keep, so a +// node never holds more than the variant it is using now. Callers hold the lock. +func dropNodeClients(nodeID int, keep string) { + for key, entry := range nodeClientsCache { + if entry.nodeID != nodeID || key == keep { + continue + } + entry.client.CloseIdleConnections() + delete(nodeClientsCache, key) + } +} + +// HTTPClientForNode returns the pooled client for n, building it on first use +// and whenever the node's identity or TLS material changes. func HTTPClientForNode(n *model.Node, proxyURL string) (*http.Client, error) { mode := n.TlsVerifyMode if mode == "" { mode = "verify" } + if mode == "verify" || n.Scheme == "http" { + // Shared across nodes and not node-specific: nothing to key on. + if proxyURL == "" { + nodeClientsMu.Lock() + dropNodeClients(n.Id, "") + nodeClientsMu.Unlock() + return defaultNodeHTTPClient, nil + } + } + + identity := nodeClientIdentity(n, mode) + key := identity + "|" + proxyURL + nodeClientsMu.Lock() + if entry, ok := nodeClientsCache[key]; ok { + nodeClientsMu.Unlock() + return entry.client, nil + } + nodeClientsMu.Unlock() + + client, err := buildNodeHTTPClient(n, mode, proxyURL) + if err != nil { + return nil, err + } + + nodeClientsMu.Lock() + if entry, ok := nodeClientsCache[key]; ok { + // A concurrent caller won the race; keep its client and drop ours. + nodeClientsMu.Unlock() + client.CloseIdleConnections() + return entry.client, nil + } + // Any other variant is dead weight: a stale identity's pool fits no trust + // decision now, and an ephemeral proxy URL is never asked for twice. + dropNodeClients(n.Id, key) + nodeClientsCache[key] = nodeClientEntry{nodeID: n.Id, client: client} + nodeClientsMu.Unlock() + return client, nil +} + +func buildNodeHTTPClient(n *model.Node, mode, proxyURL string) (*http.Client, error) { if proxyURL != "" { if mode == "mtls" && n.Scheme != "http" { timeout := remoteHTTPTimeout @@ -191,22 +274,13 @@ func HTTPClientForNode(n *model.Node, proxyURL string) (*http.Client, error) { transport.TLSClientConfig = tlsCfg return client, nil } - if mode == "verify" || n.Scheme == "http" { - return defaultNodeHTTPClient, nil - } if mode == "mtls" { build := func() (idleClosingRoundTripper, error) { tlsCfg, err := tlsConfigForNode(n) if err != nil { return nil, err } - return &http.Transport{ - MaxIdleConns: 64, - MaxIdleConnsPerHost: 4, - IdleConnTimeout: 60 * time.Second, - DialContext: netsafe.SSRFGuardedDialContext, - TLSClientConfig: tlsCfg, - }, nil + return newNodeTransport(tlsCfg), nil } transport, err := newCredentialRotatingTransport(build) if err != nil { @@ -218,15 +292,7 @@ func HTTPClientForNode(n *model.Node, proxyURL string) (*http.Client, error) { if err != nil { return nil, err } - return &http.Client{ - Transport: &http.Transport{ - MaxIdleConns: 64, - MaxIdleConnsPerHost: 4, - IdleConnTimeout: 60 * time.Second, - DialContext: netsafe.SSRFGuardedDialContext, - TLSClientConfig: tlsCfg, - }, - }, nil + return &http.Client{Transport: newNodeTransport(tlsCfg)}, nil } func tlsConfigForNode(n *model.Node) (*tls.Config, error) { diff --git a/internal/web/runtime/tls_client_test.go b/internal/web/runtime/tls_client_test.go index 9059e9f45..2d151b81c 100644 --- a/internal/web/runtime/tls_client_test.go +++ b/internal/web/runtime/tls_client_test.go @@ -6,6 +6,8 @@ import ( "crypto/tls" "encoding/base64" "encoding/hex" + "io" + "net" "net/http" "net/http/httptest" "net/url" @@ -17,6 +19,7 @@ import ( "github.com/mhsanaei/3x-ui/v3/internal/database/model" "github.com/mhsanaei/3x-ui/v3/internal/util/crypto" + "github.com/mhsanaei/3x-ui/v3/internal/util/netsafe" ) type generationProbeTransport struct { @@ -77,6 +80,130 @@ func TestCredentialRotatingTransportDropsOldPoolBeforeNextRequest(t *testing.T) } } +// Heartbeat and traffic sync ask for a client every few seconds; a rebuilt one +// owns an empty pool, so each tick paid a fresh TCP+TLS handshake per node. +func TestHTTPClientForNodeReusesOneConnectionAcrossCalls(t *testing.T) { + var handshakes atomic.Int32 + server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + })) + server.Config.ConnState = func(_ net.Conn, state http.ConnState) { + if state == http.StateNew { + handshakes.Add(1) + } + } + server.StartTLS() + defer server.Close() + + u, err := url.Parse(server.URL) + if err != nil { + t.Fatalf("parse test server url: %v", err) + } + port, err := strconv.Atoi(u.Port()) + if err != nil { + t.Fatalf("test server port: %v", err) + } + node := &model.Node{ + Id: 31, Address: u.Hostname(), Port: port, Scheme: "https", + TlsVerifyMode: "skip", AllowPrivateAddress: true, + } + + for tick := range 2 { + client, err := HTTPClientForNode(node, "") + if err != nil { + t.Fatalf("tick %d: HTTPClientForNode: %v", tick, err) + } + req, err := http.NewRequestWithContext( + netsafe.ContextWithAllowPrivate(context.Background(), true), http.MethodGet, server.URL, nil) + if err != nil { + t.Fatalf("tick %d: new request: %v", tick, err) + } + resp, err := client.Do(req) + if err != nil { + t.Fatalf("tick %d: request: %v", tick, err) + } + _, _ = io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() + } + + if got := handshakes.Load(); got != 1 { + t.Fatalf("TLS handshakes = %d, want 1: a rebuilt client re-handshakes on every tick", got) + } +} + +// A node that switches to pinning (or gains a proxy) must not be served by the +// client built for its previous trust decision. +func TestHTTPClientForNodeRebuildsWhenNodeIdentityChanges(t *testing.T) { + pin := base64.StdEncoding.EncodeToString(make([]byte, sha256.Size)) + node := &model.Node{Id: 32, Address: "node.example.test", Port: 443, Scheme: "https", TlsVerifyMode: "skip"} + skipped, err := HTTPClientForNode(node, "") + if err != nil { + t.Fatalf("skip client: %v", err) + } + + pinned := *node + pinned.TlsVerifyMode = "pin" + pinned.PinnedCertSha256 = pin + pinnedClient, err := HTTPClientForNode(&pinned, "") + if err != nil { + t.Fatalf("pin client: %v", err) + } + if skipped == pinnedClient { + t.Fatal("a pinned node must not reuse the client built to skip verification") + } + + proxied, err := HTTPClientForNode(node, "socks5://127.0.0.1:1080") + if err != nil { + t.Fatalf("proxied client: %v", err) + } + if skipped == proxied { + t.Fatal("a proxied node must not reuse the direct client") + } + + if again, err := HTTPClientForNode(node, ""); err != nil || again == pinnedClient { + t.Fatalf("a skip request must never be served the pinned client; again=%p err=%v", again, err) + } +} + +func nodeClientEntries(id int) int { + nodeClientsMu.Lock() + defer nodeClientsMu.Unlock() + count := 0 + for _, entry := range nodeClientsCache { + if entry.nodeID == id { + count++ + } + } + return count +} + +// withOutboundBridge mints a fresh loopback port per call, so the variant it +// asks for can never be hit again; only the variant in use may stay cached. +func TestHTTPClientForNodeKeepsOneClientPerNode(t *testing.T) { + node := &model.Node{Id: 77, Address: "node.example.test", Port: 443, Scheme: "https", TlsVerifyMode: "skip"} + variants := []string{"socks5://127.0.0.1:41001", "socks5://127.0.0.1:41002", ""} + for _, variant := range variants { + if _, err := HTTPClientForNode(node, variant); err != nil { + t.Fatalf("HTTPClientForNode(%q): %v", variant, err) + } + if got := nodeClientEntries(node.Id); got != 1 { + t.Fatalf("cached clients for the node after %q = %d, want 1", variant, got) + } + if client, err := HTTPClientForNode(node, variant); err != nil || client == nil { + t.Fatalf("repeat HTTPClientForNode(%q): client=%p err=%v", variant, client, err) + } + } + + verify := *node + verify.TlsVerifyMode = "verify" + if client, err := HTTPClientForNode(&verify, ""); err != nil || client != defaultNodeHTTPClient { + t.Fatalf("verify client = %p, want the shared one (%p); err=%v", client, defaultNodeHTTPClient, err) + } + if got := nodeClientEntries(node.Id); got != 0 { + t.Fatalf("cached clients for a node now on verify = %d, want 0", got) + } +} + func TestReloadMasterClientConnectionsValidatesProviderBeforeInvalidation(t *testing.T) { before := masterCertEpoch.Load() SetMasterClientCertProvider(func() (tls.Certificate, error) {