feat(net): cap payload accepted before bridge hello

crates/bridge-protocol/src/lib.rs

@@ -16,6 +16,9 @@ pub const VERSION: u16 = 0;
 /// The largest payload a frame may carry.
 pub const MAX_PAYLOAD: usize = 16 * 1024 * 1024;
 
+/// The largest payload the hello frame may carry. A connection that sends more before hello is closed.
+pub const MAX_HELLO_PAYLOAD: usize = 1024;
+
 /// The size of the key that opens a game traffic stream.
 pub const STREAM_KEY_LEN: usize = 16;
 
@@ -142,17 +145,42 @@ impl Frame {
 }
 
 /// Collects bytes as they arrive and cuts them into frames.
-#[derive(Debug, Default)]
+#[derive(Debug)]
 pub struct Decoder {
     buf: Vec<u8>,
     start: usize,
+    max_payload: usize,
+}
+
+impl Default for Decoder {
+    fn default() -> Self {
+        Self {
+            buf: Vec::new(),
+            start: 0,
+            max_payload: MAX_PAYLOAD,
+        }
+    }
 }
 
 impl Decoder {
     pub fn new() -> Self {
         Self::default()
     }
 
+    /// A decoder that limits the payload of frames to `max_payload`.
+    pub fn with_max_payload(max_payload: usize) -> Self {
+        Self {
+            buf: Vec::new(),
+            start: 0,
+            max_payload,
+        }
+    }
+
+    /// Sets the maximum payload size accepted for subsequent frames.
+    pub fn set_max_payload(&mut self, max_payload: usize) {
+        self.max_payload = max_payload;
+    }
+
     pub fn push(&mut self, bytes: &[u8]) {
         if self.start > 0 && self.start == self.buf.len() {
             self.buf.clear();
@@ -169,7 +197,7 @@ impl Decoder {
             return Ok(None);
         }
         let len = u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize;
-        if len > MAX_PAYLOAD {
+        if len > self.max_payload {
             return Err(Error::TooLarge(len));
         }
         if buf.len() < HEADER + len {

crates/bridge-protocol/src/tests.rs

@@ -114,6 +114,30 @@ fn an_oversized_length_is_refused_before_the_payload_arrives() {
     assert_eq!(decoder.next_frame(), Err(Error::TooLarge(MAX_PAYLOAD + 1)));
 }
 
+#[test]
+fn decoder_with_custom_max_payload_refuses_oversized_length() {
+    let mut header = ((MAX_HELLO_PAYLOAD + 1) as u32).to_le_bytes().to_vec();
+    header.extend_from_slice(&[1, 0, 0, 0, 0, 0, 0, 0]);
+    let mut decoder = Decoder::with_max_payload(MAX_HELLO_PAYLOAD);
+    decoder.push(&header);
+    assert_eq!(
+        decoder.next_frame(),
+        Err(Error::TooLarge(MAX_HELLO_PAYLOAD + 1))
+    );
+}
+
+#[test]
+fn decoder_with_custom_max_payload_accepts_exact_boundary() {
+    let mut wire = (MAX_HELLO_PAYLOAD as u32).to_le_bytes().to_vec();
+    wire.extend_from_slice(&[1, 0, 0, 0, 0, 0, 0, 0]);
+    wire.extend(vec![0xAA; MAX_HELLO_PAYLOAD]);
+    let mut decoder = Decoder::with_max_payload(MAX_HELLO_PAYLOAD);
+    decoder.push(&wire);
+    let frame = decoder.next_frame().unwrap().unwrap();
+    assert_eq!(frame.payload.len(), MAX_HELLO_PAYLOAD);
+    assert_eq!(frame.payload, vec![0xAA; MAX_HELLO_PAYLOAD]);
+}
+
 #[test]
 fn an_oversized_payload_is_not_encoded() {
     let frame = Frame {

crates/net/src/bridge.rs

@@ -6,7 +6,8 @@
 use std::net::Ipv4Addr;
 use std::time::Duration;
 
-use anyplay_bridge_protocol::{Decoder, Frame, Kind, Message, Reject};
+pub use anyplay_bridge_protocol::MAX_HELLO_PAYLOAD;
+use anyplay_bridge_protocol::{Decoder, Frame, Kind, MAX_PAYLOAD, Message, Reject};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::{TcpListener, TcpStream};
 use tokio::sync::mpsc::{self, UnboundedSender};
@@ -59,7 +60,7 @@ pub async fn listen(out: UnboundedSender<BridgeInput>) -> std::io::Result<u16> {
 async fn serve(connection: u64, stream: TcpStream, out: &UnboundedSender<BridgeInput>) {
     let _ = stream.set_nodelay(true);
     let (mut read, mut write) = stream.into_split();
-    let mut decoder = Decoder::new();
+    let mut decoder = Decoder::with_max_payload(MAX_HELLO_PAYLOAD);
     let mut buf = vec![0u8; 64 * 1024];
 
     let hello = tokio::time::timeout(Duration::from_secs(5), async {
@@ -83,6 +84,8 @@ async fn serve(connection: u64, stream: TcpStream, out: &UnboundedSender<BridgeI
         return;
     };
 
+    decoder.set_max_payload(MAX_PAYLOAD);
+
     let (frames, mut outgoing) = mpsc::unbounded_channel();
     let (verdict, answer) = oneshot::channel();
     if out
@@ -157,3 +160,157 @@ async fn serve(connection: u64, stream: TcpStream, out: &UnboundedSender<BridgeI
         _ = reader => {}
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use anyplay_bridge_protocol::{Frame, Hello, Message, VERSION};
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    use tokio::net::TcpStream;
+
+    #[tokio::test]
+    async fn a_hello_within_the_limit_is_accepted() {
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let port = listen(tx).await.unwrap();
+        let mut stream = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
+
+        let hello = Frame::event(&Message::Hello(Hello {
+            version: VERSION,
+            token: vec![1, 2, 3, 4],
+        }))
+        .unwrap();
+        let mut bytes = Vec::new();
+        hello.encode(&mut bytes).unwrap();
+        stream.write_all(&bytes).await.unwrap();
+
+        match rx.recv().await.unwrap() {
+            BridgeInput::Hello {
+                version,
+                token,
+                verdict,
+                ..
+            } => {
+                assert_eq!(version, VERSION);
+                assert_eq!(token, vec![1, 2, 3, 4]);
+                let _ = verdict.send(Ok(()));
+            }
+            other => panic!("expected Hello, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn an_oversized_payload_before_hello_is_refused() {
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let port = listen(tx).await.unwrap();
+        let mut stream = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
+
+        let oversized = Frame {
+            kind: Kind::HELLO,
+            request: 0,
+            payload: vec![0u8; MAX_HELLO_PAYLOAD + 1],
+        };
+        let mut bytes = Vec::new();
+        oversized.encode(&mut bytes).unwrap();
+        stream.write_all(&bytes).await.unwrap();
+
+        let mut answer = [0u8; 1];
+        let read = stream.read(&mut answer).await.unwrap();
+        assert_eq!(read, 0);
+
+        match rx.recv().await.unwrap() {
+            BridgeInput::Closed { .. } => {}
+            other => panic!("expected Closed without Hello, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn an_oversized_declared_length_before_hello_is_refused_at_once() {
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let port = listen(tx).await.unwrap();
+        let mut stream = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
+
+        let declared_len = (MAX_HELLO_PAYLOAD as u32) + 1;
+        let mut header = Vec::new();
+        header.extend_from_slice(&declared_len.to_le_bytes());
+        header.extend_from_slice(&Kind::HELLO.0.to_le_bytes());
+        header.extend_from_slice(&0u16.to_le_bytes());
+        header.extend_from_slice(&0u32.to_le_bytes());
+        stream.write_all(&header).await.unwrap();
+
+        let mut answer = [0u8; 1];
+        let read = stream.read(&mut answer).await.unwrap();
+        assert_eq!(read, 0);
+
+        match rx.recv().await.unwrap() {
+            BridgeInput::Closed { .. } => {}
+            other => panic!("expected Closed without Hello, got {other:?}"),
+        }
+    }
+    #[tokio::test]
+    async fn a_hello_with_exact_max_payload_is_accepted_at_boundary() {
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let port = listen(tx).await.unwrap();
+        let mut stream = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
+
+        // Exactly MAX_HELLO_PAYLOAD bytes: 2 bytes version + 1022 bytes token.
+        let token = vec![0x42u8; MAX_HELLO_PAYLOAD - 2];
+        let hello = Frame::event(&Message::Hello(Hello {
+            version: VERSION,
+            token: token.clone(),
+        }))
+        .unwrap();
+        assert_eq!(hello.payload.len(), MAX_HELLO_PAYLOAD);
+
+        let mut bytes = Vec::new();
+        hello.encode(&mut bytes).unwrap();
+        stream.write_all(&bytes).await.unwrap();
+
+        match rx.recv().await.unwrap() {
+            BridgeInput::Hello {
+                version,
+                token: received_token,
+                verdict,
+                ..
+            } => {
+                assert_eq!(version, VERSION);
+                assert_eq!(received_token, token);
+                let _ = verdict.send(Ok(()));
+            }
+            other => panic!("expected Hello at exact boundary, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn a_hello_split_over_tiny_writes_is_accepted() {
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let port = listen(tx).await.unwrap();
+        let mut stream = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
+
+        let hello = Frame::event(&Message::Hello(Hello {
+            version: VERSION,
+            token: vec![9, 8, 7, 6, 5],
+        }))
+        .unwrap();
+        let mut bytes = Vec::new();
+        hello.encode(&mut bytes).unwrap();
+
+        for chunk in bytes.chunks(2) {
+            stream.write_all(chunk).await.unwrap();
+            tokio::time::sleep(Duration::from_millis(1)).await;
+        }
+
+        match rx.recv().await.unwrap() {
+            BridgeInput::Hello {
+                version,
+                token,
+                verdict,
+                ..
+            } => {
+                assert_eq!(version, VERSION);
+                assert_eq!(token, vec![9, 8, 7, 6, 5]);
+                let _ = verdict.send(Ok(()));
+            }
+            other => panic!("expected Hello, got {other:?}"),
+        }
+    }
+}

spec/bridge-protocol.md

@@ -10,7 +10,7 @@ The Rust reference is [`crates/bridge-protocol`](../crates/bridge-protocol). The
   - `ANYPS5_NET_ENDPOINT`: `127.0.0.1:<port>` where the net service listens;
   - `ANYPS5_NET_TOKEN`: a random secret for this one game start, as lowercase hex.
 - Without both variables the bridge stays off, and every network library behaves as it does today: offline and signed out.
-- The bridge opens one TCP connection to the endpoint and sends **hello** as its first frame. The net service answers **welcome** or **reject**, then closes the connection after a reject. Nothing else is sent before welcome.
+- The bridge opens one TCP connection to the endpoint and sends **hello** as its first frame. The payload of the hello frame is at most 1024 bytes, and a larger first frame closes the connection. The net service answers **welcome** or **reject**, then closes the connection after a reject. Nothing else is sent before welcome.
 - The net service accepts connections from loopback addresses only, and one connection per token at a time; a new connection with the same token replaces the old one.
 - When the connection breaks, the bridge reports the player as signed out to the game and connects again every 2 seconds with the same token.
 
@@ -54,7 +54,7 @@ A payload may be longer than the fields its receiver knows; the receiver ignores
 
 | Kind | Name | Direction | Payload |
 |---|---|---|---|
-| `0x0001` | hello | bridge → service | `version: u16`, then the token as raw bytes (the decoded hex) |
+| `0x0001` | hello | bridge → service | `version: u16`, then the token as raw bytes (the decoded hex); at most 1024 bytes |
 | `0x0002` | welcome | service → bridge | `version: u16` (the version both sides will use), `udp_port: u16`, `tcp_port: u16`, `stream_key: 16 bytes` |
 | `0x0003` | reject | service → bridge | a UTF-8 reason for the log |
 | `0x8000` | unsupported | either way | empty; the reply to a request of unknown kind |