fix: warn when a ReadsBody plugin lands in SSE passthrough

A plugin that declares ReadsBody but is not a StreamingResponder falls
into streamPassthrough, where its OnResponse runs against an empty
pctx.ResponseBody (the stream is forwarded unbuffered to avoid the #642
timeout). Emit a slog.Warn -- mirroring the existing WritesBody fallback
-- so the misconfiguration surfaces instead of silently starving the
plugin of the body. Buffering is intentionally not the fix; such a plugin
should implement StreamingResponder.

Add TestForwardProxy_SSE_ReadsBodyPluginWarnsAndStreams: the stream is
still delivered verbatim, OnResponse sees an empty body, and the warning
fires.

Addresses review feedback on #675.

Assisted-By: Claude (Anthropic AI) <noreply@anthropic.com>
Signed-off-by: Hai Huang <huang195@gmail.com>

Author: huang195

authbridge/authlib/listener/forwardproxy/mcp_sse_stream_test.go

@@ -5,9 +5,11 @@ import (
 	"bytes"
 	"context"
 	"io"
+	"log/slog"
 	"net/http"
 	"net/http/httptest"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -139,27 +141,54 @@ func TestForwardProxy_SSE_StreamsWithoutResponder(t *testing.T) {
 // responseProbePlugin is a minimal non-StreamingResponder plugin whose
 // OnResponse uses only status/headers (like opa / litellm-budgettrack). It
 // records that it ran and can deny — used to prove streamPassthrough still runs
-// the response-phase pipeline before streaming.
+// the response-phase pipeline before streaming. When readsBody is set it also
+// declares ReadsBody, so it exercises the ReadsBody-but-not-StreamingResponder
+// footgun (see TestForwardProxy_SSE_ReadsBodyPluginWarnsAndStreams).
 type responseProbePlugin struct {
-	deny    bool
-	ranResp atomic.Bool
+	deny      bool
+	readsBody bool
+	ranResp   atomic.Bool
+	// sawBodyLen is the length of pctx.ResponseBody observed in OnResponse.
+	sawBodyLen atomic.Int64
 }
 
 func (p *responseProbePlugin) Name() string { return "response-probe" }
 func (p *responseProbePlugin) Capabilities() pipeline.PluginCapabilities {
-	return pipeline.PluginCapabilities{} // no ReadsBody/WritesBody/StreamingResponder → streamPassthrough path
+	// Never a StreamingResponder → streamPassthrough path. ReadsBody is opt-in
+	// so most callers stay on the status/header-only shape.
+	return pipeline.PluginCapabilities{ReadsBody: p.readsBody}
 }
 func (p *responseProbePlugin) OnRequest(context.Context, *pipeline.Context) pipeline.Action {
 	return pipeline.Action{Type: pipeline.Continue}
 }
-func (p *responseProbePlugin) OnResponse(_ context.Context, _ *pipeline.Context) pipeline.Action {
+func (p *responseProbePlugin) OnResponse(_ context.Context, pctx *pipeline.Context) pipeline.Action {
 	p.ranResp.Store(true)
+	p.sawBodyLen.Store(int64(len(pctx.ResponseBody)))
 	if p.deny {
 		return pipeline.DenyStatus(403, "test.denied", "denied by response probe")
 	}
 	return pipeline.Action{Type: pipeline.Continue}
 }
 
+// syncBuffer is a goroutine-safe io.Writer for capturing slog output written
+// from the proxy's request-handler goroutine while the test reads it.
+type syncBuffer struct {
+	mu  sync.Mutex
+	buf bytes.Buffer
+}
+
+func (b *syncBuffer) Write(p []byte) (int, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.Write(p)
+}
+
+func (b *syncBuffer) String() string {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.String()
+}
+
 // sseProxy wires an SSE upstream + a forward proxy with the given pipeline and
 // returns a client bound to the proxy. The upstream writes one event and
 // returns (closes) — enough to exercise the response-phase decision.
@@ -245,3 +274,52 @@ func TestForwardProxy_SSE_HeaderOnResponseRuns(t *testing.T) {
 		t.Errorf("SSE body not delivered verbatim: %q", body)
 	}
 }
+
+// TestForwardProxy_SSE_ReadsBodyPluginWarnsAndStreams covers the ReadsBody
+// footgun raised in review: a plugin that declares ReadsBody but is NOT a
+// StreamingResponder falls into streamPassthrough. The proxy must NOT buffer to
+// feed it a body (that would reintroduce the #642 timeout on a live stream), so
+// the plugin's OnResponse sees an empty pctx.ResponseBody. The stream must still
+// be delivered verbatim, and the misconfiguration must be surfaced via a warning
+// rather than silently starving the plugin of the body.
+func TestForwardProxy_SSE_ReadsBodyPluginWarnsAndStreams(t *testing.T) {
+	// Capture slog output. The warning is emitted from the proxy's handler
+	// goroutine, so use a mutex-guarded buffer to stay race-clean; tests in this
+	// package don't run in parallel, so swapping the default logger is safe.
+	logBuf := &syncBuffer{}
+	prev := slog.Default()
+	slog.SetDefault(slog.New(slog.NewTextHandler(logBuf, &slog.HandlerOptions{Level: slog.LevelWarn})))
+	t.Cleanup(func() { slog.SetDefault(prev) })
+
+	probe := &responseProbePlugin{readsBody: true}
+	client, url := sseProxy(t, []pipeline.Plugin{probe})
+
+	req, _ := http.NewRequest("POST", url, bytes.NewReader([]byte(`{"jsonrpc":"2.0","id":1,"method":"tools/list"}`)))
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := client.Do(req)
+	if err != nil {
+		t.Fatalf("request failed: %v", err)
+	}
+	defer resp.Body.Close()
+	body, _ := io.ReadAll(resp.Body)
+
+	if !probe.ranResp.Load() {
+		t.Error("OnResponse did not run on the SSE response")
+	}
+	// The plugin declared ReadsBody, but the stream is passed through unbuffered,
+	// so it sees no body — the documented limitation this warning surfaces.
+	if n := probe.sawBodyLen.Load(); n != 0 {
+		t.Errorf("ReadsBody plugin saw %d body bytes, want 0 (stream is not buffered for it)", n)
+	}
+	// The stream must still be delivered verbatim — no regression to the #642 timeout.
+	if resp.StatusCode != http.StatusOK {
+		t.Errorf("status = %d, want 200", resp.StatusCode)
+	}
+	if !strings.Contains(string(body), "event: message") || !strings.Contains(string(body), "id: 7") {
+		t.Errorf("SSE body not delivered verbatim: %q", body)
+	}
+	// The misconfiguration must be logged, not silent.
+	if got := logBuf.String(); !strings.Contains(got, "ReadsBody plugin that is not a StreamingResponder") {
+		t.Errorf("expected a warning about the ReadsBody+SSE misconfiguration, got logs: %q", got)
+	}
+}

authbridge/authlib/listener/forwardproxy/server.go

@@ -395,6 +395,21 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, isBridge
 				// with per-write flushing. Re-framing (handleStreamingResponse)
 				// would drop the event:/id:/retry: lines that generic SSE
 				// clients (e.g. an MCP Streamable HTTP client) depend on. Fixes #642.
+				//
+				// A plugin that declares ReadsBody (but not WritesBody, and is
+				// not a StreamingResponder) also lands here, and its OnResponse
+				// runs against an empty pctx.ResponseBody: streamPassthrough
+				// forwards the stream without buffering it. We deliberately don't
+				// buffer to satisfy such a plugin — that would reintroduce the
+				// #642 timeout on a live stream. A plugin that must inspect a
+				// streamed body should implement StreamingResponder. Warn
+				// (mirroring the WritesBody fallback above) so the
+				// misconfiguration surfaces instead of the plugin silently seeing
+				// no body. WritesBody is already false in this branch, so
+				// NeedsBody() here implies ReadsBody.
+				if s.OutboundPipeline.NeedsBody() {
+					slog.Warn("forward-proxy: text/event-stream response with a ReadsBody plugin that is not a StreamingResponder — streaming byte-for-byte; its OnResponse will see an empty body (implement StreamingResponder to inspect a streamed body)", "host", r.Host)
+				}
 				s.streamPassthrough(w, r, resp, pctx)
 				return
 			}