From f67520d98ce463f5032e26b43ff4a1c6fcfc73d7 Mon Sep 17 00:00:00 2001 From: Jakub Hrozek Date: Tue, 28 Jul 2026 00:01:49 +0200 Subject: [PATCH 1/4] Parse SSE responses per event, not per line MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The response filter processed a text/event-stream body line by line, but SSE semantics are per event: within one event all "data:" field values concatenate, and a blank line is structural. Processing lines meant we parsed the body differently than the client does, and every such disagreement is a way to smuggle an unfiltered list past the filter (#5257). Five were reachable: - A payload split across two "data:" lines. Each half failed to decode on its own, so both passed through and the client reassembled the full list. - Two JSON-RPC messages as two "data:" fields of one event. Assembling produced two concatenated values, which json.Unmarshal rejects as trailing data, so carriesResult missed it. - Mixed line terminators. SSE permits CR, LF and CRLF interchangeably within one stream; sniffing one convention for the whole body left entire events buried inside what we treated as a single line. - A leading UTF-8 BOM. Clients strip it per the WHATWG decode algorithm before parsing lines, so our "data:" prefix match failed where theirs succeeded. - A Response carrying both error and result. filterListResponse returned early on any error, and the result went out untouched. The reference client's schema strips the stray error and accepts the result. Scan (line, terminator) pairs and group them into events, assemble each event's data payload, then decode and filter once. Every line is re-emitted with the terminator it was scanned with, so the body is reproduced structurally and the old separator reconciliation — which wrote back at most one separator however many it dropped, collapsing multi-event bodies — is deleted rather than patched. A frame that fails closed now carries a correlatable JSON-RPC error instead of an empty data buffer, which per WHATWG is never dispatched and left the client hanging on its own timeout. At the SSE framing layer, every payload a strict client can parse and correlate as this request's Response is now assembled and decoded before filtering. The residual pass-through paths carry only payloads no strict parser accepts: a split inside a string literal leaves a raw control character in a JSON string, which the client's own parser rejects. So the gap is closed by construction rather than by luck. The per-method result-shape fallbacks in filterToolsResponse and filterFindToolResponse are a separate matter, untouched here. Refs #6037 Co-Authored-By: Claude Opus 5 --- pkg/authz/response_filter.go | 302 +++++++++----- pkg/authz/response_filter_test.go | 663 +++++++++++++++++++++++++----- 2 files changed, 765 insertions(+), 200 deletions(-) diff --git a/pkg/authz/response_filter.go b/pkg/authz/response_filter.go index 4d41ca1dcc..fb83b8f54b 100644 --- a/pkg/authz/response_filter.go +++ b/pkg/authz/response_filter.go @@ -203,62 +203,74 @@ func (rfw *ResponseFilteringWriter) filterAndEncode(response *jsonrpc2.Response) return jsonrpc2.EncodeMessage(filteredResponse) } +// sseLine pairs one scanned SSE line with the terminator that followed it in +// the raw body: "\r\n", "\n", "\r", or nil for a final line with none. SSE +// permits mixing all three within one stream, so each line keeps the +// terminator it actually had rather than the body being forced onto one +// stream-wide convention. +type sseLine struct { + text []byte + term []byte +} + +// processSSEResponse rewrites a text/event-stream body per SSE framing rules +// rather than treating it as an arbitrary sequence of lines: within one +// event, all `data:` field values concatenate (joined by LF, per spec) into a +// single payload, and a blank line is structural — it dispatches the event +// rather than being emittable content in its own right. +// +// Lines are scanned one terminator at a time instead of sniffing a single +// separator convention for the whole body: sniffing disagrees with a client +// that (per spec) may see CR, LF, and CRLF mixed within one stream, and that +// disagreement is exactly the class of bug this rewrite exists to close. +// Every line is re-emitted with the same terminator it was scanned with, so +// the output is structurally identical to the input (minus a leading UTF-8 +// BOM, stripped below); only `data:` payloads are rewritten. func (rfw *ResponseFilteringWriter) processSSEResponse(rawResponse []byte) error { // Note: this routine is adapted from the one in pkg/mcp/tool_filter.go. // I don't see an obvious way to factor out the commonalities, so I'm // duplicating it here, but we should refactor response parsing // respecting mime types to a common routine. - var linesep []byte - if bytes.Contains(rawResponse, []byte("\r\n")) { - linesep = []byte("\r\n") - } else if bytes.Contains(rawResponse, []byte("\n")) { - linesep = []byte("\n") - } else if bytes.Contains(rawResponse, []byte("\r")) { - linesep = []byte("\r") - } else { - // Length only, not the body: rawResponse is the full pre-filter payload - // (e.g. the unfiltered tools/list), and this error is logged upstream. - return fmt.Errorf("unsupported SSE line separator in %d-byte response", len(rawResponse)) - } - - var linesepTotal, linesepCount int - linesepTotal = bytes.Count(rawResponse, linesep) - lines := bytes.Split(rawResponse, linesep) - for _, line := range lines { - if len(line) == 0 { - continue - } - var written bool - if data, ok := bytes.CutPrefix(line, []byte("data:")); ok { - w, err := rfw.writeSSEDataLine(data) - if err != nil { - return err - } - written = w + // A client strips a leading BOM per the WHATWG UTF-8 decode algorithm + // before parsing lines, so strip it here too: otherwise the first line's + // "data:" prefix wouldn't match and the event would pass through + // unfiltered. + rawResponse = bytes.TrimPrefix(rawResponse, []byte("\xEF\xBB\xBF")) + + var outputLines []sseLine + var event []sseLine + for len(rawResponse) > 0 { + idx := bytes.IndexAny(rawResponse, "\r\n") + var line, term []byte + switch { + case idx == -1: + line, term, rawResponse = rawResponse, nil, nil + case rawResponse[idx] == '\r' && idx+1 < len(rawResponse) && rawResponse[idx+1] == '\n': + line, term, rawResponse = rawResponse[:idx], rawResponse[idx:idx+2], rawResponse[idx+2:] + default: + line, term, rawResponse = rawResponse[:idx], rawResponse[idx:idx+1], rawResponse[idx+1:] } - if !written { - _, err := rfw.ResponseWriter.Write(line) - if err != nil { - return fmt.Errorf("%w: %w", errBug, err) - } + if len(line) == 0 { + outputLines = append(outputLines, rfw.resolveSSEEvent(event)...) + outputLines = append(outputLines, sseLine{term: term}) + event = nil + continue } + event = append(event, sseLine{text: line, term: term}) + } + // A trailing event with no closing blank line is never dispatched by a + // spec-compliant client, but the backend did send these bytes, so write + // them (filtered) rather than drop them — without fabricating a + // terminator the backend didn't send. + outputLines = append(outputLines, rfw.resolveSSEEvent(event)...) - _, err := rfw.ResponseWriter.Write(linesep) - if err != nil { + for _, l := range outputLines { + if _, err := rfw.ResponseWriter.Write(l.text); err != nil { return fmt.Errorf("%w: %w", errBug, err) } - linesepCount++ - } - - // This ensures we don't send too few line separators, which might break - // SSE parsing. Note it writes at most one separator back, however many - // were dropped — see writeSSEErrorLine's doc comment for the multi-event - // case that leaves mis-framed. - if linesepCount < linesepTotal { - _, err := rfw.ResponseWriter.Write(linesep) - if err != nil { + if _, err := rfw.ResponseWriter.Write(l.term); err != nil { return fmt.Errorf("%w: %w", errBug, err) } } @@ -266,55 +278,113 @@ func (rfw *ResponseFilteringWriter) processSSEResponse(rawResponse []byte) error return nil } -// writeSSEDataLine classifies a single SSE `data:` payload and writes the -// filtered response when it is a clean Response. written reports whether the -// caller should skip the raw passthrough for this line. A non-nil err means a -// transport write failed and processSSEResponse should abort; a filtering or -// encoding failure is instead written as a substitute error `data:` line via -// writeSSEErrorLine and does not stop processing of the remaining frames. The -// substitute line is reported as written so the caller skips the raw -// passthrough: emitting the unfiltered frame after a failed filter would leak -// the list this filter exists to scrub (#5257). -func (rfw *ResponseFilteringWriter) writeSSEDataLine(data []byte) (written bool, err error) { +// resolveSSEEvent assembles the `data:` payload of one SSE event (the lines +// between two blank lines, exclusive) and returns the lines to emit in its +// place. An event with no `data:` field (e.g. a bare "event: ping") or whose +// assembled payload passes filterSSEEventData unchanged is returned as-is. +func (rfw *ResponseFilteringWriter) resolveSSEEvent(event []sseLine) []sseLine { + if len(event) == 0 { + return nil + } + + var dataValues [][]byte + for _, l := range event { + // A bare "data" line with no colon doesn't match the "data:" prefix + // below, so it's excluded from assembly here, unlike strict WHATWG + // grammar (which treats it as a data field with an empty value). + // That's safe: its only effect on a spec-compliant client's buffer + // would be one extra LF, which is legal JSON whitespace outside a + // string literal and equally fatal to us and the client inside one. + data, ok := bytes.CutPrefix(l.text, []byte("data:")) + if !ok { + continue + } + // Per the WHATWG EventSource grammar, exactly one leading space after + // "data:" is part of the field delimiter, not the payload. Stripping + // more (or a differently-positioned space) would corrupt a payload + // split mid-string-literal across lines. + data, _ = bytes.CutPrefix(data, []byte(" ")) + dataValues = append(dataValues, data) + } + if len(dataValues) == 0 { + return event + } + + assembled := bytes.Join(dataValues, []byte("\n")) + if len(assembled) == 0 { + // A client never dispatches an empty data buffer, so there's nothing + // to classify or fail closed on. + return event + } + replacement := rfw.filterSSEEventData(assembled) + if replacement == nil { + return event + } + + out := make([]sseLine, 0, len(event)) + replaced := false + for _, l := range event { + if _, ok := bytes.CutPrefix(l.text, []byte("data:")); ok { + if replaced { + // Subsequent data: lines are already folded into the + // assembled payload above; only the first line carries it. + continue + } + // jsonrpc2.EncodeMessage and json.Marshal both escape newlines, + // so replacement can never contain a raw line separator and is + // always safe to emit as a single data: line. + out = append(out, sseLine{text: append([]byte("data: "), replacement...), term: l.term}) + replaced = true + continue + } + out = append(out, l) + } + return out +} + +// filterSSEEventData classifies an event's assembled data payload and returns +// the payload to emit in its place. A nil replacement means the event's lines +// pass through unchanged. +func (rfw *ResponseFilteringWriter) filterSSEEventData(data []byte) []byte { message, decodeErr := jsonrpc2.DecodeMessage(data) response, isResponse := message.(*jsonrpc2.Response) switch { case isResponse: filteredData, ferr := rfw.filterAndEncode(response) if ferr != nil { - slog.Warn("failed to filter/encode SSE response; emitting a JSON-RPC error in place of the frame", - "method", rfw.method, "error", ferr) - return true, rfw.writeSSEErrorLine(response.ID, ferr) - } - // The caller writes linesep after this line, so do not append a - // terminator here. A hardcoded "\n" desyncs \r\n streams. - if _, werr := rfw.ResponseWriter.Write([]byte("data: " + string(filteredData))); werr != nil { - return false, fmt.Errorf("%w: %w", errBug, werr) + // errorResponseBody logs ferr itself, so don't repeat it here. + slog.Warn("emitting a JSON-RPC error in place of an SSE frame that failed to filter", + "method", rfw.method) + return rfw.errorResponseBody(response.ID, ferr) } - return true, nil + return filteredData case carriesResult(data): // The frame is not a clean Response but still carries a result. This // covers a non-Response type (a request/notification frame smuggling a // result), a decode error (missing or invalid jsonrpc tag, a response // frame with no or a non-scalar id), and a batch array. All are - // upstream-controlled shapes that would otherwise fall through and leak - // the very list this filter scrubs (#5257). Fail closed and drop the line. - slog.Warn("SSE data line carried a result outside a clean Response frame; dropping as a protocol violation", + // upstream-controlled shapes that would otherwise leak the very list + // this filter scrubs (#5257). Fail closed with an explicit error + // envelope rather than dropping the event silently: an event + // dispatched with an empty data buffer isn't delivered to the + // client's handler at all, leaving it hanging on its own timeout + // (#6037). + slog.Warn("SSE event carried a result outside a clean Response frame; failing closed as a protocol violation", "method", rfw.method) - return true, nil + return rfw.errorResponseBody(rfw.requestID(), + fmt.Errorf("dropped a frame carrying a result outside a clean Response")) case decodeErr != nil: - // Genuinely undecodable and no smuggled result. Pass this line through - // unfiltered (this line only). - slog.Warn("SSE data line could not be decoded as JSON-RPC; passing through unfiltered", + // Genuinely undecodable and no smuggled result. Pass through unfiltered. + slog.Warn("SSE event data could not be decoded as JSON-RPC; passing through unfiltered", "method", rfw.method, "error", decodeErr) - return false, nil + return nil default: - // Genuine non-Response frame (e.g. an interleaved notifications/* message) - // with no result payload. Routine SSE traffic, so log at Debug to keep the - // suspicious branches above from being buried. Pass through this line only. - slog.Debug("SSE data line was not a JSON-RPC Response; passing through unfiltered", + // Genuine non-Response frame (e.g. an interleaved notifications/* + // message) with no result payload. Routine SSE traffic, so log at + // Debug to keep the suspicious branches above from being buried. + slog.Debug("SSE event data was not a JSON-RPC Response; passing through unfiltered", "method", rfw.method) - return false, nil + return nil } } @@ -334,8 +404,29 @@ func requiresResponseFiltering(method string) bool { // result, a shape DecodeMessage rejects, or a batch array) but still carries a // result must not pass through: it would leak the list the filter exists to // scrub. See issue #5257. +// +// The payload may hold more than one top-level JSON value: an SSE event's +// assembled data can concatenate several messages (e.g. a notification and a +// response, one per data: line) into something DecodeMessage rejects as a +// single value. A json.Decoder walks all of them instead of Unmarshal-ing the +// whole payload as one document, so a result-bearing value after the first is +// still caught rather than making the whole payload look undecodable. func carriesResult(data []byte) bool { - trimmed := bytes.TrimSpace(data) + dec := json.NewDecoder(bytes.NewReader(data)) + for { + var value json.RawMessage + if dec.Decode(&value) != nil { + return false // EOF, or malformed JSON with nothing left to check + } + if valueCarriesResult(value) { + return true + } + } +} + +// valueCarriesResult applies carriesResult's check to a single JSON value. +func valueCarriesResult(value json.RawMessage) bool { + trimmed := bytes.TrimSpace(value) if len(trimmed) == 0 { return false } @@ -346,14 +437,24 @@ func carriesResult(data []byte) bool { } return json.Unmarshal(trimmed, &probe) == nil && probe.Result != nil case '[': - var batch []struct { - Result json.RawMessage `json:"result"` - } + // A single level, not recursive: a JSON-RPC batch is a flat array of + // message objects, so that's the whole legitimate surface. Recursing + // through nested arrays is O(d^2) on encoding/json's own nesting cap + // (10000), an easy amplification handle for something no client + // actually flattens. + var batch []json.RawMessage if json.Unmarshal(trimmed, &batch) != nil { return false } - for i := range batch { - if batch[i].Result != nil { + for _, el := range batch { + elTrimmed := bytes.TrimSpace(el) + if len(elTrimmed) == 0 || elTrimmed[0] != '{' { + continue + } + var probe struct { + Result json.RawMessage `json:"result"` + } + if json.Unmarshal(elTrimmed, &probe) == nil && probe.Result != nil { return true } } @@ -380,7 +481,16 @@ func sseCarriesResult(rawResponse []byte) bool { // filterListResponse filters the list response based on authorization policies func (rfw *ResponseFilteringWriter) filterListResponse(response *jsonrpc2.Response) (*jsonrpc2.Response, error) { if response.Error != nil { - // If there's an error in the response, don't filter + // A Response carrying both error and result is a protocol violation: + // DecodeMessage populates both fields as given, and some clients + // (e.g. the reference TS SDK's zod schema) silently strip the + // unexpected "error" key and treat it as a successful result. Fail + // closed rather than passing the smuggled list through under cover + // of the error field. See #5257. + if response.Result != nil { + return nil, fmt.Errorf("response carried both error and result") + } + // If there's an error and no result, don't filter return response, nil } @@ -614,33 +724,17 @@ func (rfw *ResponseFilteringWriter) errorResponseBody(id jsonrpc2.ID, err error) // writeErrorResponse writes an error response for the application/json path. // The SSE path must NOT use this: WriteHeader is a no-op once Flush() has -// committed the headers, which is the case for any real SSE response. Use -// writeSSEErrorLine instead. +// committed the headers, which is the case for any real SSE response. On the +// SSE path a filtering failure instead becomes the replacement payload of the +// event's data: line (see filterSSEEventData), written through +// processSSEResponse's normal line writer alongside every other line — no +// separate write path or WriteHeader call is needed there. func (rfw *ResponseFilteringWriter) writeErrorResponse(id jsonrpc2.ID, err error) error { rfw.ResponseWriter.WriteHeader(http.StatusInternalServerError) _, writeErr := rfw.ResponseWriter.Write(rfw.errorResponseBody(id, err)) return writeErr } -// writeSSEErrorLine writes a filtering failure as an SSE `data:` line, in place -// of the frame that failed. It writes no terminator: processSSEResponse frames -// this line exactly as it frames a successfully filtered one, so the error is -// delivered wherever a filtered response would have been. Bodies carrying more -// than one event are mis-framed by the separator reconciliation either way, -// independently of whether filtering succeeded. -// -// Deliberately no WriteHeader: earlier lines are already on the wire and Flush() -// has committed the headers, so a status change would be dropped — and -// text/event-stream with a bare JSON body is unreadable by an SSE client -// regardless. See issue #6037. -func (rfw *ResponseFilteringWriter) writeSSEErrorLine(id jsonrpc2.ID, err error) error { - _, writeErr := rfw.ResponseWriter.Write([]byte("data: " + string(rfw.errorResponseBody(id, err)))) - if writeErr != nil { - return fmt.Errorf("%w: %w", errBug, writeErr) - } - return nil -} - // requestID recovers the JSON-RPC id of the original request so an error // response written mid-filtering can still be correlated by the client. // Falls back to an empty jsonrpc2.ID (which EncodeMessage omits from the diff --git a/pkg/authz/response_filter_test.go b/pkg/authz/response_filter_test.go index cea960ecc3..d341a65887 100644 --- a/pkg/authz/response_filter_test.go +++ b/pkg/authz/response_filter_test.go @@ -4,6 +4,7 @@ package authz import ( + "bytes" "context" "encoding/json" "errors" @@ -449,7 +450,7 @@ func TestResponseFilteringWriter_NonListOperations(t *testing.T) { Result: json.RawMessage(responseData), } - responseBytes, err := json.Marshal(jsonrpcResponse) + responseBytes, err := jsonrpc2.EncodeMessage(jsonrpcResponse) require.NoError(t, err, "Failed to marshal JSON-RPC response") // Create an HTTP request @@ -867,17 +868,17 @@ func TestOptimizerPassThroughToolsInResponseFilter(t *testing.T) { }) } -// TestResponseFilteringWriter_SSE_PerLineFallthrough is a regression test for -// issue #5257: when an SSE upstream interleaves a non-Response data line (e.g. -// an MCP notification) or an undecodable data line with a real list response, -// the filter previously wrote the entire raw upstream payload and returned, +// TestResponseFilteringWriter_SSE_PerEventFallthrough is a regression test for +// issue #5257: when an SSE upstream interleaves a non-Response event (e.g. an +// MCP notification) or an undecodable event with a real list response, the +// filter previously wrote the entire raw upstream payload and returned, // leaking the unfiltered list past Cedar. It must instead pass only the -// offending line through and continue filtering the rest of the stream. +// offending event through and continue filtering the rest of the stream. // // The same code path runs for every method covered by // requiresResponseFiltering, so each of tools/list, prompts/list, and // resources/list is exercised below. -func TestResponseFilteringWriter_SSE_PerLineFallthrough(t *testing.T) { +func TestResponseFilteringWriter_SSE_PerEventFallthrough(t *testing.T) { t.Parallel() authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ @@ -1021,7 +1022,14 @@ func TestResponseFilteringWriter_SSE_PerLineFallthrough(t *testing.T) { rfw := NewResponseFilteringWriter(rr, authorizer, req, mc.method, nil, nil) rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") - body := strings.Join([]string{plc.line, mc.respLine, ""}, "\n") + // A blank line separates plc.line and mc.respLine into two + // distinct SSE events, matching how a real MCP server frames + // SSE (one JSON-RPC message per event): otherwise, with no + // blank line between them, they'd be two data: fields of the + // SAME event and their values would concatenate, which is not + // what this test means to model. The trailing "" pair + // terminates the second event. + body := strings.Join([]string{plc.line, "", mc.respLine, "", ""}, "\n") _, err = rfw.Write([]byte(body)) require.NoError(t, err) @@ -1059,6 +1067,13 @@ func TestResponseFilteringWriter_SSE_PerLineFallthrough(t *testing.T) { // must not appear in the wire output. assert.NotContains(t, out, `"`+mc.unauthorizedName+`"`, "unfiltered list payload leaked into SSE output") + + // The blank-line event terminator must survive the rewrite: + // processSSEResponse's structural invariant is that every + // line the input had, including blank lines, is written + // exactly once in its original position. + assert.True(t, strings.HasSuffix(out, "\n\n"), + "output must end with a blank-line SSE terminator") }) } } @@ -1067,8 +1082,17 @@ func TestResponseFilteringWriter_SSE_PerLineFallthrough(t *testing.T) { // TestResponseFilteringWriter_SSE_DisguisedResponseFrame is a regression test // for a residual of issue #5257: a frame that carries both a method field (so // jsonrpc2.DecodeMessage classifies it as a request/notification rather than a -// Response) and a result field smuggling a real list. Such a frame must be -// dropped, not passed through, or the unfiltered list leaks past Cedar. +// Response) and a result field smuggling a real list. Such a frame must fail +// closed with an error envelope in its place — not pass the smuggled list +// through, and not silently drop the event's data (which per WHATWG SSE +// semantics leaves the client waiting on a dispatch that never carries a +// usable payload; see #6037). +// +// The envelope must also be one the client can actually correlate: it's +// decoded and checked for a non-nil Error carrying the *request's* id, not +// just a substring match, because an id-less envelope (the zero jsonrpc2.ID, +// which EncodeMessage omits from the wire) reproduces the #6037 hang even +// though an "error" key is present on the wire. func TestResponseFilteringWriter_SSE_DisguisedResponseFrame(t *testing.T) { t.Parallel() @@ -1099,7 +1123,7 @@ func TestResponseFilteringWriter_SSE_DisguisedResponseFrame(t *testing.T) { // Each frame smuggles a tools/list result outside a clean Response, via a // different upstream-controlled shape. DecodeMessage either classifies them // as non-Response or rejects them outright, so the pre-fix code passed them - // through unfiltered. All must be dropped, not leaked. + // through unfiltered. All must fail closed into an error envelope instead. frames := []struct { name string payload string @@ -1109,6 +1133,10 @@ func TestResponseFilteringWriter_SSE_DisguisedResponseFrame(t *testing.T) { {"no id", `data: {"jsonrpc":"2.0","result":{"tools":[{"name":"admin_tool"}]}}`}, {"non-scalar id", `data: {"jsonrpc":"2.0","id":{"x":1},"result":{"tools":[{"name":"admin_tool"}]}}`}, {"batch array", `data: [{"jsonrpc":"2.0","id":1,"result":{"tools":[{"name":"admin_tool"}]}}]`}, + // A batch array whose first element isn't an object: Unmarshal-ing + // the whole array into []struct{Result} would previously fail + // outright, missing the result-bearing element further down. + {"heterogeneous batch array", `data: [1,{"jsonrpc":"2.0","id":1,"result":{"tools":[{"name":"admin_tool"}]}}]`}, } for _, f := range frames { @@ -1116,15 +1144,28 @@ func TestResponseFilteringWriter_SSE_DisguisedResponseFrame(t *testing.T) { t.Run(f.name, func(t *testing.T) { t.Parallel() - req, err := http.NewRequest(http.MethodPost, "/messages", nil) + // Run the real MCP parsing middleware so the request context + // carries a ParsedMCPRequest with a real id, matching production: + // requestID() reads it back out to correlate the error envelope. + reqBody := `{"jsonrpc":"2.0","id":99,"method":"tools/list"}` + req, err := http.NewRequest(http.MethodPost, "/messages", strings.NewReader(reqBody)) require.NoError(t, err) + req.Header.Set("Content-Type", "application/json") req = req.WithContext(auth.WithIdentity(req.Context(), identity)) + var parsedReq *http.Request + mcpparser.ParsingMiddleware(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { + parsedReq = r + })).ServeHTTP(httptest.NewRecorder(), req) + require.NotNil(t, parsedReq) rr := httptest.NewRecorder() - rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) + rfw := NewResponseFilteringWriter(rr, authorizer, parsedReq, string(mcp.MethodToolsList), nil, nil) rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") - body := strings.Join([]string{f.payload, "data: " + string(realResp), ""}, "\n") + // A blank line separates the disguised frame from the genuine + // response into two distinct SSE events, matching how a real MCP + // server frames SSE (one JSON-RPC message per event). + body := strings.Join([]string{f.payload, "", "data: " + string(realResp), "", ""}, "\n") _, err = rfw.Write([]byte(body)) require.NoError(t, err) require.NoError(t, rfw.FlushAndFilter()) @@ -1133,7 +1174,260 @@ func TestResponseFilteringWriter_SSE_DisguisedResponseFrame(t *testing.T) { assert.NotContains(t, out, "admin_tool", "smuggled result leaked past the filter") assert.Contains(t, out, "weather", - "genuine list response on a later line must still be delivered") + "genuine list response in a later event must still be delivered") + + // The disguised frame's event is first on the wire; decode it and + // require an error envelope the client can correlate. TrimSuffix + // guards a CRLF body leaving a trailing \r on the line. + firstLine := strings.TrimSuffix(strings.SplitN(out, "\n", 2)[0], "\r") + payload := strings.TrimPrefix(firstLine, "data: ") + msg, err := jsonrpc2.DecodeMessage([]byte(payload)) + require.NoError(t, err, "the fail-closed envelope must itself be valid JSON-RPC") + resp, ok := msg.(*jsonrpc2.Response) + require.True(t, ok, "the fail-closed envelope must be a clean Response") + require.NotNil(t, resp.Error, "the disguised frame's event must fail closed into an error envelope, not vanish silently") + assert.Equal(t, jsonrpc2.Int64ID(99), resp.ID, + "the error envelope must carry the request's id so the client can correlate it, or the #6037 hang reproduces") + }) + } +} + +// TestResponseFilteringWriter_SSE_ConcatenatedEventBypass is a regression test +// for a #5257-class leak the event-based rewrite itself introduced: two +// data: fields in ONE event (a notification followed by a result-bearing +// frame) assemble, per WHATWG semantics, into "{notif}\n{result}". That +// payload fails jsonrpc2.DecodeMessage (trailing data after the first JSON +// value), and carriesResult must still catch the embedded result — Unmarshal +// on the whole concatenated payload rejects it the same way DecodeMessage +// does, which would otherwise route it to the "pass through unfiltered" +// branch and leak the unauthorized tool straight through. +func TestResponseFilteringWriter_SSE_ConcatenatedEventBypass(t *testing.T) { + t.Parallel() + + authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ + Policies: []string{ + `permit(principal, action == Action::"call_tool", resource == Tool::"weather");`, + }, + EntitiesJSON: `[]`, + }, "") + require.NoError(t, err) + + identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ + Subject: "user1", + Claims: map[string]interface{}{"sub": "user1"}, + }} + + notificationLine := `data: {"jsonrpc":"2.0","method":"notifications/message","params":{}}` + resultLine := `data: {"jsonrpc":"2.0","id":1,"result":{"tools":[{"name":"admin_tool"}]}}` + + req, err := http.NewRequest(http.MethodPost, "/messages", nil) + require.NoError(t, err) + req = req.WithContext(auth.WithIdentity(req.Context(), identity)) + + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) + rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") + + // No blank line between the two data: lines: one event, two fields. + body := strings.Join([]string{notificationLine, resultLine, "", ""}, "\n") + _, err = rfw.Write([]byte(body)) + require.NoError(t, err) + require.NoError(t, rfw.FlushAndFilter()) + + out := rr.Body.String() + assert.NotContains(t, out, "admin_tool", + "a result concatenated after a notification within one event must not bypass the filter") + // NotContains alone would also pass if the event were dropped silently + // (#6037's hang); require the fail-closed envelope actually went out. + assert.Contains(t, out, `"error"`, + "the concatenated event must fail closed into an error envelope, not vanish silently") +} + +// TestResponseFilteringWriter_SSE_MixedSeparatorsBypass is a regression test +// for a #5257-class leak: SSE permits CR, LF, and CRLF mixed within one +// stream, but sniffing a single stream-wide separator (the pre-fix approach) +// disagrees with how a spec-compliant client actually splits the body into +// lines and events. Here the body sniffs as CRLF, so a conformant client's +// LF-terminated lines collapse into one giant "line" for the sniffing filter, +// whose data: prefix strip leaves an undecodable payload that falls through +// unfiltered — while the client itself correctly sees three separate events, +// the third a clean Response carrying the unauthorized tool. +func TestResponseFilteringWriter_SSE_MixedSeparatorsBypass(t *testing.T) { + t.Parallel() + + authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ + Policies: []string{ + `permit(principal, action == Action::"call_tool", resource == Tool::"weather");`, + }, + EntitiesJSON: `[]`, + }, "") + require.NoError(t, err) + + identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ + Subject: "user1", + Claims: map[string]interface{}{"sub": "user1"}, + }} + + req, err := http.NewRequest(http.MethodPost, "/messages", nil) + require.NoError(t, err) + req = req.WithContext(auth.WithIdentity(req.Context(), identity)) + + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) + rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") + + resultJSON, err := json.Marshal(mcp.ListToolsResult{ + Tools: []mcp.Tool{ + {Name: "weather", Description: "Get weather information"}, + {Name: "admin_tool", Description: "Sensitive admin operations"}, + }, + }) + require.NoError(t, err) + resultLine := "data: " + `{"jsonrpc":"2.0","id":1,"result":` + string(resultJSON) + "}" + + body := "data: {\"jsonrpc\":\"2.0\",\"method\":\"notifications/message\",\"params\":{}}\r\n" + + "\r\n" + + "data: x\n" + + "\n" + + resultLine + "\n" + + "\n" + _, err = rfw.Write([]byte(body)) + require.NoError(t, err) + require.NoError(t, rfw.FlushAndFilter()) + + out := rr.Body.String() + assert.NotContains(t, out, "admin_tool", + "mixed line-terminator conventions within one SSE body must not bypass the filter") + // NotContains alone would also pass if the event were dropped rather than + // filtered (#6037's hang); prove it was actually filtered, not eaten. + assert.Contains(t, out, "weather", "the authorized tool must survive filtering") +} + +// TestResponseFilteringWriter_SSE_LeadingBOMBypass is a regression test for a +// #5257-class leak: a client strips a leading UTF-8 BOM per the WHATWG decode +// algorithm before parsing lines, but the filter compared the raw first line +// against "data:". The BOM byte prefix made that comparison fail, so the +// event (and the unauthorized tool inside it) passed through unfiltered. +func TestResponseFilteringWriter_SSE_LeadingBOMBypass(t *testing.T) { + t.Parallel() + + authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ + Policies: []string{ + `permit(principal, action == Action::"call_tool", resource == Tool::"weather");`, + }, + EntitiesJSON: `[]`, + }, "") + require.NoError(t, err) + + identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ + Subject: "user1", + Claims: map[string]interface{}{"sub": "user1"}, + }} + + req, err := http.NewRequest(http.MethodPost, "/messages", nil) + require.NoError(t, err) + req = req.WithContext(auth.WithIdentity(req.Context(), identity)) + + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) + rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") + + resultJSON, err := json.Marshal(mcp.ListToolsResult{ + Tools: []mcp.Tool{ + {Name: "weather", Description: "Get weather information"}, + {Name: "admin_tool", Description: "Sensitive admin operations"}, + }, + }) + require.NoError(t, err) + body := "\xEF\xBB\xBF" + `data: {"jsonrpc":"2.0","id":1,"result":` + string(resultJSON) + "}\n\n" + + _, err = rfw.Write([]byte(body)) + require.NoError(t, err) + require.NoError(t, rfw.FlushAndFilter()) + + out := rr.Body.String() + assert.NotContains(t, out, "admin_tool", + "a leading UTF-8 BOM must not bypass the filter") + assert.Contains(t, out, "weather", "the authorized tool must survive filtering") +} + +// TestResponseFilteringWriter_SSE_ErrorAndResultBypass is a regression test +// for a #5257-class leak: jsonrpc2.DecodeMessage and EncodeMessage both +// populate/re-emit "error" and "result" together on one Response, and +// filterListResponse's `response.Error != nil` check returned the whole +// response, list intact, as soon as an error field was present. The reference +// TS SDK's zod schema silently strips the unknown "error" key rather than +// rejecting the message, so it would accept this as a successful response +// carrying the full unfiltered list. +// +// The error-only row distinguishes "failed closed" from "passed through +// unfiltered": both leave no admin_tool on the wire, so a substring check +// alone can't tell a fix from a regression that fails closed on every +// legitimate upstream error. Only the error *code* does — 500 is our +// envelope, anything else is the upstream's own error surviving untouched. +func TestResponseFilteringWriter_SSE_ErrorAndResultBypass(t *testing.T) { + t.Parallel() + + authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ + Policies: []string{ + `permit(principal, action == Action::"call_tool", resource == Tool::"weather");`, + }, + EntitiesJSON: `[]`, + }, "") + require.NoError(t, err) + + identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ + Subject: "user1", + Claims: map[string]interface{}{"sub": "user1"}, + }} + + testCases := []struct { + name string + body string + wantErrorCode float64 + }{ + { + name: "error and result together fails closed", + body: `data: {"jsonrpc":"2.0","id":1,"error":{"code":1,"message":"x"},` + + `"result":{"tools":[{"name":"admin_tool"}]}}` + "\n\n", + wantErrorCode: float64(mcpparser.CodeInternalError), // our envelope, not the upstream's code 1 + }, + { + name: "error alone passes through unfiltered", + body: `data: {"jsonrpc":"2.0","id":1,"error":{"code":1,"message":"x"}}` + "\n\n", + wantErrorCode: 1, // the upstream's own error, untouched + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + req, err := http.NewRequest(http.MethodPost, "/messages", nil) + require.NoError(t, err) + req = req.WithContext(auth.WithIdentity(req.Context(), identity)) + + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) + rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") + + _, err = rfw.Write([]byte(tc.body)) + require.NoError(t, err) + require.NoError(t, rfw.FlushAndFilter()) + + out := rr.Body.String() + assert.NotContains(t, out, "admin_tool", + "a Response carrying both error and result must not pass the list through under cover of the error field") + + firstLine := strings.TrimSuffix(strings.SplitN(out, "\n", 2)[0], "\r") + var decoded struct { + Error struct { + Code float64 `json:"code"` + } `json:"error"` + } + require.NoError(t, json.Unmarshal([]byte(strings.TrimPrefix(firstLine, "data: ")), &decoded)) + assert.Equal(t, tc.wantErrorCode, decoded.Error.Code, + "the error code distinguishes fail-closed (our internal-error envelope) from pass-through (the upstream's own code)") }) } } @@ -1243,6 +1537,264 @@ func TestResponseFilteringWriter_JSON_DisguisedResponseFrame(t *testing.T) { } } +// TestErrorResponseBody pins the wire shape of the JSON-RPC error envelope +// errorResponseBody produces for a filter/encode failure: a "2.0" jsonrpc tag, +// the standard internal-error code, a generic message that must NOT echo the +// wrapped error (it can name tools the caller is not authorized to see, so +// #6066 and security.md forbid forwarding it), and an id +// that round-trips for a real request id but is entirely ABSENT (not present +// and null) for the zero-value id. Absence matters because MCP types the +// error response id as optional (id?: RequestId), and the reference +// TypeScript SDK's strict schema admits undefined but not null — a null id +// would make that client throw inside its transport. +func TestErrorResponseBody(t *testing.T) { + t.Parallel() + + wrapped := errors.New("leaky-tool-name-sentinel") + + testCases := []struct { + name string + id jsonrpc2.ID + wantID any // nil means the id key must be absent + }{ + {name: "int64 id round-trips", id: jsonrpc2.Int64ID(7), wantID: float64(7)}, + {name: "string id round-trips", id: jsonrpc2.StringID("abc"), wantID: "abc"}, + {name: "zero-value id key is absent, not null", id: jsonrpc2.ID{}}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, nil, nil, string(mcp.MethodToolsList), nil, nil) + body := rfw.errorResponseBody(tc.id, wrapped) + + var decoded map[string]any + require.NoError(t, json.Unmarshal(body, &decoded)) + + assert.Equal(t, "2.0", decoded["jsonrpc"]) + + errObj, ok := decoded["error"].(map[string]any) + require.True(t, ok, "error field must decode as an object") + assert.Equal(t, float64(mcpparser.CodeInternalError), errObj["code"]) + assert.Equal(t, "internal error", errObj["message"], + "the client-visible message must be generic") + assert.NotContains(t, string(body), wrapped.Error(), + "the wrapped error must never reach the client: it can name tools the caller cannot see") + + idValue, hasID := decoded["id"] + require.Equal(t, tc.wantID != nil, hasID, "id key presence mismatch") + if tc.wantID != nil { + assert.Equal(t, tc.wantID, idValue) + } + }) + } +} + +// TestResponseFilteringWriter_SSE_SplitPayloadAcrossDataLines is a regression +// test for the #6037 rewrite: SSE data fields belonging to the same event +// concatenate (per WHATWG semantics) before the event is decoded, so a +// JSON-RPC frame split across two data: lines within one event must still be +// reconstructed and filtered, not smuggled past the filter because each half +// independently fails to decode. +// +// The split point is chosen right after a comma (a whitespace-legal position +// in JSON grammar), so each half is independently invalid JSON on its own, +// but the two halves, reassembled with the LF the SSE spec mandates between +// concatenated data values, form the original valid JSON-RPC frame again. +func TestResponseFilteringWriter_SSE_SplitPayloadAcrossDataLines(t *testing.T) { + t.Parallel() + + authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ + Policies: []string{ + `permit(principal, action == Action::"call_tool", resource == Tool::"weather");`, + }, + EntitiesJSON: `[]`, + }, "") + require.NoError(t, err) + + identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ + Subject: "user1", + Claims: map[string]interface{}{"sub": "user1"}, + }} + + resultJSON, err := json.Marshal(mcp.ListToolsResult{ + Tools: []mcp.Tool{ + {Name: "weather", Description: "Get weather information"}, + {Name: "admin_tool", Description: "Sensitive admin operations"}, + }, + }) + require.NoError(t, err) + encoded, err := jsonrpc2.EncodeMessage(&jsonrpc2.Response{ + ID: jsonrpc2.Int64ID(1), + Result: json.RawMessage(resultJSON), + }) + require.NoError(t, err) + + splitAt := bytes.Index(encoded, []byte(`,"result"`)) + 1 + require.Greater(t, splitAt, 0, "test fixture assumption broken: no comma before \"result\" in the encoded frame") + firstHalf, secondHalf := encoded[:splitAt], encoded[splitAt:] + + // Each half is independently undecodable and carries no result of its + // own, so the pre-rewrite per-line filter passed both through unfiltered. + _, err = jsonrpc2.DecodeMessage(firstHalf) + require.Error(t, err, "test fixture assumption broken: first half must not decode on its own") + require.False(t, carriesResult(firstHalf), "test fixture assumption broken: first half must not independently carry a result") + _, err = jsonrpc2.DecodeMessage(secondHalf) + require.Error(t, err, "test fixture assumption broken: second half must not decode on its own") + require.False(t, carriesResult(secondHalf), "test fixture assumption broken: second half must not independently carry a result") + + body := strings.Join([]string{"data: " + string(firstHalf), "data: " + string(secondHalf), "", ""}, "\n") + + req, err := http.NewRequest(http.MethodPost, "/messages", nil) + require.NoError(t, err) + req = req.WithContext(auth.WithIdentity(req.Context(), identity)) + + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) + rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") + + _, err = rfw.Write([]byte(body)) + require.NoError(t, err) + require.NoError(t, rfw.FlushAndFilter()) + + out := rr.Body.String() + assert.NotContains(t, out, "admin_tool", + "a JSON-RPC frame split across two data: lines within one event must still be filtered") + assert.Contains(t, out, "weather", "the authorized tool must survive filtering") +} + +// TestResponseFilteringWriter_SSE_MultiEventBody is a regression test for the +// #6037 rewrite: a body carrying two complete SSE events (each terminated by +// its own blank line) must deliver both events, with the blank line between +// them preserved, and the second event's result still filtered. The +// pre-rewrite per-line implementation dropped blank lines outright and +// reconciled the deficit with at most one substitute separator, collapsing +// any two-event body into something that looks like one malformed event. +func TestResponseFilteringWriter_SSE_MultiEventBody(t *testing.T) { + t.Parallel() + + authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ + Policies: []string{ + `permit(principal, action == Action::"call_tool", resource == Tool::"weather");`, + }, + EntitiesJSON: `[]`, + }, "") + require.NoError(t, err) + + identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ + Subject: "user1", + Claims: map[string]interface{}{"sub": "user1"}, + }} + + notificationLine := `data: {"jsonrpc":"2.0","method":"notifications/message","params":{"level":"info","data":"warming up"}}` + + resultJSON, err := json.Marshal(mcp.ListToolsResult{ + Tools: []mcp.Tool{ + {Name: "weather", Description: "Get weather information"}, + {Name: "admin_tool", Description: "Sensitive admin operations"}, + }, + }) + require.NoError(t, err) + encoded, err := jsonrpc2.EncodeMessage(&jsonrpc2.Response{ + ID: jsonrpc2.Int64ID(1), + Result: json.RawMessage(resultJSON), + }) + require.NoError(t, err) + respLine := "data: " + string(encoded) + + // Two complete events, each terminated by its own blank line. + body := strings.Join([]string{notificationLine, "", respLine, "", ""}, "\n") + + req, err := http.NewRequest(http.MethodPost, "/messages", nil) + require.NoError(t, err) + req = req.WithContext(auth.WithIdentity(req.Context(), identity)) + + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) + rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") + + _, err = rfw.Write([]byte(body)) + require.NoError(t, err) + require.NoError(t, rfw.FlushAndFilter()) + + out := rr.Body.String() + + // Both events must survive as separate events: the notification verbatim, + // separated by its own blank line from the (filtered) list response. + assert.Contains(t, out, notificationLine+"\n\n", + "the notification event must be delivered with its terminating blank line intact") + assert.NotContains(t, out, "admin_tool", "the second event's result must still be filtered") + assert.Contains(t, out, "weather", "the authorized tool must survive filtering") +} + +// TestResponseFilteringWriter_SSE_StructuralRoundTrip verifies the core +// structural invariant of the #6037 rewrite: a body needing no filtering +// (here, an interleaved notification whose event carries no result) is +// reproduced byte-for-byte, regardless of how the body's separators trail. +func TestResponseFilteringWriter_SSE_StructuralRoundTrip(t *testing.T) { + t.Parallel() + + authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ + Policies: []string{ + `permit(principal, action == Action::"call_tool", resource == Tool::"weather");`, + }, + EntitiesJSON: `[]`, + }, "") + require.NoError(t, err) + + identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ + Subject: "user1", + Claims: map[string]interface{}{"sub": "user1"}, + }} + + // The notification event carries no result, so it never reaches Cedar + // evaluation; the policy content is irrelevant to this test. + const dataLine = `data: {"jsonrpc":"2.0","method":"notifications/message","params":{"level":"info","data":"hi"}}` + + testCases := []struct { + name string + body string + }{ + {name: "ends with a blank line", body: "event: message\n" + dataLine + "\n\n"}, + {name: "ends with a single separator, no blank line", body: "event: message\n" + dataLine + "\n"}, + {name: "no separator anywhere in the body", body: dataLine}, + {name: "CRLF body ending with a blank line", body: "event: message\r\n" + dataLine + "\r\n\r\n"}, + // One event terminated by CRLF then a lone-LF blank line, followed by + // a second event terminated by a lone CR then a CRLF blank line: all + // three SSE-legal terminators appear in one body, which per WHATWG an + // upstream may do freely. + {name: "mixed CRLF, LF and CR within one body", body: dataLine + "\r\n" + "\n" + "event: ping" + "\r" + "\r\n"}, + {name: "consecutive blank lines", body: dataLine + "\n\n\n"}, + {name: "blank lines only", body: "\n\n\n"}, + // Guards the idx+1 < len(rawResponse) bounds check: a lone trailing + // CR with nothing after it must not index out of range. + {name: "ends with a lone CR", body: dataLine + "\r"}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + req, err := http.NewRequest(http.MethodPost, "/messages", nil) + require.NoError(t, err) + req = req.WithContext(auth.WithIdentity(req.Context(), identity)) + + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) + rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") + + _, err = rfw.Write([]byte(tc.body)) + require.NoError(t, err) + require.NoError(t, rfw.FlushAndFilter()) + + assert.Equal(t, tc.body, rr.Body.String(), + "a body needing no filtering must be reproduced byte-for-byte") + }) + } +} + // TestResponseFilteringWriter_FilterBypassAttempts verifies that list results a // backend tries to slip past the filter -- via media-type casing or whitespace // (RFC 9110 makes both legal), a non-200 2xx status (fetch-based clients accept @@ -1367,84 +1919,3 @@ func TestResponseFilteringWriter_FilterBypassAttempts(t *testing.T) { }) } } - -// TestErrorResponseBody pins the wire shape of the JSON-RPC error envelope -// errorResponseBody produces for a filter/encode failure: a "2.0" jsonrpc tag, -// the standard internal-error code, a generic message that must NOT echo the -// wrapped error (it can name tools the caller is not authorized to see, so -// #6066 and security.md forbid forwarding it), and an id that round-trips for -// a real request id but is entirely ABSENT (not present and null) for the -// zero-value id. Absence matters because MCP types the -// error response id as optional (id?: RequestId), and the reference -// TypeScript SDK's strict schema admits undefined but not null — a null id -// would make that client throw inside its transport. -func TestErrorResponseBody(t *testing.T) { - t.Parallel() - - wrapped := errors.New("leaky-tool-name-sentinel") - - testCases := []struct { - name string - id jsonrpc2.ID - wantID any // nil means the id key must be absent - }{ - {name: "int64 id round-trips", id: jsonrpc2.Int64ID(7), wantID: float64(7)}, - {name: "string id round-trips", id: jsonrpc2.StringID("abc"), wantID: "abc"}, - {name: "zero-value id key is absent, not null", id: jsonrpc2.ID{}}, - } - - for _, tc := range testCases { - t.Run(tc.name, func(t *testing.T) { - t.Parallel() - - rr := httptest.NewRecorder() - rfw := NewResponseFilteringWriter(rr, nil, nil, string(mcp.MethodToolsList), nil, nil) - body := rfw.errorResponseBody(tc.id, wrapped) - - var decoded map[string]any - require.NoError(t, json.Unmarshal(body, &decoded)) - - assert.Equal(t, "2.0", decoded["jsonrpc"]) - - errObj, ok := decoded["error"].(map[string]any) - require.True(t, ok, "error field must decode as an object") - assert.Equal(t, float64(mcpparser.CodeInternalError), errObj["code"]) - assert.Equal(t, "internal error", errObj["message"], - "the client-visible message must be generic") - assert.NotContains(t, string(body), wrapped.Error(), - "the wrapped error must never reach the client: it can name tools the caller cannot see") - - idValue, hasID := decoded["id"] - require.Equal(t, tc.wantID != nil, hasID, "id key presence mismatch") - if tc.wantID != nil { - assert.Equal(t, tc.wantID, idValue) - } - }) - } -} - -// TestWriteSSEErrorLine pins the two invariants writeSSEErrorLine must hold -// on the text/event-stream path (issue #6037): the written line is exactly -// "data: " followed by the same envelope errorResponseBody produces, with no -// trailing terminator (the caller, processSSEResponse, owns the separator — -// a hardcoded one here would desync a \r\n stream), and the status code is -// left untouched, since headers are already committed by the time an SSE -// frame fails mid-stream. -func TestWriteSSEErrorLine(t *testing.T) { - t.Parallel() - - // writeSSEErrorLine only reads rfw.ResponseWriter; the authorizer, request, - // and Content-Type are irrelevant on this path, so nils are safe here. - rr := httptest.NewRecorder() - rfw := NewResponseFilteringWriter(rr, nil, nil, string(mcp.MethodToolsList), nil, nil) - - id := jsonrpc2.Int64ID(9) - wrapped := errors.New("boom") - require.NoError(t, rfw.writeSSEErrorLine(id, wrapped)) - - want := "data: " + string(rfw.errorResponseBody(id, wrapped)) - assert.Equal(t, want, rr.Body.String(), - "writeSSEErrorLine must write exactly the data-prefixed envelope, nothing else") - assert.Equal(t, http.StatusOK, rr.Code, - "writeSSEErrorLine must not set a non-200 status; the SSE headers are already committed by the time it runs") -} From 8ace864f06f766908a570f8845bc9a47cc30ff53 Mon Sep 17 00:00:00 2001 From: Jakub Hrozek Date: Tue, 28 Jul 2026 11:08:14 +0200 Subject: [PATCH 2/4] Tighten SSE filter comments and close test gaps External review of the two preceding commits found no correctness regressions but surfaced a comment that would have led a future contributor to loosen a security guard, and a test hole that let a real regression pass the whole suite. The comment: objectCarriesResult was documented as reporting a non-null "result" field. RawMessage's UnmarshalJSON runs even for a JSON null, so probe.Result holds the 4 bytes "null" whenever the key exists at all and an explicit null counts as carrying a result. That is the fail-closed direction and is deliberate, but anyone adding a null check to match the old wording would have widened the smuggling surface. Say what the code does and why. The test hole: every StructuralRoundTrip row asserts byte-identity for a body needing no filtering, and every bypass test used LF bodies, so nothing exercised a filtered event on a CRLF stream. Hardcoding the replacement line's terminator as "\n" passed the entire suite. TestResponseFilteringWriter_SSE_FilteredEventTerminators covers that and the unterminated trailing event, which also pins the deliberate choice not to fabricate a terminator. Also: carriesResult now has direct coverage, including the null case above and the two deliberate false results (an undecodable prefix, and a nested array we do not recurse into) so neither gets "fixed" into a regression. Three verb-free fmt.Errorf calls become errors.New, the duplicated result probe is extracted, and the cedar-authorizer and request boilerplate repeated across ten SSE tests moves into helpers that assert the parsed request actually reached the context. Deliberately unchanged: the JSON-RPC error code stays 500 (#6039), and an absent id stays omitted rather than null, matching #6068. Refs #6037 Co-Authored-By: Claude Opus 5 --- pkg/authz/response_filter.go | 68 +++-- pkg/authz/response_filter_test.go | 493 ++++++++++++++++++------------ 2 files changed, 345 insertions(+), 216 deletions(-) diff --git a/pkg/authz/response_filter.go b/pkg/authz/response_filter.go index fb83b8f54b..3eadc208d6 100644 --- a/pkg/authz/response_filter.go +++ b/pkg/authz/response_filter.go @@ -176,8 +176,15 @@ func (rfw *ResponseFilteringWriter) processJSONResponse(rawResponse []byte) erro if carriesResult(rawResponse) { slog.Warn("JSON response carried a result outside a clean Response frame; dropping as a protocol violation", "method", rfw.method) + // A batch is replaced wholesale by this single error envelope + // even if only one of many elements carried the smuggled + // result: JSON-RPC does define batch responses, but we don't + // attempt a partial one here, and mcpparser.ParsingMiddleware + // already rejects batch requests outright (see #5745), so a + // batch response on this path is already off-spec traffic. + // Fail-closed is the right side to err on regardless. return rfw.writeErrorResponse(rfw.requestID(), - fmt.Errorf("dropped a frame carrying a result outside a clean Response")) + errors.New("dropped a frame carrying a result outside a clean Response")) } rfw.ResponseWriter.WriteHeader(rfw.statusCode) _, werr := rfw.ResponseWriter.Write(rawResponse) @@ -263,7 +270,11 @@ func (rfw *ResponseFilteringWriter) processSSEResponse(rawResponse []byte) error // A trailing event with no closing blank line is never dispatched by a // spec-compliant client, but the backend did send these bytes, so write // them (filtered) rather than drop them — without fabricating a - // terminator the backend didn't send. + // terminator the backend didn't send. Consequence: if that event needed + // filtering, the error envelope substituted here is never dispatched + // either, so the client hangs on its own timeout instead of seeing the + // error. Real backends terminate their events, so this is accepted, not + // fixed. outputLines = append(outputLines, rfw.resolveSSEEvent(event)...) for _, l := range outputLines { @@ -289,12 +300,17 @@ func (rfw *ResponseFilteringWriter) resolveSSEEvent(event []sseLine) []sseLine { var dataValues [][]byte for _, l := range event { - // A bare "data" line with no colon doesn't match the "data:" prefix - // below, so it's excluded from assembly here, unlike strict WHATWG - // grammar (which treats it as a data field with an empty value). - // That's safe: its only effect on a spec-compliant client's buffer - // would be one extra LF, which is legal JSON whitespace outside a - // string literal and equally fatal to us and the client inside one. + // A bare "data" line with no colon, or one with a space before the + // colon ("data :foo"), doesn't match the "data:" prefix below, so both + // are excluded from assembly here, unlike strict WHATWG grammar (which + // treats a bare "data" as a field with an empty value). That's safe + // for two reasons: (1) the only effect on a spec-compliant client's + // assembled buffer is one extra LF, which is JSON-insignificant + // whitespace outside a string literal and can never turn JSON we'd + // reject into JSON a client would accept; (2) for + // "data :foo", bytes.Cut on the raw line yields before == "data " (not + // "data"), so the go-sdk's own line parser ignores it too, exactly + // like us and strict WHATWG. data, ok := bytes.CutPrefix(l.text, []byte("data:")) if !ok { continue @@ -372,7 +388,7 @@ func (rfw *ResponseFilteringWriter) filterSSEEventData(data []byte) []byte { slog.Warn("SSE event carried a result outside a clean Response frame; failing closed as a protocol violation", "method", rfw.method) return rfw.errorResponseBody(rfw.requestID(), - fmt.Errorf("dropped a frame carrying a result outside a clean Response")) + errors.New("dropped a frame carrying a result outside a clean Response")) case decodeErr != nil: // Genuinely undecodable and no smuggled result. Pass through unfiltered. slog.Warn("SSE event data could not be decoded as JSON-RPC; passing through unfiltered", @@ -408,9 +424,13 @@ func requiresResponseFiltering(method string) bool { // The payload may hold more than one top-level JSON value: an SSE event's // assembled data can concatenate several messages (e.g. a notification and a // response, one per data: line) into something DecodeMessage rejects as a -// single value. A json.Decoder walks all of them instead of Unmarshal-ing the -// whole payload as one document, so a result-bearing value after the first is -// still caught rather than making the whole payload look undecodable. +// single value. json.Decoder reads one JSON value at a time and stops at the +// first one it can't decode, rather than Unmarshal-ing the whole payload as +// one document, so a result-bearing value after the first is still caught +// instead of making the whole payload look undecodable. Stopping there +// (rather than trying to resync past the bad value) is safe: a payload the +// decoder can't parse value-by-value isn't a single parseable JSON document +// either, so no strict client can read past that point. func carriesResult(data []byte) bool { dec := json.NewDecoder(bytes.NewReader(data)) for { @@ -432,10 +452,7 @@ func valueCarriesResult(value json.RawMessage) bool { } switch trimmed[0] { case '{': - var probe struct { - Result json.RawMessage `json:"result"` - } - return json.Unmarshal(trimmed, &probe) == nil && probe.Result != nil + return objectCarriesResult(trimmed) case '[': // A single level, not recursive: a JSON-RPC batch is a flat array of // message objects, so that's the whole legitimate surface. Recursing @@ -451,10 +468,7 @@ func valueCarriesResult(value json.RawMessage) bool { if len(elTrimmed) == 0 || elTrimmed[0] != '{' { continue } - var probe struct { - Result json.RawMessage `json:"result"` - } - if json.Unmarshal(elTrimmed, &probe) == nil && probe.Result != nil { + if objectCarriesResult(elTrimmed) { return true } } @@ -478,6 +492,18 @@ func sseCarriesResult(rawResponse []byte) bool { return false } +// objectCarriesResult reports whether a trimmed JSON object value has a +// "result" key present, including an explicit `"result":null` — RawMessage's +// UnmarshalJSON runs even for a JSON null, so probe.Result is non-nil (the +// 4 bytes "null") whenever the key exists at all. Treating a null result as +// carrying one is deliberate: it's the fail-closed direction. +func objectCarriesResult(trimmedObject json.RawMessage) bool { + var probe struct { + Result json.RawMessage `json:"result"` + } + return json.Unmarshal(trimmedObject, &probe) == nil && probe.Result != nil +} + // filterListResponse filters the list response based on authorization policies func (rfw *ResponseFilteringWriter) filterListResponse(response *jsonrpc2.Response) (*jsonrpc2.Response, error) { if response.Error != nil { @@ -488,7 +514,7 @@ func (rfw *ResponseFilteringWriter) filterListResponse(response *jsonrpc2.Respon // closed rather than passing the smuggled list through under cover // of the error field. See #5257. if response.Result != nil { - return nil, fmt.Errorf("response carried both error and result") + return nil, errors.New("response carried both error and result") } // If there's an error and no result, don't filter return response, nil diff --git a/pkg/authz/response_filter_test.go b/pkg/authz/response_filter_test.go index d341a65887..423db8f9fb 100644 --- a/pkg/authz/response_filter_test.go +++ b/pkg/authz/response_filter_test.go @@ -24,6 +24,7 @@ import ( "github.com/stacklok/toolhive-core/mcpcompat/mcp" "github.com/stacklok/toolhive/pkg/auth" + "github.com/stacklok/toolhive/pkg/authz/authorizers" "github.com/stacklok/toolhive/pkg/authz/authorizers/cedar" mcpparser "github.com/stacklok/toolhive/pkg/mcp" "github.com/stacklok/toolhive/pkg/vmcp/optimizer" @@ -450,7 +451,7 @@ func TestResponseFilteringWriter_NonListOperations(t *testing.T) { Result: json.RawMessage(responseData), } - responseBytes, err := jsonrpc2.EncodeMessage(jsonrpcResponse) + responseBytes, err := json.Marshal(jsonrpcResponse) require.NoError(t, err, "Failed to marshal JSON-RPC response") // Create an HTTP request @@ -868,6 +869,153 @@ func TestOptimizerPassThroughToolsInResponseFilter(t *testing.T) { }) } +// TestCarriesResult pins carriesResult's (and, through it, valueCarriesResult's) +// classification of the shapes the SSE and JSON filters route through it: no +// clean single Response, but the caller needs to know whether a "result" key +// is hiding somewhere in the payload regardless. +func TestCarriesResult(t *testing.T) { + t.Parallel() + + const notification = `{"jsonrpc":"2.0","method":"notifications/message","params":{}}` + const result = `{"jsonrpc":"2.0","id":1,"result":{"tools":[]}}` + + testCases := []struct { + name string + data string + want bool + }{ + { + name: "concatenated notification then result", + // The exact shape an SSE event's assembled data takes when two + // data: lines (a notification, then a result-bearing frame) + // concatenate into one payload: DecodeMessage rejects the whole + // thing as a single value, but the embedded result must still be + // caught. + data: notification + "\n" + result, + want: true, + }, + { + name: "heterogeneous batch array", + // A batch whose first element isn't an object: Unmarshal-ing the + // whole array into a single probe struct would fail outright, + // missing the result-bearing element further down. + data: `[1,` + result + `]`, + want: true, + }, + { + name: "undecodable prefix before a result stops at the first value", + // Deliberate: json.Decoder stops at the first value it can't + // decode ("garbage") and never reaches the result-bearing value + // after it. This is safe rather than a missed catch, because no + // strict client can read past that point either. Don't "fix" + // this into scanning past bad input looking for a result. + data: "garbage\n" + result, + want: false, + }, + { + name: "plain notification carries no result", + data: notification, + want: false, + }, + { + name: "empty input", + data: "", + want: false, + }, + { + name: "nested array is not recursed into", + // Deliberate: a JSON-RPC batch is a flat array of message + // objects, so that's the only legitimate batch surface. Recursing + // into nested arrays is an O(d^2) amplification handle on + // encoding/json's own nesting cap for something no client + // actually flattens. Don't "fix" this into recursive descent. + data: `[[` + result + `]]`, + want: false, + }, + { + name: "explicit null result still counts as carrying one", + // Deliberate, the fail-closed direction: RawMessage's + // UnmarshalJSON runs even for a JSON null, so a "result" key + // present with a null value is still treated as carrying a + // result. Don't "fix" this into skipping nulls. + data: `{"jsonrpc":"2.0","id":1,"result":null}`, + want: true, + }, + { + name: "top-level scalar carries no result", + data: `"just a string"`, + want: false, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + assert.Equal(t, tc.want, carriesResult([]byte(tc.data))) + }) + } +} + +// newWeatherOnlyAuthorizer builds the Cedar authorizer most SSE regression +// tests below share: a single policy permitting call_tool on "weather" plus +// any extraPolicies the caller needs for other resource types. +func newWeatherOnlyAuthorizer(t *testing.T, extraPolicies ...string) authorizers.Authorizer { + t.Helper() + policies := append([]string{ + `permit(principal, action == Action::"call_tool", resource == Tool::"weather");`, + }, extraPolicies...) + authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ + Policies: policies, + EntitiesJSON: `[]`, + }, "") + require.NoError(t, err) + return authorizer +} + +// newUser1Request builds a POST /messages request carrying the "user1" +// identity most SSE regression tests below share. +func newUser1Request(t *testing.T) *http.Request { + t.Helper() + identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ + Subject: "user1", + Claims: map[string]interface{}{"sub": "user1"}, + }} + req, err := http.NewRequest(http.MethodPost, "/messages", nil) + require.NoError(t, err) + return req.WithContext(auth.WithIdentity(req.Context(), identity)) +} + +// newParsedUser1Request is newUser1Request plus mcpparser.ParsingMiddleware, +// for tests whose error envelope must correlate against the request's real +// JSON-RPC id: requestID() reads the parsed request back out of context, so +// a plain request (no parsed MCP request in context) would always produce a +// zero, id-less envelope regardless of what the fix is supposed to do. +func newParsedUser1Request(t *testing.T, reqBody string) *http.Request { + t.Helper() + identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ + Subject: "user1", + Claims: map[string]interface{}{"sub": "user1"}, + }} + req, err := http.NewRequest(http.MethodPost, "/messages", strings.NewReader(reqBody)) + require.NoError(t, err) + req.Header.Set("Content-Type", "application/json") + req = req.WithContext(auth.WithIdentity(req.Context(), identity)) + + var parsedReq *http.Request + mcpparser.ParsingMiddleware(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { + parsedReq = r + })).ServeHTTP(httptest.NewRecorder(), req) + require.NotNil(t, parsedReq) + // ParsingMiddleware calls next even when parsing fails, so a nil parsedReq + // alone wouldn't catch a typo'd reqBody that fails to parse — that would + // silently degrade every caller to the zero-ID path. A notification (no + // "id" in reqBody) still parses to a non-nil ParsedMCPRequest with a nil + // ID, so this assertion holds for every caller, not just id-bearing ones. + require.NotNil(t, mcpparser.GetParsedMCPRequest(parsedReq.Context()), + "reqBody must have parsed into a ParsedMCPRequest, or requestID() silently falls back to the zero ID") + return parsedReq +} + // TestResponseFilteringWriter_SSE_PerEventFallthrough is a regression test for // issue #5257: when an SSE upstream interleaves a non-Response event (e.g. an // MCP notification) or an undecodable event with a real list response, the @@ -881,20 +1029,10 @@ func TestOptimizerPassThroughToolsInResponseFilter(t *testing.T) { func TestResponseFilteringWriter_SSE_PerEventFallthrough(t *testing.T) { t.Parallel() - authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ - Policies: []string{ - `permit(principal, action == Action::"call_tool", resource == Tool::"weather");`, - `permit(principal, action == Action::"get_prompt", resource == Prompt::"greeting");`, - `permit(principal, action == Action::"read_resource", resource == Resource::"data");`, - }, - EntitiesJSON: `[]`, - }, "") - require.NoError(t, err) - - identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ - Subject: "user1", - Claims: map[string]interface{}{"sub": "user1"}, - }} + authorizer := newWeatherOnlyAuthorizer(t, + `permit(principal, action == Action::"get_prompt", resource == Prompt::"greeting");`, + `permit(principal, action == Action::"read_resource", resource == Resource::"data");`, + ) // encodeListResponse marshals a list result type into a JSON-RPC Response // data line. @@ -1014,9 +1152,7 @@ func TestResponseFilteringWriter_SSE_PerEventFallthrough(t *testing.T) { t.Run(mc.name+"/"+plc.name, func(t *testing.T) { t.Parallel() - req, err := http.NewRequest(http.MethodPost, "/messages", nil) - require.NoError(t, err) - req = req.WithContext(auth.WithIdentity(req.Context(), identity)) + req := newUser1Request(t) rr := httptest.NewRecorder() rfw := NewResponseFilteringWriter(rr, authorizer, req, mc.method, nil, nil) @@ -1030,7 +1166,7 @@ func TestResponseFilteringWriter_SSE_PerEventFallthrough(t *testing.T) { // what this test means to model. The trailing "" pair // terminates the second event. body := strings.Join([]string{plc.line, "", mc.respLine, "", ""}, "\n") - _, err = rfw.Write([]byte(body)) + _, err := rfw.Write([]byte(body)) require.NoError(t, err) require.NoError(t, rfw.FlushAndFilter()) @@ -1096,18 +1232,7 @@ func TestResponseFilteringWriter_SSE_PerEventFallthrough(t *testing.T) { func TestResponseFilteringWriter_SSE_DisguisedResponseFrame(t *testing.T) { t.Parallel() - authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ - Policies: []string{ - `permit(principal, action == Action::"call_tool", resource == Tool::"weather");`, - }, - EntitiesJSON: `[]`, - }, "") - require.NoError(t, err) - - identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ - Subject: "user1", - Claims: map[string]interface{}{"sub": "user1"}, - }} + authorizer := newWeatherOnlyAuthorizer(t) // A genuine, filterable tools/list response on a later line. realResultJSON, err := json.Marshal(mcp.ListToolsResult{ @@ -1147,16 +1272,7 @@ func TestResponseFilteringWriter_SSE_DisguisedResponseFrame(t *testing.T) { // Run the real MCP parsing middleware so the request context // carries a ParsedMCPRequest with a real id, matching production: // requestID() reads it back out to correlate the error envelope. - reqBody := `{"jsonrpc":"2.0","id":99,"method":"tools/list"}` - req, err := http.NewRequest(http.MethodPost, "/messages", strings.NewReader(reqBody)) - require.NoError(t, err) - req.Header.Set("Content-Type", "application/json") - req = req.WithContext(auth.WithIdentity(req.Context(), identity)) - var parsedReq *http.Request - mcpparser.ParsingMiddleware(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { - parsedReq = r - })).ServeHTTP(httptest.NewRecorder(), req) - require.NotNil(t, parsedReq) + parsedReq := newParsedUser1Request(t, `{"jsonrpc":"2.0","id":99,"method":"tools/list"}`) rr := httptest.NewRecorder() rfw := NewResponseFilteringWriter(rr, authorizer, parsedReq, string(mcp.MethodToolsList), nil, nil) @@ -1166,7 +1282,7 @@ func TestResponseFilteringWriter_SSE_DisguisedResponseFrame(t *testing.T) { // response into two distinct SSE events, matching how a real MCP // server frames SSE (one JSON-RPC message per event). body := strings.Join([]string{f.payload, "", "data: " + string(realResp), "", ""}, "\n") - _, err = rfw.Write([]byte(body)) + _, err := rfw.Write([]byte(body)) require.NoError(t, err) require.NoError(t, rfw.FlushAndFilter()) @@ -1204,25 +1320,17 @@ func TestResponseFilteringWriter_SSE_DisguisedResponseFrame(t *testing.T) { func TestResponseFilteringWriter_SSE_ConcatenatedEventBypass(t *testing.T) { t.Parallel() - authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ - Policies: []string{ - `permit(principal, action == Action::"call_tool", resource == Tool::"weather");`, - }, - EntitiesJSON: `[]`, - }, "") - require.NoError(t, err) - - identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ - Subject: "user1", - Claims: map[string]interface{}{"sub": "user1"}, - }} + authorizer := newWeatherOnlyAuthorizer(t) notificationLine := `data: {"jsonrpc":"2.0","method":"notifications/message","params":{}}` resultLine := `data: {"jsonrpc":"2.0","id":1,"result":{"tools":[{"name":"admin_tool"}]}}` - req, err := http.NewRequest(http.MethodPost, "/messages", nil) - require.NoError(t, err) - req = req.WithContext(auth.WithIdentity(req.Context(), identity)) + // Run the real MCP parsing middleware so the request context carries a + // ParsedMCPRequest with a real id, matching production: requestID() reads + // it back out to correlate the error envelope. Without this, requestID() + // falls back to the zero ID, the envelope carries no "id", and this test + // would accept an envelope no client could actually correlate. + req := newParsedUser1Request(t, `{"jsonrpc":"2.0","id":99,"method":"tools/list"}`) rr := httptest.NewRecorder() rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) @@ -1230,17 +1338,26 @@ func TestResponseFilteringWriter_SSE_ConcatenatedEventBypass(t *testing.T) { // No blank line between the two data: lines: one event, two fields. body := strings.Join([]string{notificationLine, resultLine, "", ""}, "\n") - _, err = rfw.Write([]byte(body)) + _, err := rfw.Write([]byte(body)) require.NoError(t, err) require.NoError(t, rfw.FlushAndFilter()) out := rr.Body.String() assert.NotContains(t, out, "admin_tool", "a result concatenated after a notification within one event must not bypass the filter") + // NotContains alone would also pass if the event were dropped silently - // (#6037's hang); require the fail-closed envelope actually went out. - assert.Contains(t, out, `"error"`, - "the concatenated event must fail closed into an error envelope, not vanish silently") + // (#6037's hang); require the fail-closed envelope actually went out and + // carries the request's id so the client can correlate it. + firstLine := strings.TrimSuffix(strings.SplitN(out, "\n", 2)[0], "\r") + payload := strings.TrimPrefix(firstLine, "data: ") + msg, err := jsonrpc2.DecodeMessage([]byte(payload)) + require.NoError(t, err, "the fail-closed envelope must itself be valid JSON-RPC") + resp, ok := msg.(*jsonrpc2.Response) + require.True(t, ok, "the fail-closed envelope must be a clean Response") + require.NotNil(t, resp.Error, "the concatenated event must fail closed into an error envelope, not vanish silently") + assert.Equal(t, jsonrpc2.Int64ID(99), resp.ID, + "the error envelope must carry the request's id so the client can correlate it, or the #6037 hang reproduces") } // TestResponseFilteringWriter_SSE_MixedSeparatorsBypass is a regression test @@ -1255,22 +1372,8 @@ func TestResponseFilteringWriter_SSE_ConcatenatedEventBypass(t *testing.T) { func TestResponseFilteringWriter_SSE_MixedSeparatorsBypass(t *testing.T) { t.Parallel() - authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ - Policies: []string{ - `permit(principal, action == Action::"call_tool", resource == Tool::"weather");`, - }, - EntitiesJSON: `[]`, - }, "") - require.NoError(t, err) - - identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ - Subject: "user1", - Claims: map[string]interface{}{"sub": "user1"}, - }} - - req, err := http.NewRequest(http.MethodPost, "/messages", nil) - require.NoError(t, err) - req = req.WithContext(auth.WithIdentity(req.Context(), identity)) + authorizer := newWeatherOnlyAuthorizer(t) + req := newUser1Request(t) rr := httptest.NewRecorder() rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) @@ -1311,22 +1414,8 @@ func TestResponseFilteringWriter_SSE_MixedSeparatorsBypass(t *testing.T) { func TestResponseFilteringWriter_SSE_LeadingBOMBypass(t *testing.T) { t.Parallel() - authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ - Policies: []string{ - `permit(principal, action == Action::"call_tool", resource == Tool::"weather");`, - }, - EntitiesJSON: `[]`, - }, "") - require.NoError(t, err) - - identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ - Subject: "user1", - Claims: map[string]interface{}{"sub": "user1"}, - }} - - req, err := http.NewRequest(http.MethodPost, "/messages", nil) - require.NoError(t, err) - req = req.WithContext(auth.WithIdentity(req.Context(), identity)) + authorizer := newWeatherOnlyAuthorizer(t) + req := newUser1Request(t) rr := httptest.NewRecorder() rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) @@ -1349,6 +1438,8 @@ func TestResponseFilteringWriter_SSE_LeadingBOMBypass(t *testing.T) { assert.NotContains(t, out, "admin_tool", "a leading UTF-8 BOM must not bypass the filter") assert.Contains(t, out, "weather", "the authorized tool must survive filtering") + assert.False(t, strings.HasPrefix(out, "\xEF\xBB\xBF"), + "the leading BOM must be dropped from the output, not stripped-for-matching then re-emitted") } // TestResponseFilteringWriter_SSE_ErrorAndResultBypass is a regression test @@ -1363,23 +1454,12 @@ func TestResponseFilteringWriter_SSE_LeadingBOMBypass(t *testing.T) { // The error-only row distinguishes "failed closed" from "passed through // unfiltered": both leave no admin_tool on the wire, so a substring check // alone can't tell a fix from a regression that fails closed on every -// legitimate upstream error. Only the error *code* does — 500 is our -// envelope, anything else is the upstream's own error surviving untouched. +// legitimate upstream error. Only the error *code* does — mcpparser.CodeInternalError +// is our envelope, anything else is the upstream's own error surviving untouched. func TestResponseFilteringWriter_SSE_ErrorAndResultBypass(t *testing.T) { t.Parallel() - authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ - Policies: []string{ - `permit(principal, action == Action::"call_tool", resource == Tool::"weather");`, - }, - EntitiesJSON: `[]`, - }, "") - require.NoError(t, err) - - identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ - Subject: "user1", - Claims: map[string]interface{}{"sub": "user1"}, - }} + authorizer := newWeatherOnlyAuthorizer(t) testCases := []struct { name string @@ -1403,15 +1483,13 @@ func TestResponseFilteringWriter_SSE_ErrorAndResultBypass(t *testing.T) { t.Run(tc.name, func(t *testing.T) { t.Parallel() - req, err := http.NewRequest(http.MethodPost, "/messages", nil) - require.NoError(t, err) - req = req.WithContext(auth.WithIdentity(req.Context(), identity)) + req := newUser1Request(t) rr := httptest.NewRecorder() rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") - _, err = rfw.Write([]byte(tc.body)) + _, err := rfw.Write([]byte(tc.body)) require.NoError(t, err) require.NoError(t, rfw.FlushAndFilter()) @@ -1427,7 +1505,7 @@ func TestResponseFilteringWriter_SSE_ErrorAndResultBypass(t *testing.T) { } require.NoError(t, json.Unmarshal([]byte(strings.TrimPrefix(firstLine, "data: ")), &decoded)) assert.Equal(t, tc.wantErrorCode, decoded.Error.Code, - "the error code distinguishes fail-closed (our internal-error envelope) from pass-through (the upstream's own code)") + "the error code distinguishes fail-closed (mcpparser.CodeInternalError, our envelope) from pass-through (the upstream's own code)") }) } } @@ -1443,18 +1521,7 @@ func TestResponseFilteringWriter_SSE_ErrorAndResultBypass(t *testing.T) { func TestResponseFilteringWriter_JSON_DisguisedResponseFrame(t *testing.T) { t.Parallel() - authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ - Policies: []string{ - `permit(principal, action == Action::"call_tool", resource == Tool::"weather");`, - }, - EntitiesJSON: `[]`, - }, "") - require.NoError(t, err) - - identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ - Subject: "user1", - Claims: map[string]interface{}{"sub": "user1"}, - }} + authorizer := newWeatherOnlyAuthorizer(t) // A batch array smuggling a tools/list result. DecodeMessage rejects the // array, so the pre-fix JSON path wrote it through raw. @@ -1491,27 +1558,18 @@ func TestResponseFilteringWriter_JSON_DisguisedResponseFrame(t *testing.T) { t.Run(tc.name, func(t *testing.T) { t.Parallel() - reqBody := `{"jsonrpc":"2.0",` + tc.reqIDJSON + `"method":"tools/list"}` - req, err := http.NewRequest(http.MethodPost, "/messages", strings.NewReader(reqBody)) - require.NoError(t, err) - req.Header.Set("Content-Type", "application/json") - req = req.WithContext(auth.WithIdentity(req.Context(), identity)) - // Run the real MCP parsing middleware so the request context carries a // ParsedMCPRequest, matching how the request would look at this point // in the real middleware chain. writeErrorResponse's id-recovery path // reads the request id back out of that context. - var parsedReq *http.Request - mcpparser.ParsingMiddleware(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { - parsedReq = r - })).ServeHTTP(httptest.NewRecorder(), req) - require.NotNil(t, parsedReq) + reqBody := `{"jsonrpc":"2.0",` + tc.reqIDJSON + `"method":"tools/list"}` + parsedReq := newParsedUser1Request(t, reqBody) rr := httptest.NewRecorder() rfw := NewResponseFilteringWriter(rr, authorizer, parsedReq, string(mcp.MethodToolsList), nil, nil) rfw.ResponseWriter.Header().Set("Content-Type", "application/json") - _, err = rfw.Write([]byte(smuggled)) + _, err := rfw.Write([]byte(smuggled)) require.NoError(t, err) require.NoError(t, rfw.FlushAndFilter()) @@ -1538,18 +1596,23 @@ func TestResponseFilteringWriter_JSON_DisguisedResponseFrame(t *testing.T) { } // TestErrorResponseBody pins the wire shape of the JSON-RPC error envelope -// errorResponseBody produces for a filter/encode failure: a "2.0" jsonrpc tag, -// the standard internal-error code, a generic message that must NOT echo the -// wrapped error (it can name tools the caller is not authorized to see, so -// #6066 and security.md forbid forwarding it), and an id -// that round-trips for a real request id but is entirely ABSENT (not present -// and null) for the zero-value id. Absence matters because MCP types the -// error response id as optional (id?: RequestId), and the reference -// TypeScript SDK's strict schema admits undefined but not null — a null id -// would make that client throw inside its transport. +// errorResponseBody produces for a filter/encode failure: a "2.0" jsonrpc +// tag, the standard JSON-RPC internal-error code (mcpparser.CodeInternalError) +// with a fixed generic client-visible message that never echoes the wrapped +// error's text (#6066 — that error can originate in policy evaluation and +// name tools or resources), and an id that round-trips for a real request id +// but is entirely ABSENT (not present and null) for the zero-value id. +// Absence matters because MCP types the error response id as optional +// (id?: RequestId), and the reference TypeScript SDK's strict schema admits +// undefined but not null — a null id would make that client throw inside +// its transport. func TestErrorResponseBody(t *testing.T) { t.Parallel() + authorizer := newWeatherOnlyAuthorizer(t) + // A distinctive sentinel, not a generic message: its absence from the + // encoded body is what actually pins the #6066 scrub, rather than + // passing by the coincidence of a short or common wrapped error string. wrapped := errors.New("leaky-tool-name-sentinel") testCases := []struct { @@ -1566,8 +1629,8 @@ func TestErrorResponseBody(t *testing.T) { t.Run(tc.name, func(t *testing.T) { t.Parallel() - rr := httptest.NewRecorder() - rfw := NewResponseFilteringWriter(rr, nil, nil, string(mcp.MethodToolsList), nil, nil) + req := newUser1Request(t) + rfw := NewResponseFilteringWriter(httptest.NewRecorder(), authorizer, req, string(mcp.MethodToolsList), nil, nil) body := rfw.errorResponseBody(tc.id, wrapped) var decoded map[string]any @@ -1579,9 +1642,9 @@ func TestErrorResponseBody(t *testing.T) { require.True(t, ok, "error field must decode as an object") assert.Equal(t, float64(mcpparser.CodeInternalError), errObj["code"]) assert.Equal(t, "internal error", errObj["message"], - "the client-visible message must be generic") + "the client-visible message must be the fixed generic string, not the wrapped error") assert.NotContains(t, string(body), wrapped.Error(), - "the wrapped error must never reach the client: it can name tools the caller cannot see") + "the wrapped error's text must never reach the wire (#6066)") idValue, hasID := decoded["id"] require.Equal(t, tc.wantID != nil, hasID, "id key presence mismatch") @@ -1606,18 +1669,7 @@ func TestErrorResponseBody(t *testing.T) { func TestResponseFilteringWriter_SSE_SplitPayloadAcrossDataLines(t *testing.T) { t.Parallel() - authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ - Policies: []string{ - `permit(principal, action == Action::"call_tool", resource == Tool::"weather");`, - }, - EntitiesJSON: `[]`, - }, "") - require.NoError(t, err) - - identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ - Subject: "user1", - Claims: map[string]interface{}{"sub": "user1"}, - }} + authorizer := newWeatherOnlyAuthorizer(t) resultJSON, err := json.Marshal(mcp.ListToolsResult{ Tools: []mcp.Tool{ @@ -1647,9 +1699,7 @@ func TestResponseFilteringWriter_SSE_SplitPayloadAcrossDataLines(t *testing.T) { body := strings.Join([]string{"data: " + string(firstHalf), "data: " + string(secondHalf), "", ""}, "\n") - req, err := http.NewRequest(http.MethodPost, "/messages", nil) - require.NoError(t, err) - req = req.WithContext(auth.WithIdentity(req.Context(), identity)) + req := newUser1Request(t) rr := httptest.NewRecorder() rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) @@ -1675,18 +1725,7 @@ func TestResponseFilteringWriter_SSE_SplitPayloadAcrossDataLines(t *testing.T) { func TestResponseFilteringWriter_SSE_MultiEventBody(t *testing.T) { t.Parallel() - authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ - Policies: []string{ - `permit(principal, action == Action::"call_tool", resource == Tool::"weather");`, - }, - EntitiesJSON: `[]`, - }, "") - require.NoError(t, err) - - identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ - Subject: "user1", - Claims: map[string]interface{}{"sub": "user1"}, - }} + authorizer := newWeatherOnlyAuthorizer(t) notificationLine := `data: {"jsonrpc":"2.0","method":"notifications/message","params":{"level":"info","data":"warming up"}}` @@ -1707,9 +1746,7 @@ func TestResponseFilteringWriter_SSE_MultiEventBody(t *testing.T) { // Two complete events, each terminated by its own blank line. body := strings.Join([]string{notificationLine, "", respLine, "", ""}, "\n") - req, err := http.NewRequest(http.MethodPost, "/messages", nil) - require.NoError(t, err) - req = req.WithContext(auth.WithIdentity(req.Context(), identity)) + req := newUser1Request(t) rr := httptest.NewRecorder() rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) @@ -1736,18 +1773,7 @@ func TestResponseFilteringWriter_SSE_MultiEventBody(t *testing.T) { func TestResponseFilteringWriter_SSE_StructuralRoundTrip(t *testing.T) { t.Parallel() - authorizer, err := cedar.NewCedarAuthorizer(cedar.ConfigOptions{ - Policies: []string{ - `permit(principal, action == Action::"call_tool", resource == Tool::"weather");`, - }, - EntitiesJSON: `[]`, - }, "") - require.NoError(t, err) - - identity := &auth.Identity{PrincipalInfo: auth.PrincipalInfo{ - Subject: "user1", - Claims: map[string]interface{}{"sub": "user1"}, - }} + authorizer := newWeatherOnlyAuthorizer(t) // The notification event carries no result, so it never reaches Cedar // evaluation; the policy content is irrelevant to this test. @@ -1777,15 +1803,13 @@ func TestResponseFilteringWriter_SSE_StructuralRoundTrip(t *testing.T) { t.Run(tc.name, func(t *testing.T) { t.Parallel() - req, err := http.NewRequest(http.MethodPost, "/messages", nil) - require.NoError(t, err) - req = req.WithContext(auth.WithIdentity(req.Context(), identity)) + req := newUser1Request(t) rr := httptest.NewRecorder() rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") - _, err = rfw.Write([]byte(tc.body)) + _, err := rfw.Write([]byte(tc.body)) require.NoError(t, err) require.NoError(t, rfw.FlushAndFilter()) @@ -1795,6 +1819,85 @@ func TestResponseFilteringWriter_SSE_StructuralRoundTrip(t *testing.T) { } } +// TestResponseFilteringWriter_SSE_FilteredEventTerminators covers a hole +// TestResponseFilteringWriter_SSE_StructuralRoundTrip leaves open: every row +// there asserts byte-identity for a body that needs no filtering, so nothing +// ever reaches resolveSSEEvent's replacement branch. Swapping that branch's +// `term: l.term` for a hardcoded []byte("\n") would still pass the whole +// suite. These cases exercise a body that IS filtered, so they assert the +// (changed) output rather than a reproduced input and don't belong under the +// byte-identity test's name. +func TestResponseFilteringWriter_SSE_FilteredEventTerminators(t *testing.T) { + t.Parallel() + + authorizer := newWeatherOnlyAuthorizer(t) + + resultJSON, err := json.Marshal(mcp.ListToolsResult{ + Tools: []mcp.Tool{ + {Name: "weather", Description: "Get weather information"}, + {Name: "admin_tool", Description: "Sensitive admin operations"}, + }, + }) + require.NoError(t, err) + encoded, err := jsonrpc2.EncodeMessage(&jsonrpc2.Response{ + ID: jsonrpc2.Int64ID(1), + Result: json.RawMessage(resultJSON), + }) + require.NoError(t, err) + + t.Run("CRLF body: filtered data line and blank line stay CRLF-terminated", func(t *testing.T) { + t.Parallel() + + req := newUser1Request(t) + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) + rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") + + body := "data: " + string(encoded) + "\r\n\r\n" + _, err := rfw.Write([]byte(body)) + require.NoError(t, err) + require.NoError(t, rfw.FlushAndFilter()) + + out := rr.Body.String() + assert.NotContains(t, out, "admin_tool", "the filtered event must not leak the denied tool") + assert.Contains(t, out, "weather", "the authorized tool must survive filtering") + require.True(t, strings.HasSuffix(out, "\r\n\r\n"), + "the replacement data line and the following blank line must both stay CRLF-terminated: %q", out) + + // Every LF in the output must be immediately preceded by a CR: a bare + // LF proves the replacement line (or the blank line after it) fell + // back to a hardcoded "\n" instead of the CRLF the input actually used. + for i := 0; i < len(out); i++ { + if out[i] == '\n' { + require.True(t, i > 0 && out[i-1] == '\r', "found a bare LF at byte %d in %q", i, out) + } + } + }) + + t.Run("trailing filtered event with no closing blank line", func(t *testing.T) { + t.Parallel() + + req := newUser1Request(t) + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) + rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") + + // No trailing terminator at all: the backend closed the stream + // mid-event, same fixture as the unterminated-body case documented + // in processSSEResponse. + body := "data: " + string(encoded) + _, err := rfw.Write([]byte(body)) + require.NoError(t, err) + require.NoError(t, rfw.FlushAndFilter()) + + out := rr.Body.String() + assert.NotContains(t, out, "admin_tool", "the filtered event must not leak the denied tool") + assert.Contains(t, out, "weather", "the authorized tool must survive filtering") + assert.False(t, strings.HasSuffix(out, "\n") || strings.HasSuffix(out, "\r"), + "an unterminated input must not gain a fabricated terminator: %q", out) + }) +} + // TestResponseFilteringWriter_FilterBypassAttempts verifies that list results a // backend tries to slip past the filter -- via media-type casing or whitespace // (RFC 9110 makes both legal), a non-200 2xx status (fetch-based clients accept From 3fed5609ff10abb1a4e8c83c4d6c4802a41a52ef Mon Sep 17 00:00:00 2001 From: Jakub Hrozek Date: Tue, 28 Jul 2026 13:35:16 +0200 Subject: [PATCH 3/4] Close two SSE filter bypasses found in review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both are regressions the per-event rewrite introduced, found by review on #6088 and reproduced before fixing. Assembling the event's data values left leading whitespace on the payload. Go's JSON scanner treats only SP, TAB, CR and LF as whitespace, but unicode.IsSpace also covers U+000B, U+000C, U+0085, U+00A0, U+1680 and U+3000 — and the go-sdk client trims its data buffer with exactly bytes.TrimSpace. So a payload led by any of those six was rejected by both DecodeMessage and carriesResult while the client parsed it happily, and the event fell through to the undecodable branch with the unfiltered list intact. Trim on assembly instead, which aligns us with the client's own normalization: all six now filter correctly rather than merely failing closed. The second is a coverage regression rather than a new hole. Processing per event means one malformed data value makes the whole event skip filtering, where the old per-line loop filtered each line independently. A result-bearing value sitting after a malformed one in the same event therefore passed through raw. Probe each data value on its own before conceding the event needs no filtering. This only ever turns a pass-through into a fail-closed envelope, never the reverse, so it cannot reintroduce the split-payload bypass the rewrite closed. Tests for both, plus the two the review asked for: a strict WHATWG-grammar SSE parser so the error-path assertions test delivery rather than bytes — conformant payloads that no client dispatches are the whole of #6037 — and the fail-closed invariant for the rest of the stream after a mid-stream failure, which nothing pinned before. That last test asserts an authorized tool survives the second event, not just that the denied one is absent. Absence alone would also hold if the loop truncated the stream, which is the regression the test exists to catch; verified against that mutation specifically. Refs #6037 Co-Authored-By: Claude Opus 5 --- pkg/authz/response_filter.go | 32 ++- pkg/authz/response_filter_test.go | 314 ++++++++++++++++++++++++++++++ 2 files changed, 344 insertions(+), 2 deletions(-) diff --git a/pkg/authz/response_filter.go b/pkg/authz/response_filter.go index 3eadc208d6..c17108b9b5 100644 --- a/pkg/authz/response_filter.go +++ b/pkg/authz/response_filter.go @@ -326,7 +326,14 @@ func (rfw *ResponseFilteringWriter) resolveSSEEvent(event []sseLine) []sseLine { return event } - assembled := bytes.Join(dataValues, []byte("\n")) + // TrimSpace here, not inside the decoders. Go's JSON scanner treats only + // SP, TAB, CR and LF as whitespace, but unicode.IsSpace also covers U+000B, + // U+000C, U+0085, U+00A0, U+1680 and U+3000 — and the go-sdk client trims + // its data buffer with exactly this function (mcp/event.go). Leaving those + // bytes on meant DecodeMessage and carriesResult both rejected the payload + // while the client happily parsed it, so the event fell through to the + // undecodable branch and the unfiltered list went out (#5257). + assembled := bytes.TrimSpace(bytes.Join(dataValues, []byte("\n"))) if len(assembled) == 0 { // A client never dispatches an empty data buffer, so there's nothing // to classify or fail closed on. @@ -334,7 +341,28 @@ func (rfw *ResponseFilteringWriter) resolveSSEEvent(event []sseLine) []sseLine { } replacement := rfw.filterSSEEventData(assembled) if replacement == nil { - return event + // filterSSEEventData's own carriesResult scan stops at the first + // undecodable JSON value in the assembled (joined) payload, so a + // well-formed result-bearing value sitting right after a malformed + // one within the SAME event would otherwise ride through as + // "undecodable, nothing to filter" -- reopening the split-payload + // bypass this rewrite exists to close (#5257), just one line lower + // than before. Probe each data: value independently before + // conceding the event needs no filtering: this only ever turns a + // pass-through into a fail-closed error envelope, never the + // reverse, so it cannot itself reintroduce a bypass. + for _, v := range dataValues { + if carriesResult(v) { + slog.Warn("SSE event's assembled payload was undecodable but an individual data value carries a result; failing closed", + "method", rfw.method) + replacement = rfw.errorResponseBody(rfw.requestID(), + errors.New("dropped a frame carrying a result outside a clean Response")) + break + } + } + if replacement == nil { + return event + } } out := make([]sseLine, 0, len(event)) diff --git a/pkg/authz/response_filter_test.go b/pkg/authz/response_filter_test.go index 423db8f9fb..629f7885a0 100644 --- a/pkg/authz/response_filter_test.go +++ b/pkg/authz/response_filter_test.go @@ -2022,3 +2022,317 @@ func TestResponseFilteringWriter_FilterBypassAttempts(t *testing.T) { }) } } + +// TestResponseFilteringWriter_SSE_LeadingNonJSONWhitespaceStillFiltered is a +// regression test for the assembled-payload TrimSpace fix: Go's JSON scanner +// only treats SP/TAB/CR/LF as whitespace, but unicode.IsSpace (and the +// go-sdk client's own buffer trim, mcp/event.go) also covers U+000B, U+000C, +// U+0085, U+00A0, U+1680 and U+3000. Any of those left on the front of the +// assembled payload made DecodeMessage and carriesResult both reject it while +// the client parsed it fine, so the event fell to the undecodable-passthrough +// branch and the unfiltered list went out (#5257). +// +// NotContains on admin_tool alone would also be satisfied by the event +// failing closed, which would silently degrade every legitimate response +// that happens to carry leading whitespace -- so each case also requires the +// permitted tool to survive, proving the event was filtered, not dropped. +func TestResponseFilteringWriter_SSE_LeadingNonJSONWhitespaceStillFiltered(t *testing.T) { + t.Parallel() + + authorizer := newWeatherOnlyAuthorizer(t) + + resultJSON, err := json.Marshal(mcp.ListToolsResult{ + Tools: []mcp.Tool{ + {Name: "weather", Description: "Get weather information"}, + {Name: "admin_tool", Description: "Sensitive admin operations"}, + }, + }) + require.NoError(t, err) + encoded, err := jsonrpc2.EncodeMessage(&jsonrpc2.Response{ + ID: jsonrpc2.Int64ID(1), + Result: json.RawMessage(resultJSON), + }) + require.NoError(t, err) + + testCases := []struct { + name string + prefix string + }{ + {name: "no prefix (baseline)", prefix: ""}, + {name: "U+000B line tabulation", prefix: "\u000B"}, + {name: "U+000C form feed", prefix: "\u000C"}, + {name: "U+0085 next line", prefix: "\u0085"}, + {name: "U+00A0 no-break space", prefix: "\u00A0"}, + {name: "U+1680 ogham space mark", prefix: "\u1680"}, + {name: "U+3000 ideographic space", prefix: "\u3000"}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + req := newUser1Request(t) + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) + rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") + + body := "data: " + tc.prefix + string(encoded) + "\n\n" + _, err := rfw.Write([]byte(body)) + require.NoError(t, err) + require.NoError(t, rfw.FlushAndFilter()) + + out := rr.Body.String() + assert.NotContains(t, out, "admin_tool", + "a leading non-JSON-whitespace byte before the payload must not bypass the filter") + assert.Contains(t, out, "weather", + "the permitted tool must survive filtering, proving the event was filtered rather than dropped") + }) + } +} + +// TestResponseFilteringWriter_SSE_PerValueProbeCatchesConcatenatedGarbage is a +// regression test for the per-value probe added to resolveSSEEvent: one +// malformed data: value anywhere in an event used to give up filtering for +// the WHOLE event. Concretely, "garbage" followed (no blank line, so same +// event) by a genuine result-bearing frame assembles into +// "garbage\n{...result...}"; jsonrpc2.DecodeMessage stops at "garbage" and +// carriesResult on the whole joined payload does too, so both lines went out +// unfiltered. The fix probes each data: value independently before +// conceding the event needs no filtering. +func TestResponseFilteringWriter_SSE_PerValueProbeCatchesConcatenatedGarbage(t *testing.T) { + t.Parallel() + + authorizer := newWeatherOnlyAuthorizer(t) + + resultJSON, err := json.Marshal(mcp.ListToolsResult{ + Tools: []mcp.Tool{ + {Name: "weather", Description: "Get weather information"}, + {Name: "admin_tool", Description: "Sensitive admin operations"}, + }, + }) + require.NoError(t, err) + encoded, err := jsonrpc2.EncodeMessage(&jsonrpc2.Response{ + ID: jsonrpc2.Int64ID(1), + Result: json.RawMessage(resultJSON), + }) + require.NoError(t, err) + + testCases := []struct { + name string + garbage string + }{ + {name: "plain garbage prefix", garbage: "garbage"}, + {name: "comment-like prefix", garbage: "//comment"}, + {name: "bare NaN prefix", garbage: "NaN"}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + joined := tc.garbage + "\n" + string(encoded) + _, decErr := jsonrpc2.DecodeMessage([]byte(joined)) + require.Error(t, decErr, + "test fixture assumption broken: assembled payload must not decode as a single Response") + require.False(t, carriesResult([]byte(joined)), + "test fixture assumption broken: the whole-payload scan must not itself catch the smuggled result "+ + "-- otherwise this test would not exercise the per-value probe at all") + + req := newUser1Request(t) + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) + rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") + + // One event, two data: lines with no blank line between them, so + // WHATWG assembly joins them with LF into one payload the + // decoder gives up on at the first byte. + body := "data: " + tc.garbage + "\n" + "data: " + string(encoded) + "\n\n" + _, err := rfw.Write([]byte(body)) + require.NoError(t, err) + require.NoError(t, rfw.FlushAndFilter()) + + out := rr.Body.String() + assert.NotContains(t, out, "admin_tool", + "a result-bearing data: line following a malformed one in the same event must not bypass the filter") + }) + } +} + +// TestResponseFilteringWriter_SSE_FailClosedDoesNotLeakSubsequentEvents pins +// the #6037 fail-closed property that neither the event-based rewrite nor the +// TrimSpace fix's tests cover: after a mid-stream event fails to filter and +// is replaced with an error envelope, the loop must keep filtering every +// event that follows, not fall back to passing the remainder of the stream +// through raw. failingFrame is a Response carrying both "error" and "result" +// (see TestResponseFilteringWriter_SSE_ErrorAndResultBypass), a real +// reachable trigger for the fail-closed branch on this path. +func TestResponseFilteringWriter_SSE_FailClosedDoesNotLeakSubsequentEvents(t *testing.T) { + t.Parallel() + + authorizer := newWeatherOnlyAuthorizer(t) + + failingFrame := `{"jsonrpc":"2.0","id":1,"error":{"code":1,"message":"x"},` + + `"result":{"tools":[{"name":"admin_tool"}]}}` + + // The second event carries an authorized tool alongside the denied one on + // purpose. A NotContains check alone would also be satisfied by the loop + // TRUNCATING the stream after the failure — writing nothing downstream — so + // it would pass under exactly the regression this test exists to catch. + // Asserting the authorized tool survives is what proves the loop kept + // filtering rather than giving up. Do not reduce this to one assertion. + stream := "data: " + failingFrame + "\n\n" + + "data: " + `{"jsonrpc":"2.0","id":9,"result":{"tools":[{"name":"weather"},{"name":"admin_tool"}]}}` + "\n\n" + + req := newUser1Request(t) + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) + rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") + + _, err := rfw.Write([]byte(stream)) + require.NoError(t, err) + require.NoError(t, rfw.FlushAndFilter()) + + out := rr.Body.String() + assert.NotContains(t, out, "admin_tool", + "a smuggled-result event after a mid-stream filter failure must still be filtered, not passed through raw") + assert.Contains(t, out, "weather", + "the authorized tool proves the loop kept filtering after the failure instead of truncating the stream") + assert.Contains(t, out, `"error"`, + "the failing event must still emit its correlatable error envelope") +} + +// sseEvent is one event dispatched by parseSSEStream. +type sseEvent struct { + eventType string + data string +} + +// parseSSEStream implements the WHATWG EventSource parsing grammar strictly, +// so tests assert actual client-visible delivery rather than merely +// conformant-looking bytes: after #6066 the error payload can look correct +// on the wire while never being dispatched to a client at all (e.g. a bare +// JSON object with no "data:" prefix). Only what this dispatches counts as +// delivered. +// +// Per the grammar: lines end at CRLF, LF, or CR; a leading ':' marks a +// comment line with no effect; a "name:value" line strips exactly one +// leading space from the value; a line with no colon is a field with an +// empty value; only "data" and "event" have any effect, every other field +// name is ignored outright; a blank line dispatches the event only if the +// data buffer is non-empty (and is reset either way); any data left +// buffered at EOF is discarded, never dispatched. +func parseSSEStream(t *testing.T, body []byte) []sseEvent { + t.Helper() + + var events []sseEvent + var eventType string + var dataLines []string + + dispatch := func() { + if len(dataLines) > 0 { + events = append(events, sseEvent{eventType: eventType, data: strings.Join(dataLines, "\n")}) + } + eventType = "" + dataLines = nil + } + + for len(body) > 0 { + idx := bytes.IndexAny(body, "\r\n") + var line []byte + switch { + case idx == -1: + line, body = body, nil + case body[idx] == '\r' && idx+1 < len(body) && body[idx+1] == '\n': + line, body = body[:idx], body[idx+2:] + default: + line, body = body[:idx], body[idx+1:] + } + + if len(line) == 0 { + dispatch() + continue + } + if line[0] == ':' { + continue + } + + field, value, hasColon := bytes.Cut(line, []byte(":")) + if !hasColon { + field, value = line, nil + } + value = bytes.TrimPrefix(value, []byte(" ")) + + switch string(field) { + case "data": + dataLines = append(dataLines, string(value)) + case "event": + eventType = string(value) + } + // Every other field name is ignored outright. + } + // Pending data at EOF is never dispatched. + + return events +} + +// TestParseSSEStream_DeliversOnlyDispatchedEvents guards the SSE error path +// with the strict grammar parser rather than a raw byte/substring check: it +// asserts the fail-closed envelope is actually dispatched as a single +// default-typed event, which is what an MCP client (default/"message" event +// handler only) will actually see. +func TestParseSSEStream_DeliversOnlyDispatchedEvents(t *testing.T) { + t.Parallel() + + authorizer := newWeatherOnlyAuthorizer(t) + parsedReq := newParsedUser1Request(t, `{"jsonrpc":"2.0","id":99,"method":"tools/list"}`) + + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, authorizer, parsedReq, string(mcp.MethodToolsList), nil, nil) + rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") + + body := `data: {"jsonrpc":"2.0","method":"notifications/initialized","id":1,` + + `"result":{"tools":[{"name":"admin_tool"}]}}` + "\n\n" + _, err := rfw.Write([]byte(body)) + require.NoError(t, err) + require.NoError(t, rfw.FlushAndFilter()) + + wantErrorJSON := string(rfw.errorResponseBody(jsonrpc2.Int64ID(99), + errors.New("dropped a frame carrying a result outside a clean Response"))) + + events := parseSSEStream(t, rr.Body.Bytes()) + require.Len(t, events, 1, "body: %q", rr.Body.String()) + assert.Empty(t, events[0].eventType, "MCP clients only process default/'message' events") + assert.JSONEq(t, wantErrorJSON, events[0].data) +} + +// TestParseSSEStream_RejectsBareJSONWithoutDataPrefix guards the guard: a +// bare JSON-RPC envelope with no "data:" prefix -- exactly what the pre-#6066 +// code wrote directly to the response body -- must never be recognized as a +// dispatched SSE event, with or without a trailing newline. If +// parseSSEStream ever loosened enough to accept bare JSON, this fails first. +func TestParseSSEStream_RejectsBareJSONWithoutDataPrefix(t *testing.T) { + t.Parallel() + + authorizer := newWeatherOnlyAuthorizer(t) + req := newUser1Request(t) + rfw := NewResponseFilteringWriter(httptest.NewRecorder(), authorizer, req, string(mcp.MethodToolsList), nil, nil) + + bareBody := rfw.errorResponseBody(jsonrpc2.Int64ID(1), errors.New("x")) + + testCases := []struct { + name string + body []byte + }{ + {name: "no trailing newline", body: bareBody}, + {name: "with trailing newline", body: append(append([]byte{}, bareBody...), '\n')}, + } + + for _, tc := range testCases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + events := parseSSEStream(t, tc.body) + assert.Empty(t, events, "a bare JSON envelope with no data: prefix must never dispatch as an SSE event") + }) + } +} From fefd7b46a9a7c61b0f9e5df17c1a3fed34a9ddb1 Mon Sep 17 00:00:00 2001 From: Jakub Hrozek Date: Tue, 28 Jul 2026 14:55:59 +0200 Subject: [PATCH 4/4] Match the client's SSE line model, not the spec's MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Panel review on #6088 found the rewrite parses less than the client that consumes these streams, and where the two disagree an upstream can put a list through the gap. go-sdk's scanEvents — used by ToolHive's own vMCP and mcpcompat stacks — reads lines with ReadBytes('\n'), so a lone CR is not a terminator, and it applies TrimSpace to each data value rather than to the assembled buffer. The scanner here did the opposite on both counts, following the WHATWG grammar instead. A bare CR inside a data payload therefore split a line for us and not for the client: our second piece lost its "data:" prefix, the assembled payload was truncated, nothing decoded, and the body went out verbatim with the list intact. main filtered that body, so this was a regression against it. Separately, one of the six whitespace bytes Go's JSON scanner rejects but unicode.IsSpace accepts, sitting at an interior value boundary, survived our single whole-buffer trim while the client stripped it per value. main leaked that one too. Scan on LF only and trim each value before joining. That parses a superset of what we saw before, so it can only decode and filter more, never less. A browser EventSource does treat a lone CR as a terminator and will mis-frame a pass-through body carrying one — but identically to how it would mis-frame the backend's own bytes, since the output stays byte-preserving, so the worst case is an unparseable payload rather than a disclosure. Under LF-only scanning a CRLF line's CR lives in the line text, not the terminator, so replacing a data line would have silently downgraded it to bare LF. The CR is re-attached when a line is replaced. The test oracle had the same bug as the code: it reimplemented WHATWG line splitting, so the suite agreed with the implementation by construction and could not have caught any of this. It now models the go-sdk client, which is the reading that decides whether a payload reaches a caller. Also from the review: the fail-closed envelope inherited the offending event's name, and go-sdk skips events not named "message", so on a named event the #6037 hang survived the fix that was meant to end it. And an upstream error carrying an explicit "result":null lost its real code to our generic envelope; a null result cannot carry a list, so that check now exempts it. Three comments asserted things about go-sdk that reading it disproves, including the one that made the trim granularity look safe. Corrected. Refs #6037 Co-Authored-By: Claude Opus 5 --- pkg/authz/response_filter.go | 246 +++++++++++++++++++----------- pkg/authz/response_filter_test.go | 179 +++++++++++++++++++++- 2 files changed, 336 insertions(+), 89 deletions(-) diff --git a/pkg/authz/response_filter.go b/pkg/authz/response_filter.go index c17108b9b5..027a38261a 100644 --- a/pkg/authz/response_filter.go +++ b/pkg/authz/response_filter.go @@ -211,10 +211,10 @@ func (rfw *ResponseFilteringWriter) filterAndEncode(response *jsonrpc2.Response) } // sseLine pairs one scanned SSE line with the terminator that followed it in -// the raw body: "\r\n", "\n", "\r", or nil for a final line with none. SSE -// permits mixing all three within one stream, so each line keeps the -// terminator it actually had rather than the body being forced onto one -// stream-wide convention. +// the raw body: "\n", or nil for a final line with none. Lines are split on +// LF only (see processSSEResponse), so a line that was actually terminated by +// "\r\n" keeps its trailing "\r" as the last byte of text -- text+term always +// reproduces exactly the bytes that were scanned. type sseLine struct { text []byte term []byte @@ -226,10 +226,22 @@ type sseLine struct { // single payload, and a blank line is structural — it dispatches the event // rather than being emittable content in its own right. // -// Lines are scanned one terminator at a time instead of sniffing a single -// separator convention for the whole body: sniffing disagrees with a client -// that (per spec) may see CR, LF, and CRLF mixed within one stream, and that -// disagreement is exactly the class of bug this rewrite exists to close. +// Lines are split on LF only, matching the client that actually consumes +// these streams (github.com/modelcontextprotocol/go-sdk's mcp/event.go, +// which ReadBytes('\n')s and never treats a lone CR as a line terminator) -- +// not the WHATWG EventSource grammar, which additionally splits on a bare CR. +// The two disagree on a body containing an interior "\r" (e.g. inside a +// data: payload): WHATWG would cut the line there, but the client reads it as +// one line and parses through. Following the WHATWG grammar here decoded a +// SUPERSET of what the client would treat as one line, letting the shorter, +// wrongly-split pieces slip past filtering undetected -- a real regression, +// not a hypothetical one. Splitting on LF only means we parse a superset of +// what a stricter reader would call one line, so we decode and filter *more* +// than before, never less: worst case for a client that does honor a lone CR +// (e.g. a browser EventSource) is a payload it mis-frames identically to how +// it would mis-frame the backend's raw bytes directly, since output stays +// byte-preserving -- denial, never disclosure. +// // Every line is re-emitted with the same terminator it was scanned with, so // the output is structurally identical to the input (minus a leading UTF-8 // BOM, stripped below); only `data:` payloads are rewritten. @@ -248,33 +260,34 @@ func (rfw *ResponseFilteringWriter) processSSEResponse(rawResponse []byte) error var outputLines []sseLine var event []sseLine for len(rawResponse) > 0 { - idx := bytes.IndexAny(rawResponse, "\r\n") + idx := bytes.IndexByte(rawResponse, '\n') var line, term []byte - switch { - case idx == -1: + if idx == -1 { line, term, rawResponse = rawResponse, nil, nil - case rawResponse[idx] == '\r' && idx+1 < len(rawResponse) && rawResponse[idx+1] == '\n': - line, term, rawResponse = rawResponse[:idx], rawResponse[idx:idx+2], rawResponse[idx+2:] - default: + } else { line, term, rawResponse = rawResponse[:idx], rawResponse[idx:idx+1], rawResponse[idx+1:] } - if len(line) == 0 { + // A line is structurally blank once its trailing CR (left over from + // a "\r\n" terminator the LF-only split doesn't consume) is ignored; + // without this a CRLF-terminated blank line would scan as the + // single byte "\r" and never be recognised as the event separator. + // The blank line's own bytes (that "\r", if present) are preserved + // in the output below rather than dropped, so structure stays + // byte-identical to the input. + if len(bytes.TrimRight(line, "\r")) == 0 { outputLines = append(outputLines, rfw.resolveSSEEvent(event)...) - outputLines = append(outputLines, sseLine{term: term}) + outputLines = append(outputLines, sseLine{text: line, term: term}) event = nil continue } event = append(event, sseLine{text: line, term: term}) } - // A trailing event with no closing blank line is never dispatched by a - // spec-compliant client, but the backend did send these bytes, so write - // them (filtered) rather than drop them — without fabricating a - // terminator the backend didn't send. Consequence: if that event needed - // filtering, the error envelope substituted here is never dispatched - // either, so the client hangs on its own timeout instead of seeing the - // error. Real backends terminate their events, so this is accepted, not - // fixed. + // A trailing event with no closing blank line: the go-sdk client (see + // mcp/event.go's yieldEvent, called once more after its read loop hits + // EOF) does still dispatch whatever is buffered at that point, so this + // is filtered like any other event, not dropped. We write these bytes + // (filtered) without fabricating a terminator the backend didn't send. outputLines = append(outputLines, rfw.resolveSSEEvent(event)...) for _, l := range outputLines { @@ -298,8 +311,40 @@ func (rfw *ResponseFilteringWriter) resolveSSEEvent(event []sseLine) []sseLine { return nil } + dataValues := eventDataValues(event) + if len(dataValues) == 0 { + return event + } + + // No whole-buffer trim needed here: the per-value trim in eventDataValues + // already leaves nothing but JSON-significant bytes at either edge of the join. + assembled := bytes.Join(dataValues, []byte("\n")) + if len(assembled) == 0 { + // A client never dispatches an empty data buffer, so there's nothing + // to classify or fail closed on. + return event + } + + replacement, failedClosed := rfw.filterSSEEventData(assembled) + if replacement == nil { + replacement, failedClosed = rfw.probeValuesForSmuggledResult(dataValues) + } + if replacement == nil { + return event + } + return rebuildEventWithPayload(event, replacement, failedClosed) +} + +// eventDataValues extracts one event's `data:` field values, in order, each +// trimmed exactly as the client trims it. +func eventDataValues(event []sseLine) [][]byte { var dataValues [][]byte for _, l := range event { + // A line ending in "\r\n" keeps that "\r" as the trailing byte of + // l.text (see sseLine), so trim it before matching the "data:" + // prefix -- mirroring the client, which strips trailing "\r\n" from + // every line before inspecting it (mcp/event.go). + trimmed := bytes.TrimRight(l.text, "\r") // A bare "data" line with no colon, or one with a space before the // colon ("data :foo"), doesn't match the "data:" prefix below, so both // are excluded from assembly here, unlike strict WHATWG grammar (which @@ -311,7 +356,7 @@ func (rfw *ResponseFilteringWriter) resolveSSEEvent(event []sseLine) []sseLine { // "data :foo", bytes.Cut on the raw line yields before == "data " (not // "data"), so the go-sdk's own line parser ignores it too, exactly // like us and strict WHATWG. - data, ok := bytes.CutPrefix(l.text, []byte("data:")) + data, ok := bytes.CutPrefix(trimmed, []byte("data:")) if !ok { continue } @@ -320,76 +365,96 @@ func (rfw *ResponseFilteringWriter) resolveSSEEvent(event []sseLine) []sseLine { // more (or a differently-positioned space) would corrupt a payload // split mid-string-literal across lines. data, _ = bytes.CutPrefix(data, []byte(" ")) + // TrimSpace each value independently, before joining, not on the + // assembled buffer afterward. Go's JSON scanner treats only SP, TAB, + // CR and LF as whitespace, but unicode.IsSpace also covers U+000B, + // U+000C, U+0085, U+00A0, U+1680 and U+3000 -- and the go-sdk client + // trims each data value with exactly this function before joining + // (mcp/event.go), not the whole buffer once at the end. A + // whole-buffer trim only catches one of those bytes sitting at the + // very front or back of the assembled payload; one sitting at an + // interior boundary between two data: lines of the SAME event + // survived undetected, because it was never at either edge of the + // joined buffer. Trimming here catches it regardless of position. + data = bytes.TrimSpace(data) dataValues = append(dataValues, data) } - if len(dataValues) == 0 { - return event - } + return dataValues +} - // TrimSpace here, not inside the decoders. Go's JSON scanner treats only - // SP, TAB, CR and LF as whitespace, but unicode.IsSpace also covers U+000B, - // U+000C, U+0085, U+00A0, U+1680 and U+3000 — and the go-sdk client trims - // its data buffer with exactly this function (mcp/event.go). Leaving those - // bytes on meant DecodeMessage and carriesResult both rejected the payload - // while the client happily parsed it, so the event fell through to the - // undecodable branch and the unfiltered list went out (#5257). - assembled := bytes.TrimSpace(bytes.Join(dataValues, []byte("\n"))) - if len(assembled) == 0 { - // A client never dispatches an empty data buffer, so there's nothing - // to classify or fail closed on. - return event - } - replacement := rfw.filterSSEEventData(assembled) - if replacement == nil { - // filterSSEEventData's own carriesResult scan stops at the first - // undecodable JSON value in the assembled (joined) payload, so a - // well-formed result-bearing value sitting right after a malformed - // one within the SAME event would otherwise ride through as - // "undecodable, nothing to filter" -- reopening the split-payload - // bypass this rewrite exists to close (#5257), just one line lower - // than before. Probe each data: value independently before - // conceding the event needs no filtering: this only ever turns a - // pass-through into a fail-closed error envelope, never the - // reverse, so it cannot itself reintroduce a bypass. - for _, v := range dataValues { - if carriesResult(v) { - slog.Warn("SSE event's assembled payload was undecodable but an individual data value carries a result; failing closed", - "method", rfw.method) - replacement = rfw.errorResponseBody(rfw.requestID(), - errors.New("dropped a frame carrying a result outside a clean Response")) - break - } - } - if replacement == nil { - return event +// probeValuesForSmuggledResult is the fallback for an event whose assembled +// payload was undecodable as a whole. filterSSEEventData's carriesResult scan +// stops at the first undecodable JSON value in the joined payload, so a +// well-formed result-bearing value sitting right after a malformed one within +// the SAME event would otherwise ride through as "undecodable, nothing to +// filter" -- reopening the split-payload bypass this rewrite exists to close +// (#5257), just one line lower than before. Probing each value independently +// only ever turns a pass-through into a fail-closed error envelope, never the +// reverse, so it cannot itself reintroduce a bypass. A nil replacement means +// no value carried a result and the event genuinely needs no filtering. +func (rfw *ResponseFilteringWriter) probeValuesForSmuggledResult(dataValues [][]byte) (replacement []byte, failedClosed bool) { + for _, v := range dataValues { + if carriesResult(v) { + slog.Warn("SSE event's assembled payload was undecodable but an individual data value carries a result; failing closed", + "method", rfw.method) + return rfw.errorResponseBody(rfw.requestID(), + errors.New("dropped a frame carrying a result outside a clean Response")), true } } + return nil, false +} +// rebuildEventWithPayload re-emits one event's lines with replacement standing +// in for the first `data:` line's value, dropping the later `data:` lines whose +// content is already folded into it. +func rebuildEventWithPayload(event []sseLine, replacement []byte, failedClosed bool) []sseLine { out := make([]sseLine, 0, len(event)) replaced := false for _, l := range event { - if _, ok := bytes.CutPrefix(l.text, []byte("data:")); ok { - if replaced { - // Subsequent data: lines are already folded into the - // assembled payload above; only the first line carries it. - continue - } - // jsonrpc2.EncodeMessage and json.Marshal both escape newlines, - // so replacement can never contain a raw line separator and is - // always safe to emit as a single data: line. - out = append(out, sseLine{text: append([]byte("data: "), replacement...), term: l.term}) - replaced = true + trimmed := bytes.TrimRight(l.text, "\r") + // The go-sdk streamable client only dispatches unnamed ("message") + // events, so a named event's own "event:" field would silently + // swallow the fail-closed envelope substituted below, reproducing + // the #6037 hang. Drop it: the substituted payload is either the + // error envelope itself or a freshly filtered result, neither of + // which the original event name describes any more. + if failedClosed && bytes.HasPrefix(trimmed, []byte("event:")) { + continue + } + if !bytes.HasPrefix(trimmed, []byte("data:")) { + out = append(out, l) continue } - out = append(out, l) + if replaced { + // Subsequent data: lines are already folded into the assembled + // payload; only the first line carries it. + continue + } + // A "\r\n"-terminated line keeps that "\r" as l.text's trailing + // byte (see sseLine), not as part of l.term. Since the replacement + // text discards l.text entirely, re-attach any such trailing "\r" to + // the terminator here -- otherwise a CRLF-terminated data line would + // silently downgrade to bare LF once filtered, while an untouched + // (pass-through) line keeps its CRLF because it reuses l.text verbatim. + term := l.term + if cr := l.text[len(trimmed):]; len(cr) > 0 { + term = append(append([]byte{}, cr...), term...) + } + // jsonrpc2.EncodeMessage and json.Marshal both escape newlines, so + // replacement can never contain a raw line separator and is always + // safe to emit as a single data: line. + out = append(out, sseLine{text: append([]byte("data: "), replacement...), term: term}) + replaced = true } return out } // filterSSEEventData classifies an event's assembled data payload and returns -// the payload to emit in its place. A nil replacement means the event's lines -// pass through unchanged. -func (rfw *ResponseFilteringWriter) filterSSEEventData(data []byte) []byte { +// the payload to emit in its place, plus whether that payload is a fail-closed +// error envelope (as opposed to a successfully filtered result). A nil +// replacement means the event's lines pass through unchanged, and failedClosed +// is meaningless in that case. +func (rfw *ResponseFilteringWriter) filterSSEEventData(data []byte) (replacement []byte, failedClosed bool) { message, decodeErr := jsonrpc2.DecodeMessage(data) response, isResponse := message.(*jsonrpc2.Response) switch { @@ -399,9 +464,9 @@ func (rfw *ResponseFilteringWriter) filterSSEEventData(data []byte) []byte { // errorResponseBody logs ferr itself, so don't repeat it here. slog.Warn("emitting a JSON-RPC error in place of an SSE frame that failed to filter", "method", rfw.method) - return rfw.errorResponseBody(response.ID, ferr) + return rfw.errorResponseBody(response.ID, ferr), true } - return filteredData + return filteredData, false case carriesResult(data): // The frame is not a clean Response but still carries a result. This // covers a non-Response type (a request/notification frame smuggling a @@ -416,19 +481,19 @@ func (rfw *ResponseFilteringWriter) filterSSEEventData(data []byte) []byte { slog.Warn("SSE event carried a result outside a clean Response frame; failing closed as a protocol violation", "method", rfw.method) return rfw.errorResponseBody(rfw.requestID(), - errors.New("dropped a frame carrying a result outside a clean Response")) + errors.New("dropped a frame carrying a result outside a clean Response")), true case decodeErr != nil: // Genuinely undecodable and no smuggled result. Pass through unfiltered. slog.Warn("SSE event data could not be decoded as JSON-RPC; passing through unfiltered", "method", rfw.method, "error", decodeErr) - return nil + return nil, false default: // Genuine non-Response frame (e.g. an interleaved notifications/* // message) with no result payload. Routine SSE traffic, so log at // Debug to keep the suspicious branches above from being buried. slog.Debug("SSE event data was not a JSON-RPC Response; passing through unfiltered", "method", rfw.method) - return nil + return nil, false } } @@ -507,7 +572,7 @@ func valueCarriesResult(value json.RawMessage) bool { // sseCarriesResult reports whether rawResponse contains an SSE "data:" line // whose payload carries a JSON-RPC result. It is a lightweight detector, used // only to decide whether a 2xx body with an unrecognized media type needs the -// full SSE processing path (which applies its own per-line filtering and +// full SSE processing path (which applies its own event-based filtering and // fail-closed rules); it mirrors sniffSSEToolsList in pkg/mcp/tool_filter.go. func sseCarriesResult(rawResponse []byte) bool { normalized := bytes.ReplaceAll(rawResponse, []byte("\r\n"), []byte("\n")) @@ -541,10 +606,17 @@ func (rfw *ResponseFilteringWriter) filterListResponse(response *jsonrpc2.Respon // unexpected "error" key and treat it as a successful result. Fail // closed rather than passing the smuggled list through under cover // of the error field. See #5257. - if response.Result != nil { + // A literal `"result":null` is exempted: DecodeMessage sets Result to + // the non-nil 4 bytes "null" whenever the key is present at all (see + // objectCarriesResult), so without this a legitimate upstream error + // that happens to carry an explicit null result would lose its own + // code (e.g. -32601) to our generic internal-error envelope for no + // security benefit -- a null result can never carry a list, so + // failing closed on it buys nothing. + if response.Result != nil && !bytes.Equal(bytes.TrimSpace(response.Result), []byte("null")) { return nil, errors.New("response carried both error and result") } - // If there's an error and no result, don't filter + // If there's an error and no result (or an explicit null result), don't filter return response, nil } diff --git a/pkg/authz/response_filter_test.go b/pkg/authz/response_filter_test.go index 629f7885a0..f09cc8c3d7 100644 --- a/pkg/authz/response_filter_test.go +++ b/pkg/authz/response_filter_test.go @@ -1308,6 +1308,54 @@ func TestResponseFilteringWriter_SSE_DisguisedResponseFrame(t *testing.T) { } } +// TestResponseFilteringWriter_SSE_FailClosedDropsEventName is a regression +// test for #6037: the go-sdk streamable client only dispatches unnamed +// ("message"-typed) events to its MCP request handler, so substituting a +// fail-closed error envelope in place of a NAMED event's data (while leaving +// its "event:" field untouched) meant the envelope was never delivered and +// the caller hung on its own timeout -- the very hang the #6037 envelope was +// meant to prevent. This asserts what parseSSEStream (the go-sdk client +// model, not the WHATWG grammar) actually dispatches, since that's what +// determines delivery. +func TestResponseFilteringWriter_SSE_FailClosedDropsEventName(t *testing.T) { + t.Parallel() + + authorizer := newWeatherOnlyAuthorizer(t) + parsedReq := newParsedUser1Request(t, `{"jsonrpc":"2.0","id":99,"method":"tools/list"}`) + + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, authorizer, parsedReq, string(mcp.MethodToolsList), nil, nil) + rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") + + // A method+result frame that fails closed (see + // TestResponseFilteringWriter_SSE_DisguisedResponseFrame), framed under a + // custom event name the way a real backend might tag a tool-list push. + body := "event: toolsUpdate\n" + + `data: {"jsonrpc":"2.0","method":"notifications/initialized","id":1,` + + `"result":{"tools":[{"name":"admin_tool"}]}}` + "\n\n" + _, err := rfw.Write([]byte(body)) + require.NoError(t, err) + require.NoError(t, rfw.FlushAndFilter()) + + out := rr.Body.String() + assert.NotContains(t, out, "admin_tool", "smuggled result leaked past the filter") + assert.NotContains(t, out, "event:", + "the offending event's event: field must be dropped, or the go-sdk client never dispatches the fail-closed envelope") + + events := parseSSEStream(t, rr.Body.Bytes()) + require.Len(t, events, 1, "body: %q", out) + assert.Empty(t, events[0].eventType, + "the go-sdk client only dispatches unnamed/'message' events to its MCP handler; a named event here reproduces the #6037 hang") + + msg, err := jsonrpc2.DecodeMessage([]byte(events[0].data)) + require.NoError(t, err, "the dispatched envelope must itself be valid JSON-RPC") + resp, ok := msg.(*jsonrpc2.Response) + require.True(t, ok, "the dispatched envelope must be a clean Response") + require.NotNil(t, resp.Error, "the event must fail closed into an error envelope") + assert.Equal(t, jsonrpc2.Int64ID(99), resp.ID, + "the error envelope must carry the request's id so the client can correlate it") +} + // TestResponseFilteringWriter_SSE_ConcatenatedEventBypass is a regression test // for a #5257-class leak the event-based rewrite itself introduced: two // data: fields in ONE event (a notification followed by a result-bearing @@ -1477,6 +1525,16 @@ func TestResponseFilteringWriter_SSE_ErrorAndResultBypass(t *testing.T) { body: `data: {"jsonrpc":"2.0","id":1,"error":{"code":1,"message":"x"}}` + "\n\n", wantErrorCode: 1, // the upstream's own error, untouched }, + { + // A literal `"result":null` is exempted from the both-error- + // and-result fail-closed rule (see filterListResponse): a null + // result can never carry a list, so failing closed on it only + // destroys the upstream's real error code for no security + // benefit. Regression for that exemption. + name: "error with explicit null result passes through unfiltered", + body: `data: {"jsonrpc":"2.0","id":1,"error":{"code":404,"message":"not found"},"result":null}` + "\n\n", + wantErrorCode: 404, // the upstream's own error, untouched + }, } for _, tc := range testCases { @@ -2099,6 +2157,13 @@ func TestResponseFilteringWriter_SSE_LeadingNonJSONWhitespaceStillFiltered(t *te // carriesResult on the whole joined payload does too, so both lines went out // unfiltered. The fix probes each data: value independently before // conceding the event needs no filtering. +// +// NotContains(admin_tool) alone would also be satisfied by the event being +// dropped silently (the #6037 hang), so this also requires the fail-closed +// error envelope to actually go out, correlated to the request's real id via +// newParsedUser1Request -- an unparsed request would let requestID() fall +// back to the zero ID and this test would accept an envelope no client could +// actually correlate. func TestResponseFilteringWriter_SSE_PerValueProbeCatchesConcatenatedGarbage(t *testing.T) { t.Parallel() @@ -2138,9 +2203,9 @@ func TestResponseFilteringWriter_SSE_PerValueProbeCatchesConcatenatedGarbage(t * "test fixture assumption broken: the whole-payload scan must not itself catch the smuggled result "+ "-- otherwise this test would not exercise the per-value probe at all") - req := newUser1Request(t) + parsedReq := newParsedUser1Request(t, `{"jsonrpc":"2.0","id":99,"method":"tools/list"}`) rr := httptest.NewRecorder() - rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) + rfw := NewResponseFilteringWriter(rr, authorizer, parsedReq, string(mcp.MethodToolsList), nil, nil) rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") // One event, two data: lines with no blank line between them, so @@ -2154,6 +2219,116 @@ func TestResponseFilteringWriter_SSE_PerValueProbeCatchesConcatenatedGarbage(t * out := rr.Body.String() assert.NotContains(t, out, "admin_tool", "a result-bearing data: line following a malformed one in the same event must not bypass the filter") + + firstLine := strings.TrimSuffix(strings.SplitN(out, "\n", 2)[0], "\r") + payload := strings.TrimPrefix(firstLine, "data: ") + msg, decErr := jsonrpc2.DecodeMessage([]byte(payload)) + require.NoError(t, decErr, "the fail-closed envelope must itself be valid JSON-RPC") + resp, ok := msg.(*jsonrpc2.Response) + require.True(t, ok, "the fail-closed envelope must be a clean Response") + require.NotNil(t, resp.Error, "the event must fail closed into an error envelope, not vanish silently") + assert.Equal(t, jsonrpc2.Int64ID(99), resp.ID, + "the error envelope must carry the request's id so the client can correlate it, or the #6037 hang reproduces") + }) + } +} + +// TestResponseFilteringWriter_SSE_InteriorBareCRNotALineBreak is a regression +// test for a client-divergence bypass: the client that actually consumes +// these streams (github.com/modelcontextprotocol/go-sdk's mcp/event.go) +// splits lines on LF only, so a bare "\r" sitting inside a single data: +// payload is insignificant JSON whitespace to it, not a line break. Splitting +// on "\r" as well (as strict WHATWG grammar does) cuts the line there +// instead, leaving a "data:"-less second half that contributes nothing to +// the assembled payload -- so the first half (truncated, undecodable) is +// classified as carrying no result and re-emitted verbatim, leaking +// admin_tool. NotContains(admin_tool) alone would also pass if the whole +// event were merely dropped, so this also requires weather to survive, +// proving the event was filtered rather than eaten. +func TestResponseFilteringWriter_SSE_InteriorBareCRNotALineBreak(t *testing.T) { + t.Parallel() + + authorizer := newWeatherOnlyAuthorizer(t) + req := newUser1Request(t) + + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) + rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") + + // A bare "\r" sits between "id":1, and "result", inside what a + // non-conformant-to-WHATWG (but client-accurate) reader treats as ONE + // data: line. "\r" is legal JSON whitespace between a comma and the next + // key, so the client parses this payload intact. + body := "data: {\"jsonrpc\":\"2.0\",\"id\":1,\r\"result\":" + + `{"tools":[{"name":"weather"},{"name":"admin_tool"}]}}` + "\n\n" + + _, err := rfw.Write([]byte(body)) + require.NoError(t, err) + require.NoError(t, rfw.FlushAndFilter()) + + out := rr.Body.String() + assert.NotContains(t, out, "admin_tool", + "a bare CR inside a single data: payload must not split the line and truncate it past the filter") + assert.Contains(t, out, "weather", + "the authorized tool must survive filtering, proving the event was filtered rather than dropped") +} + +// TestResponseFilteringWriter_SSE_InteriorExoticWhitespaceAcrossDataLines is a +// regression test for a client-divergence bypass distinct from the leading- +// whitespace case above: the go-sdk client TrimSpaces each data: value +// independently before joining them with LF (mcp/event.go), but a single +// whole-buffer trim (applied once, after joining) only reaches the very +// front and back of the assembled payload. An exotic unicode.IsSpace byte +// (one of U+000B, U+000C, U+0085, U+00A0, U+1680, U+3000 -- whitespace to the +// client's trim but not to Go's own JSON scanner) sitting at the boundary +// BETWEEN two data: lines of the same event survives a whole-buffer trim, +// because it's never at either edge of the joined result. That leaves an +// illegal byte mid-token for our decoder while the client's own per-value +// trim removes it before joining, so the client parses fine while ours +// rejects the payload as undecodable and (pre-fix) re-emitted it raw. +func TestResponseFilteringWriter_SSE_InteriorExoticWhitespaceAcrossDataLines(t *testing.T) { + t.Parallel() + + exoticBytes := []struct { + name string + suffix string + }{ + {name: "U+000B line tabulation", suffix: "\u000B"}, + {name: "U+000C form feed", suffix: "\u000C"}, + {name: "U+0085 next line", suffix: "\u0085"}, + {name: "U+00A0 no-break space", suffix: "\u00A0"}, + {name: "U+1680 ogham space mark", suffix: "\u1680"}, + {name: "U+3000 ideographic space", suffix: "\u3000"}, + } + + for _, tc := range exoticBytes { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + authorizer := newWeatherOnlyAuthorizer(t) + req := newUser1Request(t) + + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) + rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") + + // Two data: lines, no blank line between them, so they belong to + // the SAME event and their values concatenate. The exotic byte + // trails the first value, sitting exactly at the interior join + // boundary once assembled -- never at the front or back of the + // whole buffer. + body := `data: {"jsonrpc":"2.0","id":1,` + tc.suffix + "\n" + + `data: "result":{"tools":[{"name":"weather"},{"name":"admin_tool"}]}}` + "\n\n" + + _, err := rfw.Write([]byte(body)) + require.NoError(t, err) + require.NoError(t, rfw.FlushAndFilter()) + + out := rr.Body.String() + assert.NotContains(t, out, "admin_tool", + "exotic whitespace at an interior data: value boundary must not bypass the filter") + assert.Contains(t, out, "weather", + "the authorized tool must survive filtering, proving the event was filtered rather than dropped") }) } }