diff --git a/README.md b/README.md index b986c26..d94d84a 100644 --- a/README.md +++ b/README.md @@ -1758,14 +1758,23 @@ Session-scoped standalone notifications (`resources/updated`, `elicitation/compl broadcast notifications (`tools/list_changed`, etc.) still flow to clients connected to the GET SSE stream. This mode is suitable for simple tool servers that do not need server-initiated requests. -By default, sessions do not expire. To mitigate session hijacking risks, you can set a `session_idle_timeout` (in seconds). -When configured, sessions that receive no HTTP requests for this duration are automatically expired and cleaned up: +By default, stateful sessions are bounded so an `initialize` flood cannot retain sessions until memory is exhausted: +they expire after `session_idle_timeout` seconds of inactivity (default 1800, i.e. 30 minutes) and the concurrent +session count is capped at `max_sessions` (default 10000). A session's idle timer is reset by activity that touches it +(a GET, or a regular-request POST), and expired sessions are collected by a background reaper roughly once a minute, +so cleanup lags inactivity by up to that interval. At the cap, the transport first reclaims any already-expired slots +and then, if still full, rejects a new `initialize` with HTTP 503 (it does not evict an existing session). ```ruby -# Session timeout of 30 minutes -transport = MCP::Server::Transports::StreamableHTTPTransport.new(server, session_idle_timeout: 1800) +# Tune the limits +transport = MCP::Server::Transports::StreamableHTTPTransport.new(server, session_idle_timeout: 900, max_sessions: 5000) + +# Opt out of expiry and/or the cap (not recommended on internet-facing deployments) +transport = MCP::Server::Transports::StreamableHTTPTransport.new(server, session_idle_timeout: nil, max_sessions: nil) ``` +Stateless mode (`stateless: true`) retains no sessions, so neither limit applies to it. + ### Pagination The MCP Ruby SDK supports [pagination](https://modelcontextprotocol.io/specification/2025-11-25/server/utilities/pagination) diff --git a/lib/mcp/server/transports/streamable_http_transport.rb b/lib/mcp/server/transports/streamable_http_transport.rb index 4b8f8ad..e79017e 100644 --- a/lib/mcp/server/transports/streamable_http_transport.rb +++ b/lib/mcp/server/transports/streamable_http_transport.rb @@ -24,7 +24,23 @@ module MCP "Connection" => "keep-alive", }.freeze - def initialize(server, stateless: false, enable_json_response: false, session_idle_timeout: nil) + # Secure defaults for stateful mode. Without a finite idle timeout, sessions live until an explicit client DELETE, + # so an unauthenticated `initialize` flood retains unbounded `ServerSession` objects until memory is exhausted. + # These defaults expire idle sessions and cap the concurrent count, like the C# SDK (the only reference SDK that + # hardens this by default, with a 2h idle timeout and a 10k idle-session count). One difference: at the cap this transport + # rejects a new `initialize` with 503 (after reclaiming any already-expired slots), whereas the C# SDK evicts + # the oldest idle session. Rejecting keeps established sessions stable and avoids evicting a legitimate idle session on + # an attacker's behalf, at the cost of refusing new sessions while genuinely full. Pass `session_idle_timeout: nil` to + # opt out of expiry and `max_sessions: nil` to opt out of the cap. + DEFAULT_SESSION_IDLE_TIMEOUT = 1800 + DEFAULT_MAX_SESSIONS = 10_000 + + # Distinguishes "argument omitted, apply the secure default" from an explicit `nil` (opt out of expiry). + UNSET_IDLE_TIMEOUT = Object.new.freeze + private_constant :UNSET_IDLE_TIMEOUT + + def initialize(server, stateless: false, enable_json_response: false, + session_idle_timeout: UNSET_IDLE_TIMEOUT, max_sessions: DEFAULT_MAX_SESSIONS) super(server) # Maps `session_id` to `{ get_sse_stream: stream_object, server_session: ServerSession, last_active_at: float_from_monotonic_clock }`. @sessions = {} @@ -32,9 +48,16 @@ module MCP @stateless = stateless @enable_json_response = enable_json_response - @session_idle_timeout = session_idle_timeout @pending_responses = {} + # Resolve the idle timeout: an explicit value (including `nil` to opt out) wins; otherwise apply the secure default, + # which does not apply to stateless mode since it retains no sessions. + @session_idle_timeout = if session_idle_timeout.equal?(UNSET_IDLE_TIMEOUT) + stateless ? nil : DEFAULT_SESSION_IDLE_TIMEOUT + else + session_idle_timeout + end + if @session_idle_timeout if @stateless raise ArgumentError, "session_idle_timeout is not supported in stateless mode." @@ -43,6 +66,13 @@ module MCP end end + unless max_sessions.nil? || (max_sessions.is_a?(Integer) && max_sessions > 0) + raise ArgumentError, "max_sessions must be a positive Integer or nil" + end + + # The cap guards the stateful session store; stateless mode keeps none. + @max_sessions = stateless ? nil : max_sessions + start_reaper_thread if @session_idle_timeout end @@ -622,13 +652,30 @@ module MCP session_id = SecureRandom.uuid server_session = ServerSession.new(server: @server, transport: self, session_id: session_id) - @mutex.synchronize do + # Cap the concurrent session count so an `initialize` flood cannot retain unbounded sessions until memory is exhausted. + # The check and insert share the mutex so concurrent initializes cannot race past the limit. + reclaimed = [] + inserted = @mutex.synchronize do + # When at the cap, first reclaim slots held by already-expired sessions the 60s reaper has not yet collected, + # so the cap rejects only when genuinely full rather than up to a reaper interval after sessions expired. + if @max_sessions && @sessions.size >= @max_sessions + @sessions.each_key.select { |id| session_expired?(@sessions[id]) }.each do |id| + cleanup_and_collect_stream(id, reclaimed) + end + end + + next false if @max_sessions && @sessions.size >= @max_sessions + @sessions[session_id] = { get_sse_stream: nil, server_session: server_session, last_active_at: Process.clock_gettime(Process::CLOCK_MONOTONIC), } + true end + + reclaimed.each { |stream| close_stream_safely(stream) } + return too_many_sessions_response unless inserted end response = server_session.handle_json(body_string) @@ -655,6 +702,14 @@ module MCP [202, {}, []] end + def too_many_sessions_response + json_rpc_error_response( + status: 503, + code: JsonRpcHandler::ErrorCode::INTERNAL_ERROR, + message: "Service unavailable: maximum concurrent sessions (#{@max_sessions}) reached", + ) + end + def handle_regular_request(body_string, session_id, related_request_id: nil) server_session = nil diff --git a/test/mcp/server/transports/streamable_http_transport_test.rb b/test/mcp/server/transports/streamable_http_transport_test.rb index 9b26949..ece1f77 100644 --- a/test/mcp/server/transports/streamable_http_transport_test.rb +++ b/test/mcp/server/transports/streamable_http_transport_test.rb @@ -3808,9 +3808,14 @@ module MCP transport.close end - test "default session_idle_timeout is nil and sessions do not expire" do + test "applies a finite session_idle_timeout by default and starts the reaper" do + # Secure default: stateful mode expires idle sessions out of the box. transport = StreamableHTTPTransport.new(@server) - assert_nil(transport.instance_variable_get(:@reaper_thread)) + assert_equal( + StreamableHTTPTransport::DEFAULT_SESSION_IDLE_TIMEOUT, + transport.instance_variable_get(:@session_idle_timeout), + ) + assert(transport.instance_variable_get(:@reaper_thread).alive?) init_request = create_rack_request( "POST", @@ -3818,25 +3823,101 @@ module MCP { "CONTENT_TYPE" => "application/json" }, { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json, ) - init_response = transport.handle_request(init_request) - session_id = init_response[1]["Mcp-Session-Id"] - + # A freshly created session is well within the timeout, so normal use is unaffected. + session_id = transport.handle_request(init_request)[1]["Mcp-Session-Id"] request = create_rack_request( "POST", "/", - { - "CONTENT_TYPE" => "application/json", - "HTTP_MCP_SESSION_ID" => session_id, - }, + { "CONTENT_TYPE" => "application/json", "HTTP_MCP_SESSION_ID" => session_id }, { jsonrpc: "2.0", method: "ping", id: "456" }.to_json, ) - response = transport.handle_request(request) - assert_equal(200, response[0]) + assert_equal(200, transport.handle_request(request)[0]) ensure transport.close end + test "session_idle_timeout: nil opts out of expiry" do + transport = StreamableHTTPTransport.new(@server, session_idle_timeout: nil) + assert_nil(transport.instance_variable_get(:@session_idle_timeout)) + assert_nil(transport.instance_variable_get(:@reaper_thread)) + ensure + transport.close + end + + test "stateless mode keeps no idle timeout despite the secure default" do + transport = StreamableHTTPTransport.new(@server, stateless: true) + assert_nil(transport.instance_variable_get(:@session_idle_timeout)) + assert_nil(transport.instance_variable_get(:@reaper_thread)) + ensure + transport.close + end + + test "rejects new initialize with 503 once max_sessions is reached" do + transport = StreamableHTTPTransport.new(@server, max_sessions: 2) + + 2.times do + request = create_rack_request( + "POST", + "/", + { "CONTENT_TYPE" => "application/json" }, + { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json, + ) + assert_equal(200, transport.handle_request(request)[0]) + end + + overflow = create_rack_request( + "POST", + "/", + { "CONTENT_TYPE" => "application/json" }, + { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json, + ) + response = transport.handle_request(overflow) + + assert_equal(503, response[0]) + assert_equal(2, transport.instance_variable_get(:@sessions).size) + ensure + transport.close + end + + test "at the cap, reclaims an already-expired session before rejecting" do + # A very short idle timeout so the first session is expired by the time the cap would otherwise reject + # a new initialize; the expired slot is reclaimed. + transport = StreamableHTTPTransport.new(@server, max_sessions: 1, session_idle_timeout: 0.01) + init = lambda do + transport.handle_request(create_rack_request( + "POST", + "/", + { "CONTENT_TYPE" => "application/json" }, + { jsonrpc: "2.0", method: "initialize", id: "init" }.to_json, + )) + end + + assert_equal(200, init.call[0]) + sleep(0.05) # let the first session's idle timer expire + second = init.call + + # The expired session was reclaimed in-line, so the second initialize succeeds (200) rather than + # being rejected with 503, and the store stays at the cap. + assert_equal(200, second[0]) + assert_equal(1, transport.instance_variable_get(:@sessions).size) + ensure + transport.close + end + + test "max_sessions: nil opts out of the session cap" do + transport = StreamableHTTPTransport.new(@server, max_sessions: nil) + assert_nil(transport.instance_variable_get(:@max_sessions)) + ensure + transport.close + end + + test "raises ArgumentError for an invalid max_sessions" do + [0, -1, 1.5, "10"].each do |invalid| + assert_raises(ArgumentError) { StreamableHTTPTransport.new(@server, max_sessions: invalid) } + end + end + test "raises ArgumentError when session_idle_timeout is zero" do error = assert_raises(ArgumentError) do StreamableHTTPTransport.new(@server, session_idle_timeout: 0)