From 84e8d063d2a0725894364aabb84ec93bd95bcd48 Mon Sep 17 00:00:00 2001 From: cppla Date: Sat, 3 Oct 2026 01:48:11 +0800 Subject: [PATCH] fix(web): handle informational responses safely --- docs/WEB_COVER.md | 14 + internal/cover/handler.go | 24 +- internal/cover/informational_response_test.go | 265 +++++++++++++++ .../tunnel/web_cover_informational_test.go | 124 +++++++ internal/tunnel/web_h3_client.go | 2 +- .../web_h3_informational_success_test.go | 316 +++++++++++++++++ internal/tunnel/web_h3_informational_test.go | 319 ++++++++++++++++++ internal/tunnel/web_h3_response.go | 43 +++ internal/tunnel/web_h3_response_test.go | 262 ++++++++++++++ internal/tunnel/web_udp.go | 2 +- 10 files changed, 1368 insertions(+), 3 deletions(-) create mode 100644 internal/cover/informational_response_test.go create mode 100644 internal/tunnel/web_cover_informational_test.go create mode 100644 internal/tunnel/web_h3_informational_success_test.go create mode 100644 internal/tunnel/web_h3_informational_test.go create mode 100644 internal/tunnel/web_h3_response.go create mode 100644 internal/tunnel/web_h3_response_test.go diff --git a/docs/WEB_COVER.md b/docs/WEB_COVER.md index ca32105..e029590 100644 --- a/docs/WEB_COVER.md +++ b/docs/WEB_COVER.md @@ -126,6 +126,20 @@ remain available, and the body is still streamed rather than buffered in full. This is defensive handling of upstream metadata, not an additional tunnel authentication mechanism. +Source builds also apply that filter to upstream informational responses, +including `103 Early Hints`, before forwarding them. Ordinary `Link` hints and +other end-to-end fields remain available; credentials, hop-by-hop fields and +fields named by `Connection` do not bypass filtering through an early response. +This hardening is not included in v1.0.1. + +Source-built H3 clients also consume up to five non-final `1xx` responses before +checking the final CONNECT or CONNECT-UDP response and its authentication proof. +This permits ordinary informational responses without treating them as tunnel +authentication failures. A sixth informational response is rejected; each hint +does not restart the existing establishment deadline. Informational headers are +never accepted as a replacement for the final signed proof. This client fix is +not included in v1.0.1. + In web mode: - `--listen` is the UDP/H3 bind address and `--tcp-listen` is the TCP/H1/H2 diff --git a/internal/cover/handler.go b/internal/cover/handler.go index f069db1..8048ca8 100644 --- a/internal/cover/handler.go +++ b/internal/cover/handler.go @@ -8,7 +8,9 @@ import ( "io" "log" "net/http" + "net/http/httptrace" "net/http/httputil" + "net/textproto" "net/url" "os" "path/filepath" @@ -76,7 +78,7 @@ func NewReverseProxyHandler(origin *url.URL, transport http.RoundTripper) (http. } proxy := &httputil.ReverseProxy{ - Transport: transport, + Transport: &informationalHeaderTransport{base: transport}, Rewrite: func(request *httputil.ProxyRequest) { request.SetURL(target) request.Out.Host = target.Host @@ -99,6 +101,26 @@ func NewReverseProxyHandler(origin *url.URL, transport http.RoundTripper) (http. return proxy, nil } +// informationalHeaderTransport filters upstream headers before ReverseProxy's +// trace copies an informational response to the downstream writer. Its +// ModifyResponse hook only sees the final response. Install a fresh trace here, +// after ReverseProxy installs its own, so filtering runs before older hooks +// without changing their order or return values or wrapping the response writer. +type informationalHeaderTransport struct { + base http.RoundTripper +} + +func (t *informationalHeaderTransport) RoundTrip(request *http.Request) (*http.Response, error) { + trace := &httptrace.ClientTrace{ + Got1xxResponse: func(_ int, header textproto.MIMEHeader) error { + removeUnsafeHeaders(http.Header(header)) + return nil + }, + } + request = request.WithContext(httptrace.WithClientTrace(request.Context(), trace)) + return t.base.RoundTrip(request) +} + // responseTrailerBody filters fields that a transport discovers only at EOF // or Close, including replacement Trailer maps. It does not buffer the body. // Trailer must not be inspected while Read is in progress. Close may interrupt diff --git a/internal/cover/informational_response_test.go b/internal/cover/informational_response_test.go new file mode 100644 index 0000000..93f0b87 --- /dev/null +++ b/internal/cover/informational_response_test.go @@ -0,0 +1,265 @@ +package cover + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/http/httptrace" + "net/http/httputil" + "net/textproto" + "net/url" + "reflect" + "strings" + "testing" + "time" +) + +func TestReverseProxyScrubsInformationalResponsesOnWire(t *testing.T) { + for _, protocol := range []struct { + name string + h2 bool + }{ + {name: "h1"}, + {name: "h2", h2: true}, + } { + for _, explicitFinal := range []bool{true, false} { + t.Run(fmt.Sprintf("%s/explicit_final_%t", protocol.name, explicitFinal), func(t *testing.T) { + statuses := []int{http.StatusEarlyHints, http.StatusProcessing, http.StatusEarlyHints} + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + for index, status := range statuses { + for name, values := range informationalTestHeaders(index) { + w.Header()[name] = values + } + w.WriteHeader(status) + clear(w.Header()) + } + w.Header().Set("Authorization", "fictional-final-origin") + w.Header().Set("Proxy-Authorization", "fictional-final-proxy") + w.Header().Set("Connection", "X-Final-Hop") + w.Header().Set("X-Final-Hop", "fictional-final-hop") + w.Header().Set("X-Final-End-To-End", "retained-final") + w.Header().Set("Trailer", "X-End-Trailer") + if explicitFinal { + w.WriteHeader(http.StatusOK) + } + _, _ = io.WriteString(w, "ordinary cover body") + w.Header().Set("X-End-Trailer", "retained-trailer") + })) + t.Cleanup(origin.Close) + originURL, err := url.Parse(origin.URL) + if err != nil { + t.Fatal(err) + } + upstream := &http.Transport{Proxy: nil} + t.Cleanup(upstream.CloseIdleConnections) + handler, err := NewReverseProxyHandler(originURL, upstream) + if err != nil { + t.Fatal(err) + } + front := httptest.NewUnstartedServer(handler) + front.EnableHTTP2 = protocol.h2 + front.StartTLS() + t.Cleanup(front.Close) + client := front.Client() + client.Timeout = 3 * time.Second + var gotStatuses []int + var gotHeaders []http.Header + trace := &httptrace.ClientTrace{ + Got1xxResponse: func(status int, header textproto.MIMEHeader) error { + gotStatuses = append(gotStatuses, status) + gotHeaders = append(gotHeaders, http.Header(header).Clone()) + return nil + }, + } + request, err := http.NewRequest(http.MethodGet, front.URL, nil) + if err != nil { + t.Fatal(err) + } + request = request.WithContext(httptrace.WithClientTrace(request.Context(), trace)) + response, err := client.Do(request) + if err != nil { + t.Fatal(err) + } + body, err := io.ReadAll(response.Body) + _ = response.Body.Close() + if err != nil { + t.Fatal(err) + } + wantMajor := 1 + if protocol.h2 { + wantMajor = 2 + } + if response.ProtoMajor != wantMajor || response.StatusCode != http.StatusOK || string(body) != "ordinary cover body" { + t.Fatalf("response = %s %d %q, want HTTP/%d 200 with ordinary body", response.Proto, response.StatusCode, body, wantMajor) + } + if !reflect.DeepEqual(gotStatuses, statuses) { + t.Fatalf("informational statuses = %v, want %v", gotStatuses, statuses) + } + for index, header := range gotHeaders { + assertInformationalTestHeaders(t, header, index) + } + for _, name := range []string{"Authorization", "Proxy-Authorization", "Connection", "X-Final-Hop"} { + if value := response.Header.Get(name); value != "" { + t.Errorf("unsafe final %s escaped filtering: %q", name, value) + } + } + if response.Header.Get("X-Final-End-To-End") != "retained-final" { + t.Errorf("ordinary final field was lost: %v", response.Header) + } + if response.Trailer.Get("X-End-Trailer") != "retained-trailer" { + t.Errorf("ordinary trailer was lost: %v", response.Trailer) + } + }) + } + } +} + +func TestReverseProxyInformationalTracePreservesCallerAndTransport(t *testing.T) { + type contextKey struct{} + ctx, cancel := context.WithTimeout(context.WithValue(context.Background(), contextKey{}, "caller-value"), 3*time.Second) + t.Cleanup(cancel) + observerError := errors.New("observer error") + var observed []http.Header + var statuses []int + getConnCalls := 0 + callerTrace := &httptrace.ClientTrace{ + Got1xxResponse: func(status int, header textproto.MIMEHeader) error { + statuses = append(statuses, status) + observed = append(observed, http.Header(header).Clone()) + return observerError + }, + GetConn: func(string) { getConnCalls++ }, + } + callerContext := httptrace.WithClientTrace(ctx, callerTrace) + var requestBody io.ReadCloser + var gotTraces []*httptrace.ClientTrace + transportCalls := 0 + upstream := roundTripFunc(func(request *http.Request) (*http.Response, error) { + transportCalls++ + if request.Body != requestBody || request.Header.Get("X-Caller") != "retained" { + t.Error("transport wrapper changed the caller's body or headers") + } + if request.Context().Value(contextKey{}) != "caller-value" || request.Context().Done() != callerContext.Done() { + t.Error("transport wrapper lost the caller's context value or cancellation") + } + gotDeadline, gotOK := request.Context().Deadline() + wantDeadline, wantOK := callerContext.Deadline() + if gotDeadline != wantDeadline || gotOK != wantOK { + t.Error("transport wrapper changed the caller's deadline") + } + trace := httptrace.ContextClientTrace(request.Context()) + if trace == nil || trace.Got1xxResponse == nil { + t.Fatal("transport did not receive an informational trace") + } + if trace == callerTrace { + t.Error("transport reused the caller's trace instead of composing a request-local filter") + } + for _, previous := range gotTraces { + if trace == previous { + t.Error("transport reused a trace between requests") + } + } + gotTraces = append(gotTraces, trace) + if trace.GetConn == nil { + t.Error("transport wrapper lost an unrelated trace hook") + } else { + trace.GetConn("origin.invalid:80") + } + header := informationalTestHeaders(transportCalls - 1) + // A custom transport (and Go's H2 transport) can report 101 through + // Got1xxResponse. This checks that callback boundary, rather than + // claiming support for a legal H1 protocol upgrade. + status := http.StatusEarlyHints + if transportCalls == 2 { + status = http.StatusSwitchingProtocols + } + if err := trace.Got1xxResponse(status, textproto.MIMEHeader(header)); !errors.Is(err, observerError) { + t.Errorf("composed observer error = %v, want original error", err) + } + return &http.Response{StatusCode: http.StatusOK, Header: make(http.Header), Body: io.NopCloser(strings.NewReader("transport response"))}, nil + }) + handler, err := NewReverseProxyHandler(&url.URL{Scheme: "http", Host: "origin.invalid"}, upstream) + if err != nil { + t.Fatal(err) + } + proxy, ok := handler.(*httputil.ReverseProxy) + if !ok { + t.Fatalf("handler type = %T, want *httputil.ReverseProxy", handler) + } + for index := 0; index < 2; index++ { + request, err := http.NewRequestWithContext(callerContext, http.MethodPost, "http://origin.invalid/", strings.NewReader("caller body")) + if err != nil { + t.Fatal(err) + } + request.Header.Set("X-Caller", "retained") + requestBody = request.Body + response, err := proxy.Transport.RoundTrip(request) + if err != nil { + t.Fatal(err) + } + body, err := io.ReadAll(response.Body) + _ = response.Body.Close() + if err != nil || string(body) != "transport response" { + t.Fatalf("delegated response body/error = %q/%v", body, err) + } + if request.Context() != callerContext || httptrace.ContextClientTrace(request.Context()) != callerTrace { + t.Error("transport wrapper mutated the caller's request context or trace") + } + if len(observed) != index+1 || getConnCalls != index+1 { + t.Fatalf("old trace calls = informational %d / GetConn %d, want %d each", len(observed), getConnCalls, index+1) + } + assertInformationalTestHeaders(t, observed[index], index) + } + if transportCalls != 2 || !reflect.DeepEqual(statuses, []int{http.StatusEarlyHints, http.StatusSwitchingProtocols}) { + t.Fatalf("delegated transport/statuses = %d/%v, want unchanged 103 and 101 callback events", transportCalls, statuses) + } + // The same caller hook remains usable independently, without an injected + // sanitizer or a callback retained from either completed request. + originalHeaders := informationalTestHeaders(2) + if err := callerTrace.Got1xxResponse(http.StatusProcessing, textproto.MIMEHeader(originalHeaders)); !errors.Is(err, observerError) { + t.Errorf("original observer error = %v, want original error", err) + } + if len(observed) != 3 || observed[2].Get("Authorization") != "fictional-origin-credential" || originalHeaders.Get("Authorization") != "fictional-origin-credential" { + t.Error("transport wrapper modified the caller's original trace hook") + } +} + +func informationalTestHeaders(index int) http.Header { + return http.Header{ + "Authorization": {"fictional-origin-credential"}, + "Proxy-Authorization": {"fictional-proxy-credential"}, + "Connection": {"X-Origin-Hop", "X-Other-Hop, Keep-Alive"}, + "X-Origin-Hop": {"fictional-first-hop"}, + "X-Other-Hop": {"fictional-other-hop"}, + "Proxy-Connection": {"keep-alive"}, + "Keep-Alive": {"timeout=5"}, + "Proxy-Authenticate": {"Basic realm=fictional"}, + "Te": {"trailers"}, + "Trailer": {"X-Informational-Trailer"}, + "Transfer-Encoding": {"chunked"}, + "Upgrade": {"fictional-protocol"}, + "Link": {fmt.Sprintf("; rel=preload", index)}, + "X-End-To-End": {fmt.Sprintf("retained-%d", index)}, + } +} + +func assertInformationalTestHeaders(t *testing.T, header http.Header, index int) { + t.Helper() + for _, name := range []string{ + "Authorization", "Proxy-Authorization", "Connection", "X-Origin-Hop", "X-Other-Hop", + "Proxy-Connection", "Keep-Alive", "Proxy-Authenticate", "Te", "Trailer", "Transfer-Encoding", "Upgrade", + } { + if value := header.Get(name); value != "" { + t.Errorf("informational response %d unsafe %s = %q", index, name, value) + } + } + if got, want := header.Get("Link"), fmt.Sprintf("; rel=preload", index); got != want { + t.Errorf("informational response %d Link = %q, want %q", index, got, want) + } + if got, want := header.Get("X-End-To-End"), fmt.Sprintf("retained-%d", index); got != want { + t.Errorf("informational response %d end-to-end field = %q, want %q", index, got, want) + } +} diff --git a/internal/tunnel/web_cover_informational_test.go b/internal/tunnel/web_cover_informational_test.go new file mode 100644 index 0000000..98b9fd0 --- /dev/null +++ b/internal/tunnel/web_cover_informational_test.go @@ -0,0 +1,124 @@ +package tunnel + +import ( + "context" + "errors" + "io" + "net" + "net/http" + "net/http/httptest" + "net/http/httptrace" + "net/textproto" + "net/url" + "reflect" + "testing" + "time" + + "github.com/apernet/quic-go/http3" + "github.com/cppla/autocar/internal/cover" + "github.com/cppla/autocar/internal/transport" +) + +func TestWebH3CoverFiltersInformationalResponses(t *testing.T) { + statuses := []int{http.StatusEarlyHints, http.StatusProcessing, http.StatusEarlyHints} + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Proxy-Authorization") != "" || r.Header.Get("Authorization") != "" { + t.Error("request credentials reached the fixed origin") + } + for _, status := range statuses { + w.Header().Set("Authorization", "fictional-origin-credential") + w.Header().Set("Proxy-Authorization", "fictional-proxy-credential") + w.Header().Set("Connection", "X-Origin-Hop") + w.Header().Set("X-Origin-Hop", "fictional-hop-value") + w.Header().Set("Link", "; rel=preload") + w.WriteHeader(status) + clear(w.Header()) + } + w.Header().Set("Authorization", "fictional-final-credential") + w.Header().Set("Trailer", "X-End-Trailer") + _, _ = io.WriteString(w, "ordinary H3 cover") + w.Header().Set("X-End-Trailer", "retained") + })) + t.Cleanup(origin.Close) + target, err := url.Parse(origin.URL) + if err != nil { + t.Fatal(err) + } + upstream := &http.Transport{Proxy: nil} + t.Cleanup(upstream.CloseIdleConnections) + handler, err := cover.NewReverseProxyHandler(target, upstream) + if err != nil { + t.Fatal(err) + } + serverTLS, clientTLS := testTLSConfigs(t) + server, err := ListenWebH3(WebH3ServerConfig{ + Address: "127.0.0.1:0", Token: webTestToken, TLSConfig: serverTLS, Cover: handler, + Dialer: transport.DialFunc(func(context.Context, string, string) (net.Conn, error) { + t.Error("ordinary cover request attempted a tunnel destination") + return nil, errors.New("cover-only test rejects tunnel destinations") + }), + }) + if err != nil { + t.Fatal(err) + } + serverContext, stopServer := context.WithCancel(context.Background()) + serverDone := make(chan error, 1) + go func() { serverDone <- server.Serve(serverContext) }() + t.Cleanup(func() { + stopServer() + _ = server.Close() + select { + case err := <-serverDone: + if err != nil { + t.Errorf("H3 cover Serve: %v", err) + } + case <-time.After(2 * time.Second): + t.Error("H3 cover Serve did not join") + } + }) + clientTLS = clientTLS.Clone() + clientTLS.NextProtos = []string{http3.NextProtoH3} + client := &http3.Transport{TLSClientConfig: clientTLS} + t.Cleanup(func() { _ = client.Close() }) + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + var observed []int + trace := &httptrace.ClientTrace{ + Got1xxResponse: func(status int, header textproto.MIMEHeader) error { + observed = append(observed, status) + for _, name := range []string{"Authorization", "Proxy-Authorization", "Connection", "X-Origin-Hop"} { + if value := header.Get(name); value != "" { + t.Errorf("H3 informational %d leaked %s: %q", status, name, value) + } + } + if got := header.Get("Link"); got != "; rel=preload" { + t.Errorf("ordinary H3 hint lost: %q", got) + } + return nil + }, + } + request, err := http.NewRequestWithContext(httptrace.WithClientTrace(ctx, trace), http.MethodGet, "https://"+server.Addr().String()+"/", nil) + if err != nil { + t.Fatal(err) + } + request.Header.Set("Authorization", "fictional-request-credential") + request.Header.Set("Proxy-Authorization", "fictional-invalid-proxy-credential") + response, err := client.RoundTrip(request) + if err != nil { + t.Fatal(err) + } + body, readErr := io.ReadAll(response.Body) + _ = response.Body.Close() + if readErr != nil { + t.Fatal(readErr) + } + if response.ProtoMajor != 3 || response.StatusCode != http.StatusOK || string(body) != "ordinary H3 cover" { + t.Fatalf("H3 cover = %s %d %q", response.Proto, response.StatusCode, body) + } + if !reflect.DeepEqual(observed, statuses) { + t.Fatalf("H3 hints = %v, want %v", observed, statuses) + } + if response.Header.Get("Authorization") != "" || response.Trailer.Get("X-End-Trailer") != "retained" { + t.Fatalf("H3 final headers/trailers = %v/%v", response.Header, response.Trailer) + } +} diff --git a/internal/tunnel/web_h3_client.go b/internal/tunnel/web_h3_client.go index ac79fa5..d71a400 100644 --- a/internal/tunnel/web_h3_client.go +++ b/internal/tunnel/web_h3_client.go @@ -324,7 +324,7 @@ func (c *WebH3Client) DialContext(ctx context.Context, network, address string) } return nil, fmt.Errorf("tunnel: send web-cover H3 CONNECT: %w", err) } - response, err := stream.ReadResponse() + response, err := readWebH3FinalResponse(stream) if err != nil { fail() if reservation.bootstrap { diff --git a/internal/tunnel/web_h3_informational_success_test.go b/internal/tunnel/web_h3_informational_success_test.go new file mode 100644 index 0000000..866f6c0 --- /dev/null +++ b/internal/tunnel/web_h3_informational_success_test.go @@ -0,0 +1,316 @@ +package tunnel + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/netip" + "sync" + "testing" + "time" + + "github.com/cppla/autocar/internal/transport" +) + +type webH3InfoSuccessResponse struct { + proof string + bearerLen int +} + +func webH3InfoSuccessServer(t *testing.T, dialer transport.Dialer, resolver UDPResolver) (*WebH3Server, *webTunnelHandler, <-chan webH3InfoSuccessResponse, *webH3InformationalClient) { + t.Helper() + serverTLS, clientTLS := testTLSConfigs(t) + server, err := ListenWebH3(WebH3ServerConfig{ + Address: "127.0.0.1:0", Token: webTestToken, TLSConfig: serverTLS, + Cover: http.NotFoundHandler(), Dialer: dialer, UDPResolver: resolver, + }) + if err != nil { + t.Fatal(err) + } + handler := server.server.Handler.(*webTunnelHandler) + responses := make(chan webH3InfoSuccessResponse, 2) + server.server.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + for _, status := range []int{100, 102, 103} { + w.Header().Set("Link", "; rel=preload") + w.WriteHeader(status) + clear(w.Header()) + } + handler.ServeHTTP(w, r) + responses <- webH3InfoSuccessResponse{proof: w.Header().Get(webAuthResponseHeader), bearerLen: len(r.Header.Get("Proxy-Authorization"))} + }) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- server.Serve(ctx) }() + t.Cleanup(func() { + _ = server.Close() + cancel() + select { + case err := <-done: + if err != nil { + t.Errorf("successful informational server Serve: %v", err) + } + case <-time.After(2 * time.Second): + t.Error("successful informational server Serve did not join") + } + }) + client := newWebH3InformationalClient(t, "h3-tcp", server.Addr().String(), clientTLS, time.Second) + return server, handler, responses, client +} + +func webH3InfoSuccessWait(t *testing.T, done <-chan struct{}, label string) { + t.Helper() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Errorf("%s did not join", label) + } +} + +// Exactly two real destination connections, processed sequentially by one +// owned worker. Every socket I/O has a deadline and cleanup independently +// closes both the listener and any currently accepted connection before join. +func webH3InfoSuccessTCPOrigin(t *testing.T) (string, <-chan error) { + t.Helper() + listener, err := net.ListenTCP("tcp", &net.TCPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatal(err) + } + var mu sync.Mutex + var current *net.TCPConn + closed := false + results := make(chan error, 2) + done := make(chan struct{}) + go func() { + defer close(done) + for range 2 { + if err := listener.SetDeadline(time.Now().Add(2 * time.Second)); err != nil { + results <- err + return + } + conn, err := listener.AcceptTCP() + if err != nil { + results <- err + return + } + mu.Lock() + if closed { + mu.Unlock() + _ = conn.Close() + return + } + current = conn + mu.Unlock() + err = conn.SetDeadline(time.Now().Add(2 * time.Second)) + var payload []byte + if err == nil { + payload, err = io.ReadAll(conn) + } + if err == nil { + _, err = io.Copy(conn, bytes.NewReader(append([]byte("reply:"), payload...))) + } + if err == nil { + err = conn.CloseWrite() + } + _ = conn.Close() + mu.Lock() + current = nil + mu.Unlock() + results <- err + } + }() + t.Cleanup(func() { + _ = listener.Close() + mu.Lock() + closed = true + if current != nil { + _ = current.Close() + } + mu.Unlock() + webH3InfoSuccessWait(t, done, "real TCP destination worker") + }) + return listener.Addr().String(), results +} + +func webH3InfoSuccessHandler(t *testing.T, responses <-chan webH3InfoSuccessResponse) webH3InfoSuccessResponse { + t.Helper() + select { + case response := <-responses: + if response.proof == "" { + t.Error("successful original handler omitted its final proof") + } + return response + case <-time.After(2 * time.Second): + t.Fatal("successful original handler did not join") + return webH3InfoSuccessResponse{} + } +} + +func TestWebH3InformationalTCP200FullPayloadAndReuse(t *testing.T) { + target, targetResults := webH3InfoSuccessTCPOrigin(t) + dialer := transport.DialFunc(func(ctx context.Context, network, address string) (net.Conn, error) { + if network != "tcp" || address != target { + return nil, errors.New("unexpected informational success destination") + } + return (&net.Dialer{}).DialContext(ctx, network, address) + }) + _, handler, responses, client := webH3InfoSuccessServer(t, dialer, nil) + var first any + var bootstrapLen int + for request := range 2 { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + conn, err := client.h3.DialContext(ctx, "tcp", target) + cancel() // Successful stream ownership must already be detached. + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close() }) + if err := conn.SetDeadline(time.Now().Add(2 * time.Second)); err != nil { + t.Fatal(err) + } + payload := bytes.Repeat([]byte(fmt.Sprintf("informational-tcp-%d|", request)), 1024) + if _, err := io.Copy(conn, bytes.NewReader(payload)); err != nil { + t.Fatal(err) + } + if err := conn.(interface{ CloseWrite() error }).CloseWrite(); err != nil { + t.Fatal(err) + } + got, err := io.ReadAll(conn) + _ = conn.Close() + if err != nil || !bytes.Equal(got, append([]byte("reply:"), payload...)) { + t.Fatalf("real TCP response bytes=%d err=%v, want exact %d-byte reply", len(got), err, len(payload)+6) + } + response := webH3InfoSuccessHandler(t, responses) + select { + case err := <-targetResults: + if err != nil { + t.Fatal(err) + } + case <-time.After(2 * time.Second): + t.Fatal("real TCP destination did not complete") + } + session, ready := client.session() + if !ready || len(handler.core.sem) != 0 { + t.Error("successful TCP response lost authentication or retained admission") + } + if request == 0 { + first, bootstrapLen = session, response.bearerLen + } else if session != first || response.bearerLen >= bootstrapLen { + t.Error("second successful TCP CONNECT replaced physical session or used another full ticket") + } + } + if got := client.entropy.nonceReads.Load(); got != 1 { + t.Errorf("successful TCP full bootstrap tickets=%d, want one", got) + } +} + +func webH3InfoSuccessUDPOrigin(t *testing.T) netip.AddrPort { + t.Helper() + socket, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatal(err) + } + done := make(chan struct{}) + go func() { + defer close(done) + var buf [2048]byte + for range 2 { + if socket.SetDeadline(time.Now().Add(2*time.Second)) != nil { + return + } + n, source, err := socket.ReadFromUDPAddrPort(buf[:]) + if err != nil { + return + } + if _, err := socket.WriteToUDPAddrPort(buf[:n], source); err != nil { + return + } + } + }() + t.Cleanup(func() { + _ = socket.Close() + webH3InfoSuccessWait(t, done, "real UDP destination worker") + }) + return socket.LocalAddr().(*net.UDPAddr).AddrPort() +} + +func TestWebH3InformationalUDP200EchoAndReuse(t *testing.T) { + endpoint := webH3InfoSuccessUDPOrigin(t) + targets := []string{fmt.Sprintf("first.example:%d", endpoint.Port()), fmt.Sprintf("second.example:%d", endpoint.Port())} + resolver := UDPResolverFunc(func(_ context.Context, address string) ([]netip.AddrPort, error) { + if address != targets[0] && address != targets[1] { + return nil, errors.New("unexpected informational success UDP target") + } + return []netip.AddrPort{endpoint}, nil + }) + dialer := transport.DialFunc(func(context.Context, string, string) (net.Conn, error) { + return nil, errors.New("TCP destination is not expected in UDP success control") + }) + _, handler, responses, client := webH3InfoSuccessServer(t, dialer, resolver) + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + packet, err := client.h3.DialPacket(ctx) + cancel() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = packet.Close() }) + var first any + for request, target := range targets { + payload := []byte(fmt.Sprintf("real informational UDP echo %d", request)) + if err := packet.Send(payload, target); err != nil { + t.Fatal(err) + } + type reply struct { + payload []byte + target string + err error + } + result := make(chan reply, 1) + done := make(chan struct{}) + go func() { + defer close(done) + payload, target, err := packet.Receive() + result <- reply{payload, target, err} + }() + t.Cleanup(func() { + _ = packet.Close() + webH3InfoSuccessWait(t, done, "UDP Receive worker") + }) + select { + case got := <-result: + if got.err != nil || got.target != target || !bytes.Equal(got.payload, payload) { + t.Fatalf("real UDP reply=%q target=%q err=%v", got.payload, got.target, got.err) + } + case <-time.After(2 * time.Second): + t.Fatal("real UDP reply timed out") + } + webH3InfoSuccessWait(t, done, "completed UDP Receive worker") + session, ready := client.session() + if !ready { + t.Error("successful UDP target lost ready physical authentication") + } + if request == 0 { + first = session + } else if session != first { + t.Error("second successful UDP CONNECT used a replacement physical session") + } + } + if err := packet.Close(); err != nil { + t.Fatal(err) + } + bootstrap := webH3InfoSuccessHandler(t, responses) + continuation := webH3InfoSuccessHandler(t, responses) + if continuation.bearerLen >= bootstrap.bearerLen { + // Handler completion order need not match stream creation order. + bootstrap, continuation = continuation, bootstrap + } + if continuation.bearerLen >= bootstrap.bearerLen || client.entropy.nonceReads.Load() != 1 { + t.Error("successful UDP streams did not use one bootstrap and one shorter continuation") + } + if len(client.h3.udpSlots) != 0 || len(handler.core.sem) != 0 || len(handler.udp.slots) != 0 { + t.Error("closed successful UDP targets retained client or server admission") + } +} diff --git a/internal/tunnel/web_h3_informational_test.go b/internal/tunnel/web_h3_informational_test.go new file mode 100644 index 0000000..51574ce --- /dev/null +++ b/internal/tunnel/web_h3_informational_test.go @@ -0,0 +1,319 @@ +package tunnel + +import ( + "context" + "crypto/tls" + "errors" + "fmt" + "net" + "net/http" + "net/netip" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/cppla/autocar/internal/transport" +) + +type webH3InformationalResult struct { + proof string + bearerLen int + slots int + calls int64 + canceled bool +} + +// This test-only middleware writes legitimate informational blocks before +// invoking the unmodified production authentication / failure handler. It +// leaves the actual H2 / H3 response writer and its interfaces untouched. +func webH3InformationalServer(t *testing.T, mode string, tlsConfig *tls.Config, hints []int, stall bool) (string, <-chan webH3InformationalResult) { + t.Helper() + var calls atomic.Int64 + privateErr := errors.New("private informational target failure") + dialer := transport.DialFunc(func(context.Context, string, string) (net.Conn, error) { + calls.Add(1) + return nil, privateErr + }) + resolver := UDPResolverFunc(func(context.Context, string) ([]netip.AddrPort, error) { + calls.Add(1) + return nil, privateErr + }) + results := make(chan webH3InformationalResult, 4) + release := make(chan struct{}) + releaseStall := sync.OnceFunc(func() { close(release) }) + wrap := func(next http.Handler) http.Handler { + core := next.(*webTunnelHandler).core + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + for _, status := range hints { + w.Header().Set("Link", "; rel=preload") + w.WriteHeader(status) + clear(w.Header()) + } + if stall { + select { + case <-r.Context().Done(): + case <-release: + } + } else { + next.ServeHTTP(w, r) + } + results <- webH3InformationalResult{ + proof: w.Header().Get(webAuthResponseHeader), bearerLen: len(r.Header.Get("Proxy-Authorization")), + slots: len(core.sem), calls: calls.Load(), canceled: r.Context().Err() != nil, + } + }) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + var closeServer func() error + var address string + if mode == "h2" { + server, err := ListenWebH2(WebH2ServerConfig{ + Address: "127.0.0.1:0", Token: webTestToken, TLSConfig: tlsConfig, + Cover: http.NotFoundHandler(), Dialer: dialer, + }) + if err != nil { + cancel() + t.Fatal(err) + } + server.server.Handler = wrap(server.server.Handler) + closeServer, address = server.Close, server.Addr().String() + go func() { done <- server.Serve(ctx) }() + } else { + server, err := ListenWebH3(WebH3ServerConfig{ + Address: "127.0.0.1:0", Token: webTestToken, TLSConfig: tlsConfig, + Cover: http.NotFoundHandler(), Dialer: dialer, UDPResolver: resolver, + }) + if err != nil { + cancel() + t.Fatal(err) + } + server.server.Handler = wrap(server.server.Handler) + closeServer, address = server.Close, server.Addr().String() + go func() { done <- server.Serve(ctx) }() + } + t.Cleanup(func() { + releaseStall() + _ = closeServer() + cancel() + select { + case err := <-done: + if err != nil { + t.Errorf("informational Serve join: %v", err) + } + case <-time.After(2 * time.Second): + t.Error("informational Serve did not join") + } + }) + return address, results +} + +type webH3InformationalClient struct { + h2 *WebH2Client + h3 *WebH3Client + packet transport.PacketConn + entropy *webH2AuthEntropyCounter +} + +func newWebH3InformationalClient(t *testing.T, mode, address string, tlsConfig *tls.Config, handshakeTimeout time.Duration) *webH3InformationalClient { + t.Helper() + c := &webH3InformationalClient{entropy: &webH2AuthEntropyCounter{}} + signer := newWebAuthSigner(mustWebAuthKey(t, webTestToken), nil, c.entropy) + var err error + if mode == "h2" { + c.h2, err = newWebH2ClientWithSigner(WebH2ClientConfig{ + ServerAddress: address, Token: webTestToken, TLSConfig: tlsConfig, + HandshakeTimeout: handshakeTimeout, + }, signer, webAuthClaims{}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = c.h2.Close() }) + } else { + c.h3, err = NewWebH3Client(WebH3ClientConfig{ + ServerAddress: address, Token: webTestToken, TLSConfig: tlsConfig, + HandshakeTimeout: handshakeTimeout, + }) + if err != nil { + t.Fatal(err) + } + c.h3.signer = signer + t.Cleanup(func() { _ = c.h3.Close() }) + if mode == "h3-udp" { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + c.packet, err = c.h3.DialPacket(ctx) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = c.packet.Close() }) + } + } + return c +} + +func (c *webH3InformationalClient) request(target string) error { + if c.packet != nil { + return c.packet.Send([]byte("not delivered"), target) + } + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + var conn net.Conn + var err error + if c.h2 != nil { + conn, err = c.h2.DialContext(ctx, "tcp", target) + } else { + conn, err = c.h3.DialContext(ctx, "tcp", target) + } + if conn != nil { + _ = conn.Close() + return errors.New("rejected destination returned a connection") + } + return err +} + +func (c *webH3InformationalClient) session() (any, bool) { + if c.h2 != nil { + c.h2.mu.Lock() + defer c.h2.mu.Unlock() + s := c.h2.current + return s, s != nil && s.authState == webH2ClientAuthReady && s.auth != nil + } + c.h3.mu.Lock() + defer c.h3.mu.Unlock() + s := c.h3.conns[c.h3.conn] + return s, s != nil && s.authState == webH3ClientAuthReady && s.auth != nil +} + +func webH3InformationalAwait(t *testing.T, results <-chan webH3InformationalResult) webH3InformationalResult { + t.Helper() + select { + case result := <-results: + if result.slots != 0 { + t.Errorf("handler retained %d admission slots", result.slots) + } + return result + case <-time.After(2 * time.Second): + t.Fatal("informational handler did not join") + return webH3InformationalResult{} + } +} + +func TestWebH3InformationalSignedFailurePreservesConnection(t *testing.T) { + serverTLS, clientTLS := testTLSConfigs(t) + for _, mode := range []string{"h2", "h3-tcp", "h3-udp"} { + for _, hints := range [][]int{nil, {100, 102, 103}, {100, 102, 103, 102, 103}} { + t.Run(fmt.Sprintf("%s/hints_%d", mode, len(hints)), func(t *testing.T) { + address, results := webH3InformationalServer(t, mode, serverTLS, hints, false) + client := newWebH3InformationalClient(t, mode, address, clientTLS, time.Second) + var first any + var bootstrapLen int + for request := range 2 { + err := client.request(fmt.Sprintf("missing%d.example:443", request)) + result := webH3InformationalAwait(t, results) + var rejection *WebConnectError + if !errors.As(err, &rejection) || rejection.StatusCode != http.StatusBadGateway || rejection.Transport != map[string]string{"h2": "h2", "h3-tcp": "h3", "h3-udp": "h3"}[mode] { + t.Errorf("target failure = %T %v, want verified %s 502", err, err, mode) + } + if mode == "h3-udp" && (!errors.Is(err, transport.ErrPacketTargetUnavailable) || len(client.h3.udpSlots) != 0) { + t.Error("signed UDP target failure lost recoverability or retained admission") + } + if result.proof == "" || result.calls != int64(request+1) { + t.Errorf("original handler proof-present=%t targetcalls=%d, want %d", result.proof != "", result.calls, request+1) + } + session, ready := client.session() + if !ready { + t.Error("signed 502 did not retain ready connection authentication") + } + if request == 0 { + first, bootstrapLen = session, result.bearerLen + } else if session != first || result.bearerLen >= bootstrapLen { + t.Errorf("continuation replaced physical session or did not use a shorter ticket: same=%t bearer=%d bootstrap=%d", session == first, result.bearerLen, bootstrapLen) + } + } + if got := client.entropy.nonceReads.Load(); got != 1 { + t.Errorf("full bootstrap ticket count = %d, want one", got) + } + }) + } + } +} + +func TestWebH3InformationalLimitRejectsFinalProof(t *testing.T) { + serverTLS, clientTLS := testTLSConfigs(t) + for _, mode := range []string{"h3-tcp", "h3-udp"} { + t.Run(mode, func(t *testing.T) { + address, results := webH3InformationalServer(t, mode, serverTLS, []int{100, 102, 103, 102, 103, 103}, false) + client := newWebH3InformationalClient(t, mode, address, clientTLS, time.Second) + err := client.request("missing.example:443") + result := webH3InformationalAwait(t, results) + var rejection *WebConnectError + if err == nil || errors.As(err, &rejection) || errors.Is(err, transport.ErrPacketTargetUnavailable) { + t.Errorf("six informational blocks accepted final signed rejection: %v", err) + } + if err == nil || !strings.Contains(err.Error(), "too many web-cover H3 informational responses") { + t.Errorf("six informational blocks error = %v, want the informational limit failure", err) + } + if _, ready := client.session(); ready { + t.Error("excessive informational response retained authenticated bootstrap") + } + if mode == "h3-udp" && len(client.h3.udpSlots) != 0 { + t.Error("excessive informational response retained UDP admission") + } + // Depending on cancellation scheduling, the peer may finish writing + // the valid final proof or notice stream cancellation first. Neither + // outcome may be accepted after the informational limit is exceeded. + t.Logf("six-hint rejection=%v; server final proof present=%t, calls=%d", err, result.proof != "", result.calls) + }) + } +} + +func TestWebH3InformationalStalledFinalHonorsDeadline(t *testing.T) { + serverTLS, clientTLS := testTLSConfigs(t) + for _, mode := range []string{"h3-tcp", "h3-udp"} { + t.Run(mode, func(t *testing.T) { + address, results := webH3InformationalServer(t, mode, serverTLS, []int{103}, true) + client := newWebH3InformationalClient(t, mode, address, clientTLS, time.Second) + // Establish QUIC separately so an overloaded TLS handshake cannot + // masquerade as the target response deadline under test. No request + // stream or authentication exists yet. The target budget stays 80ms. + warmContext, cancelWarm := context.WithTimeout(context.Background(), 2*time.Second) + _, _, warmErr := client.h3.connection(warmContext) + cancelWarm() + if warmErr != nil { + t.Fatal(warmErr) + } + client.h3.mu.Lock() + fresh := client.h3.conns[client.h3.conn] + isFresh := fresh != nil && fresh.authState == webH3ClientAuthFresh && fresh.users == 0 + client.h3.mu.Unlock() + if !isFresh { + t.Fatal("warm physical connection already has authentication or stream owners") + } + select { + case <-results: + t.Fatal("physical warmup invoked the request handler") + default: + } + // This client is test-exclusive with no concurrent opening request. + client.h3.handshakeTimeout = 80 * time.Millisecond + err := client.request("missing.example:443") + result := webH3InformationalAwait(t, results) + var timeout net.Error + if !errors.As(err, &timeout) || !timeout.Timeout() { + t.Errorf("stalled final response error = %T %v, want setup deadline", err, err) + } + if !result.canceled || result.calls != 0 || result.proof != "" { + t.Errorf("stalled handler canceled=%t calls=%d proof-present=%t", result.canceled, result.calls, result.proof != "") + } + if _, ready := client.session(); ready { + t.Error("stalled final response authenticated a bootstrap") + } + if mode == "h3-udp" && len(client.h3.udpSlots) != 0 { + t.Error("setup deadline retained UDP admission") + } + }) + } +} diff --git a/internal/tunnel/web_h3_response.go b/internal/tunnel/web_h3_response.go new file mode 100644 index 0000000..937cc67 --- /dev/null +++ b/internal/tunnel/web_h3_response.go @@ -0,0 +1,43 @@ +package tunnel + +import ( + "errors" + "net/http" + + "github.com/apernet/quic-go" + "github.com/apernet/quic-go/http3" +) + +// Match the pinned HTTP/3 library's complete ClientConn response loop. The +// low-level request stream used by CONNECT returns one HEADERS block at a time. +const webH3MaxInformationalResponses = 5 + +type webH3ResponseStream interface { + ReadResponse() (*http.Response, error) + CancelRead(quic.StreamErrorCode) + CancelWrite(quic.StreamErrorCode) +} + +func readWebH3FinalResponse(stream webH3ResponseStream) (*http.Response, error) { + for informational := 0; ; { + response, err := stream.ReadResponse() + if err != nil { + return nil, err + } + if response == nil { + return nil, errors.New("tunnel: empty web-cover H3 response") + } + if response.StatusCode < 100 || response.StatusCode >= 200 || response.StatusCode == http.StatusSwitchingProtocols { + return response, nil + } + informational++ + if informational > webH3MaxInformationalResponses { + stream.CancelRead(quic.StreamErrorCode(http3.ErrCodeExcessiveLoad)) + stream.CancelWrite(quic.StreamErrorCode(http3.ErrCodeExcessiveLoad)) + return nil, errors.New("tunnel: too many web-cover H3 informational responses") + } + // Do not close or drain this body: it is backed by the same request + // stream as the final response. Keep the original establishment + // deadline and cancellation active while reading the next HEADERS. + } +} diff --git a/internal/tunnel/web_h3_response_test.go b/internal/tunnel/web_h3_response_test.go new file mode 100644 index 0000000..5542008 --- /dev/null +++ b/internal/tunnel/web_h3_response_test.go @@ -0,0 +1,262 @@ +package tunnel + +import ( + "context" + "errors" + "io" + "net/http" + "strconv" + "strings" + "testing" + "time" + + "github.com/apernet/quic-go" + "github.com/apernet/quic-go/http3" +) + +func TestWebH3FinalResponsePreservesFinalMessage(t *testing.T) { + for _, status := range []int{http.StatusOK, http.StatusBadGateway} { + t.Run(http.StatusText(status), func(t *testing.T) { + body := &webH3ResponseTestBody{reader: strings.NewReader("final body")} + final := &http.Response{ + StatusCode: status, + Header: http.Header{ + webAuthResponseHeader: {"final-authenticated-proof"}, + "X-Final": {"retained"}, + }, + Trailer: http.Header{"X-End-Trailer": {"retained-trailer"}}, + Body: body, + ContentLength: 10, + } + stream := newWebH3ResponseTestStream(t) + for _, status := range []int{http.StatusContinue, http.StatusProcessing, http.StatusEarlyHints} { + stream.steps = append(stream.steps, webH3ResponseTestStep{response: stream.informational(status)}) + } + stream.steps = append(stream.steps, + webH3ResponseTestStep{response: final}, + webH3ResponseTestStep{err: errors.New("must not read past final")}, + ) + response, err := readWebH3FinalResponse(stream) + if err != nil || response != final { + t.Fatalf("final response = %p/%v, want original %p", response, err, final) + } + if len(response.Header) != 2 || response.Header.Get(webAuthResponseHeader) != "final-authenticated-proof" || response.Header.Get("X-Final") != "retained" { + t.Fatalf("final headers were replaced or mixed with informational headers: %v", response.Header) + } + if response.Body != body || response.ContentLength != 10 || response.Trailer.Get("X-End-Trailer") != "retained-trailer" { + t.Fatalf("final body, length, or trailer changed: %+v", response) + } + if body.reads != 0 || body.closes != 0 { + t.Fatal("helper read or closed the final response body") + } + stream.assertUnchanged(t, 4, false) + payload, err := io.ReadAll(response.Body) + if err != nil || string(payload) != "final body" { + t.Fatalf("returned final body = %q/%v", payload, err) + } + }) + } +} + +func TestWebH3FinalResponseDiscardsInformationalProof(t *testing.T) { + stream := newWebH3ResponseTestStream(t) + final := &http.Response{StatusCode: http.StatusOK, Header: http.Header{"X-Final": {"retained"}}, Body: http.NoBody} + stream.steps = []webH3ResponseTestStep{ + {response: stream.informational(http.StatusEarlyHints)}, + {response: final}, + } + response, err := readWebH3FinalResponse(stream) + if err != nil || response != final || len(response.Header.Values(webAuthResponseHeader)) != 0 || response.Header.Get("X-Informational") != "" { + t.Fatalf("informational fields reached final response: response=%+v err=%v", response, err) + } + stream.assertUnchanged(t, 2, false) +} + +func TestWebH3FinalResponseInformationalLimit(t *testing.T) { + for _, count := range []int{0, 1, webH3MaxInformationalResponses, webH3MaxInformationalResponses + 1} { + t.Run(strconv.Itoa(count)+"-hints", func(t *testing.T) { + stream := newWebH3ResponseTestStream(t) + for range count { + stream.steps = append(stream.steps, webH3ResponseTestStep{response: stream.informational(http.StatusEarlyHints)}) + } + final := &http.Response{StatusCode: http.StatusOK, Header: make(http.Header), Body: http.NoBody} + stream.steps = append(stream.steps, webH3ResponseTestStep{response: final}) + response, err := readWebH3FinalResponse(stream) + if count > webH3MaxInformationalResponses { + if response != nil || err == nil { + t.Fatalf("excessive hints accepted: response=%+v err=%v", response, err) + } + stream.assertUnchanged(t, count, true) + return + } + if err != nil || response != final { + t.Fatalf("allowed hints rejected: response=%p err=%v, want %p", response, err, final) + } + stream.assertUnchanged(t, count+1, false) + }) + } +} + +func TestWebH3FinalResponseSwitchingProtocolsIsTerminal(t *testing.T) { + for _, precedingHint := range []bool{false, true} { + t.Run(map[bool]string{false: "direct", true: "after-hint"}[precedingHint], func(t *testing.T) { + stream := newWebH3ResponseTestStream(t) + if precedingHint { + stream.steps = append(stream.steps, webH3ResponseTestStep{response: stream.informational(http.StatusEarlyHints)}) + } + terminalBody := &webH3ResponseTestBody{reader: strings.NewReader("terminal body")} + terminal := &http.Response{StatusCode: http.StatusSwitchingProtocols, Header: http.Header{"X-Terminal": {"retained"}}, Body: terminalBody} + stream.steps = append(stream.steps, + webH3ResponseTestStep{response: terminal}, + webH3ResponseTestStep{err: errors.New("must not read past 101")}, + ) + response, err := readWebH3FinalResponse(stream) + if err != nil || response != terminal || response.Body != terminalBody || response.Header.Get("X-Terminal") != "retained" { + t.Fatalf("101 terminal response changed: response=%+v err=%v", response, err) + } + reads := 1 + if precedingHint { + reads++ + } + stream.assertUnchanged(t, reads, false) + if terminalBody.reads != 0 || terminalBody.closes != 0 { + t.Fatal("helper read or closed the terminal response body") + } + }) + } +} + +func TestWebH3FinalResponseReadFailures(t *testing.T) { + readError := errors.New("response read failed") + for _, test := range []struct { + name string + precedingHint bool + response *http.Response + err error + }{ + {name: "early-error", err: readError}, + {name: "response-and-error", response: &http.Response{StatusCode: http.StatusOK}, err: readError}, + {name: "error-after-hint", precedingHint: true, err: readError}, + {name: "canceled-after-hint", precedingHint: true, err: context.Canceled}, + {name: "deadline-after-hint", precedingHint: true, err: context.DeadlineExceeded}, + {name: "nil-response"}, + {name: "nil-response-after-hint", precedingHint: true}, + } { + t.Run(test.name, func(t *testing.T) { + stream := newWebH3ResponseTestStream(t) + if test.precedingHint { + stream.steps = append(stream.steps, webH3ResponseTestStep{response: stream.informational(http.StatusEarlyHints)}) + } + stream.steps = append(stream.steps, + webH3ResponseTestStep{response: test.response, err: test.err}, + webH3ResponseTestStep{err: errors.New("must not read past failure")}, + ) + response, err := readWebH3FinalResponse(stream) + if response != nil || err == nil { + t.Fatalf("failed read accepted: response=%+v err=%v", response, err) + } + if test.err != nil && err != test.err { + t.Fatalf("read error identity changed: %v, want original %v", err, test.err) + } + reads := 1 + if test.precedingHint { + reads++ + } + stream.assertUnchanged(t, reads, false) + }) + } +} + +type webH3ResponseTestStep struct { + response *http.Response + err error +} + +type webH3ResponseTestStream struct { + t *testing.T + steps []webH3ResponseTestStep + reads int + readCancel []quic.StreamErrorCode + writeCancel []quic.StreamErrorCode + infoBodies []*webH3ResponseTestBody + deadline time.Time + initialDeadline time.Time + deadlineCalls int +} + +func newWebH3ResponseTestStream(t *testing.T) *webH3ResponseTestStream { + deadline := time.Now().Add(time.Second) + return &webH3ResponseTestStream{t: t, deadline: deadline, initialDeadline: deadline} +} + +func (s *webH3ResponseTestStream) ReadResponse() (*http.Response, error) { + s.t.Helper() + if s.reads >= len(s.steps) { + s.t.Fatal("read past configured response sequence") + } + step := s.steps[s.reads] + s.reads++ + return step.response, step.err +} + +func (s *webH3ResponseTestStream) CancelRead(code quic.StreamErrorCode) { + s.readCancel = append(s.readCancel, code) +} + +func (s *webH3ResponseTestStream) CancelWrite(code quic.StreamErrorCode) { + s.writeCancel = append(s.writeCancel, code) +} + +func (s *webH3ResponseTestStream) SetDeadline(deadline time.Time) error { + s.deadlineCalls++ + s.deadline = deadline + return nil +} + +func (s *webH3ResponseTestStream) informational(status int) *http.Response { + body := &webH3ResponseTestBody{reader: strings.NewReader("must not drain informational body")} + s.infoBodies = append(s.infoBodies, body) + return &http.Response{StatusCode: status, Header: http.Header{ + webAuthResponseHeader: {"fictional-informational-proof"}, + "X-Informational": {"not-final"}, + }, Body: body} +} + +func (s *webH3ResponseTestStream) assertUnchanged(t *testing.T, reads int, excessive bool) { + t.Helper() + if s.reads != reads { + t.Errorf("ReadResponse calls = %d, want %d", s.reads, reads) + } + if s.deadlineCalls != 0 || s.deadline != s.initialDeadline { + t.Error("helper reset the existing request deadline") + } + for index, body := range s.infoBodies { + if body.reads != 0 || body.closes != 0 { + t.Errorf("informational body %d read=%d close=%d, want neither", index, body.reads, body.closes) + } + } + if excessive { + wantCode := quic.StreamErrorCode(http3.ErrCodeExcessiveLoad) + if len(s.readCancel) != 1 || s.readCancel[0] != wantCode || len(s.writeCancel) != 1 || s.writeCancel[0] != wantCode { + t.Errorf("excess cancellation read=%v write=%v, want one %d each", s.readCancel, s.writeCancel, wantCode) + } + } else if len(s.readCancel) != 0 || len(s.writeCancel) != 0 { + t.Errorf("unexpected cancellation read=%v write=%v", s.readCancel, s.writeCancel) + } +} + +type webH3ResponseTestBody struct { + reader io.Reader + reads int + closes int +} + +func (b *webH3ResponseTestBody) Read(p []byte) (int, error) { + b.reads++ + return b.reader.Read(p) +} + +func (b *webH3ResponseTestBody) Close() error { + b.closes++ + return nil +} diff --git a/internal/tunnel/web_udp.go b/internal/tunnel/web_udp.go index 9c2c176..555b919 100644 --- a/internal/tunnel/web_udp.go +++ b/internal/tunnel/web_udp.go @@ -938,7 +938,7 @@ func (c *WebH3Client) openConnectUDPSession(ctx context.Context, target string) } return nil, fmt.Errorf("tunnel: send CONNECT-UDP request for %s: %w", canonicalTarget, err) } - response, err := stream.ReadResponse() + response, err := readWebH3FinalResponse(stream) if err != nil { fail() if reservation.bootstrap {