6cbabc3515
Caught in review. Cosmetic, but a comment that misstates the constant beside it is the kind of thing the next reader trusts instead of the code.
519 lines
22 KiB
Rust
519 lines
22 KiB
Rust
//! Connecting to the swarm message queue as an authenticated client.
|
|
//!
|
|
//! Shared by every process that needs the queue — the swarm controller reads
|
|
//! hive status out of it, a hive publishes its own status into it — because
|
|
//! the *connect* is identical for all of them and only the use differs.
|
|
//! Duplicating it per binary would put credential handling in two places, and
|
|
//! a token-refresh fix would then have to be found twice.
|
|
//!
|
|
//! The queue admits every non-responder client through `auth_callout`: a
|
|
//! client presents a token at CONNECT, the callout responder introspects it
|
|
//! against authelia and mints a user JWT if it is good. So each participant is
|
|
//! an ordinary client that needs an identity of its own — the controller is
|
|
//! not a hive, and the per-hive clients issued from the roster are not its to
|
|
//! use.
|
|
//!
|
|
//! Two things about that shape drive everything here:
|
|
//!
|
|
//! - **A token expires.** Authelia issues `client_credentials` access tokens
|
|
//! with `expires_in: 3599`. Authentication happens at CONNECT, so a
|
|
//! long-lived connection is fine — but a *reconnect* an hour later needs a
|
|
//! token that was minted an hour later.
|
|
//! - **`async-nats` re-runs an auth callback per connection attempt** (it is
|
|
//! handed that attempt's nonce). So the refresh belongs in the callback and
|
|
//! not in a timer: there is no window in which the client presents a token
|
|
//! that has expired since it was minted.
|
|
//!
|
|
//! ⚠️ That second point is also a trap that cost a production incident —
|
|
//! **anything in a retry path runs at the FAILURE rate.** See `CachedToken`
|
|
//! and `MAX_RECONNECT_DELAY`.
|
|
|
|
use std::path::PathBuf;
|
|
|
|
/// Everything that can go wrong reaching the swarm queue.
|
|
///
|
|
/// `thiserror` and not `anyhow` because this is a library: a caller gets a
|
|
/// type it can match on, and the binaries that consume it keep using
|
|
/// `anyhow` — `?` converts for free, so nothing downstream is more verbose
|
|
/// for it. The same split `hive-claude` uses.
|
|
///
|
|
/// The variants are the failures an operator acts on differently: a
|
|
/// half-configured environment is a deployment bug, a refused token is an
|
|
/// identity-provider config problem, an unreachable queue is a network
|
|
/// one. Collapsing them into one string would make that distinction a
|
|
/// matter of reading prose.
|
|
#[derive(Debug, thiserror::Error)]
|
|
pub enum Error {
|
|
#[error(
|
|
"swarm queue is half-configured: {prefix}_NATS_URL, \
|
|
{prefix}_OIDC_TOKEN_ENDPOINT, {prefix}_OIDC_CLIENT_ID and \
|
|
{prefix}_OIDC_CLIENT_SECRET_FILE must be set together or not at all"
|
|
)]
|
|
PartialConfig { prefix: String },
|
|
|
|
#[error("reading the queue client secret from {path}")]
|
|
ClientSecret {
|
|
path: String,
|
|
#[source]
|
|
source: std::io::Error,
|
|
},
|
|
|
|
#[error("building the token-endpoint HTTP client")]
|
|
HttpClient(#[source] reqwest::Error),
|
|
|
|
/// Distinct from `HttpClient` because the operator's next move differs:
|
|
/// this one names a path they configured, and it fires before any
|
|
/// network call.
|
|
#[error("reading the token-endpoint CA certificate from {path}")]
|
|
CaFile {
|
|
path: String,
|
|
#[source]
|
|
source: std::io::Error,
|
|
},
|
|
|
|
#[error("parsing the token-endpoint CA certificate from {path} as PEM")]
|
|
CaParse {
|
|
path: String,
|
|
#[source]
|
|
source: reqwest::Error,
|
|
},
|
|
|
|
#[error("requesting an access token from authelia")]
|
|
TokenRequest(#[source] reqwest::Error),
|
|
|
|
/// The body carries authelia's own `error_description`, and it is far
|
|
/// more useful than the status alone: a wrong grant says
|
|
/// `unauthorized_client`, a wrong secret says `invalid_client`, and
|
|
/// those point at different config.
|
|
#[error("authelia refused the token request ({status}): {body}")]
|
|
TokenRefused {
|
|
status: reqwest::StatusCode,
|
|
body: String,
|
|
},
|
|
|
|
#[error("parsing authelia's token response")]
|
|
TokenResponse(#[source] serde_json::Error),
|
|
|
|
#[error("connecting to the swarm queue at {url}")]
|
|
Connect {
|
|
url: String,
|
|
#[source]
|
|
source: async_nats::ConnectError,
|
|
},
|
|
|
|
/// Distinct from [`Error::Connect`] on purpose: that one is a connect
|
|
/// that was attempted and refused, this one is a request made before
|
|
/// any connection exists. The second is a caller-ordering problem and
|
|
/// the first is a deployment one.
|
|
#[error("not connected to the swarm queue (client state: {0:?})")]
|
|
NotConnected(async_nats::connection::State),
|
|
|
|
#[cfg(feature = "kv")]
|
|
#[error("creating the {bucket} bucket")]
|
|
CreateBucket {
|
|
bucket: &'static str,
|
|
#[source]
|
|
source: async_nats::jetstream::context::CreateKeyValueError,
|
|
},
|
|
}
|
|
|
|
/// Render an error and its source chain on one line.
|
|
///
|
|
/// **Use this instead of `{:#}` anywhere an [`Error`] is rendered without
|
|
/// first being `?`-converted into an `anyhow::Error`.** `anyhow`'s
|
|
/// `Display` special-cases `f.alternate()` to walk the source chain;
|
|
/// `thiserror`'s derive does not, so `{e:#}` and `{e}` render
|
|
/// identically for this type. A call site that formatted an
|
|
/// `anyhow::Error` with `{:#}` and now holds an [`Error`] therefore keeps
|
|
/// compiling, keeps looking right, and silently drops the cause — which
|
|
/// is the half that says *why*, and is exactly what a one-shot boot
|
|
/// warning with no retry needs most.
|
|
///
|
|
/// Public for that reason: the fix cannot live only inside this crate's
|
|
/// own auth callback while the callers it was written for reach for
|
|
/// `{:#}` and get nothing.
|
|
pub fn chain(error: &dyn std::error::Error) -> String {
|
|
let mut rendered = error.to_string();
|
|
let mut source = error.source();
|
|
while let Some(cause) = source {
|
|
rendered.push_str(": ");
|
|
rendered.push_str(&cause.to_string());
|
|
source = cause.source();
|
|
}
|
|
rendered
|
|
}
|
|
|
|
/// The hive-status KV bucket, shared by the hive that writes it and the
|
|
/// controller that reads it. Behind the `kv` feature — see the module doc
|
|
/// for why a bucket name and its config belong to neither end alone.
|
|
#[cfg(feature = "kv")]
|
|
pub mod status;
|
|
|
|
/// Only the fields this needs; authelia returns several.
|
|
#[derive(serde::Deserialize)]
|
|
struct TokenResponse {
|
|
access_token: String,
|
|
/// Seconds the token stays valid — authelia says 3599 for
|
|
/// `client_credentials`. Read rather than assumed, because it is what
|
|
/// decides when the cache below stops handing the same token back, and a
|
|
/// hardcoded hour would silently become wrong the day the identity
|
|
/// provider is retuned.
|
|
expires_in: u64,
|
|
}
|
|
|
|
/// A minted token and the moment it stops being usable.
|
|
///
|
|
/// Exists because the auth callback runs **per connection attempt**, not per
|
|
/// hour: without a cache, every reconnect mints, and a queue that cannot
|
|
/// connect turns its retry loop into a token-request loop against the
|
|
/// identity provider. That is not hypothetical — the first version of this
|
|
/// crate did exactly that, and authelia answered `429 Too Many Requests`
|
|
/// continuously until the client was changed.
|
|
struct CachedToken {
|
|
token: String,
|
|
expires_at: std::time::Instant,
|
|
}
|
|
|
|
/// Re-mint this long before expiry rather than at it. A token that expires
|
|
/// mid-CONNECT is refused by the server, which costs a whole reconnect cycle
|
|
/// to discover — far more than minting slightly early costs.
|
|
const TOKEN_REFRESH_SKEW: std::time::Duration = std::time::Duration::from_mins(2);
|
|
|
|
/// Cap on how long `async-nats` waits between reconnect attempts.
|
|
///
|
|
/// ⚠️ Deliberately far above the library's default, and the default is the
|
|
/// bug: `reconnect_delay_callback_default` backs off exponentially but clamps
|
|
/// at **4 seconds** (`connector.rs`), forever. Every attempt runs the auth
|
|
/// callback, so an unreachable queue means one token request every 4s
|
|
/// indefinitely — measured in production as a continuous
|
|
/// `429 Too Many Requests / temporarily_unavailable` from authelia, which then
|
|
/// keeps *itself* alive: the rate limit outlives whatever first broke the
|
|
/// connection, and its log volume buries the original cause.
|
|
///
|
|
/// The cost of raising it is slower recovery from a brief outage. That is the
|
|
/// right trade for an infrastructure service: a minute of staleness is a
|
|
/// dashboard reading `stale`, whereas a hot loop against the identity provider
|
|
/// degrades every other client of it.
|
|
const MAX_RECONNECT_DELAY: std::time::Duration = std::time::Duration::from_mins(1);
|
|
|
|
/// Where the controller finds the queue and what it authenticates with.
|
|
///
|
|
/// Every field comes from an environment variable the NixOS module sets, the
|
|
/// same way `load_hives` takes the roster — a config change is a redeploy, and
|
|
/// this process reads no file it was not pointed at.
|
|
#[derive(Debug, Clone)]
|
|
pub struct QueueConfig {
|
|
/// `nats://host:port` for the swarm queue.
|
|
pub url: String,
|
|
/// Authelia's token endpoint, e.g. `https://auth.<swarm>/api/oidc/token`.
|
|
pub token_endpoint: String,
|
|
/// The controller's own `OAuth2` client id.
|
|
pub client_id: String,
|
|
/// File holding the client secret's PLAINTEXT.
|
|
///
|
|
/// A path and not a value: the secret is minted on the authelia host and
|
|
/// read here, and putting it in the environment would publish it to
|
|
/// anything that can read `/proc/<pid>/environ`.
|
|
pub client_secret_file: PathBuf,
|
|
/// Extra trust anchor for the token endpoint, when it is not signed by
|
|
/// a publicly-trusted CA.
|
|
///
|
|
/// Optional, and deliberately NOT part of the all-or-none group below: a
|
|
/// swarm fronted by a public certificate needs no extra anchor, and
|
|
/// making this required would break that deployment to fix ours. Absent
|
|
/// means "the platform's roots are enough", which is the correct default
|
|
/// for a client that might talk to anything.
|
|
///
|
|
/// ⚠️ Without it, a swarm using its own CA fails at TLS with
|
|
/// `invalid peer certificate: UnknownIssuer` — the anchor exists on the
|
|
/// host and this client simply never looked at it.
|
|
pub ca_file: Option<PathBuf>,
|
|
}
|
|
|
|
impl QueueConfig {
|
|
/// Read the config from `<prefix>_NATS_URL`, `<prefix>_OIDC_TOKEN_ENDPOINT`,
|
|
/// `<prefix>_OIDC_CLIENT_ID` and `<prefix>_OIDC_CLIENT_SECRET_FILE`, or
|
|
/// `None` when the queue was not wired up for this deployment.
|
|
///
|
|
/// The prefix is a parameter rather than a constant because the variables
|
|
/// belong to the *consuming unit* — a NixOS module sets them alongside its
|
|
/// other options, and two daemons sharing one name would be a worse
|
|
/// coupling than passing four characters. What is shared is the RULE
|
|
/// below, not the spelling.
|
|
///
|
|
/// `None` rather than an error on purpose: a daemon serves its other
|
|
/// surfaces on hosts where the queue is not enabled, and refusing to start
|
|
/// there would trade a missing feature for a dead process. What must NOT
|
|
/// happen is a *half* configuration silently behaving like an absent one —
|
|
/// hence the explicit partial check below.
|
|
pub fn from_env(prefix: &str) -> Result<Option<Self>, Error> {
|
|
let url = std::env::var(format!("{prefix}_NATS_URL")).ok();
|
|
let token_endpoint = std::env::var(format!("{prefix}_OIDC_TOKEN_ENDPOINT")).ok();
|
|
let client_id = std::env::var(format!("{prefix}_OIDC_CLIENT_ID")).ok();
|
|
let secret = std::env::var(format!("{prefix}_OIDC_CLIENT_SECRET_FILE")).ok();
|
|
|
|
// Read outside the match on purpose: this one is INDEPENDENT of the
|
|
// all-or-none rule, so it must not participate in the tuple that
|
|
// decides whether the queue is configured at all. A CA path with no
|
|
// queue is meaningless rather than half-configured.
|
|
let ca_file = std::env::var(format!("{prefix}_OIDC_CA_FILE"))
|
|
.ok()
|
|
.map(PathBuf::from);
|
|
|
|
match (url, token_endpoint, client_id, secret) {
|
|
(None, None, None, None) => Ok(None),
|
|
(Some(url), Some(token_endpoint), Some(client_id), Some(secret)) => Ok(Some(Self {
|
|
url,
|
|
token_endpoint,
|
|
client_id,
|
|
client_secret_file: PathBuf::from(secret),
|
|
ca_file,
|
|
})),
|
|
// A partially-set environment is a deployment bug, and the failure
|
|
// it would otherwise produce is the expensive kind: the process
|
|
// comes up "fine", never connects, and the data it was supposed to
|
|
// move silently stops moving. Naming the variables costs one line.
|
|
_ => Err(Error::PartialConfig {
|
|
prefix: prefix.to_owned(),
|
|
}),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Mint a fresh access token for the controller's own client.
|
|
///
|
|
/// `client_credentials`, because there is no user here: the controller
|
|
/// authenticates as itself. Authelia refuses the `openid` scope for this grant
|
|
/// (a machine client receives an access token and never an id-token), so no
|
|
/// scope is requested.
|
|
async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result<CachedToken, Error> {
|
|
// Read per call rather than caching: the file is small, and a cached
|
|
// secret would survive a rotation that the operator believes took effect.
|
|
let secret = tokio::fs::read_to_string(&cfg.client_secret_file)
|
|
.await
|
|
.map_err(|source| Error::ClientSecret {
|
|
path: cfg.client_secret_file.display().to_string(),
|
|
source,
|
|
})?;
|
|
|
|
let response = http
|
|
.post(&cfg.token_endpoint)
|
|
.form(&[
|
|
("grant_type", "client_credentials"),
|
|
("client_id", cfg.client_id.as_str()),
|
|
("client_secret", secret.trim()),
|
|
])
|
|
.send()
|
|
.await
|
|
.map_err(Error::TokenRequest)?;
|
|
|
|
// Why the body and not just the status: see `Error::TokenRefused`.
|
|
let status = response.status();
|
|
let body = response.text().await.unwrap_or_default();
|
|
if !status.is_success() {
|
|
return Err(Error::TokenRefused { status, body });
|
|
}
|
|
|
|
let parsed: TokenResponse = serde_json::from_str(&body).map_err(Error::TokenResponse)?;
|
|
Ok(CachedToken {
|
|
token: parsed.access_token,
|
|
expires_at: std::time::Instant::now() + std::time::Duration::from_secs(parsed.expires_in),
|
|
})
|
|
}
|
|
|
|
/// Fail fast unless the client is actually connected.
|
|
///
|
|
/// **Call this before every `JetStream` request.** `retry_on_initial_connect`
|
|
/// means the client exists before it is usable, and a `JetStream` request made
|
|
/// in that window does not fail — it *waits*, for longer than any caller
|
|
/// should (measured: still going at 15s against a queue that simply refuses
|
|
/// the credential). On a request path that hangs a poll; on a periodic task it
|
|
/// hangs the task, including the shutdown branch it never reaches.
|
|
///
|
|
/// Testing for `!= Connected` rather than `== Disconnected` is the whole
|
|
/// point: a client that has never connected once sits in `Pending`, so the
|
|
/// `Disconnected` test passes it straight through to the hang it was written
|
|
/// to prevent. That is exactly the case that matters — a process whose
|
|
/// credential is wrong from boot never reaches `Disconnected`, because it was
|
|
/// never connected to begin with.
|
|
///
|
|
/// Naming the state is also the better error: "not connected" is actionable,
|
|
/// a timeout is not.
|
|
pub fn ensure_connected(client: &async_nats::Client) -> Result<(), Error> {
|
|
let state = client.connection_state();
|
|
if state != async_nats::connection::State::Connected {
|
|
return Err(Error::NotConnected(state));
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Connect to the swarm queue, presenting a token on each connection attempt
|
|
/// and minting a fresh one only when the cached one is near expiry.
|
|
///
|
|
/// The rejected alternative — mint once, pass a static `auth_token`, own the
|
|
/// reconnect loop — fails in the way this subsystem exists to prevent: the
|
|
/// token expires, the controller keeps serving, its status data quietly stops
|
|
/// updating, and nothing says so until someone reads a dashboard.
|
|
pub async fn connect(cfg: QueueConfig) -> Result<async_nats::Client, Error> {
|
|
// A timeout, because this client runs INSIDE the auth callback: a token
|
|
// endpoint that accepts the connection and then never answers would hang
|
|
// the callback, and with it the connection attempt that invoked it, with
|
|
// no retry and nothing in the log to say why. Failing fast lets
|
|
// `async-nats` do what it already does well — back off and try again.
|
|
// 10s is generous for a form POST to a local IdP.
|
|
let mut builder = reqwest::Client::builder().timeout(std::time::Duration::from_secs(10));
|
|
|
|
// The swarm's own CA, when the token endpoint is signed by it. ADDED, not
|
|
// substituted: `add_root_certificate` extends the default set rather than
|
|
// replacing it, so a swarm can front authelia publicly and still have
|
|
// this work.
|
|
//
|
|
// Failing here rather than falling back to the platform roots is the
|
|
// point — an operator who named a CA file wants that anchor, and a
|
|
// silent fallback would turn their typo into `UnknownIssuer` five layers
|
|
// away, inside an auth callback, on a retry loop.
|
|
if let Some(path) = &cfg.ca_file {
|
|
let pem = std::fs::read(path).map_err(|source| Error::CaFile {
|
|
path: path.display().to_string(),
|
|
source,
|
|
})?;
|
|
let cert = reqwest::Certificate::from_pem(&pem).map_err(|source| Error::CaParse {
|
|
path: path.display().to_string(),
|
|
source,
|
|
})?;
|
|
builder = builder.add_root_certificate(cert);
|
|
}
|
|
|
|
let http = builder.build().map_err(Error::HttpClient)?;
|
|
let url = cfg.url.clone();
|
|
|
|
// Shared across every invocation of the callback below, which is the
|
|
// whole point: the callback fires per connection ATTEMPT, so without
|
|
// somewhere to remember the last token, "mint on demand" and "mint on
|
|
// every retry" are the same code.
|
|
let cache: std::sync::Arc<tokio::sync::Mutex<Option<CachedToken>>> =
|
|
std::sync::Arc::new(tokio::sync::Mutex::new(None));
|
|
|
|
let client = async_nats::ConnectOptions::with_auth_callback(move |_nonce| {
|
|
let http = http.clone();
|
|
let cfg = cfg.clone();
|
|
let cache = cache.clone();
|
|
async move {
|
|
let mut slot = cache.lock().await;
|
|
|
|
// Reuse while there is comfortably more than the skew left. The
|
|
// freshness property this callback exists for is preserved — a
|
|
// reconnect an hour later finds an expired entry and re-mints —
|
|
// but a reconnect ten seconds later does NOT ask the IdP again.
|
|
let reuse = slot
|
|
.as_ref()
|
|
.filter(|t| {
|
|
t.expires_at
|
|
.saturating_duration_since(std::time::Instant::now())
|
|
> TOKEN_REFRESH_SKEW
|
|
})
|
|
.map(|t| t.token.clone());
|
|
|
|
let token = if let Some(token) = reuse {
|
|
token
|
|
} else {
|
|
let minted = mint_token(&http, &cfg)
|
|
.await
|
|
// The callback's error type carries a string, so the
|
|
// source chain would be lost; flatten it rather than
|
|
// dropping it.
|
|
.map_err(|e| async_nats::AuthError::new(chain(&e)))?;
|
|
let token = minted.token.clone();
|
|
*slot = Some(minted);
|
|
token
|
|
};
|
|
|
|
let mut auth = async_nats::Auth::new();
|
|
auth.token = Some(token);
|
|
Ok(auth)
|
|
}
|
|
})
|
|
// See `MAX_RECONNECT_DELAY`. The library's default clamps at 4s forever,
|
|
// which turns an unreachable queue into a permanent 4s poll — and, before
|
|
// the cache above, a permanent 4s token-request loop against authelia.
|
|
// Exponential from 500ms so a momentary blip still reconnects promptly.
|
|
.reconnect_delay_callback(|attempts| {
|
|
let exp = u32::try_from(attempts.saturating_sub(1)).unwrap_or(u32::MAX);
|
|
std::cmp::min(
|
|
std::time::Duration::from_millis(
|
|
500u64.saturating_mul(2u64.saturating_pow(exp.min(8))),
|
|
),
|
|
MAX_RECONNECT_DELAY,
|
|
)
|
|
})
|
|
// The controller and the queue are separate units on (possibly)
|
|
// separate hosts, and nothing orders them. Without this, a queue that
|
|
// comes up one second later leaves the controller permanently
|
|
// queue-less until someone restarts it — a boot-order race that
|
|
// presents as "status has been unavailable since Tuesday".
|
|
//
|
|
// It also composes with the callback above rather than fighting it:
|
|
// each background attempt runs the callback, which hands over a token
|
|
// that is still valid — minting a new one only when the cached one is
|
|
// near expiry. (Before the cache, "runs the callback" meant "mints",
|
|
// and this retry loop was a token-request loop. See
|
|
// `MAX_RECONNECT_DELAY`.)
|
|
.retry_on_initial_connect()
|
|
.connect(&url)
|
|
.await
|
|
.map_err(|source| Error::Connect {
|
|
url: url.clone(),
|
|
source,
|
|
})?;
|
|
|
|
Ok(client)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
/// The all-unset case is the common one — most hosts do not run the queue.
|
|
///
|
|
/// Uses a prefix no deployment sets, so it cannot pass vacuously by
|
|
/// running inside a configured environment (the guard below covers the
|
|
/// same ground, and both are cheap).
|
|
#[test]
|
|
fn an_absent_environment_is_not_an_error() {
|
|
for k in [
|
|
"SWARM_QUEUE_TEST_NATS_URL",
|
|
"SWARM_QUEUE_TEST_OIDC_TOKEN_ENDPOINT",
|
|
"SWARM_QUEUE_TEST_OIDC_CLIENT_ID",
|
|
"SWARM_QUEUE_TEST_OIDC_CLIENT_SECRET_FILE",
|
|
] {
|
|
assert!(std::env::var(k).is_err(), "{k} must be unset for this test");
|
|
}
|
|
assert!(
|
|
QueueConfig::from_env("SWARM_QUEUE_TEST")
|
|
.expect("absent is not an error")
|
|
.is_none()
|
|
);
|
|
}
|
|
|
|
/// The half-set case is the one the rule exists for: a deployment bug that
|
|
/// would otherwise look exactly like "no queue configured".
|
|
///
|
|
/// SAFETY: single-threaded mutation of a process env var under a prefix no
|
|
/// other test or deployment uses; removed before returning.
|
|
#[test]
|
|
fn a_half_set_environment_is_a_hard_error() {
|
|
unsafe {
|
|
std::env::set_var("SWARM_QUEUE_HALF_NATS_URL", "nats://127.0.0.1:4222");
|
|
}
|
|
let err = QueueConfig::from_env("SWARM_QUEUE_HALF")
|
|
.expect_err("a partial set must not read as absent");
|
|
let msg = format!("{err}");
|
|
assert!(
|
|
msg.contains("SWARM_QUEUE_HALF_OIDC_CLIENT_ID"),
|
|
"the error must name the missing variables, got: {msg}"
|
|
);
|
|
unsafe {
|
|
std::env::remove_var("SWARM_QUEUE_HALF_NATS_URL");
|
|
}
|
|
}
|
|
}
|