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

@@ -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<PrivResponse> {
-    let mut stream = UnixStream::connect(PRIV_SOCK)
+    call_at(Path::new(PRIV_SOCK), req).await
+}
+
+async fn call_at(sock: &Path, req: &PrivRequest) -> Result<PrivResponse> {
+    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<usize>
 /// attached, or hive-priv answers with something other than a terminal
 /// event.
 pub async fn call_with_fd(req: &PrivRequest, fd: OwnedFd) -> Result<PrivResponse> {
-    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<PrivResponse> {
+    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<PrivResponse
 /// `PrivEvent::Line` as it arrives, then returning the terminal
 /// `PrivResponse`. Used for long-running ops (`create` / `update`).
 pub async fn call_streaming(
+    req: &PrivRequest,
+    on_line: impl FnMut(PrivStream, &str),
+) -> Result<PrivResponse> {
+    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<PrivResponse> {
-    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<String>,
+    ) {
+        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<u8>, 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::<RawFd>());
+            (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");
+    }
 }