Merge a9486b3 into 62ddb98

.changelog/unreleased/bug-fixes/1393-p2p-data-corruption.md

@@ -0,0 +1,2 @@
+- `[tendermint-p2p]` Fix data corruption on sending long messages via `SecretConnection`
+  ([\#1393](https://github.com/informalsystems/tendermint-rs/pull/1393))

p2p/src/secret_connection.rs

@@ -538,15 +538,15 @@ fn encrypt_and_write<IoHandler: Write>(
     data: &[u8],
 ) -> io::Result<usize> {
     let mut n = 0_usize;
-    let mut data_copy = data;
-    while !data_copy.is_empty() {
+    let mut remaining = data;
+    while !remaining.is_empty() {
         let chunk: &[u8];
-        if DATA_MAX_SIZE < data.len() {
-            chunk = &data[..DATA_MAX_SIZE];
-            data_copy = &data_copy[DATA_MAX_SIZE..];
+        if DATA_MAX_SIZE < remaining.len() {
+            chunk = &remaining[..DATA_MAX_SIZE];
+            remaining = &remaining[DATA_MAX_SIZE..];
         } else {
-            chunk = data_copy;
-            data_copy = &[0_u8; 0];
+            chunk = remaining;
+            remaining = &[];
         }
         let sealed_frame = &mut [0_u8; TAG_SIZE + TOTAL_FRAME_SIZE];
         encrypt(chunk, &send_state.cipher, &send_state.nonce, sealed_frame)

test/src/test/unit/p2p/secret_connection.rs

@@ -62,6 +62,35 @@ fn test_read_write_single_message() {
     receiver.join().expect("receiver thread has panicked");
 }
 
+#[test]
+fn test_read_write_long_message() {
+    let mut message: [u8; 1025] = [0x5a; 1025];
+    message[1024] = 0xa5;
+
+    let (pipe1, pipe2) = pipe::async_bipipe_buffered();
+
+    let sender = thread::spawn(move || {
+        let mut conn1 = new_peer_conn(pipe2).expect("handshake to succeed");
+
+        conn1
+            .write_all(&message)
+            .expect("expected to write message");
+    });
+
+    let receiver = thread::spawn(move || {
+        let mut conn2 = new_peer_conn(pipe1).expect("handshake to succeed");
+
+        let mut buf = [0; 1025];
+        conn2
+            .read_exact(&mut buf)
+            .expect("expected to read message");
+        assert_eq!(&message, &buf);
+    });
+
+    sender.join().expect("sender thread has panicked");
+    receiver.join().expect("receiver thread has panicked");
+}
+
 #[test]
 fn test_evil_peer_shares_invalid_eph_key() {
     let csprng = OsRng {};