Repository: rustls/rustls (language: rust)

Issue:
unbuffered: No way to refuse a ReadTraffic if there is no space
**Checklist**
* [x] I've searched the issue tracker for similar bugs.

**Describe the bug**

The scenario is that Rustls unbuffered is being used between four fixed-sized buffers (in/out on each side), each of which might have a varying level of fill depending on downstream/upstream processing. So depending on the size of the record being decoded, there might not be space to output the results. In simple terms both data to decode and space to put the result must be present before a Rustls operation can occur successfully. 

ReadTraffic appears to support this in terms of having a `peek_len` method to get the length that is required. If there is not space, presumably the intention is that I should not call `next_record` and I shouldn't apply the `discard`.  However even if `next_record` is not called, the operation already seems to have been committed to, and the data is lost.

**To Reproduce**

The code I have to reproduce this is in a couple of crate versions which aren't published. I'd have to give you a TGZ.

**Applicable Version(s)**

0.23.10 on Linux

**Expected behavior**

If the intention of unbuffered is to allow working with fixed-size buffers, then there always needs to be a way to refuse an operation if the caller doesn't have enough space to store the result. So I'd expect that if I didn't commit (via `next_record` or whatever) then the operation wouldn't be commited internally in Rustls either. But it's not the details of the API that are important, but just having the means to handle the situation. If there was some other way to check how much space is required at some other point, e.g. before calling `process_tls_records`, that would also be okay. Even something like "don't call it unless you have 32KB free in both outgoing buffers". That's more wasteful but could also work. 

**Additional context**

I think it may be worth putting unbuffered under a cargo feature (e.g. "beta") and specifying that it isn't under semver, to give you some flexibility to iterate until things are more solid.

I also have some difficulties with WriteTraffic, but more solvable. Since in this scenario my output buffers may be fixed-size but not fully allocated, I don't want to reallocate the buffers to the maximum unless really necessary. So I need to estimate how much space you require for the `WriteTraffic::encrypt` call. So I'm estimating 100 bytes more than the unencrypted data length. This mostly works, but sometimes I was seeing much larger requirements in terms of InsufficientSize being returned. It seems that other stuff may get included in the write unrelated to the data I'm sending, which I can't account for in my estimate. 

Also, when my output space is limited, I was attempting to send data 100 bytes shorter than the space I have free. But this may be complicated by the other stuff that Rustls wants to send. I tried to back off if I get a failure, and send zero bytes to a zero-sized buffer get a measurement of the hidden data that needs sending. However I have not yet fully debugged this code because of the issues on the ReadTraffic side.

So it may be helpful to have better documentation on buffer sizing requirements and the expectations that Rustls has of the caller.

Again, as with the other issue (#1895) there is no rush. I'm trying to implement this to test the feasibility of running with fixed-sized buffers. But I don't have a deadline.



Discussion / hints:

Fix patch:
diff --git a/rustls/src/conn/unbuffered.rs b/rustls/src/conn/unbuffered.rs
index e3fe4f6e7c..59c5a51fb5 100644
--- a/rustls/src/conn/unbuffered.rs
+++ b/rustls/src/conn/unbuffered.rs
@@ -19,7 +19,7 @@ impl UnbufferedConnectionCommon<ClientConnectionData> {
         &'c mut self,
         incoming_tls: &'i mut [u8],
     ) -> UnbufferedStatus<'c, 'i, ClientConnectionData> {
-        self.process_tls_records_common(incoming_tls, |_| None, |_, _, ()| unreachable!())
+        self.process_tls_records_common(incoming_tls, |_| false, |_, _| unreachable!())
     }
 }
 
