//! 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`] (the swarm-signature
//! check before an apply, and the cache-facts refresh). 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 anyhow::{Context as _, Result, bail};
use hive_priv_sock::{PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
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/4-implementation/trust-boundary/boundary-mechanism.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)
}

async fn handle(stream: UnixStream) {
    let (reader, mut writer) = stream.into_split();
    let mut lines = BufReader::new(reader).lines();
    while let Ok(Some(line)) = lines.next_line().await {
        let resp = dispatch(&line, &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;
        }
    }
}

async fn dispatch(line: &str, writer: &mut OwnedWriteHalf) -> PrivResponse {
    response_of(run(line, 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 and execute it.
///
/// Split out of [`dispatch`] so the two failure modes collapse into one
/// `Result` instead of nested matches building the same struct.
async fn run(line: &str, writer: &mut OwnedWriteHalf) -> Result<(String, String)> {
    let req = serde_json::from_str::<PrivRequest>(line).context("parse request")?;
    exec(req, 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.
// 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, 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::RefreshCacheFacts => closure::refresh_cache_facts().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,
            nix_builds,
        } => {
            container_config::handle_write_nspawn_flags_blocking(
                container,
                binds,
                isolation,
                load_credentials,
                *hive_facts,
                private_users,
                nix_builds,
            )
            .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::SyncAgentTmpfiles => {
            files::remove_legacy_tmpfiles()?;
            Ok((String::new(), String::new()))
        }
    }
}

#[cfg(test)]
mod tests {
    use hive_priv_sock::{PrivEvent, PrivRequest, PrivResponse};
    use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
    use tokio::net::UnixStream;

    use super::{check_socket_activation, dispatch, handle, response_of};

    /// 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());
    }

    // ---- 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}");
        }
    }

    // ---- dispatch: parsing, 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) -> String {
        let (mut w, _keep) = sink();
        let resp = dispatch(line, &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).await;
            assert!(msg.starts_with("parse request"), "{line}: {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).await;
            assert!(msg.contains(want), "{line}: {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)).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)).await;
            assert!(msg.contains("invalid"), "{label} {parent} {dest}: {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}"#).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 sends non-UTF-8 or nothing gets no response and
    /// nothing runs. A line cut off by EOF is still one line: it fails to
    /// parse and is answered with the parse error, never executed.
    #[tokio::test]
    async fn a_broken_request_closes_the_connection_without_running_anything() {
        assert!(serve(b"\xff\xfe\n").await.is_empty());
        assert!(serve(b"").await.is_empty());
        let cut = serve(b"{\"op\":\"start_cont").await;
        assert_eq!(cut.len(), 1);
        assert!(!cut[0].ok);
        assert!(
            cut[0]
                .error
                .as_deref()
                .unwrap()
                .starts_with("parse request")
        );
    }
}