diff --git a/pkg/authz/response_filter.go b/pkg/authz/response_filter.go index 4d41ca1dcc..027a38261a 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) @@ -203,62 +210,91 @@ 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: "\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 +} + +// 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 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. 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.IndexByte(rawResponse, '\n') + var line, term []byte + if idx == -1 { + line, term, rawResponse = rawResponse, nil, nil + } else { + 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) - } + // 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{text: line, term: term}) + event = nil + continue } + event = append(event, sseLine{text: line, term: term}) + } + // 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)...) - _, 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 +302,198 @@ 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 + } + + 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 + // 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(trimmed, []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(" ")) + // 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) + } + return dataValues +} + +// 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 { + 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 + } + 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, 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 { 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), true } - return true, nil + 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 // 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(), + errors.New("dropped a frame carrying a result outside a clean Response")), true 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, 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. 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, false } } @@ -334,26 +513,55 @@ 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. 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 { - 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 } 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 '[': - 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 + } + if objectCarriesResult(elTrimmed) { return true } } @@ -364,7 +572,7 @@ func carriesResult(data []byte) 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")) @@ -377,10 +585,38 @@ 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 { - // 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. + // 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 (or an explicit null result), don't filter return response, nil } @@ -614,33 +850,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..f09cc8c3d7 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" @@ -23,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" @@ -867,33 +869,170 @@ 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, -// leaking the unfiltered list past Cedar. It must instead pass only the -// offending line 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) { +// 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() - 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");`, + 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 +// filter previously wrote the entire raw upstream payload and returned, +// leaking the unfiltered list past Cedar. It must instead pass only the +// 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_PerEventFallthrough(t *testing.T) { + t.Parallel() + + 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. @@ -1013,16 +1152,21 @@ func TestResponseFilteringWriter_SSE_PerLineFallthrough(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) rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") - body := strings.Join([]string{plc.line, mc.respLine, ""}, "\n") - _, err = rfw.Write([]byte(body)) + // 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) require.NoError(t, rfw.FlushAndFilter()) @@ -1059,6 +1203,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,23 +1218,21 @@ 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() - 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{ @@ -1099,7 +1248,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 +1258,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,16 +1269,20 @@ 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) - 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. + 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") - body := strings.Join([]string{f.payload, "data: " + string(realResp), ""}, "\n") - _, err = rfw.Write([]byte(body)) + // 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 +1290,280 @@ 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_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 +// 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 := newWeatherOnlyAuthorizer(t) + + notificationLine := `data: {"jsonrpc":"2.0","method":"notifications/message","params":{}}` + resultLine := `data: {"jsonrpc":"2.0","id":1,"result":{"tools":[{"name":"admin_tool"}]}}` + + // 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) + 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 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 +// 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 := 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") + + 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 := 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") + + 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") + 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 +// 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 — 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 := newWeatherOnlyAuthorizer(t) + + 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 + }, + { + // 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 { + 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") + + _, 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 (mcpparser.CodeInternalError, our envelope) from pass-through (the upstream's own code)") }) } } @@ -1149,18 +1579,7 @@ func TestResponseFilteringWriter_SSE_DisguisedResponseFrame(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. @@ -1197,27 +1616,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()) @@ -1243,6 +1653,309 @@ 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 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 { + 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() + + 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 + 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 the fixed generic string, not the wrapped error") + assert.NotContains(t, string(body), wrapped.Error(), + "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") + 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 := 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) + + 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 := 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(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 := newWeatherOnlyAuthorizer(t) + + 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 := 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(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 := newWeatherOnlyAuthorizer(t) + + // 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 := 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)) + 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_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 @@ -1368,83 +2081,433 @@ 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) { +// 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() - wrapped := errors.New("leaky-tool-name-sentinel") + 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 - id jsonrpc2.ID - wantID any // nil means the id key must be absent + prefix string }{ - {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{}}, + {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, nil, nil, string(mcp.MethodToolsList), nil, nil) - body := rfw.errorResponseBody(tc.id, wrapped) + rfw := NewResponseFilteringWriter(rr, authorizer, req, string(mcp.MethodToolsList), nil, nil) + rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") - var decoded map[string]any - require.NoError(t, json.Unmarshal(body, &decoded)) + body := "data: " + tc.prefix + string(encoded) + "\n\n" + _, err := rfw.Write([]byte(body)) + require.NoError(t, err) + require.NoError(t, rfw.FlushAndFilter()) - assert.Equal(t, "2.0", decoded["jsonrpc"]) + 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") + }) + } +} - 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") +// 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. +// +// 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() - 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) - } + 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") + + 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") + + // 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") + + 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") }) } } -// 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) { +// 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() - // writeSSEErrorLine only reads rfw.ResponseWriter; the authorizer, request, - // and Content-Type are irrelevant on this path, so nils are safe here. + authorizer := newWeatherOnlyAuthorizer(t) + parsedReq := newParsedUser1Request(t, `{"jsonrpc":"2.0","id":99,"method":"tools/list"}`) + rr := httptest.NewRecorder() - rfw := NewResponseFilteringWriter(rr, nil, nil, 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 := `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"))) - id := jsonrpc2.Int64ID(9) - wrapped := errors.New("boom") - require.NoError(t, rfw.writeSSEErrorLine(id, wrapped)) + 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) - 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") + 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") + }) + } }