feat(#702): hive-priv privileged helper binary

hive-priv/src/main.rs

@@ -0,0 +1,320 @@
+//! 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, or known sibling services) are accepted.
+//!
+//! **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.
+
+use std::os::unix::fs::PermissionsExt as _;
+use std::path::{Path, PathBuf};
+
+use anyhow::{Context as _, Result, bail};
+use hive_sh4re::priv_proto::{PRIV_SOCK, PrivRequest, PrivResponse};
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::net::{UnixListener, UnixStream};
+use tokio::process::Command;
+
+/// Sub-agent container prefix (mirrors `lifecycle::AGENT_PREFIX`).
+const AGENT_PREFIX: &str = "h-";
+
+/// Manager container name (mirrors `lifecycle::MANAGER_NAME`).
+const MANAGER_NAME: &str = "hm1nd";
+
+/// Sibling service containers managed by hive-c0re.
+const SIBLING_CONTAINERS: &[&str] = &["hive-forge", "hive-matrix", "hive-gateway"];
+
+/// Allowed path prefixes for `Chown` / `Chmod` operations.
+const SAFE_PATH_PREFIXES: &[&str] = &["/run/hive-agent/", "/var/lib/hyperhive/"];
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| "info".into()),
+        )
+        .init();
+
+    let listener = socket_listener()?;
+    tracing::info!("hive-priv listening");
+
+    loop {
+        match listener.accept().await {
+            Ok((stream, _)) => {
+                tokio::spawn(handle(stream));
+            }
+            Err(e) => {
+                tracing::error!(error = %e, "accept failed");
+            }
+        }
+    }
+}
+
+fn socket_listener() -> Result<UnixListener> {
+    // Socket activation: systemd passes the socket as fd 3 when
+    // LISTEN_FDS >= 1 and LISTEN_PID matches our pid.
+    let listen_fds: Option<i32> = std::env::var("LISTEN_FDS")
+        .ok()
+        .and_then(|s| s.parse().ok());
+    let listen_pid: Option<u32> = std::env::var("LISTEN_PID")
+        .ok()
+        .and_then(|s| s.parse().ok());
+
+    if let (Some(n), Some(p)) = (listen_fds, listen_pid) {
+        if n >= 1 && p == 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");
+            return Ok(listener);
+        }
+    }
+
+    // Fallback: bind the socket ourselves.
+    let path = Path::new(PRIV_SOCK);
+    if let Some(parent) = path.parent() {
+        std::fs::create_dir_all(parent)
+            .with_context(|| format!("create {}", parent.display()))?;
+    }
+    let _ = std::fs::remove_file(path);
+    let listener =
+        UnixListener::bind(path).with_context(|| format!("bind {PRIV_SOCK}"))?;
+    // Mode 0660: only the hive-core group can connect.
+    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o660))
+        .context("chmod priv.sock")?;
+    tracing::info!(path = PRIV_SOCK, "bound priv 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).await;
+        let mut json = serde_json::to_string(&resp).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) -> PrivResponse {
+    match serde_json::from_str::<PrivRequest>(line) {
+        Ok(req) => match exec(req).await {
+            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:#}")),
+            },
+        },
+        Err(e) => PrivResponse {
+            ok: false,
+            stdout: String::new(),
+            stderr: String::new(),
+            error: Some(format!("parse request: {e}")),
+        },
+    }
+}
+
+/// Execute a validated `PrivRequest`. Returns `(stdout, stderr)` on success.
+async fn exec(req: PrivRequest) -> Result<(String, String)> {
+    match req {
+        PrivRequest::ContainerRun { ref args } => {
+            validate_container_args(args)?;
+            let out = Command::new("nixos-container")
+                .args(args)
+                .output()
+                .await
+                .context("invoke nixos-container")?;
+            let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
+            let stderr = String::from_utf8_lossy(&out.stderr).into_owned();
+            // Log each line so progress is visible in journald even
+            // without a streaming protocol.
+            for line in stdout.lines() {
+                tracing::info!(target: "nixos-container", "{line}");
+            }
+            for line in stderr.lines() {
+                tracing::warn!(target: "nixos-container", "{line}");
+            }
+            if !out.status.success() {
+                bail!(
+                    "nixos-container {} failed ({}): {}",
+                    args.join(" "),
+                    out.status,
+                    stderr.trim()
+                );
+            }
+            Ok((stdout, stderr))
+        }
+
+        PrivRequest::DaemonReload => {
+            let out = Command::new("systemctl")
+                .arg("daemon-reload")
+                .output()
+                .await
+                .context("invoke systemctl daemon-reload")?;
+            if !out.status.success() {
+                bail!(
+                    "systemctl daemon-reload failed ({}): {}",
+                    out.status,
+                    String::from_utf8_lossy(&out.stderr).trim()
+                );
+            }
+            Ok((String::new(), String::new()))
+        }
+
+        PrivRequest::WriteNspawnConf { ref container, ref content } => {
+            validate_container_name(container)?;
+            let path = format!("/etc/nixos-containers/{container}.conf");
+            std::fs::write(&path, content).with_context(|| format!("write {path}"))?;
+            Ok((String::new(), String::new()))
+        }
+
+        PrivRequest::WriteSystemdDropin {
+            ref container,
+            ref filename,
+            ref content,
+        } => {
+            validate_container_name(container)?;
+            validate_dropin_filename(filename)?;
+            let dir =
+                format!("/run/systemd/system/container@{container}.service.d");
+            std::fs::create_dir_all(&dir)
+                .with_context(|| format!("create {dir}"))?;
+            let path = format!("{dir}/{filename}");
+            std::fs::write(&path, content).with_context(|| format!("write {path}"))?;
+            Ok((String::new(), String::new()))
+        }
+
+        PrivRequest::RemoveSystemdDropin { ref container } => {
+            validate_container_name(container)?;
+            let dir =
+                format!("/run/systemd/system/container@{container}.service.d");
+            if Path::new(&dir).exists() {
+                std::fs::remove_dir_all(&dir)
+                    .with_context(|| format!("remove {dir}"))?;
+            }
+            Ok((String::new(), String::new()))
+        }
+
+        PrivRequest::Chown { ref path, uid, gid } => {
+            validate_safe_path(path)?;
+            std::os::unix::fs::chown(path, Some(uid), Some(gid))
+                .with_context(|| {
+                    format!("chown {} to {uid}:{gid}", path.display())
+                })?;
+            Ok((String::new(), String::new()))
+        }
+
+        PrivRequest::Chmod { ref path, mode } => {
+            validate_safe_path(path)?;
+            std::fs::set_permissions(path, std::fs::Permissions::from_mode(mode))
+                .with_context(|| {
+                    format!("chmod {:o} {}", mode, path.display())
+                })?;
+            Ok((String::new(), String::new()))
+        }
+
+        PrivRequest::SystemdRunMachine { ref machine, ref cmd } => {
+            validate_container_name(machine)?;
+            let mut args = vec!["--machine".to_owned(), machine.clone(), "--quiet".to_owned(), "--".to_owned()];
+            args.extend(cmd.iter().cloned());
+            let out = Command::new("systemd-run")
+                .args(&args)
+                .output()
+                .await
+                .context("invoke systemd-run")?;
+            let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
+            let stderr = String::from_utf8_lossy(&out.stderr).into_owned();
+            if !out.status.success() {
+                bail!(
+                    "systemd-run --machine={machine} failed ({}): {}",
+                    out.status,
+                    stderr.trim()
+                );
+            }
+            Ok((stdout, stderr))
+        }
+    }
+}
+
+/// Validate that nixos-container args reference only hive-managed containers.
+fn validate_container_args(args: &[String]) -> Result<()> {
+    if args.is_empty() {
+        bail!("ContainerRun: args must not be empty");
+    }
+    // Verbs whose second positional argument is a container name.
+    const CONTAINER_ARG_VERBS: &[&str] =
+        &["start", "stop", "kill", "restart", "destroy", "update", "run", "create"];
+    if CONTAINER_ARG_VERBS.contains(&args[0].as_str()) {
+        if let Some(name) = args.get(1) {
+            validate_container_name(name)?;
+        }
+    }
+    // `list` and unknown verbs take no container arg - no further validation needed.
+    Ok(())
+}
+
+/// Validate that a container name is managed by hive.
+fn validate_container_name(name: &str) -> Result<()> {
+    if name == MANAGER_NAME {
+        return Ok(());
+    }
+    if SIBLING_CONTAINERS.contains(&name) {
+        return Ok(());
+    }
+    if name.starts_with(AGENT_PREFIX) {
+        let suffix = &name[AGENT_PREFIX.len()..];
+        if !suffix.is_empty()
+            && suffix.chars().all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-')
+        {
+            return Ok(());
+        }
+    }
+    bail!("container name {name:?} is not managed by hive");
+}
+
+fn validate_dropin_filename(filename: &str) -> Result<()> {
+    if filename.is_empty() || filename.contains('/') || filename.contains("..") {
+        bail!("invalid drop-in filename: {filename:?}");
+    }
+    Ok(())
+}
+
+fn validate_safe_path(path: &Path) -> Result<()> {
+    let s = path.to_string_lossy();
+    for prefix in SAFE_PATH_PREFIXES {
+        if s.starts_with(prefix) && !s.contains("..") {
+            return Ok(());
+        }
+    }
+    bail!("path {} is outside allowed prefixes", path.display());
+}