From 33268479d304f35f9ed48fa032eebb08ac5975bc Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sun, 26 Jul 2026 00:05:31 -0400 Subject: [PATCH] Delete the SlateDB refresh token store Removes `slate/auth_tokens.rs`, `Database::refresh_tokens()`, and its `OnceCell` now that nothing reads them, and clears the retired `auth/refresh` prefix once at startup. That sweep is not housekeeping we could skip: the reaper that used to collect those records went with the store, so without it they would sit in the object store forever. A later boot finds the prefix empty and does nothing. `record/transaction.rs` goes too -- rotation was its only caller, and SQLite transactions replaced it. `KeyedMutex` stays; `AuthCodeStore` still uses it until auth codes move. Existing refresh tokens are not migrated. Everyone re-authenticates once on upgrade. Co-Authored-By: Claude Opus 5 (1M context) --- lib/apps/fabro-server/src/serve.rs | 8 + lib/components/fabro-store/src/lib.rs | 10 +- lib/components/fabro-store/src/record/mod.rs | 8 +- .../fabro-store/src/record/transaction.rs | 208 ------ .../fabro-store/src/slate/auth_tokens.rs | 598 ------------------ lib/components/fabro-store/src/slate/mod.rs | 70 +- 6 files changed, 73 insertions(+), 829 deletions(-) delete mode 100644 lib/components/fabro-store/src/record/transaction.rs delete mode 100644 lib/components/fabro-store/src/slate/auth_tokens.rs diff --git a/lib/apps/fabro-server/src/serve.rs b/lib/apps/fabro-server/src/serve.rs index 4abe6dda1..eef61311b 100644 --- a/lib/apps/fabro-server/src/serve.rs +++ b/lib/apps/fabro-server/src/serve.rs @@ -780,6 +780,14 @@ where cache_path, )); let auth_code_store = store.auth_codes().await?; + // Refresh tokens now live in SQLite. Nothing reads the old records and no + // reaper collects them any more, so clear them out once rather than + // leaving them in the object store forever. + match store.retire_refresh_token_keyspace().await { + Ok(0) => {} + Ok(removed) => info!(removed, "Removed retired SlateDB refresh token records"), + Err(err) => warn!(error = %err, "Failed to remove retired SlateDB refresh token records"), + } let (artifact_object_store, artifact_prefix) = build_artifact_object_store_with_server_secrets( &resolved_server_settings, &server_secrets, diff --git a/lib/components/fabro-store/src/lib.rs b/lib/components/fabro-store/src/lib.rs index 1da19397b..3c57d53e2 100644 --- a/lib/components/fabro-store/src/lib.rs +++ b/lib/components/fabro-store/src/lib.rs @@ -19,9 +19,9 @@ pub use artifact_store::{ ArtifactKey, ArtifactStore, NodeArtifact, StageArtifactEntry, retry_storage_segment, stage_storage_segment, }; -// `RefreshToken` and the rotation outcome stay module-qualified while the -// SlateDB-backed store below still exports types under those names. -pub use auth_session_store::{ActiveCliSession, AuthSessionRecord, AuthSessionStore}; +pub use auth_session_store::{ + ActiveCliSession, AuthSessionRecord, AuthSessionStore, RefreshToken, RotateOutcome, +}; pub use error::{Error, Result}; pub use fabro_types::{ EventEnvelope, PendingInterviewRecord, Run, RunBlobId, RunProjection, StageId, StageProjection, @@ -38,8 +38,8 @@ pub use run_summary_store::{ }; pub use serializable_projection::SerializableProjection; pub use slate::{ - AuthCode, AuthCodeStore, Blob, BlobStore, CachedRunProjection, ConsumeOutcome, Database, - RefreshToken, RefreshTokenStore, RunCatalogIndex, RunDatabase, Runs, UnreadableRun, + AuthCode, AuthCodeStore, Blob, BlobStore, CachedRunProjection, Database, RunCatalogIndex, + RunDatabase, Runs, UnreadableRun, }; pub use types::EventPayload; diff --git a/lib/components/fabro-store/src/record/mod.rs b/lib/components/fabro-store/src/record/mod.rs index 64c956120..189863171 100644 --- a/lib/components/fabro-store/src/record/mod.rs +++ b/lib/components/fabro-store/src/record/mod.rs @@ -5,22 +5,18 @@ //! type. //! - [`RecordId`]: converts the typed id to and from key segments. //! - [`Repository`]: performs the generic get/put/delete/scan/gc operations. -//! - [`transaction`]: batches multiple typed writes into one atomic SlateDB -//! write. //! //! Production callers should add a named domain store on top of this layer //! rather than exposing `Repository` directly. See `slate/auth_codes.rs`, -//! `slate/auth_tokens.rs`, `slate/blob_store.rs`, and -//! `slate/run_catalog_index.rs` for the intended pattern. +//! `slate/blob_store.rs`, and `slate/run_catalog_index.rs` for the intended +//! pattern. mod codec; mod record_id; mod repository; -mod transaction; pub(crate) use codec::{Codec, JsonCodec, MarkerCodec, RawBytesCodec}; pub(crate) use repository::Repository; -pub(crate) use transaction::transaction; use crate::Result; diff --git a/lib/components/fabro-store/src/record/transaction.rs b/lib/components/fabro-store/src/record/transaction.rs deleted file mode 100644 index 8c7c710ac..000000000 --- a/lib/components/fabro-store/src/record/transaction.rs +++ /dev/null @@ -1,208 +0,0 @@ -use slatedb::{Db, WriteBatch}; - -use super::repository::key_for_id; -use super::{Codec, Record}; -use crate::Result; - -pub(crate) async fn transaction(db: &Db, f: F) -> Result -where - F: FnOnce(&mut Tx) -> Result, -{ - let mut tx = Tx::new(); - let value = f(&mut tx)?; - if tx.has_ops { - db.write(tx.batch).await?; - } - Ok(value) -} - -pub(crate) struct Tx { - batch: WriteBatch, - /// SlateDB rejects empty `WriteBatch` commits; skip the write entirely - /// when the closure produced no operations. - has_ops: bool, -} - -impl Tx { - fn new() -> Self { - Self { - batch: WriteBatch::new(), - has_ops: false, - } - } - - pub(crate) fn put(&mut self, record: &R) -> Result<&mut Self> { - let id = record.id(); - self.put_at(&id, record) - } - - pub(crate) fn put_at(&mut self, id: &R::Id, record: &R) -> Result<&mut Self> { - self.batch - .put(key_for_id::(id)?, R::Codec::encode(record)?); - self.has_ops = true; - Ok(self) - } - - #[allow( - dead_code, - reason = "Shared transaction surface; current production callers only use put paths" - )] - pub(crate) fn delete(&mut self, id: &R::Id) -> Result<&mut Self> { - self.batch.delete(key_for_id::(id)?); - self.has_ops = true; - Ok(self) - } -} - -#[cfg(test)] -mod tests { - use std::sync::Arc; - - use object_store::memory::InMemory; - use serde::{Deserialize, Serialize}; - - use super::{Record, Tx, transaction}; - use crate::record::{Codec, JsonCodec, Repository}; - use crate::{Error, Result}; - - #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] - struct TxRecord { - id: String, - payload: String, - poisoned: bool, - } - - impl Record for TxRecord { - type Id = String; - type Codec = JsonCodec; - - const PREFIX: &'static str = "test/transaction"; - - fn id(&self) -> Self::Id { - self.id.clone() - } - } - - #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] - struct FailingRecord { - id: String, - poisoned: bool, - } - - struct FailingCodec; - - impl Codec for FailingCodec { - fn encode(value: &FailingRecord) -> Result> { - if value.poisoned { - return Err(Error::Other( - "poisoned record refused to encode".to_string(), - )); - } - serde_json::to_vec(value).map_err(Into::into) - } - - fn decode(bytes: &[u8]) -> Result { - serde_json::from_slice(bytes).map_err(Into::into) - } - } - - impl Record for FailingRecord { - type Id = String; - type Codec = FailingCodec; - - const PREFIX: &'static str = "test/failing-transaction"; - - fn id(&self) -> Self::Id { - self.id.clone() - } - } - - async fn db() -> Arc { - Arc::new( - slatedb::Db::open("transaction-tests", Arc::new(InMemory::new())) - .await - .unwrap(), - ) - } - - #[tokio::test] - async fn closure_error_short_circuits_without_writing() { - let db = db().await; - let repo = Repository::::new(Arc::clone(&db)); - let record = TxRecord { - id: "record-1".to_string(), - payload: "hello".to_string(), - poisoned: false, - }; - - let error = transaction::<(), _>(&db, |tx| { - tx.put(&record)?; - Err(Error::Other("stop before commit".to_string())) - }) - .await - .unwrap_err(); - - assert_eq!(error.to_string(), "stop before commit"); - assert!(repo.get(&record.id()).await.unwrap().is_none()); - } - - #[tokio::test] - async fn encode_failure_discards_the_entire_batch() { - let db = db().await; - let repo = Repository::::new(Arc::clone(&db)); - let good = FailingRecord { - id: "good".to_string(), - poisoned: false, - }; - let bad = FailingRecord { - id: "bad".to_string(), - poisoned: true, - }; - - let error = transaction::<(), _>(&db, |tx| { - tx.put(&good)?; - tx.put(&bad)?; - Ok(()) - }) - .await - .unwrap_err(); - - assert_eq!(error.to_string(), "poisoned record refused to encode"); - assert!(repo.get(&good.id()).await.unwrap().is_none()); - assert!(repo.get(&bad.id()).await.unwrap().is_none()); - } - - #[tokio::test] - async fn empty_transaction_returns_without_writing() { - let db = db().await; - let repo = Repository::::new(Arc::clone(&db)); - - let value = transaction(&db, |_tx: &mut Tx| Ok::<_, Error>("ok")) - .await - .unwrap(); - - assert_eq!(value, "ok"); - assert!(repo.get(&"missing".to_string()).await.unwrap().is_none()); - } - - #[tokio::test] - async fn delete_operations_are_committed() { - let db = db().await; - let repo = Repository::::new(Arc::clone(&db)); - let record = TxRecord { - id: "delete-me".to_string(), - payload: "hello".to_string(), - poisoned: false, - }; - repo.put(&record).await.unwrap(); - - transaction::<(), _>(&db, |tx| { - tx.delete::(&record.id())?; - Ok(()) - }) - .await - .unwrap(); - - assert!(repo.get(&record.id()).await.unwrap().is_none()); - } -} diff --git a/lib/components/fabro-store/src/slate/auth_tokens.rs b/lib/components/fabro-store/src/slate/auth_tokens.rs deleted file mode 100644 index 5fed6749d..000000000 --- a/lib/components/fabro-store/src/slate/auth_tokens.rs +++ /dev/null @@ -1,598 +0,0 @@ -use std::sync::Arc; - -use chrono::{DateTime, Utc}; -use dashmap::DashMap; -use fabro_types::IdpIdentity; -use futures::StreamExt; -use serde::{Deserialize, Serialize}; -use uuid::Uuid; - -use crate::record::{JsonCodec, Record, Repository, transaction}; -use crate::{KeyedMutex, Result}; - -const REPLAY_REVOCATION_TTL_SECONDS: i64 = 60; - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub struct RefreshToken { - pub token_hash: [u8; 32], - pub chain_id: Uuid, - pub identity: IdpIdentity, - pub login: String, - pub name: String, - pub email: String, - #[serde(default, skip_serializing_if = "String::is_empty")] - pub avatar_url: String, - pub issued_at: DateTime, - pub expires_at: DateTime, - pub last_used_at: DateTime, - pub used: bool, - pub user_agent: String, -} - -impl Record for RefreshToken { - type Id = [u8; 32]; - type Codec = JsonCodec; - - const PREFIX: &'static str = "auth/refresh"; - - fn id(&self) -> Self::Id { - self.token_hash - } -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum ConsumeOutcome { - Rotated(RefreshToken, Box), - Reused(RefreshToken), - Expired, - NotFound, -} - -pub struct RefreshTokenStore { - db: Arc, - repo: Repository, - consume_locks: KeyedMutex<[u8; 32]>, - /// In-memory only: persisting attacker-supplied hashes would be an - /// unbounded-growth surface under a token-stuffing attack. - replay_revocations: DashMap<[u8; 32], DateTime>, -} - -impl std::fmt::Debug for RefreshTokenStore { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("RefreshTokenStore").finish_non_exhaustive() - } -} - -impl RefreshTokenStore { - pub(crate) fn new(db: Arc) -> Self { - Self { - repo: Repository::new(Arc::clone(&db)), - db, - consume_locks: KeyedMutex::new(), - replay_revocations: DashMap::new(), - } - } - - pub async fn insert_refresh_token(&self, token: RefreshToken) -> Result<()> { - self.repo.put(&token).await - } - - pub async fn find_refresh_token(&self, token_hash: &[u8; 32]) -> Result> { - self.repo.get(token_hash).await - } - - pub async fn active_cli_sessions( - &self, - identity: &IdpIdentity, - now: DateTime, - ) -> Result> { - let mut active_by_chain = std::collections::HashMap::::new(); - let mut tokens = self.repo.scan_stream(); - - while let Some(result) = tokens.next().await { - let (_, token) = result?; - if token.identity != *identity || token.used || token.expires_at <= now { - continue; - } - - active_by_chain - .entry(token.chain_id) - .and_modify(|current| { - if token.last_used_at > current.last_used_at { - *current = token.clone(); - } - }) - .or_insert(token); - } - - Ok(active_by_chain.into_values().collect()) - } - - pub async fn consume_and_rotate( - &self, - presented_hash: [u8; 32], - new_token: RefreshToken, - now: DateTime, - ) -> Result { - let _guard = self.consume_locks.lock(presented_hash).await; - - let outcome = match self.repo.get(&presented_hash).await? { - None => ConsumeOutcome::NotFound, - Some(existing) if now >= existing.expires_at => ConsumeOutcome::Expired, - Some(existing) if existing.used => ConsumeOutcome::Reused(existing), - Some(existing) => { - let mut old_token = existing.clone(); - old_token.used = true; - old_token.last_used_at = now; - - transaction(&self.db, |tx| { - tx.put(&old_token)?; - tx.put(&new_token)?; - Ok(()) - }) - .await?; - - ConsumeOutcome::Rotated(old_token, Box::new(new_token)) - } - }; - - Ok(outcome) - } - - pub async fn delete_chain(&self, chain_id: Uuid) -> Result { - self.repo.gc(|token| token.chain_id == chain_id).await - } - - pub async fn delete_active_chain_for_identity( - &self, - identity: &IdpIdentity, - chain_id: Uuid, - now: DateTime, - ) -> Result { - let mut token_hashes = Vec::new(); - let mut has_active_token = false; - let mut tokens = self.repo.scan_stream(); - - while let Some(result) = tokens.next().await { - let (_, token) = result?; - if token.identity != *identity || token.chain_id != chain_id { - continue; - } - if !token.used && token.expires_at > now { - has_active_token = true; - } - token_hashes.push(token.token_hash); - } - - if !has_active_token { - return Ok(0); - } - - let deleted = u64::try_from(token_hashes.len()).unwrap_or(u64::MAX); - transaction(&self.db, |tx| { - for token_hash in &token_hashes { - tx.delete::(token_hash)?; - } - Ok(()) - }) - .await?; - Ok(deleted) - } - - pub async fn gc_expired(&self, cutoff: DateTime) -> Result { - self.repo.gc(|token| token.expires_at <= cutoff).await - } - - pub fn mark_refresh_token_replay(&self, token_hash: [u8; 32], now: DateTime) { - self.replay_revocations.insert( - token_hash, - now + chrono::Duration::seconds(REPLAY_REVOCATION_TTL_SECONDS), - ); - self.replay_revocations - .retain(|_, expires_at| *expires_at > now); - } - - pub fn was_recently_replay_revoked(&self, token_hash: &[u8; 32], now: DateTime) -> bool { - self.replay_revocations - .retain(|_, expires_at| *expires_at > now); - self.replay_revocations - .get(token_hash) - .is_some_and(|expires_at| *expires_at > now) - } -} - -#[cfg(test)] -mod tests { - use std::sync::Arc; - use std::time::Duration; - - use chrono::Duration as ChronoDuration; - use object_store::memory::InMemory; - use tokio::task::JoinSet; - use uuid::Uuid; - - use super::{ConsumeOutcome, RefreshToken, RefreshTokenStore}; - use crate::Database; - - async fn store() -> Arc { - let db = Database::new( - Arc::new(InMemory::new()), - "", - Duration::from_millis(1), - None, - ); - db.refresh_tokens().await.unwrap() - } - - fn refresh_token(hash: [u8; 32], chain_id: Uuid, used: bool) -> RefreshToken { - let now = chrono::Utc::now(); - RefreshToken { - token_hash: hash, - chain_id, - identity: fabro_types::IdpIdentity::new("https://github.com", "12345").unwrap(), - login: "octocat".to_string(), - name: "The Octocat".to_string(), - email: "octocat@example.com".to_string(), - avatar_url: String::new(), - issued_at: now, - expires_at: now + ChronoDuration::days(30), - last_used_at: now, - used, - user_agent: "fabro-test".to_string(), - } - } - - fn alternate_identity() -> fabro_types::IdpIdentity { - fabro_types::IdpIdentity::new("https://github.com", "67890").unwrap() - } - - #[tokio::test] - async fn insert_find_rotate_and_reuse_work() { - let store = store().await; - let chain_id = Uuid::new_v4(); - let old_hash = [1_u8; 32]; - let new_hash = [2_u8; 32]; - let old = refresh_token(old_hash, chain_id, false); - let new = refresh_token(new_hash, chain_id, false); - store.insert_refresh_token(old.clone()).await.unwrap(); - - assert_eq!( - store.find_refresh_token(&old_hash).await.unwrap(), - Some(old) - ); - - let rotated = store - .consume_and_rotate(old_hash, new.clone(), chrono::Utc::now()) - .await - .unwrap(); - let ConsumeOutcome::Rotated(old_used, new_saved) = rotated else { - panic!("expected rotation"); - }; - assert!(old_used.used); - assert_eq!(new_saved.token_hash, new_hash); - assert_eq!( - store.find_refresh_token(&old_hash).await.unwrap(), - Some(old_used.clone()) - ); - assert!( - store - .find_refresh_token(&old_hash) - .await - .unwrap() - .expect("rotated old token should still exist") - .used - ); - - let replay = store - .consume_and_rotate( - old_hash, - refresh_token([3_u8; 32], chain_id, false), - chrono::Utc::now(), - ) - .await - .unwrap(); - let ConsumeOutcome::Reused(reused) = replay else { - panic!("expected replay to return the original used row"); - }; - assert_eq!(reused.chain_id, chain_id); - } - - #[test] - fn deserializes_legacy_json_without_avatar_url() { - let entry: RefreshToken = serde_json::from_value(serde_json::json!({ - "token_hash": [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], - "chain_id": "00000000-0000-4000-8000-000000000000", - "identity": { - "issuer": "https://github.com", - "subject": "12345" - }, - "login": "octocat", - "name": "The Octocat", - "email": "octocat@example.com", - "issued_at": "2026-01-01T00:00:00Z", - "expires_at": "2026-02-01T00:00:00Z", - "last_used_at": "2026-01-01T00:00:00Z", - "used": false, - "user_agent": "fabro-test" - })) - .unwrap(); - - assert_eq!(entry.avatar_url, ""); - } - - #[test] - fn serializes_avatar_url_when_present() { - let mut entry = refresh_token([1_u8; 32], Uuid::new_v4(), false); - entry.avatar_url = "https://example.com/octocat.png".to_string(); - - let json = serde_json::to_value(&entry).unwrap(); - - assert_eq!(json["avatar_url"], "https://example.com/octocat.png"); - } - - #[tokio::test] - async fn missing_and_expired_tokens_are_reported() { - let store = store().await; - let chain_id = Uuid::new_v4(); - - assert_eq!( - store - .consume_and_rotate( - [7_u8; 32], - refresh_token([8_u8; 32], chain_id, false), - chrono::Utc::now(), - ) - .await - .unwrap(), - ConsumeOutcome::NotFound - ); - - let mut expired = refresh_token([9_u8; 32], chain_id, false); - expired.expires_at = chrono::Utc::now() - ChronoDuration::seconds(1); - store.insert_refresh_token(expired.clone()).await.unwrap(); - - assert_eq!( - store - .consume_and_rotate( - expired.token_hash, - refresh_token([10_u8; 32], chain_id, false), - chrono::Utc::now(), - ) - .await - .unwrap(), - ConsumeOutcome::Expired - ); - assert_eq!( - store.find_refresh_token(&expired.token_hash).await.unwrap(), - Some(expired) - ); - } - - #[tokio::test] - async fn concurrent_rotation_has_one_winner() { - let store = store().await; - let chain_id = Uuid::new_v4(); - let hash = [9_u8; 32]; - store - .insert_refresh_token(refresh_token(hash, chain_id, false)) - .await - .unwrap(); - - let mut tasks = JoinSet::new(); - for idx in 0..16_u8 { - let store = Arc::clone(&store); - tasks.spawn(async move { - store - .consume_and_rotate( - hash, - refresh_token([idx; 32], chain_id, false), - chrono::Utc::now(), - ) - .await - .unwrap() - }); - } - - let mut rotated = 0; - let mut reused = 0; - while let Some(result) = tasks.join_next().await { - match result.unwrap() { - ConsumeOutcome::Rotated(_, _) => rotated += 1, - ConsumeOutcome::Reused(_) => reused += 1, - other => panic!("unexpected outcome: {other:?}"), - } - } - - assert_eq!(rotated, 1); - assert_eq!(reused, 15); - } - - #[tokio::test] - async fn delete_chain_removes_all_matching_tokens() { - let store = store().await; - let chain_id = Uuid::new_v4(); - store - .insert_refresh_token(refresh_token([1_u8; 32], chain_id, false)) - .await - .unwrap(); - store - .insert_refresh_token(refresh_token([2_u8; 32], chain_id, true)) - .await - .unwrap(); - - assert_eq!(store.delete_chain(chain_id).await.unwrap(), 2); - assert!( - store - .find_refresh_token(&[1_u8; 32]) - .await - .unwrap() - .is_none() - ); - assert!( - store - .find_refresh_token(&[2_u8; 32]) - .await - .unwrap() - .is_none() - ); - } - - #[tokio::test] - async fn delete_active_chain_for_identity_requires_active_owned_token() { - let store = store().await; - let identity = fabro_types::IdpIdentity::new("https://github.com", "12345").unwrap(); - let chain_id = Uuid::new_v4(); - let other_chain_id = Uuid::new_v4(); - let active = refresh_token([1_u8; 32], chain_id, false); - let used = refresh_token([2_u8; 32], chain_id, true); - let mut other_identity = refresh_token([3_u8; 32], chain_id, false); - other_identity.identity = alternate_identity(); - let other_chain = refresh_token([4_u8; 32], other_chain_id, false); - - for token in [ - active.clone(), - used.clone(), - other_identity.clone(), - other_chain.clone(), - ] { - store.insert_refresh_token(token).await.unwrap(); - } - - assert_eq!( - store - .delete_active_chain_for_identity(&identity, chain_id, chrono::Utc::now()) - .await - .unwrap(), - 2 - ); - assert!( - store - .find_refresh_token(&active.token_hash) - .await - .unwrap() - .is_none() - ); - assert!( - store - .find_refresh_token(&used.token_hash) - .await - .unwrap() - .is_none() - ); - assert_eq!( - store - .find_refresh_token(&other_identity.token_hash) - .await - .unwrap(), - Some(other_identity) - ); - assert_eq!( - store - .find_refresh_token(&other_chain.token_hash) - .await - .unwrap(), - Some(other_chain) - ); - } - - #[tokio::test] - async fn delete_active_chain_for_identity_returns_zero_without_active_token() { - let store = store().await; - let identity = fabro_types::IdpIdentity::new("https://github.com", "12345").unwrap(); - let chain_id = Uuid::new_v4(); - let used = refresh_token([1_u8; 32], chain_id, true); - store.insert_refresh_token(used.clone()).await.unwrap(); - - assert_eq!( - store - .delete_active_chain_for_identity(&identity, chain_id, chrono::Utc::now()) - .await - .unwrap(), - 0 - ); - assert_eq!( - store.find_refresh_token(&used.token_hash).await.unwrap(), - Some(used) - ); - } - - #[tokio::test] - async fn active_cli_sessions_return_newest_active_token_per_chain_for_identity() { - let store = store().await; - let identity = fabro_types::IdpIdentity::new("https://github.com", "12345").unwrap(); - let now = chrono::Utc::now(); - let duplicate_chain_id = Uuid::new_v4(); - let other_chain_id = Uuid::new_v4(); - - let mut old_duplicate = refresh_token([1_u8; 32], duplicate_chain_id, false); - old_duplicate.last_used_at = now - ChronoDuration::minutes(10); - old_duplicate.issued_at = now - ChronoDuration::minutes(20); - let mut newest_duplicate = refresh_token([2_u8; 32], duplicate_chain_id, false); - newest_duplicate.last_used_at = now - ChronoDuration::minutes(1); - newest_duplicate.issued_at = now - ChronoDuration::minutes(15); - let mut other_active = refresh_token([3_u8; 32], other_chain_id, false); - other_active.last_used_at = now - ChronoDuration::minutes(3); - - store.insert_refresh_token(old_duplicate).await.unwrap(); - store - .insert_refresh_token(newest_duplicate.clone()) - .await - .unwrap(); - store - .insert_refresh_token(other_active.clone()) - .await - .unwrap(); - - let sessions = store.active_cli_sessions(&identity, now).await.unwrap(); - assert_eq!(sessions.len(), 2); - assert!(sessions.contains(&newest_duplicate)); - assert!(sessions.contains(&other_active)); - } - - #[tokio::test] - async fn active_cli_sessions_exclude_expired_used_and_other_identity_tokens() { - let store = store().await; - let identity = fabro_types::IdpIdentity::new("https://github.com", "12345").unwrap(); - let now = chrono::Utc::now(); - - let active = refresh_token([1_u8; 32], Uuid::new_v4(), false); - let mut expired = refresh_token([2_u8; 32], Uuid::new_v4(), false); - expired.expires_at = now - ChronoDuration::seconds(1); - let used = refresh_token([3_u8; 32], Uuid::new_v4(), true); - let mut other_identity = refresh_token([4_u8; 32], Uuid::new_v4(), false); - other_identity.identity = alternate_identity(); - - for token in [active.clone(), expired, used, other_identity] { - store.insert_refresh_token(token).await.unwrap(); - } - - let sessions = store.active_cli_sessions(&identity, now).await.unwrap(); - assert_eq!(sessions, vec![active]); - } - - #[tokio::test] - async fn gc_expired_removes_only_tokens_at_or_before_cutoff() { - let store = store().await; - let chain_id = Uuid::new_v4(); - - let mut expired = refresh_token([4_u8; 32], chain_id, true); - expired.expires_at = chrono::Utc::now() - ChronoDuration::days(8); - let live = refresh_token([5_u8; 32], chain_id, false); - - store.insert_refresh_token(expired.clone()).await.unwrap(); - store.insert_refresh_token(live.clone()).await.unwrap(); - - assert_eq!(store.gc_expired(chrono::Utc::now()).await.unwrap(), 1); - assert!( - store - .find_refresh_token(&expired.token_hash) - .await - .unwrap() - .is_none() - ); - assert_eq!( - store.find_refresh_token(&live.token_hash).await.unwrap(), - Some(live) - ); - } -} diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index 617d56565..2ee1058d0 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -1,5 +1,4 @@ mod auth_codes; -mod auth_tokens; mod blob_store; mod projection_cache; mod run_catalog_index; @@ -11,7 +10,6 @@ use std::sync::{Arc, OnceLock}; use std::time::Duration; pub use auth_codes::{AuthCode, AuthCodeStore}; -pub use auth_tokens::{ConsumeOutcome, RefreshToken, RefreshTokenStore}; pub use blob_store::{Blob, BlobStore}; use chrono::{DateTime, Utc}; use fabro_types::{Run, RunId, SessionId}; @@ -50,7 +48,6 @@ pub struct Database { blobs: Arc>>, catalog_index: Arc>>, auth_codes: Arc>>, - refresh_tokens: Arc>>, projection_cache: Arc, projection_cache_warmed: Arc>, run_summary_store: Arc>>, @@ -83,7 +80,6 @@ impl Database { blobs: Arc::new(OnceCell::new()), catalog_index: Arc::new(OnceCell::new()), auth_codes: Arc::new(OnceCell::new()), - refresh_tokens: Arc::new(OnceCell::new()), projection_cache: Arc::new(RunProjectionCache::default()), projection_cache_warmed: Arc::new(OnceCell::new()), run_summary_store: Arc::new(OnceLock::new()), @@ -442,15 +438,27 @@ impl Database { Ok(Arc::clone(store)) } - pub async fn refresh_tokens(&self) -> Result> { - let store = self - .refresh_tokens - .get_or_try_init(|| async { - let db = Arc::new(self.open_db().await?); - Ok::<_, Error>(Arc::new(RefreshTokenStore::new(db))) - }) + /// Delete every record under the retired `auth/refresh` prefix. + /// + /// Refresh tokens moved to SQLite without an import, so these records are + /// unreadable -- and the reaper that used to collect them is gone, so + /// nothing else would ever remove them. Returns the number of records + /// deleted; a later boot finds the prefix empty and does nothing. + pub async fn retire_refresh_token_keyspace(&self) -> Result { + let db = self.open_db().await?; + let mut iter = db + .scan_prefix(keys::SlateKey::new("auth").with("refresh").into_prefix()) .await?; - Ok(Arc::clone(store)) + let mut batch = slatedb::WriteBatch::new(); + let mut deletes = 0_u64; + while let Some(entry) = iter.next().await? { + batch.delete(entry.key); + deletes += 1; + } + if deletes > 0 { + db.write(batch).await?; + } + Ok(deletes) } #[must_use] @@ -559,6 +567,44 @@ mod tests { (object_store, store) } + #[tokio::test] + async fn retire_refresh_token_keyspace_clears_the_prefix_and_is_idempotent() { + let (_object_store, store) = make_store(); + let db = store.open_db().await.unwrap(); + + let refresh_keys = ["aaa", "bbb"].map(|id| { + keys::SlateKey::new("auth") + .with("refresh") + .with(id) + .as_ref() + .to_vec() + }); + // "auth/code" sorts adjacent to "auth/refresh" and is still live, so + // it is the neighbour a too-wide prefix delete would take with it. + let auth_code_key = keys::SlateKey::new("auth") + .with("code") + .with("keep") + .as_ref() + .to_vec(); + + let mut batch = slatedb::WriteBatch::new(); + for key in &refresh_keys { + batch.put(key.as_slice(), b"{}".as_slice()); + } + batch.put(auth_code_key.as_slice(), b"{}".as_slice()); + db.write(batch).await.unwrap(); + + assert_eq!(store.retire_refresh_token_keyspace().await.unwrap(), 2); + assert_eq!(store.retire_refresh_token_keyspace().await.unwrap(), 0); + for key in &refresh_keys { + assert!(db.get(key.as_slice()).await.unwrap().is_none()); + } + assert!( + db.get(auth_code_key.as_slice()).await.unwrap().is_some(), + "retiring refresh tokens must not touch the auth code prefix" + ); + } + async fn make_summary_store() -> (tempfile::TempDir, Arc) { let (directory, store) = test_util::sqlite_summary_store().await; (directory, Arc::new(store))