Fix reverse-proxy SSE/streaming fidelity (event: lines, gzip, header propagation, Host)

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: <type>" 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

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)
+	}
+}

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: <payload>\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')

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.