//! Minimal privileged helper for hive-c0re.
//!
//! Runs as root. Exposes a narrow unix socket at `/run/hive/priv.sock`
//! that accepts `PrivRequest` JSON lines and executes only the
//! operations that genuinely require root. All coordination logic,
//! broker, HTTP, and scheduling stay in the unprivileged hive-c0re
//! process.
//!
//! **Security model**: every request is validated against a strict
//! container-name allowlist before any filesystem or process operation.
//! Only containers whose names match the hive convention (`h-*`,
//! the manager container included) are accepted. Every variant maps to a single known operation — no
//! arbitrary command pass-through.
//!
//! **Socket activation**: when systemd passes the listener socket via
//! `LISTEN_FDS=1` + `LISTEN_PID=<self>`, the inherited fd 3 is used
//! instead of binding a fresh socket.
//!
//! **Layout**: this file holds the socket loop and the one-arm-per-op
//! dispatch table. Each op family's handlers and validation live in their
//! own module: [`container`] (lifecycle), [`container_config`] (nspawn flags,
//! systemd drop-ins), [`rootfs`] (the salvage of a container's root before it
//! turns ephemeral, with [`salvage_io`] for its space, reflink and lock calls),
//! [`prestart`] (the check of a container's root that runs before each start,
//! as a sub-command), [`storage`] (btrfs subvolumes, quotas, snapshots, disk
//! measurement), [`files`] (pause marker, socket dir, legacy tmpfiles),
//! [`infra`] (gateway nginx, infra containers) and [`closure`] (downloads from
//! the swarm cache). Helpers more than one family uses are in [`shared`].

mod closure;
mod container;
mod container_config;
mod files;
mod infra;
mod prestart;
mod rootfs;
mod salvage_io;
mod shared;
mod storage;
mod userns;

use std::os::fd::{AsRawFd as _, FromRawFd as _, OwnedFd, RawFd};

use anyhow::{Context as _, Result, bail};
use hive_priv_sock::{PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse};
use tokio::io::AsyncWriteExt;
use tokio::net::unix::OwnedWriteHalf;
use tokio::net::{UnixListener, UnixStream};

#[tokio::main]
async fn main() -> Result<()> {
    // `hive-priv check-rootfs <container>` is a container unit's `ExecStartPre=`,
    // not the daemon: it answers with an exit code and never touches the socket.
    let args: Vec<String> = std::env::args().skip(1).collect();
    if let Some(code) = prestart::run(&args) {
        std::process::exit(code);
    }

    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()),
        )
        // This is a systemd-managed daemon — stdout always goes to journald,
        // never a human terminal, and journald doesn't strip ANSI escapes:
        // they land in victorialogs as raw byte-array spam otherwise.
        .with_ansi(false)
        .init();

    let listener = socket_listener()?;
    if let Err(e) = files::remove_legacy_tmpfiles() {
        tracing::warn!(error = %format!("{e:#}"), "legacy tmpfiles cleanup failed");
    }
    tracing::info!("hive-priv listening");

    loop {
        match listener.accept().await {
            Ok((stream, _)) => {
                tokio::spawn(handle(stream));
            }
            Err(e) => {
                tracing::error!(error = %e, "accept failed");
            }
        }
    }
}

/// 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<i32> = listen_fds.and_then(|s| s.parse().ok());
    let listen_pid: Option<u32> = listen_pid.and_then(|s| s.parse().ok());

    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 + \
             LISTEN_PID=<self> for {PRIV_SOCK}); run it via the hive-priv.socket unit, \
             not directly"
        );
    }
    Ok(())
}

fn socket_listener() -> Result<UnixListener> {
    // 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 {
        use std::os::unix::io::FromRawFd;
        std::os::unix::net::UnixListener::from_raw_fd(3)
    };
    std_listener
        .set_nonblocking(true)
        .context("set socket non-blocking")?;
    let listener =
        tokio::net::UnixListener::from_std(std_listener).context("wrap systemd socket")?;
    tracing::info!("using systemd-activated socket");
    Ok(listener)
}

/// Ancillary-data buffer sized and aligned for one `SCM_RIGHTS` message.
///
/// `CMSG_SPACE` is not a `const fn`, so the size is a literal with room
/// to spare (24 bytes are needed for a single descriptor on x86-64). The
/// union member gives the `cmsghdr` alignment `CMSG_FIRSTHDR` requires —
/// a bare `[u8; N]` is only byte-aligned and would be undefined behaviour
/// to walk.
#[repr(C)]
union CmsgSpace {
    _align: libc::cmsghdr,
    bytes: [u8; 32],
}

