Harden native request body ownership

Author: eldryoth

Cargo.lock

@@ -8,3 +8,4 @@ publish = false
 description = "Internal Fluxheim protocol framing helpers."
 
 [dependencies]
+sanitization = { version = "1.2.4", features = ["alloc"] }

crates/fluxheim-protocol/Cargo.toml

@@ -54,7 +54,7 @@ enum Http1ChunkedDecodeState {
     End,
 }
 
-#[derive(Clone, Debug, Eq, PartialEq)]
+#[derive(Debug, Eq, PartialEq)]
 pub struct Http1ChunkedDecoder {
     limits: Http1ChunkLimits,
     input: Vec<u8>,
@@ -97,6 +97,17 @@ impl Http1ChunkedDecoder {
         self.push_with_sink(chunk, output)
     }
 
+    pub fn push_to<S: Http1ChunkSink>(
+        &mut self,
+        chunk: &[u8],
+        output: &mut S,
+    ) -> Result<Option<Http1ChunkedDecode>, Http1ParseError> {
+        if output.len() != self.decoded_len {
+            return Err(Http1ParseError::InvalidChunk);
+        }
+        self.push_with_sink(chunk, output)
+    }
+
     pub fn buffered_len(&self) -> usize {
         self.input.len()
     }
@@ -272,11 +283,27 @@ impl Http1ChunkedDecoder {
     }
 }
 
