From 6708271862e1ef1b23c6501c309ed658b4f20820 Mon Sep 17 00:00:00 2001 From: Andreea Florescu Date: Mon, 27 Jan 2020 12:28:50 +0200 Subject: [PATCH] initial commit Pulled the code from firecracker-microvm/src/micro-http. Signed-off-by: George Pisaltu Signed-off-by: Andreea Florescu --- Cargo.lock | 32 ++ Cargo.toml | 7 + src/common/headers.rs | 397 ++++++++++++++++++++ src/common/mod.rs | 355 ++++++++++++++++++ src/connection.rs | 783 +++++++++++++++++++++++++++++++++++++++ src/lib.rs | 126 +++++++ src/request.rs | 470 ++++++++++++++++++++++++ src/response.rs | 291 +++++++++++++++ src/server.rs | 834 ++++++++++++++++++++++++++++++++++++++++++ 9 files changed, 3295 insertions(+) create mode 100644 Cargo.lock create mode 100644 Cargo.toml create mode 100644 src/common/headers.rs create mode 100644 src/common/mod.rs create mode 100644 src/connection.rs create mode 100644 src/lib.rs create mode 100644 src/request.rs create mode 100644 src/response.rs create mode 100644 src/server.rs diff --git a/Cargo.lock b/Cargo.lock new file mode 100644 index 0000000..fcbbcea --- /dev/null +++ b/Cargo.lock @@ -0,0 +1,32 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +[[package]] +name = "bitflags" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "epoll" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "bitflags 1.2.1 (registry+https://github.com/rust-lang/crates.io-index)", + "libc 0.2.66 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "libc" +version = "0.2.66" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "micro_http" +version = "0.1.0" +dependencies = [ + "epoll 4.1.0 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[metadata] +"checksum bitflags 1.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "cf1de2fe8c75bc145a2f577add951f8134889b4795d47466a54a5c846d691693" +"checksum epoll 4.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "990bcfe26bea89669ede68c3f970f61d02568dbc8660317c98d805ea4e710685" +"checksum libc 0.2.66 (registry+https://github.com/rust-lang/crates.io-index)" = "d515b1f41455adea1313a4a2ac8a8a477634fbae63cc6100e3aebb207ce61558" diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 0000000..22c82e0 --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,7 @@ +[package] +name = "micro_http" +version = "0.1.0" +authors = ["Amazon Firecracker team "] + +[dependencies] +epoll = ">=4.0.1" diff --git a/src/common/headers.rs b/src/common/headers.rs new file mode 100644 index 0000000..8e99bd5 --- /dev/null +++ b/src/common/headers.rs @@ -0,0 +1,397 @@ +// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +use std::result::Result; + +use RequestError; + +/// Wrapper over an HTTP Header type. +#[derive(Debug, Eq, Hash, PartialEq)] +pub enum Header { + /// Header `Content-Length`. + ContentLength, + /// Header `Content-Type`. + ContentType, + /// Header `Expect`. + Expect, + /// Header `Transfer-Encoding`. + TransferEncoding, + /// Header `Server`. + Server, +} + +impl Header { + pub fn raw(&self) -> &'static [u8] { + match self { + Header::ContentLength => b"Content-Length", + Header::ContentType => b"Content-Type", + Header::Expect => b"Expect", + Header::TransferEncoding => b"Transfer-Encoding", + Header::Server => b"Server", + } + } + + fn try_from(string: &[u8]) -> Result { + if let Ok(utf8_string) = String::from_utf8(string.to_vec()) { + match utf8_string.trim() { + "Content-Length" => Ok(Header::ContentLength), + "Content-Type" => Ok(Header::ContentType), + "Expect" => Ok(Header::Expect), + "Transfer-Encoding" => Ok(Header::TransferEncoding), + "Server" => Ok(Header::Server), + _ => Err(RequestError::InvalidHeader), + } + } else { + Err(RequestError::InvalidRequest) + } + } +} + +/// Wrapper over the list of headers associated with a Request that we need +/// in order to parse the request correctly and be able to respond to it. +/// +/// The only `Content-Type`s supported are `text/plain` and `application/json`, which are both +/// in plain text actually and don't influence our parsing process. +/// +/// All the other possible header fields are not necessary in order to serve this connection +/// and, thus, are not of interest to us. However, we still look for header fields that might +/// invalidate our request as we don't support the full set of HTTP/1.1 specification. +/// Such header entries are "Transfer-Encoding: identity; q=0", which means a compression +/// algorithm is applied to the body of the request, or "Expect: 103-checkpoint". +#[derive(Debug)] +pub struct Headers { + /// The `Content-Length` header field tells us how many bytes we need to receive + /// from the source after the headers. + content_length: i32, + /// The `Expect` header field is set when the headers contain the entry "Expect: 100-continue". + /// This means that, per HTTP/1.1 specifications, we must send a response with the status code + /// 100 after we have received the headers in order to receive the body of the request. This + /// field should be known immediately after parsing the headers. + expect: bool, + /// `Chunked` is a possible value of the `Transfer-Encoding` header field and every HTTP/1.1 + /// server must support it. It is useful only when receiving the body of the request and should + /// be known immediately after parsing the headers. + chunked: bool, +} + +impl Headers { + /// By default Requests are created with no headers. + pub fn default() -> Headers { + Headers { + content_length: 0, + expect: false, + chunked: false, + } + } + + /// Expects one header line and parses it, updating the header structure or returning an + /// error if the header is invalid. + /// + /// # Errors + /// `UnsupportedHeader` is returned when the parsed header line is not of interest + /// to us or when it is unrecognizable. + /// `InvalidHeader` is returned when the parsed header is formatted incorrectly or suggests + /// that the client is using HTTP features that we do not support in this implementation, + /// which invalidates the request. + pub fn parse_header_line(&mut self, header_line: &[u8]) -> Result<(), RequestError> { + // Headers must be ASCII, so also UTF-8 valid. + match std::str::from_utf8(header_line) { + Ok(headers_str) => { + let entry = headers_str.split(": ").collect::>(); + if entry.len() != 2 { + return Err(RequestError::InvalidHeader); + } + if let Ok(head) = Header::try_from(entry[0].as_bytes()) { + match head { + Header::ContentLength => { + let try_numeric: Result = + std::str::FromStr::from_str(entry[1].trim()); + if let Ok(content_length) = try_numeric { + self.content_length = content_length; + Ok(()) + } else { + Err(RequestError::InvalidHeader) + } + } + Header::ContentType => { + match MediaType::try_from(entry[1].trim().as_bytes()) { + Ok(_) => Ok(()), + Err(_) => Err(RequestError::UnsupportedHeader), + } + } + Header::TransferEncoding => match entry[1].trim() { + "chunked" => { + self.chunked = true; + Ok(()) + } + "identity; q=0" => Err(RequestError::InvalidHeader), + _ => Err(RequestError::UnsupportedHeader), + }, + Header::Expect => match entry[1].trim() { + "100-continue" => { + self.expect = true; + Ok(()) + } + _ => Err(RequestError::InvalidHeader), + }, + Header::Server => Ok(()), + } + } else { + Err(RequestError::UnsupportedHeader) + } + } + _ => Err(RequestError::InvalidHeader), + } + } + + /// Returns the content length of the body. + pub fn content_length(&self) -> i32 { + self.content_length + } + + /// Returns `true` if the transfer encoding is chunked. + #[allow(unused)] + pub fn chunked(&self) -> bool { + self.chunked + } + + /// Returns `true` if the client is expecting the code 100. + #[allow(unused)] + pub fn expect(&self) -> bool { + self.expect + } + + #[cfg(test)] + pub fn new(content_length: i32, expect: bool, chunked: bool) -> Self { + Headers { + content_length, + expect, + chunked, + } + } + + /// Parses a byte slice into a Headers structure for a HTTP request. + /// + /// The byte slice is expected to have the following format:
+ /// * Request Header Lines " CRLF"- Optional
+ /// There can be any number of request headers, including none, followed by + /// an extra sequence of Carriage Return and Line Feed. + /// All header fields are parsed. However, only the ones present in the + /// [`Headers`](struct.Headers.html) struct are relevant to us and stored + /// for future use. + /// + /// # Errors + /// The function returns `InvalidHeader` when parsing the byte stream fails. + /// + /// # Examples + /// + /// ``` + /// extern crate micro_http; + /// use micro_http::Headers; + /// + /// let request_headers = Headers::try_from(b"Content-Length: 55\r\n\r\n"); + /// ``` + pub fn try_from(bytes: &[u8]) -> Result { + // Headers must be ASCII, so also UTF-8 valid. + if let Ok(text) = std::str::from_utf8(bytes) { + let mut headers = Headers::default(); + + let header_lines = text.split("\r\n"); + for header_line in header_lines { + if header_line.is_empty() { + break; + } + match headers.parse_header_line(header_line.as_bytes()) { + Ok(_) | Err(RequestError::UnsupportedHeader) => continue, + Err(e) => return Err(e), + }; + } + return Ok(headers); + } + Err(RequestError::InvalidRequest) + } +} + +/// Wrapper over supported Media Types. +#[derive(Clone, Copy, Debug, PartialEq)] +pub enum MediaType { + /// Media Type: "text/plain". + PlainText, + /// Media Type: "application/json". + ApplicationJson, +} + +impl Default for MediaType { + fn default() -> Self { + MediaType::ApplicationJson + } +} + +impl MediaType { + fn try_from(bytes: &[u8]) -> Result { + if bytes.is_empty() { + return Err(RequestError::InvalidRequest); + } + let utf8_slice = + String::from_utf8(bytes.to_vec()).map_err(|_| RequestError::InvalidRequest)?; + match utf8_slice.as_str().trim() { + "text/plain" => Ok(MediaType::PlainText), + "application/json" => Ok(MediaType::ApplicationJson), + _ => Err(RequestError::InvalidRequest), + } + } + + /// Returns a static string representation of the object. + pub fn as_str(self) -> &'static str { + match self { + MediaType::PlainText => "text/plain", + MediaType::ApplicationJson => "application/json", + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_default() { + let headers = Headers::default(); + assert_eq!(headers.content_length(), 0); + assert_eq!(headers.chunked(), false); + assert_eq!(headers.expect(), false); + } + + #[test] + fn test_try_from_media() { + assert_eq!( + MediaType::try_from(b"application/json").unwrap(), + MediaType::ApplicationJson + ); + + assert_eq!( + MediaType::try_from(b"text/plain").unwrap(), + MediaType::PlainText + ); + + assert_eq!( + MediaType::try_from(b"").unwrap_err(), + RequestError::InvalidRequest + ); + + assert_eq!( + MediaType::try_from(b"application/json-patch").unwrap_err(), + RequestError::InvalidRequest + ); + } + + #[test] + fn test_media_as_str() { + let media_type = MediaType::ApplicationJson; + assert_eq!(media_type.as_str(), "application/json"); + + let media_type = MediaType::PlainText; + assert_eq!(media_type.as_str(), "text/plain"); + } + + #[test] + fn test_try_from_headers() { + // Valid headers. + assert_eq!( + Headers::try_from( + b"Last-Modified: Tue, 15 Nov 1994 12:45:26 GMT\r\nContent-Length: 55\r\n\r\n" + ) + .unwrap() + .content_length, + 55 + ); + + let bytes: [u8; 10] = [130, 140, 150, 130, 140, 150, 130, 140, 150, 160]; + // Invalid headers. + assert!(Headers::try_from(&bytes[..]).is_err()); + } + + #[test] + fn test_parse_header_line() { + let mut header = Headers::default(); + + // Invalid header syntax. + assert_eq!( + header.parse_header_line(b"Expect"), + Err(RequestError::InvalidHeader) + ); + + // Invalid content length. + assert_eq!( + header.parse_header_line(b"Content-Length: five"), + Err(RequestError::InvalidHeader) + ); + + // Invalid transfer encoding. + assert_eq!( + header.parse_header_line(b"Transfer-Encoding: gzip"), + Err(RequestError::UnsupportedHeader) + ); + + // Invalid expect. + assert_eq!( + header + .parse_header_line(b"Expect: 102-processing") + .unwrap_err(), + RequestError::InvalidHeader + ); + + // Unsupported media type. + assert_eq!( + header + .parse_header_line(b"Content-Type: application/json-patch") + .unwrap_err(), + RequestError::UnsupportedHeader + ); + + // Invalid input format. + let input: [u8; 10] = [130, 140, 150, 130, 140, 150, 130, 140, 150, 160]; + assert_eq!( + header.parse_header_line(&input[..]).unwrap_err(), + RequestError::InvalidHeader + ); + + // Test valid transfer encoding. + assert!(header + .parse_header_line(b"Transfer-Encoding: chunked") + .is_ok()); + assert!(header.chunked()); + + // Test valid expect. + assert!(header.parse_header_line(b"Expect: 100-continue").is_ok()); + assert!(header.expect()); + + // Test valid media type. + assert!(header + .parse_header_line(b"Content-Type: application/json") + .is_ok()); + } + + #[test] + fn test_header_try_from() { + // Bad header. + assert_eq!( + Header::try_from(b"Encoding").unwrap_err(), + RequestError::InvalidHeader + ); + + // Invalid encoding. + let input: [u8; 10] = [130, 140, 150, 130, 140, 150, 130, 140, 150, 160]; + assert_eq!( + Header::try_from(&input[..]).unwrap_err(), + RequestError::InvalidRequest + ); + + // Test valid headers. + let header = Header::try_from(b"Expect").unwrap(); + assert_eq!(header.raw(), b"Expect"); + + let header = Header::try_from(b"Transfer-Encoding").unwrap(); + assert_eq!(header.raw(), b"Transfer-Encoding"); + } +} diff --git a/src/common/mod.rs b/src/common/mod.rs new file mode 100644 index 0000000..546c2ea --- /dev/null +++ b/src/common/mod.rs @@ -0,0 +1,355 @@ +// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +use std::fmt::{Display, Error, Formatter}; + +pub mod headers; + +pub mod ascii { + pub const CR: u8 = b'\r'; + pub const COLON: u8 = b':'; + pub const LF: u8 = b'\n'; + pub const SP: u8 = b' '; + pub const CRLF_LEN: usize = 2; +} + +/// Errors associated with parsing the HTTP Request from a u8 slice. +#[derive(Debug, PartialEq)] +pub enum RequestError { + /// The HTTP Method is not supported or it is invalid. + InvalidHttpMethod(&'static str), + /// Request URI is invalid. + InvalidUri(&'static str), + /// The HTTP Version in the Request is not supported or it is invalid. + InvalidHttpVersion(&'static str), + /// The header specified may be valid, but is not supported by this HTTP implementation. + UnsupportedHeader, + /// Header specified is invalid. + InvalidHeader, + /// The Request is invalid and cannot be served. + InvalidRequest, +} + +impl Display for RequestError { + fn fmt(&self, f: &mut Formatter) -> Result<(), Error> { + match self { + RequestError::InvalidHttpMethod(inner) => write!(f, "Invalid HTTP Method: {}", inner), + RequestError::InvalidUri(inner) => write!(f, "Invalid URI: {}", inner), + RequestError::InvalidHttpVersion(inner) => write!(f, "Invalid HTTP Version: {}", inner), + RequestError::UnsupportedHeader => write!(f, "Unsupported header."), + RequestError::InvalidHeader => write!(f, "Invalid header."), + RequestError::InvalidRequest => write!(f, "Invalid request."), + } + } +} + +/// Errors associated with a HTTP Connection. +#[derive(Debug)] +pub enum ConnectionError { + /// The request parsing has failed. + ParseError(RequestError), + /// Could not perform a stream operation successfully. + StreamError(std::io::Error), + /// Attempted to read or write on a closed connection. + ConnectionClosed, + /// Attempted to write on a stream when there was nothing to write. + InvalidWrite, +} + +impl Display for ConnectionError { + fn fmt(&self, f: &mut Formatter) -> Result<(), Error> { + match self { + ConnectionError::ParseError(inner) => write!(f, "Parsing error: {}", inner), + ConnectionError::StreamError(inner) => write!(f, "Stream error: {}", inner), + ConnectionError::ConnectionClosed => write!(f, "Connection closed."), + ConnectionError::InvalidWrite => write!(f, "Invalid write attempt."), + } + } +} + +/// Errors pertaining to `HttpServer`. +#[derive(Debug)] +pub enum ServerError { + /// Epoll operations failed. + IOError(std::io::Error), + /// Error from one of the connections. + ConnectionError(ConnectionError), + /// Server maximum capacity has been reached. + ServerFull, +} + +impl Display for ServerError { + fn fmt(&self, f: &mut Formatter) -> Result<(), Error> { + match self { + ServerError::IOError(inner) => write!(f, "IO error: {}", inner), + ServerError::ConnectionError(inner) => write!(f, "Connection error: {}", inner), + ServerError::ServerFull => write!(f, "Server is full."), + } + } +} + +/// The Body associated with an HTTP Request or Response. +/// +/// ## Examples +/// ``` +/// extern crate micro_http; +/// use micro_http::Body; +/// let body = Body::new("This is a test body.".to_string()); +/// assert_eq!(body.raw(), b"This is a test body."); +/// assert_eq!(body.len(), 20); +/// ``` +#[derive(Clone, Debug, PartialEq)] +pub struct Body { + /// Body of the HTTP message as bytes. + pub body: Vec, +} + +impl Body { + /// Creates a new `Body` from a `String` input. + pub fn new>>(body: T) -> Self { + Body { body: body.into() } + } + + /// Returns the body as an `u8 slice`. + pub fn raw(&self) -> &[u8] { + self.body.as_slice() + } + + /// Returns the length of the `Body`. + pub fn len(&self) -> usize { + self.body.len() + } + + /// Checks if the body is empty, ie with zero length + pub fn is_empty(&self) -> bool { + self.body.len() == 0 + } +} + +/// Supported HTTP Methods. +#[derive(Clone, Copy, Debug, PartialEq)] +pub enum Method { + /// GET Method. + Get, + /// PUT Method. + Put, + /// PATCH Method. + Patch, +} + +impl Method { + /// Returns a `Method` object if the parsing of `bytes` is successful. + /// + /// The method is case sensitive. A call to try_from with the input b"get" will return + /// an error, but when using the input b"GET", it returns Method::Get. + /// + /// # Errors + /// Returns `RequestError` if the method specified by `bytes` is unsupported. + pub fn try_from(bytes: &[u8]) -> Result { + match bytes { + b"GET" => Ok(Method::Get), + b"PUT" => Ok(Method::Put), + b"PATCH" => Ok(Method::Patch), + _ => Err(RequestError::InvalidHttpMethod("Unsupported HTTP method.")), + } + } + + /// Returns an `u8 slice` corresponding to the Method. + pub fn raw(self) -> &'static [u8] { + match self { + Method::Get => b"GET", + Method::Put => b"PUT", + Method::Patch => b"PATCH", + } + } +} + +/// Supported HTTP Versions. +/// +/// # Examples +/// ``` +/// extern crate micro_http; +/// use micro_http::Version; +/// let version = Version::try_from(b"HTTP/1.1"); +/// assert!(version.is_ok()); +/// +/// let version = Version::try_from(b"http/1.1"); +/// assert!(version.is_err()); +/// ``` +#[derive(Clone, Copy, Debug, PartialEq)] +pub enum Version { + /// HTTP/1.0 + Http10, + /// HTTP/1.1 + Http11, +} + +impl Version { + /// HTTP Version as an `u8 slice`. + pub fn raw(self) -> &'static [u8] { + match self { + Version::Http10 => b"HTTP/1.0", + Version::Http11 => b"HTTP/1.1", + } + } + + /// Creates a new HTTP Version from an `u8 slice`. + /// + /// The supported versions are HTTP/1.0 and HTTP/1.1. + /// The version is case sensitive and the accepted input is upper case. + /// + /// # Errors + /// Returns a `RequestError` when the version is not supported. + pub fn try_from(bytes: &[u8]) -> Result { + match bytes { + b"HTTP/1.0" => Ok(Version::Http10), + b"HTTP/1.1" => Ok(Version::Http11), + _ => Err(RequestError::InvalidHttpVersion( + "Unsupported HTTP version.", + )), + } + } + + /// Returns the default HTTP version = HTTP/1.1. + pub fn default() -> Self { + Version::Http11 + } +} + +#[cfg(test)] +mod tests { + use super::*; + + impl PartialEq for ConnectionError { + fn eq(&self, other: &Self) -> bool { + use self::ConnectionError::*; + match (self, other) { + (ParseError(_), ParseError(_)) => true, + (ConnectionClosed, ConnectionClosed) => true, + (StreamError(_), StreamError(_)) => true, + (InvalidWrite, InvalidWrite) => true, + _ => false, + } + } + } + + #[test] + fn test_version() { + // Tests for raw() + assert_eq!(Version::Http10.raw(), b"HTTP/1.0"); + assert_eq!(Version::Http11.raw(), b"HTTP/1.1"); + + // Tests for try_from() + assert_eq!(Version::try_from(b"HTTP/1.0").unwrap(), Version::Http10); + assert_eq!(Version::try_from(b"HTTP/1.1").unwrap(), Version::Http11); + assert_eq!( + Version::try_from(b"HTTP/2.0").unwrap_err(), + RequestError::InvalidHttpVersion("Unsupported HTTP version.") + ); + + // Test for default() + assert_eq!(Version::default(), Version::Http11); + } + + #[test] + fn test_method() { + // Test for raw + assert_eq!(Method::Get.raw(), b"GET"); + assert_eq!(Method::Put.raw(), b"PUT"); + assert_eq!(Method::Patch.raw(), b"PATCH"); + + // Tests for try_from + assert_eq!(Method::try_from(b"GET").unwrap(), Method::Get); + assert_eq!(Method::try_from(b"PUT").unwrap(), Method::Put); + assert_eq!(Method::try_from(b"PATCH").unwrap(), Method::Patch); + assert_eq!( + Method::try_from(b"POST").unwrap_err(), + RequestError::InvalidHttpMethod("Unsupported HTTP method.") + ); + } + + #[test] + fn test_body() { + let body = Body::new("".to_string()); + // Test for is_empty + assert!(body.is_empty()); + let body = Body::new("This is a body.".to_string()); + // Test for len + assert_eq!(body.len(), 15); + // Test for raw + assert_eq!(body.raw(), b"This is a body."); + } + + #[test] + fn test_display_request_error() { + assert_eq!( + format!("{}", RequestError::InvalidHttpMethod("test")), + "Invalid HTTP Method: test" + ); + assert_eq!( + format!("{}", RequestError::InvalidUri("test")), + "Invalid URI: test" + ); + assert_eq!( + format!("{}", RequestError::InvalidHttpVersion("test")), + "Invalid HTTP Version: test" + ); + assert_eq!( + format!("{}", RequestError::InvalidHeader), + "Invalid header." + ); + assert_eq!( + format!("{}", RequestError::UnsupportedHeader), + "Unsupported header." + ); + assert_eq!( + format!("{}", RequestError::InvalidRequest), + "Invalid request." + ); + } + + #[test] + fn test_display_connection_error() { + assert_eq!( + format!( + "{}", + ConnectionError::ParseError(RequestError::InvalidRequest) + ), + "Parsing error: Invalid request." + ); + assert_eq!( + format!( + "{}", + ConnectionError::StreamError(std::io::Error::from_raw_os_error(11)) + ), + "Stream error: Resource temporarily unavailable (os error 11)" + ); + assert_eq!( + format!("{}", ConnectionError::ConnectionClosed), + "Connection closed." + ); + assert_eq!( + format!("{}", ConnectionError::InvalidWrite), + "Invalid write attempt." + ); + } + + #[test] + fn test_display_server_error() { + assert_eq!( + format!( + "{}", + ServerError::ConnectionError(ConnectionError::ConnectionClosed) + ), + "Connection error: Connection closed." + ); + assert_eq!(format!("{}", ServerError::ServerFull), "Server is full."); + assert_eq!( + format!( + "{}", + ServerError::IOError(std::io::Error::from_raw_os_error(11)) + ), + "IO error: Resource temporarily unavailable (os error 11)" + ); + } +} diff --git a/src/connection.rs b/src/connection.rs new file mode 100644 index 0000000..7e10a3e --- /dev/null +++ b/src/connection.rs @@ -0,0 +1,783 @@ +// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +use std::collections::VecDeque; +use std::io::{Read, Write}; + +use common::ascii::{CR, CRLF_LEN, LF}; +use common::Body; +pub use common::{ConnectionError, RequestError}; +use headers::Headers; +use request::{find, Request, RequestLine}; +use response::{Response, StatusCode}; + +const BUFFER_SIZE: usize = 1024; + +/// Describes the state machine of an HTTP connection. +pub enum ConnectionState { + WaitingForRequestLine, + WaitingForHeaders, + WaitingForBody, + RequestReady, +} + +/// A wrapper over a HTTP Connection. +pub struct HttpConnection { + /// A partial request that is still being received. + pending_request: Option, + /// Stream implementing `Read` and `Write`, capable of sending and + /// receiving bytes. + stream: T, + /// The state of the connection regarding the current request that + /// is being processed. + state: ConnectionState, + /// Buffer where we store the bytes we read from the stream. + buffer: [u8; BUFFER_SIZE], + /// The index in the buffer from where we have to start reading in + /// the next `try_read` call. + read_cursor: usize, + /// Contains all bytes pertaining to the body of the request that + /// is currently being processed. + body_vec: Vec, + /// Represents how many bytes from the body of the request are still + /// to be read. + body_bytes_to_be_read: i32, + /// A queue of all requests that have been fully received and parsed. + parsed_requests: VecDeque, + /// A queue of requests that are waiting to be sent. + response_queue: VecDeque, + /// A buffer containing the bytes of a response that is currently + /// being sent. + response_buffer: Option>, +} + +impl HttpConnection { + /// Creates an empty connection. + pub fn new(stream: T) -> Self { + HttpConnection { + pending_request: None, + stream, + state: ConnectionState::WaitingForRequestLine, + buffer: [0; BUFFER_SIZE], + read_cursor: 0, + body_vec: vec![], + body_bytes_to_be_read: 0, + parsed_requests: VecDeque::new(), + response_queue: VecDeque::new(), + response_buffer: None, + } + } + + /// Tries to read new bytes from the stream and automatically update the request. + /// Meant to be used only with non-blocking streams and an `EPOLL` structure. + /// Should be called whenever an `EPOLLIN` event is signaled. + /// + /// # Errors + /// `StreamError` is returned when an IO operation fails. + /// `ConnectionClosed` is returned when a client prematurely closes the connection. + /// `ParseError` is returned when a parsing operation fails. + pub fn try_read(&mut self) -> Result<(), ConnectionError> { + // Read some bytes from the stream, which will be appended to what is already + // present in the buffer from a previous call of `try_read`. There are already + // `read_cursor` bytes present in the buffer. + let end_cursor = self.read_bytes()?; + + let mut line_start_index = 0; + loop { + match self.state { + ConnectionState::WaitingForRequestLine => { + if !self.parse_request_line(&mut line_start_index, end_cursor)? { + return Ok(()); + } + } + ConnectionState::WaitingForHeaders => { + if !self.parse_headers(&mut line_start_index, end_cursor)? { + return Ok(()); + } + } + ConnectionState::WaitingForBody => { + if !self.parse_body(&mut line_start_index, end_cursor)? { + return Ok(()); + } + } + ConnectionState::RequestReady => { + // This request is ready to be passed for handling. + // Update the state machine to expect a new request and push this request into + // the `parsed_requests` queue. + self.state = ConnectionState::WaitingForRequestLine; + self.body_bytes_to_be_read = 0; + self.parsed_requests + .push_back(self.pending_request.take().unwrap()); + } + }; + } + } + + // Reads a maximum of 1024 bytes from the stream into `buffer`. + // The return value represents the end index of what we have just appended. + fn read_bytes(&mut self) -> Result { + // Append new bytes to what we already have in the buffer. + let bytes_read = self + .stream + .read(&mut self.buffer[self.read_cursor..]) + .map_err(ConnectionError::StreamError)?; + + // If the read returned 0 then the client has closed the connection. + if bytes_read == 0 { + return Err(ConnectionError::ConnectionClosed); + } + + Ok(bytes_read + self.read_cursor) + } + + // Parses bytes in `buffer` for a valid request line. + // Returns `false` if there are no more bytes to be parsed in the buffer. + fn parse_request_line( + &mut self, + start: &mut usize, + end: usize, + ) -> Result { + match find(&self.buffer[*start..end], &[CR, LF]) { + Some(line_end_index) => { + let line = &self.buffer[*start..(*start + line_end_index)]; + + *start = *start + line_end_index + CRLF_LEN; + let request_line = + RequestLine::try_from(line).map_err(ConnectionError::ParseError)?; + + // Form the request with a valid request line, which is the bare minimum + // for a valid request. + self.pending_request = Some(Request { + request_line, + headers: Headers::default(), + body: None, + }); + self.state = ConnectionState::WaitingForHeaders; + Ok(true) + } + None => { + // The request line is longer than BUFFER_SIZE bytes, so the request is invalid. + if end == BUFFER_SIZE && *start == 0 { + return Err(ConnectionError::ParseError(RequestError::InvalidRequest)); + } else { + // Move the incomplete request line to the beginning of the buffer and wait + // for the next `try_read` call to complete it. + // This can only happen if another request was sent before this one, as the + // limit for the length of a request line in this implementation is 1024 bytes. + self.shift_buffer_left(*start, end); + } + Ok(false) + } + } + } + + // Parses bytes in `buffer` for header fields. + // Returns `false` if there are no more bytes to be parsed in the buffer. + fn parse_headers( + &mut self, + line_start_index: &mut usize, + end_cursor: usize, + ) -> Result { + match find(&self.buffer[*line_start_index..end_cursor], &[CR, LF]) { + // We have found the end of the headers. + // `line_start_index` points to the end of the most recently found CR LF + // sequence. That means that if we found the next CR LF sequence at this index, + // they are, in fact, a CR LF CR LF sequence, which marks the end of the header + // fields, per HTTP specification. + Some(0) => { + // If our current state is `WaitingForHeaders`, it means that we already have + // a valid request formed from a request line, so it's safe to unwrap. + let request = self.pending_request.as_mut().unwrap(); + if request.headers.content_length() == 0 { + self.state = ConnectionState::RequestReady; + } else { + if request.headers.expect() { + // Send expect. + let expect_response = + Response::new(request.http_version(), StatusCode::Continue); + self.response_queue.push_back(expect_response); + } + + self.body_bytes_to_be_read = request.headers.content_length(); + request.body = Some(Body::new(vec![])); + self.state = ConnectionState::WaitingForBody; + } + + // Update the index for the next header. + *line_start_index += CRLF_LEN; + Ok(true) + } + // We have found the end of a header line. + Some(relative_line_end_index) => { + let request = self.pending_request.as_mut().unwrap(); + // The `line_end_index` relative to the whole buffer. + let line_end_index = relative_line_end_index + *line_start_index; + + // Get the line slice and parse it. + let line = &self.buffer[*line_start_index..line_end_index]; + match request.headers.parse_header_line(line) { + // If a header is unsupported we ignore it. + Ok(_) | Err(RequestError::UnsupportedHeader) => {} + // If parsing the header invalidates the request, we propagate + // the error. + Err(e) => return Err(ConnectionError::ParseError(e)), + }; + + // Update the `line_start_index` to where we finished parsing. + *line_start_index = line_end_index + CRLF_LEN; + Ok(true) + } + // If we have an incomplete header line. + None => { + // If we have parsed BUFFER_SIZE bytes and still haven't found the header + // line end sequence. + if *line_start_index == 0 && end_cursor == BUFFER_SIZE { + // Header line is longer than BUFFER_SIZE bytes, so it is invalid. + return Err(ConnectionError::ParseError(RequestError::InvalidHeader)); + } + // Move the incomplete header line from the end of the buffer to + // the beginning, so that we can append the rest of the line and + // parse it in the next `try_read` call. + self.shift_buffer_left(*line_start_index, end_cursor); + Ok(false) + } + } + } + + // Parses bytes in `buffer` to be put into the request body, if there should be one. + // Returns `false` if there are no more bytes to be parsed in the buffer. + fn parse_body( + &mut self, + line_start_index: &mut usize, + end_cursor: usize, + ) -> Result { + // If what we have just read is not enough to complete the request and + // there are more bytes pertaining to the body of the request. + if self.body_bytes_to_be_read > end_cursor as i32 - *line_start_index as i32 { + // Append everything that we read to our current incomplete body and update + // `body_bytes_to_be_read`. + self.body_vec + .extend_from_slice(&self.buffer[*line_start_index..end_cursor]); + self.body_bytes_to_be_read -= end_cursor as i32 - *line_start_index as i32; + + // Clear the buffer and reset the starting index. + for i in 0..BUFFER_SIZE { + self.buffer[i] = 0; + } + self.read_cursor = 0; + + return Ok(false); + } + + // Append only the remaining necessary bytes to the body of the request. + self.body_vec.extend_from_slice( + &self.buffer + [*line_start_index..(*line_start_index + self.body_bytes_to_be_read as usize)], + ); + *line_start_index += self.body_bytes_to_be_read as usize; + self.body_bytes_to_be_read = 0; + + let request = self.pending_request.as_mut().unwrap(); + // If there are no more bytes to be read for this request. + // Assign the body of the request. + let placeholder: Vec<_> = self + .body_vec + .drain(..request.headers.content_length() as usize) + .collect(); + request.body = Some(Body::new(placeholder)); + + // If we read more bytes than we should have into the body of the request. + if !self.body_vec.is_empty() { + return Err(ConnectionError::ParseError(RequestError::InvalidRequest)); + } + + self.state = ConnectionState::RequestReady; + Ok(true) + } + + /// Tries to write the first available response to the provided stream. + /// Meant to be used only with non-blocking streams and an `EPOLL` structure. + /// Should be called whenever an `EPOLLOUT` event is signaled. + /// + /// # Errors + /// `StreamError` is returned when an IO operation fails. + /// `ConnectionClosed` is returned when trying to write on a closed connection. + /// `InvalidWrite` is returned when trying to write on a connection with an + /// empty outgoing buffer. + pub fn try_write(&mut self) -> Result<(), ConnectionError> { + if self.response_buffer.is_none() { + if let Some(response) = self.response_queue.pop_front() { + let mut response_buffer_vec: Vec = Vec::new(); + response + .write_all(&mut response_buffer_vec) + .map_err(ConnectionError::StreamError)?; + self.response_buffer = Some(response_buffer_vec); + } else { + return Err(ConnectionError::InvalidWrite); + } + } + + let mut response_fully_written = false; + let mut connection_closed = false; + + if let Some(response_buffer_vec) = self.response_buffer.as_mut() { + let bytes_to_be_written = response_buffer_vec.len(); + match self.stream.write(response_buffer_vec.as_slice()) { + Ok(0) | Err(_) => { + connection_closed = true; + } + Ok(bytes_written) => { + if bytes_written != bytes_to_be_written { + response_buffer_vec.drain(..bytes_written); + } else { + response_fully_written = true; + } + } + } + } + + if connection_closed { + self.clear_write_buffer(); + return Err(ConnectionError::ConnectionClosed); + } else if response_fully_written { + self.response_buffer.take(); + } + + Ok(()) + } + + fn clear_write_buffer(&mut self) { + self.response_queue.clear(); + self.response_buffer.take(); + } + + /// Send a response back to the source of a request. + pub fn enqueue_response(&mut self, response: Response) { + self.response_queue.push_back(response); + } + + fn shift_buffer_left(&mut self, line_start_index: usize, end_cursor: usize) { + // We don't want to shift something that is already at the beginning. + if line_start_index != 0 { + // Move the bytes from `line_start_index` to the beginning of the buffer. + for cursor in 0..(end_cursor - line_start_index) { + self.buffer[cursor] = self.buffer[line_start_index + cursor]; + } + + // Clear the rest of the buffer. + for cursor in (end_cursor - line_start_index)..end_cursor { + self.buffer[cursor] = 0; + } + } + + // Update `read_cursor`. + self.read_cursor = end_cursor - line_start_index; + } + + /// Returns the first parsed request in the queue or `None` if the queue + /// is empty. + pub fn pop_parsed_request(&mut self) -> Option { + self.parsed_requests.pop_front() + } + + /// Returns `true` if there are bytes waiting to be written into the stream. + pub fn pending_write(&self) -> bool { + self.response_buffer.is_some() || !self.response_queue.is_empty() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use common::{Method, Version}; + use std::os::unix::net::UnixStream; + + #[test] + fn test_try_read_expect() { + // Test request with `Expect` header. + let (mut sender, receiver) = UnixStream::pair().unwrap(); + receiver.set_nonblocking(true).expect("Can't modify socket"); + let mut conn = HttpConnection::new(receiver); + sender + .write_all( + b"PATCH http://localhost/home HTTP/1.1\r\n\ + Expect: 100-continue\r\n\ + Content-Length: 26\r\n\ + Transfer-Encoding: chunked\r\n\r\n", + ) + .unwrap(); + assert!(conn.try_read().is_ok()); + + sender.write_all(b"this is not\n\r\na json \nbody").unwrap(); + conn.try_read().unwrap(); + let request = conn.pop_parsed_request().unwrap(); + + let expected_request = Request { + request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11), + headers: Headers::new(26, true, true), + body: Some(Body::new(b"this is not\n\r\na json \nbody".to_vec())), + }; + + assert_eq!(request, expected_request); + } + + #[test] + fn test_try_read_long_headers() { + // Long request headers. + let (mut sender, receiver) = UnixStream::pair().unwrap(); + receiver.set_nonblocking(true).expect("Can't modify socket"); + let mut conn = HttpConnection::new(receiver); + sender + .write_all( + b"PATCH http://localhost/home HTTP/1.1\r\n\ + Expect: 100-continue\r\n\ + Transfer-Encoding: chunked\r\n", + ) + .unwrap(); + + for i in 0..90 { + sender.write_all(b"Custom-Header-Testing: 1").unwrap(); + sender.write_all(i.to_string().as_bytes()).unwrap(); + sender.write_all(b"\r\n").unwrap(); + } + sender + .write_all(b"Content-Length: 26\r\n\r\nthis is not\n\r\na json \nbody") + .unwrap(); + assert!(conn.try_read().is_ok()); + assert!(conn.try_read().is_ok()); + assert!(conn.try_read().is_ok()); + let request = conn.pop_parsed_request().unwrap(); + + let expected_request = Request { + request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11), + headers: Headers::new(26, true, true), + body: Some(Body::new(b"this is not\n\r\na json \nbody".to_vec())), + }; + assert_eq!(request, expected_request); + } + + #[test] + fn test_try_read_split_ending() { + // Long request with '\r\n' on BUFFER_SIZEth and 1025th positions in the request. + let (mut sender, receiver) = UnixStream::pair().unwrap(); + receiver.set_nonblocking(true).expect("Can't modify socket"); + let mut conn = HttpConnection::new(receiver); + sender + .write_all( + b"PATCH http://localhost/home HTTP/1.1\r\n\ + Expect: 100-continue\r\n\ + Transfer-Encoding: chunked\r\n", + ) + .unwrap(); + + for i in 0..32 { + sender.write_all(b"Custom-Header-Testing: 1").unwrap(); + sender.write_all(i.to_string().as_bytes()).unwrap(); + sender.write_all(b"\r\n").unwrap(); + } + sender + .write_all(b"Head: aaaaa\r\nContent-Length: 26\r\n\r\nthis is not\n\r\na json \nbody") + .unwrap(); + assert!(conn.try_read().is_ok()); + conn.try_read().unwrap(); + let request = conn.pop_parsed_request().unwrap(); + let expected_request = Request { + request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11), + headers: Headers::new(26, true, true), + body: Some(Body::new(b"this is not\n\r\na json \nbody".to_vec())), + }; + assert_eq!(request, expected_request); + } + + #[test] + fn test_try_read_invalid_request() { + // Invalid request. + let (mut sender, receiver) = UnixStream::pair().unwrap(); + receiver.set_nonblocking(true).expect("Can't modify socket"); + let mut conn = HttpConnection::new(receiver); + sender + .write_all( + b"PATCH http://localhost/home HTTP/1.1\r\n\ + Expect: 100-continue\r\n\ + Transfer-Encoding: chunked\r\n", + ) + .unwrap(); + + for i in 0..40 { + sender.write_all(b"Custom-Header-Testing: 1").unwrap(); + sender.write_all(i.to_string().as_bytes()).unwrap(); + sender.write_all(b"\r\n").unwrap(); + } + sender + .write_all(b"Content-Length: alpha\r\n\r\nthis is not\n\r\na json \nbody") + .unwrap(); + assert!(conn.try_read().is_ok()); + let request_error = conn.try_read().unwrap_err(); + assert_eq!( + request_error, + ConnectionError::ParseError(RequestError::InvalidHeader) + ); + } + + #[test] + fn test_try_read_long_request_body() { + // Long request body. + let (mut sender, receiver) = UnixStream::pair().unwrap(); + receiver.set_nonblocking(true).expect("Can't modify socket"); + let mut conn = HttpConnection::new(receiver); + sender + .write_all( + b"PATCH http://localhost/home HTTP/1.1\r\n\ + Expect: 100-continue\r\n\ + Transfer-Encoding: chunked\r\n\ + Content-Length: 1400\r\n\r\n", + ) + .unwrap(); + + let mut request_body: Vec = Vec::with_capacity(1400); + for _ in 0..100 { + request_body.write_all(b"This is a test").unwrap(); + } + sender.write_all(request_body.as_slice()).unwrap(); + assert!(conn.try_read().is_ok()); + conn.try_read().unwrap(); + let request = conn.pop_parsed_request().unwrap(); + let expected_request = Request { + request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11), + headers: Headers::new(1400, true, true), + body: Some(Body::new(request_body)), + }; + assert_eq!(request, expected_request); + } + + #[test] + fn test_try_read_large_req_line() { + // Request line longer than BUFFER_SIZE bytes. + let (mut sender, receiver) = UnixStream::pair().unwrap(); + receiver.set_nonblocking(true).expect("Can't modify socket"); + let mut conn = HttpConnection::new(receiver); + sender.write_all(b"PATCH http://localhost/home").unwrap(); + + let mut request_body: Vec = Vec::with_capacity(1400); + for _ in 0..200 { + request_body.write_all(b"/home").unwrap(); + } + sender.write_all(request_body.as_slice()).unwrap(); + assert_eq!( + conn.try_read().unwrap_err(), + ConnectionError::ParseError(RequestError::InvalidRequest) + ); + } + + #[test] + fn test_try_read_large_header_line() { + // Header line longer than BUFFER_SIZE bytes. + let (mut sender, receiver) = UnixStream::pair().unwrap(); + receiver.set_nonblocking(true).expect("Can't modify socket"); + let mut conn = HttpConnection::new(receiver); + sender + .write_all(b"PATCH http://localhost/home HTTP/1.1\r\nhead: ") + .unwrap(); + + let mut request_body: Vec = Vec::with_capacity(1030); + for _ in 0..86 { + request_body.write_all(b"abcdefghijkl").unwrap(); + } + request_body.write_all(b"\r\n\r\n").unwrap(); + sender.write_all(request_body.as_slice()).unwrap(); + assert!(conn.try_read().is_ok()); + assert_eq!( + conn.try_read().unwrap_err(), + ConnectionError::ParseError(RequestError::InvalidHeader) + ); + } + + #[test] + fn test_try_read_no_body_request() { + // Request without body. + let (mut sender, receiver) = UnixStream::pair().unwrap(); + receiver.set_nonblocking(true).expect("Can't modify socket"); + let mut conn = HttpConnection::new(receiver); + sender + .write_all( + b"PATCH http://localhost/home HTTP/1.1\r\n\ + Expect: 100-continue\r\n\ + Transfer-Encoding: chunked\r\n\r\n", + ) + .unwrap(); + conn.try_read().unwrap(); + let request = conn.pop_parsed_request().unwrap(); + let expected_request = Request { + request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11), + headers: Headers::new(0, true, true), + body: None, + }; + assert_eq!(request, expected_request); + } + + #[test] + fn test_try_read_segmented_req_line() { + // Segmented request line. + let (mut sender, receiver) = UnixStream::pair().unwrap(); + receiver.set_nonblocking(true).expect("Can't modify socket"); + let mut conn = HttpConnection::new(receiver); + sender.write_all(b"PATCH http://local").unwrap(); + assert!(conn.try_read().is_ok()); + + sender.write_all(b"host/home HTTP/1.1\r\n\r\n").unwrap(); + + conn.try_read().unwrap(); + let request = conn.pop_parsed_request().unwrap(); + let expected_request = Request { + request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11), + headers: Headers::new(0, false, false), + body: None, + }; + assert_eq!(request, expected_request); + } + + #[test] + fn test_try_read_long_req_line_b2b() { + // Long request line after another request. + let (mut sender, receiver) = UnixStream::pair().unwrap(); + receiver.set_nonblocking(true).expect("Can't modify socket"); + let mut conn = HttpConnection::new(receiver); + // Req line 23 + 10*x + 13 = 36 + 10* x 984 free in first try read + sender + .write_all(b"PATCH http://localhost/home HTTP/1.1\r\n\r\nPATCH http://localhost/") + .unwrap(); + + let mut request_line: Vec = Vec::with_capacity(980); + for _ in 0..98 { + request_line.write_all(b"localhost/").unwrap(); + } + request_line.write_all(b" HTTP/1.1\r\n\r\n").unwrap(); + sender.write_all(request_line.as_slice()).unwrap(); + + conn.try_read().unwrap(); + let request = conn.pop_parsed_request().unwrap(); + let expected_request = Request { + request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11), + headers: Headers::new(0, false, false), + body: None, + }; + assert_eq!(request, expected_request); + + conn.try_read().unwrap(); + let request = conn.pop_parsed_request().unwrap(); + let mut expected_request_as_bytes = Vec::new(); + expected_request_as_bytes + .write_all(b"http://localhost/") + .unwrap(); + expected_request_as_bytes.append(request_line.as_mut()); + let expected_request = Request { + request_line: RequestLine::new( + Method::Patch, + std::str::from_utf8(&expected_request_as_bytes[..997]).unwrap(), + Version::Http11, + ), + headers: Headers::new(0, false, false), + body: None, + }; + assert_eq!(request, expected_request); + } + + #[test] + fn test_try_read_double_request() { + // Double request in a single read. + let (mut sender, receiver) = UnixStream::pair().unwrap(); + receiver.set_nonblocking(true).expect("Can't modify socket"); + let mut conn = HttpConnection::new(receiver); + sender + .write_all( + b"PATCH http://localhost/home HTTP/1.1\r\n\ + Transfer-Encoding: chunked\r\n\ + Content-Length: 26\r\n\r\nthis is not\n\r\na json \nbody", + ) + .unwrap(); + sender + .write_all( + b"PUT http://farhost/away HTTP/1.1\r\nContent-Length: 23\r\n\r\nthis is another request", + ) + .unwrap(); + + let expected_request_first = Request { + request_line: RequestLine::new(Method::Patch, "http://localhost/home", Version::Http11), + headers: Headers::new(26, false, true), + body: Some(Body::new(b"this is not\n\r\na json \nbody".to_vec())), + }; + + conn.try_read().unwrap(); + let request_first = conn.pop_parsed_request().unwrap(); + let request_second = conn.pop_parsed_request().unwrap(); + + let expected_request_second = Request { + request_line: RequestLine::new(Method::Put, "http://farhost/away", Version::Http11), + headers: Headers::new(23, false, false), + body: Some(Body::new(b"this is another request".to_vec())), + }; + assert_eq!(request_first, expected_request_first); + assert_eq!(request_second, expected_request_second); + } + + #[test] + fn test_try_read_connection_closed() { + // Connection abruptly closed. + let (mut sender, receiver) = UnixStream::pair().unwrap(); + receiver.set_nonblocking(true).expect("Can't modify socket"); + let mut conn = HttpConnection::new(receiver); + sender + .write_all( + b"PATCH http://localhost/home HTTP/1.1\r\n\ + Transfer-Encoding: chunked\r\n\ + Content-Len", + ) + .unwrap(); + + conn.try_read().unwrap(); + sender.shutdown(std::net::Shutdown::Both).unwrap(); + + assert_eq!( + conn.try_read().unwrap_err(), + ConnectionError::ConnectionClosed + ); + } + + #[test] + fn test_enqueue_response() { + // Response without body. + let (sender, mut receiver) = UnixStream::pair().unwrap(); + receiver.set_nonblocking(true).expect("Can't modify socket"); + let mut conn = HttpConnection::new(sender); + + let response = Response::new(Version::Http11, StatusCode::OK); + let mut expected_response: Vec = vec![]; + response.write_all(&mut expected_response).unwrap(); + + conn.enqueue_response(response); + assert!(conn.try_write().is_ok()); + + let mut response_buffer = vec![0u8; expected_response.len()]; + receiver.read_exact(&mut response_buffer).unwrap(); + assert_eq!(response_buffer, expected_response); + + // Response with body. + let (sender, mut receiver) = UnixStream::pair().unwrap(); + receiver.set_nonblocking(true).expect("Can't modify socket"); + let mut conn = HttpConnection::new(sender); + let mut response = Response::new(Version::Http11, StatusCode::OK); + let mut body: Vec = vec![]; + body.write_all(br#"{ "json": "body", "hello": "world" }"#) + .unwrap(); + response.set_body(Body::new(body)); + let mut expected_response: Vec = vec![]; + response.write_all(&mut expected_response).unwrap(); + + conn.enqueue_response(response); + assert!(conn.try_write().is_ok()); + + let mut response_buffer = vec![0u8; expected_response.len()]; + receiver.read_exact(&mut response_buffer).unwrap(); + assert_eq!(response_buffer, expected_response); + } +} diff --git a/src/lib.rs b/src/lib.rs new file mode 100644 index 0000000..7eab6f9 --- /dev/null +++ b/src/lib.rs @@ -0,0 +1,126 @@ +// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +#![deny(missing_docs)] +//! Minimal implementation of the [HTTP/1.0](https://tools.ietf.org/html/rfc1945) +//! and [HTTP/1.1](https://www.ietf.org/rfc/rfc2616.txt) protocols. +//! +//! HTTP/1.1 has a mandatory header **Host**, but as this crate is only used +//! for parsing API requests, this header (if present) is ignored. +//! +//! This HTTP implementation is stateless thus it does not support chunking or +//! compression. +//! +//! ## Supported Headers +//! The **micro_http** crate has support for parsing the following **Request** +//! headers: +//! - Content-Length +//! - Expect +//! - Transfer-Encoding +//! +//! The **Response** does not have a public interface for adding headers, but whenever +//! a write to the **Body** is made, the headers **ContentLength** and **MediaType** +//! are automatically updated. +//! +//! ### Media Types +//! The supported media types are: +//! - text/plain +//! - application/json +//! +//! ## Supported Methods +//! The supported HTTP Methods are: +//! - GET +//! - PUT +//! - PATCH +//! +//! ## Supported Status Codes +//! The supported status codes are: +//! +//! - Continue - 100 +//! - OK - 200 +//! - No Content - 204 +//! - Bad Request - 400 +//! - Not Found - 404 +//! - Internal Server Error - 500 +//! - Not Implemented - 501 +//! +//! ## Example for parsing an HTTP Request from a slice +//! ``` +//! extern crate micro_http; +//! use micro_http::{Request, Version}; +//! +//! let http_request = Request::try_from(b"GET http://localhost/home HTTP/1.0\r\n\r\n").unwrap(); +//! assert_eq!(http_request.http_version(), Version::Http10); +//! assert_eq!(http_request.uri().get_abs_path(), "/home"); +//! ``` +//! +//! ## Example for creating an HTTP Response +//! ``` +//! extern crate micro_http; +//! use micro_http::{Body, MediaType, Response, StatusCode, Version}; +//! +//! let mut response = Response::new(Version::Http10, StatusCode::OK); +//! let body = String::from("This is a test"); +//! response.set_body(Body::new(body.clone())); +//! response.set_content_type(MediaType::PlainText); +//! +//! assert!(response.status() == StatusCode::OK); +//! assert_eq!(response.body().unwrap(), Body::new(body)); +//! assert_eq!(response.http_version(), Version::Http10); +//! +//! let mut response_buf: [u8; 126] = [0; 126]; +//! assert!(response.write_all(&mut response_buf.as_mut()).is_ok()); +//! ``` +//! +//! `HttpConnection` can be used for automatic data exchange and parsing when +//! handling a client, but it only supports one stream. +//! +//! For handling multiple clients use `HttpServer`, which multiplexes `HttpConnection`s +//! and offers an easy to use interface. The server can run in either blocking or +//! non-blocking mode. Non-blocking is achieved by using `epoll` to make sure +//! `requests` will never block when called. +//! +//! ## Example for using the server +//! +//! ``` +//! extern crate micro_http; +//! use micro_http::{HttpServer, Response, StatusCode}; +//! +//! let path_to_socket = "/tmp/example.sock"; +//! std::fs::remove_file(path_to_socket).unwrap_or_default(); +//! +//! // Start the server. +//! let mut server = HttpServer::new(path_to_socket).unwrap(); +//! server.start_server().unwrap(); +//! +//! // Connect a client to the server so it doesn't block in our example. +//! let mut socket = std::os::unix::net::UnixStream::connect(path_to_socket).unwrap(); +//! +//! // Server loop processing requests. +//! loop { +//! for request in server.requests().unwrap() { +//! let response = request.process(|request| { +//! // Your code here. +//! Response::new(request.http_version(), StatusCode::NoContent) +//! }); +//! server.respond(response); +//! } +//! // Break this example loop. +//! break; +//! } +//! ``` + +mod common; +mod connection; +mod request; +mod response; +mod server; +use common::ascii; +use common::headers; + +pub use connection::{ConnectionError, HttpConnection}; +pub use request::{Request, RequestError}; +pub use response::{Response, StatusCode}; +pub use server::{HttpServer, ServerError, ServerRequest, ServerResponse}; + +pub use common::headers::{Headers, MediaType}; +pub use common::{Body, Method, Version}; diff --git a/src/request.rs b/src/request.rs new file mode 100644 index 0000000..e8c5e78 --- /dev/null +++ b/src/request.rs @@ -0,0 +1,470 @@ +// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +use std::str::from_utf8; + +use common::ascii::{CR, CRLF_LEN, LF, SP}; +pub use common::RequestError; +use common::{Body, Method, Version}; +use headers::Headers; + +/// Finds the first occurence of `sequence` in the `bytes` slice. +/// +/// Returns the starting position of the `sequence` in `bytes` or `None` if the +/// `sequence` is not found. +pub fn find(bytes: &[u8], sequence: &[u8]) -> Option { + bytes + .windows(sequence.len()) + .position(|window| window == sequence) +} + +/// Wrapper over HTTP URIs. +/// +/// The `Uri` can not be used directly and it is only accessible from an HTTP Request. +#[derive(Clone, Debug, PartialEq)] +pub struct Uri { + string: String, +} + +impl Uri { + fn new(slice: &str) -> Self { + Uri { + string: String::from(slice), + } + } + + fn try_from(bytes: &[u8]) -> Result { + if bytes.is_empty() { + return Err(RequestError::InvalidUri("Empty URI not allowed.")); + } + let utf8_slice = + from_utf8(bytes).map_err(|_| RequestError::InvalidUri("Cannot parse URI as UTF-8."))?; + Ok(Uri::new(utf8_slice)) + } + + /// Returns the absolute path of the `Uri`. + /// + /// URIs can be represented in absolute form or relative form. The absolute form includes + /// the HTTP scheme, followed by the absolute path as follows: + /// "http:" "//" host [ ":" port ] [ abs_path ] + /// The relative URIs can be one of net_path | abs_path | rel_path. + /// This method only handles absolute URIs and relative URIs specified by abs_path. + /// The abs_path is expected to start with '/'. + /// + /// # Errors + /// Returns an empty byte array when the host or the path are empty/invalid. + pub fn get_abs_path(&self) -> &str { + const HTTP_SCHEME_PREFIX: &str = "http://"; + + if self.string.starts_with(HTTP_SCHEME_PREFIX) { + let without_scheme = &self.string[HTTP_SCHEME_PREFIX.len()..]; + if without_scheme.is_empty() { + return ""; + } + // The host in this case includes the port and contains the bytes after http:// up to + // the next '/'. + match without_scheme.bytes().position(|byte| byte == b'/') { + Some(len) => &without_scheme[len..], + None => "", + } + } else { + if self.string.starts_with('/') { + return self.string.as_str(); + } + + "" + } + } +} + +/// Wrapper over an HTTP Request Line. +#[derive(Debug, PartialEq)] +pub struct RequestLine { + method: Method, + uri: Uri, + http_version: Version, +} + +impl RequestLine { + fn parse_request_line(request_line: &[u8]) -> (&[u8], &[u8], &[u8]) { + if let Some(method_end) = find(request_line, &[SP]) { + let method = &request_line[..method_end]; + + let uri_and_version = &request_line[(method_end + 1)..]; + + if let Some(uri_end) = find(uri_and_version, &[SP]) { + let uri = &uri_and_version[..uri_end]; + + let version = &uri_and_version[(uri_end + 1)..]; + + return (method, uri, version); + } + + return (method, uri_and_version, b""); + } + + (b"", b"", b"") + } + + /// Tries to parse a byte stream in a request line. Fails if the request line is malformed. + pub fn try_from(request_line: &[u8]) -> Result { + let (method, uri, version) = RequestLine::parse_request_line(request_line); + + Ok(RequestLine { + method: Method::try_from(method)?, + uri: Uri::try_from(uri)?, + http_version: Version::try_from(version)?, + }) + } + + // Returns the minimum length of a valid request. The request must contain + // the method (GET), the URI (minmum 1 character), the HTTP version(HTTP/DIGIT.DIGIT) and + // 2 separators (SP). + fn min_len() -> usize { + Method::Get.raw().len() + 1 + Version::Http10.raw().len() + 2 + } + + #[cfg(test)] + pub fn new(method: Method, uri: &str, http_version: Version) -> Self { + RequestLine { + method, + uri: Uri::new(uri), + http_version, + } + } +} + +/// Wrapper over an HTTP Request. +#[allow(unused)] +#[derive(Debug)] +pub struct Request { + /// The request line of the request. + pub request_line: RequestLine, + /// The headers of the request. + pub headers: Headers, + /// The body of the request. + pub body: Option, +} + +impl Request { + /// Parses a byte slice into a HTTP Request. + /// + /// The byte slice is expected to have the following format:
+ /// * Request Line: "GET SP Request-uri SP HTTP/1.0 CRLF" - Mandatory
+ /// * Request Headers " CRLF"- Optional
+ /// * Entity Body - Optional
+ /// The request headers and the entity body is not parsed and None is returned because + /// these are not used by the MMDS server. + /// The only supported method is GET and the HTTP protocol is expected to be HTTP/1.0 + /// or HTTP/1.1. + /// + /// # Errors + /// The function returns InvalidRequest when parsing the byte stream fails. + /// + /// # Examples + /// + /// ``` + /// extern crate micro_http; + /// use micro_http::Request; + /// + /// let http_request = Request::try_from(b"GET http://localhost/home HTTP/1.0\r\n"); + /// ``` + pub fn try_from(byte_stream: &[u8]) -> Result { + // The first line of the request is the Request Line. The line ending is CR LF. + let request_line_end = match find(byte_stream, &[CR, LF]) { + Some(len) => len, + // If no CR LF is found in the stream, the request format is invalid. + None => return Err(RequestError::InvalidRequest), + }; + + let request_line_bytes = &byte_stream[..request_line_end]; + if request_line_bytes.len() < RequestLine::min_len() { + return Err(RequestError::InvalidRequest); + } + + let request_line = RequestLine::try_from(request_line_bytes)?; + + // Find the next CR LF CR LF sequence in our buffer starting at the end on the Request + // Line, including the trailing CR LF previously found. + match find(&byte_stream[request_line_end..], &[CR, LF, CR, LF]) { + // If we have found a CR LF CR LF at the end of the Request Line, the request + // is complete. + Some(0) => Ok(Request { + request_line, + headers: Headers::default(), + body: None, + }), + Some(headers_end) => { + // Parse the request headers. + // Start by removing the leading CR LF from them. + let headers_and_body = &byte_stream[(request_line_end + CRLF_LEN)..]; + let headers_end = headers_end - CRLF_LEN; + let headers = Headers::try_from(&headers_and_body[..headers_end])?; + + // Parse the body of the request. + // Firstly check if we have a body. + let body = match headers.content_length() { + 0 => { + // No request body. + None + } + content_length => { + // Headers suggest we have a body, but the buffer is shorter than the specified + // content length. + if headers_and_body.len() - (headers_end + 2 * CRLF_LEN) + < content_length as usize + { + return Err(RequestError::InvalidRequest); + } + let body_as_bytes = &headers_and_body[(headers_end + 2 * CRLF_LEN)..]; + // If the actual length of the body is different than the `Content-Length` value + // in the headers then this request is invalid. + if body_as_bytes.len() == content_length as usize { + Some(Body::new(body_as_bytes)) + } else { + return Err(RequestError::InvalidRequest); + } + } + }; + + Ok(Request { + request_line, + headers, + body, + }) + } + // If we can't find a CR LF CR LF even though the request should have headers + // the request format is invalid. + None => Err(RequestError::InvalidRequest), + } + } + + /// Returns the `Uri` from the parsed `Request`. + /// + /// The return value can be used to get the absolute path of the URI. + pub fn uri(&self) -> &Uri { + &self.request_line.uri + } + + /// Returns the HTTP `Version` of the `Request`. + pub fn http_version(&self) -> Version { + self.request_line.http_version + } + + /// Returns the HTTP `Method` of the `Request`. + pub fn method(&self) -> Method { + self.request_line.method + } +} + +#[cfg(test)] +mod tests { + use super::*; + + impl PartialEq for Request { + fn eq(&self, other: &Request) -> bool { + // Ignore the other fields of Request for now because they are not used. + self.request_line == other.request_line + && self.headers.content_length() == other.headers.content_length() + && self.headers.expect() == other.headers.expect() + && self.headers.chunked() == other.headers.chunked() + } + } + + #[test] + fn test_uri() { + let uri = Uri::new("http://localhost/home"); + assert_eq!(uri.get_abs_path(), "/home"); + + let uri = Uri::new("/home"); + assert_eq!(uri.get_abs_path(), "/home"); + + let uri = Uri::new("home"); + assert_eq!(uri.get_abs_path(), ""); + + let uri = Uri::new("http://"); + assert_eq!(uri.get_abs_path(), ""); + + let uri = Uri::new("http://192.168.0.0"); + assert_eq!(uri.get_abs_path(), ""); + } + + #[test] + fn test_find() { + let bytes: &[u8; 13] = b"abcacrgbabsjl"; + let i = find(&bytes[..], b"ac"); + assert_eq!(i.unwrap(), 3); + + let i = find(&bytes[..], b"rgb"); + assert_eq!(i.unwrap(), 5); + + let i = find(&bytes[..], b"ab"); + assert_eq!(i.unwrap(), 0); + + let i = find(&bytes[..], b"l"); + assert_eq!(i.unwrap(), 12); + + let i = find(&bytes[..], b"jle"); + assert!(i.is_none()); + + let i = find(&bytes[..], b"asdkjhasjhdjhgsadg"); + assert!(i.is_none()); + + let i = find(&bytes[..], b"abcacrgbabsjl"); + assert_eq!(i.unwrap(), 0); + } + + #[test] + fn test_into_request_line() { + let expected_request_line = RequestLine { + http_version: Version::Http10, + method: Method::Get, + uri: Uri::new("http://localhost/home"), + }; + + let request_line = b"GET http://localhost/home HTTP/1.0"; + assert_eq!( + RequestLine::try_from(request_line).unwrap(), + expected_request_line + ); + + let expected_request_line = RequestLine { + http_version: Version::Http11, + method: Method::Get, + uri: Uri::new("http://localhost/home"), + }; + + // Happy case with request line ending in CRLF. + let request_line = b"GET http://localhost/home HTTP/1.1"; + assert_eq!( + RequestLine::try_from(request_line).unwrap(), + expected_request_line + ); + + // Happy case with request line ending in LF instead of CRLF. + let request_line = b"GET http://localhost/home HTTP/1.1"; + assert_eq!( + RequestLine::try_from(request_line).unwrap(), + expected_request_line + ); + + // Test for invalid method. + let request_line = b"POST http://localhost/home HTTP/1.0"; + assert_eq!( + RequestLine::try_from(request_line).unwrap_err(), + RequestError::InvalidHttpMethod("Unsupported HTTP method.") + ); + + // Test for invalid uri. + let request_line = b"GET HTTP/1.0"; + assert_eq!( + RequestLine::try_from(request_line).unwrap_err(), + RequestError::InvalidUri("Empty URI not allowed.") + ); + + // Test for invalid HTTP version. + let request_line = b"GET http://localhost/home HTTP/2.0"; + assert_eq!( + RequestLine::try_from(request_line).unwrap_err(), + RequestError::InvalidHttpVersion("Unsupported HTTP version.") + ); + + // Test for invalid format with no method, uri or version. + let request_line = b"nothing"; + assert_eq!( + RequestLine::try_from(request_line).unwrap_err(), + RequestError::InvalidHttpMethod("Unsupported HTTP method.") + ); + + // Test for invalid format with no version. + let request_line = b"GET /"; + assert_eq!( + RequestLine::try_from(request_line).unwrap_err(), + RequestError::InvalidHttpVersion("Unsupported HTTP version.") + ); + } + + #[test] + fn test_into_request() { + let expected_request = Request { + request_line: RequestLine { + http_version: Version::Http10, + method: Method::Get, + uri: Uri::new("http://localhost/home"), + }, + body: None, + headers: Headers::default(), + }; + let request_bytes = b"GET http://localhost/home HTTP/1.0\r\n\ + Last-Modified: Tue, 15 Nov 1994 12:45:26 GMT\r\n\r\n"; + let request = Request::try_from(request_bytes).unwrap(); + assert_eq!(request, expected_request); + assert_eq!(request.uri(), &Uri::new("http://localhost/home")); + assert_eq!(request.http_version(), Version::Http10); + assert!(request.body.is_none()); + + // Test for invalid Request (length is less than minimum). + let request_bytes = b"GET"; + assert_eq!( + Request::try_from(request_bytes).unwrap_err(), + RequestError::InvalidRequest + ); + + // Test for a request with the headers we are looking for. + let request = Request::try_from( + b"PATCH http://localhost/home HTTP/1.1\r\n\ + Expect: 100-continue\r\n\ + Transfer-Encoding: chunked\r\n\ + Content-Length: 26\r\n\r\nthis is not\n\r\na json \nbody", + ) + .unwrap(); + assert_eq!(request.uri(), &Uri::new("http://localhost/home")); + assert_eq!(request.http_version(), Version::Http11); + assert_eq!(request.method(), Method::Patch); + assert_eq!(request.headers.chunked(), true); + assert_eq!(request.headers.expect(), true); + assert_eq!(request.headers.content_length(), 26); + assert_eq!( + request.body.unwrap().body, + String::from("this is not\n\r\na json \nbody") + .as_bytes() + .to_vec() + ); + + // Test for an invalid request format. + Request::try_from(b"PATCH http://localhost/home HTTP/1.1\r\n").unwrap_err(); + + // Test for an invalid encoding. + let request = Request::try_from( + b"PATCH http://localhost/home HTTP/1.1\r\n\ + Expect: 100-continue\r\n\ + Transfer-Encoding: identity; q=0\r\n\ + Content-Length: 26\r\n\r\nthis is not\n\r\na json \nbody", + ) + .unwrap_err(); + assert_eq!(request, RequestError::InvalidHeader); + + // Test for an invalid content length. + let request = Request::try_from( + b"PATCH http://localhost/home HTTP/1.1\r\n\ + Expect: 100-continue\r\n\ + Content-Length: 5000\r\n\r\nthis is a short body", + ) + .unwrap_err(); + assert_eq!(request, RequestError::InvalidRequest); + + // Test for a request without a body and an optional header. + let request = Request::try_from( + b"GET http://localhost/ HTTP/1.0\r\n\ + Accept-Encoding: gzip\r\n\r\n", + ) + .unwrap(); + assert_eq!(request.uri(), &Uri::new("http://localhost/")); + assert_eq!(request.http_version(), Version::Http10); + assert_eq!(request.method(), Method::Get); + assert_eq!(request.headers.chunked(), false); + assert_eq!(request.headers.expect(), false); + assert_eq!(request.headers.content_length(), 0); + assert!(request.body.is_none()); + } +} diff --git a/src/response.rs b/src/response.rs new file mode 100644 index 0000000..bd9ac37 --- /dev/null +++ b/src/response.rs @@ -0,0 +1,291 @@ +// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +use std::io::{Error as WriteError, Write}; + +use ascii::{COLON, CR, LF, SP}; +use common::{Body, Version}; +use headers::{Header, MediaType}; + +/// Wrapper over a response status code. +/// +/// The status code is defined as specified in the +/// [RFC](https://tools.ietf.org/html/rfc7231#section-6). +#[allow(dead_code)] +#[derive(Clone, Copy, Debug, PartialEq)] +pub enum StatusCode { + /// 100, Continue + Continue, + /// 200, OK + OK, + /// 204, No Content + NoContent, + /// 400, Bad Request + BadRequest, + /// 404, Not Found + NotFound, + /// 500, Internal Server Error + InternalServerError, + /// 501, Not Implemented + NotImplemented, +} + +impl StatusCode { + /// Returns the status code as bytes. + pub fn raw(self) -> &'static [u8; 3] { + match self { + StatusCode::Continue => b"100", + StatusCode::OK => b"200", + StatusCode::NoContent => b"204", + StatusCode::BadRequest => b"400", + StatusCode::NotFound => b"404", + StatusCode::InternalServerError => b"500", + StatusCode::NotImplemented => b"501", + } + } +} + +struct StatusLine { + http_version: Version, + status_code: StatusCode, +} + +impl StatusLine { + fn new(http_version: Version, status_code: StatusCode) -> Self { + StatusLine { + http_version, + status_code, + } + } + + fn write_all(&self, mut buf: T) -> Result<(), WriteError> { + buf.write_all(self.http_version.raw())?; + buf.write_all(&[SP])?; + buf.write_all(self.status_code.raw())?; + buf.write_all(&[SP, CR, LF])?; + + Ok(()) + } +} + +/// Wrapper over the list of headers associated with a HTTP Response. +/// When creating a ResponseHeaders object, the content type is initialized to `text/plain`. +/// The content type can be updated with a call to `set_content_type`. +pub struct ResponseHeaders { + content_length: i32, + content_type: MediaType, + server: String, +} + +impl Default for ResponseHeaders { + fn default() -> Self { + ResponseHeaders { + content_length: Default::default(), + content_type: Default::default(), + server: "Firecracker API".to_string(), + } + } +} + +impl ResponseHeaders { + /// Writes the headers to `buf` using the HTTP specification. + pub fn write_all(&self, buf: &mut T) -> Result<(), WriteError> { + buf.write_all(Header::Server.raw())?; + buf.write_all(&[COLON, SP])?; + buf.write_all(self.server.as_bytes())?; + + buf.write_all(&[CR, LF])?; + buf.write_all(b"Connection: keep-alive")?; + buf.write_all(&[CR, LF])?; + + if self.content_length != 0 { + buf.write_all(Header::ContentType.raw())?; + buf.write_all(&[COLON, SP])?; + buf.write_all(self.content_type.as_str().as_bytes())?; + buf.write_all(&[CR, LF])?; + + buf.write_all(Header::ContentLength.raw())?; + buf.write_all(&[COLON, SP])?; + buf.write_all(self.content_length.to_string().as_bytes())?; + buf.write_all(&[CR, LF])?; + } + + buf.write_all(&[CR, LF]) + } + + // Sets the content length to be written in the HTTP response. + fn set_content_length(&mut self, content_length: i32) { + self.content_length = content_length; + } + + /// Sets the HTTP response header server. + pub fn set_server(&mut self, server: &str) { + self.server = String::from(server); + } + + /// Sets the content type to be written in the HTTP response. + #[allow(unused)] + pub fn set_content_type(&mut self, content_type: MediaType) { + self.content_type = content_type; + } +} + +/// Wrapper over an HTTP Response. +/// +/// The Response is created using a `Version` and a `StatusCode`. When creating a Response object, +/// the body is initialized to `None`. The body can be updated with a call to `set_body`. +pub struct Response { + status_line: StatusLine, + headers: ResponseHeaders, + body: Option, +} + +impl Response { + /// Creates a new HTTP `Response` with an empty body. + pub fn new(http_version: Version, status_code: StatusCode) -> Response { + Response { + status_line: StatusLine::new(http_version, status_code), + headers: ResponseHeaders::default(), + body: None, + } + } + + /// Updates the body of the `Response`. + /// + /// This function has side effects because it also updates the headers: + /// - `ContentLength`: this is set to the length of the specified body. + pub fn set_body(&mut self, body: Body) { + self.headers.set_content_length(body.len() as i32); + self.body = Some(body); + } + + /// Updates the content type of the `Response`. + pub fn set_content_type(&mut self, content_type: MediaType) { + self.headers.set_content_type(content_type); + } + + /// Sets the HTTP response server. + pub fn set_server(&mut self, server: &str) { + self.headers.set_server(server); + } + + fn write_body(&self, mut buf: T) -> Result<(), WriteError> { + if let Some(ref body) = self.body { + buf.write_all(body.raw())?; + } + Ok(()) + } + + /// Writes the content of the `Response` to the specified `buf`. + /// + /// # Errors + /// Returns an error when the buffer is not large enough. + pub fn write_all(&self, mut buf: &mut T) -> Result<(), WriteError> { + self.status_line.write_all(&mut buf)?; + self.headers.write_all(&mut buf)?; + self.write_body(&mut buf)?; + + Ok(()) + } + + /// Returns the Status Code of the Response. + pub fn status(&self) -> StatusCode { + self.status_line.status_code + } + + /// Returns the Body of the response. If the response does not have a body, + /// it returns None. + pub fn body(&self) -> Option { + self.body.clone() + } + + /// Returns the HTTP Version of the response. + pub fn content_length(&self) -> i32 { + self.headers.content_length + } + + /// Returns the HTTP Version of the response. + pub fn content_type(&self) -> MediaType { + self.headers.content_type + } + + /// Returns the HTTP Version of the response. + pub fn http_version(&self) -> Version { + self.status_line.http_version + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_write_response() { + let mut response = Response::new(Version::Http10, StatusCode::OK); + let body = "This is a test"; + response.set_body(Body::new(body)); + response.set_content_type(MediaType::PlainText); + + assert!(response.status() == StatusCode::OK); + assert_eq!(response.body().unwrap(), Body::new(body)); + assert_eq!(response.http_version(), Version::Http10); + assert_eq!(response.content_length(), 14); + assert_eq!(response.content_type(), MediaType::PlainText); + + let expected_response: &'static [u8] = b"HTTP/1.0 200 \r\n\ + Server: Firecracker API\r\n\ + Connection: keep-alive\r\n\ + Content-Type: text/plain\r\n\ + Content-Length: 14\r\n\r\n\ + This is a test"; + + let mut response_buf: [u8; 126] = [0; 126]; + assert!(response.write_all(&mut response_buf.as_mut()).is_ok()); + assert!(response_buf.as_ref() == expected_response); + + // Test write failed. + let mut response_buf: [u8; 1] = [0; 1]; + assert!(response.write_all(&mut response_buf.as_mut()).is_err()); + } + + #[test] + fn test_set_server() { + let mut response = Response::new(Version::Http10, StatusCode::OK); + let body = "This is a test"; + let server = "rust-vmm API"; + response.set_body(Body::new(body)); + response.set_content_type(MediaType::PlainText); + response.set_server(server); + + assert!(response.status() == StatusCode::OK); + assert_eq!(response.body().unwrap(), Body::new(body)); + assert_eq!(response.http_version(), Version::Http10); + assert_eq!(response.content_length(), 14); + assert_eq!(response.content_type(), MediaType::PlainText); + + let expected_response = format!( + "HTTP/1.0 200 \r\n\ + Server: {}\r\n\ + Connection: keep-alive\r\n\ + Content-Type: text/plain\r\n\ + Content-Length: 14\r\n\r\n\ + This is a test", + server + ); + + let mut response_buf: [u8; 123] = [0; 123]; + assert!(response.write_all(&mut response_buf.as_mut()).is_ok()); + assert!(response_buf.as_ref() == expected_response.as_bytes()); + } + + #[test] + fn test_status_code() { + assert_eq!(StatusCode::Continue.raw(), b"100"); + assert_eq!(StatusCode::OK.raw(), b"200"); + assert_eq!(StatusCode::NoContent.raw(), b"204"); + assert_eq!(StatusCode::BadRequest.raw(), b"400"); + assert_eq!(StatusCode::NotFound.raw(), b"404"); + assert_eq!(StatusCode::InternalServerError.raw(), b"500"); + assert_eq!(StatusCode::NotImplemented.raw(), b"501"); + } +} diff --git a/src/server.rs b/src/server.rs new file mode 100644 index 0000000..12a99ec --- /dev/null +++ b/src/server.rs @@ -0,0 +1,834 @@ +// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 +extern crate epoll; + +use std::io::{Read, Write}; +use std::os::unix::io::AsRawFd; +use std::os::unix::io::RawFd; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::Path; + +use common::{Body, Version}; +pub use common::{ConnectionError, RequestError, ServerError}; +use connection::HttpConnection; +use request::Request; +use response::{Response, StatusCode}; +use std::collections::HashMap; + +static SERVER_FULL_ERROR_MESSAGE: &[u8] = b"HTTP/1.1 503\r\n\ + Server: Firecracker API\r\n\ + Connection: close\r\n\ + Content-Length: 40\r\n\r\n{ \"error\": \"Too many open connections\" }"; +const MAX_CONNECTIONS: usize = 10; + +type Result = std::result::Result; + +/// Wrapper over `Request` which adds an identification token. +pub struct ServerRequest { + /// Inner request. + pub request: Request, + /// Identification token. + id: u64, +} + +impl ServerRequest { + /// Creates a new `ServerRequest` object from an existing `Request`, + /// adding an identification token. + pub fn new(request: Request, id: u64) -> Self { + ServerRequest { request, id } + } + + /// Returns a reference to the inner request. + pub fn inner(&self) -> &Request { + &self.request + } + + /// Calls the function provided on the inner request to obtain the response. + /// The response is then wrapped in a `ServerResponse`. + /// + /// Returns a `ServerResponse` ready for yielding to the server + pub fn process(&self, callable: F) -> ServerResponse + where + F: Fn(&Request) -> Response, + { + let http_response = callable(self.inner()); + ServerResponse::new(http_response, self.id) + } +} + +/// Wrapper over `Response` which adds an identification token. +pub struct ServerResponse { + /// Inner response. + response: Response, + /// Identification token. + id: u64, +} + +impl ServerResponse { + fn new(response: Response, id: u64) -> ServerResponse { + ServerResponse { response, id } + } +} + +/// Describes the state of the connection as far as data exchange +/// on the stream is concerned. +#[derive(PartialOrd, PartialEq)] +enum ClientConnectionState { + AwaitingIncoming, + AwaitingOutgoing, + Closed, +} + +/// Wrapper over `HttpConnection` which keeps track of yielded +/// requests and absorbed responses. +struct ClientConnection { + /// The `HttpConnection` object which handles data exchange. + connection: HttpConnection, + /// The state of the connection in the `epoll` structure. + state: ClientConnectionState, + /// Represents the difference between yielded requests and + /// absorbed responses. + /// This has to be `0` if we want to drop the connection. + in_flight_response_count: u32, +} + +impl ClientConnection { + fn new(connection: HttpConnection) -> Self { + ClientConnection { + connection, + state: ClientConnectionState::AwaitingIncoming, + in_flight_response_count: 0, + } + } + + fn read(&mut self) -> Result> { + // Data came into the connection. + let mut parsed_requests = vec![]; + match self.connection.try_read() { + Err(ConnectionError::ConnectionClosed) => { + // Connection timeout. + self.state = ClientConnectionState::Closed; + // We don't want to propagate this to the server and we will + // return no requests and wait for the connection to become + // safe to drop. + return Ok(vec![]); + } + Err(ConnectionError::StreamError(inner)) => { + // Reading from the connection failed. + // We should try to write an error message regardless. + let mut internal_error_response = + Response::new(Version::Http11, StatusCode::InternalServerError); + internal_error_response.set_body(Body::new(inner.to_string())); + self.connection.enqueue_response(internal_error_response); + } + Err(ConnectionError::ParseError(inner)) => { + // An error occurred while parsing the read bytes. + // Check if there are any valid parsed requests in the queue. + while let Some(_discarded_request) = self.connection.pop_parsed_request() {} + + // Send an error response for the request that gave us the error. + let mut error_response = Response::new(Version::Http11, StatusCode::BadRequest); + error_response.set_body(Body::new( + format!( + "{{ \"error\": \"{}\nAll previous unanswered requests will be dropped.\" }}", + inner.to_string() + ) + .to_string(), + )); + self.connection.enqueue_response(error_response); + } + Err(ConnectionError::InvalidWrite) => { + // This is unreachable because `HttpConnection::try_read()` cannot return this error variant. + unreachable!(); + } + Ok(()) => { + while let Some(request) = self.connection.pop_parsed_request() { + // Add all valid requests to `parsed_requests`. + parsed_requests.push(request); + } + } + } + self.in_flight_response_count += parsed_requests.len() as u32; + // If the state of the connection has changed, we need to update + // the event set in the `epoll` structure. + if self.connection.pending_write() { + self.state = ClientConnectionState::AwaitingOutgoing; + } + + Ok(parsed_requests) + } + + fn write(&mut self) -> Result<()> { + // The stream is available for writing. + match self.connection.try_write() { + Err(ConnectionError::ConnectionClosed) | Err(ConnectionError::StreamError(_)) => { + // Writing to the stream failed so it will be removed. + self.state = ClientConnectionState::Closed; + } + Err(ConnectionError::InvalidWrite) => { + // A `try_write` call was performed on a connection that has nothing + // to write. + return Err(ServerError::ConnectionError(ConnectionError::InvalidWrite)); + } + _ => { + // Check if we still have bytes to write for this connection. + if !self.connection.pending_write() { + self.state = ClientConnectionState::AwaitingIncoming; + } + } + } + Ok(()) + } + + fn enqueue_response(&mut self, response: Response) { + if self.state != ClientConnectionState::Closed { + self.connection.enqueue_response(response); + } + self.in_flight_response_count -= 1; + } + + // Returns `true` if the connection is closed and safe to drop. + fn is_done(&self) -> bool { + self.state == ClientConnectionState::Closed + && !self.connection.pending_write() + && self.in_flight_response_count == 0 + } +} + +/// HTTP Server implementation using Unix Domain Sockets and `EPOLL` to +/// handle multiple connections on the same thread. +/// +/// The function that does the data exchange is `handle_notifications`. +/// It can be called in a loop, which will render the thread that the +/// server runs on incapable of performing other operations, or it can +/// be used in another `EPOLL` structure, as it provides its `epoll_fd`, +/// the file descriptor of the epoll structure used within the server, +/// and it can be added to another one using the `EPOLLIN` flag. Whenever +/// there is a notification on that fd, `handle_notifications` should be +/// called once. +/// +/// # Example +/// +/// ## Starting and running the server +/// +/// ``` +/// use micro_http::{HttpServer, Response, StatusCode}; +/// +/// let path_to_socket = "/tmp/example.sock"; +/// std::fs::remove_file(path_to_socket).unwrap_or_default(); +/// +/// // Start the server. +/// let mut server = HttpServer::new(path_to_socket).unwrap(); +/// server.start_server().unwrap(); +/// +/// // Connect a client to the server so it doesn't block in our example. +/// let mut socket = std::os::unix::net::UnixStream::connect(path_to_socket).unwrap(); +/// +/// // Server loop processing requests. +/// loop { +/// for request in server.requests().unwrap() { +/// let response = request.process(|request| { +/// // Your code here. +/// Response::new(request.http_version(), StatusCode::NoContent) +/// }); +/// server.respond(response); +/// } +/// // Break this example loop. +/// break; +/// } +/// ``` +pub struct HttpServer { + /// Socket on which we listen for new connections. + socket: UnixListener, + /// File descriptor of the server's epoll structure. + epoll_fd: RawFd, + /// Holds the token-connection pairs of the server. + /// Each connection has an associated identification token, which is + /// the file descriptor of the underlying stream. + /// We use the file descriptor of the stream as the key for mapping + /// connections because the 1-to-1 relation is guaranteed by the OS. + connections: HashMap>, +} + +impl HttpServer { + /// Constructor for `HttpServer`. + /// + /// Returns the newly formed `HttpServer`. + /// + /// # Errors + /// Returns an `IOError` when binding or `epoll::create` fails. + pub fn new>(path_to_socket: P) -> Result { + let socket = UnixListener::bind(path_to_socket).map_err(ServerError::IOError)?; + let epoll_fd = epoll::create(true).map_err(ServerError::IOError)?; + Ok(HttpServer { + socket, + epoll_fd, + connections: HashMap::new(), + }) + } + + /// Starts the HTTP Server. + pub fn start_server(&mut self) -> Result<()> { + // Add the socket on which we listen for new connections to the + // `epoll` structure. + Self::epoll_add(self.epoll_fd, self.socket.as_raw_fd()) + } + + /// This function is responsible for the data exchange with the clients and should + /// be called when we are either notified through `epoll` that we need to exchange + /// data with at least a client or when we don't need to perform any other operations + /// on this thread and we can afford to call it in a loop. + /// + /// Note that this function will block the current thread if there are no notifications + /// to be handled by the server. + /// + /// Returns a collection of complete and valid requests to be processed by the user + /// of the server. Once processed, responses should be sent using `enqueue_responses()`. + /// + /// # Errors + /// `IOError` is returned when `read`, `write` or `epoll::ctl` operations fail. + /// `ServerFull` is returned when a client is trying to connect to the server, but + /// full capacity has already been reached. + /// `InvalidWrite` is returned when the server attempted to perform a write operation + /// on a connection on which it is not possible. + pub fn requests(&mut self) -> Result> { + let mut parsed_requests: Vec = vec![]; + let mut events = vec![epoll::Event::new(epoll::Events::empty(), 0); MAX_CONNECTIONS]; + + // This is a wrapper over the syscall `epoll_wait` and it will block the + // current thread until at least one event is received. + // The received notifications will then populate the `events` array with + // `event_count` elements, where 1 <= event_count <= MAX_CONNECTIONS. + let event_count = + epoll::wait(self.epoll_fd, -1, &mut events[..]).map_err(ServerError::IOError)?; + // We use `take()` on the iterator over `events` as, even though only + // `events_count` events have been inserted into `events`, the size of + // the array is still `MAX_CONNECTIONS`, so we discard empty elements + // at the end of the array. + for e in events.iter().take(event_count) { + // Check the file descriptor which produced the notification `e`. + // It could be that we have a new connection, or one of our open + // connections is ready to exchange data with a client. + if e.data as RawFd == self.socket.as_raw_fd() { + // We have received a notification on the listener socket, which + // means we have a new connection to accept. + match self.handle_new_connection() { + // If the server is full, we send a message to the client + // notifying them that we will close the connection, then + // we discard it. + Err(ServerError::ServerFull) => { + self.socket + .accept() + .map_err(ServerError::IOError) + .and_then(move |(mut stream, _)| { + stream + .write(SERVER_FULL_ERROR_MESSAGE) + .map_err(ServerError::IOError) + })?; + } + // An internal error will compromise any in-flight requests. + Err(error) => return Err(error), + Ok(()) => {} + }; + } else { + // We have a notification on one of our open connections. + let fd = e.data as RawFd; + let client_connection = self.connections.get_mut(&fd).unwrap(); + if e.events & epoll::Events::EPOLLIN.bits() != 0 { + // We have bytes to read from this connection. + // If our `read` yields `Request` objects, we wrap them with an ID before + // handing them to the user. + parsed_requests.append( + &mut client_connection + .read()? + .into_iter() + .map(|request| ServerRequest::new(request, fd as u64)) + .collect(), + ); + // If the connection was incoming before we read and we now have to write + // either an error message or an `expect` response, we change its `epoll` + // event set to notify us when the stream is ready for writing. + if client_connection.state == ClientConnectionState::AwaitingOutgoing { + Self::epoll_mod(self.epoll_fd, fd, epoll::Events::EPOLLOUT)?; + } + } else if e.events & epoll::Events::EPOLLOUT.bits() != 0 { + // We have bytes to write on this connection. + client_connection.write()?; + // If the connection was outgoing before we tried to write the responses + // and we don't have any more responses to write, we change the `epoll` + // event set to notify us when we have bytes to read from the stream. + if client_connection.state == ClientConnectionState::AwaitingIncoming { + Self::epoll_mod(self.epoll_fd, fd, epoll::Events::EPOLLIN)?; + } + } + } + } + + // Remove dead connections. + self.connections + .retain(|_, client_connection| !client_connection.is_done()); + + Ok(parsed_requests) + } + + /// The file descriptor of the `epoll` structure can enable the server to become + /// a non-blocking structure in an application. + /// + /// Returns the file descriptor of the server's internal `epoll` structure. + /// + /// # Example + /// + /// ## Non-blocking server + /// ``` + /// extern crate epoll; + /// + /// use micro_http::{HttpServer, Response, StatusCode}; + /// + /// // Create our epoll manager. + /// let epoll_fd = epoll::create(true).unwrap(); + /// + /// let path_to_socket = "/tmp/epoll_example.sock"; + /// std::fs::remove_file(path_to_socket).unwrap_or_default(); + /// + /// // Start the server. + /// let mut server = HttpServer::new(path_to_socket).unwrap(); + /// server.start_server().unwrap(); + /// + /// // Add our server to the `epoll` manager. + /// epoll::ctl( + /// epoll_fd, + /// epoll::ControlOptions::EPOLL_CTL_ADD, + /// server.epoll_fd(), + /// epoll::Event::new(epoll::Events::EPOLLIN, 1234u64), + /// ) + /// .unwrap(); + /// + /// // Connect a client to the server so it doesn't block in our example. + /// let mut socket = std::os::unix::net::UnixStream::connect(path_to_socket).unwrap(); + /// + /// // Control loop of the application. + /// let mut events = Vec::with_capacity(10); + /// loop { + /// let num_ev = epoll::wait(epoll_fd, -1, events.as_mut_slice()); + /// for event in events { + /// match event.data { + /// // The server notification. + /// 1234 => { + /// let request = server.requests(); + /// // Process... + /// } + /// // Other `epoll` notifications. + /// _ => { + /// // Do other computation. + /// } + /// } + /// } + /// // Break this example loop. + /// break; + /// } + /// ``` + pub fn epoll_fd(&self) -> RawFd { + self.epoll_fd + } + + /// Enqueues the provided responses in the outgoing connection. + /// + /// # Errors + /// `IOError` is returned when an `epoll::ctl` operation fails. + pub fn enqueue_responses(&mut self, responses: Vec) -> Result<()> { + for response in responses { + self.respond(response)?; + } + + Ok(()) + } + + /// Adds the provided response to the outgoing buffer in the corresponding connection. + /// + /// # Errors + /// `IOError` is returned when an `epoll::ctl` operation fails. + pub fn respond(&mut self, response: ServerResponse) -> Result<()> { + if let Some(client_connection) = self.connections.get_mut(&(response.id as i32)) { + // If the connection was incoming before we enqueue the response, we change its + // `epoll` event set to notify us when the stream is ready for writing. + if let ClientConnectionState::AwaitingIncoming = client_connection.state { + client_connection.state = ClientConnectionState::AwaitingOutgoing; + Self::epoll_mod(self.epoll_fd, response.id as RawFd, epoll::Events::EPOLLOUT)?; + } + client_connection.enqueue_response(response.response); + } + Ok(()) + } + + /// Accepts a new incoming connection and adds it to the `epoll` notification structure. + /// + /// # Errors + /// `IOError` is returned when an `epoll::ctl` operation fails. + fn handle_new_connection(&mut self) -> Result<()> { + if self.connections.len() == MAX_CONNECTIONS { + // If we want a replacement policy for connections + // this is where we will have it. + return Err(ServerError::ServerFull); + } + + self.socket + .accept() + .map_err(ServerError::IOError) + .and_then(|(stream, _)| { + // `HttpConnection` is supposed to work with non-blocking streams. + stream + .set_nonblocking(true) + .map(|_| stream) + .map_err(ServerError::IOError) + }) + .and_then(|stream| { + // Add the stream to the `epoll` structure and listen for bytes to be read. + Self::epoll_add(self.epoll_fd, stream.as_raw_fd())?; + // Then add it to our open connections. + self.connections.insert( + stream.as_raw_fd(), + ClientConnection::new(HttpConnection::new(stream)), + ); + Ok(()) + }) + } + + /// Changes the event type for a connection to either listen for incoming bytes + /// or for when the stream is ready for writing. + fn epoll_mod(epoll_fd: RawFd, stream_fd: RawFd, evset: epoll::Events) -> Result<()> { + let event = epoll::Event::new(evset, stream_fd as u64); + epoll::ctl( + epoll_fd, + epoll::ControlOptions::EPOLL_CTL_MOD, + stream_fd, + event, + ) + .map_err(ServerError::IOError) + } + + /// Adds a stream to the `epoll` notification structure with the `EPOLLIN` event set. + fn epoll_add(epoll_fd: RawFd, stream_fd: RawFd) -> Result<()> { + epoll::ctl( + epoll_fd, + epoll::ControlOptions::EPOLL_CTL_ADD, + stream_fd, + epoll::Event::new(epoll::Events::EPOLLIN, stream_fd as u64), + ) + .map_err(ServerError::IOError) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::fs; + use std::os::unix::net::UnixStream; + + use common::Body; + use std::io::Read; + use std::io::Write; + + #[test] + fn test_wait_one_connection() { + let path_to_socket = "/tmp/test_socket_http_server1.sock"; + fs::remove_file(path_to_socket).unwrap_or_default(); + + let mut server = HttpServer::new(path_to_socket.to_string()).unwrap(); + server.start_server().unwrap(); + + // Test one incoming connection. + let mut socket = UnixStream::connect(path_to_socket).unwrap(); + assert!(server.requests().unwrap().is_empty()); + + socket + .write_all( + b"PATCH /machine-config HTTP/1.1\r\n\ + Content-Length: 13\r\n\ + Content-Type: application/json\r\n\r\nwhatever body", + ) + .unwrap(); + + let mut req_vec = server.requests().unwrap(); + let server_request = req_vec.remove(0); + + server + .respond(server_request.process(|_request| { + let mut response = Response::new(Version::Http11, StatusCode::OK); + let response_body = b"response body"; + response.set_body(Body::new(response_body.to_vec())); + response + })) + .unwrap(); + assert!(server.requests().unwrap().is_empty()); + + let mut buf: [u8; 1024] = [0; 1024]; + assert!(socket.read(&mut buf[..]).unwrap() > 0); + fs::remove_file(path_to_socket).unwrap(); + } + + #[test] + fn test_wait_concurrent_connections() { + let path_to_socket = "/tmp/test_socket_http_server2.sock"; + fs::remove_file(path_to_socket).unwrap_or_default(); + + let mut server = HttpServer::new(path_to_socket.to_string()).unwrap(); + server.start_server().unwrap(); + + // Test two concurrent connections. + let mut first_socket = UnixStream::connect(path_to_socket).unwrap(); + assert!(server.requests().unwrap().is_empty()); + + first_socket + .write_all( + b"PATCH /machine-config HTTP/1.1\r\n\ + Content-Length: 13\r\n\ + Content-Type: application/json\r\n\r\nwhatever body", + ) + .unwrap(); + let mut second_socket = UnixStream::connect(path_to_socket).unwrap(); + + let mut req_vec = server.requests().unwrap(); + let server_request = req_vec.remove(0); + + server + .respond(server_request.process(|_request| { + let mut response = Response::new(Version::Http11, StatusCode::OK); + let response_body = b"response body"; + response.set_body(Body::new(response_body.to_vec())); + response + })) + .unwrap(); + second_socket + .write_all( + b"GET /machine-config HTTP/1.1\r\n\ + Content-Length: 20\r\n\ + Content-Type: application/json\r\n\r\nwhatever second body", + ) + .unwrap(); + + let mut req_vec = server.requests().unwrap(); + let second_server_request = req_vec.remove(0); + + assert_eq!( + second_server_request.request, + Request::try_from( + b"GET /machine-config HTTP/1.1\r\n\ + Content-Length: 20\r\n\ + Content-Type: application/json\r\n\r\nwhatever second body" + ) + .unwrap() + ); + + let mut buf: [u8; 1024] = [0; 1024]; + assert!(first_socket.read(&mut buf[..]).unwrap() > 0); + first_socket.shutdown(std::net::Shutdown::Both).unwrap(); + + server + .respond(second_server_request.process(|_request| { + let mut response = Response::new(Version::Http11, StatusCode::OK); + let response_body = b"response second body"; + response.set_body(Body::new(response_body.to_vec())); + response + })) + .unwrap(); + + assert!(server.requests().unwrap().is_empty()); + let mut buf: [u8; 1024] = [0; 1024]; + assert!(second_socket.read(&mut buf[..]).unwrap() > 0); + second_socket.shutdown(std::net::Shutdown::Both).unwrap(); + assert!(server.requests().unwrap().is_empty()); + fs::remove_file(path_to_socket).unwrap(); + } + + #[test] + fn test_wait_expect_connection() { + let path_to_socket = "/tmp/test_socket_http_server3.sock"; + fs::remove_file(path_to_socket).unwrap_or_default(); + + let mut server = HttpServer::new(path_to_socket.to_string()).unwrap(); + server.start_server().unwrap(); + + // Test one incoming connection with `Expect: 100-continue`. + let mut socket = UnixStream::connect(path_to_socket).unwrap(); + assert!(server.requests().unwrap().is_empty()); + + socket + .write_all( + b"PATCH /machine-config HTTP/1.1\r\n\ + Content-Length: 13\r\n\ + Expect: 100-continue\r\n\r\n", + ) + .unwrap(); + // `wait` on server to receive what the client set on the socket. + // This will set the stream direction to `Outgoing`, as we need to send a `100 CONTINUE` response. + let req_vec = server.requests().unwrap(); + assert!(req_vec.is_empty()); + // Another `wait`, this time to send the response. + // Will be called because of an `EPOLLOUT` notification. + let req_vec = server.requests().unwrap(); + assert!(req_vec.is_empty()); + let mut buf: [u8; 1024] = [0; 1024]; + assert!(socket.read(&mut buf[..]).unwrap() > 0); + + socket.write_all(b"whatever body").unwrap(); + let mut req_vec = server.requests().unwrap(); + let server_request = req_vec.remove(0); + + server + .respond(server_request.process(|_request| { + let mut response = Response::new(Version::Http11, StatusCode::OK); + let response_body = b"response body"; + response.set_body(Body::new(response_body.to_vec())); + response + })) + .unwrap(); + + let req_vec = server.requests().unwrap(); + assert!(req_vec.is_empty()); + + let mut buf: [u8; 1024] = [0; 1024]; + assert!(socket.read(&mut buf[..]).unwrap() > 0); + fs::remove_file(path_to_socket).unwrap(); + } + + #[test] + fn test_wait_many_connections() { + let path_to_socket = "/tmp/test_socket_http_server4.sock"; + fs::remove_file(path_to_socket).unwrap_or_default(); + + let mut server = HttpServer::new(path_to_socket.to_string()).unwrap(); + server.start_server().unwrap(); + + let mut sockets: Vec = Vec::with_capacity(11); + for _ in 0..MAX_CONNECTIONS { + sockets.push(UnixStream::connect(path_to_socket).unwrap()); + assert!(server.requests().unwrap().is_empty()); + } + + sockets.push(UnixStream::connect(path_to_socket).unwrap()); + assert!(server.requests().unwrap().is_empty()); + let mut buf: [u8; 120] = [0; 120]; + sockets[MAX_CONNECTIONS].read_exact(&mut buf).unwrap(); + assert_eq!(&buf[..], SERVER_FULL_ERROR_MESSAGE); + + fs::remove_file(path_to_socket).unwrap(); + } + + #[test] + fn test_wait_parse_error() { + let path_to_socket = "/tmp/test_socket_http_server5.sock"; + fs::remove_file(path_to_socket).unwrap_or_default(); + + let mut server = HttpServer::new(path_to_socket.to_string()).unwrap(); + server.start_server().unwrap(); + + // Test one incoming connection. + let mut socket = UnixStream::connect(path_to_socket).unwrap(); + socket.set_nonblocking(true).unwrap(); + assert!(server.requests().unwrap().is_empty()); + + socket + .write_all( + b"PATCH /machine-config HTTP/1.1\r\n\ + Content-Length: alpha\r\n\ + Content-Type: application/json\r\n\r\nwhatever body", + ) + .unwrap(); + + assert!(server.requests().unwrap().is_empty()); + assert!(server.requests().unwrap().is_empty()); + let mut buf: [u8; 198] = [0; 198]; + assert!(socket.read(&mut buf[..]).unwrap() > 0); + let error_message = b"HTTP/1.1 400 \r\n\ + Server: Firecracker API\r\n\ + Connection: keep-alive\r\n\ + Content-Type: application/json\r\n\ + Content-Length: 80\r\n\r\n{ \"error\": \"Invalid header.\n\ + All previous unanswered requests will be dropped.\" }"; + assert_eq!(&buf[..], &error_message[..]); + + fs::remove_file(path_to_socket).unwrap(); + } + + #[test] + fn test_wait_in_flight_responses() { + let path_to_socket = "/tmp/test_socket_http_server6.sock"; + fs::remove_file(path_to_socket).unwrap_or_default(); + + let mut server = HttpServer::new(path_to_socket.to_string()).unwrap(); + server.start_server().unwrap(); + + // Test a connection dropped and then a new one appearing + // before the user had a chance to send the response to the + // first one. + let mut first_socket = UnixStream::connect(path_to_socket).unwrap(); + assert!(server.requests().unwrap().is_empty()); + + first_socket + .write_all( + b"PATCH /machine-config HTTP/1.1\r\n\ + Content-Length: 13\r\n\ + Content-Type: application/json\r\n\r\nwhatever body", + ) + .unwrap(); + + let mut req_vec = server.requests().unwrap(); + let server_request = req_vec.remove(0); + + first_socket.shutdown(std::net::Shutdown::Both).unwrap(); + assert!(server.requests().unwrap().is_empty()); + let mut second_socket = UnixStream::connect(path_to_socket).unwrap(); + second_socket.set_nonblocking(true).unwrap(); + assert!(server.requests().unwrap().is_empty()); + + server + .enqueue_responses(vec![server_request.process(|_request| { + let mut response = Response::new(Version::Http11, StatusCode::OK); + let response_body = b"response body"; + response.set_body(Body::new(response_body.to_vec())); + response + })]) + .unwrap(); + assert!(server.requests().unwrap().is_empty()); + assert_eq!(server.connections.len(), 1); + let mut buf: [u8; 1024] = [0; 1024]; + assert!(second_socket.read(&mut buf[..]).is_err()); + + second_socket + .write_all( + b"GET /machine-config HTTP/1.1\r\n\ + Content-Length: 20\r\n\ + Content-Type: application/json\r\n\r\nwhatever second body", + ) + .unwrap(); + + let mut req_vec = server.requests().unwrap(); + let second_server_request = req_vec.remove(0); + + assert_eq!( + second_server_request.request, + Request::try_from( + b"GET /machine-config HTTP/1.1\r\n\ + Content-Length: 20\r\n\ + Content-Type: application/json\r\n\r\nwhatever second body" + ) + .unwrap() + ); + + server + .respond(second_server_request.process(|_request| { + let mut response = Response::new(Version::Http11, StatusCode::OK); + let response_body = b"response second body"; + response.set_body(Body::new(response_body.to_vec())); + response + })) + .unwrap(); + + assert!(server.requests().unwrap().is_empty()); + let mut buf: [u8; 1024] = [0; 1024]; + assert!(second_socket.read(&mut buf[..]).unwrap() > 0); + second_socket.shutdown(std::net::Shutdown::Both).unwrap(); + assert!(server.requests().is_ok()); + fs::remove_file(path_to_socket).unwrap(); + } +}