diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index ccebaf3..ee14445 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -253,6 +253,15 @@ progress as activity, including small buffered responses. Stalled request or response bodies and blocked writes remain bounded; header and keepalive timeouts are unchanged. +Source builds also forward unknown-length HTTP response bodies and +`text/event-stream` events without waiting for the origin to finish. Fixed-size +non-streaming responses retain normal buffering. If copying an origin response +fails, the proxy aborts that response instead of emitting a successful final +chunk: HTTP/1.1 clients can detect truncation. Failure before buffered headers +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. + 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 1456093..8851edd 100644 --- a/internal/proxy/http.go +++ b/internal/proxy/http.go @@ -5,6 +5,7 @@ import ( "encoding/base64" "errors" "io" + "mime" "net" "net/http" "net/textproto" @@ -290,7 +291,13 @@ func (s *HTTPServer) serveForward(w http.ResponseWriter, r *http.Request) { // writer buffers small chunks and the client sends nothing more. body = &httpResponseActivityReader{Reader: body, conn: conn} } - _, _ = io.Copy(w, body) + if err := copyHTTPResponse(w, body, response); err != nil { + // Headers may already be on the wire. Returning normally would let + // net/http finalize a chunked response (or infer a short Content-Length), + // hiding a truncated origin body from the client. Abort only this + // response; net/http handles the sentinel without logging a stack trace. + panic(http.ErrAbortHandler) + } for key, values := range response.Trailer { if isHopByHopHeader(key) { continue @@ -299,6 +306,42 @@ func (s *HTTPServer) serveForward(w http.ResponseWriter, r *http.Request) { } } +// Unknown-length responses and server-sent events can remain open indefinitely. +// Flush their headers and each body write so small events don't wait in the +// HTTP server's response buffer until the origin finishes. Ordinary fixed-size +// responses retain net/http's buffering and copy optimizations. +func copyHTTPResponse(w http.ResponseWriter, body io.Reader, response *http.Response) error { + contentType, _, _ := mime.ParseMediaType(response.Header.Get("Content-Type")) + var destination io.Writer = w + if response.ContentLength == -1 || contentType == "text/event-stream" { + flush := http.NewResponseController(w).Flush + if err := flush(); err != nil { + // Serve's native ResponseWriter supports flushing. Preserve Handler + // compatibility for embedders whose wrappers don't expose it. + if !errors.Is(err, http.ErrNotSupported) { + return err + } + } else { + destination = httpResponseFlushWriter{writer: w, flush: flush} + } + } + _, err := io.Copy(destination, body) + return err +} + +type httpResponseFlushWriter struct { + writer io.Writer + flush func() error +} + +func (w httpResponseFlushWriter) Write(p []byte) (int, error) { + n, err := w.writer.Write(p) + if n > 0 && err == nil { + err = w.flush() + } + return n, err +} + type httpResponseActivityReader struct { io.Reader conn *trackedConn diff --git a/internal/proxy/http_response_copy_test.go b/internal/proxy/http_response_copy_test.go new file mode 100644 index 0000000..6a48558 --- /dev/null +++ b/internal/proxy/http_response_copy_test.go @@ -0,0 +1,84 @@ +package proxy + +import ( + "bytes" + "errors" + "io" + "net/http" + "strings" + "testing" + "testing/iotest" +) + +func TestCopyHTTPResponseFlushAndErrors(t *testing.T) { + copyErr := errors.New("response copy fixture") + for _, test := range []struct { + name string + length int64 + contentType string + hideFlusher bool + flushErrorAt int + writeError bool + readError bool + wantFlushes int + wantBody string + wantError bool + }{ + {name: "unknown length", length: -1, wantFlushes: 2, wantBody: "body"}, + {name: "event stream", length: 4, contentType: "text/event-stream; charset=utf-8", wantFlushes: 2, wantBody: "body"}, + {name: "ordinary fixed size", length: 4, contentType: "text/plain", wantBody: "body"}, + {name: "wrapper without flusher", length: -1, hideFlusher: true, wantBody: "body"}, + {name: "header flush error", length: -1, flushErrorAt: 1, wantFlushes: 1, wantError: true}, + {name: "body flush error", length: -1, flushErrorAt: 2, wantFlushes: 2, wantBody: "body", wantError: true}, + {name: "body write error", length: -1, writeError: true, wantFlushes: 1, wantError: true}, + {name: "body read error", length: -1, readError: true, wantFlushes: 1, wantError: true}, + } { + t.Run(test.name, func(t *testing.T) { + writer := &httpCopyTestWriter{header: make(http.Header), err: copyErr, flushErrorAt: test.flushErrorAt, writeError: test.writeError} + var destination http.ResponseWriter = writer + if test.hideFlusher { + destination = struct{ http.ResponseWriter }{writer} + } + var source io.Reader = strings.NewReader("body") + if test.readError { + source = iotest.ErrReader(copyErr) + } + response := &http.Response{ContentLength: test.length, Header: make(http.Header)} + response.Header.Set("Content-Type", test.contentType) + err := copyHTTPResponse(destination, source, response) + if (err != nil) != test.wantError || (test.wantError && !errors.Is(err, copyErr)) { + t.Fatalf("copy error = %v, want fixture error = %v", err, test.wantError) + } + if writer.flushes != test.wantFlushes || writer.body.String() != test.wantBody { + t.Fatalf("flushes/body = %d/%q, want %d/%q", writer.flushes, writer.body.String(), test.wantFlushes, test.wantBody) + } + }) + } +} + +type httpCopyTestWriter struct { + header http.Header + body bytes.Buffer + err error + flushes int + flushErrorAt int + writeError bool +} + +func (w *httpCopyTestWriter) Header() http.Header { return w.header } +func (*httpCopyTestWriter) WriteHeader(int) {} + +func (w *httpCopyTestWriter) Write(p []byte) (int, error) { + if w.writeError { + return 0, w.err + } + return w.body.Write(p) +} + +func (w *httpCopyTestWriter) FlushError() error { + w.flushes++ + if w.flushes == w.flushErrorAt { + return w.err + } + return nil +} diff --git a/internal/proxy/http_response_integrity_test.go b/internal/proxy/http_response_integrity_test.go new file mode 100644 index 0000000..87f7d76 --- /dev/null +++ b/internal/proxy/http_response_integrity_test.go @@ -0,0 +1,119 @@ +package proxy + +import ( + "bytes" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +func TestHTTPForwardRejectsTruncatedChunkedResponse(t *testing.T) { + for _, size := range []int{32, 64 << 10} { + t.Run(fmt.Sprint(size), func(t *testing.T) { + payload := bytes.Repeat([]byte("x"), size) + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + conn, rw, err := w.(http.Hijacker).Hijack() + if err != nil { + t.Error(err) + return + } + defer conn.Close() + // Send a complete data chunk but omit the terminal zero chunk. + // The body length is unknown, so a proxy that returns normally + // after a copy error can falsely turn this into a complete reply. + _, _ = fmt.Fprintf(rw, "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n%x\r\n", len(payload)) + _, _ = rw.Write(payload) + _, _ = rw.WriteString("\r\n") + _ = rw.Flush() + })) + defer origin.Close() + server, proxyURL, stop := startHTTPProxy(t, Config{Dialer: directDialer()}) + defer stop(server) + client := proxyHTTPClient(t, proxyURL, nil) + defer client.CloseIdleConnections() + response, err := client.Get(origin.URL) + if err != nil { + t.Fatalf("read streamed response headers: %v", err) + } + defer response.Body.Close() + got, err := io.ReadAll(response.Body) + if !errors.Is(err, io.ErrUnexpectedEOF) || response.StatusCode != http.StatusOK || !bytes.Equal(got, payload) { + t.Fatalf("truncated chunked response: status=%d bytes=%d error=%v; want 200, %d bytes and unexpected EOF", response.StatusCode, len(got), err, size) + } + }) + } +} + +func TestHTTPForwardStreamsBeforeOriginCompletes(t *testing.T) { + for _, kind := range []string{"unknown length", "event stream with known length"} { + t.Run(kind, func(t *testing.T) { + const first = "data: first\n\n" + const last = "data: last\n\n" + release := make(chan struct{}) + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if kind == "event stream with known length" { + w.Header().Set("Content-Type", "text/event-stream; charset=utf-8") + w.Header().Set("Content-Length", fmt.Sprint(len(first)+len(last))) + } + _, _ = io.WriteString(w, first) + w.(http.Flusher).Flush() + select { + case <-release: + _, _ = io.WriteString(w, last) + case <-r.Context().Done(): + } + })) + defer origin.Close() + server, proxyURL, stop := startHTTPProxy(t, Config{Dialer: directDialer()}) + defer stop(server) + client := proxyHTTPClient(t, proxyURL, nil) + client.Timeout = 2 * time.Second + defer client.CloseIdleConnections() + response, err := client.Get(origin.URL) + if err != nil { + t.Fatalf("receive response while origin remains open: %v", err) + } + defer response.Body.Close() + prefix := make([]byte, len(first)) + if _, err := io.ReadFull(response.Body, prefix); err != nil { + t.Fatalf("read first event before releasing origin: %v", err) + } + if string(prefix) != first { + t.Fatalf("first event = %q", prefix) + } + close(release) + rest, err := io.ReadAll(response.Body) + if err != nil || string(rest) != last { + t.Fatalf("remaining stream = %q, %v", rest, err) + } + }) + } +} + +func TestHTTPForwardCompleteChunkedResponseWithTrailers(t *testing.T) { + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Trailer", "X-Body-Complete") + _, _ = io.WriteString(w, "first") + w.(http.Flusher).Flush() + _, _ = io.WriteString(w, "last") + w.Header().Set("X-Body-Complete", "yes") + })) + defer origin.Close() + server, proxyURL, stop := startHTTPProxy(t, Config{Dialer: directDialer()}) + defer stop(server) + client := proxyHTTPClient(t, proxyURL, nil) + defer client.CloseIdleConnections() + response, err := client.Get(origin.URL) + if err != nil { + t.Fatal(err) + } + defer response.Body.Close() + body, err := io.ReadAll(response.Body) + if err != nil || string(body) != "firstlast" || response.Trailer.Get("X-Body-Complete") != "yes" { + t.Fatalf("complete response body=%q trailer=%v error=%v", body, response.Trailer, err) + } +} diff --git a/internal/proxy/http_test.go b/internal/proxy/http_test.go index 30f18e8..bbafded 100644 --- a/internal/proxy/http_test.go +++ b/internal/proxy/http_test.go @@ -375,12 +375,17 @@ func TestHTTPActiveBodyIdleTimeouts(t *testing.T) { server, proxyURL, stopProxy := startHTTPProxy(t, Config{Dialer: directDialer(), IdleTimeout: idle}) defer stopProxy(server) client := proxyHTTPClient(t, proxyURL, nil) + started := time.Now() response, err := client.Get(origin.URL + "/stall") if err != nil { - t.Fatal(err) + // A fixed-size response with no body can remain buffered. Aborting + // the failed origin copy may therefore close before headers arrive. + if !errors.Is(err, io.EOF) || time.Since(started) > time.Second { + t.Fatalf("stalled response did not abort promptly: %v", err) + } + return } defer response.Body.Close() - started := time.Now() _, err = io.ReadAll(response.Body) if err == nil { t.Fatal("truncated stalled response unexpectedly completed successfully")