fix(core/net): bind the data plane to the authenticated peer + stop adaptive FEC wedging large frames

Four defects from the punktfunk-core quality sweep, all in the data plane. transport/udp: the hole-punch adopted the source address of ANY datagram whose first 8 bytes matched PUNCH_MAGIC — a fixed public constant with no key, nonce or session id — and the authenticated QUIC peer was passed only as the no-punch fallback, so it was never used to validate. Hole-punch is the default bring-up path (it is skipped only for a fixed --data-port), and the data socket is an OS ephemeral, so spraying the ephemeral range during the 2500 ms punch wait let anyone steal the video plane: the legitimate client is then filtered out by the connect() and gets nothing, while QUIC stays healthy so no reconnect fires. With a spoofed source the same 8 bytes aim a full-rate stream at a third party. Take the authenticated peer IP and require the punch to match it — only the PORT is in question (that is what a NAT remaps and what the punch exists to discover); the client dials the same host IP as its QUIC connection, so a NAT presents one source IP for both planes. Also budget each read from the REMAINING window, so off-peer datagrams cannot stretch the wait past punch_timeout. transport/udp: the punch keepalive treated every send error as fatal and broke out of its loop permanently and silently. It holds a clone of the connected, non-blocking data socket — exactly the socket whose transient conditions this module defines and documents in is_transient_io. One ENOBUFS from a full wlan tx queue or an ENETUNREACH during an AP roam killed the only thing holding the NAT mapping open; the path recovers, video keeps flowing, and the stream dies later when the idle timer expires the mapping during a static scene. Route it through is_transient_io like every other send site in the file. packet: adaptive FEC moved fec_percent live (host bands it 1..=50 while Welcome advertises 10) but the receiver's per-block acceptance ceiling was computed once from the negotiated percentage and never re-derived. Once FEC ramped, every packet of a maximal block failed `total > max_total_shards`, the block never accumulated a shard, the frame aged out, and the resulting loss drove FEC higher still — large frames wedged at 100% loss exactly when FEC was meant to rescue the link. Fixed on both sides, because hosts and clients update independently: the sender clamps per-block parity to the ceiling the peer negotiated, and the receiver sizes that ceiling from the whole clamp range rather than a stale snapshot of it. No wire bytes and no C ABI signature change; WIRE_VERSION and ABI_VERSION are unchanged. Regression tests cover all three (the punch tests were confirmed to fail without the fix). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-20 01:09:20 +02:00
parent 15d51bc0ff
commit 7b2cdf5a7a
5 changed files with 234 additions and 14 deletions
@@ -39,10 +39,19 @@ pub struct Packetizer {
    /// DATA shards before any block's parity — all blocks' parity must stay alive until the
    /// frame's second emission pass.
    recovery: Vec<Vec<Vec<u8>>>,
+    /// The peer's per-block `data + recovery` acceptance ceiling, frozen from the **negotiated**
+    /// config exactly as the far side derives it in [`ReassemblerLimits::from_config`]. Adaptive
+    /// FEC moves `fec.fec_percent` live ([`set_fec_percent`](Self::set_fec_percent)) but the
+    /// receiver's ceiling is computed once at session construction and never re-derived, so parity
+    /// must be clamped against this or a raised percentage puts blocks over the far side's bound —
+    /// where every packet of the block is dropped wholesale, the frame never completes, and the
+    /// resulting loss pushes adaptive FEC *higher*. See the `recovery_for` clamp in `packetize_each`.
+    max_total_shards: usize,
 }

 impl Packetizer {
    pub fn new(config: &Config) -> Self {
+        let max_data = config.fec.max_data_per_block as usize;
        Packetizer {
            next_frame_index: 0,
            next_probe_index: 0,
@@ -52,6 +61,9 @@ impl Packetizer {
            version: config.phase as u8,
            tail: Vec::new(),
            recovery: Vec::new(),
+            // Mirrors `ReassemblerLimits::from_config` — keep the two in step.
+            max_total_shards: (max_data + config.fec.recovery_for(max_data))
+                .min(config.fec.scheme.max_total_shards()),
        }
    }

@@ -173,6 +185,15 @@ impl Packetizer {
        };
        // Per-block shard geometry (deterministic — recomputed in both passes).
        let block_data_count = |b: usize| ((b + 1) * max_block).min(total_data) - b * max_block;
+        // Parity for a `k`-shard block: the configured percentage, clamped so the block's wire
+        // total never exceeds what the peer will accept (see `max_total_shards`). The clamp only
+        // binds on blocks near `max_data_per_block`; smaller blocks keep the full adaptive range,
+        // so raising FEC still buys real protection wherever there is headroom. Bound as locals,
+        // not as a `&self` method: `emit_one` below would otherwise capture all of `self` and
+        // collide with the `&mut self.recovery[b]` parity borrow.
+        let (fec, max_total_shards) = (self.fec, self.max_total_shards);
+        let recovery_for =
+            move |k: usize| fec.recovery_for(k).min(max_total_shards.saturating_sub(k));

        // One parity pool per block, reused across frames (steady-state zero-alloc).
        if self.recovery.len() < block_count {
@@ -183,7 +204,7 @@ impl Packetizer {
        let mut total_recovery = 0usize;
        for b in 0..block_count {
            let k = block_data_count(b);
-            let m = self.fec.recovery_for(k);
+            let m = recovery_for(k);
            if k + m > u16::MAX as usize {
                return Err(PunktfunkError::Unsupported("block shard count exceeds u16"));
            }
@@ -204,7 +225,7 @@ impl Packetizer {
                    block_index: b as u16,
                    block_count: block_count as u16,
                    data_shards: k as u16,
-                    recovery_shards: self.fec.recovery_for(k) as u16,
+                    recovery_shards: recovery_for(k) as u16,
                    shard_index: shard_index as u16,
                    shard_bytes: payload as u16,
                    magic: PUNKTFUNK_MAGIC,
@@ -223,7 +244,7 @@ impl Packetizer {

            // This block's data shards: references into `frame` (plus the staged tail).
            let data_shards: Vec<&[u8]> = (first..first + k).map(shard_at).collect();
-            let recovery_count = self.fec.recovery_for(k);
+            let recovery_count = recovery_for(k);
            coder.encode_into(&data_shards, recovery_count, &mut self.recovery[b])?;

            for (shard_index, body) in data_shards.iter().enumerate() {
@@ -242,7 +263,7 @@ impl Packetizer {
        let mut parity_left = total_recovery;
        for b in 0..block_count {
            let k = block_data_count(b);
-            let recovery_count = self.fec.recovery_for(k);
+            let recovery_count = recovery_for(k);
            for r in 0..recovery_count {
                parity_left -= 1;
                let mut flags = FLAG_PIC;
@@ -106,8 +106,19 @@ pub struct ReassemblerLimits {
 impl ReassemblerLimits {
    pub fn from_config(c: &Config) -> Self {
        let max_data = c.fec.max_data_per_block as usize;
+        // Size the ceiling from the whole range adaptive FEC may reach, NOT from the percentage
+        // negotiated at session start: the sender moves `fec_percent` live (`Packetizer::
+        // set_fec_percent`, clamped to ≤ 90) and the wire is self-describing, so it never
+        // renegotiates. Deriving this from the start value made every packet of a large block
+        // fail the `total > max_total_shards` check once FEC ramped up — the block never
+        // accumulated a shard, the frame aged out, and the resulting loss drove FEC *higher*,
+        // wedging large frames at 100% loss exactly when FEC was meant to rescue the link. A
+        // current sender also clamps its side (`Packetizer::recovery_for`); this keeps an
+        // already-deployed sender that doesn't from wedging a current receiver. Still a hard
+        // pre-allocation bound against hostile headers — just the sender's clamp, not a stale
+        // snapshot of it.
        let max_total =
-            (max_data + c.fec.recovery_for(max_data)).min(c.fec.scheme.max_total_shards());
+            (max_data + (max_data * 90).div_ceil(100)).min(c.fec.scheme.max_total_shards());
        let total_data = c.max_frame_bytes.div_ceil(c.shard_payload.max(1)).max(1);
        ReassemblerLimits {
            shard_bytes: c.shard_payload,
@@ -579,3 +579,64 @@ fn rejects_wrong_shard_bytes_and_oversized_frame() {
        .is_none());
    assert_eq!(stats.snapshot().packets_dropped, 1);
 }
+
+/// Adaptive FEC raises `fec_percent` mid-session while the receiver's per-block acceptance
+/// ceiling is frozen at session construction and never renegotiated. A maximal block must
+/// therefore still land: the sender clamps its parity to the ceiling
+/// (`Packetizer::recovery_for`), and the receiver sizes that ceiling from the whole clamp range
+/// rather than the start percentage. Regression guard for the wedge this caused — every packet
+/// of a large block failing `total > max_total_shards`, so the frame never completed and the
+/// resulting loss drove adaptive FEC higher still.
+#[test]
+fn adaptive_fec_ramp_keeps_maximal_blocks_within_the_peers_ceiling() {
+    let cfg = e2e_config(FecScheme::Gf16, 10);
+    let coder = coder_for(FecScheme::Gf16);
+    let lim = ReassemblerLimits::from_config(&cfg);
+    let mut pk = Packetizer::new(&cfg);
+
+    // Ramp far past the negotiated 10% — exactly what `apply_fec_target` does under loss.
+    pk.set_fec_percent(50);
+
+    // A frame of full `max_data_per_block` blocks: where the ceiling actually binds.
+    let frame_len = cfg.shard_payload * cfg.fec.max_data_per_block as usize * 2;
+    let src: Vec<u8> = (0..frame_len).map(|i| (i * 131 + 7) as u8).collect();
+    let pkts = pk.packetize(&src, 1, 0, coder.as_ref()).unwrap();
+
+    let k = cfg.fec.max_data_per_block as usize;
+    let mut clamped = false;
+    for p in &pkts {
+        let hdr = PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap();
+        let total = hdr.data_shards as usize + hdr.recovery_shards as usize;
+        assert!(
+            total <= lim.max_total_shards,
+            "block total {total} exceeds the peer's ceiling {} — every packet of this block \
+             would be dropped",
+            lim.max_total_shards
+        );
+        // The unclamped 50% would put 2 parity on a full block; the negotiated 10% ceiling
+        // leaves room for 1. Proves the clamp actually bound rather than passing vacuously.
+        if hdr.data_shards as usize == k {
+            assert!(
+                (hdr.recovery_shards as usize) < cfg.fec.recovery_for(k).max(1) + 1,
+                "parity must be clamped to the peer's ceiling"
+            );
+            clamped = true;
+        }
+    }
+    assert!(clamped, "test must exercise a maximal block");
+
+    // And the frame still reassembles byte-identically.
+    let mut r = Reassembler::new(lim);
+    let stats = StatsCounters::default();
+    let mut got = None;
+    for p in &pkts {
+        if let Some(f) = r.push(p, coder.as_ref(), &stats).unwrap() {
+            got = Some(f);
+        }
+    }
+    assert_eq!(
+        got.expect("frame must complete after an adaptive-FEC ramp")
+            .data,
+        src
+    );
+}
@@ -110,8 +110,18 @@ pub fn spawn_data_punch(sock: UdpSocket, stop: std::sync::Arc<std::sync::atomic:
        .spawn(move || {
            let mut i = 0u32;
            while !stop.load(std::sync::atomic::Ordering::Relaxed) {
-                if sock.send(PUNCH_MAGIC).is_err() {
-                    break;
+                match sock.send(PUNCH_MAGIC) {
+                    Ok(_) => {}
+                    // Same contract as `Transport::send`: a momentarily full tx queue, a stale
+                    // ICMP or a network-path blip is a lossy drop, not a reason to stop holding
+                    // the NAT/firewall path open. Breaking here is silent and permanent — the
+                    // path recovers, video keeps flowing, and the stream dies later when the
+                    // idle timer expires the mapping during a static scene.
+                    Err(e) if is_transient_io(&e) => {}
+                    Err(e) => {
+                        tracing::debug!(error = %e, "data-plane punch send failed — stopping keepalive");
+                        break;
+                    }
                }
                let delay_ms = if i < 15 { 200 } else { 2000 };
                i = i.saturating_add(1);
@@ -160,34 +170,65 @@ impl UdpTransport {
    /// NAT-translated one, which can differ from the client-reported `fallback_peer`). If no punch
    /// arrives (a client that doesn't hole-punch), fall back to `fallback_peer` — the same flat-LAN
    /// behaviour as [`connect`](Self::connect). Returns `(transport, punched)`.
+    ///
+    /// `expect_ip` is the *authenticated* peer address (the QUIC connection's remote IP) — see
+    /// [`from_socket_punch`](Self::from_socket_punch) for why only punches from it are honoured.
    pub fn connect_via_punch(
        local: &str,
        fallback_peer: &str,
+        expect_ip: std::net::IpAddr,
        punch_timeout: std::time::Duration,
    ) -> std::io::Result<(Self, bool)> {
-        Self::from_socket_punch(UdpSocket::bind(local)?, fallback_peer, punch_timeout)
+        Self::from_socket_punch(
+            UdpSocket::bind(local)?,
+            fallback_peer,
+            expect_ip,
+            punch_timeout,
+        )
    }

    /// [`connect_via_punch`](Self::connect_via_punch) on an already-bound socket — see
    /// [`from_socket`](Self::from_socket) for why the host binds the data port up front.
+    ///
+    /// `expect_ip` binds the data plane to the peer the control plane already authenticated.
+    /// [`PUNCH_MAGIC`] is a fixed public constant carrying no key, nonce or session id, so without
+    /// this check *any* source that lands an 8-byte datagram on the (ephemeral, sprayable) data
+    /// port during the punch wait becomes the video destination — the legitimate client is then
+    /// filtered out by the `connect` below and receives nothing, while QUIC stays healthy so no
+    /// reconnect is triggered. Only the *port* is in question here (that is what a NAT remaps, and
+    /// what the punch exists to discover); the IP is known, because the client binds `0.0.0.0:0`
+    /// and dials the same host IP as its QUIC connection, so the kernel picks the same source IP
+    /// for both planes and any NAT on the path presents one source IP for both.
    pub fn from_socket_punch(
        socket: UdpSocket,
        fallback_peer: &str,
+        expect_ip: std::net::IpAddr,
        punch_timeout: std::time::Duration,
    ) -> std::io::Result<(Self, bool)> {
-        socket.set_read_timeout(Some(punch_timeout))?;
        let deadline = std::time::Instant::now() + punch_timeout;
        let mut buf = [0u8; 64];
        let mut observed: Option<std::net::SocketAddr> = None;
        loop {
+            // Budget the read from what's LEFT, not the full window: off-peer datagrams are
+            // discarded below, and a full-window timeout per read would let a stray flood stretch
+            // the punch wait far past `punch_timeout`.
+            let remaining = deadline.saturating_duration_since(std::time::Instant::now());
+            if remaining.is_zero() {
+                break;
+            }
+            socket.set_read_timeout(Some(remaining))?;
            match socket.recv_from(&mut buf) {
                Ok((n, src))
-                    if n >= PUNCH_MAGIC.len() && &buf[..PUNCH_MAGIC.len()] == PUNCH_MAGIC =>
+                    if src.ip() == expect_ip
+                        && n >= PUNCH_MAGIC.len()
+                        && &buf[..PUNCH_MAGIC.len()] == PUNCH_MAGIC =>
                {
                    observed = Some(src);
                    break;
                }
-                Ok(_) => {} // stray datagram — keep waiting for a real punch
+                // Stray, or a well-formed punch from someone who isn't the authenticated peer —
+                // keep waiting for a real one.
+                Ok(_) => {}
                Err(e)
                    if matches!(
                        e.kind(),
@@ -198,9 +239,6 @@ impl UdpTransport {
                }
                Err(e) => return Err(e),
            }
-            if std::time::Instant::now() >= deadline {
-                break;
-            }
        }
        let punched = observed.is_some();
        let target = observed.map(|s| s.to_string());
@@ -482,4 +520,89 @@ mod tests {
            "every datagram should be drained via recv_batch"
        );
    }
+
+    /// The punch discovers the peer's NAT-remapped *port*, so a punch from the authenticated IP on
+    /// a port that differs from the client-reported one must still be adopted — that is the whole
+    /// reason hole-punching exists, and the source-IP check must not break it.
+    #[test]
+    fn punch_adopts_remapped_port_from_the_authenticated_peer() {
+        // Stands in for the client's post-NAT data socket: same IP as the "QUIC peer", new port.
+        let puncher = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
+        puncher
+            .set_read_timeout(Some(std::time::Duration::from_millis(500)))
+            .unwrap();
+        // The client-*reported* address, which the NAT remapped — video must NOT go here.
+        let reported = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
+        reported
+            .set_read_timeout(Some(std::time::Duration::from_millis(200)))
+            .unwrap();
+
+        let host_sock = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
+        let host_addr = host_sock.local_addr().unwrap();
+        puncher.send_to(PUNCH_MAGIC, host_addr).unwrap();
+
+        let (transport, punched) = UdpTransport::from_socket_punch(
+            host_sock,
+            &reported.local_addr().unwrap().to_string(),
+            std::net::IpAddr::from([127, 0, 0, 1]),
+            std::time::Duration::from_millis(500),
+        )
+        .unwrap();
+        assert!(punched, "a punch from the authenticated IP must be adopted");
+
+        transport.send(b"video").unwrap();
+        let mut buf = [0u8; 64];
+        let n = puncher
+            .recv(&mut buf)
+            .expect("video must follow the punched (NAT-remapped) port");
+        assert_eq!(&buf[..n], b"video");
+        assert!(
+            reported.recv(&mut buf).is_err(),
+            "video must not go to the stale reported port"
+        );
+    }
+
+    /// A punch from any source other than the QUIC-authenticated peer must be ignored: `PUNCH_MAGIC`
+    /// is a fixed public constant with no key or session id, so honouring an off-peer punch lets
+    /// anyone who lands an 8-byte datagram on the ephemeral data port steal (or redirect) the video
+    /// plane while the control plane stays healthy. Falling back to the reported address is correct.
+    #[test]
+    fn punch_from_an_unauthenticated_source_is_ignored() {
+        let attacker = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
+        attacker
+            .set_read_timeout(Some(std::time::Duration::from_millis(200)))
+            .unwrap();
+        let legit = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
+        legit
+            .set_read_timeout(Some(std::time::Duration::from_millis(500)))
+            .unwrap();
+
+        let host_sock = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
+        let host_addr = host_sock.local_addr().unwrap();
+        attacker.send_to(PUNCH_MAGIC, host_addr).unwrap();
+
+        // The authenticated peer is TEST-NET-1, so nothing arriving over loopback is the peer.
+        let (transport, punched) = UdpTransport::from_socket_punch(
+            host_sock,
+            &legit.local_addr().unwrap().to_string(),
+            std::net::IpAddr::from([192, 0, 2, 1]),
+            std::time::Duration::from_millis(300),
+        )
+        .unwrap();
+        assert!(
+            !punched,
+            "an off-peer punch must not be adopted as the video destination"
+        );
+
+        transport.send(b"video").unwrap();
+        let mut buf = [0u8; 64];
+        assert!(
+            attacker.recv(&mut buf).is_err(),
+            "video must never be redirected to the punch source"
+        );
+        let n = legit
+            .recv(&mut buf)
+            .expect("video falls back to the reported peer address");
+        assert_eq!(&buf[..n], b"video");
+    }
 }
@@ -1265,6 +1265,10 @@ async fn serve_session(
                UdpTransport::from_socket_punch(
                    data_sock,
                    &client_udp.to_string(),
+                    // Only honour a punch from the peer QUIC already authenticated: the punch is
+                    // there to discover the NAT-remapped *port*, and `client_udp`'s IP is the
+                    // host-observed QUIC remote (only its port is client-reported).
+                    client_udp.ip(),
                    std::time::Duration::from_millis(2500),
                )
            };