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