port forward: a tunnel's TCP stream refuses packets over twice MAX_FRAME

The codec takes a header declaring up to 1 GiB and hands the packet up
only once it has all arrived, so the channel window bounded what the
peer may send, not what this side buffers. Both sides now cap the codec
at 2 * MAX_FRAME as soon as multiplexing is agreed: a data frame with
its envelope and MAC fits with room to spare, and a header over the cap
ends the tunnel before a byte of payload is read. TCP only; the
WebSocket and WebRTC codecs carry caps of their own.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EZ49AbZJYfm8NTp5yDPMab

src/port_forward_mux.rs

@@ -22,6 +22,11 @@ pub const INITIAL_WINDOW: u32 = 64 * 1024;
 pub const MIN_FRAME_CHARGE: u32 = 64;
 pub const CHANNEL_WINDOW: u32 = 256 * 1024;
 pub const MAX_FRAME: usize = 64 * 1024;
+/// Cap on one framed packet once the tunnel is up: a `MAX_FRAME` data frame,
+/// its protobuf envelope and the 16-byte MAC fit with room to spare. The codec
+/// otherwise takes a header declaring up to 1 GiB, and the channel window is
+/// only checked once the whole packet has arrived.
+pub const MAX_PACKET: usize = 2 * MAX_FRAME;
 pub const UPDATE_THRESHOLD: u32 = CHANNEL_WINDOW / 2;
 pub const MAX_CHANNELS: usize = 256;
 pub const DATA_QUEUE_FRAMES: usize = 128;
@@ -34,6 +39,14 @@ pub fn effective_window(advertised: u32) -> u32 {
     advertised.max(INITIAL_WINDOW)
 }
 
+/// For a tunnel's stream once multiplexing is agreed, on both sides. The
+/// WebSocket and WebRTC codecs carry caps of their own.
+pub fn cap_packet_size(stream: &mut hbb_common::Stream) {
+    if let hbb_common::Stream::Tcp(s) = stream {
+        s.0.codec_mut().set_max_packet_length(MAX_PACKET);
+    }
+}
+
 /// What a `data` frame of this length costs its channel's window.
 pub fn charge(len: usize) -> u32 {
     u32::try_from(len).unwrap_or(u32::MAX).max(MIN_FRAME_CHARGE)
@@ -466,7 +479,8 @@ mod tunnel {
             }
         }
 
-        pub fn set_muxed(&self, stream: Stream, interface: impl Interface) -> Arc<TunnelHandle> {
+        pub fn set_muxed(&self, mut stream: Stream, interface: impl Interface) -> Arc<TunnelHandle> {
+            cap_packet_size(&mut stream);
             let (data_tx, data_rx) = mpsc::channel(DATA_QUEUE_FRAMES);
             let (control_tx, control_rx) = mpsc::unbounded_channel();
             let handle = Arc::new(TunnelHandle {
@@ -1150,6 +1164,29 @@ mod tests {
             (a, b)
         }
 
+        #[test]
+        fn a_packet_declared_over_the_cap_ends_the_tunnel_before_it_arrives() {
+            rt().block_on(async {
+                use std::time::{Duration, Instant};
+                let (ours, mut theirs) = local_pair().await;
+                let addr = ours.peer_addr().unwrap();
+                let t = Tunnel::new();
+                t.claim();
+                let _h = t.set_muxed(Stream::Tcp(FramedStream::from(ours, addr)), NoUi::default());
+                assert!(matches!(t.claim(), Claim::Muxed(_)));
+                // The codec's three-byte header form, declaring one byte more
+                // than the cap, and nothing behind it: an uncapped codec waits
+                // for the whole packet and the tunnel stays up.
+                let head = ((MAX_PACKET as u32 + 1) << 2) | 0x2;
+                theirs.write_all(&head.to_le_bytes()[..3]).await.unwrap();
+                let deadline = Instant::now() + Duration::from_secs(2);
+                while !matches!(t.claim(), Claim::Claimed) {
+                    assert!(Instant::now() < deadline, "tunnel still up: the oversized header was accepted");
+                    tokio::time::sleep(Duration::from_millis(20)).await;
+                }
+            });
+        }
+
         #[test]
         fn claim_follows_the_tunnel_state() {
             let t = Tunnel::new();

src/server/connection.rs

@@ -1668,6 +1668,7 @@ impl Connection {
             return true;
         };
         if pf.multiplex {
+            crate::port_forward_mux::cap_packet_size(&mut self.stream);
             // `inner.tx` is set for the connection's whole life; `None` here is
             // unreachable, and refusing the login is the only honest answer.
             self.port_forward_mux = self.inner.tx.clone().map(|tx| {