/// One `recvmsg` into `buf`, returning the bytes read plus any file
/// descriptors that rode along as `SCM_RIGHTS`.
///
/// Why not a plain read: ancillary data is attached to a *specific*
/// `recvmsg` call, so a buffered line reader cannot surface it — it
/// reads the bytes and silently drops the descriptor.
///
/// `MSG_CMSG_CLOEXEC` is not optional: without it a received descriptor
/// is inherited by every `btrfs` / `nixos-container` child this helper
/// later spawns.
fn recv_with_fds(sock: RawFd, buf: &mut [u8]) -> std::io::Result<(usize, Vec<OwnedFd>)> {
    const FD_SIZE: usize = std::mem::size_of::<RawFd>();

    let mut iov = libc::iovec {
        iov_base: buf.as_mut_ptr().cast(),
        iov_len: buf.len(),
    };
    let mut cmsg = CmsgSpace { bytes: [0; 32] };
    // SAFETY: msghdr is a plain C struct with no invalid bit patterns;
    // every field we care about is set immediately below.
    let mut msg: libc::msghdr = unsafe { std::mem::zeroed() };
    msg.msg_iov = &raw mut iov;
    msg.msg_iovlen = 1;
    msg.msg_control = std::ptr::addr_of_mut!(cmsg.bytes).cast();
    msg.msg_controllen = 32;

    // SAFETY: `msg` points at a live iovec covering `buf` and a live,
    // correctly aligned control buffer of the length we just declared.
    let n = unsafe { libc::recvmsg(sock, &raw mut msg, libc::MSG_CMSG_CLOEXEC) };
    if n < 0 {
        return Err(std::io::Error::last_os_error());
    }

    // Take ownership of every descriptor the kernel attached, even ones
    // this protocol never expects: an `OwnedFd` we drop is closed, an
    // fd we fail to claim is leaked for the lifetime of the process.
    let mut fds = Vec::new();
    // SAFETY: `msg` was just filled in by a successful `recvmsg`.
    let mut cmsgp = unsafe { libc::CMSG_FIRSTHDR(&raw const msg) };
    while !cmsgp.is_null() {
        // SAFETY: CMSG_FIRSTHDR / CMSG_NXTHDR only ever return a pointer
        // to a complete header inside the control buffer.
        let hdr = unsafe { std::ptr::read_unaligned(cmsgp) };
        if hdr.cmsg_level == libc::SOL_SOCKET && hdr.cmsg_type == libc::SCM_RIGHTS {
            // SAFETY: same, and CMSG_LEN(0) is the header's own length.
            let payload = hdr.cmsg_len as usize - unsafe { libc::CMSG_LEN(0) } as usize;
            let count = payload / FD_SIZE;
            // SAFETY: CMSG_DATA points at `payload` bytes of descriptors.
            let data = unsafe { libc::CMSG_DATA(cmsgp) };
            for i in 0..count {
                // Copied out byte-wise rather than read through a
                // `*const RawFd`: the control buffer is only guaranteed
                // `cmsghdr`-aligned, so casting to a more strictly
                // aligned pointer would be unsound even where it happens
                // to work.
                let mut raw = [0u8; FD_SIZE];
                // SAFETY: i < count, so this reads inside the payload.
                unsafe {
                    std::ptr::copy_nonoverlapping(data.add(i * FD_SIZE), raw.as_mut_ptr(), FD_SIZE);
                }
                // SAFETY: the kernel just created this descriptor for
                // us — we are its only owner.
                fds.push(unsafe { OwnedFd::from_raw_fd(RawFd::from_ne_bytes(raw)) });
            }
        }
        // SAFETY: `cmsgp` came from this same message.
        cmsgp = unsafe { libc::CMSG_NXTHDR(&raw const msg, cmsgp) };
    }

    // `n >= 0` was checked above, so the conversion cannot fail; going
    // through `try_from` keeps it a cast-free, lint-clean widening.
    let read = usize::try_from(n).unwrap_or_default();
    Ok((read, fds))
}

