fix(transport): match explicit HTTP media types and Accept quality values

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<Item = &str> {
+    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<BoxResponse>);
 
 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)

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<S>(
     service_factory: impl Fn() -> Result<S, std::io::Error> + 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<S>(
+    service_factory: impl Fn() -> Result<S, std::io::Error> + 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"));
+}