diff --git a/authbridge/authlib/listener/internal/sseframe/reader.go b/authbridge/authlib/listener/internal/sseframe/reader.go index e5bd5e0a2..aa4328af8 100644 --- a/authbridge/authlib/listener/internal/sseframe/reader.go +++ b/authbridge/authlib/listener/internal/sseframe/reader.go @@ -46,6 +46,13 @@ type Reader struct { // Reused across ReadFrame calls to avoid per-frame allocation in // the common steady-state case. scratch []byte + // lastEvent holds the "event:" field value of the frame most + // recently returned by ReadFrame (empty if the frame named no + // event type). Reused across calls; valid until the next ReadFrame. + // Preserved so a proxy re-framing the stream can reproduce the + // upstream's SSE event type — clients like the Anthropic SDK type + // each event from the "event:" line, not the data payload. + lastEvent []byte } // NewReader wraps r with the given per-frame size cap (bytes). @@ -84,6 +91,7 @@ const DefaultMaxFrameSize = 1 << 20 // must copy. func (r *Reader) ReadFrame() ([]byte, error) { r.scratch = r.scratch[:0] + r.lastEvent = r.lastEvent[:0] hasData := false for { @@ -122,9 +130,17 @@ func (r *Reader) ReadFrame() ([]byte, error) { // value. Lines without a colon name a field with an empty // value, which we don't care about either way. field, value := splitField(line) + if field == "event" { + // Capture the event type so a re-framing proxy can reproduce + // the upstream "event:" line. Copied because value points into + // readLine's buffer, which is reused on the next line read. + r.lastEvent = append(r.lastEvent[:0], value...) + continue + } if field != "data" { - // We deliberately ignore "event", "id", "retry", and any - // unknown field — the consumer only needs the data payload. + // We deliberately ignore "id", "retry", and any unknown + // field — the consumer only needs the data payload (and the + // event type captured above). continue } @@ -145,6 +161,14 @@ func (r *Reader) ReadFrame() ([]byte, error) { } } +// LastEvent returns the "event:" field value of the frame most recently +// returned by ReadFrame, or an empty slice if that frame named no event +// type. The returned slice is owned by the Reader and reused on the next +// ReadFrame call; copy to retain. A re-framing proxy uses this to emit an +// "event: " line so downstream SSE consumers that type events from +// the event field (e.g. the Anthropic SDK) see the upstream's framing. +func (r *Reader) LastEvent() []byte { return r.lastEvent } + // readLine reads one logical SSE line off the underlying buffered // reader. Line terminators per the SSE spec are LF, CR, or CRLF — // any of the three ends a line. The returned slice excludes the diff --git a/authbridge/authlib/listener/internal/sseframe/reader_test.go b/authbridge/authlib/listener/internal/sseframe/reader_test.go index e593db4f1..c2eac3442 100644 --- a/authbridge/authlib/listener/internal/sseframe/reader_test.go +++ b/authbridge/authlib/listener/internal/sseframe/reader_test.go @@ -163,3 +163,69 @@ func TestReader_LongDataLineExceedsBufioBuffer(t *testing.T) { t.Errorf("frame len = %d, want %d", len(frame), n) } } + +// TestReader_LastEvent_Captured locks in the fix that lets a +// re-framing proxy reproduce the upstream's "event:" line: after +// ReadFrame returns the data payload, LastEvent must return the +// event type named on the preceding "event:" line. +func TestReader_LastEvent_Captured(t *testing.T) { + body := "event: message_start\ndata: {\"x\":1}\n\n" + r := NewReader(strings.NewReader(body), 0) + frame, err := r.ReadFrame() + if err != nil { + t.Fatalf("ReadFrame: %v", err) + } + if string(frame) != `{"x":1}` { + t.Errorf("frame = %q, want %q", frame, `{"x":1}`) + } + if got := string(r.LastEvent()); got != "message_start" { + t.Errorf("LastEvent() = %q, want %q", got, "message_start") + } +} + +// TestReader_LastEvent_ResetPerFrame confirms the event type does not +// stick across frames: a frame with no "event:" line must report an +// empty LastEvent even when the previous frame named one. +func TestReader_LastEvent_ResetPerFrame(t *testing.T) { + body := "event: message_start\ndata: one\n\ndata: two\n\n" + r := NewReader(strings.NewReader(body), 0) + + frame, err := r.ReadFrame() + if err != nil { + t.Fatalf("first ReadFrame: %v", err) + } + if string(frame) != "one" { + t.Errorf("first frame = %q, want one", frame) + } + if got := string(r.LastEvent()); got != "message_start" { + t.Errorf("first LastEvent() = %q, want %q", got, "message_start") + } + + frame, err = r.ReadFrame() + if err != nil { + t.Fatalf("second ReadFrame: %v", err) + } + if string(frame) != "two" { + t.Errorf("second frame = %q, want two", frame) + } + if got := r.LastEvent(); len(got) != 0 { + t.Errorf("second LastEvent() = %q, want empty (not sticky across frames)", got) + } +} + +// TestReader_LastEvent_EmptyForDataOnlyFrame preserves the existing +// data-only behavior: a frame with no "event:" line at all reports an +// empty LastEvent. +func TestReader_LastEvent_EmptyForDataOnlyFrame(t *testing.T) { + r := NewReader(strings.NewReader("data: hello\n\n"), 0) + frame, err := r.ReadFrame() + if err != nil { + t.Fatalf("ReadFrame: %v", err) + } + if string(frame) != "hello" { + t.Errorf("frame = %q, want hello", frame) + } + if got := r.LastEvent(); len(got) != 0 { + t.Errorf("LastEvent() = %q, want empty", got) + } +} diff --git a/authbridge/authlib/listener/reverseproxy/server.go b/authbridge/authlib/listener/reverseproxy/server.go index ae1701b5c..8a5eba5b6 100644 --- a/authbridge/authlib/listener/reverseproxy/server.go +++ b/authbridge/authlib/listener/reverseproxy/server.go @@ -93,6 +93,35 @@ func NewServer(inbound *pipeline.Holder, sessions *session.Store, backendURL str return nil, err } proxy := httputil.NewSingleHostReverseProxy(target) + // The default Director rewrites the outbound scheme/host/path but + // deliberately leaves req.Host as the inbound caller's Host (e.g. + // "authbridge-ab1:8080"). Cloudflare-fronted backends like + // api.anthropic.com validate Host against the request line and + // reject a mismatch, so wrap the Director to rewrite it to the + // backend target's host after the default rewrite runs. + orig := proxy.Director + proxy.Director = func(req *http.Request) { + orig(req) + req.Host = target.Host + // Strip the client's Accept-Encoding, but only when a plugin will + // actually inspect the response body: a StreamingResponder (SSE + // re-framing) or any ReadsBody/WritesBody plugin (buffered read into + // pctx.ResponseBody). Those paths must see plaintext — with no explicit + // Accept-Encoding, Go's transport negotiates gzip itself and + // transparently decompresses the response (dropping Content-Encoding / + // Content-Length), so both the inspection here and the downstream client + // get plaintext. Leaving an explicit "Accept-Encoding: gzip" through + // would yield a gzipped body the SSE re-framer reads as garbage plus a + // Content-Encoding: gzip header over re-emitted plaintext — which makes + // SSE clients (e.g. the Anthropic SDK) decode zero events. When no plugin + // reads the response body this proxy is a pure pass-through, so leave + // Accept-Encoding intact rather than force needless upstream→client + // decompression (matters for a remote backend or large non-streamed + // bodies; harmless on a localhost sidecar hop). + if inbound.NeedsBody() || inbound.HasStreamingResponders() { + req.Header.Del("Accept-Encoding") + } + } // FlushInterval -1 makes ReverseProxy flush after every Read of // the response body. Required for streaming text/event-stream // responses where each frame must hit the client immediately — @@ -221,7 +250,6 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) { slog.Debug("reverse-proxy: buffered request body", "host", r.Host, "bodyLen", len(body)) } - originalAuth := pctx.Headers.Get("Authorization") action := s.InboundPipeline.Run(r.Context(), pctx) if action.Type == pipeline.Reject { s.recordInboundReject(pctx, action) @@ -239,13 +267,26 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) { r.Header.Del("Content-Encoding") } - // Propagate an inbound Authorization mutation to the forwarded - // request. A plugin (e.g. jwt-validation in mint mode) may have - // replaced the caller's token on pctx.Headers; the proxy forwards - // r.Header, so without this the backend would still see the original - // token. Only rewrite when the value actually changed. - if newAuth := pctx.Headers.Get("Authorization"); newAuth != originalAuth { - r.Header.Set("Authorization", newAuth) + // Propagate every header mutation the inbound pipeline made to the forwarded + // request. pctx.Headers started as a clone of r.Header, so plugins' set / replace + // / delete operations on it are the intended backend-facing header set. Only + // Authorization used to be forwarded, silently dropping any other injected header + // (e.g. static-inject's x-api-key). Content-Length / Content-Encoding are managed + // by the body-rewrite block above and the transport, so leave them untouched. + skip := func(k string) bool { return k == "Content-Length" || k == "Content-Encoding" } + for k := range r.Header { + if skip(k) { + continue + } + if _, ok := pctx.Headers[k]; !ok { + r.Header.Del(k) // plugin removed it + } + } + for k, vv := range pctx.Headers { + if skip(k) { + continue + } + r.Header[k] = append([]string(nil), vv...) // set / overwrite } // Record the inbound request event whenever there is something @@ -609,7 +650,16 @@ func (b *streamingResponseBody) Read(p []byte) (int, error) { // gets its own `data: ` prefix and the downstream parser sees the // same event boundaries the upstream produced. For single-line // JSON-RPC payloads this is equivalent to one `data: \n\n`. - out := make([]byte, 0, len(frame)+8) + out := make([]byte, 0, len(frame)+16) + // Preserve the upstream SSE "event:" line. sseframe surfaces only the + // data payload, but clients like the Anthropic SDK type each event from + // the "event:" field; dropping it makes the re-framed stream parse to + // zero typed events downstream. + if ev := b.reader.LastEvent(); len(ev) > 0 { + out = append(out, "event: "...) + out = append(out, ev...) + out = append(out, '\n') + } rest := frame for len(rest) > 0 { nl := bytes.IndexByte(rest, '\n') diff --git a/authbridge/authlib/listener/reverseproxy/server_test.go b/authbridge/authlib/listener/reverseproxy/server_test.go index b418eee5f..3ff2d6daa 100644 --- a/authbridge/authlib/listener/reverseproxy/server_test.go +++ b/authbridge/authlib/listener/reverseproxy/server_test.go @@ -109,6 +109,163 @@ func TestReverseProxy_BypassPath(t *testing.T) { } } +// TestReverseProxy_RewritesHostToBackend locks in the fix for the +// Cloudflare-fronted-backend bug: httputil.NewSingleHostReverseProxy's +// default Director rewrites the outbound URL but deliberately leaves +// req.Host as the inbound caller's Host, so backends like +// api.anthropic.com that validate Host reject the forwarded request. +// The wrapped Director must set req.Host to the backend target's host +// after the default rewrite runs. +func TestReverseProxy_RewritesHostToBackend(t *testing.T) { + var gotHost string + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotHost = r.Host + w.WriteHeader(http.StatusOK) + })) + defer backend.Close() + + p, err := pipeline.New(nil) + if err != nil { + t.Fatal(err) + } + srv, err := NewServer(pipeline.NewHolder(p), nil, backend.URL, nil) + if err != nil { + t.Fatal(err) + } + + proxy := httptest.NewServer(srv.Handler()) + defer proxy.Close() + + req, _ := http.NewRequest("GET", proxy.URL+"/api/data", nil) + req.Host = "authbridge-ab1:8080" // inbound Host, deliberately different from the backend + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + backendHost := strings.TrimPrefix(strings.TrimPrefix(backend.URL, "https://"), "http://") + if gotHost != backendHost { + t.Errorf("backend received Host = %q, want %q (backend target host)", gotHost, backendHost) + } +} + +// --- Header Propagation Tests --- + +// headerMutatorPlugin applies an arbitrary mutation function to +// pctx.Headers during OnRequest. Used to simulate plugins like +// static-inject (set x-api-key, delete Authorization) and +// jwt-validation's mint mode (replace Authorization) without pulling +// in their real implementations. +type headerMutatorPlugin struct { + mutate func(h http.Header) +} + +func (p *headerMutatorPlugin) Name() string { return "header-mutator" } +func (p *headerMutatorPlugin) Capabilities() pipeline.PluginCapabilities { + return pipeline.PluginCapabilities{} +} +func (p *headerMutatorPlugin) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action { + p.mutate(pctx.Headers) + return pipeline.Action{Type: pipeline.Continue} +} +func (p *headerMutatorPlugin) OnResponse(_ context.Context, _ *pipeline.Context) pipeline.Action { + return pipeline.Action{Type: pipeline.Continue} +} + +// TestReverseProxy_HeaderMutation_SetAndDelete locks in the fix for the +// static-inject 401 bug: a plugin that sets a new header (x-api-key) +// and deletes Authorization on pctx.Headers must have BOTH mutations +// reach the backend. Before this fix, the reverse proxy only ever +// propagated an Authorization change, so x-api-key never reached +// api.anthropic.com and the stale Authorization was still forwarded. +func TestReverseProxy_HeaderMutation_SetAndDelete(t *testing.T) { + var gotAPIKey string + var gotAuthOK bool + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotAPIKey = r.Header.Get("x-api-key") + _, gotAuthOK = r.Header["Authorization"] + w.WriteHeader(http.StatusOK) + })) + defer backend.Close() + + mutator := &headerMutatorPlugin{mutate: func(h http.Header) { + h.Set("x-api-key", "REALKEY") + h.Del("Authorization") + }} + p, err := pipeline.New([]pipeline.Plugin{mutator}) + if err != nil { + t.Fatal(err) + } + srv, err := NewServer(pipeline.NewHolder(p), nil, backend.URL, nil) + if err != nil { + t.Fatal(err) + } + proxy := httptest.NewServer(srv.Handler()) + defer proxy.Close() + + req, _ := http.NewRequest("GET", proxy.URL+"/v1/messages", nil) + req.Header.Set("Authorization", "Bearer PLACEHOLDER") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + + if gotAPIKey != "REALKEY" { + t.Errorf("backend x-api-key = %q, want %q", gotAPIKey, "REALKEY") + } + if gotAuthOK { + t.Errorf("backend received an Authorization header; want it deleted") + } +} + +// Authorization-replace (mint) regression coverage already exists in +// placeholder_test.go's TestInboundPropagation_RewrittenAuthReachesBackend +// (a plugin that replaces Authorization must have the new value reach +// the backend). Confirmed still green under the full header-diff sync. + +// TestReverseProxy_HeaderMutation_PassThroughUnchanged confirms a +// header the pipeline never touched still reaches the backend +// unchanged under the new diff-sync (as opposed to only forwarding +// whatever the pipeline explicitly set). +func TestReverseProxy_HeaderMutation_PassThroughUnchanged(t *testing.T) { + var gotCustom string + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotCustom = r.Header.Get("X-Untouched") + w.WriteHeader(http.StatusOK) + })) + defer backend.Close() + + // Plugin mutates an unrelated header so pctx.Headers diverges from + // a pure clone, but never touches X-Untouched. + mutator := &headerMutatorPlugin{mutate: func(h http.Header) { + h.Set("x-api-key", "REALKEY") + }} + p, err := pipeline.New([]pipeline.Plugin{mutator}) + if err != nil { + t.Fatal(err) + } + srv, err := NewServer(pipeline.NewHolder(p), nil, backend.URL, nil) + if err != nil { + t.Fatal(err) + } + proxy := httptest.NewServer(srv.Handler()) + defer proxy.Close() + + req, _ := http.NewRequest("GET", proxy.URL+"/api", nil) + req.Header.Set("X-Untouched", "still-here") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + + if gotCustom != "still-here" { + t.Errorf("backend X-Untouched = %q, want %q (pass-through header dropped)", gotCustom, "still-here") + } +} + // --- Body Buffering Tests --- // bodyRecorderPlugin records whether it received a body during OnRequest. diff --git a/authbridge/authlib/listener/reverseproxy/streaming_test.go b/authbridge/authlib/listener/reverseproxy/streaming_test.go index 17b2fdddb..25fa7d305 100644 --- a/authbridge/authlib/listener/reverseproxy/streaming_test.go +++ b/authbridge/authlib/listener/reverseproxy/streaming_test.go @@ -8,6 +8,7 @@ import ( "io" "net/http" "net/http/httptest" + "strings" "sync" "testing" "time" @@ -232,3 +233,173 @@ func framesAsStrings(frames [][]byte) []string { } return out } + +// TestReverseProxy_Streaming_PreservesEventLine is the regression +// guard for the SSE "event:" line bug: a client reading through the +// proxy must see the upstream's "event: " line reproduced +// before each re-framed "data:" line, not just the bare data payload. +// The Anthropic SDK types each event from the "event:" field, so +// dropping it made downstream clients decode zero typed events even +// though the data bytes were intact. +func TestReverseProxy_Streaming_PreservesEventLine(t *testing.T) { + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + flusher := w.(http.Flusher) + fmt.Fprintf(w, "event: message_start\ndata: {\"type\":\"message_start\"}\n\n") + flusher.Flush() + fmt.Fprintf(w, "event: content_block_delta\ndata: {\"type\":\"content_block_delta\"}\n\n") + flusher.Flush() + })) + defer backend.Close() + + p, err := pipeline.New(nil) + if err != nil { + t.Fatalf("New pipeline: %v", err) + } + srv, err := NewServer(pipeline.NewHolder(p), nil, backend.URL, nil) + if err != nil { + t.Fatalf("NewServer: %v", err) + } + proxy := httptest.NewServer(srv.Handler()) + defer proxy.Close() + + resp, err := http.Get(proxy.URL + "/stream") + if err != nil { + t.Fatalf("Get: %v", err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !bytes.Contains(body, []byte("event: message_start\n")) { + t.Errorf("body missing %q — event: line not preserved; body=%q", "event: message_start\n", body) + } + if !bytes.Contains(body, []byte("event: content_block_delta\n")) { + t.Errorf("body missing %q — event: line not preserved; body=%q", "event: content_block_delta\n", body) + } + if got := bytes.Count(body, []byte("data:")); got != 2 { + t.Errorf("body has %d data: lines, want 2 — body=%q", got, body) + } +} + +// roundTripFunc adapts a function to http.RoundTripper, letting a test +// substitute the reverse proxy's outbound Transport with one that +// inspects the exact request the Director produced — bypassing net/http. +// Transport's own automatic Accept-Encoding negotiation, which adds +// "gzip" on the wire whenever the header is unset and would otherwise +// make an unset header indistinguishable from a deleted one at a real +// backend. +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { return f(req) } + +// TestReverseProxy_StripsAcceptEncoding is the regression guard for +// the gzip bug: when a plugin inspects the response body, the Director +// must strip the client's Accept-Encoding so Go's transport negotiates +// (and transparently decompresses) gzip itself. Leaving an explicit +// "Accept-Encoding: gzip" through produced a gzipped body the SSE +// re-framer read as garbage plus a Content-Encoding: gzip header over +// re-emitted plaintext. The pipeline here carries a body-inspecting +// StreamingResponder (streamingProbe, ReadsBody=true) so the strip is +// in scope. +// +// This asserts against the outbound request the Director hands to the +// Transport rather than what a real backend observes on the wire: +// net/http.Transport itself adds "Accept-Encoding: gzip" to the wire +// request whenever the header is unset (to negotiate transparent +// decompression), so a real backend sees "gzip" either way — the +// difference the fix controls is whether that header is explicit +// (caller-set, decompression left to the caller) or absent (Transport +// negotiates and decompresses transparently). Substituting the +// Transport lets the test observe that distinction directly. +func TestReverseProxy_StripsAcceptEncoding(t *testing.T) { + var gotAcceptEncoding string + var sawHeader bool + + p, err := pipeline.New([]pipeline.Plugin{newStreamingProbe(false)}) + if err != nil { + t.Fatalf("New pipeline: %v", err) + } + srv, err := NewServer(pipeline.NewHolder(p), nil, "http://backend.invalid", nil) + if err != nil { + t.Fatalf("NewServer: %v", err) + } + srv.proxy.Transport = roundTripFunc(func(req *http.Request) (*http.Response, error) { + gotAcceptEncoding = req.Header.Get("Accept-Encoding") + _, sawHeader = req.Header["Accept-Encoding"] + return &http.Response{ + StatusCode: http.StatusOK, + Status: "200 OK", + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader("")), + Request: req, + }, nil + }) + proxy := httptest.NewServer(srv.Handler()) + defer proxy.Close() + + req, err := http.NewRequest("GET", proxy.URL+"/api/data", nil) + if err != nil { + t.Fatalf("NewRequest: %v", err) + } + req.Header.Set("Accept-Encoding", "gzip") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("Do: %v", err) + } + resp.Body.Close() + + if sawHeader { + t.Errorf("outbound request Accept-Encoding = %q, want stripped (Director did not delete it)", gotAcceptEncoding) + } +} + +// TestReverseProxy_PreservesAcceptEncoding_PassThrough is the flip side +// of the strip: when no plugin reads the response body, the proxy is a +// pure pass-through and must NOT strip Accept-Encoding — forcing +// upstream→client decompression there is a needless regression (matters +// for a remote backend or large non-streamed bodies). The pipeline is +// empty, so neither NeedsBody() nor HasStreamingResponders() holds and +// the Director leaves the caller's explicit header intact. +func TestReverseProxy_PreservesAcceptEncoding_PassThrough(t *testing.T) { + var gotAcceptEncoding string + + p, err := pipeline.New(nil) + if err != nil { + t.Fatalf("New pipeline: %v", err) + } + srv, err := NewServer(pipeline.NewHolder(p), nil, "http://backend.invalid", nil) + if err != nil { + t.Fatalf("NewServer: %v", err) + } + srv.proxy.Transport = roundTripFunc(func(req *http.Request) (*http.Response, error) { + gotAcceptEncoding = req.Header.Get("Accept-Encoding") + return &http.Response{ + StatusCode: http.StatusOK, + Status: "200 OK", + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader("")), + Request: req, + }, nil + }) + proxy := httptest.NewServer(srv.Handler()) + defer proxy.Close() + + req, err := http.NewRequest("GET", proxy.URL+"/api/data", nil) + if err != nil { + t.Fatalf("NewRequest: %v", err) + } + req.Header.Set("Accept-Encoding", "gzip") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("Do: %v", err) + } + resp.Body.Close() + + if gotAcceptEncoding != "gzip" { + t.Errorf("outbound request Accept-Encoding = %q, want %q preserved (pure pass-through must not strip)", gotAcceptEncoding, "gzip") + } +}