diff --git a/docs/architecture/client.md b/docs/architecture/client.md index 9b2856a..264e341 100644 --- a/docs/architecture/client.md +++ b/docs/architecture/client.md @@ -69,6 +69,39 @@ half-capability, which the family forbids (ADR-003 symmetry + consumers turn capabilities on/off per connection; the shape is an assembly-mutable acceptor registry). +> **The landed acceptor shape (`w4-reverse-listen` — recorded at +> landing):** the registry is **per-type fixed slots** — exactly the +> two `-R` data-leg types the client direction can serve +> (`forwarded-tcpip`, `forwarded-streamlocal`, both under `tunnel`; +> no map, no generic key), one acceptor trait per type; install is +> **refuse-never-replace** (the `ListenRouter::register` recorded +> posture, mirrored client-side); the slots are read **at arrival +> time** by the dial handler's hooks — a pre-install/unfeatured open +> is declined at arrival, never buffered. The payload an accepted +> open hands the assembly: the channel duplex (family-typed — the +> `-L` rows' `TunnelChannel` byte duplex, one type both legs ride) + +> the open's echo/originator fields; **the pump is the acceptor's** +> from the handoff on (the family handler shape) — the crate keeps no +> accepted-open ledger. Bounded bookkeeping, the cap relation: the +> registry bounded by construction (two slots); the +> **reverse-listen count** and the **accepted-open handoff count** +> are two dedicated counters, each against the same +> `Limits::max_concurrent_channels` ceiling (the server-side +> listener/ListenerCap posture mirrored — independent counts, one +> ceiling; the listens live at the peer, the counter exists so a dial +> cannot mint an unbounded listener set). The **forward-request rows +> gate on the registered acceptor** (the issuing-side disposition: +> a listen whose data leg cannot serve is never asked for — +> install-first, `OpError::NoAcceptor` pre-wire), plus +> `OpError::ListenRefused` (the peer's `RequestDenied` — denied +> listens and unknown-listener cancels, sshd raw-string parity) and +> `OpError::ListenCapExceeded` (the ceiling, pre-wire). The +> **runtime-Unix gate** rides the unix *listen* rows only (the bind +> is the platform-bound op; the `forwarded-streamlocal` ingestion is +> platform-neutral byte pumping). The `tcpip_forward` reply's mint +> rides only for `port = 0` requests (russh/sshd parity: an explicit +> port's bound *is* the requested port, which the row surfaces). + ## Lifecycle primitives (OQ-12 resolution) - Forward listeners return the **actually-bound port** (russh diff --git a/src/client.rs b/src/client.rs index ca12402..6a309d6 100644 --- a/src/client.rs +++ b/src/client.rs @@ -69,6 +69,10 @@ use std::sync::Arc; use russh::Disconnect; use russh::keys::PrivateKeyWithHashAlg; +#[cfg(feature = "tunnel")] +use crate::channel_cap::ChannelCap; +#[cfg(feature = "tunnel")] +use crate::client_accept::AcceptorRegistry; use crate::client_handler::ClientHandler; use crate::config::{DialConfig, Limits}; use crate::error::DialError; @@ -81,6 +85,14 @@ pub async fn dial(config: DialConfig) -> Result { let (host, port) = parse_dial_address(&config.address)?; let handler = ClientHandler::new(Arc::clone(&config.host_key_policy), host, port); + #[cfg(feature = "tunnel")] + let (handler, acceptors) = { + let acceptors = Arc::new(AcceptorRegistry::new()); + ( + handler.with_tunnel_surface(Arc::clone(&acceptors), config.limits), + acceptors, + ) + }; let russh_config = Arc::new(build_client_config(&config.limits)); let mut handle = russh::client::connect(russh_config, config.address.as_str(), handler).await?; @@ -103,6 +115,10 @@ pub async fn dial(config: DialConfig) -> Result { limits: config.limits, address: config.address, username: config.username, + #[cfg(feature = "tunnel")] + acceptors, + #[cfg(feature = "tunnel")] + listener_budget: ChannelCap::default(), }) } @@ -167,6 +183,19 @@ pub struct SshConnection { limits: Limits, address: String, username: String, + /// The `-R` leg's acceptor registry (the dial constructs it empty; + /// the assembly installs/uninstalls across the live connection — + /// see [`SshConnection::acceptors`] and + /// [`crate::client_accept`]`'s recorded decisions. + #[cfg(feature = "tunnel")] + acceptors: Arc, + /// The reverse-listen count's dedicated bound — admitted at the + /// listen rows, released at the confirmed cancels and the refused/ + /// wire-error paths (the cap relation recorded in + /// [`crate::client_accept`]; [`crate::channel_cap`]'s mutex makes + /// the counts sound under the caller's task). + #[cfg(feature = "tunnel")] + listener_budget: ChannelCap, } impl SshConnection { @@ -186,6 +215,25 @@ impl SshConnection { &self.username } + /// The `-R` leg's acceptor registry (the + /// `w4-reverse-listen` surface, `client_accept`'s module docs + /// carry the recorded decisions) — install/uninstall the + /// per-type acceptors across the live connection (the + /// late-install dynamic). The dial constructs it empty: nothing + /// server-initiated is served until a slot holds, fail-closed at + /// arrival time. + #[cfg(feature = "tunnel")] + pub fn acceptors(&self) -> &AcceptorRegistry { + &self.acceptors + } + + /// The reverse-listen count's shared handle — the op rows + /// ([`crate::client_accept`]) read it; private to the crate. + #[cfg(feature = "tunnel")] + pub(crate) fn listener_budget(&self) -> &ChannelCap { + &self.listener_budget + } + /// Whether the underlying russh session is closed. pub fn is_closed(&self) -> bool { self.handle.is_closed() @@ -223,12 +271,15 @@ impl SshConnection { impl std::fmt::Debug for SshConnection { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("SshConnection") + let mut debug = f.debug_struct("SshConnection"); + debug .field("address", &self.address) .field("username", &self.username) .field("limits", &self.limits) - .field("closed", &self.handle.is_closed()) - .finish() + .field("closed", &self.handle.is_closed()); + #[cfg(feature = "tunnel")] + debug.field("acceptors", &self.acceptors); + debug.finish() } } diff --git a/src/client_accept.rs b/src/client_accept.rs new file mode 100644 index 0000000..78ca5c3 --- /dev/null +++ b/src/client_accept.rs @@ -0,0 +1,624 @@ +//! The `-R` leg's client half (`w4-reverse-listen` — `client.md` +//! §op rows row 4 + §The acceptor): the hub's reverse direction in +//! family types. Two surfaces, the acceptor registry first: +//! +//! 1. **The forward request rows** — [`SshConnection::reverse_listen`] +//! / [`SshConnection::reverse_listen_cancel`] / the unix pair +//! (`reverse_listen_unix` / `reverse_listen_unix_cancel`): the +//! vanilla-server listen requests, the lifecycle primitives +//! (`client.md` §Lifecycle — the actually-bound port surfaced, +//! every listener explicitly cancellable; the *handles*' surface is +//! `w4-lifecycle`'s growth). +//! 2. **The acceptor registry** ([`AcceptorRegistry`]) — the +//! assembly-mutable, per-channel-type registry the dial surface's +//! russh handler routes **server-initiated `forwarded-*` opens** +//! through. Served only under a registered acceptor for that type +//! and under the `tunnel` feature; unregistered/unfeatured accepted +//! opens **decline cleanly** — never russh's accept-and-drop +//! default (ADR-003 §4, the client-side fail-closed acceptor; the +//! reverse-runner fact 1 disposition is ours). +//! +//! ## The recorded decisions (the task's decide + record items) +//! +//! - **Registry shape: per-type fixed slots, not a typed map** — the +//! client direction's registry-servable type set is closed and tiny +//! (exactly the `-R` data leg's two: `forwarded-tcpip` / +//! `forwarded-streamlocal`, both under `tunnel`); one trait per type +//! ([`TcpipForwardedAcceptor`] / [`StreamlocalForwardedAcceptor`]) +//! makes *which acceptor serves which type* a compile-time fact and +//! the fail-closed gate a one-slot read. A generic typed map would +//! add runtime dispatch (a typed key + an enum payload) for a set +//! that cannot grow without an ADR (any further server-initiated +//! type is a mapping, `client.md` §op rows). The harness's +//! `ForwardedAcceptor` vocabulary is grown, not forked: the payload +//! fields are the same shapes (the open's originator fields + the +//! channel duplex), the duplex now family-typed +//! ([`TunnelChannel`](crate::TunnelChannel) — the `-L` rows' byte +//! duplex, one type both legs ride). +//! - **Install is refuse-never-replace** ([`AcceptorRegistry:: +//! install_tcpip`]'s held-slot error): the same posture recorded +//! server-side on the `ListenRouter` at `w4-router-collision` — +//! a same-slot second install is an assembly program error, declined +//! never a quiet displacement of a live acceptor. Uninstall is +//! explicit, 1:1 with the install. +//! - **Late-install across a live connection** (the alkminer +//! `install_forward_sink` dynamic — reverse-runner fact 2): the +//! registry rides the connected [`SshConnection`] +//! ([`SshConnection::acceptors`]) — the dial constructs it empty, +//! the assembly installs/uninstalls whenever it decides. The open +//! hooks read the slot **at arrival time**, fail-closed: an open +//! arriving before its acceptor installs (or after an uninstall) +//! is **declined at arrival, never buffered** — no unclaimed-open +//! queue exists anywhere in this module's posture (an unclaimed +//! open's bytes would have to be buffered unboundedly to accept- +//! then-route; the decline is the clean, bounded answer). +//! - **The listen rows gate on the registered acceptor** — the +//! *issuing*-side disposition: a listen whose data leg cannot serve +//! is a half-capability (the peer stays bound, every arrival +//! declines), so a `reverse_listen` with no registered acceptor for +//! the type answers +//! [`OpError::NoAcceptor`](crate::OpError) **before any wire op** +//! (install-first is the recorded order; the rows never buffer an +//! outstanding listen against a later install — re-issuing is the +//! assembly's retry, the listen rows are stateless global requests). +//! - **Bounded bookkeeping (AGENTS.md convention 9) — the cap +//! relation** (the server-side listener-counter posture as the +//! template): the registry is bounded by construction (two fixed +//! slots). The per-connection **listener count** is a dedicated +//! in-crate counter admitting at the listen rows and releasing at +//! the confirmed cancels and the failed/wire-error paths — bounded +//! against the *same* `Limits::max_concurrent_channels` ceiling the +//! server direction's listener counter uses (independent counts, one +//! ceiling — never a map: the outstanding listens live at the peer; +//! the counter exists so a dial cannot mint an unbounded listener +//! set the registry would have to serve). The per-connection +//! **accepted-open handoff count** is the second dedicated counter +//! against the same ceiling (admitted in the dial handler's open +//! hook, accepted-with-the-count-held, released when the acceptor's +//! `accept` call completes — no accepted-open ledger anywhere +//! else: the pump is the acceptor's from the handoff on, the +//! channel's transport bounds are russh's windows, and the slot read +//! is the arrival admission). +//! - **The runtime-Unix gate (OQ-11 symmetry)**: the streamlocal +//! *listen* rows (`reverse_listen_unix` / `reverse_listen_unix_cancel`) +//! compile everywhere and +//! decline off-Unix before any wire op (the bind is the +//! platform-bound operation — the door puts nothing on the wire it +//! cannot track). The `forwarded-streamlocal` *open* hook is +//! universal: ingestion is platform-neutral byte pumping — the +//! unix-socket leg's socket is the peer-side's. +//! - **`OpError::ListenRefused`** — the peer's `RequestDenied` on +//! every form's decline (a denied listen, a cancel of an unknown +//! listener) — channel-scoped honesty: the connection stays usable +//! (the `From` mapping's recorded producer, `error.rs`). +//! +//! ## The data leg +//! +//! Per accepted open: the hook accepts the channel and hands the +//! assembly the open — the [`TunnelChannel`](crate::TunnelChannel) +//! duplex + the open's echo/originator fields (the payload types +//! below) — on a spawned task (the spawn-and-return rule's client +//! mirror: the hook runs in the packet loop and returns fast). The +//! pump is the acceptor's from the handoff — the family handler shape +//! (alktunnels' `ForwardedHandler` recipe, POC-3 §C.4); dropping the +//! duplex is wire-clean (russh closes the channel). +//! +//! ## Test surface +//! +//! The wire vehicle is the fake `sshd` double grown with the +//! server-initiated-forward surfaces (`tests/common/sshd_double.rs`) — +//! the double binds **real** sockets for its script accepts, opens the +//! forwarded channels toward the client per accepted connection, and +//! pumps them (a real `ssh -R`-shaped round trip in-process, the task's +//! recorded smoke posture). The full-circle alkssh-client ↔ +//! alkssh-server-direction smoke (both `-R` legs through the alkcall +//! fabric, the composition the assembly makes) rides the fabric +//! doubles in `tests/tunnel_mapping.rs` — the same doubles the +//! server-side `-R` rows use; real-`sshd` `-R` is wave-5 e2e +//! territory, never silently dropped. +//! +//! ## Smoke posture (recorded) +//! +//! The in-process smoke shape is a real `ssh -R`-shaped round trip: a +//! TCP connection to the forward's actually-bound port round-trips +//! through the registered acceptor's duplex (the reverse-runner's +//! hub-side proxy shape). The full-circle variant against the alkssh +//! server direction pins the client registry against the server +//! direction's real arrival pump (the echo/originator/mint shapes +//! `w3-tunnel-listen` landed); real-`sshd` `-R` is wave-5 e2e. + +use std::sync::Arc; + +use async_trait::async_trait; +use parking_lot::Mutex; +use russh::client::ChannelOpenHandle; + +use crate::TunnelChannel; +use crate::channel_cap::ChannelCap; +use crate::client::SshConnection; +use crate::error::OpError; +use crate::observe; + +/// An accepted server-initiated `forwarded-tcpip` open — the acceptor +/// registry's payload for the +/// [`TcpipForwardedAcceptor::accept`] call: the open's echo fields +/// (the listen endpoint the arrival belongs to — the assembly's +/// routing key; the OpenSSH matcher parity verbatim) + the originator +/// (the arrival's peer, as the wire carried it) + the channel duplex. +pub struct ForwardedTcpip { + /// The connected address the open echoed — the `reverse_listen` + /// wire string (sshd parity). + pub connected_address: String, + /// The actually-bound port of the minted listener — for a + /// `port = 0` mint this is the server-picked number, never `0`. + pub connected_port: u32, + /// The arrival's peer address (the wire originator). + pub originator_address: String, + /// The arrival's peer port (the wire originator). + pub originator_port: u32, + /// The accepted channel's byte duplex — the acceptor owns the + /// pump from the handoff on (dropping it is wire-clean). + pub stream: TunnelChannel, +} + +impl std::fmt::Debug for ForwardedTcpip { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ForwardedTcpip") + .field("connected_address", &self.connected_address) + .field("connected_port", &self.connected_port) + .field("originator_address", &self.originator_address) + .field("originator_port", &self.originator_port) + .finish_non_exhaustive() + } +} + +/// An accepted server-initiated `forwarded-streamlocal` open — the +/// streamlocal acceptor's payload (the echoed socket path + the +/// channel duplex). +pub struct ForwardedStreamlocal { + /// The socket path the open echoed — the `reverse_listen_unix` + /// wire string (sshd parity). + pub socket_path: String, + /// The accepted channel's byte duplex — the acceptor owns the + /// pump from the handoff on (dropping it is wire-clean). + pub stream: TunnelChannel, +} + +impl std::fmt::Debug for ForwardedStreamlocal { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ForwardedStreamlocal") + .field("socket_path", &self.socket_path) + .finish_non_exhaustive() + } +} + +/// The `forwarded-tcpip` acceptor — the registry's per-type slot +/// trait. `accept` runs on the spawned handoff task (never the +/// packet loop): the assembly's pump/home of the open lives inside +/// it; the crate releases its handoff counter when the call returns, +/// so the family shape (hand off to a queue, return) keeps the +/// counter's meaning "open accepted, handoff in flight". +#[async_trait] +pub trait TcpipForwardedAcceptor: Send + Sync { + /// Serves one accepted open. Never invoked for an unregistered + /// type (the registry gate is first); the open is already accepted + /// on the wire — refusing one from here is a *data-plane* choice + /// (drop the duplex; the channel closes on it), never a wire + /// open-decline. + async fn accept(&self, open: ForwardedTcpip); +} + +/// The `forwarded-streamlocal` acceptor — the unix slot's trait +/// (see [`TcpipForwardedAcceptor`] for the call-shape contract). +#[async_trait] +pub trait StreamlocalForwardedAcceptor: Send + Sync { + /// Serves one accepted open (platform-neutral pumping — see the + /// module docs' unix-gate record). + async fn accept(&self, open: ForwardedStreamlocal); +} + +/// The held-slot refusal (`AcceptorRegistry::install_*`) — the +/// refuse-never-replace posture (the server-side `ListenRouter`'s +/// recorded sibling): the slot already holds a live acceptor; a second +/// install is the assembly's program error, declined, never a quiet +/// displacement. An explicit `uninstall_*` frees the slot. +#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)] +#[error("an acceptor is already registered for this channel type")] +pub struct AcceptorHeld; + +/// The assembly-mutable, per-channel-type acceptor registry — the +/// `-R` data leg's fail-closed gate (see the module docs for the +/// recorded decisions). Constructed empty at the dial +/// ([`SshConnection::acceptors`], the live-connection accessor); +/// install/uninstall is the assembly's to drive whenever it decides. +pub struct AcceptorRegistry { + tcpip: Mutex>>, + streamlocal: Mutex>>, +} + +impl AcceptorRegistry { + /// The empty registry — the dial's construct (nothing is served + /// until an acceptor installs; ADR-003 §4's fail-closed default). + pub fn new() -> Self { + Self { + tcpip: Mutex::new(None), + streamlocal: Mutex::new(None), + } + } + + /// Installs the `forwarded-tcpip` acceptor. Refuse-never-replace: + /// a held slot declines ([`AcceptorHeld`]) — uninstall first if a + /// replacement is meant (the late-install toggle's other half). + pub fn install_tcpip( + &self, + acceptor: Arc, + ) -> Result<(), AcceptorHeld> { + let mut slot = self.tcpip.lock(); + if slot.is_some() { + return Err(AcceptorHeld); + } + *slot = Some(acceptor); + Ok(()) + } + + /// Uninstalls the `forwarded-tcpip` acceptor — `true` when one was + /// held (the toggle's honest report). In-flight opens already + /// accepted keep pumping (their handoff happened); arrivals after + /// this decline at arrival time (fail-closed, no buffering). + pub fn uninstall_tcpip(&self) -> bool { + self.tcpip.lock().take().is_some() + } + + /// Installs the `forwarded-streamlocal` acceptor (see + /// [`Self::install_tcpip`] for the refuse-never-replace posture). + pub fn install_streamlocal( + &self, + acceptor: Arc, + ) -> Result<(), AcceptorHeld> { + let mut slot = self.streamlocal.lock(); + if slot.is_some() { + return Err(AcceptorHeld); + } + *slot = Some(acceptor); + Ok(()) + } + + /// Uninstalls the `forwarded-streamlocal` acceptor (see + /// [`Self::uninstall_tcpip`] for the honest-report posture). + pub fn uninstall_streamlocal(&self) -> bool { + self.streamlocal.lock().take().is_some() + } + + /// The arrival-time slot reads (the open hooks' gates) — cloned + /// out so the spawned handoff task owns its acceptor handle. + pub(crate) fn tcpip_acceptor(&self) -> Option> { + self.tcpip.lock().clone() + } + + pub(crate) fn streamlocal_acceptor(&self) -> Option> { + self.streamlocal.lock().clone() + } +} + +impl Default for AcceptorRegistry { + fn default() -> Self { + Self::new() + } +} + +impl std::fmt::Debug for AcceptorRegistry { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("AcceptorRegistry") + .field("tcpip", &self.tcpip.lock().is_some()) + .field("streamlocal", &self.streamlocal.lock().is_some()) + .finish() + } +} + +// --------------------------------------------------------------------------- +// The server-initiated opens' handoff (the dial handler's routed paths) +// --------------------------------------------------------------------------- + +/// The accepted `forwarded-tcpip` open's handoff task — the +/// spawn-and-return rule's client mirror (ADR-006 decision 4; the +/// hook runs in the packet loop and returns fast): accepts the open on +/// the wire, converts the channel to the family duplex, delivers the +/// payload to the assembly's acceptor, and releases the handoff +/// counter when the acceptor's call completes (the recorded release +/// site — the count's meaning is "accepted, handoff in flight"). +#[cfg_attr(feature = "tunnel", allow(clippy::too_many_arguments))] +pub(crate) async fn handoff_forwarded_tcpip( + acceptor: Arc, + open_cap: Arc, + channel: russh::Channel, + reply: ChannelOpenHandle, + connected_address: String, + connected_port: u32, + originator_address: String, + originator_port: u32, +) { + let channel_number = channel.id().number(); + observe::serving_path_started("forwarded-tcpip", channel_number, None); + reply.accept().await; + let open = ForwardedTcpip { + connected_address, + connected_port, + originator_address, + originator_port, + stream: TunnelChannel { + stream: channel.into_stream(), + }, + }; + acceptor.accept(open).await; + open_cap.release(); + observe::serving_path_ended("forwarded-tcpip", channel_number, None); +} + +/// The accepted `forwarded-streamlocal` open's handoff task — the +/// unix slot's mirror of [`handoff_forwarded_tcpip`]. +pub(crate) async fn handoff_forwarded_streamlocal( + acceptor: Arc, + open_cap: Arc, + channel: russh::Channel, + reply: ChannelOpenHandle, + socket_path: String, +) { + let channel_number = channel.id().number(); + observe::serving_path_started("forwarded-streamlocal", channel_number, None); + reply.accept().await; + let open = ForwardedStreamlocal { + socket_path, + stream: TunnelChannel { + stream: channel.into_stream(), + }, + }; + acceptor.accept(open).await; + open_cap.release(); + observe::serving_path_ended("forwarded-streamlocal", channel_number, None); +} + +// --------------------------------------------------------------------------- +// The reverse-listen rows (under `tunnel`) +// --------------------------------------------------------------------------- + +impl SshConnection { + /// The reverse-listen row (`-R`): asks the vanilla server to listen + /// on (address, port) and returns the **actually-bound port** — + /// for `port = 0` this is the peer-picked number (`client.md` + /// §Lifecycle: the bound-port primitive, the port-0 mint); for an + /// explicit port the wire's success reply carries no number (russh/ + /// sshd parity — the reply's data only exists for mints) and the + /// bound port *is* the requested one, which the row surfaces. + /// + /// Gates, both **before any wire op**: the `forwarded-tcpip` + /// acceptor must be registered + /// ([`OpError::NoAcceptor`](crate::OpError) — a listen whose data + /// leg cannot serve is a half-capability) and the connection's + /// listener count must have room + /// ([`OpError::ListenCapExceeded`](crate::OpError) — the + /// dedicated counter sharing + /// [`Limits::max_concurrent_channels`](crate::Limits); see the + /// module docs' cap relation). The peer's refusal is + /// [`OpError::ListenRefused`](crate::OpError) (the wire's + /// `RequestDenied`) and releases the count; the connection stays + /// usable after every refusal class. The listener is explicitly + /// cancellable ([`SshConnection::reverse_listen_cancel`]; it is + /// not implicitly unlistened by anything but the connection's + /// end). + pub async fn reverse_listen( + &self, + address: impl Into, + port: u32, + ) -> Result { + if self.acceptors().tcpip_acceptor().is_none() { + return Err(OpError::NoAcceptor); + } + if !self + .listener_budget() + .admit(self.limits().max_concurrent_channels) + { + return Err(OpError::ListenCapExceeded); + } + let bound = match self.russh_handle().tcpip_forward(address, port).await { + Ok(minted) if port == 0 => minted, + Ok(_) => port, + Err(error) => { + self.listener_budget().release(); + return Err(OpError::from(error)); + } + }; + Ok(bound) + } + + /// The reverse-listen cancel row (`-R` teardown): the explicit + /// cancellation of one bound listener (the lifecycle primitive: + /// every listener explicitly cancellable, the wire-honest path — + /// the peer tears its listen down, the bound port frees). Matches + /// by the raw wire strings (sshd parity). The peer's refusal + /// (unknown listener) is [`OpError::ListenRefused`](crate::OpError); + /// a confirmed cancel releases the listener count. + pub async fn reverse_listen_cancel( + &self, + address: impl Into, + port: u32, + ) -> Result<(), OpError> { + match self + .russh_handle() + .cancel_tcpip_forward(address, port) + .await + { + Ok(()) => { + self.listener_budget().release(); + Ok(()) + } + Err(error) => Err(OpError::from(error)), + } + } + + /// The streamlocal reverse-listen row (the unix `-R`): asks the + /// vanilla server to listen on the unix socket at `socket_path`, + /// minting no port (the global request's success is the outcome; + /// the accepted opens' echo carries the path). Same gates as + /// [`SshConnection::reverse_listen`] for the + /// `forwarded-streamlocal` slot, plus the **runtime-Unix gate** + /// (OQ-11: off-Unix, [`OpError::UnixUnavailable`](crate::OpError) + /// before any wire op — the bind is the platform-bound operation; + /// the channel ingestion is platform-neutral). + #[cfg_attr(not(unix), allow(unused_variables))] + pub async fn reverse_listen_unix(&self, socket_path: impl Into) -> Result<(), OpError> { + #[cfg(not(unix))] + { + return Err(OpError::UnixUnavailable); + } + #[cfg(unix)] + { + if self.acceptors().streamlocal_acceptor().is_none() { + return Err(OpError::NoAcceptor); + } + if !self + .listener_budget() + .admit(self.limits().max_concurrent_channels) + { + return Err(OpError::ListenCapExceeded); + } + match self.russh_handle().streamlocal_forward(socket_path).await { + Ok(()) => Ok(()), + Err(error) => { + self.listener_budget().release(); + Err(OpError::from(error)) + } + } + } + } + + /// The streamlocal cancel row (matches by the socket path, sshd + /// parity — see [`SshConnection::reverse_listen_cancel`]). + #[cfg_attr(not(unix), allow(unused_variables))] + pub async fn reverse_listen_unix_cancel( + &self, + socket_path: impl Into, + ) -> Result<(), OpError> { + #[cfg(not(unix))] + { + return Err(OpError::UnixUnavailable); + } + #[cfg(unix)] + { + match self + .russh_handle() + .cancel_streamlocal_forward(socket_path) + .await + { + Ok(()) => { + self.listener_budget().release(); + Ok(()) + } + Err(error) => Err(OpError::from(error)), + } + } + } +} + +#[cfg(test)] +mod tests { + use super::AcceptorHeld; + use crate::client_accept::{ + AcceptorRegistry, ForwardedStreamlocal, ForwardedTcpip, StreamlocalForwardedAcceptor, + TcpipForwardedAcceptor, + }; + use std::sync::Arc; + + struct StubTcpip; + + #[async_trait::async_trait] + impl TcpipForwardedAcceptor for StubTcpip { + async fn accept(&self, _open: ForwardedTcpip) {} + } + + struct StubStreamlocal; + + #[async_trait::async_trait] + impl StreamlocalForwardedAcceptor for StubStreamlocal { + async fn accept(&self, _open: ForwardedStreamlocal) {} + } + + /// - *Contract stamp:* the recorded refuse-never-replace posture — + /// a held slot declines the second install (never a quiet + /// displacement of a live acceptor); uninstall frees it. + #[test] + fn install_refuses_a_held_slot_and_uninstall_frees_it() { + let registry = AcceptorRegistry::new(); + let first: Arc = Arc::new(StubTcpip); + assert!(registry.install_tcpip(Arc::clone(&first)).is_ok()); + assert!( + registry.install_tcpip(Arc::new(StubTcpip)).is_err(), + "a held slot is refused, never replaced" + ); + let held = registry.tcpip_acceptor().expect("the first stays held"); + assert!(Arc::ptr_eq(&held, &first)); + + assert!(registry.uninstall_tcpip()); + assert!( + !registry.uninstall_tcpip(), + "the second uninstall is the honest `false`" + ); + assert!(registry.tcpip_acceptor().is_none()); + assert!( + registry.install_tcpip(first).is_ok(), + "an uninstalled slot installs again (the late-install toggle)" + ); + } + + /// - *Contract stamp:* the two slots are independent (the + /// per-type-registry decision) — the empty streamlocal slot + /// stays unfeatured while the tcpip slot serves. + #[test] + fn the_slots_are_independent() { + let registry = AcceptorRegistry::new(); + assert!(registry.tcpip_acceptor().is_none()); + assert!(registry.streamlocal_acceptor().is_none()); + + registry + .install_tcpip(Arc::new(StubTcpip)) + .expect("the tcp slot installs"); + assert!(registry.tcpip_acceptor().is_some()); + assert!( + registry.streamlocal_acceptor().is_none(), + "the streamlocal slot is untouched" + ); + + registry + .install_streamlocal(Arc::new(StubStreamlocal)) + .expect("the streamlocal slot installs independently"); + assert!(registry.streamlocal_acceptor().is_some()); + } + + /// - *Contract stamp:* the Debug shape (the installed-slot fact, + /// never internals). + #[test] + fn the_registry_debug_shapes_the_slots() { + let registry = AcceptorRegistry::new(); + let debug = format!("{registry:?}"); + assert!( + debug.contains("tcpip: false") && debug.contains("streamlocal: false"), + "{debug}" + ); + + registry + .install_tcpip(Arc::new(StubTcpip)) + .expect("installs"); + let debug = format!("{registry:?}"); + assert!(debug.contains("tcpip: true"), "{debug}"); + } + + /// - *Contract stamp:* the held-slot error's honest shape (no + /// `Other`, no catch-all). + #[test] + fn acceptor_held_error_shapes() { + let held = AcceptorHeld; + assert_eq!( + held.to_string(), + "an acceptor is already registered for this channel type" + ); + } +} diff --git a/src/client_handler.rs b/src/client_handler.rs index 273c055..29314e8 100644 --- a/src/client_handler.rs +++ b/src/client_handler.rs @@ -20,25 +20,39 @@ //! the client-side symmetry rule): russh's default posture for every //! `server_channel_open_*` hook is accept-and-drop — a //! half-capability the family forbids. Every hook is overridden -//! explicit-over-inherited and answers +//! explicit-over-inherited. The `forwarded-tcpip` / +//! `forwarded-streamlocal` hooks (the `-R` data leg's types) route +//! through the **acceptor registry** ([`crate::client_accept`] — +//! the `w4-reverse-listen` surface): served only under a registered +//! acceptor for that type and under the `tunnel` feature (the +//! [`with_tunnel_surface`](ClientHandler::with_tunnel_surface) +//! wiring; a build without `tunnel` has no registry — every hook +//! declines); unregistered/unfeatured opens answer //! [`ChannelOpenFailure::AdministrativelyProhibited`] — a clean //! wire-visible refusal the peer observes as an //! unconfirmed open, never an accepted-then-dropped channel. The -//! unknown-type gate +//! accepted open's handoff spawns (the spawn-and-return rule's +//! client mirror): the hook accepts and returns fast, the duplex + +//! the open's fields ride the payload to the assembly's acceptor +//! call on its own task. Every other type has no registry slot — +//! the unknown-type gate //! ([`Handler::should_accept_unknown_server_channel`]) answers //! `false` (the hook's reject-before-open form; its `_unknown` //! sibling stays a defensive reject — wire-reachable only if that -//! gate ever changed, also fail-closed). The acceptor registry -//! that serves these opens under its feature is -//! `w4-reverse-listen`'s surface; this skeleton pins the decline-all -//! default every unregistered open gets. +//! gate ever changed, also fail-closed), and the never-feature rows +//! (x11, agent) decline at any configuration. //! -//! Data-plane hooks stay russh's defaults: every server-initiated -//! channel is declined, so no data for one can ever arrive. The -//! `w4-op-rows` / `w4-reverse-listen` tasks grow this skeleton in -//! place (the wave's sequential client chain); the decline events ride -//! the crate's [`crate::observe`] vocabulary (`decline_open`, -//! reason `no-acceptor`). +//! The accepted-open handoff carries a **bounded bookkeeping** +//! counter (`AGENTS.md` convention 9, the cap relation recorded in +//! [`crate::client_accept`]): a dedicated per-connection counter — +//! sharing the dial's `Limits::max_concurrent_channels` ceiling the +//! server direction's per-session cap uses — admitted in the hook +//! before the accept, released when the acceptor's call completes in +//! the spawned task; a full counter declines the arrival cleanly +//! (`cap-exceeded`). + +#[cfg(feature = "tunnel")] +use std::sync::Arc; use russh::ChannelOpenFailure; use russh::client::ChannelOpenHandle; @@ -46,10 +60,30 @@ use russh::client::Msg; use russh::keys::PublicKeyOrCertificate; use russh::{Channel, ChannelId}; +#[cfg(feature = "tunnel")] +use crate::channel_cap::ChannelCap; +#[cfg(feature = "tunnel")] +use crate::client_accept::{ + AcceptorRegistry, handoff_forwarded_streamlocal, handoff_forwarded_tcpip, +}; +#[cfg(feature = "tunnel")] +use crate::config::Limits; use crate::host_key::{HostKeyPolicy, PresentedHostKey}; use crate::observe; use crate::{DialError, fingerprint}; +/// The `-R` leg's per-connection client-handler state (the +/// `w4-reverse-listen` surface, under `tunnel`): the acceptor registry +/// the two forwarded hooks route through, plus the accepted-open +/// handoff counter and its ceiling (the dial's +/// `Limits::max_concurrent_channels`). +#[cfg(feature = "tunnel")] +pub(crate) struct TunnelSurface { + acceptors: Arc, + limits: Limits, + open_cap: std::sync::Arc, +} + /// The per-connection client handler the dial surface constructs — /// see the module docs. Test surface /// (`#[doc(hidden)]` re-export from the crate root): the contract @@ -59,6 +93,11 @@ pub struct ClientHandler { policy: std::sync::Arc, host: String, port: u16, + /// The `-R` surface under `tunnel` (`Some` — the dial wires it; + /// the bare-hook tests construct without it to pin the decline-all + /// default posture in the featured builds too). + #[cfg(feature = "tunnel")] + tunnel: Option, } impl ClientHandler { @@ -74,6 +113,24 @@ impl ClientHandler { policy, host: host.into(), port, + #[cfg(feature = "tunnel")] + tunnel: None, + } + } + + /// Wires the `-R` surface (the dial's step under `tunnel`): the + /// acceptor registry (constructed empty at the dial; the assembly + /// installs through the connection's accessor) + the + /// accepted-open counter against the limits' channel ceiling. + #[cfg(feature = "tunnel")] + pub fn with_tunnel_surface(self, acceptors: Arc, limits: Limits) -> Self { + Self { + tunnel: Some(TunnelSurface { + acceptors, + limits, + open_cap: Arc::new(ChannelCap::default()), + }), + ..self } } } @@ -125,39 +182,138 @@ impl russh::client::Handler for ClientHandler { Ok(()) } + /// The `forwarded-tcpip` hook (the `-R` data leg's gate): the + /// registry's slot read at arrival time — registered + room → + /// accept and hand the assembly the open on a spawned task; no + /// acceptor / no room → the clean decline, never accept-and-drop. async fn server_channel_open_forwarded_tcpip( &mut self, channel: Channel, - _connected_address: &str, - _connected_port: u32, - _originator_address: &str, - _originator_port: u32, + #[cfg_attr(not(feature = "tunnel"), allow(unused_variables))] connected_address: &str, + #[cfg_attr(not(feature = "tunnel"), allow(unused_variables))] connected_port: u32, + #[cfg_attr(not(feature = "tunnel"), allow(unused_variables))] originator_address: &str, + #[cfg_attr(not(feature = "tunnel"), allow(unused_variables))] originator_port: u32, reply: ChannelOpenHandle, _session: &mut russh::client::Session, ) -> Result<(), Self::Error> { - observe::decline_open("forwarded-tcpip", "no-acceptor", channel.id().number()); - reply - .reject(ChannelOpenFailure::AdministrativelyProhibited) - .await; - Ok(()) + #[cfg(not(feature = "tunnel"))] + { + observe::decline_open("forwarded-tcpip", "no-acceptor", channel.id().number()); + reply + .reject(ChannelOpenFailure::AdministrativelyProhibited) + .await; + Ok(()) + } + #[cfg(feature = "tunnel")] + { + let Some(surface) = self.tunnel.as_ref() else { + observe::decline_open("forwarded-tcpip", "no-acceptor", channel.id().number()); + reply + .reject(ChannelOpenFailure::AdministrativelyProhibited) + .await; + return Ok(()); + }; + let Some(acceptor) = surface.acceptors.tcpip_acceptor() else { + observe::decline_open("forwarded-tcpip", "no-acceptor", channel.id().number()); + reply + .reject(ChannelOpenFailure::AdministrativelyProhibited) + .await; + return Ok(()); + }; + if !surface + .open_cap + .admit(surface.limits.max_concurrent_channels) + { + observe::decline_open("forwarded-tcpip", "cap-exceeded", channel.id().number()); + reply + .reject(ChannelOpenFailure::AdministrativelyProhibited) + .await; + return Ok(()); + } + tokio::spawn(handoff_forwarded_tcpip( + acceptor, + Arc::clone(&surface.open_cap), + channel, + reply, + connected_address.to_owned(), + connected_port, + originator_address.to_owned(), + originator_port, + )); + Ok(()) + } } + /// The `forwarded-streamlocal` hook — the unix-slot's mirror of + /// the tcpip gate (the ingestion is platform-neutral: no runtime + /// gate here; the listen rows carry the OQ-11 gate on the bind — + /// the recorded asymmetry, see [`crate::client_accept`]). async fn server_channel_open_forwarded_streamlocal( &mut self, channel: Channel, - _socket_path: &str, + #[cfg_attr(not(feature = "tunnel"), allow(unused_variables))] socket_path: &str, reply: ChannelOpenHandle, _session: &mut russh::client::Session, ) -> Result<(), Self::Error> { - observe::decline_open( - "forwarded-streamlocal", - "no-acceptor", - channel.id().number(), - ); - reply - .reject(ChannelOpenFailure::AdministrativelyProhibited) - .await; - Ok(()) + #[cfg(not(feature = "tunnel"))] + { + observe::decline_open( + "forwarded-streamlocal", + "no-acceptor", + channel.id().number(), + ); + reply + .reject(ChannelOpenFailure::AdministrativelyProhibited) + .await; + Ok(()) + } + #[cfg(feature = "tunnel")] + { + let Some(surface) = self.tunnel.as_ref() else { + observe::decline_open( + "forwarded-streamlocal", + "no-acceptor", + channel.id().number(), + ); + reply + .reject(ChannelOpenFailure::AdministrativelyProhibited) + .await; + return Ok(()); + }; + let Some(acceptor) = surface.acceptors.streamlocal_acceptor() else { + observe::decline_open( + "forwarded-streamlocal", + "no-acceptor", + channel.id().number(), + ); + reply + .reject(ChannelOpenFailure::AdministrativelyProhibited) + .await; + return Ok(()); + }; + if !surface + .open_cap + .admit(surface.limits.max_concurrent_channels) + { + observe::decline_open( + "forwarded-streamlocal", + "cap-exceeded", + channel.id().number(), + ); + reply + .reject(ChannelOpenFailure::AdministrativelyProhibited) + .await; + return Ok(()); + } + tokio::spawn(handoff_forwarded_streamlocal( + acceptor, + Arc::clone(&surface.open_cap), + channel, + reply, + socket_path.to_owned(), + )); + Ok(()) + } } async fn server_channel_open_direct_tcpip( @@ -247,10 +403,10 @@ impl russh::client::Handler for ClientHandler { /// the hooks directly where the wire cannot reach a hook on its own. #[cfg(test)] mod tests { - use std::sync::Arc; - use russh::client::Handler as _; + use std::sync::Arc; + use super::ClientHandler; use crate::host_key::HostKeyPolicy; diff --git a/src/client_ops.rs b/src/client_ops.rs index 4b2b7da..5825900 100644 --- a/src/client_ops.rs +++ b/src/client_ops.rs @@ -541,7 +541,11 @@ async fn write_exit_chunk( /// surface — the duplex's `poll_shutdown` carries the wire EOF. #[cfg(feature = "tunnel")] pub struct TunnelChannel { - stream: russh::ChannelStream, + /// The russh channel's byte duplex (`pub(crate)` — the `-R` leg's + /// acceptor payloads construct the type over a server-initiated + /// open's `into_stream()` conversion; downstream crates see the + /// type opaque as ever). + pub(crate) stream: russh::ChannelStream, } #[cfg(feature = "tunnel")] @@ -836,7 +840,9 @@ async fn git_output_pump( #[cfg(all(test, any(feature = "session", feature = "git", feature = "tunnel")))] mod tests { - use super::{family_exit_code, open_error}; + #[cfg(any(feature = "session", feature = "git"))] + use super::family_exit_code; + use super::open_error; use crate::error::OpError; use russh::ChannelOpenFailure; @@ -865,6 +871,7 @@ mod tests { /// - *Contract stamp:* the family exit-currency clamp (§the /// family plane) — ordinary codes ride identity, beyond-range /// ones narrow (never invent). + #[cfg(any(feature = "session", feature = "git"))] #[test] fn family_exit_code_narrows_beyond_the_i32_range() { assert_eq!(family_exit_code(0), 0); diff --git a/src/error.rs b/src/error.rs index 810469a..2d42eb2 100644 --- a/src/error.rs +++ b/src/error.rs @@ -338,6 +338,36 @@ pub enum OpError { #[error("unix socket forwarding is unavailable on this platform")] UnixUnavailable, + /// The reverse-forward op's channel type has **no acceptor + /// registered** (ADR-003 §4's client-side gate surfaced to the + /// *issuing* side: a reverse listen whose data leg could not serve + /// arrivals is issued never — the half-capability posture, + /// answered before any wire op). Featured-but-unregistered is the + /// exact class: the unset-declines convention mirrored (the + /// acceptor slots are assembly-mutable, install-first is the + /// recorded order). + #[error("no acceptor is registered for the op's forwarded channel type")] + NoAcceptor, + + /// The peer refused the op's reverse-forward global request (the + /// `RequestDenied` wire decline class) — a listen denial or the + /// cancel forms' unknown-listener answer (sshd parity: match by + /// the raw wire strings). Channel-scoped honesty: the connection + /// stays usable (row-pinned), and the wire's own reason classes + /// are never invented here — the closed family class mirrors the + /// wire shape, `RequestRefused`'s global-request counterpart. + #[error("the peer refused the reverse-forward request")] + ListenRefused, + + /// The op's reverse-listen admission hit the connection's + /// dedicated listener ceiling (the listener counter sharing + /// `Limits::max_concurrent_channels` — the server-side + /// listener-counter posture mirrored; the cap relation recorded in + /// `client_accept.rs`) — answered **before any wire op**: a + /// request the budget cannot track is issued never. + #[error("the connection's reverse-listen count is at its ceiling")] + ListenCapExceeded, + /// The sftp route is the OQ-13 decline placeholder (the feature /// exists, the route does not until alkfs lands, on either /// direction) — answered before any wire op, never a @@ -352,14 +382,18 @@ pub enum OpError { } impl From for OpError { - /// The one mirror point: a typed open's rejection is the wire - /// shape `russh::Error::ChannelOpenFailure` and maps to + /// The mirror points: a typed open's rejection is the wire shape + /// `russh::Error::ChannelOpenFailure` and maps to /// [`OpError::OpenDeclined`] (family-mirrored - /// [`OpenDeclineReason`]); every other russh error stays the + /// [`OpenDeclineReason`]); a denied reverse-forward *global* + /// request (`RequestDenied` — the listen/cancel forms' only wire + /// decline shape at this pin, the rows' only producer) maps to + /// [`OpError::ListenRefused`]; every other russh error stays the /// transparent [`OpError::Wire`]. fn from(error: russh::Error) -> Self { match error { russh::Error::ChannelOpenFailure(failure) => Self::OpenDeclined(failure.into()), + russh::Error::RequestDenied => Self::ListenRefused, other => Self::Wire(other), } } @@ -586,6 +620,9 @@ mod tests { OpError::CommandTooLong, OpError::InvalidRepoId, OpError::UnixUnavailable, + OpError::NoAcceptor, + OpError::ListenRefused, + OpError::ListenCapExceeded, OpError::SftpNotServed, OpError::Wire(russh::Error::Disconnect), ]); @@ -601,6 +638,9 @@ mod tests { OpError::CommandTooLong => "command-too-long", OpError::InvalidRepoId => "invalid-repo-id", OpError::UnixUnavailable => "unix-unavailable", + OpError::NoAcceptor => "no-acceptor", + OpError::ListenRefused => "listen-refused", + OpError::ListenCapExceeded => "listen-cap-exceeded", OpError::SftpNotServed => "sftp-not-served", OpError::Wire(_) => "wire", }; @@ -609,6 +649,16 @@ mod tests { } } + /// - *Contract stamp:* the reverse-forward global request's wire + /// decline class (`RequestDenied` — the listen/cancel forms' + /// only producer shape at this pin): `ListenRefused`, never the + /// transparent wire class; the typed-open mapping unchanged. + #[test] + fn op_error_request_denied_maps_to_listen_refused() { + let mapped: OpError = russh::Error::RequestDenied.into(); + assert!(matches!(mapped, OpError::ListenRefused), "got {mapped:?}"); + } + /// - *Contract stamp:* POC-1 §6 — a rejected channel open /// surfaces on the awaiting typed call as /// `russh::Error::ChannelOpenFailure(..)`; the op rows map it diff --git a/src/lib.rs b/src/lib.rs index 2eca99c..694d8a4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -20,14 +20,16 @@ //! //! Architecture: `docs/architecture/` (phase 1 — ADRs 001–006 accepted; //! the foundations, the server stack, the server-direction mappings, and -//! the client dial + op rows are landed through wave 4; the `-R` leg and -//! lifecycle primitives, the validation suite, and the release are later -//! implementation waves). See also `docs/research/`, +//! the client dial + op rows + the `-R` acceptor leg are landed through +//! wave 4; the lifecycle primitives, the validation suite, and the +//! release are later implementation waves). See also `docs/research/`, //! `docs/plans/implementation.md`, and family ADR-003. mod adapter; mod channel_cap; mod client; +#[cfg(feature = "tunnel")] +mod client_accept; mod client_handler; mod client_ops; mod config; @@ -49,13 +51,22 @@ mod tty_map; mod tunnel_map; pub use adapter::SshAdapter; +pub use client::SshConnection; +pub use client::dial; /// Contract-test surface (`#[doc(hidden)]`): the dial surface's /// config-mapping and handler-skeleton mechanics (the `serve.rs` /// §Test-surface mirror; see `client.rs` §Test surface). Not /// documented API. #[doc(hidden)] pub use client::russh_client_config; -pub use client::{SshConnection, dial}; +/// The `-R` leg's acceptor registry + the reverse-listen rows' +/// payload/acceptor types (`w4-reverse-listen` — under `tunnel`; the +/// recorded decisions live in the `client_accept` module docs). +#[cfg(feature = "tunnel")] +pub use client_accept::{ + AcceptorHeld, AcceptorRegistry, ForwardedStreamlocal, ForwardedTcpip, + StreamlocalForwardedAcceptor, TcpipForwardedAcceptor, +}; #[doc(hidden)] pub use client_handler::ClientHandler; /// The client op rows' surfaces (`w4-op-rows`): diff --git a/tasks/implementation/w4-reverse-listen.md b/tasks/implementation/w4-reverse-listen.md index 2986465..4d984a6 100644 --- a/tasks/implementation/w4-reverse-listen.md +++ b/tasks/implementation/w4-reverse-listen.md @@ -1,7 +1,7 @@ --- id: w4-reverse-listen name: Wave 4 — the client -R leg (forward requests + the acceptor registry) -status: pending +status: completed depends_on: [w4-op-rows] scope: moderate risk: high @@ -65,21 +65,21 @@ registry that serves: ## Acceptance Criteria -- [ ] `tcpip_forward` / `streamlocal_forward` (+ cancel forms) land +- [x] `tcpip_forward` / `streamlocal_forward` (+ cancel forms) land with the actually-bound port surfaced (the port-0 mint row); the runtime-Unix gate compiles everywhere, declines cleanly off-Unix -- [ ] The acceptor registry lands (shape decided + recorded): served +- [x] The acceptor registry lands (shape decided + recorded): served only under a registered acceptor under `tunnel`; unregistered/unfeatured server-initiated opens declined cleanly (never accept-and-drop) — wire rows per build -- [ ] Late-install works across a live connection; a pre-install +- [x] Late-install works across a live connection; a pre-install arrival is declined (decided + recorded, no unclaimed buffering) -- [ ] The data leg hands accepted opens to the assembly handler; +- [x] The data leg hands accepted opens to the assembly handler; bookkeeping bounded (the cap relation decided + recorded) -- [ ] Per-build contract rows green through the server double; the +- [x] Per-build contract rows green through the server double; the smoke posture decided + recorded -- [ ] `cargo test` (+ per-feature + `--all-features`), clippy ×2 +- [x] `cargo test` (+ per-feature + `--all-features`), clippy ×2 (`-D warnings`), fmt, `cargo doc --no-deps` pass ## References @@ -104,6 +104,113 @@ registry that serves: > To be filled by implementation agent +Implementation decisions (also recorded in `src/client_accept.rs`'s +module docs and `client.md` §The acceptor's landing block): + +- **Registry shape — per-type fixed slots, not a typed map**: the + client-direction registry-servable set is closed (exactly the two + `-R` data-leg types under `tunnel`); one acceptor trait per type + (`TcpipForwardedAcceptor` / `StreamlocalForwardedAcceptor`), the + payload an accepted open hands the assembly: the channel duplex + (family-typed — the `-L` rows' `TunnelChannel`, one type both legs + ride) + the open's echo/originator fields — the harness + `ForwardedAcceptor` vocabulary grown, not forked. Install is + **refuse-never-replace** (`AcceptorHeld` — the server-side + `ListenRouter::register` posture mirrored); uninstall is explicit + and reports honestly. A new module `src/client_accept.rs` (one + module per file; the rows' method bodies live beside the registry). +- **The listen rows gate on the registered acceptor** (the + issuing-side disposition — a listen whose data leg cannot serve is + never asked for; install-first recorded order; + `OpError::NoAcceptor` pre-wire). The cancel forms do *not* gate on + the acceptor (teardown must work after an uninstall). +- **Bound-port mapping** (recorded wire fact at this pin): russh's + client `tcpip_forward` reply carries a port **only for + `port = 0` requests** (sshd parity — an explicit request's bound + port *is* the requested port); `reverse_listen` maps both to the + actually-bound number it returns. +- **Bounded bookkeeping — the cap relation**: two dedicated + `ChannelCap` counters, each against the same + `Limits::max_concurrent_channels` ceiling (the server-side + listener/ListenerCap posture mirrored): (a) the per-connection + **reverse-listen count** — admitted at the listen rows, released at + confirmed cancels and the refused/wire-error paths (never a map — + the outstanding listens live at the peer); (b) the **accepted-open + handoff count** — admitted in the dial handler's open hook before + the accept, released when the acceptor's `accept` call completes in + the spawned handoff task; a full counter declines the arrival + (`cap-exceeded`, event vocabulary). No accepted-open ledger: the + pump is the acceptor's from the handoff on (the family handler + shape). +- **The runtime-Unix gate rides the unix *listen* rows only** (the + bind is the platform-bound op; `forwarded-streamlocal` *ingestion* + is platform-neutral byte pumping, hook universal compile). +- **Pre-existing wart fixed**: `client_ops.rs`'s test mod gated + `family_exit_code` under a `tunnel`-inclusive cfg — the + `--features tunnel` build did not compile its tests before this + task; the import/test are now gated to the exit-consuming builds. + +Surprises: russh's client reply shapes (above, the port-0-only mint) +and the `JoinHandle::drop`-is-detach-not-cancel behavior in the +double's cancel teardown (first draft removed the entry without +aborting the accept loop — the real listener stayed bound; the double +aborts explicitly now). + ## Summary -> To be filled on completion \ No newline at end of file +> To be filled on completion + +Landed (`feat(client)`: w4-reverse-listen): + +- **`src/client_accept.rs`** — the acceptor registry + (`AcceptorRegistry` + the two per-type acceptor traits + the + `ForwardedTcpip`/`ForwardedStreamlocal` payloads + `AcceptorHeld`), + the reverse-listen rows (`reverse_listen` / `reverse_listen_cancel` + / the unix pair, with the gates + dedicated counters), and the + accepted-opens' handoff tasks; exports from the crate root under + `tunnel`. +- **`src/client.rs`** — the dial wires the (empty) registry + + tunnel surface into the handler and the connection + (`SshConnection::acceptors` — the live-connection accessor; + the listener budget's private field). +- **`src/client_handler.rs`** — the two forwarded hooks route through + the registry at arrival time (registered + room → accept + spawned + handoff; unregistered/cap-exhausted → clean `Administratively- + Prohibited` decline with the event vocabulary); every other type + keeps the decline-all skeleton; the never-feature rows unchanged. +- **`src/error.rs`** — `OpError::NoAcceptor` / `ListenRefused` / + `ListenCapExceeded` + the `RequestDenied → ListenRefused` mapping + (the listen/cancel forms' only wire decline shape at this pin). +- **`tests/common/sshd_double.rs`** — the fake `sshd`'s `-R` server + half: the four listen/cancel hooks (scripted accept/deny via + `DblListen`; an accept binds a **real** socket, mints the real + bound port for `port = 0` into the out-param, spawns the accept + loop — per accepted connection a server-initiated `forwarded-*` + open echoing the listen's args + the peer, then the bytes pumped; + cancels match by the raw wire strings, tearing the real listener + down) + the listen/cancel observations. +- **`tests/client_reverse_listen.rs`** (`tunnel` build) — the + contract rows: listen + data-leg pump round trip (real + `ssh -R`-shaped), the port-0 mint, the no-acceptor pre-wire + declines (listen + open, with the event rows), the late-install + rows (pre-install declined → serve-after; uninstall takes it off), + the cancel rows (teardown observable — reconnect refused; unknown + → `ListenRefused`), the listener-ceiling row (exceeded pre-wire; + cancel recovers), the registry contract through the accessor, and + the streamlocal unix rows (+ the compiled off-Unix gate row). + Per-build matrix: the *unfeatured* builds' wire rows run in every + build through `tests/client_dial.rs`'s decline-all row (the + surface itself is the compile-time mirror). +- **`tests/tunnel_mapping.rs`** — the `client_smoke` module: the + full-circle alkssh-client ↔ alkssh-server-direction smoke (both + `-R` legs through the alkcall fabric double over a real TCP + mount) — the port-0 mint through the server direction's fabric + listen + arrival pump into the registered acceptor's duplex, bytes + round-tripped end to end; real-`sshd` `-R` remains wave-5 e2e + (recorded, never silently dropped). + +Verification: `cargo test` (default) / `--features ` / +`--all-features` all green; `cargo clippy --all-targets -- -D +warnings` × builds (none / all-features) clean; `cargo fmt --check` +clean; `cargo doc --no-deps` (+ `--all-features`; only the +pre-existing `client_ops.rs` private-link warnings remain). \ No newline at end of file diff --git a/tests/client_reverse_listen.rs b/tests/client_reverse_listen.rs new file mode 100644 index 0000000..ba93459 --- /dev/null +++ b/tests/client_reverse_listen.rs @@ -0,0 +1,795 @@ +//! The `-R` leg's contract rows (`w4-reverse-listen`) — driven through +//! the in-process fake `sshd` grown with the reverse-forward surfaces +//! ([`sshd_double`](common::sshd_double): the `tcpip-forward` / +//! `streamlocal-forward` hooks that bind **real** sockets and open the +//! `forwarded-*` channels toward the client per accepted connection): +//! the listen rows (featured + registered → served, the port-0 mint, +//! the cancels + the listener-cap ceiling), the acceptor-registry rows +//! (featured + unregistered → clean decline; late-install across a +//! live connection — decline-before-install, serve-after; uninstall; +//! the data-leg pump round trip), the off-Unix streamlocal rows +//! (OQ-11), and the connection-stays-usable row after every decline +//! class. The row/stamp discipline (`tests/common/mod.rs`'s doc) +//! applies verbatim. +//! +//! **Per-build matrix note (recorded):** this file is the +//! `tunnel`-featured build's rows (the only build whose surface can +//! issue the reverse rows at all — the unfeatured build's compile-time +//! mirror); the wire decline rows for the *unfeatured* builds run in +//! every build through `tests/client_dial.rs`'s +//! `server_initiated_opens_decline_cleanly_on_the_wire` (the +//! decline-all skeleton the dial carries in a build without `tunnel` +//! — no registry exists there, every hook declines). +#![cfg(feature = "tunnel")] + +mod common; + +use std::sync::Arc; +use std::time::Duration; + +use alkssh::{ + AcceptorHeld, AcceptorRegistry, DialConfig, ForwardedStreamlocal, ForwardedTcpip, OpError, + SshConnection, StreamlocalForwardedAcceptor, TcpipForwardedAcceptor, dial, +}; +use async_trait::async_trait; +use common::sshd_double::{DblListen, SshdDouble, SshdObserved, SshdScript, spawn_scripted}; +use common::{MemHostKeyPolicy, TestVault, fixture}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +/// A vault-backed key source over the alpha fixture — the dial +/// posture (the fixture PEM sealed, decrypt-at-dial). +fn alpha_key_source() -> Arc { + let vault = TestVault::unlocked(); + let blob = vault.seal(fixture("alpha")); + Arc::new(alkssh::VaultKeySource::new(vault.handle(), Some(blob))) +} + +/// Dials a fresh double under the given script — one dial per row. +async fn dial_scripted_limits( + script: SshdScript, + limits: alkssh::Limits, +) -> (SshConnection, SshdDouble, SshdObserved) { + let (double, accepts, observed) = + spawn_scripted(vec![fixture("bravo").private_key()], true, script).await; + accepts.register(fixture("alpha").fingerprint); + + let policy = Arc::new(MemHostKeyPolicy::with_default(true)); + let mut config = DialConfig::new( + double.address.clone(), + "runner", + alpha_key_source(), + Arc::clone(&policy) as Arc, + ); + config.limits = limits; + let connection = dial(config).await.expect("the dial succeeds"); + (connection, double, observed) +} + +async fn dial_scripted(script: SshdScript) -> (SshConnection, SshdDouble, SshdObserved) { + dial_scripted_limits(script, alkssh::Limits::default()).await +} + +/// The clean-close observable: the explicit close leaves the double's +/// session `Ok` — the "the connection stays usable" final act. +async fn expect_clean_close(connection: SshConnection, mut double: SshdDouble) { + assert!(!connection.is_closed()); + connection.close().await; + let outcome = double.next_session_outcome().await; + assert!( + matches!(outcome, Some(Ok(()))), + "the clean close is the double's clean session end, got {outcome:?}" + ); + double.shutdown(); +} + +/// A free loopback port the row's listen asks for exactly (the probe +/// binds, then frees — a tiny race, as everywhere a fixed port is +/// picked; the rows mostly mint dynamic ports instead). +async fn pick_port() -> u16 { + let probe = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("the probe binds"); + probe.local_addr().expect("addr").port() +} + +/// The byte round trip the acceptor double's echo pump runs — the +/// data leg's plain duplex (the bytes back through the accepted +/// channel; the pump is the assembly's). +async fn echo_pump(mut stream: alkssh::TunnelChannel) { + let mut buf = vec![0u8; 4096]; + loop { + match stream.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(count) => { + if stream.write_all(&buf[..count]).await.is_err() { + return; + } + } + } + } +} + +/// What the acceptor double records per accepted open — the payload's +/// echo/originator fields (the duplex rides the type, never recorded). +#[derive(Clone, Debug, PartialEq)] +struct OpenRecord { + connected_address: String, + connected_port: u32, + originator_address: String, + originator_port: u32, +} + +/// The acceptor double — records the payload's fields and, when the +/// row asks, pumps the duplex as an echo (the family handler shape: +/// the assembly owns the pump). +#[derive(Clone)] +struct AcceptorDouble { + opens: tokio::sync::mpsc::UnboundedSender, + echo: bool, +} + +#[async_trait] +impl TcpipForwardedAcceptor for AcceptorDouble { + async fn accept(&self, open: ForwardedTcpip) { + let record = OpenRecord { + connected_address: open.connected_address.clone(), + connected_port: open.connected_port, + originator_address: open.originator_address.clone(), + originator_port: open.originator_port, + }; + if self.echo { + tokio::spawn(echo_pump(open.stream)); + } else { + drop(open.stream); + } + let _ = self.opens.send(record); + } +} + +/// The streamlocal acceptor double — the unix slot's shape. +#[derive(Clone)] +struct UnixAcceptorDouble { + opens: tokio::sync::mpsc::UnboundedSender, + echo: bool, +} + +#[async_trait] +impl StreamlocalForwardedAcceptor for UnixAcceptorDouble { + async fn accept(&self, open: ForwardedStreamlocal) { + if self.echo { + tokio::spawn(echo_pump(open.stream)); + } else { + drop(open.stream); + } + let _ = self.opens.send(open.socket_path); + } +} + +/// The next accepted open's fields — the row's payload observation +/// (bounded await: a wrong row must not hang). +async fn next_open( + opens: &mut tokio::sync::mpsc::UnboundedReceiver, + context: &str, +) -> OpenRecord { + tokio::time::timeout(Duration::from_secs(15), opens.recv()) + .await + .unwrap_or_else(|_| panic!("{context}: the accepted open hands off in time")) + .expect("{context}: the acceptor sender lives") +} + +/// - *Contract stamp:* `client.md` §op rows row 4 (the `-R` leg's +/// listen + ingest row) + the module docs' listen-gate record — +/// featured + registered → served: the +/// `reverse_listen` request rides the wire (the double's +/// `tcpip-forward` hook observes it verbatim), the actually-bound +/// port returns, and the **data leg** round-trips: a connection to +/// the bound port's bytes flow through the double's accept loop → +/// the server-initiated `forwarded-tcpip` open → the registered +/// acceptor's duplex → the echo pump → back, verbatim. The payload +/// fields are the wire's echo shapes (the connected address/port == +/// the listen request's, the originator the arrival's peer). +#[tokio::test] +async fn reverse_listen_serves_and_the_data_leg_pumps() { + let scripted = SshdScript { + tcp_listen: DblListen::Accept, + ..SshdScript::default() + }; + let (connection, double, observed) = dial_scripted(scripted).await; + + let (opens_tx, mut opens_rx) = tokio::sync::mpsc::unbounded_channel(); + let registry: &AcceptorRegistry = connection.acceptors(); + registry + .install_tcpip(Arc::new(AcceptorDouble { + opens: opens_tx, + echo: true, + })) + .expect("the slot installs"); + + let port = u32::from(pick_port().await); + let bound = connection + .reverse_listen("127.0.0.1", port) + .await + .expect("the featured + registered listen serves"); + assert_eq!(bound, port, "an explicit port binds exactly that port"); + + let mut conn = + tokio::net::TcpStream::connect(("127.0.0.1", u16::try_from(port).expect("in-range"))) + .await + .expect("the double's real listener accepts"); + conn.write_all(b"leg-data") + .await + .expect("the forward's ingress carries the bytes"); + let mut echoed = [0u8; 8]; + tokio::time::timeout(Duration::from_secs(15), conn.read_exact(&mut echoed)) + .await + .expect("the round trip completes in time") + .expect("the echo reads"); + assert_eq!(&echoed, b"leg-data", "the data leg round-trips verbatim"); + + let record = next_open(&mut opens_rx, "the data-leg row").await; + assert_eq!(record.connected_address, "127.0.0.1".to_string()); + assert_eq!(record.connected_port, port); + assert_eq!(record.originator_address, "127.0.0.1".to_string()); + assert!( + record.originator_port > 0, + "the originator is the arrival's ephemeral port" + ); + + assert_eq!( + observed.listens(), + vec![("tcpip-forward", "127.0.0.1".to_string(), port)], + "the listen request rode the wire verbatim" + ); + + expect_clean_close(connection, double).await; +} + +/// - *Contract stamp:* `client.md` §Lifecycle primitives (the +/// bound-port spec) — `port = 0` mint: the peer picks and the +/// surfaced number is the actually-bound one (the row connects to +/// it; the double's listener is the real socket, so the mint is the +/// real shape sshd's `get_local_port` parity rides). +#[tokio::test] +async fn port_zero_mint_surfaces_the_actually_bound_port() { + let scripted = SshdScript { + tcp_listen: DblListen::Accept, + ..SshdScript::default() + }; + let (connection, _double, _observed) = dial_scripted(scripted).await; + + let (opens_tx, mut opens_rx) = tokio::sync::mpsc::unbounded_channel(); + connection + .acceptors() + .install_tcpip(Arc::new(AcceptorDouble { + opens: opens_tx, + echo: false, + })) + .expect("the slot installs"); + + let bound = connection + .reverse_listen("127.0.0.1", 0) + .await + .expect("the mint serves"); + assert!( + bound > 0, + "the served mint is the actually-bound port, never the requested 0" + ); + + drop( + tokio::net::TcpStream::connect(("127.0.0.1", u16::try_from(bound).expect("in-range"))) + .await + .expect("the surfaced number is a really listening port"), + ); + + let record = next_open(&mut opens_rx, "the mint row").await; + assert_eq!(record.connected_address, "127.0.0.1"); + assert_eq!( + record.connected_port, bound, + "the open echoes the bound port" + ); + + let _ = connection.close().await; +} + +/// - *Contract stamp:* ADR-003 §4's issuing-side mirror + the module +/// docs' listen-gate record — a reverse listen with no registered +/// acceptor declines **before any wire op** (`NoAcceptor`; the +/// half-capability posture: a bound-but-unserveable listener is +/// never asked for), and the connection stays usable. +#[tokio::test] +async fn reverse_listen_without_an_acceptor_declines_before_any_wire_op() { + let scripted = SshdScript { + tcp_listen: DblListen::Accept, + ..SshdScript::default() + }; + let (connection, double, observed) = dial_scripted(scripted).await; + + let result = connection + .reverse_listen("127.0.0.1", u32::from(pick_port().await)) + .await; + assert!( + matches!(result, Err(OpError::NoAcceptor)), + "the unregistered class answers the listen, got {result:?}" + ); + assert!( + observed.listens().is_empty(), + "nothing rode the wire — the gate answered before the request" + ); + + expect_clean_close(connection, double).await; +} + +/// - *Contract stamp:* ADR-003 §4 — featured + unregistered → clean +/// decline: a server-initiated `forwarded-tcpip` open on a +/// live connection with an empty registry answers +/// `AdministrativelyProhibited` (never russh's accept-and-drop +/// default, never a buffered open), the decline event rides the +/// crate vocabulary, and the connection stays usable. +#[tokio::test] +async fn unregistered_forwarded_open_declines_cleanly() { + common::event_capture::install(); + let scripted = SshdScript { + tcp_listen: DblListen::Accept, + ..SshdScript::default() + }; + let (connection, mut double, _observed) = dial_scripted(scripted).await; + + let server = double.session_handle().await; + let outcome = tokio::time::timeout( + Duration::from_secs(15), + server.channel_open_forwarded_tcpip("hub.example", 4096, "originator", 0), + ) + .await + .expect("the open resolves in time"); + assert!( + matches!( + outcome, + Err(russh::Error::ChannelOpenFailure( + russh::ChannelOpenFailure::AdministrativelyProhibited + )) + ), + "the unregistered open declines cleanly, got {outcome:?}" + ); + + let decline = common::event_capture::await_event(&[ + ("row", "forwarded-tcpip"), + ("decline_class", "open"), + ("reason", "no-acceptor"), + ]) + .await; + assert_eq!(decline.level, "DEBUG"); + assert_eq!( + decline.field("feature_build"), + Some(common::event_capture::expected_build()) + ); + + expect_clean_close(connection, double).await; +} + +/// - *Contract stamp:* the late-install dynamic (reverse-runner fact +/// 2 — consumers turn capabilities on/off per connection) + the +/// arrival-time gate — the same live connection: the pre-install +/// arrival declines, the post-install arrival serves (never +/// buffered, never half-served), and the acceptor double's late +/// install is the registry's whole surface. +#[tokio::test] +async fn late_install_serves_after_a_pre_install_arrival_declined() { + let scripted = SshdScript { + tcp_listen: DblListen::Accept, + ..SshdScript::default() + }; + let (connection, mut double, _observed) = dial_scripted(scripted).await; + + let server = double.session_handle().await; + let pre = tokio::time::timeout( + Duration::from_secs(15), + server.channel_open_forwarded_tcpip("hub.example", 4096, "originator", 0), + ) + .await + .expect("the pre-install open resolves in time"); + assert!( + matches!( + pre, + Err(russh::Error::ChannelOpenFailure( + russh::ChannelOpenFailure::AdministrativelyProhibited + )) + ), + "the pre-install arrival declines at arrival time, got {pre:?}" + ); + + let (opens_tx, mut opens_rx) = tokio::sync::mpsc::unbounded_channel(); + connection + .acceptors() + .install_tcpip(Arc::new(AcceptorDouble { + opens: opens_tx, + echo: false, + })) + .expect("the late install takes"); + + let post = tokio::time::timeout( + Duration::from_secs(15), + server.channel_open_forwarded_tcpip("hub.example", 4096, "originator", 0), + ) + .await + .expect("the post-install open resolves in time"); + assert!( + post.is_ok(), + "the post-install arrival serves, got {post:?}" + ); + let record = next_open(&mut opens_rx, "the late-install row").await; + assert_eq!( + record, + OpenRecord { + connected_address: "hub.example".into(), + connected_port: 4096, + originator_address: "originator".into(), + originator_port: 0, + } + ); + + expect_clean_close(connection, double).await; +} + +/// - *Contract stamp:* the late-install toggle's off half — the +/// uninstall takes the capability off mid-connection: the +/// post-uninstall arrival declines at arrival time (fail-closed; +/// the already-accepted open's channel is the acceptor's business). +#[tokio::test] +async fn uninstall_takes_the_capability_off_before_a_later_arrival() { + let scripted = SshdScript { + tcp_listen: DblListen::Accept, + ..SshdScript::default() + }; + let (connection, mut double, _observed) = dial_scripted(scripted).await; + + let (opens_tx, mut opens_rx) = tokio::sync::mpsc::unbounded_channel(); + connection + .acceptors() + .install_tcpip(Arc::new(AcceptorDouble { + opens: opens_tx, + echo: false, + })) + .expect("the slot installs"); + + let server = double.session_handle().await; + let served = tokio::time::timeout( + Duration::from_secs(15), + server.channel_open_forwarded_tcpip("hub.example", 4096, "originator", 0), + ) + .await + .expect("the served open resolves in time"); + assert!(served.is_ok(), "the installed slot serves, got {served:?}"); + let record = next_open(&mut opens_rx, "the uninstall row").await; + assert_eq!(record.connected_address, "hub.example"); + + assert!( + connection.acceptors().uninstall_tcpip(), + "the uninstall reports the held slot" + ); + + let post = tokio::time::timeout( + Duration::from_secs(15), + server.channel_open_forwarded_tcpip("hub.example", 4096, "originator", 0), + ) + .await + .expect("the post-uninstall open resolves in time"); + assert!( + matches!( + post, + Err(russh::Error::ChannelOpenFailure( + russh::ChannelOpenFailure::AdministrativelyProhibited + )) + ), + "the post-uninstall arrival declines, got {post:?}" + ); + + expect_clean_close(connection, double).await; +} + +/// - *Contract stamp:* `client.md` §Lifecycle primitives (every +/// listener explicitly cancellable) + the double's sshd parity — +/// the cancel tears the real listener down (the bound port +/// reconnects as refused), a cancel of an unknown listener is the +/// peer's refusal (`ListenRefused`), and the connection stays +/// usable after every cancel class. +#[tokio::test] +async fn cancel_tears_the_forward_down_and_unknown_cancels_refuse() { + let scripted = SshdScript { + tcp_listen: DblListen::Accept, + ..SshdScript::default() + }; + let (connection, double, observed) = dial_scripted(scripted).await; + + let (opens_tx, _opens_rx) = tokio::sync::mpsc::unbounded_channel(); + connection + .acceptors() + .install_tcpip(Arc::new(AcceptorDouble { + opens: opens_tx, + echo: false, + })) + .expect("the slot installs"); + + let port = u32::from(pick_port().await); + connection + .reverse_listen("127.0.0.1", port) + .await + .expect("the listen binds"); + + connection + .reverse_listen_cancel("127.0.0.1", port) + .await + .expect("the confirmed cancel"); + assert_eq!( + observed.cancels(), + vec![("cancel-tcpip-forward", "127.0.0.1".to_string(), port)], + "the cancel rode the wire verbatim" + ); + + let mut refused = false; + for _ in 0..200 { + let reconnect = + tokio::net::TcpStream::connect(("127.0.0.1", u16::try_from(port).expect("in-range"))) + .await; + match reconnect { + Ok(_) => { + tokio::time::sleep(Duration::from_millis(25)).await; + } + Err(err) if err.kind() == std::io::ErrorKind::ConnectionRefused => { + refused = true; + break; + } + Err(err) => panic!("the bound port errors, not refuses, after the cancel: {err:?}"), + } + } + assert!( + refused, + "the forward is torn down — the bound port refuses a reconnect" + ); + + let again = connection.reverse_listen_cancel("127.0.0.1", port).await; + assert!( + matches!(again, Err(OpError::ListenRefused)), + "the unknown listener's cancel refuses, got {again:?}" + ); + assert!( + matches!( + connection + .reverse_listen_cancel("never-listened", port) + .await, + Err(OpError::ListenRefused) + ), + "no match by endpoint either — the raw-string parity holds" + ); + + expect_clean_close(connection, double).await; +} + +/// - *Contract stamp:* the module docs' recorded cap relation (the +/// server-side listener-counter posture mirrored) — the dedicated +/// listener count against the `max_concurrent_channels` ceiling: +/// beyond it the listen declines **before any wire op** +/// (`ListenCapExceeded`), and a confirmed cancel recovers the +/// budget. +#[tokio::test] +async fn the_listener_ceiling_bounds_the_listen_map_and_recovers_on_cancel() { + let scripted = SshdScript { + tcp_listen: DblListen::Accept, + ..SshdScript::default() + }; + let limits = alkssh::Limits { + max_concurrent_channels: 1, + ..alkssh::Limits::default() + }; + let (connection, double, observed) = dial_scripted_limits(scripted, limits).await; + + let (opens_tx, _opens_rx) = tokio::sync::mpsc::unbounded_channel(); + connection + .acceptors() + .install_tcpip(Arc::new(AcceptorDouble { + opens: opens_tx, + echo: false, + })) + .expect("the slot installs"); + + let first = u32::from(pick_port().await); + connection + .reverse_listen("127.0.0.1", first) + .await + .expect("the first listen admits"); + + let second = connection + .reverse_listen("127.0.0.1", u32::from(pick_port().await)) + .await; + assert!( + matches!(second, Err(OpError::ListenCapExceeded)), + "beyond the ceiling the listen refuses pre-wire, got {second:?}" + ); + assert_eq!( + observed.listens().len(), + 1, + "the ceiling refusal consumed nothing on the wire" + ); + + connection + .reverse_listen_cancel("127.0.0.1", first) + .await + .expect("the cancel recovers the budget"); + + let third = u32::from(pick_port().await); + connection + .reverse_listen("127.0.0.1", third) + .await + .expect("the count admits again after the cancel"); + assert_eq!( + observed.listens().len(), + 2, + "the recovered budget's listen rode the wire" + ); + + expect_clean_close(connection, double).await; +} + +/// - *Contract stamp:* the module docs' refuse-never-replace posture +/// through the connection accessor — the second install is declined +/// (`AcceptorHeld`), the first acceptor stays served, and the +/// uninstall/uninstall-again cycle reports honestly. +#[tokio::test] +async fn the_connection_accessor_serves_the_registry_contract() { + let scripted = SshdScript { + tcp_listen: DblListen::Accept, + ..SshdScript::default() + }; + let (connection, _double, _observed) = dial_scripted(scripted).await; + + let (opens_tx, _opens_rx) = tokio::sync::mpsc::unbounded_channel(); + let registry = connection.acceptors(); + registry + .install_tcpip(Arc::new(AcceptorDouble { + opens: opens_tx, + echo: false, + })) + .expect("the first install takes"); + assert!( + matches!( + registry.install_tcpip(Arc::new(AcceptorDouble { + opens: tokio::sync::mpsc::unbounded_channel().0, + echo: false, + })), + Err(AcceptorHeld) + ), + "the held slot refuses the second install" + ); + assert!(registry.uninstall_tcpip()); + assert!(!registry.uninstall_tcpip(), "the honest `false` report"); + + let _ = connection.close().await; +} + +/// The unix `-R` rows (the runtime-gate's on-Unix leg's shape): the +/// listen + the data-leg round trip + the payload's echoed path + the +/// cancel's wire-honest teardown + the unknown cancel's refusal. +/// +/// - *Contract stamp:* OQ-11's client mirror (the +/// `streamlocal_forward` rows compile under `tunnel`, serve on +/// unix) + `mapping.md` §Global-request rows (the unix listen legs) +/// + the acceptor registry's streamlocal slot (featured + +/// registered → served). +#[cfg(unix)] +#[tokio::test] +async fn streamlocal_listen_serves_arrivals_and_cancels_through_the_unix_gate() { + let scripted = SshdScript { + unix_listen: DblListen::Accept, + ..SshdScript::default() + }; + let (connection, double, observed) = dial_scripted(scripted).await; + + let (ops_tx, mut ops_rx) = tokio::sync::mpsc::unbounded_channel(); + let registry = connection.acceptors(); + registry + .install_streamlocal(Arc::new(UnixAcceptorDouble { + opens: ops_tx, + echo: true, + })) + .expect("the streamlocal slot installs"); + + let socket_path = std::env::temp_dir() + .join(format!("alkssh-w4-rl-{}", std::process::id())) + .to_path_buf(); + let socket_path = socket_path.to_string_lossy().to_string(); + let _ = std::fs::remove_file(&socket_path); + + connection + .reverse_listen_unix(&socket_path) + .await + .expect("the unix listen serves"); + assert_eq!( + observed.listens(), + vec![("streamlocal-forward", socket_path.clone(), 0)], + "the unix listen rode the wire verbatim" + ); + + let mut conn = tokio::net::UnixStream::connect(&socket_path) + .await + .expect("the double's real listener accepts"); + conn.write_all(b"unix-data") + .await + .expect("the unix forward's ingress carries the bytes"); + let mut echoed = [0u8; 9]; + tokio::time::timeout(Duration::from_secs(15), conn.read_exact(&mut echoed)) + .await + .expect("the unix round trip completes in time") + .expect("the echo reads"); + assert_eq!( + &echoed, b"unix-data", + "the unix data leg round-trips verbatim" + ); + + let path = tokio::time::timeout(Duration::from_secs(15), ops_rx.recv()) + .await + .expect("the streamlocal handoff arrives in time") + .expect("the acceptor sender lives"); + assert_eq!( + path, socket_path, + "the open echoes the listen's socket path verbatim" + ); + + connection + .reverse_listen_unix_cancel(&socket_path) + .await + .expect("the unix cancel confirms"); + assert_eq!( + observed.cancels(), + vec![("cancel-streamlocal-forward", socket_path.clone(), 0)], + "the unix cancel rode the wire verbatim" + ); + + let reconnect = tokio::net::UnixStream::connect(&socket_path).await; + assert!( + reconnect.is_err(), + "the unix forward is torn down — the socket path refuses a reconnect, got {reconnect:?}" + ); + assert!( + matches!( + connection.reverse_listen_unix_cancel(&socket_path).await, + Err(OpError::ListenRefused) + ), + "the unknown unix listener's cancel refuses" + ); + + let _ = std::fs::remove_file(&socket_path); + expect_clean_close(connection, double).await; +} + +/// - *Contract stamp:* OQ-11 (the runtime-Unix gate; universal +/// compile, clean off-Unix decline) + the module docs' gate record — +/// off-Unix the unix listen declines **before any wire op** +/// (`UnixUnavailable`; the platform-refused request consumes +/// nothing). +#[cfg(not(unix))] +#[tokio::test] +async fn streamlocal_declines_off_unix_before_any_wire_op() { + let scripted = SshdScript { + unix_listen: DblListen::Accept, + ..SshdScript::default() + }; + let (connection, double, observed) = dial_scripted(scripted).await; + + connection + .acceptors() + .install_streamlocal(Arc::new(UnixAcceptorDouble { + opens: tokio::sync::mpsc::unbounded_channel().0, + echo: false, + })) + .expect("the slot installs"); + let result = connection.reverse_listen_unix("alkssh-row.sock").await; + assert!( + matches!(result, Err(OpError::UnixUnavailable)), + "the off-Unix gate answers the unix listen, got {result:?}" + ); + assert!( + observed.listens().is_empty(), + "nothing rode the wire — the gate answered before the request" + ); + + expect_clean_close(connection, double).await; +} diff --git a/tests/common/sshd_double.rs b/tests/common/sshd_double.rs index 1c28efc..df8d07b 100644 --- a/tests/common/sshd_double.rs +++ b/tests/common/sshd_double.rs @@ -56,6 +56,24 @@ //! changes/signals the client rows assert the wire verbatim //! against). The double stays vanilla-parity: a hand-rolled //! responder standing in for `sshd`, nothing alkssh-typed inside. +//! +//! - **The `-R` server half** (`w4-reverse-listen`): the double grows +//! the reverse-forward surfaces a vanilla `sshd` really serves — the +//! `tcpip-forward` / `streamlocal-forward` global-request hooks +//! (scripted accept/deny; an accept **binds a real socket** in +//! sshd's role, so a `port = 0` request's minted bound port is the +//! socket's real one and the data leg is a genuine +//! `ssh -R`-shaped round trip), the accept loop per bound listener +//! (per accepted connection: a server-initiated `forwarded-*` open +//! toward the client echoing the listen's own args + the +//! connection's peer as the originator, then the bytes pumped +//! bidirectionally; a client-declined open closes the accepted +//! connection — vanilla parity), and the cancel forms removing the +//! matching entry (tearing the real listener down; an unknown +//! listener denies, sshd parity). The bookkeeping is per-connection +//! handler state and bounded by the scripted listens alone — the +//! double is a scripted stand-in, not a real server: at most one +//! forward per scripted accept is what the rows drive. #![allow(dead_code)] @@ -89,6 +107,27 @@ pub enum DblDirect { Echo, } +/// The double's scripted behavior for the client's reverse-forward +/// listen requests (`tcpip-forward` / `streamlocal-forward@…) — the +/// `-R` rows' scripted class (`w4-reverse-listen`). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum DblListen { + /// Deny the listen — the vanilla sshd's refusal (the scripted + /// server-refusal class; the wire `RequestFailure` the client's + /// op row mirrors). The script's default (fail-closed double). + #[default] + Deny, + /// Accept the listen: the double binds a **real socket** in + /// sshd's role (the actually-bound port rides the reply's mint — + /// the port-0 row's real shape), records it for the cancel forms, + /// and spawns the accept loop (per accepted connection: a + /// server-initiated `forwarded-*` open toward the client echoing + /// the listen's own args + the connection's peer, then the bytes + /// pumped bidirectionally; a client-declined open closes that + /// accepted connection — vanilla parity). + Accept, +} + /// The fake `sshd`'s op-row script (the `w4-op-rows` request /// surfaces) — what the double answers and, on a success, what the /// scripted "program" produces. @@ -110,6 +149,11 @@ pub struct SshdScript { pub direct: DblDirect, /// The `direct-streamlocal` opens' behavior. pub streamlocal: DblDirect, + /// The `tcpip-forward` listen form's behavior (`tcp -R` rows). + pub tcp_listen: DblListen, + /// The `streamlocal-forward` listen form's behavior (the unix + /// `-R` rows). + pub unix_listen: DblListen, /// The counted refusal — the double declines the *first N* client /// session opens (per connection, armed by /// `accept_session_channels`): the stays-usable-after-a-decline @@ -121,7 +165,8 @@ pub struct SshdScript { impl Default for SshdScript { /// The plain dial rows' posture: session opens accepted (the /// `accept_session_channels` flag governs), requests succeed with - /// no outputs and no exit, direct opens rejected. + /// no outputs and no exit, direct opens rejected, listen forms + /// denied (the scripted fail-closed default). fn default() -> Self { Self { request_success: true, @@ -130,11 +175,18 @@ impl Default for SshdScript { exit: None, direct: DblDirect::Reject, streamlocal: DblDirect::Reject, + tcp_listen: DblListen::Deny, + unix_listen: DblListen::Deny, reject_first_session_opens: 0, } } } +/// One recorded global-request listen/cancel form: (the request +/// name, the wire endpoint string, the wire port) — the reverse-listen +/// rows assert the listens the client ops put on the wire. +pub type ListenRecord = (&'static str, String, u32); + /// The double's per-connection observations — the wire rows assert /// the verbatim translations against these (what the client's op /// rows put on the wire). @@ -145,6 +197,8 @@ pub struct SshdObserved { shells: Arc>>, windows: Arc>>, signals: Arc>>, + listens: Arc>>, + cancels: Arc>>, } /// One recorded pty request's translated shapes — (term, cols, rows, @@ -185,6 +239,17 @@ impl SshdObserved { pub fn signals(&self) -> Vec { self.signals.lock().clone() } + + /// The global-request listen forms seen (name, endpoint, port — + /// the requested port, never the mint). + pub fn listens(&self) -> Vec { + self.listens.lock().clone() + } + + /// The global-request cancel forms seen (name, endpoint, port). + pub fn cancels(&self) -> Vec { + self.cancels.lock().clone() + } } /// The double's scripted program serve — after a request's success @@ -244,6 +309,81 @@ async fn direct_echo(mut stream: russh::ChannelStream) { } } +/// The `-R` accept loop per bound listener (the double's sshd role): +/// per accepted connection a server-initiated `forwarded-*` open +/// toward the client — the connected address echoes the listen +/// request's wire string verbatim, the port the actually-bound one, +/// the originator the accepted connection's peer — then the bytes +/// pumped bidirectionally between the SSH channel and the accepted +/// socket. A client-declined open closes that accepted connection +/// (vanilla parity: the listener stays, the data has nowhere to go). +/// The loop ends with the listener itself (cancel tears it down; the +/// session's end fails the opens' sends). +async fn accept_forward_loop( + handle: Handle, + listener: tokio::net::TcpListener, + connected_address: String, + connected_port: u32, +) { + loop { + let Ok((conn, peer)) = listener.accept().await else { + return; + }; + let opened = handle + .channel_open_forwarded_tcpip( + connected_address.clone(), + connected_port, + peer.ip().to_string(), + u32::from(peer.port()), + ) + .await; + match opened { + Ok(channel) => { + tokio::spawn(forward_connection(channel, conn)); + } + Err(_) => drop(conn), + } + } +} + +/// The accepted forward connection's pump (its SSH channel ↔ its +/// socket) — the data leg's byte plane. +async fn forward_connection(channel: Channel, mut conn: tokio::net::TcpStream) { + let mut stream = channel.into_stream(); + let _ = tokio::io::copy_bidirectional(&mut stream, &mut conn).await; +} + +/// The unix accept loop's twin (the unix `-R` rows' shape): per +/// accepted socket a `forwarded-streamlocal` open toward the client +/// echoing the listen path. +#[cfg(unix)] +async fn accept_unix_forward_loop( + handle: Handle, + listener: tokio::net::UnixListener, + socket_path: String, +) { + loop { + let Ok((conn, _)) = listener.accept().await else { + return; + }; + let opened = handle + .channel_open_forwarded_streamlocal(socket_path.clone()) + .await; + match opened { + Ok(channel) => { + tokio::spawn(unix_forward_connection(channel, conn)); + } + Err(_) => drop(conn), + } + } +} + +#[cfg(unix)] +async fn unix_forward_connection(channel: Channel, mut conn: tokio::net::UnixStream) { + let mut stream = channel.into_stream(); + let _ = tokio::io::copy_bidirectional(&mut stream, &mut conn).await; +} + /// The fake `sshd`'s per-connection handler — accept-set identity /// resolution, explicit rejects everywhere, the `auth_succeeded` /// handle carrier and the session-open acceptor stub. @@ -255,6 +395,13 @@ struct SshdDoubleHandler { script: Arc, observed: SshdObserved, remaining_declines: std::sync::Arc, + /// The `-R` bookkeeping (per-connection, sshd's listener table): + /// the key is the cancel-form's match key (the wire endpoint + the + /// bound port — a `port = 0` mint's cancel matches the bound one), + /// the value the live accept loop (cancel aborts it, dropping the + /// real listener with it). + tcp_forwards: std::collections::HashMap<(String, u32), tokio::task::JoinHandle<()>>, + unix_forwards: std::collections::HashMap>, } impl russh::server::Handler for SshdDoubleHandler { @@ -485,6 +632,123 @@ impl russh::server::Handler for SshdDoubleHandler { self.observed.signals.lock().push(format!("{signal:?}")); Ok(()) } + + /// The `tcpip-forward` listen form: recorded, scripted — an accept + /// binds a real `127.0.0.1` listener in sshd's role (a `port = 0` + /// request's mint is the socket's real bound port, the out-param + /// parity), keys the listener by (wire address, bound port), and + /// spawns the accept loop ([`accept_forward_loop`]). + async fn tcpip_forward( + &mut self, + address: &str, + port: &mut u32, + session: &mut Session, + ) -> Result { + self.observed + .listens + .lock() + .push(("tcpip-forward", address.to_string(), *port)); + if self.script.tcp_listen == DblListen::Deny { + return Ok(false); + } + let bind_target = format!("127.0.0.1:{}", *port); + let Ok(listener) = tokio::net::TcpListener::bind(&bind_target).await else { + return Ok(false); + }; + let bound = listener + .local_addr() + .expect("the bound listener has an address") + .port(); + let bound = u32::from(bound); + if *port == 0 { + *port = bound; + } + let task = tokio::spawn(accept_forward_loop( + session.handle(), + listener, + address.to_string(), + bound, + )); + self.tcp_forwards.insert((address.to_string(), bound), task); + Ok(true) + } + + /// The `cancel-tcpip-forward` form: matches by the raw wire + /// strings (the bound port for a mint); a hit removes the + /// listener (tearing the real socket down — the cancel row's + /// observable), an unknown listener denies (sshd parity). + async fn cancel_tcpip_forward( + &mut self, + address: &str, + port: u32, + _session: &mut Session, + ) -> Result { + self.observed + .cancels + .lock() + .push(("cancel-tcpip-forward", address.to_string(), port)); + match self.tcp_forwards.remove(&(address.to_string(), port)) { + Some(listener_task) => { + listener_task.abort(); + Ok(true) + } + None => Ok(false), + } + } + + /// The `streamlocal-forward` listen form (the unix `-R`): records, + /// scripts, binds the real unix socket (stale file unlinked + /// first), and spawns the unix accept loop. Off-Unix the hook is + /// not implemented and falls to the trait's deny default (the + /// vanilla Windows sshd parity) — the crate's own runtime-Unix + /// gate mirrors the refusal client-side before the wire. + #[cfg(unix)] + async fn streamlocal_forward( + &mut self, + socket_path: &str, + session: &mut Session, + ) -> Result { + self.observed + .listens + .lock() + .push(("streamlocal-forward", socket_path.to_string(), 0)); + if self.script.unix_listen == DblListen::Deny { + return Ok(false); + } + let _ = std::fs::remove_file(socket_path); + let Ok(listener) = tokio::net::UnixListener::bind(socket_path) else { + return Ok(false); + }; + let task = tokio::spawn(accept_unix_forward_loop( + session.handle(), + listener, + socket_path.to_string(), + )); + self.unix_forwards.insert(socket_path.to_string(), task); + Ok(true) + } + + /// The `cancel-streamlocal-forward` form (matches by the path + /// alone — the unix cancel's sshd parity). + #[cfg(unix)] + async fn cancel_streamlocal_forward( + &mut self, + socket_path: &str, + _session: &mut Session, + ) -> Result { + self.observed.cancels.lock().push(( + "cancel-streamlocal-forward", + socket_path.to_string(), + 0, + )); + match self.unix_forwards.remove(socket_path) { + Some(listener_task) => { + listener_task.abort(); + Ok(true) + } + None => Ok(false), + } + } } impl SshdDoubleHandler { @@ -589,6 +853,8 @@ pub async fn spawn_scripted( remaining_declines: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new( script_declines, )), + tcp_forwards: std::collections::HashMap::new(), + unix_forwards: std::collections::HashMap::new(), }; let outcomes = outcomes_tx.clone(); tokio::spawn(async move { diff --git a/tests/tunnel_mapping.rs b/tests/tunnel_mapping.rs index 3390537..de51053 100644 --- a/tests/tunnel_mapping.rs +++ b/tests/tunnel_mapping.rs @@ -1894,6 +1894,166 @@ mod serving { } } +#[cfg(feature = "tunnel")] +mod client_smoke { + //! The full-circle smoke (`w4-reverse-listen`'s recorded posture): + //! the **alkssh client's** `-R` rows against the **alkssh server + //! direction's** `-R` rows — both directions of the crate in one + //! live wire, the route-1 bridge's two legs meeting through the + //! alkcall fabric double (the composition the alkforge assembly + //! makes). A connection to the minted bound port round-trips + //! through: the fabric's real TCP listener → the fabric's + //! bind-first accept loop → the door-side `tunnel/forwarded` + //! arrival → the door's server-initiated `forwarded-tcpip` open + //! (echo args + originator) → the alkssh client's acceptor registry + //! → the acceptor duplex → the echo back. Real-`sshd` `-R` is + //! wave-5 e2e territory (never silently dropped). + + use std::sync::Arc; + use std::time::Duration; + + use async_trait::async_trait; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + use alkssh::{ + AcceptorRegistry, DialConfig, ForwardedTcpip, TcpipForwardedAcceptor, TunnelListenSeam, + dial, + }; + + use super::common::harness::serve_config as harness_serve_config; + use super::common::{MemHostKeyPolicy, MemIdentityProvider, fixture, identity}; + use super::serving::wire_fabric_with_listen; + + struct EchoAcceptor { + opens: tokio::sync::mpsc::UnboundedSender<(String, u32, String)>, + } + + #[async_trait] + impl TcpipForwardedAcceptor for EchoAcceptor { + async fn accept(&self, open: ForwardedTcpip) { + let _ = self.opens.send(( + open.connected_address.clone(), + open.connected_port, + open.originator_address.clone(), + )); + tokio::spawn(async move { + let mut stream = open.stream; + let mut buf = vec![0u8; 4096]; + loop { + match stream.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(count) => { + if stream.write_all(&buf[..count]).await.is_err() { + return; + } + } + } + } + }); + } + } + + #[tokio::test] + async fn smoke_client_reverse_listen_full_circle_through_the_server_direction() { + let double = wire_fabric_with_listen(true, true).await; + + let provider = Arc::new(MemIdentityProvider::new()); + provider.register( + fixture("alpha").fingerprint, + identity("reverse-runner", &["ssh"]), + ); + let mut serve = harness_serve_config(provider, &[fixture("bravo")], Default::default()); + serve.channel_client = Some(Arc::clone(&double.client)); + serve.tunnel_listen_seam = Some(TunnelListenSeam::new(Arc::clone(&double.registry))); + let adapter = alkssh::SshAdapter::new(serve) + .await + .expect("the -R wired adapter constructs"); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("the smoke's loopback listener binds"); + let addr = listener.local_addr().expect("the smoke has an address"); + tokio::spawn(async move { + loop { + let Ok((socket, _)) = listener.accept().await else { + break; + }; + if adapter.handle(socket).await.is_err() { + break; + } + } + }); + + let policy = Arc::new(MemHostKeyPolicy::new()); + policy.pin("127.0.0.1", fixture("bravo").fingerprint); + let config = DialConfig::new( + addr.to_string(), + "runner", + crate_test_key_source(), + Arc::clone(&policy) as Arc, + ); + let connection = dial(config).await.expect("the dial succeeds"); + + let (opens_tx, mut opens_rx) = tokio::sync::mpsc::unbounded_channel(); + let registry: &AcceptorRegistry = connection.acceptors(); + registry + .install_tcpip(Arc::new(EchoAcceptor { opens: opens_tx })) + .expect("the acceptor installs"); + + let bound = connection + .reverse_listen("127.0.0.1", 0) + .await + .expect("the full-circle mint serves"); + assert!( + bound > 0, + "the server direction's fabric minted a real bound port" + ); + + let mut conn = + tokio::net::TcpStream::connect(("127.0.0.1", u16::try_from(bound).expect("in-range"))) + .await + .expect("the fabric's listener accepts the arrival"); + conn.write_all(b"full-circle-data") + .await + .expect("the arrival carries the bytes"); + let mut echoed = [0u8; 16]; + tokio::time::timeout(Duration::from_secs(30), conn.read_exact(&mut echoed)) + .await + .expect("the full-circle round trip completes in time") + .expect("the echo reads"); + assert_eq!( + &echoed, b"full-circle-data", + "the bytes returned through every leg: fabric → door arrival \ + → SSH channel → client acceptor duplex → echo → back" + ); + + let (connected_address, connected_port, originator_address) = + tokio::time::timeout(Duration::from_secs(15), opens_rx.recv()) + .await + .expect("the acceptor's handoff record arrives") + .expect("the acceptor sender lives"); + assert_eq!(connected_address, "127.0.0.1"); + assert_eq!( + connected_port, bound, + "the echo port is the minted bound one" + ); + assert_eq!( + originator_address, "127.0.0.1", + "the originator carries the fabric arrival's peer" + ); + + let _ = connection.close().await; + let _ = double.minted_token().await; // the establisher's mint observed (the fabric's bookkeeping) + } + + /// The smoke's dial key pose (the TestVault posture, client-side). + fn crate_test_key_source() -> Arc { + let vault = super::common::TestVault::unlocked(); + let blob = vault.seal(fixture("alpha")); + Arc::new(alkssh::VaultKeySource::new(vault.handle(), Some(blob))) + } +} + #[cfg(all(feature = "tunnel", unix))] mod real_tool { //! The real-tool smoke (`w3-tunnel-direct`): one `ssh -L` (+ curl)