micromegas_auth/

db_api_key.rs

//! DB-backed API key authentication.
//!
//! Holds API keys in two Postgres tables — `ingestion_api_keys` and
//! `analytics_api_keys` (migration v5, `rust/ingestion/src/sql_migration.rs`) —
//! each row storing only a SHA-256 hash of the key, plus a
//! `created_at`/`created_by`/`last_used_at`/`revoked_at`/`revoked_by` audit trail.
//! [`crate::db_api_key::DbApiKeyAuthProvider`] validates by hash-indexed lookup behind a short-TTL
//! `moka` cache.

use crate::env::resolve_prefixed_var;
use crate::ip_allowlist::IpAllowlist;
use crate::types::{AuthContext, AuthProvider, AuthType, ProviderUnavailable, RequestParts};
use anyhow::{Context, Result, anyhow};
use base64::Engine;
use chrono::Utc;
use moka::future::Cache;
use rand::Rng;
use sha2::{Digest, Sha256};
use sqlx::PgPool;
use sqlx::Row;
use std::sync::Arc;
use std::sync::atomic::{AtomicI64, Ordering};
use std::time::Duration;

/// Which key table a provider (or the management API) is bound to.
///
/// The table name is always a `&'static str` literal — never derived from caller
/// input — so the SQL built from it (below) can never be injected into.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ApiKeyTable {
    /// `ingestion_api_keys` — write credentials, minted/listed/revoked over HTTP.
    Ingestion,
    /// `analytics_api_keys` — read credentials, minted/listed/revoked over
    /// HTTP through `analytics-web-srv`'s own routes (`/api/analytics-api-keys*`,
    /// see the admin runbook) — not through this crate's ingestion-hosted API.
    Analytics,
}

impl ApiKeyTable {
    /// Static table name.
    pub fn table_name(self) -> &'static str {
        match self {
            ApiKeyTable::Ingestion => "ingestion_api_keys",
            ApiKeyTable::Analytics => "analytics_api_keys",
        }
    }

    /// Whether this table carries a write-side `audience` column (`ingestion_api_keys` only,
    /// migration v6). `analytics_api_keys`'s read-side mirror is `read_audiences`, a set-valued
    /// grant on the caller rather than a column on the key row.
    pub fn has_audience(self) -> bool {
        matches!(self, ApiKeyTable::Ingestion)
    }

    /// Whether a key from this table may delegate (mint/act as another principal) --
    /// `analytics_api_keys` only.
    pub fn allows_delegation(self) -> bool {
        matches!(self, ApiKeyTable::Analytics)
    }
}

/// Cache and audit knobs for [`DbApiKeyAuthProvider`], read from env with defaults.
#[derive(Clone, Copy, Debug)]
pub struct DbApiKeyConfig {
    /// `MICROMEGAS_API_KEY_CACHE_SIZE`, default 10_000.
    pub cache_size: u64,
    /// `MICROMEGAS_AUTH_CACHE_TTL_SECONDS`, default 60 -- a single flat,
    /// unprefixed knob (no `{prefix}_` role variant) shared with
    /// `DbAudienceGrantsConfig`/`DbGroupsConfig`'s own positive-cache TTL: one
    /// value governs the API-key, audience-grant, and group snapshot caches
    /// process-wide, across every role. Also the bound on revocation latency: a
    /// `DELETE` writes `revoked_at` but cannot invalidate a remote process's
    /// cache, so a revoked key keeps authenticating until this TTL elapses.
    pub cache_ttl_secs: u64,
    /// `MICROMEGAS_API_KEY_UNKNOWN_CACHE_TTL_SECONDS`, default 10. Shorter than
    /// `cache_ttl_secs` so a freshly minted key is not masked by an earlier probe
    /// of that same (not-yet-existing) string for longer than necessary.
    pub unknown_cache_ttl_secs: u64,
    /// `MICROMEGAS_API_KEY_UNKNOWN_CACHE_SIZE`, default 10_000.
    pub unknown_cache_size: u64,
}

