rmcp_server_kit/

auth.rs

//! Authentication middleware for MCP servers.
//!
//! Supports multiple authentication methods tried in priority order:
//! 1. mTLS client certificate (if configured and peer cert present)
//! 2. Bearer token (API key) with Argon2id hash verification
//!
//! Includes per-source-IP rate limiting on authentication attempts.

use std::{
    collections::HashSet,
    net::SocketAddr,
    num::{NonZeroU32, NonZeroUsize},
    path::PathBuf,
    sync::{
        Arc, LazyLock, Mutex,
        atomic::{AtomicU64, Ordering},
    },
    time::Duration,
};

use arc_swap::ArcSwap;
use argon2::{Argon2, PasswordHash, PasswordHasher, PasswordVerifier};
use axum::{
    body::Body,
    extract::ConnectInfo,
    http::{Request, StatusCode, header},
    middleware::Next,
    response::{IntoResponse, Response},
};
use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use secrecy::SecretString;
use serde::Deserialize;
use x509_parser::prelude::*;

use crate::{
    bounded_limiter::{BoundedKeyedLimiter, BoundedLimiterDeny, KeyEvictionPolicy},
    error::RmcpServerKitError,
    transport::RateLimitKey,
};

/// Identity of an authenticated caller.
///
/// The [`Debug`] impl is **manually written** to redact the raw bearer token
/// and the JWT `sub` claim. This prevents accidental disclosure if an
/// `AuthIdentity` is ever logged via `tracing::debug!(?identity, …)` or
/// `format!("{identity:?}")`. Only `name`, `role`, and `method` are printed
/// in the clear; `raw_token` and `sub` are rendered as `<redacted>` /
/// `<present>` / `<none>` markers.
#[derive(Clone)]
#[non_exhaustive]
pub struct AuthIdentity {
    /// Human-readable identity name (e.g. API key label or cert CN).
    pub name: String,
    /// RBAC role associated with this identity.
    pub role: String,
    /// Which authentication mechanism produced this identity.
    pub method: AuthMethod,
    /// Raw bearer token from the `Authorization` header, wrapped in
    /// [`SecretString`] so it is never accidentally logged or serialized.
    /// Present for OAuth JWT; `None` for mTLS and API-key auth.
    /// Tool handlers use this for downstream token passthrough via
    /// [`crate::rbac::current_token`].
    pub raw_token: Option<SecretString>,
    /// JWT `sub` claim (stable user identifier, e.g. Keycloak UUID).
    /// Used for token store keying. `None` for non-JWT auth.
    pub sub: Option<String>,
}

impl std::fmt::Debug for AuthIdentity {
    /// Redacts `raw_token` and `sub` to prevent secret leakage via
    /// `format!("{:?}")` or `tracing::debug!(?identity)`.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("AuthIdentity")
            .field("name", &self.name)
            .field("role", &self.role)
            .field("method", &self.method)
            .field(
                "raw_token",
                &if self.raw_token.is_some() {
                    "<redacted>"
                } else {
                    "<none>"
                },
            )
            .field(
                "sub",
                &if self.sub.is_some() {
                    "<redacted>"
                } else {
                    "<none>"
                },
            )
            .finish()
    }
}

/// How the caller authenticated.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum AuthMethod {
    /// Bearer API key (Argon2id-hashed, configured statically).
    BearerToken,
    /// Mutual TLS client certificate.
    MtlsCertificate,
    /// OAuth 2.1 JWT bearer token (validated via JWKS).
    OAuthJwt,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AuthFailureClass {
    MissingCredential,
    InvalidCredential,
    #[cfg_attr(
        not(feature = "oauth"),
        allow(
            dead_code,
            reason = "only OAuth JWT validation can report an expired credential; \
                      the variant is unconstructed in builds without that feature"
        )
    )]
    ExpiredCredential,
    /// Source IP exceeded the post-failure backoff limit.
    RateLimited,
    /// Source IP exceeded the pre-auth abuse gate (rejected before any
    /// password-hash work — see [`AuthState::pre_auth_limiter`]).
    PreAuthGate,
}

impl AuthFailureClass {
    fn as_str(self) -> &'static str {
        match self {
            Self::MissingCredential => "missing_credential",
            Self::InvalidCredential => "invalid_credential",
            Self::ExpiredCredential => "expired_credential",
            Self::RateLimited => "rate_limited",
            Self::PreAuthGate => "pre_auth_gate",
        }
    }

    fn bearer_error(self) -> (&'static str, &'static str) {
        match self {
            Self::MissingCredential => (
                "invalid_request",
                "missing bearer token or mTLS client certificate",
            ),
            Self::InvalidCredential => ("invalid_token", "token is invalid"),
            Self::ExpiredCredential => ("invalid_token", "token is expired"),
            Self::RateLimited => ("invalid_request", "too many failed authentication attempts"),
            Self::PreAuthGate => (
                "invalid_request",
                "too many unauthenticated requests from this source",
            ),
        }
    }

    fn response_body(self) -> &'static str {
        match self {
            Self::MissingCredential => "unauthorized: missing credential",
            Self::InvalidCredential => "unauthorized: invalid credential",
            Self::ExpiredCredential => "unauthorized: expired credential",
            Self::RateLimited => "rate limited",
            Self::PreAuthGate => "rate limited (pre-auth)",
        }
    }
}

/// Snapshot of authentication success/failure counters.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
#[non_exhaustive]
pub struct AuthCountersSnapshot {
    /// Successful mTLS authentications.
    pub success_mtls: u64,
    /// Successful bearer-token authentications.
    pub success_bearer: u64,
    /// Successful OAuth JWT authentications.
    pub success_oauth_jwt: u64,
    /// Failures because no credential was presented.
    pub failure_missing_credential: u64,
    /// Failures because the credential was malformed or wrong.
    pub failure_invalid_credential: u64,
    /// Failures because the credential had expired.
    pub failure_expired_credential: u64,
    /// Failures because the source IP was rate-limited (post-failure backoff).
    pub failure_rate_limited: u64,
    /// Failures because the source IP exceeded the pre-auth abuse gate.
    /// These never reach the password-hash verification path.
    pub failure_pre_auth_gate: u64,
}

/// Internal atomic counters backing [`AuthCountersSnapshot`].
#[derive(Debug, Default)]
pub(crate) struct AuthCounters {
    success_mtls: AtomicU64,
    success_bearer: AtomicU64,
    success_oauth_jwt: AtomicU64,
    failure_missing_credential: AtomicU64,
    failure_invalid_credential: AtomicU64,
    failure_expired_credential: AtomicU64,
    failure_rate_limited: AtomicU64,
    failure_pre_auth_gate: AtomicU64,
}

impl AuthCounters {
    fn record_success(&self, method: AuthMethod) {
        match method {
            AuthMethod::MtlsCertificate => {
                self.success_mtls.fetch_add(1, Ordering::Relaxed);
            }
            AuthMethod::BearerToken => {
                self.success_bearer.fetch_add(1, Ordering::Relaxed);
            }
            AuthMethod::OAuthJwt => {
                self.success_oauth_jwt.fetch_add(1, Ordering::Relaxed);
            }
        }
    }

    fn record_failure(&self, class: AuthFailureClass) {
        match class {
            AuthFailureClass::MissingCredential => {
                self.failure_missing_credential
                    .fetch_add(1, Ordering::Relaxed);
            }
            AuthFailureClass::InvalidCredential => {
                self.failure_invalid_credential
                    .fetch_add(1, Ordering::Relaxed);
            }
            AuthFailureClass::ExpiredCredential => {
                self.failure_expired_credential
                    .fetch_add(1, Ordering::Relaxed);
            }
            AuthFailureClass::RateLimited => {
                self.failure_rate_limited.fetch_add(1, Ordering::Relaxed);
            }
            AuthFailureClass::PreAuthGate => {
                self.failure_pre_auth_gate.fetch_add(1, Ordering::Relaxed);
            }
        }
    }

    fn snapshot(&self) -> AuthCountersSnapshot {
        AuthCountersSnapshot {
            success_mtls: self.success_mtls.load(Ordering::Relaxed),
            success_bearer: self.success_bearer.load(Ordering::Relaxed),
            success_oauth_jwt: self.success_oauth_jwt.load(Ordering::Relaxed),
            failure_missing_credential: self.failure_missing_credential.load(Ordering::Relaxed),
            failure_invalid_credential: self.failure_invalid_credential.load(Ordering::Relaxed),
            failure_expired_credential: self.failure_expired_credential.load(Ordering::Relaxed),
            failure_rate_limited: self.failure_rate_limited.load(Ordering::Relaxed),
            failure_pre_auth_gate: self.failure_pre_auth_gate.load(Ordering::Relaxed),
        }
    }
}

/// RFC 3339 timestamp, parsed at deserialization time.
///
/// Use this for any public field that needs to carry an RFC 3339 timestamp from
/// TOML/JSON config or builder APIs. Construction is fallible (`parse`); once
/// constructed the value is guaranteed to be a real RFC 3339 timestamp with a
/// known offset, so downstream code does not need to handle parse errors.
///
/// Wraps [`chrono::DateTime<chrono::FixedOffset>`]; the underlying value is
/// available via [`Self::as_datetime`] or [`Self::into_inner`]. `Serialize`
/// emits the canonical RFC 3339 form via [`chrono::DateTime::to_rfc3339`], so
/// the on-the-wire format for `ApiKeySummary` (admin endpoints) is unchanged.
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
#[non_exhaustive]
pub struct RfcTimestamp(chrono::DateTime<chrono::FixedOffset>);

impl RfcTimestamp {
    /// Parse an RFC 3339 timestamp.
    ///
    /// # Errors
    ///
    /// Returns the underlying [`chrono::ParseError`] when `s` is not a valid
    /// RFC 3339 timestamp (e.g. missing the `T` separator, missing the offset
    /// suffix, or out-of-range fields).
    pub fn parse(s: &str) -> Result<Self, chrono::ParseError> {
        chrono::DateTime::parse_from_rfc3339(s).map(Self)
    }

    /// Borrow the underlying [`chrono::DateTime`].
    #[must_use]
    pub fn as_datetime(&self) -> &chrono::DateTime<chrono::FixedOffset> {
        &self.0
    }

    /// Consume the wrapper and return the underlying [`chrono::DateTime`].
    #[must_use]
    pub fn into_inner(self) -> chrono::DateTime<chrono::FixedOffset> {
        self.0
    }
}

impl std::fmt::Display for RfcTimestamp {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Canonical RFC 3339 form; matches the deserialization input contract.
        write!(f, "{}", self.0.to_rfc3339())
    }
}

impl std::fmt::Debug for RfcTimestamp {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Render as the canonical RFC 3339 string (not chrono's internal
        // debug form) so existing `ApiKeyEntry` Debug-redaction tests --
        // which look for the literal `"2030-01-01T00:00:00Z"` form in the
        // formatted output -- continue to hold without bespoke handling.
        write!(f, "{}", self.0.to_rfc3339())
    }
}

impl<'de> Deserialize<'de> for RfcTimestamp {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        // Validate at deserialization time: a malformed `expires_at` in
        // TOML or JSON aborts config load with a clear serde error rather
        // than silently producing a key that fails open at runtime.
        let s = String::deserialize(deserializer)?;
        Self::parse(&s).map_err(serde::de::Error::custom)
    }
}

impl serde::Serialize for RfcTimestamp {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        serializer.serialize_str(&self.0.to_rfc3339())
    }
}

impl From<chrono::DateTime<chrono::FixedOffset>> for RfcTimestamp {
    fn from(value: chrono::DateTime<chrono::FixedOffset>) -> Self {
        Self(value)
    }
}

/// A single API key entry (stored as Argon2id hash in config).
///
/// The [`Debug`] impl is **manually written** to redact the Argon2id hash.
/// Although the hash is not directly reversible, treating it as a secret
/// prevents offline brute-force attempts from leaked logs and matches the
/// defense-in-depth posture used for [`AuthIdentity`].
#[derive(Clone, Deserialize)]
#[serde(deny_unknown_fields)]
#[non_exhaustive]
pub struct ApiKeyEntry {
    /// Human-readable key label (used in logs and audit records).
    pub name: String,
    /// Argon2id hash of the token (PHC string format).
    pub hash: String,
    /// RBAC role granted when this key authenticates successfully.
    pub role: String,
    /// Optional expiry, parsed from an RFC 3339 string at deserialization
    /// time. Construction from a raw string is fallible (see
    /// [`RfcTimestamp::parse`] and [`ApiKeyEntry::try_with_expiry`]),
    /// which guarantees `verify_bearer_token` never sees a malformed value.
    pub expires_at: Option<RfcTimestamp>,
}

impl std::fmt::Debug for ApiKeyEntry {
    /// Redacts the Argon2id `hash` to keep it out of logs, panic backtraces,
    /// and admin-endpoint responses that might `format!("{:?}", …)` an entry.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ApiKeyEntry")
            .field("name", &self.name)
            .field("hash", &"<redacted>")
            .field("role", &self.role)
            .field("expires_at", &self.expires_at)
            .finish()
    }
}

impl ApiKeyEntry {
    /// Create a new API key entry (no expiry).
    #[must_use]
    pub fn new(name: impl Into<String>, hash: impl Into<String>, role: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            hash: hash.into(),
            role: role.into(),
            expires_at: None,
        }
    }

    /// Set an RFC 3339 expiry on this key.
    ///
    /// Takes an already-parsed [`RfcTimestamp`]; for ergonomic construction
    /// from a raw string see [`Self::try_with_expiry`].
    #[must_use]
    pub fn with_expiry(mut self, expires_at: RfcTimestamp) -> Self {
        self.expires_at = Some(expires_at);
        self
    }

    /// Set an RFC 3339 expiry on this key from a raw string.
    ///
    /// # Errors
    ///
    /// Returns the underlying [`chrono::ParseError`] when `expires_at` is
    /// not a valid RFC 3339 timestamp. This is the fallible counterpart to
    /// [`Self::with_expiry`].
    pub fn try_with_expiry(
        mut self,
        expires_at: impl AsRef<str>,
    ) -> Result<Self, chrono::ParseError> {
        self.expires_at = Some(RfcTimestamp::parse(expires_at.as_ref())?);
        Ok(self)
    }
}