/// Reads newline-delimited requests off one connection, pairing each
/// with the descriptor that arrived with it.
///
/// The pairing is deliberately trivial, because the protocol is:
/// `hive-sock-client` connects per request, so a connection carries one
/// line and at most one descriptor. The loop below still handles several
/// sequential requests (the server always has), but it refuses to guess
/// — a second descriptor arriving before its line is a protocol error,
/// not something to queue and hope about.
struct Requests<'a> {
    sock: &'a UnixStream,
    buf: Vec<u8>,
    fd: Option<OwnedFd>,
}

impl Requests<'_> {
    /// Next complete request line and its descriptor, or `None` at EOF.
    async fn next(&mut self) -> Result<Option<(String, Option<OwnedFd>)>> {
        loop {
            if let Some(nl) = self.buf.iter().position(|&b| b == b'\n') {
                let line: Vec<u8> = self.buf.drain(..=nl).take(nl).collect();
                let line = String::from_utf8(line).context("request line was not valid UTF-8")?;
                return Ok(Some((line, self.fd.take())));
            }

            let mut chunk = [0u8; 8192];
            let raw = self.sock.as_raw_fd();
            let (n, fds) = self
                .sock
                .async_io(tokio::io::Interest::READABLE, || {
                    recv_with_fds(raw, &mut chunk)
                })
                .await
                .context("recvmsg on the priv socket")?;

            for fd in fds {
                if self.fd.replace(fd).is_some() {
                    bail!("more than one file descriptor passed for a single request");
                }
            }
            if n == 0 {
                if !self.buf.is_empty() {
                    bail!("connection closed mid-request ({} bytes)", self.buf.len());
                }
                return Ok(None);
            }
            self.buf.extend_from_slice(&chunk[..n]);
        }
    }
}

async fn handle(stream: UnixStream) {
    let (reader, mut writer) = stream.into_split();
    let mut requests = Requests {
        sock: reader.as_ref(),
        buf: Vec::new(),
        fd: None,
    };
    loop {
        let (line, fd) = match requests.next().await {
            Ok(Some(req)) => req,
            Ok(None) => break,
            Err(e) => {
                tracing::warn!(error = %format!("{e:#}"), "reading request failed");
                break;
            }
        };
        let resp = dispatch(&line, fd, &mut writer).await;
        // Write the terminal PrivResponse as a PrivEvent::Done. Wire-identical
        // to a bare PrivResponse (untagged), so old hive-c0re callers that
        // deserialise directly to PrivResponse continue to work.
        let event = PrivEvent::Done(resp);
        let mut json = serde_json::to_string(&event).unwrap_or_else(|e| {
            format!("{{\"ok\":false,\"stdout\":\"\",\"stderr\":\"\",\"error\":\"serialise failed: {e}\"}}")
        });
        json.push('\n');
        if let Err(e) = writer.write_all(json.as_bytes()).await {
            tracing::warn!(error = %e, "write response failed");
            break;
        }
    }
}

/// Reject a request whose descriptor and operation disagree, in either
/// direction.
///
/// No guessing when the caller didn't say: an op that streams into a
/// passed descriptor cannot invent one, and an op that takes none must
/// not silently accept one. Returning the `Err` here drops the
/// `OwnedFd`, which closes it.
fn check_fd_agreement(req: &PrivRequest, fd: Option<&OwnedFd>) -> Result<()> {
    let wants_fd = matches!(req, PrivRequest::SendAgentSnapshotToFd { .. });
    match (wants_fd, fd.is_some()) {
        (true, false) => bail!("this operation requires a passed file descriptor, none arrived"),
        (false, true) => bail!("this operation does not take a passed file descriptor"),
        _ => Ok(()),
    }
}

async fn dispatch(line: &str, fd: Option<OwnedFd>, writer: &mut OwnedWriteHalf) -> PrivResponse {
    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,
            stderr,
            error: None,
        },
        Err(e) => PrivResponse {
            ok: false,
            stdout: String::new(),
            stderr: String::new(),
            error: Some(format!("{e:#}")),
        },
    }
}

/// Parse one request line, check it agrees with the descriptor that
/// arrived with it, and execute it.
///
/// Split out of [`dispatch`] so the three failure modes collapse into one
/// `Result` instead of three nested matches building the same struct.
async fn run(
    line: &str,
    fd: Option<OwnedFd>,
    writer: &mut OwnedWriteHalf,
) -> Result<(String, String)> {
    let req = serde_json::from_str::<PrivRequest>(line).context("parse request")?;
    check_fd_agreement(&req, fd.as_ref())?;
    exec(req, fd, writer).await
}

