Watch
0
0
Fork
You've already forked hyperhive
0
hyperhive/swarm-nats-auth/src/main.rs
atlas 91fdf96a67 docs: drop harness systemd settings; swarm-nats-auth: refuse the hive's shared agent client
Docs (review on #4948): the harness's restart settings are not useful to
a reader, so the `Restart`/`RestartSec` bullet in docs/turn-loop and the
"30 seconds later, no start limit" wording in docs/swarm are removed.
What stays is the behaviour: the harness exits and comes back until the
queue accepts it.

Grant: with the harness's shared-client fallback gone, nothing presents
the hive's shared agent client (`hive-<hive>-agent`) to the queue;
swarm-logs uses that OIDC client over HTTP only. swarm-nats-auth now
refuses it outright instead of granting the configured
`$SWARM.term.{hive}.>` / `$SWARM.agent-state.{hive}.>` publish and its
`_INBOX.client.<id>.>` subscribe, so an agent holding it can no longer
publish as a sibling. The agent arm still runs before the hive arm, so
that id never falls through to its hive's grant.

Removed with it: the `--agent-publish-subject` flag and its two uses in
swarm-nats.nix, `Policy`'s agent subject list, and
swarm-queue-client's `client_inbox_prefix`. The module-eval now asserts
the flag is absent; the policy tests assert the refusal and that the
hive itself is still admitted.
2026-10-04 22:24:51 +02:00

404 lines
17 KiB
Rust

//! Auth-callout responder for the swarm's NATS queue.
//!
//! `nix/host-modules/swarm-nats.nix` configures `nats-server` with an
//! `auth_callout` block and no responder, which is the fail-closed state: the
//! server answers `"auth_required":true` and admits nobody. This binary is what
//! makes it able to say *yes*.
//!
//! It connects as the one callout-exempt user (by nkey, never by name — the
//! server refuses to start if that entry carries a username), subscribes to
//! `$SYS.REQ.USER.AUTH`, validates the presented bearer token against
//! authelia's introspection endpoint (or, for an agent's own token, against the
//! secret store: see `agent_token`), and answers with a NATS user JWT signed
//! by the account key. A rejection is answered explicitly: silence is
//! indistinguishable from the responder being down, and the queue is the
//! swarm's control path.
//!
//! # Secrets
//!
//! Every credential is taken as a **path**, never a value. Two reasons, both
//! previously learned the hard way here: a value in nix config is rendered into
//! the world-readable store, and a value in `argv` is readable by anyone via
//! `/proc/<pid>/cmdline`, which is `0444`. Paths are not secrets, so passing
//! them as flags is fine.
use std::path::PathBuf;
use anyhow::Context;
use clap::Parser;
use futures_util::StreamExt;
mod agent_token;
mod introspect;
mod policy;
mod request;
mod respond;
/// Subject the NATS server publishes authorization requests on.
const AUTH_SUBJECT: &str = "$SYS.REQ.USER.AUTH";
#[derive(Debug, Parser)]
#[command(
name = "swarm-nats-auth",
about = "Auth-callout responder for the swarm NATS queue"
)]
struct Args {
/// NATS server to connect to.
#[arg(long, default_value = "nats://127.0.0.1:4222")]
nats_url: String,
/// Path to the seed of the callout-exempt user this responder connects as.
/// Its public half is `services.hyperhive.deploy.nats.calloutUserPublicKey`.
#[arg(long)]
user_seed_file: PathBuf,
/// Path to the account signing seed used to sign issued user JWTs. Its
/// public half is `services.hyperhive.deploy.nats.calloutIssuerPublicKey`.
#[arg(long)]
issuer_seed_file: PathBuf,
/// Authelia's OIDC introspection endpoint.
#[arg(long)]
introspection_url: String,
/// `OAuth2` client id this responder introspects as. Must match
/// `services.hyperhive.swarm.nats.clientId`, whose default this mirrors.
#[arg(long, default_value = "swarm-nats")]
client_id: String,
/// Account an admitted client is placed in. Must name an entry in the
/// server's own `accounts` block — in server-config mode the account is
/// resolved by *name*, so a value the server does not know is a grant it
/// refuses. The module passes its `clientAccount`; the default mirrors it.
#[arg(long, default_value = "APP")]
account: String,
/// Path to this responder's own OIDC client secret.
#[arg(long)]
client_secret_file: PathBuf,
/// Client-id prefix that marks a hive. `swarm-authelia.nix` mints one
/// machine client per roster entry as `hive-<name>`, while the KV key is
/// the bare `<name>` — this is the contract between the two, declared
/// rather than inferred from the shape of an id.
#[arg(long, default_value = "hive-")]
hive_client_prefix: String,
/// Suffix that marks an agent container, appended to the hive's own client
/// id: `swarm-authelia.nix` mints one machine client per roster entry as
/// `hive-<name>-agent` — per **hive**, not per agent, because agents are
/// created at runtime and a per-agent client would make creating one a
/// config change plus a reload.
///
/// So this identity says which hive an agent belongs to and never which
/// agent: two agents on one hive are indistinguishable under it. The
/// responder refuses it; the suffix is what tells it apart from a hive.
///
/// A suffix on the hive's id rather than a prefix of its own, because
/// `agent-<name>` reads as *the agent called `<name>`* — the one thing
/// this identity does not carry.
///
/// A suffix starts with `-`, so it has to be accepted as a value rather
/// than read as a short-flag cluster — without this the default below
/// cannot be passed back in on the command line that produced it.
#[arg(long, default_value = "-agent", allow_hyphen_values = true)]
agent_client_suffix: String,
/// Client ids allowed to read every hive's status. Repeatable. The
/// default is the swarm controller, which is the only reader that exists.
#[arg(long = "reader-client", default_values_t = [String::from("swarm-controller")])]
reader_clients: Vec<String>,
/// Additional subjects a hive may publish to, with `{hive}` standing for
/// its own name. Repeatable, empty by default.
///
/// The extension point for a second stream published by the same
/// `hive-<name>` identity — lifecycle notices, say. Without it, adding one
/// means changing this responder; with it, a deployment says so and the
/// subject still lands inside that hive's own namespace.
#[arg(long = "hive-publish-subject")]
hive_publish_subjects: Vec<String>,
/// Subjects an agent that presented its own credential may publish to,
/// with `{agent}` standing for its name. Repeatable, empty by default,
/// which denies every agent token.
#[arg(long = "agent-token-publish-subject")]
agent_token_publish_subjects: Vec<String>,
/// Role on the secret store's `cert` auth mount this responder logs in
/// as to read agent credentials. The store is found through the `BAO_*`
/// environment; with none, every agent token is denied.
#[arg(long, default_value = "swarm-nats-auth")]
store_cert_role: String,
}
/// Read a secret file and strip surrounding whitespace.
///
/// The trim matters: an `echo`-created seed file ends in a newline, and an
/// nkey seed with a trailing byte is not a seed — it fails at parse with a
/// message about encoding rather than about the file, which sends you looking
/// in the wrong place. The value is never logged, and the error deliberately
/// names only the path.
fn read_secret(path: &std::path::Path) -> anyhow::Result<String> {
let raw = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
let trimmed = raw.trim();
if trimmed.is_empty() {
anyhow::bail!("{} is empty", path.display());
}
Ok(trimmed.to_owned())
}
#[tokio::main]
async fn main() -> anyhow::Result<()> {
swarm_queue_client::install_crypto_provider();
tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
)
// This is a systemd-managed daemon — stdout always goes to journald,
// never a human terminal, and journald doesn't strip ANSI escapes:
// they land in victorialogs as raw byte-array spam otherwise.
.with_ansi(false)
.init();
let args = Args::parse();
let client_secret = read_secret(&args.client_secret_file)?;
// The bucket is NOT a flag. `swarm_queue_client::status`'s own docs say
// why: reader and writer must name the same bucket, and an option is a
// way for two deployments to disagree about which one that is. This
// responder is the third end that names it, so it takes the same
// constant rather than a copy of the literal.
// Fails the process rather than warning: a policy that cannot express a
// per-hive namespace is not a policy this responder should run with, and
// the queue's fail-closed state (no responder) is a legible outage where a
// silently over-broad grant is not.
let policy = policy::Policy::new(
args.hive_client_prefix.clone(),
args.agent_client_suffix.clone(),
swarm_queue_client::status::BUCKET.to_owned(),
args.reader_clients.clone(),
args.hive_publish_subjects.clone(),
)?
.with_agent_token_subjects(args.agent_token_publish_subjects.clone())?;
let store = agent_token::Store::from_env(args.store_cert_role.clone());
let http = reqwest::Client::new();
let issuer = nkeys::KeyPair::from_seed(&read_secret(&args.issuer_seed_file)?)
.context("parse the account signing seed")?;
let user_seed = read_secret(&args.user_seed_file)?;
let client = async_nats::ConnectOptions::with_nkey(user_seed)
.name("swarm-nats-auth")
.connect(&args.nats_url)
.await
.with_context(|| format!("connect to {}", args.nats_url))?;
let mut requests = client
.subscribe(AUTH_SUBJECT)
.await
.with_context(|| format!("subscribe to {AUTH_SUBJECT}"))?;
tracing::info!(
nats_url = %args.nats_url,
subject = AUTH_SUBJECT,
"swarm-nats-auth: connected, awaiting authorization requests"
);
while let Some(msg) = requests.next().await {
// Decode failures are logged and dropped, never propagated: this loop
// is the swarm's login path, and exiting on one malformed payload
// would let any client take authentication down for everyone.
let req = match request::decode(&msg.payload) {
Ok(req) => req,
Err(e) => {
tracing::warn!(error = ?e, "undecodable auth request, ignoring");
continue;
}
};
// No token is a denial, not an error: an anonymous connect is a
// normal thing for a client to attempt and an abnormal thing to
// grant. Neither the store nor introspection is reached until
// something was presented.
let (caller, permissions) = match req
.connect_opts
.auth_token
.as_deref()
.map(agent_token::classify)
{
// The caller is the name the token claims, logged whether or not
// the secret proved it; `granted` says which.
Some(agent_token::Presented::Agent(token)) => (
Some(format!("agent:{}", token.agent)),
agent_token::authorize(&policy, store.as_ref(), &token).await,
),
Some(agent_token::Presented::Malformed(e)) => {
tracing::warn!(error = %e, "malformed agent token; denying");
(None, None)
}
Some(agent_token::Presented::Bearer(token)) => {
introspected(&policy, &http, &args, &client_secret, token).await
}
None => (None, None),
};
// The caller is an identifier, not a credential: a client id, or
// `agent:<name>` for an agent token. One line per auth request, a connect
// or a reconnect, so `hive-<h>-agent` lines count requests made on a
// hive's shared credential, not agents.
tracing::info!(
user_nkey = %req.user_nkey,
server_id = %req.server_id.id,
granted = permissions.is_some(),
caller = caller.as_deref().unwrap_or("-"),
"auth request"
);
// Always reply, including on a denial. A server that hears nothing
// cannot tell a refusing responder from a dead one, so silence turns
// every rejection into a 2s timeout and hides an outage inside what
// looks like ordinary denials.
let Some(reply_to) = msg.reply.clone() else {
tracing::warn!("auth request had no reply subject; dropping");
continue;
};
let token = match &permissions {
Some(permissions) => respond::grant(
&issuer,
&args.account,
&req.server_id.id,
&req.user_nkey,
permissions,
),
None => respond::deny(&issuer, &req.server_id.id, &req.user_nkey),
};
if let Err(e) = client.publish(reply_to, token.into()).await {
tracing::warn!(error = ?e, "failed to publish auth response");
}
}
anyhow::bail!("subscription to {AUTH_SUBJECT} ended")
}
/// The OIDC path: who the `IdP` says presented `token`, and what the policy
/// grants that client. `None` permissions is a denial.
///
/// The caller is an identity or nothing — see `introspect`'s module docs.
/// There is no "admitted, identity unknown" branch to write here because
/// there is no such value to receive.
async fn introspected(
policy: &policy::Policy,
http: &reqwest::Client,
args: &Args,
client_secret: &str,
token: &str,
) -> (Option<String>, Option<policy::Permissions>) {
let caller = introspect::identify_caller(
http,
&args.introspection_url,
&args.client_id,
client_secret,
token,
)
.await
// An introspection that could not be *made* is a denial too. The
// failure modes of an HTTP call are exactly the conditions under
// which an attacker would most like this to fall open.
.unwrap_or_else(|e| {
tracing::warn!(error = ?e, "introspection failed; denying");
None
});
// Admission said who; the policy says what. A caller the `IdP`
// vouches for but no rule matches is denied — see `policy`'s module
// docs for why that is deny and not "connect with nothing".
let permissions = caller.as_deref().and_then(|id| policy.permissions(id));
if let (Some(id), None) = (caller.as_deref(), permissions.as_ref()) {
// Loud, and the one case an operator has to be able to find: a
// valid credential refused by our own policy. The alternative is
// a client that authenticates fine and mysteriously cannot work.
tracing::warn!(
caller = %id,
"authenticated client matches no policy rule; denying"
);
}
(caller, permissions)
}
#[cfg(test)]
mod tests {
use super::*;
/// The argument vector `swarm-nats.nix` actually builds, in the form it
/// builds it: every flag space-separated, and the suffix's value opening
/// with `-`. Read as a short-flag cluster it aborts before `main` runs,
/// and a responder that never starts denies every client — which reaches
/// the operator as `authorization violation` on an unrelated process.
#[test]
fn the_units_own_argument_vector_parses() {
let args = Args::try_parse_from([
"swarm-nats-auth",
"--user-seed-file",
"/run/credentials/callout-user.seed",
"--issuer-seed-file",
"/run/credentials/issuer.seed",
"--client-secret-file",
"/run/credentials/oidc-client.secret",
"--introspection-url",
"https://auth.example/api/oidc/introspection",
"--hive-client-prefix",
"hive-",
"--agent-client-suffix",
"-agent",
"--agent-token-publish-subject",
"$SWARM.term.{agent}",
"--agent-token-publish-subject",
"$SWARM.agent-state.{agent}",
"--agent-token-publish-subject",
"$KV.agent-icons.{agent}",
"--agent-token-publish-subject",
"$SWARM.term.{agent}.sub.>",
"--agent-token-publish-subject",
"$JS.API.DIRECT.GET.KV_agent-schedules.$KV.agent-schedules.{agent}",
"--store-cert-role",
"swarm-nats-auth",
])
.expect("the unit's own argument vector must parse");
assert_eq!(args.agent_client_suffix, "-agent");
assert_eq!(args.agent_token_publish_subjects.len(), 5);
}
/// The control for the case above: an ordinary value parses through the
/// same flag, so a failure there is about the leading `-` and not about
/// the argument being unknown.
#[test]
fn an_agent_suffix_without_a_leading_dash_parses_too() {
let args = Args::try_parse_from([
"swarm-nats-auth",
"--user-seed-file",
"/run/credentials/callout-user.seed",
"--issuer-seed-file",
"/run/credentials/issuer.seed",
"--client-secret-file",
"/run/credentials/oidc-client.secret",
"--introspection-url",
"https://auth.example/api/oidc/introspection",
"--agent-client-suffix",
"agent",
])
.expect("an ordinary suffix must parse");
assert_eq!(args.agent_client_suffix, "agent");
}
/// The default is the value the deployment passes, so the two cannot
/// drift into disagreeing about what an agent client is called.
#[test]
fn the_default_suffix_is_what_the_module_passes() {
let args = Args::try_parse_from([
"swarm-nats-auth",
"--user-seed-file",
"/run/credentials/callout-user.seed",
"--issuer-seed-file",
"/run/credentials/issuer.seed",
"--client-secret-file",
"/run/credentials/oidc-client.secret",
"--introspection-url",
"https://auth.example/api/oidc/introspection",
])
.expect("the required flags alone must parse");
assert_eq!(args.agent_client_suffix, "-agent");
}
}