krabka_security/

oauthbearer.rs

//! Pure SASL/OAUTHBEARER logic (KIP-255 / RFC 7628).
//!
//! Two pieces, both I/O-free so the broker can unit-test them without a
//! socket:
//!
//! 1. [`parse_client_initial_response`] decodes the RFC 7628 client initial
//!    response (`n,,\x01auth=Bearer <token>\x01\x01`) into its bearer token
//!    and optional authzid.
//! 2. [`UnsecuredJwsValidator`] validates the bearer token as an *unsecured*
//!    JWS (`alg: none`) and reads the connection principal from a claim.
//!    This mirrors Kafka's `OAuthBearerUnsecuredValidatorCallbackHandler`,
//!    the built-in development/testing validator. Signed-token (JWKS)
//!    validation is separate.

use std::{fmt, str::from_utf8, sync::Arc};

use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD as B64URL};
use jsonpath_rust::{parser::model::JpQuery, query::js_path_process};
use krabka_units::{Time, convert::TimeExt, fmt::Human as _, secs};
use serde_json::Value;

use crate::{AuthError, AuthMethod, Principal, jwks::JwksHandle};

/// Outcome of an OAUTHBEARER validation: the authenticated principal plus the
/// token's expiry.
///
/// The expiry populates `SaslAuthenticateResponse.session_lifetime_ms`. The
/// dispatch loop also uses it to schedule per-connection re-auth deadlines
/// (KIP-368).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AuthOutcome {
    pub principal: Principal,
    /// Token expiry as Unix epoch milliseconds. `None` means "no expiry / no
    /// re-auth required" and is reserved for future non-OAuth paths. For
    /// OAUTHBEARER this is always `Some`, because validators reject tokens
    /// without `exp`.
    pub expires_at_ms: Option<i64>,
}

/// Parsed RFC 7628 client initial response.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClientInitialResponse {
    /// The bearer token (the value after `auth=Bearer` and its spaces).
    pub token: String,
    /// The GS2 authorization id after `a=`, verbatim, if the client supplied
    /// one.
    pub authzid: Option<String>,
}

/// Parse the SASL/OAUTHBEARER client initial response (RFC 7628 §3.1).
///
/// The grammar is the one that Kafka 4.3.1's
/// `org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerClientInitialResponse`
/// matches against the whole message (`^A` = `\x01`):
///
/// ```text
/// message  = "n," ["a=" saslname] "," ^A *(key "=" value ^A) ^A
/// saslname = 1*([\x01-\x7F] except "=" and "," / "=2C" / "=3D")
/// key      = 1*[A-Za-z]
/// value    = 1*[\x21-\x7E SP HTAB CR LF]
/// auth     = scheme 1*SP token
/// scheme   = 1*[A-Za-z0-9_]
/// token    = 1*[-_~+/.A-Za-z0-9] *"="
/// ```
///
/// The channel-binding flag must be `n`. The authzid is returned verbatim:
/// Kafka does not decode `=2C` or `=3D`. An empty authzid is `None`. The
/// `auth` kvpair is required, and its scheme must be `Bearer` in any letter
/// case. If `auth` occurs more than once, the last one wins, as in Kafka's
/// map. The other kvpairs (SASL extensions such as `host` and `port`) must
/// match the grammar, but this function does not return them.
///
/// # Errors
///
/// [`AuthError::MalformedMessage`] when the message does not match the
/// grammar, when the `auth` kvpair is missing, or when its scheme is not
/// `Bearer`.
pub fn parse_client_initial_response(bytes: &[u8]) -> Result<ClientInitialResponse, AuthError> {
    let s = from_utf8(bytes).map_err(|_| AuthError::MalformedMessage)?;
    let (authzid, rest) = parse_gs2_header(s)?;

    // `rest` is `*(key "=" value ^A) ^A`, with nothing after the final `^A`.
    let kvpairs = rest
        .strip_suffix(KVSEP)
        .ok_or(AuthError::MalformedMessage)?;
    let mut auth = None;
    if !kvpairs.is_empty() {
        let kvpairs = kvpairs
            .strip_suffix(KVSEP)
            .ok_or(AuthError::MalformedMessage)?;
        for pair in kvpairs.split(KVSEP) {
            let (key, value) = pair.split_once('=').ok_or(AuthError::MalformedMessage)?;
            if !is_kvpair_key(key) || !is_kvpair_value(value) {
                return Err(AuthError::MalformedMessage);
            }
            if key == "auth" {
                auth = Some(value);
            }
        }
    }

    let token = parse_bearer_auth(auth.ok_or(AuthError::MalformedMessage)?)?;
    Ok(ClientInitialResponse {
        token: token.to_owned(),
        authzid,
    })
}

/// The RFC 7628 kvpair separator.
const KVSEP: char = '\u{1}';

/// Parse the GS2 header `"n," ["a=" saslname] "," ^A` at the start of a
/// client initial response.
///
/// This function returns the authzid verbatim (`None` when it is absent) and
/// the input after the header's trailing `^A`. Kafka accepts only the `n`
/// channel-binding flag: OAUTHBEARER never negotiates channel binding.
fn parse_gs2_header(message: &str) -> Result<(Option<String>, &str), AuthError> {
    let rest = message
        .strip_prefix("n,")
        .ok_or(AuthError::MalformedMessage)?;
    let (authzid, rest) = if let Some(rest) = rest.strip_prefix(',') {
        (None, rest)
    } else {
        // A saslname never contains ",", so the first "," ends it.
        let (field, rest) = rest.split_once(',').ok_or(AuthError::MalformedMessage)?;
        let name = field
            .strip_prefix("a=")
            .ok_or(AuthError::MalformedMessage)?;
        if !is_saslname(name) {
            return Err(AuthError::MalformedMessage);
        }
        (Some(name.to_owned()), rest)
    };
    let rest = rest
        .strip_prefix(KVSEP)
        .ok_or(AuthError::MalformedMessage)?;
    Ok((authzid, rest))
}

/// Whether `name` matches Kafka's `SASLNAME`: one or more of an ASCII
/// character from `\x01` to `\x7F` other than `=` and `,`, `=2C`, or `=3D`.
fn is_saslname(name: &str) -> bool {
    let bytes = name.as_bytes();
    let mut i = 0;
    while let Some(&byte) = bytes.get(i) {
        match byte {
            b'=' => {
                let escape = bytes.get(i + 1..i + 3);
                if escape != Some(b"2C".as_slice()) && escape != Some(b"3D".as_slice()) {
                    return false;
                }
                i += 3;
            }
            b',' | 0x00 | 0x80.. => return false,
            _ => i += 1,
        }
    }
    !bytes.is_empty()
}

/// Whether `key` matches Kafka's kvpair `KEY`: one or more ASCII letters.
fn is_kvpair_key(key: &str) -> bool {
    !key.is_empty() && key.bytes().all(|b| b.is_ascii_alphabetic())
}

/// Whether `value` matches Kafka's kvpair `VALUE`: one or more of a visible
/// ASCII character, space, tab, CR, or LF.
fn is_kvpair_value(value: &str) -> bool {
    !value.is_empty()
        && value
            .bytes()
            .all(|b| matches!(b, 0x21..=0x7E | b' ' | b'\t' | b'\r' | b'\n'))
}

/// Return the token of an `auth` kvpair value, which must be
/// `scheme 1*SP token` with the scheme `Bearer` in any letter case.
///
/// Kafka's `AUTH_PATTERN` is `(?<scheme>[\w]+)[ ]+(?<token>[-_~+/\.a-zA-Z0-9]+([=]*))`,
/// where `\w` is the ASCII class `[A-Za-z0-9_]`.
fn parse_bearer_auth(auth: &str) -> Result<&str, AuthError> {
    let (scheme, rest) = auth.split_once(' ').ok_or(AuthError::MalformedMessage)?;
    let token = rest.trim_start_matches(' ');
    let is_scheme_byte = |b: u8| b.is_ascii_alphanumeric() || b == b'_';
    let is_token_byte = |b: u8| b.is_ascii_alphanumeric() || b"-_~+/.".contains(&b);
    let unpadded = token.trim_end_matches('=');
    let well_formed = !scheme.is_empty()
        && scheme.bytes().all(is_scheme_byte)
        && !unpadded.is_empty()
        && unpadded.bytes().all(is_token_byte);
    if well_formed && scheme.eq_ignore_ascii_case("bearer") {
        Ok(token)
    } else {
        Err(AuthError::MalformedMessage)
    }
}

/// Validates an *unsecured* JWS bearer token (`alg: none`) and derives the
/// connection principal. Mirrors Kafka's
/// `OAuthBearerUnsecuredValidatorCallbackHandler`.
#[derive(Debug, Clone, PartialEq)]
pub struct UnsecuredJwsValidator {
    /// Claim whose string value becomes the principal name. Default `sub`.
    pub principal_claim_name: String,
    /// Tolerance applied to the `exp` / `iat` temporal checks to absorb clock
    /// drift between the client and broker. Default 0, Kafka's
    /// `unsecuredValidatorAllowableClockSkewMs`.
    pub allowable_clock_skew: Time,
    /// Precompiled `JsonPath` expression that the validator evaluates against
    /// the token's claim set. The validator rejects the token when the
    /// expression yields empty/null/false. Compile it once at validator
    /// construction.
    pub custom_claim_check: Option<JpQuery>,
    /// When set, the JWT `typ` header field must equal this
    /// string. The validator ignores this field when unset.
    pub valid_token_type: Option<String>,
    /// Alternate claim name to read the principal name from
    /// when `principal_claim_name` is absent or empty. Strimzi calls this the
    /// "service-account fallback": `sub` typically holds a UUID, and
    /// `client_id` is the readable name.
    pub fallback_user_name_claim: Option<String>,
    /// Prepended to the resolved principal name ONLY when
    /// the fallback claim fires. Strimzi convention: "service-account-".
    pub fallback_user_name_prefix: Option<String>,
    /// Precompiled `JsonPath` expression that extracts group
    /// memberships from the token claims. Compile it once at startup.
    pub groups_claim: Option<JpQuery>,
    /// When `groups_claim` resolves to a string and not an
    /// array, split on this delimiter. Common values: "," or " ".
    pub groups_claim_delimiter: Option<String>,
    /// Claim that holds the token's scope, Kafka's
    /// `unsecuredValidatorScopeClaimName`. Default `scope`.
    pub scope_claim_name: String,
    /// Scope items the token must all carry, Kafka's
    /// `unsecuredValidatorRequiredScope`. Empty, the default, requires none.
    /// [`parse_required_scope`] reads the option's space-delimited value.
    pub required_scope: Vec<String>,
}

impl Default for UnsecuredJwsValidator {
    fn default() -> Self {
        Self {
            principal_claim_name: "sub".to_owned(),
            allowable_clock_skew: <Time as TimeExt>::ZERO,
            custom_claim_check: None,
            valid_token_type: None,
            fallback_user_name_claim: None,
            fallback_user_name_prefix: None,
            groups_claim: None,
            groups_claim_delimiter: None,
            scope_claim_name: "scope".to_owned(),
            required_scope: Vec::new(),
        }
    }
}

impl UnsecuredJwsValidator {
    /// Validate `token` against `now_ms` and return the authenticated
    /// [`Principal`] on success.
    ///
    /// `now_ms` is Unix epoch milliseconds.
    ///
    /// # Errors
    ///
    /// [`AuthError::InvalidToken`] for any structural, signature, temporal,
    /// scope-claim, or principal-claim failure, and
    /// [`AuthError::InsufficientScope`] for an otherwise valid token that lacks
    /// an item of [`required_scope`](Self::required_scope). Kafka's
    /// `OAuthBearerUnsecuredValidatorCallbackHandler` makes the same split.
    /// [`oauth_error_json`] renders either one as the RFC 7628 server error
    /// challenge.
    // skip_all keeps the bearer `token` out of span fields; only the
    // non-sensitive mechanism is recorded. `err` surfaces the failure reason
    // (Debug) without the token contents.
    #[tracing::instrument(
        level = "info",
        skip_all,
        fields(mechanism = "OAUTHBEARER", validator = "unsecured"),
        err
    )]
    pub fn validate(&self, token: &str, now_ms: i64) -> Result<AuthOutcome, AuthError> {
        // JWS compact serialization: header.payload.signature. For `alg:none`
        // the signature segment is empty.
        let mut segs = token.split('.');
        let header_b64 = segs.next().ok_or(AuthError::InvalidToken)?;
        let payload_b64 = segs.next().ok_or(AuthError::InvalidToken)?;
        let sig = segs.next().ok_or(AuthError::InvalidToken)?;
        if segs.next().is_some() {
            return Err(AuthError::InvalidToken);
        }
        if !sig.is_empty() {
            // Signed token — needs JWKS signature verification.
            return Err(AuthError::InvalidToken);
        }

        let header: Value = decode_json_segment(header_b64)?;
        if header.get("alg").and_then(Value::as_str) != Some("none") {
            return Err(AuthError::InvalidToken);
        }
        // Optional JWT `typ` header check (JWT-mode validator only).
        if let Some(expected_typ) = &self.valid_token_type
            && header.get("typ").and_then(Value::as_str) != Some(expected_typ.as_str())
        {
            return Err(AuthError::InvalidToken);
        }

        let claims: Value = decode_json_segment(payload_b64)?;
        // Kafka reads the scope claim while it parses the token, so a scope
        // claim of the wrong type fails the token whatever the required scope.
        let token_scope = scope_claim(&claims, &self.scope_claim_name)?;

        let skew_ms = skew_millis(self.allowable_clock_skew);
        // `exp` is required and must be in the future (within skew).
        let exp_ms = numeric_date_ms(&claims, "exp").ok_or(AuthError::InvalidToken)?;
        if exp_ms + skew_ms <= now_ms {
            return Err(AuthError::InvalidToken);
        }
        // `iat` is optional. When present it must be a number, must not be in
        // the future (within skew), and must be before `exp`. Kafka's
        // `validateTimeConsistency` compares the two claims in seconds, without
        // skew.
        if let Some(iat) = claims.get("iat") {
            let iat_ms = numeric_date_ms(&claims, "iat").ok_or(AuthError::InvalidToken)?;
            if iat_ms - skew_ms > now_ms {
                return Err(AuthError::InvalidToken);
            }
            let (Some(issued_at_secs), Some(expires_at_secs)) =
                (iat.as_f64(), claims.get("exp").and_then(Value::as_f64))
            else {
                return Err(AuthError::InvalidToken);
            };
            if expires_at_secs <= issued_at_secs {
                return Err(AuthError::InvalidToken);
            }
        }

        // Optional JsonPath custom_claim_check.
        if let Some(path) = &self.custom_claim_check
            && !evaluate_custom_claim_check(path, &claims)
        {
            return Err(AuthError::InvalidToken);
        }

        // Primary → fallback → reject. Prefix applied only
        // when fallback fires. Kafka rejects a blank principal with
        // `Utils.isBlank`, so a blank claim counts as absent.
        let (raw_name, used_fallback) = if let Some(n) = claims
            .get(&self.principal_claim_name)
            .and_then(Value::as_str)
            .filter(|s| !is_java_blank(s))
        {
            (n.to_owned(), false)
        } else {
            let fallback_claim = self
                .fallback_user_name_claim
                .as_deref()
                .ok_or(AuthError::InvalidToken)?;
            let raw = claims
                .get(fallback_claim)
                .and_then(Value::as_str)
                .filter(|s| !is_java_blank(s))
                .ok_or(AuthError::InvalidToken)?;
            (raw.to_owned(), true)
        };
        let name = if used_fallback {
            match &self.fallback_user_name_prefix {
                Some(prefix) => format!("{prefix}{raw_name}"),
                None => raw_name,
            }
        } else {
            raw_name
        };

        // Groups extraction.
        let groups = self.groups_claim.as_ref().map_or_else(Vec::new, |path| {
            extract_groups(path, &claims, self.groups_claim_delimiter.as_deref())
        });

        // Kafka's `validateScope` runs last, so only a token that passes every
        // other check reports `insufficient_scope`.
        if !self
            .required_scope
            .iter()
            .all(|item| token_scope.contains(item))
        {
            return Err(AuthError::InsufficientScope {
                required_scope: self.required_scope.clone(),
            });
        }

        Ok(AuthOutcome {
            principal: Principal {
                name,
                auth_method: AuthMethod::SaslOAuthBearer,
                groups,
            },
            expires_at_ms: Some(exp_ms),
        })
    }
}

