890 lines
32 KiB
Rust
890 lines
32 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::path::{Path, PathBuf};
|
|
|
|
use anyhow::{Context as _, Result, bail};
|
|
use hive_sh4re::priv_proto::{
|
|
AGENT_PREFIX, AGENT_STATE_ROOT, BindMount, JournalOutput, MANAGER_NAME, META_DIR,
|
|
NetworkIsolation, PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse, PrivStream, PrivStreamLine,
|
|
SIBLING_CONTAINERS,
|
|
};
|
|
use tokio::io::{AsyncBufReadExt, 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> {
|
|
use std::os::unix::fs::PermissionsExt as _;
|
|
// Socket activation: systemd passes the socket as fd 3 when
|
|
// LISTEN_FDS >= 1 and LISTEN_PID matches our pid.
|
|
let listen_fds: Option<i32> = std::env::var("LISTEN_FDS")
|
|
.ok()
|
|
.and_then(|s| s.parse().ok());
|
|
let listen_pid: Option<u32> = std::env::var("LISTEN_PID")
|
|
.ok()
|
|
.and_then(|s| s.parse().ok());
|
|
|
|
if let (Some(n), Some(p)) = (listen_fds, listen_pid)
|
|
&& n >= 1
|
|
&& p == std::process::id()
|
|
{
|
|
// SAFETY: systemd has passed us a ready UnixListener on fd 3.
|
|
let std_listener = unsafe {
|
|
use std::os::unix::io::FromRawFd;
|
|
std::os::unix::net::UnixListener::from_raw_fd(3)
|
|
};
|
|
std_listener
|
|
.set_nonblocking(true)
|
|
.context("set socket non-blocking")?;
|
|
let listener =
|
|
tokio::net::UnixListener::from_std(std_listener).context("wrap systemd socket")?;
|
|
tracing::info!("using systemd-activated socket");
|
|
return Ok(listener);
|
|
}
|
|
|
|
// Fallback: bind the socket ourselves.
|
|
let path = Path::new(PRIV_SOCK);
|
|
if let Some(parent) = path.parent() {
|
|
std::fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?;
|
|
}
|
|
let _ = std::fs::remove_file(path);
|
|
let listener = UnixListener::bind(path).with_context(|| format!("bind {PRIV_SOCK}"))?;
|
|
// Mode 0660: only the hive-core group can connect.
|
|
std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o660))
|
|
.context("chmod priv.sock")?;
|
|
tracing::info!(path = PRIV_SOCK, "bound priv socket");
|
|
Ok(listener)
|
|
}
|
|
|
|
async fn handle(stream: UnixStream) {
|
|
let (reader, mut writer) = stream.into_split();
|
|
let mut lines = BufReader::new(reader).lines();
|
|
while let Ok(Some(line)) = lines.next_line().await {
|
|
let resp = dispatch(&line, &mut writer).await;
|
|
// Write the terminal PrivResponse as a PrivEvent::Done. Wire-identical
|
|
// to a bare PrivResponse (untagged), so old hive-c0re callers that
|
|
// deserialise directly to PrivResponse continue to work.
|
|
let event = PrivEvent::Done(resp);
|
|
let mut json = serde_json::to_string(&event).unwrap_or_else(|e| {
|
|
format!("{{\"ok\":false,\"stdout\":\"\",\"stderr\":\"\",\"error\":\"serialise failed: {e}\"}}")
|
|
});
|
|
json.push('\n');
|
|
if let Err(e) = writer.write_all(json.as_bytes()).await {
|
|
tracing::warn!(error = %e, "write response failed");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn dispatch(line: &str, writer: &mut OwnedWriteHalf) -> PrivResponse {
|
|
match serde_json::from_str::<PrivRequest>(line) {
|
|
Ok(req) => match exec(req, 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:#}")),
|
|
},
|
|
},
|
|
Err(e) => PrivResponse {
|
|
ok: false,
|
|
stdout: String::new(),
|
|
stderr: String::new(),
|
|
error: Some(format!("parse request: {e}")),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// 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.
|
|
#[allow(clippy::too_many_lines)]
|
|
async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String, String)> {
|
|
match req {
|
|
PrivRequest::StartContainer { ref name } => {
|
|
validate_container_name(name)?;
|
|
container_run(&["start", &container_system_name(name)]).await
|
|
}
|
|
|
|
PrivRequest::StopContainer { ref name } => {
|
|
validate_container_name(name)?;
|
|
container_run(&["stop", &container_system_name(name)]).await
|
|
}
|
|
|
|
PrivRequest::KillContainer { ref name } => {
|
|
validate_container_name(name)?;
|
|
container_run(&["kill", &container_system_name(name)]).await
|
|
}
|
|
|
|
PrivRequest::UpdateContainer { ref name, stream } => {
|
|
validate_container_name(name)?;
|
|
let flake_ref = agent_flake_ref(name);
|
|
let args = [
|
|
"update",
|
|
&container_system_name(name),
|
|
"--flake",
|
|
&flake_ref,
|
|
];
|
|
if stream {
|
|
container_run_streaming(&args, writer).await
|
|
} else {
|
|
container_run(&args).await
|
|
}
|
|
}
|
|
|
|
PrivRequest::CreateContainer { ref name, stream } => {
|
|
validate_container_name(name)?;
|
|
let flake_ref = agent_flake_ref(name);
|
|
let args = [
|
|
"create",
|
|
&container_system_name(name),
|
|
"--flake",
|
|
&flake_ref,
|
|
];
|
|
if stream {
|
|
container_run_streaming(&args, writer).await
|
|
} else {
|
|
container_run(&args).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,
|
|
lines,
|
|
boot,
|
|
output,
|
|
ref unit,
|
|
ref priority,
|
|
ref grep,
|
|
ref since,
|
|
ref until,
|
|
} => {
|
|
validate_container_system_name(container)?;
|
|
read_container_journal(
|
|
container,
|
|
lines,
|
|
boot,
|
|
output,
|
|
unit.as_deref(),
|
|
priority.as_deref(),
|
|
grep.as_deref(),
|
|
since.as_deref(),
|
|
until.as_deref(),
|
|
)
|
|
.await
|
|
}
|
|
|
|
PrivRequest::WriteNspawnFlags {
|
|
ref container,
|
|
ref binds,
|
|
ref isolation,
|
|
} => {
|
|
validate_container_system_name(container)?;
|
|
for bind in binds {
|
|
validate_bind_path(&bind.host_path)?;
|
|
validate_bind_path(&bind.container_path)?;
|
|
}
|
|
write_nspawn_flags(container, binds, isolation.as_ref())?;
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
PrivRequest::WriteResourceLimits {
|
|
ref container,
|
|
ref memory_max,
|
|
ref cpu_quota,
|
|
} => {
|
|
validate_container_system_name(container)?;
|
|
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 = format!("[Service]\nMemoryMax={memory_max}\nCPUQuota={cpu_quota}\n");
|
|
std::fs::write(&path, content).with_context(|| format!("write {path}"))?;
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
PrivRequest::RemoveServiceDropin { ref container } => {
|
|
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()))
|
|
}
|
|
|
|
PrivRequest::DaemonReload => {
|
|
let out = Command::new("systemctl")
|
|
.arg("daemon-reload")
|
|
.output()
|
|
.await
|
|
.context("invoke systemctl daemon-reload")?;
|
|
if !out.status.success() {
|
|
bail!(
|
|
"systemctl daemon-reload failed ({}): {}",
|
|
out.status,
|
|
String::from_utf8_lossy(&out.stderr).trim()
|
|
);
|
|
}
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
PrivRequest::ReloadGatewayNginx => sync_gateway_nginx().await,
|
|
|
|
PrivRequest::ChownSocketDir {
|
|
ref agent_name,
|
|
uid,
|
|
gid,
|
|
} => {
|
|
validate_agent_name(agent_name)?;
|
|
let path = socket_dir_path(agent_name);
|
|
std::os::unix::fs::chown(&path, Some(uid), Some(gid))
|
|
.with_context(|| format!("chown {} to {uid}:{gid}", path.display()))?;
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
PrivRequest::ChmodSocketDir {
|
|
ref agent_name,
|
|
mode,
|
|
} => {
|
|
use std::os::unix::fs::PermissionsExt as _;
|
|
validate_agent_name(agent_name)?;
|
|
let path = socket_dir_path(agent_name);
|
|
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(mode))
|
|
.with_context(|| format!("chmod {:o} {}", mode, path.display()))?;
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
|
|
PrivRequest::RunForgeAdmin { ref args } => {
|
|
for arg in args {
|
|
validate_forge_admin_arg(arg)?;
|
|
}
|
|
run_forge_admin(args).await
|
|
}
|
|
|
|
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,
|
|
} => {
|
|
validate_agent_name(agent_name)?;
|
|
write_agent_state_file(agent_name, "matrix-token", &format!("{token}\n"))
|
|
}
|
|
|
|
PrivRequest::RestartMatrixDaemon { ref agent_name } => {
|
|
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(),
|
|
))
|
|
}
|
|
}
|
|
}
|
|
|
|
/// 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)> {
|
|
use std::os::unix::fs::{MetadataExt as _, PermissionsExt as _};
|
|
|
|
let state_dir = PathBuf::from(AGENT_STATE_ROOT)
|
|
.join(agent_name)
|
|
.join("state");
|
|
// 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()))?;
|
|
|
|
let path = state_dir.join(filename);
|
|
std::fs::write(&path, content).with_context(|| format!("write {}", path.display()))?;
|
|
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600))
|
|
.with_context(|| format!("chmod 600 {}", path.display()))?;
|
|
|
|
// Chown to the state dir's owner so the agent process can read the file.
|
|
// 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) => {
|
|
if let Err(e) = std::os::unix::fs::chown(&path, Some(meta.uid()), Some(meta.gid())) {
|
|
tracing::warn!(
|
|
agent = %agent_name,
|
|
path = %path.display(),
|
|
error = %e,
|
|
"write_agent_state_file: chown 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, file = %filename, "wrote agent state file");
|
|
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(())
|
|
}
|
|
|
|
/// 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();
|
|
for line in stdout.lines() {
|
|
tracing::info!(target: "forgejo-admin", "{line}");
|
|
}
|
|
for line in stderr.lines() {
|
|
tracing::warn!(target: "forgejo-admin", "{line}");
|
|
}
|
|
if !out.status.success() {
|
|
bail!(
|
|
"forgejo admin {} failed ({}): {}",
|
|
args.join(" "),
|
|
out.status,
|
|
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();
|
|
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 `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.
|
|
#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
|
|
async fn read_container_journal(
|
|
container: &str,
|
|
lines: u32,
|
|
boot: bool,
|
|
output: JournalOutput,
|
|
unit: Option<&str>,
|
|
priority: Option<&str>,
|
|
grep: Option<&str>,
|
|
since: Option<&str>,
|
|
until: Option<&str>,
|
|
) -> Result<(String, String)> {
|
|
let mut args: Vec<String> = vec![
|
|
"-M".to_owned(),
|
|
container.to_owned(),
|
|
"--no-pager".to_owned(),
|
|
format!("--output={}", output.as_journalctl()),
|
|
"-n".to_owned(),
|
|
lines.to_string(),
|
|
];
|
|
if boot {
|
|
args.push("-b".to_owned());
|
|
}
|
|
if let Some(u) = unit {
|
|
args.push("-u".to_owned());
|
|
args.push(u.to_owned());
|
|
}
|
|
if let Some(p) = priority {
|
|
args.push("-p".to_owned());
|
|
args.push(p.to_owned());
|
|
}
|
|
// `--grep=`/`--since=`/`--until=` use the `=`-joined form so a value
|
|
// can never be parsed as a separate journalctl flag.
|
|
if let Some(g) = grep {
|
|
args.push(format!("--grep={g}"));
|
|
}
|
|
if let Some(s) = since {
|
|
args.push(format!("--since={s}"));
|
|
}
|
|
if let Some(u) = 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 nginx unit inside the `hive-gateway` container.
|
|
///
|
|
/// Queries `ActiveState` via `systemctl --machine=hive-gateway` (requires
|
|
/// root — machine-bus transport enters the container namespace), then
|
|
/// dispatches:
|
|
/// - `active` → `systemctl reload nginx` (SIGHUP, zero-downtime)
|
|
/// - `failed` → `systemctl reset-failed nginx` + `systemctl start nginx`
|
|
/// - otherwise → `systemctl start nginx`
|
|
///
|
|
/// 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([
|
|
"--machine=hive-gateway",
|
|
"show",
|
|
"--property=ActiveState",
|
|
"--value",
|
|
"nginx",
|
|
])
|
|
.output()
|
|
.await
|
|
.context("query nginx ActiveState in hive-gateway")?;
|
|
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 container 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(["--machine=hive-gateway", "reload", "nginx"])
|
|
.output()
|
|
.await
|
|
.context("reload nginx in hive-gateway")?;
|
|
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(["--machine=hive-gateway", "reset-failed", "nginx"])
|
|
.status()
|
|
.await;
|
|
let out = Command::new("systemctl")
|
|
.args(["--machine=hive-gateway", "start", "nginx"])
|
|
.output()
|
|
.await
|
|
.context("start nginx after reset-failed in hive-gateway")?;
|
|
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(["--machine=hive-gateway", "start", "nginx"])
|
|
.output()
|
|
.await
|
|
.context("start nginx in hive-gateway")?;
|
|
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}")
|
|
}
|
|
|
|
/// Path of the per-agent unix-socket dir on the host.
|
|
fn socket_dir_path(agent_name: &str) -> PathBuf {
|
|
PathBuf::from(format!("{SOCKET_DIR_ROOT}/{agent_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<()> {
|
|
if name == MANAGER_NAME {
|
|
return Ok(());
|
|
}
|
|
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 name == MANAGER_NAME {
|
|
return Ok(());
|
|
}
|
|
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`.
|
|
fn write_nspawn_flags(
|
|
container: &str,
|
|
binds: &[BindMount],
|
|
isolation: Option<&NetworkIsolation>,
|
|
) -> 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");
|
|
out.push_str("HOST_ADDRESS=\n");
|
|
let _ = writeln!(out, "LOCAL_ADDRESS={}", iso.agent_ip);
|
|
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 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();
|
|
let flags_joined = flags.join(" ");
|
|
let _ = writeln!(out, "EXTRA_NSPAWN_FLAGS=\"{flags_joined}\"");
|
|
std::fs::write(&path, out).with_context(|| format!("write {path}"))
|
|
}
|