use alloc::vec::Vec;
use crate::{
body::{Decoder, Encoder, Step, push_framing},
buffer::Buffer,
date::HttpDate,
error::HttpError,
head::{
HeadKind, HeadScanner, Headers, RequestHead, RequestInfo, Scan, Version, interpret_request,
},
limits::Limits,
message::{
OutBody, Response, check_response_field, check_trailers, check_upgrade_protocol,
push_last_chunk_with_trailers, write_field,
},
method::Method,
preface::is_h2_preface,
status::write_status_line,
write::{BodyPlan, OutgoingBody, Output, ResponseHeadRef},
};
#[non_exhaustive]
#[derive(Debug)]
pub enum ServerEvent<'a> {
Request(RequestHead<'a>),
Body(&'a [u8]),
Trailers(Headers<'a>),
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,
}
#[non_exhaustive]
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum Progress {
Idle,
Head,
Body,
AwaitingResponse,
WritingBody,
Upgraded,
Closed,
}
pub struct ServerConn {
buf: Buffer,
limits: Limits,
scanner: HeadScanner,
trailers: HeadScanner,
decoder: Decoder,
info: RequestInfo,
state: ReqState,
resp: RespState,
keep_alive: bool,
allow_keep_alive: bool,
continue_sent: bool,
body_seen: bool,
resp_chunked: bool,
report_trailers: bool,
requests_served: u64,
preface_undecided: 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 {
Self::from_buffer(Buffer::new(limits.buffer_capacity()), limits)
}
#[must_use]
pub fn with_buffer(storage: Vec<u8>, limits: Limits) -> Self {
Self::from_buffer(
Buffer::with_storage(storage, limits.buffer_capacity()),
limits,
)
}
fn from_buffer(mut buf: Buffer, limits: Limits) -> Self {
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,
allow_keep_alive: true,
continue_sent: false,
body_seen: false,
resp_chunked: false,
report_trailers: false,
requests_served: 0,
preface_undecided: true,
}
}
#[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_read() {
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();
if self.preface_undecided {
match is_h2_preface(self.buf.from(start)) {
Some(true) => return Err(self.fail(HttpError::Http2Preface)),
Some(false) => self.preface_undecided = false,
None if self.buf.is_eof() => self.preface_undecided = false,
None => return Ok(ServerEvent::NeedInput),
}
}
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 {
Scan::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));
}
Scan::Partial => return Ok(ServerEvent::NeedInput),
Scan::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,
self.report_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);
self.body_seen = true;
Ok(ServerEvent::Body(self.buf.range(start, start + n)))
}
Step::Trailers => Ok(ServerEvent::Trailers(Headers::new(
self.buf.unread(),
&self.trailers.fields,
))),
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.wipe_consumed();
self.buf.unpin();
self.buf.pin_at_pos();
self.scanner.reset();
self.resp = RespState::Idle;
self.info = RequestInfo::EMPTY;
self.continue_sent = false;
self.body_seen = 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_read())
}
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 fn respond_with_date(
&mut self,
response: &Response,
date: &HttpDate,
out: &mut Vec<u8>,
) -> Result<(), HttpError> {
self.respond_dated(response, Some(date.as_bytes()), out)
}
pub fn respond_head_only(
&mut self,
response: &Response,
out: &mut Vec<u8>,
) -> Result<BodyPlan, HttpError> {
self.start_response(response, None, out)
}
#[cfg(feature = "std")]
pub(crate) fn respond_head_dated(
&mut self,
response: &Response,
date: Option<&[u8]>,
out: &mut Vec<u8>,
) -> Result<BodyPlan, HttpError> {
self.start_response(response, date, out)
}
pub fn send_informational(
&mut self,
interim: &Response,
out: &mut Vec<u8>,
) -> Result<(), HttpError> {
if let Some(e) = interim.error() {
return Err(e);
}
let status = interim.status();
if !(102..=199).contains(&status) {
return Err(HttpError::InvalidStatus);
}
if !matches!(&interim.body, OutBody::Full(b) if b.is_empty()) {
return Err(HttpError::BodyNotAllowed);
}
if !matches!(self.state, ReqState::Body | ReqState::Complete) {
return Err(HttpError::InvalidState(
"an interim response needs a request to answer",
));
}
if self.info.version != Version::Http11 || self.resp != RespState::Idle {
return Err(HttpError::InterimNotAllowed);
}
write_status_line(out, status);
out.extend_from_slice(interim.fields());
out.extend_from_slice(b"\r\n");
Ok(())
}
pub fn set_keep_alive(&mut self, keep: bool) {
self.allow_keep_alive = keep;
if !keep && self.resp != RespState::Idle {
self.keep_alive = false;
}
}
#[must_use]
pub fn keep_alive(&self) -> bool {
self.allow_keep_alive
&& match self.state {
ReqState::Head => true,
ReqState::Body | ReqState::Complete => self.keep_alive,
ReqState::Upgraded | ReqState::Failed | ReqState::Closed => false,
}
}
#[must_use]
pub fn requests_served(&self) -> u64 {
self.requests_served
}
#[must_use]
pub fn progress(&self) -> Progress {
if matches!(self.resp, RespState::Streaming(_)) {
return Progress::WritingBody;
}
match self.state {
ReqState::Head if self.buf.from(self.buf.pin()).is_empty() => Progress::Idle,
ReqState::Head => Progress::Head,
ReqState::Body if self.resp == RespState::Sent => {
if self.decoder.is_read() && self.keep_alive {
Progress::Idle
} else {
Progress::Closed
}
}
ReqState::Body if !self.decoder.is_read() => Progress::Body,
ReqState::Body => Progress::AwaitingResponse,
ReqState::Complete if self.resp != RespState::Sent => Progress::AwaitingResponse,
ReqState::Complete if self.keep_alive => Progress::Idle,
ReqState::Complete => Progress::Closed,
ReqState::Failed if self.resp == RespState::Idle => Progress::AwaitingResponse,
ReqState::Upgraded => Progress::Upgraded,
ReqState::Failed | ReqState::Closed => Progress::Closed,
}
}
pub fn set_report_trailers(&mut self, report: bool) {
self.report_trailers = report;
}
pub fn reset(&mut self) {
self.buf.reset();
self.buf.pin_at_pos();
self.scanner.reset();
self.trailers.reset();
self.decoder = Decoder::new();
self.info = RequestInfo::EMPTY;
self.state = ReqState::Head;
self.resp = RespState::Idle;
self.keep_alive = false;
self.allow_keep_alive = true;
self.continue_sent = false;
self.body_seen = false;
self.resp_chunked = false;
self.requests_served = 0;
self.preface_undecided = true;
}
#[must_use]
pub fn into_buffer(mut self) -> Vec<u8> {
self.buf.take_storage()
}
pub fn shrink_idle(&mut self) -> bool {
if self.progress() != Progress::Idle || self.state != ReqState::Head {
return false;
}
self.buf.unpin();
let freed = self.buf.shrink();
self.buf.pin_at_pos();
if freed {
self.scanner.fields = Vec::new();
self.trailers.fields = Vec::new();
}
freed
}
pub fn set_wipe(&mut self, wipe: bool) {
self.buf.set_wipe(wipe);
}
#[inline]
pub(crate) fn respond_dated<O: Output + ?Sized>(
&mut self,
response: &Response,
date: Option<&[u8]>,
out: &mut O,
) -> Result<(), HttpError> {
let plan = self.start_response(response, date, out)?;
if plan == BodyPlan::Raw {
if let OutBody::Full(body) = &response.body {
out.put(body);
}
}
Ok(())
}
#[inline(always)]
pub(crate) fn start_response<S: HeadSource + ?Sized, O: Output + ?Sized>(
&mut self,
spec: &S,
date: Option<&[u8]>,
out: &mut O,
) -> Result<BodyPlan, HttpError> {
let has_date = spec.check()?;
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, body, upgrade) = (spec.status(), spec.body(), spec.upgrade());
if !(100..=599).contains(&status) || ((100..200).contains(&status) && status != 101) {
return Err(HttpError::InvalidStatus);
}
if upgrade.is_some() && status != 101 {
return Err(HttpError::InvalidState(
"an upgrade protocol needs status 101",
));
}
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) && !matches!(body, OutgoingBody::Empty | OutgoingBody::Full(0)) {
return Err(HttpError::BodyNotAllowed);
}
if tunnel {
let complete_upgrade =
self.body_read() && (self.info.upgrade || method == Method::Connect);
if !complete_upgrade || (status == 101 && upgrade.is_none()) || 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 && !upgrade.is_some_and(|p| self.view().offers(p)) {
return Err(HttpError::UpgradeNotOffered);
}
}
let mut close = failed
|| spec.close()
|| !self.keep_alive
|| !self.allow_keep_alive
|| !self.body_read();
let head_only = method == Method::Head || bodiless || tunnel;
self.resp_chunked = body == OutgoingBody::Chunked;
let (mut encoder, declared) = match body {
OutgoingBody::Empty => (Encoder::Length { remaining: 0 }, Some(0)),
OutgoingBody::Full(n) | OutgoingBody::Length(n) => {
(Encoder::Length { remaining: n }, Some(n))
}
OutgoingBody::Chunked if version == Version::Http10 => {
close = true;
(Encoder::CloseDelimited, None)
}
OutgoingBody::Chunked => (Encoder::Chunked, None),
};
write_status_line(out, status);
if !(bodiless || tunnel) {
push_framing(out, encoder, declared);
}
if tunnel {
close = false;
} else if close {
out.put(b"connection: close\r\n");
} else if version == Version::Http10 {
out.put(b"connection: keep-alive\r\n");
}
if let Some(date) = date.filter(|_| !has_date) {
out.put(b"date: ");
out.put(date);
out.put(b"\r\n");
}
spec.write_fields(out);
out.put(b"\r\n");
if head_only {
encoder = Encoder::Discard;
}
self.keep_alive = !close;
let plan = match body {
OutgoingBody::Empty | OutgoingBody::Full(_) => {
self.resp = RespState::Sent;
self.requests_served += 1;
match body {
OutgoingBody::Full(n) if n > 0 && !head_only => BodyPlan::Raw,
_ => BodyPlan::None,
}
}
OutgoingBody::Length(_) | OutgoingBody::Chunked => {
self.resp = RespState::Streaming(encoder);
BodyPlan::Stream
}
};
if tunnel {
self.state = ReqState::Upgraded;
} else if failed {
self.state = ReqState::Closed;
}
Ok(plan)
}
pub(crate) fn stream_body<O: Output + ?Sized>(
&mut self,
data: &[u8],
out: &mut O,
) -> Result<(), HttpError> {
match &mut self.resp {
RespState::Streaming(encoder) => encoder.write(data, out),
_ => Err(HttpError::InvalidState("no streamed response in progress")),
}
}
pub(crate) fn finish_body<O: Output + ?Sized>(
&mut self,
trailers: &[(&str, &[u8])],
out: &mut O,
) -> Result<(), HttpError> {
let RespState::Streaming(encoder) = &mut self.resp else {
return Err(HttpError::InvalidState("no streamed response in progress"));
};
if !trailers.is_empty() && !self.resp_chunked {
return Err(HttpError::TrailerNotAllowed);
}
match encoder {
Encoder::Chunked => push_last_chunk_with_trailers(out, trailers)?,
_ => {
check_trailers(trailers)?;
encoder.finish(out)?;
}
}
if *encoder == Encoder::CloseDelimited {
self.keep_alive = false;
}
self.resp = RespState::Sent;
self.requests_served += 1;
Ok(())
}
pub fn send_continue(&mut self, out: &mut Vec<u8>) -> Result<(), HttpError> {
if !matches!(self.state, ReqState::Body | ReqState::Complete) {
return Err(HttpError::InvalidState(
"100 Continue needs a request whose body is still to come",
));
}
if self.state == ReqState::Complete
|| self.decoder.is_read()
|| self.body_seen
|| self.resp != RespState::Idle
|| self.info.version != Version::Http11
|| self.continue_sent
{
return Err(HttpError::InterimNotAllowed);
}
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> {
self.stream_body(data, out)
}
pub fn end_body(&mut self, out: &mut Vec<u8>) -> Result<(), HttpError> {
self.end_body_with_trailers(&[], out)
}
pub fn end_body_with_trailers(
&mut self,
trailers: &[(&str, &[u8])],
out: &mut Vec<u8>,
) -> Result<(), HttpError> {
self.finish_body(trailers, out)
}
}
pub(crate) trait HeadSource {
fn status(&self) -> u16;
fn body(&self) -> OutgoingBody;
fn close(&self) -> bool;
fn upgrade(&self) -> Option<&[u8]>;
fn check(&self) -> Result<bool, HttpError>;
fn write_fields<O: Output + ?Sized>(&self, out: &mut O);
}
impl HeadSource for Response {
#[inline]
fn status(&self) -> u16 {
Response::status(self)
}
#[inline]
fn body(&self) -> OutgoingBody {
match &self.body {
OutBody::Full(b) => OutgoingBody::Full(b.len() as u64),
OutBody::Length(n) => OutgoingBody::Length(*n),
OutBody::Chunked => OutgoingBody::Chunked,
}
}
#[inline]
fn close(&self) -> bool {
self.close
}
#[inline]
fn upgrade(&self) -> Option<&[u8]> {
self.upgrade_protocol()
}
#[inline]
fn check(&self) -> Result<bool, HttpError> {
match self.error() {
Some(e) => Err(e),
None => Ok(self.has_date),
}
}
#[inline]
fn write_fields<O: Output + ?Sized>(&self, out: &mut O) {
out.put(self.fields());
}
}
impl HeadSource for ResponseHeadRef<'_> {
#[inline]
fn status(&self) -> u16 {
self.status
}
#[inline]
fn body(&self) -> OutgoingBody {
self.body
}
#[inline]
fn close(&self) -> bool {
self.close
}
#[inline]
fn upgrade(&self) -> Option<&[u8]> {
self.upgrade.map(str::as_bytes)
}
fn check(&self) -> Result<bool, HttpError> {
if let Some(protocol) = self.upgrade {
check_upgrade_protocol(protocol.as_bytes())?;
}
let mut has_date = false;
for &(name, value) in self.fields {
has_date |= check_response_field(name, value)?;
}
Ok(has_date)
}
fn write_fields<O: Output + ?Sized>(&self, out: &mut O) {
if let Some(protocol) = self.upgrade {
out.put(b"connection: upgrade\r\nupgrade: ");
out.put(protocol.as_bytes());
out.put(b"\r\n");
}
for &(name, value) in self.fields {
write_field(out, name, value);
}
}
}
#[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());
}
}