diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index ee4cc6f..ff31948 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -58,6 +58,13 @@ relay joins both pumps and any started cancellation close callback before returning; it never closes the shared transport dialer. HTTP CONNECT installs this owner before replaying any bytes buffered during request parsing. +Both TCP frontends reject a custom dialer's successful socket if the setup +context is already canceled or expired when dialing returns. They close that +owned result and retain the setup error classification (HTTP 504 for a timeout +and the SOCKS timeout reply). This check precedes local timer cleanup; a timely +success remains usable after setup ends, including reusable HTTP origin sockets. +An error returned by the dialer itself is not replaced by a timer or Close error. + ## Native TCP flow 1. The local frontend authenticates the local user when configured and parses @@ -85,6 +92,13 @@ sibling flows. A per-connection pacer sees aggregate QUIC statistics, so concurrent streams share one path estimate rather than each overestimating the same bottleneck. +Native QUIC client closure shuts the physical-dial registration gate before +canceling pending attempts. It waits for each registered attempt to return and +close any late owned connection; concurrent `Close` callers share that completion +barrier. Canceling one caller still ends only its wait, not the shared physical +attempt. An operation that ignores cancellation can delay closure, and best-effort +observability callbacks remain outside this join guarantee. + ## Web-cover TCP flow The web-cover server binds TCP and UDP on the same numeric port. Its TCP side diff --git a/internal/proxy/http.go b/internal/proxy/http.go index baaeb10..0e65449 100644 --- a/internal/proxy/http.go +++ b/internal/proxy/http.go @@ -383,16 +383,18 @@ func validateAbsoluteTarget(target *url.URL) error { } func (s *HTTPServer) dialContext(ctx context.Context, network, address string) (net.Conn, error) { - var ( - conn net.Conn - err error - ) - if s.cfg.dialTimeout <= 0 { - conn, err = s.cfg.dialer.DialContext(ctx, network, address) - } else { - dialCtx, cancel := context.WithTimeout(ctx, s.cfg.dialTimeout) + dialCtx := ctx + if s.cfg.dialTimeout > 0 { + var cancel context.CancelFunc + dialCtx, cancel = context.WithTimeout(ctx, s.cfg.dialTimeout) defer cancel() - conn, err = s.cfg.dialer.DialContext(dialCtx, network, address) + } + conn, err := s.cfg.dialer.DialContext(dialCtx, network, address) + if err == nil { + // Reject an owned late result, but keep the dialer's own error when it + // failed. Check before the deferred timer cleanup so a timely result + // remains reusable after this setup context is detached. + err = dialCtx.Err() } if err != nil || conn == nil { // Own a custom dialer's non-nil failed result before discarding it. diff --git a/internal/proxy/socks5.go b/internal/proxy/socks5.go index 23ffd78..e04d57d 100644 --- a/internal/proxy/socks5.go +++ b/internal/proxy/socks5.go @@ -155,6 +155,12 @@ func (s *SOCKS5Server) serveConnect(client net.Conn, request socksRequest) { if err == nil { upstream, err = s.cfg.dialer.DialContext(ctx, "tcp", request.address) } + if err == nil { + // A custom dialer can return a socket after setup was canceled or + // expired. Check before our own cancel, which must not invalidate a + // timely successful result or become the established relay's owner. + err = ctx.Err() + } cancel() if s.cfg.handshakeTimeout > 0 { _ = client.SetWriteDeadline(time.Now().Add(s.cfg.handshakeTimeout)) diff --git a/internal/proxy/tcp_dial_late_success_test.go b/internal/proxy/tcp_dial_late_success_test.go new file mode 100644 index 0000000..2716498 --- /dev/null +++ b/internal/proxy/tcp_dial_late_success_test.go @@ -0,0 +1,124 @@ +package proxy + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/cppla/autocar/internal/transport" +) + +// A custom dialer may finish establishing a socket after its caller's deadline. +// Return an actual TCP endpoint only after observing that deadline, rather than +// sleeping and hoping to race the timer. Cleanup closes the underlying socket +// without changing the independently observed production Close count. +func TestTCPDialLateSuccessRejectedByFrontends(t *testing.T) { + for _, frontend := range []string{"socks-connect", "http-connect", "http-forward"} { + t.Run(frontend, func(t *testing.T) { + var requests atomic.Int32 + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + requests.Add(1) + _, _ = io.WriteString(w, "unexpected late request") + })) + t.Cleanup(origin.Close) + target := strings.TrimPrefix(origin.URL, "http://") + conn, err := net.DialTimeout("tcp", target, time.Second) + if err != nil { + t.Fatal(err) + } + owned := &tcpDialErrorOwnedConn{Conn: conn, closeError: errors.New("late-result close error")} + t.Cleanup(func() { _ = conn.Close() }) + returned := make(chan error, 1) + var calls atomic.Int32 + dialer := transport.DialFunc(func(ctx context.Context, _, _ string) (net.Conn, error) { + if calls.Add(1) != 1 { + return nil, errors.New("unexpected repeated fixture dial") + } + <-ctx.Done() + returned <- ctx.Err() + return owned, nil + }) + cfg := Config{Dialer: dialer, DialTimeout: 25 * time.Millisecond, HandshakeTimeout: time.Second, IdleTimeout: time.Second, MaxConnections: 1} + var client net.Conn + var tracker *connTracker + rejected := true + if frontend == "socks-connect" { + server, address, stop := startSOCKS5(t, cfg) + t.Cleanup(func() { stop(server) }) + tracker = server.lifecycle.tracker + client = dialTCP(t, address) + t.Cleanup(func() { _ = client.Close() }) + socksGreeting(t, client, nil) + host, port := splitAddress(t, target) + mustWrite(t, client, domainSOCKSRequest(socksCommandConnect, host, port)) + if got := readSOCKSReply(t, client); got != socksReplyTTLExpired { + rejected = false + t.Errorf("late success SOCKS reply = %d, want timeout reply %d", got, socksReplyTTLExpired) + } + } else { + server, proxyURL, stop := startHTTPProxy(t, cfg) + t.Cleanup(func() { stop(server) }) + tracker = server.lifecycle.tracker + client = dialTCP(t, strings.TrimPrefix(proxyURL, "http://")) + t.Cleanup(func() { _ = client.Close() }) + method, requestTarget := http.MethodConnect, target + if frontend == "http-forward" { + method, requestTarget = http.MethodGet, origin.URL+"/late" + } + mustWrite(t, client, []byte(fmt.Sprintf("%s %s HTTP/1.1\r\nHost: %s\r\nConnection: close\r\n\r\n", method, requestTarget, target))) + response, err := http.ReadResponse(bufio.NewReader(client), &http.Request{Method: method}) + if err != nil { + t.Fatal(err) + } + if response.StatusCode != http.StatusGatewayTimeout { + rejected = false + t.Errorf("late success HTTP status = %d, want 504", response.StatusCode) + } + // On the unfixed CONNECT path, the 200 body is the live tunnel; + // do not block reading it merely to report the failed assertion. + if method != http.MethodConnect || response.StatusCode != http.StatusOK { + if _, err := io.ReadAll(response.Body); err != nil { + t.Error("failure response body: ", err) + } + } + _ = response.Body.Close() + } + select { + case err := <-returned: + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("fixture dial context = %v, want deadline exceeded", err) + } + case <-time.After(time.Second): + t.Fatal("custom dial did not return after its deadline") + } + // Release a regressed successful relay before waiting for its slot. + _ = client.Close() + if !rejected { + _ = conn.Close() + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if err := tracker.wait(ctx); err != nil { + t.Fatal("late result retained the frontend slot: ", err) + } + if got := owned.closeCalls.Load(); got != 1 { + t.Errorf("late result Close calls = %d, want exactly one", got) + } + if got := calls.Load(); got != 1 { + t.Errorf("custom dial calls = %d, want one", got) + } + if got := requests.Load(); got != 0 { + t.Errorf("origin received %d requests on an expired dial result", got) + } + }) + } +} diff --git a/internal/tunnel/native_quic_close_join_test.go b/internal/tunnel/native_quic_close_join_test.go new file mode 100644 index 0000000..c7392e3 --- /dev/null +++ b/internal/tunnel/native_quic_close_join_test.go @@ -0,0 +1,213 @@ +package tunnel + +import ( + "context" + "crypto/tls" + "errors" + "net" + "sync" + "testing" + "time" + + "github.com/cppla/autocar/internal/protocol" + quic "github.com/quic-go/quic-go" +) + +// The gate changes only when a real successful DialAddr result is handed back +// to the client. It does not fake a connection, a handshake, or admission. +// Cancellation can race the pinned QUIC dial's successful handshake select; +// the client must own and finish cleaning up that late result before Close. +func TestNativeQUICCloseJoinsLatePhysicalDial(t *testing.T) { + serverTLS, clientTLS := testTLSConfigs(t) + listener, err := quic.ListenAddr("127.0.0.1:0", mustServerTLSConfig(t, serverTLS), hardenedQUICServerConfig(nil, 2)) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + client, err := NewClient(ClientConfig{ServerAddress: listener.Addr().String(), Token: testToken, + TLSConfig: clientTLS, QUICDialTimeout: 2 * time.Second}) + if err != nil { + cancel() + _ = listener.Close() + t.Fatal(err) + } + + peerReady := make(chan *quic.Conn, 1) + peerDone := make(chan struct{}) + peerErr := make(chan error, 1) + go func() { + defer close(peerDone) + peer, err := listener.Accept(ctx) + if err != nil { + peerErr <- err + return + } + peerReady <- peer + <-peer.Context().Done() + peerErr <- nil + }() + + release := make(chan struct{}) + var releaseOnce sync.Once + openGate := func() { releaseOnce.Do(func() { close(release) }) } + dialReady := make(chan *quic.Conn, 1) + dialReturned := make(chan struct{}) + realDial := client.dialQUIC + client.dialQUIC = func(dialCtx context.Context, address string, tlsConfig *tls.Config, cfg *quic.Config) (*quic.Conn, error) { + defer close(dialReturned) + conn, err := realDial(dialCtx, address, tlsConfig, cfg) + if err != nil { + return nil, err + } + dialReady <- conn + // Once the underlying call has succeeded it no longer owns the result. + // This bounded gate models late return, not another cancellable dial. + select { + case <-release: + case <-ctx.Done(): + } + return conn, nil + } + callerDone := make(chan struct{}) + callerResult := make(chan error, 1) + go func() { + defer close(callerDone) + conn, err := client.connection(ctx) + if conn != nil { + _ = conn.CloseWithError(applicationShutdown, "unexpected handoff") + } + callerResult <- err + }() + closeResults := make(chan error, 2) + closeDone := [2]chan struct{}{make(chan struct{}), make(chan struct{})} + var closeStarted [2]bool + var physical, peer *quic.Conn + var attemptDone <-chan struct{} + + t.Cleanup(func() { + // Release every synthetic wait before joining production Close. Exact + // sockets are independently closed if a failed assertion left them live. + openGate() + cancel() + if physical != nil { + _ = physical.CloseWithError(applicationShutdown, "test cleanup") + } + if peer != nil { + _ = peer.CloseWithError(applicationShutdown, "test cleanup") + } + _ = listener.Close() + _ = client.Close() + nativeQUICCloseJoinWait(t, dialReturned, "dial callback cleanup") + if attemptDone != nil { + nativeQUICCloseJoinWait(t, attemptDone, "physical attempt cleanup") + } + nativeQUICCloseJoinWait(t, callerDone, "caller cleanup") + for index, started := range closeStarted { + if started { + nativeQUICCloseJoinWait(t, closeDone[index], "Close worker cleanup") + } + } + nativeQUICCloseJoinWait(t, peerDone, "peer cleanup") + }) + + select { + case physical = <-dialReady: + case <-callerDone: + t.Fatalf("real physical handshake failed: %v", <-callerResult) + case <-ctx.Done(): + t.Fatal("real physical handshake did not complete") + } + select { + case peer = <-peerReady: + case <-peerDone: + t.Fatalf("real peer Accept failed: %v", <-peerErr) + case <-ctx.Done(): + t.Fatal("real peer did not accept the completed handshake") + } + state := physical.ConnectionState() + if state.TLS.Version != tls.VersionTLS13 || state.TLS.NegotiatedProtocol != protocol.ALPN || state.Used0RTT { + t.Fatalf("physical TLS=%#x ALPN=%q 0-RTT=%v", state.TLS.Version, state.TLS.NegotiatedProtocol, state.Used0RTT) + } + client.mu.Lock() + if client.dialing != nil { + attemptDone = client.dialing.done + } + unpublished := client.conn == nil + client.mu.Unlock() + if attemptDone == nil || !unpublished { + t.Fatal("gate did not retain the actual unregistered physical attempt") + } + + startClose := func(index int) { + closeStarted[index] = true + go func() { + defer close(closeDone[index]) + closeResults <- client.Close() + }() + } + startClose(0) + select { + case <-client.ctx.Done(): // The first Close has crossed its closed gate. + case <-ctx.Done(): + t.Fatal("first Close did not cancel the client lifetime") + } + startClose(1) + for index := range closeDone { + select { + case <-closeDone[index]: + t.Errorf("Close worker %d returned before the late physical result could be cleaned up", index) + case <-time.After(75 * time.Millisecond): + } + } + if physical.Context().Err() != nil || peer.Context().Err() != nil { + t.Error("independent cleanup or a caller closed the gated real physical result") + } + select { + case <-attemptDone: + t.Error("attempt completed while its real physical return was still gated") + default: + } + + openGate() + for index := range closeDone { + if !nativeQUICCloseJoinWait(t, closeDone[index], "Close completion") { + return + } + if err := <-closeResults; err != nil { + t.Errorf("Close returned %v", err) + } + } + if !nativeQUICCloseJoinWait(t, dialReturned, "real dial callback") || + !nativeQUICCloseJoinWait(t, attemptDone, "physical attempt") || + !nativeQUICCloseJoinWait(t, callerDone, "connection caller") || + !nativeQUICCloseJoinWait(t, physical.Context().Done(), "physical close") || + !nativeQUICCloseJoinWait(t, peerDone, "actual peer close") { + return + } + if err := <-callerResult; !errors.Is(err, net.ErrClosed) { + t.Errorf("closed caller result=%v, want net.ErrClosed", err) + } + if err := <-peerErr; err != nil { + t.Errorf("peer worker=%v", err) + } + client.mu.Lock() + retained := client.conn != nil || client.dialing != nil + client.mu.Unlock() + if retained { + t.Error("completed Close retained physical connection or dial attempt") + } + if _, err := client.DialContext(ctx, "tcp", "127.0.0.1:1"); !errors.Is(err, net.ErrClosed) { + t.Errorf("future dial=%v, want net.ErrClosed", err) + } +} + +func nativeQUICCloseJoinWait(t *testing.T, done <-chan struct{}, phase string) bool { + t.Helper() + select { + case <-done: + return true + case <-time.After(2 * time.Second): + t.Errorf("%s did not join", phase) + return false + } +} diff --git a/internal/tunnel/quic.go b/internal/tunnel/quic.go index 29c8813..76833df 100644 --- a/internal/tunnel/quic.go +++ b/internal/tunnel/quic.go @@ -464,6 +464,11 @@ type Client struct { pacer *connectionPacer dialing *quicDialAttempt closed bool + // Registration and the closed gate share mu, so Close can safely join + // every physical dial, including an unregistered late successful result. + dialWorkers sync.WaitGroup + closeDone chan struct{} + closeErr error // primaryFailedAt and primaryProbeID implement a small circuit breaker. // They are guarded by mu together with the QUIC connection state. primaryFailedAt time.Time @@ -577,6 +582,7 @@ func NewClient(config ClientConfig) (*Client, error) { txProfile: txProfile, ctx: ctx, cancel: cancel, + closeDone: make(chan struct{}), udpConfig: udpConfig, udpDispatchers: make(map[*quic.Conn]*clientDatagramDispatcher), localMode: basePacer.label(), @@ -1035,6 +1041,7 @@ func (c *Client) connection(ctx context.Context) (*quic.Conn, error) { healthGeneration: c.primaryHealthGeneration, } c.dialing = attempt + c.dialWorkers.Add(1) go c.runQUICDial(attempt, dialCtx) } c.mu.Unlock() @@ -1057,25 +1064,33 @@ func (c *Client) connection(ctx context.Context) (*quic.Conn, error) { } func (c *Client) runQUICDial(attempt *quicDialAttempt, dialCtx context.Context) { + defer c.dialWorkers.Done() defer attempt.cancel() conn, err := c.dialQUIC(dialCtx, c.address, c.tlsConfig.Clone(), c.quicConfig.Clone()) var pacer *connectionPacer + cleanupReason := "client closed" if err == nil { pacer, err = newConnectionPacer(c.pacing, c.maxTx) if err != nil { - _ = conn.CloseWithError(applicationShutdown, "pacing unavailable") + cleanupReason = "pacing unavailable" } } c.mu.Lock() - if c.dialing == attempt { - c.dialing = nil - } - if err == nil && !c.closed { + accepted := err == nil && !c.closed + if accepted { c.conn = conn c.pacer = pacer - } else if conn != nil { - _ = conn.CloseWithError(applicationShutdown, "client closed") + } + c.mu.Unlock() + // Closing a rejected physical result can block. Keep it owned by the + // dial worker, outside mu, and publish completion only after cleanup. + if !accepted && conn != nil { + _ = conn.CloseWithError(applicationShutdown, cleanupReason) + } + c.mu.Lock() + if c.dialing == attempt { + c.dialing = nil } if err != nil { attempt.err = fmt.Errorf("tunnel: dial QUIC: %w", err) @@ -1109,26 +1124,33 @@ func (c *Client) invalidate(conn *quic.Conn) { func (c *Client) Close() error { c.mu.Lock() if c.closed { + done := c.closeDone c.mu.Unlock() - return nil + <-done + return c.closeErr } c.closed = true - c.cancel() - if c.dialing != nil { - c.dialing.cancel() - } + attempt := c.dialing conn := c.conn c.conn = nil c.pacer = nil c.mu.Unlock() + c.cancel() + if attempt != nil { + attempt.cancel() + } if conn != nil { _ = conn.CloseWithError(applicationShutdown, "client closed") } c.closeDatagramDispatchers() + c.dialWorkers.Wait() + var err error if c.fallback != nil { - return c.fallback.Close() + err = c.fallback.Close() } - return nil + c.closeErr = err + close(c.closeDone) + return err } func (c *Client) connectionPacer(conn *quic.Conn) *connectionPacer { diff --git a/internal/tunnel/web_cover_public_origin_test.go b/internal/tunnel/web_cover_public_origin_test.go index 5a4dc24..abe839b 100644 --- a/internal/tunnel/web_cover_public_origin_test.go +++ b/internal/tunnel/web_cover_public_origin_test.go @@ -377,7 +377,7 @@ func newWebPublicOriginFixture(t *testing.T, proto int, mode bool) *webPublicOri }) serverTLS, clientTLS := testTLSConfigs(t) f.frontTLS = clientTLS.Clone() - server, err := ListenWeb(WebServerConfig{TCPAddress: "127.0.0.1:0", UDPAddress: "127.0.0.1:0", Token: webTestToken, TLSConfig: serverTLS, Cover: website, + server, err := listenWebEphemeralFixture(WebServerConfig{TCPAddress: "127.0.0.1:0", UDPAddress: "127.0.0.1:0", Token: webTestToken, TLSConfig: serverTLS, Cover: website, Dialer: transport.DialFunc(func(context.Context, string, string) (net.Conn, error) { f.targetDials.Add(1) return nil, errors.New("public vhost fixture forbids tunnel dial")