/// Extract group memberships from token claims with a precompiled `JsonPath`.
///
/// This function reads each result element by its JSON type:
/// - `String`: if `delimiter` is set, split + trim + drop empty;
///   otherwise the whole string becomes one group.
/// - `Array`: each string element becomes a group.
/// - `Number` / `Object` / `Null`: ignored, with no error.
///
/// This function returns `vec![]` for empty matches. Zero groups is not an
/// error, because the token can legitimately have no groups.
fn extract_groups(path: &JpQuery, claims: &Value, delimiter: Option<&str>) -> Vec<String> {
    let Ok(refs) = js_path_process(path, claims) else {
        return Vec::new();
    };
    let mut out = Vec::new();
    for r in refs {
        match r.val {
            Value::String(s) => match delimiter {
                Some(d) => out.extend(
                    s.split(d)
                        .map(str::trim)
                        .filter(|s| !s.is_empty())
                        .map(String::from),
                ),
                None => out.push(s.clone()),
            },
            Value::Array(items) => {
                out.extend(items.iter().filter_map(Value::as_str).map(String::from));
            }
            _ => {} // ignore numbers, objects, nulls
        }
    }
    out
}

/// Evaluate a precompiled `JsonPath` expression against the token claims.
///
/// This function returns true when the result is truthy, which means non-empty
/// with no element that is null or false. It returns false in every other case.
/// This matches Strimzi's "expression yields truthy" semantics. An error during
/// evaluation counts as falsy and rejects the token.
fn evaluate_custom_claim_check(path: &JpQuery, claims: &Value) -> bool {
    let Ok(refs) = js_path_process(path, claims) else {
        return false;
    };
    if refs.is_empty() {
        return false;
    }
    for r in refs {
        match r.val {
            Value::Null | Value::Bool(false) => return false,
            _ => {}
        }
    }
    true
}

/// Whether the JWT `aud` claim contains `expected`. `aud` is a single string
/// or an array of strings (RFC 7519 §4.1.3).
fn audience_contains(claims: &Value, expected: &str) -> bool {
    match claims.get("aud") {
        Some(Value::String(s)) => s == expected,
        Some(Value::Array(items)) => items
            .iter()
            .filter_map(Value::as_str)
            .any(|a| a == expected),
        _ => false,
    }
}

/// base64url-decode a JWS segment and parse it as JSON.
fn decode_json_segment(seg: &str) -> Result<Value, AuthError> {
    let bytes = B64URL.decode(seg).map_err(|_| AuthError::InvalidToken)?;
    serde_json::from_slice(&bytes).map_err(|_| AuthError::InvalidToken)
}

/// A clock-skew tolerance in whole milliseconds, the unit this module
/// normalizes the JWT `NumericDate` claims to.
///
/// The temporal checks compare epoch-millisecond instants, which stay raw
/// integers, so the tolerance must use that unit. Rounding to nearest matches
/// every other timeout that crosses this seam. No wire field carries this
/// value, so truncation is not necessary.
fn skew_millis(skew: Time) -> i64 {
    skew.millis_i64()
}

/// Read a JWT `NumericDate` claim and convert it to integer milliseconds.
///
/// A `NumericDate` is a count of seconds since the epoch and can be fractional.
fn numeric_date_ms(claims: &Value, key: &str) -> Option<i64> {
    let v = claims.get(key)?;
    // The common case: an integer second count. Avoids any float rounding.
    if let Some(secs) = v.as_i64() {
        return secs.checked_mul(1000);
    }
    // Fractional NumericDate (rare): truncate to whole milliseconds.
    let ms = v.as_f64()? * 1000.0;
    if ms.is_finite() {
        ms.trunc().to_string().parse().ok()
    } else {
        None
    }
}

/// The RFC 7628 server error challenge for a failed OAUTHBEARER validation,
/// byte for byte what Kafka 4.3.1's `OAuthBearerSaslServer.jsonErrorResponse`
/// sends.
///
/// [`AuthError::InsufficientScope`] renders as
/// `{"status":"insufficient_scope", "scope":"[a, b]"}`: the required scope in
/// Java's `List.toString` form, after the `", "` separator Kafka writes before
/// the `scope` field. Every other error renders as
/// `{"status":"invalid_token"}`. No krabka validator produces an
/// `openid-configuration` field, so the challenge never carries one.
///
/// The broker sends this as the server's first message after a failed
/// validation. The JVM `OAuthBearerSaslClient` treats any non-empty first
/// server message as an error and replies with a single `\x01` kvsep, and the
/// broker then completes the failure handshake.
#[must_use]
pub fn oauth_error_json(err: &AuthError) -> String {
    match err {
        // A required scope item matches Kafka's scope-token grammar, which has
        // no `"` or `\`, so the scope needs no JSON escaping. Kafka escapes
        // nothing either.
        AuthError::InsufficientScope { required_scope } => format!(
            "{{\"status\":\"insufficient_scope\", \"scope\":\"[{}]\"}}",
            required_scope.join(", ")
        ),
        _ => "{\"status\":\"invalid_token\"}".to_owned(),
    }
}

/// A required-scope item outside Kafka's scope-token grammar
/// (`%x21 / %x23-5B / %x5D-7E`, RFC 6749 §3.3), as
/// [`parse_required_scope`] reports it.
#[derive(Debug, thiserror::Error, Clone, PartialEq, Eq)]
#[error("invalid scope value: {0}")]
pub struct InvalidScopeItem(pub String);

/// Parse Kafka's `unsecuredValidatorRequiredScope` option the way Kafka 4.3.1's
/// `OAuthBearerUnsecuredValidatorCallbackHandler` does.
///
/// The option is trimmed, then split on single spaces, and empty items are
/// skipped. A blank option requires no scope. The result is the value of
/// [`UnsecuredJwsValidator::required_scope`].
///
/// # Errors
///
/// [`InvalidScopeItem`] for the first item that has a character outside the
/// scope-token grammar, such as `"`, `\`, a tab, or a non-ASCII character.
/// Kafka fails the validator's configuration on it.
pub fn parse_required_scope(option: &str) -> Result<Vec<String>, InvalidScopeItem> {
    java_trim(option)
        .split(' ')
        .filter(|item| !item.is_empty())
        .map(|item| {
            if item
                .chars()
                .all(|c| matches!(c, '\x21' | '\x23'..='\x5B' | '\x5D'..='\x7E'))
            {
                Ok(item.to_owned())
            } else {
                Err(InvalidScopeItem(item.to_owned()))
            }
        })
        .collect()
}

/// Java's `String.trim`: strip every leading and trailing character at or
/// below U+0020.
fn java_trim(s: &str) -> &str {
    s.trim_matches(|c: char| c <= ' ')
}

/// Kafka's `Utils.isBlank`: empty after [`java_trim`].
fn is_java_blank(s: &str) -> bool {
    java_trim(s).is_empty()
}

/// The text Jackson's `JsonNode.asText` gives a JSON value. Kafka's unsecured
/// JWS reads every non-number claim, and every element of an array claim,
/// through it. A container has no text.
fn jackson_text(v: &Value) -> String {
    match v {
        Value::String(s) => s.clone(),
        Value::Number(n) => n.to_string(),
        Value::Bool(b) => b.to_string(),
        Value::Null => "null".to_owned(),
        Value::Array(_) | Value::Object(_) => String::new(),
    }
}

/// The token's scope as Kafka's `OAuthBearerUnsecuredJws.calculateScope`
/// reads it from the `claim_name` claim.
///
/// An absent claim is no scope. An array is one item per non-blank element,
/// trimmed. A string is one item, trimmed and not split, or no scope when
/// blank, and Kafka reads a boolean, `null`, or object claim as such a string.
///
/// # Errors
///
/// [`AuthError::InvalidToken`] for a numeric claim, which Kafka cannot cast to
/// a string or a list.
fn scope_claim(claims: &Value, claim_name: &str) -> Result<Vec<String>, AuthError> {
    let items = match claims.get(claim_name) {
        None => Vec::new(),
        Some(Value::Number(_)) => return Err(AuthError::InvalidToken),
        Some(Value::Array(elements)) => elements.iter().map(jackson_text).collect(),
        Some(other) => vec![jackson_text(other)],
    };
    Ok(items
        .iter()
        .filter(|item| !is_java_blank(item))
        .map(|item| java_trim(item).to_owned())
        .collect())
}

/// Validates a *signed* JWS bearer token (`RS256` / `ES256`) against a JWKS
/// key set from the identity provider.
///
/// The validator then checks the standard JWT claims and derives the connection
/// principal.
///
/// The key set lives behind a [`JwksHandle`] so the broker's background
/// refresher can rotate keys without a broker restart and without a lock. Each
/// call to [`validate`](Self::validate) reads the current set.
#[derive(Debug, Clone)]
pub struct SignedJwsValidator {
    /// Claim whose string value becomes the principal name. Default `sub`.
    pub principal_claim_name: String,
    /// Tolerance applied to `exp` / `iat` / `nbf`.
    pub allowable_clock_skew: Time,
    /// When set, the token `iss` claim must equal this exactly.
    pub valid_issuer: Option<String>,
    /// When set, the token `aud` claim must contain this value.
    pub expected_audience: Option<String>,
    /// Precompiled `JsonPath` `custom_claim_check`. See
    /// [`UnsecuredJwsValidator`] for semantics.
    pub custom_claim_check: Option<JpQuery>,
    /// JWT `typ` header check. Ignored when unset.
    pub valid_token_type: Option<String>,
    /// Alternate principal claim. See [`UnsecuredJwsValidator`].
    pub fallback_user_name_claim: Option<String>,
    /// Prepended to the principal name only on fallback.
    pub fallback_user_name_prefix: Option<String>,
    /// Precompiled `JsonPath` that extracts group memberships.
    pub groups_claim: Option<JpQuery>,
    /// Delimiter when `groups_claim` resolves to a string.
    pub groups_claim_delimiter: Option<String>,
    /// Hard cache-expiry threshold. When set, the validator rejects tokens if
    /// the paired refresher has not had a successful fetch within this window.
    /// The check reads [`JwksHandle::last_successful_fetch_ms`]. `None` = no
    /// expiry check. The validator fails closed on a long `IdP` outage, so a
    /// rotated-out key can't keep signing valid tokens indefinitely.
    pub cache_expiry: Option<Time>,
    /// The live JWKS. The broker's refresher swaps it in.
    keys: JwksHandle,
}

impl SignedJwsValidator {
    /// A validator backed by `keys`, with the unsecured validator's claim
    /// defaults, Kafka's 30 s `sasl.oauthbearer.clock.skew.seconds`, and no
    /// issuer / audience constraint.
    #[must_use]
    pub fn new(keys: JwksHandle) -> Self {
        Self {
            principal_claim_name: "sub".to_owned(),
            allowable_clock_skew: secs(30),
            valid_issuer: None,
            expected_audience: None,
            custom_claim_check: None,
            valid_token_type: None,
            fallback_user_name_claim: None,
            fallback_user_name_prefix: None,
            groups_claim: None,
            groups_claim_delimiter: None,
            cache_expiry: None,
            keys,
        }
    }

    /// The shared key-set handle, so the broker can hand the same cell to its
    /// JWKS refresher task.
    #[must_use]
    pub fn key_handle(&self) -> JwksHandle {
        self.keys.clone()
    }

