diff --git a/docs/architecture/client.md b/docs/architecture/client.md index 6e20806..9b2856a 100644 --- a/docs/architecture/client.md +++ b/docs/architecture/client.md @@ -1,6 +1,6 @@ --- status: reviewed -last_updated: 2026-10-28 +last_updated: 2026-10-29 --- # Client — client direction: driving vanilla SSH servers as family ops @@ -99,6 +99,27 @@ half-capability, which the family forbids (ADR-003 symmetry + ## Correctness facts to carry (POC-recorded) +> **The landed op-row shape (`w4-op-rows` — recorded at landing):** +> the exit-*consuming* rows (session exec/pty, git exec) ride the +> **split-halves wait-pump** (`Channel::split()` + a spawned task +> draining `wait()`), never `into_stream()` — the conversion's reader +> side swallows exactly the non-`Data` messages the exit reports ride +> on, so a wait-pump is the only shape that can consume them (the +> same reason `w4-git-exit-close` pinned the split-halves shape +> server-side). Only the exit-less rows (the tunnel directs) use +> `into_stream()`, straight after the typed open — no queue +> consumption ever precedes a conversion, which is POC-1 §6's caveat +> respected by construction. The session rows' family plane is the +> alktty chunk framing both directions (stdin/`CTRL_IN` out, stdout/ +> stderr in, the family exit chunk in-band, last — `exit_code.rs`'s +> two-arm contract: the server arm derives wire reports from family +> codes, the client arm mirrors wire reports into family codes, +> `−signum` through the one table, the `−1` sentinel never widened). +> The op surface carries no russh types: `OpError` mirrors the typed +> open's decline classes (`OpenDeclineReason`), the `wait()`-drain's +> `Failure` reply is `RequestRefused`, and the sftp placeholder +> declines in every build before any wire op (OQ-13). + - `Channel::exec()` (no `request_exec` at this pin); pty/subsystem request failures surface as `ChannelMsg::Failure` via `wait()`, not `Err` from the call — the caller must drain until the expected diff --git a/src/client.rs b/src/client.rs index 087cd8a..ca12402 100644 --- a/src/client.rs +++ b/src/client.rs @@ -191,6 +191,24 @@ impl SshConnection { self.handle.is_closed() } + /// The russh handle the op rows compose into (`w4-op-rows` — the + /// method bodies live in [`crate::client_ops`], a sibling module, + /// so the field stays private to this module). + #[cfg(any(feature = "session", feature = "git", feature = "tunnel"))] + pub(crate) fn russh_handle( + &self, + ) -> &russh::client::Handle { + &self.handle + } + + /// The dial's bounded-resource budget, verbatim — the op rows + /// read the knobs they consume (the exec-command length cap + /// drives the command-carrying rows' client-mirror gate). + #[cfg(any(feature = "session", feature = "git"))] + pub(crate) fn budgets(&self) -> &Limits { + &self.limits + } + /// The explicit clean-close path (OQ-12's raw material; the full /// cancellable-listener lifecycle surface is `w4-lifecycle`'s): /// `Disconnect::ByApplication`, the honest close — the peer diff --git a/src/client_ops.rs b/src/client_ops.rs new file mode 100644 index 0000000..4b2b7da --- /dev/null +++ b/src/client_ops.rs @@ -0,0 +1,874 @@ +//! The client op rows (`w4-op-rows`) — `client.md` §op rows in code: +//! the family op → vanilla translation the reverse-runner shape +//! drives (the hub dialed [`SshConnection`](crate::SshConnection) and +//! now opens every channel itself). **Type symmetry at the seam is +//! the contract:** family-protocol types in, vanilla wire ops out, no +//! vanilla-concept leaks up — the rows' inputs are the mapped +//! sibling crates' types (the alktty session shapes for the exec/pty +//! rows, alkgit's `Service` for the git row) and the surfaces are +//! crate-owned (`RemoteSession` / `TunnelChannel` / `GitPackSession`), +//! the failure taxonomy [`OpError`](crate::OpError). All rows share +//! the russh 0.65 mechanics POC-3 §D.3 recorded: +//! +//! - `Channel::exec()` — there is **no `request_exec` at this pin**. +//! - A request's Success/Failure replies arrive **on the channel +//! queue via `wait()`**, not as `Err` from the request call — every +//! request-carrying row drains until the expected message, +//! `continue`-ing past the interleaved rest, and maps a `Failure` +//! to [`OpError::RequestRefused`](crate::OpError). The drain is +//! message-capped (bounded resources: a peer that never answers +//! cannot loop it unboundedly; the *connection's* wall clock — the +//! dial's always-set [`Limits::inactivity_timeout`](crate::Limits) — +//! bounds the idle waits). +//! - **The typed opens resolve `Err` directly on a server reject** +//! (POC-1 §6) — [`OpError::OpenDeclined`](crate::OpError), family-mirrored +//! [`OpenDeclineReason`](crate::OpenDeclineReason), **and the +//! connection stays usable after the decline** (row-pinned per +//! decline class). +//! - **No channel-queue consumption where a `into_stream()` conversion +//! follows** (POC-1 §6's fresh-`wait()` caveat) — the tunnel rows +//! (the only `into_stream` consumers) go typed-open → straight +//! conversion, nothing between. The exit-**consuming** rows +//! (exec/pty, git) cannot ride `into_stream` at all — the +//! conversion's reader side swallows every non-`Data` message, +//! which is exactly where `ExitStatus`/`ExitSignal` live — so they +//! run the **split-halves wait-pump** (`Channel::split()` + a +//! spawned task draining `wait()`), the shape the server +//! direction's git row landed for the same reason +//! (`w4-git-exit-close`). The exit-ride is the family plane's (§the +//! session rows) or the consumed-report oneshot's (§the git row). +//! - `window_change`/`signal` are fire-and-forget on the wire (no SSH +//! reply exists) — the pumps send and drop the outcome, never +//! awaiting one. +//! +//! ## The family plane (the session rows) +//! +//! The exec/pty session's data plane is the **alktty chunk framing in +//! both directions** — the exact plane the server direction's session +//! bridge speaks ([`crate::tty_map`]), consumed not re-derived: +//! stdin chunks (`STREAM_STDIN`) ride the wire `data` (the +//! zero-length sentinel = wire EOF, the bridge's decided translation — +//! the session keeps pumping after it); `CTRL_IN` chunks translate to +//! the wire request rows (`Resize` → `window_change` (the session +//! dimension clamp is not needed — the family plane is already +//! `u16`), `Signal{name}` → `signal` through the shared table's +//! name→`Sig` map, `Eof` → wire EOF); wire `data` → stdout chunks, +//! wire `extended_data(1)` → stderr chunks; and the exit rides +//! **in-band as the family's exit chunk** (`STREAM_CTRL_OUT`, +//! exit-chunk-is-last — alktty's ADR-004, mapping.md's contract): +//! wire `ExitStatus(N)` → code `N`, wire `ExitSignal(name)` → +//! `−signum` through the shared table's inverse read +//! ([`crate::exit_code`] — the derivation contract's client arm; a +//! name with no table entry rides the `−1` sentinel, the mapping +//! never invents). A session that ends with **no exit report** +//! delivers no exit chunk — the output half just ends (mapping.md's +//! no-exit-chunk close family: "reports nothing"). The session's +//! ttymodes never ride the wire pty request (the OQ-44 +//! reserved-JSON posture mirrored: the door never invents wire +//! terminal modes from family data); a family `term: None` rides the +//! wire's empty-terminal-string value. +//! +//! ## The row inventory +//! +//! | Op | Feature | Vanilla translation | Surface | +//! |---|---|---|---| +//! | pipe-mode exec | `session` | `channel_open_session` + `exec` | [`SshConnection::exec`] | +//! | shell with/without a pty | `session` | `channel_open_session` + `pty_request`? + `shell_request` | [`SshConnection::shell`] | +//! | exec in a pty (`ssh -t host "cmd"`) | `session` | `pty_request` then `exec` | [`SshConnection::pty_exec`] | +//! | tunnel target (`-L`; the alksocks `DialFn`) | `tunnel` | `channel_open_direct_tcpip` | [`SshConnection::tunnel_direct`] | +//! | the streamlocal twin | `tunnel` | `channel_open_direct_streamlocal`, runtime-Unix-gated | [`SshConnection::tunnel_direct_unix`] | +//! | upstream git service | `git` | the shared grammar's command + `exec`; the pack duplex | [`SshConnection::git_service`] | +//! | sftp | *every build* | **none** — the OQ-13 decline placeholder | [`SshConnection::sftp`] | +//! +//! The exec-command **length budget** applies identically to the +//! command-carrying rows (the server direction's pre-parse cap +//! mirrored at the command-construction gate — bounded resources; +//! an unbounded command buffer can never ride a dial): +//! [`OpError::CommandTooLong`](crate::OpError) before any wire op. +//! +//! No client-side in-crate channel cap *admits* ops (the server +//! direction's cap is a wire-battery defense; the client rows are +//! hub-driven — the family app chose the dial): bounded resources on +//! this direction are the config's always-on wall clock/keepalives, +//! the russh windows, the bounded duplexes/mpscs the pumps ride, and +//! the drain cap — every spawn bounded by its own task's end +//! conditions. +//! +//! ## Spawn posture +//! +//! The op methods await their launch (typed open, requests, the +//! reply drains) on the *caller's* task — never the russh packet +//! loop's — so the server direction's spawn-and-return deadlock class +//! does not exist here; the long-lived pumps spawn at launch and end +//! on their terminal conditions (the exit chunk, the channel's end, +//! or the hub dropping the surface — a drop is wire-clean: the input +//! pump's terminal path closes the channel, the pump pair reaping as +//! their duplex halves drop). + +#[cfg(any(feature = "session", feature = "git"))] +use russh::ChannelMsg; +#[cfg(any(feature = "session", feature = "git"))] +use russh::client::Msg; +#[cfg(feature = "git")] +use tokio::io::AsyncWriteExt; + +use crate::client::SshConnection; +#[cfg(feature = "session")] +use crate::client_handler::ClientHandler; +use crate::error::OpError; + +/// The op-row pump duplex capacity — the crate's bridge-pair +/// convention (the wave-2 loopback harness's per-direction capacity), +/// keeping every pump's backpressure bounded. +#[cfg(any(feature = "session", feature = "git"))] +const OP_DUPLEX_CAPACITY: usize = 65_536; + +/// The pump-side read buffer — the SSH-side granularity (per-chunk +/// payloads are bounded by the dial's `maximum_packet_size`). +#[cfg(feature = "git")] +const PUMP_BUF: usize = 16 * 1024; + +/// The request-reply drain's message cap (bounded resources, the +/// §module docs' drain row): a peer answering only with non-reply +/// messages cannot loop the drain unboundedly. +#[cfg(any(feature = "session", feature = "git"))] +const DRAIN_MESSAGE_CAP: usize = 1024; + +/// The originator label the door puts on the wire's `direct-tcpip` +/// opens (the SSH originator fields are what a vanilla client fills +/// in for its own local listener; this door's hub-driven dials carry +/// the door's label and no port — recorded decision: nothing +/// invented, nothing leaked, the peer's logs name the door). +#[cfg(feature = "tunnel")] +const TUNNEL_ORIGINATOR: &str = "alk-ssh"; + +/// The wire exit status → the family `i32` exit-code convention: a +/// code beyond the family `i32` range is unwritable as an exit-chunk +/// code; the clamp rides (defensive corner — vanilla command codes +/// are far below it; the mapping never invents a value, it narrows). +#[cfg(any(feature = "session", feature = "git"))] +fn family_exit_code(status: u32) -> i32 { + status.min(i32::MAX as u32) as i32 +} + +/// The typed-open decline's translation point: the russh error → the +/// op taxonomy (a channel-open failure → +/// [`OpError::OpenDeclined`](crate::OpError::OpenDeclined), +/// family-mirrored; everything else transparent). The russh level's +/// test observable (`POC-1 §6`'s `Err(ChannelOpenFailure)` — the +/// awaiting typed call's resolution) rides this point; the +/// connection stays usable after every decline (row-pinned). +#[cfg(any(feature = "session", feature = "git", feature = "tunnel"))] +fn open_error(error: russh::Error) -> OpError { + OpError::from(error) +} + +/// The request-reply drain (the §module docs' recorded mechanics, +/// shared by the exec/pty rows and the git row): Success/Failure via +/// `wait()` — never an `Err` from the request call — `continue` past +/// the interleaved rest (the POC-3 §D.3 break-on-first-message trap, +/// not repeated), message-capped. `Failure` is the wire +/// request-refusal class ([`OpError::RequestRefused`]); the +/// channel's end ahead of any reply is [`OpError::ChannelEnded`]. +#[cfg(any(feature = "session", feature = "git"))] +async fn drain_request_reply(channel: &mut russh::Channel) -> Result<(), OpError> { + for _ in 0..DRAIN_MESSAGE_CAP { + match channel.wait().await { + Some(ChannelMsg::Success) => return Ok(()), + Some(ChannelMsg::Failure) => return Err(OpError::RequestRefused), + Some(ChannelMsg::Close) => return Err(OpError::ChannelEnded), + Some(_) => continue, + None => return Err(OpError::ChannelEnded), + } + } + Err(OpError::ChannelEnded) +} + +/// Drains one request's reply; on any refusal, closes the channel +/// behind it (the row owned the channel; the launch never +/// half-serves) and returns the error. The launch rows' shared step. +#[cfg(any(feature = "session", feature = "git"))] +async fn drain_launch_request(channel: &mut russh::Channel) -> Result<(), OpError> { + let outcome = drain_request_reply(channel).await; + if outcome.is_err() { + let _ = channel.close().await; + } + outcome +} + +// --------------------------------------------------------------------------- +// The sftp placeholder — every build (OQ-13; no feature, no dep) +// --------------------------------------------------------------------------- + +impl SshConnection { + /// The sftp op — the client-side half of the OQ-13 decline + /// placeholder: **declines in every build, before any wire op** + /// (never a half-opened subsystem channel). The feature exists + /// and ships in the build string; the route does not, and arrives + /// with alkfs as its own ADR — both directions' placeholders are + /// then replaced by real rows (`client.md` §What the crate does + /// *not* include). Compiles feature-free; the `Ok` type is + /// `Infallible` by design — nothing can ever populate it here. + pub async fn sftp(&self) -> Result { + Err(OpError::SftpNotServed) + } +} + +// --------------------------------------------------------------------------- +// The session rows — exec / pty (under `session`) +// --------------------------------------------------------------------------- + +/// A remote **session** — the family session plane over the runner +/// (the session rows' surface): the alktty chunk framing in both +/// directions, exactly the plane the server direction's session +/// bridge speaks ([`crate::tty_map`]), reversed. Two duplex halves +/// the hub drives; the pumps are spawned at the op's launch (§the +/// family plane in the module docs for the full chunk map). Dropping +/// the session (both halves) is wire-clean: the input pump's +/// terminal path closes the channel. +#[cfg(feature = "session")] +pub struct RemoteSession { + /// The hub→wire chunk plane: stdin chunks + `CTRL_IN` controls + /// (`alktty::wire::ChunkWriter` framing; the zero-length stdin + /// chunk = wire EOF; `Resize`/`Signal`/`Eof` controls = the wire + /// request rows). + pub input: tokio::io::DuplexStream, + /// The wire→hub chunk plane: stdout/stderr chunks + the exit + /// chunk in-band, last (exit-chunk-is-last). + pub output: tokio::io::DuplexStream, +} + +#[cfg(feature = "session")] +impl std::fmt::Debug for RemoteSession { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RemoteSession").finish_non_exhaustive() + } +} + +/// Which request launches a session — the family op's payload +/// (§the row inventory; pipe-mode exec vs the shell form). +#[cfg(feature = "session")] +enum LaunchRequest { + Exec(String), + Shell, +} + +#[cfg(feature = "session")] +impl SshConnection { + /// The pipe-mode exec row: `ssh host "cmd"` in, family-typed — + /// the session channel's open + the wire `exec` request, then the + /// chunk-framed session (§the family plane; stderr separated). + /// The command rides the wire verbatim (the remote end's shell + /// parses it — the vanilla exec grammar; never argv-split). + pub async fn exec(&self, command: impl Into) -> Result { + let command = command.into(); + if command.len() > self.budgets().max_exec_command_len { + return Err(OpError::CommandTooLong); + } + open_remote_session(self.russh_handle(), None, LaunchRequest::Exec(command)).await + } + + /// The pty-mode exec row (`ssh -t host "cmd"`): the pty request + /// carries the family [`TerminalParamsWire`](alktty::negotiation:: + /// TerminalParamsWire) first (its acceptance drained), then the + /// exec. **A refused pty refuses the whole row** — never a + /// half-capability, never a silent pty-less fall. + pub async fn pty_exec( + &self, + params: alktty::negotiation::TerminalParamsWire, + command: impl Into, + ) -> Result { + let command = command.into(); + if command.len() > self.budgets().max_exec_command_len { + return Err(OpError::CommandTooLong); + } + open_remote_session( + self.russh_handle(), + Some(params), + LaunchRequest::Exec(command), + ) + .await + } + + /// The shell row ("own a shell on the runner"): with + /// `params` — the pty mode (the interactive `ssh` shape); with + /// `None` — the pipe-mode shell (the `ssh -T` shape; a pty-less + /// shell request is legitimate vanilla and served, never declined, + /// mirroring the server direction's decided parity). + pub async fn shell( + &self, + params: Option, + ) -> Result { + open_remote_session(self.russh_handle(), params, LaunchRequest::Shell).await + } +} + +/// The session rows' launcher (§module docs): typed open → the pty? → +/// the launch request (each request-reply drained; a refusal closes +/// the channel behind itself) → the split-halves pumps. +#[cfg(feature = "session")] +async fn open_remote_session( + handle: &russh::client::Handle, + pty: Option, + request: LaunchRequest, +) -> Result { + let mut channel = handle.channel_open_session().await.map_err(open_error)?; + + if let Some(params) = &pty { + channel + .request_pty( + true, + params.term.as_deref().unwrap_or_default(), + u32::from(params.cols), + u32::from(params.rows), + u32::from(params.pixel_width), + u32::from(params.pixel_height), + &[], + ) + .await + .map_err(open_error)?; + drain_launch_request(&mut channel).await?; + } + + match request { + LaunchRequest::Exec(command) => { + channel + .exec(true, command.into_bytes()) + .await + .map_err(open_error)?; + } + LaunchRequest::Shell => { + channel.request_shell(true).await.map_err(open_error)?; + } + } + drain_launch_request(&mut channel).await?; + + let (read_half, write_half) = channel.split(); + let (kill, kill_rx) = tokio::sync::watch::channel(false); + let (in_tx, in_rx) = tokio::io::duplex(OP_DUPLEX_CAPACITY); + let (out_tx, out_rx) = tokio::io::duplex(OP_DUPLEX_CAPACITY); + tokio::spawn(session_input_pump(write_half, in_rx, kill_rx)); + tokio::spawn(session_output_pump(read_half, out_tx, kill)); + Ok(RemoteSession { + input: in_tx, + output: out_rx, + }) +} + +/// The session row's **input pump**: the hub's chunk plane → the +/// wire (§the family plane). The zero-length stdin sentinel / a +/// `CTRL_IN` `Eof` / the hub's input half dropping = the wire EOF +/// *once*, then the idle-for-kill state (the session keeps pumping); +/// the terminal kill (from the output pump's terminal paths) → the +/// wire `channel_close` — the session's drop is wire-clean. +#[cfg(feature = "session")] +async fn session_input_pump( + write_half: russh::ChannelWriteHalf, + input: tokio::io::DuplexStream, + mut kill: tokio::sync::watch::Receiver, +) { + let mut done = false; + let mut reader = alktty::wire::ChunkReader::new(input); + loop { + tokio::select! { + frame = hub_frame(&mut reader), if !done => { + match frame { + HubFrame::Chunk(chunk) => match chunk.stream_type { + alktty::wire::STREAM_STDIN => { + if chunk.bytes.is_empty() { + let _ = write_half.eof().await; + done = true; + } else if write_half.data_bytes(chunk.bytes).await.is_err() { + return; + } + } + alktty::wire::STREAM_CTRL_IN => { + match alktty::ControlMessage::from_slice(&chunk.bytes) { + Ok(alktty::ControlMessage::Resize { + cols, + rows, + pixel_width, + pixel_height, + }) => { + let _ = write_half + .window_change( + u32::from(cols), + u32::from(rows), + u32::from(pixel_width), + u32::from(pixel_height), + ) + .await; + } + Ok(alktty::ControlMessage::Signal { name }) => { + let _ = write_half + .signal(crate::exit_code::sig_from_name(&name)) + .await; + } + Ok(alktty::ControlMessage::Eof) => { + let _ = write_half.eof().await; + done = true; + } + _ => {} + } + } + _ => {} + }, + HubFrame::Break => { + let _ = write_half.close().await; + return; + } + HubFrame::Closed => { + let _ = write_half.eof().await; + done = true; + } + } + } + _ = kill.changed() => { + let _ = write_half.close().await; + return; + } + } + } +} + +/// One frame read from the hub→door plane — the *no-disambiguation* +/// read direction: this plane's chunks are the family protocol's own +/// (stdin `0` / `CTRL_IN` `3`); no negotiation-error frames exist +/// toward the door, so the read is the protocol's direct +/// [`ChunkReader::read_chunk`] (a `0x00`-leading frame here is a +/// zero-length stdin chunk — the sentinel — and reads as exactly +/// that). A rogue frame is the bounds-checked [`RawError`], mapped to +/// the pump's [`HubFrame::Break`]. +#[cfg(feature = "session")] +enum HubFrame { + Chunk(alktty::wire::Chunk), + Closed, + Break, +} + +#[cfg(feature = "session")] +async fn hub_frame(reader: &mut alktty::wire::ChunkReader) -> HubFrame { + match reader.read_chunk().await { + Ok(chunk) => HubFrame::Chunk(chunk), + Err(alktty::wire::RawError::ConnectionClosed) => HubFrame::Closed, + Err(_) => HubFrame::Break, + } +} + +/// The session row's **output pump**: the wire → the hub's chunk +/// plane (§the family plane). `data` → stdout chunks, +/// `extended_data(1)` → stderr chunks, and the `ExitStatus` / +/// `ExitSignal` terminal → the in-band exit chunk (last) — then the +/// kill fires and the input pump closes the wire. A channel end with +/// no report → no exit chunk; the output half just ends. +#[cfg(feature = "session")] +async fn session_output_pump( + mut read_half: russh::ChannelReadHalf, + output: tokio::io::DuplexStream, + kill: tokio::sync::watch::Sender, +) { + let mut writer = alktty::wire::ChunkWriter::new(output); + loop { + match read_half.wait().await { + None | Some(ChannelMsg::Close) => { + let _ = kill.send(true); + return; + } + Some(ChannelMsg::Data { data }) => { + let chunk = alktty::wire::Chunk { + stream_type: alktty::wire::STREAM_STDOUT, + bytes: data, + }; + if writer.write_chunk(&chunk).await.is_err() { + let _ = kill.send(true); + return; + } + } + Some(ChannelMsg::ExtendedData { data, ext: 1 }) => { + let chunk = alktty::wire::Chunk { + stream_type: alktty::wire::STREAM_STDERR, + bytes: data, + }; + if writer.write_chunk(&chunk).await.is_err() { + let _ = kill.send(true); + return; + } + } + Some(ChannelMsg::ExitStatus { exit_status }) => { + write_exit_chunk(&mut writer, family_exit_code(exit_status)).await; + let _ = kill.send(true); + return; + } + Some(ChannelMsg::ExitSignal { signal_name, .. }) => { + let name = crate::exit_code::sig_control_name(&signal_name); + write_exit_chunk(&mut writer, crate::exit_code::signal_code_from_name(&name)).await; + let _ = kill.send(true); + return; + } + Some(ChannelMsg::Eof) => continue, + Some(_) => {} + } + } +} + +/// The family exit chunk's write — `{"type":"exit","code":N}` as the +/// in-band terminal chunk (exit-chunk-is-last), serialized by the +/// family currency's own serializer. An unserializable exit (not a +/// reachable shape) writes nothing — the no-report family, never a +/// corrupted frame. +#[cfg(feature = "session")] +async fn write_exit_chunk( + writer: &mut alktty::wire::ChunkWriter, + code: i32, +) { + if let Ok(bytes) = (alktty::ControlMessage::Exit { code }).to_json() { + let _ = writer + .write_chunk(&alktty::wire::Chunk { + stream_type: alktty::wire::STREAM_CTRL_OUT, + bytes, + }) + .await; + } +} + +// --------------------------------------------------------------------------- +// The tunnel rows — direct-tcpip / direct-streamlocal (under `tunnel`) +// --------------------------------------------------------------------------- + +/// A tunnel channel — the tunnel rows' surface: a plain byte duplex +/// over the open channel (`-L`'s pump; the alksocks `DialFn` composes +/// exactly this type). Nothing vanilla-typed rides the family +/// surface — the duplex's `poll_shutdown` carries the wire EOF. +#[cfg(feature = "tunnel")] +pub struct TunnelChannel { + stream: russh::ChannelStream, +} + +#[cfg(feature = "tunnel")] +impl std::fmt::Debug for TunnelChannel { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("TunnelChannel").finish_non_exhaustive() + } +} + +#[cfg(feature = "tunnel")] +impl tokio::io::AsyncRead for TunnelChannel { + fn poll_read( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + std::pin::Pin::new(&mut self.stream).poll_read(cx, buf) + } +} + +#[cfg(feature = "tunnel")] +impl tokio::io::AsyncWrite for TunnelChannel { + fn poll_write( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &[u8], + ) -> std::task::Poll> { + std::pin::Pin::new(&mut self.stream).poll_write(cx, buf) + } + + fn poll_flush( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::pin::Pin::new(&mut self.stream).poll_flush(cx) + } + + fn poll_shutdown( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::pin::Pin::new(&mut self.stream).poll_shutdown(cx) + } +} + +#[cfg(feature = "tunnel")] +impl SshConnection { + /// The tunnel-target row (`-L`; the alksocks `DialFn` composes + /// over it): the typed open — server reject → + /// [`OpError::OpenDeclined`](crate::OpError) with the mirrored + /// class, and the connection stays usable (row-pinned per + /// decline class). Between the open and the `into_stream()` + /// conversion **nothing else happens** — no channel-queue + /// consumption (POC-1 §6's caveat, respected by construction; the + /// data pump is the raw byte duplex). + pub async fn tunnel_direct( + &self, + host: impl Into, + port: u32, + ) -> Result { + let channel = self + .russh_handle() + .channel_open_direct_tcpip(host, port, TUNNEL_ORIGINATOR, 0) + .await + .map_err(open_error)?; + Ok(TunnelChannel { + stream: channel.into_stream(), + }) + } + + /// The streamlocal twin (`direct-streamlocal@openssh.com`): same + /// typed-open/pump shape, the OQ-11 runtime-Unix gate mirrored — + /// **compiles wherever `tunnel` compiles**, and off-Unix the op + /// declines cleanly ([`OpError::UnixUnavailable`](crate::OpError)) + /// before any wire op. + pub async fn tunnel_direct_unix( + &self, + socket_path: impl Into, + ) -> Result { + #[cfg(not(unix))] + { + let _ = socket_path; + Err(OpError::UnixUnavailable) + } + #[cfg(unix)] + { + let channel = self + .russh_handle() + .channel_open_direct_streamlocal(socket_path) + .await + .map_err(open_error)?; + Ok(TunnelChannel { + stream: channel.into_stream(), + }) + } + } +} + +// --------------------------------------------------------------------------- +// The git row — upstream exec consumption (under `git`) +// --------------------------------------------------------------------------- + +/// An upstream **git service channel** — the git row's surface: the +/// raw pack duplex (the git protocol's own byte plane rides the +/// channel untouched), the sshd-side stderr split out, and the +/// consumed exit report at the family's `i32` exit-code convention +/// (`ExitStatus(N)` → `N`; `ExitSignal` → `−signum` through the +/// shared table — the mapping never invents; a receive `Err` on +/// `exit` = the pump ended with **no exit report at all**, the +/// no-exit-chunk close family's mirror). +#[cfg(feature = "git")] +pub struct GitPackSession { + /// The pack bytes the hub pushes to the wire (request lines, + /// haves, the pushed pack). Ending the half = wire EOF. + pub input: tokio::io::DuplexStream, + /// The wire `data` → the pack bytes (the advertisement, acks, + /// the pack). + pub output: tokio::io::DuplexStream, + /// The wire `extended_data(1)` (the remote program's stderr + /// diagnostics) — raw bytes; a hub uninterested in diagnostics + /// drops the half (the pump then discards further stderr). + pub stderr: tokio::io::DuplexStream, + /// The consumed exit report (see the type docs). + pub exit: tokio::sync::oneshot::Receiver, +} + +#[cfg(feature = "git")] +impl std::fmt::Debug for GitPackSession { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("GitPackSession").finish_non_exhaustive() + } +} + +#[cfg(feature = "git")] +impl SshConnection { + /// The upstream git row: the shared exec grammar's command + /// (`git-{upload,receive}-pack '/'` — built through + /// [`crate::dispatch`]'s construction arm, the parse arm's exact + /// inverse; no forked grammar), pointed at the vanilla `sshd`: + /// typed open + `exec`, the pack duplex pumped, the exit report + /// consumed. + /// + /// **The pre-wire gates, riding the shared grammar's verdicts:** + /// the repo id validates against the ADR-008 grammar (the git + /// row's server-direction gate mirrored, + /// [`crate::git_map`]) → [`OpError::InvalidRepoId`](crate::OpError); + /// the constructed command is checked against the exec-length + /// budget ([`Limits::max_exec_command_len`]) → + /// [`OpError::CommandTooLong`](crate::OpError) — both **before any + /// wire op**. (The grammar's other decline classes — + /// `unknown-service` / `service-only` — are structurally + /// unreachable here: the [`alkgit::Service`] input type carries + /// only the two serving services; the construction arm can emit + /// no other token.) + pub async fn git_service( + &self, + service: alkgit::Service, + repo_id: &str, + ) -> Result { + let route = match service { + alkgit::Service::UploadPack => crate::dispatch::GitServiceRoute::UploadPack, + alkgit::Service::ReceivePack => crate::dispatch::GitServiceRoute::ReceivePack, + }; + if !crate::git_map::repo_id_validates(repo_id) { + return Err(OpError::InvalidRepoId); + } + let command = crate::dispatch::build_git_exec_command(route, repo_id); + if command.len() > self.budgets().max_exec_command_len { + return Err(OpError::CommandTooLong); + } + + let mut channel = self + .russh_handle() + .channel_open_session() + .await + .map_err(open_error)?; + channel + .exec(true, command.into_bytes()) + .await + .map_err(open_error)?; + drain_launch_request(&mut channel).await?; + + let (read_half, write_half) = channel.split(); + let (kill, kill_rx) = tokio::sync::watch::channel(false); + let (in_tx, in_rx) = tokio::io::duplex(OP_DUPLEX_CAPACITY); + let (out_tx, out_rx) = tokio::io::duplex(OP_DUPLEX_CAPACITY); + let (err_tx, err_rx) = tokio::io::duplex(OP_DUPLEX_CAPACITY); + let (exit_tx, exit_rx) = tokio::sync::oneshot::channel(); + tokio::spawn(git_input_pump(write_half, in_rx, kill_rx)); + tokio::spawn(git_output_pump(read_half, out_tx, err_tx, exit_tx, kill)); + Ok(GitPackSession { + input: in_tx, + output: out_rx, + stderr: err_rx, + exit: exit_rx, + }) + } +} + +/// The git row's **raw input pump**: the hub's pack bytes → the wire; +/// the input half's end (or any read error) = the wire EOF *once*, +/// then the idle-for-kill state (the advertisement/acks/exit keep +/// flowing); the kill (the output pump's terminal paths) → the wire +/// `channel_close`. +#[cfg(feature = "git")] +async fn git_input_pump( + write_half: russh::ChannelWriteHalf, + mut input: tokio::io::DuplexStream, + mut kill: tokio::sync::watch::Receiver, +) { + let mut done = false; + let mut buf = vec![0u8; PUMP_BUF]; + loop { + if done { + if kill.changed().await.is_err() { + let _ = write_half.close().await; + return; + } + let _ = write_half.close().await; + return; + } + tokio::select! { + n = tokio::io::AsyncReadExt::read(&mut input, &mut buf) => { + match n { + Ok(0) | Err(_) => { + let _ = write_half.eof().await; + done = true; + } + Ok(count) => { + if write_half.data(&buf[..count]).await.is_err() + { + return; + } + } + } + } + _ = kill.changed() => { + let _ = write_half.close().await; + return; + } + } + } +} + +/// The git row's **output pump**: the wire → the pack duplex + +/// stderr; the exit report consumed here (the family code) and the +/// kill fired — then the input pump closes the wire. A channel end +/// with no report → no exit fired; the `exit` oneshot drops (the +/// hub's receive reads the no-report mirror). +#[cfg(feature = "git")] +async fn git_output_pump( + mut read_half: russh::ChannelReadHalf, + mut output: tokio::io::DuplexStream, + mut stderr: tokio::io::DuplexStream, + exit: tokio::sync::oneshot::Sender, + kill: tokio::sync::watch::Sender, +) { + let mut stderr_gone = false; + loop { + match read_half.wait().await { + None | Some(ChannelMsg::Close) => { + let _ = kill.send(true); + return; + } + Some(ChannelMsg::Data { data }) => { + if output.write_all(&data).await.is_err() { + let _ = kill.send(true); + return; + } + } + Some(ChannelMsg::ExtendedData { data, ext: 1 }) => { + if !stderr_gone && stderr.write_all(&data).await.is_err() { + stderr_gone = true; + } + } + Some(ChannelMsg::ExitStatus { exit_status }) => { + let _ = exit.send(family_exit_code(exit_status)); + let _ = kill.send(true); + return; + } + Some(ChannelMsg::ExitSignal { signal_name, .. }) => { + let name = crate::exit_code::sig_control_name(&signal_name); + let _ = exit.send(crate::exit_code::signal_code_from_name(&name)); + let _ = kill.send(true); + return; + } + Some(ChannelMsg::Eof) => continue, + Some(_) => {} + } + } +} + +#[cfg(all(test, any(feature = "session", feature = "git", feature = "tunnel")))] +mod tests { + use super::{family_exit_code, open_error}; + use crate::error::OpError; + use russh::ChannelOpenFailure; + + /// - *Contract stamp:* `client.md` §op rows (no vanilla-concept + /// leaks up) + POC-1 §6 — the typed open's decline mapping + /// preserves the wire class (`error.rs`'s mirror rows; this + /// pins the op rows' own consumption point). + #[test] + fn typed_open_decline_maps_through_the_mirror() { + let mapped = open_error(russh::Error::ChannelOpenFailure( + ChannelOpenFailure::ConnectFailed, + )); + assert!(matches!( + mapped, + OpError::OpenDeclined(crate::error::OpenDeclineReason::ConnectFailed) + )); + let mapped = open_error(russh::Error::ChannelOpenFailure( + ChannelOpenFailure::AdministrativelyProhibited, + )); + assert!(matches!( + mapped, + OpError::OpenDeclined(crate::error::OpenDeclineReason::Prohibited) + )); + } + + /// - *Contract stamp:* the family exit-currency clamp (§the + /// family plane) — ordinary codes ride identity, beyond-range + /// ones narrow (never invent). + #[test] + fn family_exit_code_narrows_beyond_the_i32_range() { + assert_eq!(family_exit_code(0), 0); + assert_eq!(family_exit_code(255), 255); + assert_eq!(family_exit_code(u32::MAX), i32::MAX); + } +} diff --git a/src/dispatch.rs b/src/dispatch.rs index 400f932..93e5cfc 100644 --- a/src/dispatch.rs +++ b/src/dispatch.rs @@ -174,6 +174,26 @@ pub(crate) fn parse_exec(command: &str) -> ExecRoute { } } +/// The grammar's construction arm (`w4-op-rows`) — the client +/// direction's mirror of [`parse_exec`]: the git route + repo id → +/// the vanilla wire command string, byte-for-byte the shape the +/// grammar parses back (`git-{upload,receive}-pack '/'` — the +/// fully-single-quote-wrapped path with the leading `/`, POC-3 §A's +/// verified shape). The client row reuses — never forks — the +/// grammar: it hands the constructed command back through +/// [`parse_exec`] before any wire op, so the shared verdicts (the +/// per-length cap, the decline classes) ride both arms +/// (`client.md` §op rows: "the same exec grammar the server side +/// parses, reused, not forked"). +#[cfg(feature = "git")] +pub(crate) fn build_git_exec_command(service: GitServiceRoute, repo_id: &str) -> String { + let token = match service { + GitServiceRoute::UploadPack => "git-upload-pack", + GitServiceRoute::ReceivePack => "git-receive-pack", + }; + format!("{} '/{}'", token, repo_id) +} + /// The argument-grammar stage: a path-shaped argument (fully wrapped /// in single quotes — the vanilla wire shape — or unquoted absolute; /// anything else is not git) normalized to the repo path: quotes @@ -395,4 +415,56 @@ mod tests { assert_eq!(route("\u{0}\u{1}"), ExecRoute::Pipe); assert_eq!(git_repo("git-upload-pack '/\u{e9}'"), "\u{e9}"); } + + /// - *Contract stamp:* `client.md` §op rows (the same exec + /// grammar the server side parses, reused not forked); + /// ADR-006 decision 2 — the construction arm's command parses + /// back through [`parse_exec`] to exactly the route it was + /// built from, for both services and every grammar shape. + #[cfg(feature = "git")] + #[test] + fn the_construction_arm_round_trips_through_the_parse_arm() { + for (service, repo) in [ + (GitServiceRoute::UploadPack, "owner/demo"), + (GitServiceRoute::UploadPack, "owner/demo.git"), + (GitServiceRoute::ReceivePack, "r"), + (GitServiceRoute::UploadPack, "a/b/c"), + (GitServiceRoute::ReceivePack, "u\u{e9}"), + ] { + let command = super::build_git_exec_command(service, repo); + match super::parse_exec(&command) { + super::ExecRoute::Git { + service: parsed, + repo: parsed_repo, + } => { + assert_eq!(parsed, service, "{command:?}"); + // the round-trip holds past the one `.git` suffix + // strip (the construction pre-applies it: the + // built command never carries a `.git` suffix) + assert!( + parsed_repo == repo || (parsed_repo == repo.strip_suffix(".git").unwrap()), + "{command:?} → {parsed_repo:?}" + ); + } + other => panic!("{command:?} must parse back as git, got {other:?}"), + } + } + } + + /// - *Contract stamp:* the construction arm is grammar-loyal — + /// it is only ever reachable for the two *serving* services + /// (the type is the gate); the wire shape is fully + /// single-quote-wrapped with the leading `/`. + #[cfg(feature = "git")] + #[test] + fn the_construction_arm_emits_the_wire_shape() { + assert_eq!( + super::build_git_exec_command(GitServiceRoute::UploadPack, "owner/demo"), + "git-upload-pack '/owner/demo'" + ); + assert_eq!( + super::build_git_exec_command(GitServiceRoute::ReceivePack, "r"), + "git-receive-pack '/r'" + ); + } } diff --git a/src/error.rs b/src/error.rs index 0230e52..810469a 100644 --- a/src/error.rs +++ b/src/error.rs @@ -231,9 +231,143 @@ impl From for DialError { } } +/// The family-typed mirror of the SSH `CHANNEL_OPEN` rejection +/// reasons (RFC 4250 §1.5 / RFC 4254 §5.2) — the op rows translate +/// the wire's typed-open decline into this closed vocabulary +/// (`client.md` §op rows: the family surface carries no vanilla types). +/// The mapping is one-to-one with russh's +/// [`russh::ChannelOpenFailure`], whose `Other` arm mirrors its +/// `code`/`reason` pair verbatim — a decline reason the client +/// *could not map* is never silently collapsed onto a known class. +#[derive(Debug, Clone, Error, PartialEq, Eq)] +pub enum OpenDeclineReason { + /// The SSH `administratively prohibited` class — a policy + /// refusal at the server (the door's own unfeatured/cap + /// declines answer with this family's server-direction mirror). + #[error("the server refused the open (administratively prohibited)")] + Prohibited, + + /// The SSH `connect failed` class — the target could not be + /// bridged (a dead `-L` target, an impossible one). + #[error("the server rejected the open (connect failed)")] + ConnectFailed, + + /// The SSH `unknown channel type` class. + #[error("the server rejected the open (unknown channel type)")] + UnknownChannelType, + + /// The SSH `resource shortage` class. + #[error("the server rejected the open (resource shortage)")] + ResourceShortage, + + /// A rejection reason with no standard code — the wire's own + /// `code`/`reason` pair rides (the mirror never invents a class + /// for an unmappable one). + #[error("the server rejected the open (code {code}: {reason})")] + Other { code: u32, reason: String }, +} + +impl From for OpenDeclineReason { + /// The one-to-one mirror (see the type docs). + fn from(failure: russh::ChannelOpenFailure) -> Self { + match failure { + russh::ChannelOpenFailure::AdministrativelyProhibited => Self::Prohibited, + russh::ChannelOpenFailure::ConnectFailed => Self::ConnectFailed, + russh::ChannelOpenFailure::UnknownChannelType => Self::UnknownChannelType, + russh::ChannelOpenFailure::ResourceShortage => Self::ResourceShortage, + russh::ChannelOpenFailure::Other { code, reason } => Self::Other { code, reason }, + } + } +} + +/// The client direction's op-row failure classes (`w4-op-rows`) — +/// the exec/pty / tunnel / git-pack op surfaces' closed taxonomy +/// (`RemoteSession` / `TunnelChannel` / `GitPackSession` — the +/// feature-gated members live in the crate root's exports), deliberately distinct from [`DialError`]: +/// an op runs on an *already-dialled* connection, so its failure is +/// never a no-connection outcome — the connection stays usable +/// ([`OpError::OpenDeclined`] / [`OpError::RequestRefused`] and +/// [`OpError::ChannelEnded`] are channel/op-scoped, mirroring the +/// server direction's decline shapes: channel-scoped, auth intact, +/// no state poisoning), and every op row's decline is fail-closed. +/// +/// There is no catch-all variant: everything an op can fail is one +/// of these classes or a decline class of its own. +/// +#[derive(Debug, Error)] +pub enum OpError { + /// The server refused the op's **channel open** — the typed + /// open's decline (POC-1 §6: a rejected open resolves on the + /// awaiting call, never a `wait()` match). The family-mirrored + /// reason rides; the connection stays usable (row-pinned). + #[error("the server declined the op's channel open: {0}")] + OpenDeclined(#[from] OpenDeclineReason), + + /// The server refused the op's **request** on the opened channel + /// — the SSH-REQUEST reply's `Failure` class, observed through + /// the `wait()`-drain (POC-3 §D.3: Success/Failure arrive on the + /// channel queue, not as `Err` from the request call). The + /// channel is closed behind the refusal (the row owned it). + #[error("the server refused the op's channel request")] + RequestRefused, + + /// The channel ended before the op's request reply arrived (the + /// `wait()`-drain saw the channel's end) — the op cannot + /// proceed, nothing is fabricated. + #[error("the channel ended before the op's request reply arrived")] + ChannelEnded, + + /// The op's command exceeds the exec-length budget + /// (`Limits::max_exec_command_len`) — the per-length cap's + /// client-mirror verdict, answered **before any wire op** (bounded + /// resources; the server direction's pre-parse cap mirrored at + /// the client's command-construction gate). + #[error("the op's command exceeds the configured exec-command length cap")] + CommandTooLong, + + /// The git row's repo id is outside the ADR-008 grammar — the + /// git row's validation gate mirrored (the server direction's + /// `repo_id_validates` verdict), answered before any wire op. + #[error("the repository id is outside the repo-id grammar")] + InvalidRepoId, + + /// Unix sockets are unavailable on this platform — the + /// `direct-streamlocal` twin's runtime-Unix gate (OQ-11: + /// universal compile, clean off-Unix decline **before any wire + /// op**), mirrored from the server direction's gate. + #[error("unix socket forwarding is unavailable on this platform")] + UnixUnavailable, + + /// The sftp route is the OQ-13 decline placeholder (the feature + /// exists, the route does not until alkfs lands, on either + /// direction) — answered before any wire op, never a + /// half-opened subsystem channel. + #[error("the sftp route is not served (the OQ-13 decline placeholder)")] + SftpNotServed, + + /// A russh session transport error underlying the op (the + /// transparent class, as [`DialError::Wire`]). + #[error("ssh op wire error: {0}")] + Wire(russh::Error), +} + +impl From for OpError { + /// The one mirror point: a typed open's rejection is the wire + /// shape `russh::Error::ChannelOpenFailure` and maps to + /// [`OpError::OpenDeclined`] (family-mirrored + /// [`OpenDeclineReason`]); every other russh error stays the + /// transparent [`OpError::Wire`]. + fn from(error: russh::Error) -> Self { + match error { + russh::Error::ChannelOpenFailure(failure) => Self::OpenDeclined(failure.into()), + other => Self::Wire(other), + } + } +} + #[cfg(test)] mod tests { - use super::{AdapterError, DialError, KeySourceError}; + use super::{AdapterError, DialError, KeySourceError, OpError, OpenDeclineReason}; fn all_variants() -> [KeySourceError; 5] { [ @@ -408,4 +542,87 @@ mod tests { )); assert_eq!(mapped.to_string(), "ssh session error: Unknown server key"); } + + fn all_open_decline_reasons() -> Vec<(russh::ChannelOpenFailure, OpenDeclineReason)> { + use russh::ChannelOpenFailure as F; + vec![ + (F::AdministrativelyProhibited, OpenDeclineReason::Prohibited), + (F::ConnectFailed, OpenDeclineReason::ConnectFailed), + (F::UnknownChannelType, OpenDeclineReason::UnknownChannelType), + (F::ResourceShortage, OpenDeclineReason::ResourceShortage), + ( + F::Other { + code: 919, + reason: "because".into(), + }, + OpenDeclineReason::Other { + code: 919, + reason: "because".into(), + }, + ), + ] + } + + /// - *Contract stamp:* `client.md` §op rows (no vanilla-concept + /// leaks up) + POC-1 §6 (the typed open's decline resolves on + /// the awaiting call) — the wire class mirrors one-to-one into + /// the family vocabulary, `Other` verbatim, nothing collapsed. + #[test] + fn open_decline_mirror_preserves_every_class() { + for (wire, expected) in all_open_decline_reasons() { + let mapped: OpenDeclineReason = wire.into(); + assert_eq!(mapped, expected); + } + } + + fn all_op_variants() -> Vec { + let mut all: Vec = all_open_decline_reasons() + .into_iter() + .map(|(_, reason)| OpError::OpenDeclined(reason)) + .collect(); + all.extend([ + OpError::RequestRefused, + OpError::ChannelEnded, + OpError::CommandTooLong, + OpError::InvalidRepoId, + OpError::UnixUnavailable, + OpError::SftpNotServed, + OpError::Wire(russh::Error::Disconnect), + ]); + all + } + + #[test] + fn op_error_closed_taxonomy_no_catch_all() { + let exhaustive: fn(&OpError) -> &'static str = |e| match e { + OpError::OpenDeclined(_) => "open-declined", + OpError::RequestRefused => "request-refused", + OpError::ChannelEnded => "channel-ended", + OpError::CommandTooLong => "command-too-long", + OpError::InvalidRepoId => "invalid-repo-id", + OpError::UnixUnavailable => "unix-unavailable", + OpError::SftpNotServed => "sftp-not-served", + OpError::Wire(_) => "wire", + }; + for v in all_op_variants() { + assert!(!exhaustive(&v).is_empty()); + } + } + + /// - *Contract stamp:* POC-1 §6 — a rejected channel open + /// surfaces on the awaiting typed call as + /// `russh::Error::ChannelOpenFailure(..)`; the op rows map it + /// into [`OpError::OpenDeclined`] and every other wire error + /// stays transparent. + #[test] + fn op_error_open_failure_maps_to_the_class() { + let mapped: OpError = + russh::Error::ChannelOpenFailure(russh::ChannelOpenFailure::ConnectFailed).into(); + assert!(matches!( + mapped, + OpError::OpenDeclined(OpenDeclineReason::ConnectFailed) + )); + let mapped: OpError = russh::Error::Disconnect.into(); + assert!(matches!(mapped, OpError::Wire(russh::Error::Disconnect))); + } } diff --git a/src/exit_code.rs b/src/exit_code.rs new file mode 100644 index 0000000..6db084f --- /dev/null +++ b/src/exit_code.rs @@ -0,0 +1,325 @@ +//! The exit-currency module — mapping.md §Exit chunk's derivation +//! contract (rules 1–6, normative) as code, *shared by both +//! directions* (`w4-op-rows` extracted it out of `tty_map.rs`, whose +//! server-side session rows were its first and only consumer). +//! +//! The alktty exit chunk (`{"type":"exit","code":i32}`, +//! exit-chunk-is-last per alktty's ADR-004 on the alknet ADR-055 +//! lineage) is the family's sole exit currency; the wire speaks SSH +//! exit reports. The two arms of the same contract: +//! +//! - **The server direction** ([`derive_exit`]): the family chunk → +//! exactly one SSH exit report shape, **never widened**. +//! - **The client direction** (`w4-op-rows`): the SSH exit report → +//! the family `i32` code ([`signal_code_from_name`] and the +//! direct `u32 → i32` identity) — the exact inverse arm, riding the +//! same table. The mirror consumes the convention as-is, including +//! its one collision (the `−1` sentinel vs a genuine SIGHUP — +//! mapping.md rule 3's symmetric-fidelity cost). +//! +//! No widening in either arm: the chunk is not enriched, the wire +//! report is not enriched, and the *core_dumped* bool is never +//! asserted (alktty discards it — nothing better is knowable). +//! +//! ## The server arm's rules (mapping.md, normative) +//! +//! 1. `code ≥ 0` → `exit-status(N)`. +//! 2. negative `−N`, N ≥ 2 → `exit-signal(name)` via the crate-local +//! **Unix signum→name table** ([`SIGNAL_NAMES`]). Server-side +//! work — the signal semantics belong to the serving platform +//! where the backend reaps. Deliberately *not* derived from +//! alktty's `signal_from_name` inverse, which is a forward +//! name→number map for the control row, not a derivation table. +//! `core_dumped` is **never asserted** — SSH's `exit-signal` bool +//! stays false on every row. +//! 3. `−1` → `exit-status(255)`, never a signal claim — the +//! could-not-determine sentinel collides with the SIGHUP encoding +//! on the pipe path, accepted symmetrically per the mapping: +//! neither a fabricated signal for a wait failure nor a real one. +//! 4. Non-Unix adds no other negatives: the table is the only +//! negative recognizer. +//! 5. **The PTY-path fidelity limit (upstream)**: portable-pty 0.9 +//! reaps a signal-terminated Unix child as a plain code (the +//! signal number is unrecoverable through its trait surface), so +//! signal-terminated PTY sessions arrive as `exit-status(1)` — a +//! documented alktty/portable-pty ask; this table in practice +//! sees the *pipe* path's negative encoding only. Faithful +//! `exit-signal` there needs the alktty backend to conform to its +//! own ADR's negative-code convention; alkssh's contract adapts +//! automatically. +//! 6. Unmapped negatives → rule 3 (`exit-status(255)`) — the mapping +//! never invents a name (defensive case: the convention's +//! producers are `−1` and `−signum` only). +//! +//! ## The client arm's rules (`w4-op-rows`, the mirror) +//! +//! 1. `exit-status(N)` → the family code `N` (direct identity). +//! 2. `exit-signal(name)` → the family code `−N` via the same +//! [`SIGNAL_NAMES`] table read backwards +//! ([`signal_code_from_name`]) — the mapping never invents a +//! signum. +//! 3. A name with no table entry → `−1` (the sentinel; the same +//! honest could-not-determine semantic rule 3 consumes on the +//! server arm — including a genuine SIGHUP, whose mirror rides +//! `−1` by the same accepted collision). +//! 4. No exit report at all ends a session with none — the +//! no-exit-chunk close family's mirror ("reports nothing", +//! mapping.md): the caller observes the session's end, never a +//! fabricated code. + +/// One SSH exit report shape derived from an exit chunk — +/// [`derive_exit`]'s outcome. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[cfg(feature = "session")] +pub(crate) enum ExitReport { + /// `exit-status(N)` — codes ≥ 0, the `−1` sentinel, and every + /// unmapped negative (rules 1, 3, 4, 6). + Status(u32), + /// `exit-signal(name)` — a mapped negative `−N`, N ≥ 2 (rule 2). + /// The name is the table's value, never an invention. + Signal(&'static str), +} + +/// The crate-local Unix signum→name table (server arm rule 2 / client +/// arm rule 2): the 1–31 conventional Unix signals, as the exit +/// chunk's producers encode them (`−signum`) and as the wire's +/// `exit-signal` senders name them. Server-side work — the signal +/// semantics belong to the serving platform where the backend reaps. +/// Deliberately **not** derived from alktty's `signal_from_name` (a +/// forward name→number map for the control row, not a derivation +/// table; deliberately narrower too — it exists for what a front-end +/// sends, not for what a child can die by). +pub(crate) const SIGNAL_NAMES: &[(u32, &str)] = &[ + (1, "HUP"), + (2, "INT"), + (3, "QUIT"), + (4, "ILL"), + (5, "TRAP"), + (6, "ABRT"), + (7, "BUS"), + (8, "FPE"), + (9, "KILL"), + (10, "USR1"), + (11, "SEGV"), + (12, "USR2"), + (13, "PIPE"), + (14, "ALRM"), + (15, "TERM"), + (16, "STKFLT"), + (17, "CHLD"), + (18, "CONT"), + (19, "STOP"), + (20, "TSTP"), + (21, "TTIN"), + (22, "TTOU"), + (23, "URG"), + (24, "XCPU"), + (25, "XFSZ"), + (26, "VTALRM"), + (27, "PROF"), + (28, "WINCH"), + (29, "IO"), + (30, "PWR"), + (31, "SYS"), +]; + +/// The derivation contract's whole decision procedure (the §module +/// docs' rules 1–6 — normative). +#[cfg(feature = "session")] +pub(crate) fn derive_exit(code: i32) -> ExitReport { + match code { + code if code >= 0 => ExitReport::Status(code as u32), + -1 => ExitReport::Status(255), + other => match other.checked_abs() { + Some(signum) => match SIGNAL_NAMES.iter().find(|(n, _)| *n == signum as u32) { + Some((_, name)) => ExitReport::Signal(name), + None => ExitReport::Status(255), + }, + None => ExitReport::Status(255), + }, + } +} + +/// The client arm's inverse read (§module docs, rules 1–3): the SSH +/// `exit-signal` sender's name → the family exit-code convention's +/// value (`−signum`), `−1` for a name with no table entry (the +/// sentinel — the mapping never invents a signum; a genuine SIGHUP +/// rides it by the same accepted collision). +pub(crate) fn signal_code_from_name(name: &str) -> i32 { + match SIGNAL_NAMES.iter().find(|(_, n)| *n == name) { + Some((signum, _)) => -(*signum as i32), + None => -1, + } +} + +/// The table's name → russh's `Sig` vocabulary (the +/// `exit_signal_request` parameter). russh's known set rides its +/// variant names; anything else rides `Sig::Custom` (RFC 4254 §6.10's +/// signal-name encoding — vanilla clients parse arbitrary names). +#[cfg(feature = "session")] +pub(crate) fn sig_from_name(name: &str) -> russh::Sig { + match name { + "ABRT" => russh::Sig::ABRT, + "ALRM" => russh::Sig::ALRM, + "FPE" => russh::Sig::FPE, + "HUP" => russh::Sig::HUP, + "ILL" => russh::Sig::ILL, + "INT" => russh::Sig::INT, + "KILL" => russh::Sig::KILL, + "PIPE" => russh::Sig::PIPE, + "QUIT" => russh::Sig::QUIT, + "SEGV" => russh::Sig::SEGV, + "TERM" => russh::Sig::TERM, + "USR1" => russh::Sig::USR1, + other => russh::Sig::Custom(other.to_owned()), + } +} + +/// The SSH signal row's name translation: russh's `Sig` → the name +/// string alktty's control vocabulary carries (the uppercase names; +/// `Custom` carrying its own). Names flow to the backend verbatim — +/// no alkssh-side filter (mapping.md §the control rows; the client +/// arm's `exit-signal` consumption reads through this before +/// [`signal_code_from_name`]). +#[cfg(any(feature = "session", feature = "git"))] +pub(crate) fn sig_control_name(sig: &russh::Sig) -> String { + match sig { + russh::Sig::ABRT => "ABRT".to_owned(), + russh::Sig::ALRM => "ALRM".to_owned(), + russh::Sig::FPE => "FPE".to_owned(), + russh::Sig::HUP => "HUP".to_owned(), + russh::Sig::ILL => "ILL".to_owned(), + russh::Sig::INT => "INT".to_owned(), + russh::Sig::KILL => "KILL".to_owned(), + russh::Sig::PIPE => "PIPE".to_owned(), + russh::Sig::QUIT => "QUIT".to_owned(), + russh::Sig::SEGV => "SEGV".to_owned(), + russh::Sig::TERM => "TERM".to_owned(), + russh::Sig::USR1 => "USR1".to_owned(), + russh::Sig::Custom(name) => name.clone(), + } +} + +#[cfg(test)] +#[cfg(feature = "session")] +mod tests { + use super::{ExitReport, SIGNAL_NAMES, derive_exit, sig_control_name, sig_from_name}; + use russh::Sig; + + /// - *Contract stamp:* mapping.md §Exit chunk rule 1 (`code ≥ 0` + /// → `exit-status(N)` — the chunk carries the whole truth for + /// clean exits). + #[test] + fn non_negative_codes_map_to_exit_status() { + assert_eq!(derive_exit(0), ExitReport::Status(0)); + assert_eq!(derive_exit(1), ExitReport::Status(1)); + assert_eq!(derive_exit(42), ExitReport::Status(42)); + assert_eq!(derive_exit(255), ExitReport::Status(255)); + } + + /// - *Contract stamp:* mapping.md §Exit chunk rule 2 (negative + /// `−N`, N ≥ 2 → `exit-signal(name)` via the table — the POC-3 + /// §D.2 live evidence: `−15` → `exit-signal("TERM")`). + #[test] + fn negative_signals_map_through_the_table() { + for (signum, name) in SIGNAL_NAMES.iter().skip(1) { + assert_eq!(derive_exit(-(*signum as i32)), ExitReport::Signal(name)); + } + assert_eq!(derive_exit(-15), ExitReport::Signal("TERM")); + assert_eq!(derive_exit(-2), ExitReport::Signal("INT")); + } + + /// - *Contract stamp:* mapping.md §Exit chunk rule 3 (`−1` → + /// `exit-status(255)`, never a signal claim — the sentinel/ + /// SIGHUP collision accepted symmetrically). + #[test] + fn the_sentinel_is_not_a_signal() { + assert_eq!(derive_exit(-1), ExitReport::Status(255)); + } + + /// - *Contract stamp:* mapping.md §Exit chunk rule 6 (unmapped + /// negatives → rule 3 — the mapping never invents a name). + #[test] + fn unmapped_negatives_report_255() { + for code in [i32::MIN, -33, -32, -100, -1_000_000] { + assert_eq!(derive_exit(code), ExitReport::Status(255)); + } + } + + /// - *Contract stamp:* the table bound — every entry is the + /// conventional Unix signal set 1–31, sorted, names unique and + /// uppercase (the vocabulary the derivation table promises). + #[test] + fn the_table_is_the_conventional_unix_set() { + for (n, name) in SIGNAL_NAMES.iter() { + assert!((1..=31).contains(n), "signum {n} outside 1..=31"); + assert!( + name.chars() + .all(|c| c.is_ascii_uppercase() || c.is_ascii_digit()), + "signal names are uppercase-and-digit vocabulary: {name}" + ); + } + let mut nums: Vec = SIGNAL_NAMES.iter().map(|(n, _)| *n).collect(); + nums.sort_unstable(); + nums.dedup(); + assert_eq!(nums.len(), SIGNAL_NAMES.len(), "no duplicate signum"); + assert_eq!(nums.first(), Some(&1)); + assert_eq!(nums.last(), Some(&31)); + } + + /// The name→`Sig` translation: russh's known set rides variant + /// names; everything else rides `Sig::Custom` (the RFC's + /// signal-name encoding). + #[test] + fn sig_from_name_covers_known_and_custom() { + assert!(matches!(sig_from_name("TERM"), Sig::TERM)); + assert!(matches!(sig_from_name("USR1"), Sig::USR1)); + assert!(matches!(sig_from_name("USR2"), Sig::Custom(_))); + assert!(matches!(sig_from_name("TSTP"), Sig::Custom(_))); + assert_eq!(sig_control_name(&Sig::TERM), "TERM"); + assert_eq!(sig_control_name(&Sig::USR1), "USR1"); + assert_eq!(sig_control_name(&Sig::Custom("ABRT".to_owned())), "ABRT"); + } +} + +/// The client arm's mirror row — runnable in every build the arm +/// compiles in (session or git; the git-only builds carry the client +/// rows too). +#[cfg(test)] +#[cfg(any(feature = "session", feature = "git"))] +mod client_arm_rows { + use super::SIGNAL_NAMES; + + /// - *Contract stamp:* the client arm (§module docs' rules 1–3) — + /// the inverse read through the same table, never widened: + /// mapped names ride `−signum`, unmapped names ride the `−1` + /// sentinel (the mapping never invents a signum). + #[test] + fn signal_code_from_name_mirrors_through_the_same_table() { + for (signum, name) in SIGNAL_NAMES.iter() { + assert_eq!(super::signal_code_from_name(name), -(*signum as i32)); + } + assert_eq!(super::signal_code_from_name("WHATEVER"), -1); + } + + /// - *Contract stamp:* the two arms round-trip through the one + /// table (client arm → server arm); the `−1` sentinel collision + /// rides rule 3 both directions, accepted. + #[cfg(feature = "session")] + #[test] + fn the_arms_round_trip_and_the_collision_rides_rule_3() { + use super::{ExitReport, derive_exit}; + for (_, name) in SIGNAL_NAMES.iter().skip(1) { + assert_eq!( + derive_exit(super::signal_code_from_name(name)), + ExitReport::Signal(name), + "the arms round-trip for {name}" + ); + } + assert_eq!( + derive_exit(super::signal_code_from_name("HUP")), + ExitReport::Status(255), + "the HUP collision rides rule 3 both directions, accepted" + ); + } +} diff --git a/src/handler.rs b/src/handler.rs index 66ddb55..8a180f2 100644 --- a/src/handler.rs +++ b/src/handler.rs @@ -1068,7 +1068,7 @@ impl russh::server::Handler for SessionHandler { { if let Some(surface) = self.bridges.get(&channel) { let message = alktty::ControlMessage::Signal { - name: tty_map::sig_control_name(&signal), + name: crate::exit_code::sig_control_name(&signal), }; let _ = surface.forward_control(&message); } diff --git a/src/lib.rs b/src/lib.rs index 13ce015..2eca99c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -20,18 +20,21 @@ //! //! Architecture: `docs/architecture/` (phase 1 — ADRs 001–006 accepted; //! the foundations, the server stack, the server-direction mappings, and -//! the client direction's dial surface are landed through wave 4's dial -//! task; the client op rows, the validation suite, and the release are -//! later implementation waves). See also `docs/research/`, +//! the client dial + op rows are landed through wave 4; the `-R` leg and +//! lifecycle primitives, the validation suite, and the release are later +//! implementation waves). See also `docs/research/`, //! `docs/plans/implementation.md`, and family ADR-003. mod adapter; mod channel_cap; mod client; mod client_handler; +mod client_ops; mod config; mod dispatch; mod error; +#[cfg(any(feature = "session", feature = "git"))] +mod exit_code; mod fingerprint; #[cfg(feature = "git")] mod git_map; @@ -55,6 +58,22 @@ pub use client::russh_client_config; pub use client::{SshConnection, dial}; #[doc(hidden)] pub use client_handler::ClientHandler; +/// The client op rows' surfaces (`w4-op-rows`): +/// [`RemoteSession`](client_ops::RemoteSession) — the exec/pty rows' +/// family chunk plane (under `session`), +/// [`TunnelChannel`](client_ops::TunnelChannel) — the direct +/// rows' byte duplex (under `tunnel`), +/// [`GitPackSession`](client_ops::GitPackSession) — the upstream git +/// row's pack duplex + consumed exit (under `git`). The op methods +/// live on [`SshConnection`] (the op-row surface). Not a re-export +/// here: the `sftp` row is [`SshConnection::sftp`] itself — the +/// every-build OQ-13 decline placeholder. +#[cfg(feature = "git")] +pub use client_ops::GitPackSession; +#[cfg(feature = "session")] +pub use client_ops::RemoteSession; +#[cfg(feature = "tunnel")] +pub use client_ops::TunnelChannel; #[cfg(feature = "git")] pub use config::GitSeam; #[cfg(feature = "session")] @@ -62,7 +81,7 @@ pub use config::SessionSeam; #[cfg(feature = "tunnel")] pub use config::TunnelListenSeam; pub use config::{DialConfig, Limits, ServeConfig}; -pub use error::{AdapterError, DialError, KeySourceError}; +pub use error::{AdapterError, DialError, KeySourceError, OpError, OpenDeclineReason}; pub use fingerprint::fingerprint; /// Contract-test surface (`#[doc(hidden)]`): the per-connection auth /// bridge handler — constructible so the tests drive its hooks directly diff --git a/src/tty_map.rs b/src/tty_map.rs index f4fba8f..21729b7 100644 --- a/src/tty_map.rs +++ b/src/tty_map.rs @@ -137,44 +137,20 @@ //! vanilla sshd's own ordering); a resize with neither a serving //! bridge nor a stored pty is dropped. //! -//! ## The exit derivation contract (mapping.md §Exit chunk, rules -//! 1–6 — normative) +//! ## The exit derivation contract (mapping.md §Exit chunk, rules 1–6) //! -//! The alktty exit chunk (`{"type":"exit","code":i32}`, -//! exit-chunk-is-last per alktty's ADR-004 on the alknet ADR-055 -//! lineage) is the sole -//! fidelity input, **never widened**: [`derive_exit`] maps it to -//! exactly one SSH exit report shape: +//! Normative text and the whole derivation machinery now live in +//! [`crate::exit_code`] — the shared exit-currency module consumed by +//! *both* directions (`w4-op-rows` extracted it out of this module: +//! the client rows consume wire exit reports back into the family's +//! `i32` exit-code convention, the exact inverse arm of the same +//! contract, and the table/name plumbing must not be forked). This +//! module keeps the server-direction arm: the exit chunk → +//! [`ExitReport`] ([`derive_exit`]) → the wire report, rules 1–6 +//! verbatim (rule 5's PTY-path fidelity limit included). //! -//! 1. `code ≥ 0` → `exit-status(N)` (rule 1). -//! 2. negative `−N`, N ≥ 2 → `exit-signal(name)` via the -//! crate-local **Unix signum→name table** ([`SIGNAL_NAMES`]). -//! Server-side work — the signal semantics belong to the serving -//! platform where the backend reaps. Deliberately *not* derived -//! from alktty's `signal_from_name` inverse, which is a forward -//! name→number map for the control row, not a derivation table. -//! `core_dumped` is **never asserted** — alktty discards the -//! core-dump bit entirely, so nothing better is knowable from the -//! chunk, and SSH's `exit-signal` bool stays false on every row. -//! 3. `−1` → `exit-status(255)`, never a signal claim (rule 3 — the -//! could-not-determine sentinel collides with the SIGHUP -//! encoding on the pipe path, accepted symmetrically per the -//! mapping: neither a fabricated signal for a wait failure nor a -//! real one). -//! 4. Non-Unix adds no other negatives (rule 4): the table above is -//! the only negative recognizer. -//! 5. **The PTY-path fidelity limit (rule 5, upstream)**: -//! portable-pty 0.9 reaps a signal-terminated Unix child as a -//! plain code (the signal number is unrecoverable through its -//! trait surface), so signal-terminated PTY sessions arrive as -//! `exit-status(1)` — a documented alktty/portable-pty ask; this -//! module's table in practice sees the *pipe* path's negative -//! encoding only. Faithful `exit-signal` there needs the alktty -//! backend to conform to its own ADR's negative-code convention; -//! alkssh's contract adapts automatically. -//! 6. Unmapped negatives → rule 3 (`exit-status(255)`) — the -//! mapping never invents a name (defensive case: the -//! convention's producers are `−1` and `−signum` only). +//! The report is sent **before** close, from the spawned serving +//! task (ADR-006 decision 4, via the cloned `Session::handle()`). //! //! The report is sent **before** close, from the spawned serving //! task (ADR-006 decision 4, via the cloned `Session::handle()`). @@ -206,6 +182,7 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::sync::{mpsc, watch}; use crate::config::SessionSeam; +use crate::exit_code::{ExitReport, derive_exit, sig_from_name}; use crate::observe; /// The negotiation frame the bridge speaks locally — every locally @@ -227,120 +204,6 @@ const BRIDGE_DUPLEX_CAPACITY: usize = 65_536; /// `maximum_packet_size`, here by [`MAX_CHUNK_LEN`] on the frame). const STDIN_PUMP_BUF: usize = 16 * 1024; -/// One SSH exit report shape derived from an exit chunk — -/// [`derive_exit`]'s outcome. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) enum ExitReport { - /// `exit-status(N)` — codes ≥ 0, the `−1` sentinel, and every - /// unmapped negative (rules 1, 3, 4, 6). - Status(u32), - /// `exit-signal(name)` — a mapped negative `−N`, N ≥ 2 (rule 2). - /// The name is the table's value, never an invention. - Signal(&'static str), -} - -/// The crate-local Unix signum→name table (rule 2): the 1–31 -/// conventional Unix signals, as the exit chunk's producers encode -/// them (`−signum`). Server-side work — the serving platform owns -/// the signal semantics where the backend reaps. Deliberately **not** -/// derived from alktty's `signal_from_name` (a forward name→number -/// map for the control row, not a derivation table; deliberately -/// narrower too — it exists for what a front-end sends, not for what -/// a child can die by). -pub(crate) const SIGNAL_NAMES: &[(u32, &str)] = &[ - (1, "HUP"), - (2, "INT"), - (3, "QUIT"), - (4, "ILL"), - (5, "TRAP"), - (6, "ABRT"), - (7, "BUS"), - (8, "FPE"), - (9, "KILL"), - (10, "USR1"), - (11, "SEGV"), - (12, "USR2"), - (13, "PIPE"), - (14, "ALRM"), - (15, "TERM"), - (16, "STKFLT"), - (17, "CHLD"), - (18, "CONT"), - (19, "STOP"), - (20, "TSTP"), - (21, "TTIN"), - (22, "TTOU"), - (23, "URG"), - (24, "XCPU"), - (25, "XFSZ"), - (26, "VTALRM"), - (27, "PROF"), - (28, "WINCH"), - (29, "IO"), - (30, "PWR"), - (31, "SYS"), -]; - -/// The derivation contract's whole decision procedure (§module docs, -/// mapping.md rules 1–6 — normative). -pub(crate) fn derive_exit(code: i32) -> ExitReport { - match code { - code if code >= 0 => ExitReport::Status(code as u32), - -1 => ExitReport::Status(255), - other => match other.checked_abs() { - Some(signum) => match SIGNAL_NAMES.iter().find(|(n, _)| *n == signum as u32) { - Some((_, name)) => ExitReport::Signal(name), - None => ExitReport::Status(255), - }, - None => ExitReport::Status(255), - }, - } -} - -/// The table's name → russh's `Sig` vocabulary (the -/// `exit_signal_request` parameter). russh's known set rides its -/// variant names; anything else rides `Sig::Custom` (RFC 4254 §6.10's -/// signal-name encoding — vanilla clients parse arbitrary names). -pub(crate) fn sig_from_name(name: &str) -> russh::Sig { - match name { - "ABRT" => russh::Sig::ABRT, - "ALRM" => russh::Sig::ALRM, - "FPE" => russh::Sig::FPE, - "HUP" => russh::Sig::HUP, - "ILL" => russh::Sig::ILL, - "INT" => russh::Sig::INT, - "KILL" => russh::Sig::KILL, - "PIPE" => russh::Sig::PIPE, - "QUIT" => russh::Sig::QUIT, - "SEGV" => russh::Sig::SEGV, - "TERM" => russh::Sig::TERM, - "USR1" => russh::Sig::USR1, - other => russh::Sig::Custom(other.to_owned()), - } -} - -/// The SSH signal row's name translation: russh's `Sig` → the name -/// string alktty's control vocabulary carries (the uppercase names; -/// `Custom` carrying its own). Names flow to the backend verbatim — -/// no alkssh-side filter (§the control rows). -pub(crate) fn sig_control_name(sig: &russh::Sig) -> String { - match sig { - russh::Sig::ABRT => "ABRT".to_owned(), - russh::Sig::ALRM => "ALRM".to_owned(), - russh::Sig::FPE => "FPE".to_owned(), - russh::Sig::HUP => "HUP".to_owned(), - russh::Sig::ILL => "ILL".to_owned(), - russh::Sig::INT => "INT".to_owned(), - russh::Sig::KILL => "KILL".to_owned(), - russh::Sig::PIPE => "PIPE".to_owned(), - russh::Sig::QUIT => "QUIT".to_owned(), - russh::Sig::SEGV => "SEGV".to_owned(), - russh::Sig::TERM => "TERM".to_owned(), - russh::Sig::USR1 => "USR1".to_owned(), - russh::Sig::Custom(name) => name.clone(), - } -} - /// The session dimension clamp: u32 wire dims → the wire's u16 (a /// dimension beyond 65535 is nonsense on both papers; the cap, not /// a truncation of useful data). @@ -394,7 +257,7 @@ pub(crate) fn negotiate_request( /// One parsed bridge-side frame from alktty's `client_send` half — /// the 4-byte first-window disambiguation's outcome (§the /// translation bridge). -enum BridgeFrame { +pub(crate) enum BridgeFrame { /// A raw alktty chunk (data or server→client control). Chunk(Chunk), /// The length-prefixed negotiation error frame — the session @@ -414,7 +277,9 @@ enum BridgeFrame { /// byte; anything ≤ 4 is a chunk `stream_type`. Bounds-checked /// against [`MAX_CHUNK_LEN`] both directions — a rogue length never /// triggers an oversized read. -async fn read_bridge_frame(reader: &mut (impl tokio::io::AsyncRead + Unpin)) -> BridgeFrame { +pub(crate) async fn read_bridge_frame( + reader: &mut (impl tokio::io::AsyncRead + Unpin), +) -> BridgeFrame { let mut head = [0u8; 4]; match reader.read_exact(&mut head).await { Ok(_) => {} @@ -825,84 +690,6 @@ async fn drive_chunks_to_ssh( #[cfg(test)] mod tests { - use super::{ExitReport, SIGNAL_NAMES, derive_exit, sig_control_name, sig_from_name}; - use russh::Sig; - - /// - *Contract stamp:* mapping.md §Exit chunk rule 1 (`code ≥ 0` - /// → `exit-status(N)` — the chunk carries the whole truth for - /// clean exits). - #[test] - fn non_negative_codes_map_to_exit_status() { - assert_eq!(derive_exit(0), ExitReport::Status(0)); - assert_eq!(derive_exit(1), ExitReport::Status(1)); - assert_eq!(derive_exit(42), ExitReport::Status(42)); - assert_eq!(derive_exit(255), ExitReport::Status(255)); - } - - /// - *Contract stamp:* mapping.md §Exit chunk rule 2 (negative - /// `−N`, N ≥ 2 → `exit-signal(name)` via the table — the POC-3 - /// §D.2 live evidence: `−15` → `exit-signal("TERM")`). - #[test] - fn negative_signals_map_through_the_table() { - for (signum, name) in SIGNAL_NAMES.iter().skip(1) { - assert_eq!(derive_exit(-(*signum as i32)), ExitReport::Signal(name)); - } - assert_eq!(derive_exit(-15), ExitReport::Signal("TERM")); - assert_eq!(derive_exit(-2), ExitReport::Signal("INT")); - } - - /// - *Contract stamp:* mapping.md §Exit chunk rule 3 (`−1` → - /// `exit-status(255)`, never a signal claim — the sentinel/ - /// SIGHUP collision accepted symmetrically). - #[test] - fn the_sentinel_is_not_a_signal() { - assert_eq!(derive_exit(-1), ExitReport::Status(255)); - } - - /// - *Contract stamp:* mapping.md §Exit chunk rule 6 (unmapped - /// negatives → rule 3 — the mapping never invents a name). - #[test] - fn unmapped_negatives_report_255() { - for code in [i32::MIN, -33, -32, -100, -1_000_000] { - assert_eq!(derive_exit(code), ExitReport::Status(255)); - } - } - - /// - *Contract stamp:* the table bound — every entry is the - /// conventional Unix signal set 1–31, sorted, names unique and - /// uppercase (the vocabulary the derivation table promises). - #[test] - fn the_table_is_the_conventional_unix_set() { - for (n, name) in SIGNAL_NAMES.iter() { - assert!((1..=31).contains(n), "signum {n} outside 1..=31"); - assert!( - name.chars() - .all(|c| c.is_ascii_uppercase() || c.is_ascii_digit()), - "signal names are uppercase-and-digit vocabulary: {name}" - ); - } - let mut nums: Vec = SIGNAL_NAMES.iter().map(|(n, _)| *n).collect(); - nums.sort_unstable(); - nums.dedup(); - assert_eq!(nums.len(), SIGNAL_NAMES.len(), "no duplicate signa"); - assert_eq!(nums.first(), Some(&1)); - assert_eq!(nums.last(), Some(&31)); - } - - /// The name→`Sig` translation: russh's known set rides variant - /// names; everything else rides `Sig::Custom` (the RFC's - /// signal-name encoding). - #[test] - fn sig_from_name_covers_known_and_custom() { - assert!(matches!(sig_from_name("TERM"), Sig::TERM)); - assert!(matches!(sig_from_name("USR1"), Sig::USR1)); - assert!(matches!(sig_from_name("USR2"), Sig::Custom(_))); - assert!(matches!(sig_from_name("TSTP"), Sig::Custom(_))); - assert_eq!(sig_control_name(&Sig::TERM), "TERM"); - assert_eq!(sig_control_name(&Sig::USR1), "USR1"); - assert_eq!(sig_control_name(&Sig::Custom("ABRT".to_owned())), "ABRT"); - } - /// - *Contract stamp:* the no-exit-chunk close family's frame /// classification (`w4-tty-coverage` — the wire row in /// `tests/tty_mapping.rs` lands the family's serving outcome): a diff --git a/tasks/implementation/w4-op-rows.md b/tasks/implementation/w4-op-rows.md index e45dbce..fb3cdac 100644 --- a/tasks/implementation/w4-op-rows.md +++ b/tasks/implementation/w4-op-rows.md @@ -1,7 +1,7 @@ --- id: w4-op-rows name: Wave 4 — the client op rows (exec/pty, direct-tcpip, git exec, sftp decline) -status: pending +status: completed depends_on: [w4-dial] scope: moderate risk: medium @@ -63,20 +63,20 @@ mechanics are the body of knowledge this task lands in code: ## Acceptance Criteria -- [ ] Exec/pty rows land under `session` with the POC-3 §D.3 +- [x] Exec/pty rows land under `session` with the POC-3 §D.3 mechanics pinned in code + rows (`exec()` not `request_exec`; `wait()`-drain Success/Failure; `window_change`/`signal` before `into_stream`; queue never consumed before conversion) -- [ ] `direct-tcpip` (+ streamlocal twin, runtime-Unix gate) lands +- [x] `direct-tcpip` (+ streamlocal twin, runtime-Unix gate) lands under `tunnel`: typed open, reject → `Err(ChannelOpenFailure)`, the connection stays usable after a decline (row-pinned) -- [ ] Git exec consumption lands under `git` over the shared grammar +- [x] Git exec consumption lands under `git` over the shared grammar (no forked parser); the pack duplex pumps; exit-status consumed -- [ ] The sftp subsystem placeholder declines in every build (the +- [x] The sftp subsystem placeholder declines in every build (the client-side OQ-13 mirror) -- [ ] Per-build contract rows green through the server double; the +- [x] Per-build contract rows green through the server double; the smoke posture decided + recorded -- [ ] `cargo test` (+ per-feature + `--all-features`), clippy ×2 +- [x] `cargo test` (+ per-feature + `--all-features`), clippy ×2 (`-D warnings`), fmt, `cargo doc --no-deps` pass ## References @@ -98,8 +98,233 @@ mechanics are the body of knowledge this task lands in code: ## Notes -> To be filled by implementation agent +### What landed (2026-10-29) + +- `src/client_ops.rs` (new, every build; rows feature-gated inside): + the op rows — the family op → vanilla translation surfaces, all on + `SshConnection` (`dial`'s handle; sibling-module method impls over + the crate accessors `russh_handle()` / `budgets()`): + - *session*: `exec` (pipe), `pty_exec` (`ssh -t`), `shell(Option< + TerminalParamsWire>)` (pty with params; the pipe shell with + `None`, mirroring the server direction's decided pty-less shell + parity) → `RemoteSession`, the **family chunk plane** (§below). + - *tunnel*: `tunnel_direct(host, port)` / + `tunnel_direct_unix(path)` → `TunnelChannel` (a plain byte + duplex — the alksocks `DialFn` composes it; the `-D` leg's + composition needs no dependency, ADR-003 §1a). + - *git*: `git_service(alkgit::Service, repo_id)` → + `GitPackSession` (raw pack duplex + stderr duplex + the + consumed-exit oneshot). + - *every build*: `sftp()` → `Err(OpError::SftpNotServed)` (the + `Ok` type is `Infallible` — the feature ships but the route + doesn't, OQ-13, now tested in both the featureless and the + featured build). +- `src/exit_code.rs` (new, extracted from `tty_map.rs` — the wave's + structural edit): the derivation contract's machinery + (`SIGNAL_NAMES` / `ExitReport` / `derive_exit` / + `sig_from_name` / `sig_control_name`) is now **two-direction + territory** — the module gained the client arm's + `signal_code_from_name` (the wire `exit-signal` name → the family + `−signum` code, `−1` for the unmapped/the sentinel collision) and + the module docs carry the client arm's rules as the same + contract's inverse. `tty_map.rs` consumes it (no fork anywhere); + `handler.rs`'s signal row follows the moved path. +- `src/error.rs`: [`OpError`] — the op rows' closed taxonomy + (`OpenDecline(OpenDeclineReason)` — the *family-mirrored* copy of + the typed open's wire class, + `Other{code,reason}` verbatim, nothing collapsed / + `RequestRefused` (the `wait()`-drain's `Failure` reply — POC-3 + §D.3's not-an-`Err` fact surfaced as the op's decline class) / + `ChannelEnded` (the drain saw the channel's end; nothing + fabricated) / `CommandTooLong` (the exec-length budget's + client-mirror verdict, pre-wire) / `InvalidRepoId` (the git + validation gate mirrored, pre-wire) / `UnixUnavailable` (the OQ-11 + gate, pre-wire) / `SftpNotServed` (the placeholder) / + `Wire(russh::Error)` (transparent); the `From` maps + the typed open's `ChannelOpenFailure` into the family mirror — + one conversion point, unit rows pin every class). +- `src/dispatch.rs`: the shared grammar's **construction arm** — + `build_git_exec_command(route, repo_id)` beside `parse_exec`, with + unit rows proving the construction round-trips the parse for both + services (the grammar's gate is the `Service` *type*: the + construction arm can emit no other token, so the client rows are + structurally unreachable by `unknown-service`/`service-only` — + recorded). +- `src/client.rs`: the crate accessors (`russh_handle`/`budgets`, + feature-gated; no public-surface change); `src/lib.rs`: the + `client_ops` module + the per-feature surface re-exports + (`RemoteSession` / `TunnelChannel` / `GitPackSession`, + `OpError`/`OpenDeclineReason` public). +- `tests/common/sshd_double.rs`: the fake `sshd` grown with the + op-row script — the session requests' reply knob, the scripted + program (stdout/stderr/exit — `exit-then-close`, the vanilla + discipline; a no-output/no-exit script launches *nothing*, keeping + interactive sessions open), the direct opens' echo-or-reject + behavior, a counted-refusal knob for the + stays-usable-after-a-decline rows, and the observation surface + (`SshdObserved`: recorded execs/ptys/windows/signals — the wire + translations' witnesses); `tests/client_ops.rs`: the contract rows + (below). + +### Decided + recorded + +1. **The family plane (the session rows) is the alktty chunk framing + in both directions** — the exact plane the server direction's + bridge speaks, consumed not re-derived: stdin chunks + `CTRL_IN` + (sentinel/`Resize`/`Signal`/`Eof`) toward the wire; stdout/stderr + chunks + the **in-band exit chunk, last** toward the hub. The + exit ride is the family exit-currency (`ControlMessage::Exit`'s + `i32`): wire `ExitStatus(N)` → `N`, wire `ExitSignal(name)` → + `−signum` via the table's inverse read (unmapped names ride the + `−1` sentinel — including a genuine SIGHUP, the same collision + rule 3 consumes server-side). A session ending with no report + delivers no exit chunk (the no-exit-chunk close family's mirror). +2. **The exec/pty/git rows never `into_stream()`; the tunnel rows + never wait.** The mechanics facts pin the *shapes*: the + conversion's reader swallows exactly the non-`Data` messages the + exit reports ride on, so an exit-consuming row is structurally a + **split-halves wait-pump** (`Channel::split()` + spawned drain of + `wait()` — the same shape `w4-git-exit-close` pinned server-side; + the write half stays alive, so the fire-and-forget + `window_change`/`signal` rows ride mid-session — the ordering + fact's constraint, that controls precede `into_stream()`, is + respected by *construction* across both row kinds). The tunnel + rows go typed-open → straight `into_stream()`, no queue + consumption ever (POC-1 §6's fresh-`wait()` caveat, honored by + construction); `client.md` §Correctness facts records the landed + shape. +3. **The hub→door plane reads without the error-frame + disambiguation** (`ChunkReader::read_chunk`, not + `tty_map::read_bridge_frame`): that plane's chunk types are + stdin `0` / `CTRL_IN` `3`; a negotiation-error frame never exists + toward the door — and a `0x00`-leading read there is a legitimately + valid zero-length stdin chunk, so `read_bridge_frame`'s + `0x00`-disambiguation (whose home is the alktty→hub direction, + where error frames live) would misparse it. The wave-3 harness's + mistake became this module's first live-row bug, caught by the + contract rows before landing; recorded so the disambiguation rule + ("the server never sends 0 or 3", alktty ADR-052 §5) is understood + as *directional*. +4. **The pumps' terminal/kill structure** (the drop-is-wire-clean + path): every pump pair shares a kill watch the *output* pump's + terminal paths fire (the exit chunk / the channel's end / a hub + write failure); the *input* pump closes the wire channel on it. + stdin-EOF (the sentinel, a `CTRL_IN` Eof, or the hub's input half + dropping) sends the wire EOF *once* and parks the pump in the + idle-for-kill state (the session keeps pumping — stdin closure + never ends the session). Dropping the surface (both halves) closes + the channel — wire-clean, nothing dangling outside the dial's + own wall clock. +5. **The launch drains are message-capped** (`DRAIN_MESSAGE_CAP`, + 1024): a peer answering only with non-reply messages cannot loop + the drain unboundedly (bounded resources; the drain's idle waits + are bounded by the dial's always-set inactivity timeout). A + refusal closes the channel behind itself (`drain_launch_request` + — the row owned it; never half-serves). +6. **The pty translation**: family `TerminalParamsWire` → wire + `pty_request` verbatim (term/cols/rows/pixels, `term: None` → the + empty-terminal-string value; **the modes JSON never rides the + wire** — always the empty set, the OQ-44 reserved posture + mirrored: the door never invents wire terminal modes); a refused + pty request refuses the whole row (never a pty-less fall) — the + never-half-capability row. +7. **The git row's gates before any wire op** (the shared grammar's + verdicts): `validate_repo_id` → `InvalidRepoId`; the built + command's length → `CommandTooLong`. The other grammar decline + classes are structurally unreachable client-side (the + `Service`-typed input carries only the two serving services). + stderr rides a split raw duplex (the hub may drop it); the pack + duplex is the protocol's own byte plane, untouched. +8. **The tunnel rows' originator**: the door's label + (`"alk-ssh"`, port 0) — nothing invented, nothing leaked; peer + logs name the door (recorded; a real originator would be the + assembly's local-listener fact, and the `DialFn` shape carries + the target only). +9. **No client-side in-crate channel cap admits ops**: the server + direction's cap is a wire-battery defense; the client rows are + hub-driven (the family app chose the dial). Bounded resources on + this direction: the always-on wall clock/keepalives, the russh + windows, the bounded duplexes/mpscs, the drain cap. +10. **No crate events on the client op rows**: the outcomes ride the + typed surfaces (`Result`/channels) — the decline* event + vocabulary takes a wire channel id, and a *client-side* declined + open has none to name (the typed open carries no id back); + the events remain the server-direction (and dial-skeleton) rows' + surface. Recorded so the asymmetry is a decision, not an + omission. +11. **The smoke posture** (decided + recorded, per the task's item): + the session rows' cheap real-stack smoke is the + alkssh-client-op-row ↔ alkssh-server loopback (the client exec + row against the server direction's *real* session rows — both + chunk-bridge arms over one live wire), landed as the row + `smoke_client_exec_row_against_the_server_directions_session_rows` + (gated on `sh` like the wave-3 smokes, skipped-with-reason when + absent). A **real-vanilla-`sshd`** dial smoke is wave-5 e2e + territory explicitly — nothing silently dropped. + +### Test rows (`tests/client_ops.rs`, one per decline/contract class; per-build binaries) + +- *every build*: the sftp placeholder's decline (before any wire op, + no channel opened at the peer) + the featured-build variant (under + `sftp`) — the connection-stays-usable close on both. +- *session* (10 rows + the smoke): the exec family-plane row + (command rides the wire verbatim; stdout/stderr chunks distinct; + the exit chunk in-band, last; the stdin sentinel = wire EOF), the + refused-exec `RequestRefused` row, the pre-wire budget row (no + channel opened), the pty-shell param-translation row (the family + `TerminalParamsWire` verbatim, the reserved modes never riding), + the pipe-shell parity row (no pty request), the live-pump control + rows (CTRL_IN Resize → wire `window_change`; Signal → wire + `signal`, name verbatim), the exit-signal mirror row (terminal + `exit-signal("TERM")` → the family code `−15` in-band), the + never-half-capability row (a refused pty refuses the whole row — + no exec behind it ever rode the wire), the server-rejected-open + decline row (`OpenDeclined(Prohibited)`) with the + stays-usable-after-the-decline launch on the same connection, and + the smoke. +- *tunnel*: the direct-tcpip pump row (raw byte duplex + round-trips; `poll_shutdown` EOF rides), the + reject-maps-the-class + stays-usable row (`ConnectFailed` mirror; + the streamlocal twin then opens on the same connection on-Unix; + `#[cfg(not(unix))]`'s OQ-11 decline row compiles off-Unix only — + the universal-compile/clean-decline gate mirrored from the server + side), the twin's on-Unix pump served in the same row. +- *git*: the grammar's construction riding the wire verbatim + the + pack duplex pump + the `exit-status(0)` consumed into the family + code, the pre-wire `InvalidRepoId` / budget rows (no channel + opened), and the receive-pack-refused `RequestRefused` row. +- The `PumpBuf`/drain caps and the exit-currency mirror are unit + rows inside `client_ops.rs` / `exit_code.rs` / `error.rs` + (taxonomy) / `dispatch.rs` (the construction arm's round-trip). + +### Verification + +`cargo test` (default), per-feature (`--features {session, git, +tunnel, sftp}`), pairs, and `--all-features` — all 15 test binaries +green in the full builds (the client_ops binary: 18 rows under +`--all-features`, the per-build sets: session 12 / git 5 / tunnel 3 +/ featureless 1); `cargo clippy --all-features --all-targets` and +`cargo clippy --all-targets` both `-D warnings` clean; `cargo fmt +--check` clean; `cargo doc --no-deps` warning-free. ## Summary -> To be filled on completion \ No newline at end of file +The client op rows landed: family ops in (`alktty` session shapes / +alkgit `Service`), vanilla wire ops out, no vanilla concepts riding +up (`OpError`'s family-mirrored taxonomy; `OpenDeclineReason` as the +typed-open decline's one-to-one mirror). The session rows give the +hub the alktty chunk plane in both directions with the exit +consumed in-band as the family exit chunk (the derivation contract's +client arm, extracted into the shared two-direction `exit_code.rs` +module — table and rules never forked); the tunnel rows are the +typed `direct-tcpip`/`direct-streamlocal` opens into raw byte +duplexes (rejects mirror to `OpenDeclineReason`, the connection +stays usable, row-pinned per decline class); the git row builds the +vanilla exec command through the shared grammar's new construction +arm and pumps the raw pack duplex with the exit consumed; the sftp +placeholder declines in every build before any wire op. The +mechanics facts are pinned in code (the `wait()`-drain shapes, the +split-halves pump for exit-consuming rows, the +typed-open-then-straight-conversion for the exit-less ones), the +fake-`sshd` double grew the scripted op-row surfaces + observations, +and 18 contract rows per-build + one smoke landed through it. \ No newline at end of file diff --git a/tests/client_ops.rs b/tests/client_ops.rs new file mode 100644 index 0000000..a3fb2fc --- /dev/null +++ b/tests/client_ops.rs @@ -0,0 +1,838 @@ +//! The op rows' contract rows (`w4-op-rows`) — driven through the +//! in-process fake `sshd` grown with the op-row script +//! ([`sshd_double`](common::sshd_double), the client-direction +//! vehicle): featured + working rows (exec exit-status, pty +//! window-change riding the live pump, the direct-tcpip pump, the git +//! exec round trip), the decline rows (server-rejected opens, the +//! sftp placeholder, the off-Unix streamlocal twin — OQ-11), and the +//! connection-stays-usable row after each decline class. The +//! row/stamp discipline (`tests/common/mod.rs`'s doc) applies +//! verbatim. +//! +//! **Smoke posture** (the task's decided + recorded item): the session +//! rows' cheap real-stack smoke is the alkssh-client-op-row ↔ +//! alkssh-server-direction loopback (both russh stacks; the client +//! exec row driven against the server direction's *real* session rows +//! — the chunk bridge's two arms meeting over one live wire), gated +//! on tooling like the wave-3 smokes — skipped-with-reason when +//! absent, never silently green. A real-vanilla-`sshd` dial is wave-5 +//! e2e territory explicitly (see the task's Notes). + +mod common; + +use std::sync::Arc; +use std::time::Duration; + +use alkssh::{DialConfig, Limits, OpError, SshConnection, dial}; + +#[cfg(any(feature = "session", feature = "tunnel"))] +use alkssh::OpenDeclineReason; +#[cfg(feature = "session")] +use alkssh::fingerprint; + +#[cfg(feature = "tunnel")] +use common::sshd_double::DblDirect; +#[cfg(any(feature = "session", feature = "git"))] +use common::sshd_double::DblExit; +use common::sshd_double::{SshdDouble, SshdObserved, SshdScript, spawn_scripted}; +use common::{MemHostKeyPolicy, TestVault, fixture}; + +/// A vault-backed key source over the alpha fixture — the dial +/// posture (the fixture PEM sealed, decrypt-at-dial). +fn alpha_key_source() -> Arc { + let vault = TestVault::unlocked(); + let blob = vault.seal(fixture("alpha")); + Arc::new(alkssh::VaultKeySource::new(vault.handle(), Some(blob))) +} + +/// Dials a fresh double under the given script — one dial per row per +/// double. Returns the connection, the double (its outcome channel +/// for the clean-close rows) and the double's observations. +async fn dial_scripted(script: SshdScript) -> (SshConnection, SshdDouble, SshdObserved) { + dial_scripted_limits(script, Limits::default()).await +} + +/// [`dial_scripted`] tuned with an explicit budget (the client-side +/// caps are [`Limits`], carried by the dial). +async fn dial_scripted_limits( + script: SshdScript, + limits: Limits, +) -> (SshConnection, SshdDouble, SshdObserved) { + let (double, accepts, observed) = + spawn_scripted(vec![fixture("bravo").private_key()], true, script).await; + accepts.register(fixture("alpha").fingerprint); + + let policy = Arc::new(MemHostKeyPolicy::new()); + policy.pin("127.0.0.1", fixture("bravo").fingerprint); + let mut config = DialConfig::new( + double.address.clone(), + "runner", + alpha_key_source(), + Arc::clone(&policy) as Arc, + ); + config.limits = limits; + let connection = dial(config).await.expect("the dial succeeds"); + (connection, double, observed) +} + +/// The clean-close observable: the explicit close leaves the double's +/// session `Ok` — the decline rows' "the connection stays usable" +/// final act. +async fn expect_clean_close(connection: SshConnection, mut double: SshdDouble) { + assert!(!connection.is_closed()); + connection.close().await; + let outcome = double.next_session_outcome().await; + assert!( + matches!(outcome, Some(Ok(()))), + "the clean close is the double's clean session end, got {outcome:?}" + ); + double.shutdown(); +} + +/// `next_opened_session` bounded for the "no channel ever opened" +/// assertions (an unbounded `recv` would hang a wrong row). +async fn no_open_observed(double: &mut SshdDouble, within: &str) { + let deadline = tokio::time::Instant::now() + Duration::from_millis(400); + let opened = tokio::time::timeout_at(deadline, double.next_opened_session()).await; + assert!( + matches!(opened, Ok(None) | Err(_)), + "no session channel may open before {within}: got {opened:?}" + ); +} + +/// OQ-13's client mirror: the sftp op declines cleanly in **every** +/// build (the `sftp` feature ships in the build string but gates no +/// route — the featured-build variant below pins that half too, +/// `--all-features` running both). +/// +/// - *Contract stamp:* OQ-13 (the decline placeholder is the shipped +/// behavior, both directions) + `client.md` §op rows (row 5) + §What +/// the crate does *not* include — before any wire op: no session +/// channel ever opens at the peer, no subsystem request ever rides, +/// and the connection stays usable (the clean close is the double's +/// clean session end). +#[tokio::test] +async fn sftp_placeholder_declines_in_every_build_before_any_wire_op() { + let (connection, mut double, observed) = dial_scripted(SshdScript::default()).await; + + let err = connection + .sftp() + .await + .expect_err("the OQ-13 placeholder declines the sftp op"); + assert!(matches!(err, OpError::SftpNotServed), "got {err:?}"); + assert!( + observed.execs().is_empty(), + "the placeholder consulted no wire op" + ); + no_open_observed(&mut double, "the sftp placeholder's decline").await; + expect_clean_close(connection, double).await; +} + +#[cfg(feature = "session")] +#[tokio::test] +async fn sftp_feature_on_the_placeholder_still_declines() { + let (connection, mut double, observed) = dial_scripted(SshdScript::default()).await; + let err = connection + .sftp() + .await + .expect_err("the feature never serves the route (OQ-13 — the placeholder ships)"); + assert!(matches!(err, OpError::SftpNotServed), "got {err:?}"); + assert!(observed.execs().is_empty()); + no_open_observed(&mut double, "the featured placeholder's decline").await; + expect_clean_close(connection, double).await; +} + +#[cfg(feature = "session")] +mod session_rows { + use super::*; + use alktty::wire::{ChunkReader, STREAM_CTRL_IN, STREAM_STDIN}; + use alktty::{ControlMessage, negotiation::TerminalParamsWire}; + use tokio::io::AsyncWriteExt; + + /// Frames one chunk by hand (`stream_type` byte + the big-endian + /// `u32` length + payload — the published header format, read + /// back only through alktty's own [`ChunkReader`]). + fn frame(stream_type: u8, payload: &[u8]) -> Vec { + let mut out = vec![stream_type]; + out.extend_from_slice(&(payload.len() as u32).to_be_bytes()); + out.extend_from_slice(payload); + out + } + + /// Frames one control chunk — the family currency's own + /// serializer builds the payload. + async fn control_frame(message: &ControlMessage) -> Vec { + let payload = message.to_json().expect("the control serializes"); + frame(STREAM_CTRL_IN, &payload) + } + + /// Drains the hub's chunk plane until the exit chunk or the end — + /// returns (stdout, stderr, the exit code in-band, if any). + async fn read_to_exit(mut session: alkssh::RemoteSession) -> (Vec, Vec, Option) { + let mut stdout = Vec::new(); + let mut stderr = Vec::new(); + let mut exit = None; + let mut reader = ChunkReader::new(&mut session.output); + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + loop { + let chunk = match tokio::time::timeout_at(deadline, reader.read_chunk()).await { + Ok(Ok(chunk)) => chunk, + _ => break, + }; + match chunk.stream_type { + alktty::wire::STREAM_STDOUT => stdout.extend_from_slice(&chunk.bytes), + alktty::wire::STREAM_STDERR => stderr.extend_from_slice(&chunk.bytes), + alktty::wire::STREAM_CTRL_OUT => { + if let Ok(ControlMessage::Exit { code }) = + ControlMessage::from_slice(&chunk.bytes) + { + exit = Some(code); + } + break; + } + _ => {} + } + } + (stdout, stderr, exit) + } + + /// The exit row: the sentinel rides, the pump delivers. + async fn send_stdin_eof(session: &mut alkssh::RemoteSession) { + session + .input + .write_all(&frame(STREAM_STDIN, &[])) + .await + .expect("the stdin sentinel writes"); + } + + /// - *Contract stamp:* `client.md` §op rows (row 1, the exec + /// translation + the family plane) — the wire carries the + /// family command verbatim (`exec()`, never `request_exec`); + /// stdout/stderr come back as distinct chunks and the exit + /// status rides the family chunk in-band, last. + #[tokio::test] + async fn exec_row_pumps_the_program_and_the_exit_through_the_family_plane() { + let script = SshdScript { + stdout: b"out-ok", + stderr: b"err-why", + exit: Some(DblExit::Status(0)), + ..SshdScript::default() + }; + let (connection, double, observed) = dial_scripted(script).await; + let mut session = connection + .exec("echo out-ok") + .await + .expect("the exec row launches"); + + assert_eq!( + observed.execs(), + vec![b"echo out-ok".to_vec()], + "the wire carries the family command, never altered" + ); + send_stdin_eof(&mut session).await; + let (stdout, stderr, exit) = read_to_exit(session).await; + assert_eq!(stdout, b"out-ok"); + assert_eq!(stderr, b"err-why"); + assert_eq!( + exit, + Some(0), + "exit-status(0) rode the wire; the family chunk mirrors it in-band" + ); + expect_clean_close(connection, double).await; + } + + /// - *Contract stamp:* POC-3 §D.3 (the Success/Failure semantics — + /// **the Failure reply surfaces on the channel queue's drain, + /// not as `Err` from the request call**) → + /// [`OpError::RequestRefused`]; the refused row closes its + /// channel behind the refusal and the connection stays usable + /// (the clean-close observable). + #[tokio::test] + async fn refused_exec_reply_surfaces_the_request_refusal() { + let script = SshdScript { + request_success: false, + ..SshdScript::default() + }; + let (connection, double, observed) = dial_scripted(script).await; + let err = connection + .exec("echo refused") + .await + .expect_err("the scripted refusal surfaces"); + assert!(matches!(err, OpError::RequestRefused), "got {err:?}"); + assert_eq!( + observed.execs(), + vec![b"echo refused".to_vec()], + "the wire carried the request (its Failure reply is the class)" + ); + expect_clean_close(connection, double).await; + } + + /// - *Contract stamp:* the command-carrying rows' length budget + /// (the server direction's pre-parse cap, `Limits:: + /// max_exec_command_len`, mirrored at the command-construction + /// gate): a beyond-budget command declines **before any wire + /// op** — no channel, no request, bounded by construction. + #[tokio::test] + async fn oversized_exec_declines_before_any_wire_op() { + let limits = Limits { + max_exec_command_len: 4, + ..Limits::default() + }; + let script = SshdScript { + stdout: b"never", + exit: Some(DblExit::Status(0)), + ..SshdScript::default() + }; + let (connection, mut double, observed) = dial_scripted_limits(script, limits).await; + let err = connection + .exec("echo too long") + .await + .expect_err("the budget gate declines before the wire"); + assert!(matches!(err, OpError::CommandTooLong), "got {err:?}"); + assert!(observed.execs().is_empty()); + no_open_observed(&mut double, "the budget gate's decline").await; + expect_clean_close(connection, double).await; + } + + /// - *Contract stamp:* `client.md` §op rows (row 1, the pty/shell + /// translation) — the family [`TerminalParamsWire`] rides the + /// wire `pty_request` verbatim (the reserved modes never ride: + /// the double sees the params, and the wire request carries the + /// empty terminal-modes set), and the shell request launches + /// the session. + #[tokio::test] + async fn pty_shell_translates_the_family_params_onto_the_wire() { + let (connection, double, observed) = dial_scripted(SshdScript::default()).await; + let params = TerminalParamsWire { + term: Some("xterm-256color".to_owned()), + cols: 80, + rows: 24, + pixel_width: 10, + pixel_height: 20, + modes: serde_json::json!({}), + }; + let session = connection + .shell(Some(params)) + .await + .expect("the pty shell row launches"); + assert_eq!( + observed.ptys(), + vec![("xterm-256color".to_owned(), 80, 24, 10, 20)], + "the family params ride the wire pty request verbatim" + ); + assert_eq!(observed.shells().len(), 1, "the shell request rides"); + drop(session); + expect_clean_close(connection, double).await; + } + + /// - *Contract stamp:* the pipe-mode shell parity (mirroring the + /// server direction's decided row — a pty-less shell is + /// legitimate vanilla and served, never declined): the + /// `params: None` shell carries **no** pty request. + #[tokio::test] + async fn pipe_shell_launches_without_a_pty_request() { + let (connection, double, observed) = dial_scripted(SshdScript::default()).await; + let session = connection + .shell(None) + .await + .expect("the pipe shell launches"); + assert!( + observed.ptys().is_empty(), + "no pty request rides the pipe-mode shell" + ); + assert_eq!(observed.shells().len(), 1); + drop(session); + expect_clean_close(connection, double).await; + } + + /// - *Contract stamp:* the pty control rows riding the **live + /// pump** (the POC-3 §D.3 window-change/signal mechanics', + /// client arm): the session never `into_stream()`s, so the + /// fire-and-forget rows ride the write half mid-session — the + /// `CTRL_IN` `Resize` translates to the wire `window_change` + /// and `Signal` to the wire `signal` (the double's + /// observations are the witnesses). + #[tokio::test] + async fn pty_control_rows_ride_the_live_session_pump() { + let (connection, double, observed) = dial_scripted(SshdScript::default()).await; + let params = TerminalParamsWire { + term: Some("xterm-256color".to_owned()), + cols: 80, + rows: 24, + pixel_width: 0, + pixel_height: 0, + modes: serde_json::json!({}), + }; + let mut session = connection + .shell(Some(params)) + .await + .expect("the pty shell row launches"); + + session + .input + .write_all( + &control_frame(&ControlMessage::Resize { + cols: 120, + rows: 40, + pixel_width: 0, + pixel_height: 0, + }) + .await, + ) + .await + .expect("the resize writes"); + session + .input + .write_all( + &control_frame(&ControlMessage::Signal { + name: "TERM".to_owned(), + }) + .await, + ) + .await + .expect("the signal writes"); + + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + let witnessed = || { + observed.windows() == vec![(120, 40, 0, 0)] + && observed.signals().iter().any(|s| s.contains("TERM")) + }; + while !witnessed() { + if tokio::time::Instant::now() >= deadline { + panic!("the control translations never landed: {observed:?}"); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert_eq!( + observed.windows(), + vec![(120, 40, 0, 0)], + "the CTRL_IN Resize translated to the wire window_change" + ); + assert!( + observed.signals().iter().any(|s| s.contains("TERM")), + "the CTRL_IN Signal translated to the wire signal, name verbatim" + ); + + drop(session); + expect_clean_close(connection, double).await; + } + + /// - *Contract stamp:* the family exit-currency's signal mirror + /// (mapping.md's derivation contract, the client arm — consumed + /// not re-derived): a wire `exit-signal("TERM")` reports the + /// family code `−15` in-band (rule 2's value, the table read + /// backwards; the derivation never invents). + #[tokio::test] + async fn exit_signal_report_mirrors_the_family_code_in_band() { + let script = SshdScript { + stdout: b"killed", + exit: Some(DblExit::Signal("TERM")), + ..SshdScript::default() + }; + let (connection, double, observed) = dial_scripted(script).await; + let mut session = connection + .exec("start a session") + .await + .expect("the exec row launches"); + assert_eq!(observed.execs(), vec![b"start a session".to_vec()]); + send_stdin_eof(&mut session).await; + let (stdout, _, exit) = read_to_exit(session).await; + assert_eq!(stdout, b"killed"); + assert_eq!( + exit, + Some(-15), + "the −signum mirror, table-read, no invention" + ); + expect_clean_close(connection, double).await; + } + + /// - *Contract stamp:* the never-half-capability posture (ADR-003 + /// decision 3, client mirror) — a **refused pty refuses the + /// whole row**: the exec behind it never rides the wire (the + /// double's exec observations stay empty). + #[tokio::test] + async fn refused_pty_refuses_the_whole_row_never_half_serves() { + let script = SshdScript { + request_success: false, + ..SshdScript::default() + }; + let (connection, double, observed) = dial_scripted(script).await; + let params = TerminalParamsWire { + term: Some("xterm".to_owned()), + cols: 80, + rows: 24, + pixel_width: 0, + pixel_height: 0, + modes: serde_json::json!({}), + }; + let err = connection + .pty_exec(params, "echo never") + .await + .expect_err("the refused pty refuses the row"); + assert!(matches!(err, OpError::RequestRefused), "got {err:?}"); + assert!( + observed.execs().is_empty() && observed.shells().is_empty(), + "the requests behind the refused pty never rode the wire (the row's channel closed behind the refusal)" + ); + expect_clean_close(connection, double).await; + } + + /// - *Contract stamp:* the server-rejected **open** row (POC-1 §6, + /// client arm) + the connection-stays-usable row (per decline + /// class, `client.md` §Correctness facts): the session open + /// refused (the scripted count) → `OpenDeclined(Prohibited)`, + /// then a later launch on the *same connection* succeeds and + /// pumps — the usable-after-decline posture — and everything + /// closes clean. + #[tokio::test] + async fn server_rejected_open_declines_and_the_connection_stays_usable() { + let script = SshdScript { + reject_first_session_opens: 1, + stdout: b"still-usable", + exit: Some(DblExit::Status(0)), + ..SshdScript::default() + }; + let (connection, double, observed) = dial_scripted(script).await; + + let err = connection + .exec("echo declined") + .await + .expect_err("the server-rejected open declines"); + assert!( + matches!(err, OpError::OpenDeclined(OpenDeclineReason::Prohibited)), + "got {err:?}" + ); + assert!( + !connection.is_closed(), + "the decline never poisons the connection" + ); + + let mut session = connection + .exec("echo still-usable") + .await + .expect("the same connection serves the next launch"); + send_stdin_eof(&mut session).await; + let (stdout, _, exit) = read_to_exit(session).await; + assert_eq!(stdout, b"still-usable"); + assert_eq!(exit, Some(0)); + assert_eq!( + observed.execs(), + vec![b"echo still-usable".to_vec()], + "the declined open carried no request; the usable launch rode the wire" + ); + expect_clean_close(connection, double).await; + } + + /// - *Contract stamp:* the task's smoke item (the §module docs' + /// decided posture; like the wave-3 smokes, skipped-with-reason + /// when the tooling is absent): the client exec row driven + /// against the alkssh *server direction*'s real session rows — + /// both russh stacks over one live wire, the chunk bridge's two + /// arms meeting (the client's family plane consuming the + /// server's family plane), data and the exit-status + /// round-tripping end to end. + #[tokio::test] + async fn smoke_client_exec_row_against_the_server_directions_session_rows() { + if std::process::Command::new("sh") + .arg("-c") + .arg("exit 0") + .output() + .is_err() + { + eprintln!("SKIPPED (with reason): the op-row smoke — sh is unavailable"); + return; + } + + let provider = Arc::new(common::MemIdentityProvider::new()); + provider.register( + &fingerprint(&fixture("alpha").public_key()), + common::identity("op-row-smoke", &["ssh"]), + ); + let mut serve = + common::harness::serve_config(provider, &[fixture("bravo")], Limits::default()); + serve.exec_wrapper_argv = Some(vec!["/bin/sh".to_owned(), "-c".to_owned()]); + { + let mut backends = std::collections::HashMap::new(); + backends.insert( + "local".to_owned(), + Arc::new(alktty::LocalTtyBackend::new()) as Arc, + ); + serve.session_seam = Some(alkssh::SessionSeam { + backends: Arc::new(backends), + backend: "local".to_owned(), + ownership: None, + }); + } + 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 addr = listener.local_addr().expect("the smoke has an address"); + 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 policy = Arc::new(MemHostKeyPolicy::new()); + policy.pin("127.0.0.1", fixture("bravo").fingerprint); + let config = DialConfig::new( + addr.to_string(), + "runner", + alpha_key_source(), + Arc::clone(&policy) as Arc, + ); + let connection = dial(config).await.expect("the smoke dial succeeds"); + let session = connection + .exec("echo alkssh-op-row-smoke") + .await + .expect("the client exec row drives the alkssh server direction"); + + let (stdout, stderr, exit) = read_to_exit(session).await; + assert!( + String::from_utf8_lossy(&stdout).contains("alkssh-op-row-smoke"), + "the program's stdout rode both chunk bridges: {stdout:?}" + ); + assert_eq!(stderr, Vec::::new(), "pipe mode keeps stderr separate"); + assert_eq!( + exit, + Some(0), + "the exit chunk rode the family plane, both arms" + ); + + accept_task.abort(); + connection.close().await; + } +} + +#[cfg(feature = "tunnel")] +mod tunnel_rows { + use super::*; + use alkssh::TunnelChannel; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + /// - *Contract stamp:* `client.md` §op rows (row 2, the typed + /// `channel_open_direct_tcpip` open — the alksocks `DialFn` + /// composes over this surface) + POC-3 §C.1's recipe reversed — + /// the pump is the raw byte duplex, round-tripping, no channel + /// queue consumed anywhere between the open and the + /// `into_stream()` conversion. + #[tokio::test] + async fn direct_tcpip_pump_round_trips_the_raw_duplex() { + let script = SshdScript { + direct: DblDirect::Echo, + ..SshdScript::default() + }; + let (connection, double, _observed) = dial_scripted(script).await; + let mut channel: TunnelChannel = connection + .tunnel_direct("hub.example", 4096) + .await + .expect("the typed open succeeds"); + + let payload = b"tunnel-payload-round-trip"; + channel.write_all(payload).await.expect("the pump writes"); + channel.flush().await.expect("the pump flushes"); + let mut echo = vec![0u8; payload.len()]; + tokio::time::timeout(Duration::from_secs(10), channel.read_exact(&mut echo)) + .await + .expect("the echo arrives") + .expect("the echo reads"); + assert_eq!(echo, payload, "the byte plane round-trips"); + + channel.shutdown().await.expect("the EOF rides"); + expect_clean_close(connection, double).await; + } + + /// - *Contract stamp:* the server-rejected open's family mirror + /// (POC-1 §6 — the typed open resolves `Err` on the reject, + /// family-mapped) + the connection-stays-usable row **per + /// decline class**: after the `ConnectFailed` class the + /// streamlocal twin still opens and pumps through the same + /// connection, and the whole thing closes clean. + #[cfg(unix)] + #[tokio::test] + async fn server_rejected_direct_open_leaves_the_connection_usable() { + let script = SshdScript { + direct: DblDirect::Reject, + streamlocal: DblDirect::Echo, + ..SshdScript::default() + }; + let (connection, double, _observed) = dial_scripted(script).await; + + let err = connection + .tunnel_direct("hub.example", 4096) + .await + .expect_err("the scripted server refuses the direct open"); + assert!( + matches!(err, OpError::OpenDeclined(OpenDeclineReason::ConnectFailed)), + "the wire class mirrors, got {err:?}" + ); + assert!( + !connection.is_closed(), + "the decline never poisons the connection" + ); + + let mut channel = connection + .tunnel_direct_unix("/run/alkssh-op-row.sock") + .await + .expect("the usable-after-decline row: the twin opens on the same connection"); + let payload = b"after-the-decline"; + channel.write_all(payload).await.expect("the pump writes"); + channel.flush().await.expect("the pump flushes"); + let mut echo = vec![0u8; payload.len()]; + tokio::time::timeout(Duration::from_secs(10), channel.read_exact(&mut echo)) + .await + .expect("the echo arrives") + .expect("the echo reads"); + assert_eq!(echo, payload); + + expect_clean_close(connection, double).await; + } + + /// - *Contract stamp:* OQ-11 (the streamlocal twin's runtime-Unix + /// gate, client arm — universal compile, clean off-Unix decline + /// **before any wire op**): this row compiles only off-Unix; + /// the on-Unix builds ride the usable-after-decline row's + /// success half. + #[cfg(not(unix))] + #[tokio::test] + async fn streamlocal_twin_declines_off_unix_before_any_wire_op() { + let (connection, mut double, _observed) = dial_scripted(SshdScript::default()).await; + let err = connection + .tunnel_direct_unix("/run/alkssh-off-unix.sock") + .await + .expect_err("the OQ-11 gate declines off-Unix"); + assert!(matches!(err, OpError::UnixUnavailable), "got {err:?}"); + no_open_observed(&mut double, "the off-Unix twin's decline").await; + expect_clean_close(connection, double).await; + } +} + +#[cfg(feature = "git")] +mod git_rows { + use super::*; + use alkgit::Service; + use alkssh::GitPackSession; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + /// The scripted "advertisement" bytes the double's program + /// writes (a placeholder pack-side payload for the pump row). + const ADVERT: &[u8] = b"0008version 2\n0000"; + + /// - *Contract stamp:* `client.md` §op rows (row 4 — the shared + /// exec grammar's command, *reused, not forked*; the wire + /// carries `git-upload-pack '/owner/demo'` byte-verbatim) + the + /// pack duplex pump + the exit-status consumption. + #[tokio::test] + async fn git_row_round_trips_the_pack_duplex_and_consumes_the_exit() { + let script = SshdScript { + stdout: ADVERT, + exit: Some(DblExit::Status(0)), + ..SshdScript::default() + }; + let (connection, double, observed) = dial_scripted(script).await; + let mut pack: GitPackSession = connection + .git_service(Service::UploadPack, "owner/demo") + .await + .expect("the git row opens"); + assert_eq!( + observed.execs(), + vec![b"git-upload-pack '/owner/demo'".to_vec()], + "the shared grammar's command rode the wire byte-verbatim" + ); + + pack.output + .write_all(b"pack bytes upstream") + .await + .expect("the pack pump writes"); + pack.input.shutdown().await.expect("the EOF rides"); + let mut advert = Vec::new(); + tokio::time::timeout( + Duration::from_secs(10), + pack.output.read_to_end(&mut advert), + ) + .await + .expect("the advertisement arrives") + .expect("the advertisement reads"); + assert_eq!(advert, ADVERT, "the pack duplex pumped"); + + let exit = tokio::time::timeout(Duration::from_secs(10), pack.exit) + .await + .expect("the exit report arrives") + .expect("the pump consumed it"); + assert_eq!(exit, 0, "exit-status(0) consumed into the family code"); + expect_clean_close(connection, double).await; + } + + /// - *Contract stamp:* the git row's pre-wire gates riding the + /// shared grammar's verdicts (`client.md` §op rows, row 4; the + /// git row's server-direction gate mirrored): an id outside the + /// ADR-008 grammar declines (`OpError::InvalidRepoId`) **before + /// any wire op**. + #[tokio::test] + async fn invalid_repo_id_declines_before_any_wire_op() { + let (connection, mut double, observed) = dial_scripted(SshdScript::default()).await; + let err = connection + .git_service(Service::UploadPack, "a/b/c") + .await + .expect_err("the validation gate declines"); + assert!(matches!(err, OpError::InvalidRepoId), "got {err:?}"); + assert!(observed.execs().is_empty()); + no_open_observed(&mut double, "the invalid-repo-id decline").await; + expect_clean_close(connection, double).await; + } + + /// - *Contract stamp:* the per-length cap's client mirror (the + /// shared grammar's verdict — `Limits:: + /// max_exec_command_len`): before any wire op. + #[tokio::test] + async fn oversized_git_command_declines_before_any_wire_op() { + let limits = Limits { + max_exec_command_len: 4, + ..Limits::default() + }; + let (connection, mut double, observed) = + dial_scripted_limits(SshdScript::default(), limits).await; + let err = connection + .git_service(Service::ReceivePack, "owner/demo") + .await + .expect_err("the length gate declines"); + assert!(matches!(err, OpError::CommandTooLong), "got {err:?}"); + assert!(observed.execs().is_empty()); + no_open_observed(&mut double, "the length gate's decline").await; + expect_clean_close(connection, double).await; + } + + /// - *Contract stamp:* the upstream's own refusal is the + /// request-refusal class (the drain's `Failure` — POC-3 §D.3's + /// git-row shape): declined cleanly, the connection intact. + #[tokio::test] + async fn refused_git_exec_surfaces_the_request_refusal() { + let script = SshdScript { + request_success: false, + ..SshdScript::default() + }; + let (connection, double, observed) = dial_scripted(script).await; + let err = connection + .git_service(Service::ReceivePack, "owner/demo") + .await + .expect_err("the scripted upstream refuses"); + assert!(matches!(err, OpError::RequestRefused), "got {err:?}"); + assert_eq!( + observed.execs(), + vec![b"git-receive-pack '/owner/demo'".to_vec()], + "the receive-pack command rode the wire (the refusal is its reply)" + ); + expect_clean_close(connection, double).await; + } +} diff --git a/tests/common/sshd_double.rs b/tests/common/sshd_double.rs index e02da13..1c28efc 100644 --- a/tests/common/sshd_double.rs +++ b/tests/common/sshd_double.rs @@ -36,13 +36,26 @@ //! client `channel_open_*`; the dialed client's skeleton declines; //! the server's open call observes //! `Err(ChannelOpenFailure(..))` — the wire rows' observable). +//! - **The session handle**: the double's `auth_succeeded` hook hands +//! the per-connection [`russh::server::Handle`] to the row through +//! a one-shot carrier ([`SshdDouble::session_handle`]) — the row's +//! initiator for the server-initiated-open decline rows (server → +//! client `channel_open_*`; the dialed client's skeleton declines; +//! the server's open call observes +//! `Err(ChannelOpenFailure(..))` — the wire rows' observable). //! - **The acceptor stub**: rows that need a served session channel //! (the stays-usable-after-declines row here; the op-row vehicles //! later) arm the double to accept client session opens and hand //! the channel out ([`SshdDouble::next_opened_session`]). -//! -//! The row/stamp discipline (`tests/common/mod.rs`'s doc) applies -//! verbatim. +//! - **The op-row script** (`w4-op-rows`): the double's +//! request/open surfaces grow a small script — the session +//! requests' reply (success/failure), the scripted stdout/stderr +//! data + exit report a launched program would produce, and the +//! direct opens' echo-or-reject behavior — plus an observation +//! surface (the recorded commands/pty params/window +//! changes/signals the client rows assert the wire verbatim +//! against). The double stays vanilla-parity: a hand-rolled +//! responder standing in for `sshd`, nothing alkssh-typed inside. #![allow(dead_code)] @@ -52,7 +65,184 @@ use std::time::Duration; use parking_lot::Mutex; use russh::keys::PrivateKey; use russh::server::{Auth, Handle, Msg, PublicKeyOffer, Response, Session, VerifiedPublicKey}; -use russh::{Channel, MethodKind, MethodSet}; +use russh::{Channel, ChannelOpenFailure, MethodKind, MethodSet}; + +/// The double's scripted exit report — what a launched program's +/// death would produce (`exit-status` / `exit-signal`). +#[derive(Debug, Clone, Copy)] +pub enum DblExit { + /// `exit-status(N)`. + Status(u32), + /// `exit-signal(name)` — the wire's signal-name class. + Signal(&'static str), +} + +/// The double's scripted behavior for a client-initiated direct open +/// (`direct-tcpip` / `direct-streamlocal`). +#[derive(Debug, Clone, Copy)] +pub enum DblDirect { + /// Reject the open — the scripted server-refusal class + /// (`ConnectFailed`, the vanilla sshd's dead-target shape). + Reject, + /// Accept the open and echo-pump the channel (the pump rows' + /// round-trip shape). + Echo, +} + +/// The fake `sshd`'s op-row script (the `w4-op-rows` request +/// surfaces) — what the double answers and, on a success, what the +/// scripted "program" produces. +#[derive(Debug, Clone, Copy)] +pub struct SshdScript { + /// The session requests' reply (`exec`/`pty`/`shell`): + /// `true` → `channel_success`, `false` → `channel_failure` (the + /// request-refusal rows' scripted class). + pub request_success: bool, + /// The channel data written as the program's stdout, after the + /// request's success reply. + pub stdout: &'static [u8], + /// The channel data written as `extended_data(1)` stderr. + pub stderr: &'static [u8], + /// The exit report after the outputs (`None` = none — the + /// no-report rows). + pub exit: Option, + /// The `direct-tcpip` opens' behavior. + pub direct: DblDirect, + /// The `direct-streamlocal` opens' behavior. + pub streamlocal: DblDirect, + /// The counted refusal — the double declines the *first N* client + /// session opens (per connection, armed by + /// `accept_session_channels`): the stays-usable-after-a-decline + /// rows' scripted shape (one connection, first open refused, the + /// next served). + pub reject_first_session_opens: usize, +} + +impl Default for SshdScript { + /// The plain dial rows' posture: session opens accepted (the + /// `accept_session_channels` flag governs), requests succeed with + /// no outputs and no exit, direct opens rejected. + fn default() -> Self { + Self { + request_success: true, + stdout: &[], + stderr: &[], + exit: None, + direct: DblDirect::Reject, + streamlocal: DblDirect::Reject, + reject_first_session_opens: 0, + } + } +} + +/// The double's per-connection observations — the wire rows assert +/// the verbatim translations against these (what the client's op +/// rows put on the wire). +#[derive(Clone, Debug, Default)] +pub struct SshdObserved { + execs: Arc>>, + ptys: Arc>>, + shells: Arc>>, + windows: Arc>>, + signals: Arc>>, +} + +/// One recorded pty request's translated shapes — (term, cols, rows, +/// pixel_width, pixel_height). +pub type PtyRecord = (String, u32, u32, u32, u32); + +/// One recorded window change's shapes — (cols, rows, pixel_width, +/// pixel_height). +pub type WindowRecord = (u32, u32, u32, u32); + +/// One recorded exec request's command, verbatim bytes. +pub type ExecRecord = Vec; + +impl SshdObserved { + /// The exec commands seen (`exec_request`'s wire string, + /// verbatim bytes). + pub fn execs(&self) -> Vec> { + self.execs.lock().clone() + } + + /// The pty requests seen (term, cols, rows, pixel dims). + pub fn ptys(&self) -> Vec { + self.ptys.lock().clone() + } + + /// Whether any `shell_request` was seen. + pub fn shells(&self) -> Vec<()> { + self.shells.lock().clone() + } + + /// The window changes seen (cols, rows, pixel dims). + pub fn windows(&self) -> Vec { + self.windows.lock().clone() + } + + /// The signals seen (russh's `Sig` debug names — "TERM", + /// `Custom("..")`, …). + pub fn signals(&self) -> Vec { + self.signals.lock().clone() + } +} + +/// The double's scripted program serve — after a request's success +/// reply: stdout, then `extended_data(1)` stderr, then the exit +/// report, then the close (the vanilla sshd's exit-then-close +/// discipline). +async fn scripted_program(handle: Handle, channel_id: russh::ChannelId, script: Arc) { + if !script.stdout.is_empty() + && handle + .data(channel_id, script.stdout.to_vec()) + .await + .is_err() + { + return; + } + if !script.stderr.is_empty() + && handle + .extended_data(channel_id, 1, script.stderr.to_vec()) + .await + .is_err() + { + return; + } + match script.exit { + Some(DblExit::Status(code)) => { + let _ = handle.exit_status_request(channel_id, code).await; + } + Some(DblExit::Signal(name)) => { + let sig = match name { + "TERM" => russh::Sig::TERM, + "INT" => russh::Sig::INT, + other => russh::Sig::Custom(other.to_owned()), + }; + let _ = handle + .exit_signal_request(channel_id, sig, false, String::new(), "en".to_owned()) + .await; + } + None => {} + } + let _ = handle.close(channel_id).await; +} + +/// The direct opens' echo pump — the accepted-typed-open shape's +/// server half, POC-1 §7's byte round-trip. +async fn direct_echo(mut stream: russh::ChannelStream) { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let mut buf = vec![0u8; 4096]; + loop { + match stream.read(&mut buf).await { + Ok(0) | Err(_) => return, + Ok(count) => { + if stream.write_all(&buf[..count]).await.is_err() { + return; + } + } + } + } +} /// The fake `sshd`'s per-connection handler — accept-set identity /// resolution, explicit rejects everywhere, the `auth_succeeded` @@ -62,6 +252,9 @@ struct SshdDoubleHandler { accept_session_channels: bool, sessions_out: Arc>>>, opened_sessions_out: tokio::sync::mpsc::UnboundedSender>, + script: Arc, + observed: SshdObserved, + remaining_declines: std::sync::Arc, } impl russh::server::Handler for SshdDoubleHandler { @@ -126,13 +319,39 @@ impl russh::server::Handler for SshdDoubleHandler { /// The acceptor stub: armed rows accept client session opens and /// hand the channel out; unarmed rows decline cleanly (the - /// vehicle never accept-and-drops either). + /// vehicle never accept-and-drops either); a counted refusal + /// declines the first opens per connection (the + /// stays-usable-after-a-decline rows' script). async fn channel_open_session( &mut self, channel: Channel, reply: russh::server::ChannelOpenHandle, _session: &mut Session, ) -> Result<(), Self::Error> { + let mut declined = false; + while !declined { + let current = self + .remaining_declines + .load(std::sync::atomic::Ordering::SeqCst); + if current == 0 { + break; + } + declined = self + .remaining_declines + .compare_exchange( + current, + current - 1, + std::sync::atomic::Ordering::SeqCst, + std::sync::atomic::Ordering::SeqCst, + ) + .is_ok(); + } + if declined { + reply + .reject(russh::ChannelOpenFailure::AdministrativelyProhibited) + .await; + return Ok(()); + } if self.accept_session_channels { self.opened_sessions_out .send(channel) @@ -145,6 +364,151 @@ impl russh::server::Handler for SshdDoubleHandler { } Ok(()) } + + /// The `direct-tcpip` script: reject (the scripted + /// server-refusal class, `ConnectFailed` — the vanilla sshd's + /// dead-target shape) or accept + echo pump. + async fn channel_open_direct_tcpip( + &mut self, + channel: Channel, + _host_to_connect: &str, + _port_to_connect: u32, + _originator_address: &str, + _originator_port: u32, + reply: russh::server::ChannelOpenHandle, + _session: &mut Session, + ) -> Result<(), Self::Error> { + match self.script.direct { + DblDirect::Reject => reply.reject(ChannelOpenFailure::ConnectFailed).await, + DblDirect::Echo => { + reply.accept().await; + tokio::spawn(direct_echo(channel.into_stream())); + } + } + Ok(()) + } + + /// The `direct-streamlocal` script (same behaviors). + async fn channel_open_direct_streamlocal( + &mut self, + channel: Channel, + _socket_path: &str, + reply: russh::server::ChannelOpenHandle, + _session: &mut Session, + ) -> Result<(), Self::Error> { + match self.script.streamlocal { + DblDirect::Reject => reply.reject(ChannelOpenFailure::ConnectFailed).await, + DblDirect::Echo => { + reply.accept().await; + tokio::spawn(direct_echo(channel.into_stream())); + } + } + Ok(()) + } + + /// The pty script: the request is recorded and answered per the + /// script (the request-refusal rows' class). + #[allow(clippy::too_many_arguments)] + async fn pty_request( + &mut self, + channel: russh::ChannelId, + term: &str, + col_width: u32, + row_height: u32, + pix_width: u32, + pix_height: u32, + _modes: &[(russh::Pty, u32)], + session: &mut Session, + ) -> Result<(), Self::Error> { + self.observed.ptys.lock().push(( + term.to_string(), + col_width, + row_height, + pix_width, + pix_height, + )); + self.reply_script_request(channel, session).await; + Ok(()) + } + + /// The shell script: recorded + answered per the script; a + /// success launches the scripted program. + async fn shell_request( + &mut self, + channel: russh::ChannelId, + session: &mut Session, + ) -> Result<(), Self::Error> { + self.observed.shells.lock().push(()); + self.reply_script_request(channel, session).await; + Ok(()) + } + + /// The exec script: the wire command is recorded verbatim (the + /// client rows' construction/translation assertions), the reply + /// per the script, and a success launches the scripted program. + async fn exec_request( + &mut self, + channel: russh::ChannelId, + command: &[u8], + session: &mut Session, + ) -> Result<(), Self::Error> { + self.observed.execs.lock().push(command.to_vec()); + self.reply_script_request(channel, session).await; + Ok(()) + } + + /// The window-change observation (the pty control rows' translation + /// witness). + async fn window_change_request( + &mut self, + _channel: russh::ChannelId, + col_width: u32, + row_height: u32, + pix_width: u32, + pix_height: u32, + _session: &mut Session, + ) -> Result<(), Self::Error> { + self.observed + .windows + .lock() + .push((col_width, row_height, pix_width, pix_height)); + Ok(()) + } + + /// The signal observation (names as russh carries them). + async fn signal( + &mut self, + _channel: russh::ChannelId, + signal: russh::Sig, + _session: &mut Session, + ) -> Result<(), Self::Error> { + self.observed.signals.lock().push(format!("{signal:?}")); + Ok(()) + } +} + +impl SshdDoubleHandler { + /// The scripted request reply: success launches the scripted + /// program through the session's handle (spawned — the packet + /// loop returns fast) *when a program is scripted at all* — an + /// output-less/no-exit script launches nothing (the interactive + /// sessions stay open; close is the client's or the connection's); + /// the failure class answers `channel_failure`. + async fn reply_script_request(&self, channel: russh::ChannelId, session: &mut Session) { + if !self.script.request_success { + let _ = session.channel_failure(channel); + return; + } + let _ = session.channel_success(channel); + let has_program = !self.script.stdout.is_empty() + || !self.script.stderr.is_empty() + || self.script.exit.is_some(); + if has_program { + let (handle, channel_id) = (session.handle(), channel); + let script = Arc::clone(&self.script); + tokio::spawn(scripted_program(handle, channel_id, script)); + } + } } /// The fake `sshd`'s accepted-identity register — exact-string @@ -174,20 +538,23 @@ pub struct SshdDouble { accept_task: tokio::task::JoinHandle<()>, } -/// Spawns the double on `127.0.0.1:0`: `host_keys` rides the -/// `russh::server::Config`; `accept_session_channels` arms the -/// session-open acceptor stub. Returns the double and its -/// accept-set register (register before dialing — the wire's auth -/// reads it at the dial). -pub async fn spawn( +/// Spawns the double on `127.0.0.1:0` under the given op-row script: +/// `host_keys` rides the `russh::server::Config`; +/// `accept_session_channels` arms the session-open acceptor stub. +/// Returns the double, its accept-set register (register before +/// dialing — the wire's auth reads it at the dial), and the +/// observation surface (see [`SshdObserved`]). +pub async fn spawn_scripted( host_keys: Vec, accept_session_channels: bool, -) -> (SshdDouble, SshdAccepts) { + script: SshdScript, +) -> (SshdDouble, SshdAccepts, SshdObserved) { let accepts = Arc::new(parking_lot::RwLock::new(Vec::new())); let (session_tx, session_rx) = tokio::sync::oneshot::channel(); let (opened_tx, opened_rx) = tokio::sync::mpsc::unbounded_channel(); let (outcomes_tx, outcomes_rx) = tokio::sync::mpsc::unbounded_channel(); let sessions_out = Arc::new(Mutex::new(Some(session_tx))); + let script = Arc::new(script); let config = Arc::new(russh::server::Config { methods: MethodSet::from(&[MethodKind::PublicKey][..]), @@ -202,6 +569,9 @@ pub async fn spawn( let addr = listener.local_addr().expect("the double has an address"); let accepts_for_loop = Arc::clone(&accepts); + let observed = SshdObserved::default(); + let observed_for_loop = observed.clone(); + let script_declines = script.reject_first_session_opens; let accept_task = tokio::spawn(async move { loop { let (socket, _) = match listener.accept().await { @@ -214,6 +584,11 @@ pub async fn spawn( accept_session_channels, sessions_out: Arc::clone(&sessions_out), opened_sessions_out: opened_tx.clone(), + script: Arc::clone(&script), + observed: observed_for_loop.clone(), + remaining_declines: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new( + script_declines, + )), }; let outcomes = outcomes_tx.clone(); tokio::spawn(async move { @@ -238,9 +613,22 @@ pub async fn spawn( accept_task, }, SshdAccepts { inner: accepts }, + observed, ) } +/// Spawns the double on `127.0.0.1:0` with the plain dial rows' +/// default script — the `w4-dial` shape, kept: the double and its +/// accept-set register. +pub async fn spawn( + host_keys: Vec, + accept_session_channels: bool, +) -> (SshdDouble, SshdAccepts) { + let (double, accepts, _observed) = + spawn_scripted(host_keys, accept_session_channels, SshdScript::default()).await; + (double, accepts) +} + impl SshdDouble { /// The dialed session's server-side handle — the initiator for /// the server-initiated-open rows. Awaits the `auth_succeeded`