702 lines
27 KiB
Rust
702 lines
27 KiB
Rust
//! 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<Output = anyhow::Result<Option<String>>> + 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<Option<String>> {
|
||
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<Stored> = 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<Utc>,
|
||
}
|
||
|
||
#[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<Cached>,
|
||
/// 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<Instant>,
|
||
/// 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<String> {
|
||
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<String> {
|
||
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<S> {
|
||
source: S,
|
||
ttl: Duration,
|
||
cached: tokio::sync::Mutex<Cache>,
|
||
}
|
||
|
||
/// 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<impl BearerSource>,
|
||
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<Arc<Responder<Store>>>, 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<Box<dyn Future<Output = anyhow::Result<Option<String>>> + 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<MockBearerSource> {
|
||
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<Answer> = 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);
|
||
}
|
||
}
|