diff --git a/Cargo.lock b/Cargo.lock index c80af6aa3..5a229fbad 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..ef00f14c0 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 + .auth_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.auth_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 + .auth_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 + .auth_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.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"; + + 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..833082a2d 100644 --- a/lib/apps/fabro-server/src/serve.rs +++ b/lib/apps/fabro-server/src/serve.rs @@ -783,10 +783,12 @@ where ) .await .context("activating SQLite blob storage")?; - 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. + // 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"), @@ -857,7 +859,7 @@ where .await?; spawn_auth_store_reapers( - Arc::clone(&auth_code_store), + Arc::clone(&state.stores.auth_codes), Arc::clone(&state.stores.auth_sessions), shutdown.clone(), ); diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index 9500f578b..a9f58b3d3 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -85,8 +85,8 @@ 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, + ArtifactKey, ArtifactStore, AuthCodeStore, AuthSessionStore, CachedRunProjection, Database, + EventEnvelope, EventPayload, KeyedMutex, NodeArtifact, PendingInterviewRecord, RunSummaryStore, StageArtifactEntry, StageId, }; #[cfg(test)] @@ -1154,6 +1154,7 @@ pub struct AppState { pub(crate) struct AppStores { 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, @@ -1170,6 +1171,12 @@ impl AppState { pub fn test_auth_session_store(&self) -> &Arc { &self.stores.auth_sessions } + + /// Access the auth-code store used by this router. + #[must_use] + pub fn test_auth_code_store(&self) -> &Arc { + &self.stores.auth_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_auth_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/auth_code_store.rs b/lib/components/fabro-store/src/auth_code_store.rs new file mode 100644 index 000000000..003da7e64 --- /dev/null +++ b/lib/components/fabro-store/src/auth_code_store.rs @@ -0,0 +1,370 @@ +//! 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::{Result, sqlite_row}; + +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 AuthCodeStore { + pool: SqlitePool, +} + +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 { + #[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 { + Ok(PendingCliAuthorization { + 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: sqlite_row::timestamp_from_row(row, RECORD_NAME, "expires_at_ms")?, + }) +} + +#[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::{AuthCodeStore, 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 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(); + + 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 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(AuthCodeStore::new(store.pool.clone())), + Arc::new(AuthCodeStore::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 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))) + .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 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)), + ("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 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(); + store.pool.close().await; + + let database = fabro_db::Database::connect(directory.path().join("fabro.sqlite3")) + .await + .unwrap(); + database.migrate().await.unwrap(); + let reopened = AuthCodeStore::new(database.clone_pool()); + assert_eq!( + reopened.consume("durable-code", now).await.unwrap(), + Some(expected) + ); + } + + #[tokio::test] + 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)); + 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 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(); + + 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 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))) + .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 invalid_stored_timestamp_is_typed() { + let (_directory, store) = test_support::sqlite_auth_code_store().await; + 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()) + .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/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/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/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..3fd34521d 100644 --- a/lib/components/fabro-store/src/lib.rs +++ b/lib/components/fabro-store/src/lib.rs @@ -1,6 +1,7 @@ use chrono::{DateTime, Utc}; mod artifact_store; +mod auth_code_store; pub mod auth_session_store; mod blob_store; mod error; @@ -13,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; @@ -21,6 +23,7 @@ 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, }; @@ -44,10 +47,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..daecabb7a 100644 --- a/lib/components/fabro-store/src/record/mod.rs +++ b/lib/components/fabro-store/src/record/mod.rs @@ -4,18 +4,19 @@ //! - [`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/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; @@ -26,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 60cbb2541..6713e2190 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,7 +68,7 @@ use std::sync::Arc; use futures::stream::{self}; use futures::{Stream, StreamExt}; -use slatedb::{Db, KeyValue, WriteBatch}; +use slatedb::{Db, KeyValue}; use super::{Codec, Record, RecordId}; use crate::{Error, Result, keys}; @@ -78,7 +77,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, @@ -94,6 +93,7 @@ impl Repository { } } + #[cfg(test)] pub(crate) async fn get(&self, id: &R::Id) -> Result> { self.db .get(key_for_id::(id)?) @@ -102,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 @@ -160,29 +161,6 @@ impl Repository { Err(err) => Box::pin(stream::once(async move { Err(err) })), } } - - 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>>; @@ -467,25 +445,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/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..0e7d48d90 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 @@ -585,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") 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") } 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 a77004142..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, RunSummaryStore}; +use crate::{AuthCodeStore, AuthSessionStore, RunSummaryStore}; use crate::{BlobStore, Database, Result}; /// Returns an isolated SQLite blob authority backed by its own in-memory @@ -146,14 +146,28 @@ 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_auth_code_store() -> (tempfile::TempDir, AuthCodeStore) { + let directory = tempfile::tempdir().unwrap(); + let store = AuthCodeStore::new(sqlite_test_pool(directory.path()).await); + (directory, store) } #[cfg(test)] @@ -165,9 +179,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/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..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 = ?", ) @@ -583,6 +587,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,