/// mTLS client certificate authentication configuration.
#[derive(Debug, Clone, Deserialize)]
#[serde(deny_unknown_fields)]
#[allow(
    clippy::struct_excessive_bools,
    reason = "mTLS CRL behavior is intentionally configured as independent booleans"
)]
#[non_exhaustive]
pub struct MtlsConfig {
    /// Path to CA certificate(s) for verifying client certs (PEM format).
    pub ca_cert_path: PathBuf,
    /// If true, clients MUST present a valid certificate.
    /// If false, client certs are optional (verified if presented).
    #[serde(default)]
    pub required: bool,
    /// Default RBAC role for mTLS-authenticated clients.
    /// The client cert CN becomes the identity name.
    #[serde(default = "default_mtls_role")]
    pub default_role: String,
    /// Enable CRL-based certificate revocation checks using CDP URLs from the
    /// configured CA chain and connecting client certificates.
    #[serde(default = "default_true")]
    pub crl_enabled: bool,
    /// Optional fixed refresh interval for known CRLs. When omitted, refresh
    /// cadence is derived from `nextUpdate` and clamped internally.
    #[serde(default, with = "humantime_serde::option")]
    pub crl_refresh_interval: Option<Duration>,
    /// Timeout for individual CRL fetches.
    #[serde(default = "default_crl_fetch_timeout", with = "humantime_serde")]
    pub crl_fetch_timeout: Duration,
    /// Retry-retention window: how long a CRL whose refresh keeps failing is
    /// retained in the cache so the background refresher can keep retrying it,
    /// measured past the CRL's `nextUpdate`.
    ///
    /// This does **not** permit use of an expired CRL. When
    /// `crl_enforce_expiration` is set (the default), webpki rejects any CRL
    /// past its `nextUpdate` during validation; this window only bounds how
    /// long a persistently-failing entry is kept for retry before the verifier
    /// gives up and evicts it (at which point `crl_deny_on_unavailable` governs
    /// the handshake outcome).
    ///
    /// The preferred config key is `crl_retry_retention`; `crl_stale_grace` is
    /// accepted as a deprecated alias for backward compatibility.
    #[serde(
        default = "default_crl_stale_grace",
        alias = "crl_retry_retention",
        with = "humantime_serde"
    )]
    pub crl_stale_grace: Duration,
    /// When true, missing or unavailable CRLs cause revocation checks to fail
    /// closed.
    ///
    /// Defaults to `true`. RFC 5280 §6.3 treats a certificate whose
    /// revocation status cannot be determined as unverified, so a client
    /// certificate advertising CRL distribution points is rejected when
    /// *every* relevant CDP is uncached and unfetchable. Denial requires all
    /// relevant CDPs to be unavailable, not merely one -- otherwise an
    /// attacker who blocks a single mirror could deny service.
    ///
    /// Set to `false` to restore the pre-3.9 fail-open behaviour, in which an
    /// unreachable CRL lets the handshake proceed. That is strongly
    /// discouraged: a revoked certificate is then accepted whenever its CRL
    /// is unreachable, which is precisely the condition an attacker holding a
    /// revoked certificate can induce.
    #[serde(default = "default_true")]
    pub crl_deny_on_unavailable: bool,
    /// When true, apply revocation checks only to the end-entity certificate.
    #[serde(default)]
    pub crl_end_entity_only: bool,
    /// Allow HTTP CRL distribution-point URLs in addition to HTTPS.
    ///
    /// Defaults to `true` because RFC 5280 §4.2.1.13 designates HTTP (and
    /// LDAP) as the canonical transport for CRL distribution points.
    /// SSRF defense for HTTP CDPs is provided by the IP-allowlist guard
    /// (private/loopback/link-local/multicast/cloud-metadata addresses are
    /// always rejected), redirect=none, body-size cap, and per-host
    /// concurrency limit -- not by forcing HTTPS.
    #[serde(default = "default_true")]
    pub crl_allow_http: bool,
    /// Enforce CRL expiration during certificate validation.
    #[serde(default = "default_true")]
    pub crl_enforce_expiration: bool,
    /// Maximum concurrent CRL fetches across all hosts. Defense in depth
    /// against SSRF amplification: even if many CDPs are discovered, no
    /// more than this many fetches run in parallel. Per-host concurrency
    /// is independently capped at 1 regardless of this value.
    /// Default: `4`.
    #[serde(default = "default_crl_max_concurrent_fetches")]
    pub crl_max_concurrent_fetches: usize,
    /// Hard cap on each CRL response body in bytes. Fetches exceeding this
    /// are aborted mid-stream to bound memory and prevent gzip-bomb-style
    /// amplification. Default: 5 MiB (`5 * 1024 * 1024`).
    #[serde(default = "default_crl_max_response_bytes")]
    pub crl_max_response_bytes: u64,
    /// Global CDP discovery rate limit, in URLs per minute. Throttles
    /// how many *new* CDP URLs the verifier may admit into the fetch
    /// pipeline across the whole process, bounding asymmetric `DoS`
    /// amplification when attacker-controlled certificates carry large
    /// CDP lists. The limit is global (not per-source-IP) in this
    /// release; per-IP scoping is deferred to a future version because
    /// it requires plumbing the peer `SocketAddr` through the rustls
    /// verifier hook (a different subsystem than ordinary request
    /// middleware). Note: the **bearer pre-auth limiter** that gates
    /// API-key / OAuth `Authorization` headers is already per-IP — see
    /// [`RateLimitConfig::pre_auth_max_per_minute`] and the keyed
    /// governor built by `build_pre_auth_limiter`. URLs that lose the
    /// rate-limiter race are *not* marked as seen, so subsequent
    /// handshakes observing the same URL can retry admission.
    /// Default: `60`.
    #[serde(default = "default_crl_discovery_rate_per_min")]
    pub crl_discovery_rate_per_min: u32,
    /// Maximum number of distinct hosts that may hold a CRL fetch
    /// semaphore at any time. At the cap, idle entries (no in-flight
    /// fetch) are evicted on demand so new hosts keep working; only when
    /// every entry has a concurrent in-flight fetch does the request
    /// return [`RmcpServerKitError::Config`] containing the literal substring
    /// `"crl_host_semaphore_cap_exceeded"`. Bounds memory growth from
    /// attacker-controlled CDP URLs pointing at unique hostnames.
    /// Default: 1024.
    #[serde(default = "default_crl_max_host_semaphores")]
    pub crl_max_host_semaphores: usize,
    /// Maximum number of distinct URLs tracked in the "seen" set.
    /// Beyond this, additional discovered URLs are silently dropped
    /// with a rate-limited warn! log; no error surfaces. Default: 4096.
    #[serde(default = "default_crl_max_seen_urls")]
    pub crl_max_seen_urls: usize,
    /// Maximum number of cached CRL entries. Beyond this, new
    /// successful fetches are silently dropped with a rate-limited
    /// warn! log (newest-rejected, not LRU-evicted). Default: 1024.
    #[serde(default = "default_crl_max_cache_entries")]
    pub crl_max_cache_entries: usize,
}

fn default_mtls_role() -> String {
    "viewer".into()
}

const fn default_true() -> bool {
    true
}

const fn default_crl_fetch_timeout() -> Duration {
    Duration::from_secs(30)
}

const fn default_crl_stale_grace() -> Duration {
    Duration::from_hours(24)
}

const fn default_crl_max_concurrent_fetches() -> usize {
    4
}

const fn default_crl_max_response_bytes() -> u64 {
    5 * 1024 * 1024
}

const fn default_crl_discovery_rate_per_min() -> u32 {
    60
}

const fn default_crl_max_host_semaphores() -> usize {
    1024
}

const fn default_crl_max_seen_urls() -> usize {
    4096
}

const fn default_crl_max_cache_entries() -> usize {
    1024
}

/// Rate limiting configuration for authentication attempts.
///
/// rmcp-server-kit uses two independent per-IP token-bucket limiters for auth:
///
/// 1. **Pre-auth abuse gate** ([`Self::pre_auth_max_per_minute`]): consulted
///    *before* any password-hash work. Throttles unauthenticated traffic from
///    a single source IP so an attacker cannot pin the CPU on Argon2id by
///    spraying invalid bearer tokens. Sized generously (default = 10× the
///    post-failure quota) so legitimate clients are unaffected. mTLS-
///    authenticated connections bypass this gate entirely (the TLS handshake
///    already performed expensive crypto with a verified peer).
/// 2. **Post-failure backoff** ([`Self::max_attempts_per_minute`]): consulted
///    *after* an authentication attempt fails. Provides explicit backpressure
///    on bad credentials.
#[derive(Debug, Clone, Deserialize)]
#[serde(deny_unknown_fields)]
#[non_exhaustive]
pub struct RateLimitConfig {
    /// Maximum failed authentication attempts per source IP per minute.
    /// Successful authentications do not consume this budget.
    #[serde(default = "default_max_attempts")]
    pub max_attempts_per_minute: u32,
    /// Maximum *unauthenticated* requests per source IP per minute admitted
    /// to the password-hash verification path. When `None`, defaults to
    /// `max_attempts_per_minute * 10` at limiter-construction time.
    ///
    /// Set higher than [`Self::max_attempts_per_minute`] so honest clients
    /// retrying with the wrong key never trip this gate; its purpose is only
    /// to bound CPU usage under spray attacks.
    #[serde(default)]
    pub pre_auth_max_per_minute: Option<u32>,
    /// Hard cap on the number of distinct source IPs tracked per limiter.
    /// When reached, idle entries are pruned first; if still full, the
    /// oldest (LRU) entry is evicted to make room for the new one. This
    /// bounds memory under IP-spray attacks. Default: `10_000`.
    #[serde(default = "default_max_tracked_keys")]
    pub max_tracked_keys: usize,
    /// Per-IP entries idle for longer than this are eligible for
    /// opportunistic pruning. Default: 15 minutes.
    #[serde(default = "default_idle_eviction", with = "humantime_serde")]
    pub idle_eviction: Duration,
    /// Burst capacity for the post-failure limiter: the maximum number
    /// of failed attempts admitted back-to-back before the sustained
    /// `max_attempts_per_minute` rate applies. `None` (default) keeps
    /// governor's default of burst = rate. Must be greater than zero
    /// when set. May be smaller than the rate (smoothing) or larger
    /// (spike tolerance).
    #[serde(default)]
    pub burst: Option<u32>,
    /// Burst capacity for the pre-auth abuse gate. `None` (default)
    /// keeps burst = the gate's resolved rate. Legal regardless of
    /// whether [`Self::pre_auth_max_per_minute`] is set — the gate's
    /// base rate always resolves (`max_attempts_per_minute * 10` when
    /// unset). Must be greater than zero when set.
    #[serde(default)]
    pub pre_auth_burst: Option<u32>,
    /// Full-table policy when a rate limiter sees a new source IP after
    /// reaching [`Self::max_tracked_keys`]. Default: [`KeyEvictionPolicy::EvictLru`].
    #[serde(default)]
    pub key_eviction_policy: KeyEvictionPolicy,
}

impl Default for RateLimitConfig {
    fn default() -> Self {
        Self {
            max_attempts_per_minute: default_max_attempts(),
            pre_auth_max_per_minute: None,
            max_tracked_keys: default_max_tracked_keys(),
            idle_eviction: default_idle_eviction(),
            burst: None,
            pre_auth_burst: None,
            key_eviction_policy: KeyEvictionPolicy::default(),
        }
    }
}

impl RateLimitConfig {
    /// Create a rate limit config with the given max failed attempts per minute.
    /// Pre-auth gate defaults to `10x` this value at limiter-construction time.
    /// Memory-bound defaults are `10_000` tracked keys with 15-minute idle eviction.
    #[must_use]
    pub fn new(max_attempts_per_minute: u32) -> Self {
        Self {
            max_attempts_per_minute,
            ..Self::default()
        }
    }

    /// Override the pre-auth abuse-gate quota (per source IP per minute).
    /// When unset, defaults to `max_attempts_per_minute * 10`.
    #[must_use]
    pub fn with_pre_auth_max_per_minute(mut self, quota: u32) -> Self {
        self.pre_auth_max_per_minute = Some(quota);
        self
    }

    /// Override the per-limiter cap on tracked source-IP keys (default `10_000`).
    #[must_use]
    pub fn with_max_tracked_keys(mut self, max: usize) -> Self {
        self.max_tracked_keys = max;
        self
    }

    /// Override the idle-eviction window (default 15 minutes).
    #[must_use]
    pub fn with_idle_eviction(mut self, idle: Duration) -> Self {
        self.idle_eviction = idle;
        self
    }

    /// Set the burst capacity for the post-failure limiter. Must be
    /// greater than zero (validated at server-config validation time).
    #[must_use]
    pub fn with_burst(mut self, burst: u32) -> Self {
        self.burst = Some(burst);
        self
    }

    /// Set the burst capacity for the pre-auth abuse gate. Must be
    /// greater than zero (validated at server-config validation time).
    #[must_use]
    pub fn with_pre_auth_burst(mut self, burst: u32) -> Self {
        self.pre_auth_burst = Some(burst);
        self
    }

    /// Set the tracked-key full-table policy for auth limiters.
    #[must_use]
    pub const fn with_key_eviction_policy(mut self, policy: KeyEvictionPolicy) -> Self {
        self.key_eviction_policy = policy;
        self
    }
}

fn default_max_attempts() -> u32 {
    30
}

fn default_max_tracked_keys() -> usize {
    10_000
}

fn default_idle_eviction() -> Duration {
    Duration::from_mins(15)
}

/// Authentication configuration.
#[derive(Debug, Clone, Default, Deserialize)]
#[serde(deny_unknown_fields)]
#[non_exhaustive]
pub struct AuthConfig {
    /// Master switch - when false, all requests are allowed through.
    #[serde(default)]
    pub enabled: bool,
    /// Bearer token API keys.
    #[serde(default)]
    pub api_keys: Vec<ApiKeyEntry>,
    /// mTLS client certificate authentication.
    pub mtls: Option<MtlsConfig>,
    /// Rate limiting for auth attempts.
    pub rate_limit: Option<RateLimitConfig>,
    /// OAuth 2.1 JWT bearer token authentication.
    #[cfg(feature = "oauth")]
    pub oauth: Option<crate::oauth::OAuthConfig>,
    /// Presence-only placeholder for `auth.oauth` in builds without the
    /// `oauth` cargo feature.
    ///
    /// `deny_unknown_fields` (above) would otherwise reject an `[auth.oauth]`
    /// table with `unknown field \`oauth\``, which never mentions the feature
    /// flag and sends operators hunting for a typo that does not exist.
    /// Accepting the key here and rejecting it in
    /// [`AuthConfig::check_oauth_feature`] turns that into an actionable
    /// message. `IgnoredAny` records presence without retaining the value, so
    /// no OAuth secret is held in memory by a build that cannot use it.
    #[cfg(not(feature = "oauth"))]
    #[serde(default)]
    pub(crate) oauth: Option<serde::de::IgnoredAny>,
}

