//! The metrics store gateway's `auth_request` responder for Grafana. //! //! nginx sends it every request to the store's query API that carries a //! `Bearer` (`swarm-victoriametrics.nix`). It answers 200 when the bearer is //! the value stored at `--store-path`, which Grafana's host creates and its //! datasource presents, and 401 otherwise: `auth_request` turns any other //! status into a 500 for the caller. A store that does not answer is a 401 //! only until a value has been read; see below. //! //! It reads the store under its own certificate, whose policy reaches that one //! path. It serves a unix socket whose directory only the gateway's nginx may //! enter. //! //! 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, 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. //! //! When the store does not answer, requests are checked against the last //! value it did return, up to `MAX_KEPT_AGE` after that read, and a wrong //! bearer is still refused. The first such request of each failure streak logs //! it. A revoked or rotated bearer therefore stays accepted for as long as the //! store cannot be reached, for at most `MAX_KEPT_AGE`; the first answer //! replaces the value, or drops it when the store holds none. Past that age the //! value is dropped, and a process that has not yet read the store has none: //! both deny. use std::future::Future; use std::os::unix::fs::PermissionsExt as _; use std::path::PathBuf; use std::sync::Arc; use std::time::{Duration, Instant}; use anyhow::Context as _; use axum::extract::State; use axum::http::{HeaderMap, StatusCode, header}; use chrono::{DateTime, SecondsFormat, Utc}; use clap::Parser; use subtle::ConstantTimeEq as _; use swarm_secret_client::client::{DEFAULT_CERT_MOUNT, Settings}; /// How long one store lookup may take before the store counts as not /// answering. Well under nginx's 60s `proxy_read_timeout`, so a slow store /// reads as a 401 or a fallback answer rather than a gateway timeout. const LOOKUP_TIMEOUT: Duration = Duration::from_secs(5); /// How long requests skip the 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); /// How old the last bearer read from the store may be and still be checked /// against when the store does not answer. It covers the store's slow minutes /// after a boot (5 to 20 so far), and caps how long a revoked or rotated /// bearer can survive an outage. Counted from the last successful read. const MAX_KEPT_AGE: Duration = Duration::from_hours(1); #[derive(Parser)] #[command(about = "auth_request responder: checks Grafana's metrics bearer against the store")] struct Args { /// Unix socket to serve on. #[arg(long)] socket: PathBuf, /// Cert-auth role this responder logs in to the store as. #[arg(long, default_value = "swarm-metrics-auth")] store_cert_role: String, /// KV path, under the `secret` mount, holding the bearer in its `value` /// field. #[arg(long)] store_path: String, /// Seconds the stored bearer is used without reading the store again; /// 0 reads it on every request, except for a few seconds after a read /// failed. #[arg(long, default_value_t = 60)] cache_ttl: u64, } /// Where the stored bearer comes from. #[cfg_attr(test, mockall::automock)] trait BearerSource { /// The stored bearer, `None` when nothing is stored, or an error when the /// store could not answer. fn lookup(&self) -> impl Future>> + Send; } /// The object at the bearer's path. `bao kv put … value=-` on Grafana's host /// writes this field. #[derive(serde::Deserialize)] struct Stored { value: String, } /// The swarm secret store, reached with the responder's own certificate. struct Store { settings: Settings, cert_role: String, path: String, } impl BearerSource for Store { /// Logs in per lookup, so no token is left to expire inside a long-lived /// process. async fn lookup(&self) -> anyhow::Result> { let store = swarm_secret_client::SecretStore::connect( &self.settings, &self.cert_role, DEFAULT_CERT_MOUNT, ) .await .context("logging in to the secret store")?; let stored: Option = store .read_optional(&self.path) .await .with_context(|| format!("reading {}", self.path))?; Ok(stored.map(|s| s.value)) } } /// The last value the store returned. struct Cached { value: String, /// When it was read, on the clock the TTL is measured on. read_at: Instant, /// The same moment, for the log. read_at_wall: DateTime, } #[derive(Default)] struct Cache { /// Kept past `--cache-ttl`, up to `MAX_KEPT_AGE`: an expired value is /// replaced by the next answer from the store, and checked against while /// there is none. stored: Option, /// Until when requests skip the 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, /// Whether this failure streak has logged that it checks against `stored`. fallback_logged: bool, } impl Cache { /// Drops `stored` once it is `MAX_KEPT_AGE` old at `now`, returning the line /// to log: `Some` once, when it goes. fn expire_kept(&mut self, now: Instant) -> Option { let kept = self.stored.as_ref()?; if now.saturating_duration_since(kept.read_at) < MAX_KEPT_AGE { return None; } let read_at = kept.read_at_wall; self.stored = None; Some(format!( "the store did not answer for too long; dropping the bearer read at {}", read_at.to_rfc3339_opts(SecondsFormat::Secs, true) )) } /// The line to log when a request is checked against `stored` because the /// store did not answer: `Some` for the first such request of a failure /// streak, `None` for the rest. fn fallback_notice(&mut self) -> Option { let read_at = self.stored.as_ref()?.read_at_wall; if std::mem::replace(&mut self.fallback_logged, true) { return None; } Some(format!( "the store did not answer; checking against the bearer read at {}", read_at.to_rfc3339_opts(SecondsFormat::Secs, true) )) } } struct Responder { source: S, ttl: Duration, cached: tokio::sync::Mutex, } /// What reading the store for an expired copy came to. enum Refresh { /// The store returned a bearer, now in the cache. Replaced, /// The store answered that it holds no usable bearer; the cache is empty. Empty, /// The store failed or timed out, now or a moment ago. Unanswered, } /// Reads the store into `cache`, unless a lookup failed less than /// `LOOKUP_FAILURE_WINDOW` ago. async fn refresh(source: &impl BearerSource, cache: &mut Cache, now: Instant) -> Refresh { if cache.failed_until.is_some_and(|until| now < until) { return Refresh::Unanswered; } match tokio::time::timeout(LOOKUP_TIMEOUT, source.lookup()).await { Ok(Ok(Some(value))) if !value.is_empty() => { cache.stored = Some(Cached { value, read_at: now, read_at_wall: Utc::now(), }); cache.failed_until = None; cache.fallback_logged = false; Refresh::Replaced } Ok(Ok(_)) => { tracing::warn!("no bearer is stored; denying"); *cache = Cache::default(); Refresh::Empty } Ok(Err(e)) => { tracing::warn!(error = format!("{e:#}"), "bearer lookup failed"); cache.failed_until = Some(Instant::now() + LOOKUP_FAILURE_WINDOW); Refresh::Unanswered } Err(_) => { tracing::warn!("bearer lookup timed out"); cache.failed_until = Some(Instant::now() + LOOKUP_FAILURE_WINDOW); Refresh::Unanswered } } } /// The answer for one subrequest's `Authorization` header, at `now`. async fn decide( responder: &Responder, authorization: Option<&str>, now: Instant, ) -> StatusCode { let Some(bearer) = authorization.and_then(|v| v.strip_prefix("Bearer ")) else { tracing::warn!("no bearer token; denying"); return StatusCode::UNAUTHORIZED; }; // Held across the lookup, so concurrent requests wait for one read // rather than each starting their own. 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 { match refresh(&responder.source, &mut cache, now).await { Refresh::Replaced => {} Refresh::Empty => return StatusCode::UNAUTHORIZED, Refresh::Unanswered => { if let Some(line) = cache.expire_kept(now) { tracing::warn!("{line}"); } if cache.stored.is_none() { tracing::warn!("the store did not answer and no bearer is held; denying"); return StatusCode::UNAUTHORIZED; } if let Some(line) = cache.fallback_notice() { tracing::warn!("{line}"); } } } } 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 // bearer was right. if bool::from(stored.value.as_bytes().ct_eq(bearer.as_bytes())) { StatusCode::OK } else { tracing::warn!("bearer does not match the stored one; denying"); StatusCode::UNAUTHORIZED } } async fn handle(State(responder): State>>, headers: HeaderMap) -> StatusCode { let authorization = headers .get(header::AUTHORIZATION) .and_then(|v| v.to_str().ok()); decide(&responder, authorization, Instant::now()).await } #[tokio::main] async fn main() -> anyhow::Result<()> { swarm_secret_client::install_crypto_provider(); tracing_subscriber::fmt() .with_env_filter( tracing_subscriber::EnvFilter::try_from_default_env() .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), ) // stdout is journald, which keeps ANSI escapes as raw bytes. .with_ansi(false) .init(); let args = Args::parse(); let settings = Settings::from_env() .context("reading the secret store's address and this responder's identity")?; let responder = Arc::new(Responder { source: Store { settings, cert_role: args.store_cert_role, path: args.store_path, }, ttl: Duration::from_secs(args.cache_ttl), cached: tokio::sync::Mutex::new(Cache::default()), }); // `RuntimeDirectoryPreserve=yes` can leave the previous run's socket; the // directory is this unit's alone, so unlinking it is safe. if let Err(e) = std::fs::remove_file(&args.socket) && e.kind() != std::io::ErrorKind::NotFound { return Err(e) .with_context(|| format!("clearing stale socket at {}", args.socket.display())); } let listener = tokio::net::UnixListener::bind(&args.socket) .with_context(|| format!("binding {}", args.socket.display()))?; // nginx is another user and `connect(2)` needs write. The directory is // the access control, not this mode. std::fs::set_permissions(&args.socket, std::fs::Permissions::from_mode(0o666)) .with_context(|| format!("chmod {}", args.socket.display()))?; tracing::info!(socket = %args.socket.display(), "swarm-metrics-auth listening"); let app = axum::Router::new().fallback(handle).with_state(responder); axum::serve(listener, app).await.context("serving") } #[cfg(test)] mod tests { use super::*; use std::pin::Pin; type Lookup = Pin>> + Send>>; type Answer = fn() -> Lookup; fn holds(value: &str) -> Lookup { let value = value.to_owned(); Box::pin(async move { Ok(Some(value)) }) } fn holds_nothing() -> Lookup { Box::pin(async { Ok(None) }) } fn fails() -> Lookup { Box::pin(async { Err(anyhow::anyhow!("store unreachable")) }) } /// Accepts the lookup and never answers. fn hangs() -> Lookup { Box::pin(std::future::pending()) } /// A store that answers each lookup in turn with the next of `answers`, /// and fails the test on one more. fn store(answers: &[Answer]) -> MockBearerSource { let mut store = MockBearerSource::new(); let mut order = mockall::Sequence::new(); for &answer in answers { store .expect_lookup() .times(1) .in_sequence(&mut order) .returning(answer); } store } fn secret() -> Lookup { holds(SECRET) } fn replacement() -> Lookup { holds("replaced") } const TTL: Duration = Duration::from_mins(1); const SECRET: &str = "0f1e2d3c4b5a"; const GOOD: &str = "Bearer 0f1e2d3c4b5a"; const REPLACED: &str = "Bearer replaced"; fn responder(source: MockBearerSource) -> Responder { Responder { source, ttl: TTL, cached: tokio::sync::Mutex::new(Cache::default()), } } /// A request time far enough back that a value read then has expired, and /// that is still before the failure window a lookup now opens. fn long_ago() -> Instant { Instant::now() .checked_sub(TTL * 2) .expect("the monotonic clock is older than two minutes") } #[tokio::test] async fn the_stored_bearer_is_200() { let r = responder(store(&[secret])); assert_eq!(decide(&r, Some(GOOD), Instant::now()).await, StatusCode::OK); } #[tokio::test] async fn everything_else_is_401() { let r = responder(store(&[secret])); for authorization in [ None, Some(SECRET), Some("Basic 0f1e2d3c4b5a"), Some("Bearer 0f1e2d3c4b5b"), Some("Bearer 0f1e2d3c4b5"), Some("Bearer "), Some("Bearer authelia_at_abc.def"), ] { assert_eq!( decide(&r, authorization, Instant::now()).await, StatusCode::UNAUTHORIZED, "{authorization:?}" ); } } /// Nothing stored, an empty value, and a store that cannot answer are /// all denials, whatever the bearer, when no value was ever read. #[tokio::test] async fn no_usable_stored_value_is_401() { for answer in [holds_nothing, || holds(""), fails] { let mut source = MockBearerSource::new(); source.expect_lookup().returning(answer); let r = responder(source); for authorization in [GOOD, "Bearer "] { assert_eq!( decide(&r, Some(authorization), Instant::now()).await, StatusCode::UNAUTHORIZED ); } } } /// Within the TTL no request reaches the store, a wrong bearer included, /// and a replacement the store already holds goes unseen; after it the /// store is read again, and its answer replaces the old value at once. #[tokio::test] async fn the_stored_value_is_reused_until_the_ttl() { let r = responder(store(&[secret, replacement])); let t0 = Instant::now(); assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::OK); let within = t0 + Duration::from_secs(59); assert_eq!(decide(&r, Some(GOOD), within).await, StatusCode::OK); assert_eq!( decide(&r, Some(REPLACED), within).await, StatusCode::UNAUTHORIZED ); assert_eq!( decide(&r, Some(GOOD), t0 + TTL).await, StatusCode::UNAUTHORIZED ); assert_eq!(decide(&r, Some(REPLACED), t0 + TTL).await, StatusCode::OK); } /// 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_after_the_failure_window() { let r = responder(store(&[fails, secret])); let t0 = Instant::now(); assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::UNAUTHORIZED); assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::UNAUTHORIZED); let after = t0 + LOOKUP_FAILURE_WINDOW + Duration::from_secs(1); assert_eq!(decide(&r, Some(GOOD), after).await, StatusCode::OK); } /// With a value read before, a store that errors or sleeps past /// `LOOKUP_TIMEOUT` leaves the requests checked against that value: the /// matching bearer is accepted, a wrong one refused, and the store is not /// asked again inside the failure window. #[tokio::test(start_paused = true)] async fn an_unanswering_store_leaves_the_last_good_bearer_in_force() { for (name, outage) in [("erroring", fails as Answer), ("hanging", hangs)] { let r = responder(store(&[secret, outage])); assert_eq!( decide(&r, Some(GOOD), long_ago()).await, StatusCode::OK, "{name}" ); for _ in 0..3 { assert_eq!( decide(&r, Some(GOOD), Instant::now()).await, StatusCode::OK, "{name}" ); assert_eq!( decide(&r, Some("Bearer 0f1e2d3c4b5b"), Instant::now()).await, StatusCode::UNAUTHORIZED, "{name}" ); } } } /// With no value ever read, a store that sleeps past `LOOKUP_TIMEOUT` /// denies the bearer the last test accepts. #[tokio::test(start_paused = true)] async fn a_timeout_without_a_last_good_bearer_is_401() { let r = responder(store(&[hangs])); for _ in 0..3 { assert_eq!( decide(&r, Some(GOOD), Instant::now()).await, StatusCode::UNAUTHORIZED ); } } /// The store's next answer replaces the value kept through the outage. #[tokio::test(start_paused = true)] async fn the_first_answer_after_an_outage_replaces_the_value() { let r = responder(store(&[secret, hangs, replacement])); assert_eq!(decide(&r, Some(GOOD), long_ago()).await, StatusCode::OK); assert_eq!(decide(&r, Some(GOOD), Instant::now()).await, StatusCode::OK); let recovered = Instant::now() + LOOKUP_FAILURE_WINDOW + Duration::from_secs(1); assert_eq!( decide(&r, Some(GOOD), recovered).await, StatusCode::UNAUTHORIZED ); assert_eq!(decide(&r, Some(REPLACED), recovered).await, StatusCode::OK); } /// A store that answers it holds no bearer drops the value: a later /// outage does not bring it back. #[tokio::test] async fn an_empty_answer_drops_the_last_good_bearer() { let r = responder(store(&[secret, holds_nothing, fails])); assert_eq!(decide(&r, Some(GOOD), long_ago()).await, StatusCode::OK); assert_eq!( decide(&r, Some(GOOD), Instant::now()).await, StatusCode::UNAUTHORIZED ); assert_eq!( decide(&r, Some(GOOD), Instant::now()).await, StatusCode::UNAUTHORIZED ); } /// A burst against a store that errors, or that sleeps past /// `LOOKUP_TIMEOUT`, costs one lookup, and every request is answered /// within that one timeout rather than queueing behind one timeout each: /// denied with no value read before, checked against it with one. #[tokio::test(start_paused = true)] async fn a_burst_against_an_unanswering_store_costs_one_lookup() { const BURST: usize = 20; for (name, outage) in [("erroring", fails as Answer), ("hanging", hangs)] { for (value_read, expected) in [(false, StatusCode::UNAUTHORIZED), (true, StatusCode::OK)] { let answers: Vec = if value_read { vec![secret, outage] } else { vec![outage] }; let r = Arc::new(responder(store(&answers))); if value_read { assert_eq!(decide(&r, Some(GOOD), long_ago()).await, StatusCode::OK); } 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(), expected, "{name}, value read: {value_read}" ); } assert!(started.elapsed() <= LOOKUP_TIMEOUT, "{name}"); } } } /// Each failure streak yields the fallback line once, naming when the /// value was read; a streak that ends in an answer, and a cache holding no /// value, yield none. #[test] fn the_fallback_is_noticed_once_per_failure_streak() { let mut cache = Cache::default(); assert_eq!(cache.fallback_notice(), None); cache.stored = Some(Cached { value: SECRET.to_owned(), read_at: Instant::now(), read_at_wall: "2026-11-04T07:12:11Z".parse().unwrap(), }); assert_eq!( cache.fallback_notice().as_deref(), Some( "the store did not answer; checking against the bearer read at \ 2026-11-04T07:12:11Z" ) ); for _ in 0..3 { assert_eq!(cache.fallback_notice(), None); } cache.fallback_logged = false; assert!(cache.fallback_notice().is_some()); } /// `decide` marks the streak as noticed on the first request checked /// against the kept value, and an answer from the store ends the streak. #[tokio::test(start_paused = true)] async fn an_answer_from_the_store_ends_the_failure_streak() { let r = responder(store(&[secret, hangs, replacement])); assert_eq!(decide(&r, Some(GOOD), long_ago()).await, StatusCode::OK); assert!(!r.cached.lock().await.fallback_logged); for _ in 0..3 { decide(&r, Some(GOOD), Instant::now()).await; assert!(r.cached.lock().await.fallback_logged); } let recovered = Instant::now() + LOOKUP_FAILURE_WINDOW + Duration::from_secs(1); assert_eq!(decide(&r, Some(REPLACED), recovered).await, StatusCode::OK); assert!(!r.cached.lock().await.fallback_logged); } /// A store that answers the leading `answers` once each, in order, then /// fails every lookup. fn then_down(answers: &[Answer]) -> MockBearerSource { let mut source = store(answers); source.expect_lookup().returning(fails); source } /// Within an hour of the read an unanswering store leaves the /// matching bearer accepted and a wrong one refused; from that age on both /// are refused. #[tokio::test] async fn the_kept_bearer_is_checked_against_only_until_the_max_age() { let r = responder(then_down(&[secret])); let t0 = Instant::now(); assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::OK); let last_moment = t0 + Duration::from_secs(59 * 60 + 59); assert_eq!(decide(&r, Some(GOOD), last_moment).await, StatusCode::OK); assert_eq!( decide(&r, Some("Bearer 0f1e2d3c4b5b"), last_moment).await, StatusCode::UNAUTHORIZED ); let expired = t0 + Duration::from_hours(1); assert_eq!( decide(&r, Some(GOOD), expired).await, StatusCode::UNAUTHORIZED ); assert_eq!( decide(&r, Some(GOOD), expired + Duration::from_secs(1)).await, StatusCode::UNAUTHORIZED ); } /// An answer from the store restarts the age: a value read again after /// 50 minutes is still checked against 20 minutes into the next outage, /// 70 minutes after the first read. #[tokio::test] async fn an_answer_from_the_store_resets_the_age() { let r = responder(then_down(&[secret, secret])); let t0 = Instant::now(); assert_eq!(decide(&r, Some(GOOD), t0).await, StatusCode::OK); let second_read = t0 + Duration::from_mins(50); assert_eq!(decide(&r, Some(GOOD), second_read).await, StatusCode::OK); let outage = t0 + Duration::from_mins(70); assert_eq!(decide(&r, Some(GOOD), outage).await, StatusCode::OK); } /// The expiry yields its line once, naming the read time, and leaves /// nothing to check against; a younger value is left alone. #[test] fn the_expiry_of_the_kept_bearer_is_noticed_once() { let t0 = Instant::now(); let mut cache = Cache { stored: Some(Cached { value: SECRET.to_owned(), read_at: t0, read_at_wall: "2026-11-04T07:12:11Z".parse().unwrap(), }), ..Cache::default() }; assert_eq!( cache.expire_kept(t0 + MAX_KEPT_AGE.saturating_sub(Duration::from_secs(1))), None ); assert!(cache.stored.is_some()); assert_eq!( cache.expire_kept(t0 + MAX_KEPT_AGE).as_deref(), Some( "the store did not answer for too long; dropping the bearer read at \ 2026-11-04T07:12:11Z" ) ); assert!(cache.stored.is_none()); assert_eq!(cache.expire_kept(t0 + MAX_KEPT_AGE * 2), None); } }