Add a Session-Ownership Hook to StreamableHTTPTransport

## Motivation and Context

`StreamableHTTPTransport` issues a random `SecureRandom.uuid` session ID and validates
an incoming request by session existence and idle timeout only, with no binding to a client identity.
An attacker who obtains a victim's session ID (network sniffing, logs, etc.) can therefore POST
a `tools/call` on that session and the server executes it, silently poisoning the victim's session.
The previously shipped 409 fix guards the GET SSE stream slot but does nothing for the POST write path,
so this is not covered by it.

Binding a session to a user is fundamentally the deploying application's responsibility:
the transport hands the server only the JSON-RPC body string and never receives the authenticated principal
(`Authorization`/`Origin` headers do not reach handlers), and the MCP spec frames session-to-user binding as
a SHOULD. The Python, Go, and C# SDKs bind automatically only when an authenticated identity is present.
This gives the Ruby SDK the seam to do the same:

- A `session_request_validator` constructor option, called as `->(request, session_id) { true | false }`
  on every non-`initialize` POST, GET, and DELETE against an existing session; a falsy return rejects
  with HTTP 403. The gate is applied uniformly to regular requests, notifications, and client responses,
  not just tool calls - otherwise a stolen session ID could POST `notifications/cancelled` to cancel
  a victim's in-flight request, or a forged client response, by taking a branch the gate did not cover.
  `initialize` is exempt because it establishes the session. An application that has an authenticated principal
  uses the validator to enforce that the caller owns the session; without one, ownership is not enforced.
