diff --git a/tasks/implementation/w5-e2e-client.md b/tasks/implementation/w5-e2e-client.md index f6abc5e..4c645ae 100644 --- a/tasks/implementation/w5-e2e-client.md +++ b/tasks/implementation/w5-e2e-client.md @@ -1,7 +1,7 @@ --- id: w5-e2e-client name: Wave 5 — real-tool e2e, client direction (the door against a real sshd) -status: pending +status: completed depends_on: [review-wave-4] scope: moderate risk: medium @@ -50,16 +50,16 @@ matrix task's binaries), under the check-gated ## Acceptance Criteria -- [ ] The sshd fixture lands (spawned real server, ed25519 host + +- [x] The sshd fixture lands (spawned real server, ed25519 host + user keys, the policy pin posture); its shape decided + recorded -- [ ] Exec, direct-tcpip, and the `-R` leg (port-0 mint + accepted +- [x] Exec, direct-tcpip, and the `-R` leg (port-0 mint + accepted arrival + cancel) green against real sshd (or reasoned-skipped) -- [ ] The decline rows green: sftp placeholder, unfeatured +- [x] The decline rows green: sftp placeholder, unfeatured direct-tcpip (connection stays usable), host-key mismatch hard-decline -- [ ] Every finding fixed in-task or booked for the gate (recorded) -- [ ] `cargo test` (+ the builds the new binary rides), clippy ×2 +- [x] Every finding fixed in-task or booked for the gate (recorded) +- [x] `cargo test` (+ the builds the new binary rides), clippy ×2 (`-D warnings`), fmt, `cargo doc --no-deps` green ## References @@ -80,8 +80,59 @@ matrix task's binaries), under the check-gated ## Notes -> To be filled by implementation agent +**The sshd fixture decided**: `tests/e2e_client.rs` (the new binary), reusing `tests/common` — the +wave-4 client binaries stay single-writer per the decomposition's file map; this task edited only +its own binary (`tests/e2e_client.rs`, an in-file `support` module). **Fixture shape: per-row +spawned real `sshd`** — each row's config knobs differ (`AllowTcpForwarding` yes/no) and each row +dials its own server, so a shared per-binary instance buys nothing; per-row isolation (own workdir, +own minted ed25519 host key via `ssh-keygen`, own probe-verified port from the e2e-server's +monotonic allocator posture) beats the spawn cost. Knob set: publickey-only, `StrictModes no` (the +throwaway workdir carries none of the real home-dir hygiene), `UsePAM no`, `UseDNS no`, +`PidFile none`, `PermitRootLogin prohibit-password`, host key + authorized_keys (fixture `alpha`'s +public blob) at absolute paths, per-row `AllowTcpForwarding`. Readiness is a bounded TCP probe; a +server that never comes up is the row's **reasoned skip** with the server's own `-e` log tail as +the reason (worked as non-root in this environment — verified against OpenSSH 9.6p1). The dial +rides the decrypt-at-dial path over the in-memory vault double (`VaultKeySource` over `TestVault`, +`alpha` sealed) and the case-1 pin posture (`MemHostKeyPolicy::pin`) accepts only the minted host +fingerprint. + +**Row inventory** (all check-gated, `#[cfg(unix)]` + feature-gated per row group; deadline-bounded +end to end — the POC-1 §5 posture; `Contract stamp:` markers cite `client.md` / `reverse-runner.md`): + +| Row | Build(s) | What it pins | +|---|---|---| +| `exec_row::real_sshd_exec_pumps_streams_and_reports_the_exit_status` | session (+all) | the wave-4 double-driven exec row re-landed: stdout/stderr separated through a real server, the **nonzero** exit (3) reported in-band last — the real server's reaping, parity not a canned 0; the case-1 pin accept decision on record against the really-presented key | +| `direct_tcpip_row::real_sshd_direct_tcpip_round_trips_a_live_tcp_connection` | tunnel (+all) | the `-L`-shaped dial through a real server's forwarding into a live loopback echo server, bytes end to end | +| `reverse_rows::real_sshd_port_zero_mint_serves_the_accepted_arrival_and_the_data_leg` | tunnel (+all) | the port-0 mint surfaces the actually-bound port (verified by dialing it), the server-initiated `forwarded-tcpip` arrives into the installed acceptor, the external dialer's bytes round trip — the reverse-runner consumer pattern end to end | +| `reverse_rows::real_sshd_cancel_tears_the_listen_down_and_unknown_cancels_refuse` | tunnel (+all) | explicit-port listen actually bound; the listener handle's `cancel()` frees the **real** bound port (later dials refused); a stateless cancel of the unknown listener is `ListenRefused`; connection stays usable | +| `declines::real_connection_sftp_placeholder_declines_with_the_connection_intact` | **every build** | the OQ-13 placeholder against a real authenticated connection: typed decline before any wire op, connection intact | +| `declines::real_presented_key_mismatch_hard_declines_and_the_right_pin_dials` | **every build** | the wrong-pinned policy hard-declines the dial (`DialError::HostKeyRejected`), the policy log records the **real** fingerprint as rejected, and the correctly-pinned dial to the same server succeeds — the decline is the policy's decision, not the server's | +| `unfeatured_direct_decline::real_server_refused_direct_tcpip_declines_and_the_connection_stays_usable` | tunnel+session (all-features) | a real server with `AllowTcpForwarding no` refuses the typed open → `OpenDeclined(Prohibited)`, and a following exec is served — the connection-stays-usable fact end to end | + +**Decisions recorded** (for the wave-5 gate): + +- The "connection stays usable" decline row is gated `tunnel` **and** `session` (rides + `--all-features` only): the post-decline exec is the stays-usable proof's vehicle, and exec is + a `session`-gated op. Under tunnel-only builds the row is absent; the fake-double mirror + (`tests/client_ops.rs`) carries the tunnel-only stays-usable rows. +- The sftp and host-key-mismatch rows run in **every** build (feature-free ops on + `SshConnection`), so even the no-feature matrix build exercises two real-server rows. +- The fixture skips are honest: no `sshd` in the environment / no `ssh-keygen` / spawn refusal / + never-listening (the server's `-e` log tail carried in the reason) — never silently green. + +**Findings**: no src/ changes needed — every observed issue was e2e-graph local +(compile/fmt-level only). No bookings for the gate beyond the decisions above. ## Summary -> To be filled on completion \ No newline at end of file +Completed 2026-10-30. The client-direction real-tool e2e landed as `tests/e2e_client.rs` (7 rows, +fixture shape above): exec (exit-status parity, nonzero through real reaping), the direct-tcpip +dial, the `-R` leg (port-0 mint + accepted arrival + data leg + cancel tearing the real listen +down + unknown-cancel `ListenRefused`), and the three decline rows (sftp placeholder, host-key +mismatch hard-decline with the real presented fingerprint on record, server-refused direct-tcpip +with the connection staying usable). Verification: full `scripts/matrix.sh` run green — `cargo +test` per build (none / git / session / tunnel / sftp / --all-features; the new binary compiles +in every build and runs only its rows' builds: 2 / 2 / 3 / 5 / 2 / 7 rows), both clippy runs +`-D warnings` clean, fmt clean, `cargo doc --no-deps` clean; plus 10 consecutive all-features +e2e_client runs green (no flakes; per-row workdirs + the e2e-server's port allocator carried +over). Skips are reasoned prints only; on this host sshd 9.6p1 is present, so every row ran. \ No newline at end of file diff --git a/tests/e2e_client.rs b/tests/e2e_client.rs new file mode 100644 index 0000000..fa17dbc --- /dev/null +++ b/tests/e2e_client.rs @@ -0,0 +1,978 @@ +//! The client-direction real-tool e2e (`w5-e2e-client`) — the +//! reverse-runner shape re-landed against a **real `sshd`** +//! (`docs/plans/implementation.md` §Wave 5, the real-tool e2e bullet's +//! client half). The wave-4 client rows ran against the in-process +//! fake `sshd` double (`tests/common/sshd_double.rs`); the wave-4 gate +//! named real-`sshd` e2e wave-5 territory explicitly. This binary +//! dials a spawned OpenSSH server with the shipped crate: exec +//! (exit-status parity through a real server's reaping), the +//! direct-tcpip dial (`-L`-shaped), the `-R` leg (the port-0 mint +//! against a real server's actually-bound reply, the server-initiated +//! `forwarded-tcpip` open served through the acceptor registry — the +//! reverse-runner consumer pattern end to end), the cancel form +//! tearing the real listen down, and the decline rows (the sftp +//! client-side placeholder decline, the server-refused direct-tcpip +//! open with the connection staying usable, the host-key policy +//! mismatch hard-decline — the case-1 pin posture against a real +//! presented key). +//! +//! **Placement decision (recorded, the task's acceptance criterion +//! 1):** a dedicated `tests/e2e_client.rs` binary reusing +//! `tests/common` — the wave-4 client binaries (`tests/client_dial.rs` +//! / `client_ops.rs` / `client_reverse_listen.rs`) keep their +//! single-writer posture for `w5-decline-matrix`; this binary owns +//! only its own rows and its in-file `support` fixture module, +//! disjoint from `tests/e2e_server.rs` (the matrix task's binaries). +//! +//! **The sshd fixture shape (decided + recorded):** a **per-row +//! spawned** real `sshd` — each row's config knobs differ +//! (`AllowTcpForwarding` yes/no) and each row dials its own server, +//! so a shared per-binary instance buys nothing: per-row isolation +//! (own workdir, own minted ed25519 host key, own probed port) beats +//! the spawn cost. The knob set: publickey-only +//! (`PubkeyAuthentication yes`, password + keyboard-interactive off), +//! `StrictModes no` (the fixture's throwaway workdir carries none of +//! the real home-dir hygiene), `UsePAM no`, `UseDNS no`, `PidFile +//! none`, the minted host key + the fixture-`alpha`-authorized +//! `authorized_keys` at absolute paths, and per-row +//! `AllowTcpForwarding`. Readiness is a bounded TCP probe against the +//! dialed port; a server that never comes up (binary absent, +//! `ssh-keygen` absent, spawn refused, privilege-separation directory +//! missing, config refused) is the row's **reasoned skip** with the +//! server's own `-e` log as the reason — never silently green. The +//! dial rides the decrypt-at-dial path over the in-memory vault +//! double (`VaultKeySource` over `TestVault`, fixture `alpha` sealed), +//! and the case-1 pin posture (`MemHostKeyPolicy::pin`) accepts only +//! the fixture's minted host fingerprint. +//! +//! Rows are `#[cfg(unix)]` (the fixture spawns processes) and +//! feature-gated per row group under the established check-gated +//! **skipped-with-reason** posture. Every real-tool interaction is +//! deadline-bounded — a decline must surface a typed value to the +//! dialer, never a hang. Rows carry `Contract stamp:` doc markers +//! against `client.md` / `reverse-runner.md` (the greppable +//! `tests/common` discipline). + +mod common; + +#[cfg(unix)] +mod support { + #![allow(dead_code)] + + use std::io::Read; + use std::path::{Path, PathBuf}; + use std::process::Command; + use std::sync::Arc; + use std::time::Duration; + + use alkssh::{DialConfig, DialError, HostKeyPolicy, SshConnection, VaultKeySource, dial}; + + use super::common::{MemHostKeyPolicy, TestVault, fixture}; + + /// One shared row deadline — every real-process interaction is + /// bounded (the POC-1 §5 posture: a decline never hangs). + pub const ROW_DEADLINE: Duration = Duration::from_secs(30); + + /// The sshd candidates probed in order (`sbin` paths first — the + /// server binary rarely sits on the user's `PATH`). + const SSHD_CANDIDATES: &[&str] = &[ + "/usr/sbin/sshd", + "/usr/local/sbin/sshd", + "/sbin/sshd", + "/usr/bin/sshd", + ]; + + /// Prints the row's reasoned skip (never silently green). + pub fn skip(row: &str, reason: &str) { + eprintln!("SKIPPED (with reason): {row} — {reason}"); + } + + /// The `sshd` binary's location — probed once, `None` when the + /// environment carries no server (every row's reasoned skip). + pub fn sshd_binary() -> Option { + SSHD_CANDIDATES + .iter() + .map(PathBuf::from) + .find(|path| path.is_file()) + } + + /// True when spawning the probe succeeds at all (the minting + /// tool's own exit codes are its business). + pub fn keygen_present() -> bool { + Command::new("ssh-keygen").arg("-h").output().is_ok() + } + + /// The per-row fixture workdir — a throwaway directory under the + /// temp dir, one per row (the wave-5 e2e-server lesson: no shared + /// workdirs, no cross-row races). + fn workdir(row: &str) -> PathBuf { + let work = + std::env::temp_dir().join(format!("alkssh-e2e-client-{row}-{}", std::process::id())); + std::fs::create_dir_all(&work).expect("the row workdir creates"); + work + } + + /// Mints one ed25519 host key into the workdir (the independent + /// issuer) and returns its recorded fingerprint — byte-verbatim + /// from `ssh-keygen -lf`, the same `SHA256:` format the + /// fingerprint parity contract pins (the policy's pin comparison + /// is exact string equality by contract). + fn mint_host_key(work: &Path) -> String { + let key_path = work.join("hostkey"); + let minted = Command::new("ssh-keygen") + .args(["-q", "-t", "ed25519", "-N", "", "-C", "alkssh-e2e-fixture"]) + .arg("-f") + .arg(&key_path) + .output() + .expect("ssh-keygen spawns"); + assert!( + minted.status.success(), + "the host key mints — stderr: {}", + String::from_utf8_lossy(&minted.stderr) + ); + let listed = Command::new("ssh-keygen") + .arg("-lf") + .arg(key_path.with_extension("pub")) + .output() + .expect("ssh-keygen -lf spawns"); + let text = String::from_utf8_lossy(&listed.stdout).to_string(); + text.split_whitespace() + .find(|token| token.starts_with("SHA256:")) + .expect("the fingerprint line carries a SHA256: fingerprint") + .to_owned() + } + + /// The in-process port allocator for the rows' named binds (the + /// e2e-server posture verbatim: a monotonic counter over a window + /// below the kernel's ephemeral range, probe-verified per + /// allocation — parallel rows never collide). + pub async fn pick_port() -> u16 { + static NEXT: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(2000); + loop { + let raw = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let port = 2000 + (raw % 15_000) as u16; + match tokio::net::TcpListener::bind(("127.0.0.1", port)).await { + Ok(listener) => { + drop(listener); + return port; + } + Err(_) => continue, + } + } + } + + /// The spawned real `sshd` — the per-row fixture (the shape + /// recorded in the module docs). The dial rides the + /// decrypt-at-dial path over the in-memory vault double; the + /// case-1 pinned policy accepts exactly the minted host key. + pub struct SshdFixture { + pub work: PathBuf, + pub port: u16, + pub host_fingerprint: String, + pub username: String, + child: tokio::process::Child, + } + + impl SshdFixture { + /// A fixture dialing publickey-only with TCP forwarding + /// enabled (`AllowTcpForwarding yes`) — the op rows' server. + pub async fn spawn(row: &str) -> Result { + Self::spawn_with(row, true).await + } + + /// A fixture with an explicit `AllowTcpForwarding` knob (the + /// server-refused direct-tcpip row's `no` posture). + pub async fn spawn_with(row: &str, forwarding: bool) -> Result { + let Some(server) = sshd_binary() else { + return Err("no sshd binary in the environment".to_owned()); + }; + if !keygen_present() { + return Err("ssh-keygen is unavailable".to_owned()); + } + + let work = workdir(row); + let host_fingerprint = mint_host_key(&work); + + let authorized = work.join("authorized_keys"); + std::fs::write( + &authorized, + format!("{} alkssh-e2e-user\n", fixture("alpha").blob), + ) + .expect("the authorized_keys file writes"); + + let port = pick_port().await; + let config_path = work.join("sshd_config"); + let log_path = work.join("sshd.log"); + std::fs::write( + &config_path, + format!( + "Port {port}\n\ + ListenAddress 127.0.0.1\n\ + HostKey {host}\n\ + AuthorizedKeysFile {authorized}\n\ + PubkeyAuthentication yes\n\ + PasswordAuthentication no\n\ + KbdInteractiveAuthentication no\n\ + PermitRootLogin prohibit-password\n\ + UsePAM no\n\ + UseDNS no\n\ + StrictModes no\n\ + PidFile none\n\ + LogLevel INFO\n\ + AllowTcpForwarding {forwarding}\n\ + AllowAgentForwarding no\n\ + X11Forwarding no\n", + host = work.join("hostkey").display(), + authorized = authorized.display(), + ), + ) + .expect("the sshd_config writes"); + + let log = std::fs::File::create(&log_path).expect("the sshd log file opens"); + let mut child = tokio::process::Command::new(&server) + .args(["-D", "-e", "-f"]) + .arg(&config_path) + .kill_on_drop(true) + .stderr(log) + .spawn() + .map_err(|error| format!("sshd refuses to spawn ({error})"))?; + + let username = Command::new("id") + .arg("-un") + .output() + .ok() + .map(|out| String::from_utf8_lossy(&out.stdout).trim().to_owned()) + .filter(|user| !user.is_empty()) + .ok_or_else(|| "the connecting user cannot be resolved (id -un)".to_owned())?; + + let readiness = tokio::time::timeout(Duration::from_secs(15), async { + loop { + if tokio::net::TcpStream::connect(("127.0.0.1", port)) + .await + .is_ok() + { + return Ok(()); + } + match child.try_wait() { + Ok(Some(status)) => { + return Err(format!( + "sshd exits at startup (status {status}): {}", + log_tail(&log_path) + )); + } + Ok(None) => {} + Err(error) => { + return Err(format!( + "sshd cannot be polled ({error}): {}", + log_tail(&log_path) + )); + } + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .await; + match readiness { + Ok(Ok(())) => {} + Ok(Err(reason)) => { + let _ = child.kill().await; + return Err(reason); + } + Err(_) => { + let _ = child.kill().await; + return Err(format!( + "sshd never opened its listen port: {}", + log_tail(&log_path) + )); + } + } + + Ok(Self { + work, + port, + host_fingerprint, + username, + child, + }) + } + + /// The dialed address (the dial config's `host:port` form). + pub fn address(&self) -> String { + format!("127.0.0.1:{}", self.port) + } + + /// The case-1 pinned policy — accepts exactly this fixture's + /// minted host fingerprint at `127.0.0.1` (the pin posture + /// consumed against a real presented key). A `Some` pin + /// substitutes another fixture's fingerprint (the mismatch + /// row's wrong-pin posture). + pub fn policy(&self, wrong_pin: Option<&'static str>) -> Arc { + let policy = Arc::new(MemHostKeyPolicy::new()); + match wrong_pin { + Some(pin) => policy.pin("127.0.0.1", pin), + None => policy.pin("127.0.0.1", &self.host_fingerprint), + } + policy + } + + /// The authenticated dial over the fixture under the given + /// policy — the decrypt-at-dial path (fixture `alpha` sealed + /// into the in-memory vault, decrypted and loaded exactly at + /// the dial), deadline-bounded. + pub async fn dial_from( + &self, + policy: Arc, + ) -> Result { + let vault = TestVault::unlocked(); + let source = Arc::new(VaultKeySource::new( + vault.handle(), + Some(vault.seal(fixture("alpha"))), + )); + let config = DialConfig::new( + self.address(), + self.username.clone(), + source, + Arc::clone(&policy) as Arc, + ); + tokio::time::timeout(ROW_DEADLINE, dial(config)) + .await + .expect("the dial completes within the row deadline") + } + + /// The dial the op rows use — pinned to the fixture's own + /// key, so the authenticated connection is the expected `Ok`. + pub async fn dial_default(&self) -> SshConnection { + self.dial_from(self.policy(None)) + .await + .expect("the pinned dial succeeds against the real server") + } + + /// Stops the spawned server and removes the workdir (best + /// effort; a panicking row leaves both to the OS). + pub async fn stop(mut self) { + let _ = self.child.kill().await; + let _ = self.child.wait().await; + let _ = std::fs::remove_dir_all(&self.work); + } + } + + /// The server's captured `-e` diagnostics — the skip reason's raw + /// input, tail-truncated (best effort; a missing log is no text). + fn log_tail(log: &Path) -> String { + let Ok(mut file) = std::fs::File::open(log) else { + return String::new(); + }; + let mut text = String::new(); + if file.read_to_string(&mut text).is_err() { + return String::new(); + } + let from = text.len().saturating_sub(512); + text[from..] + .lines() + .rev() + .take(4) + .collect::>() + .join(" | ") + } + + /// A tiny TCP echo server — the direct-tcpip row's real `-L` + /// target (the bytes round trip through the server-side bridge + /// and this server, exactly the `-L` topology). + pub async fn spawn_echo_server() -> u16 { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("the echo listener binds"); + let port = listener.local_addr().expect("local addr").port(); + tokio::spawn(async move { + loop { + let Ok((mut socket, _)) = listener.accept().await else { + return; + }; + tokio::spawn(async move { + let (mut read, mut write) = socket.split(); + let _ = tokio::io::copy(&mut read, &mut write).await; + }); + } + }); + port + } + + /// The bounded byte round trip a data leg runs: the request in, + /// the half-close (the wire EOF), the echo back, the end. + pub async fn round_trip(mut socket: S, request: &[u8]) -> Vec + where + S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, + { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + tokio::time::timeout(ROW_DEADLINE, async move { + socket.write_all(request).await.expect("the request writes"); + socket.shutdown().await.expect("the half-close writes"); + let mut responded = Vec::new(); + socket + .read_to_end(&mut responded) + .await + .expect("the echo reads to the end"); + responded + }) + .await + .expect("the data leg completes within the row deadline") + } +} + +/// The exec row (the `session` build): the family-typed exec through +/// a real server — stdout rides `data`, stderr rides +/// `extended_data(1)`, the exit status rides in-band last, and the +/// server's reaping reports the program's real exit code (a nonzero +/// exit proves the parity is the server's, not a canned 0). +#[cfg(all(unix, feature = "session"))] +mod exec_row { + use std::time::Duration; + + use alktty::ControlMessage; + use alktty::wire::{ChunkReader, STREAM_CTRL_OUT, STREAM_STDERR, STREAM_STDOUT}; + + use super::support; + + /// Drains the session's chunk plane until the exit chunk — + /// returns (stdout, stderr, the exit code in-band, if any). The + /// deadline bounds a wrong row instead of hanging it. + 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(15); + loop { + let chunk = match tokio::time::timeout_at(deadline, reader.read_chunk()).await { + Ok(Ok(chunk)) => chunk, + _ => break, + }; + match chunk.stream_type { + STREAM_STDOUT => stdout.extend_from_slice(&chunk.bytes), + STREAM_STDERR => stderr.extend_from_slice(&chunk.bytes), + STREAM_CTRL_OUT => { + if let Ok(ControlMessage::Exit { code }) = + ControlMessage::from_slice(&chunk.bytes) + { + exit = Some(code); + } + break; + } + _ => {} + } + } + (stdout, stderr, exit) + } + + /// - *Contract stamp:* `client.md` §op rows (row 1, the exec + /// parity) + §The dial surface — the wave-4 double-driven exec + /// row re-landed against a real `sshd`: the family-typed op + /// (`exec()`, the command verbatim) through the real server's + /// own session reaping — stdout separated from stderr, the + /// real exit status reported in-band last (`exit-status(3)`, + /// parity — not a canned 0), and the case-1 pinned policy's + /// accept decision on record against the really-presented key. + #[tokio::test] + async fn real_sshd_exec_pumps_streams_and_reports_the_exit_status() { + let fixture = match support::SshdFixture::spawn("exec").await { + Ok(fixture) => fixture, + Err(reason) => { + support::skip("exec row", &reason); + return; + } + }; + + let policy = fixture.policy(None); + let connection = fixture + .dial_from(policy.clone()) + .await + .expect("the pinned dial succeeds against the real server"); + + let session = tokio::time::timeout( + support::ROW_DEADLINE, + connection.exec("printf out-ok && printf err-bad 1>&2 && exit 3"), + ) + .await + .expect("the exec opens within the row deadline") + .expect("the exec opens against the real server"); + + let (stdout, stderr, exit) = read_to_exit(session).await; + assert_eq!(stdout, b"out-ok", "stdout rode the data stream"); + assert_eq!(stderr, b"err-bad", "stderr rode extended_data(1)"); + assert_eq!( + exit, + Some(3), + "the real server's reaping reports the program's exit status — parity, not a canned code" + ); + assert!( + !connection.is_closed(), + "the connection outlives a finished exec" + ); + + let decisions = policy.decisions(); + assert_eq!( + decisions.len(), + 1, + "the dial called the policy exactly once — the real presented key" + ); + let decision = &decisions[0]; + assert!(decision.accepted, "the pinned accept is the dial's verdict"); + assert_eq!(decision.presented.host, "127.0.0.1"); + assert_eq!(decision.presented.port, fixture.port); + assert_eq!( + decision.presented.fingerprint, fixture.host_fingerprint, + "the presented fingerprint is the minted host key's recorded one" + ); + + connection.close().await; + fixture.stop().await; + } +} + +/// The direct-tcpip row (the `tunnel` build): the `-L`-shaped dial +/// through a real server's forwarding — a real TCP connection to a +/// live loopback echo server, bytes end to end. +#[cfg(all(unix, feature = "tunnel"))] +mod direct_tcpip_row { + use super::support; + + /// - *Contract stamp:* `client.md` §op rows (row 2, the + /// direct-tcpip dial) — the wave-4 double-driven `-L` row + /// re-landed against a real server: `tunnel_direct` opens the + /// channel on the real wire, the plain byte duplex pumps, and + /// a **live** TCP listener on the far side answers + /// byte-for-byte (the server's own connect is exercised, not a + /// canned echo). + #[tokio::test] + async fn real_sshd_direct_tcpip_round_trips_a_live_tcp_connection() { + let fixture = match support::SshdFixture::spawn("direct-tcpip").await { + Ok(fixture) => fixture, + Err(reason) => { + support::skip("direct-tcpip row", &reason); + return; + } + }; + + let connection = fixture.dial_default().await; + let target_port = support::spawn_echo_server().await; + + let channel = tokio::time::timeout( + support::ROW_DEADLINE, + connection.tunnel_direct("127.0.0.1", u32::from(target_port)), + ) + .await + .expect("the tunnel op completes within the row deadline") + .expect("the real server serves the direct-tcpip open"); + + let responded = support::round_trip(channel, b"ping-alkssh-e2e\n").await; + assert_eq!( + responded, b"ping-alkssh-e2e\n", + "the bytes cross the real server's forwarding end to end" + ); + assert!(!connection.is_closed()); + + connection.close().await; + fixture.stop().await; + } +} + +/// The `-R` rows (the `tunnel` build): the port-0 mint against a real +/// server's actually-bound reply, the server-initiated +/// `forwarded-tcpip` open served through the acceptor registry (the +/// reverse-runner consumer pattern end to end), and the cancel form +/// tearing the real listen down. +#[cfg(all(unix, feature = "tunnel"))] +mod reverse_rows { + use std::sync::Arc; + use std::time::Duration; + + use alkssh::{ForwardedTcpip, ReverseEndpoint, TcpipForwardedAcceptor}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + use super::support; + + /// The acceptor double — records each open's payload fields and + /// pumps the duplex as an echo (the assembly's pump shape). + #[derive(Clone)] + struct AcceptorDouble { + opens: tokio::sync::mpsc::UnboundedSender<(String, u32, String, u32)>, + } + + #[async_trait::async_trait] + impl TcpipForwardedAcceptor for AcceptorDouble { + async fn accept(&self, open: ForwardedTcpip) { + let mut stream = open.stream; + tokio::spawn(async move { + 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; + } + } + } + } + }); + let _ = self.opens.send(( + open.connected_address, + open.connected_port, + open.originator_address, + open.originator_port, + )); + } + } + + /// - *Contract stamp:* `client.md` §op rows (row 4, the `-R` + /// listen + ingest) + `reverse-runner.md` (the consumer + /// pattern) — the wave-4 double-driven row re-landed against a + /// real server: the port-0 mint returns the **actually-bound** + /// port (a real server's reply, verified by dialing it), the + /// server-initiated `forwarded-tcpip` open arrives into the + /// installed acceptor, and the end-to-end consumer pattern + /// pumps: an external dialer → sshd's listen → the arrival → + /// the acceptor's duplex → the echo back. + #[tokio::test] + async fn real_sshd_port_zero_mint_serves_the_accepted_arrival_and_the_data_leg() { + let fixture = match support::SshdFixture::spawn("reverse-mint").await { + Ok(fixture) => fixture, + Err(reason) => { + support::skip("-R mint row", &reason); + return; + } + }; + + let connection = fixture.dial_default().await; + let (opens_tx, mut opens_rx) = tokio::sync::mpsc::unbounded_channel(); + connection + .acceptors() + .install_tcpip(Arc::new(AcceptorDouble { opens: opens_tx })) + .expect("the acceptor slot installs"); + + let listener = tokio::time::timeout( + support::ROW_DEADLINE, + connection.reverse_listen("127.0.0.1", 0), + ) + .await + .expect("the listen completes within the row deadline") + .expect("the real server grants the forward"); + + let bound = match listener.endpoint() { + ReverseEndpoint::Tcp { address, port } => { + assert_eq!( + address, "127.0.0.1", + "the wire address is the requested one" + ); + *port + } + other => panic!("the tcp endpoint is the mint's shape, got {other:?}"), + }; + assert!( + bound != 0, + "the mint surfaces the peer-picked port, never the requested 0" + ); + + let mut dialer = tokio::time::timeout( + support::ROW_DEADLINE, + tokio::net::TcpStream::connect(( + "127.0.0.1", + u16::try_from(bound).expect("the bound port fits u16"), + )), + ) + .await + .expect("the dialer connects within the row deadline") + .expect("the minted port is actually bound — a real server's reply, dialable locally"); + + let (connected_address, connected_port, originator_address, originator_port) = + tokio::time::timeout(Duration::from_secs(15), opens_rx.recv()) + .await + .expect("the arrival lands in time") + .expect("the acceptor sender lives"); + assert_eq!(connected_address, "127.0.0.1"); + assert_eq!( + connected_port, bound, + "the arrival echoes the listen request" + ); + assert_eq!(originator_address, "127.0.0.1"); + assert!( + originator_port != 0, + "the originator is the external dialer's ephemeral source port" + ); + + let responded = support::round_trip(&mut dialer, b"reverse-leg-marker\n").await; + assert_eq!( + responded, b"reverse-leg-marker\n", + "the consumer pattern's data leg round trips through the real server" + ); + + assert!(!connection.is_closed()); + drop(listener); + connection.close().await; + fixture.stop().await; + } + + /// - *Contract stamp:* `client.md` §Lifecycle primitives (the + /// cancel form) — the wave-4 double-driven cancel row + /// re-landed against a real server: an explicit-port listen is + /// actually bound (dialable), the listener handle's `cancel()` + /// tears the **real** listen down (a later dial is refused), a + /// stateless cancel of the unknown listener is the wire's + /// `ListenRefused`, and the connection stays usable. + #[tokio::test] + async fn real_sshd_cancel_tears_the_listen_down_and_unknown_cancels_refuse() { + let fixture = match support::SshdFixture::spawn("reverse-cancel").await { + Ok(fixture) => fixture, + Err(reason) => { + support::skip("-R cancel row", &reason); + return; + } + }; + + let connection = fixture.dial_default().await; + let (opens_tx, _opens_rx) = tokio::sync::mpsc::unbounded_channel(); + connection + .acceptors() + .install_tcpip(Arc::new(AcceptorDouble { opens: opens_tx })) + .expect("the acceptor slot installs"); + + let port = support::pick_port().await; + let listener = tokio::time::timeout( + support::ROW_DEADLINE, + connection.reverse_listen("127.0.0.1", u32::from(port)), + ) + .await + .expect("the listen completes within the row deadline") + .expect("the real server grants the forward"); + assert_eq!( + listener.endpoint().bound_port(), + Some(u32::from(port)), + "an explicit-port listen surfaces the requested port (the wire's mint data is absent)" + ); + + let _bound_conn = tokio::time::timeout( + support::ROW_DEADLINE, + tokio::net::TcpStream::connect(("127.0.0.1", port)), + ) + .await + .expect("the probe connects within the row deadline") + .expect("the explicit port is actually bound before the cancel"); + + tokio::time::timeout(support::ROW_DEADLINE, listener.cancel()) + .await + .expect("the cancel completes within the row deadline") + .expect("the real listen tears down"); + + let mut torn_down = false; + for _ in 0..40 { + match tokio::net::TcpStream::connect(("127.0.0.1", port)).await { + Ok(_) => tokio::time::sleep(Duration::from_millis(100)).await, + Err(_) => { + torn_down = true; + break; + } + } + } + assert!( + torn_down, + "the cancel frees the real bound port — connects are refused afterwards" + ); + + let again = connection + .reverse_listen_cancel("127.0.0.1", u32::from(port)) + .await; + assert!( + matches!(again, Err(alkssh::OpError::ListenRefused)), + "the real server refuses the unknown listener, got {again:?}" + ); + assert!( + !connection.is_closed(), + "the connection stays usable after every cancel class" + ); + + connection.close().await; + fixture.stop().await; + } +} + +/// The decline rows — the sftp placeholder (every build) and the +/// host-key policy mismatch (every build): both dial a real server +/// and pin typed declines, feature-free. +#[cfg(unix)] +mod declines { + use std::sync::Arc; + + use alkssh::{DialError, OpError}; + + use super::common::fixture as key_fixture; + use super::support; + + /// - *Contract stamp:* OQ-13 + `client.md` §op rows (row 5) — + /// the sftp placeholder declines in **every** build against a + /// real authenticated connection: typed decline before any + /// wire op (the real server observes no subsystem request), + /// connection intact. + #[tokio::test] + async fn real_connection_sftp_placeholder_declines_with_the_connection_intact() { + let fixture = match support::SshdFixture::spawn("sftp-decline").await { + Ok(fixture) => fixture, + Err(reason) => { + support::skip("sftp decline row", &reason); + return; + } + }; + + let connection = fixture.dial_default().await; + let err = tokio::time::timeout(support::ROW_DEADLINE, connection.sftp()) + .await + .expect("the op completes within the row deadline") + .expect_err("the OQ-13 placeholder declines against the real connection"); + assert!(matches!(err, OpError::SftpNotServed), "got {err:?}"); + assert!( + !connection.is_closed(), + "the pre-wire decline touches no wire state — the connection stays up" + ); + connection.close().await; + fixture.stop().await; + } + + /// - *Contract stamp:* `client.md` §The dial surface (the policy + /// call at `check_server_key`, OQ-05) + Host-key §case 1 — the + /// pin posture consumed against a real presented key: a + /// mismatch hard-declines the dial (`DialError:: + /// HostKeyRejected`, no connection), the policy's decision log + /// records the **real** fingerprint as rejected, and a + /// correctly-pinned dial to the same server succeeds — the + /// decline is the policy's decision, not the server's. + #[tokio::test] + async fn real_presented_key_mismatch_hard_declines_and_the_right_pin_dials() { + let fixture = match support::SshdFixture::spawn("host-key-mismatch").await { + Ok(fixture) => fixture, + Err(reason) => { + support::skip("host-key mismatch row", &reason); + return; + } + }; + + let wrong = fixture.policy(Some(key_fixture("delta").fingerprint)); + let declined = fixture + .dial_from(Arc::clone(&wrong)) + .await + .expect_err("the mismatch hard-declines the dial"); + assert!( + matches!(declined, DialError::HostKeyRejected), + "got {declined:?}" + ); + + let decisions = wrong.decisions(); + assert_eq!( + decisions.len(), + 1, + "the dial called the policy exactly once — the real presented key" + ); + let decision = &decisions[0]; + assert!( + !decision.accepted, + "the pinned mismatch is the reject verdict" + ); + assert_eq!(decision.presented.host, "127.0.0.1"); + assert_eq!(decision.presented.port, fixture.port); + assert_eq!( + decision.presented.fingerprint, fixture.host_fingerprint, + "the presented fingerprint is the real server's minted host key" + ); + + let connection = fixture.dial_default().await; + assert!( + !connection.is_closed(), + "the correctly-pinned dial to the same server succeeds" + ); + connection.close().await; + fixture.stop().await; + } +} + +/// The server-refused direct-tcpip row (the `tunnel` **and** +/// `session` builds — the all-features posture): a real server with +/// `AllowTcpForwarding no` refuses the typed open, and the +/// connection stays usable (a following exec is served). The session +/// gate is the stays-usable proof's vehicle (recorded: under +/// tunnel-only builds the row is absent; the fake-double mirror, +/// `tests/client_ops.rs`, carries the tunnel-only stays-usable rows). +#[cfg(all(unix, feature = "tunnel", feature = "session"))] +mod unfeatured_direct_decline { + use std::time::Duration; + + use alkssh::{OpError, OpenDeclineReason}; + + use super::support; + + /// - *Contract stamp:* `client.md` §op rows (row 2's declined + /// mirror) + §Correctness facts (a decline is channel-scoped, + /// auth intact, connection usable) — the wave-4 typed-open + /// decline row re-landed against a real server's own refusal: + /// `tunnel_direct` against `AllowTcpForwarding no` → + /// `OpenDeclined(Prohibited)` (the `administratively + /// prohibited` class), and a following exec is served — the + /// connection outlives the decline end to end. + #[tokio::test] + async fn real_server_refused_direct_tcpip_declines_and_the_connection_stays_usable() { + let fixture = match support::SshdFixture::spawn_with("direct-decline", false).await { + Ok(fixture) => fixture, + Err(reason) => { + support::skip("direct-tcpip decline row", &reason); + return; + } + }; + + let connection = fixture.dial_default().await; + + let declined = tokio::time::timeout( + support::ROW_DEADLINE, + connection.tunnel_direct("127.0.0.1", 443), + ) + .await + .expect("the tunnel op completes within the row deadline") + .expect_err("the server that forbids forwarding refuses the open"); + assert!( + matches!( + declined, + OpError::OpenDeclined(OpenDeclineReason::Prohibited) + ), + "the administratively-prohibited class mirrors through, got {declined:?}" + ); + assert!( + !connection.is_closed(), + "the decline is channel-scoped — the connection survives it" + ); + + let mut session = + tokio::time::timeout(support::ROW_DEADLINE, connection.exec("echo still-usable")) + .await + .expect("the exec opens within the row deadline") + .expect("the post-decline exec is served — the connection stays usable"); + + let mut stdout = Vec::new(); + let mut reader = alktty::wire::ChunkReader::new(&mut session.output); + let deadline = tokio::time::Instant::now() + Duration::from_secs(15); + loop { + match tokio::time::timeout_at(deadline, reader.read_chunk()).await { + Ok(Ok(chunk)) if chunk.stream_type == alktty::wire::STREAM_STDOUT => { + stdout.extend_from_slice(&chunk.bytes); + } + Ok(Ok(chunk)) if chunk.stream_type == alktty::wire::STREAM_CTRL_OUT => { + break; + } + _ => break, + } + } + assert_eq!( + stdout, b"still-usable\n", + "the exec after the decline answers — auth intact, connection usable" + ); + + connection.close().await; + fixture.stop().await; + } +}