    /// Validate a signed bearer token against `now_ms`.
    ///
    /// `now_ms` is Unix epoch milliseconds.
    ///
    /// # Errors
    ///
    /// [`AuthError::InvalidToken`] for any structural, signature, temporal,
    /// issuer, audience, scope, or principal-claim failure.
    // skip_all keeps the bearer `token` (and its signature) out of span
    // fields; only the non-sensitive mechanism is recorded. `err` surfaces the
    // rejection reason (Debug) without leaking the token.
    #[tracing::instrument(
        level = "info",
        skip_all,
        fields(mechanism = "OAUTHBEARER", validator = "signed"),
        err
    )]
    pub fn validate(&self, token: &str, now_ms: i64) -> Result<AuthOutcome, AuthError> {
        // JWS compact serialization: header.payload.signature, all non-empty.
        let mut segs = token.split('.');
        let header_b64 = segs.next().ok_or(AuthError::InvalidToken)?;
        let payload_b64 = segs.next().ok_or(AuthError::InvalidToken)?;
        let sig_b64 = segs.next().ok_or(AuthError::InvalidToken)?;
        if segs.next().is_some() || sig_b64.is_empty() {
            return Err(AuthError::InvalidToken);
        }

        let header: Value = decode_json_segment(header_b64)?;
        let alg = header
            .get("alg")
            .and_then(Value::as_str)
            .ok_or(AuthError::InvalidToken)?;
        if alg != "RS256" && alg != "ES256" {
            return Err(AuthError::InvalidToken);
        }
        // Optional JWT `typ` check (JWT-mode validator only).
        if let Some(expected_typ) = &self.valid_token_type
            && header.get("typ").and_then(Value::as_str) != Some(expected_typ.as_str())
        {
            return Err(AuthError::InvalidToken);
        }

        // Hard cache-expiry. If the last successful refresh is older than
        // `cache_expiry`, reject all tokens until the refresher succeeds
        // again. The `last_fetch > 0` guard skips this on a never-fetched
        // handle so the "broker is still starting up" path stays open (the
        // verify-level check below will reject anyway because the key set is
        // empty).
        if let Some(cache_expiry) = self.cache_expiry {
            let last_fetch = self.keys.last_successful_fetch_ms();
            if last_fetch > 0 && now_ms.saturating_sub(last_fetch) > cache_expiry.millis_i64() {
                tracing::debug!(
                    last_fetch_ms = last_fetch,
                    now_ms,
                    cache_expiry = %cache_expiry.human(),
                    "JWKS cache expired; rejecting token until next successful refresh",
                );
                return Err(AuthError::InvalidToken);
            }
        }

        let kid = header.get("kid").and_then(Value::as_str);

        let signing_input = format!("{header_b64}.{payload_b64}");
        let sig = B64URL
            .decode(sig_b64)
            .map_err(|_| AuthError::InvalidToken)?;
        // On any verify failure (unknown kid or bad signature)
        // signal the refresher to attempt an on-demand JWKS fetch — the
        // signing key may have rotated since the last periodic refresh.
        // The current token still rejects; a subsequent reconnect will
        // see the rotated keys. The refresher's `min_on_demand_pause`
        // caps the signal-storm cost.
        if let Err(e) = self
            .keys
            .load()
            .verify(kid, alg, signing_input.as_bytes(), &sig)
        {
            self.keys.signal_refresh();
            return Err(e);
        }

        let claims: Value = decode_json_segment(payload_b64)?;
        self.check_claims(&claims, now_ms)
    }

    /// Apply the claim policy to already-signature-verified `claims`.
    ///
    /// The policy covers the temporal, issuer, audience, scope, and principal
    /// checks. It is a separate function so tests can drive it without minting
    /// signed tokens.
    fn check_claims(&self, claims: &Value, now_ms: i64) -> Result<AuthOutcome, AuthError> {
        let skew_ms = skew_millis(self.allowable_clock_skew);
        // `exp` required and in the future (within skew).
        let exp_ms = numeric_date_ms(claims, "exp").ok_or(AuthError::InvalidToken)?;
        if exp_ms + skew_ms <= now_ms {
            return Err(AuthError::InvalidToken);
        }
        // `iat` optional: must not be in the future (within skew).
        if let Some(iat_ms) = numeric_date_ms(claims, "iat")
            && iat_ms - skew_ms > now_ms
        {
            return Err(AuthError::InvalidToken);
        }
        // `nbf` optional: token not valid before it (within skew).
        if let Some(nbf_ms) = numeric_date_ms(claims, "nbf")
            && nbf_ms - skew_ms > now_ms
        {
            return Err(AuthError::InvalidToken);
        }

        if let Some(expected) = &self.valid_issuer
            && claims.get("iss").and_then(Value::as_str) != Some(expected.as_str())
        {
            return Err(AuthError::InvalidToken);
        }

        if let Some(expected) = &self.expected_audience
            && !audience_contains(claims, expected)
        {
            return Err(AuthError::InvalidToken);
        }

        // Optional JsonPath custom_claim_check.
        if let Some(path) = &self.custom_claim_check
            && !evaluate_custom_claim_check(path, claims)
        {
            return Err(AuthError::InvalidToken);
        }

        // Primary → fallback → reject. Prefix on fallback only.
        let (raw_name, used_fallback) = if let Some(n) = claims
            .get(&self.principal_claim_name)
            .and_then(Value::as_str)
            .filter(|s| !s.is_empty())
        {
            (n.to_owned(), false)
        } else {
            let fallback_claim = self
                .fallback_user_name_claim
                .as_deref()
                .ok_or(AuthError::InvalidToken)?;
            let raw = claims
                .get(fallback_claim)
                .and_then(Value::as_str)
                .filter(|s| !s.is_empty())
                .ok_or(AuthError::InvalidToken)?;
            (raw.to_owned(), true)
        };
        let name = if used_fallback {
            match &self.fallback_user_name_prefix {
                Some(prefix) => format!("{prefix}{raw_name}"),
                None => raw_name,
            }
        } else {
            raw_name
        };

        let groups = self.groups_claim.as_ref().map_or_else(Vec::new, |path| {
            extract_groups(path, claims, self.groups_claim_delimiter.as_deref())
        });

        Ok(AuthOutcome {
            principal: Principal {
                name,
                auth_method: AuthMethod::SaslOAuthBearer,
                groups,
            },
            expires_at_ms: Some(exp_ms),
        })
    }
}

/// The broker's configured OAUTHBEARER token validator.
///
/// The choices are the development-only unsecured-JWS path, production
/// signed-JWT validation against a JWKS endpoint, or RFC 7662 opaque-token
/// introspection. The default is unsecured.
#[derive(Debug, Clone)]
pub enum OAuthBearerValidator {
    /// Unsecured JWS (`alg:none`). Development / testing only.
    Unsecured(UnsecuredJwsValidator),
    /// Signed JWS verified against a JWKS key set.
    Signed(SignedJwsValidator),
    /// RFC 7662 opaque-token introspection.
    Introspection(IntrospectionValidator),
}

impl Default for OAuthBearerValidator {
    fn default() -> Self {
        Self::Unsecured(UnsecuredJwsValidator::default())
    }
}

impl OAuthBearerValidator {
    /// Validate `token` against `now_ms` and dispatch to the configured path.
    ///
    /// # Errors
    ///
    /// - [`AuthError::InvalidToken`] when the token fails validation.
    /// - [`AuthError::IntrospectionTransport`] when the introspection variant's
    ///   HTTP call fails at the transport layer.
    // Dispatcher over the configured validator. skip_all keeps `token` out of
    // span fields. The match arms each produce a `Result<AuthOutcome, _>`
    // value (the introspection arm is `.await`ed to a value), NOT a bare
    // future, so `async_yields_async` does not fire. `err` surfaces the
    // rejection reason (Debug) without leaking the token.
    #[tracing::instrument(level = "info", skip_all, fields(mechanism = "OAUTHBEARER"), err)]
    pub async fn validate(&self, token: &str, now_ms: i64) -> Result<AuthOutcome, AuthError> {
        match self {
            Self::Unsecured(v) => v.validate(token, now_ms),
            Self::Signed(v) => v.validate(token, now_ms),
            Self::Introspection(v) => v.validate(token, now_ms).await,
        }
    }

    /// The JWKS handle when this is a signed validator, so the broker can wire
    /// a refresher to the same key cell. `None` for the unsecured + introspection paths.
    #[must_use]
    pub fn jwks_handle(&self) -> Option<JwksHandle> {
        match self {
            Self::Unsecured(_) | Self::Introspection(_) => None,
            Self::Signed(v) => Some(v.key_handle()),
        }
    }
}

/// HTTP transport contract for RFC 7662 introspection + OIDC userinfo.
///
/// This trait lives in this crate to keep `crates/security` as the validator
/// surface. The concrete reqwest-backed impl lives in `crates/broker`, in
/// `oauth_introspection.rs`, so this crate stays I/O-free.
#[async_trait::async_trait]
pub trait IntrospectionClient: Send + Sync + fmt::Debug {
    /// POST the `IdP`'s introspection endpoint with `token` in a
    /// form-encoded body. The caller checks `active` + claims.
    async fn introspect(&self, token: &str) -> Result<serde_json::Value, IntrospectionError>;

    /// GET the `IdP`'s userinfo endpoint with `Authorization: Bearer
    /// <token>`. `Ok(None)` when the validator is configured without
    /// userinfo enrichment.
    async fn userinfo(&self, token: &str) -> Result<Option<serde_json::Value>, IntrospectionError>;
}

/// Transport-layer failures that [`IntrospectionClient`] reports.
///
/// The validator maps these onto [`AuthError::IntrospectionTransport`] for the
/// SASL handler.
#[derive(Debug, thiserror::Error)]
pub enum IntrospectionError {
    #[error("transport: {0}")]
    Transport(String),
    #[error("non-2xx response: {0}")]
    Status(u16),
    #[error("invalid JSON body")]
    Parse,
}

/// RFC 7662 opaque-token introspection validator.
///
/// The validator calls the introspection endpoint once per token and does no
/// caching. RFC 7662 §4 discourages caching without explicit lifetime info, and
/// SASL runs once per connection, so the cost is acceptable.
///
/// After a successful introspection the validator can also call OIDC userinfo
/// and merge the profile claims over the introspection claims.
#[derive(Debug, Clone)]
pub struct IntrospectionValidator {
    pub client: Arc<dyn IntrospectionClient>,
    /// Claim whose string value becomes the principal name. Default `sub`
    /// for generic OAuth flows; commonly `client_id` for Keycloak
    /// client-credentials.
    pub principal_claim_name: String,
    /// Precompiled `JsonPath` `custom_claim_check`. See
    /// [`UnsecuredJwsValidator`] for semantics. Introspection has no JWT
    /// header, so there is no `valid_token_type` field here.
    pub custom_claim_check: Option<JpQuery>,
    /// `true` iff a `userinfo_endpoint_uri` is configured. The validator then
    /// calls `client.userinfo(token)` after a successful introspection and
    /// merges the response over the introspection claims.
    pub call_userinfo: bool,
    /// Clock-skew tolerance for `exp`/`iat`/`nbf` checks on
    /// introspection-response timestamps, when present.
    pub allowable_clock_skew: Time,
    /// When set, the introspection-response `aud` claim (RFC 7662 §2.2) must
    /// contain this value. The default is `None`, which applies no audience
    /// restriction. This check stops a token minted for another resource server
    /// of the same `IdP` from authenticating here, even though that token still
    /// introspects as `active: true`.
    pub expected_audience: Option<String>,
    /// Alternate principal claim. See [`UnsecuredJwsValidator`].
    pub fallback_user_name_claim: Option<String>,
    /// Prepended to the principal name only on fallback.
    pub fallback_user_name_prefix: Option<String>,
    /// Precompiled `JsonPath` that extracts group memberships from the merged
    /// claims, which are the introspection claims plus the optional userinfo
    /// claims.
    pub groups_claim: Option<JpQuery>,
    /// Delimiter when `groups_claim` resolves to a string.
    pub groups_claim_delimiter: Option<String>,
}

impl IntrospectionValidator {
    /// Validate a bearer token with RFC 7662 introspection and optional
    /// userinfo enrichment.
    ///
    /// # Errors
    ///
    /// - [`AuthError::IntrospectionTransport`] on HTTP transport / parse failures.
    /// - [`AuthError::InvalidToken`] on `active != true`, missing `exp`,
    ///   missing principal claim, scope mismatch, or temporal-claim failure.
    ///   `exp` is required so the SASL handler can populate
    ///   `session_lifetime_ms` for KIP-368 re-authentication.
    // skip_all keeps the bearer `token` out of span fields; only the
    // non-sensitive mechanism is recorded. `err` surfaces the rejection /
    // transport error (Debug) without leaking the token. The async body ends
    // by returning an `AuthOutcome` value (not another future), so
    // `async_yields_async` does not apply.
    #[tracing::instrument(
        level = "info",
        skip_all,
        fields(mechanism = "OAUTHBEARER", validator = "introspection"),
        err
    )]
    pub async fn validate(&self, token: &str, now_ms: i64) -> Result<AuthOutcome, AuthError> {
        let mut claims = self
            .client
            .introspect(token)
            .await
            .map_err(|e| AuthError::IntrospectionTransport(e.to_string()))?;
        if claims.get("active").and_then(Value::as_bool) != Some(true) {
            return Err(AuthError::InvalidToken);
        }
        // Audience restriction. When configured, the introspection-response
        // `aud` claim (RFC 7662 §2.2) must contain the expected value, mirroring
        // the signed-JWS path. Guards against an `IdP` that serves multiple
        // resource servers handing out a token that introspects as active here
        // but was minted for a different audience.
        if let Some(expected) = &self.expected_audience
            && !audience_contains(&claims, expected)
        {
            return Err(AuthError::InvalidToken);
        }
        check_temporal_claims(&claims, now_ms, self.allowable_clock_skew)?;
        // Capture `exp_ms` from the introspection response BEFORE any userinfo
        // merge. Introspection's `exp` is the authoritative session expiry
        // (RFC 7662); userinfo typically doesn't carry `exp`, and the
        // `merge_userinfo_over_introspection` precedence already reserves
        // `exp` to introspection, but pulling it out here makes the
        // ordering explicit. Required for OAUTHBEARER (validators reject
        // tokens without `exp`).
        let exp_ms = numeric_date_ms(&claims, "exp").ok_or(AuthError::InvalidToken)?;
        if self.call_userinfo
            && let Some(ui) = self
                .client
                .userinfo(token)
                .await
                .map_err(|e| AuthError::IntrospectionTransport(e.to_string()))?
        {
            merge_userinfo_over_introspection(&mut claims, ui);
        }
        // Optional JsonPath custom_claim_check. Evaluated against the merged
        // claims (introspection plus optional userinfo).
        if let Some(path) = &self.custom_claim_check
            && !evaluate_custom_claim_check(path, &claims)
        {
            return Err(AuthError::InvalidToken);
        }
        // Primary → fallback → reject. Prefix on fallback only.
        let (raw_name, used_fallback) = if let Some(n) = claims
            .get(&self.principal_claim_name)
            .and_then(Value::as_str)
            .filter(|s| !s.is_empty())
        {
            (n.to_owned(), false)
        } else {
            let fallback_claim = self
                .fallback_user_name_claim
                .as_deref()
                .ok_or(AuthError::InvalidToken)?;
            let raw = claims
                .get(fallback_claim)
                .and_then(Value::as_str)
                .filter(|s| !s.is_empty())
                .ok_or(AuthError::InvalidToken)?;
            (raw.to_owned(), true)
        };
        let name = if used_fallback {
            match &self.fallback_user_name_prefix {
                Some(prefix) => format!("{prefix}{raw_name}"),
                None => raw_name,
            }
        } else {
            raw_name
        };

        let groups = self.groups_claim.as_ref().map_or_else(Vec::new, |path| {
            extract_groups(path, &claims, self.groups_claim_delimiter.as_deref())
        });

        Ok(AuthOutcome {
            principal: Principal {
                name,
                auth_method: AuthMethod::SaslOAuthBearer,
                groups,
            },
            expires_at_ms: Some(exp_ms),
        })
    }
}

/// Skew-tolerant temporal-claims check for introspection responses.
///
/// RFC 7662 does not mandate exp/iat/nbf, but this check honors them when they
/// are present.
fn check_temporal_claims(claims: &Value, now_ms: i64, skew: Time) -> Result<(), AuthError> {
    let skew_ms = skew_millis(skew);
    if let Some(exp_s) = claims.get("exp").and_then(Value::as_i64) {
        let exp_ms = exp_s.saturating_mul(1000);
        if now_ms.saturating_sub(skew_ms) > exp_ms {
            return Err(AuthError::InvalidToken);
        }
    }
    if let Some(iat_s) = claims.get("iat").and_then(Value::as_i64) {
        let iat_ms = iat_s.saturating_mul(1000);
        if iat_ms.saturating_sub(skew_ms) > now_ms {
            return Err(AuthError::InvalidToken);
        }
    }
    if let Some(nbf_s) = claims.get("nbf").and_then(Value::as_i64) {
        let nbf_ms = nbf_s.saturating_mul(1000);
        if nbf_ms.saturating_sub(skew_ms) > now_ms {
            return Err(AuthError::InvalidToken);
        }
    }
    Ok(())
}

