ppoppo_sdk_core/verifier/

jwt.rs

//! Production [`BearerVerifier`] adapter — verifies PAS-issued JWTs.
//!
//! [`JwtVerifier`] is the only place inside the SDK that knows the
//! token format. It calls [`ppoppo_token::access_token::verify`] under a TTL-cached
//! JWKS ([`super::keyset::JwksCache`]), maps the engine's [`Claims`]
//! payload to an SDK-shaped [`VerifiedClaims`], and routes any
//! [`ppoppo_token::access_token::AuthError`] to the boundary's [`VerifyError`]
//! enum so audit logs retain the M-code via Display fallback.
//!
//! Single textbook constructor [`Self::from_jwks_url`] — Findings 2/3
//! collapsed `KeySet` and `with_config` out of the public surface.

use std::collections::BTreeMap;
use std::str::FromStr;
use std::sync::Arc;

use async_trait::async_trait;
use jiff::Timestamp;
use ppoppo_clock::ArcClock;
use ppoppo_clock::native::WallClock;

// Engine `VerifyConfig` is the cryptographic verify config (audience,
// issuer, required EpochEnforcement stance). Aliased to disambiguate from
// the SDK boundary's [`super::VerifyConfig`] (was `Expectations`) which
// carries the per-deployment iss/aud expectations folded into the SDK
// verifier at construction.
use ppoppo_token::access_token::{
    AuthError, Claims, EpochEnforcement, VerifyConfig as EngineVerifyConfig,
};
use ppoppo_token::SharedAuthError;
use super::jwks_cache::JwksCache;
use super::{BearerVerifier, VerifiedClaims, VerifyConfig, VerifyError};
use crate::audit::{AuditEvent, AuditSink, VerifyErrorKind};
use crate::session_liveness::{SessionLiveness, SessionLivenessError};
use crate::types::{Ppnum, PpnumId, SessionId};

/// PAS JWT verifier (RFC 9068, EdDSA).
///
/// Constructed once per consumer deployment with the JWKS URL and the
/// per-deployment [`VerifyConfig`]. Cheap to clone — internal state is
/// `Arc`-shared, so you can store the verifier behind
/// `Arc<dyn BearerVerifier>` in a request extension or per-route layer
/// without measurable overhead.
///
/// `JwksCache` and `ArcClock` have no useful `Debug` representation,
/// so this is a manual impl that surfaces only the expectations shape.
#[derive(Clone)]
pub struct JwtVerifier {
    jwks: JwksCache,
    expectations: VerifyConfig,
    clock: ArcClock,
    /// M48 audit emission port. `None` (the default from
    /// [`Self::from_jwks_url`]) means the verifier silently skips
    /// audit emission on failure — matching pre-Phase-9 behavior so
    /// existing consumers (RCW / CTW today) are unaffected. Opt in
    /// with [`Self::with_audit`].
    audit_sink: Option<Arc<dyn AuditSink>>,
    /// Required sv-axis stance (RFC_202607150428 Q3 — the former
    /// `Option<Arc<dyn EpochRevocation>>` silent default is retired).
    /// [`EpochEnforcement::Enforced`] gates every verify through the
    /// engine's `check_epoch`; the named non-`Enforced` stances skip
    /// the gate but are declared at construction via
    /// [`EpochEnforcement::declare`] (boot log visibility).
    epoch: EpochEnforcement,
    /// Phase 11.Z 0.10.0 (RFC_2026-05-08 §4.2 lock) — L2
    /// [`SessionLiveness`] port for per-request session-row revocation
    /// enforcement. `None` (the default) means the verifier
    /// short-circuits the L2 check — every verify admits past the
    /// session-row axis. Opt in with [`Self::with_session_liveness`]
    /// to enforce per-session logout / `LogoutAll` row-revocation.
    /// Skipped silently when the bearer's `sid` claim is `None`
    /// (machine credentials, AI-agent flows — lenient per the
    /// existing [`VerifiedClaims::session_id`] invariant).
    session_liveness: Option<Arc<dyn SessionLiveness>>,
}

impl std::fmt::Debug for JwtVerifier {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("JwtVerifier")
            .field("expectations", &self.expectations)
            .finish_non_exhaustive()
    }
}