impl AuthConfig {
    /// Create an enabled auth config with the given API keys.
    #[must_use]
    pub fn with_keys(keys: Vec<ApiKeyEntry>) -> Self {
        Self {
            enabled: true,
            api_keys: keys,
            mtls: None,
            rate_limit: None,
            #[cfg(feature = "oauth")]
            oauth: None,
            #[cfg(not(feature = "oauth"))]
            oauth: None,
        }
    }

    /// Set rate limiting on this auth config.
    #[must_use]
    pub fn with_rate_limit(mut self, rate_limit: RateLimitConfig) -> Self {
        self.rate_limit = Some(rate_limit);
        self
    }

    /// Reject an `[auth.oauth]` table in a build compiled without the `oauth`
    /// cargo feature.
    ///
    /// Fails closed on purpose. Ignoring the table would start the server with
    /// OAuth silently disabled while the operator's configuration says it is
    /// on -- for a bearer-token deployment that is an unauthenticated server.
    ///
    /// # Errors
    ///
    /// Returns [`RmcpServerKitError::Config`] when `auth.oauth` is present and
    /// the `oauth` feature is disabled. Always `Ok` when the feature is
    /// enabled, where the table is parsed into
    /// [`oauth::OAuthConfig`](crate::oauth::OAuthConfig) instead.
    pub fn check_oauth_feature(&self) -> Result<(), RmcpServerKitError> {
        #[cfg(not(feature = "oauth"))]
        {
            (self.oauth.is_none()).ok_or_else(|| {
                RmcpServerKitError::Config(
                    "auth.oauth is configured but this build of rmcp-server-kit was compiled \
                     without the `oauth` cargo feature; rebuild with `--features oauth` or \
                     remove the [auth.oauth] table"
                        .into(),
                )
            })?;
        }
        Ok(())
    }
}

/// Summary of a single API key suitable for admin endpoints.
///
/// Intentionally omits the Argon2id hash - only metadata is exposed.
#[derive(Debug, Clone, serde::Serialize)]
#[non_exhaustive]
pub struct ApiKeySummary {
    /// Human-readable key label.
    pub name: String,
    /// RBAC role granted when this key authenticates.
    pub role: String,
    /// Optional RFC 3339 expiry timestamp. Serialized as a canonical
    /// RFC 3339 string so the admin-endpoint wire format is preserved.
    pub expires_at: Option<RfcTimestamp>,
}

/// Snapshot of the enabled authentication methods for admin endpoints.
#[derive(Debug, Clone, serde::Serialize)]
#[allow(
    clippy::struct_excessive_bools,
    reason = "this is a flat summary of independent auth-method booleans"
)]
#[non_exhaustive]
pub struct AuthConfigSummary {
    /// Master enabled flag from config.
    pub enabled: bool,
    /// Whether API-key bearer auth is configured.
    pub bearer: bool,
    /// Whether mTLS client auth is configured.
    pub mtls: bool,
    /// Whether OAuth JWT validation is configured.
    pub oauth: bool,
    /// Current API-key list (no hashes).
    pub api_keys: Vec<ApiKeySummary>,
}

impl AuthConfig {
    /// Produce a hash-free summary of the auth config for admin endpoints.
    #[must_use]
    pub fn summary(&self) -> AuthConfigSummary {
        AuthConfigSummary {
            enabled: self.enabled,
            bearer: !self.api_keys.is_empty(),
            mtls: self.mtls.is_some(),
            #[cfg(feature = "oauth")]
            oauth: self.oauth.is_some(),
            #[cfg(not(feature = "oauth"))]
            oauth: false,
            api_keys: self
                .api_keys
                .iter()
                .map(|k| ApiKeySummary {
                    name: k.name.clone(),
                    role: k.role.clone(),
                    expires_at: k.expires_at,
                })
                .collect(),
        }
    }
}

/// Keyed rate limiter type (per source IP). Memory-bounded by
/// [`RateLimitConfig::max_tracked_keys`] to defend against IP-spray `DoS`.
pub(crate) type KeyedLimiter = BoundedKeyedLimiter<RateLimitKey>;

/// Connection info for TLS connections, carrying the peer socket address
/// and (when mTLS is configured) the verified client identity extracted
/// from the peer certificate during the TLS handshake.
///
/// Defined as a local type so we can implement axum's `Connected` trait
/// for our custom `TlsListener` without orphan rule issues. The `identity`
/// field travels with the connection itself (via the wrapping IO type),
/// so there is no shared map to race against, no port-reuse aliasing, and
/// no eviction policy to maintain.
#[derive(Clone, Debug)]
#[non_exhaustive]
pub(crate) struct TlsConnInfo {
    /// Remote peer socket address.
    pub addr: SocketAddr,
    /// Verified mTLS client identity, if a client certificate was presented
    /// and successfully extracted during the TLS handshake.
    pub identity: Option<AuthIdentity>,
}

impl TlsConnInfo {
    /// Construct a new [`TlsConnInfo`].
    #[must_use]
    pub(crate) const fn new(addr: SocketAddr, identity: Option<AuthIdentity>) -> Self {
        Self { addr, identity }
    }
}

/// Default hard cap on the number of distinct authenticated identities
/// remembered by [`SeenIdentitySet`].
///
/// Sized to comfortably exceed realistic identity churn for an MCP server
/// while bounding worst-case memory at roughly `4096 * avg_name_len`
/// (~256 KiB at 64-byte names). Honest clients will never trigger eviction;
/// hostile churn (rotating mTLS subjects or OAuth `sub` values) is bounded.
const DEFAULT_SEEN_IDENTITY_CAP: usize = 4096;

/// Bounded set tracking which authenticated identities have already been
/// logged at INFO level (subsequent auths fall back to DEBUG).
///
/// # Why bounded?
///
/// `id.name` is attacker-influenced under mTLS (SAN/CN) and OAuth (`sub`).
/// An unbounded [`std::collections::HashSet`] would grow with churn,
/// producing both a slow memory leak and unbounded log-cardinality
/// downstream (Loki/ES). The cap follows the same trade-off documented in
/// [`crate::bounded_limiter`]: when an evicted identity reappears it
/// re-fires INFO once. This is acceptable for diagnostic logging.
///
/// # Concurrency
///
/// Uses [`std::sync::Mutex`] because [`Self::insert_is_first`] is purely
/// synchronous and the critical section never `.await`s. The mutex is
/// poison-tolerant: a poisoned set is still logically consistent
/// (only writer is `insert_is_first`, which performs an atomic insert
/// + bounded eviction; no torn invariants are possible).
pub(crate) struct SeenIdentitySet {
    inner: Mutex<SeenInner>,
}

struct SeenInner {
    set: HashSet<String>,
    /// Insertion-order FIFO used for bounded eviction. Tracking strict LRU
    /// would require touching the queue on every hit (under the mutex);
    /// FIFO is sufficient because the contract only promises "bounded
    /// memory", not "remember the most recently seen identities".
    order: std::collections::VecDeque<String>,
    cap: usize,
}

impl SeenIdentitySet {
    /// Construct with the default cap of [`DEFAULT_SEEN_IDENTITY_CAP`].
    #[must_use]
    pub(crate) fn new() -> Self {
        Self::with_cap(DEFAULT_SEEN_IDENTITY_CAP)
    }

    /// Construct with an explicit cap. A `cap` of `0` is silently raised
    /// to `1` to keep the invariant `set.len() <= cap` non-vacuous.
    #[must_use]
    pub(crate) fn with_cap(cap: usize) -> Self {
        let cap = cap.max(1);
        Self {
            inner: Mutex::new(SeenInner {
                set: HashSet::with_capacity(cap.min(64)),
                order: std::collections::VecDeque::with_capacity(cap.min(64)),
                cap,
            }),
        }
    }

    /// Insert `name`. Returns `true` if this is the first time `name` was
    /// inserted (or it was previously evicted and reinserted), `false`
    /// if it was already present.
    ///
    /// When the cap is reached, the oldest inserted entry is evicted to
    /// make room. Eviction never blocks the caller.
    pub(crate) fn insert_is_first(&self, name: &str) -> bool {
        // SAFETY: the only writer is this method; a poisoned set remains
        // logically consistent (atomic insert + bounded eviction preserve
        // the `set.len() <= cap` invariant). Continuing past poison only
        // affects diagnostic logging granularity, not correctness or
        // security.
        let mut guard = self
            .inner
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);

        if guard.set.contains(name) {
            return false;
        }
        // Cap enforcement: evict-then-insert keeps the invariant
        // `set.len() <= cap` even when the cap is `1`.
        if guard.set.len() >= guard.cap
            && let Some(evicted) = guard.order.pop_front()
        {
            guard.set.remove(&evicted);
        }
        let owned = name.to_owned();
        guard.set.insert(owned.clone());
        guard.order.push_back(owned);
        true
    }

    /// Test-only snapshot of the current size.
    #[cfg(test)]
    pub(crate) fn len(&self) -> usize {
        self.inner
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .set
            .len()
    }
}

impl Default for SeenIdentitySet {
    fn default() -> Self {
        Self::new()
    }
}

/// Shared state for the auth middleware.
///
/// `api_keys` uses [`ArcSwap`] so the SIGHUP handler can atomically
/// swap in a new key list without blocking in-flight requests.
#[allow(
    missing_debug_implementations,
    reason = "contains governor RateLimiter and JwksCache without Debug impls"
)]
#[non_exhaustive]
pub(crate) struct AuthState {
    /// Active set of API keys (hot-swappable).
    pub api_keys: ArcSwap<Vec<ApiKeyEntry>>,
    /// Optional per-IP post-failure rate limiter (consulted *after* auth fails).
    pub rate_limiter: Option<Arc<KeyedLimiter>>,
    /// Optional per-IP pre-auth abuse gate (consulted *before* password-hash work).
    /// mTLS-authenticated connections bypass this gate.
    pub pre_auth_limiter: Option<Arc<KeyedLimiter>>,
    #[cfg(feature = "oauth")]
    /// Optional JWKS cache for OAuth JWT validation.
    pub jwks_cache: Option<Arc<crate::oauth::JwksCache>>,
    /// Tracks identity names that have already been logged at INFO level.
    /// Subsequent auths for the same identity are logged at DEBUG.
    /// Bounded to prevent attacker-driven memory growth via churned
    /// mTLS subjects or OAuth `sub` claims (see [`SeenIdentitySet`]).
    pub seen_identities: SeenIdentitySet,
    /// Lightweight in-memory auth success/failure counters for diagnostics.
    pub counters: AuthCounters,
    /// Absolute URL of this server's RFC 9728 Protected Resource Metadata,
    /// advertised in the `WWW-Authenticate` challenge.
    ///
    /// RFC 9728 5.1 defines `resource_metadata` as a URL; emitting an
    /// absolute one lets a client resolve it without knowing the origin it
    /// was challenged from. `None` falls back to the well-known path, which
    /// stays correct for same-origin clients.
    pub resource_metadata_url: Option<String>,
}

impl AuthState {
    /// Atomically replace the API key list (lock-free, wait-free).
    ///
    /// New requests immediately see the updated keys.
    /// In-flight requests that already loaded the old list finish
    /// using it -- no torn reads.
    pub(crate) fn reload_keys(&self, keys: Vec<ApiKeyEntry>) {
        let count = keys.len();
        self.api_keys.store(Arc::new(keys));
        tracing::info!(keys = count, "API keys reloaded");
    }

    /// Snapshot auth counters for diagnostics and tests.
    #[must_use]
    pub(crate) fn counters_snapshot(&self) -> AuthCountersSnapshot {
        self.counters.snapshot()
    }

    /// Produce the admin-endpoint list of API keys (metadata only, no hashes).
    #[must_use]
    pub(crate) fn api_key_summaries(&self) -> Vec<ApiKeySummary> {
        self.api_keys
            .load()
            .iter()
            .map(|k| ApiKeySummary {
                name: k.name.clone(),
                role: k.role.clone(),
                expires_at: k.expires_at,
            })
            .collect()
    }

    /// Log auth success: INFO on first occurrence per identity, DEBUG after.
    ///
    /// Backed by [`SeenIdentitySet`], a bounded FIFO set that caps
    /// retained identities to prevent attacker-driven memory growth.
    /// FIFO (not LRU) is intentional: this cache de-duplicates INFO logs,
    /// not security state, so per-hit eviction-order mutation is not
    /// justified. See [`SeenIdentitySet`] for the full trade-off rationale.
    fn log_auth(&self, id: &AuthIdentity, method: &str) {
        self.counters.record_success(id.method);
        let first = self.seen_identities.insert_is_first(&id.name);
        if first {
            tracing::info!(name = %id.name, role = %id.role, "{method} authenticated");
        } else {
            tracing::debug!(name = %id.name, role = %id.role, "{method} authenticated");
        }
    }
}

/// Default auth rate limit: 30 attempts per minute per source IP.
// SAFETY: unwrap() is safe - literal 30 is provably non-zero (const-evaluated).
const DEFAULT_AUTH_RATE: NonZeroU32 = NonZeroU32::new(30).unwrap();

/// Apply an optional burst capacity to a quota. `None` keeps governor's
/// default (burst = rate). Zero values are rejected at config-validation
/// time; the `NonZeroU32` filter here is defensive only.
fn apply_burst(quota: governor::Quota, burst: Option<u32>) -> governor::Quota {
    match burst.and_then(NonZeroU32::new) {
        Some(b) => quota.allow_burst(b),
        None => quota,
    }
}

/// Create a post-failure rate limiter from config.
#[must_use]
pub(crate) fn build_rate_limiter(config: &RateLimitConfig) -> Arc<KeyedLimiter> {
    // Defense in depth: `serve()` and `serve_with_listener()` require a
    // `Validated<McpServerConfig>` and reject zero before startup, but
    // `auth::tests` construct limiters directly from raw `RateLimitConfig`
    // values to exercise limiter behavior without building a full server.
    let quota = governor::Quota::per_minute(
        NonZeroU32::new(config.max_attempts_per_minute).unwrap_or(DEFAULT_AUTH_RATE),
    );
    let quota = apply_burst(quota, config.burst);
    // Defense in depth: Phase-1 config validation rejects `0` upstream, but
    // tests can still exercise this helper directly with raw config values.
    let max_tracked_keys = NonZeroUsize::new(config.max_tracked_keys).unwrap_or(NonZeroUsize::MIN);
    Arc::new(BoundedKeyedLimiter::new_with_policy(
        quota,
        max_tracked_keys,
        config.idle_eviction,
        config.key_eviction_policy,
    ))
}

