Refuse subscriptions/listen requests that cannot be encoded

## Motivation and Context

A `subscriptions/listen` request whose id, or one of whose honored resource URIs, is a string holding bytes that
are not valid UTF-8 parses, since the JSON parser accepts such bytes, but cannot be written back as JSON.
The transport answered 200, registered the stream, and then failed to encode the acknowledgement that echoes both;
`JSON::GeneratorError` is not among the write errors the stream handles, so the registered entry was never removed
and kept one of the `max_listen_subscriptions` slots until the transport closed.

The acknowledgement is now encoded before the stream is committed to. An id that cannot be encoded is refused with
HTTP 400 and JSON-RPC `-32600`, the error carrying a null id since the id itself cannot be written;
an honored filter that cannot be encoded is refused with HTTP 400 and `-32602`. Neither registers a stream.

## How Has This Been Tested?

New tests in `test/mcp/server/transports/streamable_http_transport_test.rb` send a listen request whose id and
one whose resource URI hold an invalid byte, and check the 400 and that no stream is registered.
Against the previous library both answer 200.
Two more send the invalid URI to a server without the `subscribe` capability, which acknowledges without it,
and to a transport at its stream cap, which answers 503 first.

## Breaking Changes

None.

Author: koic

lib/mcp/server/transports/streamable_http_transport.rb

@@ -904,6 +904,14 @@ def handle_subscriptions_listen(body)
             return invalid_request_response("Invalid Request: subscriptions/listen requires an id")
           end
 
+          # The id is echoed in every message of the stream, so one that cannot be written back as JSON
+          # (a String holding bytes that are not valid UTF-8 parses, but does not generate) is refused here,
+          # while the refusal can still answer with a null id, rather than surfacing as a failed write after
+          # the stream has been registered.
+          unless json_encodable?(request_id)
+            return invalid_request_response("Invalid Request: subscriptions/listen id must be valid UTF-8")
+          end
+
           begin
             if RequestEnvelope.modern?(params)
               RequestEnvelope.parse!(params, request: params)
@@ -946,7 +954,43 @@ def handle_subscriptions_listen(body)
             return too_many_listen_subscriptions_response(request_id)
           end
 
-          [200, SSE_HEADERS.dup, listen_sse_body(request_id, honored_filter(filter))]
+          honored = honored_filter(filter)
+
+          # The acknowledgement echoes the honored filter, whose resource URIs are client-supplied strings,
+          # so it is encoded before the stream is committed to: a URI that cannot be written as JSON is
+          # refused with a 400 instead of failing the first write after the stream was registered.
+          unless (acknowledgement = listen_acknowledgement_json(request_id, honored))
+            return json_rpc_error_response(
+              status: 400,
+              code: JsonRpcHandler::ErrorCode::INVALID_PARAMS,
+              message: "Invalid params: subscriptions/listen `notifications` must be valid UTF-8",
+              id: request_id,
+            )
+          end
+
+          [200, SSE_HEADERS.dup, listen_sse_body(request_id, honored, acknowledgement)]
+        end
+
+        # The `notifications/subscriptions/acknowledged` frame for a listen stream as JSON, or `nil` when
+        # the request id or the honored filter holds a string JSON cannot encode.
+        def listen_acknowledgement_json(request_id, honored)
+          {
+            jsonrpc: "2.0",
+            method: Methods::NOTIFICATIONS_SUBSCRIPTIONS_ACKNOWLEDGED,
+            params: {
+              notifications: honored,
+              _meta: { RequestEnvelope::SUBSCRIPTION_ID_META_KEY.to_sym => request_id },
+            },
+          }.to_json
+        rescue JSON::GeneratorError
+          nil
+        end
+
+        def json_encodable?(value)
+          value.to_json
+          true
+        rescue JSON::GeneratorError
+          false
         end
 
         def listen_subscriptions_full?
@@ -1005,7 +1049,7 @@ def first
         # The entry is keyed by an identifier minted here, not by the request id: that id is unique only among
         # the requesting client's own in-flight requests, and two clients that pick the same one must each get
         # their stream, stamped with the id they sent.
-        def listen_sse_body(request_id, honored)
+        def listen_sse_body(request_id, honored, acknowledgement)
           ListenStreamBody.new do |stream|
             subscription_key = SecureRandom.uuid
             subscription = nil
