use alloc::vec::Vec;
use crate::{
body::{Decoder, Encoder, Step, push_framing},
buffer::Buffer,
error::HttpError,
head::{HeadKind, HeadScanner, Progress, RequestHead, RequestInfo, Version, interpret_request},
limits::Limits,
message::{OutBody, Response},
method::Method,
status::reason_phrase,
syntax::push_decimal,
};
#[non_exhaustive]
#[derive(Debug)]
pub enum ServerEvent<'a> {
Request(RequestHead<'a>),
Body(&'a [u8]),
RequestEnd,
NeedInput,
AwaitingResponse,
Upgraded,
Closed,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum ReqState {
Head,
Body,
Complete,
Upgraded,
Failed,
Closed,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum RespState {
Idle,
Streaming(Encoder),
Sent,
}
pub struct ServerConn {
buf: Buffer,
limits: Limits,
scanner: HeadScanner,
trailers: HeadScanner,
decoder: Decoder,
info: RequestInfo,
state: ReqState,
resp: RespState,
keep_alive: bool,
continue_sent: bool,
}
impl Default for ServerConn {
fn default() -> Self {
Self::new()
}
}
impl core::fmt::Debug for ServerConn {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("ServerConn")
.field("state", &self.state)
.field("response", &self.resp)
.field("buffered", &self.buf.unread().len())
.field("limits", &self.limits)
.finish()
}
}
impl ServerConn {
#[must_use]
pub fn new() -> Self {
Self::with_limits(Limits::new())
}
#[must_use]
pub fn with_limits(limits: Limits) -> Self {
let mut buf = Buffer::new(limits.buffer_capacity());
buf.pin_at_pos();
Self {
buf,
limits,
scanner: HeadScanner::new(),
trailers: HeadScanner::new(),
decoder: Decoder::new(),
info: RequestInfo::EMPTY,
state: ReqState::Head,
resp: RespState::Idle,
keep_alive: false,
continue_sent: false,
}
}
#[must_use]
pub fn limits(&self) -> &Limits {
&self.limits
}
#[must_use = "bytes beyond the returned count were not taken"]
pub fn feed(&mut self, bytes: &[u8]) -> usize {
self.buf.feed(bytes)
}
pub fn read_buf(&mut self) -> &mut [u8] {
self.buf.spare()
}
pub fn commit(&mut self, n: usize) {
self.buf.commit(n);
}
pub fn feed_eof(&mut self) {
self.buf.set_eof();
}
pub fn poll(&mut self) -> Result<ServerEvent<'_>, HttpError> {
loop {
match self.state {
ReqState::Head => return self.poll_head(),
ReqState::Body => {
if self.resp == RespState::Sent && !self.decoder.is_done() {
self.state = ReqState::Closed;
return Ok(ServerEvent::Closed);
}
return self.poll_body();
}
ReqState::Complete => {
if self.resp != RespState::Sent {
return Ok(ServerEvent::AwaitingResponse);
}
if !self.keep_alive {
self.state = ReqState::Closed;
return Ok(ServerEvent::Closed);
}
self.next_exchange();
}
ReqState::Upgraded => return Ok(ServerEvent::Upgraded),
ReqState::Failed | ReqState::Closed => return Ok(ServerEvent::Closed),
}
}
}
fn poll_head(&mut self) -> Result<ServerEvent<'_>, HttpError> {
let start = self.buf.pin();
let msg = self.buf.from(start);
let progress = self
.scanner
.advance(msg, &self.limits, HeadKind::Request, true)
.map_err(|e| self.fail(e))?;
let len = match progress {
Progress::Partial if self.buf.is_eof() => {
if self.buf.from(start).is_empty() {
self.state = ReqState::Closed;
return Ok(ServerEvent::Closed);
}
return Err(self.fail(HttpError::UnexpectedEof));
}
Progress::Partial => return Ok(ServerEvent::NeedInput),
Progress::Done(len) => len,
};
let info = interpret_request(
self.buf.from(start),
self.scanner.start_line(),
&self.scanner.fields,
&self.limits,
)
.map_err(|e| self.fail(e))?;
self.buf.consume(len);
self.buf.extend_pin_to_pos();
self.info = info;
self.keep_alive = info.keep_alive;
self.decoder.start(info.framing);
self.state = ReqState::Body;
Ok(ServerEvent::Request(self.view()))
}
fn poll_body(&mut self) -> Result<ServerEvent<'_>, HttpError> {
let mut framing_bytes = 0;
let step = self.decoder.step(
self.buf.unread(),
self.buf.is_eof(),
&self.limits,
&mut self.trailers,
&mut framing_bytes,
);
self.buf.consume(framing_bytes);
match step.map_err(|e| self.fail(e))? {
Step::Data(n) => {
let start = self.buf.pos();
self.buf.consume(n);
Ok(ServerEvent::Body(self.buf.range(start, start + n)))
}
Step::Done => {
self.state = ReqState::Complete;
Ok(ServerEvent::RequestEnd)
}
Step::NeedMore => Ok(ServerEvent::NeedInput),
}
}
fn fail(&mut self, e: HttpError) -> HttpError {
self.state = ReqState::Failed;
self.keep_alive = false;
e
}
fn next_exchange(&mut self) {
self.buf.unpin();
self.buf.pin_at_pos();
self.scanner.reset();
self.resp = RespState::Idle;
self.info = RequestInfo::EMPTY;
self.continue_sent = false;
self.state = ReqState::Head;
}
#[cfg(feature = "std")]
pub(crate) fn response_started(&self) -> bool {
self.resp != RespState::Idle
}
fn body_read(&self) -> bool {
self.state == ReqState::Complete || (self.state == ReqState::Body && self.decoder.is_done())
}
fn view(&self) -> RequestHead<'_> {
RequestHead::new(
self.buf.from(self.buf.pin()),
&self.scanner.fields,
&self.info,
)
}
#[must_use]
pub fn request(&self) -> Option<RequestHead<'_>> {
match self.state {
ReqState::Body | ReqState::Complete | ReqState::Upgraded => Some(self.view()),
_ => None,
}
}
#[must_use]
pub fn is_upgraded(&self) -> bool {
self.state == ReqState::Upgraded
}
#[must_use]
pub fn remaining_input(&self) -> &[u8] {
self.buf.unread()
}
pub fn respond(&mut self, response: &Response, out: &mut Vec<u8>) -> Result<(), HttpError> {
self.respond_dated(response, None, out)
}
pub(crate) fn respond_dated(
&mut self,
response: &Response,
date: Option<&[u8]>,
out: &mut Vec<u8>,
) -> Result<(), HttpError> {
if let Some(e) = response.error() {
return Err(e);
}
let failed = self.state == ReqState::Failed;
match self.state {
ReqState::Body | ReqState::Complete | ReqState::Failed => {}
ReqState::Head => return Err(HttpError::InvalidState("no request to respond to")),
ReqState::Upgraded | ReqState::Closed => {
return Err(HttpError::InvalidState(
"the connection is closed or upgraded",
));
}
}
if self.resp != RespState::Idle {
return Err(HttpError::InvalidState("a response was already started"));
}
let status = response.status();
if !(100..=599).contains(&status) || ((100..200).contains(&status) && status != 101) {
return Err(HttpError::InvalidStatus);
}
let (method, version) = if failed {
(Method::Get, Version::Http11)
} else {
(self.info.method, self.info.version)
};
let bodiless = (100..200).contains(&status) || status == 204 || status == 304;
let tunnel = status == 101 || (method == Method::Connect && (200..300).contains(&status));
if bodiless || tunnel {
let empty = matches!(&response.body, OutBody::Full(b) if b.is_empty());
if !empty {
return Err(HttpError::BodyNotAllowed);
}
}
if tunnel {
let complete_upgrade =
self.body_read() && (self.info.upgrade || method == Method::Connect);
if !complete_upgrade || (status == 101 && !response.upgrade) || failed {
return Err(HttpError::InvalidState(
"101 (or 2xx to CONNECT) must answer a complete upgrade request; \
build it with Response::switching_protocols",
));
}
if status == 101
&& !response
.upgrade_protocol()
.is_some_and(|p| self.view().offers(p))
{
return Err(HttpError::UpgradeNotOffered);
}
}
let mut close = failed || response.close || !self.keep_alive || !self.body_read();
let head_only = method == Method::Head || bodiless || tunnel;
let (mut encoder, declared) = match &response.body {
OutBody::Full(b) => (
Encoder::Length {
remaining: b.len() as u64,
},
Some(b.len() as u64),
),
OutBody::Length(n) => (Encoder::Length { remaining: *n }, Some(*n)),
OutBody::Chunked if version == Version::Http10 => {
close = true;
(Encoder::CloseDelimited, None)
}
OutBody::Chunked => (Encoder::Chunked, None),
};
out.extend_from_slice(b"HTTP/1.1 ");
push_decimal(out, u64::from(status));
out.push(b' ');
out.extend_from_slice(reason_phrase(status).as_bytes());
out.extend_from_slice(b"\r\n");
if !(bodiless || tunnel) {
push_framing(out, encoder, declared);
}
if tunnel {
close = false;
} else if close {
out.extend_from_slice(b"connection: close\r\n");
} else if version == Version::Http10 {
out.extend_from_slice(b"connection: keep-alive\r\n");
}
if let Some(date) = date.filter(|_| !response.has_date) {
out.extend_from_slice(b"date: ");
out.extend_from_slice(date);
out.extend_from_slice(b"\r\n");
}
out.extend_from_slice(response.fields());
out.extend_from_slice(b"\r\n");
if head_only {
encoder = Encoder::Discard;
}
self.keep_alive = !close;
match &response.body {
OutBody::Full(b) => {
encoder.write(b, out)?;
self.resp = RespState::Sent;
}
_ => self.resp = RespState::Streaming(encoder),
}
if tunnel {
self.state = ReqState::Upgraded;
} else if failed {
self.state = ReqState::Closed;
}
Ok(())
}
pub fn send_continue(&mut self, out: &mut Vec<u8>) -> Result<(), HttpError> {
if self.state != ReqState::Body
|| self.decoder.is_done()
|| self.resp != RespState::Idle
|| self.info.version != Version::Http11
|| self.continue_sent
{
return Err(HttpError::InvalidState(
"100 Continue answers an HTTP/1.1 request whose body has not been read, \
once, before the final response",
));
}
self.continue_sent = true;
out.extend_from_slice(b"HTTP/1.1 100 Continue\r\n\r\n");
Ok(())
}
pub fn send_body(&mut self, data: &[u8], out: &mut Vec<u8>) -> Result<(), HttpError> {
match &mut self.resp {
RespState::Streaming(encoder) => encoder.write(data, out),
_ => Err(HttpError::InvalidState("no streamed response in progress")),
}
}
pub fn end_body(&mut self, out: &mut Vec<u8>) -> Result<(), HttpError> {
match &mut self.resp {
RespState::Streaming(encoder) => {
encoder.finish(out)?;
if *encoder == Encoder::CloseDelimited {
self.keep_alive = false;
}
self.resp = RespState::Sent;
Ok(())
}
_ => Err(HttpError::InvalidState("no streamed response in progress")),
}
}
}
#[cfg(test)]
mod tests {
use alloc::vec;
use super::*;
fn events(conn: &mut ServerConn) -> Vec<alloc::string::String> {
let mut log = Vec::new();
loop {
match conn.poll() {
Ok(ServerEvent::Request(r)) => log.push(alloc::format!("req {}", r.target())),
Ok(ServerEvent::Body(b)) => log.push(alloc::format!("body {}", b.len())),
Ok(ServerEvent::RequestEnd) => {
log.push("end".into());
return log;
}
Ok(other) => {
log.push(alloc::format!("{other:?}"));
return log;
}
Err(e) => {
log.push(alloc::format!("err {e:?}"));
return log;
}
}
}
}
#[test]
fn test_early_response_closes_connection() {
let mut conn = ServerConn::new();
let _ = conn.feed(b"POST / HTTP/1.1\r\nHost: a\r\nContent-Length: 10\r\n\r\nab");
assert!(matches!(conn.poll(), Ok(ServerEvent::Request(_))));
let mut out = Vec::new();
conn.respond(&Response::new(413), &mut out).unwrap();
assert!(out.windows(17).any(|w| w == b"connection: close"));
assert!(matches!(conn.poll(), Ok(ServerEvent::Closed)));
}
#[test]
fn test_respond_after_failure_is_once_and_closes() {
let mut conn = ServerConn::new();
let _ = conn.feed(b"GET / HTTP/1.1\r\n\r\n");
assert_eq!(conn.poll().unwrap_err(), HttpError::MissingHost);
let mut out = Vec::new();
conn.respond(&Response::new(400), &mut out).unwrap();
assert!(out.starts_with(b"HTTP/1.1 400 Bad Request\r\n"));
assert!(conn.respond(&Response::new(400), &mut out).is_err());
assert!(matches!(conn.poll(), Ok(ServerEvent::Closed)));
}
#[test]
fn test_state_errors() {
let mut conn = ServerConn::new();
let mut out = Vec::new();
assert!(matches!(
conn.respond(&Response::new(200), &mut out),
Err(HttpError::InvalidState(_))
));
assert!(matches!(
conn.send_body(b"x", &mut out),
Err(HttpError::InvalidState(_))
));
assert!(matches!(
conn.end_body(&mut out),
Err(HttpError::InvalidState(_))
));
let _ = conn.feed(b"GET / HTTP/1.1\r\nHost: a\r\n\r\n");
assert_eq!(events(&mut conn), vec!["req /", "end"]);
assert_eq!(
conn.respond(&Response::new(99), &mut out),
Err(HttpError::InvalidStatus)
);
assert_eq!(
conn.respond(&Response::new(100), &mut out),
Err(HttpError::InvalidStatus)
);
assert_eq!(
conn.respond(&Response::new(204).body(&b"x"[..]), &mut out),
Err(HttpError::BodyNotAllowed)
);
assert!(matches!(
conn.respond(&Response::switching_protocols("x"), &mut out),
Err(HttpError::InvalidState(_))
));
conn.respond(&Response::new(204), &mut out).unwrap();
assert!(matches!(
conn.respond(&Response::new(200), &mut out),
Err(HttpError::InvalidState(_))
));
}
#[test]
fn test_http10_keep_alive_and_chunked_fallback() {
let mut conn = ServerConn::new();
let _ = conn.feed(b"GET / HTTP/1.0\r\nConnection: keep-alive\r\n\r\n");
assert_eq!(events(&mut conn), vec!["req /", "end"]);
let mut out = Vec::new();
conn.respond(&Response::text(200, "x"), &mut out).unwrap();
assert!(out.windows(22).any(|w| w == b"connection: keep-alive"));
assert!(matches!(conn.poll(), Ok(ServerEvent::NeedInput)));
let _ = conn.feed(b"GET / HTTP/1.0\r\n\r\n");
assert_eq!(events(&mut conn), vec!["req /", "end"]);
let mut out = Vec::new();
conn.respond(&Response::new(200).chunked(), &mut out)
.unwrap();
conn.send_body(b"abc", &mut out).unwrap();
conn.end_body(&mut out).unwrap();
assert!(out.ends_with(b"\r\n\r\nabc"));
assert!(!out.windows(17).any(|w| w == b"transfer-encoding"));
assert!(matches!(conn.poll(), Ok(ServerEvent::Closed)));
}
fn upgrade_request(conn: &mut ServerConn, offer: &str) {
let msg = alloc::format!(
"GET / HTTP/1.1\r\nHost: a\r\nConnection: upgrade\r\nUpgrade: {offer}\r\n\r\n"
);
let _ = conn.feed(msg.as_bytes());
assert_eq!(events(conn), vec!["req /", "end"]);
}
#[test]
fn test_switching_protocols_must_name_an_offered_protocol() {
let mut conn = ServerConn::new();
upgrade_request(&mut conn, "websocket/13, foo/2");
let mut out = Vec::new();
for refused in ["ftp", "foo/3", "websocket/12", "h2c"] {
assert_eq!(
conn.respond(&Response::switching_protocols(refused), &mut out),
Err(HttpError::UpgradeNotOffered),
"{refused}"
);
assert!(out.is_empty());
assert!(!conn.is_upgraded());
}
conn.respond(&Response::switching_protocols("WebSocket"), &mut out)
.unwrap();
assert!(conn.is_upgraded());
assert!(out.windows(18).any(|w| w == b"upgrade: WebSocket"));
}
#[test]
fn test_refused_switch_leaves_a_normal_response_possible() {
let mut conn = ServerConn::new();
upgrade_request(&mut conn, "websocket");
let mut out = Vec::new();
assert_eq!(
conn.respond(&Response::switching_protocols("irc"), &mut out),
Err(HttpError::UpgradeNotOffered)
);
conn.respond(&Response::new(426), &mut out).unwrap();
assert!(out.starts_with(b"HTTP/1.1 426 Upgrade Required\r\n"));
assert!(matches!(conn.poll(), Ok(ServerEvent::NeedInput)));
}
#[test]
fn test_h2c_only_offer_is_served_as_http11() {
let mut conn = ServerConn::new();
let _ = conn.feed(
b"GET / HTTP/1.1\r\nHost: a\r\nConnection: Upgrade, HTTP2-Settings\r\nUpgrade: h2c\r\nHTTP2-Settings: AAMAAABkAAQAoAAA\r\n\r\n",
);
assert!(matches!(conn.poll(), Ok(ServerEvent::Request(r)) if !r.is_upgrade()));
assert!(matches!(conn.poll(), Ok(ServerEvent::RequestEnd)));
let mut out = Vec::new();
assert!(matches!(
conn.respond(&Response::switching_protocols("websocket"), &mut out),
Err(HttpError::InvalidState(_))
));
conn.respond(&Response::new(200), &mut out).unwrap();
assert!(matches!(conn.poll(), Ok(ServerEvent::NeedInput)));
}
#[test]
fn test_unregistered_status_keeps_the_space_before_an_empty_reason() {
let mut conn = ServerConn::new();
let _ = conn.feed(b"GET / HTTP/1.1\r\nHost: a\r\n\r\n");
assert_eq!(events(&mut conn), vec!["req /", "end"]);
let mut out = Vec::new();
conn.respond(&Response::new(299), &mut out).unwrap();
assert!(out.starts_with(b"HTTP/1.1 299 \r\n"));
}
#[test]
fn test_connect_tunnel() {
let mut conn = ServerConn::new();
let _ = conn.feed(b"CONNECT example.com:443 HTTP/1.1\r\nHost: example.com:443\r\n\r\nTLS");
assert_eq!(events(&mut conn), vec!["req example.com:443", "end"]);
let mut out = Vec::new();
conn.respond(&Response::new(200), &mut out).unwrap();
assert_eq!(out, b"HTTP/1.1 200 OK\r\n\r\n");
assert!(matches!(conn.poll(), Ok(ServerEvent::Upgraded)));
assert_eq!(conn.remaining_input(), b"TLS");
assert!(conn.request().is_some());
}
}