Every admitted client got the same unrestricted grant, so any hive could write any other hive's status key. The responder now derives a permission set from the caller's identity and mints it into the user JWT. A hive may publish to its own KV key and the two JetStream subjects needed to reach it; the controller may list and fetch every key and write none; anything else is denied outright. Deny is the default because every other shape fails open, and silently: a client that matched no rule and kept the old grant would make the policy advisory. The subject sets are measured rather than reasoned about, and two of them are counter-intuitive. `$KV.<bucket>.<key>` alone does not let a client write that key, because the client resolves the bucket first. And `$JS.API.>` is not "the JetStream permission": it also covers `$JS.API.STREAM.DELETE`, with which a hive correctly refused on a neighbour's key can delete the whole bucket and every hive's data with it. Granting it would have made per-key scoping decorative, so the subjects are named individually and a test asserts the wildcard does not come back as a convenience. Minimality is by removal: each subject was dropped in turn to confirm the client breaks without it. That is not pedantry — an additive search had called a set minimal while two of its five subjects were never needed, which ships an unnecessary grant with a measurement attached making it look earned. Both grants include `STREAM.CREATE` on the one named stream, because `status::open_or_create` is called by both ends: either may arrive first on a fresh swarm, and without it a new swarm never gets a bucket at all. `CREATE` is not `UPDATE`, so a second arrival cannot reshape the bucket the first one made. `status::BUCKET` moves out from behind the `kv` feature so this responder can share it. The name is a `&str` with no dependencies and only `open_or_create` needs JetStream; gating the name forced a third consumer to choose between a stack it does not use and a copied literal, and the copied literal is exactly the disagreement that module exists to prevent. Only publish is scoped. Subscription permissions are unrestricted and unmeasured, and the module docs say so rather than implying a property nothing established.
647 lines
29 KiB
Rust
647 lines
29 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. See the module doc for why a bucket name and
|
|
/// its config belong to neither end alone.
|
|
///
|
|
/// The module itself is unconditional; only the parts that *open* the bucket
|
|
/// need the `kv` feature. The name is a `&str` with no dependencies, and a
|
|
/// third end names it too — the auth-callout responder, which derives the
|
|
/// subjects a hive may publish to from it without ever speaking `jetstream`.
|
|
/// Gating the name behind `kv` would have forced that consumer to choose
|
|
/// between pulling a JetStream stack it does not use and copying the literal,
|
|
/// which is the disagreement this module exists to prevent.
|
|
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 = token_request(http, cfg, 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),
|
|
})
|
|
}
|
|
|
|
/// Build the token request: `client_credentials`, authenticated with HTTP
|
|
/// Basic.
|
|
///
|
|
/// 🩸 **The credentials go in the `Authorization` header, not the form body.**
|
|
/// Both are legal OAuth 2.0 — `client_secret_basic` and `client_secret_post` —
|
|
/// but a client registration names *one*, and authelia's default (and ours) is
|
|
/// Basic. Sending them in the body got every token request refused with
|
|
/// `Client authentication failed … the registered client is configured to only
|
|
/// support 'client_secret_basic'`, which reached the operator as an endless
|
|
/// `429` because the retries tripped a rate limiter whose penalty grew faster
|
|
/// than the retry interval. The 429 then arrived *before* the credentials were
|
|
/// ever evaluated, so the one line naming the real cause appeared once an hour.
|
|
///
|
|
/// RFC 6749 §2.3.1 says clients SHOULD use Basic, both introspection callers in
|
|
/// this workspace already do, and a secret in a header is one fewer place for a
|
|
/// proxy to log it.
|
|
///
|
|
/// Split out of [`mint_token`] so the request's *shape* is testable without a
|
|
/// running identity provider — see the tests at the bottom of this file.
|
|
fn token_request(
|
|
http: &reqwest::Client,
|
|
cfg: &QueueConfig,
|
|
secret: &str,
|
|
) -> reqwest::RequestBuilder {
|
|
http.post(&cfg.token_endpoint)
|
|
.basic_auth(&cfg.client_id, Some(secret))
|
|
.form(&[("grant_type", "client_credentials")])
|
|
}
|
|
|
|
/// Build the HTTP client used to reach `cfg.token_endpoint`, trusting
|
|
/// `cfg.ca_file` when set. Shared by [`connect`]'s auth callback and by
|
|
/// [`mint_token_for`] — anything presenting this identity's credentials to
|
|
/// the token endpoint needs the swarm's own CA trusted the same way, so the
|
|
/// trust-anchor logic lives here once rather than once per caller.
|
|
fn build_http_client(cfg: &QueueConfig) -> Result<reqwest::Client, Error> {
|
|
let mut builder = reqwest::Client::builder().timeout(std::time::Duration::from_secs(10));
|
|
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);
|
|
}
|
|
builder.build().map_err(Error::HttpClient)
|
|
}
|
|
|
|
/// Mint a fresh token for `cfg`'s identity and hand back just the string —
|
|
/// no caching, a fresh HTTP client per call.
|
|
///
|
|
/// Public because the queue connection is not the only thing this identity
|
|
/// authenticates: "one identity per principal" means a caller that already
|
|
/// holds a [`QueueConfig`] for its queue connection authenticates anywhere
|
|
/// else it needs to prove who it is from the exact same client, rather than
|
|
/// a second identity being provisioned per destination. No caching here
|
|
/// unlike [`connect`]'s callback: that one exists because `async-nats` reruns
|
|
/// its callback per reconnect *attempt*, a hot path this isn't — a caller
|
|
/// outside that loop (e.g. `swarm-controller::auth`'s bridge client) mints
|
|
/// per call, same as this crate did before the reconnect-storm fix added the
|
|
/// cache.
|
|
pub async fn mint_token_for(cfg: &QueueConfig) -> Result<String, Error> {
|
|
let http = build_http_client(cfg)?;
|
|
Ok(mint_token(&http, cfg).await?.token)
|
|
}
|
|
|
|
/// 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> {
|
|
// Trusts `cfg.ca_file` when set (the swarm's own CA, when the token
|
|
// endpoint is signed by it) — see `build_http_client`. A timeout too,
|
|
// 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.
|
|
let http = build_http_client(&cfg)?;
|
|
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");
|
|
}
|
|
}
|
|
|
|
fn token_cfg() -> QueueConfig {
|
|
QueueConfig {
|
|
url: "nats://127.0.0.1:4222".to_owned(),
|
|
token_endpoint: "https://auth.example.com/api/oidc/token".to_owned(),
|
|
client_id: "hive-alpha".to_owned(),
|
|
client_secret_file: PathBuf::from("/nonexistent"),
|
|
ca_file: None,
|
|
}
|
|
}
|
|
|
|
/// 🩸 THE REGRESSION TEST FOR AN OUTAGE THAT RAN FOR WEEKS.
|
|
///
|
|
/// The credentials used to go in the form body (`client_secret_post`).
|
|
/// Authelia's client registration allows only `client_secret_basic`, so
|
|
/// every token request was refused — and the refusals tripped a rate
|
|
/// limiter whose 429 then arrived *before* the credentials were evaluated,
|
|
/// so the error naming the cause appeared roughly once an hour inside a
|
|
/// continuous storm of a different error.
|
|
///
|
|
/// This asserts the *shape of the request* rather than a server's reply,
|
|
/// which is the whole point: it fails on the old code with no identity
|
|
/// provider, no deployment and no network.
|
|
/// A client for inspecting a request, never for sending one.
|
|
///
|
|
/// 🩸 `reqwest::Client::new()` **panics in the nix build sandbox**, which
|
|
/// has no system CA store: `ClientBuilder::build()` reaches
|
|
/// `rustls_platform_verifier::Verifier::new()` and fails with "No CA
|
|
/// certificates were loaded from the system", and `new()` is
|
|
/// `build().expect(..)`. The test passed locally — a devshell has
|
|
/// `/etc/ssl/certs` — and failed in CI.
|
|
///
|
|
/// Turning verification off takes the `!certs_verification` branch, which
|
|
/// installs a no-op verifier and never consults the platform store, so
|
|
/// this builds anywhere. It is sound *here specifically* because nothing
|
|
/// is ever sent: the request is built and its bytes inspected.
|
|
fn offline_client() -> reqwest::Client {
|
|
reqwest::Client::builder()
|
|
.danger_accept_invalid_certs(true)
|
|
.build()
|
|
.expect("a client that verifies nothing needs no system trust store")
|
|
}
|
|
|
|
#[test]
|
|
fn the_token_request_authenticates_with_http_basic() {
|
|
let req = token_request(&offline_client(), &token_cfg(), "s3cret")
|
|
.build()
|
|
.expect("the token request must build");
|
|
|
|
let auth = req
|
|
.headers()
|
|
.get(reqwest::header::AUTHORIZATION)
|
|
.expect("credentials must travel in the Authorization header")
|
|
.to_str()
|
|
.expect("the header is ascii");
|
|
assert!(
|
|
auth.starts_with("Basic "),
|
|
"must be client_secret_basic, got: {auth}"
|
|
);
|
|
|
|
let body = std::str::from_utf8(
|
|
req.body()
|
|
.and_then(reqwest::Body::as_bytes)
|
|
.expect("the request has an in-memory body"),
|
|
)
|
|
.expect("the body is utf-8");
|
|
assert!(
|
|
body.contains("grant_type=client_credentials"),
|
|
"the grant type still belongs in the body, got: {body}"
|
|
);
|
|
// The half that was actually broken: a secret in the body is both the
|
|
// wrong auth method for our registration and a value proxies log.
|
|
assert!(
|
|
!body.contains("client_secret"),
|
|
"the secret must not be in the request body, got: {body}"
|
|
);
|
|
assert!(
|
|
!body.contains("client_id"),
|
|
"the client id belongs in the Basic credentials, got: {body}"
|
|
);
|
|
}
|
|
}
|