Deliver SSE filter failures as parseable events

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

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