impl JwtVerifier {
    /// Single textbook constructor — collapses {JWKS construction,
    /// VerifyConfig, VerifyConfig} into one boundary call.
    ///
    /// Consumer never sees the underlying `JwksCache`. The default
    /// engine `VerifyConfig::access_token(issuer, audience)` covers
    /// every concrete RCW/CTW/3rd-party scenario.
    ///
    /// **Required stance**: `epoch` is the sv-axis [`EpochEnforcement`]
    /// declaration (RFC_202607150428 Q3). There is no default — a
    /// consumer that does not enforce the axis must pass a NAMED
    /// [`EpochEnforcement::Unenforced`] (WARN-declared at boot) or
    /// [`EpochEnforcement::EnforcedElsewhere`]; production perimeters
    /// pass [`EpochEnforcement::Enforced`] over the canonical
    /// `Cache + Fetcher` composition (`pas_external::epoch::
    /// CompositeEpochRevocation`).
    ///
    /// **Builder family**:
    ///
    /// - [`Self::with_audit`] — wire an [`AuditSink`] for M48
    ///   verify-failure emission (Phase 9). Defaults to no emission;
    ///   wrap in [`crate::RateLimitedAuditSink`] for log-flood DoS
    ///   defense (M49).
    /// - [`Self::with_session_liveness`] — L2 session-row revocation.
    ///
    /// # Errors
    ///
    /// Returns [`VerifyError::KeysetUnavailable`] if the initial JWKS
    /// fetch fails. The verifier cannot serve verifications without
    /// at least one usable key snapshot.
    pub async fn from_jwks_url(
        jwks_url: impl Into<String>,
        expectations: VerifyConfig,
        epoch: EpochEnforcement,
    ) -> Result<Self, VerifyError> {
        let jwks = JwksCache::fetch(jwks_url).await?;
        epoch.declare(&format!("JwtVerifier[aud={}]", expectations.audience));
        Ok(Self {
            jwks,
            expectations,
            clock: Arc::new(WallClock),
            audit_sink: None,
            epoch,
            session_liveness: None,
        })
    }

    /// Wire an audit sink for M48 verify-failure emission.
    ///
    /// Every Err path inside [`BearerVerifier::verify`] will, after
    /// this builder, also emit an [`AuditEvent`] through the supplied
    /// sink. The sink is expected to be cheap (`Arc`-shared); it is
    /// not consulted on the success path.
    ///
    /// **Composition**: for log-flood DoS defense (M49), wrap the
    /// real sink in [`crate::RateLimitedAuditSink`] before passing
    /// here:
    ///
    /// ```ignore
    /// use std::sync::Arc;
    /// use pas_external::{
    ///     AuditSink, MemoryRateLimiter, JwtVerifier, RateLimitedAuditSink,
    /// };
    /// # async fn wire(real_sink: Arc<dyn AuditSink>) -> Result<(), Box<dyn std::error::Error>> {
    /// let limited: Arc<dyn AuditSink> = Arc::new(RateLimitedAuditSink::new(
    ///     real_sink,
    ///     Arc::new(MemoryRateLimiter::default()),
    /// ));
    /// let verifier = JwtVerifier::from_jwks_url(
    ///     "https://accounts.ppoppo.com/.well-known/jwks.json",
    ///     pas_external::VerifyConfig::new("accounts.ppoppo.com", "client-id"),
    ///     pas_external::EpochEnforcement::Unenforced {
    ///         reason: "doc example — pass Enforced(port) in production",
    ///     },
    /// )
    /// .await?
    /// .with_audit(limited);
    /// # let _ = verifier;
    /// # Ok(()) }
    /// ```
    ///
    /// Calling repeatedly replaces the sink (no chaining); the last
    /// call wins. By design, only one sink per verifier — composition
    /// for multiple destinations belongs in the adapter layer
    /// (a `FanoutAuditSink` is a future possibility).
    /// Inject an [`ArcClock`] for deterministic time control.
    ///
    /// Sets the clock used for:
    /// - `now: i64` passed to the token engine on every verify call
    /// - `created_at` timestamp in [`AuditEvent`] failure records
    /// - JWKS staleness check in the underlying [`JwksCache`]
    ///
    /// Defaults to [`WallClock`]. Tests inject `FrozenClock` to
    /// control time-sensitive behavior without sleeping.
    ///
    /// This builder is on `JwtVerifier` only (NOT on `BearerVerifier`
    /// trait) — no breaking change for consumers holding
    /// `Arc<dyn BearerVerifier>`.
    #[must_use]
    pub fn with_clock(mut self, clock: ArcClock) -> Self {
        self.jwks = self.jwks.with_clock(clock.clone());
        self.clock = clock;
        self
    }

    #[must_use]
    pub fn with_audit(mut self, sink: Arc<dyn AuditSink>) -> Self {
        self.audit_sink = Some(sink);
        self
    }

