use std::io::{self, BufRead, BufReader, Read, Write};
use std::net::{Shutdown, TcpStream};
use std::time::{Duration, Instant};
use ureq::http::{HeaderMap, HeaderName, HeaderValue, Method, Uri};
use super::{error::Error, headers};
const HEADER_BYTES: usize = 16_384;
const LINGER_TIMEOUT: Duration = Duration::from_secs(1);
#[cfg(test)]
#[path = "wire_failure_tests.rs"]
mod failure_tests;
pub(super) struct Request {
pub(super) method: Method,
pub(super) path: String,
pub(super) headers: HeaderMap,
pub(super) length: usize,
}
pub(super) struct Response {
pub(super) status: u16,
pub(super) body: Vec<u8>,
pub(super) headers: Vec<(&'static str, String)>,
}
impl Response {
pub(super) fn refused(error: &Error, metadata: Option<&str>) -> Result<Self, Error> {
let body = serde_json::to_vec(error).map_err(|_| {
Error::new(
500,
"mcp_http_encoding_failed",
"The refusal could not be encoded",
)
})?;
let mut headers = Vec::new();
if error.status == 401 {
let challenge = metadata.map_or_else(
|| "Bearer".to_owned(),
|url| format!("Bearer resource_metadata=\"{url}\""),
);
headers.push(("WWW-Authenticate", challenge));
}
if error.status == 405 {
headers.push(("Allow", "POST".to_owned()));
}
Ok(Self {
status: error.status,
body,
headers,
})
}
pub(super) fn write(&self, socket: &mut TcpStream, timeout: Duration) -> Result<(), Error> {
let deadline = deadline(timeout)?;
let mut head = format!(
"HTTP/1.1 {} \r\nContent-Length: {}\r\nContent-Type: application/json\r\nCache-Control: no-store\r\nConnection: close\r\n",
self.status,
self.body.len(),
);
for (name, value) in &self.headers {
head.push_str(name);
head.push_str(": ");
head.push_str(value);
head.push_str("\r\n");
}
head.push_str("\r\n");
write(socket, head.as_bytes(), deadline)?;
write(socket, &self.body, deadline)
}
}
fn deadline(timeout: Duration) -> Result<Instant, Error> {
Instant::now()
.checked_add(timeout)
.ok_or_else(|| Error::config("HTTP deadline is not representable"))
}
fn remaining(deadline: Instant, code: &'static str) -> Result<Duration, Error> {
deadline
.checked_duration_since(Instant::now())
.filter(|value| !value.is_zero())
.ok_or_else(|| Error::new(408, code, "The configured HTTP deadline was reached"))
}
fn read_error(error: &io::Error, timeout: &'static str, code: &'static str) -> Error {
if matches!(
error.kind(),
io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock
) {
Error::new(408, timeout, "The configured HTTP deadline was reached")
} else {
Error::new(400, code, "The HTTP request could not be read")
}
.with_io(error)
}
pub(super) fn head(
socket: &mut BufReader<TcpStream>,
timeout: Duration,
cap: usize,
) -> Result<Request, Error> {
let deadline = deadline(timeout)?;
let mut head = Vec::new();
loop {
socket
.get_ref()
.set_read_timeout(Some(remaining(deadline, "mcp_http_header_timeout")?))
.map_err(|error| {
Error::new(
500,
"mcp_http_socket_failed",
"The socket deadline could not be set",
)
.with_io(&error)
})?;
let available = match socket.fill_buf() {
Ok(available) => available,
Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
Err(error) => {
return Err(read_error(
&error,
"mcp_http_header_timeout",
"mcp_http_header_unreadable",
));
}
};
if available.is_empty() {
return Err(Error::new(
400,
"mcp_http_header_unreadable",
"The caller closed before its headers completed",
));
}
let before = head.len();
let count = available.len().min(HEADER_BYTES - before);
head.extend_from_slice(&available[..count]);
let from = before.saturating_sub(3);
if let Some(start) = head[from..]
.windows(4)
.position(|bytes| bytes == b"\r\n\r\n")
{
let end = from + start + 4;
socket.consume(end - before);
head.truncate(end);
return parse(&head, cap);
}
socket.consume(count);
if head.len() == HEADER_BYTES {
return Err(Error::new(
431,
"mcp_http_headers_too_large",
"The request headers exceed their byte limit",
));
}
}
}
pub(super) fn parse(bytes: &[u8], cap: usize) -> Result<Request, Error> {
let bad = || {
Error::new(
400,
"mcp_http_headers_invalid",
"The request requires valid HTTP/1.1 headers",
)
};
if !bytes.ends_with(b"\r\n\r\n") {
return Err(bad());
}
let text = std::str::from_utf8(bytes).map_err(|_| bad())?;
let mut lines = text.split("\r\n");
let mut first = lines.next().ok_or_else(bad)?.split_ascii_whitespace();
let method = Method::from_bytes(first.next().ok_or_else(bad)?.as_bytes()).map_err(|_| bad())?;
let target = first.next().ok_or_else(bad)?;
if !target.starts_with('/')
|| target.contains('#')
|| first.next() != Some("HTTP/1.1")
|| first.next().is_some()
{
return Err(bad());
}
let uri: Uri = target.parse().map_err(|_| bad())?;
let mut headers = HeaderMap::new();
for line in lines {
if line.is_empty() {
break;
}
let (name, value) = line.split_once(':').ok_or_else(bad)?;
let name = HeaderName::from_bytes(name.as_bytes()).map_err(|_| bad())?;
let value = HeaderValue::from_str(value.trim_matches([' ', '\t'])).map_err(|_| bad())?;
headers.append(name, value);
}
if headers::one(
&headers,
"host",
400,
"mcp_http_headers_invalid",
"Exactly one Host header is required",
)?
.is_none_or(str::is_empty)
{
return Err(bad());
}
if headers.contains_key("transfer-encoding") {
return Err(Error::new(
400,
"mcp_http_transfer_encoding_refused",
"Only a bounded Content-Length body is served",
));
}
if headers.contains_key("expect") {
return Err(Error::new(
417,
"mcp_http_expect_unsupported",
"Expect negotiation is not served",
));
}
let length = headers::one(
&headers,
"content-length",
400,
"mcp_http_length_invalid",
"Exactly one numeric Content-Length is required",
)?;
let length = match length {
Some(value) if !value.is_empty() && value.bytes().all(|byte| byte.is_ascii_digit()) => {
value.parse::<usize>().map_err(|_| bad())?
}
Some(_) => return Err(bad()),
None if method == Method::POST => {
return Err(Error::new(
411,
"mcp_http_length_required",
"POST requires Content-Length",
));
}
None => 0,
};
if length > cap {
return Err(Error::new(
413,
"mcp_http_body_refused",
"The request body exceeds its byte limit",
));
}
Ok(Request {
method,
path: uri.path().to_owned(),
headers,
length,
})
}
pub(super) fn body(
socket: &mut BufReader<TcpStream>,
length: usize,
timeout: Duration,
) -> Result<Vec<u8>, Error> {
let deadline = deadline(timeout)?;
let mut body = vec![0; length];
let mut read = 0;
while read < length {
socket
.get_ref()
.set_read_timeout(Some(remaining(deadline, "mcp_http_body_timeout")?))
.map_err(|error| {
Error::new(
500,
"mcp_http_socket_failed",
"The socket deadline could not be set",
)
.with_io(&error)
})?;
let count = match socket.read(&mut body[read..]) {
Ok(0) => {
return Err(Error::new(
400,
"mcp_http_body_refused",
"The caller closed before its body completed",
));
}
Ok(count) => count,
Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
Err(error) => {
return Err(read_error(
&error,
"mcp_http_body_timeout",
"mcp_http_body_refused",
));
}
};
read += count;
}
Ok(body)
}
fn write(socket: &mut TcpStream, mut bytes: &[u8], deadline: Instant) -> Result<(), Error> {
while !bytes.is_empty() {
socket
.set_write_timeout(Some(remaining(deadline, "mcp_http_write_timeout")?))
.map_err(|error| {
Error::new(
500,
"mcp_http_socket_failed",
"The socket deadline could not be set",
)
.with_io(&error)
})?;
let count = match socket.write(bytes) {
Ok(0) => {
return Err(Error::new(
500,
"mcp_http_write_failed",
"The HTTP answer could not be written",
));
}
Ok(count) => count,
Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
Err(error) => return Err(write_error(&error)),
};
bytes = &bytes[count..];
}
Ok(())
}
fn write_error(error: &io::Error) -> Error {
if matches!(
error.kind(),
io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock
) {
Error::new(
408,
"mcp_http_write_timeout",
"The configured HTTP write deadline was reached",
)
} else {
Error::new(
500,
"mcp_http_write_failed",
"The HTTP answer could not be written",
)
}
.with_io(error)
}
pub(super) fn linger(socket: &mut BufReader<TcpStream>) -> Result<(), Error> {
let deadline = deadline(LINGER_TIMEOUT)?;
set_linger_timeout(socket.get_ref(), deadline, false)?;
close_write(socket.get_ref())?;
let buffered = socket.buffer().len();
socket.consume(buffered);
let mut bytes = [0; 4096];
loop {
set_linger_timeout(socket.get_ref(), deadline, true)?;
match socket.get_mut().read(&mut bytes) {
Ok(0) => return Ok(()),
Ok(_) => {}
Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
Err(error) => {
return Err(read_error(
&error,
"mcp_http_linger_timeout",
"mcp_http_linger_unreadable",
));
}
}
}
}
fn set_linger_timeout(
socket: &TcpStream,
deadline: Instant,
closed_write: bool,
) -> Result<(), Error> {
let timeout = remaining(deadline, "mcp_http_linger_timeout")?;
match socket.set_read_timeout(Some(timeout)) {
Ok(()) => Ok(()),
Err(error)
if closed_write && cfg!(target_os = "macos") && error.raw_os_error() == Some(22) =>
{
Ok(())
}
Err(error) => Err(Error::new(
500,
"mcp_http_socket_failed",
"The lingering close deadline could not be set",
)
.with_io(&error)),
}
}
pub(super) fn close_write(socket: &TcpStream) -> Result<(), Error> {
socket.shutdown(Shutdown::Write).map_err(|error| {
Error::new(
500,
"mcp_http_close_failed",
"The HTTP answer could not be half-closed",
)
.with_io(&error)
})
}