/// Merge the userinfo response over the introspection claims.
///
/// Userinfo wins for profile-style claims such as `preferred_username`, email,
/// name, `given_name`, and `family_name`. Introspection wins for the small set
/// of authorization claims that `RESERVED` lists.
fn merge_userinfo_over_introspection(introspection: &mut Value, userinfo: Value) {
    const RESERVED: &[&str] = &["active", "exp", "iat", "nbf", "scope", "client_id", "sub"];
    let (Some(obj), Value::Object(ui_map)) = (introspection.as_object_mut(), userinfo) else {
        return;
    };
    for (k, v) in ui_map {
        if !RESERVED.contains(&k.as_str()) {
            obj.insert(k, v);
        }
    }
}

#[cfg(test)]
mod tests {

    use jsonpath_rust::parser::parse_json_path;
    use krabka_units::{millis, minutes};

    use super::*;

    fn jws(header: &str, claims: &str) -> String {
        format!(
            "{}.{}.",
            B64URL.encode(header.as_bytes()),
            B64URL.encode(claims.as_bytes())
        )
    }

    fn unsecured(sub: &str, iat_s: i64, exp_s: i64) -> String {
        jws(
            "{\"alg\":\"none\"}",
            &format!("{{\"sub\":\"{sub}\",\"iat\":{iat_s},\"exp\":{exp_s}}}"),
        )
    }

    /// Build an unsecured-JWS from an explicit header + claim object, so
    /// callers can drive the `typ` header for `typ`-check tests.
    ///
    /// This helper leaves the signature segment empty, as `alg:none` requires.
    fn make_unsecured_jws_with_header(
        header: &serde_json::Value,
        claims: &serde_json::Value,
    ) -> String {
        format!(
            "{}.{}.",
            B64URL.encode(serde_json::to_vec(header).unwrap()),
            B64URL.encode(serde_json::to_vec(claims).unwrap()),
        )
    }

    fn make_unsecured_jws(claims: &serde_json::Value) -> String {
        make_unsecured_jws_with_header(&serde_json::json!({"alg": "none"}), claims)
    }

    fn parse_jp(expr: &str) -> JpQuery {
        parse_json_path(expr).expect("expression compiles")
    }

    #[test]
    fn parse_accepts_client_initial_responses_that_kafka_accepts() {
        // Each case is accepted by Kafka 4.3.1's
        // `OAuthBearerClientInitialResponse` grammar.
        let cases: [(&str, &[u8], ClientInitialResponse); 14] = [
            (
                "empty authzid",
                b"n,,\x01auth=Bearer tok.en.\x01\x01",
                ClientInitialResponse {
                    token: "tok.en.".into(),
                    authzid: None,
                },
            ),
            (
                "authzid and extensions around auth",
                b"n,a=alice,\x01host=example.com\x01auth=Bearer abc\x01port=443\x01\x01",
                ClientInitialResponse {
                    token: "abc".into(),
                    authzid: Some("alice".into()),
                },
            ),
            (
                "lowercase scheme",
                b"n,,\x01auth=bearer T\x01\x01",
                ClientInitialResponse {
                    token: "T".into(),
                    authzid: None,
                },
            ),
            (
                "uppercase scheme",
                b"n,,\x01auth=BEARER T\x01\x01",
                ClientInitialResponse {
                    token: "T".into(),
                    authzid: None,
                },
            ),
            (
                "several spaces after the scheme",
                b"n,,\x01auth=Bearer   T\x01\x01",
                ClientInitialResponse {
                    token: "T".into(),
                    authzid: None,
                },
            ),
            (
                "token with padding",
                b"n,,\x01auth=Bearer abc==\x01\x01",
                ClientInitialResponse {
                    token: "abc==".into(),
                    authzid: None,
                },
            ),
            (
                "every token character class",
                b"n,,\x01auth=Bearer -_~+/.azAZ09=\x01\x01",
                ClientInitialResponse {
                    token: "-_~+/.azAZ09=".into(),
                    authzid: None,
                },
            ),
            (
                "authzid escapes stay verbatim",
                b"n,a=a=2Cb=3Dc,\x01auth=Bearer T\x01\x01",
                ClientInitialResponse {
                    token: "T".into(),
                    authzid: Some("a=2Cb=3Dc".into()),
                },
            ),
            (
                "authzid with a kvsep inside",
                b"n,a=x\x01y,\x01auth=Bearer T\x01\x01",
                ClientInitialResponse {
                    token: "T".into(),
                    authzid: Some("x\x01y".into()),
                },
            ),
            (
                "extension value with spaces and a tab",
                b"n,,\x01ext=val ue\tx\x01auth=Bearer T\x01\x01",
                ClientInitialResponse {
                    token: "T".into(),
                    authzid: None,
                },
            ),
            (
                "extension value with an equals sign",
                b"n,,\x01auth=Bearer T\x01ext=a=b\x01\x01",
                ClientInitialResponse {
                    token: "T".into(),
                    authzid: None,
                },
            ),
            (
                "later auth kvpair wins",
                b"n,,\x01auth=Bearer A\x01auth=Bearer B\x01\x01",
                ClientInitialResponse {
                    token: "B".into(),
                    authzid: None,
                },
            ),
            (
                "extension key that differs from auth only in case",
                b"n,,\x01auth=Bearer T\x01Auth=x\x01\x01",
                ClientInitialResponse {
                    token: "T".into(),
                    authzid: None,
                },
            ),
            (
                "authzid with DEL and punctuation",
                b"n,a=u\x7f!~,\x01auth=Bearer T\x01\x01",
                ClientInitialResponse {
                    token: "T".into(),
                    authzid: Some("u\x7f!~".into()),
                },
            ),
        ];
        for (name, bytes, expected) in cases {
            let parsed = parse_client_initial_response(bytes);
            assert2::assert!(parsed == Ok(expected), "case: {name}");
        }
    }

    #[test]
    fn parse_rejects_client_initial_responses_that_kafka_rejects() {
        // Each case is rejected by Kafka 4.3.1's
        // `OAuthBearerClientInitialResponse` grammar.
        let cases: [(&str, &[u8]); 34] = [
            ("missing auth kvpair", b"n,,\x01host=example.com\x01\x01"),
            ("no kvpairs", b"n,,\x01\x01"),
            (
                "auth key is case sensitive",
                b"n,,\x01AUTH=Bearer T\x01\x01",
            ),
            ("basic scheme", b"n,,\x01auth=Basic abc\x01\x01"),
            (
                "scheme with a non-word character",
                b"n,,\x01auth=Bear-er T\x01\x01",
            ),
            ("no scheme", b"n,,\x01auth= T\x01\x01"),
            ("no space after the scheme", b"n,,\x01auth=BearerT\x01\x01"),
            ("tab after the scheme", b"n,,\x01auth=Bearer\tT\x01\x01"),
            ("no token", b"n,,\x01auth=Bearer \x01\x01"),
            (
                "trailing space after the token",
                b"n,,\x01auth=Bearer T \x01\x01",
            ),
            (
                "token outside the token class",
                b"n,,\x01auth=Bearer to*ken\x01\x01",
            ),
            (
                "equals sign inside the token",
                b"n,,\x01auth=Bearer a=b\x01\x01",
            ),
            ("padding only", b"n,,\x01auth=Bearer ==\x01\x01"),
            ("unknown cb flag", b"z,,\x01auth=Bearer abc\x01\x01"),
            ("y cb flag", b"y,,\x01auth=Bearer T\x01\x01"),
            ("p cb flag", b"p=tls-unique,,\x01auth=Bearer T\x01\x01"),
            ("authzid without a=", b"n,alice,\x01auth=Bearer T\x01\x01"),
            ("empty authzid after a=", b"n,a=,\x01auth=Bearer T\x01\x01"),
            (
                "authzid with a raw equals sign",
                b"n,a=a=b,\x01auth=Bearer T\x01\x01",
            ),
            (
                "authzid with a lowercase escape",
                b"n,a=a=2c,\x01auth=Bearer T\x01\x01",
            ),
            (
                "authzid with a non-ASCII character",
                "n,a=b\u{e9},\x01auth=Bearer T\x01\x01".as_bytes(),
            ),
            (
                "authzid with a NUL",
                b"n,a=a\x00b,\x01auth=Bearer T\x01\x01",
            ),
            ("no kvsep after the gs2 header", b"n,,auth=Bearer T\x01\x01"),
            ("missing final kvsep", b"n,,\x01auth=Bearer T\x01"),
            ("missing both final kvseps", b"n,,\x01auth=Bearer T"),
            ("trailing junk", b"n,,\x01auth=Bearer T\x01\x01x"),
            ("extra trailing kvsep", b"n,,\x01auth=Bearer T\x01\x01\x01"),
            (
                "extension key with a digit",
                b"n,,\x01k1=v\x01auth=Bearer T\x01\x01",
            ),
            ("empty extension key", b"n,,\x01=v\x01auth=Bearer T\x01\x01"),
            (
                "empty extension value",
                b"n,,\x01k=\x01auth=Bearer T\x01\x01",
            ),
            (
                "extension without an equals sign",
                b"n,,\x01k\x01auth=Bearer T\x01\x01",
            ),
            (
                "extension value with a non-ASCII character",
                "n,,\x01k=v\u{e9}\x01auth=Bearer T\x01\x01".as_bytes(),
            ),
            (
                "extension value with DEL",
                b"n,,\x01k=v\x7f\x01auth=Bearer T\x01\x01",
            ),
            ("invalid UTF-8", b"n,,\x01auth=Bearer T\x01k=\xff\x01\x01"),
        ];
        for (name, bytes) in cases {
            let parsed = parse_client_initial_response(bytes);
            assert2::assert!(parsed == Err(AuthError::MalformedMessage), "case: {name}");
        }
    }

    #[test]
    fn validate_accepts_fresh_unsecured_token() {
        let v = UnsecuredJwsValidator::default();
        let now = 1_000_000_000_000;
        let token = unsecured("admin", 999_999_000, 1_000_000_900); // seconds
        let outcome = v.validate(&token, now).unwrap();
        assert2::assert!(
            (
                outcome.principal.name.as_str(),
                outcome.principal.auth_method
            ) == ("admin", AuthMethod::SaslOAuthBearer)
        );
    }

    #[test]
    fn unsecured_validate_surfaces_exp_in_auth_outcome() {
        // exp = 2000 sec = 2_000_000 ms; now = 1_000_000 ms.
        let exp_secs: i64 = 2_000;
        let now_ms: i64 = 1_000_000;
        let token = unsecured("alice", 999, exp_secs);
        let v = UnsecuredJwsValidator::default();
        let outcome = v.validate(&token, now_ms).expect("token valid");
        assert2::assert!(
            (outcome.principal.name.as_str(), outcome.expires_at_ms)
                == ("alice", Some(exp_secs * 1000))
        );
    }

    /// The temporal checks compare epoch-millisecond instants, so the
    /// configured skew must arrive there in whole milliseconds. This test pins
    /// the seam. It also pins that a sub-millisecond tolerance rounds and does
    /// not collapse to zero the way an `as i64` truncation would.
    #[test]
    fn clock_skew_crosses_into_the_temporal_checks_in_whole_milliseconds() {
        for (_name, skew, expected_ms) in [
            ("zero", <Time as TimeExt>::ZERO, 0),
            ("default", secs(30), 30_000),
            ("one millisecond", millis(1), 1),
            ("sub-millisecond rounds up", Time::from_secs_f64(0.0006), 1),
            (
                "sub-millisecond rounds down",
                Time::from_secs_f64(0.0004),
                0,
            ),
            ("five minutes", minutes(5), 300_000),
        ] {
            assert2::check!(skew_millis(skew) == expected_ms);
        }
    }

    /// `exp` is compared as `exp_ms + skew <= now_ms`, so the accept/reject
    /// boundary must sit exactly at the configured tolerance.
    #[test]
    fn unsecured_exp_boundary_tracks_the_configured_skew_exactly() {
        // exp = 1000 s ⇒ exp_ms = 1_000_000. With 30 s of skew the token is
        // still good at now = 1_030_000 and dead one millisecond later.
        let token = unsecured("admin", 900, 1_000);
        let v = UnsecuredJwsValidator {
            allowable_clock_skew: secs(30),
            ..Default::default()
        };
        assert2::check!(v.validate(&token, 1_029_999).is_ok());
        assert2::check!(v.validate(&token, 1_030_000).is_err());
    }

    /// Kafka's `unsecuredValidatorAllowableClockSkewMs` defaults to 0, so the
    /// default validator refuses a token from the millisecond its `exp` falls.
    #[test]
    fn unsecured_default_allows_no_clock_skew() {
        let token = unsecured("admin", 900, 1_000);
        let v = UnsecuredJwsValidator::default();
        assert2::check!(v.validate(&token, 999_999).is_ok());
        assert2::check!(v.validate(&token, 1_000_000).is_err());
    }

    #[test]
    fn validate_rejects_expired_token() {
        let v = UnsecuredJwsValidator {
            allowable_clock_skew: <Time as TimeExt>::ZERO,
            ..Default::default()
        };
        let now = 2_000_000_000_000;
        let token = unsecured("admin", 1_000_000_000, 1_000_000_100);
        assert2::assert!(v.validate(&token, now) == Err(AuthError::InvalidToken));
    }

    #[test]
    fn validate_rejects_future_iat() {
        let v = UnsecuredJwsValidator {
            allowable_clock_skew: <Time as TimeExt>::ZERO,
            ..Default::default()
        };
        let now = 1_000_000_000_000;
        // iat far in the future, exp even further.
        let token = unsecured("admin", 5_000_000_000, 5_000_000_100);
        assert2::assert!(v.validate(&token, now) == Err(AuthError::InvalidToken));
    }

    #[test]
    fn validate_accepts_exp_and_iat_at_skew_boundaries() {
        let v = UnsecuredJwsValidator {
            allowable_clock_skew: secs(1),
            ..Default::default()
        };

        let exp_boundary = make_unsecured_jws(&serde_json::json!({
            "sub": "admin",
            "exp": 1_999,
        }));
        assert2::assert!(v.validate(&exp_boundary, 2_000_000).is_err());
        let exp_inside = make_unsecured_jws(&serde_json::json!({
            "sub": "admin",
            "exp": 2_000,
        }));
        assert2::assert!(v.validate(&exp_inside, 2_000_000).is_ok());

        let iat_boundary = make_unsecured_jws(&serde_json::json!({
            "sub": "admin",
            "iat": 2_001,
            "exp": 3_000,
        }));
        assert2::assert!(v.validate(&iat_boundary, 2_000_000).is_ok());
        let iat_outside = make_unsecured_jws(&serde_json::json!({
            "sub": "admin",
            "iat": 2_002,
            "exp": 3_000,
        }));
        assert2::assert!(v.validate(&iat_outside, 2_000_000).is_err());
    }

    #[test]
    fn validate_rejects_signed_token() {
        let v = UnsecuredJwsValidator::default();
        let now = 1_000_000_000_000;
        // alg RS256 + a non-empty signature segment.
        let token = format!(
            "{}.{}.{}",
            B64URL.encode(b"{\"alg\":\"RS256\"}"),
            B64URL.encode(b"{\"sub\":\"admin\",\"exp\":1000000900}"),
            B64URL.encode(b"sig")
        );
        assert2::assert!(v.validate(&token, now) == Err(AuthError::InvalidToken));
    }