/// Execute a validated `PrivRequest`. Returns `(stdout, stderr)` on success.
/// For streaming ops (`CreateContainer`/`UpdateContainer` with `stream: true`)
/// output lines are forwarded to `writer` as `PrivEvent::Line` messages and
/// the returned strings are empty.
///
/// `fd` is the descriptor that arrived with this request, already checked
/// against the operation by [`check_fd_agreement`]: `Some` exactly for
/// the variants that stream into a caller-supplied descriptor, `None`
/// for every other operation.
// One match arm per priv op — a flat 1:1 dispatch table. Every arm either
// delegates directly or validates then delegates; an op whose handling is
// more than that gets its own named function instead, so the match's length
// tracks the op count, not complexity.
#[allow(clippy::too_many_lines)]
async fn exec(
    req: PrivRequest,
    fd: Option<OwnedFd>,
    writer: &mut OwnedWriteHalf,
) -> Result<(String, String)> {
    match req {
        PrivRequest::StartContainer { ref name } => container::start_container(name).await,

        PrivRequest::StartFailureLog {
            ref name,
            since_unix,
        } => container::start_failure_log(name, since_unix).await,

        PrivRequest::StopContainer { ref name } => container::stop_container(name).await,

        PrivRequest::KillContainer { ref name } => container::kill_container(name).await,

        PrivRequest::UpdateContainer {
            ref name,
            stream,
            ref toplevel,
        } => container::container_toplevel_action("update", name, toplevel, stream, writer).await,

        PrivRequest::CreateContainer {
            ref name,
            stream,
            ref toplevel,
        } => container::container_toplevel_action("create", name, toplevel, stream, writer).await,

        PrivRequest::RealiseClosure {
            ref agent,
            ref store_path,
        } => closure::realise_closure(agent, store_path).await,

        PrivRequest::DestroyContainer { ref name } => container::destroy_container(name).await,

        PrivRequest::ListContainers => container::container_run(&["list"]).await,

        PrivRequest::WriteNspawnFlags {
            container,
            binds,
            isolation,
            load_credentials,
            hive_facts,
            private_users,
        } => {
            container_config::handle_write_nspawn_flags_blocking(
                container,
                binds,
                isolation,
                load_credentials,
                *hive_facts,
                private_users,
            )
            .await
        }

        PrivRequest::EnsureAgentSocketDir {
            ref name,
            private_users,
        } => files::ensure_agent_socket_dir(name, private_users),

        PrivRequest::WriteResourceLimits {
            ref container,
            ref memory_max,
            ref cpu_quota,
            cpu_weight,
            io_weight,
        } => container_config::write_resource_limits(
            container, memory_max, cpu_quota, cpu_weight, io_weight,
        ),

        PrivRequest::RemoveServiceDropin { ref container } => {
            container_config::remove_service_dropin(container)
        }

        PrivRequest::DaemonReload => container_config::daemon_reload().await,

        PrivRequest::ReloadGatewayNginx => infra::sync_gateway_nginx().await,

        PrivRequest::SetAgentPaused {
            ref agent_name,
            paused,
        } => files::exec_set_agent_paused(agent_name, paused),

        PrivRequest::ControlInfraContainer { container, action } => {
            infra::control_infra_container(container, action).await
        }

        PrivRequest::EnsureAgentSubvolume { ref agent_name } => {
            storage::exec_ensure_agent_subvolume(agent_name).await
        }

        PrivRequest::DeleteAgentSubvolume { ref agent_name } => {
            storage::exec_delete_agent_subvolume(agent_name).await
        }

        PrivRequest::EnsureBtrfsQuota => storage::ensure_btrfs_quota().await,

        PrivRequest::ReadSubvolumeUsage { ref agent_name } => {
            storage::exec_read_subvolume_usage(agent_name).await
        }

        PrivRequest::SetSubvolumeQuota {
            ref agent_name,
            limit_bytes,
        } => storage::exec_set_subvolume_quota(agent_name, limit_bytes).await,

        PrivRequest::MeasureAgentDisk { ref agent_name } => {
            storage::exec_measure_agent_disk(agent_name).await
        }

        PrivRequest::UpgradeAgentSubvolume { ref agent_name } => {
            storage::exec_upgrade_agent_subvolume(agent_name).await
        }

        PrivRequest::SnapshotAgentSubvolume {
            ref agent_name,
            ref snapshot_name,
        } => storage::exec_snapshot_agent_subvolume(agent_name, snapshot_name).await,

        PrivRequest::DeleteAgentSnapshot {
            ref agent_name,
            ref snapshot_name,
        } => storage::exec_delete_agent_snapshot(agent_name, snapshot_name).await,

        PrivRequest::SendAgentSnapshotToFile {
            ref agent_name,
            ref snapshot_name,
            ref parent_snapshot_name,
            ref dest_file_name,
        } => {
            storage::exec_send_agent_snapshot_to_file(
                agent_name,
                snapshot_name,
                parent_snapshot_name.as_deref(),
                dest_file_name,
            )
            .await
        }

        PrivRequest::SendAgentSnapshotToFd {
            ref agent_name,
            ref snapshot_name,
            ref parent_snapshot_name,
        } => {
            storage::exec_send_agent_snapshot_to_fd(
                agent_name,
                snapshot_name,
                parent_snapshot_name.as_deref(),
                fd,
            )
            .await
        }

        PrivRequest::SyncAgentTmpfiles => {
            files::remove_legacy_tmpfiles()?;
            Ok((String::new(), String::new()))
        }
    }
}
#[cfg(test)]
mod tests {
    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 {
        PrivRequest::SendAgentSnapshotToFd {
            agent_name: "atlas".to_owned(),
            snapshot_name: "hive-migrate".to_owned(),
            parent_snapshot_name: None,
        }
    }

