diff --git a/crates/punktfunk-core/src/packet/packetize.rs b/crates/punktfunk-core/src/packet/packetize.rs index 0112b527..e9f875ef 100644 --- a/crates/punktfunk-core/src/packet/packetize.rs +++ b/crates/punktfunk-core/src/packet/packetize.rs @@ -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>>, + /// 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; diff --git a/crates/punktfunk-core/src/packet/reassemble.rs b/crates/punktfunk-core/src/packet/reassemble.rs index 5fb498e5..b91f6b14 100644 --- a/crates/punktfunk-core/src/packet/reassemble.rs +++ b/crates/punktfunk-core/src/packet/reassemble.rs @@ -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, diff --git a/crates/punktfunk-core/src/packet/tests.rs b/crates/punktfunk-core/src/packet/tests.rs index 2736aac6..003fb0bc 100644 --- a/crates/punktfunk-core/src/packet/tests.rs +++ b/crates/punktfunk-core/src/packet/tests.rs @@ -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 = (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 + ); +} diff --git a/crates/punktfunk-core/src/transport/udp/mod.rs b/crates/punktfunk-core/src/transport/udp/mod.rs index cb58ba85..7c147020 100644 --- a/crates/punktfunk-core/src/transport/udp/mod.rs +++ b/crates/punktfunk-core/src/transport/udp/mod.rs @@ -110,8 +110,18 @@ pub fn spawn_data_punch(sock: UdpSocket, stop: std::sync::Arc {} + // 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 = 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"); + } } diff --git a/crates/punktfunk-host/src/native.rs b/crates/punktfunk-host/src/native.rs index c29d5b25..4bfe60f3 100644 --- a/crates/punktfunk-host/src/native.rs +++ b/crates/punktfunk-host/src/native.rs @@ -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), ) };