scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-07 08:27:12 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Delete the SlateDB refresh token store

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-07-26 00:05:31 -04:00
parent be6dd7df97
commit 33268479d3
No known key found for this signature in database
6 changed files with 73 additions and 829 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
lib/apps/fabro-server/src/serve.rs
View file

@ -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,

10
lib/components/fabro-store/src/lib.rs
View file

@ -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;

8
lib/components/fabro-store/src/record/mod.rs
View file

@ -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<R>` 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;

208
lib/components/fabro-store/src/record/transaction.rs
View file

@ -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<T, F>(db: &Db, f: F) -> Result<T>
where
F: FnOnce(&mut Tx) -> Result<T>,
{
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<R: Record>(&mut self, record: &R) -> Result<&mut Self> {
let id = record.id();
self.put_at(&id, record)
}
pub(crate) fn put_at<R: Record>(&mut self, id: &R::Id, record: &R) -> Result<&mut Self> {
self.batch
.put(key_for_id::<R>(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<R: Record>(&mut self, id: &R::Id) -> Result<&mut Self> {
self.batch.delete(key_for_id::<R>(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<FailingRecord> for FailingCodec {
fn encode(value: &FailingRecord) -> Result<Vec<u8>> {
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<FailingRecord> {
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<slatedb::Db> {
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::<TxRecord>::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::<FailingRecord>::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::<TxRecord>::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::<TxRecord>::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::<TxRecord>(&record.id())?;
Ok(())
})
.await
.unwrap();
assert!(repo.get(&record.id()).await.unwrap().is_none());
}
}

598
lib/components/fabro-store/src/slate/auth_tokens.rs
View file

@ -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<Utc>,
pub expires_at: DateTime<Utc>,
pub last_used_at: DateTime<Utc>,
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<RefreshToken>),
Reused(RefreshToken),
Expired,
NotFound,
}
pub struct RefreshTokenStore {
db: Arc<slatedb::Db>,
repo: Repository<RefreshToken>,
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<Utc>>,
}
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<slatedb::Db>) -> 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<Option<RefreshToken>> {
self.repo.get(token_hash).await
}
pub async fn active_cli_sessions(
&self,
identity: &IdpIdentity,
now: DateTime<Utc>,
) -> Result<Vec<RefreshToken>> {
let mut active_by_chain = std::collections::HashMap::<Uuid, RefreshToken>::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<Utc>,
) -> Result<ConsumeOutcome> {
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<u64> {
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<Utc>,
) -> Result<u64> {
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::<RefreshToken>(token_hash)?;
}
Ok(())
})
.await?;
Ok(deleted)
}
pub async fn gc_expired(&self, cutoff: DateTime<Utc>) -> Result<u64> {
self.repo.gc(|token| token.expires_at <= cutoff).await
}
pub fn mark_refresh_token_replay(&self, token_hash: [u8; 32], now: DateTime<Utc>) {
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<Utc>) -> 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<RefreshTokenStore> {
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)
);
}
}

70
lib/components/fabro-store/src/slate/mod.rs
View file

@ -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<OnceCell<Arc<BlobStore>>>,
catalog_index: Arc<OnceCell<Arc<RunCatalogIndex>>>,
auth_codes: Arc<OnceCell<Arc<AuthCodeStore>>>,
refresh_tokens: Arc<OnceCell<Arc<RefreshTokenStore>>>,
projection_cache: Arc<RunProjectionCache>,
projection_cache_warmed: Arc<OnceCell<()>>,
run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
@ -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<Arc<RefreshTokenStore>> {
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<u64> {
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<RunSummaryStore>) {
let (directory, store) = test_util::sqlite_summary_store().await;
(directory, Arc::new(store))