From c57451545d4fda30d6009cc60f3804bb4e9f1cd6 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Thu, 24 Sep 2026 07:15:27 +0000 Subject: [PATCH] =?UTF-8?q?consumer:=20AssociateSession=20=E2=80=94=20the?= =?UTF-8?q?=20framed=20datagram=20stage=20+=20e2e=20suite?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The UDP ASSOCIATE consumer session (POC #2's shape, consumer.md §The associate session, ADR-003): - `AssociateSession::associate` — open → no-auth → ASSOCIATE → reply; the sentinel BND.ADDR is received (`relay_addr()`, RFC fidelity) and never used for addressing (the handle is the channel). The ASSOCIATE DST.ADDR is the vanilla-mirror unspecified address. - `send_datagram` — frames exactly once via the crate's wire `frame()` (the framing-once mirror; the u16 joint bound enforced at frame time). Takes `impl ToTargetAddr` (ADR-012 posture); a `ToTargetAddr for &TargetAddr` impl was added to the forked codec so a `TargetAddr` value survives across sends. - `recv_datagram` — the incremental `DatagramReader` + batched- datagram pending queue, state persisting across calls (the POC's reader-state finding); `Ok(None)` on channel EOF; fail-loud on a stream end mid-frame and on undecodable frames (`WireError` → the forked `AddrError`/`UdpHeaderError` RFC classes — the consumer mirror of the producer stage's `UndecodableFrame` exit). - `shutdown` — the half-close; the mux's implicit-EOF sentinel reaches the producer's stage as EOF (association lifetime = channel lifetime). `close()`/`Drop` own the teardown (sync reap + structural EOF, the CONNECT/BIND posture); teardown-at-birth on in-stream refusals; no `Clone` (compile-fail asserted). - Typed error surface: open failures are `SocksSessionError::Open` (ADR-049); the missing-egress 0x07 and factory-failure codes surface from `associate` as `Rfc`. E2E suite (`tests/consumer-associate.rs`, 8 tests over the harness): round-trips (multiple in-order datagrams, one association, sentinel relay address asserted), the split-datagram persistence test (a 60 KiB datagram reassembling across the call boundary + the pending- queue drain — both halves of the persistence invariant), the empty- payload round-trip (F-2, never EOF), shutdown → a fresh associate succeeds immediately (the POC's test 4), Drop teardown + connection health, the missing-egress refusal (typed 0x07, no phantom channel), the egress-factory failure mapping, and the CONNECT+ASSOCIATE regression guard (interleaved on one registration). Unit tests (4) pin the reader internals over a duplex (`skip_auth`'s handshake-free path). Verification: cargo test 305 (7 suites, all green), clippy --all-targets -D warnings, fmt, cargo doc --no-deps, and both wasm32-unknown-unknown checks clean. --- src/consumer.rs | 526 +++++++++++++++++- src/lib.rs | 5 +- src/rfc1928/util/target_addr.rs | 6 + .../implementation/core-consumer-associate.md | 97 +++- tests/consumer-associate.rs | 479 ++++++++++++++++ 5 files changed, 1095 insertions(+), 18 deletions(-) create mode 100644 tests/consumer-associate.rs diff --git a/src/consumer.rs b/src/consumer.rs index 613e735..3ced946 100644 --- a/src/consumer.rs +++ b/src/consumer.rs @@ -12,32 +12,51 @@ //! BIND session: the reply#2 reader (consumer.md §The BIND client, //! ADR-007) — reply#1 rides the forked client `request()`, reply#2 is //! hand-read with the forked `read_address` codec (no second codec), -//! then the pass-through data plane. The ASSOCIATE session lands with -//! its task (`core-consumer-associate`). +//! then the pass-through data plane. [`AssociateSession`] is the UDP +//! ASSOCIATE session (POC #2's shape — see its docs). //! //! Target construction rides [`ToTargetAddr`]: IP-string forms parse //! to IP targets; the domain form passes through unresolved (ADR-012 //! — resolution is the egress's job, never the protocol's). //! +//! [`AssociateSession`] is the UDP ASSOCIATE session (POC #2's shape, +//! consumer.md §The associate session, ADR-003): it opens the +//! channel, negotiates, sends ASSOCIATE, reads the reply — **receives +//! the sentinel `BND.ADDR` and never uses it for addressing** (the +//! handle is the channel). The datagram plane is the framed stage: +//! `send_datagram` frames once (`[len: u16 BE][addr + data]` — the +//! framing-once mirror, ADR-003) and `recv_datagram` decodes +//! incrementally with **reader state that persists across calls** +//! (mid-frame position + batched-datagram pending queue — a datagram +//! split across channel chunks survives a caller-driven receive +//! loop; POC #2's finding: the first cut dropped the reader state +//! per call, the 2-byte-prefix chunk test exposed it). `shutdown` +//! half-closes the stream — the mux's implicit-EOF sentinel reaches +//! the producer's stage as EOF and the association ends (association +//! lifetime = channel lifetime). +//! //! Teardown ownership (alktunnels ADR-005 posture, inherited via //! ADR-004): the session owns the adopted channel entry. On Drop it //! reaps it (`ChannelManager::teardown_channel` — sync, best-effort; //! `Drop` cannot await) and the data plane EOFs structurally (the //! mux's implicit-EOF sentinel on the dropped write half). The //! teardown-at-birth discipline covers the in-stream refusal path: -//! a refused CONNECT tears the adopted channel down before the +//! a refused ASSOCIATE tears the adopted channel down before the //! error returns — a refused session never leaves a phantom channel. //! The session does **not** implement `Clone` (compile-asserted on -//! the struct): aliasing a session would alias its teardown. +//! the struct): aliasing a session would alias its teardown and its +//! reader state. //! //! Error surface (consumer.md §Error surface): open failures are the //! typed [`SocksOpenError`] reasons (ADR-049 §4 — a refused session -//! never yields a phantom channel); a refused target is an RFC reply -//! code surfaced through [`SocksSessionError::Rfc`]. +//! never yields a phantom channel); a refused ASSOCIATE is an RFC +//! reply code surfaced through [`SocksSessionError::Rfc`]. use alkcall::channels::client::ChannelClient; use alkcall::channels::manager::ChannelManager; use alkcall::core::types::BiStream; +use bytes::Bytes; +use std::net::{IpAddr, Ipv6Addr, SocketAddr}; use tokio::io::{AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf}; use crate::error::{SocksOpenError, SocksSessionError}; @@ -45,7 +64,8 @@ use crate::params::{Socks5Params, OP_SOCKS5_OPEN, SOCKS5_ALPN}; use crate::rfc1928::client::{Config, Socks5Stream}; use crate::rfc1928::consts; use crate::rfc1928::util::target_addr::{read_address, TargetAddr, ToTargetAddr}; -use crate::rfc1928::{ReplyError, Socks5Command, SocksError}; +use crate::rfc1928::{ReplyError, Socks5Command, SocksError, UdpHeaderError}; +use crate::wire::{frame, DatagramReader, WireError}; /// A completed SOCKS5 CONNECT session on the consumer side: the /// adopted channel + the post-reply byte pipe to the target. @@ -177,6 +197,302 @@ impl Drop for Socks5ChannelSession { } } +/// The UDP ASSOCIATE consumer session (POC #2's shape, consumer.md +/// §The associate session, ADR-003): the channel is the relay handle +/// — the reply's sentinel `BND.ADDR` is received and never used for +/// addressing, the datagram plane is the framed stage on the channel +/// (`[len: u16 BE][addr + data]`, both directions), and the +/// association's lifetime is the channel's. +/// +/// The session is a handle to one channel (one session = one channel +/// = one ASSOCIATE, convention 10) and owns its reader state: the +/// incremental [`DatagramReader`] (mid-frame position) and the +/// pending-queue of batched datagrams both **persist across +/// `recv_datagram` calls** — a datagram split across channel chunks +/// survives a caller-driven receive loop (POC #2's finding, the +/// first cut dropped the reader state per call). It does **not** +/// implement `Clone`: +/// +/// ```compile_fail +/// use alksocks::consumer::AssociateSession; +/// fn assert_clone() {} +/// assert_clone::(); +/// ``` +/// +/// The inner stream is `Option` so `Drop`/`shutdown`/`close` can +/// empty the type (the alktunnels `Drop`-impl posture); a spent +/// session drops as a teardown-only no-op. +pub struct AssociateSession { + channel_id: u32, + manager: Option, + inner: Option>, + /// The incremental length-codec decoder — its mid-frame state + /// persists across `recv_datagram` calls (a datagram may be split + /// across channel chunks; the same `DatagramReader` the producer + /// stage runs). + reader: DatagramReader, + /// Datagrams decoded beyond the one returned (a chunk may batch + /// several datagrams — drained by later `recv_datagram` calls). + /// The batched datagrams are stored decoded (`(addr, payload)` — + /// the same shape `recv_datagram` returns). + pending: std::collections::VecDeque<(TargetAddr, Bytes)>, + /// The reply's BND.ADDR — received for RFC fidelity, unused for + /// addressing (wrapper-aware clients sendto the channel; ADR-003). + relay_addr: Option, +} + +impl std::fmt::Debug for AssociateSession { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("AssociateSession") + .field("channel_id", &self.channel_id) + .field("associated", &self.inner.is_some()) + .field("relay_addr", &self.relay_addr) + .field("pending_datagrams", &self.pending.len()) + .finish() + } +} + +impl AssociateSession { + /// ASSOCIATE through a produced SOCKS5 resource: opens the + /// channel (`open_channel(OP_SOCKS5_OPEN, {}, SOCKS5_ALPN)` — the + /// trivial params, ADR-001), drives the no-auth handshake and the + /// UDP ASSOCIATE request in-stream over the channel's `BiStream`, + /// and reads the reply. The DST.ADDR of the ASSOCIATE command + /// mirrors the vanilla shape — the unspecified address (`[::]:0`), + /// as the forked `Socks5Datagram` sends. The reply's BND.ADDR is + /// the producer's sentinel (`127.0.0.1:0` on the in-band path) — + /// captured as [`Self::relay_addr`], never used for addressing + /// (ADR-003: the handle is the channel). + /// + /// Error surface: an open failure is [`SocksSessionError::Open`] + /// (the typed ADR-049 reasons; no channel was ever returned); a + /// refused ASSOCIATE (the missing-egress `CommandNotSupported` + /// 0x07 or a factory failure's mapped code) surfaces the RFC + /// reply code through [`SocksSessionError::Rfc`] — and the + /// adopted channel is torn down before the error returns (no + /// phantom channel). + pub async fn associate(client: &ChannelClient) -> Result { + let (channel_id, send, recv) = client + .open_channel(OP_SOCKS5_OPEN, Socks5Params::as_json(), SOCKS5_ALPN) + .await + .map_err(SocksOpenError::from)?; + + let bidi = BiStream::from_joined(recv, send); + let negotiated = async { + let mut inner = Socks5Stream::use_stream(bidi, None, Config::default()).await?; + let relay_addr = inner + .request( + Socks5Command::UDPAssociate, + TargetAddr::Ip(SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 0)), + ) + .await?; + Result::<(TargetAddr, Socks5Stream), SocksSessionError>::Ok(( + relay_addr, inner, + )) + }; + + match negotiated.await { + Ok((relay_addr, inner)) => Ok(Self { + channel_id, + manager: Some(client.manager().clone()), + inner: Some(inner), + reader: DatagramReader::new(), + pending: std::collections::VecDeque::new(), + relay_addr: Some(relay_addr), + }), + Err(err) => { + let _ = client.manager().teardown_channel(channel_id); + Err(err) + } + } + } + + /// The adopted channel's id (observability + correlation; one + /// session = one channel). + pub fn channel_id(&self) -> u32 { + self.channel_id + } + + /// The reply's sentinel `BND.ADDR` (RFC fidelity: the relay + /// address the vanilla shape would sendto; the wrapper-aware + /// session never uses it — the handle is the channel, ADR-003). + /// `None` after a taken stream (the spent-session posture). + pub fn relay_addr(&self) -> Option { + self.relay_addr.clone() + } + + /// Send one datagram to `dst`: the RFC address encoding + the + /// payload, length-framed **exactly once** by the wire codec + /// ([`frame`] — the framing-once mirror, ADR-003) and written to + /// the channel. A zero-length payload is a legal empty datagram + /// (the F-2 invariant at payload level). Rejects + /// [`SocksSessionError::Rfc`] ([`WireError`]) at frame time — + /// the joint bound (`len(addr) + len(payload) > 65535`, the u16 + /// length field would wrap) and the address-encoding refusals. + pub async fn send_datagram( + &mut self, + dst: impl ToTargetAddr, + payload: &[u8], + ) -> Result<(), SocksSessionError> { + let dst: TargetAddr = dst.to_target_addr()?; + let framed = frame(&dst, payload).map_err(wire_error)?; + let Some(inner) = self.inner.as_mut() else { + return Err(SocksSessionError::Io(std::io::Error::new( + std::io::ErrorKind::NotConnected, + "the association's stream is spent", + ))); + }; + let socket = inner.get_socket_mut(); + socket.write_all(&framed).await?; + socket.flush().await?; + Ok(()) + } + + /// Receive the next datagram: decode incrementally from the + /// channel's read half — `(source_addr, payload)` per datagram, + /// `Ok(None)` on channel EOF (the association's end — the + /// association lifetime is the channel's, ADR-003). + /// + /// **The reader state persists across calls** (POC #2's finding, + /// ADR-003): the incremental [`DatagramReader`]'s mid-frame + /// position survives a caller-driven receive loop — a datagram + /// split across channel chunks reassembles across calls — and + /// frames decoded beyond the one returned (a chunk may batch + /// several datagrams) queue in `pending` for later calls. + /// + /// Fail-loud shapes (never a silent partial datagram): a stream + /// end mid-frame is the truncated-datagram condition (an + /// `UnexpectedEof` I/O error through [`SocksSessionError::Rfc`]); + /// a frame the codec refuses (garbage ATYP — any post-reply + /// control byte) errors loud through [`WireError`]'s + /// `UdpHeaderError` mapping and ends the caller's loop. + pub async fn recv_datagram( + &mut self, + ) -> Result, SocksSessionError> { + // The batched-datagram pending queue first (a chunk may have + // batched several frames — drain in order). + if let Some(dg) = self.pending.pop_front() { + return Ok(Some(dg)); + } + + let mut chunk = vec![0u8; 16 * 1024]; + let Some(inner) = self.inner.as_mut() else { + return Ok(None); + }; + let socket = inner.get_socket_mut(); + loop { + let n = socket.read(&mut chunk).await?; + if n == 0 { + // Channel EOF = association end (the clean exit) — + // fail-loud if a frame was in flight (never a silent + // partial datagram; the reader state is the teardown + // diagnostic). + if self.reader.is_mid_frame() { + return Err(SocksSessionError::Rfc(SocksError::Io(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "association ended mid-datagram (fail-loud)", + )))); + } + return Ok(None); + } + let mut dgs = self.reader.feed(&chunk[..n]).into_iter(); + if let Some(first) = dgs.next() { + for res in dgs { + match res { + Ok(pair) => self.pending.push_back(pair), + Err(err) => return Err(wire_error(err)), + } + } + match first { + Ok(pair) => return Ok(Some(pair)), + Err(err) => return Err(wire_error(err)), + } + } + } + } + + /// The underlying stream as split read/write halves (the front + /// door's bridge seam: a raw pipe to a vanilla client over the + /// framed stage — the door's RFC header codec owns the frames at + /// the boundary, ADR-003). `None` after the stream was taken (the + /// spent-session posture). + pub fn split(&mut self) -> Option> { + let inner = self.inner.as_mut()?; + Some(tokio::io::split(inner)) + } + + /// Take the underlying [`Socks5Stream`] (advanced use — + /// the forked client type's socket accessors, owned-half pumps; + /// the POC's `raw_mut` seam, widened). The session rides along + /// still owning the channel's teardown: drop the pair (or + /// `shutdown`/`close` the session) when done. + pub fn into_inner(mut self) -> TakenAssociateInner { + let inner = self.inner.take(); + TakenAssociateInner { + session: self, + inner: inner.expect("into_inner on a spent session"), + } + } + + /// Half-close the stream (the association's graceful end): the + /// mux's implicit-EOF sentinel reaches the producer's stage as + /// EOF and the association ends (association lifetime = channel + /// lifetime, ADR-003). Returns whether the adopted channel entry + /// existed (the reap result; the session is spent either way). + pub async fn shutdown(mut self) -> Result { + let reaped = if let Some(inner) = self.inner.as_mut() { + let socket = inner.get_socket_mut(); + socket.shutdown().await?; + self.inner.take(); + self.reap() + } else { + self.reap() + }; + Ok(reaped) + } + + /// Tear the association down gracefully (the CONNECT session's + /// `close` posture): half-close the stream (the EOF sentinel + /// reaches the producer's stage), then reap the adopted channel + /// entry. Returns whether the entry existed. + pub async fn close(mut self) -> bool { + if let Some(inner) = self.inner.as_mut() { + let _ = inner.shutdown().await; + } + self.inner.take(); + self.reap() + } + + fn reap(&mut self) -> bool { + self.manager + .take() + .map(|m| m.teardown_channel(self.channel_id).is_ok()) + .unwrap_or(false) + } +} + +impl Drop for AssociateSession { + /// Sync-reap the adopted channel entry (`teardown_channel` is + /// sync — best-effort; `Drop` cannot await) and drop the pipe + /// (the mux's implicit-EOF sentinel EOFs the producer's stage — + /// the association ends with the session's lifetime). Dropping + /// without `shutdown`/`close` never leaks the channel entry. + fn drop(&mut self) { + self.inner.take(); + let _ = self.reap(); + } +} + +/// The owned stream [`AssociateSession::into_inner`] yields, paired +/// with the session that still owns the channel's teardown. +pub struct TakenAssociateInner { + /// The teardown owner — reap via `shutdown`/`close`/`Drop`. + pub session: AssociateSession, + /// The framed datagram stage's stream (the forked client type + /// over the channel's `BiStream`). + pub inner: Socks5Stream, +} + /// The owned pipe [`Socks5ChannelSession::into_inner`] yields, paired /// with the session that still owns the channel's teardown. pub struct TakenInner { @@ -194,7 +510,6 @@ pub type SplitPipe<'s> = ( ReadHalf<&'s mut Socks5Stream>, WriteHalf<&'s mut Socks5Stream>, ); - /// A completed BIND reply#1 (consumer.md §The BIND client, ADR-007's /// client-side phase reading): the channel is open, reply#1 carried /// BND.ADDR — the advertised address the app server dials (the one @@ -464,6 +779,28 @@ impl Drop for BindSession { } } +/// Map a datagram-frame decode error ([`WireError`]) into the RFC +/// vocabulary the session surface carries: the address-encoding +/// refusal is the forked [`AddrError`] class (the same sub-layout +/// the forked codec validates — one vocabulary), the protocol-garbage +/// shapes (bad ATYP / a short frame / non-UTF-8 domain / a `len == 0` +/// frame) surface as the RFC UDP-header error class +/// ([`UdpHeaderError`]) — the post-reply-garbage semantic, the +/// consumer-side mirror of the producer stage's +/// `UndecodableFrame` exit. The oversize family is unreachable on +/// the receive path (the length codec's frame-time bound makes a +/// declared length undecodable only via the garbage shapes) — it +/// maps through the UDP-header class too (fail-loud either way). +fn wire_error(err: WireError) -> SocksSessionError { + match err { + WireError::Address(addr) => SocksError::AddrError(addr).into(), + other => SocksError::UdpHeaderError(UdpHeaderError::ReadingError(std::io::Error::other( + other.to_string(), + ))) + .into(), + } +} + /// Read reply#2 off the raw stream (the RFC's second reply: VER, REP, /// RSV, ATYP, addr, port — the POC #5 `read_reply2` sketch): the 4-byte /// head, the version pin, the reply code mapped through the forked @@ -494,6 +831,7 @@ where mod tests { use super::*; use alkcall::channels::client::ChannelOpenError; + use futures::future::FutureExt as _; fn assert_send() {} @@ -534,4 +872,176 @@ mod tests { let session: SocksSessionError = open.into(); assert!(matches!(session, SocksSessionError::Open(_))); } + + #[test] + fn associate_session_shape_is_send_and_not_clone() { + assert_send::(); + assert_send::(); + } + + #[test] + fn associate_debug_impl_names_the_channel_and_liveness() { + let session = AssociateSession { + channel_id: 7, + manager: None, + inner: None, + reader: DatagramReader::new(), + pending: Default::default(), + relay_addr: None, + }; + let rendered = format!("{session:?}"); + assert!(rendered.contains("channel_id: 7"), "{rendered}"); + assert!(rendered.contains("associated: false"), "{rendered}"); + } + + #[test] + fn spent_associate_reap_is_a_no_op() { + let mut session = AssociateSession { + channel_id: 9, + manager: None, + inner: None, + reader: DatagramReader::new(), + pending: Default::default(), + relay_addr: None, + }; + assert!(!session.reap(), "no manager — nothing reaped"); + assert!(session.split().is_none(), "no stream after teardown"); + assert!(session.relay_addr().is_none()); + } + + #[tokio::test] + async fn recv_datagram_persists_reader_state_across_calls() { + use crate::wire::frame as wire_frame; + use tokio::io::AsyncWriteExt as _; + + // The frame-source side: two frames, the first one split + // across two writes (the 2-byte-prefix chunk shape — the POC + // finding's trigger) and the second batched into the same + // read batch. The session's reader state must survive the + // call boundary: the first call returns the first datagram + // (reassembled across reads), the second drains the pending + // queue. + let (mut writer, reader) = tokio::io::duplex(1024); + let addr = TargetAddr::Ip("10.0.0.1:53".parse().unwrap()); + let wire = { + let mut v = wire_frame(&addr, b"0123456789").unwrap().to_vec(); + v.extend_from_slice(&wire_frame(&addr, b"second").unwrap()); + v + }; + writer.write_all(&wire[..5]).await.unwrap(); + writer.write_all(&wire[5..]).await.unwrap(); + drop(writer); + + let mut session = build_test_session(reader); + + let (src, first) = session + .recv_datagram() + .await + .expect("recv") + .expect("a datagram"); + assert_eq!(src, addr); + assert_eq!(&first[..], b"0123456789"); + + // The second frame was batched in the same read batch: the + // pending queue carries it to the next call. + let (src, second) = session + .recv_datagram() + .await + .expect("recv") + .expect("a datagram"); + assert_eq!(src, addr); + assert_eq!(&second[..], b"second"); + + // EOF: both frames consumed, the queue empty — `None`. + assert!(session.recv_datagram().await.expect("eof").is_none()); + } + + #[tokio::test] + async fn recv_datagram_fails_loud_when_stream_ends_mid_frame() { + use tokio::io::AsyncWriteExt as _; + + let (mut writer, reader) = tokio::io::duplex(1024); + let addr = TargetAddr::Ip("10.0.0.1:53".parse().unwrap()); + let framed = crate::wire::frame(&addr, b"payload").unwrap(); + writer.write_all(&framed[..4]).await.unwrap(); + drop(writer); + + let mut session = build_test_session(reader); + let err = session + .recv_datagram() + .await + .expect_err("a stream end mid-frame is the truncated-datagram condition"); + assert!( + matches!(err, SocksSessionError::Rfc(SocksError::Io(_))), + "fail-loud, not a silent partial: {err:?}" + ); + } + + #[tokio::test] + async fn recv_datagram_fails_loud_on_an_undecodable_frame() { + use tokio::io::AsyncWriteExt as _; + + let (mut writer, reader) = tokio::io::duplex(1024); + // Hand-craft a frame whose inner bytes start with an + // unassigned ATYP (the post-reply-garbage semantic). + let inner_bytes = vec![0x7Fu8, 1, 2, 3]; + let mut garbage = (inner_bytes.len() as u16).to_be_bytes().to_vec(); + garbage.extend_from_slice(&inner_bytes); + writer.write_all(&garbage).await.unwrap(); + drop(writer); + + let mut session = build_test_session(reader); + let err = session + .recv_datagram() + .await + .expect_err("garbage ATYP is loud, never a silent drop"); + assert!(matches!(err, SocksSessionError::Rfc(_)), "{err:?}"); + } + + #[test] + fn wire_error_maps_the_decode_garbage_shapes() { + // The address-encoding refusal rides the forked AddrError + // class; the garbage shapes ride the UDP-header class. + let addr_err = + crate::wire::frame(&TargetAddr::Domain("a".repeat(256), 80), b"").unwrap_err(); + let mapped = wire_error(addr_err); + assert!(matches!( + mapped, + SocksSessionError::Rfc(SocksError::AddrError(_)) + )); + + let garbage = crate::wire::DatagramReader::new() + .feed(&[0x00, 0x04, 0x7f, 1, 2, 3]) + .into_iter() + .next() + .unwrap() + .unwrap_err(); + let mapped = wire_error(garbage); + assert!(matches!( + mapped, + SocksSessionError::Rfc(SocksError::UdpHeaderError(_)) + )); + } + + /// A session over a raw duplex (the reader internals' test + /// harness — the handshake-free path: `skip_auth` bypasses the + /// greeting, no reply is expected, the datagram stage rides + /// raw). + fn build_test_session(session_end: tokio::io::DuplexStream) -> AssociateSession { + let mut config = Config::default(); + config.set_skip_auth(true); + let (read, write) = tokio::io::split(session_end); + let inner = Socks5Stream::use_stream(BiStream::from_joined(read, write), None, config) + .now_or_never() + .expect("skip_auth's use_stream is instantly ready") + .expect("skip_auth's use_stream succeeds"); + AssociateSession { + channel_id: 1, + manager: None, + inner: Some(inner), + reader: DatagramReader::new(), + pending: Default::default(), + relay_addr: None, + } + } } diff --git a/src/lib.rs b/src/lib.rs index 3824b36..90f9f26 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -29,7 +29,10 @@ pub mod params; pub mod producer; pub mod wire; -pub use consumer::{BindDataPlane, BindReply1, BindSession, Socks5ChannelSession}; +pub use consumer::{ + AssociateSession, BindDataPlane, BindReply1, BindSession, Socks5ChannelSession, + TakenAssociateInner, +}; pub use egress::{ egress_factory, BindAccept, BindAcceptError, BindAcceptFactory, BindAcceptFactoryError, BindTargetHandle, DialFn, Egress, EgressError, EgressFactory, SocksDialError, TargetHandle, diff --git a/src/rfc1928/util/target_addr.rs b/src/rfc1928/util/target_addr.rs index bbd148b..e88cac8 100644 --- a/src/rfc1928/util/target_addr.rs +++ b/src/rfc1928/util/target_addr.rs @@ -200,6 +200,12 @@ impl ToTargetAddr for TargetAddr { } } +impl ToTargetAddr for &TargetAddr { + fn to_target_addr(&self) -> io::Result { + Ok((*self).clone()) + } +} + #[derive(Debug)] pub enum Addr { V4([u8; 4]), diff --git a/tasks/implementation/core-consumer-associate.md b/tasks/implementation/core-consumer-associate.md index 729f8eb..9c4eabd 100644 --- a/tasks/implementation/core-consumer-associate.md +++ b/tasks/implementation/core-consumer-associate.md @@ -1,7 +1,7 @@ --- id: implementation/core-consumer-associate name: "Core — consumer ASSOCIATE: AssociateSession + e2e tests" -status: pending +status: completed depends_on: [implementation/core-consumer-connect, implementation/core-producer-associate] scope: moderate risk: medium @@ -51,16 +51,16 @@ adapted to the crate's wire codec): ## Acceptance Criteria -- [ ] `AssociateSession` with `associate`/`send_datagram`/ +- [x] `AssociateSession` with `associate`/`send_datagram`/ `recv_datagram`/`shutdown` per the POC shape -- [ ] Reader state persists across `recv_datagram` calls (the +- [x] Reader state persists across `recv_datagram` calls (the split-datagram test) -- [ ] Framing-once mirror (send frames exactly once; recv decodes +- [x] Framing-once mirror (send frames exactly once; recv decodes incrementally) -- [ ] Teardown ownership on Drop; no `Clone` (compile-asserted) -- [ ] E2E tests through the harness (round-trips, shutdown, Drop +- [x] Teardown ownership on Drop; no `Clone` (compile-asserted) +- [x] E2E tests through the harness (round-trips, shutdown, Drop teardown, regression guard) -- [ ] `cargo test` green; clippy/fmt clean; both wasm checks clean +- [x] `cargo test` green; clippy/fmt clean; both wasm checks clean ## References @@ -72,8 +72,87 @@ adapted to the crate's wire codec): ## Notes -> To be filled by implementation agent +- **`send_datagram` takes `impl ToTargetAddr`** (the CONNECT + session's target-construction posture — ADR-012: IP-string forms + parse, domains pass through unresolved), not the POC's + `&TargetAddr`: the same entry shape as `connect`. A + `ToTargetAddr for &TargetAddr` impl was added to the forked + address codec (our owned code, ADR-013) so callers can reuse a + `TargetAddr` value across sends. +- **Teardown ownership matches the CONNECT/BIND sessions** (the + crate's posture, strengthened from the POC's "close/join"): the + session owns the adopted channel entry — Drop sync-reaps + (`teardown_channel`, best-effort; `Drop` cannot await) + the data + plane EOFs structurally (the mux's implicit-EOF sentinel); + `shutdown()` half-closes (the graceful path, returning the reap + result); `close()` is the `shutdown`+reap alias-shaped variant + kept for the connect/bind symmetry; teardown-at-birth covers the + in-stream refusal path (a refused ASSOCIATE never leaves a + phantom channel). +- **`recv_datagram` returns `Result>`** + (the POC's shape): `Ok(None)` = channel EOF (the association's + end); a stream end mid-frame is fail-loud (an `UnexpectedEof` + I/O error through `SocksSessionError::Rfc` — never a silent + partial datagram); an undecodable frame (garbage ATYP) errors + loud through the wire codec's `WireError` → RFC mapping + (`AddrError` class for address-encoding refusals, + `UdpHeaderError` for the garbage shapes — the consumer-side + mirror of the producer stage's `UndecodableFrame` exit). +- **The pending queue stores decoded datagrams** (`(TargetAddr, + Bytes)`) — the POC stored pre-decode inner bytes; decoded storage + keeps the delivery shape uniform (decode once per frame either + way; no wire.rs change needed — `decode_inner` stays private). +- **Unit tests ride a duplex-backed session** (`skip_auth`'s + handshake-free `use_stream` path over a `BiStream::from_joined` + duplex — no channels needed) for the reader-internals probes: + reader-state persistence (the 2-byte-prefix chunk shape, the POC + finding's trigger), mid-frame EOF fail-loud, garbage-ATYP fail- + loud, and the wire-error mapping table. +- **The e2e split-datagram test uses a 60 KiB payload** (60007 + inner bytes — legal under the u16 joint bound, well over the 16 + KB read buffer): first → big → third, in order, with the big + datagram reassembling across the call boundary and the third + drained through the pending queue — both halves of the + persistence invariant pinned end to end. +- **No harness changes**: the consumer-side topology's + `wire_consumer_side_with` + the producer-associate task's + `echo_egress`/`refusing_egress` closures cover every test. +- The lib.rs re-exports gained `AssociateSession` + + `TakenAssociateInner`; the forked `target_addr.rs` gained the + `&TargetAddr` `ToTargetAddr` impl (ergonomics; owned code, + ADR-013). ## Summary -> To be filled on completion \ No newline at end of file +Implemented the UDP ASSOCIATE consumer session in `src/consumer.rs` +(POC #2's shape, adapted to the crate's teardown-owning session +posture): `AssociateSession::associate` (open → no-auth → ASSOCIATE +→ reply, capturing the sentinel BND.ADDR as `relay_addr()` — +received, never used for addressing; the ASSOCIATE DST.ADDR is the +vanilla-mirror unspecified address), `send_datagram` (frames once +via the crate's wire `frame()` — the framing-once mirror, the joint +bound enforced at frame time), `recv_datagram` (the incremental +`DatagramReader` + batched-datagram pending queue, **state that +persists across calls** — the POC's reader-state finding, with the +fail-loud shapes: mid-frame EOF and undecodable frames), `shutdown` +(the half-close; the mux's implicit-EOF sentinel reaches the +producer's stage as EOF — association lifetime = channel lifetime), +plus `split`/`into_inner` accessors and the CONNECT sessions' +teardown ownership (Drop sync-reap + structural EOF, `close()` +graceful, teardown-at-birth on in-stream refusals, no `Clone` — +`compile_fail`-asserted). Typed error surface throughout: open +failures are `SocksSessionError::Open` (ADR-049 reasons); the +missing-egress 0x07 and factory-failure codes surface from +`associate` as `Rfc`. E2E suite (`tests/consumer-associate.rs`, 8 +tests over the harness): the round-trip (sentinel relay address → +multiple in-order datagrams, one association), the split-datagram +persistence test (60 KiB datagram across the call boundary + +pending-queue drain), empty-payload round-trip (F-2, never EOF), +shutdown → fresh associate succeeds immediately (the POC's test 4), +Drop teardown + connection health, the missing-egress refusal +(typed 0x07, no phantom channel), the egress-factory failure +mapping, and the CONNECT+ASSOCIATE regression guard on one +registration (interleaved, both serving). Verification: `cargo +test` 305 (incl. the 3 doctests — the compile-fail assertions ride +among them), clippy `-D warnings` (all targets), fmt, `cargo doc +--no-deps`, and both wasm checks clean. \ No newline at end of file diff --git a/tests/consumer-associate.rs b/tests/consumer-associate.rs new file mode 100644 index 0000000..e7e3200 --- /dev/null +++ b/tests/consumer-associate.rs @@ -0,0 +1,479 @@ +//! The consumer UDP ASSOCIATE integration tests — `AssociateSession` +//! end to end through the harness's consumer-side topology (the +//! producer's real command-dispatch path over channels; the framed +//! datagram stage, ADR-003, driven by the wrapper-aware client, POC +//! #2's shape). +//! +//! Covered: associate → sentinel reply (received, never used for +//! addressing) → send/recv round-trips (multiple in-order datagrams, +//! one association); the split-datagram persistence test (reader +//! state persists across `recv_datagram` calls — the POC's +//! 2-byte-prefix chunk finding); empty-payload datagram round-trip +//! (F-2, never EOF); shutdown → the producer's stage ends and a +//! fresh associate succeeds immediately; Drop teardown (drop +//! mid-stream → the producer's stage sees EOF, the connection stays +//! healthy); the ASSOCIATE refusal surfaced as a typed error with no +//! phantom channel; CONNECT + ASSOCIATE on the same registration +//! (the regression guard extended). + +mod harness; + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +use alksocks::consumer::{AssociateSession, Socks5ChannelSession}; +use alksocks::egress::EgressError; +use alksocks::error::SocksSessionError; +use alksocks::rfc1928::consts; +use alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr}; +use alksocks::SocksError as SocksErrorKind; + +use harness::{echo_dial, echo_egress, refusing_egress, wire_consumer_side_with}; + +/// The no-leaked-channels probe (the connect/bind suites' shape, +/// re-pinned for the ASSOCIATE session): both managers hold only +/// channel 0 — the producer's entry reaps by its open-op wrapper; +/// the probe polls past the reap race. +async fn no_leaked_channels(topo: &harness::ConsumerSideTopology) { + let deadline = std::time::Duration::from_secs(5); + let consumer_ids = topo.consumer.manager().channel_ids(); + assert!( + consumer_ids.iter().all(|&id| id == 0), + "consumer leaked channel entries: {consumer_ids:?}" + ); + tokio::time::timeout(deadline, async { + loop { + let producer_ids = topo.producer_manager.channel_ids(); + if producer_ids.iter().all(|&id| id == 0) { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .expect("the producer's handler-exit teardown reaps the entry"); +} + +/// The happy path: `associate` → the reply's sentinel `BND.ADDR` is +/// received (RFC fidelity) and never used for addressing → send/recv +/// round-trips through the echo egress — multiple datagrams, in +/// order, on one association; the establisher witnessed the opener +/// identity (CF-006). +#[tokio::test] +async fn associate_round_trips_multiple_datagrams_one_association() { + let topo = wire_consumer_side_with( + echo_dial(), + Some(echo_egress()), + None, + Some(harness::consumer_identity()), + ) + .await; + + let mut assoc = AssociateSession::associate(topo.consumer.as_ref()) + .await + .expect("the association associates"); + + // The reply's BND.ADDR is the producer's sentinel (127.0.0.1:0 — + // ADR-003): received for RFC fidelity, never used for addressing. + let relay = assoc + .relay_addr() + .expect("the reply carried BND.ADDR") + .into_string_and_port(); + assert_eq!(relay.0, "127.0.0.1", "the in-band sentinel BND.ADDR"); + assert_eq!(relay.1, 0); + + let target = TargetAddr::Ip("10.0.0.1:53".parse().unwrap()); + for i in 0..5u8 { + let payload = vec![i; 10]; + assoc + .send_datagram(&target, &payload) + .await + .expect("send datagram"); + } + for i in 0..5u8 { + let (src, payload) = + tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram()) + .await + .expect("recv within timeout") + .expect("a datagram (not eof)") + .expect("a datagram"); + assert_eq!(src, target, "the echo reports the destination"); + assert_eq!(payload, bytes::Bytes::from(vec![i; 10]), "in-order echo"); + } + + // The association is alive: another datagram round-trips. + assoc.send_datagram(&target, b"alive").await.expect("send"); + let (_, payload) = + tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram()) + .await + .expect("recv timed out") + .expect("a datagram (not eof)") + .expect("a datagram"); + assert_eq!(&payload[..], b"alive"); + + assert!( + topo.consumer.manager().has_channel(assoc.channel_id()), + "the adopted channel entry exists while associated" + ); + + let reaped = assoc.shutdown().await.expect("half-close"); + assert!(reaped, "the adopted channel entry existed"); + no_leaked_channels(&topo).await; +} + +/// The split-datagram persistence test (POC #2's finding — the +/// reader state must persist across `recv_datagram` calls): a large +/// datagram's frame (under the u16 joint bound but well over the 16 +/// KB read buffer) spans multiple channel reads AND multiple +/// `recv_datagram` calls — the first call returns the earlier small +/// datagram while the large one is mid-flight, the next call +/// reassembles it across its internal reads — and a trailing small +/// datagram batched in the same read batch drains through the +/// pending queue. +#[tokio::test] +async fn recv_datagram_state_persists_across_calls() { + let topo = wire_consumer_side_with(echo_dial(), Some(echo_egress()), None, None).await; + + let mut assoc = AssociateSession::associate(topo.consumer.as_ref()) + .await + .expect("the association associates"); + let target = TargetAddr::Ip("10.0.0.2:53".parse().unwrap()); + + // The order tells the story: "first" completes early; the 60 KiB + // datagram (60007 inner bytes — legal, but over the 16 KB read + // buffer) is mid-flight when the first call returns; "third" + // batches after it. + assoc + .send_datagram(&target, b"first") + .await + .expect("send the first datagram"); + let big = vec![7u8; 60 * 1024]; + assoc + .send_datagram(&target, &big) + .await + .expect("send the big datagram"); + assoc + .send_datagram(&target, b"third") + .await + .expect("send the third datagram"); + + let (src, first) = + tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram()) + .await + .expect("recv timed out") + .expect("a datagram (not eof)") + .expect("a datagram"); + assert_eq!(src, target); + assert_eq!(&first[..], b"first", "the first datagram, in order"); + + // The large datagram reassembles across the call boundary (its + // mid-frame state persisted into this call) — the POC finding's + // shape. + let (src, second) = + tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram()) + .await + .expect("recv timed out") + .expect("a datagram (not eof)") + .expect("a datagram"); + assert_eq!(src, target); + assert_eq!(&second[..], &big[..], "the split datagram reassembled"); + + // The trailing datagram rode the same read batches: the pending + // queue carries it to this call. + let (_, third) = + tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram()) + .await + .expect("recv timed out") + .expect("a datagram (not eof)") + .expect("a datagram"); + assert_eq!(&third[..], b"third", "the batched datagram drained"); + + let reaped = assoc.shutdown().await.expect("half-close"); + assert!(reaped); + no_leaked_channels(&topo).await; +} + +/// An empty-payload datagram is a legal datagram — never EOF (the +/// F-2 invariant, end to end through the echo egress), and the +/// association survives it (pre → empty → post, in order). +#[tokio::test] +async fn empty_payload_datagram_round_trips_and_survives() { + let topo = wire_consumer_side_with(echo_dial(), Some(echo_egress()), None, None).await; + + let mut assoc = AssociateSession::associate(topo.consumer.as_ref()) + .await + .expect("the association associates"); + let target = TargetAddr::Ip("10.0.0.3:53".parse().unwrap()); + + assoc + .send_datagram(&target, b"pre") + .await + .expect("send pre"); + assoc + .send_datagram(&target, b"") + .await + .expect("an empty payload is a legal datagram"); + assoc + .send_datagram(&target, b"post") + .await + .expect("send post"); + + let (_, pre) = tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram()) + .await + .expect("recv timed out") + .expect("a datagram (not eof)") + .expect("a datagram"); + assert_eq!(&pre[..], b"pre"); + + let (src, empty) = + tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram()) + .await + .expect("recv timed out") + .expect("a datagram (not eof)") + .expect("a datagram"); + assert_eq!(src, target); + assert!( + empty.is_empty(), + "the empty-payload datagram round-trips (never EOF)" + ); + + let (_, post) = tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram()) + .await + .expect("recv timed out") + .expect("a datagram (not eof)") + .expect("a datagram"); + assert_eq!(&post[..], b"post"); + + let reaped = assoc.shutdown().await.expect("half-close"); + assert!(reaped); + no_leaked_channels(&topo).await; +} + +/// Shutdown half-closes the stream: the mux's implicit-EOF sentinel +/// reaches the producer's stage as EOF and the association ends +/// (association lifetime = channel lifetime) — and a fresh associate +/// succeeds immediately after (the POC's test 4, from the consumer +/// side; the same registration is unpoisoned). +#[tokio::test] +async fn shutdown_ends_the_association_and_a_fresh_associate_succeeds() { + let topo = wire_consumer_side_with(echo_dial(), Some(echo_egress()), None, None).await; + + let mut assoc = AssociateSession::associate(topo.consumer.as_ref()) + .await + .expect("the association associates"); + let channel_id = assoc.channel_id(); + let target = TargetAddr::Ip("10.0.0.4:53".parse().unwrap()); + assoc + .send_datagram(&target, b"before") + .await + .expect("send before shutdown"); + let (_, echoed) = + tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram()) + .await + .expect("recv timed out") + .expect("a datagram (not eof)") + .expect("a datagram"); + assert_eq!(&echoed[..], b"before"); + + // Half-close: the producer's stage exits on the EOF sentinel. + assoc.shutdown().await.expect("shutdown"); + assert!( + !topo.consumer.manager().has_channel(channel_id), + "the adopted channel entry was reaped" + ); + + // The connection is healthy: a fresh associate round-trips. + let mut fresh = AssociateSession::associate(topo.consumer.as_ref()) + .await + .expect("fresh associate after the shutdown"); + fresh + .send_datagram(&target, b"fresh") + .await + .expect("send after fresh associate"); + let (_, echoed) = + tokio::time::timeout(std::time::Duration::from_secs(10), fresh.recv_datagram()) + .await + .expect("recv timed out") + .expect("a datagram (not eof)") + .expect("a datagram"); + assert_eq!(&echoed[..], b"fresh"); + let reaped = fresh.shutdown().await.expect("shutdown"); + assert!(reaped); + no_leaked_channels(&topo).await; +} + +/// Drop teardown: dropping a live association mid-stream reaps the +/// adopted channel entry (no leak), the producer's stage sees the +/// structural EOF (the mux's implicit-EOF sentinel on the dropped +/// write half), and the connection stays healthy — a fresh associate +/// succeeds immediately after. +#[tokio::test] +async fn drop_teardown_reaps_and_connection_stays_healthy() { + let topo = wire_consumer_side_with(echo_dial(), Some(echo_egress()), None, None).await; + + let mut assoc = AssociateSession::associate(topo.consumer.as_ref()) + .await + .expect("the association associates"); + let channel_id = assoc.channel_id(); + assoc + .send_datagram(TargetAddr::Ip("10.0.0.5:53".parse().unwrap()), b"mid") + .await + .expect("send before the drop"); + let (_, echoed) = + tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram()) + .await + .expect("recv timed out") + .expect("a datagram (not eof)") + .expect("a datagram"); + assert_eq!(&echoed[..], b"mid"); + + // Drop mid-stream (the teardown-ownership posture). + drop(assoc); + assert!( + !topo.consumer.manager().has_channel(channel_id), + "the adopted channel entry was reaped on Drop" + ); + no_leaked_channels(&topo).await; + + // The connection is healthy: a fresh associate round-trips. + let mut fresh = AssociateSession::associate(topo.consumer.as_ref()) + .await + .expect("fresh associate after the drop"); + let target = TargetAddr::Ip("10.0.0.5:53".parse().unwrap()); + fresh + .send_datagram(&target, b"again") + .await + .expect("send after the drop"); + let (_, echoed) = + tokio::time::timeout(std::time::Duration::from_secs(10), fresh.recv_datagram()) + .await + .expect("recv timed out") + .expect("a datagram (not eof)") + .expect("a datagram"); + assert_eq!(&echoed[..], b"again"); + fresh.shutdown().await.expect("shutdown"); + no_leaked_channels(&topo).await; +} + +/// A missing egress seam: the ASSOCIATE refusal surfaces the RFC +/// reply code through the session type (`CommandNotSupported` 0x07 — +/// ADR-008's missing-callback class), and no phantom channel remains +/// on either side (teardown-at-birth). +#[tokio::test] +async fn missing_egress_refusal_surfaces_rfc_reply_code_no_phantom_channel() { + let topo = wire_consumer_side_with(echo_dial(), None, None, None).await; + + let err = AssociateSession::associate(topo.consumer.as_ref()) + .await + .expect_err("no egress wired — the ASSOCIATE must refuse"); + + match &err { + SocksSessionError::Rfc(SocksErrorKind::ReplyError(code)) => { + assert_eq!(code.as_u8(), consts::SOCKS5_REPLY_COMMAND_NOT_SUPPORTED); + } + other => panic!("expected an Rfc reply-code refusal, got {other:?}"), + } + + no_leaked_channels(&topo).await; + + // The connection is healthy: wiring the egress serves ASSOCIATE. + let topo2 = wire_consumer_side_with(echo_dial(), Some(echo_egress()), None, None).await; + let mut assoc = AssociateSession::associate(topo2.consumer.as_ref()) + .await + .expect("fresh associate after the refusal"); + assoc + .send_datagram(TargetAddr::Ip("10.0.0.6:53".parse().unwrap()), b"fresh") + .await + .expect("send"); + let (_, echoed) = + tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram()) + .await + .expect("recv timed out") + .expect("a datagram (not eof)") + .expect("a datagram"); + assert_eq!(&echoed[..], b"fresh"); + assoc.shutdown().await.expect("shutdown"); + no_leaked_channels(&topo2).await; +} + +/// An egress-factory failure: the mapped RFC reply code surfaces from +/// `associate` (reply#1's code — the factory refused before the +/// reply), and no phantom channel remains. +#[tokio::test] +async fn egress_factory_refusal_surfaces_the_mapped_reply_code() { + let topo = wire_consumer_side_with( + echo_dial(), + Some(refusing_egress(EgressError::Unreachable( + "no egress".into(), + ))), + None, + None, + ) + .await; + + let err = AssociateSession::associate(topo.consumer.as_ref()) + .await + .expect_err("the egress factory refused"); + match &err { + SocksSessionError::Rfc(SocksErrorKind::ReplyError(code)) => { + assert_eq!(code.as_u8(), consts::SOCKS5_REPLY_NETWORK_UNREACHABLE); + } + other => panic!("expected an Rfc reply-code refusal, got {other:?}"), + } + no_leaked_channels(&topo).await; +} + +/// CONNECT + ASSOCIATE on the same registration (the regression +/// guard extended): one ALPN, both commands, interleaved — an +/// association and a CONNECT session live concurrently on the same +/// registration, both serving. +#[tokio::test] +async fn connect_and_associate_served_on_the_same_registration() { + let topo = wire_consumer_side_with(echo_dial(), Some(echo_egress()), None, None).await; + + let mut assoc = AssociateSession::associate(topo.consumer.as_ref()) + .await + .expect("the association associates"); + let mut connect = Socks5ChannelSession::connect( + topo.consumer.as_ref(), + ("echo.test", 443u16).to_target_addr().unwrap(), + ) + .await + .expect("the CONNECT session connects"); + + // The association round-trips while the CONNECT pipe is open. + let target = TargetAddr::Ip("10.0.0.7:53".parse().unwrap()); + assoc + .send_datagram(&target, b"udp") + .await + .expect("send datagram"); + let (_, echoed) = + tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram()) + .await + .expect("recv timed out") + .expect("a datagram (not eof)") + .expect("a datagram"); + assert_eq!(&echoed[..], b"udp"); + + // The CONNECT pipe round-trips while the association is open. + connect + .split() + .expect("pipe") + .1 + .write_all(b"tcp") + .await + .expect("write"); + let (mut read, _) = connect.split().expect("pipe"); + let mut buf = [0u8; 3]; + tokio::time::timeout( + std::time::Duration::from_secs(10), + read.read_exact(&mut buf), + ) + .await + .expect("echo timed out") + .expect("read the echo"); + assert_eq!(&buf, b"tcp"); + + assoc.shutdown().await.expect("shutdown"); + connect.close().await; + no_leaked_channels(&topo).await; +}