    #[test]
    fn validate_rejects_missing_exp() {
        let v = UnsecuredJwsValidator::default();
        let token = jws("{\"alg\":\"none\"}", "{\"sub\":\"admin\"}");
        assert2::assert!(v.validate(&token, 1_000_000_000_000) == Err(AuthError::InvalidToken));
    }

    #[test]
    fn validate_rejects_missing_principal_claim() {
        let v = UnsecuredJwsValidator::default();
        let token = jws("{\"alg\":\"none\"}", "{\"exp\":5000000000}");
        assert2::assert!(v.validate(&token, 1_000_000_000_000) == Err(AuthError::InvalidToken));
    }

    // ---- custom_claim_check (JsonPath) + valid_token_type ---

    #[test]
    fn unsecured_validate_rejects_when_custom_claim_check_fails() {
        let exp_secs: i64 = 2_000;
        let now_ms: i64 = 1_000_000;
        let token = make_unsecured_jws(&serde_json::json!({
            "sub": "alice",
            "exp": exp_secs,
            "scope": ["kafka.read"],
        }));
        let v = UnsecuredJwsValidator {
            custom_claim_check: Some(parse_jp("$.scope[?@ == 'kafka.admin']")),
            ..Default::default()
        };
        let result = v.validate(&token, now_ms);
        assert2::assert!(result.unwrap_err() == AuthError::InvalidToken);
    }

    #[test]
    fn custom_claim_check_rejects_false_and_null_matches() {
        for (_case, claims, want) in [
            ("false claim", serde_json::json!({"enabled": false}), false),
            ("null claim", serde_json::json!({"enabled": null}), false),
            ("true claim", serde_json::json!({"enabled": true}), true),
        ] {
            assert2::assert!(evaluate_custom_claim_check(&parse_jp("$.enabled"), &claims) == want);
        }
    }

    #[test]
    fn unsecured_validate_accepts_when_custom_claim_check_passes() {
        let exp_secs: i64 = 2_000;
        let now_ms: i64 = 1_000_000;
        let token = make_unsecured_jws(&serde_json::json!({
            "sub": "alice",
            "exp": exp_secs,
            "scope": ["kafka.admin", "kafka.read"],
        }));
        let v = UnsecuredJwsValidator {
            custom_claim_check: Some(parse_jp("$.scope[?@ == 'kafka.admin']")),
            ..Default::default()
        };
        let outcome = v.validate(&token, now_ms).expect("valid token");
        assert2::assert!(outcome.principal.name == "alice");
    }

    #[test]
    fn unsecured_validate_rejects_when_valid_token_type_mismatch() {
        let exp_secs: i64 = 2_000;
        let now_ms: i64 = 1_000_000;
        // Token with typ=OPAQUE in the header.
        let token = make_unsecured_jws_with_header(
            &serde_json::json!({"alg": "none", "typ": "OPAQUE"}),
            &serde_json::json!({"sub": "alice", "exp": exp_secs}),
        );
        let v = UnsecuredJwsValidator {
            valid_token_type: Some("JWT".into()),
            ..Default::default()
        };
        let result = v.validate(&token, now_ms);
        assert2::assert!(result.unwrap_err() == AuthError::InvalidToken);
    }

    #[test]
    fn unsecured_validate_accepts_when_valid_token_type_match() {
        let exp_secs: i64 = 2_000;
        let now_ms: i64 = 1_000_000;
        let token = make_unsecured_jws_with_header(
            &serde_json::json!({"alg": "none", "typ": "JWT"}),
            &serde_json::json!({"sub": "alice", "exp": exp_secs}),
        );
        let v = UnsecuredJwsValidator {
            valid_token_type: Some("JWT".into()),
            ..Default::default()
        };
        assert2::assert!(v.validate(&token, now_ms).is_ok());
    }

    #[test]
    fn unsecured_validate_accepts_when_valid_token_type_unset_regardless_of_header() {
        let exp_secs: i64 = 2_000;
        let now_ms: i64 = 1_000_000;
        let token = make_unsecured_jws_with_header(
            &serde_json::json!({"alg": "none", "typ": "OPAQUE"}),
            &serde_json::json!({"sub": "alice", "exp": exp_secs}),
        );
        let v = UnsecuredJwsValidator::default();
        // No valid_token_type set → header `typ` ignored.
        assert2::assert!(v.validate(&token, now_ms).is_ok());
    }

    // ---- name fallback chain + groups extraction --------------

    #[test]
    fn unsecured_validate_uses_primary_principal_claim_when_present() {
        // Regression: primary claim present → use primary, no prefix.
        let exp_secs: i64 = 2_000;
        let now_ms: i64 = 1_000_000;
        let token = make_unsecured_jws_with_header(
            &serde_json::json!({"alg": "none", "typ": "JWT"}),
            &serde_json::json!({"sub": "alice", "exp": exp_secs}),
        );
        let v = UnsecuredJwsValidator {
            fallback_user_name_claim: Some("client_id".into()),
            fallback_user_name_prefix: Some("service-account-".into()),
            ..Default::default()
        };
        let outcome = v.validate(&token, now_ms).expect("valid");
        assert2::assert!(outcome.principal.name == "alice"); // primary, no prefix
    }

    #[test]
    fn unsecured_validate_falls_back_to_alt_claim_when_primary_absent() {
        let exp_secs: i64 = 2_000;
        let now_ms: i64 = 1_000_000;
        let token = make_unsecured_jws_with_header(
            &serde_json::json!({"alg": "none", "typ": "JWT"}),
            // No `sub` — primary lookup fails.
            &serde_json::json!({"client_id": "svc1", "exp": exp_secs}),
        );
        let v = UnsecuredJwsValidator {
            fallback_user_name_claim: Some("client_id".into()),
            ..Default::default()
        };
        let outcome = v.validate(&token, now_ms).expect("valid");
        assert2::assert!(outcome.principal.name == "svc1"); // fallback, no prefix
    }

    #[test]
    fn unsecured_validate_applies_fallback_prefix_only_on_fallback() {
        let exp_secs: i64 = 2_000;
        let now_ms: i64 = 1_000_000;
        let token = make_unsecured_jws_with_header(
            &serde_json::json!({"alg": "none", "typ": "JWT"}),
            &serde_json::json!({"client_id": "svc1", "exp": exp_secs}),
        );
        let v = UnsecuredJwsValidator {
            fallback_user_name_claim: Some("client_id".into()),
            fallback_user_name_prefix: Some("service-account-".into()),
            ..Default::default()
        };
        let outcome = v.validate(&token, now_ms).expect("valid");
        assert2::assert!(outcome.principal.name == "service-account-svc1");
    }

    #[test]
    fn unsecured_validate_rejects_when_neither_primary_nor_fallback_present() {
        let exp_secs: i64 = 2_000;
        let now_ms: i64 = 1_000_000;
        let token = make_unsecured_jws_with_header(
            &serde_json::json!({"alg": "none", "typ": "JWT"}),
            // Neither sub nor client_id.
            &serde_json::json!({"exp": exp_secs}),
        );
        let v = UnsecuredJwsValidator {
            fallback_user_name_claim: Some("client_id".into()),
            ..Default::default()
        };
        assert2::assert!(v.validate(&token, now_ms) == Err(AuthError::InvalidToken));
    }

    #[test]
    fn unsecured_validate_extracts_groups_from_array_claim() {
        let exp_secs: i64 = 2_000;
        let now_ms: i64 = 1_000_000;
        let token = make_unsecured_jws_with_header(
            &serde_json::json!({"alg": "none", "typ": "JWT"}),
            &serde_json::json!({
                "sub": "alice",
                "exp": exp_secs,
                "groups": ["admin", "ops"],
            }),
        );
        let v = UnsecuredJwsValidator {
            groups_claim: Some(parse_jp("$.groups")),
            ..Default::default()
        };
        let outcome = v.validate(&token, now_ms).expect("valid");
        assert2::assert!(outcome.principal.groups == vec!["admin".to_owned(), "ops".to_owned()]);
    }

    #[test]
    fn unsecured_validate_extracts_groups_from_delimited_string() {
        let exp_secs: i64 = 2_000;
        let now_ms: i64 = 1_000_000;
        let token = make_unsecured_jws_with_header(
            &serde_json::json!({"alg": "none", "typ": "JWT"}),
            &serde_json::json!({
                "sub": "alice",
                "exp": exp_secs,
                "groups": "admin,ops, kafka",
            }),
        );
        let v = UnsecuredJwsValidator {
            groups_claim: Some(parse_jp("$.groups")),
            groups_claim_delimiter: Some(",".into()),
            ..Default::default()
        };
        let outcome = v.validate(&token, now_ms).expect("valid");
        assert2::assert!(
            outcome.principal.groups
                == vec!["admin".to_owned(), "ops".to_owned(), "kafka".to_owned()]
        );
    }

    #[test]
    fn unsecured_validate_extracts_groups_from_nested_claim_via_jsonpath() {
        let exp_secs: i64 = 2_000;
        let now_ms: i64 = 1_000_000;
        let token = make_unsecured_jws_with_header(
            &serde_json::json!({"alg": "none", "typ": "JWT"}),
            &serde_json::json!({
                "sub": "alice",
                "exp": exp_secs,
                "realm_access": { "roles": ["admin", "ops"] },
            }),
        );
        let v = UnsecuredJwsValidator {
            groups_claim: Some(parse_jp("$.realm_access.roles[*]")),
            ..Default::default()
        };
        let outcome = v.validate(&token, now_ms).expect("valid");
        assert2::assert!(outcome.principal.groups == vec!["admin".to_owned(), "ops".to_owned()]);
    }

    #[test]
    fn unsecured_validate_returns_empty_groups_when_claim_unset() {
        let exp_secs: i64 = 2_000;
        let now_ms: i64 = 1_000_000;
        let token = make_unsecured_jws_with_header(
            &serde_json::json!({"alg": "none", "typ": "JWT"}),
            &serde_json::json!({
                "sub": "alice",
                "exp": exp_secs,
                "groups": ["admin"],
            }),
        );
        let v = UnsecuredJwsValidator::default(); // no groups_claim
        let outcome = v.validate(&token, now_ms).expect("valid");
        assert2::assert!(outcome.principal.groups == Vec::<String>::new());
    }

    #[test]
    fn unsecured_validate_returns_empty_groups_when_claim_resolves_to_empty() {
        let exp_secs: i64 = 2_000;
        let now_ms: i64 = 1_000_000;
        let token = make_unsecured_jws_with_header(
            &serde_json::json!({"alg": "none", "typ": "JWT"}),
            &serde_json::json!({
                "sub": "alice",
                "exp": exp_secs,
            }),
        );
        let v = UnsecuredJwsValidator {
            groups_claim: Some(parse_jp("$.nonexistent")),
            ..Default::default()
        };
        let outcome = v.validate(&token, now_ms).expect("valid");
        assert2::assert!(outcome.principal.groups == Vec::<String>::new());
    }

    #[test]
    fn validate_custom_principal_claim() {
        let v = UnsecuredJwsValidator {
            principal_claim_name: "client_id".to_owned(),
            ..Default::default()
        };
        let token = jws(
            "{\"alg\":\"none\"}",
            "{\"client_id\":\"svc-1\",\"exp\":5000000000}",
        );
        let outcome = v.validate(&token, 1_000_000_000_000).unwrap();
        assert2::assert!(outcome.principal.name == "svc-1");
    }

    /// The [`AuthOutcome`] the unsecured validator returns for `name` and an
    /// `exp` of `exp_ms`, with no groups.
    fn outcome(name: &str, exp_ms: i64) -> AuthOutcome {
        AuthOutcome {
            principal: Principal {
                name: name.to_owned(),
                auth_method: AuthMethod::SaslOAuthBearer,
                groups: Vec::new(),
            },
            expires_at_ms: Some(exp_ms),
        }
    }

    /// Each case is a claim set Kafka 4.3.1's
    /// `OAuthBearerUnsecuredValidatorCallbackHandler` answers the same way:
    /// the validator runs with a 30 s skew at `now` = 1000 s.
    #[test]
    fn unsecured_validate_matches_kafka_claim_checks() {
        let now_ms = 1_000_000;
        let cases: Vec<(&str, serde_json::Value, Result<AuthOutcome, AuthError>)> = vec![
            (
                "exp before iat, both inside the skew",
                serde_json::json!({"sub": "admin", "iat": 1_010, "exp": 1_005}),
                Err(AuthError::InvalidToken),
            ),
            (
                "exp equal to iat",
                serde_json::json!({"sub": "admin", "iat": 1_005, "exp": 1_005}),
                Err(AuthError::InvalidToken),
            ),
            (
                "exp a fraction of a millisecond after iat",
                serde_json::json!({"sub": "admin", "iat": 1_005.000_1, "exp": 1_005.000_2}),
                Ok(outcome("admin", 1_005_000)),
            ),
            (
                "exp after iat",
                serde_json::json!({"sub": "admin", "iat": 1_005, "exp": 1_006}),
                Ok(outcome("admin", 1_006_000)),
            ),
            (
                "no iat skips the consistency check",
                serde_json::json!({"sub": "admin", "exp": 1_006}),
                Ok(outcome("admin", 1_006_000)),
            ),
            (
                "string iat",
                serde_json::json!({"sub": "admin", "iat": "999", "exp": 1_006}),
                Err(AuthError::InvalidToken),
            ),
            (
                "null iat",
                serde_json::json!({"sub": "admin", "iat": null, "exp": 1_006}),
                Err(AuthError::InvalidToken),
            ),
            (
                "space-only principal",
                serde_json::json!({"sub": " ", "exp": 1_006}),
                Err(AuthError::InvalidToken),
            ),
            (
                "control-character-only principal",
                serde_json::json!({"sub": "\t\n\u{1}", "exp": 1_006}),
                Err(AuthError::InvalidToken),
            ),
            (
                "principal with surrounding spaces stays verbatim",
                serde_json::json!({"sub": " alice ", "exp": 1_006}),
                Ok(outcome(" alice ", 1_006_000)),
            ),
            (
                "numeric scope",
                serde_json::json!({"sub": "admin", "exp": 1_006, "scope": 5}),
                Err(AuthError::InvalidToken),
            ),
            (
                "string scope",
                serde_json::json!({"sub": "admin", "exp": 1_006, "scope": "a b"}),
                Ok(outcome("admin", 1_006_000)),
            ),
            (
                "list scope with non-string elements, which Kafka reads as text",
                serde_json::json!({"sub": "admin", "exp": 1_006, "scope": ["a", 5, true, null, {}]}),
                Ok(outcome("admin", 1_006_000)),
            ),
            (
                "boolean scope, which Kafka reads as text",
                serde_json::json!({"sub": "admin", "exp": 1_006, "scope": true}),
                Ok(outcome("admin", 1_006_000)),
            ),
            (
                "object scope, which Kafka reads as blank",
                serde_json::json!({"sub": "admin", "exp": 1_006, "scope": {"a": 1}}),
                Ok(outcome("admin", 1_006_000)),
            ),
        ];
        // The cases put `iat` up to 10 s ahead of `now`, inside this skew.
        let v = UnsecuredJwsValidator {
            allowable_clock_skew: secs(30),
            ..Default::default()
        };
        for (name, claims, want) in cases {
            let token = make_unsecured_jws(&claims);
            assert2::check!(v.validate(&token, now_ms) == want, "case: {name}");
        }
    }

