crates/gunnar-loadbench/src/ssh.rs
//! **The git front door on `:2222`.**
//!
//! Two probes, because they cost the estate very different amounts and the
//! bench should be able to run the cheap one always.
//!
//! ## `ssh-kexinit` — always compiled, `std::net` only
//!
//! RFC 4253 §4.2 and §7.1: exchange identification strings, then read the
//! server's `SSH_MSG_KEXINIT` (message number 20) — the algorithm list — and
//! disconnect without completing the key exchange. That is a real, full
//! round trip against the real `SSH-2.0-gunnar_*` listener: it costs the
//! appliance an `accept`, a session task, a host-key lookup and the KEXINIT
//! it composes from its configured algorithms. It does NOT cost it a
//! Diffie-Hellman, which is what makes it safe to point a rate at.
//!
//! Zero dependencies. The identification exchange is bytes on a socket and
//! `std::net::TcpStream` speaks it. Pulling russh in to read one packet would
//! be the opposite of the reuse law.
//!
//! ## `git-ls-refs` — behind `--features git-ssh`
//!
//! A REAL git read verb: protocol-v2 `ls-refs` through
//! [`gunnar_client::ls_remote`], which dials over **russh** — the estate's one
//! SSH dialler, pure Rust, no `git` subprocess anywhere. `ls-refs` is the
//! cheapest complete git read there is: it advertises refs and sends no pack.
//!
//! **A full `clone` is deliberately not benched.** Pack generation is the most
//! expensive thing this appliance does, the estate is three small VMs, and the
//! brief says the goal is to find where it saturates without leaving it
//! unusable. `ls-refs` exercises the same door — TCP, KEX, publickey auth, the
//! exec dispatch, the repository path sandbox, the protocol-v2 handshake — and
//! stops before the part that would cost gigabytes of CPU and egress.
use std::io::{Read, Write};
use std::net::{TcpStream, ToSocketAddrs};
use std::time::{Duration, Instant};
use crate::outcome::{classify_transport, Outcome};
use crate::stats::Sample;
/// What we announce. A server that logs client versions should be able to see
/// exactly who made this traffic.
const OUR_IDENT: &[u8] = b"SSH-2.0-gunnar_loadbench\r\n";
/// SSH_MSG_KEXINIT.
const MSG_KEXINIT: u8 = 20;
/// The identification string is at most 255 bytes including CRLF (RFC 4253 §4.2),
/// and a server may send any number of other lines before it.
const MAX_IDENT: usize = 255;
const MAX_IDENT_LINES: usize = 16;
/// A `:2222` endpoint to knock on.
pub struct KexProbe {
#[allow(dead_code)] // carried so a probe can name itself in a panic-free error path
pub name: &'static str,
pub host: String,
pub port: u16,
pub timeout: Duration,
}
/// What the server said it is, read once before the load starts so the report
/// can name the software it measured.
pub fn identify(host: &str, port: u16, timeout: Duration) -> Result<String, String> {
let mut s = connect(host, port, timeout).map_err(|e| e.to_string())?;
s.write_all(OUR_IDENT).map_err(|e| e.to_string())?;
let (ident, _) = read_ident(&mut s).map_err(|e| e.to_string())?;
Ok(ident)
}
fn connect(host: &str, port: u16, timeout: Duration) -> std::io::Result<TcpStream> {
let addr = (host, port)
.to_socket_addrs()?
.next()
.ok_or_else(|| std::io::Error::new(std::io::ErrorKind::NotFound, "no address"))?;
let s = TcpStream::connect_timeout(&addr, timeout)?;
s.set_read_timeout(Some(timeout))?;
s.set_write_timeout(Some(timeout))?;
s.set_nodelay(true)?;
Ok(s)
}
/// Read lines until one begins `SSH-`. Returns that line and any bytes that
/// were read past it (a server that pipelines its KEXINIT behind the banner).
fn read_ident(s: &mut TcpStream) -> std::io::Result<(String, Vec<u8>)> {
let mut buf = Vec::with_capacity(MAX_IDENT * 2);
let mut byte = [0u8; 1];
let mut lines = 0usize;
loop {
let mut line = Vec::with_capacity(MAX_IDENT);
loop {
let n = s.read(&mut byte)?;
if n == 0 {
return Err(std::io::Error::new(
std::io::ErrorKind::UnexpectedEof,
"closed before an identification string",
));
}
if byte[0] == b'\n' {
break;
}
if byte[0] != b'\r' {
line.push(byte[0]);
}
if line.len() > MAX_IDENT {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"identification string over 255 bytes",
));
}
}
if line.starts_with(b"SSH-") {
return Ok((String::from_utf8_lossy(&line).into_owned(), buf));
}
lines += 1;
if lines > MAX_IDENT_LINES {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"too many pre-banner lines",
));
}
buf.extend_from_slice(&line);
}
}
/// Read one binary packet (RFC 4253 §6, unencrypted form) and return its
/// payload's first byte — the message number.
fn read_msg_number(s: &mut TcpStream) -> std::io::Result<u8> {
let mut len = [0u8; 4];
s.read_exact(&mut len)?;
let packet_len = u32::from_be_bytes(len) as usize;
// A KEXINIT is a few hundred bytes. Anything enormous here is not a KEXINIT
// and reading it would be the bench letting a peer allocate on its behalf.
if packet_len < 2 || packet_len > 65_536 {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!("packet_length {packet_len} is not a plausible KEXINIT"),
));
}
let mut head = [0u8; 2];
s.read_exact(&mut head)?;
Ok(head[1])
}
impl KexProbe {
/// One knock: connect, exchange identification strings, read the server's
/// KEXINIT, hang up. Never panics.
pub fn once(&self, released: Instant) -> Sample {
let r = (|| -> std::io::Result<u8> {
let mut s = connect(&self.host, self.port, self.timeout)?;
s.write_all(OUR_IDENT)?;
s.flush()?;
let (_ident, _rest) = read_ident(&mut s)?;
read_msg_number(&mut s)
})();
let dt = released.elapsed().as_nanos() as u64;
match r {
Ok(MSG_KEXINIT) => Sample {
latency_ns: dt,
late_ns: 0,
bytes: 0,
outcome: Outcome::Ok,
},
// It answered, but not with the algorithm list. On this door that
// is a refusal (russh sends DISCONNECT, msg 1, when it will not
// serve), and it is a different fact from a dead socket.
Ok(_) => Sample::refused(Outcome::SshAuthRefused, dt, 0),
Err(e) => {
let o = match e.kind() {
std::io::ErrorKind::InvalidData | std::io::ErrorKind::UnexpectedEof => {
Outcome::SshNoBanner
}
k => classify_transport(Some(k), &e.to_string()),
};
Sample::refused(o, dt, 0)
}
}
}
}
/// The real git read verb, protocol-v2 `ls-refs` over russh.
#[cfg(feature = "git-ssh")]
pub struct LsRefsProbe {
#[allow(dead_code)] // the report prints the catalogue name; this is the probe's own copy
pub name: &'static str,
pub url: String,
pub key: std::path::PathBuf,
}
#[cfg(feature = "git-ssh")]
impl LsRefsProbe {
pub fn once(&self, released: Instant) -> Sample {
match gunnar_client::ls_remote(&self.url, &self.key) {
Ok(refs) => Sample {
latency_ns: released.elapsed().as_nanos() as u64,
late_ns: 0,
bytes: refs.len().min(u32::MAX as usize) as u32,
outcome: Outcome::Ok,
},
Err(e) => {
let dt = released.elapsed().as_nanos() as u64;
let t = e.to_string().to_ascii_lowercase();
// An auth refusal and a dead socket are not the same fact.
let o = if t.contains("auth") || t.contains("permission") || t.contains("key") {
Outcome::SshAuthRefused
} else {
classify_transport(None, &t)
};
Sample::refused(o, dt, 0)
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::net::TcpListener;
/// A server that speaks the first two steps correctly.
fn fake_ssh(payload_msg: u8) -> (String, u16) {
let l = TcpListener::bind("127.0.0.1:0").unwrap();
let port = l.local_addr().unwrap().port();
std::thread::spawn(move || {
for c in l.incoming().take(4) {
let mut c = match c {
Ok(c) => c,
Err(_) => continue,
};
let _ = c.write_all(b"SSH-2.0-fake_1.0 test\r\n");
// packet_length = 2 (padding_length + one payload byte)
let _ = c.write_all(&2u32.to_be_bytes());
let _ = c.write_all(&[0u8, payload_msg]);
let _ = c.flush();
std::thread::sleep(Duration::from_millis(30));
}
});
("127.0.0.1".to_string(), port)
}
#[test]
fn a_kexinit_is_an_ok() {
let (h, p) = fake_ssh(MSG_KEXINIT);
let probe = KexProbe {
name: "t",
host: h,
port: p,
timeout: Duration::from_secs(2),
};
assert_eq!(probe.once(Instant::now()).outcome, Outcome::Ok);
}
#[test]
fn a_disconnect_instead_of_a_kexinit_is_a_refusal_not_an_ok() {
let (h, p) = fake_ssh(1); // SSH_MSG_DISCONNECT
let probe = KexProbe {
name: "t",
host: h,
port: p,
timeout: Duration::from_secs(2),
};
assert_eq!(probe.once(Instant::now()).outcome, Outcome::SshAuthRefused);
}
#[test]
fn nothing_listening_is_connect_refused_and_never_a_panic() {
// port 1 on loopback: reliably refused, never a timeout.
let probe = KexProbe {
name: "t",
host: "127.0.0.1".into(),
port: 1,
timeout: Duration::from_millis(500),
};
let o = probe.once(Instant::now()).outcome;
assert!(
matches!(o, Outcome::ConnectRefused | Outcome::Timeout),
"{o}"
);
}
#[test]
fn a_server_that_hangs_up_before_the_banner_is_its_own_bucket() {
let l = TcpListener::bind("127.0.0.1:0").unwrap();
let port = l.local_addr().unwrap().port();
std::thread::spawn(move || {
if let Some(Ok(c)) = l.incoming().next() {
drop(c);
}
});
let probe = KexProbe {
name: "t",
host: "127.0.0.1".into(),
port,
timeout: Duration::from_secs(2),
};
let o = probe.once(Instant::now()).outcome;
assert!(
matches!(o, Outcome::SshNoBanner | Outcome::ConnectionReset),
"{o}"
);
}
#[test]
fn an_absurd_packet_length_is_refused_rather_than_allocated() {
let l = TcpListener::bind("127.0.0.1:0").unwrap();
let port = l.local_addr().unwrap().port();
std::thread::spawn(move || {
if let Some(Ok(mut c)) = l.incoming().next() {
let _ = c.write_all(b"SSH-2.0-fake\r\n");
let _ = c.write_all(&u32::MAX.to_be_bytes());
std::thread::sleep(Duration::from_millis(50));
}
});
let probe = KexProbe {
name: "t",
host: "127.0.0.1".into(),
port,
timeout: Duration::from_secs(2),
};
assert_eq!(probe.once(Instant::now()).outcome, Outcome::SshNoBanner);
}
}