refactor(tls-client): make tls_client compatible with a synchronous API (#1027)

* refactor(tls-client): make `write_plaintext` sync and remove async api

* restore `complete_io`

* do not potentially block in `write_all_plaintext`

Author: th4s

crates/tls/client/src/conn.rs

@@ -457,6 +457,9 @@ impl ConnectionCommon {
             return Err(Error::CorruptMessage);
         }
 
+        // Process outgoing plaintext buffer and encrypt messages.
+        self.flush_plaintext().await?;
+
         // Process new messages.
         while let Some(msg) = self.message_deframer.frames.pop_front() {
             // If we're not decrypting yet, we process it immediately. Otherwise it will be
@@ -508,25 +511,22 @@ impl ConnectionCommon {
         Ok(state)
     }
 
-    /// Write buffer into connection.
-    pub async fn write_plaintext(&mut self, buf: &[u8]) -> Result<usize, Error> {
-        if let Ok(st) = &mut self.state {
-            st.perhaps_write_key_update(&mut self.common_state).await;
+    /// Writes plaintext `buf` into an internal buffer. May not fully process the
+    /// whole buffer and returns the processed length.
+    pub fn write_plaintext(&mut self, buf: &[u8]) -> Result<usize, Error> {
+        if buf.is_empty() {
+            // Don't send empty fragments.
+            return Ok(0);
         }
-        self.common_state.send_some_plaintext(buf).await
+
+        let len = self.sendable_plaintext.append_limited_copy(buf);
+        Ok(len)
     }
 
-    /// Write entire buffer into connection.
-    pub async fn write_all_plaintext(&mut self, buf: &[u8]) -> Result<usize, Error> {
-        let mut pos = 0;
-        while pos < buf.len() {
-            pos += self.write_plaintext(&buf[pos..]).await?;
-        }
-        self.backend.flush().await?;
-        while let Some(msg) = self.backend.next_outgoing().await? {
-            self.queue_tls_message(msg);
-        }
-        Ok(pos)
+    /// Writes the entire plaintext `buf` into an internal buffer.
+    pub fn write_all_plaintext(&mut self, buf: &[u8]) -> Result<(), Error> {
+        self.sendable_plaintext.append(buf.to_vec());
+        Ok(())
     }
 
     /// Read TLS content from `rd`.  This method does internal
@@ -782,15 +782,6 @@ impl CommonState {
         }
     }
 
-    /// Send plaintext application data, fragmenting and
-    /// encrypting it as it goes out.
-    ///
-    /// If internal buffers are too small, this function will not accept
-    /// all the data.
-    pub(crate) async fn send_some_plaintext(&mut self, data: &[u8]) -> Result<usize, Error> {
-        self.send_plain(data, Limit::Yes).await
-    }
-
     // Changing the keys must not span any fragmented handshake
     // messages.  Otherwise the defragmented messages will have
     // been protected with two different record layer protections,
@@ -931,32 +922,6 @@ impl CommonState {
         self.sendable_tls.write_to_async(wr).await
     }
 
-    /// Encrypt and send some plaintext `data`.  `limit` controls
-    /// whether the per-connection buffer limits apply.
-    ///
-    /// Returns the number of bytes written from `data`: this might
-    /// be less than `data.len()` if buffer limits were exceeded.
-    async fn send_plain(&mut self, data: &[u8], limit: Limit) -> Result<usize, Error> {
-        if !self.may_send_application_data {
-            // If we haven't completed handshaking, buffer
-            // plaintext to send once we do.
-            let len = match limit {
-                Limit::Yes => self.sendable_plaintext.append_limited_copy(data),
-                Limit::No => self.sendable_plaintext.append(data.to_vec()),
-            };
-            return Ok(len);
-        }
-
-        debug_assert!(self.record_layer.is_encrypting());
-
-        if data.is_empty() {
-            // Don't send empty fragments.
-            return Ok(0);
-        }
-
-        self.send_appdata_encrypt(data, limit).await
-    }
-
     pub(crate) async fn start_outgoing_traffic(&mut self) -> Result<(), Error> {
         self.may_send_application_data = true;
         self.flush_plaintext().await
@@ -1012,15 +977,14 @@ impl CommonState {
         self.sendable_tls.set_limit(limit);
     }
 
-    /// Send any buffered plaintext.  Plaintext is buffered if
-    /// written during handshake.
-    async fn flush_plaintext(&mut self) -> Result<(), Error> {
+    /// Send and encrypt any buffered plaintext. Does nothing during handshake.
+    pub async fn flush_plaintext(&mut self) -> Result<(), Error> {
         if !self.may_send_application_data {
             return Ok(());
         }
 
         while let Some(buf) = self.sendable_plaintext.pop() {
-            self.send_plain(&buf, Limit::No).await?;
+            self.send_appdata_encrypt(&buf, Limit::No).await?;
         }
 
         Ok(())

crates/tls/client/tests/api.rs

@@ -247,7 +247,8 @@ async fn servered_client_data_sent() {
         let (mut client, mut server) =
             make_pair_for_arc_configs(&Arc::new(client_config), &server_config).await;
 
-        assert_eq!(5, client.write_plaintext(b"hello").await.unwrap());
+        assert_eq!(5, client.write_plaintext(b"hello").unwrap());
+        client.flush_plaintext().await.unwrap();
 
         do_handshake(&mut client, &mut server).await;
         send(&mut client, &mut server);
@@ -286,7 +287,7 @@ async fn servered_both_data_sent() {
             make_pair_for_arc_configs(&Arc::new(client_config), &server_config).await;
 
         assert_eq!(12, server.writer().write(b"from-server!").unwrap());
-        assert_eq!(12, client.write_plaintext(b"from-client!").await.unwrap());
+        assert_eq!(12, client.write_plaintext(b"from-client!").unwrap());
 
         do_handshake(&mut client, &mut server).await;
 
@@ -432,7 +433,7 @@ async fn server_close_notify() {
 
         // check that alerts don't overtake appdata
         assert_eq!(12, server.writer().write(b"from-server!").unwrap());
-        assert_eq!(12, client.write_plaintext(b"from-client!").await.unwrap());
+        assert_eq!(12, client.write_plaintext(b"from-client!").unwrap());
         server.send_close_notify();
 
         receive(&mut server, &mut client);
@@ -460,7 +461,8 @@ async fn client_close_notify() {
 
         // check that alerts don't overtake appdata
         assert_eq!(12, server.writer().write(b"from-server!").unwrap());
-        assert_eq!(12, client.write_plaintext(b"from-client!").await.unwrap());
+        assert_eq!(12, client.write_plaintext(b"from-client!").unwrap());
+        client.flush_plaintext().await.unwrap();
         client.send_close_notify().await.unwrap();
 
         send(&mut client, &mut server);
@@ -487,7 +489,7 @@ async fn server_closes_uncleanly() {
 
         // check that unclean EOF reporting does not overtake appdata
         assert_eq!(12, server.writer().write(b"from-server!").unwrap());
-        assert_eq!(12, client.write_plaintext(b"from-client!").await.unwrap());
+        assert_eq!(12, client.write_plaintext(b"from-client!").unwrap());
 
         receive(&mut server, &mut client);
         transfer_eof(&mut client);
@@ -518,7 +520,7 @@ async fn client_closes_uncleanly() {
 
         // check that unclean EOF reporting does not overtake appdata
         assert_eq!(12, server.writer().write(b"from-server!").unwrap());
-        assert_eq!(12, client.write_plaintext(b"from-client!").await.unwrap());
+        assert_eq!(12, client.write_plaintext(b"from-client!").unwrap());
         client.process_new_packets().await.unwrap();
 
         send(&mut client, &mut server);
@@ -900,20 +902,9 @@ async fn client_respects_buffer_limit_pre_handshake() {
 
     client.set_buffer_limit(Some(32));
 
-    assert_eq!(
-        client
-            .write_plaintext(b"01234567890123456789")
-            .await
-            .unwrap(),
-        20
-    );
-    assert_eq!(
-        client
-            .write_plaintext(b"01234567890123456789")
-            .await
-            .unwrap(),
-        12
-    );
+    assert_eq!(client.write_plaintext(b"01234567890123456789").unwrap(), 20);
+    assert_eq!(client.write_plaintext(b"01234567890123456789").unwrap(), 12);
+    client.flush_plaintext().await.unwrap();
 
     do_handshake(&mut client, &mut server).await;
     send(&mut client, &mut server);
@@ -953,20 +944,9 @@ async fn client_respects_buffer_limit_post_handshake() {
     do_handshake(&mut client, &mut server).await;
     client.set_buffer_limit(Some(48));
 
-    assert_eq!(
-        client
-            .write_plaintext(b"01234567890123456789")
-            .await
-            .unwrap(),
-        20
-    );
-    assert_eq!(
-        client
-            .write_plaintext(b"01234567890123456789")
-            .await
-            .unwrap(),
-        6
-    );
+    assert_eq!(client.write_plaintext(b"01234567890123456789").unwrap(), 20);
+    assert_eq!(client.write_plaintext(b"01234567890123456789").unwrap(), 6);
+    client.flush_plaintext().await.unwrap();
 
     send(&mut client, &mut server);
     server.process_new_packets().unwrap();
@@ -1211,14 +1191,8 @@ async fn client_complete_io_for_write() {
 
         do_handshake(&mut client, &mut server).await;
 
-        client
-            .write_plaintext(b"01234567890123456789")
-            .await
-            .unwrap();
-        client
-            .write_plaintext(b"01234567890123456789")
-            .await
-            .unwrap();
+        client.write_plaintext(b"01234567890123456789").unwrap();
+        client.write_plaintext(b"01234567890123456789").unwrap();
         {
             let mut pipe = ServerSession::new(&mut server);
             let (rdlen, wrlen) = client
@@ -1350,7 +1324,8 @@ async fn server_stream_read() {
     for kt in ALL_KEY_TYPES.iter() {
         let (mut client, mut server) = make_pair(*kt).await;
 
-        client.write_all_plaintext(b"world").await.unwrap();
+        client.write_all_plaintext(b"world").unwrap();
+        client.process_new_packets().await.unwrap();
 
         {
             let mut pipe = ClientSession::new(&mut client);
@@ -1366,7 +1341,8 @@ async fn server_streamowned_read() {
     for kt in ALL_KEY_TYPES.iter() {
         let (mut client, server) = make_pair(*kt).await;
 
-        client.write_all_plaintext(b"world").await.unwrap();
+        client.write_all_plaintext(b"world").unwrap();
+        client.process_new_packets().await.unwrap();
 
         {
             let pipe = ClientSession::new(&mut client);
@@ -1385,7 +1361,9 @@ async fn server_streamowned_read() {
 //         errkind: io::ErrorKind::ConnectionAborted,
 //         after: 0,
 //     };
-//     client.write_all_plaintext(b"hello").await.unwrap();
+//     client.write_all_plaintext(b"hello").unwrap();
+//     client.process_new_packets().await.unwrap();
+//
 //     let mut client_stream = Stream::new(&mut client, &mut pipe);
 //     let rc = client_stream.write(b"world");
 //     assert!(rc.is_err());
@@ -1402,7 +1380,9 @@ async fn server_streamowned_read() {
 //         errkind: io::ErrorKind::ConnectionAborted,
 //         after: 1,
 //     };
-//     client.write_all_plaintext(b"hello").await.unwrap();
+//     client.write_all_plaintext(b"hello").unwrap();
+//     client.process_new_packets().await.unwrap();
+//
 //     let mut client_stream = Stream::new(&mut client, &mut pipe);
 //     let rc = client_stream.write(b"world");
 //     assert_eq!(format!("{:?}", rc), "Ok(5)");
@@ -1900,14 +1880,9 @@ async fn servered_write_for_client_appdata() {
     let (mut client, mut server) = make_pair(KeyType::Rsa).await;
     do_handshake(&mut client, &mut server).await;
 
-    client
-        .write_all_plaintext(b"01234567890123456789")
-        .await
-        .unwrap();
-    client
-        .write_all_plaintext(b"01234567890123456789")
-        .await
-        .unwrap();
+    client.write_all_plaintext(b"01234567890123456789").unwrap();
+    client.write_all_plaintext(b"01234567890123456789").unwrap();
+    client.process_new_packets().await.unwrap();
     {
         let mut pipe = ServerSession::new(&mut server);
         let wrlen = client.write_tls(&mut pipe).unwrap();
@@ -2019,11 +1994,10 @@ async fn servered_write_for_server_handshake_no_half_rtt_by_default() {
 async fn servered_write_for_client_handshake() {
     let (mut client, mut server) = make_pair(KeyType::Rsa).await;
 
-    client
-        .write_all_plaintext(b"01234567890123456789")
-        .await
-        .unwrap();
-    client.write_all_plaintext(b"0123456789").await.unwrap();
+    client.write_all_plaintext(b"01234567890123456789").unwrap();
+    client.write_all_plaintext(b"0123456789").unwrap();
+    client.process_new_packets().await.unwrap();
+
     {
         let mut pipe = ServerSession::new(&mut server);
         let wrlen = client.write_tls(&mut pipe).unwrap();