From 05999036aaf1afcbb29adccbc731703cd7356b44 Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Mon, 24 Aug 2026 11:31:43 -0400 Subject: [PATCH 1/5] Move pending CLI authorizations to SQLite --- Cargo.lock | 1 + .../administration/server-configuration.mdx | 2 +- docs/public/changelog/2026-07-26.mdx | 4 + lib/apps/fabro-server/src/auth/cli_flow.rs | 102 +++-- lib/apps/fabro-server/src/auth/mod.rs | 2 +- lib/apps/fabro-server/src/serve.rs | 39 +- lib/apps/fabro-server/src/server.rs | 31 +- .../tests/it/api/cli_auth_token.rs | 17 +- lib/components/fabro-store/Cargo.toml | 1 + .../src/authorization_code_store.rs | 399 ++++++++++++++++++ lib/components/fabro-store/src/error.rs | 14 +- lib/components/fabro-store/src/lib.rs | 7 +- .../fabro-store/src/record/codec.rs | 4 + lib/components/fabro-store/src/record/mod.rs | 9 +- .../fabro-store/src/record/repository.rs | 7 +- .../fabro-store/src/slate/auth_codes.rs | 214 ---------- lib/components/fabro-store/src/slate/mod.rs | 81 +++- .../fabro-store/src/test_support/mod.rs | 16 +- .../2026082201_oauth_authorization_codes.sql | 22 + lib/foundation/fabro-db/tests/sqlite.rs | 76 ++++ 20 files changed, 743 insertions(+), 305 deletions(-) create mode 100644 lib/components/fabro-store/src/authorization_code_store.rs delete mode 100644 lib/components/fabro-store/src/slate/auth_codes.rs create mode 100644 lib/foundation/fabro-db/migrations/2026082201_oauth_authorization_codes.sql diff --git a/Cargo.lock b/Cargo.lock index 44a8a8e64..36feb7e4d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3161,6 +3161,7 @@ dependencies = [ "percent-encoding", "serde", "serde_json", + "sha2 0.10.9", "slatedb", "sqlx", "strum 0.28.0", diff --git a/docs/public/administration/server-configuration.mdx b/docs/public/administration/server-configuration.mdx index 001c94a16..750923752 100644 --- a/docs/public/administration/server-configuration.mdx +++ b/docs/public/administration/server-configuration.mdx @@ -259,7 +259,7 @@ honors those hand-edited values even though the browser wizard does not manage t Shared relational state, including vault entries, server-managed definitions, and CLI auth sessions, lives at `/db/fabro.sqlite3`. Run events continue to use the `[server.slatedb]` object store. -CLI auth sessions are stored as an `auth_sessions` row per signed-in CLI, with the rotating refresh tokens for that session in `refresh_tokens`. Revoking a session from **Settings → Sessions**, or with `DELETE /api/v1/auth/sessions/{id}`, deletes the session row and its tokens together. +CLI auth sessions are stored as an `auth_sessions` row per signed-in CLI, with the rotating refresh tokens for that session in `refresh_tokens`. Pending browser-to-CLI handoffs live briefly in `oauth_authorization_codes`; the table contains a SHA-256 hash of each one-time code, never the raw bearer value. Revoking a session from **Settings → Sessions**, or with `DELETE /api/v1/auth/sessions/{id}`, deletes the session row and its tokens together. Before applying pending SQLite migrations, Fabro creates `/db/fabro.sqlite3.pre-migration.bak` with SQLite's `VACUUM INTO`. Each migration run replaces the previous snapshot, so only the most recent pre-migration backup is retained. diff --git a/docs/public/changelog/2026-07-26.mdx b/docs/public/changelog/2026-07-26.mdx index 1a94f3a93..a9ae5d33a 100644 --- a/docs/public/changelog/2026-07-26.mdx +++ b/docs/public/changelog/2026-07-26.mdx @@ -15,6 +15,10 @@ Two dates on **Settings → Sessions** were wrong as a result and are now correc Listing and revoking sessions no longer reads every refresh token the server has ever issued, so both stay fast as a workspace accumulates logins. Revoking a session removes its tokens in the same operation. +## Pending CLI logins + +Pending CLI authorization codes now live in SQLite as SHA-256 hashes and are consumed atomically on the first exchange attempt. A login that is already between browser approval and token exchange when the server upgrades cannot carry across the storage cutover; run `fabro auth login` again. These codes expire after 60 seconds, and completed logins are unaffected. + ## Refresh token replay Replaying a refresh token still revokes its whole chain immediately. One detail changed: when several requests present the same already-rotated token at once, later ones now report `refresh_token_expired` where they previously reported `refresh_token_revoked`. The CLI treats both the same way — it discards the stored credentials and prompts you to sign in again. diff --git a/lib/apps/fabro-server/src/auth/cli_flow.rs b/lib/apps/fabro-server/src/auth/cli_flow.rs index af8598405..bb64a7c17 100644 --- a/lib/apps/fabro-server/src/auth/cli_flow.rs +++ b/lib/apps/fabro-server/src/auth/cli_flow.rs @@ -28,8 +28,8 @@ use url::{Host, Url}; use crate::auth::browser_shell::browser_shell; use crate::auth::{ - self, AuthCode, AuthErrorCode, AuthSessionRecord, InitialRefreshToken, JwtSubject, - REFRESH_TOKEN_PREFIX, RotateOutcome, + self, AuthErrorCode, AuthSessionRecord, InitialRefreshToken, JwtSubject, + PendingCliAuthorization, REFRESH_TOKEN_PREFIX, RotateOutcome, }; use crate::jwt_auth::{AuthMode, ConfiguredAuth, bearer_token_from_headers}; use crate::principal_middleware::{ @@ -390,18 +390,12 @@ async fn token( ); } - let auth_codes = match state.store_ref().auth_codes().await { - Ok(store) => store, - Err(err) => { - warn!(error = %err, "Failed to open auth code store"); - return oauth_error( - StatusCode::INTERNAL_SERVER_ERROR, - "server_error", - "Could not complete authentication", - ); - } - }; - let Some(entry) = (match auth_codes.consume(code).await { + let Some(entry) = (match state + .stores + .authorization_codes + .consume(code, chrono::Utc::now()) + .await + { Ok(entry) => entry, Err(err) => { warn!(error = %err, "Failed to consume auth code"); @@ -1117,8 +1111,7 @@ async fn issue_auth_code_response( let Some(redirect_uri) = canonical_loopback_redirect_uri(redirect_uri) else { return static_error_page(INVALID_REDIRECT_URI); }; - let entry = AuthCode { - code: code.clone(), + let entry = PendingCliAuthorization { identity, login: session.login.clone(), name: session.name.clone(), @@ -1129,20 +1122,7 @@ async fn issue_auth_code_response( expires_at: chrono::Utc::now() + chrono::Duration::seconds(60), }; - let store = match state.store_ref().auth_codes().await { - Ok(store) => store, - Err(err) => { - warn!(error = %err, "Failed to open auth code store"); - return redirect_with_error( - &redirect_uri, - state_token, - "server_error", - "Could not complete GitHub sign-in", - ); - } - }; - - if let Err(err) = store.insert(entry).await { + if let Err(err) = state.stores.authorization_codes.issue(&code, &entry).await { warn!(error = %err, "Failed to persist auth code"); return redirect_with_error( &redirect_uri, @@ -1185,7 +1165,9 @@ mod tests { CliFlowCookie, DEV_TOKEN_LOGIN_INSTRUCTIONS, add_cli_flow_cookie, read_private_cli_flow, user_agent_fingerprint, web_routes, }; - use crate::auth::{self, AuthCode, AuthErrorCode, AuthSessionRecord, InitialRefreshToken}; + use crate::auth::{ + self, AuthErrorCode, AuthSessionRecord, InitialRefreshToken, PendingCliAuthorization, + }; use crate::jwt_auth::{AuthMode, ConfiguredAuth}; use crate::principal_middleware::{AuthStatus, RequestAuthContext}; use crate::server::AppState; @@ -1329,10 +1311,10 @@ client_id = "github-client-id" } async fn insert_auth_code(state: &crate::server::AppState, code: &str, verifier: &str) { - let auth_codes = state.store_ref().auth_codes().await.unwrap(); - auth_codes - .insert(AuthCode { - code: code.to_string(), + state + .stores + .authorization_codes + .issue(code, &PendingCliAuthorization { identity: fabro_types::IdpIdentity::new("https://github.com", "12345") .expect("identity should be valid"), login: "octocat".to_string(), @@ -1721,9 +1703,10 @@ client_id = "github-client-id" .nth(1) .and_then(|segment| segment.split('&').next()) .expect("auth code should be present"); - let auth_codes = state.store_ref().auth_codes().await.unwrap(); - let entry = auth_codes - .consume(code) + let entry = state + .stores + .authorization_codes + .consume(code, chrono::Utc::now()) .await .unwrap() .expect("code should exist"); @@ -2078,6 +2061,49 @@ client_id = "github-client-id" assert_eq!(body["error"], "invalid_code"); } + #[tokio::test] + async fn token_storage_failure_returns_safe_oauth_error() { + let (app, state) = test_router(github_settings("https://fabro.example")); + state.stores.authorization_codes.test_close().await; + let raw_code = "raw-code-that-must-not-escape"; + let raw_verifier = "raw-verifier-that-must-not-escape"; + let redirect_uri = "http://127.0.0.1:4444/callback"; + + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/auth/cli/token") + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from( + json!({ + "grant_type": "authorization_code", + "code": raw_code, + "code_verifier": raw_verifier, + "redirect_uri": redirect_uri + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let rendered = String::from_utf8(bytes.to_vec()).unwrap(); + assert_eq!( + serde_json::from_str::(&rendered).unwrap(), + json!({ + "error": "server_error", + "error_description": "Could not complete authentication" + }) + ); + for sensitive in [raw_code, raw_verifier, redirect_uri] { + assert!(!rendered.contains(sensitive)); + } + } + #[tokio::test] async fn token_rejects_userinfo_injected_redirect_uri() { let (app, state) = test_router(github_settings("https://fabro.example")); diff --git a/lib/apps/fabro-server/src/auth/mod.rs b/lib/apps/fabro-server/src/auth/mod.rs index b9b0244b1..a4908a914 100644 --- a/lib/apps/fabro-server/src/auth/mod.rs +++ b/lib/apps/fabro-server/src/auth/mod.rs @@ -30,7 +30,7 @@ pub(crate) const REFRESH_TOKEN_PREFIX: &str = "fabro_refresh_"; pub(crate) use browser_shell::browser_shell; pub(crate) use cli_flow::web_routes; -pub(crate) use fabro_store::AuthCode; +pub(crate) use fabro_store::PendingCliAuthorization; pub(crate) use fabro_store::auth_session_store::{ AuthSessionRecord, InitialRefreshToken, RotateOutcome, }; diff --git a/lib/apps/fabro-server/src/serve.rs b/lib/apps/fabro-server/src/serve.rs index a9cfc51af..7d0f04842 100644 --- a/lib/apps/fabro-server/src/serve.rs +++ b/lib/apps/fabro-server/src/serve.rs @@ -783,7 +783,13 @@ where ) .await .context("activating SQLite blob storage")?; - let auth_code_store = store.auth_codes().await?; + let retired_authorization_codes = retire_authorization_code_records(&store).await?; + if retired_authorization_codes > 0 { + info!( + removed = retired_authorization_codes, + "Removed retired SlateDB authorization code records" + ); + } // 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. @@ -857,7 +863,7 @@ where .await?; spawn_auth_store_reapers( - Arc::clone(&auth_code_store), + Arc::clone(&state.stores.authorization_codes), Arc::clone(&state.stores.auth_sessions), shutdown.clone(), ); @@ -1127,7 +1133,7 @@ async fn shutdown_signal() { } fn spawn_auth_store_reapers( - auth_codes: Arc, + auth_codes: Arc, auth_sessions: Arc, shutdown: CancellationToken, ) { @@ -1135,8 +1141,15 @@ fn spawn_auth_store_reapers( spawn_refresh_token_reaper(auth_sessions, shutdown); } +async fn retire_authorization_code_records(store: &fabro_store::Database) -> anyhow::Result { + store + .retire_authorization_code_keyspace() + .await + .context("retiring SlateDB authorization code records") +} + fn spawn_auth_code_reaper( - auth_codes: Arc, + auth_codes: Arc, shutdown: CancellationToken, ) { tokio::spawn(async move { @@ -1287,6 +1300,24 @@ mod tests { } } + #[tokio::test] + async fn authorization_code_retirement_failure_is_fatal_before_startup() { + let object_store: Arc = + Arc::new(object_store::memory::InMemory::new()); + let store = fabro_store::Database::new(object_store, "", Duration::from_millis(1), None); + store.test_close_slate().await.unwrap(); + + let err = super::retire_authorization_code_records(&store) + .await + .expect_err("retirement failure must abort startup"); + let chain: Vec = err.chain().map(ToString::to_string).collect(); + assert_eq!(chain[0], "retiring SlateDB authorization code records"); + assert!( + chain.len() > 1, + "retirement error should preserve its source chain: {chain:?}" + ); + } + #[tokio::test(start_paused = true)] async fn force_exit_after_shutdown_does_not_resolve_before_cancellation() { let token = CancellationToken::new(); diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index 9500f578b..bdbe33742 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -85,9 +85,9 @@ use fabro_slack::threads::ThreadRegistry; use fabro_slack::{blocks as slack_blocks, connection as slack_connection}; use fabro_static::EnvVars; use fabro_store::{ - ArtifactKey, ArtifactStore, AuthSessionStore, CachedRunProjection, Database, EventEnvelope, - EventPayload, KeyedMutex, NodeArtifact, PendingInterviewRecord, RunSummaryStore, - StageArtifactEntry, StageId, + ArtifactKey, ArtifactStore, AuthSessionStore, AuthorizationCodeStore, CachedRunProjection, + Database, EventEnvelope, EventPayload, KeyedMutex, NodeArtifact, PendingInterviewRecord, + RunSummaryStore, StageArtifactEntry, StageId, }; #[cfg(test)] use fabro_types::BlockedReason; @@ -1152,14 +1152,15 @@ pub struct AppState { } pub(crate) struct AppStores { - pub(crate) runs: Arc, - pub(crate) run_summaries: Arc, - pub(crate) auth_sessions: Arc, - pub(crate) automations: Arc, - pub(crate) environments: Arc, - pub(crate) mcp_servers: Arc, - pub(crate) vault: Arc, - pub(crate) variables: Arc, + pub(crate) runs: Arc, + pub(crate) run_summaries: Arc, + pub(crate) authorization_codes: Arc, + pub(crate) auth_sessions: Arc, + pub(crate) automations: Arc, + pub(crate) environments: Arc, + pub(crate) mcp_servers: Arc, + pub(crate) vault: Arc, + pub(crate) variables: Arc, } #[cfg(any(test, feature = "test-support"))] @@ -1170,6 +1171,12 @@ impl AppState { pub fn test_auth_session_store(&self) -> &Arc { &self.stores.auth_sessions } + + /// Access the authorization-code store used by this router. + #[must_use] + pub fn test_authorization_code_store(&self) -> &Arc { + &self.stores.authorization_codes + } } impl AppState { @@ -2442,6 +2449,7 @@ pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result anyhow::Result (axum::Router, Arc, Arc) { +fn test_app(source: &str) -> (axum::Router, Arc) { let settings = settings_from_toml(source); let object_store: Arc = Arc::new(InMemory::new()); let store = Arc::new(fabro_store::test_support::test_database( @@ -41,7 +41,7 @@ fn test_app(source: &str) -> (axum::Router, Arc, Arc) { artifact_store, ); let app = build_router_with_options(Arc::clone(&state), &auth_mode, RouterOptions::default()); - (app, store, state) + (app, state) } fn pkce_challenge(verifier: &str) -> String { @@ -54,7 +54,7 @@ fn hash_refresh_secret(secret: &str) -> [u8; 32] { #[tokio::test] async fn cli_auth_token_exchanges_code_over_public_router() { - let (app, store, _state) = test_app( + let (app, state) = test_app( r#" _version = 1 @@ -71,10 +71,9 @@ url = "https://fabro.example" client_id = "Iv1.test" "#, ); - let auth_codes = store.auth_codes().await.unwrap(); - auth_codes - .insert(AuthCode { - code: "integration-code".to_string(), + state + .test_authorization_code_store() + .issue("integration-code", &PendingCliAuthorization { identity: fabro_types::IdpIdentity::new("https://github.com", "12345").unwrap(), login: "octocat".to_string(), name: "The Octocat".to_string(), @@ -122,7 +121,7 @@ client_id = "Iv1.test" #[tokio::test] async fn cli_auth_refresh_replay_revokes_chain_over_public_router() { - let (app, _store, state) = test_app( + let (app, state) = test_app( r#" _version = 1 diff --git a/lib/components/fabro-store/Cargo.toml b/lib/components/fabro-store/Cargo.toml index 0ad54c39a..56d8e3acd 100644 --- a/lib/components/fabro-store/Cargo.toml +++ b/lib/components/fabro-store/Cargo.toml @@ -28,6 +28,7 @@ tokio-stream.workspace = true dashmap.workspace = true serde.workspace = true serde_json.workspace = true +sha2.workspace = true sqlx.workspace = true strum.workspace = true chrono = { workspace = true, features = ["serde"] } diff --git a/lib/components/fabro-store/src/authorization_code_store.rs b/lib/components/fabro-store/src/authorization_code_store.rs new file mode 100644 index 000000000..44ab3d049 --- /dev/null +++ b/lib/components/fabro-store/src/authorization_code_store.rs @@ -0,0 +1,399 @@ +//! SQLite-backed storage for pending CLI authorizations. +//! +//! The raw authorization code is a one-time bearer credential. It remains at +//! the HTTP boundary and is hashed before every database operation; the +//! stored domain type owns only the approved authorization the code unlocks. + +use chrono::{DateTime, Utc}; +use fabro_types::IdpIdentity; +use sha2::{Digest as _, Sha256}; +use sqlx::sqlite::SqliteRow; +use sqlx::{Row as _, SqlitePool}; + +use crate::{Error, Result}; + +const RECORD_NAME: &str = "pending CLI authorization"; + +/// Approved identity and OAuth context waiting for a CLI code exchange. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PendingCliAuthorization { + pub identity: IdpIdentity, + pub login: String, + pub name: String, + pub email: String, + pub avatar_url: String, + pub code_challenge: String, + pub redirect_uri: String, + pub expires_at: DateTime, +} + +/// Issues, consumes, and expires pending CLI authorizations in SQLite. +pub struct AuthorizationCodeStore { + pool: SqlitePool, +} + +impl std::fmt::Debug for AuthorizationCodeStore { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("AuthorizationCodeStore") + .finish_non_exhaustive() + } +} + +impl AuthorizationCodeStore { + #[must_use] + pub fn new(pool: SqlitePool) -> Self { + Self { pool } + } + + /// Persist a pending authorization under the SHA-256 digest of `code`. + pub async fn issue(&self, code: &str, pending: &PendingCliAuthorization) -> Result<()> { + let code_hash = hash_code(code); + sqlx::query( + r" +INSERT INTO oauth_authorization_codes ( + code_hash, identity_issuer, identity_subject, login, name, email, + avatar_url, code_challenge, redirect_uri, expires_at_ms +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +", + ) + .bind(code_hash.as_slice()) + .bind(pending.identity.issuer()) + .bind(pending.identity.subject()) + .bind(&pending.login) + .bind(&pending.name) + .bind(&pending.email) + .bind(&pending.avatar_url) + .bind(&pending.code_challenge) + .bind(&pending.redirect_uri) + .bind(pending.expires_at.timestamp_millis()) + .execute(&self.pool) + .await?; + Ok(()) + } + + /// Atomically remove the authorization for `code` and return it if live. + /// + /// Expiry is checked after deletion so every exchange attempt burns a + /// matching code, including an expired one. + pub async fn consume( + &self, + code: &str, + now: DateTime, + ) -> Result> { + let code_hash = hash_code(code); + let row = sqlx::query( + r" +DELETE FROM oauth_authorization_codes +WHERE code_hash = ? +RETURNING identity_issuer, identity_subject, login, name, email, avatar_url, + code_challenge, redirect_uri, expires_at_ms +", + ) + .bind(code_hash.as_slice()) + .fetch_optional(&self.pool) + .await?; + + let Some(row) = row else { + return Ok(None); + }; + let pending = pending_from_row(&row)?; + if pending.expires_at <= now { + return Ok(None); + } + Ok(Some(pending)) + } + + /// Delete authorizations expiring at or before `cutoff`. + pub async fn gc_expired(&self, cutoff: DateTime) -> Result { + let result = sqlx::query("DELETE FROM oauth_authorization_codes WHERE expires_at_ms <= ?") + .bind(cutoff.timestamp_millis()) + .execute(&self.pool) + .await?; + Ok(result.rows_affected()) + } + + /// Close the shared pool to exercise storage-failure paths in consumers. + #[cfg(any(test, feature = "test-support"))] + pub async fn test_close(&self) { + self.pool.close().await; + } +} + +fn hash_code(code: &str) -> [u8; 32] { + Sha256::digest(code.as_bytes()).into() +} + +fn pending_from_row(row: &SqliteRow) -> Result { + let identity = IdpIdentity::new( + row.try_get::("identity_issuer")?, + row.try_get::("identity_subject")?, + ) + .map_err(|source| Error::InvalidStoredIdentity { + record: RECORD_NAME, + source, + })?; + Ok(PendingCliAuthorization { + identity, + login: row.try_get("login")?, + name: row.try_get("name")?, + email: row.try_get("email")?, + avatar_url: row.try_get("avatar_url")?, + code_challenge: row.try_get("code_challenge")?, + redirect_uri: row.try_get("redirect_uri")?, + expires_at: timestamp_from_row(row, "expires_at_ms")?, + }) +} + +fn timestamp_from_row(row: &SqliteRow, field: &'static str) -> Result> { + let value: i64 = row.try_get(field)?; + DateTime::from_timestamp_millis(value).ok_or(Error::InvalidStoredTimestamp { + record: RECORD_NAME, + field, + value, + }) +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use chrono::{Duration, Utc}; + use fabro_types::IdpIdentity; + use sha2::{Digest as _, Sha256}; + use tokio::fs; + use tokio::task::JoinSet; + + use super::{AuthorizationCodeStore, PendingCliAuthorization}; + use crate::{Error, test_support}; + + fn pending(expires_at: chrono::DateTime) -> PendingCliAuthorization { + PendingCliAuthorization { + identity: IdpIdentity::new("https://github.com", "12345").unwrap(), + login: "octocat".to_string(), + name: "The Octocat".to_string(), + email: "octocat@example.com".to_string(), + avatar_url: "https://example.com/octocat.png".to_string(), + code_challenge: "challenge".to_string(), + redirect_uri: "http://127.0.0.1:4444/callback".to_string(), + expires_at, + } + } + + fn now() -> chrono::DateTime { + chrono::DateTime::from_timestamp_millis(Utc::now().timestamp_millis()).unwrap() + } + + #[tokio::test] + async fn authorization_code_issue_and_consume_round_trips_once() { + let (_directory, store) = test_support::sqlite_authorization_code_store().await; + let now = now(); + let expected = pending(now + Duration::seconds(60)); + store.issue("one-time-code", &expected).await.unwrap(); + + assert_eq!( + store.consume("one-time-code", now).await.unwrap(), + Some(expected) + ); + assert!(store.consume("one-time-code", now).await.unwrap().is_none()); + } + + #[tokio::test] + async fn authorization_code_concurrent_consume_has_one_winner_across_store_instances() { + let (_directory, store) = test_support::sqlite_authorization_code_store().await; + let now = now(); + store + .issue("contended-code", &pending(now + Duration::seconds(60))) + .await + .unwrap(); + let stores = [ + Arc::new(AuthorizationCodeStore::new(store.pool.clone())), + Arc::new(AuthorizationCodeStore::new(store.pool.clone())), + ]; + + let mut tasks = JoinSet::new(); + for index in 0..16 { + let store = Arc::clone(&stores[index % stores.len()]); + tasks.spawn(async move { + store + .consume("contended-code", now) + .await + .unwrap() + .is_some() + }); + } + + let mut winners = 0; + while let Some(result) = tasks.join_next().await { + if result.unwrap() { + winners += 1; + } + } + assert_eq!(winners, 1); + } + + #[tokio::test] + async fn authorization_code_expired_consume_deletes_the_row() { + let (_directory, store) = test_support::sqlite_authorization_code_store().await; + let now = now(); + store + .issue("expired-code", &pending(now - Duration::seconds(1))) + .await + .unwrap(); + + assert!(store.consume("expired-code", now).await.unwrap().is_none()); + let rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM oauth_authorization_codes") + .fetch_one(&store.pool) + .await + .unwrap(); + assert_eq!(rows, 0); + } + + #[tokio::test] + async fn authorization_code_gc_removes_only_rows_at_or_before_cutoff() { + let (_directory, store) = test_support::sqlite_authorization_code_store().await; + let now = now(); + for (code, expiry) in [ + ("before", now - Duration::seconds(1)), + ("at", now), + ("after", now + Duration::seconds(1)), + ] { + store.issue(code, &pending(expiry)).await.unwrap(); + } + + assert_eq!(store.gc_expired(now).await.unwrap(), 2); + assert!(store.consume("before", now).await.unwrap().is_none()); + assert!(store.consume("at", now).await.unwrap().is_none()); + assert!(store.consume("after", now).await.unwrap().is_some()); + } + + #[tokio::test] + async fn authorization_code_survives_reopening_the_sqlite_pool() { + let (directory, store) = test_support::sqlite_authorization_code_store().await; + let now = now(); + let expected = pending(now + Duration::seconds(60)); + store.issue("durable-code", &expected).await.unwrap(); + store.pool.close().await; + + let database = fabro_db::Database::connect(directory.path().join("fabro.sqlite3")) + .await + .unwrap(); + database.migrate().await.unwrap(); + let reopened = AuthorizationCodeStore::new(database.clone_pool()); + assert_eq!( + reopened.consume("durable-code", now).await.unwrap(), + Some(expected) + ); + } + + #[tokio::test] + async fn authorization_code_duplicate_hash_fails_without_overwriting() { + let (_directory, store) = test_support::sqlite_authorization_code_store().await; + let now = now(); + let first = pending(now + Duration::seconds(60)); + let mut second = pending(now + Duration::seconds(120)); + second.login = "different-login".to_string(); + store.issue("duplicate-code", &first).await.unwrap(); + + assert!(store.issue("duplicate-code", &second).await.is_err()); + assert_eq!( + store.consume("duplicate-code", now).await.unwrap(), + Some(first) + ); + } + + #[tokio::test] + async fn authorization_code_errors_do_not_expose_sensitive_fields() { + let (_directory, store) = test_support::sqlite_authorization_code_store().await; + let raw_code = "raw-authorization-code"; + let entry = pending(now() + Duration::seconds(60)); + store.issue(raw_code, &entry).await.unwrap(); + + let err = store.issue(raw_code, &entry).await.unwrap_err(); + let rendered = err.to_string(); + let code_hash = hex::encode(super::hash_code(raw_code)); + for sensitive in [ + raw_code, + code_hash.as_str(), + entry.code_challenge.as_str(), + entry.redirect_uri.as_str(), + entry.login.as_str(), + entry.name.as_str(), + entry.email.as_str(), + entry.avatar_url.as_str(), + ] { + assert!( + !rendered.contains(sensitive), + "storage error exposed sensitive field {sensitive:?}: {rendered}" + ); + } + } + + #[tokio::test] + async fn authorization_code_persistence_contains_hash_but_not_raw_code() { + let (directory, store) = test_support::sqlite_authorization_code_store().await; + let raw_code = "raw-authorization-code-that-must-never-be-persisted"; + store + .issue(raw_code, &pending(Utc::now() + Duration::seconds(60))) + .await + .unwrap(); + + let persisted_hash: Vec = + sqlx::query_scalar("SELECT code_hash FROM oauth_authorization_codes") + .fetch_one(&store.pool) + .await + .unwrap(); + let expected_hash: [u8; 32] = Sha256::digest(raw_code.as_bytes()).into(); + assert_eq!(persisted_hash, expected_hash); + sqlx::query("PRAGMA wal_checkpoint(TRUNCATE)") + .execute(&store.pool) + .await + .unwrap(); + store.pool.close().await; + let bytes = fs::read(directory.path().join("fabro.sqlite3")) + .await + .unwrap(); + assert!( + !bytes + .windows(raw_code.len()) + .any(|window| window == raw_code.as_bytes()) + ); + } + + #[tokio::test] + async fn authorization_code_invalid_stored_timestamp_is_typed() { + let (_directory, store) = test_support::sqlite_authorization_code_store().await; + let pending = pending(Utc::now() + Duration::seconds(60)); + let code_hash = super::hash_code("invalid-timestamp-code"); + sqlx::query( + r" +INSERT INTO oauth_authorization_codes ( + code_hash, identity_issuer, identity_subject, login, name, email, + avatar_url, code_challenge, redirect_uri, expires_at_ms +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +", + ) + .bind(code_hash.as_slice()) + .bind(pending.identity.issuer()) + .bind(pending.identity.subject()) + .bind(&pending.login) + .bind(&pending.name) + .bind(&pending.email) + .bind(&pending.avatar_url) + .bind(&pending.code_challenge) + .bind(&pending.redirect_uri) + .bind(i64::MAX) + .execute(&store.pool) + .await + .unwrap(); + + let err = store + .consume("invalid-timestamp-code", Utc::now()) + .await + .unwrap_err(); + assert!(matches!(err, Error::InvalidStoredTimestamp { + record: "pending CLI authorization", + field: "expires_at_ms", + value: i64::MAX, + })); + } +} diff --git a/lib/components/fabro-store/src/error.rs b/lib/components/fabro-store/src/error.rs index 7cc31d9f5..c75784716 100644 --- a/lib/components/fabro-store/src/error.rs +++ b/lib/components/fabro-store/src/error.rs @@ -1,4 +1,4 @@ -use fabro_types::BlobHash; +use fabro_types::{BlobHash, IdpIdentityError}; pub type Result = std::result::Result; @@ -12,6 +12,18 @@ pub enum Error { Serde(#[from] serde_json::Error), #[error("SQLite error: {0}")] Sqlite(#[from] sqlx::Error), + #[error("stored {record} has an invalid identity")] + InvalidStoredIdentity { + record: &'static str, + #[source] + source: IdpIdentityError, + }, + #[error("stored {record} has an invalid {field} timestamp: {value}")] + InvalidStoredTimestamp { + record: &'static str, + field: &'static str, + value: i64, + }, #[error("stored blob {blob_hash} has bytes that conflict with its hash")] BlobHashConflict { blob_hash: BlobHash }, #[error("stored blob data does not match requested hash {blob_hash}")] diff --git a/lib/components/fabro-store/src/lib.rs b/lib/components/fabro-store/src/lib.rs index 91363b5be..c3667744e 100644 --- a/lib/components/fabro-store/src/lib.rs +++ b/lib/components/fabro-store/src/lib.rs @@ -2,6 +2,7 @@ use chrono::{DateTime, Utc}; mod artifact_store; pub mod auth_session_store; +mod authorization_code_store; mod blob_store; mod error; mod keyed_mutex; @@ -24,6 +25,7 @@ pub use artifact_store::{ pub use auth_session_store::{ ActiveCliSession, AuthSessionRecord, AuthSessionStore, InitialRefreshToken, RotateOutcome, }; +pub use authorization_code_store::{AuthorizationCodeStore, PendingCliAuthorization}; pub use blob_store::{Blob, BlobStore}; pub use error::{Error, Result}; pub use fabro_types::{ @@ -44,10 +46,7 @@ pub use run_summary_store::{ RunSummarySortDirection, RunSummaryStore, RunSummaryVisibility, }; pub use serializable_projection::SerializableProjection; -pub use slate::{ - AuthCode, AuthCodeStore, CachedRunProjection, Database, RunCatalogIndex, RunDatabase, Runs, - UnreadableRun, -}; +pub use slate::{CachedRunProjection, Database, RunCatalogIndex, RunDatabase, Runs, UnreadableRun}; pub use types::EventPayload; #[derive(Debug, Default, Clone, PartialEq, Eq)] diff --git a/lib/components/fabro-store/src/record/codec.rs b/lib/components/fabro-store/src/record/codec.rs index 2a1eef181..6c0a8c67a 100644 --- a/lib/components/fabro-store/src/record/codec.rs +++ b/lib/components/fabro-store/src/record/codec.rs @@ -1,5 +1,7 @@ use bytes::Bytes; +#[cfg(test)] use serde::Serialize; +#[cfg(test)] use serde::de::DeserializeOwned; use crate::{Error, Result}; @@ -10,8 +12,10 @@ pub(crate) trait Codec: Send + Sync + 'static { fn decode(bytes: &[u8]) -> Result; } +#[cfg(test)] pub(crate) struct JsonCodec; +#[cfg(test)] impl Codec for JsonCodec where R: Serialize + DeserializeOwned, diff --git a/lib/components/fabro-store/src/record/mod.rs b/lib/components/fabro-store/src/record/mod.rs index 189863171..0550d34f5 100644 --- a/lib/components/fabro-store/src/record/mod.rs +++ b/lib/components/fabro-store/src/record/mod.rs @@ -7,15 +7,16 @@ //! - [`Repository`]: performs the generic get/put/delete/scan/gc operations. //! //! Production callers should add a named domain store on top of this layer -//! rather than exposing `Repository` directly. See `slate/auth_codes.rs`, -//! `slate/blob_store.rs`, and `slate/run_catalog_index.rs` for the intended -//! pattern. +//! rather than exposing `Repository` directly. See `slate/blob_store.rs` +//! and `slate/run_catalog_index.rs` for the intended pattern. mod codec; mod record_id; mod repository; -pub(crate) use codec::{Codec, JsonCodec, MarkerCodec, RawBytesCodec}; +#[cfg(test)] +pub(crate) use codec::JsonCodec; +pub(crate) use codec::{Codec, MarkerCodec, RawBytesCodec}; pub(crate) use repository::Repository; use crate::Result; diff --git a/lib/components/fabro-store/src/record/repository.rs b/lib/components/fabro-store/src/record/repository.rs index 60cbb2541..5f067b1e8 100644 --- a/lib/components/fabro-store/src/record/repository.rs +++ b/lib/components/fabro-store/src/record/repository.rs @@ -69,7 +69,9 @@ use std::sync::Arc; use futures::stream::{self}; use futures::{Stream, StreamExt}; -use slatedb::{Db, KeyValue, WriteBatch}; +#[cfg(test)] +use slatedb::WriteBatch; +use slatedb::{Db, KeyValue}; use super::{Codec, Record, RecordId}; use crate::{Error, Result, keys}; @@ -78,7 +80,7 @@ use crate::{Error, Result, keys}; /// stores. /// /// This type is intentionally `pub(crate)`: callers should interact through a -/// named store such as `AuthCodeStore` or `BlobStore`, which can add +/// named store such as `RunCatalogIndex` or `BlobStore`, which can add /// domain-specific behavior on top of the generic storage primitives here. pub(crate) struct Repository { db: Arc, @@ -161,6 +163,7 @@ impl Repository { } } + #[cfg(test)] pub(crate) async fn gc(&self, predicate: F) -> Result where F: Fn(&R) -> bool + Send + Sync, diff --git a/lib/components/fabro-store/src/slate/auth_codes.rs b/lib/components/fabro-store/src/slate/auth_codes.rs deleted file mode 100644 index 0fbdb2ab9..000000000 --- a/lib/components/fabro-store/src/slate/auth_codes.rs +++ /dev/null @@ -1,214 +0,0 @@ -use std::sync::Arc; - -use chrono::{DateTime, Utc}; -use fabro_types::IdpIdentity; -use serde::{Deserialize, Serialize}; - -use crate::record::{JsonCodec, Record, Repository}; -use crate::{KeyedMutex, Result}; - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub struct AuthCode { - pub code: String, - 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 code_challenge: String, - pub redirect_uri: String, - pub expires_at: DateTime, -} - -impl Record for AuthCode { - type Id = String; - type Codec = JsonCodec; - - const PREFIX: &'static str = "auth/code"; - - fn id(&self) -> Self::Id { - self.code.clone() - } -} - -pub struct AuthCodeStore { - repo: Repository, - consume_locks: KeyedMutex, -} - -impl std::fmt::Debug for AuthCodeStore { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("AuthCodeStore").finish_non_exhaustive() - } -} - -impl AuthCodeStore { - pub(crate) fn new(db: Arc) -> Self { - Self { - repo: Repository::new(db), - consume_locks: KeyedMutex::new(), - } - } - - pub async fn insert(&self, entry: AuthCode) -> Result<()> { - self.repo.put(&entry).await - } - - pub async fn consume(&self, code: &str) -> Result> { - let code = code.to_string(); - let _guard = self.consume_locks.lock(code.clone()).await; - let entry = self.repo.get(&code).await?; - let result = match entry { - Some(entry) if entry.expires_at > Utc::now() => { - self.repo.delete(&code).await?; - Some(entry) - } - Some(_) => { - self.repo.delete(&code).await?; - None - } - None => None, - }; - - Ok(result) - } - - pub async fn gc_expired(&self, cutoff: DateTime) -> Result { - self.repo - .gc(|auth_code| auth_code.expires_at <= cutoff) - .await - } -} - -#[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 super::{AuthCode, AuthCodeStore}; - use crate::test_support; - - async fn store() -> Arc { - let db = test_support::test_database( - Arc::new(InMemory::new()), - "", - Duration::from_millis(1), - None, - ); - db.auth_codes().await.unwrap() - } - - fn auth_code(code: &str, expires_at: chrono::DateTime) -> AuthCode { - AuthCode { - code: code.to_string(), - 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(), - code_challenge: "challenge".to_string(), - redirect_uri: "http://127.0.0.1/callback".to_string(), - expires_at, - } - } - - #[tokio::test] - async fn insert_and_consume_is_single_use() { - let store = store().await; - store - .insert(auth_code( - "code-1", - chrono::Utc::now() + ChronoDuration::seconds(60), - )) - .await - .unwrap(); - - assert!(store.consume("code-1").await.unwrap().is_some()); - assert!(store.consume("code-1").await.unwrap().is_none()); - } - - #[test] - fn deserializes_legacy_json_without_avatar_url() { - let entry: AuthCode = serde_json::from_value(serde_json::json!({ - "code": "legacy-code", - "identity": { - "issuer": "https://github.com", - "subject": "12345" - }, - "login": "octocat", - "name": "The Octocat", - "email": "octocat@example.com", - "code_challenge": "challenge", - "redirect_uri": "http://127.0.0.1/callback", - "expires_at": "2026-01-01T00:00:00Z" - })) - .unwrap(); - - assert_eq!(entry.avatar_url, ""); - } - - #[test] - fn serializes_avatar_url_when_present() { - let mut entry = auth_code("avatar-code", chrono::Utc::now()); - 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 concurrent_consume_has_one_winner() { - let store = store().await; - store - .insert(auth_code( - "code-2", - chrono::Utc::now() + ChronoDuration::seconds(60), - )) - .await - .unwrap(); - - let mut tasks = JoinSet::new(); - for _ in 0..16 { - let store = Arc::clone(&store); - tasks.spawn(async move { store.consume("code-2").await.unwrap().is_some() }); - } - - let mut successes = 0; - while let Some(result) = tasks.join_next().await { - if result.unwrap() { - successes += 1; - } - } - - assert_eq!(successes, 1); - } - - #[tokio::test] - async fn gc_expired_removes_only_expired_codes() { - let store = store().await; - store - .insert(auth_code( - "expired", - chrono::Utc::now() - ChronoDuration::seconds(1), - )) - .await - .unwrap(); - store - .insert(auth_code( - "live", - chrono::Utc::now() + ChronoDuration::seconds(60), - )) - .await - .unwrap(); - - assert_eq!(store.gc_expired(chrono::Utc::now()).await.unwrap(), 1); - assert!(store.consume("expired").await.unwrap().is_none()); - assert!(store.consume("live").await.unwrap().is_some()); - } -} diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index 77624e46a..df323ef19 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -1,4 +1,3 @@ -mod auth_codes; mod projection_cache; mod run_catalog_index; mod run_store; @@ -8,7 +7,6 @@ use std::path::PathBuf; use std::sync::{Arc, OnceLock}; use std::time::Duration; -pub use auth_codes::{AuthCode, AuthCodeStore}; use chrono::{DateTime, Utc}; use fabro_types::{Run, RunId, SessionId}; use object_store::ObjectStore; @@ -45,7 +43,6 @@ pub struct Database { active_runs: Arc>>>, blobs: Arc, catalog_index: Arc>>, - auth_codes: Arc>>, projection_cache: Arc, projection_cache_warmed: Arc>, run_summary_store: Arc>>, @@ -78,7 +75,6 @@ impl Database { active_runs: Arc::new(Mutex::new(HashMap::new())), blobs, catalog_index: Arc::new(OnceCell::new()), - auth_codes: Arc::new(OnceCell::new()), projection_cache: Arc::new(RunProjectionCache::default()), projection_cache_warmed: Arc::new(OnceCell::new()), run_summary_store: Arc::new(OnceLock::new()), @@ -417,17 +413,6 @@ impl Database { Ok(()) } - pub async fn auth_codes(&self) -> Result> { - let store = self - .auth_codes - .get_or_try_init(|| async { - let db = Arc::new(self.open_db().await?); - Ok::<_, Error>(Arc::new(AuthCodeStore::new(db))) - }) - .await?; - Ok(Arc::clone(store)) - } - pub async fn catalog_index(&self) -> Result> { let store = self .catalog_index @@ -467,6 +452,36 @@ impl Database { Ok(deletes) } + /// Delete every record under the retired `auth/code` prefix. + /// + /// Authorization codes move to SQLite without an import. Their short + /// lifetime makes them safe to discard, while deletion prevents an older + /// binary from accepting a code issued before the storage cutover. + /// Returns the number of records deleted; later boots are no-ops. + pub async fn retire_authorization_code_keyspace(&self) -> Result { + let db = self.open_db().await?; + let mut iter = db + .scan_prefix(keys::SlateKey::new("auth").with("code").into_prefix()) + .await?; + 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) + } + + /// Close the shared SlateDB handle to exercise storage-failure paths. + #[cfg(any(test, feature = "test-support"))] + pub async fn test_close_slate(&self) -> Result<()> { + self.open_db().await?.close().await?; + Ok(()) + } + #[must_use] pub fn runs(&self) -> Runs { Runs { db: self.clone() } @@ -611,6 +626,42 @@ mod tests { ); } + #[tokio::test] + async fn retire_authorization_code_keyspace_clears_only_its_prefix_and_is_idempotent() { + let (_object_store, store) = make_store(); + let db = store.open_db().await.unwrap(); + + let authorization_code_keys = ["aaa", "bbb"].map(|id| { + keys::SlateKey::new("auth") + .with("code") + .with(id) + .as_ref() + .to_vec() + }); + let neighboring_key = keys::SlateKey::new("auth") + .with("refresh") + .with("keep") + .as_ref() + .to_vec(); + + let mut batch = slatedb::WriteBatch::new(); + for key in &authorization_code_keys { + batch.put(key.as_slice(), b"{}".as_slice()); + } + batch.put(neighboring_key.as_slice(), b"{}".as_slice()); + db.write(batch).await.unwrap(); + + assert_eq!(store.retire_authorization_code_keyspace().await.unwrap(), 2); + assert_eq!(store.retire_authorization_code_keyspace().await.unwrap(), 0); + for key in &authorization_code_keys { + assert!(db.get(key.as_slice()).await.unwrap().is_none()); + } + assert!( + db.get(neighboring_key.as_slice()).await.unwrap().is_some(), + "retiring authorization codes must not touch neighboring auth prefixes" + ); + } + async fn make_summary_store() -> (tempfile::TempDir, Arc) { let (directory, store) = store_test_support::sqlite_summary_store().await; (directory, Arc::new(store)) diff --git a/lib/components/fabro-store/src/test_support/mod.rs b/lib/components/fabro-store/src/test_support/mod.rs index a77004142..c43c1544b 100644 --- a/lib/components/fabro-store/src/test_support/mod.rs +++ b/lib/components/fabro-store/src/test_support/mod.rs @@ -8,7 +8,7 @@ use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; use crate::keys::SlateKey; #[cfg(test)] -use crate::{AuthSessionStore, RunSummaryStore}; +use crate::{AuthSessionStore, AuthorizationCodeStore, RunSummaryStore}; use crate::{BlobStore, Database, Result}; /// Returns an isolated SQLite blob authority backed by its own in-memory @@ -156,6 +156,20 @@ pub(crate) async fn sqlite_auth_session_store() -> (tempfile::TempDir, AuthSessi (directory, AuthSessionStore::new(database.clone_pool())) } +#[cfg(test)] +pub(crate) async fn sqlite_authorization_code_store() -> (tempfile::TempDir, AuthorizationCodeStore) +{ + let directory = tempfile::tempdir().unwrap(); + let database = fabro_db::Database::connect(directory.path().join("fabro.sqlite3")) + .await + .unwrap(); + database.migrate().await.unwrap(); + ( + directory, + AuthorizationCodeStore::new(database.clone_pool()), + ) +} + #[cfg(test)] pub(crate) async fn sqlite_summary_store() -> (tempfile::TempDir, RunSummaryStore) { let directory = tempfile::tempdir().unwrap(); diff --git a/lib/foundation/fabro-db/migrations/2026082201_oauth_authorization_codes.sql b/lib/foundation/fabro-db/migrations/2026082201_oauth_authorization_codes.sql new file mode 100644 index 000000000..c7a9a35c0 --- /dev/null +++ b/lib/foundation/fabro-db/migrations/2026082201_oauth_authorization_codes.sql @@ -0,0 +1,22 @@ +-- A pending CLI authorization is short-lived application state. Its raw +-- bearer code stays at the HTTP boundary; SQLite receives only its SHA-256 +-- digest so a database disclosure cannot reveal an exchangeable code. +CREATE TABLE oauth_authorization_codes ( + code_hash BLOB PRIMARY KEY NOT NULL, + identity_issuer TEXT NOT NULL, + identity_subject TEXT NOT NULL, + login TEXT NOT NULL, + name TEXT NOT NULL, + email TEXT NOT NULL, + avatar_url TEXT NOT NULL DEFAULT '', + code_challenge TEXT NOT NULL, + redirect_uri TEXT NOT NULL, + expires_at_ms INTEGER NOT NULL, + + CHECK (length(code_hash) = 32), + CHECK (length(identity_issuer) > 0), + CHECK (length(identity_subject) > 0) +); + +CREATE INDEX oauth_authorization_codes_by_expiry + ON oauth_authorization_codes (expires_at_ms); diff --git a/lib/foundation/fabro-db/tests/sqlite.rs b/lib/foundation/fabro-db/tests/sqlite.rs index 4cb930b9b..3eefd7655 100644 --- a/lib/foundation/fabro-db/tests/sqlite.rs +++ b/lib/foundation/fabro-db/tests/sqlite.rs @@ -90,6 +90,14 @@ async fn connect_creates_parent_directory_and_migrate_is_idempotent() -> anyhow: assert_eq!(count, 1, "{table} table should exist"); } + let authorization_code_table_count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM sqlite_master \ + WHERE type = 'table' AND name = 'oauth_authorization_codes'", + ) + .fetch_one(database.pool()) + .await?; + assert_eq!(authorization_code_table_count, 1); + let legacy_import_table_count: i64 = sqlx::query_scalar( "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'legacy_imports'", ) @@ -583,6 +591,74 @@ async fn auth_sessions_schema_rejects_invalid_rows() -> anyhow::Result<()> { Ok(()) } +#[tokio::test] +async fn authorization_code_schema_enforces_hash_identity_and_expiry_index() -> anyhow::Result<()> { + let dir = tempfile::tempdir()?; + let database = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?; + database.migrate().await?; + + let columns = sqlx::query("PRAGMA table_info(oauth_authorization_codes)") + .fetch_all(database.pool()) + .await?; + assert_eq!(columns.len(), 10); + assert_eq!(columns[0].get::("name"), "code_hash"); + assert_eq!(columns[0].get::("type"), "BLOB"); + assert_eq!(columns[0].get::("notnull"), 1); + assert_eq!(columns[0].get::("pk"), 1); + + insert_authorization_code(database.pool(), &[1_u8; 32], "https://github.com", "12345").await?; + for (hash, issuer, subject) in [ + (vec![2_u8; 31], "https://github.com", "12345"), + (vec![2_u8; 33], "https://github.com", "12345"), + (vec![2_u8; 32], "", "12345"), + (vec![2_u8; 32], "https://github.com", ""), + ] { + assert!( + insert_authorization_code(database.pool(), &hash, issuer, subject) + .await + .is_err(), + "invalid authorization code row should be rejected: hash_len={}, issuer={issuer:?}, subject={subject:?}", + hash.len() + ); + } + + let expiry_index: Option = sqlx::query_scalar( + "SELECT name FROM sqlite_master \ + WHERE type = 'index' AND name = 'oauth_authorization_codes_by_expiry'", + ) + .fetch_optional(database.pool()) + .await?; + assert_eq!( + expiry_index.as_deref(), + Some("oauth_authorization_codes_by_expiry") + ); + + Ok(()) +} + +async fn insert_authorization_code( + pool: &fabro_db::DbPool, + code_hash: &[u8], + identity_issuer: &str, + identity_subject: &str, +) -> Result<(), sqlx::Error> { + sqlx::query( + r" +INSERT INTO oauth_authorization_codes ( + code_hash, identity_issuer, identity_subject, login, name, email, + code_challenge, redirect_uri, expires_at_ms +) VALUES (?, ?, ?, 'octocat', 'The Octocat', 'octocat@example.com', + 'challenge', 'http://127.0.0.1/callback', 1000) +", + ) + .bind(code_hash) + .bind(identity_issuer) + .bind(identity_subject) + .execute(pool) + .await?; + Ok(()) +} + async fn insert_auth_session( pool: &fabro_db::DbPool, id: &str, From a2f0167844a35a4c72d9d598ac0265781702039a Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Mon, 24 Aug 2026 13:09:32 -0400 Subject: [PATCH 2/5] Simplify SQLite authorization-code cutover Cleanup pass over the pending-CLI-authorization move to SQLite: - Extract a shared Database::retire_keyspace helper; the refresh-token and authorization-code retirements are now one-line wrappers over it. - Inline the startup retirement call (dropping the single-use wrapper, its context-chain test, and the test_close_slate hook it required) and run both SlateDB retirement scans concurrently. Error policies are unchanged: authorization codes fatal, refresh tokens best-effort. - Add a shared sqlite_row module with typed identity/timestamp row decoding, used by both AuthorizationCodeStore and AuthSessionStore; the session store's stringly Error::Other corruption errors become the typed InvalidStoredIdentity/InvalidStoredTimestamp variants. - Delete Repository::gc, which had no production callers left and was kept alive by its own test; update the record-layer docs to match. - Deduplicate the SQLite test-support bootstrap into sqlite_test_pool, reuse issue() in the invalid-timestamp test instead of a copied INSERT, and fold the new table into the existing existence-check loop in the fabro-db schema test. Co-Authored-By: Claude Fable 5 --- lib/apps/fabro-server/src/serve.rs | 42 ++++-------- .../fabro-store/src/auth_session_store.rs | 40 ++++------- .../src/authorization_code_store.rs | 68 ++++++------------- lib/components/fabro-store/src/lib.rs | 1 + lib/components/fabro-store/src/record/mod.rs | 2 +- .../fabro-store/src/record/repository.rs | 52 +------------- lib/components/fabro-store/src/slate/mod.rs | 32 +++------ lib/components/fabro-store/src/sqlite_row.rs | 31 +++++++++ .../fabro-store/src/test_support/mod.rs | 31 ++++----- lib/foundation/fabro-db/tests/sqlite.rs | 14 ++-- 10 files changed, 109 insertions(+), 204 deletions(-) create mode 100644 lib/components/fabro-store/src/sqlite_row.rs diff --git a/lib/apps/fabro-server/src/serve.rs b/lib/apps/fabro-server/src/serve.rs index 7d0f04842..bc90e604e 100644 --- a/lib/apps/fabro-server/src/serve.rs +++ b/lib/apps/fabro-server/src/serve.rs @@ -783,17 +783,24 @@ where ) .await .context("activating SQLite blob storage")?; - let retired_authorization_codes = retire_authorization_code_records(&store).await?; + // Refresh tokens and authorization codes 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. Retiring the + // authorization-code keyspace is fatal on failure; refresh-token cleanup + // stays best-effort. + let (retired_authorization_codes, retired_refresh_tokens) = tokio::join!( + store.retire_authorization_code_keyspace(), + store.retire_refresh_token_keyspace(), + ); + let retired_authorization_codes = + retired_authorization_codes.context("retiring SlateDB authorization code records")?; if retired_authorization_codes > 0 { info!( removed = retired_authorization_codes, "Removed retired SlateDB authorization code records" ); } - // 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 { + match retired_refresh_tokens { 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"), @@ -1141,13 +1148,6 @@ fn spawn_auth_store_reapers( spawn_refresh_token_reaper(auth_sessions, shutdown); } -async fn retire_authorization_code_records(store: &fabro_store::Database) -> anyhow::Result { - store - .retire_authorization_code_keyspace() - .await - .context("retiring SlateDB authorization code records") -} - fn spawn_auth_code_reaper( auth_codes: Arc, shutdown: CancellationToken, @@ -1300,24 +1300,6 @@ mod tests { } } - #[tokio::test] - async fn authorization_code_retirement_failure_is_fatal_before_startup() { - let object_store: Arc = - Arc::new(object_store::memory::InMemory::new()); - let store = fabro_store::Database::new(object_store, "", Duration::from_millis(1), None); - store.test_close_slate().await.unwrap(); - - let err = super::retire_authorization_code_records(&store) - .await - .expect_err("retirement failure must abort startup"); - let chain: Vec = err.chain().map(ToString::to_string).collect(); - assert_eq!(chain[0], "retiring SlateDB authorization code records"); - assert!( - chain.len() > 1, - "retirement error should preserve its source chain: {chain:?}" - ); - } - #[tokio::test(start_paused = true)] async fn force_exit_after_shutdown_does_not_resolve_before_cancellation() { let token = CancellationToken::new(); diff --git a/lib/components/fabro-store/src/auth_session_store.rs b/lib/components/fabro-store/src/auth_session_store.rs index ffb43e601..21c0ab28c 100644 --- a/lib/components/fabro-store/src/auth_session_store.rs +++ b/lib/components/fabro-store/src/auth_session_store.rs @@ -11,7 +11,9 @@ use sqlx::sqlite::SqliteRow; use sqlx::{Row as _, SqlitePool}; use uuid::Uuid; -use crate::{Error, Result}; +use crate::{Error, Result, sqlite_row}; + +const RECORD_NAME: &str = "auth session"; /// A CLI auth session: one rotation chain, owned by one identity. #[derive(Debug, Clone, PartialEq, Eq)] @@ -176,7 +178,8 @@ INSERT INTO auth_sessions ( .await? .into_iter() .map(|row| { - let expires_at = timestamp_from_row(&row, "expires_at_ms")?; + let expires_at = + sqlite_row::timestamp_from_row(&row, RECORD_NAME, "expires_at_ms")?; Ok(ActiveCliSession { session: session_from_row(&row)?, expires_at, @@ -376,34 +379,19 @@ async fn load_session( } fn session_from_row(row: &SqliteRow) -> Result { - let identity = IdpIdentity::new( - row.try_get::("identity_issuer")?, - row.try_get::("identity_subject")?, - ) - .map_err(|err| { - Error::Other(format!( - "stored auth session has an invalid identity: {err}" - )) - })?; Ok(AuthSessionRecord { - id: parse_uuid(&row.try_get::("id")?)?, - identity, - login: row.try_get("login")?, - name: row.try_get("name")?, - email: row.try_get("email")?, - avatar_url: row.try_get("avatar_url")?, - user_agent: row.try_get("user_agent")?, - created_at: timestamp_from_row(row, "created_at_ms")?, - last_used_at: timestamp_from_row(row, "last_used_at_ms")?, + id: parse_uuid(&row.try_get::("id")?)?, + identity: sqlite_row::identity_from_row(row, RECORD_NAME)?, + login: row.try_get("login")?, + name: row.try_get("name")?, + email: row.try_get("email")?, + avatar_url: row.try_get("avatar_url")?, + user_agent: row.try_get("user_agent")?, + created_at: sqlite_row::timestamp_from_row(row, RECORD_NAME, "created_at_ms")?, + last_used_at: sqlite_row::timestamp_from_row(row, RECORD_NAME, "last_used_at_ms")?, }) } -fn timestamp_from_row(row: &SqliteRow, column: &str) -> Result> { - let millis: i64 = row.try_get(column)?; - DateTime::from_timestamp_millis(millis) - .ok_or_else(|| Error::Other(format!("stored auth session has an invalid {column}"))) -} - fn parse_uuid(value: &str) -> Result { Uuid::parse_str(value) .map_err(|err| Error::Other(format!("stored auth session has an invalid id: {err}"))) diff --git a/lib/components/fabro-store/src/authorization_code_store.rs b/lib/components/fabro-store/src/authorization_code_store.rs index 44ab3d049..2a0ac3a09 100644 --- a/lib/components/fabro-store/src/authorization_code_store.rs +++ b/lib/components/fabro-store/src/authorization_code_store.rs @@ -10,7 +10,7 @@ use sha2::{Digest as _, Sha256}; use sqlx::sqlite::SqliteRow; use sqlx::{Row as _, SqlitePool}; -use crate::{Error, Result}; +use crate::{Result, sqlite_row}; const RECORD_NAME: &str = "pending CLI authorization"; @@ -124,32 +124,15 @@ fn hash_code(code: &str) -> [u8; 32] { } fn pending_from_row(row: &SqliteRow) -> Result { - let identity = IdpIdentity::new( - row.try_get::("identity_issuer")?, - row.try_get::("identity_subject")?, - ) - .map_err(|source| Error::InvalidStoredIdentity { - record: RECORD_NAME, - source, - })?; Ok(PendingCliAuthorization { - identity, - login: row.try_get("login")?, - name: row.try_get("name")?, - email: row.try_get("email")?, - avatar_url: row.try_get("avatar_url")?, + identity: sqlite_row::identity_from_row(row, RECORD_NAME)?, + login: row.try_get("login")?, + name: row.try_get("name")?, + email: row.try_get("email")?, + avatar_url: row.try_get("avatar_url")?, code_challenge: row.try_get("code_challenge")?, - redirect_uri: row.try_get("redirect_uri")?, - expires_at: timestamp_from_row(row, "expires_at_ms")?, - }) -} - -fn timestamp_from_row(row: &SqliteRow, field: &'static str) -> Result> { - let value: i64 = row.try_get(field)?; - DateTime::from_timestamp_millis(value).ok_or(Error::InvalidStoredTimestamp { - record: RECORD_NAME, - field, - value, + redirect_uri: row.try_get("redirect_uri")?, + expires_at: sqlite_row::timestamp_from_row(row, RECORD_NAME, "expires_at_ms")?, }) } @@ -362,29 +345,18 @@ mod tests { #[tokio::test] async fn authorization_code_invalid_stored_timestamp_is_typed() { let (_directory, store) = test_support::sqlite_authorization_code_store().await; - let pending = pending(Utc::now() + Duration::seconds(60)); - let code_hash = super::hash_code("invalid-timestamp-code"); - sqlx::query( - r" -INSERT INTO oauth_authorization_codes ( - code_hash, identity_issuer, identity_subject, login, name, email, - avatar_url, code_challenge, redirect_uri, expires_at_ms -) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) -", - ) - .bind(code_hash.as_slice()) - .bind(pending.identity.issuer()) - .bind(pending.identity.subject()) - .bind(&pending.login) - .bind(&pending.name) - .bind(&pending.email) - .bind(&pending.avatar_url) - .bind(&pending.code_challenge) - .bind(&pending.redirect_uri) - .bind(i64::MAX) - .execute(&store.pool) - .await - .unwrap(); + store + .issue( + "invalid-timestamp-code", + &pending(Utc::now() + Duration::seconds(60)), + ) + .await + .unwrap(); + sqlx::query("UPDATE oauth_authorization_codes SET expires_at_ms = ?") + .bind(i64::MAX) + .execute(&store.pool) + .await + .unwrap(); let err = store .consume("invalid-timestamp-code", Utc::now()) diff --git a/lib/components/fabro-store/src/lib.rs b/lib/components/fabro-store/src/lib.rs index c3667744e..eac3556df 100644 --- a/lib/components/fabro-store/src/lib.rs +++ b/lib/components/fabro-store/src/lib.rs @@ -14,6 +14,7 @@ mod run_state; mod run_summary_store; mod serializable_projection; mod slate; +mod sqlite_row; #[cfg(any(test, feature = "test-support"))] pub mod test_support; mod types; diff --git a/lib/components/fabro-store/src/record/mod.rs b/lib/components/fabro-store/src/record/mod.rs index 0550d34f5..61d9e5435 100644 --- a/lib/components/fabro-store/src/record/mod.rs +++ b/lib/components/fabro-store/src/record/mod.rs @@ -4,7 +4,7 @@ //! - [`Record`]: declares the key prefix, id type, and codec for one persisted //! type. //! - [`RecordId`]: converts the typed id to and from key segments. -//! - [`Repository`]: performs the generic get/put/delete/scan/gc operations. +//! - [`Repository`]: performs the generic get/put/delete/scan operations. //! //! Production callers should add a named domain store on top of this layer //! rather than exposing `Repository` directly. See `slate/blob_store.rs` diff --git a/lib/components/fabro-store/src/record/repository.rs b/lib/components/fabro-store/src/record/repository.rs index 5f067b1e8..f69b3a5aa 100644 --- a/lib/components/fabro-store/src/record/repository.rs +++ b/lib/components/fabro-store/src/record/repository.rs @@ -52,13 +52,12 @@ //! async fn get(&self, id: &str) -> Result> { //! self.repo.get(&id.to_string()).await //! } -//! -//! async fn gc_expired(&self, now: DateTime) -> Result { -//! self.repo.gc(|session| session.expires_at <= now).await -//! } //! } //! ``` //! +//! `JsonCodec` is currently compiled only for tests; un-gate it when the +//! first production JSON-encoded record type appears. +//! //! Keep `Repository` internal. Domain-specific invariants such as consume //! locks, token rotation, or marker-only behavior belong in the named store //! that wraps it, not in this generic layer. @@ -69,8 +68,6 @@ use std::sync::Arc; use futures::stream::{self}; use futures::{Stream, StreamExt}; -#[cfg(test)] -use slatedb::WriteBatch; use slatedb::{Db, KeyValue}; use super::{Codec, Record, RecordId}; @@ -162,30 +159,6 @@ impl Repository { Err(err) => Box::pin(stream::once(async move { Err(err) })), } } - - #[cfg(test)] - pub(crate) async fn gc(&self, predicate: F) -> Result - where - F: Fn(&R) -> bool + Send + Sync, - { - let mut iter = self.db.scan_prefix(prefix_key::(&[])?).await?; - let mut batch = WriteBatch::new(); - let mut deletes = 0_u64; - - while let Some(entry) = iter.next().await? { - let value = R::Codec::decode(&entry.value)?; - if predicate(&value) { - batch.delete(entry.key); - deletes += 1; - } - } - - if deletes > 0 { - self.db.write(batch).await?; - } - - Ok(deletes) - } } pub(crate) type RepositoryStream<'a, T> = Pin> + Send + 'a>>; @@ -470,25 +443,6 @@ mod tests { assert!(repo.get(&saved.id()).await.unwrap().is_none()); } - #[tokio::test] - async fn gc_deletes_matching_records() { - let repo = Repository::::new(db().await); - for record in [ - record("bucket-a", "keep", false), - record("bucket-a", "delete", true), - record("bucket-b", "keep", false), - record("bucket-b", "delete", true), - ] { - repo.put(&record).await.unwrap(); - } - - assert_eq!(repo.gc(|record| record.delete_me).await.unwrap(), 2); - - let remaining = repo.scan_stream().try_collect::>().await.unwrap(); - assert_eq!(remaining.len(), 2); - assert!(remaining.iter().all(|(_, record)| !record.delete_me)); - } - #[tokio::test] async fn marker_records_use_put_at_exists_and_scan_ids() { let repo = Repository::::new(db().await); diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index df323ef19..aac69c659 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -436,20 +436,8 @@ impl Database { /// 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?; - 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) + self.retire_keyspace(keys::SlateKey::new("auth").with("refresh")) + .await } /// Delete every record under the retired `auth/code` prefix. @@ -459,10 +447,13 @@ impl Database { /// binary from accepting a code issued before the storage cutover. /// Returns the number of records deleted; later boots are no-ops. pub async fn retire_authorization_code_keyspace(&self) -> Result { + self.retire_keyspace(keys::SlateKey::new("auth").with("code")) + .await + } + + async fn retire_keyspace(&self, keyspace: keys::SlateKey) -> Result { let db = self.open_db().await?; - let mut iter = db - .scan_prefix(keys::SlateKey::new("auth").with("code").into_prefix()) - .await?; + let mut iter = db.scan_prefix(keyspace.into_prefix()).await?; let mut batch = slatedb::WriteBatch::new(); let mut deletes = 0_u64; while let Some(entry) = iter.next().await? { @@ -475,13 +466,6 @@ impl Database { Ok(deletes) } - /// Close the shared SlateDB handle to exercise storage-failure paths. - #[cfg(any(test, feature = "test-support"))] - pub async fn test_close_slate(&self) -> Result<()> { - self.open_db().await?.close().await?; - Ok(()) - } - #[must_use] pub fn runs(&self) -> Runs { Runs { db: self.clone() } diff --git a/lib/components/fabro-store/src/sqlite_row.rs b/lib/components/fabro-store/src/sqlite_row.rs new file mode 100644 index 000000000..166bfd982 --- /dev/null +++ b/lib/components/fabro-store/src/sqlite_row.rs @@ -0,0 +1,31 @@ +//! Shared decoding helpers for columns the SQLite-backed stores have in +//! common. `record` names the stored domain type (e.g. "auth session") so +//! corruption errors say which table failed without repeating the schema. + +use chrono::{DateTime, Utc}; +use fabro_types::IdpIdentity; +use sqlx::Row as _; +use sqlx::sqlite::SqliteRow; + +use crate::{Error, Result}; + +pub(crate) fn identity_from_row(row: &SqliteRow, record: &'static str) -> Result { + IdpIdentity::new( + row.try_get::("identity_issuer")?, + row.try_get::("identity_subject")?, + ) + .map_err(|source| Error::InvalidStoredIdentity { record, source }) +} + +pub(crate) fn timestamp_from_row( + row: &SqliteRow, + record: &'static str, + field: &'static str, +) -> Result> { + let value: i64 = row.try_get(field)?; + DateTime::from_timestamp_millis(value).ok_or(Error::InvalidStoredTimestamp { + record, + field, + value, + }) +} diff --git a/lib/components/fabro-store/src/test_support/mod.rs b/lib/components/fabro-store/src/test_support/mod.rs index c43c1544b..6a9cbd548 100644 --- a/lib/components/fabro-store/src/test_support/mod.rs +++ b/lib/components/fabro-store/src/test_support/mod.rs @@ -146,28 +146,29 @@ pub async fn put_unvalidated_run_event( .await } +/// Connects to a migrated `fabro.sqlite3` in `directory` and returns its pool. #[cfg(test)] -pub(crate) async fn sqlite_auth_session_store() -> (tempfile::TempDir, AuthSessionStore) { - let directory = tempfile::tempdir().unwrap(); - let database = fabro_db::Database::connect(directory.path().join("fabro.sqlite3")) +async fn sqlite_test_pool(directory: &Path) -> sqlx::SqlitePool { + let database = fabro_db::Database::connect(directory.join("fabro.sqlite3")) .await .unwrap(); database.migrate().await.unwrap(); - (directory, AuthSessionStore::new(database.clone_pool())) + database.clone_pool() +} + +#[cfg(test)] +pub(crate) async fn sqlite_auth_session_store() -> (tempfile::TempDir, AuthSessionStore) { + let directory = tempfile::tempdir().unwrap(); + let store = AuthSessionStore::new(sqlite_test_pool(directory.path()).await); + (directory, store) } #[cfg(test)] pub(crate) async fn sqlite_authorization_code_store() -> (tempfile::TempDir, AuthorizationCodeStore) { let directory = tempfile::tempdir().unwrap(); - let database = fabro_db::Database::connect(directory.path().join("fabro.sqlite3")) - .await - .unwrap(); - database.migrate().await.unwrap(); - ( - directory, - AuthorizationCodeStore::new(database.clone_pool()), - ) + let store = AuthorizationCodeStore::new(sqlite_test_pool(directory.path()).await); + (directory, store) } #[cfg(test)] @@ -179,9 +180,5 @@ pub(crate) async fn sqlite_summary_store() -> (tempfile::TempDir, RunSummaryStor #[cfg(test)] pub(crate) async fn sqlite_summary_store_at(directory: &Path) -> RunSummaryStore { - let database = fabro_db::Database::connect(directory.join("fabro.sqlite3")) - .await - .unwrap(); - database.migrate().await.unwrap(); - RunSummaryStore::new(database.clone_pool()) + RunSummaryStore::new(sqlite_test_pool(directory).await) } diff --git a/lib/foundation/fabro-db/tests/sqlite.rs b/lib/foundation/fabro-db/tests/sqlite.rs index 3eefd7655..62cd1fcd7 100644 --- a/lib/foundation/fabro-db/tests/sqlite.rs +++ b/lib/foundation/fabro-db/tests/sqlite.rs @@ -80,7 +80,11 @@ async fn connect_creates_parent_directory_and_migrate_is_idempotent() -> anyhow: .await?; assert_eq!(blobs_table_count, 1); - for table in ["auth_sessions", "refresh_tokens"] { + for table in [ + "auth_sessions", + "refresh_tokens", + "oauth_authorization_codes", + ] { let count: i64 = sqlx::query_scalar( "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?", ) @@ -90,14 +94,6 @@ async fn connect_creates_parent_directory_and_migrate_is_idempotent() -> anyhow: assert_eq!(count, 1, "{table} table should exist"); } - let authorization_code_table_count: i64 = sqlx::query_scalar( - "SELECT COUNT(*) FROM sqlite_master \ - WHERE type = 'table' AND name = 'oauth_authorization_codes'", - ) - .fetch_one(database.pool()) - .await?; - assert_eq!(authorization_code_table_count, 1); - let legacy_import_table_count: i64 = sqlx::query_scalar( "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'legacy_imports'", ) From 68ef8c7e89aaa96a8e656659429aa8a2e52c5b2e Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Mon, 24 Aug 2026 13:24:13 -0400 Subject: [PATCH 3/5] Leave old SlateDB authorization-code records in place Drop the startup retirement of the auth/code keyspace instead of carrying one-shot cleanup code forever. The records it deleted are inert: at most a handful exist at cutover, every binary (old or new) rejects them within 60 seconds of issue via the expiry check, and nothing reads the keyspace after the move to SQLite. The refresh-token retirement keeps its original inline shape. Co-Authored-By: Claude Fable 5 --- lib/apps/fabro-server/src/serve.rs | 25 +++------ lib/components/fabro-store/src/slate/mod.rs | 60 ++------------------- 2 files changed, 12 insertions(+), 73 deletions(-) diff --git a/lib/apps/fabro-server/src/serve.rs b/lib/apps/fabro-server/src/serve.rs index bc90e604e..3e0752c82 100644 --- a/lib/apps/fabro-server/src/serve.rs +++ b/lib/apps/fabro-server/src/serve.rs @@ -783,24 +783,13 @@ where ) .await .context("activating SQLite blob storage")?; - // Refresh tokens and authorization codes 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. Retiring the - // authorization-code keyspace is fatal on failure; refresh-token cleanup - // stays best-effort. - let (retired_authorization_codes, retired_refresh_tokens) = tokio::join!( - store.retire_authorization_code_keyspace(), - store.retire_refresh_token_keyspace(), - ); - let retired_authorization_codes = - retired_authorization_codes.context("retiring SlateDB authorization code records")?; - if retired_authorization_codes > 0 { - info!( - removed = retired_authorization_codes, - "Removed retired SlateDB authorization code records" - ); - } - match retired_refresh_tokens { + // 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. Pending authorization codes + // also moved to SQLite, but their old records are left in place: at most a + // handful exist at cutover, every binary rejects them within 60 seconds of + // issue, and nothing reads their keyspace again. + 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"), diff --git a/lib/components/fabro-store/src/slate/mod.rs b/lib/components/fabro-store/src/slate/mod.rs index aac69c659..0e7d48d90 100644 --- a/lib/components/fabro-store/src/slate/mod.rs +++ b/lib/components/fabro-store/src/slate/mod.rs @@ -436,24 +436,10 @@ impl Database { /// 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 { - self.retire_keyspace(keys::SlateKey::new("auth").with("refresh")) - .await - } - - /// Delete every record under the retired `auth/code` prefix. - /// - /// Authorization codes move to SQLite without an import. Their short - /// lifetime makes them safe to discard, while deletion prevents an older - /// binary from accepting a code issued before the storage cutover. - /// Returns the number of records deleted; later boots are no-ops. - pub async fn retire_authorization_code_keyspace(&self) -> Result { - self.retire_keyspace(keys::SlateKey::new("auth").with("code")) - .await - } - - async fn retire_keyspace(&self, keyspace: keys::SlateKey) -> Result { let db = self.open_db().await?; - let mut iter = db.scan_prefix(keyspace.into_prefix()).await?; + let mut iter = db + .scan_prefix(keys::SlateKey::new("auth").with("refresh").into_prefix()) + .await?; let mut batch = slatedb::WriteBatch::new(); let mut deletes = 0_u64; while let Some(entry) = iter.next().await? { @@ -584,8 +570,8 @@ mod tests { .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. + // "auth/code" sorts adjacent to "auth/refresh", 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") @@ -610,42 +596,6 @@ mod tests { ); } - #[tokio::test] - async fn retire_authorization_code_keyspace_clears_only_its_prefix_and_is_idempotent() { - let (_object_store, store) = make_store(); - let db = store.open_db().await.unwrap(); - - let authorization_code_keys = ["aaa", "bbb"].map(|id| { - keys::SlateKey::new("auth") - .with("code") - .with(id) - .as_ref() - .to_vec() - }); - let neighboring_key = keys::SlateKey::new("auth") - .with("refresh") - .with("keep") - .as_ref() - .to_vec(); - - let mut batch = slatedb::WriteBatch::new(); - for key in &authorization_code_keys { - batch.put(key.as_slice(), b"{}".as_slice()); - } - batch.put(neighboring_key.as_slice(), b"{}".as_slice()); - db.write(batch).await.unwrap(); - - assert_eq!(store.retire_authorization_code_keyspace().await.unwrap(), 2); - assert_eq!(store.retire_authorization_code_keyspace().await.unwrap(), 0); - for key in &authorization_code_keys { - assert!(db.get(key.as_slice()).await.unwrap().is_none()); - } - assert!( - db.get(neighboring_key.as_slice()).await.unwrap().is_some(), - "retiring authorization codes must not touch neighboring auth prefixes" - ); - } - async fn make_summary_store() -> (tempfile::TempDir, Arc) { let (directory, store) = store_test_support::sqlite_summary_store().await; (directory, Arc::new(store)) From f3ff7f27a45c378c0c646447407afb385566adf9 Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Mon, 24 Aug 2026 17:12:52 -0400 Subject: [PATCH 4/5] Keep auth code store naming consistent --- lib/apps/fabro-server/src/auth/cli_flow.rs | 10 ++-- lib/apps/fabro-server/src/serve.rs | 6 +-- lib/apps/fabro-server/src/server.rs | 34 ++++++------ .../tests/it/api/cli_auth_token.rs | 2 +- ...ation_code_store.rs => auth_code_store.rs} | 53 +++++++++---------- lib/components/fabro-store/src/lib.rs | 4 +- .../fabro-store/src/test_support/mod.rs | 7 ++- 7 files changed, 57 insertions(+), 59 deletions(-) rename lib/components/fabro-store/src/{authorization_code_store.rs => auth_code_store.rs} (84%) diff --git a/lib/apps/fabro-server/src/auth/cli_flow.rs b/lib/apps/fabro-server/src/auth/cli_flow.rs index bb64a7c17..ef00f14c0 100644 --- a/lib/apps/fabro-server/src/auth/cli_flow.rs +++ b/lib/apps/fabro-server/src/auth/cli_flow.rs @@ -392,7 +392,7 @@ async fn token( let Some(entry) = (match state .stores - .authorization_codes + .auth_codes .consume(code, chrono::Utc::now()) .await { @@ -1122,7 +1122,7 @@ async fn issue_auth_code_response( expires_at: chrono::Utc::now() + chrono::Duration::seconds(60), }; - if let Err(err) = state.stores.authorization_codes.issue(&code, &entry).await { + if let Err(err) = state.stores.auth_codes.issue(&code, &entry).await { warn!(error = %err, "Failed to persist auth code"); return redirect_with_error( &redirect_uri, @@ -1313,7 +1313,7 @@ client_id = "github-client-id" async fn insert_auth_code(state: &crate::server::AppState, code: &str, verifier: &str) { state .stores - .authorization_codes + .auth_codes .issue(code, &PendingCliAuthorization { identity: fabro_types::IdpIdentity::new("https://github.com", "12345") .expect("identity should be valid"), @@ -1705,7 +1705,7 @@ client_id = "github-client-id" .expect("auth code should be present"); let entry = state .stores - .authorization_codes + .auth_codes .consume(code, chrono::Utc::now()) .await .unwrap() @@ -2064,7 +2064,7 @@ client_id = "github-client-id" #[tokio::test] async fn token_storage_failure_returns_safe_oauth_error() { let (app, state) = test_router(github_settings("https://fabro.example")); - state.stores.authorization_codes.test_close().await; + state.stores.auth_codes.test_close().await; let raw_code = "raw-code-that-must-not-escape"; let raw_verifier = "raw-verifier-that-must-not-escape"; let redirect_uri = "http://127.0.0.1:4444/callback"; diff --git a/lib/apps/fabro-server/src/serve.rs b/lib/apps/fabro-server/src/serve.rs index 3e0752c82..833082a2d 100644 --- a/lib/apps/fabro-server/src/serve.rs +++ b/lib/apps/fabro-server/src/serve.rs @@ -859,7 +859,7 @@ where .await?; spawn_auth_store_reapers( - Arc::clone(&state.stores.authorization_codes), + Arc::clone(&state.stores.auth_codes), Arc::clone(&state.stores.auth_sessions), shutdown.clone(), ); @@ -1129,7 +1129,7 @@ async fn shutdown_signal() { } fn spawn_auth_store_reapers( - auth_codes: Arc, + auth_codes: Arc, auth_sessions: Arc, shutdown: CancellationToken, ) { @@ -1138,7 +1138,7 @@ fn spawn_auth_store_reapers( } fn spawn_auth_code_reaper( - auth_codes: Arc, + auth_codes: Arc, shutdown: CancellationToken, ) { tokio::spawn(async move { diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index bdbe33742..a9f58b3d3 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -85,9 +85,9 @@ use fabro_slack::threads::ThreadRegistry; use fabro_slack::{blocks as slack_blocks, connection as slack_connection}; use fabro_static::EnvVars; use fabro_store::{ - ArtifactKey, ArtifactStore, AuthSessionStore, AuthorizationCodeStore, CachedRunProjection, - Database, EventEnvelope, EventPayload, KeyedMutex, NodeArtifact, PendingInterviewRecord, - RunSummaryStore, StageArtifactEntry, StageId, + ArtifactKey, ArtifactStore, AuthCodeStore, AuthSessionStore, CachedRunProjection, Database, + EventEnvelope, EventPayload, KeyedMutex, NodeArtifact, PendingInterviewRecord, RunSummaryStore, + StageArtifactEntry, StageId, }; #[cfg(test)] use fabro_types::BlockedReason; @@ -1152,15 +1152,15 @@ pub struct AppState { } pub(crate) struct AppStores { - pub(crate) runs: Arc, - pub(crate) run_summaries: Arc, - pub(crate) authorization_codes: Arc, - pub(crate) auth_sessions: Arc, - pub(crate) automations: Arc, - pub(crate) environments: Arc, - pub(crate) mcp_servers: Arc, - pub(crate) vault: Arc, - pub(crate) variables: Arc, + pub(crate) runs: Arc, + pub(crate) run_summaries: Arc, + pub(crate) auth_codes: Arc, + pub(crate) auth_sessions: Arc, + pub(crate) automations: Arc, + pub(crate) environments: Arc, + pub(crate) mcp_servers: Arc, + pub(crate) vault: Arc, + pub(crate) variables: Arc, } #[cfg(any(test, feature = "test-support"))] @@ -1172,10 +1172,10 @@ impl AppState { &self.stores.auth_sessions } - /// Access the authorization-code store used by this router. + /// Access the auth-code store used by this router. #[must_use] - pub fn test_authorization_code_store(&self) -> &Arc { - &self.stores.authorization_codes + pub fn test_auth_code_store(&self) -> &Arc { + &self.stores.auth_codes } } @@ -2449,7 +2449,7 @@ pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result anyhow::Result) -> std::fmt::Result { - f.debug_struct("AuthorizationCodeStore") - .finish_non_exhaustive() + f.debug_struct("AuthCodeStore").finish_non_exhaustive() } } -impl AuthorizationCodeStore { +impl AuthCodeStore { #[must_use] pub fn new(pool: SqlitePool) -> Self { Self { pool } @@ -146,7 +145,7 @@ mod tests { use tokio::fs; use tokio::task::JoinSet; - use super::{AuthorizationCodeStore, PendingCliAuthorization}; + use super::{AuthCodeStore, PendingCliAuthorization}; use crate::{Error, test_support}; fn pending(expires_at: chrono::DateTime) -> PendingCliAuthorization { @@ -167,8 +166,8 @@ mod tests { } #[tokio::test] - async fn authorization_code_issue_and_consume_round_trips_once() { - let (_directory, store) = test_support::sqlite_authorization_code_store().await; + async fn issue_and_consume_round_trips_once() { + let (_directory, store) = test_support::sqlite_auth_code_store().await; let now = now(); let expected = pending(now + Duration::seconds(60)); store.issue("one-time-code", &expected).await.unwrap(); @@ -181,16 +180,16 @@ mod tests { } #[tokio::test] - async fn authorization_code_concurrent_consume_has_one_winner_across_store_instances() { - let (_directory, store) = test_support::sqlite_authorization_code_store().await; + async fn concurrent_consume_has_one_winner_across_store_instances() { + let (_directory, store) = test_support::sqlite_auth_code_store().await; let now = now(); store .issue("contended-code", &pending(now + Duration::seconds(60))) .await .unwrap(); let stores = [ - Arc::new(AuthorizationCodeStore::new(store.pool.clone())), - Arc::new(AuthorizationCodeStore::new(store.pool.clone())), + Arc::new(AuthCodeStore::new(store.pool.clone())), + Arc::new(AuthCodeStore::new(store.pool.clone())), ]; let mut tasks = JoinSet::new(); @@ -215,8 +214,8 @@ mod tests { } #[tokio::test] - async fn authorization_code_expired_consume_deletes_the_row() { - let (_directory, store) = test_support::sqlite_authorization_code_store().await; + async fn expired_consume_deletes_the_row() { + let (_directory, store) = test_support::sqlite_auth_code_store().await; let now = now(); store .issue("expired-code", &pending(now - Duration::seconds(1))) @@ -232,8 +231,8 @@ mod tests { } #[tokio::test] - async fn authorization_code_gc_removes_only_rows_at_or_before_cutoff() { - let (_directory, store) = test_support::sqlite_authorization_code_store().await; + async fn gc_removes_only_rows_at_or_before_cutoff() { + let (_directory, store) = test_support::sqlite_auth_code_store().await; let now = now(); for (code, expiry) in [ ("before", now - Duration::seconds(1)), @@ -250,8 +249,8 @@ mod tests { } #[tokio::test] - async fn authorization_code_survives_reopening_the_sqlite_pool() { - let (directory, store) = test_support::sqlite_authorization_code_store().await; + async fn survives_reopening_the_sqlite_pool() { + let (directory, store) = test_support::sqlite_auth_code_store().await; let now = now(); let expected = pending(now + Duration::seconds(60)); store.issue("durable-code", &expected).await.unwrap(); @@ -261,7 +260,7 @@ mod tests { .await .unwrap(); database.migrate().await.unwrap(); - let reopened = AuthorizationCodeStore::new(database.clone_pool()); + let reopened = AuthCodeStore::new(database.clone_pool()); assert_eq!( reopened.consume("durable-code", now).await.unwrap(), Some(expected) @@ -269,8 +268,8 @@ mod tests { } #[tokio::test] - async fn authorization_code_duplicate_hash_fails_without_overwriting() { - let (_directory, store) = test_support::sqlite_authorization_code_store().await; + async fn duplicate_hash_fails_without_overwriting() { + let (_directory, store) = test_support::sqlite_auth_code_store().await; let now = now(); let first = pending(now + Duration::seconds(60)); let mut second = pending(now + Duration::seconds(120)); @@ -285,8 +284,8 @@ mod tests { } #[tokio::test] - async fn authorization_code_errors_do_not_expose_sensitive_fields() { - let (_directory, store) = test_support::sqlite_authorization_code_store().await; + async fn errors_do_not_expose_sensitive_fields() { + let (_directory, store) = test_support::sqlite_auth_code_store().await; let raw_code = "raw-authorization-code"; let entry = pending(now() + Duration::seconds(60)); store.issue(raw_code, &entry).await.unwrap(); @@ -312,8 +311,8 @@ mod tests { } #[tokio::test] - async fn authorization_code_persistence_contains_hash_but_not_raw_code() { - let (directory, store) = test_support::sqlite_authorization_code_store().await; + async fn persistence_contains_hash_but_not_raw_code() { + let (directory, store) = test_support::sqlite_auth_code_store().await; let raw_code = "raw-authorization-code-that-must-never-be-persisted"; store .issue(raw_code, &pending(Utc::now() + Duration::seconds(60))) @@ -343,8 +342,8 @@ mod tests { } #[tokio::test] - async fn authorization_code_invalid_stored_timestamp_is_typed() { - let (_directory, store) = test_support::sqlite_authorization_code_store().await; + async fn invalid_stored_timestamp_is_typed() { + let (_directory, store) = test_support::sqlite_auth_code_store().await; store .issue( "invalid-timestamp-code", diff --git a/lib/components/fabro-store/src/lib.rs b/lib/components/fabro-store/src/lib.rs index eac3556df..3fd34521d 100644 --- a/lib/components/fabro-store/src/lib.rs +++ b/lib/components/fabro-store/src/lib.rs @@ -1,8 +1,8 @@ use chrono::{DateTime, Utc}; mod artifact_store; +mod auth_code_store; pub mod auth_session_store; -mod authorization_code_store; mod blob_store; mod error; mod keyed_mutex; @@ -23,10 +23,10 @@ pub use artifact_store::{ ArtifactKey, ArtifactStore, NodeArtifact, StageArtifactEntry, retry_storage_segment, stage_storage_segment, }; +pub use auth_code_store::{AuthCodeStore, PendingCliAuthorization}; pub use auth_session_store::{ ActiveCliSession, AuthSessionRecord, AuthSessionStore, InitialRefreshToken, RotateOutcome, }; -pub use authorization_code_store::{AuthorizationCodeStore, PendingCliAuthorization}; pub use blob_store::{Blob, BlobStore}; pub use error::{Error, Result}; pub use fabro_types::{ diff --git a/lib/components/fabro-store/src/test_support/mod.rs b/lib/components/fabro-store/src/test_support/mod.rs index 6a9cbd548..d1a7f4848 100644 --- a/lib/components/fabro-store/src/test_support/mod.rs +++ b/lib/components/fabro-store/src/test_support/mod.rs @@ -8,7 +8,7 @@ use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; use crate::keys::SlateKey; #[cfg(test)] -use crate::{AuthSessionStore, AuthorizationCodeStore, RunSummaryStore}; +use crate::{AuthCodeStore, AuthSessionStore, RunSummaryStore}; use crate::{BlobStore, Database, Result}; /// Returns an isolated SQLite blob authority backed by its own in-memory @@ -164,10 +164,9 @@ pub(crate) async fn sqlite_auth_session_store() -> (tempfile::TempDir, AuthSessi } #[cfg(test)] -pub(crate) async fn sqlite_authorization_code_store() -> (tempfile::TempDir, AuthorizationCodeStore) -{ +pub(crate) async fn sqlite_auth_code_store() -> (tempfile::TempDir, AuthCodeStore) { let directory = tempfile::tempdir().unwrap(); - let store = AuthorizationCodeStore::new(sqlite_test_pool(directory.path()).await); + let store = AuthCodeStore::new(sqlite_test_pool(directory.path()).await); (directory, store) } From dc1f235c4867b6b2f6a5291c63bb970f1cb3708e Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Mon, 24 Aug 2026 17:31:15 -0400 Subject: [PATCH 5/5] Keep retired Slate helpers test-only --- lib/apps/fabro-server/tests/it/api/cli_auth_token.rs | 2 +- lib/components/fabro-store/src/blob_store.rs | 1 + lib/components/fabro-store/src/record/mod.rs | 1 + lib/components/fabro-store/src/record/repository.rs | 2 ++ lib/components/fabro-store/src/slate/run_catalog_index.rs | 1 + 5 files changed, 6 insertions(+), 1 deletion(-) diff --git a/lib/apps/fabro-server/tests/it/api/cli_auth_token.rs b/lib/apps/fabro-server/tests/it/api/cli_auth_token.rs index 4a79d4e21..4094f7133 100644 --- a/lib/apps/fabro-server/tests/it/api/cli_auth_token.rs +++ b/lib/apps/fabro-server/tests/it/api/cli_auth_token.rs @@ -8,7 +8,7 @@ use fabro_server::jwt_auth::resolve_auth_mode_with_lookup; use fabro_server::server::{AppState, RouterOptions, build_router_with_options}; use fabro_server::test_support::test_app_state_with_store_and_runtime_settings; use fabro_store::auth_session_store::{AuthSessionRecord, InitialRefreshToken}; -use fabro_store::{ArtifactStore, Database, PendingCliAuthorization}; +use fabro_store::{ArtifactStore, PendingCliAuthorization}; use object_store::memory::InMemory; use sha2::{Digest, Sha256}; use tower::ServiceExt; diff --git a/lib/components/fabro-store/src/blob_store.rs b/lib/components/fabro-store/src/blob_store.rs index e5d0dc128..592305c7b 100644 --- a/lib/components/fabro-store/src/blob_store.rs +++ b/lib/components/fabro-store/src/blob_store.rs @@ -31,6 +31,7 @@ impl Record for Blob { const PREFIX: &'static str = "blobs/sha256"; + #[cfg(test)] fn id(&self) -> Self::Id { BlobHash::new(&self.0) } diff --git a/lib/components/fabro-store/src/record/mod.rs b/lib/components/fabro-store/src/record/mod.rs index 61d9e5435..daecabb7a 100644 --- a/lib/components/fabro-store/src/record/mod.rs +++ b/lib/components/fabro-store/src/record/mod.rs @@ -27,6 +27,7 @@ pub(crate) trait Record: Sized + Send + Sync + 'static { const PREFIX: &'static str; + #[cfg(test)] fn id(&self) -> Self::Id; } diff --git a/lib/components/fabro-store/src/record/repository.rs b/lib/components/fabro-store/src/record/repository.rs index f69b3a5aa..6713e2190 100644 --- a/lib/components/fabro-store/src/record/repository.rs +++ b/lib/components/fabro-store/src/record/repository.rs @@ -93,6 +93,7 @@ impl Repository { } } + #[cfg(test)] pub(crate) async fn get(&self, id: &R::Id) -> Result> { self.db .get(key_for_id::(id)?) @@ -101,6 +102,7 @@ impl Repository { .transpose() } + #[cfg(test)] pub(crate) async fn put(&self, record: &R) -> Result<()> { let id = record.id(); self.put_at(&id, record).await diff --git a/lib/components/fabro-store/src/slate/run_catalog_index.rs b/lib/components/fabro-store/src/slate/run_catalog_index.rs index 33efce158..cfaae3a98 100644 --- a/lib/components/fabro-store/src/slate/run_catalog_index.rs +++ b/lib/components/fabro-store/src/slate/run_catalog_index.rs @@ -16,6 +16,7 @@ impl Record for RunCatalogEntry { const PREFIX: &'static str = "runs/_index/by-start"; + #[cfg(test)] fn id(&self) -> Self::Id { unreachable!("marker records must use put_at") }