pub(crate) fn resolve_u64(prefix: &str, suffix: &str, default: u64) -> u64 {
    let var = resolve_prefixed_var(prefix, suffix);
    let raw = std::env::var(&var).ok();
    match raw {
        Some(s) => s.parse::<u64>().unwrap_or_else(|_| {
            micromegas_tracing::warn!("Invalid {var} value '{s}', using default {default}");
            default
        }),
        None => default,
    }
}

impl DbApiKeyConfig {
    /// Resolves each `API_KEY_`-specific knob as `{prefix}_*` first, falling back to the
    /// unprefixed name, the same `{prefix}_*`-with-fallback convention
    /// [`crate::env::resolve_prefixed_var`] implements for every other knob in this crate. With
    /// an empty prefix this is identical to the unprefixed vars, so an unprefixed caller just
    /// passes `""`.
    ///
    /// `cache_ttl_secs` is the exception: it reads the **flat, unprefixed**
    /// `MICROMEGAS_AUTH_CACHE_TTL_SECONDS` knob directly (`prefix` is ignored for this one
    /// field), the same single process-wide value `DbAudienceGrantsConfig`/`DbGroupsConfig`
    /// read -- there is exactly one value for this knob, no per-role distinction.
    pub fn from_env_with_prefix(prefix: &str) -> Self {
        Self {
            cache_size: resolve_u64(prefix, "API_KEY_CACHE_SIZE", 10_000),
            cache_ttl_secs: resolve_u64("", "AUTH_CACHE_TTL_SECONDS", 60),
            unknown_cache_ttl_secs: resolve_u64(prefix, "API_KEY_UNKNOWN_CACHE_TTL_SECONDS", 10),
            unknown_cache_size: resolve_u64(prefix, "API_KEY_UNKNOWN_CACHE_SIZE", 10_000),
        }
    }
}

/// SHA-256 of the full key string. The only place a key is hashed; the mint
/// routes (`rust/analytics-web-srv/src/ingestion_keys.rs`, `analytics_keys.rs`)
/// and this provider's lookup go through the same digest definition.
pub fn hash_key(key: &str) -> [u8; 32] {
    let mut hasher = Sha256::new();
    hasher.update(key.as_bytes());
    hasher.finalize().into()
}

/// 256 bits of OS entropy, base64url-nopad, `mmk_`-prefixed. `mmk_` makes minted
/// keys recognizable to secret scanners; it is cosmetic to validation, since
/// [`hash_key`] covers the whole string rather than parsing the prefix.
pub fn generate_key() -> String {
    let mut rng = rand::rng();
    let bytes: [u8; 32] = rng.random();
    let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(bytes);
    format!("mmk_{encoded}")
}

/// Builds a small, dedicated connection pool for key-store lookups from an
/// existing pool's options — rather than sharing the caller's lake pool
/// directly. Inherits the lake pool's options wholesale (including any
/// `after_connect`/`before_acquire` hooks, e.g. the read-only-connection checks in
/// `micromegas_ingestion::data_lake_connection`), not just its connect options, so a strict or
/// prefer-fallback lake pool's write-session guarantee carries over to the key store.
/// `max_connections(4)`, `acquire_timeout(2s)`: a credential flood or a
/// DB outage on the key-lookup path can therefore never starve the write path of
/// connections, and a lookup fails fast into its 503 rather than blocking up to
/// sqlx's default 30s `acquire_timeout`.
pub fn dedicated_key_store_pool(lake_pool: &PgPool) -> PgPool {
    let connect_options = (*lake_pool.connect_options()).clone();
    lake_pool
        .options()
        .clone()
        .max_connections(4)
        .acquire_timeout(Duration::from_secs(2))
        .connect_lazy_with(connect_options)
}

/// True when `err`'s SQLSTATE is `25006` (`cannot execute ... in a read-only transaction`) --
/// the code the `last_used_at` update raises when its connection turned read-only mid-query.
/// A local copy of `micromegas_ingestion::data_lake_connection::is_read_only_violation`'s check:
/// `auth` doesn't depend on `ingestion`.
pub fn is_read_only_error(err: &sqlx::Error) -> bool {
    err.as_database_error()
        .and_then(|db_err| db_err.code())
        .is_some_and(|code| code == "25006")
}

