diff --git a/pkg/authz/response_filter.go b/pkg/authz/response_filter.go index 7402fa9e3f..e298de46fe 100644 --- a/pkg/authz/response_filter.go +++ b/pkg/authz/response_filter.go @@ -192,7 +192,7 @@ func (rfw *ResponseFilteringWriter) processSSEResponse(rawResponse []byte) error var written bool if data, ok := bytes.CutPrefix(line, []byte("data:")); ok { - w, stop, err := rfw.writeSSEDataLine(data) + w, stop, err := rfw.writeSSEDataLine(data, linesep) if stop { return err } @@ -228,20 +228,22 @@ func (rfw *ResponseFilteringWriter) processSSEResponse(rawResponse []byte) error // 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. stop reports that -// processSSEResponse should return err immediately (an error response has -// already been written on the filter/encode error paths). -func (rfw *ResponseFilteringWriter) writeSSEDataLine(data []byte) (written, stop bool, err error) { +// processSSEResponse should return err immediately (a self-terminated SSE +// error event has already been written on the filter/encode error paths, so +// skipping the loop's later line-separator reconciliation is safe). linesep +// is the stream's detected line separator, used to frame that error event. +func (rfw *ResponseFilteringWriter) writeSSEDataLine(data, linesep []byte) (written, stop bool, err error) { message, decodeErr := jsonrpc2.DecodeMessage(data) response, isResponse := message.(*jsonrpc2.Response) switch { case isResponse: filteredResponse, ferr := rfw.filterListResponse(response) if ferr != nil { - return false, true, rfw.writeErrorResponse(response.ID, ferr) + return false, true, rfw.writeSSEError(response.ID, ferr, linesep) } filteredData, ferr := jsonrpc2.EncodeMessage(filteredResponse) if ferr != nil { - return false, true, rfw.writeErrorResponse(response.ID, ferr) + return false, true, rfw.writeSSEError(response.ID, ferr, linesep) } // The caller writes linesep after this line, so do not append a // terminator here. A hardcoded "\n" desyncs \r\n streams. @@ -528,18 +530,18 @@ func (rfw *ResponseFilteringWriter) filterResourcesResponse(response *jsonrpc2.R return filteredResponse, nil } -// writeErrorResponse logs the full filtering error server-side and sends the client a -// generic message. err can originate in policy evaluation and name tools or -// resources, so security.md forbids returning it verbatim. -func (rfw *ResponseFilteringWriter) writeErrorResponse(id jsonrpc2.ID, err error) error { - slog.Error("error filtering response", "method", rfw.method, "error", err) +// filterErrorBody renders the JSON-RPC error response body for a filtering +// failure: the standard Internal Error code and a generic message. The +// underlying error is deliberately NOT included — it can originate in policy +// evaluation and name tools or resources, so security.md forbids returning it +// verbatim; callers log it server-side instead. The original request's id is +// included when recoverable so the client can correlate the error. +func filterErrorBody(id jsonrpc2.ID) []byte { errorResponse := &jsonrpc2.Response{ ID: id, Error: jsonrpc2.NewError(mcpparser.CodeInternalError, "internal error"), } - // Encode before writing any header, so an encode failure never leaves a - // half-written response. body, encErr := jsonrpc2.EncodeMessage(errorResponse) if encErr != nil { // Unreachable in practice: errorResponse is always a well-formed @@ -548,12 +550,70 @@ func (rfw *ResponseFilteringWriter) writeErrorResponse(id jsonrpc2.ID, err error slog.Error("failed to encode JSON-RPC error response", "error", encErr) body = fmt.Appendf(nil, `{"jsonrpc":"2.0","error":{"code":%d,"message":"internal error"}}`, mcpparser.CodeInternalError) } + return body +} +// writeErrorResponse writes a filtering-failure error on the +// application/json path: HTTP 500 with the JSON-RPC error as the body. Do +// NOT use it on the SSE path — a bare JSON object written into a +// text/event-stream parses as an unrecognized field under the WHATWG +// event-stream grammar and is silently discarded, leaving the client to hang +// to its own timeout; use writeSSEError there instead (#6037). +func (rfw *ResponseFilteringWriter) writeErrorResponse(id jsonrpc2.ID, err error) error { + // Full error server-side only; the client gets the generic body. + slog.Error("error filtering response", "method", rfw.method, "error", err) + // The body is built before writing any header, so an encode failure never + // leaves a half-written response. + body := filterErrorBody(id) rfw.ResponseWriter.WriteHeader(http.StatusInternalServerError) _, writeErr := rfw.ResponseWriter.Write(body) return writeErr } +// writeSSEError delivers a filtering failure on the SSE path as a properly +// framed event, so the client actually receives the JSON-RPC error instead +// of hanging: the stream then ends after a complete event, which is the +// spec's shape for a request that failed after the stream opened (the SSE +// stream carries the JSON-RPC response for the POSTed request; an error +// response is a response). +// +// Unlike writeErrorResponse, this writes no status header: by the time a +// data line fails filtering, earlier stream bytes and the proxy's Flush have +// typically already committed the 200/text/event-stream header, so the old +// WriteHeader(500) here was a usually-no-op that Go logged as superfluous — +// and even when it did land, a JSON body under an event-stream Content-Type +// was unreadable by any SSE client (#6037). +// +// Framing, per the WHATWG event-stream grammar: +// - a leading blank line first terminates any partially written preceding +// event, so this error is dispatched standalone; at an event boundary the +// blank line is a no-op (an empty data buffer dispatches nothing); +// - the payload rides a "data: " field line; +// - a trailing blank line dispatches the event. The frame is fully +// self-terminated because the caller stops stream processing (and its +// line-separator reconciliation) immediately after this write. +// +// Returns nil once the event is written: the failure has been delivered to +// the client in-band, and is logged here for server-side visibility. +func (rfw *ResponseFilteringWriter) writeSSEError(id jsonrpc2.ID, ferr error, linesep []byte) error { + // Full error server-side only (matching writeErrorResponse's scrub); the + // client's event carries the generic body. + slog.Error("delivering response-filtering failure to the SSE client as an error event", + "method", rfw.method, "error", ferr) + + body := filterErrorBody(id) + frame := make([]byte, 0, len("data: ")+len(body)+3*len(linesep)) + frame = append(frame, linesep...) + frame = append(frame, "data: "...) + frame = append(frame, body...) + frame = append(frame, linesep...) + frame = append(frame, linesep...) + if _, werr := rfw.ResponseWriter.Write(frame); werr != nil { + return fmt.Errorf("%w: %w", errBug, werr) + } + 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 3ffd15197e..5ee81480c1 100644 --- a/pkg/authz/response_filter_test.go +++ b/pkg/authz/response_filter_test.go @@ -6,6 +6,7 @@ package authz import ( "context" "encoding/json" + "errors" "io" "net/http" "net/http/httptest" @@ -1239,3 +1240,203 @@ func TestResponseFilteringWriter_JSON_DisguisedResponseFrame(t *testing.T) { }) } } + +// sseEvent is one event dispatched by parseSSEStream. +type sseEvent struct { + eventType string + data string +} + +// parseSSEStream interprets body per the WHATWG HTML "server-sent events" +// event-stream grammar (§"Interpreting an event stream"): lines end at CRLF, +// LF, or CR; a line starting with ':' is a comment; a line containing ':' +// splits into a field name and value (one leading space of the value is +// stripped); a line without ':' is a field with an empty value. Only +// recognized fields have any effect — "data" appends value+"\n" to the data +// buffer, "event" sets the event type — and an UNRECOGNIZED field is ignored +// outright. A blank line dispatches an event only when the data buffer is +// non-empty, and pending data left at end-of-stream is discarded. +// +// The strictness is the point of these tests: a bare JSON object written +// into a stream is one giant unrecognized field ('{"jsonrpc"' up to the +// first colon) and yields no event at all — the silent client hang of #6037 +// — so a conformant-looking payload is not evidence of delivery. +func parseSSEStream(t *testing.T, body []byte) []sseEvent { + t.Helper() + + normalized := strings.ReplaceAll(string(body), "\r\n", "\n") + normalized = strings.ReplaceAll(normalized, "\r", "\n") + lines := strings.Split(normalized, "\n") + // A trailing line terminator makes Split produce a final "" that is not a + // blank LINE (a blank line is two consecutive terminators); drop it so an + // unterminated final event is correctly left undispatched. + if len(lines) > 0 && lines[len(lines)-1] == "" { + lines = lines[:len(lines)-1] + } + + var events []sseEvent + var dataBuf strings.Builder + var eventType string + for _, line := range lines { + switch { + case line == "": + if dataBuf.Len() > 0 { + events = append(events, sseEvent{ + eventType: eventType, + data: strings.TrimSuffix(dataBuf.String(), "\n"), + }) + } + dataBuf.Reset() + eventType = "" + case strings.HasPrefix(line, ":"): + // Comment; ignored. + default: + name, value, hasColon := strings.Cut(line, ":") + if hasColon { + value = strings.TrimPrefix(value, " ") + } else { + value = "" + } + switch name { + case "data": + dataBuf.WriteString(value) + dataBuf.WriteString("\n") + case "event": + eventType = value + default: + // Unrecognized field: ignored per the grammar. A bare JSON + // line lands here. + } + } + } + return events +} + +// newSSEFilteringWriter builds a ResponseFilteringWriter over a recorder +// whose response is already committed as an SSE stream, the state +// writeSSEError runs in. +func newSSEFilteringWriter(t *testing.T) (*httptest.ResponseRecorder, *ResponseFilteringWriter) { + t.Helper() + req := httptest.NewRequest(http.MethodPost, "/messages", nil) + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, nil, req, string(mcp.MethodToolsList), nil, nil) + rfw.ResponseWriter.Header().Set("Content-Type", "text/event-stream") + return rr, rfw +} + +// TestWriteSSEErrorIsParseableByAnSSEClient is the delivery guard for #6037: +// the filtering-failure error on the SSE path must arrive as a dispatched, +// default-typed SSE event whose data is the conformant JSON-RPC error — run +// through a WHATWG-grammar parser, because conformant bytes that no SSE +// client can see (the pre-fix behavior) must fail this test. +func TestWriteSSEErrorIsParseableByAnSSEClient(t *testing.T) { + t.Parallel() + + testCases := []struct { + name string + id jsonrpc2.ID + linesep string + wantData string + }{ + { + name: "LF stream with int64 id", + id: jsonrpc2.Int64ID(42), + linesep: "\n", + wantData: `{"jsonrpc":"2.0","id":42,"error":{"code":-32603,"message":"internal error"}}`, + }, + { + name: "CRLF stream with string id", + id: jsonrpc2.StringID("abc"), + linesep: "\r\n", + wantData: `{"jsonrpc":"2.0","id":"abc","error":{"code":-32603,"message":"internal error"}}`, + }, + { + // An unrecoverable id omits the key entirely (never "id":null), + // matching the empty-ID encoding asserted elsewhere (#6038). + name: "empty id omits the id key", + id: jsonrpc2.ID{}, + linesep: "\n", + wantData: `{"jsonrpc":"2.0","error":{"code":-32603,"message":"internal error"}}`, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + rr, rfw := newSSEFilteringWriter(t) + require.NoError(t, rfw.writeSSEError(tc.id, errors.New("boom"), []byte(tc.linesep))) + + events := parseSSEStream(t, rr.Body.Bytes()) + require.Len(t, events, 1, "the error must arrive as exactly one dispatched SSE event; body: %q", rr.Body.String()) + ev := events[0] + assert.Empty(t, ev.eventType, + "the error event must be default-typed: MCP clients only process events with no type or type 'message'") + assert.JSONEq(t, tc.wantData, ev.data) + + var decoded map[string]json.RawMessage + require.NoError(t, json.Unmarshal([]byte(ev.data), &decoded)) + _, hasID := decoded["id"] + assert.Equal(t, tc.id != jsonrpc2.ID{}, hasID, "id key presence mismatch") + + // No mid-stream status rewrite: the old WriteHeader(500) was a + // usually-no-op on a committed stream (and a recorder pins it + // reliably); the in-band JSON-RPC error rides the 200 stream. + assert.Equal(t, http.StatusOK, rr.Code, + "writeSSEError must not attempt a status rewrite mid-stream") + }) + } +} + +// TestWriteSSEErrorAfterPartialEvent covers the truncation interaction from +// #6037: if the failure lands while a previous event is partially written +// (field lines emitted, no dispatching blank line yet), the error frame's +// leading separator must first terminate that partial event so the error +// still arrives as its own cleanly parseable event, rather than being +// concatenated into the partial one's data buffer. +func TestWriteSSEErrorAfterPartialEvent(t *testing.T) { + t.Parallel() + + const wantData = `{"jsonrpc":"2.0","id":7,"error":{"code":-32603,"message":"internal error"}}` + + testCases := []struct { + name string + partial string + }{ + {name: "after a typed field line", partial: "event: message\n"}, + {name: "after an undispatched data line", partial: "data: {\"jsonrpc\":\"2.0\",\n"}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + rr, rfw := newSSEFilteringWriter(t) + _, err := rr.WriteString(tc.partial) + require.NoError(t, err) + require.NoError(t, rfw.writeSSEError(jsonrpc2.Int64ID(7), errors.New("boom"), []byte("\n"))) + + events := parseSSEStream(t, rr.Body.Bytes()) + require.NotEmpty(t, events, "the error event must be dispatched; body: %q", rr.Body.String()) + last := events[len(events)-1] + assert.JSONEq(t, wantData, last.data, + "the error must parse standalone, not merged into the partial event's data") + }) + } +} + +// TestBareJSONInEventStreamYieldsNoSSEEvent documents the #6037 failure mode +// and validates parseSSEStream's strictness (guarding the guard): the +// pre-fix writeErrorResponse wrote exactly these bytes into the stream, and +// under the event-stream grammar they parse as an unrecognized field and are +// discarded — zero events, client hangs. If parseSSEStream ever loosens +// enough to accept this shape, the delivery guard above stops guarding. +func TestBareJSONInEventStreamYieldsNoSSEEvent(t *testing.T) { + t.Parallel() + + body := filterErrorBody(jsonrpc2.Int64ID(1)) + assert.Empty(t, parseSSEStream(t, body), + "a bare JSON object must not parse as an SSE event") + assert.Empty(t, parseSSEStream(t, append(body, '\n')), + "a line terminator alone must not turn a bare JSON object into an event") +}