/// Create a pre-auth abuse-gate rate limiter from config.
///
/// Quota: `pre_auth_max_per_minute` if set, otherwise
/// `max_attempts_per_minute * 10` (capped at `u32::MAX`). The 10× factor
/// keeps the gate generous enough for honest retries while still bounding
/// attacker CPU on Argon2 verification.
#[must_use]
pub(crate) fn build_pre_auth_limiter(config: &RateLimitConfig) -> Arc<KeyedLimiter> {
    let resolved = config.pre_auth_max_per_minute.unwrap_or_else(|| {
        config
            .max_attempts_per_minute
            .saturating_mul(PRE_AUTH_DEFAULT_MULTIPLIER)
    });
    let quota =
        governor::Quota::per_minute(NonZeroU32::new(resolved).unwrap_or(DEFAULT_PRE_AUTH_RATE));
    let quota = apply_burst(quota, config.pre_auth_burst);
    // Defense in depth: Phase-1 config validation rejects `0` upstream, but
    // tests can still exercise this helper directly with raw config values.
    let max_tracked_keys = NonZeroUsize::new(config.max_tracked_keys).unwrap_or(NonZeroUsize::MIN);
    Arc::new(BoundedKeyedLimiter::new_with_policy(
        quota,
        max_tracked_keys,
        config.idle_eviction,
        config.key_eviction_policy,
    ))
}

/// Default multiplier applied to `max_attempts_per_minute` when the operator
/// does not set `pre_auth_max_per_minute` explicitly.
const PRE_AUTH_DEFAULT_MULTIPLIER: u32 = 10;

/// Default pre-auth abuse-gate rate (used only if both the configured value
/// and the multiplied fallback are zero, which `NonZeroU32::new` rejects).
// SAFETY: unwrap() is safe - literal 300 is provably non-zero (const-evaluated).
const DEFAULT_PRE_AUTH_RATE: NonZeroU32 = NonZeroU32::new(300).unwrap();

/// Parse an mTLS client certificate and extract an `AuthIdentity`.
///
/// Reads the Subject CN as the identity name. Falls back to the first
/// DNS SAN if CN is absent. The role is taken from the `MtlsConfig`.
#[must_use]
pub fn extract_mtls_identity(cert_der: &[u8], default_role: &str) -> Option<AuthIdentity> {
    let (_, cert) = X509Certificate::from_der(cert_der).ok()?;

    // Try CN from Subject first.
    let cn = cert
        .subject()
        .iter_common_name()
        .next()
        .and_then(|attr| attr.as_str().ok())
        .map(String::from);

    // Fall back to first DNS SAN.
    let name = cn.or_else(|| {
        cert.subject_alternative_name()
            .ok()
            .flatten()
            .and_then(|san| {
                #[allow(
                    clippy::wildcard_enum_match_arm,
                    reason = "x509-parser GeneralName is a large external enum; only DNSName is meaningful here"
                )]
                san.value.general_names.iter().find_map(|gn| match gn {
                    GeneralName::DNSName(dns) => Some((*dns).to_owned()),
                    _ => None,
                })
            })
    })?;

    // Reject identities with characters unsafe for logging and RBAC matching.
    if !name
        .chars()
        .all(|c| c.is_alphanumeric() || matches!(c, '-' | '.' | '_' | '@'))
    {
        tracing::warn!(cn = %name, "mTLS identity rejected: invalid characters in CN/SAN");
        return None;
    }

    Some(AuthIdentity {
        name,
        role: default_role.to_owned(),
        method: AuthMethod::MtlsCertificate,
        raw_token: None,
        sub: None,
    })
}

/// Extract the bearer token from an `Authorization` header value.
///
/// Implements RFC 7235 §2.1: the auth-scheme token is **case-insensitive**.
/// `Bearer`, `bearer`, `BEARER`, and `BeArEr` all parse equivalently. Any
/// leading whitespace between the scheme and the token is trimmed (per
/// RFC 7235 the separator is one or more SP characters; we accept the
/// common single-space form plus tolerate extras).
///
/// Returns `None` if the header value:
/// - does not contain a space (no scheme/credentials boundary), or
/// - uses a scheme other than `Bearer` (case-insensitively), or
/// - carries a credential containing embedded whitespace.
///
/// # Why whitespace only, and not full `token68`
///
/// RFC 7235 §2.1 defines the credential as `token68`, which excludes
/// whitespace. Accepting an embedded SP/HTAB creates a parser differential
/// against a fronting proxy that splits on any whitespace.
///
/// Enforcing the whole `token68` character class would be a breaking
/// change: [`ApiKeyEntry::new`] accepts an arbitrary caller-supplied hash
/// and [`verify_bearer_token`] verifies the raw presented string, so
/// consumers may have hashed opaque tokens containing punctuation outside
/// `token68`. Those must keep authenticating -- do not "complete" this
/// check without a major-version note.
///
/// ASCII semantics suffice because [`http::HeaderValue::to_str`] rejects
/// every non-visible byte before this helper runs.
fn extract_bearer(value: &str) -> Option<&str> {
    let (scheme, rest) = value.split_once(' ')?;
    if !scheme.eq_ignore_ascii_case("Bearer") {
        return None;
    }
    let token = rest.trim_start_matches(' ');
    if token.is_empty() || token.bytes().any(|b| b.is_ascii_whitespace()) {
        return None;
    }
    Some(token)
}

/// Verify a bearer token against configured API keys.
///
/// Argon2id verification is CPU-intensive, so this should be called via
/// `spawn_blocking`. Returns the matching identity if the token is valid.
///
/// # Timing-side-channel resistance
///
/// Always performs **exactly one Argon2id verification per configured key**,
/// regardless of:
///
/// * which slot (if any) matches the presented token, or
/// * whether a key has expired.
///
/// Expired and post-match slots are verified against an internal dummy PHC hash,
/// a fixed Argon2id PHC string with the same cost parameters as the real
/// hashes. This bounds the timing observable to "one Argon2 per configured
/// key" regardless of which (if any) slot held the matching credential,
/// closing the first-match latency oracle (CWE-208) and the expired-slot
/// timing leak.
///
/// `subtle::ConstantTimeEq` folds each slot's match bit into the running
/// result without comparing the token bytes in short-circuiting fashion.
///
/// The guarantee this function provides is the Argon2 count, not full
/// branchlessness: selecting `verify_against` and recording `matched_index`
/// are both ordinary data-dependent branches. They are cheap, predictable,
/// and operate on locals, so they are dwarfed by the Argon2id verification
/// that dominates every iteration -- but the timing claim stops at
/// "one verification per configured key". Do not read this as a
/// constant-time selection routine.
///
/// # Panics
///
/// Panics if the internal dummy PHC hash cannot be parsed as an Argon2id PHC string.
/// This is impossible by construction: the static is generated by
/// [`argon2::Argon2::hash_password`] which always emits a valid PHC string.
#[must_use]
pub fn verify_bearer_token(token: &str, keys: &[ApiKeyEntry]) -> Option<AuthIdentity> {
    use subtle::ConstantTimeEq as _;

    let now = chrono::Utc::now();
    #[allow(
        clippy::expect_used,
        reason = "DUMMY_PHC_HASH is a static LazyLock built from a fixed Argon2id PHC string by construction; PasswordHash::new on it is infallible. See DUMMY_PHC_HASH definition."
    )]
    let dummy_hash = PasswordHash::new(&DUMMY_PHC_HASH)
        .expect("DUMMY_PHC_HASH is a valid Argon2id PHC string by construction");

    let mut matched_index: usize = usize::MAX;
    let mut any_match: u8 = 0;

    for (idx, key) in keys.iter().enumerate() {
        let expired = key.expires_at.is_some_and(|exp| exp.as_datetime() < &now);

        let real_hash = PasswordHash::new(&key.hash);
        let verify_against = match (&real_hash, expired, any_match) {
            (Ok(h), false, 0) => h,
            _ => &dummy_hash,
        };

        let slot_ok = u8::from(
            Argon2::default()
                .verify_password(token.as_bytes(), verify_against)
                .is_ok(),
        );

        let real_match = slot_ok & u8::from(!expired) & u8::from(real_hash.is_ok());
        let first_real_match = real_match & (1 - any_match);
        if first_real_match.ct_eq(&1).into() {
            matched_index = idx;
        }
        any_match |= real_match;
    }

    if any_match == 0 {
        return None;
    }
    let key = keys.get(matched_index)?;
    Some(AuthIdentity {
        name: key.name.clone(),
        role: key.role.clone(),
        method: AuthMethod::BearerToken,
        raw_token: None,
        sub: None,
    })
}

/// Fixed Argon2id PHC hash used as a constant-time placeholder when an
/// API-key slot is expired, malformed, or follows the matching slot.
///
/// Generated once on first access using the same default Argon2 cost
/// parameters as live verifications, so the dummy verify takes
/// indistinguishable wall time from a real one. The plaintext
/// (`"rmcp-server-kit-dummy"`) and the fixed salt are unrelated to any
/// real credential — randomness is unnecessary because this hash is
/// only ever compared against attacker-supplied input on slots that
/// will be discarded regardless of match result. Argon2's work factor is
/// set by the PHC `m`/`t`/`p` parameters, not by the salt value, so a
/// fixed salt costs exactly what a random one would;
/// `dummy_and_real_hashes_share_cost_parameters` pins that equivalence.
static DUMMY_PHC_HASH: LazyLock<String> = LazyLock::new(|| {
    #[allow(
        clippy::expect_used,
        reason = "Argon2::default() over a fixed plaintext and a fixed 16-byte salt is infallible; it fails only on invalid params or salt length, both constants here"
    )]
    Argon2::default()
        .hash_password_with_salt(b"rmcp-server-kit-dummy", &[0u8; 16])
        .expect("Argon2 default params hash a fixed plaintext")
        .to_string()
});

/// Generate a new API key: 256-bit random token + Argon2id hash.
///
/// Returns `(plaintext_token, argon2id_hash_phc_string)`.
/// The plaintext is shown once to the user and never stored.
///
/// # Errors
///
/// Returns an error if Argon2id hashing fails (should not happen with valid
/// inputs, but we avoid panicking).
pub fn generate_api_key() -> Result<(String, String), RmcpServerKitError> {
    let mut token_bytes = [0u8; 32];
    rand::fill(&mut token_bytes);
    let token = URL_SAFE_NO_PAD.encode(token_bytes);

    let mut salt_bytes = [0u8; 16];
    rand::fill(&mut salt_bytes);
    let hash = Argon2::default()
        .hash_password_with_salt(token.as_bytes(), &salt_bytes)
        .map_err(|e| RmcpServerKitError::Internal(format!("argon2id hashing failed: {e}")))?
        .to_string();

    Ok((token, hash))
}

fn build_www_authenticate_value(
    resource_metadata: Option<&str>,
    failure: AuthFailureClass,
) -> String {
    let (error, error_description) = failure.bearer_error();
    if let Some(url) = resource_metadata {
        return format!(
            "Bearer resource_metadata=\"{url}\", error=\"{error}\", error_description=\"{error_description}\""
        );
    }
    format!("Bearer error=\"{error}\", error_description=\"{error_description}\"")
}

fn auth_method_label(method: AuthMethod) -> &'static str {
    match method {
        AuthMethod::MtlsCertificate => "mTLS",
        AuthMethod::BearerToken => "bearer token",
        AuthMethod::OAuthJwt => "OAuth JWT",
    }
}

#[cfg_attr(
    not(feature = "oauth"),
    allow(
        unused_variables,
        reason = "`state` is only read to decide whether to advertise OAuth \
                  protected-resource metadata; without the `oauth` feature that \
                  decision is a compile-time `false`"
    )
)]
fn unauthorized_response(state: &AuthState, failure_class: AuthFailureClass) -> Response {
    #[cfg(feature = "oauth")]
    let advertise_resource_metadata = state.jwks_cache.is_some();
    #[cfg(not(feature = "oauth"))]
    let advertise_resource_metadata = false;

    let resource_metadata = advertise_resource_metadata.then(|| {
        state
            .resource_metadata_url
            .as_deref()
            .unwrap_or("/.well-known/oauth-protected-resource")
    });
    let challenge = build_www_authenticate_value(resource_metadata, failure_class);
    (
        StatusCode::UNAUTHORIZED,
        [(header::WWW_AUTHENTICATE, challenge)],
        failure_class.response_body(),
    )
        .into_response()
}

// cancel-safe: no shared-state mutation. The Argon2 verification is offloaded
// to `spawn_blocking`; dropping its `JoinHandle` on cancellation detaches the
// task (the hash completes off-task, harmlessly) rather than tearing partial
// state. The OAuth branch delegates to `validate_token_with_reason`, which is
// itself cancel-safe (read-only JWKS lookup + pure claim checks).
async fn authenticate_bearer_identity(
    state: &AuthState,
    token: &str,
) -> Result<AuthIdentity, AuthFailureClass> {
    let mut failure_class = AuthFailureClass::MissingCredential;

    #[cfg(feature = "oauth")]
    if let Some(ref cache) = state.jwks_cache
        && crate::oauth::looks_like_jwt(token)
    {
        match cache.validate_token_with_reason(token).await {
            Ok(mut id) => {
                id.raw_token = Some(SecretString::from(token.to_owned()));
                return Ok(id);
            }
            Err(crate::oauth::JwtValidationFailure::Expired) => {
                failure_class = AuthFailureClass::ExpiredCredential;
            }
            Err(crate::oauth::JwtValidationFailure::Invalid) => {
                failure_class = AuthFailureClass::InvalidCredential;
            }
        }
    }

    let token = token.to_owned();
    let keys = state.api_keys.load_full(); // Arc clone, lock-free

    // Argon2id is CPU-bound - offload to blocking thread pool.
    let identity = tokio::task::spawn_blocking(move || verify_bearer_token(&token, &keys))
        .await
        .ok()
        .flatten();

    if let Some(id) = identity {
        return Ok(id);
    }

    if failure_class == AuthFailureClass::MissingCredential {
        failure_class = AuthFailureClass::InvalidCredential;
    }

    Err(failure_class)
}