/// `SELECT EXISTS(SELECT 1 FROM <table> WHERE revoked_at IS NULL)` — used by
/// `ProviderBuilder::build()` (`rust/auth/src/default_provider.rs`) to decide
/// whether a non-empty key store counts as "auth configured" on its own. A
/// missing relation (e.g. a schema that has not reached migration v5 yet)
/// surfaces as an `Err` here and must be propagated, never treated as "empty".
pub async fn key_store_has_live_rows(pool: &PgPool, table: ApiKeyTable) -> Result<bool> {
    let row = sqlx::query(&format!(
        "SELECT EXISTS(SELECT 1 FROM {} WHERE revoked_at IS NULL) AS has_rows",
        table.table_name()
    ))
    .fetch_one(pool)
    .await
    .with_context(|| format!("checking whether {} has any live key", table.table_name()))?;
    let has_rows: bool = row
        .try_get("has_rows")
        .with_context(|| "reading has_rows")?;
    Ok(has_rows)
}

/// hash -> (key_id, name, audience, allowlist) for a key known to be live.
struct KeyRow {
    key_id: uuid::Uuid,
    name: String,
    /// `Some` for `Ingestion` (the column is `NOT NULL` as of migration v6), `None` for
    /// `Analytics` (which carries no `audience` column at all).
    audience: Option<String>,
    /// Parsed once per cache fill (not per request) from the `allowed_cidrs` column --
    /// `IpAllowlist::parse` is fed `&[]` when the column read back `None` (every pre-v11 row),
    /// giving the same "no restriction" allowlist as an empty stored array.
    allowlist: IpAllowlist,
}

/// Distinguishes "the DB answered: no such live key" and "the row exists but is
/// malformed" -- both credential rejections -- from "the DB could not be reached
/// at all", which alone becomes a [`ProviderUnavailable`] and must never populate
/// either cache.
#[derive(thiserror::Error, Debug)]
enum LookupError {
    #[error("no such live key")]
    NotFound,
    #[error("{0}")]
    Db(anyhow::Error),
    /// A row was read successfully but its `allowed_cidrs` column could not be decoded or
    /// parsed -- e.g. a hand-edited row (`mkdocs/docs/admin/api-keys.md` documents that as
    /// supported) with a CIDR the column has no `CHECK` constraint to reject. Not a DB outage:
    /// retrying the same query only reproduces the same malformed row.
    #[error("{0}")]
    Invalid(anyhow::Error),
}

/// Shared throttle primitive: true at most once per `window_secs`, checked-and-set
/// via a single `AtomicI64` "last logged at" timestamp. Backs both `maybe_log_error`
/// and the read-only-fallback `warn!` in `lookup_key_row` -- each keeps its own
/// `AtomicI64` so a burst of one kind of event can never suppress the other's log.
fn should_log(last_logged_at: &AtomicI64, window_secs: i64) -> bool {
    let now = Utc::now().timestamp();
    let prev = last_logged_at.load(Ordering::Relaxed);
    now.saturating_sub(prev) >= window_secs
        && last_logged_at
            .compare_exchange(prev, now, Ordering::Relaxed, Ordering::Relaxed)
            .is_ok()
}

/// Rate-limits the outage `error!` log to at most once per `window_secs` (per
/// table) -- not once per rejected request. `DbApiKeyAuthProvider` sits last in the
/// auth chain, so during an outage every non-env-key, non-JWT request reaches it; an
/// unconditional `error!` would flood `log_entries` with the outage's own noise
/// on the highest-volume service in the deployment.
fn maybe_log_error(last_logged_at: &AtomicI64, window_secs: i64, table: &str, err: &anyhow::Error) {
    if should_log(last_logged_at, window_secs) {
        micromegas_tracing::error!("db_api_key store error (table={table}): {err:#}");
    }
}

