pas_external/

refresh_source.rs

//! # Renewing `TokenSource` — refresh-grant + rotated-token persistence
//!
//! A [`RefreshTokenSource`] is a [`ppoppo_sdk_core::token_cache::TokenSource`]
//! for clients that hold a **durable refresh token** and want a
//! [`ppoppo_sdk_core::token_cache::TokenCache`] to renew the access token on
//! its own: on cache lapse the cache calls [`TokenSource::fetch_token`], which
//! runs the OAuth `refresh_token` grant, **persists the rotated refresh token**
//! (PAS applies RTR — RFC 9700 §2.2.2 — so every refresh returns a new token),
//! and hands back the fresh access token + its TTL.
//!
//! The persistence seam is [`TokenStore`]: the consumer plugs its keystore —
//! CNC uses the macOS Keychain; a short-lived process can use
//! [`MemoryTokenStore`]. RCW/CTW do **not** use this type — their OIDC RP flow
//! refreshes per request and re-stores inline; this is for native / CLI clients
//! that feed a long-lived `TokenCache`.
//!
//! ## The `Send` discipline
//!
//! [`TokenStore`] and [`TokenSource`] are `#[async_trait]` (their futures are
//! boxed `+ Send`), and [`crate::pas_port::PasAuthPort::refresh`] is `+ Send`.
//! So [`RefreshTokenSource`]'s `fetch_token` future is `Send` **by
//! construction** — the impl only compiles if its body holds `Send` values
//! across every `await`. Never introduce an `async` closure on this path: it
//! carries a higher-ranked lifetime whose `Send`-ness fails to prove on a
//! multi-threaded runtime (the yanked-0.5.2 lesson). A `#[tokio::test(flavor =
//! "multi_thread")]` that `spawn`s the future guards it.

use std::marker::PhantomData;
use std::time::Duration;

use async_trait::async_trait;
use ppoppo_sdk_core::scopes::{ConsentScopes, ScopedTokenSource};
use ppoppo_sdk_core::token_cache::{TokenCacheError, TokenSource};

use crate::pas_port::{PasAuthPort, PasFailure};
use crate::scope_grant::ensure_covers;
#[cfg(doc)]
use crate::oauth::AuthClient;

/// A [`TokenStore`] backend error. Consumers wrap their native keystore error
/// in the message; it is tunnelled through [`TokenCacheError::Source`] so the
/// SDK layer can surface it intact.
#[derive(Debug, thiserror::Error)]
#[error("token store error: {0}")]
pub struct TokenStoreError(String);

impl TokenStoreError {
    /// Wrap a backend failure message.
    #[must_use]
    pub fn new(message: impl Into<String>) -> Self {
        Self(message.into())
    }
}

/// Durable persistence seam for a rotated refresh token. All methods are
/// `Send`-safe for multi-threaded runtimes.
///
/// The consumer plugs the keystore: CNC → macOS Keychain, a short-lived process
/// → [`MemoryTokenStore`]. The store owns exactly one refresh token per session
/// (the latest, after rotation).
#[async_trait]
pub trait TokenStore: Send + Sync {
    /// The currently-persisted refresh token, or `None` if unauthenticated.
    async fn load(&self) -> Result<Option<String>, TokenStoreError>;
    /// Persist a (freshly-rotated) refresh token, replacing any prior one.
    async fn save(&self, refresh_token: &str) -> Result<(), TokenStoreError>;
    /// Drop the persisted refresh token (logout / terminal refresh failure).
    async fn clear(&self) -> Result<(), TokenStoreError>;
}

/// An ephemeral in-memory [`TokenStore`] — a reference adapter and test double.
///
/// **Not durable**: the token lives only for the process lifetime. Real clients
/// plug a persistent keystore (Keychain, encrypted DB column). Useful for a
/// short-lived CLI invocation or a consumer's tests.
#[derive(Default)]
pub struct MemoryTokenStore {
    inner: std::sync::Mutex<Option<String>>,
}

