swarm: carry the agent queue credential from the host into the harness

hive-c0re stats the two files `swarm-bao-queue-agent` lands and forwards
them into every agent container as systemd credentials, and the harness
resolves a `QueueConfig` out of them at boot. Nothing connects yet.

A credential and not a bind mount, and the mode is what forces it: the
secret is root:0600 and the harness runs as the unprivileged agent user,
so a bind would deliver a file that user cannot open. nspawn's
`--load-credential` is read by the container manager as root and
re-exposed under the consuming unit's own `User=`. hive-c0re never reads
the bytes either way, which is just as well — it runs as `hive-core`.

Absent files stay legal and become visible rather than silent: the
publisher lives on the authelia host and mints on its first boot, so
"nothing at that path" is the ordinary early state of a swarm. c0re
forwards nothing and logs why; the harness logs that it has no queue.

The client id comes out of the delivered file rather than being rebuilt
from `hiveName` in nix, which is the agreement `swarm-secret-client`
states. `QueueConfig::from_env` wants it as a value, so the harness reads
the file itself — assigning the variable instead would need
`std::env::set_var` in a process that has already spawned threads.

Refs #3805
This commit is contained in:
atlas 2026-09-12 22:41:36 +02:00 committed by atlas
commit 353cdd9264
6 changed files with 436 additions and 7 deletions

1
Cargo.lock generated
View file