- A limited built-in Origin-consistency check as defense in depth (not authentication): the `Origin` header
  is recorded at `initialize` and a later request whose `Origin` differs is rejected, but only when
  both are present. A non-browser client that omits `Origin` (curl, a script - i.e. the advisory's own PoC)
  is not stopped by this check; it only raises the bar against a browser-based cross-origin attacker.
  Enforcing ownership against a determined attacker requires the validator.

Stateless mode retains no sessions, so neither check applies to it.

## How Has This Been Tested?

New tests in `test/mcp/server/transports/streamable_http_transport_test.rb`: a POST whose Origin differs
from `initialize` is rejected with 403; a matching Origin and a no-Origin request both succeed;
a custom `session_request_validator` both rejects (403) and allows (200) and receives the request and session ID;
the validator also gates notification (`notifications/cancelled`) and client-response POSTs, not just regular requests;
and stateless mode never consults the validator.

## Breaking Changes

None. `session_request_validator` defaults to `nil`, and the built-in Origin check only rejects when both
the session and the request carry an `Origin` and they differ, so existing applications are unaffected.

test/mcp/server/transports/streamable_http_transport_test.rb

@@ -4741,6 +4741,196 @@ module MCP
           transport.close
         end
 
+        test "rejects a POST to a session whose Origin differs from initialize with 403" do
+          # Both origins are allow-listed so they pass DNS rebinding protection; the rejection
+          # below comes from the per-session Origin-consistency gate, not from the rebinding check.
+          transport = StreamableHTTPTransport.new(
+            @server,
+            enable_json_response: true,
+            allowed_origins: ["https://app.example.com", "https://evil.example.com"],
+          )
+          init = create_rack_request(
+            "POST",
+            "/",
+            { "CONTENT_TYPE" => "application/json", "HTTP_ORIGIN" => "https://app.example.com" },
+            { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json,
+          )
+          session_id = transport.handle_request(init)[1]["Mcp-Session-Id"]
+
+          attacker = create_rack_request(
+            "POST",
+            "/",
+            {
+              "CONTENT_TYPE" => "application/json",
+              "HTTP_MCP_SESSION_ID" => session_id,
+              "HTTP_ORIGIN" => "https://evil.example.com",
+            },
+            { jsonrpc: "2.0", method: "ping", id: "1" }.to_json,
+          )
+          response = transport.handle_request(attacker)
+
+          assert_equal(403, response[0])
+        ensure
+          transport.close
+        end
+
+        test "allows a POST with a matching Origin and one with no Origin" do
+          transport = StreamableHTTPTransport.new(
+            @server,
+            enable_json_response: true,
+            allowed_origins: ["https://app.example.com"],
+          )
+          init = create_rack_request(
+            "POST",
+            "/",
+            { "CONTENT_TYPE" => "application/json", "HTTP_ORIGIN" => "https://app.example.com" },
+            { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json,
+          )
+          session_id = transport.handle_request(init)[1]["Mcp-Session-Id"]
+
+          same_origin = create_rack_request(
+            "POST",
+            "/",
+            {
+              "CONTENT_TYPE" => "application/json",
+              "HTTP_MCP_SESSION_ID" => session_id,
+              "HTTP_ORIGIN" => "https://app.example.com",
+            },
+            { jsonrpc: "2.0", method: "ping", id: "1" }.to_json,
+          )
+
+          # A non-browser client that sends no Origin is not rejected by the built-in check.
+          no_origin = create_rack_request(
+            "POST",
+            "/",
+            { "CONTENT_TYPE" => "application/json", "HTTP_MCP_SESSION_ID" => session_id },
+            { jsonrpc: "2.0", method: "ping", id: "2" }.to_json,
+          )
+
+          assert_equal(200, transport.handle_request(same_origin)[0])
+          assert_equal(200, transport.handle_request(no_origin)[0])
+        ensure
+          transport.close
+        end
+
+        test "a custom session_request_validator rejects and allows requests" do
+          allowed = { value: true }
+          validator = ->(_request, _session_id) { allowed[:value] }
+          transport = StreamableHTTPTransport.new(@server, enable_json_response: true, session_request_validator: validator)
+          init = create_rack_request(
+            "POST",
+            "/",
+            { "CONTENT_TYPE" => "application/json" },
+            { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json,
+          )
+          session_id = transport.handle_request(init)[1]["Mcp-Session-Id"]
+
+          ping = lambda do
+            create_rack_request(
+              "POST",
+              "/",
+              { "CONTENT_TYPE" => "application/json", "HTTP_MCP_SESSION_ID" => session_id },
+              { jsonrpc: "2.0", method: "ping", id: "1" }.to_json,
+            )
+          end
+
+          assert_equal(200, transport.handle_request(ping.call)[0])
+          allowed[:value] = false
+          assert_equal(403, transport.handle_request(ping.call)[0])
+        ensure
+          transport.close
+        end
+
+        test "the validator also gates notification and response POSTs, not just regular requests" do
+          # A stolen session ID must not be able to POST `notifications/cancelled` (which would cancel a victim's in-flight request)
+          # or a client response by bypassing the ownership gate. `initialize` establishes the session and is exempt,
+          # so a single transport whose validator always rejects still initializes.
+          transport = StreamableHTTPTransport.new(
+            @server,
+            enable_json_response: true,
+            session_request_validator: ->(_request, _session_id) { false },
+          )
+          init = create_rack_request(
+            "POST",
+            "/",
+            { "CONTENT_TYPE" => "application/json" },
+            { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json,
+          )
+          session_id = transport.handle_request(init)[1]["Mcp-Session-Id"]
+          refute_nil(session_id)
+
+          headers = { "CONTENT_TYPE" => "application/json", "HTTP_MCP_SESSION_ID" => session_id }
+          cancelled = create_rack_request(
+            "POST",
+            "/",
+            headers,
+            { jsonrpc: "2.0", method: "notifications/cancelled", params: { requestId: "1" } }.to_json,
+          )
+          client_response = create_rack_request(
+            "POST",
+            "/",
+            headers,
+            { jsonrpc: "2.0", id: "1", result: {} }.to_json,
+          )
+
+          assert_equal(403, transport.handle_request(cancelled)[0])
+          assert_equal(403, transport.handle_request(client_response)[0])
+        ensure
+          transport.close
+        end
+
+        test "session_request_validator receives the request and session id" do
+          seen = {}
+          validator = ->(request, session_id) {
+            seen[:origin] = request.env["HTTP_ORIGIN"]
+            seen[:session_id] = session_id
+            true
+          }
+          transport = StreamableHTTPTransport.new(
+            @server,
+            enable_json_response: true,
+            allowed_origins: ["https://app.example.com"],
+            session_request_validator: validator,
+          )
+          init = create_rack_request(
+            "POST",
+            "/",
+            { "CONTENT_TYPE" => "application/json" },
+            { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json,
+          )
+          session_id = transport.handle_request(init)[1]["Mcp-Session-Id"]
+
+          transport.handle_request(create_rack_request(
+            "POST",
+            "/",
+            { "CONTENT_TYPE" => "application/json", "HTTP_MCP_SESSION_ID" => session_id, "HTTP_ORIGIN" => "https://app.example.com" },
+            { jsonrpc: "2.0", method: "ping", id: "1" }.to_json,
+          ))
+
+          assert_equal(session_id, seen[:session_id])
+          assert_equal("https://app.example.com", seen[:origin])
+        ensure
+          transport.close
+        end
+
+        test "session validation does not apply in stateless mode" do
+          rejecting = ->(_request, _session_id) { false }
+          transport = StreamableHTTPTransport.new(@server, stateless: true, enable_json_response: true, session_request_validator: rejecting)
+
+          request = create_rack_request(
+            "POST",
+            "/",
+            { "CONTENT_TYPE" => "application/json" },
+            { jsonrpc: "2.0", method: "ping", id: "1" }.to_json,
+          )
+          response = transport.handle_request(request)
+
+          # Stateless mode keeps no sessions, so the validator is never consulted.
+          assert_equal(200, response[0])
+        ensure
+          transport.close
+        end
+
         private
 
         def create_rack_request(method, path, headers, body = nil)