/// Consult the pre-auth abuse gate for the given peer.
///
/// Returns `Some(response)` if the request should be rejected (limiter
/// configured AND quota exhausted for this source IP). Returns `None`
/// otherwise (limiter absent, peer address unknown, or quota available),
/// in which case the caller should proceed with credential verification.
///
/// Side effects on rejection: increments the `pre_auth_gate` failure
/// counter and emits a warn-level log. mTLS-authenticated requests must
/// be admitted by the caller *before* invoking this helper.
fn pre_auth_gate(state: &AuthState, client_key: Option<&RateLimitKey>) -> Option<Response> {
    let limiter = state.pre_auth_limiter.as_ref()?;
    let key = client_key?;
    match limiter.check_key_detailed(key) {
        Ok(()) => None,
        Err(BoundedLimiterDeny::RateLimited(wait)) => {
            state.counters.record_failure(AuthFailureClass::PreAuthGate);
            tracing::warn!(
                rate_limit_key = %key,
                "auth rate limited by pre-auth gate (request rejected before credential verification)"
            );
            Some(
                RmcpServerKitError::RateLimitedFor {
                    message: "too many unauthenticated requests from this source".into(),
                    retry_after: wait,
                }
                .into_response(),
            )
        }
        Err(BoundedLimiterDeny::CapacityFull) => {
            tracing::warn!(
                rate_limit_key = %key,
                "auth pre-auth gate rejected unseen key because tracked-key capacity is full"
            );
            Some(
                (
                    StatusCode::SERVICE_UNAVAILABLE,
                    "rate limiter capacity exhausted",
                )
                    .into_response(),
            )
        }
    }
}

#[cfg_attr(
    not(feature = "metrics"),
    allow(
        unused_variables,
        reason = "`extensions` is read only to record the \
                  `rmcp_server_kit_rate_limited_total` metric; without the \
                  `metrics` feature there is no recording site"
    )
)]
fn post_failure_rate_limit_response(
    limiter: &KeyedLimiter,
    key: &RateLimitKey,
    extensions: &axum::http::Extensions,
) -> Option<Response> {
    match limiter.check_key_detailed(key) {
        Ok(()) => None,
        Err(BoundedLimiterDeny::RateLimited(wait)) => {
            #[cfg(feature = "metrics")]
            crate::metrics::record_rate_limit_deny(extensions, "auth_post");
            tracing::warn!(rate_limit_key = %key, "auth rate limited after repeated failures");
            Some(
                RmcpServerKitError::RateLimitedFor {
                    message: "too many failed authentication attempts".into(),
                    retry_after: wait,
                }
                .into_response(),
            )
        }
        Err(BoundedLimiterDeny::CapacityFull) => {
            tracing::warn!(
                rate_limit_key = %key,
                "auth post-failure limiter rejected unseen key because tracked-key capacity is full"
            );
            Some(
                (
                    StatusCode::SERVICE_UNAVAILABLE,
                    "rate limiter capacity exhausted",
                )
                    .into_response(),
            )
        }
    }
}

/// Axum middleware that enforces authentication.
///
/// Tries authentication methods in priority order:
/// 1. mTLS client certificate identity (populated by TLS acceptor)
/// 2. Bearer token from `Authorization` header
///
/// Failed authentication attempts are rate-limited per source IP.
/// Successful authentications do not consume rate limit budget.
// cancel-safe: `TimeoutLayer` may drop this future, but limiter mutations are
// deliberate attempt accounting: pre-auth prices bearer/JWT verification,
// post-failure prices failed auth, and identity extensions die with the request.
pub(crate) async fn auth_middleware(
    state: Arc<AuthState>,
    req: Request<Body>,
    next: Next,
) -> Response {
    // Extract the mTLS identity from ConnectInfo (TLS / mTLS:
    // ConnectInfo<TlsConnInfo> carries the verified identity directly on
    // the connection — no shared map, no port-reuse aliasing) and the
    // rate-limit key (resolved client IP when trusted-forwarder mode is
    // active, else the direct peer; see transport::limiter_client_ip).
    let tls_info = req.extensions().get::<ConnectInfo<TlsConnInfo>>().cloned();
    // Resolved only when a limiter will actually consult it, so servers
    // with no rate limiting never trip the unattributed-fallback warning.
    let client_key = (state.pre_auth_limiter.is_some() || state.rate_limiter.is_some())
        .then(|| crate::transport::limiter_client_key(req.extensions()));

    // 1. Try mTLS identity (extracted by the TLS acceptor during handshake
    //    and attached to the connection itself).
    //
    //    mTLS connections bypass the pre-auth abuse gate below: the TLS
    //    handshake already performed expensive crypto with a verified peer,
    //    so we trust them not to be a CPU-spray attacker.
    if let Some(id) = tls_info.and_then(|ci| ci.0.identity) {
        state.log_auth(&id, "mTLS");
        let mut req = req;
        req.extensions_mut().insert(id);
        return next.run(req).await;
    }

    // 2. Pre-auth abuse gate: rejects CPU-spray attacks BEFORE the Argon2id
    //    verification path runs. Keyed by source IP. mTLS connections (above)
    //    are exempt; this gate only protects the bearer/JWT verification path.
    if let Some(blocked) = pre_auth_gate(&state, client_key.as_ref()) {
        #[cfg(feature = "metrics")]
        crate::metrics::record_rate_limit_deny(req.extensions(), "auth_pre");
        return blocked;
    }

    let failure_class = if let Some(value) = req.headers().get(header::AUTHORIZATION) {
        match value.to_str().ok().and_then(extract_bearer) {
            Some(token) => match authenticate_bearer_identity(&state, token).await {
                Ok(id) => {
                    state.log_auth(&id, auth_method_label(id.method));
                    let mut req = req;
                    req.extensions_mut().insert(id);
                    return next.run(req).await;
                }
                Err(class) => class,
            },
            None => AuthFailureClass::InvalidCredential,
        }
    } else {
        AuthFailureClass::MissingCredential
    };

    tracing::warn!(failure_class = %failure_class.as_str(), "auth failed");

    // Rate limit check (applied after auth failure only).
    // Successful authentications do not consume rate limit budget.
    if let (Some(limiter), Some(key)) = (&state.rate_limiter, client_key.as_ref())
        && let Some(resp) = post_failure_rate_limit_response(limiter, key, req.extensions())
    {
        if resp.status() == StatusCode::TOO_MANY_REQUESTS {
            state.counters.record_failure(AuthFailureClass::RateLimited);
        }
        return resp;
    }

    state.counters.record_failure(failure_class);
    unauthorized_response(&state, failure_class)
}

#[cfg(test)]
mod tests {
    use std::net::IpAddr;

    use super::*;
    use crate::transport::RateLimitKey;

    /// A PHC string produced by **argon2 0.5.3** through the same code path as
    /// [`generate_api_key`] (16 salt bytes, `Argon2::default()`).
    ///
    /// Pinned so the argon2 0.6 upgrade cannot silently invalidate credentials
    /// that are already deployed: if this stops verifying, every stored API key
    /// stops working. Captured before the upgrade and asserted after it.
    const ARGON2_0_5_TOKEN: &str = "golden-vector-token-0p5p3";
    const ARGON2_0_5_HASH: &str = "$argon2id$v=19$m=19456,t=2,p=1$BwcHBwcHBwcHBwcHBwcHBw$spS8B9AhHG1LikfhGlssVMfP8mq37+8/mXnl98ps0NU";

    #[test]
    fn argon2_0_5_produced_hash_still_verifies() {
        let parsed =
            PasswordHash::new(ARGON2_0_5_HASH).expect("a 0.5-era PHC string must still parse");
        Argon2::default()
            .verify_password(ARGON2_0_5_TOKEN.as_bytes(), &parsed)
            .expect("already-deployed API keys must keep verifying across the argon2 upgrade");
    }

    /// The dummy hash burned on a miss must cost the same as a real one.
    ///
    /// Argon2 work is set by the PHC parameters, not the salt, so asserting the
    /// dummy and a freshly generated key share `argon2id`, `v=19` and identical
    /// `m`/`t`/`p` pins the constant-time property without a flaky wall-clock
    /// measurement. Forcing `DUMMY_PHC_HASH` here also surfaces a `LazyLock`
    /// panic in CI rather than at the first production auth.
    #[test]
    fn dummy_and_real_hashes_share_cost_parameters() {
        let (_token, real_hash) = generate_api_key().expect("key generation must succeed");
        let real = PasswordHash::new(&real_hash).expect("generated hash must parse");
        let dummy = PasswordHash::new(&DUMMY_PHC_HASH).expect("dummy hash must parse");

        assert_eq!(dummy.algorithm, real.algorithm, "algorithm must match");
        assert_eq!(dummy.version, real.version, "PHC version must match");
        assert_eq!(
            dummy.params, real.params,
            "m/t/p must match or the dummy no longer costs what a real verification costs"
        );
    }

    #[test]
    fn generate_and_verify_api_key() {
        let (token, hash) = generate_api_key().unwrap();

        // Token is 43 chars (256-bit base64url, no padding)
        assert_eq!(token.len(), 43);

        // Hash is a valid PHC string
        assert!(hash.starts_with("$argon2id$"));

        // Verification succeeds with correct token
        let keys = vec![ApiKeyEntry {
            name: "test".into(),
            hash,
            role: "viewer".into(),
            expires_at: None,
        }];
        let id = verify_bearer_token(&token, &keys);
        assert!(id.is_some());
        let id = id.unwrap();
        assert_eq!(id.name, "test");
        assert_eq!(id.role, "viewer");
        assert_eq!(id.method, AuthMethod::BearerToken);
    }

    #[test]
    fn wrong_token_rejected() {
        let (_token, hash) = generate_api_key().unwrap();
        let keys = vec![ApiKeyEntry {
            name: "test".into(),
            hash,
            role: "viewer".into(),
            expires_at: None,
        }];
        assert!(verify_bearer_token("wrong-token", &keys).is_none());
    }

    #[test]
    fn expired_key_rejected() {
        let (token, hash) = generate_api_key().unwrap();
        let keys = vec![ApiKeyEntry {
            name: "test".into(),
            hash,
            role: "viewer".into(),
            expires_at: Some(RfcTimestamp::parse("2020-01-01T00:00:00Z").unwrap()),
        }];
        assert!(verify_bearer_token(&token, &keys).is_none());
    }

    #[test]
    fn match_in_last_slot_still_authenticates() {
        let (token, hash) = generate_api_key().unwrap();
        let (_other_token, other_hash) = generate_api_key().unwrap();
        let keys = vec![
            ApiKeyEntry {
                name: "first".into(),
                hash: other_hash.clone(),
                role: "viewer".into(),
                expires_at: None,
            },
            ApiKeyEntry {
                name: "second".into(),
                hash: other_hash,
                role: "viewer".into(),
                expires_at: None,
            },
            ApiKeyEntry {
                name: "match".into(),
                hash,
                role: "ops".into(),
                expires_at: None,
            },
        ];
        let id = verify_bearer_token(&token, &keys).expect("last-slot match must authenticate");
        assert_eq!(id.name, "match");
        assert_eq!(id.role, "ops");
    }

    #[test]
    fn expired_slot_before_valid_match_does_not_short_circuit() {
        let (token, hash) = generate_api_key().unwrap();
        let (_, other_hash) = generate_api_key().unwrap();
        let keys = vec![
            ApiKeyEntry {
                name: "expired".into(),
                hash: other_hash,
                role: "viewer".into(),
                expires_at: Some(RfcTimestamp::parse("2020-01-01T00:00:00Z").unwrap()),
            },
            ApiKeyEntry {
                name: "valid".into(),
                hash,
                role: "ops".into(),
                expires_at: None,
            },
        ];
        let id = verify_bearer_token(&token, &keys)
            .expect("valid slot following an expired slot must authenticate");
        assert_eq!(id.name, "valid");
    }

    #[test]
    fn malformed_hash_slot_does_not_short_circuit() {
        let (token, hash) = generate_api_key().unwrap();
        let keys = vec![
            ApiKeyEntry {
                name: "broken".into(),
                hash: "this-is-not-a-phc-string".into(),
                role: "viewer".into(),
                expires_at: None,
            },
            ApiKeyEntry {
                name: "valid".into(),
                hash,
                role: "ops".into(),
                expires_at: None,
            },
        ];
        let id = verify_bearer_token(&token, &keys)
            .expect("valid slot following a malformed-hash slot must authenticate");
        assert_eq!(id.name, "valid");
    }

    // Regression tests for H3 (api_key_expires_at_fail_open).
    //
    // Prior to 1.6.0 the runtime expiry check used a chained
    // `if let Some(_) && let Ok(exp) = parse(_) && exp < now` which
    // silently fell through on parse error, letting a key with
    // `expires_at = "not-a-date"` authenticate forever. These tests
    // pin the type-system fix: malformed RFC 3339 is rejected at
    // deserialization time (no `RfcTimestamp` can ever be malformed),
    // and the runtime check is a pure comparison with no parse path.

    #[test]
    fn rfc_timestamp_parse_rejects_malformed() {
        for bad in [
            "not-a-date",
            "",
            "2025-13-01T00:00:00Z", // month 13
            "2025-01-32T00:00:00Z", // day 32
            "2025-01-01T00:00:00",  // missing offset
            "01/01/2025",           // wrong format
            "2025-01-01T25:00:00Z", // hour 25
        ] {
            assert!(
                RfcTimestamp::parse(bad).is_err(),
                "RfcTimestamp::parse must reject {bad:?}"
            );
        }
    }

    #[test]
    fn rfc_timestamp_parse_accepts_valid() {
        for good in [
            "2025-01-01T00:00:00Z",
            "2025-01-01T00:00:00+00:00",
            "2025-12-31T23:59:59-08:00",
            "2099-01-01T00:00:00.123456789Z",
        ] {
            assert!(
                RfcTimestamp::parse(good).is_ok(),
                "RfcTimestamp::parse must accept {good:?}"
            );
        }
    }

    #[test]
    fn api_key_entry_deserialize_rejects_malformed_expires_at() {
        // TOML with a malformed expires_at must fail to deserialize.
        // This is the load-time defense: a typo in auth.toml aborts
        // config load with a clear serde error, instead of producing
        // a key that authenticates forever (the H3 fail-open).
        let toml = r#"
            name = "bad-key"
            hash = "$argon2id$v=19$m=19456,t=2,p=1$c2FsdA$h4sh"
            role = "viewer"
            expires_at = "not-a-date"
        "#;
        let result: Result<ApiKeyEntry, _> = toml::from_str(toml);
        assert!(
            result.is_err(),
            "deserialization must reject malformed expires_at"
        );
    }

