p2p: apply Core's message size and count limits, and bound queued sends per peer

node/src/net/connection.rs

@@ -6,7 +6,7 @@ use std::io;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
-use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt, ReadBuf, ReadHalf, WriteHalf};
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf, ReadHalf, WriteHalf};
 use tokio::net::TcpStream;
 
 use crate::net::stats::PeerStats;
@@ -225,8 +225,17 @@ pub struct V1Writer {
 
 /// Size of a P2P message header: 4 (magic) + 12 (command) + 4 (length) + 4 (checksum).
 const HEADER_SIZE: usize = 24;
-/// Maximum message payload size (32 MB).
-const MAX_PAYLOAD_SIZE: usize = 32 * 1024 * 1024;
+/// Maximum message payload size: Core's `MAX_PROTOCOL_MESSAGE_LENGTH`
+/// (`net.h`). A header announcing more ends the connection
+/// (`V1Transport::readHeader`); satd took up to 32 MiB.
+const MAX_PAYLOAD_SIZE: usize = 4_000_000;
+/// How far ahead of the payload bytes actually received the receive buffer
+/// is grown: Core's `V1Transport::readData` ("Allocate up to 256 KiB ahead,
+/// but never more than the total message size").
+const PAYLOAD_ALLOC_STEP: usize = 256 * 1024;
+/// Receive-buffer capacity kept for the next message. A buffer grown past it
+/// by a larger message is released once that message is decoded.
+const RECV_BUF_KEEP: usize = 64 * 1024;
 /// Maximum bytes to scan when resyncing to magic after stream misalignment.
 const MAX_RESYNC_BYTES: usize = 256 * 1024;
 
@@ -326,15 +335,39 @@ impl V1Writer {
     pub async fn send(&mut self, msg: NetworkMessage) -> io::Result<()> {
         let raw = RawNetworkMessage::new(self.magic, msg);
         let bytes = serialize(&raw);
-        self.stream.write_all(&bytes).await?;
+        write_counted(&mut self.stream, &bytes, self.counters.as_ref()).await?;
         if let Some(c) = &self.counters {
-            c.record_sent(bytes.len());
             c.attribute_sent(raw.cmd(), bytes.len());
         }
         Ok(())
     }
 }
 
+/// Write `bytes` in full, counting each chunk as the socket takes it.
+///
+/// Core moves `nSendBytes` and `m_last_send` on every partial send
+/// (`SocketSendData`). Counting only whole messages made a large message that
+/// a peer reads slowly look like no progress at all, and the write loop's send
+/// timeout is measured against these counts.
+pub(crate) async fn write_counted<W: AsyncWrite + Unpin>(
+    w: &mut W,
+    bytes: &[u8],
+    counters: Option<&Arc<PeerStats>>,
+) -> io::Result<()> {
+    let mut written = 0;
+    while written < bytes.len() {
+        let n = w.write(&bytes[written..]).await?;
+        if n == 0 {
+            return Err(io::ErrorKind::WriteZero.into());
+        }
+        written += n;
+        if let Some(c) = counters {
+            c.record_sent(n);
+        }
+    }
+    Ok(())
+}
+
 impl V1Reader {
     /// Receive the next network message. Skips messages that fail to deserialize.
     ///
@@ -383,10 +416,19 @@ async fn recv_message<R: AsyncReadExt + Unpin>(
             ));
         }
 
