From d2b6d63d577a98cf393787fe38ec611b17144922 Mon Sep 17 00:00:00 2001 From: Hai Huang Date: Fri, 30 Oct 2026 09:50:47 -0400 Subject: [PATCH] fix: strip hop-by-hop response headers, share SSE normalization, and dedupe the listener helpers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three drift defects from the listener/parser audit (#1567), where one copy got a fix its twins did not: 1. Hop-by-hop response headers reached the client. The forward proxy's four response-header copy loops relayed resp.Header verbatim — requests were stripped (stripForUpstream), responses were not, and a comment on one loop even claimed the stripping happened. Through the TLS bridge an HTTP/1.1 origin's Keep-Alive lands inside an HTTP/2 HEADERS frame, which Node's http2 rejects with ERR_HTTP2_STREAM_ERROR. All four sites now share copyResponseHeaders, which skips the hop-by-hop set plus anything the response's own Connection header names. 2. Only the inference parsers handled a BOM or CR-only line endings. A BOM-prefixed or CR-only MCP/A2A stream matched no data: line, the extension stayed empty, and IBAC treats an unclassified exchange as one that passes by default. normalizeSSE moved to parsercommon.NormalizeSSE and is now used by all six buffered scanners. 3. Request helpers copied between listeners. isEventStream was defined three times and setBody/syncHeaders were inlined in the reverse proxy; all now live in core/listener/httpx (IsEventStream, SetRequestBody, SyncRequestHeaders) with one definition and six call sites. Every fix is pinned by a test that was red on main: hop-by-hop headers absent end-to-end on both the buffered and streaming paths, and BOM/CR-only MCP and A2A streams classifying. Assisted-By: Claude (Anthropic AI) Signed-off-by: Hai Huang --- core/listener/extproc/server.go | 19 +-- core/listener/forwardproxy/hopbyhop_test.go | 92 +++++++++++++++ core/listener/forwardproxy/server.go | 121 ++++++++------------ core/listener/httpx/eventstream.go | 22 ++++ core/listener/httpx/request.go | 56 +++++++++ core/listener/reverseproxy/server.go | 43 +------ core/plugins/a2aparser/plugin.go | 9 +- core/plugins/a2aparser/sse_shapes_test.go | 71 ++++++++++++ core/plugins/inferenceparser/anthropic.go | 7 +- core/plugins/inferenceparser/plugin.go | 42 +------ core/plugins/inferenceparser/responses.go | 2 +- core/plugins/internal/parsercommon/sse.go | 42 +++++++ core/plugins/mcpparser/plugin.go | 5 +- core/plugins/mcpparser/sse_shapes_test.go | 75 ++++++++++++ 14 files changed, 432 insertions(+), 174 deletions(-) create mode 100644 core/listener/forwardproxy/hopbyhop_test.go create mode 100644 core/listener/httpx/eventstream.go create mode 100644 core/listener/httpx/request.go create mode 100644 core/plugins/a2aparser/sse_shapes_test.go create mode 100644 core/plugins/internal/parsercommon/sse.go create mode 100644 core/plugins/mcpparser/sse_shapes_test.go diff --git a/core/listener/extproc/server.go b/core/listener/extproc/server.go index 239ed08ff..d4fa680d7 100644 --- a/core/listener/extproc/server.go +++ b/core/listener/extproc/server.go @@ -912,7 +912,7 @@ func (s *Server) handleResponseBody(ctx context.Context, body []byte, pctx *pipe // unparseable halves — and on the Anthropic dialect the event at risk is message_delta, so // what goes missing is the OUTPUT tally while the response still settles at the prompt-only // floor. - if isEventStream(pctx.ResponseHeaders.Get("Content-Type")) { + if httpx.IsEventStream(pctx.ResponseHeaders.Get("Content-Type")) { // BOUNDED LIKE THE OTHER ARM. The carry itself is capped at one frame's worth, but the join // below is carry + THIS MESSAGE, and nothing else caps the message: the maxBodySize check in // Process covers the REQUEST body only. appendBoundedBody puts both under the same ceiling @@ -1396,7 +1396,7 @@ func countHeader(headers *corev3.HeaderMap, key string) int { // counts of every event that DID complete were dispatched when their message arrived, and the // terminal marker below is what turns them into a settled figure. func dispatchTerminalFrame(ctx context.Context, p *pipeline.Holder, pctx *pipeline.Context) { - if isEventStream(pctx.ResponseHeaders.Get("Content-Type")) { + if httpx.IsEventStream(pctx.ResponseHeaders.Get("Content-Type")) { _ = p.RunResponseFrame(ctx, pctx, nil, true) return } @@ -1405,7 +1405,7 @@ func dispatchTerminalFrame(ctx context.Context, p *pipeline.Holder, pctx *pipeli func dispatchBufferedFrames(ctx context.Context, p *pipeline.Holder, pctx *pipeline.Context, last bool) pipeline.Action { contentType := pctx.ResponseHeaders.Get("Content-Type") - if isEventStream(contentType) && len(pctx.ResponseBody) > 0 { + if httpx.IsEventStream(contentType) && len(pctx.ResponseBody) > 0 { // PARSE ONLY WHAT IS COMPLETE, AND KEEP THE REST FOR THE NEXT MESSAGE. sseframe // delivers an unterminated trailing event at EOF — correct for a stream that really // ended, wrong for one that merely ran out of THIS chunk — so a straddling event would @@ -1448,16 +1448,3 @@ func dispatchBufferedFrames(ctx context.Context, p *pipeline.Holder, pctx *pipel } return p.RunResponseFrame(ctx, pctx, pctx.ResponseBody, true) } - -// isEventStream reports whether a Content-Type header value names the -// SSE media type. Tolerates parameters and ASCII case differences. -// Mirrors the helpers in forwardproxy/reverseproxy. -func isEventStream(contentType string) bool { - if contentType == "" { - return false - } - if idx := strings.IndexByte(contentType, ';'); idx >= 0 { - contentType = contentType[:idx] - } - return strings.EqualFold(strings.TrimSpace(contentType), "text/event-stream") -} diff --git a/core/listener/forwardproxy/hopbyhop_test.go b/core/listener/forwardproxy/hopbyhop_test.go new file mode 100644 index 000000000..ff1ef6570 --- /dev/null +++ b/core/listener/forwardproxy/hopbyhop_test.go @@ -0,0 +1,92 @@ +package forwardproxy + +import ( + "io" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/rossoctl/cortex/core/pipeline" +) + +// The tests in this file pin issue #1567 item 1. Hop-by-hop headers describe ONE +// connection and must not be forwarded. The four response-header copy loops used to +// relay them verbatim — requests were stripped (stripForUpstream) but responses were +// not, and a comment on one of the loops even claimed the stripping happened. Through +// the TLS bridge an HTTP/1.1 origin's Keep-Alive reaches an HTTP/2 client inside a +// HEADERS frame, which Node's http2 rejects with ERR_HTTP2_STREAM_ERROR. + +// hopByHopUpstream answers every request with a body carrying the headers under test. +func hopByHopUpstream(t *testing.T, contentType string) *httptest.Server { + t.Helper() + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", contentType) + // A hop-by-hop header by name... + w.Header().Set("Keep-Alive", "timeout=5") + // ...and one made hop-by-hop by being named in Connection. + w.Header().Set("Connection", "X-Hop-Extra") + w.Header().Set("X-Hop-Extra", "hop value") + // And one that must survive. + w.Header().Set("X-Normal", "kept") + w.WriteHeader(http.StatusOK) + _, _ = io.WriteString(w, "{}") + })) + t.Cleanup(upstream.Close) + return upstream +} + +// proxyGet fetches url through a plain forward proxy (no plugins) and returns the +// response the proxy handed the client, body consumed. +func proxyGet(t *testing.T, url string) *http.Response { + t.Helper() + pipe, err := pipeline.New(nil) + if err != nil { + t.Fatalf("pipeline.New: %v", err) + } + srv, err := NewServer(pipeline.NewHolder(pipe), nil, nil) + if err != nil { + t.Fatalf("NewServer: %v", err) + } + proxy := httptest.NewServer(srv.Handler()) + t.Cleanup(proxy.Close) + client := &http.Client{ + Transport: &http.Transport{Proxy: http.ProxyURL(mustParseURL(proxy.URL))}, + Timeout: 10 * time.Second, + } + resp, err := client.Get(url + "/data") + if err != nil { + t.Fatalf("GET through the proxy: %v", err) + } + t.Cleanup(func() { _ = resp.Body.Close() }) + _, _ = io.Copy(io.Discard, resp.Body) + return resp +} + +func assertNoHopByHop(t *testing.T, resp *http.Response) { + t.Helper() + if v := resp.Header.Get("Keep-Alive"); v != "" { + t.Errorf("Keep-Alive = %q, want absent — hop-by-hop headers describe one connection "+ + "and must not be forwarded to the client", v) + } + if v := resp.Header.Get("X-Hop-Extra"); v != "" { + t.Errorf("X-Hop-Extra = %q, want absent — a header named in Connection is hop-by-hop "+ + "for this connection, whatever its usual meaning", v) + } + if v := resp.Header.Get("X-Normal"); v != "kept" { + t.Errorf("X-Normal = %q, want %q — end-to-end headers must still be forwarded", v, "kept") + } +} + +func TestForwardProxy_HopByHopResponseHeadersStripped(t *testing.T) { + resp := proxyGet(t, hopByHopUpstream(t, "application/json").URL) + assertNoHopByHop(t, resp) +} + +// The streaming path has its own copy loop (streamPassthrough); the same rule must +// hold there. No StreamingResponder is configured, so the passthrough loop is the +// one that runs. +func TestForwardProxy_HopByHopResponseHeadersStrippedOnStream(t *testing.T) { + resp := proxyGet(t, hopByHopUpstream(t, "text/event-stream").URL) + assertNoHopByHop(t, resp) +} diff --git a/core/listener/forwardproxy/server.go b/core/listener/forwardproxy/server.go index 28ab012c0..e29a5dc89 100644 --- a/core/listener/forwardproxy/server.go +++ b/core/listener/forwardproxy/server.go @@ -658,13 +658,13 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne // x-api-key). Content-Length / Content-Encoding are managed by the // body-rewrite block below and the transport, so leave them untouched. // Mirrors reverseproxy's forwarded-request header sync. - syncHeaders(r, pctx.Headers) + httpx.SyncRequestHeaders(r, pctx.Headers) // If a WritesRequestBody plugin rewrote pctx.Body, ship the new bytes // upstream and clear Content-Encoding (see forwardproxy response // path for the rationale). if pctx.BodyMutated() { - setBody(r, pctx.Body) + httpx.SetRequestBody(r, pctx.Body) } // The pipeline this request runs on for the rest of the handler: every @@ -851,7 +851,7 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne // `{"detail":"Method Not Allowed"}` arrived as the literal string "Unknown error" // once isKnownMislabeledSSE started matching every status, not just success. mislabeled := resp.StatusCode/100 == 2 && isKnownMislabeledSSE(r.Host, r.URL.Path) - if (isEventStream(resp.Header.Get("Content-Type")) || mislabeled) && resp.Body != nil { + if (httpx.IsEventStream(resp.Header.Get("Content-Type")) || mislabeled) && resp.Body != nil { if pinned.WritesResponseBody() { // A response mutator needs the whole response to rewrite it, so // it can't stream — fall back to the buffered path with a warning. @@ -956,18 +956,14 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne } - for key, values := range resp.Header { - for _, value := range values { - w.Header().Add(key, value) - } - } + copyResponseHeaders(w, resp) w.WriteHeader(resp.StatusCode) // A skipped response of unknown length — an event stream, a long poll — is relayed as // it arrives, as httputil.ReverseProxy does; buffered, its first bytes would wait for // the next 4KB. skip_hosts never sent one here, but self-traffic does: OpenCode's TUI // follows its service's event stream through this path. var dst io.Writer = w - if f, ok := w.(http.Flusher); ok && skipped && (resp.ContentLength == -1 || isEventStream(resp.Header.Get("Content-Type"))) { + if f, ok := w.(http.Flusher); ok && skipped && (resp.ContentLength == -1 || httpx.IsEventStream(resp.Header.Get("Content-Type"))) { f.Flush() dst = flushWriter{w: w, f: f} } @@ -1893,31 +1889,50 @@ func (s *Server) attributeForRecording(pctx *pipeline.Context, clientHeaders htt // recordOutboundRequestEvent records pctx's request row under sid. Its pair is // recordOutboundResponseEvent, which files under pctx.OutboundSessionID, so the // caller pins that to sid first. -// syncHeaders makes r's headers the pipeline's: pctx.Headers started as a clone of -// r.Header, so a plugin's set, replace and delete on it are the upstream-facing -// header set. Content-Length and Content-Encoding are left to setBody and the -// transport. -func syncHeaders(r *http.Request, h http.Header) { - skip := func(k string) bool { - return strings.EqualFold(k, "Content-Length") || strings.EqualFold(k, "Content-Encoding") - } - for k := range r.Header { - if skip(k) { - continue - } - if _, ok := h[k]; !ok { - r.Header.Del(k) // plugin removed it + +// hopByHopHeaders is the response-side twin of the request-side list stripForUpstream +// deletes. RFC 9110 §7.6.3: these describe one connection, not the resource, and must +// not be forwarded in either direction. +var hopByHopHeaders = map[string]bool{ + "Connection": true, + "Keep-Alive": true, + "Proxy-Authenticate": true, + "Proxy-Authorization": true, + "Proxy-Connection": true, + "Te": true, + "Trailer": true, + "Transfer-Encoding": true, + "Upgrade": true, +} + +// copyResponseHeaders forwards resp's end-to-end headers to w, skipping the hop-by-hop +// set — including anything the response's own Connection header names as hop-by-hop for +// this connection (#1567). The four response paths used to copy resp.Header verbatim +// while requests were stripped, so an HTTP/1.1 origin's Keep-Alive reached the client; +// through the TLS bridge that lands inside an HTTP/2 HEADERS frame, which clients like +// Node's http2 reject outright. +// +// Content-Length is NOT skipped here: the buffered paths forward a body that matches it, +// and the streaming paths delete it themselves after this call. +func copyResponseHeaders(w http.ResponseWriter, resp *http.Response) { + // The Connection header itself is in the skip list; its values are read first + // because a response can demote a normally end-to-end header to hop-by-hop by + // naming it ("Connection: X-Hop-Extra" makes X-Hop-Extra per-connection). + named := map[string]bool{} + for _, v := range resp.Header.Values("Connection") { + for _, name := range strings.Split(v, ",") { + if name = strings.TrimSpace(name); name != "" { + named[http.CanonicalHeaderKey(name)] = true + } } } - for k, vv := range h { - if skip(k) { + for key, values := range resp.Header { + if hopByHopHeaders[http.CanonicalHeaderKey(key)] || named[http.CanonicalHeaderKey(key)] { continue } - if len(vv) == 0 { - r.Header.Del(k) // pctx.Headers[k] = nil is a delete, same as Del(k) - continue + for _, value := range values { + w.Header().Add(key, value) } - r.Header[k] = append([]string(nil), vv...) // set / overwrite } } @@ -1979,15 +1994,6 @@ func stripForUpstream(r *http.Request, ownEncoding, keepUpgrade bool) { } } -// setBody gives r the body a plugin wrote, and clears Content-Encoding, since the -// bytes are the plugin's and carry none. -func setBody(r *http.Request, body []byte) { - r.Body = io.NopCloser(bytes.NewReader(body)) - r.ContentLength = int64(len(body)) - r.Header.Set("Content-Length", fmt.Sprintf("%d", len(body))) - r.Header.Del("Content-Encoding") -} - // resendIfAsked gives the outbound pipeline's Resenders a look at each refused answer, // at most pipeline.MaxResends times, and sends the request again when one asks. It // returns the answer to relay: resp itself, or the last send's. @@ -2031,8 +2037,8 @@ func (s *Server) resendIfAsked(r *http.Request, pctx *pipeline.Context, client * s.recordOutboundRequestEvent(tl, pctx, pctx.OutboundSessionID) } again := r.Clone(r.Context()) - syncHeaders(again, pctx.Headers) - setBody(again, pctx.Body) + httpx.SyncRequestHeaders(again, pctx.Headers) + httpx.SetRequestBody(again, pctx.Body) // No keepUpgrade: a resent request is never a forwarded upgrade — the // 101 branch in serveOutbound skips resendIfAsked entirely, and only that // branch leaves the headers unstripped. @@ -2461,21 +2467,6 @@ func classifyBrokenStream(pctx *pipeline.Context, r *http.Request, resp *http.Re return fail, true } -// isEventStream reports whether a Content-Type header value names the -// SSE media type. Content-Type may carry parameters (charset=, boundary=, -// etc.) so we match on the bare type/subtype prefix and tolerate any -// suffix. Case-insensitive per RFC 9110 §8.3.1. -func isEventStream(contentType string) bool { - if contentType == "" { - return false - } - // Strip parameters: "text/event-stream; charset=utf-8" → "text/event-stream". - if idx := strings.IndexByte(contentType, ';'); idx >= 0 { - contentType = contentType[:idx] - } - return strings.EqualFold(strings.TrimSpace(contentType), "text/event-stream") -} - // isKnownMislabeledSSE reports whether host+path is a known endpoint that sends real SSE // wire bytes (event:/data: lines) under the wrong Content-Type. // @@ -2557,12 +2548,8 @@ func (s *Server) handleStreamingResponse(w http.ResponseWriter, r *http.Request, // Forward headers and the streaming status code BEFORE the first // frame is written. Strip Content-Length since we'll be writing - // chunked, and clear hop-by-hop headers as net/http would. - for key, values := range resp.Header { - for _, value := range values { - w.Header().Add(key, value) - } - } + // chunked; copyResponseHeaders clears the hop-by-hop set as net/http would. + copyResponseHeaders(w, resp) w.Header().Del("Content-Length") w.WriteHeader(resp.StatusCode) flusher.Flush() @@ -2676,11 +2663,7 @@ func (s *Server) streamPassthrough(w http.ResponseWriter, r *http.Request, resp // Forward headers + status before the first byte. Drop Content-Length since // we relay an open-ended chunked stream. - for key, values := range resp.Header { - for _, value := range values { - w.Header().Add(key, value) - } - } + copyResponseHeaders(w, resp) w.Header().Del("Content-Length") w.WriteHeader(resp.StatusCode) flusher.Flush() @@ -2793,11 +2776,7 @@ func (s *Server) streamFallbackBuffered(w http.ResponseWriter, r *http.Request, } } s.recordOutboundResponseEvent(pctx, resp.StatusCode, nil) - for key, values := range resp.Header { - for _, value := range values { - w.Header().Add(key, value) - } - } + copyResponseHeaders(w, resp) w.WriteHeader(resp.StatusCode) if _, err := io.Copy(w, resp.Body); err != nil { slog.Debug("response copy error", "host", r.Host, "error", err) diff --git a/core/listener/httpx/eventstream.go b/core/listener/httpx/eventstream.go new file mode 100644 index 000000000..d20e699d7 --- /dev/null +++ b/core/listener/httpx/eventstream.go @@ -0,0 +1,22 @@ +package httpx + +import "strings" + +// IsEventStream reports whether a Content-Type header value names the SSE media +// type. Tolerates parameters ("text/event-stream; charset=utf-8") and ASCII case +// differences. +// +// ONE definition for all three listeners (#1567): the forward proxy, reverse proxy +// and ext_proc listeners each carried their own copy, and a fix to one — the +// parameter tolerance — reached the others only by re-editing every copy. The +// copies were identical only because someone kept them that way. +func IsEventStream(contentType string) bool { + if contentType == "" { + return false + } + // Strip parameters: "text/event-stream; charset=utf-8" → "text/event-stream". + if idx := strings.IndexByte(contentType, ';'); idx >= 0 { + contentType = contentType[:idx] + } + return strings.EqualFold(strings.TrimSpace(contentType), "text/event-stream") +} diff --git a/core/listener/httpx/request.go b/core/listener/httpx/request.go new file mode 100644 index 000000000..75cd6f104 --- /dev/null +++ b/core/listener/httpx/request.go @@ -0,0 +1,56 @@ +package httpx + +import ( + "bytes" + "fmt" + "io" + "net/http" + "strings" +) + +// SetRequestBody replaces r's body with body and fixes the headers that describe it: +// Content-Length to the new size, Content-Encoding dropped (a rewritten body is +// plaintext — the caller decompressed or rebuilt it). +// +// ONE definition for every listener (#1567): the forward proxy carried this as setBody +// and the reverse proxy inlined the same four lines, which is the copy pattern that +// let every other duplicated helper drift. +func SetRequestBody(r *http.Request, body []byte) { + r.Body = io.NopCloser(bytes.NewReader(body)) + r.ContentLength = int64(len(body)) + r.Header.Set("Content-Length", fmt.Sprintf("%d", len(body))) + r.Header.Del("Content-Encoding") +} + +// SyncRequestHeaders propagates every header mutation a pipeline made back onto the +// request that will be sent: h started as a clone of r.Header, so entries missing from +// h were deleted by a plugin, empty entries are deletes, and the rest are set / +// overwritten. Content-Length and Content-Encoding are skipped — the body-rewrite +// path and the transport own those. +// +// Shared for the same reason as SetRequestBody: the forward proxy's syncHeaders and +// the reverse proxy's inline copy of it were identical only because someone kept +// them that way. +func SyncRequestHeaders(r *http.Request, h http.Header) { + skip := func(k string) bool { + return strings.EqualFold(k, "Content-Length") || strings.EqualFold(k, "Content-Encoding") + } + for k := range r.Header { + if skip(k) { + continue + } + if _, ok := h[k]; !ok { + r.Header.Del(k) // plugin removed it + } + } + for k, vv := range h { + if skip(k) { + continue + } + if len(vv) == 0 { + r.Header.Del(k) // h[k] = nil is a delete, same as Del(k) + continue + } + r.Header[k] = append([]string(nil), vv...) // set / overwrite + } +} diff --git a/core/listener/reverseproxy/server.go b/core/listener/reverseproxy/server.go index fb0ce5a2d..40941b372 100644 --- a/core/listener/reverseproxy/server.go +++ b/core/listener/reverseproxy/server.go @@ -15,7 +15,6 @@ import ( "net/http/httputil" "net/url" "strconv" - "strings" "sync" "time" @@ -410,10 +409,7 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) { // the backend and clear Content-Encoding (same rationale as the // response path — plugin may have decompressed). if pctx.BodyMutated() { - r.Body = io.NopCloser(bytes.NewReader(pctx.Body)) - r.ContentLength = int64(len(pctx.Body)) - r.Header.Set("Content-Length", fmt.Sprintf("%d", len(pctx.Body))) - r.Header.Del("Content-Encoding") + httpx.SetRequestBody(r, pctx.Body) } // Propagate every header mutation the inbound pipeline made to the forwarded @@ -422,27 +418,7 @@ func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) { // 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 strings.EqualFold(k, "Content-Length") || strings.EqualFold(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 - } - if len(vv) == 0 { - r.Header.Del(k) // pctx.Headers[k] = nil is a delete, same as Del(k) - continue - } - r.Header[k] = append([]string(nil), vv...) // set / overwrite - } + httpx.SyncRequestHeaders(r, pctx.Headers) // Record the inbound request event whenever there is something // observable: an A2A conversation, plugin invocations, or plugin-public @@ -535,7 +511,7 @@ func (s *Server) modifyResponse(resp *http.Response) error { // WritesResponseBody is incompatible with streaming (we can't rewrite a // body we've already started forwarding) — fall back to buffered // with a warning. - if isEventStream(resp.Header.Get("Content-Type")) && + if httpx.IsEventStream(resp.Header.Get("Content-Type")) && pinned.HasStreamingResponders() && resp.Body != nil { if pinned.WritesResponseBody() { @@ -1052,19 +1028,6 @@ func (b *streamingResponseBody) Close() error { return b.upstream.Close() } -// isEventStream reports whether a Content-Type header value names the -// SSE media type. Tolerates parameters and ASCII case differences. -// Mirrors the forwardproxy helper of the same name. -func isEventStream(contentType string) bool { - if contentType == "" { - return false - } - if idx := strings.IndexByte(contentType, ';'); idx >= 0 { - contentType = contentType[:idx] - } - return strings.EqualFold(strings.TrimSpace(contentType), "text/event-stream") -} - // inboundSessionID returns the bucket ID for an inbound event. Mirrors // extproc's inboundSessionID: trusts the A2A-stated contextId when // non-empty, otherwise routes to DefaultSessionID. Does NOT fall back diff --git a/core/plugins/a2aparser/plugin.go b/core/plugins/a2aparser/plugin.go index 9e2047702..4e63a2a7a 100644 --- a/core/plugins/a2aparser/plugin.go +++ b/core/plugins/a2aparser/plugin.go @@ -320,8 +320,12 @@ func extractSendResponse(body []byte, ext *pipeline.A2AExtension) bool { // extractStreamResponse handles message/stream SSE responses (the // buffered path). For each "data:" line it extracts a single event // and folds it into the extension via extractStreamEvent. +// +// parsercommon.NormalizeSSE first: a leading BOM or CR-only line endings are +// wire-legal SSE, and a raw LF split reads such a body as one unparseable +// line — leaving the extension empty and the exchange unclassified (#1567). func extractStreamResponse(body []byte, ext *pipeline.A2AExtension) { - for _, line := range bytes.Split(body, []byte("\n")) { + for _, line := range bytes.Split(parsercommon.NormalizeSSE(body), []byte("\n")) { line = bytes.TrimSpace(line) if !bytes.HasPrefix(line, []byte("data:")) { continue @@ -399,7 +403,8 @@ func extractSessionID(body []byte) string { return sid } // SSE format: scan "data:" lines for the first event that carries a session ID. - for _, line := range bytes.Split(body, []byte("\n")) { + // Normalized for the same reason as extractStreamResponse above. + for _, line := range bytes.Split(parsercommon.NormalizeSSE(body), []byte("\n")) { line = bytes.TrimSpace(line) if !bytes.HasPrefix(line, []byte("data:")) { continue diff --git a/core/plugins/a2aparser/sse_shapes_test.go b/core/plugins/a2aparser/sse_shapes_test.go new file mode 100644 index 000000000..7b479f594 --- /dev/null +++ b/core/plugins/a2aparser/sse_shapes_test.go @@ -0,0 +1,71 @@ +package a2aparser + +import ( + "context" + "testing" + + "github.com/rossoctl/cortex/core/pipeline" +) + +// The two tests in this file pin issue #1567 item 2 for A2A. The event-stream format +// allows a leading UTF-8 BOM and CR-only line endings, and the buffered message/stream +// scanner used to split on LF over the raw body — so a stream shaped that way matched +// no "data:" line, the A2A extension stayed empty, and IBAC treated the exchange as +// unclassified, which passes by default. inferenceparser already normalizes these +// shapes (normalizeSSE); these tests hold the A2A side to the same bar. + +// bufferedStreamResponse builds a message/stream body in the wire-legal shape named +// by bom and crOnly. The body starts directly with a data: line when bom is set — +// that is the case the BOM breaks: a BOM before an event: or comment line leaves the +// data: lines untouched, while a BOM before data: defeats the prefix check on the +// first line. +func bufferedStreamResponse(bom, crOnly bool) []byte { + eol := "\n" + if crOnly { + eol = "\r" + } + body := `data: {"result":{"kind":"artifact-update","taskId":"task-1","artifact":{"parts":[{"kind":"text","text":"Hello world"}]}}}` + eol + eol + body += `data: {"result":{"kind":"status-update","final":true,"status":{"state":"completed"}}}` + eol + eol + if bom { + body = "\xef\xbb\xbf" + body + } + return []byte(body) +} + +func TestA2AParser_OnResponse_BufferedSSELeadingBOM(t *testing.T) { + p := NewA2AParser() + pctx := &pipeline.Context{ + Extensions: pipeline.Extensions{A2A: &pipeline.A2AExtension{Method: "message/stream"}}, + ResponseBody: bufferedStreamResponse(true, false), + } + if action := p.OnResponse(context.Background(), pctx); action.Type != pipeline.Continue { + t.Fatalf("action = %v, want Continue", action.Type) + } + ext := pctx.Extensions.A2A + if ext.Artifact != "Hello world" { + t.Errorf("Artifact = %q, want %q — a leading BOM is wire-legal SSE and must not "+ + "hide the events from the buffered scanner", ext.Artifact, "Hello world") + } + if ext.FinalStatus != "completed" { + t.Errorf("FinalStatus = %q, want %q", ext.FinalStatus, "completed") + } +} + +func TestA2AParser_OnResponse_BufferedSSECROnlyLineEndings(t *testing.T) { + p := NewA2AParser() + pctx := &pipeline.Context{ + Extensions: pipeline.Extensions{A2A: &pipeline.A2AExtension{Method: "message/stream"}}, + ResponseBody: bufferedStreamResponse(false, true), + } + if action := p.OnResponse(context.Background(), pctx); action.Type != pipeline.Continue { + t.Fatalf("action = %v, want Continue", action.Type) + } + ext := pctx.Extensions.A2A + if ext.Artifact != "Hello world" { + t.Errorf("Artifact = %q, want %q — CR-only line endings are wire-legal SSE; a body "+ + "so shaped is ONE line to an LF split and every data: check missed it", ext.Artifact, "Hello world") + } + if ext.FinalStatus != "completed" { + t.Errorf("FinalStatus = %q, want %q", ext.FinalStatus, "completed") + } +} diff --git a/core/plugins/inferenceparser/anthropic.go b/core/plugins/inferenceparser/anthropic.go index 172b6c415..2302f6416 100644 --- a/core/plugins/inferenceparser/anthropic.go +++ b/core/plugins/inferenceparser/anthropic.go @@ -497,9 +497,10 @@ func mergeAnthropicUsageMaxSeen(state *inferenceStreamState, incoming parsercomm func parseAnthropicSSE(body []byte, ext *pipeline.InferenceExtension) { markStreamedResponse(body, ext) state := &inferenceStreamState{} - // normalizeSSE for the reason given where it is defined: a BOM or CR-only line endings are - // wire-legal, and splitting on LF alone reads such a body as one unparseable line. - for _, line := range bytes.Split(normalizeSSE(body), []byte("\n")) { + // parsercommon.NormalizeSSE for the reason given where it is defined: a BOM or CR-only + // line endings are wire-legal, and splitting on LF alone reads such a body as one + // unparseable line. + for _, line := range bytes.Split(parsercommon.NormalizeSSE(body), []byte("\n")) { line = bytes.TrimSpace(line) if !bytes.HasPrefix(line, []byte("data:")) { continue diff --git a/core/plugins/inferenceparser/plugin.go b/core/plugins/inferenceparser/plugin.go index 5afd8c194..2f1e378ff 100644 --- a/core/plugins/inferenceparser/plugin.go +++ b/core/plugins/inferenceparser/plugin.go @@ -585,7 +585,7 @@ func (p *InferenceParser) OnResponseFrame(_ context.Context, pctx *pipeline.Cont // parses nothing, and feeding a bare payload to the SSE reader parses nothing — each of // them silently, which is why the shapes are told apart here instead. func carriesSSEFraming(frame []byte) bool { - frame = normalizeSSE(frame) + frame = parsercommon.NormalizeSSE(frame) if bytes.HasPrefix(bytes.TrimLeft(frame, " \t\r\n"), []byte("data:")) { return true } @@ -619,44 +619,6 @@ func markStreamedResponse(body []byte, ext *pipeline.InferenceExtension) { } } -// normalizeSSE puts a buffered SSE body into the one shape the parsers below read: no leading -// byte-order mark, and LF line endings. -// -// BOTH HALVES ARE WIRE-LEGAL AND BOTH WERE MISSED. The event-stream format allows CRLF, LF or -// CR as the line terminator and requires a decoder to strip one leading BOM — while detection -// looked for "data:" after trimming " \t\r\n" (which does not include a BOM) or for a literal -// "\ndata:" (which a CR-only stream never contains), and the parsers split on LF alone. So a -// BOM-prefixed body, or a CR-only body whose first field is not `data:`, fell through to the -// JSON arm: unmarshalling failed, usage stayed unset, and the request kept only whatever a cost -// header happened to say. With no cost header there was nothing left to settle from. -// -// USED BY DETECTION AND PARSING, deliberately the same function. Fixing only the detector would -// route such a body to a parser that still cannot read it, which trades a silent miss for a -// silent empty parse. -// -// Allocation-free for the overwhelming majority: a body with no BOM and no CR is returned as -// it came. -func normalizeSSE(body []byte) []byte { - body = bytes.TrimPrefix(body, []byte("\xef\xbb\xbf")) - if !bytes.ContainsRune(body, '\r') { - return body - } - // WHAT IS LOAD-BEARING HERE, MEASURED RATHER THAN ASSUMED. The BOM strip and the CR-only - // replacement are: a CR-only body is ONE line to bytes.Split(body, "\n"), so every `data:` - // prefix check misses it and the usage is lost, and a leading BOM defeats the same check on the - // first line. Both have rows in sse_shapes_test.go that fail without them. - // - // THE CRLF-FIRST ORDERING IS NOT. Both parsers are line-based and TrimSpace each line, so a - // trailing \r is already handled — dropping this replacement entirely, leaving CR->LF alone, - // changes no observable behaviour in either dialect (mutation-tested). It stays as defence in - // depth for a future consumer that splits on a blank line rather than scanning lines, where - // CRLF collapsed by CR->LF alone WOULD become a blank line between every field; the PR body's - // claim that CRLF was a fix is wrong, and the CRLF rows in the tests are exercise rather than - // discrimination. - out := bytes.ReplaceAll(body, []byte("\r\n"), []byte("\n")) - return bytes.ReplaceAll(out, []byte("\r"), []byte("\n")) -} - // foldResponseFrame folds one streamed frame into the running state via the dialect the // endpoint speaks. Extracted so the mid-stream and terminal call sites cannot drift apart // on which parser a path gets. @@ -788,7 +750,7 @@ func parseInferenceJSON(body []byte, ext *pipeline.InferenceExtension) { func parseInferenceSSE(body []byte, ext *pipeline.InferenceExtension) { markStreamedResponse(body, ext) state := &inferenceStreamState{} - for _, line := range bytes.Split(normalizeSSE(body), []byte("\n")) { + for _, line := range bytes.Split(parsercommon.NormalizeSSE(body), []byte("\n")) { line = bytes.TrimSpace(line) if !bytes.HasPrefix(line, []byte("data:")) { continue diff --git a/core/plugins/inferenceparser/responses.go b/core/plugins/inferenceparser/responses.go index 1408ba4e1..effc70373 100644 --- a/core/plugins/inferenceparser/responses.go +++ b/core/plugins/inferenceparser/responses.go @@ -563,7 +563,7 @@ func foldResponsesFrame(frame []byte, state *inferenceStreamState, ext *pipeline func parseResponsesSSE(body []byte, ext *pipeline.InferenceExtension) { markStreamedResponse(body, ext) state := &inferenceStreamState{} - for _, line := range bytes.Split(normalizeSSE(body), []byte("\n")) { + for _, line := range bytes.Split(parsercommon.NormalizeSSE(body), []byte("\n")) { line = bytes.TrimSpace(line) if !bytes.HasPrefix(line, []byte("data:")) { continue diff --git a/core/plugins/internal/parsercommon/sse.go b/core/plugins/internal/parsercommon/sse.go new file mode 100644 index 000000000..bf96aab04 --- /dev/null +++ b/core/plugins/internal/parsercommon/sse.go @@ -0,0 +1,42 @@ +package parsercommon + +import "bytes" + +// NormalizeSSE puts a buffered SSE body into the one shape the line scanners read: no +// leading byte-order mark, and LF line endings. +// +// BOTH HALVES ARE WIRE-LEGAL AND BOTH WERE MISSED. The event-stream format allows CRLF, +// LF or CR as the line terminator and requires a decoder to strip one leading BOM — +// while detection looked for "data:" after trimming " \t\r\n" (which does not include a +// BOM) or for a literal "\ndata:" (which a CR-only stream never contains), and the +// parsers split on LF alone. So a BOM-prefixed body, or a CR-only body whose first field +// is not `data:`, matched no data: line at all. +// +// SHARED BECAUSE THE MISS WAS A COPY PATTERN (#1567): inference-parser had this fix and +// the a2a and mcp parsers did not — a BOM-prefixed or CR-only MCP/A2A stream left the +// extension empty, and IBAC treats an unclassified exchange as one that passes by +// default. Detection and parsing use the same function deliberately: fixing only a +// detector would route such a body to a parser that still cannot read it, which trades +// a silent miss for a silent empty parse. +// +// Allocation-free for the overwhelming majority: a body with no BOM and no CR is +// returned as it came. +func NormalizeSSE(body []byte) []byte { + body = bytes.TrimPrefix(body, []byte("\xef\xbb\xbf")) + if !bytes.ContainsRune(body, '\r') { + return body + } + // WHAT IS LOAD-BEARING HERE, MEASURED RATHER THAN ASSUMED. The BOM strip and the + // CR-only replacement are: a CR-only body is ONE line to bytes.Split(body, "\n"), so + // every `data:` prefix check misses it, and a leading BOM defeats the same check on + // the first line. + // + // THE CRLF-FIRST ORDERING IS NOT. The scanners are line-based and TrimSpace each + // line, so a trailing \r is already handled — dropping this replacement entirely, + // leaving CR->LF alone, changes no observable behaviour (mutation-tested on the + // inference parsers). It stays as defence in depth for a future consumer that splits + // on a blank line rather than scanning lines, where CRLF collapsed by CR->LF alone + // WOULD become a blank line between every field. + out := bytes.ReplaceAll(body, []byte("\r\n"), []byte("\n")) + return bytes.ReplaceAll(out, []byte("\r"), []byte("\n")) +} diff --git a/core/plugins/mcpparser/plugin.go b/core/plugins/mcpparser/plugin.go index 31d20f666..4453dfd7f 100644 --- a/core/plugins/mcpparser/plugin.go +++ b/core/plugins/mcpparser/plugin.go @@ -475,7 +475,10 @@ func parseMCPResponse(body []byte) (jsonRPCResponse, bool) { if json.Unmarshal(body, &rpc) == nil && (rpc.Result != nil || rpc.Error != nil) { return rpc, true } - for _, line := range bytes.Split(body, []byte("\n")) { + // Normalized before the LF split: a leading BOM or CR-only line endings are + // wire-legal SSE, and a raw split reads such a body as one unparseable line — + // leaving the extension empty and the exchange unclassified (#1567). + for _, line := range bytes.Split(parsercommon.NormalizeSSE(body), []byte("\n")) { line = bytes.TrimSpace(line) if !bytes.HasPrefix(line, []byte("data:")) { continue diff --git a/core/plugins/mcpparser/sse_shapes_test.go b/core/plugins/mcpparser/sse_shapes_test.go new file mode 100644 index 000000000..d5eee8149 --- /dev/null +++ b/core/plugins/mcpparser/sse_shapes_test.go @@ -0,0 +1,75 @@ +package mcpparser + +import ( + "context" + "testing" + + "github.com/rossoctl/cortex/core/pipeline" +) + +// The two tests in this file pin issue #1567 item 2 for MCP. The event-stream format +// allows a leading UTF-8 BOM and CR-only line endings, and the buffered response +// scanner used to split on LF over the raw body — so a Streamable-HTTP response +// shaped that way matched no "data:" line, the MCP extension stayed empty, and IBAC +// treated the exchange as unclassified, which passes by default. inferenceparser +// already normalizes these shapes (normalizeSSE); these tests hold MCP to the same +// bar. + +// mcpStreamBody builds a Streamable-HTTP response in the wire-legal shape named by +// bom and crOnly. The body starts directly with a data: line when bom is set — that +// is the case the BOM breaks (see the a2a fixture for the reasoning). +func mcpStreamBody(bom, crOnly bool) []byte { + eol := "\n" + if crOnly { + eol = "\r" + } + body := `data: {"jsonrpc":"2.0","id":1,"result":{"tools":[{"name":"get_weather"}]}}` + eol + eol + // A SECOND frame that is not a response — a notification. One data: frame + // alone happens to survive a raw LF split (the whole body is one line that + // starts with "data:", and TrimSpace eats the trailing CR), and under a BOM + // the LATER frames still parse, so a second result frame would let both + // shapes pass for the wrong reason. A notification makes the first frame the + // only parseable one, which is the failure shape that matters. + body += `data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progress":50}}` + eol + eol + if bom { + body = "\xef\xbb\xbf" + body + } + return []byte(body) +} + +func TestMCPParser_OnResponse_SSELeadingBOM(t *testing.T) { + p := NewMCPParser() + pctx := &pipeline.Context{ + Extensions: pipeline.Extensions{MCP: &pipeline.MCPExtension{Method: "tools/list"}}, + ResponseBody: mcpStreamBody(true, false), + } + if action := p.OnResponse(context.Background(), pctx); action.Type != pipeline.Continue { + t.Fatalf("action = %v, want Continue", action.Type) + } + if pctx.Extensions.MCP.Result == nil { + t.Fatal("Result should be populated from the SSE data frame — a leading BOM is " + + "wire-legal SSE and must not hide the frame from the buffered scanner") + } + if _, ok := pctx.Extensions.MCP.Result["tools"]; !ok { + t.Error("Result should contain tools from the SSE data") + } +} + +func TestMCPParser_OnResponse_SSECROnlyLineEndings(t *testing.T) { + p := NewMCPParser() + pctx := &pipeline.Context{ + Extensions: pipeline.Extensions{MCP: &pipeline.MCPExtension{Method: "tools/list"}}, + ResponseBody: mcpStreamBody(false, true), + } + if action := p.OnResponse(context.Background(), pctx); action.Type != pipeline.Continue { + t.Fatalf("action = %v, want Continue", action.Type) + } + if pctx.Extensions.MCP.Result == nil { + t.Fatal("Result should be populated from the SSE data frame — CR-only line endings " + + "are wire-legal SSE; a body so shaped is ONE line to an LF split and the data: " + + "check missed it") + } + if _, ok := pctx.Extensions.MCP.Result["tools"]; !ok { + t.Error("Result should contain tools from the SSE data") + } +}