/// DB-backed API key authentication provider, bound to one [`ApiKeyTable`].
///
/// Every DB error is wrapped in [`ProviderUnavailable`] before being returned and
/// emits `imetric!("db_api_key_error_count", "count", table = name, 1)` unconditionally
/// — the metric is unconditional even on requests whose `error!` line was
/// suppressed by `maybe_log_error`'s rate limit.
pub struct DbApiKeyAuthProvider {
    pool: PgPool,
    table: ApiKeyTable,
    /// hash -> (key_id, name) for keys known to be live.
    valid: Cache<[u8; 32], Arc<KeyRow>>,
    /// hash -> () for tokens the DB answered "no such live key" for, and for
    /// tokens whose row exists but failed to decode/parse (`LookupError::Invalid`) --
    /// both are credential rejections from the caller's point of view.
    unknown: Cache<[u8; 32], ()>,
    /// Rate-limit window for the outage `error!` log — mirrors `cache_ttl_secs`,
    /// floored so a zero or very low cache TTL doesn't also disable log
    /// rate-limiting.
    error_log_window_secs: i64,
    last_logged_at: Arc<AtomicI64>,
    /// Separate throttle state for the read-only-fallback `warn!` in `lookup_key_row` --
    /// its own `AtomicI64` so a flood of unknown tokens against a read-only connection can
    /// never suppress `last_logged_at`'s genuine DB-error log, or vice versa.
    last_read_only_warn_at: Arc<AtomicI64>,
}

impl DbApiKeyAuthProvider {
    /// Creates a new provider bound to `table`, using `pool` for lookups (the
    /// caller is expected to pass a [`dedicated_key_store_pool`], not the lake
    /// pool itself).
    pub fn new(pool: PgPool, table: ApiKeyTable, config: DbApiKeyConfig) -> Self {
        let valid = Cache::builder()
            .max_capacity(config.cache_size)
            .time_to_live(Duration::from_secs(config.cache_ttl_secs))
            .build();
        let unknown = Cache::builder()
            .max_capacity(config.unknown_cache_size)
            .time_to_live(Duration::from_secs(config.unknown_cache_ttl_secs))
            .build();
        Self {
            pool,
            table,
            valid,
            unknown,
            error_log_window_secs: config.cache_ttl_secs.max(60) as i64,
            last_logged_at: Arc::new(AtomicI64::new(0)),
            last_read_only_warn_at: Arc::new(AtomicI64::new(0)),
        }
    }
}

/// Looks up a live key's row, bumping `last_used_at` on the way when possible.
///
/// Tries the `UPDATE ... RETURNING` first. On a `25006` (the connection turned read-only, e.g.
/// a `WritablePolicy::Prefer` FlightSQL pool that fell back to a replica), falls back to a plain
/// `SELECT` of the same columns and skips the bump rather than failing the whole lookup --
/// `last_used_at` is best-effort telemetry, not something a caller's auth should depend on.
/// Unlike valid keys (cached in `self.valid`), an unknown token reaches this fallback each time
/// it's a distinct token missing `self.unknown`, and again each time the same token is retried
/// after `self.unknown`'s TTL (`unknown_cache_ttl_secs`) has expired -- so the `warn!` is
/// throttled via `last_read_only_warn_at`/`window_secs` rather than logged unconditionally.
async fn lookup_key_row(
    pool: &PgPool,
    table: ApiKeyTable,
    hash: &[u8; 32],
    returning: &str,
    last_read_only_warn_at: &AtomicI64,
    window_secs: i64,
) -> Result<Option<sqlx::postgres::PgRow>, sqlx::Error> {
    let update_result = sqlx::query(&format!(
        "UPDATE {} SET last_used_at = now() WHERE key_hash = $1 AND revoked_at IS NULL RETURNING {returning}",
        table.table_name()
    ))
    .bind(&hash[..])
    .fetch_optional(pool)
    .await;
    match update_result {
        Err(e) if is_read_only_error(&e) => {
            if should_log(last_read_only_warn_at, window_secs) {
                micromegas_tracing::warn!(
                    "db_api_key lookup on a read-only connection (table={}): skipping last_used_at bump",
                    table.table_name()
                );
            }
            sqlx::query(&format!(
                "SELECT {returning} FROM {} WHERE key_hash = $1 AND revoked_at IS NULL",
                table.table_name()
            ))
            .bind(&hash[..])
            .fetch_optional(pool)
            .await
        }
        other => other,
    }
}