-        // Read payload
-        let mut payload = vec![0u8; payload_len];
-        if payload_len > 0 {
-            stream.read_exact(&mut payload).await?;
+        // Read the payload into `buf`, behind the header, growing it as the
+        // bytes arrive and never more than PAYLOAD_ALLOC_STEP ahead of them
+        // (Core's `V1Transport::readData`). Sizing it from the header alone
+        // let a peer that announced a large payload and sent nothing hold
+        // that much memory per connection.
+        buf.clear();
+        buf.extend_from_slice(&header);
+        let frame_len = HEADER_SIZE + payload_len;
+        while buf.len() < frame_len {
+            let start = buf.len();
+            let end = frame_len.min(start + PAYLOAD_ALLOC_STEP);
+            buf.resize(end, 0);
+            stream.read_exact(&mut buf[start..end]).await?;
         }
 
         // Record on-wire bytes for this framed message (counted even if it
@@ -399,23 +441,16 @@ async fn recv_message<R: AsyncReadExt + Unpin>(
             c.record_recv(wire_len);
         }
 
-        // Combine header + payload and deserialize
-        buf.clear();
-        buf.extend_from_slice(&header);
-        buf.extend_from_slice(&payload);
-
-        match deserialize::<RawNetworkMessage>(buf) {
+        let result = match deserialize::<RawNetworkMessage>(buf) {
+            Ok(raw) if *raw.magic() != magic => Some(Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "wrong network magic",
+            ))),
             Ok(raw) => {
-                if *raw.magic() != magic {
-                    return Err(io::Error::new(
-                        io::ErrorKind::InvalidData,
-                        "wrong network magic",
-                    ));
-                }
                 if let Some(c) = counters {
                     c.attribute_recv(raw.cmd(), wire_len);
                 }
-                return Ok(raw.payload().clone());
+                Some(Ok(raw.into_payload()))
             }
             Err(_) if &header[4..16] == b"pong\0\0\0\0\0\0\0\0" && payload_len < 8 => {
                 // Too short to hold a nonce. Core still acts on it -- it
@@ -424,20 +459,29 @@ async fn recv_message<R: AsyncReadExt + Unpin>(
                 if let Some(c) = counters {
                     c.attribute_recv("pong", wire_len);
                 }
-                return Ok(NetworkMessage::Unknown {
+                Some(Ok(NetworkMessage::Unknown {
                     command: bitcoin::p2p::message::CommandString::try_from_static("pong")
                         .expect("static command"),
-                    payload,
-                });
+                    payload: buf[HEADER_SIZE..].to_vec(),
+                }))
             }
             Err(_) => {
                 if let Some(c) = counters {
                     c.attribute_recv(crate::net::stats::MSG_TYPE_OTHER, wire_len);
                 }
-                let cmd = String::from_utf8_lossy(&header[4..16]);
-                tracing::debug!(cmd = %cmd.trim_end_matches('\0'), "Skipping unparseable message");
-                continue;
+                // The peer's own text: sanitized as Core's
+                // `SanitizeString(msg_type)`, so it cannot break a log line.
+                let cmd = crate::net::limits::sanitize_string(&String::from_utf8_lossy(&header[4..16]));
+                tracing::debug!(cmd = %cmd, "Skipping unparseable message");
+                None
             }
+        };
+        // Kept for the next message, but not at the size of a large one.
+        if buf.capacity() > RECV_BUF_KEEP {
+            *buf = Vec::new();
+        }
+        if let Some(result) = result {
+            return result;
         }
     }
 }
@@ -500,3 +544,7 @@ async fn resync_to_magic<R: AsyncReadExt + Unpin>(
         }
     }
 }
+
+#[cfg(test)]
+#[path = "connection_bounds_tests.rs"]
+mod bounds_tests;

node/src/net/connection_bounds_tests.rs

