Merge pull request #675 from fabro-sh/fix/cross-process-refresh-lock

Fix cross-process CLI token refresh races

lib/foundation/fabro-client/src/auth_store.rs

@@ -18,6 +18,8 @@ use fs2::FileExt;
 use rand::Rng;
 use serde::{Deserialize, Serialize};
 use thiserror::Error;
+#[cfg(unix)]
+use tokio::task::{self, JoinError};
 
 use crate::target::ServerTarget;
 
@@ -106,6 +108,9 @@ pub enum LockError {
         path:   PathBuf,
         source: std::io::Error,
     },
+    #[cfg(unix)]
+    #[error("failed to wait for auth store refresh lock at {path}: {source}")]
+    Task { path: PathBuf, source: JoinError },
 }
 
 #[derive(Debug, Clone)]
@@ -113,6 +118,15 @@ pub struct AuthStore {
     path: PathBuf,
 }
 
+/// Holds the cross-process refresh lock until dropped.
+#[cfg(unix)]
+pub(crate) struct RefreshLockGuard {
+    _file: std::fs::File,
+}
+
+#[cfg(not(unix))]
+pub(crate) struct RefreshLockGuard;
+
 #[derive(Debug, Default, Serialize, Deserialize)]
 struct AuthFile {
     #[serde(default)]
@@ -193,6 +207,28 @@ impl AuthStore {
         })
     }
 