-trait Http1ChunkSink {
+impl Drop for Http1ChunkedDecoder {
+    fn drop(&mut self) {
+        sanitization::unsafe_wipe::volatile_sanitize_vec(&mut self.input);
+    }
+}
+
+pub trait Http1ChunkSink {
+    fn len(&self) -> usize;
+
+    fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
     fn append(&mut self, bytes: &[u8]) -> Result<(), Http1ParseError>;
 }
 
 impl Http1ChunkSink for Vec<u8> {
+    fn len(&self) -> usize {
+        Vec::len(self)
+    }
+
     fn append(&mut self, bytes: &[u8]) -> Result<(), Http1ParseError> {
         self.try_reserve(bytes.len())
             .map_err(|_| Http1ParseError::BodyTooLarge)?;
@@ -291,6 +318,10 @@ struct Http1SliceChunkSink<'a> {
 }
 
 impl Http1ChunkSink for Http1SliceChunkSink<'_> {
+    fn len(&self) -> usize {
+        self.len
+    }
+
     fn append(&mut self, bytes: &[u8]) -> Result<(), Http1ParseError> {
         let end = self
             .len

crates/fluxheim-protocol/src/http1_chunked.rs

@@ -29,7 +29,7 @@ pub use http1_chunked::{
     DEFAULT_HTTP1_MAX_BODY_BYTES, DEFAULT_HTTP1_MAX_CHUNK_COUNT,
     DEFAULT_HTTP1_MAX_CHUNK_EXTENSION_BYTES, DEFAULT_HTTP1_MAX_CHUNK_LINE_BYTES,
     DEFAULT_HTTP1_MAX_CHUNK_SIZE, DEFAULT_HTTP1_MAX_ENCODED_BYTES, Http1ChunkLimits,
-    Http1ChunkedDecode, Http1ChunkedDecoder, decode_http1_chunked_body,
+    Http1ChunkSink, Http1ChunkedDecode, Http1ChunkedDecoder, decode_http1_chunked_body,
 };
 pub use http1_connection::{
     Http1ConnectionDirective, Http1ConnectionOptions, http1_connection_directive,

crates/fluxheim-protocol/src/lib.rs

@@ -119,7 +119,7 @@ pub use native_http1::serve_native_http1_rustls_listener;
 pub use native_http1::serve_native_http1_unix_listener;
 pub use native_http1::{
     NativeHttp1ConnectionStream, NativeHttp1Error, NativeHttp1GeoContext, NativeHttp1Handler,
-    NativeHttp1Request, NativeHttp1RequestContext, NativeHttp1Response,
+    NativeHttp1Request, NativeHttp1RequestBody, NativeHttp1RequestContext, NativeHttp1Response,
     NativeHttp1ResponseWritePolicy, NativeHttp1TlsClientIdentity, serve_native_http1_connection,
     serve_native_http1_listener, serve_native_http1_listener_with_proxy_protocol,
 };

crates/fluxheim-server/src/lib.rs

@@ -10,7 +10,6 @@ use fluxheim_protocol::{
 };
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite};
 use tokio::time::timeout;
-use zeroize::Zeroizing;
 
 #[cfg(any(feature = "tls-rustls-backend", feature = "tls-openssl-backend"))]
 use crate::DownstreamHttp2Policy;
@@ -38,7 +37,7 @@ pub use openssl_listener::{
 };
 use proxy_protocol::read_proxy_protocol_source;
 pub use request::{
-    NativeHttp1GeoContext, NativeHttp1Request, NativeHttp1RequestContext,
+    NativeHttp1GeoContext, NativeHttp1Request, NativeHttp1RequestBody, NativeHttp1RequestContext,
     NativeHttp1TlsClientIdentity,
 };
 use response::write_response;
@@ -322,7 +321,7 @@ where
             Err(error) => return Err(error),
         };
         let mut request = request;
-        request.body = Zeroizing::new(body);
+        request.body = body;
         request_handler.prepare_request_context(&mut request);
         if request_handler.handles_connection_takeover(&request) {
             let prebuffered = std::mem::take(&mut buffer);
@@ -445,7 +444,7 @@ fn owned_request_from_head(
         target: head.target().to_owned(),
         version: head.version(),
         headers: owned_headers(head.headers(), head.effective_authority()),
-        body: Zeroizing::new(Vec::new()),
+        body: NativeHttp1RequestBody::empty(),
         trailers: Vec::new(),
     }
 }

crates/fluxheim-server/src/native_http1.rs

@@ -5,7 +5,7 @@ use tokio::io::{AsyncRead, AsyncReadExt};
 use tokio::time::timeout;
 
 use crate::request_body_budget::NativeRequestBodyReservation;
-use crate::{DownstreamHttp1Policy, NativeHttp1Error};
+use crate::{DownstreamHttp1Policy, NativeHttp1Error, NativeHttp1RequestBody};
 
 const READ_CHUNK_BYTES: usize = 8192;
 
@@ -17,7 +17,7 @@ pub(super) async fn read_body<S>(
     head_len: usize,
     framing: Http1BodyFraming,
     reservation: &mut NativeRequestBodyReservation,
-) -> Result<Vec<u8>, NativeHttp1Error>
+) -> Result<NativeHttp1RequestBody, NativeHttp1Error>
 where
     S: AsyncRead + Unpin,
 {
@@ -48,33 +48,45 @@ async fn read_body_inner<S>(
     framing: Http1BodyFraming,
     max_body_bytes: usize,
     reservation: &mut NativeRequestBodyReservation,
-) -> Result<Vec<u8>, NativeHttp1Error>
+) -> Result<NativeHttp1RequestBody, NativeHttp1Error>
 where
     S: AsyncRead + Unpin,
 {
     match framing {
         Http1BodyFraming::NoBody => {
+            sanitization::unsafe_wipe::volatile_sanitize_bytes(&mut buffer[..head_len]);
             buffer.drain(..head_len);
-            Ok(Vec::new())
+            Ok(NativeHttp1RequestBody::empty())
         }
         Http1BodyFraming::ContentLength(length) => {
             let length = usize::try_from(length).map_err(|_| Http1ParseError::BodyTooLarge)?;
             if length > max_body_bytes {
                 return Err(Http1ParseError::BodyTooLarge.into());
             }
-            let required = head_len
-                .checked_add(length)
+            let available = buffer.len().saturating_sub(head_len).min(length);
+            let body_end = head_len
+                .checked_add(available)
                 .ok_or(Http1ParseError::BodyTooLarge)?;
-            while buffer.len() < required {
+            let mut initial = std::mem::take(buffer);
+            let remainder = initial.split_off(body_end);
+            sanitization::unsafe_wipe::volatile_sanitize_bytes(&mut initial[..head_len]);
+            initial.drain(..head_len);
+            *buffer = remainder;
+            let mut body = NativeHttp1RequestBody::from_vec(initial);
+            body.replace_capacity(length)?;
+            while body.len() < length {
+                let remaining = length - body.len();
                 let mut chunk = [0u8; READ_CHUNK_BYTES];
-                let read = stream.read(&mut chunk).await?;
+                let read = stream
+                    .read(&mut chunk[..remaining.min(READ_CHUNK_BYTES)])
+                    .await?;
                 if read == 0 {
                     return Err(Http1ParseError::InvalidContentLength.into());
                 }
-                buffer.extend_from_slice(&chunk[..read]);
+                let append_result = body.extend_from_slice(&chunk[..read]);
+                sanitization::unsafe_wipe::volatile_sanitize_bytes(&mut chunk[..read]);
+                append_result?;
             }
-            let body = buffer[head_len..required].to_vec();
-            buffer.drain(..required);
             Ok(body)
         }
         Http1BodyFraming::Chunked => {
@@ -89,19 +101,17 @@ async fn read_chunked_body<S>(
     head_len: usize,
     max_body_bytes: usize,
     reservation: &mut NativeRequestBodyReservation,
-) -> Result<Vec<u8>, NativeHttp1Error>
+) -> Result<NativeHttp1RequestBody, NativeHttp1Error>
 where
     S: AsyncRead + Unpin,
 {
     buffer.drain(..head_len);
     let limits = Http1ChunkLimits::default().with_max_body_bytes(max_body_bytes);
     let mut decoder = Http1ChunkedDecoder::new(limits);
-    let mut body = Vec::new();
-    let initial = std::mem::take(buffer);
-    reservation
-        .grow_to(initial.len().min(max_body_bytes))
-        .await?;
-    if let Some(decoded) = decoder.push(&initial, &mut body)? {
+    let mut body = NativeHttp1RequestBody::empty();
+    let initial = NativeHttp1RequestBody::from_vec(std::mem::take(buffer));
+    reserve_chunked_body_growth(&mut body, reservation, initial.len(), max_body_bytes).await?;
+    if let Some(decoded) = decoder.push_to(&initial, &mut body)? {
         buffer.extend_from_slice(&initial[decoded.consumed_len..]);
         return Ok(body);
     }
@@ -115,10 +125,10 @@ where
         let prospective_encoded = fed_len
             .checked_add(read)
             .ok_or(Http1ParseError::EncodedBodyTooLarge)?;
-        reservation
-            .grow_to(prospective_encoded.min(max_body_bytes))
-            .await?;
-        if let Some(decoded) = decoder.push(&chunk[..read], &mut body)? {
+        reserve_chunked_body_growth(&mut body, reservation, read, max_body_bytes).await?;
+        let decoded = decoder.push_to(&chunk[..read], &mut body);
+        sanitization::unsafe_wipe::volatile_sanitize_bytes(&mut chunk[..read]);
+        if let Some(decoded) = decoded? {
             let consumed_from_chunk = decoded
                 .consumed_len
                 .checked_sub(fed_len)
@@ -132,3 +142,58 @@ where
         fed_len = prospective_encoded;
     }
 }
+
+async fn reserve_chunked_body_growth(
+    body: &mut NativeHttp1RequestBody,
+    reservation: &mut NativeRequestBodyReservation,
+    possible_growth: usize,
+    max_body_bytes: usize,
+) -> Result<(), NativeHttp1Error> {
+    let required = body
+        .len()
+        .checked_add(possible_growth)
+        .ok_or(Http1ParseError::BodyTooLarge)?
+        .min(max_body_bytes);
+    if required <= body.capacity() {
+        return Ok(());
+    }
+    let geometric = body.capacity().max(READ_CHUNK_BYTES).saturating_mul(2);
+    let admitted_capacity = required.max(geometric).min(max_body_bytes);
+    reservation.grow_to(admitted_capacity).await?;
+    body.reserve_capacity(admitted_capacity)?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::NativeRequestBodyBudget;
+
+    #[tokio::test]
+    async fn content_length_moves_body_and_releases_connection_buffer_capacity() {
+        let head = b"POST / HTTP/1.1\r\nHost: local.test\r\nContent-Length: 5\r\n\r\n";
+        let pipeline = b"GET /next HTTP/1.1\r\nHost: local.test\r\n\r\n";
+        let mut buffer = Vec::with_capacity(1024 * 1024);
+        buffer.extend_from_slice(head);
+        buffer.extend_from_slice(b"hello");
+        buffer.extend_from_slice(pipeline);
+        let original_capacity = buffer.capacity();
+        let mut reservation = NativeRequestBodyBudget::new(1024).reserve(5).await.unwrap();
+
+        let body = read_body_inner(
+            &mut tokio::io::empty(),
+            &mut buffer,
+            head.len(),
+            Http1BodyFraming::ContentLength(5),
+            1024,
+            &mut reservation,
+        )
+        .await
+        .unwrap();
+
+        assert_eq!(body.as_ref(), b"hello");
+        assert_eq!(buffer, pipeline);
+        assert!(buffer.capacity() < original_capacity);
+        assert!(body.capacity() >= body.len());
+    }
+}