@@ -0,0 +1,116 @@
+//! The v1 frame limit, the receive buffer's growth and release, and send
+//! progress, against Bitcoin Core's `V1Transport` and `SocketSendData`.
+
+use super::*;
+use bitcoin::p2p::message::CommandString;
+use std::io::Cursor;
+use std::time::Duration;
+
+/// A reader that records the largest read it was asked for.
+struct MaxAsk<R> {
+    inner: R,
+    max: usize,
+}
+
+impl<R: AsyncRead + Unpin> AsyncRead for MaxAsk<R> {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        self.max = self.max.max(buf.remaining());
+        Pin::new(&mut self.inner).poll_read(cx, buf)
+    }
+}
+
+fn unknown(len: usize) -> NetworkMessage {
+    NetworkMessage::Unknown {
+        command: CommandString::try_from_static("bigmsg").unwrap(),
+        payload: vec![0x5a; len],
+    }
+}
+
+fn frame(msg: NetworkMessage) -> Vec<u8> {
+    serialize(&RawNetworkMessage::new(Magic::REGTEST, msg))
+}
+
+/// Core's `MAX_PROTOCOL_MESSAGE_LENGTH` is 4,000,000 bytes, and a header
+/// announcing more ends the connection (`V1Transport::readHeader`). satd
+/// took frames up to 32 MiB.
+#[tokio::test]
+async fn a_frame_over_four_million_bytes_is_refused() {
+    let mut buf = Vec::new();
+    let mut ok = Cursor::new(frame(unknown(4_000_000)));
+    let got = recv_message(&mut ok, Magic::REGTEST, &mut buf, None).await;
+    assert!(matches!(got, Ok(NetworkMessage::Unknown { ref payload, .. }) if payload.len() == 4_000_000));
+
+    let mut over = Cursor::new(frame(unknown(4_000_001)));
+    let err = recv_message(&mut over, Magic::REGTEST, &mut buf, None)
+        .await
+        .expect_err("a 4,000,001-byte payload is refused");
+    assert_eq!(err.kind(), io::ErrorKind::InvalidData, "{err}");
+}
+
+/// Core grows the receive buffer as payload bytes arrive, never more than
+/// 256 KiB ahead of them (`V1Transport::readData`). satd allocated the whole
+/// announced payload before reading any of it, so a peer could make it hold
+/// a full frame per connection by announcing one and sending nothing.
+#[tokio::test]
+async fn the_payload_buffer_grows_with_the_bytes_received() {
+    let mut bytes = frame(unknown(4_000_000));
+    bytes.truncate(HEADER_SIZE + 1000);
+    let mut reader = MaxAsk { inner: Cursor::new(bytes), max: 0 };
+    let mut buf = Vec::new();
+    let err = recv_message(&mut reader, Magic::REGTEST, &mut buf, None)
+        .await
+        .expect_err("the payload never arrives");
+    assert_eq!(err.kind(), io::ErrorKind::UnexpectedEof);
+    assert!(reader.max <= PAYLOAD_ALLOC_STEP, "asked for {} bytes at once", reader.max);
+    assert!(buf.capacity() <= HEADER_SIZE + 2 * PAYLOAD_ALLOC_STEP, "holding {} bytes", buf.capacity());
+}
+
+/// The receive buffer is reused between messages, but not at the size of
+/// the largest message the connection ever carried.
+#[tokio::test]
+async fn the_receive_buffer_is_released_after_a_large_message() {
+    let mut bytes = frame(unknown(1_000_000));
+    bytes.extend(frame(NetworkMessage::Ping(7)));
+    let mut reader = Cursor::new(bytes);
+    let mut buf = Vec::new();
+    let big = recv_message(&mut reader, Magic::REGTEST, &mut buf, None).await.unwrap();
+    assert!(matches!(big, NetworkMessage::Unknown { .. }));
+    assert!(buf.capacity() <= RECV_BUF_KEEP, "kept {} bytes after a 1 MB message", buf.capacity());
+    let small = recv_message(&mut reader, Magic::REGTEST, &mut buf, None).await.unwrap();
+    assert_eq!(small, NetworkMessage::Ping(7));
+}
+
+/// Bytes are counted as the socket takes them, not once the whole message
+/// is out: Core's `m_last_send` and `nSendBytes` move on every partial send
+/// (`SocketSendData`), and the send timeout is measured against that.
+#[tokio::test]
+async fn a_send_counts_its_bytes_as_the_socket_takes_them() {
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let (dialed, accepted) = tokio::join!(
+        TcpStream::connect(listener.local_addr().unwrap()),
+        listener.accept()
+    );
+    // The far end never reads.
+    let (_far, _) = accepted.unwrap();
+    let conn = Connection::with_magic(dialed.unwrap(), Magic::REGTEST);
+    let (_reader, mut writer) = conn.split();
+    let stats = PeerStats::new(crate::net::stats::NetTotals::new());
+    writer.set_counters(stats.clone());
+
+    let msg_len = frame(unknown(3_900_000)).len() as u64;
+    let mut completed = 0u64;
+    while let Ok(r) = tokio::time::timeout(Duration::from_millis(500), writer.send(unknown(3_900_000))).await {
+        r.unwrap();
+        completed += msg_len;
+        assert!(completed < 512 * 1024 * 1024, "the socket never filled");
+    }
+    assert!(
+        stats.bytes_sent() > completed,
+        "a send the socket only partly took is not counted: {} sent, {completed} in whole messages",
+        stats.bytes_sent()
+    );
+}

