Watch
0
0
Fork
You've already forked hyperhive
0
hyperhive/hive-priv/src/main.rs
atlas 7c0a622d41 build agents from internal/meta when it pins their config
A rebuild's MetaSync resolves the head of the swarm's internal/meta once and
records it; the prebuild and the swap build that flake instead of the hive's
own, and a first deploy does the same in Create. hive-priv takes the commit as
a validated meta_rev on UpdateContainer and CreateContainer and builds from its
own forge URL (HIVE_PRIV_FORGE_URL).

The head is used only when its flake lists the agent at the config commit the
forge's main is at; otherwise, and when the repo has no commit or this hive has
no core token file for nix's git, the hive renders locally as before.

hive-c0re's and hive-priv's gitconfig now share the forge credential helper and
rewrite the forge's https URLs to http, which hive-priv lacked.
2026-10-24 21:13:24 +02:00

564 lines
21 KiB
Rust

//! 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), [`storage`] (btrfs subvolumes, quotas, snapshots, disk
//! measurement), [`files`] (pause marker, socket dir, legacy tmpfiles) and
//! [`infra`] (gateway nginx, forge admin, infra containers). Helpers more than
//! one family uses are in [`shared`].
mod container;
mod container_config;
mod files;
mod infra;
mod shared;
mod storage;
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<()> {
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");
}
}
}
}
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.
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());
let activated =
matches!(listen_fds, Some(n) if n >= 1) && listen_pid == Some(std::process::id());
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"
);
}
// 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 {
match run(line, fd, writer).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:#}")),
},
}
}
/// 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::StopContainer { ref name } => container::stop_container(name).await,
PrivRequest::KillContainer { ref name } => container::kill_container(name).await,
PrivRequest::UpdateContainer {
ref name,
stream,
ref meta_rev,
} => {
container::container_flake_action("update", name, meta_rev.as_deref(), stream, writer)
.await
}
PrivRequest::CreateContainer {
ref name,
stream,
ref meta_rev,
} => {
container::container_flake_action("create", name, meta_rev.as_deref(), stream, writer)
.await
}
PrivRequest::DestroyContainer { ref name } => container::destroy_container(name).await,
PrivRequest::ListContainers => container::container_run(&["list"]).await,
PrivRequest::WriteNspawnFlags {
ref container,
ref binds,
ref isolation,
ref load_credentials,
ref hive_facts,
} => container_config::handle_write_nspawn_flags(
container,
binds,
isolation,
load_credentials,
hive_facts,
),
PrivRequest::EnsureAgentSocketDir { ref name } => files::ensure_agent_socket_dir(name),
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::RunForgeAdmin { ref args } => infra::exec_forge_admin(args).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 super::{OwnedFd, PrivRequest, check_fd_agreement};
/// 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());
}
}