diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index 5394929..55fe696 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -284,6 +284,13 @@ shutdown semantics or TCP CONNECT handling, and is not included in v1.0.1. Cancellation requires context-aware resolvers/dialers; custom implementations that ignore cancellation cannot be forcibly interrupted by the proxy. +Source builds reject late destination TCP dial results across native QUIC/TLS +and web H2/H3: a connection returned after the configured outbound dial deadline +is closed even if a custom dialer reports success. Rejection retains the +existing sanitized native dial-failure or authenticated web 502 response and +does not close the shared dialer. Established healthy streams still outlive +their temporary dial context. This change is not included in v1.0.1. + Source builds also cancel an HTTP connection's request contexts when that tracked connection is closed. This lets forced shutdown cancel a pending upstream request even when HTTP/1.1 pipelining has stopped the background diff --git a/docs/WEB_COVER.md b/docs/WEB_COVER.md index b712672..25dc3aa 100644 --- a/docs/WEB_COVER.md +++ b/docs/WEB_COVER.md @@ -408,6 +408,21 @@ does not change authentication, wire profiles, timeout defaults or sibling stream ownership, and is not a general connection-speed or browser-equivalence claim. The published v1.0.1 binary does not contain this change. +Source builds also recheck a queued H3 caller's cancellation before physical +connection selection and authentication reservation. A canceled caller must +not initiate a new shared handshake or claim an otherwise Fresh session. +`web-auto` rejects and closes a logical CONNECT-UDP result returned after its +caller cancellation or primary opening budget; this never triggers H2 UDP +fallback or resets the TCP cooldown. Successful established packets remain +detached from the caller's temporary opening context. + +Both web TCP handlers reject and close a destination connection returned after +the configured destination dial deadline, even if a custom dialer reports +success. The existing authenticated 502 response and connection-bound session +are retained, so a later healthy request can reuse that session. This does not +change authentication, wire profiles, or timeout defaults and is not included +in the published v1.0.1 binary. + After a failed H3 attempt, new TCP streams avoid repeating the UDP timeout. `--fallback-cooldown` is a base duration; each failure independently selects a retry point within +/-20% so clients do not probe in a fixed synchronized diff --git a/internal/tunnel/common.go b/internal/tunnel/common.go index 94b4b8a..19d430f 100644 --- a/internal/tunnel/common.go +++ b/internal/tunnel/common.go @@ -217,6 +217,11 @@ func (s *serverCore) handleStream( dialCtx, cancel := context.WithTimeout(ctx, s.dialTimeout) upstream, err := s.dialer.DialContext(dialCtx, req.Network.String(), req.Address) + // Reject a successful late result before our own context cleanup. This + // bounds establishment without binding a healthy stream to its dial timer. + if err == nil { + err = contextError(dialCtx) + } cancel() if err != nil || upstream == nil { // A custom dialer may return an owned connection even on failure. diff --git a/internal/tunnel/destination_dial_deadline_test.go b/internal/tunnel/destination_dial_deadline_test.go new file mode 100644 index 0000000..ce9ae86 --- /dev/null +++ b/internal/tunnel/destination_dial_deadline_test.go @@ -0,0 +1,488 @@ +package tunnel + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "net" + "net/http" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cppla/autocar/internal/protocol" +) + +// The configured Dialer really connects to an owned TCP endpoint, but returns +// that connection only after the server's outbound deadline. This is a causal +// late-result model, not a claim about the scheduling of the default Dialer. +func TestDestinationDialDeadlineRejectsLateOwnedConnection(t *testing.T) { + for _, mode := range []string{"quic", "tls", "h2", "h3"} { + t.Run(mode, func(t *testing.T) { + f := newDestinationDeadlineFixture(t, mode) + ctx, cancel := context.WithTimeout(f.ctx, 2*time.Second) + conn, err := f.client.DialContext(ctx, "tcp", f.target.Addr().String()) + callerLive := ctx.Err() == nil + cancel() + f.wait(f.dialReturned, "late destination callback returned") + if !callerLive || !f.connectedBeforeDeadline || !errors.Is(f.returnCause, context.DeadlineExceeded) { + t.Fatalf("late-result cause=%v, connected before deadline=%v, caller still live=%v", f.returnCause, f.connectedBeforeDeadline, callerLive) + } + lateSucceeded := conn != nil && err == nil + if conn != nil { + if !lateSucceeded { + _ = conn.Close() + t.Fatalf("destination returned connection and error: %v", err) + } + t.Error("server accepted a connection returned after its destination dial deadline") + // Keep the old-code negative recoverable: actually exercise and + // close its successful stream, then run both healthy controls. + f.echo(conn, []byte("late result incorrectly relayed\x00\xff")) + } else { + f.assertRejection(err) + } + f.wait(f.late.closed, "actual late destination Close completed") + firstBytes := f.awaitTarget() + if !lateSucceeded && firstBytes != 0 { + t.Errorf("rejected late destination received %d bytes", firstBytes) + } + if !lateSucceeded && f.late.closes.Load() != 1 { + t.Errorf("rejected late destination Close count=%d, want 1 before cleanup", f.late.closes.Load()) + } + f.awaitRequest() + var before destinationPreserveSession + if mode == "h2" || mode == "h3" { + before = destinationPreserveWebSession(t, f.client) + } + for i := 0; i < 2; i++ { + ctx, cancel := context.WithTimeout(f.ctx, 2*time.Second) + conn, err := f.client.DialContext(ctx, "tcp", f.target.Addr().String()) + cancel() + if err != nil || conn == nil { + t.Fatalf("healthy request %d: %v", i, err) + } + payload := bytes.Repeat([]byte{byte(i), 'h', 'e', 'a', 'l', 't', 'h', 'y', 0, 255}, 1024) + f.echo(conn, payload) + if received := f.awaitTarget(); received != len(payload) { + t.Fatalf("healthy target %d received %d/%d bytes", i, received, len(payload)) + } + f.awaitRequest() + if mode == "h2" || mode == "h3" { + if after := destinationPreserveWebSession(t, f.client); before != after { + t.Fatal("late rejection or healthy continuation replaced the authenticated physical session") + } + } + } + if calls, closes := f.dialCalls.Load(), f.dialerCloses.Load(); calls != 3 || closes != 0 { + t.Errorf("destination Dialer calls/closes=%d/%d, want 3/0", calls, closes) + } + if got := f.covers.Load(); got != 0 { + t.Errorf("authenticated requests entered cover %d times", got) + } + if mode == "h2" || mode == "h3" { + f.mu.Lock() + requests := append([]destinationDeadlineWebRequest(nil), f.requests...) + f.mu.Unlock() + if len(requests) != 3 { + t.Fatalf("web request count=%d, want 3", len(requests)) + } + if requests[0].physical == nil || requests[0].short { + t.Fatal("first web request did not bootstrap its actual physical connection") + } + for _, request := range requests[1:] { + if request.physical != requests[0].physical || !request.short { + t.Fatal("healthy requests did not use short authentication on the same physical connection") + } + } + if got := f.entropy.nonceReads.Load(); got != 1 { + t.Errorf("full-auth nonce reads=%d, want 1", got) + } + } + t.Logf("%s: late callback deadline=%v, old-success=%v; actual late Close=%d and target EOF; two healthy echoes, destination calls=3, admission=0 before cleanup", mode, f.returnCause, lateSucceeded, f.late.closes.Load()) + }) + } +} + +type destinationDeadlineConn struct { + *net.TCPConn + closes atomic.Int64 + once sync.Once + closed chan struct{} +} + +func (c *destinationDeadlineConn) Close() error { + c.closes.Add(1) + err := c.TCPConn.Close() + c.once.Do(func() { close(c.closed) }) + return err +} + +type destinationDeadlineDialer struct{ fixture *destinationDeadlineFixture } + +func (d *destinationDeadlineDialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) { + f := d.fixture + call := f.dialCalls.Add(1) + if network != "tcp" || address != f.target.Addr().String() { + return nil, fmt.Errorf("fixture denied unowned destination %q/%q", network, address) + } + conn, err := (&net.Dialer{Timeout: time.Second}).DialContext(ctx, "tcp4", address) + if err != nil { + return conn, err + } + raw := conn.(*net.TCPConn) + f.mu.Lock() + if f.closing { + f.mu.Unlock() + _ = raw.Close() + return nil, net.ErrClosed + } + f.raw = append(f.raw, raw) + f.mu.Unlock() + if call != 1 { + return raw, nil + } + f.late = &destinationDeadlineConn{TCPConn: raw, closed: make(chan struct{})} + deadline, ok := ctx.Deadline() + f.connectedBeforeDeadline = ok && time.Now().Before(deadline) && ctx.Err() == nil + select { + case <-ctx.Done(): + f.returnCause = context.Cause(ctx) + case <-f.ctx.Done(): // Independent fatal cleanup releases this callback. + f.returnCause = context.Cause(f.ctx) + } + close(f.dialReturned) + return f.late, nil +} + +func (d *destinationDeadlineDialer) Close() error { + d.fixture.dialerCloses.Add(1) + return nil +} + +type destinationDeadlineWebRequest struct { + physical *webServerConnectionAuth + short bool +} + +type destinationDeadlineTargetResult struct { + bytes int + err error +} + +type destinationDeadlineFixture struct { + t *testing.T + mode string + ctx context.Context + cancel context.CancelFunc + target *net.TCPListener + client destinationDialOwnershipDialer + server destinationDialOwnershipServer + stop context.CancelFunc + serveErr error + + mu sync.Mutex + closing bool + raw []*net.TCPConn + joins []<-chan struct{} + requests []destinationDeadlineWebRequest + webDone chan (<-chan struct{}) + results chan destinationDeadlineTargetResult + admission *StreamAdmission + nativeDone func() bool + late *destinationDeadlineConn + dialReturned chan struct{} + connectedBeforeDeadline bool + returnCause error + dialCalls atomic.Int64 + dialerCloses atomic.Int64 + covers atomic.Int64 + entropy webH2AuthEntropyCounter +} + +func newDestinationDeadlineFixture(t *testing.T, mode string) *destinationDeadlineFixture { + t.Helper() + // Six seconds of independent work plus the two-second cleanup join cap. + ctx, cancel := context.WithTimeout(context.Background(), 6*time.Second) + f := &destinationDeadlineFixture{t: t, mode: mode, ctx: ctx, cancel: cancel, + results: make(chan destinationDeadlineTargetResult, 3), webDone: make(chan (<-chan struct{}), 3), dialReturned: make(chan struct{})} + t.Cleanup(f.cleanup) + var err error + f.target, err = net.ListenTCP("tcp4", &net.TCPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatal(err) + } + done := make(chan struct{}) + f.joins = append(f.joins, done) + go func() { defer close(done); f.serveTarget() }() + f.admission, err = NewStreamAdmission(1) + if err != nil { + t.Fatal(err) + } + serverTLS, clientTLS := testTLSConfigs(t) + outbound := &destinationDeadlineDialer{fixture: f} + cover := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { f.covers.Add(1); http.NotFound(w, r) }) + const serverDialTimeout = 100 * time.Millisecond + switch mode { + case "quic": + s, e := ListenQUIC(QUICServerConfig{Address: "127.0.0.1:0", Token: testToken, TLSConfig: serverTLS, Dialer: outbound, StreamAdmission: f.admission, DialTimeout: serverDialTimeout, HandshakeTimeout: 2 * time.Second}) + if e != nil { + t.Fatal(e) + } + f.server = s + f.nativeDone = func() bool { return len(s.streamSem) == 0 } + c, e := NewClient(ClientConfig{ServerAddress: s.Addr().String(), Token: testToken, TLSConfig: clientTLS, QUICDialTimeout: 2 * time.Second, HandshakeTimeout: 2 * time.Second}) + err = e + if e == nil { + f.client = c + } + case "tls": + s, e := ListenTLS(TLSServerConfig{Address: "127.0.0.1:0", Token: testToken, TLSConfig: serverTLS, Dialer: outbound, StreamAdmission: f.admission, DialTimeout: serverDialTimeout, HandshakeTimeout: 2 * time.Second}) + if e != nil { + t.Fatal(e) + } + f.server = s + f.nativeDone = func() bool { return len(s.connSem) == 0 } + c, e := NewTLSClient(TLSClientConfig{ServerAddress: s.Addr().String(), Token: testToken, TLSConfig: clientTLS, DialTimeout: 2 * time.Second, HandshakeTimeout: 2 * time.Second}) + err = e + if e == nil { + f.client = c + } + case "h2": + s, e := ListenWebH2(WebH2ServerConfig{Address: "127.0.0.1:0", Token: testToken, TLSConfig: serverTLS, Dialer: outbound, StreamAdmission: f.admission, DialTimeout: serverDialTimeout, HandshakeTimeout: 2 * time.Second, Cover: cover}) + if e != nil { + t.Fatal(e) + } + f.server = s + s.server.Handler = f.observeWeb(s.server.Handler) + c, e := NewWebH2Client(WebH2ClientConfig{ServerAddress: s.Addr().String(), Token: testToken, TLSConfig: clientTLS, DialTimeout: 2 * time.Second, HandshakeTimeout: 2 * time.Second, FingerprintProfile: FingerprintNative}) + err = e + if e == nil { + f.client = c + c.auth = newWebAuthSigner(mustWebAuthKey(t, testToken), nil, &f.entropy) + } + case "h3": + s, e := ListenWebH3(WebH3ServerConfig{Address: "127.0.0.1:0", Token: testToken, TLSConfig: serverTLS, Dialer: outbound, StreamAdmission: f.admission, DialTimeout: serverDialTimeout, HandshakeTimeout: 2 * time.Second, Cover: cover}) + if e != nil { + t.Fatal(e) + } + f.server = s + s.server.Handler = f.observeWeb(s.server.Handler) + c, e := NewWebH3Client(WebH3ClientConfig{ServerAddress: s.Addr().String(), Token: testToken, TLSConfig: clientTLS, DialTimeout: 2 * time.Second, HandshakeTimeout: 2 * time.Second, FingerprintProfile: H3FingerprintNative}) + err = e + if e == nil { + f.client = c + c.signer = newWebAuthSigner(mustWebAuthKey(t, testToken), nil, &f.entropy) + } + } + if err != nil { + t.Fatal(err) + } + serveCtx, stop := context.WithCancel(context.Background()) + f.stop = stop + serveDone := make(chan struct{}) + f.mu.Lock() + f.joins = append(f.joins, serveDone) + f.mu.Unlock() + go func() { defer close(serveDone); f.serveErr = f.server.Serve(serveCtx) }() + return f +} + +func (f *destinationDeadlineFixture) observeWeb(handler http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + done := make(chan struct{}) + f.mu.Lock() + if f.closing { + f.mu.Unlock() + return + } + f.joins = append(f.joins, done) + physical, _ := r.Context().Value(webServerConnectionAuthContextKey{}).(*webServerConnectionAuth) + values := r.Header.Values("Proxy-Authorization") + short := false + if len(values) == 1 { + _, short = parseWebSessionBearer(values[0]) + } + f.requests = append(f.requests, destinationDeadlineWebRequest{physical: physical, short: short}) + f.mu.Unlock() + defer close(done) + f.webDone <- done + handler.ServeHTTP(w, r) + }) +} + +func (f *destinationDeadlineFixture) serveTarget() { + for i := 0; i < 3; i++ { + conn, err := f.target.AcceptTCP() + if err != nil { + return + } + f.mu.Lock() + if f.closing { + f.mu.Unlock() + _ = conn.Close() + return + } + f.raw = append(f.raw, conn) + f.mu.Unlock() + err = conn.SetDeadline(time.Now().Add(3 * time.Second)) + var got []byte + if err == nil { + got, err = io.ReadAll(conn) + } + if err == nil { + _, err = io.Copy(conn, bytes.NewReader(got)) + } + if err == nil { + err = conn.CloseWrite() + } + closeErr := conn.Close() + if err == nil { + err = closeErr + } + f.results <- destinationDeadlineTargetResult{bytes: len(got), err: err} + } +} + +func (f *destinationDeadlineFixture) echo(conn net.Conn, payload []byte) { + f.t.Helper() + defer conn.Close() + if err := conn.SetDeadline(time.Now().Add(2 * time.Second)); err != nil { + f.t.Fatal(err) + } + if _, err := io.Copy(conn, bytes.NewReader(payload)); err != nil { + f.t.Fatal(err) + } + writer, ok := conn.(interface{ CloseWrite() error }) + if !ok { + f.t.Fatal("successful stream lost CloseWrite") + } + if err := writer.CloseWrite(); err != nil { + f.t.Fatal(err) + } + got, err := io.ReadAll(conn) + if err != nil || !bytes.Equal(got, payload) { + f.t.Fatalf("real echo=%d/%d bytes, error=%v", len(got), len(payload), err) + } +} + +func (f *destinationDeadlineFixture) assertRejection(err error) { + f.t.Helper() + if f.mode == "quic" || f.mode == "tls" { + var remote *RemoteError + if !errors.As(err, &remote) || remote.Status != protocol.StatusDialFailed || remote.Message != "destination unavailable" { + f.t.Errorf("late destination rejection=%T %v, want sanitized StatusDialFailed", err, err) + } + } else { + var remote *WebConnectError + if !errors.As(err, &remote) || remote.Transport != f.mode || remote.StatusCode != http.StatusBadGateway || err.Error() != fmt.Sprintf("tunnel: %s cover CONNECT rejected with HTTP status 502", f.mode) { + f.t.Errorf("late destination rejection=%T %v, want authenticated typed 502", err, err) + } + } +} + +func (f *destinationDeadlineFixture) wait(done <-chan struct{}, what string) { + f.t.Helper() + select { + case <-done: + case <-f.ctx.Done(): + f.t.Fatalf("%s: %v", what, context.Cause(f.ctx)) + } +} + +func (f *destinationDeadlineFixture) awaitTarget() int { + f.t.Helper() + select { + case result := <-f.results: + if result.err != nil { + f.t.Fatalf("real target EOF/echo: %v", result.err) + } + return result.bytes + case <-f.ctx.Done(): + f.t.Fatal("real target worker did not reach EOF") + } + return 0 +} + +func (f *destinationDeadlineFixture) awaitRequest() { + f.t.Helper() + if f.mode == "h2" || f.mode == "h3" { + select { + case done := <-f.webDone: + f.wait(done, "web handler joined") + case <-f.ctx.Done(): + f.t.Fatal("web request was not observed") + } + } + tick := time.NewTicker(5 * time.Millisecond) + defer tick.Stop() + for len(f.admission.sem) != 0 || (f.nativeDone != nil && !f.nativeDone()) { + select { + case <-tick.C: + case <-f.ctx.Done(): + f.t.Fatal("actual handler did not release admission") + } + } +} + +func (f *destinationDeadlineFixture) cleanup() { + f.mu.Lock() + f.closing = true + raw := append([]*net.TCPConn(nil), f.raw...) + f.mu.Unlock() + f.cancel() // Releases the configured late callback independently of Serve. + if f.target != nil { + _ = f.target.Close() + } + for _, conn := range raw { + _ = conn.Close() + } // Never calls the observed wrapper. + done := make(chan struct{}) + var closeErr error + go func() { + defer close(done) + if f.client != nil { + closeErr = f.client.Close() + } + if f.server != nil { + closeErr = errors.Join(closeErr, f.server.Close()) + } + if f.stop != nil { + f.stop() + } // Close precedes Serve cancellation (H3 socket owner). + }() + timer := time.NewTimer(2 * time.Second) + defer timer.Stop() + select { + case <-done: + if !destinationDialOwnershipOnlyClosed(closeErr) { + f.t.Errorf("client/server cleanup: %v", closeErr) + } + case <-timer.C: + f.t.Error("client/server cleanup did not join") + return + } + // Drain a fresh index, not an early snapshot: accepted children register + // before their parent completes, and cleanup's closed gate blocks new work. + for i := 0; ; i++ { + f.mu.Lock() + if i == len(f.joins) { + f.mu.Unlock() + break + } + join := f.joins[i] + f.mu.Unlock() + select { + case <-join: + case <-timer.C: + f.t.Error("owned worker cleanup did not join") + return + } + } + if f.serveErr != nil { + f.t.Errorf("server Serve cleanup: %v", f.serveErr) + } + f.t.Log("all owned target/handler/server workers and client/server Close joined") +} diff --git a/internal/tunnel/web_client.go b/internal/tunnel/web_client.go index 44a9798..0838ce1 100644 --- a/internal/tunnel/web_client.go +++ b/internal/tunnel/web_client.go @@ -474,6 +474,9 @@ func (c *WebClient) DialPacket(ctx context.Context) (transport.PacketConn, error if ctx == nil { return nil, errors.New("tunnel: nil web-cover packet dial context") } + if err := contextError(ctx); err != nil { + return nil, err + } packetDialer, ok := c.primary.dialer.(transport.PacketDialer) if !ok { return nil, errors.New("tunnel: web-cover primary does not support CONNECT-UDP") @@ -485,8 +488,17 @@ func (c *WebClient) DialPacket(ctx context.Context) (transport.PacketConn, error stopClient := context.AfterFunc(c.ctx, cancelDial) packet, err := packetDialer.DialPacket(dialContext) stopClient() + // Inspect the attempt before our own cleanup cancels it. A successful late + // result still belongs to this failed opening, not to an established user. + attemptErr := contextError(dialContext) cancelDial() + if err == nil { + err = attemptErr + } if err != nil { + if packet != nil { + _ = packet.Close() + } if callerErr := contextError(ctx); callerErr != nil { return nil, callerErr } @@ -495,6 +507,10 @@ func (c *WebClient) DialPacket(ctx context.Context) (transport.PacketConn, error } return nil, fmt.Errorf("tunnel: web-cover %s CONNECT-UDP: %w", c.primary.name, err) } + if callerErr := contextError(ctx); callerErr != nil { + _ = packet.Close() + return nil, callerErr + } if c.isClosed() { _ = packet.Close() return nil, net.ErrClosed diff --git a/internal/tunnel/web_client_packet_handoff_test.go b/internal/tunnel/web_client_packet_handoff_test.go new file mode 100644 index 0000000..e70bf3f --- /dev/null +++ b/internal/tunnel/web_client_packet_handoff_test.go @@ -0,0 +1,382 @@ +package tunnel + +import ( + "bytes" + "context" + "crypto/tls" + "errors" + "fmt" + "net" + "net/http" + "net/netip" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cppla/autocar/internal/transport" +) + +// The decorator only schedules the return of a real, successful H3 PacketConn. +// It does not synthesize SETTINGS, authentication, packets, or network latency. +type webClientPacketHandoffPrimary struct { + *WebH3Client + mu sync.Mutex + packets []*webUDPPacketConn + handoff func(context.Context) +} + +func (d *webClientPacketHandoffPrimary) DialPacket(ctx context.Context) (transport.PacketConn, error) { + packet, err := d.WebH3Client.DialPacket(ctx) + if err != nil { + return packet, err + } + actual, ok := packet.(*webUDPPacketConn) + if !ok { + _ = packet.Close() + return nil, errors.New("handoff fixture expected a real H3 PacketConn") + } + d.mu.Lock() + d.packets = append(d.packets, actual) + handoff := d.handoff + d.mu.Unlock() + if handoff != nil { + handoff(ctx) + } + return actual, nil +} + +type webClientPacketHandoffFallback struct{ calls atomic.Int64 } + +func (d *webClientPacketHandoffFallback) DialContext(context.Context, string, string) (net.Conn, error) { + d.calls.Add(1) + return nil, errors.New("H2 fallback is forbidden for UDP handoff") +} + +func (*webClientPacketHandoffFallback) Close() error { return nil } + +type webClientPacketHandoffFixture struct { + client *WebClient + primary *webClientPacketHandoffPrimary + fallback *webClientPacketHandoffFallback + server *WebH3Server + echo *net.UDPConn + target netip.AddrPort + serveCancel context.CancelFunc + serveResult chan error + serveDone chan struct{} + echoDone chan struct{} + echoResult chan error + handlerDone chan struct{} + joinedHandlers int + resolver atomic.Int64 + tcpDials atomic.Int64 + cover atomic.Int64 + requests atomic.Int64 + unexpected atomic.Int64 + receivers []<-chan struct{} +} + +func webClientPacketHandoffWait(t *testing.T, done <-chan struct{}, label string) bool { + t.Helper() + select { + case <-done: + return true + case <-time.After(2 * time.Second): + t.Errorf("%s did not join", label) + return false + } +} + +func newWebClientPacketHandoffFixture(t *testing.T) *webClientPacketHandoffFixture { + t.Helper() + echo, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatal(err) + } + f := &webClientPacketHandoffFixture{ + echo: echo, target: echo.LocalAddr().(*net.UDPAddr).AddrPort(), + echoDone: make(chan struct{}), echoResult: make(chan error, 1), + serveDone: make(chan struct{}), serveResult: make(chan error, 1), + handlerDone: make(chan struct{}, 4), fallback: &webClientPacketHandoffFallback{}, + } + t.Cleanup(func() { + if f.primary != nil { + f.primary.mu.Lock() + packets := append([]*webUDPPacketConn(nil), f.primary.packets...) + f.primary.mu.Unlock() + for _, packet := range packets { + _ = packet.Close() + } + } + if f.client != nil { + if err := f.client.Close(); err != nil { + t.Errorf("client Close: %v", err) + } + } else if f.primary != nil { + _ = f.primary.Close() + } + for _, done := range f.receivers { + webClientPacketHandoffWait(t, done, "owned PacketConn Receive worker") + } + if f.server != nil { + if err := normalizeWebServerCloseError(f.server.Close()); err != nil { + t.Errorf("server Close: %v", err) + } + f.serveCancel() + if webClientPacketHandoffWait(t, f.serveDone, "owned H3 Serve worker") { + if err := <-f.serveResult; err != nil { + t.Errorf("server Serve: %v", err) + } + } + for want := int(f.requests.Load()); f.joinedHandlers < want; f.joinedHandlers++ { + select { + case <-f.handlerDone: + case <-time.After(2 * time.Second): + t.Error("owned H3 request handler did not join during independent cleanup") + f.joinedHandlers = want + } + } + } + _ = f.echo.Close() + if webClientPacketHandoffWait(t, f.echoDone, "owned UDP echo worker") { + if err := <-f.echoResult; err != nil { + t.Errorf("UDP echo: %v", err) + } + } + }) + go func() { + defer close(f.echoDone) + buffer := make([]byte, 2048) + for { + if err := echo.SetDeadline(time.Now().Add(4 * time.Second)); err != nil { + f.echoResult <- err + return + } + n, source, err := echo.ReadFromUDPAddrPort(buffer) + if err != nil { + if errors.Is(err, net.ErrClosed) { + err = nil + } + f.echoResult <- err + return + } + if _, err := echo.WriteToUDPAddrPort(buffer[:n], source); err != nil { + f.echoResult <- err + return + } + } + }() + serverTLS, clientTLS := testTLSConfigs(t) + f.server, err = ListenWebH3(WebH3ServerConfig{ + Address: "127.0.0.1:0", Token: webTestToken, TLSConfig: serverTLS, + HandshakeTimeout: time.Second, DialTimeout: time.Second, + Dialer: transport.DialFunc(func(context.Context, string, string) (net.Conn, error) { + f.tcpDials.Add(1) + return nil, errors.New("TCP destination forbidden in UDP handoff fixture") + }), + Cover: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + f.cover.Add(1) + http.NotFound(w, nil) + }), + UDPResolver: UDPResolverFunc(func(_ context.Context, address string) ([]netip.AddrPort, error) { + f.resolver.Add(1) + if address != f.target.String() { + f.unexpected.Add(1) + return nil, errors.New("only the numeric owned UDP target is permitted") + } + return []netip.AddrPort{f.target}, nil + }), + }) + if err != nil { + t.Fatal(err) + } + original := f.server.server.Handler + f.server.server.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + f.requests.Add(1) + defer func() { f.handlerDone <- struct{}{} }() + original.ServeHTTP(w, r) + }) + serveCtx, cancelServe := context.WithCancel(context.Background()) + f.serveCancel = cancelServe + go func() { + defer close(f.serveDone) + f.serveResult <- f.server.Serve(serveCtx) + }() + h3, err := NewWebH3Client(WebH3ClientConfig{ + ServerAddress: f.server.Addr().String(), Token: webTestToken, TLSConfig: clientTLS, + FingerprintProfile: H3FingerprintNative, DialTimeout: time.Second, HandshakeTimeout: time.Second, + }) + if err != nil { + t.Fatal(err) + } + f.primary = &webClientPacketHandoffPrimary{WebH3Client: h3} + f.client, err = newWebClientWithPaths( + webClientPath{name: webAuthTransportH3, dialer: f.primary}, + webClientPath{name: webAuthTransportH2, dialer: f.fallback}, + 80*time.Millisecond, time.Minute, time.Now, + ) + if err != nil { + t.Fatal(err) + } + // Physical handshake and DATAGRAM SETTINGS use an independent budget. The + // 80ms tested budget below measures only a warm logical PacketConn handoff. + warmCtx, cancelWarm := context.WithTimeout(context.Background(), 2*time.Second) + warm, err := h3.DialPacket(warmCtx) + cancelWarm() + if err != nil { + t.Fatal(err) + } + _ = warm.Close() + h3.mu.Lock() + physical := h3.conn + h3.mu.Unlock() + if physical == nil { + t.Fatal("missing warm real H3 physical connection") + } + state := physical.ConnectionState().TLS + if state.Version != tls.VersionTLS13 || state.NegotiatedProtocol != "h3" || len(state.VerifiedChains) == 0 { + t.Fatalf("unverified warm H3 TLS state: %+v", state) + } + return f +} + +func (f *webClientPacketHandoffFixture) echoPayload(t *testing.T, packet transport.PacketConn, payload []byte) { + t.Helper() + if err := packet.Send(payload, f.target.String()); err != nil { + t.Fatal(err) + } + result := make(chan webClientPacketHandoffResult, 1) + done := make(chan struct{}) + f.receivers = append(f.receivers, done) + go func() { + defer close(done) + got, address, err := packet.Receive() + result <- webClientPacketHandoffResult{payload: got, address: address, err: err} + }() + if !webClientPacketHandoffWait(t, done, "owned healthy UDP Receive worker") { + t.FailNow() + } + got := <-result + if got.err != nil || !bytes.Equal(got.payload, payload) || got.address != f.target.String() { + t.Fatalf("real UDP echo=%q/%q/%v, want=%q/%q", got.payload, got.address, got.err, payload, f.target) + } +} + +type webClientPacketHandoffResult struct { + payload []byte + address string + err error +} + +func testWebClientPacketHandoff(t *testing.T, deadline bool) { + f := newWebClientPacketHandoffFixture(t) + f.primary.WebH3Client.mu.Lock() + physical := f.primary.conn + f.primary.WebH3Client.mu.Unlock() + caller, cancelCaller := context.WithCancelCause(context.Background()) + t.Cleanup(func() { cancelCaller(context.Canceled) }) + cause := errors.New("private UDP handoff caller cancellation") + if deadline { + cause = context.DeadlineExceeded + } + var handoffErr error + f.primary.handoff = func(budget context.Context) { + if err := contextError(budget); err != nil { + handoffErr = fmt.Errorf("warm packet returned after budget before handoff: %w", err) + return + } + if deadline { + <-budget.Done() + handoffErr = context.Cause(budget) + } else { + cancelCaller(cause) + } + } + f.client.mu.Lock() + f.client.nextPrimaryID++ + generation := f.client.nextPrimaryID + f.client.mu.Unlock() + if !f.client.recordPrimaryFailure(generation, time.Now()) { + t.Fatal("failed to install the healthy-control circuit generation") + } + packet, err := f.client.DialPacket(caller) + if deadline && !errors.Is(handoffErr, context.DeadlineExceeded) { + t.Fatalf("primary deadline handoff was not observed: %v", handoffErr) + } + if !deadline && handoffErr != nil { + t.Fatalf("caller handoff was not observed: %v", handoffErr) + } + correctCause := err == cause + if deadline { + correctCause = errors.Is(err, context.DeadlineExceeded) + } + if packet != nil || !correctCause { + t.Errorf("late PacketConn=%T error=%v, want nil and exact %v", packet, err, cause) + } + f.primary.mu.Lock() + packets := append([]*webUDPPacketConn(nil), f.primary.packets...) + f.primary.mu.Unlock() + if len(packets) != 1 { + t.Fatalf("handoff did not return exactly one real H3 PacketConn: %d", len(packets)) + } + late := packets[0] + select { + case <-late.done: + default: + t.Error("late real H3 PacketConn remains open before independent cleanup") + } + if deadline && contextError(caller) != nil { + t.Error("primary-only budget canceled the still-live original caller") + } + if f.requests.Load() != 0 || f.resolver.Load() != 0 || f.cover.Load() != 0 || f.tcpDials.Load() != 0 || f.fallback.calls.Load() != 0 { + t.Error("logical handoff unexpectedly emitted a target/cover/fallback request") + } + assertWebPrimaryFailureGeneration(t, f.client, generation) + if f.client.SelectedTransport() != "" { + t.Error("unestablished logical packet falsely selected a transport") + } + // Independent cleanup follows the ownership oracle even on the old-source + // failure, so a leaked result cannot pollute the real healthy control. + _ = late.Close() + f.primary.mu.Lock() + f.primary.handoff = nil + f.primary.mu.Unlock() + live, cancelLive := context.WithCancelCause(context.Background()) + t.Cleanup(func() { cancelLive(context.Canceled) }) + healthy, err := f.client.DialPacket(live) + if err != nil { + t.Fatal(err) + } + assertWebPrimaryFailureGeneration(t, f.client, generation) + f.echoPayload(t, healthy, []byte("complete real H3 datagram before caller cancellation")) + cancelLive(errors.New("successful packet must be detached from caller")) + f.echoPayload(t, healthy, []byte("complete real H3 datagram after caller cancellation")) + f.primary.WebH3Client.mu.Lock() + samePhysical := f.primary.conn == physical + f.primary.WebH3Client.mu.Unlock() + if !samePhysical || f.fallback.calls.Load() != 0 || f.requests.Load() != 1 || f.resolver.Load() != 1 || f.unexpected.Load() != 0 { + t.Error("healthy UDP control did not reuse only the authenticated H3 physical path") + } + if f.client.SelectedTransport() != webAuthTransportH3 { + t.Error("a real newly authenticated UDP target did not select H3") + } + _ = healthy.Close() + select { + case <-f.handlerDone: + f.joinedHandlers++ + case <-time.After(2 * time.Second): + t.Fatal("real authenticated UDP handler did not join after packet Close") + } + if got := len(f.primary.udpSlots); got != 0 { + t.Errorf("UDP session admission remains occupied after packet Close: %d", got) + } +} + +func TestWebClientPacketHandoffHonorsCallerCancellation(t *testing.T) { + testWebClientPacketHandoff(t, false) +} + +func TestWebClientPacketHandoffHonorsPrimaryDeadline(t *testing.T) { + testWebClientPacketHandoff(t, true) +} diff --git a/internal/tunnel/web_h3_client.go b/internal/tunnel/web_h3_client.go index 397b68d..e20141e 100644 --- a/internal/tunnel/web_h3_client.go +++ b/internal/tunnel/web_h3_client.go @@ -416,6 +416,10 @@ func (c *WebH3Client) reserveAuthenticatedSession(ctx context.Context) (webH3Ses c.mu.Unlock() return webH3SessionReservation{}, net.ErrClosed } + if err := contextError(ctx); err != nil { + c.mu.Unlock() + return webH3SessionReservation{}, err + } if session == nil || session.retired || conn.Context().Err() != nil { c.mu.Unlock() continue @@ -484,6 +488,12 @@ func (c *WebH3Client) connection(ctx context.Context) (*quic.Conn, *http3.Client c.mu.Unlock() return nil, nil, net.ErrClosed } + // A caller can be canceled while queued on mu. Recheck before selecting + // an unclaimed session or starting client-owned physical initialization. + if err := contextError(ctx); err != nil { + c.mu.Unlock() + return nil, nil, err + } if c.conn != nil && c.client != nil && c.conn.Context().Err() == nil { conn, client := c.conn, c.client c.mu.Unlock() diff --git a/internal/tunnel/web_h3_selection_cancellation_test.go b/internal/tunnel/web_h3_selection_cancellation_test.go new file mode 100644 index 0000000..74beaad --- /dev/null +++ b/internal/tunnel/web_h3_selection_cancellation_test.go @@ -0,0 +1,311 @@ +package tunnel + +import ( + "bytes" + "context" + "crypto/tls" + "errors" + "fmt" + "io" + "net" + "net/http" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/apernet/quic-go" + "github.com/cppla/autocar/internal/transport" +) + +// Deadline is reached only after contextError has read the real context's +// still-nil Cause. Compute its actual deadline before publishing the witness; +// cancellation must not retroactively change that preflight result. Value, +// Err and Done all retain the real WithCancelCause context's semantics. +type webH3SelectionPreflightContext struct { + context.Context + entered chan struct{} + once sync.Once +} + +func (c *webH3SelectionPreflightContext) Deadline() (time.Time, bool) { + deadline, ok := c.Context.Deadline() + c.once.Do(func() { close(c.entered) }) + return deadline, ok +} + +type webH3SelectionObservation struct { + connection *webServerConnectionAuth + full bool + short bool + proof bool + proto int + tlsVersion uint16 + joined <-chan struct{} +} + +type webH3SelectionFixture struct { + f *webH2DialSharingFixture + client *WebH3Client + target string + entropy *webH2AuthEntropyCounter + observed chan webH3SelectionObservation + verified chan struct{} + verifyCalls atomic.Int32 + dials atomic.Int32 + covers atomic.Int32 + unexpected atomic.Int32 +} + +func newWebH3SelectionFixture(t *testing.T) *webH3SelectionFixture { + t.Helper() + f := newWebH2DialSharingFixture(t) + x := &webH3SelectionFixture{ + f: f, target: f.target(), entropy: &webH2AuthEntropyCounter{}, + observed: make(chan webH3SelectionObservation, 2), verified: make(chan struct{}, 4), + } + serverTLS, clientTLS := testTLSConfigs(t) + server, err := ListenWebH3(WebH3ServerConfig{ + Address: "127.0.0.1:0", Token: webTestToken, TLSConfig: serverTLS, + HandshakeTimeout: time.Second, DialTimeout: time.Second, + Cover: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + x.covers.Add(1) + http.NotFound(w, r) + }), + Dialer: transport.DialFunc(func(ctx context.Context, network, address string) (net.Conn, error) { + x.dials.Add(1) + if network != "tcp" || address != x.target { + return nil, errors.New("selection fixture refuses a non-owned destination") + } + return (&net.Dialer{Timeout: time.Second}).DialContext(ctx, "tcp4", x.target) + }), + }) + if err != nil { + t.Fatal(err) + } + original := server.server.Handler + server.server.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + joined := make(chan struct{}) + f.register(joined) + defer close(joined) + connection, _ := r.Context().Value(webServerConnectionAuthContextKey{}).(*webServerConnectionAuth) + bearer := r.Header.Get("Proxy-Authorization") + _, short := parseWebSessionBearer(bearer) + original.ServeHTTP(w, r) + observation := webH3SelectionObservation{ + connection: connection, full: len(bearer) >= 385 && len(bearer) <= 2047, + short: short, proof: w.Header().Get(webAuthResponseHeader) != "", + proto: r.ProtoMajor, tlsVersion: webRequestTLSVersion(r), joined: joined, + } + select { + case x.observed <- observation: + default: + x.unexpected.Add(1) + } + }) + serveJoined := make(chan struct{}) + f.register(serveJoined) + var serveErr error + f.checks = append(f.checks, func() { + if serveErr != nil { + t.Errorf("owned selection H3 Serve: %v", serveErr) + } + }) + f.close = append(f.close, func() { + if err := server.Close(); err != nil && !errors.Is(err, net.ErrClosed) { + t.Errorf("owned selection H3 server Close: %v", err) + } + }) + go func() { defer close(serveJoined); serveErr = server.Serve(f.ctx) }() + + // Native's standard certificate chain/name verification still runs before + // this callback. It is an actual successful TLS handshake observation, not + // a synthetic physical dial hook or a browser-profile claim. + clientTLS = clientTLS.Clone() + clientTLS.VerifyConnection = func(state tls.ConnectionState) error { + if state.Version != tls.VersionTLS13 || len(state.VerifiedChains) == 0 || state.NegotiatedProtocol != "h3" { + return errors.New("owned selection TLS verification state is incomplete") + } + x.verifyCalls.Add(1) + select { + case x.verified <- struct{}{}: + default: + x.unexpected.Add(1) + } + return nil + } + x.client, err = NewWebH3Client(WebH3ClientConfig{ + ServerAddress: server.Addr().String(), Token: webTestToken, TLSConfig: clientTLS, + FingerprintProfile: H3FingerprintNative, DialTimeout: time.Second, HandshakeTimeout: time.Second, + }) + if err != nil { + t.Fatal(err) + } + x.client.signer = newWebAuthSigner(mustWebAuthKey(t, webTestToken), nil, x.entropy) + f.close = append(f.close, func() { + if err := x.client.Close(); err != nil { + t.Errorf("owned selection H3 client Close: %v", err) + } + }) + return x +} + +func (x *webH3SelectionFixture) canceledAtPreflight(call func(context.Context) error) error { + x.f.t.Helper() + ctx, cancel := context.WithCancelCause(x.f.ctx) + defer cancel(context.Canceled) + witness := &webH3SelectionPreflightContext{Context: ctx, entered: make(chan struct{})} + cause := errors.New("private cancellation after successful selection preflight") + result, joined := make(chan error, 1), make(chan struct{}) + x.f.register(joined) + x.client.mu.Lock() + unlock := sync.OnceFunc(x.client.mu.Unlock) + // This is appended last and therefore runs before client.Close on Fatal. + x.f.close = append(x.f.close, unlock) + go func() { defer close(joined); result <- call(witness) }() + x.f.wait(witness.entered, "preflight read nil Cause before actual mutex wait") + cancel(cause) + x.f.wait(ctx.Done(), "real caller cancellation publication while mutex held") + if context.Cause(witness) != cause { + x.f.t.Fatal("wrapper did not retain the real caller's cancellation cause") + } + unlock() + x.f.wait(joined, "canceled selection caller completion") + err := <-result + if err != cause { + x.f.t.Errorf("canceled selection result=%v, want exact private caller cause", err) + } + return err +} + +// These later controls deliberately still run after a negative assertion. +// On old source they can prove healthy replacement/retry even when the +// cancellation incorrectly changed a Fresh connection's authentication state. +func (x *webH3SelectionFixture) healthy(expected *quic.Conn) { + x.f.t.Helper() + noncesBefore, verifiesBefore := x.entropy.nonceReads.Load(), x.verifyCalls.Load() + var physical *quic.Conn + for round := range 2 { + ctx, cancel := context.WithTimeout(x.f.ctx, 2*time.Second) + conn, err := x.client.DialContext(ctx, "tcp", x.target) + if err != nil { + cancel() + x.f.t.Fatalf("later healthy public H3 CONNECT %d: %v", round, err) + } + if err := conn.SetDeadline(time.Now().Add(time.Second)); err != nil { + _ = conn.Close() + cancel() + x.f.t.Fatalf("later healthy H3 I/O deadline: %v", err) + } + payload := []byte(fmt.Sprintf("owned-selection-cancellation-echo-%d", round)) + _, writeErr := conn.Write(payload) + closeWriteErr := conn.(*webH3Conn).CloseWrite() + got, readErr := io.ReadAll(io.LimitReader(conn, int64(len(payload))+1)) + closeErr := conn.Close() + cancel() + if writeErr != nil || closeWriteErr != nil || readErr != nil || closeErr != nil || !bytes.Equal(got, payload) { + x.f.t.Fatalf("full H3/TCP echo %d=%q write/FIN/read/close=%v/%v/%v/%v", round, got, writeErr, closeWriteErr, readErr, closeErr) + } + x.client.mu.Lock() + current := x.client.conn + x.client.mu.Unlock() + if current == nil || (round == 1 && current != physical) { + x.f.t.Error("two later healthy public flows did not reuse one actual H3 connection") + } + physical = current + } + var observations [2]webH3SelectionObservation + for i := range observations { + select { + case observations[i] = <-x.observed: + case <-x.f.ctx.Done(): + x.f.t.Fatal("later authenticated handler observation exceeded fixture budget") + } + x.f.wait(observations[i].joined, "later actual authenticated H3 handler joined") + if !observations[i].proof || observations[i].proto != 3 || observations[i].tlsVersion != tls.VersionTLS13 || observations[i].connection == nil { + x.f.t.Errorf("healthy handler %d proof/proto/TLS/connection=%t/%d/%x/%t", i, observations[i].proof, observations[i].proto, observations[i].tlsVersion, observations[i].connection != nil) + } + } + if observations[0].connection != observations[1].connection || !observations[0].full || observations[0].short || !observations[1].short || x.entropy.nonceReads.Load()-noncesBefore != 1 || x.dials.Load() != 2 || x.covers.Load() != 0 || x.unexpected.Load() != 0 { + x.f.t.Errorf("healthy full/short authentication control failed: full/short=%t/%t newnonce=%d destination/cover/unexpected=%d/%d/%d", observations[0].full, observations[1].short, x.entropy.nonceReads.Load()-noncesBefore, x.dials.Load(), x.covers.Load(), x.unexpected.Load()) + } + if expected != nil && physical != expected { + x.f.t.Error("caller cancellation replaced the previously initialized Fresh physical connection") + } + if expected != nil && x.verifyCalls.Load() != verifiesBefore { + x.f.t.Error("reusing the initialized Fresh connection performed another actual TLS handshake") + } + x.f.t.Log("later public H3 two complete TCP echoes, same physical connection and full/short authentication controls evaluated") +} + +func TestWebH3ConnectionCanceledWhileWaitingForSelection(t *testing.T) { + x := newWebH3SelectionFixture(t) + x.canceledAtPreflight(func(ctx context.Context) error { + conn, client, err := x.client.connection(ctx) + if conn != nil || client != nil { + x.f.t.Error("canceled cold selection returned a physical H3 connection") + } + return err + }) + x.client.mu.Lock() + attempt := x.client.dial + started := attempt != nil || x.client.conn != nil || x.client.client != nil || len(x.client.conns) != 0 + x.client.mu.Unlock() + if started { + t.Error("caller canceled while waiting for mutex nevertheless started client-owned physical H3 setup") + } + // Old source can already have cleared dial after publishing a connection. + // A captured attempt is awaited through its immutable result, not a private + // WaitGroup/runtime-stack introspection. Standard TLS callback is wire proof. + if attempt != nil { + x.f.wait(attempt.done, "unexpected old-source actual physical attempt completed") + } + if calls := x.verifyCalls.Load(); calls != 0 { + t.Errorf("canceled-before-selection performed %d actual verified TLS handshakes", calls) + } + if started { + select { + case <-x.verified: + t.Log("negative boundary also reached actual owned native TLS certificate verification") + default: + t.Log("negative boundary observed physical setup state before verified TLS completion") + } + } + if x.entropy.nonceReads.Load() != 0 || x.dials.Load() != 0 || x.covers.Load() != 0 { + t.Error("canceled cold selection attempted application authentication, destination dialing or public cover") + } + x.healthy(nil) +} + +func TestWebH3FreshSessionCancellationDoesNotClaimAuthentication(t *testing.T) { + x := newWebH3SelectionFixture(t) + physical, _, err := x.client.connection(x.f.ctx) + if err != nil || physical == nil { + t.Fatalf("real owned H3 initialization without authentication: %v", err) + } + x.client.mu.Lock() + session := x.client.conns[physical] + fresh := session != nil && session.authState == webH3ClientAuthFresh && session.users == 0 && !session.retired && physical.Context().Err() == nil + x.client.mu.Unlock() + if !fresh || x.verifyCalls.Load() != 1 || x.entropy.nonceReads.Load() != 0 || x.dials.Load() != 0 { + t.Fatal("actual verified initialized Fresh H3 connection was not established before scheduling") + } + x.canceledAtPreflight(func(ctx context.Context) error { + conn, err := x.client.DialContext(ctx, "tcp", x.target) + if conn != nil { + _ = conn.Close() + t.Error("canceled Fresh selection returned an application stream") + } + return err + }) + x.client.mu.Lock() + preserved := x.client.conn == physical && x.client.conns[physical] == session && session.authState == webH3ClientAuthFresh && session.users == 0 && !session.retired && physical.Context().Err() == nil + x.client.mu.Unlock() + if !preserved || x.entropy.nonceReads.Load() != 0 || x.dials.Load() != 0 || x.covers.Load() != 0 { + t.Errorf("canceled Fresh selection changed physical/authentication state: preserved=%t nonce/destination/cover=%d/%d/%d", preserved, x.entropy.nonceReads.Load(), x.dials.Load(), x.covers.Load()) + } + // This public path causally exercises connection's mutex recheck. The + // narrower return-to-reserve mutex window remains a source-order defense, + // not a separately injected scheduling claim in this test. + x.healthy(physical) +} diff --git a/internal/tunnel/web_handler.go b/internal/tunnel/web_handler.go index ee16fdf..c102a0f 100644 --- a/internal/tunnel/web_handler.go +++ b/internal/tunnel/web_handler.go @@ -64,6 +64,11 @@ func (h *webTunnelHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { dialCtx, cancel := context.WithTimeout(r.Context(), h.core.dialTimeout) upstream, err := h.core.dialer.DialContext(dialCtx, "tcp", r.Host) + // The dialer's successful result is not accepted after the opening budget. + // Read its cause before cancel() so healthy results remain detached. + if err == nil { + err = contextError(dialCtx) + } cancel() if err != nil || upstream == nil { // Keep cleanup errors out of the authenticated failure response.