Aggregate SecretConnection chunks with unmarshal protobuf retry (#903)

```go
// tendermint/cometbft proposal:
type Proposal struct {
	Type      SignedMsgType
	Height    int64
	Round     int32
	PolRound  int32
	BlockID   BlockID
	Timestamp time.Time
	Signature []byte
}
```

```go
// vs sei-tendermint proposal
type Proposal struct {
	Type            SignedMsgType
	Height          int64
	Round           int32
	PolRound        int32
	BlockID         BlockID
	Timestamp       time.Time
	Signature       []byte

        // this is a list, and can be very long...
	TxKeys          []*TxKey
	Evidence        *EvidenceList
	LastCommit      *Commit
	Header          Header
	ProposerAddress []byte
}
```

Since Proposal has TxKeys and other lists, Proposal has variable length
It is easily goes > 1024 bytes if block has big mount of txs. And it
is not a problem of canonical tendermint/cometbft implementations
since due to its message structure, it has a fixed max length < 1024 (DATA_MAX_SIZE)

sei-tendermint, when it connects to remote signer over tcp, sends
proposal divided by chunk of DATA_MAX_SIZE (1024) each, which kind of
fits the expectation of tmkms. However, tmkms never tries to
aggregate chunks. In fact, it is impossible for tmkms to implement
aggregation properly without knowing the length beforehand: which is
not provided by tendermint protocol.

There might be a confusion also, because all implementations of
tendermint send lenght-delimited protobufs, and tmkms also reads with
a function "length delimited". However, it actually means that the
protobuf msg is prepended by it's length: so that when tmkms reads
1024 bytes it knows which zeroes are payload and which a need to be
cut. Another words, it has nothing to do with multi-chunk payload.

Which means that sei-tendermint just doesn't bother about tcp
remote signer, and it is impossible to make it work with tmkms without
rewriting both and adding this custom protocol of "aggregate chunks until
you get full message length".

--
This code implements aggregation by trying to unmarshal aggregated
message each time it gets a new chunk. I don't think it is a good idea
in a long run, however, the alternative would be to adjust both Sei
and tmkms, rolling out new length-aware protocol between them -- I'm
not sure how sufficient it is and definitely needs a
discussion. Current solution is compartable with both
cometbft/tendermint and sei-tendermint, however, way less efficient then
the original `read` implementation of tmkms.

---------

Co-authored-by: Mateusz Kaczanowski <mateusz@chorus.one>

src/rpc.rs

@@ -7,12 +7,9 @@ use crate::privval::SignableMsg;
 use prost::Message as _;
 use std::io::Read;
 use tendermint::{chain, Proposal, Vote};
+use tendermint_p2p::secret_connection::DATA_MAX_SIZE;
 use tendermint_proto as proto;
 
-// TODO(tarcieri): use `tendermint_p2p::secret_connection::DATA_MAX_SIZE`
-// See informalsystems/tendermint-rs#1356
-const DATA_MAX_SIZE: usize = 262144;
-
 use crate::{
     error::{Error, ErrorKind},
     prelude::*,
@@ -31,12 +28,36 @@ pub enum Request {
 impl Request {
     /// Read a request from the given readable.
     pub fn read(conn: &mut impl Read, expected_chain_id: &chain::Id) -> Result<Self, Error> {
-        let msg_bytes = read_msg(conn)?;
-
-        // Parse Protobuf-encoded request message
-        let msg = proto::privval::Message::decode_length_delimited(msg_bytes.as_ref())
-            .map_err(|e| format_err!(ErrorKind::ProtocolError, "malformed message packet: {}", e))?
-            .sum;
+        let mut msg_bytes: Vec<u8> = vec![];
+        let msg;
+
+        // fix for Sei: collect incoming bytes of Protobuf from incoming msg
+        loop {
+            let mut msg_chunk = read_msg(conn)?;
+            let chunk_len = msg_chunk.len();
+            msg_bytes.append(&mut msg_chunk);
+
+            // if we can decode it, great, break the loop
+            match proto::privval::Message::decode_length_delimited(msg_bytes.as_ref()) {
+                Ok(m) => {
+                    msg = m.sum;
+                    break;
+                }
+                Err(e) => {
+                    // if chunk_len < DATA_MAX_SIZE (1024) we assume it was the end of the message and it is malformed
+                    if chunk_len < DATA_MAX_SIZE {
+                        return Err(format_err!(
+                            ErrorKind::ProtocolError,
+                            "malformed message packet: {}",
+                            e
+                        )
+                        .into());
+                    }
+                    // otherwise, we go to start of the loop assuming next chunk(s)
+                    // will fill the message
+                }
+            }
+        }
 
         let (req, chain_id) = match msg {
             Some(proto::privval::message::Sum::SignVoteRequest(

tests/integration.rs

@@ -5,6 +5,7 @@ use chrono::{DateTime, Utc};
 use prost::Message;
 use rand::Rng;
 use signature::Verifier;
+use std::fs::File;
 use std::{
     fs,
     io::{self, Cursor, Read, Write},
@@ -617,6 +618,20 @@ fn test_handle_and_sign_ping_pong() {
     });
 }
 
+#[test]
+fn test_buffer_underflow_sign_proposal() {
+    let key_type = KeyType::Consensus;
+    ProtocolTester::apply(&key_type, |mut pt| {
+        send_buffer_underflow_request(&mut pt);
+        let response: Result<(), ()> = match read_response(&mut pt) {
+            proto::privval::message::Sum::SignedProposalResponse(_) => Ok(()),
+            other => panic!("unexpected message type in response: {other:?}"),
+        };
+
+        assert!(response.is_ok());
+    });
+}
+
 /// Encode request as a Protobuf message
 fn send_request(request: proto::privval::message::Sum, pt: &mut ProtocolTester) {
     let mut buf = vec![];
@@ -627,6 +642,15 @@ fn send_request(request: proto::privval::message::Sum, pt: &mut ProtocolTester)
     pt.write_all(&buf).unwrap();
 }
 
+/// Opens a binary file with big proposal (> 1024 bytes, from Sei network)
+/// and sends via protocol tester
+fn send_buffer_underflow_request(pt: &mut ProtocolTester) {
+    let mut file = File::open("tests/support/buffer-underflow-proposal.bin").unwrap();
+    let mut buf = Vec::<u8>::new();
+    file.read_to_end(&mut buf).unwrap();
+    pt.write_all(&buf).unwrap();
+}
+
 /// Read the response as a Protobuf message
 fn read_response(pt: &mut ProtocolTester) -> proto::privval::message::Sum {
     let mut resp_buf = vec![0u8; 4096];