From 6a1e47532dbc3e281225d65cc274304c631530a8 Mon Sep 17 00:00:00 2001 From: atlas Date: Tue, 3 Nov 2026 02:35:19 +0100 Subject: [PATCH] hive-priv: test main.rs's request dispatch and validation, and infra.rs check_socket_activation and response_of are extracted from socket_listener and dispatch unchanged, and infra.rs takes a &impl Spawn, so tests run stand-ins instead of systemctl. --- hive-priv/src/infra.rs | 227 +++++++++++++++- hive-priv/src/main.rs | 590 ++++++++++++++++++++++++++++++++++++++++- 2 files changed, 795 insertions(+), 22 deletions(-) diff --git a/hive-priv/src/infra.rs b/hive-priv/src/infra.rs index 8723c1b7..71ae1035 100644 --- a/hive-priv/src/infra.rs +++ b/hive-priv/src/infra.rs @@ -2,9 +2,9 @@ //! [`InfraContainer`] unit) and `ReloadGatewayNginx`. Neither takes a unit or //! container name from the caller. +use crate::shared::{RealSpawn, Spawn}; use anyhow::{Context as _, Result, bail}; use hive_priv_sock::{InfraAction, InfraContainer}; -use tokio::process::Command; /// `ControlInfraContainer` — start/stop a hive infrastructure service via /// `systemctl `. The [`InfraContainer`] enum is the allowlist: @@ -18,10 +18,19 @@ use tokio::process::Command; pub(crate) async fn control_infra_container( container: InfraContainer, action: InfraAction, +) -> Result<(String, String)> { + control_infra_container_with(&RealSpawn, container, action).await +} + +async fn control_infra_container_with( + spawn: &impl Spawn, + container: InfraContainer, + action: InfraAction, ) -> Result<(String, String)> { let verb = action.systemctl_verb(); let unit = container.service_unit(); - let out = Command::new("systemctl") + let out = spawn + .command("systemctl") .args([verb, unit]) .output() .await @@ -55,7 +64,12 @@ pub(crate) async fn control_infra_container( /// Returns `(String::new(), String::new())` on success so it fits the /// `exec` return type directly. pub(crate) async fn sync_gateway_nginx() -> Result<(String, String)> { - let state_out = Command::new("systemctl") + sync_gateway_nginx_with(&RealSpawn).await +} + +async fn sync_gateway_nginx_with(spawn: &impl Spawn) -> Result<(String, String)> { + let state_out = spawn + .command("systemctl") .args(["show", "--property=ActiveState", "--value", "nginx"]) .output() .await @@ -72,7 +86,8 @@ pub(crate) async fn sync_gateway_nginx() -> Result<(String, String)> { // start-limit failure; plain start when inactive or unknown. match state.as_str() { "active" => { - let out = Command::new("systemctl") + let out = spawn + .command("systemctl") .args(["reload", "nginx"]) .output() .await @@ -87,11 +102,13 @@ pub(crate) async fn sync_gateway_nginx() -> Result<(String, String)> { } "failed" => { // Clear start-limit so the next start can proceed. - let _ = Command::new("systemctl") + let _ = spawn + .command("systemctl") .args(["reset-failed", "nginx"]) .status() .await; - let out = Command::new("systemctl") + let out = spawn + .command("systemctl") .args(["start", "nginx"]) .output() .await @@ -106,7 +123,8 @@ pub(crate) async fn sync_gateway_nginx() -> Result<(String, String)> { } _ => { // inactive, activating, deactivating, unknown — just start. - let out = Command::new("systemctl") + let out = spawn + .command("systemctl") .args(["start", "nginx"]) .output() .await @@ -125,7 +143,12 @@ pub(crate) async fn sync_gateway_nginx() -> Result<(String, String)> { #[cfg(test)] mod tests { - use super::{InfraAction, InfraContainer}; + use std::path::PathBuf; + + use super::{ + InfraAction, InfraContainer, control_infra_container_with, sync_gateway_nginx_with, + }; + use crate::shared::test_support::{MissingTool, ScriptSpawn, scratch}; use hive_priv_sock::PrivRequest; /// A start/stop naming the forge or matrix container fails to decode, so @@ -148,4 +171,192 @@ mod tests { }) )); } + + /// Stand-in for `systemctl`: every call is logged as `systemctl args…`, + /// then `behave` (a shell script over `$@`) decides output and status. + struct Stub { + dir: PathBuf, + script: String, + } + + impl Stub { + fn new(behave: &str) -> Self { + let dir = scratch(); + let script = format!("echo \"$0 $*\" >> {d}/calls\n{behave}", d = dir.display()); + Self { dir, script } + } + + fn spawn(&self) -> ScriptSpawn { + ScriptSpawn(self.script.clone()) + } + + fn calls(&self) -> Vec { + std::fs::read_to_string(self.dir.join("calls")) + .map(|log| log.lines().map(str::to_owned).collect()) + .unwrap_or_default() + } + } + + impl Drop for Stub { + fn drop(&mut self) { + std::fs::remove_dir_all(&self.dir).ok(); + } + } + + const SHOW: &str = "systemctl show --property=ActiveState --value nginx"; + + /// A host whose nginx reports `state`; every other verb succeeds. + fn nginx_in(state: &str) -> Stub { + Stub::new(&format!("case \"$1\" in show) echo {state};; esac")) + } + + async fn sync(stub: &Stub) -> anyhow::Result<(String, String)> { + sync_gateway_nginx_with(&stub.spawn()).await + } + + async fn control( + stub: &Stub, + container: InfraContainer, + action: InfraAction, + ) -> anyhow::Result<(String, String)> { + control_infra_container_with(&stub.spawn(), container, action).await + } + + fn err_text(r: anyhow::Result<(String, String)>) -> String { + format!("{:#}", r.unwrap_err()) + } + + // ---- ControlInfraContainer ---- + + /// The unit comes from the variant and the verb from the action: the + /// whole argv is `systemctl nginx.service`, nothing else. + #[tokio::test] + async fn control_runs_exactly_the_verb_on_the_variants_unit() { + for (action, verb) in [(InfraAction::Start, "start"), (InfraAction::Stop, "stop")] { + let stub = Stub::new("echo out; echo err >&2"); + let (stdout, stderr) = control(&stub, InfraContainer::Gateway, action) + .await + .unwrap(); + assert_eq!( + stub.calls(), + [format!("systemctl {verb} nginx.service")], + "{verb}" + ); + assert_eq!((stdout.as_str(), stderr.as_str()), ("out\n", "err\n")); + } + } + + /// A non-zero exit is an error naming the verb, the unit, the status and + /// the trimmed stderr, so the operator sees why `hivectl` failed. + #[tokio::test] + async fn control_reports_a_failed_systemctl_with_its_stderr() { + let stub = Stub::new("echo 'unit is masked' >&2; exit 5"); + let msg = err_text(control(&stub, InfraContainer::Gateway, InfraAction::Stop).await); + assert!(msg.contains("systemctl stop nginx.service exited"), "{msg}"); + assert!(msg.ends_with("unit is masked"), "{msg}"); + } + + /// A host without `systemctl` is an error that names the command. + #[tokio::test] + async fn control_reports_a_missing_systemctl_by_name() { + let msg = err_text( + control_infra_container_with(&MissingTool, InfraContainer::Gateway, InfraAction::Start) + .await, + ); + assert!(msg.contains("systemctl start nginx.service"), "{msg}"); + } + + // ---- ReloadGatewayNginx ---- + + /// A running nginx is reloaded, never restarted. + #[tokio::test] + async fn an_active_nginx_is_reloaded() { + let stub = nginx_in("active"); + assert_eq!(sync(&stub).await.unwrap(), (String::new(), String::new())); + assert_eq!(stub.calls(), [SHOW, "systemctl reload nginx"]); + } + + /// A failed nginx has its start-limit cleared before the start. + #[tokio::test] + async fn a_failed_nginx_is_reset_then_started() { + let stub = nginx_in("failed"); + sync(&stub).await.unwrap(); + assert_eq!( + stub.calls(), + [ + SHOW, + "systemctl reset-failed nginx", + "systemctl start nginx" + ] + ); + } + + /// Any other state, including one the query could not produce, is a + /// plain start: no reload of a unit that is not running. + #[tokio::test] + async fn every_other_state_is_a_plain_start() { + for state in ["inactive", "activating", "deactivating", "reloading", ""] { + let stub = nginx_in(state); + sync(&stub).await.unwrap(); + assert_eq!(stub.calls(), [SHOW, "systemctl start nginx"], "{state:?}"); + } + } + + /// A failing state query is only logged: the sync still starts nginx + /// from whatever the query printed, here nothing. + #[tokio::test] + async fn a_failed_state_query_still_starts_nginx() { + let stub = Stub::new("case \"$1\" in show) echo boom >&2; exit 1;; esac"); + sync(&stub).await.unwrap(); + assert_eq!(stub.calls(), [SHOW, "systemctl start nginx"]); + } + + /// A failed reload is an error carrying systemctl's stderr; a failed + /// `reset-failed` is not, because the start that follows is what counts. + #[tokio::test] + async fn a_failed_reload_is_an_error_and_a_failed_reset_is_not() { + let stub = Stub::new( + "case \"$1\" in show) echo active;; reload) echo 'bad config' >&2; exit 1;; esac", + ); + let msg = err_text(sync(&stub).await); + assert!(msg.contains("gateway nginx reload failed"), "{msg}"); + assert!(msg.ends_with("bad config"), "{msg}"); + + let stub = Stub::new("case \"$1\" in show) echo failed;; reset-failed) exit 1;; esac"); + sync(&stub).await.unwrap(); + assert_eq!( + stub.calls(), + [ + SHOW, + "systemctl reset-failed nginx", + "systemctl start nginx" + ] + ); + } + + /// A start that fails is an error on both start paths, and the plain + /// one names the state it started from. + #[tokio::test] + async fn a_failed_start_is_an_error_on_both_start_paths() { + let stub = Stub::new( + "case \"$1\" in show) echo failed;; start) echo 'start limit hit' >&2; exit 1;; esac", + ); + let msg = err_text(sync(&stub).await); + assert!(msg.contains("start (after reset-failed) failed"), "{msg}"); + assert!(msg.ends_with("start limit hit"), "{msg}"); + + let stub = Stub::new( + "case \"$1\" in show) echo inactive;; start) echo 'no such unit' >&2; exit 1;; esac", + ); + let msg = err_text(sync(&stub).await); + assert!(msg.contains("state=\"inactive\""), "{msg}"); + assert!(msg.ends_with("no such unit"), "{msg}"); + } + + /// Without `systemctl` the state query itself fails, and says so. + #[tokio::test] + async fn a_missing_systemctl_fails_the_state_query() { + let msg = err_text(sync_gateway_nginx_with(&MissingTool).await); + assert!(msg.contains("query gateway nginx ActiveState"), "{msg}"); + } } diff --git a/hive-priv/src/main.rs b/hive-priv/src/main.rs index 27f4f51f..faf1c845 100644 --- a/hive-priv/src/main.rs +++ b/hive-priv/src/main.rs @@ -84,19 +84,18 @@ async fn main() -> Result<()> { } } -fn socket_listener() -> Result { - // hive-priv is ALWAYS socket-activated by the `hive-priv.socket` unit - // (fd 3 via LISTEN_FDS). There is intentionally no self-bind fallback, - // so dev and prod take the same path; see docs/trust-boundary/boundary.md. - let listen_fds: Option = std::env::var("LISTEN_FDS") - .ok() - .and_then(|s| s.parse().ok()); - let listen_pid: Option = std::env::var("LISTEN_PID") - .ok() - .and_then(|s| s.parse().ok()); +/// Refuse to run unless systemd handed this process its socket: `LISTEN_FDS` +/// is at least 1 and `LISTEN_PID` is `own_pid`. A `LISTEN_PID` naming another +/// process means the variables were inherited from a parent, not meant for us. +fn check_socket_activation( + listen_fds: Option<&str>, + listen_pid: Option<&str>, + own_pid: u32, +) -> Result<()> { + let listen_fds: Option = listen_fds.and_then(|s| s.parse().ok()); + let listen_pid: Option = listen_pid.and_then(|s| s.parse().ok()); - let activated = - matches!(listen_fds, Some(n) if n >= 1) && listen_pid == Some(std::process::id()); + let activated = matches!(listen_fds, Some(n) if n >= 1) && listen_pid == Some(own_pid); if !activated { bail!( "hive-priv requires systemd socket activation (expected LISTEN_FDS>=1 + \ @@ -104,6 +103,18 @@ fn socket_listener() -> Result { not directly" ); } + Ok(()) +} + +fn socket_listener() -> Result { + // hive-priv is ALWAYS socket-activated by the `hive-priv.socket` unit + // (fd 3 via LISTEN_FDS). There is intentionally no self-bind fallback, + // so dev and prod take the same path; see docs/trust-boundary/boundary.md. + check_socket_activation( + std::env::var("LISTEN_FDS").ok().as_deref(), + std::env::var("LISTEN_PID").ok().as_deref(), + std::process::id(), + )?; // SAFETY: systemd has passed us a ready UnixListener on fd 3. let std_listener = unsafe { @@ -307,7 +318,13 @@ fn check_fd_agreement(req: &PrivRequest, fd: Option<&OwnedFd>) -> Result<()> { } async fn dispatch(line: &str, fd: Option, writer: &mut OwnedWriteHalf) -> PrivResponse { - match run(line, fd, writer).await { + response_of(run(line, fd, writer).await) +} + +/// The wire answer for one request's outcome: the handler's output on +/// success, otherwise the whole error chain and no output. +fn response_of(outcome: Result<(String, String)>) -> PrivResponse { + match outcome { Ok((stdout, stderr)) => PrivResponse { ok: true, stdout, @@ -515,7 +532,18 @@ async fn exec( } #[cfg(test)] mod tests { - use super::{OwnedFd, PrivRequest, check_fd_agreement}; + use std::io::Write as _; + use std::os::fd::{AsRawFd as _, OwnedFd, RawFd}; + use std::os::unix::fs::MetadataExt as _; + + use hive_priv_sock::{PrivEvent, PrivRequest, PrivResponse}; + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; + use tokio::net::UnixStream; + + use super::{ + Requests, check_fd_agreement, check_socket_activation, dispatch, handle, recv_with_fds, + response_of, + }; /// A request that streams into a caller-supplied descriptor. fn fd_taking_request() -> PrivRequest { @@ -586,4 +614,538 @@ mod tests { )); assert!(serde_json::from_str::(r#"{"op":"sync_agent_tmpfilez"}"#).is_err()); } + + /// `sendmsg` `bytes` with every descriptor of `fds` attached as one + /// `SCM_RIGHTS` message: the client side of what `recv_with_fds` reads. + fn send_with_fds(sock: RawFd, bytes: &[u8], fds: &[RawFd]) { + const FD_SIZE: usize = std::mem::size_of::(); + let payload = u32::try_from(FD_SIZE * fds.len()).unwrap(); + let mut ctrl = [0u64; 8]; + let mut iov = libc::iovec { + iov_base: bytes.as_ptr().cast_mut().cast(), + iov_len: bytes.len(), + }; + // SAFETY: msghdr is a plain C struct with no invalid bit patterns. + let mut msg: libc::msghdr = unsafe { std::mem::zeroed() }; + msg.msg_iov = &raw mut iov; + msg.msg_iovlen = 1; + // SAFETY: CMSG_SPACE is a pure size computation. + msg.msg_controllen = unsafe { libc::CMSG_SPACE(payload) } as _; + msg.msg_control = ctrl.as_mut_ptr().cast(); + // SAFETY: `ctrl` is aligned for `cmsghdr` and large enough for the + // header and `fds.len()` descriptors. + unsafe { + let hdr = libc::CMSG_FIRSTHDR(&raw const msg); + (*hdr).cmsg_level = libc::SOL_SOCKET; + (*hdr).cmsg_type = libc::SCM_RIGHTS; + (*hdr).cmsg_len = libc::CMSG_LEN(payload) as _; + std::ptr::copy_nonoverlapping( + fds.as_ptr().cast::(), + libc::CMSG_DATA(hdr), + payload as usize, + ); + assert!(libc::sendmsg(sock, &raw const msg, 0) >= 0, "sendmsg"); + } + } + + fn is_cloexec(fd: &OwnedFd) -> bool { + // SAFETY: `fd` is an open descriptor. + let flags = unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_GETFD) }; + flags >= 0 && flags & libc::FD_CLOEXEC != 0 + } + + fn same_file(fd: &OwnedFd, other: &std::fs::File) -> bool { + let ours = std::fs::File::from(fd.try_clone().unwrap()) + .metadata() + .unwrap(); + let theirs = other.metadata().unwrap(); + (ours.dev(), ours.ino()) == (theirs.dev(), theirs.ino()) + } + + fn dev_null() -> std::fs::File { + std::fs::File::open("/dev/null").expect("open /dev/null") + } + + // ---- socket activation ---- + + /// Only systemd's own handover starts the daemon: at least one fd, and + /// `LISTEN_PID` naming this process. Every other shape is refused, and the + /// refusal says how to run it. + #[test] + fn only_a_handover_addressed_to_this_process_starts_the_daemon() { + check_socket_activation(Some("1"), Some("42"), 42).expect("the normal handover"); + check_socket_activation(Some("3"), Some("42"), 42).expect("more fds is still activation"); + + for (fds, pid, why) in [ + (None, None, "neither variable set"), + (Some("1"), None, "no LISTEN_PID"), + (None, Some("42"), "no LISTEN_FDS"), + (Some("0"), Some("42"), "zero fds"), + (Some("-1"), Some("42"), "negative fds"), + (Some("many"), Some("42"), "unparsable LISTEN_FDS"), + (Some("1"), Some("pid"), "unparsable LISTEN_PID"), + (Some("1"), Some("41"), "LISTEN_PID of another process"), + (Some("1"), Some("-42"), "negative LISTEN_PID"), + (Some("1"), Some(""), "empty LISTEN_PID"), + ] { + let err = check_socket_activation(fds, pid, 42).expect_err(why); + let msg = format!("{err:#}"); + assert!( + msg.contains("requires systemd socket activation"), + "{why}: {msg}" + ); + assert!(msg.contains("/run/hive/priv.sock"), "{why}: {msg}"); + } + } + + // ---- recv_with_fds ---- + + /// Plain bytes come back with no descriptors, and a closed peer reads as + /// zero bytes. + #[test] + fn recv_reads_plain_bytes_and_end_of_stream() { + let (a, b) = std::os::unix::net::UnixStream::pair().unwrap(); + (&a).write_all(b"hello").unwrap(); + drop(a); + let mut buf = [0u8; 16]; + let (n, fds) = recv_with_fds(b.as_raw_fd(), &mut buf).unwrap(); + assert_eq!((&buf[..n], fds.len()), (&b"hello"[..], 0)); + let (n, fds) = recv_with_fds(b.as_raw_fd(), &mut buf).unwrap(); + assert_eq!((n, fds.len()), (0, 0)); + } + + /// A descriptor sent with the bytes arrives as the same open file, and + /// close-on-exec: without that every `btrfs` child the helper spawns + /// would inherit it. + #[test] + fn recv_returns_the_passed_descriptor_close_on_exec() { + let (a, b) = std::os::unix::net::UnixStream::pair().unwrap(); + let file = dev_null(); + send_with_fds(a.as_raw_fd(), b"x", &[file.as_raw_fd()]); + let mut buf = [0u8; 16]; + let (n, mut fds) = recv_with_fds(b.as_raw_fd(), &mut buf).unwrap(); + assert_eq!(n, 1); + assert_eq!(fds.len(), 1); + let fd = fds.pop().unwrap(); + assert!(is_cloexec(&fd), "a received descriptor must be CLOEXEC"); + assert!(same_file(&fd, &file)); + } + + /// Every descriptor of the message is claimed, even though the protocol + /// takes one: an unclaimed one would leak for the life of the process. + #[test] + fn recv_claims_every_descriptor_of_a_message() { + let (a, b) = std::os::unix::net::UnixStream::pair().unwrap(); + let (f1, f2) = (dev_null(), dev_null()); + send_with_fds(a.as_raw_fd(), b"x", &[f1.as_raw_fd(), f2.as_raw_fd()]); + let mut buf = [0u8; 16]; + let (_, fds) = recv_with_fds(b.as_raw_fd(), &mut buf).unwrap(); + assert_eq!(fds.len(), 2); + assert!(fds.iter().all(is_cloexec)); + } + + /// A read on a descriptor that is not a socket is the OS error, not a + /// panic or an empty read. + #[test] + fn recv_on_a_bad_descriptor_is_an_error() { + let mut buf = [0u8; 4]; + assert!(recv_with_fds(-1, &mut buf).is_err()); + } + + // ---- Requests: line framing and descriptor pairing ---- + + fn requests(sock: &UnixStream) -> Requests<'_> { + Requests { + sock, + buf: Vec::new(), + fd: None, + } + } + + /// Lines are split at the newline, several in one read come out in + /// order, and end of stream is `None`. + #[tokio::test] + async fn requests_split_lines_in_order_and_end_at_eof() { + let (mut client, server) = UnixStream::pair().unwrap(); + client.write_all(b"one\ntwo\n\nthree\n").await.unwrap(); + drop(client); + let mut reqs = requests(&server); + for want in ["one", "two", "", "three"] { + let (line, fd) = reqs.next().await.unwrap().unwrap(); + assert_eq!(line, want); + assert!(fd.is_none()); + } + assert!(reqs.next().await.unwrap().is_none()); + } + + /// A line that arrives in pieces is one request. + #[tokio::test] + async fn requests_join_a_line_split_across_reads() { + let (mut client, server) = UnixStream::pair().unwrap(); + let writer = tokio::spawn(async move { + client.write_all(b"{\"op\":\"daemon").await.unwrap(); + client.flush().await.unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + client.write_all(b"_reload\"}\n").await.unwrap(); + }); + let mut reqs = requests(&server); + let (line, _) = reqs.next().await.unwrap().unwrap(); + assert_eq!(line, r#"{"op":"daemon_reload"}"#); + writer.await.unwrap(); + } + + /// A connection that closes before the newline is an error naming how + /// much was cut off, never a request. + #[tokio::test] + async fn requests_refuse_a_line_cut_off_by_eof() { + let (mut client, server) = UnixStream::pair().unwrap(); + client.write_all(b"{\"op\":").await.unwrap(); + drop(client); + let err = requests(&server).next().await.unwrap_err(); + assert!( + format!("{err:#}").contains("closed mid-request (6 bytes)"), + "{err:#}" + ); + } + + /// A line that is not UTF-8 is refused outright. + #[tokio::test] + async fn requests_refuse_a_line_that_is_not_utf8() { + let (mut client, server) = UnixStream::pair().unwrap(); + client.write_all(b"\xff\xfe\n").await.unwrap(); + let err = requests(&server).next().await.unwrap_err(); + assert!(format!("{err:#}").contains("not valid UTF-8"), "{err:#}"); + } + + /// The descriptor that came with a line is handed out with that line + /// and with no other. + #[tokio::test] + async fn requests_pair_a_descriptor_with_its_line_only() { + let (client, server) = UnixStream::pair().unwrap(); + let file = dev_null(); + send_with_fds(client.as_raw_fd(), b"first\n", &[file.as_raw_fd()]); + send_with_fds(client.as_raw_fd(), b"second\n", &[]); + let mut reqs = requests(&server); + let (line, fd) = reqs.next().await.unwrap().unwrap(); + assert_eq!(line, "first"); + assert!(same_file( + &fd.expect("the descriptor rides with its line"), + &file + )); + let (line, fd) = reqs.next().await.unwrap().unwrap(); + assert_eq!(line, "second"); + assert!(fd.is_none(), "the descriptor must not carry over"); + } + + /// A second descriptor before the first one's line is complete is a + /// protocol error, not something to queue. + #[tokio::test] + async fn requests_refuse_a_second_descriptor_for_one_line() { + let (client, server) = UnixStream::pair().unwrap(); + let file = dev_null(); + send_with_fds(client.as_raw_fd(), b"par", &[file.as_raw_fd()]); + send_with_fds(client.as_raw_fd(), b"tial\n", &[file.as_raw_fd()]); + let err = requests(&server).next().await.unwrap_err(); + assert!( + format!("{err:#}").contains("more than one file descriptor"), + "{err:#}" + ); + } + + // ---- dispatch: parsing, fd agreement, validation, error mapping ---- + + /// Success carries the handler's output and no error; failure carries the + /// whole context chain and no output. + #[test] + fn the_outcome_maps_to_the_wire_response() { + let ok = response_of(Ok(("out".to_owned(), "err".to_owned()))); + assert!(ok.ok); + assert_eq!((ok.stdout.as_str(), ok.stderr.as_str()), ("out", "err")); + assert!(ok.error.is_none()); + + let failed = response_of(Err(anyhow::anyhow!("root cause").context("outer step"))); + assert!(!failed.ok); + assert!(failed.stdout.is_empty() && failed.stderr.is_empty()); + assert_eq!(failed.error.as_deref(), Some("outer step: root cause")); + } + + /// A write half nobody reads, for ops that would stream. + fn sink() -> (tokio::net::unix::OwnedWriteHalf, UnixStream) { + let (ours, theirs) = UnixStream::pair().unwrap(); + (ours.into_split().1, theirs) + } + + /// The failed response for `line`, asserting nothing else came back. + async fn refused(line: &str, fd: Option) -> String { + let (mut w, _keep) = sink(); + let resp = dispatch(line, fd, &mut w).await; + assert!(!resp.ok, "{line} was accepted"); + assert!(resp.stdout.is_empty() && resp.stderr.is_empty(), "{line}"); + resp.error.expect("a failure carries its error") + } + + /// A line that is not a request, or names no op, is answered with the + /// parse error and runs nothing. + #[tokio::test] + async fn malformed_and_unknown_requests_are_refused_as_parse_errors() { + for line in [ + "", + "not json", + "{", + "[]", + "null", + r#"{"op":"rm_rf"}"#, + r#"{"op":"start_container"}"#, + r#"{"name":"bob"}"#, + r#"{"op":"start_container","name":7}"#, + r#"{"op":"control_infra_container","container":"Forge","action":"stop"}"#, + ] { + let msg = refused(line, None).await; + assert!(msg.starts_with("parse request"), "{line}: {msg}"); + } + } + + /// A descriptor on an op that takes none, and a missing one on an op that + /// needs it, are answered with the refusal before the op runs. + #[tokio::test] + async fn a_fd_sent_to_an_op_that_takes_none_is_refused_before_it_runs() { + // The name is bad too: only the descriptor check can produce this + // message, so the op's own validation was not reached. + let msg = refused( + r#"{"op":"set_agent_paused","agent_name":"../x","paused":true}"#, + Some(dev_null().into()), + ) + .await; + assert!( + msg.contains("does not take a passed file descriptor"), + "{msg}" + ); + + let msg = refused( + r#"{"op":"send_agent_snapshot_to_fd","agent_name":"atlas","snapshot_name":"hive-x","parent_snapshot_name":null}"#, + None, + ) + .await; + assert!(msg.contains("requires a passed file descriptor"), "{msg}"); + } + + /// Every op that names an agent or container refuses a bad name from the + /// request line itself, before any command runs or path is built. A + /// control (a valid name on the same op) is in the lifecycle and storage + /// tests; here the point is that each wire op reaches its validator. + #[tokio::test] + async fn every_named_op_refuses_a_bad_name_from_the_wire() { + let store = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-bob"; + let bad = "../x"; + let named = |op: &str, field: &str, extra: &str| { + format!(r#"{{"op":"{op}","{field}":"{bad}"{extra}}}"#) + }; + let mut cases: Vec<(String, &str)> = Vec::new(); + for (op, field) in [ + ("start_container", "name"), + ("stop_container", "name"), + ("kill_container", "name"), + ("destroy_container", "name"), + ("ensure_agent_socket_dir", "name"), + ("ensure_agent_subvolume", "agent_name"), + ("delete_agent_subvolume", "agent_name"), + ("read_subvolume_usage", "agent_name"), + ("measure_agent_disk", "agent_name"), + ("upgrade_agent_subvolume", "agent_name"), + ] { + cases.push((named(op, field, ""), "invalid name")); + } + cases.push(( + named("start_failure_log", "name", r#","since_unix":0"#), + "invalid name", + )); + for op in ["update_container", "create_container"] { + let extra = format!(r#","toplevel":"{store}""#); + cases.push((named(op, "name", &extra), "invalid name")); + } + cases.push(( + format!(r#"{{"op":"realise_closure","agent":"{bad}","store_path":"{store}"}}"#), + "invalid name", + )); + cases.push(( + named("set_agent_paused", "agent_name", r#","paused":true"#), + "invalid name", + )); + cases.push(( + named( + "set_subvolume_quota", + "agent_name", + r#","limit_bytes":null"#, + ), + "invalid name", + )); + for op in ["snapshot_agent_subvolume", "delete_agent_snapshot"] { + let extra = r#","snapshot_name":"hive-x""#; + cases.push((named(op, "agent_name", extra), "invalid name")); + } + cases.push(( + named( + "send_agent_snapshot_to_file", + "agent_name", + r#","snapshot_name":"hive-x","parent_snapshot_name":null,"dest_file_name":"out""#, + ), + "invalid name", + )); + // The container-name ops take the `h-` system name. + for name in ["h-../x", "evil", "hive-forge", "h-"] { + cases.push(( + format!(r#"{{"op":"remove_service_dropin","container":"{name}"}}"#), + if name.starts_with("h-") { + "invalid name" + } else { + "not managed by hive" + }, + )); + } + cases.push(( + r#"{"op":"write_resource_limits","container":"evil","memory_max":"1G","cpu_quota":"100%"}"#.to_owned(), + "not managed by hive", + )); + cases.push(( + r#"{"op":"write_nspawn_flags","container":"evil","binds":[],"isolation":{"bridge":"br","gateway_ip":"10.0.0.1"}}"#.to_owned(), + "not managed by hive", + )); + + for (line, want) in &cases { + let msg = refused(line, None).await; + assert!(msg.contains(want), "{line}: {msg}"); + } + // The fd op validates before it looks for its descriptor. + let msg = refused( + &named( + "send_agent_snapshot_to_fd", + "agent_name", + r#","snapshot_name":"hive-x","parent_snapshot_name":null"#, + ), + Some(dev_null().into()), + ) + .await; + assert!(msg.contains("invalid name"), "{msg}"); + } + + /// Snapshot labels and the export file name are validated from the wire + /// too, for a valid agent: a label must carry the `hive-` prefix and the + /// credential charset, so no label or file name can climb out of the + /// agent's directory. + #[tokio::test] + async fn snapshot_labels_and_export_names_are_validated_from_the_wire() { + let snap = |op: &str, label: &str| { + format!(r#"{{"op":"{op}","agent_name":"atlas","snapshot_name":"{label}"}}"#) + }; + for op in ["snapshot_agent_subvolume", "delete_agent_snapshot"] { + for label in ["", "x", "../hive-x", "hive-../x", "hive-a/b", "hive-a.b"] { + let msg = refused(&snap(op, label), None).await; + assert!(msg.contains("invalid"), "{op} {label:?}: {msg}"); + } + } + let export = |label: &str, parent: &str, dest: &str| { + format!( + r#"{{"op":"send_agent_snapshot_to_file","agent_name":"atlas","snapshot_name":"{label}","parent_snapshot_name":{parent},"dest_file_name":"{dest}"}}"# + ) + }; + for (label, parent, dest) in [ + ("../x", "null", "out"), + ("hive-x", r#""../x""#, "out"), + ("hive-x", "null", "../out"), + ("hive-x", "null", "a/b"), + ("hive-x", "null", ""), + ("hive-x", "null", "out.img"), + ] { + let msg = refused(&export(label, parent, dest), None).await; + assert!(msg.contains("invalid"), "{label} {parent} {dest}: {msg}"); + } + let fd_op = r#"{"op":"send_agent_snapshot_to_fd","agent_name":"atlas","snapshot_name":"hive-x","parent_snapshot_name":"../x"}"#; + let msg = refused(fd_op, Some(dev_null().into())).await; + assert!(msg.contains("invalid"), "{msg}"); + } + + /// The error is the whole anyhow chain, so a caller sees the cause and + /// the context in one string. + #[tokio::test] + async fn the_error_carries_the_whole_context_chain() { + let msg = refused(r#"{"op":"start_container","name":7}"#, None).await; + assert!(msg.starts_with("parse request: "), "{msg}"); + assert!(msg.contains("invalid type"), "{msg}"); + } + + // ---- handle: the whole connection ---- + + /// Every response line of a connection, in order, once the server closes. + async fn serve(input: &[u8]) -> Vec { + let (mut client, server) = UnixStream::pair().unwrap(); + let task = tokio::spawn(handle(server)); + client.write_all(input).await.unwrap(); + client.shutdown().await.unwrap(); + let mut raw = String::new(); + client.read_to_string(&mut raw).await.unwrap(); + task.await.unwrap(); + raw.lines() + .map(|l| { + // The wire is bare-`PrivResponse` compatible (the event is + // untagged), and also a `Done` to a new client. + let bare: PrivResponse = serde_json::from_str(l).unwrap(); + match serde_json::from_str::(l).unwrap() { + PrivEvent::Done(done) => assert_eq!(done.ok, bare.ok), + PrivEvent::Line(_) => panic!("unexpected stream line: {l}"), + } + bare + }) + .collect() + } + + /// One connection answers each request line with one response line, in + /// order, and a refused request does not close the connection. + #[tokio::test] + async fn a_connection_answers_each_line_and_survives_a_refusal() { + let resps = + serve(b"garbage\n{\"op\":\"start_container\",\"name\":\"../x\"}\n{\"op\":\"nope\"}\n") + .await; + assert_eq!(resps.len(), 3); + let errs: Vec<_> = resps.iter().map(|r| r.error.as_deref().unwrap()).collect(); + assert!(errs[0].starts_with("parse request"), "{errs:?}"); + assert!(errs[1].contains("invalid name"), "{errs:?}"); + assert!(errs[2].starts_with("parse request"), "{errs:?}"); + assert!(resps.iter().all(|r| !r.ok)); + } + + /// A connection that dies mid-line or sends non-UTF-8 gets no response + /// and nothing runs. + #[tokio::test] + async fn a_broken_request_closes_the_connection_without_a_response() { + assert!(serve(b"{\"op\":\"start_cont").await.is_empty()); + assert!(serve(b"\xff\xfe\n").await.is_empty()); + assert!(serve(b"").await.is_empty()); + } + + /// The descriptor check holds end to end: a stray descriptor on a + /// request that takes none is answered with the refusal. + #[tokio::test] + async fn a_stray_descriptor_is_refused_over_the_wire() { + let (client, server) = UnixStream::pair().unwrap(); + let task = tokio::spawn(handle(server)); + let file = dev_null(); + send_with_fds( + client.as_raw_fd(), + b"{\"op\":\"set_agent_paused\",\"agent_name\":\"../x\",\"paused\":true}\n", + &[file.as_raw_fd()], + ); + let (mut rd, mut wr) = client.into_split(); + wr.shutdown().await.unwrap(); + let mut raw = String::new(); + rd.read_to_string(&mut raw).await.unwrap(); + task.await.unwrap(); + let resp: PrivResponse = serde_json::from_str(raw.trim()).unwrap(); + assert!(!resp.ok); + assert!( + resp.error + .unwrap() + .contains("does not take a passed file descriptor"), + "{raw}" + ); + } }