alkdev/alkssh
6
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(e2e): client-direction real-sshd e2e (w5-e2e-client)

Browse Source 
The wave-5 client-direction real-tool e2e: a dedicated
`tests/e2e_client.rs` binary dialing a per-row spawned real OpenSSH
server with the shipped crate — exec (exit-status parity 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 decline rows (sftp placeholder, host-key
mismatch hard-decline, server-refused direct-tcpip with the connection
staying usable). Fixture shape decided + recorded in the task file:
per-row spawned sshd, minted ed25519 host key, publickey-only knobs,
the decrypt-at-dial vault posture, the case-1 pin posture. No src/
changes; every row check-gated skipped-with-reason and deadline-bounded.

Verification: `scripts/matrix.sh` green (cargo test per build none/git/
session/tunnel/sftp/--all-features — e2e_client rides 2/2/3/5/2/7 rows,
clippy x2 `-D warnings`, fmt, doc), plus 10 consecutive all-features
e2e_client runs green. taskgraph validate: 42 tasks valid.
This commit is contained in:
glm-5.3-flash committed 2026-10-30 07:55:13 +00:00
1 parent 567a34a5e1
commit 69d888a231
2 files changed
+1037 -8
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tasks/implementation/w5-e2e-client.md
+59 -8
View File
@@ -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
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.
tests/e2e_client.rs
+978
View File
@@ -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<PathBuf> {
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, String> {
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<Self, String> {
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<MemHostKeyPolicy> {
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<MemHostKeyPolicy>,
) -> Result<SshConnection, DialError> {
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<dyn HostKeyPolicy>,
);
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::<Vec<_>>()
.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<S>(mut socket: S, request: &[u8]) -> Vec<u8>
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<u8>, Vec<u8>, Option<i32>) {
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;
}
}