From 081a7681c96d30f4c0e93a47751bada5bf55600d Mon Sep 17 00:00:00 2001 From: Jakub Hrozek Date: Mon, 27 Jul 2026 22:23:42 +0200 Subject: [PATCH] Frame SSE filter-failure errors as SSE events On the text/event-stream path, writeErrorResponse wrote a bare JSON object straight into the stream body with no "data: " prefix. Per the WHATWG event-stream grammar that parses as a field named {"jsonrpc", and an unrecognized field is ignored, so the client never saw the error at all: it saw a stream that stopped with no response to its request and hung until its own timeout. That is the worst available failure mode, because the caller cannot tell it apart from a dead server. Emit the error as a "data:" line in place of the frame that failed and let the loop keep going, so the separator and the trailing reconciliation frame it exactly as they frame a successfully filtered frame. Remaining frames are unrelated to the failed one, so continuing also removes the truncation that the old early return caused by skipping the reconciliation. The mid-stream WriteHeader(500) is gone: Flush() has already committed the headers by then, and text/event-stream carrying a bare JSON body is unreadable by an SSE client regardless. writeSSEDataLine's stop return value collapses into err != nil, which also fixes a latent wart where the caller discarded a non-nil error whenever stop was false. This buys framing parity with the success path, not general SSE correctness. A body carrying more than one event is still mis-framed by the separator reconciliation, and a body with no trailing blank line still gets no terminator. Both predate this change, apply equally when filtering succeeds, and are left alone. The failure branch is unreachable from any real backend response, so the tests cover it at the seam rather than through FlushAndFilter: no test-only injection point was added to production code. Refs #6037 Co-Authored-By: Claude Opus 5 --- pkg/authz/response_filter.go | 100 ++++++++++++++++++++---------- pkg/authz/response_filter_test.go | 82 ++++++++++++++++++++++++ 2 files changed, 150 insertions(+), 32 deletions(-) diff --git a/pkg/authz/response_filter.go b/pkg/authz/response_filter.go index bd99931117..4d41ca1dcc 100644 --- a/pkg/authz/response_filter.go +++ b/pkg/authz/response_filter.go @@ -184,12 +184,7 @@ func (rfw *ResponseFilteringWriter) processJSONResponse(rawResponse []byte) erro return werr } - filteredResponse, err := rfw.filterListResponse(response) - if err != nil { - return rfw.writeErrorResponse(response.ID, err) - } - - filteredData, err := jsonrpc2.EncodeMessage(filteredResponse) + filteredData, err := rfw.filterAndEncode(response) if err != nil { return rfw.writeErrorResponse(response.ID, err) } @@ -199,6 +194,15 @@ func (rfw *ResponseFilteringWriter) processJSONResponse(rawResponse []byte) erro return err } +// filterAndEncode filters a Response and encodes the result for the wire. +func (rfw *ResponseFilteringWriter) filterAndEncode(response *jsonrpc2.Response) ([]byte, error) { + filteredResponse, err := rfw.filterListResponse(response) + if err != nil { + return nil, err + } + return jsonrpc2.EncodeMessage(filteredResponse) +} + 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 @@ -227,8 +231,8 @@ 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) - if stop { + w, err := rfw.writeSSEDataLine(data) + if err != nil { return err } written = w @@ -249,7 +253,9 @@ func (rfw *ResponseFilteringWriter) processSSEResponse(rawResponse []byte) error } // This ensures we don't send too few line separators, which might break - // SSE parsing. + // 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 { @@ -262,28 +268,30 @@ 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) { +// 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) { 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) - } - filteredData, ferr := jsonrpc2.EncodeMessage(filteredResponse) + filteredData, ferr := rfw.filterAndEncode(response) if ferr != nil { - return false, true, rfw.writeErrorResponse(response.ID, ferr) + 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, true, fmt.Errorf("%w: %w", errBug, werr) + return false, fmt.Errorf("%w: %w", errBug, werr) } - return true, false, nil + return true, nil 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 @@ -293,20 +301,20 @@ func (rfw *ResponseFilteringWriter) writeSSEDataLine(data []byte) (written, stop // 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", "method", rfw.method) - return true, false, nil + return true, nil 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", "method", rfw.method, "error", decodeErr) - return false, false, nil + return false, nil default: // Genuine non-Response frame (e.g. an interleaved notifications/* message) // with no result payload. Routine SSE traffic, so log at Debug to keep the // suspicious branches above from being buried. Pass through this line only. slog.Debug("SSE data line was not a JSON-RPC Response; passing through unfiltered", "method", rfw.method) - return false, false, nil + return false, nil } } @@ -579,32 +587,60 @@ 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 { +// errorResponseBody logs the full filtering error server-side and encodes a +// JSON-RPC error response carrying a deliberately generic client-visible +// message: err can originate in policy evaluation and name tools or resources, +// so security.md forbids returning it verbatim (#6066). Both transports build +// their body here, so neither can leak it. id is echoed so the client can +// correlate the error with its request. +func (rfw *ResponseFilteringWriter) errorResponseBody(id jsonrpc2.ID, err error) []byte { slog.Error("error filtering response", "method", rfw.method, "error", err) 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 // jsonrpc2.Response built from a valid ID and message above. Fall back // to a hardcoded valid JSON-RPC error body rather than writing nothing. 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 fmt.Appendf(nil, `{"jsonrpc":"2.0","error":{"code":%d,"message":"internal error"}}`, + mcpparser.CodeInternalError) } + return body +} +// 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. +func (rfw *ResponseFilteringWriter) writeErrorResponse(id jsonrpc2.ID, err error) error { rfw.ResponseWriter.WriteHeader(http.StatusInternalServerError) - _, writeErr := rfw.ResponseWriter.Write(body) + _, 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 373f7561bb..cea960ecc3 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" @@ -1366,3 +1367,84 @@ func TestResponseFilteringWriter_FilterBypassAttempts(t *testing.T) { }) } } + +// TestErrorResponseBody pins the wire shape of the JSON-RPC error envelope +// errorResponseBody produces for a filter/encode failure: a "2.0" jsonrpc tag, +// the standard internal-error code, a generic message that must NOT echo the +// wrapped error (it can name tools the caller is not authorized to see, so +// #6066 and security.md forbid forwarding it), and an id that round-trips for +// a real request id but is entirely ABSENT (not present and null) for the +// zero-value id. Absence matters because MCP types the +// error response id as optional (id?: RequestId), and the reference +// TypeScript SDK's strict schema admits undefined but not null — a null id +// would make that client throw inside its transport. +func TestErrorResponseBody(t *testing.T) { + t.Parallel() + + wrapped := errors.New("leaky-tool-name-sentinel") + + testCases := []struct { + name string + id jsonrpc2.ID + wantID any // nil means the id key must be absent + }{ + {name: "int64 id round-trips", id: jsonrpc2.Int64ID(7), wantID: float64(7)}, + {name: "string id round-trips", id: jsonrpc2.StringID("abc"), wantID: "abc"}, + {name: "zero-value id key is absent, not null", id: jsonrpc2.ID{}}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, nil, nil, string(mcp.MethodToolsList), nil, nil) + body := rfw.errorResponseBody(tc.id, wrapped) + + var decoded map[string]any + require.NoError(t, json.Unmarshal(body, &decoded)) + + assert.Equal(t, "2.0", decoded["jsonrpc"]) + + errObj, ok := decoded["error"].(map[string]any) + require.True(t, ok, "error field must decode as an object") + assert.Equal(t, float64(mcpparser.CodeInternalError), errObj["code"]) + assert.Equal(t, "internal error", errObj["message"], + "the client-visible message must be generic") + assert.NotContains(t, string(body), wrapped.Error(), + "the wrapped error must never reach the client: it can name tools the caller cannot see") + + idValue, hasID := decoded["id"] + require.Equal(t, tc.wantID != nil, hasID, "id key presence mismatch") + if tc.wantID != nil { + assert.Equal(t, tc.wantID, idValue) + } + }) + } +} + +// TestWriteSSEErrorLine pins the two invariants writeSSEErrorLine must hold +// on the text/event-stream path (issue #6037): the written line is exactly +// "data: " followed by the same envelope errorResponseBody produces, with no +// trailing terminator (the caller, processSSEResponse, owns the separator — +// a hardcoded one here would desync a \r\n stream), and the status code is +// left untouched, since headers are already committed by the time an SSE +// frame fails mid-stream. +func TestWriteSSEErrorLine(t *testing.T) { + t.Parallel() + + // writeSSEErrorLine only reads rfw.ResponseWriter; the authorizer, request, + // and Content-Type are irrelevant on this path, so nils are safe here. + rr := httptest.NewRecorder() + rfw := NewResponseFilteringWriter(rr, nil, nil, string(mcp.MethodToolsList), nil, nil) + + id := jsonrpc2.Int64ID(9) + wrapped := errors.New("boom") + require.NoError(t, rfw.writeSSEErrorLine(id, wrapped)) + + want := "data: " + string(rfw.errorResponseBody(id, wrapped)) + assert.Equal(t, want, rr.Body.String(), + "writeSSEErrorLine must write exactly the data-prefixed envelope, nothing else") + assert.Equal(t, http.StatusOK, rr.Code, + "writeSSEErrorLine must not set a non-200 status; the SSE headers are already committed by the time it runs") +}