diff --git a/src/consumer.rs b/src/consumer.rs index 7a4cc2e..01ce393 100644 --- a/src/consumer.rs +++ b/src/consumer.rs @@ -58,11 +58,13 @@ //! 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::client::{ChannelClient, ChannelOpenError}; use alkcall::channels::manager::ChannelManager; use alkcall::core::types::BiStream; +use alkcall::protocol::connection::CallConnection; use bytes::Bytes; use std::net::{IpAddr, Ipv6Addr, SocketAddr}; +use std::sync::Arc; use tokio::io::{AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf}; use crate::error::{SocksOpenError, SocksSessionError}; @@ -107,6 +109,51 @@ where } } +/// The cred-bearing front door's re-origination prologue (the +/// token-carrying twin of [`open_and_negotiate`]): open the channel +/// with the payload `auth_token` riding the `call.requested` payload +/// (CF-005 precedence (1) — the producer-side dispatch resolves the +/// token through its `ServingConfig.identity_provider` and runs the +/// ACL gate against the mapped identity), join the halves, run the +/// command negotiation, and reap on failure — the same phantom-channel +/// property. `Ok` yields the channel id, the negotiated stream, and +/// the cloned manager. +/// +/// The token path needs the raw `call.requested` payload (`auth_token` +/// is a payload-level field, ADR-017 §7), so the open rides +/// `CallConnection::call_with_payload` + `adopt_channel` — the same +/// two-step `open_channel` performs, with the token added. The +/// client's call connection is taken once and cached in the returned +/// client (the door's `TokenClient` owns it for the door's lifetime). +async fn open_with_token_and_negotiate( + client: &crate::consumer::TokenClient, + token: Vec, + negotiate: F, +) -> Result<(u32, T, ChannelManager), SocksSessionError> +where + F: FnOnce(Socks5Stream) -> Fut, + Fut: std::future::Future>, +{ + let channel_id = client.open_with_token(token).await?; + + let (send, recv) = client + .manager() + .adopt_channel(channel_id, SOCKS5_ALPN, None) + .await + .map_err(|e| SocksOpenError::from(ChannelOpenError::AdoptFailed(e)))?; + let bidi = BiStream::from_joined(recv, send); + + let negotiated = negotiate(Socks5Stream::use_stream(bidi, None, Config::default()).await?); + + match negotiated.await { + Ok(inner) => Ok((channel_id, inner, client.manager().clone())), + Err(err) => { + let _ = client.manager().teardown_channel(channel_id); + Err(err) + } + } +} + /// The spent-session marker (the gate review's R-2, /// tasks/implementation/review-protocol-core.md): [`Socks5ChannelSession /// ::into_inner`] / [`AssociateSession::into_inner`] return it when @@ -136,6 +183,11 @@ pub struct Socks5ChannelSession { channel_id: u32, manager: Option, inner: Option>, + /// The CONNECT reply's BND.ADDR — captured at `connect` (the + /// forked client `request()` returns it once; the front door's + /// re-origination re-encodes the vanilla reply from it — the RFC + /// codec at the door boundary only, ADR-003). + reply_addr: Option, } impl std::fmt::Debug for Socks5ChannelSession { @@ -165,15 +217,45 @@ impl Socks5ChannelSession { ) -> Result { let target: TargetAddr = target.to_target_addr()?; let (channel_id, inner, manager) = open_and_negotiate(client, |mut inner| async move { - inner.request(Socks5Command::TCPConnect, target).await?; - Ok(inner) + let reply_addr = inner.request(Socks5Command::TCPConnect, target).await?; + Ok((reply_addr, inner)) }) .await?; + let (reply_addr, inner) = inner; Ok(Self { channel_id, manager: Some(manager), inner: Some(inner), + reply_addr: Some(reply_addr), + }) + } + + /// The cred-bearing front door's CONNECT re-origination (the + /// token-carrying twin of [`Self::connect`]): the open rides the + /// payload `auth_token` (CF-005 precedence (1) — the producer's + /// identity provider resolves the door's mapped identity and its + /// scopes gate the open), then the same in-stream negotiation over + /// the channel's `BiStream`. Error surface identical. + pub async fn connect_with_token( + client: &TokenClient, + target: impl ToTargetAddr, + token: Vec, + ) -> Result { + let target: TargetAddr = target.to_target_addr()?; + let (channel_id, inner, manager) = + open_with_token_and_negotiate(client, token, |mut inner| async move { + let reply_addr = inner.request(Socks5Command::TCPConnect, target).await?; + Ok((reply_addr, inner)) + }) + .await?; + let (reply_addr, inner) = inner; + + Ok(Self { + channel_id, + manager: Some(manager), + inner: Some(inner), + reply_addr: Some(reply_addr), }) } @@ -183,6 +265,16 @@ impl Socks5ChannelSession { self.channel_id } + /// The CONNECT reply's BND.ADDR — the in-band reply's bound + /// address, captured at `connect` (the forked client `request()` + /// returns it once). The front door's re-origination re-encodes + /// the vanilla reply from it (the RFC codec at the door boundary + /// only, ADR-003); `None` after the pipe was taken (the + /// spent-session posture). + pub fn reply_addr(&self) -> Option { + self.reply_addr.clone() + } + /// The post-CONNECT pipe as split read/write halves (0 B framing /// — raw bytes to/from the target). The halves borrow the /// session: teardown stays owned (`close`/`Drop`). `None` after @@ -346,6 +438,39 @@ impl AssociateSession { }) } + /// The cred-bearing front door's ASSOCIATE re-origination (the + /// token-carrying twin of [`Self::associate`]): the open rides + /// the payload `auth_token` (CF-005 precedence (1)). The TCP door + /// uses this only for its RFC-faithful UDP refusal probe (the + /// door's datagram plane is the UDP door's task); the session is + /// otherwise identical. + pub async fn associate_with_token( + client: &TokenClient, + token: Vec, + ) -> Result { + let (channel_id, inner, manager) = + open_with_token_and_negotiate(client, token, |mut inner| async move { + let relay_addr = inner + .request( + Socks5Command::UDPAssociate, + TargetAddr::Ip(SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 0)), + ) + .await?; + Ok((relay_addr, inner)) + }) + .await?; + let (relay_addr, inner) = inner; + + Ok(Self { + channel_id, + manager: Some(manager), + inner: Some(inner), + reader: DatagramReader::new(), + pending: std::collections::VecDeque::new(), + relay_addr: Some(relay_addr), + }) + } + /// The adopted channel's id (observability + correlation; one /// session = one channel). pub fn channel_id(&self) -> u32 { @@ -751,6 +876,35 @@ impl BindSession { }) } + /// The cred-bearing front door's BIND re-origination (the + /// token-carrying twin of [`Self::bind`]): the open rides the + /// payload `auth_token` (CF-005 precedence (1) — the producer's + /// identity provider resolves the door's mapped identity; the + /// BIND command gate reads `socks5:bind` from it, ADR-006), then + /// the same in-stream BIND phase model over the channel's + /// `BiStream`. Error surface identical. + pub async fn bind_with_token( + client: &TokenClient, + target: impl ToTargetAddr, + token: Vec, + ) -> Result { + let target: TargetAddr = target.to_target_addr()?; + let (channel_id, inner, manager) = + open_with_token_and_negotiate(client, token, |mut inner| async move { + let bind_addr = inner.request(Socks5Command::TCPBind, target).await?; + Ok((bind_addr, inner)) + }) + .await?; + let (bind_addr, inner) = inner; + + Ok(Self { + channel_id, + manager: Some(manager), + inner: Some(inner), + bind_addr: Some(bind_addr), + }) + } + /// The adopted channel's id (observability + correlation; one /// session = one channel). pub fn channel_id(&self) -> u32 { @@ -900,6 +1054,82 @@ where Ok(peer) } +/// The cred-bearing front door's channel surface (the +/// token-carrying twin of a `ChannelClient`'s open path): wraps the +/// client's channel-0 [`CallConnection`] so an open call can carry the +/// payload `auth_token` (CF-005 precedence (1) — the producer-side +/// dispatch resolves the token through its +/// `ServingConfig.identity_provider` and runs the ACL gate against the +/// resolved identity, ADR-017 §7). The client's call connection is +/// taken once at construction (the client's own calling surface ends +/// — a door client is a door client); the manager stays shared for +/// `adopt_channel`/teardown. +/// +/// The token is the credential table's *identity id* (the resolution +/// seam's vocabulary — the SOCKS password itself never rides it, and +/// the vanilla client never sees the identity). The door's producer +/// resolves `token → Identity` through its +/// `ServingConfig.identity_provider`; the identity's scopes govern the +/// open gate and the BIND command gate exactly as any per-call opener +/// identity does (CF-005/CF-006). +pub struct TokenClient { + call: Arc, + manager: ChannelManager, +} + +impl TokenClient { + /// Wrap a `ChannelClient`: take its call connection (the open-op + /// calling surface) and clone the manager. The door's + /// token-bearing sessions then ride this client. + pub async fn wrap(client: &ChannelClient) -> Self { + let call = client + .take_call_connection() + .await + .expect("the channel client's call connection exists until taken"); + Self { + call, + manager: client.manager().clone(), + } + } + + /// The manager (adopt/teardown — the token sessions' prologue uses + /// it; the sessions' teardown owns it per session). + pub fn manager(&self) -> &ChannelManager { + &self.manager + } + + /// The raw open call with the `auth_token` payload (CF-005 + /// precedence (1)): the payload shape mirrors + /// [`ChannelClient::call_open_op`]'s `call.requested` envelope with + /// the token added (ADR-017 §7 — the hub-forwarding path's shape). + /// Returns the adopted `channel_id` (the reply's `channel_id` + /// extracted — the adopt is the caller's next step, mirroring + /// `open_channel`'s two halves). + pub async fn open_with_token(&self, token: Vec) -> Result { + let payload = serde_json::json!({ + "operationId": OP_SOCKS5_OPEN, + "input": Socks5Params::as_json(), + "auth_token": String::from_utf8_lossy(&token).into_owned(), + }); + let response = self.call.call_with_payload(payload).await; + let out = response + .result + .map_err(|error| ChannelOpenError::CallFailed { error }) + .map_err(SocksOpenError::from)?; + let channel_id = out + .get("channel_id") + .and_then(|v: &serde_json::Value| v.as_u64()) + .ok_or(ChannelOpenError::MissingChannelId)? as u32; + Ok(channel_id) + } +} + +impl std::fmt::Debug for TokenClient { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("TokenClient").finish_non_exhaustive() + } +} + #[cfg(test)] mod tests { use super::*; @@ -947,6 +1177,7 @@ mod tests { channel_id: 7, manager: None, inner: None, + reply_addr: None, }; let rendered = format!("{session:?}"); assert!(rendered.contains("channel_id: 7"), "{rendered}"); @@ -959,6 +1190,7 @@ mod tests { channel_id: 9, manager: None, inner: None, + reply_addr: None, }; assert!(!session.reap(), "no manager — nothing reaped"); assert!(session.split().is_none(), "no pipe after teardown"); @@ -973,6 +1205,7 @@ mod tests { channel_id: 9, manager: None, inner: None, + reply_addr: None, }; assert!( session.into_inner().is_err(), diff --git a/src/door.rs b/src/door.rs new file mode 100644 index 0000000..1b9b228 --- /dev/null +++ b/src/door.rs @@ -0,0 +1,652 @@ +//! The consumer's TCP front door (feature `local`, ADR-005/ADR-014; +//! consumer.md §The Front Doors): the ssh `-D` local exposure — a real +//! `socks5://127.0.0.1:port` endpoint vanilla clients (curl, browsers, +//! tun2proxy) point at. The door is the vanilla↔ACL boundary; the RFC +//! codec lives only here (ADR-003's "codec at front doors only"). +//! +//! Two door variants (ADR-005), selected by constructor config: +//! +//! - **The NoAuth door** (POC #1's raw-pipe shape): per accepted +//! connection, a literal raw pipe onto a channel `BiStream` +//! (`copy_bidirectional`, **zero protocol translation**). The +//! producer runs the whole RFC conversation through the pipe — the +//! door adds nothing. The loopback-only / wrapper-aware shape. +//! - **The cred-bearing door** (RFC 1929): the door terminates the +//! local RFC conversation itself — the forked +//! `accept_password_auth` shape (`start` → +//! `negotiate_auth(&[PasswordAuthentication])` → +//! `read_username_password` → check → `accept`/`reject`) — then +//! maps the accepted credential through the **static config table** +//! (`user:pass →` channel identity, ADR-005) and **re-originates +//! the SOCKS conversation in-band** via the token-carrying consumer +//! sessions (`Socks5ChannelSession::connect_with_token` / +//! `BindSession::bind_with_token`), re-encoding the in-band replies +//! to the vanilla client and piping the post-auth data-plane bytes +//! raw. **The door never pipes raw post-auth**: the vanilla +//! client's command bytes were consumed by the door's own RFC 1929 +//! half; the in-band path re-speaks them from the parsed target. +//! The SOCKS password never crosses the channel as a SOCKS +//! password; the channel identity never reaches the vanilla client. +//! +//! **The identity mechanism** (the CF-005 seam; decided at +//! implementation, recorded in the task Notes): the credential→identity +//! mapping yields an alkcall [`Identity`] (scopes included); the +//! per-credential identity rides the open-op call's `auth_token` +//! payload (CF-005 precedence (1): payload `auth_token` → +//! `identity_provider` resolution — alkcall `dispatch:: +//! resolve_identity`, ADR-017 §7), never a per-credential +//! `ChannelClient`. The token is the mapped identity's id — the +//! resolution seam's vocabulary; the password itself never rides it. +//! The producer resolves it through its `ServingConfig +//! .identity_provider` and the ACL gate reads the resolved scopes +//! exactly as for any per-call opener identity (CF-005/CF-006). The +//! door checks credentials itself — no forked `AuthMethod` impl +//! (ADR-005). +//! +//! **The accept path owns the idle defense** (ADR-014): an accepted +//! connection that sends no first byte within the configured window +//! is closed door-side (an accepted socket is not a channel — the cap +//! regime does not bound it); plus the max-concurrent-sessions cap +//! (an accept beyond the bound is refused door-side, no channel +//! opened). The two v1 knobs are constructor config. +//! +//! **The BIND door**: the same listener shape; a vanilla BIND +//! conversation rides [`BindSession`] (reply#1's advertised address +//! passes through to the vanilla client unchanged — the conversation +//! is the client's own; the door adds nothing, consumer.md §The BIND +//! door's reply#2 fidelity). The BIND door rides the same door +//! variant split (NoAuth raw-pipe vs cred-bearing re-origination). +//! +//! Wiring (consumer.md §Wiring): `FrontDoor::builder() +//! .credentials(cred_table).bind("127.0.0.1:1080").build()` → +//! `serve(channel_client)` — the NoAuth door omits `.credentials`. +//! `serve` runs the accept loop; per-connection failures are logged + +//! isolated (the loop continues — the alktunnels per-accept isolation +//! posture). + +use std::collections::HashMap; +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::sync::Arc; +use std::time::Duration; + +use alkcall::channels::client::ChannelClient; +use alkcall::core::auth::Identity; +use tokio::io::AsyncWriteExt as _; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::Semaphore; + +use crate::consumer::{TakenInner, TokenClient}; +use crate::rfc1928::consts; +use crate::rfc1928::server::{new_reply, states, Socks5ServerProtocol, SocksServerError}; +use crate::rfc1928::util::target_addr::TargetAddr; +use crate::rfc1928::{ReplyError, Socks5Command, SocksError}; + +/// One static credential-table entry (ADR-005): `user:pass →` channel +/// identity. The table is the door's credential→identity mapping — +/// static, declarative, assembly-layer. +#[derive(Clone, Debug)] +pub struct DoorCredential { + /// The RFC 1929 username the vanilla client presents. + pub username: String, + /// The RFC 1929 password the vanilla client presents. + pub password: String, + /// The channel identity the accepted credential maps to (scopes + /// included — the open gate and the BIND command gate read them + /// producer-side). The identity's `id` doubles as the open-op + /// `auth_token` (the resolution seam's vocabulary — the password + /// never rides it). + pub identity: Identity, +} + +/// The credential→identity static config table (ADR-005): exact +/// `username`/`password` lookup. A miss rejects the RFC 1929 +/// conversation door-side (no channel opened — the ACL boundary held; +/// the vanilla client got the RFC 1929 rejection). +#[derive(Clone, Debug, Default)] +pub struct CredentialTable { + entries: HashMap<(String, String), Identity>, +} + +impl CredentialTable { + /// Build the table from [`DoorCredential`] entries. + pub fn new(entries: impl IntoIterator) -> Self { + Self { + entries: entries + .into_iter() + .map(|c| ((c.username, c.password), c.identity)) + .collect(), + } + } + + /// The exact-match lookup: `(username, password) → Option`. + pub fn lookup(&self, username: &str, password: &str) -> Option<&Identity> { + self.entries + .get(&(username.to_string(), password.to_string())) + } + + /// The number of entries (observability). + pub fn len(&self) -> usize { + self.entries.len() + } + + /// Whether the table is empty (an empty cred-bearing table + /// rejects every conversation door-side). + pub fn is_empty(&self) -> bool { + self.entries.is_empty() + } +} + +/// Why one accepted connection's serving task ended (the door's +/// per-connection teardown diagnostic — logged + isolated). +#[derive(Debug)] +enum DoorExit { + /// The vanilla client went away (clean end, both variants). + ClientGone, + /// The data plane (local pipe ↔ in-band session) ended. + PlaneEnded, + /// The local RFC half failed (a rejected credential, a malformed + /// conversation, protocol garbage) — the vanilla client's fault. + LocalRfc(SocksError), + /// The in-band open failed (the producer refused the mapped + /// identity, the target dial refused, the channel open errored). + Open(String), + /// A plain I/O failure on the local socket. + Io(std::io::Error), +} + +impl std::fmt::Display for DoorExit { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + DoorExit::ClientGone => f.write_str("client gone"), + DoorExit::PlaneEnded => f.write_str("data plane ended"), + DoorExit::LocalRfc(err) => write!(f, "local rfc: {err}"), + DoorExit::Open(msg) => write!(f, "open failed: {msg}"), + DoorExit::Io(err) => write!(f, "io: {err}"), + } + } +} + +impl From for DoorExit { + fn from(err: SocksServerError) -> Self { + DoorExit::LocalRfc(SocksError::ServerError(err)) + } +} + +/// The TCP front door (feature `local`): the vanilla↔ACL boundary. +/// Variant selection is constructor config (ADR-005): credentials +/// configured = the cred-bearing door (RFC 1929 terminates locally, +/// re-origination in-band); credentials absent = the NoAuth door (the +/// literal raw pipe). The accept path's defense knobs (ADR-014) are +/// constructor config: the idle-first-byte window and the +/// max-concurrent-sessions cap. +pub struct FrontDoor { + bind_addr: SocketAddr, + credentials: Option, + idle_first_byte: Duration, + max_sessions: usize, +} + +impl FrontDoor { + /// The builder (consumer.md §Wiring). + pub fn builder() -> FrontDoorBuilder { + FrontDoorBuilder::default() + } +} + +/// The builder for [`FrontDoor`] (consumer.md §Wiring: `builder() +/// .credentials(cred_table).bind("127.0.0.1:1080").build()`). The +/// NoAuth door omits `.credentials`. +#[derive(Default)] +pub struct FrontDoorBuilder { + bind_addr: Option, + credentials: Option, + idle_first_byte: Option, + max_sessions: Option, +} + +impl FrontDoorBuilder { + /// The credential→identity static config table (ADR-005). + /// Configuring it selects the cred-bearing door (RFC 1929); + /// omitting it selects the NoAuth door (the raw-pipe shape). + pub fn credentials(mut self, table: CredentialTable) -> Self { + self.credentials = Some(table); + self + } + + /// The local bind address — the door's explicit bind decision + /// (the never-binds posture: the door 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's accept-path defense + /// 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 local conversation — the NoAuth door's + /// greeting bytes (read producer-side through the pipe) and the + /// cred-bearing door's greeting (read by the door's own RFC 1929 + /// half). + pub fn idle_first_byte(mut self, window: Duration) -> Self { + self.idle_first_byte = Some(window); + self + } + + /// The max-concurrent-sessions cap (ADR-014's accept-path defense + /// knob 2): an accept beyond the bound is refused door-side — the + /// socket is closed immediately, no channel opened. + pub fn max_sessions(mut self, cap: usize) -> Self { + self.max_sessions = Some(cap); + self + } + + pub fn build(self) -> FrontDoor { + FrontDoor { + bind_addr: self.bind_addr.expect("the door binds an address"), + credentials: self.credentials, + idle_first_byte: self.idle_first_byte.unwrap_or(Duration::from_secs(30)), + max_sessions: self.max_sessions.unwrap_or(256), + } + } +} + +impl FrontDoor { + /// Bind the configured address and run the accept loop + /// (consumer.md §Wiring: `door.serve(channel_client)`). Per- + /// connection failures are logged + isolated — the loop continues + /// (the alktunnels per-accept isolation posture). Runs until the + /// listener fails fatally; the caller owns the task's lifetime. + pub async fn serve(self, client: Arc) -> std::io::Result<()> { + let listener = TcpListener::bind(self.bind_addr).await?; + tracing::info!(bind = %self.bind_addr, "front door: listening"); + self.serve_on(listener, client).await + } + + /// 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, + client: Arc, + ) -> std::io::Result<()> { + let credentials = self.credentials; + let idle = self.idle_first_byte; + let live = Arc::new(Semaphore::new(self.max_sessions)); + + // The cred-bearing door's channel surface: the client's call + // connection is taken once for the door's lifetime (one + // transport, one call connection, many per-credential + // identities riding the open-op `auth_token` payloads). The + // NoAuth door keeps the client's own calling surface + // (`open_channel` — the raw-pipe shape never carries a + // token). + let token_client = if credentials.is_some() { + Some(Arc::new(TokenClient::wrap(&client).await)) + } else { + None + }; + + loop { + let (stream, peer) = match listener.accept().await { + Ok(accepted) => accepted, + Err(err) => { + tracing::warn!(error = %err, "front 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, "front door: session cap reached — refusing"); + drop(stream); + continue; + }; + + let client = Arc::clone(&client); + let per_accept_credentials = credentials.clone(); + let per_accept_tokens = token_client.clone(); + tokio::spawn(async move { + let _permit = permit; + let exit = serve_connection( + stream, + peer, + &client, + per_accept_credentials.as_ref(), + per_accept_tokens.as_ref(), + idle, + ) + .await; + tracing::debug!(peer = %peer, "front door: connection ended: {exit}"); + }); + } + } +} + +/// Serve one accepted connection through the door variant split: the +/// idle-first-byte gate first (ADR-014 knob 1 — the door's accept +/// path owns the defense; an accepted socket is not a channel), then +/// the variant. +async fn serve_connection( + stream: TcpStream, + peer: SocketAddr, + client: &Arc, + credentials: Option<&CredentialTable>, + token_client: Option<&Arc>, + idle: Duration, +) -> DoorExit { + // The idle-first-byte defense: an accepted connection that sends + // no first byte within the window is closed door-side (ADR-014's + // slowloris sliver). The peek waits for readability without + // consuming — the variant's own protocol read picks the byte up. + let mut first = [0u8; 1]; + match tokio::time::timeout(idle, stream.peek(&mut first)).await { + Ok(Ok(_)) => {} + Ok(Err(err)) => return DoorExit::Io(err), + Err(_) => { + tracing::debug!(peer = %peer, "front door: idle-first-byte timeout"); + return DoorExit::ClientGone; + } + } + + match (credentials, token_client) { + // The NoAuth door: the literal raw pipe (POC #1's shape — + // zero protocol translation). + (None, _) => raw_pipe(stream, client).await, + // The cred-bearing door: RFC 1929 terminates locally, + // re-origination rides the token sessions. + (Some(table), Some(tokens)) => cred_bearing(stream, client, table, tokens).await, + // Unreachable by construction (`serve_on` builds the pair) — + // the arm exists to keep the match exhaustive without a + // panic path (convention 2). + (Some(_), None) => { + DoorExit::Open("the cred-bearing door's token surface was not built".to_string()) + } + } +} + +/// The NoAuth door's raw pipe (POC #1's `front_door_raw_pipe_vanilla_ +/// client` shape): open a channel, then `copy_bidirectional` the +/// accepted socket onto the channel `BiStream` — the producer runs +/// the whole RFC conversation through the pipe; the door adds +/// nothing. +async fn raw_pipe(mut stream: TcpStream, client: &Arc) -> DoorExit { + let (_channel_id, send, recv) = match client + .open_channel( + crate::params::OP_SOCKS5_OPEN, + crate::params::Socks5Params::as_json(), + crate::params::SOCKS5_ALPN, + ) + .await + { + Ok(open) => open, + Err(err) => return DoorExit::Open(err.to_string()), + }; + let mut bidi = alkcall::core::types::BiStream::from_joined(recv, send); + let _ = tokio::io::copy_bidirectional(&mut stream, &mut bidi).await; + DoorExit::PlaneEnded +} + +/// The cred-bearing door (ADR-005): terminate the vanilla RFC 1929 +/// conversation's auth locally (the forked surface), map the accepted +/// credential through the static table, re-originate the SOCKS +/// conversation in-band via the token-carrying consumer sessions, +/// re-encode the in-band replies to the vanilla client, and pipe the +/// post-auth data-plane bytes raw. +/// +/// The door never pipes raw post-auth: the vanilla client's command +/// bytes were consumed by the door's own RFC half; the in-band path +/// re-speaks them from the parsed target. The password never crosses +/// the channel as a SOCKS password (the in-band handshake is NoAuth — +/// the mapped identity is the authorization); the channel identity +/// never reaches the vanilla client. +async fn cred_bearing( + mut stream: TcpStream, + _client: &Arc, + table: &CredentialTable, + token_client: &Arc, +) -> DoorExit { + // The local RFC 1929 half (the forked surface) + the command read. + let outcome = local_rfc1929_half(&mut stream, table).await; + let (cmd, target, identity) = match outcome { + Ok(triple) => triple, + Err(exit) => return exit, + }; + + // The in-band re-origination: the per-credential identity rides + // the open-op call's `auth_token` payload (CF-005 precedence + // (1)). The token is the mapped identity's id — the resolution + // seam's vocabulary; the password itself never rides it. + let token = identity.id.as_bytes().to_vec(); + + match cmd { + Socks5Command::TCPConnect => { + let session = match crate::consumer::Socks5ChannelSession::connect_with_token( + token_client, + target, + token, + ) + .await + { + Ok(session) => session, + Err(err) => return DoorExit::Open(err.to_string()), + }; + pipe_connect_plane(stream, session).await + } + Socks5Command::TCPBind => { + let session = + match crate::consumer::BindSession::bind_with_token(token_client, target, token) + .await + { + Ok(session) => session, + Err(err) => return DoorExit::Open(err.to_string()), + }; + pipe_bind_plane(stream, session).await + } + // UDP ASSOCIATE through the TCP door is the UDP door's shape + // (the datagram plane's RFC codec boundary — ADR-003), not + // this one's: refuse RFC-faithfully — the vanilla client gets + // the refused reply through the door's own codec (no phantom + // data plane, no in-band session opened). + Socks5Command::UDPAssociate => { + let reply = new_reply( + &ReplyError::CommandNotSupported, + SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0), + ); + let _ = stream.write_all(&reply).await; + let _ = stream.flush().await; + DoorExit::LocalRfc(SocksError::ReplyError(ReplyError::CommandNotSupported)) + } + } +} + +/// The door's local RFC 1929 half (the forked surface — ADR-005's +/// accept_password_auth shape): `negotiate_auth(&[ +/// PasswordAuthentication])` → `read_username_password` → check +/// through the static table → `accept`/`reject`, then +/// `read_command` for the re-origination's parsed target. The check +/// rides the forked `CheckResult`'s `bool` impl; the mapped identity +/// is recorded by the same table lookup the check ran (one table, +/// one truth) and re-yielded to the caller. +async fn local_rfc1929_half( + stream: &mut TcpStream, + table: &CredentialTable, +) -> Result<(Socks5Command, TargetAddr, Identity), DoorExit> { + // The check closure is sync (`FnMut(String, String) -> bool` over + // the forked `CheckResult`'s `bool` impl); the mapped identity is + // recorded through a sync witness (a `std` mutex — the closure + // runs once, synchronously). + let found: Arc>> = Arc::new(std::sync::Mutex::new(None)); + let found_for_check = Arc::clone(&found); + + let (proto, _ok) = Socks5ServerProtocol::<_, states::Authenticated>::accept_password_auth( + stream, + |user, pass| { + let mapped = table.lookup(&user, &pass).cloned(); + // A poisoned lock would mean a panic mid-check — the + // `unwrap_or_else(into_inner)` convention keeps the door + // alive (AGENTS.md convention 2). + let mut slot = found_for_check.lock().unwrap_or_else(|e| e.into_inner()); + let accepted = mapped.is_some(); + *slot = mapped; + // The check IS the table lookup: a miss rejects (the + // forked `accept_password_auth` writes the RFC 1929 + // rejection when the checker returns `false`). + accepted + }, + ) + .await + .map_err(DoorExit::from)?; + + let identity = found + .lock() + .unwrap_or_else(|e| e.into_inner()) + .take() + .ok_or_else(|| DoorExit::from(SocksServerError::AuthenticationRejected))?; + + let (_proto, cmd, target) = proto.read_command().await.map_err(DoorExit::from)?; + Ok((cmd, target, identity)) +} + +/// Pipe the re-originated CONNECT session: the vanilla reply (the +/// in-band reply's BND.ADDR re-encoded — the RFC codec at the door +/// boundary only, ADR-003), then the data plane ↔ the accepted socket. +/// The door already consumed the vanilla client's greeting + command +/// bytes; the post-auth bytes ride the session's data plane raw from +/// here. The pump shape is the taken inner stream +/// (`Socks5Stream` is itself `AsyncRead + AsyncWrite` — the +/// forked client type's intended use) against the local socket via +/// `copy_bidirectional`; the session stays the channel-teardown owner +/// (the `into_inner` pairing; dropped at the end — the EOF sentinel +/// reaches the producer's pump). +async fn pipe_connect_plane( + mut stream: TcpStream, + session: crate::consumer::Socks5ChannelSession, +) -> DoorExit { + // The vanilla CONNECT reply: the in-band reply's BND.ADDR + // re-encoded (the vanilla client is waiting for it — the door's + // RFC half consumed its local bytes; the in-band conversation is + // fresh and re-originated). The sentinel address is the + // RFC-compatibility surface the producer already chose; the door + // adds nothing. + let reply = match session.reply_addr() { + Some(TargetAddr::Ip(sa)) => new_reply(&ReplyError::Succeeded, sa), + Some(TargetAddr::Domain(_, port)) => new_reply( + &ReplyError::Succeeded, + SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), port), + ), + None => { + session.close().await; + return DoorExit::PlaneEnded; + } + }; + if stream.write_all(&reply).await.is_err() { + session.close().await; + return DoorExit::ClientGone; + } + let _ = stream.flush().await; + + let Ok(TakenInner { session, inner }) = session.into_inner() else { + return DoorExit::PlaneEnded; + }; + let _ = tokio::io::copy_bidirectional(&mut stream, &mut { inner }).await; + let _ = session; + DoorExit::PlaneEnded +} + +/// Pipe the re-originated BIND session (the BIND door, cred-bearing +/// variant): reply#1's advertised address is re-encoded to the +/// vanilla client from the session's captured `bind_addr` (the +/// conversation is the vanilla client's own — the door's RFC half +/// consumed the local greeting + command bytes, and the in-band +/// session's reply#1 address passes through re-encoded; consumer.md +/// §The BIND door's reply#2 fidelity). The quiet wait rides +/// `await_inbound` (the take-guarded reply#2 reader); reply#2's peer +/// is likewise re-encoded; the data plane passes through raw. The +/// session stays the channel-teardown owner. +async fn pipe_bind_plane( + mut stream: TcpStream, + mut session: crate::consumer::BindSession, +) -> DoorExit { + // Reply#1 (the vanilla form): the in-band reply's BND.ADDR + // re-encoded — the advertised address passes through unchanged. + let bind_addr = session.reply1().map(|reply1| reply1.bind_addr.clone()); + let Some(bind_addr) = bind_addr else { + session.close().await; + return DoorExit::PlaneEnded; + }; + let wire = match &bind_addr { + TargetAddr::Ip(sa) => new_reply(&ReplyError::Succeeded, *sa), + TargetAddr::Domain(name, port) => { + let mut wire = vec![ + consts::SOCKS5_VERSION, + ReplyError::Succeeded.as_u8(), + 0x00, + consts::SOCKS5_ADDR_TYPE_DOMAIN_NAME, + ]; + wire.push(name.len() as u8); + wire.extend_from_slice(name.as_bytes()); + wire.extend_from_slice(&port.to_be_bytes()); + wire + } + }; + if write_vanilla_reply(&mut stream, &wire).await.is_err() { + session.close().await; + return DoorExit::ClientGone; + } + + bind_quiet_and_pipe(stream, session).await +} + +/// The BIND door's quiet wait + reply#2 + the data plane: await the +/// inbound accept through the session (`await_inbound` — the one +/// normative reply#2 reader), re-encode reply#2's peer to the vanilla +/// client, then pipe the pass-through plane raw (the plane owns the +/// in-band stream — `BindDataPlane` impls `AsyncRead + AsyncWrite`, +/// the pass-through to the accepted peer; the session stays the +/// teardown owner, dropped at the end). +async fn bind_quiet_and_pipe( + mut stream: TcpStream, + mut session: crate::consumer::BindSession, +) -> DoorExit { + let plane = match session.await_inbound().await { + Ok(Some(plane)) => plane, + Ok(None) => return DoorExit::PlaneEnded, + // An accept-stage refusal (reply#2's mapped code was written + // in-band; the producer's pump ended the session): the door + // closes the vanilla connection — the conversation ends where + // the in-band one did. + Err(_) => { + session.close().await; + return DoorExit::PlaneEnded; + } + }; + + // Reply#2 (the vanilla form): the accepted peer re-encoded. + let reply = match plane.peer { + TargetAddr::Ip(sa) => new_reply(&ReplyError::Succeeded, sa), + TargetAddr::Domain(_, port) => new_reply( + &ReplyError::Succeeded, + SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), port), + ), + }; + if write_vanilla_reply(&mut stream, &reply).await.is_err() { + session.close().await; + return DoorExit::ClientGone; + } + + let _ = tokio::io::copy_bidirectional(&mut stream, &mut { plane }).await; + session.close().await; + DoorExit::PlaneEnded +} + +/// Write one vanilla reply frame (the door's RFC codec at the +/// boundary — ADR-003): write + flush; a failed write is the client +/// gone. +async fn write_vanilla_reply(stream: &mut TcpStream, reply: &[u8]) -> std::io::Result<()> { + stream.write_all(reply).await?; + stream.flush().await +} diff --git a/src/lib.rs b/src/lib.rs index 50a8d8b..3178ec4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -68,6 +68,9 @@ pub mod wire; #[cfg(feature = "local")] pub mod local; +#[cfg(feature = "local")] +pub mod door; + /// 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 @@ -109,3 +112,8 @@ 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 consumer::TokenClient; +#[cfg(feature = "local")] +pub use door::{CredentialTable, DoorCredential, FrontDoor}; diff --git a/tasks/implementation/door-tcp.md b/tasks/implementation/door-tcp.md index 5fb6146..a4f67fd 100644 --- a/tasks/implementation/door-tcp.md +++ b/tasks/implementation/door-tcp.md @@ -1,7 +1,7 @@ --- id: implementation/door-tcp name: Front doors — the TCP door (raw-pipe + RFC 1929 cred-bearing) and the BIND door -status: pending +status: completed depends_on: [implementation/local-providers] scope: moderate risk: medium @@ -89,23 +89,23 @@ Deliverables: ## Acceptance Criteria -- [ ] `cargo test` passes; `cargo test --all-features` passes -- [ ] `cargo clippy --all-targets -- -D warnings` clean; `cargo fmt +- [x] `cargo test` passes; `cargo test --all-features` passes +- [x] `cargo clippy --all-targets -- -D warnings` clean; `cargo fmt --check` clean -- [ ] Default wasm checks stay green (door code is `local`-gated) -- [ ] The NoAuth door is a literal raw pipe (zero protocol +- [x] Default wasm checks stay green (door code is `local`-gated) +- [x] The NoAuth door is a literal raw pipe (zero protocol translation) — pinned by test -- [ ] The cred-bearing door terminates RFC 1929 locally and +- [x] The cred-bearing door terminates RFC 1929 locally and re-originates in-band (never pipes raw post-auth) — pinned by test (a vanilla client's command bytes must not reach the producer as greeting bytes) -- [ ] The credential→identity mapping is the static config table +- [x] The credential→identity mapping is the static config table (ADR-005); the password never crosses the channel as a SOCKS password — pinned by test (the producer's establisher witness sees the mapped identity, never the password) -- [ ] The idle-first-byte timeout + session cap land as constructor +- [x] The idle-first-byte timeout + session cap land as constructor config — pinned by tests -- [ ] Per-connection failure isolation — pinned by test +- [x] Per-connection failure isolation — pinned by test ## References @@ -123,8 +123,123 @@ Deliverables: ## Notes -> To be filled by implementation agent +> Filled by the implementation agent (2026-09-25). + +- **The identity mechanism — decided (the task's open choice)**: the + per-credential identity rides the open-op call's `auth_token` + payload (CF-005 precedence (1)), never a per-credential + `ChannelClient`. Mechanically: the door wraps the client's + channel-0 `CallConnection` once (`TokenClient::wrap` — the raw + `call_with_payload` surface, since the token is a payload-level + field, ADR-017 §7) and each credential's open rides the payload + `auth_token`; the producer's `ServingConfig.identity_provider` + resolves it and the ACL gate reads the resolved scopes exactly as + for any per-call opener identity. The token is the mapped + identity's *id* (the resolution seam's vocabulary — the password + never rides it). Rationale: one transport, one call connection, + many identities — a per-credential `ChannelClient` would need a + full transport connection per credential, which the ssh `-D` + single-socket posture cannot afford. +- **New consumer surface (`src/consumer.rs`)**: `TokenClient` (the + token-carrying open surface over the taken `CallConnection` + + shared manager) and the `*_with_token` twins of the three session + constructors (`Socks5ChannelSession::connect_with_token`, + `BindSession::bind_with_token`, + `AssociateSession::associate_with_token`) through a + `open_with_token_and_negotiate` prologue (the mirror of + `open_and_negotiate`: open-with-token → adopt → negotiate → + teardown-at-birth on failure). `Socks5ChannelSession` gained the + `reply_addr` field + accessor (the CONNECT reply's BND.ADDR, + captured at `connect` — the door's vanilla-reply re-encoding rides + it). The existing constructor surfaces are untouched. +- **The door (`src/door.rs`, feature `local`)**: + - `FrontDoor::builder().credentials(...).bind(...).build()` → + `serve(client)`; `serve_on(listener, client)` is the pre-bound + seam (tests/assembly wire their own listener — the bind decision + stays the caller's). + - **The NoAuth door**: accept → `open_channel` → + `copy_bidirectional(accepted_socket, BiStream)` — the literal + raw pipe; zero translation (POC #1's shape). + - **The cred-bearing door**: the forked + `accept_password_auth` shape locally (`negotiate_auth(&[ + PasswordAuthentication])` → `read_username_password` → check → + `accept`/`reject`), the checker closing over the static + `CredentialTable` (`user:pass → Identity`, the ADR-005 static + mapping — the check and the mapping are one lookup), then + `read_command` for the re-origination's parsed target, then the + in-band sessions with the mapped identity's id as the token. + Replies are re-encoded to the vanilla client (the RFC codec at + the door boundary only, ADR-003): the CONNECT reply from + `reply_addr()`, the BIND door's reply#1 from `reply1() + .bind_addr`, reply#2 from `await_inbound()`'s plane peer; the + post-reply data planes ride `copy_bidirectional` against the + taken in-band streams. **Never pipes raw post-auth** — the + vanilla client's greeting/command bytes were consumed by the + door's RFC half; the in-band path re-speaks them from the parsed + target (pinned by test: the cred-bearing CONNECT round-trip + succeeding at all is the pin — raw piping would misparse the + command bytes as a greeting). + - **The BIND door**: same listener shape, riding + `BindSession::bind_with_token`; the quiet wait rides the + session's `await_inbound`; reply#1's advertised address passes + through unchanged (the door adds nothing). + - UDP ASSOCIATE through the TCP door: refused RFC-faithfully at + the door (reply code 0x07 — the UDP door's task, not this one). + - **The accept-path defense (ADR-014)**: the idle-first-byte + window (a `peek`-based readability wait before any protocol + work — an accepted connection silent past the window is closed + door-side) and the max-sessions cap (a `Semaphore`; an accept + beyond the bound is refused immediately, no channel opened) as + constructor config. Defaults: 30s / 256. + - **Failure isolation**: per-accept `tokio::spawn`; failures log + + isolate — the accept loop continues (the alktunnels posture). +- **Harness**: `wire_consumer_side_with_provider` added (the + consumer-side topology with the producer's + `ServingConfig.identity_provider` configurable — the CF-005 + precedence (1) seam the token sessions ride); + `wire_consumer_side_with` delegates unchanged. +- **Test-provider note**: the test `DoorTokenProvider` resolves + `door-`-prefixed tokens (the tests' credential-table identity ids) + to open-scoped identities (bind-scoped when listed). The token→ + identity resolution is producer policy, not wire shape. ## Summary -> To be filled on completion \ No newline at end of file +> Filled on completion (2026-09-25). + +Implemented the consumer's TCP front door (feature `local`): the +NoAuth door (the literal raw pipe — accept → channel `BiStream` via +`copy_bidirectional`, zero protocol translation), the cred-bearing +door (RFC 1929 terminates locally through the forked +`accept_password_auth` shape; the static credential→identity table +maps the accepted credential; the SOCKS conversation re-originates +in-band via the token-carrying consumer sessions with the vanilla +replies re-encoded at the door; the password never crosses the +channel as a SOCKS password), and the BIND door (reply#1 fidelity — +the advertised address passes through unchanged; the quiet wait + +reply#2 + the pass-through data plane). The identity mechanism +decided + recorded: the per-credential identity rides the open-op +`auth_token` payload (CF-005 precedence (1)) via the new +`TokenClient` (the client's call connection taken once per door, +shared manager), resolved producer-side through +`ServingConfig.identity_provider`. The ADR-014 accept-path defense +landed as constructor config: the idle-first-byte window (peek-based, +door-side close) + the max-sessions cap (semaphore-refused accepts), +plus per-accept failure isolation (the loop continues). New +consumer-session surface: `TokenClient` + the `*_with_token` +constructor twins + `Socks5ChannelSession::reply_addr` (existing +surfaces untouched). 10 integration tests in `tests/door-tcp.rs` +pin: the NoAuth raw-pipe round-trip (the POC #1 curl shape), the +cred-bearing good/bad-credential paths (the witness sees the mapped +identity id, never the password; a rejection opens no channel), the +never-pipes-raw-post-auth pin, the BIND door's vanilla conversation +(reply#1/reply#2 fidelity + the data plane), the idle timeout, the +session cap, failure isolation, and the regression guards (native +sessions beside a NoAuth door; `bind_with_token` driven directly). + +Verification: `cargo test` (375 green across 10 targets) + +`cargo test --all-features` (423 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`) green, and the default +dependency graph stays net-free (the door rides `local` alone). \ No newline at end of file diff --git a/tests/door-tcp.rs b/tests/door-tcp.rs new file mode 100644 index 0000000..527e241 --- /dev/null +++ b/tests/door-tcp.rs @@ -0,0 +1,858 @@ +//! The TCP front door's integration tests (feature `local`): the +//! NoAuth raw pipe with a vanilla RFC 1928 client (the POC #1 curl +//! shape), the cred-bearing door's RFC 1929 (good creds → CONNECT +//! round-trip; bad creds → rejection with no channel opened; the +//! command bytes never reach the producer as greeting bytes), the +//! BIND door (reply#1/reply#2 fidelity), the ADR-014 accept-path +//! defense (the idle-first-byte timeout + the session cap), and the +//! per-connection failure isolation. +#![cfg(feature = "local")] + +mod harness; + +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::time::timeout; + +use alkcall::core::auth::{Identity, IdentityProvider}; +use alksocks::consumer::{BindSession, Socks5ChannelSession}; +use alksocks::rfc1928::consts; +use alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr}; +use alksocks::{FrontDoor, SOCKS5_BIND_SCOPE}; + +use harness::consumer_identity; + +const BOUND: Duration = Duration::from_secs(10); + +/// The token→identity provider the cred-bearing door's producer runs +/// (CF-005 precedence (1)): the token is the mapped identity's id — +/// the resolution seam's vocabulary; the password never rides it. +/// Resolves the credential-table identity ids (the `door-` prefixed +/// forms the tests configure) to identities carrying `socks5:open` +/// (plus `socks5:bind` when the entry is bind-scoped); anything else +/// resolves to none (the dispatch falls through to the transport +/// identity per the CF-005 precedence order). +#[derive(Clone)] +struct DoorTokenProvider { + bind_scoped: Vec, +} + +impl IdentityProvider for DoorTokenProvider { + fn resolve_from_fingerprint(&self, _: &str) -> Option { + None + } + fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option { + let id = String::from_utf8_lossy(&token.raw).into_owned(); + if id.starts_with("door-") { + let mut scopes = vec![alksocks::SOCKS5_OPEN_SCOPE.to_string()]; + if self.bind_scoped.contains(&id) { + scopes.push(SOCKS5_BIND_SCOPE.to_string()); + } + return Some(Identity { + id, + scopes, + resources: Default::default(), + }); + } + None + } +} + +/// An echo TCP server on an ephemeral loopback port (the vanilla +/// client's CONNECT target — the POC #1 curl shape). +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 +} + +/// The door's raw RFC 1928 client (the differential-suite pattern): a +/// `TcpStream` dialed to the door, greeting + method selection + +/// command + reply read. +struct VanillaClient { + stream: TcpStream, +} + +impl VanillaClient { + async fn dial(addr: SocketAddr) -> Self { + Self { + stream: TcpStream::connect(addr).await.expect("dial the door"), + } + } + + /// Greeting `05 02 00` (password auth offered — the cred-bearing + /// door's shape) → the method-selection read. + async fn greet_password(&mut self) -> u8 { + self.stream + .write_all(&[0x05, 0x01, 0x02]) + .await + .expect("greeting written"); + let mut chosen = [0u8; 2]; + self.stream + .read_exact(&mut chosen) + .await + .expect("method selected"); + assert_eq!(chosen[0], consts::SOCKS5_VERSION); + chosen[1] + } + + /// Greeting `05 01 00` (no-auth offered) → the method-selection + /// read. + async fn greet_no_auth(&mut self) -> u8 { + self.stream + .write_all(&[0x05, 0x01, 0x00]) + .await + .expect("greeting written"); + let mut chosen = [0u8; 2]; + self.stream + .read_exact(&mut chosen) + .await + .expect("method selected"); + assert_eq!(chosen[0], consts::SOCKS5_VERSION); + chosen[1] + } + + /// The RFC 1929 credential sub-negotiation → the status byte. + async fn send_credentials(&mut self, user: &str, pass: &str) -> u8 { + let mut packet = vec![0x01, user.len() as u8]; + packet.extend_from_slice(user.as_bytes()); + packet.push(pass.len() as u8); + packet.extend_from_slice(pass.as_bytes()); + self.stream.write_all(&packet).await.expect("creds sent"); + let mut status = [0u8; 2]; + self.stream.read_exact(&mut status).await.expect("status"); + assert_eq!(status[0], 0x01, "the sub-negotiation version is 1"); + status[1] + } + + /// A CONNECT request (v4 target) → the reply code. + async fn connect_v4(&mut self, target: SocketAddr) -> u8 { + let mut req = [0u8; 10]; + req[..3].copy_from_slice(&[0x05, consts::SOCKS5_CMD_TCP_CONNECT, 0x00]); + req[3] = 0x01; + let SocketAddr::V4(v4) = target else { + panic!("v4 targets only"); + }; + req[4..8].copy_from_slice(&v4.ip().octets()); + req[8..10].copy_from_slice(&v4.port().to_be_bytes()); + self.stream.write_all(&req).await.expect("request sent"); + let (code, _addr) = self.read_reply_addr().await; + code + } + + /// A BIND request (v4 target) → reply#1's code + the advertised + /// BND.ADDR. + async fn bind_v4(&mut self, ip: [u8; 4], port: u16) -> (u8, TargetAddr) { + let mut req = [0u8; 10]; + req[..3].copy_from_slice(&[0x05, consts::SOCKS5_CMD_TCP_BIND, 0x00]); + req[3] = 0x01; + req[4..8].copy_from_slice(&ip); + req[8..10].copy_from_slice(&port.to_be_bytes()); + self.stream.write_all(&req).await.expect("bind sent"); + self.read_reply_addr().await + } + + /// Read one reply (any BND.ADDR form) → (code, addr). + async fn read_reply_addr(&mut self) -> (u8, TargetAddr) { + let mut head = [0u8; 4]; + self.stream.read_exact(&mut head).await.expect("reply head"); + assert_eq!(head[0], consts::SOCKS5_VERSION, "reply version 5"); + let addr = match head[3] { + consts::SOCKS5_ADDR_TYPE_IPV4 => { + let mut rest = [0u8; 6]; + self.stream.read_exact(&mut rest).await.expect("v4 addr"); + TargetAddr::Ip(SocketAddr::from(( + std::net::Ipv4Addr::from([rest[0], rest[1], rest[2], rest[3]]), + u16::from_be_bytes([rest[4], rest[5]]), + ))) + } + consts::SOCKS5_ADDR_TYPE_DOMAIN_NAME => { + let mut len = [0u8; 1]; + self.stream.read_exact(&mut len).await.expect("domain len"); + let mut rest = vec![0u8; len[0] as usize + 2]; + self.stream.read_exact(&mut rest).await.expect("domain"); + TargetAddr::Domain( + String::from_utf8(rest[..rest.len() - 2].to_vec()).expect("domain utf8"), + u16::from_be_bytes([rest[rest.len() - 2], rest[rest.len() - 1]]), + ) + } + other => panic!("unknown reply ATYP {other:#x}"), + }; + (head[1], addr) + } +} + +/// The NoAuth door end to end (the POC #1 curl shape): a vanilla RFC +/// 1928 client through the raw pipe → CONNECT round-trip to a real +/// TCP echo target. Zero protocol translation pinned: the door adds +/// nothing — the producer runs the whole RFC conversation through the +/// pipe (the vanilla client's greeting crosses it verbatim). +#[tokio::test] +async fn noauth_door_raw_pipe_vanilla_connect_round_trip() { + let echo = echo_tcp_server().await; + let dial = alksocks::tcp_dial(); + let topo = harness::wire_consumer_side_with(dial, None, None, None).await; + + // The door takes a pre-bound listener (`serve_on`), so the test + // owns the port (the assembly seam — the same accept loop without + // the door's own bind). + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("door addr"); + let door = FrontDoor::builder().bind(addr).build(); + let _serve = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { + door.serve_on(listener, client) + .await + .expect("the door serves") + } + }); + + let mut vanilla = VanillaClient::dial(addr).await; + let method = timeout(BOUND, vanilla.greet_no_auth()) + .await + .expect("greeting round-trips through the raw pipe"); + assert_eq!(method, consts::SOCKS5_AUTH_METHOD_NONE); + + let code = timeout(BOUND, vanilla.connect_v4(echo)) + .await + .expect("CONNECT round-trips through the raw pipe"); + assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); + + // The data plane is the literal pipe: echo both directions. + vanilla.stream.write_all(b"vanilla-echo").await.unwrap(); + let mut buf = [0u8; 12]; + timeout(BOUND, vanilla.stream.read_exact(&mut buf)) + .await + .expect("echo within bound") + .expect("read"); + assert_eq!(&buf, b"vanilla-echo", "the raw pipe round-trips"); + topo.consumer.manager(); +} + +/// The cred-bearing door: good creds → RFC 1929 success → CONNECT +/// round-trip. The mapped identity reaches the producer (the +/// establisher witness sees the mapped identity id, never the +/// password); the in-band conversation is re-originated (the vanilla +/// client's command bytes were consumed by the door's own RFC half). +#[tokio::test] +async fn cred_door_good_creds_connect_round_trip_with_mapped_identity() { + let echo = echo_tcp_server().await; + let dial = alksocks::tcp_dial(); + + let topo = harness::wire_consumer_side_with_provider( + dial, + None, + None, + Some(consumer_identity()), + Arc::new(DoorTokenProvider { + bind_scoped: vec![], + }), + ) + .await; + let _ = echo; + + let table = alksocks::CredentialTable::new(vec![alksocks::DoorCredential { + username: "alice".to_string(), + password: "secret".to_string(), + identity: Identity { + id: "door-alice".to_string(), + scopes: vec![alksocks::SOCKS5_OPEN_SCOPE.to_string()], + resources: Default::default(), + }, + }]); + + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("door addr"); + let door = FrontDoor::builder().credentials(table).bind(addr).build(); + let _serve = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { + door.serve_on(listener, client) + .await + .expect("the door serves") + } + }); + + let mut vanilla = VanillaClient::dial(addr).await; + let method = timeout(BOUND, vanilla.greet_password()) + .await + .expect("greeting"); + assert_eq!(method, consts::SOCKS5_AUTH_METHOD_PASSWORD); + + let status = timeout(BOUND, vanilla.send_credentials("alice", "secret")) + .await + .expect("credential exchange"); + assert_eq!(status, 0x00, "accepted"); + assert_eq!(status, 0x00, "RFC 1929 success"); + + let code = timeout(BOUND, vanilla.connect_v4(echo)) + .await + .expect("CONNECT after auth"); + assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); + + vanilla.stream.write_all(b"cred-echo").await.unwrap(); + let mut buf = [0u8; 9]; + timeout(BOUND, vanilla.stream.read_exact(&mut buf)) + .await + .expect("echo within bound") + .expect("read"); + assert_eq!(&buf, b"cred-echo"); + + // The establisher witness saw the mapped identity (CF-006 via the + // token resolution — CF-005 precedence (1)), never the password. + timeout(BOUND, async { + loop { + let seen = topo.identity_witness.lock().await.clone(); + if seen.as_deref() == Some("door-alice") { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("the establisher witnessed the mapped identity"); +} + +/// The cred-bearing door: bad creds → the RFC 1929 rejection (status +/// 0xff), and no channel opened (the identity witness never fires; +/// the consumer's manager holds only channel 0 — the ACL boundary +/// held door-side). +#[tokio::test] +async fn cred_door_bad_creds_rejected_no_channel_opened() { + let dial = alksocks::tcp_dial(); + let table = alksocks::CredentialTable::new(vec![alksocks::DoorCredential { + username: "alice".to_string(), + password: "secret".to_string(), + identity: Identity { + id: "door-alice".to_string(), + scopes: vec![alksocks::SOCKS5_OPEN_SCOPE.to_string()], + resources: Default::default(), + }, + }]); + + let topo = harness::wire_consumer_side_with_provider( + dial, + None, + None, + Some(consumer_identity()), + Arc::new(DoorTokenProvider { + bind_scoped: vec![], + }), + ) + .await; + + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("door addr"); + let door = FrontDoor::builder().credentials(table).bind(addr).build(); + let _serve = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { + door.serve_on(listener, client) + .await + .expect("the door serves") + } + }); + + let mut vanilla = VanillaClient::dial(addr).await; + let method = timeout(BOUND, vanilla.greet_password()) + .await + .expect("greeting"); + assert_eq!(method, consts::SOCKS5_AUTH_METHOD_PASSWORD); + + let status = timeout(BOUND, vanilla.send_credentials("alice", "wrong")) + .await + .expect("credential exchange"); + assert_eq!(status, 0xff, "RFC 1929 rejection"); + + // No channel opened: the witness never fired (the door-side + // rejection never reached the producer), and the consumer's + // manager holds only channel 0. + tokio::time::sleep(Duration::from_millis(200)).await; + assert_eq!( + topo.identity_witness.lock().await.as_deref(), + None, + "the rejection never opened a channel" + ); + assert!( + topo.consumer + .manager() + .channel_ids() + .iter() + .all(|&id| id == 0), + "no channel entry exists opener-side" + ); +} + +/// The pin: a vanilla client's command bytes must not reach the +/// producer as greeting bytes (the door never pipes raw post-auth). +/// The producer's in-band conversation is re-originated — the +/// producer's `accept_no_auth` reads a fresh greeting written by the +/// *consumer session* (`Socks5Stream::use_stream`), not the vanilla +/// client's consumed bytes. Observable as: the cred-bearing CONNECT +/// round-trip works at all (raw post-auth piping would misparse the +/// command bytes as a greeting — the producer would reply 0xff-shaped +/// garbage), and the witness carries the mapped identity (the +/// password never crossed). +#[tokio::test] +async fn cred_door_never_pipes_raw_post_auth() { + let echo = echo_tcp_server().await; + let dial = alksocks::tcp_dial(); + let table = alksocks::CredentialTable::new(vec![alksocks::DoorCredential { + username: "bob".to_string(), + password: "pw".to_string(), + identity: Identity { + id: "door-bob".to_string(), + scopes: vec![alksocks::SOCKS5_OPEN_SCOPE.to_string()], + resources: Default::default(), + }, + }]); + + let topo = harness::wire_consumer_side_with_provider( + dial, + None, + None, + Some(consumer_identity()), + Arc::new(DoorTokenProvider { + bind_scoped: vec![], + }), + ) + .await; + + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("door addr"); + let door = FrontDoor::builder().credentials(table).bind(addr).build(); + let _serve = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { + door.serve_on(listener, client) + .await + .expect("the door serves") + } + }); + + // The vanilla client speaks greeting → password-auth → command — + // if the door piped these bytes raw post-auth, the producer's + // in-band greeting read would consume `05 01 00 01 ...` (the + // command request) as a method list and the conversation would + // fail. The round-trip succeeding IS the pin (the re-origination + // re-spoke the bytes in-band). + let mut vanilla = VanillaClient::dial(addr).await; + let method = timeout(BOUND, vanilla.greet_password()) + .await + .expect("greeting"); + assert_eq!(method, consts::SOCKS5_AUTH_METHOD_PASSWORD, "password auth"); + let status = timeout(BOUND, vanilla.send_credentials("bob", "pw")) + .await + .expect("creds"); + assert_eq!(status, 0x00, "accepted"); + let code = timeout(BOUND, vanilla.connect_v4(echo)) + .await + .expect("the in-band conversation is fresh (never the consumed bytes)"); + assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); + vanilla.stream.write_all(b"pin").await.unwrap(); + let mut buf = [0u8; 3]; + timeout(BOUND, vanilla.stream.read_exact(&mut buf)) + .await + .expect("echo") + .expect("read"); + assert_eq!(&buf, b"pin"); +} + +/// The BIND door (cred-bearing variant): a vanilla BIND conversation +/// through the door — reply#1's advertised address passes through +/// unchanged (the door adds nothing), reply#2 carries the real peer, +/// and the data plane round-trips (the POC #5 door shape). +#[tokio::test] +async fn bind_door_vanilla_conversation_reply_fidelity() { + let dial = alksocks::tcp_dial(); + let egress = None; + let accept = Some(alksocks::bind_accept([127, 0, 0, 1])); + let table = alksocks::CredentialTable::new(vec![alksocks::DoorCredential { + username: "carol".to_string(), + password: "pw".to_string(), + identity: Identity { + id: "door-carol".to_string(), + scopes: vec![ + alksocks::SOCKS5_OPEN_SCOPE.to_string(), + SOCKS5_BIND_SCOPE.to_string(), + ], + resources: Default::default(), + }, + }]); + + let topo = harness::wire_consumer_side_with_provider( + dial, + egress, + accept, + Some(consumer_identity()), + Arc::new(DoorTokenProvider { + bind_scoped: vec!["door-carol".to_string()], + }), + ) + .await; + + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("door addr"); + let door = FrontDoor::builder().credentials(table).bind(addr).build(); + let _serve = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { + door.serve_on(listener, client) + .await + .expect("the door serves") + } + }); + + let mut vanilla = VanillaClient::dial(addr).await; + let method = timeout(BOUND, vanilla.greet_password()) + .await + .expect("greeting"); + assert_eq!(method, consts::SOCKS5_AUTH_METHOD_PASSWORD, "password auth"); + let status = timeout(BOUND, vanilla.send_credentials("carol", "pw")) + .await + .expect("creds"); + assert_eq!(status, 0x00, "accepted"); + + // BIND request: 127.0.0.1:0 (ephemeral — the local accept + // provider binds the real listener). + let (code1, advertised) = timeout(BOUND, vanilla.bind_v4([127, 0, 0, 1], 0)) + .await + .expect("reply#1"); + assert_eq!( + code1, + consts::SOCKS5_REPLY_SUCCEEDED, + "BIND served under the gate" + ); + let TargetAddr::Ip(advertised_sa) = advertised else { + panic!("an IP advertised address, got {advertised:?}") + }; + assert!( + advertised_sa.ip().is_loopback() && advertised_sa.port() != 0, + "reply#1 carries the real listener address: {advertised_sa}" + ); + + // The app server dials the advertised address. + let mut app = timeout(BOUND, TcpStream::connect(advertised_sa)) + .await + .expect("dial within bound") + .expect("app dials the advertised address"); + app.write_all(b"vanilla-bind").await.unwrap(); + app.flush().await.unwrap(); + + // Reply#2: success + the real peer (the door adds nothing). + let (code2, peer) = timeout(BOUND, vanilla.read_reply_addr()) + .await + .expect("reply#2"); + assert_eq!(code2, consts::SOCKS5_REPLY_SUCCEEDED, "reply#2 success"); + let TargetAddr::Ip(peer_sa) = peer else { + panic!("an IP peer, got {peer:?}") + }; + assert_eq!( + peer_sa.port(), + app.local_addr().expect("app local").port(), + "reply#2 carries the real peer port (byte-exact)" + ); + + // The data plane through the door. + let mut buf = [0u8; 12]; + timeout(BOUND, vanilla.stream.read_exact(&mut buf)) + .await + .expect("data plane round trip") + .expect("read"); + assert_eq!(&buf, b"vanilla-bind"); + drop(app); +} + +/// The idle-first-byte timeout (ADR-014 knob 1): an accepted +/// connection that sends no first byte within the window is closed +/// door-side. +#[tokio::test] +async fn idle_first_byte_timeout_closes_a_silent_connection() { + let dial = alksocks::tcp_dial(); + let topo = harness::wire_consumer_side_with(dial, None, None, None).await; + + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("door addr"); + let door = FrontDoor::builder() + .bind(addr) + .idle_first_byte(Duration::from_millis(300)) + .build(); + let _serve = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { + door.serve_on(listener, client) + .await + .expect("the door serves") + } + }); + + // Connect and stay silent: the door must close the socket within + // (window + slack). + let mut silent = TcpStream::connect(addr).await.expect("dial the door"); + let mut probe = [0u8; 1]; + let start = std::time::Instant::now(); + let outcome = timeout(Duration::from_secs(3), silent.read(&mut probe)).await; + match outcome { + Err(_) => panic!("the silent connection must be closed door-side"), + Ok(Ok(0)) | Ok(Err(_)) => {} + Ok(Ok(n)) => panic!("EOF expected, got {n} bytes"), + } + let elapsed = start.elapsed(); + assert!( + elapsed >= Duration::from_millis(250), + "the door waited the window before closing (not instant): {elapsed:?}" + ); + assert!( + elapsed < Duration::from_secs(2), + "the close arrived near the window, not arbitrarily late: {elapsed:?}" + ); + + // No channel was opened for the silent connection. + tokio::time::sleep(Duration::from_millis(100)).await; + assert!( + topo.consumer + .manager() + .channel_ids() + .iter() + .all(|&id| id == 0), + "the idle timeout fired before any channel opened" + ); +} + +/// The session cap (ADR-014 knob 2): an accept beyond the bound is +/// refused door-side — no channel opened for the refused accept, and +/// the door keeps serving (the loop continues). +#[tokio::test] +async fn session_cap_refuses_beyond_the_bound() { + let dial = alksocks::tcp_dial(); + let topo = harness::wire_consumer_side_with(dial, None, None, None).await; + + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("door addr"); + let door = FrontDoor::builder() + .bind(addr) + .idle_first_byte(Duration::from_secs(30)) + .max_sessions(1) + .build(); + let _serve = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { + door.serve_on(listener, client) + .await + .expect("the door serves") + } + }); + + // Session 1: connects and stays silent (holds the one slot). + let mut holder = TcpStream::connect(addr).await.expect("first dial"); + let _ = &mut holder; + + // Session 2: beyond the cap — refused. The socket closes without + // a reply (the door-side refusal; no RFC bytes ever flow). + let refused = TcpStream::connect(addr).await.expect("second dial"); + let mut probe = [0u8; 1]; + let outcome = timeout(Duration::from_secs(3), { + let mut refused = refused; + async move { refused.read(&mut probe).await } + }) + .await; + match outcome { + Err(_) => panic!("the beyond-cap connection must be refused"), + Ok(Ok(0)) | Ok(Err(_)) => {} + Ok(Ok(n)) => panic!("no bytes for a beyond-cap accept, got {n}"), + } + + // Exactly one channel exists (the holder's — opened when it went + // silent-past... wait, no: the holder sent no bytes either, so the + // idle timeout governs it; the cap test needs the holder to stay + // within the window while the second dial arrives. The holder's + // channel opens only after its first byte — silence keeps it + // channel-less. Assert: no channels at all for either connection + // (both silent), and the door is still serving (a fresh dial + // succeeds). + tokio::time::sleep(Duration::from_millis(100)).await; + assert!( + topo.consumer + .manager() + .channel_ids() + .iter() + .all(|&id| id == 0), + "neither silent connection opened a channel" + ); + + // The door keeps serving (the loop continued). + let fresh = TcpStream::connect(addr).await; + assert!(fresh.is_ok(), "the door keeps accepting after refusals"); + drop(fresh); +} + +/// Per-connection failure isolation (the alktunnels per-accept +/// posture): one dead client doesn't kill the door — a client that +/// speaks garbage (an invalid greeting) gets its connection ended, +/// and the door keeps serving fresh clients end to end. +#[tokio::test] +async fn per_connection_failure_isolation() { + let echo = echo_tcp_server().await; + let dial = alksocks::tcp_dial(); + let topo = harness::wire_consumer_side_with(dial, None, None, None).await; + + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("door addr"); + let door = FrontDoor::builder().bind(addr).build(); + let _serve = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { + door.serve_on(listener, client) + .await + .expect("the door serves") + } + }); + + // Client A: protocol garbage (a bad version byte through the raw + // pipe — the producer's typestate refuses in-band; the door's + // pipe ends with the session). + let mut garbage = VanillaClient::dial(addr).await; + garbage.stream.write_all(&[0x04, 0x01, 0x00]).await.unwrap(); + // Give the producer's refusal a beat; the door's pipe tears down. + tokio::time::sleep(Duration::from_millis(200)).await; + drop(garbage); + + // Client B: the full vanilla CONNECT round-trip still works (the + // door survived the dead client). + let mut good = VanillaClient::dial(addr).await; + let method = timeout(BOUND, good.greet_no_auth()) + .await + .expect("the door is alive"); + assert_eq!(method, consts::SOCKS5_AUTH_METHOD_NONE); + let code = timeout(BOUND, good.connect_v4(echo)) + .await + .expect("the door survived the dead client"); + assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED); + good.stream.write_all(b"alive").await.unwrap(); + let mut buf = [0u8; 5]; + timeout(BOUND, good.stream.read_exact(&mut buf)) + .await + .expect("echo") + .expect("read"); + assert_eq!(&buf, b"alive"); +} + +/// The channels-native sessions still work beside the door (the +/// regression guard: the door's `TokenClient::wrap` took the client's +/// call connection — a NoAuth door never wraps, so the client's own +/// sessions keep serving; a cred-bearing door's client is door-only +/// by design — the wrap's contract, asserted here at the surface +/// level via a NoAuth topology). +#[tokio::test] +async fn native_sessions_work_beside_a_noauth_door() { + let echo = echo_tcp_server().await; + let dial = alksocks::tcp_dial(); + let topo = harness::wire_consumer_side_with(dial, None, None, None).await; + + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("door addr"); + let door = FrontDoor::builder().bind(addr).build(); + let _serve = tokio::spawn({ + let client = Arc::clone(&topo.consumer); + async move { + door.serve_on(listener, client) + .await + .expect("the door serves") + } + }); + + // The channels-native CONNECT session rides the same client. + let mut session = Socks5ChannelSession::connect(topo.consumer.as_ref(), TargetAddr::Ip(echo)) + .await + .expect("the native session connects"); + let (mut read, mut write) = session.split().expect("post-reply pipe"); + write.write_all(b"native").await.unwrap(); + write.flush().await.unwrap(); + let mut buf = [0u8; 6]; + timeout(BOUND, read.read_exact(&mut buf)) + .await + .expect("echo") + .expect("read"); + assert_eq!(&buf, b"native"); + session.close().await; +} + +/// The BIND door via the channels-native session (the regression +/// guard on the door's BIND path machinery: the same +/// `BindSession::bind_with_token` surface the door rides, driven +/// directly with the token provider wired). +#[tokio::test] +async fn bind_session_with_token_round_trip() { + let dial = alksocks::tcp_dial(); + let accept = Some(alksocks::bind_accept([127, 0, 0, 1])); + + let topo = harness::wire_consumer_side_with_provider( + dial, + None, + accept, + Some(consumer_identity()), + Arc::new(DoorTokenProvider { + bind_scoped: vec!["door-native-bind".to_string()], + }), + ) + .await; + + let token_client = + Arc::new(alksocks::consumer::TokenClient::wrap(topo.consumer.as_ref()).await); + let mut session = BindSession::bind_with_token( + token_client.as_ref(), + ("bind.token", 0u16).to_target_addr().unwrap(), + b"door-native-bind".to_vec(), + ) + .await + .expect("the token-carrying BIND session binds"); + + let TargetAddr::Ip(sa) = session.reply1().expect("reply#1").bind_addr.clone() else { + panic!("an IP BND.ADDR") + }; + assert!(sa.ip().is_loopback() && sa.port() != 0, "a real listener"); + + let mut app = timeout(BOUND, TcpStream::connect(sa)) + .await + .expect("dial") + .expect("app dials"); + let mut plane = timeout(BOUND, session.await_inbound()) + .await + .expect("reply#2") + .expect("plane") + .expect("accept resolved"); + plane.write_all(b"token-bind").await.unwrap(); + plane.flush().await.unwrap(); + let mut buf = [0u8; 10]; + timeout(BOUND, app.read_exact(&mut buf)) + .await + .expect("read") + .expect("read"); + assert_eq!(&buf, b"token-bind"); + session.close().await; +} diff --git a/tests/harness.rs b/tests/harness.rs index fdf5e09..0b26856 100644 --- a/tests/harness.rs +++ b/tests/harness.rs @@ -672,6 +672,28 @@ pub async fn wire_consumer_side_with( egress: Option, accept: Option, identity: Option, +) -> ConsumerSideTopology { + wire_consumer_side_with_provider( + dial, + egress, + accept, + identity, + Arc::new(alkcall::core::auth::NoopIdentityProvider), + ) + .await +} + +/// The consumer-side topology with the producer-side dispatch's +/// identity provider configured (the CF-005 precedence (1) seam — the +/// payload `auth_token` → `identity_provider` resolution path the +/// cred-bearing door's token sessions ride). Everything else matches +/// [`wire_consumer_side_with`]. +pub async fn wire_consumer_side_with_provider( + dial: DialFn, + egress: Option, + accept: Option, + identity: Option, + provider: Arc, ) -> ConsumerSideTopology { let producer_op_registry = Arc::new(OperationRegistry::new()); let (producer_manager_tx, mut producer_manager_rx) = @@ -681,12 +703,14 @@ pub async fn wire_consumer_side_with( let resolved = Arc::new(identity.unwrap_or_else(consumer_identity)); let resolved_for_hook = Arc::clone(&resolved); + let provider_for_hook = Arc::clone(&provider); let dial_for_hook = Arc::clone(&dial); let install_hook: InstallChannelZero = Arc::new( move |manager: ChannelManager, channel0_conn: CoreConnection, _auth: AuthContext| { let registry = Arc::clone(&producer_op_registry); let manager_tx = producer_manager_tx.clone(); let witness = Arc::clone(&witness_for_hook); + let provider = Arc::clone(&provider_for_hook); let resolved = Arc::clone(&resolved_for_hook); let dial = Arc::clone(&dial_for_hook); let egress = egress.clone(); @@ -718,10 +742,7 @@ pub async fn wire_consumer_side_with( ) .expect("register socks5 openable (the consumer-side topology)"); let _ = manager_tx.send(manager).await; - let dp = Dispatcher::new( - registry, - Arc::new(alkcall::core::auth::NoopIdentityProvider), - ); + let dp = Dispatcher::new(registry, provider); dp.serve_single_stream(call_connection, reader, writer) .await; })