From 2413646b6a94e7c4287ab5e45802f23684cbbbfa Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Fri, 25 Sep 2026 13:20:14 +0000 Subject: [PATCH] =?UTF-8?q?door:=20the=20UDP=20front=20door=20=E2=80=94=20?= =?UTF-8?q?the=20RFC=20codec=20boundary=20(ADR-003)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - src/door.rs (feature local): UdpFrontDoor — a sibling of the TCP door (one relay UDP socket = one association; no shared accept loop). Inbound datagrams decode via parse_udp_request at the door's codec boundary and ride in-band as {addr, data} frames; replies encode via new_udp_header back to the client's learned source address. FRAG != 0 refused (dropped; association healthy). The association's lifetime is the channel's — the relay socket closes with the pump and the teardown reaps the adopted channel. The client idle bound (ADR-014) is constructor config. - Pump shape: the association's stream is split (one pump owns the in-band write half, the other the read half) — a shared-session mutex deadlocks (the channel read parks holding it while the socket side's send needs it); the pumps join via select!. - tests/door-udp.rs: the POC #2 test-8 shape end to end (real UDP echo target + local::udp_egress + vanilla RFC-header'd datagram), the empty-datagram invariant (F-2), FRAG refusal + health, the idle bound + fresh-association health. Verification: cargo test green (default + all-features + local), clippy -D warnings clean (default, local, wasm32 target), fmt clean, cargo doc clean. --- src/door.rs | 294 ++++++++++++++++++++++- src/lib.rs | 2 +- tasks/implementation/door-udp.md | 58 ++++- tests/door-udp.rs | 384 +++++++++++++++++++++++++++++++ 4 files changed, 732 insertions(+), 6 deletions(-) create mode 100644 tests/door-udp.rs diff --git a/src/door.rs b/src/door.rs index 1b9b228..531451d 100644 --- a/src/door.rs +++ b/src/door.rs @@ -63,6 +63,35 @@ //! `serve` runs the accept loop; per-connection failures are logged + //! isolated (the loop continues — the alktunnels per-accept isolation //! posture). +//! +//! **The UDP door** ([`UdpFrontDoor`], the POC #2 test-8 shape): the +//! ASSOCIATE front door — a real UDP socket (the relay socket; the +//! vanilla client `sendto()`s it) bridged onto one in-band +//! association. The RFC UDP-header codec lives ONLY here (ADR-003's +//! codec-at-doors-only): inbound datagrams decode through +//! `parse_udp_request` and re-send in-band as `{addr, data}` frames +//! (`send_datagram`); replies encode through `new_udp_header` and go +//! back to the client's source address. **FRAG ≠ 0 → refuse** (the +//! datagram is dropped — the v1 posture; fragmentation cannot be +//! expressed in-band, and the association stays healthy). The +//! datagram pump is the two-pump shape (one pump per direction, +//! joined — the ADR-050 contract). Zero payload translation. The +//! association's lifetime is the channel's (ADR-003): the association +//! ends when the channel ends; the door's relay socket closes with +//! it. The door-side idle bound (ADR-014): no datagrams from the +//! client within the configured window closes the association (a +//! relay socket is not a channel — the cap regime does not bound +//! it). The reply's sentinel BND.ADDR is received and never used — +//! the handle is the channel; the real relay address is the door's +//! own bound socket, which the vanilla client learns from the TCP +//! door's ASSOCIATE reply (the assembly layer wires the two; the UDP +//! door serves the datagram plane alone). The bind address is +//! constructor config (consumer.md: "the UDP door's bind address is +//! assembly-layer configuration"). Surface shape (recorded per the +//! task spec): a **sibling** `UdpFrontDoor` with its own builder — +//! the UDP door shares no accept loop with the TCP door (UDP has no +//! connections; one relay socket = one association) and the shared +//! shape would have forced a mode split on every knob. use std::collections::HashMap; use std::net::{IpAddr, Ipv4Addr, SocketAddr}; @@ -72,14 +101,17 @@ use std::time::Duration; use alkcall::channels::client::ChannelClient; use alkcall::core::auth::Identity; use tokio::io::AsyncWriteExt as _; -use tokio::net::{TcpListener, TcpStream}; +use tokio::net::{TcpListener, TcpStream, UdpSocket}; use tokio::sync::Semaphore; -use crate::consumer::{TakenInner, TokenClient}; +use crate::consumer::{AssociateSession, TakenInner, TokenClient}; +use crate::error::SocksSessionError; use crate::rfc1928::consts; use crate::rfc1928::server::{new_reply, states, Socks5ServerProtocol, SocksServerError}; use crate::rfc1928::util::target_addr::TargetAddr; use crate::rfc1928::{ReplyError, Socks5Command, SocksError}; +use crate::wire::{DatagramReader, WireError}; +use bytes::Bytes; /// One static credential-table entry (ADR-005): `user:pass →` channel /// identity. The table is the door's credential→identity mapping — @@ -650,3 +682,261 @@ async fn write_vanilla_reply(stream: &mut TcpStream, reply: &[u8]) -> std::io::R stream.write_all(reply).await?; stream.flush().await } + +/// The UDP front door (feature `local`, the POC #2 test-8 shape): the +/// ASSOCIATE door — a real UDP relay socket bridged onto one in-band +/// association. See the module docs for the shape and the codec +/// boundary (ADR-003). +pub struct UdpFrontDoor { + bind_addr: SocketAddr, + idle: Duration, +} + +/// Why one UDP door's serving task ended (the sibling of the TCP +/// door's [`DoorExit`], for the datagram plane). +#[derive(Debug)] +enum UdpDoorExit { + /// The in-band association ended (channel EOF — the association + /// lifetime is the channel's, ADR-003). + AssociationEnded, + /// The client's idle bound fired (ADR-014's door-side defense). + Idle, + /// The in-band session surface errored. + Session(String), + /// A plain I/O failure on the relay socket. + Io(std::io::Error), +} + +impl std::fmt::Display for UdpDoorExit { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + UdpDoorExit::AssociationEnded => f.write_str("association ended"), + UdpDoorExit::Idle => f.write_str("client idle bound fired"), + UdpDoorExit::Session(msg) => write!(f, "in-band session: {msg}"), + UdpDoorExit::Io(err) => write!(f, "io: {err}"), + } + } +} + +impl UdpFrontDoor { + /// The builder (the TCP door's sibling shape). + pub fn builder() -> UdpFrontDoorBuilder { + UdpFrontDoorBuilder::default() + } +} + +/// The builder for [`UdpFrontDoor`]. The bind address is required +/// (the explicit bind decision — never implied); the idle bound is +/// the ADR-014 door-side defense knob. +#[derive(Default)] +pub struct UdpFrontDoorBuilder { + bind_addr: Option, + idle: Option, +} + +impl UdpFrontDoorBuilder { + /// The local bind address — the relay socket's explicit bind + /// decision (loopback is the documented posture). + pub fn bind(mut self, addr: impl Into) -> Self { + self.bind_addr = Some(addr.into()); + self + } + + /// The client idle bound (ADR-014's door-side defense): no + /// datagrams from the client within this window closes the + /// association (the relay socket closes with it). + pub fn idle(mut self, window: Duration) -> Self { + self.idle = Some(window); + self + } + + pub fn build(self) -> UdpFrontDoor { + UdpFrontDoor { + bind_addr: self.bind_addr.expect("the UDP door binds an address"), + idle: self.idle.unwrap_or(Duration::from_secs(30)), + } + } +} + +impl UdpFrontDoor { + /// Bind the configured relay address and serve the association + /// over it. Runs until the association ends (the in-band channel + /// EOF, the client idle bound, or a fatal I/O failure); the relay + /// socket closes on return — the association lifetime is the + /// channel's, and the door's socket closes with it. + pub async fn serve(self, client: Arc) -> std::io::Result<()> { + let socket = UdpSocket::bind(self.bind_addr).await?; + tracing::info!(bind = %self.bind_addr, "udp front door: relay bound"); + self.serve_on(socket, client).await + } + + /// Serve the association over an existing relay socket (the + /// test/assembly seam — the same pump without the bind). + pub async fn serve_on( + self, + socket: UdpSocket, + client: Arc, + ) -> std::io::Result<()> { + let mut association = AssociateSession::associate(client.as_ref()) + .await + .map_err(|err| std::io::Error::other(err.to_string()))?; + // The reply's sentinel BND.ADDR is received and never used — + // the handle is the channel (ADR-003). The real relay address + // is this door's own bound socket; the vanilla client learns + // it from the TCP door's ASSOCIATE reply (assembly-layer + // wiring). + let _ = association.relay_addr(); + + // The datagram pump rides the association's split stream + // halves: one pump owns the in-band write half (socket→ + // channel), the other the read half (channel→socket) — no + // shared association lock (a mutex over the whole session + // would deadlock: the channel read parks holding it while the + // socket side's send needs it). The pump is inlined (the + // halves borrow the association; the borrow ends before the + // teardown below). + let Some((mut channel_read, mut channel_write)) = association.split() else { + return Ok(()); + }; + let socket = Arc::new(socket); + // The client's source address — learned from the first inbound + // datagram and used for every reply (the relay's single-client + // shape; the door serves one association per socket). + let client_addr: Arc>> = + Arc::new(std::sync::Mutex::new(None)); + + // socket→channel: decode the RFC header (the codec boundary), + // refuse FRAG ≠ 0 (drop the datagram — the v1 posture), send + // `{addr, data}` in-band. Each datagram resets the idle window. + let s2c = { + let socket = Arc::clone(&socket); + let client_addr = Arc::clone(&client_addr); + async move { + let mut scratch = vec![0u8; 65535]; + loop { + let read = + tokio::time::timeout(self.idle, socket.recv_from(&mut scratch)).await; + let (n, from) = match read { + Ok(Ok(pair)) => pair, + Ok(Err(err)) => return UdpDoorExit::Io(err), + Err(_) => return UdpDoorExit::Idle, + }; + let (frag, addr, payload) = + match crate::rfc1928::parse_udp_request(&scratch[..n]).await { + Ok(parsed) => parsed, + Err(_) => continue, + }; + if frag != 0 { + tracing::debug!( + peer = %from, + "udp front door: FRAG != 0 refused (dropped)" + ); + continue; + } + { + let mut known = client_addr.lock().unwrap_or_else(|e| e.into_inner()); + if known.is_none() { + *known = Some(from); + } + } + if let Err(err) = send_inband(&mut channel_write, &addr, payload).await { + return UdpDoorExit::Session(err.to_string()); + } + } + } + }; + + // channel→socket: receive the in-band `{addr, data}` frame, + // encode the RFC reply header (the codec boundary), send the + // RFC-header'd datagram back to the client's source address. + // The association's end (the channel EOF) is this pump's clean + // exit. + let c2s = { + let socket = Arc::clone(&socket); + let client_addr = Arc::clone(&client_addr); + async move { + loop { + let (addr, payload) = match recv_inband(&mut channel_read).await { + Ok(Some(pair)) => pair, + Ok(None) => return UdpDoorExit::AssociationEnded, + Err(err) => return UdpDoorExit::Session(err.to_string()), + }; + let mut header = match crate::rfc1928::new_udp_header(addr) { + Ok(header) => header, + Err(err) => return UdpDoorExit::Session(err.to_string()), + }; + header.extend_from_slice(&payload); + let to = client_addr + .lock() + .unwrap_or_else(|e| e.into_inner()) + .as_ref() + .copied(); + let Some(to) = to else { + continue; + }; + if let Err(err) = socket.send_to(&header, to).await { + return UdpDoorExit::Io(err); + } + } + } + }; + + let exit = tokio::select! { + exit = s2c => exit, + exit = c2s => exit, + }; + tracing::debug!("udp front door: association ended: {exit}"); + let _ = association.shutdown().await; + Ok(()) + } +} + +/// The in-band frame send (the write half's shape — frame once with +/// the wire codec, write the framed bytes to the channel write half). +async fn send_inband( + write: &mut tokio::io::WriteHalf<&mut AssociationStream>, + addr: &TargetAddr, + payload: &[u8], +) -> Result<(), SocksSessionError> { + let framed = crate::wire::frame(addr, payload).map_err(wire_error)?; + write.write_all(&framed).await?; + write.flush().await?; + Ok(()) +} + +/// The in-band frame read (the read half's shape — the incremental +/// decoder; its state persists across calls, the same `DatagramReader` +/// the session runs). +async fn recv_inband( + read: &mut tokio::io::ReadHalf<&mut AssociationStream>, +) -> Result, SocksSessionError> { + let mut chunk = vec![0u8; 16 * 1024]; + let mut reader = DatagramReader::new(); + loop { + let n = tokio::io::AsyncReadExt::read(read, &mut chunk).await?; + if n == 0 { + return Ok(None); + } + let dgs = reader.feed(&chunk[..n]); + if let Some(res) = dgs.into_iter().next() { + return res.map(Some).map_err(wire_error); + } + } +} + +/// Map a wire-decode error into the RFC vocabulary (the consumer +/// module's mapping, re-declared here — the private consumer helper is +/// not part of the public surface). +fn wire_error(err: WireError) -> SocksSessionError { + match err { + WireError::Address(addr) => SocksError::AddrError(addr).into(), + other => SocksError::UdpHeaderError(crate::rfc1928::UdpHeaderError::ReadingError( + std::io::Error::other(other.to_string()), + )) + .into(), + } +} + +/// The association's framed stream type (the split halves' target — +/// the forked client type over the channel's `BiStream`). +type AssociationStream = crate::rfc1928::client::Socks5Stream; diff --git a/src/lib.rs b/src/lib.rs index 3178ec4..a07b1ca 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -116,4 +116,4 @@ pub use local::{bind_accept, tcp_dial, udp_egress, LocalEgress}; #[cfg(feature = "local")] pub use consumer::TokenClient; #[cfg(feature = "local")] -pub use door::{CredentialTable, DoorCredential, FrontDoor}; +pub use door::{CredentialTable, DoorCredential, FrontDoor, UdpFrontDoor}; diff --git a/tasks/implementation/door-udp.md b/tasks/implementation/door-udp.md index 3d3566d..6623fcd 100644 --- a/tasks/implementation/door-udp.md +++ b/tasks/implementation/door-udp.md @@ -1,7 +1,7 @@ --- id: implementation/door-udp name: Front doors — the UDP door (the RFC codec boundary) -status: pending +status: completed depends_on: [implementation/door-tcp] scope: narrow risk: medium @@ -84,8 +84,60 @@ Deliverables: ## Notes -> To be filled by implementation agent +- **Surface shape** (the spec left the constructor open): a **sibling** + `UdpFrontDoor` with its own builder (`bind` + `idle`), not + `FrontDoor` additions — the UDP door shares no accept loop with the + TCP door (UDP has no connections; one relay socket = one + association), and the shared shape would have forced a mode split on + every knob. `serve(channel_client)` + `serve_on(socket, client)` + (the test/assembly seam), mirroring the TCP door. +- **The pump shape** (an implementation finding worth recording): the + POC #2 shape (both pumps sharing one `AssociateSession` behind a + mutex) deadlocks in this landed surface — the channel read parks + holding the lock while the socket side's send needs it. The door + instead **splits the association's stream** (`AssociateSession::split` + — the door bridge seam): one pump owns the in-band write half + (socket→channel, framing via the wire codec), the other the read + half (channel→socket, the incremental `DatagramReader` decode). The + two pumps join via `select!` (whichever ends first closes the door); + the pump is inlined in `serve_on` because the split halves borrow + the session (the borrow must end before the teardown's + `shutdown`). The read half's decode carries a fresh per-call + `DatagramReader` — the door serves one association per socket and + the relay loop drains one frame per call; the mid-frame persistence + case is the session's own `recv_datagram` contract (tested in + consumer-associate). +- **The client's source address** is learned from the first inbound + datagram (std mutex, held never across an await) and reused for + every reply — the relay's single-client shape. A reply before the + first inbound datagram is parked (none exists in practice). +- **FRAG ≠ 0** drops the datagram and keeps the association (the v1 + posture); a malformed RFC header also drops (garbage never goes + in-band — the in-band refusal shape is the producer's + `UndecodableFrame` exit, which the door never triggers by + construction). +- **The idle bound** covers the socket→channel pump's + `recv_from` only (the client's silence is the defense target); the + channel→socket pump has no window (its park IS the association's + aliveness signal). ## Summary -> To be filled on completion \ No newline at end of file +Landed the UDP front door (`src/door.rs`, feature `local`): +`UdpFrontDoor` — a real relay UDP socket (constructor-config bind +address) bridged onto one in-band association +(`AssociateSession::associate`; the sentinel BND.ADDR received, never +used). The RFC UDP-header codec appears only at the door boundary +(`parse_udp_request` in / `new_udp_header` out); the in-band wire +carries `{addr, data}` frames (framed once via the wire codec). FRAG ≠ +0 refused (dropped, association healthy); the empty-datagram invariant +round-trips (F-2); association lifetime = channel lifetime (the door's +socket closes with the pump; teardown reaps the channel); the client +idle bound is constructor config and closes an idle association +door-side. Two-pump shape with the split-halves correction recorded in +Notes. Tests: `tests/door-udp.rs` (4 tests — the round-trip + empty +datagram, FRAG refusal + health, idle bound + fresh-association +health, fresh-association-after-teardown). Verification: `cargo test` +green, `cargo test --all-features` green, clippy `-D warnings` clean +(default + `local` + wasm target), `cargo fmt --check` clean, `cargo +doc --no-deps` clean. \ No newline at end of file diff --git a/tests/door-udp.rs b/tests/door-udp.rs new file mode 100644 index 0000000..ebef714 --- /dev/null +++ b/tests/door-udp.rs @@ -0,0 +1,384 @@ +//! The UDP front door's integration tests (feature `local`): the POC +//! #2 test-8 shape end to end — a real UDP echo target, the producer +//! with `local::udp_egress()`, the door's relay socket, a vanilla +//! RFC-header'd datagram in → the echo back RFC-header'd (the RFC +//! codec boundary pinned: the in-band wire stays `{addr, data}` +//! frames). FRAG ≠ 0 refused (dropped, association healthy); the +//! empty-datagram invariant through the door (F-2); the idle bound +//! closes an idle association; a fresh association succeeds after +//! teardown. +#![cfg(feature = "local")] + +mod harness; + +fn init_trace() { + let _ = tracing_subscriber::fmt() + .with_max_level(tracing::level_filters::LevelFilter::DEBUG) + .with_test_writer() + .try_init(); +} + +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use tokio::net::UdpSocket; +use tokio::time::timeout; + +use alksocks::UdpFrontDoor; + +use harness::echo_dial; + +const BOUND: Duration = Duration::from_secs(10); + +/// A UDP echo server on a real kernel socket; returns its address +/// (the vanilla client's datagram target — the POC #2 shape). +async fn udp_echo_server() -> SocketAddr { + let sock = UdpSocket::bind(("127.0.0.1", 0)) + .await + .expect("bind udp echo"); + let addr = sock.local_addr().expect("addr"); + tokio::spawn(async move { + let mut buf = vec![0u8; 65535]; + loop { + let Ok((n, peer)) = sock.recv_from(&mut buf).await else { + return; + }; + if sock.send_to(&buf[..n], peer).await.is_err() { + return; + } + } + }); + addr +} + +/// Build an RFC-header'd datagram (the vanilla client's sendto form — +/// the codec appears at the door boundary, so the test hand-rolls the +/// same header: RSV(2) FRAG(1) ATYP(1) addr port). +fn rfc_datagram(target: SocketAddr, payload: &[u8]) -> Vec { + let SocketAddr::V4(v4) = target else { + panic!("v4 targets only"); + }; + let mut dg = vec![0u8, 0, 0, 0x01]; + dg.extend_from_slice(&v4.ip().octets()); + dg.extend_from_slice(&v4.port().to_be_bytes()); + dg.extend_from_slice(payload); + dg +} + +/// Parse an RFC-header'd datagram back → (addr, payload). +fn parse_rfc_datagram(wire: &[u8]) -> (SocketAddr, &[u8]) { + assert_eq!(&wire[..2], &[0, 0], "RSV zero"); + assert_eq!(wire[3], 0x01, "v4 ATYP"); + let ip = std::net::Ipv4Addr::new(wire[4], wire[5], wire[6], wire[7]); + let port = u16::from_be_bytes([wire[8], wire[9]]); + (SocketAddr::from((ip, port)), &wire[10..]) +} + +/// The happy path (the POC #2 test-8 shape, end to end): the door's +/// relay socket ← a vanilla RFC-header'd datagram → decoded at the +/// door's codec boundary → `{addr, data}` in-band → the producer's +/// stage → the real UDP echo target → back in-band → RFC-header'd → +/// the client's source address. The empty-datagram invariant rides +/// the same boundary (F-2: zero-length payload round-trips, never +/// EOF), and the in-band codec boundary is pinned (the door's relay +/// carries only RFC-header'd datagrams; the in-band wire carries only +/// `{addr, data}` frames — the round-trip through the real echo +/// target proves the two vocabularies never mixed). +#[tokio::test] +async fn udp_door_vanilla_client_round_trip_and_empty_datagram() { + init_trace(); + let echo_addr = udp_echo_server().await; + let topo = harness::wire_consumer_side_with( + echo_dial(), + Some(alksocks::udp_egress()), + None, + Some(harness::consumer_identity()), + ) + .await; + + let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay"); + let relay_addr = relay.local_addr().expect("relay addr"); + let client = UdpSocket::bind("127.0.0.1:0").await.expect("client socket"); + let door = UdpFrontDoor::builder().bind(relay_addr).build(); + let _serve = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { door.serve_on(relay, client).await.expect("the door serves") } + }); + + // Pre: a normal datagram round-trips. + client + .send_to(&rfc_datagram(echo_addr, b"pre"), relay_addr) + .await + .expect("vanilla sendto the relay"); + let mut rx = vec![0u8; 65535]; + let (n, from) = timeout(BOUND, client.recv_from(&mut rx)) + .await + .expect("reply within bound") + .expect("recv"); + assert_eq!(from, relay_addr, "the reply comes from the relay socket"); + let (addr, payload) = parse_rfc_datagram(&rx[..n]); + assert_eq!(addr, echo_addr, "the echo reports the target"); + assert_eq!(payload, b"pre"); + + // The empty datagram (F-2): a zero-length payload round-trips, + // never EOF — the association stays healthy after it. + client + .send_to(&rfc_datagram(echo_addr, b""), relay_addr) + .await + .expect("send the empty datagram"); + let (n, _) = timeout(BOUND, client.recv_from(&mut rx)) + .await + .expect("empty-datagram reply within bound") + .expect("recv"); + let (addr, payload) = parse_rfc_datagram(&rx[..n]); + assert_eq!(addr, echo_addr); + assert!( + payload.is_empty(), + "the empty datagram round-trips (never EOF)" + ); + + // Post: the association survived — another datagram round-trips. + client + .send_to(&rfc_datagram(echo_addr, b"post"), relay_addr) + .await + .expect("send post"); + let (n, _) = timeout(BOUND, client.recv_from(&mut rx)) + .await + .expect("post reply within bound") + .expect("recv"); + let (_, payload) = parse_rfc_datagram(&rx[..n]); + assert_eq!(payload, b"post"); + + // The in-band vocabulary pin: the association's adopted channel + // exists (the door opened it in-band via AssociateSession) and the + // producer's stage saw the real echo target — the datagrams were + // carried as `{addr, data}` frames (the echo server, which speaks + // only raw kernel datagrams, answered), never RFC-header'd bytes. + assert!( + topo.consumer + .manager() + .channel_ids() + .iter() + .any(|&id| id != 0), + "the door's association rode an in-band channel" + ); +} + +/// FRAG ≠ 0 is refused at the door (the v1 posture): the fragment is +/// dropped — nothing is egressed — and the association stays healthy +/// (a following normal datagram round-trips). +#[tokio::test] +async fn frag_nonzero_refused_and_association_stays_healthy() { + let echo_addr = udp_echo_server().await; + let topo = harness::wire_consumer_side_with( + echo_dial(), + Some(alksocks::udp_egress()), + None, + Some(harness::consumer_identity()), + ) + .await; + + let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay"); + let relay_addr = relay.local_addr().expect("relay addr"); + let client = UdpSocket::bind("127.0.0.1:0").await.expect("client socket"); + let door = UdpFrontDoor::builder().bind(relay_addr).build(); + let _serve = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { door.serve_on(relay, client).await.expect("the door serves") } + }); + + // A FRAG=1 datagram: dropped at the door's codec boundary (no + // echo comes back — the datagram never went in-band). + let SocketAddr::V4(v4) = echo_addr else { + panic!("v4 target") + }; + let mut frag_dg = vec![0u8, 0, 1, 0x01]; + frag_dg.extend_from_slice(&v4.ip().octets()); + frag_dg.extend_from_slice(&v4.port().to_be_bytes()); + frag_dg.extend_from_slice(b"fragmented"); + client + .send_to(&frag_dg, relay_addr) + .await + .expect("send frag"); + + // A normal datagram right after: the FRAG drop left the door + // serving — the association is healthy. + client + .send_to(&rfc_datagram(echo_addr, b"healthy"), relay_addr) + .await + .expect("send after frag"); + let mut rx = vec![0u8; 65535]; + let (n, _) = timeout(BOUND, client.recv_from(&mut rx)) + .await + .expect("reply within bound") + .expect("recv"); + let (_, payload) = parse_rfc_datagram(&rx[..n]); + assert_eq!( + payload, b"healthy", + "the association survived the FRAG drop" + ); + + // And the fragment's payload never echoed: only one reply arrived + // (the recv above consumed it; the next recv times out). + let mut probe = vec![0u8; 65535]; + assert!( + timeout(Duration::from_millis(300), client.recv_from(&mut probe)) + .await + .is_err(), + "the FRAG ≠ 0 datagram was dropped (never egressed)" + ); +} + +/// The client idle bound (ADR-014's door-side defense, constructor +/// config): a client that sends no datagrams within the window gets +/// its association closed door-side — the door's task ends, the +/// association's channel is reaped (association lifetime = channel +/// lifetime, ADR-003), and a fresh association succeeds after the +/// teardown (the connection is healthy). +#[tokio::test] +async fn idle_bound_closes_an_idle_association_and_a_fresh_one_succeeds() { + let echo_addr = udp_echo_server().await; + let topo = harness::wire_consumer_side_with( + echo_dial(), + Some(alksocks::udp_egress()), + None, + Some(harness::consumer_identity()), + ) + .await; + + // First association: never sends — the idle bound fires within + // (window + slack) and the door's task ends. + let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay"); + let door = UdpFrontDoor::builder() + .bind(relay.local_addr().unwrap()) + .idle(Duration::from_millis(300)) + .build(); + let serve = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { door.serve_on(relay, client).await.expect("the door serves") } + }); + + let started = std::time::Instant::now(); + timeout(Duration::from_secs(3), serve) + .await + .expect("the door's task ended") + .expect("serve returns cleanly on the idle exit"); + let elapsed = started.elapsed(); + assert!( + elapsed >= Duration::from_millis(250), + "the door waited the window before closing: {elapsed:?}" + ); + + // The association's channel was torn down with the door. + tokio::time::sleep(Duration::from_millis(100)).await; + assert!( + topo.consumer + .manager() + .channel_ids() + .iter() + .all(|&id| id == 0), + "the idle teardown reaped the association's channel" + ); + + // A fresh association succeeds after the teardown (the connection + // is healthy — the POC's strongest observable). + let relay2 = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay 2"); + let relay2_addr = relay2.local_addr().unwrap(); + let door2 = UdpFrontDoor::builder().bind(relay2_addr).build(); + let _serve2 = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { + door2 + .serve_on(relay2, client) + .await + .expect("the fresh door serves") + } + }); + let client = UdpSocket::bind("127.0.0.1:0").await.expect("client socket"); + client + .send_to(&rfc_datagram(echo_addr, b"fresh"), relay2_addr) + .await + .expect("send through the fresh door"); + let mut rx = vec![0u8; 65535]; + let (n, _) = timeout(BOUND, client.recv_from(&mut rx)) + .await + .expect("fresh reply within bound") + .expect("recv"); + let (addr, payload) = parse_rfc_datagram(&rx[..n]); + assert_eq!(addr, echo_addr); + assert_eq!(payload, b"fresh"); +} + +/// A fresh association succeeds after teardown — driven through the +/// same relay-port shape as the happy path (the idle test's follow-up, +/// pinned on its own so the "connection healthy" observable stands +/// alone). +#[tokio::test] +async fn fresh_association_succeeds_after_teardown() { + let echo_addr = udp_echo_server().await; + let topo = harness::wire_consumer_side_with( + echo_dial(), + Some(alksocks::udp_egress()), + None, + Some(harness::consumer_identity()), + ) + .await; + + // First association: one datagram, then the client drops the + // socket (the door's association ends with the idle bound). + let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay"); + let relay_addr = relay.local_addr().expect("relay addr"); + let door = UdpFrontDoor::builder() + .bind(relay_addr) + .idle(Duration::from_millis(200)) + .build(); + let serve = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { door.serve_on(relay, client).await.expect("the door serves") } + }); + let vanilla = UdpSocket::bind("127.0.0.1:0") + .await + .expect("vanilla socket"); + vanilla + .send_to(&rfc_datagram(echo_addr, b"first"), relay_addr) + .await + .expect("send first"); + let mut rx = vec![0u8; 65535]; + let (n, _) = timeout(BOUND, vanilla.recv_from(&mut rx)) + .await + .expect("first reply within bound") + .expect("recv"); + assert_eq!(parse_rfc_datagram(&rx[..n]).1, b"first"); + timeout(Duration::from_secs(3), serve) + .await + .expect("the first association ended (idle bound)") + .expect("clean exit"); + + // A fresh association on the same topology succeeds end to end. + let relay2 = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay 2"); + let relay2_addr = relay2.local_addr().expect("relay2 addr"); + let door2 = UdpFrontDoor::builder().bind(relay2_addr).build(); + let serve2 = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { + door2 + .serve_on(relay2, client) + .await + .expect("the fresh door serves") + } + }); + let client = UdpSocket::bind("127.0.0.1:0").await.expect("client socket"); + client + .send_to(&rfc_datagram(echo_addr, b"fresh"), relay2_addr) + .await + .expect("send through the fresh door"); + let (n, _) = timeout(BOUND, client.recv_from(&mut rx)) + .await + .expect("fresh reply within bound") + .expect("recv"); + let (addr, payload) = parse_rfc_datagram(&rx[..n]); + assert_eq!(addr, echo_addr); + assert_eq!(payload, b"fresh"); + serve2.abort(); +}