    /// Wire the L2 [`SessionLiveness`] port for per-request session-row
    /// revocation enforcement (Phase 11.Z 0.10.0, RFC_2026-05-08 §4.2
    /// lock). The L2 sibling of the required L1 `EpochEnforcement` ctor stance.
    ///
    /// With no port wired, the verifier short-circuits the L2 check —
    /// every token admits past the session-row axis (matching pre-0.10.0
    /// behavior, where L2 was inline-composed in the consumer's
    /// `AuthProvider::verify_token`).
    ///
    /// With a port wired, every verify call (after engine signature +
    /// claims + L1 sv-axis succeed) consults `port.check(sid)` for the
    /// bearer's `sid` claim:
    ///
    /// - `Ok(())` → admit.
    /// - `Err(`[`SessionLivenessError::Revoked`]`)` → reject as
    ///   [`VerifyError::SessionRevoked`] (HTTP 401 + clear cookies in
    ///   the perimeter).
    /// - `Err(`[`SessionLivenessError::Transient`]`)` → reject as
    ///   [`VerifyError::SessionLivenessLookupUnavailable`] (HTTP 503,
    ///   fail-closed per `STANDARDS_AUTH_INVALIDATION` §3).
    ///
    /// **Lenient on no-`sid` claim**: tokens without `sid` (machine
    /// credentials, AI-agent flows, R6 legacy admit per
    /// [`VerifiedClaims::session_id`]) admit without consulting the port.
    /// Non-session-bound tokens have no row to look up. RFC_2026-05-08
    /// §4.2 lock decision.
    ///
    /// Canonical wiring shape — RCW post-0.10.0:
    ///
    /// ```ignore
    /// use std::sync::Arc;
    /// use pas_external::SessionLiveness;
    ///
    /// # async fn wire(
    /// #     verifier: pas_external::JwtVerifier,
    /// #     liveness: Arc<dyn SessionLiveness>,
    /// # ) -> pas_external::JwtVerifier {
    /// // RCW's PgSessionLiveness adapter against scrcall.user_sessions.
    /// // CTW mirror: same shape over scctime.user_sessions.
    /// verifier.with_session_liveness(liveness)
    /// # }
    /// ```
    ///
    /// Calling repeatedly replaces the port (no chaining); the last
    /// call wins.
    #[must_use]
    pub fn with_session_liveness(mut self, port: Arc<dyn SessionLiveness>) -> Self {
        self.session_liveness = Some(port);
        self
    }

    /// Test-support ctor — constructs a verifier without performing
    /// a JWKS fetch.
    ///
    /// The internal keyset is empty, so any engine verify path
    /// rejects on `KidUnknown` (mapped to
    /// [`VerifyError::SignatureInvalid`] per `map_auth_error`).
    /// Adapter-side rejection paths
    /// ([`VerifyError::InvalidFormat`], [`VerifyError::IdTokenAsBearer`])
    /// are fully exercisable since they reject before consulting the
    /// keyset.
    ///
    /// Used by Phase 9.D's audit-emission integration tests to drive
    /// the verify path without a wiremock-shaped JWKS endpoint.
    /// NOT for production use — use [`Self::from_jwks_url`] instead.
    #[cfg(any(test, feature = "test-support"))]
    #[must_use]
    pub fn for_test_skip_fetch(expectations: VerifyConfig) -> Self {
        Self {
            jwks: JwksCache::for_test_empty(),
            expectations,
            clock: Arc::new(WallClock),
            audit_sink: None,
            // Provably inert: the empty keyset rejects every token at the
            // signature step, BEFORE the engine's epoch gate — no epoch
            // stance is reachable from this ctor. Declared, not silent.
            epoch: EpochEnforcement::Unenforced {
                reason: "test ctor (empty keyset — epoch gate unreachable)",
            },
            session_liveness: None,
        }
    }

    /// Test-support ctor with a **caller-supplied keyset**, so `verify` can run
    /// a full signature-passing path — reaching engine checks that run *after*
    /// signature verification, the `sv` epoch gate in particular — without a
    /// wiremock JWKS endpoint. Pair with `ppoppo_token::SigningKey::test_pair()`
    /// to mint tokens the returned verifier will accept.
    #[cfg(any(test, feature = "test-support"))]
    #[must_use]
    pub fn for_test_with_keys(
        expectations: VerifyConfig,
        keyset: ppoppo_token::KeySet,
        epoch: EpochEnforcement,
    ) -> Self {
        Self {
            jwks: JwksCache::for_test_with_keys(keyset),
            expectations,
            clock: Arc::new(WallClock),
            audit_sink: None,
            epoch,
            session_liveness: None,
        }
    }

