diff --git a/Cargo.toml b/Cargo.toml index c5cf66e..da64bbb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -91,6 +91,11 @@ gix-hash = { version = "0.28", features = ["sha1"] } # `ReceiverStream` — test-graph only (the crate proper hand-rolls no # streams). tokio-stream = "0.1" +# The leg-2 fabric double's establisher/params serialization +# (`w3-tunnel-listen`) — test-graph only (the door's own lib code +# stays serde-free under the `tunnel` feature; the double's +# dynamic-bind producer serializes params/reply fields). +serde_json = "1" # The real-tool pty smoke (`w3-tty-session`): alktty's `local` backend # composition — the dev-dep `features = ["local"]` unification applies # to the TEST graph only (feature unification folds `local` into the diff --git a/docs/architecture/mapping.md b/docs/architecture/mapping.md index 7d5012b..52dbcdc 100644 --- a/docs/architecture/mapping.md +++ b/docs/architecture/mapping.md @@ -1,6 +1,6 @@ --- status: reviewed -last_updated: 2026-10-28 +last_updated: 2026-10-29 --- # Mapping — server direction: SSH channels × family seams (the spine) @@ -34,7 +34,7 @@ bridge primitives every mapping uses (POC-1 §7). | `session` → `exec_request` git | `git clone/push ssh://` | `git` | alkgit: parse command → `Service` + repo → `serve_duplex_resolved(repo_id, identity, service, duplex, …)` | **exec-only** (ADR-006 decision 2): the repo rides inside the exec command string (`git-upload-pack '/repo'`); vanilla git never speaks subsystem for git; process-mode dialect (server-speaks-first advertisement); exit-status proven both directions | | `session` → `subsystem_request` | `sftp` clients; manual `ssh -s` | `sftp` (placeholder) / `git` rows | — | **`sftp` name**: declines until alkfs lands — the feature exists, the route doesn't (OQ-13 resolved: the tested decline placeholder is the shipped behavior until an alkfs-mapping ADR changes it). **git service names**: decline with reason — a subsystem request carries only a name, never a repo id (POC-3 §A.3); never half-serves | | `direct-tcpip` | `ssh -L`; all `ssh -D` traffic | `tunnel` | alktunnels **route 1**: adapter-side bridge — `TunnelSession::open_direct` on a held `ChannelClient`, pump `take_halves()` vs the channel | target = `SubstrateAddr::HostPort`/`Path`; `Substrate` = Tcp/Unix; accept only **after** the fabric open succeeds — failure maps to an explicit `ConnectFailed` reject (POC-3 §C.2); vanilla `-D`'s dynamic targets dispatch here (POC-3 correction 2) | -| `forwarded-tcpip` | `ssh -R` data channels | `tunnel` | the `register_tunnel_forwarded_openable`'s `ForwardedHandler` shape — hand the channel duplex to the handler | listen flow = the global-request rows below; the data leg is untested in POC-3 beyond refusal (see the POC-3 note below) | +| `forwarded-tcpip` | `ssh -R` data channels | `tunnel` | the `register_tunnel_forwarded_openable`'s `ForwardedHandler` shape — hand the channel duplex to the handler | per fabric arrival the door opens a `forwarded-tcpip` channel **toward the vanilla client** (a vanilla client never initiates this type — sshd parity; an inbound open is the unvanilla class, declined `ConnectFailed`); echo args = the original `tcpip_forward` request's wire string verbatim + the actually-bound port (the sshd `c->path` + `get_local_port` parity — a `port = 0` mint echoes the allocated number); originator = `TunnelForwardedParams.peer`; correlation-miss → teardown-at-birth (never a half-open channel); the accept is the crate's per-session channel cap (the acceptor vocabulary) | | `direct-streamlocal` | unix-socket forward | `tunnel` | `SubstrateAddr::Path` treatment | compiles wherever `tunnel` compiles; serving is **runtime-Unix-gated** — on Unix serves, on other platforms a clean open-decline (OQ-11 resolved: universal compile + runtime decline, no per-platform feature fork) | | `x11` | X forwarding | — (never) | — | always declines; not feature-gated (ADR-003 §5) | | `auth-agent` | agent forwarding | — (never) | — | always declines; the anti-pattern of the alkvault custody posture (ADR-003 §5) | @@ -56,14 +56,67 @@ command → parse-then-fail). ## Global-request rows (the `-R` listen flow) `Handler::tcpip_forward(address, &mut port) -> bool` + -`Handler::streamlocal_forward` — gate on `tunnel` (decline both the -listen request and the data channels when absent). `streamlocal_forward` -additionally carries the same runtime-Unix gate as its channel row -(OQ-11): off-Unix it declines. The bound-port -out-param (`port = 0` → server picks) is a spec detail in -`client.md`'s lifecycle primitives; disposition follows `tunnel`. -*POC-3 note: only the refusal shape was exercised (egress-only build); -the served listen flow is implementation-phase contract-test territory.* +`Handler::streamlocal_forward` (+ both cancel forms) — gate on +`tunnel` (unfeatured declines both the listen requests and the data +channels). The served flow (`w3-tunnel-listen`, decidable against +alktunnels 0.4.1's actual surfaces with POC-3 §C.4 as the template): + +- **The listen dial** — family-ward through the consumer surface: + `TunnelListener::open(channel_client, TunnelParams {resource: , substrate: Tcp/Unix})`. The resource carries + the wire endpoint string verbatim — the OpenSSH client normalizes + the no-address case to `"localhost"` before it sends + (`channel_rfwd_bind_host` parity), so the door passes through; whose + bind policy governs (wildcard vs loopback — the sshd `GatewayPorts` + posture) is the fabric's establisher/bind closure, assembly policy + at the family's provider. The dial is bounded (`LISTEN_DIAL_TIMEOUT` + = 10 s, matching alkcall's own establisher deadline) and awaited + **inline** in the hook by wire necessity (the global request's + answer + the `&mut u32` out-param are the hook's own; the fabric + leg's answer doesn't ride this session's routing, so the stall is + bounded, not deadlock-class) — recorded as the flow's one inline + serving step. +- **The reply contract**: the minted `listen` token (alkcall ADR-049 + amendment 3 establisher field) is REQUIRED — a producer that minted + no token (or no `bound` report) gets a declined listen with its + fabric channel reaped immediately (a listen the door cannot + correlate arrivals onto never answers success). The bound port + writes the `&mut u32` out-param (`port = 0` → the actually-bound + number — sshd's `allocated_listen_port` reply parity; the + `client.md` lifecycle primitive). +- **The door-side arrival surface** — the fabric's bind-first accept + dispatcher (`make_bindfirst_accept_handler`) dials the + `tunnel/forwarded` op **toward the door** per arrival (the + `ForwardedOpener` closure); the door serves it through the alkcall + openable registered into the shared per-connection registry — the + serve-config's `TunnelListenSeam` (the `w3-seam-config` second + course-correction: the assembly's `Arc` alone + carries no inbound-op serving loop; the seam fronts the + `ServingConfig.registry` the assembly wired at the leg's + `from_connection_with_serving`). The registration runs at + `SshAdapter::new` (a failure is a failed construction — + `AdapterError::ListenRegistration`); alkcall's own gates carry the + per-identity cap + the `tunnel:forwarded` scope per arrival, then + the crate's arrival pump takes over. +- **The arrival pump** — per the `forwarded-tcpip` row above; unix + substrate arrivals (a minted `streamlocal_forward` route) open + `forwarded-streamlocal@openssh.com` with the path echoed (russh's + server `Handle` carries the channel-open at this pin; the OQ-11 + runtime-Unix gate rides the unix listen row the same way it rides + the channel row: off-Unix decline, universal compile). +- **The cancel forms + bounded bookkeeping**: the per-session + listener map (packet-loop-only, like the `channels` store) matches + cancels by (the wire endpoint as the request carried it, the bound + port — sshd's `c->path == host && listening_port == port` parity; + the unix forms by path alone) 1:1 into the family listener's + teardown (`TunnelListener::close`/`Drop` — the route leaves the + shared router *before* the teardown races), `false` for an unknown + pair. The listen count is a bounded resource: a dedicated counter + against the same `Limits::max_concurrent_channels` ceiling (the + listener/channels mix stays unconstrained; both stay bounded — no + unbounded listener map). A session's teardown removes its tokens + from the shared router (a fabric arrival after the session is gone + is the correlation-miss class). ## Request-level rows on a `session` channel @@ -156,7 +209,7 @@ the whole spine, not an implementation choice. | none (default) | — | everything | | `git` | git-parseable exec | pty/shell, plain exec, all other channel types, all `session` requests | | `session` | pty/shell/pipe-exec (requires the configured shell/exec wrapper — OQ-08; an unset field declines its row though the feature is on) | git-parseable exec (the `git` parse is its gate), other channel types | -| `tunnel` | direct/forwarded-tcpip, streamlocal (unix), `-R` listens | session channels (auto-reject), all others | +| `tunnel` | direct/forwarded-tcpip, streamlocal (unix), `-R` listens + arrivals (the listen forms additionally require the configured `TunnelListenSeam` — an egress-only door declines just the `-R` forms) | session channels (auto-reject), all others | | any build + `sftp` slot | unchanged from the underlying build | `sftp` subsystems decline-with-reason until alkfs | Every row is tested in *every* build it can appear in — the decline diff --git a/src/adapter.rs b/src/adapter.rs index 63bb029..0a9664f 100644 --- a/src/adapter.rs +++ b/src/adapter.rs @@ -50,6 +50,8 @@ use tokio::io::{AsyncRead, AsyncWrite}; use crate::config::ServeConfig; use crate::error::{AdapterError, KeySourceError}; use crate::handler::SessionHandler; +#[cfg(feature = "tunnel")] +use crate::tunnel_map; /// The constructed server stack — the substrate every channel-type /// mapping composes into (ADR-001's `SshAdapter`). @@ -70,6 +72,21 @@ impl SshAdapter { /// server, with the [`KeySourceError`] refused through /// [`AdapterError::Construction`]. pub async fn new(serve: ServeConfig) -> Result { + let serve = { + #[cfg(not(feature = "tunnel"))] + { + serve + } + #[cfg(feature = "tunnel")] + { + let mut serve = serve; + if serve.channel_client.is_some() && serve.tunnel_listen_seam.is_some() { + tunnel_map::install_forwarded_surface(&mut serve) + .map_err(AdapterError::ListenRegistration)?; + } + serve + } + }; let keys = serve.host_key_source.load().await?; if keys.is_empty() { return Err(AdapterError::Construction(KeySourceError::Absent)); diff --git a/src/channel_cap.rs b/src/channel_cap.rs index e81fff4..caec428 100644 --- a/src/channel_cap.rs +++ b/src/channel_cap.rs @@ -16,13 +16,15 @@ //! [`crate::dispatch::cap_exceeded_open_decline`]). Release //! ([`ChannelCap::release`]) fires from the `channel_close` hook for //! established channels — and, since `w3-tunnel-direct`, from a -//! spawned route-1 bridge task on a fabric-open failure, where the -//! open was *admitted but never established* (the reject shape means -//! no `CHANNEL_CLOSE` ever cycles back): see the release-sites section. +//! spawned route-1 bridge task on a fabric-open failure — and, since +//! `w3-tunnel-listen`, from the spawned `-R` arrival pumps on the +//! never-established paths (client-rejected open, dead session): see +//! the release-sites section. //! -//! ## Release sites (the `w3-tunnel-direct` recorded decision) +//! ## Release sites (the recorded decision, three sites since +//! `w3-tunnel-listen`) //! -//! Every admission has exactly one release, and the two paths are +//! Every admission has exactly one release, and the paths are //! disjoint by the wire shape: //! //! 1. **The established-channel close** — the `channel_close` hook. @@ -31,37 +33,39 @@ //! is consumed at most once (russh removes the channel before the //! hook). //! 2. **The admitted-but-rejected bridge** — the route-1 bridge's -//! failure path. The wave-3 flow admits at the open hook (the -//! packet-loop single-writer site — no race window against the cap -//! check) and holds the count across the fabric open; on fabric -//! failure the open is rejected with the explicit `ConnectFailed` +//! failure path (`w3-tunnel-direct`). The open hook admits and +//! holds the count across the fabric open; on fabric failure the +//! open is rejected with the explicit `ConnectFailed` //! (ADR-006 decision 1: accept only after the fabric open -//! succeeds), the channel never establishes, and *no `channel_close` -//! can ever fire* — so the count must be released from where the -//! failure is known: inside the spawned bridge task, immediately -//! after the reject is answered. A success never releases there -//! (its close rides path 1), so a failed bridge cannot leak budget -//! and cannot double-release. +//! succeeds), the channel never establishes, and *no +//! `channel_close` can ever fire* — so the count must be released +//! from where the failure is known: inside the spawned bridge +//! task, immediately after the reject is answered. A success never +//! releases there (its close rides path 1). +//! 3. **The `-R` arrival pumps** (`w3-tunnel-listen`): a fabric +//! arrival is admitted into the owning route's count and opens the +//! SSH `forwarded-tcpip`/`forwarded-streamlocal` channel toward +//! the vanilla client through the session handle. A rejected open — +//! the client's own refusal, or a dead session's send failure — +//! never establishes the channel, so the count releases inside the +//! arrival task exactly once (an established channel's release is +//! path 1's). These are the mirror of site 2 on the server-initiated +//! open direction — the acceptor vocabulary's cap rows. //! -//! ## Why the mutex (the single-writer reasoning, updated) +//! ## Why the mutex (the single-writer reasoning, updated twice) //! //! The wave-2 shape was a bare `usize` field written only by the //! session's packet loop — sound because every hook runs inline on //! that loop with exclusive `&mut self` access. The bridge release -//! breaks the single-writer property: the spawned task owns the -//! failure answer and the packet loop cannot be reached from it. The -//! recorded decision is a `parking_lot::Mutex`: **admissions -//! stay packet-loop-only** (every `admit` call site is an -//! `channel_open_*` hook — the cap check and the increment are one -//! atomic section against a shared writer), while **releases have the -//! two sites above** (packet loop + spawned bridge tasks). The mutex -//! is poisoned-lock-resistant (`unwrap_or_else(|e| e.into_inner())` — -//! AGENTS.md convention 2: a panic in one operation must not cascade -//! to other operations), and both release sites saturate -//! (`saturating_sub`), so no path can manufacture capacity with a -//! stray decrement. Channels never closed before session teardown -//! stay counted until the handler dies with the session — per-session -//! state, gone with the connection. +//! (`w3-tunnel-direct`) and the arrival admission (`w3-tunnel-listen`) +//! both break the single-writer property: those tasks are alkcall +//! serving tasks, unreachable from the packet loop. The recorded +//! decision is a `parking_lot::Mutex`: **packet-loop hooks and +//! spawned tasks both touch the count only through the mutex**, their +//! critical sections are single integer ops (no await inside the +//! lock), and all sites saturating. Channels never closed before +//! session teardown stay counted until the handler dies with the +//! session — per-session state, gone with the connection. use parking_lot::Mutex; @@ -76,22 +80,15 @@ pub(crate) struct ChannelCap { } impl ChannelCap { - /// Whether the cap has room — the check form for the rows whose - /// decline is ordered *before* the accept decision without - /// consuming a count: the `forwarded-tcpip` stub (the `-R` data - /// leg's flow is wave-3 material — a declined stub opens nothing, - /// so it must not consume an admission). The accept paths use - /// [`ChannelCap::admit`]; a plain check never mutates. - pub(crate) fn has_room(&self, cap: usize) -> bool { - *self.active.lock() < cap - } - /// The accept-path admission decision: room under `cap` increments /// the count and admits; otherwise refuses (fail-closed — the /// caller answers the open with a clean decline). `cap == 0` - /// refuses every open. The caller site is the packet loop's open - /// hook (the single admit writer — the held count is what the - /// spawned bridge holds across the fabric open). + /// refuses every open. Call sites: the packet loop's open hooks + /// `w2`-style, and — since `w3-tunnel-listen` — the `-R` arrival + /// tasks (the door's forwarded-arrival admission: an arrival is + /// admitted on the owning route's count inside the spawned + /// arrival pump — see the module doc's updated single-writer + /// reasoning). The mutex makes both sound. pub(crate) fn admit(&self, cap: usize) -> bool { let mut active = self.active.lock(); if *active < cap { @@ -134,7 +131,6 @@ mod tests { assert!(cap.admit(CAP)); assert!(cap.admit(CAP)); assert!(!cap.admit(CAP), "beyond the cap: refused"); - assert!(!cap.has_room(CAP)); assert_eq!(cap.active(), CAP); } @@ -148,7 +144,6 @@ mod tests { assert!(!cap.admit(CAP)); cap.release(); - assert!(cap.has_room(CAP)); assert!(cap.admit(CAP), "an open succeeds again after a close"); assert!(!cap.admit(CAP)); } diff --git a/src/config.rs b/src/config.rs index 2864a8b..92db1fc 100644 --- a/src/config.rs +++ b/src/config.rs @@ -31,9 +31,15 @@ //! decline even though the `session` feature is on; //! [`ServeConfig::git_seam`] unset → the git row declines even //! though `git` is on (the decline matrix's config-driven notes; -//! the feature alone configures nothing). The tunnel handle stays -//! the flat `Option>` as landed — a single-field -//! seam needs no struct. +//! the feature alone configures nothing). The tunnel branch carries +//! its flat `Option>` — an egress-only +//! (route-1-only) door needs no struct for leg 1 — and the `-R` +//! listen leg's registration seam landed with `w3-tunnel-listen` +//! (`ServeConfig::tunnel_listen_seam` — the wave-3 seam-config's +//! second course-correction; the shape decision recorded on +//! [`TunnelListenSeam`]: unset → the listen rows decline even with +//! a fabric leg present, because the door-side inbound-op serving +//! loop only exists where the assembly wired it). //! - **Injected handles, never crate-resolved identity or trust**: //! the [`IdentityProvider`], [`HostKeySource`] and //! [`HostKeyPolicy`] handles are `Arc` the assembly @@ -61,6 +67,8 @@ use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; +#[cfg(feature = "tunnel")] +use crate::tunnel_map::ListenRouter; use alkcall::channels::client::ChannelClient; use alkcall::core::auth::IdentityProvider; #[cfg(feature = "session")] @@ -298,6 +306,69 @@ impl std::fmt::Debug for GitSeam { } } +/// The `tunnel` feature's `-R` listen-leg registration seam (the +/// `w3-tunnel-listen` shape — the wave-3 seam-config's second +/// course-correction, resolved per its own decision record): the +/// assembly-supplied `Arc` alone cannot carry the +/// door-side acceptance of the fabric's inbound forwarded opens, +/// because alkcall serves those opens on the **door's channel-0 +/// serving loop**, which exists only when the consumer constructed its +/// `ChannelClient` with +/// [`ChannelClient::from_connection_with_serving`] against a registry +/// the assembly owns. The published alkcall surface has no +/// post-construction way to add that loop, so the serve-config's +/// tunnel handle grows this seam. +/// +/// Same whole-`Option`-struct decision as [`SessionSeam`]/[`GitSeam`]: +/// **`ServeConfig::tunnel_listen_seam` unset → the `-R` listen rows +/// (`tcpip_forward` / `streamlocal_forward` / both cancel forms) +/// decline even under the `tunnel` feature with a fabric leg present** +/// — an egress-only door (the POC-3 stage-C posture) is legitimate +/// config; the feature alone configures nothing. Checked at request +/// time by the mappings; no construction-time validation. The +/// registration itself (the alkcall openable registration the seam +/// fronts) is alkssh-side work at [`crate::SshAdapter::new`] — a +/// failed registration is a failed construction (fail-closed, never a +/// half-configured adapter). +#[cfg(feature = "tunnel")] +pub struct TunnelListenSeam { + /// The **shared** per-connection `OperationRegistry` the fabric + /// leg's inbound ops dispatch against: the exact `Arc` the + /// assembly passed in this leg's `ChannelClient` construction + /// (`ServingConfig.registry`) — alkssh registers the + /// `tunnel/forwarded` openable into it, and alkcall's dispatch + /// reads it per call (the W2 post-hoc-registration posture). The + /// door's own ACL entry rides the registration; arrivals lacking + /// the `tunnel:forwarded` scope are gated by alkcall's + /// `AccessControl::check` fail-closed. + pub registry: Arc, +} + +/// The `OperationRegistry` type behind the seam's shared dispatcher. +/// (Re-imported here for the field's shape; the registration work +/// lives in [`crate::tunnel_map`].) +#[cfg(feature = "tunnel")] +use alkcall::registry::registration::OperationRegistry; + +#[cfg(feature = "tunnel")] +impl TunnelListenSeam { + /// Builds the seam over the assembly's shared per-connection + /// registry (the `ServingConfig.registry` the same fabric leg's + /// `ChannelClient` was constructed with — one `Arc`, both sides). + pub fn new(registry: Arc) -> Self { + Self { registry } + } +} + +#[cfg(feature = "tunnel")] +impl std::fmt::Debug for TunnelListenSeam { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("TunnelListenSeam") + .field("registry", &"") + .finish() + } +} + /// The server direction's assembly-supplied serving configuration /// (`overview.md` §Server stack): everything wave-2's serving loop /// takes at construction. Pure data + injected handles — no russh @@ -376,6 +447,20 @@ pub struct ServeConfig { /// 2: absent + unfeatured both decline; present + featured /// serves. pub channel_client: Option>, + /// The `-R` listen leg's registration seam ([`TunnelListenSeam`] — + /// the shared per-connection registry the fabric's inbound + /// forwarded opens dispatch against). **Unset → the listen rows + /// decline even under the `tunnel` feature with a fabric leg + /// present** (the egress-only door is legitimate config — the + /// POC-3 stage-C posture; shape decision recorded on the type). + #[cfg(feature = "tunnel")] + pub tunnel_listen_seam: Option, + /// The listen leg's per-arrival bookkeeping — installed by + /// [`crate::SshAdapter::new`] when the fabric leg + seam are both + /// present (the registration site). Crate-private: never assembly + /// data. + #[cfg(feature = "tunnel")] + pub(crate) listen_router: Option>, /// The bounded-resources budget this serving direction enforces /// (see [`Limits`]). pub limits: Limits, @@ -404,6 +489,10 @@ impl ServeConfig { identity_provider, host_key_source, channel_client: None, + #[cfg(feature = "tunnel")] + tunnel_listen_seam: None, + #[cfg(feature = "tunnel")] + listen_router: None, limits: Limits::default(), } } @@ -431,6 +520,16 @@ impl ServeConfig { self.channel_client.is_some() } + /// Whether the `-R` listen rows' registration seam is configured + /// in this configuration ([`TunnelListenSeam`] present; the + /// `tunnel` feature gate applies at the dispatch on top): `false` + /// means the listen rows decline — the unset-declines contract, + /// checked at request time by the mappings, never here. + #[cfg(feature = "tunnel")] + pub fn tunnel_listen_seam_configured(&self) -> bool { + self.tunnel_listen_seam.is_some() + } + /// Whether the session rows' family seam is configured in this /// configuration ([`SessionSeam`] present; the `session` feature /// gate applies at the dispatch on top): `false` means the rows @@ -470,10 +569,12 @@ impl std::fmt::Debug for ServeConfig { { debug.field("git_seam", &self.git_seam.is_some()); } - debug - .field("channel_client", &self.channel_client.as_ref().is_some()) - .field("limits", &self.limits) - .finish() + debug.field("channel_client", &self.channel_client.as_ref().is_some()); + #[cfg(feature = "tunnel")] + { + debug.field("tunnel_listen_seam", &self.tunnel_listen_seam.is_some()); + } + debug.field("limits", &self.limits).finish() } } @@ -982,6 +1083,36 @@ mod tests { } } + /// - *Contract stamp:* the `w3-tunnel-listen` seam shape (the + /// unset-declines contract's tunnel extension): the seam starts + /// unset (the listen rows decline — an egress-only door is + /// legitimate config), sets independently of the fabric handle, + /// and Debug-shapes without registry internals. + #[cfg(feature = "tunnel")] + #[test] + fn serve_config_listen_seam_unset_declines_and_sets_independently() { + let serve = ServeConfig::new(Arc::new(NoopIdentityProvider), host_key_source()); + + assert!(serve.tunnel_listen_seam.is_none()); + assert!(!serve.tunnel_listen_seam_configured()); + assert!(!serve.tunnel_configured()); + + let debug = format!("{serve:?}"); + assert!(debug.contains("tunnel_listen_seam: false")); + + let mut serve = ServeConfig::new(Arc::new(NoopIdentityProvider), host_key_source()); + serve.tunnel_listen_seam = Some(super::super::TunnelListenSeam::new(Default::default())); + assert!(serve.tunnel_listen_seam_configured()); + + let debug = format!("{serve:?}"); + assert!(debug.contains("tunnel_listen_seam: true")); + let seam_debug = format!("{:?}", serve.tunnel_listen_seam.as_ref().expect("holds")); + assert!( + seam_debug.contains(""), + "the seam Debug shapes the registry without internals: {seam_debug}" + ); + } + #[tokio::test] async fn dial_config_key_source_stays_fail_closed_through_the_config() { let dial = DialConfig::new("fleet.example:22", key_source(), Arc::new(RejectAll)); diff --git a/src/error.rs b/src/error.rs index 501c61d..9a6851b 100644 --- a/src/error.rs +++ b/src/error.rs @@ -104,15 +104,19 @@ pub enum KeySourceError { Decode(String), } -/// The adapter (server-direction) error surface — the two failure -/// classes the mount point can reach, deliberately closed: +/// The adapter (server-direction) error surface — the failure classes +/// the mount point can reach, deliberately closed: /// /// - **Construction** — the [`SshAdapter::new`](crate::SshAdapter::new) -/// host-key load failed: the [`KeySourceError`] variant is preserved -/// verbatim (locked / absent / decryption / unknown-version / decode -/// each stays itself) and **no adapter is constructed** (ADR-005 -/// decision 6 — never a half-configured adapter, never a keyless -/// server that fails later). +/// setup failed *before any session exists*: a failed host-key load +/// (the [`KeySourceError`] variant is preserved verbatim — locked / +/// absent / decryption / unknown-version / decode each stays itself) +/// or a refused `-R` listen-leg registration (`ListenRegistration`, +/// `w3-tunnel-listen` — a fabric that could mint listens the door +/// cannot serve is worse than a half-configured adapter). +/// **No adapter is constructed** (ADR-005 decision 6 — never a +/// half-configured adapter, never a keyless server that fails +/// later). /// - **Wire** — the russh session stack's own error, mapped /// transparently (`From`, the wave-1 error-model's /// reserved `From` requirement of the per-connection @@ -122,7 +126,7 @@ pub enum KeySourceError { /// dispatch's SSH-shaped responses, never this taxonomy. /// /// There is no catch-all variant: everything the crate's server -/// direction can fail is one of these two classes or a decline. +/// direction can fail is one of these classes or a decline. #[derive(Debug, Error)] pub enum AdapterError { /// The host-key load at adapter construction refused — fail-closed @@ -131,6 +135,14 @@ pub enum AdapterError { #[error("ssh server construction refused: {0}")] Construction(#[from] KeySourceError), + /// The `-R` listen leg's door-side openable registration refused + /// at construction (`w3-tunnel-listen`): fail-closed, never a + /// half-configured adapter — a construction where the fabric could + /// mint listens the door cannot serve is worse than no adapter. + /// The string carries the registration failure detail. + #[error("ssh server construction refused (listen registration): {0}")] + ListenRegistration(String), + /// A mounted russh session errored (transport / KEX / IO class). #[error("ssh session error: {0}")] Wire(#[from] russh::Error), @@ -209,21 +221,27 @@ mod tests { .into_iter() .map(AdapterError::Construction) .collect(); + all.push(AdapterError::ListenRegistration("duplicate".into())); all.push(AdapterError::Wire(russh::Error::UnknownKey)); all } #[test] fn adapter_error_closed_taxonomy_two_classes() { - let exhaustive: fn(&AdapterError) -> &'static str = |e| match e { - AdapterError::Construction(_) => "construction", - AdapterError::Wire(_) => "wire", - }; + let exhaustive: fn(&AdapterError) -> &'static str = match_fn(); for v in all_adapter_variants() { assert!(!exhaustive(&v).is_empty()); } } + fn match_fn() -> fn(&AdapterError) -> &'static str { + |e| match e { + AdapterError::Construction(_) => "construction", + AdapterError::ListenRegistration(_) => "construction", + AdapterError::Wire(_) => "wire", + } + } + #[test] fn construction_preserves_the_key_source_variant() { for source in all_variants() { diff --git a/src/handler.rs b/src/handler.rs index d56f58c..f0258d8 100644 --- a/src/handler.rs +++ b/src/handler.rs @@ -56,8 +56,12 @@ //! only after the fabric open succeeds, `ConnectFailed` on failure //! ([`crate::dispatch::connect_failed_open_decline`], the wave-2 //! seam stub's reason surviving as the real decline). -//! `forwarded-tcpip` stays the stub until `w3-tunnel-listen` lands -//! the `-R` data leg's flow (the mapping row's listen-flow shape). +//! An **inbound** `forwarded-tcpip` open — arriving *from* the vanilla +//! client — declines with the same explicit `ConnectFailed` as the +//! unvanilla-arrival class (`w3-tunnel-listen`: the real `-R` data +//! leg is the door's own server-initiated opens, driven by the +//! [`crate::tunnel_map`] arrival pump; a vanilla client never +//! initiates the type, OpenSSH parity). //! The bare //! `auth-agent@openssh.com` channel *type* //! (`forwarded-streamlocal@openssh.com` likewise) has no russh handler @@ -154,7 +158,14 @@ //! the open *before* any channel is owned — nothing is served yet, //! so nothing spawns. The route-1 bridge (`w3-tunnel-direct`) moved //! the direct rows' accept past the fabric open: the accept and the -//! pump ride the spawned task, per the same constraint. +//! pump ride the spawned task, per the same constraint. The `-R` +//! listen rows' family-ward dial is the one **inline await** on the +//! loop (`w3-tunnel-listen`'s recorded decision: the global request's +//! reply + the `&mut u32` bound-port out-param are the hook's own — +//! no spawned-task answer exists at this pin — and the fabric's +//! answer rides the fabric leg's connections, not this session's +//! routing; bounded by `LISTEN_DIAL_TIMEOUT`), while everything under +//! the reply — the arrivals, the pumps — stays spawned. //! //! Request-level declines: `subsystem_request` declines every name — //! `sftp` is the OQ-13 placeholder (the feature exists, the route @@ -164,9 +175,11 @@ //! declines unconditionally (OQ-09 — no allowlist ships: wire env is //! not family config). `x11_request` and `agent_request` decline //! (ADR-003 §5). The global listen requests (`tcpip_forward` -//! / `streamlocal_forward` / the cancel forms) decline — they are -//! `tunnel` rows whose serving flow is wave-3 material, so featured -//! builds decline them too this wave. +//! / `streamlocal_forward` + their cancel forms) — the `-R` rows +//! (`w3-tunnel-listen`): the real flow under `tunnel` with the listen +//! seam configured (the family-ward `TunnelListener::open` dial, the +//! arrivals' arrival pump through the shared router); every decline +//! class keeps a global-request-level reason (see the hooks). //! //! ## Wire decline reasons (the wave-1 error-model reservation) //! @@ -179,7 +192,8 @@ //! prohibited; the direct rows' cannot-bridge classes (the fabric //! entry unconfigured under the feature, the failed fabric open, an //! impossible wire target), the off-Unix `direct-streamlocal` -//! decline, and the `forwarded-tcpip` stub → connect failed). These +//! decline, and the unvanilla inbound `forwarded-tcpip` open → +//! connect failed). These //! happen before the channel exists, so the SSH open-failure code is //! the honest, richest reason surface. //! - **Request-level declines** — [`Session::channel_failure`]: the @@ -233,7 +247,14 @@ //! spawned task on a fabric-open failure (the admitted-but-rejected //! shape: ADR-006 decision 1 rejects the open, so no `channel_close` //! can ever fire for it; the release mechanism is the shared -//! [`ChannelCap`] `Arc`, recorded in that module). The order of the +//! [`ChannelCap`] `Arc`, recorded in that module). The `-R` leg adds +//! its two budget surfaces (`w3-tunnel-listen`, recorded in +//! `channel_cap.rs`): the arrivals admit into the owning route's +//! count inside the spawned arrival tasks (a beyond-cap arrival never +//! opens an SSH channel — teardown-at-birth), and the listen count is +//! its own dedicated counter against the same ceiling (the +//! listener/channels mix stays unconstrained; both stay bounded). The +//! order of the //! gates: the feature gate first (an unfeatured build declines //! everything before the cap matters), then the pure-decline classes //! (the tunnel rows' unset-handle/unix-gate/identity/invalid-target @@ -277,7 +298,7 @@ use crate::fingerprint::fingerprint; #[cfg(feature = "git")] use crate::git_map; use crate::observe; -#[cfg(any(feature = "git", feature = "session"))] +#[cfg(any(feature = "git", feature = "session", feature = "tunnel"))] use crate::serve; #[cfg(feature = "session")] use crate::tty_map; @@ -298,14 +319,30 @@ use alktty::negotiation::NegotiateRequest; /// [`Self::resolved_identity`] — the wave-3 seam reads it). pub struct SessionHandler { serve: Arc, - /// The in-crate concurrent-channel cap (`ChannelCap`), shared as - /// an `Arc` since `w3-tunnel-direct`: the route-1 bridge's spawned - /// task releases the count on a fabric-open failure (the + /// The in-crate concurrent-channel cap (`w2-bounded-resources`), + /// shared as an `Arc` since `w3-tunnel-direct`: the route-1 bridge's + /// spawned task releases the count on a fabric-open failure (the /// admitted-but-rejected shape — no `channel_close` can ever fire - /// for it; see `channel_cap.rs`'s release-sites record). Admits - /// stay packet-loop-only (every `admit` call site is an open - /// hook); the mutex makes the bridge-task releases sound. + /// for it; see `channel_cap.rs`'s release-sites record). Admissions + /// have two sites since `w3-tunnel-listen` — the packet loop's + /// open hooks and the `-R` arrival tasks — and releases three; the + /// mutex makes them sound (the reasoning, recorded in + /// `channel_cap.rs`). channel_cap: Arc, + /// The `-R` listen leg's per-session bookkeeping + /// (`w3-tunnel-listen`): the minted-listen map (token → pending + /// listen; written only by the packet loop's hooks — the listen + /// global requests mints, the cancel forms + this handler's + /// teardown remove, single-writer/no-lock like the `channels` + /// store) and the listen count's dedicated + /// `max_concurrent_channels`-ceiling counter (a listen is at least + /// one future channel's potential; its own counter keeps the + /// listener/channels mix unconstrained while both stay bounded — + /// no unbounded listener map). + #[cfg(feature = "tunnel")] + listeners: HashMap, + #[cfg(feature = "tunnel")] + listener_cap: tunnel_map::ListenerCap, resolved_identity: Option, /// The store-and-take handoff (`w3-git-mapping`): confirmed /// session channels held from the open until their row takes or @@ -334,14 +371,28 @@ pub struct SessionHandler { teardown: tty_map::Teardown, } -#[cfg(feature = "session")] impl Drop for SessionHandler { /// Fires the session rows' teardown signal (`w3-tty-session`): /// every spawned serving task of this connection watches it and /// cancels its alktty session (the kill-on-drop cleanup) instead - /// of outliving the connection. + /// of outliving the connection. The `tunnel` build adds the `-R` + /// leg's teardown (`w3-tunnel-listen`): the session's pending + /// listens' tokens leave the shared router (a fabric arrival after + /// the session is gone is the correlation-miss class — + /// teardown-at-birth) and the listener objects drop here (the + /// fabric-side reap). Any in-flight arrival already past the + /// route lookup fails its dead-handle open and releases its + /// admission — the never-established path. fn drop(&mut self) { + #[cfg(feature = "session")] self.teardown.fire(); + #[cfg(feature = "tunnel")] + { + let tokens: Vec = self.listeners.keys().copied().collect(); + if let Some(router) = &self.serve.listen_router { + router.remove_tokens(&tokens); + } + } } } @@ -362,6 +413,10 @@ impl SessionHandler { bridges: HashMap::new(), #[cfg(feature = "session")] teardown: tty_map::Teardown::new(), + #[cfg(feature = "tunnel")] + listeners: HashMap::new(), + #[cfg(feature = "tunnel")] + listener_cap: tunnel_map::ListenerCap::default(), } } @@ -405,19 +460,6 @@ impl SessionHandler { .admit(self.serve.limits.max_concurrent_channels) } - /// Whether the cap is already exhausted — the gate slot for the - /// one row whose decline is still ordered *before* an accept - /// decision without consuming an admission: the `forwarded-tcpip` - /// stub (the `-R` data leg's flow is `w3-tunnel-listen` material — - /// a declined stub opens nothing, so it must not consume a count; - /// the direct rows moved to [`Self::admit_channel_open`]-at-the- - /// accept flow when `w3-tunnel-direct` landed the route-1 bridge). - fn channel_cap_exhausted(&self) -> bool { - !self - .channel_cap - .has_room(self.serve.limits.max_concurrent_channels) - } - /// The ADR-004 decision 1 bridge core, structured out of the /// [`Handler::auth_publickey`] hook so the contract tests can drive /// it directly: the hook's russh-owned argument shape @@ -690,12 +732,16 @@ impl russh::server::Handler for SessionHandler { Ok(()) } - /// The `forwarded-tcpip` open gate (`tunnel` — the `-R` data - /// leg): the same gate/cap/stub split as `direct-tcpip`. The - /// harness's russh client cannot initiate this channel type (the - /// server opens it toward the client), so its wire rows are - /// wave-3 acceptor material; the gate itself is straight-line - /// identical to the other tunnel rows. + /// The `forwarded-tcpip` open gate (`tunnel`): an **inbound** open + /// of this channel type — arriving *from* the vanilla client — is + /// not the `-R` leg at all (the `-R` data leg is the door's own + /// server-initiated opens via [`crate::tunnel_map`]'s arrival + /// pump; a vanilla client never initiates `forwarded-tcpip`, + /// OpenSSH parity). The gate + the unvanilla decline stay + /// straight-line and inline: nothing serves, nothing was ever in + /// flight, so no admission is touched, and the explicit + /// `ConnectFailed` reject keeps the wave-2 wire shape (the event's + /// class records it as the unvanilla-arrival decline). async fn channel_open_forwarded_tcpip( &mut self, channel: Channel, @@ -712,12 +758,8 @@ impl russh::server::Handler for SessionHandler { reply.reject(dispatch::unfeatured_open_decline()).await; return Ok(()); } - if self.channel_cap_exhausted() { - observe::decline_open(ROW, "cap-exceeded", channel.id().number()); - reply.reject(dispatch::cap_exceeded_open_decline()).await; - return Ok(()); - } - tunnel_row_stub(ROW, channel.id().number(), reply).await; + observe::decline_open(ROW, "unvanilla-inbound", channel.id().number()); + reply.reject(dispatch::connect_failed_open_decline()).await; Ok(()) } @@ -1098,57 +1140,297 @@ impl russh::server::Handler for SessionHandler { Ok(false) } - /// The `tcpip-forward` listen request (`-R` ingress): a `tunnel` - /// row whose serving flow is wave-3 material — the gate exists - /// ([`Self::serves_tunnels`]) but even a featured build declines - /// this wave (never advertise a listen that cannot serve). The - /// bound-port `port = 0` picking rule rides the wave-3 flow. + /// The `tcpip-forward` listen request (`-R` ingress, `w3-tunnel- + /// listen`): the gate is [`Self::serves_tunnels`]; the real flow + /// rides [`listen_request_dispatch`] (`tunnel` builds) — the + /// family-ward `TunnelListener::open` dial, the bound-port out-param + /// (`port = 0` → the fabric's actually-bound number, the sshd + /// parity), the route + listener registration. Unfeatured builds + /// decline with the global-request class; never a minted listen a + /// door cannot serve. async fn tcpip_forward( &mut self, - _address: &str, - _port: &mut u32, + address: &str, + port: &mut u32, _session: &mut Session, ) -> Result { - observe::decline_global_request("tcpip_forward", "unimplemented"); - Ok(false) + #[cfg(feature = "tunnel")] + { + listen_request_dispatch( + self, + "tcpip_forward", + address, + alktunnels::params::Substrate::Tcp, + Some(port), + _session, + ) + .await + } + #[cfg(not(feature = "tunnel"))] + { + let _ = (address, port); + observe::decline_global_request("tcpip_forward", "unfeatured"); + Ok(false) + } } - /// Nothing is ever registered this wave — the cancel form - /// declines accordingly. + /// The `cancel-tcpip-forward` cancel form (`-KR`, mux teardown, + /// `ExitOnForwardFailure`): 1:1 into the family listener's teardown — + /// the pending listen matched by (the wire address as the request + /// carried it, the bound port), the route removed **before** any + /// teardown (a racing arrival is already the correlation-miss + /// class), the listener count released, the fabric channel's + /// teardown spawned (`TunnelListener::close`). `false` for a + /// (address, port) pair with no pending listen — the sshd parity + /// (`cancel of a not-found forward`). async fn cancel_tcpip_forward( &mut self, - _address: &str, - _port: u32, + address: &str, + port: u32, _session: &mut Session, ) -> Result { - observe::decline_global_request("cancel_tcpip_forward", "unimplemented"); - Ok(false) + #[cfg(feature = "tunnel")] + { + cancel_listen_dispatch(self, "cancel_tcpip_forward", address, Some(port)) + } + #[cfg(not(feature = "tunnel"))] + { + let _ = (address, port); + observe::decline_global_request("cancel_tcpip_forward", "unfeatured"); + Ok(false) + } } - /// The `streamlocal-forward` listen request: the same decline as - /// `tcpip-forward`, plus its runtime-Unix gate (OQ-11) riding the - /// wave-3 serving flow it will gate. + /// The `streamlocal-forward` listen request: the same listen flow + /// as `tcpip_forward` over the unix substrate — with the OQ-11 + /// runtime-Unix gate riding the serving flow it serves (off-Unix + /// it declines before any fabric call; universal compile, no + /// per-platform feature fork). No port out-param (the unix listen + /// carries no port semantics; the path echoes verbatim). async fn streamlocal_forward( &mut self, - _socket_path: &str, + socket_path: &str, _session: &mut Session, ) -> Result { - observe::decline_global_request("streamlocal_forward", "unimplemented"); - Ok(false) + #[cfg(feature = "tunnel")] + { + if !cfg!(unix) { + observe::decline_global_request("streamlocal_forward", "unix-unavailable"); + return Ok(false); + } + listen_request_dispatch( + self, + "streamlocal_forward", + socket_path, + alktunnels::params::Substrate::Unix, + None, + _session, + ) + .await + } + #[cfg(not(feature = "tunnel"))] + { + let _ = socket_path; + observe::decline_global_request("streamlocal_forward", "unfeatured"); + Ok(false) + } } - /// Nothing is ever registered this wave — the cancel form - /// declines accordingly. + /// The `cancel-streamlocal-forward` form: the tcpip cancel's shape + /// over the unix substrate (path-match only; no port). async fn cancel_streamlocal_forward( &mut self, - _socket_path: &str, + socket_path: &str, _session: &mut Session, ) -> Result { - observe::decline_global_request("cancel_streamlocal_forward", "unimplemented"); - Ok(false) + #[cfg(all(feature = "tunnel", unix))] + { + cancel_listen_dispatch(self, "cancel_streamlocal_forward", socket_path, None) + } + #[cfg(all(feature = "tunnel", not(unix)))] + { + let _ = socket_path; + observe::decline_global_request("cancel_streamlocal_forward", "unix-unavailable"); + Ok(false) + } + #[cfg(not(feature = "tunnel"))] + { + let _ = socket_path; + observe::decline_global_request("cancel_streamlocal_forward", "unfeatured"); + Ok(false) + } } } +/// The `tcpip_forward` / `streamlocal_forward` global requests' real +/// flow (`w3-tunnel-listen`; `mapping.md` §Global-request rows): the +/// gate order is decline-first-alloc-never, each consumer checked +/// before the next allocation is taken: +/// +/// 1. **The `no-identity` guard** — unreachable on the wire (russh +/// authorizes before requests serve) and fail-closed: never +/// anonymous listens. +/// 2. **The seam checks** — the fabric handle or the listen seam +/// (`listen_router` absent: an egress-only door — the seam unset or +/// the registration never ran) decline `seam-unconfigured` before +/// any dial. +/// 3. **The OQ-11 unix gate** — the streamlocal variant's off-Unix +/// decline rides its hook (checked before this fn runs, so the +/// unreachable-on-Unix-only row never here). +/// 4. **The listener cap** — the listen count's dedicated +/// `max_concurrent_channels`-ceiling admission; beyond it +/// `cap-exceeded`. +/// +/// Then the family-ward dial ([`crate::tunnel_map::open_listen`], the +/// `resource = ` naming convention) — awaited +/// inline (the recorded inline decision: the global request's reply + +/// the `&mut u32` out-param are the hook's; the fabric answer rides +/// the fabric leg, not this session's packet loop; the dial is +/// bounded by `LISTEN_DIAL_TIMEOUT`, so the stall is bounded). The +/// reply's `bound` port writes the out-param (`port = 0` → the +/// actually-bound number, sshd parity); the route registers into the +/// shared router (session-scope: the owning session's handle + its +/// channel-cap `Arc`) and the pending listen into the session's +/// packet-loop-only map. `Ok(true)`'s answer is the only success +/// shape; every decline path answers the vanilla client's +/// `RequestDenied` via `Ok(false)`. +#[cfg(feature = "tunnel")] +async fn listen_request_dispatch( + handler: &mut SessionHandler, + row: &'static str, + wire_endpoint: &str, + substrate: alktunnels::params::Substrate, + port: Option<&mut u32>, + session: &mut Session, +) -> Result { + if handler.resolved_identity().is_none() { + observe::decline_global_request(row, "no-identity"); + return Ok(false); + } + let (Some(client), Some(router)) = ( + handler.serve.channel_client.as_ref(), + handler.serve.listen_router.as_ref(), + ) else { + observe::decline_global_request(row, "seam-unconfigured"); + return Ok(false); + }; + if !handler + .listener_cap + .admit(handler.serve.limits.max_concurrent_channels) + { + observe::decline_global_request(row, "cap-exceeded"); + return Ok(false); + } + let minted = match tunnel_map::open_listen( + client, + tunnel_map::listen_resource(wire_endpoint, substrate, port.as_deref().copied()), + substrate, + ) + .await + { + Ok(minted) => minted, + Err(_) => { + handler.listener_cap.release(); + observe::decline_global_request(row, "listen-failed"); + return Ok(false); + } + }; + let bound = minted.bound; + let echo = match (&substrate, &bound) { + ( + alktunnels::params::Substrate::Tcp, + alktunnels::params::SubstrateAddr::HostPort { + port: bound_port, .. + }, + ) => tunnel_map::RouteEcho::Tcp { + address: wire_endpoint.to_owned(), + port: u32::from(*bound_port), + }, + (alktunnels::params::Substrate::Unix, alktunnels::params::SubstrateAddr::Path { path }) => { + tunnel_map::RouteEcho::Unix { path: path.clone() } + } + _ => { + handler.listener_cap.release(); + observe::decline_global_request(row, "listen-failed"); + return Ok(false); + } + }; + if let (Some(port), Some(bound_port)) = (port, tunnel_map::bound_port(&bound)) { + *port = bound_port; + } + let token = minted.token; + router.register( + token, + tunnel_map::ListenRoute { + echo, + substrate, + handle: Arc::new(session.handle()), + cap: Arc::clone(&handler.channel_cap), + cap_value: handler.serve.limits.max_concurrent_channels, + }, + ); + handler.listeners.insert( + token, + tunnel_map::PendingListen { + wire_address: wire_endpoint.to_owned(), + bound_port: tunnel_map::bound_port(&bound).unwrap_or(0), + substrate, + listener: minted.listener, + }, + ); + Ok(true) +} + +/// The `cancel_tcpip_forward` / `cancel_streamlocal_forward` global +/// requests' real flow (`w3-tunnel-listen`): 1:1 into the family +/// listener's teardown. Match rule (sshd parity): the wire endpoint as +/// the original request carried it + the bound port for the tcp +/// variants (the client sends the same normalized string it sent +/// originally, and the *allocated* port for `port = 0` mints — the +/// value the listen reply carried); the path alone for the unix +/// variant. Order: the route leaves the shared router **before** the +/// entry leaves the session map (so a racing arrival is already the +/// correlation-miss class — no post-cancel half-serve), then the +/// listener count releases, then the family teardown spawns +/// (`TunnelListener::close` inside the spawn — a bounded local +/// shutdown + entry reap, not a serving path the hook may await). +/// `Ok(true)` iff a pending listen matched. +#[cfg(feature = "tunnel")] +fn cancel_listen_dispatch( + handler: &mut SessionHandler, + row: &'static str, + wire_endpoint: &str, + port: Option, +) -> Result { + if handler.resolved_identity().is_none() { + observe::decline_global_request(row, "no-identity"); + return Ok(false); + } + let token = handler.listeners.iter().find_map(|(token, pending)| { + let port_matched = match port { + Some(port) => pending.bound_port == port, + None => pending.substrate == alktunnels::params::Substrate::Unix, + }; + (pending.wire_address == wire_endpoint && port_matched).then_some(*token) + }); + let Some(token) = token else { + observe::decline_global_request(row, "no-such-listen"); + return Ok(false); + }; + let Some(entry) = handler.listeners.remove(&token) else { + observe::decline_global_request(row, "no-such-listen"); + return Ok(false); + }; + if let Some(router) = &handler.serve.listen_router { + router.cancel(&token); + } + handler.listener_cap.release(); + serve::spawn_serving_path(async move { + entry.listener.close().await; + }); + Ok(true) +} + /// The request-level decline shape for an unfeatured/unimplemented /// capability: fail the request and leave the channel usable (POC-1 /// §5 — session survives declines, no state poisoning). The decline @@ -1440,14 +1722,3 @@ async fn direct_open_dispatch( Arc::clone(&handler.channel_cap), ); } - -/// The `forwarded-tcpip` seam stub — the `-R` data leg's decline until -/// `w3-tunnel-listen` lands the listen flow this leg rides (the -/// harness's russh client cannot even initiate this channel type — -/// the server opens it toward the client — so its wire rows are -/// acceptor material). The open-level decline event is emitted here: -/// the explicit `ConnectFailed` reject ADR-006 decision 1 pins. -async fn tunnel_row_stub(row: &'static str, channel_id: u32, reply: ChannelOpenHandle) { - observe::decline_open(row, "connect-failed", channel_id); - reply.reject(dispatch::connect_failed_open_decline()).await; -} diff --git a/src/lib.rs b/src/lib.rs index a7d82a6..71747bb 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -47,6 +47,8 @@ pub use adapter::SshAdapter; pub use config::GitSeam; #[cfg(feature = "session")] pub use config::SessionSeam; +#[cfg(feature = "tunnel")] +pub use config::TunnelListenSeam; pub use config::{DialConfig, Limits, ServeConfig}; pub use error::{AdapterError, KeySourceError}; pub use fingerprint::fingerprint; diff --git a/src/observe.rs b/src/observe.rs index 87bbd95..1ab53c4 100644 --- a/src/observe.rs +++ b/src/observe.rs @@ -55,7 +55,8 @@ //! //! Decline events additionally carry exactly: //! -//! - `decline_class` — `"open"` | `"request"` | `"serving"` | `"auth"`: +//! - `decline_class` — `"open"` | `"request"` | `"serving"` | +//! `"arrival"` | `"auth"`: //! the two wire decline classes of `mapping.md` §Wire decline //! reasons, plus the serving-path class (this wave the seam stub's //! decline *is* the serving path; the wave-3 mappings emit the same @@ -93,8 +94,16 @@ //! path guards — unreachable in practice, fail-closed; the tunnel //! rows' open-level `no-identity` guard rides the same value — //! reachable never, on the wire, for the same reason), -//! `"unimplemented"` (the subsystem rows — OQ-13 — and the wave-3 -//! listen forms), `"declined-by-design"` (env, OQ-09), and on the +//! `"unimplemented"` (the subsystem rows — OQ-13), +//! `"declined-by-design"` (env, OQ-09), +//! `"correlation-miss"` / `"cap-exceeded"` / `"accept-refused"` / +//! `"unvanilla-inbound"` (`w3-tunnel-listen`: the `-R` leg's arrival +//! declines — the class whose events carry the minted `listen` token +//! instead of a channel id — and the unvanilla inbound +//! `forwarded-tcpip` open), +//! `"listen-failed"` / `"no-such-listen"` (`w3-tunnel-listen`: the +//! listen dial's failed/bounded outcome and the cancel forms' +//! not-found pair) — the global-request and arrival rows, and on the //! auth class `"method-unimplemented"` / `"cert-unsupported"` / //! `"non-ed25519"` / `"fingerprint-unresolved"`. //! - `exec_route` — only on `exec_request`-row events: the dispatch @@ -225,6 +234,28 @@ pub(crate) fn decline_global_request(row: &'static str, reason: &'static str) { ); } +/// A decline on the `-R` leg's **arrival** surface (`w3-tunnel- +/// listen`): a fabric inbound forwarded open that cannot become an +/// SSH channel — `correlation-miss` (no route for the minted token — +/// teardown-at-birth), `cap-exceeded` (the per-arrival admission), or +/// `accept-refused` (the vanilla client rejected the SSH open — the +/// count released, never a half-open channel). No SSH channel id +/// exists yet at this surface (that is the point of the declines), so +/// the minted `listen` token — the correlation key itself — rides +/// instead; the token is bookkeeping metadata (the producer's minted +/// correlation number), never secret material. +#[cfg(feature = "tunnel")] +pub(crate) fn decline_arrival(row: &'static str, reason: &'static str, listen: u32) { + tracing::debug!( + feature_build = FEATURE_BUILD, + row, + listen, + decline_class = "arrival", + reason, + "decline: forwarded arrival rejected" + ); +} + /// A serving-path decline: a routed row whose serving task answers /// failure-then-close rather than serving. This wave that is the seam /// stub (the wire shape pinned in `tests/spawn_return.rs`); the wave-3 @@ -467,6 +498,31 @@ mod tests { assert_eq!(field(&fields, "error"), "Probe"); } + /// - *Contract stamp:* the arrival vocabulary (`w3-tunnel-listen`) + /// — an arrival decline carries the minted `listen` token + /// instead of a channel id (none exists yet — that is the point + /// of the declines; the token is bookkeeping metadata, never + /// secret material). + #[cfg(feature = "tunnel")] + #[test] + fn decline_arrival_carries_the_listen_token_not_a_channel_id() { + let fields = capture(|| decline_arrival("forwarded-tcpip", "correlation-miss", 9182)); + assert_eq!(field(&fields, "feature_build"), FEATURE_BUILD); + assert_eq!(field(&fields, "row"), "forwarded-tcpip"); + assert_eq!(field(&fields, "listen"), "9182"); + assert_eq!(field(&fields, "decline_class"), "arrival"); + assert_eq!(field(&fields, "reason"), "correlation-miss"); + assert_eq!( + field(&fields, "message"), + "decline: forwarded arrival rejected" + ); + let names: Vec = fields.iter().map(|(n, _)| n.clone()).collect(); + assert!( + !names.iter().any(|n| n == "channel_id"), + "an arrival decline carries the listen token, never a channel id" + ); + } + /// The feature-build string: every enabled feature appears; a /// featureless build reads `none`. The expected half is derived /// from `cfg!` independently of the crate's match — divergence diff --git a/src/tunnel_map.rs b/src/tunnel_map.rs index f044d47..9a921d2 100644 --- a/src/tunnel_map.rs +++ b/src/tunnel_map.rs @@ -30,6 +30,129 @@ //! observes a clean, channel-scoped open refusal (POC-1 §5 posture) //! — vanilla clients retry-observe a down forward instead of a hang. //! +//! ## The `-R` listen flow (`w3-tunnel-listen` — the fabric listens, +//! the client dials) +//! +//! The server direction of the route-1 bridge (`mapping.md` +//! §Channel-type rows, the `forwarded-tcpip` row + §Global-request +//! rows). Per the family door semantics: the door owns nothing local +//! — **the fabric listens**; per arrival the data flows back toward +//! the vanilla SSH client, which dials its own target. Three moving +//! parts, all decided against alktunnels 0.4.1's actual surfaces with +//! POC-3 §C.4 as the template: +//! +//! 1. **The listen dial** (`tcpip_forward` / `streamlocal_forward`): +//! the door maps the vanilla listen request family-ward as +//! `TunnelParams { resource: , +//! substrate: Tcp/Unix }` — the **resource-naming convention +//! recorded** (the one open surface this flow's composition had): +//! ADR-001 keeps the wire field a *resource identifier*, and a +//! fabric serving vanilla `-R` opt into dynamic-address binding by +//! its own establisher/bind-closure treating the mapped string as +//! its bind endpoint (`bindfirst_establisher` + +//! `register_openable_with_establisher`, the alkcall ADR-049 +//! establisher shape — who may bind what is the fabric's ACL + +//! its bind closure, assembly policy at the family's provider, +//! never the door's). The dial is the published consumer surface — +//! `TunnelListener::open(client, params)` on the held +//! `ChannelClient` — bounded by [`LISTEN_DIAL_TIMEOUT`]; the +//! reply's `bound` + minted `listen` token (the alkcall ADR-049 +//! amendment-3 establisher fields) are read through +//! `parse-listen-reply`-class accessors; **a reply without both is +//! a declined listen with the fabric listener torn down +//! immediately** (never a minted-but-uncorrelatable listen — a +//! declined listen never left a half state). +//! 2. **The door-side arrival surface** — the composition decision +//! (the task's second `w3-seam-config` course correction): the +//! fabric dials `tunnel/forwarded` *toward the door* per alktunnels' +//! bind-first dispatch (`make_bindfirst_accept_handler` + +//! `ForwardedOpener` — the producer calls the op on the channel-0 +//! call plane toward the listen opener, and the door reply carries +//! the door-allocated channel id the producer adopts; alkcall's +//! odd/even split keeps the two open directions collision-free). +//! Serving that op needs the door's channel-0 serving loop — an +//! alkcall `ChannelClient` built with +//! `from_connection_with_serving`, whose registry the assembly owns +//! — so the serve-config carries [`crate::config::TunnelListenSeam`] +//! (the same `Arc` the assembly wired); alkssh +//! registers the forwarded openable itself in +//! [`install_forwarded_surface`] (called from +//! [`crate::SshAdapter::new`]; the registration failure is a failed +//! construction): a `ChannelCore` over a clone of the held +//! client's manager + alktunnels' `tunnel_forwarded_spec()` + +//! `make_forwarded_op_handler` over the crate's arrival pump, the +//! wire-level ACL (`tunnel:forwarded` scope) and the per-connection +//! channel budget enforced inside alkcall's dispatch. +//! 3. **The arrival pump** — the `forwarded-tcpip` data leg: the +//! registered [`ForwardedHandler`]-closure correlates the +//! arrival's `listen` token through [`ListenRouter`] (the route was +//! registered at listen acceptance with the owning session's +//! handle + cap); a miss is **teardown-at-birth** (drop the fabric +//! half, never a half-open SSH channel). On a hit, per vanilla `-R` +//! convention: the door opens a `forwarded-tcpip` channel *toward +//! the client* echoing **the listen request's own args** — the +//! connected address as the wire `tcpip_forward` request carried it +//! verbatim (the OpenSSH client matches that exact string; for +//! `port = 0` mints the *actually bound* port from the fabric's +//! `bound` reply — the sshd parity recorded in +//! [`ListenRoute`]) — with the `TunnelForwardedParams.peer` arrival +//! mapped to the originator fields (`HostPort` → address/port; +//! `Path` → path + port 0 — the degenerate but typed mapping, +//! unreachable for a TCP-minted listen). The accepted-or-rejected +//! decision is the in-crate cap (the wave-2 gate's booked acceptor +//! vocabulary — the per-arrival admission; a client-side rejection +//! or a dead session releases the count, the established channel's +//! release rides [`crate::handler`]'s `channel_close` like every +//! established channel). Accept → the fabric `BiStream` pumped +//! against the SSH channel ([`alkcall::channels::pump::pump_bidi`]). +//! Unix-substrate arrivals (a `streamlocal_forward`-minted +//! route) open `forwarded-streamlocal@openssh.com` with the listen +//! path echoed (russh's server `Handle` carries the channel-open at +//! this pin, and the vanilla client's `-R` unix listener is the +//! consumer of that leg). +//! +//! **The inline listen dial (recorded decision):** the +//! spawn-and-return rule keeps *serving paths* out of the packet +//! loop — but the `tcpip_forward` global request's reply is bound to +//! the hook itself (russh answers the want-reply from the hook's +//! return + the `&mut u32` out-param; no handle surface responds to a +//! global request from a spawned task at this pin), so the listen +//! dial awaits inline. This cannot deadlock against the peer's +//! follow-up bytes the way an inline serving path can (the vanilla +//! client waits for the global reply before proceeding; the fabric's +//! answer rides the fabric leg's connections, not this session's +//! packet loop), and its stall is bounded: other channels of the +//! connection pause only for [`LISTEN_DIAL_TIMEOUT`], and the +//! fabric-dial failure answers a denied listen. All serving below the +//! reply (none — the listen returns) and all arrival paths stay +//! spawned. +//! +//! ## Bounded bookkeeping (AGENTS.md convention 9) +//! +//! - **The per-session listener map** (the session-handler's +//! `listeners` — packet-loop-only writer, like the `channels`/ +//! `ptys` stores) maps the minted token → the pending listen (wire +//! args, bound port, the [`alktunnels::consumer::TunnelListener`] +//! whose `close`/`Drop` perform the fabric teardown). Its bound: +//! the listen count shares the `Limits::max_concurrent_channels` +//! *ceiling* in a dedicated counter — a listen is at least one +//! future channel's potential, sessions minting unbounded listeners +//! would mint an unbounded fabric-channel map on the shared leg. +//! - **The route map** ([`ListenRouter`], shared with the registered +//! arrival pump across every session riding one fabric leg) is +//! bounded by construction: entries are minted exactly at accepted +//! listens (listener-cap gated), removed 1:1 at the cancel forms +//! and at the owning session's teardown (the handler's `Drop` +//! removes its tokens — a stale route to a dead handle only ever +//! costs one failed open that returns teardown-at-birth, never a +//! half-open SSH channel). +//! - **A cancelled / torn-down listen never half-serves**: the cancel +//! path removes the route before any teardown (a racing arrival +//! misses → teardown-at-birth), and `cancel_*` of an unknown +//! (address, bound-port) pair answers `false` — the vanilla +//! cancel contract (sshd parity: match by the raw wire string + +//! the bound port). +//! //! ## The runtime-Unix gate (OQ-11) //! //! `direct-streamlocal` compiles wherever `tunnel` compiles (universal @@ -38,7 +161,8 @@ //! (never a partial bridge), on-Unix the real bridge. The gate lives //! in the dispatch hook ([`crate::handler`]'s `direct-streamlocal` //! row), before the cap admission — a platform-refused open consumes -//! nothing. +//! nothing. The gate rides the `streamlocal_forward` listen row the +//! same way (`w3-tunnel-listen`). //! //! ## The identity ride (the visible-surface invariant's gate) //! @@ -51,53 +175,50 @@ //! open that proceeds then rides the channels connection into the //! fabric, whose own `AccessControl::check(peer_identity)` gates the //! mapped `channels/tunnel/direct` open (alkcall CF-005/CF-006) — -//! the composed-gate witness is the harness's identity rows. +//! the composed-gate witness is the harness's identity rows. The +//! `-R` leg composes the same gates on both of its directions: the +//! listen dial rides the fabric leg's identity (`tunnel:open` scope — +//! the assembly's dial-identity composition, unchanged), and the +//! arrivals are gated by the door-side registry's ACL +//! (`tunnel:forwarded` scope, per arrival) exactly when the assembly's +//! fabric leg carries it. //! -//! **`set_identity` mechanics (recorded — the task's decided point):** -//! the alkhub `hub_leg.rs` one-time-set precedent -//! (`channel0_conn.set_identity(identity)`) is a **serving-side** set — -//! the hub resolves the peer identity and sets it on its accept-side -//! channel-0 `Connection` inside the channel-0 install, before any -//! dispatch. The client-side surface -//! (`ChannelClient::take_call_connection()` → `CallConnection:: -//! connection()` → `Connection::set_identity`) is **rejected for this -//! door**, verified against alkcall 0.10.1: the take is a one-shot -//! single-slot detach ("taking it detaches the client's own calling -//! surface" — every later `call_open_op`, and with it every -//! `TunnelSession::open_direct`, errors "channel client closed"), and -//! the slot is the **shared assembly handle**: `SshAdapter` hands one -//! `Arc` to every mounted session, so the first session's -//! take would detach the open surface for every later session of the -//! whole serving direction — a strictly worse version of the relay -//! note's "the detach is wrong for hubs" (three claimants). The -//! identity that alkcall's protocol consults at the fabric's serving -//! dispatch is the channels connection's dial identity (CF-005: set -//! once at dial, propagated to channel 0 at `from_connection`) — -//! crossing sides is not an alkcall wire concept. The composition — -//! which channel identity the fabric leg dials with, against which the -//! fabric's ACL gates alkssh's mapped opens — is therefore **assembly -//! policy**: the assembly supplies both the injected -//! `IdentityProvider` (the SSH-peer resolution) and the fabric leg -//! (the handle and its dial identity); alkssh carries no deployment -//! policy, and it structurally refuses anonymous serving (the -//! `no-identity` guard) so no SSH-shaped exception to the -//! visible-surface invariant exists. The upstream ask that would let -//! alkssh do more (a non-detaching post-construction identity seam on -//! `ChannelClient`, per-session fabric legs) is booked for the -//! wave-3 gate. +//! **`set_identity` mechanics (recorded — `w3-tunnel-direct`'s decided +//! point, carried unchanged):** the alkhub `hub_leg.rs` one-time-set +//! precedent (`channel0_conn.set_identity(identity)`) is a +//! **serving-side** set — the identity the fabric leg dials with, and +//! against which the fabric's ACL gates alkssh's mapped opens, is the +//! assembly-supplied `ChannelClient`'s own dial identity (CF-005: set +//! once at dial, propagated to channel 0 at `from_connection`); the +//! door's `ChannelClient::take_call_connection` surface is rejected +//! for this door (the detach would strand every later session), and +//! the identity composition is assembly policy, structurally refused +//! anonymous by the `no-identity` guard. The door-side arrival surface +//! adds no new identity mechanics: alkssh registers the openable with +//! an install-time anonymous `AuthContext` (transport-truthful fields +//! are not alkssh's to know) and alkcall's per-call CF-005 overlay + +//! the assembly's `ServingConfig` composition decides what identity +//! the arriving forwarded call carries. +use std::collections::HashMap; use std::sync::Arc; +use std::time::Duration; use alkcall::channels::client::ChannelClient; +use alkcall::channels::operations::ChannelCore; use alkcall::channels::pump::pump_bidi; -use alktunnels::consumer::TunnelSession; -use alktunnels::params::{Substrate, SubstrateAddr}; +use alkcall::core::auth::AuthContext; +use alkcall::core::types::BiStream; +use alktunnels::consumer::{TunnelListener, TunnelSession}; +use alktunnels::params::{Substrate, SubstrateAddr, TunnelForwardedParams, TunnelParams}; +use alktunnels::producer::{ForwardedHandler, register_tunnel_forwarded_openable}; +use parking_lot::Mutex; use russh::Channel; -use russh::server::ChannelOpenHandle; +use russh::server::{ChannelOpenHandle, Handle}; use crate::channel_cap::ChannelCap; use crate::serve; -use crate::{dispatch, observe}; +use crate::{config, dispatch, observe}; /// The SSH data channel type: russh's server-side channel over its own /// message type (the store-and-take handoff and the serving rows use @@ -159,3 +280,356 @@ pub(crate) fn spawn_direct_row( row, channel, reply, target, substrate, client, cap, )); } + +// --------------------------------------------------------------------------- +// The `-R` listen flow (w3-tunnel-listen) +// --------------------------------------------------------------------------- + +/// The wall-clock bound on the family-ward listen dial (`TunnelListener:: +/// open`): the fabric's establisher is already bound by alkcall's own +/// establisher deadline (10 s, alkcall ADR-049 §2), so this matches it — +/// a fabric that never answers a listen dial cannot hold the packet +/// loop longer than a bounded dial. +pub(crate) const LISTEN_DIAL_TIMEOUT: Duration = Duration::from_secs(10); + +/// One route in the listen-arrival bookkeeping — registered at listen +/// acceptance, consulted per arrival. The echo fields are the vanilla +/// `-R` convention decoded (sshd parity, recorded on [`ListenRouter`]): +/// the connected address rides the wire `tcpip_forward` string +/// **verbatim** (the OpenSSH client sent that exact string and matches +/// the open's address field against it exactly), the port is the +/// **actually bound** one (the fabric's `bound` reply — the sshd +/// `get_local_port`-of-accepted-socket parity, which is what makes +/// `port = 0` mints round-trip), and the unix path rides the +/// `streamlocal_forward` request verbatim. +#[derive(Clone, Debug)] +pub(crate) struct ListenRoute { + /// The SSH args the arrivals' opens echo — keyed by the route's + /// substrate (`Tcp`/`Udp` → the vanilla `-R` trio's first two + /// fields; `Unix` → the `forwarded-streamlocal` path). + pub(crate) echo: RouteEcho, + pub(crate) substrate: Substrate, + /// The owning session's packet-loop handle — the arrival's SSH + /// channel opens through it; a session that died fails the open + /// (the arrival's teardown-at-birth path). + pub(crate) handle: Arc, + /// The owning session's in-crate concurrent-channel cap: the + /// per-arrival admission (the acceptor vocabulary) + the release + /// on the never-established paths. + pub(crate) cap: Arc, + pub(crate) cap_value: usize, +} + +/// The echoed SSH args of a route (see [`ListenRoute`]). +#[derive(Clone, Debug)] +pub(crate) enum RouteEcho { + Tcp { address: String, port: u32 }, + Unix { path: String }, +} + +/// The shared per-fabric-leg map of accepted listens +/// (`token → route`), the arrival pump's correlation registry and the +/// bounded bookkeeping's spine: entries are minted exactly at accepted +/// listens (the per-session listener cap gated), taken at the cancel +/// forms, and removed at the owning session's teardown (the handler's +/// `Drop`). Arrivals that find no route are the correlation-miss +/// class — teardown-at-birth, never a half-open SSH channel. +/// +/// The map is behind a `parking_lot::Mutex` because arrivals arrive on +/// alkcall serving tasks while the packet loop mints/removes — the +/// held region is a HashMap op, never an await. +pub(crate) struct ListenRouter { + routes: Mutex>, +} + +impl ListenRouter { + /// Minted-entry registration (the listen-acceptance step). + pub(crate) fn register(&self, token: u32, route: ListenRoute) { + self.routes.lock().insert(token, route); + } + + /// The per-arrival lookup — cloned out, never taken: one route + /// serves unboundedly many arrivals; only the cancel forms and + /// session teardown remove. + pub(crate) fn route_for(&self, token: u32) -> Option { + self.routes.lock().get(&token).cloned() + } + + /// The cancel-form removal (1:1 into the family listener's + /// teardown): removes first, so a racing arrival already misses — + /// teardown-at-birth, never a post-cancel half-serve. + pub(crate) fn cancel(&self, token: &u32) -> bool { + self.routes.lock().remove(token).is_some() + } + + /// The owning session's teardown (the handler's `Drop`): remove + /// the session's tokens; any listener objects left drop with the + /// handler (the fabric-side reap — see `PendingListen`). + pub(crate) fn remove_tokens(&self, tokens: &[u32]) { + let mut routes = self.routes.lock(); + for token in tokens { + routes.remove(token); + } + } +} + +impl Default for ListenRouter { + fn default() -> Self { + Self { + routes: Mutex::new(HashMap::new()), + } + } +} + +/// The door-side arrival surface's install step (called from +/// [`crate::SshAdapter::new`] when the fabric leg + seam are both +/// present): builds the router, registers alktunnels' forwarded +/// openable into the seam's shared registry over a `ChannelCore` +/// wrapping a clone of the held `ChannelClient`'s manager — the +/// alkcall shape whose wrapper performs the door-side arrival's +/// channel allocation, ACL (`tunnel:forwarded`, per call) and budget +/// (alkcall's secure default per-identity policy, `256`, plus the +/// `DEFAULT_MAX_CHANNELS` per-connection bound — alkssh composes no +/// deployment policy; an assembly wanting a different per-identity +/// cap shapes it through its own `ServingConfig`/`registry` +/// composition upstream) — and spawns alktunnels' forward-op handler +/// over the crate's arrival pump. The registration is one-per-op-id- +/// per-registry (alkcall's rule): a conflicting registration is a +/// failed construction (fail-closed — never a route to a handler the +/// door does not own). +/// +/// On success the router rides back into the serve-config (the +/// session handlers read it) — `Ok` with the router installed, `Err` +/// is the caller's constructor failure. +pub(crate) fn install_forwarded_surface(serve: &mut config::ServeConfig) -> Result<(), String> { + let Some(seam) = serve.tunnel_listen_seam.as_mut() else { + return Err("tunnel listen surface requested without a tunnel_listen_seam".to_string()); + }; + let Some(client) = serve.channel_client.as_ref() else { + return Err("tunnel listen seam configured without a fabric leg".to_string()); + }; + let router = Arc::new(ListenRouter::default()); + let handler: ForwardedHandler = { + let router = Arc::clone(&router); + Arc::new(move |params: TunnelForwardedParams, bidi: BiStream| { + tokio::spawn(pump_forwarded_arrival(Arc::clone(&router), params, bidi)) + }) + }; + let core = ChannelCore::new( + client.manager().clone(), + alkcall::channels::policy::default_policy(), + ); + register_tunnel_forwarded_openable( + &core, + &seam.registry, + AuthContext::anonymous(b"alk/tunnel"), + handler, + ) + .map_err(|failure| format!("tunnel/forwarded openable registration: {failure}"))?; + serve.listen_router = Some(Arc::clone(&router)); + Ok(()) +} + +/// The registered arrival pump — the `forwarded-tcpip` data leg +/// (mapping.md's row, "hand the channel duplex to the handler"): the +/// alkcall openable registration hands every fabric arrival here with +/// the parsed [`TunnelForwardedParams`] and the arrival's `BiStream`; +/// the flow is the recorded shape: +/// +/// 1. **Correlation** — the route for the arrival's minted token; +/// a miss is teardown-at-birth (drop the fabric half; the alkcall +/// wrapper's awaited join ends, the channel reaps). +/// 2. **The in-crate cap admission** (bounded resources at the accept +/// surface — the wave-2 gate's booked acceptor vocabulary): no +/// room → the fabric half drops, the arrival never opens an SSH +/// channel. The count releases here on every never-established +/// path; an established one releases at its close (the packet +/// loop's release site), the direct rows' disjoint-release shape. +/// 3. **The SSH open toward the client** — the echoed listen args +/// (`forwarded-tcpip` for the tcp/udp substrates with the originator +/// fields mapped from `TunnelForwardedParams.peer`; `forwarded- +/// streamlocal@openssh.com` with the echoed path for `Unix`) — per +/// the spawn-and-return constraint this runs on this spawned task; +/// a rejected/dead-open releases the count and drops the fabric +/// half (never a half-open SSH channel). +/// 4. **The pump** — the fabric `BiStream` ↔ the SSH channel duplex +/// (the route-1 bridge's primitive, reversed). +async fn pump_forwarded_arrival( + router: Arc, + params: TunnelForwardedParams, + bidi: BiStream, +) { + let row: &'static str = match params.substrate { + Substrate::Unix => "forwarded-streamlocal", + Substrate::Tcp | Substrate::Udp => "forwarded-tcpip", + }; + let Some(route) = router.route_for(params.listen) else { + observe::decline_arrival(row, "correlation-miss", params.listen); + drop(bidi); + return; + }; + if !route.cap.admit(route.cap_value) { + observe::decline_arrival(row, "cap-exceeded", params.listen); + drop(bidi); + return; + } + let opened = match route.substrate { + Substrate::Unix => match route.echo { + RouteEcho::Unix { path } => route.handle.channel_open_forwarded_streamlocal(path).await, + _ => { + route.cap.release(); + observe::decline_arrival(row, "correlation-miss", params.listen); + return; + } + }, + Substrate::Tcp | Substrate::Udp => match (route.echo, ¶ms.peer) { + ( + RouteEcho::Tcp { address, port }, + SubstrateAddr::HostPort { + host, + port: peer_port, + }, + ) => { + route + .handle + .channel_open_forwarded_tcpip( + address, + port, + host.clone(), + u32::from(*peer_port), + ) + .await + } + (RouteEcho::Tcp { address, port }, SubstrateAddr::Path { path }) => { + // The degenerate-but-typed mapping (a path-shaped peer on + // a tcp listen — unreachable from a well-behaved producer): + // the path rides the originator string, port 0. + route + .handle + .channel_open_forwarded_tcpip(address, port, path.clone(), 0) + .await + } + _ => { + route.cap.release(); + observe::decline_arrival(row, "correlation-miss", params.listen); + return; + } + }, + }; + match opened { + Ok(channel) => { + let channel_number = channel.id().number(); + observe::serving_path_started(row, channel_number, None); + let (ssh_read, ssh_write) = tokio::io::split(channel.into_stream()); + pump_bidi(bidi, ssh_read, ssh_write).await; + observe::serving_path_ended(row, channel_number, None); + } + Err(_) => { + route.cap.release(); + observe::decline_arrival(row, "accept-refused", params.listen); + } + } +} + +/// A minted listen — the door-side outcome of a `tcpip_forward` / +/// `streamlocal_forward` request's family-ward dial (the published +/// `TunnelListener::open`, bounded by [`LISTEN_DIAL_TIMEOUT`], with +/// the reply's `bound` + `listen` establisher fields required: a +/// producer that minted no correlation token or no bound report +/// cannot serve — [`Result`]'s `Err` declines the SSH request before +/// anything registers). +pub(crate) struct MintedListen { + pub(crate) listener: TunnelListener, + pub(crate) token: u32, + pub(crate) bound: SubstrateAddr, +} + +/// The listen dial (the mapping.md §Global-request rows' family-ward +/// step): `TunnelParams { resource: , substrate }` +/// — the wire-named bind endpoint as the resource, the recorded +/// door-side naming convention whose fabric-side interpretation is +/// the assembly's establisher — then the reply projection. Called +/// inline from the handler hook (see the recorded inline-decision in +/// the module doc). +pub(crate) async fn open_listen( + client: &Arc, + resource: String, + substrate: Substrate, +) -> Result { + let future = TunnelListener::open( + client, + TunnelParams { + resource, + substrate, + }, + ); + let listener = match tokio::time::timeout(LISTEN_DIAL_TIMEOUT, future).await { + Ok(Ok(listener)) => listener, + Ok(Err(error)) => return Err(format!("listen dial failed: {error:?}")), + Err(_) => return Err("listen dial timed out".to_string()), + }; + let Some(token) = listener.listen() else { + return Err("the fabric minted no listen token".to_string()); + }; + let Some(bound) = listener.bound().cloned() else { + return Err("the fabric reported no bound endpoint".to_string()); + }; + Ok(MintedListen { + listener, + token, + bound, + }) +} + +/// The bound port the listen request's out-param carries +/// (the `&mut u32` sshd parity): the fabric's bound report's port. +/// Unix listens have no port semantics — the tcp variants only. +pub(crate) fn bound_port(bound: &SubstrateAddr) -> Option { + match bound { + SubstrateAddr::HostPort { port, .. } => Some(u32::from(*port)), + _ => None, + } +} + +/// The family-ward resource string for a listen request — the wire +/// bind endpoint verbatim (`"localhost"` / `"0.0.0.0"` / explicit +/// host: the OpenSSH client has already normalized the no-address case +/// to `"localhost"` and `""` before this surface exists — sshd +/// `channel_rfwd_bind_host` parity), with the requested port for the +/// tcp substrates (`host:port`; port `0` stays `0` — the OS-picking is +/// the fabric's bind step, whose bound report feeds the out-param). +pub(crate) fn listen_resource(endpoint: &str, substrate: Substrate, port: Option) -> String { + match substrate { + Substrate::Unix => endpoint.to_owned(), + _ => format!("{}:{}", endpoint, port.unwrap_or(0)), + } +} + +/// The per-session pending listen (the packet-loop-only store, like +/// the `channels`/`ptys` stores): the minted token is the map's key; +/// the entry carries what the cancel forms match on (the wire endpoint +/// verbatim + the bound port — the sshd `c->path == host && +/// c->listening_port == port` parity) over the substrate (a unix +/// entry's port is 0-carrying and never matched — the unix cancel +/// forms match on the path alone) and the session-owned +/// [`TunnelListener`] whose close/Drop performs the family teardown. +pub(crate) struct PendingListen { + pub(crate) wire_address: String, + pub(crate) bound_port: u32, + pub(crate) substrate: Substrate, + pub(crate) listener: TunnelListener, +} + +/// The listen count's dedicated bound — the same mechanism as +/// [`ChannelCap`], instantiated per session next to it, against the +/// same `Limits::max_concurrent_channels` ceiling: listeners and +/// channels stay independently bounded while their mix stays +/// unconstrained (AGENTS.md convention 9 — no unbounded listener +/// map). Admits are packet-loop-only (the listen hooks are the only +/// mint site); releases fire at the cancel forms, the listen-dial +/// failures (admitted-but-never-minted — the route 1 bridge's +/// release-site shape), and nowhere else (a minted listener's release +/// is... never — listeners hold until cancel or session teardown, so +/// the session's count dies with the handler, per the channels +/// cap's same posture). +pub(crate) type ListenerCap = ChannelCap; diff --git a/tasks/implementation/w3-tunnel-listen.md b/tasks/implementation/w3-tunnel-listen.md index 1198130..b5a8e9d 100644 --- a/tasks/implementation/w3-tunnel-listen.md +++ b/tasks/implementation/w3-tunnel-listen.md @@ -1,7 +1,7 @@ --- id: w3-tunnel-listen name: Wave 3 — tunnel leg 2 (the -R listen flow + the forwarded-tcpip data leg) -status: pending +status: completed depends_on: [w3-tunnel-direct] scope: moderate risk: high @@ -87,24 +87,24 @@ target): ## Acceptance Criteria -- [ ] `tcpip_forward` / `streamlocal_forward` (+ cancel forms) carry +- [x] `tcpip_forward` / `streamlocal_forward` (+ cancel forms) carry the real mapped flow under `tunnel` (+ the runtime-Unix gate where it binds); decline classes keep their wire shapes -- [ ] The family-ward listen composition decided against alktunnels +- [x] The family-ward listen composition decided against alktunnels 0.4.1's actual surfaces and recorded; the door-side inbound-open mechanics closed (or a serve-config amendment landed + recorded; or stopped-and-booked for the gate — nothing silent) -- [ ] The `forwarded-tcpip` data leg serves per the handler shape +- [x] The `forwarded-tcpip` data leg serves per the handler shape (target-arg echo per pending listen, cap-gated acceptor, the pump, teardown-at-birth on correlation-miss) -- [ ] The per-session listener bookkeeping bounded; the bound-port +- [x] The per-session listener bookkeeping bounded; the bound-port out-param returns the actually-bound number -- [ ] The per-build decline-matrix rows green through the harness with +- [x] The per-build decline-matrix rows green through the harness with the client-side accept hook in place (the wave-2 gate's booked harness item discharged) -- [ ] The real-tool smoke posture decided + recorded; the wave-2 stub +- [x] The real-tool smoke posture decided + recorded; the wave-2 stub listen rows migrated -- [ ] `cargo test` (+ per-feature + `--all-features` rows), clippy ×2 +- [x] `cargo test` (+ per-feature + `--all-features` rows), clippy ×2 (`-D warnings`), fmt, `cargo doc --no-deps` pass ## References @@ -132,4 +132,128 @@ target): ## Notes -> To be filled at implementation time. \ No newline at end of file +**Landed.** The `-R` leg is served for real: the listen flow +(`tcpip_forward`/`streamlocal_forward` + cancel forms under `tunnel`) +and the `forwarded-tcpip`/`forwarded-streamlocal` data leg, verified +against the published alktunnels 0.4.1 / alkcall 0.10.1 surfaces +through the harness and (tooling present) a real OpenSSH `ssh -R` +smoke. 18 server-direction rows + 2 real-tool smokes green in +`tests/tunnel_mapping.rs`. + +- **The composition decisions (recorded in `src/tunnel_map.rs`'s + module doc + `mapping.md` §Global-request rows):** + - *Who dials `tunnel/forwarded`*: the fabric (not the door) — + alktunnels' bind-first accept dispatcher + (`make_bindfirst_accept_handler` + the `ForwardedOpener` closure) + dials the op on the channel-0 call plane toward the listen opener + per arrival; the door reply carries the door-allocated channel id + the producer adopts (alkcall's odd/even split keeps the two open + directions collision-free: the door allocates odd as the connect + side, the fabric even as the accept side). + - *The door-side inbound-open mechanics*: closed over the published + surfaces — the `Arc` alone cannot carry them (a + client serves inbound ops on its channel-0 loop only when built + with `from_connection_with_serving` against an `OperationRegistry` + the assembly owns; no post-construction way in) — so the + serve-config's tunnel handle gained `tunnel_listen_seam` + (`TunnelListenSeam`, the wave-3 seam-config's second + course-correction): the same `Arc` the assembly + wired into the leg's `ServingConfig`. alkssh registers the + forwarded openable itself (`register_tunnel_forwarded_openable` + over a `ChannelCore` wrapping a clone of the held client's + manager, alkcall's secure default per-identity policy) at + `SshAdapter::new` — a registration failure is a failed + construction (`AdapterError::ListenRegistration`), never a + half-configured adapter. A scratch probe (published crates only, + no `alktunnels/local`) proved the topology end to end before any + of this landed: door-side registration → fabric bind-first + producer → arrival → `ForwardedHandler` (params + `BiStream`) → + pump round trip, with the door's own pending calls still resolving + while serving is enabled. + - *The resource-naming convention (the flow's one open surface)*: + the vanilla wire bind endpoint rides family-ward as the resource + string verbatim (`"localhost:9200"` / `"0.0.0.0:9200"` / the + socket path) — the OpenSSH client normalizes the no-address case + to `"localhost"` before sending (sshd `channel_rfwd_bind_host` + parity), the door passes through, and the fabric's + establisher/bind closure (assembly policy — the dynamic-address + producer opt-in the door's convention fronts) treats it as the + bind endpoint. ADR-001's "resource is not an address" discipline + stays a producer-side naming rule; this is a recorded door-to- + fabric convention, not a wire change. +- **The `forwarded-tcpip` data leg** (the row the POC only exercised + as refusal): per arrival → token correlation through the shared + `ListenRouter` (one route serves unboundedly many arrivals; + removed only at the cancel forms/session teardown) → the in-crate + per-session channel cap admits at the accept surface — the acceptor + vocabulary's first shape, the wave-2 gate's booked item now + discharged (never-established arrival paths release inside the + arrival pump; established closes release through the packet loop's + `channel_close` as ever) → `Handle::channel_open_forwarded_tcpip` + echoing the listen request's own args: the wire address string + verbatim + the actually-bound port (sshd `c->path` + + `get_local_port` parity — the OpenSSH client matches that exact + string + port and dials its own target), the + `TunnelForwardedParams.peer` arrival mapped to the originator + fields (HostPort → address/port; Path → path + 0, + degenerate-but-typed) → `pump_bidi(ssh, fabric)` (half-close + semantics: EOF-shutdown in the completed direction; the vanilla + client closes its half when its target socket dies — the row's + stand-in — and the cap release follows through the normal close). + Correlation-miss → teardown-at-birth (drop the fabric half; the + open op's channel reaps — never a half-open SSH channel). Unix + arrivals (`streamlocal_forward` mints) open + `forwarded-streamlocal@openssh.com` with the path echoed (russh's + server `Handle` carries that channel-open at this pin; the client + hook exists) — the "no russh hook" concern in the wave-2 stub notes + was about the door's *inbound* handlers of that type, not the + server-initiated direction. +- **Bounded bookkeeping** (convention 9): the per-session listener + map is packet-loop-only (like the `channels`/`ptys` stores); its + bound is a dedicated counter against the same + `max_concurrent_channels` ceiling — a listen is at least one future + channel's potential; the listener/channels mix stays unconstrained + while both stay bounded (no unbound listener map). The route map is + bounded by construction: entries minted exactly at accepted listens, + removed 1:1 at the cancels and at the owning session's teardown (the + handler's `Drop`); a route to a dead handle only ever costs one + failed open → teardown-at-birth. `cancel` of an unknown + (endpoint, bound-port) pair answers `false` (sshd + cancel-of-unknown parity). Admissions have two sites and releases + three since this task — the cap's recorded single-writer reasoning + updated in `channel_cap.rs` (mutex-guarded, saturating, both). +- **The real-tool smoke**: implemented and green here (tooling + check-gated like the -L smoke's posture): a real `ssh -R + 127.0.0.1:PORT:HTTP` (OpenSSH client) through the TCP-mounted + adapter → the fabric double with a real bind → arrival → the + forwarded-tcpip open → the client dials its own endpoint → curled + body parity. Never silently dropped. +- **The decline matrix** per build, all through the harness with the + client-side accept hook in place (the harness handler's + `server_channel_open_forwarded_tcpip` / + `server_channel_open_forwarded_streamlocal`, acceptor-driven — no + acceptor → clean decline, the family's client-side contract): the + wave-2 stub listen rows migrated (`tcpip_forward_decline_emits_*` + split per build: `unfeatured` vs `seam-unconfigured`); the + featured + configured rows serve/arrive/cancel (incl. the port-0 + mint resolving into the `&mut u32` out-param and the echo); the + unset listen seam declines while the egress bridge still routes + (the row's control leg); the listener cap and the per-arrival + channel cap rows (refusal + release proof); the composed-identity + rows (the fabric-scope gate on the listen dial, the door-side ACL + gate on arrivals); the OQ-11 off-Unix rows; the correlation-miss + row; the unvanilla inbound `forwarded-tcpip` open declined + (event-level contract — russh's client has no arbitrary-type open + to drive that row on the wire). +- **`tests/common` grew**: the forwarding acceptor + (`ForwardedAcceptor` + `ForwardedOpen` + `drive_with_acceptor`) + and a minimal hand-rolled tracing subscriber (`trace_global` — the + family posture: no `tracing-subscriber` dependency; opted in per + row) as the leg-2 rows' failure-triage vehicle. +- **Verification (all green)**: `cargo test` across default / + `tunnel` / `git` / `session` / `sftp` / `tunnel,git` / + `tunnel,session` / `git,session` / `tunnel,sftp` / `git,sftp` / + `--all-features`; `cargo clippy --all-targets -- -D warnings` ×2 + (default + `--all-features`) plus the combo builds (`tunnel,git` / + `tunnel,session` / `tunnel,sftp`); `cargo fmt --check`; `cargo doc + --no-deps` (0 warnings); `taskgraph validate` (26 tasks). diff --git a/tests/common/harness.rs b/tests/common/harness.rs index c25fd88..482a57c 100644 --- a/tests/common/harness.rs +++ b/tests/common/harness.rs @@ -76,7 +76,7 @@ use std::sync::Arc; use alkcall::core::auth::IdentityProvider; use russh::client::{AuthResult, Handle, Msg}; use russh::keys::PrivateKeyWithHashAlg; -use russh::{Channel, ChannelMsg, Disconnect, Error, MethodSet}; +use russh::{Channel, ChannelMsg, ChannelOpenFailure, Disconnect, Error, MethodSet}; use alkssh::{ AdapterError, HostKeyPolicy, Limits, PresentedHostKey, ServeConfig, SshAdapter, fingerprint, @@ -112,20 +112,86 @@ pub fn client_config(limits: Limits) -> Arc { }) } +/// One server-initiated forwarded open the rows observe at the +/// harness client (`w3-tunnel-listen`'s accept-hook harness row — the +/// `-R` data leg arrives here). The `Channel` only carries data after +/// the row accepts (the harness hook answers the open itself). +#[derive(Debug)] +pub enum ForwardedOpen { + /// The `forwarded-tcpip` row — the echo args + originator the door + /// opened with, verbatim as the wire carried them. + Tcpip { + connected_address: String, + connected_port: u32, + originator_address: String, + originator_port: u32, + channel: Channel, + }, + /// The `forwarded-streamlocal@openssh.com` row (the unix listen + /// leg's arrival shape). + Streamlocal { + socket_path: String, + channel: Channel, + }, +} + +/// The harness client's acceptor — what a vanilla client's matching +/// `-R` listener stands-in as: when installed, server-initiated +/// forwarded opens are accepted and handed to the row through the +/// channel the acceptor was built with; **no acceptor installed → the +/// open declines cleanly** (`AdministrativelyProhibited` — the +/// family's client-side contract, ADR-003 symmetry: a vanillla stand- +/// in with no matching forward never accept-and-drops). +pub struct ForwardedAcceptor { + opens: tokio::sync::mpsc::UnboundedSender, +} + +impl ForwardedAcceptor { + /// Builds the acceptor over its event channel; the receiving half + /// rides the rows (`drain_forwarded`). + pub fn new() -> (Self, tokio::sync::mpsc::UnboundedReceiver) { + let (opens, receiver) = tokio::sync::mpsc::unbounded_channel(); + (Self { opens }, receiver) + } + + fn hand_off(&self, open: ForwardedOpen) -> Result<(), Error> { + self.opens + .send(open) + .map_err(|_| Error::ChannelOpenFailure(ChannelOpenFailure::UnknownChannelType)) + } +} + /// The harness's russh client handler — the [`MemHostKeyPolicy`] /// decision double bridged into russh's `check_server_key` hook, with /// the decision input exactly as the wave-4 dial wiring will shape it -/// ([`PresentedHostKey`] over the crate's fingerprint path). +/// ([`PresentedHostKey`] over the crate's fingerprint path), plus the +/// `w3-tunnel-listen` accept hook: the server-initiated forwarded +/// opens the contract rows observe (the ADR-003 §4 client-side +/// acceptor symmetry). pub struct HarnessClientHandler { policy: Arc, + acceptor: Option>, } impl HarnessClientHandler { /// Builds the handler over a policy double — `with_default(true)` /// for the routine permissive posture, `with_default(false)` (or a - /// pinned rule) for the fail-closed rows. + /// pinned rule) for the fail-closed rows — with the forwarded + /// acceptor absent (opens decline cleanly; install one with + /// [`HarnessClientHandler::with_forwarded_acceptor`]). pub fn new(policy: Arc) -> Self { - Self { policy } + Self { + policy, + acceptor: None, + } + } + + /// Installs the acceptor (the matching-`-R`-listener stand-in). + pub fn with_forwarded_acceptor(self, acceptor: Arc) -> Self { + Self { + policy: self.policy, + acceptor: Some(acceptor), + } } } @@ -147,6 +213,67 @@ impl russh::client::Handler for HarnessClientHandler { }; Ok(self.policy.decide(&presented).await) } + + /// The `forwarded-tcpip` accept hook: accept-into-the-acceptor + /// (the row observes the echoed args + originator verbatim and + /// pumps the `Channel`), or the clean decline when no acceptor is + /// installed. + async fn server_channel_open_forwarded_tcpip( + &mut self, + channel: Channel, + connected_address: &str, + connected_port: u32, + originator_address: &str, + originator_port: u32, + reply: russh::client::ChannelOpenHandle, + _session: &mut russh::client::Session, + ) -> Result<(), Self::Error> { + match self.acceptor.as_ref() { + Some(acceptor) => { + let open = ForwardedOpen::Tcpip { + connected_address: connected_address.to_owned(), + connected_port, + originator_address: originator_address.to_owned(), + originator_port, + channel, + }; + acceptor.hand_off(open)?; + reply.accept().await; + } + None => { + reply + .reject(ChannelOpenFailure::AdministrativelyProhibited) + .await; + } + } + Ok(()) + } + + /// The `forwarded-streamlocal` accept hook: the unix leg's shape. + async fn server_channel_open_forwarded_streamlocal( + &mut self, + channel: Channel, + socket_path: &str, + reply: russh::client::ChannelOpenHandle, + _session: &mut russh::client::Session, + ) -> Result<(), Self::Error> { + match self.acceptor.as_ref() { + Some(acceptor) => { + let open = ForwardedOpen::Streamlocal { + socket_path: socket_path.to_owned(), + channel, + }; + acceptor.hand_off(open)?; + reply.accept().await; + } + None => { + reply + .reject(ChannelOpenFailure::AdministrativelyProhibited) + .await; + } + } + Ok(()) + } } /// Assembles the server side from the wave-1 doubles — a @@ -218,16 +345,28 @@ pub async fn drive( adapter: SshAdapter, host_key_policy: Arc, limits: Limits, +) -> Result { + drive_with_acceptor(adapter, host_key_policy, limits, None).await +} + +/// [`drive`] with the client-side forwarded acceptor installed — the +/// `w3-tunnel-listen` rows' vehicle (the harness handler's accept +/// hook consumes the installed acceptor; `None` = the clean-decline +/// posture [`drive`] carries). +pub async fn drive_with_acceptor( + adapter: SshAdapter, + host_key_policy: Arc, + limits: Limits, + acceptor: Option>, ) -> Result { let (client_io, server_io) = tokio::io::duplex(DUPLEX_CAPACITY); let server_task = tokio::spawn(async move { adapter.handle(server_io).await }); - let client = russh::client::connect_stream( - client_config(limits), - client_io, - HarnessClientHandler::new(Arc::clone(&host_key_policy)), - ) - .await?; + let mut handler = HarnessClientHandler::new(Arc::clone(&host_key_policy)); + if let Some(acceptor) = acceptor { + handler = handler.with_forwarded_acceptor(acceptor); + } + let client = russh::client::connect_stream(client_config(limits), client_io, handler).await?; Ok(LoopbackSession { client, @@ -353,7 +492,9 @@ pub async fn request_agent_forward(channel: &Channel) -> Result<(), Error> /// Requests a `tcpip-forward` listen (`-R` ingress) — the /// success/failure shape is the result verbatim -/// ([`Error::RequestDenied`] on a declined listen). +/// ([`Error::RequestDenied`] on a declined listen; a `port = 0` +/// request's actually-bound number rides the success value — sshd +/// parity). pub async fn tcpip_forward( client: &Handle, address: &str, @@ -362,6 +503,40 @@ pub async fn tcpip_forward( client.tcpip_forward(address, port).await } +/// Requests a `cancel-tcpip-forward`. +pub async fn cancel_tcpip_forward( + client: &Handle, + address: &str, + port: u32, +) -> Result<(), Error> { + client.cancel_tcpip_forward(address, port).await +} + +/// Requests a `streamlocal-forward` listen (the unix `-R` ingress). +pub async fn streamlocal_forward( + client: &Handle, + socket_path: &str, +) -> Result<(), Error> { + client.streamlocal_forward(socket_path).await +} + +/// Requests a `cancel-streamlocal-forward`. +pub async fn cancel_streamlocal_forward( + client: &Handle, + socket_path: &str, +) -> Result<(), Error> { + client.cancel_streamlocal_forward(socket_path).await +} + +/// Drains the acceptor's event channel — the row's observation shape +/// (`None` once the acceptor's sender went, i.e. after the session's +/// end). +pub async fn next_forwarded_open( + receiver: &mut tokio::sync::mpsc::UnboundedReceiver, +) -> Option { + receiver.recv().await +} + /// Observes the next channel message — `None` after the channel has /// closed (russh's EOF-of-conversation shape). pub async fn next_message(channel: &mut Channel) -> Option { diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 0ff2dc1..5c94ed3 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -425,3 +425,52 @@ impl HostKeyPolicy for MemHostKeyPolicy { verdict } } + +/// The leg-2 rows' tracing visibility (`w3-tunnel-listen`): a minimal +/// hand-rolled global [`tracing::Subscriber`] printing every event's +/// field pairs to stderr — the same hand-rolled convention the +/// crate-side vocabulary rows use (no `tracing-subscriber` +/// dependency; events fire from spawned tasks on both sides of the +/// fabric, so the global subscriber is the only capture shape that +/// works). Rows opt in: `trace_global()` once per row. +pub fn trace_global() { + use std::sync::atomic::{AtomicU64, Ordering}; + struct TraceSubscriber(AtomicU64); + struct FieldPrinter(u64); + + impl tracing::field::Visit for FieldPrinter { + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + eprintln!("trace[{}]: {} = {value:?}", self.0, field.name()); + } + + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + eprintln!("trace[{}]: {} = {value:?}", self.0, field.name()); + } + } + + impl tracing::Subscriber for TraceSubscriber { + fn enabled(&self, _metadata: &tracing::Metadata<'_>) -> bool { + true + } + + fn new_span(&self, _span: &tracing::span::Attributes<'_>) -> tracing::Id { + tracing::Id::from_u64(1) + } + + fn record(&self, _span: &tracing::Id, _values: &tracing::span::Record<'_>) {} + + fn record_follows_from(&self, _span: &tracing::Id, _follows: &tracing::Id) {} + + fn event(&self, event: &tracing::Event<'_>) { + event.record(&mut FieldPrinter(self.0.fetch_add(1, Ordering::Relaxed))); + } + + fn enter(&self, _span: &tracing::Id) {} + fn exit(&self, _span: &tracing::Id) {} + fn clone_span(&self, _id: &tracing::Id) -> tracing::Id { + tracing::Id::from_u64(1) + } + } + + let _ = tracing::subscriber::set_global_default(TraceSubscriber(AtomicU64::new(0))); +} diff --git a/tests/observe.rs b/tests/observe.rs index 35eb757..bd7241c 100644 --- a/tests/observe.rs +++ b/tests/observe.rs @@ -261,23 +261,51 @@ async fn x11_open_decline_emits_the_never_feature_event() { } /// - *Contract stamp:* mapping.md §Global-request rows (the `-R` -/// listen form declines — the event carries no channel id: global -/// requests have none). +/// listen form declines this build's flow — the event carries no +/// channel id: global requests have none). The reason is the +/// build's gate: `unfeatured` where the feature is off, +/// `seam-unconfigured` where the tunnel feature is on but no +/// fabric leg/listen seam is configured (the unset-declines +/// posture — a bare `authenticated()` session configures neither). +#[cfg(not(feature = "tunnel"))] #[tokio::test] async fn tcpip_forward_decline_emits_the_global_request_event() { let session = authenticated("forward-row").await; tcpip_forward(&session.client, "127.0.0.1", 8080) .await .expect_err("the listen request declines"); - let event = await_event(&[ - ("row", "tcpip_forward"), - ("decline_class", "request"), - ("reason", "unimplemented"), - ]) - .await; assert_decline( - &event, - &[("row", "tcpip_forward"), ("reason", "unimplemented")], + &await_event(&[ + ("row", "tcpip_forward"), + ("decline_class", "request"), + ("reason", "unfeatured"), + ]) + .await, + &[("row", "tcpip_forward"), ("reason", "unfeatured")], + ); + session.unmount().await; +} + +/// - *Contract stamp:* mapping.md §Global-request rows + the +/// decline-matrix's config-driven notes (the `w3-tunnel-listen` +/// migrated stub row): a featured build whose fabric leg/listen +/// seam is unconfigured declines `seam-unconfigured` — the feature +/// alone configures nothing. +#[cfg(feature = "tunnel")] +#[tokio::test] +async fn tcpip_forward_decline_emits_the_seam_unconfigured_event() { + let session = authenticated("forward-row").await; + tcpip_forward(&session.client, "127.0.0.1", 8080) + .await + .expect_err("the unconfigured listen declines"); + assert_decline( + &await_event(&[ + ("row", "tcpip_forward"), + ("decline_class", "request"), + ("reason", "seam-unconfigured"), + ]) + .await, + &[("row", "tcpip_forward"), ("reason", "seam-unconfigured")], ); session.unmount().await; } diff --git a/tests/tunnel_mapping.rs b/tests/tunnel_mapping.rs index 3293a14..6ffb4af 100644 --- a/tests/tunnel_mapping.rs +++ b/tests/tunnel_mapping.rs @@ -42,31 +42,36 @@ mod common; mod serving { use std::collections::HashMap; use std::sync::Arc; - use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::atomic::{AtomicU32, AtomicUsize, Ordering}; use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero}; - use alkcall::channels::client::ChannelClient; - use alkcall::channels::operations::{ChannelCore, ChannelOperations}; + use alkcall::channels::client::{ChannelClient, ServingConfig}; + use alkcall::channels::operations::{ChannelCore, ChannelOperations, Establishment}; use alkcall::core::auth::{AuthContext, Identity, NoopIdentityProvider}; use alkcall::core::types::{Connection as CoreConnection, ProtocolHandler}; use alkcall::protocol::connection::{CallConnection, split_single_stream}; use alkcall::protocol::dispatch::Dispatcher; use alkcall::registry::registration::OperationRegistry; - use alktunnels::params::{Substrate, TUNNEL_DIRECT_SCOPE}; + use alktunnels::params::{ + Substrate, SubstrateAddr, TUNNEL_DIRECT_SCOPE, TUNNEL_FORWARDED_SCOPE, TUNNEL_OPEN_SCOPE, + }; use alktunnels::producer::{ - DialFn, IdentityWitness, TargetHandle, TunnelEstablishError, + BindFirstAcceptLoop, BindFirstPlan, BindFn, BoundListener, DialFn, ForwardedOpener, + IdentityWitness, TargetHandle, TunnelEstablishError, make_bindfirst_accept_handler, register_tunnel_direct_openable, }; use russh::client::Msg; use russh::{Channel, ChannelMsg, ChannelOpenFailure, Error}; + use std::time::Duration; use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::sync::mpsc; use super::common::harness::{ - LoopbackSession, drive, open_direct_streamlocal, open_direct_tcpip, - serve_config as harness_serve_config, + ForwardedOpen, LoopbackSession, cancel_tcpip_forward, drive, open_direct_streamlocal, + open_direct_tcpip, serve_config as harness_serve_config, tcpip_forward, }; use super::common::{MemHostKeyPolicy, MemIdentityProvider, fixture, identity}; - use alkssh::{Limits, SshAdapter}; + use alkssh::{Limits, SshAdapter, TunnelListenSeam}; type TargetMap = Arc>>; @@ -150,6 +155,394 @@ mod serving { ) } + /// The leg-2 fabric double (`w3-tunnel-listen`'s recorded + /// composition): the leg-1 wire fabric extended with the + /// **bind-first listen half** — the assembly-composed shape the + /// vanilla `-R` listen flow rides, built exactly on the published + /// alktunnels pieces: a `bindfirst_establisher`-shaped + /// establisher, `make_bindfirst_accept_handler`, and the + /// `ForwardedOpener` dialing the forwarded op back at the door. + /// + /// The **door-side serving surface**: the `Arc` + /// constructed with `from_connection_with_serving` against a + /// shared registry the row hands alkssh through the + /// `TunnelListenSeam` (one `Arc`, both sides). + /// + /// The double's bind seam is the **dynamic-address producer** the + /// door's naming convention front: the establisher treats the + /// wire-mapped resource string as the bind endpoint directly + /// (`resource = "127.0.0.1:9200"` binds `127.0.0.1:9200`) and mints + /// the correlation token per the ADR-008 discipline. Who may bind + /// what is the fabric's policy — the double binds what the room + /// asks; a real fabric's ACL is the same seam. `localhost` + /// renders to `127.0.0.1` so the real-tool smoke's curl is + /// deterministic about the family. + pub(super) struct FabricDouble { + pub(super) client: Arc, + pub(super) registry: Arc, + opener: Arc>, + /// The establisher's mint stream (the rows' real-token vehicle + /// — the ACL row needs the genuinely minted token its listen's + /// fabric established; the SSH layer never carries the token). + minted: tokio::sync::Mutex>, + } + + impl FabricDouble { + /// The fabric's `ForwardedOpener` (the call plane), exported + /// for the rows' direct drives (an arrival the fabric would + /// make, dialed by the row itself — the unknown-token and + /// composed-gate rows' vehicle). + pub(super) async fn forwarded_opener(&self) -> ForwardedOpener { + for _ in 0..600 { + if let Some(opener) = self.opener.get() { + return Arc::clone(opener); + } + tokio::task::yield_now().await; + } + panic!("the fabric's forwarded opener never installed"); + } + + /// Drives the fabric's forwarded-open closure directly: + /// `Ok` carries the adopted fabric-side halves (the door + /// allocated the channel — the correlation-miss row's + /// teardown-at-birth is then observable as an EOF on the + /// `recv` half), `Err` is the door-side refusal (the + /// composed-gate row's ACL answer — the arrival never became + /// anything). + /// The next minted correlation token (one per accepted listen, + /// in mint order — the rows' real-token vehicle). + pub(super) async fn minted_token(&self) -> u32 { + self.minted + .lock() + .await + .recv() + .await + .expect("the fabric's establisher minted a token in time") + } + + pub(super) async fn open_forwarded_directly( + &self, + params: alktunnels::params::TunnelForwardedParams, + ) -> Result< + ( + alkcall::channels::reassembly::MpscSendStream, + alkcall::channels::reassembly::MpscRecvStream, + ), + TunnelEstablishError, + > { + let opener = self.forwarded_opener().await; + opener(params).await + } + } + + /// The dynamic-bind producer's establisher (the double's shape of + /// the `bindfirst_establisher` recipe): parse + /// `TunnelParams {resource, substrate}`, bind via the injected + /// [`BindFn`] directly on the resource string, mint the token per + /// the ADR-008 discipline (non-zero-seeded counter, never 0), and + /// contribute the `bound` + `listen` reply fields (ADR-049 + /// amendment 3). + fn dynamic_bindfirst_establisher( + bind: BindFn, + minted: Arc>, + ) -> alkcall::channels::operations::OpenEstablisher { + let counter = Arc::new(AtomicU32::new(1)); + Arc::new(move |input: serde_json::Value, _auth| { + let bind = Arc::clone(&bind); + let minted = Arc::clone(&minted); + let counter = Arc::clone(&counter); + Box::pin(async move { + let params: alktunnels::params::TunnelParams = + serde_json::from_value(input.clone()).map_err(|e| { + TunnelEstablishError::HandlerError(format!("invalid params: {e}")) + })?; + let token = counter.fetch_add(1, Ordering::Relaxed); + let token = if token == 0 { 1 } else { token }; + let _ = minted.send(token); + let BoundListener { listener, bound } = + bind(params.substrate, ¶ms.resource).await?; + let plan = BindFirstPlan { + listener, + token, + substrate: params.substrate, + }; + Ok(Establishment::new(Arc::new(plan)) + .with_reply_field( + "bound", + serde_json::to_value(&bound) + .map_err(|e| TunnelEstablishError::HandlerError(format!("{e}")))?, + ) + .with_reply_field("listen", serde_json::json!(token))) + }) + }) + } + + /// The double's real-socket bind seam: real TCP/unix binds of the + /// wire-named endpoint (the platform boundary the + /// `tunnels/local` feature serves in a real deployment — + /// hand-rolled here, never a sibling-checkout path). Unix binds + /// unlink a stale socket file first (`bind_unix`'s caller + /// posture). + fn real_bind() -> BindFn { + Arc::new(move |substrate, endpoint: &str| { + let endpoint = endpoint.to_owned(); + Box::pin(async move { + match substrate { + alktunnels::params::Substrate::Unix => { + let _ = std::fs::remove_file(&endpoint); + let listener = tokio::net::UnixListener::bind(&endpoint).map_err(|e| { + TunnelEstablishError::ResourceShortage(format!( + "bind `{endpoint}`: {e}" + )) + })?; + Ok(BoundListener { + listener: Arc::new(listener), + bound: SubstrateAddr::Path { + path: endpoint.clone(), + }, + }) + } + _ => { + let hostport = endpoint.replace("localhost:", "127.0.0.1:"); + let listener = + tokio::net::TcpListener::bind(&hostport) + .await + .map_err(|e| { + TunnelEstablishError::ResourceShortage(format!( + "bind `{hostport}`: {e}" + )) + })?; + let addr = listener.local_addr().expect("bound tcp addr"); + Ok(BoundListener { + listener: Arc::new(listener), + bound: SubstrateAddr::HostPort { + host: addr.ip().to_string(), + port: addr.port(), + }, + }) + } + } + }) + }) + } + + /// The double's accept loop: real TCP/unix accept of ONE pair per + /// invocation (the dispatcher's per-accept contract), downcasting + /// the plan's opaque listener to the bind seam's own wrappers. The + /// unix leg's `peer` is fabricated (an unnamed peer cannot ride + /// the forwarded op's required `peer` param — the shipped + /// accept-pair posture). + fn real_accept_loop() -> BindFirstAcceptLoop { + Arc::new(move |listener: Arc| { + Box::pin(async move { + if let Some(tcp) = listener.downcast_ref::() { + let (stream, peer) = tcp + .accept() + .await + .map_err(|e| TunnelEstablishError::DialFailed(format!("accept: {e}")))?; + let (read, write) = stream.into_split(); + return Ok(( + SubstrateAddr::HostPort { + host: peer.ip().to_string(), + port: peer.port(), + }, + TargetHandle { + read: Box::new(read), + write: Box::new(write), + }, + )); + } + if let Some(unix) = listener.downcast_ref::() { + let (stream, _peer) = unix + .accept() + .await + .map_err(|e| TunnelEstablishError::DialFailed(format!("accept: {e}")))?; + let (read, write) = stream.into_split(); + return Ok(( + SubstrateAddr::Path { + path: "/fabric-arrival-peer.sock".to_owned(), + }, + TargetHandle { + read: Box::new(read), + write: Box::new(write), + }, + )); + } + Err(TunnelEstablishError::HandlerError( + "the plan's listener is not one of the double's binds".into(), + )) + }) + }) + } + + /// Wires the leg-2 fabric double (leg 1 unchanged underneath — the + /// dial seam + direct openable stay exactly `wire_fabric`'s + /// composition, against a leg-1 fail dial the -R rows never + /// exercise). `door_scoped_caller` composes the fabric-side + /// serving identity (what gates the door's listen op's ACL there); + /// `arrival_scoped` composes the door-side `ServingConfig.identity` + /// with or without the `tunnel:forwarded` scope (the composed-gate + /// row's negative leg). + pub(super) async fn wire_fabric_with_listen( + door_scoped_caller: bool, + arrival_scoped: bool, + ) -> FabricDouble { + // The two sides' registries are independent (the fabric's own + // dispatch registry; and the DOOR-side shared one the row + // hands alkssh's seam — the registration into the fabric's + // registry would be the silent-fork shape). The earlier draft + // wired one Arc to both — pinned wrong here for exactly one + // reason: the door's `tunnel/forwarded` openable must live in + // the DOOR's dispatch surface alone. + let door_registry = Arc::new(OperationRegistry::new()); + let registry = Arc::new(OperationRegistry::new()); + let registry_for_hook = Arc::clone(®istry); + let door_caller = if door_scoped_caller { + identity("door-caller", &[TUNNEL_OPEN_SCOPE]) + } else { + identity("door-caller-no-scope", &["ssh"]) + }; + let arrival_identity = if arrival_scoped { + identity("fabric-arrival", &[TUNNEL_FORWARDED_SCOPE]) + } else { + identity("fabric-arrival-plain", &["ssh"]) + }; + let dial: DialFn = Arc::new( + move |_substrate: alktunnels::params::Substrate, target: &str| { + let target = target.to_owned(); + Box::pin(async move { + Err::(TunnelEstablishError::DialFailed( + format!("the leg-1 dial double is not part of the -R rows: {target}"), + )) + }) + }, + ); + let opener_cell = Arc::new(tokio::sync::OnceCell::new()); + let opener_for_hook = Arc::clone(&opener_cell); + let (minted_sender, minted_receiver) = mpsc::unbounded_channel::(); + let minted_tx: Arc> = Arc::new(minted_sender); + let minted_for_hook = minted_tx.clone(); + + let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, _auth| { + let registry = Arc::clone(®istry_for_hook); + let door_caller = door_caller.clone(); + let minted_tx = minted_for_hook.clone(); + let dial = Arc::clone(&dial); + let opener_cell = Arc::clone(&opener_for_hook); + tokio::spawn(async move { + let Ok(channel0_bidi) = channel0_conn.accept_bi().await else { + return; + }; + let (writer, reader) = split_single_stream(channel0_bidi); + let _ = channel0_conn.set_identity(door_caller); + let call_connection = Arc::new(CallConnection::new_single_stream( + channel0_conn, + Arc::clone(&writer), + )); + + // The fabric's ForwardedOpener (the call plane): + // dial the forwarded op at the door, adopt the + // door-allocated id on the fabric's manager, and + // return the adopted channel's halves + // (make_bindfirst_accept_handler composes the + // pump). Exported for the rows' direct drives. + let open_forwarded: ForwardedOpener = { + let call = Arc::clone(&call_connection); + let manager = manager.clone(); + Arc::new(move |params: alktunnels::params::TunnelForwardedParams| { + let call = Arc::clone(&call); + let manager = manager.clone(); + Box::pin(async move { + let payload = serde_json::to_value(¶ms) + .map_err(|e| TunnelEstablishError::HandlerError(format!("{e}")))?; + let response = call + .call(alktunnels::params::OP_TUNNEL_FORWARDED, payload) + .await; + let out = response.result.map_err(|e| { + TunnelEstablishError::HandlerError(format!("forwarded open: {e:?}")) + })?; + let id = + out.get("channel_id") + .and_then(|v| v.as_u64()) + .ok_or_else(|| { + TunnelEstablishError::HandlerError("no channel_id".into()) + })? as u32; + manager + .adopt_channel(id, alktunnels::params::TUNNEL_ALPN, None) + .await + .map_err(|e| { + TunnelEstablishError::HandlerError(format!("adopt: {e:?}")) + }) + }) + }) + }; + let _ = opener_cell.set(Arc::clone(&open_forwarded)); + + let ops = ChannelOperations::with_default_policy(manager.clone()); + ops.register_on(®istry).expect("channel ops register"); + register_tunnel_direct_openable( + &ChannelCore::new(manager.clone(), alkcall::channels::policy::default_policy()), + ®istry, + AuthContext::anonymous(b"alk/tunnel"), + dial, + None, + ) + .expect("the direct openable registers"); + let core = + ChannelCore::new(manager.clone(), alkcall::channels::policy::default_policy()); + core.register_openable_with_establisher( + alktunnels::params::tunnel_open_spec(), + Some(dynamic_bindfirst_establisher( + real_bind(), + Arc::clone(&minted_tx), + )), + make_bindfirst_accept_handler(real_accept_loop(), open_forwarded), + ®istry, + AuthContext::anonymous(b"alk/tunnel"), + None, + ) + .expect("the bind-first listen openable registers"); + + let dp = Dispatcher::new(registry, Arc::new(NoopIdentityProvider)); + dp.serve_single_stream(call_connection, reader, writer) + .await; + }) + }); + + let (consumer_end, producer_end) = tokio::io::duplex(64 * 1024); + let consumer_conn = CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None); + let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None); + + let adapter = + ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap)); + tokio::spawn(async move { + let _ = ProtocolHandler::handle( + &adapter, + producer_conn, + &AuthContext::anonymous(b"alk/channels"), + ) + .await; + }); + + let client = ChannelClient::from_connection_with_serving( + consumer_conn, + Some(ServingConfig { + registry: Arc::clone(&door_registry), + identity_provider: Arc::new(NoopIdentityProvider), + identity: Some(arrival_identity), + }), + ) + .await + .expect("the fabric's channels client establishes"); + + FabricDouble { + client: Arc::new(client), + registry: door_registry, + opener: opener_cell, + minted: tokio::sync::Mutex::new(minted_receiver), + } + } + /// A pre-created duplex target — the "in-process dial" half: the /// test holds the returned side, the map hands the dial seam the /// other half when the establisher renders a matching target. @@ -658,6 +1051,651 @@ mod serving { session.unmount().await; } + + // --- the `-R` listen rows (w3-tunnel-listen) --------------------------- + + /// The tunnel build's authenticated loopback session over the + /// leg-2 double, with the fabric leg + listen seam wired (the + /// `-R` bridge's serve-config halves) and the harness client's + /// acceptor installed (the matching-`-R`-forward stand-in). The + /// returned receiver is the forwarded-opens observation channel. + async fn listen_session( + double: &FabricDouble, + limits: Limits, + ) -> ( + LoopbackSession, + tokio::sync::mpsc::UnboundedReceiver, + ) { + let provider = Arc::new(MemIdentityProvider::new()); + provider.register( + fixture("alpha").fingerprint, + identity("tunnel-rider", &["ssh"]), + ); + let mut serve = harness_serve_config(provider, &[fixture("bravo")], limits); + serve.channel_client = Some(Arc::clone(&double.client)); + serve.tunnel_listen_seam = Some(TunnelListenSeam::new(Arc::clone(&double.registry))); + let adapter = SshAdapter::new(serve) + .await + .expect("the -R wired adapter constructs"); + + let policy = Arc::new(MemHostKeyPolicy::with_default(true)); + let (acceptor, receiver) = super::common::harness::ForwardedAcceptor::new(); + let mut session = super::common::harness::drive_with_acceptor( + adapter, + Arc::clone(&policy), + Limits::default(), + Some(Arc::new(acceptor)), + ) + .await + .expect("the loopback mounts"); + + let result = + authenticate_publickey(&mut session.client, "tunnel-rider", fixture("alpha")).await; + assert!( + result.success(), + "the fixture identity authenticates before every -R row" + ); + (session, receiver) + } + + /// A free loopback port (the rows' pick helper; the listen binds + /// exactly what it was asked). + 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 arrival row's pump legs: the arrival's request reaches the + /// forwarded SSH channel, the channel's reply reaches the arrival + /// socket — the fabric `BiStream` ↔ SSH channel duplex, zero + /// framing both directions (the route-1 bridge's primitive, + /// reversed). + async fn arrival_pump( + arrival: &mut tokio::net::TcpStream, + channel: &mut Channel, + request: &[u8], + reply: &[u8], + context: &str, + ) { + let mut seen: Vec = Vec::new(); + tokio::time::timeout(Duration::from_secs(15), async { + while seen.len() < request.len() { + match channel.wait().await { + Some(ChannelMsg::Data { data }) => seen.extend_from_slice(&data), + Some(_) => continue, + None => panic!("the forwarded channel ended before the request arrived"), + } + } + }) + .await + .expect("the arrival's request reaches the forwarded channel in time"); + assert_eq!( + seen, request, + "{context}: the fabric → SSH pump carries the arrival bytes unframed" + ); + + channel + .data_bytes(reply.to_vec()) + .await + .expect("{context}: the SSH channel carries the reply toward the arrival"); + let mut seen_reply = vec![0u8; reply.len()]; + tokio::time::timeout(Duration::from_secs(15), arrival.read_exact(&mut seen_reply)) + .await + .expect("the reply reaches the arrival socket in time") + .expect("{context}: the SSH → fabric pump carries the reply bytes unframed"); + assert_eq!(seen_reply, reply, "{context}: the reply arrives verbatim"); + } + + /// - *Contract stamp:* mapping.md §Global-request rows + /// (`tcpip_forward` → the family-ward listen dial) + §Channel-type + /// rows, the `forwarded-tcpip` row (the data leg: per fabric + /// arrival the door opens a `forwarded-tcpip` channel **toward + /// the vanilla client**, echoing the listen request's own args — + /// the OpenSSH matcher parity — and pumping the fabric + /// `BiStream` ↔ the SSH channel). Observed end to end through the + /// harness's accept hook. + #[tokio::test] + async fn tcpip_forward_serves_through_the_family_listen() { + let double = wire_fabric_with_listen(true, true).await; + let (session, mut opens) = listen_session(&double, Limits::default()).await; + + let port = pick_port().await; + // A nonzero request's success reply carries no port (russh/sshd + // parity); the arrival's echo below is the actually-bound + // assertion. + let _reply = tcpip_forward(&session.client, "127.0.0.1", u32::from(port)) + .await + .expect("the featured + configured listen serves"); + + let mut arrival = tokio::net::TcpStream::connect(("127.0.0.1", port)) + .await + .expect("the fabric's listener accepts the arrival"); + arrival + .write_all(b"arrival-request") + .await + .expect("the arrival socket carries the request"); + + let open = tokio::time::timeout(Duration::from_secs(15), opens.recv()) + .await + .expect("the forwarded open arrives in time") + .expect("the acceptor observed the server-initiated open"); + let ForwardedOpen::Tcpip { + connected_address, + connected_port, + originator_address, + originator_port, + mut channel, + } = open + else { + panic!("the -R data leg opens forwarded-tcpip, got another open shape"); + }; + assert_eq!( + connected_address, "127.0.0.1", + "the connected address echoes the listen request's wire string verbatim (sshd parity)" + ); + assert_eq!( + connected_port, + u32::from(port), + "the port echoes the actually-bound number (the port the fabric bound)" + ); + assert_eq!( + originator_address, "127.0.0.1", + "the originator carries the arrival's peer (TunnelForwardedParams.peer)" + ); + assert!( + originator_port > 0, + "the originator's port is the arrival's ephemeral port" + ); + + arrival_pump( + &mut arrival, + &mut channel, + b"arrival-request", + b"forwarded-reply", + "the tcpip_forward serve row", + ) + .await; + + session.unmount().await; + } + + /// - *Contract stamp:* `client.md` §Lifecycle primitives (the + /// bound-port spec: `port = 0` → the server picks and the + /// actually-bound number returns) — the out-param carries the + /// fabric's `bound` reply field's port, and the arrival's echo + /// port is that same allocated number (sshd's `get_local_port` + /// parity — a forwarded open for a `0`-minted listener never + /// claims port 0). + #[tokio::test] + async fn port_zero_mint_is_resolved_into_the_bound_port_out_param() { + let double = wire_fabric_with_listen(true, true).await; + let (session, mut opens) = listen_session(&double, Limits::default()).await; + + let bound = tcpip_forward(&session.client, "127.0.0.1", 0) + .await + .expect("the port-0 mint serves"); + assert_ne!(bound, 0, "the fabric's actually-bound port rides out"); + + let mut arrival = + tokio::net::TcpStream::connect(("127.0.0.1", u16::try_from(bound).expect("u16"))) + .await + .expect("the minted listener accepts"); + arrival.write_all(b"zero-mint").await.expect("rides"); + + let open = tokio::time::timeout(Duration::from_secs(15), opens.recv()) + .await + .expect("the forwarded open arrives in time") + .expect("the acceptor observed the open"); + let ForwardedOpen::Tcpip { + connected_address, + connected_port, + mut channel, + .. + } = open + else { + panic!("the tcp forward opened forwarded-tcpip"); + }; + assert_eq!(connected_address, "127.0.0.1"); + assert_eq!(connected_port, bound, "the echo port is the minted one"); + let mut seen: Vec = Vec::new(); + tokio::time::timeout(Duration::from_secs(15), async { + while seen.len() < 9 { + match channel.wait().await { + Some(ChannelMsg::Data { data }) => seen.extend_from_slice(&data), + Some(_) => continue, + None => panic!("the forwarded channel ended before the arrival rode"), + } + } + }) + .await + .expect("the pump carries the arrival in time"); + assert_eq!(seen, b"zero-mint"); + + session.unmount().await; + } + + /// - *Contract stamp:* the cancel forms (`cancel_tcpip_forward`): + /// 1:1 into the family listener's teardown — the matching listen + /// cancels `Ok`, the fabric's bound listener goes away (a + /// re-connect is refused), and a second cancel of the same + /// (address, port) pair answers the not-found denial + /// (`RequestDenied` — the sshd cancel-of-unknown parity). + #[tokio::test] + async fn cancel_tcpip_forward_tears_the_fabric_listener_down() { + let double = wire_fabric_with_listen(true, true).await; + let (session, mut opens) = listen_session(&double, Limits::default()).await; + + let port = pick_port().await; + let bound = u32::from(port); + tcpip_forward(&session.client, "127.0.0.1", bound) + .await + .expect("the listen serves before its cancel"); + + cancel_tcpip_forward(&session.client, "127.0.0.1", bound) + .await + .expect("the pending listen's cancel answers success"); + let denied = cancel_tcpip_forward(&session.client, "127.0.0.1", bound) + .await + .expect_err("a second cancel of the same pair is the not-found denial"); + assert!( + matches!(denied, Error::RequestDenied), + "expected RequestDenied, got {denied:?}" + ); + + // The fabric listener is gone (the teardown cascaded: the + // door's listener session dropped, the far accept loop's EOF + // sentinel fired, the plan's listener dropped — the bind + // releases). Bounded retry: the connect is refused. + tokio::time::timeout(Duration::from_secs(10), async { + loop { + match tokio::net::TcpStream::connect(("127.0.0.1", port)).await { + Err(_) => return, + Ok(stream) => drop(stream), + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + }) + .await + .expect("the fabric listener never came back after the cancel"); + + // The torn-down listen could not have served a late arrival as + // an SSH channel: nothing rode the acceptor. + tokio::time::timeout(Duration::from_millis(500), opens.recv()) + .await + .expect_err("no forwarded open may follow a cancelled listen"); + assert!(opens.is_empty()); + + session.unmount().await; + } + + /// - *Contract stamp:* the arrival pump's correlation-miss class + /// (`w3-tunnel-listen` — teardown-at-birth, never a half-open SSH + /// channel): an arrival whose `listen` token has no route + /// (nothing was ever minted for this door) is dropped by the + /// door-side handler — the open op's channel tears down at + /// birth, and the fabric side's adopted halves EOF immediately. + #[tokio::test] + async fn a_correlation_miss_adopts_no_ssh_channel() { + let double = wire_fabric_with_listen(true, true).await; + let provider = Arc::new(MemIdentityProvider::new()); + provider.register( + fixture("alpha").fingerprint, + identity("tunnel-rider", &["ssh"]), + ); + let mut serve = harness_serve_config(provider, &[fixture("bravo")], Limits::default()); + serve.channel_client = Some(Arc::clone(&double.client)); + serve.tunnel_listen_seam = Some(TunnelListenSeam::new(Arc::clone(&double.registry))); + let _adapter = SshAdapter::new(serve) + .await + .expect("the -R wired adapter constructs (the registration is adapter-scope)"); + + let params = alktunnels::params::TunnelForwardedParams { + listen: 424_242, + substrate: alktunnels::params::Substrate::Tcp, + peer: alktunnels::params::SubstrateAddr::HostPort { + host: "127.0.0.1".to_owned(), + port: 1234, + }, + }; + let (mut send, mut recv) = double + .open_forwarded_directly(params) + .await + .expect("the door-side openable allocated the arrival's channel"); + send.write_all(b"orphan arrival") + .await + .expect("the adopted halves still write (the door dropped its end — EOF, not refusal)"); + + // The door-side handler dropped the fabric half at birth: the + // EOF arrives bounded, the channel never half-serves. + tokio::time::timeout(Duration::from_secs(10), recv.read_u8()) + .await + .expect("teardown-at-birth is observable in time") + .expect_err("the adoptee's read half EOFs (the channel reapped)"); + } + + /// - *Contract stamp:* AGENTS.md convention 7 (the visible-surface + /// invariant's alkcall half on the `-R` leg): the door-side + /// registry's `AccessControl::check` gates arrivals by the fabric + /// caller's identity — an identity lacking the + /// `tunnel:forwarded` scope is refused by alkcall's dispatch (the + /// openable's own spec) and no SSH channel opens, even for a + /// genuinely minted token. + #[tokio::test] + async fn the_door_side_acl_refuses_an_unscoped_arrival() { + let double = wire_fabric_with_listen(true, false).await; + let (session, mut opens) = listen_session(&double, Limits::default()).await; + + let port = pick_port().await; + let _bound = tcpip_forward(&session.client, "127.0.0.1", u32::from(port)) + .await + .expect( + "the listen itself is gated by the fabric-side identity, which carries the scope", + ); + + // A real token (minted for this listener), unscoped arrival + // identity: the door-side ACL refuses the open op — the fabric + // side sees the refusal, never a channel allocation. + let token = double.minted_token().await; + let params = alktunnels::params::TunnelForwardedParams { + listen: token, + substrate: alktunnels::params::Substrate::Tcp, + peer: alktunnels::params::SubstrateAddr::HostPort { + host: "127.0.0.1".to_owned(), + port: 4321, + }, + }; + let outcome = double.open_forwarded_directly(params).await; + assert!( + outcome.is_err(), + "the unscoped arrival identity is refused at the door's dispatch" + ); + + // Nothing rode the acceptor. + tokio::time::timeout(Duration::from_millis(500), opens.recv()) + .await + .expect_err("an ACL-refused arrival never opens an SSH channel"); + assert!(opens.is_empty()); + + session.unmount().await; + } + + /// - *Contract stamp:* AGENTS.md convention 9 (the listener count + /// is a bounded resource): the per-session listener map is + /// bounded through its dedicated counter against the same + /// `max_concurrent_channels` ceiling — the listen beyond the + /// count answers the vanilla `RequestDenied` while channels stay + /// free, and the cancel releases the count again. + #[tokio::test] + async fn the_listener_cap_bounds_the_session_listen_map() { + let double = wire_fabric_with_listen(true, true).await; + let limits = Limits { + max_concurrent_channels: 1, + ..Limits::default() + }; + let (session, mut opens) = listen_session(&double, limits).await; + + let first = tcpip_forward(&session.client, "127.0.0.1", 0) + .await + .expect("the first listen is admitted"); + assert_ne!(first, 0); + let err = tcpip_forward(&session.client, "127.0.0.1", 0) + .await + .expect_err("the second listen rides beyond the listener cap"); + assert!( + matches!(err, Error::RequestDenied), + "expected RequestDenied, got {err:?}" + ); + + // The cap recovery: the cancel of the first listen frees the + // count; the next mint is admitted. + cancel_tcpip_forward(&session.client, "127.0.0.1", first) + .await + .expect("the cancel releases the count"); + let second = tcpip_forward(&session.client, "127.0.0.1", 0) + .await + .expect("the listener count recovered at the cancel"); + assert_ne!(second, 0); + + // Nothing in this row ever manufactured an arrival. + tokio::time::timeout(Duration::from_millis(300), opens.recv()) + .await + .expect_err("no arrivals were manufactured in this row"); + assert!(opens.is_empty()); + + session.unmount().await; + } + + /// - *Contract stamp:* AGENTS.md convention 9 / the acceptor + /// vocabulary's cap rows (the wave-2 gate's booked shape): the + /// per-arrival admission is the owning session's in-crate + /// channel cap — an arrival onto an exhausted count is refused + /// at the accept surface (no SSH channel opens; the fabric half + /// drops, teardown-at-birth), and an established channel's close + /// recovers the count (the next arrival is admitted — the + /// release proof). + #[tokio::test] + async fn the_arrival_admission_is_cap_gated_at_the_accept_surface() { + let double = wire_fabric_with_listen(true, true).await; + let limits = Limits { + max_concurrent_channels: 1, + ..Limits::default() + }; + let (session, mut opens) = listen_session(&double, limits).await; + + let port = pick_port().await; + let _reply = tcpip_forward(&session.client, "127.0.0.1", u32::from(port)) + .await + .expect("a minted listen costs no channel count"); + + // The first arrival admits itself: its SSH channel opens and + // the count is full (held by the arrival socket + channel, + // both deliberately kept alive un-served). + let first_arrival = tokio::net::TcpStream::connect(("127.0.0.1", port)) + .await + .expect("the first arrival is admitted"); + let first_open = tokio::time::timeout(Duration::from_secs(15), opens.recv()) + .await + .expect("the first arrival's open arrives") + .expect("observed"); + let ForwardedOpen::Tcpip { + channel: first_channel, + .. + } = first_open + else { + panic!("the tcp arrival opened forwarded-tcpip"); + }; + + // The second arrival meets the full count: no SSH channel + // opens (its fabric half teardown-at-births), and its socket + // EOFs bounded. + let mut second_arrival = tokio::net::TcpStream::connect(("127.0.0.1", port)) + .await + .expect("the fabric listener still accepts arrivals"); + tokio::time::timeout(Duration::from_millis(700), opens.recv()) + .await + .expect_err("an arrival beyond the cap never opens an SSH channel"); + let eof = tokio::time::timeout(Duration::from_secs(10), second_arrival.read_u8()) + .await + .expect("the refused arrival's teardown is observable in time") + .expect_err("the refused arrival's socket EOFs"); + let _ = eof; + + // The release proof: the first arrival's socket goes — the + // pump's fabric→SSH direction EOFs (the pump's + // shutdown-on-completion half-closes the SSH channel — the + // OpenSSH client, whose target socket died with the arrival, + // closes its half; the harness's stand-in does exactly that). + // That close — not the pump — is what fires the packet loop's + // established-channel release. + drop(first_arrival); + let _ = first_channel.close().await; + + let mut third_arrival = tokio::net::TcpStream::connect(("127.0.0.1", port)) + .await + .expect("the fabric listener still serves"); + third_arrival.write_all(b"again").await.expect("rides"); + let third_open = tokio::time::timeout(Duration::from_secs(15), opens.recv()) + .await + .expect("the next arrival's open arrives — the released count recovered") + .expect("observed"); + let ForwardedOpen::Tcpip { + channel: mut third_channel, + .. + } = third_open + else { + panic!("the tcp arrival opened forwarded-tcpip"); + }; + let mut seen: Vec = Vec::new(); + tokio::time::timeout(Duration::from_secs(15), async { + while seen.len() < 5 { + match third_channel.wait().await { + Some(ChannelMsg::Data { data }) => seen.extend_from_slice(&data), + Some(_) => continue, + None => panic!("the forwarded channel ended before the bytes rode"), + } + } + }) + .await + .expect("the pump rides in time"); + assert_eq!(seen, b"again"); + + session.unmount().await; + } + + /// - *Contract stamp:* OQ-11 on the listen flow (the runtime-Unix + /// gate riding the `streamlocal_forward` serving flow) + the unix + /// row of the `-R` leg: the unix listen serves through the fabric + /// (the path echoes verbatim — no port semantics, no port + /// out-param), the arrival opens + /// `forwarded-streamlocal@openssh.com` toward the client, and the + /// cancel form tears it down 1:1 (path-matched). + #[cfg(unix)] + #[tokio::test] + async fn streamlocal_forward_serves_arrivals_and_cancels_through_the_family() { + let double = wire_fabric_with_listen(true, true).await; + let (session, mut opens) = listen_session(&double, Limits::default()).await; + + let path = format!("/tmp/alkssh-w3-listen-{}-listen.sock", std::process::id()); + super::common::harness::streamlocal_forward(&session.client, &path) + .await + .expect("the unix listen serves"); + + let mut arrival = tokio::net::UnixStream::connect(&path) + .await + .expect("the fabric's unix listener accepts the arrival"); + arrival.write_all(b"unix-arrival").await.expect("rides"); + + let open = tokio::time::timeout(Duration::from_secs(15), opens.recv()) + .await + .expect("the forwarded open arrives in time") + .expect("observed"); + let ForwardedOpen::Streamlocal { + socket_path, + mut channel, + } = open + else { + panic!("the unix arrival opens forwarded-streamlocal"); + }; + assert_eq!( + socket_path, path, + "the server_socket_path echoes the listen request's path verbatim" + ); + let mut seen: Vec = Vec::new(); + tokio::time::timeout(Duration::from_secs(15), async { + while seen.len() < 12 { + match channel.wait().await { + Some(ChannelMsg::Data { data }) => seen.extend_from_slice(&data), + Some(_) => continue, + None => panic!("the forwarded channel ended before the arrival rode"), + } + } + }) + .await + .expect("the unix pump carries the arrival"); + assert_eq!(seen, b"unix-arrival"); + channel + .data_bytes(b"unix-reply".to_vec()) + .await + .expect("the SSH channel carries the reply"); + let mut reply = vec![0u8; 10]; + tokio::time::timeout(Duration::from_secs(15), arrival.read_exact(&mut reply)) + .await + .expect("the reply reaches the arrival stream in time") + .expect("rides"); + assert_eq!(reply, b"unix-reply"); + + super::common::harness::cancel_streamlocal_forward(&session.client, &path) + .await + .expect("the pending unix listen's cancel answers success"); + let denied = super::common::harness::cancel_streamlocal_forward(&session.client, &path) + .await + .expect_err("a second cancel of the same path is the not-found denial"); + assert!(matches!(denied, Error::RequestDenied), "got {denied:?}"); + + session.unmount().await; + let _ = std::fs::remove_file(&path); + } + + /// - *Contract stamp:* the unset-declines contract on the listen + /// seam (`w3-tunnel-listen`'s whole-`Option` shape — an + /// egress-only door is legitimate config): a `tunnel` build with + /// a fabric leg but no `tunnel_listen_seam` declines every listen + /// form with the vanilla `RequestDenied`, while the leg-1 rows + /// still route (the egress bridge is not dead). + #[tokio::test] + async fn listen_forms_decline_when_the_listen_seam_is_unset() { + let targets: TargetMap = Arc::default(); + let client = wire_fabric( + inproc_dial(targets, Arc::new(AtomicUsize::new(0))), + scoped_caller(), + Arc::new(tokio::sync::Mutex::new(None)), + ) + .await; + let provider = Arc::new(MemIdentityProvider::new()); + provider.register( + fixture("alpha").fingerprint, + identity("tunnel-rider", &["ssh"]), + ); + let mut serve = harness_serve_config(provider, &[fixture("bravo")], Limits::default()); + serve.channel_client = Some(client); + let adapter = SshAdapter::new(serve) + .await + .expect("the egress-only door constructs without the seam"); + + let policy = Arc::new(MemHostKeyPolicy::with_default(true)); + let mut session = + super::common::harness::drive(adapter, Arc::clone(&policy), Limits::default()) + .await + .expect("the loopback mounts"); + let result = + authenticate_publickey(&mut session.client, "tunnel-rider", fixture("alpha")).await; + assert!(result.success()); + + let err = tcpip_forward(&session.client, "127.0.0.1", 8080) + .await + .expect_err("the unset seam declines the listen"); + assert!(matches!(err, Error::RequestDenied), "got {err:?}"); + let err = cancel_tcpip_forward(&session.client, "127.0.0.1", 8080) + .await + .expect_err("the unset seam declines the cancel too"); + assert!(matches!(err, Error::RequestDenied), "got {err:?}"); + + // The egress door still routes: the direct row reaches the + // fabric (the fail-dial class — `ConnectFailed`, not + // `AdministrativelyProhibited`). + let err = open_direct_tcpip(&session.client, "127.0.0.1", 36005) + .await + .expect_err("the direct row still routes onto the bridge"); + open_failure( + err, + ChannelOpenFailure::ConnectFailed, + "the egress-only door's direct row (the seam-unconfigured row's control leg)", + ); + + session.unmount().await; + } } #[cfg(all(feature = "tunnel", unix))] @@ -671,13 +1709,14 @@ mod real_tool { use std::sync::Arc; - use alktunnels::params::{Substrate, TUNNEL_DIRECT_SCOPE}; - use alktunnels::producer::{DialFn, TargetHandle, TunnelEstablishError}; + use alktunnels::params::{Substrate, SubstrateAddr, TUNNEL_DIRECT_SCOPE}; + use alktunnels::producer::{BindFn, BoundListener, DialFn, TargetHandle, TunnelEstablishError}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use super::common::harness::serve_config as harness_serve_config; use super::common::{MemIdentityProvider, fixture, identity}; - use super::serving::wire_fabric; + use super::serving::{wire_fabric, wire_fabric_with_listen}; + use alkssh::TunnelListenSeam; /// The smoke's dial seam: a real-TCP dial closure (hand-rolled — /// the assembly's `DialFn` injected over the substrate for real; @@ -865,6 +1904,172 @@ mod real_tool { let _ = std::fs::remove_dir_all(&work); } + /// The smoke's bind seam: the leg-2 dynamic-address producer's + /// real-TCP bind (the wire-named endpoint bound for real; shared + /// shape with the harness rows' `real_bind`). + fn tcp_bind() -> BindFn { + Arc::new(move |_substrate, endpoint: &str| { + let endpoint = endpoint.to_owned(); + let endpoint = endpoint.replace("localhost:", "127.0.0.1:"); + Box::pin(async move { + let listener = tokio::net::TcpListener::bind(&endpoint) + .await + .map_err(|e| { + TunnelEstablishError::ResourceShortage(format!("bind `{endpoint}`: {e}")) + })?; + let addr = listener.local_addr().expect("bound addr"); + Ok(BoundListener { + listener: Arc::new(listener), + bound: SubstrateAddr::HostPort { + host: addr.ip().to_string(), + port: addr.port(), + }, + }) + }) + }) + } + + #[tokio::test] + async fn real_ssh_dash_r_smoke() { + if std::process::Command::new("ssh") + .arg("-V") + .output() + .is_err() + { + eprintln!("SKIPPED (with reason): real-tool -R smoke — ssh is unavailable"); + return; + } + if std::process::Command::new("curl") + .arg("--version") + .output() + .is_err() + { + eprintln!("SKIPPED (with reason): real-tool -R smoke — curl is unavailable"); + return; + } + + // The vanilla client's own dial target (the `-R` convention: + // the client dials its own endpoint per arrival, from the + // machine the SSH client runs on). + let (http_addr, body) = spawn_http_endpoint().await; + let double = wire_fabric_with_listen(true, true).await; + + let provider = Arc::new(MemIdentityProvider::new()); + provider.register( + fixture("alpha").fingerprint, + identity("tunnel-rider", &["ssh"]), + ); + let mut serve = + harness_serve_config(provider, &[fixture("bravo")], alkssh::Limits::default()); + serve.channel_client = Some(Arc::clone(&double.client)); + serve.tunnel_listen_seam = Some(TunnelListenSeam::new(Arc::clone(&double.registry))); + let adapter = Arc::new( + alkssh::SshAdapter::new(serve) + .await + .expect("the smoke adapter constructs"), + ); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("the smoke listener binds"); + let ssh_port = listener.local_addr().expect("local addr").port(); + let accept_task = tokio::spawn(async move { + loop { + let Ok((stream, _)) = listener.accept().await else { + return; + }; + let adapter = Arc::clone(&adapter); + tokio::spawn(async move { + let _ = adapter.handle(stream).await; + }); + } + }); + + let fwd_port: u16 = pick_local_port().await; + let mut ssh_child = tokio::process::Command::new("ssh") + .args([ + "-N", + "-i", + work_key_path().to_str().expect("tmp path"), + "-p", + &ssh_port.to_string(), + "-o", + "BatchMode=yes", + "-o", + "IdentitiesOnly=yes", + "-o", + "StrictHostKeyChecking=no", + "-o", + "UserKnownHostsFile=/dev/null", + "-o", + "ConnectTimeout=10", + "-o", + "ExitOnForwardFailure=yes", + "-R", + &format!( + "127.0.0.1:{fwd_port}:{}:{}", + http_addr.ip(), + http_addr.port() + ), + ]) + .arg("tunnel-rider@127.0.0.1") + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .spawn() + .expect("vanilla ssh spawns"); + + // The fabric bound the -R listener; curl arrives on it, the + // door maps the arrival to a forwarded-tcpip channel the + // vanilla client accepts and dials its own endpoint through. + let through = tokio::time::timeout(std::time::Duration::from_secs(30), async { + let mut attempts = 0u32; + loop { + let out = tokio::process::Command::new("curl") + .args(["-s", "--max-time", "10"]) + .arg(format!("http://127.0.0.1:{fwd_port}/")) + .output() + .await + .expect("curl spawns"); + if out.status.success() { + return out; + } + attempts += 1; + assert!( + attempts < 20, + "curl never got through: exit {:?}", + out.status.code() + ); + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + } + }) + .await + .expect("the -R smoke resolves in time"); + + let _ = ssh_child.kill().await; + accept_task.abort(); + + let stdout = String::from_utf8_lossy(&through.stdout).to_string(); + assert!( + through.status.success() && stdout.trim_end() == body.trim_end(), + "vanilla ssh -R + curl through the fabric → arrival → forwarded-tcpip → client-dialed endpoint returns the body — got {stdout:?}" + ); + } + + /// The smoke's per-run workdir key path (the -L smoke's inline + /// shape extracted). + fn work_key_path() -> std::path::PathBuf { + use std::os::unix::fs::PermissionsExt; + + let work = + std::env::temp_dir().join(format!("alkssh-tunnel-r-smoke-{}", std::process::id())); + std::fs::create_dir_all(&work).expect("the smoke workdir creates"); + let key_path = work.join("id_ed25519"); + std::fs::write(&key_path, fixture("alpha").pem).expect("the fixture key writes"); + std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o600)) + .expect("the key file perms set"); + key_path + } + async fn pick_local_port() -> u16 { let listener = tokio::net::TcpListener::bind("127.0.0.1:0") .await