diff --git a/crates/rmcp/src/transport/streamable_http_server/tower.rs b/crates/rmcp/src/transport/streamable_http_server/tower.rs index 2b0772b21..76046935f 100644 --- a/crates/rmcp/src/transport/streamable_http_server/tower.rs +++ b/crates/rmcp/src/transport/streamable_http_server/tower.rs @@ -55,6 +55,59 @@ use crate::{ pub(crate) const DEFAULT_MAX_REQUEST_BODY_BYTES: usize = 4 * 1024 * 1024; const STATELESS_STREAM_CHANNEL_CAPACITY: usize = 16; +// HTTP lists and parameter lists may contain delimiters in quoted strings. +// A quoted-pair keeps an escaped quote from ending the string. +fn split_unquoted(value: &str, delimiter: char) -> impl Iterator { + let mut quoted = false; + let mut escaped = false; + value.split(move |character| { + if escaped { + escaped = false; + } else if quoted && character == '\\' { + escaped = true; + } else if character == '"' { + quoted = !quoted; + } else if !quoted && character == delimiter { + return true; + } + false + }) +} + +// MCP requires explicit media types, rather than general wildcard negotiation. +fn accepts_media_type(headers: &HeaderMap, expected: &str) -> bool { + headers.get_all(http::header::ACCEPT).iter().any(|value| { + let Ok(value) = value.to_str() else { + return false; + }; + split_unquoted(value, ',').any(|entry| { + let mut parts = split_unquoted(entry, ';'); + parts + .next() + .is_some_and(|essence| essence.trim().eq_ignore_ascii_case(expected)) + && parts.all(|parameter| { + let Some((name, value)) = parameter.split_once('=') else { + return true; + }; + !name.trim().eq_ignore_ascii_case("q") || quality_is_acceptable(value.trim()) + }) + }) + }) +} + +// RFC 9110 section 12.4.2 permits up to three fractional digits. +fn quality_is_acceptable(quality: &str) -> bool { + let (whole, fraction) = quality.split_once('.').unwrap_or((quality, "")); + if fraction.len() > 3 || !fraction.bytes().all(|digit| digit.is_ascii_digit()) { + return false; + } + match whole { + "0" => fraction.bytes().any(|digit| digit != b'0'), + "1" => fraction.bytes().all(|digit| digit == b'0'), + _ => false, + } +} + struct ErrorResponse(Box); impl ErrorResponse { @@ -1847,12 +1900,7 @@ where B::Error: Display, { // check accept header - if !request - .headers() - .get(http::header::ACCEPT) - .and_then(|header| header.to_str().ok()) - .is_some_and(|header| header.contains(EVENT_STREAM_MIME_TYPE)) - { + if !accepts_media_type(request.headers(), EVENT_STREAM_MIME_TYPE) { return Ok(Response::builder() .status(http::StatusCode::NOT_ACCEPTABLE) .body( @@ -1986,13 +2034,8 @@ where B::Error: Display, { // check accept header - if !request - .headers() - .get(http::header::ACCEPT) - .and_then(|header| header.to_str().ok()) - .is_some_and(|header| { - header.contains(JSON_MIME_TYPE) && header.contains(EVENT_STREAM_MIME_TYPE) - }) + if !accepts_media_type(request.headers(), JSON_MIME_TYPE) + || !accepts_media_type(request.headers(), EVENT_STREAM_MIME_TYPE) { return Ok(Response::builder() .status(http::StatusCode::NOT_ACCEPTABLE) @@ -2005,7 +2048,12 @@ where .headers() .get(http::header::CONTENT_TYPE) .and_then(|header| header.to_str().ok()) - .is_some_and(|header| header.starts_with(JSON_MIME_TYPE)) + .is_some_and(|header| { + header + .split(';') + .next() + .is_some_and(|essence| essence.trim().eq_ignore_ascii_case(JSON_MIME_TYPE)) + }) { return Ok(Response::builder() .status(http::StatusCode::UNSUPPORTED_MEDIA_TYPE) diff --git a/crates/rmcp/tests/test_streamable_http_standard_headers.rs b/crates/rmcp/tests/test_streamable_http_standard_headers.rs index e531ac00f..f6f1e4dba 100644 --- a/crates/rmcp/tests/test_streamable_http_standard_headers.rs +++ b/crates/rmcp/tests/test_streamable_http_standard_headers.rs @@ -45,11 +45,21 @@ async fn spawn_server() -> (reqwest::Client, String, CancellationToken) { async fn spawn_server_with( service_factory: impl Fn() -> Result + Send + Sync + 'static, ) -> (reqwest::Client, String, CancellationToken) +where + S: ServerHandler + Send + 'static, +{ + spawn_server_with_mode(service_factory, false).await +} + +async fn spawn_server_with_mode( + service_factory: impl Fn() -> Result + Send + Sync + 'static, + legacy_session_mode: bool, +) -> (reqwest::Client, String, CancellationToken) where S: ServerHandler + Send + 'static, { let config = StreamableHttpServerConfig::default() - .with_legacy_session_mode(false) + .with_legacy_session_mode(legacy_session_mode) .with_json_response(true) .with_sse_keep_alive(None) .with_cancellation_token(CancellationToken::new()); @@ -495,3 +505,134 @@ async fn accepts_initialize_with_matching_mcp_method() -> anyhow::Result<()> { ct.cancel(); Ok(()) } + +#[tokio::test] +async fn media_type_headers_follow_http_semantics() { + let (client, url, ct) = spawn_server().await; + let cases: &[(&str, &[&str], &str, u16)] = &[ + ( + "ordinary", + &["application/json, text/event-stream"], + "application/json", + 202, + ), + ( + "content type case", + &["application/json, text/event-stream"], + "Application/JSON", + 202, + ), + ( + "accept case", + &["APPLICATION/JSON, TEXT/EVENT-STREAM"], + "application/json", + 202, + ), + ( + "split accept", + &["application/json", "text/event-stream"], + "application/json", + 202, + ), + ( + "content type suffix", + &["application/json, text/event-stream"], + "application/jsonx", + 415, + ), + ( + "accept suffix", + &["application/json-seq, text/event-stream"], + "application/json", + 406, + ), + ( + "zero quality", + &["application/json;q=0, text/event-stream"], + "application/json", + 406, + ), + ( + "zero quality decimal", + &["application/json;Q=0.000, text/event-stream"], + "application/json", + 406, + ), + ( + "positive quality", + &["application/json;q=0.001, text/event-stream"], + "application/json", + 202, + ), + ( + "parameters", + &["application/json;profile=\"a,b;c\", text/event-stream"], + "application/json; charset=utf-8", + 202, + ), + ( + "parameter is not media", + &["text/plain;profile=\"application/json, text/event-stream\""], + "application/json", + 406, + ), + ( + "escaped quoted parameter", + &[r#"application/json;profile="a\",b;c", text/event-stream"#], + "application/json", + 202, + ), + ( + "quoted quality is not a weight", + &[r#"application/json;profile="a;q=0", text/event-stream"#], + "application/json", + 202, + ), + ( + "escaped parameter is not media", + &[r#"text/plain;profile="a\",application/json, text/event-stream""#], + "application/json", + 406, + ), + ("wildcard alone", &["*/*"], "application/json", 406), + ]; + let mut failures = Vec::new(); + for (name, accept, content_type, expected) in cases { + let mut request = client + .post(&url) + .header("Content-Type", *content_type) + .body(r#"{"jsonrpc":"2.0","method":"notifications/initialized"}"#); + for value in *accept { + request = request.header("Accept", *value); + } + let actual = request.send().await.unwrap().status().as_u16(); + eprintln!("POST {name}: expected {expected}, got {actual}"); + if actual != *expected { + failures.push(format!("{name}: expected {expected}, got {actual}")); + } + } + ct.cancel(); + // Legacy mode dispatches GET through media validation before requiring a session ID. + let (client, url, ct) = spawn_server_with_mode(|| Ok(HeaderValidationServer), true).await; + for (accept, expected) in [ + ("text/event-stream", 400), + ("TEXT/EVENT-STREAM", 400), + ("text/event-streamx", 406), + ("text/event-stream;q=0.0", 406), + ] { + let actual = client + .get(&url) + .header("Accept", accept) + .send() + .await + .unwrap() + .status() + .as_u16(); + eprintln!("GET {accept}: expected {expected}, got {actual}"); + if actual != expected { + failures.push(format!("GET {accept}: expected {expected}, got {actual}")); + } + } + ct.cancel(); + assert!(failures.is_empty(), "{}", failures.join("\n")); +}