    #[test]
    fn api_key_entry_deserialize_accepts_valid_expires_at() {
        let toml = r#"
            name = "good-key"
            hash = "$argon2id$v=19$m=19456,t=2,p=1$c2FsdA$h4sh"
            role = "viewer"
            expires_at = "2099-01-01T00:00:00Z"
        "#;
        let entry: ApiKeyEntry = toml::from_str(toml).expect("valid RFC 3339 must deserialize");
        assert!(entry.expires_at.is_some());
    }

    #[test]
    fn api_key_entry_deserialize_accepts_missing_expires_at() {
        // Omitting expires_at must continue to mean "no expiry"; this
        // is the documented contract and must survive the H3 fix.
        let toml = r#"
            name = "eternal-key"
            hash = "$argon2id$v=19$m=19456,t=2,p=1$c2FsdA$h4sh"
            role = "viewer"
        "#;
        let entry: ApiKeyEntry = toml::from_str(toml).expect("missing expires_at must deserialize");
        assert!(entry.expires_at.is_none());
    }

    #[test]
    fn mtls_crl_deny_on_unavailable_defaults_to_fail_closed() {
        // Every in-crate test helper builds MtlsConfig via a struct literal,
        // which bypasses serde defaults entirely. Only a deserialization from
        // TOML that omits the key exercises the shipped default.
        let toml = r#"
            ca_cert_path = "/etc/certs/clients-ca.pem"
        "#;
        let cfg: MtlsConfig = toml::from_str(toml).expect("minimal mtls config must deserialize");
        assert!(
            cfg.crl_deny_on_unavailable,
            "omitting crl_deny_on_unavailable must fail closed (RFC 5280 6.3)"
        );
    }

    #[test]
    fn mtls_crl_deny_on_unavailable_opt_out_is_honoured() {
        let toml = r#"
            ca_cert_path = "/etc/certs/clients-ca.pem"
            crl_deny_on_unavailable = false
        "#;
        let cfg: MtlsConfig = toml::from_str(toml).expect("opt-out config must deserialize");
        assert!(
            !cfg.crl_deny_on_unavailable,
            "an explicit false must still select fail-open"
        );
    }

    #[test]
    fn try_with_expiry_rejects_malformed() {
        let entry = ApiKeyEntry::new("k", "hash", "viewer");
        assert!(entry.try_with_expiry("not-a-date").is_err());
    }

    #[test]
    fn try_with_expiry_accepts_valid() {
        let entry = ApiKeyEntry::new("k", "hash", "viewer")
            .try_with_expiry("2099-01-01T00:00:00Z")
            .expect("valid RFC 3339 must be accepted");
        assert!(entry.expires_at.is_some());
    }

    #[test]
    fn api_key_summary_serializes_expires_at_as_rfc3339() {
        // The admin endpoint wire format is `{"expires_at": "RFC 3339 str"}`.
        // Pinning this prevents an accidental serialization-format change
        // (e.g. chrono's debug form, a Unix timestamp) that would silently
        // break operator tooling that parses these payloads.
        let summary = ApiKeySummary {
            name: "k".into(),
            role: "viewer".into(),
            expires_at: Some(RfcTimestamp::parse("2030-01-01T00:00:00Z").unwrap()),
        };
        let json = serde_json::to_string(&summary).unwrap();
        assert!(
            json.contains(r#""expires_at":"2030-01-01T00:00:00+00:00""#),
            "wire format regressed: {json}"
        );
    }

    #[test]
    fn future_expiry_accepted() {
        let (token, hash) = generate_api_key().unwrap();
        let keys = vec![ApiKeyEntry {
            name: "test".into(),
            hash,
            role: "viewer".into(),
            expires_at: Some(RfcTimestamp::parse("2099-01-01T00:00:00Z").unwrap()),
        }];
        assert!(verify_bearer_token(&token, &keys).is_some());
    }

    #[test]
    fn multiple_keys_first_match_wins() {
        let (token, hash) = generate_api_key().unwrap();
        let keys = vec![
            ApiKeyEntry {
                name: "wrong".into(),
                hash: "$argon2id$v=19$m=19456,t=2,p=1$invalid$invalid".into(),
                role: "ops".into(),
                expires_at: None,
            },
            ApiKeyEntry {
                name: "correct".into(),
                hash,
                role: "deploy".into(),
                expires_at: None,
            },
        ];
        let id = verify_bearer_token(&token, &keys).unwrap();
        assert_eq!(id.name, "correct");
        assert_eq!(id.role, "deploy");
    }

    #[test]
    fn rate_limiter_allows_within_quota() {
        let config = RateLimitConfig {
            max_attempts_per_minute: 5,
            pre_auth_max_per_minute: None,
            max_tracked_keys: default_max_tracked_keys(),
            idle_eviction: default_idle_eviction(),
            burst: None,
            pre_auth_burst: None,
            key_eviction_policy: KeyEvictionPolicy::default(),
        };
        let limiter = build_rate_limiter(&config);
        let ip = RateLimitKey::Ip("10.0.0.1".parse::<IpAddr>().unwrap());

        // First 5 should succeed.
        for _ in 0..5 {
            assert!(limiter.check_key(&ip).is_ok());
        }
        // 6th should fail.
        assert!(limiter.check_key(&ip).is_err());
    }

    #[test]
    fn rate_limiter_separate_ips() {
        let config = RateLimitConfig {
            max_attempts_per_minute: 2,
            pre_auth_max_per_minute: None,
            max_tracked_keys: default_max_tracked_keys(),
            idle_eviction: default_idle_eviction(),
            burst: None,
            pre_auth_burst: None,
            key_eviction_policy: KeyEvictionPolicy::default(),
        };
        let limiter = build_rate_limiter(&config);
        let ip1 = RateLimitKey::Ip("10.0.0.1".parse::<IpAddr>().unwrap());
        let ip2 = RateLimitKey::Ip("10.0.0.2".parse::<IpAddr>().unwrap());

        // Exhaust ip1's quota.
        assert!(limiter.check_key(&ip1).is_ok());
        assert!(limiter.check_key(&ip1).is_ok());
        assert!(limiter.check_key(&ip1).is_err());

        // ip2 should still have quota.
        assert!(limiter.check_key(&ip2).is_ok());
    }

    #[test]
    fn extract_mtls_identity_from_cn() {
        // Generate a cert with explicit CN.
        let mut params = rcgen::CertificateParams::new(vec!["test-client.local".into()]).unwrap();
        params.distinguished_name = rcgen::DistinguishedName::new();
        params
            .distinguished_name
            .push(rcgen::DnType::CommonName, "test-client");
        let cert = params
            .self_signed(&rcgen::KeyPair::generate().unwrap())
            .unwrap();
        let der = cert.der();

        let id = extract_mtls_identity(der, "ops").unwrap();
        assert_eq!(id.name, "test-client");
        assert_eq!(id.role, "ops");
        assert_eq!(id.method, AuthMethod::MtlsCertificate);
    }

    #[test]
    fn extract_mtls_identity_falls_back_to_san() {
        // Cert with no CN but has a DNS SAN.
        let mut params =
            rcgen::CertificateParams::new(vec!["san-only.example.com".into()]).unwrap();
        params.distinguished_name = rcgen::DistinguishedName::new();
        // No CN set - should fall back to DNS SAN.
        let cert = params
            .self_signed(&rcgen::KeyPair::generate().unwrap())
            .unwrap();
        let der = cert.der();

        let id = extract_mtls_identity(der, "viewer").unwrap();
        assert_eq!(id.name, "san-only.example.com");
        assert_eq!(id.role, "viewer");
    }

    #[test]
    fn extract_mtls_identity_invalid_der() {
        assert!(extract_mtls_identity(b"not-a-cert", "viewer").is_none());
    }

    // -- auth_middleware integration tests --

    use axum::{
        body::Body,
        http::{Request, StatusCode},
    };
    use tower::ServiceExt as _;

    fn auth_router(state: Arc<AuthState>) -> axum::Router {
        axum::Router::new()
            .route("/mcp", axum::routing::post(|| async { "ok" }))
            .layer(axum::middleware::from_fn(move |req, next| {
                let s = Arc::clone(&state);
                auth_middleware(s, req, next)
            }))
    }

    fn test_auth_state(keys: Vec<ApiKeyEntry>) -> Arc<AuthState> {
        Arc::new(AuthState {
            api_keys: ArcSwap::new(Arc::new(keys)),
            rate_limiter: None,
            pre_auth_limiter: None,
            #[cfg(feature = "oauth")]
            jwks_cache: None,
            seen_identities: SeenIdentitySet::new(),
            counters: AuthCounters::default(),
            resource_metadata_url: None,
        })
    }

    #[tokio::test]
    async fn middleware_rejects_no_credentials() {
        let state = test_auth_state(vec![]);
        let app = auth_router(Arc::clone(&state));
        let req = Request::builder()
            .method(axum::http::Method::POST)
            .uri("/mcp")
            .body(Body::empty())
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
        let challenge = resp
            .headers()
            .get(header::WWW_AUTHENTICATE)
            .unwrap()
            .to_str()
            .unwrap();
        assert!(challenge.contains("error=\"invalid_request\""));

        let counters = state.counters_snapshot();
        assert_eq!(counters.failure_missing_credential, 1);
    }

    #[tokio::test]
    async fn middleware_accepts_valid_bearer() {
        let (token, hash) = generate_api_key().unwrap();
        let keys = vec![ApiKeyEntry {
            name: "test-key".into(),
            hash,
            role: "ops".into(),
            expires_at: None,
        }];
        let state = test_auth_state(keys);
        let app = auth_router(Arc::clone(&state));
        let req = Request::builder()
            .method(axum::http::Method::POST)
            .uri("/mcp")
            .header("authorization", format!("Bearer {token}"))
            .body(Body::empty())
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);

        let counters = state.counters_snapshot();
        assert_eq!(counters.success_bearer, 1);
    }

    #[tokio::test]
    async fn middleware_rejects_wrong_bearer() {
        let (_token, hash) = generate_api_key().unwrap();
        let keys = vec![ApiKeyEntry {
            name: "test-key".into(),
            hash,
            role: "ops".into(),
            expires_at: None,
        }];
        let state = test_auth_state(keys);
        let app = auth_router(Arc::clone(&state));
        let req = Request::builder()
            .method(axum::http::Method::POST)
            .uri("/mcp")
            .header("authorization", "Bearer wrong-token-here")
            .body(Body::empty())
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
        let challenge = resp
            .headers()
            .get(header::WWW_AUTHENTICATE)
            .unwrap()
            .to_str()
            .unwrap();
        assert!(challenge.contains("error=\"invalid_token\""));

        let counters = state.counters_snapshot();
        assert_eq!(counters.failure_invalid_credential, 1);
    }

    #[tokio::test]
    async fn middleware_rate_limits() {
        let state = Arc::new(AuthState {
            api_keys: ArcSwap::new(Arc::new(vec![])),
            rate_limiter: Some(build_rate_limiter(&RateLimitConfig {
                max_attempts_per_minute: 1,
                pre_auth_max_per_minute: None,
                max_tracked_keys: default_max_tracked_keys(),
                idle_eviction: default_idle_eviction(),
                burst: None,
                pre_auth_burst: None,
                key_eviction_policy: KeyEvictionPolicy::default(),
            })),
            pre_auth_limiter: None,
            #[cfg(feature = "oauth")]
            jwks_cache: None,
            seen_identities: SeenIdentitySet::new(),
            counters: AuthCounters::default(),
            resource_metadata_url: None,
        });
        let app = auth_router(state);

        // First request: UNAUTHORIZED (no credentials, but not rate limited)
        let req = Request::builder()
            .method(axum::http::Method::POST)
            .uri("/mcp")
            .body(Body::empty())
            .unwrap();
        let resp = app.clone().oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);

