ppoppo_sdk_core/

session_liveness.rs

//! [`SessionLiveness`] port — per-request session-row liveness check.
//!
//! Phase 11.Z 0.10.0 (RFC_2026-05-08 §4.2 lock). Promoted from RCW + CTW's
//! identical consumer-local `check_session_alive` shape.
//!
//! # The L2 axis (vs the L1 sv-axis)
//!
//! - **L1 sv-axis** ([`crate::epoch`]): "is the token's `sv` claim still
//!   current against PAS's authoritative substrate?" — answered against
//!   the canonical `sv:{sub}` cache + a fetcher fallback. Catches
//!   break-glass / LogoutAll propagation across services.
//! - **L2 session liveness** (this module): "is the bearer token's
//!   session row still alive in the consumer's own DB?" — answered by
//!   the consumer's per-deployment substrate. Catches per-session
//!   logout (`revoked_at IS NOT NULL`) without depending on PAS.
//!
//! Both axes are wired into [`crate::PasJwtVerifier`] as engine slots
//! ([`crate::PasJwtVerifier::with_epoch_revocation`] for L1,
//! [`crate::PasJwtVerifier::with_session_liveness`] for L2). With no
//! port wired, the verifier short-circuits the corresponding check.
//!
//! # Why this lives in `session_liveness` (alongside `TokenCipher`)
//!
//! The existing `session_liveness` module ships `TokenCipher` +
//! [`super::attempt_liveness_refresh`] for the periodic *PAS-callback*
//! refresh-token check (consumer asks PAS "is my refresh_token still
//! good?"). This new port is the per-request *consumer-DB* row check.
//! Both answer "is this user's session valid?" at different layers and
//! cadences — one shared umbrella module keeps the surface coherent.

use async_trait::async_trait;

use crate::types::SessionId;

/// Per-request session-row liveness check.
///
/// Wired into [`crate::PasJwtVerifier::with_session_liveness`] as a
/// verifier slot symmetric to
/// [`crate::PasJwtVerifier::with_epoch_revocation`]. With no port wired,
/// the verifier short-circuits the L2 check (matches pre-0.10.0
/// behavior).
///
/// # 3-state contract
///
/// - `Ok(())` → session is live; admit the token.
/// - `Err(`[`SessionLivenessError::Revoked`]`)` → session row absent OR
///   `revoked_at` is set; the verifier maps to
///   [`crate::VerifyError::SessionRevoked`]. Actionable for `LogoutAll`
///   / per-session-revoke flows.
/// - `Err(`[`SessionLivenessError::Transient`]`)` → substrate down (DB
///   connection lost, schema unavailable, etc.); the verifier maps to
///   [`crate::VerifyError::SessionLivenessLookupUnavailable`] (HTTP 503).
///   Fail-closed per `STANDARDS_AUTH_INVALIDATION` §3.
///
/// # Lenient on no-`sid` claim
///
/// When the bearer's `sid` claim is `None` (machine credentials,
/// AI-agent flows, R6 legacy admit per
/// [`crate::AuthSession::session_id`]), the verifier admits without
/// consulting this port — non-session-bound tokens have no row to look
/// up. RFC_2026-05-08 §4.2 lock decision (lenient — matches the existing
/// `AuthSession::session_id` invariant).
///
/// # Implementations
///
/// Consumer-side adapters: RCW ships `PgSessionLiveness` over
/// `scrcall.user_sessions`; CTW ships the same shape over
/// `scctime.user_sessions`. Each is ~10 lines of consumer-local
/// code — schema name + DB pool are deployment-specific and never
/// shipped from the SDK.
///
/// ```ignore
/// use async_trait::async_trait;
/// use pas_external::session_liveness::{SessionLiveness, SessionLivenessError};
/// use pas_external::types::SessionId;
/// use sqlx::PgPool;
///
/// pub struct PgSessionLiveness { pool: PgPool }
///
/// #[async_trait]
/// impl SessionLiveness for PgSessionLiveness {
///     async fn check(&self, sid: &SessionId) -> Result<(), SessionLivenessError> {
///         let row: Option<(Option<time::OffsetDateTime>,)> =
///             sqlx::query_as("SELECT revoked_at FROM scrcall.user_sessions WHERE id = $1")
///                 .bind(&sid.0)
///                 .fetch_optional(&self.pool)
///                 .await
///                 .map_err(|e| SessionLivenessError::Transient(format!("session lookup: {e}")))?;
///         match row {
///             None | Some((Some(_),)) => Err(SessionLivenessError::Revoked),
///             Some((None,)) => Ok(()),
///         }
///     }
/// }
/// ```
#[async_trait]
pub trait SessionLiveness: std::fmt::Debug + Send + Sync {
    async fn check(&self, sid: &SessionId) -> Result<(), SessionLivenessError>;
}