    /// A blank principal claim falls through to the fallback claim, and a
    /// blank fallback claim is rejected the way an absent one is.
    #[test]
    fn unsecured_validate_treats_blank_principal_claims_as_absent() {
        let v = UnsecuredJwsValidator {
            fallback_user_name_claim: Some("client_id".to_owned()),
            fallback_user_name_prefix: Some("service-account-".to_owned()),
            ..Default::default()
        };
        for (name, claims, want) in [
            (
                "blank primary, present fallback",
                serde_json::json!({"sub": "  ", "client_id": "svc", "exp": 1_006}),
                Ok(outcome("service-account-svc", 1_006_000)),
            ),
            (
                "blank primary, blank fallback",
                serde_json::json!({"sub": "  ", "client_id": "\t", "exp": 1_006}),
                Err(AuthError::InvalidToken),
            ),
        ] {
            let token = make_unsecured_jws(&claims);
            assert2::check!(v.validate(&token, 1_000_000) == want, "case: {name}");
        }
    }

    /// Kafka's `unsecuredValidatorRequiredScope`: every required scope item
    /// must be in the token's scope claim, or the validation fails with
    /// `insufficient_scope` and the whole required list.
    #[test]
    fn unsecured_validate_enforces_required_scope() {
        let required = vec!["a".to_owned(), "b".to_owned()];
        let insufficient = Err(AuthError::InsufficientScope {
            required_scope: required.clone(),
        });
        let scope_claim = UnsecuredJwsValidator {
            required_scope: required.clone(),
            ..Default::default()
        };
        let scp_claim = UnsecuredJwsValidator {
            required_scope: required,
            scope_claim_name: "scp".to_owned(),
            ..Default::default()
        };
        let cases = [
            (
                "list holding every required item",
                &scope_claim,
                serde_json::json!({"sub": "admin", "exp": 1_006, "scope": ["b", "a"]}),
                Ok(outcome("admin", 1_006_000)),
            ),
            (
                "list items are trimmed, extra items are fine",
                &scope_claim,
                serde_json::json!({"sub": "admin", "exp": 1_006, "scope": [" a ", "b", "c"]}),
                Ok(outcome("admin", 1_006_000)),
            ),
            (
                "list missing one item",
                &scope_claim,
                serde_json::json!({"sub": "admin", "exp": 1_006, "scope": ["a"]}),
                insufficient.clone(),
            ),
            (
                "a string scope is one item, not split on spaces",
                &scope_claim,
                serde_json::json!({"sub": "admin", "exp": 1_006, "scope": "a b"}),
                insufficient.clone(),
            ),
            (
                "no scope claim",
                &scope_claim,
                serde_json::json!({"sub": "admin", "exp": 1_006}),
                insufficient.clone(),
            ),
            (
                "configured scope claim name",
                &scp_claim,
                serde_json::json!({"sub": "admin", "exp": 1_006, "scp": ["a", "b"]}),
                Ok(outcome("admin", 1_006_000)),
            ),
            (
                "default claim name ignored once another is configured",
                &scp_claim,
                serde_json::json!({"sub": "admin", "exp": 1_006, "scope": ["a", "b"]}),
                insufficient,
            ),
            (
                "an expired token is invalid_token, not insufficient_scope",
                &scope_claim,
                serde_json::json!({"sub": "admin", "exp": 900}),
                Err(AuthError::InvalidToken),
            ),
            (
                "a blank principal is invalid_token, not insufficient_scope",
                &scope_claim,
                serde_json::json!({"sub": " ", "exp": 1_006}),
                Err(AuthError::InvalidToken),
            ),
        ];
        for (name, v, claims, want) in cases {
            let token = make_unsecured_jws(&claims);
            assert2::check!(v.validate(&token, 1_000_000) == want, "case: {name}");
        }
    }

    #[test]
    fn unsecured_validate_without_required_scope_accepts_a_token_without_scope() {
        let token = make_unsecured_jws(&serde_json::json!({"sub": "admin", "exp": 1_006}));
        assert2::assert!(
            UnsecuredJwsValidator::default().validate(&token, 1_000_000)
                == Ok(outcome("admin", 1_006_000))
        );
    }

    /// Kafka's `OAuthBearerScopeUtils.parseScope` after the option's trim.
    #[test]
    fn parse_required_scope_matches_kafka() {
        let items = |v: &[&str]| v.iter().map(ToString::to_string).collect::<Vec<_>>();
        for (input, want) in [
            ("", Ok(Vec::new())),
            (" \t ", Ok(Vec::new())),
            ("a", Ok(items(&["a"]))),
            ("a b", Ok(items(&["a", "b"]))),
            ("\t a   b \n", Ok(items(&["a", "b"]))),
            ("a,b", Ok(items(&["a,b"]))),
            ("!#[]~", Ok(items(&["!#[]~"]))),
            ("a\"b", Err(InvalidScopeItem("a\"b".to_owned()))),
            ("a\\b", Err(InvalidScopeItem("a\\b".to_owned()))),
            ("a\tb", Err(InvalidScopeItem("a\tb".to_owned()))),
            ("a é", Err(InvalidScopeItem("é".to_owned()))),
        ] {
            assert2::check!(parse_required_scope(input) == want, "input: {input:?}");
        }
    }