+    #[cfg(unix)]
+    pub(crate) async fn acquire_refresh_lock(&self) -> Result<RefreshLockGuard, AuthStoreError> {
+        let store = self.clone();
+        let lock_path = self.refresh_lock_path();
+        // Another CLI can hold this lock through a network request, so keep
+        // the blocking wait off the Tokio worker threads.
+        task::spawn_blocking(move || store.acquire_refresh_lock_blocking())
+            .await
+            .map_err(|source| LockError::Task {
+                path: lock_path,
+                source,
+            })?
+    }
+
+    // Matches the other lock helpers, which are no-op passthroughs off Unix.
+    // Failing here instead would break re-installing a stored dev token, which
+    // needs no lock because it never writes.
+    #[cfg(not(unix))]
+    pub(crate) async fn acquire_refresh_lock(&self) -> Result<RefreshLockGuard, AuthStoreError> {
+        Ok(RefreshLockGuard)
+    }
+
     fn read_auth_file(&self) -> Result<AuthFile, AuthStoreError> {
         match fs::read_to_string(&self.path) {
             Ok(contents) => {
@@ -225,16 +261,7 @@ impl AuthStore {
         &self,
         f: impl FnOnce() -> Result<T, AuthStoreError>,
     ) -> Result<T, AuthStoreError> {
-        let lock_file = self.open_lock_file()?;
-        match FileExt::try_lock_shared(&lock_file) {
-            Ok(()) => {}
-            Err(source) if source.kind() == std::io::ErrorKind::WouldBlock => {
-                lock_file
-                    .lock_shared()
-                    .map_err(|source| self.lock_error(source))?;
-            }
-            Err(source) => return Err(self.lock_error(source)),
-        }
+        let _lock = open_locked_file(self.lock_path(), LockMode::Shared)?;
         f()
     }
 
@@ -251,38 +278,25 @@ impl AuthStore {
         &self,
         f: impl FnOnce() -> Result<T, AuthStoreError>,
     ) -> Result<T, AuthStoreError> {
-        let lock_file = self.open_lock_file()?;
-        match FileExt::try_lock_exclusive(&lock_file) {
-            Ok(()) => {}
-            Err(source) if source.kind() == std::io::ErrorKind::WouldBlock => {
-                lock_file
-                    .lock_exclusive()
-                    .map_err(|source| self.lock_error(source))?;
-            }
-            Err(source) => return Err(self.lock_error(source)),
-        }
+        let _lock = open_locked_file(self.lock_path(), LockMode::Exclusive)?;
         f()
     }
 
-    #[cfg(unix)]
-    fn open_lock_file(&self) -> Result<std::fs::File, AuthStoreError> {
-        let path = self.lock_path();
-        std::fs::OpenOptions::new()
-            .create(true)
-            .read(true)
-            .write(true)
-            .truncate(false)
-            .open(&path)
-            .map_err(|source| LockError::Io { path, source }.into())
-    }
-
     fn lock_path(&self) -> PathBuf {
         self.path.with_extension("lock")
     }
 
     #[cfg(unix)]
-    fn lock_error(&self, source: std::io::Error) -> AuthStoreError {
-        classify_lock_error(self.lock_path(), source).into()
+    fn refresh_lock_path(&self) -> PathBuf {
+        self.path.with_extension("refresh.lock")
+    }
+
+    #[cfg(unix)]
+    fn acquire_refresh_lock_blocking(&self) -> Result<RefreshLockGuard, AuthStoreError> {
+        self.ensure_parent_dir()?;
+        Ok(RefreshLockGuard {
+            _file: open_locked_file(self.refresh_lock_path(), LockMode::Exclusive)?,
+        })
     }
 
     #[cfg(unix)]
@@ -360,6 +374,62 @@ fn write_private_file(path: &Path, contents: &str) -> Result<(), AuthStoreError>
     Ok(())
 }
 
+#[cfg(unix)]
+#[derive(Clone, Copy)]
+enum LockMode {
+    Shared,
+    Exclusive,
+}
+
+#[cfg(unix)]
+impl LockMode {
+    // Qualify these as `FileExt` calls. `std::fs::File` has inherent locking
+    // methods with different return types, and inherent methods take
+    // precedence over trait methods.
+    fn try_lock(self, file: &std::fs::File) -> std::io::Result<()> {
+        match self {
+            Self::Shared => FileExt::try_lock_shared(file),
+            Self::Exclusive => FileExt::try_lock_exclusive(file),
+        }
+    }
+
+    fn lock(self, file: &std::fs::File) -> std::io::Result<()> {
+        match self {
+            Self::Shared => FileExt::lock_shared(file),
+            Self::Exclusive => FileExt::lock_exclusive(file),
+        }
+    }
+}
+
+/// Opens `path`, creating it if absent, and takes an advisory lock on the
+/// returned handle. Dropping the handle releases the lock.
+///
+/// The non-blocking attempt comes first so that a filesystem which cannot lock
+/// at all reports `EOPNOTSUPP`/`ENOLCK` right away. Only plain contention
+/// reports `WouldBlock`, and that is the one case worth waiting on.
+#[cfg(unix)]
+fn open_locked_file(path: PathBuf, mode: LockMode) -> Result<std::fs::File, AuthStoreError> {
+    let file = std::fs::OpenOptions::new()
+        .create(true)
+        .read(true)
+        .write(true)
+        .truncate(false)
+        .open(&path)
+        .map_err(|source| LockError::Io {
+            path: path.clone(),
+            source,
+        })?;
+    match mode.try_lock(&file) {
+        Ok(()) => {}
+        Err(source) if source.kind() == std::io::ErrorKind::WouldBlock => {
+            mode.lock(&file)
+                .map_err(|source| classify_lock_error(path, source))?;
+        }
+        Err(source) => return Err(classify_lock_error(path, source).into()),
+    }
+    Ok(file)
+}
+
 #[cfg(unix)]
 fn classify_lock_error(path: PathBuf, source: std::io::Error) -> LockError {
     match source.raw_os_error() {

lib/foundation/fabro-client/src/client.rs

@@ -77,7 +77,9 @@ struct ClientState {
 pub struct Client {
     state:               Arc<RwLock<ClientState>>,
     oauth_session:       Option<OAuthSession>,
-    refresh_lock:        Arc<Mutex<()>>,
+    /// Serializes rotation between this client's own tasks. The cross-process
+    /// half lives in `AuthStore::acquire_refresh_lock`.
+    local_refresh_lock:  Arc<Mutex<()>>,
     transport_connector: Option<TransportConnector>,
     request_timeout:     Option<std::time::Duration>,
 }
@@ -304,7 +306,7 @@ impl ClientBuilder {
         Ok(Client {
             state: Arc::new(RwLock::new(state)),
             oauth_session: self.oauth_session,
-            refresh_lock: Arc::new(Mutex::new(())),
+            local_refresh_lock: Arc::new(Mutex::new(())),
             transport_connector,
             request_timeout,
         })
@@ -329,7 +331,7 @@ impl Client {
                 None,
             ))),
             oauth_session:       None,
-            refresh_lock:        Arc::new(Mutex::new(())),
+            local_refresh_lock:  Arc::new(Mutex::new(())),
             transport_connector: None,
             request_timeout:     None,
         }
@@ -443,12 +445,17 @@ impl Client {
             return Err(session_expired());
         };
 
-        let _guard = self.refresh_lock.lock().await;
+        let _guard = self.local_refresh_lock.lock().await;
         let current_state = self.current_state();
         if current_state.bearer_token.as_deref() != Some(failed_access_token) {
             return Ok(());
         }
 
+        // Refresh tokens are single-use, so rotation has to be serialized
+        // across processes too, not just across this client's tasks. The
+        // AuthStore calls below take the shorter auth-file lock inside this
+        // guard; nothing takes the two in the other order.
+        let _refresh_guard = oauth_session.auth_store.acquire_refresh_lock().await?;
         let Some(entry) = oauth_session.auth_store.get(&oauth_session.target)? else {
             self.rebuild_with_fallback(oauth_session).await?;
             return Err(session_expired());
@@ -460,6 +467,16 @@ impl Client {
             }
             AuthEntry::OAuth(entry) => entry,
         };
+        // Adopt a token another process already rotated, but only while it is
+        // still usable. The caller retries once and does not refresh again, so
+        // installing an expired token here would surface a 401. An expired one
+        // falls through and rotates with the refresh token just read.
+        if oauth_entry.access_token != failed_access_token
+            && oauth_entry.access_token_expires_at > chrono::Utc::now()
+        {
+            self.rebuild_client(Some(oauth_entry.access_token)).await?;
+            return Ok(());
+        }
         if oauth_entry.refresh_token_expires_at <= chrono::Utc::now() {
             oauth_session.auth_store.remove(&oauth_session.target)?;
             self.rebuild_with_fallback(oauth_session).await?;
@@ -2573,6 +2590,160 @@ mod tests {
         ]);
     }
 
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn concurrent_clients_refresh_a_rotating_token_once() {
+        let server = MockServer::start_async().await;
+        let refresh_mock = server
+            .mock_async(|when, then| {
+                when.method(POST)
+                    .path("/auth/cli/refresh")
+                    .header("authorization", "Bearer refresh-octocat");
+                then.status(200)
+                    .delay(Duration::from_millis(100))
+                    .header("Content-Type", "application/json")
+                    .json_body(json!({
+                        "access_token": "access-refreshed",
+                        "access_token_expires_at": (chrono::Utc::now()
+                            + ChronoDuration::minutes(10))
+                            .to_rfc3339(),
+                        "refresh_token": "refresh-refreshed",
+                        "refresh_token_expires_at": (chrono::Utc::now()
+                            + ChronoDuration::days(30))
+                            .to_rfc3339(),
+                        "subject": {
+                            "idp_issuer": "https://github.com",
+                            "idp_subject": "12345",
+                            "login": "octocat",
+                            "name": "Name octocat",
+                            "email": "octocat@example.com"
+                        }
+                    }));
+            })
+            .await;
+        let temp = tempfile::tempdir().unwrap();
+        let auth_store = AuthStore::new(temp.path().join("auth.json"));
+        let target = ServerTarget::http_url(server.base_url()).unwrap();
+        let entry = oauth_entry("octocat");
+        auth_store
+            .put(&target, AuthEntry::OAuth(entry.clone()))
+            .unwrap();
+
+        // Two separately built clients hold separate in-process mutexes, so the
+        // only thing serializing them is the lock file. That works in one
+        // process because flock conflicts across distinct descriptors.
+        let no_proxy_connector = || {
+            let base_url = server.base_url();
+            TransportConnector::new(move |_bearer_token| {
+                let base_url = base_url.clone();
+                async move { Ok((fabro_http::test_http_client().unwrap(), base_url)) }
+            })
+        };
+        let first = Client::builder()
+            .target(target.clone())
+            .credential(Credential::OAuth(entry.clone()))
+            .oauth_session(OAuthSession::new(target.clone(), auth_store.clone()))
+            .transport_connector(no_proxy_connector())
+            .connect()
+            .await
+            .unwrap();
+        let second = Client::builder()
+            .target(target.clone())
+            .credential(Credential::OAuth(entry))
+            .oauth_session(OAuthSession::new(target.clone(), auth_store.clone()))
+            .transport_connector(no_proxy_connector())
+            .connect()
+            .await
+            .unwrap();
+
+        let (first_result, second_result) = tokio::join!(
+            first.refresh_access_token("access-octocat"),
+            second.refresh_access_token("access-octocat"),
+        );
+
+        first_result.unwrap();
+        second_result.unwrap();
+        refresh_mock.assert_calls_async(1).await;
+        assert_eq!(
+            first.current_state().bearer_token.as_deref(),
+            Some("access-refreshed")
+        );
+        assert_eq!(
+            second.current_state().bearer_token.as_deref(),
+            Some("access-refreshed")
+        );
+        // The rotated refresh token must be what landed in the store, or the
+        // next rotation would replay a spent one.
+        let stored = match auth_store.get(&target).unwrap().unwrap() {
+            AuthEntry::OAuth(stored) => stored,
+            AuthEntry::DevToken(_) => panic!("expected an OAuth entry"),
+        };
+        assert_eq!(stored.access_token, "access-refreshed");
+        assert_eq!(stored.refresh_token, "refresh-refreshed");
+    }
+
+    #[tokio::test]
+    async fn refresh_access_token_rotates_when_the_stored_token_is_also_expired() {
+        let server = MockServer::start_async().await;
+        let refresh_mock = server
+            .mock_async(|when, then| {
+                when.method(POST)
+                    .path("/auth/cli/refresh")
+                    .header("authorization", "Bearer refresh-octocat");
+                then.status(200)
+                    .header("Content-Type", "application/json")
+                    .json_body(json!({
+                        "access_token": "access-refreshed",
+                        "access_token_expires_at": (chrono::Utc::now()
+                            + ChronoDuration::minutes(10))
+                            .to_rfc3339(),
+                        "refresh_token": "refresh-refreshed",
+                        "refresh_token_expires_at": (chrono::Utc::now()
+                            + ChronoDuration::days(30))
+                            .to_rfc3339(),
+                        "subject": {
+                            "idp_issuer": "https://github.com",
+                            "idp_subject": "12345",
+                            "login": "octocat",
+                            "name": "Name octocat",
+                            "email": "octocat@example.com"
+                        }
+                    }));
+            })
+            .await;
+        let temp = tempfile::tempdir().unwrap();
+        let auth_store = AuthStore::new(temp.path().join("auth.json"));
+        let target = ServerTarget::http_url(server.base_url()).unwrap();
+
+        // A sibling process rotated the store a while ago, and that token has
+        // since expired too. Adopting it would 401 on the caller's single retry.
+        let mut stored = oauth_entry("octocat");
+        stored.access_token = "access-stale".to_string();
+        stored.access_token_expires_at = chrono::Utc::now() - ChronoDuration::minutes(1);
+        auth_store.put(&target, AuthEntry::OAuth(stored)).unwrap();
+
+        let base_url = server.base_url();
+        let client = Client::builder()
+            .target(target.clone())
+            .credential(Credential::OAuth(oauth_entry("octocat")))
+            .oauth_session(OAuthSession::new(target, auth_store))
+            .transport_connector(TransportConnector::new(move |_bearer_token| {
+                let base_url = base_url.clone();
+                async move { Ok((fabro_http::test_http_client().unwrap(), base_url)) }
+            }))
+            .connect()
+            .await
+            .unwrap();
+
+        client.refresh_access_token("access-octocat").await.unwrap();
+
+        refresh_mock.assert_calls_async(1).await;
+        assert_eq!(
+            client.current_state().bearer_token.as_deref(),
+            Some("access-refreshed")
+        );
+    }
+
     #[tokio::test]
     async fn refresh_access_token_classifies_expired_refresh_tokens() {
         let server = MockServer::start();