From 0e2ee787bc51105cb027b1903b7d4e81f542c0f4 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Wed, 29 Jul 2026 10:07:03 -0400 Subject: [PATCH 1/4] fix(auth): serialize CLI token refreshes across processes --- lib/foundation/fabro-client/src/auth_store.rs | 60 ++++++++++++++ lib/foundation/fabro-client/src/client.rs | 80 +++++++++++++++++++ 2 files changed, 140 insertions(+) diff --git a/lib/foundation/fabro-client/src/auth_store.rs b/lib/foundation/fabro-client/src/auth_store.rs index b6bdf8760..4cde4d48b 100644 --- a/lib/foundation/fabro-client/src/auth_store.rs +++ b/lib/foundation/fabro-client/src/auth_store.rs @@ -18,6 +18,9 @@ use fs2::FileExt; use rand::Rng; use serde::{Deserialize, Serialize}; use thiserror::Error; +#[cfg(unix)] +use tokio::task; +use tokio::task::JoinError; use crate::target::ServerTarget; @@ -106,6 +109,8 @@ pub enum LockError { path: PathBuf, source: std::io::Error, }, + #[error("failed to wait for auth store lock at {path}: {source}")] + Task { path: PathBuf, source: JoinError }, } #[derive(Debug, Clone)] @@ -113,6 +118,11 @@ pub struct AuthStore { path: PathBuf, } +#[cfg(unix)] +pub(crate) struct RefreshLockGuard { + _file: std::fs::File, +} + #[derive(Debug, Default, Serialize, Deserialize)] struct AuthFile { #[serde(default)] @@ -193,6 +203,25 @@ impl AuthStore { }) } + #[cfg(unix)] + pub(crate) async fn acquire_refresh_lock(&self) -> Result { + 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, + })? + } + + #[cfg(not(unix))] + pub(crate) async fn acquire_refresh_lock(&self) -> Result<(), AuthStoreError> { + Err(AuthStoreError::UnsupportedPlatform) + } + fn read_auth_file(&self) -> Result { match fs::read_to_string(&self.path) { Ok(contents) => { @@ -280,6 +309,37 @@ impl AuthStore { self.path.with_extension("lock") } + #[cfg(unix)] + fn refresh_lock_path(&self) -> PathBuf { + self.path.with_extension("refresh.lock") + } + + #[cfg(unix)] + fn acquire_refresh_lock_blocking(&self) -> Result { + self.ensure_parent_dir()?; + let path = self.refresh_lock_path(); + let lock_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 FileExt::try_lock_exclusive(&lock_file) { + Ok(()) => {} + Err(source) if source.kind() == std::io::ErrorKind::WouldBlock => { + lock_file + .lock_exclusive() + .map_err(|source| classify_lock_error(path, source))?; + } + Err(source) => return Err(classify_lock_error(path, source).into()), + } + Ok(RefreshLockGuard { _file: lock_file }) + } + #[cfg(unix)] fn lock_error(&self, source: std::io::Error) -> AuthStoreError { classify_lock_error(self.lock_path(), source).into() diff --git a/lib/foundation/fabro-client/src/client.rs b/lib/foundation/fabro-client/src/client.rs index 531b9f2ba..dc3c86823 100644 --- a/lib/foundation/fabro-client/src/client.rs +++ b/lib/foundation/fabro-client/src/client.rs @@ -449,6 +449,10 @@ impl Client { return Ok(()); } + // Refresh tokens are single-use. Hold the cross-process lock from the + // fresh store read through rotation and persistence. AuthStore methods + // take the shorter auth-file lock inside this guard. + 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 +464,10 @@ impl Client { } AuthEntry::OAuth(entry) => entry, }; + if oauth_entry.access_token != failed_access_token { + 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 +2581,78 @@ 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(); + + let first = Client::builder() + .target(target.clone()) + .credential(Credential::OAuth(entry.clone())) + .oauth_session(OAuthSession::new(target.clone(), auth_store.clone())) + .connect() + .await + .unwrap(); + let second = Client::builder() + .target(target.clone()) + .credential(Credential::OAuth(entry)) + .oauth_session(OAuthSession::new(target, auth_store)) + .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") + ); + } + #[tokio::test] async fn refresh_access_token_classifies_expired_refresh_tokens() { let server = MockServer::start(); From 69976fc308d0c797aa9e9312bbc512ddb7794cdc Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sat, 1 Aug 2026 09:59:17 -0400 Subject: [PATCH 2/4] refactor(client): dedupe auth store lock acquisition The cross-process refresh lock added a third copy of the open-file, try-lock, then block-on-contention sequence. Collapse all three into one `open_locked_file` helper parameterized by `LockMode`, which removes `open_lock_file` and `lock_error`. Lock calls are now qualified as `FileExt` calls throughout, since `std::fs::File` has inherent locking methods with different return types that take precedence over trait methods. Also give `acquire_refresh_lock` one signature on all platforms by defining `RefreshLockGuard` for non-Unix targets too, instead of returning `Result<(), _>` there and `Result` on Unix. Co-Authored-By: Claude Opus 5 (1M context) --- lib/foundation/fabro-client/src/auth_store.rs | 129 ++++++++++-------- 1 file changed, 70 insertions(+), 59 deletions(-) diff --git a/lib/foundation/fabro-client/src/auth_store.rs b/lib/foundation/fabro-client/src/auth_store.rs index 4cde4d48b..76d95fcf4 100644 --- a/lib/foundation/fabro-client/src/auth_store.rs +++ b/lib/foundation/fabro-client/src/auth_store.rs @@ -118,11 +118,19 @@ pub struct AuthStore { path: PathBuf, } +/// Holds the cross-process refresh lock until dropped. #[cfg(unix)] pub(crate) struct RefreshLockGuard { _file: std::fs::File, } +#[allow( + dead_code, + reason = "Non-Unix targets never acquire the lock, so this is never constructed." +)] +#[cfg(not(unix))] +pub(crate) struct RefreshLockGuard; + #[derive(Debug, Default, Serialize, Deserialize)] struct AuthFile { #[serde(default)] @@ -218,7 +226,7 @@ impl AuthStore { } #[cfg(not(unix))] - pub(crate) async fn acquire_refresh_lock(&self) -> Result<(), AuthStoreError> { + pub(crate) async fn acquire_refresh_lock(&self) -> Result { Err(AuthStoreError::UnsupportedPlatform) } @@ -254,16 +262,7 @@ impl AuthStore { &self, f: impl FnOnce() -> Result, ) -> Result { - 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() } @@ -280,31 +279,10 @@ impl AuthStore { &self, f: impl FnOnce() -> Result, ) -> Result { - 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 { - 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") } @@ -317,32 +295,9 @@ impl AuthStore { #[cfg(unix)] fn acquire_refresh_lock_blocking(&self) -> Result { self.ensure_parent_dir()?; - let path = self.refresh_lock_path(); - let lock_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 FileExt::try_lock_exclusive(&lock_file) { - Ok(()) => {} - Err(source) if source.kind() == std::io::ErrorKind::WouldBlock => { - lock_file - .lock_exclusive() - .map_err(|source| classify_lock_error(path, source))?; - } - Err(source) => return Err(classify_lock_error(path, source).into()), - } - Ok(RefreshLockGuard { _file: lock_file }) - } - - #[cfg(unix)] - fn lock_error(&self, source: std::io::Error) -> AuthStoreError { - classify_lock_error(self.lock_path(), source).into() + Ok(RefreshLockGuard { + _file: open_locked_file(self.refresh_lock_path(), LockMode::Exclusive)?, + }) } #[cfg(unix)] @@ -420,6 +375,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 { + 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() { From 58c8df593ab54e18b85a68c3c6a6d5fed8a9dfe5 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sat, 1 Aug 2026 10:06:11 -0400 Subject: [PATCH 3/4] fix(client): only adopt a peer-rotated token while it is usable The staleness check treated "different from the token that failed" as "usable". A long-lived process could read an entry a sibling rotated an hour earlier, whose access token had since expired, install it, and return Ok. The caller retries once and does not refresh again, so that surfaced a 401. Require the stored token to be unexpired; an expired one now falls through and rotates with the refresh token just read. Also: - Give the non-Unix `acquire_refresh_lock` a no-op passthrough, matching the other lock helpers off Unix. Returning an error there broke re-installing a stored dev token, which needs no lock because it never writes. - Rename `Client::refresh_lock` to `local_refresh_lock`. Two different locks were sharing one word four lines apart. - Gate `LockError::Task` on Unix, where its only construction site is. - Give the concurrency test a no-proxy transport connector. Building clients without one goes through `connect_target_transport`, which does not disable proxy discovery, against localhost. - Assert the rotated refresh token reaches the store, which is the invariant behind single-use rotation. Co-Authored-By: Claude Opus 5 (1M context) --- lib/foundation/fabro-client/src/auth_store.rs | 13 +-- lib/foundation/fabro-client/src/client.rs | 109 ++++++++++++++++-- 2 files changed, 106 insertions(+), 16 deletions(-) diff --git a/lib/foundation/fabro-client/src/auth_store.rs b/lib/foundation/fabro-client/src/auth_store.rs index 76d95fcf4..0b7606320 100644 --- a/lib/foundation/fabro-client/src/auth_store.rs +++ b/lib/foundation/fabro-client/src/auth_store.rs @@ -19,8 +19,7 @@ use rand::Rng; use serde::{Deserialize, Serialize}; use thiserror::Error; #[cfg(unix)] -use tokio::task; -use tokio::task::JoinError; +use tokio::task::{self, JoinError}; use crate::target::ServerTarget; @@ -109,6 +108,7 @@ pub enum LockError { path: PathBuf, source: std::io::Error, }, + #[cfg(unix)] #[error("failed to wait for auth store lock at {path}: {source}")] Task { path: PathBuf, source: JoinError }, } @@ -124,10 +124,6 @@ pub(crate) struct RefreshLockGuard { _file: std::fs::File, } -#[allow( - dead_code, - reason = "Non-Unix targets never acquire the lock, so this is never constructed." -)] #[cfg(not(unix))] pub(crate) struct RefreshLockGuard; @@ -225,9 +221,12 @@ impl AuthStore { })? } + // 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 { - Err(AuthStoreError::UnsupportedPlatform) + Ok(RefreshLockGuard) } fn read_auth_file(&self) -> Result { diff --git a/lib/foundation/fabro-client/src/client.rs b/lib/foundation/fabro-client/src/client.rs index dc3c86823..24a7fd0c9 100644 --- a/lib/foundation/fabro-client/src/client.rs +++ b/lib/foundation/fabro-client/src/client.rs @@ -77,7 +77,9 @@ struct ClientState { pub struct Client { state: Arc>, oauth_session: Option, - refresh_lock: Arc>, + /// Serializes rotation between this client's own tasks. The cross-process + /// half lives in `AuthStore::acquire_refresh_lock`. + local_refresh_lock: Arc>, transport_connector: Option, request_timeout: Option, } @@ -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,15 +445,16 @@ 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. Hold the cross-process lock from the - // fresh store read through rotation and persistence. AuthStore methods - // take the shorter auth-file lock inside this guard. + // 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?; @@ -464,7 +467,13 @@ impl Client { } AuthEntry::OAuth(entry) => entry, }; - if oauth_entry.access_token != failed_access_token { + // 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(()); } @@ -2620,17 +2629,29 @@ mod tests { .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, auth_store)) + .oauth_session(OAuthSession::new(target.clone(), auth_store.clone())) + .transport_connector(no_proxy_connector()) .connect() .await .unwrap(); @@ -2651,6 +2672,76 @@ mod tests { 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] From 5f72bfe3ac7ebdedea9d9a352fee64ef649d3bc6 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sat, 1 Aug 2026 10:17:00 -0400 Subject: [PATCH 4/4] fix(client): name the refresh lock in its wait error LockError::Task is only constructed while waiting for the refresh sidecar lock, so "auth store lock" pointed at the wrong file. Co-Authored-By: Claude Opus 5 (1M context) --- lib/foundation/fabro-client/src/auth_store.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/foundation/fabro-client/src/auth_store.rs b/lib/foundation/fabro-client/src/auth_store.rs index 0b7606320..b2cd0ebb6 100644 --- a/lib/foundation/fabro-client/src/auth_store.rs +++ b/lib/foundation/fabro-client/src/auth_store.rs @@ -109,7 +109,7 @@ pub enum LockError { source: std::io::Error, }, #[cfg(unix)] - #[error("failed to wait for auth store lock at {path}: {source}")] + #[error("failed to wait for auth store refresh lock at {path}: {source}")] Task { path: PathBuf, source: JoinError }, }