diff --git a/Cargo.lock b/Cargo.lock index 792e78b..91e69c2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -114,6 +114,7 @@ dependencies = [ "alkcall", "alkgit", "alktty", + "alktunnels", "alkvault", "async-trait", "bytes", @@ -150,6 +151,22 @@ dependencies = [ "tracing", ] +[[package]] +name = "alktunnels" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee56da894f2d02120d3602df32c7f6a9db7f930a401df66d40b632cd9767af41" +dependencies = [ + "alkcall", + "bytes", + "futures", + "serde", + "serde_json", + "thiserror 2.0.21", + "tokio", + "tracing", +] + [[package]] name = "alkvault" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index a25254d..c5cf66e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,12 +31,12 @@ name = "alkssh" # sibling dep is gated feature-to-dep so a git-build never compiles # alktty and vice versa — the wave-1 manifest rule: per-feature sibling # deps land with their first consuming wave, which for alktty/alkgit is -# the wave-3 config fields; alktunnels still lands with its first -# consuming code, w3-tunnel-direct). +# the wave-3 config fields and for alktunnels `w3-tunnel-direct`, the +# route-1 bridge). default = [] session = ["dep:alktty", "dep:serde_json", "dep:bytes"] git = ["dep:alkgit"] -tunnel = [] +tunnel = ["dep:alktunnels"] sftp = [] [dependencies] @@ -58,6 +58,16 @@ alkvault = "0.1" # touches a real backend; the `local` shapes stay alktty-side. alktty = { version = "0.8.1", default-features = false, optional = true } alkgit = { version = "0.1.0", default-features = false, optional = true } +# tunnel-only: the route-1 bridge's consumer surface (`TunnelSession:: +# open_direct`/`take_halves`, `SubstrateAddr`/`Substrate`). Pin +# re-verified against crates.io at `w3-tunnel-direct` (2026-10-29): +# 0.4.1 is `max_stable_version`/`newest_version`, MIT OR Apache-2.0, +# rust_version 1.88 ≤ our 1.89, alkcall ^0.10.1 pinned exactly +# (single-versioned against our `alkcall = "0.10"`). The `local` +# feature stays off — the dial seam is injected (`DialFn`); the real +# dial/backends belong to the assembly, and the tests hand-roll their +# own dial/dial-fail seams (`tests/tunnel_mapping.rs`). +alktunnels = { version = "0.4.1", default-features = false, optional = true } # session-only: the OQ-44 reserved ttymodes JSON (`TerminalParamsWire.modes`) # and the exit-chunk / negotiation-error-frame JSON reads in `tty_map.rs`. # Same pin alktty carries (crate resolution stays single-versioned). diff --git a/src/channel_cap.rs b/src/channel_cap.rs index fa9f220..e81fff4 100644 --- a/src/channel_cap.rs +++ b/src/channel_cap.rs @@ -14,72 +14,108 @@ //! open is refused — a clean decline, **never queued, never half-open** //! (the fail-closed posture; the refusal reason is //! [`crate::dispatch::cap_exceeded_open_decline`]). Release -//! ([`ChannelCap::release`]) fires from the `channel_close` hook, so -//! the cap recovers as channels close. +//! ([`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. //! -//! ## Why no lock +//! ## Release sites (the `w3-tunnel-direct` recorded decision) //! -//! The count is **handler-owned, not shared**: russh at this pin runs -//! every handler hook inline in the session's packet loop with -//! exclusive `&mut self` access (ADR-006 decision 4 puts serving -//! paths on spawned tasks, but those report through the session -//! handle — their channel closes cycle back through the packet loop -//! as `CHANNEL_CLOSE`, firing the same `channel_close` hook). The -//! only writer is that single loop, so the plain field is sound and a -//! lock would only mislead about shared mutation that does not exist. -//! The per-identity policy layer (when the alkcall bridge carries it) -//! is additional and arrives with the bridge, not here. +//! Every admission has exactly one release, and the two paths are +//! disjoint by the wire shape: //! -//! ## Soundness of the decrement invariant +//! 1. **The established-channel close** — the `channel_close` hook. +//! russh routes `CHANNEL_CLOSE` only for channels it considers +//! established; a declined open never established one, and a close +//! 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` +//! (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. //! -//! Every `channel_close` delivery corresponds to exactly one -//! previously-admitted open: russh only routes `CHANNEL_CLOSE` for -//! channels it considers established, and a declined open never -//! establishes one — so no refusal path can ever decrement for a -//! channel it never counted. A `CHANNEL_CLOSE` is also consumed at -//! most once (russh removes the channel before the hook), so one -//! admitted channel decrements exactly once. [`ChannelCap::release`] -//! additionally saturates, making a stray decrement impossible to -//! manufacture capacity with. Channels never closed before session -//! teardown stay counted until the handler dies with the session — -//! per-session state, gone with the connection. +//! ## Why the mutex (the single-writer reasoning, updated) +//! +//! 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. + +use parking_lot::Mutex; /// The per-session active-channel count for the in-crate /// concurrent-channel cap — see the module docs for the shape, the -/// single-writer reasoning, and the decrement invariant. +/// release sites, and the writer/soundness reasoning. Shared with the +/// spawned route-1 bridge tasks ([`Arc`]) exactly so the +/// fabric-failure release can fire from the task that knows it. #[derive(Debug, Default)] pub(crate) struct ChannelCap { - active: usize, + active: Mutex, } impl ChannelCap { - /// Whether an open has room under `cap` — the pure check, for the - /// paths whose decline is ordered *before* the accept decision - /// (this wave: the tunnel rows, whose seam stub declines without - /// ever opening; wave 3 moves the admission call to the - /// accept-after-fabric-open point). + /// 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 < cap + *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. - pub(crate) fn admit(&mut self, cap: usize) -> bool { - if self.has_room(cap) { - self.active += 1; + /// 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). + pub(crate) fn admit(&self, cap: usize) -> bool { + let mut active = self.active.lock(); + if *active < cap { + *active += 1; true } else { false } } - /// A counted channel closed — the `channel_close` hook's - /// decrement, recovery of the cap. Saturating: a count never goes - /// below zero, so no decrement path can manufacture capacity. - pub(crate) fn release(&mut self) { - self.active = self.active.saturating_sub(1); + /// The release: the established channel's close (the hook) or the + /// admitted-but-rejected bridge's failure answer (the spawned + /// task). Disjoint per admission by the wire shape (see the + /// release-sites section). Saturating: no path manufactures + /// capacity. + pub(crate) fn release(&self) { + let mut active = self.active.lock(); + *active = active.saturating_sub(1); + } + + /// The active count — in-crate inspection for the unit rows. + #[cfg(test)] + pub(crate) fn active(&self) -> usize { + *self.active.lock() } } @@ -94,18 +130,19 @@ mod tests { /// the configured ceiling and refuses beyond it). #[test] fn admits_up_to_the_cap_then_refuses() { - let mut cap = ChannelCap::default(); + let cap = ChannelCap::default(); 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); } /// - *Contract stamp:* the task's cap mechanics (a close recovers /// the cap — the count decrements on close). #[test] fn release_recovers_the_cap() { - let mut cap = ChannelCap::default(); + let cap = ChannelCap::default(); assert!(cap.admit(CAP)); assert!(cap.admit(CAP)); assert!(!cap.admit(CAP)); @@ -121,7 +158,7 @@ mod tests { /// bypass). #[test] fn zero_cap_refuses_every_open() { - let mut cap = ChannelCap::default(); + let cap = ChannelCap::default(); assert!(!cap.admit(0)); cap.release(); assert!( @@ -135,10 +172,30 @@ mod tests { /// decrement invariant this crate's own gates make unreachable). #[test] fn release_saturates_at_zero() { - let mut cap = ChannelCap::default(); + let cap = ChannelCap::default(); cap.release(); cap.release(); - assert_eq!(cap.active, 0); + assert_eq!(cap.active(), 0); assert!(cap.admit(CAP)); } + + /// - *Contract stamp:* `w3-tunnel-direct`'s recorded release + /// mechanism (the admitted-but-rejected bridge releases inside + /// the spawned task): the cap is shared safely — admission from + /// the packet-loop site, release from anywhere that holds the + /// `Arc` — and disjoint releases never double-count. + #[test] + fn the_cap_is_shareable_between_the_loop_and_bridge_tasks() { + let cap = std::sync::Arc::new(ChannelCap::default()); + let shared = std::sync::Arc::clone(&cap); + + std::thread::spawn(move || { + shared.release(); + }) + .join() + .expect("the bridge-site release thread runs"); + + assert!(cap.admit(CAP), "a foreign-thread release stays sound"); + assert_eq!(cap.active(), 1); + } } diff --git a/src/dispatch.rs b/src/dispatch.rs index f75c8cc..400f932 100644 --- a/src/dispatch.rs +++ b/src/dispatch.rs @@ -213,14 +213,20 @@ pub(crate) fn unfeatured_open_decline() -> ChannelOpenFailure { ChannelOpenFailure::AdministrativelyProhibited } -/// The wire decline reason for the **tunnel seam stub** — the row is -/// feature-gated `tunnel` but the route-1 bridge (wave 3) does not -/// exist yet, so the fabric open never happened: `connect failed`, the -/// explicit reject shape ADR-006 decision 1 pins (accept only after -/// the fabric open succeeds; failure maps to an explicit SSH -/// `ConnectFailed`). Wave 3 replaces the stub with the real -/// accept-after-open flow. -pub(crate) fn tunnel_seam_stub_decline() -> ChannelOpenFailure { +/// The wire decline reason for a **cannot-bridge** tunnel open +/// (`w3-tunnel-direct` — the wave-2 seam stub's reason surviving as +/// the real decline): `connect failed`, the explicit reject shape +/// ADR-006 decision 1 pins — accept only after the fabric open +/// succeeds, and a fabric-open failure maps to this explicit `SSH +/// `ConnectFailed` (the channel never establishes, so the client +/// observes a channel-scoped refusal, no hang). The same code covers +/// the other cannot-bridge classes of the direct rows (the fabric +/// entry unconfigured under the feature — the unset-declines posture — +/// and wire arguments naming an impossible target; a vanilla client +/// cannot distinguish these from a dead target, and the event +/// vocabulary records which). Policy refusals stay +/// [`unfeatured_open_decline`] / [`cap_exceeded_open_decline`]. +pub(crate) fn connect_failed_open_decline() -> ChannelOpenFailure { ChannelOpenFailure::ConnectFailed } diff --git a/src/handler.rs b/src/handler.rs index 8c3792f..d56f58c 100644 --- a/src/handler.rs +++ b/src/handler.rs @@ -47,9 +47,18 @@ //! feature-gated (ADR-003 §5). Unfeatured and never-feature rows //! decline **at the open** with a clean wire reason //! ([`crate::dispatch::unfeatured_open_decline`]) — never a -//! half-open channel; the `tunnel`-featured rows decline at the -//! family-seam stub ([`crate::dispatch::tunnel_seam_stub_decline`]) -//! until wave 3's route-1 bridge replaces it. The bare +//! half-open channel. The `direct-tcpip` / `direct-streamlocal` +//! featured rows are the route-1 bridge (`w3-tunnel-direct`, +//! [`crate::tunnel_map`]): the pre-admit decline classes (the unset +//! fabric handle under the feature — the unset-declines posture; the +//! unreachable-in-practice `no-identity`; an impossible wire target), +//! the cap admission at the hook, then the spawned bridge — accept +//! 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). +//! The bare //! `auth-agent@openssh.com` channel *type* //! (`forwarded-streamlocal@openssh.com` likewise) has no russh handler //! hook at this pin — russh rejects those opens inline @@ -140,11 +149,12 @@ //! end-to-end by the mechanics rows in `tests/spawn_return.rs` and the //! mapping rows in `tests/git_mapping.rs` / `tests/tty_mapping.rs`. //! -//! The open-level declines (unfeatured/never-feature rows, the tunnel -//! seam stub) stay inline by class: they answer the open *before* any -//! channel is owned — nothing is served yet, so nothing spawns. Wave -//! 3's accept-only-after-fabric-open flow is where the tunnel row -//! gains its spawned serving path, per the same constraint. +//! The open-level declines (unfeatured/never-feature rows, the +//! direct rows' pre-admit classes) stay inline by class: they answer +//! 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. //! //! Request-level declines: `subsystem_request` declines every name — //! `sftp` is the OQ-13 placeholder (the feature exists, the route @@ -164,10 +174,14 @@ //! §5: channel-scoped decline, auth intact, no hangs): //! //! - **Open-level declines** — channel-open failure codes (`x11` -//! never-feature, unfeatured feature rows → administratively -//! prohibited; the tunnel seam stub → connect failed). These happen -//! before the channel exists, so the SSH open-failure code is the -//! honest, richest reason surface. +//! never-feature, unfeatured feature rows, the tunnel rows' +//! identity-less guard and beyond-cap refusals → administratively +//! 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 +//! happen before the channel exists, so the SSH open-failure code is +//! the honest, richest reason surface. //! - **Request-level declines** — [`Session::channel_failure`]: the //! SSH request-reply vocabulary has no failure text, so a request //! decline's "reason" is the routing decision itself (documented @@ -213,12 +227,20 @@ //! enforced at every `channel_open_*` hook: an accept path beyond //! `Limits::max_concurrent_channels` is refused fail-closed at the //! open ([`crate::dispatch::cap_exceeded_open_decline`] — never -//! queued, never half-open), and the count releases through the -//! `channel_close` hook so the cap recovers. The order of the gates: -//! the feature gate first (an unfeatured build declines everything -//! before the cap matters), the cap gate before any accept — and -//! before the tunnel seam stub, whose decline opens nothing and so -//! must not consume a count. The `max_exec_command_len` cap is +//! queued, never half-open), and the count releases so the cap +//! recovers: through the `channel_close` hook for established +//! channels, and — `w3-tunnel-direct` — inside the route-1 bridge's +//! 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 +//! 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 +//! checks precede the admission — decline-first-alloc-never applies +//! to the budget too), the cap admission before any accept; the +//! admitted count is held across the fabric open and released exactly +//! once on either path. The `max_exec_command_len` cap is //! enforced at `exec_request` *before* the parse and any allocation //! (`w3-git-mapping`): an over-length wire command declines //! request-level, unfeatured-or-not. The per-identity policy layer @@ -259,6 +281,8 @@ use crate::observe; use crate::serve; #[cfg(feature = "session")] use crate::tty_map; +#[cfg(feature = "tunnel")] +use crate::tunnel_map; #[cfg(feature = "session")] use alktty::negotiation::NegotiateRequest; @@ -274,7 +298,14 @@ use alktty::negotiation::NegotiateRequest; /// [`Self::resolved_identity`] — the wave-3 seam reads it). pub struct SessionHandler { serve: Arc, - channel_cap: ChannelCap, + /// 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 + /// 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. + channel_cap: Arc, resolved_identity: Option, /// The store-and-take handoff (`w3-git-mapping`): confirmed /// session channels held from the open until their row takes or @@ -322,7 +353,7 @@ impl SessionHandler { pub fn new(serve: Arc) -> Self { Self { serve, - channel_cap: ChannelCap::default(), + channel_cap: Arc::new(ChannelCap::default()), resolved_identity: None, channels: HashMap::new(), #[cfg(feature = "session")] @@ -356,24 +387,31 @@ impl SessionHandler { /// The accept-path admission under the in-crate concurrent-channel /// cap (bounded resources, `w2-bounded-resources`): room under /// `Limits::max_concurrent_channels` admits and counts the channel; - /// beyond the cap the open is refused fail-closed — never queued, - /// never half-open. Called exactly at accept sites: a decline path - /// (unfeatured, never-feature, the tunnel seam stub) never touches - /// the count, and the only writer is this connection's packet loop - /// ([`ChannelCap`]'s single-writer reasoning). - fn admit_channel_open(&mut self) -> bool { + /// beyond the cap the open is refused fail-closed with a clean + /// wire reason ([`crate::dispatch::cap_exceeded_open_decline`]) — + /// never queued, never half-open. Called exactly at accept sites — + /// the packet loop owns every admit call (the single admit writer, + /// [`ChannelCap`]'s recorded reasoning); a pure-decline path + /// (unfeatured, never-feature, the tunnel rows' pre-admit classes) + /// never touches the count. Since `w3-tunnel-direct`, the tunnel + /// rows admit **at the hook and hold the count across the fabric + /// open**: a fabric failure releases inside the spawned bridge + /// task (the admitted-but-rejected shape — no `channel_close` can + /// ever fire); a success releases when the established channel + /// closes ([`Self::channel_close`]). The release sites are disjoint + /// per admission by the wire shape. + fn admit_channel_open(&self) -> bool { self.channel_cap .admit(self.serve.limits.max_concurrent_channels) } /// Whether the cap is already exhausted — the gate slot for the - /// rows whose decline is ordered *before* the accept decision. - /// This wave that is the tunnel rows, whose seam stub declines - /// without ever opening a channel (a counted admission it never - /// opens would leak against the cap — declined-first-alloc-never, - /// the ADR-006 dispatch rule, applies to the budget too). Wave 3's - /// accept-after-fabric-open flow replaces the stub and moves the - /// admission ([`Self::admit_channel_open`]) to the accept point. + /// 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 @@ -597,20 +635,25 @@ impl russh::server::Handler for SessionHandler { Ok(()) } - /// The `direct-tcpip` open gate (`tunnel`): unfeatured declines - /// at the open; featured routes to the tunnel seam stub — the - /// route-1 bridge is wave-3 material, and the stub's decline is - /// the explicit `ConnectFailed` reject shape ADR-006 decision 1 - /// pins (accept only after the fabric open succeeds; the skeleton - /// never opens the fabric). The in-crate cap gates before the stub - /// ([`Self::channel_cap_exhausted`]): a cap-exhausted build - /// answers the cap reason — the stub opens nothing, so admission - /// moves to wave 3's real accept point. + /// The `direct-tcpip` open gate (`tunnel`) and its route-1 bridge + /// dispatch (`w3-tunnel-direct`): unfeatured declines at the open; + /// featured rows build the bridge target from the SSH open + /// arguments (`SubstrateAddr::HostPort` + `Substrate::Tcp` — an + /// out-of-range wire port is an impossible target, declined + /// pre-admission) and route to [`Self::dispatch_direct_open`]: + /// pre-admit decline classes first (the unset fabric handle + /// declining under the feature — the unset-declines posture — and + /// the unreachable-in-practice `no-identity` guard), then the cap + /// admission **at the hook, held across the fabric open**, then + /// the spawned bridge per spawn-and-return — + /// `TunnelSession::open_direct` → accept only after the fabric + /// open succeeds (`ConnectFailed` on failure, the count released + /// in the task), `take_halves()` pumped against the taken channel. async fn channel_open_direct_tcpip( &mut self, channel: Channel, - _host_to_connect: &str, - _port_to_connect: u32, + host_to_connect: &str, + port_to_connect: u32, _originator_address: &str, _originator_port: u32, reply: ChannelOpenHandle, @@ -622,12 +665,28 @@ 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(()); + #[cfg(feature = "tunnel")] + { + let target = u16::try_from(port_to_connect).ok().map(|port| { + alktunnels::params::SubstrateAddr::HostPort { + host: host_to_connect.to_owned(), + port, + } + }); + direct_open_dispatch( + self, + ROW, + channel, + reply, + target, + alktunnels::params::Substrate::Tcp, + ) + .await; + } + #[cfg(not(feature = "tunnel"))] + { + let _ = (host_to_connect, port_to_connect); } - tunnel_row_stub(ROW, channel.id().number(), reply).await; Ok(()) } @@ -662,10 +721,15 @@ impl russh::server::Handler for SessionHandler { Ok(()) } - /// The `direct-streamlocal` open gate (`tunnel`): the same - /// gate/cap/stub split as the tcp rows. The runtime-Unix gate (OQ-11) - /// gates *serving* on Unix; this wave the stub declines - /// everywhere, so the runtime gate has nothing to ride yet. + /// The `direct-streamlocal` open gate (`tunnel`) and its route-1 + /// bridge dispatch: the `direct-tcpip` shape with the OQ-11 + /// **runtime-Unix gate** — the row compiles wherever `tunnel` + /// compiles (universal compile, no per-platform feature fork), but + /// off-Unix it declines at the open **before any fabric call** + /// (never a partial bridge — the gate precedes the unset-handle + /// check, the identity guard, and the cap admission), and on-Unix + /// it rides the same bridge with `SubstrateAddr::Path` + + /// `Substrate::Unix`. async fn channel_open_direct_streamlocal( &mut self, channel: Channel, @@ -679,12 +743,33 @@ 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(()); + #[cfg(feature = "tunnel")] + { + #[cfg(unix)] + { + let target = Some(alktunnels::params::SubstrateAddr::Path { + path: _socket_path.to_owned(), + }); + direct_open_dispatch( + self, + ROW, + channel, + reply, + target, + alktunnels::params::Substrate::Unix, + ) + .await; + } + #[cfg(not(unix))] + { + observe::decline_open(ROW, "unix-unavailable", channel.id().number()); + reply.reject(dispatch::connect_failed_open_decline()).await; + } + } + #[cfg(not(feature = "tunnel"))] + { + let _ = _socket_path; } - tunnel_row_stub(ROW, channel.id().number(), reply).await; Ok(()) } @@ -1286,13 +1371,83 @@ fn spawn_tty_row( let _ = session.channel_success(channel); } -/// The tunnel row's seam stub: the explicit `ConnectFailed` reject -/// ([`crate::dispatch::tunnel_seam_stub_decline`]) — wave 3 lands the -/// route-1 bridge here (`TunnelSession::open_direct` → accept only -/// after the fabric open succeeds → `take_halves` pump). The -/// open-level decline event is emitted here: the stub's decline -/// reason is the `connect-failed` class ADR-006 decision 1 pins. +/// The direct tunnel rows' dispatch (`w3-tunnel-direct`): the +/// pre-admit decline classes first — **decline first, alloc never** +/// applies to the cap budget too, so every consumer is checked before +/// the admission is taken. Order (the recorded admission mechanics: +/// the admission is taken at the hook — the packet-loop's single +/// admit-writer site, no race window against the cap check — and held +/// across the fabric open): +/// +/// 1. **The unset fabric handle decline** (`seam-unconfigured`, the +/// unset-declines posture — the feature alone configures nothing): +/// `ConnectFailed` ([`crate::dispatch::connect_failed_open_decline`]) +/// — the bridge cannot be attempted at all. +/// 2. **The `no-identity` guard** — unreachable on the wire (russh +/// authorizes before channels exist) and fail-closed: never +/// anonymous serving (the git/tty rows' shape, discharged here for +/// the tunnel row — the parked `w2-auth-bridge` item; the +/// `set_identity` mechanics the rides records live in +/// [`crate::tunnel_map`]). +/// 3. **The impossible-target decline** (`invalid-target` — an +/// out-of-range wire port): `ConnectFailed`. +/// +/// Then the cap admission ([`Self::admit_channel_open`]; beyond it → +/// `cap-exceeded`), and the spawn: the admitted count rides into +/// [`crate::tunnel_map::spawn_direct_row`], which releases it exactly +/// once on the failure path (no `channel_close` can ever fire for a +/// rejected open) and never on the success path (whose close fires +/// [`Self::channel_close`] when the established channel closes). +#[cfg(feature = "tunnel")] +async fn direct_open_dispatch( + handler: &SessionHandler, + row: &'static str, + channel: Channel, + reply: ChannelOpenHandle, + target: Option, + substrate: alktunnels::params::Substrate, +) { + let channel_number = channel.id().number(); + let Some(client) = handler.serve.channel_client.clone() else { + observe::decline_open(row, "seam-unconfigured", channel_number); + reply.reject(dispatch::connect_failed_open_decline()).await; + return; + }; + if handler.resolved_identity().is_none() { + observe::decline_open(row, "no-identity", channel_number); + reply + .reject(russh::ChannelOpenFailure::AdministrativelyProhibited) + .await; + return; + } + let Some(target) = target else { + observe::decline_open(row, "invalid-target", channel_number); + reply.reject(dispatch::connect_failed_open_decline()).await; + return; + }; + if !handler.admit_channel_open() { + observe::decline_open(row, "cap-exceeded", channel_number); + reply.reject(dispatch::cap_exceeded_open_decline()).await; + return; + } + tunnel_map::spawn_direct_row( + row, + channel, + reply, + target, + substrate, + client, + 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::tunnel_seam_stub_decline()).await; + reply.reject(dispatch::connect_failed_open_decline()).await; } diff --git a/src/lib.rs b/src/lib.rs index 2ef719a..a7d82a6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -39,6 +39,8 @@ mod observe; mod serve; #[cfg(feature = "session")] mod tty_map; +#[cfg(feature = "tunnel")] +mod tunnel_map; pub use adapter::SshAdapter; #[cfg(feature = "git")] diff --git a/src/observe.rs b/src/observe.rs index ff5d2cf..87bbd95 100644 --- a/src/observe.rs +++ b/src/observe.rs @@ -64,10 +64,15 @@ //! - `reason` — the decline cause, mirroring the architecture's own //! classifications: `"unfeatured"` (feature-gate absence), //! `"never-feature"` (ADR-003 §5 rows), `"cap-exceeded"`, -//! `"connect-failed"` (the tunnel seam stub's `ConnectFailed`), +//! `"connect-failed"` (`w3-tunnel-direct`: the tunnel rows' +//! cannot-bridge class — the failed fabric open, the wave-2 seam +//! stub's reason surviving as the real decline; also the +//! `forwarded-tcpip` stub's standing decline class), //! `"seam-unconfigured"` (the config-driven row decline — the git -//! row and, now, the session rows: feature on, seam unset; -//! `w3-git-mapping` / `w3-tty-session`), +//! row and, now, the session rows (feature on, seam unset; +//! `w3-git-mapping` / `w3-tty-session`) and the tunnel rows' absent +//! fabric handle even under the feature (`w3-tunnel-direct`; the +//! unset-declines posture)), //! `"shell-unconfigured"` / `"wrapper-unconfigured"` (`w3-tty- //! session`: the OQ-08 unset-field rows — feature on, the //! serve-config field unset, never a fallback), @@ -79,8 +84,15 @@ //! the git row's validation gate; no service token or repo string //! ever rides — only the vocabulary value), //! `"exec-oversize"` (the pre-parse `max_exec_command_len` cap), +//! `"invalid-target"` (`w3-tunnel-direct`: the direct rows' +//! impossible wire target — an out-of-range `direct-tcpip` port), +//! `"unix-unavailable"` (`w3-tunnel-direct`: the OQ-11 runtime-Unix +//! gate — the off-Unix `direct-streamlocal` decline, answered +//! before any fabric call), //! `"no-identity"` / `"channel-not-held"` (the git row's serving- -//! path guards — unreachable in practice, fail-closed), +//! 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 //! auth class `"method-unimplemented"` / `"cert-unsupported"` / @@ -239,7 +251,7 @@ pub(crate) fn decline_serving( /// Spawned-path started — emitted as the spawned task's first action, /// so "started" is the task's own honest start, not the hook's spawn /// call. The mapping tasks' serving paths inherit this shape. -#[cfg(any(feature = "git", feature = "session"))] +#[cfg(any(feature = "git", feature = "session", feature = "tunnel"))] pub(crate) fn serving_path_started( row: &'static str, channel_id: u32, @@ -257,7 +269,7 @@ pub(crate) fn serving_path_started( /// Spawned-path ended — the serving future finished (cleanly, or with /// the decline it was wired to answer); emitted *after* the path's /// wire reporting has completed. -#[cfg(any(feature = "git", feature = "session"))] +#[cfg(any(feature = "git", feature = "session", feature = "tunnel"))] pub(crate) fn serving_path_ended( row: &'static str, channel_id: u32, @@ -278,6 +290,9 @@ pub(crate) fn serving_path_ended( /// the family's recoverable-failure level. Only russh protocol errors /// ride the `error` field (§Safety posture); the mapping tasks emit /// this from their real paths when serving fails before reporting. +/// (The tunnel bridge never calls this: its wire reporting is +/// open-level — `accept`/`reject` cannot fail send-wise — so the fn +/// stays gated to the git/session rows.) #[cfg(any(feature = "git", feature = "session"))] pub(crate) fn serving_path_failed( row: &'static str, diff --git a/src/tunnel_map.rs b/src/tunnel_map.rs new file mode 100644 index 0000000..f044d47 --- /dev/null +++ b/src/tunnel_map.rs @@ -0,0 +1,161 @@ +//! The `direct-tcpip` / `direct-streamlocal` mapping — the tunnel +//! branch's route-1 bridge (`w3-tunnel-direct`, ADR-006 decision 1): +//! the adapter-side bridge POC-3 stage C proved, re-landed as contract +//! against published alktunnels 0.4.1. +//! +//! ## The route-1 bridge (ADR-006 decision 1) +//! +//! alkssh holds the assembly-supplied `Arc` into the +//! family fabric (the serve-config's `channel_client` — the wave-1 +//! config field's consuming code). Per accepted `direct-tcpip` / +//! `direct-streamlocal` open, the target/substrate are constructed +//! from the SSH open arguments (`direct-tcpip`(host, port) → +//! `SubstrateAddr::HostPort` + `Substrate::Tcp`; +//! `direct-streamlocal`(socket path) → `SubstrateAddr::Path` + +//! `Substrate::Unix` — the mapping.md row's treatment) and the flow is +//! exactly the POC recipe: **`TunnelSession::open_direct(&client, +//! target, substrate)` → accept only after the fabric open succeeds** +//! (`ChannelOpenHandle::accept` consumes the handle — the reject shape +//! is only possible pre-accept, which is what makes accept-after-open +//! the only honest flow) → on success `take_halves()` pumped against +//! the taken channel's stream ([`alkcall::channels::pump::pump_bidi`], +//! the alkcall primitive the POC recipe used) on the spawned task, per +//! the spawn-and-return constraint. The open hook returns fast with +//! the fabric open in flight. +//! +//! **Fabric failure → the explicit `ConnectFailed` reject** — the +//! wave-2 seam stub's reason surviving as the real decline +//! ([`crate::dispatch::connect_failed_open_decline`]; the stub itself +//! is retired). The channel is never established, so the SSH client +//! observes a clean, channel-scoped open refusal (POC-1 §5 posture) +//! — vanilla clients retry-observe a down forward instead of a hang. +//! +//! ## The runtime-Unix gate (OQ-11) +//! +//! `direct-streamlocal` compiles wherever `tunnel` compiles (universal +//! compile — no per-platform feature fork); **serving is runtime-Unix- +//! gated**: off-Unix a clean open-decline **before any fabric call** +//! (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. +//! +//! ## The identity ride (the visible-surface invariant's gate) +//! +//! The resolved SSH identity is the auth bridge's ride-along in +//! handler state (`auth_publickey` stores; the row dispatches never +//! serve anonymously — the `no-identity` guard, unreachable on the +//! wire because russh authorizes before channels exist, fail-closed +//! like the git/tty rows' `w2-auth-bridge` parked item discharged +//! there). The guard gates every tunnel open **at the door**; the +//! 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. +//! +//! **`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. + +use std::sync::Arc; + +use alkcall::channels::client::ChannelClient; +use alkcall::channels::pump::pump_bidi; +use alktunnels::consumer::TunnelSession; +use alktunnels::params::{Substrate, SubstrateAddr}; +use russh::Channel; +use russh::server::ChannelOpenHandle; + +use crate::channel_cap::ChannelCap; +use crate::serve; +use crate::{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 +/// the same shape). +pub(crate) type WireChannel = Channel; + +/// The spawned route-1 bridge (`w3-tunnel-direct`, the POC-3 §C +/// recipe): the fabric open runs here — accept only after it succeeds, +/// `take_halves()` pumped against the taken channel on completion, and +/// on failure the explicit `ConnectFailed` reject **with the cap count +/// released** (the recorded release mechanism — the admitted open this +/// task holds was never established, so no `channel_close` can ever +/// fire; the count recovers here, in the task that knows the failure, +/// exactly once: a success never releases). The channel closes on +/// drop (russh's `ChannelCloseOnDrop`) after the pump ends. +pub(crate) async fn serve_direct_row( + row: &'static str, + channel: WireChannel, + reply: ChannelOpenHandle, + target: SubstrateAddr, + substrate: Substrate, + client: Arc, + cap: Arc, +) { + let channel_number = channel.id().number(); + observe::serving_path_started(row, channel_number, None); + + match TunnelSession::open_direct(&client, target, substrate).await { + Ok(session) => { + reply.accept().await; + let taken = session.take_halves(); + pump_bidi(channel.into_stream(), taken.read, taken.write).await; + observe::serving_path_ended(row, channel_number, None); + } + Err(_) => { + cap.release(); + observe::decline_open(row, "connect-failed", channel_number); + reply.reject(dispatch::connect_failed_open_decline()).await; + observe::serving_path_ended(row, channel_number, None); + } + } +} + +/// The open hook's spawn step (ADR-006 decision 4): the handler clones +/// the cap `Arc` (the failure-path release's handle — the count was +/// admitted at the hook and is held across the fabric open) and hands +/// the rest over; the hook returns fast with the fabric open in +/// flight. +pub(crate) fn spawn_direct_row( + row: &'static str, + channel: WireChannel, + reply: ChannelOpenHandle, + target: SubstrateAddr, + substrate: Substrate, + client: Arc, + cap: Arc, +) { + serve::spawn_serving_path(serve_direct_row( + row, channel, reply, target, substrate, client, cap, + )); +} diff --git a/tasks/implementation/w3-tunnel-direct.md b/tasks/implementation/w3-tunnel-direct.md index 05ad7be..30322ff 100644 --- a/tasks/implementation/w3-tunnel-direct.md +++ b/tasks/implementation/w3-tunnel-direct.md @@ -1,7 +1,7 @@ --- id: w3-tunnel-direct name: Wave 3 — tunnel leg 1 (the route-1 bridge — direct-tcpip + streamlocal) -status: pending +status: completed depends_on: [w3-tty-session] scope: moderate risk: medium @@ -79,25 +79,25 @@ adapter-side route-1 bridge for `direct-tcpip` and `direct-streamlocal` ## Acceptance Criteria -- [ ] `direct-tcpip` serves through the route-1 bridge: fabric open → +- [x] `direct-tcpip` serves through the route-1 bridge: fabric open → accept → `take_halves()` pump on the spawned task; failure → the explicit `ConnectFailed` reject, the stub retired; the open hook returns fast -- [ ] The cap admission across the fabric window pinned (admit at the +- [x] The cap admission across the fabric window pinned (admit at the hook, release on fabric failure inside the spawned task — no leaked count, no double-admit) -- [ ] The identity propagates to the channels connection once +- [x] The identity propagates to the channels connection once (`set_identity` mechanics + hook placement decided and recorded); the wave-2 `auth_succeeded` item discharged -- [ ] `direct-streamlocal` carries the runtime-Unix gate +- [x] `direct-streamlocal` carries the runtime-Unix gate (compile-everywhere, clean open-decline off-Unix before any fabric call) -- [ ] The per-build decline-matrix rows green through the harness +- [x] The per-build decline-matrix rows green through the harness (unfeatured/unset/fabric-failure/cap/serve classes), each with its contract stamp; the `ssh -L` smoke row -- [ ] The `alktunnels` optional dep lands, pin re-verified; +- [x] The `alktunnels` optional dep lands, pin re-verified; `tunnel`-off builds carry no alktunnels types in code paths -- [ ] `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 @@ -122,4 +122,128 @@ adapter-side route-1 bridge for `direct-tcpip` and `direct-streamlocal` ## Notes -> To be filled at implementation time. \ No newline at end of file +> Recorded at implementation time (2026-10-29). All decisions live in +> the module docs (`src/tunnel_map.rs`, `src/channel_cap.rs`, +> `src/handler.rs`) and are summarized here for the gate. + +- **The cap release mechanism (the task's first decide-and-record)** — + `ChannelCap` became a `parking_lot::Mutex` shared as an + `Arc`: **admissions stay packet-loop-only** (every `admit` call site + is an open hook — the check-and-increment is one atomic section + against the shared writer), while **releases have exactly two + disjoint sites** — the `channel_close` hook for *established* + channels, and *inside the spawned bridge task* on a fabric-open + failure (the admitted-but-rejected shape: ADR-006 decision 1 rejects + the open, so no `CHANNEL_CLOSE` can ever cycle back, and the count + must be released where the failure is known — before the reject is + answered, so the very next open observes the recovered budget). + Both release sites saturate. Pinning rows: + `a_failed_bridge_releases_the_cap_count_before_the_next_open` + (failure-then-serve under `max_concurrent_channels: 1` — a leaked + count would answer `AdministrativelyProhibited`) and + `the_cap_refuses_beyond_the_ceiling_and_recovers_on_close` + (refusal + close-hook recovery). + +- **`set_identity` mechanics (the task's second decide-and-record; the + `w2-auth-bridge` parked item discharged)** — the alkhub `hub_leg.rs` + one-time-set precedent 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 the task named + (`ChannelClient::take_call_connection()` → + `CallConnection::connection()` → `Connection::set_identity`) was + **checked and rejected for this door**, verified against alkcall + 0.10.1's docs: 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 — the relay note's "the detach is wrong for hubs" + applies a fortiori. 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`) — identity crossing the alkcall wire is not + a protocol concept. Decision: the composition — which identity the + fabric leg dials with, against which the fabric's ACL gates + alkssh's mapped opens — is **assembly policy** (the assembly + supplies both the injected `IdentityProvider` and the fabric leg); + alkssh structurally refuses anonymous serving (the `no-identity` + open guard, unreachable on the wire — russh authorizes before + channels exist — fail-closed like the git/tty rows) so no + SSH-shaped exception to the visible-surface invariant exists, and + the harness rows witness the composed gate end to end: + `the_fabric_acl_refuses_an_unscoped_channels_identity` (the + fabric's `AccessControl::check` refusing a scopeless channels + identity — the establisher never dialed) and + `direct_tcpip_serves_through_the_route_1_bridge`'s identity-witness + assertion (the establisher's per-call auth seeing the channels + identity — POC-3 §C.3). **Hook placement**: the identity store + stays the `auth_publickey` ride-along in handler state; the + tunnel rows' dispatch reads it; `auth_succeeded` stays + unimplemented (russh default) — nothing rides it that the store + does not carry. **Upstream ask booked for the wave-3 gate**: a + non-detaching post-construction identity seam on `ChannelClient` + (or per-session fabric legs) would let the door attach per-peer + identities to a shared leg; nothing is silently absorbed. + +- **Wire decline classes (recorded)** — two vanilla-observable classes + at the open: *policy refusals* → `AdministrativelyProhibited` + (unfeatured, cap-exceeded, the `no-identity` guard) vs + *cannot-bridge* → `ConnectFailed` + ([`dispatch::connect_failed_open_decline`] — the wave-2 stub's + reason surviving as the real decline): the failed fabric open, the + absent fabric handle under the feature (`seam-unconfigured`, the + unset-declines posture), an impossible wire target + (`invalid-target`, an out-of-range port), and the off-Unix + `direct-streamlocal` OQ-11 gate (`unix-unavailable`). The classes + are distinguishable in the event vocabulary, which gained + `invalid-target` and `unix-unavailable`. + +- **Admission mechanics** — the pre-admit decline classes are checked + in order (unset handle → no-identity → invalid target), so every + pure-decline path consumes nothing; the admission is taken at the + hook and held across the fabric open; the success release rides the + established channel's close. `forwarded-tcpip` kept its + pre-admit `has_room` cap gate and stub — its `-R` data-leg flow is + `w3-tunnel-listen` material (the harness's russh client cannot + initiate the type; its wire rows are acceptor vocabulary). + +- **The fabric double** — the POC-3 §C.4 in-process recipe as test + composition (`wire_fabric` in `tests/tunnel_mapping.rs`): + `ChannelsAdapter` + `InstallChannelZero` + `ChannelOperations` + + `register_tunnel_direct_openable` + `Dispatcher::serve_single_stream` + over a tokio duplex, the connect side the `Arc` the + serve-config holds. The hand-rolled dial/dial-fail seam is an + in-process map of duplex halves keyed by the establisher-rendered + target string (consume-once — the cap-recovery row reads the + consume-once dial as its release signal), with an atomics call + counter for the "never called" assertions (the ACL row and the + OQ-11 row); the smoke's dial seam is a real-TCP closure (alktunnels' + `local` feature stays off everywhere — the shipped crate and the + tests). + +- **The real-tool smoke** — tooling posture decided like the git task + (conditional skip with a printed reason when `ssh`/`curl` is + absent), and it ran green here: one `ssh -N -L` (OpenSSH client) + through a TCP-mounted adapter → the route-1 bridge → the fabric + double with a real-TCP dial → a local HTTP endpoint, curled through + the forwarded port (body parity asserted). 8/8 rows green in + `tests/tunnel_mapping.rs`. + +- **Pins re-verified against crates.io (2026-10-29)**: alktunnels + 0.4.1 is `max_stable_version`/`newest_version`, MIT OR Apache-2.0, + `rust_version = 1.88` ≤ ours 1.89, alkcall `0.10.1` exact + (single-versioned against our `alkcall = "0.10"`); wired + `tunnel = ["dep:alktunnels"]`, `default-features = false` (the + `local` feature never enabled). `tunnel`-off builds carry no + alktunnels types in code paths (dep-to-feature gating; verified by + the per-feature test rows). + +- **Verification (all green)**: `cargo test` (default + `--features + tunnel` / `git` / `session` / `git,session` / `tunnel,git` / + `tunnel,session` / `--all-features`), `cargo clippy --all-targets + -- -D warnings` ×2 (default + `--all-features`), `cargo fmt + --check`, `cargo doc --no-deps` 0 warnings, `taskgraph validate` + (26 tasks). \ No newline at end of file diff --git a/tests/bounded_resources.rs b/tests/bounded_resources.rs index 34c5929..116eb2d 100644 --- a/tests/bounded_resources.rs +++ b/tests/bounded_resources.rs @@ -260,9 +260,17 @@ async fn zero_cap_refuses_every_open_and_poisons_nothing() { session.unmount().await; } +/// - *Contract stamp:* `w3-tunnel-direct`'s recorded admission +/// mechanics (decline-first-alloc-never applies to the cap budget +/// too): the tunnel row's pre-admit classes — here the **absent +/// fabric handle** under the feature — answer before the admission +/// is taken, so under an exhausted cap the absent-handle class +/// (`ConnectFailed`, the cannot-bridge wire class) wins over the cap +/// reason. The cap class on a *configured* direct row rides +/// tests/tunnel_mapping.rs's fabric double. #[cfg(all(feature = "tunnel", any(feature = "session", feature = "git")))] #[tokio::test] -async fn an_exhausted_cap_answers_the_tunnel_row_with_the_cap_reason() { +async fn the_tunnel_rows_pre_admit_class_answers_before_the_cap_gate() { use common::harness::{open_direct_tcpip, open_session_channel}; let session = authenticated_with(Limits { @@ -277,11 +285,11 @@ async fn an_exhausted_cap_answers_the_tunnel_row_with_the_cap_reason() { let err = open_direct_tcpip(&session.client, "internal.example", 5432) .await - .expect_err("the cap gates the tunnel row before the seam stub"); + .expect_err("the pre-admit class answers before the cap gate is even consulted"); open_failure( err, - ChannelOpenFailure::AdministrativelyProhibited, - "the cap-exhausted tunnel open", + ChannelOpenFailure::ConnectFailed, + "the absent-handle decline precedes the admission", ); session.unmount().await; diff --git a/tests/dispatch_core.rs b/tests/dispatch_core.rs index 453a640..49edfce 100644 --- a/tests/dispatch_core.rs +++ b/tests/dispatch_core.rs @@ -554,9 +554,11 @@ mod session_build { // --------------------------------------------------------------------------- // tunnel-only build: session channels decline (auto-reject); the -// tunnel rows route to the tunnel seam stub even with the feature on -// (the route-1 bridge is wave 3) — mapping.md §Decline matrix (tunnel -// row, as landed this wave). +// tunnel direct rows decline on the **absent handle** even with the +// feature on (the route-1 bridge is `w3-tunnel-direct`: the pre-admit +// unset-handle class — the unset-declines posture; the fabric double's +// serve/cap/fabric-failure rows ride tests/tunnel_mapping.rs) — +// mapping.md §Decline matrix (tunnel row). // --------------------------------------------------------------------------- #[cfg(all(feature = "tunnel", not(feature = "git"), not(feature = "session")))] @@ -582,30 +584,34 @@ mod tunnel_build { session.unmount().await; } - /// - *Contract stamp:* the task's channel-open gate (featured - /// tunnel rows route to the seam stub; the stub's decline is - /// the explicit `ConnectFailed` reject ADR-006 decision 1 pins - /// for failed fabric opens). + /// - *Contract stamp:* mapping.md §Channel-type rows + the + /// unset-declines posture (`w3-seam-config` — the feature alone + /// configures nothing): the direct rows decline on the **absent + /// fabric handle** even with `tunnel` on — + /// [`connect failed`], the cannot-bridge class ADR-006 decision + /// 1 pins (`w3-tunnel-direct`'s recorded wire classes: policy + /// refusals are administratively-prohibited, cannot-bridge is + /// connect-failed). #[tokio::test] - async fn tunnel_rows_route_to_the_seam_stub_in_a_featured_build() { + async fn tunnel_direct_rows_decline_seam_unconfigured_without_a_handle() { let session = authenticated().await; let err = open_direct_tcpip(&session.client, "internal.example", 5432) .await - .expect_err("the featured direct-tcpip row is a declining seam stub"); + .expect_err("the direct-tcpip row declines on the absent handle"); open_failure( err, ChannelOpenFailure::ConnectFailed, - "the direct-tcpip seam stub", + "the direct-tcpip unset-handle decline", ); let err = open_direct_streamlocal(&session.client, "/run/alkssh.sock") .await - .expect_err("the featured direct-streamlocal row is a declining seam stub"); + .expect_err("the direct-streamlocal row declines on the absent handle"); open_failure( err, ChannelOpenFailure::ConnectFailed, - "the direct-streamlocal seam stub", + "the direct-streamlocal unset-handle decline", ); session.unmount().await; @@ -687,29 +693,32 @@ mod all_features { session.unmount().await; } - /// - *Contract stamp:* the task's channel-open gate (the tunnel - /// rows are seam stubs even with the feature on — the route-1 - /// bridge is wave 3; `ConnectFailed` is the stub's class). + /// - *Contract stamp:* the unset-declines posture (`w3-seam-config`, + /// carried through `w3-tunnel-direct`: the route-1 bridge's + /// pre-admit unset-handle class) — the direct rows decline on + /// the **absent fabric handle** even with the full feature graph + /// on (`ConnectFailed`, the cannot-bridge class); the served + /// surface rides tests/tunnel_mapping.rs's fabric double. #[tokio::test] - async fn tunnel_rows_route_to_the_seam_stub_with_all_features() { + async fn tunnel_direct_rows_decline_seam_unconfigured_without_a_handle_with_all_features() { let session = authenticated().await; let err = open_direct_tcpip(&session.client, "internal.example", 5432) .await - .expect_err("the full-featured direct-tcpip row is a declining seam stub"); + .expect_err("the full-featured direct-tcpip row declines on the absent handle"); open_failure( err, ChannelOpenFailure::ConnectFailed, - "the direct-tcpip seam stub", + "the direct-tcpip unset-handle decline", ); let err = open_direct_streamlocal(&session.client, "/run/alkssh.sock") .await - .expect_err("the full-featured direct-streamlocal row is a declining seam stub"); + .expect_err("the full-featured direct-streamlocal row declines on the absent handle"); open_failure( err, ChannelOpenFailure::ConnectFailed, - "the direct-streamlocal seam stub", + "the direct-streamlocal unset-handle decline", ); session.unmount().await; diff --git a/tests/observe.rs b/tests/observe.rs index f7cc589..35eb757 100644 --- a/tests/observe.rs +++ b/tests/observe.rs @@ -673,25 +673,26 @@ mod tunnel_build { use super::*; use common::harness::open_session_channel; - /// - *Contract stamp:* ADR-006 decision 1 (accept only after the - /// fabric open succeeds; failure maps to the explicit - /// `ConnectFailed` — this wave the seam stub; the event pins the - /// decline class the wave-3 bridge replaces). + /// - *Contract stamp:* the unset-declines posture (`w3-seam-config`, + /// carried through `w3-tunnel-direct`: the route-1 bridge's + /// pre-admit classes) — the direct row's absent-handle decline + /// emits the `seam-unconfigured` open-class event (the fabric + /// double's real decline events ride tests/tunnel_mapping.rs). #[tokio::test] - async fn tunnel_seam_stub_decline_emits_the_connect_failed_event() { + async fn tunnel_direct_row_absent_handle_decline_emits_the_event() { let session = authenticated("tunnel-stub-row").await; common::harness::open_direct_tcpip(&session.client, "internal.example", 5432) .await - .expect_err("the tunnel row declines at the seam stub"); + .expect_err("the tunnel row declines on the absent handle"); let event = await_event(&[ ("row", "direct-tcpip"), ("decline_class", "open"), - ("reason", "connect-failed"), + ("reason", "seam-unconfigured"), ]) .await; assert_decline( &event, - &[("row", "direct-tcpip"), ("reason", "connect-failed")], + &[("row", "direct-tcpip"), ("reason", "seam-unconfigured")], ); session.unmount().await; } diff --git a/tests/tunnel_mapping.rs b/tests/tunnel_mapping.rs new file mode 100644 index 0000000..3293a14 --- /dev/null +++ b/tests/tunnel_mapping.rs @@ -0,0 +1,876 @@ +//! The `w3-tunnel-direct` contract rows — the tunnel branch served for +//! real: the route-1 bridge (ADR-006 decision 1) lands +//! `direct-tcpip`/`direct-streamlocal` through the published alktunnels +//! 0.4.1 seam (`TunnelSession::open_direct` → accept only after the +//! fabric open succeeds → `take_halves()` pumped against the taken +//! channel), observed through the loopback harness against the **fabric +//! double** (`w3-tunnel-direct`'s recorded test composition: the +//! POC-3 §C.4 in-process recipe — `ChannelsAdapter` + +//! `InstallChannelZero` + the alktunnels openables + a hand-rolled +//! dial/dial-fail seam, hand-rolled here, never a sibling-checkout +//! path). +//! +//! The decline-matrix classes *per feature build* stay in +//! `tests/dispatch_core.rs` (unfeatured rows in the featureless/git/ +//! session builds; the absent-handle rows in the tunnel builds); this +//! binary adds the featured-build rows the fabric double makes +//! observable: the serve class (data round trip, both substrates), the +//! fabric-failure class (`ConnectFailed`), the cap classes (refusal + +//! the release mechanism: an admitted-but-rejected bridge releases its +//! count, a closed serving channel recovers), the composed identity +//! gate (the fabric's `AccessControl::check` refusing an unscoped +//! channels identity — the visible-surface invariant's alkcall half, +//! with the establisher never reached), the establisher's identity +//! witness (the POC-3 §C.3 shape), the OQ-11 off-Unix gate, and the +//! real-tool `ssh -L` smoke. +//! +//! The fabric double's dial seam is in-process by design: the dial +//! target is matched against a map of pre-created duplex halves keyed +//! by the rendered target string (`host:port` / socket path — exactly +//! the shape the establisher hands the `DialFn`), so the serve rows +//! are fully deterministic (no sockets on the target substrate), while +//! the real-tool smoke hand-rolls a real-TCP dial closure against a +//! real loopback endpoint (the smoke exercises the substrate for real; +//! tooling posture recorded like the git task — skipped-with-reason +//! when `ssh`/`curl` is unavailable, never silently green). + +#![allow(dead_code)] + +mod common; + +#[cfg(feature = "tunnel")] +mod serving { + use std::collections::HashMap; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + + use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero}; + use alkcall::channels::client::ChannelClient; + use alkcall::channels::operations::{ChannelCore, ChannelOperations}; + 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::producer::{ + DialFn, IdentityWitness, TargetHandle, TunnelEstablishError, + register_tunnel_direct_openable, + }; + use russh::client::Msg; + use russh::{Channel, ChannelMsg, ChannelOpenFailure, Error}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + use super::common::harness::{ + LoopbackSession, drive, open_direct_streamlocal, open_direct_tcpip, + serve_config as harness_serve_config, + }; + use super::common::{MemHostKeyPolicy, MemIdentityProvider, fixture, identity}; + use alkssh::{Limits, SshAdapter}; + + type TargetMap = Arc>>; + + /// The in-process fabric (`w3-tunnel-direct`'s recorded double, the + /// POC-3 §C.4 recipe): an `alk/channels` connection over a tokio + /// duplex whose accept side registers `ChannelOperations` + the + /// `channels/tunnel/direct` openable (the establisher's dial seam + /// injected), and whose connect side is the `Arc` a + /// tunnel-featured serve-config holds. The accept side's channel-0 + /// identity (`caller`) is what the fabric's + /// `AccessControl::check(peer_identity)` gates the mapped opens + /// against — the assembly-composed identity of the real deployment; + /// `witness` records what the establisher's per-call auth (alkcall + /// CF-006) saw. + pub(super) async fn wire_fabric( + dial: DialFn, + caller: Identity, + witness: IdentityWitness, + ) -> Arc { + let registry = Arc::new(OperationRegistry::new()); + let registry_for_hook = Arc::clone(®istry); + let witness_for_hook = Arc::clone(&witness); + let caller_for_hook = caller.clone(); + let dial_for_hook = Arc::clone(&dial); + + let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, _auth| { + let registry = Arc::clone(®istry_for_hook); + let witness = Arc::clone(&witness_for_hook); + let caller = caller_for_hook.clone(); + let dial = Arc::clone(&dial_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); + // The serving-side identity set once at install — the + // hub_leg.rs one-time-set precedent the door's + // `set_identity` mechanics decision reads (the identity + // the serving dispatch resolves for the mapped opens' + // ACL). + let _ = channel0_conn.set_identity(caller); + let call_connection = Arc::new(CallConnection::new_single_stream( + channel0_conn, + Arc::clone(&writer), + )); + 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, + Some(witness), + ) + .expect("tunnel direct 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; + }); + + Arc::new( + ChannelClient::from_connection(consumer_conn) + .await + .expect("the fabric's channels client establishes"), + ) + } + + /// 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. + fn inproc_target(targets: &TargetMap, key: &str, capacity: usize) -> tokio::io::DuplexStream { + let (target_side, test_side) = tokio::io::duplex(capacity); + targets + .lock() + .unwrap_or_else(|e| e.into_inner()) + .insert(key.to_owned(), target_side); + test_side + } + + /// The hand-rolled dial seam (the assembly's injected `DialFn`, + /// doubled): an exact `host:port` / socket-path key pops the + /// pre-created half; an absent key is the dial-fail seam + /// (`TunnelEstablishError::DialFailed`) — both the fabric-failure + /// rows' shape and the release rows' failure leg. The call counter + /// witnesses the "no fabric call" assertions. + fn inproc_dial(targets: TargetMap, calls: Arc) -> DialFn { + Arc::new(move |_substrate: Substrate, target: &str| { + let targets = Arc::clone(&targets); + let calls = Arc::clone(&calls); + let target = target.to_owned(); + Box::pin(async move { + calls.fetch_add(1, Ordering::SeqCst); + let half = targets + .lock() + .unwrap_or_else(|e| e.into_inner()) + .remove(&target) + .ok_or_else(|| { + TunnelEstablishError::DialFailed(format!( + "in-process double target `{target}` absent" + )) + })?; + let (read, write) = tokio::io::split(half); + Ok(TargetHandle { + read: Box::new(read), + write: Box::new(write), + }) + }) + }) + } + + /// The fabric caller identity with the direct op's scope — the + /// composed-gate identity the serving side resolves, so the ACL + /// admits the mapped opens. + fn scoped_caller() -> Identity { + identity("tunnel-caller", &[TUNNEL_DIRECT_SCOPE]) + } + + /// The tunnel build's authenticated loopback session over the + /// doubles, with the fabric client wired in: the fixture identity + /// resolves through the provider and the serve-config carries the + /// `channel_client` (the route-1 bridge's family-fabric entry); + /// `limits` rides the serve-config (the cap rows tune them). + async fn tunnel_session(client: Arc, limits: Limits) -> LoopbackSession { + 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(client); + let adapter = SshAdapter::new(serve) + .await + .expect("the tunnel build's adapter constructs"); + + let policy = Arc::new(MemHostKeyPolicy::with_default(true)); + let mut session = 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(), + "the fixture identity authenticates before every tunnel row" + ); + session + } + + /// The harness import that needs the loopback session's client + /// (kept near the top of the module for the rows' readability). + use super::common::harness::authenticate_publickey; + + /// Matches an open failure's SSH reason — the wire decline reason + /// the rows assert. + fn open_failure(err: Error, expected: ChannelOpenFailure, context: &str) { + assert!( + matches!(&err, Error::ChannelOpenFailure(reason) if *reason == expected), + "{context}: expected open decline with {expected:?}, got {err:?}" + ); + } + + /// The serve row's data leg: the SSH channel against the in-process + /// target — one request down `direct-tcpip`, one reply back (the + /// bidirectional pump of the route-1 bridge, the POC-3 §C.1 shape: + /// zero framing, both directions live). + async fn round_trip( + channel: &mut Channel, + mut target_side: tokio::io::DuplexStream, + request: &[u8], + reply: &[u8], + context: &str, + ) { + channel + .data_bytes(request.to_vec()) + .await + .expect("the SSH channel carries the request down"); + + let mut seen_request = vec![0u8; request.len()]; + tokio::time::timeout( + std::time::Duration::from_secs(15), + target_side.read_exact(&mut seen_request), + ) + .await + .expect("the request arrives through the bridge in time") + .expect("the pump delivers the request to the substrate"); + assert_eq!( + seen_request, request, + "{context}: the bridge pumps the SSH channel bytes unframed" + ); + + target_side + .write_all(reply) + .await + .expect("the target half writes the reply"); + target_side.flush().await.expect("the flush rides"); + + let mut seen_reply = Vec::new(); + tokio::time::timeout(std::time::Duration::from_secs(15), async { + while seen_reply.len() < reply.len() { + match channel.wait().await { + Some(ChannelMsg::Data { data }) => seen_reply.extend_from_slice(&data), + Some(_) => continue, + None => panic!("{context}: the channel ended before the reply arrived"), + } + } + }) + .await + .expect("the reply arrives through the bridge in time"); + seen_reply.truncate(reply.len()); + assert_eq!( + seen_reply, reply, + "{context}: the bridge pumps the substrate bytes back unframed" + ); + } + + /// - *Contract stamp:* ADR-006 decision 1 (the route-1 bridge: the + /// `direct-tcpip` open → `TunnelSession::open_direct` over a held + /// `ChannelClient` → accept only after the fabric open succeeds → + /// `take_halves()` pumped bidirectionally — the POC-3 §C.1 shape, + /// zero framing both directions; the open hook returns fast). + #[tokio::test] + async fn direct_tcpip_serves_through_the_route_1_bridge() { + let targets: TargetMap = Arc::default(); + let calls = Arc::new(AtomicUsize::new(0)); + let witness = Arc::new(tokio::sync::Mutex::new(None)); + let target_side = inproc_target(&targets, "127.0.0.1:36001", 4096); + let client = wire_fabric( + inproc_dial(Arc::clone(&targets), Arc::clone(&calls)), + scoped_caller(), + Arc::clone(&witness), + ) + .await; + let session = tunnel_session(client, Limits::default()).await; + + let mut channel = open_direct_tcpip(&session.client, "127.0.0.1", 36001) + .await + .expect("the featured direct-tcpip row serves through the bridge"); + round_trip( + &mut channel, + target_side, + b"GET /hello HTTP/1.1\r\nHost: target.example\r\n\r\n", + b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello", + "the direct-tcpip serve row", + ) + .await; + + let witnessed = witness.lock().await.clone(); + assert_eq!( + witnessed.as_deref(), + Some("tunnel-caller"), + "the direct establisher's per-call auth saw the channels connection's identity (CF-006, POC-3 §C.3's witness)" + ); + assert_eq!( + calls.load(Ordering::SeqCst), + 1, + "one fabric dial rode the open" + ); + + session.unmount().await; + } + + /// - *Contract stamp:* mapping.md §Channel-type rows + /// (`direct-streamlocal` → `SubstrateAddr::Path` + + /// `Substrate::Unix`, the route-1 bridge's second substrate; + /// ADR-006 decision 1). + #[cfg(unix)] + #[tokio::test] + async fn direct_streamlocal_serves_through_the_route_1_bridge_on_unix() { + let targets: TargetMap = Arc::default(); + let calls = Arc::new(AtomicUsize::new(0)); + let target_side = inproc_target(&targets, "/inproc/tunnel-target.sock", 4096); + let client = wire_fabric( + inproc_dial(Arc::clone(&targets), calls), + scoped_caller(), + Arc::new(tokio::sync::Mutex::new(None)), + ) + .await; + let session = tunnel_session(client, Limits::default()).await; + + let mut channel = open_direct_streamlocal(&session.client, "/inproc/tunnel-target.sock") + .await + .expect("the on-Unix direct-streamlocal row serves through the bridge"); + round_trip( + &mut channel, + target_side, + b"unix-request-bytes", + b"unix", + "the direct-streamlocal serve row", + ) + .await; + + session.unmount().await; + } + + /// - *Contract stamp:* ADR-006 decision 1 (fabric-open failure → + /// the explicit `ConnectFailed` reject — accept only after the + /// fabric open succeeds; the wave-2 stub's reason surviving as + /// the real decline; POC-3 §C.2's failure shape). + #[tokio::test] + async fn a_failed_fabric_open_rejects_the_open_connect_failed() { + let targets: TargetMap = Arc::default(); + let calls = Arc::new(AtomicUsize::new(0)); + let client = wire_fabric( + inproc_dial(targets, Arc::clone(&calls)), + scoped_caller(), + Arc::new(tokio::sync::Mutex::new(None)), + ) + .await; + let session = tunnel_session(client, Limits::default()).await; + + let err = open_direct_tcpip(&session.client, "unreachable.example", 9) + .await + .expect_err("the fabric open fails on the absent dial target"); + open_failure( + err, + ChannelOpenFailure::ConnectFailed, + "the fabric-failure direct-tcpip row", + ); + assert_eq!( + calls.load(Ordering::SeqCst), + 1, + "the bridge dialed the fabric exactly once — the SSH open carried the fabric's answer" + ); + + session.unmount().await; + } + + /// - *Contract stamp:* `w3-tunnel-direct`'s recorded cap release + /// mechanism (the admission is taken at the hook and held across + /// the fabric open; a fabric failure releases the count **inside + /// the spawned task** — the `channel_close` release point only + /// fires for established channels, so without this release a + /// failed bridge would leak budget): a failed bridge under a + /// full cap does not strand the budget — the next open of the + /// same session is admitted (it would answer + /// `AdministrativelyProhibited` if the count had leaked). + #[tokio::test] + async fn a_failed_bridge_releases_the_cap_count_before_the_next_open() { + let targets: TargetMap = Arc::default(); + let calls = Arc::new(AtomicUsize::new(0)); + let target_side = inproc_target(&targets, "127.0.0.1:36002", 512); + let client = wire_fabric( + inproc_dial(Arc::clone(&targets), calls), + scoped_caller(), + Arc::new(tokio::sync::Mutex::new(None)), + ) + .await; + let limits = Limits { + max_concurrent_channels: 1, + ..Limits::default() + }; + let session = tunnel_session(client, limits).await; + + let err = open_direct_tcpip(&session.client, "unreachable.example", 9) + .await + .expect_err("the failing bridge declines"); + open_failure( + err, + ChannelOpenFailure::ConnectFailed, + "the cap-release row's failing leg", + ); + + // The failure answer landed before the client saw it + // (release-then-reject inside the task), so the very next open + // of this same session observes the released count. + let mut channel = open_direct_tcpip(&session.client, "127.0.0.1", 36002) + .await + .expect("the next open is admitted — the failed bridge released its count"); + round_trip( + &mut channel, + target_side, + b"req", + b"rep", + "the cap-release row's serving leg", + ) + .await; + + session.unmount().await; + } + + /// - *Contract stamp:* AGENTS.md convention 9 / `overview.md` + /// §Bounded-resources policy + `ChannelCap`'s recorded + /// disjoint-release shape: a *closed serving* channel releases + /// through the `channel_close` hook — a full cap refuses with the + /// cap reason, and a recovered cap admits again. + #[tokio::test] + async fn the_cap_refuses_beyond_the_ceiling_and_recovers_on_close() { + let targets: TargetMap = Arc::default(); + let calls = Arc::new(AtomicUsize::new(0)); + let target_side = inproc_target(&targets, "127.0.0.1:36003", 512); + let client = wire_fabric( + inproc_dial(Arc::clone(&targets), calls), + scoped_caller(), + Arc::new(tokio::sync::Mutex::new(None)), + ) + .await; + let limits = Limits { + max_concurrent_channels: 1, + ..Limits::default() + }; + let session = tunnel_session(client, limits).await; + + let mut channel = open_direct_tcpip(&session.client, "127.0.0.1", 36003) + .await + .expect("the first open admits under the cap"); + round_trip( + &mut channel, + target_side, + b"q", + b"a", + "the cap-refusal row's serving channel", + ) + .await; + + let err = open_direct_tcpip(&session.client, "127.0.0.1", 36003) + .await + .expect_err("the second open rides beyond the cap"); + open_failure( + err, + ChannelOpenFailure::AdministrativelyProhibited, + "the cap-exceeded direct-tcpip row", + ); + + channel.close().await.expect("the serving channel closes"); + // The close cycles back through the packet loop, the + // established-channel release fires, and the cap recovers. The + // recovery is observable because the dial target is consumed + // once: the moment the cap admits again, the open proceeds + // onto the bridge (and fails at the fabric dial with the + // connect-failed class) — an `AdministrativelyProhibited` + // answer here would mean a leaked count. + tokio::time::timeout(std::time::Duration::from_secs(10), async { + let mut attempts = 0u32; + loop { + match open_direct_tcpip(&session.client, "127.0.0.1", 36003).await { + Ok(opened) => { + drop(opened); + return; + } + Err(Error::ChannelOpenFailure( + ChannelOpenFailure::AdministrativelyProhibited, + )) if attempts < 40 => { + attempts += 1; + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + Err(Error::ChannelOpenFailure( + ChannelOpenFailure::AdministrativelyProhibited, + )) => { + panic!("the cap never recovered after the close — the count leaked") + } + Err(Error::ChannelOpenFailure(_)) => { + // The open went past the cap answer and onto the + // bridge (the fabric dial's connect-failed) — the + // released count is what this row observes. + return; + } + Err(other) => panic!("unexpected open error during recovery: {other:?}"), + } + } + }) + .await + .expect("the cap recovery row resolves in time"); + + session.unmount().await; + } + + /// - *Contract stamp:* AGENTS.md convention 7 (the visible-surface + /// invariant's alkcall half): the fabric's + /// `AccessControl::check(peer_identity)` gates the mapped open — + /// a channels connection whose identity lacks the direct op's + /// scope is refused by the fabric's dispatch (`FORBIDDEN`), the + /// establisher never runs, and the door answers the SSH open + /// with the explicit `ConnectFailed`. The auth bridge's + /// SSH-resolved identity and the fabric leg's dial identity + /// compose at the assembly (the deployment policy) — this row is + /// the composed-gate witness through the double. + #[tokio::test] + async fn the_fabric_acl_refuses_an_unscoped_channels_identity() { + let targets: TargetMap = Arc::default(); + let calls = Arc::new(AtomicUsize::new(0)); + let witness = Arc::new(tokio::sync::Mutex::new(None)); + let _ = inproc_target(&targets, "127.0.0.1:36004", 512); + let client = wire_fabric( + inproc_dial(Arc::clone(&targets), Arc::clone(&calls)), + identity("tunnel-caller-no-scope", &["ssh"]), + Arc::clone(&witness), + ) + .await; + let session = tunnel_session(client, Limits::default()).await; + + let err = open_direct_tcpip(&session.client, "127.0.0.1", 36004) + .await + .expect_err("the unscoped channels identity is gated at the fabric"); + open_failure( + err, + ChannelOpenFailure::ConnectFailed, + "the fabric-ACL direct-tcpip row", + ); + assert_eq!( + calls.load(Ordering::SeqCst), + 0, + "the ACL check precedes the establisher — a refused open never dialed" + ); + + session.unmount().await; + } + + /// - *Contract stamp:* mapping.md §Channel-type rows (`direct-tcpip`'s + /// target is the SSH open arguments verbatim; an impossible wire + /// target — a port outside the u16 space — is a declined + /// pre-admission, `invalid-target`, connect-failed on the wire). + #[tokio::test] + async fn an_out_of_range_wire_port_declines_pre_admission() { + let targets: TargetMap = Arc::default(); + let calls = Arc::new(AtomicUsize::new(0)); + let client = wire_fabric( + inproc_dial(targets, Arc::clone(&calls)), + scoped_caller(), + Arc::new(tokio::sync::Mutex::new(None)), + ) + .await; + let session = tunnel_session(client, Limits::default()).await; + + let err = open_direct_tcpip(&session.client, "target.example", u32::from(u16::MAX) + 1) + .await + .expect_err("an out-of-range wire port declines"); + open_failure( + err, + ChannelOpenFailure::ConnectFailed, + "the invalid-target direct-tcpip row", + ); + assert_eq!( + calls.load(Ordering::SeqCst), + 0, + "decline-first, alloc never: an impossible target never reaches the fabric" + ); + + session.unmount().await; + } + + /// - *Contract stamp:* OQ-11 (the runtime-Unix gate: universal + /// compile, clean open-decline off-Unix **before any fabric + /// call** — never a partial bridge): this row compiles in this + /// binary only off-Unix and pins the gate with its dial-call + /// witness at zero. On-Unix builds ride the serve row above. + #[cfg(not(unix))] + #[tokio::test] + async fn direct_streamlocal_declines_off_unix_before_any_fabric_call() { + let targets: TargetMap = Arc::default(); + let calls = Arc::new(AtomicUsize::new(0)); + let _ = inproc_target(&targets, "/inproc/off-unix.sock", 512); + let client = wire_fabric( + inproc_dial(Arc::clone(&targets), Arc::clone(&calls)), + scoped_caller(), + Arc::new(tokio::sync::Mutex::new(None)), + ) + .await; + let session = tunnel_session(client, Limits::default()).await; + + let err = open_direct_streamlocal(&session.client, "/inproc/off-unix.sock") + .await + .expect_err("the off-Unix streamlocal open declines"); + open_failure( + err, + ChannelOpenFailure::ConnectFailed, + "the OQ-11 off-Unix direct-streamlocal row", + ); + assert_eq!( + calls.load(Ordering::SeqCst), + 0, + "the runtime gate precedes any fabric call — no partial bridge" + ); + + session.unmount().await; + } +} + +#[cfg(all(feature = "tunnel", unix))] +mod real_tool { + //! The real-tool smoke (`w3-tunnel-direct`): one `ssh -L` (+ curl) + //! through a TCP-mounted adapter against the fabric double — the + //! POC-3 §C.1 gate re-landed. Tooling posture (recorded, like the + //! git task's): the smoke requires `ssh`/`curl` in the environment; + //! when absent it is **skipped with a printed reason** + //! (skipped-with-reason at most, never silently green). + + use std::sync::Arc; + + use alktunnels::params::{Substrate, TUNNEL_DIRECT_SCOPE}; + use alktunnels::producer::{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; + + /// The smoke's dial seam: a real-TCP dial closure (hand-rolled — + /// the assembly's `DialFn` injected over the substrate for real; + /// alktunnels' `local` feature stays off). + fn tcp_dial() -> DialFn { + Arc::new(move |_substrate: Substrate, target: &str| { + let target = target.to_owned(); + Box::pin(async move { + let stream = tokio::net::TcpStream::connect(&target).await.map_err(|e| { + TunnelEstablishError::DialFailed(format!("dial `{target}`: {e}")) + })?; + let (read, write) = stream.into_split(); + Ok(TargetHandle { + read: Box::new(read), + write: Box::new(write), + }) + }) + }) + } + + /// A tiny HTTP/1.1 endpoint (the POC's shape) — the substrate the + /// forwarded traffic dials through the fabric. + async fn spawn_http_endpoint() -> (std::net::SocketAddr, String) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("the smoke listener binds"); + let addr = listener.local_addr().expect("local addr"); + let body = "alkssh w3-tunnel-direct through the fabric\n".to_owned(); + let body2 = body.clone(); + tokio::spawn(async move { + loop { + let Ok((mut conn, _)) = listener.accept().await else { + return; + }; + let body = body2.clone(); + tokio::spawn(async move { + let mut buf = vec![0u8; 8192]; + let _ = conn.read(&mut buf).await; + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + body + ); + let _ = conn.write_all(head.as_bytes()).await; + let _ = conn.shutdown().await; + }); + } + }); + (addr, body) + } + + #[tokio::test] + async fn real_ssh_dash_l_smoke() { + if std::process::Command::new("ssh") + .arg("-V") + .output() + .is_err() + { + eprintln!("SKIPPED (with reason): real-tool tunnel smoke — ssh is unavailable"); + return; + } + if std::process::Command::new("curl") + .arg("--version") + .output() + .is_err() + { + eprintln!("SKIPPED (with reason): real-tool tunnel smoke — curl is unavailable"); + return; + } + + let (http_addr, body) = spawn_http_endpoint().await; + let client = wire_fabric( + tcp_dial(), + identity("tunnel-caller", &[TUNNEL_DIRECT_SCOPE]), + Arc::new(tokio::sync::Mutex::new(None)), + ) + .await; + + let work = std::env::temp_dir().join(format!("alkssh-tunnel-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"); + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o600)) + .expect("the key file perms set"); + + 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(client); + 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", + 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", + "-L", + &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 -L listener binds client-side; give it a bounded settle + // window then curl through it. + 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 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 -L + curl through the door → fabric → endpoint returns the body — got {stdout:?}" + ); + + let _ = std::fs::remove_dir_all(&work); + } + + async fn pick_local_port() -> u16 { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("the probe binds"); + let port = listener.local_addr().expect("addr").port(); + drop(listener); + port + } +}