diff --git a/docs/adr/ADR-380-loopback-udp-tee.md b/docs/adr/ADR-380-loopback-udp-tee.md new file mode 100644 index 000000000..bd7924e68 --- /dev/null +++ b/docs/adr/ADR-380-loopback-udp-tee.md @@ -0,0 +1,58 @@ +# ADR 380: Loopback UDP tee for second consumers of the CSI stream + +## Status + +Proposed. Unit tested, and smoke tested on loopback: a datagram sent to the +server arrived byte for byte at the tee target. Not yet exercised against a +live multi node fleet. + +Date: 2026-09-28 + +Numbering: On 2026-10-02 `main` had up to ADR 378 and open PRs claimed nothing +higher; ADR 379 is the OTA first-boot health check. If another branch also +takes 380, renumber whichever merges second. + +## Context + +A five node deployment (four ESP32 S3, one ESP32 C6) streams ADR 018 frames to +one sensing server. A second consumer cannot see the raw stream. + +Each node sends to a single `target_ip` (`firmware/esp32-csi-node/main/stream_sender.c`). +The server's WebSocket and REST outputs carry per node amplitude only: no +phase, no I/Q, no sequence number, and no `0xC511A110` sync packets. A fusion +host that aligns CSI with camera, radar, or lidar needs all of those. Binding a +second listener to the CSI port is not an option, and an external relay in +front of the port would make every source loopback, which the ADR 296 +allowlist always admits. + +## Decision + +`--udp-tee` (env `RUVIEW_UDP_TEE`) copies every datagram the source allowlist +admits, byte for byte, to up to four `ip:port` targets. + +- Targets must be loopback, so raw CSI never leaves the host through the tee. +- A target may not use port 0 or the listener's own port, which would loop. +- Sends are non blocking. A copy that cannot be queued is dropped and counted, + so a slow consumer never stalls the receive loop. +- The copy happens after the ADR 296 allowlist check and before any parsing, + so consumers see exactly what the server admitted, in every wire format. +- The tee is off unless configured. + +## Consequences + +A fusion host or Cognitum cog can subscribe to the full data plane by reading +a loopback port. A consumer that today binds the CSI port itself can keep its +code and move to the tee port instead. + +The tee is byte transparent, so it needs no change when the ADR 018 header or +a later wire version changes. + +Each target costs one `send_to` per admitted datagram. Four targets is the +limit. + +## Validation + +```bash +cd v2 +cargo test -p wifi-densepose-sensing-server --no-default-features --lib udp_tee +``` diff --git a/docs/adr/README.md b/docs/adr/README.md index c5051feaa..8fb7eaf39 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -2,6 +2,7 @@ Latest proposed decisions: +- [ADR-380: Loopback UDP tee for second consumers of the CSI stream](ADR-380-loopback-udp-tee.md) - [ADR-378: `ruview-live` showcase views — CSI waterfall, radar fan, animated cells](ADR-378-ruview-live-showcase-views.md) - [ADR-377: `ruview-live` — a Claude Code mod shipped inside `@ruvnet/ruview`](ADR-377-ruview-live-claude-code-mod.md) - [ADR-376: `ruview` — one npm install for every RuView component](ADR-376-ruview-umbrella-npm-package.md) @@ -181,6 +182,7 @@ Statuses: **Proposed** (under discussion), **Accepted** (approved and/or impleme | [ADR-294](ADR-294-wifi-veil-integration.md) | WiFi Veil integration — emission-shaping countermeasure as advisory BFLD dependency | Accepted (initial implementation) | | [ADR-295](ADR-295-source-provenance-state-machine.md) | Source provenance state machine — synthetic can never present as live | Accepted (initial implementation) | | [ADR-296](ADR-296-sensor-data-plane-bind-hardening.md) | Sensor data-plane hardening — UDP bind control and source allowlist (step one) | Accepted (initial implementation) | +| [ADR-380](ADR-380-loopback-udp-tee.md) | Loopback UDP tee for second consumers of the CSI stream (`--udp-tee`) | Proposed | | [ADR-297](ADR-297-multi-node-semantic-correctness.md) | Multi-node semantic correctness — per-node inference, node-keyed rate limiting, stale state | Accepted (initial implementation) | | [ADR-298](ADR-298-model-release-sanity-gates.md) | Model release sanity gates — block degenerate and mislabeled model artifacts | Accepted (initial implementation) | | [ADR-299](ADR-299-csi-data-incident-repo-controls.md) | Repository CSI data-incident controls — ignore rules and pre-commit/CI policy check | Accepted (controls implemented; tree remediation gated) | diff --git a/v2/crates/wifi-densepose-sensing-server/src/lib.rs b/v2/crates/wifi-densepose-sensing-server/src/lib.rs index 0cf0214f1..72f31b3c1 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/lib.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/lib.rs @@ -43,6 +43,8 @@ pub mod telemetry; pub mod trainer; /// ADR-296: UDP data-plane bind scope decision + source IP/CIDR allowlist. pub mod udp_bind; +/// ADR-380: loopback raw-datagram tee for second consumers of the CSI stream. +pub mod udp_tee; pub mod vital_signs; /// ADR-270 Mist and NETGEAR telemetry providers. pub mod vendor_mist_netgear; diff --git a/v2/crates/wifi-densepose-sensing-server/src/main.rs b/v2/crates/wifi-densepose-sensing-server/src/main.rs index 180ae6f53..aa3e69c7d 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/main.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/main.rs @@ -129,6 +129,14 @@ struct Args { #[arg(long, env = "RUVIEW_UDP_INSECURE_LAN")] udp_insecure_lan: bool, + /// Copy every admitted UDP datagram, unchanged, to these loopback + /// `ip:port` targets (comma-separated, repeatable; env `RUVIEW_UDP_TEE`; + /// ADR-380). Lets a second consumer read full I/Q and sync packets without + /// binding the CSI port. Off by default. + /// Example: `--udp-tee 127.0.0.1:5007`. + #[arg(long = "udp-tee", value_name = "IP:PORT", env = "RUVIEW_UDP_TEE")] + udp_tee: Vec, + /// Path to UI static files (repo `ui/`; from `v2/` use `../ui` or rely on auto-detect) #[arg(long, default_value = "../ui")] ui_path: PathBuf, @@ -9709,6 +9717,7 @@ async fn udp_receiver_task( bind_ip: std::net::IpAddr, udp_port: u16, allowlist: std::sync::Arc, + tee: Option>, ) { let addr = format!("{bind_ip}:{udp_port}"); let socket = match UdpSocket::bind(&addr).await { @@ -9735,6 +9744,10 @@ async fn udp_receiver_task( ); continue; } + // ADR-380: second consumers see exactly what was admitted. + if let Some(tee) = &tee { + tee.forward(&buf[..len]); + } if len > 0 && buf[0] == b'{' { match serde_json::from_slice::(&buf[..len]) .map_err(|error| error.to_string()) @@ -12289,7 +12302,7 @@ async fn main() { if plan.bind_udp { // ADR-296: resolve the UDP bind scope + source allowlist and fail closed // on an unguarded routable bind, mirroring the OAuth boot refusal below. - use wifi_densepose_sensing_server::udp_bind; + use wifi_densepose_sensing_server::{udp_bind, udp_tee}; let udp_bind_ip: std::net::IpAddr = match args.udp_bind.parse() { Ok(ip) => ip, Err(_) => { @@ -12323,11 +12336,29 @@ async fn main() { std::process::exit(1); } } + let udp_tee = match udp_tee::parse_targets(args.udp_tee.iter(), args.udp_port) { + Ok(targets) if targets.is_empty() => None, + Ok(targets) => match udp_tee::UdpTee::bind(targets) { + Ok(tee) => { + info!("UDP tee: copying admitted datagrams to {:?}", tee.targets()); + Some(std::sync::Arc::new(tee)) + } + Err(e) => { + error!("Failed to open UDP tee socket: {e}"); + std::process::exit(1); + } + }, + Err(e) => { + error!("Invalid --udp-tee: {e}"); + std::process::exit(1); + } + }; tokio::spawn(udp_receiver_task( state.clone(), udp_bind_ip, args.udp_port, udp_allowlist, + udp_tee, )); tokio::spawn(broadcast_tick_task(state.clone(), args.tick_ms)); } diff --git a/v2/crates/wifi-densepose-sensing-server/src/udp_tee.rs b/v2/crates/wifi-densepose-sensing-server/src/udp_tee.rs new file mode 100644 index 000000000..95f644994 --- /dev/null +++ b/v2/crates/wifi-densepose-sensing-server/src/udp_tee.rs @@ -0,0 +1,178 @@ +//! ADR-380: raw-datagram tee for the UDP data plane. +//! +//! The sensing server owns the CSI UDP port, and its WebSocket/REST outputs +//! carry amplitude only. A second consumer (a fusion host, a Cognitum cog, a +//! recorder) that needs full I/Q, sequence numbers and the `0xC511A110` sync +//! packets gets them from this tee: every datagram the source allowlist admits +//! is copied unchanged to each configured target. +//! +//! Targets are loopback-only, so the tee can never send raw CSI off the host, +//! and may not use the listener's own port, which would loop datagrams back +//! into the receiver (loopback is always admitted by the allowlist). Sends are +//! non-blocking; a datagram that cannot be queued is dropped and counted rather +//! than stalling the receive loop. + +use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr, UdpSocket}; +use std::sync::atomic::{AtomicU64, Ordering}; + +/// Upper bound on tee targets; each admitted datagram costs one send per target. +pub const MAX_TEE_TARGETS: usize = 4; + +/// Parse tee targets from CLI/env specs. Each item may be a comma-separated +/// list of `ip:port`. Returns an error on the first unusable entry rather than +/// silently dropping it. +pub fn parse_targets(specs: I, listen_port: u16) -> Result, String> +where + I: IntoIterator, + S: AsRef, +{ + let mut targets: Vec = Vec::new(); + for spec in specs { + for part in spec.as_ref().split(',') { + let part = part.trim(); + if part.is_empty() { + continue; + } + let addr: SocketAddr = part + .parse() + .map_err(|_| format!("tee target '{part}' is not ip:port"))?; + if !addr.ip().is_loopback() { + return Err(format!("tee target '{part}' must be a loopback address")); + } + if addr.port() == 0 || addr.port() == listen_port { + return Err(format!( + "tee target '{part}' may not use port 0 or the listener port {listen_port}" + )); + } + if targets.contains(&addr) { + continue; + } + if targets.len() >= MAX_TEE_TARGETS { + return Err(format!("too many tee targets (limit {MAX_TEE_TARGETS})")); + } + targets.push(addr); + } + } + Ok(targets) +} + +/// Non-blocking forwarder that copies admitted datagrams to loopback targets. +#[derive(Debug)] +pub struct UdpTee { + v4: Option, + v6: Option, + targets: Vec, + forwarded: AtomicU64, + dropped: AtomicU64, +} + +impl UdpTee { + /// Bind one ephemeral loopback socket per address family in use. + pub fn bind(targets: Vec) -> std::io::Result { + let open = |ip: std::net::IpAddr| -> std::io::Result { + let socket = UdpSocket::bind(SocketAddr::new(ip, 0))?; + socket.set_nonblocking(true)?; + Ok(socket) + }; + let v4 = if targets.iter().any(SocketAddr::is_ipv4) { + Some(open(Ipv4Addr::LOCALHOST.into())?) + } else { + None + }; + let v6 = if targets.iter().any(SocketAddr::is_ipv6) { + Some(open(Ipv6Addr::LOCALHOST.into())?) + } else { + None + }; + Ok(Self { + v4, + v6, + targets, + forwarded: AtomicU64::new(0), + dropped: AtomicU64::new(0), + }) + } + + /// Copy one datagram to every target. Never blocks. + pub fn forward(&self, datagram: &[u8]) { + for target in &self.targets { + let socket = if target.is_ipv4() { &self.v4 } else { &self.v6 }; + let sent = socket + .as_ref() + .is_some_and(|s| s.send_to(datagram, target).is_ok()); + let counter = if sent { &self.forwarded } else { &self.dropped }; + counter.fetch_add(1, Ordering::Relaxed); + } + } + + /// Configured targets, for the startup summary. + pub fn targets(&self) -> &[SocketAddr] { + &self.targets + } + + /// Datagram copies successfully handed to the OS. + pub fn forwarded(&self) -> u64 { + self.forwarded.load(Ordering::Relaxed) + } + + /// Datagram copies that could not be sent (buffer full, no listener error). + pub fn dropped(&self) -> u64 { + self.dropped.load(Ordering::Relaxed) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + + #[test] + fn parses_loopback_targets_and_ignores_blanks_and_duplicates() { + let t = parse_targets(["127.0.0.1:5007, ,127.0.0.1:5007", "[::1]:5008"], 5006).unwrap(); + assert_eq!(t.len(), 2); + } + + #[test] + fn rejects_non_loopback_targets() { + let err = parse_targets(["192.0.2.10:5007"], 5006).unwrap_err(); + assert!(err.contains("loopback"), "{err}"); + } + + #[test] + fn rejects_the_listener_port_to_prevent_a_loop() { + assert!(parse_targets(["127.0.0.1:5006"], 5006).is_err()); + assert!(parse_targets(["127.0.0.1:0"], 5006).is_err()); + } + + #[test] + fn rejects_malformed_and_excess_targets() { + assert!(parse_targets(["localhost:5007"], 5006).is_err()); + let many: Vec = (0..=MAX_TEE_TARGETS) + .map(|i| format!("127.0.0.1:{}", 6000 + i)) + .collect(); + assert!(parse_targets(many, 5006).is_err()); + } + + #[test] + fn empty_spec_means_no_tee() { + assert!(parse_targets(Vec::::new(), 5006) + .unwrap() + .is_empty()); + } + + #[test] + fn forwards_datagrams_byte_for_byte() { + let sink = UdpSocket::bind("127.0.0.1:0").unwrap(); + sink.set_read_timeout(Some(Duration::from_secs(2))).unwrap(); + let tee = UdpTee::bind(vec![sink.local_addr().unwrap()]).unwrap(); + + let datagram = [0x01, 0x00, 0x11, 0xC5, 7, 1, 64, 0, 0xFF]; + tee.forward(&datagram); + + let mut buf = [0u8; 64]; + let (len, _) = sink.recv_from(&mut buf).unwrap(); + assert_eq!(&buf[..len], &datagram); + assert_eq!(tee.forwarded(), 1); + assert_eq!(tee.dropped(), 0); + } +}