diff --git a/src/lib.rs b/src/lib.rs index c6759bd..3bb6d59 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -45,4 +45,4 @@ pub use producer::{ make_socks5_pump_handler, register_socks5_openable, serve_socks5_over, socks5_session_establisher, IdentityWitness, Socks5Producer, Socks5ProducerBuilder, }; -pub use wire::{addr_encoding_len, frame, DatagramReader, WireError, MAX_INNER_LEN}; +pub use wire::{addr_encoding_len, frame, frame_inner, DatagramReader, WireError, MAX_INNER_LEN}; diff --git a/src/producer.rs b/src/producer.rs index be3044f..00ecb65 100644 --- a/src/producer.rs +++ b/src/producer.rs @@ -40,9 +40,16 @@ //! birth). Refusal: `reply_error` with the mapped RFC code //! (dial-failure → `ConnectionRefused` 0x05; scope-class → //! `ConnectionNotAllowed` 0x02) — no phantom data plane. -//! - **UDP ASSOCIATE** → RFC-faithful refusal in this task's scope -//! (`CommandNotSupported` — ADR-008's missing-callback class); the -//! producer-associate task replaces it. +//! - **UDP ASSOCIATE** → the datagram stage (ADR-003): the sentinel +//! reply (`127.0.0.1:0` — the relay handle IS the channel; the +//! address is pure RFC-compatibility surface), then the +//! `select!`-driven stage (wire.md §UDP ASSOCIATE): channel reads ∥ +//! egress replies, framing applied exactly once by the wire layer, +//! the egress readiness contract (`Pending` when nothing is ready), +//! association lifetime = channel lifetime. Missing egress seam: +//! RFC-faithful refusal (`CommandNotSupported` — ADR-008's +//! missing-callback class). Egress-factory failure: reply#1 with +//! the mapped code. //! - **TCP BIND** → the command gate first (`socks5:bind`, ADR-006, //! from the per-call opener identity — a scope-less identity gets //! `ConnectionNotAllowed` 0x02 while CONNECT still serves it), then @@ -97,11 +104,14 @@ use alkcall::registry::registration::OperationRegistry; use serde_json::Value; use tokio::io::{AsyncRead, AsyncWrite}; -use crate::egress::{BindAccept, BindAcceptFactory, BindTargetHandle, DialFn, EgressFactory}; +use crate::egress::{ + BindAccept, BindAcceptFactory, BindTargetHandle, DialFn, Egress, EgressFactory, +}; use crate::params::{socks5_open_spec, SOCKS5_BIND_SCOPE}; use crate::rfc1928::server::{new_reply, Socks5ServerProtocol, SocksServerError}; use crate::rfc1928::util::target_addr::TargetAddr; use crate::rfc1928::{ReplyError, Socks5Command, SocksError}; +use crate::wire::{frame_inner, DatagramReader}; pub use crate::params::{OP_SOCKS5_OPEN, SOCKS5_ALPN, SOCKS5_OPEN_SCOPE}; @@ -260,9 +270,16 @@ where Ok(()) } -/// UDP ASSOCIATE: RFC-faithful refusal in this task's scope — the -/// datagram stage lands with the producer-associate task (the -/// pre-vertical posture; ADR-008's missing-callback class). +/// UDP ASSOCIATE: the datagram stage (ADR-003, POC #2-validated). +/// The typestate is driven directly (no `run_udp_proxy_custom` — its +/// unconditional kernel bind is unusable on channels); the reply +/// carries the sentinel address (the relay handle IS the channel; +/// the address is pure RFC-compatibility surface for wrapper-aware +/// clients, who never sendto() it); the post-reply stream is the +/// framed `{addr, data}` datagram stage. A missing egress seam +/// refuses RFC-faithfully (ADR-008's missing-callback class — an +/// unprovided capability, never a panic); an egress-factory failure +/// refuses with the mapped code. async fn serve_udp_associate( proto: Socks5ServerProtocol, egress: Option, @@ -270,11 +287,146 @@ async fn serve_udp_associate( where T: AsyncRead + AsyncWrite + Unpin, { - let _ = egress; - proto.reply_error(&ReplyError::CommandNotSupported).await?; - Err(SocksError::ServerError(SocksServerError::UnknownCommand( - Socks5Command::UDPAssociate.as_u8(), - ))) + let Some(egress) = egress else { + proto.reply_error(&ReplyError::CommandNotSupported).await?; + return Err(SocksError::ServerError(SocksServerError::UnknownCommand( + Socks5Command::UDPAssociate.as_u8(), + ))); + }; + + let mut egress = match egress().await { + Ok(egress) => egress, + Err(err) => { + proto.reply_error(&err.to_reply_error()).await?; + return Err(SocksError::ServerError(SocksServerError::Bug( + "egress refused in-stream", + ))); + } + }; + + let inner = proto.reply_success(sentinel_reply_addr()).await?; + + run_datagram_stage(inner, egress.as_mut()).await; + Ok(()) +} + +/// The post-reply datagram stage (wire.md §UDP ASSOCIATE; the POC #2 +/// shape). Three POC invariants, load-bearing: +/// +/// - **The concurrent select shape** — channel reads and the egress's +/// reply stream run concurrently (the honest full-duplex shape; a +/// poll-then-block loop loses replies arriving while the channel +/// read is parked — the POC's first iteration, measured). +/// - **The egress readiness contract** (`Egress::next_reply`): it +/// must be `Pending` when no reply is ready; an instantly-ready +/// `Ok(None)` starves a biased select into livelock. The stage +/// tolerates a spurious `Ok(None)` by re-parking (never a hot +/// loop); an `Err` ends the association (the egress drops with it). +/// - **Framing applied exactly once, by the wire layer** — the stage +/// reads frames via `DatagramReader` (channel → egress `send` gets +/// the payload) and writes frames via `frame_inner` (egress +/// `next_reply`'s inner bytes → the channel). The egress never sees +/// a length prefix. +/// +/// Association lifetime = channel lifetime: channel EOF ends the +/// stage; no extra control vocabulary. A undecodable frame (garbage +/// ATYP — any post-reply control byte) fails the association loud +/// (the RFC's "post-reply control data is garbage" semantic). A +/// stream end mid-frame is fail-loud too (never a silent partial +/// datagram — the reader state is the teardown diagnostic). The +/// stage owns the egress: on every exit path the egress binding +/// drops here (teardown-at-birth — no `_` bindings of owning +/// values). +/// +/// Per-datagram addressing in the frame does the demultiplexing (no +/// flow table); one relay loop, no spawn-per-datagram. A frame's +/// egress `send` is awaited inline before the next select iteration — +/// the stage interleaves egress sends with reply drains (the POC's +/// measured shape; sufficient for relay postures). +async fn run_datagram_stage(mut inner: T, egress: &mut (dyn Egress + Send)) -> StageExit +where + T: AsyncRead + AsyncWrite + Unpin, +{ + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let mut reader = DatagramReader::new(); + let mut chunk = vec![0u8; 16 * 1024]; + loop { + tokio::select! { + // The egress's reply stream (its own pump; `None` = a + // spurious wakeup, the stage re-parks — never a hot loop). + reply = egress.next_reply() => { + match reply { + Ok(Some(frame)) => { + match frame_inner(&frame) { + Ok(wire) => { + if inner.write_all(&wire).await.is_err() { + return StageExit::Io; + } + } + Err(err) => { + tracing::debug!("udp stage: reply frame rejected: {err}"); + return StageExit::ReplyFrameRejected; + } + } + } + Ok(None) => {} + Err(err) => { + tracing::debug!("udp egress relay ended: {err:#}"); + return StageExit::EgressEnded; + } + } + } + // The channel's read side: frames in, EOF = end. + read = inner.read(&mut chunk) => { + match read { + // Channel EOF = association end (the clean exit — + // the association lifetime is the channel's). + Ok(0) => return StageExit::ChannelEof, + Ok(n) => { + for dg in reader.feed(&chunk[..n]) { + match dg { + Err(err) => { + tracing::debug!("udp stage: undecodable frame: {err}"); + return StageExit::UndecodableFrame; + } + Ok((addr, payload)) => { + if let Err(err) = egress.send(addr, payload).await { + tracing::debug!("udp egress send failed: {err:#}"); + return StageExit::EgressEnded; + } + } + } + } + } + Err(err) => { + tracing::debug!("udp stage: channel read error: {err}"); + return StageExit::Io; + } + } + } + } + } +} + +/// Why the datagram stage ended (the stage's teardown diagnostic — +/// each exit drops the egress with the stage; the channel EOF is the +/// one clean exit, per the association-lifetime contract). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum StageExit { + /// Channel EOF — the association's clean end (the one the + /// association-lifetime contract names). + ChannelEof, + /// A frame the codec refused (garbage ATYP / a `len == 0` frame — + /// post-reply control bytes). Loud fail: never a silent drop. + UndecodableFrame, + /// An egress `next_reply` `Err` or a `send` failure (the egress + /// relay ended; the lossy-datagram posture). + EgressEnded, + /// A reply the wire codec rejected (the egress violated the + /// joint bound — never a wire overflow). + ReplyFrameRejected, + /// The channel stream itself failed (I/O). + Io, } /// TCP BIND: the phase model (ADR-007; POC #5-validated). Phases: @@ -546,6 +698,7 @@ impl Socks5ProducerBuilder { mod tests { use super::*; use crate::egress::{BindAcceptError, BindAcceptFactoryError, SocksDialError, TargetHandle}; + use bytes::Bytes; use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream}; /// The test dial closure (the harness's echo shape, inline): one @@ -779,29 +932,616 @@ mod tests { )); } - /// The ASSOCIATE dispatch arm refuses RFC-faithfully in this - /// task's scope (`CommandNotSupported` — the pre-vertical - /// posture; the producer-associate task replaces it). BIND now - /// serves under the gate — its tests live below. - #[tokio::test] - async fn associate_arm_refuses_command_not_supported() { + // -- UDP ASSOCIATE: the datagram stage (ADR-003, POC #2 shape, + // in-process over duplexes) ------------------------------------ + + /// The test egress (the harness's ChanEgress shape, inline): one + /// unbounded queue; `send` frames the reply inner bytes for echo + /// (`addr + data` — the egress speaks inner bytes, no length + /// prefix); `next_reply` parks on the queue (`Pending` when + /// nothing is ready — the readiness contract). + struct QueueEgress { + inbound: tokio::sync::mpsc::UnboundedSender, + outbound: tokio::sync::Mutex>, + } + + impl crate::egress::Egress for QueueEgress { + fn send( + &mut self, + addr: TargetAddr, + payload: Bytes, + ) -> futures::future::BoxFuture<'_, Result<(), crate::egress::EgressError>> { + Box::pin(async move { + let mut inner = addr.to_be_bytes().map_err(|e| { + crate::egress::EgressError::General(format!("addr encode: {e}")) + })?; + inner.extend_from_slice(&payload); + self.inbound + .send(Bytes::from(inner)) + .map_err(|_| crate::egress::EgressError::General("egress closed".into())) + }) + } + + fn next_reply( + &mut self, + ) -> futures::future::BoxFuture<'_, Result, crate::egress::EgressError>> + { + Box::pin(async move { + match self.outbound.lock().await.recv().await { + Some(inner) => Ok(Some(inner)), + None => Err(crate::egress::EgressError::General("egress closed".into())), + } + }) + } + } + + use crate::egress::EgressError; + + /// The frame a raw client sends on the datagram stage: + /// `[len: u16 BE][addr + data]`. + fn datagram_frame(addr: &TargetAddr, payload: &[u8]) -> Vec { + crate::wire::frame(addr, payload).unwrap().to_vec() + } + + /// Split the wire into (addr, payload) — the raw client's reply + /// decode (one frame per read batch, in order). + fn decode_datagram(wire: &[u8]) -> (TargetAddr, Bytes) { + let mut reader = crate::wire::DatagramReader::new(); + let mut got = None; + for dg in reader.feed(wire) { + got = Some(dg); + } + got.expect("one datagram decoded").expect("frame decodes") + } + + /// The full ASSOCIATE conversation over a duplex: greet → + /// associate request → sentinel reply. Returns (server handle, + /// client duplex). `egress_factory_err` simulates a refusing + /// factory; `None` wires the echo egress. + async fn raw_associate( + egress: Option, + ) -> ( + tokio::task::JoinHandle>, + tokio::io::DuplexStream, + ) { + let (client_end, server_end) = tokio::io::duplex(64 * 1024); let dial = echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new()))); - let (handle, mut client) = raw_connect(dial).await; + let handle = tokio::spawn(serve_socks5_over( + server_end, + Some(dial), + egress, + None, + false, + )); + let mut client = client_end; greet(&mut client).await; - let req_target = TargetAddr::Domain("later.task".into(), 1080); - write_request(&mut client, &Socks5Command::UDPAssociate, &req_target).await; + write_request( + &mut client, + &Socks5Command::UDPAssociate, + &TargetAddr::Domain("relay.test".into(), 1080), + ) + .await; + let (code, addr) = read_reply(&mut client).await; + assert_eq!(code, 0x00, "the associate reply succeeds"); + assert_eq!( + (addr[0], addr[1], addr[2], addr[3], addr[4], addr[5]), + (127, 0, 0, 1, 0, 0), + "the reply carries the sentinel BND.ADDR 127.0.0.1:0" + ); + (handle, client) + } + + /// The echo egress over the queue shape (the harness's + /// `echo_egress`, inline): datagrams sent echo back as inner + /// bytes. + fn echo_egress_factory() -> EgressFactory { + Arc::new(|| { + Box::pin(async { + let (inbound, outbound) = tokio::sync::mpsc::unbounded_channel(); + Ok(Box::new(QueueEgress { + inbound, + outbound: tokio::sync::Mutex::new(outbound), + }) as Box) + }) + }) + } + + /// The ASSOCIATE happy path: sentinel reply → datagram + /// round-trip through the echo egress (the frame shape both + /// directions, in-order across multiple datagrams on one + /// association). + #[tokio::test] + async fn udp_associate_round_trip_through_the_echo_egress() { + let (handle, mut client) = raw_associate(Some(echo_egress_factory())).await; + + // Multiple datagrams, in order, one association. + let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 1], 53))); + for i in 0..5u8 { + let payload = vec![i; 10]; + client + .write_all(&datagram_frame(&target, &payload)) + .await + .unwrap(); + } + for i in 0..5u8 { + let mut buf = vec![0u8; 2 + 7 + 10]; + client.read_exact(&mut buf).await.expect("reply frame"); + let (src, echoed) = decode_datagram(&buf); + assert_eq!(src, target, "the echo reports the destination"); + assert_eq!(&echoed[..], &vec![i; 10][..], "in-order echo"); + } + + // The association is alive: another datagram round-trips. + client + .write_all(&datagram_frame(&target, b"alive")) + .await + .unwrap(); + let mut buf = vec![0u8; 2 + 7 + 5]; + client.read_exact(&mut buf).await.expect("reply frame"); + let (_, echoed) = decode_datagram(&buf); + assert_eq!(&echoed[..], b"alive"); + + // Channel EOF ends the stage cleanly (the association + // lifetime = the channel's). + drop(client); + let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("session ends on channel EOF") + .expect("join"); + result.expect("clean session"); + } + + /// An empty-payload datagram is a legal datagram — never EOF (the + /// F-2 invariant at payload level, end to end): it round-trips + /// through the echo egress and the association survives it. + #[tokio::test] + async fn udp_empty_payload_datagram_is_legal_not_eof() { + let (handle, mut client) = raw_associate(Some(echo_egress_factory())).await; + let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 2], 53))); + + // pre → empty (len = 7 inner: address only) → post, in order; + // the empty datagram must echo as an empty datagram (a real + // datagram), and the association must survive it. + client + .write_all(&datagram_frame(&target, b"pre")) + .await + .unwrap(); + let empty = datagram_frame(&target, b""); + assert_eq!( + empty.len(), + 2 + 7, + "the empty payload's frame is addr-only inner" + ); + client.write_all(&empty).await.unwrap(); + client + .write_all(&datagram_frame(&target, b"post")) + .await + .unwrap(); + + let mut buf = vec![0u8; 2 + 7 + 3]; + client.read_exact(&mut buf).await.expect("pre echo"); + let (_, echoed) = decode_datagram(&buf); + assert_eq!(&echoed[..], b"pre"); + + let mut buf = vec![0u8; 2 + 7]; + client.read_exact(&mut buf).await.expect("empty echo"); + let (src, echoed) = decode_datagram(&buf); + assert_eq!(src, target); + assert!( + echoed.is_empty(), + "the empty-payload datagram round-trips (never EOF)" + ); + + let mut buf = vec![0u8; 2 + 7 + 4]; + client.read_exact(&mut buf).await.expect("post echo"); + let (_, echoed) = decode_datagram(&buf); + assert_eq!(&echoed[..], b"post"); + + drop(client); + let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("session ends") + .expect("join"); + result.expect("clean session"); + } + + /// A undecodable frame (garbage ATYP — any post-reply control + /// byte parses as one) fails the association loud: EOF on the + /// next read, no silent drop. + #[tokio::test] + async fn udp_undecodable_frame_ends_association() { + let (handle, mut client) = raw_associate(Some(echo_egress_factory())).await; + + // Hand-craft a frame whose inner bytes start with an + // unassigned ATYP. + let inner = vec![0x7Fu8, 1, 2, 3]; + let mut garbage_frame = (inner.len() as u16).to_be_bytes().to_vec(); + garbage_frame.extend_from_slice(&inner); + client.write_all(&garbage_frame).await.unwrap(); + + // The association ends: EOF (no reply datagram follows). + let mut buf = [0u8; 16]; + let outcome = + tokio::time::timeout(std::time::Duration::from_secs(5), client.read(&mut buf)).await; + match outcome { + Err(_) => panic!("the association must end after an undecodable frame"), + Ok(Ok(0)) | Ok(Err(_)) => {} + Ok(Ok(n)) => panic!("EOF expected after an undecodable frame — got {n} bytes"), + } + drop(client); + tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("session ends") + .expect("join") + .expect("the stage exited"); + } + + /// Channel EOF ends the association and the connection stays + /// healthy (the POC's test 4, in-process shape): after the client + /// half-closes one association, a fresh ASSOCIATE on a fresh + /// duplex succeeds immediately (the same registration is + /// unpoisoned). + #[tokio::test] + async fn udp_channel_eof_ends_association_and_connection_stays_healthy() { + let egress = echo_egress_factory(); + let (handle, mut client) = raw_associate(Some(Arc::clone(&egress))).await; + + // Half-close: EOF to the stage → the association ends. + use tokio::io::AsyncWriteExt as _; + client.shutdown().await.unwrap(); + let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("session ends on channel EOF") + .expect("join"); + result.expect("clean session (EOF is the clean exit)"); + drop(client); + + // A fresh associate succeeds immediately after (the same + // registration, unpoisoned). + let (handle, mut client) = raw_associate(Some(Arc::clone(&egress))).await; + let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 3], 53))); + client + .write_all(&datagram_frame(&target, b"fresh")) + .await + .unwrap(); + let mut buf = vec![0u8; 2 + 7 + 5]; + client.read_exact(&mut buf).await.expect("fresh echo"); + let (_, echoed) = decode_datagram(&buf); + assert_eq!(&echoed[..], b"fresh"); + drop(client); + tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("session ends") + .expect("join") + .expect("clean"); + } + + /// The readiness contract (the POC's load-bearing API finding, + /// re-pinned by this suite per ADR-003): an egress whose + /// `next_reply` is an instantly-ready `Ok(None)` must not starve + /// the stage's channel branch — datagrams still egress and echo. + /// (The stage tolerates spurious `Ok(None)`s by re-parking.) + #[tokio::test] + async fn udp_instant_ok_none_egress_does_not_starve_the_stage() { + struct HotNoneEgress { + inbound: tokio::sync::mpsc::UnboundedSender, + outbound: tokio::sync::Mutex>, + } + + impl crate::egress::Egress for HotNoneEgress { + fn send( + &mut self, + addr: TargetAddr, + payload: Bytes, + ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> { + Box::pin(async move { + let mut inner = addr + .to_be_bytes() + .map_err(|e| EgressError::General(format!("addr encode: {e}")))?; + inner.extend_from_slice(&payload); + self.inbound + .send(Bytes::from(inner)) + .map_err(|_| EgressError::General("egress closed".into())) + }) + } + + fn next_reply( + &mut self, + ) -> futures::future::BoxFuture<'_, Result, EgressError>> { + Box::pin(async move { + // Drain one queued reply if any, then — instead of + // parking — return an instantly-ready `Ok(None)` + // (the contract violation the stage must tolerate; + // an actual biased-select livelock would hang this + // test's channel branch). + match self.outbound.lock().await.try_recv() { + Ok(inner) => Ok(Some(inner)), + Err(_) => Ok(None), + } + }) + } + } + + let factory: EgressFactory = Arc::new(|| { + Box::pin(async { + let (inbound, outbound) = tokio::sync::mpsc::unbounded_channel(); + Ok(Box::new(HotNoneEgress { + inbound, + outbound: tokio::sync::Mutex::new(outbound), + }) as Box) + }) + }); + + let (handle, mut client) = raw_associate(Some(factory)).await; + let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 4], 53))); + + // The datagrams must still egress (the channel branch + // progresses) despite the hot `Ok(None)`s. + client + .write_all(&datagram_frame(&target, b"hot")) + .await + .unwrap(); + let mut buf = vec![0u8; 2 + 7 + 3]; + tokio::time::timeout(std::time::Duration::from_secs(5), async { + client.read_exact(&mut buf).await + }) + .await + .expect("the channel branch is not starved — the echo arrived") + .expect("read the echo"); + let (_, echoed) = decode_datagram(&buf); + assert_eq!(&echoed[..], b"hot"); + + drop(client); + tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("session ends") + .expect("join") + .expect("clean"); + } + + /// A missing egress callback refuses ASSOCIATE RFC-faithfully + /// (`CommandNotSupported` 0x07 — ADR-008's missing-callback class, + /// never a panic), and CONNECT still serves the same registration + /// (the regression guard). + #[tokio::test] + async fn missing_egress_refuses_associate_command_not_supported() { + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let dial = echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new()))); + let handle = tokio::spawn(serve_socks5_over(server_end, Some(dial), None, None, false)); + let mut client = client_end; + greet(&mut client).await; + write_request( + &mut client, + &Socks5Command::UDPAssociate, + &TargetAddr::Domain("no.egress".into(), 1080), + ) + .await; let (code, _addr) = read_reply(&mut client).await; - assert_eq!(code, 0x07, "ASSOCIATE refuses CommandNotSupported"); + assert_eq!(code, 0x07, "missing egress → CommandNotSupported"); drop(client); let err = tokio::time::timeout(std::time::Duration::from_secs(5), handle) .await .expect("session ends") .expect("join") - .expect_err("refusal surfaced in-stream"); + .expect_err("the refusal surfaces in-stream"); assert!(matches!( err, SocksError::ServerError(SocksServerError::UnknownCommand(_)) )); + + // CONNECT still serves the same registration shape (the + // regression guard). + let (handle, mut client) = + raw_connect(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new())))).await; + greet(&mut client).await; + let (code, _addr) = read_reply_after_connect(&mut client).await; + assert_eq!(code, 0x00, "CONNECT still serves"); + drop(client); + tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("session ends") + .expect("join") + .expect("clean"); + } + + /// The egress-factory failure (post-command, pre-reply) maps to + /// reply#1's code — no phantom datagram plane. + #[tokio::test] + async fn egress_factory_failure_maps_to_reply_code() { + let factory: EgressFactory = + Arc::new(|| Box::pin(async { Err(EgressError::Unreachable("no egress".into())) })); + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let dial = echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new()))); + let handle = tokio::spawn(serve_socks5_over( + server_end, + Some(dial), + Some(factory), + None, + false, + )); + let mut client = client_end; + greet(&mut client).await; + write_request( + &mut client, + &Socks5Command::UDPAssociate, + &TargetAddr::Domain("down.egress".into(), 1080), + ) + .await; + let (code, _addr) = read_reply(&mut client).await; + assert_eq!(code, 0x03, "egress-factory refusal → NetworkUnreachable"); + drop(client); + let err = tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("session ends") + .expect("join") + .expect_err("the refusal surfaces in-stream"); + assert!(matches!( + err, + SocksError::ServerError(SocksServerError::Bug("egress refused in-stream")) + )); + } + + /// The egress lifecycle: the factory is called once per + /// ASSOCIATE, the stage owns the egress, and stage exit drops it + /// (the teardown-at-birth witness — the drop flag flips when the + /// channel EOFs, not before). + #[tokio::test] + async fn egress_factory_called_once_and_dropped_with_the_stage() { + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + + static CALLS: AtomicUsize = AtomicUsize::new(0); + let dropped = Arc::new(AtomicBool::new(false)); + let dropped_for_factory = Arc::clone(&dropped); + + let factory: EgressFactory = Arc::new(move || { + CALLS.fetch_add(1, Ordering::SeqCst); + let dropped = Arc::clone(&dropped_for_factory); + Box::pin(async move { + let (inbound, outbound) = tokio::sync::mpsc::unbounded_channel(); + struct Witness { + flag: Arc, + inbound: tokio::sync::mpsc::UnboundedSender, + outbound: tokio::sync::Mutex>, + } + impl crate::egress::Egress for Witness { + fn send( + &mut self, + addr: TargetAddr, + payload: Bytes, + ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> + { + let mut inner = addr.to_be_bytes().unwrap().to_vec(); + inner.extend_from_slice(&payload); + Box::pin(async move { + self.inbound + .send(Bytes::from(inner)) + .map_err(|_| EgressError::General("egress closed".into())) + }) + } + + fn next_reply( + &mut self, + ) -> futures::future::BoxFuture<'_, Result, EgressError>> + { + Box::pin(async move { + match self.outbound.lock().await.recv().await { + Some(inner) => Ok(Some(inner)), + None => Err(EgressError::General("egress closed".into())), + } + }) + } + } + impl Drop for Witness { + fn drop(&mut self) { + self.flag.store(true, Ordering::SeqCst); + } + } + Ok(Box::new(Witness { + flag: dropped, + inbound, + outbound: tokio::sync::Mutex::new(outbound), + }) as Box) + }) + }); + + let (handle, mut client) = raw_associate(Some(factory)).await; + assert_eq!(CALLS.load(Ordering::SeqCst), 1, "the factory ran once"); + assert!( + !dropped.load(Ordering::SeqCst), + "the egress lives while the stage runs" + ); + + // One round-trip (the stage is alive). + let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 4], 53))); + client + .write_all(&datagram_frame(&target, b"witness")) + .await + .unwrap(); + let mut buf = vec![0u8; 2 + 7 + 7]; + client.read_exact(&mut buf).await.expect("echo"); + let (_, echoed) = decode_datagram(&buf); + assert_eq!(&echoed[..], b"witness"); + + // Channel EOF → stage exit → the egress drops. + drop(client); + tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("session ends") + .expect("join") + .expect("clean"); + assert!( + dropped.load(Ordering::SeqCst), + "stage exit drops the egress (teardown-at-birth)" + ); + } + + /// CONNECT and ASSOCIATE coexist on one `serve_socks5_over` (the + /// regression guard, extended): both commands serve their data + /// planes over separate duplexes on the same seam set. + #[tokio::test] + async fn connect_and_associate_serve_on_the_same_registration() { + let targets = Arc::new(tokio::sync::Mutex::new(Vec::new())); + let dial = echo_dial(Arc::clone(&targets)); + + // CONNECT first. + let (client_end, server_end) = tokio::io::duplex(64 * 1024); + let serve = tokio::spawn(serve_socks5_over( + server_end, + Some(Arc::clone(&dial)), + Some(echo_egress_factory()), + None, + false, + )); + let mut client = client_end; + greet(&mut client).await; + write_request( + &mut client, + &Socks5Command::TCPConnect, + &TargetAddr::Domain("tcp.test".into(), 443), + ) + .await; + let (code, _addr) = read_reply(&mut client).await; + assert_eq!(code, 0x00); + client.write_all(b"tcp").await.unwrap(); + let mut buf = [0u8; 3]; + client.read_exact(&mut buf).await.unwrap(); + assert_eq!(&buf, b"tcp"); + drop(client); + tokio::time::timeout(std::time::Duration::from_secs(5), serve) + .await + .expect("session ends") + .expect("join") + .expect("clean"); + + // ASSOCIATE second — the same seams. + let (handle, mut client) = raw_associate(Some(echo_egress_factory())).await; + let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 5], 53))); + client + .write_all(&datagram_frame(&target, b"udp")) + .await + .unwrap(); + let mut buf = vec![0u8; 2 + 7 + 3]; + client.read_exact(&mut buf).await.expect("echo"); + let (_, echoed) = decode_datagram(&buf); + assert_eq!(&echoed[..], b"udp"); + drop(client); + tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("session ends") + .expect("join") + .expect("clean"); + } + + /// The CONNECT conversation's front half (the regression guard's + /// helper: greet + request + reply). + async fn read_reply_after_connect(client: &mut tokio::io::DuplexStream) -> (u8, [u8; 6]) { + write_request( + client, + &Socks5Command::TCPConnect, + &TargetAddr::Domain("echo.test".into(), 443), + ) + .await; + read_reply(client).await } /// The dispatch skeleton + the two-pump shape coexist: CONNECT diff --git a/src/wire.rs b/src/wire.rs index 34ae767..64bb5ec 100644 --- a/src/wire.rs +++ b/src/wire.rs @@ -158,6 +158,23 @@ pub fn frame(addr: &TargetAddr, payload: &[u8]) -> Result { Ok(buf.freeze()) } +/// Frame pre-encoded inner bytes (`addr + data` — exactly what +/// [`crate::egress::Egress::next_reply`] returns) as one +/// `[len: u16 BE][inner]` frame. The reply-direction half of the +/// framing-once invariant: the egress speaks inner bytes (no length +/// prefix), and this adds the single prefix at the wire layer. Rejects +/// [`WireError::Oversize`] when the inner bytes exceed +/// [`MAX_INNER_LEN`] (the u16 length field would wrap). +pub fn frame_inner(inner: &[u8]) -> Result { + if inner.len() > MAX_INNER_LEN { + return Err(WireError::Oversize(inner.len(), MAX_INNER_LEN)); + } + let mut buf = BytesMut::with_capacity(FRAME_LEN_FIELD + inner.len()); + buf.put_u16(inner.len() as u16); + buf.put_slice(inner); + Ok(buf.freeze()) +} + /// Incremental datagram decoder over the framed stage's read /// direction. Feed it whatever `read()` yields; it reassembles /// `[len: u16 BE][addr + data]` frames across chunk boundaries — a @@ -507,6 +524,46 @@ mod tests { } } + #[test] + fn frame_inner_frames_the_egress_inner_bytes_exactly_once() { + let inner = { + let mut v = v4(53).to_be_bytes().unwrap().to_vec(); + v.extend_from_slice(b"answer"); + v + }; + let framed = frame_inner(&inner).unwrap(); + assert_eq!(&framed[..2], &(inner.len() as u16).to_be_bytes()); + assert_eq!(&framed[2..], &inner[..]); + // The read direction decodes it back: the same (addr, payload) + // pair the egress encoded (the round-trip shape). + let mut r = DatagramReader::new(); + let dgs = r.feed(&framed); + assert_eq!(dgs.len(), 1); + let (src, payload) = dgs.into_iter().next().unwrap().unwrap(); + assert_eq!(src, v4(53)); + assert_eq!(&payload[..], b"answer"); + } + + #[test] + fn frame_inner_rejects_inner_bytes_over_the_u16_bound() { + let err = frame_inner(&vec![0u8; MAX_INNER_LEN + 1]).unwrap_err(); + assert!(matches!(err, WireError::Oversize(65536, 65535))); + let framed = frame_inner(&vec![0u8; MAX_INNER_LEN]).unwrap(); + assert_eq!(framed.len(), 2 + MAX_INNER_LEN); + } + + #[test] + fn frame_and_frame_inner_produce_identical_wires() { + // The reply path (frame_inner over pre-encoded inner bytes) + // and the forward path (frame over addr + payload) agree + // byte-for-byte — one codec, two callers. + let addr = domain("example.com", 53); + let payload = b"payload"; + let mut inner = addr.to_be_bytes().unwrap().to_vec(); + inner.extend_from_slice(payload); + assert_eq!(frame_inner(&inner).unwrap(), frame(&addr, payload).unwrap()); + } + #[test] fn oversize_rejected_at_frame_time() { let addr = v4(80); diff --git a/tasks/implementation/core-producer-associate.md b/tasks/implementation/core-producer-associate.md index 7ee7c69..64e6f77 100644 --- a/tasks/implementation/core-producer-associate.md +++ b/tasks/implementation/core-producer-associate.md @@ -1,7 +1,7 @@ --- id: implementation/core-producer-associate name: "Core — producer ASSOCIATE: the select!-driven datagram stage" -status: pending +status: completed depends_on: [implementation/core-wire-codec, implementation/core-producer-connect] scope: moderate risk: medium @@ -62,16 +62,16 @@ Deliverables (port from `/workspace/alksocks-udp-poc/src/producer.rs` ## Acceptance Criteria -- [ ] The datagram stage per the three POC invariants (concurrent +- [x] The datagram stage per the three POC invariants (concurrent select, readiness contract, framing-once) -- [ ] Sentinel BND.ADDR in the reply; association lifetime = channel +- [x] Sentinel BND.ADDR in the reply; association lifetime = channel lifetime (EOF teardown tested) -- [ ] Readiness-contract test present (instant-Ok(None) egress does +- [x] Readiness-contract test present (instant-Ok(None) egress does not starve the stage) -- [ ] Missing-egress refusal RFC-faithful -- [ ] CONNECT regression guard passes (one registration, both +- [x] Missing-egress refusal RFC-faithful +- [x] CONNECT regression guard passes (one registration, both commands) -- [ ] `cargo test` green; clippy/fmt clean; both wasm checks clean +- [x] `cargo test` green; clippy/fmt clean; both wasm checks clean ## References @@ -86,8 +86,75 @@ Deliverables (port from `/workspace/alksocks-udp-poc/src/producer.rs` ## Notes -> To be filled by implementation agent +- **Wire codec addition** (`src/wire.rs`): `frame_inner()` — the + reply-direction half of the framing-once invariant (frames + pre-encoded inner bytes from the egress; the forward-path + `frame()` and the reply-path `frame_inner()` produce identical + wires — pinned byte-for-byte). Exported from lib.rs. `DatagramReader` + and `frame()` were already in place from the wire-codec task. +- **Stage shape** (`src/producer.rs`): `serve_udp_associate` drives + the typestate directly (POC #2's shape — no `run_udp_proxy_custom`), + then `run_datagram_stage` is a `select!` loop over channel reads ∥ + `egress.next_reply()`, returning a typed `StageExit` diagnostic + (ChannelEof / UndecodableFrame / EgressEnded / ReplyFrameRejected / + Io). The egress `send` is awaited inline before the next select + iteration (the POC's measured interleaved shape); no + spawn-per-datagram, no flow table. +- **Exit semantics**: channel EOF is the clean exit (the association + lifetime); undecodable frames (garbage ATYP, `len == 0`) and + egress `Err` end the association loud; a reply the joint bound + rejects (`ReplyFrameRejected`) is its own exit (never a wire + overflow). The stage owns the egress structurally — the egress + binding is moved into the stage call; every exit drops it + (drop-witness test pins the teardown-at-birth discipline). +- **Missing-egress refusal**: `CommandNotSupported` (0x07) — the + ADR-008 missing-callback class, same shape as CONNECT's + missing-dial refusal. Egress-factory failure (post-command, + pre-reply) → reply#1 with the mapped RFC code + (`NetworkUnreachable` for the Unreachable variant), surfaced + in-stream as `SocksServerError::Bug("egress refused in-stream")`. +- The stale ASSOCIATE-refuses tests (`associate_arm_refuses_…` + in producer.rs, `producer_associate_refuses_0x07_…` in the + harness) were re-pinned to the missing-egress semantics; the + harness three-command regression guard now drives all three + commands serving (CONNECT + BIND + a datagram round-trip). +- The consumer side (the `AssociateSession` mirror) is the + consumer-associate task; its `send_datagram` will frame via + `frame()` and its `recv_datagram` will decode via the persistent + `DatagramReader` — both already exported. ## Summary -> To be filled on completion \ No newline at end of file +The producer ASSOCIATE path is implemented and verified. The +dispatch arm in `serve_socks5_over` now branches to the datagram +stage: sentinel reply (`127.0.0.1:0`, ADR-003's RFC-compatibility +surface) → the `select!`-driven stage (channel reads ∥ egress +replies) with the three POC invariants as code: + +1. **Concurrent select shape** — both branches of the loop are + awaited concurrently; no poll-then-block window (the reply-loss + hazard the POC measured). +2. **The egress readiness contract** — spurious `Ok(None)`s are + tolerated by re-parking; the contract test (`udp_instant_ok_none_egress_does_not_starve_the_stage`) + pins that an instantly-ready `Ok(None)` egress must not starve + the channel branch (the POC's load-bearing API finding, + re-pinned by this suite). +3. **Framing applied exactly once, by the wire layer** — the stage + reads frames via `DatagramReader` (egress `send` gets the bare + payload) and writes frames via the new `frame_inner` (egress + `next_reply`'s inner bytes → the channel); the egress never sees + a length prefix. + +Association lifetime = channel lifetime (EOF is the clean exit; +tested in-process and over channels — a fresh associate succeeds +immediately after, the POC's test 4). Missing egress refuses +RFC-faithfully (0x07); egress-factory failure maps to reply#1's +code. CONNECT still serves on the same registration (regression +guards at both layers). + +Verification: `cargo test` 174/174 (107 lib + 30 harness + 37 +differential), `cargo clippy --all-targets -- -D warnings` clean, +`cargo fmt --check` clean, `cargo check --target +wasm32-unknown-unknown` + `cargo clippy --target +wasm32-unknown-unknown -- -D warnings` clean, `cargo doc --no-deps` +clean. \ No newline at end of file diff --git a/tests/harness.rs b/tests/harness.rs index 5f38bde..800f4f2 100644 --- a/tests/harness.rs +++ b/tests/harness.rs @@ -1349,16 +1349,15 @@ mod tests { assert_eq!(code, consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED); } - /// The ASSOCIATE dispatch arm refuses RFC-faithfully - /// (`CommandNotSupported` 0x07 — the pre-vertical posture; the - /// producer-associate task replaces it) while CONNECT still - /// serves on the same registration. BIND now serves under the - /// gate — its integration tests live below. + /// ASSOCIATE refuses RFC-faithfully when the egress seam is not + /// wired (`CommandNotSupported` 0x07 — ADR-008's missing-callback + /// class) while CONNECT still serves the same registration (the + /// missing-callback posture, over channels). #[tokio::test] async fn producer_associate_refuses_0x07_connect_still_serves() { let topo = wire(echo_dial()).await; - // ASSOCIATE refuses. + // ASSOCIATE refuses (no egress seam wired). let (mut raw, _pump) = open_channel_and_pump(&topo).await; raw.greet().await.expect("greet"); raw.request( @@ -1384,6 +1383,264 @@ mod tests { assert_eq!(&buf, b"still"); } + // -- the producer ASSOCIATE datagram stage over channels ------------- + + /// Frame one datagram the way the in-band wire carries it + /// (`[len: u16 BE][addr + data]` — the wire codec's shape). + fn udp_frame(addr: &TargetAddr, payload: &[u8]) -> Vec { + alksocks::wire::frame(addr, payload) + .expect("frame") + .to_vec() + } + + /// Decode a datagram from the wire (one frame; in-order batches). + fn udp_unframe(wire: &[u8]) -> (TargetAddr, Bytes) { + let mut reader = alksocks::wire::DatagramReader::new(); + let mut got = None; + for dg in reader.feed(wire) { + got = Some(dg); + } + got.expect("one datagram").expect("frame decodes") + } + + /// The ASSOCIATE conversation over a channel: open + adopt + + /// pump, greet, request, reply — asserting the sentinel + /// `127.0.0.1:0` BND.ADDR (ADR-003's RFC-compatibility surface). + async fn raw_associate( + topo: &Topology, + ) -> ( + RawRfcClient, + tokio::task::JoinHandle<(u64, u64)>, + ) { + let (mut raw, pump) = open_channel_and_pump(topo).await; + raw.greet().await.expect("greet"); + raw.request( + Socks5Command::UDPAssociate, + &TargetAddr::Domain("relay.test".into(), 1080), + ) + .await + .expect("request written"); + let (code, addr) = raw.read_reply().await.expect("reply read"); + assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED, "the associate reply"); + assert_eq!( + (addr[0], addr[1], addr[2], addr[3], addr[4], addr[5]), + (127, 0, 0, 1, 0, 0), + "the reply carries the sentinel BND.ADDR" + ); + (raw, pump) + } + + /// The datagram stage end to end over channels (the POC #2 + /// shape): sentinel reply → datagrams round-trip through the echo + /// egress in order, one association; EOF (channel shutdown) ends + /// the association and a fresh associate succeeds immediately + /// (the POC's test 4); the empty-payload datagram is legal (never + /// EOF); a garbage-ATYP frame ends the association loud. + #[tokio::test] + async fn producer_udp_associate_datagram_stage_over_channels() { + init_trace(); + let topo = wire_with( + RegisterMode::Producer { + dial: echo_dial(), + egress: Some(echo_egress()), + accept: None, + }, + Some(consumer_identity()), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await; + + // Associate: the sentinel reply, then the datagram stage. + let (mut raw, pump) = raw_associate(&topo).await; + let target = TargetAddr::Ip("10.0.0.1:53".parse().unwrap()); + + // Multiple datagrams, in order, one association. + for i in 0..5u8 { + raw.stream + .write_all(&udp_frame(&target, &[i; 10])) + .await + .expect("datagram sent"); + } + for i in 0..5u8 { + let mut buf = vec![0u8; 2 + 7 + 10]; + tokio::time::timeout( + std::time::Duration::from_secs(10), + raw.stream.read_exact(&mut buf), + ) + .await + .expect("echo timed out") + .expect("reply frame read"); + let (src, payload) = udp_unframe(&buf); + assert_eq!(src, target, "the echo reports the destination"); + assert_eq!(&payload[..], &vec![i; 10][..], "in-order echo"); + } + + // The empty-payload datagram: legal, never EOF (the F-2 + // invariant at payload level). + raw.stream + .write_all(&udp_frame(&target, b"")) + .await + .expect("empty datagram sent"); + let mut buf = vec![0u8; 2 + 7]; + tokio::time::timeout( + std::time::Duration::from_secs(10), + raw.stream.read_exact(&mut buf), + ) + .await + .expect("empty echo timed out") + .expect("reply frame read"); + let (_, payload) = udp_unframe(&buf); + assert!(payload.is_empty(), "the empty datagram round-trips"); + + // A garbage-ATYP frame ends the association loud: EOF on the + // next read. + raw.stream + .write_all(&[0x00, 0x04, 0x7f, 1, 2, 3]) + .await + .expect("garbage frame sent"); + let mut buf = [0u8; 16]; + let outcome = tokio::time::timeout( + std::time::Duration::from_secs(10), + raw.stream.read(&mut buf), + ) + .await + .expect("read resolves"); + match outcome { + Ok(0) | Err(_) => {} + Ok(n) => panic!("EOF expected after an undecodable frame — got {n} bytes"), + } + drop(raw); + tokio::time::timeout(std::time::Duration::from_secs(10), pump) + .await + .expect("the stage pump ends") + .expect("join"); + } + + /// Channel EOF ends the association and the connection stays + /// healthy (the POC's test 4, over channels): after one + /// association's channel is torn down, a fresh associate succeeds + /// immediately and round-trips. + #[tokio::test] + async fn producer_udp_associate_eof_teardown_connection_stays_healthy() { + init_trace(); + let topo = wire_with( + RegisterMode::Producer { + dial: echo_dial(), + egress: Some(echo_egress()), + accept: None, + }, + Some(consumer_identity()), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await; + + // First association: EOF ends it (half-close the channel). + let (raw, pump) = raw_associate(&topo).await; + drop(raw); + tokio::time::timeout(std::time::Duration::from_secs(10), pump) + .await + .expect("the stage pump ends on channel EOF") + .expect("join"); + + // The connection is healthy: a fresh associate round-trips. + let (mut raw, _pump) = raw_associate(&topo).await; + let target = TargetAddr::Ip("10.0.0.2:53".parse().unwrap()); + raw.stream + .write_all(&udp_frame(&target, b"fresh")) + .await + .expect("datagram sent"); + let mut buf = vec![0u8; 2 + 7 + 5]; + tokio::time::timeout( + std::time::Duration::from_secs(10), + raw.stream.read_exact(&mut buf), + ) + .await + .expect("fresh echo timed out") + .expect("reply frame read"); + let (_, payload) = udp_unframe(&buf); + assert_eq!(&payload[..], b"fresh"); + } + + /// The egress-factory refusal maps to reply#1's mapped code over + /// channels (the pre-reply refusal path) — no phantom datagram + /// plane. + #[tokio::test] + async fn producer_udp_associate_factory_refusal_maps_to_reply_code() { + let topo = wire_with( + RegisterMode::Producer { + dial: echo_dial(), + egress: Some(refusing_egress(EgressError::Unreachable( + "no egress".into(), + ))), + accept: None, + }, + Some(consumer_identity()), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await; + + let (mut raw, _pump) = open_channel_and_pump(&topo).await; + raw.greet().await.expect("greet"); + raw.request( + Socks5Command::UDPAssociate, + &TargetAddr::Domain("relay.test".into(), 1080), + ) + .await + .expect("request written"); + let (code, _addr) = raw.read_reply().await.expect("reply read"); + assert_eq!( + code, + consts::SOCKS5_REPLY_NETWORK_UNREACHABLE, + "the factory refusal maps to reply#1's code" + ); + } + + /// CONNECT + ASSOCIATE on the same registration over channels + /// (the regression guard, extended — one registration, both + /// commands). + #[tokio::test] + async fn producer_connect_and_associate_on_the_same_registration() { + let topo = wire_with( + RegisterMode::Producer { + dial: echo_dial(), + egress: Some(echo_egress()), + accept: None, + }, + Some(consumer_identity()), + None, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await; + + // CONNECT serves. + let (mut raw, _pump) = open_channel_and_pump(&topo).await; + let code = raw + .connect(&TargetAddr::Domain("echo.test".into(), 443)) + .await + .expect("connect"); + assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); + raw.stream.write_all(b"tcp").await.expect("write"); + let mut buf = [0u8; 3]; + raw.stream.read_exact(&mut buf).await.expect("echo"); + assert_eq!(&buf, b"tcp"); + drop(raw); + + // ASSOCIATE serves the same registration. + let (mut raw, _pump) = raw_associate(&topo).await; + let target = TargetAddr::Ip("10.0.0.3:53".parse().unwrap()); + raw.stream + .write_all(&udp_frame(&target, b"udp")) + .await + .expect("datagram sent"); + let mut buf = vec![0u8; 2 + 7 + 3]; + raw.stream.read_exact(&mut buf).await.expect("echo"); + let (_, payload) = udp_unframe(&buf); + assert_eq!(&payload[..], b"udp"); + } + /// BIND over channels (the phase model end to end): the accept /// factory resolves at command dispatch, reply#1 carries the /// advertised BND.ADDR, reply#2 carries the accepted peer, and @@ -1623,16 +1880,16 @@ mod tests { .expect("join"); } - /// The three-command regression guard over channels (POC #5 test - /// 7): one registration, all three commands — CONNECT serves, - /// BIND serves under its gate, ASSOCIATE refuses (its task - /// pending). Fresh channel per session. + /// The three-command regression guard (POC #5 test 7): one + /// registration, all three commands — CONNECT serves, BIND serves + /// under its gate, ASSOCIATE serves with the echo egress. Fresh + /// channel per session. #[tokio::test] async fn producer_three_command_regression_guard() { let topo = wire_with( RegisterMode::Producer { dial: echo_dial(), - egress: None, + egress: Some(echo_egress()), accept: Some(echo_accept()), }, Some(both_scopes_identity()), @@ -1675,18 +1932,24 @@ mod tests { assert_eq!(code2, consts::SOCKS5_REPLY_SUCCEEDED); drop(raw); - // ASSOCIATE still refuses (the producer-associate task is - // pending). - let (mut raw, _pump) = open_channel_and_pump(&topo).await; - raw.greet().await.expect("greet"); - raw.request( - Socks5Command::UDPAssociate, - &TargetAddr::Domain("relay.test".into(), 1080), + // ASSOCIATE serves: the sentinel reply + one datagram + // round-trip. + let (mut raw, _pump) = raw_associate(&topo).await; + let target = TargetAddr::Ip("10.0.0.9:53".parse().unwrap()); + raw.stream + .write_all(&udp_frame(&target, b"all3")) + .await + .expect("datagram sent"); + let mut buf = vec![0u8; 2 + 7 + 4]; + tokio::time::timeout( + std::time::Duration::from_secs(10), + raw.stream.read_exact(&mut buf), ) .await - .expect("request written"); - let (code, _addr) = raw.read_reply().await.expect("reply read"); - assert_eq!(code, consts::SOCKS5_REPLY_COMMAND_NOT_SUPPORTED); + .expect("echo timed out") + .expect("reply frame read"); + let (_, payload) = udp_unframe(&buf); + assert_eq!(&payload[..], b"all3"); } /// BIND two-stage failure mapping over channels: factory failure