node/src/net/limits.rs

@@ -0,0 +1,60 @@
+//! Bitcoin Core's limits on what one P2P message may carry, and its
+//! sanitizing of peer-supplied text.
+//!
+//! rust-bitcoin decodes a vector of any length the frame can hold
+//! (`MAX_INV_SIZE` is documented as "not currently enforced"), so these are
+//! applied to the decoded message, with Core's reaction to each.
+
+/// Core's `MAX_HEADERS_RESULTS` (`net_processing.h`): the most headers a
+/// `headers` message may carry. More is `Misbehaving` ("headers message
+/// size = N").
+pub const MAX_HEADERS_RESULTS: usize = 2000;
+
+/// Core's `MAX_ADDR_TO_SEND` (`net_processing.cpp`): the most entries an
+/// `addr` or `addrv2` message may carry. More is `Misbehaving` ("addr
+/// message size = N", `ProcessAddrs`).
+pub const MAX_ADDR_TO_SEND: usize = 1000;
+
+/// The most entries of a `notfound` Core looks at: `MAX_PEER_TX_ANNOUNCEMENTS
+/// + MAX_BLOCKS_IN_TRANSIT_PER_PEER` (5000 + 16, `net_processing.cpp`
+/// `NOTFOUND`). A longer one is ignored whole, with no penalty.
+pub const MAX_NOTFOUND_SZ: usize = 5000 + 16;
+
+/// Characters Core's `SanitizeString` keeps under `SAFE_CHARS_DEFAULT`
+/// (`util/strencodings.cpp`): alphanumerics plus ` .,;-_/:?@()`.
+fn is_safe_char(c: char) -> bool {
+    c.is_ascii_alphanumeric() || " .,;-_/:?@()".contains(c)
+}
+
+/// Core's `SanitizeString(str)`: drop every character outside
+/// `SAFE_CHARS_DEFAULT`. Core applies it to a peer's user agent before it
+/// stores or logs it (`cleanSubVer`), and to message types in log lines, so
+/// nothing a peer sends can put a line break, a control character or an
+/// escape sequence into the log or into `getpeerinfo`.
+pub fn sanitize_string(s: &str) -> String {
+    s.chars().filter(|c| is_safe_char(*c)).collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Core's `SAFE_CHARS_DEFAULT`, character for character: everything it
+    /// lists survives, and nothing else in the byte range does.
+    #[test]
+    fn sanitize_string_keeps_exactly_cores_safe_chars() {
+        let core_safe: String = ('a'..='z')
+            .chain('A'..='Z')
+            .chain('0'..='9')
+            .chain(" .,;-_/:?@()".chars())
+            .collect();
+        assert_eq!(sanitize_string(&core_safe), core_safe);
+        for b in 0u8..=255 {
+            let c = char::from(b);
+            let kept = sanitize_string(&c.to_string());
+            assert_eq!(kept.is_empty(), !core_safe.contains(c), "byte {b:#04x}");
+        }
+        assert_eq!(sanitize_string("/Satoshi:27.0.0/\nfake log line\x1b[31m"), "/Satoshi:27.0.0/fake log line31m");
+        assert_eq!(sanitize_string("caf\u{e9}"), "caf");
+    }
+}