    /// Golden server challenges: the bytes of Kafka 4.3.1's
    /// `OAuthBearerSaslServer.jsonErrorResponse`.
    #[test]
    fn oauth_error_json_matches_kafka_json_error_response() {
        for (err, want) in [
            (AuthError::InvalidToken, r#"{"status":"invalid_token"}"#),
            (
                AuthError::IntrospectionTransport("connection refused".to_owned()),
                r#"{"status":"invalid_token"}"#,
            ),
            (
                AuthError::InsufficientScope {
                    required_scope: vec!["a".to_owned()],
                },
                r#"{"status":"insufficient_scope", "scope":"[a]"}"#,
            ),
            (
                AuthError::InsufficientScope {
                    required_scope: vec!["a".to_owned(), "b".to_owned()],
                },
                r#"{"status":"insufficient_scope", "scope":"[a, b]"}"#,
            ),
        ] {
            assert2::check!(oauth_error_json(&err) == want, "error: {err:?}");
        }
    }

    // ---- SignedJwsValidator -------------------------------------

    use tokio::sync::mpsc;

    use crate::jwks::{Jwks, JwksHandle, mint_es256, mint_rs256, mint_rs256_with_header};

    /// Build a `SignedJwsValidator` whose key set comes from `jwks_json`.
    fn signed(jwks_json: &str) -> (SignedJwsValidator, JwksHandle) {
        let handle = JwksHandle::new(Jwks::from_json(jwks_json, false).unwrap());
        (SignedJwsValidator::new(handle.clone()), handle)
    }

    #[test]
    fn signed_accepts_fresh_rs256_token() {
        let (token, jwks) = mint_rs256("k1", "{\"sub\":\"admin\",\"exp\":9999999999}");
        let (v, _h) = signed(&jwks);
        let outcome = v.validate(&token, 1_000_000_000_000).unwrap();
        assert2::assert!(
            (
                outcome.principal.name.as_str(),
                outcome.principal.auth_method
            ) == ("admin", AuthMethod::SaslOAuthBearer)
        );
    }

    #[test]
    fn signed_validate_surfaces_exp_in_auth_outcome() {
        let exp_secs: i64 = 2_000;
        let now_ms: i64 = 1_000_000;
        let (token, jwks) = mint_rs256("k1", &format!("{{\"sub\":\"alice\",\"exp\":{exp_secs}}}"));
        let (v, _h) = signed(&jwks);
        let outcome = v.validate(&token, now_ms).expect("token valid");
        assert2::assert!(
            (outcome.principal.name.as_str(), outcome.expires_at_ms)
                == ("alice", Some(exp_secs * 1000))
        );
    }

    #[test]
    fn signed_rejects_unsecured_alg_none() {
        // An `alg:none` token must never pass the signed validator even if a
        // key happens to be present.
        let (_token, jwks) = mint_rs256("k1", "{\"sub\":\"a\",\"exp\":9999999999}");
        let (v, _h) = signed(&jwks);
        let unsecured = jws("{\"alg\":\"none\"}", "{\"sub\":\"a\",\"exp\":9999999999}");
        assert2::assert!(v.validate(&unsecured, 1_000_000_000_000) == Err(AuthError::InvalidToken));
    }

    #[test]
    fn signed_rejects_malformed_compact_jws_segments() {
        let (token, jwks) = mint_rs256("k1", "{\"sub\":\"a\",\"exp\":9999999999}");
        let (v, _h) = signed(&jwks);
        let header = B64URL.encode(b"{\"alg\":\"RS256\",\"kid\":\"k1\"}");
        let payload = B64URL.encode(b"{\"sub\":\"a\",\"exp\":9999999999}");

        let empty_signature = format!("{header}.{payload}.");
        let extra_segment = format!("{token}.extra");
        for (_name, malformed) in [
            ("empty signature", empty_signature.as_str()),
            ("extra segment", extra_segment.as_str()),
        ] {
            assert2::assert!(v.validate(malformed, 1_000_000_000_000).is_err());
        }
    }

    #[test]
    fn signed_rejects_expired() {
        let (token, jwks) = mint_rs256("k1", "{\"sub\":\"a\",\"exp\":1000}");
        let (mut v, _h) = signed(&jwks);
        v.allowable_clock_skew = <Time as TimeExt>::ZERO;
        // now (ms) far past exp (1000 s).
        assert2::assert!(v.validate(&token, 5_000_000_000_000) == Err(AuthError::InvalidToken));
    }

    #[test]
    fn signed_rejects_future_nbf() {
        let (token, jwks) = mint_rs256(
            "k1",
            "{\"sub\":\"a\",\"exp\":9999999999,\"nbf\":5000000000}",
        );
        let (mut v, _h) = signed(&jwks);
        v.allowable_clock_skew = <Time as TimeExt>::ZERO;
        // now = 1e12 ms = 1e9 s, which is before nbf (5e9 s).
        assert2::assert!(v.validate(&token, 1_000_000_000_000) == Err(AuthError::InvalidToken));
    }

    #[test]
    fn signed_honors_issuer() {
        let (token, jwks) = mint_rs256(
            "k1",
            "{\"sub\":\"a\",\"exp\":9999999999,\"iss\":\"https://idp\"}",
        );
        let (mut v, _h) = signed(&jwks);
        v.valid_issuer = Some("https://idp".to_owned());
        assert2::assert!(v.validate(&token, 1_000_000_000_000).is_ok());
        v.valid_issuer = Some("https://other".to_owned());
        assert2::assert!(v.validate(&token, 1_000_000_000_000) == Err(AuthError::InvalidToken));
    }

    #[test]
    fn signed_rejects_missing_issuer_when_required() {
        let (token, jwks) = mint_rs256("k1", "{\"sub\":\"a\",\"exp\":9999999999}");
        let (mut v, _h) = signed(&jwks);
        v.valid_issuer = Some("https://idp".to_owned());
        assert2::assert!(v.validate(&token, 1_000_000_000_000) == Err(AuthError::InvalidToken));
    }

    #[test]
    fn signed_honors_audience_string_and_array() {
        let (tok_str, jwks) =
            mint_rs256("k1", "{\"sub\":\"a\",\"exp\":9999999999,\"aud\":\"kafka\"}");
        let (mut v, _h) = signed(&jwks);
        v.expected_audience = Some("kafka".to_owned());
        assert2::assert!(v.validate(&tok_str, 1_000_000_000_000).is_ok());

        let (tok_arr, jwks2) = mint_rs256(
            "k1",
            "{\"sub\":\"a\",\"exp\":9999999999,\"aud\":[\"other\",\"kafka\"]}",
        );
        let (mut v2, _h2) = signed(&jwks2);
        v2.expected_audience = Some("kafka".to_owned());
        assert2::assert!(v2.validate(&tok_arr, 1_000_000_000_000).is_ok());

        let (tok_bad, jwks3) =
            mint_rs256("k1", "{\"sub\":\"a\",\"exp\":9999999999,\"aud\":\"web\"}");
        let (mut v3, _h3) = signed(&jwks3);
        v3.expected_audience = Some("kafka".to_owned());
        assert2::assert!(v3.validate(&tok_bad, 1_000_000_000_000) == Err(AuthError::InvalidToken));
    }

    // ---- SignedJwsValidator custom_claim_check + valid_token_type

    #[test]
    fn signed_validate_rejects_when_custom_claim_check_fails() {
        let (token, jwks) = mint_rs256(
            "k1",
            "{\"sub\":\"alice\",\"exp\":9999999999,\"scope\":[\"kafka.read\"]}",
        );
        let (mut v, _h) = signed(&jwks);
        v.custom_claim_check = Some(parse_jp("$.scope[?@ == 'kafka.admin']"));
        let result = v.validate(&token, 1_000_000_000_000);
        assert2::assert!(result.unwrap_err() == AuthError::InvalidToken);
    }

    #[test]
    fn signed_validate_accepts_when_custom_claim_check_passes() {
        let (token, jwks) = mint_rs256(
            "k1",
            "{\"sub\":\"alice\",\"exp\":9999999999,\"scope\":[\"kafka.admin\",\"kafka.read\"]}",
        );
        let (mut v, _h) = signed(&jwks);
        v.custom_claim_check = Some(parse_jp("$.scope[?@ == 'kafka.admin']"));
        let outcome = v.validate(&token, 1_000_000_000_000).expect("valid token");
        assert2::assert!(outcome.principal.name == "alice");
    }

    #[test]
    fn signed_validate_rejects_when_valid_token_type_mismatch() {
        let (token, jwks) = mint_rs256_with_header(
            "{\"alg\":\"RS256\",\"kid\":\"k1\",\"typ\":\"OPAQUE\"}",
            "{\"sub\":\"alice\",\"exp\":9999999999}",
        );
        let (mut v, _h) = signed(&jwks);
        v.valid_token_type = Some("JWT".into());
        let result = v.validate(&token, 1_000_000_000_000);
        assert2::assert!(result.unwrap_err() == AuthError::InvalidToken);
    }

    #[test]
    fn signed_validate_accepts_when_valid_token_type_match() {
        let (token, jwks) = mint_rs256_with_header(
            "{\"alg\":\"RS256\",\"kid\":\"k1\",\"typ\":\"JWT\"}",
            "{\"sub\":\"alice\",\"exp\":9999999999}",
        );
        let (mut v, _h) = signed(&jwks);
        v.valid_token_type = Some("JWT".into());
        assert2::assert!(v.validate(&token, 1_000_000_000_000).is_ok());
    }

    #[test]
    fn signed_validate_accepts_when_valid_token_type_unset_regardless_of_header() {
        let (token, jwks) = mint_rs256_with_header(
            "{\"alg\":\"RS256\",\"kid\":\"k1\",\"typ\":\"OPAQUE\"}",
            "{\"sub\":\"alice\",\"exp\":9999999999}",
        );
        let (v, _h) = signed(&jwks);
        // No valid_token_type set → header `typ` ignored.
        assert2::assert!(v.validate(&token, 1_000_000_000_000).is_ok());
    }

    #[test]
    fn signed_rejects_missing_principal() {
        let (token, jwks) = mint_rs256("k1", "{\"exp\":9999999999}");
        let (v, _h) = signed(&jwks);
        assert2::assert!(v.validate(&token, 1_000_000_000_000) == Err(AuthError::InvalidToken));
    }

    #[test]
    fn signed_custom_principal_claim() {
        let (token, jwks) = mint_rs256("k1", "{\"client_id\":\"svc-1\",\"exp\":9999999999}");
        let (mut v, _h) = signed(&jwks);
        v.principal_claim_name = "client_id".to_owned();
        assert2::assert!(
            v.validate(&token, 1_000_000_000_000)
                .unwrap()
                .principal
                .name
                == "svc-1"
        );
    }

    #[test]
    fn signed_key_rotation_via_handle() {
        // Token A verifies under key set A. ES256 with a fresh key per mint, so
        // set B is a genuinely different key (RS256's fixed test key can't be).
        let (token_a, jwks_a) = mint_es256("k1", "{\"sub\":\"a\",\"exp\":9999999999}");
        let (v, handle) = signed(&jwks_a);
        assert2::assert!(v.validate(&token_a, 1_000_000_000_000).is_ok());

        // Rotate to a fresh key set (same kid, new key). Token A no longer
        // verifies; a token under the new key does. Same validator instance.
        let (token_b, jwks_b) = mint_es256("k1", "{\"sub\":\"b\",\"exp\":9999999999}");
        handle.store(Jwks::from_json(&jwks_b, false).unwrap());
        assert2::assert!(
            (
                v.validate(&token_a, 1_000_000_000_000),
                v.validate(&token_b, 1_000_000_000_000)
                    .map(|outcome| outcome.principal.name),
            ) == (Err(AuthError::InvalidToken), Ok("b".to_owned()))
        );
    }

    #[test]
    fn signed_key_handle_shares_validator_key_cell() {
        let (token_a, jwks_a) = mint_es256("k1", "{\"sub\":\"a\",\"exp\":9999999999}");
        let (v, _handle) = signed(&jwks_a);
        assert2::assert!(v.validate(&token_a, 1_000_000_000_000).is_ok());

        let handle = v.key_handle();
        let (token_b, jwks_b) = mint_es256("k1", "{\"sub\":\"b\",\"exp\":9999999999}");
        handle.store(Jwks::from_json(&jwks_b, false).expect("parse rotated jwks"));

        assert2::assert!(
            (
                v.validate(&token_a, 1_000_000_000_000),
                v.validate(&token_b, 1_000_000_000_000)
                    .map(|outcome| outcome.principal.name),
            ) == (Err(AuthError::InvalidToken), Ok("b".to_owned()))
        );
    }

    #[test]
    fn signed_rejects_tokens_with_expected_issuer_or_audience_missing() {
        let (token, jwks) = mint_rs256("k1", "{\"sub\":\"a\",\"exp\":9999999999}");
        let (mut issuer_validator, _h) = signed(&jwks);
        issuer_validator.valid_issuer = Some("https://idp".to_owned());
        assert2::assert!(
            issuer_validator.validate(&token, 1_000_000_000_000) == Err(AuthError::InvalidToken)
        );

        let (mut audience_validator, _h) = signed(&jwks);
        audience_validator.expected_audience = Some("kafka".to_owned());
        assert2::assert!(
            audience_validator.validate(&token, 1_000_000_000_000) == Err(AuthError::InvalidToken)
        );
    }

    #[test]
    fn signed_accepts_temporal_claims_at_skew_boundaries() {
        let (token, jwks) = mint_rs256(
            "k1",
            "{\"sub\":\"a\",\"exp\":2000,\"iat\":2001,\"nbf\":2001}",
        );
        let (mut v, _h) = signed(&jwks);
        v.allowable_clock_skew = secs(1);
        assert2::assert!(v.validate(&token, 2_000_000).is_ok());
    }

    #[test]
    fn signed_rejects_temporal_claims_outside_skew_boundaries() {
        let (expired, expired_jwks) = mint_rs256("k1", "{\"sub\":\"a\",\"exp\":1999}");
        let (mut expired_v, _h) = signed(&expired_jwks);
        expired_v.allowable_clock_skew = secs(1);
        assert2::assert!(expired_v.validate(&expired, 2_000_000).is_err());

        let (future_iat, iat_jwks) = mint_rs256("k1", "{\"sub\":\"a\",\"exp\":3000,\"iat\":2002}");
        let (mut iat_v, _h) = signed(&iat_jwks);
        iat_v.allowable_clock_skew = secs(1);
        assert2::assert!(iat_v.validate(&future_iat, 2_000_000).is_err());

        let (future_nbf, nbf_jwks) = mint_rs256("k1", "{\"sub\":\"a\",\"exp\":3000,\"nbf\":2002}");
        let (mut nbf_v, _h) = signed(&nbf_jwks);
        nbf_v.allowable_clock_skew = secs(1);
        assert2::assert!(nbf_v.validate(&future_nbf, 2_000_000).is_err());
    }

    #[test]
    fn signed_rejects_when_keyset_empty() {
        let (token, _jwks) = mint_rs256("k1", "{\"sub\":\"a\",\"exp\":9999999999}");
        let v = SignedJwsValidator::new(JwksHandle::default());
        assert2::assert!(v.validate(&token, 1_000_000_000_000) == Err(AuthError::InvalidToken));
    }

    // ---- signed-validator parity --------------------------------

    #[test]
    fn signed_validate_falls_back_to_alt_claim_when_primary_absent() {
        // Mirror the unsecured fallback test using the signed validator.
        // No `sub` claim → fallback to `client_id`, then apply prefix.
        let (token, jwks) = mint_rs256_with_header(
            "{\"alg\":\"RS256\",\"kid\":\"k1\",\"typ\":\"JWT\"}",
            "{\"client_id\":\"svc1\",\"exp\":9999999999,\"iss\":\"https://test.example\"}",
        );
        let (mut v, _h) = signed(&jwks);
        v.fallback_user_name_claim = Some("client_id".into());
        v.fallback_user_name_prefix = Some("service-account-".into());
        let outcome = v.validate(&token, 1_000_000_000_000).expect("valid");
        assert2::assert!(outcome.principal.name == "service-account-svc1");
    }

    #[test]
    fn signed_validate_extracts_groups_from_array_claim() {
        let (token, jwks) = mint_rs256(
            "k1",
            "{\"sub\":\"alice\",\"exp\":9999999999,\"groups\":[\"admin\",\"ops\"]}",
        );
        let (mut v, _h) = signed(&jwks);
        v.groups_claim = Some(parse_jp("$.groups"));
        let outcome = v.validate(&token, 1_000_000_000_000).expect("valid");
        assert2::assert!(outcome.principal.groups == vec!["admin".to_owned(), "ops".to_owned()]);
    }

    // ---- cache expiry + signal-on-verify-failure ----------------

    /// Build a signed validator whose paired `JwksHandle` carries an explicit
    /// `last_successful_fetch_ms` and a fresh signal channel.
    ///
    /// This helper returns the validator and the receiver, so tests can assert
    /// on the emitted signals.
    fn signed_with_handles(
        jwks_json: &str,
        last_successful_fetch_ms: i64,
    ) -> (SignedJwsValidator, mpsc::Receiver<()>) {
        use std::sync::{Arc, atomic::AtomicI64};
        let (tx, rx) = mpsc::channel::<()>(1);
        let ts = Arc::new(AtomicI64::new(last_successful_fetch_ms));
        let handle = JwksHandle::new_with_refresher_handles(
            Jwks::from_json(jwks_json, false).unwrap(),
            ts,
            tx,
        );
        (SignedJwsValidator::new(handle), rx)
    }

    #[test]
    fn signed_validate_rejects_when_jwks_cache_expired() {
        let now_ms: i64 = 10_000_000;
        let exp_secs: i64 = (now_ms / 1000) + 60;
        let (token, jwks) = mint_rs256("k1", &format!("{{\"sub\":\"alice\",\"exp\":{exp_secs}}}"));
        // Last fetch 2s ago; expiry threshold 1s ⇒ expired.
        let (mut v, _rx) = signed_with_handles(&jwks, now_ms - 2_000);
        v.cache_expiry = Some(secs(1));
        assert2::assert!(v.validate(&token, now_ms) == Err(AuthError::InvalidToken));
    }

    #[test]
    fn signed_validate_accepts_when_jwks_cache_within_expiry() {
        let now_ms: i64 = 10_000_000;
        let exp_secs: i64 = (now_ms / 1000) + 60;
        let (token, jwks) = mint_rs256("k1", &format!("{{\"sub\":\"alice\",\"exp\":{exp_secs}}}"));
        // Last fetch 500ms ago; expiry threshold 1s ⇒ still fresh.
        let (mut v, _rx) = signed_with_handles(&jwks, now_ms - 500);
        v.cache_expiry = Some(secs(1));
        let outcome = v.validate(&token, now_ms).expect("valid");
        assert2::assert!(outcome.principal.name == "alice");
    }

    #[test]
    fn signed_validate_accepts_when_jwks_cache_age_equals_expiry() {
        let now_ms: i64 = 10_000_000;
        let exp_secs: i64 = (now_ms / 1000) + 60;
        let (token, jwks) = mint_rs256("k1", &format!("{{\"sub\":\"alice\",\"exp\":{exp_secs}}}"));
        let (mut v, _rx) = signed_with_handles(&jwks, now_ms - 1_000);
        v.cache_expiry = Some(secs(1));
        let outcome = v.validate(&token, now_ms).expect("valid");
        assert2::assert!(outcome.principal.name == "alice");
    }

    #[test]
    fn signed_validate_accepts_when_expiry_unset_regardless_of_cache_age() {
        let now_ms: i64 = 10_000_000;
        let exp_secs: i64 = (now_ms / 1000) + 60;
        let (token, jwks) = mint_rs256("k1", &format!("{{\"sub\":\"alice\",\"exp\":{exp_secs}}}"));
        // Cache "is" very stale, but cache_expiry = None ⇒ no check fires.
        let (v, _rx) = signed_with_handles(&jwks, now_ms - 999_999_999);
        // cache_expiry left at default None.
        assert2::assert!(v.validate(&token, now_ms).is_ok());
    }

    #[test]
    fn signed_validate_skips_expiry_check_when_never_fetched() {
        // last_successful_fetch_ms still at 0 ⇒ no expiry math runs (the
        // never-fetched broker-startup window stays open; verify will fail
        // anyway because the served key set is intact, but the expiry path
        // must not preempt with a spurious rejection).
        let now_ms: i64 = 10_000_000;
        let exp_secs: i64 = (now_ms / 1000) + 60;
        let (token, jwks) = mint_rs256("k1", &format!("{{\"sub\":\"alice\",\"exp\":{exp_secs}}}"));
        let (mut v, _rx) = signed_with_handles(&jwks, 0);
        v.cache_expiry = Some(millis(1));
        // Cache "age" math would say "very expired" if we ran it, but the
        // sentinel-zero guard skips. Verification succeeds.
        assert2::assert!(v.validate(&token, now_ms).is_ok());
    }

    #[test]
    fn signed_validate_signals_refresh_on_unknown_kid() {
        let now_ms: i64 = 10_000_000;
        let exp_secs: i64 = (now_ms / 1000) + 60;
        // Token's header advertises kid="k1" (the default rs256 mint), but the
        // served JWKS has a different kid ⇒ verify() returns InvalidToken AND
        // signal_refresh() fires.
        let (token, _jwks_with_k1) =
            mint_rs256("k1", &format!("{{\"sub\":\"alice\",\"exp\":{exp_secs}}}"));
        // Hand-craft a JWKS with a different kid so verify can't find k1.
        let mismatched_jwks =
            r#"{"keys":[{"kty":"RSA","kid":"other","n":"AQAB","e":"AQAB"}]}"#.to_owned();
        let (v, mut rx) = signed_with_handles(&mismatched_jwks, now_ms);
        assert2::assert!(
            (v.validate(&token, now_ms), rx.try_recv().is_ok())
                == (Err(AuthError::InvalidToken), true)
        );
    }

    #[test]
    fn signed_validate_does_not_signal_when_verification_succeeds() {
        let now_ms: i64 = 10_000_000;
        let exp_secs: i64 = (now_ms / 1000) + 60;
        let (token, jwks) = mint_rs256("k1", &format!("{{\"sub\":\"alice\",\"exp\":{exp_secs}}}"));
        let (v, mut rx) = signed_with_handles(&jwks, now_ms);
        assert2::assert!(
            (v.validate(&token, now_ms).is_ok(), rx.try_recv().is_err()) == (true, true)
        );
    }

    #[tokio::test]
    async fn enum_dispatches_unsecured_and_signed() {
        // Unsecured default.
        let unsecured = OAuthBearerValidator::default();
        let tok = unsecured_token("admin", 999_999_000, 9_999_999_999);
        assert2::assert!(
            (
                unsecured.jwks_handle().is_none(),
                unsecured.validate(&tok, 1_000_000_000_000).await.is_ok(),
            ) == (true, true)
        );

        // Signed.
        let (token, jwks) = mint_rs256("k1", "{\"sub\":\"x\",\"exp\":9999999999}");
        let (sv, _h) = signed(&jwks);
        let signed_enum = OAuthBearerValidator::Signed(sv);
        assert2::assert!(
            (
                signed_enum.jwks_handle().is_some(),
                signed_enum
                    .validate(&token, 1_000_000_000_000)
                    .await
                    .map(|outcome| outcome.principal.name),
            ) == (true, Ok("x".to_owned()))
        );
    }

    fn unsecured_token(sub: &str, iat_s: i64, exp_s: i64) -> String {
        jws(
            "{\"alg\":\"none\"}",
            &format!("{{\"sub\":\"{sub}\",\"iat\":{iat_s},\"exp\":{exp_s}}}"),
        )
    }

    #[test]
    fn custom_claim_check_compile_error_at_validator_construction() {
        // Operators paste a malformed expression. We catch it at parse
        // time (validator construction), not per-token validation.
        let result = parse_json_path("@.unterminated");
        assert2::assert!(result.is_err());
    }
}

#[cfg(test)]
mod introspection_tests {
    use std::{collections::HashMap, sync::Mutex};

    use jsonpath_rust::parser::parse_json_path;
    use serde_json::{Value, json};

    use super::*;
    use crate::{AuthError, AuthMethod};

    // --- pure temporal/claim helpers (no IdP/JWKS fixtures needed) ----------