#[async_trait::async_trait]
impl AuthProvider for DbApiKeyAuthProvider {
    async fn validate_request(&self, parts: &dyn RequestParts) -> Result<AuthContext> {
        let token = parts
            .bearer_token()
            .ok_or_else(|| anyhow!("missing bearer token"))?;
        let hash = hash_key(token);

        // A repeated probe of the same bogus token is free after the first
        // attempt; a flood of distinct bogus tokens still costs one DB round trip
        // per request.
        if self.unknown.get(&hash).await.is_some() {
            anyhow::bail!("invalid API token");
        }

        let pool = self.pool.clone();
        let table = self.table;
        let last_logged_at = self.last_logged_at.clone();
        let last_read_only_warn_at = self.last_read_only_warn_at.clone();
        let window_secs = self.error_log_window_secs;

        // `try_get_with`, not a plain `get`/`insert` pair: among concurrent
        // callers for the same hash, exactly one runs this loader — the rest
        // await its result. This is what rate-limits the `last_used_at` write to
        // once per cache TTL per key, not once per concurrent request.
        let result = self
            .valid
            .try_get_with(hash, async move {
                // Built from `&'static str` literals alongside `table.table_name()`, so
                // nothing caller-supplied ever reaches this SQL.
                let returning = if table.has_audience() {
                    "key_id, name, audience, allowed_cidrs"
                } else {
                    "key_id, name, allowed_cidrs"
                };
                let row = lookup_key_row(
                    &pool,
                    table,
                    &hash,
                    returning,
                    &last_read_only_warn_at,
                    window_secs,
                )
                .await
                .map_err(|e| {
                    let err = anyhow::Error::from(e)
                        .context(format!("looking up key in {}", table.table_name()));
                    // Unconditional: fires on every DB error, independent of the
                    // rate-limited `error!` line below.
                    micromegas_tracing::imetric!(
                        "db_api_key_error_count",
                        "count",
                        table = table.table_name(),
                        1_u64
                    );
                    maybe_log_error(&last_logged_at, window_secs, table.table_name(), &err);
                    LookupError::Db(err)
                })?;

                match row {
                    Some(row) => {
                        let key_id: uuid::Uuid = row.try_get("key_id").map_err(|e| {
                            LookupError::Db(anyhow::Error::from(e).context("reading key_id"))
                        })?;
                        let name: String = row.try_get("name").map_err(|e| {
                            LookupError::Db(anyhow::Error::from(e).context("reading name"))
                        })?;
                        let audience: Option<String> = if table.has_audience() {
                            Some(row.try_get("audience").map_err(|e| {
                                LookupError::Db(anyhow::Error::from(e).context("reading audience"))
                            })?)
                        } else {
                            None
                        };
                        // A nullable `TEXT[]` decodes into `Option<Vec<String>>`, not a bare
                        // `Vec<String>` -- sqlx returns `UnexpectedNull` on a bare `Vec<String>`
                        // target when the column is `NULL`, which every pre-v11 row is. `None`
                        // is treated as an empty slice before handing it to `IpAllowlist::parse`.
                        let allowed_cidrs: Option<Vec<String>> =
                            row.try_get("allowed_cidrs").map_err(|e| {
                                let err = anyhow::Error::from(e).context(format!(
                                    "reading allowed_cidrs for key_id={key_id} in {}",
                                    table.table_name()
                                ));
                                micromegas_tracing::error!(
                                    "db_api_key store: malformed row (table={}): {err:#}",
                                    table.table_name()
                                );
                                LookupError::Invalid(err)
                            })?;
                        let allowlist =
                            IpAllowlist::parse(allowed_cidrs.as_deref().unwrap_or_default())
                                .map_err(|e| {
                                    let err = e.context(format!(
                                        "parsing allowed_cidrs for key_id={key_id} in {}",
                                        table.table_name()
                                    ));
                                    micromegas_tracing::error!(
                                        "db_api_key store: malformed row (table={}): {err:#}",
                                        table.table_name()
                                    );
                                    LookupError::Invalid(err)
                                })?;
                        Ok(Arc::new(KeyRow {
                            key_id,
                            name,
                            audience,
                            allowlist,
                        }))
                    }
                    None => Err(LookupError::NotFound),
                }
            })
            .await;

        match result {
            Ok(row) => {
                // Checked after a cache hit or a fresh load, before returning `Ok(AuthContext)`.
                // Deliberately not cached in `self.unknown` (that cache means "no such live
                // key", not "this key exists but this caller may not use it from here") and
                // must not increment `db_api_key_error_count` (that metric means "the DB was
                // unreachable", not "a credential was rejected") -- both untouched, since this
                // runs after the `try_get_with` closure above already returned successfully.
                if !row.allowlist.allows(parts.client_ip()) {
                    micromegas_tracing::warn!(
                        "db api key rejected by allowlist: table={} key_id={} name={} client_ip={:?}",
                        self.table.table_name(),
                        row.key_id,
                        row.name,
                        parts.client_ip()
                    );
                    return Err(anyhow!("invalid API token: source IP not permitted"));
                }
                micromegas_tracing::trace!(
                    "db api key validated: table={} key_id={} name={}",
                    self.table.table_name(),
                    row.key_id,
                    row.name
                );
                Ok(AuthContext {
                    subject: row.name.clone(),
                    // Deliberately left `None`, not `created_by`: under the grant-map model,
                    // `email` is what `user:` selectors match on, so populating it from the
                    // minting admin would hand this key every audience granted to *them*.
                    email: None,
                    issuer: "api_key".to_string(),
                    audience: None,
                    expires_at: None,
                    auth_type: AuthType::ApiKey,
                    // `false` for ingestion keys: an ingestion write credential is not a
                    // delegating service account, and can never reach the gRPC path
                    // `allow_delegation` governs anyway (it lives in the other table).
                    // Unchanged (`true`) for analytics keys.
                    allow_delegation: self.table.allows_delegation(),
                    // `Some(..)` for every ingestion key (the column is `NOT NULL`), `None`
                    // for analytics keys -- `read_audiences` populates that side instead.
                    bound_audience: row.audience.clone(),
                    read_audiences: vec![],
                    // API keys carry no email for a `user:` member to match, so only a
                    // wildcard-admin group could ever make one admin.
                    memberships: Arc::from([]),
                })
            }
            // A DB *error* is propagated and cached in neither map: caching an
            // outage as `unknown` would turn a transient failure into a
            // TTL-long outage for every affected key.
            Err(arc_err) => match arc_err.as_ref() {
                // `Invalid` is already logged (with `key_id`) where it's raised, above --
                // this arm only needs to render the generic client-facing rejection and stop
                // the malformed row from being re-queried on every subsequent request.
                LookupError::NotFound | LookupError::Invalid(_) => {
                    self.unknown.insert(hash, ()).await;
                    Err(anyhow!("invalid API token"))
                }
                LookupError::Db(e) => Err(ProviderUnavailable(anyhow!(
                    "{} key store unavailable: {e:#}",
                    self.table.table_name()
                ))
                .into()),
            },
        }
    }
}

