door: the UDP front door — the RFC codec boundary (ADR-003)

- 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

@@ -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<SocketAddr>,
+    idle: Option<Duration>,
+}
+
+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<SocketAddr>) -> 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<ChannelClient>) -> 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<ChannelClient>,
+    ) -> 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<std::sync::Mutex<Option<SocketAddr>>> =
+            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<Option<(TargetAddr, Bytes)>, 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<alkcall::core::types::BiStream>;

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};