    #[test]
    fn temporal_claims_exp_boundary_is_inclusive() {
        // now - skew == exp_ms is NOT expired (the guard is `>`, not `>=`).
        let claims = json!({ "exp": 10 }); // exp_ms = 10_000
        for (_name, now_ms, expected) in [
            ("at boundary", 10_000, true),
            ("past expiry", 10_001, false),
        ] {
            assert2::assert!(
                check_temporal_claims(&claims, now_ms, <Time as TimeExt>::ZERO).is_ok() == expected
            );
        }
    }

    #[test]
    fn temporal_claims_iat_and_nbf_future_rejected() {
        // iat: at the boundary it's valid; issued in the future is rejected.
        let iat = json!({ "iat": 10 }); // iat_ms = 10_000
        assert2::assert!(check_temporal_claims(&iat, 10_000, <Time as TimeExt>::ZERO).is_ok());
        assert2::assert!(check_temporal_claims(&iat, 9_999, <Time as TimeExt>::ZERO).is_err());
        // nbf (not-before): same shape.
        let nbf = json!({ "nbf": 10 });
        assert2::assert!(check_temporal_claims(&nbf, 10_000, <Time as TimeExt>::ZERO).is_ok());
        assert2::assert!(check_temporal_claims(&nbf, 9_999, <Time as TimeExt>::ZERO).is_err());
    }

    #[test]
    fn numeric_date_fractional_seconds_to_ms() {
        // Fractional NumericDate path: 10.5 s -> 10_500 ms (pins the `* 1000`).
        assert2::assert!(numeric_date_ms(&json!({ "k": 10.5 }), "k") == Some(10_500));
    }

    #[test]
    fn audience_contains_array_membership() {
        // `aud` array membership: `any(a == expected)` (kills `==` -> `!=`).
        for (_name, audience, expected) in [
            ("present audience", "svc-a", true),
            ("absent audience", "svc-b", false),
        ] {
            assert2::assert!(audience_contains(&json!({ "aud": ["svc-a"] }), audience) == expected);
        }
    }

    /// Per-token canned responses.
    ///
    /// `introspect` returns the entry for the matching token. When there is no
    /// entry it returns a Transport error, so a test can drive the
    /// transport-error path.
    #[derive(Debug, Default)]
    struct MockIntrospectionClient {
        introspect_responses: Mutex<HashMap<String, Result<Value, IntrospectionError>>>,
        userinfo_responses: Mutex<HashMap<String, Result<Option<Value>, IntrospectionError>>>,
    }

    impl MockIntrospectionClient {
        fn arc() -> Arc<Self> {
            Arc::new(Self::default())
        }
        fn set_introspect(&self, token: &str, resp: Result<Value, IntrospectionError>) {
            self.introspect_responses
                .lock()
                .unwrap()
                .insert(token.into(), resp);
        }
        fn set_userinfo(&self, token: &str, resp: Result<Option<Value>, IntrospectionError>) {
            self.userinfo_responses
                .lock()
                .unwrap()
                .insert(token.into(), resp);
        }
    }

    #[async_trait::async_trait]
    impl IntrospectionClient for MockIntrospectionClient {
        async fn introspect(&self, token: &str) -> Result<Value, IntrospectionError> {
            self.introspect_responses
                .lock()
                .unwrap()
                .remove(token)
                .unwrap_or_else(|| Err(IntrospectionError::Transport("no canned response".into())))
        }
        async fn userinfo(&self, token: &str) -> Result<Option<Value>, IntrospectionError> {
            self.userinfo_responses
                .lock()
                .unwrap()
                .remove(token)
                .unwrap_or(Ok(None))
        }
    }

    fn validator(client: Arc<MockIntrospectionClient>) -> IntrospectionValidator {
        IntrospectionValidator {
            client,
            principal_claim_name: "sub".into(),
            custom_claim_check: None,
            call_userinfo: false,
            allowable_clock_skew: secs(30),
            expected_audience: None,
            fallback_user_name_claim: None,
            fallback_user_name_prefix: None,
            groups_claim: None,
            groups_claim_delimiter: None,
        }
    }

    fn parse_jp(expr: &str) -> JpQuery {
        parse_json_path(expr).expect("expression compiles")
    }

    const NOW_MS: i64 = 1_700_000_000_000;

    #[tokio::test]
    async fn introspection_active_true_with_principal_returns_ok() {
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "tok",
            Ok(json!({"active": true, "sub": "alice", "exp": NOW_MS/1000 + 60})),
        );
        let v = validator(Arc::clone(&mock));
        let outcome = v.validate("tok", NOW_MS).await.unwrap();
        assert2::assert!(outcome.principal.name.as_str() == "alice");
        assert2::assert!(outcome.principal.auth_method == AuthMethod::SaslOAuthBearer);
    }

    #[tokio::test]
    async fn introspection_active_false_rejected() {
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect("tok", Ok(json!({"active": false})));
        let v = validator(Arc::clone(&mock));
        assert2::assert!(matches!(
            v.validate("tok", NOW_MS).await,
            Err(AuthError::InvalidToken)
        ));
    }

    #[tokio::test]
    async fn introspection_missing_active_field_rejected() {
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect("tok", Ok(json!({"sub": "alice"})));
        let v = validator(Arc::clone(&mock));
        assert2::assert!(matches!(
            v.validate("tok", NOW_MS).await,
            Err(AuthError::InvalidToken)
        ));
    }

    #[tokio::test]
    async fn introspection_expired_exp_rejected() {
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "tok",
            Ok(json!({"active": true, "sub": "alice", "exp": NOW_MS/1000 - 3600})),
        );
        let v = validator(Arc::clone(&mock));
        assert2::assert!(matches!(
            v.validate("tok", NOW_MS).await,
            Err(AuthError::InvalidToken)
        ));
    }

    // ---- IntrospectionValidator custom_claim_check -------------

    #[tokio::test]
    async fn introspection_validate_rejects_when_custom_claim_check_fails() {
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "tok",
            Ok(json!({
                "active": true,
                "sub": "alice",
                "exp": NOW_MS/1000 + 60,
                "scope": ["kafka.read"],
            })),
        );
        let mut v = validator(Arc::clone(&mock));
        v.custom_claim_check = Some(parse_jp("$.scope[?@ == 'kafka.admin']"));
        let result = v.validate("tok", NOW_MS).await;
        assert2::assert!(result.unwrap_err() == AuthError::InvalidToken);
    }

    #[tokio::test]
    async fn introspection_validate_accepts_when_custom_claim_check_passes() {
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "tok",
            Ok(json!({
                "active": true,
                "sub": "alice",
                "exp": NOW_MS/1000 + 60,
                "scope": ["kafka.admin", "kafka.read"],
            })),
        );
        let mut v = validator(Arc::clone(&mock));
        v.custom_claim_check = Some(parse_jp("$.scope[?@ == 'kafka.admin']"));
        let outcome = v.validate("tok", NOW_MS).await.expect("valid");
        assert2::assert!(outcome.principal.name == "alice");
    }

    // ---- IntrospectionValidator expected_audience --------------

    #[tokio::test]
    async fn introspection_honors_audience_string_and_array() {
        // Matching string `aud`.
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "tok",
            Ok(json!({"active": true, "sub": "a", "exp": NOW_MS/1000 + 60, "aud": "kafka"})),
        );
        let mut v = validator(Arc::clone(&mock));
        v.expected_audience = Some("kafka".to_owned());
        assert2::assert!(v.validate("tok", NOW_MS).await.is_ok());

        // Matching value inside an `aud` array.
        let mock2 = MockIntrospectionClient::arc();
        mock2.set_introspect(
            "tok",
            Ok(json!({
                "active": true, "sub": "a", "exp": NOW_MS/1000 + 60,
                "aud": ["other", "kafka"],
            })),
        );
        let mut v2 = validator(Arc::clone(&mock2));
        v2.expected_audience = Some("kafka".to_owned());
        assert2::assert!(v2.validate("tok", NOW_MS).await.is_ok());
    }

    #[tokio::test]
    async fn introspection_rejects_non_matching_audience() {
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "tok",
            Ok(json!({"active": true, "sub": "a", "exp": NOW_MS/1000 + 60, "aud": "web"})),
        );
        let mut v = validator(Arc::clone(&mock));
        v.expected_audience = Some("kafka".to_owned());
        assert2::assert!(v.validate("tok", NOW_MS).await == Err(AuthError::InvalidToken));
    }

    #[tokio::test]
    async fn introspection_rejects_missing_audience_when_expected() {
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "tok",
            Ok(json!({"active": true, "sub": "a", "exp": NOW_MS/1000 + 60})),
        );
        let mut v = validator(Arc::clone(&mock));
        v.expected_audience = Some("kafka".to_owned());
        assert2::assert!(v.validate("tok", NOW_MS).await == Err(AuthError::InvalidToken));
    }

    #[tokio::test]
    async fn introspection_ignores_audience_when_unset() {
        // Default (expected_audience == None): any `aud` is accepted.
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "tok",
            Ok(json!({"active": true, "sub": "a", "exp": NOW_MS/1000 + 60, "aud": "web"})),
        );
        let v = validator(Arc::clone(&mock));
        assert2::assert!(v.validate("tok", NOW_MS).await.is_ok());
    }

    #[tokio::test]
    async fn introspection_validate_does_not_check_valid_token_type() {
        // Introspection responses have no JWT header → typ check is N/A.
        // The struct doesn't even expose a valid_token_type field; this
        // is a regression test that validation passes regardless of any
        // hypothetical typ in the response (introspection responses
        // don't carry `typ`).
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "tok",
            Ok(json!({
                "active": true,
                "sub": "alice",
                "exp": NOW_MS/1000 + 60,
            })),
        );
        let v = validator(Arc::clone(&mock));
        let outcome = v.validate("tok", NOW_MS).await.expect("valid");
        assert2::assert!(outcome.principal.name == "alice");
    }

    #[tokio::test]
    async fn introspection_userinfo_claims_override_introspection_for_profile_keys() {
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "tok",
            Ok(json!({
                "active": true,
                "sub": "alice",
                "exp": NOW_MS/1000 + 60,
                "preferred_username": "intros-name",
            })),
        );
        mock.set_userinfo(
            "tok",
            Ok(Some(
                json!({"preferred_username": "userinfo-name", "email": "[email protected]"}),
            )),
        );
        let mut v = validator(Arc::clone(&mock));
        v.call_userinfo = true;
        v.principal_claim_name = "preferred_username".into();
        let outcome = v.validate("tok", NOW_MS).await.unwrap();
        assert2::assert!(outcome.principal.name == "userinfo-name");
    }

    #[tokio::test]
    async fn introspection_userinfo_does_not_override_authorization_keys() {
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "tok",
            Ok(json!({"active": true, "sub": "alice", "exp": NOW_MS/1000 + 60})),
        );
        mock.set_userinfo("tok", Ok(Some(json!({"active": false, "sub": "mallory"}))));
        let mut v = validator(Arc::clone(&mock));
        v.call_userinfo = true;
        let outcome = v.validate("tok", NOW_MS).await.unwrap();
        assert2::assert!(outcome.principal.name == "alice");
    }

    #[tokio::test]
    async fn introspection_userinfo_disabled_when_call_userinfo_false() {
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "tok",
            Ok(json!({"active": true, "sub": "alice", "exp": NOW_MS/1000 + 60})),
        );
        // Deliberately set a userinfo response — should be ignored.
        mock.set_userinfo("tok", Ok(Some(json!({"preferred_username": "ignored"}))));
        let v = validator(Arc::clone(&mock)); // call_userinfo: false (default)
        let outcome = v.validate("tok", NOW_MS).await.unwrap();
        assert2::assert!(outcome.principal.name == "alice");
    }

    #[tokio::test]
    async fn introspection_transport_error_becomes_introspection_transport() {
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "tok",
            Err(IntrospectionError::Transport("connection refused".into())),
        );
        let v = validator(Arc::clone(&mock));
        let err = v.validate("tok", NOW_MS).await.unwrap_err();
        assert2::assert!(
            matches!(err, AuthError::IntrospectionTransport(ref msg) if msg.contains("connection refused"))
        );
    }

    #[tokio::test]
    async fn introspection_default_principal_claim_sub() {
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "tok",
            Ok(json!({"active": true, "sub": "sub-name", "exp": NOW_MS/1000 + 60})),
        );
        let v = validator(Arc::clone(&mock));
        assert2::assert!(v.validate("tok", NOW_MS).await.unwrap().principal.name == "sub-name");
    }

    #[tokio::test]
    async fn introspection_custom_principal_claim_client_id() {
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "tok",
            Ok(json!({
                "active": true,
                "sub": "sub-name",
                "exp": NOW_MS/1000 + 60,
                "client_id": "my-client",
            })),
        );
        let mut v = validator(Arc::clone(&mock));
        v.principal_claim_name = "client_id".into();
        assert2::assert!(v.validate("tok", NOW_MS).await.unwrap().principal.name == "my-client");
    }

    #[tokio::test]
    async fn introspection_rejects_empty_fallback_principal_claim() {
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "tok",
            Ok(json!({
                "active": true,
                "exp": NOW_MS/1000 + 60,
                "client_id": "",
            })),
        );
        let mut v = validator(Arc::clone(&mock));
        v.fallback_user_name_claim = Some("client_id".into());
        assert2::assert!(v.validate("tok", NOW_MS).await == Err(AuthError::InvalidToken));
    }

    #[tokio::test]
    async fn enum_dispatch_introspection_async() {
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "tok",
            Ok(json!({"active": true, "sub": "alice", "exp": NOW_MS/1000 + 60})),
        );
        let v = validator(Arc::clone(&mock));
        let enum_v = OAuthBearerValidator::Introspection(v);
        let outcome = enum_v.validate("tok", NOW_MS).await.unwrap();
        assert2::assert!(outcome.principal.name == "alice");
    }

    #[tokio::test]
    async fn introspection_validate_surfaces_exp_from_introspection_response() {
        let exp_secs: i64 = 2_000;
        let now_ms: i64 = 1_000_000;
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "opaque-token",
            Ok(json!({
                "active": true,
                "sub": "alice",
                "exp": exp_secs,
                "scope": "kafka.write",
            })),
        );
        let v = validator(Arc::clone(&mock));
        let outcome = v
            .validate("opaque-token", now_ms)
            .await
            .expect("token valid");
        assert2::assert!(outcome.principal.name.as_str() == "alice");
        assert2::assert!(outcome.expires_at_ms == Some(exp_secs * 1000));
    }

    // ---- introspection parity -----------------------------------

    #[tokio::test]
    async fn introspection_validate_extracts_groups_from_introspection_response() {
        let exp_secs: i64 = 2_000;
        let now_ms: i64 = 1_000_000;
        let mock = MockIntrospectionClient::arc();
        mock.set_introspect(
            "opaque-token",
            Ok(json!({
                "active": true,
                "sub": "alice",
                "exp": exp_secs,
                "groups": ["admin", "ops"],
            })),
        );
        let mut v = validator(Arc::clone(&mock));
        v.groups_claim = Some(parse_jp("$.groups"));
        let outcome = v.validate("opaque-token", now_ms).await.expect("valid");
        assert2::assert!(outcome.principal.groups == vec!["admin".to_owned(), "ops".to_owned()]);
    }
}