@@ -32,36 +32,39 @@ impl UnbufferedConnectionCommon<ServerConnectionData> {
     ) -> UnbufferedStatus<'c, 'i, ServerConnectionData> {
         self.process_tls_records_common(
             incoming_tls,
-            |conn| conn.pop_early_data(),
-            |conn, incoming_tls, chunk| ReadEarlyData::new(conn, incoming_tls, chunk).into(),
+            |conn| conn.peek_early_data().is_some(),
+            |conn, incoming_tls| ReadEarlyData::new(conn, incoming_tls).into(),
         )
     }
 }
 
 impl<Data> UnbufferedConnectionCommon<Data> {
-    fn process_tls_records_common<'c, 'i, T>(
+    fn process_tls_records_common<'c, 'i>(
         &'c mut self,
         incoming_tls: &'i mut [u8],
-        mut check: impl FnMut(&mut Self) -> Option<T>,
-        execute: impl FnOnce(&'c mut Self, &'i mut [u8], T) -> ConnectionState<'c, 'i, Data>,
+        mut early_data_available: impl FnMut(&mut Self) -> bool,
+        early_data_state: impl FnOnce(&'c mut Self, &'i mut [u8]) -> ConnectionState<'c, 'i, Data>,
     ) -> UnbufferedStatus<'c, 'i, Data> {
         let mut buffer = DeframerSliceBuffer::new(incoming_tls);
         let mut buffer_progress = self.core.hs_deframer.progress();
 
         let (discard, state) = loop {
-            if let Some(value) = check(self) {
-                break (buffer.pending_discard(), execute(self, incoming_tls, value));
+            if early_data_available(self) {
+                break (
+                    buffer.pending_discard(),
+                    early_data_state(self, incoming_tls),
+                );
             }
 
-            if let Some(chunk) = self
+            if !self
                 .core
                 .common_state
                 .received_plaintext
-                .pop()
+                .is_empty()
             {
                 break (
                     buffer.pending_discard(),
-                    ReadTraffic::new(self, incoming_tls, chunk).into(),
+                    ReadTraffic::new(self, incoming_tls).into(),
                 );
             }
 
@@ -316,101 +319,97 @@ impl<Data> fmt::Debug for ConnectionState<'_, '_, Data> {
 
 /// Application data is available
 pub struct ReadTraffic<'c, 'i, Data> {
-    _conn: &'c mut UnbufferedConnectionCommon<Data>,
+    conn: &'c mut UnbufferedConnectionCommon<Data>,
     // for forwards compatibility; to support in-place decryption in the future
     _incoming_tls: &'i mut [u8],
-    chunk: Vec<u8>,
-    taken: bool,
+
+    // owner of the latest chunk obtained in `next_record`, as borrowed by
+    // `AppDataRecord`
+    chunk: Option<Vec<u8>>,
 }
 
 impl<'c, 'i, Data> ReadTraffic<'c, 'i, Data> {
-    fn new(
-        _conn: &'c mut UnbufferedConnectionCommon<Data>,
-        _incoming_tls: &'i mut [u8],
-        chunk: Vec<u8>,
-    ) -> Self {
+    fn new(conn: &'c mut UnbufferedConnectionCommon<Data>, _incoming_tls: &'i mut [u8]) -> Self {
         Self {
-            _conn,
+            conn,
             _incoming_tls,
-            chunk,
-            taken: false,
+            chunk: None,
         }
     }
 
     /// Decrypts and returns the next available app-data record
     // TODO deprecate in favor of `Iterator` implementation, which requires in-place decryption
     pub fn next_record(&mut self) -> Option<Result<AppDataRecord<'_>, Error>> {
-        if self.taken {
-            None
-        } else {
-            self.taken = true;
-            Some(Ok(AppDataRecord {
+        self.chunk = self
+            .conn
+            .core
+            .common_state
+            .received_plaintext
+            .pop();
+        self.chunk.as_ref().map(|chunk| {
+            Ok(AppDataRecord {
                 discard: 0,
-                payload: &self.chunk,
-            }))
-        }
+                payload: chunk,
+            })
+        })
     }
 
     /// Returns the payload size of the next app-data record *without* decrypting it
     ///
     /// Returns `None` if there are no more app-data records
     pub fn peek_len(&self) -> Option<NonZeroUsize> {
-        if self.taken {
-            None
-        } else {
-            NonZeroUsize::new(self.chunk.len())
-        }
+        self.conn
+            .core
+            .common_state
+            .received_plaintext
+            .peek()
+            .and_then(|ch| NonZeroUsize::new(ch.len()))
     }
 }
 
 /// Early application-data is available.
 pub struct ReadEarlyData<'c, 'i, Data> {
-    _conn: &'c mut UnbufferedConnectionCommon<Data>,
+    conn: &'c mut UnbufferedConnectionCommon<Data>,
+
     // for forwards compatibility; to support in-place decryption in the future
     _incoming_tls: &'i mut [u8],
-    chunk: Vec<u8>,
-    taken: bool,
+
+    // owner of the latest chunk obtained in `next_record`, as borrowed by
+    // `AppDataRecord`
+    chunk: Option<Vec<u8>>,
 }
 
-impl<'c, 'i, Data> ReadEarlyData<'c, 'i, Data> {
+impl<'c, 'i> ReadEarlyData<'c, 'i, ServerConnectionData> {
     fn new(
-        _conn: &'c mut UnbufferedConnectionCommon<Data>,
+        conn: &'c mut UnbufferedConnectionCommon<ServerConnectionData>,
         _incoming_tls: &'i mut [u8],
-        chunk: Vec<u8>,
     ) -> Self {
         Self {
-            _conn,
+            conn,
             _incoming_tls,
-            chunk,
-            taken: false,
+            chunk: None,
         }
     }
-}
 
-impl ReadEarlyData<'_, '_, ServerConnectionData> {
     /// decrypts and returns the next available app-data record
     // TODO deprecate in favor of `Iterator` implementation, which requires in-place decryption
     pub fn next_record(&mut self) -> Option<Result<AppDataRecord<'_>, Error>> {
-        if self.taken {
-            None
-        } else {
-            self.taken = true;
-            Some(Ok(AppDataRecord {
+        self.chunk = self.conn.pop_early_data();
+        self.chunk.as_ref().map(|chunk| {
+            Ok(AppDataRecord {
                 discard: 0,
-                payload: &self.chunk,
-            }))
-        }
+                payload: chunk,
+            })
+        })
     }
 
     /// returns the payload size of the next app-data record *without* decrypting it
     ///
     /// returns `None` if there are no more app-data records
     pub fn peek_len(&self) -> Option<NonZeroUsize> {
-        if self.taken {
-            None
-        } else {
-            NonZeroUsize::new(self.chunk.len())
-        }
+        self.conn
+            .peek_early_data()
+            .and_then(|ch| NonZeroUsize::new(ch.len()))
     }
 }
 
diff --git a/rustls/src/server/server_conn.rs b/rustls/src/server/server_conn.rs
index 0541ee0a93..8108701e14 100644
--- a/rustls/src/server/server_conn.rs
+++ b/rustls/src/server/server_conn.rs
@@ -918,6 +918,10 @@ impl UnbufferedConnectionCommon<ServerConnectionData> {
     pub(crate) fn pop_early_data(&mut self) -> Option<Vec<u8>> {
         self.core.data.early_data.pop()
     }
+
+    pub(crate) fn peek_early_data(&self) -> Option<&[u8]> {
+        self.core.data.early_data.peek()
+    }
 }
 
 /// Represents a `ClientHello` message received through the [`Acceptor`].
@@ -1058,6 +1062,13 @@ impl EarlyDataState {
         matches!(self, Self::Rejected)
     }
 
+    fn peek(&self) -> Option<&[u8]> {
+        match self {
+            Self::Accepted { received, .. } => received.peek(),
+            _ => None,
+        }
+    }
+
     fn pop(&mut self) -> Option<Vec<u8>> {
         match self {
             Self::Accepted { received, .. } => received.pop(),
diff --git a/rustls/src/vecbuf.rs b/rustls/src/vecbuf.rs
index 1343db5fec..ddc150d529 100644
--- a/rustls/src/vecbuf.rs
+++ b/rustls/src/vecbuf.rs
@@ -113,6 +113,13 @@ impl ChunkVecBuffer {
 
         Ok(())
     }
+
+    /// Inspect the first chunk from this object.
+    pub(crate) fn peek(&self) -> Option<&[u8]> {
+        self.chunks
+            .front()
+            .map(|ch| ch.as_slice())
+    }
 }
 
 #[cfg(feature = "std")]


Test patch:
diff --git a/rustls/tests/unbuffered.rs b/rustls/tests/unbuffered.rs
index 4ceb3472f8..850f149bd0 100644
--- a/rustls/tests/unbuffered.rs
+++ b/rustls/tests/unbuffered.rs
@@ -919,6 +919,53 @@ fn rejects_junk() {
     confirm_transmit_tls_data(server.process_tls_records(&mut []));
 }
 
+#[test]
+fn read_traffic_not_consumed_too_early() {
+    let mut outcome = handshake(&rustls::version::TLS13);
+    let mut client = outcome.client.take().unwrap();
+    let mut server = outcome.server.take().unwrap();
+
+    let mut client_to_server_buf = Buffer::default();
+    write_traffic(client.process_tls_records(&mut []), |mut wt| {
+        encrypt(&mut wt, b"hello", &mut client_to_server_buf)
+    });
+
+    // if we just peek, we are presented the same data again
+    let (_, discard) = read_traffic(
+        server.process_tls_records(client_to_server_buf.filled()),
+        |rt| assert_eq!(rt.peek_len(), NonZeroUsize::new(5)),
+    );
+    assert!(discard > 0);
+    client_to_server_buf.discard(discard);
+
+    // ditto
+    let (_, discard) = read_traffic(
+        server.process_tls_records(client_to_server_buf.filled()),
+        |rt| assert_eq!(rt.peek_len(), NonZeroUsize::new(5)),
+    );
+    assert_eq!(discard, 0);
+
+    // now consume
+    let (data, discard) = read_traffic(
+        server.process_tls_records(client_to_server_buf.filled()),
+        |mut rt| {
+            rt.next_record()
+                .unwrap()
+                .unwrap()
+                .payload
+                .to_vec()
+        },
+    );
+    assert_eq!(discard, 0);
+    assert_eq!(data, b"hello");
+
+    // server is now idle
+    write_traffic(
+        server.process_tls_records(client_to_server_buf.filled()),
+        |_| (),
+    );
+}
+
 fn write_traffic<T: SideData, R, F: FnMut(WriteTraffic<T>) -> R>(
     status: UnbufferedStatus<'_, '_, T>,
     mut f: F,