Skip to content

Commit c818296

Browse files
committed
fix(transport): follow same-origin 307/308 redirects in the default Streamable HTTP client
The default reqwest client used `Policy::none()`, so any redirect from an MCP server failed with `UnexpectedServerResponse("HTTP 307: ...")`. It now follows 307 and 308 when the target has the same origin as the original request, up to 10 hops. Those keep the method and body, so the JSON-RPC message and the session, auth and custom headers arrive intact. 301, 302 and 303 are still returned to the caller, because following them turns the POST into a bodyless GET. Cross-origin redirects are still returned too, which keeps the header-leak guard from #936. Reported-by: Mo (@trickyfalcon)
1 parent 8d226f4 commit c818296

1 file changed

Lines changed: 279 additions & 3 deletions

File tree

‎crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs‎

Lines changed: 279 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -395,17 +395,45 @@ impl StreamableHttpClientTransport<reqwest::Client> {
395395
/// TCP Delayed ACK on Linux when the previous response body was not
396396
/// fully consumed before the pool attempts to reuse the connection.
397397
///
398-
/// Automatic redirects are disabled so caller-supplied custom headers
399-
/// cannot be replayed to a redirect target.
398+
/// Only same-origin 307 and 308 redirects are followed, up to
399+
/// [`MAX_REDIRECTS`] hops. Those keep the method and body, so the
400+
/// JSON-RPC message reaches the new location intact. A 301, 302 or 303
401+
/// would turn a POST into a bodyless GET and lose the message, and a
402+
/// cross-origin redirect would replay the session id and caller-supplied
403+
/// custom headers to another host, so both are returned to the caller as
404+
/// the redirect response instead.
400405
fn default_http_client() -> reqwest::Client {
401406
reqwest::Client::builder()
402407
.pool_max_idle_per_host(0)
403-
.redirect(reqwest::redirect::Policy::none())
408+
.redirect(same_origin_redirect_policy())
404409
.build()
405410
.expect("failed to build default reqwest client")
406411
}
407412
}
408413