// Hand-written so the refresh token never reaches a log line. The derived
// impl printed `Some("<the actual refresh token>")` — a durable credential,
// rendered by the one trait people reach for while debugging. Reports
// whether a credential is held, which is the only part worth seeing.
impl std::fmt::Debug for MemoryTokenStore {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let held = self
            .inner
            .lock()
            .map(|g| g.is_some())
            .unwrap_or_else(|e| e.into_inner().is_some());
        f.debug_struct("MemoryTokenStore")
            .field("refresh_token", &if held { "[redacted]" } else { "[none]" })
            .finish()
    }
}

impl MemoryTokenStore {
    /// An empty store (unauthenticated).
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// A store pre-seeded with an initial refresh token (e.g. straight from the
    /// authorization_code exchange, before the first renewal).
    #[must_use]
    pub fn with_token(refresh_token: impl Into<String>) -> Self {
        Self { inner: std::sync::Mutex::new(Some(refresh_token.into())) }
    }
}

#[async_trait]
impl TokenStore for MemoryTokenStore {
    async fn load(&self) -> Result<Option<String>, TokenStoreError> {
        let guard = self.inner.lock().map_err(|_| TokenStoreError::new("lock poisoned"))?;
        Ok(guard.clone())
    }

    async fn save(&self, refresh_token: &str) -> Result<(), TokenStoreError> {
        let mut guard = self.inner.lock().map_err(|_| TokenStoreError::new("lock poisoned"))?;
        *guard = Some(refresh_token.to_owned());
        Ok(())
    }

    async fn clear(&self) -> Result<(), TokenStoreError> {
        let mut guard = self.inner.lock().map_err(|_| TokenStoreError::new("lock poisoned"))?;
        *guard = None;
        Ok(())
    }
}

/// A renewing [`TokenSource`]: mints access tokens via the OAuth `refresh_token`
/// grant and persists the rotated refresh token through a [`TokenStore`].
///
/// Depends on the [`PasAuthPort`] *port* (not the concrete `AuthClient`), so it
/// is unit-tested against an in-memory adapter with no HTTP. Wire it into a
/// [`ppoppo_sdk_core::token_cache::TokenCache`] so the cache renews on lapse.
///
/// ## The scope parameter — a per-refresh check, not a construction-time one
///
/// `Sc` is the tier this credential is expected to carry. Every refresh
/// validates the server's scope echo against it ([`ensure_covers`]) before
/// handing the access token back, so a grant that narrows **mid-session** —
/// which OAuth 2.1 §1.3.2 explicitly permits — surfaces as a terminal error
/// rather than as an unexplained 403 from the resource server later.
///
/// This is why the check lives here and not in a wrapper:
/// [`TokenSource::fetch_token`] returns only `(String, Duration)`, so nothing
/// composed *around* this type can ever see the scope echo. The seam has to be
/// inside.
///
/// That also makes [`ScopedTokenSource<Sc>`] a stronger witness than a
/// construction-time check would be. The marker holds for every instance
/// because the guarantee is intrinsic to *use*: any token this source yields
/// has just been checked. [`new`](Self::new) therefore needs no proof from its
/// caller — there is no window in which an unchecked token escapes.
pub struct RefreshTokenSource<P: PasAuthPort, T: TokenStore, Sc: ConsentScopes> {
    auth: P,
    store: T,
    _scope: PhantomData<Sc>,
}

impl<P: PasAuthPort, T: TokenStore, Sc: ConsentScopes> RefreshTokenSource<P, T, Sc> {
    /// `auth` performs the refresh grant (typically an [`AuthClient`] carrying
    /// the token endpoint + client_id + RFC 8707 resource); `store` holds the
    /// durable refresh token (seed it with the initial token first).
    ///
    /// `Sc` is usually inferred from the surrounding binding; give it
    /// explicitly (`RefreshTokenSource::<_, _, Notify>::new(..)`) where it is
    /// not.
    ///
    /// Most consumers do not call this directly — [`AuthClient`] cannot be
    /// constructed outside this crate, so a native client obtains a ready
    /// source from
    /// [`NativeAuthFlow::token_source`](crate::NativeAuthFlow::token_source)
    /// or [`NativeAuthFlow::complete`](crate::NativeAuthFlow::complete). This
    /// constructor is for consumers supplying their own [`PasAuthPort`].
    pub fn new(auth: P, store: T) -> Self {
        Self { auth, store, _scope: PhantomData }
    }
}