#[cfg(test)]
mod tests {
    //! `KeyRow` and the `valid`/`unknown` caches are private, so the no-DB allowlist coverage
    //! the plan called for lives here rather than in `tests/db_api_key_tests.rs` -- there is no
    //! public seam from that integration-test crate to pre-populate a cache entry directly.
    //! Mirrors that file's "no DB" pattern: a lazily-connected pool that is never actually
    //! queried, since the cache is pre-populated and the loader closure never runs.

    use super::*;
    use crate::policy::PUBLIC_AUDIENCE;
    use crate::types::{HttpRequestParts, RequestParts};
    use micromegas_tracing::event::in_memory_sink::InMemorySink;
    use micromegas_tracing::metrics::MetricsMsgQueueAny;
    use micromegas_tracing::test_utils::init_in_memory_tracing;
    use micromegas_transit::HeterogeneousQueue;
    use serial_test::serial;
    use sqlx::postgres::PgPoolOptions;
    use std::net::{IpAddr, Ipv4Addr};

    fn unreachable_pool() -> PgPool {
        PgPoolOptions::new()
            .acquire_timeout(Duration::from_millis(50))
            .connect_lazy("postgres://localhost/unused")
            .expect("lazy pool creation is infallible")
    }

    fn test_config() -> DbApiKeyConfig {
        DbApiKeyConfig {
            cache_size: 100,
            cache_ttl_secs: 60,
            unknown_cache_ttl_secs: 10,
            unknown_cache_size: 100,
        }
    }

