diff --git a/docs/WEB_COVER.md b/docs/WEB_COVER.md index 25dc3aa..7531091 100644 --- a/docs/WEB_COVER.md +++ b/docs/WEB_COVER.md @@ -416,6 +416,14 @@ 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. +Source builds also preserve the opening context's cancellation cause for H3 +CONNECT-UDP I/O failures and recheck cancellation after verifying a successful +TCP or UDP response, including at the final stream handoff lock. A rejected +handoff cancels only that request stream and releases its reservation; proven +connection authentication and sibling streams remain usable. Cancellation +after successful ownership transfer does not close the established stream. +These changes are not included in the published v1.0.1 binary. + 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 diff --git a/internal/tunnel/web_h3_client.go b/internal/tunnel/web_h3_client.go index e20141e..cb9348a 100644 --- a/internal/tunnel/web_h3_client.go +++ b/internal/tunnel/web_h3_client.go @@ -378,9 +378,10 @@ func (c *WebH3Client) DialContext(ctx context.Context, network, address string) fail() return nil, &WebConnectError{Transport: webAuthTransportH3, StatusCode: response.StatusCode} } - if !stopDial() && ctx.Err() != nil { + stopDial() + if err := contextError(ctx); err != nil { cancelStream() - return nil, ctx.Err() + return nil, err } if err := stream.SetDeadline(time.Time{}); err != nil { cancelStream() @@ -392,6 +393,13 @@ func (c *WebH3Client) DialContext(ctx context.Context, network, address string) cancelStream() return nil, net.ErrClosed } + // Stopping the opening watcher does not transfer ownership. Cancellation + // while waiting for this lock must still reclaim only this request stream. + if err := contextError(ctx); err != nil { + c.mu.Unlock() + cancelStream() + return nil, err + } c.selected = true c.mu.Unlock() result := newWebH3Conn(stream, conn.LocalAddr(), conn.RemoteAddr()) diff --git a/internal/tunnel/web_h3_stream_handoff_test.go b/internal/tunnel/web_h3_stream_handoff_test.go new file mode 100644 index 0000000..66688a0 --- /dev/null +++ b/internal/tunnel/web_h3_stream_handoff_test.go @@ -0,0 +1,347 @@ +package tunnel + +import ( + "bytes" + "context" + "crypto/tls" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/netip" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cppla/autocar/internal/transport" +) + +// Done always returns the exact real context's channel. With a Fresh session, +// the first Done evaluation after verified bootstrap becomes Ready is the +// standard context.AfterFunc stop's removeChild -> parentCancelCtx evaluation. +// The pinned Go 1.25/1.27 stop has already claimed its once at that point; this +// gate delays its return, not cancellation publication or a protocol response. +type webH3StreamHandoffContext struct { + context.Context + client *WebH3Client + entered chan struct{} + release <-chan struct{} + fixture context.Context + gated atomic.Bool + expired atomic.Bool +} + +func (c *webH3StreamHandoffContext) Done() <-chan struct{} { + done := c.Context.Done() + c.client.mu.Lock() + session := c.client.conns[c.client.conn] + ready := session != nil && session.authState == webH3ClientAuthReady && session.auth != nil + c.client.mu.Unlock() + if ready && c.gated.CompareAndSwap(false, true) { + close(c.entered) + select { + case <-c.release: + case <-c.fixture.Done(): + c.expired.Store(true) + } + } + return done +} + +type webH3StreamHandoffObservation struct { + connection *webServerConnectionAuth + full, short bool + proto int + tlsVersion uint16 + proof bool + joined <-chan struct{} +} + +type webH3StreamHandoffFixture struct { + f *webH2DialSharingFixture + client *WebH3Client + server *WebH3Server + tcp string + udp netip.AddrPort + entropy *webH2AuthEntropyCounter + observed chan webH3StreamHandoffObservation + covers, unexpected atomic.Int32 +} + +func newWebH3StreamHandoffFixture(t *testing.T) *webH3StreamHandoffFixture { + t.Helper() + f := newWebH2DialSharingFixture(t) // Owns bounded cleanup and every worker join. + x := &webH3StreamHandoffFixture{f: f, tcp: f.target(), entropy: &webH2AuthEntropyCounter{}, observed: make(chan webH3StreamHandoffObservation, 3)} + echo, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatal(err) + } + x.udp = echo.LocalAddr().(*net.UDPAddr).AddrPort() + f.close = append(f.close, func() { _ = echo.Close() }) + echoJoined, echoResult := make(chan struct{}), make(chan error, 1) + f.register(echoJoined) + f.checks = append(f.checks, func() { + if err := <-echoResult; err != nil { + t.Errorf("owned UDP echo: %v", err) + } + }) + go func() { + defer close(echoJoined) + buffer := make([]byte, 2048) + for { + if err := echo.SetDeadline(time.Now().Add(4 * time.Second)); err != nil { + echoResult <- err + return + } + n, source, err := echo.ReadFromUDPAddrPort(buffer) + if err != nil { + if errors.Is(err, net.ErrClosed) { + err = nil + } + echoResult <- err + return + } + if _, err := echo.WriteToUDPAddrPort(buffer[:n], source); err != nil { + echoResult <- err + return + } + } + }() + serverTLS, clientTLS := testTLSConfigs(t) + x.server, err = ListenWebH3(WebH3ServerConfig{ + Address: "127.0.0.1:0", Token: webTestToken, TLSConfig: serverTLS, + DialTimeout: time.Second, HandshakeTimeout: 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) { + if network != "tcp" || address != x.tcp { + x.unexpected.Add(1) + return nil, errors.New("non-owned TCP destination") + } + return (&net.Dialer{Timeout: time.Second}).DialContext(ctx, "tcp4", x.tcp) + }), + UDPResolver: UDPResolverFunc(func(_ context.Context, address string) ([]netip.AddrPort, error) { + if address != x.udp.String() { + x.unexpected.Add(1) + return nil, errors.New("non-owned UDP destination") + } + return []netip.AddrPort{x.udp}, nil + }), + }) + if err != nil { + t.Fatal(err) + } + original := x.server.server.Handler + x.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 := webH3StreamHandoffObservation{connection: connection, full: len(bearer) >= 385 && len(bearer) <= 2047, short: short, + proto: r.ProtoMajor, tlsVersion: webRequestTLSVersion(r), proof: w.Header().Get(webAuthResponseHeader) != "", joined: joined} + select { + case x.observed <- observation: + default: + x.unexpected.Add(1) + } + }) + serveJoined, serveResult := make(chan struct{}), make(chan error, 1) + f.register(serveJoined) + f.close = append(f.close, func() { + if err := normalizeWebServerCloseError(x.server.Close()); err != nil { + t.Errorf("owned H3 server Close: %v", err) + } + }) + f.checks = append(f.checks, func() { + if err := <-serveResult; err != nil { + t.Errorf("owned H3 Serve: %v", err) + } + }) + go func() { defer close(serveJoined); serveResult <- x.server.Serve(f.ctx) }() + x.client, err = NewWebH3Client(WebH3ClientConfig{ + ServerAddress: x.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 H3 client Close: %v", err) + } + }) + return x +} + +func (x *webH3StreamHandoffFixture) observation() webH3StreamHandoffObservation { + x.f.t.Helper() + select { + case observation := <-x.observed: + x.f.wait(observation.joined, "actual signed H3 handler completed") + if observation.connection == nil || !observation.proof || observation.proto != 3 || observation.tlsVersion != tls.VersionTLS13 { + x.f.t.Errorf("incomplete real H3 authentication observation: %+v", observation) + } + return observation + case <-x.f.ctx.Done(): + x.f.t.Fatal("owned H3 handler observation exceeded fixture budget") + return webH3StreamHandoffObservation{} + } +} + +func (x *webH3StreamHandoffFixture) healthyTCP() { + t := x.f.t + ctx, cancel := context.WithCancelCause(x.f.ctx) + defer cancel(context.Canceled) + conn, err := x.client.DialContext(ctx, "tcp", x.tcp) + if err != nil { + t.Fatal(err) + } + x.f.close = append(x.f.close, func() { _ = conn.Close() }) + // Cancellation after successful API handoff must not kill this live stream. + cancel(errors.New("post-success TCP caller no longer owns the stream")) + if err := conn.SetDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + payload := []byte("complete-detached-H3-TCP-payload") + _, writeErr := conn.Write(payload) + finErr := conn.(*webH3Conn).CloseWrite() + body, readErr := io.ReadAll(io.LimitReader(conn, int64(len(payload))+1)) + closeErr := conn.Close() + if writeErr != nil || finErr != nil || readErr != nil || closeErr != nil || !bytes.Equal(body, payload) { + t.Fatalf("detached TCP echo=%q errors=%v/%v/%v/%v", body, writeErr, finErr, readErr, closeErr) + } +} + +func (x *webH3StreamHandoffFixture) healthyUDP() { + t := x.f.t + ctx, cancel := context.WithCancelCause(x.f.ctx) + defer cancel(context.Canceled) + opened, err := x.client.openConnectUDPSession(ctx, x.udp.String()) + if err != nil { + t.Fatal(err) + } + x.f.close = append(x.f.close, opened.close) + cancel(errors.New("post-success UDP caller no longer owns the stream")) + payload := []byte("complete-detached-H3-UDP-payload") + type datagramResult struct { + frame []byte + err error + } + result, joined := make(chan datagramResult, 1), make(chan struct{}) + x.f.register(joined) + go func() { + defer close(joined) + if err := opened.stream.SendDatagram(encodeConnectUDPDatagram(payload)); err != nil { + result <- datagramResult{err: err} + return + } + readCtx, stop := context.WithTimeout(x.f.ctx, time.Second) + defer stop() + frame, err := opened.stream.ReceiveDatagram(readCtx) + result <- datagramResult{frame, err} + }() + x.f.wait(joined, "actual detached UDP datagram sender and receiver") + got := <-result + body, ok := parseConnectUDPDatagram(got.frame) + if got.err != nil || !ok || !bytes.Equal(body, payload) { + t.Fatalf("detached UDP echo=%q parsed=%t error=%v", body, ok, got.err) + } + opened.close() +} + +func webH3StreamHandoffCanceled(t *testing.T, udp bool) { + x := newWebH3StreamHandoffFixture(t) + ctx, cancel := context.WithCancelCause(x.f.ctx) + defer cancel(context.Canceled) + release := make(chan struct{}) + ungate := sync.OnceFunc(func() { close(release) }) + witness := &webH3StreamHandoffContext{Context: ctx, client: x.client, entered: make(chan struct{}), release: release, fixture: x.f.ctx} + // Appended last: even Fatal cleanup releases Done before client.Close. + x.f.close = append(x.f.close, ungate) + type openResult struct { + tcp net.Conn + udp *webConnectUDPStream + err error + } + result, joined := make(chan openResult, 1), make(chan struct{}) + x.f.register(joined) + go func() { + defer close(joined) + var got openResult + if udp { + got.udp, got.err = x.client.openConnectUDPSession(witness, x.udp.String()) + } else { + got.tcp, got.err = x.client.DialContext(witness, "tcp", x.tcp) + } + result <- got + }() + x.f.wait(witness.entered, "real AfterFunc stop held after first verified signed 200") + x.client.mu.Lock() + physical := x.client.conn + session := x.client.conns[physical] + ready := session != nil && session.authState == webH3ClientAuthReady && session.auth != nil && session.users == 1 + x.client.mu.Unlock() + if !ready || physical == nil || x.entropy.nonceReads.Load() != 1 { + t.Fatal("stop gate was not after exactly one real Fresh bootstrap") + } + state := physical.ConnectionState().TLS + if state.Version != tls.VersionTLS13 || state.NegotiatedProtocol != "h3" || len(state.VerifiedChains) == 0 { + t.Fatal("gated physical TLS was not actually verified TLS1.3/H3") + } + cause := errors.New("private cancellation after verified H3 stream proof") + cancel(cause) + // Do not call witness.Done here: the unmodified real context is the signal. + x.f.wait(ctx.Done(), "real original caller cancellation published") + ungate() + x.f.wait(joined, "post-proof canceled stream opener") + got := <-result + if got.tcp != nil || got.udp != nil || got.err != cause { + t.Errorf("post-proof handoff tcp=%t udp=%t err=%v, want nil objects and exact private cause", got.tcp != nil, got.udp != nil, got.err) + } + x.client.mu.Lock() + current, users, readyAfter, selected := x.client.conn, session.users, session.authState == webH3ClientAuthReady, x.client.selected + x.client.mu.Unlock() + if current != physical || !readyAfter || users != 0 || selected { + t.Errorf("rejected handoff changed ownership/auth/selection: same=%t ready=%t users=%d selected=%t", current == physical, readyAfter, users, selected) + } + if witness.expired.Load() { + t.Error("Done gate was released by fixture expiry rather than explicit test release") + } + // Independent disposal preserves bounded old-source failure and allows the + // later real healthy controls to run even if a live object leaked to caller. + if got.tcp != nil { + _ = got.tcp.Close() + } + if got.udp != nil { + got.udp.close() + } + first := x.observation() + if !first.full || first.short { + t.Error("first real request was not the one full connection credential") + } + x.healthyTCP() + second := x.observation() + x.healthyUDP() + third := x.observation() + x.client.mu.Lock() + current, users = x.client.conn, session.users + x.client.mu.Unlock() + if current != physical || users != 0 || physical.Context().Err() != nil || !second.short || !third.short || + first.connection != second.connection || first.connection != third.connection || x.entropy.nonceReads.Load() != 1 || x.covers.Load() != 0 || x.unexpected.Load() != 0 { + t.Errorf("later healthy TCP/UDP reuse failed: same=%t users=%d physicalErr=%v short=%t/%t fullNonces=%d cover/unexpected=%d/%d", current == physical, users, physical.Context().Err(), second.short, third.short, x.entropy.nonceReads.Load(), x.covers.Load(), x.unexpected.Load()) + } + t.Log(fmt.Sprintf("real signed %s bootstrap stop gate, exact caller cause, healthy short TCP/UDP complete echoes and detached cancellation controls evaluated", map[bool]string{false: "TCP", true: "UDP"}[udp])) +} + +func TestWebH3TCPStreamHandoffPreservesCallerCancellation(t *testing.T) { + webH3StreamHandoffCanceled(t, false) +} + +func TestWebH3UDPStreamHandoffPreservesCallerCancellation(t *testing.T) { + webH3StreamHandoffCanceled(t, true) +} diff --git a/internal/tunnel/web_udp.go b/internal/tunnel/web_udp.go index 555b919..20dcb93 100644 --- a/internal/tunnel/web_udp.go +++ b/internal/tunnel/web_udp.go @@ -888,9 +888,10 @@ func (c *WebH3Client) openConnectUDPSession(ctx context.Context, target string) if reservation.bootstrap { c.failSessionAuthentication(session) } - if ctx.Err() == nil { - c.retire(conn) + if cause := contextError(ctx); cause != nil { + return nil, cause } + c.retire(conn) return nil, fmt.Errorf("tunnel: open CONNECT-UDP request stream: %w", err) } cancelStream := func() { @@ -936,6 +937,9 @@ func (c *WebH3Client) openConnectUDPSession(ctx context.Context, target string) if reservation.bootstrap { c.failSessionAuthentication(session) } + if cause := contextError(ctx); cause != nil { + return nil, cause + } return nil, fmt.Errorf("tunnel: send CONNECT-UDP request for %s: %w", canonicalTarget, err) } response, err := readWebH3FinalResponse(stream) @@ -944,6 +948,9 @@ func (c *WebH3Client) openConnectUDPSession(ctx context.Context, target string) if reservation.bootstrap { c.failSessionAuthentication(session) } + if cause := contextError(ctx); cause != nil { + return nil, cause + } if conn.Context().Err() != nil { c.retire(conn) } @@ -980,9 +987,10 @@ func (c *WebH3Client) openConnectUDPSession(ctx context.Context, target string) fail() return nil, errors.New("tunnel: CONNECT-UDP response did not negotiate the Capsule Protocol") } - if !stopOpen() && ctx.Err() != nil { + stopOpen() + if err := contextError(ctx); err != nil { cancelStream() - return nil, ctx.Err() + return nil, err } if err := stream.SetDeadline(time.Time{}); err != nil { cancelStream() @@ -994,6 +1002,13 @@ func (c *WebH3Client) openConnectUDPSession(ctx context.Context, target string) cancelStream() return nil, net.ErrClosed } + // A verified response establishes authentication, but stream ownership is + // not handed off until cancellation is checked under the selection lock. + if err := contextError(ctx); err != nil { + c.mu.Unlock() + cancelStream() + return nil, err + } c.selected = true c.mu.Unlock() owned = false diff --git a/internal/tunnel/web_udp_open_cancellation_test.go b/internal/tunnel/web_udp_open_cancellation_test.go new file mode 100644 index 0000000..70d76d7 --- /dev/null +++ b/internal/tunnel/web_udp_open_cancellation_test.go @@ -0,0 +1,386 @@ +package tunnel + +import ( + "bytes" + "context" + "crypto/tls" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/netip" + "sync/atomic" + "testing" + "time" + + "github.com/cppla/autocar/internal/transport" +) + +type webUDPOpenCancellationRequest struct { + auth *webServerConnectionAuth + full bool + short bool + udp bool + proto int + tls uint16 + proof bool // Read only after joined closes. + joined chan struct{} +} + +type webUDPOpenCancellationFixture struct { + f *webH2DialSharingFixture + client *WebH3Client + handler *webTunnelHandler + tcpTarget string + udpTarget netip.AddrPort + entropy *webH2AuthEntropyCounter + requests chan *webUDPOpenCancellationRequest + resolveEntered chan struct{} + resolveJoined chan struct{} + requestCalls atomic.Int32 + resolves atomic.Int32 + tcpDials atomic.Int32 + covers atomic.Int32 + unexpected atomic.Int32 + verified atomic.Int32 +} + +// Reuse only the frozen owned-socket/deadline/join infrastructure. The first +// real authenticated CONNECT-UDP resolver blocks before any response headers; +// no transport error, authentication result, or response is synthesized. +func newWebUDPOpenCancellationFixture(t *testing.T) *webUDPOpenCancellationFixture { + t.Helper() + f := newWebH2DialSharingFixture(t) + x := &webUDPOpenCancellationFixture{ + f: f, tcpTarget: f.target(), entropy: &webH2AuthEntropyCounter{}, + requests: make(chan *webUDPOpenCancellationRequest, 4), + resolveEntered: make(chan struct{}), resolveJoined: make(chan struct{}), + } + echo, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatal(err) + } + x.udpTarget = echo.LocalAddr().(*net.UDPAddr).AddrPort() + f.close = append(f.close, func() { _ = echo.Close() }) + echoJoined := make(chan struct{}) + f.register(echoJoined) + var echoErr error + f.checks = append(f.checks, func() { + if echoErr != nil { + t.Errorf("owned UDP echo worker: %v", echoErr) + } + }) + go func() { + defer close(echoJoined) + buffer := make([]byte, 2048) + for { + if err := echo.SetDeadline(time.Now().Add(7 * time.Second)); err != nil { + echoErr = err + return + } + n, source, err := echo.ReadFromUDPAddrPort(buffer) + if err != nil { + if !errors.Is(err, net.ErrClosed) { + echoErr = err + } + return + } + if _, err := echo.WriteToUDPAddrPort(buffer[:n], source); err != nil { + if !errors.Is(err, net.ErrClosed) { + echoErr = err + } + return + } + } + }() + serverTLS, clientTLS := testTLSConfigs(t) + server, err := ListenWebH3(WebH3ServerConfig{ + Address: "127.0.0.1:0", Token: webTestToken, TLSConfig: serverTLS, + HandshakeTimeout: 2 * time.Second, DialTimeout: 2 * time.Second, + MaxUDPSessions: 2, MaxClientUDPSessions: 2, + 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.tcpDials.Add(1) + if network != "tcp" || address != x.tcpTarget { + x.unexpected.Add(1) + return nil, errors.New("UDP cancellation fixture refuses a non-owned TCP target") + } + return (&net.Dialer{Timeout: time.Second}).DialContext(ctx, "tcp4", x.tcpTarget) + }), + UDPResolver: UDPResolverFunc(func(ctx context.Context, address string) ([]netip.AddrPort, error) { + call := x.resolves.Add(1) + joined := make(chan struct{}) + f.register(joined) + defer close(joined) + if address != x.udpTarget.String() { + x.unexpected.Add(1) + return nil, errors.New("UDP cancellation fixture refuses a non-owned numeric target") + } + if call == 1 { + defer close(x.resolveJoined) + close(x.resolveEntered) + select { + case <-ctx.Done(): + return nil, context.Cause(ctx) + case <-f.ctx.Done(): + return nil, context.Cause(f.ctx) + } + } + if call != 2 { + x.unexpected.Add(1) + } + return []netip.AddrPort{x.udpTarget}, nil + }), + }) + if err != nil { + t.Fatal(err) + } + x.handler = server.server.Handler.(*webTunnelHandler) + original := server.server.Handler + server.server.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + x.requestCalls.Add(1) + joined := make(chan struct{}) + f.register(joined) + defer close(joined) + auth, _ := r.Context().Value(webServerConnectionAuthContextKey{}).(*webServerConnectionAuth) + bearer := r.Header.Get("Proxy-Authorization") + _, short := parseWebSessionBearer(bearer) + request := &webUDPOpenCancellationRequest{ + auth: auth, full: len(bearer) >= 385 && len(bearer) <= 2047, short: short, + udp: r.Proto == webConnectUDPProtocol, proto: r.ProtoMajor, + tls: webRequestTLSVersion(r), joined: joined, + } + select { + case x.requests <- request: + default: + x.unexpected.Add(1) + } + original.ServeHTTP(w, r) + request.proof = w.Header().Get(webAuthResponseHeader) != "" + }) + serveJoined := make(chan struct{}) + f.register(serveJoined) + var serveErr error + f.checks = append(f.checks, func() { + if serveErr != nil { + t.Errorf("owned UDP cancellation H3 Serve: %v", serveErr) + } + }) + f.close = append(f.close, func() { + if err := server.Close(); err != nil && !errors.Is(err, net.ErrClosed) { + t.Errorf("owned UDP cancellation H3 server Close: %v", err) + } + }) + go func() { defer close(serveJoined); serveErr = server.Serve(f.ctx) }() + clientTLS = clientTLS.Clone() + clientTLS.VerifyConnection = func(state tls.ConnectionState) error { + if state.Version != tls.VersionTLS13 || state.NegotiatedProtocol != "h3" || len(state.VerifiedChains) == 0 { + return errors.New("owned UDP cancellation TLS verification is incomplete") + } + x.verified.Add(1) + return nil + } + x.client, err = NewWebH3Client(WebH3ClientConfig{ + ServerAddress: server.Addr().String(), Token: webTestToken, TLSConfig: clientTLS, + FingerprintProfile: H3FingerprintNative, DialTimeout: time.Second, HandshakeTimeout: 2 * 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 UDP cancellation H3 client Close: %v", err) + } + }) + return x +} + +func (x *webUDPOpenCancellationFixture) request(udp, full bool, auth *webServerConnectionAuth) *webUDPOpenCancellationRequest { + x.f.t.Helper() + var request *webUDPOpenCancellationRequest + select { + case request = <-x.requests: + case <-x.f.ctx.Done(): + x.f.t.Fatal("actual authenticated H3 request observation exceeded fixture budget") + } + if request.udp != udp || request.full != full || request.short == full || request.auth == nil || + (auth != nil && request.auth != auth) || request.proto != 3 || request.tls != tls.VersionTLS13 { + x.f.t.Errorf("actual request UDP/full/short/auth/proto/TLS=%t/%t/%t/%t/%d/%x", request.udp, request.full, request.short, request.auth == auth, request.proto, request.tls) + } + return request +} + +func (x *webUDPOpenCancellationFixture) tcpEcho(conn net.Conn, payload string, fin bool) { + x.f.t.Helper() + if err := conn.SetDeadline(time.Now().Add(time.Second)); err != nil { + x.f.t.Fatal(err) + } + if n, err := conn.Write([]byte(payload)); err != nil || n != len(payload) { + x.f.t.Fatalf("owned TCP sibling write=%d/%d, %v", n, len(payload), err) + } + if fin { + if err := conn.(*webH3Conn).CloseWrite(); err != nil { + x.f.t.Fatal(err) + } + } + got := make([]byte, len(payload)) + if _, err := io.ReadFull(conn, got); err != nil || string(got) != payload { + x.f.t.Fatalf("owned TCP sibling echo=%q, %v", got, err) + } + if fin { + extra, err := io.ReadAll(io.LimitReader(conn, 1)) + if err != nil || len(extra) != 0 { + x.f.t.Fatalf("owned TCP sibling FIN remainder=%q, %v", extra, err) + } + } +} + +func (x *webUDPOpenCancellationFixture) state(session *webH3ClientSession, users int) { + x.f.t.Helper() + x.client.mu.Lock() + same := x.client.conn == session.conn && x.client.conns[session.conn] == session + ready, retired, gotUsers := session.authState == webH3ClientAuthReady, session.retired, session.users + x.client.mu.Unlock() + session.auth.mu.Lock() + pending := len(session.auth.pending) + session.auth.mu.Unlock() + if !same || !ready || retired || gotUsers != users || pending != 0 || session.conn.Context().Err() != nil { + x.f.t.Errorf("same physical/Ready/retired/users/pending/live=%t/%t/%t/%d/%d/%t; want users=%d", same, ready, retired, gotUsers, pending, session.conn.Context().Err() == nil, users) + } +} + +func (x *webUDPOpenCancellationFixture) healthyUDP() *webUDPOpenCancellationRequest { + x.f.t.Helper() + packet, err := x.client.DialPacket(x.f.ctx) + if err != nil { + x.f.t.Fatal(err) + } + x.f.close = append(x.f.close, func() { _ = packet.Close() }) + result, joined := make(chan error, 1), make(chan struct{}) + x.f.register(joined) + payload := []byte("healthy-owned-UDP-after-opening-cancellation") + go func() { + defer close(joined) + if err := packet.Send(payload, x.udpTarget.String()); err != nil { + result <- err + return + } + got, address, err := packet.Receive() + if err == nil && (!bytes.Equal(got, payload) || address != x.udpTarget.String()) { + err = fmt.Errorf("owned UDP echo=%q address=%q", got, address) + } + result <- err + }() + x.f.wait(joined, "later public PacketConn Send/Receive worker") + if err := <-result; err != nil { + x.f.t.Fatal(err) + } + request := x.request(true, false, nil) + if err := packet.Close(); err != nil { + x.f.t.Fatal(err) + } + x.f.wait(request.joined, "later healthy CONNECT-UDP server workers") + if !request.proof { + x.f.t.Error("later healthy CONNECT-UDP response lacked authenticated proof") + } + return request +} + +// This tests the internal opening-context contract, not a per-Send context: +// PacketConn.Send has no context argument, and successful DialPacket detaches +// its caller context from the logical packet association's lifetime. +func TestWebH3ConnectUDPReadCancellationPreservesCallerCause(t *testing.T) { + for _, mode := range []string{"private_cancel", "operation_deadline"} { + t.Run(mode, func(t *testing.T) { + x := newWebUDPOpenCancellationFixture(t) + sibling, err := x.client.DialContext(x.f.ctx, "tcp", x.tcpTarget) + if err != nil { + t.Fatal(err) + } + x.f.close = append(x.f.close, func() { _ = sibling.Close() }) + warm := x.request(false, true, nil) + x.tcpEcho(sibling, "live-sibling-before-UDP-cancel", false) + x.client.mu.Lock() + session := x.client.conns[x.client.conn] + x.client.mu.Unlock() + if session == nil || session.auth == nil { + t.Fatal("healthy TCP bootstrap did not establish authentication") + } + x.state(session, 1) + ctx, cancelCause := context.WithCancelCause(x.f.ctx) + defer cancelCause(context.Canceled) + var expected error = errors.New("private CONNECT-UDP response-wait cancellation") + if mode == "operation_deadline" { + var cancelDeadline context.CancelFunc + ctx, cancelDeadline = context.WithTimeout(ctx, 500*time.Millisecond) + defer cancelDeadline() + expected = context.DeadlineExceeded + } + type openResult struct { + err error + hadStream bool + } + result, joined := make(chan openResult, 1), make(chan struct{}) + x.f.register(joined) + go func() { + defer close(joined) + stream, err := x.client.openConnectUDPSession(ctx, x.udpTarget.String()) + if stream != nil { + stream.close() // Independently release any unexpected success. + } + result <- openResult{err: err, hadStream: stream != nil} + }() + x.f.wait(x.resolveEntered, "real authenticated CONNECT-UDP reached owned blocked resolver") + failed := x.request(true, false, warm.auth) + if ctx.Err() != nil { + t.Fatal("opening context expired before the actual response-wait cancellation witness") + } + if mode == "private_cancel" { + cancelCause(expected) + } + x.f.wait(joined, "canceled CONNECT-UDP opening worker") + got := <-result + if got.hadStream || got.err != expected { + t.Errorf("canceled CONNECT-UDP returned stream=%t error=%T %v; want exact caller cause %v", got.hadStream, got.err, got.err, expected) + } + x.f.wait(x.resolveJoined, "canceled real resolver callback") + x.f.wait(failed.joined, "canceled real CONNECT-UDP server handler") + x.state(session, 1) + if slots := len(x.handler.udp.slots); slots != 0 { + t.Errorf("canceled CONNECT-UDP retained %d server UDP admission slots", slots) + } + x.tcpEcho(sibling, "live-sibling-after-UDP-cancel", false) + healthy := x.healthyUDP() + if healthy.auth != warm.auth { + t.Error("later healthy UDP did not retain the original authenticated server connection") + } + x.state(session, 1) + later, err := x.client.DialContext(x.f.ctx, "tcp", x.tcpTarget) + if err != nil { + t.Fatal(err) + } + x.f.close = append(x.f.close, func() { _ = later.Close() }) + laterRequest := x.request(false, false, warm.auth) + x.tcpEcho(later, "later-TCP-complete-FIN-echo", true) + _ = later.Close() + x.f.wait(laterRequest.joined, "later healthy TCP server forwarding worker") + x.tcpEcho(sibling, "original-sibling-complete-FIN-echo", true) + _ = sibling.Close() + x.f.wait(warm.joined, "original live TCP sibling server forwarding worker") + x.state(session, 0) + if !warm.proof || !laterRequest.proof || x.entropy.nonceReads.Load() != 1 || x.verified.Load() != 1 || + x.requestCalls.Load() != 4 || x.resolves.Load() != 2 || x.tcpDials.Load() != 2 || x.covers.Load() != 0 || x.unexpected.Load() != 0 { + t.Errorf("healthy proof/nonce/TLS/requests/resolves/TCP/cover/unexpected=%t/%t/%d/%d/%d/%d/%d/%d/%d", warm.proof, laterRequest.proof, x.entropy.nonceReads.Load(), x.verified.Load(), x.requestCalls.Load(), x.resolves.Load(), x.tcpDials.Load(), x.covers.Load(), x.unexpected.Load()) + } + if len(x.handler.udp.slots) != 0 || len(x.handler.core.sem) != 0 || len(x.client.udpSlots) != 0 { + t.Error("completed owned flows retained server/client admission slots") + } + t.Log("real authenticated resolver cancellation joined; original TCP sibling survived; later complete TCP/UDP echoes reused one verified TLS/H3 connection and one full-authentication nonce") + }) + } +}