// Neither `P` nor `T` is required to be `Debug` (a consumer's keystore
// adapter usually is not), so this reports the tier and nothing else — which
// is the field actually worth seeing when a credential misbehaves.
impl<P: PasAuthPort, T: TokenStore, Sc: ConsentScopes> std::fmt::Debug
    for RefreshTokenSource<P, T, Sc>
{
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RefreshTokenSource")
            .field("scope", &Sc::scope_line())
            .finish_non_exhaustive()
    }
}

/// The witness: every token this source yields has passed [`ensure_covers`]
/// against `Sc`. See the type's docs for why the guarantee attaches to use
/// rather than construction.
impl<P: PasAuthPort, T: TokenStore, Sc: ConsentScopes> ScopedTokenSource<Sc>
    for RefreshTokenSource<P, T, Sc>
{
}

#[async_trait]
impl<P: PasAuthPort, T: TokenStore, Sc: ConsentScopes> TokenSource
    for RefreshTokenSource<P, T, Sc>
{
    async fn fetch_token(&self) -> Result<(String, Duration), TokenCacheError> {
        let refresh_token = self
            .store
            .load()
            .await
            .map_err(|e| TokenCacheError::Source(Box::new(e)))?
            .ok_or_else(|| {
                TokenCacheError::Source(Box::new(TokenStoreError::new(
                    "no refresh token stored — client is not authenticated",
                )))
            })?;

        match self.auth.refresh(&refresh_token).await {
            Ok(response) => {
                // The refresh-leg grant check (OAuth 2.1 §1.3.2 permits a
                // refresh response to narrow the scope). Runs BEFORE the
                // rotated token is persisted and before the access token is
                // handed out: if the tier is gone, neither the caller nor the
                // store should advance on a credential that no longer means
                // what `Sc` says it means.
                //
                // Deliberately does NOT clear the store. Unlike the 4xx arm
                // below, this credential is not dead — it is merely narrower
                // than this source's tier. Clearing would conflate "the server
                // reduced your grant" with "your session was revoked", and
                // destroy a refresh token that is still perfectly valid for a
                // client built at a lower tier. Recovery is re-authorization
                // (which re-consents and mints a new token anyway), and that is
                // the consumer's call to make.
                ensure_covers::<Sc>(response.scope.as_deref())
                    .map_err(|e| TokenCacheError::Fetch(format!("refresh {e}")))?;

                // RTR: persist the rotated token so the NEXT renewal presents it.
                // Absent (a non-rotating server) → keep the current token.
                if let Some(rotated) = response.refresh_token.as_deref() {
                    self.store
                        .save(rotated)
                        .await
                        .map_err(|e| TokenCacheError::Source(Box::new(e)))?;
                }
                let ttl = Duration::from_secs(response.expires_in.unwrap_or(3600));
                Ok((response.access_token, ttl))
            }
            // 4xx: the refresh token is dead (expired / revoked / logged out
            // elsewhere, incl. RTR family-invalidation). Terminal — clear the
            // store so the next load reads "unauthenticated" rather than
            // replaying a dead token; the consumer must re-authenticate.
            Err(PasFailure::Rejected { detail, .. }) => {
                // Best-effort: we are already returning a terminal error.
                let _ = self.store.clear().await;
                Err(TokenCacheError::Fetch(format!(
                    "refresh rejected (credential dead): {detail}"
                )))
            }
            // 5xx / transport: transient. Keep the token; the caller may retry.
            Err(PasFailure::ServerError { detail, .. }) => {
                Err(TokenCacheError::Fetch(format!("PAS refresh 5xx: {detail}")))
            }
            Err(PasFailure::Transport { detail }) => {
                Err(TokenCacheError::Fetch(format!("PAS refresh transport error: {detail}")))
            }
        }
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use super::*;
    use crate::oauth::TokenResponse;
    use std::sync::Mutex;

    /// The tier under test — mirrors `pcs_session::scopes::Notify`, which is
    /// not reachable from here (no SDK↔SDK edge, by design).
    struct Notify;
    impl ConsentScopes for Notify {
        const SCOPES: &'static [&'static str] = &["chat.read", "contact.read", "chat.ack"];
    }

    /// Shorthand: the concrete source used by every test below.
    fn source<T: TokenStore>(auth: MockAuth, store: T) -> RefreshTokenSource<MockAuth, T, Notify> {
        RefreshTokenSource::new(auth, store)
    }

    /// A canned `PasAuthPort` — returns a preset outcome, recording the refresh
    /// token it was presented so a test can assert the rotation chain.
    struct MockAuth {
        outcome: Mutex<Result<TokenResponse, PasFailure>>,
        seen: Mutex<Vec<String>>,
    }

    impl MockAuth {
        fn ok(access: &str, rotated: Option<&str>, expires_in: Option<u64>) -> Self {
            Self::with_scope(access, rotated, expires_in, None)
        }

        /// `scope` is the RFC 6749 §5.1 grant echo: `None` = granted as
        /// requested, `Some` = the exact atoms the server conceded.
        fn with_scope(
            access: &str,
            rotated: Option<&str>,
            expires_in: Option<u64>,
            scope: Option<&str>,
        ) -> Self {
            Self {
                outcome: Mutex::new(Ok(TokenResponse {
                    access_token: access.to_owned(),
                    token_type: "Bearer".to_owned(),
                    expires_in,
                    refresh_token: rotated.map(str::to_owned),
                    id_token: None,
                    scope: scope.map(str::to_owned),
                })),
                seen: Mutex::new(Vec::new()),
            }
        }
        fn rejected() -> Self {
            Self {
                outcome: Mutex::new(Err(PasFailure::Rejected {
                    status: 400,
                    detail: "invalid_grant".to_owned(),
                })),
                seen: Mutex::new(Vec::new()),
            }
        }
    }

    // `PasAuthPort::refresh` is native RPITIT (`-> impl Future + Send`), not
    // `#[async_trait]` — the mock matches that shape. The returned future owns
    // `token` and borrows `&self` (Send, since `MockAuth: Sync`).
    impl PasAuthPort for MockAuth {
        fn refresh(
            &self,
            refresh_token: &str,
        ) -> impl std::future::Future<Output = Result<TokenResponse, PasFailure>> + Send {
            let token = refresh_token.to_owned();
            async move {
                self.seen.lock().unwrap().push(token);
                self.outcome.lock().unwrap().clone()
            }
        }
    }

    #[tokio::test]
    async fn rotation_is_persisted_to_the_store() {
        let auth = MockAuth::ok("AT-1", Some("RT-2"), Some(900));
        let store = MemoryTokenStore::with_token("RT-1");
        let source = source(auth, store);

        let (access, ttl) = source.fetch_token().await.expect("fetch");
        assert_eq!(access, "AT-1");
        assert_eq!(ttl, Duration::from_secs(900));
        // The presented token was RT-1; the rotated RT-2 is now persisted.
        assert_eq!(source.auth.seen.lock().unwrap().as_slice(), ["RT-1"]);
        assert_eq!(source.store.load().await.unwrap().as_deref(), Some("RT-2"));

        // A second renewal presents the rotated token, not the original.
        let _ = source.fetch_token().await.expect("second fetch");
        assert_eq!(source.auth.seen.lock().unwrap().as_slice(), ["RT-1", "RT-2"]);
    }

    #[tokio::test]
    async fn absent_rotation_keeps_the_current_token() {
        let auth = MockAuth::ok("AT-1", None, None);
        let store = MemoryTokenStore::with_token("RT-1");
        let source = source(auth, store);

        let (_, ttl) = source.fetch_token().await.expect("fetch");
        assert_eq!(ttl, Duration::from_secs(3600), "missing expires_in defaults to 1h");
        assert_eq!(source.store.load().await.unwrap().as_deref(), Some("RT-1"));
    }

    #[tokio::test]
    async fn missing_credential_is_a_source_error() {
        let auth = MockAuth::ok("AT", Some("RT-2"), None);
        let source = source(auth, MemoryTokenStore::new());
        let err = source.fetch_token().await.expect_err("empty store must fail");
        assert!(matches!(err, TokenCacheError::Source(_)));
    }

    #[tokio::test]
    async fn rejected_refresh_clears_the_store() {
        let auth = MockAuth::rejected();
        let store = MemoryTokenStore::with_token("RT-dead");
        let source = source(auth, store);

        let err = source.fetch_token().await.expect_err("dead credential must fail");
        assert!(matches!(err, TokenCacheError::Fetch(_)));
        // The dead token is cleared so no retry replays it.
        assert_eq!(source.store.load().await.unwrap(), None);
    }

    /// OAuth 2.1 §1.3.2 lets a refresh response narrow the scope. Before the
    /// check existed, this returned a perfectly good-looking access token and
    /// the loss surfaced as an unexplained 403 from PCS on the next call that
    /// happened to need `chat.ack`.
    #[tokio::test]
    async fn narrowed_refresh_is_a_terminal_error_naming_the_lost_atom() {
        let auth = MockAuth::with_scope("AT-1", Some("RT-2"), None, Some("chat.read contact.read"));
        let store = MemoryTokenStore::with_token("RT-1");
        let source = source(auth, store);

        let err = source.fetch_token().await.expect_err("a narrowed grant must not pass");
        assert!(matches!(err, TokenCacheError::Fetch(_)));
        assert!(err.to_string().contains("chat.ack"), "must name the lost atom: {err}");

        // The credential is NOT cleared: it is narrower, not dead. Recovery is
        // re-authorization, which the consumer decides — clearing here would
        // conflate a reduced grant with a revoked session.
        assert_eq!(source.store.load().await.unwrap().as_deref(), Some("RT-1"));
    }

    /// The rotated token must not be persisted when the grant check fails —
    /// otherwise the store advances to a token whose tier no longer matches
    /// this source, and the next refresh reports the same failure from a
    /// different starting point.
    #[tokio::test]
    async fn a_failed_grant_check_does_not_advance_the_stored_token() {
        let auth = MockAuth::with_scope("AT-1", Some("RT-2"), None, Some("chat.read"));
        let source = source(auth, MemoryTokenStore::with_token("RT-1"));

        let _ = source.fetch_token().await.expect_err("narrowed grant");
        assert_eq!(
            source.store.load().await.unwrap().as_deref(),
            Some("RT-1"),
            "rotation must not be persisted past a failed grant check"
        );
    }

    /// RFC 6749 §5.1: an omitted echo means "granted as requested". A
    /// published SDK must accept that rather than fail closed on it.
    #[tokio::test]
    async fn absent_scope_echo_passes_the_grant_check() {
        let source = source(
            MockAuth::ok("AT-1", Some("RT-2"), None),
            MemoryTokenStore::with_token("RT-1"),
        );
        let (access, _) = source.fetch_token().await.expect("absent echo == granted as requested");
        assert_eq!(access, "AT-1");
    }

    /// A server granting more than asked is not an error.
    #[tokio::test]
    async fn superset_scope_echo_passes_the_grant_check() {
        let auth = MockAuth::with_scope(
            "AT-1",
            None,
            None,
            Some("chat.read contact.read chat.ack chat.send"),
        );
        let source = source(auth, MemoryTokenStore::with_token("RT-1"));
        assert!(source.fetch_token().await.is_ok());
    }

    /// The load-bearing `Send` guard: on a multi-thread runtime, `spawn`
    /// requires the future be `Send`. This only compiles if `fetch_token`'s
    /// body is `Send` across every await (the yanked-0.5.2 regression).
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn fetch_token_future_is_send() {
        let source = source(
            MockAuth::ok("AT", Some("RT-2"), None),
            MemoryTokenStore::with_token("RT-1"),
        );
        let handle = tokio::spawn(async move { source.fetch_token().await });
        let (access, _) = handle.await.expect("join").expect("fetch");
        assert_eq!(access, "AT");
    }
}