2996 lines
122 KiB
Rust
2996 lines
122 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, or known sibling service containers) 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.
|
|
|
|
use std::os::fd::{AsRawFd as _, FromRawFd as _, OwnedFd, RawFd};
|
|
use std::path::{Path, PathBuf};
|
|
|
|
use anyhow::{Context as _, Result, bail};
|
|
use hive_priv_sock::{
|
|
AGENT_PREFIX, AGENT_RUNTIME_ROOT, AGENT_STATE_ROOT, AgentTmpfilesEntry, BindMount,
|
|
CredentialMount, InfraAction, InfraContainer, JournalQuery, META_DIR, MIGRATE_STAGING_ROOT,
|
|
NetworkIsolation, PAUSED_MARKER_FILE, PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse,
|
|
PrivStream, PrivStreamLine, SIBLING_CONTAINERS,
|
|
};
|
|
use serde::Serialize;
|
|
use tokio::io::{AsyncWriteExt, BufReader};
|
|
use tokio::net::unix::OwnedWriteHalf;
|
|
use tokio::net::{UnixListener, UnixStream};
|
|
use tokio::process::Command;
|
|
|
|
/// Root of the per-agent unix-socket dirs on the host.
|
|
const SOCKET_DIR_ROOT: &str = "/run/hive-agent";
|
|
|
|
#[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> {
|
|
// 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/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
|
|
}
|
|
|
|
/// Write one `PrivEvent::Line` to the client. Best-effort: a write
|
|
/// failure is logged but doesn't abort the running subprocess.
|
|
async fn write_line_event(writer: &mut OwnedWriteHalf, stream: PrivStream, data: &str) {
|
|
let event = PrivEvent::Line(PrivStreamLine {
|
|
stream,
|
|
data: data.to_owned(),
|
|
});
|
|
if let Ok(mut json) = serde_json::to_string(&event) {
|
|
json.push('\n');
|
|
if let Err(e) = writer.write_all(json.as_bytes()).await {
|
|
tracing::warn!(error = %e, "write_line_event: write failed");
|
|
}
|
|
}
|
|
}
|
|
|
|
/// 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. The length tracks
|
|
// the op count, not complexity; splitting it would just scatter the mapping.
|
|
#[allow(clippy::too_many_lines)]
|
|
/// Execute one validated request.
|
|
///
|
|
/// `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.
|
|
async fn exec(
|
|
req: PrivRequest,
|
|
fd: Option<OwnedFd>,
|
|
writer: &mut OwnedWriteHalf,
|
|
) -> Result<(String, String)> {
|
|
match req {
|
|
PrivRequest::StartContainer { ref name } => {
|
|
validate_container_name(name)?;
|
|
let machine = container_system_name(name);
|
|
// Clear any start-limit lockout left by earlier failures so a
|
|
// now-correct start isn't blocked. nixos-container start does not
|
|
// do this itself. Best-effort: if the unit doesn't exist yet
|
|
// (first-time create) reset-failed is a no-op and we proceed.
|
|
let _ = Command::new("systemctl")
|
|
.args(["reset-failed", &format!("container@{machine}.service")])
|
|
.status()
|
|
.await;
|
|
container_run(&["start", &machine]).await
|
|
}
|
|
|
|
PrivRequest::StopContainer { ref name } => {
|
|
validate_container_name(name)?;
|
|
stop_and_release(&container_system_name(name)).await
|
|
}
|
|
|
|
PrivRequest::KillContainer { ref name } => {
|
|
validate_container_name(name)?;
|
|
// nixos-container has no kill verb. Use machinectl to send SIGKILL
|
|
// to all processes in the container — the right semantics for a
|
|
// forced shutdown after a graceful stop has already been attempted.
|
|
let machine = container_system_name(name);
|
|
machinectl_run(&["kill", &machine, "--signal=SIGKILL"]).await
|
|
}
|
|
|
|
PrivRequest::UpdateContainer { ref name, stream } => {
|
|
container_flake_action("update", name, stream, writer).await
|
|
}
|
|
|
|
PrivRequest::CreateContainer { ref name, stream } => {
|
|
container_flake_action("create", name, stream, writer).await
|
|
}
|
|
|
|
PrivRequest::DestroyContainer { ref name } => {
|
|
validate_container_name(name)?;
|
|
container_run(&["destroy", &container_system_name(name)]).await
|
|
}
|
|
|
|
PrivRequest::ListContainers => container_run(&["list"]).await,
|
|
|
|
PrivRequest::ReadContainerJournal {
|
|
ref container,
|
|
ref query,
|
|
} => {
|
|
validate_container_system_name(container)?;
|
|
read_container_journal(container, query).await
|
|
}
|
|
|
|
PrivRequest::WriteNspawnFlags {
|
|
ref container,
|
|
ref binds,
|
|
ref isolation,
|
|
ref load_credentials,
|
|
} => handle_write_nspawn_flags(container, binds, isolation.as_ref(), load_credentials),
|
|
|
|
PrivRequest::WriteResourceLimits {
|
|
ref container,
|
|
ref memory_max,
|
|
ref cpu_quota,
|
|
cpu_weight,
|
|
io_weight,
|
|
} => write_resource_limits(container, memory_max, cpu_quota, cpu_weight, io_weight),
|
|
|
|
PrivRequest::RemoveServiceDropin { ref container } => remove_service_dropin(container),
|
|
|
|
PrivRequest::DaemonReload => daemon_reload().await,
|
|
|
|
PrivRequest::ReloadGatewayNginx => sync_gateway_nginx().await,
|
|
|
|
PrivRequest::RunForgeAdmin { ref args } => {
|
|
for arg in args {
|
|
validate_forge_admin_arg(arg)?;
|
|
}
|
|
run_forge_admin(args).await
|
|
}
|
|
|
|
PrivRequest::SetAgentPaused {
|
|
ref agent_name,
|
|
paused,
|
|
} => {
|
|
validate_agent_name(agent_name)?;
|
|
set_agent_paused(agent_name, paused)
|
|
}
|
|
|
|
PrivRequest::WriteAgentForgeToken {
|
|
ref agent_name,
|
|
ref token,
|
|
} => {
|
|
validate_agent_name(agent_name)?;
|
|
write_agent_state_file(agent_name, "forge-token", &format!("{token}\n"))
|
|
}
|
|
|
|
PrivRequest::WriteAgentMatrixToken {
|
|
ref agent_name,
|
|
ref token,
|
|
ref account,
|
|
ref homeserver,
|
|
} => {
|
|
validate_agent_name(agent_name)?;
|
|
// Build the token filename. `None` → the hive account's
|
|
// `matrix-token`; `Some(a)` → `matrix-token-<a>`. The account
|
|
// suffix MUST be validated as a plain identifier (no `/`, `.`,
|
|
// `..`) before it goes into the filename, or a crafted account
|
|
// could traverse out of the state dir — `write_agent_state_file`
|
|
// trusts its `filename` argument.
|
|
let filename = match account {
|
|
None => "matrix-token".to_owned(),
|
|
Some(a) => {
|
|
validate_name_chars(a)?;
|
|
format!("matrix-token-{a}")
|
|
}
|
|
};
|
|
let res = write_agent_state_file(agent_name, &filename, &format!("{token}\n"))?;
|
|
// For an extra account, persist its homeserver in a sidecar
|
|
// (`matrix-account-<a>.json`) so the daemon can auto-discover the
|
|
// account without a static `matrixAccounts` config entry. Only
|
|
// when both `account` and `homeserver` are present; the account
|
|
// suffix is already validated above.
|
|
if let (Some(a), Some(hs)) = (account, homeserver) {
|
|
let meta = serde_json::to_string(&MatrixAccountSidecar {
|
|
homeserver: hs.as_str(),
|
|
})
|
|
.context("serialize matrix account sidecar")?;
|
|
write_agent_state_file(agent_name, &format!("matrix-account-{a}.json"), &meta)?;
|
|
}
|
|
Ok(res)
|
|
}
|
|
|
|
PrivRequest::WriteAgentGithubToken {
|
|
ref agent_name,
|
|
ref token,
|
|
} => {
|
|
validate_agent_name(agent_name)?;
|
|
write_agent_state_file(agent_name, "github-token", &format!("{token}\n"))
|
|
}
|
|
|
|
PrivRequest::WriteAgentExtraForgeAccount {
|
|
ref agent_name,
|
|
ref label,
|
|
ref base_url,
|
|
ref token,
|
|
} => {
|
|
validate_agent_name(agent_name)?;
|
|
validate_name_chars(label)?;
|
|
let res = write_agent_state_file(
|
|
agent_name,
|
|
&format!("forge-{label}-token"),
|
|
&format!("{token}\n"),
|
|
)?;
|
|
// Sidecar carries the base URL — there's no host-side nix config
|
|
// for extra forges, so this is the only place it's persisted.
|
|
let meta = serde_json::to_string(&ForgeSidecar {
|
|
base_url: base_url.as_str(),
|
|
})
|
|
.context("serialize forge account sidecar")?;
|
|
write_agent_state_file(agent_name, &format!("forge-{label}.json"), &meta)?;
|
|
Ok(res)
|
|
}
|
|
|
|
PrivRequest::DeleteAgentExtraForgeAccount {
|
|
ref agent_name,
|
|
ref label,
|
|
} => {
|
|
validate_agent_name(agent_name)?;
|
|
validate_name_chars(label)?;
|
|
delete_agent_state_file(agent_name, &format!("forge-{label}-token"))?;
|
|
delete_agent_state_file(agent_name, &format!("forge-{label}.json"))
|
|
}
|
|
|
|
PrivRequest::RestartMatrixDaemon { ref agent_name } => {
|
|
restart_matrix_daemon(agent_name).await
|
|
}
|
|
|
|
PrivRequest::RegisterCiRunner { ref token } => register_ci_runner(token).await,
|
|
|
|
PrivRequest::ControlInfraContainer { container, action } => {
|
|
control_infra_container(container, action).await
|
|
}
|
|
|
|
PrivRequest::EnsureAgentSubvolume { ref agent_name } => {
|
|
validate_agent_name(agent_name)?;
|
|
ensure_agent_subvolume(agent_name).await
|
|
}
|
|
|
|
PrivRequest::DeleteAgentSubvolume { ref agent_name } => {
|
|
validate_agent_name(agent_name)?;
|
|
delete_agent_subvolume(agent_name).await
|
|
}
|
|
|
|
PrivRequest::EnsureBtrfsQuota => ensure_btrfs_quota().await,
|
|
|
|
PrivRequest::ReadSubvolumeUsage { ref agent_name } => {
|
|
validate_agent_name(agent_name)?;
|
|
read_subvolume_usage(agent_name).await
|
|
}
|
|
|
|
PrivRequest::SetSubvolumeQuota {
|
|
ref agent_name,
|
|
limit_bytes,
|
|
} => {
|
|
validate_agent_name(agent_name)?;
|
|
set_subvolume_quota(agent_name, limit_bytes).await
|
|
}
|
|
|
|
PrivRequest::UpgradeAgentSubvolume { ref agent_name } => {
|
|
validate_agent_name(agent_name)?;
|
|
upgrade_agent_subvolume(agent_name).await
|
|
}
|
|
|
|
PrivRequest::SnapshotAgentSubvolume {
|
|
ref agent_name,
|
|
ref snapshot_name,
|
|
} => {
|
|
validate_agent_name(agent_name)?;
|
|
validate_snapshot_name(snapshot_name)?;
|
|
snapshot_agent_subvolume(agent_name, snapshot_name).await
|
|
}
|
|
|
|
PrivRequest::DeleteAgentSnapshot {
|
|
ref agent_name,
|
|
ref snapshot_name,
|
|
} => {
|
|
validate_agent_name(agent_name)?;
|
|
validate_snapshot_name(snapshot_name)?;
|
|
delete_agent_snapshot(agent_name, snapshot_name).await
|
|
}
|
|
|
|
PrivRequest::SendAgentSnapshotToFile {
|
|
ref agent_name,
|
|
ref snapshot_name,
|
|
ref parent_snapshot_name,
|
|
ref dest_file_name,
|
|
} => {
|
|
validate_agent_name(agent_name)?;
|
|
validate_snapshot_name(snapshot_name)?;
|
|
if let Some(parent) = parent_snapshot_name {
|
|
validate_snapshot_name(parent)?;
|
|
}
|
|
validate_credential_name(dest_file_name)?;
|
|
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,
|
|
} => {
|
|
validate_agent_name(agent_name)?;
|
|
validate_snapshot_name(snapshot_name)?;
|
|
if let Some(parent) = parent_snapshot_name {
|
|
validate_snapshot_name(parent)?;
|
|
}
|
|
let dest = fd.context("no descriptor to stream into")?;
|
|
send_agent_snapshot_to_fd(
|
|
agent_name,
|
|
snapshot_name,
|
|
parent_snapshot_name.as_deref(),
|
|
dest,
|
|
)
|
|
.await
|
|
}
|
|
|
|
PrivRequest::SyncAgentTmpfiles { ref agents } => sync_agent_tmpfiles(agents).await,
|
|
}
|
|
}
|
|
|
|
/// Shared body for `CreateContainer` / `UpdateContainer`: validate the
|
|
/// name, build the `nixos-container <verb> … --flake <ref>` argv, and
|
|
/// run it (streaming line events to `writer` when `stream` is set).
|
|
async fn container_flake_action(
|
|
verb: &str,
|
|
name: &str,
|
|
stream: bool,
|
|
writer: &mut OwnedWriteHalf,
|
|
) -> Result<(String, String)> {
|
|
validate_container_name(name)?;
|
|
let flake_ref = agent_flake_ref(name);
|
|
let args = [verb, &container_system_name(name), "--flake", &flake_ref];
|
|
if stream {
|
|
container_run_streaming(&args, writer).await
|
|
} else {
|
|
container_run(&args).await
|
|
}
|
|
}
|
|
|
|
/// `WriteNspawnFlags` — validate the container + every bind path + every
|
|
/// credential entry, then write the container's nspawn flag overrides.
|
|
fn handle_write_nspawn_flags(
|
|
container: &str,
|
|
binds: &[BindMount],
|
|
isolation: Option<&NetworkIsolation>,
|
|
load_credentials: &[CredentialMount],
|
|
) -> Result<(String, String)> {
|
|
validate_container_system_name(container)?;
|
|
for bind in binds {
|
|
validate_bind_path(&bind.host_path)?;
|
|
validate_bind_path(&bind.container_path)?;
|
|
}
|
|
for cred in load_credentials {
|
|
validate_credential_name(&cred.name)?;
|
|
// Same path rules as binds (absolute, no colon/newline/quote/null):
|
|
// the colon ban is essential since `--load-credential=name:path`
|
|
// uses `:` as the name/path separator.
|
|
validate_bind_path(&cred.host_path)?;
|
|
}
|
|
write_nspawn_flags(container, binds, isolation, load_credentials)?;
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
/// A btrfs snapshot label must start with `hive-` — this doubles as an
|
|
/// allow-list: only names hivectl itself constructs (or an operator who
|
|
/// knows the convention) can reach the `btrfs subvolume snapshot`/`delete`
|
|
/// shellouts, so an arbitrary caller can't use the snapshot ops to probe or
|
|
/// churn unrelated paths under `AGENT_STATE_ROOT`. Beyond the prefix, the
|
|
/// same charset restriction as [`validate_credential_name`] applies (it's
|
|
/// interpolated straight into a filesystem path).
|
|
fn validate_snapshot_name(name: &str) -> Result<()> {
|
|
if !name.starts_with("hive-") {
|
|
bail!("invalid snapshot label {name:?}: must start with \"hive-\"");
|
|
}
|
|
validate_credential_name(name)
|
|
}
|
|
|
|
/// A systemd credential id must be a short token — restrict to
|
|
/// `[A-Za-z0-9_-]` (no `.`) so it can't inject extra `--load-credential`
|
|
/// argv or break the `name:path` shape. `.` is deliberately excluded, not
|
|
/// just a bare `..`: this name gets interpolated into filesystem paths
|
|
/// (snapshot labels) and there's no legitimate need for a dot in either a
|
|
/// systemd credential id or a `hive-`-prefixed snapshot label — we're
|
|
/// defining this token format from scratch, so keep it maximally strict
|
|
/// rather than allow-then-patch each traversal-adjacent character
|
|
/// (mara: "we are making up the rules here, lets go strict").
|
|
fn validate_credential_name(name: &str) -> Result<()> {
|
|
if name.is_empty()
|
|
|| !name
|
|
.bytes()
|
|
.all(|b| b.is_ascii_alphanumeric() || matches!(b, b'_' | b'-'))
|
|
{
|
|
bail!("invalid credential name {name:?}: must be non-empty [A-Za-z0-9_-]");
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// `RemoveServiceDropin` — remove the container service's drop-in dir
|
|
/// if present (idempotent).
|
|
fn remove_service_dropin(container: &str) -> Result<(String, String)> {
|
|
validate_container_system_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()))
|
|
}
|
|
|
|
/// `WriteResourceLimits` — drop the systemd resource settings into the
|
|
/// container service's drop-in dir, together with a
|
|
/// `ConditionPathIsDirectory=` guard on the agent's MCP runtime dir.
|
|
///
|
|
/// Two different kinds of setting land in the same file. `MemoryMax=` /
|
|
/// `CPUQuota=` are hard caps that throttle even on an idle host;
|
|
/// `CPUWeight=` / `IOWeight=` are cgroup v2 relative shares that only
|
|
/// decide who yields *under contention*. A weight of `None` means "not
|
|
/// configured" and omits the line, so a hive-c0re built before the weights
|
|
/// existed keeps producing the old two-line drop-in.
|
|
///
|
|
/// The condition causes systemd to *skip* (not *fail*) the unit when the
|
|
/// bind-mount source dir is absent — result is `condition`, which does not
|
|
/// increment the start-limit counter. This is belt-and-braces on top of
|
|
/// the tmpfiles.d entries written by `SyncAgentTmpfiles`: in the unlikely
|
|
/// event the dir is missing at start time, the unit idles rather than
|
|
/// restart-looping into `start-limit-hit`.
|
|
fn write_resource_limits(
|
|
container: &str,
|
|
memory_max: &str,
|
|
cpu_quota: &str,
|
|
cpu_weight: Option<u32>,
|
|
io_weight: Option<u32>,
|
|
) -> Result<(String, String)> {
|
|
validate_container_system_name(container)?;
|
|
// Derive the logical agent name (strip h- prefix) to form the runtime
|
|
// dir path. Falls back to the full container name for infra containers
|
|
// that don't use the h- prefix.
|
|
let logical = container.strip_prefix(AGENT_PREFIX).unwrap_or(container);
|
|
let runtime_dir = format!("{AGENT_RUNTIME_ROOT}/{logical}");
|
|
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}/hyperhive-limits.conf");
|
|
let content = limits_dropin_body(&runtime_dir, memory_max, cpu_quota, cpu_weight, io_weight);
|
|
std::fs::write(&path, content).with_context(|| format!("write {path}"))?;
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
/// How long a window the start-limit counts over, and how many starts it
|
|
/// allows inside it.
|
|
///
|
|
/// `container@.service` sets `Restart=on-failure` and **no** start limit, so
|
|
/// systemd's defaults apply: 5 starts per 10s, `RestartSec` 100ms. That
|
|
/// makes the bound depend on *how fast* a container dies — one that fails
|
|
/// instantly trips the limit in under a second, one that takes longer than
|
|
/// ~2s never trips it and restarts forever. Whether an agent gets bounded
|
|
/// is not meant to be a function of its failure speed.
|
|
///
|
|
/// The window has to exceed the worst-case time to burn the burst, or the
|
|
/// counter ages out between attempts and the limit is again unreachable:
|
|
/// `TimeoutStartSec` is 1min, so `BURST` slow failures plus their backoff
|
|
/// can span several minutes. 10min covers that with room.
|
|
///
|
|
/// Giving up is cheap here **because it is not terminal** — hive-c0re's
|
|
/// reconcile sweep retries later, and `reset-failed` (see `StartContainer`)
|
|
/// clears the latch first. That is what makes a tight burst safe.
|
|
const START_LIMIT_INTERVAL_SEC: u32 = 600;
|
|
/// One start plus two retries — the operator's ruling was "retry once or
|
|
/// twice", with the reconcile sweep as the slow path after that.
|
|
const START_LIMIT_BURST: u32 = 3;
|
|
/// Backoff between those retries. The 100ms default is for processes that
|
|
/// respawn instantly; a container that just failed to boot gains nothing
|
|
/// from being retried a tenth of a second later.
|
|
const RESTART_SEC: u32 = 5;
|
|
|
|
/// Render the body of `hyperhive-limits.conf`.
|
|
///
|
|
/// `[Unit]`: the condition is checked at start time — it skips (not fails)
|
|
/// the unit when the MCP socket dir is absent, avoiding restart loops —
|
|
/// plus the bounded start limit (see the constants above; `StartLimit*` are
|
|
/// `[Unit]` settings since systemd 229 and are silently ignored under
|
|
/// `[Service]`).
|
|
/// `[Service]`: the restart backoff, the hard caps, then the relative
|
|
/// weights. A weight of `None` means "not configured" and omits its line
|
|
/// entirely, so a request from a hive-c0re built before the weights
|
|
/// existed — or one whose nix option is `null` — renders no weight lines.
|
|
fn limits_dropin_body(
|
|
runtime_dir: &str,
|
|
memory_max: &str,
|
|
cpu_quota: &str,
|
|
cpu_weight: Option<u32>,
|
|
io_weight: Option<u32>,
|
|
) -> String {
|
|
// Built as two possibly-empty lines rather than pushed onto the
|
|
// string: `format!` appended to a `String` trips clippy::pedantic's
|
|
// `format_push_string`, and a `write!` would need an unwrap.
|
|
let cpu_weight_line = cpu_weight.map_or_else(String::new, |w| format!("CPUWeight={w}\n"));
|
|
let io_weight_line = io_weight.map_or_else(String::new, |w| format!("IOWeight={w}\n"));
|
|
format!(
|
|
"[Unit]\n\
|
|
ConditionPathIsDirectory={runtime_dir}\n\
|
|
StartLimitIntervalSec={START_LIMIT_INTERVAL_SEC}\n\
|
|
StartLimitBurst={START_LIMIT_BURST}\n\
|
|
\n\
|
|
[Service]\n\
|
|
RestartSec={RESTART_SEC}\n\
|
|
MemoryMax={memory_max}\n\
|
|
CPUQuota={cpu_quota}\n\
|
|
{cpu_weight_line}{io_weight_line}"
|
|
)
|
|
}
|
|
|
|
/// `DaemonReload` — `systemctl daemon-reload` on the host.
|
|
async fn daemon_reload() -> Result<(String, String)> {
|
|
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()))
|
|
}
|
|
|
|
/// `RestartMatrixDaemon` — restart the matrix daemon unit inside the
|
|
/// agent's container.
|
|
async fn restart_matrix_daemon(agent_name: &str) -> Result<(String, String)> {
|
|
validate_agent_name(agent_name)?;
|
|
let machine = format!("--machine=h-{agent_name}");
|
|
let unit = "hive-matrix-daemon.service";
|
|
let out = Command::new("systemctl")
|
|
.args([&machine, "restart", unit])
|
|
.output()
|
|
.await
|
|
.with_context(|| format!("systemctl restart {unit} in container h-{agent_name}"))?;
|
|
if !out.status.success() {
|
|
bail!(
|
|
"systemctl restart {unit} in h-{agent_name} exited {}: {}",
|
|
out.status,
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
Ok((
|
|
String::from_utf8_lossy(&out.stdout).into_owned(),
|
|
String::from_utf8_lossy(&out.stderr).into_owned(),
|
|
))
|
|
}
|
|
|
|
/// `RegisterCiRunner` — write the runner registration token to the host-side
|
|
/// `/run/hive-ci/runner-token` env-file, then restart the in-container runner
|
|
/// so it re-registers. The forge admin token never enters the container; only
|
|
/// the registration token c0re passes here is written, and it lands on a host
|
|
/// path bind-mounted read-only into hive-ci.
|
|
async fn register_ci_runner(token: &str) -> Result<(String, String)> {
|
|
use std::os::unix::fs::PermissionsExt as _;
|
|
// Reject anything that could corrupt the `KEY=VALUE` env-file or smuggle a
|
|
// second line — a forge registration token is an opaque single-line string.
|
|
if token.is_empty() || token.contains(['\n', '\r', '\0']) {
|
|
bail!("ci runner registration token empty or contains control characters");
|
|
}
|
|
let token_path = "/run/hive-ci/runner-token";
|
|
// In-place truncate+write of the existing inode (mirrors the prefetch's
|
|
// `echo > $FILE`), NOT a temp+rename: nspawn pins this file's inode into
|
|
// hive-ci at container start, so a rename would leave the running runner
|
|
// reading the old content. Format + perms match the tmpfiles seed and the
|
|
// prefetch: `TOKEN=<tok>`, mode 0600, root-owned.
|
|
std::fs::write(token_path, format!("TOKEN={token}\n"))
|
|
.with_context(|| format!("write {token_path}"))?;
|
|
std::fs::set_permissions(token_path, std::fs::Permissions::from_mode(0o600))
|
|
.with_context(|| format!("chmod {token_path}"))?;
|
|
// Restart the in-container runner so it reads the new token and registers.
|
|
let out = Command::new("systemctl")
|
|
.args(["--machine=hive-ci", "restart", "gitea-runner-hive.service"])
|
|
.output()
|
|
.await
|
|
.context("systemctl restart gitea-runner-hive.service in hive-ci")?;
|
|
if !out.status.success() {
|
|
bail!(
|
|
"systemctl restart gitea-runner-hive.service in hive-ci exited {}: {}",
|
|
out.status,
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
Ok((
|
|
String::from_utf8_lossy(&out.stdout).into_owned(),
|
|
String::from_utf8_lossy(&out.stderr).into_owned(),
|
|
))
|
|
}
|
|
|
|
/// `ControlInfraContainer` — start/stop/restart a hive infrastructure
|
|
/// service via `systemctl <verb> <unit>`. The [`InfraContainer`] enum is
|
|
/// the allowlist: serde already rejected any unknown / unsafe name
|
|
/// (hive-c0re has no variant, so a stop can't sever the daemon socket) at
|
|
/// deserialisation, so no root-side `.contains()` check is needed here.
|
|
/// Serves both the hive-wide `hivectl stop`/`start` flow and an
|
|
/// `infra_admin` agent's `restart` (action = Restart).
|
|
///
|
|
/// ⚠️ The unit is derived from the variant, never sent by the caller —
|
|
/// which is what keeps this from being a general `systemctl` pass-through.
|
|
/// It is not always `container@<name>.service`: the gateway resolves to the
|
|
/// host's `nginx.service`.
|
|
async fn control_infra_container(
|
|
container: InfraContainer,
|
|
action: InfraAction,
|
|
) -> Result<(String, String)> {
|
|
let verb = action.systemctl_verb();
|
|
let unit = container.service_unit();
|
|
let out = Command::new("systemctl")
|
|
.args([verb, &unit])
|
|
.output()
|
|
.await
|
|
.with_context(|| format!("systemctl {verb} {unit}"))?;
|
|
if !out.status.success() {
|
|
bail!(
|
|
"systemctl {verb} {unit} exited {}: {}",
|
|
out.status,
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
tracing::info!(target: "infra-control", "{verb} {unit}");
|
|
Ok((
|
|
String::from_utf8_lossy(&out.stdout).into_owned(),
|
|
String::from_utf8_lossy(&out.stderr).into_owned(),
|
|
))
|
|
}
|
|
|
|
/// Create/overwrite `dir/filename` at 0600 without following a symlink at the
|
|
/// leaf, returning the open fd for the caller to `fchown`. `filename` must be a
|
|
/// single plain component (no `/`, `.`, `..`) — the leaf sits in an
|
|
/// agent-writable dir, so `O_NOFOLLOW` refuses a planted symlink (`ELOOP`)
|
|
/// instead of letting this root-privileged write/chmod be redirected at another
|
|
/// file; `O_WRONLY` refuses a directory leaf (`EISDIR`); `O_TRUNC` keeps the
|
|
/// overwrite semantics for an existing regular file. `.mode(0o600)` sets the
|
|
/// create mode; the explicit `fchmod` after (on the fd, not a re-resolved path)
|
|
/// tightens an already-existing file and dodges umask. The returned fd is the
|
|
/// exact inode the write hit, so the caller's `fchown` is TOCTOU-immune.
|
|
fn write_state_file_nofollow(dir: &Path, filename: &str, content: &str) -> Result<std::fs::File> {
|
|
use std::io::Write as _;
|
|
use std::os::unix::fs::{OpenOptionsExt as _, PermissionsExt as _};
|
|
|
|
if filename.is_empty() || filename == "." || filename == ".." || filename.contains('/') {
|
|
bail!("write_state_file_nofollow: refusing non-plain filename {filename:?}");
|
|
}
|
|
let path = dir.join(filename);
|
|
let mut file = std::fs::OpenOptions::new()
|
|
.write(true)
|
|
.create(true)
|
|
.truncate(true)
|
|
.mode(0o600)
|
|
.custom_flags(libc::O_NOFOLLOW)
|
|
.open(&path)
|
|
.with_context(|| format!("open (no-follow) {}", path.display()))?;
|
|
file.write_all(content.as_bytes())
|
|
.with_context(|| format!("write {}", path.display()))?;
|
|
file.set_permissions(std::fs::Permissions::from_mode(0o600))
|
|
.with_context(|| format!("chmod 600 {}", path.display()))?;
|
|
Ok(file)
|
|
}
|
|
|
|
/// Sidecar written alongside an extra matrix account's token
|
|
/// (`matrix-account-<name>.json`) so `hive-matrix-mcp` can auto-discover
|
|
/// the account's homeserver without a static `matrixAccounts` config
|
|
/// entry. Read side: `hive-matrix-mcp/src/accounts.rs`'s
|
|
/// `read_account_homeserver` (deliberately reads via a bare
|
|
/// `serde_json::Value` rather than this shape — that side treats a
|
|
/// malformed/missing sidecar as "skip this account" rather than an
|
|
/// error, so it stays loosely typed; this side is the one place the
|
|
/// file is written, so it gets the precise shape).
|
|
#[derive(Serialize)]
|
|
struct MatrixAccountSidecar<'a> {
|
|
homeserver: &'a str,
|
|
}
|
|
|
|
/// Sidecar written alongside a dashboard-provisioned extra forge
|
|
/// account's token (`forge-<label>.json`) so `hive-forge` can resolve
|
|
/// the account's base URL. Read side: `hive-forge/src/client.rs`'s own
|
|
/// (separately defined, deserialize-only) `ForgeSidecar` — same field
|
|
/// name (`base_url`), no shared crate between `hive-priv` and
|
|
/// `hive-forge` to hang a common type off, so the two structs are
|
|
/// pinned to the same JSON key by convention, not by the compiler.
|
|
#[derive(Serialize)]
|
|
struct ForgeSidecar<'a> {
|
|
base_url: &'a str,
|
|
}
|
|
|
|
/// Shared helper for `WriteAgentForgeToken` and `WriteAgentMatrixToken`.
|
|
/// Writes `content` to `AGENT_STATE_ROOT/<agent_name>/state/<filename>`,
|
|
/// chowns to the agent user (derived from the state dir's existing owner),
|
|
/// and chmods 0600. Running as root (hive-priv), so this succeeds
|
|
/// regardless of the file's prior owner/permissions.
|
|
fn write_agent_state_file(
|
|
agent_name: &str,
|
|
filename: &str,
|
|
content: &str,
|
|
) -> Result<(String, String)> {
|
|
let state_dir = PathBuf::from(AGENT_STATE_ROOT)
|
|
.join(agent_name)
|
|
.join("state");
|
|
write_agent_dir_file(agent_name, &state_dir, filename, content)
|
|
}
|
|
|
|
/// Create or remove an agent's pause marker under its harness dir. The
|
|
/// marker is written empty and chowned to the harness dir's owner (the
|
|
/// agent), matching how the harness itself would have created it.
|
|
///
|
|
/// Both directions are idempotent: re-pausing truncates the existing empty
|
|
/// marker rather than failing, and a `NotFound` on removal is the
|
|
/// already-resumed case, not an error.
|
|
fn set_agent_paused(agent_name: &str, paused: bool) -> Result<(String, String)> {
|
|
let harness_dir = PathBuf::from(AGENT_STATE_ROOT)
|
|
.join(agent_name)
|
|
.join("harness");
|
|
if paused {
|
|
return write_agent_dir_file(agent_name, &harness_dir, PAUSED_MARKER_FILE, "");
|
|
}
|
|
remove_marker_in(&harness_dir, PAUSED_MARKER_FILE)?;
|
|
tracing::info!(agent = %agent_name, "cleared pause marker");
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
/// Unlink `dir/filename`, treating "already gone" as success.
|
|
///
|
|
/// `remove_file` unlinks the leaf itself and never follows a symlink, so an
|
|
/// agent-planted link at the marker path cannot redirect this root unlink
|
|
/// at another file — the same threat `write_state_file_nofollow` closes on
|
|
/// the create side.
|
|
fn remove_marker_in(dir: &Path, filename: &str) -> Result<()> {
|
|
let path = dir.join(filename);
|
|
match std::fs::remove_file(&path) {
|
|
Ok(()) => Ok(()),
|
|
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
|
|
Err(e) => Err(e).with_context(|| format!("remove {}", path.display())),
|
|
}
|
|
}
|
|
|
|
/// Write `content` to `dir/filename` as root, chowning the result to `dir`'s
|
|
/// owner so the agent process can read it back. Shared by the credential
|
|
/// writes (which target `state/`) and the pause marker (which targets
|
|
/// `harness/`) — both write into a directory owned by the agent, which is
|
|
/// precisely why they need hive-priv at all.
|
|
fn write_agent_dir_file(
|
|
agent_name: &str,
|
|
dir: &Path,
|
|
filename: &str,
|
|
content: &str,
|
|
) -> Result<(String, String)> {
|
|
use std::os::fd::AsRawFd as _;
|
|
use std::os::unix::fs::MetadataExt as _;
|
|
|
|
let state_dir = dir.to_path_buf();
|
|
// NOTE: `create_dir_all` is normally a no-op — lifecycle creates and chowns
|
|
// the state dir during spawn. On the rare edge where the dir doesn't exist
|
|
// yet (container being provisioned for the first time), the newly created dir
|
|
// is root:root. The `stat state_dir` chown below will then see uid=0 and
|
|
// leave the file root-owned (0600). The agent won't be able to read it until
|
|
// its lifecycle completes. If that happens, a `systemctl restart hive-c0re`
|
|
// after provisioning will re-mint and re-write the token correctly.
|
|
std::fs::create_dir_all(&state_dir)
|
|
.with_context(|| format!("create state dir {}", state_dir.display()))?;
|
|
|
|
// Security-critical: refuses a symlink the (state/-owning) agent may have
|
|
// planted at the leaf, so this root-privileged create/write/chmod/chown
|
|
// can't be redirected at an arbitrary file. See `write_state_file_nofollow`.
|
|
let path = state_dir.join(filename);
|
|
let file = write_state_file_nofollow(&state_dir, filename, content)?;
|
|
|
|
// Chown to the state dir's owner so the agent process can read the file.
|
|
// fchown on the same fd — TOCTOU-immune (the inode the write hit, never a
|
|
// swapped path). If stat fails (e.g. dir just created, owner is root), the
|
|
// file stays root-owned and 0600 — still unreadable by others, just not
|
|
// agent-readable. Log a warning so operators can diagnose.
|
|
match std::fs::metadata(&state_dir) {
|
|
Ok(meta) => {
|
|
// SAFETY: `file` is an open, owned fd live for the whole call;
|
|
// `fchown` only mutates that inode's uid/gid.
|
|
let rc = unsafe { libc::fchown(file.as_raw_fd(), meta.uid(), meta.gid()) };
|
|
if rc != 0 {
|
|
let e = std::io::Error::last_os_error();
|
|
tracing::warn!(
|
|
agent = %agent_name,
|
|
path = %path.display(),
|
|
error = %e,
|
|
"write_agent_state_file: fchown failed"
|
|
);
|
|
}
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
agent = %agent_name,
|
|
error = %e,
|
|
"write_agent_state_file: stat state_dir failed, leaving file root-owned"
|
|
);
|
|
}
|
|
}
|
|
tracing::info!(
|
|
agent = %agent_name,
|
|
dir = %state_dir.display(),
|
|
file = %filename,
|
|
"wrote agent file"
|
|
);
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
/// Remove `AGENT_STATE_ROOT/<agent_name>/state/<filename>` if present.
|
|
/// Idempotent revoke counterpart to [`write_agent_state_file`] — a
|
|
/// missing file is success, not an error. `filename` must be a single
|
|
/// plain component (no `/`, `.`, `..`); callers pass a pre-validated
|
|
/// label into a fixed `forge-<label>-token` shape, same as the write
|
|
/// side.
|
|
fn delete_agent_state_file(agent_name: &str, filename: &str) -> Result<(String, String)> {
|
|
if filename.is_empty() || filename == "." || filename == ".." || filename.contains('/') {
|
|
bail!("delete_agent_state_file: refusing non-plain filename {filename:?}");
|
|
}
|
|
let path = PathBuf::from(AGENT_STATE_ROOT)
|
|
.join(agent_name)
|
|
.join("state")
|
|
.join(filename);
|
|
match std::fs::remove_file(&path) {
|
|
Ok(()) => {
|
|
tracing::info!(agent = %agent_name, file = %filename, "removed agent state file");
|
|
}
|
|
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
|
|
Err(e) => return Err(e).with_context(|| format!("remove {}", path.display())),
|
|
}
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
/// btrfs superblock magic, as reported by `statfs(2)`'s `f_type`.
|
|
const BTRFS_SUPER_MAGIC: i64 = 0x9123_683E;
|
|
|
|
/// Inode number of a btrfs subvolume root (`BTRFS_FIRST_FREE_OBJECTID`).
|
|
/// Every subvolume's top directory has this inode; plain directories do
|
|
/// not, so `statfs == btrfs && st_ino == 256` reliably identifies a
|
|
/// subvolume root.
|
|
const BTRFS_SUBVOL_ROOT_INO: u64 = 256;
|
|
|
|
/// Whether `path` lives on a btrfs filesystem (via `statfs(2)`).
|
|
fn is_on_btrfs(path: &Path) -> Result<bool> {
|
|
use std::os::unix::ffi::OsStrExt as _;
|
|
let c_path = std::ffi::CString::new(path.as_os_str().as_bytes())
|
|
.with_context(|| format!("path {} has an interior null byte", path.display()))?;
|
|
// SAFETY: `c_path` is a valid NUL-terminated C string that outlives the
|
|
// call; `statfs` only writes into the zero-initialised `buf`.
|
|
let mut buf: libc::statfs = unsafe { std::mem::zeroed() };
|
|
let rc = unsafe { libc::statfs(c_path.as_ptr(), &raw mut buf) };
|
|
if rc != 0 {
|
|
return Err(std::io::Error::last_os_error())
|
|
.with_context(|| format!("statfs {}", path.display()));
|
|
}
|
|
Ok(buf.f_type == BTRFS_SUPER_MAGIC)
|
|
}
|
|
|
|
/// Whether `path` is the root of a btrfs subvolume (on btrfs and inode 256).
|
|
fn is_btrfs_subvolume(path: &Path) -> bool {
|
|
use std::os::unix::fs::MetadataExt as _;
|
|
let on_btrfs = is_on_btrfs(path).unwrap_or(false);
|
|
let ino_match = std::fs::metadata(path).is_ok_and(|m| m.ino() == BTRFS_SUBVOL_ROOT_INO);
|
|
on_btrfs && ino_match
|
|
}
|
|
|
|
/// `EnsureAgentSubvolume` — make the agent's state root a btrfs subvolume
|
|
/// when the FS supports it. Idempotent + progressive: no-op when the root
|
|
/// already exists or the FS isn't btrfs. See the wire doc on the variant.
|
|
async fn ensure_agent_subvolume(agent_name: &str) -> Result<(String, String)> {
|
|
use std::os::unix::fs::MetadataExt as _;
|
|
|
|
let root = PathBuf::from(AGENT_STATE_ROOT);
|
|
let agent_root = root.join(agent_name);
|
|
|
|
// Progressive: existing agents (plain dir OR already a subvol) are left
|
|
// untouched — never auto-migrated.
|
|
if agent_root.exists() {
|
|
return Ok((String::new(), String::new()));
|
|
}
|
|
|
|
// Only btrfs supports subvolumes; on anything else hive-c0re's normal
|
|
// `create_dir_all` makes a plain dir (the pre-subvolume behaviour). The
|
|
// parent must exist for both statfs and `btrfs subvolume create`.
|
|
std::fs::create_dir_all(&root)
|
|
.with_context(|| format!("create agents root {}", root.display()))?;
|
|
if !is_on_btrfs(&root)? {
|
|
return Ok((String::new(), String::new()));
|
|
}
|
|
|
|
let out = Command::new("btrfs")
|
|
.args(["subvolume", "create"])
|
|
.arg(&agent_root)
|
|
.output()
|
|
.await
|
|
.with_context(|| format!("spawn btrfs subvolume create {}", agent_root.display()))?;
|
|
if !out.status.success() {
|
|
bail!(
|
|
"btrfs subvolume create {} failed: {}",
|
|
agent_root.display(),
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
|
|
// The subvol root is created root-owned, but hive-c0re (the `hive-core`
|
|
// user) must be able to mkdir state/ claude/ harness/ inside it — exactly
|
|
// as it would in a plain dir. Chown it to AGENT_STATE_ROOT's owner
|
|
// (hive-core). This MUST succeed: a root-owned subvol would make the
|
|
// downstream dir creation fail with a confusing permission error, and the
|
|
// c0re-side exists-check would then skip re-running this op on retry,
|
|
// wedging the agent. So on any failure roll the subvol back and bail — the
|
|
// create path surfaces a clear error and a retry starts clean.
|
|
let chown_result = std::fs::metadata(&root)
|
|
.with_context(|| format!("stat agents root {} for ownership", root.display()))
|
|
.and_then(|meta| {
|
|
std::os::unix::fs::chown(&agent_root, Some(meta.uid()), Some(meta.gid())).with_context(
|
|
|| format!("chown subvol {} to agents-root owner", agent_root.display()),
|
|
)
|
|
});
|
|
if let Err(e) = chown_result {
|
|
// Best-effort rollback so we never leave a root-owned subvol behind.
|
|
let _ = Command::new("btrfs")
|
|
.args(["subvolume", "delete"])
|
|
.arg(&agent_root)
|
|
.output()
|
|
.await;
|
|
return Err(e.context(format!(
|
|
"rolled back subvolume {} after chown failed",
|
|
agent_root.display()
|
|
)));
|
|
}
|
|
tracing::info!(agent = %agent_name, path = %agent_root.display(), "created agent state subvolume");
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
/// `DeleteAgentSubvolume` — delete the agent's state root iff it's a btrfs
|
|
/// subvolume (purge path only). No-op for plain dirs / missing paths;
|
|
/// hive-c0re's own `remove_dir_all` covers those. See the wire doc.
|
|
async fn delete_agent_subvolume(agent_name: &str) -> Result<(String, String)> {
|
|
let agent_root = PathBuf::from(AGENT_STATE_ROOT).join(agent_name);
|
|
if !is_btrfs_subvolume(&agent_root) {
|
|
return Ok((String::new(), String::new()));
|
|
}
|
|
let out = Command::new("btrfs")
|
|
.args(["subvolume", "delete"])
|
|
.arg(&agent_root)
|
|
.output()
|
|
.await
|
|
.with_context(|| format!("spawn btrfs subvolume delete {}", agent_root.display()))?;
|
|
if !out.status.success() {
|
|
bail!(
|
|
"btrfs subvolume delete {} failed: {}",
|
|
agent_root.display(),
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
tracing::info!(agent = %agent_name, path = %agent_root.display(), "deleted agent state subvolume");
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
/// `EnsureBtrfsQuota` — enable btrfs qgroup accounting on the filesystem
|
|
/// holding `AGENT_STATE_ROOT`. Idempotent + statfs-gated (no-op off btrfs).
|
|
/// Operator opt-in only; see the wire doc.
|
|
async fn ensure_btrfs_quota() -> Result<(String, String)> {
|
|
let root = PathBuf::from(AGENT_STATE_ROOT);
|
|
std::fs::create_dir_all(&root)
|
|
.with_context(|| format!("create agents root {}", root.display()))?;
|
|
if !is_on_btrfs(&root)? {
|
|
// Non-btrfs host: quota/qgroups don't apply. No-op success so the
|
|
// operator-facing verb degrades cleanly.
|
|
return Ok((
|
|
String::new(),
|
|
"filesystem is not btrfs — quota not applicable".to_owned(),
|
|
));
|
|
}
|
|
let out = Command::new("btrfs")
|
|
.args(["quota", "enable"])
|
|
.arg(&root)
|
|
.output()
|
|
.await
|
|
.with_context(|| format!("spawn btrfs quota enable {}", root.display()))?;
|
|
if !out.status.success() {
|
|
bail!(
|
|
"btrfs quota enable {} failed: {}",
|
|
root.display(),
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
tracing::info!(path = %root.display(), "enabled btrfs qgroup accounting");
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
/// `ReadSubvolumeUsage` — return an agent subvolume's qgroup rows
|
|
/// (`btrfs qgroup show -f --raw <…/agent_name>`) verbatim in stdout for
|
|
/// hive-c0re to parse. `-f` lists the qgroups impacting the given path,
|
|
/// excluding ancestral qgroups (per btrfs-qgroup-show(8)) — so it scopes
|
|
/// to this subvolume and never mixes in other agents'. hive-c0re then
|
|
/// selects the level-0 (`0/<subvolid>`) leaf row. See the wire doc.
|
|
async fn read_subvolume_usage(agent_name: &str) -> Result<(String, String)> {
|
|
let agent_root = PathBuf::from(AGENT_STATE_ROOT).join(agent_name);
|
|
let out = Command::new("btrfs")
|
|
.args(["qgroup", "show", "-f", "--raw"])
|
|
.arg(&agent_root)
|
|
.output()
|
|
.await
|
|
.with_context(|| format!("spawn btrfs qgroup show {}", agent_root.display()))?;
|
|
if !out.status.success() {
|
|
// The common failure is "quota not enabled" — pass the stderr
|
|
// through so hive-c0re can surface it gracefully.
|
|
bail!(
|
|
"btrfs qgroup show {} failed: {}",
|
|
agent_root.display(),
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
Ok((
|
|
String::from_utf8_lossy(&out.stdout).into_owned(),
|
|
String::new(),
|
|
))
|
|
}
|
|
|
|
/// Best-effort removal of a leftover migration path from a prior aborted
|
|
/// upgrade: try `btrfs subvolume delete` (in case it's a half-created
|
|
/// subvolume) then a plain recursive remove. Both failures are ignored —
|
|
/// the path may simply not exist.
|
|
async fn cleanup_stale_path(path: &Path) {
|
|
if path.exists() {
|
|
let _ = Command::new("btrfs")
|
|
.args(["subvolume", "delete"])
|
|
.arg(path)
|
|
.output()
|
|
.await;
|
|
let _ = std::fs::remove_dir_all(path);
|
|
}
|
|
}
|
|
|
|
/// Stage a populated subvolume at `tmp` mirroring `agent_root`: create the
|
|
/// subvolume, copy `agent_root`'s contents into it preserving
|
|
/// ownership/permissions/xattrs, then match the subvolume root's owner + mode
|
|
/// to the original. On any failure the partially-staged `tmp` is cleaned up
|
|
/// (so the caller can bail with the original dir still untouched).
|
|
async fn stage_upgrade_subvolume(agent_root: &Path, tmp: &Path) -> Result<()> {
|
|
use std::os::unix::fs::{MetadataExt as _, PermissionsExt as _};
|
|
|
|
// Fresh subvolume to receive the copy.
|
|
let out = Command::new("btrfs")
|
|
.args(["subvolume", "create"])
|
|
.arg(tmp)
|
|
.output()
|
|
.await
|
|
.with_context(|| format!("spawn btrfs subvolume create {}", tmp.display()))?;
|
|
if !out.status.success() {
|
|
bail!(
|
|
"btrfs subvolume create {} failed: {}",
|
|
tmp.display(),
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
|
|
// Copy contents preserving everything (`-a` = --preserve=all → mode,
|
|
// ownership, timestamps, links, xattrs); reflink for fast CoW clones on the
|
|
// same btrfs. `<src>/.` copies the directory's contents (incl. dotfiles)
|
|
// into the subvolume rather than nesting it.
|
|
let copy = Command::new("cp")
|
|
.arg("-a")
|
|
.arg("--reflink=auto")
|
|
.arg(format!("{}/.", agent_root.display()))
|
|
.arg(tmp)
|
|
.output()
|
|
.await
|
|
.with_context(|| format!("spawn cp into {}", tmp.display()))?;
|
|
if !copy.status.success() {
|
|
cleanup_stale_path(tmp).await;
|
|
bail!(
|
|
"copy {} -> {} failed (original left untouched): {}",
|
|
agent_root.display(),
|
|
tmp.display(),
|
|
String::from_utf8_lossy(©.stderr).trim()
|
|
);
|
|
}
|
|
|
|
// Match the new subvolume root's ownership + mode to the original dir.
|
|
// `cp -a <src>/.` copies the *contents* but the subvolume root keeps its
|
|
// create-time root ownership, so set it explicitly — the swapped-in
|
|
// subvolume must be indistinguishable from the original to hive-c0re.
|
|
let apply = std::fs::metadata(agent_root)
|
|
.with_context(|| format!("stat {} for ownership", agent_root.display()))
|
|
.and_then(|m| {
|
|
std::os::unix::fs::chown(tmp, Some(m.uid()), Some(m.gid()))
|
|
.with_context(|| format!("chown {} to match original", tmp.display()))?;
|
|
std::fs::set_permissions(tmp, std::fs::Permissions::from_mode(m.mode()))
|
|
.with_context(|| format!("chmod {} to match original", tmp.display()))?;
|
|
Ok(())
|
|
});
|
|
if let Err(e) = apply {
|
|
cleanup_stale_path(tmp).await;
|
|
return Err(e.context("upgrade aborted before swap; original left untouched"));
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// `UpgradeAgentSubvolume` — convert an existing plain-dir agent state root
|
|
/// into a btrfs subvolume in place. Operator opt-in; the caller (hivectl)
|
|
/// stops the agent first and restarts it after. See the wire doc.
|
|
///
|
|
/// Migration: stage a sibling subvolume mirroring the dir
|
|
/// ([`stage_upgrade_subvolume`]), then rename the original aside and the
|
|
/// subvolume into place, then remove the original. Any failure before the
|
|
/// rename-swap leaves the original dir untouched.
|
|
async fn upgrade_agent_subvolume(agent_name: &str) -> Result<(String, String)> {
|
|
let root = PathBuf::from(AGENT_STATE_ROOT);
|
|
let agent_root = root.join(agent_name);
|
|
|
|
if !agent_root.exists() {
|
|
// A crash between the two swap renames (original → `.<name>.old`
|
|
// succeeded, `.<name>.migrating` → agent_root did not) leaves the
|
|
// agent root missing but the original data intact under `.<name>.old`.
|
|
// Point at the recovery rather than a bare "nothing to do" so the
|
|
// operator isn't left guessing where the data went.
|
|
let old = root.join(format!(".{agent_name}.old"));
|
|
if old.exists() {
|
|
bail!(
|
|
"no state dir at {agent} — but {old} holds the original data from an \
|
|
interrupted upgrade (host crashed mid-swap). Restore it with \
|
|
`mv {old} {agent}`, then re-run the upgrade.",
|
|
agent = agent_root.display(),
|
|
old = old.display(),
|
|
);
|
|
}
|
|
bail!(
|
|
"no state dir to upgrade at {} — nothing to do",
|
|
agent_root.display()
|
|
);
|
|
}
|
|
// Idempotent: already a subvolume → nothing to do.
|
|
if is_btrfs_subvolume(&agent_root) {
|
|
return Ok((String::new(), String::new()));
|
|
}
|
|
if !is_on_btrfs(&root)? {
|
|
bail!(
|
|
"{} is not on btrfs — subvolumes are unsupported, cannot upgrade",
|
|
root.display()
|
|
);
|
|
}
|
|
|
|
// Sibling temp paths on the same filesystem (so the copy can reflink and
|
|
// the swap renames are atomic). Leading dots keep them out of the agent
|
|
// namespace (`validate_agent_name` rejects dot-prefixed names).
|
|
let tmp = root.join(format!(".{agent_name}.migrating"));
|
|
let old = root.join(format!(".{agent_name}.old"));
|
|
// Clear any debris from a previously interrupted run before starting.
|
|
cleanup_stale_path(&tmp).await;
|
|
cleanup_stale_path(&old).await;
|
|
|
|
stage_upgrade_subvolume(&agent_root, &tmp).await?;
|
|
|
|
// Swap. `rename` is atomic within a filesystem. The window between the
|
|
// two renames is the only unsafe point: a crash there leaves the agent
|
|
// root missing but both `.old` (original) and the new subvolume present
|
|
// — recoverable by hand, hence the loud logging.
|
|
if let Err(e) = std::fs::rename(&agent_root, &old) {
|
|
cleanup_stale_path(&tmp).await;
|
|
return Err(anyhow::Error::new(e).context(format!(
|
|
"rename {} -> {} failed; original left untouched",
|
|
agent_root.display(),
|
|
old.display()
|
|
)));
|
|
}
|
|
if let Err(e) = std::fs::rename(&tmp, &agent_root) {
|
|
// Restore the original from its renamed-aside copy.
|
|
let restored = std::fs::rename(&old, &agent_root).is_ok();
|
|
cleanup_stale_path(&tmp).await;
|
|
return Err(anyhow::Error::new(e).context(format!(
|
|
"rename {} -> {} failed; original {}",
|
|
tmp.display(),
|
|
agent_root.display(),
|
|
if restored {
|
|
"restored"
|
|
} else {
|
|
"COULD NOT BE RESTORED — manual recovery needed"
|
|
}
|
|
)));
|
|
}
|
|
|
|
// 5. Success: drop the original (a plain dir) and report.
|
|
if let Err(e) = std::fs::remove_dir_all(&old) {
|
|
// The migration succeeded; a leftover `.old` is cosmetic. Warn only.
|
|
tracing::warn!(
|
|
agent = %agent_name, path = %old.display(),
|
|
"upgraded subvolume but failed to remove old dir: {e}"
|
|
);
|
|
}
|
|
tracing::info!(
|
|
agent = %agent_name, path = %agent_root.display(),
|
|
"upgraded agent state dir to btrfs subvolume"
|
|
);
|
|
Ok((
|
|
format!("upgraded {} to a btrfs subvolume", agent_root.display()),
|
|
String::new(),
|
|
))
|
|
}
|
|
|
|
/// Derive a snapshot's path from the agent name + label: a dot-prefixed
|
|
/// sibling of the agent's state root so it can never collide with a real
|
|
/// agent directory (`validate_agent_name` rejects dot-prefixed names).
|
|
fn snapshot_path(agent_name: &str, snapshot_name: &str) -> PathBuf {
|
|
PathBuf::from(AGENT_STATE_ROOT).join(format!(".{agent_name}.snapshot.{snapshot_name}"))
|
|
}
|
|
|
|
/// `SnapshotAgentSubvolume` — create a read-only btrfs snapshot of an
|
|
/// agent's state subvolume, for `btrfs send` to stream from during
|
|
/// inter-hive migration. See the wire doc.
|
|
async fn snapshot_agent_subvolume(
|
|
agent_name: &str,
|
|
snapshot_name: &str,
|
|
) -> Result<(String, String)> {
|
|
let agent_root = PathBuf::from(AGENT_STATE_ROOT).join(agent_name);
|
|
if !is_btrfs_subvolume(&agent_root) {
|
|
bail!(
|
|
"{} is not a btrfs subvolume — nothing to snapshot (run `hivectl agent <name> subvol upgrade` first)",
|
|
agent_root.display()
|
|
);
|
|
}
|
|
let snap = snapshot_path(agent_name, snapshot_name);
|
|
if snap.exists() {
|
|
bail!(
|
|
"snapshot {} already exists — delete it first or pick a different name",
|
|
snap.display()
|
|
);
|
|
}
|
|
let out = Command::new("btrfs")
|
|
.args(["subvolume", "snapshot", "-r"])
|
|
.arg(&agent_root)
|
|
.arg(&snap)
|
|
.output()
|
|
.await
|
|
.with_context(|| {
|
|
format!(
|
|
"spawn btrfs subvolume snapshot -r {} {}",
|
|
agent_root.display(),
|
|
snap.display()
|
|
)
|
|
})?;
|
|
if !out.status.success() {
|
|
bail!(
|
|
"btrfs subvolume snapshot -r {} {} failed: {}",
|
|
agent_root.display(),
|
|
snap.display(),
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
tracing::info!(
|
|
agent = %agent_name, snapshot = %snap.display(),
|
|
"created read-only agent state snapshot"
|
|
);
|
|
Ok((snap.display().to_string(), String::new()))
|
|
}
|
|
|
|
/// `DeleteAgentSnapshot` — delete a previously-created read-only snapshot.
|
|
/// No-op if the path doesn't exist. See the wire doc.
|
|
async fn delete_agent_snapshot(agent_name: &str, snapshot_name: &str) -> Result<(String, String)> {
|
|
let snap = snapshot_path(agent_name, snapshot_name);
|
|
if !snap.exists() {
|
|
return Ok((String::new(), String::new()));
|
|
}
|
|
let out = Command::new("btrfs")
|
|
.args(["subvolume", "delete"])
|
|
.arg(&snap)
|
|
.output()
|
|
.await
|
|
.with_context(|| format!("spawn btrfs subvolume delete {}", snap.display()))?;
|
|
if !out.status.success() {
|
|
bail!(
|
|
"btrfs subvolume delete {} failed: {}",
|
|
snap.display(),
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
tracing::info!(agent = %agent_name, snapshot = %snap.display(), "deleted agent state snapshot");
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
/// `SendAgentSnapshotToFile` — stream a read-only snapshot (optionally
|
|
/// incremental against `parent_name`) to a file under
|
|
/// `MIGRATE_STAGING_ROOT` via `btrfs send`. Local-file half of the
|
|
/// inter-hive migration transport; see `PrivRequest::SendAgentSnapshotToFile`
|
|
/// for the cross-hive follow-up.
|
|
async fn send_agent_snapshot_to_file(
|
|
agent_name: &str,
|
|
snapshot_name: &str,
|
|
parent_name: Option<&str>,
|
|
dest_file_name: &str,
|
|
) -> Result<(String, String)> {
|
|
let snap = snapshot_path(agent_name, snapshot_name);
|
|
if !snap.exists() {
|
|
bail!(
|
|
"snapshot {} does not exist — create it with `subvol snapshot create` first",
|
|
snap.display()
|
|
);
|
|
}
|
|
|
|
std::fs::create_dir_all(MIGRATE_STAGING_ROOT)
|
|
.with_context(|| format!("create {MIGRATE_STAGING_ROOT}"))?;
|
|
let dest = Path::new(MIGRATE_STAGING_ROOT).join(dest_file_name);
|
|
// `create_new` (O_CREAT|O_EXCL) makes the no-overwrite guarantee atomic
|
|
// instead of a check-then-create race against a concurrent request.
|
|
let dest_file = match std::fs::File::options()
|
|
.write(true)
|
|
.create_new(true)
|
|
.open(&dest)
|
|
{
|
|
Ok(f) => f,
|
|
Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => bail!(
|
|
"{} already exists — pick a different destination or remove it first \
|
|
(send never overwrites an existing export)",
|
|
dest.display()
|
|
),
|
|
Err(e) => {
|
|
return Err(e).with_context(|| format!("create {}", dest.display()));
|
|
}
|
|
};
|
|
|
|
let mut cmd = Command::new("btrfs");
|
|
cmd.arg("send");
|
|
if let Some(parent) = parent_name {
|
|
let parent_path = snapshot_path(agent_name, parent);
|
|
if !parent_path.exists() {
|
|
bail!(
|
|
"parent snapshot {} does not exist — pick an existing parent or omit it for a full send",
|
|
parent_path.display()
|
|
);
|
|
}
|
|
cmd.arg("-p").arg(&parent_path);
|
|
}
|
|
cmd.arg(&snap);
|
|
cmd.stdout(std::process::Stdio::from(dest_file));
|
|
cmd.stderr(std::process::Stdio::piped());
|
|
|
|
let out = cmd
|
|
.spawn()
|
|
.with_context(|| format!("spawn btrfs send {}", snap.display()))?
|
|
.wait_with_output()
|
|
.await
|
|
.with_context(|| format!("wait on btrfs send {}", snap.display()))?;
|
|
if !out.status.success() {
|
|
// Clean up a partial/failed export so a retry doesn't trip the
|
|
// "already exists" guard on garbage. Best-effort: warn (don't fail
|
|
// the whole call over it) if removal itself fails, so a stuck
|
|
// partial file that later masquerades as a completed export is at
|
|
// least visible in the log.
|
|
if let Err(rm_err) = std::fs::remove_file(&dest) {
|
|
tracing::warn!(
|
|
dest = %dest.display(), error = %rm_err,
|
|
"failed to remove partial export after btrfs send failure — \
|
|
next attempt at this dest will hit the already-exists guard"
|
|
);
|
|
}
|
|
bail!(
|
|
"btrfs send {} failed: {}",
|
|
snap.display(),
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
tracing::info!(
|
|
agent = %agent_name, snapshot = %snap.display(), dest = %dest.display(),
|
|
parent = ?parent_name, "exported agent snapshot to file"
|
|
);
|
|
Ok((dest.display().to_string(), String::new()))
|
|
}
|
|
|
|
/// `SendAgentSnapshotToFd` — stream a read-only snapshot (optionally
|
|
/// incremental against `parent_name`) straight into a descriptor the
|
|
/// caller passed us.
|
|
///
|
|
/// The network half of the inter-hive migration transport, arranged so
|
|
/// this helper never learns there *is* a network: hive-c0re connects to
|
|
/// the peer's snapshot store, writes the header itself, and hands the
|
|
/// connected socket over. We only ever see "a thing to write bytes into",
|
|
/// which keeps a root process out of any address, protocol or trust
|
|
/// decision — and keeps everyone out of the data path once `btrfs send`
|
|
/// starts, which matters at multi-gigabyte sizes.
|
|
async fn send_agent_snapshot_to_fd(
|
|
agent_name: &str,
|
|
snapshot_name: &str,
|
|
parent_name: Option<&str>,
|
|
dest: OwnedFd,
|
|
) -> Result<(String, String)> {
|
|
let snap = snapshot_path(agent_name, snapshot_name);
|
|
if !snap.exists() {
|
|
bail!(
|
|
"snapshot {} does not exist — create it with `subvol snapshot create` first",
|
|
snap.display()
|
|
);
|
|
}
|
|
|
|
let mut cmd = Command::new("btrfs");
|
|
cmd.arg("send");
|
|
if let Some(parent) = parent_name {
|
|
let parent_path = snapshot_path(agent_name, parent);
|
|
if !parent_path.exists() {
|
|
bail!(
|
|
"parent snapshot {} does not exist — pick an existing parent or omit it for a full send",
|
|
parent_path.display()
|
|
);
|
|
}
|
|
cmd.arg("-p").arg(&parent_path);
|
|
}
|
|
cmd.arg(&snap);
|
|
cmd.stdout(std::process::Stdio::from(dest));
|
|
cmd.stderr(std::process::Stdio::piped());
|
|
|
|
let out = cmd
|
|
.spawn()
|
|
.with_context(|| format!("spawn btrfs send {}", snap.display()))?
|
|
.wait_with_output()
|
|
.await
|
|
.with_context(|| format!("wait on btrfs send {}", snap.display()))?;
|
|
if !out.status.success() {
|
|
// Nothing to clean up: the destination isn't ours. A partial
|
|
// stream is the receiving end's problem, and `btrfs receive`
|
|
// refuses to commit an incomplete subvolume anyway.
|
|
bail!(
|
|
"btrfs send {} failed: {}",
|
|
snap.display(),
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
tracing::info!(
|
|
agent = %agent_name, snapshot = %snap.display(), parent = ?parent_name,
|
|
"streamed agent snapshot into a passed descriptor"
|
|
);
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
/// `SetSubvolumeQuota` — set or clear a qgroup size limit on an agent
|
|
/// subvolume (`btrfs qgroup limit <bytes|none> <…/agent_name>`). See the
|
|
/// wire doc.
|
|
async fn set_subvolume_quota(
|
|
agent_name: &str,
|
|
limit_bytes: Option<u64>,
|
|
) -> Result<(String, String)> {
|
|
let agent_root = PathBuf::from(AGENT_STATE_ROOT).join(agent_name);
|
|
let limit = limit_bytes.map_or_else(|| "none".to_owned(), |n| n.to_string());
|
|
let out = Command::new("btrfs")
|
|
.args(["qgroup", "limit", &limit])
|
|
.arg(&agent_root)
|
|
.output()
|
|
.await
|
|
.with_context(|| format!("spawn btrfs qgroup limit {}", agent_root.display()))?;
|
|
if !out.status.success() {
|
|
bail!(
|
|
"btrfs qgroup limit {limit} {} failed: {}",
|
|
agent_root.display(),
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
tracing::info!(agent = %agent_name, %limit, "set agent subvolume quota");
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
/// Validate a single argument destined for `forgejo admin`. Rejects
|
|
/// null bytes and newlines (which could corrupt the subprocess args list
|
|
/// or log output). Shell metacharacters are harmless since the command
|
|
/// is spawned directly (no shell), but we reject them defensively.
|
|
fn validate_forge_admin_arg(arg: &str) -> Result<()> {
|
|
if arg.bytes().any(|b| b == 0 || b == b'\n' || b == b'\r') {
|
|
bail!("forge admin arg {arg:?} contains null byte or newline");
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Redact a line before it hits the (root-readable, but still
|
|
/// unnecessarily exposed) host journal.
|
|
///
|
|
/// Two independent rules, because the previous single rule failed open.
|
|
/// It matched only the substring "password", chosen to be robust against
|
|
/// forgejo *rewording* its password line — and the leak arrived from the
|
|
/// other axis entirely: a **different kind of secret** on a differently
|
|
/// worded line. `forgejo admin user generate-access-token` prints
|
|
/// `Access token was successfully created: <40 hex>`, which contains no
|
|
/// "password" and went to the journal verbatim for every agent ever
|
|
/// provisioned.
|
|
///
|
|
/// So the second rule matches on **shape, not vocabulary**: a long
|
|
/// unbroken run of secret-alphabet characters. A new secret type is then
|
|
/// caught by default rather than by someone remembering to add a keyword.
|
|
///
|
|
/// ⚠️ This deliberately over-matches. A nix store hash is also a long
|
|
/// opaque run and will redact its line. That is the correct direction to
|
|
/// be wrong in: the cost of a false positive is one less log line, and
|
|
/// the cost of a false negative is a live credential in a journal that
|
|
/// any `read_host_journal` holder can read.
|
|
fn redact_secret_line(line: &str) -> std::borrow::Cow<'_, str> {
|
|
if line.to_ascii_lowercase().contains("password") {
|
|
return std::borrow::Cow::Borrowed("[redacted: line mentions a password]");
|
|
}
|
|
if contains_secret_shaped_run(line) {
|
|
return std::borrow::Cow::Borrowed("[redacted: line contains a secret-shaped token]");
|
|
}
|
|
std::borrow::Cow::Borrowed(line)
|
|
}
|
|
|
|
/// True when the line contains an unbroken run of at least 32 characters
|
|
/// from the hex / base64url alphabet. 32 sits below forgejo's 40-hex
|
|
/// access token and above the ordinary words and path segments that
|
|
/// appear in `forgejo admin` output.
|
|
///
|
|
/// ⚠️ Scans for a RUN, not for a whitespace-delimited word. An earlier
|
|
/// version split on whitespace and required the whole word to match,
|
|
/// which a secret with punctuation glued to it defeats: `"<token>,"` and
|
|
/// `"[<token>]"` both fail an all-chars check on the word while still
|
|
/// containing the credential in full. Whitespace is not what delimits a
|
|
/// secret — the alphabet is (thanks @argus for catching it).
|
|
fn contains_secret_shaped_run(line: &str) -> bool {
|
|
const MIN: usize = 32;
|
|
let mut run = 0usize;
|
|
for b in line.bytes() {
|
|
if b.is_ascii_alphanumeric() || matches!(b, b'+' | b'/' | b'=' | b'_' | b'-') {
|
|
run += 1;
|
|
if run >= MIN {
|
|
return true;
|
|
}
|
|
} else {
|
|
run = 0;
|
|
}
|
|
}
|
|
false
|
|
}
|
|
|
|
/// Run `forgejo admin <args>` inside the `hive-forge` container as the
|
|
/// `forgejo` unix user. Requires root (for nsenter into the container's
|
|
/// namespaces). Returns `(stdout, stderr)`.
|
|
async fn run_forge_admin(args: &[String]) -> Result<(String, String)> {
|
|
let mut cmd_args: Vec<&str> = vec![
|
|
"run",
|
|
"hive-forge",
|
|
"--",
|
|
"runuser",
|
|
"-u",
|
|
"forgejo",
|
|
"--",
|
|
"forgejo",
|
|
"--work-path",
|
|
"/var/lib/forgejo",
|
|
"admin",
|
|
];
|
|
for a in args {
|
|
cmd_args.push(a.as_str());
|
|
}
|
|
let out = Command::new("nixos-container")
|
|
.args(&cmd_args)
|
|
.output()
|
|
.await
|
|
.context("invoke nixos-container run hive-forge -- forgejo admin")?;
|
|
let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
|
|
let stderr = String::from_utf8_lossy(&out.stderr).into_owned();
|
|
// stdout at DEBUG, not INFO: on the success path this stream carries
|
|
// the *product* of the command (the freshly minted token, the created
|
|
// user's details) and nothing an operator needs at default verbosity.
|
|
// Redaction stays on as the second layer — the level decides who sees
|
|
// it, the redactor decides what it says, and neither alone is enough.
|
|
for line in stdout.lines() {
|
|
tracing::debug!(target: "forgejo-admin", "{}", redact_secret_line(line));
|
|
}
|
|
for line in stderr.lines() {
|
|
tracing::warn!(target: "forgejo-admin", "{}", redact_secret_line(line));
|
|
}
|
|
if !out.status.success() {
|
|
// Redact here too. The error string is propagated to the caller and
|
|
// ends up logged; a partial-failure stderr can carry the same
|
|
// material stdout would have. Redacting the log but not the error
|
|
// is the same "two of three sites" gap that makes these leaks
|
|
// survive a fix.
|
|
let safe_stderr: String = stderr
|
|
.lines()
|
|
.map(|l| redact_secret_line(l).into_owned())
|
|
.collect::<Vec<_>>()
|
|
.join("; ");
|
|
bail!(
|
|
"forgejo admin {} failed ({}): {}",
|
|
args.join(" "),
|
|
out.status,
|
|
safe_stderr.trim()
|
|
);
|
|
}
|
|
Ok((stdout, stderr))
|
|
}
|
|
|
|
/// Invoke `nixos-container` with the given args, log output to journald.
|
|
async fn container_run(args: &[&str]) -> Result<(String, String)> {
|
|
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();
|
|
// `list` is a read-only enumeration called on the hot path (dashboard
|
|
// rescan, forge + boot sweeps) — its stdout is the return value, not
|
|
// progress, so logging every container name on every call floods the
|
|
// journal. Log stdout only for the mutating ops, where each line is
|
|
// genuine progress. stderr is always logged (errors matter regardless).
|
|
if args.first() != Some(&"list") {
|
|
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))
|
|
}
|
|
|
|
/// Invoke `machinectl` with the given args, log output to journald.
|
|
/// Used for operations that nixos-container doesn't expose (e.g. sending
|
|
/// signals to running containers).
|
|
async fn machinectl_run(args: &[&str]) -> Result<(String, String)> {
|
|
let out = Command::new("machinectl")
|
|
.args(args)
|
|
.output()
|
|
.await
|
|
.context("invoke machinectl")?;
|
|
let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
|
|
let stderr = String::from_utf8_lossy(&out.stderr).into_owned();
|
|
for line in stdout.lines() {
|
|
tracing::info!(target: "machinectl", "{line}");
|
|
}
|
|
for line in stderr.lines() {
|
|
tracing::warn!(target: "machinectl", "{line}");
|
|
}
|
|
if !out.status.success() {
|
|
bail!(
|
|
"machinectl {} failed ({}): {}",
|
|
args.join(" "),
|
|
out.status,
|
|
stderr.trim()
|
|
);
|
|
}
|
|
Ok((stdout, stderr))
|
|
}
|
|
|
|
/// How long to wait for machined to drop a machine's registration after a
|
|
/// shutdown has been asked for. Generous: a container with slow-stopping
|
|
/// units legitimately takes a while, and escalating early would SIGKILL a
|
|
/// shutdown that was going to finish on its own.
|
|
const NAME_RELEASE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20);
|
|
|
|
/// Poll cadence while waiting for the registration to go away.
|
|
const NAME_RELEASE_POLL: std::time::Duration = std::time::Duration::from_millis(500);
|
|
|
|
/// Cap on a single registration probe. The probe is a D-Bus round trip to
|
|
/// machined; if machined itself is wedged the call would otherwise sit on
|
|
/// the D-Bus method timeout, which is far longer than the whole stop
|
|
/// sequence should take.
|
|
const MACHINE_PROBE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
|
|
|
|
/// True when machined still holds a registration for `machine`.
|
|
///
|
|
/// This asks machined the same question the registration itself answers —
|
|
/// `machinectl show` resolves the name through `GetMachine`, the very lookup
|
|
/// that makes a later boot fail with `Failed to register machine: already
|
|
/// exists`. So a positive answer here is exactly the condition that breaks
|
|
/// the next start, not a proxy for it.
|
|
///
|
|
/// Deliberately NOT the container's systemd unit state: the unit can be
|
|
/// `inactive` while the registration is still held, and that gap is the
|
|
/// whole bug this probe exists to catch.
|
|
///
|
|
/// Errors and timeouts answer "still registered". Being wrong that way costs
|
|
/// a redundant SIGKILL to something already gone; being wrong the other way
|
|
/// hands back a stop that silently leaked the name.
|
|
async fn machine_registered(machine: &str) -> bool {
|
|
let probe = Command::new("machinectl")
|
|
.args(["show", machine, "--property=Name"])
|
|
// Don't leave a probe behind when the timeout below fires.
|
|
.kill_on_drop(true)
|
|
.output();
|
|
match tokio::time::timeout(MACHINE_PROBE_TIMEOUT, probe).await {
|
|
Ok(Ok(out)) => out.status.success(),
|
|
Ok(Err(e)) => {
|
|
tracing::warn!(%machine, error = %e, "machinectl show failed to run; assuming still registered");
|
|
true
|
|
}
|
|
Err(_) => {
|
|
tracing::warn!(%machine, "machinectl show timed out; assuming still registered");
|
|
true
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Poll [`machine_registered`] until the name is free or the timeout expires.
|
|
/// Returns true once the registration is gone.
|
|
async fn wait_name_released(machine: &str) -> bool {
|
|
let deadline = tokio::time::Instant::now() + NAME_RELEASE_TIMEOUT;
|
|
loop {
|
|
if !machine_registered(machine).await {
|
|
return true;
|
|
}
|
|
if tokio::time::Instant::now() >= deadline {
|
|
return false;
|
|
}
|
|
tokio::time::sleep(NAME_RELEASE_POLL).await;
|
|
}
|
|
}
|
|
|
|
/// Stop `machine` and only report success once machined has actually released
|
|
/// the name.
|
|
///
|
|
/// `nixos-container stop` exiting 0 does not mean the machine is gone. A
|
|
/// process that sits in the machine's cgroup without being a child of the
|
|
/// container's init — a shell exec'd in from outside, say — never receives
|
|
/// the shutdown's SIGTERM if it has been stopped with SIGSTOP, so the
|
|
/// registration outlives the "successful" stop. Every later start of that
|
|
/// container then fails with `Failed to register machine: already exists`,
|
|
/// and machined re-persists the stale record across its own restart, so
|
|
/// there is no cleaning it up after the fact. The only place to catch it is
|
|
/// here, in the stop.
|
|
///
|
|
/// So: ask for the stop, wait for the name, and fail loudly if it is still
|
|
/// held — a caller that is told the stop worked will go on to start the
|
|
/// container and hit the confusing registration error instead of this one.
|
|
///
|
|
/// This used to escalate to `machinectl kill --signal=SIGKILL` here. Dropped
|
|
/// per real incident data: the case this guards is a container genuinely
|
|
/// wedged (e.g. a root-login process survived the stop), and SIGKILL
|
|
/// doesn't recover that in practice — only a host-level reboot has.
|
|
/// Pretending a kill attempt handled it hides a condition that needs a human
|
|
/// to look at the host, so this now just reports the failure instead of
|
|
/// quietly (and ineffectually) trying to force it.
|
|
///
|
|
/// The verify-then-fail lives in the helper rather than at a call site so
|
|
/// that every stop gets it: dashboard, reconcile, destroy, cold-start
|
|
/// fallback. The start path already distrusts its own exit code the same way;
|
|
/// this is the missing half of that pair.
|
|
async fn stop_and_release(machine: &str) -> Result<(String, String)> {
|
|
let stop = container_run(&["stop", machine]).await;
|
|
if wait_name_released(machine).await {
|
|
// Happy path, and also the path where a stop that reported failure
|
|
// nonetheless brought the machine down. Either way the caller gets
|
|
// the original result untouched.
|
|
return stop;
|
|
}
|
|
|
|
tracing::error!(
|
|
%machine,
|
|
"stop finished but machined still holds the registration — the \
|
|
container is likely wedged (a process outside its init tree \
|
|
survived the stop); this needs a host-level look, not another \
|
|
stop attempt"
|
|
);
|
|
bail!(
|
|
"stop {machine}: machined still holds the machine name after {}s. \
|
|
This container is likely wedged and starting it again will fail to \
|
|
register — needs host-level intervention (a reboot has been the \
|
|
only reliable fix in practice).",
|
|
NAME_RELEASE_TIMEOUT.as_secs()
|
|
)
|
|
}
|
|
|
|
/// Invoke `nixos-container` with the given args and forward output lines
|
|
/// to the caller as `PrivEvent::Line` messages in real time, logging each
|
|
/// line to journald as it arrives. Returns `(String::new(), String::new())`
|
|
/// on success (all output was streamed); the error string includes stderr
|
|
/// tail on failure.
|
|
async fn container_run_streaming(
|
|
args: &[&str],
|
|
writer: &mut OwnedWriteHalf,
|
|
) -> Result<(String, String)> {
|
|
use tokio::io::AsyncBufReadExt as _;
|
|
use tokio::process::Command;
|
|
|
|
let mut child = Command::new("nixos-container")
|
|
.args(args)
|
|
.stdout(std::process::Stdio::piped())
|
|
.stderr(std::process::Stdio::piped())
|
|
.spawn()
|
|
.context("invoke nixos-container (streaming)")?;
|
|
|
|
let stdout = child.stdout.take().expect("stdout piped");
|
|
let stderr = child.stderr.take().expect("stderr piped");
|
|
|
|
let mut stdout_lines = BufReader::new(stdout).lines();
|
|
let mut stderr_lines = BufReader::new(stderr).lines();
|
|
|
|
// Collect stderr for the error message; stream both to the client.
|
|
let mut stderr_buf = String::new();
|
|
|
|
// Drive stdout and stderr concurrently. `tokio::select!` interleaves
|
|
// them without bias — both streams drain at roughly the same rate
|
|
// as the subprocess produces output.
|
|
loop {
|
|
tokio::select! {
|
|
line = stdout_lines.next_line() => {
|
|
match line {
|
|
Ok(Some(l)) => {
|
|
tracing::info!(target: "nixos-container", "{l}");
|
|
write_line_event(writer, PrivStream::Stdout, &l).await;
|
|
}
|
|
Ok(None) => break,
|
|
Err(e) => {
|
|
tracing::warn!(error = %e, "nixos-container stdout read error");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
line = stderr_lines.next_line() => {
|
|
match line {
|
|
Ok(Some(l)) => {
|
|
tracing::warn!(target: "nixos-container", "{l}");
|
|
write_line_event(writer, PrivStream::Stderr, &l).await;
|
|
if !stderr_buf.is_empty() {
|
|
stderr_buf.push('\n');
|
|
}
|
|
stderr_buf.push_str(&l);
|
|
}
|
|
Ok(None) => {}
|
|
Err(e) => {
|
|
tracing::warn!(error = %e, "nixos-container stderr read error");
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Drain any remaining stderr after stdout closed.
|
|
while let Ok(Some(l)) = stderr_lines.next_line().await {
|
|
tracing::warn!(target: "nixos-container", "{l}");
|
|
write_line_event(writer, PrivStream::Stderr, &l).await;
|
|
if !stderr_buf.is_empty() {
|
|
stderr_buf.push('\n');
|
|
}
|
|
stderr_buf.push_str(&l);
|
|
}
|
|
|
|
let status = child.wait().await.context("wait nixos-container")?;
|
|
if !status.success() {
|
|
// Only the last stderr line is embedded — the full stderr was
|
|
// already forwarded line-by-line as PrivEvent::Line messages and
|
|
// is captured in build_logs.sqlite by the caller. Keeping the
|
|
// error message short avoids bloating the anyhow chain.
|
|
bail!(
|
|
"nixos-container {} failed ({}): {}",
|
|
args.join(" "),
|
|
status,
|
|
stderr_buf.lines().last().unwrap_or("").trim()
|
|
);
|
|
}
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
/// Read a container's journal as root via `journalctl -M`. Returns
|
|
/// `(stdout, stderr)`. Unlike `container_run` a non-zero exit is *not* a
|
|
/// hard error — journalctl's own diagnostic (folded into `stderr` with
|
|
/// the exit status) is what the caller surfaces to the operator, so the
|
|
/// helper never bails.
|
|
async fn read_container_journal(container: &str, query: &JournalQuery) -> Result<(String, String)> {
|
|
let mut args: Vec<String> = vec![
|
|
"-M".to_owned(),
|
|
container.to_owned(),
|
|
"--no-pager".to_owned(),
|
|
format!("--output={}", query.output.as_journalctl()),
|
|
"-n".to_owned(),
|
|
query.lines.to_string(),
|
|
];
|
|
if query.boot {
|
|
args.push("-b".to_owned());
|
|
}
|
|
if let Some(u) = &query.unit {
|
|
args.push("-u".to_owned());
|
|
args.push(u.clone());
|
|
}
|
|
if let Some(p) = &query.priority {
|
|
args.push("-p".to_owned());
|
|
args.push(p.clone());
|
|
}
|
|
// `--grep=`/`--since=`/`--until=` use the `=`-joined form so a value
|
|
// can never be parsed as a separate journalctl flag.
|
|
if let Some(g) = &query.grep {
|
|
args.push(format!("--grep={g}"));
|
|
}
|
|
if let Some(s) = &query.since {
|
|
args.push(format!("--since={s}"));
|
|
}
|
|
if let Some(u) = &query.until {
|
|
args.push(format!("--until={u}"));
|
|
}
|
|
let out = Command::new("journalctl")
|
|
.args(&args)
|
|
.output()
|
|
.await
|
|
.context("invoke journalctl -M")?;
|
|
let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
|
|
let stderr = if out.status.success() {
|
|
String::from_utf8_lossy(&out.stderr).into_owned()
|
|
} else {
|
|
format!(
|
|
"journalctl -M {container} exited {}: {}",
|
|
out.status,
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
)
|
|
};
|
|
Ok((stdout, stderr))
|
|
}
|
|
|
|
/// Synchronise the host's nginx unit after an `agents.conf` write.
|
|
///
|
|
/// Queries `ActiveState` and dispatches:
|
|
/// - `active` → `systemctl reload nginx` (SIGHUP, zero-downtime)
|
|
/// - `failed` → `systemctl reset-failed nginx` + `systemctl start nginx`
|
|
/// - otherwise → `systemctl start nginx`
|
|
///
|
|
/// ⚠️ `nginx` is hard-coded on purpose — see `PrivRequest::ReloadGatewayNginx`.
|
|
/// The unit name is the scope of this verb: nginx is a host unit, so no
|
|
/// namespace bounds it and the literal is the only thing standing between
|
|
/// "reload the gateway" and "reload anything".
|
|
///
|
|
/// Returns `(String::new(), String::new())` on success so it fits the
|
|
/// `exec` return type directly.
|
|
async fn sync_gateway_nginx() -> Result<(String, String)> {
|
|
let state_out = Command::new("systemctl")
|
|
.args(["show", "--property=ActiveState", "--value", "nginx"])
|
|
.output()
|
|
.await
|
|
.context("query gateway nginx ActiveState")?;
|
|
if !state_out.status.success() {
|
|
tracing::warn!(
|
|
exit_code = ?state_out.status.code(),
|
|
stderr = %String::from_utf8_lossy(&state_out.stderr).trim(),
|
|
"systemctl show ActiveState exited non-zero — gateway nginx may be down"
|
|
);
|
|
}
|
|
let state = String::from_utf8_lossy(&state_out.stdout).trim().to_owned();
|
|
// State-aware dispatch: reload when running; reset+start after
|
|
// start-limit failure; plain start when inactive or unknown.
|
|
match state.as_str() {
|
|
"active" => {
|
|
let out = Command::new("systemctl")
|
|
.args(["reload", "nginx"])
|
|
.output()
|
|
.await
|
|
.context("reload gateway nginx")?;
|
|
if !out.status.success() {
|
|
bail!(
|
|
"gateway nginx reload failed ({}): {}",
|
|
out.status,
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
}
|
|
"failed" => {
|
|
// Clear start-limit so the next start can proceed.
|
|
let _ = Command::new("systemctl")
|
|
.args(["reset-failed", "nginx"])
|
|
.status()
|
|
.await;
|
|
let out = Command::new("systemctl")
|
|
.args(["start", "nginx"])
|
|
.output()
|
|
.await
|
|
.context("start gateway nginx after reset-failed")?;
|
|
if !out.status.success() {
|
|
bail!(
|
|
"gateway nginx start (after reset-failed) failed ({}): {}",
|
|
out.status,
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
}
|
|
_ => {
|
|
// inactive, activating, deactivating, unknown — just start.
|
|
let out = Command::new("systemctl")
|
|
.args(["start", "nginx"])
|
|
.output()
|
|
.await
|
|
.context("start gateway nginx")?;
|
|
if !out.status.success() {
|
|
bail!(
|
|
"gateway nginx start failed (state={state:?}) ({}): {}",
|
|
out.status,
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
}
|
|
}
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
/// Return the system container name for a logical agent name.
|
|
/// All agents (including the manager) use the `h-` prefix.
|
|
fn container_system_name(name: &str) -> String {
|
|
format!("{AGENT_PREFIX}{name}")
|
|
}
|
|
|
|
/// Validate a logical agent name (the name hive-c0re uses internally,
|
|
/// before the `h-` container prefix is applied).
|
|
fn validate_agent_name(name: &str) -> Result<()> {
|
|
validate_name_chars(name)?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Validate a logical agent name and check it maps to a hive-managed container.
|
|
fn validate_container_name(name: &str) -> Result<()> {
|
|
if SIBLING_CONTAINERS.contains(&name) {
|
|
return Ok(());
|
|
}
|
|
validate_name_chars(name)?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Validate a system-level container name (already has `h-` prefix for
|
|
/// all agents including the manager, or is a sibling service name).
|
|
fn validate_container_system_name(name: &str) -> Result<()> {
|
|
if SIBLING_CONTAINERS.contains(&name) {
|
|
return Ok(());
|
|
}
|
|
if let Some(suffix) = name.strip_prefix(AGENT_PREFIX) {
|
|
validate_name_chars(suffix)?;
|
|
return Ok(());
|
|
}
|
|
bail!("container name {name:?} is not managed by hive");
|
|
}
|
|
|
|
fn validate_name_chars(name: &str) -> Result<()> {
|
|
if name.is_empty()
|
|
|| !name
|
|
.chars()
|
|
.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-')
|
|
{
|
|
bail!("invalid name {name:?}: must be non-empty lowercase ascii + digits + hyphens");
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Derive the meta-flake ref for an agent by name.
|
|
fn agent_flake_ref(name: &str) -> String {
|
|
format!("{META_DIR}#{name}")
|
|
}
|
|
|
|
/// Validate a bind-mount path: must be absolute, non-empty, and contain
|
|
/// no newlines, null bytes, or double-quotes (which would break the
|
|
/// `EXTRA_NSPAWN_FLAGS="..."` conf line format).
|
|
fn validate_bind_path(path: &str) -> Result<()> {
|
|
if path.is_empty()
|
|
|| !path.starts_with('/')
|
|
|| path
|
|
.bytes()
|
|
.any(|b| b == 0 || b == b'\n' || b == b'"' || b == b':')
|
|
{
|
|
bail!(
|
|
"invalid bind path {path:?}: must be an absolute path with no colons, newlines, null bytes, or double-quotes"
|
|
);
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Update `/etc/nixos-containers/<container>.conf`: strips network-isolation
|
|
/// vars (`PRIVATE_NETWORK`, `HOST_ADDRESS*`, `LOCAL_ADDRESS*`, `HOST_BRIDGE`,
|
|
/// Update `/etc/nixos-containers/<container>.conf`: strip old network vars,
|
|
/// write network isolation settings, then append `EXTRA_NSPAWN_FLAGS`.
|
|
/// When `isolation` is `Some`, writes `PRIVATE_NETWORK=1` + veth wiring;
|
|
/// when `None`, writes `PRIVATE_NETWORK=0`.
|
|
/// `--tmpfs=<mount>/.git` for every bound git repo, hiding its metadata
|
|
/// from inside the container.
|
|
///
|
|
/// Two kinds of mount qualify, for one reason: **the agent is given a
|
|
/// working tree, never a repository.** `/knowledge` is the hive's shared
|
|
/// docs — whose `.git/config` has held a credential the host-side worker
|
|
/// embedded — and `/agents/<name>/config` is a config repo, an agent's own
|
|
/// or a parent's read-only view of a child's. In both cases `.git` carries
|
|
/// every branch and the full history of a document whose *currently
|
|
/// deployed* value is the only thing a reader may act on, and an abandoned
|
|
/// branch is indistinguishable from a live one.
|
|
///
|
|
/// An overlay rather than an exported copy: there is no second tree to
|
|
/// keep in sync, so nothing can go stale, and no code path has to remember
|
|
/// to refresh it.
|
|
///
|
|
/// ⚠️ Ordering matters — these must be appended **after** the `--bind`
|
|
/// flags so nspawn mounts them on top of the already-mounted trees.
|
|
/// Config mounts are matched by shape, not by a name list: the set is
|
|
/// dynamic, growing with each child bound into a parent.
|
|
fn git_overlay_flags(binds: &[BindMount]) -> Vec<String> {
|
|
binds
|
|
.iter()
|
|
.map(|b| b.container_path.as_str())
|
|
.filter(|p| *p == "/knowledge" || (p.starts_with("/agents/") && p.ends_with("/config")))
|
|
.map(|p| format!("--tmpfs={p}/.git"))
|
|
.collect()
|
|
}
|
|
|
|
fn write_nspawn_flags(
|
|
container: &str,
|
|
binds: &[BindMount],
|
|
isolation: Option<&NetworkIsolation>,
|
|
load_credentials: &[CredentialMount],
|
|
) -> Result<()> {
|
|
use std::fmt::Write as _;
|
|
let path = format!("/etc/nixos-containers/{container}.conf");
|
|
let original = std::fs::read_to_string(&path).with_context(|| format!("read {path}"))?;
|
|
let lines: Vec<&str> = original
|
|
.lines()
|
|
.filter(|line| {
|
|
let t = line.trim_start();
|
|
!t.starts_with("EXTRA_NSPAWN_FLAGS=")
|
|
&& !t.starts_with("PRIVATE_NETWORK=")
|
|
&& !t.starts_with("HOST_ADDRESS=")
|
|
&& !t.starts_with("LOCAL_ADDRESS=")
|
|
&& !t.starts_with("HOST_ADDRESS6=")
|
|
&& !t.starts_with("LOCAL_ADDRESS6=")
|
|
&& !t.starts_with("HOST_BRIDGE=")
|
|
})
|
|
.collect();
|
|
let mut out = lines.join("\n");
|
|
if !out.is_empty() {
|
|
out.push('\n');
|
|
}
|
|
if let Some(iso) = isolation {
|
|
out.push_str("PRIVATE_NETWORK=1\n");
|
|
// HOST_ADDRESS = the bridge gateway IP. nixos-container's
|
|
// container-side setup only installs a default route
|
|
// (`ip route add default via $HOST_ADDRESS`) when HOST_ADDRESS is
|
|
// non-empty; leaving it blank gave the container an address but no
|
|
// route off the bridge subnet (no internet, no api.anthropic.com).
|
|
// In bridge mode (HOST_BRIDGE set) the host-side address/route
|
|
// setup is skipped, so this only affects the container's route —
|
|
// exactly what we want.
|
|
let _ = writeln!(out, "HOST_ADDRESS={}", iso.gateway_ip);
|
|
// LOCAL_ADDRESS is intentionally empty: agent containers receive their
|
|
// IP dynamically via DHCP from the bridge dnsmasq pool. HOST_ADDRESS
|
|
// (the gateway IP) is still written so nixos-container's container-side
|
|
// init installs a default route before the DHCP lease arrives.
|
|
out.push_str("LOCAL_ADDRESS=\n");
|
|
out.push_str("HOST_ADDRESS6=\n");
|
|
out.push_str("LOCAL_ADDRESS6=\n");
|
|
let _ = writeln!(out, "HOST_BRIDGE={}", iso.bridge);
|
|
} else {
|
|
out.push_str("PRIVATE_NETWORK=0\n");
|
|
out.push_str("HOST_ADDRESS=\n");
|
|
out.push_str("LOCAL_ADDRESS=\n");
|
|
out.push_str("HOST_ADDRESS6=\n");
|
|
out.push_str("LOCAL_ADDRESS6=\n");
|
|
out.push_str("HOST_BRIDGE=\n");
|
|
}
|
|
let mut flags: Vec<String> = binds
|
|
.iter()
|
|
.map(|b| {
|
|
let flag = if b.read_only { "--bind-ro" } else { "--bind" };
|
|
format!("{flag}={}:{}", b.host_path, b.container_path)
|
|
})
|
|
.collect();
|
|
flags.extend(git_overlay_flags(binds));
|
|
// Credential forwarding: nspawn loads each host secret into the
|
|
// container's credential store under `<name>`; inner units inherit it
|
|
// via `LoadCredential=<name>`. Validated (name charset + bind-path
|
|
// rules) in handle_write_nspawn_flags above.
|
|
for cred in load_credentials {
|
|
flags.push(format!(
|
|
"--load-credential={}:{}",
|
|
cred.name, cred.host_path
|
|
));
|
|
}
|
|
let flags_joined = flags.join(" ");
|
|
let _ = writeln!(out, "EXTRA_NSPAWN_FLAGS=\"{flags_joined}\"");
|
|
std::fs::write(&path, out).with_context(|| format!("write {path}"))?;
|
|
|
|
// DNS marker for the in-container resolver oneshot. nixos-container
|
|
// copies the *host's* /etc/resolv.conf into the container at every
|
|
// start (its host resolver — e.g. 127.0.0.53 — is unreachable from a
|
|
// private netns, and isn't authoritative for the hive's own zones
|
|
// anyway). The `hyperhive-isolated-dns` oneshot in harness-base.nix
|
|
// rewrites resolv.conf to point at the bridge resolver, but only when
|
|
// this marker exists; it carries the gateway IP so the container
|
|
// doesn't have to re-derive it. Written on isolate, removed otherwise,
|
|
// so the same shared container toplevel behaves correctly in both modes.
|
|
write_bridge_dns_marker(container, isolation)?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Path to the in-container DNS marker (the container's own `/etc`).
|
|
fn bridge_dns_marker_path(container: &str) -> String {
|
|
format!("/var/lib/nixos-containers/{container}/etc/hyperhive-bridge-dns")
|
|
}
|
|
|
|
/// Write (isolated) or remove (host-netns) the bridge-DNS marker the
|
|
/// `hyperhive-isolated-dns` oneshot keys off. The marker file contains
|
|
/// just the gateway IP. Best-effort on removal (absence is the goal).
|
|
fn write_bridge_dns_marker(container: &str, isolation: Option<&NetworkIsolation>) -> Result<()> {
|
|
let path = bridge_dns_marker_path(container);
|
|
match isolation {
|
|
Some(iso) => {
|
|
// On a fresh install the container's `/etc` may not exist yet
|
|
// (rootfs not fully materialised before the first start), so
|
|
// `write` would fail with ENOENT. Create the parent dir first
|
|
// — it's the container's own `/etc`, which nixos-container
|
|
// populates on start; a pre-created dir + our marker persist.
|
|
if let Some(parent) = std::path::Path::new(&path).parent() {
|
|
std::fs::create_dir_all(parent).with_context(|| {
|
|
format!("create bridge-DNS marker dir {}", parent.display())
|
|
})?;
|
|
}
|
|
std::fs::write(&path, format!("{}\n", iso.gateway_ip))
|
|
.with_context(|| format!("write bridge-DNS marker {path}"))?;
|
|
}
|
|
None => match std::fs::remove_file(&path) {
|
|
Ok(()) => {}
|
|
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
|
|
Err(e) => return Err(e).with_context(|| format!("remove bridge-DNS marker {path}")),
|
|
},
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// `SyncAgentTmpfiles` — write `/etc/tmpfiles.d/hyperhive-agents.conf` for
|
|
/// the given agent set and immediately apply it with `systemd-tmpfiles --create`.
|
|
///
|
|
/// Each call atomically replaces the file with entries for all current agents,
|
|
/// then creates any missing dirs on the running host. The file survives reboots
|
|
/// and is read by `systemd-tmpfiles-setup.service` (runs in `sysinit.target`,
|
|
/// before any container units can start), so bind-mount source dirs are always
|
|
/// pre-created regardless of whether hive-c0re has reached `ensure_agent_runtime_dir`.
|
|
///
|
|
/// Directories written per agent:
|
|
/// - `/run/hyperhive/agents/<name>` (MCP socket dir, bind-mounted into container
|
|
/// as `/run/hive`)
|
|
/// - `/run/hive-agent/<name>` (web socket dir, bind-mounted into container)
|
|
const TMPFILES_PATH: &str = "/etc/tmpfiles.d/hyperhive-agents.conf";
|
|
|
|
async fn sync_agent_tmpfiles(agents: &[AgentTmpfilesEntry]) -> Result<(String, String)> {
|
|
use std::fmt::Write as _;
|
|
for entry in agents {
|
|
validate_agent_name(&entry.name)?;
|
|
}
|
|
|
|
// Build tmpfiles.d content. Root dirs first, then per-agent.
|
|
let mut content =
|
|
String::from("# managed by hive-c0re — do not edit (regenerated on spawn/destroy)\n");
|
|
// Parent dirs — created with permissive mode so hive-c0re can make subdirs.
|
|
// /run/hyperhive itself is also a RuntimeDirectory of hive-c0re.service; the
|
|
// tmpfiles.d entry here ensures it exists before hive-c0re starts (boot race).
|
|
content.push_str("d /run/hyperhive 0750 hive-core hive-core -\n");
|
|
writeln!(content, "d {AGENT_RUNTIME_ROOT} 0755 hive-core hive-core -").ok();
|
|
// `hive-core`, not root: c0re does the `create_dir_all` for a new agent's
|
|
// subdir itself, so a root-owned parent EACCESes on the first spawn of a
|
|
// fresh host. This must stay in step with the identical rule in
|
|
// `nix/host-modules/hive-gateway/default.nix` — the two files declared
|
|
// different owners for this one path, and which won depended on the order
|
|
// systemd happened to read them in.
|
|
writeln!(content, "d {SOCKET_DIR_ROOT} 0755 hive-core hive-core -").ok();
|
|
// Per-agent dirs.
|
|
for entry in agents {
|
|
let name = &entry.name;
|
|
writeln!(
|
|
content,
|
|
"d {AGENT_RUNTIME_ROOT}/{name} 0755 hive-core hive-core -"
|
|
)
|
|
.ok();
|
|
// The agent's socket dir. Three principals need it and no two share a
|
|
// group, so the mode has to say so explicitly:
|
|
//
|
|
// owner = the agent user rwx binds + unlinks agent.sock/web.sock
|
|
// other = --x traverse only, no listing
|
|
//
|
|
// "other" covers hive-c0re (dials agent.sock) and the gateway's nginx
|
|
// (dials web.sock, and has all of /run/hive-agent bind-mounted in).
|
|
// Both sockets are 0666, so traversal is all they need.
|
|
//
|
|
// 0751 rather than the historical 0777 is a fix, not a tidy-up:
|
|
// write permission on a *directory* is what confers the right to
|
|
// unlink its entries, whoever owns them — the sticky bit is the only
|
|
// thing that would restrain that, and it was never set here. So the
|
|
// old world-writable mode let anything able to reach the path delete
|
|
// an agent's socket, bind its own, and receive that agent's todos.
|
|
// Dropping `o=w` removes that permission outright rather than
|
|
// qualifying it. Declaring the owner here also ends the tug-of-war
|
|
// with the
|
|
// old ChownSocketDir: `d` re-applies on every sync, so a chown made
|
|
// afterwards was reset by the next agent's spawn.
|
|
if let (Some(uid), Some(gid)) = (entry.uid, entry.gid) {
|
|
writeln!(content, "d {SOCKET_DIR_ROOT}/{name} 0751 {uid} {gid} -").ok();
|
|
} else {
|
|
// Before the container's /etc/passwd exists there is no uid to
|
|
// name, and the harness must still be able to bind. Keep the old
|
|
// permissive mode for that agent alone; the next sync (any spawn
|
|
// or destroy, or c0re restart) resolves the uid and tightens it.
|
|
tracing::info!(%name, "tmpfiles.d: agent uid unknown, deferring 0751 on socket dir");
|
|
writeln!(content, "d {SOCKET_DIR_ROOT}/{name} 0777 root root -").ok();
|
|
}
|
|
}
|
|
|
|
// Atomic write: write to a tmp file then rename so a concurrent reader
|
|
// always sees a complete file.
|
|
let tmp = format!("{TMPFILES_PATH}.tmp");
|
|
std::fs::write(&tmp, &content).with_context(|| format!("write {tmp}"))?;
|
|
std::fs::rename(&tmp, TMPFILES_PATH)
|
|
.with_context(|| format!("rename {TMPFILES_PATH}.tmp -> {TMPFILES_PATH}"))?;
|
|
tracing::info!(agents = agents.len(), "tmpfiles.d: wrote {TMPFILES_PATH}");
|
|
|
|
// Apply immediately so dirs exist on the running host, not just after next boot.
|
|
let out = Command::new("systemd-tmpfiles")
|
|
.args(["--create", TMPFILES_PATH])
|
|
.output()
|
|
.await
|
|
.context("systemd-tmpfiles --create")?;
|
|
if !out.status.success() {
|
|
let stderr = String::from_utf8_lossy(&out.stderr).trim().to_owned();
|
|
anyhow::bail!(
|
|
"systemd-tmpfiles --create failed ({}): {stderr}",
|
|
out.status
|
|
);
|
|
}
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::{
|
|
BindMount, OwnedFd, PAUSED_MARKER_FILE, PrivRequest, check_fd_agreement,
|
|
contains_secret_shaped_run, git_overlay_flags, limits_dropin_body, redact_secret_line,
|
|
remove_marker_in, write_state_file_nofollow,
|
|
};
|
|
use std::path::PathBuf;
|
|
use std::sync::atomic::{AtomicU32, Ordering};
|
|
|
|
fn bind(container_path: &str) -> BindMount {
|
|
BindMount {
|
|
host_path: "/var/lib/hyperhive/whatever".to_owned(),
|
|
container_path: container_path.to_owned(),
|
|
read_only: true,
|
|
}
|
|
}
|
|
|
|
/// Every bound git repo gets its `.git` overlaid — the knowledge tree
|
|
/// and *each* config mount, an agent's own plus every child's.
|
|
///
|
|
/// The child case is the one worth pinning: that set grows at runtime
|
|
/// as agents gain children, so a rule written as a list of names would
|
|
/// silently stop covering new ones.
|
|
#[test]
|
|
fn every_bound_git_repo_gets_its_dot_git_hidden() {
|
|
let flags = git_overlay_flags(&[
|
|
bind("/knowledge"),
|
|
bind("/agents/atlas/config"),
|
|
bind("/agents/kiddo/config"),
|
|
]);
|
|
assert_eq!(
|
|
flags,
|
|
[
|
|
"--tmpfs=/knowledge/.git",
|
|
"--tmpfs=/agents/atlas/config/.git",
|
|
"--tmpfs=/agents/kiddo/config/.git",
|
|
]
|
|
);
|
|
}
|
|
|
|
/// ...and nothing else does. A blanket "overlay .git on every bind"
|
|
/// would mask a real `.git` under `state/`, where an agent legitimately
|
|
/// keeps working clones of its own.
|
|
#[test]
|
|
fn non_repo_mounts_are_left_alone() {
|
|
let flags = git_overlay_flags(&[
|
|
bind("/agents/atlas/state"),
|
|
bind("/shared"),
|
|
bind("/applied"),
|
|
bind("/agents/atlas/config-notes"),
|
|
]);
|
|
assert!(flags.is_empty(), "overlaid a non-repo mount: {flags:?}");
|
|
}
|
|
|
|
/// 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");
|
|
}
|
|
|
|
/// The whole rendered body, pinned literally — including the values,
|
|
/// so changing the restart policy is a visible test edit rather than a
|
|
/// silent one.
|
|
///
|
|
/// Placement is the part most worth pinning: `StartLimit*` are `[Unit]`
|
|
/// settings and systemd **silently ignores** them under `[Service]`, so
|
|
/// a bound that moved sections would look configured and do nothing.
|
|
/// An unset weight still emits no line at all, which is what keeps a
|
|
/// hive-c0re older than that field from changing what lands on disk.
|
|
#[test]
|
|
fn dropin_body_is_pinned_exactly() {
|
|
assert_eq!(
|
|
limits_dropin_body("/run/hyperhive/agents/iris", "4G", "200%", None, None),
|
|
"[Unit]\n\
|
|
ConditionPathIsDirectory=/run/hyperhive/agents/iris\n\
|
|
StartLimitIntervalSec=600\n\
|
|
StartLimitBurst=3\n\
|
|
\n\
|
|
[Service]\n\
|
|
RestartSec=5\n\
|
|
MemoryMax=4G\n\
|
|
CPUQuota=200%\n"
|
|
);
|
|
}
|
|
|
|
/// The bound is hive-wide **policy**, not a per-agent parameter: it is
|
|
/// rendered from constants and no caller-supplied value can omit or
|
|
/// alter it. This is the property that justifies keeping it out of the
|
|
/// wire protocol — if it ever varies by request, that argument is gone.
|
|
#[test]
|
|
fn the_start_limit_is_present_whatever_the_caller_passes() {
|
|
for (mem, cpu, cw, iw) in [
|
|
("4G", "200%", None, None),
|
|
("512M", "50%", Some(10), Some(10)),
|
|
("infinity", "infinity", Some(10_000), None),
|
|
] {
|
|
let body = limits_dropin_body("/rt/x", mem, cpu, cw, iw);
|
|
let unit = body
|
|
.split("[Service]")
|
|
.next()
|
|
.expect("the body always has a [Unit] section before [Service]");
|
|
assert!(
|
|
unit.contains("StartLimitIntervalSec=600") && unit.contains("StartLimitBurst=3"),
|
|
"start limit missing from [Unit] for ({mem}, {cpu}): {body}"
|
|
);
|
|
assert!(
|
|
body.contains("\nRestartSec=5\n"),
|
|
"restart backoff missing for ({mem}, {cpu}): {body}"
|
|
);
|
|
}
|
|
}
|
|
|
|
/// Weights are appended to the `[Service]` section, each omitted
|
|
/// independently when `None`.
|
|
#[test]
|
|
fn weights_are_emitted_only_when_set() {
|
|
let both = limits_dropin_body("/rt/x", "4G", "200%", Some(80), Some(80));
|
|
assert!(
|
|
both.ends_with("CPUQuota=200%\nCPUWeight=80\nIOWeight=80\n"),
|
|
"{both}"
|
|
);
|
|
|
|
let cpu_only = limits_dropin_body("/rt/x", "4G", "200%", Some(80), None);
|
|
assert!(
|
|
cpu_only.ends_with("CPUQuota=200%\nCPUWeight=80\n"),
|
|
"{cpu_only}"
|
|
);
|
|
assert!(!cpu_only.contains("IOWeight"), "{cpu_only}");
|
|
|
|
let io_only = limits_dropin_body("/rt/x", "4G", "200%", None, Some(80));
|
|
assert!(
|
|
io_only.ends_with("CPUQuota=200%\nIOWeight=80\n"),
|
|
"{io_only}"
|
|
);
|
|
assert!(!io_only.contains("CPUWeight"), "{io_only}");
|
|
}
|
|
|
|
#[test]
|
|
fn redacts_lines_mentioning_password_case_insensitively() {
|
|
assert_eq!(
|
|
redact_secret_line("New password: hunter2"),
|
|
"[redacted: line mentions a password]"
|
|
);
|
|
assert_eq!(
|
|
redact_secret_line("PASSWORD=hunter2"),
|
|
"[redacted: line mentions a password]"
|
|
);
|
|
assert_eq!(
|
|
redact_secret_line("User \"foo\" was successfully created."),
|
|
"User \"foo\" was successfully created."
|
|
);
|
|
}
|
|
|
|
/// The regression this function exists for. The keyword rule passes
|
|
/// this line straight through — it says nothing about a password — so
|
|
/// only the shape rule catches it.
|
|
#[test]
|
|
fn redacts_access_token_line_which_mentions_no_password() {
|
|
let line =
|
|
"Access token was successfully created: 0123456789abcdef0123456789abcdef01234567";
|
|
assert!(
|
|
!line.to_ascii_lowercase().contains("password"),
|
|
"fixture must not contain the keyword, or it proves nothing"
|
|
);
|
|
assert_eq!(
|
|
redact_secret_line(line),
|
|
"[redacted: line contains a secret-shaped token]"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn secret_shape_boundaries() {
|
|
// Ordinary forgejo-admin output survives: no run is long enough.
|
|
assert_eq!(
|
|
redact_secret_line("Command 'user' 'create' finished with no errors."),
|
|
"Command 'user' 'create' finished with no errors."
|
|
);
|
|
// 31 chars is below the floor, 32 is at it.
|
|
assert!(!contains_secret_shaped_run(&"a".repeat(31)));
|
|
assert!(contains_secret_shaped_run(&"a".repeat(32)));
|
|
// Punctuation breaks the run — a sentence never trips it however long.
|
|
assert!(!contains_secret_shaped_run(
|
|
"this.is.a.very.long.dotted.identifier.but.not.a.secret"
|
|
));
|
|
// base64url and hex alphabets both count.
|
|
assert!(contains_secret_shaped_run(
|
|
"ZGVhZGJlZWZkZWFkYmVlZmRlYWRiZWVmZGVhZA=="
|
|
));
|
|
assert!(contains_secret_shaped_run(
|
|
"aG93-dy_there-aG93dy1theresomething"
|
|
));
|
|
}
|
|
|
|
/// argus on the review: a whitespace-delimited check is defeated by
|
|
/// punctuation glued to the secret — the punctuation joins the "word"
|
|
/// and fails the alphabet test for the whole run, while the credential
|
|
/// sits there in full. Scanning for a RUN rather than a WORD closes it.
|
|
/// These are the shapes that used to slip through.
|
|
#[test]
|
|
fn secret_is_caught_with_punctuation_glued_to_it() {
|
|
const TOK: &str = "0123456789abcdef0123456789abcdef01234567";
|
|
for line in [
|
|
format!("token: {TOK},"),
|
|
format!("token: {TOK}."),
|
|
format!("using [{TOK}] now"),
|
|
format!("value=\"{TOK}\""),
|
|
format!("(created {TOK})"),
|
|
// no whitespace anywhere -- one glued blob
|
|
format!("Bearer:{TOK};next"),
|
|
] {
|
|
assert_eq!(
|
|
redact_secret_line(&line),
|
|
"[redacted: line contains a secret-shaped token]",
|
|
"leaked through: {line}"
|
|
);
|
|
}
|
|
}
|
|
|
|
/// Unique scratch dir per test, no external tempfile dep.
|
|
fn scratch() -> PathBuf {
|
|
static CTR: AtomicU32 = AtomicU32::new(0);
|
|
let n = CTR.fetch_add(1, Ordering::Relaxed);
|
|
let dir = std::env::temp_dir().join(format!(
|
|
"hive-priv-nofollow-test-{}-{n}",
|
|
std::process::id()
|
|
));
|
|
std::fs::create_dir_all(&dir).unwrap();
|
|
dir
|
|
}
|
|
|
|
#[test]
|
|
fn rejects_non_plain_filenames() {
|
|
let dir = scratch();
|
|
for bad in ["", ".", "..", "a/b", "/etc/passwd", "../escape", "sub/tok"] {
|
|
assert!(
|
|
write_state_file_nofollow(&dir, bad, "x").is_err(),
|
|
"must reject filename {bad:?}"
|
|
);
|
|
}
|
|
std::fs::remove_dir_all(&dir).ok();
|
|
}
|
|
|
|
#[test]
|
|
fn refuses_symlink_leaf_and_leaves_target_untouched() {
|
|
let dir = scratch();
|
|
let target = dir.join("target");
|
|
std::fs::write(&target, "original").unwrap();
|
|
// Agent plants a symlink where the token would be written.
|
|
std::os::unix::fs::symlink(&target, dir.join("forge-token")).unwrap();
|
|
|
|
let res = write_state_file_nofollow(&dir, "forge-token", "PWNED");
|
|
assert!(res.is_err(), "O_NOFOLLOW must refuse a symlink leaf");
|
|
// The root-privileged write must NOT have followed the link.
|
|
assert_eq!(
|
|
std::fs::read_to_string(&target).unwrap(),
|
|
"original",
|
|
"symlink target must be untouched"
|
|
);
|
|
std::fs::remove_dir_all(&dir).ok();
|
|
}
|
|
|
|
#[test]
|
|
fn writes_plain_file_0600() {
|
|
use std::os::unix::fs::PermissionsExt as _;
|
|
let dir = scratch();
|
|
write_state_file_nofollow(&dir, "forge-token", "secret").unwrap();
|
|
let path = dir.join("forge-token");
|
|
assert_eq!(std::fs::read_to_string(&path).unwrap(), "secret");
|
|
let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
|
|
assert_eq!(mode, 0o600, "token file must be 0600");
|
|
// Overwrite truncates cleanly (O_TRUNC), no residue.
|
|
write_state_file_nofollow(&dir, "forge-token", "new").unwrap();
|
|
assert_eq!(std::fs::read_to_string(&path).unwrap(), "new");
|
|
std::fs::remove_dir_all(&dir).ok();
|
|
}
|
|
|
|
/// The pause marker round-trips through the same root-only path the
|
|
/// credential writes use, and BOTH directions are idempotent — the
|
|
/// dashboard toggle and `hivectl pause|resume` fire blind, without
|
|
/// reading the current state first.
|
|
#[test]
|
|
fn pause_marker_create_and_remove_are_idempotent() {
|
|
let dir = scratch();
|
|
let path = dir.join(PAUSED_MARKER_FILE);
|
|
|
|
for _ in 0..2 {
|
|
write_state_file_nofollow(&dir, PAUSED_MARKER_FILE, "").unwrap();
|
|
assert!(path.exists(), "marker must exist after pause");
|
|
assert_eq!(std::fs::read_to_string(&path).unwrap(), "");
|
|
}
|
|
for _ in 0..2 {
|
|
remove_marker_in(&dir, PAUSED_MARKER_FILE).unwrap();
|
|
assert!(!path.exists(), "marker must be gone after resume");
|
|
}
|
|
std::fs::remove_dir_all(&dir).ok();
|
|
}
|
|
|
|
/// A resume must never follow an agent-planted symlink at the marker
|
|
/// path: this unlink runs as root, so following it would let an agent
|
|
/// delete an arbitrary file on the host.
|
|
#[test]
|
|
fn resume_unlinks_the_symlink_not_its_target() {
|
|
let dir = scratch();
|
|
let target = dir.join("target");
|
|
std::fs::write(&target, "original").unwrap();
|
|
let link = dir.join(PAUSED_MARKER_FILE);
|
|
std::os::unix::fs::symlink(&target, &link).unwrap();
|
|
|
|
remove_marker_in(&dir, PAUSED_MARKER_FILE).unwrap();
|
|
// `exists()` follows the link, so it can't tell "link removed" from
|
|
// "target removed, dangling link left" — stat the link itself.
|
|
assert!(
|
|
std::fs::symlink_metadata(&link).is_err(),
|
|
"the link itself must be unlinked"
|
|
);
|
|
assert_eq!(
|
|
std::fs::read_to_string(&target).unwrap(),
|
|
"original",
|
|
"symlink target must survive the root unlink"
|
|
);
|
|
std::fs::remove_dir_all(&dir).ok();
|
|
}
|
|
}
|