use std::io::Write;
use std::net::{Shutdown, TcpStream};
use std::num::NonZeroU64;
use std::sync::mpsc;
use std::time::Duration;
use super::fixtures::{LIST, Reply, Server, TestResult, config, refusal, request};
use super::listener;
use super::native_call_fixture::Store;
fn start(
store: &Store,
) -> Result<(Server, mpsc::Receiver<String>), Box<dyn std::error::Error + Send + Sync>> {
let (logged, lines) = mpsc::channel();
let server = listener::Server::new(
config()?,
store.path.clone(),
Duration::from_secs(1),
131_072,
1,
)?
.with_diagnostics(logged)?;
Ok((Server::from_mcp(server)?, lines))
}
#[test]
fn an_unauthenticated_large_body_receives_the_whole_challenge_and_refusal() -> TestResult {
let store = Store::start()?;
let mut config = config()?;
config.resource_metadata_url = Some("https://identity.example/resource-metadata".to_owned());
let mcp = Server::mcp(config, store.path.clone(), 131_072, 1)?;
let response = Reply::read(request(&mcp, "POST", &"x".repeat(65_536), false, &[])?)?;
refusal(&response, 401, "mcp_sign_in_required")?;
assert_eq!(
response
.headers
.get("www-authenticate")
.ok_or("challenge missing")?
.to_str()?,
"Bearer resource_metadata=\"https://identity.example/resource-metadata\""
);
assert!(store.asked()?.is_empty());
mcp.finish()?;
store.finish()
}
fn raw_refusal(bytes: &str, status: u16, code: &str, has_cause: bool) -> TestResult {
let store = Store::start()?;
let (mcp, lines) = start(&store)?;
let mut socket = TcpStream::connect(mcp.address)?;
socket.write_all(bytes.as_bytes())?;
let response = Reply::read(socket)?;
refusal(&response, status, code)?;
let line = lines.try_recv()?;
let body: serde_json::Value = serde_json::from_slice(&response.body)?;
let message = body["message"].as_str().ok_or("reason missing")?;
assert!(line.contains(code));
assert_eq!(line.matches(message).count(), 1);
assert_eq!(line.contains("; cause="), has_cause);
assert_eq!(line.lines().count(), 1);
assert!(store.asked()?.is_empty());
mcp.finish()?;
store.finish()
}
#[test]
fn headers_over_the_byte_bound_refuse_431_before_sign_in() -> TestResult {
raw_refusal(
&format!(
concat!(
"POST /mcp HTTP/1.1\r\nHost: fixture\r\nPadding: {}\r\n",
"Content-Length: 0\r\n\r\n"
),
"x".repeat(16_384)
),
431,
"mcp_http_headers_too_large",
false,
)
}
#[test]
fn expect_negotiation_refuses_417_before_sign_in() -> TestResult {
raw_refusal(
concat!(
"POST /mcp HTTP/1.1\r\nHost: fixture\r\n",
"Expect: 100-continue\r\nContent-Length: 0\r\n\r\n"
),
417,
"mcp_http_expect_unsupported",
false,
)
}
#[test]
fn an_unknown_route_refuses_404_before_sign_in() -> TestResult {
raw_refusal(
"POST /other HTTP/1.1\r\nHost: fixture\r\nContent-Length: 0\r\n\r\n",
404,
"mcp_http_route_missing",
false,
)
}
#[test]
fn an_incomplete_header_eof_is_named_and_logged_once() -> TestResult {
let store = Store::start()?;
let (mcp, lines) = start(&store)?;
let mut socket = TcpStream::connect(mcp.address)?;
socket.write_all(b"POST /mcp HTTP/1.1\r\nHost: fixture")?;
socket.shutdown(Shutdown::Write)?;
refusal(&Reply::read(socket)?, 400, "mcp_http_header_unreadable")?;
let lines = lines.try_iter().collect::<Vec<_>>();
assert_eq!(lines.len(), 1);
assert!(lines[0].contains("mcp_http_header_unreadable"));
assert!(store.asked()?.is_empty());
mcp.finish()?;
store.finish()
}
#[test]
fn a_partial_body_expires_on_the_configured_body_deadline() -> TestResult {
let store = Store::start()?;
let mut config = config()?;
config.body_timeout_ms = NonZeroU64::new(50).ok_or("body timeout missing")?;
let mcp = Server::mcp(config, store.path.clone(), 131_072, 1)?;
let mut socket = TcpStream::connect(mcp.address)?;
write!(
socket,
"POST /mcp HTTP/1.1\r\nHost: fixture\r\nContent-Length: {}\r\nContent-Type: application/json\r\nAccept: application/json, text/event-stream\r\nAuthorization: Bearer fixture-token\r\n\r\n{{",
LIST.len()
)?;
refusal(&Reply::read(socket)?, 408, "mcp_http_body_timeout")?;
assert!(store.asked()?.is_empty());
mcp.finish()?;
store.finish()
}
#[derive(Clone, Copy)]
enum Behind {
Accepting,
Refusing(&'static str),
Absent,
}
type LogCase = (
Behind,
bool,
Vec<(&'static str, &'static str)>,
u16,
&'static str,
);
fn logged_case((behind, token, extra, status, code): LogCase) -> TestResult {
let refused = match behind {
Behind::Refusing(name) => Some(name),
Behind::Accepting | Behind::Absent => None,
};
let store = match behind {
Behind::Accepting | Behind::Absent => Store::start()?,
Behind::Refusing(code) => Store::refusing(code)?,
};
let (logged, lines) = mpsc::channel();
let directory = tempfile::tempdir()?;
let socket = match behind {
Behind::Absent => directory.path().join("absent.sock"),
Behind::Accepting | Behind::Refusing(_) => store.path.clone(),
};
let server = listener::Server::new(config()?, socket, Duration::from_secs(1), 131_072, 1)?
.with_diagnostics(logged)?;
let mcp = Server::from_mcp(server)?;
let response = Reply::read(request(&mcp, "POST", LIST, token, &extra)?)?;
refusal(&response, status, code)?;
let line = lines.try_recv()?;
let body: serde_json::Value = serde_json::from_slice(&response.body)?;
let message = body["message"].as_str().ok_or("reason missing")?;
assert!(line.contains(code));
assert_eq!(line.matches(message).count(), 1);
match refused {
Some(name) => assert!(line.ends_with(&format!("; cause=store:{name}")), "{line}"),
None => assert!(!line.contains("; cause="), "{line}"),
}
assert!(!line.contains("fixture-token"));
assert_eq!(line.lines().count(), 1);
mcp.finish()?;
store.finish()
}
#[test]
fn sign_in_protocol_and_store_refusals_have_safe_named_log_lines() -> TestResult {
for case in [
(
Behind::Accepting,
false,
Vec::new(),
401,
"mcp_sign_in_required",
),
(
Behind::Accepting,
true,
vec![("mcp-protocol-version", "invalid")],
400,
"mcp_http_protocol_version_unsupported",
),
(
Behind::Refusing("pass_expired"),
true,
Vec::new(),
401,
"mcp_pass_refused",
),
(
Behind::Refusing("lys_could_not_be_asked"),
true,
Vec::new(),
503,
"mcp_store_unavailable",
),
(
Behind::Absent,
true,
Vec::new(),
503,
"mcp_store_unavailable",
),
] {
logged_case(case)?;
}
Ok(())
}
#[test]
fn a_configured_path_is_the_only_one_answered() -> TestResult {
let store = Store::start()?;
let mut config = config()?;
config.path = "/mcp/studio".to_owned();
config.validate()?;
let mcp = Server::mcp(config, store.path.clone(), 131_072, 1)?;
for (path, status) in [("/mcp", 404), ("/mcp/studio/", 404), ("/mcp/studio", 200)] {
let mut socket = TcpStream::connect(mcp.address)?;
socket.write_all(
format!(
"POST {path} HTTP/1.1\r\nHost: fixture\r\nContent-Type: application/json\r\nAccept: application/json, text/event-stream\r\nAuthorization: Bearer fixture-token\r\nContent-Length: {}\r\n\r\n{LIST}",
LIST.len()
)
.as_bytes(),
)?;
let response = Reply::read(socket)?;
assert_eq!(response.status, status, "{path}");
if status == 404 {
refusal(&response, 404, "mcp_http_route_missing")?;
assert!(store.asked()?.is_empty(), "{path} reached the store");
}
}
assert!(
!store.asked()?.is_empty(),
"the configured path never reached the store"
);
mcp.finish()?;
store.finish()
}