/// Per-request liveness failure surface.
///
/// Two variants — `Revoked` (substrate said "no") and `Transient`
/// (substrate couldn't answer) — collapse to typed
/// [`crate::VerifyError`] variants in the verifier so audit logs pivot
/// L2-revoked vs L2-substrate-down distinct from L1 sv-axis events.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum SessionLivenessError {
    /// Session row absent OR `revoked_at` is set. Token rejected as
    /// [`crate::VerifyError::SessionRevoked`]. The two surface causes
    /// (absent row vs revoked row) collapse to one variant because
    /// audit-time distinction is rarely actionable: a user whose row
    /// was deleted is in the same security state as one whose row was
    /// flipped to revoked. Substrate-side logging records the cause if
    /// needed.
    #[error("session revoked or not found")]
    Revoked,
    /// Substrate could not answer (connection error, query timeout,
    /// schema unavailable). Carries a free-form detail for audit logs;
    /// the verifier collapses every Transient variant onto
    /// [`crate::VerifyError::SessionLivenessLookupUnavailable`]
    /// (fail-closed) so the engine sees a uniform contract regardless
    /// of substrate flavor.
    #[error("session liveness substrate unavailable: {0}")]
    Transient(String),
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    //! Pins the contract — Display strings, error categorization, and
    //! the no-`sid` lenient-skip via in-process port construction. The
    //! verifier-slot integration is exercised by
    //! `tests/session_liveness_lookup_boundary.rs`.
    use super::*;
    use std::sync::Mutex;

    #[derive(Debug)]
    struct CountingLiveness {
        responses: Mutex<Vec<Result<(), SessionLivenessError>>>,
        calls: Mutex<u32>,
    }

    #[async_trait]
    impl SessionLiveness for CountingLiveness {
        async fn check(&self, _sid: &SessionId) -> Result<(), SessionLivenessError> {
            *self.calls.lock().unwrap() += 1;
            self.responses
                .lock()
                .unwrap()
                .pop()
                .unwrap_or_else(|| Err(SessionLivenessError::Transient("exhausted".into())))
        }
    }

    #[tokio::test]
    async fn live_admits() {
        let port = CountingLiveness {
            responses: Mutex::new(vec![Ok(())]),
            calls: Mutex::new(0),
        };
        let sid = SessionId::from("01HZAA00000000000000000000".to_string());
        assert!(matches!(port.check(&sid).await, Ok(())));
    }

    #[tokio::test]
    async fn revoked_surfaces_revoked_variant() {
        let port = CountingLiveness {
            responses: Mutex::new(vec![Err(SessionLivenessError::Revoked)]),
            calls: Mutex::new(0),
        };
        let sid = SessionId::from("01HZAA00000000000000000000".to_string());
        assert!(matches!(port.check(&sid).await, Err(SessionLivenessError::Revoked)));
    }

    #[tokio::test]
    async fn transient_carries_detail_for_audit() {
        let port = CountingLiveness {
            responses: Mutex::new(vec![Err(SessionLivenessError::Transient(
                "connection refused: pgpool dead".into(),
            ))]),
            calls: Mutex::new(0),
        };
        let sid = SessionId::from("01HZAA00000000000000000000".to_string());
        match port.check(&sid).await {
            Err(SessionLivenessError::Transient(detail)) => {
                assert!(detail.contains("connection refused"), "{detail}");
            }
            other => panic!("expected Transient, got {other:?}"),
        }
    }

    #[test]
    fn display_strings_stable_for_audit_pivot() {
        // Audit dashboards may match on these strings — keep them stable
        // unless deliberately rewriting the audit pivot path. The eng
        // log format is the same as the engine's transient/stale
        // disambiguation in STANDARDS_AUTH_INVALIDATION §3.
        assert_eq!(
            SessionLivenessError::Revoked.to_string(),
            "session revoked or not found"
        );
        assert_eq!(
            SessionLivenessError::Transient("foo".into()).to_string(),
            "session liveness substrate unavailable: foo"
        );
    }
}