        // Second request from same "IP" (no ConnectInfo in test, so peer_addr is None
        // and rate limiter won't fire). That's expected -- rate limiting requires
        // ConnectInfo which isn't available in unit tests without a real server.
        // This test verifies the middleware wiring doesn't panic.
    }

    /// Verify that rate limit semantics: only failed auth attempts consume budget.
    ///
    /// This is a unit test of the limiter behavior. The middleware integration
    /// is that on auth failure, `check_key` is called; on auth success, it is NOT.
    /// Full e2e tests verify the middleware routing but require `ConnectInfo`.
    #[test]
    fn rate_limit_semantics_failed_only() {
        let config = RateLimitConfig {
            max_attempts_per_minute: 3,
            pre_auth_max_per_minute: None,
            max_tracked_keys: default_max_tracked_keys(),
            idle_eviction: default_idle_eviction(),
            burst: None,
            pre_auth_burst: None,
            key_eviction_policy: KeyEvictionPolicy::default(),
        };
        let limiter = build_rate_limiter(&config);
        let ip = RateLimitKey::Ip("192.168.1.100".parse::<IpAddr>().unwrap());

        // Simulate: 3 failed attempts should exhaust quota.
        assert!(
            limiter.check_key(&ip).is_ok(),
            "failure 1 should be allowed"
        );
        assert!(
            limiter.check_key(&ip).is_ok(),
            "failure 2 should be allowed"
        );
        assert!(
            limiter.check_key(&ip).is_ok(),
            "failure 3 should be allowed"
        );
        assert!(
            limiter.check_key(&ip).is_err(),
            "failure 4 should be blocked"
        );

        // In the actual middleware flow:
        // - Successful auth: verify_bearer_token returns Some, we return early
        //   WITHOUT calling check_key, so no budget consumed.
        // - Failed auth: verify_bearer_token returns None, we call check_key
        //   THEN return 401, so budget is consumed.
        //
        // This means N successful requests followed by M failed requests
        // will only count M toward the rate limit, not N+M.
    }

    // -- pre-auth abuse gate (H-S1) --

    /// The pre-auth gate must default to ~10x the post-failure quota so honest
    /// retry storms never trip it but a Argon2-spray attacker is throttled.
    #[test]
    fn pre_auth_default_multiplier_is_10x() {
        let config = RateLimitConfig {
            max_attempts_per_minute: 5,
            pre_auth_max_per_minute: None,
            max_tracked_keys: default_max_tracked_keys(),
            idle_eviction: default_idle_eviction(),
            burst: None,
            pre_auth_burst: None,
            key_eviction_policy: KeyEvictionPolicy::default(),
        };
        let limiter = build_pre_auth_limiter(&config);
        let ip = RateLimitKey::Ip("10.0.0.1".parse::<IpAddr>().unwrap());

        // Quota should be 50 (5 * 10), not 5. We expect the first 50 to pass.
        for i in 0..50 {
            assert!(
                limiter.check_key(&ip).is_ok(),
                "pre-auth attempt {i} (of expected 50) should be allowed under default 10x multiplier"
            );
        }
        // The 51st attempt must be blocked: confirms quota is bounded, not infinite.
        assert!(
            limiter.check_key(&ip).is_err(),
            "pre-auth attempt 51 should be blocked (quota is 50, not unbounded)"
        );
    }

    /// An explicit `pre_auth_max_per_minute` override must win over the
    /// 10x-multiplier default.
    #[test]
    fn pre_auth_explicit_override_wins() {
        let config = RateLimitConfig {
            max_attempts_per_minute: 100,     // would default to 1000 pre-auth quota
            pre_auth_max_per_minute: Some(2), // but operator caps at 2
            max_tracked_keys: default_max_tracked_keys(),
            idle_eviction: default_idle_eviction(),
            burst: None,
            pre_auth_burst: None,
            key_eviction_policy: KeyEvictionPolicy::default(),
        };
        let limiter = build_pre_auth_limiter(&config);
        let ip = RateLimitKey::Ip("10.0.0.2".parse::<IpAddr>().unwrap());

        assert!(limiter.check_key(&ip).is_ok(), "attempt 1 allowed");
        assert!(limiter.check_key(&ip).is_ok(), "attempt 2 allowed");
        assert!(
            limiter.check_key(&ip).is_err(),
            "attempt 3 must be blocked (explicit override of 2 wins over 10x default of 1000)"
        );
    }

    /// The pre-auth gate's 429 must carry a Retry-After header.
    #[test]
    fn pre_auth_gate_deny_sets_retry_after() {
        let config = RateLimitConfig::new(100).with_pre_auth_max_per_minute(1);
        let state = AuthState {
            api_keys: ArcSwap::new(Arc::new(vec![])),
            rate_limiter: None,
            pre_auth_limiter: Some(build_pre_auth_limiter(&config)),
            #[cfg(feature = "oauth")]
            jwks_cache: None,
            seen_identities: SeenIdentitySet::new(),
            counters: AuthCounters::default(),
            resource_metadata_url: None,
        };
        let ip = RateLimitKey::Ip("10.7.7.7".parse::<IpAddr>().unwrap());
        assert!(
            pre_auth_gate(&state, Some(&ip)).is_none(),
            "first request within quota"
        );
        let resp = pre_auth_gate(&state, Some(&ip)).expect("second request must be gated");
        assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
        let retry_after = resp
            .headers()
            .get(header::RETRY_AFTER)
            .expect("Retry-After present")
            .to_str()
            .unwrap()
            .parse::<u64>()
            .unwrap();
        assert!(retry_after >= 1, "delta-seconds must be >= 1");
    }

    #[test]
    fn pre_auth_gate_capacity_full_returns_503_without_retry_after() {
        let config = RateLimitConfig::new(100)
            .with_pre_auth_max_per_minute(10)
            .with_max_tracked_keys(1)
            .with_key_eviction_policy(KeyEvictionPolicy::RejectNew);
        let state = AuthState {
            api_keys: ArcSwap::new(Arc::new(vec![])),
            rate_limiter: None,
            pre_auth_limiter: Some(build_pre_auth_limiter(&config)),
            #[cfg(feature = "oauth")]
            jwks_cache: None,
            seen_identities: SeenIdentitySet::new(),
            counters: AuthCounters::default(),
            resource_metadata_url: None,
        };
        let established = RateLimitKey::Ip("10.7.7.7".parse::<IpAddr>().unwrap());
        let unseen = RateLimitKey::Ip("10.7.7.8".parse::<IpAddr>().unwrap());
        assert!(pre_auth_gate(&state, Some(&established)).is_none());

        let resp = pre_auth_gate(&state, Some(&unseen)).expect("unseen key must be rejected");

        assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
        assert!(resp.headers().get(header::RETRY_AFTER).is_none());
    }

    /// Post-failure limiter honors an explicit burst capacity.
    #[test]
    fn post_failure_limiter_burst_allows_initial_spike() {
        let config = RateLimitConfig::new(1).with_burst(3);
        let limiter = build_rate_limiter(&config);
        let ip = RateLimitKey::Ip("10.6.6.6".parse::<IpAddr>().unwrap());
        for i in 0..3 {
            assert!(limiter.check_key(&ip).is_ok(), "burst attempt {i}");
        }
        assert!(
            limiter.check_key(&ip).is_err(),
            "attempt 4 must exceed the burst bucket"
        );
    }

    /// End-to-end: the pre-auth gate must reject before the bearer-verification
    /// path runs. We exhaust the gate's quota (Some(1)) with one bad-bearer
    /// request, then the second request must be rejected with 429 + the
    /// `pre_auth_gate` failure counter incremented (NOT
    /// `failure_invalid_credential`, which would prove Argon2 ran).
    #[tokio::test]
    async fn pre_auth_gate_blocks_before_argon2_verification() {
        let (_token, hash) = generate_api_key().unwrap();
        let keys = vec![ApiKeyEntry {
            name: "test-key".into(),
            hash,
            role: "ops".into(),
            expires_at: None,
        }];
        let config = RateLimitConfig {
            max_attempts_per_minute: 100,
            pre_auth_max_per_minute: Some(1),
            max_tracked_keys: default_max_tracked_keys(),
            idle_eviction: default_idle_eviction(),
            burst: None,
            pre_auth_burst: None,
            key_eviction_policy: KeyEvictionPolicy::default(),
        };
        let state = Arc::new(AuthState {
            api_keys: ArcSwap::new(Arc::new(keys)),
            rate_limiter: None,
            pre_auth_limiter: Some(build_pre_auth_limiter(&config)),
            #[cfg(feature = "oauth")]
            jwks_cache: None,
            seen_identities: SeenIdentitySet::new(),
            counters: AuthCounters::default(),
            resource_metadata_url: None,
        });
        let app = auth_router(Arc::clone(&state));
        let peer: SocketAddr = "10.0.0.10:54321".parse().unwrap();

        // First bad-bearer request: gate has quota, bearer verification runs,
        // returns 401 (invalid credential).
        let mut req1 = Request::builder()
            .method(axum::http::Method::POST)
            .uri("/mcp")
            .header("authorization", "Bearer obviously-not-a-real-token")
            .body(Body::empty())
            .unwrap();
        req1.extensions_mut().insert(ConnectInfo(peer));
        let resp1 = app.clone().oneshot(req1).await.unwrap();
        assert_eq!(
            resp1.status(),
            StatusCode::UNAUTHORIZED,
            "first attempt: gate has quota, falls through to bearer auth which fails with 401"
        );

        // Second bad-bearer request from same IP: gate quota exhausted, must
        // reject with 429 BEFORE the Argon2 verification path runs.
        let mut req2 = Request::builder()
            .method(axum::http::Method::POST)
            .uri("/mcp")
            .header("authorization", "Bearer also-not-a-real-token")
            .body(Body::empty())
            .unwrap();
        req2.extensions_mut().insert(ConnectInfo(peer));
        let resp2 = app.oneshot(req2).await.unwrap();
        assert_eq!(
            resp2.status(),
            StatusCode::TOO_MANY_REQUESTS,
            "second attempt from same IP: pre-auth gate must reject with 429"
        );

        let counters = state.counters_snapshot();
        assert_eq!(
            counters.failure_pre_auth_gate, 1,
            "exactly one request must have been rejected by the pre-auth gate"
        );
        // Critical: Argon2 verification must NOT have run on the gated request.
        // The first request's 401 increments `failure_invalid_credential` to 1;
        // the second (gated) request must NOT increment it further.
        assert_eq!(
            counters.failure_invalid_credential, 1,
            "bearer verification must run exactly once (only the un-gated first request)"
        );
    }

    /// mTLS-authenticated requests must bypass the pre-auth gate entirely.
    /// The TLS handshake already performed expensive crypto with a verified
    /// peer, so mTLS callers should never be throttled by this gate.
    ///
    /// Setup: a pre-auth gate with quota 1 (very tight). Submit two mTLS
    /// requests in quick succession from the same IP. Both must succeed.
    #[tokio::test]
    async fn pre_auth_gate_does_not_throttle_mtls() {
        let config = RateLimitConfig {
            max_attempts_per_minute: 100,
            pre_auth_max_per_minute: Some(1), // tight: would block 2nd plain request
            max_tracked_keys: default_max_tracked_keys(),
            idle_eviction: default_idle_eviction(),
            burst: None,
            pre_auth_burst: None,
            key_eviction_policy: KeyEvictionPolicy::default(),
        };
        let state = Arc::new(AuthState {
            api_keys: ArcSwap::new(Arc::new(vec![])),
            rate_limiter: None,
            pre_auth_limiter: Some(build_pre_auth_limiter(&config)),
            #[cfg(feature = "oauth")]
            jwks_cache: None,
            seen_identities: SeenIdentitySet::new(),
            counters: AuthCounters::default(),
            resource_metadata_url: None,
        });
        let app = auth_router(Arc::clone(&state));
        let peer: SocketAddr = "10.0.0.20:54321".parse().unwrap();
        let identity = AuthIdentity {
            name: "cn=test-client".into(),
            role: "viewer".into(),
            method: AuthMethod::MtlsCertificate,
            raw_token: None,
            sub: None,
        };
        let tls_info = TlsConnInfo::new(peer, Some(identity));

        for i in 0..3 {
            let mut req = Request::builder()
                .method(axum::http::Method::POST)
                .uri("/mcp")
                .body(Body::empty())
                .unwrap();
            req.extensions_mut().insert(ConnectInfo(tls_info.clone()));
            let resp = app.clone().oneshot(req).await.unwrap();
            assert_eq!(
                resp.status(),
                StatusCode::OK,
                "mTLS request {i} must succeed: pre-auth gate must not apply to mTLS callers"
            );
        }

        let counters = state.counters_snapshot();
        assert_eq!(
            counters.failure_pre_auth_gate, 0,
            "pre-auth gate counter must remain at zero: mTLS bypasses the gate"
        );
        assert_eq!(
            counters.success_mtls, 3,
            "all three mTLS requests must have been counted as successful"
        );
    }

    /// Pre-auth-gate denial must increment the `auth_pre` deny counter
    /// via the metrics handle in the request extensions.
    #[cfg(feature = "metrics")]
    #[tokio::test]
    async fn pre_auth_gate_deny_increments_counter() {
        let config = RateLimitConfig {
            max_attempts_per_minute: 100,
            pre_auth_max_per_minute: Some(1),
            max_tracked_keys: default_max_tracked_keys(),
            idle_eviction: default_idle_eviction(),
            burst: None,
            pre_auth_burst: None,
            key_eviction_policy: KeyEvictionPolicy::default(),
        };
        let state = Arc::new(AuthState {
            api_keys: ArcSwap::new(Arc::new(vec![])),
            rate_limiter: None,
            pre_auth_limiter: Some(build_pre_auth_limiter(&config)),
            #[cfg(feature = "oauth")]
            jwks_cache: None,
            seen_identities: SeenIdentitySet::new(),
            counters: AuthCounters::default(),
            resource_metadata_url: None,
        });
        let app = auth_router(Arc::clone(&state));
        let metrics = Arc::new(crate::metrics::McpMetrics::new().expect("metrics registry"));
        let peer: SocketAddr = "10.0.0.30:54321".parse().expect("addr parses");
        let mk = || {
            let mut req = Request::builder()
                .method(axum::http::Method::POST)
                .uri("/mcp")
                .header("authorization", "Bearer not-a-real-token")
                .body(Body::empty())
                .expect("request builds");
            req.extensions_mut().insert(ConnectInfo(peer));
            req.extensions_mut().insert(Arc::clone(&metrics));
            req
        };
        let counter = |label: &str| metrics.rate_limited_total.with_label_values(&[label]).get();

        let first = app.clone().oneshot(mk()).await.expect("first request");
        assert_eq!(first.status(), StatusCode::UNAUTHORIZED);
        assert_eq!(counter("auth_pre"), 0, "un-gated request must not count");

        let gated = app.oneshot(mk()).await.expect("second request");
        assert_eq!(gated.status(), StatusCode::TOO_MANY_REQUESTS);
        assert_eq!(counter("auth_pre"), 1, "gated request must count once");
        assert_eq!(counter("auth_post"), 0, "post limiter never fired");
    }

    /// Post-failure limiter denial must increment the `auth_post` deny
    /// counter via the metrics handle in the request extensions.
    #[cfg(feature = "metrics")]
    #[tokio::test]
    async fn post_failure_limiter_deny_increments_counter() {
        let config = RateLimitConfig {
            max_attempts_per_minute: 1, // tight: 2nd failure trips the limiter
            pre_auth_max_per_minute: None,
            max_tracked_keys: default_max_tracked_keys(),
            idle_eviction: default_idle_eviction(),
            burst: None,
            pre_auth_burst: None,
            key_eviction_policy: KeyEvictionPolicy::default(),
        };
        let state = Arc::new(AuthState {
            api_keys: ArcSwap::new(Arc::new(vec![])),
            rate_limiter: Some(build_rate_limiter(&config)),
            pre_auth_limiter: None,
            #[cfg(feature = "oauth")]
            jwks_cache: None,
            seen_identities: SeenIdentitySet::new(),
            counters: AuthCounters::default(),
            resource_metadata_url: None,
        });
        let app = auth_router(Arc::clone(&state));
        let metrics = Arc::new(crate::metrics::McpMetrics::new().expect("metrics registry"));
        let peer: SocketAddr = "10.0.0.31:54321".parse().expect("addr parses");
        let mk = || {
            let mut req = Request::builder()
                .method(axum::http::Method::POST)
                .uri("/mcp")
                .header("authorization", "Bearer not-a-real-token")
                .body(Body::empty())
                .expect("request builds");
            req.extensions_mut().insert(ConnectInfo(peer));
            req.extensions_mut().insert(Arc::clone(&metrics));
            req
        };
        let counter = |label: &str| metrics.rate_limited_total.with_label_values(&[label]).get();

        // First failure consumes the budget but is NOT itself limited.
        let first = app.clone().oneshot(mk()).await.expect("first request");
        assert_eq!(first.status(), StatusCode::UNAUTHORIZED);
        assert_eq!(counter("auth_post"), 0);

        // Second failure trips the post-failure limiter.
        let limited = app.oneshot(mk()).await.expect("second request");
        assert_eq!(limited.status(), StatusCode::TOO_MANY_REQUESTS);
        assert_eq!(counter("auth_post"), 1, "deny must count once");
        assert_eq!(counter("auth_pre"), 0, "pre-auth gate disabled here");
    }

    // -------------------------------------------------------------------
    // RFC 7235 §2.1 case-insensitive scheme parsing for `extract_bearer`.
    // -------------------------------------------------------------------

    #[test]
    fn extract_bearer_accepts_canonical_case() {
        assert_eq!(extract_bearer("Bearer abc123"), Some("abc123"));
    }

    #[test]
    fn extract_bearer_is_case_insensitive_per_rfc7235() {
        // RFC 7235 §2.1: "auth-scheme is case-insensitive".
        // Real-world clients (curl, browsers, custom HTTP libs) emit varied
        // casings; rejecting any of them is a spec violation.
        for header in &[
            "bearer abc123",
            "BEARER abc123",
            "BeArEr abc123",
            "bEaReR abc123",
        ] {
            assert_eq!(
                extract_bearer(header),
                Some("abc123"),
                "header {header:?} must parse as a Bearer token (RFC 7235 §2.1)"
            );
        }
    }

    #[test]
    fn extract_bearer_rejects_other_schemes() {
        assert_eq!(extract_bearer("Basic dXNlcjpwYXNz"), None);
        assert_eq!(extract_bearer("Digest username=\"x\""), None);
        assert_eq!(extract_bearer("Token abc123"), None);
    }

    #[test]
    fn extract_bearer_rejects_malformed() {
        // Empty string, no separator, scheme-only, scheme + only whitespace.
        assert_eq!(extract_bearer(""), None);
        assert_eq!(extract_bearer("Bearer"), None);
        assert_eq!(extract_bearer("Bearer "), None);
        assert_eq!(extract_bearer("Bearer    "), None);
    }

    #[test]
    fn extract_bearer_tolerates_extra_separator_whitespace() {
        // Some non-conformant clients emit two spaces; we should still parse.
        assert_eq!(extract_bearer("Bearer  abc123"), Some("abc123"));
        assert_eq!(extract_bearer("Bearer   abc123"), Some("abc123"));
    }

    #[test]
    fn extract_bearer_rejects_embedded_whitespace() {
        assert_eq!(extract_bearer("Bearer abc 123"), None);
        assert_eq!(extract_bearer("Bearer abc\t123"), None);
        assert_eq!(extract_bearer("Bearer abc123 "), None);
        assert_eq!(extract_bearer("Bearer abc123\r\n"), None);
    }

    #[test]
    fn extract_bearer_still_accepts_opaque_non_token68_credentials() {
        // Compatibility guard. `ApiKeyEntry::new` accepts an arbitrary
        // caller-supplied hash, so consumers may have hashed opaque tokens
        // using punctuation outside RFC 7235 `token68`. Narrowing this to a
        // strict token68 charset would silently 401 them on upgrade.
        assert_eq!(
            extract_bearer("Bearer aBc!@#$%^&*()"),
            Some("aBc!@#$%^&*()")
        );
        assert_eq!(extract_bearer("Bearer tok{en}|v1"), Some("tok{en}|v1"));
    }

    #[test]
    fn extract_bearer_accepts_generated_key_and_jwt_shapes() {
        let (token, _hash) = generate_api_key().unwrap();
        let header = format!("Bearer {token}");
        assert_eq!(extract_bearer(&header), Some(token.as_str()));

        let jwt = "eyJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJ4In0.c2ln-_bmF0dXJl";
        let jwt_header = format!("Bearer {jwt}");
        assert_eq!(extract_bearer(&jwt_header), Some(jwt));
    }

    // -------------------------------------------------------------------
    // Debug redaction: ensure `AuthIdentity` and `ApiKeyEntry` never leak
    // secret material via `format!("{:?}", …)` or `tracing::debug!(?…)`.
    // -------------------------------------------------------------------

    #[test]
    fn auth_identity_debug_redacts_raw_token() {
        let id = AuthIdentity {
            name: "alice".into(),
            role: "admin".into(),
            method: AuthMethod::OAuthJwt,
            raw_token: Some(SecretString::from("super-secret-jwt-payload-xyz")),
            sub: Some("keycloak-uuid-2f3c8b".into()),
        };
        let dbg = format!("{id:?}");

        // Plaintext fields must be visible (they are not secrets).
        assert!(dbg.contains("alice"), "name should be visible: {dbg}");
        assert!(dbg.contains("admin"), "role should be visible: {dbg}");
        assert!(dbg.contains("OAuthJwt"), "method should be visible: {dbg}");

        // Secret fields must NOT leak.
        assert!(
            !dbg.contains("super-secret-jwt-payload-xyz"),
            "raw_token must be redacted in Debug output: {dbg}"
        );
        assert!(
            !dbg.contains("keycloak-uuid-2f3c8b"),
            "sub must be redacted in Debug output: {dbg}"
        );
        assert!(
            dbg.contains("<redacted>"),
            "redaction marker missing: {dbg}"
        );
    }

    #[test]
    fn auth_identity_debug_marks_absent_secrets() {
        // For non-OAuth identities (mTLS / API key) the secret fields are
        // None; redacted Debug output should distinguish that from "present".
        let id = AuthIdentity {
            name: "viewer-key".into(),
            role: "viewer".into(),
            method: AuthMethod::BearerToken,
            raw_token: None,
            sub: None,
        };
        let dbg = format!("{id:?}");
        assert!(
            dbg.contains("<none>"),
            "absent secrets should be marked: {dbg}"
        );
        assert!(
            !dbg.contains("<redacted>"),
            "no <redacted> marker when secrets are absent: {dbg}"
        );
    }

    #[test]
    fn api_key_entry_debug_redacts_hash() {
        let entry = ApiKeyEntry {
            name: "viewer-key".into(),
            // Realistic Argon2id PHC string (must NOT leak).
            hash: "$argon2id$v=19$m=19456,t=2,p=1$c2FsdHNhbHQ$h4sh3dPa55w0rd".into(),
            role: "viewer".into(),
            expires_at: Some(RfcTimestamp::parse("2030-01-01T00:00:00Z").unwrap()),
        };
        let dbg = format!("{entry:?}");

        // Non-secret fields visible.
        assert!(dbg.contains("viewer-key"));
        assert!(dbg.contains("viewer"));
        assert!(dbg.contains("2030-01-01T00:00:00+00:00"));

        // Hash material must NOT leak.
        assert!(
            !dbg.contains("$argon2id$"),
            "argon2 hash leaked into Debug output: {dbg}"
        );
        assert!(
            !dbg.contains("h4sh3dPa55w0rd"),
            "hash digest leaked into Debug output: {dbg}"
        );
        assert!(
            dbg.contains("<redacted>"),
            "redaction marker missing: {dbg}"
        );
    }

    // -- AuthFailureClass exact-string contract tests --
    //
    // These tests pin the exact wire strings emitted for each failure
    // class. They exist to kill mutation-test mutants that replace the
    // match-arm string literals (e.g. with `""` or with the value from
    // another arm). Operators and dashboards rely on these literals
    // for metric labels and audit-log filters; any change is a
    // breaking observability change and must be reflected in
    // CHANGELOG.md.

    #[test]
    fn auth_failure_class_as_str_exact_strings() {
        assert_eq!(
            AuthFailureClass::MissingCredential.as_str(),
            "missing_credential"
        );
        assert_eq!(
            AuthFailureClass::InvalidCredential.as_str(),
            "invalid_credential"
        );
        assert_eq!(
            AuthFailureClass::ExpiredCredential.as_str(),
            "expired_credential"
        );
        assert_eq!(AuthFailureClass::RateLimited.as_str(), "rate_limited");
        assert_eq!(AuthFailureClass::PreAuthGate.as_str(), "pre_auth_gate");
    }

    #[test]
    fn auth_failure_class_response_body_exact_strings() {
        assert_eq!(
            AuthFailureClass::MissingCredential.response_body(),
            "unauthorized: missing credential"
        );
        assert_eq!(
            AuthFailureClass::InvalidCredential.response_body(),
            "unauthorized: invalid credential"
        );
        assert_eq!(
            AuthFailureClass::ExpiredCredential.response_body(),
            "unauthorized: expired credential"
        );
        assert_eq!(
            AuthFailureClass::RateLimited.response_body(),
            "rate limited"
        );
        assert_eq!(
            AuthFailureClass::PreAuthGate.response_body(),
            "rate limited (pre-auth)"
        );
    }

    #[test]
    fn auth_failure_class_bearer_error_exact_strings() {
        assert_eq!(
            AuthFailureClass::MissingCredential.bearer_error(),
            (
                "invalid_request",
                "missing bearer token or mTLS client certificate"
            )
        );
        assert_eq!(
            AuthFailureClass::InvalidCredential.bearer_error(),
            ("invalid_token", "token is invalid")
        );
        assert_eq!(
            AuthFailureClass::ExpiredCredential.bearer_error(),
            ("invalid_token", "token is expired")
        );
        assert_eq!(
            AuthFailureClass::RateLimited.bearer_error(),
            ("invalid_request", "too many failed authentication attempts")
        );
        assert_eq!(
            AuthFailureClass::PreAuthGate.bearer_error(),
            (
                "invalid_request",
                "too many unauthenticated requests from this source"
            )
        );
    }

    // -- AuthConfig::summary boolean-flag contract tests --
    //
    // These tests pin the boolean flags emitted by `AuthConfig::summary`
    // so that mutations like deleting `!` (which would invert the
    // semantics of `bearer`) or replacing `is_some()` with `is_none()`
    // are caught immediately. The summary is consumed by `/admin/*`
    // diagnostics so any inversion is an operator-visible regression.

    #[test]
    fn auth_config_summary_bearer_true_when_keys_present() {
        let (_token, hash) = generate_api_key().unwrap();
        let cfg = AuthConfig::with_keys(vec![ApiKeyEntry::new("k", hash, "viewer")]);
        let s = cfg.summary();
        assert!(s.enabled, "summary.enabled must reflect AuthConfig.enabled");
        assert!(
            s.bearer,
            "summary.bearer must be true when api_keys is non-empty (kills `!` deletion at L615)"
        );
        assert!(!s.mtls, "summary.mtls must be false when mtls is None");
        assert!(!s.oauth, "summary.oauth must be false when oauth is None");
        assert_eq!(s.api_keys.len(), 1);
        assert_eq!(s.api_keys[0].name, "k");
        assert_eq!(s.api_keys[0].role, "viewer");
    }

    #[test]
    fn auth_config_summary_bearer_false_when_no_keys() {
        let cfg = AuthConfig::with_keys(vec![]);
        let s = cfg.summary();
        assert!(
            !s.bearer,
            "summary.bearer must be false when api_keys is empty (kills `!` deletion at L615)"
        );
        assert!(s.api_keys.is_empty());
    }

    #[test]
    fn seen_identity_set_first_then_repeat() {
        let set = SeenIdentitySet::new();
        assert!(set.insert_is_first("alice"), "first sighting is first");
        assert!(
            !set.insert_is_first("alice"),
            "second sighting is not first"
        );
        assert!(set.insert_is_first("bob"));
        assert_eq!(set.len(), 2);
    }

    #[test]
    fn seen_identity_set_evicts_oldest_at_cap() {
        let set = SeenIdentitySet::with_cap(2);
        assert!(set.insert_is_first("a"));
        assert!(set.insert_is_first("b"));
        // Cap reached; inserting "c" evicts "a".
        assert!(set.insert_is_first("c"));
        assert_eq!(set.len(), 2);
        // "a" was evicted, so it re-fires as "first" (matches the documented
        // bounded trade-off: re-INFO once on reappearance). Inserting "a"
        // here evicts "b" (next oldest), leaving {c, a}.
        assert!(set.insert_is_first("a"));
        assert_eq!(set.len(), 2);
        // "b" has now been evicted in turn, so it re-fires as "first" too.
        assert!(set.insert_is_first("b"));
        // Sanity: cap is never exceeded regardless of churn pattern.
        for i in 0..32 {
            set.insert_is_first(&format!("churn-{i}"));
            assert!(set.len() <= 2, "cap invariant must hold");
        }
    }

    #[test]
    fn seen_identity_set_cap_zero_is_raised_to_one() {
        let set = SeenIdentitySet::with_cap(0);
        assert!(set.insert_is_first("only"));
        assert_eq!(set.len(), 1);
        // Next insert evicts "only".
        assert!(set.insert_is_first("next"));
        assert_eq!(set.len(), 1);
    }

    #[test]
    fn seen_identity_set_fifo_does_not_refresh_on_repeat_hit() {
        // Locks in the FIFO contract: repeat hits MUST NOT bump an entry
        // to the back of the eviction queue (that would be LRU).
        let set = SeenIdentitySet::with_cap(2);
        assert!(set.insert_is_first("a")); // order=[a]
        assert!(set.insert_is_first("b")); // order=[a,b]
        // Repeat hit on "a" - if this were LRU, "a" would move to the back
        // and "b" would be the next eviction victim. Under FIFO, "a" stays
        // at the front (oldest by insertion).
        assert!(!set.insert_is_first("a"));
        // Insert "c" forces eviction. Under FIFO, "a" (oldest by insertion)
        // is evicted; "b" survives. Under LRU, "b" would have been evicted.
        assert!(set.insert_is_first("c"));
        // Prove "a" was evicted: re-inserting fires as first again.
        assert!(set.insert_is_first("a"));
        // Prove "b" was NOT evicted: re-inserting does NOT fire as first.
        // (If LRU semantics had snuck in, this assertion would fail.)
        // After the previous step, "a" eviction pushed out "b" as the new
        // oldest, so we must re-add "b" via a fresh insert path. To keep
        // the test deterministic we rebuild a small scenario:
        let set = SeenIdentitySet::with_cap(2);
        assert!(set.insert_is_first("x")); // order=[x]
        assert!(set.insert_is_first("y")); // order=[x,y]
        assert!(!set.insert_is_first("x")); // repeat hit (under FIFO: order unchanged)
        assert!(set.insert_is_first("z")); // evicts "x" under FIFO
        assert!(
            !set.insert_is_first("y"),
            "y must still be present (FIFO did not evict it)"
        );
        assert!(
            set.insert_is_first("x"),
            "x must have been evicted by FIFO (would NOT have been evicted under LRU)"
        );
    }
}