diff --git a/Cargo.toml b/Cargo.toml index 6a5114e..cbcf463 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,14 +7,14 @@ license-file = "LICENSE" homepage = "https://github.com/OctaneWeb/Octane" readme = "README.md" edition = "2018" +rust-version = "1.85" [dependencies] -lazy_static = "1.4.0" -tokio-rustls = { version = "0.20", optional = true } +tokio-rustls = { version = "0.26.6", default-features = false, features = ["ring", "tls12"], optional = true } openssl = { version = "0.10.30", optional = true } -tokio-openssl = { version = "0.5", optional = true } -tokio = { version = "0.3", features = ["net", "io-util", "stream", "rt-multi-thread", "macros", "fs"] } -socket2 = "0.3.19" +tokio-openssl = { version = "0.6.5", optional = true } +tokio = { version = "1.54.1", features = ["net", "io-util", "rt", "fs", "sync", "time"] } +socket2 = "0.6.5" serde = { version = "1.0.123", optional = true } serde_json = { version = "1.0.59", optional = true } @@ -42,10 +42,11 @@ openSSL = ["openssl", "tokio-openssl"] fuzzing = [] [dev-dependencies] -tokio = { version = "0.3", features = ["net", "io-util", "rt-multi-thread", "macros"] } +tokio = { version = "1.54.1", features = ["macros", "rt-multi-thread"] } curl = "0.4.33" [workspace] +resolver = "2" members = [ "octane_macros", "octane_http" diff --git a/octane_http/Cargo.toml b/octane_http/Cargo.toml index 198bf68..f1d2810 100644 --- a/octane_http/Cargo.toml +++ b/octane_http/Cargo.toml @@ -3,6 +3,7 @@ name = "octane_http" version = "0.1.0" authors = ["Daksh14 <41485688+Daksh14@users.noreply.github.com>"] edition = "2018" +rust-version = "1.85" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html @@ -11,5 +12,8 @@ path = "../octane_macros" version = "0.1.2" [dependencies] -tokio = { version = "0.3", features = ["net", "io-util", "stream", "rt-multi-thread"] } +tokio = "1.54.1" + +[dev-dependencies] +tokio = { version = "1.54.1", features = ["rt"] } diff --git a/octane_http/fuzz/Cargo.toml b/octane_http/fuzz/Cargo.toml index daa7f45..38a4fb1 100644 --- a/octane_http/fuzz/Cargo.toml +++ b/octane_http/fuzz/Cargo.toml @@ -3,13 +3,14 @@ name = "octane_http-fuzz" version = "0.0.0" publish = false edition = "2018" +rust-version = "1.85" [package.metadata] cargo-fuzz = true [dependencies] libfuzzer-sys = "0.4" -tokio = { version = "0.3", features = ["io-util", "rt"] } +tokio = { version = "1.54.1", features = ["rt"] } [dependencies.octane_http] path = ".." diff --git a/octane_http/fuzz/fuzz_targets/http1x_read.rs b/octane_http/fuzz/fuzz_targets/http1x_read.rs index 9b5e022..e771dd3 100644 --- a/octane_http/fuzz/fuzz_targets/http1x_read.rs +++ b/octane_http/fuzz/fuzz_targets/http1x_read.rs @@ -52,15 +52,18 @@ fuzz_target!(|data: &[u8]| { // The returned request borrows `buffered`, so copy its fields out // before inspecting the buffer again. - let owned = Http1xReader::new(&mut io, &mut buffered).await.ok().map(|req| { - ( - req.request_method.to_vec(), - req.request_url.to_vec(), - req.request_version.to_vec(), - req.headers.to_vec(), - req.body_remainder.to_vec(), - ) - }); + let owned = Http1xReader::new(&mut io, &mut buffered) + .await + .ok() + .map(|req| { + ( + req.request_method.to_vec(), + req.request_url.to_vec(), + req.request_version.to_vec(), + req.headers.to_vec(), + req.body_remainder.to_vec(), + ) + }); // The reader only ever buffers bytes the connection produced, in order. assert!(wire.starts_with(&buffered)); diff --git a/octane_http/fuzz/fuzz_targets/http2_frame.rs b/octane_http/fuzz/fuzz_targets/http2_frame.rs index 74b5802..c634c92 100644 --- a/octane_http/fuzz/fuzz_targets/http2_frame.rs +++ b/octane_http/fuzz/fuzz_targets/http2_frame.rs @@ -126,7 +126,10 @@ fuzz_target!(|data: &[u8]| { let mut full = Vec::with_capacity(PREFACE.len() + wire.len()); full.extend_from_slice(PREFACE); full.extend_from_slice(wire); - let mut io = ChunkedIo { input: &full, chunk }; + let mut io = ChunkedIo { + input: &full, + chunk, + }; if Http2FrameReader::read_preface(&mut io).await.is_err() { return; diff --git a/octane_http/fuzz/fuzz_targets/http2_hpack.rs b/octane_http/fuzz/fuzz_targets/http2_hpack.rs index 9ab385c..746a480 100644 --- a/octane_http/fuzz/fuzz_targets/http2_hpack.rs +++ b/octane_http/fuzz/fuzz_targets/http2_hpack.rs @@ -30,7 +30,10 @@ fuzz_target!(|data: &[u8]| { // Huffman: decoding arbitrary bytes may fail but must not panic, and // anything we encode must decode back to itself. let _ = huffman::decode(rest); - assert_eq!(huffman::decode(&huffman::encode(rest)).as_deref(), Some(rest)); + assert_eq!( + huffman::decode(&huffman::encode(rest)).as_deref(), + Some(rest) + ); // Encoder round-trip: a header field encoded by the server must survive a // trip through its own decoder byte-for-byte. diff --git a/octane_http/src/http1x/mod.rs b/octane_http/src/http1x/mod.rs index 938402a..5f319ea 100644 --- a/octane_http/src/http1x/mod.rs +++ b/octane_http/src/http1x/mod.rs @@ -1,17 +1,116 @@ use crate::http1x::raw_request::RawRequest1x; use crate::{StatusCode, DOUBLE_CRLF}; -use tokio::io::{AsyncRead, AsyncReadExt}; +use std::future::Future; +use std::io; +use std::mem::MaybeUninit; +use std::pin::Pin; +use std::task::{Context, Poll}; +use tokio::io::{AsyncRead, ReadBuf}; pub mod raw_request; // 8 kb static read buffer, the standard chunk size across nginx and friends. const BUFFER_SIZE: usize = 8192; -// Upper bound on the request head (request-line + headers). A peer that never -// sends the CRLFCRLF terminator must not be able to grow `data` without limit. +// Upper bound on the complete request head (request-line, headers and the +// terminating CRLFCRLF). A peer that never sends the terminator must not be +// able to grow `data` without limit. const MAX_HEAD_SIZE: usize = 64 * 1024; +/// Incremental CRLFCRLF search state. +/// +/// Once a prefix has been searched, a future terminator can only start in the +/// last three bytes of that prefix or in newly-read bytes. Remembering that +/// boundary makes fragmented, one-byte-at-a-time input linear rather than +/// repeatedly searching the whole accumulated head. +#[derive(Default)] +struct HeadTerminatorSearch { + next_start: usize, +} + +impl HeadTerminatorSearch { + fn find(&mut self, data: &[u8]) -> Option { + let start = self.next_start.min(data.len()); + let found = find_head_end(data, start); + + if found.is_none() { + self.next_start = data.len().saturating_sub(DOUBLE_CRLF.len() - 1); + } + + found + } +} + +/// Find the end (rather than the start) of CRLFCRLF at or after `start`. +#[inline] +fn find_head_end(data: &[u8], start: usize) -> Option { + let last_start = data.len().checked_sub(DOUBLE_CRLF.len())?; + let mut i = start; + + while i <= last_start { + // This is intentionally specialized for HTTP's four-byte delimiter. + // It avoids the inner dynamic loop in the public generic helper. + if data[i] == b'\r' && data[i + 1] == b'\n' && data[i + 2] == b'\r' && data[i + 3] == b'\n' + { + return Some(i + DOUBLE_CRLF.len()); + } + i += 1; + } + + None +} + +/// Poll one read without retaining an 8 KiB scratch array in every suspended +/// reader future. The temporary storage lives only on the executor's stack +/// while this closure is actively being polled, and `MaybeUninit` avoids +/// clearing bytes that the I/O source immediately overwrites. +struct ReadChunk<'a, R> { + reader: &'a mut R, + data: &'a mut Vec, + read_limit: usize, +} + +impl Future for ReadChunk<'_, R> +where + R: AsyncRead + Unpin, +{ + type Output = io::Result; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.get_mut(); + debug_assert!(this.read_limit <= BUFFER_SIZE); + let mut storage = [MaybeUninit::::uninit(); BUFFER_SIZE]; + let mut buf = ReadBuf::uninit(&mut storage[..this.read_limit]); + + match Pin::new(&mut *this.reader).poll_read(cx, &mut buf) { + Poll::Ready(Ok(())) => { + let bytes = buf.filled(); + let count = bytes.len(); + this.data.extend_from_slice(bytes); + Poll::Ready(Ok(count)) + } + Poll::Ready(Err(error)) => Poll::Ready(Err(error)), + Poll::Pending => Poll::Pending, + } + } +} + +fn read_chunk<'a, R>( + reader: &'a mut R, + data: &'a mut Vec, + read_limit: usize, +) -> ReadChunk<'a, R> +where + R: AsyncRead + Unpin, +{ + ReadChunk { + reader, + data, + read_limit, + } +} + /// Reads and parses the head of an HTTP/1.x request straight off an /// [`AsyncRead`] source, with zero copies of the parsed fields. /// @@ -32,30 +131,39 @@ impl Http1xReader { where R: AsyncRead + Unpin, { - let mut buf = [0u8; BUFFER_SIZE]; + let mut terminator = HeadTerminatorSearch::default(); // Keep reading until the head is complete. We only need the head // buffered to parse the request; the caller can stream the rest of the // body afterwards from the same reader. - while find_in_slice(&data[..], DOUBLE_CRLF).is_none() { - if data.len() > MAX_HEAD_SIZE { + loop { + // Only the first MAX_HEAD_SIZE bytes can contain an admissible + // terminator. This also bounds work when a caller supplies a large + // pre-populated buffer containing an already-buffered body. + let searchable_len = data.len().min(MAX_HEAD_SIZE); + if let Some(head_end) = terminator.find(&data[..searchable_len]) { + // Reuse the delimiter position so the zero-copy parser does + // not make a second pass over a potentially 64 KiB head. + return RawRequest1x::parse_with_head_end(&data[..], head_end); + } + + // At exactly the limit, an incomplete delimiter can only finish + // beyond it. Checking before the next read keeps the cap exact and + // independent of the source's chunk size. + if searchable_len == MAX_HEAD_SIZE { return Err(StatusCode::RequestHeaderFieldsTooLarge); } - let n = reader - .read(&mut buf) + let read_limit = BUFFER_SIZE.min(MAX_HEAD_SIZE - data.len()); + let n = read_chunk(reader, data, read_limit) .await .map_err(|_| StatusCode::BadRequest)?; if n == 0 { // EOF before a complete head arrived. - break; + return Err(StatusCode::BadRequest); } - - data.extend_from_slice(&buf[..n]); } - - RawRequest1x::parse(&data[..]) } } @@ -91,19 +199,58 @@ pub fn find_in_slice(haystack: &[T], needle: &[T]) -> Option { None } -// Helper function for extracting some headers -pub(crate) fn parse_without_body(data: &str) -> Option<(&str, String)> { - let n = data.find("\r\n")?; - let (line, rest) = data.split_at(n); - let request_line = line; - let headers = (&rest[2..]).to_owned(); - Some((request_line, headers)) -} - #[cfg(test)] mod tests { use super::*; + struct ChunkedReader<'a> { + remaining: &'a [u8], + max_chunk: usize, + reads: usize, + } + + impl<'a> ChunkedReader<'a> { + fn new(bytes: &'a [u8], max_chunk: usize) -> Self { + assert!(max_chunk > 0); + Self { + remaining: bytes, + max_chunk, + reads: 0, + } + } + } + + impl AsyncRead for ChunkedReader<'_> { + fn poll_read( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let this = self.get_mut(); + let count = this + .remaining + .len() + .min(this.max_chunk) + .min(buf.remaining()); + let (chunk, remaining) = this.remaining.split_at(count); + buf.put_slice(chunk); + this.remaining = remaining; + this.reads += 1; + Poll::Ready(Ok(())) + } + } + + fn request_with_head_len(head_len: usize) -> Vec { + const PREFIX: &[u8] = b"GET / HTTP/1.1\r\nX: "; + assert!(head_len >= PREFIX.len() + DOUBLE_CRLF.len()); + + let mut request = Vec::with_capacity(head_len); + request.extend_from_slice(PREFIX); + request.resize(head_len - DOUBLE_CRLF.len(), b'a'); + request.extend_from_slice(DOUBLE_CRLF); + request + } + fn block_on(fut: F) -> F::Output { tokio::runtime::Builder::new_current_thread() .build() @@ -144,4 +291,85 @@ mod tests { assert_eq!(find_in_slice(b"abcdef", b"XY"), None); assert_eq!(find_in_slice(b"a", b"abc"), None); } + + #[test] + fn accepts_a_head_exactly_at_the_limit_for_any_chunking() { + let request = request_with_head_len(MAX_HEAD_SIZE); + + for chunk_size in [1, 17, BUFFER_SIZE, usize::MAX] { + let mut reader = ChunkedReader::new(&request, chunk_size); + let mut data = Vec::new(); + + let parsed = block_on(Http1xReader::new(&mut reader, &mut data)).unwrap(); + assert_eq!(parsed.request_method, b"GET"); + assert!(parsed.body_remainder.is_empty()); + assert_eq!(data.len(), MAX_HEAD_SIZE, "chunk size {}", chunk_size); + } + } + + #[test] + fn rejects_a_head_one_byte_over_the_limit_for_any_chunking() { + let request = request_with_head_len(MAX_HEAD_SIZE + 1); + + for chunk_size in [1, 17, BUFFER_SIZE, usize::MAX] { + let mut reader = ChunkedReader::new(&request, chunk_size); + let mut data = Vec::new(); + + let result = block_on(Http1xReader::new(&mut reader, &mut data)); + assert_eq!( + result, + Err(StatusCode::RequestHeaderFieldsTooLarge), + "chunk size {}", + chunk_size + ); + assert_eq!(data.len(), MAX_HEAD_SIZE, "chunk size {}", chunk_size); + } + } + + #[test] + fn does_not_charge_an_already_buffered_body_to_the_head_limit() { + let head = b"POST / HTTP/1.1\r\nContent-Length: 131072\r\n\r\n"; + let body_len = MAX_HEAD_SIZE * 2; + let mut data = Vec::with_capacity(head.len() + body_len); + data.extend_from_slice(head); + data.resize(head.len() + body_len, b'x'); + let mut reader = ChunkedReader::new(&[], 1); + + let parsed = block_on(Http1xReader::new(&mut reader, &mut data)).unwrap(); + + assert_eq!(parsed.body_remainder.len(), body_len); + assert_eq!(reader.reads, 0); + } + + #[test] + fn one_byte_fragmentation_searches_each_start_position_once() { + let mut data = Vec::with_capacity(MAX_HEAD_SIZE); + let mut search = HeadTerminatorSearch::default(); + let mut candidate_starts = 0usize; + + assert_eq!(search.find(&data), None); + for _ in 0..MAX_HEAD_SIZE { + data.push(b'a'); + candidate_starts += + (data.len() - search.next_start).saturating_sub(DOUBLE_CRLF.len() - 1); + assert_eq!(search.find(&data), None); + } + + assert_eq!(candidate_starts, MAX_HEAD_SIZE - DOUBLE_CRLF.len() + 1); + } + + #[test] + fn suspended_reader_does_not_retain_the_scratch_buffer() { + let raw = b""; + let mut data = Vec::new(); + let mut reader = &raw[..]; + let future = Http1xReader::new(&mut reader, &mut data); + let future_size = std::mem::size_of_val(&future); + + assert!( + future_size < 1024, + "reader future unexpectedly retained {} bytes", + future_size + ); + } } diff --git a/octane_http/src/http1x/raw_request.rs b/octane_http/src/http1x/raw_request.rs index 6feb834..9a19491 100644 --- a/octane_http/src/http1x/raw_request.rs +++ b/octane_http/src/http1x/raw_request.rs @@ -46,6 +46,19 @@ impl<'a> RawRequest1x<'a> { /// terminated by `CRLFCRLF`; any bytes after that terminator are exposed /// through [`RawRequest1x::body_remainder`]. No bytes are copied. pub fn parse(buf: &'a [u8]) -> Result { + Self::parse_inner(buf, None) + } + + /// Parse using a CRLFCRLF end position already found by the streaming + /// reader. Valid requests avoid searching the complete head a second time. + pub(super) fn parse_with_head_end( + buf: &'a [u8], + complete_head_end: usize, + ) -> Result { + Self::parse_inner(buf, Some(complete_head_end)) + } + + fn parse_inner(buf: &'a [u8], complete_head_end: Option) -> Result { // METHOD SP ... let method_end = eat_till(buf, SP)?; let request_method = &buf[..method_end]; @@ -74,7 +87,28 @@ impl<'a> RawRequest1x<'a> { // request with zero headers (where the request-line CRLF is itself the // first half of the terminator) parse without a special case. let after_version = &after_url[version_end..]; - let head_end = eat_till_four_bytes(after_version, DOUBLE_CRLF)?; + let after_version_start = buf.len() - after_version.len(); + let head_end = match complete_head_end { + Some(complete_head_end) => { + let terminator_start = complete_head_end + .checked_sub(DOUBLE_CRLF.len()) + .ok_or(StatusCode::BadRequest)?; + + if buf.get(terminator_start..complete_head_end) != Some(&DOUBLE_CRLF[..]) { + return Err(StatusCode::BadRequest); + } + + if terminator_start >= after_version_start { + terminator_start - after_version_start + } else { + // Preserve `parse`'s behavior for structurally malformed + // request-lines containing an earlier CRLFCRLF. Normal + // requests always take the zero-rescan branch above. + eat_till_four_bytes(after_version, DOUBLE_CRLF)? + } + } + None => eat_till_four_bytes(after_version, DOUBLE_CRLF)?, + }; // Skip the request-line CRLF (2 bytes) to reach the header block, which // runs up to the start of the terminating CRLFCRLF. When there are no diff --git a/src/config.rs b/src/config.rs index e235e54..59e5724 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,11 +1,12 @@ use crate::default; use core::time::Duration; +#[cfg(feature = "rustls")] +use std::fs::File; +#[cfg(feature = "rustls")] +use std::io; use std::path::PathBuf; #[cfg(feature = "rustls")] -use tokio_rustls::rustls::{ - internal::pemfile::{certs, rsa_private_keys}, - Certificate, PrivateKey, -}; +use tokio_rustls::rustls::pki_types::{pem::PemObject, CertificateDer, PrivateKeyDer}; /// Ssl struct contains the key and cert /// required to setup SSL with the selected @@ -28,7 +29,7 @@ use tokio_rustls::rustls::{ /// ``` #[derive(Clone)] pub struct Ssl { - /// Location of the private key file, should be an `RSA_PRIVATE_KEY`. + /// Location of a PEM-encoded private key file. pub key: PathBuf, /// Location of the certificate file. pub cert: PathBuf, @@ -103,6 +104,16 @@ impl Ssl { pub struct OctaneConfig { /// The duration for keep alive requests. It is 5 seconds by default pub keep_alive: Option, + /// Maximum number of connections being served at once. Once the limit is + /// reached, each listener stops draining the operating system's backlog + /// while it waits for capacity. It is 1024 by default; zero is treated as + /// one, and values too large for the runtime return an error from `listen`. + pub max_connections: usize, + /// Maximum total time spent on one connection after it acquires serving + /// capacity, including a TLS handshake when TLS is enabled. `None` + /// disables the deadline. It is 30 seconds by default. Custom Tokio + /// runtimes must enable the time driver with `enable_time` or `enable_all`. + pub connection_timeout: Option, /// An instance of the `Ssl` struct to store the values of key and certificates. pub ssl: Ssl, /// The largest request body in bytes the server will read, requests @@ -209,6 +220,8 @@ impl OctaneConfig { OctaneConfig { ssl: Ssl::new(), keep_alive: Some(Duration::from_secs(5)), + max_connections: 1024, + connection_timeout: Some(Duration::from_secs(30)), max_body_size: 2 * 1024 * 1024, worker_threads: None, } @@ -217,33 +230,36 @@ impl OctaneConfig { pub(crate) fn append(&mut self, settings: Self) { self.ssl = settings.ssl; self.keep_alive = settings.keep_alive; + self.max_connections = settings.max_connections; + self.connection_timeout = settings.connection_timeout; self.max_body_size = settings.max_body_size; } /// Sets the number of worker threads, this is settings /// which will be applied to the `core_threads` method /// on the runtime builder struct. Alias for - /// [tokio's core_threads](https://docs.rs/tokio/0.2.13/tokio/runtime/struct.Builder.html#method.core_threads) + /// [Tokio's worker_threads](https://docs.rs/tokio/latest/tokio/runtime/struct.Builder.html#method.worker_threads) pub fn worker_threads(&mut self, threads: usize) -> &mut Self { self.worker_threads = Some(threads); self } - // Get the certs as a Vec, a user will not have to + // Get the certificate chain, a user will not have to // use this directly, this is used and done for them #[cfg(feature = "rustls")] - pub(crate) fn get_cert(&self) -> std::io::Result> { - let mut buf = std::io::BufReader::new(std::fs::File::open(&self.ssl.cert)?); - certs(&mut buf) - .map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidInput, "Invalid Certs")) + pub(crate) fn get_cert(&self) -> io::Result>> { + let file = File::open(&self.ssl.cert)?; + CertificateDer::pem_reader_iter(file) + .collect::, _>>() + .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error)) } - // Get the private key as a Vec, a user will not have to + // Get the first supported private key, a user will not have to // use this directly #[cfg(feature = "rustls")] - pub(crate) fn get_key(&self) -> std::io::Result> { - let mut buf = std::io::BufReader::new(std::fs::File::open(&self.ssl.key)?); - rsa_private_keys(&mut buf) - .map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidInput, "Invalid Key")) + pub(crate) fn get_key(&self) -> io::Result> { + let file = File::open(&self.ssl.key)?; + PrivateKeyDer::from_pem_reader(file) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error)) } } diff --git a/src/constants.rs b/src/constants.rs index 6239c2a..eb7aea9 100644 --- a/src/constants.rs +++ b/src/constants.rs @@ -1,7 +1,4 @@ #![allow(dead_code)] -use std::collections::HashSet; -use std::iter::FromIterator; - pub const SP: char = ' '; pub const HT: char = '\t'; pub const CR: char = '\r'; @@ -21,10 +18,32 @@ pub const DAYS_PER_4Y: i64 = 365 * 4 + 1; pub static DAYS_IN_MONTH: [i64; 12] = [31, 30, 31, 30, 31, 31, 30, 31, 30, 31, 31, 29]; pub const NOT_FOUND: &str = r#"404 NOT FOUND - OCTANE

404 NOT FOUND

OCTANE - 0.1
"#; -lazy_static! { - pub static ref TOKEN_CHARS: HashSet = HashSet::from_iter( - "!#$%&'*+-.0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ^_`abcdefghijklmnopqrstuvwxyz|~".chars() - ); +#[inline] +pub(crate) fn is_token(value: &str) -> bool { + !value.is_empty() + && value.bytes().all(|byte| { + matches!( + byte, + b'0'..=b'9' + | b'A'..=b'Z' + | b'a'..=b'z' + | b'!' + | b'#' + | b'$' + | b'%' + | b'&' + | b'\'' + | b'*' + | b'+' + | b'-' + | b'.' + | b'^' + | b'_' + | b'`' + | b'|' + | b'~' + ) + }) } #[macro_export] diff --git a/src/cookie.rs b/src/cookie.rs index 5446188..ec90d0c 100644 --- a/src/cookie.rs +++ b/src/cookie.rs @@ -5,7 +5,7 @@ //! through, while serialising is strict so the server never sends a header //! that a browser would reject or misread. -use crate::constants::{HT, SP, TOKEN_CHARS}; +use crate::constants::{is_token, HT, SP}; use crate::path::is_ctl; use crate::time::Time; use std::borrow::Cow; @@ -375,10 +375,6 @@ fn unquote(value: &str) -> &str { } } -fn is_token(s: &str) -> bool { - !s.is_empty() && s.chars().all(|c| TOKEN_CHARS.contains(&c)) -} - // cookie-octet: printable US-ASCII except `"`, `,`, `;` and `\` fn is_cookie_octet(b: u8) -> bool { matches!(b, 0x21 | 0x23..=0x2B | 0x2D..=0x3A | 0x3C..=0x5B | 0x5D..=0x7E) diff --git a/src/error.rs b/src/error.rs index 52a58f3..714f980 100644 --- a/src/error.rs +++ b/src/error.rs @@ -49,12 +49,12 @@ impl Error { where S: AsyncWrite + Unpin, { - let mut res = Response::new_from_slice(NOT_FOUND.as_bytes()); + let mut res = Response::new_from_static(NOT_FOUND.as_bytes()); res.status(self.kind) .default_headers() .set("Content-Type", "text/html"); - let response = res.get_data(); + let response = res.into_parts(); Octane::send(response, stream).await?; Ok(()) } diff --git a/src/lib.rs b/src/lib.rs index 0da5935..7ba0b48 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -15,7 +15,7 @@ //! //! ```toml //! octane = "0.1.1" -//! tokio = "0.2.22" +//! tokio = { version = "1.54.1", features = ["macros", "rt-multi-thread"] } //! ``` //! //! and then in your main.rs, @@ -64,8 +64,6 @@ //! The rust code you would use to setup openssl will be same as rustls (not including certs and keys). //! If you want to switch between any of them, its just the matter of editing your cargo.toml //! to enable the feature. -#[macro_use] -extern crate lazy_static; /// Configurations for Octane web server pub mod config; pub(crate) mod constants; diff --git a/src/request.rs b/src/request.rs index bd09c29..afbb065 100644 --- a/src/request.rs +++ b/src/request.rs @@ -15,8 +15,7 @@ use std::collections::HashMap; use std::marker::PhantomData; use std::str; use std::string::ToString; -use tokio::io::ReadHalf; -use tokio::prelude::*; +use tokio::io::{AsyncRead, AsyncReadExt}; /// Holds the type of request method, like GET, /// POST etc. @@ -90,18 +89,18 @@ impl RequestLine { pub(crate) fn parse(request_line: &str) -> Option { let mut toks = request_line.split(SP); let method = toks.next()?; - // only the path is used for routing, the query is kept apart let target = toks.next()?; + let version = toks.next()?; + Self::from_parts(method, target, version, toks.next().is_some()) + } + + fn from_parts(method: &str, target: &str, version: &str, has_extra: bool) -> Option { + // only the path is used for routing, the query is kept apart let (path, query) = target.split_once('?').unwrap_or((target, "")); let path = PathBuf::parse(path).ok()?; - let version = toks.next()?; - // `split_at` panics on a version shorter than 5 bytes or with a - // multi-byte character over byte 5 - let (first, ver) = if version.is_char_boundary(5) { - version.split_at(5) - } else { - (version, "") - }; + // `get` avoids slicing through a multi-byte character and handles + // versions shorter than the `HTTP/` prefix without panicking. + let ver = version.get(5..).unwrap_or(""); let enum_ver = match ver { "1.1" => HttpVersion::Http11, "1.0" => HttpVersion::Http10, @@ -110,7 +109,7 @@ impl RequestLine { _ => HttpVersion::HttpInvalid, }; - if cfg!(feature = "faithful") && (first != "HTTP/" || toks.next().is_some()) { + if cfg!(feature = "faithful") && (version.get(..5) != Some("HTTP/") || has_extra) { return None; } let request_method = match method { @@ -141,6 +140,13 @@ pub(crate) fn request_line_parse(method: &[u8], url: &[u8], version: &[u8]) -> O str::from_utf8(url), str::from_utf8(version), ) { + (Ok(method), Ok(url), Ok(version)) + if !method.contains(SP) && !url.contains(SP) && !version.contains(SP) => + { + RequestLine::from_parts(method, url, version, false) + } + // Keep the exact legacy behavior for malformed fields containing a + // space. Valid parser output takes the allocation-free branch above. (Ok(method), Ok(url), Ok(version)) => { RequestLine::parse(&format!("{} {} {}", method, url, version)) } @@ -165,16 +171,15 @@ pub struct Header { impl Header { /// Parses a `key: value` header string and /// returns a Header struct - pub(crate) fn parse(header: String) -> Option { + pub(crate) fn parse(header: impl AsRef) -> Option { + let header = header.as_ref(); let mut toks = header.splitn(2, ':'); let name = toks.next()?; if name.is_empty() { return None; } - if cfg!(feature = "faithful") { - for c in name.chars() { - TOKEN_CHARS.get(&c)?; - } + if cfg!(feature = "faithful") && !is_token(name) { + return None; } // whitespace around a field value isn't part of it, and HTAB is the // only control character allowed inside one @@ -229,7 +234,8 @@ pub struct Headers { impl Headers { /// Parse all the headers on a request - pub(crate) fn parse(request: String) -> Option { + pub(crate) fn parse(request: impl AsRef) -> Option { + let request = request.as_ref(); let mut toks = Spliterator::new(request.as_bytes(), B_CRLF); // an empty block is a request without any header lines toks.finished = request.is_empty(); @@ -237,20 +243,37 @@ impl Headers { #[cfg(feature = "raw_headers")] let mut raw_headers: Vec
= Vec::new(); for tok in toks { - let parsed = Header::parse(match str::from_utf8(tok) { - Ok(s) => s.to_owned(), - Err(_) => return None, - })?; - let name = parsed.name.to_ascii_lowercase(); - // cookie pairs are joined with "; " (RFC 6265 section 5.4) - let sep = if name == "cookie" { "; " } else { ", " }; - headers - .entry(name) - .and_modify(|v| { - v.push_str(sep); - v.push_str(&parsed.value); - }) - .or_insert_with(|| parsed.value.to_owned()); + let parsed = Header::parse(str::from_utf8(tok).ok()?)?; + #[cfg(feature = "raw_headers")] + { + let name = parsed.name.to_ascii_lowercase(); + // cookie pairs are joined with "; " (RFC 6265 section 5.4) + let sep = if name == "cookie" { "; " } else { ", " }; + headers + .entry(name) + .and_modify(|v| { + v.push_str(sep); + v.push_str(&parsed.value); + }) + .or_insert_with(|| parsed.value.to_owned()); + } + #[cfg(not(feature = "raw_headers"))] + { + // Without raw-header preservation, reuse the allocations made + // by `Header::parse` instead of cloning both strings. + let Header { mut name, value } = parsed; + name.make_ascii_lowercase(); + let sep = if name == "cookie" { "; " } else { ", " }; + match headers.entry(name) { + std::collections::hash_map::Entry::Occupied(mut entry) => { + entry.get_mut().push_str(sep); + entry.get_mut().push_str(&value); + } + std::collections::hash_map::Entry::Vacant(entry) => { + entry.insert(value); + } + } + } #[cfg(feature = "raw_headers")] raw_headers.push(parsed); } @@ -308,7 +331,10 @@ impl Headers { let sep = if lower == "cookie" { "; " } else { ", " }; parsed .entry(lower) - .and_modify(|v| *v = format!("{}{}{}", v, sep, value)) + .and_modify(|v| { + v.push_str(sep); + v.push_str(value); + }) .or_insert_with(|| value.clone()); #[cfg(feature = "raw_headers")] raw.push(Header { @@ -370,20 +396,19 @@ impl<'a> Request<'a> { /// and body and return a Request struct pub(crate) fn parse( request_line: RequestLine, - headers: &'a Headers, + headers: Headers, body: &'a [u8], ) -> Option { #[cfg(feature = "cookies")] - let cookies: Vec>; - #[cfg(feature = "cookies")] - if let Some(v) = headers.get("cookie") { - cookies = Cookie::parse(v); - } else { - cookies = Default::default(); - } + let cookies: Vec> = headers.get("cookie").map_or_else(Vec::new, |value| { + Cookie::parse(value) + .into_iter() + .map(|cookie| Cookie::new(cookie.name().to_owned(), cookie.value().to_owned())) + .collect() + }); Some(Self { request_line, - headers: headers.clone(), + headers, #[cfg(feature = "cookies")] cookies, body, @@ -392,38 +417,47 @@ impl<'a> Request<'a> { /// Build a [`Request`] from an already-parsed request line and header set, /// reading any body bytes that weren't already buffered alongside the head. /// - /// `body_remainder` is the slice of the body that arrived in the same read - /// as the head; if `Content-Length` calls for more, the rest is streamed - /// off `reader` into `body_vec`. Bodies longer than `max_body_size` are - /// refused with a `413` before anything is read. + /// `body_start` is the offset in `data` immediately after the request head. + /// If `Content-Length` calls for more bytes than are already buffered, the + /// rest is streamed off `reader` into that same allocation. Bodies longer + /// than `max_body_size` are refused with a `413` before anything is read. pub(crate) async fn from_raw( - headers: &'a Headers, + headers: Headers, request_line: RequestLine, - body_remainder: &'a [u8], - body_vec: &'a mut Vec, - mut reader: ReadHalf, + data: &'a mut Vec, + body_start: usize, + mut reader: T, max_body_size: usize, ) -> Result, StatusCode> { let body_len = headers.content_length(max_body_size)?; - let body: &'a [u8] = if body_len <= body_remainder.len() { - // bytes past the declared length aren't part of this body - &body_remainder[..body_len] - } else { + let buffered_body_len = data + .len() + .checked_sub(body_start) + .ok_or(StatusCode::BadRequest)?; + let body_end = body_start + .checked_add(body_len) + .ok_or(StatusCode::PayloadTooLarge)?; + + if body_len > buffered_body_len { // grow the buffer as bytes arrive instead of allocating the // declared length up front - body_vec.extend_from_slice(body_remainder); - let missing = (body_len - body_remainder.len()) as u64; + let missing = (body_len - buffered_body_len) as u64; (&mut reader) .take(missing) - .read_to_end(body_vec) + .read_to_end(data) .await .map_err(|_| StatusCode::BadRequest)?; // the client hung up before sending the whole body - if body_vec.len() != body_len { + if data.len() != body_end { return Err(StatusCode::BadRequest); } - &body_vec[..] - }; + } + + // Bytes already buffered or sent after the declared length are not + // part of this request body. + let body = data + .get(body_start..body_end) + .ok_or(StatusCode::BadRequest)?; Self::parse(request_line, headers, body).ok_or(StatusCode::BadRequest) } @@ -591,6 +625,25 @@ mod test { assert_eq!(req.version, HttpVersion::Http11); } + #[test] + fn byte_request_line_parser_matches_string_parser() { + let cases = [ + ("GET", "/", "HTTP/1.1"), + ("PATCH", "/a/b?x=1", "HTTP/1.0"), + ("UNKNOWN", "/resource", "HTTP/9.9"), + // Inputs containing spaces take the compatibility fallback. + ("GET extra", "/", "HTTP/1.1"), + ("GET", "/two paths", "HTTP/1.1"), + ]; + for (method, target, version) in cases { + let combined = format!("{} {} {}", method, target, version); + assert_eq!( + request_line_parse(method.as_bytes(), target.as_bytes(), version.as_bytes()), + RequestLine::parse(&combined) + ); + } + } + #[test] fn sucess_non_documented() { // Non-documented methods should also work. @@ -754,6 +807,14 @@ mod test { assert!(Headers::parse(String::new()).unwrap().is_empty()); } + #[test] + fn headers_accept_borrowed_blocks() { + let block = String::from("Host: localhost\r\nX-Test: value"); + let headers = Headers::parse(block.as_str()).unwrap(); + assert_eq!(headers.get("host").unwrap(), "localhost"); + assert_eq!(headers.get("x-test").unwrap(), "value"); + } + #[test] fn success_value_whitespace_and_tabs() { // Whitespace around a value is dropped, tabs inside it are kept. @@ -802,4 +863,76 @@ mod test { assert_eq!(parse("Content-Length: 101"), too_large); assert_eq!(parse("Content-Length: 99999999999999999999999"), too_large); } + + #[tokio::test] + async fn from_raw_appends_missing_body_to_the_request_buffer() { + let mut data = Vec::with_capacity(16); + data.extend_from_slice(b"HEADhe"); + let request = Request::from_raw( + Headers::parse("Content-Length: 5").unwrap(), + RequestLine::parse("POST / HTTP/1.1").unwrap(), + &mut data, + 4, + &b"lloignored"[..], + 16, + ) + .await + .unwrap(); + + assert_eq!(request.body, b"hello"); + let body_pointer = request.body.as_ptr() as usize; + drop(request); + assert_eq!(body_pointer, data.as_ptr() as usize + 4); + assert_eq!(data, b"HEADhello"); + } + + #[tokio::test] + async fn from_raw_excludes_already_buffered_trailing_bytes() { + let mut data = b"HEADhello".to_vec(); + let request = Request::from_raw( + Headers::parse("Content-Length: 2").unwrap(), + RequestLine::parse("POST / HTTP/1.1").unwrap(), + &mut data, + 4, + &b""[..], + 16, + ) + .await + .unwrap(); + + assert_eq!(request.body, b"he"); + drop(request); + assert_eq!(data, b"HEADhello"); + } + + #[tokio::test] + async fn from_raw_rejects_oversized_and_incomplete_bodies() { + let mut oversized_data = b"HEAD".to_vec(); + let oversized = Request::from_raw( + Headers::parse("Content-Length: 6").unwrap(), + RequestLine::parse("POST / HTTP/1.1").unwrap(), + &mut oversized_data, + 4, + &b"unused"[..], + 5, + ) + .await; + assert!(matches!(oversized, Err(StatusCode::PayloadTooLarge))); + drop(oversized); + assert_eq!(oversized_data, b"HEAD"); + + let mut incomplete_data = b"HEADhe".to_vec(); + let incomplete = Request::from_raw( + Headers::parse("Content-Length: 5").unwrap(), + RequestLine::parse("POST / HTTP/1.1").unwrap(), + &mut incomplete_data, + 4, + &b"l"[..], + 5, + ) + .await; + assert!(matches!(incomplete, Err(StatusCode::BadRequest))); + drop(incomplete); + assert_eq!(incomplete_data, b"HEADhel"); + } } diff --git a/src/responder.rs b/src/responder.rs index 9b0910a..f6f6d71 100644 --- a/src/responder.rs +++ b/src/responder.rs @@ -2,12 +2,15 @@ use crate::constants::*; #[cfg(feature = "cookies")] use crate::cookie::Cookie; use crate::file_handler::FileHandler; -use crate::path::is_ctl; use crate::time::Time; use octane_http::{HttpVersion, StatusCode}; +#[cfg(feature = "json")] +use serde::Serialize; use std::collections::HashMap; use std::error::Error; +#[cfg(feature = "cookies")] use std::fmt; +use std::fmt::Write; use std::io::Cursor; use std::marker::PhantomData; use std::path::PathBuf; @@ -17,8 +20,9 @@ pub(crate) type BoxReader = Box; pub(crate) enum ResBody { None, + Bytes(Vec), + Static(&'static [u8]), Sized(usize, BoxReader), - Unsized(BoxReader), } impl ResBody { @@ -26,7 +30,8 @@ impl ResBody { match self { // never send more than the Content-Length we announced ResBody::Sized(len, reader) => Box::new(reader.take(len as u64)) as BoxReader, - ResBody::Unsized(reader) => reader, + ResBody::Bytes(body) => Box::new(Cursor::new(body)) as BoxReader, + ResBody::Static(body) => Box::new(Cursor::new(body)) as BoxReader, ResBody::None => Box::new(Cursor::new(Vec::new())) as BoxReader, } } @@ -95,7 +100,7 @@ impl<'a> Response<'a> { /// ); /// ``` pub fn set(&mut self, key: &str, value: &str) -> &mut Self { - self.headers.insert(key.to_owned(), value.to_owned()); + self.set_owned(key, value.to_owned()); self } /// Get the header value by name @@ -138,29 +143,30 @@ impl<'a> Response<'a> { /// /// ``` pub fn send>(&mut self, body: T) { - let body_slice = body.as_ref(); - let len = body_slice.len(); - self.body = ResBody::Sized(len, Box::new(Cursor::new(body_slice.to_vec())) as BoxReader); - self.content_len = Some(body_slice.len()); - self.default_headers(); + self.set_body(body.as_ref().to_vec()); + } + /// Sets an owned response body without copying it. Prefer this when the + /// body is already a `Vec` or `String`. + pub fn send_owned>>(&mut self, body: T) { + self.set_body(body.into()); } /// Automatically set headers like date, content /// length, and sent content header to "text/html" /// if no content header is sent pub fn default_headers(&mut self) -> &mut Self { - if let Some(x) = self.content_len { - self.headers - .insert("Content-Length".to_string(), x.to_string()); + if let Some(content_len) = self.content_len { + self.set_owned("Content-Length", content_len.to_string()); } - if let Some(date) = Time::now() { - self.headers.insert("Date".to_string(), date.format()); + if let Some(date) = Time::now_http_date() { + self.set_owned("Date", date); } - if self.headers.get("Content-Type").is_none() { + if !self.headers.contains_key("Content-Type") { let mut format = String::from("text/html"); if let Some(charset) = &self.charset { - format.push_str(&format!(" ;charset={:?}", charset)); + write!(&mut format, " ;charset={:?}", charset) + .expect("writing a charset to a String cannot fail"); } - self.set("Content-Type", &format); + self.set_owned("Content-Type", format); } self } @@ -193,10 +199,8 @@ impl<'a> Response<'a> { /// Consume the response and get the final formed http /// response that the server will send in bytes pub fn get_data(self) -> (String, BoxReader) { - ( - format!("{}{}{}", self.status_line(), self.headers(), CRLF), - self.body.get_reader(), - ) + let (head, body) = self.into_parts(); + (head, body.get_reader()) } /// Send a file as the response, automatically detect the /// mime type and set the headers accordingly @@ -261,10 +265,13 @@ impl<'a> Response<'a> { /// ``` #[cfg(feature = "json")] pub fn json(&mut self, structure: T) -> serde_json::Result<()> { - self.body = - ResBody::Unsized(Box::new(Cursor::new(serde_json::to_vec(structure))) as BoxReader); + let body = serde_json::to_vec(&structure)?; + let len = body.len(); + self.body = ResBody::Bytes(body); + self.content_len = Some(len); self.with_type("application/json"); self.default_headers(); + Ok(()) } /// Set the status code from the status code enum /// @@ -441,10 +448,11 @@ impl<'a> Response<'a> { // Creates a new response from a slice pub(crate) fn new_from_slice>(body: T) -> Self { let body_slice = body.as_ref(); - Self::new( - Box::new(Cursor::new(body_slice.to_vec())) as BoxReader, - Some(body_slice.len()), - ) + Self::with_body(ResBody::Bytes(body_slice.to_vec()), Some(body_slice.len())) + } + // Creates a response backed by static bytes without copying the body. + pub(crate) fn new_from_static(body: &'static [u8]) -> Self { + Self::with_body(ResBody::Static(body), Some(body.len())) } // Creates a new response with empty body pub(crate) fn new_empty() -> Self { @@ -461,16 +469,10 @@ impl<'a> Response<'a> { } } // Generates a new empty response - fn new(body: BoxReader, content_len: Option) -> Self { - let body_res: ResBody; - if let Some(x) = content_len { - body_res = ResBody::Sized(x, body) - } else { - body_res = ResBody::Unsized(body) - } + fn with_body(body: ResBody, content_len: Option) -> Self { Response { status_code: StatusCode::Ok, - body: body_res, + body, content_len, http_version: "1.1".to_owned(), headers: HashMap::new(), @@ -480,26 +482,38 @@ impl<'a> Response<'a> { marker: PhantomData, } } - fn reason_phrase(&self) -> String { - self.status_code.to_string().to_uppercase() - } - fn status_code(&self) -> i32 { - self.status_code.into() - } - fn status_line(&self) -> String { - format!( - "{}/{}{}{}{}{}{}", - "HTTP", - self.http_version, - SP, - self.status_code(), - SP, - self.reason_phrase(), - CRLF - ) - } - fn headers(&self) -> String { - let mut headers_str = String::new(); + fn set_body(&mut self, body: Vec) { + let len = body.len(); + self.body = ResBody::Bytes(body); + self.content_len = Some(len); + self.default_headers(); + } + fn set_owned(&mut self, key: &str, value: String) { + if let Some(existing) = self.headers.get_mut(key) { + *existing = value; + } else { + self.headers.insert(key.to_owned(), value); + } + } + pub(crate) fn into_parts(self) -> (String, ResBody) { + let header_bytes = self + .headers + .iter() + .map(|(name, value)| name.len() + value.len() + 4) + .sum::(); + let (status_code, reason_phrase) = self.status_code.fetch(); + let mut headers_str = String::with_capacity( + 32 + self.http_version.len() + reason_phrase.len() + header_bytes, + ); + headers_str.push_str("HTTP/"); + headers_str.push_str(&self.http_version); + headers_str.push(SP); + write!(&mut headers_str, "{}", status_code) + .expect("writing a status code to a String cannot fail"); + headers_str.push(SP); + headers_str.extend(reason_phrase.chars().flat_map(char::to_uppercase)); + headers_str.push_str(CRLF); + // push normal headers self.headers .iter() @@ -511,7 +525,8 @@ impl<'a> Response<'a> { .iter() .filter_map(|cookie| cookie.serialise().ok()) .for_each(|cookie| push_header(&mut headers_str, "Set-Cookie", &cookie)); - headers_str + headers_str.push_str(CRLF); + (headers_str, self.body) } } @@ -519,17 +534,26 @@ impl<'a> Response<'a> { // characters in the value, CR and LF included, are replaced by spaces, so no // value can start another header line or end the head early. fn push_header(headers_str: &mut String, name: &str, value: &str) { - if name.is_empty() || !name.chars().all(|c| TOKEN_CHARS.contains(&c)) { + if !is_token(name) { return; } headers_str.push_str(name); headers_str.push(':'); headers_str.push(SP); - headers_str.extend( - value - .chars() - .map(|c| if is_ctl(c) && c != HT { SP } else { c }), - ); + if value + .bytes() + .any(|byte| (byte <= b'\x1f' && byte != b'\t') || byte == b'\x7f') + { + headers_str.extend(value.chars().map(|c| { + if (c <= '\x1f' && c != HT) || c == '\x7f' { + SP + } else { + c + } + })); + } else { + headers_str.push_str(value); + } headers_str.push_str(CRLF); } @@ -619,6 +643,31 @@ mod test { assert!(data_to_string(res.get_data()).await.ends_with("\r\n\r\nhe")); } + #[cfg(feature = "json")] + #[tokio::test] + async fn json_body_is_owned_and_sized() { + let mut res = Response::new_empty(); + res.json([1, 2]).unwrap(); + let data = data_to_string(res.get_data()).await; + + assert!(data.contains("\r\nContent-Length: 5\r\n")); + assert!(data.contains("\r\nContent-Type: application/json\r\n")); + assert!(data.ends_with("\r\n\r\n[1,2]")); + } + + #[test] + fn send_owned_reuses_the_body_allocation() { + let body = Vec::with_capacity(64); + let allocation = body.as_ptr(); + let mut res = Response::new_empty(); + res.send_owned(body); + + match &res.body { + ResBody::Bytes(body) => assert_eq!(body.as_ptr(), allocation), + _ => panic!("owned response body was not stored as bytes"), + } + } + #[cfg(feature = "cookies")] #[tokio::test] async fn cookies_get_their_own_header_lines() { diff --git a/src/router.rs b/src/router.rs index 4c7704e..f9364e6 100644 --- a/src/router.rs +++ b/src/router.rs @@ -1,7 +1,7 @@ use crate::default; use crate::error::InvalidPathError; use crate::middlewares::Closures; -use crate::path::{MatchedPath, PathNode}; +use crate::path::PathNode; use crate::request::{MatchedRequest, Request, RequestMethod}; use crate::responder::Response; use std::collections::HashMap; @@ -256,59 +256,83 @@ impl Router { // Fetch the closure according to the request path, run that // specific closure. - pub(crate) fn run(&self, parsed_request: Request<'_>, mut res: &mut Response) { - let req = &parsed_request.request_line; + pub(crate) fn run(&self, parsed_request: Request<'_>, res: &mut Response) { + let method = parsed_request.request_line.method; + // `MatchedPath` values borrow their input path when URL variables are + // enabled. A path-only clone lets the router move the much larger + // request (including all headers) into `MatchedRequest` unchanged. + let match_path = parsed_request.request_line.path.clone(); - let mut matches: Vec>> = Vec::new(); - if let Some(functions) = self.paths.get(&req.method) { - let mut routes = functions.get(&req.path); - routes.sort_by_key(|v| v.index); - matches.push(routes); - }; - // run RequestMethod::All regardless of the request method - if let Some(functions) = self.paths.get(&RequestMethod::All) { - let mut routes = functions.get(&req.path); - routes.sort_by_key(|v| v.index); - matches.push(routes); - } - - matches.push( - self.middlewares - .iter() - .map(|c| MatchedPath { - data: c, - #[cfg(feature = "url_variables")] - vars: HashMap::new(), - }) - .collect(), - ); + let mut method_routes = self + .paths + .get(&method) + .map_or_else(Vec::new, |functions| { + let mut routes = functions.get(&match_path); + routes.sort_unstable_by_key(|route| route.index); + routes + }) + .into_iter() + .peekable(); + // Run RequestMethod::All regardless of the request method. + let mut all_routes = self + .paths + .get(&RequestMethod::All) + .map_or_else(Vec::new, |functions| { + let mut routes = functions.get(&match_path); + routes.sort_unstable_by_key(|route| route.index); + routes + }) + .into_iter() + .peekable(); + let mut middlewares = self.middlewares.iter().peekable(); - let mut indices = vec![0_usize; matches.len()]; - let total: usize = matches.iter().map(Vec::len).sum(); #[cfg(feature = "url_variables")] let mut matched = MatchedRequest { - request: parsed_request.clone(), + request: parsed_request, vars: HashMap::new(), }; #[cfg(not(feature = "url_variables"))] let matched = MatchedRequest { - request: parsed_request.clone(), + request: parsed_request, }; - for _ in 0..total { - let mut minind = 0; - let mut minval = usize::MAX; - for (n, (v, i)) in matches.iter().zip(indices.iter()).enumerate() { - if *i < v.len() && v[*i].index < minval { - minval = v[*i].index; - minind = n; - } - } - #[cfg(feature = "url_variables")] + + loop { + let method_index = method_routes.peek().map_or(usize::MAX, |route| route.index); + let all_index = all_routes.peek().map_or(usize::MAX, |route| route.index); + let middleware_index = middlewares + .peek() + .map_or(usize::MAX, |middleware| middleware.index); + + if method_index == usize::MAX + && all_index == usize::MAX + && middleware_index == usize::MAX { - matched.vars = matches[minind][indices[minind]].vars.clone(); + break; } - let flow = (matches[minind][indices[minind]].closure)(&matched, &mut res); - indices[minind] += 1; + + // Registration indexes are globally ordered. Merge the three + // already-sorted streams directly instead of allocating a vector + // for middleware matches plus bookkeeping vectors on every request. + let flow = if method_index <= all_index && method_index <= middleware_index { + let route = method_routes.next().unwrap(); + #[cfg(feature = "url_variables")] + { + matched.vars = route.vars; + } + (&route.data.closure)(&matched, res) + } else if all_index <= middleware_index { + let route = all_routes.next().unwrap(); + #[cfg(feature = "url_variables")] + { + matched.vars = route.vars; + } + (&route.data.closure)(&matched, res) + } else { + let middleware = middlewares.next().unwrap(); + #[cfg(feature = "url_variables")] + matched.vars.clear(); + (&middleware.closure)(&matched, res) + }; if !flow.should_continue() { break; } @@ -412,7 +436,6 @@ macro_rules! route_stop { default!(Router); -#[macro_use] macro_rules! inject_method { ( $instance: expr, $path: expr, $closure: expr, $method: expr ) => { use crate::middlewares::Closures; @@ -475,6 +498,15 @@ impl Route for Router { mod test { use super::*; use crate::path::PathBuf; + use crate::request::{Headers, RequestLine}; + use std::sync::{Arc, Mutex}; + + fn record(seen: Arc>>, value: usize) -> Closure { + Box::new(move |_, _| { + seen.lock().unwrap().push(value); + Flow::Next + }) + } #[test] pub fn router_test() { @@ -581,4 +613,64 @@ mod test { .len() ); } + + #[test] + fn run_merges_routes_and_middleware_in_registration_order() { + let seen = Arc::new(Mutex::new(Vec::new())); + let mut router = Router::new(); + router.add(record(Arc::clone(&seen), 0)).unwrap(); + router.add_route("/", record(Arc::clone(&seen), 1)).unwrap(); + router.get("/", record(Arc::clone(&seen), 2)).unwrap(); + router.add(record(Arc::clone(&seen), 3)).unwrap(); + + let request = Request::parse( + RequestLine::parse("GET / HTTP/1.1").unwrap(), + Headers::parse("").unwrap(), + b"", + ) + .unwrap(); + router.run(request, &mut Response::new_empty()); + + assert_eq!(*seen.lock().unwrap(), vec![0, 1, 2, 3]); + } + + #[test] + #[cfg(feature = "url_variables")] + fn run_moves_route_variables_and_clears_them_for_middleware() { + let seen = Arc::new(Mutex::new(Vec::new())); + let mut router = Router::new(); + let route_seen = Arc::clone(&seen); + router + .get( + "/thing/:id", + Box::new(move |request, _| { + route_seen + .lock() + .unwrap() + .push(request.vars.get("id").map(|value| value.to_string())); + Flow::Next + }), + ) + .unwrap(); + let middleware_seen = Arc::clone(&seen); + router + .add(Box::new(move |request, _| { + middleware_seen + .lock() + .unwrap() + .push(request.vars.get("id").map(|value| value.to_string())); + Flow::Next + })) + .unwrap(); + + let request = Request::parse( + RequestLine::parse("GET /thing/42 HTTP/1.1").unwrap(), + Headers::parse("").unwrap(), + b"", + ) + .unwrap(); + router.run(request, &mut Response::new_empty()); + + assert_eq!(*seen.lock().unwrap(), vec![Some("42".to_owned()), None]); + } } diff --git a/src/server.rs b/src/server.rs index c0491a9..c076b86 100644 --- a/src/server.rs +++ b/src/server.rs @@ -2,22 +2,34 @@ use crate::config::{Config, OctaneConfig, Ssl}; use crate::error::Error; use crate::http::Http; use crate::middlewares::Closures; -use crate::request::{Headers, Request, RequestLine}; -use crate::responder::{BoxReader, Response}; +use crate::request::{Headers, Request}; +use crate::responder::{ResBody, Response}; use crate::router::{Closure, Route, Router, RouterResult}; use crate::server_builder::ServerBuilder; use crate::tls::AsMutStream; use crate::{declare_error, default}; -use octane_http::http1x::raw_request::RawRequest1x; use octane_http::http1x::Http1xReader; use octane_http::StatusCode; use std::error::Error as StdError; +use std::io; use std::marker::Unpin; use std::str; use std::sync::Arc; use std::time::Duration; -use tokio::io::{copy, split, AsyncWriteExt}; -use tokio::prelude::*; +use tokio::io::{copy, split, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use tokio::sync::Semaphore; + +fn checked_connection_limit(configured: usize) -> io::Result { + let configured = configured.max(1); + if configured > Semaphore::MAX_PERMITS { + Err(io::Error::new( + io::ErrorKind::InvalidInput, + "max_connections exceeds the supported semaphore limit", + )) + } else { + Ok(configured) + } +} /// The Octane server /// @@ -108,6 +120,10 @@ impl Octane { /// function also starts the Ssl server if the features /// are enabled and the key/certs are provided /// + /// The default connection deadline uses Tokio's time driver. A custom + /// runtime must call `enable_time` or `enable_all`; runtimes created by + /// `#[tokio::main]` already enable it. + /// /// # Example /// /// ```no_run @@ -126,28 +142,35 @@ impl Octane { where F: FnOnce(), { + let max_connections = checked_connection_limit(self.settings.max_connections)?; let server = Arc::new(self); + let connections = Arc::new(Semaphore::new(max_connections)); let mut _ssl = false; #[cfg(any(feature = "openSSL", feature = "rustls"))] { use crate::task; let clone = Arc::clone(&server); + let tls_connections = Arc::clone(&connections); _ssl = true; - async fn listen_ssl(server: Arc) -> Result<(), Box> { + async fn listen_ssl( + server: Arc, + connections: Arc, + ) -> Result<(), Box> { let server_builder = ServerBuilder::new(server.settings.ssl.port); server_builder? .listen_ssl( |stream, server| async { Octane::serve(stream, server).await }, server, + connections, ) .await?; Ok(()) } task!({ - if let Err(x) = listen_ssl(clone).await { + if let Err(x) = listen_ssl(clone, tls_connections).await { println!("WARNING: {}", x); } }); @@ -158,6 +181,7 @@ impl Octane { .listen( move |stream, server| async move { Octane::serve(stream, server).await }, server, + connections, ) .await?; @@ -182,33 +206,36 @@ impl Octane { let (mut reader, writer) = split(stream_async); let mut data = Vec::new(); - let mut body_vec = Vec::new(); // only http1.1x path for now :) - let raw = match Http1xReader::new(&mut reader, &mut data).await { - Ok(raw) => raw, - Err(code) => { - declare_error!(writer, code); - } - }; - - let request_line = crate::request::request_line_parse( - raw.request_method, - raw.request_url, - raw.request_version, - ); + let (request_line, headers, buffered_body_len) = { + let raw = match Http1xReader::new(&mut reader, &mut data).await { + Ok(raw) => raw, + Err(code) => { + declare_error!(writer, code); + } + }; - // headers in htt1.1x are valid utf8 strs - let headers = str::from_utf8(raw.headers) - .ok() - .and_then(|h| Headers::parse(h.to_owned())); + let buffered_body_len = raw.body_remainder.len(); + let request_line = crate::request::request_line_parse( + raw.request_method, + raw.request_url, + raw.request_version, + ); + // Headers are owned after parsing, so the raw view can end before + // the request body grows `data`. + let headers = str::from_utf8(raw.headers).ok().and_then(Headers::parse); + (request_line, headers, buffered_body_len) + }; + // The raw view has ended, so `data` can be borrowed mutably again. + let body_start = data.len() - buffered_body_len; if let (Some(request_line), Some(headers)) = (request_line, headers) { let request = match Request::from_raw( - &headers, + headers, request_line, - raw.body_remainder, - &mut body_vec, + &mut data, + body_start, reader, server.settings.max_body_size, ) @@ -219,21 +246,22 @@ impl Octane { declare_error!(writer, code); } }; - let request_line = &request.request_line; let mut res = Response::new_empty(); // Detect http version and validate - let checker = Http::validate(&request); - if checker.is_malformed() { - declare_error!(writer, checker.err_code.unwrap()); + { + let checker = Http::validate(&request); + if checker.is_malformed() { + declare_error!(writer, checker.err_code.unwrap()); + } } - if request_line.method.is_some() { + if request.request_line.method.is_some() { // run closures - server.router.run(request.clone(), &mut res); + server.router.run(request, &mut res); if !res.has_body() { declare_error!(writer, StatusCode::NotFound); } - Octane::send(res.get_data(), writer).await?; + Octane::send(res.into_parts(), writer).await?; } else { declare_error!(writer, StatusCode::NotImplemented); } @@ -248,14 +276,22 @@ impl Octane { self.router.run(request, res); } pub(crate) async fn send( - mut response: (String, BoxReader), + response: (String, ResBody), mut stream_async: S, ) -> Result<(), Box> where S: AsyncWrite + Unpin, { stream_async.write_all(response.0.as_bytes()).await?; - copy(&mut response.1, &mut stream_async).await?; + match response.1 { + ResBody::None => {} + ResBody::Bytes(body) => stream_async.write_all(&body).await?, + ResBody::Static(body) => stream_async.write_all(body).await?, + ResBody::Sized(len, reader) => { + let mut reader = reader.take(len as u64); + copy(&mut reader, &mut stream_async).await?; + } + } Ok(()) } } @@ -307,3 +343,19 @@ impl Config for Octane { &mut self.settings.ssl } } + +#[cfg(test)] +mod tests { + use super::checked_connection_limit; + use tokio::sync::Semaphore; + + #[test] + fn connection_limit_is_validated_before_building_the_semaphore() { + assert_eq!(checked_connection_limit(0).unwrap(), 1); + assert_eq!( + checked_connection_limit(Semaphore::MAX_PERMITS).unwrap(), + Semaphore::MAX_PERMITS + ); + assert!(checked_connection_limit(usize::MAX).is_err()); + } +} diff --git a/src/server_builder.rs b/src/server_builder.rs index a2eda80..560c14e 100644 --- a/src/server_builder.rs +++ b/src/server_builder.rs @@ -1,12 +1,14 @@ use crate::server::Octane; use crate::task; -use socket2::{Domain, Protocol, SockAddr, Socket, Type}; +use socket2::{Domain, SockAddr, Socket, Type}; use std::future::Future; use std::io::Result; use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::sync::Arc; +use std::time::Duration; use tokio::net::{TcpListener, TcpStream}; -use tokio::stream::StreamExt; +use tokio::sync::{OwnedSemaphorePermit, Semaphore}; +use tokio::time::{sleep, timeout}; #[cfg(feature = "openSSL")] use tokio_openssl::SslStream; #[cfg(feature = "rustls")] @@ -16,10 +18,53 @@ pub struct ServerBuilder { socket: TcpListener, } +const ACCEPT_ERROR_BACKOFF: Duration = Duration::from_millis(50); + +async fn reserve_connection(connections: &Arc) -> OwnedSemaphorePermit { + Arc::clone(connections) + .acquire_owned() + .await + .expect("connection semaphore is never closed") +} + +async fn accept_bounded( + listener: &TcpListener, + connections: &Arc, +) -> (TcpStream, OwnedSemaphorePermit) { + loop { + match listener.accept().await { + Ok((stream, _)) => { + // Only one socket per listener can wait here. Further sockets + // stay in the kernel backlog until serving capacity returns. + // Acquiring after accept also prevents an idle TLS listener + // from reserving capacity needed by the plain listener (and + // vice versa). + let permit = reserve_connection(connections).await; + return (stream, permit); + } + Err(_) => { + // An intermittent accept error must not take down the server; + // back off on persistent failures to avoid a hot error loop. + sleep(ACCEPT_ERROR_BACKOFF).await; + } + } + } +} + +async fn run_with_timeout(future: F, connection_timeout: Option) +where + F: Future, +{ + if let Some(duration) = connection_timeout { + let _ = timeout(duration, future).await; + } else { + future.await; + } +} + impl ServerBuilder { pub fn new(port: u16) -> Result { - let stream = Type::stream(); - let socket = Socket::new(Domain::ipv4(), stream, None)?; + let socket = Socket::new(Domain::IPV4, Type::STREAM, None)?; socket.set_nonblocking(true)?; // must come before bind, or a restart fails while old connections // are still in TIME_WAIT @@ -27,50 +72,66 @@ impl ServerBuilder { let bind_add = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), port); socket.bind(&SockAddr::from(bind_add))?; socket.listen(2048)?; + let listener: std::net::TcpListener = socket.into(); Ok(ServerBuilder { - socket: TcpListener::from_std(socket.into_tcp_listener())?, + socket: TcpListener::from_std(listener)?, }) } - pub async fn listen(mut self, exec: C, server: Arc) -> Result<()> + pub async fn listen( + self, + exec: C, + server: Arc, + connections: Arc, + ) -> Result<()> where - T: Future + Send, + T: Future + Send + 'static, C: FnOnce(TcpStream, Arc) -> T + Send + 'static + Copy, { - while let Some(stream) = self.socket.next().await { - stream.map(|stream| { - let server = Arc::clone(&server); - task!({ - exec(stream, server).await; - }) - })?; + loop { + let (stream, permit) = accept_bounded(&self.socket, &connections).await; + let server = Arc::clone(&server); + let connection_timeout = server.settings.connection_timeout; + task!({ + let _permit = permit; + run_with_timeout(exec(stream, server), connection_timeout).await; + }); } - Ok(()) } #[cfg(feature = "openSSL")] - pub async fn listen_ssl(self, exec: C, server: Arc) -> Result<()> + pub async fn listen_ssl( + self, + exec: C, + server: Arc, + connections: Arc, + ) -> Result<()> where - T: Future + Send, + T: Future + Send + 'static, C: FnOnce(SslStream, Arc) -> T + Send + 'static + Copy, { - let mut ssl_listener = self.socket; let acceptor = crate::tls::openssl::acceptor(&server.settings)?; - while let Some(stream) = ssl_listener.next().await { + loop { + let (tcp_stream, permit) = accept_bounded(&self.socket, &connections).await; let acceptor = acceptor.clone(); - let tcp_stream = stream?; let server = Arc::clone(&server); + let connection_timeout = server.settings.connection_timeout; task!({ - let stream = tokio_openssl::accept(&acceptor, tcp_stream).await; - if let Ok(stream_ssl) = stream { - exec(stream_ssl, server).await; - } else { - stream.map_err(|e| println!("{:?}", e)).err(); - } + let _permit = permit; + run_with_timeout( + async move { + if let Ok(stream_ssl) = + crate::tls::openssl::accept(&acceptor, tcp_stream).await + { + exec(stream_ssl, server).await; + } + }, + connection_timeout, + ) + .await; }); } - Ok(()) } #[cfg(feature = "rustls")] @@ -78,30 +139,80 @@ impl ServerBuilder { self, exec: C, server: Arc, + connections: Arc, ) -> std::result::Result<(), Box> where - T: Future + Send, + T: Future + Send + 'static, C: FnOnce(TlsStream, Arc) -> T + Send + 'static + Copy, { - let mut ssl_listener = self.socket; let acceptor = crate::tls::rustls::acceptor(&server.settings)?; - while let Some(stream) = ssl_listener.next().await { + loop { + let (tcp_stream, permit) = accept_bounded(&self.socket, &connections).await; let acceptor = acceptor.clone(); - let server = Arc::clone(&server); + let connection_timeout = server.settings.connection_timeout; - let tcp_stream = stream?; task!({ - let stream = acceptor.accept(tcp_stream).await; - if let Ok(stream_ssl) = stream { - exec(stream_ssl, server).await; - } else { - stream - .map_err(|e| println!("WARNING: {:?}", e.kind())) - .err(); - } + let _permit = permit; + run_with_timeout( + async move { + if let Ok(stream_ssl) = acceptor.accept(tcp_stream).await { + exec(stream_ssl, server).await; + } + }, + connection_timeout, + ) + .await; }); } - Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::{reserve_connection, run_with_timeout}; + use std::future::pending; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; + use std::time::Duration; + use tokio::sync::Semaphore; + use tokio::time::timeout; + + #[tokio::test] + async fn connection_capacity_blocks_until_released() { + let connections = Arc::new(Semaphore::new(1)); + let held_permit = reserve_connection(&connections).await; + let task_connections = Arc::clone(&connections); + + let mut waiting = tokio::spawn(async move { reserve_connection(&task_connections).await }); + + assert!(timeout(Duration::from_millis(25), &mut waiting) + .await + .is_err()); + drop(held_permit); + + let permit = timeout(Duration::from_secs(1), waiting).await.unwrap(); + assert!(permit.is_ok()); + } + + #[tokio::test] + async fn timeout_cancels_connection_work() { + struct DropFlag(Arc); + + impl Drop for DropFlag { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } + } + + let dropped = Arc::new(AtomicBool::new(false)); + let task_flag = Arc::clone(&dropped); + let work = async move { + let _drop_flag = DropFlag(task_flag); + pending::<()>().await; + }; + + run_with_timeout(work, Some(Duration::from_millis(10))).await; + assert!(dropped.load(Ordering::SeqCst)); } } diff --git a/src/time.rs b/src/time.rs index 2866d0f..a1a04d9 100644 --- a/src/time.rs +++ b/src/time.rs @@ -1,9 +1,18 @@ use crate::constants::{ DAYS_IN_MONTH, DAYS_PER_100Y, DAYS_PER_400Y, DAYS_PER_4Y, LEAPOCH, MONTHS, SP, WEEKS, }; +use std::cell::RefCell; use std::convert::TryInto; +use std::fmt::Write; use std::time::{SystemTime, UNIX_EPOCH}; +thread_local! { + // HTTP dates have one-second precision. Keeping the formatted value per + // worker thread avoids recalculating the calendar for every response and + // needs neither a lock nor another dependency. + static HTTP_DATE_CACHE: RefCell> = RefCell::new(None); +} + pub(crate) struct Time { min: i64, sec: i64, @@ -14,86 +23,67 @@ pub(crate) struct Time { day: i64, } -trait Padding { - fn pad_zero(self) -> String; -} - -impl Padding for i64 { - fn pad_zero(self) -> String { - let day_string = format!("{}{}", 0, self); - (&day_string[day_string.len() - 2..]).to_owned() - } -} - impl Time { - pub(crate) fn now() -> Option { - let stamp = match SystemTime::now().duration_since(UNIX_EPOCH) { - Ok(n) => { - if let Ok(x) = n.as_secs().try_into() { - x - } else { - return None; + pub(crate) fn now_http_date() -> Option { + let stamp = Self::unix_timestamp()?; + HTTP_DATE_CACHE.with(|cache| { + let mut cache = cache.borrow_mut(); + if let Some((cached_stamp, cached_date)) = cache.as_ref() { + if *cached_stamp == stamp { + return Some(cached_date.clone()); } } - Err(m) => panic!("Got time error: {}", m), - }; - Self::time(stamp) + let stamp_i64 = stamp.try_into().ok()?; + let date = Self::time(stamp_i64)?.format(); + *cache = Some((stamp, date.clone())); + Some(date) + }) + } + fn unix_timestamp() -> Option { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .ok() + .map(|duration| duration.as_secs()) } pub(crate) fn format(&self) -> String { - if let Some(month_day) = self.month_day() { - if let Some(week_day) = self.week_day() { - let date = format!( - "{}{}{}{}{}", - self.day.pad_zero(), - SP, - month_day, - SP, - self.year - ); - let time = format!( - "{}:{}:{}", - self.hour.pad_zero(), - self.min.pad_zero(), - self.sec.pad_zero() - ); - format!("{},{}{}{}{}{}GMT", week_day, SP, date, SP, time, SP) - } else { - String::new() - } - } else { - String::new() - } + let (month_day, week_day) = match (self.month_day(), self.week_day()) { + (Some(month_day), Some(week_day)) => (month_day, week_day), + _ => return String::new(), + }; + // An IMF-fixdate is always 29 ASCII bytes. Reserving it once avoids the + // temporary strings previously created for every individual field. + let mut output = String::with_capacity(29); + write!( + &mut output, + "{},{}{:02}{}{}{}{}{}{:02}:{:02}:{:02}{}GMT", + week_day, + SP, + self.day, + SP, + month_day, + SP, + self.year, + SP, + self.hour, + self.min, + self.sec, + SP + ) + .expect("writing a date to a String cannot fail"); + output } /* pub fn with_stamp(self, stamp: i64) -> Option { Self::time(stamp) } */ - fn week_day(&self) -> Option { - if let Some(week) = WEEKS.iter().enumerate().find(|(i, _)| { - if let Ok(week_day) = self.week.try_into() { - *i == week_day - } else { - false - } - }) { - Some(week.1.to_string()) - } else { - None - } + fn week_day(&self) -> Option<&'static str> { + let week: usize = self.week.try_into().ok()?; + WEEKS.get(week).copied() } - fn month_day(&self) -> Option { - if let Some(month) = MONTHS.iter().enumerate().find(|(i, _)| { - if let Ok(month_day) = self.month.try_into() { - *i == month_day - } else { - false - } - }) { - Some(month.1.to_string()) - } else { - None - } + fn month_day(&self) -> Option<&'static str> { + let month: usize = self.month.try_into().ok()?; + MONTHS.get(month).copied() } pub(crate) fn time(stamp: i64) -> Option { diff --git a/src/tls/openssl.rs b/src/tls/openssl.rs index 824b346..a14b165 100644 --- a/src/tls/openssl.rs +++ b/src/tls/openssl.rs @@ -1,8 +1,10 @@ #![cfg(feature = "openSSL")] use crate::config::OctaneConfig; use crate::tls::AsMutStream; -use openssl::ssl::{SslAcceptor, SslFiletype, SslMethod}; +use openssl::ssl::{Ssl, SslAcceptor, SslFiletype, SslMethod}; +use std::error::Error; use std::io::Result; +use std::pin::Pin; use tokio::net::TcpStream; use tokio_openssl::SslStream; @@ -15,6 +17,17 @@ pub fn acceptor(settings: &OctaneConfig) -> Result { Ok(acceptor) } +/// Wrap a Tokio stream and complete its server-side OpenSSL handshake. +pub async fn accept( + acceptor: &SslAcceptor, + stream: TcpStream, +) -> std::result::Result, Box> { + let ssl = Ssl::new(acceptor.context())?; + let mut stream = SslStream::new(ssl, stream)?; + Pin::new(&mut stream).accept().await?; + Ok(stream) +} + impl AsMutStream for SslStream { fn stream_mut(&mut self) -> &mut TcpStream { self.get_mut() diff --git a/src/tls/rustls.rs b/src/tls/rustls.rs index 1a72d28..5b82689 100644 --- a/src/tls/rustls.rs +++ b/src/tls/rustls.rs @@ -6,20 +6,15 @@ use std::sync::Arc; use tokio::net::TcpStream; use tokio_rustls::server::TlsStream; use tokio_rustls::{ - rustls::{NoClientAuth, ServerConfig}, + rustls::{crypto::ring, ServerConfig}, TlsAcceptor, }; pub fn acceptor(settings: &OctaneConfig) -> Result> { - let mut config = ServerConfig::new(NoClientAuth::new()); - let mut key = settings.get_key()?; - if key.get(0).is_none() { - panic!( - "{:?}", - "rustls expects a RSA_PRIVATE_KEY, invalid key provided" - ); - } - config.set_single_cert(settings.get_cert()?, key.remove(0))?; + let config = ServerConfig::builder_with_provider(Arc::new(ring::default_provider())) + .with_safe_default_protocol_versions()? + .with_no_client_auth() + .with_single_cert(settings.get_cert()?, settings.get_key()?)?; // config.set_protocols(&[b"h2".to_vec()]); let acceptor = TlsAcceptor::from(Arc::new(config)); Ok(acceptor)