From a6935e57c336969ec13b9762dd156854ae0b9ce8 Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 12 Oct 2026 20:48:45 +0200 Subject: [PATCH] swarm-queue-client: re-read a held OIDC client secret when the IdP refuses it hive-c0re (`hive-`), swarm-controller and swarm-review-classifier read their queue client secret from the store once and held it as `ClientSecret::Value` for the life of the process. After authelia re-minted a secret and swarm-secret-publish stored the new one, every later token mint presented the old value and was refused until the daemon restarted. `ClientSecret::Stored` replaces `Value`: a `StoredSecret` holds the secret in memory with a caller-supplied read callback, so the crate takes no store dependency. When the token endpoint answers 400 or 401, `mint_token` (and its blocking sibling, used by the controller's OTLP exporters) re-reads it and, if the store holds a different secret, mints once more with it. One re-read runs at a time and is spawned so it outlives a caller that gives up. A re-read that fails or finds the refused value delays the next one, 5 s doubling to 5 min, so a secret that stays wrong costs one store read per delay. Logs name the client id and never the secret. swarm-nats-auth's `oidc_secret` was the template; its re-read machinery now sits on `StoredSecret`, so the tree has one copy. The responder gains the same re-read backoff. The controller keeps its `IDENTITY` OnceLock: it now holds a `QueueConfig` whose secret is re-read, and the lazily built vcs-metrics exporter reads the identity from it. agent_renewal.rs's module doc said a re-minted agent queue secret only reaches the agent at its next start; hive-agent reads the store on every connection attempt after its first, so the doc now says that. Part of #5017 (slice Q). --- Cargo.lock | 4 + docs/swarm/credentials.md | 4 +- hive-c0re/src/swarm_queue.rs | 15 +- swarm-controller/src/agent_renewal.rs | 8 +- swarm-controller/src/queue_identity.rs | 19 +- swarm-nats-auth/src/main.rs | 11 +- swarm-nats-auth/src/oidc_secret.rs | 393 ++++---------------- swarm-queue-client/Cargo.toml | 14 + swarm-queue-client/src/lib.rs | 262 ++++++++++++-- swarm-queue-client/src/stored_secret.rs | 459 ++++++++++++++++++++++++ swarm-review-classifier/src/main.rs | 12 +- 11 files changed, 822 insertions(+), 379 deletions(-) create mode 100644 swarm-queue-client/src/stored_secret.rs diff --git a/Cargo.lock b/Cargo.lock index 0509a77e..f524ee11 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4954,6 +4954,9 @@ name = "swarm-queue-client" version = "0.1.0" dependencies = [ "async-nats", + "axum", + "data-encoding", + "futures-util", "reqwest", "rustls", "serde", @@ -4962,6 +4965,7 @@ dependencies = [ "thiserror 2.0.18", "tokio", "tracing", + "tracing-subscriber", "utoipa", ] diff --git a/docs/swarm/credentials.md b/docs/swarm/credentials.md index cfee085e..d3241665 100644 --- a/docs/swarm/credentials.md +++ b/docs/swarm/credentials.md @@ -60,7 +60,7 @@ of the cell says how. | `swarm/agents//matrix/main` | `swarm-controller`, with the swarm's appservice token, at agent creation and in a five-minute pass | the agent container itself, under the certificate its hive passed in | ✅ the pass re-mints when the stored token is missing, unknown to the homeserver, or someone else's | ✅ `hive-matrix-daemon` exits when the homeserver rejects its token, and a five-minute timer restarts it, which reads the store again | | `swarm/agents//matrix/` | `swarm-controller` | the agent container itself, under the certificate its hive passed in | must be stated | must be stated | | `swarm/controller/swarm-controller/matrix/appservice-token` | `swarm-matrix-ctl`, inside the `hive-matrix` container, once | `swarm-controller`, under its own certificate | ❌ `swarm-matrix-ctl` mints it once; the container keeps its copy and republishes it when the store's differs | ✅ the controller reads it on every five-minute matrix pass | -| `swarm/controller/swarm-controller/oidc/client` | authelia, at its first boot; `swarm-secret-publish` copies it in | `swarm-controller`, under its own certificate, once at start | ❌ authelia mints it once. A re-mint is republished by `swarm-secret-publish`'s path unit | ❌ read once at start; the controller holds the old value until it restarts | +| `swarm/controller/swarm-controller/oidc/client` | authelia, at its first boot; `swarm-secret-publish` copies it in | `swarm-controller`, under its own certificate, at start and on refusal | ❌ authelia mints it once. A re-mint is republished by `swarm-secret-publish`'s path unit | ✅ read at start and held in memory; read again when authelia refuses it, at most once per backoff of 5 s doubling to 5 min while the store holds the refused value | | `swarm/controller/swarm-controller/forge/token` | the forge, inside the `hive-forge` container, once; `swarm-bao-forge-publisher` on the forge's host copies it in under its own certificate, with a write-only grant | `swarm-controller`, under its own certificate, at start | ❌ the forge mints it once and keeps its copy in the container; the publisher writes that copy each time the forge container starts and each time the mint writes the file | ❌ the controller reads it at start, and in the background with backoff until it first finds it; a replaced token reaches it with the next restart | | `swarm/agents//bao-mtls` | the store's agent PKI mount (`deploy.bao.agentPkiMountPath`), which generates the key, at `swarm-controller`'s request at agent creation | `hive-c0re`, under the hive's own certificate, when it writes the agent's container config; then `hive-agent-bao-identity` in the agent container, under the leaf it holds | ✅ `swarm-controller`'s five-minute pass re-issues a live agent's leaf once it's past half its validity (45 of 90 days, read from the certificate itself) | ✅ `hive-agent-bao-identity` re-reads it every 10 minutes and swaps a new leaf into `/run/hive-agent-bao-identity/`, which every store login in the container presents. It reads under the leaf it holds, so it renews only while that leaf is valid | | `swarm/agents//queue` | `swarm-controller`, at agent creation | `hive-agent` in the agent container, under the agent's own certificate, held in memory — the identity it presents to the swarm queue, naming that one agent rather than its hive; `swarm-logs` in the agent container, under the same certificate, on every query, as the bearer it presents at the log store; `swarm-nats-auth` and `swarm-logs-auth`, each under its own certificate, to check a presented secret | ✅ `swarm-controller`'s five-minute pass re-mints a live agent's secret once it's 45 days old by `minted_at` on the stored object; a secret with no `minted_at` gets one stamped, value unchanged. The pass skips agents declared `Destroyed` — declaring an agent destroyed deletes every version of the path instead, the undo of the mint rather than another one | ✅ `hive-agent` reads the path before its first connect and again on every reconnect attempt, so a reconnect after a re-mint presents the new secret. An open connection keeps the secret it connected with; after a revocation the agent keeps retrying under the queue client's backoff. `swarm-logs` reads it on every query and once more after a 401 | @@ -70,7 +70,7 @@ of the cell says how. | `swarm/hives//matrix/sender-token` | `swarm-controller`, with the swarm's appservice token, for every hive in its directory in a five-minute pass | `swarm-controller` under its own certificate, before it decides whether to mint, and hive-c0re's `stored_sender_token()`, under the hive's own certificate | ✅ the controller's pass re-mints when the stored token is missing, unknown to the homeserver, or someone else's | ✅ hive-c0re's matrix sweep reads the store every run and overwrites its token file when the store's token differs | | `swarm/hives//forge/token` | `swarm-controller`, with the forge's admin API, for every hive in its directory in a five-minute pass | `swarm-controller` under its own certificate, before it decides whether to mint, and hive-c0re's `forge::reader_token()`, under the hive's own certificate | ✅ the controller's pass re-mints when the stored token is missing or no longer matches the forge (last eight characters and scopes) | ✅ hive-c0re reads the store on every forge read | | `swarm/hives//queue/agent` | authelia | `swarm-bao-queue-agent` on the hive's host, under its own per-hive certificate; no agent's policy reaches it | must be stated | must be stated | -| `swarm/hives//oidc/client` | authelia, at its first boot; `swarm-secret-publish` copies it in | hive-c0re, under the hive's own certificate, when it first connects to the queue | ❌ authelia mints it once. A re-mint is republished by `swarm-secret-publish`'s path unit | ❌ read until the first connect succeeds, retried with a backoff of up to five minutes while it's missing; then held until hive-c0re restarts | +| `swarm/hives//oidc/client` | authelia, at its first boot; `swarm-secret-publish` copies it in | hive-c0re, under the hive's own certificate, when it first connects to the queue | ❌ authelia mints it once. A re-mint is republished by `swarm-secret-publish`'s path unit | ✅ read until the first connect succeeds, retried with a backoff of up to five minutes while it's missing; then held in memory and read again when authelia refuses it | | `swarm/services//oidc/client` | authelia | that service's own reader, under its own certificate — the queue responder's under `swarm-nats-auth`'s, the homeserver's under `swarm-matrix-ctl`'s, the forge's under `swarm-forge-publisher`'s, the review classifier's under `swarm-review-classifier`'s; the bao UI's by the store's granter | must be stated | must be stated | | `swarm/services/ci-runner/registration-token` | `swarm-controller`, from the forge's admin API, at start and in a five-minute pass; its grant on the path is write-only | `swarm-bao-ci-runner` on the runner's host, under the runner's own certificate, into a file only `swarm-ci-runner-register` in the runner's container loads; no other principal's policy reads it | ❌ the forge hands out one token until an admin resets it; the controller writes a reset token within five minutes | ✅ `swarm-bao-ci-runner` reads it at boot, on deploy and every ten minutes, and restarts the runner when the value changed | | `swarm/services/review-classifier/anthropic-api-key` | the operator, by hand in the store's UI, field `api_key` | `swarm-review-classifier` in its container, under its own certificate, at the start of every run; held for that run and handed to `claude`'s environment only; no other principal's policy reads it | ❌ an operator's key; replaced by writing a new one | ✅ read on every run, so a written or removed key takes effect on the next run | diff --git a/hive-c0re/src/swarm_queue.rs b/hive-c0re/src/swarm_queue.rs index 664ca87b..b8b6cecf 100644 --- a/hive-c0re/src/swarm_queue.rs +++ b/hive-c0re/src/swarm_queue.rs @@ -13,7 +13,8 @@ //! The queue's address, the token endpoint and the client id come from the //! environment. The client secret comes from the swarm's secret store, at //! `swarm_secret_client::queue::hive_client_path`, read under this hive's own -//! certificate and held only in memory. `swarm-secret-publish`, on the host +//! certificate, held only in memory and read again when authelia refuses it +//! (`swarm_queue_client::stored_secret`). `swarm-secret-publish`, on the host //! that runs authelia, puts it there. //! //! Connects lazily on first use rather than at boot — nothing here @@ -27,6 +28,7 @@ use std::path::PathBuf; use std::time::Duration; use anyhow::{Context, Result, anyhow, bail}; +use swarm_queue_client::stored_secret::StoredSecret; use swarm_queue_client::{ClientSecret, QueueConfig}; use swarm_secret_client::queue::{self, HiveClientCredential}; use tokio::sync::Mutex; @@ -229,11 +231,14 @@ struct Coordinates { impl Coordinates { fn with_secret(self, secret: String) -> QueueConfig { + let stored = StoredSecret::new(self.client_id.clone(), secret, || async { + fetch_secret().await.map_err(Into::into) + }); QueueConfig { url: self.url, token_endpoint: self.token_endpoint, client_id: self.client_id, - client_secret: ClientSecret::Value(secret), + client_secret: ClientSecret::Stored(stored), ca_file: self.ca_file, } } @@ -327,14 +332,14 @@ mod tests { ); } - #[test] - fn a_full_environment_carries_the_stored_secret_as_a_value() { + #[tokio::test] + async fn a_full_environment_carries_the_stored_secret_in_memory() { let cfg = coordinates_from(env(&FULL)) .expect("complete") .expect("configured") .with_secret("s3cret".to_owned()); assert_eq!(cfg.client_id, "hive-alpha"); - assert!(matches!(&cfg.client_secret, ClientSecret::Value(v) if v == "s3cret")); + assert!(matches!(&cfg.client_secret, ClientSecret::Stored(s) if s.current() == "s3cret")); } /// The secret is not in the store when this daemon first asks, and is diff --git a/swarm-controller/src/agent_renewal.rs b/swarm-controller/src/agent_renewal.rs index 8fd470e0..a3e38bd9 100644 --- a/swarm-controller/src/agent_renewal.rs +++ b/swarm-controller/src/agent_renewal.rs @@ -10,10 +10,10 @@ //! //! A running agent re-reads its certificate every ten minutes under the one it //! holds (`nix/agent-modules/bao.nix`); the old one stays valid until it -//! expires. A new secret reaches the agent at its next start, when the -//! container fetches it. The old secret stops working at once: the queue checks -//! it only at `CONNECT`, so an open connection is unaffected, but a reconnect -//! before that restart presents the old secret and is denied. +//! expires. The old queue secret stops working at once, but the queue checks it +//! only at `CONNECT`, so an open connection is unaffected. The agent reads its +//! secret from the store on every connection attempt after its first +//! (`hive-agent/src/swarm_queue.rs`), so a reconnect presents the new one. //! //! [`spawn`]'s pass, once the queue is up and every [`RECONCILE_INTERVAL`], //! inserts a job per agent with anything due: `MintAgentIdentity` for the diff --git a/swarm-controller/src/queue_identity.rs b/swarm-controller/src/queue_identity.rs index 840c71c4..3246493a 100644 --- a/swarm-controller/src/queue_identity.rs +++ b/swarm-controller/src/queue_identity.rs @@ -3,9 +3,10 @@ //! //! The queue's address, the token endpoint and the client id come from the //! environment. The client secret comes from the swarm's secret store, at -//! `swarm_secret_client::queue::controller_client_path`, read once at start -//! under this daemon's own certificate and held only in memory — it is never -//! on this host's disk. `swarm-secret-publish`, on the host that runs +//! `swarm_secret_client::queue::controller_client_path`, read at start under +//! this daemon's own certificate, held only in memory and read again when +//! authelia refuses it (`swarm_queue_client::stored_secret`). It is never on +//! this host's disk. `swarm-secret-publish`, on the host that runs //! authelia, puts it there. //! //! [`init`] runs once in `main`, before anything calls [`get`]. @@ -16,6 +17,7 @@ use std::sync::OnceLock; use std::time::Duration; use anyhow::{Context, Result, bail}; +use swarm_queue_client::stored_secret::StoredSecret; use swarm_queue_client::{ClientSecret, QueueConfig}; use swarm_secret_client::queue::{self, ControllerCredential}; @@ -40,11 +42,14 @@ struct Coordinates { impl Coordinates { fn with_secret(self, secret: String) -> QueueConfig { + let stored = StoredSecret::new(self.client_id.clone(), secret, || async { + fetch_secret().await.map_err(Into::into) + }); QueueConfig { url: self.url, token_endpoint: self.token_endpoint, client_id: self.client_id, - client_secret: ClientSecret::Value(secret), + client_secret: ClientSecret::Stored(stored), ca_file: self.ca_file, } } @@ -201,14 +206,14 @@ mod tests { assert_eq!(c.ca_file, None); } - #[test] - fn the_fetched_secret_is_held_as_a_value_not_a_path() { + #[tokio::test] + async fn the_fetched_secret_is_held_in_memory_not_a_path() { let cfg = coordinates_from(env(&FULL)) .expect("complete") .expect("configured") .with_secret("s3cret".to_owned()); assert!( - matches!(&cfg.client_secret, ClientSecret::Value(v) if v == "s3cret"), + matches!(&cfg.client_secret, ClientSecret::Stored(s) if s.current() == "s3cret"), "{:?}", cfg.client_secret ); diff --git a/swarm-nats-auth/src/main.rs b/swarm-nats-auth/src/main.rs index 49bb645b..31c2b79e 100644 --- a/swarm-nats-auth/src/main.rs +++ b/swarm-nats-auth/src/main.rs @@ -385,16 +385,17 @@ async fn load_callout_keys( async fn load_client_secret( keys: &impl keystore::KeyStore, args: &Args, -) -> anyhow::Result> { +) -> anyhow::Result { let path = oidc_secret::path(&args.client_id)?; let secret = oidc_secret::read(keys, &path).await?; tracing::info!(%path, "loaded the OIDC client secret"); - let source = oidc_secret::Store::new( + Ok(oidc_secret::OidcSecret::from_store( + &args.client_id, + secret, swarm_secret_client::client::Settings::from_env()?, args.store_cert_role.clone(), path, - ); - Ok(oidc_secret::OidcSecret::new(source, secret)) + )) } /// The `ensure-callout-keys` verb: see [`xkey::ensure`] and [`nkey::ensure`]. @@ -436,7 +437,7 @@ async fn introspected( policy: &policy::Policy, http: &reqwest::Client, args: &Args, - client_secret: &oidc_secret::OidcSecret, + client_secret: &oidc_secret::OidcSecret, token: &str, ) -> (Option, Option) { let deadline = tokio::time::Instant::now() + introspect::INTROSPECTION_TIMEOUT; diff --git a/swarm-nats-auth/src/oidc_secret.rs b/swarm-nats-auth/src/oidc_secret.rs index ea158762..4dd4fa9a 100644 --- a/swarm-nats-auth/src/oidc_secret.rs +++ b/swarm-nats-auth/src/oidc_secret.rs @@ -4,24 +4,20 @@ //! //! The secret can change under a running responder: authelia's host //! re-publishes it when it is re-minted. So when the `IdP` refuses an -//! introspection (a non-2xx), the secret is read from the store again and, -//! when it changed, the introspection is tried once more. Both fit in the one -//! request's [`INTROSPECTION_TIMEOUT`](crate::introspect::INTROSPECTION_TIMEOUT). -//! A re-read that does not is left running: that request is denied, and the -//! next one introspects with what the re-read landed. One re-read runs at a -//! time; a refusal while one is running waits for it rather than starting -//! another. +//! introspection (a non-2xx), the secret is read from the store again +//! ([`StoredSecret::after_refusal`]) and, when it changed, the introspection +//! is tried once more. Both fit in the one request's +//! [`INTROSPECTION_TIMEOUT`](crate::introspect::INTROSPECTION_TIMEOUT). A +//! re-read that does not is left running: that request is denied, and the +//! next one introspects with what the re-read landed. use std::future::Future; -use std::sync::{Arc, Mutex, RwLock}; -use std::time::Duration; +use std::sync::Arc; use anyhow::{Context, Result}; -use futures_util::FutureExt; -use futures_util::future::{BoxFuture, Shared}; +use swarm_queue_client::stored_secret::StoredSecret; use swarm_secret_client::client::{DEFAULT_CERT_MOUNT, Settings}; use swarm_secret_client::path::{Kind, principal_prefix}; -use tokio::task::AbortHandle; use tokio::time::Instant; use crate::introspect::Answer; @@ -55,73 +51,41 @@ pub async fn read(store: &impl KeyStore, path: &str) -> Result { Ok(trimmed.to_owned()) } -/// Where a re-read of the secret comes from. -pub trait Source: Send + Sync + 'static { - /// The secret as the store holds it now. - fn read(&self) -> impl Future> + Send; +/// The secret in use, and how it is re-read. +pub struct OidcSecret { + secret: StoredSecret, } -/// The secret store, logged in to per read: a re-read can come long after -/// the start's token has expired. -pub struct Store { - settings: Settings, - cert_role: String, - path: String, -} - -impl Store { - /// The secret at `path`, read as `cert_role` from the store at `settings`. - pub fn new(settings: Settings, cert_role: String, path: String) -> Self { - Self { - settings, - cert_role, - path, - } - } -} - -impl Source for Store { - async fn read(&self) -> Result { - let store = swarm_secret_client::SecretStore::connect( - &self.settings, - &self.cert_role, - DEFAULT_CERT_MOUNT, - ) - .await - .context("logging in to the secret store")?; - read(&store, &self.path).await - } -} - -/// The longest one re-read runs. The store client sets no timeout of its -/// own, and while a re-read runs every refusal waits on it instead of -/// starting another. -const REREAD_TIMEOUT: Duration = Duration::from_secs(10); - -/// A re-read in progress, shared by every request waiting on it. It yields -/// whether the store was read; a failed or timed-out read logs for itself. -type Reread = Shared>; - -/// The secret in use, and the re-read that replaces it. No `Debug`: it holds -/// the secret. -pub struct OidcSecret { - source: Arc, - current: Arc>, - reread: Mutex>, -} - -impl OidcSecret { - /// `initial`, re-read from `source` when the `IdP` refuses it. - pub fn new(source: S, initial: String) -> Self { - Self { - source: Arc::new(source), - current: Arc::new(RwLock::new(initial)), - reread: Mutex::new(None), - } +impl OidcSecret { + /// `client_id`'s secret, `initial` until a re-read replaces it. A re-read + /// logs in to the store at `settings` as `cert_role` and reads `path`: it + /// can come long after the start's token has expired. + pub fn from_store( + client_id: &str, + initial: String, + settings: Settings, + cert_role: String, + path: String, + ) -> Self { + let source = Arc::new((settings, cert_role, path)); + Self::new(StoredSecret::new(client_id, initial, move || { + let source = Arc::clone(&source); + async move { + let (settings, cert_role, path) = &*source; + let store = swarm_secret_client::SecretStore::connect( + settings, + cert_role, + DEFAULT_CERT_MOUNT, + ) + .await + .context("logging in to the secret store")?; + Ok(read(&store, path).await?) + } + })) } - fn current(&self) -> String { - self.current.read().expect("unpoisoned").clone() + fn new(secret: StoredSecret) -> Self { + Self { secret } } /// The caller `attempt` names, introspecting with the current secret. @@ -136,82 +100,26 @@ impl OidcSecret { F: Fn(String) -> Fut, Fut: Future>, { - let used = self.current(); + let used = self.secret.current(); if let Answer::Caller(caller) = attempt(used.clone()).await? { return Ok(caller); } - match tokio::time::timeout_at(deadline, self.reread()).await { - Err(_) => { - tracing::warn!( - "re-reading the OIDC client secret did not finish inside the introspection budget; \ - denying, and the next request introspects with what it reads" - ); - return Ok(None); - } - Ok(false) => return Ok(None), - Ok(true) => {} - } - let fresh = self.current(); - if fresh == used { - tracing::warn!("the stored OIDC client secret is the one the IdP refused; denying"); + let Ok(fresh) = tokio::time::timeout_at(deadline, self.secret.after_refusal(&used)).await + else { + tracing::warn!( + "re-reading the OIDC client secret did not finish inside the introspection budget; \ + denying, and the next request introspects with what it reads" + ); return Ok(None); - } - tracing::info!("re-read a changed OIDC client secret; introspecting again"); + }; + let Some(fresh) = fresh else { + return Ok(None); + }; match attempt(fresh).await? { Answer::Caller(caller) => Ok(caller), Answer::Refused => Ok(None), } } - - /// The running re-read, or a new one when none is running. Spawned, so it - /// finishes even when every request waiting on it has given up. - fn reread(&self) -> Reread { - let mut slot = self.reread.lock().expect("unpoisoned"); - if let Some((handle, running)) = slot.as_ref() - && !handle.is_finished() - { - return running.clone(); - } - let source = Arc::clone(&self.source); - let current = Arc::clone(&self.current); - let task = tokio::spawn(async move { - match tokio::time::timeout(REREAD_TIMEOUT, source.read()).await { - Ok(Ok(secret)) => { - *current.write().expect("unpoisoned") = secret; - true - } - Ok(Err(e)) => { - tracing::warn!( - error = format!("{e:#}"), - "could not re-read the OIDC client secret; keeping the one in use" - ); - false - } - Err(_) => { - tracing::warn!( - timeout = ?REREAD_TIMEOUT, - "re-reading the OIDC client secret timed out; keeping the one in use" - ); - false - } - } - }); - let handle = task.abort_handle(); - let shared = task - .map(|read| { - read.unwrap_or_else(|e| { - tracing::warn!( - error = %e, - "the OIDC client secret re-read task died; keeping the one in use" - ); - false - }) - }) - .boxed() - .shared(); - *slot = Some((handle, shared.clone())); - shared - } } #[cfg(test)] @@ -223,29 +131,23 @@ mod tests { use crate::introspect::INTROSPECTION_TIMEOUT; use crate::keystore::tests::Fake as FakeKeys; - /// A store holding `secret`, or failing, that takes `delay` to answer. - struct Fake { - secret: Option<&'static str>, + /// `initial` in use, re-read from a store holding `stored` that takes + /// `delay` to answer; and the count of its reads. + fn secret( + initial: &str, + stored: &'static str, delay: Duration, - reads: Arc, - } - - impl Fake { - fn holding(secret: &'static str) -> Self { - Self { - secret: Some(secret), - delay: Duration::ZERO, - reads: Arc::default(), + ) -> (OidcSecret, Arc) { + let reads = Arc::new(AtomicUsize::new(0)); + let counted = Arc::clone(&reads); + let secret = StoredSecret::new("swarm-nats", initial.to_owned(), move || { + counted.fetch_add(1, Ordering::SeqCst); + async move { + tokio::time::sleep(delay).await; + Ok(stored.to_owned()) } - } - } - - impl Source for Fake { - async fn read(&self) -> Result { - self.reads.fetch_add(1, Ordering::SeqCst); - tokio::time::sleep(self.delay).await; - self.secret.map(str::to_owned).context("store unreachable") - } + }); + (OidcSecret::new(secret), reads) } /// An `IdP` that admits as `hive-alpha` only when introspected with @@ -289,9 +191,7 @@ mod tests { #[tokio::test] async fn an_answered_introspection_reads_nothing() { - let store = Fake::holding("new"); - let reads = Arc::clone(&store.reads); - let secret = OidcSecret::new(store, "old".to_owned()); + let (secret, reads) = secret("old", "new", Duration::ZERO); let calls = AtomicUsize::new(0); let caller = secret .introspect(budget(), idp("old", &calls)) @@ -307,9 +207,7 @@ mod tests { /// admitted with it. #[tokio::test] async fn a_refused_secret_is_re_read_and_the_request_retried_once() { - let store = Fake::holding("new"); - let reads = Arc::clone(&store.reads); - let secret = OidcSecret::new(store, "old".to_owned()); + let (secret, reads) = secret("old", "new", Duration::ZERO); let calls = AtomicUsize::new(0); let caller = secret .introspect(budget(), idp("new", &calls)) @@ -322,9 +220,7 @@ mod tests { #[tokio::test] async fn a_refusal_after_the_re_read_is_a_denial_with_no_third_attempt() { - let store = Fake::holding("newer"); - let reads = Arc::clone(&store.reads); - let secret = OidcSecret::new(store, "old".to_owned()); + let (secret, reads) = secret("old", "newer", Duration::ZERO); let calls = AtomicUsize::new(0); let caller = secret .introspect(budget(), idp("unknown", &calls)) @@ -337,7 +233,7 @@ mod tests { #[tokio::test] async fn an_unchanged_secret_is_not_retried() { - let secret = OidcSecret::new(Fake::holding("old"), "old".to_owned()); + let (secret, _) = secret("old", "old", Duration::ZERO); let calls = AtomicUsize::new(0); let caller = secret .introspect(budget(), idp("unknown", &calls)) @@ -347,99 +243,12 @@ mod tests { assert_eq!(calls.load(Ordering::SeqCst), 1); } - thread_local! { - static LOG: std::cell::RefCell> = const { std::cell::RefCell::new(Vec::new()) }; - } - - /// What `tracing` records on the calling thread. The subscriber is - /// process-wide, because a scoped one loses to the callsite cache when - /// another test reaches the same event first; the buffer is per thread, - /// which is where a current-thread test runtime runs the spawned re-read. - struct Logs; - - impl std::io::Write for Logs { - fn write(&mut self, buf: &[u8]) -> std::io::Result { - LOG.with(|log| log.borrow_mut().extend_from_slice(buf)); - Ok(buf.len()) - } - - fn flush(&mut self) -> std::io::Result<()> { - Ok(()) - } - } - - impl Logs { - /// Starts this thread's capture empty. - fn capture() { - static INSTALLED: std::sync::Once = std::sync::Once::new(); - INSTALLED.call_once(|| { - tracing_subscriber::fmt() - .with_writer(|| Logs) - .with_ansi(false) - .init(); - }); - LOG.with(|log| log.borrow_mut().clear()); - } - - fn text() -> String { - LOG.with(|log| String::from_utf8(log.borrow().clone()).expect("utf-8")) - } - } - - #[tokio::test] - async fn a_failed_re_read_denies_and_keeps_the_secret_in_use() { - Logs::capture(); - let store = Fake { - secret: None, - delay: Duration::ZERO, - reads: Arc::default(), - }; - let secret = OidcSecret::new(store, "old".to_owned()); - let calls = AtomicUsize::new(0); - let caller = secret - .introspect(budget(), idp("new", &calls)) - .await - .expect("answered"); - assert_eq!(caller, None); - assert_eq!(calls.load(Ordering::SeqCst), 1); - assert_eq!(secret.current(), "old"); - - let text = Logs::text(); - assert!(text.contains("could not re-read the OIDC client secret")); - assert!(!text.contains("is the one the IdP refused"), "{text}"); - assert!(!text.contains("old"), "no secret in the log: {text}"); - } - - /// The store was read and holds what the `IdP` already refused: the one - /// case that says so. - #[tokio::test] - async fn the_unchanged_secret_is_logged_as_the_one_the_idp_refused() { - Logs::capture(); - let secret = OidcSecret::new(Fake::holding("old"), "old".to_owned()); - let calls = AtomicUsize::new(0); - let caller = secret - .introspect(budget(), idp("new", &calls)) - .await - .expect("answered"); - assert_eq!(caller, None); - - let text = Logs::text(); - assert!(text.contains("is the one the IdP refused"), "{text}"); - assert!(!text.contains("could not re-read"), "{text}"); - } - /// A store slower than the budget: the request is denied at its /// deadline, the re-read finishes behind it, and the next request is /// admitted on its first attempt without reading again. #[tokio::test(start_paused = true)] async fn a_re_read_past_the_budget_serves_the_next_request() { - let store = Fake { - secret: Some("new"), - delay: Duration::from_secs(3), - reads: Arc::default(), - }; - let reads = Arc::clone(&store.reads); - let secret = OidcSecret::new(store, "old".to_owned()); + let (secret, reads) = secret("old", "new", Duration::from_secs(3)); let calls = AtomicUsize::new(0); let started = Instant::now(); @@ -462,64 +271,6 @@ mod tests { assert_eq!(reads.load(Ordering::SeqCst), 1); } - /// A refusal while a re-read is still running waits on that one, and is - /// admitted when it lands inside this request's budget. - #[tokio::test(start_paused = true)] - async fn a_refusal_during_a_running_re_read_joins_it() { - let store = Fake { - secret: Some("new"), - delay: Duration::from_secs(2), - reads: Arc::default(), - }; - let reads = Arc::clone(&store.reads); - let secret = OidcSecret::new(store, "old".to_owned()); - let calls = AtomicUsize::new(0); - - let first = secret - .introspect(budget(), idp("new", &calls)) - .await - .expect("answered"); - assert_eq!(first, None); - - let second = secret - .introspect(budget(), idp("new", &calls)) - .await - .expect("answered"); - assert_eq!(second.as_deref(), Some("hive-alpha")); - assert_eq!(calls.load(Ordering::SeqCst), 3); - assert_eq!(reads.load(Ordering::SeqCst), 1); - } - - /// A re-read that stalls is given up at [`REREAD_TIMEOUT`]: the secret in - /// use stays, and the next refusal starts a fresh re-read rather than - /// waiting on the stalled one. - #[tokio::test(start_paused = true)] - async fn a_stalled_re_read_is_dropped_at_its_timeout() { - let store = Fake { - secret: Some("new"), - delay: REREAD_TIMEOUT * 6, - reads: Arc::default(), - }; - let reads = Arc::clone(&store.reads); - let secret = OidcSecret::new(store, "old".to_owned()); - let calls = AtomicUsize::new(0); - - let first = secret - .introspect(budget(), idp("new", &calls)) - .await - .expect("answered"); - assert_eq!(first, None); - - tokio::time::sleep(REREAD_TIMEOUT).await; - assert_eq!(secret.current(), "old"); - let second = secret - .introspect(budget(), idp("new", &calls)) - .await - .expect("answered"); - assert_eq!(second, None); - assert_eq!(reads.load(Ordering::SeqCst), 2); - } - /// An `IdP` on a loopback port that admits as `hive-alpha` a request /// whose basic auth is `swarm-nats:{registered}`, and answers any other /// with a 401. @@ -558,9 +309,7 @@ mod tests { .no_proxy() .build() .expect("client"); - let store = Fake::holding("new"); - let reads = Arc::clone(&store.reads); - let secret = OidcSecret::new(store, "old".to_owned()); + let (secret, reads) = secret("old", "new", Duration::ZERO); let deadline = budget(); let caller = secret .introspect(deadline, |own| { diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index 5496599d..0fb0389f 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -55,6 +55,9 @@ agent-limits = ["async-nats/jetstream"] # Bare (no `kv`/`jetstream`) unless a consumer opts into the `kv` feature # above - the connect itself needs none of them. async-nats.workspace = true +# `Shared`, so every caller refused while a secret re-read runs waits on that +# one read (`stored_secret`). +futures-util.workspace = true # `blocking` on top of the workspace default (`form`/`json`/`rustls`) — # `mint_token_for_blocking` needs `reqwest::blocking::Client` for a caller # with no tokio reactor to `.await` an async request on (an OTLP exporter's @@ -76,5 +79,16 @@ tokio.workspace = true tracing.workspace = true utoipa = { workspace = true, optional = true } +[dev-dependencies] +# A loopback stand-in token endpoint for the client secret re-read tests. +axum.workspace = true +data-encoding.workspace = true +# `test-util` for `#[tokio::test(start_paused = true)]`: the re-read backoff +# tests run on paused time. +tokio = { workspace = true, features = ["test-util"] } +# Capturing what a re-read logs, to check it names the client and never the +# secret. +tracing-subscriber.workspace = true + [lints] workspace = true diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 6cd0da3a..7ba6f233 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -402,6 +402,8 @@ pub mod subagent_term; /// same shapes. pub mod review_classifier; +pub mod stored_secret; + /// Only the fields this needs; authelia returns several. #[derive(serde::Deserialize)] struct TokenResponse { @@ -489,28 +491,36 @@ pub struct QueueConfig { pub ca_file: Option, } -/// Where a [`QueueConfig`]'s client secret comes from. -#[derive(Clone)] +/// Where a [`QueueConfig`]'s client secret comes from. `Debug` never prints +/// a held secret. +#[derive(Clone, Debug)] pub enum ClientSecret { /// A file holding the plaintext, read at every token mint so a rotated /// file takes effect without a restart. File(PathBuf), - /// The plaintext itself, held only in this process's memory. - Value(String), -} - -/// Redacts [`ClientSecret::Value`]: a [`QueueConfig`] is `Debug`, and printing -/// one must not put the secret in a log. -impl std::fmt::Debug for ClientSecret { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::File(path) => f.debug_tuple("File").field(path).finish(), - Self::Value(_) => f.write_str("Value()"), - } - } + /// The plaintext, held only in this process's memory and read again + /// when the token endpoint refuses it. + Stored(stored_secret::StoredSecret), } impl ClientSecret { + /// The secret to read again after `error`: a held one the token endpoint + /// refused as a credential (400 or 401). A 429 or a 5xx says nothing + /// about the secret. + fn refused(&self, error: &Error) -> Option<&stored_secret::StoredSecret> { + match (self, error) { + (Self::Stored(stored), Error::TokenRefused { status, .. }) + if matches!( + *status, + reqwest::StatusCode::BAD_REQUEST | reqwest::StatusCode::UNAUTHORIZED + ) => + { + Some(stored) + } + _ => None, + } + } + async fn read(&self) -> Result { match self { Self::File(path) => { @@ -521,7 +531,7 @@ impl ClientSecret { source, }) } - Self::Value(value) => Ok(value.clone()), + Self::Stored(stored) => Ok(stored.current()), } } @@ -533,7 +543,7 @@ impl ClientSecret { source, }) } - Self::Value(value) => Ok(value.clone()), + Self::Stored(stored) => Ok(stored.current()), } } } @@ -545,6 +555,10 @@ impl ClientSecret { /// (a machine client receives an access token and never an id-token), so a /// caller passing `None` for `scope` requests none at all — see /// [`token_request`] for the one scope a caller does have to ask for. +/// +/// A [`ClientSecret::Stored`] the endpoint refuses is read again, and when +/// that lands a different secret the request is sent once more with it; see +/// [`stored_secret`] for how often a re-read may run. async fn mint_token( http: &reqwest::Client, cfg: &QueueConfig, @@ -554,7 +568,32 @@ async fn mint_token( // 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 = cfg.client_secret.read().await?; + let refused = match mint_with(http, cfg, &secret, audience, scope).await { + Err(e) => e, + minted => return minted, + }; + let Some(stored) = cfg.client_secret.refused(&refused) else { + return Err(refused); + }; + tracing::warn!( + client_id = %cfg.client_id, + error = %chain(&refused), + "the token endpoint refused this client's secret; re-reading it" + ); + match stored.after_refusal(&secret).await { + Some(fresh) => mint_with(http, cfg, &fresh, audience, scope).await, + None => Err(refused), + } +} +/// One token request presenting `secret`. +async fn mint_with( + http: &reqwest::Client, + cfg: &QueueConfig, + secret: &str, + audience: Option<&str>, + scope: Option<&str>, +) -> Result { let response = token_request(http, cfg, secret.trim(), audience, scope) .send() .await @@ -700,13 +739,46 @@ pub fn mint_token_for_blocking( audience: Option<&str>, scope: Option<&str>, ) -> Result { - let http = build_blocking_http_client(cfg)?; + mint_token_blocking(&build_blocking_http_client(cfg)?, cfg, audience, scope) +} +/// [`mint_token`] on the blocking client. +fn mint_token_blocking( + http: &reqwest::blocking::Client, + cfg: &QueueConfig, + audience: Option<&str>, + scope: Option<&str>, +) -> Result { // Read per call rather than caching — see `mint_token`'s identical // comment on the async path; the reasoning does not change with the // client type. let secret = cfg.client_secret.read_blocking()?; + let refused = match mint_with_blocking(http, cfg, &secret, audience, scope) { + Err(e) => e, + minted => return minted, + }; + let Some(stored) = cfg.client_secret.refused(&refused) else { + return Err(refused); + }; + tracing::warn!( + client_id = %cfg.client_id, + error = %chain(&refused), + "the token endpoint refused this client's secret; re-reading it" + ); + match stored.after_refusal_blocking(&secret) { + Some(fresh) => mint_with_blocking(http, cfg, &fresh, audience, scope), + None => Err(refused), + } +} +/// [`mint_with`] on the blocking client. +fn mint_with_blocking( + http: &reqwest::blocking::Client, + cfg: &QueueConfig, + secret: &str, + audience: Option<&str>, + scope: Option<&str>, +) -> Result { let mut form = vec![("grant_type", "client_credentials")]; if let Some(audience) = audience { form.push(("audience", audience)); @@ -947,6 +1019,9 @@ where #[cfg(test)] mod tests { + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + use super::*; /// The names the publishers, the relays and the queue's grant all spell. @@ -983,24 +1058,149 @@ mod tests { .with_no_client_auth(); } - #[test] - fn a_held_secret_is_read_back_as_itself() { - let secret = ClientSecret::Value("s3cret".to_owned()); - assert_eq!( - secret.read_blocking().expect("a value needs no I/O"), - "s3cret" - ); - } - - #[test] - fn a_debug_print_of_a_config_does_not_carry_a_held_secret() { + #[tokio::test] + async fn a_debug_print_of_a_config_does_not_carry_a_held_secret() { let cfg = QueueConfig { - client_secret: ClientSecret::Value("s3cret".to_owned()), + client_secret: ClientSecret::Stored( + stored_secret::tests::Fake::holding("new").secret("hive-alpha", "s3cret"), + ), ..token_cfg() }; let printed = format!("{cfg:?}"); assert!(!printed.contains("s3cret"), "{printed}"); - assert!(printed.contains(""), "{printed}"); + assert!(printed.contains("StoredSecret"), "{printed}"); + } + + /// A token endpoint on a loopback port that mints for + /// `hive-alpha:{registered}` and answers any other secret with a 401, + /// counting its requests. + async fn stub_idp(registered: &'static str, calls: Arc) -> String { + use axum::http::{HeaderMap, StatusCode, header::AUTHORIZATION}; + + let expected = format!( + "Basic {}", + data_encoding::BASE64.encode(format!("hive-alpha:{registered}").as_bytes()) + ); + let app = axum::Router::new().fallback(move |headers: HeaderMap| async move { + calls.fetch_add(1, Ordering::SeqCst); + if headers.get(AUTHORIZATION).and_then(|v| v.to_str().ok()) == Some(&*expected) { + ( + StatusCode::OK, + r#"{"access_token":"a-token","expires_in":3599}"#, + ) + } else { + (StatusCode::UNAUTHORIZED, r#"{"error":"invalid_client"}"#) + } + }); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let url = format!("http://{}/token", listener.local_addr().expect("addr")); + tokio::spawn(async move { axum::serve(listener, app).await }); + url + } + + /// A plain-HTTP loopback client: no proxy, and no platform trust store, + /// which the nix build sandbox does not have (see [`offline_client`]). + fn loopback_client() -> reqwest::Client { + reqwest::Client::builder() + .no_proxy() + .danger_accept_invalid_certs(true) + .build() + .expect("a client that verifies nothing needs no system trust store") + } + + /// The rotation the re-read exists for: the token endpoint refuses the + /// secret held since start, the store holds the new one, and the same + /// mint succeeds with it. + #[tokio::test] + async fn a_refused_secret_is_re_read_and_the_mint_retried_with_the_new_one() { + let calls = Arc::new(AtomicUsize::new(0)); + let store = stored_secret::tests::Fake::holding("secret-b"); + let cfg = QueueConfig { + token_endpoint: stub_idp("secret-b", Arc::clone(&calls)).await, + client_secret: ClientSecret::Stored(store.secret("hive-alpha", "secret-a")), + ..token_cfg() + }; + let minted = mint_token(&loopback_client(), &cfg, None, None) + .await + .expect("minted with the re-read secret"); + assert_eq!(minted.token, "a-token"); + assert_eq!(calls.load(Ordering::SeqCst), 2); + assert_eq!(store.reads(), 1); + } + + /// A secret the store still holds as refused: the mint fails, and the + /// mints after it inside the backoff ask the token endpoint once each + /// and read nothing. + #[tokio::test] + async fn a_permanently_refused_secret_is_not_re_read_on_every_mint() { + let calls = Arc::new(AtomicUsize::new(0)); + let store = stored_secret::tests::Fake::holding("secret-a"); + let cfg = QueueConfig { + token_endpoint: stub_idp("secret-b", Arc::clone(&calls)).await, + client_secret: ClientSecret::Stored(store.secret("hive-alpha", "secret-a")), + ..token_cfg() + }; + let http = loopback_client(); + for _ in 0..5 { + let err = mint_token(&http, &cfg, None, None) + .await + .err() + .expect("refused"); + assert!( + matches!(err, Error::TokenRefused { status, .. } if status == 401), + "{err}" + ); + } + assert_eq!(calls.load(Ordering::SeqCst), 5); + assert_eq!(store.reads(), 1); + } + + /// A refusal that is not about the credential reads nothing. + #[tokio::test] + async fn a_rate_limited_mint_does_not_re_read() { + let store = stored_secret::tests::Fake::holding("secret-b"); + let cfg = QueueConfig { + client_secret: ClientSecret::Stored(store.secret("hive-alpha", "secret-a")), + ..token_cfg() + }; + let limited = Error::TokenRefused { + status: reqwest::StatusCode::TOO_MANY_REQUESTS, + body: String::new(), + }; + assert!(cfg.client_secret.refused(&limited).is_none()); + let refused = Error::TokenRefused { + status: reqwest::StatusCode::UNAUTHORIZED, + body: String::new(), + }; + assert!(cfg.client_secret.refused(&refused).is_some()); + } + + /// The blocking mint re-reads the same way, from a thread with no + /// runtime of its own. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn the_blocking_mint_re_reads_a_refused_secret() { + let calls = Arc::new(AtomicUsize::new(0)); + let store = stored_secret::tests::Fake::holding("secret-b"); + let cfg = QueueConfig { + token_endpoint: stub_idp("secret-b", Arc::clone(&calls)).await, + client_secret: ClientSecret::Stored(store.secret("hive-alpha", "secret-a")), + ..token_cfg() + }; + let token = tokio::task::spawn_blocking(move || { + let http = reqwest::blocking::Client::builder() + .no_proxy() + .danger_accept_invalid_certs(true) + .build() + .expect("client"); + mint_token_blocking(&http, &cfg, None, None) + }) + .await + .expect("joined") + .expect("minted with the re-read secret"); + assert_eq!(token, "a-token"); + assert_eq!(store.reads(), 1); } fn token_cfg() -> QueueConfig { diff --git a/swarm-queue-client/src/stored_secret.rs b/swarm-queue-client/src/stored_secret.rs new file mode 100644 index 00000000..878b0716 --- /dev/null +++ b/swarm-queue-client/src/stored_secret.rs @@ -0,0 +1,459 @@ +//! A client secret held in memory and read again from where it is kept when +//! the `IdP` refuses it. +//! +//! The secret can change under a running process: authelia's host +//! re-publishes it when it is re-minted. A refusal asks for a re-read +//! ([`StoredSecret::after_refusal`]); when the read lands a different secret, +//! the caller tries once more with it. One re-read runs at a time, and a +//! refusal while one is running waits for it rather than starting another. +//! The re-read is spawned, so it finishes even when every caller waiting on +//! it has given up, and the next refusal finds what it landed. +//! +//! A re-read that fails, or finds the secret the `IdP` refused, delays the +//! next one: [`FIRST_REREAD_DELAY`], doubled per such re-read up to +//! [`MAX_REREAD_DELAY`]. A refusal inside that delay reads nothing, so a +//! secret that stays wrong costs the store one read per delay, not one per +//! refusal. + +use std::future::Future; +use std::pin::Pin; +use std::sync::{Arc, Mutex, RwLock}; +use std::time::Duration; + +use futures_util::FutureExt as _; +use futures_util::future::{BoxFuture, Shared}; +use tokio::task::AbortHandle; +use tokio::time::Instant; + +/// What a failed read of the secret reports. `anyhow::Error` converts into it. +pub type ReadError = Box; + +/// The longest one re-read runs. A store client may set no timeout of its +/// own, and while a re-read runs every refusal waits on it. +pub const REREAD_TIMEOUT: Duration = Duration::from_secs(10); + +/// The wait before the next re-read after one that brought no new secret. +pub const FIRST_REREAD_DELAY: Duration = Duration::from_secs(5); + +/// The longest wait between two re-reads. +pub const MAX_REREAD_DELAY: Duration = Duration::from_mins(5); + +type Read = + dyn Fn() -> Pin> + Send>> + Send + Sync; + +/// A re-read in progress, shared by every caller waiting on it. +type Reread = Shared>; + +/// The secret in use and how to read it again. Clones share one secret: a +/// re-read through any of them is seen by all. `Debug` names the client and +/// never the secret. +#[derive(Clone)] +pub struct StoredSecret { + inner: Arc, +} + +struct Inner { + client_id: String, + read: Box, + runtime: tokio::runtime::Handle, + current: RwLock, + rereads: Mutex, +} + +#[derive(Default)] +struct Rereads { + running: Option<(AbortHandle, Reread)>, + /// Re-reads in a row that brought no new secret. + fruitless: u32, + /// No re-read starts before this. + not_before: Option, +} + +impl std::fmt::Debug for StoredSecret { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("StoredSecret") + .field("client_id", &self.inner.client_id) + .finish_non_exhaustive() + } +} + +impl StoredSecret { + /// `client_id`'s secret, `initial` until `read` replaces it. `client_id` + /// names the client in logs. + /// + /// # Panics + /// Outside a tokio runtime: re-reads are spawned on the one this is made + /// in. + pub fn new(client_id: impl Into, initial: String, read: F) -> Self + where + F: Fn() -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, + { + Self { + inner: Arc::new(Inner { + client_id: client_id.into(), + read: Box::new(move || Box::pin(read())), + runtime: tokio::runtime::Handle::current(), + current: RwLock::new(initial), + rereads: Mutex::default(), + }), + } + } + + /// The secret in use. + #[must_use] + pub fn current(&self) -> String { + self.inner.current.read().expect("unpoisoned").clone() + } + + /// After the `IdP` refused `refused`: the secret to try once more, or + /// `None` when there is no other one to try. Waits for a re-read, at most + /// [`REREAD_TIMEOUT`]; a caller with a shorter budget bounds this itself. + pub async fn after_refusal(&self, refused: &str) -> Option { + let current = self.current(); + if current != refused { + return Some(current); + } + let Some(reread) = self.reread() else { + tracing::debug!( + client_id = %self.inner.client_id, + "the IdP refused this client's secret inside the re-read backoff; not re-reading" + ); + return None; + }; + reread.await; + let fresh = self.current(); + (fresh != refused).then_some(fresh) + } + + /// [`after_refusal`](Self::after_refusal), for a thread with no tokio + /// runtime of its own. + /// + /// # Panics + /// Inside a tokio runtime, as [`tokio::runtime::Handle::block_on`] does. + #[must_use] + pub fn after_refusal_blocking(&self, refused: &str) -> Option { + self.inner.runtime.block_on(self.after_refusal(refused)) + } + + /// The running re-read, or a new one when none is running and the + /// backoff allows it. + fn reread(&self) -> Option { + let mut rereads = self.inner.rereads.lock().expect("unpoisoned"); + if let Some((handle, running)) = rereads.running.as_ref() + && !handle.is_finished() + { + return Some(running.clone()); + } + if rereads.not_before.is_some_and(|at| Instant::now() < at) { + return None; + } + let inner = Arc::clone(&self.inner); + let task = self + .inner + .runtime + .spawn(async move { inner.reread().await }); + let handle = task.abort_handle(); + let client_id = self.inner.client_id.clone(); + let shared = task + .map(move |done| { + if let Err(e) = done { + tracing::warn!( + %client_id, + error = %e, + "the client secret re-read task died; keeping the one in use" + ); + } + }) + .boxed() + .shared(); + rereads.running = Some((handle, shared.clone())); + Some(shared) + } +} + +impl Inner { + /// One read of the secret, kept when it differs from the one in use. A + /// read that brings no new secret pushes the next one back. + async fn reread(&self) { + let before = self.current.read().expect("unpoisoned").clone(); + let outcome = tokio::time::timeout(REREAD_TIMEOUT, (self.read)()).await; + let mut rereads = self.rereads.lock().expect("unpoisoned"); + let failure = match outcome { + Ok(Ok(secret)) if secret != before => { + *self.current.write().expect("unpoisoned") = secret; + rereads.fruitless = 0; + rereads.not_before = None; + tracing::info!( + client_id = %self.client_id, + "re-read a changed client secret after the IdP refused the one in use" + ); + return; + } + Ok(Ok(_)) => "the stored client secret is the one the IdP refused".to_owned(), + Ok(Err(e)) => format!("could not re-read the client secret: {}", crate::chain(&*e)), + Err(_) => format!("re-reading the client secret timed out after {REREAD_TIMEOUT:?}"), + }; + rereads.fruitless = rereads.fruitless.saturating_add(1); + let delay = delay_after(rereads.fruitless); + rereads.not_before = Some(Instant::now() + delay); + tracing::warn!( + client_id = %self.client_id, + next_reread_in = ?delay, + "{failure}; keeping the one in use" + ); + } +} + +/// [`FIRST_REREAD_DELAY`] doubled per fruitless re-read after the first, +/// capped at [`MAX_REREAD_DELAY`]. +fn delay_after(fruitless: u32) -> Duration { + let doublings = fruitless.saturating_sub(1).min(16); + FIRST_REREAD_DELAY + .saturating_mul(1 << doublings) + .min(MAX_REREAD_DELAY) +} + +#[cfg(test)] +pub(crate) mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + + use super::*; + + /// A store holding `secret`, or failing, that takes `delay` to answer. + pub(crate) struct Fake { + pub(crate) secret: Arc>>, + pub(crate) delay: Duration, + pub(crate) reads: Arc, + } + + impl Fake { + pub(crate) fn holding(secret: &'static str) -> Self { + Self { + secret: Arc::new(Mutex::new(Some(secret))), + delay: Duration::ZERO, + reads: Arc::default(), + } + } + + /// `client_id`'s secret, `initial` in use, re-read from this store. + pub(crate) fn secret(&self, client_id: &str, initial: &str) -> StoredSecret { + let (secret, delay, reads) = ( + Arc::clone(&self.secret), + self.delay, + Arc::clone(&self.reads), + ); + StoredSecret::new(client_id, initial.to_owned(), move || { + let (secret, reads) = (Arc::clone(&secret), Arc::clone(&reads)); + async move { + reads.fetch_add(1, Ordering::SeqCst); + tokio::time::sleep(delay).await; + let held = *secret.lock().expect("unpoisoned"); + held.map(str::to_owned) + .ok_or_else(|| "store unreachable".into()) + } + }) + } + + pub(crate) fn reads(&self) -> usize { + self.reads.load(Ordering::SeqCst) + } + } + + thread_local! { + static LOG: std::cell::RefCell> = const { std::cell::RefCell::new(Vec::new()) }; + } + + /// What `tracing` records on the calling thread. The subscriber is + /// process-wide, because a scoped one loses to the callsite cache when + /// another test reaches the same event first; the buffer is per thread, + /// which is where a current-thread test runtime runs the spawned re-read. + pub(crate) struct Logs; + + impl std::io::Write for Logs { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + LOG.with(|log| log.borrow_mut().extend_from_slice(buf)); + Ok(buf.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl Logs { + /// Starts this thread's capture empty. + pub(crate) fn capture() { + static INSTALLED: std::sync::Once = std::sync::Once::new(); + INSTALLED.call_once(|| { + tracing_subscriber::fmt() + .with_writer(|| Logs) + .with_ansi(false) + .init(); + }); + LOG.with(|log| log.borrow_mut().clear()); + } + + pub(crate) fn text() -> String { + LOG.with(|log| String::from_utf8(log.borrow().clone()).expect("utf-8")) + } + } + + #[tokio::test] + async fn a_refused_secret_is_re_read_and_the_new_one_offered() { + Logs::capture(); + let store = Fake::holding("new"); + let secret = store.secret("hive-alpha", "old"); + assert_eq!(secret.after_refusal("old").await.as_deref(), Some("new")); + assert_eq!(secret.current(), "new"); + assert_eq!(store.reads(), 1); + + let text = Logs::text(); + assert!(text.contains("re-read a changed client secret"), "{text}"); + assert!(text.contains("hive-alpha"), "{text}"); + assert!(!text.contains("old") && !text.contains("new"), "{text}"); + } + + /// A refusal of a secret another caller already replaced offers the + /// replacement without reading again. + #[tokio::test] + async fn a_refusal_of_a_replaced_secret_offers_the_replacement() { + let store = Fake::holding("new"); + let secret = store.secret("hive-alpha", "old"); + assert_eq!(secret.after_refusal("old").await.as_deref(), Some("new")); + assert_eq!(secret.after_refusal("old").await.as_deref(), Some("new")); + assert_eq!(store.reads(), 1); + } + + /// A secret that stays wrong: the first refusal reads, refusals inside + /// the backoff read nothing, and the backoff doubles per fruitless read. + #[tokio::test(start_paused = true)] + async fn a_permanently_refused_secret_backs_off() { + Logs::capture(); + let store = Fake::holding("old"); + let secret = store.secret("hive-alpha", "old"); + + assert_eq!(secret.after_refusal("old").await, None); + assert_eq!(store.reads(), 1); + let text = Logs::text(); + assert!(text.contains("is the one the IdP refused"), "{text}"); + assert!(text.contains("hive-alpha"), "{text}"); + assert!(!text.contains("\"old\""), "{text}"); + + for _ in 0..10 { + assert_eq!(secret.after_refusal("old").await, None); + } + assert_eq!(store.reads(), 1); + + tokio::time::sleep(FIRST_REREAD_DELAY).await; + assert_eq!(secret.after_refusal("old").await, None); + assert_eq!(store.reads(), 2); + + tokio::time::sleep(FIRST_REREAD_DELAY).await; + assert_eq!(secret.after_refusal("old").await, None); + assert_eq!( + store.reads(), + 2, + "the second fruitless read doubles the wait" + ); + + tokio::time::sleep(FIRST_REREAD_DELAY).await; + assert_eq!(secret.after_refusal("old").await, None); + assert_eq!(store.reads(), 3); + } + + /// Once the store holds a new secret, the first refusal past the backoff + /// picks it up. + #[tokio::test(start_paused = true)] + async fn a_secret_published_during_the_backoff_is_read_after_it() { + let store = Fake::holding("old"); + let secret = store.secret("hive-alpha", "old"); + assert_eq!(secret.after_refusal("old").await, None); + + *store.secret.lock().expect("unpoisoned") = Some("new"); + assert_eq!(secret.after_refusal("old").await, None); + tokio::time::sleep(FIRST_REREAD_DELAY).await; + assert_eq!(secret.after_refusal("old").await.as_deref(), Some("new")); + assert_eq!(store.reads(), 2); + } + + #[tokio::test] + async fn a_failed_re_read_keeps_the_secret_in_use() { + Logs::capture(); + let store = Fake::holding("new"); + *store.secret.lock().expect("unpoisoned") = None; + let secret = store.secret("hive-alpha", "old"); + assert_eq!(secret.after_refusal("old").await, None); + assert_eq!(secret.current(), "old"); + + let text = Logs::text(); + assert!( + text.contains("could not re-read the client secret"), + "{text}" + ); + assert!(text.contains("store unreachable"), "{text}"); + assert!(!text.contains("is the one the IdP refused"), "{text}"); + assert!(!text.contains("old"), "no secret in the log: {text}"); + } + + /// A refusal while a re-read is still running waits on that one rather + /// than starting another. + #[tokio::test(start_paused = true)] + async fn a_refusal_during_a_running_re_read_joins_it() { + let mut store = Fake::holding("new"); + store.delay = Duration::from_secs(2); + let secret = store.secret("hive-alpha", "old"); + let (first, second) = + tokio::join!(secret.after_refusal("old"), secret.after_refusal("old")); + assert_eq!(first.as_deref(), Some("new")); + assert_eq!(second.as_deref(), Some("new")); + assert_eq!(store.reads(), 1); + } + + /// A caller that gives up leaves the re-read running; the next refusal + /// finds what it landed without reading again. + #[tokio::test(start_paused = true)] + async fn a_re_read_outlives_a_caller_that_gave_up() { + let mut store = Fake::holding("new"); + store.delay = Duration::from_secs(3); + let secret = store.secret("hive-alpha", "old"); + let gave_up = + tokio::time::timeout(Duration::from_secs(1), secret.after_refusal("old")).await; + assert!(gave_up.is_err()); + + tokio::time::sleep(Duration::from_secs(3)).await; + assert_eq!(secret.current(), "new"); + assert_eq!(secret.after_refusal("old").await.as_deref(), Some("new")); + assert_eq!(store.reads(), 1); + } + + /// A re-read that stalls is given up at [`REREAD_TIMEOUT`], keeping the + /// secret in use, and counts as fruitless. + #[tokio::test(start_paused = true)] + async fn a_stalled_re_read_is_dropped_at_its_timeout() { + let mut store = Fake::holding("new"); + store.delay = REREAD_TIMEOUT * 6; + let secret = store.secret("hive-alpha", "old"); + assert_eq!(secret.after_refusal("old").await, None); + assert_eq!(secret.current(), "old"); + + tokio::time::sleep(FIRST_REREAD_DELAY).await; + assert_eq!(secret.after_refusal("old").await, None); + assert_eq!(store.reads(), 2); + } + + #[test] + fn the_backoff_doubles_up_to_its_cap() { + assert_eq!(delay_after(1), FIRST_REREAD_DELAY); + assert_eq!(delay_after(2), FIRST_REREAD_DELAY * 2); + assert_eq!(delay_after(40), MAX_REREAD_DELAY); + } + + #[tokio::test] + async fn a_debug_print_does_not_carry_the_secret() { + let secret = Fake::holding("new").secret("hive-alpha", "s3cret"); + let printed = format!("{secret:?}"); + assert!(!printed.contains("s3cret"), "{printed}"); + assert!(printed.contains("hive-alpha"), "{printed}"); + } +} diff --git a/swarm-review-classifier/src/main.rs b/swarm-review-classifier/src/main.rs index 77d86af1..c322e28d 100644 --- a/swarm-review-classifier/src/main.rs +++ b/swarm-review-classifier/src/main.rs @@ -7,8 +7,8 @@ //! the controller reads the PR and sends it along. //! //! The store identity comes from the `BAO_*` environment. The queue client -//! secret is read once at start and the API key once per run, both from the -//! store under that identity. +//! secret is read at start and again when authelia refuses it, the API key +//! once per run, both from the store under that identity. use std::path::PathBuf; use std::time::Duration; @@ -16,6 +16,7 @@ use std::time::Duration; use anyhow::{Context, Result, bail}; use futures_util::StreamExt as _; use swarm_queue_client::review_classifier::{QUEUE_GROUP, Reply, SUBJECT, TryOut}; +use swarm_queue_client::stored_secret::StoredSecret; use swarm_queue_client::{ClientSecret, QueueConfig}; mod aggregate; @@ -59,11 +60,16 @@ async fn serve() -> Result<()> { .context("finding this binary for the claude trampoline")?, }; let secret = client_secret_with_retry(&client_id).await?; + let reread_id = client_id.clone(); + let secret = StoredSecret::new(client_id.clone(), secret, move || { + let client_id = reread_id.clone(); + async move { client_secret(&client_id).await.map_err(Into::into) } + }); let queue = swarm_queue_client::connect(QueueConfig { url: var("NATS_URL")?, token_endpoint: var("OIDC_TOKEN_ENDPOINT")?, client_id, - client_secret: ClientSecret::Value(secret), + client_secret: ClientSecret::Stored(secret), ca_file: std::env::var(format!("{PREFIX}_OIDC_CA_FILE")) .ok() .map(PathBuf::from),