Merge commit from fork
Bound stdio frame reads with max_line_bytes
This commit is contained in:
commit
c8930758ad
7 changed files with 285 additions and 5 deletions
|
|
@ -102,6 +102,8 @@ transport = MCP::Server::Transports::StdioTransport.new(server)
|
|||
transport.open
|
||||
```
|
||||
|
||||
`StdioTransport.new` accepts an optional `max_line_bytes:` keyword that caps the byte length of a single newline-delimited request frame. A frame that reaches this limit without a newline is rejected and the connection is closed, preventing unbounded memory growth from a peer that never emits a newline. It defaults to `4 * 1024 * 1024` (4 MiB).
|
||||
|
||||
You can run this script and then type in requests to the server at the command line.
|
||||
|
||||
```console
|
||||
|
|
@ -1958,6 +1960,7 @@ Use the `MCP::Client::Stdio` transport to interact with MCP servers running as s
|
|||
| `args:` | No | An array of arguments passed to the command. Defaults to `[]`. |
|
||||
| `env:` | No | A hash of environment variables to set for the server process. Defaults to `nil`. |
|
||||
| `read_timeout:` | No | Timeout in seconds for waiting for a server response. Defaults to `nil` (no timeout). |
|
||||
| `max_line_bytes:` | No | Maximum byte length of a single newline-delimited response frame. A frame that reaches this limit without a newline is rejected as a transport error, preventing unbounded memory growth from a server that never emits a newline. Defaults to `4 * 1024 * 1024` (4 MiB). |
|
||||
|
||||
Example usage:
|
||||
|
||||
|
|
|
|||
|
|
@ -65,6 +65,7 @@ stdio_transport.close
|
|||
| `args:` | No | An array of arguments passed to the command. Defaults to `[]`. |
|
||||
| `env:` | No | A hash of environment variables for the server process. Defaults to `nil`. |
|
||||
| `read_timeout:` | No | Timeout in seconds for waiting for a server response. Defaults to `nil`. |
|
||||
| `max_line_bytes:` | No | Maximum byte length of a single newline-delimited response frame. A frame that reaches this limit without a newline is rejected as a transport error, preventing unbounded memory growth from a server that never emits a newline. Defaults to `4 * 1024 * 1024` (4 MiB). |
|
||||
|
||||
## HTTP Transport
|
||||
|
||||
|
|
|
|||
|
|
@ -59,6 +59,8 @@ transport = MCP::Server::Transports::StdioTransport.new(server)
|
|||
transport.open
|
||||
```
|
||||
|
||||
`StdioTransport.new` accepts an optional `max_line_bytes:` keyword that caps the byte length of a single newline-delimited request frame. A frame that reaches this limit without a newline is rejected and the connection is closed, preventing unbounded memory growth from a peer that never emits a newline. It defaults to `4 * 1024 * 1024` (4 MiB).
|
||||
|
||||
## Streamable HTTP Transport
|
||||
|
||||
`MCP::Server::Transports::StreamableHTTPTransport` is a standard Rack app, so it can be mounted in any Rack-compatible framework.
|
||||
|
|
|
|||
|
|
@ -19,13 +19,28 @@ module MCP
|
|||
CLOSE_TIMEOUT = 2
|
||||
STDERR_READ_SIZE = 4096
|
||||
|
||||
# Default upper bound on a single newline-delimited frame read from the
|
||||
# server's stdout. CRuby's `IO#gets` without a limit accumulates bytes until a
|
||||
# newline arrives, so a spawned server that never emits one can grow a single
|
||||
# String until the host process is OOM-killed. 4 MiB is large enough for any
|
||||
# realistic JSON-RPC frame, including base64-embedded images.
|
||||
MAX_LINE_BYTES = 4 * 1024 * 1024
|
||||
|
||||
attr_reader :command, :args, :env, :server_info
|
||||
|
||||
def initialize(command:, args: [], env: nil, read_timeout: nil)
|
||||
def initialize(command:, args: [], env: nil, read_timeout: nil, max_line_bytes: MAX_LINE_BYTES)
|
||||
# Reject `nil` or non-positive values: `IO#gets("\n", nil)` and a negative
|
||||
# limit read without an upper bound, which would silently disable the
|
||||
# protection this option exists to provide.
|
||||
unless max_line_bytes.is_a?(Integer) && max_line_bytes > 0
|
||||
raise ArgumentError, "max_line_bytes must be a positive Integer"
|
||||
end
|
||||
|
||||
@command = command
|
||||
@args = args
|
||||
@env = env
|
||||
@read_timeout = read_timeout
|
||||
@max_line_bytes = max_line_bytes
|
||||
@stdin = nil
|
||||
@stdout = nil
|
||||
@stderr = nil
|
||||
|
|
@ -239,7 +254,7 @@ module MCP
|
|||
loop do
|
||||
ensure_running!
|
||||
wait_for_readable!(method, params) if @read_timeout
|
||||
line = @stdout.gets
|
||||
line = read_line(method, params)
|
||||
raise_connection_error!(method, params) if line.nil?
|
||||
|
||||
parsed = JSON.parse(line.strip)
|
||||
|
|
@ -278,6 +293,22 @@ module MCP
|
|||
)
|
||||
end
|
||||
|
||||
# Reads one newline-delimited frame from the server's stdout, bounded by `@max_line_bytes`.
|
||||
# Returns the line (including its trailing newline) or `nil` at EOF. Raises when the limit
|
||||
# is reached before a newline arrives, which signals a server streaming an unbounded frame.
|
||||
# A short final frame without a trailing newline (EOF) is still returned, since its length
|
||||
# stays under the limit.
|
||||
def read_line(method, params)
|
||||
line = @stdout.gets("\n", @max_line_bytes)
|
||||
return line unless line && !line.end_with?("\n") && line.bytesize >= @max_line_bytes
|
||||
|
||||
raise RequestHandlerError.new(
|
||||
"Server response frame exceeds #{@max_line_bytes} bytes without a newline",
|
||||
{ method: method, params: params },
|
||||
error_type: :internal_error,
|
||||
)
|
||||
end
|
||||
|
||||
def raise_connection_error!(method, params)
|
||||
raise RequestHandlerError.new(
|
||||
"Server process closed stdout unexpectedly",
|
||||
|
|
|
|||
|
|
@ -9,10 +9,25 @@ module MCP
|
|||
class StdioTransport < Transport
|
||||
STATUS_INTERRUPTED = Signal.list["INT"] + 128
|
||||
|
||||
def initialize(server)
|
||||
# Default upper bound on a single newline-delimited frame. CRuby's `IO#gets`
|
||||
# without a limit accumulates bytes until a newline arrives, so a peer that
|
||||
# never emits one can grow a single String until the process is OOM-killed.
|
||||
# 4 MiB is large enough for any realistic JSON-RPC frame, including
|
||||
# base64-embedded images.
|
||||
MAX_LINE_BYTES = 4 * 1024 * 1024
|
||||
|
||||
def initialize(server, max_line_bytes: MAX_LINE_BYTES)
|
||||
super(server)
|
||||
@open = false
|
||||
@session = nil
|
||||
# Reject `nil` or non-positive values: `IO#gets("\n", nil)` and a negative
|
||||
# limit read without an upper bound, which would silently disable the
|
||||
# protection this option exists to provide.
|
||||
unless max_line_bytes.is_a?(Integer) && max_line_bytes > 0
|
||||
raise ArgumentError, "max_line_bytes must be a positive Integer"
|
||||
end
|
||||
|
||||
@max_line_bytes = max_line_bytes
|
||||
$stdin.set_encoding("UTF-8")
|
||||
$stdout.set_encoding("UTF-8")
|
||||
end
|
||||
|
|
@ -20,7 +35,20 @@ module MCP
|
|||
def open
|
||||
@open = true
|
||||
@session = ServerSession.new(server: @server, transport: self)
|
||||
while @open && (line = $stdin.gets)
|
||||
while @open
|
||||
begin
|
||||
line = read_line($stdin)
|
||||
rescue RequestHandlerError => e
|
||||
# Stop accumulating and end the connection gracefully rather than
|
||||
# letting an unbounded read exhaust memory or escape as an uncaught
|
||||
# backtrace. Scoped to the read so genuine request errors raised while
|
||||
# handling a frame are not swallowed here.
|
||||
@open = false
|
||||
MCP.configuration.exception_reporter.call(e, { error: "stdio frame exceeds limit" })
|
||||
break
|
||||
end
|
||||
break if line.nil?
|
||||
|
||||
response = @session.handle_json(line.strip)
|
||||
send_response(response) if response
|
||||
end
|
||||
|
|
@ -73,7 +101,7 @@ module MCP
|
|||
raise
|
||||
end
|
||||
|
||||
while @open && (line = $stdin.gets)
|
||||
while @open && (line = read_line($stdin))
|
||||
begin
|
||||
parsed = JSON.parse(line.strip, symbolize_names: true)
|
||||
rescue JSON::ParserError => e
|
||||
|
|
@ -95,6 +123,26 @@ module MCP
|
|||
|
||||
raise "Transport closed while waiting for response to #{method} request."
|
||||
end
|
||||
|
||||
private
|
||||
|
||||
# Reads one newline-delimited frame, bounded by `@max_line_bytes`. Returns
|
||||
# the line (including its trailing newline) or `nil` at EOF. Raises when the
|
||||
# limit is reached before a newline arrives, which signals a peer streaming
|
||||
# an unbounded frame. A short final frame without a trailing newline (EOF) is
|
||||
# still returned, since its length stays under the limit.
|
||||
def read_line(io)
|
||||
line = io.gets("\n", @max_line_bytes)
|
||||
if line && !line.end_with?("\n") && line.bytesize >= @max_line_bytes
|
||||
raise RequestHandlerError.new(
|
||||
"stdio frame exceeds #{@max_line_bytes} bytes without a newline",
|
||||
nil,
|
||||
error_type: :internal_error,
|
||||
)
|
||||
end
|
||||
|
||||
line
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -1060,6 +1060,116 @@ module MCP
|
|||
end
|
||||
end
|
||||
|
||||
def test_new_raises_argument_error_when_max_line_bytes_is_not_a_positive_integer
|
||||
[nil, 0, -1, 1.5, "1024"].each do |invalid|
|
||||
error = assert_raises(ArgumentError) do
|
||||
Stdio.new(command: "ruby", args: ["server.rb"], max_line_bytes: invalid)
|
||||
end
|
||||
assert_equal("max_line_bytes must be a positive Integer", error.message)
|
||||
end
|
||||
end
|
||||
|
||||
def test_send_request_raises_when_response_frame_exceeds_max_line_bytes
|
||||
stdin_read, stdin_write = IO.pipe
|
||||
stdout_read, stdout_write = IO.pipe
|
||||
stderr_read, _ = IO.pipe
|
||||
|
||||
Open3.stubs(:popen3).returns([stdin_write, stdout_read, stderr_read, mock_wait_thread])
|
||||
|
||||
transport = Stdio.new(command: "ruby", args: ["server.rb"], max_line_bytes: 1024)
|
||||
|
||||
request = { jsonrpc: "2.0", id: "test-id", method: "tools/list" }
|
||||
|
||||
server_thread = Thread.new do
|
||||
init_line = stdin_read.gets
|
||||
init_request = JSON.parse(init_line)
|
||||
stdout_write.puts(JSON.generate({
|
||||
jsonrpc: "2.0",
|
||||
id: init_request["id"],
|
||||
result: {
|
||||
protocolVersion: "2025-11-25",
|
||||
capabilities: {},
|
||||
serverInfo: { name: "test-server", version: "1.0.0" },
|
||||
},
|
||||
}))
|
||||
stdout_write.flush
|
||||
|
||||
# Read initialized notification.
|
||||
stdin_read.gets
|
||||
|
||||
# Read tools/list request, then stream an unbounded frame with no newline.
|
||||
stdin_read.gets
|
||||
stdout_write.write("A" * 2000)
|
||||
stdout_write.flush
|
||||
end
|
||||
|
||||
transport.connect
|
||||
|
||||
error = assert_raises(RequestHandlerError) do
|
||||
transport.send_request(request: request)
|
||||
end
|
||||
|
||||
assert_match(/exceeds 1024 bytes without a newline/, error.message)
|
||||
assert_equal(:internal_error, error.error_type)
|
||||
ensure
|
||||
server_thread.join
|
||||
stdin_read.close
|
||||
stdin_write.close
|
||||
stdout_read.close
|
||||
stdout_write.close
|
||||
end
|
||||
|
||||
def test_send_request_accepts_response_within_custom_max_line_bytes
|
||||
stdin_read, stdin_write = IO.pipe
|
||||
stdout_read, stdout_write = IO.pipe
|
||||
stderr_read, _ = IO.pipe
|
||||
|
||||
Open3.stubs(:popen3).returns([stdin_write, stdout_read, stderr_read, mock_wait_thread])
|
||||
|
||||
transport = Stdio.new(command: "ruby", args: ["server.rb"], max_line_bytes: 1024)
|
||||
|
||||
request = { jsonrpc: "2.0", id: "test-id", method: "tools/list" }
|
||||
|
||||
server_thread = Thread.new do
|
||||
init_line = stdin_read.gets
|
||||
init_request = JSON.parse(init_line)
|
||||
stdout_write.puts(JSON.generate({
|
||||
jsonrpc: "2.0",
|
||||
id: init_request["id"],
|
||||
result: {
|
||||
protocolVersion: "2025-11-25",
|
||||
capabilities: {},
|
||||
serverInfo: { name: "test-server", version: "1.0.0" },
|
||||
},
|
||||
}))
|
||||
stdout_write.flush
|
||||
|
||||
stdin_read.gets
|
||||
|
||||
tools_line = stdin_read.gets
|
||||
tools_request = JSON.parse(tools_line)
|
||||
stdout_write.puts(JSON.generate({
|
||||
jsonrpc: "2.0",
|
||||
id: tools_request["id"],
|
||||
result: { tools: [] },
|
||||
}))
|
||||
stdout_write.flush
|
||||
end
|
||||
|
||||
transport.connect
|
||||
|
||||
response = transport.send_request(request: request)
|
||||
|
||||
assert_equal("test-id", response["id"])
|
||||
assert_equal({ "tools" => [] }, response["result"])
|
||||
ensure
|
||||
server_thread.join
|
||||
stdin_read.close
|
||||
stdin_write.close
|
||||
stdout_read.close
|
||||
stdout_write.close
|
||||
end
|
||||
|
||||
def test_send_notification_writes_json_line_without_waiting_for_response
|
||||
stdin_read, stdin_write = IO.pipe
|
||||
stdout_read, stdout_write = IO.pipe
|
||||
|
|
|
|||
|
|
@ -441,6 +441,91 @@ module MCP
|
|||
end
|
||||
end
|
||||
end
|
||||
|
||||
test "raises ArgumentError when max_line_bytes is not a positive Integer" do
|
||||
[nil, 0, -1, 1.5, "1024"].each do |invalid|
|
||||
error = assert_raises(ArgumentError) do
|
||||
StdioTransport.new(@server, max_line_bytes: invalid)
|
||||
end
|
||||
assert_equal("max_line_bytes must be a positive Integer", error.message)
|
||||
end
|
||||
end
|
||||
|
||||
test "open stops and reports when a stdin frame exceeds max_line_bytes without a newline" do
|
||||
transport = StdioTransport.new(@server, max_line_bytes: 64)
|
||||
input = StringIO.new("A" * 200) # No newline: an unbounded frame.
|
||||
output = StringIO.new
|
||||
|
||||
reported = []
|
||||
original_stdin = $stdin
|
||||
original_stdout = $stdout
|
||||
original_reporter = MCP.configuration.exception_reporter
|
||||
|
||||
begin
|
||||
$stdin = input
|
||||
$stdout = output
|
||||
MCP.configuration.exception_reporter = ->(e, ctx) { reported << [e, ctx] }
|
||||
|
||||
transport.open
|
||||
|
||||
refute(transport.instance_variable_get(:@open))
|
||||
assert_equal(1, reported.size)
|
||||
error, _ctx = reported.first
|
||||
assert_instance_of(RequestHandlerError, error)
|
||||
assert_match(/exceeds 64 bytes without a newline/, error.message)
|
||||
ensure
|
||||
$stdin = original_stdin
|
||||
$stdout = original_stdout
|
||||
MCP.configuration.exception_reporter = original_reporter
|
||||
end
|
||||
end
|
||||
|
||||
test "open processes a valid final frame that ends at EOF without a trailing newline" do
|
||||
request = { jsonrpc: "2.0", method: "ping", id: "eof" }
|
||||
input = StringIO.new(JSON.generate(request)) # No trailing newline.
|
||||
output = StringIO.new
|
||||
|
||||
original_stdin = $stdin
|
||||
original_stdout = $stdout
|
||||
|
||||
begin
|
||||
$stdin = input
|
||||
$stdout = output
|
||||
|
||||
@transport.open
|
||||
|
||||
response = JSON.parse(output.string, symbolize_names: true)
|
||||
assert_equal("eof", response[:id])
|
||||
assert_equal({}, response[:result])
|
||||
ensure
|
||||
$stdin = original_stdin
|
||||
$stdout = original_stdout
|
||||
end
|
||||
end
|
||||
|
||||
test "send_request raises when a response frame exceeds max_line_bytes without a newline" do
|
||||
transport = StdioTransport.new(@server, max_line_bytes: 64)
|
||||
input = StringIO.new("A" * 200) # No newline: an unbounded frame.
|
||||
output = StringIO.new
|
||||
|
||||
original_stdin = $stdin
|
||||
original_stdout = $stdout
|
||||
|
||||
begin
|
||||
$stdin = input
|
||||
$stdout = output
|
||||
transport.instance_variable_set(:@open, true)
|
||||
|
||||
error = assert_raises(RequestHandlerError) do
|
||||
transport.send_request("sampling/createMessage", { messages: [] })
|
||||
end
|
||||
|
||||
assert_match(/exceeds 64 bytes without a newline/, error.message)
|
||||
ensure
|
||||
$stdin = original_stdin
|
||||
$stdout = original_stdout
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
Loading…
Reference in a new issue