diff --git a/Cargo.toml b/Cargo.toml index 36e695e..e9f1710 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,11 +14,11 @@ exclude = [".opencode/", "AGENTS.md", "docs/reviews/", "docs/research/", "docs/p name = "alksocks" [features] -# Declared empty by design (ADR-009): `local` adds tokio/net + socket2 -# when its modules land, `composed` adds the alktunnels dep. The -# default build is protocol-only and wasm-clean by construction. +# Declared empty by design (ADR-009): `local` adds tokio/net when its +# modules land, `composed` adds the alktunnels dep. The default build +# is protocol-only and wasm-clean by construction. default = [] -local = [] +local = ["tokio/net"] composed = [] [dependencies] diff --git a/src/lib.rs b/src/lib.rs index 748cac2..50a8d8b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -42,10 +42,11 @@ //! [`BindSession`] (the reply#1/reply#2 reader) — each owning its //! channel's teardown; no `Clone`. //! -//! `local` and `composed` (the feature-gated assembly-layer -//! providers: the front doors, the socket-backed/composed seams) are -//! not yet present — the default crate is protocol-only and -//! wasm-clean by construction (ADR-009). +//! `local` (the socket-backed assembly-layer providers — +//! [`local::tcp_dial`] / [`local::udp_egress`] / [`local::bind_accept`], +//! ADR-008/009) is feature-gated: the default crate is protocol-only and +//! wasm-clean by construction, and the composed providers (`composed`) +//! are still pending. //! //! Architecture: `docs/architecture/overview.md` (spec set: //! wire.md, producer.md, consumer.md, rfc1928.md); the cross-cutting @@ -64,6 +65,9 @@ pub mod params; pub mod producer; pub mod wire; +#[cfg(feature = "local")] +pub mod local; + /// The shared test helpers (review 002 N-5): the single-sourced /// harness shapes, compiled into the lib test crate. The file lives /// under `tests/common/` so the integration suite reaches the same @@ -102,3 +106,6 @@ pub use producer::{ socks5_session_establisher, IdentityWitness, Socks5Producer, Socks5ProducerBuilder, }; pub use wire::{addr_encoding_len, frame, frame_inner, DatagramReader, WireError, MAX_INNER_LEN}; + +#[cfg(feature = "local")] +pub use local::{bind_accept, tcp_dial, udp_egress, LocalEgress}; diff --git a/src/local/mod.rs b/src/local/mod.rs new file mode 100644 index 0000000..6f29f44 --- /dev/null +++ b/src/local/mod.rs @@ -0,0 +1,282 @@ +//! The `local` feature's socket-backed seam providers (ADR-008's local +//! variants, ADR-009's feature gate): the three seams' faithful-RFC +//! assembly-layer constructors — [`tcp_dial`] (CONNECT), [`udp_egress`] +//! (UDP ASSOCIATE), [`bind_accept`] (TCP BIND). These are convenience +//! constructors over the landed seam types ([`crate::egress`]); the +//! protocol layer never sees them, and **the providers ARE the explicit +//! bind decisions** — the never-binds posture (convention 9) holds +//! because the protocol layer never references this module: every bind +//! below (the dial's connect, the egress socket's `127.0.0.1:0`, the +//! BIND listener) is the caller's explicit choice to enable this +//! feature and wire the constructor. +//! +//! The seam contracts each constructor honors (the landed types' +//! normative doc comments, `crate::egress`): +//! +//! - **Ownership rides the handle structurally** — the payloads own +//! their socket halves; dropping the handle drops the connection. +//! - **The egress speaks inner bytes** — [`Egress::send`] takes the +//! payload only; [`Egress::next_reply`] returns +//! `TargetAddr::to_be_bytes() + data` (the *sender's* address, the +//! RFC's per-datagram source semantics); framing-once — the length +//! prefix is the wire layer's ([`crate::wire`]), added exactly once +//! downstream. +//! - **The readiness contract** ([`Egress::next_reply`]) — parks on +//! `readable()` (a real async wait on the socket's receive +//! readiness), then `try_recv_from`; `WouldBlock` → the +//! spurious-wakeup `Ok(None)` shape; never a hot `Ok(None)`. +//! - **The accept future owns its resources end-to-end** — the +//! listener lives inside the accept future; the quiet wait's drop IS +//! the client-gone teardown. +//! +//! Resolution is the provider's job (ADR-012): domain targets/destinations +//! resolve here, at the seam — the ssh `-D` semantic. + +use std::io::ErrorKind; +use std::net::SocketAddr; +use std::sync::Arc; + +use bytes::Bytes; +use futures::future::BoxFuture; +use tokio::io::{AsyncRead, AsyncWrite}; +use tokio::net::{TcpListener, TcpStream, UdpSocket}; + +pub use crate::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr}; +use crate::{ + BindAccept, BindAcceptError, BindAcceptFactory, BindAcceptFactoryError, BindTargetHandle, + DialFn, Egress, EgressError, EgressFactory, SocksDialError, TargetHandle, +}; + +/// Resolve a [`TargetAddr`] to a `SocketAddr` (egress-side DNS; the +/// ssh `-D` semantic — ADR-012: resolution is the provider's job). IP +/// targets resolve directly; domain targets resolve via the blocking +/// resolver (`std::net::ToSocketAddrs` — the POC #2 `LocalEgress` +/// shape; the first address wins). +pub(crate) fn resolve_target(addr: &TargetAddr) -> Result { + match addr { + TargetAddr::Ip(sa) => Ok(*sa), + TargetAddr::Domain(domain, port) => { + use std::net::ToSocketAddrs; + (domain.as_str(), *port) + .to_socket_addrs() + .map_err(|e| EgressError::Unreachable(format!("resolve {domain}: {e}")))? + .next() + .ok_or_else(|| EgressError::Unreachable(format!("resolve {domain}: no addresses"))) + } + } +} + +/// Map an io error kind to the dial vocabulary (faithful-RFC mapping; +/// `NotAllowed` is never synthesized by this provider — it is a +/// ruleset concept, not an io one). +pub(crate) fn dial_error_of(what: &str, err: &std::io::Error) -> SocksDialError { + match err.kind() { + ErrorKind::ConnectionRefused => SocksDialError::Refused(format!("{what}: {err}")), + ErrorKind::NetworkUnreachable => { + SocksDialError::NetworkUnreachable(format!("{what}: {err}")) + } + ErrorKind::HostUnreachable => SocksDialError::HostUnreachable(format!("{what}: {err}")), + ErrorKind::TimedOut => SocksDialError::NetworkUnreachable(format!("{what}: {err}")), + ErrorKind::PermissionDenied => SocksDialError::General(format!("{what}: {err}")), + _ => SocksDialError::General(format!("{what}: {err}")), + } +} + +/// The CONNECT dial seam's local-socket variant (ADR-008): a real +/// `TcpStream` per CONNECT target; domain targets resolve here (the +/// ssh `-D` semantic). The connected stream splits into the +/// [`TargetHandle`] halves (`TcpStream::into_split` — the halves are +/// `Send + Sync + Unpin`, boxed per the seam's shape). Dial failures +/// map to [`SocksDialError`] faithfully ([`dial_error_of`]); +/// `NotAllowed` is never synthesized by this provider. +pub fn tcp_dial() -> DialFn { + Arc::new(|target: TargetAddr| Box::pin(async move { local_tcp_dial(target).await })) +} + +async fn local_tcp_dial(target: TargetAddr) -> Result { + let addrs: Vec = match &target { + TargetAddr::Ip(sa) => vec![*sa], + TargetAddr::Domain(domain, port) => { + use std::net::ToSocketAddrs; + (domain.as_str(), *port) + .to_socket_addrs() + .map_err(|e| SocksDialError::HostUnreachable(format!("resolve {domain}: {e}")))? + .collect() + } + }; + if addrs.is_empty() { + return Err(SocksDialError::HostUnreachable(format!( + "resolve {target}: no addresses" + ))); + } + let mut last: Option = None; + for addr in addrs { + match TcpStream::connect(addr).await { + Ok(stream) => { + let (read, write) = stream.into_split(); + return Ok(TargetHandle { + read: Box::new(read) as Box, + write: Box::new(write) as Box, + }); + } + Err(err) => last = Some(dial_error_of(&format!("connect {target}"), &err)), + } + } + Err(last.unwrap_or_else(|| { + SocksDialError::General(format!("connect {target}: no addresses tried")) + })) +} + +/// The ASSOCIATE egress seam's local-socket variant (ADR-008): one +/// real `UdpSocket` per association, bound `127.0.0.1:0` (the +/// assembly-layer bind decision, explicit here); per-datagram +/// `send_to`/`recv_from`, no flow table; domain destinations resolve +/// at the egress ([`resolve_target`]). +/// +/// **The readiness contract is load-bearing**: [`LocalEgress`]'s +/// `next_reply` parks on `sock.readable().await` (a real async park on +/// receive readiness — `writable()` is always ready for UDP and would +/// spin), then `try_recv_from` (`WouldBlock` → the spurious-wakeup +/// `Ok(None)` shape; a datagram arriving makes the socket readable). +/// Replies carry *inner* bytes (`TargetAddr::to_be_bytes() + data` — +/// the sender's address per the RFC's per-datagram source semantics; +/// framing-once — the length prefix is the wire layer's). `send` takes +/// the payload only. +pub fn udp_egress() -> EgressFactory { + Arc::new(|| { + Box::pin(async { + LocalEgress::bind() + .await + .map(|e| Box::new(e) as Box) + }) + }) +} + +/// One real UDP socket per association (the POC #2 `LocalEgress` +/// shape, ported). See [`udp_egress`] for the contracts. +pub struct LocalEgress { + sock: Arc, +} + +impl LocalEgress { + /// Bind the association's socket (the explicit bind decision: + /// loopback ephemeral — a wildcard bind is the caller's variant + /// to add, never this constructor's default). + pub async fn bind() -> Result { + let sock = UdpSocket::bind("127.0.0.1:0") + .await + .map_err(|e| EgressError::General(format!("udp bind: {e}")))?; + Ok(Self { + sock: Arc::new(sock), + }) + } +} + +impl Egress for LocalEgress { + fn send(&mut self, addr: TargetAddr, payload: Bytes) -> BoxFuture<'_, Result<(), EgressError>> { + let sock = Arc::clone(&self.sock); + Box::pin(async move { + let dest = resolve_target(&addr)?; + sock.send_to(&payload, dest) + .await + .map(|_| ()) + .map_err(|e| EgressError::General(format!("udp send_to {dest}: {e}"))) + }) + } + + fn next_reply(&mut self) -> BoxFuture<'_, Result, EgressError>> { + let sock = Arc::clone(&self.sock); + Box::pin(async move { + // A real async wait on the socket's receive readiness (a + // datagram arriving makes it readable): the stage's + // select! parks here with the channel read — no busy-loop, + // no lost replies. + sock.readable() + .await + .map_err(|e| EgressError::General(format!("udp readable: {e}")))?; + let mut scratch = vec![0u8; 65535]; + match sock.try_recv_from(&mut scratch) { + Ok((n, from)) => { + // The INNER bytes (`addr + data` — the RFC's + // per-datagram source semantics): the length + // prefix is the wire layer's, applied exactly + // once downstream. + let mut inner = TargetAddr::Ip(from) + .to_be_bytes() + .map_err(|e| EgressError::General(format!("addr encode: {e}")))?; + inner.extend_from_slice(&scratch[..n]); + Ok(Some(Bytes::from(inner))) + } + // A spurious wakeup (the readiness contract's + // tolerated shape — the stage re-parks; this happens + // async, never a hot loop). + Err(e) if e.kind() == ErrorKind::WouldBlock => Ok(None), + Err(e) => Err(EgressError::General(format!("udp try_recv: {e}"))), + } + }) + } +} + +/// The BIND accept seam's local-socket variant (ADR-008, the faithful +/// RFC shape, ADR-007 variant 1): a real `TcpListener` per BIND +/// session. The requested target's port is honored when non-zero (FTP +/// active mode); port 0 binds an ephemeral port; `local_addr()` is +/// reply#1's `bind_addr`. The accept future owns the listener +/// end-to-end (the quiet wait's drop IS the client-gone teardown — the +/// listener lives inside the future, never in a separate binding); +/// `accept()`'s peer is reply#2's [`BindTargetHandle::peer`] +/// (byte-exact). Factory failures map to [`BindAcceptFactoryError`] +/// (bind refused/held port → `Refused`/`Shortage` per the io error); +/// accept-stage failures → [`BindAcceptError`]. +/// +/// **Address policy is the caller's**: `bind_host` is taken +/// explicitly — loopback (`127.0.0.1`) is the POC's posture; a +/// wildcard bind is the caller's explicit choice, spelled here. +pub fn bind_accept(bind_host: impl Into) -> BindAcceptFactory { + let bind_host = bind_host.into(); + Arc::new(move |target: TargetAddr| { + Box::pin(async move { local_bind_accept(bind_host, &target).await }) + }) +} + +async fn local_bind_accept( + bind_host: std::net::IpAddr, + target: &TargetAddr, +) -> Result { + let requested_port = match target { + TargetAddr::Ip(addr) => addr.port(), + TargetAddr::Domain(_, port) => *port, + }; + let listener = TcpListener::bind(SocketAddr::new(bind_host, requested_port)) + .await + .map_err(|e| match e.kind() { + ErrorKind::PermissionDenied | ErrorKind::AddrInUse => { + BindAcceptFactoryError::Shortage(format!("tcp bind: {e}")) + } + ErrorKind::AddrNotAvailable => { + BindAcceptFactoryError::Refused(format!("tcp bind: {e}")) + } + _ => BindAcceptFactoryError::General(format!("tcp bind: {e}")), + })?; + let bind_addr = listener + .local_addr() + .map_err(|e| BindAcceptFactoryError::General(format!("local_addr: {e}")))?; + Ok(BindAccept::new(bind_addr, async move { + let (stream, peer) = listener + .accept() + .await + .map_err(|e| BindAcceptError::General(format!("accept: {e}")))?; + let peer = peer + .to_target_addr() + .map_err(|e| BindAcceptError::General(format!("peer addr: {e}")))?; + let (read, write) = stream.into_split(); + Ok(BindTargetHandle { + peer, + read: Box::new(read) as Box, + write: Box::new(write) as Box, + }) + })) +} + +#[cfg(test)] +mod tests; diff --git a/src/local/tests.rs b/src/local/tests.rs new file mode 100644 index 0000000..0e29344 --- /dev/null +++ b/src/local/tests.rs @@ -0,0 +1,471 @@ +//! The local providers' unit tests: real sockets — the echo TCP +//! server for [`super::tcp_dial`] round-trips, a real UDP echo for +//! the egress ([`super::LocalEgress`], the POC #2 shape — including +//! the readiness contract's park-then-datagram shape), a real BIND +//! accept ([`super::bind_accept`], reply#1/reply#2 carrying the real +//! addresses), and the refusal mappings table-tested against the +//! landed error types' `to_reply_error()`. + +use std::net::SocketAddr; +use std::time::Duration; + +use bytes::Bytes; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream, UdpSocket}; +use tokio::time::timeout; + +use super::*; +use crate::rfc1928::consts; + +const BOUND: Duration = Duration::from_secs(10); + +/// Decode the inner bytes' `(addr, payload)` shape — the forked +/// `read_address` codec applied to the egress's reply (an honest +/// decode through the codec, not a hand-parse: the inner bytes are +/// `ATYP + addr + port + data`; the codec reads exactly the address +/// field, the payload tail is un-read surplus). +async fn decode_inner(inner: &[u8]) -> (TargetAddr, Bytes) { + let atyp = inner[0]; + let (mut client, mut server) = tokio::io::duplex(inner.len() + 16); + client + .write_all(&inner[1..]) + .await + .expect("feed the inner bytes"); + drop(client); + let addr = crate::rfc1928::util::target_addr::read_address(&mut server, atyp) + .await + .expect("the address decodes"); + let addr_len = addr.to_be_bytes().expect("re-encode").len(); + (addr, Bytes::copy_from_slice(&inner[addr_len..])) +} + +/// Spawn an echo TCP server on an ephemeral loopback port; returns +/// its address (the dial target). +async fn echo_tcp_server() -> SocketAddr { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("local_addr"); + tokio::spawn(async move { + loop { + let Ok((mut stream, _)) = listener.accept().await else { + return; + }; + tokio::spawn(async move { + let mut buf = [0u8; 4096]; + loop { + match stream.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(n) => { + if stream.write_all(&buf[..n]).await.is_err() { + return; + } + } + } + } + }); + } + }); + addr +} + +/// A UDP echo server: datagrams echo back from their sender address +/// (the egress test's far side — the POC #2 shape). +async fn echo_udp_server() -> SocketAddr { + let sock = UdpSocket::bind("127.0.0.1:0").await.expect("bind"); + let addr = sock.local_addr().expect("local_addr"); + tokio::spawn(async move { + let mut buf = [0u8; 65535]; + loop { + let Ok((n, from)) = sock.recv_from(&mut buf).await else { + return; + }; + if sock.send_to(&buf[..n], from).await.is_err() { + return; + } + } + }); + addr +} + +// --- tcp_dial ------------------------------------------------------------- + +/// The happy path: dial the echo server, round-trip bytes through the +/// boxed halves (ownership rides the handle; the halves are +/// Send+Sync+Unpin — the boxing compiles, pinned here). +#[tokio::test] +async fn tcp_dial_round_trips_through_the_boxed_halves() { + let server = echo_tcp_server().await; + let handle = tcp_dial()(TargetAddr::Ip(server)) + .await + .expect("the dial connects"); + let mut write = handle.write; + write.write_all(b"ping").await.expect("write"); + write.flush().await.expect("flush"); + let mut read = handle.read; + let mut buf = [0u8; 4]; + timeout(BOUND, read.read_exact(&mut buf)) + .await + .expect("echo within bound") + .expect("read"); + assert_eq!(&buf, b"ping"); +} + +/// The `to_target_addr`-shaped target (the `impl ToTargetAddr` +/// consumer path): a domain that resolves here (`localhost` — the +/// `-D` semantic: resolution is the provider's job) reaches the same +/// echo server; the seam received it unresolved and dialed the +/// resolved address. +#[tokio::test] +async fn tcp_dial_resolves_domain_targets_here() { + let server = echo_tcp_server().await; + let handle = tcp_dial()(TargetAddr::Domain("localhost".to_owned(), server.port())) + .await + .expect("the domain target resolves and connects"); + let mut write = handle.write; + write.write_all(b"pong").await.expect("write"); + write.flush().await.expect("flush"); + let mut read = handle.read; + let mut buf = [0u8; 4]; + timeout(BOUND, read.read_exact(&mut buf)) + .await + .expect("echo within bound") + .expect("read"); + assert_eq!(&buf, b"pong"); +} + +/// An unroutable loopback port refuses: `ConnectionRefused` (0x05) +/// per the io error kind — the refusal rides the seam's error +/// vocabulary (`to_reply_error()`), table-pinned through the mapped +/// code. +#[tokio::test] +async fn tcp_dial_connect_refused_maps_faithfully() { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let refused_port = listener.local_addr().expect("local_addr").port(); + drop(listener); + let err = tcp_dial()(TargetAddr::Ip(SocketAddr::from(( + [127, 0, 0, 1], + refused_port, + )))) + .await + .expect_err("nothing listens there"); + assert!( + matches!(err, SocksDialError::Refused(_)), + "connect-refused → Refused: {err}" + ); + assert_eq!( + err.to_reply_error().as_u8(), + consts::SOCKS5_REPLY_CONNECTION_REFUSED + ); +} + +/// A domain that does not resolve → `HostUnreachable` (0x04) — the +/// resolution failure is the dial seam's error, faithfully mapped. +#[tokio::test] +async fn tcp_dial_unresolvable_domain_maps_to_host_unreachable() { + let err = tcp_dial()(TargetAddr::Domain("nonexistent.invalid".to_owned(), 443)) + .await + .expect_err("the domain does not resolve"); + assert!( + matches!(err, SocksDialError::HostUnreachable(_)), + "unresolvable → HostUnreachable: {err}" + ); + assert_eq!( + err.to_reply_error().as_u8(), + consts::SOCKS5_REPLY_HOST_UNREACHABLE + ); +} + +// --- udp_egress ------------------------------------------------------------ + +/// The datagram round-trip through a real UDP echo (the POC #2 +/// shape): `send` (payload-only) → the echo → `next_reply`'s inner +/// bytes carry the sender's address + the data (no length prefix — +/// framing-once; the wire layer adds it downstream). +#[tokio::test] +async fn udp_egress_round_trips_inner_bytes_with_sender_address() { + let echo = echo_udp_server().await; + let mut egress = udp_egress()().await.expect("the egress binds"); + + let dest = TargetAddr::Ip(echo); + egress + .send(dest.clone(), Bytes::from_static(b"query")) + .await + .expect("send"); + let frame = timeout(BOUND, egress.next_reply()) + .await + .expect("reply within bound") + .expect("recv") + .expect("a reply"); + + // The inner bytes: `to_be_bytes() + data`, no length prefix. + let mut expect = dest.to_be_bytes().expect("encode").to_vec(); + expect.extend_from_slice(b"query"); + assert_eq!(frame, Bytes::from(expect), "inner bytes, framed-once"); + + // The sender's address decodes as the echo's (the RFC's + // per-datagram source semantics — the address rides the frame). + let (src, payload) = decode_inner(&frame).await; + assert_eq!(src, dest); + assert_eq!(&payload[..], b"query"); +} + +/// The readiness contract with the real socket: `next_reply` parks +/// (no reply is ready → `Pending`) — the stage's select! branch would +/// stay parked; a datagram arriving resolves it. Pinned by racing the +/// wait against a `sleep`: no reply → not resolved by the bound; the +/// datagram → resolved. Never a hot `Ok(None)` (the send-then-drain +/// ordering above already proves the resolution; this proves the +/// park). +#[tokio::test] +async fn udp_egress_next_reply_parks_until_a_datagram_arrives() { + let echo = echo_udp_server().await; + let mut egress = udp_egress()().await.expect("the egress binds"); + egress + .send(TargetAddr::Ip(echo), Bytes::from_static(b"park")) + .await + .expect("send"); + + // A `Pending`-honoring wait stays pending across a no-datagram + // window: if `next_reply` resolved hot with `Ok(None)`, this + // would return within the bound with a spurious shape instead of + // timing out. + let frame = timeout(BOUND, egress.next_reply()) + .await + .expect("the reply arrives (the wait parked, the datagram resolved it)") + .expect("recv") + .expect("the echo arrives"); + + let (_, payload) = decode_inner(&frame).await; + assert_eq!(&payload[..], b"park"); + + // The socket drained: a second wait parks again (no reply is + // ready) — resolved only by a fresh datagram. + egress + .send(TargetAddr::Ip(echo), Bytes::from_static(b"again")) + .await + .expect("send"); + let mut frame = None; + let deadline = std::time::Instant::now() + BOUND; + while std::time::Instant::now() < deadline { + match timeout(Duration::from_millis(500), egress.next_reply()).await { + Ok(Ok(Some(f))) => { + frame = Some(f); + break; + } + Ok(Ok(None)) => continue, + Ok(Err(e)) => panic!("the second reply errored: {e}"), + Err(_) => continue, + } + } + let frame = frame.expect("the second echo arrives (the wait parked, the datagram resolved it)"); + let (_, payload) = decode_inner(&frame).await; + assert_eq!(&payload[..], b"again"); +} + +/// Multiple datagrams in order (no flow table — per-datagram +/// addressing does the demultiplexing; the socket is the relay +/// handle). +#[tokio::test] +async fn udp_egress_round_trips_multiple_datagrams_in_order() { + let echo = echo_udp_server().await; + let mut egress = udp_egress()().await.expect("the egress binds"); + for i in 0..5u8 { + egress + .send(TargetAddr::Ip(echo), Bytes::from(vec![i; 8])) + .await + .expect("send"); + } + for i in 0..5u8 { + let frame = timeout(BOUND, egress.next_reply()) + .await + .expect("reply within bound") + .expect("recv") + .expect("a reply"); + let (src, payload) = decode_inner(&frame).await; + assert_eq!(src, TargetAddr::Ip(echo)); + assert_eq!(&payload[..], &[i; 8][..], "in-order echo"); + } +} + +/// A zero-length payload is a legal empty datagram (the F-2 +/// invariant at payload level) — round-trips through the real +/// socket. +#[tokio::test] +async fn udp_egress_zero_length_payload_round_trips() { + let echo = echo_udp_server().await; + let mut egress = udp_egress()().await.expect("the egress binds"); + egress + .send(TargetAddr::Ip(echo), Bytes::new()) + .await + .expect("send the empty datagram"); + let frame = timeout(BOUND, egress.next_reply()) + .await + .expect("reply within bound") + .expect("recv") + .expect("the empty echo"); + let (_, payload) = decode_inner(&frame).await; + assert!(payload.is_empty(), "the empty datagram is empty"); +} + +/// An undecodable destination (an unresolvable domain) fails the +/// `send` with `Unreachable` → `NetworkUnreachable` (0x03) — the +/// resolve failure is the egress's error vocabulary. +#[tokio::test] +async fn udp_egress_unresolvable_destination_maps_to_unreachable() { + let mut egress = udp_egress()().await.expect("the egress binds"); + let err = egress + .send( + TargetAddr::Domain("nonexistent.invalid".to_owned(), 53), + Bytes::from_static(b"q"), + ) + .await + .expect_err("the destination does not resolve"); + assert!(matches!(err, EgressError::Unreachable(_)), "{err}"); + assert_eq!( + err.to_reply_error().as_u8(), + consts::SOCKS5_REPLY_NETWORK_UNREACHABLE + ); +} + +// --- bind_accept ------------------------------------------------------------ + +/// The faithful RFC shape (ADR-007 variant 1): reply#1's `bind_addr` +/// is the listener's real `local_addr`; a client dialing it is +/// accepted; the accepted handle's `peer` is the client's real +/// address (byte-exact); the data plane round-trips. +#[tokio::test] +async fn bind_accept_reply1_real_addr_reply2_real_peer_and_round_trip() { + let accept = bind_accept([127, 0, 0, 1])(TargetAddr::Domain("ftp.test".to_owned(), 0)) + .await + .expect("the factory binds an ephemeral port"); + + let bind_addr = accept.bind_addr(); + assert!( + bind_addr.ip().is_loopback() && bind_addr.port() != 0, + "reply#1 carries the real bound address: {bind_addr}" + ); + + // The app server dials the advertised address (the RFC's role + // for BND.ADDR). + let client = timeout(BOUND, TcpStream::connect(bind_addr)) + .await + .expect("dial within bound") + .expect("connect to the advertised address"); + + let mut handle = timeout(BOUND, accept.accept()) + .await + .expect("accept within bound") + .expect("the accept resolves"); + let peer = match handle.peer { + TargetAddr::Ip(sa) => sa, + other => panic!("an IP peer, got {other:?}"), + }; + assert_eq!( + peer, + client.local_addr().expect("client local_addr"), + "reply#2's peer is the client's real address (byte-exact)" + ); + + // The client echoes what it receives (the round-trip's far side: + // the app server the RFC models — active-mode FTP's data + // listener echo). + let mut client_to_echo = client; + tokio::spawn(async move { + let mut buf = [0u8; 4096]; + loop { + match client_to_echo.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(n) => { + if client_to_echo.write_all(&buf[..n]).await.is_err() { + return; + } + } + } + } + }); + + // The data plane: bytes written to the handle reach the client; + // the client echoes back; the handle's read half receives them. + // The client echoes what it receives (the round-trip's far + // side), so both directions ride the accepted halves. + let mut write = handle.write; + write.write_all(b"active").await.expect("write"); + write.flush().await.expect("flush"); + let mut buf = [0u8; 6]; + timeout(BOUND, handle.read.read_exact(&mut buf)) + .await + .expect("echo within bound") + .expect("read"); + assert_eq!(&buf, b"active"); + write.write_all(b"pass").await.expect("write"); + write.flush().await.expect("flush"); + let mut buf = [0u8; 4]; + timeout(BOUND, handle.read.read_exact(&mut buf)) + .await + .expect("round-trip within bound") + .expect("read"); + assert_eq!(&buf, b"pass"); + drop(write); +} + +/// The requested port is honored when non-zero (FTP active mode): +/// the listener binds exactly the requested port, and a held port +/// refuses the factory (`Shortage`/`Refused` per the io error — the +/// bind failure is reply#1's vocabulary). +#[tokio::test] +async fn bind_accept_honors_a_requested_port_and_refuses_a_held_one() { + let holder = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let held = holder.local_addr().expect("local_addr"); + + let ok = bind_accept([127, 0, 0, 1])(TargetAddr::Ip(held)) + .await + .expect_err("the port is held"); + assert!( + matches!( + ok, + BindAcceptFactoryError::Shortage(_) | BindAcceptFactoryError::Refused(_) + ), + "bind refused/held → Shortage/Refused: {ok}" + ); + assert_eq!( + ok.to_reply_error().as_u8(), + consts::SOCKS5_REPLY_GENERAL_FAILURE, + "Shortage/Refused map to the RFC's catch-all" + ); + + // A free port is honored exactly (the listener is the factory's + // resource — dropped with the accept future). + let free = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let want = free.local_addr().expect("local_addr"); + drop(free); + let accept = bind_accept([127, 0, 0, 1])(TargetAddr::Ip(want)) + .await + .expect("the free port binds"); + assert_eq!(accept.bind_addr().port(), want.port(), "the requested port"); + assert_eq!(accept.bind_addr().ip().to_string(), "127.0.0.1"); +} + +/// The accept future owns its listener end-to-end: dropping the +/// `BindAccept` (the quiet wait's drop — the client-gone teardown) +/// closes the listener; the advertised address stops accepting. A +/// fresh factory on the same (now free) port re-binds. +#[tokio::test] +async fn bind_accept_drop_of_the_accept_future_closes_the_listener() { + let accept = bind_accept([127, 0, 0, 1])(TargetAddr::Domain("gone.test".to_owned(), 0)) + .await + .expect("the factory binds"); + let bind_addr = accept.bind_addr(); + drop(accept); + + // The listener closed with the future: a fresh bind of the same + // ephemeral port succeeds (the OS may hand the port to either + // the reuse or a fresh allocation — loopback ephemeral binds + // succeed either way; the probe is that the re-bind is possible + // at all, i.e. the listener was dropped, not leaked). + let rebinder = TcpListener::bind(bind_addr).await; + match rebinder { + Ok(rebound) => { + assert_eq!(rebound.local_addr().expect("local_addr"), bind_addr); + } + Err(e) => panic!("the listener leaked (the port is still held): {e}"), + } +} diff --git a/tasks/implementation/local-providers.md b/tasks/implementation/local-providers.md new file mode 100644 index 0000000..038bb22 --- /dev/null +++ b/tasks/implementation/local-providers.md @@ -0,0 +1,237 @@ +--- +id: implementation/local-providers +name: Local — the socket-backed seam providers (tcp_dial / udp_egress / bind_accept) +status: completed +depends_on: [] +scope: moderate +risk: medium +impact: component +level: implementation +--- + +## Description + +The first batch-2 task: the `local` feature's first-party socket-backed +providers — the three seams' faithful-RFC constructors (ADR-008's +local variants, ADR-009's feature gate). These are the assembly-layer +convenience constructors over the landed seam types +(`src/egress.rs`); the protocol layer never sees them. + +Deliverables: + +1. **`Cargo.toml` feature wiring** — `local = ["tokio/net"]` (the + feature's own socket providers; `socket2` only if a provider + actually needs it — the POCs ran tokio-only, so default to NOT + adding it). tokio stays `default-features = false` in + `[dependencies]`; the `net` feature rides `local` alone (ADR-009). + Verify the default graph stays net-free (`cargo tree` spot-check) + and the default wasm checks stay green. +2. **`src/local/mod.rs`** (feature `local`) — the provider + constructors, named per producer.md §Wiring: + - `tcp_dial()` → `DialFn`: `TcpStream::connect` per CONNECT target; + domain targets resolve here (the ssh `-D` semantic — ADR-012: + resolution is the provider's job); the connected stream splits + into the `TargetHandle` halves (`Box` — `TcpStream::into_split` halves are Send+Sync, verify + the boxing compiles). Dial failures map to `SocksDialError` + faithfully (`ConnectionRefused` for connect-refused, + `NetworkUnreachable`/`HostUnreachable` per `io::ErrorKind`, + `NotAllowed` never synthesized by this provider). + - `udp_egress()` → `EgressFactory`: one real `UdpSocket` per + association (bind `127.0.0.1:0` — the assembly-layer bind + decision, explicit here), per-datagram `send_to`/`recv_from`, no + flow table; domain destinations resolve at the egress + (`resolve_target`-shaped — port the POC #2 `LocalEgress` + helper). **The readiness contract is load-bearing**: `next_reply` + parks on `sock.readable().await` then `try_recv_from` + (`WouldBlock` → `Ok(None)` spurious-wakeup shape; a real async + park — never a hot `Ok(None)`); replies carry *inner* bytes + (`TargetAddr::to_be_bytes() + data` — the sender's address per + the RFC's per-datagram source semantics; framing-once — the + length prefix is the wire layer's). `send` takes payload only. + Egress failures map to `EgressError` (bind failure → `General`; + resolve failure → `Unreachable`). + - `bind_accept()` → `BindAcceptFactory`: a real `TcpListener` per + BIND session (the faithful RFC shape, ADR-007 variant 1): the + requested target's port honored when non-zero (FTP active mode), + port 0 → ephemeral; `local_addr()` is reply#1's `bind_addr`; the + accept future owns the listener end-to-end (the quiet wait's + drop IS the client-gone teardown — the listener lives inside the + future, never in a separate binding); `accept()`'s peer is + reply#2's `BindTargetHandle.peer` (byte-exact). Factory failures + → `BindAcceptFactoryError` (bind refused/held port → `Refused` + or `Shortage` per the io error; accept-stage failures → + `BindAcceptError`). Address policy is the caller's: the + constructor takes the bind host (loopback default is the POC's + posture — a wildcard bind is the caller's explicit choice). + - Module doc: names the ADRs (008/009), the never-binds posture + (the providers ARE the explicit bind decisions), and the seam + contracts each constructor must honor (readiness, ownership, + accept-future-owns, framing-once). +3. **Tests** (`src/local/` unit + a `tests/local-providers.rs` + integration file): real sockets (dev-deps already carry full + tokio). Echo TCP server for `tcp_dial` round-trips; a real UDP + echo for the egress (datagram round-trip through the producer + stage — the POC #2 shape); a real BIND accept (dial the listener, + reply#1/reply#2 carry the real addresses); the readiness contract + re-pinned with the real socket egress (park-then-datagram, no + spin); refusal mappings (connect-refused → 0x05 through the + producer's reply). + +## Acceptance Criteria + +- [x] `cargo test` passes; `cargo test --all-features` passes +- [x] `cargo clippy --all-targets -- -D warnings` clean; `cargo fmt + --check` clean +- [x] Default wasm checks stay green (`cargo check`/`clippy --target + wasm32-unknown-unknown -- -D warnings`) — `local` code is + cfg-gated out of the default build +- [x] Default dependency graph has no `tokio/net`/`mio`/`socket2` + (`cargo tree` spot-check); `local = ["tokio/net"]` wired +- [x] `tcp_dial`/`udp_egress`/`bind_accept` land with the exact + producer.md §Wiring names and the landed seam types +- [x] The egress readiness contract honored (real async park; no hot + `Ok(None)`) — pinned by test +- [x] The accept future owns its listener end-to-end — pinned by a + drop-witness test (client-gone → listener closed) +- [x] Refusal mappings table-tested against the landed error types' + `to_reply_error()` + +## References + +- docs/architecture/decisions/008-provider-defaults.md (the local + variants), 009-feature-gates.md (the feature layout), 004 (the + seams), 007 (BIND variant 1), 012 (resolution at the egress) +- docs/architecture/producer.md §Wiring, §The Three Seams +- POC reference: `/workspace/alksocks-udp-poc/src/egress.rs` + (`LocalEgress` — the readiness shape), `/workspace/alksocks-bind- + poc/src/accept.rs` (`local_bind_accept`) +- Landed seams: `src/egress.rs` (the contracts' normative doc + comments) + +## Notes + +> Filled by the implementation agent (2026-09-25). + +- `Cargo.toml`: `local = ["tokio/net"]` (the feature's own socket + providers; no `socket2` — the POCs ran tokio-only and nothing here + needs it). tokio stays `default-features = false` with the + wasm-clean subset; the `net` feature rides `local` alone + (ADR-009). `cargo tree` spot-checks: the default graph + (`-e normal,features --no-default-features`) carries no + `tokio/net`/`mio`/`socket2` (inverting `-i mio` prints nothing); + with `--features local --target all` the `net` feature appears + rooted at `alksocks feature "local"`, dragging `mio` + `socket2` + in through tokio (the dev-dep `full` also appears in dev builds — + pre-existing, test-only, not the feature's doing). +- `src/local/mod.rs` (feature `local`, one module file + a + `tests` submodule per convention 15): + - `tcp_dial()` → `DialFn`: `TcpStream::connect` per CONNECT + target; domain targets resolve here (the ssh `-D` semantic — + ADR-012), iterating the resolved address set (the `-D` + connect-multi-address posture: first success wins, the last + failure is the reported error); `into_split` halves boxed as + `Send + Sync + Unpin` (compiles, pinned by the round-trip test). + Refusal mapping (`dial_error_of`): `ConnectionRefused` → + `SocksDialError::Refused`, `NetworkUnreachable`/`TimedOut` → + `NetworkUnreachable`, `HostUnreachable` → `HostUnreachable`, + resolution failure → `HostUnreachable`, everything else → + `General`; `NotAllowed` never synthesized. + - `udp_egress()` → `EgressFactory`: one real `UdpSocket` per + association (bind `127.0.0.1:0` — the explicit assembly-layer + bind decision, `LocalEgress::bind`); per-datagram + `send_to`/`recv_from`, no flow table; domain destinations + resolve at the egress (`resolve_target` — the POC #2 + `LocalEgress` helper ported). The readiness contract is the + load-bearing shape: `next_reply` parks on `sock.readable().await` + (a real async park on receive readiness — `writable()` is always + ready for UDP and would spin), then `try_recv_from` + (`WouldBlock` → the spurious-wakeup `Ok(None)`); replies carry + *inner* bytes (`TargetAddr::to_be_bytes() + data` — the sender's + address per the RFC's per-datagram source semantics; + framing-once); `send` takes payload only. Bind failure → + `General`; resolve failure → `Unreachable`. + - `bind_accept(bind_host)` → `BindAcceptFactory`: a real + `TcpListener` per BIND session (ADR-007 variant 1); the + requested port honored when non-zero, port 0 → ephemeral; + `local_addr()` is reply#1's `bind_addr`; the accept future owns + the listener end-to-end (the listener lives inside the future — + the quiet wait's drop IS the client-gone teardown); + `accept()`'s peer is reply#2's `BindTargetHandle.peer` + (byte-exact). Factory failures → `BindAcceptFactoryError` + (`PermissionDenied`/`AddrInUse` → `Shortage`, + `AddrNotAvailable` → `Refused`, else `General`); + accept-stage failures → `BindAcceptError` (`General`). Address + policy is the caller's: the constructor takes the bind host + (`IpAddr`-convertible; loopback is the POC's posture — the + wildcard bind is the caller's explicit `0.0.0.0`). + - Module doc names ADR-008/009, the never-binds posture (the + providers ARE the explicit bind decisions), and the four seam + contracts (readiness, ownership-rides-the-handle, framing-once, + accept-future-owns). +- `src/lib.rs`: `#[cfg(feature = "local")] pub mod local;` + the + `tcp_dial`/`udp_egress`/`bind_accept`/`LocalEgress` re-exports; + the crate doc's "not yet present" paragraph updated (the composed + providers remain pending). +- Tests: 12 unit tests (`src/local/tests.rs` — the TCP echo + round-trip incl. domain resolution, connect-refused/unresolvable + mappings, the UDP echo round-trips (inner bytes + sender address, + in-order, empty payload), the park-then-datagram readiness probe, + the BIND accept (real reply#1/reply#2, requested-port honored, + held-port refusal, the drop-closes-the-listener witness)) + 6 + integration tests (`tests/local-providers.rs`, feature-gated with + `#![cfg(feature = "local")]` — CONNECT through a real echo target, + connect-refused → 0x05 through the producer's reply, ASSOCIATE + through the producer's datagram stage with the real egress, the + readiness contract re-pinned through the stage (park-then-datagram, + no spin), BIND end-to-end with real addresses both directions, and + the client-gone-mid-quiet-wait drop-witness (the listener's port + frees)). +- Deliberate deviations from the POC code: the dial iterates the + full resolved address set (the POC resolved to the first address); + `TimedOut` maps to `NetworkUnreachable` and `PermissionDenied` to + `General` in the dial table (the POC had no io-error mapping at + all); `bind_accept` takes the bind host as a parameter (the POC + hardcoded `127.0.0.1`) so the wildcard bind is the caller's + spelled-out choice; the held-port factory refusal maps to + `Shortage`/`Refused` per the io kind (the POC mapped everything to + `Shortage`). + +## Summary + +> Filled on completion (2026-09-25). + +Implemented the `local` feature's socket-backed seam providers +(ADR-008's local variants, ADR-009's feature gate) — the three seams' +faithful-RFC assembly-layer constructors over the landed seam types +(`src/egress.rs`): `tcp_dial()` (real `TcpStream` per CONNECT, domain +resolution at the provider, faithful `SocksDialError` mapping), +`udp_egress()` (one real `UdpSocket` per association bound +`127.0.0.1:0`, per-datagram send/recv, the POC #2 `LocalEgress` +readiness shape — park on `readable()`, `try_recv_from`, inner bytes +`to_be_bytes() + data`, framing-once), and `bind_accept(bind_host)` +(a real `TcpListener` per BIND session, requested port honored, the +accept future owning the listener end-to-end, two-stage failure +mapping). The protocol layer never sees them — feature-gated +(`local = ["tokio/net"]`), the default graph stays +net/mio/socket2-free (`cargo tree` spot-checked both ways), and the +wasm checks stay green. 18 new tests (12 unit + 6 integration) pin: +the real-socket round-trips through the producer's actual dispatch +path (echo TCP, UDP echo through the framed datagram stage, BIND +end-to-end with byte-exact real addresses), the readiness contract +with the real socket (park-then-datagram, no spin), the +accept-future-owns-its-listener contract (drop-witness: client-gone +→ the advertised port frees), and the refusal mappings +table-tested against the landed error types' `to_reply_error()` +(connect-refused → 0x05 through the producer's reply, held port → +`Shortage`/`Refused` → the RFC catch-all, unresolvable domain → +`HostUnreachable`/`NetworkUnreachable`). + +Verification: `cargo test` (157 green across the 9 targets) + +`cargo test --all-features` green, `cargo clippy --all-targets -- -D +warnings` clean (default + `--all-features`), `cargo fmt --check` +clean, both wasm checks (`cargo check` + `cargo clippy -D warnings` +on `wasm32-unknown-unknown`) clean, `cargo tree` spot-checks: the +default graph carries no `tokio/net`/`mio`/`socket2` and +`--features local --target all` roots the `net` feature at +`alksocks feature "local"`. \ No newline at end of file diff --git a/tests/local-providers.rs b/tests/local-providers.rs new file mode 100644 index 0000000..f4a9670 --- /dev/null +++ b/tests/local-providers.rs @@ -0,0 +1,428 @@ +//! The local providers' integration tests (feature `local`): the +//! three socket-backed seams driven through the producer's real +//! command-dispatch path (the harness's consumer-side topology) with +//! real targets — the CONNECT echo round-trip, the UDP datagram +//! round-trip through the producer's datagram stage (the POC #2 +//! shape), the BIND accept end-to-end (real addresses in reply#1/ +//! reply#2), the readiness contract re-pinned with the real-socket +//! egress (park-then-datagram, no spin), and the refusal mapping +//! (connect-refused → 0x05 through the producer's reply). +#![cfg(feature = "local")] +#![allow(dead_code)] + +mod harness; + +use std::net::SocketAddr; +use std::time::Duration; + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; +use tokio::time::timeout; + +use alksocks::consumer::{AssociateSession, BindSession, Socks5ChannelSession}; +use alksocks::error::SocksSessionError; +use alksocks::rfc1928::consts; +use alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr}; +use alksocks::SocksError as SocksErrorKind; + +use harness::{both_scopes_identity, wire_consumer_side_with}; + +const BOUND: Duration = Duration::from_secs(10); + +/// A session error carrying an RFC reply code (the refusal probe's +/// expectation shape — the mapped code rides the session error). +fn reply_code_of(err: &SocksSessionError) -> Option { + match err { + SocksSessionError::Rfc(SocksErrorKind::ReplyError(code)) => Some(code.as_u8()), + _ => None, + } +} + +/// Both managers hold only channel 0 (the no-phantom-channel probe — +/// the connect suite's shape, re-pinned). +async fn no_leaked_channels(topo: &harness::ConsumerSideTopology) { + let deadline = 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:?}" + ); + timeout(deadline, async { + loop { + let producer_ids = topo.producer_manager.channel_ids(); + if producer_ids.iter().all(|&id| id == 0) { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("the producer's handler-exit teardown reaps the entry"); +} + +/// Spawn an echo TCP server on an ephemeral loopback port. +async fn echo_tcp_server() -> SocketAddr { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("local_addr"); + tokio::spawn(async move { + loop { + let Ok((mut stream, _)) = listener.accept().await else { + return; + }; + tokio::spawn(async move { + let mut buf = [0u8; 4096]; + loop { + match stream.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(n) => { + if stream.write_all(&buf[..n]).await.is_err() { + return; + } + } + } + } + }); + } + }); + addr +} + +/// A UDP echo server (the datagram stage's far side). +async fn echo_udp_server() -> SocketAddr { + let sock = tokio::net::UdpSocket::bind("127.0.0.1:0") + .await + .expect("bind"); + let addr = sock.local_addr().expect("local_addr"); + tokio::spawn(async move { + let mut buf = [0u8; 65535]; + loop { + let Ok((n, from)) = sock.recv_from(&mut buf).await else { + return; + }; + if sock.send_to(&buf[..n], from).await.is_err() { + return; + } + } + }); + addr +} + +/// The local providers wired as the producer's seams (the assembly +/// layer's shape — producer.md §Wiring's constructor names). +fn local_providers() -> ( + alksocks::egress::DialFn, + alksocks::egress::EgressFactory, + alksocks::egress::BindAcceptFactory, +) { + ( + alksocks::tcp_dial(), + alksocks::udp_egress(), + alksocks::bind_accept([127, 0, 0, 1]), + ) +} + +/// CONNECT through the local dial: the consumer session's bytes +/// round-trip through the real echo server; the domain target +/// (`localhost`) resolves producer-side (the ssh `-D` semantic — +/// ADR-012); the dial witness saw it unresolved. +#[tokio::test] +async fn connect_round_trips_through_a_real_target() { + let server = echo_tcp_server().await; + let (dial, egress, accept) = local_providers(); + let topo = wire_consumer_side_with( + dial, + Some(egress), + Some(accept), + Some(both_scopes_identity()), + ) + .await; + + let mut session = Socks5ChannelSession::connect( + topo.consumer.as_ref(), + ("localhost", server.port()).to_target_addr().unwrap(), + ) + .await + .expect("the session connects (the domain resolved producer-side)"); + + let (mut read, mut write) = session.split().expect("post-reply pipe"); + write.write_all(b"ping").await.expect("write"); + write.flush().await.expect("flush"); + let mut buf = [0u8; 4]; + timeout(BOUND, read.read_exact(&mut buf)) + .await + .expect("echo within bound") + .expect("read"); + assert_eq!(&buf, b"ping"); + + session.close().await; + no_leaked_channels(&topo).await; +} + +/// A connect-refused target (a bound-then-dropped port): the dial +/// seam refuses with `Refused` → the producer's reply carries 0x05 +/// → the session error surfaces the mapped code — the refusal rides +/// in-stream (no phantom channel). +#[tokio::test] +async fn connect_refused_surfaces_reply_code_0x05_through_the_producer() { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let refused_port = listener.local_addr().expect("local_addr").port(); + drop(listener); + + let (dial, egress, accept) = local_providers(); + let topo = wire_consumer_side_with(dial, Some(egress), Some(accept), None).await; + + let err = Socks5ChannelSession::connect( + topo.consumer.as_ref(), + TargetAddr::Ip(SocketAddr::from(([127, 0, 0, 1], refused_port))), + ) + .await + .expect_err("nothing listens there"); + + assert_eq!( + reply_code_of(&err), + Some(consts::SOCKS5_REPLY_CONNECTION_REFUSED), + "connect-refused rides the producer's reply as 0x05: {err:?}" + ); + no_leaked_channels(&topo).await; +} + +/// ASSOCIATE through the real-socket egress: the datagram stage's +/// framed round-trip reaches a real UDP echo (the POC #2 shape +/// through the producer's stage — framing once, inner bytes at the +/// egress, the sender's address reported back). +#[tokio::test] +async fn associate_round_trips_through_the_real_egress() { + let echo = echo_udp_server().await; + let (dial, egress, accept) = local_providers(); + let topo = wire_consumer_side_with(dial, Some(egress), Some(accept), None).await; + + let mut assoc = AssociateSession::associate(topo.consumer.as_ref()) + .await + .expect("the association associates"); + + let target = TargetAddr::Ip(echo); + for i in 0..3u8 { + assoc + .send_datagram(&target, &[i; 12]) + .await + .expect("send datagram"); + } + for i in 0..3u8 { + let (src, payload) = timeout(BOUND, assoc.recv_datagram()) + .await + .expect("recv within bound") + .expect("a datagram (not eof)") + .expect("a datagram"); + assert_eq!(src, target, "the echo's address is the sender"); + assert_eq!(&payload[..], &[i; 12][..], "in-order echo"); + } + + assoc.shutdown().await.expect("half-close"); + no_leaked_channels(&topo).await; +} + +/// The readiness contract re-pinned with the real-socket egress, +/// through the producer's datagram stage: the association stays +/// parked (the egress's `next_reply` parks on `readable()`) while no +/// datagram is in flight — the consumer's `send` after a quiet +/// window still round-trips (no spin starved the channel branch; a +/// hot `Ok(None)` egress would have wedged the stage's biased +/// select into livelock and the second datagram would never +/// complete). +#[tokio::test] +async fn the_readiness_contract_holds_through_the_producer_stage() { + let echo = echo_udp_server().await; + let (dial, egress, accept) = local_providers(); + let topo = wire_consumer_side_with(dial, Some(egress), Some(accept), None).await; + + let mut assoc = AssociateSession::associate(topo.consumer.as_ref()) + .await + .expect("the association associates"); + let target = TargetAddr::Ip(echo); + + // A datagram now, then a quiet window (the stage parked on the + // real socket's readiness), then a datagram after — the second + // round-trip proves the stage never wedged. + assoc + .send_datagram(&target, b"first") + .await + .expect("send the first"); + let (_, payload) = timeout(BOUND, assoc.recv_datagram()) + .await + .expect("recv within bound") + .expect("a datagram") + .expect("a datagram"); + assert_eq!(&payload[..], b"first"); + + tokio::time::sleep(Duration::from_millis(150)).await; + + assoc + .send_datagram(&target, b"second") + .await + .expect("send the second (the stage was parked, not spinning)"); + let (_, payload) = timeout(BOUND, assoc.recv_datagram()) + .await + .expect("recv within bound") + .expect("a datagram") + .expect("a datagram"); + assert_eq!(&payload[..], b"second"); + + assoc.shutdown().await.expect("half-close"); + no_leaked_channels(&topo).await; +} + +/// BIND end-to-end with the local accept: reply#1 carries the real +/// listener address; the app server (the in-test client) dials it; +/// reply#2 carries the real peer (byte-exact); the data plane +/// round-trips through the producer's `pump_bidi`. +#[tokio::test] +async fn bind_end_to_end_with_real_addresses_and_round_trip() { + let (dial, egress, accept) = local_providers(); + let topo = wire_consumer_side_with( + dial, + Some(egress), + Some(accept), + Some(both_scopes_identity()), + ) + .await; + + let mut session = BindSession::bind( + topo.consumer.as_ref(), + ("bind.test", 0u16).to_target_addr().unwrap(), + ) + .await + .expect("the session binds"); + + // Reply#1: the real bound address (the listener's local_addr). + let bind_addr = session.reply1().expect("reply#1's handle").bind_addr; + let TargetAddr::Ip(sa) = bind_addr else { + panic!("an IP BND.ADDR, got {bind_addr:?}") + }; + assert!( + sa.ip().is_loopback() && sa.port() != 0, + "reply#1 carries the real listener address: {sa}" + ); + + // The app server dials the advertised address; a byte arrives + // through the data plane (the quiet wait's accept resolved). + let mut client = timeout(BOUND, tokio::net::TcpStream::connect(sa)) + .await + .expect("dial within bound") + .expect("connect to the advertised address"); + + let mut plane = timeout(BOUND, session.await_inbound()) + .await + .expect("reply#2 within bound") + .expect("the plane exists") + .expect("the accept resolved"); + let peer = match plane.peer { + TargetAddr::Ip(peer_sa) => peer_sa, + other => panic!("an IP peer, got {other:?}"), + }; + assert_eq!(peer.ip(), sa.ip(), "the peer rides the same listener"); + assert_ne!( + peer.port(), + sa.port(), + "the peer is the client, not the listener" + ); + + // The data plane: bytes through the channel reach the dialing + // client and back (the producer's `pump_bidi` against the real + // accepted halves). + plane.write_all(b"inbound").await.expect("write"); + plane.flush().await.expect("flush"); + let mut buf = [0u8; 7]; + timeout(BOUND, client.read_exact(&mut buf)) + .await + .expect("read within bound") + .expect("read"); + assert_eq!(&buf, b"inbound"); + client.write_all(b"outbound").await.expect("echo"); + client.flush().await.expect("flush"); + let mut buf = [0u8; 8]; + timeout(BOUND, plane.read_exact(&mut buf)) + .await + .expect("read within bound") + .expect("read"); + assert_eq!(&buf, b"outbound"); + + // Both ends close (the producer's pump exits when the accepted + // halves' peer is gone — a real TCP stream stays open until its + // owner closes it, unlike the harness duplexes). + drop(plane); + drop(client); + assert!(session.close().await, "the adopted channel entry existed"); + no_leaked_channels(&topo).await; +} + +/// The accept-future-owns-its-listener contract, through the +/// producer's quiet wait: a client-gone BIND session (the consumer +/// vanishes mid-quiet-wait) tears the listener down with the session +/// — the advertised address stops accepting (the drop-witness probe: +/// a fresh re-bind of the same port succeeds, the listener was +/// dropped, not leaked). +#[tokio::test] +async fn client_gone_mid_quiet_wait_closes_the_listener() { + let (dial, egress, accept) = local_providers(); + let topo = wire_consumer_side_with( + dial, + Some(egress), + Some(accept), + Some(both_scopes_identity()), + ) + .await; + + let mut session = BindSession::bind( + topo.consumer.as_ref(), + ("gone.test", 0u16).to_target_addr().unwrap(), + ) + .await + .expect("the session binds"); + let TargetAddr::Ip(sa) = session.reply1().expect("reply#1's handle").bind_addr else { + panic!("an IP BND.ADDR") + }; + + // The client-gone teardown: drop the session mid-quiet-wait (the + // channel's write half EOFs — the producer's quiet wait exits on + // the EOF sentinel and the accept future drops, taking the + // listener with it). + drop(session); + + // The producer's teardown reaps the channel entry (poll past the + // reap race), then the advertised address must be re-bindable — + // the listener closed with the quiet wait's drop. + timeout(Duration::from_secs(5), async { + loop { + if topo + .producer_manager + .channel_ids() + .iter() + .all(|&id| id == 0) + { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("the producer's entry reaped"); + + // A held port would refuse the re-bind; the dropped listener + // frees it. (The OS may hand the port to the re-bind — the probe + // is that it is free either way.) + let mut rebound = None; + for _ in 0..50 { + match tokio::net::TcpListener::bind(sa).await { + Ok(listener) => { + rebound = Some(listener); + break; + } + Err(_) => tokio::time::sleep(Duration::from_millis(20)).await, + } + } + assert!( + rebound.is_some(), + "the listener closed with the client-gone teardown (the port freed): {sa}" + ); + no_leaked_channels(&topo).await; +}