    /// Internal: emit an audit event before returning a `VerifyError`.
    ///
    /// Returns the original `err` so call sites stay terse:
    ///
    /// ```ignore
    /// return Err(self.emit_failure(token, VerifyError::InvalidFormat).await);
    /// ```
    ///
    /// When no sink is wired, this is a no-op that returns the error
    /// directly. The `Option` check happens before any token-decode
    /// work so the no-audit path stays zero-overhead.
    async fn emit_failure(&self, bearer_token: &str, err: VerifyError) -> VerifyError {
        let Some(sink) = self.audit_sink.as_ref() else {
            return err;
        };

        let kind = VerifyErrorKind::from(&err);
        let (client_id_hint, kid_hint) = peek_token_hints(bearer_token);
        let mut metadata = BTreeMap::new();
        if let VerifyError::Other(msg) = &err {
            metadata.insert(
                "engine_msg".to_owned(),
                serde_json::Value::String(msg.clone()),
            );
        }
        let event = AuditEvent::from_hints(
            kind,
            self.clock.now(),
            client_id_hint,
            kid_hint,
            metadata,
        );
        sink.record_failure(event).await;
        err
    }
}

/// Map [`VerifyError`] to the audit-layer kind enum.
///
/// Lives here (in `token/jwt.rs`, gated by `well-known-fetch`) rather
/// than in `audit/sink.rs` to preserve the dependency direction: the
/// always-compiled audit module knows nothing about [`VerifyError`]
/// (which is gated behind `feature = "token"`); the JWT adapter, which
/// produces [`VerifyError`], maps to the audit-side kind enum at the
/// boundary. Inverting this would force `feature = "token"` to become
/// transitively required for the audit module, which is the opposite
/// of Phase 9 refinement #3.
impl From<&VerifyError> for VerifyErrorKind {
    fn from(err: &VerifyError) -> Self {
        match err {
            VerifyError::InvalidFormat => Self::InvalidFormat,
            VerifyError::SignatureInvalid => Self::SignatureInvalid,
            VerifyError::Expired => Self::Expired,
            VerifyError::IssuerInvalid => Self::IssuerInvalid,
            VerifyError::AudienceInvalid => Self::AudienceInvalid,
            VerifyError::MissingClaim(claim) => Self::MissingClaim((*claim).to_owned()),
            VerifyError::KeysetUnavailable => Self::KeysetUnavailable,
            VerifyError::IdTokenAsBearer => Self::IdTokenAsBearer,
            VerifyError::SessionVersionStale => Self::SessionVersionStale,
            VerifyError::SessionVersionLookupUnavailable => {
                Self::SessionVersionLookupUnavailable
            }
            VerifyError::SessionRevoked => Self::SessionRevoked,
            VerifyError::SessionLivenessLookupUnavailable => {
                Self::SessionLivenessLookupUnavailable
            }
            VerifyError::Other(_) => Self::Other,
        }
    }
}

/// Best-effort decode the rejected token's `client_id` (payload) and
/// `kid` (header) for audit-event hinting.
///
/// Returns `(None, None)` for any token that fails to parse — the
/// token has already been rejected by definition, so its claims are
/// untrusted; consumers MUST NOT treat absence as a security signal.
/// The hints exist purely for grouping / dashboard pivot.
///
/// Defensive decode mirrors [`peek_id_token_shape`] — no panics on
/// malformed input. A garbage token returns `(None, None)`, which
/// composes via [`crate::audit::compose_source_id`] into the
/// `"anon::nokid"` canonical bucket.
fn peek_token_hints(token: &str) -> (Option<String>, Option<String>) {
    use base64::Engine as _;

    let mut parts = token.split('.');
    let header_b64 = parts.next();
    let payload_b64 = parts.next();

    let kid_hint = header_b64.and_then(|h| {
        let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
            .decode(h)
            .ok()?;
        let value: serde_json::Value = serde_json::from_slice(&bytes).ok()?;
        value
            .get("kid")
            .and_then(|k| k.as_str())
            .map(|s| s.to_owned())
    });

    let client_id_hint = payload_b64.and_then(|p| {
        let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
            .decode(p)
            .ok()?;
        let value: serde_json::Value = serde_json::from_slice(&bytes).ok()?;
        value
            .get("client_id")
            .and_then(|c| c.as_str())
            .map(|s| s.to_owned())
    });

    (client_id_hint, kid_hint)
}

