fix: bound request body size in SSE and Streamable HTTP transports (#839)
Caps the size of incoming HTTP POST bodies in both server transports so an oversized request can no longer exhaust server memory ## How Has This Been Tested? new tests ## Breaking Changes binary incompatible for SseServerTransport constuctors ## Types of changes - [x] Bug fix (non-breaking change which fixes a - [x] New feature (non-breaking change which adds functionality) - [x] Breaking change (fix or feature that would to change) - [ ] Documentation update ## Checklist - [x] I have read the [MCP Documentation](https: - [x] My code follows the repository's style guidelines - [x] New and existing tests pass locally - [x] I have added appropriate error handling - [x] I have added or updated documentation as needed
This commit is contained in:
Pavel Gorgulov
GitHub
parent
bbf77bae93
commit
35da063c23
8 changed files with 308 additions and 36 deletions
|
|
@ -51,12 +51,12 @@ public final class io/modelcontextprotocol/kotlin/sdk/server/HostValidationKt {
|
|||
}
|
||||
|
||||
public final class io/modelcontextprotocol/kotlin/sdk/server/KtorServerKt {
|
||||
public static final fun mcp (Lio/ktor/server/application/Application;ZLjava/util/List;Ljava/util/List;Lkotlin/jvm/functions/Function1;)V
|
||||
public static final fun mcp (Lio/ktor/server/routing/Route;Ljava/lang/String;ZLjava/util/List;Ljava/util/List;Lkotlin/jvm/functions/Function1;)V
|
||||
public static final fun mcp (Lio/ktor/server/routing/Route;ZLjava/util/List;Ljava/util/List;Lkotlin/jvm/functions/Function1;)V
|
||||
public static synthetic fun mcp$default (Lio/ktor/server/application/Application;ZLjava/util/List;Ljava/util/List;Lkotlin/jvm/functions/Function1;ILjava/lang/Object;)V
|
||||
public static synthetic fun mcp$default (Lio/ktor/server/routing/Route;Ljava/lang/String;ZLjava/util/List;Ljava/util/List;Lkotlin/jvm/functions/Function1;ILjava/lang/Object;)V
|
||||
public static synthetic fun mcp$default (Lio/ktor/server/routing/Route;ZLjava/util/List;Ljava/util/List;Lkotlin/jvm/functions/Function1;ILjava/lang/Object;)V
|
||||
public static final fun mcp (Lio/ktor/server/application/Application;ZLjava/util/List;Ljava/util/List;JLkotlin/jvm/functions/Function1;)V
|
||||
public static final fun mcp (Lio/ktor/server/routing/Route;Ljava/lang/String;ZLjava/util/List;Ljava/util/List;JLkotlin/jvm/functions/Function1;)V
|
||||
public static final fun mcp (Lio/ktor/server/routing/Route;ZLjava/util/List;Ljava/util/List;JLkotlin/jvm/functions/Function1;)V
|
||||
public static synthetic fun mcp$default (Lio/ktor/server/application/Application;ZLjava/util/List;Ljava/util/List;JLkotlin/jvm/functions/Function1;ILjava/lang/Object;)V
|
||||
public static synthetic fun mcp$default (Lio/ktor/server/routing/Route;Ljava/lang/String;ZLjava/util/List;Ljava/util/List;JLkotlin/jvm/functions/Function1;ILjava/lang/Object;)V
|
||||
public static synthetic fun mcp$default (Lio/ktor/server/routing/Route;ZLjava/util/List;Ljava/util/List;JLkotlin/jvm/functions/Function1;ILjava/lang/Object;)V
|
||||
public static final fun mcpStatelessStreamableHttp (Lio/ktor/server/application/Application;Ljava/lang/String;ZLjava/util/List;Ljava/util/List;Lio/modelcontextprotocol/kotlin/sdk/server/EventStore;Lkotlin/jvm/functions/Function1;)V
|
||||
public static synthetic fun mcpStatelessStreamableHttp$default (Lio/ktor/server/application/Application;Ljava/lang/String;ZLjava/util/List;Ljava/util/List;Lio/modelcontextprotocol/kotlin/sdk/server/EventStore;Lkotlin/jvm/functions/Function1;ILjava/lang/Object;)V
|
||||
public static final fun mcpStreamableHttp (Lio/ktor/server/application/Application;Ljava/lang/String;ZLjava/util/List;Ljava/util/List;Lio/modelcontextprotocol/kotlin/sdk/server/EventStore;Lkotlin/jvm/functions/Function1;)V
|
||||
|
|
@ -208,7 +208,8 @@ public class io/modelcontextprotocol/kotlin/sdk/server/ServerSession : io/modelc
|
|||
}
|
||||
|
||||
public final class io/modelcontextprotocol/kotlin/sdk/server/SseServerTransport : io/modelcontextprotocol/kotlin/sdk/shared/AbstractTransport {
|
||||
public fun <init> (Ljava/lang/String;Lio/ktor/server/sse/ServerSSESession;)V
|
||||
public fun <init> (Ljava/lang/String;Lio/ktor/server/sse/ServerSSESession;J)V
|
||||
public synthetic fun <init> (Ljava/lang/String;Lio/ktor/server/sse/ServerSSESession;JILkotlin/jvm/internal/DefaultConstructorMarker;)V
|
||||
public fun close (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
|
||||
public final fun getSessionId ()Ljava/lang/String;
|
||||
public final fun handleMessage (Ljava/lang/String;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
|
||||
|
|
|
|||
|
|
@ -44,6 +44,8 @@ private val logger = KotlinLogging.logger {}
|
|||
* When `null` while the localhost host defaults are in effect (no custom `allowedHosts`),
|
||||
* the `Origin` header is validated against `localhost`, `127.0.0.1`, `[::1]`.
|
||||
* With custom `allowedHosts`, `null` skips origin validation.
|
||||
* @param maxRequestBodySize maximum allowed size, in bytes, of an incoming POST body; larger requests are
|
||||
* rejected with `413 Payload Too Large`. Defaults to 4 MiB.
|
||||
* @param block factory block with access to the [ServerSSESession]
|
||||
* that creates and returns the [Server] to handle the connection.
|
||||
* @throws IllegalStateException if the [SSE] plugin is not installed.
|
||||
|
|
@ -55,10 +57,11 @@ public fun Route.mcp(
|
|||
enableDnsRebindingProtection: Boolean = true,
|
||||
allowedHosts: List<String>? = null,
|
||||
allowedOrigins: List<String>? = null,
|
||||
maxRequestBodySize: Long = DEFAULT_MAX_REQUEST_BODY_SIZE,
|
||||
block: ServerSSESession.() -> Server,
|
||||
) {
|
||||
route(path) {
|
||||
mcp(enableDnsRebindingProtection, allowedHosts, allowedOrigins, block)
|
||||
mcp(enableDnsRebindingProtection, allowedHosts, allowedOrigins, maxRequestBodySize, block)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -76,15 +79,19 @@ public fun Route.mcp(
|
|||
* When `null` while the localhost host defaults are in effect (no custom `allowedHosts`),
|
||||
* the `Origin` header is validated against `localhost`, `127.0.0.1`, `[::1]`.
|
||||
* With custom `allowedHosts`, `null` skips origin validation.
|
||||
* @param maxRequestBodySize maximum allowed size, in bytes, of an incoming POST body; larger requests are
|
||||
* rejected with `413 Payload Too Large`. Defaults to 4 MiB.
|
||||
* @param block factory block with access to the [ServerSSESession]
|
||||
* that creates and returns the [Server] to handle the connection.
|
||||
* @throws IllegalStateException if the [SSE] plugin is not installed.
|
||||
*/
|
||||
@KtorDsl
|
||||
@Suppress("LongParameterList")
|
||||
public fun Route.mcp(
|
||||
enableDnsRebindingProtection: Boolean = true,
|
||||
allowedHosts: List<String>? = null,
|
||||
allowedOrigins: List<String>? = null,
|
||||
maxRequestBodySize: Long = DEFAULT_MAX_REQUEST_BODY_SIZE,
|
||||
block: ServerSSESession.() -> Server,
|
||||
) {
|
||||
try {
|
||||
|
|
@ -103,7 +110,7 @@ public fun Route.mcp(
|
|||
val transportManager = TransportManager<SseServerTransport>()
|
||||
|
||||
sse {
|
||||
mcpSseEndpoint("", transportManager, block)
|
||||
mcpSseEndpoint("", transportManager, maxRequestBodySize, block)
|
||||
}
|
||||
|
||||
post {
|
||||
|
|
@ -126,21 +133,25 @@ public fun Route.mcp(
|
|||
* When `null` while the localhost host defaults are in effect (no custom `allowedHosts`),
|
||||
* the `Origin` header is validated against `localhost`, `127.0.0.1`, `[::1]`.
|
||||
* With custom `allowedHosts`, `null` skips origin validation.
|
||||
* @param maxRequestBodySize maximum allowed size, in bytes, of an incoming POST body; larger requests are
|
||||
* rejected with `413 Payload Too Large`. Defaults to 4 MiB.
|
||||
* @param block factory block with access to the [ServerSSESession]
|
||||
* that creates and returns the [Server] to handle the connection.
|
||||
*/
|
||||
@KtorDsl
|
||||
@Suppress("LongParameterList")
|
||||
public fun Application.mcp(
|
||||
enableDnsRebindingProtection: Boolean = true,
|
||||
allowedHosts: List<String>? = null,
|
||||
allowedOrigins: List<String>? = null,
|
||||
maxRequestBodySize: Long = DEFAULT_MAX_REQUEST_BODY_SIZE,
|
||||
block: ServerSSESession.() -> Server,
|
||||
) {
|
||||
installMcpContentNegotiation()
|
||||
install(SSE)
|
||||
|
||||
routing {
|
||||
mcp(enableDnsRebindingProtection, allowedHosts, allowedOrigins, block)
|
||||
mcp(enableDnsRebindingProtection, allowedHosts, allowedOrigins, maxRequestBodySize, block)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -327,9 +338,10 @@ public fun Application.mcpStatelessStreamableHttp(
|
|||
private suspend fun ServerSSESession.mcpSseEndpoint(
|
||||
postEndpoint: String,
|
||||
transportManager: TransportManager<SseServerTransport>,
|
||||
maxRequestBodySize: Long,
|
||||
block: ServerSSESession.() -> Server,
|
||||
) {
|
||||
val transport = mcpSseTransport(postEndpoint, transportManager)
|
||||
val transport = mcpSseTransport(postEndpoint, transportManager, maxRequestBodySize)
|
||||
|
||||
val server = block()
|
||||
|
||||
|
|
@ -348,8 +360,9 @@ private suspend fun ServerSSESession.mcpSseEndpoint(
|
|||
private fun ServerSSESession.mcpSseTransport(
|
||||
postEndpoint: String,
|
||||
transportManager: TransportManager<SseServerTransport>,
|
||||
maxRequestBodySize: Long,
|
||||
): SseServerTransport {
|
||||
val transport = SseServerTransport(postEndpoint, this)
|
||||
val transport = SseServerTransport(postEndpoint, this, maxRequestBodySize)
|
||||
transportManager.addTransport(transport.sessionId, transport)
|
||||
logger.info { "New SSE connection established and stored with sessionId: ${transport.sessionId}" }
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,34 @@
|
|||
package io.modelcontextprotocol.kotlin.sdk.server
|
||||
|
||||
import io.ktor.server.application.ApplicationCall
|
||||
import io.ktor.server.request.receiveChannel
|
||||
import io.ktor.utils.io.readRemaining
|
||||
import kotlinx.io.Buffer
|
||||
import kotlinx.io.IOException
|
||||
import kotlinx.io.readString
|
||||
|
||||
/** Default maximum size (in bytes) for an incoming HTTP request body: 4 MiB. */
|
||||
internal const val DEFAULT_MAX_REQUEST_BODY_SIZE: Long = 4L * 1024 * 1024
|
||||
|
||||
/** Signals that an incoming request body exceeded the maximum size a transport is willing to read. */
|
||||
internal class RequestBodyTooLargeException(val maxBodySize: Long) :
|
||||
IOException("Request body exceeds the maximum of $maxBodySize bytes.")
|
||||
|
||||
/**
|
||||
* Reads the request body as a UTF-8 string, rejecting bodies larger than [maxBytes].
|
||||
*
|
||||
* At most `maxBytes + 1` bytes are ever pulled from the request channel, so an oversized or unframed
|
||||
* body cannot exhaust memory regardless of the (untrusted) `Content-Length` header.
|
||||
*
|
||||
* @throws RequestBodyTooLargeException if the body exceeds [maxBytes]
|
||||
*/
|
||||
internal suspend fun ApplicationCall.receiveTextWithLimit(maxBytes: Long): String {
|
||||
val buffer = Buffer()
|
||||
val readLimit = if (maxBytes == Long.MAX_VALUE) maxBytes else maxBytes + 1
|
||||
val read = receiveChannel().readRemaining(readLimit).transferTo(buffer)
|
||||
if (read > maxBytes) {
|
||||
buffer.clear()
|
||||
throw RequestBodyTooLargeException(maxBytes)
|
||||
}
|
||||
return buffer.readString()
|
||||
}
|
||||
|
|
@ -5,7 +5,6 @@ import io.ktor.http.HttpStatusCode
|
|||
import io.ktor.http.encodeURLPath
|
||||
import io.ktor.server.application.ApplicationCall
|
||||
import io.ktor.server.request.contentType
|
||||
import io.ktor.server.request.receiveText
|
||||
import io.ktor.server.response.respondText
|
||||
import io.ktor.server.sse.ServerSSESession
|
||||
import io.modelcontextprotocol.kotlin.sdk.shared.AbstractTransport
|
||||
|
|
@ -29,10 +28,19 @@ internal const val SESSION_ID_PARAM = "sessionId"
|
|||
*
|
||||
* @param endpoint relative or absolute URL the client will POST messages to
|
||||
* @param session active SSE session used to deliver server-to-client events
|
||||
* @param maxRequestBodySize maximum allowed size, in bytes, of an incoming POST body; larger requests are
|
||||
* rejected with `413 Payload Too Large` without being buffered in full. Defaults to 4 MiB.
|
||||
*/
|
||||
@OptIn(ExperimentalAtomicApi::class)
|
||||
public class SseServerTransport(private val endpoint: String, private val session: ServerSSESession) :
|
||||
AbstractTransport() {
|
||||
public class SseServerTransport(
|
||||
private val endpoint: String,
|
||||
private val session: ServerSSESession,
|
||||
private val maxRequestBodySize: Long = DEFAULT_MAX_REQUEST_BODY_SIZE,
|
||||
) : AbstractTransport() {
|
||||
init {
|
||||
require(maxRequestBodySize > 0) { "maxRequestBodySize must be greater than 0" }
|
||||
}
|
||||
|
||||
private val initialized: AtomicBoolean = AtomicBoolean(false)
|
||||
|
||||
/** Unique identifier for this transport session, generated randomly on creation. */
|
||||
|
|
@ -77,6 +85,7 @@ public class SseServerTransport(private val endpoint: String, private val sessio
|
|||
val message = "SSE connection not established"
|
||||
call.respondText(message, status = HttpStatusCode.InternalServerError)
|
||||
_onError.invoke(IllegalStateException(message))
|
||||
return
|
||||
}
|
||||
|
||||
val body = try {
|
||||
|
|
@ -85,9 +94,13 @@ public class SseServerTransport(private val endpoint: String, private val sessio
|
|||
error("Unsupported content-type: $ct")
|
||||
}
|
||||
|
||||
call.receiveText()
|
||||
call.receiveTextWithLimit(maxRequestBodySize)
|
||||
} catch (e: CancellationException) {
|
||||
throw e
|
||||
} catch (e: RequestBodyTooLargeException) {
|
||||
call.respondText(e.message ?: "Request body too large", status = HttpStatusCode.PayloadTooLarge)
|
||||
_onError.invoke(e)
|
||||
return
|
||||
} catch (e: Exception) {
|
||||
call.respondText("Invalid message: ${e.message}", status = HttpStatusCode.BadRequest)
|
||||
_onError.invoke(e)
|
||||
|
|
|
|||
|
|
@ -8,7 +8,6 @@ import io.ktor.server.application.ApplicationCall
|
|||
import io.ktor.server.request.contentType
|
||||
import io.ktor.server.request.header
|
||||
import io.ktor.server.request.httpMethod
|
||||
import io.ktor.server.request.receiveText
|
||||
import io.ktor.server.response.header
|
||||
import io.ktor.server.response.respond
|
||||
import io.ktor.server.response.respondNullable
|
||||
|
|
@ -48,7 +47,6 @@ import kotlin.uuid.Uuid
|
|||
internal const val MCP_SESSION_ID_HEADER = "mcp-session-id"
|
||||
private const val MCP_PROTOCOL_VERSION_HEADER = "mcp-protocol-version"
|
||||
private const val MCP_RESUMPTION_TOKEN_HEADER = "Last-Event-ID"
|
||||
private const val DEFAULT_MAX_REQUEST_BODY_SIZE: Long = 4L * 1024 * 1024 // 4 MB
|
||||
private const val MIN_PRIMING_EVENT_PROTOCOL_VERSION = "2025-11-25"
|
||||
|
||||
/**
|
||||
|
|
@ -703,23 +701,13 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
|
|||
}
|
||||
|
||||
private suspend fun parseBody(call: ApplicationCall): List<JSONRPCMessage>? {
|
||||
val maxSize = configuration.maxRequestBodySize
|
||||
val contentLength = call.request.header(HttpHeaders.ContentLength)?.toLongOrNull() ?: 0L
|
||||
if (contentLength > maxSize) {
|
||||
val body = try {
|
||||
call.receiveTextWithLimit(configuration.maxRequestBodySize)
|
||||
} catch (e: RequestBodyTooLargeException) {
|
||||
call.reject(
|
||||
HttpStatusCode.PayloadTooLarge,
|
||||
RPCError.ErrorCode.INVALID_REQUEST,
|
||||
"Invalid Request: message size exceeds maximum of $maxSize bytes",
|
||||
)
|
||||
return null
|
||||
}
|
||||
|
||||
val body = call.receiveText()
|
||||
if (body.length.toLong() > maxSize) {
|
||||
call.reject(
|
||||
HttpStatusCode.PayloadTooLarge,
|
||||
RPCError.ErrorCode.INVALID_REQUEST,
|
||||
"Invalid Request: message size exceeds maximum of $maxSize bytes",
|
||||
"Invalid Request: message size exceeds maximum of ${e.maxBodySize} bytes",
|
||||
)
|
||||
return null
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,103 @@
|
|||
package io.modelcontextprotocol.kotlin.sdk.server
|
||||
|
||||
import io.kotest.matchers.shouldBe
|
||||
import io.ktor.client.request.post
|
||||
import io.ktor.client.request.setBody
|
||||
import io.ktor.client.statement.bodyAsText
|
||||
import io.ktor.http.HttpStatusCode
|
||||
import io.ktor.server.response.respondText
|
||||
import io.ktor.server.routing.post
|
||||
import io.ktor.server.routing.routing
|
||||
import io.ktor.server.testing.ApplicationTestBuilder
|
||||
import io.ktor.server.testing.testApplication
|
||||
import kotlin.test.Test
|
||||
|
||||
class RequestBodyTest {
|
||||
|
||||
/** Wires a POST endpoint that echoes the body when it fits, or replies 413 when it is too large. */
|
||||
private fun ApplicationTestBuilder.installEchoEndpoint(maxBytes: Long) {
|
||||
install(io.ktor.server.sse.SSE)
|
||||
application {
|
||||
routing {
|
||||
post("/echo") {
|
||||
val text = try {
|
||||
call.receiveTextWithLimit(maxBytes)
|
||||
} catch (e: RequestBodyTooLargeException) {
|
||||
call.respondText(e.message ?: "too large", status = HttpStatusCode.PayloadTooLarge)
|
||||
return@post
|
||||
}
|
||||
call.respondText(text)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `body under the limit is returned intact`() = testApplication {
|
||||
installEchoEndpoint(maxBytes = 1024)
|
||||
val payload = "hello world"
|
||||
|
||||
val response = client.post("/echo") { setBody(payload) }
|
||||
|
||||
response.status shouldBe HttpStatusCode.OK
|
||||
response.bodyAsText() shouldBe payload
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `body exactly at the limit is accepted`() = testApplication {
|
||||
val payload = "x".repeat(64)
|
||||
installEchoEndpoint(maxBytes = payload.length.toLong())
|
||||
|
||||
val response = client.post("/echo") { setBody(payload) }
|
||||
|
||||
response.status shouldBe HttpStatusCode.OK
|
||||
response.bodyAsText() shouldBe payload
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `body exceeding the limit is rejected with 413`() = testApplication {
|
||||
val payload = "x".repeat(65)
|
||||
installEchoEndpoint(maxBytes = 64)
|
||||
|
||||
val response = client.post("/echo") { setBody(payload) }
|
||||
|
||||
response.status shouldBe HttpStatusCode.PayloadTooLarge
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `large body exceeding the limit is rejected without buffering it whole`() = testApplication {
|
||||
// 8 MB body against a 4 MB limit: must be rejected.
|
||||
installEchoEndpoint(maxBytes = 4L * 1024 * 1024)
|
||||
val payload = "x".repeat(8 * 1024 * 1024)
|
||||
|
||||
val response = client.post("/echo") { setBody(payload) }
|
||||
|
||||
response.status shouldBe HttpStatusCode.PayloadTooLarge
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `multibyte UTF-8 body larger than a read chunk decodes intact`() = testApplication {
|
||||
// Emoji is 4 bytes in UTF-8; a payload larger than the internal read chunk forces the
|
||||
// accumulated bytes to span multiple chunks. Decoding once at the end must not corrupt
|
||||
// characters that straddle a chunk boundary.
|
||||
val unit = "a😀b" // mixes 1- and 4-byte code units so boundaries land mid-character
|
||||
val payload = unit.repeat(40_000) // well over a 64 KiB read chunk
|
||||
installEchoEndpoint(maxBytes = payload.encodeToByteArray().size.toLong())
|
||||
|
||||
val response = client.post("/echo") { setBody(payload) }
|
||||
|
||||
response.status shouldBe HttpStatusCode.OK
|
||||
response.bodyAsText() shouldBe payload
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `body is returned intact when the limit is Long MAX_VALUE`() = testApplication {
|
||||
installEchoEndpoint(maxBytes = Long.MAX_VALUE)
|
||||
val payload = "hello world"
|
||||
|
||||
val response = client.post("/echo") { setBody(payload) }
|
||||
|
||||
response.status shouldBe HttpStatusCode.OK
|
||||
response.bodyAsText() shouldBe payload
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,90 @@
|
|||
package io.modelcontextprotocol.kotlin.sdk.server
|
||||
|
||||
import io.kotest.matchers.shouldBe
|
||||
import io.ktor.client.request.post
|
||||
import io.ktor.client.request.prepareGet
|
||||
import io.ktor.client.request.setBody
|
||||
import io.ktor.client.statement.HttpResponse
|
||||
import io.ktor.client.statement.bodyAsChannel
|
||||
import io.ktor.http.ContentType
|
||||
import io.ktor.http.HttpStatusCode
|
||||
import io.ktor.http.contentType
|
||||
import io.ktor.server.application.install
|
||||
import io.ktor.server.routing.post
|
||||
import io.ktor.server.routing.routing
|
||||
import io.ktor.server.sse.SSE
|
||||
import io.ktor.server.sse.ServerSSESession
|
||||
import io.ktor.server.testing.testApplication
|
||||
import io.ktor.utils.io.readLine
|
||||
import io.mockk.mockk
|
||||
import java.util.concurrent.atomic.AtomicBoolean
|
||||
import kotlin.test.Test
|
||||
|
||||
class SseServerTransportTest : AbstractKtorExtensionsTest() {
|
||||
|
||||
@Test
|
||||
fun `handlePostMessage on a not-started transport does not deliver the message`() = testApplication {
|
||||
// A registered onMessage callback opens the delivery gate, so if the not-initialized branch
|
||||
// falls through it would wrongly hand the message to the application. It must return instead.
|
||||
val transport = SseServerTransport("/messages", mockk<ServerSSESession>(relaxed = true))
|
||||
val delivered = AtomicBoolean(false)
|
||||
transport.onMessage { delivered.set(true) }
|
||||
|
||||
application {
|
||||
routing {
|
||||
post("/messages") { transport.handlePostMessage(call) }
|
||||
}
|
||||
}
|
||||
|
||||
val response = client.post("/messages") {
|
||||
contentType(ContentType.Application.Json)
|
||||
setBody("""{"jsonrpc":"2.0","id":1,"method":"ping"}""")
|
||||
}
|
||||
|
||||
response.status shouldBe HttpStatusCode.InternalServerError
|
||||
delivered.get() shouldBe false
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `SSE POST exceeding maxRequestBodySize is rejected with 413`() = testApplication {
|
||||
application {
|
||||
install(SSE)
|
||||
routing {
|
||||
mcp(enableDnsRebindingProtection = false, maxRequestBodySize = MAX_BODY) { testServer() }
|
||||
}
|
||||
}
|
||||
|
||||
client.prepareGet("/").execute { response ->
|
||||
val sessionId = response.readSessionId()
|
||||
requireNotNull(sessionId) { "sessionId not found in SSE endpoint event" }
|
||||
|
||||
// A body one byte over the configured limit must be rejected before processing.
|
||||
val oversized = "x".repeat((MAX_BODY + 1).toInt())
|
||||
val postResponse = client.post("/?sessionId=$sessionId") {
|
||||
contentType(ContentType.Application.Json)
|
||||
setBody(oversized)
|
||||
}
|
||||
postResponse.status shouldBe HttpStatusCode.PayloadTooLarge
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun HttpResponse.readSessionId(): String? {
|
||||
val channel = bodyAsChannel()
|
||||
var eventName: String? = null
|
||||
while (!channel.isClosedForRead) {
|
||||
val line = channel.readLine() ?: break
|
||||
when {
|
||||
line.startsWith("event:") -> eventName = line.substringAfter("event:").trim()
|
||||
|
||||
line.startsWith("data:") && eventName == "endpoint" -> {
|
||||
return line.substringAfter("data:").trim().substringAfter("sessionId=").ifEmpty { null }
|
||||
}
|
||||
}
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
private companion object {
|
||||
const val MAX_BODY = 1024L
|
||||
}
|
||||
}
|
||||
|
|
@ -24,6 +24,8 @@ import io.ktor.server.routing.post
|
|||
import io.ktor.server.routing.routing
|
||||
import io.ktor.server.testing.ApplicationTestBuilder
|
||||
import io.ktor.server.testing.testApplication
|
||||
import io.ktor.utils.io.ByteReadChannel
|
||||
import io.ktor.utils.io.readLine
|
||||
import io.ktor.utils.io.readUTF8Line
|
||||
import io.modelcontextprotocol.kotlin.sdk.types.ClientCapabilities
|
||||
import io.modelcontextprotocol.kotlin.sdk.types.EmptyResult
|
||||
|
|
@ -421,6 +423,34 @@ class StreamableHttpServerTransportTest {
|
|||
response.status shouldBe HttpStatusCode.PayloadTooLarge
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `POST with oversized chunked body and no Content-Length returns 413`() = testApplication {
|
||||
configTestServer()
|
||||
|
||||
val client = createTestClient()
|
||||
|
||||
val transport = StreamableHttpServerTransport(
|
||||
StreamableHttpServerTransport.Configuration(enableJsonResponse = true, maxRequestBodySize = 1024),
|
||||
)
|
||||
transport.onMessage { message ->
|
||||
if (message is JSONRPCRequest) {
|
||||
transport.send(JSONRPCResponse(message.id, EmptyResult()))
|
||||
}
|
||||
}
|
||||
|
||||
configureTransportEndpoint(transport)
|
||||
|
||||
// Stream the body as a channel so the client uses chunked transfer-encoding with no
|
||||
// Content-Length header: the size limit must hold without trusting the (absent) header.
|
||||
val oversized = ByteReadChannel("x".repeat(4096).encodeToByteArray())
|
||||
val response = client.post(path) {
|
||||
addStreamableHeaders()
|
||||
setBody(oversized)
|
||||
}
|
||||
|
||||
response.status shouldBe HttpStatusCode.PayloadTooLarge
|
||||
}
|
||||
|
||||
@ParameterizedTest
|
||||
@MethodSource("maxBodySizeTestCases")
|
||||
fun `POST with custom max request body size validates payload size`(
|
||||
|
|
@ -507,7 +537,7 @@ class StreamableHttpServerTransportTest {
|
|||
|
||||
// New stream is alive
|
||||
val secondChannel = secondResponse.bodyAsChannel()
|
||||
val firstLine = secondChannel.readUTF8Line()
|
||||
val firstLine = secondChannel.readLine()
|
||||
firstLine.shouldNotBeNull()
|
||||
secondChannel.isClosedForRead shouldBe false
|
||||
}
|
||||
|
|
@ -546,7 +576,7 @@ class StreamableHttpServerTransportTest {
|
|||
header("mcp-protocol-version", LATEST_PROTOCOL_VERSION)
|
||||
}.execute { response ->
|
||||
response.status shouldBe HttpStatusCode.OK
|
||||
response.bodyAsChannel().readUTF8Line()
|
||||
response.bodyAsChannel().readLine()
|
||||
}
|
||||
|
||||
// Step 3: Immediately reconnect — the transport should close the stale
|
||||
|
|
@ -561,7 +591,7 @@ class StreamableHttpServerTransportTest {
|
|||
response.headers[MCP_SESSION_ID_HEADER] shouldBe sessionId
|
||||
|
||||
val channel = response.bodyAsChannel()
|
||||
val firstLine = channel.readUTF8Line()
|
||||
val firstLine = channel.readLine()
|
||||
firstLine.shouldNotBeNull()
|
||||
channel.isClosedForRead shouldBe false
|
||||
}
|
||||
|
|
@ -604,7 +634,7 @@ class StreamableHttpServerTransportTest {
|
|||
|
||||
// Verify the stream is alive by reading at least one line (flush event)
|
||||
val channel = response.bodyAsChannel()
|
||||
val firstLine = channel.readUTF8Line()
|
||||
val firstLine = channel.readLine()
|
||||
firstLine.shouldNotBeNull()
|
||||
channel.isClosedForRead shouldBe false
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue