crates/nornir-git/tests/liveness_common/mod.rs
//! The harness the liveness tests share: a real russh SERVER, a TCP PROXY that
//! can go silent or hang up mid-transfer, and a WATCHDOG so a client that parks
//! forever shows up as a failed test, not as a CI job that never ends.
//!
//! Why a proxy and not the server itself: the defect (cov-13 finding 4,
//! 2026-10-06) was a link that went SILENT — a WireGuard tunnel that stopped
//! carrying packets for six minutes — and a peer that hung up while the client
//! was parked. Neither is something a well-behaved server does on its own. The
//! proxy sits on the client's TCP connection and does exactly those two things
//! on command:
//!
//! * [`Mode::Silent`] — stop reading and stop forwarding, both ways, but keep
//! both sockets OPEN. The client's kernel keeps getting ACKs (loopback) and
//! then a zero window, so TCP itself never times out: only the SSH layer can
//! notice. That is the worst case, and the incident's shape (CLOSE-WAIT, bytes
//! unread, every thread parked).
//! * [`Mode::Close`] — drop both sockets: the peer hung up.
//!
//! Every server and proxy runs on its OWN thread and runtime, because the client
//! under test builds a runtime of its own and blocks the calling thread on it.
#![allow(dead_code)]
use std::io::{Read, Write};
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use russh::server::Server as _;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::sync::watch;
// ── the watchdog ────────────────────────────────────────────────────────────
/// Run `f` on its own thread and wait at most `bound` for it.
///
/// A client that parks forever is THE bug, so it must surface as a failure with
/// a name, never as a hung test binary. On timeout the stuck thread is left
/// behind (it is parked; the test process ends without it) and the test panics
/// naming what hung.
pub fn within<T: Send + 'static>(
bound: Duration,
what: &str,
f: impl FnOnce() -> T + Send + 'static,
) -> (T, Duration) {
let (tx, rx) = std::sync::mpsc::channel();
let started = Instant::now();
std::thread::Builder::new()
.name(format!("watched: {what}"))
.spawn(move || {
let out = f();
let _ = tx.send(out);
})
.expect("spawn the watched thread");
match rx.recv_timeout(bound) {
Ok(out) => (out, started.elapsed()),
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
panic!("{what}: the watched thread panicked (see its message above)")
}
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => panic!(
"HUNG: {what} did not return within {} s — the client is parked \
(the cov-13 finding-4 defect: a dropped or silent link never surfaces)",
bound.as_secs()
),
}
}
// ── the proxy ───────────────────────────────────────────────────────────────
/// What a proxied connection does with its bytes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Mode {
/// Pass everything through.
Forward,
/// Read nothing, forward nothing, keep both sockets open.
Silent,
/// Drop both sockets.
Close,
}
#[derive(Default)]
struct ProxyCtl {
/// For connections accepted from now on: switch to this mode once this many
/// client→server bytes have passed.
plan: Mutex<Option<(u64, Mode)>>,
/// Every live connection's mode switch.
conns: Mutex<Vec<watch::Sender<Mode>>>,
/// Client→server bytes forwarded, over all connections.
up_bytes: AtomicU64,
}
/// A loopback TCP proxy in front of a server, with a hand on the link.
pub struct Proxy {
pub port: u16,
ctl: Arc<ProxyCtl>,
}
impl Proxy {
/// Listen on an ephemeral loopback port and forward every connection to
/// `upstream`.
pub fn start(upstream: SocketAddr) -> Proxy {
let ctl = Arc::new(ProxyCtl::default());
let ctl_thread = ctl.clone();
let (tx, rx) = std::sync::mpsc::channel::<u16>();
std::thread::Builder::new()
.name("test-proxy".into())
.spawn(move || {
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.expect("proxy runtime");
rt.block_on(async move {
let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
.await
.expect("bind proxy");
tx.send(listener.local_addr().expect("addr").port()).expect("publish port");
loop {
let Ok((client, _)) = listener.accept().await else { continue };
let Ok(server) = tokio::net::TcpStream::connect(upstream).await else {
continue;
};
let _ = client.set_nodelay(true);
let _ = server.set_nodelay(true);
let plan = *ctl_thread.plan.lock().unwrap();
let (mode_tx, mode_rx) = watch::channel(Mode::Forward);
ctl_thread.conns.lock().unwrap().push(mode_tx.clone());
let (cr, cw) = client.into_split();
let (sr, sw) = server.into_split();
let ctl_up = ctl_thread.clone();
tokio::spawn(pump(cr, sw, mode_rx.clone(), mode_tx.clone(), Some((ctl_up, plan))));
tokio::spawn(pump(sr, cw, mode_rx, mode_tx, None));
}
});
})
.expect("spawn proxy thread");
let port = rx.recv().expect("the proxy must publish its port");
Proxy { port, ctl }
}
/// Connections accepted from now on switch to `mode` once `bytes`
/// client→server bytes have passed through the proxy in total.
pub fn arm(&self, bytes: u64, mode: Mode) {
*self.ctl.plan.lock().unwrap() = Some((bytes, mode));
}
/// Connections accepted from now on forward forever.
pub fn disarm(&self) {
*self.ctl.plan.lock().unwrap() = None;
}
/// Switch every live connection to `mode` now.
pub fn set_all(&self, mode: Mode) {
for c in self.ctl.conns.lock().unwrap().iter() {
let _ = c.send(mode);
}
}
/// Hang up every connection and forget them, so the next scenario starts
/// from nothing.
pub fn close_all(&self) {
self.set_all(Mode::Close);
self.ctl.conns.lock().unwrap().clear();
}
pub fn up_bytes(&self) -> u64 {
self.ctl.up_bytes.load(Ordering::SeqCst)
}
/// An `ssh://` URL that dials the server THROUGH this proxy.
pub fn url(&self, path: &str) -> String {
format!("ssh://git@127.0.0.1:{}{}", self.port, path)
}
}
/// The client→server half's counter and its armed switch.
type Upstream = Option<(Arc<ProxyCtl>, Option<(u64, Mode)>)>;
/// One direction of a proxied connection.
///
/// `up` is `Some` on the client→server half: it counts bytes and fires the
/// connection's armed mode switch when the count is reached.
async fn pump(
mut src: tokio::net::tcp::OwnedReadHalf,
mut dst: tokio::net::tcp::OwnedWriteHalf,
mut mode: watch::Receiver<Mode>,
mode_tx: watch::Sender<Mode>,
up: Upstream,
) {
let mut buf = vec![0u8; 64 * 1024];
let mut passed: u64 = 0;
loop {
let now = *mode.borrow_and_update();
match now {
Mode::Forward => {}
Mode::Close => return,
Mode::Silent => {
// Hold both sockets, touch neither, until told to hang up.
loop {
if mode.changed().await.is_err() {
return;
}
if *mode.borrow_and_update() == Mode::Close {
return;
}
}
}
}
tokio::select! {
r = src.read(&mut buf) => {
let n = match r {
Ok(0) | Err(_) => {
// One side went away: the connection is over both ways.
let _ = mode_tx.send(Mode::Close);
return;
}
Ok(n) => n,
};
if *mode.borrow() != Mode::Forward {
continue; // the link died while this read was in flight
}
if dst.write_all(&buf[..n]).await.is_err() {
let _ = mode_tx.send(Mode::Close);
return;
}
if let Some((ctl, plan)) = &up {
passed += n as u64;
ctl.up_bytes.fetch_add(n as u64, Ordering::SeqCst);
if let Some((at, then)) = plan {
if passed >= *at {
let _ = mode_tx.send(*then);
}
}
}
}
_ = mode.changed() => {}
}
}
}
// ── the server ──────────────────────────────────────────────────────────────
/// What the test server does on `exec`, chosen by the command string:
///
/// * `sink` — answer `ready\n`, then swallow everything; on a trailing `END\n`
/// answer `ok\n` after `answer_after` (or never, if it is `None`) and close.
/// * `git-upload-pack …` — answer ONE ref line of a v0 advertisement and then
/// nothing (no flush): a far end that is alive but stuck mid-list.
#[derive(Clone)]
pub struct ServerPlan {
/// How long after `END\n` the `sink` answers; `None` = never.
pub answer_after: Option<Duration>,
}
#[derive(Default)]
pub struct ServerSeen {
/// Bytes the `sink` swallowed.
pub sunk: AtomicU64,
/// The `git-upload-pack` half-advertisement has gone out.
pub partial_sent: std::sync::atomic::AtomicBool,
}
struct SrvHandler {
plan: ServerPlan,
seen: Arc<ServerSeen>,
tail: Vec<u8>,
answered: bool,
}
impl russh::server::Handler for SrvHandler {
type Error = russh::Error;
async fn auth_publickey(
&mut self,
_user: &str,
_key: russh::server::VerifiedPublicKey<'_>,
) -> Result<russh::server::Auth, Self::Error> {
Ok(russh::server::Auth::Accept)
}
async fn channel_open_session(
&mut self,
_channel: russh::Channel<russh::server::Msg>,
reply: russh::server::ChannelOpenHandle,
_session: &mut russh::server::Session,
) -> Result<(), Self::Error> {
reply.accept().await;
Ok(())
}
async fn env_request(
&mut self,
channel: russh::ChannelId,
_name: &str,
_value: &str,
session: &mut russh::server::Session,
) -> Result<(), Self::Error> {
session.channel_success(channel)?;
Ok(())
}
async fn exec_request(
&mut self,
channel: russh::ChannelId,
data: &[u8],
session: &mut russh::server::Session,
) -> Result<(), Self::Error> {
let cmd = String::from_utf8_lossy(data).to_string();
session.channel_success(channel)?;
let handle = session.handle();
if cmd.starts_with("git-upload-pack") {
let seen = self.seen.clone();
tokio::spawn(async move {
// ONE complete v0 ref line, and then nothing ever: no flush,
// no second ref. A reader that takes a cut here for the end of
// the list returns a short list as if it were the whole one.
let body = "7fd1a60b01f91b314f59955a4e4d4e80d8edf11d HEAD\0multi_ack\n";
let line = format!("{:04x}{body}", body.len() + 4);
let _ = handle.data(channel, line.into_bytes()).await;
seen.partial_sent.store(true, Ordering::SeqCst);
});
} else {
tokio::spawn(async move {
let _ = handle.data(channel, b"ready\n".to_vec()).await;
});
}
Ok(())
}
async fn data(
&mut self,
channel: russh::ChannelId,
data: &[u8],
session: &mut russh::server::Session,
) -> Result<(), Self::Error> {
self.seen.sunk.fetch_add(data.len() as u64, Ordering::SeqCst);
self.tail.extend_from_slice(data);
if self.tail.len() > 8 {
let cut = self.tail.len() - 8;
self.tail.drain(..cut);
}
if !self.answered && self.tail.ends_with(b"END\n") {
self.answered = true;
if let Some(after) = self.plan.answer_after {
let handle = session.handle();
tokio::spawn(async move {
tokio::time::sleep(after).await;
let _ = handle.data(channel, b"ok\n".to_vec()).await;
let _ = handle.exit_status_request(channel, 0).await;
let _ = handle.eof(channel).await;
let _ = handle.close(channel).await;
});
}
}
Ok(())
}
}
struct SrvFactory {
plan: ServerPlan,
seen: Arc<ServerSeen>,
}
impl russh::server::Server for SrvFactory {
type Handler = SrvHandler;
fn new_client(&mut self, _peer: Option<SocketAddr>) -> SrvHandler {
SrvHandler { plan: self.plan.clone(), seen: self.seen.clone(), tail: Vec::new(), answered: false }
}
}
/// A running test sshd.
pub struct TestServer {
pub addr: SocketAddr,
pub seen: Arc<ServerSeen>,
}
impl TestServer {
pub fn start(dir: &Path, plan: ServerPlan) -> TestServer {
let keydir = dir.join("hostkey");
nornir_git::sshkey::generate_ed25519_openssh(&keydir, "liveness-sshd@localhost")
.expect("mint host key");
let pem = std::fs::read_to_string(keydir.join(nornir_git::sshkey::PRIV_FILE))
.expect("read host key");
let seen = Arc::new(ServerSeen::default());
let seen_srv = seen.clone();
let (tx, rx) = std::sync::mpsc::channel::<SocketAddr>();
std::thread::Builder::new()
.name("test-sshd".into())
.spawn(move || {
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.expect("server runtime");
rt.block_on(async move {
let key = russh::keys::PrivateKey::from_openssh(&pem).expect("host key");
let config = Arc::new(russh::server::Config {
keys: vec![key],
// The SERVER keeps no keepalive of its own here: what is
// under test is the client noticing on its own.
inactivity_timeout: Some(Duration::from_secs(300)),
auth_rejection_time: Duration::ZERO,
nodelay: true,
..Default::default()
});
let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
.await
.expect("bind sshd");
tx.send(listener.local_addr().expect("addr")).expect("publish addr");
let mut factory = SrvFactory { plan, seen: seen_srv };
let _ = factory.run_on_socket(config, &listener).await;
});
})
.expect("spawn sshd thread");
let addr = rx.recv().expect("the server must publish its address");
TestServer { addr, seen }
}
}
/// A listener that accepts and then says NOTHING — not even the SSH banner.
pub fn mute_listener() -> u16 {
let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).expect("bind mute");
let port = listener.local_addr().expect("addr").port();
std::thread::Builder::new()
.name("mute-listener".into())
.spawn(move || {
let mut held = Vec::new();
for s in listener.incoming().flatten() {
held.push(s); // hold it open, say nothing
}
})
.expect("spawn mute listener");
port
}
// ── the client side of a scenario ───────────────────────────────────────────
/// Mint a client key.
pub fn mint_key(dir: &Path) -> PathBuf {
let d = dir.join("client");
nornir_git::sshkey::generate_ed25519_openssh(&d, "liveness-client@localhost").expect("mint key");
d.join(nornir_git::sshkey::PRIV_FILE)
}
/// Read exactly `n` bytes, as the protocol code does.
pub fn read_n(r: &mut dyn Read, n: usize) -> std::io::Result<Vec<u8>> {
let mut v = vec![0u8; n];
r.read_exact(&mut v)?;
Ok(v)
}
/// Push-shaped load: write 64 KiB chunks until `limit` bytes or an error.
pub fn write_until_error(w: &mut dyn Write, limit: u64) -> std::io::Result<u64> {
let chunk = vec![0x5au8; 64 * 1024];
let mut sent = 0u64;
while sent < limit {
w.write_all(&chunk)?;
sent += chunk.len() as u64;
}
w.flush()?;
Ok(sent)
}
// ── leak accounting ─────────────────────────────────────────────────────────
/// Threads in this process.
pub fn threads() -> usize {
std::fs::read_dir("/proc/self/task").map(|d| d.count()).unwrap_or(0)
}
/// Open file descriptors in this process.
pub fn fds() -> usize {
std::fs::read_dir("/proc/self/fd").map(|d| d.count()).unwrap_or(0)
}
/// TCP sockets in CLOSE-WAIT whose REMOTE port is `port` — i.e. client sockets
/// to the proxy whose peer hung up and which nobody ever read to the end.
pub fn close_wait_to(port: u16) -> usize {
let mut n = 0;
for f in ["/proc/net/tcp", "/proc/net/tcp6"] {
let Ok(text) = std::fs::read_to_string(f) else { continue };
for line in text.lines().skip(1) {
let cols: Vec<&str> = line.split_whitespace().collect();
if cols.len() < 4 {
continue;
}
let rem_port = cols[2].rsplit(':').next().and_then(|p| u16::from_str_radix(p, 16).ok());
if rem_port == Some(port) && cols[3] == "08" {
n += 1;
}
}
}
n
}
/// Poll until `f` holds or `bound` passes; the last value either way.
pub fn settle(bound: Duration, mut f: impl FnMut() -> bool) -> bool {
let until = Instant::now() + bound;
loop {
if f() {
return true;
}
if Instant::now() >= until {
return false;
}
std::thread::sleep(Duration::from_millis(50));
}
}