Repository: rustls/rustls (language: rust)

Issue:
unbuffered: After `Closed` no `WriteTraffic` state arrives
**Checklist**
* [x] I've searched the issue tracker for similar bugs.

**Describe the bug**
I've updated my `pipebuf` crate Rustls wrapper in crate `pipebuf_rustls` to add unbuffered Rustls support. However my tests which succeed in the buffered mode fail for the unbuffered mode. The failing test does the following:

- Client sends one byte
- Client closes connection
- Server sends one byte
- Server closes connection

The byte sent by the server cannot be processed through the unbuffered interface because the `process_tls_records` call does not give a `WriteTraffic` state again after `Closed`. Without access to `WriteTraffic`, I can't send the data. The buffered interface does not have this limitation.

**To Reproduce**
Check out `pipebuf_rustls` from github (https://github.com/uazu/pipebuf_rustls).  Run a single test using `cargo test --no-default-features --features unbuffered byte_each_way`. There is an assertion failure due to 1 byte sent but 0 bytes received.

**Applicable Version(s)**
0.23.4 on Linux

**Expected behavior**
I'd expect this case to be supported. I don't know how you'd want to support it with your unbuffered API though.

**Additional context**
The reason I added unbuffered support to `pipebuf_rustls` was that this is a much better fit to the `pipebuf` model. I added `PBufRd::data_mut` to `pipebuf` 0.3.1 to support Rustls unbuffered mode (which requires `&mut [u8]` for incoming data). So I support the unbuffered effort. Just it doesn't cover all necessary cases yet.


Discussion / hints:
Thanks for the report. Yes, the current API treats a half-close as a full-close. I think we should add additional `ConnectionState`s so the server in your example can finish its connection with:

- `ReadTraffic` reads the client's close_notify
- `PeerClosed` informs the application that no further data will be forthcoming from peer
- `WriteTraffic` for additional server-sent traffic, followed by `queue_close_notify`
- `EncodeTlsData` for outputting that data
- `TransmitTlsData` for confirming the data was sent 
- `Closed` nothing further can happen with this connection

(This arrangement retains `Closed` as a terminal, full-close state.)
Will this need semver-incompatible changes? If so, it might take us a while to get this released.
The `ConnectionState` enum is `non_exhaustive` so adding new states there isn't semver breaking in itself. I think it would be more a question of how the semantics of the overall API are updated and whether it affects existing usages.
Thanks for the quick response. There is no rush from my point of view as I also have it working with the buffered interface, and I can continue to use that. So if it's a breaking change for 0.24 that's not a problem. The API changes you suggested sound fine, but I'll code against whatever you come up with. If you want me to test against a pre-release version of Rustls I can do that.

Also it would be helpful to have some guidance about buffer-sizing requirements, either in the docs or as API calls. For example I'm allowing 18KiB for `EncodeTlsData` because that appears to be the maximum required by protocol, but typically it will be much less. I can add another issue for this if you wish.

Fix patch:
diff --git a/rustls/src/common_state.rs b/rustls/src/common_state.rs
index 3ff4609df8c..12aca203a4b 100644
--- a/rustls/src/common_state.rs
+++ b/rustls/src/common_state.rs
@@ -40,6 +40,8 @@ pub struct CommonState {
     pub(crate) may_receive_application_data: bool,
     pub(crate) early_traffic: bool,
     sent_fatal_alert: bool,
+    /// If we signaled end of stream.
+    pub(crate) has_sent_close_notify: bool,
     /// If the peer has signaled end of stream.
     pub(crate) has_received_close_notify: bool,
     #[cfg(feature = "std")]
@@ -74,6 +76,7 @@ impl CommonState {
             may_receive_application_data: false,
             early_traffic: false,
             sent_fatal_alert: false,
+            has_sent_close_notify: false,
             has_received_close_notify: false,
             #[cfg(feature = "std")]
             has_seen_eof: false,
@@ -573,6 +576,7 @@ impl CommonState {
         }
         debug!("Sending warning alert {:?}", AlertDescription::CloseNotify);
         self.sent_fatal_alert = true;
+        self.has_sent_close_notify = true;
         self.send_warning_alert_no_log(AlertDescription::CloseNotify);
     }
 
diff --git a/rustls/src/conn.rs b/rustls/src/conn.rs
index 43706241452..426d3770cce 100644
--- a/rustls/src/conn.rs
+++ b/rustls/src/conn.rs
@@ -811,6 +811,7 @@ impl<Data> From<ConnectionCore<Data>> for ConnectionCommon<Data> {
 pub struct UnbufferedConnectionCommon<Data> {
     pub(crate) core: ConnectionCore<Data>,
     wants_write: bool,
+    emitted_peer_closed_state: bool,
 }
 
 impl<Data> From<ConnectionCore<Data>> for UnbufferedConnectionCommon<Data> {
@@ -818,6 +819,7 @@ impl<Data> From<ConnectionCore<Data>> for UnbufferedConnectionCommon<Data> {
         Self {
             core,
             wants_write: false,
+            emitted_peer_closed_state: false,
         }
     }
 }
diff --git a/rustls/src/conn/unbuffered.rs b/rustls/src/conn/unbuffered.rs
index ef8146e098e..ffe14c75f7d 100644
--- a/rustls/src/conn/unbuffered.rs
+++ b/rustls/src/conn/unbuffered.rs
@@ -131,6 +131,18 @@ impl<Data> UnbufferedConnectionCommon<Data> {
                 .core
                 .common_state
                 .has_received_close_notify
+                && !self.emitted_peer_closed_state
+            {
+                self.emitted_peer_closed_state = true;
+                break (buffer.pending_discard(), ConnectionState::PeerClosed);
+            } else if self
+                .core
+                .common_state
+                .has_received_close_notify
+                && self
+                    .core
+                    .common_state
+                    .has_sent_close_notify
             {
                 break (buffer.pending_discard(), ConnectionState::Closed);
             } else if self
@@ -185,7 +197,26 @@ pub enum ConnectionState<'c, 'i, Data> {
     /// the received data.
     ReadTraffic(ReadTraffic<'c, 'i, Data>),
 
-    /// Connection has been cleanly closed by the peer
+    /// Connection has been cleanly closed by the peer.
+    ///
+    /// This state is encountered at most once by each connection -- it is
+    /// "edge" triggered, rather than "level" triggered.
+    ///
+    /// It delimits the data received from the peer, meaning you can be sure you
+    /// have received all the data the peer sent.
+    ///
+    /// No further application data will be received from the peer, so no further
+    /// `ReadTraffic` states will be produced.
+    ///
+    /// However, it is possible to _send_ further application data via `WriteTraffic`
+    /// states, or close the connection cleanly by calling
+    /// [`WriteTraffic::queue_close_notify()`].
+    PeerClosed,
+
+    /// Connection has been cleanly closed by both us and the peer.
+    ///
+    /// This is a terminal state.  No other states will be produced for this
+    /// connection.
     Closed,
 
     /// One, or more, early (RTT-0) data records are available
@@ -262,6 +293,8 @@ impl<Data> fmt::Debug for ConnectionState<'_, '_, Data> {
         match self {
             Self::ReadTraffic(..) => f.debug_tuple("ReadTraffic").finish(),
 
+            Self::PeerClosed => write!(f, "PeerClosed"),
+
             Self::Closed => write!(f, "Closed"),
 
             Self::ReadEarlyData(..) => f.debug_tuple("ReadEarlyData").finish(),


Test patch:
diff --git a/rustls/tests/unbuffered.rs b/rustls/tests/unbuffered.rs
index 7dbc2fc70d6..e37030b5b5e 100644
--- a/rustls/tests/unbuffered.rs
+++ b/rustls/tests/unbuffered.rs
@@ -5,8 +5,8 @@ use std::num::NonZeroUsize;
 use rustls::client::{ClientConnectionData, EarlyDataError, UnbufferedClientConnection};
 use rustls::server::{ServerConnectionData, UnbufferedServerConnection};
 use rustls::unbuffered::{
-    ConnectionState, EncodeError, EncryptError, InsufficientSizeError, UnbufferedConnectionCommon,
-    UnbufferedStatus, WriteTraffic,
+    ConnectionState, EncodeError, EncryptError, InsufficientSizeError, ReadTraffic,
+    UnbufferedConnectionCommon, UnbufferedStatus, WriteTraffic,
 };
 use rustls::version::TLS13;
 use rustls::{
@@ -329,10 +329,10 @@ fn run(
                     .client_received_app_data
                     .extend(records);
             }
-            State::Closed => {
-                client_handshake_done = true;
-                outcome.client_reached_connection_closed_state = true
+            State::PeerClosed => {
+                outcome.client_saw_peer_closed_state = true;
             }
+            State::Closed => {}
             state => unreachable!("{state:?}"),
         }
 
@@ -385,10 +385,10 @@ fn run(
                     .server_received_app_data
                     .extend(records);
             }
-            State::Closed => {
-                server_handshake_done = true;
-                outcome.server_reached_connection_closed_state = true
+            State::PeerClosed => {
+                outcome.server_saw_peer_closed_state = true;
             }
+            State::Closed => {}
         }
 
         count += 1;
@@ -435,7 +435,7 @@ fn close_notify_client_to_server() {
         );
 
         assert!(!client_actions.send_close_notify);
-        assert!(outcome.server_reached_connection_closed_state);
+        assert!(outcome.server_saw_peer_closed_state);
     }
 }
 
@@ -459,7 +459,59 @@ fn close_notify_server_to_client() {
         );
 
         assert!(!server_actions.send_close_notify);
-        assert!(outcome.client_reached_connection_closed_state);
+        assert!(outcome.client_saw_peer_closed_state);
+    }
+}
+
+#[test]
+fn full_closure_server_to_client() {
+    for version in rustls::ALL_VERSIONS {
+        eprintln!("{version:?}");
+        let mut outcome = handshake(version);
+        let mut client = outcome.client.take().unwrap();
+        let mut server = outcome.server.take().unwrap();
+
+        let mut buf = Buffer::default();
+
+        // server sends message followed by close_notify, in one flight
+        write_traffic(
+            server.process_tls_records(&mut []),
+            |mut wt: WriteTraffic<_>| {
+                encrypt(&mut wt, b"hello", &mut buf);
+                queue_close_notify(&mut wt, &mut buf);
+            },
+        );
+
+        let (_, discard) = read_traffic(client.process_tls_records(buf.filled()), |mut rt| {
+            assert_eq!(rt.peek_len(), NonZeroUsize::new(5));
+            let app_data = rt.next_record().unwrap().unwrap();
+            assert_eq!(app_data.payload, b"hello");
+        });
+        buf.discard(discard);
+
+        let discard = peer_closed(client.process_tls_records(buf.filled()));
+        buf.discard(discard);
+        assert_eq!(buf.used, 0);
+
+        // client replies with its own data and close_notify
+        write_traffic(client.process_tls_records(&mut []), |mut wt| {
+            encrypt(&mut wt, b"goodbye", &mut buf);
+            queue_close_notify(&mut wt, &mut buf);
+        });
+
+        let (_, discard) = read_traffic(server.process_tls_records(buf.filled()), |mut rt| {
+            assert_eq!(rt.peek_len(), NonZeroUsize::new(7));
+            let app_data = rt.next_record().unwrap().unwrap();
+            assert_eq!(app_data.payload, b"goodbye");
+        });
+        buf.discard(discard);
+
+        let discard = peer_closed(server.process_tls_records(buf.filled()));
+        buf.discard(discard);
+        assert_eq!(buf.used, 0);
+
+        closed(client.process_tls_records(&mut []));
+        closed(server.process_tls_records(&mut []));
     }
 }
 
@@ -483,13 +535,13 @@ fn junk_after_close_notify_received() {
     let discard = match dbg!(server.process_tls_records(dbg!(&mut client_send_buf[..len]))) {
         UnbufferedStatus {
             discard,
-            state: Ok(ConnectionState::Closed),
+            state: Ok(ConnectionState::PeerClosed),
         } => {
             assert_eq!(discard, 24);
             discard
         }
         st => {
-            panic!("unexpected server state {st:?} (wanted Closed)");
+            panic!("unexpected server state {st:?} (wanted PeerClosed)");
         }
     };
 
@@ -785,7 +837,7 @@ fn tls12_connection_fails_after_key_reaches_confidentiality_limit() {
     match server.process_tls_records(&mut data) {
         UnbufferedStatus {
             discard,
-            state: Ok(ConnectionState::Closed),
+            state: Ok(ConnectionState::PeerClosed),
         } if discard == data_len => {}
         st => panic!("unexpected server state {st:?}"),
     }
@@ -862,11 +914,36 @@ fn write_traffic<T: SideData, R, F: FnMut(WriteTraffic<T>) -> R>(
 ) -> R {
     let UnbufferedStatus { discard, state } = status;
     assert_eq!(discard, 0);
-    let state = state.unwrap();
-    if let ConnectionState::WriteTraffic(state) = state {
-        f(state)
-    } else {
-        panic!("unexpected client state {state:?} (wanted WriteTraffic)");
+    match state.unwrap() {
+        ConnectionState::WriteTraffic(state) => f(state),
+        other => panic!("unexpected state {other:?} (wanted WriteTraffic)"),
+    }
+}
+
+fn read_traffic<T: SideData, R, F: FnMut(ReadTraffic<T>) -> R>(
+    status: UnbufferedStatus<'_, '_, T>,
+    mut f: F,
+) -> (R, usize) {
+    let UnbufferedStatus { discard, state } = status;
+    match state.unwrap() {
+        ConnectionState::ReadTraffic(state) => (f(state), discard),
+        other => panic!("unexpected state {other:?} (wanted ReadTraffic)"),
+    }
+}
+
+fn peer_closed<T: SideData>(status: UnbufferedStatus<'_, '_, T>) -> usize {
+    let UnbufferedStatus { discard, state } = status;
+    match state.unwrap() {
+        ConnectionState::PeerClosed => discard,
+        other => panic!("unexpected state {other:?} (wanted PeerClosed)"),
+    }
+}
+
+fn closed<T: SideData>(status: UnbufferedStatus<'_, '_, T>) -> usize {
+    let UnbufferedStatus { discard, state } = status;
+    match state.unwrap() {
+        ConnectionState::Closed => discard,
+        other => panic!("unexpected state {other:?} (wanted Closed)"),
     }
 }
 
@@ -909,6 +986,7 @@ fn confirm_transmit_tls_data<T: SideData>(status: UnbufferedStatus<'_, '_, T>) {
 #[derive(Debug)]
 enum State {
     Closed,
+    PeerClosed,
     EncodedTlsData,
     TransmitTlsData {
         sent_app_data: bool,
@@ -955,11 +1033,11 @@ struct Outcome {
     server_transcript: Vec<String>,
     server_received_early_data: Vec<Vec<u8>>,
     server_received_app_data: Vec<Vec<u8>>,
-    server_reached_connection_closed_state: bool,
+    server_saw_peer_closed_state: bool,
     client: Option<UnbufferedClientConnection>,
     client_transcript: Vec<String>,
     client_received_app_data: Vec<Vec<u8>>,
-    client_reached_connection_closed_state: bool,
+    client_saw_peer_closed_state: bool,
 }
 
 fn advance_client(
@@ -976,23 +1054,21 @@ fn advance_client(
     let state = match state {
         ConnectionState::TransmitTlsData(mut state) => {
             let mut sent_early_data = false;
-            if let Some(early_data) = actions.early_data_to_send {
-                if let Some(mut state) = state.may_encrypt_early_data() {
-                    write_with_buffer_size_checks(
-                        |out_buf| state.encrypt(early_data, out_buf),
-                        |e| {
-                            println!("encrypt error: {e}");
-                            if let EarlyDataError::Encrypt(EncryptError::InsufficientSize(ise)) = e
-                            {
-                                ise
-                            } else {
-                                unreachable!()
-                            }
-                        },
-                        &mut buffers.outgoing,
-                    );
-                    sent_early_data = true;
-                }
+            if let (Some(early_data), Some(mut state)) =
+                (actions.early_data_to_send, state.may_encrypt_early_data())
+            {
+                write_with_buffer_size_checks(
+                    |out_buf| state.encrypt(early_data, out_buf),
+                    |e| {
+                        println!("encrypt error: {e}");
+                        match e {
+                            EarlyDataError::Encrypt(EncryptError::InsufficientSize(ise)) => ise,
+                            _ => unreachable!(),
+                        }
+                    },
+                    &mut buffers.outgoing,
+                );
+                sent_early_data = true;
             }
             state.done();
             State::TransmitTlsData {
@@ -1055,10 +1131,9 @@ fn handle_state<Data>(
                 |out_buf| state.encode(out_buf),
                 |e| {
                     println!("encode error: {e}");
-                    if let EncodeError::InsufficientSize(ise) = e {
-                        ise
-                    } else {
-                        unreachable!()
+                    match e {
+                        EncodeError::InsufficientSize(ise) => ise,
+                        _ => unreachable!(),
                     }
                 },
                 outgoing,
@@ -1074,11 +1149,11 @@ fn handle_state<Data>(
 
         ConnectionState::TransmitTlsData(mut state) => {
             let mut sent_app_data = false;
-            if let Some(app_data) = actions.app_data_to_send {
-                if let Some(mut state) = state.may_encrypt_app_data() {
-                    encrypt(&mut state, app_data, outgoing);
-                    sent_app_data = true;
-                }
+            if let (Some(app_data), Some(mut state)) =
+                (actions.app_data_to_send, state.may_encrypt_app_data())
+            {
+                encrypt(&mut state, app_data, outgoing);
+                sent_app_data = true;
             }
 
             let mut sent_close_notify = false;
@@ -1136,6 +1211,7 @@ fn handle_state<Data>(
             State::ReceivedAppData { records }
         }
 
+        ConnectionState::PeerClosed => State::PeerClosed,
         ConnectionState::Closed => State::Closed,
 
         _ => unreachable!(),
@@ -1159,10 +1235,9 @@ fn encrypt<Data>(state: &mut WriteTraffic<'_, Data>, app_data: &[u8], outgoing:
 }
 
 fn map_encrypt_error(e: EncryptError) -> InsufficientSizeError {
-    if let EncryptError::InsufficientSize(ise) = e {
-        ise
-    } else {
-        unreachable!()
+    match e {
+        EncryptError::InsufficientSize(ise) => ise,
+        _ => unreachable!(),
     }
 }