    fn bearer_parts(token: &str, client_ip: Option<IpAddr>) -> HttpRequestParts {
        let mut headers = http::HeaderMap::new();
        headers.insert(
            http::header::AUTHORIZATION,
            format!("Bearer {token}").parse().expect("valid header"),
        );
        HttpRequestParts {
            headers,
            method: http::Method::GET,
            uri: "/test".parse().expect("valid uri"),
            client_ip,
        }
    }

    fn count_integer_metric(sink: &InMemorySink, name: &str) -> u64 {
        let state = sink.state.lock().expect("sink lock");
        let mut count = 0u64;
        for block in &state.metrics_blocks {
            for evt in block.events.iter() {
                match evt {
                    MetricsMsgQueueAny::IntegerMetricEvent(e) if e.desc.name == name => count += 1,
                    MetricsMsgQueueAny::TaggedIntegerMetricEvent(e) if e.desc.name == name => {
                        count += 1
                    }
                    _ => {}
                }
            }
        }
        count
    }

    /// Pre-populates the `valid` cache with a canned `KeyRow` carrying a populated
    /// `allowed_cidrs`-derived `IpAllowlist`, bypassing `try_get_with`'s loader (and therefore
    /// the DB) entirely -- `validate_request` only ever reaches the DB on a cache miss.
    async fn provider_with_cached_allowlisted_key(
        token: &str,
        allowed_cidrs: &[&str],
    ) -> DbApiKeyAuthProvider {
        let provider =
            DbApiKeyAuthProvider::new(unreachable_pool(), ApiKeyTable::Ingestion, test_config());
        let allowlist = IpAllowlist::parse(
            &allowed_cidrs
                .iter()
                .map(|s| s.to_string())
                .collect::<Vec<_>>(),
        )
        .expect("valid CIDR");
        provider
            .valid
            .insert(
                hash_key(token),
                Arc::new(KeyRow {
                    key_id: uuid::Uuid::new_v4(),
                    name: "allowlisted-test-key".to_string(),
                    audience: Some(PUBLIC_AUDIENCE.to_string()),
                    allowlist,
                }),
            )
            .await;
        provider
    }

    #[tokio::test]
    #[serial]
    async fn allowlisted_key_authenticates_from_in_range_ip_without_db_access() {
        let token = "mmk_allowlist_in_range";
        let provider = provider_with_cached_allowlisted_key(token, &["10.0.0.0/24"]).await;
        let parts = bearer_parts(token, Some(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 5))));

        let result = provider.validate_request(&parts as &dyn RequestParts).await;
        assert!(result.is_ok(), "in-range IP must be permitted");
    }

    #[tokio::test]
    #[serial]
    async fn allowlisted_key_is_rejected_from_out_of_range_ip_without_db_access() {
        let guard = init_in_memory_tracing();
        let token = "mmk_allowlist_out_of_range";
        let provider = provider_with_cached_allowlisted_key(token, &["10.0.0.0/24"]).await;
        let parts = bearer_parts(token, Some(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1))));

        let result = provider.validate_request(&parts as &dyn RequestParts).await;
        assert!(result.is_err(), "out-of-range IP must be rejected");

        // Rejection on an allowlist mismatch is a distinct case from "no such live key" (which
        // populates `unknown`) and from a DB outage (which increments the error metric) -- this
        // is neither, so both must stay untouched.
        assert!(
            provider.unknown.get(&hash_key(token)).await.is_none(),
            "an allowlist rejection must not poison the `unknown` cache for a key that exists"
        );
        micromegas_tracing::dispatch::flush_metrics_buffer();
        assert_eq!(
            count_integer_metric(&guard.sink, "db_api_key_error_count"),
            0,
            "an allowlist rejection is not a DB error and must not increment the error metric"
        );
    }

    #[tokio::test]
    #[serial]
    async fn allowlisted_key_is_rejected_when_client_ip_is_unresolved() {
        let token = "mmk_allowlist_no_ip";
        let provider = provider_with_cached_allowlisted_key(token, &["10.0.0.0/24"]).await;
        let parts = bearer_parts(token, None);

        let result = provider.validate_request(&parts as &dyn RequestParts).await;
        assert!(
            result.is_err(),
            "an unresolved client IP must never satisfy a non-empty allowlist"
        );
    }
}