    /// A real descriptor — `/dev/null` rather than a fake, so the drop
    /// that closes it on a rejection path is genuinely exercised.
    fn some_fd() -> OwnedFd {
        std::fs::File::open("/dev/null")
            .expect("open /dev/null")
            .into()
    }

    /// An op that streams into a passed descriptor cannot invent one:
    /// falling back to anything (a temp file, the response socket) would
    /// send an agent's state somewhere the caller never asked for.
    #[test]
    fn an_fd_taking_op_without_a_descriptor_is_rejected() {
        let err = check_fd_agreement(&fd_taking_request(), None)
            .expect_err("no descriptor arrived, so this must not proceed");
        let msg = format!("{err:#}");
        assert!(msg.contains("requires a passed file descriptor"), "{msg}");
    }

    /// The mirror case: a descriptor sent alongside an op that takes
    /// none is a protocol error, not something to ignore. Returning the
    /// error drops the `OwnedFd`, which closes it — the alternative
    /// leaks one descriptor per stray request in a long-lived root
    /// process.
    #[test]
    fn a_descriptor_sent_to_an_op_that_takes_none_is_rejected() {
        let fd = some_fd();
        let err = check_fd_agreement(&PrivRequest::DaemonReload, Some(&fd))
            .expect_err("an unexpected descriptor must not be silently ignored");
        let msg = format!("{err:#}");
        assert!(
            msg.contains("does not take a passed file descriptor"),
            "{msg}"
        );
    }

    /// Both agreeing combinations pass, so the check rejects mismatches
    /// rather than descriptors in general.
    #[test]
    fn agreeing_combinations_are_accepted() {
        let fd = some_fd();
        check_fd_agreement(&fd_taking_request(), Some(&fd))
            .expect("an fd-taking op with its descriptor is the normal case");
        check_fd_agreement(&PrivRequest::DaemonReload, None)
            .expect("every ordinary request arrives without a descriptor");
    }

    /// An older hive-c0re's `SyncAgentTmpfiles` still decodes, payload and
    /// all, so it gets the legacy cleanup rather than a parse error. Control:
    /// an unknown op does not decode.
    #[test]
    fn legacy_sync_agent_tmpfiles_request_still_decodes() {
        let legacy =
            r#"{"op":"sync_agent_tmpfiles","agents":[{"name":"atlas","uid":1000,"gid":994}]}"#;
        assert!(matches!(
            serde_json::from_str::<PrivRequest>(legacy),
            Ok(PrivRequest::SyncAgentTmpfiles)
        ));
        assert!(serde_json::from_str::<PrivRequest>(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::<RawFd>();
        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::<u8>(),
                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<OwnedFd>) -> 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<PrivResponse> {
        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::<PrivEvent>(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}"
        );
    }
}