@@ -1029,15 +1073,6 @@ def listen_sse_body(request_id, honored)
             if subscription.nil?
               close_stream_safely(stream)
             else
-              acknowledgement = {
-                jsonrpc: "2.0",
-                method: Methods::NOTIFICATIONS_SUBSCRIPTIONS_ACKNOWLEDGED,
-                params: {
-                  notifications: honored,
-                  _meta: { RequestEnvelope::SUBSCRIPTION_ID_META_KEY.to_sym => request_id },
-                },
-              }
-
               begin
                 acknowledged = subscription[:write_mutex].synchronize do
                   next false if subscription[:closed]

test/mcp/server/transports/streamable_http_transport_test.rb

@@ -6564,6 +6564,80 @@ def string
           refute(events.any? { |event| event["method"] == "notifications/tools/list_changed" })
         end
 
+        test "subscriptions/listen refuses an id that is not valid UTF-8 without registering a stream" do
+          body = modern_listen_body(id: "listen-BAD", params: { notifications: { toolsListChanged: true } }).b.sub("BAD", "\xFF".b)
+
+          response = @transport.handle_request(modern_rack_request(body))
+
+          assert_equal 400, response[0]
+
+          parsed = JSON.parse(response[2][0])
+
+          # The member is present and null, as JSON-RPC 2.0 requires when the id cannot be determined.
+          assert parsed.key?("id")
+          assert_nil parsed["id"]
+          assert_equal JsonRpcHandler::ErrorCode::INVALID_REQUEST, parsed.dig("error", "code")
+          assert_empty @transport.instance_variable_get(:@listen_subscriptions)
+        end
+
+        test "subscriptions/listen refuses a resource URI that is not valid UTF-8 without registering a stream" do
+          server = Server.new(name: "listen_test", capabilities: { resources: { subscribe: true } })
+          transport = StreamableHTTPTransport.new(server, listen_keepalive_interval: nil)
+          body = modern_listen_body(
+            id: "listen-1",
+            params: { notifications: { resourceSubscriptions: ["file:///BAD"] } }
+          ).b.sub("BAD", "\xFF".b)
+
+          response = transport.handle_request(modern_rack_request(body))
+
+          assert_equal 400, response[0]
+
+          parsed = JSON.parse(response[2][0])
+
+          assert_equal "listen-1", parsed["id"]
+          assert_equal(-32602, parsed.dig("error", "code"))
+          assert_empty transport.instance_variable_get(:@listen_subscriptions)
+        ensure
+          transport.close
+        end
+
+        test "subscriptions/listen acknowledges without a resource URI that is not valid UTF-8 when subscribe is not declared" do
+          # Without the `subscribe` capability the URIs are not honored, so the acknowledgement never echoes
+          # the unencodable one and the request is served.
+          body = modern_listen_body(
+            id: "listen-1",
+            params: { notifications: { toolsListChanged: true, resourceSubscriptions: ["file:///BAD"] } },
+          ).b.sub("BAD", "\xFF".b)
+
+          response = @transport.handle_request(modern_rack_request(body))
+
+          assert_equal 200, response[0]
+
+          io = StringIO.new
+          response[2].call(io)
+          acknowledgement = sse_events(io).first
+
+          assert_equal "notifications/subscriptions/acknowledged", acknowledgement["method"]
+          refute acknowledgement.dig("params", "notifications").key?("resourceSubscriptions")
+        end
+
+        test "subscriptions/listen refuses a request past the stream cap before checking its resource URIs" do
+          server = Server.new(name: "listen_test", capabilities: { resources: { subscribe: true } })
+          transport = StreamableHTTPTransport.new(server, max_listen_subscriptions: 1, listen_keepalive_interval: nil)
+          open_listen_stream(id: "listen-1", notifications: { toolsListChanged: true }, transport: transport)
+          body = modern_listen_body(
+            id: "listen-2",
+            params: { notifications: { resourceSubscriptions: ["file:///BAD"] } }
+          ).b.sub("BAD", "\xFF".b)
+
+          response = transport.handle_request(modern_rack_request(body))
+
+          assert_equal 503, response[0]
+          assert_equal "listen-2", JSON.parse(response[2][0])["id"]
+        ensure
+          transport.close
+        end
+
         test "transport close during the acknowledgement write still sends the acknowledgement first" do
           # The stream blocks its first write, the acknowledgement, until released, so the close arrives while
           # that write is in progress and has to queue behind it on the stream's write mutex.