Aggregate SecretConnection chunks with unmarshal protobuf retry

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];