diff --git a/lib/foundation/fabro-client/src/auth_store.rs b/lib/foundation/fabro-client/src/auth_store.rs index b6bdf8760..b2cd0ebb6 100644 --- a/lib/foundation/fabro-client/src/auth_store.rs +++ b/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 { + 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 { + Ok(RefreshLockGuard) + } + fn read_auth_file(&self) -> Result { match fs::read_to_string(&self.path) { Ok(contents) => { @@ -225,16 +261,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() } @@ -251,38 +278,25 @@ 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") } #[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 { + 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 { + 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() { diff --git a/lib/foundation/fabro-client/src/client.rs b/lib/foundation/fabro-client/src/client.rs index 531b9f2ba..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,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();