414+
/// Maximum number of redirects the default client follows for one request.
415+
const MAX_REDIRECTS: usize = 10;
416+
417+
fn same_origin_redirect_policy() -> reqwest::redirect::Policy {
418+
reqwest::redirect::Policy::custom(|attempt| {
419+
let preserves_body = matches!(
420+
attempt.status(),
421+
reqwest::StatusCode::TEMPORARY_REDIRECT | reqwest::StatusCode::PERMANENT_REDIRECT
422+
);
423+
let same_origin = attempt
424+
.previous()
425+
.first()
426+
.is_some_and(|original| original.origin() == attempt.url().origin());
427+
if !preserves_body || !same_origin {
428+
attempt.stop()
429+
} else if attempt.previous().len() > MAX_REDIRECTS {
430+
attempt.error(format!("exceeded {MAX_REDIRECTS} redirects"))
431+
} else {
432+
attempt.follow()
433+
}
434+
})
435+
}
436+
409437
#[cfg(test)]
410438
mod tests {
411439
use rstest::rstest;
@@ -640,4 +668,252 @@ mod tests {
640668

641669
Ok(())
642670
}
671+
672+
/// What the redirect target saw for a single request.
673+
#[derive(Debug, Default)]
674+
struct CapturedRequest {
675+
method: String,
676+
body: Vec<u8>,
677+
session_id: Option<String>,
678+
authorization: Option<String>,
679+
api_key: Option<String>,
680+
}
681+
682+
/// Serves `POST /mcp` with a redirect of `status` to `/v2/mcp` on the same
683+
/// origin, and records every request that reaches `/v2/mcp`.
684+
async fn spawn_same_origin_redirect_server(
685+
status: http::StatusCode,
686+
) -> anyhow::Result<(
687+
std::net::SocketAddr,
688+
std::sync::Arc<tokio::sync::Mutex<Vec<CapturedRequest>>>,
689+
tokio::task::JoinHandle<std::io::Result<()>>,
690+
)> {
691+
use std::{net::SocketAddr, sync::Arc};
692+
693+
use axum::{
694+
Router,
695+
body::Bytes,
696+
extract::State,
697+
response::IntoResponse,
698+
routing::{any, post},
699+
};
700+
use http::{HeaderMap, Method, header::LOCATION};
701+
use tokio::sync::Mutex;
702+
703+
use crate::transport::common::http_header::HEADER_SESSION_ID;
704+
705+
type Captured = Arc<Mutex<Vec<CapturedRequest>>>;
706+
707+
fn header(headers: &HeaderMap, name: &str) -> Option<String> {
708+
headers
709+
.get(name)
710+
.and_then(|value| value.to_str().ok())
711+
.map(str::to_owned)
712+
}
713+
714+
async fn target_handler(
715+
State(captured): State<Captured>,
716+
method: Method,
717+
headers: HeaderMap,
718+
body: Bytes,
719+
) -> impl IntoResponse {
720+
captured.lock().await.push(CapturedRequest {
721+
method: method.to_string(),
722+
body: body.to_vec(),
723+
session_id: header(&headers, HEADER_SESSION_ID),
724+
authorization: header(&headers, http::header::AUTHORIZATION.as_str()),
725+
api_key: header(&headers, "x-api-key"),
726+
});
727+
(
728+
http::StatusCode::OK,
729+
[(http::header::CONTENT_TYPE, "application/json")],
730+
r#"{"jsonrpc":"2.0","id":1,"result":{}}"#,
731+
)
732+
}
733+
734+
let listener = tokio::net::TcpListener::bind(SocketAddr::from(([127, 0, 0, 1], 0))).await?;
735+
let addr = listener.local_addr()?;
736+
let location = format!("http://{addr}/v2/mcp");
737+
let captured: Captured = Arc::new(Mutex::new(Vec::new()));
738+
let app = Router::new()
739+
.route(
740+
"/mcp",
741+
post(move || {
742+
let location = location.clone();
743+
async move { (status, [(LOCATION, location)], "") }
744+
}),
745+
)
746+
.route("/v2/mcp", any(target_handler))
747+
.with_state(captured.clone());
748+
let server = tokio::spawn(async move { axum::serve(listener, app).await });
749+
Ok((addr, captured, server))
750+
}
751+
752+
fn ping_message() -> ClientJsonRpcMessage {
753+
ClientJsonRpcMessage::request(
754+
ClientRequest::PingRequest(PingRequest::default()),
755+
RequestId::Number(1),
756+
)
757+
}
758+
759+
#[rstest]
760+
#[case::temporary_redirect(http::StatusCode::TEMPORARY_REDIRECT)]
761+
#[case::permanent_redirect(http::StatusCode::PERMANENT_REDIRECT)]
762+
#[tokio::test]
763+
async fn default_http_client_follows_same_origin_body_preserving_redirect(
764+
#[case] status: http::StatusCode,
765+
) -> anyhow::Result<()> {
766+
use std::{collections::HashMap, sync::Arc};
767+
768+
use http::{HeaderName, HeaderValue};
769+
770+
use super::StreamableHttpClientTransport;
771+
use crate::transport::streamable_http_client::{
772+
StreamableHttpClient, StreamableHttpPostResponse,
773+
};
774+
775+
let (addr, captured, server) = spawn_same_origin_redirect_server(status).await?;
776+
let mut custom_headers = HashMap::new();
777+
custom_headers.insert(
778+
HeaderName::from_static("x-api-key"),
779+
HeaderValue::from_static("secret"),
780+
);
781+
let message = ping_message();
782+
let expected_body = serde_json::to_value(&message)?;
783+
784+
let client = StreamableHttpClientTransport::<reqwest::Client>::default_http_client();
785+
let result = client
786+
.post_message(
787+
Arc::<str>::from(format!("http://{addr}/mcp")),
788+
message,
789+
Some(Arc::from("session-1")),
790+
Some("token".to_owned()),
791+
custom_headers,
792+
)
793+
.await;
794+
server.abort();
795+
796+
assert!(
797+
matches!(result, Ok(StreamableHttpPostResponse::Json(..))),
798+
"same-origin {status} should be followed, got {result:?}"
799+
);
800+
let captured = captured.lock().await;
801+
assert_eq!(captured.len(), 1, "target should be hit exactly once");
802+
let request = &captured[0];
803+
assert_eq!(request.method, "POST", "{status} must keep the method");
804+
// reqwest can only replay a reusable body on 307/308; the JSON-RPC
805+
// message must arrive intact, not empty.
806+
assert_eq!(
807+
serde_json::from_slice::<serde_json::Value>(&request.body)?,
808+
expected_body,
809+
"{status} must replay the JSON-RPC body"
810+
);
811+
assert_eq!(request.session_id.as_deref(), Some("session-1"));
812+
assert_eq!(request.authorization.as_deref(), Some("Bearer token"));
813+
assert_eq!(request.api_key.as_deref(), Some("secret"));
814+
Ok(())
815+
}
816+
817+
#[tokio::test]
818+
async fn default_http_client_stops_redirect_loop() -> anyhow::Result<()> {
819+
use std::{
820+
collections::HashMap,
821+
net::SocketAddr,
822+
sync::{
823+
Arc,
824+
atomic::{AtomicUsize, Ordering},
825+
},
826+
};
827+
828+
use axum::{Router, routing::post};
829+
use http::header::LOCATION;
830+
831+
use super::{MAX_REDIRECTS, StreamableHttpClientTransport};
832+
use crate::transport::streamable_http_client::{StreamableHttpClient, StreamableHttpError};
833+
834+
let listener = tokio::net::TcpListener::bind(SocketAddr::from(([127, 0, 0, 1], 0))).await?;
835+
let addr = listener.local_addr()?;
836+
let hits = Arc::new(AtomicUsize::new(0));
837+
let app = Router::new().route(
838+
"/mcp",
839+
post({
840+
let hits = hits.clone();
841+
move || {
842+
hits.fetch_add(1, Ordering::SeqCst);
843+
async {
844+
(
845+
http::StatusCode::TEMPORARY_REDIRECT,
846+
[(LOCATION, "/mcp")],
847+
"",
848+
)
849+
}
850+
}
851+
}),
852+
);
853+
let server = tokio::spawn(async move { axum::serve(listener, app).await });
854+
855+
let client = StreamableHttpClientTransport::<reqwest::Client>::default_http_client();
856+
let result = client
857+
.post_message(
858+
Arc::<str>::from(format!("http://{addr}/mcp")),
859+
ping_message(),
860+
None,
861+
None,
862+
HashMap::new(),
863+
)
864+
.await;
865+
server.abort();
866+
867+
assert!(
868+
matches!(&result, Err(StreamableHttpError::Client(error)) if error.is_redirect()),
869+
"redirect loop should end in a redirect error, got {result:?}"
870+
);
871+
assert_eq!(hits.load(Ordering::SeqCst), MAX_REDIRECTS + 1);
872+
Ok(())
873+
}
874+
875+
#[rstest]
876+
#[case::moved_permanently(http::StatusCode::MOVED_PERMANENTLY)]
877+
#[case::found(http::StatusCode::FOUND)]
878+
#[case::see_other(http::StatusCode::SEE_OTHER)]
879+
#[tokio::test]
880+
async fn default_http_client_does_not_follow_method_rewriting_redirect(
881+
#[case] status: http::StatusCode,
882+
) -> anyhow::Result<()> {
883+
use std::{collections::HashMap, sync::Arc};
884+
885+
use super::StreamableHttpClientTransport;
886+
use crate::transport::streamable_http_client::{StreamableHttpClient, StreamableHttpError};
887+
888+
let (addr, captured, server) = spawn_same_origin_redirect_server(status).await?;
889+
890+
let client = StreamableHttpClientTransport::<reqwest::Client>::default_http_client();
891+
let result = client
892+
.post_message(
893+
Arc::<str>::from(format!("http://{addr}/mcp")),
894+
ping_message(),
895+
None,
896+
None,
897+
HashMap::new(),
898+
)
899+
.await;
900+
server.abort();
901+
902+
// Following would turn the POST into a bodyless GET, which a
903+
// Streamable HTTP server may accept as an SSE stream request, so the
904+
// message would be lost silently.
905+
assert!(
906+
matches!(
907+
&result,
908+
Err(StreamableHttpError::UnexpectedServerResponse(message))
909+
if message.contains(status.as_str())
910+
),
911+
"{status} should surface as an error, got {result:?}"
912+
);
913+
assert!(
914+
captured.lock().await.is_empty(),
915+
"{status} must not be followed"
916+
);
917+
Ok(())
918+
}
643919
}

0 commit comments

Comments
 (0)