#[async_trait]
impl BearerVerifier for JwtVerifier {
    async fn verify(&self, bearer_token: &str) -> Result<VerifiedClaims, VerifyError> {
        // Adapter-side reject before engine entry — JWS Compact has
        // exactly three segments separated by `.`. The engine also
        // checks structurally, but rejecting upstream keeps the audit
        // log signal cleaner ("malformed at SDK boundary" vs "engine
        // M07/M11/M15").
        if bearer_token.is_empty() || !looks_like_jws_compact(bearer_token) {
            return Err(self.emit_failure(bearer_token, VerifyError::InvalidFormat).await);
        }

        // M73 — id_token presented as a Bearer. Refuse BEFORE engine
        // entry so the audit log gets a dedicated `IdTokenAsBearer`
        // signal instead of being masked by the engine's
        // `TypMismatch → SignatureInvalid` collapse (γ1 — defense in
        // depth: check both `typ="JWT"` header and `cat="id"` payload).
        // Order matters: a token with valid signature would otherwise
        // fail at `check_header::run` and surface as `SignatureInvalid`,
        // hiding the developer-misuse signal.
        if peek_id_token_shape(bearer_token) {
            return Err(self
                .emit_failure(bearer_token, VerifyError::IdTokenAsBearer)
                .await);
        }

        let cfg = EngineVerifyConfig::access_token(
            self.expectations.issuer.clone(),
            self.expectations.audience.clone(),
            self.epoch.clone(),
        );
        let keyset = self.jwks.snapshot().await;
        let now = self.clock.now().as_second();
        let claims = match ppoppo_token::access_token::verify(bearer_token, &cfg, &keyset, now).await {
            Ok(c) => c,
            Err(e) => {
                let mapped = map_auth_error(e, &self.expectations);
                return Err(self.emit_failure(bearer_token, mapped).await);
            }
        };

        // Phase 11.Z 0.10.0 — L2 session-row liveness check
        // (RFC_2026-05-08 §4.2 lock). Lenient on no-`sid`: tokens
        // without a `sid` claim admit without consulting the port,
        // matching the VerifiedClaims::session_id None invariant for
        // machine credentials / AI-agent flows / R6 legacy admit.
        if let Some(port) = self.session_liveness.as_ref()
            && let Some(sid_str) = claims.sid.as_deref() {
                let sid = SessionId::from(sid_str.to_owned());
                match port.check(&sid).await {
                    Ok(()) => {}
                    Err(SessionLivenessError::Revoked) => {
                        return Err(self
                            .emit_failure(bearer_token, VerifyError::SessionRevoked)
                            .await);
                    }
                    Err(SessionLivenessError::Transient(_)) => {
                        return Err(self
                            .emit_failure(
                                bearer_token,
                                VerifyError::SessionLivenessLookupUnavailable,
                            )
                            .await);
                    }
                }
            }
            // No `sid` claim → admit without consulting the port (lenient).

        match claims_to_verified(claims) {
            Ok(session) => Ok(session),
            Err(err) => Err(self.emit_failure(bearer_token, err).await),
        }
    }
}

fn looks_like_jws_compact(token: &str) -> bool {
    token.split('.').count() == 3
}

