feat(tunnel): land the route-1 bridge — direct-tcpip + streamlocal served for real (w3-tunnel-direct)

- The route-1 bridge (ADR-006 decision 1, the POC-3 stage-C recipe
  re-landed as contract): per accepted `direct-tcpip` /
  `direct-streamlocal` open, the target/substrate are constructed from
  the SSH open arguments and the spawned task runs
  `TunnelSession::open_direct` over the held `Arc<ChannelClient>` →
  accept only after the fabric open succeeds → `take_halves()` pumped
  against the taken channel (`pump_bidi`) — the open hook returns fast;
  a fabric failure answers the explicit `ConnectFailed` reject (the
  wave-2 stub's reason surviving as the real decline; the stub retired
  from the direct rows, kept on `forwarded-tcpip` for w3-tunnel-listen).
- The cap across the fabric window (recorded): the admission is taken
  at the hook and held across the fabric open; a fabric failure
  releases the count inside the spawned task — the admitted-but-
  rejected shape no `channel_close` can ever fire for. `ChannelCap`
  became a mutex-backed `Arc`: admissions stay packet-loop-only,
  releases have exactly two disjoint sites (close hook, bridge task),
  both saturating.
- The identity ride recorded (the `w2-auth-bridge` parked item): the
  alkhub `hub_leg.rs` one-time-set precedent is a serving-side set; the
  client-side `take_call_connection()` seam is rejected for this door
  (one-shot detach of the shared assembly handle's open surface — a
  fortiori the relay note's "wrong for hubs"); the identity the fabric's
  `AccessControl::check` consults is the dial identity (CF-005), the
  composition is assembly policy, and the door structurally refuses
  anonymous serving (`no-identity`, unreachable on the wire). Upstream
  ask booked for the gate.
- The OQ-11 runtime-Unix gate: `direct-streamlocal` compiles wherever
  `tunnel` compiles; off-Unix a clean open-decline before any fabric
  call (dial-call witness at zero), on-Unix the real bridge.
- Decline classes recorded: policy → `AdministrativelyProhibited`
  (unfeatured/cap/`no-identity`); cannot-bridge → `ConnectFailed`
  (failed fabric open, absent handle under the feature, invalid
  target, off-Unix streamlocal); the event vocabulary gains
  `invalid-target` and `unix-unavailable`.
- The optional dep lands: `alktunnels 0.4.1` (pin re-verified against
  crates.io — max stable, MIT OR Apache-2.0, rust_version 1.88, alkcall
  0.10.1 exact), wired `tunnel = ["dep:alktunnels"]`, `local` never
  enabled.
- The fabric double (`tests/tunnel_mapping.rs`): the POC-3 §C.4
  in-process recipe — `ChannelsAdapter` + `InstallChannelZero` + the
  alktunnels openables + a hand-rolled dial/dial-fail seam (in-process
  duplex targets keyed by the rendered string; the smoke's dial is real
  TCP). Rows: both-substrate serve round trips, fabric-failure,
  cap refusal + release + recovery, the fabric-ACL composed-gate
  witness, the CF-006 identity witness, the OQ-11 gate, and the
  `ssh -L` + curl real-tool smoke (conditional-skip posture, ran green).

Verification: `cargo test` rows green per feature build (default,
tunnel, git, session, git+session, tunnel+git, tunnel+session,
--all-features); `cargo clippy --all-targets -- -D warnings` ×2
(default + --all-features) clean; `cargo fmt --check`; `cargo doc
--no-deps` 0 warnings; `taskgraph validate` (26 tasks).
This commit is contained in:
glm-5.3-flash committed 2026-10-29 03:23:20 +00:00
1 parent b1e569d1a8
commit 6443feac3e
13 files changed
+1611 -170

No files matched your search

Generated
+17
View File
@@ -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"
+13 -3
View File
@@ -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).
+104 -47
View File
@@ -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<usize>`: **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<ChannelCap>`]) 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<usize>,
}
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);
}
}
+14 -8
View File
@@ -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
}
+220 -65
View File
@@ -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<ServeConfig>,
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<ChannelCap>,
resolved_identity: Option<Identity>,
/// 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<ServeConfig>) -> 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<Msg>,
_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<Msg>,
@@ -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<Msg>,
reply: ChannelOpenHandle,
target: Option<alktunnels::params::SubstrateAddr>,
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;
}
+2
View File
@@ -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")]
+21 -6
View File
@@ -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<E: std::fmt::Debug>(
row: &'static str,
+161
View File
@@ -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<ChannelClient>` 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<ServeConfig>` 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<russh::server::Msg>;
/// 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<ChannelClient>,
cap: Arc<ChannelCap>,
) {
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<ChannelClient>,
cap: Arc<ChannelCap>,
) {
serve::spawn_serving_path(serve_direct_row(
row, channel, reply, target, substrate, client, cap,
));
}
+133 -9
View File
@@ -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.
> 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<usize>` 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<ServeConfig>` 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<ChannelClient>` 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).
+12 -4
View File
@@ -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;
+29 -20
View File
@@ -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;
+9 -8
View File
@@ -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;
}
+876
View File
@@ -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<std::sync::Mutex<HashMap<String, tokio::io::DuplexStream>>>;
/// 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<ChannelClient>` 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<ChannelClient> {
let registry = Arc::new(OperationRegistry::new());
let registry_for_hook = Arc::clone(&registry);
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(&registry_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(&registry).expect("channel ops register");
register_tunnel_direct_openable(
&ChannelCore::new(manager.clone(), alkcall::channels::policy::default_policy()),
&registry,
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<AtomicUsize>) -> 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<ChannelClient>, 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<Msg>,
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
}
}