diff --git a/Cargo.lock b/Cargo.lock index f524ee11..0509a77e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4954,9 +4954,6 @@ name = "swarm-queue-client" version = "0.1.0" dependencies = [ "async-nats", - "axum", - "data-encoding", - "futures-util", "reqwest", "rustls", "serde", @@ -4965,7 +4962,6 @@ dependencies = [ "thiserror 2.0.18", "tokio", "tracing", - "tracing-subscriber", "utoipa", ] diff --git a/docs/swarm/credentials.md b/docs/swarm/credentials.md index d3241665..cfee085e 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, 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/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/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 in memory and read again when authelia refuses it | +| `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/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 b8b6cecf..664ca87b 100644 --- a/hive-c0re/src/swarm_queue.rs +++ b/hive-c0re/src/swarm_queue.rs @@ -13,8 +13,7 @@ //! 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, held only in memory and read again when authelia refuses it -//! (`swarm_queue_client::stored_secret`). `swarm-secret-publish`, on the host +//! certificate and held only in memory. `swarm-secret-publish`, on the host //! that runs authelia, puts it there. //! //! Connects lazily on first use rather than at boot — nothing here @@ -28,7 +27,6 @@ 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; @@ -231,14 +229,11 @@ 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::Stored(stored), + client_secret: ClientSecret::Value(secret), ca_file: self.ca_file, } } @@ -332,14 +327,14 @@ mod tests { ); } - #[tokio::test] - async fn a_full_environment_carries_the_stored_secret_in_memory() { + #[test] + fn a_full_environment_carries_the_stored_secret_as_a_value() { 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::Stored(s) if s.current() == "s3cret")); + assert!(matches!(&cfg.client_secret, ClientSecret::Value(v) if v == "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 a3e38bd9..8fd470e0 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. 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. +//! 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. //! //! [`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 3246493a..840c71c4 100644 --- a/swarm-controller/src/queue_identity.rs +++ b/swarm-controller/src/queue_identity.rs @@ -3,10 +3,9 @@ //! //! 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 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 +//! `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 //! authelia, puts it there. //! //! [`init`] runs once in `main`, before anything calls [`get`]. @@ -17,7 +16,6 @@ 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}; @@ -42,14 +40,11 @@ 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::Stored(stored), + client_secret: ClientSecret::Value(secret), ca_file: self.ca_file, } } @@ -206,14 +201,14 @@ mod tests { assert_eq!(c.ca_file, None); } - #[tokio::test] - async fn the_fetched_secret_is_held_in_memory_not_a_path() { + #[test] + fn the_fetched_secret_is_held_as_a_value_not_a_path() { let cfg = coordinates_from(env(&FULL)) .expect("complete") .expect("configured") .with_secret("s3cret".to_owned()); assert!( - matches!(&cfg.client_secret, ClientSecret::Stored(s) if s.current() == "s3cret"), + matches!(&cfg.client_secret, ClientSecret::Value(v) if v == "s3cret"), "{:?}", cfg.client_secret ); diff --git a/swarm-nats-auth/src/introspect.rs b/swarm-nats-auth/src/introspect.rs index bd2b7d85..dfce9c0f 100644 --- a/swarm-nats-auth/src/introspect.rs +++ b/swarm-nats-auth/src/introspect.rs @@ -47,11 +47,10 @@ pub const INTROSPECTION_TIMEOUT: Duration = Duration::from_millis(1500); pub enum Answer { /// A 2xx: the caller, or `None` for every shape that is not an admission. Caller(Option), - /// A non-2xx, with its status: the `IdP` did not answer about the token - /// at all. A 400 or 401 is how it refuses this responder's own - /// credential, which `oidc_secret` re-reads and retries once before this - /// denies. - Refused(reqwest::StatusCode), + /// A non-2xx: the `IdP` did not answer about the token at all, which is + /// also how it refuses this responder's own credential. `oidc_secret` + /// re-reads that credential and retries once before this denies. + Refused, } /// The two fields this responder acts on — one per decision. Authelia returns @@ -93,7 +92,7 @@ impl IntrospectionResponse { /// Returns `Ok(Answer::Caller(Some(client_id)))` **only** on a 2xx whose /// body says `active: true` and names the client. Every other 2xx is /// `Ok(Answer::Caller(None))` with the reason logged, a non-2xx is -/// `Ok(Answer::Refused(_))`, and `Err` is a call that could not be made at all +/// `Ok(Answer::Refused)`, and `Err` is a call that could not be made at all /// or did not finish by `deadline` — callers must treat all but the first as /// a denial. /// @@ -124,7 +123,7 @@ pub async fn identify_caller( let status = resp.status(); if !status.is_success() { tracing::warn!(%status, "introspection returned non-2xx"); - return Ok(Answer::Refused(status)); + return Ok(Answer::Refused); } let body: IntrospectionResponse = match resp.json().await { Ok(b) => b, diff --git a/swarm-nats-auth/src/main.rs b/swarm-nats-auth/src/main.rs index 31c2b79e..49bb645b 100644 --- a/swarm-nats-auth/src/main.rs +++ b/swarm-nats-auth/src/main.rs @@ -385,17 +385,16 @@ 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"); - Ok(oidc_secret::OidcSecret::from_store( - &args.client_id, - secret, + let source = oidc_secret::Store::new( 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`]. @@ -437,7 +436,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 2db27083..ea158762 100644 --- a/swarm-nats-auth/src/oidc_secret.rs +++ b/swarm-nats-auth/src/oidc_secret.rs @@ -4,21 +4,24 @@ //! //! 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 as a credential (a 400 or 401), 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. Any other non-2xx is -//! denied without a re-read. +//! 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. use std::future::Future; -use std::sync::Arc; +use std::sync::{Arc, Mutex, RwLock}; +use std::time::Duration; use anyhow::{Context, Result}; -use swarm_queue_client::stored_secret::{StoredSecret, is_credential_refusal}; +use futures_util::FutureExt; +use futures_util::future::{BoxFuture, Shared}; 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; @@ -52,48 +55,79 @@ pub async fn read(store: &impl KeyStore, path: &str) -> Result { Ok(trimmed.to_owned()) } -/// The secret in use, and how it is re-read. -pub struct OidcSecret { - secret: StoredSecret, +/// 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; } -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?) - } - })) +/// 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), + } } - fn new(secret: StoredSecret) -> Self { - Self { secret } + fn current(&self) -> String { + self.current.read().expect("unpoisoned").clone() } /// The caller `attempt` names, introspecting with the current secret. - /// When the `IdP` refuses that as a credential (a 400 or 401), the secret - /// is re-read and, if it changed and `deadline` has not passed, `attempt` - /// runs once more with the new one. Any other refusal is a denial. - /// `attempt` bounds itself by the same `deadline`. + /// When the `IdP` refuses that, the secret is re-read and, if it changed + /// and `deadline` has not passed, `attempt` runs once more with the new + /// one. `attempt` bounds itself by the same `deadline`. /// /// # Errors /// When an attempt could not be made at all; the caller denies. @@ -102,28 +136,82 @@ impl OidcSecret { F: Fn(String) -> Fut, Fut: Future>, { - let used = self.secret.current(); - match attempt(used.clone()).await? { - Answer::Caller(caller) => return Ok(caller), - Answer::Refused(status) if !is_credential_refusal(status) => return Ok(None), - Answer::Refused(_) => {} + let used = self.current(); + if let Answer::Caller(caller) = attempt(used.clone()).await? { + return Ok(caller); } - 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" - ); + 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"); return Ok(None); - }; - let Some(fresh) = fresh else { - return Ok(None); - }; + } + tracing::info!("re-read a changed OIDC client secret; introspecting again"); match attempt(fresh).await? { Answer::Caller(caller) => Ok(caller), - Answer::Refused(_) => Ok(None), + 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)] @@ -135,23 +223,29 @@ mod tests { use crate::introspect::INTROSPECTION_TIMEOUT; use crate::keystore::tests::Fake as FakeKeys; - /// `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, + /// A store holding `secret`, or failing, that takes `delay` to answer. + struct Fake { + secret: Option<&'static str>, delay: Duration, - ) -> (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()) + reads: Arc, + } + + impl Fake { + fn holding(secret: &'static str) -> Self { + Self { + secret: Some(secret), + delay: Duration::ZERO, + reads: Arc::default(), } - }); - (OidcSecret::new(secret), reads) + } + } + + 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") + } } /// An `IdP` that admits as `hive-alpha` only when introspected with @@ -165,7 +259,7 @@ mod tests { std::future::ready(Ok(if secret == registered { Answer::Caller(Some("hive-alpha".to_owned())) } else { - Answer::Refused(reqwest::StatusCode::UNAUTHORIZED) + Answer::Refused })) } } @@ -195,7 +289,9 @@ mod tests { #[tokio::test] async fn an_answered_introspection_reads_nothing() { - let (secret, reads) = secret("old", "new", Duration::ZERO); + let store = Fake::holding("new"); + let reads = Arc::clone(&store.reads); + let secret = OidcSecret::new(store, "old".to_owned()); let calls = AtomicUsize::new(0); let caller = secret .introspect(budget(), idp("old", &calls)) @@ -211,7 +307,9 @@ mod tests { /// admitted with it. #[tokio::test] async fn a_refused_secret_is_re_read_and_the_request_retried_once() { - let (secret, reads) = secret("old", "new", Duration::ZERO); + let store = Fake::holding("new"); + let reads = Arc::clone(&store.reads); + let secret = OidcSecret::new(store, "old".to_owned()); let calls = AtomicUsize::new(0); let caller = secret .introspect(budget(), idp("new", &calls)) @@ -224,7 +322,9 @@ mod tests { #[tokio::test] async fn a_refusal_after_the_re_read_is_a_denial_with_no_third_attempt() { - let (secret, reads) = secret("old", "newer", Duration::ZERO); + let store = Fake::holding("newer"); + let reads = Arc::clone(&store.reads); + let secret = OidcSecret::new(store, "old".to_owned()); let calls = AtomicUsize::new(0); let caller = secret .introspect(budget(), idp("unknown", &calls)) @@ -237,7 +337,7 @@ mod tests { #[tokio::test] async fn an_unchanged_secret_is_not_retried() { - let (secret, _) = secret("old", "old", Duration::ZERO); + let secret = OidcSecret::new(Fake::holding("old"), "old".to_owned()); let calls = AtomicUsize::new(0); let caller = secret .introspect(budget(), idp("unknown", &calls)) @@ -247,39 +347,85 @@ mod tests { assert_eq!(calls.load(Ordering::SeqCst), 1); } - /// An `IdP` outage answers 503s and 429s with the new secret already in - /// the store: none of them reads it, so none pushes the re-read back, and - /// the first 401 once the `IdP` is back re-reads and is admitted. - #[tokio::test] - async fn only_a_credential_refusal_re_reads_the_secret() { - use reqwest::StatusCode; + thread_local! { + static LOG: std::cell::RefCell> = const { std::cell::RefCell::new(Vec::new()) }; + } - let (secret, reads) = secret("old", "new", Duration::ZERO); - let calls = AtomicUsize::new(0); - for status in [ - StatusCode::SERVICE_UNAVAILABLE, - StatusCode::BAD_GATEWAY, - StatusCode::TOO_MANY_REQUESTS, - ] { - let caller = secret - .introspect(budget(), |_| { - calls.fetch_add(1, Ordering::SeqCst); - std::future::ready(Ok(Answer::Refused(status))) - }) - .await - .expect("answered"); - assert_eq!(caller, None, "{status}"); + /// 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()) } - assert_eq!(calls.load(Ordering::SeqCst), 3); - assert_eq!(reads.load(Ordering::SeqCst), 0); + 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.as_deref(), Some("hive-alpha")); - assert_eq!(calls.load(Ordering::SeqCst), 5); - assert_eq!(reads.load(Ordering::SeqCst), 1); + 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 @@ -287,7 +433,13 @@ mod tests { /// 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 (secret, reads) = secret("old", "new", Duration::from_secs(3)); + 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 calls = AtomicUsize::new(0); let started = Instant::now(); @@ -310,6 +462,64 @@ 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. @@ -348,7 +558,9 @@ mod tests { .no_proxy() .build() .expect("client"); - let (secret, reads) = secret("old", "new", Duration::ZERO); + let store = Fake::holding("new"); + let reads = Arc::clone(&store.reads); + let secret = OidcSecret::new(store, "old".to_owned()); let deadline = budget(); let caller = secret .introspect(deadline, |own| { diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index 0fb0389f..5496599d 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -55,9 +55,6 @@ 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 @@ -79,16 +76,5 @@ 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/README.md b/swarm-queue-client/README.md index 0d4708b1..15cc1b69 100644 --- a/swarm-queue-client/README.md +++ b/swarm-queue-client/README.md @@ -44,10 +44,9 @@ The client secret is never in the environment, which would publish it to anything that can read `/proc//environ`. A caller passes it either as a **path** (`ClientSecret::File`), read per token request rather than cached, so a rotation the operator believes took effect actually did, or as a value it -fetched itself (`ClientSecret::Stored`), held in memory and fetched again -through the caller's callback when the token endpoint refuses it as a -credential. `swarm-controller`, `hive-c0re` and `swarm-review-classifier` do the -latter with the secret they read from the swarm's secret store. +fetched itself (`ClientSecret::Value`), held in memory. `swarm-controller` and +`hive-c0re` both do the latter with the secret they read from the swarm's +secret store. ## Token request shape: HTTP Basic, `audience`, `scope` diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 849c85b3..6cd0da3a 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -402,8 +402,6 @@ 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 { @@ -491,32 +489,28 @@ pub struct QueueConfig { pub ca_file: Option, } -/// Where a [`QueueConfig`]'s client secret comes from. `Debug` never prints -/// a held secret. -#[derive(Clone, Debug)] +/// Where a [`QueueConfig`]'s client secret comes from. +#[derive(Clone)] 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, held only in this process's memory and read again - /// when the token endpoint refuses it. - Stored(stored_secret::StoredSecret), + /// 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()"), + } + } } impl ClientSecret { - /// The secret to read again after `error`: a held one the token endpoint - /// refused as a credential ([`stored_secret::is_credential_refusal`]). - fn refused(&self, error: &Error) -> Option<&stored_secret::StoredSecret> { - match (self, error) { - (Self::Stored(stored), Error::TokenRefused { status, .. }) - if stored_secret::is_credential_refusal(*status) => - { - Some(stored) - } - _ => None, - } - } - async fn read(&self) -> Result { match self { Self::File(path) => { @@ -527,7 +521,7 @@ impl ClientSecret { source, }) } - Self::Stored(stored) => Ok(stored.current()), + Self::Value(value) => Ok(value.clone()), } } @@ -539,7 +533,7 @@ impl ClientSecret { source, }) } - Self::Stored(stored) => Ok(stored.current()), + Self::Value(value) => Ok(value.clone()), } } } @@ -551,10 +545,6 @@ 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, @@ -564,32 +554,7 @@ 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 @@ -735,46 +700,13 @@ pub fn mint_token_for_blocking( audience: Option<&str>, scope: Option<&str>, ) -> Result { - mint_token_blocking(&build_blocking_http_client(cfg)?, cfg, audience, scope) -} + let http = build_blocking_http_client(cfg)?; -/// [`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)); @@ -1015,9 +947,6 @@ 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. @@ -1054,164 +983,24 @@ mod tests { .with_no_client_auth(); } - #[tokio::test] - async fn a_debug_print_of_a_config_does_not_carry_a_held_secret() { + #[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() { let cfg = QueueConfig { - client_secret: ClientSecret::Stored( - stored_secret::tests::Fake::holding("new").secret("hive-alpha", "s3cret"), - ), + client_secret: ClientSecret::Value("s3cret".to_owned()), ..token_cfg() }; let printed = format!("{cfg:?}"); assert!(!printed.contains("s3cret"), "{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_or_failing_mint_does_not_re_read() { - use axum::http::StatusCode; - - let http = loopback_client(); - for status in [ - StatusCode::TOO_MANY_REQUESTS, - StatusCode::SERVICE_UNAVAILABLE, - ] { - let app = axum::Router::new().fallback(move || async move { (status, "") }); - 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 }); - - let store = stored_secret::tests::Fake::holding("secret-b"); - let cfg = QueueConfig { - token_endpoint: url, - client_secret: ClientSecret::Stored(store.secret("hive-alpha", "secret-a")), - ..token_cfg() - }; - let err = mint_token(&http, &cfg, None, None) - .await - .err() - .expect("refused"); - assert!( - matches!(err, Error::TokenRefused { status: s, .. } if s == status.as_u16()), - "{err}" - ); - assert_eq!(store.reads(), 0, "{status}"); - } - } - - /// 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); + assert!(printed.contains(""), "{printed}"); } fn token_cfg() -> QueueConfig { diff --git a/swarm-queue-client/src/stored_secret.rs b/swarm-queue-client/src/stored_secret.rs deleted file mode 100644 index 9e2320e4..00000000 --- a/swarm-queue-client/src/stored_secret.rs +++ /dev/null @@ -1,470 +0,0 @@ -//! 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" - ); - } -} - -/// Whether an `IdP` answering `status` refused the client's credential (400 or -/// 401), the one answer a re-read can fix. A 429 or a 5xx says nothing about -/// the secret, and re-reading on one would spend the backoff on an outage. -#[must_use] -pub fn is_credential_refusal(status: reqwest::StatusCode) -> bool { - matches!( - status, - reqwest::StatusCode::BAD_REQUEST | reqwest::StatusCode::UNAUTHORIZED - ) -} - -/// [`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 c322e28d..77d86af1 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 at start and again when authelia refuses it, the API key -//! once per run, both from the store under that identity. +//! secret is read once at start and the API key once per run, both from the +//! store under that identity. use std::path::PathBuf; use std::time::Duration; @@ -16,7 +16,6 @@ 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; @@ -60,16 +59,11 @@ 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::Stored(secret), + client_secret: ClientSecret::Value(secret), ca_file: std::env::var(format!("{PREFIX}_OIDC_CA_FILE")) .ok() .map(PathBuf::from),