feat(mcp): loopback bind + bearer-token gate for Streamable HTTP (T10.4, FR-MCP-1/6)

bind_loopback() refuses to bind anything that doesn't resolve to a
loopback address -- unit tested directly (non-loopback IP, unparseable
host, and a real 127.0.0.1:0 bind). serve() runs a hyper HTTP/1 accept
loop in front of rmcp's StreamableHttpService (a bare tower_service, not
an axum app); every request needs `Authorization: Bearer <token>`
(constant-time compared via mcp::token::verify) before it ever reaches
the MCP service -- an unauthenticated request never reaches rmcp at all.
This commit is contained in:
iamdoubz committed 2026-07-07 00:20:14 -05:00
1 parent c48c67760d
commit 9eed88e5bd
1 file changed
+164
+164
View File
@@ -0,0 +1,164 @@
//! Streamable HTTP transport (FR-MCP-6): loopback-only bind + a bearer-token
//! gate that runs in front of every connection, before a single byte reaches
//! the MCP service. `rmcp`'s `StreamableHttpService` is a bare
//! `tower_service::Service` (not an axum app), so this module supplies the
//! actual TCP accept loop + HTTP/1 framing via `hyper`.
use crate::mcp::handler::WaMcpHandler;
use crate::mcp::{token, McpError};
use bytes::Bytes;
use http_body_util::{combinators::BoxBody, BodyExt, Full};
use hyper::body::Incoming;
use hyper::service::service_fn;
use hyper::{Request, Response, StatusCode};
use hyper_util::rt::TokioIo;
use rmcp::transport::streamable_http_server::session::local::LocalSessionManager;
use rmcp::transport::streamable_http_server::{StreamableHttpServerConfig, StreamableHttpService};
use std::convert::Infallible;
use std::net::{IpAddr, SocketAddr};
use std::sync::Arc;
use tokio::net::TcpListener;
use tokio_util::sync::CancellationToken;
/// Binds `host:port`, refusing anything that doesn't resolve to a loopback
/// address (127.0.0.0/8 or ::1) -- the only way this crate ever opens a
/// listening socket for MCP (FR-MCP-1, NFR-SEC-5). Kept generic over `host`
/// purely so the refusal path is directly unit-testable; the only production
/// caller (`mcp::server`) always passes `"127.0.0.1"`.
pub(crate) async fn bind_loopback(host: &str, port: u16) -> Result<TcpListener, McpError> {
let ip: IpAddr = host.parse().map_err(|_| McpError::NonLoopback)?;
if !ip.is_loopback() {
return Err(McpError::NonLoopback);
}
TcpListener::bind(SocketAddr::new(ip, port))
.await
.map_err(|e| McpError::Server(e.to_string()))
}
/// A running HTTP server; `stop()` cancels the accept loop and all live
/// connections and waits for cleanup.
pub(crate) struct HttpServerHandle {
pub local_addr: SocketAddr,
shutdown: CancellationToken,
join: tokio::task::JoinHandle<()>,
}
impl HttpServerHandle {
pub async fn stop(self) {
self.shutdown.cancel();
let _ = self.join.await;
}
}
/// Serves the MCP Streamable HTTP endpoint (`/mcp`, per the config's session
/// routing) on an already-bound loopback listener. Every request must present
/// `Authorization: Bearer <token>` matching the stored token (constant-time
/// compare, `mcp::token::verify`) or it never reaches `rmcp`.
pub(crate) fn serve(listener: TcpListener, expected_token: String, handler: WaMcpHandler) -> HttpServerHandle {
let local_addr = listener
.local_addr()
.expect("a just-bound TcpListener has a local addr");
let shutdown = CancellationToken::new();
let config = StreamableHttpServerConfig {
stateful_mode: true,
..Default::default()
};
let session_manager = Arc::new(LocalSessionManager::default());
let service = StreamableHttpService::new(
move || Ok(handler.clone()),
session_manager,
config,
);
let accept_ct = shutdown.clone();
let join = tokio::spawn(async move {
loop {
tokio::select! {
_ = accept_ct.cancelled() => break,
accepted = listener.accept() => {
let Ok((stream, _peer)) = accepted else { continue };
let io = TokioIo::new(stream);
let svc = service.clone();
let token = expected_token.clone();
let conn_ct = accept_ct.clone();
tokio::spawn(async move {
let guarded = service_fn(move |req: Request<Incoming>| {
let mut svc = svc.clone();
let token = token.clone();
async move { Ok::<_, Infallible>(handle_request(req, &mut svc, &token).await) }
});
let conn = hyper::server::conn::http1::Builder::new().serve_connection(io, guarded);
tokio::select! {
_ = conn_ct.cancelled() => {}
_ = conn => {}
}
});
}
}
}
});
HttpServerHandle { local_addr, shutdown, join }
}
async fn handle_request(
req: Request<Incoming>,
svc: &mut StreamableHttpService<WaMcpHandler>,
expected_token: &str,
) -> Response<BoxBody<Bytes, Infallible>> {
if !is_authorized(&req, expected_token) {
return unauthorized_response();
}
let (parts, body) = req.into_parts();
let req = Request::from_parts(parts, body.boxed());
tower_service::Service::call(svc, req)
.await
.unwrap_or_else(|never: Infallible| match never {})
}
fn is_authorized(req: &Request<Incoming>, expected_token: &str) -> bool {
let Some(header) = req.headers().get(hyper::header::AUTHORIZATION) else {
return false;
};
let Ok(header) = header.to_str() else {
return false;
};
let Some(presented) = header.strip_prefix("Bearer ") else {
return false;
};
token::verify(presented, expected_token)
}
fn unauthorized_response() -> Response<BoxBody<Bytes, Infallible>> {
Response::builder()
.status(StatusCode::UNAUTHORIZED)
.header(hyper::header::CONTENT_TYPE, "application/json")
.body(Full::new(Bytes::from_static(b"{\"error\":\"unauthorized\"}")).boxed())
.expect("valid response")
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn refuses_a_non_loopback_bind() {
// A real routable address is never allowed regardless of port
// availability -- the check happens before any socket syscall.
let err = bind_loopback("8.8.8.8", 0).await.unwrap_err();
assert!(matches!(err, McpError::NonLoopback));
}
#[tokio::test]
async fn refuses_an_unparseable_host() {
let err = bind_loopback("not-an-ip", 0).await.unwrap_err();
assert!(matches!(err, McpError::NonLoopback));
}
#[tokio::test]
async fn binds_127_0_0_1_on_an_os_assigned_port() {
let listener = bind_loopback("127.0.0.1", 0).await.expect("loopback bind");
assert!(listener.local_addr().unwrap().ip().is_loopback());
}
}