Skip to main content

matrix_sdk/authentication/oauth/
cross_process.rs

1use std::sync::Arc;
2
3#[cfg(feature = "e2e-encryption")]
4use matrix_sdk_base::crypto::{
5    CryptoStoreError,
6    store::{LockableCryptoStore, Store},
7};
8use matrix_sdk_common::cross_process_lock::{
9    CrossProcessLock, CrossProcessLockError, CrossProcessLockGuard,
10};
11use sha2::{Digest as _, Sha256};
12use thiserror::Error;
13use tokio::sync::{Mutex, OwnedMutexGuard};
14use tracing::trace;
15
16use crate::SessionTokens;
17
18/// Key in the database for the custom value holding the current session tokens
19/// hash.
20const OIDC_SESSION_HASH_KEY: &str = "oidc_session_hash";
21
22/// Newtype to identify that a value is a session tokens' hash.
23#[derive(Clone, PartialEq, Eq)]
24struct SessionHash(Vec<u8>);
25
26impl SessionHash {
27    fn to_hex(&self) -> String {
28        const CHARS: &[char; 16] =
29            &['0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f'];
30        let mut res = String::with_capacity(2 * self.0.len() + 2);
31        if !self.0.is_empty() {
32            res.push('0');
33            res.push('x');
34        }
35        for &c in &self.0 {
36            // We don't really care about little vs big endianness, since we
37            // only need a stable format, so we pick one: little endian (print
38            // high bits first).
39            res.push(CHARS[(c >> 4) as usize]);
40            res.push(CHARS[(c & 0b1111) as usize]);
41        }
42        res
43    }
44}
45
46impl std::fmt::Debug for SessionHash {
47    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
48        f.debug_tuple("SessionHash").field(&self.to_hex()).finish()
49    }
50}
51
52/// Compute a hash uniquely identifying the OAuth 2.0 session tokens.
53fn compute_session_hash(tokens: &SessionTokens) -> SessionHash {
54    let mut hash = Sha256::new().chain_update(tokens.access_token.as_bytes());
55    if let Some(refresh_token) = &tokens.refresh_token {
56        hash = hash.chain_update(refresh_token.as_bytes());
57    }
58    SessionHash(hash.finalize().to_vec())
59}
60
61#[derive(Clone)]
62pub(super) struct CrossProcessRefreshManager {
63    store: Store,
64    store_lock: CrossProcessLock<LockableCryptoStore>,
65    known_session_hash: Arc<Mutex<Option<SessionHash>>>,
66}
67
68impl CrossProcessRefreshManager {
69    /// Create a new `CrossProcessRefreshManager`.
70    pub fn new(store: Store, lock: CrossProcessLock<LockableCryptoStore>) -> Self {
71        Self { store, store_lock: lock, known_session_hash: Arc::new(Mutex::new(None)) }
72    }
73
74    /// Wait for up to 60 seconds to get a cross-process store lock, then either
75    /// timeout (as an error) or return a lock guard.
76    ///
77    /// The guard also contains information useful to react upon another
78    /// background refresh having happened in the database already.
79    pub async fn spin_lock(
80        &self,
81    ) -> Result<CrossProcessRefreshLockGuard, CrossProcessRefreshLockError> {
82        // Acquire the intra-process mutex, to avoid multiple requests across
83        // threads in the current process.
84        trace!("Waiting for intra-process lock...");
85        let prev_hash = self.known_session_hash.clone().lock_owned().await;
86
87        // Acquire the cross-process mutex, to avoid multiple requests across
88        // different processus.
89        trace!("Waiting for inter-process lock...");
90        let store_guard = self
91            .store_lock
92            .spin_lock(Some(60000))
93            .await
94            .map_err(|err| {
95                CrossProcessRefreshLockError::LockError(CrossProcessLockError::TryLock(Arc::new(
96                    err,
97                )))
98            })?
99            .map_err(|err| CrossProcessRefreshLockError::LockError(err.into()))?;
100
101        // Read the previous session hash in the database.
102        let current_db_session_bytes = self.store.get_custom_value(OIDC_SESSION_HASH_KEY).await?;
103
104        let db_hash = current_db_session_bytes.map(SessionHash);
105
106        let hash_mismatch = match (&db_hash, &*prev_hash) {
107            (None, _) => false,
108            (Some(_), None) => true,
109            (Some(db), Some(known)) => db != known,
110        };
111
112        trace!(hash_mismatch, ?prev_hash, ?db_hash);
113
114        let guard = CrossProcessRefreshLockGuard {
115            hash_guard: prev_hash,
116            _store_guard: store_guard.into_guard(),
117            hash_mismatch,
118            db_hash,
119            store: self.store.clone(),
120        };
121
122        Ok(guard)
123    }
124
125    pub async fn restore_session(&self, tokens: &SessionTokens) {
126        let prev_tokens_hash = compute_session_hash(tokens);
127        *self.known_session_hash.lock().await = Some(prev_tokens_hash);
128    }
129
130    pub async fn on_logout(&self) -> Result<(), CrossProcessRefreshLockError> {
131        self.store
132            .remove_custom_value(OIDC_SESSION_HASH_KEY)
133            .await
134            .map_err(CrossProcessRefreshLockError::StoreError)?;
135        *self.known_session_hash.lock().await = None;
136        Ok(())
137    }
138}
139
140pub(super) struct CrossProcessRefreshLockGuard {
141    /// The hash for the latest session, either the one we knew, or the latest
142    /// one read from the database, if it was more up to date.
143    hash_guard: OwnedMutexGuard<Option<SessionHash>>,
144
145    /// Cross-process lock being hold.
146    _store_guard: CrossProcessLockGuard,
147
148    /// Reference to the underlying store, for storing the hash of the latest
149    /// known session (as a custom value).
150    store: Store,
151
152    /// Do the in-memory hash and database hash mismatch?
153    ///
154    /// If so, this indicates that another process may have refreshed the token
155    /// in the background.
156    ///
157    /// We don't consider it a mismatch if there was no previous value in the
158    /// database. We do consider it a mismatch if there was no in-memory value
159    /// known, but one was known in the database.
160    pub hash_mismatch: bool,
161
162    /// Session hash previously stored in the DB.
163    ///
164    /// Used for debugging and testing purposes.
165    db_hash: Option<SessionHash>,
166}
167
168impl CrossProcessRefreshLockGuard {
169    /// Updates the `SessionTokens` hash in-memory only.
170    fn save_in_memory(&mut self, hash: SessionHash) {
171        *self.hash_guard = Some(hash);
172    }
173
174    /// Updates the `SessionTokens` hash in the database only.
175    async fn save_in_database(
176        &self,
177        hash: &SessionHash,
178    ) -> Result<(), CrossProcessRefreshLockError> {
179        self.store.set_custom_value(OIDC_SESSION_HASH_KEY, hash.0.clone()).await?;
180        Ok(())
181    }
182
183    /// Updates the `SessionTokens` hash in both memory and database.
184    ///
185    /// Must be called after a successful refresh.
186    pub async fn save_in_memory_and_db(
187        &mut self,
188        tokens: &SessionTokens,
189    ) -> Result<(), CrossProcessRefreshLockError> {
190        let hash = compute_session_hash(tokens);
191        self.save_in_database(&hash).await?;
192        self.save_in_memory(hash);
193        Ok(())
194    }
195
196    /// Handle a mismatch by making sure values in the database and memory match
197    /// tokens we trust.
198    pub async fn handle_mismatch(
199        &mut self,
200        trusted_tokens: &SessionTokens,
201    ) -> Result<(), CrossProcessRefreshLockError> {
202        let new_hash = compute_session_hash(trusted_tokens);
203        trace!("Trusted OAuth 2.0 tokens have hash {new_hash:?}; db had {:?}", self.db_hash);
204
205        if let Some(db_hash) = &self.db_hash
206            && new_hash != *db_hash
207        {
208            // That should never happen, unless we got into an impossible
209            // situation! In this case, we assume the value returned by the
210            // callback is always correct, so override that in the database too.
211            tracing::error!("error: DB and trusted disagree. Overriding in DB.");
212            self.save_in_database(&new_hash).await?;
213        }
214
215        self.save_in_memory(new_hash);
216        Ok(())
217    }
218}
219
220/// An error that happened when interacting with the cross-process store lock
221/// during a token refresh.
222#[derive(Debug, Error)]
223pub enum CrossProcessRefreshLockError {
224    /// Underlying error caused by the store.
225    #[error(transparent)]
226    StoreError(#[from] CryptoStoreError),
227
228    /// The locking itself failed.
229    #[error(transparent)]
230    LockError(#[from] CrossProcessLockError),
231
232    /// The previous hash isn't valid.
233    #[error("the previous stored hash isn't a valid integer")]
234    InvalidPreviousHash,
235
236    /// The lock hasn't been set up.
237    #[error("the cross-process lock hasn't been set up with `enable_cross_process_refresh_lock")]
238    MissingLock,
239
240    /// Cross-process lock was set, but without session callbacks.
241    #[error(
242        "reload session callback must be set with Client::set_session_callbacks() \
243         for the cross-process lock to work"
244    )]
245    MissingReloadSession,
246
247    /// The store has been created twice.
248    #[error(
249        "the cross-process lock has been set up twice with `enable_cross_process_refresh_lock`"
250    )]
251    DuplicatedLock,
252}
253
254#[cfg(all(test, feature = "e2e-encryption", feature = "sqlite", not(target_family = "wasm")))]
255mod tests {
256
257    use anyhow::Context as _;
258    use futures_util::future::join_all;
259    use matrix_sdk_base::{SessionMeta, store::RoomLoadSettings};
260    use matrix_sdk_test::{User, async_test};
261    use ruma::owned_device_id;
262
263    use super::compute_session_hash;
264    use crate::{
265        Error,
266        authentication::oauth::cross_process::SessionHash,
267        test_utils::{
268            client::{
269                MockClientBuilder, mock_prev_session_tokens_with_refresh,
270                mock_session_tokens_with_refresh, oauth::mock_session,
271            },
272            mocks::MatrixMockServer,
273        },
274    };
275
276    #[async_test]
277    async fn test_restore_session_lock() -> Result<(), Error> {
278        // Create a client that will use sqlite databases.
279
280        let tmp_dir = tempfile::tempdir()?;
281        let client = MockClientBuilder::new(None)
282            .on_builder(|builder| builder.sqlite_store(&tmp_dir, None))
283            .unlogged()
284            .build()
285            .await;
286
287        let tokens = mock_session_tokens_with_refresh();
288
289        client.oauth().enable_cross_process_refresh_lock("test".to_owned()).await?;
290
291        client.set_session_callbacks(
292            Box::new({
293                // This is only called because of extra checks in the code.
294                let tokens = tokens.clone();
295                move |_| Ok(tokens.clone())
296            }),
297            Box::new(|_| panic!("save_session_callback shouldn't be called here")),
298        )?;
299
300        let session_hash = compute_session_hash(&tokens);
301        client
302            .oauth()
303            .restore_session(mock_session(tokens.clone()), RoomLoadSettings::default())
304            .await?;
305
306        assert_eq!(client.session_tokens().unwrap(), tokens);
307
308        let oauth = client.oauth();
309        let xp_manager = oauth.ctx().cross_process_token_refresh_manager.get().unwrap();
310
311        {
312            let known_session = xp_manager.known_session_hash.lock().await;
313            assert_eq!(known_session.as_ref().unwrap(), &session_hash);
314        }
315
316        {
317            let lock = xp_manager.spin_lock().await.unwrap();
318            assert!(!lock.hash_mismatch);
319            assert_eq!(lock.db_hash.unwrap(), session_hash);
320        }
321
322        Ok(())
323    }
324
325    #[async_test]
326    async fn test_finish_login() -> anyhow::Result<()> {
327        let server = MatrixMockServer::new().await;
328        server.mock_who_am_i().ok().expect(1).named("whoami").mount().await;
329
330        let tmp_dir = tempfile::tempdir()?;
331        let client = server
332            .client_builder()
333            .on_builder(|builder| builder.sqlite_store(&tmp_dir, None))
334            .registered_with_oauth()
335            .build()
336            .await;
337        let oauth = client.oauth();
338
339        // Enable cross-process lock.
340        oauth.enable_cross_process_refresh_lock("lock".to_owned()).await?;
341
342        // Simulate we've done finalize_authorization / restore_session before.
343        let session_tokens = mock_session_tokens_with_refresh();
344        client.auth_ctx().set_session_tokens(session_tokens.clone());
345
346        // Now, finishing logging will get the user ID.
347        oauth.load_session(owned_device_id!("D3V1C31D")).await?;
348
349        let session_meta = client.session_meta().context("should have session meta now")?;
350        assert_eq!(
351            *session_meta,
352            SessionMeta { user_id: User::Alice.into(), device_id: owned_device_id!("D3V1C31D") }
353        );
354
355        {
356            // The cross process lock has been correctly updated, and the next
357            // attempt to take it won't result in a mismatch.
358            let xp_manager =
359                oauth.ctx().cross_process_token_refresh_manager.get().context("must have lock")?;
360            let guard = xp_manager.spin_lock().await?;
361            let actual_hash = compute_session_hash(&session_tokens);
362            assert_eq!(guard.db_hash.as_ref(), Some(&actual_hash));
363            assert_eq!(guard.hash_guard.as_ref(), Some(&actual_hash));
364            assert!(!guard.hash_mismatch);
365        }
366
367        Ok(())
368    }
369
370    #[async_test]
371    async fn test_refresh_access_token_twice() -> anyhow::Result<()> {
372        // This tests that refresh token works, and that it doesn't cause
373        // multiple token refreshes whenever one spawns two refreshes around the
374        // same time.
375
376        let server = MatrixMockServer::new().await;
377
378        let oauth_server = server.oauth();
379        oauth_server.mock_server_metadata().ok().expect(1..).named("server_metadata").mount().await;
380        oauth_server.mock_token().ok().expect(1).named("token").mount().await;
381
382        let tmp_dir = tempfile::tempdir()?;
383        let client = server
384            .client_builder()
385            .on_builder(|builder| builder.sqlite_store(&tmp_dir, None))
386            .unlogged()
387            .build()
388            .await;
389        let oauth = client.oauth();
390
391        let next_tokens = mock_session_tokens_with_refresh();
392
393        // Enable cross-process lock.
394        oauth.enable_cross_process_refresh_lock("lock".to_owned()).await?;
395
396        // Restore the session.
397        oauth
398            .restore_session(
399                mock_session(mock_prev_session_tokens_with_refresh()),
400                RoomLoadSettings::default(),
401            )
402            .await?;
403
404        // Immediately try to refresh the access token twice in parallel.
405        for result in join_all([oauth.refresh_access_token(), oauth.refresh_access_token()]).await {
406            result?;
407        }
408
409        {
410            // The cross process lock has been correctly updated, and the next
411            // attempt to take it won't result in a mismatch.
412            let xp_manager =
413                oauth.ctx().cross_process_token_refresh_manager.get().context("must have lock")?;
414            let guard = xp_manager.spin_lock().await?;
415            let actual_hash = compute_session_hash(&next_tokens);
416            assert_eq!(guard.db_hash.as_ref(), Some(&actual_hash));
417            assert_eq!(guard.hash_guard.as_ref(), Some(&actual_hash));
418            assert!(!guard.hash_mismatch);
419        }
420
421        Ok(())
422    }
423
424    #[async_test]
425    async fn test_cross_process_concurrent_refresh() -> anyhow::Result<()> {
426        let server = MatrixMockServer::new().await;
427
428        let oauth_server = server.oauth();
429        oauth_server.mock_server_metadata().ok().expect(1..).named("server_metadata").mount().await;
430        oauth_server.mock_token().ok().expect(1).named("token").mount().await;
431
432        let prev_tokens = mock_prev_session_tokens_with_refresh();
433        let next_tokens = mock_session_tokens_with_refresh();
434
435        // Create the first client.
436        let tmp_dir = tempfile::tempdir()?;
437        let client = server
438            .client_builder()
439            .on_builder(|builder| builder.sqlite_store(&tmp_dir, None))
440            .unlogged()
441            .build()
442            .await;
443
444        let oauth = client.oauth();
445        oauth.enable_cross_process_refresh_lock("client1".to_owned()).await?;
446
447        oauth
448            .restore_session(mock_session(prev_tokens.clone()), RoomLoadSettings::default())
449            .await?;
450
451        // Create a second client, without restoring it, to test that a token
452        // update before restoration doesn't cause new issues.
453        let unrestored_client = server
454            .client_builder()
455            .on_builder(|builder| builder.sqlite_store(&tmp_dir, None))
456            .unlogged()
457            .build()
458            .await;
459        let unrestored_oauth = unrestored_client.oauth();
460        unrestored_oauth.enable_cross_process_refresh_lock("unrestored_client".to_owned()).await?;
461
462        {
463            // Create a third client that will run a refresh while the others
464            // two are doing nothing.
465            let client3 = server
466                .client_builder()
467                .on_builder(|builder| builder.sqlite_store(&tmp_dir, None))
468                .unlogged()
469                .build()
470                .await;
471
472            let oauth3 = client3.oauth();
473            oauth3.enable_cross_process_refresh_lock("client3".to_owned()).await?;
474            oauth3
475                .restore_session(mock_session(prev_tokens.clone()), RoomLoadSettings::default())
476                .await?;
477
478            // Run a refresh in the second client; this will invalidate the
479            // tokens from the first token.
480            oauth3.refresh_access_token().await?;
481
482            assert_eq!(client3.session_tokens(), Some(next_tokens.clone()));
483
484            // Reading from the cross-process lock for the second client only
485            // shows the new tokens.
486            let xp_manager =
487                oauth3.ctx().cross_process_token_refresh_manager.get().context("must have lock")?;
488            let guard = xp_manager.spin_lock().await?;
489            let actual_hash = compute_session_hash(&next_tokens);
490            assert_eq!(guard.db_hash.as_ref(), Some(&actual_hash));
491            assert_eq!(guard.hash_guard.as_ref(), Some(&actual_hash));
492            assert!(!guard.hash_mismatch);
493        }
494
495        {
496            // Restoring the client that was not restored yet will work Just
497            // Fine.
498            let oauth = unrestored_oauth;
499
500            unrestored_client.set_session_callbacks(
501                Box::new({
502                    // This is only called because of extra checks in the code.
503                    let tokens = next_tokens.clone();
504                    move |_| Ok(tokens.clone())
505                }),
506                Box::new(|_| panic!("save_session_callback shouldn't be called here")),
507            )?;
508
509            oauth
510                .restore_session(mock_session(prev_tokens.clone()), RoomLoadSettings::default())
511                .await?;
512
513            // And this client is now aware of the latest tokens.
514            let xp_manager =
515                oauth.ctx().cross_process_token_refresh_manager.get().context("must have lock")?;
516            let guard = xp_manager.spin_lock().await?;
517            let next_hash = compute_session_hash(&next_tokens);
518            assert_eq!(guard.db_hash.as_ref(), Some(&next_hash));
519            assert_eq!(guard.hash_guard.as_ref(), Some(&next_hash));
520            assert!(!guard.hash_mismatch);
521
522            drop(oauth);
523            drop(unrestored_client);
524        }
525
526        {
527            // The cross process lock has been correctly updated, and the next
528            // attempt to take it will result in a mismatch.
529            let xp_manager =
530                oauth.ctx().cross_process_token_refresh_manager.get().context("must have lock")?;
531            let guard = xp_manager.spin_lock().await?;
532            let previous_hash = compute_session_hash(&prev_tokens);
533            let next_hash = compute_session_hash(&next_tokens);
534            assert_eq!(guard.db_hash, Some(next_hash));
535            assert_eq!(guard.hash_guard.as_ref(), Some(&previous_hash));
536            assert!(guard.hash_mismatch);
537        }
538
539        client.set_session_callbacks(
540            Box::new({
541                // This is only called because of extra checks in the code.
542                let tokens = next_tokens.clone();
543                move |_| Ok(tokens.clone())
544            }),
545            Box::new(|_| panic!("save_session_callback shouldn't be called here")),
546        )?;
547
548        oauth.refresh_access_token().await?;
549
550        {
551            // The next attempt to take the lock isn't a mismatch.
552            let xp_manager =
553                oauth.ctx().cross_process_token_refresh_manager.get().context("must have lock")?;
554            let guard = xp_manager.spin_lock().await?;
555            let actual_hash = compute_session_hash(&next_tokens);
556            assert_eq!(guard.db_hash.as_ref(), Some(&actual_hash));
557            assert_eq!(guard.hash_guard.as_ref(), Some(&actual_hash));
558            assert!(!guard.hash_mismatch);
559        }
560
561        Ok(())
562    }
563
564    #[async_test]
565    async fn test_logout() -> anyhow::Result<()> {
566        let server = MatrixMockServer::new().await;
567
568        let oauth_server = server.oauth();
569        oauth_server
570            .mock_server_metadata()
571            .ok_https()
572            .expect(1..)
573            .named("server_metadata")
574            .mount()
575            .await;
576        oauth_server.mock_revocation().ok().expect(1).named("revocation").mount().await;
577
578        let tmp_dir = tempfile::tempdir()?;
579        let client = server
580            .client_builder()
581            .on_builder(|builder| builder.sqlite_store(&tmp_dir, None))
582            .unlogged()
583            .build()
584            .await;
585        let oauth = client.oauth().insecure_rewrite_https_to_http();
586
587        // Enable cross-process lock.
588        oauth.enable_cross_process_refresh_lock("lock".to_owned()).await?;
589
590        // Restore the session.
591        let tokens = mock_session_tokens_with_refresh();
592        oauth.restore_session(mock_session(tokens.clone()), RoomLoadSettings::default()).await?;
593
594        oauth.logout().await.unwrap();
595
596        {
597            // The cross process lock has been correctly updated, and all the
598            // hashes are empty after a logout.
599            let xp_manager =
600                oauth.ctx().cross_process_token_refresh_manager.get().context("must have lock")?;
601            let guard = xp_manager.spin_lock().await?;
602            assert!(guard.db_hash.is_none());
603            assert!(guard.hash_guard.is_none());
604            assert!(!guard.hash_mismatch);
605        }
606
607        Ok(())
608    }
609
610    #[test]
611    fn test_session_hash_to_hex() {
612        let hash = SessionHash(vec![]);
613        assert_eq!(hash.to_hex(), "");
614
615        let hash = SessionHash(vec![0x13, 0x37, 0x42, 0xde, 0xad, 0xca, 0xfe]);
616        assert_eq!(hash.to_hex(), "0x133742deadcafe");
617    }
618
619    /// The refresh token can be rotated by another process while the app is
620    /// suspended in the middle of its own refresh. The race, in order:
621    ///
622    /// 1. the app starts a refresh and takes the cross-process lock,
623    /// 2. the OS suspends it while it waits for `server_metadata()`,
624    /// 3. the 500ms lock lease lapses, as the task renewing it is frozen too,
625    /// 4. the NSE takes the lock, refreshes, and rotates the refresh token,
626    /// 5. the app resumes, still holding the token it captured in step 1.
627    ///
628    /// The app must notice the rotation, adopt the token the NSE stored, and
629    /// stay signed in.
630    #[async_test]
631    async fn test_refresh_interrupted_by_suspension_does_not_sign_out() {
632        use std::{thread, time::Duration};
633
634        let server = MatrixMockServer::new().await;
635        let oauth_server = server.oauth();
636
637        // The token endpoint behaves like a rotating MAS. Only the first
638        // exchange succeeds and rotates the token: that one is the NSE's
639        // refresh, which the app is suspended through. Any later exchange
640        // presents the token that rotation consumed, and is rejected with
641        // `invalid_grant`.
642        oauth_server
643            .mock_token()
644            .ok_with_tokens("1234", "ZYXWV") // == mock_session_tokens_with_refresh()
645            .mock_once()
646            .with_priority(1)
647            .mount()
648            .await;
649        oauth_server.mock_token().invalid_grant().with_priority(2).mount().await;
650
651        // The app's (first) metadata request is delayed, to keep its refresh
652        // parked until after the NSE has rotated the token. The NSE's own
653        // request is answered immediately.
654        oauth_server
655            .mock_server_metadata()
656            .with_delay(Duration::from_secs(1))
657            .ok()
658            .mock_once()
659            .with_priority(1)
660            .mount()
661            .await;
662        oauth_server.mock_server_metadata().ok().with_priority(2).mount().await;
663
664        // The app and the NSE are two clients over one shared sqlite store,
665        // both restored with the same (prev) session.
666        let tmp_dir = tempfile::tempdir().unwrap();
667
668        let app = server
669            .client_builder()
670            .on_builder(|b| b.sqlite_store(&tmp_dir, None))
671            .unlogged()
672            .build()
673            .await;
674        app.oauth().enable_cross_process_refresh_lock("app".to_owned()).await.unwrap();
675        app.oauth()
676            .restore_session(
677                mock_session(mock_prev_session_tokens_with_refresh()),
678                RoomLoadSettings::default(),
679            )
680            .await
681            .unwrap();
682        app.set_session_callbacks(
683            Box::new(|_| Ok(mock_session_tokens_with_refresh())),
684            Box::new(|_| Ok(())),
685        )
686        .unwrap();
687
688        let nse = server
689            .client_builder()
690            .on_builder(|b| b.sqlite_store(&tmp_dir, None))
691            .unlogged()
692            .build()
693            .await;
694        nse.oauth().enable_cross_process_refresh_lock("nse".to_owned()).await.unwrap();
695        nse.oauth()
696            .restore_session(
697                mock_session(mock_prev_session_tokens_with_refresh()),
698                RoomLoadSettings::default(),
699            )
700            .await
701            .unwrap();
702        nse.set_session_callbacks(
703            Box::new(|_| Ok(mock_session_tokens_with_refresh())),
704            Box::new(|_| Ok(())),
705        )
706        .unwrap();
707
708        // Start the app's refresh; it takes the lock and then parks in the
709        // delayed `server_metadata()` request.
710        let app_oauth = app.oauth();
711        let app_refresh = tokio::spawn(async move { app_oauth.refresh_access_token().await });
712
713        // Wait until the app has actually issued that request — by then it
714        // holds the lock and, in the buggy ordering, has already captured the
715        // refresh token.
716        let mut waited = Duration::ZERO;
717        while !server
718            .received_requests()
719            .await
720            .unwrap_or_default()
721            .iter()
722            .any(|request| request.url.path().contains("auth_metadata"))
723        {
724            assert!(waited < Duration::from_secs(5), "the app never asked for the metadata");
725            tokio::time::sleep(Duration::from_millis(10)).await;
726            waited += Duration::from_millis(10);
727        }
728
729        // "Suspend" the app: blocking the current-thread runtime freezes every
730        // task, including the one renewing the lease, so the 500ms lock lease
731        // lapses.
732        thread::sleep(Duration::from_millis(700));
733
734        // The NSE steals the lapsed lock and refreshes, rotating prev -> next
735        // and consuming the prev token at the server.
736        nse.oauth().refresh_access_token().await.unwrap();
737        assert_eq!(nse.session_tokens(), Some(mock_session_tokens_with_refresh()));
738
739        // The app resumes when its metadata delay elapses. It must notice the
740        // rotation and recover, rather than exchange the token it captured
741        // earlier.
742        let app_refresh = app_refresh.await.expect("the app refresh task shouldn't panic");
743
744        assert!(
745            app_refresh.is_ok(),
746            "the app was signed out after the NSE rotated the refresh token: {app_refresh:?}"
747        );
748    }
749}