From 58d0b86ec5d385dce4da0c4ee4953cccdf2edd7b Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Wed, 28 Oct 2026 17:50:49 +0000 Subject: [PATCH] feat(observe): land the decline-path event vocabulary (w3-decline-tracing) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The wave-2 gate's booked finding: observability on the decline paths. `tracing` rides the manifest per the ADR-002 toolkit but nothing in src/ emitted — a decline's "reason" was only the routing decision documented per build. - src/observe.rs: the event vocabulary, adopting the family posture read from alktty/alktunnels/alkcall/alkhttp (free tracing fns, message-last lowercase, flat fields; no info!/trace!, no macros, no spans): feature_build + row + channel_id + decline_class (open/request/serving/auth) + reason + exec_route (exec rows); declines at debug, the serving-path failure arm the vocabulary's one warn - handler.rs: every decline site emits — auth dispositions (non-ed25519 / fingerprint-unresolved / the four unimplemented methods), open-level declines (unfeatured / never-feature / cap-exceeded / tunnel connect-failed stub), request-level capability declines (exec route-aware, subsystem, env, x11, agent, the -R listen forms), and the seam-stub serving declines with the spawned-path lifecycle (started / ended / failed — the shape the wave-3 mappings inherit) - tests/observe.rs: the event-presence pin — a hand-rolled global capture subscriber (the recorded test convention; no event-capture precedent exists in the family), field-matching rows across the default/git/session/tunnel/--all-features builds, and the safety scan (no username, no fingerprint ever in an event) - helper-level exact shapes pinned in src/observe.rs unit rows Verification: cargo test (default, --features git/session/tunnel, --all-features all green), clippy x2 --all-targets -D warnings, fmt --check, cargo doc --no-deps 0 warnings, taskgraph validate. --- src/handler.rs | 130 +++- src/lib.rs | 1 + src/observe.rs | 458 ++++++++++++ src/serve.rs | 3 + tasks/implementation/w3-decline-tracing.md | 162 ++++- tests/observe.rs | 765 +++++++++++++++++++++ 6 files changed, 1490 insertions(+), 29 deletions(-) create mode 100644 src/observe.rs create mode 100644 tests/observe.rs diff --git a/src/handler.rs b/src/handler.rs index b390faa..088f56d 100644 --- a/src/handler.rs +++ b/src/handler.rs @@ -128,6 +128,18 @@ //! serving task (ADR-006 decision 4, §The spawn-and-return //! structural constraint above); the wire shape is unchanged. //! +//! ## The event vocabulary on every decline (`w3-decline-tracing`) +//! +//! Every decline site here emits its event through +//! [`crate::observe`] — the wave-2 gate's booked finding: the +//! deciding context (row, feature build, channel id, decline class, +//! reason) rides the tracing event, not just the routing decision +//! documented per build. The vocabulary and its safety posture (no +//! secrets, no username, no fingerprint — structurally) live in that +//! module; the serving-path lifecycle (started / ended / failure) +//! hangs off the spawned stub body and is what the wave-3 mappings' +//! real serving paths inherit. +//! //! ## Bounded resources at the handler surface (`w2-bounded-resources`) //! //! (AGENTS.md convention 9; `overview.md` §Bounded-resources policy — @@ -171,6 +183,7 @@ use crate::config::ServeConfig; use crate::dispatch; use crate::error::AdapterError; use crate::fingerprint::fingerprint; +use crate::observe; use crate::serve; /// The per-connection handler (one per mounted stream): russh constructs @@ -261,11 +274,17 @@ impl SessionHandler { if *signature_algorithm != Algorithm::Ed25519 || public_key.algorithm() != Algorithm::Ed25519 { + observe::decline_auth("auth_publickey", "non-ed25519"); return None; } - self.serve + let identity = self + .serve .identity_provider - .resolve_from_fingerprint(&fingerprint(public_key)) + .resolve_from_fingerprint(&fingerprint(public_key)); + if identity.is_none() { + observe::decline_auth("auth_publickey", "fingerprint-unresolved"); + } + identity } /// The `session` channel-open gate (ADR-003 feature graph): @@ -334,6 +353,7 @@ impl russh::server::Handler for SessionHandler { /// ADR-004 decision 3: explicit family-contract decline — never the /// inherited default (a pin bump could silently change it). async fn auth_none(&mut self, _user: &str) -> Result { + observe::decline_auth("auth_none", "method-unimplemented"); Ok(Auth::reject()) } @@ -341,6 +361,7 @@ impl russh::server::Handler for SessionHandler { /// auth does not exist on this door; ADR-005's key custody is the /// whole story. async fn auth_password(&mut self, _user: &str, _password: &str) -> Result { + observe::decline_auth("auth_password", "method-unimplemented"); Ok(Auth::reject()) } @@ -352,6 +373,7 @@ impl russh::server::Handler for SessionHandler { _user: &str, _certificate: &russh::keys::Certificate, ) -> Result { + observe::decline_auth("auth_openssh_certificate", "cert-unsupported"); Ok(Auth::reject()) } @@ -362,6 +384,7 @@ impl russh::server::Handler for SessionHandler { _submethods: &str, _response: Option>, ) -> Result { + observe::decline_auth("auth_keyboard_interactive", "method-unimplemented"); Ok(Auth::reject()) } @@ -403,12 +426,14 @@ impl russh::server::Handler for SessionHandler { _session: &mut Session, ) -> Result<(), Self::Error> { if !self.serves_session_channels() { + observe::decline_open("session", "unfeatured", _channel.id().number()); reply.reject(dispatch::unfeatured_open_decline()).await; return Ok(()); } if self.admit_channel_open() { reply.accept().await; } else { + observe::decline_open("session", "cap-exceeded", _channel.id().number()); reply.reject(dispatch::cap_exceeded_open_decline()).await; } Ok(()) @@ -420,12 +445,13 @@ impl russh::server::Handler for SessionHandler { /// path, and this hook never has one. async fn channel_open_x11( &mut self, - _channel: Channel, + channel: Channel, _originator_address: &str, _originator_port: u32, reply: ChannelOpenHandle, _session: &mut Session, ) -> Result<(), Self::Error> { + observe::decline_open("x11", "never-feature", channel.id().number()); reply.reject(dispatch::unfeatured_open_decline()).await; Ok(()) } @@ -441,7 +467,7 @@ impl russh::server::Handler for SessionHandler { /// moves to wave 3's real accept point. async fn channel_open_direct_tcpip( &mut self, - _channel: Channel, + channel: Channel, _host_to_connect: &str, _port_to_connect: u32, _originator_address: &str, @@ -449,15 +475,18 @@ impl russh::server::Handler for SessionHandler { reply: ChannelOpenHandle, _session: &mut Session, ) -> Result<(), Self::Error> { + const ROW: &str = "direct-tcpip"; if !self.serves_tunnels() { + observe::decline_open(ROW, "unfeatured", channel.id().number()); reply.reject(dispatch::unfeatured_open_decline()).await; return Ok(()); } if self.channel_cap_exhausted() { + observe::decline_open(ROW, "cap-exceeded", channel.id().number()); reply.reject(dispatch::cap_exceeded_open_decline()).await; return Ok(()); } - tunnel_row_stub(reply).await; + tunnel_row_stub(ROW, channel.id().number(), reply).await; Ok(()) } @@ -469,7 +498,7 @@ impl russh::server::Handler for SessionHandler { /// identical to the other tunnel rows. async fn channel_open_forwarded_tcpip( &mut self, - _channel: Channel, + channel: Channel, _host_to_connect: &str, _port_to_connect: u32, _originator_address: &str, @@ -477,15 +506,18 @@ impl russh::server::Handler for SessionHandler { reply: ChannelOpenHandle, _session: &mut Session, ) -> Result<(), Self::Error> { + const ROW: &str = "forwarded-tcpip"; if !self.serves_tunnels() { + observe::decline_open(ROW, "unfeatured", channel.id().number()); reply.reject(dispatch::unfeatured_open_decline()).await; return Ok(()); } if self.channel_cap_exhausted() { + observe::decline_open(ROW, "cap-exceeded", channel.id().number()); reply.reject(dispatch::cap_exceeded_open_decline()).await; return Ok(()); } - tunnel_row_stub(reply).await; + tunnel_row_stub(ROW, channel.id().number(), reply).await; Ok(()) } @@ -495,20 +527,23 @@ impl russh::server::Handler for SessionHandler { /// everywhere, so the runtime gate has nothing to ride yet. async fn channel_open_direct_streamlocal( &mut self, - _channel: Channel, + channel: Channel, _socket_path: &str, reply: ChannelOpenHandle, _session: &mut Session, ) -> Result<(), Self::Error> { + const ROW: &str = "direct-streamlocal"; if !self.serves_tunnels() { + observe::decline_open(ROW, "unfeatured", channel.id().number()); reply.reject(dispatch::unfeatured_open_decline()).await; return Ok(()); } if self.channel_cap_exhausted() { + observe::decline_open(ROW, "cap-exceeded", channel.id().number()); reply.reject(dispatch::cap_exceeded_open_decline()).await; return Ok(()); } - tunnel_row_stub(reply).await; + tunnel_row_stub(ROW, channel.id().number(), reply).await; Ok(()) } @@ -531,6 +566,12 @@ impl russh::server::Handler for SessionHandler { if self.serves_git_row() { git_row_stub(channel, session); } else { + observe::decline_request( + "exec_request", + "unfeatured", + channel.number(), + Some("git"), + ); decline_unfeatured_request(channel, session).await; } } @@ -538,6 +579,12 @@ impl russh::server::Handler for SessionHandler { if self.serves_session_rows() { pipe_row_stub(channel, session); } else { + observe::decline_request( + "exec_request", + "unfeatured", + channel.number(), + Some("pipe"), + ); decline_unfeatured_request(channel, session).await; } } @@ -561,6 +608,7 @@ impl russh::server::Handler for SessionHandler { if self.serves_session_rows() { pty_row_stub(channel, session); } else { + observe::decline_request("pty_request", "unfeatured", channel.number(), None); decline_unfeatured_request(channel, session).await; } Ok(()) @@ -578,6 +626,7 @@ impl russh::server::Handler for SessionHandler { if self.serves_session_rows() { shell_row_stub(channel, session); } else { + observe::decline_request("shell_request", "unfeatured", channel.number(), None); decline_unfeatured_request(channel, session).await; } Ok(()) @@ -597,6 +646,7 @@ impl russh::server::Handler for SessionHandler { _name: &str, session: &mut Session, ) -> Result<(), Self::Error> { + observe::decline_request("subsystem_request", "unimplemented", channel.number(), None); decline_unfeatured_request(channel, session).await; Ok(()) } @@ -611,6 +661,7 @@ impl russh::server::Handler for SessionHandler { _variable_value: &str, session: &mut Session, ) -> Result<(), Self::Error> { + observe::decline_request("env_request", "declined-by-design", channel.number(), None); decline_unfeatured_request(channel, session).await; Ok(()) } @@ -626,6 +677,7 @@ impl russh::server::Handler for SessionHandler { _x11_screen_number: u32, session: &mut Session, ) -> Result<(), Self::Error> { + observe::decline_request("x11_request", "never-feature", channel.number(), None); decline_unfeatured_request(channel, session).await; Ok(()) } @@ -636,9 +688,10 @@ impl russh::server::Handler for SessionHandler { /// decline; russh replies the channel failure. async fn agent_request( &mut self, - _channel: ChannelId, + channel: ChannelId, _session: &mut Session, ) -> Result { + observe::decline_request("agent_request", "never-feature", channel.number(), None); Ok(false) } @@ -653,6 +706,7 @@ impl russh::server::Handler for SessionHandler { _port: &mut u32, _session: &mut Session, ) -> Result { + observe::decline_global_request("tcpip_forward", "unimplemented"); Ok(false) } @@ -664,6 +718,7 @@ impl russh::server::Handler for SessionHandler { _port: u32, _session: &mut Session, ) -> Result { + observe::decline_global_request("cancel_tcpip_forward", "unimplemented"); Ok(false) } @@ -675,6 +730,7 @@ impl russh::server::Handler for SessionHandler { _socket_path: &str, _session: &mut Session, ) -> Result { + observe::decline_global_request("streamlocal_forward", "unimplemented"); Ok(false) } @@ -685,13 +741,16 @@ impl russh::server::Handler for SessionHandler { _socket_path: &str, _session: &mut Session, ) -> Result { + observe::decline_global_request("cancel_streamlocal_forward", "unimplemented"); Ok(false) } } /// The request-level decline shape for an unfeatured/unimplemented /// capability: fail the request and leave the channel usable (POC-1 -/// §5 — session survives declines, no state poisoning). +/// §5 — session survives declines, no state poisoning). The decline +/// event is emitted by the caller (the hook carries the row/reason +/// context); this is the wire shape only. async fn decline_unfeatured_request(channel: ChannelId, session: &mut Session) { let _ = session.channel_failure(channel); } @@ -705,17 +764,41 @@ async fn decline_unfeatured_request(channel: ChannelId, session: &mut Session) { /// IS the row this wave). Decline-first-alloc-never holds structurally: /// nothing is allocated before the failure lands — the spawn's handle /// clone is the only resource ever taken. -async fn seam_stub_decline(handle: Handle, channel: ChannelId) { - let _ = handle.channel_failure(channel).await; - let _ = handle.close(channel).await; +/// +/// The serving-path lifecycle events ride this body +/// ([`crate::observe`]): started (the task's first action), the +/// serving-path decline (class `serving`, reason `seam-stub`), and — +/// after the wire reporting has completed — ended, or failed when a +/// report send could not complete (the one warn in the vocabulary). +async fn seam_stub_decline( + handle: Handle, + channel: ChannelId, + row: &'static str, + exec_route: Option<&'static str>, +) { + observe::serving_path_started(row, channel.number(), exec_route); + observe::decline_serving(row, "seam-stub", channel.number(), exec_route); + let error = match handle.channel_failure(channel).await { + Ok(()) => handle.close(channel).await.err(), + Err(error) => Some(error), + }; + match error { + Some(error) => observe::serving_path_failed(row, channel.number(), &error, exec_route), + None => observe::serving_path_ended(row, channel.number(), exec_route), + } } /// Spawns the seam-stub decline as the row's serving path (ADR-006 /// decision 4): the handle is cloned here, the decline runs on its own /// task, and the hook returns — no serving path awaits inline. -fn spawn_seam_stub(channel: ChannelId, session: &Session) { +fn spawn_seam_stub( + channel: ChannelId, + session: &Session, + row: &'static str, + exec_route: Option<&'static str>, +) { let handle = session.handle(); - serve::spawn_serving_path(seam_stub_decline(handle, channel)); + serve::spawn_serving_path(seam_stub_decline(handle, channel, row, exec_route)); } /// The git row's seam stub — wave 3 lands the alkgit seam here @@ -723,7 +806,7 @@ fn spawn_seam_stub(channel: ChannelId, session: &Session) { /// per ADR-006 decision 4; the skeleton declines at the seam, on its /// own task. fn git_row_stub(channel: ChannelId, session: &Session) { - spawn_seam_stub(channel, session); + spawn_seam_stub(channel, session, "exec_request", Some("git")); } /// The session pipe row's seam stub — wave 3 lands the alktty pipe @@ -731,7 +814,7 @@ fn git_row_stub(channel: ChannelId, session: &Session) { /// ADR-006 decision 4; the skeleton declines at the seam, on its own /// task. fn pipe_row_stub(channel: ChannelId, session: &Session) { - spawn_seam_stub(channel, session); + spawn_seam_stub(channel, session, "exec_request", Some("pipe")); } /// The session pty row's seam stub — wave 3 lands the alktty @@ -739,7 +822,7 @@ fn pipe_row_stub(channel: ChannelId, session: &Session) { /// (`drive_session_pre_negotiated`), spawned per ADR-006 decision 4; /// the skeleton declines at the seam, on its own task. fn pty_row_stub(channel: ChannelId, session: &Session) { - spawn_seam_stub(channel, session); + spawn_seam_stub(channel, session, "pty_request", None); } /// The session shell row's seam stub — wave 3 lands the alktty @@ -747,13 +830,16 @@ fn pty_row_stub(channel: ChannelId, session: &Session) { /// crate default), spawned per ADR-006 decision 4; the skeleton /// declines at the seam, on its own task. fn shell_row_stub(channel: ChannelId, session: &Session) { - spawn_seam_stub(channel, session); + spawn_seam_stub(channel, session, "shell_request", None); } /// 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). -async fn tunnel_row_stub(reply: ChannelOpenHandle) { +/// 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. +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; } diff --git a/src/lib.rs b/src/lib.rs index 1771108..2b091b8 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -33,6 +33,7 @@ mod fingerprint; mod handler; mod host_key; mod key_source; +mod observe; mod serve; pub use adapter::SshAdapter; diff --git a/src/observe.rs b/src/observe.rs new file mode 100644 index 0000000..734c522 --- /dev/null +++ b/src/observe.rs @@ -0,0 +1,458 @@ +//! The decline-path + serving-path event vocabulary +//! (`w3-decline-tracing` — the wave-2 gate's booked finding): the +//! observability surface every decline site in `src/` emits through, +//! and the serving-path lifecycle skeleton ([`serving_path_started`] / +//! [`serving_path_ended`] / [`serving_path_failed`]) the wave-3 +//! mappings inherit. +//! +//! ## The family posture read (2026-10-28) +//! +//! Read before choosing this shape — no alkssh-private dialect: +//! +//! - `alktty` (`adapter.rs`, `session.rs`, `local/pipe.rs`, +//! `local/pty.rs`): free `tracing::{debug, warn}` imports, +//! message-last lowercase sentences with a component prefix +//! (`"tty: backend stdin write failed: {e}"`), no spans, no macros. +//! - `alktunnels` (`producer.rs`, `consumer.rs`, `params.rs`): the +//! same free-fn posture, structured fields where identity exists +//! (`listen = …`, `error = %e`, `reason = …`), levels `debug` for +//! structural observations, `warn` for recoverable failures, +//! `error` for abnormal pump terminations. +//! - `alkcall` (`protocol/*`, `channels/*`): the widest use — the same +//! fields convention (`channel_id`, `request_id`, `error = %e`), +//! component-prefixed messages (`"channel client: …"`, +//! `"open wrapper: …"`), `debug`/`warn` only. +//! - `alkhttp` (the closest structural analog): `warn`/`error` with +//! `error = %e` (`"channels session ended"`). +//! - `alkgit`: no tracing at all at this writing. +//! +//! Never observed in the family: `info!` / `trace!` events, subscriber +//! setup in library code, macro or span wrappers — so none here +//! either. Adopted verbatim: free `tracing` fn calls, message-last +//! lowercase with the `"decline: …"` / `"serving path …"` phrasing, +//! flat per-event fields (`feature_build`, `row`, `channel_id`, +//! `decline_class`, `reason`, `exec_route`, `error`). +//! +//! ## The event vocabulary +//! +//! Every event carries exactly: +//! +//! - `feature_build = FEATURE_BUILD` — the crate's compile-time +//! feature set as a stable string (`"none"`, `"git"`, +//! `"session+git"`, …; features joined in the order `session, git, +//! tunnel, sftp`), so a captured event is self-describing per +//! build. The `sftp` feature gates no route, but ships in the +//! build string: it is part of what the assembly deployed. +//! - `row` — the greppable wire row the event is about: the channel +//! type for open-level (e.g. `"session"`, `"direct-tcpip"`, `"x11"`), +//! the request name for request-level (e.g. `"pty_request"`, +//! `"tcpip_forward"`), the handler hook name for auth (e.g. +//! `"auth_publickey"`), the session-row hook names for the +//! serving-path lifecycle. +//! - `channel_id` — the wire channel id (u32), on every event of a +//! channel; the auth class has none (the auth hooks carry no +//! channel at this surface). +//! +//! Decline events additionally carry exactly: +//! +//! - `decline_class` — `"open"` | `"request"` | `"serving"` | `"auth"`: +//! the two wire decline classes of `mapping.md` §Wire decline +//! reasons, plus the serving-path class (this wave the seam stub's +//! decline *is* the serving path; the wave-3 mappings emit the same +//! class from their real spawned paths) and the auth class (the +//! disposition declines of ADR-004). +//! - `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`), +//! `"seam-stub"` (the session-row seam stubs), `"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"` / +//! `"non-ed25519"` / `"fingerprint-unresolved"`. +//! - `exec_route` — only on `exec_request`-row events: the dispatch +//! route the parse routed to (`"git"` | `"pipe"`), so a git row's +//! decline is distinguishable from a pipe row's in combined builds. +//! Other rows carry none (a `None` field is not recorded by +//! `tracing`). +//! +//! Levels: declines and the lifecycle's started/ended emit at +//! `debug` — a decline is the door's normal, honest posture (vanilla +//! clients probe unfeatured rows routinely; nothing abnormal +//! happened), matching the family's structural-observation level. +//! Only [`serving_path_failed`] is `warn` — a serving path that could +//! not complete its wire reporting is a real failure (the family's +//! recoverable-failure level). +//! +//! ## Safety posture (conventions 6, OQ-03) +//! +//! **No secret material ever enters an event; never the username.** +//! Structurally enforced, not recalled: the helpers take only +//! `&'static str` vocabulary values and the wire channel id — there +//! is no parameter a username, command string, env name/value, repo +//! path, key, or fingerprint could ride. In particular *the +//! fingerprint stays out of the events* (`auth.md` §Dispositions: log +//! exposure is deployment policy — the crate's events are +//! safe-by-default at every level, so the fingerprint is a +//! deployer-side computation over its own logs, never crate-emitted). +//! The `error` field carries only russh protocol errors. The wire +//! rows in `tests/observe.rs` pin the posture: the username never +//! appears in any captured field. If a future decline site needs +//! request-argument detail (e.g. the git grammar's decline reasons in +//! `w3-git-mapping`), it must add an explicitly-vocabulary'd field and +//! record its safety reasoning — dynamic strings do not ride this +//! module by default. + +/// The compile-time feature set, as a stable string (`"none"`, +/// `"session"`, `"session+git"`, …). Features join in the order +/// `session, git, tunnel, sftp`; the featureless build is `"none"`. +pub(crate) const FEATURE_BUILD: &str = feature_build(); + +const fn feature_build() -> &'static str { + match ( + cfg!(feature = "session"), + cfg!(feature = "git"), + cfg!(feature = "tunnel"), + cfg!(feature = "sftp"), + ) { + (false, false, false, false) => "none", + (true, false, false, false) => "session", + (false, true, false, false) => "git", + (false, false, true, false) => "tunnel", + (false, false, false, true) => "sftp", + (true, true, false, false) => "session+git", + (true, false, true, false) => "session+tunnel", + (true, false, false, true) => "session+sftp", + (false, true, true, false) => "git+tunnel", + (false, true, false, true) => "git+sftp", + (false, false, true, true) => "tunnel+sftp", + (true, true, true, false) => "session+git+tunnel", + (true, true, false, true) => "session+git+sftp", + (true, false, true, true) => "session+tunnel+sftp", + (false, true, true, true) => "git+tunnel+sftp", + (true, true, true, true) => "session+git+tunnel+sftp", + } +} + +/// An open-level decline: the channel-open gate refused an open +/// (`unfeatured` / `never-feature` / `cap-exceeded` / the tunnel seam +/// stub's `connect-failed`). The wire answer is the open-failure code +/// ([`crate::dispatch`]); this is the event for the decision behind it. +pub(crate) fn decline_open(row: &'static str, reason: &'static str, channel_id: u32) { + tracing::debug!( + feature_build = FEATURE_BUILD, + row, + channel_id, + decline_class = "open", + reason, + "decline: channel open rejected" + ); +} + +/// A request-level decline on a session channel: a capability the +/// build does not serve (`subsystem` / `env` / `x11` / agent, an +/// unfeatured parse route) or a not-yet-landed listen form. The wire +/// answer is `channel_failure` (the channel stays usable) or the +/// denied-request reply — the event is the deciding context. +pub(crate) fn decline_request( + row: &'static str, + reason: &'static str, + channel_id: u32, + exec_route: Option<&'static str>, +) { + tracing::debug!( + feature_build = FEATURE_BUILD, + row, + channel_id, + decline_class = "request", + reason, + exec_route, + "decline: channel request rejected" + ); +} + +/// An auth-class decline: a disposition of the ADR-004 bridge turned +/// the method away. No channel id exists at this surface; the +/// *fingerprint never rides the event* (see §Safety posture) — the +/// `reason` codes distinguish the decline causes without exposing who. +pub(crate) fn decline_auth(row: &'static str, reason: &'static str) { + tracing::debug!( + feature_build = FEATURE_BUILD, + row, + decline_class = "auth", + reason, + "decline: auth disposition rejected" + ); +} + +/// A request-level decline on a **global** request — the `-R` listen +/// forms (`mapping.md` §Global-request rows). No channel id exists at +/// this surface; the fields are otherwise the `request` vocabulary. +pub(crate) fn decline_global_request(row: &'static str, reason: &'static str) { + tracing::debug!( + feature_build = FEATURE_BUILD, + row, + decline_class = "request", + reason, + "decline: global request rejected" + ); +} + +/// A serving-path decline: a routed row whose serving task answers +/// failure-then-close rather than serving. This wave that is the seam +/// stub (the wire shape pinned in `tests/spawn_return.rs`); the wave-3 +/// mappings emit it from their real serving tasks wherever serving +/// itself cannot proceed — the class the spawned paths inherit. +pub(crate) fn decline_serving( + row: &'static str, + reason: &'static str, + channel_id: u32, + exec_route: Option<&'static str>, +) { + tracing::debug!( + feature_build = FEATURE_BUILD, + row, + channel_id, + decline_class = "serving", + reason, + exec_route, + "decline: serving path rejected" + ); +} + +/// 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. +pub(crate) fn serving_path_started( + row: &'static str, + channel_id: u32, + exec_route: Option<&'static str>, +) { + tracing::debug!( + feature_build = FEATURE_BUILD, + row, + channel_id, + exec_route, + "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. +pub(crate) fn serving_path_ended( + row: &'static str, + channel_id: u32, + exec_route: Option<&'static str>, +) { + tracing::debug!( + feature_build = FEATURE_BUILD, + row, + channel_id, + exec_route, + "serving path ended" + ); +} + +/// Spawned-path failure — the serving path could not complete its +/// wire reporting (the session handle's sends failed: the session is +/// gone, or the report never landed). The vocabulary's one `warn` — +/// 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. +pub(crate) fn serving_path_failed( + row: &'static str, + channel_id: u32, + error: &E, + exec_route: Option<&'static str>, +) { + tracing::warn!( + feature_build = FEATURE_BUILD, + row, + channel_id, + exec_route, + error = ?error, + "serving path failed" + ); +} + +#[cfg(test)] +mod tests { + use tracing::subscriber::with_default; + + /// A fake error shape — the failure row only pins that a Debug + /// error value rides the `error` field. + #[derive(Debug)] + struct Probe; + static PROBE: Probe = Probe; + + use super::*; + + /// A minimal field-capturing visitor shared by these rows and the + /// wire rows in `tests/observe.rs` (the test convention recorded + /// in the task notes — the family has no event-capture precedent, + /// only alktunnels' fmt-subscriber dev-dep for debug output, so + /// the convention is a hand-rolled minimal [`tracing::Subscriber`] + /// — no `tracing-subscriber` dependency). + struct Visitor(Vec<(String, String)>); + + impl tracing::field::Visit for Visitor { + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + self.0.push((field.name().to_owned(), value.to_owned())); + } + + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + self.0.push((field.name().to_owned(), format!("{value:?}"))); + } + } + + /// Captures the fields of the one event the emit fn fires, via + /// the thread-local default subscriber (`with_default` — these + /// are direct helper rows; the spawned-task surfaces are driven + /// over the wire in `tests/observe.rs` against the global + /// capture). + fn capture(emit: impl FnOnce()) -> Vec<(String, String)> { + struct Capture(std::sync::Mutex>); + impl tracing::Subscriber for Capture { + fn enabled(&self, _metadata: &tracing::Metadata<'_>) -> bool { + true + } + fn new_span(&self, _span: &tracing::span::Attributes<'_>) -> tracing::Id { + tracing::Id::from_u64(1) + } + fn record(&self, _span: &tracing::Id, _values: &tracing::span::Record<'_>) {} + fn record_follows_from(&self, _span: &tracing::Id, _follows: &tracing::Id) {} + fn event(&self, event: &tracing::Event<'_>) { + let mut v = Visitor(Vec::new()); + event.record(&mut v); + self.0 + .lock() + .unwrap_or_else(|e| e.into_inner()) + .append(&mut v.0); + } + fn enter(&self, _span: &tracing::Id) {} + fn exit(&self, _span: &tracing::Id) {} + fn clone_span(&self, _id: &tracing::Id) -> tracing::Id { + tracing::Id::from_u64(1) + } + } + let holder = std::sync::Arc::new(Capture(Default::default())); + with_default(std::sync::Arc::clone(&holder), || { + let prior = holder.0.lock().unwrap().len(); + emit(); + assert!(holder.0.lock().unwrap().len() > prior); + }); + holder.0.lock().unwrap().clone() + } + + fn field<'a>(fields: &'a [(String, String)], name: &str) -> &'a str { + fields + .iter() + .find(|(n, _)| n == name) + .map(|(_, v)| v.as_str()) + .unwrap_or_else(|| panic!("no field {name} in {fields:?}")) + } + + /// - *Contract stamp:* the event vocabulary (`w3-decline-tracing`) + /// — an open decline carries exactly the base shape, values + /// included. + #[test] + fn decline_open_carries_the_full_vocabulary() { + let fields = capture(|| decline_open("x11", "never-feature", 7)); + assert_eq!(field(&fields, "feature_build"), FEATURE_BUILD); + assert_eq!(field(&fields, "row"), "x11"); + assert_eq!(field(&fields, "channel_id"), "7"); + assert_eq!(field(&fields, "decline_class"), "open"); + assert_eq!(field(&fields, "reason"), "never-feature"); + assert_eq!(field(&fields, "message"), "decline: channel open rejected"); + let names: Vec = fields.iter().map(|(n, _)| n.clone()).collect(); + assert_eq!( + names, + vec![ + "message".to_owned(), + "feature_build".to_owned(), + "row".to_owned(), + "channel_id".to_owned(), + "decline_class".to_owned(), + "reason".to_owned(), + ], + "an open decline carries exactly the vocabulary, nothing else \ + (the message field first — the event! macro's shape)" + ); + } + + /// - *Contract stamp:* the `exec_route` field — present on the + /// exec row, absent elsewhere (a `None` field is not recorded + /// by `tracing`). + #[test] + fn decline_request_carries_the_route_when_the_parse_routed() { + let fields = capture(|| decline_request("exec_request", "unfeatured", 3, Some("git"))); + assert_eq!(field(&fields, "decline_class"), "request"); + assert_eq!(field(&fields, "exec_route"), "git"); + + let fields = capture(|| decline_request("subsystem_request", "unimplemented", 3, None)); + assert_eq!(field(&fields, "decline_class"), "request"); + assert!( + !fields.iter().any(|(n, _)| n == "exec_route"), + "a None route field is not recorded at all" + ); + } + + #[test] + fn decline_auth_carries_no_channel() { + let fields = capture(|| decline_auth("auth_password", "method-unimplemented")); + let names: Vec = fields.iter().map(|(n, _)| n.clone()).collect(); + assert!(!names.iter().any(|n| n.starts_with("channel"))); + assert_eq!(field(&fields, "decline_class"), "auth"); + assert_eq!(field(&fields, "row"), "auth_password"); + assert_eq!(field(&fields, "reason"), "method-unimplemented"); + } + + /// - *Contract stamp:* the `exec_route` field — present on the + /// exec row, absent elsewhere (a `None` field is not recorded + /// by `tracing`). The pty/shell lifecycle rows carry no route. + #[test] + fn serving_path_lifecycle_carries_started_ended_and_the_warn_failure() { + let fields = capture(|| serving_path_started("pty_request", 1, None)); + assert_eq!(field(&fields, "message"), "serving path started"); + let names: Vec = fields.iter().map(|(n, _)| n.clone()).collect(); + assert!( + !names.iter().any(|n| n == "decline_class"), + "lifecycle events carry no decline class" + ); + + let fields = capture(|| serving_path_ended("exec_request", 1, Some("pipe"))); + assert_eq!(field(&fields, "exec_route"), "pipe"); + assert_eq!(field(&fields, "message"), "serving path ended"); + + let fields = capture(|| serving_path_failed("shell_request", 1, &PROBE, None)); + assert_eq!(field(&fields, "message"), "serving path failed"); + assert_eq!(field(&fields, "error"), "Probe"); + } + + /// The feature-build string: every enabled feature appears; a + /// featureless build reads `none`. The expected half is derived + /// from `cfg!` independently of the crate's match — divergence + /// breaks this row loudly. + #[test] + fn feature_build_string_reflects_the_cfgs() { + let mut expected = String::new(); + for (on, name) in [ + (cfg!(feature = "session"), "session"), + (cfg!(feature = "git"), "git"), + (cfg!(feature = "tunnel"), "tunnel"), + (cfg!(feature = "sftp"), "sftp"), + ] { + if on { + if !expected.is_empty() { + expected.push('+'); + } + expected.push_str(name); + } + } + if expected.is_empty() { + expected.push_str("none"); + } + assert_eq!(FEATURE_BUILD, expected); + } +} diff --git a/src/serve.rs b/src/serve.rs index d3771fa..97ac288 100644 --- a/src/serve.rs +++ b/src/serve.rs @@ -13,6 +13,9 @@ //! and returns. No serving path is ever awaited inline; the dispatch //! seam stubs ([`crate::handler`]) hang off this helper, so the //! wave-3 mappings inherit the pattern mechanically from the stubs. +//! The serving-path *events* (started / ended / failure — +//! [`crate::observe`]) are the spawned task's own actions: the stub +//! body emits them, the mappings' real paths will too. //! - **Exit/status reporting rides the cloned handle** — the spawned //! task's only report surface is the handle clone it was handed; //! [`report_exit_status_then_close`] preserves the wire ordering the diff --git a/tasks/implementation/w3-decline-tracing.md b/tasks/implementation/w3-decline-tracing.md index a891b59..7b96fb9 100644 --- a/tasks/implementation/w3-decline-tracing.md +++ b/tasks/implementation/w3-decline-tracing.md @@ -1,7 +1,7 @@ --- id: w3-decline-tracing name: Wave 3 — decline-path tracing (the wave-2 gate's booked finding) -status: pending +status: completed depends_on: [w3-seam-config] scope: narrow risk: low @@ -46,16 +46,16 @@ inconsistent events. ## Acceptance Criteria -- [ ] The family tracing posture read and recorded; the event +- [x] The family tracing posture read and recorded; the event vocabulary established (row + build + channel + decline class) and used at every decline site in `src/` -- [ ] Serve-path skeleton events in place (start/ended/failure) so the +- [x] Serve-path skeleton events in place (start/ended/failure) so the mapping tasks' spawned paths inherit shaped rows -- [ ] The safety posture recorded (no secrets, no username; what +- [x] The safety posture recorded (no secrets, no username; what identity detail may appear decided) -- [ ] A test pins the event presence at the decline sites (or the +- [x] A test pins the event presence at the decline sites (or the doc-test convention the family uses — decide + record) -- [ ] `cargo test`, `cargo clippy --all-targets -- -D warnings`, +- [x] `cargo test`, `cargo clippy --all-targets -- -D warnings`, `cargo fmt --check`, `cargo doc --no-deps` pass ## References @@ -70,4 +70,152 @@ inconsistent events. ## Notes -> To be filled at implementation time. \ No newline at end of file +### The family posture read (2026-10-28) + +Read before choosing the event shape (the task's first bullet), across +the family's `src/`: + +- **alktty** (`adapter.rs`, `session.rs`, `channels.rs`, + `local/pipe.rs`, `local/pty.rs`): free `tracing::{debug, warn}` + imports; message-last lowercase sentences with a component prefix + (`"tty: backend stdin write failed: {e}"`); interpolation in + messages, no spans. +- **alktunnels** (`producer.rs`, `consumer.rs`, `params.rs`): the same + free-fn posture, plus structured fields where identity exists + (`listen = …`, `error = %e`, `reason = …`); levels `debug` for + structural observations, `warn` for recoverable failures, `error` + for abnormal pump terminations. +- **alkcall** (`protocol/*`, `channels/*`): the widest use — the same + fields convention (`channel_id`, `request_id`, `error = %e`), + component-prefixed messages (`"channel client: …"`, `"open wrapper: + …"`), `debug`/`warn` only. +- **alkhttp** (the closest structural analog): `warn`/`error` with + `error = %e` (`"channels session ended"`). +- **alkgit**: no tracing at all at this writing. + +Never observed: `info!`/`trace!` events, subscriber setup in library +code, macro or span wrappers. Adopted verbatim in `src/observe.rs`: +free `tracing` fn calls, message-last lowercase with the +`"decline: …"` / `"serving path …"` phrasing, flat per-event fields. + +### The event vocabulary (landed in `src/observe.rs`) + +- **`feature_build`** — the compile-time feature set as a stable + string (`"none"`, `"git"`, `"session+git"`, …; features join in the + order `session, git, tunnel, sftp`), so a captured event is + self-describing per build; `sftp` ships in the string even though it + gates no route (it is part of what the assembly deployed). +- **`row`** — the greppable wire row: the channel type for open-level + (`"session"`, `"direct-tcpip"`, `"x11"`, …), the request name for + request-level (`"pty_request"`, `"tcpip_forward"`, …), the handler + hook name for auth (`"auth_publickey"`, `"auth_none"`, …). +- **`decline_class`** — `"open"` | `"request"` | `"serving"` | + `"auth"`: the two wire decline classes (`mapping.md` §Wire decline + reasons), plus the serving-path class (this wave the seam stub's + decline *is* the serving path; the mappings emit the same class from + their real spawned paths) and the auth class (the ADR-004 + disposition declines — added by this task: the auth hooks are + decline sites too, and `auth.md`'s dispositions are the crate's + auth-decline contract). +- **`reason`** — the decline cause, mirroring the architecture's own + classifications: `"unfeatured"`, `"never-feature"` (ADR-003 §5), + `"cap-exceeded"`, `"connect-failed"` (tunnel stub / ADR-006 + decision 1), `"seam-stub"` (session-row stubs), `"unimplemented"` + (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"` / `"non-ed25519"` / + `"fingerprint-unresolved"`. +- **`channel_id`** — the wire channel id (u32), on every decline of a + channel (notably: the event's id is *russh's server-side* id — + `ChannelId` is not publicly constructible, so the helpers take the + raw u32 via `ChannelId::number()`); the auth class and the global + `-R` listen forms carry none (a separate `decline_global_request` + helper — global requests have no channel). +- **`exec_route`** — only on `exec_request`-row events: the parse + route (`"git"` | `"pipe"`), so a git row's decline is + distinguishable from a pipe row's in combined builds. A `None` route + field is not recorded at all by `tracing` (discovered by the unit + rows; recorded in the module docs). The `message` field rides first + (the `event!` macro's shape — the unit rows pin the field order). + +Levels: every decline and the lifecycle's started/ended emit at +`debug` — a decline is the door's normal, honest posture (vanilla +clients probe unfeatured rows routinely), matching the family's +structural-observation level. Only [`serving_path_failed`] is `warn` +(a serving path that could not complete its wire reporting is a real +failure — the family's recoverable-failure level). + +### The serving-path skeleton + +`src/serve.rs` stays the mechanics only (spawn + exit-report); the +lifecycle events hang off the spawned task's body in +`handler.rs`'s `seam_stub_decline`: started (the spawned task's first +action, so "started" is the task's honest start, not the hook's spawn +call) → the serving-path decline (class `serving`, reason `seam-stub`) +→ ended — or failed (`warn`, with the russh protocol error) when a +report send could not complete. The wave-3 mapping branches +(`w3-git-mapping`, `w3-tty-session`, `w3-tunnel-direct`) inherit the +shaped rows by emitting the same helpers from their real serving tasks. + +### The safety posture (conventions 6, OQ-03 — recorded) + +**No secret material ever enters an event; never the username; +the fingerprint stays out too.** Structurally enforced, not recalled: +the vocabulary helpers take only `&'static str` values and the wire +channel id — no parameter a username, command string, env name/value, +repo path, key, or fingerprint could ride. The `reason` codes carry +the decline's *cause class*, and the deciding context is the row + +build + channel — deliberately *not* the identity behind the wire +attempt. This is the recording of auth.md §Dispositions' standing +rule at the crate level: log exposure is deployment policy, so the +crate's events are safe-by-default at every level — the fingerprint +(and any identity detail) is a deployer-side computation over its own +logs, never crate-emitted. If a future decline site needs +request-argument detail (e.g. the git grammar's decline reasons in +`w3-git-mapping`), it must add an explicitly-vocabulary'd field and +record its safety reasoning — dynamic strings do not ride +`src/observe.rs` by default. The wire rows pin the posture by +scanning every captured event: no username string, no `SHA256:` +fingerprint, no `user*`-named field ever appears. + +### The test convention (decided + recorded) + +The family has no event-capture precedent to adopt — alktunnels' +`tracing-subscriber` dev-dep drives fmt output, not assertions. The +convention these rows pin: **a hand-rolled minimal +`tracing::Subscriber`** capturing events into a shared buffer, +installed once as the global default (`tests/observe.rs`) — the +server side runs on spawned tasks, so a thread-local +`with_default` capture cannot see them, and a global capture needs no +new dependency. Two levels of pinning: + +- Helper-level (`src/observe.rs` unit rows, thread-local + `with_default`): exact field sets and order (message first), the + `exec_route` presence semantics, the no-channel auth shape, the + lifecycle messages, the `warn` failure arm, and the + cfg-derived `feature_build` string. +- Wire-level (`tests/observe.rs`, the loopback harness drives the + real wire; capture installed up front via the mount helpers — a + lazily-installed global loses the earliest emissions): presence by + field-matching at every decline site across the builds — x11 open, + `-R` listen (global), the fingerprint-unresolved and auth_none + dispositions (driven on fresh mounts — a second auth exchange on an + already-authenticated session is not a supported russh surface, as + the probe discovered), the unfeatured session/pipe/pty rows, the + git seam-stub's full lifecycle, the tunnel stub's `connect-failed`, + the cap-exceeded row (zero-cap limits must ride the *server* config + — `serve_config`, not just the client's `Limits`), and the + username/fingerprint safety scan. Build-specific rows follow the + `dispatch_core.rs` per-build cfg convention and were verified in + the default, `git`, `session`, `tunnel`, and `--all-features` + builds. + +### Verification (all green) + +`cargo test` default build 110 rows across 7 binaries + doctests · +`cargo test --all-features` ✓ · `cargo test --features git/session/tunnel` +✓ (each build runs its decline rows) · +`cargo clippy --all-targets -- -D warnings` ✓ · +`cargo clippy --all-features --all-targets -- -D warnings` ✓ · +`cargo fmt --check` ✓ · `cargo doc --no-deps` 0 warnings · +`taskgraph validate` ✓. \ No newline at end of file diff --git a/tests/observe.rs b/tests/observe.rs new file mode 100644 index 0000000..ed33b23 --- /dev/null +++ b/tests/observe.rs @@ -0,0 +1,765 @@ +//! The `w3-decline-tracing` wire rows — the event presence pin: every +//! decline class in `src/` (the open-level declines, the request-level +//! capability declines, the seam-stub serving declines with their +//! spawned-path lifecycle, and the auth dispositions) emits its +//! event through the vocabulary (`src/observe.rs`), driven over the +//! real wire through the loopback harness. +//! +//! **Test convention (recorded):** the family's only tracing-test +//! precedent is alktunnels' fmt-subscriber dev-dep, used for debug +//! *output* — there is no event-capture convention to adopt, so these +//! rows pin the convention: a hand-rolled minimal +//! [`tracing::Subscriber`] capturing events into a shared buffer, +//! installed once as the global default (the server side runs on +//! spawned tasks — a thread-local default cannot see them; a global +//! capture needs no extra dependency). The install is up front — the +//! mount helpers call [`capture`] before any wire exchange, so no +//! event is emitted below the not-yet-installed global. Rows assert +//! presence by field-matching (`row` + `decline_class` + `reason` + +//! build), and the vocabulary assertion checks each captured decline's exact +//! field set — the shape the three wave-3 mapping branches inherit. +//! The helper-level shapes (field order, the `warn` level of the +//! failure arm) are pinned in `src/observe.rs`'s own unit rows. +//! +//! Feature builds are compile-time, so the build-specific rows are +//! feature-gated to the build whose decline matrix they drive (the +//! `dispatch_core.rs` convention); the shared decline rows run in +//! every build. + +#![allow(dead_code)] + +mod common; + +use std::sync::{Arc, Mutex, OnceLock}; +use std::time::{Duration, Instant}; + +use common::harness::{ + adapter, authenticate_publickey, await_close, drive, next_message, open_x11_channel, + tcpip_forward, +}; +use common::{MemHostKeyPolicy, MemIdentityProvider, fixture, identity}; +use russh::client::Msg; +use russh::{Channel, ChannelMsg}; + +use alkssh::Limits; + +// --------------------------------------------------------------------------- +// The capture subscriber — the recorded test convention (see module docs). +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone)] +struct Captured { + level: String, + fields: Vec<(String, String)>, +} + +impl Captured { + fn field(&self, name: &str) -> Option<&str> { + self.fields + .iter() + .find(|(n, _)| n == name) + .map(|(_, v)| v.as_str()) + } + + fn has(&self, want: &[(&str, &str)]) -> bool { + want.iter() + .all(|(n, v)| self.field(n).is_some_and(|got| got == *v)) + } +} + +struct FieldVisitor(Vec<(String, String)>); + +impl tracing::field::Visit for FieldVisitor { + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + self.0.push((field.name().to_owned(), value.to_owned())); + } + + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + self.0.push((field.name().to_owned(), format!("{value:?}"))); + } +} + +#[derive(Default)] +struct GlobalCapture(Mutex>); + +impl tracing::Subscriber for GlobalCapture { + fn enabled(&self, _metadata: &tracing::Metadata<'_>) -> bool { + true + } + + fn new_span(&self, _span: &tracing::span::Attributes<'_>) -> tracing::Id { + static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1); + tracing::Id::from_u64(NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed)) + } + + fn record(&self, _span: &tracing::Id, _values: &tracing::span::Record<'_>) {} + + fn record_follows_from(&self, _span: &tracing::Id, _follows: &tracing::Id) {} + + fn event(&self, event: &tracing::Event<'_>) { + let mut visitor = FieldVisitor(Vec::new()); + event.record(&mut visitor); + self.0.lock().unwrap().push(Captured { + level: event.metadata().level().to_string(), + fields: visitor.0, + }); + } + + fn enter(&self, _span: &tracing::Id) {} + fn exit(&self, _span: &tracing::Id) {} + fn clone_span(&self, _id: &tracing::Id) -> tracing::Id { + tracing::Id::from_u64(1) + } +} + +fn capture() -> &'static Arc { + static CAPTURE: OnceLock> = OnceLock::new(); + CAPTURE.get_or_init(|| { + let capture = Arc::new(GlobalCapture::default()); + tracing::subscriber::set_global_default(Arc::clone(&capture)) + .expect("the observe suite installs the global capture exactly once"); + capture + }) +} + +/// The shared capture snapshot — every test reads the whole buffer and +/// matches fields, so parallel rows never order against each other. +fn captured() -> Vec { + capture().0.lock().unwrap().clone() +} + +/// Waits until an event matching every field pair appears (the +/// spawned-task surfaces land their events asynchronously), then +/// returns it. +async fn await_event(want: &[(&str, &str)]) -> Captured { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + if let Some(event) = captured().into_iter().find(|e| e.has(want)) { + return event; + } + if Instant::now() > deadline { + panic!("no tracing event matching {want:?} within the window"); + } + tokio::time::sleep(Duration::from_millis(10)).await; + } +} + +/// The expected `feature_build` value, derived from `cfg!` — the +/// vocabulary assertion checks every captured decline against it (the +/// same convention as `src/observe.rs`'s cfg-derived unit row). +fn expected_build() -> &'static str { + match ( + cfg!(feature = "session"), + cfg!(feature = "git"), + cfg!(feature = "tunnel"), + cfg!(feature = "sftp"), + ) { + (false, false, false, false) => "none", + (true, false, false, false) => "session", + (false, true, false, false) => "git", + (false, false, true, false) => "tunnel", + (false, false, false, true) => "sftp", + (true, true, false, false) => "session+git", + (true, false, true, false) => "session+tunnel", + (true, false, false, true) => "session+sftp", + (false, true, true, false) => "git+tunnel", + (false, true, false, true) => "git+sftp", + (false, false, true, true) => "tunnel+sftp", + (true, true, true, false) => "session+git+tunnel", + (true, true, false, true) => "session+git+sftp", + (true, false, true, true) => "session+tunnel+sftp", + (false, true, true, true) => "git+tunnel+sftp", + (true, true, true, true) => "session+git+tunnel+sftp", + } +} + +/// The global-request rows — the no-channel surfaces (the event +/// vocabulary assertion needs to know the field set these carry). +fn is_global_row(row: Option<&str>) -> bool { + row.is_some_and(|row| { + matches!( + row, + "tcpip_forward" + | "cancel_tcpip_forward" + | "streamlocal_forward" + | "cancel_streamlocal_forward" + ) + }) +} + +/// The vocabulary assertion: a captured decline carries the exact +/// field set (nothing more — the safety posture rides on this: no +/// dynamic extra fields exist), with the build/class/reason/row the +/// caller matched on. +fn assert_decline(event: &Captured, want: &[(&str, &str)]) { + assert!( + event.has(want), + "decline event missing the matched fields {want:?}: {event:?}" + ); + assert_eq!( + event.field("feature_build"), + Some(expected_build()), + "every event is self-describing per build" + ); + let class = event + .field("decline_class") + .expect("a decline event carries its class"); + let row = event.field("row"); + let mut expected: Vec = ["feature_build", "row", "decline_class", "reason", "message"] + .iter() + .map(|n| n.to_string()) + .collect(); + let channel_carrying = class != "auth" && !is_global_row(row); + if channel_carrying { + assert!( + event.field("channel_id").is_some(), + "a channel decline carries the wire channel id: {event:?}" + ); + expected.push("channel_id".to_owned()); + } else { + assert!( + event.field("channel_id").is_none(), + "an auth/global decline has no channel id: {event:?}" + ); + } + if row == Some("exec_request") { + expected.push("exec_route".to_owned()); + assert!( + event.field("exec_route").is_some(), + "an exec-row decline carries the parse route: {event:?}" + ); + } + let mut names: Vec = event.fields.iter().map(|(n, _)| n.clone()).collect(); + names.sort(); + expected.sort(); + assert_eq!( + names, expected, + "a decline carries exactly the vocabulary, nothing more (no dynamic fields)" + ); +} + +// --------------------------------------------------------------------------- +// Shared rows — run in every feature build. +// --------------------------------------------------------------------------- + +/// - *Contract stamp:* ADR-003 §5 (x11 is a never-feature — the open +/// declines in every build; the event pins the deciding context). +#[tokio::test] +async fn x11_open_decline_emits_the_never_feature_event() { + let session = authenticated("x11-open-row").await; + open_x11_channel(&session.client, "127.0.0.1", 0) + .await + .expect_err("x11 opens decline"); + let event = await_event(&[ + ("row", "x11"), + ("decline_class", "open"), + ("reason", "never-feature"), + ]) + .await; + assert_decline(&event, &[("row", "x11"), ("reason", "never-feature")]); + session.unmount().await; +} + +/// - *Contract stamp:* mapping.md §Global-request rows (the `-R` +/// listen form declines — the event carries no channel id: global +/// requests have none). +#[tokio::test] +async fn tcpip_forward_decline_emits_the_global_request_event() { + let session = authenticated("forward-row").await; + tcpip_forward(&session.client, "127.0.0.1", 8080) + .await + .expect_err("the listen request declines"); + let event = await_event(&[ + ("row", "tcpip_forward"), + ("decline_class", "request"), + ("reason", "unimplemented"), + ]) + .await; + assert_decline( + &event, + &[("row", "tcpip_forward"), ("reason", "unimplemented")], + ); + session.unmount().await; +} + +/// - *Contract stamp:* ADR-004 (the bridge rejects fail-closed — +/// `None` resolution → reject disposition; the event pins the +/// decline without exposing who). +#[tokio::test] +async fn unresolved_fingerprint_auth_decline_emits_the_auth_event() { + let mut session = mounted().await; + let result = + authenticate_publickey(&mut session.client, "unregistered", fixture("charlie")).await; + assert!(!result.success(), "the unregistered fixture rejects"); + let event = await_event(&[ + ("row", "auth_publickey"), + ("decline_class", "auth"), + ("reason", "fingerprint-unresolved"), + ]) + .await; + assert_decline( + &event, + &[ + ("row", "auth_publickey"), + ("reason", "fingerprint-unresolved"), + ], + ); + session.unmount().await; +} + +/// - *Contract stamp:* ADR-004 decision 3 (auth_none: explicit +/// family-contract decline). +#[tokio::test] +async fn auth_none_decline_emits_the_auth_event() { + let mut session = mounted().await; + let result = session + .client + .authenticate_none("observe-none") + .await + .expect("the auth exchange completes"); + assert!( + !result.success(), + "auth_none rejects (the explicit disposition)" + ); + let event = await_event(&[ + ("row", "auth_none"), + ("decline_class", "auth"), + ("reason", "method-unimplemented"), + ]) + .await; + assert_decline( + &event, + &[("row", "auth_none"), ("reason", "method-unimplemented")], + ); + session.unmount().await; +} + +/// - *Contract stamp:* auth.md §Dispositions (log exposure is +/// deployment policy — the crate's events are safe-by-default: no +/// username, secret, or fingerprint ever enters an event field or +/// message; `w3-decline-tracing`'s safety posture, pinned by +/// scanning every captured event across the whole binary's rows). +#[tokio::test] +async fn no_captured_event_ever_carries_a_username_or_fingerprint() { + let session = authenticated("safety-row").await; + open_x11_channel(&session.client, "127.0.0.1", 0) + .await + .expect_err("x11 declines"); + session.unmount().await; + + await_event(&[("row", "x11")]).await; + for event in captured() { + let message = event.field("message").unwrap_or(""); + for username in [ + "x11-open-row", + "forward-row", + "fp-row", + "none-row", + "safety-row", + "unregistered", + "observe-none", + ] { + assert!( + !message.contains(username), + "no event message names the username, got {message:?}" + ); + } + for (name, value) in &event.fields { + let lowered = name.to_lowercase(); + assert!( + !lowered.contains("user"), + "no field ever names a username-like input: {name}" + ); + assert!( + !lowered.contains("fingerprint"), + "the fingerprint is not an event field (auth.md: safe-by-default), got {name}" + ); + assert!( + !value.contains("SHA256:"), + "no event value carries a fingerprint, got {value:?}" + ); + for username in ["x11-open-row", "fp-row", "unregistered"] { + assert!( + !value.contains(username), + "no event value names the username, got {value:?}" + ); + } + } + } +} + +// --------------------------------------------------------------------------- +// Feature-build-specific rows (the dispatch_core.rs convention). +// --------------------------------------------------------------------------- + +#[cfg(not(any(feature = "session", feature = "git", feature = "tunnel")))] +mod default_build { + use super::*; + use common::harness::open_session_channel; + + /// - *Contract stamp:* mapping.md §Decline matrix (the none-build + /// row: everything declines — the event answers the booked + /// finding: the decline's deciding context is observable). + #[tokio::test] + async fn unfeatured_session_open_emits_the_open_event() { + let session = authenticated("none-build-row").await; + open_session_channel(&session.client) + .await + .expect_err("a default build declines session channels"); + let event = await_event(&[ + ("row", "session"), + ("decline_class", "open"), + ("reason", "unfeatured"), + ]) + .await; + assert_decline(&event, &[("row", "session"), ("reason", "unfeatured")]); + session.unmount().await; + } +} + +#[cfg(all(feature = "git", not(feature = "session"), not(feature = "tunnel")))] +mod git_build { + use super::*; + use common::harness::{ + open_session_channel, request_agent_forward, request_env, request_exec, request_pty, + request_subsystem, request_x11, + }; + + /// - *Contract stamp:* ADR-006 decision 4 (spawn-and-return — the + /// serving path is spawned; the lifecycle events started / + /// decline / ended ride the spawned task, in wiring order). + #[tokio::test] + async fn git_seam_stub_declines_with_the_full_serving_lifecycle() { + let session = authenticated("git-stub-row").await; + let mut channel = open_session_channel(&session.client) + .await + .expect("session channels open where a session-serving feature exists"); + request_exec(&channel, "git-upload-pack '/owner/demo'") + .await + .expect("the exec request rides the wire"); + expect_failure_then_close(&mut channel).await; + + let decline = await_event(&[ + ("row", "exec_request"), + ("decline_class", "serving"), + ("reason", "seam-stub"), + ("exec_route", "git"), + ]) + .await; + assert_decline(&decline, &[("row", "exec_request"), ("exec_route", "git")]); + await_event(&[("message", "serving path started"), ("exec_route", "git")]).await; + await_event(&[("message", "serving path ended"), ("exec_route", "git")]).await; + session.unmount().await; + } + + #[tokio::test] + async fn unfeatured_pipe_exec_decline_emits_the_request_event_with_the_route() { + let session = authenticated("git-pipe-row").await; + let mut channel = open_session_channel(&session.client) + .await + .expect("session channels open where a session-serving feature exists"); + request_exec(&channel, "uptime") + .await + .expect("the exec rides the wire"); + expect_failure(&mut channel).await; + let event = await_event(&[ + ("row", "exec_request"), + ("decline_class", "request"), + ("reason", "unfeatured"), + ("exec_route", "pipe"), + ]) + .await; + assert_decline(&event, &[("row", "exec_request"), ("exec_route", "pipe")]); + session.unmount().await; + } + + #[tokio::test] + async fn subsystem_env_pty_and_never_feature_request_declines_emit_events() { + let session = authenticated("git-req-row").await; + let mut channel = open_session_channel(&session.client) + .await + .expect("session channels open where a session-serving feature exists"); + + request_subsystem(&channel, "sftp") + .await + .expect("rides the wire"); + expect_failure(&mut channel).await; + let event = await_event(&[ + ("row", "subsystem_request"), + ("decline_class", "request"), + ("reason", "unimplemented"), + ]) + .await; + assert_decline(&event, &[("row", "subsystem_request")]); + + request_pty(&channel).await.expect("rides the wire"); + expect_failure(&mut channel).await; + let event = await_event(&[ + ("row", "pty_request"), + ("decline_class", "request"), + ("reason", "unfeatured"), + ]) + .await; + assert_decline(&event, &[("row", "pty_request")]); + + request_env(&channel, "SOME_VAR", "value") + .await + .expect("rides the wire"); + expect_failure(&mut channel).await; + let event = await_event(&[ + ("row", "env_request"), + ("decline_class", "request"), + ("reason", "declined-by-design"), + ]) + .await; + assert_decline(&event, &[("row", "env_request")]); + + request_x11(&channel).await.expect("rides the wire"); + expect_failure(&mut channel).await; + let event = await_event(&[ + ("row", "x11_request"), + ("decline_class", "request"), + ("reason", "never-feature"), + ]) + .await; + assert_decline(&event, &[("row", "x11_request")]); + + request_agent_forward(&channel) + .await + .expect("rides the wire"); + expect_failure(&mut channel).await; + let event = await_event(&[ + ("row", "agent_request"), + ("decline_class", "request"), + ("reason", "never-feature"), + ]) + .await; + assert_decline(&event, &[("row", "agent_request")]); + + session.unmount().await; + } + + /// - *Contract stamp:* overview.md §Bounded-resources policy (the + /// cap is refused fail-closed at the open; the event carries the + /// cap reason). + #[tokio::test] + async fn cap_exceeded_session_open_emits_the_cap_event() { + let provider = Arc::new(MemIdentityProvider::new()); + provider.register( + fixture("alpha").fingerprint, + identity("observe-cap", &["ssh"]), + ); + let mut limits = Limits::default(); + limits.max_concurrent_channels = 0; + let adapter = alkssh::SshAdapter::new(common::harness::serve_config( + provider, + &[fixture("bravo")], + limits, + )) + .await + .expect("adapter constructs"); + let mut session = drive( + adapter, + Arc::new(MemHostKeyPolicy::with_default(true)), + limits, + ) + .await + .expect("the loopback mounts"); + let result = + authenticate_publickey(&mut session.client, "observe-cap", fixture("alpha")).await; + assert!(result.success(), "the fixture identity authenticates"); + open_session_channel(&session.client) + .await + .expect_err("the zero-cap build refuses session opens"); + + let event = await_event(&[ + ("row", "session"), + ("decline_class", "open"), + ("reason", "cap-exceeded"), + ]) + .await; + assert_decline(&event, &[("row", "session"), ("reason", "cap-exceeded")]); + session.unmount().await; + } +} + +#[cfg(all(feature = "session", not(feature = "git"), not(feature = "tunnel")))] +mod session_build { + use super::*; + use common::harness::{open_session_channel, request_exec, request_pty}; + + /// - *Contract stamp:* ADR-006 decision 4 (the pty serving path is + /// spawned; the lifecycle rides the task — what the wave-3 tty + /// mapping's real path inherits). + #[tokio::test] + async fn pty_seam_stub_declines_with_the_full_serving_lifecycle() { + let session = authenticated("session-pty-row").await; + let mut channel = open_session_channel(&session.client) + .await + .expect("session channels open"); + request_pty(&channel).await.expect("rides the wire"); + expect_failure_then_close(&mut channel).await; + + let decline = await_event(&[ + ("row", "pty_request"), + ("decline_class", "serving"), + ("reason", "seam-stub"), + ]) + .await; + assert_decline(&decline, &[("row", "pty_request")]); + await_event(&[("message", "serving path started"), ("row", "pty_request")]).await; + await_event(&[("message", "serving path ended"), ("row", "pty_request")]).await; + session.unmount().await; + } + + #[tokio::test] + async fn unfeatured_git_exec_decline_emits_the_request_event_with_the_route() { + let session = authenticated("session-git-row").await; + let mut channel = open_session_channel(&session.client) + .await + .expect("session channels open"); + request_exec(&channel, "git-upload-pack '/owner/demo'") + .await + .expect("rides the wire"); + expect_failure(&mut channel).await; + let event = await_event(&[ + ("row", "exec_request"), + ("decline_class", "request"), + ("reason", "unfeatured"), + ("exec_route", "git"), + ]) + .await; + assert_decline(&event, &[("row", "exec_request"), ("exec_route", "git")]); + session.unmount().await; + } + + #[tokio::test] + async fn pipe_seam_stub_declines_with_the_full_serving_lifecycle() { + let session = authenticated("session-pipe-row").await; + let mut channel = open_session_channel(&session.client) + .await + .expect("session channels open"); + request_exec(&channel, "uptime") + .await + .expect("rides the wire"); + expect_failure_then_close(&mut channel).await; + + let decline = await_event(&[ + ("row", "exec_request"), + ("decline_class", "serving"), + ("reason", "seam-stub"), + ("exec_route", "pipe"), + ]) + .await; + assert_decline(&decline, &[("row", "exec_request"), ("exec_route", "pipe")]); + await_event(&[("message", "serving path ended"), ("exec_route", "pipe")]).await; + session.unmount().await; + } +} + +#[cfg(all(feature = "tunnel", not(feature = "session"), not(feature = "git")))] +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). + #[tokio::test] + async fn tunnel_seam_stub_decline_emits_the_connect_failed_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"); + let event = await_event(&[ + ("row", "direct-tcpip"), + ("decline_class", "open"), + ("reason", "connect-failed"), + ]) + .await; + assert_decline( + &event, + &[("row", "direct-tcpip"), ("reason", "connect-failed")], + ); + session.unmount().await; + } + + #[tokio::test] + async fn unfeatured_session_open_in_a_tunnel_build_emits_the_event() { + let session = authenticated("tunnel-none-row").await; + open_session_channel(&session.client) + .await + .expect_err("a tunnel-only build declines session channels"); + let event = await_event(&[ + ("row", "session"), + ("decline_class", "open"), + ("reason", "unfeatured"), + ]) + .await; + assert_decline(&event, &[("row", "session"), ("reason", "unfeatured")]); + session.unmount().await; + } +} + +// --------------------------------------------------------------------------- +// Local drivers (dispatch_core.rs's shapes, copied per-binary). +// --------------------------------------------------------------------------- + +async fn authenticated(username: &str) -> common::harness::LoopbackSession { + capture(); + let provider = Arc::new(MemIdentityProvider::new()); + provider.register(fixture("alpha").fingerprint, identity(username, &["ssh"])); + let adapter = adapter(provider, &[fixture("bravo")]).await; + let mut session = drive( + adapter, + Arc::new(MemHostKeyPolicy::with_default(true)), + Limits::default(), + ) + .await + .expect("the loopback mounts"); + let result = authenticate_publickey(&mut session.client, username, fixture("alpha")).await; + assert!( + result.success(), + "the fixture identity authenticates before every observe row" + ); + session +} + +/// A mounted (kex-complete, **unauthenticated**) loopback session — +/// the auth-decline rows' driver: an auth disposition is observed +/// before any acceptance exists, so these rows must not pre-derive +/// themselves through [`authenticated`] (a second auth exchange on an +/// authenticated session is not a supported russh surface — mounted +/// fresh is the honest shape, matching `auth_bridge.rs`'s rows). +async fn mounted() -> common::harness::LoopbackSession { + capture(); + let provider = Arc::new(MemIdentityProvider::new()); + let adapter = adapter(provider, &[fixture("bravo")]).await; + drive( + adapter, + Arc::new(MemHostKeyPolicy::with_default(true)), + Limits::default(), + ) + .await + .expect("the loopback mounts") +} + +async fn expect_failure(channel: &mut Channel) { + match next_message(channel).await { + Some(ChannelMsg::Failure) => {} + other => panic!("expected a channel failure, got {other:?}"), + } +} + +async fn expect_failure_then_close(channel: &mut Channel) { + expect_failure(channel).await; + assert!( + await_close(channel).await, + "the seam stub closes the channel" + ); + // Yield so the spawned task's post-close events (ended) are likely + // to have landed before the unmount — await_event's polling covers + // the rest. + tokio::time::sleep(Duration::from_millis(50)).await; +}