origin_auth/
provider.rs

1use crate::{AuthorizationFlow, TokenSet, TokenStore};
2use origin_domain::{AccountId, AppError, Clock, ConnectorId, Result};
3use origin_secrets::Secret;
4use std::sync::Arc;
5use time::Duration;
6use tokio::sync::Mutex;
7
8/// Refresh this long before the token actually expires, to survive clock skew and the
9/// time the request spends in flight.
10const REFRESH_SKEW: Duration = Duration::seconds(60);
11
12/// Hands out a valid access token, refreshing when needed.
13///
14/// Connectors depend on this rather than on [`TokenStore`], so nothing outside this
15/// type has to reason about expiry.
16#[derive(Debug)]
17pub struct AccessTokenProvider {
18    connector: ConnectorId,
19    flow: AuthorizationFlow,
20    tokens: TokenStore,
21    clock: Arc<dyn Clock>,
22    /// Serialises refreshes.
23    ///
24    /// Without it, ten concurrent requests on an expired token trigger ten refreshes,
25    /// and a provider that rotates refresh tokens invalidates nine of them.
26    refresh_lock: Mutex<()>,
27}
28
29impl AccessTokenProvider {
30    pub fn new(
31        connector: ConnectorId,
32        flow: AuthorizationFlow,
33        tokens: TokenStore,
34        clock: Arc<dyn Clock>,
35    ) -> Self {
36        Self {
37            connector,
38            flow,
39            tokens,
40            clock,
41            refresh_lock: Mutex::new(()),
42        }
43    }
44
45    /// A usable access token for `account`.
46    ///
47    /// Fails with `Authentication` when the account has never been connected, or when
48    /// the token expired and cannot be refreshed — both mean the user has to act.
49    pub async fn access_token(&self, account: &AccountId) -> Result<Secret> {
50        let current = self.load(account).await?;
51
52        if !current.expires_within(self.clock.as_ref(), REFRESH_SKEW) {
53            return Ok(current.access_token);
54        }
55
56        let _guard = self.refresh_lock.lock().await;
57
58        // Another task may have refreshed while we waited for the lock.
59        let current = self.load(account).await?;
60        if !current.expires_within(self.clock.as_ref(), REFRESH_SKEW) {
61            return Ok(current.access_token);
62        }
63
64        let refreshed = self.refresh(account, &current).await?;
65        Ok(refreshed.access_token)
66    }
67
68    /// Force a refresh, e.g. after a `401` from an API that expires tokens early.
69    pub async fn force_refresh(&self, account: &AccountId) -> Result<Secret> {
70        let _guard = self.refresh_lock.lock().await;
71        let current = self.load(account).await?;
72        Ok(self.refresh(account, &current).await?.access_token)
73    }
74
75    /// Store the tokens of a freshly authorized account.
76    pub async fn store(&self, account: &AccountId, tokens: &TokenSet) -> Result<()> {
77        self.tokens.save(&self.connector, account, tokens).await
78    }
79
80    /// Forget an account's credentials.
81    pub async fn forget(&self, account: &AccountId) -> Result<()> {
82        self.tokens.delete(&self.connector, account).await
83    }
84
85    async fn load(&self, account: &AccountId) -> Result<TokenSet> {
86        self.tokens
87            .load(&self.connector, account)
88            .await?
89            .ok_or_else(|| {
90                AppError::Authentication(format!(
91                    "account {account} is not connected to {}",
92                    self.connector
93                ))
94            })
95    }
96
97    async fn refresh(&self, account: &AccountId, current: &TokenSet) -> Result<TokenSet> {
98        let Some(refresh_token) = current.refresh_token.as_ref() else {
99            return Err(AppError::Authentication(format!(
100                "the session for {account} expired and cannot be renewed — please \
101                 reconnect the account"
102            )));
103        };
104
105        tracing::info!(connector = %self.connector, %account, "refreshing access token");
106
107        let refreshed = match self.flow.refresh(refresh_token.expose()).await {
108            Ok(refreshed) => refreshed,
109            Err(error) => {
110                // A rejected refresh token is final: keeping it would retry forever.
111                if error.kind() == origin_domain::ErrorKind::Authentication {
112                    tracing::warn!(%account, "refresh token rejected, discarding credentials");
113                    self.tokens.delete(&self.connector, account).await?;
114                }
115                return Err(error);
116            }
117        };
118
119        let merged = current.merge_refreshed(refreshed);
120        self.tokens.save(&self.connector, account, &merged).await?;
121        Ok(merged)
122    }
123}
origin_auth/provider.rs

origin_auth/
provider.rs