/// M73 — return `true` when the JWS Compact `token` carries an
/// id_token-shaped header or payload. γ1 (defense in depth): both
/// signals are cheap to read and either alone is conclusive.
///
/// **Header `typ`**: RFC 9068 access tokens carry `typ="at+jwt"`;
/// OIDC id_tokens carry `typ="JWT"`. Anything other than `at+jwt` is
/// treated as not-an-access-token here. We deliberately do NOT only
/// match the literal string `"JWT"` — a token whose header omits `typ`
/// or carries some third value (`id+jwt`, `application/jwt`) is also
/// not-an-access-token, and lumping them all into the M73 reject
/// surfaces the same audit signal: "this isn't the access token shape".
///
/// **Payload `cat`**: ppoppo's domain claim distinguishing access
/// (`cat="access"`) from id (`cat="id"`). Specifically `cat="id"` is
/// the conclusive id_token signal; absence is admitted (legacy /
/// non-ppoppo issuers may not carry it — the engine will surface that
/// downstream).
///
/// **Defense in depth**: catching both prevents two failure modes —
/// (a) a forger who sets `typ="at+jwt"` on a payload with `cat="id"`
/// (the engine's typ check would pass, M45 would reject `cat="id"` as
/// `UnknownClaim` since it's not in the access-allowlist, but only
/// after signature verify), and (b) a misconfigured IdP that emits
/// `typ="JWT"` on an access token (rare but possible during a buggy
/// rollout). Either alone returns `true`.
///
/// **Best-effort decode**: a malformed token (bad base64, bad JSON)
/// returns `false` — the engine's M01-M16a + M31-M34 are the
/// authoritative layer for parse rejection. We only signal positively
/// when we successfully decode AND find an id-token-shape marker.
pub(crate) fn peek_id_token_shape(token: &str) -> bool {
    use base64::Engine as _;

    let mut parts = token.split('.');
    let Some(header_b64) = parts.next() else { return false; };
    let Some(payload_b64) = parts.next() else { return false; };

    let header_bytes = match base64::engine::general_purpose::URL_SAFE_NO_PAD.decode(header_b64) {
        Ok(b) => b,
        Err(_) => return false,
    };
    let payload_bytes = match base64::engine::general_purpose::URL_SAFE_NO_PAD.decode(payload_b64) {
        Ok(b) => b,
        Err(_) => return false,
    };

    let header_json: serde_json::Value = match serde_json::from_slice(&header_bytes) {
        Ok(v) => v,
        Err(_) => return false,
    };
    let payload_json: serde_json::Value = match serde_json::from_slice(&payload_bytes) {
        Ok(v) => v,
        Err(_) => return false,
    };

    let typ = header_json.get("typ").and_then(|v| v.as_str());
    if matches!(typ, Some(t) if t != "at+jwt") {
        return true;
    }

    let cat = payload_json.get("cat").and_then(|v| v.as_str());
    if cat == Some("id") {
        return true;
    }

    false
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod m73_tests {
    //! M73 unit coverage for `peek_id_token_shape`.
    //!
    //! Tested at the helper level (not through `JwtVerifier::verify`)
    //! because constructing a `JwtVerifier` requires a live JWKS
    //! fetch; pulling in `wiremock` for one boundary check would be
    //! disproportionate. The helper is `pub(crate)`, the verify wiring
    //! is one `if peek...() { return Err(...); }` line, and the test
    //! covers every branch the wiring depends on.
    use super::*;
    use base64::Engine as _;

    fn forge(header: serde_json::Value, payload: serde_json::Value) -> String {
        let h = base64::engine::general_purpose::URL_SAFE_NO_PAD
            .encode(serde_json::to_vec(&header).unwrap());
        let p = base64::engine::general_purpose::URL_SAFE_NO_PAD
            .encode(serde_json::to_vec(&payload).unwrap());
        format!("{h}.{p}.<sig>")
    }

    #[test]
    fn typ_jwt_alone_is_id_token_shape() {
        // Even if `cat` says "access", `typ="JWT"` is the canonical
        // OIDC id_token header — not a Bearer-class access token.
        let token = forge(
            serde_json::json!({"alg": "EdDSA", "typ": "JWT", "kid": "k"}),
            serde_json::json!({"cat": "access", "sub": "01HSAB00000000000000000000"}),
        );
        assert!(peek_id_token_shape(&token));
    }

    #[test]
    fn cat_id_alone_is_id_token_shape() {
        // Even if `typ="at+jwt"` (header looks like access), `cat="id"`
        // is the conclusive ppoppo-domain signal. γ1 defense in depth.
        let token = forge(
            serde_json::json!({"alg": "EdDSA", "typ": "at+jwt", "kid": "k"}),
            serde_json::json!({"cat": "id", "sub": "01HSAB00000000000000000000"}),
        );
        assert!(peek_id_token_shape(&token));
    }

    #[test]
    fn typ_jwt_and_cat_id_both_signal() {
        // Canonical id_token shape — both markers present.
        let token = forge(
            serde_json::json!({"alg": "EdDSA", "typ": "JWT", "kid": "k"}),
            serde_json::json!({"cat": "id", "sub": "01HSAB00000000000000000000"}),
        );
        assert!(peek_id_token_shape(&token));
    }

    #[test]
    fn proper_access_token_shape_returns_false() {
        // Canonical RFC 9068 access token — `typ="at+jwt"` + `cat="access"`.
        // Must fall through to the engine for full verification.
        let token = forge(
            serde_json::json!({"alg": "EdDSA", "typ": "at+jwt", "kid": "k"}),
            serde_json::json!({"cat": "access", "sub": "01HSAB00000000000000000000"}),
        );
        assert!(!peek_id_token_shape(&token));
    }

    #[test]
    fn missing_typ_and_cat_admits_to_engine() {
        // No typ, no cat — ambiguous. The engine's M13 + M45 layers will
        // catch the violation downstream; the M73 peek doesn't itself
        // signal because it only fires on positive id_token markers.
        let token = forge(
            serde_json::json!({"alg": "EdDSA", "kid": "k"}),
            serde_json::json!({"sub": "01HSAB00000000000000000000"}),
        );
        assert!(!peek_id_token_shape(&token));
    }

    #[test]
    fn malformed_base64_returns_false_not_panic() {
        // Defense: a garbage token must defer to the engine for parse
        // rejection (it will surface as `InvalidFormat` /
        // `SignatureInvalid`), not crash here.
        assert!(!peek_id_token_shape("not.valid.token"));
        assert!(!peek_id_token_shape("!!!.@@@.###"));
        assert!(!peek_id_token_shape(""));
    }

    #[test]
    fn unrecognized_typ_value_is_id_token_shape() {
        // Anything other than `at+jwt` is treated as not-an-access-token
        // — surfaces the same audit signal as `typ="JWT"`. Rationale:
        // the BearerVerifier surface is meant to receive RFC 9068
        // tokens; any other `typ` value is a class mismatch.
        let token = forge(
            serde_json::json!({"alg": "EdDSA", "typ": "id+jwt", "kid": "k"}),
            serde_json::json!({"cat": "access", "sub": "01HSAB00000000000000000000"}),
        );
        assert!(peek_id_token_shape(&token));
    }
}

/// Map engine [`AuthError`] to the SDK boundary [`VerifyError`].
///
/// Algorithm + header rejections collapse to `SignatureInvalid` /
/// `InvalidFormat` because their semantic at the SDK boundary is "the
/// token cannot be trusted." Audit logs that need the precise M-code
/// route through the [`VerifyError::Other`] Display fallback (engine
/// `AuthError` Display retains the row identifier).
fn map_auth_error(err: AuthError, _expectations: &VerifyConfig) -> VerifyError {
    use AuthError as E;
    use SharedAuthError as S;
    match err {
        // JOSE wire-format errors travel via the `Jose` carrier variant
        // since the 10.1.B SharedAuthError split. Algorithm + header +
        // kid + typ collapse to SignatureInvalid; serialization-shape
        // (oversize / JSON-form / JWE / lax base64) collapse to
        // InvalidFormat. The audit log disambiguates via Display
        // fallback when the precise M-row matters.
        E::Jose(
            S::AlgNone
            | S::AlgNotWhitelisted
            | S::AlgHmacRejected
            | S::AlgRsaRejected
            | S::AlgEcdsaRejected
            | S::HeaderJku
            | S::HeaderX5u
            | S::HeaderJwk
            | S::HeaderX5c
            | S::HeaderCrit
            | S::HeaderExtraParam
            | S::HeaderB64False
            | S::KidUnknown
            | S::TypMismatch
            | S::NestedJws
            | S::DuplicateJsonKeys
            | S::HeaderUnparseable
            | S::PayloadUnparseable
            | S::NotJwsCompact,
        ) => VerifyError::SignatureInvalid,
        E::Jose(
            S::OversizedToken | S::JwsJsonRejected | S::JwePayload | S::LaxBase64,
        ) => VerifyError::InvalidFormat,
        // Time-bound rejections — caller treats these as "refresh".
        E::Expired | E::ExpUpperBound | E::IatFuture | E::NotYetValid => VerifyError::Expired,
        // Issuer / audience — typed for audit pivot.
        E::IssMismatch => VerifyError::IssuerInvalid,
        E::AudMismatch => VerifyError::AudienceInvalid,
        // Missing claims — surface the canonical claim name.
        E::ExpMissing => VerifyError::MissingClaim("exp"),
        E::AudMissing => VerifyError::MissingClaim("aud"),
        E::IatMissing => VerifyError::MissingClaim("iat"),
        E::JtiMissing => VerifyError::MissingClaim("jti"),
        E::SubMissing => VerifyError::MissingClaim("sub"),
        E::ClientIdMissing => VerifyError::MissingClaim("client_id"),
        // Phase 11.Z (RFC §3 Row 3) — sv-axis enforcement typed
        // variants. Pre-11.Z these collapsed into `Other(String)`
        // along with every engine variant the SDK didn't typify.
        E::SessionVersionStale => VerifyError::SessionVersionStale,
        E::SessionVersionLookupUnavailable => VerifyError::SessionVersionLookupUnavailable,
        // Catch-all keeps the engine row identifier in the Display so
        // the audit log can pivot on the M-code without the SDK
        // pre-translating every variant — engine evolution doesn't
        // require lockstep enum updates here.
        other => VerifyError::Other(other.to_string()),
    }
}

fn claims_to_verified(claims: Claims) -> Result<VerifiedClaims, VerifyError> {
    // `sub` is a ULID per the engine's Phase 4 domain rules. A failure
    // here would mean the issuer drifted from the contract — fail
    // closed with a typed error.
    let ppnum_id = ulid::Ulid::from_str(&claims.sub)
        .map(PpnumId)
        .map_err(|_| VerifyError::MissingClaim("sub"))?;

    let ppnum = match claims.active_ppnum {
        Some(p) => Some(Ppnum::try_from(p).map_err(|_| VerifyError::MissingClaim("active_ppnum"))?),
        None => None,
    };

    let session_id = claims.sid.map(SessionId::from);

    let expires_at = Timestamp::from_second(claims.exp)
        .map_err(|_| VerifyError::MissingClaim("exp"))?;

    Ok(VerifiedClaims::new(ppnum_id, ppnum, session_id, expires_at))
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod epoch_wiring_tests {
    //! The `sv` epoch gate reached **through** `JwtVerifier::verify` — the
    //! link `:3202` / RCW / CTW depend on: the required `EpochEnforcement`
    //! stance forwards the port into the engine `VerifyConfig` at verify
    //! time (RFC_202607150428 Q3 — the opt-in builder is retired). The
    //! engine's `check_epoch` reject logic and the
    //! `CompositeEpochRevocation` composer are tested elsewhere; this closes the
    //! one link between them — that an *Enforced* verifier actually consults the
    //! port and refuses, and that the named non-Enforced stances admit.
    //! `for_test_with_keys` supplies a real in-memory keyset so
    //! verify reaches post-signature checks without a wiremock JWKS endpoint,
    //! the pattern this crate deliberately keeps out of boundary tests
    //! (RFC_202607150428 S5/S6).
    use super::*;
    use ppoppo_token::SigningKey;
    use ppoppo_token::access_token::{
        EpochRevocation, EpochRevocationError, IssueConfig, IssueRequest, issue,
    };
    use std::sync::Arc;

    const ISSUER: &str = "accounts.ppoppo.com";
    const CLIENT: &str = "rcw-client-id";
    const SUB: &str = "01HSAB00000000000000000000";

    /// An `EpochRevocation` stub pinned to a fixed current epoch.
    #[derive(Debug)]
    struct FixedEpoch(i64);
    #[async_trait::async_trait]
    impl EpochRevocation for FixedEpoch {
        async fn current(&self, _sub: &str) -> Result<i64, EpochRevocationError> {
            Ok(self.0)
        }
    }

    fn mint(signer: &SigningKey, sv: Option<u64>) -> String {
        let cfg = IssueConfig::access_token(ISSUER, CLIENT, signer.kid());
        let mut req = IssueRequest::new(SUB, CLIENT, std::time::Duration::from_secs(3600))
            .with_account_type("human");
        if let Some(v) = sv {
            req = req.with_session_version(v);
        }
        let now = jiff::Timestamp::now().as_second();
        issue(&req, &cfg, signer, now).expect("mint")
    }

    fn wired(keyset: ppoppo_token::KeySet, current: i64) -> JwtVerifier {
        JwtVerifier::for_test_with_keys(
            VerifyConfig::new(ISSUER, CLIENT),
            keyset,
            EpochEnforcement::Enforced(Arc::new(FixedEpoch(current))),
        )
    }

    /// A wired verifier ADMITS a current token — anti-vacuous anchor: proves
    /// the rejections below are the port acting, not the verifier being inert.
    #[tokio::test]
    async fn wired_verifier_admits_current_sv() {
        let (signer, keyset) = SigningKey::test_pair();
        let token = mint(&signer, Some(5)); // 5 >= current 5
        wired(keyset, 5)
            .verify(&token)
            .await
            .expect("a token at the current epoch must be admitted");
    }

    /// A wired verifier REFUSES a token whose `sv` is behind the account epoch
    /// — proof that `JwtVerifier::verify` consults the port (break-glass reach).
    #[tokio::test]
    async fn wired_verifier_rejects_stale_sv() {
        let (signer, keyset) = SigningKey::test_pair();
        let token = mint(&signer, Some(3)); // 3 < current 10 → stale
        let err = wired(keyset, 10)
            .verify(&token)
            .await
            .expect_err("stale sv must reject");
        assert!(
            matches!(err, VerifyError::SessionVersionStale),
            "stale token must surface SessionVersionStale, got {err:?}",
        );
    }

    /// S6: a wired verifier REFUSES a token with **no `sv` claim** (R6
    /// legacy-admit retired). This is the missing-sv reject reaching through the
    /// SDK verify path — the exact enforcement `:3202` gains from S5-PCS+S6.
    #[tokio::test]
    async fn wired_verifier_rejects_missing_sv() {
        let (signer, keyset) = SigningKey::test_pair();
        let token = mint(&signer, None); // no sv claim
        let err = wired(keyset, 0)
            .verify(&token)
            .await
            .expect_err("missing sv must reject (S6)");
        assert!(
            matches!(err, VerifyError::SessionVersionStale),
            "missing-sv token must reject post-S6, got {err:?}",
        );
    }

    /// Q3 typed stance: a NAMED `Unenforced` verifier admits past the sv
    /// axis — the declared-gap semantics RCW/CTW run under until their
    /// substrate lands (RFC_202607150428 §6·S5). Pins that the stance
    /// short-circuit is behavior-identical to the retired `epoch: None`
    /// (a stale AND a missing sv both admit), so flipping a consumer to
    /// `Unenforced` is a declaration, never a behavior change.
    #[tokio::test]
    async fn unenforced_stance_admits_stale_and_missing_sv() {
        let (signer, keyset) = SigningKey::test_pair();
        let verifier = JwtVerifier::for_test_with_keys(
            VerifyConfig::new(ISSUER, CLIENT),
            keyset,
            EpochEnforcement::Unenforced {
                reason: "test: declared-gap semantics under verification",
            },
        );
        let stale = mint(&signer, Some(3));
        verifier
            .verify(&stale)
            .await
            .expect("Unenforced stance must admit a stale-sv token (declared gap)");
        let missing = mint(&signer, None);
        verifier
            .verify(&missing)
            .await
            .expect("Unenforced stance must admit a missing-sv token (declared gap)");
    }
}