diff --git a/tasks/implementation/w5-e2e-server.md b/tasks/implementation/w5-e2e-server.md index 0d8dfab..24137e6 100644 --- a/tasks/implementation/w5-e2e-server.md +++ b/tasks/implementation/w5-e2e-server.md @@ -1,7 +1,7 @@ --- id: w5-e2e-server name: Wave 5 — real-tool e2e, server direction (OpenSSH / git / curl against the door) -status: pending +status: completed depends_on: [review-wave-4] scope: moderate risk: medium @@ -76,8 +76,65 @@ check-gated **skipped-with-reason** posture (never silently green): ## Notes -> To be filled by implementation agent +**The e2e home decided**: `tests/e2e_server.rs` (the new binary), reusing `tests/common` — the +mapping binaries stay single-writer per the decomposition's file map; this task edits only its own +binary. Its fixtures live in-file: a `support` module (tool gates, the fixture-key/option builders, +the registered identity provider, the port allocator, the real-TCP mount) and a tunnel-gated +`fabric` module (the trimmed leg-1/leg-2 fabric double — TCP-only, happy-path: the decline and +collision machinery stays the mapping binaries'). Everything is `#[cfg(unix)]` with the body +feature-gated per row group; no src/ changes were needed — every observed issue was e2e-graph +local. + +**Row inventory** (all check-gated, **skipped-with-reason** when a tool is absent / short, never +silently green; each row's real-process interaction is deadline-bounded and asserts the door's +decline event through the capture's deadline-bounded poll): + +| Row | Build(s) | What it pins | +|---|---|---| +| `git_rows::real_git_clones_the_empty_repo_through_the_door` | git (+all) | the wave-3 clone smoke re-landed: real `git clone ssh://…` → unborn-HEAD advertisement, normalized resolved id | +| `git_rows::real_git_push_creates_the_ref_through_the_door` | git (+all) | **push** (the wave-3 smoke's gap): receive-pack through the real client, rev parity (`refs/heads/main` new-oid == local `rev-parse HEAD`), CAS-application row | +| `session_rows::real_openssh_pty_shell_session_serves_through_the_door` | session (+all) | the wave-3 pty smoke re-landed + extended: the `echo` marker rides the real PTY round trip, exit-status(0) before close | +| `session_rows::real_openssh_pipe_exec_separates_stdout_and_stderr_through_the_door` | session (+all) | `ssh host "cmd"` through the exec wrapper (`wrapper ++ [cmd]`): stdout rides SSH data, stderr rides `extended_data(1)` — POC-3 §D.1's pipe split end to end | +| `tunnel_rows::real_ssh_local_forward_serves_the_direct_tcpip_leg` | tunnel (+all) | `-L` smoke re-landed: route-1 bridge + real `tcp_dial`, body parity | +| `tunnel_rows::real_ssh_remote_forward_serves_the_arrival_and_data_legs` | tunnel (+all) | `-R`: listen, fabric-bound port, the server-initiated `forwarded-tcpip` arrival on the real wire, the client-dialed data leg — the leg the in-process smokes covered only through the fabric double | +| `tunnel_rows::real_ssh_dynamic_forward_serves_curl_through_the_direct_tcpip_leg` | tunnel (+all) | **`-D`**: curl `--socks5-hostname` through vanilla OpenSSH's client-side SOCKS → the door's `direct-tcpip` leg (POC-3 correction 2 end to end); two requests pin the per-connection channel opens | +| `unfeatured_session_decline::real_openssh_pty_attempt_declines_cleanly_in_a_feature_limited_build` | not-session (none/git/tunnel/sftp) | real `ssh -tt … echo` → nonzero exit, clean diagnostic, no hang; the decline event pins the build's class (`pty_request` "unfeatured" request-decline in the git build, `session` "unfeatured" open-decline otherwise) | +| `unfeatured_tunnel_declines::real_openssh_remote_forward_attempt_declines_cleanly` | not-tunnel | real `ssh -N -R -o ExitOnForwardFailure=yes` → exit + OpenSSH's `forwarding failed` diagnostic; `tcpip_forward` global-request decline event | +| `unfeatured_tunnel_declines::real_openssh_local_forward_connections_decline_per_connection` | not-tunnel | real `ssh -N -L` + repeated curls: every connection fails promptly (exit 7-class, no hang); `direct-tcpip` open-decline event | +| `subsystem_decline::real_sftp_subsystem_attempt_declines_with_reason` | any(session,git) | the OQ-13 placeholder from a real `sftp` client: nonzero exit, clean diagnostic; `subsystem_request` request-decline event (`unimplemented`) | +| `unknown_repo_decline::real_git_clone_of_an_unknown_repo_collapses_to_the_generic_denial` | git (+all) | real `git clone` of a nonexistent repo → exit 128, the collapsed `access denied` band naming the normalized id, **no existence oracle** | + +**Findings fixed in-task** (no src/ changes; all e2e-graph local): + +1. **Parallel-row flake — the shared key workdir**: the three tunnel rows initially shared one + `write_key` workdir, so concurrent rows raced on the same key file and one row's cleanup + deleted the dir under the others (observed as refused forwards / curl exit 7). Fixed: per-row + workdirs (`tunnel-l` / `tunnel-r` / `tunnel-d`). +2. **Parallel-row flake — port collisions**: the probe-and-drop ephemeral port picker let parallel + rows (and under load, re-entries) collide on named binds. Fixed: an in-process monotonic + allocator over the 2000–16999 window (below the kernel's ephemeral range, so `bind(0)` can + never steal one), probe-verified per allocation. +3. **Fabric-registration race**: the leg-2 fabric's openables register on a spawned task; an + `ssh -R` listen dial ahead of registration is refused, not retried (`curl never got through` + exit 7). Fixed: both fabric constructors settle on the direct/listen openables' registration + before returning (`settle`, bounded poll on `OperationRegistry::registration`). +4. **Event-window flake under CI load**: the decline rows' fixed 10 s `await_event` window can + expire on a loaded host. Fixed: `event_capture::captured_with` (additive, non-blocking peek) + + `support::decline_event` — the rows poll within the row deadline instead of a fixed window. ## Summary -> To be filled on completion \ No newline at end of file +Completed 2026-10-30. The server-direction real-tool e2e landed as `tests/e2e_server.rs` (the +placement decision above): 12 rows over the six build shapes — OpenSSH pty/shell + pipe exec +(stderr separation), `-L`/`-R`/`-D`+curl (with curl's two-request per-connection pin), git +clone **and** push (rev parity + CAS row pinned through the recording quartet), and the six +decline-observability rows (unfeatured pty/`-L`/`-R` from real clients, the sftp subsystem +decline, the unknown-repo collapsed generic denial — each also pinned by its door-side decline +event). Every finding fixed in-task (four, listed in Notes — flake hardening only, no src/ +changes). Verification: full `scripts/matrix.sh` run green — `cargo test` per build (none / git +/ session / tunnel / sftp / --all-features; the new binary rides each build compiling and +running only its rows' builds), both clippy runs `-D warnings` clean, fmt clean, +`cargo doc --no-deps` clean; plus 10 consecutive all-features e2e runs green after the flake +fixes (the parallel-port + shared-workdir + fabric-settle + event-poll hardening). Skips are +reasoned prints only (tool absence; on this host Git 2.43 / OpenSSH 9.6 / curl 8.5 / sftp are +present, so every row ran). \ No newline at end of file diff --git a/tests/common/event_capture.rs b/tests/common/event_capture.rs index b8068f7..3f56e8b 100644 --- a/tests/common/event_capture.rs +++ b/tests/common/event_capture.rs @@ -108,6 +108,22 @@ pub fn install() { let _ = capture(); } +/// The non-blocking peek — `Some` when a matching event is already on +/// record, `None` otherwise. The e2e rows poll this alongside their +/// bounded curls (the awaiting rows' `await_event` stays fixed-window; +/// a slow CI host turns the wait-then-window shape into a flake, the +/// poll-with-row-deadline shape does not). +pub fn captured_with(want: &[(&str, &str)]) -> Option { + let capture = capture(); + capture + .0 + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone() + .into_iter() + .find(|e| e.has(want)) +} + /// Waits until an event matching every field pair appears, then /// returns it — the spawned-task surfaces land their events /// asynchronously (the deadline loop is the observe-suite convention). diff --git a/tests/e2e_server.rs b/tests/e2e_server.rs new file mode 100644 index 0000000..39b6545 --- /dev/null +++ b/tests/e2e_server.rs @@ -0,0 +1,1751 @@ +//! The server-direction real-tool e2e (`w5-e2e-server`) — the POC-3 +//! gates re-landed against the shipped crate: real vanilla processes +//! (`ssh` / `git` / `curl` / `sftp`) against the adapter mounted on a +//! real TCP listener over the established in-memory doubles (the +//! wave-3 smoke composition, `tests/common/`). The row set is the +//! full server direction: OpenSSH pty/shell and pipe exec, `-L`, +//! `-R`, `-D` (the `direct-tcpip` leg), the decline-observability +//! rows (unfeatured declines from real attempts, the sftp subsystem +//! decline, the unknown-repo collapsed generic denial), and git +//! clone **and** push through the git mapping. +//! +//! **Placement decision (recorded, the task's acceptance criterion +//! 1):** a dedicated `tests/e2e_server.rs` binary reusing +//! `tests/common` — the wave-3/4 mapping binaries keep their +//! single-writer posture for `w5-decline-matrix`; this binary owns +//! only its own rows and its in-file fixture helpers (the trimmed +//! leg-1/leg-2 fabric double for the tunnel rows, the support +//! helpers). Rows arrive under the established check-gated +//! **skipped-with-reason** posture (tool absent / version short — +//! never silently green) and under a wall-clock deadline — a decline +//! must surface a clean diagnostic to the real client, never a hang +//! (POC-1 §5 posture). Rows carry `Contract stamp:` doc markers +//! against `mapping.md` / the POC-3 findings (the greppable +//! `tests/common` discipline). + +mod common; + +/// The per-binary shared helpers: tool gates, the fixture-key / +/// option builders, the registered identity provider, the ephemeral +/// port picker, and the real-TCP mount every row composes. +#[cfg(unix)] +mod support { + #![allow(dead_code)] + + use std::path::{Path, PathBuf}; + use std::process::Command; + use std::sync::Arc; + + use alkssh::{ServeConfig, SshAdapter}; + + use super::common::{MemIdentityProvider, fixture, identity}; + + /// One shared row deadline — every real-process interaction is + /// bounded (POC-1 §5: a decline never hangs). + pub const ROW_DEADLINE: std::time::Duration = std::time::Duration::from_secs(30); + + /// True when spawning the probe succeeds at all (exit codes are + /// the probe's own business — `ssh -V` writes to stderr, `sftp -h` + /// exits 1 with usage). + pub fn tool_present(binary: &str, probe: &[&str]) -> bool { + Command::new(binary).args(probe).output().is_ok() + } + + /// Prints the row's reasoned skip (never silently green). + pub fn skip(row: &str, reason: &str) { + eprintln!("SKIPPED (with reason): {row} — {reason}"); + } + + /// The per-row throwaway workdir + the fixture key written into + /// it (0600 — the real client's required file mode). + pub fn write_key(row: &str) -> PathBuf { + let work = + std::env::temp_dir().join(format!("alkssh-e2e-server-{row}-{}", std::process::id())); + std::fs::create_dir_all(&work).expect("the row workdir creates"); + let key_path = work.join("id_ed25519"); + std::fs::write(&key_path, fixture("alpha").pem).expect("the fixture key writes"); + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o600)) + .expect("the key file perms set"); + key_path + } + + /// A row's cleanup (best-effort on the success path; a panicking + /// row leaves the temp dir for the OS). + pub fn cleanup(work: &Path) { + let _ = std::fs::remove_dir_all(work); + } + + /// The decline-event observation for the e2e rows: poll the + /// capture until a matching event is on record — the row deadline + /// bounds the wait (not the capture's fixed window), so a slow CI + /// host never turns an honest decline into a flake. + pub async fn decline_event(want: &[(&str, &str)]) -> super::common::event_capture::Captured { + let deadline = std::time::Instant::now() + ROW_DEADLINE; + loop { + if let Some(event) = super::common::event_capture::captured_with(want) { + return event; + } + assert!( + std::time::Instant::now() < deadline, + "no decline event matching {want:?} within the row deadline" + ); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + } + + /// The OpenSSH client option set every row shares (batch mode, no + /// known-hosts persistence, bounded connect). + pub fn ssh_options() -> Vec { + vec![ + "-o".to_owned(), + "BatchMode=yes".to_owned(), + "-o".to_owned(), + "IdentitiesOnly=yes".to_owned(), + "-o".to_owned(), + "StrictHostKeyChecking=no".to_owned(), + "-o".to_owned(), + "UserKnownHostsFile=/dev/null".to_owned(), + "-o".to_owned(), + "ConnectTimeout=10".to_owned(), + ] + } + + /// The `GIT_SSH_COMMAND` transport line for the git rows (the + /// same option set, one string). + pub fn git_ssh_command(key: &Path) -> String { + format!( + "ssh -i {} -o BatchMode=yes -o IdentitiesOnly=yes \ + -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null \ + -o ConnectTimeout=10", + key.display() + ) + } + + /// The git 2.43+ availability gate (the wave-3 smoke's version + /// gate, shared by every git row). + pub fn git_available() -> (bool, String) { + let output = Command::new("git").arg("--version").output(); + let Ok(output) = output else { + return (false, "git is unavailable".to_owned()); + }; + let text = String::from_utf8_lossy(&output.stdout).to_string(); + let version = text + .split_whitespace() + .last() + .unwrap_or_default() + .split('.') + .filter_map(|c| c.parse::().ok()) + .collect::>(); + if version.first() == Some(&2) && version.get(1).is_some_and(|minor| *minor >= 43) { + (true, text.trim().to_owned()) + } else { + (false, format!("git {text:?} is older than 2.43.0")) + } + } + + /// The registered identity provider every row serves against: + /// fixture `alpha`'s fingerprint resolves to the row identity + /// (the ADR-004 decision 1 resolution shape). + pub fn provider(identity_id: &str) -> Arc { + let provider = Arc::new(MemIdentityProvider::new()); + provider.register( + fixture("alpha").fingerprint, + identity(identity_id, &["ssh"]), + ); + provider + } + + /// The in-process port allocator for the rows' named binds + /// (`-L`/`-D` client-side listeners, the `-R` fabric's establisher + /// bind): a monotonic counter over a window **below** the kernel's + /// ephemeral port range, so parallel rows can never collect the + /// same port and bind(0) allocations can never steal one — the + /// probe-bind re-verifies availability (a previous run's TIME_WAIT + /// tuples) and skips taken ports. + pub async fn pick_local_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, + } + } + } + + /// Mounts the door over a real TCP loopback listener over the + /// in-memory doubles (the wave-3 smoke's accept-loop shape): the + /// adapter constructs over the serve config, binds an ephemeral + /// port, and serves each accepted connection to completion on a + /// spawned task. The returned port is what the real tools dial. + pub async fn mount(serve: ServeConfig) -> (u16, tokio::task::JoinHandle<()>) { + let adapter = Arc::new( + SshAdapter::new(serve) + .await + .expect("the e2e adapter constructs"), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("the e2e listener binds"); + let port = listener.local_addr().expect("local addr").port(); + let accept_task = tokio::spawn(async move { + loop { + let Ok((stream, _)) = listener.accept().await else { + return; + }; + let adapter = Arc::clone(&adapter); + tokio::spawn(async move { + let _ = adapter.handle(stream).await; + }); + } + }); + (port, accept_task) + } +} + +/// The git rows — `git` build: clone (the wave-3 smoke re-landed) and +/// push (the POC-3 stage-B receive-pack gate), through the git +/// mapping over the hand-rolled recording quartet. +#[cfg(unix)] +#[cfg(feature = "git")] +mod git_rows { + use std::process::Output; + use std::sync::Arc; + use std::time::Duration; + + use alkssh::Limits; + + use super::common::git_doubles as gd; + use super::common::harness::serve_config as harness_serve_config; + use super::support; + + /// The unborn-HEAD advertisement every row serves (the wave-3 + /// smoke's repo record). + fn unborn_demo(doubles: &Arc) { + doubles.install(gd::RepoSetup::private_granted( + "owner/demo", + "git-rider", + vec![gd::RefLine { + name: "HEAD".into(), + oid: None, + peeled: None, + symref_target: Some("refs/heads/main".into()), + }], + )); + } + + async fn git_mount(doubles: &Arc) -> (u16, tokio::task::JoinHandle<()>) { + let provider = support::provider("git-rider"); + let mut serve = harness_serve_config( + provider, + &[super::common::fixture("bravo")], + Limits::default(), + ); + serve.git_seam = Some(gd::git_seam(doubles)); + support::mount(serve).await + } + + async fn run_git(args: &[&str]) -> Output { + tokio::process::Command::new("git") + .args(args) + .output() + .await + .expect("git spawns") + } + + async fn run_git_ssh(args: &[&str], ssh: &str) -> Output { + tokio::process::Command::new("git") + .args(args) + .env("GIT_SSH_COMMAND", ssh) + .output() + .await + .expect("git spawns") + } + + /// - *Contract stamp:* mapping.md §Channel-type rows (the git exec + /// row — `git clone ssh://…`; POC-3 §A.1 re-landed against the + /// shipped crate). + #[tokio::test] + async fn real_git_clones_the_empty_repo_through_the_door() { + let (git_ok, git_reason) = support::git_available(); + if !git_ok { + support::skip("git clone row", &git_reason); + return; + } + if !support::tool_present("ssh", &["-V"]) { + support::skip("git clone row", "ssh is unavailable"); + return; + } + + let key = support::write_key("git-clone"); + let work = key + .parent() + .expect("the workdir sits above the key") + .to_path_buf(); + + let doubles = Arc::new(gd::MemGit::new()); + unborn_demo(&doubles); + let (port, accept_task) = git_mount(&doubles).await; + + let ssh = support::git_ssh_command(&key); + let target = work.join("clone"); + let outcome = run_git_ssh( + &[ + "clone", + &format!("ssh://git-rider@127.0.0.1:{port}/owner/demo.git"), + target.to_str().expect("the clone target path"), + ], + &ssh, + ) + .await; + + accept_task.abort(); + + let stderr = String::from_utf8_lossy(&outcome.stderr).to_string(); + assert!( + outcome.status.success(), + "vanilla git clone (2.43+) through the door succeeds — stderr: {stderr}" + ); + assert!( + target.join(".git").is_dir(), + "the clone materializes a repository (the empty-repo clone: unborn HEAD)" + ); + assert!( + stderr.contains("empty repository"), + "the unborn-HEAD advertisement is what the clone consumed — stderr: {stderr}" + ); + assert!( + doubles.resolved_ids().iter().any(|id| id == "owner/demo"), + "the resolved id is the normalized one (the .git suffix rode off)" + ); + support::cleanup(&work); + } + + /// - *Contract stamp:* mapping.md §Channel-type rows (the git exec + /// row — `git push ssh://…` = receive-pack through the real + /// client; POC-3 §A stage-B gate re-landed, the wave-3 smoke's + /// clone-only coverage completed). + #[tokio::test] + async fn real_git_push_creates_the_ref_through_the_door() { + let (git_ok, git_reason) = support::git_available(); + if !git_ok { + support::skip("git push row", &git_reason); + return; + } + if !support::tool_present("ssh", &["-V"]) { + support::skip("git push row", "ssh is unavailable"); + return; + } + + let key = support::write_key("git-push"); + let work = key + .parent() + .expect("the workdir sits above the key") + .to_path_buf(); + + let doubles = Arc::new(gd::MemGit::new()); + unborn_demo(&doubles); + let (port, accept_task) = git_mount(&doubles).await; + + let repo = work.join("local"); + let init = run_git(&[ + "-c", + "init.defaultBranch=main", + "init", + repo.to_str().expect("the local repo path"), + ]) + .await; + assert!( + init.status.success(), + "the local repo initializes — stderr: {}", + String::from_utf8_lossy(&init.stderr) + ); + std::fs::write(repo.join("hello.txt"), "alkssh e2e push\n") + .expect("the commit file writes"); + let add = run_git(&["-C", repo.to_str().expect("path"), "add", "-A"]).await; + assert!(add.status.success(), "the staged tree adds"); + let commit = run_git(&[ + "-C", + repo.to_str().expect("path"), + "-c", + "user.name=e2e-rider", + "-c", + "user.email=e2e@alk.dev", + "commit", + "-m", + "alkssh e2e push row", + ]) + .await; + assert!( + commit.status.success(), + "the commit lands — stderr: {}", + String::from_utf8_lossy(&commit.stderr) + ); + let head = run_git(&["-C", repo.to_str().expect("path"), "rev-parse", "HEAD"]).await; + assert!(head.status.success(), "rev-parse HEAD"); + let head_hex = String::from_utf8_lossy(&head.stdout).trim().to_owned(); + + let ssh = support::git_ssh_command(&key); + let push = tokio::time::timeout( + Duration::from_secs(30), + run_git_ssh( + &[ + "-C", + repo.to_str().expect("path"), + "push", + &format!("ssh://git-rider@127.0.0.1:{port}/owner/demo.git"), + "main", + ], + &ssh, + ), + ) + .await; + + accept_task.abort(); + + let outcome = match push { + Ok(outcome) => outcome, + Err(_) => panic!("the real git push row exceeded its 30 s deadline"), + }; + let stderr = String::from_utf8_lossy(&outcome.stderr).to_string(); + assert!( + outcome.status.success(), + "vanilla git push (receive-pack) through the door succeeds — stderr: {stderr}" + ); + + let prepared = doubles.prepared(); + let update = prepared + .first() + .and_then(|pack| pack.updates.iter().find(|u| u.name == "refs/heads/main")) + .expect("the push prepared one receive-pack carrying refs/heads/main"); + let new = update.new.expect("a branch push carries the new ref oid"); + assert_eq!( + format!("{new}"), + head_hex, + "the ref the door's receive-pack applied is the pushed commit (rev parity)" + ); + assert!( + doubles + .applied() + .iter() + .any(|applied| applied.updates.iter().any(|u| u.name == "refs/heads/main")), + "the CAS application rides apply_updates (the wire layer's single ingestion home)" + ); + support::cleanup(&work); + } +} + +/// The session rows — `session` build: the interactive pty/shell +/// session (the wave-3 smoke re-landed, extended to the echo marker) +/// and the plain pipe-exec row with its separate stderr. +#[cfg(unix)] +#[cfg(feature = "session")] +mod session_rows { + use std::collections::HashMap; + use std::sync::Arc; + use std::time::Duration; + + use alkssh::{Limits, SessionSeam}; + use alktty::LocalTtyBackend; + use alktty::backend::TtyBackend; + + use super::common::harness::serve_config as harness_serve_config; + use super::support; + + /// The `local` composition for the session rows — alktty's own + /// backend in the seam map (dev-dep unification: the `local` + /// feature rides the test graph only). + fn local_seam() -> SessionSeam { + let mut backends = HashMap::new(); + backends.insert( + "local".to_owned(), + Arc::new(LocalTtyBackend::new()) as Arc, + ); + SessionSeam { + backends: Arc::new(backends), + backend: "local".to_owned(), + ownership: None, + } + } + + /// - *Contract stamp:* mapping.md §Channel-type rows (the + /// `session` → `pty_request` + `shell_request` row; POC-3 §D.1 + /// re-landed as a full OpenSSH interactive session). + #[tokio::test] + async fn real_openssh_pty_shell_session_serves_through_the_door() { + if !support::tool_present("ssh", &["-V"]) { + support::skip("pty/shell row", "ssh is unavailable"); + return; + } + if !support::tool_present("sh", &["-c", "exit 0"]) { + support::skip("pty/shell row", "sh is unavailable"); + return; + } + + let key = support::write_key("pty-shell"); + let work = key + .parent() + .expect("the workdir sits above the key") + .to_path_buf(); + + let mut serve = harness_serve_config( + support::provider("e2e-rider"), + &[super::common::fixture("bravo")], + Limits::default(), + ); + serve.shell_argv = Some(vec!["/bin/sh".to_owned(), "-i".to_owned()]); + serve.session_seam = Some(local_seam()); + let (port, accept_task) = support::mount(serve).await; + + let outcome = tokio::time::timeout(Duration::from_secs(30), async { + let mut child = tokio::process::Command::new("ssh") + .args(["-tt", "-p", &port.to_string(), "-i"]) + .arg(&key) + .args(support::ssh_options()) + .arg("e2e-rider@127.0.0.1") + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .spawn() + .expect("ssh spawns"); + { + use tokio::io::AsyncWriteExt; + let mut stdin = child.stdin.take().expect("the child stdin is piped"); + stdin + .write_all(b"echo alkssh-e2e-pty; exit\n") + .await + .expect("the row stdin writes"); + stdin.flush().await.expect("the row stdin flushes"); + } + child.wait_with_output().await.expect("ssh completes") + }) + .await; + + accept_task.abort(); + + let output = match outcome { + Ok(output) => output, + Err(_) => panic!("the pty/shell row exceeded its 30 s deadline"), + }; + let stdout = String::from_utf8_lossy(&output.stdout).to_string(); + assert!( + output.status.success(), + "vanilla OpenSSH's pty+shell session through the door exits 0 \ + (the exit chunk derived exit-status(0) before close) — stdout: {stdout:?}" + ); + assert!( + stdout.contains("alkssh-e2e-pty"), + "the keystrokes and the shell's echo rode the real PTY round trip — stdout: {stdout:?}" + ); + support::cleanup(&work); + } + + /// - *Contract stamp:* mapping.md §Request-level rows (the + /// `exec_request` pipe row — `wrapper ++ [cmd]`, stderr rides + /// `extended_data(1)`; POC-3 §D.1's pipe split, end to end). + #[tokio::test] + async fn real_openssh_pipe_exec_separates_stdout_and_stderr_through_the_door() { + if !support::tool_present("ssh", &["-V"]) { + support::skip("pipe exec row", "ssh is unavailable"); + return; + } + if !support::tool_present("sh", &["-c", "exit 0"]) { + support::skip("pipe exec row", "sh is unavailable"); + return; + } + + let key = support::write_key("pipe-exec"); + let work = key + .parent() + .expect("the workdir sits above the key") + .to_path_buf(); + + let mut serve = harness_serve_config( + support::provider("e2e-rider"), + &[super::common::fixture("bravo")], + Limits::default(), + ); + serve.exec_wrapper_argv = Some(vec!["/bin/sh".to_owned(), "-c".to_owned()]); + serve.session_seam = Some(local_seam()); + let (port, accept_task) = support::mount(serve).await; + + let outcome = tokio::time::timeout(Duration::from_secs(30), async { + tokio::process::Command::new("ssh") + .args(["-p", &port.to_string(), "-i"]) + .arg(&key) + .args(support::ssh_options()) + .arg("e2e-rider@127.0.0.1") + .arg("echo alkssh-e2e-stdout; echo alkssh-e2e-stderr >&2") + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .output() + .await + .expect("ssh spawns and completes") + }) + .await; + + accept_task.abort(); + + let output = match outcome { + Ok(output) => output, + Err(_) => panic!("the pipe-exec row exceeded its 30 s deadline"), + }; + let stdout = String::from_utf8_lossy(&output.stdout).to_string(); + let stderr = String::from_utf8_lossy(&output.stderr).to_string(); + assert!( + output.status.success(), + "vanilla `ssh host \"cmd\"` through the exec wrapper exits 0 — stdout: {stdout:?} stderr: {stderr:?}" + ); + assert!( + stdout.contains("alkssh-e2e-stdout"), + "the pipe row's stdout rides SSH data — stdout: {stdout:?}" + ); + assert!( + stderr.contains("alkssh-e2e-stderr"), + "the pipe row's stderr rides extended_data(1) separately — stderr: {stderr:?}" + ); + support::cleanup(&work); + } +} + +/// The tunnel rows — `tunnel` build: `-L` (the route-1 bridge), `-R` +/// (listen + arrival + data leg through the leg-2 fabric), and `-D` +/// (curl's SOCKS traffic takes the `direct-tcpip` leg — POC-3 +/// correction 2's posture). +#[cfg(unix)] +#[cfg(feature = "tunnel")] +mod tunnel_rows { + use std::sync::Arc; + use std::time::Duration; + + use alkssh::{Limits, ServeConfig, TunnelListenSeam}; + + use super::common::harness::serve_config as harness_serve_config; + use super::common::{fixture, identity}; + use super::support; + + /// The in-file fabric fixtures — the trimmed leg-1/leg-2 fabric + /// double (the wave-3 smoke's recorded composition, POC-3 §C.4 + /// recipe), TCP-only and happy-path only: the decline/collision + /// machinery stays the mapping binaries'. + #[cfg(unix)] + mod fabric { + #![allow(dead_code)] + + use std::sync::Arc; + use std::sync::atomic::{AtomicU32, Ordering}; + + use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero}; + use alkcall::channels::client::{ChannelClient, ServingConfig}; + use alkcall::channels::operations::{ + ChannelCore, ChannelOperations, Establishment, OpenEstablisher, + }; + use alkcall::core::auth::{AuthContext, Identity, NoopIdentityProvider}; + use alkcall::core::types::{Connection as CoreConnection, ProtocolHandler}; + use alkcall::protocol::connection::{CallConnection, split_single_stream}; + use alkcall::protocol::dispatch::Dispatcher; + use alkcall::registry::registration::OperationRegistry; + use alktunnels::params::{ + OP_TUNNEL_DIRECT, OP_TUNNEL_OPEN, Substrate, SubstrateAddr, TUNNEL_FORWARDED_SCOPE, + TUNNEL_OPEN_SCOPE, + }; + use alktunnels::producer::{ + BindFirstAcceptLoop, BindFirstPlan, BindFn, BoundListener, DialFn, ForwardedOpener, + IdentityWitness, TargetHandle, TunnelEstablishError, make_bindfirst_accept_handler, + register_tunnel_direct_openable, + }; + + use crate::common::identity; + + /// The registration settle: the fabric's openables register on + /// a spawned task, so a row that drives a client before the + /// registrations land races (a listen dial before + /// `channels/tunnel/sub` is registered is refused, not + /// retried). Every fabric-consumer waits for the named ops' + /// registrations with a bounded poll. + async fn settle(registry: &Arc, ops: &[&str]) { + for _ in 0..6000 { + if ops.iter().all(|op| registry.registration(op).is_some()) { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + panic!("the fabric never registered {ops:?}"); + } + + /// The dial seam: a real-TCP dial closure (the assembly's + /// injected `DialFn` for real; alktunnels' `local` feature + /// stays off). + pub(super) fn tcp_dial() -> DialFn { + Arc::new(move |_substrate: Substrate, target: &str| { + let target = target.to_owned(); + Box::pin(async move { + let stream = tokio::net::TcpStream::connect(&target).await.map_err(|e| { + TunnelEstablishError::DialFailed(format!("dial `{target}`: {e}")) + })?; + let (read, write) = stream.into_split(); + Ok(TargetHandle { + read: Box::new(read), + write: Box::new(write), + }) + }) + }) + } + + /// The leg-1 fabric (`w3-tunnel-direct`'s recorded double): an + /// `alk/channels` connection whose accept side registers + /// `ChannelOperations` + the `channels/tunnel/direct` openable + /// (the injected dial seam), and whose connect side is the + /// `Arc` a tunnel-featured serve-config holds. + pub(super) async fn wire_fabric( + dial: DialFn, + caller: Identity, + witness: IdentityWitness, + ) -> Arc { + let registry = Arc::new(OperationRegistry::new()); + let registry_for_hook = Arc::clone(®istry); + let caller_for_hook = caller.clone(); + let dial_for_hook = Arc::clone(&dial); + let witness_for_hook = Some(witness); + + let install_hook: InstallChannelZero = + Arc::new(move |manager, channel0_conn, _auth| { + let registry = Arc::clone(®istry_for_hook); + let caller = caller_for_hook.clone(); + let dial = Arc::clone(&dial_for_hook); + let witness = witness_for_hook.clone(); + tokio::spawn(async move { + let Ok(channel0_bidi) = channel0_conn.accept_bi().await else { + return; + }; + let (writer, reader) = split_single_stream(channel0_bidi); + let _ = channel0_conn.set_identity(caller); + let call_connection = Arc::new(CallConnection::new_single_stream( + channel0_conn, + Arc::clone(&writer), + )); + let ops = ChannelOperations::with_default_policy(manager.clone()); + ops.register_on(®istry).expect("channel ops register"); + register_tunnel_direct_openable( + &ChannelCore::new( + manager.clone(), + alkcall::channels::policy::default_policy(), + ), + ®istry, + AuthContext::anonymous(b"alk/tunnel"), + dial, + witness, + ) + .expect("tunnel direct openable registers"); + let dp = Dispatcher::new(registry, Arc::new(NoopIdentityProvider)); + dp.serve_single_stream(call_connection, reader, writer) + .await; + }) + }); + + let (consumer_end, producer_end) = tokio::io::duplex(64 * 1024); + let consumer_conn = + CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None); + let producer_conn = + CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None); + + let adapter = + ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap)); + tokio::spawn(async move { + let _ = ProtocolHandler::handle( + &adapter, + producer_conn, + &AuthContext::anonymous(b"alk/channels"), + ) + .await; + }); + + settle(®istry, &[OP_TUNNEL_DIRECT]).await; + Arc::new( + ChannelClient::from_connection(consumer_conn) + .await + .expect("the fabric's channels client establishes"), + ) + } + + /// The e2e fabric double for the `-R` row: the leg-1 shape + /// extended with the **bind-first listen half** — the + /// establisher binds the wire-named endpoint for real (the + /// dynamic-address producer), mints the correlation token per + /// the ADR-008 discipline, and the accept handler dials the + /// `tunnel/forwarded` op at the door per arrival. The door + /// side rides the `ServingConfig.registry` the row hands + /// alkssh through the `TunnelListenSeam` (one `Arc`). + pub(super) struct FabricDouble { + pub(super) client: Arc, + pub(super) registry: Arc, + } + + fn real_tcp_bind() -> BindFn { + Arc::new(move |_substrate, endpoint: &str| { + let endpoint = endpoint.to_owned(); + let endpoint = endpoint.replace("localhost:", "127.0.0.1:"); + Box::pin(async move { + let listener = tokio::net::TcpListener::bind(&endpoint) + .await + .map_err(|e| { + TunnelEstablishError::ResourceShortage(format!( + "bind `{endpoint}`: {e}" + )) + })?; + let addr = listener.local_addr().expect("bound addr"); + Ok(BoundListener { + listener: Arc::new(listener), + bound: SubstrateAddr::HostPort { + host: addr.ip().to_string(), + port: addr.port(), + }, + }) + }) + }) + } + + fn real_accept_loop() -> BindFirstAcceptLoop { + Arc::new(move |listener: Arc| { + Box::pin(async move { + let tcp = listener.downcast_ref::().ok_or( + TunnelEstablishError::HandlerError( + "the plan's listener is not the double's bind".into(), + ), + )?; + let (stream, peer) = tcp + .accept() + .await + .map_err(|e| TunnelEstablishError::DialFailed(format!("accept: {e}")))?; + let (read, write) = stream.into_split(); + Ok(( + SubstrateAddr::HostPort { + host: peer.ip().to_string(), + port: peer.port(), + }, + TargetHandle { + read: Box::new(read), + write: Box::new(write), + }, + )) + }) + }) + } + + fn counter_mint() -> OpenEstablisher { + let counter = Arc::new(AtomicU32::new(1)); + Arc::new(move |input: serde_json::Value, _auth| { + let counter = Arc::clone(&counter); + Box::pin(async move { + let params: alktunnels::params::TunnelParams = + serde_json::from_value(input.clone()).map_err(|e| { + TunnelEstablishError::HandlerError(format!("invalid params: {e}")) + })?; + let next = counter.fetch_add(1, Ordering::Relaxed); + let token = if next == 0 { 1 } else { next }; + let BoundListener { listener, bound } = + real_tcp_bind()(params.substrate, ¶ms.resource).await?; + let plan = BindFirstPlan { + listener, + token, + substrate: params.substrate, + }; + Ok(Establishment::new(Arc::new(plan)) + .with_reply_field( + "bound", + serde_json::to_value(&bound) + .map_err(|e| TunnelEstablishError::HandlerError(format!("{e}")))?, + ) + .with_reply_field("listen", serde_json::json!(token))) + }) + }) + } + + pub(super) async fn wire_fabric_with_listen(dial: DialFn) -> FabricDouble { + let door_registry = Arc::new(OperationRegistry::new()); + let registry = Arc::new(OperationRegistry::new()); + let registry_for_hook = Arc::clone(®istry); + let door_caller = identity("door-caller", &[TUNNEL_OPEN_SCOPE]); + let arrival_identity = identity("fabric-arrival", &[TUNNEL_FORWARDED_SCOPE]); + + let install_hook: InstallChannelZero = + Arc::new(move |manager, channel0_conn, _auth| { + let registry = Arc::clone(®istry_for_hook); + let door_caller = door_caller.clone(); + let dial = Arc::clone(&dial); + tokio::spawn(async move { + let Ok(channel0_bidi) = channel0_conn.accept_bi().await else { + return; + }; + let (writer, reader) = split_single_stream(channel0_bidi); + let _ = channel0_conn.set_identity(door_caller); + let call_connection = Arc::new(CallConnection::new_single_stream( + channel0_conn, + Arc::clone(&writer), + )); + + let open_forwarded: ForwardedOpener = { + let call = Arc::clone(&call_connection); + let manager = manager.clone(); + Arc::new(move |params: alktunnels::params::TunnelForwardedParams| { + let call = Arc::clone(&call); + let manager = manager.clone(); + Box::pin(async move { + let payload = serde_json::to_value(¶ms).map_err(|e| { + TunnelEstablishError::HandlerError(format!("{e}")) + })?; + let response = call + .call(alktunnels::params::OP_TUNNEL_FORWARDED, payload) + .await; + let out = response.result.map_err(|e| { + TunnelEstablishError::HandlerError(format!( + "forwarded open: {e:?}" + )) + })?; + let id = out + .get("channel_id") + .and_then(|v| v.as_u64()) + .ok_or_else(|| { + TunnelEstablishError::HandlerError( + "no channel_id".into(), + ) + })? as u32; + manager + .adopt_channel(id, alktunnels::params::TUNNEL_ALPN, None) + .await + .map_err(|e| { + TunnelEstablishError::HandlerError(format!( + "adopt: {e:?}" + )) + }) + }) + }) + }; + + let ops = ChannelOperations::with_default_policy(manager.clone()); + ops.register_on(®istry).expect("channel ops register"); + let direct_core = ChannelCore::new( + manager.clone(), + alkcall::channels::policy::default_policy(), + ); + register_tunnel_direct_openable( + &direct_core, + ®istry, + AuthContext::anonymous(b"alk/tunnel"), + dial, + None, + ) + .expect("the direct openable registers"); + direct_core + .register_openable_with_establisher( + alktunnels::params::tunnel_open_spec(), + Some(counter_mint()), + make_bindfirst_accept_handler(real_accept_loop(), open_forwarded), + ®istry, + AuthContext::anonymous(b"alk/tunnel"), + None, + ) + .expect("the bind-first listen openable registers"); + + let dp = Dispatcher::new(registry, Arc::new(NoopIdentityProvider)); + dp.serve_single_stream(call_connection, reader, writer) + .await; + }) + }); + + let (consumer_end, producer_end) = tokio::io::duplex(64 * 1024); + let consumer_conn = + CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None); + let producer_conn = + CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None); + + let adapter = + ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap)); + tokio::spawn(async move { + let _ = ProtocolHandler::handle( + &adapter, + producer_conn, + &AuthContext::anonymous(b"alk/channels"), + ) + .await; + }); + + let client = ChannelClient::from_connection_with_serving( + consumer_conn, + Some(ServingConfig { + registry: Arc::clone(&door_registry), + identity_provider: Arc::new(NoopIdentityProvider), + identity: Some(arrival_identity), + }), + ) + .await + .expect("the fabric's channels client establishes"); + + settle(®istry, &[OP_TUNNEL_DIRECT, OP_TUNNEL_OPEN]).await; + + FabricDouble { + client: Arc::new(client), + registry: door_registry, + } + } + } + + /// A tiny HTTP/1.1 endpoint (the POC's shape) — the substrate the + /// forwarded traffic dials through the fabric. + async fn spawn_http_endpoint() -> (std::net::SocketAddr, String) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("the row endpoint binds"); + let addr = listener.local_addr().expect("local addr"); + let body = "alkssh w5-e2e-server through the fabric\n".to_owned(); + let body2 = body.clone(); + tokio::spawn(async move { + loop { + let Ok((mut conn, _)) = listener.accept().await else { + return; + }; + let body = body2.clone(); + tokio::spawn(async move { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let mut buf = vec![0u8; 8192]; + let _ = conn.read(&mut buf).await; + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + body + ); + let _ = conn.write_all(head.as_bytes()).await; + let _ = conn.shutdown().await; + }); + } + }); + (addr, body) + } + + /// The curl-until-settled loop (the -L listener binds client-side; + /// the -R fabric binds after the listen flow): a bounded settle + /// window, the first successful body wins. + async fn curl_until_success(url: &str, extra: &[&str]) -> std::process::Output { + let mut attempts = 0u32; + loop { + let out = tokio::process::Command::new("curl") + .args(["-s", "--max-time", "10"]) + .args(extra) + .arg(url) + .output() + .await + .expect("curl spawns"); + if out.status.success() { + return out; + } + attempts += 1; + assert!( + attempts < 20, + "curl never got through: exit {:?}", + out.status.code() + ); + tokio::time::sleep(Duration::from_millis(250)).await; + } + } + + async fn tunnel_mount(serve: ServeConfig) -> (u16, tokio::task::JoinHandle<()>) { + support::mount(serve).await + } + + fn key_arg(row: &str) -> (std::path::PathBuf, std::path::PathBuf) { + let key = support::write_key(row); + let work = key + .parent() + .expect("the workdir sits above the key") + .to_path_buf(); + (work, key) + } + + fn ssh_forward_args( + port: u16, + key: &std::path::Path, + forward: &[&str], + ) -> tokio::process::Command { + let mut cmd = tokio::process::Command::new("ssh"); + cmd.arg("-N") + .arg("-i") + .arg(key) + .arg("-p") + .arg(port.to_string()) + .args(support::ssh_options()) + .args(["-o", "ExitOnForwardFailure=yes"]) + .args(forward) + .arg("e2e-rider@127.0.0.1") + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::piped()); + cmd + } + + /// - *Contract stamp:* mapping.md §Channel-type rows (the + /// `direct-tcpip` row — `ssh -L` through the route-1 bridge; + /// POC-3 §C.1 re-landed with the shipped crate). + #[tokio::test] + async fn real_ssh_local_forward_serves_the_direct_tcpip_leg() { + if !support::tool_present("ssh", &["-V"]) { + support::skip("-L row", "ssh is unavailable"); + return; + } + if !support::tool_present("curl", &["--version"]) { + support::skip("-L row", "curl is unavailable"); + return; + } + + let (http_addr, body) = spawn_http_endpoint().await; + let caller = identity("tunnel-caller", &[alktunnels::params::TUNNEL_DIRECT_SCOPE]); + let fabric = fabric::wire_fabric( + fabric::tcp_dial(), + caller, + Arc::new(tokio::sync::Mutex::new(None)), + ) + .await; + + let (work, key) = key_arg("tunnel-l"); + let mut serve = harness_serve_config( + support::provider("tunnel-rider"), + &[fixture("bravo")], + Limits::default(), + ); + serve.channel_client = Some(fabric); + let (ssh_port, accept_task) = tunnel_mount(serve).await; + + let fwd_port = support::pick_local_port().await; + let mut ssh_child = ssh_forward_args( + ssh_port, + &key, + &[ + "-L", + &format!( + "127.0.0.1:{fwd_port}:{}:{}", + http_addr.ip(), + http_addr.port() + ), + ], + ) + .spawn() + .expect("vanilla ssh spawns"); + + let through = tokio::time::timeout(Duration::from_secs(30), async { + curl_until_success(&format!("http://127.0.0.1:{fwd_port}/"), &[]).await + }) + .await; + + let _ = ssh_child.kill().await; + accept_task.abort(); + + let out = through.expect("the -L row resolves in time"); + let stdout = String::from_utf8_lossy(&out.stdout).to_string(); + assert!( + stdout.trim_end() == body.trim_end(), + "vanilla ssh -L + curl through the door → fabric → endpoint returns the body — got {stdout:?}" + ); + support::cleanup(&work); + } + + /// - *Contract stamp:* mapping.md §Global-request rows (the `-R` + /// listen flow — `ssh -R`'s listen, the fabric-bound port, the + /// server-initiated `forwarded-tcpip` arrival, the client-dialed + /// data leg; the `w3-tunnel-listen` flow re-landed against a + /// real OpenSSH client). + #[tokio::test] + async fn real_ssh_remote_forward_serves_the_arrival_and_data_legs() { + use alktunnels::params::Substrate; + + if !support::tool_present("ssh", &["-V"]) { + support::skip("-R row", "ssh is unavailable"); + return; + } + if !support::tool_present("curl", &["--version"]) { + support::skip("-R row", "curl is unavailable"); + return; + } + + let (http_addr, body) = spawn_http_endpoint().await; + let fail_dial: alktunnels::producer::DialFn = + Arc::new(move |_substrate: Substrate, target: &str| { + let target = target.to_owned(); + Box::pin(async move { + Err::< + alktunnels::producer::TargetHandle, + alktunnels::producer::TunnelEstablishError, + >(alktunnels::producer::TunnelEstablishError::DialFailed( + format!("the leg-1 dial double is not part of the -R row: {target}"), + )) + }) + }); + let double = fabric::wire_fabric_with_listen(fail_dial).await; + + let (work, key) = key_arg("tunnel-r"); + let mut serve = harness_serve_config( + support::provider("tunnel-rider"), + &[fixture("bravo")], + Limits::default(), + ); + serve.channel_client = Some(Arc::clone(&double.client)); + serve.tunnel_listen_seam = Some(TunnelListenSeam::new(Arc::clone(&double.registry))); + let (ssh_port, accept_task) = tunnel_mount(serve).await; + + let fwd_port = support::pick_local_port().await; + let mut ssh_child = ssh_forward_args( + ssh_port, + &key, + &[ + "-R", + &format!( + "127.0.0.1:{fwd_port}:{}:{}", + http_addr.ip(), + http_addr.port() + ), + ], + ) + .spawn() + .expect("vanilla ssh spawns"); + + let through = tokio::time::timeout(Duration::from_secs(30), async { + curl_until_success(&format!("http://127.0.0.1:{fwd_port}/"), &[]).await + }) + .await; + + let _ = ssh_child.kill().await; + accept_task.abort(); + + let out = through.expect("the -R row resolves in time"); + let stdout = String::from_utf8_lossy(&out.stdout).to_string(); + assert!( + stdout.trim_end() == body.trim_end(), + "vanilla ssh -R + curl: fabric arrival → forwarded-tcpip → client-dialed endpoint returns \ + the body — got {stdout:?}" + ); + support::cleanup(&work); + } + + /// - *Contract stamp:* mapping.md §Channel-type rows (`direct-tcpip` + /// serves **all** `ssh -D` traffic per POC-3 correction 2 — the + /// SOCKS handshake happens client-side in OpenSSH; POC-3 §B.2's + /// corrected posture re-landed with real curl). + #[tokio::test] + async fn real_ssh_dynamic_forward_serves_curl_through_the_direct_tcpip_leg() { + if !support::tool_present("ssh", &["-V"]) { + support::skip("-D row", "ssh is unavailable"); + return; + } + if !support::tool_present("curl", &["--version"]) { + support::skip("-D row", "curl is unavailable"); + return; + } + + let (http_addr, body) = spawn_http_endpoint().await; + let caller = identity("tunnel-caller", &[alktunnels::params::TUNNEL_DIRECT_SCOPE]); + let fabric = fabric::wire_fabric( + fabric::tcp_dial(), + caller, + Arc::new(tokio::sync::Mutex::new(None)), + ) + .await; + + let (work, key) = key_arg("tunnel-d"); + let mut serve = harness_serve_config( + support::provider("tunnel-rider"), + &[fixture("bravo")], + Limits::default(), + ); + serve.channel_client = Some(fabric); + let (ssh_port, accept_task) = tunnel_mount(serve).await; + + let socks_port = support::pick_local_port().await; + let mut ssh_child = + ssh_forward_args(ssh_port, &key, &["-D", &format!("127.0.0.1:{socks_port}")]) + .spawn() + .expect("vanilla ssh spawns"); + + let through = tokio::time::timeout(Duration::from_secs(30), async { + let first = curl_until_success( + &format!("http://{}:{}/", http_addr.ip(), http_addr.port()), + &["--socks5-hostname", &format!("127.0.0.1:{socks_port}")], + ) + .await; + let second = curl_until_success( + &format!("http://{}:{}/", http_addr.ip(), http_addr.port()), + &["--socks5-hostname", &format!("127.0.0.1:{socks_port}")], + ) + .await; + (first, second) + }) + .await; + + let _ = ssh_child.kill().await; + accept_task.abort(); + + let (first, second) = through.expect("the -D row resolves in time"); + for out in [first, second] { + let stdout = String::from_utf8_lossy(&out.stdout).to_string(); + assert!( + stdout.trim_end() == body.trim_end(), + "curl --socks5-hostname through vanilla ssh -D rides the door's direct-tcpip leg — \ + got {stdout:?}" + ); + } + support::cleanup(&work); + } +} + +/// The decline-observability row where the `session` feature is off — +/// a real `ssh` pty attempt meets a clean, channel-scoped decline +/// (auth intact, no hang), with the door's decline event on record. +#[cfg(unix)] +#[cfg(not(feature = "session"))] +mod unfeatured_session_decline { + use std::time::Duration; + + use alkssh::Limits; + + use super::common::event_capture::{expected_build, install}; + use super::common::fixture; + use super::common::harness::serve_config as harness_serve_config; + use super::support; + + /// - *Contract stamp:* mapping.md §Decline matrix (none/git/tunnel + /// builds decline session rows) + POC-1 §5 (decline + /// observability: channel-scoped decline, auth intact, no hangs). + #[tokio::test] + async fn real_openssh_pty_attempt_declines_cleanly_in_a_feature_limited_build() { + if !support::tool_present("ssh", &["-V"]) { + support::skip("pty decline row", "ssh is unavailable"); + return; + } + + install(); + let key = support::write_key("pty-decline"); + let work = key + .parent() + .expect("the workdir sits above the key") + .to_path_buf(); + + let serve = harness_serve_config( + support::provider("decline-rider"), + &[fixture("bravo")], + Limits::default(), + ); + let (port, accept_task) = support::mount(serve).await; + + let outcome = tokio::time::timeout(Duration::from_secs(30), async { + tokio::process::Command::new("ssh") + .args(["-tt", "-p", &port.to_string(), "-i"]) + .arg(&key) + .args(support::ssh_options()) + .arg("decline-rider@127.0.0.1") + .arg("echo unreachable") + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .output() + .await + .expect("ssh spawns") + }) + .await; + + accept_task.abort(); + + let output = match outcome { + Ok(output) => output, + Err(_) => panic!("the pty decline row hung past its 30 s deadline"), + }; + let stderr = String::from_utf8_lossy(&output.stderr).to_string(); + assert!( + !output.status.success(), + "the unfeatured pty attempt surfaces a failure to the real client — stderr: {stderr:?}" + ); + assert!( + !stderr.trim().is_empty(), + "the decline surfaces a clean diagnostic — stderr: {stderr:?}" + ); + if cfg!(feature = "git") { + let event = support::decline_event(&[ + ("row", "pty_request"), + ("message", "decline: channel request rejected"), + ("feature_build", expected_build()), + ]) + .await; + assert_eq!(event.field("decline_class"), Some("request")); + assert_eq!(event.field("reason"), Some("unfeatured")); + } else { + let event = support::decline_event(&[ + ("row", "session"), + ("message", "decline: channel open rejected"), + ("feature_build", expected_build()), + ]) + .await; + assert_eq!(event.field("decline_class"), Some("open")); + assert_eq!(event.field("reason"), Some("unfeatured")); + } + support::cleanup(&work); + } +} + +/// The decline-observability rows where the `tunnel` feature is off — +/// a real `ssh -R` dies at the listen gate with a clean diagnostic; +/// a real `ssh -L`'s curls fail per connection without hanging the +/// client; the door's decline events hold for both. +#[cfg(unix)] +#[cfg(not(feature = "tunnel"))] +mod unfeatured_tunnel_declines { + use std::time::Duration; + + use alkssh::Limits; + + use super::common::event_capture::{expected_build, install}; + use super::common::fixture; + use super::common::harness::serve_config as harness_serve_config; + use super::support; + + async fn plain_mount() -> (u16, tokio::task::JoinHandle<()>) { + let serve = harness_serve_config( + support::provider("decline-rider"), + &[fixture("bravo")], + Limits::default(), + ); + support::mount(serve).await + } + + fn spawn_forward(forward: &[&str], port: u16, key: &std::path::Path) -> tokio::process::Child { + tokio::process::Command::new("ssh") + .arg("-N") + .arg("-i") + .arg(key) + .arg("-p") + .arg(port.to_string()) + .args(support::ssh_options()) + .args(forward) + .arg("decline-rider@127.0.0.1") + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::piped()) + .spawn() + .expect("vanilla ssh spawns") + } + + /// - *Contract stamp:* mapping.md §Global-request rows (the gate: + /// unfeatured `tcpip_forward` declines — POC-3 §C.2's refusal + /// shape re-landed with a real OpenSSH client and + /// `ExitOnForwardFailure`). The -R row is gated here + /// (check-gated, skipped-with-reason, not silently green). + #[tokio::test] + async fn real_openssh_remote_forward_attempt_declines_cleanly() { + if !support::tool_present("ssh", &["-V"]) { + support::skip("-R decline row", "ssh is unavailable"); + return; + } + + install(); + let key = support::write_key("rfwd-decline"); + let work = key + .parent() + .expect("the workdir sits above the key") + .to_path_buf(); + + let (port, accept_task) = plain_mount().await; + let listen_port = support::pick_local_port().await; + + let outcome = tokio::time::timeout(Duration::from_secs(30), async { + let child = spawn_forward( + &[ + "-o", + "ExitOnForwardFailure=yes", + "-R", + &format!("127.0.0.1:{listen_port}:127.0.0.1:9365"), + ], + port, + &key, + ); + child.wait_with_output().await.expect("ssh completes") + }) + .await; + + accept_task.abort(); + + let output = match outcome { + Ok(output) => output, + Err(_) => panic!("the -R decline row hung past its 30 s deadline"), + }; + let stderr = String::from_utf8_lossy(&output.stderr).to_string(); + assert!( + !output.status.success(), + "the declined listen surfaces a failure to the real client — stderr: {stderr:?}" + ); + assert!( + stderr.contains("forwarding failed"), + "OpenSSH's remote-forward-failed diagnostic is on record — stderr: {stderr:?}" + ); + support::decline_event(&[ + ("row", "tcpip_forward"), + ("message", "decline: global request rejected"), + ("feature_build", expected_build()), + ]) + .await; + support::cleanup(&work); + } + + /// - *Contract stamp:* mapping.md §Channel-type rows (the + /// `direct-tcpip` open gate — unfeatured declines; POC-1 §5: + /// the real client observes a clean per-connection failure, + /// no hang). + #[tokio::test] + async fn real_openssh_local_forward_connections_decline_per_connection() { + if !support::tool_present("ssh", &["-V"]) { + support::skip("-L decline row", "ssh is unavailable"); + return; + } + if !support::tool_present("curl", &["--version"]) { + support::skip("-L decline row", "curl is unavailable"); + return; + } + + install(); + let key = support::write_key("lfwd-decline"); + let work = key + .parent() + .expect("the workdir sits above the key") + .to_path_buf(); + + let (port, accept_task) = plain_mount().await; + let fwd_port = support::pick_local_port().await; + + let mut ssh_child = spawn_forward( + &["-L", &format!("127.0.0.1:{fwd_port}:127.0.0.1:9365")], + port, + &key, + ); + let want: [(&str, &str); 3] = [ + ("row", "direct-tcpip"), + ("message", "decline: channel open rejected"), + ("feature_build", expected_build()), + ]; + let outcome = tokio::time::timeout(Duration::from_secs(30), async { + let deadline = std::time::Instant::now() + Duration::from_secs(28); + loop { + let out = tokio::process::Command::new("curl") + .args(["-s", "--max-time", "5"]) + .arg(format!("http://127.0.0.1:{fwd_port}/")) + .output() + .await + .expect("curl spawns"); + assert!( + !out.status.success(), + "the declined direct-tcpip open must fail the forward, got exit {:?}", + out.status.code() + ); + match super::common::event_capture::captured_with(&want) { + Some(event) => return event, + None => { + assert!( + std::time::Instant::now() < deadline, + "no direct-tcpip decline event within the row deadline" + ); + tokio::time::sleep(Duration::from_millis(200)).await; + } + } + } + }) + .await; + + let _ = ssh_child.kill().await; + accept_task.abort(); + + let event = match outcome { + Ok(event) => event, + Err(_) => panic!("the -L decline row hung past its 30 s deadline"), + }; + assert_eq!(event.field("decline_class"), Some("open")); + assert_eq!(event.field("reason"), Some("unfeatured")); + support::cleanup(&work); + } +} + +/// The decline-observability row for the sftp subsystem decline +/// (OQ-13) — a real `sftp` client's subsystem request meets the +/// named, channel-scoped decline with a clean diagnostic and no hang. +/// +/// Row placement note: the sftp slot gates nothing, and the row can +/// only be exercised where session channels open (git / session +/// builds — in the none/tunnel builds the session-open decline is the +/// observable, already pinned above and by the matrix suite). +#[cfg(unix)] +#[cfg(any(feature = "session", feature = "git"))] +mod subsystem_decline { + use std::time::Duration; + + use alkssh::Limits; + + use super::common::event_capture::{expected_build, install}; + use super::common::fixture; + use super::common::harness::serve_config as harness_serve_config; + use super::support; + + /// - *Contract stamp:* mapping.md §Channel-type rows (the + /// subsystem row: the `sftp` name declines-until-alkfs, OQ-13; + /// POC-3 §A.3's forced decline re-landed from a real sftp + /// client). + #[tokio::test] + async fn real_sftp_subsystem_attempt_declines_with_reason() { + if !support::tool_present("sftp", &["-h"]) { + support::skip("sftp decline row", "sftp is unavailable"); + return; + } + + install(); + let key = support::write_key("sftp-decline"); + let work = key + .parent() + .expect("the workdir sits above the key") + .to_path_buf(); + + let serve = harness_serve_config( + support::provider("decline-rider"), + &[fixture("bravo")], + Limits::default(), + ); + let (port, accept_task) = support::mount(serve).await; + + let batch = work.join("sftp-batch"); + std::fs::write(&batch, "ls\n").expect("the sftp batch file writes"); + + let outcome = tokio::time::timeout(Duration::from_secs(30), async { + tokio::process::Command::new("sftp") + .arg("-b") + .arg(&batch) + .arg("-P") + .arg(port.to_string()) + .arg("-i") + .arg(&key) + .args(support::ssh_options()) + .arg("decline-rider@127.0.0.1") + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .output() + .await + .expect("sftp spawns") + }) + .await; + + accept_task.abort(); + + let output = match outcome { + Ok(output) => output, + Err(_) => panic!("the sftp decline row hung past its 30 s deadline"), + }; + let stderr = String::from_utf8_lossy(&output.stderr).to_string(); + assert!( + !output.status.success(), + "the sftp decline surfaces a failure to the real client — stderr: {stderr:?}" + ); + assert!( + !stderr.trim().is_empty(), + "the decline surfaces a clean diagnostic — stderr: {stderr:?}" + ); + let event = support::decline_event(&[ + ("row", "subsystem_request"), + ("message", "decline: channel request rejected"), + ("feature_build", expected_build()), + ]) + .await; + assert_eq!(event.field("reason"), Some("unimplemented")); + support::cleanup(&work); + } +} + +/// The decline-observability row for the unknown-repo collapsed +/// generic denial — a real `git clone` of a nonexistent repo dies +/// with the collapsed `access denied` band before any protocol byte +/// (no existence oracle) and exit 128. +#[cfg(unix)] +#[cfg(feature = "git")] +mod unknown_repo_decline { + use std::sync::Arc; + use std::time::Duration; + + use alkssh::Limits; + + use super::common::git_doubles as gd; + use super::common::harness::serve_config as harness_serve_config; + use super::support; + + /// - *Contract stamp:* mapping.md §Session-channel dispatch + /// (unknown repo → the collapsed generic denial *before any + /// protocol byte*; POC-3 §A.5's vanilla diagnostics: exit 128, + /// `fatal: remote error: access denied: owner/absent`). + #[tokio::test] + async fn real_git_clone_of_an_unknown_repo_collapses_to_the_generic_denial() { + let (git_ok, git_reason) = support::git_available(); + if !git_ok { + support::skip("unknown-repo row", &git_reason); + return; + } + if !support::tool_present("ssh", &["-V"]) { + support::skip("unknown-repo row", "ssh is unavailable"); + return; + } + + let key = support::write_key("repo-denial"); + let work = key + .parent() + .expect("the workdir sits above the key") + .to_path_buf(); + + let doubles = Arc::new(gd::MemGit::new()); + let provider = support::provider("git-rider"); + let mut serve = harness_serve_config( + provider, + &[super::common::fixture("bravo")], + Limits::default(), + ); + serve.git_seam = Some(gd::git_seam(&doubles)); + let (port, accept_task) = support::mount(serve).await; + + let ssh = support::git_ssh_command(&key); + let outcome = tokio::time::timeout(Duration::from_secs(30), async { + tokio::process::Command::new("git") + .arg("clone") + .arg(format!("ssh://git-rider@127.0.0.1:{port}/owner/absent.git")) + .arg(work.join("absent")) + .env("GIT_SSH_COMMAND", &ssh) + .output() + .await + .expect("git clone spawns") + }) + .await; + + accept_task.abort(); + + let outcome = match outcome { + Ok(outcome) => outcome, + Err(_) => panic!("the unknown-repo row hung past its 30 s deadline"), + }; + let stderr = String::from_utf8_lossy(&outcome.stderr).to_string(); + assert_eq!( + outcome.status.code(), + Some(128), + "the collapsed denial is git's fatal-remote-error exit — stderr: {stderr:?}" + ); + assert!( + stderr.contains("access denied") && stderr.contains("owner/absent"), + "the denial names the collapsed `access denied` band with the normalized id, no existence \ + oracle — stderr: {stderr:?}" + ); + assert!( + !stderr.contains("not found"), + "no repository-existence oracle leaks — stderr: {stderr:?}" + ); + support::cleanup(&work); + } +}