From 29569d9afdc3ab43517406fd9f6c17f61d9a3a36 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Fri, 25 Sep 2026 14:23:18 +0000 Subject: [PATCH] =?UTF-8?q?local:=20the=20producer-side=20local=20backend?= =?UTF-8?q?=20=E2=80=94=20the=20producer's=20front=20door?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ProducerDoor (feature local, src/local/producer_door.rs): a real TcpListener accept loop handing each accepted connection to the landed Socks5Producer::serve_socks5_over — the NoAuth in-stream RFC conversation with the local seam providers dialing/egressing/accepting; no new protocol-layer code. - The accept path owns ADR-014's two knobs: the idle-first-byte peek (non-consuming) and the max-concurrent-sessions cap (try_acquire — an accept beyond the bound is refused door-side). - ProducerDoorGuard: the listener's lifetime is the caller's — graceful close() awaits the loop's end; Drop aborts best-effort (the alktunnels TunnelListener teardown posture). In-flight sessions are never force-closed. - Per-connection failures are logged + isolated (the loop continues). - The BIND gate on this path is the builder's bind_policy (no open-op identity here; the pump handler's identity-derived gate does not apply) — pinned by the 0x02-refusal test. - tests/local-backend-producer.rs (9): vanilla CONNECT round-trip on real sockets, faithful-RFC BIND (real reply#1/reply#2 addresses), UDP ASSOCIATE through the real egress, domain resolution at the seam, idle timeout, session cap, failure isolation, bind_policy(false). Verification: cargo test + --all-features green; clippy clean (default, all-features, wasm32 targets); cargo fmt --check clean; wasm checks green (backend is local-gated). --- src/lib.rs | 3 + src/local/mod.rs | 4 + src/local/producer_door.rs | 297 ++++++++++ .../implementation/local-backend-producer.md | 45 +- tests/local-backend-producer.rs | 508 ++++++++++++++++++ 5 files changed, 854 insertions(+), 3 deletions(-) create mode 100644 src/local/producer_door.rs create mode 100644 tests/local-backend-producer.rs diff --git a/src/lib.rs b/src/lib.rs index a07b1ca..3f78411 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -113,6 +113,9 @@ pub use wire::{addr_encoding_len, frame, frame_inner, DatagramReader, WireError, #[cfg(feature = "local")] pub use local::{bind_accept, tcp_dial, udp_egress, LocalEgress}; +#[cfg(feature = "local")] +pub use local::{ProducerDoor, ProducerDoorBuilder, ProducerDoorGuard}; + #[cfg(feature = "local")] pub use consumer::TokenClient; #[cfg(feature = "local")] diff --git a/src/local/mod.rs b/src/local/mod.rs index 6f29f44..469f32d 100644 --- a/src/local/mod.rs +++ b/src/local/mod.rs @@ -280,3 +280,7 @@ async fn local_bind_accept( #[cfg(test)] mod tests; + +mod producer_door; + +pub use producer_door::{ProducerDoor, ProducerDoorBuilder, ProducerDoorGuard}; diff --git a/src/local/producer_door.rs b/src/local/producer_door.rs new file mode 100644 index 0000000..462181a --- /dev/null +++ b/src/local/producer_door.rs @@ -0,0 +1,297 @@ +//! The producer-side local backend (feature `local`; producer.md +//! §Producer-Side Local Backend): the same door shape in the other +//! direction — the producer serves from a real `TcpListener`, so +//! vanilla clients reach a *locally-served* SOCKS5 resource (the +//! ssh `-D` capability hosted on the producer side). Each accepted +//! TCP connection is handed to the landed substrate-free entry +//! [`Socks5Producer::serve_socks5_over`] directly — the RFC +//! conversation runs on the accepted socket, the seams dial locally +//! (`tcp_dial()` / `udp_egress()` / `bind_accept()`). No new +//! protocol-layer code: this module is pure accept-loop assembly. +//! +//! **The door surface** (ADR-005): the NoAuth shape is the primary +//! one here — the producer IS the RFC conversation on this path +//! (`serve_socks5_over` runs `accept_no_auth` in-stream); there is no +//! door-side protocol half to re-originate. A cred-bearing variant +//! (RFC 1929) would wrap `accept_password_auth` before +//! `serve_socks5_over` — documented follow-up, not landed (the task +//! spec lands the NoAuth shape unless a test drives the other). +//! +//! **The accept path owns the idle defense + the session cap** +//! (ADR-014's accept-path defense; an accepted socket is not a +//! channel — the cap regime does not bound it): the idle-first-byte +//! peek (non-consuming — the producer's own greeting read picks the +//! byte up) and the max-concurrent-sessions cap (an accept beyond +//! the bound is refused door-side, the socket closed immediately, no +//! session spawned) are the same two knobs the consumer door carries +//! ([`crate::door::FrontDoor`]). +//! +//! **The BIND command gate** on this path is the builder's +//! `bind_policy` (the substrate-free entry's gate — ADR-006's +//! posture). The pump handler's identity-derived gate does **not** +//! apply here: this path has no open-op identity — the accepted +//! socket is the whole caller; whatever policy governs it is the +//! builder's explicit `bind_policy(bool)` and the two accept-path +//! knobs. +//! +//! **Lifetime** (the alktunnels `TunnelListener` teardown posture): +//! the accept loop runs until the caller ends it — [`ProducerDoor`] +//! hands back a guard whose `close()` awaits the loop's end; dropping +//! the guard is a best-effort abort (the spawned sessions own their +//! sockets; an in-flight session is not force-closed by a drop, only +//! `close()` is the graceful full stop). + +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use tokio::net::TcpListener; +use tokio::sync::{watch, Semaphore}; + +use crate::producer::Socks5Producer; + +/// Why one accepted connection's serving task ended (the backend's +/// per-connection teardown diagnostic — logged + isolated; the loop +/// never exits on a per-connection failure). +#[derive(Debug)] +enum BackendExit { + /// The vanilla client went away before/after the greeting. + ClientGone, + /// The RFC conversation ended (its `SocksError` — the mapped + /// refusal, the protocol garbage, or the session end). + Session(crate::SocksError), + /// A plain I/O failure on the accepted socket. + Io(std::io::Error), +} + +impl std::fmt::Display for BackendExit { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + BackendExit::ClientGone => f.write_str("client gone"), + BackendExit::Session(err) => write!(f, "session: {err:#}"), + BackendExit::Io(err) => write!(f, "io: {err}"), + } + } +} + +/// The producer-side local backend (feature `local`): a real +/// `TcpListener` front door for [`Socks5Producer`] — each accepted +/// TCP connection is served via `serve_socks5_over` (the NoAuth +/// in-stream RFC conversation; the seams dial locally). The accept +/// path owns the two ADR-014 defense knobs (the idle-first-byte +/// window + the max-concurrent-sessions cap). +pub struct ProducerDoor { + bind_addr: Option, + producer: Socks5Producer, + idle_first_byte: Duration, + max_sessions: usize, +} + +impl ProducerDoor { + /// The builder (the consumer door's sibling shape). + pub fn builder() -> ProducerDoorBuilder { + ProducerDoorBuilder::default() + } +} + +/// The builder for [`ProducerDoor`]. The producer and the bind +/// address are required (the explicit bind decision — never implied); +/// the two defense knobs carry the consumer door's defaults. +#[derive(Default)] +pub struct ProducerDoorBuilder { + producer: Option, + bind_addr: Option, + idle_first_byte: Option, + max_sessions: Option, +} + +impl ProducerDoorBuilder { + /// The producer whose `serve_socks5_over` serves every accepted + /// connection (the landed assembly shape — its seams dial/egress/ + /// accept; its `bind_policy` is this path's BIND gate). + pub fn producer(mut self, producer: Socks5Producer) -> Self { + self.producer = Some(producer); + self + } + + /// The local bind address — the backend's explicit bind decision + /// (the never-binds posture: the protocol layer never binds; this + /// is the assembly layer's spelled-out choice — loopback is the + /// documented posture). + pub fn bind(mut self, addr: impl Into) -> Self { + self.bind_addr = Some(addr.into()); + self + } + + /// The idle-first-byte window (ADR-014 knob 1): an accepted + /// connection that sends no first byte within this window is + /// closed door-side. The window covers the first byte of the RFC + /// conversation — the greeting (`05 ..`), read producer-side by + /// `serve_socks5_over`'s own handshake (the peek is non-consuming). + pub fn idle_first_byte(mut self, window: Duration) -> Self { + self.idle_first_byte = Some(window); + self + } + + /// The max-concurrent-sessions cap (ADR-014 knob 2): an accept + /// beyond the bound is refused door-side — the socket is closed + /// immediately, no session spawned. + pub fn max_sessions(mut self, cap: usize) -> Self { + self.max_sessions = Some(cap); + self + } + + pub fn build(self) -> ProducerDoor { + ProducerDoor { + bind_addr: self.bind_addr, + producer: self.producer.expect("the backend serves a producer"), + idle_first_byte: self.idle_first_byte.unwrap_or(Duration::from_secs(30)), + max_sessions: self.max_sessions.unwrap_or(256), + } + } +} + +/// The serving guard: [`ProducerDoor::serve`] returns one; `close()` +/// ends the accept loop gracefully (the listener is dropped, the +/// loop's end is awaited — the spawned sessions already running own +/// their sockets and finish on their own), and dropping it aborts the +/// loop task best-effort (the alktunnels `TunnelListener` teardown +/// posture). +pub struct ProducerDoorGuard { + done_rx: Option>, + task: Option>>, +} + +impl ProducerDoorGuard { + /// End the accept loop gracefully: the loop task returns, and the + /// guard awaits its end. In-flight sessions are NOT force-closed — + /// they own their sockets and finish with their clients (per- + /// connection isolation cuts both ways). Returns the loop task's + /// fatal outcome, if any (`None` = the loop ended by close). + pub async fn close(mut self) -> Option { + self.done_rx = None; + let task = self.task.take(); + drop(self); + match task { + Some(task) => task.await.ok().flatten(), + None => None, + } + } +} + +impl Drop for ProducerDoorGuard { + /// Best-effort abort: drop the loop task's handle (the loop ends + /// when its listener drops — the listener is owned by the task; + /// an abort cancels the accept wait). Sync-only, never awaits. + fn drop(&mut self) { + if let Some(task) = self.task.take() { + task.abort(); + } + } +} + +impl ProducerDoor { + /// Bind the configured address and run the accept loop. Per- + /// connection failures are logged + isolated — the loop continues + /// (the alktunnels per-accept isolation posture). The listener's + /// lifetime is the caller's: the returned [`ProducerDoorGuard`] + /// ends it (`close()` graceful; drop best-effort abort). + pub async fn serve(self) -> Result { + let ProducerDoor { + bind_addr, + producer, + idle_first_byte, + max_sessions, + } = self; + let bind_addr = bind_addr.expect("the backend binds an address (builder .bind)"); + let listener = TcpListener::bind(bind_addr).await?; + Ok(serve_on_parts( + listener, + idle_first_byte, + max_sessions, + producer, + )) + } + + /// Run the accept loop over an existing listener (the test/ + /// assembly seam — the same accept loop without the bind). + pub async fn serve_on(self, listener: TcpListener) -> ProducerDoorGuard { + let ProducerDoor { + producer, + idle_first_byte, + max_sessions, + .. + } = self; + serve_on_parts(listener, idle_first_byte, max_sessions, producer) + } +} + +fn serve_on_parts( + listener: TcpListener, + idle_first_byte: Duration, + max_sessions: usize, + producer: Socks5Producer, +) -> ProducerDoorGuard { + let (done_tx, done_rx) = watch::channel(()); + let task = tokio::spawn(async move { + let live = Arc::new(Semaphore::new(max_sessions)); + loop { + tokio::select! { + _ = done_tx.closed() => return None, + accepted = listener.accept() => { + let (stream, peer) = match accepted { + Ok(pair) => pair, + Err(err) => { + tracing::warn!(error = %err, "producer door: accept failed"); + continue; + } + }; + + // The session cap (ADR-014 knob 2): refuse + // beyond the bound — `try_acquire` keeps the + // accept loop free (a permit returns when a + // serving task ends and drops its guard). + let Ok(permit) = live.clone().try_acquire_owned() else { + tracing::warn!(peer = %peer, "producer door: session cap reached — refusing"); + drop(stream); + continue; + }; + + let producer = producer.clone(); + tokio::spawn(async move { + let _permit = permit; + let exit = serve_accepted(stream, &producer, idle_first_byte).await; + tracing::debug!(peer = %peer, "producer door: connection ended: {exit}"); + }); + } + } + } + }); + ProducerDoorGuard { + done_rx: Some(done_rx), + task: Some(task), + } +} + +/// Serve one accepted connection: the idle-first-byte defense first +/// (ADR-014 knob 1 — the peek waits for readability without +/// consuming; the producer's own greeting read picks the byte up), +/// then the substrate-free entry (`serve_socks5_over` — the NoAuth +/// in-stream RFC conversation; the seams dial locally). +async fn serve_accepted( + stream: tokio::net::TcpStream, + producer: &Socks5Producer, + idle: Duration, +) -> BackendExit { + let mut first = [0u8; 1]; + match tokio::time::timeout(idle, stream.peek(&mut first)).await { + Ok(Ok(_)) => {} + Ok(Err(err)) => return BackendExit::Io(err), + Err(_) => return BackendExit::ClientGone, + } + + match producer.serve_socks5_over(stream).await { + Ok(()) => BackendExit::ClientGone, + Err(err) => BackendExit::Session(err), + } +} diff --git a/tasks/implementation/local-backend-producer.md b/tasks/implementation/local-backend-producer.md index c91d350..c1c4eac 100644 --- a/tasks/implementation/local-backend-producer.md +++ b/tasks/implementation/local-backend-producer.md @@ -1,7 +1,7 @@ --- id: implementation/local-backend-producer name: Producer-side local backend — the producer's front door (vanilla clients → a locally-served SOCKS5 resource) -status: pending +status: completed depends_on: [implementation/local-providers] scope: narrow risk: low @@ -86,8 +86,47 @@ Deliverables: ## Notes -> To be filled by implementation agent +- **Symbol choice**: the constructor is `ProducerDoor` (+ `ProducerDoorBuilder`/ + `ProducerDoorGuard`) in `src/local/producer_door.rs` — the consumer door's + sibling name (`FrontDoor`/`UdpFrontDoor`), not `serve_local`: the surface is a + builder + accept-loop object (`serve()` binds, `serve_on(listener)` is the + test/assembly seam over an existing listener), matching the consumer door's + shape rather than a free function. +- **Door surface decision**: the NoAuth shape landed; the cred-bearing + producer-side door is a documented follow-up (no test drives it). On this + path the producer IS the RFC conversation (`serve_socks5_over` runs + `accept_no_auth` in-stream) — there is no door-side protocol half to + re-originate; a cred-bearing variant would wrap `accept_password_auth` + before handing the stream to `serve_socks5_over`. +- **Guard shape** (the alktunnels `TunnelListener` teardown posture): + `serve`/`serve_on` return a `ProducerDoorGuard`; `close()` ends the accept + loop gracefully and awaits the loop task; a Drop aborts it best-effort + (sync, never awaits). In-flight sessions are never force-closed by the + guard — they own their sockets (per-connection isolation cuts both ways). +- **BIND gate documentation**: the per-caller gate (the pump handler's + identity-derived one) does not apply here — this path has no open-op + identity; the builder's `bind_policy` is the whole gate, pinned by the + `bind_policy(false)` test (0x02 refusal, CONNECT still serves). +- **No protocol-layer code**: the accept loop rides + `Socks5Producer::serve_socks5_over` directly; the only additions are + accept-loop assembly (`peek` idle gate, `Semaphore` session cap, + `try_acquire_owned` door-side refusal) and the `watch`-signaled guard. ## Summary -> To be filled on completion \ No newline at end of file +Landed the producer-side local backend (feature `local`): `ProducerDoor` +(`src/local/producer_door.rs`, exported from the crate root) — a real +`TcpListener` front door handing every accepted connection to the landed +`Socks5Producer::serve_socks5_over` (NoAuth in-stream RFC conversation; +`local::tcp_dial()` / `udp_egress()` / `bind_accept()` dial locally). The +accept path owns ADR-014's two knobs (idle-first-byte peek + session-cap +semaphore, door-side refusals); per-connection failures are logged + +isolated; the `ProducerDoorGuard` carries the listener's lifetime +(graceful `close()` / best-effort abort Drop). Pinned by +`tests/local-backend-producer.rs` (9 tests): vanilla CONNECT round-trip on +real sockets, faithful-RFC BIND (real reply#1/reply#2 addresses), UDP +ASSOCIATE through the real egress, domain-target resolution at the seam, +the idle timeout, the session cap (refusal while live sessions serve), +failure isolation, and the `bind_policy(false)` gate. Verified: `cargo +test` + `--all-features` green, clippy clean (default/all/wasm targets), +`cargo fmt --check` clean, wasm checks stay green. \ No newline at end of file diff --git a/tests/local-backend-producer.rs b/tests/local-backend-producer.rs new file mode 100644 index 0000000..a6d558c --- /dev/null +++ b/tests/local-backend-producer.rs @@ -0,0 +1,508 @@ +//! The producer-side local backend's integration tests (feature +//! `local`; tasks/implementation/local-backend-producer.md): a vanilla +//! RFC 1928 client against the locally-served producer — the whole +//! stack on real sockets (`ProducerDoor` → `serve_socks5_over` → +//! `local::tcp_dial()` / `local::bind_accept()`), BIND through the +//! local backend (the faithful-RFC variant — real reply#1/reply#2 +//! addresses), the idle-first-byte timeout, the session cap, +//! per-connection failure isolation, and the `bind_policy(false)` +//! gate's RFC-faithful BIND refusal (0x02). +#![cfg(feature = "local")] +#![allow(dead_code)] + +use std::time::Duration; + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::time::timeout; + +use alksocks::local::ProducerDoor; +use alksocks::producer::Socks5Producer; +use alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr}; + +const BOUND: Duration = Duration::from_secs(10); +const IDLE: Duration = Duration::from_millis(300); + +/// Spawn a real echo TCP server on an ephemeral loopback port +/// (CONNECT's target — the dial seam dials it for real). +async fn echo_tcp_server() -> std::net::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 +} + +/// The producer under test: local socket-backed seams all three +/// (the whole stack on real sockets), BIND policy as given. +fn local_producer(bind_policy: Option) -> Socks5Producer { + let builder = Socks5Producer::builder() + .dial(alksocks::local::tcp_dial()) + .egress(alksocks::local::udp_egress()) + .accept(alksocks::local::bind_accept(std::net::IpAddr::from([ + 127, 0, 0, 1, + ]))); + let builder = match bind_policy { + Some(can) => builder.bind_policy(can), + None => builder, + }; + builder.build() +} + +/// Serve the producer behind a real listener (ephemeral loopback +/// port); returns the vanilla client's door address and the guard. +async fn serve( + producer: Socks5Producer, + idle: Duration, + cap: usize, +) -> (std::net::SocketAddr, alksocks::local::ProducerDoorGuard) { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("local_addr"); + let door = ProducerDoor::builder() + .producer(producer) + .idle_first_byte(idle) + .max_sessions(cap) + .build(); + let guard = door.serve_on(listener).await; + (addr, guard) +} + +/// A vanilla client: connect to the door, run the full NoAuth +/// greeting + CONNECT request, read the reply, and echo round-trip +/// through the dialed target. Returns the reply code. +async fn vanilla_connect( + door: std::net::SocketAddr, + target: std::net::SocketAddr, +) -> (TcpStream, u8, [u8; 6]) { + let mut stream = TcpStream::connect(door).await.expect("door connect"); + stream + .write_all(&[0x05, 0x01, 0x00]) + .await + .expect("greeting"); + let mut chosen = [0u8; 2]; + stream + .read_exact(&mut chosen) + .await + .expect("method selection"); + assert_eq!(chosen, [0x05, 0x00], "no-auth selected"); + + let mut req = vec![0x05, 0x01, 0x00, 0x01]; + req.extend_from_slice(&[127, 0, 0, 1]); + req.extend_from_slice(&target.port().to_be_bytes()); + stream.write_all(&req).await.expect("request"); + stream.flush().await.expect("flush"); + + let mut head = [0u8; 4]; + stream.read_exact(&mut head).await.expect("reply head"); + assert_eq!(head[0], 0x05, "reply is v5"); + assert_eq!(head[3], 0x01, "v4 BND.ADDR"); + let mut addr = [0u8; 6]; + stream.read_exact(&mut addr).await.expect("BND.ADDR"); + (stream, head[1], addr) +} + +#[tokio::test] +async fn vanilla_client_connect_round_trips_through_local_seams() { + let target = echo_tcp_server().await; + let (door, _guard) = serve(local_producer(None), IDLE, 16).await; + + let (mut stream, code, _addr) = timeout(BOUND, vanilla_connect(door, target)) + .await + .expect("round-trip"); + assert_eq!( + code, 0x00, + "the CONNECT resolved through the real dial seam" + ); + + // The data plane is raw pass-through to the real echo server. + stream.write_all(b"ping").await.expect("probe"); + let mut buf = [0u8; 4]; + timeout(BOUND, stream.read_exact(&mut buf)) + .await + .expect("echo") + .expect("read"); + assert_eq!(&buf, b"ping"); + + drop(stream); + _guard.close().await; +} + +/// BIND through the local backend — the faithful-RFC variant: a real +/// inbound peer connects to the reply#1-advertised listener, and +/// reply#2 carries the real peer address. +#[tokio::test] +async fn vanilla_client_bind_round_trips_with_real_addresses() { + let (door, _guard) = serve(local_producer(None), IDLE, 16).await; + + let mut client = TcpStream::connect(door).await.expect("door connect"); + client + .write_all(&[0x05, 0x01, 0x00]) + .await + .expect("greeting"); + let mut chosen = [0u8; 2]; + client + .read_exact(&mut chosen) + .await + .expect("method selection"); + assert_eq!(chosen, [0x05, 0x00]); + + // BIND for the ephemeral port (ATYP v4, port 0). + client + .write_all(&[0x05, 0x02, 0x00, 0x01, 0, 0, 0, 0, 0, 0]) + .await + .expect("bind request"); + + // Reply#1: success + the real bound listener address. + let mut head = [0u8; 4]; + client.read_exact(&mut head).await.expect("reply#1 head"); + assert_eq!(head[1], 0x00, "BIND reply#1 succeeded"); + let mut bnd = [0u8; 6]; + client.read_exact(&mut bnd).await.expect("reply#1 BND.ADDR"); + assert_eq!(bnd[0], 127, "the reply carries the real loopback bind"); + let bound_port = u16::from_be_bytes([bnd[4], bnd[5]]); + assert!(bound_port != 0, "a real listener was bound"); + + // The inbound peer connects to the advertised address. + let advertised = std::net::SocketAddr::from(([127, 0, 0, 1], bound_port)); + let mut peer = TcpStream::connect(advertised).await.expect("peer connects"); + peer.write_all(b"hello-bind").await.expect("peer writes"); + + // Reply#2: success + the peer's real address. + let mut head2 = [0u8; 4]; + client.read_exact(&mut head2).await.expect("reply#2 head"); + assert_eq!(head2[1], 0x00, "BIND reply#2 succeeded"); + let mut peer_addr = [0u8; 6]; + client + .read_exact(&mut peer_addr) + .await + .expect("reply#2 DST.ADDR"); + assert_eq!(peer_addr[0], 127, "reply#2 carries the real peer address"); + let peer_port = u16::from_be_bytes([peer_addr[4], peer_addr[5]]); + assert!(peer_port != 0, "the peer's ephemeral port is real"); + + // The data plane is raw pass-through to the accepted peer. + let mut buf = [0u8; 10]; + timeout(BOUND, client.read_exact(&mut buf)) + .await + .expect("data plane") + .expect("read"); + assert_eq!(&buf, b"hello-bind"); + + client.write_all(b"pong").await.expect("client writes"); + let mut echo = [0u8; 4]; + timeout(BOUND, peer.read_exact(&mut echo)) + .await + .expect("reverse data plane") + .expect("read"); + assert_eq!(&echo, b"pong"); + + drop(peer); + drop(client); + _guard.close().await; +} + +/// The idle-first-byte defense (ADR-014 knob 1): an accepted +/// connection that sends nothing within the window is closed +/// door-side — EOF arrives, no greeting ever answered. +#[tokio::test] +async fn idle_first_byte_timeout_closes_the_accepted_connection() { + let (door, _guard) = serve(local_producer(None), IDLE, 16).await; + let mut stream = TcpStream::connect(door).await.expect("door connect"); + + let mut buf = [0u8; 2]; + match timeout(IDLE * 4, stream.read(&mut buf)).await { + Err(_) => panic!("the idle bound must fire"), + Ok(Ok(0)) | Ok(Err(_)) => {} + Ok(Ok(n)) => panic!("EOF expected after the idle bound — got {n} bytes"), + } + _guard.close().await; +} + +/// The session cap (ADR-014 knob 2): an accept beyond the bound is +/// refused door-side (immediate EOF), while the sessions under the +/// bound keep serving. +#[tokio::test] +async fn session_cap_refuses_beyond_the_bound_while_live_sessions_serve() { + let target = echo_tcp_server().await; + let (door, _guard) = serve(local_producer(None), IDLE, 1).await; + + // One live session fills the cap (holding the reply open). + let (mut held, code, _) = timeout(BOUND, vanilla_connect(door, target)) + .await + .expect("held session"); + assert_eq!(code, 0x00, "the cap's first session serves"); + + // The second accept is beyond the bound: closed immediately. + let mut refused = TcpStream::connect(door).await.expect("door connect"); + let mut buf = [0u8; 2]; + match timeout(BOUND, refused.read(&mut buf)).await { + Err(_) => panic!("the capped accept must end"), + Ok(Ok(0)) | Ok(Err(_)) => {} + Ok(Ok(n)) => panic!("the capped accept is refused — got {n} bytes"), + } + + // The held session still serves (isolation: the refusal did not + // touch it). + held.write_all(b"still-alive").await.expect("held writes"); + let mut echo = [0u8; 11]; + timeout(BOUND, held.read_exact(&mut echo)) + .await + .expect("held echoes") + .expect("read"); + assert_eq!(&echo, b"still-alive"); + + drop(held); + _guard.close().await; +} + +/// Per-connection failure isolation: a connection speaking protocol +/// garbage ends loud door-side while the loop continues — the next +/// connection on the same door serves normally. +#[tokio::test] +async fn per_connection_failure_is_isolated_and_the_loop_continues() { + let target = echo_tcp_server().await; + let (door, _guard) = serve(local_producer(None), IDLE, 16).await; + + // Garbage: a valid greeting, then a malformed command head. + let mut bad = TcpStream::connect(door).await.expect("door connect"); + bad.write_all(&[0x05, 0x01, 0x00]).await.expect("greeting"); + let mut chosen = [0u8; 2]; + bad.read_exact(&mut chosen).await.expect("method selection"); + bad.write_all(&[0x05, 0x01, 0x00, 0xFF]) + .await + .expect("garbage head"); + drop(bad); + tokio::time::sleep(Duration::from_millis(50)).await; + + // The loop continues: a well-formed client serves normally. + let (mut stream, code, _) = timeout(BOUND, vanilla_connect(door, target)) + .await + .expect("next session"); + assert_eq!(code, 0x00, "the loop survived the failed connection"); + + stream.write_all(b"ping").await.expect("probe"); + let mut buf = [0u8; 4]; + timeout(BOUND, stream.read_exact(&mut buf)) + .await + .expect("echo") + .expect("read"); + assert_eq!(&buf, b"ping"); + + drop(stream); + _guard.close().await; +} + +/// The `bind_policy(false)` gate is honored on this path (ADR-006's +/// posture): BIND refuses RFC-faithfully with 0x02 — the command is +/// known, the caller is not allowed — while CONNECT still serves. +#[tokio::test] +async fn bind_policy_false_refuses_bind_rfc_faithfully_but_connect_serves() { + let target = echo_tcp_server().await; + let (door, _guard) = serve(local_producer(Some(false)), IDLE, 16).await; + + // BIND → 0x02 (ConnectionNotAllowed), through the producer's own + // in-stream gate (no open-op identity here — the builder's + // bind_policy is the whole gate on this path). + let mut client = TcpStream::connect(door).await.expect("door connect"); + client + .write_all(&[0x05, 0x01, 0x00]) + .await + .expect("greeting"); + let mut chosen = [0u8; 2]; + client + .read_exact(&mut chosen) + .await + .expect("method selection"); + client + .write_all(&[0x05, 0x02, 0x00, 0x01, 0, 0, 0, 0, 0, 0]) + .await + .expect("bind request"); + let mut head = [0u8; 4]; + client.read_exact(&mut head).await.expect("reply head"); + assert_eq!(head[1], 0x02, "bind_policy(false) → ConnectionNotAllowed"); + drop(client); + + // CONNECT still serves through the same closed-gate producer. + let (stream, code, _) = timeout(BOUND, vanilla_connect(door, target)) + .await + .expect("connect session"); + assert_eq!(code, 0x00, "the gate closes BIND alone"); + + drop(stream); + _guard.close().await; +} + +/// The default BIND posture holds too: with the accept seam wired and +/// no explicit policy, BIND is permitted (the backward-compat +/// reading) and serves through the real listener seam. +#[tokio::test] +async fn default_bind_posture_serves_through_the_accept_seam() { + let (door, _guard) = serve(local_producer(None), IDLE, 16).await; + + let mut client = TcpStream::connect(door).await.expect("door connect"); + client + .write_all(&[0x05, 0x01, 0x00]) + .await + .expect("greeting"); + let mut chosen = [0u8; 2]; + client + .read_exact(&mut chosen) + .await + .expect("method selection"); + client + .write_all(&[0x05, 0x02, 0x00, 0x01, 0, 0, 0, 0, 0, 0]) + .await + .expect("bind request"); + let mut head = [0u8; 4]; + client.read_exact(&mut head).await.expect("reply#1 head"); + assert_eq!(head[1], 0x00, "the default posture permits BIND"); + drop(client); + _guard.close().await; +} + +/// UDP ASSOCIATE through the local backend: the sentinel reply rides +/// the door, the datagram stage relays through the real UDP egress +/// (a real loopback UDP echo target). +#[tokio::test] +async fn vanilla_client_udp_associate_relays_through_the_local_egress() { + let sock = tokio::net::UdpSocket::bind("127.0.0.1:0") + .await + .expect("bind"); + let udp_target = 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; + } + } + }); + + let (door, _guard) = serve(local_producer(None), IDLE, 16).await; + + let mut client = TcpStream::connect(door).await.expect("door connect"); + client + .write_all(&[0x05, 0x01, 0x00]) + .await + .expect("greeting"); + let mut chosen = [0u8; 2]; + client + .read_exact(&mut chosen) + .await + .expect("method selection"); + client + .write_all(&[0x05, 0x03, 0x00, 0x01, 0, 0, 0, 0, 0, 0]) + .await + .expect("associate request"); + let mut head = [0u8; 4]; + client.read_exact(&mut head).await.expect("reply head"); + assert_eq!( + head[1], 0x00, + "the associate resolved through the real egress seam" + ); + let mut bnd = [0u8; 6]; + client.read_exact(&mut bnd).await.expect("BND.ADDR"); + assert_eq!(&bnd, &[127, 0, 0, 1, 0, 0], "the sentinel BND.ADDR"); + + // One datagram frame (the in-band wire shape) echoes through the + // real UDP egress. + let dst = TargetAddr::Ip(udp_target); + let frame = alksocks::wire::frame(&dst, b"udp-probe").expect("frame"); + client.write_all(&frame).await.expect("datagram frame"); + + let mut reply = vec![0u8; frame.len()]; + timeout(BOUND, client.read_exact(&mut reply)) + .await + .expect("echo reply") + .expect("read"); + // The reply's inner carries the *sender's* address (the echo's + // source is the egress socket, not the target) — decode and match + // the payload only. + let n = u16::from_be_bytes([reply[0], reply[1]]) as usize; + let inner = &reply[2..2 + n]; + let atyp = inner[0]; + let addr_len = match atyp { + 0x01 => 1 + 4 + 2, + 0x03 => 1 + 1 + inner[1] as usize + 2, + 0x04 => 1 + 16 + 2, + other => panic!("unexpected ATYP {other}"), + }; + assert_eq!(&inner[addr_len..], b"udp-probe", "the datagram echoed"); + + drop(client); + _guard.close().await; +} + +/// The door's address target form: a domain-form CONNECT target +/// resolves at the seam (the ssh `-D` semantic — ADR-012) — the echo +/// server is reachable as `localhost`. +#[tokio::test] +async fn domain_form_targets_resolve_at_the_seam() { + let target = echo_tcp_server().await; + let (door, _guard) = serve(local_producer(None), IDLE, 16).await; + + let mut client = TcpStream::connect(door).await.expect("door connect"); + client + .write_all(&[0x05, 0x01, 0x00]) + .await + .expect("greeting"); + let mut chosen = [0u8; 2]; + client + .read_exact(&mut chosen) + .await + .expect("method selection"); + + let domain = "localhost".to_owned(); + let mut req = vec![0x05, 0x01, 0x00, 0x03, domain.len() as u8]; + req.extend_from_slice(domain.as_bytes()); + req.extend_from_slice(&target.port().to_be_bytes()); + client.write_all(&req).await.expect("request"); + client.flush().await.expect("flush"); + + let mut head = [0u8; 4]; + client.read_exact(&mut head).await.expect("reply head"); + assert_eq!(head[1], 0x00, "the domain resolved at the seam"); + let mut addr = [0u8; 6]; + client.read_exact(&mut addr).await.expect("BND.ADDR"); + + client.write_all(b"dom").await.expect("probe"); + let mut buf = [0u8; 3]; + timeout(BOUND, client.read_exact(&mut buf)) + .await + .expect("echo") + .expect("read"); + assert_eq!(&buf, b"dom"); + + drop(client); + _guard.close().await; +} + +/// The `ToTargetAddr` import stays exercised (the compile-time shape +/// the domain test rides). +#[allow(dead_code)] +fn _target_addr_shape() { + let _: TargetAddr = std::net::SocketAddr::from(([127, 0, 0, 1], 1)) + .to_target_addr() + .expect("encode"); +}