@ -1668,6 +1668,7 @@ dependencies = [
"schemars", "schemars",
"serde", "serde",
"serde_json", "serde_json",
"swarm-queue-client",
"tempfile", "tempfile",
"tokio", "tokio",
"tokio-stream", "tokio-stream",

View file

@ -31,6 +31,9 @@ rusqlite.workspace = true
schemars.workspace = true schemars.workspace = true
serde.workspace = true serde.workspace = true
serde_json.workspace = true serde_json.workspace = true
# Bare: `kv`/`notices` name buckets and streams this harness opens neither
# end of. All it wants from the crate is `QueueConfig` and, next, a connect.
swarm-queue-client.workspace = true
tokio.workspace = true tokio.workspace = true
tokio-stream.workspace = true tokio-stream.workspace = true
tower-http.workspace = true tower-http.workspace = true

View file

@ -28,6 +28,7 @@ mod serve_common;
mod state_entry_watch; mod state_entry_watch;
mod stats; mod stats;
mod stream_enrich; mod stream_enrich;
mod swarm_queue;
mod term_msg; mod term_msg;
mod todo_server; mod todo_server;
mod todos; mod todos;
@ -495,6 +496,9 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
let claude_dir = login::default_dir(); let claude_dir = login::default_dir();
let initial = LoginState::from_dir(&claude_dir); let initial = LoginState::from_dir(&claude_dir);
tracing::info!(state = ?initial, claude_dir = %claude_dir.display(), "harness boot"); tracing::info!(state = ?initial, claude_dir = %claude_dir.display(), "harness boot");
// Resolve the swarm-queue credential while the boot log is still the
// thing an operator is reading. Nothing connects yet.
swarm_queue::init();
// Config fact, stamped once — see `harness_state::write_api_key_mode`'s // Config fact, stamped once — see `harness_state::write_api_key_mode`'s
// doc for why hive-c0re needs this to stop reporting `needs_login` for // doc for why hive-c0re needs this to stop reporting `needs_login` for
// an agent whose `~/.claude/` is empty by design. // an agent whose `~/.claude/` is empty by design.

View file

@ -0,0 +1,262 @@
//! This agent's swarm-queue credentials, resolved once at boot.
//!
//! Three of the four coordinates arrive as environment variables the meta
//! flake renders into the harness unit. The fourth — the OIDC client id —
//! arrives as a *file*, delivered beside the secret as one systemd
//! credential pair (`nix/agent-modules/queue.nix`). That is the whole reason
//! this module exists rather than a bare [`QueueConfig::from_env`] call: the
//! id rides with the secret so a reader never has to spell `hive-<name>-agent`
//! a second time, and `from_env` wants it as a value.
//!
//! Reading the file and assigning the variable would put the same rule back
//! in `from_env`'s hands, but `std::env::set_var` is unsound in a process that
//! has already spawned threads, and this one has. So the rule is restated
//! here over the inputs this consumer actually has, and [`decide`] is the
//! single place it lives.
use std::path::Path;
use std::sync::OnceLock;
use swarm_queue_client::QueueConfig;
/// Variable prefix for this agent's coordinates. Distinct from `HIVE_C0RE`'s
/// on purpose: an agent authenticates as its own client, not as its hive.
const ENV_PREFIX: &str = "HIVE_AGENT";
/// Resolved once at boot, so the publisher that connects to the queue reads
/// one answer rather than re-deriving it per call.
static CONFIG: OnceLock<Option<QueueConfig>> = OnceLock::new();
/// The four variables the harness unit sets, before the client-id file is
/// read. Collected into a struct so [`decide`] is pure over them and the
/// process env is touched in exactly one place.
struct QueueEnv {
nats_url: Option<String>,
token_endpoint: Option<String>,
client_id_file: Option<String>,
client_secret_file: Option<String>,
/// Independent of the all-or-none group below, exactly as in
/// `QueueConfig::from_env`: a swarm behind a publicly-trusted
/// certificate needs no extra anchor. Nothing in this tree sets it for
/// an agent — a container already trusts the swarm root, which
/// `hive_c0re::meta` embeds at build time — so it is here for a
/// deployment that needs a different one, not for ours.
ca_file: Option<String>,
}
impl QueueEnv {
fn from_env() -> Self {
let var = |suffix: &str| std::env::var(format!("{ENV_PREFIX}_{suffix}")).ok();
Self {
nats_url: var("NATS_URL"),
token_endpoint: var("OIDC_TOKEN_ENDPOINT"),
client_id_file: var("OIDC_CLIENT_ID_FILE"),
client_secret_file: var("OIDC_CLIENT_SECRET_FILE"),
ca_file: var("OIDC_CA_FILE"),
}
}
}
/// What the environment plus the client-id file add up to.
enum Resolution {
/// Everything is here; the agent can reach the queue.
Configured(Box<QueueConfig>),
/// No queue for this agent, and that is a legal state — carries why.
Absent(&'static str),
/// Some of the environment, not all of it. A deployment bug rather than
/// an absent integration, so it is reported and then survived.
Partial,
}
/// Read the client id out of the file the credential landed at.
///
/// `None` for a missing or empty file, which is the ordinary state of a hive
/// whose secret store has nothing published yet — the credential simply is
/// not there, and nspawn forwards nothing. Trailing newline stripped: the
/// reader unit writes one and an id with a newline in it authenticates as
/// nobody.
fn read_client_id(path: &Path) -> Option<String> {
match std::fs::read_to_string(path) {
Ok(raw) => {
let id = raw.trim();
(!id.is_empty()).then(|| id.to_owned())
}
Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
Err(e) => {
tracing::warn!(path = %path.display(), error = %e, "reading the queue client id failed");
None
}
}
}
/// Decide what this agent's queue configuration is, given the environment and
/// whatever the client-id file held.
///
/// All four variables or none, for the reason `QueueConfig::from_env` gives:
/// a half-set environment produces a process that comes up fine and never
/// connects. The client id is graded separately from the four because its
/// file has a legal absence the variables do not — the unit names the path a
/// credential *would* arrive at whether or not one has been published yet.
fn decide(env: &QueueEnv, client_id: Option<String>) -> Resolution {
match (
env.nats_url.as_ref(),
env.token_endpoint.as_ref(),
env.client_id_file.as_ref(),
env.client_secret_file.as_ref(),
) {
(None, None, None, None) => Resolution::Absent("this hive has no swarm queue configured"),
(Some(url), Some(token_endpoint), Some(_), Some(secret)) => match client_id {
Some(client_id) => Resolution::Configured(Box::new(QueueConfig {
url: url.clone(),
token_endpoint: token_endpoint.clone(),
client_id,
client_secret_file: secret.into(),
ca_file: env.ca_file.as_ref().map(Into::into),
})),
None => {
Resolution::Absent("the queue credential has not been published to this hive yet")
}
},
_ => Resolution::Partial,
}
}
/// Resolve this agent's queue configuration and record it for later use.
///
/// Never fails: a harness that cannot reach the queue still serves its
/// operator, its web UI and its turn loop, so every outcome here is a log
/// line and not an exit.
pub fn init() {
let env = QueueEnv::from_env();
let client_id = env
.client_id_file
.as_deref()
.map(Path::new)
.and_then(read_client_id);
let resolved = match decide(&env, client_id) {
Resolution::Configured(cfg) => {
// url + client id only. The secret is a path in this struct and
// stays one: neither it nor its contents belong in a log.
tracing::info!(url = %cfg.url, client_id = %cfg.client_id, "swarm queue configured");
Some(*cfg)
}
Resolution::Absent(why) => {
tracing::info!(why, "no swarm queue for this agent");
None
}
Resolution::Partial => {
tracing::error!(
prefix = ENV_PREFIX,
"swarm queue half-configured: {ENV_PREFIX}_NATS_URL, \
{ENV_PREFIX}_OIDC_TOKEN_ENDPOINT, {ENV_PREFIX}_OIDC_CLIENT_ID_FILE and \
{ENV_PREFIX}_OIDC_CLIENT_SECRET_FILE are set together or not at all \
this agent will not connect"
);
None
}
};
let _ = CONFIG.set(resolved);
}
#[cfg(test)]
mod tests {
use super::{QueueEnv, Resolution, decide, read_client_id};
fn env(parts: [Option<&str>; 4]) -> QueueEnv {
let [nats_url, token_endpoint, client_id_file, client_secret_file] = parts;
QueueEnv {
nats_url: nats_url.map(str::to_owned),
token_endpoint: token_endpoint.map(str::to_owned),
client_id_file: client_id_file.map(str::to_owned),
client_secret_file: client_secret_file.map(str::to_owned),
ca_file: None,
}
}
fn full() -> QueueEnv {
env([
Some("nats://10.42.0.1:4222"),
Some("https://auth.t.local/api/oidc/token"),
Some("/run/credentials/hive-agent.service/hive-queue-agent-client-id"),
Some("/run/credentials/hive-agent.service/hive-queue-agent-secret"),
])
}
/// The reader unit writes the id with a trailing newline; an id carrying
/// one is a client authelia has never heard of.
#[test]
fn a_client_id_file_is_read_without_its_newline() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("client_id");
std::fs::write(&path, "hive-h1-agent\n").expect("write");
assert_eq!(read_client_id(&path).as_deref(), Some("hive-h1-agent"));
}
/// Both shapes of "no credential here": never delivered, or delivered
/// empty. Neither is an id, and treating an empty string as one would
/// authenticate as the anonymous client rather than failing.
#[test]
fn a_missing_or_empty_client_id_file_reads_as_no_id() {
let dir = tempfile::tempdir().expect("tempdir");
let missing = dir.path().join("client_id");
assert_eq!(read_client_id(&missing), None);
std::fs::write(&missing, "\n").expect("write");
assert_eq!(read_client_id(&missing), None);
}
#[test]
fn a_complete_environment_with_a_published_id_configures_the_queue() {
let e = full();
let Resolution::Configured(cfg) = decide(&e, Some("hive-h1-agent".to_owned())) else {
panic!("expected a configured queue");
};
assert_eq!(cfg.url, "nats://10.42.0.1:4222");
assert_eq!(cfg.client_id, "hive-h1-agent");
assert!(
cfg.client_secret_file.ends_with("hive-queue-agent-secret"),
"the secret stays a path: {}",
cfg.client_secret_file.display()
);
}
/// A hive with no swarm queue at all. Silence here is correct, and it has
/// to be distinguishable from the half-set case below — that distinction
/// is the only thing that makes the error branch worth logging.
#[test]
fn an_empty_environment_is_no_queue_rather_than_an_error() {
let e = env([None, None, None, None]);
assert!(matches!(decide(&e, None), Resolution::Absent(_)));
}
/// The credential's own absence. The unit names the path unconditionally
/// once the hive has a queue, so this is the state of every agent on a
/// swarm whose publisher has not run — legal, and not the error branch.
#[test]
fn a_complete_environment_with_no_published_id_is_no_queue() {
let e = full();
assert!(matches!(decide(&e, None), Resolution::Absent(_)));
}
/// Each single-variable omission, because the failure a partial set
/// produces is a harness that looks healthy and publishes nothing.
#[test]
fn any_missing_variable_is_a_partial_configuration() {
for drop in 0..4 {
let mut parts = [
Some("nats://10.42.0.1:4222"),
Some("https://auth.t.local/api/oidc/token"),
Some("/run/credentials/hive-agent.service/hive-queue-agent-client-id"),
Some("/run/credentials/hive-agent.service/hive-queue-agent-secret"),
];
parts[drop] = None;
let e = env(parts);
assert!(
matches!(
decide(&e, Some("hive-h1-agent".to_owned())),
Resolution::Partial
),
"dropping variable {drop} must report a partial configuration"
);
}
}
}

View file

@ -130,6 +130,71 @@ fn bind_child_agent_dirs(child: &str, binds: &mut Vec<BindMount>) {
} }
} }
/// Env var naming the host directory `swarm-bao-queue-agent.service` lands
/// this hive's agent queue credential in. Set by the hive-c0re NixOS module
/// from `deploy.hive-controller.queue.agentCredentialDir`; absent means this
/// daemon runs outside its unit, which is the same "no queue" answer as an
/// empty directory.
const QUEUE_CREDENTIAL_DIR_ENV: &str = "HIVE_C0RE_AGENT_QUEUE_CREDENTIAL_DIR";
/// systemd credential ids the two files arrive under inside the container.
/// `nix/agent-modules/queue.nix` spells the same two names in the harness
/// unit's `LoadCredential=`; neither side can discover the other's, so a
/// rename here is a rename there.
const QUEUE_SECRET_CREDENTIAL: &str = "hive-queue-agent-secret";
const QUEUE_CLIENT_ID_CREDENTIAL: &str = "hive-queue-agent-client-id";
/// Forward this hive's agent queue credential into `agent_name`'s container
/// as two systemd credentials, or nothing when the publisher has not landed
/// it yet.
///
/// **A credential and not a bind mount, because of the mode.** The secret is
/// `root:0600` on the host and the harness runs as the unprivileged agent
/// user, so a bind would deliver a file that user cannot open. nspawn's
/// `--load-credential` is read by the container manager as root and
/// re-exposed under the consuming unit's own `User=`, which is the whole
/// difference. hive-c0re never reads the bytes either way — it runs as
/// `hive-core` and only ever needs to know the file is *there*, which a
/// `0755` directory permits.
///
/// **Absent files are legal.** Authelia mints the secret on its first boot
/// and a publisher on that host puts it in the store, so "nothing at that
/// path" is the ordinary early state of a swarm rather than a fault. Saying
/// so at `info!` is what keeps it from being invisible: without a line here,
/// an agent that never connects looks identical to one that was never
/// configured.
fn queue_agent_credentials(agent_name: &str, dir: Option<&Path>) -> Vec<CredentialMount> {
let Some(dir) = dir else {
tracing::info!(
%agent_name,
"no {QUEUE_CREDENTIAL_DIR_ENV} in this daemon's environment — agent gets no swarm queue"
);
return Vec::new();
};
let secret = dir.join("secret");
let client_id = dir.join("client_id");
// Both or neither, for the same reason `QueueConfig::from_env` refuses a
// half-set environment: a client holding one of the two comes up fine and
// never connects.
if !secret.is_file() || !client_id.is_file() {
tracing::info!(
%agent_name, dir = %dir.display(),
"swarm queue credential not published yet — agent gets no swarm queue"
);
return Vec::new();
}
vec![
CredentialMount {
name: QUEUE_SECRET_CREDENTIAL.to_owned(),
host_path: secret.to_string_lossy().into_owned(),
},
CredentialMount {
name: QUEUE_CLIENT_ID_CREDENTIAL.to_owned(),
host_path: client_id.to_string_lossy().into_owned(),
},
]
}
/// Idempotently rewrite the lines in `/etc/nixos-containers/<container>.conf` /// Idempotently rewrite the lines in `/etc/nixos-containers/<container>.conf`
/// that hive-c0re owns: `PRIVATE_NETWORK` (always 1), `HOST_ADDRESS` (the /// that hive-c0re owns: `PRIVATE_NETWORK` (always 1), `HOST_ADDRESS` (the
/// bridge gateway IP) and `EXTRA_NSPAWN_FLAGS` (the runtime-dir bind). What /// bridge gateway IP) and `EXTRA_NSPAWN_FLAGS` (the runtime-dir bind). What
@ -185,12 +250,11 @@ async fn set_nspawn_flags(
// is needed here — the bind alone is enough. // is needed here — the bind alone is enough.
let claude_mount = container_claude_mount(agent_name); let claude_mount = container_claude_mount(agent_name);
// No hive-wide secrets are forwarded into agent containers. hive-priv // The agent's own swarm-queue credential — the one thing forwarded in
// still accepts a credential list (see `write_nspawn_flags`), but // here, and forwarded *as a credential* rather than a bind for the
// nothing produces one: the only entry was the OTEL upstream token, // reason `queue_agent_credentials` records.
// and an agent has no business holding the hive's credential for let queue_credential_dir = std::env::var_os(QUEUE_CREDENTIAL_DIR_ENV).map(PathBuf::from);
// anything outside it. let load_creds = queue_agent_credentials(agent_name, queue_credential_dir.as_deref());
let load_creds: Vec<CredentialMount> = Vec::new();
let mut binds: Vec<BindMount> = vec![ let mut binds: Vec<BindMount> = vec![
BindMount { BindMount {
@ -324,7 +388,10 @@ async fn set_nspawn_flags(
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::{BindMount, bind_child_agent_dirs}; use super::{
BindMount, QUEUE_CLIENT_ID_CREDENTIAL, QUEUE_SECRET_CREDENTIAL, bind_child_agent_dirs,
queue_agent_credentials,
};
fn child_binds() -> Vec<BindMount> { fn child_binds() -> Vec<BindMount> {
let mut binds = Vec::new(); let mut binds = Vec::new();
@ -393,4 +460,49 @@ mod tests {
bind_child_agent_dirs("../escape", &mut binds); bind_child_agent_dirs("../escape", &mut binds);
assert!(binds.is_empty()); assert!(binds.is_empty());
} }
/// The ordinary state of a swarm before the publisher on the authelia
/// host has run: the directory is named and empty. Forwarding a
/// credential whose source does not exist would make every agent
/// container refuse to start, so this has to be silence-with-a-log
/// rather than a partial list.
#[test]
fn an_unpublished_queue_credential_forwards_nothing() {
let dir = tempfile::tempdir().expect("tempdir");
assert!(queue_agent_credentials("iris", None).is_empty());
assert!(queue_agent_credentials("iris", Some(dir.path())).is_empty());
std::fs::write(dir.path().join("secret"), "s").expect("write secret");
assert!(
queue_agent_credentials("iris", Some(dir.path())).is_empty(),
"a secret with no client id is not a usable credential"
);
}
/// Both files present: two credentials, named the same two ids the
/// harness unit imports, pointing at the two files the reader unit
/// wrote.
#[test]
fn a_published_queue_credential_forwards_both_files() {
let dir = tempfile::tempdir().expect("tempdir");
std::fs::write(dir.path().join("secret"), "s").expect("write secret");
std::fs::write(dir.path().join("client_id"), "hive-h1-agent").expect("write client_id");
let creds = queue_agent_credentials("iris", Some(dir.path()));
let named: Vec<(String, String)> = creds
.into_iter()
.map(|c| (c.name, c.host_path))
.collect::<Vec<_>>();
assert_eq!(
named,
[
(
QUEUE_SECRET_CREDENTIAL.to_owned(),
dir.path().join("secret").to_string_lossy().into_owned(),
),
(
QUEUE_CLIENT_ID_CREDENTIAL.to_owned(),
dir.path().join("client_id").to_string_lossy().into_owned(),
),
]
);
}
} }

View file

@ -700,6 +700,12 @@ const CANONICAL_INPUTS: &[&str] = &["nixpkgs"];
/// trivial to stub from tests (which build their own slice instead of /// trivial to stub from tests (which build their own slice instead of
/// touching process-wide env). /// touching process-wide env).
const FORWARDED_VARS: &[&str] = &[ const FORWARDED_VARS: &[&str] = &[
// Where the swarm queue is and where its tokens are minted, as an agent
// container reaches them. Both are addresses the *host* computes — the
// queue's from the bridge IP an agent routes over, never a loopback
// address, which inside a container is the agent itself.
"HIVE_AGENT_NATS_URL",
"HIVE_AGENT_OIDC_TOKEN_ENDPOINT",
"HIVE_FORGE_URL", "HIVE_FORGE_URL",
"HIVE_FORGE_PUBLIC_URL", "HIVE_FORGE_PUBLIC_URL",
"HIVE_MATRIX_URL", "HIVE_MATRIX_URL",
@ -724,6 +730,16 @@ const FORWARDED_VARS: &[&str] = &[
/// answer. Half-wiring this map is not a missing nicety; it is a value that /// answer. Half-wiring this map is not a missing nicety; it is a value that
/// evaluates fine and is silently wrong. /// evaluates fine and is silently wrong.
const FORWARDED_VAR_OPTIONS: &[(&str, &str)] = &[ const FORWARDED_VAR_OPTIONS: &[(&str, &str)] = &[
// Read at build time to decide whether the harness unit declares the
// queue credential at all — see `nix/agent-modules/queue.nix`. An agent
// whose option says no queue and whose env says otherwise logs the
// partial-config error rather than half-connecting, which is the same
// failure this map's doc describes and the reason both halves are wired.
("HIVE_AGENT_NATS_URL", "hyperhive.queue.natsUrl"),
(
"HIVE_AGENT_OIDC_TOKEN_ENDPOINT",
"hyperhive.queue.tokenEndpoint",
),
("HIVE_FORGE_URL", "hyperhive.forge.url"), ("HIVE_FORGE_URL", "hyperhive.forge.url"),
("HIVE_MATRIX_URL", "hyperhive.matrix.url"), ("HIVE_MATRIX_URL", "hyperhive.matrix.url"),
("HYPERHIVE_HIVE_NAME", "hyperhive.hiveName"), ("HYPERHIVE_HIVE_NAME", "hyperhive.hiveName"),
@ -2027,6 +2043,37 @@ mod tests {
); );
} }
/// The queue coordinates travel as a pair and are wired through both
/// halves of the forwarded-var machinery: the env var the harness reads at
/// runtime and the option the harness *unit* is built from. An agent that
/// got only the env var would declare no credential and then report a
/// partial config — visible, but a rebuild away from working.
#[test]
fn queue_coordinates_are_forwarded_as_both_env_and_option() {
for var in ["HIVE_AGENT_NATS_URL", "HIVE_AGENT_OIDC_TOKEN_ENDPOINT"] {
assert!(
FORWARDED_VARS.contains(&var),
"{var} must be forwarded into every agent's env"
);
}
let mut out = String::new();
push_forwarded_var_options(
&mut out,
&[
("HIVE_AGENT_NATS_URL", "nats://10.42.0.1:4222".to_string()),
(
"HIVE_AGENT_OIDC_TOKEN_ENDPOINT",
"https://auth.t.local/api/oidc/token".to_string(),
),
],
);
assert_eq!(
out,
" hyperhive.queue.natsUrl = \"nats://10.42.0.1:4222\";\n\
\x20 hyperhive.queue.tokenEndpoint = \"https://auth.t.local/api/oidc/token\";\n"
);
}
#[test] #[test]
fn require_service_urls_accepts_a_rendered_forge_url() { fn require_service_urls_accepts_a_rendered_forge_url() {
require_service_urls(&[ require_service_urls(&[