From 0bc0b86d1d2c996d40dcb232d7b16ace8ac5defa Mon Sep 17 00:00:00 2001 From: damocles Date: Fri, 6 Nov 2026 18:59:24 +0100 Subject: [PATCH] hive-c0re: priv_client can dial any socket path, so its wire path is tested against a fake hive-priv --- hive-c0re/src/priv_client.rs | 197 ++++++++++++++++++++++++++++++++++- 1 file changed, 193 insertions(+), 4 deletions(-) diff --git a/hive-c0re/src/priv_client.rs b/hive-c0re/src/priv_client.rs index 962a879c..a742b5de 100644 --- a/hive-c0re/src/priv_client.rs +++ b/hive-c0re/src/priv_client.rs @@ -13,6 +13,7 @@ use hive_priv_sock::{ }; use hive_types::NixBuilds; use std::os::fd::{AsRawFd as _, OwnedFd, RawFd}; +use std::path::Path; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Interest}; use tokio::net::UnixStream; @@ -20,7 +21,11 @@ use tokio::net::UnixStream; /// Send a single request to `hive-priv` and return the response. /// For streaming ops use `call_streaming` instead. pub async fn call(req: &PrivRequest) -> Result { - let mut stream = UnixStream::connect(PRIV_SOCK) + call_at(Path::new(PRIV_SOCK), req).await +} + +async fn call_at(sock: &Path, req: &PrivRequest) -> Result { + let mut stream = UnixStream::connect(sock) .await .context("connect to hive-priv socket")?; let line = serde_json::to_string(req).context("serialise PrivRequest")? + "\n"; @@ -139,7 +144,11 @@ fn send_with_fd(sock: RawFd, bytes: &[u8], fd: RawFd) -> std::io::Result /// attached, or hive-priv answers with something other than a terminal /// event. pub async fn call_with_fd(req: &PrivRequest, fd: OwnedFd) -> Result { - let mut stream = UnixStream::connect(PRIV_SOCK) + call_with_fd_at(Path::new(PRIV_SOCK), req, fd).await +} + +async fn call_with_fd_at(path: &Path, req: &PrivRequest, fd: OwnedFd) -> Result { + let mut stream = UnixStream::connect(path) .await .context("connect to hive-priv socket (fd-passing)")?; let line = serde_json::to_string(req).context("serialise PrivRequest")? + "\n"; @@ -181,10 +190,18 @@ pub async fn call_with_fd(req: &PrivRequest, fd: OwnedFd) -> Result Result { + call_streaming_at(Path::new(PRIV_SOCK), req, on_line).await +} + +async fn call_streaming_at( + path: &Path, req: &PrivRequest, mut on_line: impl FnMut(PrivStream, &str), ) -> Result { - let mut stream = UnixStream::connect(PRIV_SOCK) + let mut stream = UnixStream::connect(path) .await .context("connect to hive-priv socket (streaming)")?; let line = serde_json::to_string(req).context("serialise PrivRequest")? + "\n"; @@ -666,10 +683,14 @@ fn ok(resp: PrivResponse) -> Result<()> { mod tests { use super::{ BindMount, CredentialMount, HiveFacts, NetworkIsolation, NixBuilds, PrivRequest, - PrivResponse, check, create_container_request, parse_qgroup_usage, resource_limits_request, + PrivResponse, PrivStream, call_at, call_streaming_at, call_with_fd_at, check, + create_container_request, parse_qgroup_usage, resource_limits_request, send_snapshot_request, update_container_request, within, write_nspawn_flags_request, }; use serde_json::{Value, json}; + use std::io::{Read as _, Write as _}; + use std::os::fd::{AsRawFd as _, FromRawFd as _, OwnedFd, RawFd}; + use std::os::unix::net::{UnixListener, UnixStream}; use std::time::Duration; fn wire(req: &PrivRequest) -> Value { @@ -841,4 +862,172 @@ mod tests { let got = within(Duration::from_secs(5), async { Ok(7_u64) }).await; assert_eq!(got.expect("answered in time"), 7); } + + /// A stand-in for hive-priv on a tempdir socket: it accepts one + /// connection, reads the request until the client shuts down its write + /// half, writes `reply`, and hands back what it read. + fn fake_priv( + reply: &'static str, + ) -> ( + tempfile::TempDir, + std::path::PathBuf, + std::thread::JoinHandle, + ) { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("priv.sock"); + let listener = UnixListener::bind(&path).expect("bind"); + let server = std::thread::spawn(move || { + let (mut conn, _) = listener.accept().expect("accept"); + let mut got = String::new(); + conn.read_to_string(&mut got).expect("read request"); + conn.write_all(reply.as_bytes()).expect("write reply"); + got + }); + (dir, path, server) + } + + const DONE: &str = "{\"ok\":true,\"stdout\":\"hi\",\"stderr\":\"\"}\n"; + + #[tokio::test] + async fn a_call_sends_one_json_line_and_returns_the_done_event() { + let (_dir, path, server) = fake_priv(DONE); + let resp = call_at(&path, &PrivRequest::DaemonReload) + .await + .expect("answered"); + assert!(resp.ok); + assert_eq!(resp.stdout, "hi"); + let sent = server.join().expect("server"); + assert_eq!(sent, "{\"op\":\"daemon_reload\"}\n"); + } + + #[tokio::test] + async fn a_call_to_a_missing_socket_names_the_connect_step() { + let dir = tempfile::tempdir().expect("tempdir"); + let err = call_at(&dir.path().join("none.sock"), &PrivRequest::DaemonReload) + .await + .expect_err("nothing listens there"); + assert!( + err.to_string().contains("connect to hive-priv socket"), + "{err}" + ); + } + + #[tokio::test] + async fn a_stream_line_answering_a_plain_call_is_an_error() { + let (_dir, path, server) = fake_priv("{\"stream\":\"stdout\",\"data\":\"x\"}\n"); + let err = call_at(&path, &PrivRequest::DaemonReload) + .await + .expect_err("a non-streaming op must end in Done"); + assert!(err.to_string().contains("unexpected stream line"), "{err}"); + server.join().expect("server"); + } + + #[tokio::test] + async fn a_garbled_answer_is_a_parse_error() { + let (_dir, path, server) = fake_priv("not json\n"); + let err = call_at(&path, &PrivRequest::DaemonReload) + .await + .expect_err("not an event"); + assert!(err.to_string().contains("parse PrivResponse"), "{err}"); + server.join().expect("server"); + } + + #[tokio::test] + async fn streaming_hands_each_line_over_in_order_before_the_done_event() { + let (_dir, path, server) = fake_priv( + "{\"stream\":\"stdout\",\"data\":\"one\"}\n\ + {\"stream\":\"stderr\",\"data\":\"two\"}\n\ + {\"ok\":true,\"stdout\":\"\",\"stderr\":\"\"}\n", + ); + let mut seen = Vec::new(); + let resp = call_streaming_at(&path, &PrivRequest::DaemonReload, |s, d| { + seen.push((s, d.to_owned())); + }) + .await + .expect("done"); + assert!(resp.ok); + assert_eq!( + seen, + vec![ + (PrivStream::Stdout, "one".to_owned()), + (PrivStream::Stderr, "two".to_owned()), + ] + ); + server.join().expect("server"); + } + + #[tokio::test] + async fn streaming_that_ends_without_a_done_event_is_an_error() { + let (_dir, path, server) = fake_priv("{\"stream\":\"stdout\",\"data\":\"one\"}\n"); + let err = call_streaming_at(&path, &PrivRequest::DaemonReload, |_, _| {}) + .await + .expect_err("closed early"); + assert!(err.to_string().contains("before sending Done"), "{err}"); + server.join().expect("server"); + } + + /// `recvmsg` the request and the one descriptor that rides with it. + fn recv_with_fd(conn: &UnixStream) -> (Vec, OwnedFd) { + let mut buf = [0_u8; 256]; + let mut iov = libc::iovec { + iov_base: buf.as_mut_ptr().cast(), + iov_len: buf.len(), + }; + // u64 elements give the control buffer the 8-byte alignment a + // `cmsghdr` read needs; a byte array would be misaligned. + let mut cmsg = [0_u64; 8]; + // SAFETY: an all-zero msghdr is valid; every pointer set below + // refers to a buffer that outlives the recvmsg call. + let (n, fd) = unsafe { + let mut msg: libc::msghdr = std::mem::zeroed(); + msg.msg_iov = &raw mut iov; + msg.msg_iovlen = 1; + msg.msg_control = cmsg.as_mut_ptr().cast(); + msg.msg_controllen = std::mem::size_of_val(&cmsg); + let n = libc::recvmsg(conn.as_raw_fd(), &raw mut msg, 0); + assert!(n > 0, "recvmsg failed"); + let hdr = libc::CMSG_FIRSTHDR(&raw const msg); + assert!(!hdr.is_null(), "no ancillary data arrived"); + assert_eq!((*hdr).cmsg_level, libc::SOL_SOCKET); + assert_eq!((*hdr).cmsg_type, libc::SCM_RIGHTS); + let fd = std::ptr::read_unaligned(libc::CMSG_DATA(hdr).cast::()); + (n, OwnedFd::from_raw_fd(fd)) + }; + (buf[..usize::try_from(n).expect("positive")].to_vec(), fd) + } + + #[tokio::test] + async fn a_descriptor_arrives_with_the_request_and_is_closed_on_our_side() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("priv.sock"); + let listener = UnixListener::bind(&path).expect("bind"); + let server = std::thread::spawn(move || { + let (mut conn, _) = listener.accept().expect("accept"); + let (request, fd) = recv_with_fd(&conn); + let mut rest = String::new(); + conn.read_to_string(&mut rest).expect("read to EOF"); + conn.write_all(DONE.as_bytes()).expect("reply"); + (request, fd) + }); + + let (mut read_end, write_end) = std::io::pipe().expect("pipe"); + let resp = call_with_fd_at(&path, &PrivRequest::DaemonReload, write_end.into()) + .await + .expect("answered"); + assert!(resp.ok); + + let (request, received) = server.join().expect("server"); + assert_eq!(request, b"{\"op\":\"daemon_reload\"}\n"); + // The descriptor that arrived is the pipe's write end, and our + // copy is already closed: writing through the received one and + // then dropping it is what ends the read. + let mut write_end = std::fs::File::from(received); + write_end.write_all(b"ping").expect("write via received fd"); + drop(write_end); + let mut out = String::new(); + read_end + .read_to_string(&mut out) + .expect("EOF after both ends closed"); + assert_eq!(out, "ping"); + } }