From 5b7d99a83dc633f087b9e58abd79f51115f29548 Mon Sep 17 00:00:00 2001 From: cppla Date: Sat, 3 Oct 2026 16:58:46 +0800 Subject: [PATCH] fix(proxy): preserve streamed HTTP request trailers --- docs/DEPLOYMENT.md | 9 + internal/proxy/http.go | 1 + internal/proxy/http_request_trailer.go | 115 +++++++ .../http_request_trailer_integration_test.go | 302 ++++++++++++++++++ internal/proxy/http_request_trailer_test.go | 256 +++++++++++++++ 5 files changed, 683 insertions(+) create mode 100644 internal/proxy/http_request_trailer.go create mode 100644 internal/proxy/http_request_trailer_integration_test.go create mode 100644 internal/proxy/http_request_trailer_test.go diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index 55fe696..30315ef 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -347,6 +347,15 @@ are sent can instead appear as a connection error. Declared trailers are forwarded only after the body completes successfully. This does not add HTTP Upgrade support or change CONNECT tunnels, and is not included in v1.0.1. +Source builds preserve declared safe trailers on streamed HTTP forward-proxy +uploads, including integrity metadata finalized only at body EOF. Uploads are +not buffered; trailer values are copied before the outbound transport writes +them. Forbidden trailer fields, proxy credentials and initially +connection-nominated hop fields are removed. A nomination received only at EOF +removes its value but cannot retract a declaration already streamed upstream. +Ordinary origin `Authorization` headers retain their existing behavior. This +is not included in v1.0.1. + The relay's separate `--destination-write-timeout` defaults to `5m`; `0` also selects `5m`, rather than disabling the limit. It bounds the completion of each TCP destination write, split into chunks of at most 32 KiB. Each diff --git a/internal/proxy/http.go b/internal/proxy/http.go index 0e65449..9b19595 100644 --- a/internal/proxy/http.go +++ b/internal/proxy/http.go @@ -264,6 +264,7 @@ func (s *HTTPServer) serveForward(w http.ResponseWriter, r *http.Request) { } out := r.Clone(r.Context()) + prepareHTTPForwardRequestTrailers(r, out) out.RequestURI = "" // For absolute-form requests the URI authority is authoritative. A proxy // must not let a conflicting inbound Host field select a different virtual diff --git a/internal/proxy/http_request_trailer.go b/internal/proxy/http_request_trailer.go new file mode 100644 index 0000000..56400b5 --- /dev/null +++ b/internal/proxy/http_request_trailer.go @@ -0,0 +1,115 @@ +package proxy + +import ( + "io" + "net/http" + "strings" + "sync" + + "golang.org/x/net/http/httpguts" +) + +// Clone owns Trailer but shares Body. The server body fills the original map +// only at EOF; publish its safe declared values before the transport sees EOF. +func prepareHTTPForwardRequestTrailers(in, out *http.Request) { + blocked := httpForwardTrailerNominations(in.Header, in.Trailer) + destination := make(http.Header, len(out.Trailer)) + declared := make(map[string]string, len(out.Trailer)) + for field, values := range out.Trailer { + if !httpForwardTrailerAllowed(field, blocked) { + continue + } + key := http.CanonicalHeaderKey(field) + declared[strings.ToLower(field)] = key + destination[key] = append(destination[key], values...) + } + out.Trailer = destination + if len(declared) == 0 || out.Body == nil || out.Body == http.NoBody { + return + } + out.Body = &httpForwardTrailerBody{ + body: out.Body, source: in, destination: destination, + declared: declared, blocked: blocked, + } + // Known-length embedder requests can have trailers too. Keep the incoming + // body's length validation but use H1 framing which actually emits trailers. + out.ContentLength = -1 + out.TransferEncoding = []string{"chunked"} + for field := range out.Header { + if strings.EqualFold(field, "Content-Length") { + delete(out.Header, field) + } + } + // A replay getter would bypass the EOF bridge. Do not modify the caller's. + out.GetBody = nil +} + +func httpForwardTrailerNominations(headers ...http.Header) map[string]bool { + blocked := make(map[string]bool) + for _, header := range headers { + for field, values := range header { + if !strings.EqualFold(field, "Connection") { + continue + } + for _, value := range values { + for _, token := range strings.Split(value, ",") { + token = strings.TrimSpace(token) + if httpguts.ValidHeaderFieldName(token) { + blocked[strings.ToLower(token)] = true + } + } + } + } + } + return blocked +} + +func httpForwardTrailerAllowed(field string, blocked map[string]bool) bool { + return httpguts.ValidHeaderFieldName(field) && httpguts.ValidTrailerHeader(field) && + !isHopByHopHeader(field) && !strings.EqualFold(field, "Proxy-Authentication-Info") && + !blocked[strings.ToLower(field)] +} + +// One body reader may run concurrently with Close, as net/http requires. +// Close delegates without taking an I/O lock so it can unblock that reader. +// Only successful EOF publishes metadata; neither errors nor Close do so. +type httpForwardTrailerBody struct { + body io.ReadCloser + source *http.Request + destination http.Header + declared map[string]string + blocked map[string]bool + once sync.Once +} + +func (b *httpForwardTrailerBody) Read(p []byte) (int, error) { + n, err := b.body.Read(p) + if err == io.EOF { + b.once.Do(func() { + current := b.source.Trailer // Custom bodies may replace the map at EOF. + for field := range httpForwardTrailerNominations(current) { + b.blocked[field] = true + } + // Keep the map identity already captured by the transport. + clear(b.destination) + for _, key := range b.declared { + if httpForwardTrailerAllowed(key, b.blocked) { + b.destination[key] = nil + } + } + // One ASCII alias contributes values once to its logical declaration. + // Copy slices so the outbound metadata never aliases caller storage. + for field, values := range current { + if !httpForwardTrailerAllowed(field, b.blocked) { + continue + } + if key, ok := b.declared[strings.ToLower(field)]; ok { + b.destination[key] = append(b.destination[key], values...) + } + } + }) + } + return n, err +} + +func (b *httpForwardTrailerBody) Close() error { return b.body.Close() } diff --git a/internal/proxy/http_request_trailer_integration_test.go b/internal/proxy/http_request_trailer_integration_test.go new file mode 100644 index 0000000..7f94b8f --- /dev/null +++ b/internal/proxy/http_request_trailer_integration_test.go @@ -0,0 +1,302 @@ +package proxy + +import ( + "bufio" + "bytes" + "context" + "crypto/sha256" + "encoding/base64" + "errors" + "fmt" + "io" + "net" + "net/http" + "strings" + "sync" + "testing" + "time" + + "github.com/cppla/autocar/internal/transport" +) + +// This regression uses raw owned TCP on both sides. The origin must +// have consumed the first complete body chunk before the visitor sends EOF and +// the two trailing fields. It does not simulate a prepopulated request map. +func TestHTTPForwardRequestTrailersStream(t *testing.T) { + for _, mode := range []string{"direct", "proxy", "proxy_filters_late_fields"} { + t.Run(mode, func(t *testing.T) { + httpRequestTrailerRun(t, mode) + }) + } +} + +func TestHTTPForwardRequestTrailersVisitorCancellation(t *testing.T) { + httpRequestTrailerRun(t, "proxy_canceled_upload") +} + +func httpRequestTrailerRun(t *testing.T, mode string) { + t.Helper() + payload := bytes.Repeat([]byte("streamed-trailer-body\n"), 3277) + payload = payload[:64<<10] + sum := sha256.Sum256(payload) + digest := "sha-256=:" + base64.StdEncoding.EncodeToString(sum[:]) + ":" + origin, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + firstBody := make(chan struct{}) + originDone := make(chan struct{}) + type observation struct { + body []byte + headers http.Header + trailers http.Header + status int + err error + } + observed := make(chan observation, 1) + var originMu sync.Mutex + var originConn net.Conn + originClosed := false + var visitor net.Conn + var proxyServer *HTTPServer + var proxyListener net.Listener + var proxyDone chan struct{} + var proxyResult chan error + var handlerEntered, handlerDone chan struct{} + // Register independent cleanup before launching any owned worker. + t.Cleanup(func() { + if visitor != nil { + _ = visitor.Close() + } + originMu.Lock() + originClosed = true + owned := originConn + originMu.Unlock() + _ = origin.Close() + if owned != nil { + _ = owned.Close() + } + if proxyServer != nil { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + err := proxyServer.Shutdown(ctx) + cancel() + if err != nil && !errors.Is(err, net.ErrClosed) { + t.Errorf("proxy cleanup shutdown: %v", err) + } + } + if proxyListener != nil { + _ = proxyListener.Close() + } + auditTrailerJoin(t, originDone, "origin worker") + if proxyDone != nil { + auditTrailerJoin(t, proxyDone, "proxy Serve") + select { + case <-handlerEntered: + auditTrailerJoin(t, handlerDone, "proxy handler") + default: + } + select { + case err := <-proxyResult: + if err != nil { + t.Errorf("proxy Serve: %v", err) + } + default: + t.Error("joined proxy Serve has no result") + } + } + }) + go func() { + defer close(originDone) + result := observation{} + defer func() { observed <- result }() + conn, err := origin.Accept() + if err != nil { + result.err = err + return + } + originMu.Lock() + if originClosed { + originMu.Unlock() + _ = conn.Close() + result.err = net.ErrClosed + return + } + originConn = conn + originMu.Unlock() + defer conn.Close() + _ = conn.SetDeadline(time.Now().Add(3 * time.Second)) + request, err := http.ReadRequest(bufio.NewReader(conn)) + if err != nil { + result.err = err + return + } + defer request.Body.Close() + result.headers = request.Header.Clone() + first := make([]byte, len(payload)) + if _, err := io.ReadFull(request.Body, first); err != nil { + result.err = err + return + } + close(firstBody) + rest, err := io.ReadAll(request.Body) + result.body = append(first, rest...) + result.trailers = request.Trailer.Clone() + if err != nil { + result.err = err + return + } + result.status = http.StatusUnprocessableEntity + if bytes.Equal(result.body, payload) && result.trailers.Get("Content-Digest") == digest && result.trailers.Get("X-Upload-End") == "complete" { + result.status = http.StatusOK + } + _, result.err = fmt.Fprintf(conn, "HTTP/1.1 %d %s\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok", result.status, http.StatusText(result.status)) + }() + address := origin.Addr().String() + requestTarget := "/upload" + if mode != "direct" { + proxyServer, err = NewHTTPServer(Config{Dialer: transport.DialFunc(func(ctx context.Context, network, target string) (net.Conn, error) { + if network != "tcp" || target != origin.Addr().String() { + return nil, fmt.Errorf("diagnostic rejected non-owned target %q %q", network, target) + } + return (&net.Dialer{}).DialContext(ctx, network, target) + })}) + if err != nil { + t.Fatal(err) + } + handlerEntered, handlerDone = make(chan struct{}), make(chan struct{}) + originalHandler := proxyServer.server.Handler + proxyServer.server.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + close(handlerEntered) + defer close(handlerDone) + originalHandler.ServeHTTP(w, r) + }) + proxyListener, err = net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + proxyDone, proxyResult = make(chan struct{}), make(chan error, 1) + go func() { + defer close(proxyDone) + proxyResult <- proxyServer.Serve(proxyListener) + }() + address = proxyListener.Addr().String() + requestTarget = "http://" + origin.Addr().String() + "/upload" + } + visitor, err = net.DialTimeout("tcp", address, time.Second) + if err != nil { + t.Fatal(err) + } + _ = visitor.SetDeadline(time.Now().Add(3 * time.Second)) + declared := "Content-Digest, X-Upload-End" + connection := "close" + extraHeaders := "" + if mode == "proxy_filters_late_fields" { + // Go's H1 parser accepts these adversarial fields. Authentication + // and hop fields are not claimed to be valid end-to-end trailers. + declared += ", Connection, X-Initial-Nominated, X-Late-Nominated, Proxy-Authorization, Proxy-Authenticate, Proxy-Authentication-Info, Proxy-Connection, Te" + connection += ", X-Initial-Nominated" + extraHeaders = "Authorization: Bearer website-dummy\r\nProxy-Authorization: Basic proxy-dummy\r\nX-Initial-Nominated: header-dummy\r\n" + } + firstWire := fmt.Sprintf("POST %s HTTP/1.1\r\nHost: %s\r\nTransfer-Encoding: chunked\r\nTrailer: %s\r\nConnection: %s\r\n%s\r\n%x\r\n", requestTarget, origin.Addr(), declared, connection, extraHeaders, len(payload)) + if err := auditTrailerWriteAll(visitor, append(append([]byte(firstWire), payload...), []byte("\r\n")...)); err != nil { + t.Fatal(err) + } + select { + case <-firstBody: + t.Logf("origin consumed first %d bytes before visitor EOF/trailers", len(payload)) + case <-originDone: + t.Fatal("origin ended before first-body witness") + case <-time.After(2 * time.Second): + t.Fatal("origin first-body witness timeout") + } + if mode == "proxy_canceled_upload" { + // This explicit visitor closure, not fixture cleanup, is the + // cancellation event while the source Body still awaits EOF. + if err := visitor.Close(); err != nil { + t.Fatal(err) + } + if !auditTrailerJoin(t, originDone, "canceled origin worker") || !auditTrailerJoin(t, handlerDone, "canceled proxy handler") { + t.FailNow() + } + result := <-observed + if !errors.Is(result.err, io.EOF) && !errors.Is(result.err, io.ErrUnexpectedEOF) { + t.Errorf("visitor closure did not interrupt origin body with EOF: %v", result.err) + } + if !bytes.Equal(result.body, payload) { + t.Errorf("canceled origin first body bytes=%d want=%d", len(result.body), len(payload)) + } + t.Logf("explicit visitor Close joined origin and proxy handler before cleanup: bytes=%d error=%v", len(result.body), result.err) + return + } + lastWire := "0\r\nContent-Digest: " + digest + "\r\nX-Upload-End: complete\r\n" + if mode == "proxy_filters_late_fields" { + lastWire += "Connection: X-Late-Nominated\r\nX-Initial-Nominated: initial-dummy\r\nX-Late-Nominated: late-dummy\r\nProxy-Authorization: Basic trailer-dummy\r\nProxy-Authenticate: Basic dummy\r\nProxy-Authentication-Info: dummy\r\nProxy-Connection: keep-alive\r\nTe: trailers\r\n" + } + if err := auditTrailerWriteAll(visitor, []byte(lastWire+"\r\n")); err != nil { + t.Fatal(err) + } + response, err := http.ReadResponse(bufio.NewReader(visitor), nil) + if err != nil { + t.Fatal(err) + } + body, readErr := io.ReadAll(response.Body) + closeErr := response.Body.Close() + if readErr != nil || closeErr != nil || string(body) != "ok" { + t.Fatalf("response body=%q read=%v close=%v", body, readErr, closeErr) + } + if !auditTrailerJoin(t, originDone, "completed origin worker") { + t.FailNow() + } + result := <-observed // buffered and published before originDone closes + gotSum := sha256.Sum256(result.body) + t.Logf("status=%d bodyBytes=%d sha256=%x trailers=%v originError=%v", response.StatusCode, len(result.body), gotSum, result.trailers, result.err) + if result.err != nil || !bytes.Equal(result.body, payload) { + t.Fatalf("origin body integrity error=%v bytes=%d", result.err, len(result.body)) + } + if response.StatusCode != http.StatusOK || result.trailers.Get("Content-Digest") != digest || result.trailers.Get("X-Upload-End") != "complete" { + t.Errorf("streamed request trailers lost: status=%d Content-Digest=%q X-Upload-End=%q", response.StatusCode, result.trailers.Get("Content-Digest"), result.trailers.Get("X-Upload-End")) + } + if mode == "proxy_filters_late_fields" { + for _, key := range []string{"Connection", "X-Initial-Nominated", "X-Late-Nominated", "Proxy-Authorization", "Proxy-Authenticate", "Proxy-Authentication-Info", "Proxy-Connection", "Te"} { + for field := range result.trailers { + if strings.EqualFold(field, key) { + // The initial declaration is already on the wire before the + // late Connection nomination appears at EOF. Streaming can + // filter all values, but cannot retract that declaration. + if key == "X-Late-Nominated" && len(result.trailers[field]) == 0 { + continue + } + t.Errorf("unsafe trailer declaration/value reached origin: %s=%v", field, result.trailers[field]) + } + } + } + if result.headers.Get("Authorization") != "Bearer website-dummy" || result.headers.Get("Proxy-Authorization") != "" || result.headers.Get("X-Initial-Nominated") != "" { + t.Errorf("request header policy changed/leaked: %v", result.headers) + } + } +} + +func auditTrailerWriteAll(conn net.Conn, data []byte) error { + for len(data) != 0 { + n, err := conn.Write(data) + data = data[n:] + if err != nil { + return err + } + if n == 0 { + return io.ErrNoProgress + } + } + return nil +} + +func auditTrailerJoin(t *testing.T, done <-chan struct{}, what string) bool { + t.Helper() + select { + case <-done: + return true + case <-time.After(2 * time.Second): + t.Errorf("%s failed to join", what) + return false + } +} diff --git a/internal/proxy/http_request_trailer_test.go b/internal/proxy/http_request_trailer_test.go new file mode 100644 index 0000000..0b8a01f --- /dev/null +++ b/internal/proxy/http_request_trailer_test.go @@ -0,0 +1,256 @@ +package proxy + +import ( + "context" + "errors" + "io" + "net/http" + "reflect" + "slices" + "strings" + "sync" + "testing" + "time" +) + +type forwardTrailerTestBody struct { + read func([]byte) (int, error) + close func() error +} + +func (b *forwardTrailerTestBody) Read(p []byte) (int, error) { return b.read(p) } +func (b *forwardTrailerTestBody) Close() error { + if b.close != nil { + return b.close() + } + return nil +} + +func TestHTTPForwardRequestTrailerEOF(t *testing.T) { + for _, dataWithEOF := range []bool{false, true} { + name := "separate-EOF" + if dataWithEOF { + name = "data-with-EOF" + } + t.Run(name, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + in := &http.Request{Header: http.Header{"Content-Length": {"4"}, "Authorization": {"origin-token"}}, Trailer: http.Header{"X-End": {"stale"}, "X-Absent": {"stale"}}, ContentLength: 4} + in = in.WithContext(ctx) + in.GetBody = func() (io.ReadCloser, error) { return io.NopCloser(strings.NewReader("body")), nil } + calls := 0 + closed := false + closeErr := errors.New("owned close error") + in.Body = &forwardTrailerTestBody{ + read: func(p []byte) (int, error) { + calls++ + if calls == 1 { + if !dataWithEOF { + return copy(p, "body"), nil + } + } else if calls > 2 || dataWithEOF { + return 0, io.EOF + } + in.Trailer = http.Header{"X-End": {"complete"}, "X-Undeclared": {"hidden"}} + if dataWithEOF { + return copy(p, "body"), io.EOF + } + return 0, io.EOF + }, + close: func() error { closed = true; return closeErr }, + } + out := in.Clone(in.Context()) + prepareHTTPForwardRequestTrailers(in, out) + captured := out.Trailer + if out.Context() != ctx || out.Header.Get("Authorization") != "origin-token" || out.ContentLength != -1 || !reflect.DeepEqual(out.TransferEncoding, []string{"chunked"}) || out.GetBody != nil || out.Header.Get("Content-Length") != "" { + t.Fatal("outbound context, authorization or trailer framing changed incorrectly") + } + if in.ContentLength != 4 || in.GetBody == nil || in.Header.Get("Content-Length") != "4" || in.TransferEncoding != nil { + t.Fatal("modified caller's framing or replay getter") + } + p := make([]byte, 4) + n, err := out.Body.Read(p) + if n != 4 || string(p) != "body" || (dataWithEOF && err != io.EOF) || (!dataWithEOF && err != nil) { + t.Fatalf("first Read=(%d,%v), payload=%q", n, err, p) + } + if !dataWithEOF { + if captured.Get("X-End") != "stale" { + t.Fatal("published before EOF") + } + if n, err := out.Body.Read(p); n != 0 || err != io.EOF { + t.Fatalf("second Read=(%d,%v)", n, err) + } + } + if captured.Get("X-End") != "complete" || len(captured) != 2 || len(captured["X-Absent"]) != 0 || captured.Get("X-Undeclared") != "" { + t.Fatalf("EOF metadata in captured map=%v", captured) + } + in.Trailer["X-End"][0] = "caller-mutated" + if _, err := out.Body.Read(p); err != io.EOF || captured.Get("X-End") != "complete" { + t.Fatal("aliased source values or republished after EOF") + } + if err := out.Body.Close(); !closed || err != closeErr { + t.Fatalf("Close error not preserved: %v closed=%v", err, closed) + } + }) + } +} + +func TestHTTPForwardRequestTrailerPolicy(t *testing.T) { + unsafe := []string{"Connection", "Authorization", "Proxy-Authorization", "Proxy-Authenticate", "Proxy-Authentication-Info", "Proxy-Connection", "Content-Length", "Transfer-Encoding", "Trailer", "Upgrade", "Host", "If-Match", "bad key", "X-Ünicode"} + in := &http.Request{ + Header: http.Header{"connection": {" X-Initial , Upgrade"}, "Authorization": {"origin-token"}}, + Trailer: http.Header{ + "X-End": {"initial-a"}, "x-end": {"initial-b"}, "X-Initial": {"hidden"}, + "X-Early-Trailer": {"hidden"}, "X-Late": nil, "Connection": {"X-Early-Trailer"}, + }, + } + for _, key := range unsafe { + if key != "Connection" { + in.Trailer[key] = []string{"hidden"} + } + } + initial := in.Trailer.Clone() + in.Body = &forwardTrailerTestBody{read: func([]byte) (int, error) { + in.Trailer = http.Header{ + "X-End": {"final-a"}, "x-end": {"final-b"}, "X-Initial": {"hidden"}, + "X-Early-Trailer": {"hidden"}, "X-Late": {"hidden"}, "connection": {" x-late "}, + "X-Undeclared": {"hidden"}, + } + for _, key := range unsafe { + if !strings.EqualFold(key, "Connection") { + in.Trailer[key] = []string{"hidden"} + } + } + return 0, io.EOF + }} + out := in.Clone(in.Context()) + prepareHTTPForwardRequestTrailers(in, out) + if !reflect.DeepEqual(in.Trailer, initial) { + t.Fatal("preparation mutated original map") + } + if len(out.Trailer) != 2 || len(out.Trailer["X-End"]) != 2 { + t.Fatalf("initial policy or alias merge=%v", out.Trailer) + } + out.Trailer["X-End"][0] = "out-mutated" + if !reflect.DeepEqual(in.Trailer, initial) || in.Header.Get("Authorization") != "origin-token" { + t.Fatal("outbound metadata aliases input") + } + // Header scrubbing must not erase the saved nominations. + removeHopByHopHeaders(out.Header) + if _, err := out.Body.Read(nil); err != io.EOF { + t.Fatal(err) + } + got := slices.Clone(out.Trailer["X-End"]) + slices.Sort(got) + if len(out.Trailer) != 1 || !reflect.DeepEqual(got, []string{"final-a", "final-b"}) || out.Header.Get("Authorization") != "origin-token" { + t.Fatalf("EOF policy, aliases or origin authorization=%v headers=%v", out.Trailer, out.Header) + } + if in.Trailer.Get("Proxy-Authorization") != "hidden" || in.Trailer["connection"][0] != " x-late " { + t.Fatal("EOF policy changed source metadata") + } +} + +func TestHTTPForwardRequestTrailerNoPublishOnErrorOrClose(t *testing.T) { + readErr, closeErr := errors.New("body read failure"), errors.New("body close failure") + in := &http.Request{Trailer: http.Header{"X-End": {"initial"}}} + in.Body = &forwardTrailerTestBody{ + read: func(p []byte) (int, error) { + in.Trailer["X-End"] = []string{"must-not-publish"} + return copy(p, "x"), readErr + }, + close: func() error { return closeErr }, + } + out := in.Clone(in.Context()) + prepareHTTPForwardRequestTrailers(in, out) + if n, err := out.Body.Read(make([]byte, 1)); n != 1 || err != readErr { + t.Fatalf("Read error/count=(%d,%v)", n, err) + } + if err := out.Body.Close(); err != closeErr || out.Trailer.Get("X-End") != "initial" { + t.Fatalf("Close error=%v trailer=%v", err, out.Trailer) + } +} + +func TestHTTPForwardRequestTrailerCloseInterruptsRead(t *testing.T) { + entered, closed, done := make(chan struct{}), make(chan struct{}), make(chan struct{}) + closeDone := make(chan struct{}) + var readErr, closeErr error + closerStarted := false + var once sync.Once + in := &http.Request{Trailer: http.Header{"X-End": nil}} + in.Body = &forwardTrailerTestBody{ + read: func([]byte) (int, error) { close(entered); <-closed; return 0, io.ErrClosedPipe }, + close: func() error { once.Do(func() { close(closed) }); return nil }, + } + out := in.Clone(in.Context()) + prepareHTTPForwardRequestTrailers(in, out) + // Cleanup independently unblocks and joins even if the success oracle fails. + t.Cleanup(func() { + _ = in.Body.Close() + select { + case <-done: + case <-time.After(time.Second): + t.Error("owned reader did not join") + } + if closerStarted { + select { + case <-closeDone: + case <-time.After(time.Second): + t.Error("owned closer did not join") + } + } + }) + go func() { _, readErr = out.Body.Read(nil); close(done) }() + select { + case <-entered: + case <-time.After(time.Second): + t.Fatal("reader did not enter underlying body") + } + // Run Close with a separate joined worker so a regression cannot deadlock + // the test goroutine before independent cleanup can close the source body. + closerStarted = true + go func() { closeErr = out.Body.Close(); close(closeDone) }() + select { + case <-closeDone: + if closeErr != nil { + t.Fatal(closeErr) + } + case <-time.After(time.Second): + t.Fatal("Close blocked behind body Read") + } + select { + case <-done: + if readErr != io.ErrClosedPipe || len(out.Trailer["X-End"]) != 0 { + t.Fatalf("read error=%v trailer=%v", readErr, out.Trailer) + } + case <-time.After(time.Second): + t.Fatal("Close did not interrupt body Read") + } +} + +func TestHTTPForwardRequestTrailerNoBridge(t *testing.T) { + for _, mode := range []string{"nil-body", "no-body", "no-declaration", "unsafe-only"} { + t.Run(mode, func(t *testing.T) { + in := &http.Request{Header: http.Header{"Content-Length": {"4"}}, ContentLength: 4, TransferEncoding: []string{"identity"}, Trailer: http.Header{"X-End": nil}} + in.Body = io.NopCloser(strings.NewReader("body")) + in.GetBody = func() (io.ReadCloser, error) { return io.NopCloser(strings.NewReader("body")), nil } + switch mode { + case "nil-body": + in.Body = nil + case "no-body": + in.Body = http.NoBody + case "no-declaration": + in.Trailer = nil + case "unsafe-only": + in.Trailer = http.Header{"Authorization": {"secret"}} + } + out := in.Clone(in.Context()) + prepareHTTPForwardRequestTrailers(in, out) + if out.Body != in.Body || out.ContentLength != 4 || out.Header.Get("Content-Length") != "4" || out.GetBody == nil || !reflect.DeepEqual(out.TransferEncoding, in.TransferEncoding) { + t.Fatal("unnecessary body/framing/replay mutation") + } + if mode == "unsafe-only" && len(out.Trailer) != 0 { + t.Fatal("unsafe declarations retained") + } + }) + } +}