diff --git a/docs/swarm/credentials.md b/docs/swarm/credentials.md index 46a2af60..419f61d5 100644 --- a/docs/swarm/credentials.md +++ b/docs/swarm/credentials.md @@ -76,7 +76,7 @@ of the cell says how. | `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 CI runner's, Grafana's, the log store's, the metrics store's, the forge's and the homeserver's journal forwarders' under `swarm-ci-runner`'s, `swarm-grafana-oidc`'s, `swarm-logs-auth`'s, `swarm-metrics-auth`'s, `swarm-forge-publisher`'s and `swarm-matrix-ctl`'s, authelia's journal forwarder's under `swarm-secret-publisher`'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 under the host's `/run` that 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//metrics/bearer` | `swarm-bao-grafana-metrics-bearer` on Grafana's host, under Grafana's certificate, when the store holds none | `swarm-metrics-auth` on the metrics store's host, under its own certificate, at most once a minute, to check the bearer Grafana's datasource presents at the store's read API, and `swarm-logs-grafana-auth` on the log store's host, under `swarm-logs-auth`'s certificate, to check the same bearer at the log store's query API | ❌ nothing re-mints it; an operator deletes it and restarts `swarm-bao-grafana-metrics-bearer` | ✅ `swarm-metrics-auth` reads it again once its copy is a minute old. ❌ Grafana reads its file copy only as it starts; the bearer unit restarts Grafana when the copy changes | +| `swarm/services//metrics/bearer` | `swarm-bao-grafana-metrics-bearer` on Grafana's host, under Grafana's certificate, when the store holds none | `swarm-metrics-auth` on the metrics store's host, under its own certificate, at most once a minute (every three seconds while the store fails), to check the bearer Grafana's datasource presents at the store's read API, and `swarm-logs-grafana-auth` on the log store's host, under `swarm-logs-auth`'s certificate, to check the same bearer at the log store's query API | ❌ nothing re-mints it; an operator deletes it and restarts `swarm-bao-grafana-metrics-bearer` | ✅ `swarm-metrics-auth` reads it again once its copy is a minute old. ❌ Grafana reads its file copy only as it starts; the bearer unit restarts Grafana when the copy changes | | `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 API key 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 | | `swarm/services/review-classifier/acp-provider` | 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 ACP run; held for that run and handed to `opencode`'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 | | `swarm/services/review-classifier/forge/token` | `swarm-controller`, with the forge's admin API, for its `review-classifier` account, at start and in a five-minute pass; the secret publisher's policy denies it | `swarm-controller` under its own certificate, before it decides whether to mint, and on every review it posts as the account; the classifier service's policy doesn't read it | ✅ the controller's pass re-mints when the stored token is missing or no longer matches the forge (last eight characters and scopes) | ✅ the controller reads the store on every review it posts | diff --git a/docs/swarm/services.md b/docs/swarm/services.md index 2b40d099..f239e148 100644 --- a/docs/swarm/services.md +++ b/docs/swarm/services.md @@ -238,7 +238,8 @@ Where the secret store runs on the same host, these default to the leaf it mints. On any other host you place a leaf whose CN is `deploy.bao.metricsAuthCommonName`, and evaluation refuses the config until you set both. `swarm-metrics-auth` reads the stored bearer at most once a -minute, so a replaced bearer starts working up to a minute later. +minute, so a replaced bearer starts working up to a minute later. After a +failed or timed-out read it denies for three seconds without trying again. `swarm-metrics-agent-auth` is the log store's `swarm-logs-auth` binary, with the same grant TTL and rate limit, under `deploy.victoriametrics.agentAuth*` (table above). diff --git a/swarm-metrics-auth/Cargo.toml b/swarm-metrics-auth/Cargo.toml index 6d90939f..33891503 100644 --- a/swarm-metrics-auth/Cargo.toml +++ b/swarm-metrics-auth/Cargo.toml @@ -23,3 +23,8 @@ tracing-subscriber.workspace = true [lints] workspace = true + +# `test-util` for `#[tokio::test(start_paused = true)]`: the lookup-timeout +# test waits out `LOOKUP_TIMEOUT` on paused time. +[dev-dependencies] +tokio = { workspace = true, features = ["test-util"] } diff --git a/swarm-metrics-auth/src/main.rs b/swarm-metrics-auth/src/main.rs index 41076d78..be478d21 100644 --- a/swarm-metrics-auth/src/main.rs +++ b/swarm-metrics-auth/src/main.rs @@ -12,8 +12,10 @@ //! //! The stored value is kept for `--cache-ttl` seconds, and no request reaches //! the store within that window: a replaced bearer is refused, and the one it -//! replaced still accepted, for up to that long. Lookups are serialised, so a -//! burst of requests against an expired copy costs one. +//! replaced still accepted, for up to that long. Lookups are serialised, and a +//! failed or timed-out one is remembered for `LOOKUP_FAILURE_WINDOW`, so a +//! burst of requests against an expired copy costs one lookup, success or +//! failure. use std::future::Future; use std::os::unix::fs::PermissionsExt as _; @@ -33,6 +35,13 @@ use swarm_secret_client::client::{DEFAULT_CERT_MOUNT, Settings}; /// rather than a gateway timeout. const LOOKUP_TIMEOUT: Duration = Duration::from_secs(5); +/// How long requests are denied without a lookup after one failed or timed +/// out. Each request queued behind a failing lookup would otherwise run its +/// own, and the k-th would wait k × `LOOKUP_TIMEOUT`, past nginx's 60s read +/// timeout. Short against `--cache-ttl`, so a recovered store is used again +/// within seconds. +const LOOKUP_FAILURE_WINDOW: Duration = Duration::from_secs(3); + #[derive(Parser)] #[command(about = "auth_request responder: checks Grafana's metrics bearer against the store")] struct Args { @@ -47,7 +56,8 @@ struct Args { #[arg(long)] store_path: String, /// Seconds the stored bearer is used without reading the store again; - /// 0 reads it on every request. + /// 0 reads it on every request, except for a few seconds after a read + /// failed. #[arg(long, default_value_t = 60)] cache_ttl: u64, } @@ -97,10 +107,19 @@ struct Cached { read_at: Instant, } +#[derive(Default)] +struct Cache { + stored: Option, + /// Until when requests are denied without a lookup, after one failed or + /// timed out. Stamped when that lookup ends: the requests queued behind + /// it carry the time they arrived, which is earlier. + failed_until: Option, +} + struct Responder { source: S, ttl: Duration, - cached: tokio::sync::Mutex>, + cached: tokio::sync::Mutex, } /// The answer for one subrequest's `Authorization` header, at `now`. @@ -115,17 +134,23 @@ async fn decide( }; // Held across the lookup, so concurrent requests wait for one read // rather than each starting their own. - let mut cached = responder.cached.lock().await; - let fresh = cached + let mut cache = responder.cached.lock().await; + let fresh = cache + .stored .as_ref() .is_some_and(|c| now.saturating_duration_since(c.read_at) < responder.ttl); if !fresh { + if cache.failed_until.is_some_and(|until| now < until) { + tracing::warn!("bearer lookup failed a moment ago; denying without another"); + return StatusCode::UNAUTHORIZED; + } match tokio::time::timeout(LOOKUP_TIMEOUT, responder.source.lookup()).await { Ok(Ok(Some(value))) if !value.is_empty() => { - *cached = Some(Cached { + cache.stored = Some(Cached { value, read_at: now, }); + cache.failed_until = None; } Ok(Ok(_)) => { tracing::warn!("no bearer is stored; denying"); @@ -133,15 +158,17 @@ async fn decide( } Ok(Err(e)) => { tracing::warn!(error = format!("{e:#}"), "bearer lookup failed; denying"); + cache.failed_until = Some(Instant::now() + LOOKUP_FAILURE_WINDOW); return StatusCode::UNAUTHORIZED; } Err(_) => { tracing::warn!("bearer lookup timed out; denying"); + cache.failed_until = Some(Instant::now() + LOOKUP_FAILURE_WINDOW); return StatusCode::UNAUTHORIZED; } } } - let Some(stored) = cached.as_ref() else { + let Some(stored) = cache.stored.as_ref() else { return StatusCode::UNAUTHORIZED; }; // Constant time, so the time to refuse says nothing about how much of the @@ -183,7 +210,7 @@ async fn main() -> anyhow::Result<()> { path: args.store_path, }, ttl: Duration::from_secs(args.cache_ttl), - cached: tokio::sync::Mutex::new(None), + cached: tokio::sync::Mutex::new(Cache::default()), }); // `RuntimeDirectoryPreserve=yes` can leave the previous run's socket; the @@ -212,10 +239,11 @@ mod tests { use std::sync::Mutex; use std::sync::atomic::{AtomicUsize, Ordering}; - /// A store answering `stored` (or failing when it is `Err`), counting - /// lookups. + /// A store answering `stored` (or failing when it is `Err`) after `delay`, + /// counting lookups. struct Fake { stored: Mutex, ()>>, + delay: Duration, lookups: AtomicUsize, } @@ -223,10 +251,18 @@ mod tests { fn holding(value: &str) -> Self { Self { stored: Mutex::new(Ok(Some(value.to_owned()))), + delay: Duration::ZERO, lookups: AtomicUsize::new(0), } } + fn hanging() -> Self { + Self { + delay: Duration::from_hours(1), + ..Self::holding(SECRET) + } + } + fn set(&self, stored: Result, ()>) { *self.stored.lock().unwrap() = stored.map(|v| v.map(str::to_owned)); } @@ -239,6 +275,7 @@ mod tests { impl BearerSource for &Fake { async fn lookup(&self) -> anyhow::Result> { self.lookups.fetch_add(1, Ordering::SeqCst); + tokio::time::sleep(self.delay).await; self.stored .lock() .unwrap() @@ -247,6 +284,12 @@ mod tests { } } + impl BearerSource for Arc { + async fn lookup(&self) -> anyhow::Result> { + (&**self).lookup().await + } + } + const TTL: Duration = Duration::from_mins(1); const SECRET: &str = "0f1e2d3c4b5a"; const GOOD: &str = "Bearer 0f1e2d3c4b5a"; @@ -255,7 +298,7 @@ mod tests { Responder { source: store, ttl: TTL, - cached: tokio::sync::Mutex::new(None), + cached: tokio::sync::Mutex::new(Cache::default()), } } @@ -338,17 +381,51 @@ mod tests { assert_eq!(store.lookups(), 2); } - /// A failed read leaves nothing usable cached, so the next request reads - /// the store again rather than waiting out a TTL. + /// A failed read is remembered for `LOOKUP_FAILURE_WINDOW`, not for a TTL: + /// requests inside it are denied without a lookup, the first one after it + /// reads the store again. #[tokio::test] - async fn a_failed_lookup_is_retried_on_the_next_request() { + async fn a_failed_lookup_is_retried_after_the_failure_window() { let store = Fake::holding(SECRET); store.set(Err(())); let r = responder(&store); let t0 = Instant::now(); assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::UNAUTHORIZED); store.set(Ok(Some(SECRET))); - assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::OK); + assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::UNAUTHORIZED); + assert_eq!(store.lookups(), 1); + let after = t0 + LOOKUP_FAILURE_WINDOW + Duration::from_secs(1); + assert_eq!(decide(&r, Some(GOOD), after).await, StatusCode::OK); assert_eq!(store.lookups(), 2); } + + /// A burst against a store that errors, or that sleeps past + /// `LOOKUP_TIMEOUT`, costs one lookup, and every request is denied + /// within that one timeout rather than queueing behind one timeout each. + #[tokio::test(start_paused = true)] + async fn a_burst_against_a_failing_store_costs_one_lookup() { + const BURST: usize = 20; + let erroring = Fake::holding(SECRET); + erroring.set(Err(())); + for (name, store) in [("erroring", erroring), ("hanging", Fake::hanging())] { + let store = Arc::new(store); + let r = Arc::new(Responder { + source: Arc::clone(&store), + ttl: TTL, + cached: tokio::sync::Mutex::new(Cache::default()), + }); + let started = tokio::time::Instant::now(); + let now = Instant::now(); + let mut burst = tokio::task::JoinSet::new(); + for _ in 0..BURST { + let r = Arc::clone(&r); + burst.spawn(async move { decide(&r, Some(GOOD), now).await }); + } + while let Some(status) = burst.join_next().await { + assert_eq!(status.unwrap(), StatusCode::UNAUTHORIZED, "{name}"); + } + assert_eq!(store.lookups(), 1, "{name}"); + assert!(started.elapsed() <= LOOKUP_TIMEOUT, "{name}"); + } + } }