scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Merge pull request #792 from fabro-sh/codex/sqlite-authorization-codes

Browse source 
Move pending CLI authorizations to SQLite
This commit is contained in:
Scott Werner 2026-08-25 11:46:10 -04:00 • committed by GitHub
commit 679d20cb52
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
24 changed files with 665 additions and 380 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -3161,6 +3161,7 @@ dependencies = [
"percent-encoding",
"serde",
"serde_json",
"sha2 0.10.9",
"slatedb",
"sqlx",
"strum 0.28.0",

2
docs/public/administration/server-configuration.mdx
View file

@ -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 `<storage_root>/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 `<storage_root>/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.

4
docs/public/changelog/2026-07-26.mdx
View file

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

102
lib/apps/fabro-server/src/auth/cli_flow.rs
View file

@ -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::<serde_json::Value>(&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"));

2
lib/apps/fabro-server/src/auth/mod.rs
View file

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

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

@ -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(),
);

13
lib/apps/fabro-server/src/server.rs
View file

@ -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<Database>,
pub(crate) run_summaries: Arc<RunSummaryStore>,
pub(crate) auth_codes: Arc<AuthCodeStore>,
pub(crate) auth_sessions: Arc<AuthSessionStore>,
pub(crate) automations: Arc<AutomationStore>,
pub(crate) environments: Arc<EnvironmentStore>,
@ -1170,6 +1171,12 @@ impl AppState {
pub fn test_auth_session_store(&self) -> &Arc<AuthSessionStore> {
&self.stores.auth_sessions
}
/// Access the auth-code store used by this router.
#[must_use]
pub fn test_auth_code_store(&self) -> &Arc<AuthCodeStore> {
&self.stores.auth_codes
}
}
impl AppState {
@ -2442,6 +2449,7 @@ pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result<Arc<AppS
);
let run_summaries =
store.attach_run_summary_store(Arc::new(RunSummaryStore::new(db_pool.clone())));
let auth_codes = Arc::new(AuthCodeStore::new(db_pool.clone()));
let auth_sessions = Arc::new(AuthSessionStore::new(db_pool.clone()));
let mcp_server_dir = mcp_server_dir_for_active_config(&active_config_path);
let mcp_server_pool = db_pool.clone();
@ -2549,6 +2557,7 @@ pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result<Arc<AppS
stores: AppStores {
runs: store,
run_summaries,
auth_codes,
auth_sessions,
automations: automation_store,
environments: environment_store,

17
lib/apps/fabro-server/tests/it/api/cli_auth_token.rs
View file

@ -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, AuthCode, Database};
use fabro_store::{ArtifactStore, PendingCliAuthorization};
use object_store::memory::InMemory;
use sha2::{Digest, Sha256};
use tower::ServiceExt;
@ -16,7 +16,7 @@ use uuid::Uuid;
use crate::helpers::{body_json, settings_from_toml};
fn test_app(source: &str) -> (axum::Router, Arc<Database>, Arc<AppState>) {
fn test_app(source: &str) -> (axum::Router, Arc<AppState>) {
let settings = settings_from_toml(source);
let object_store: Arc<dyn object_store::ObjectStore> = 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<Database>, Arc<AppState>) {
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

1
lib/components/fabro-store/Cargo.toml
View file

@ -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"] }

370
lib/components/fabro-store/src/auth_code_store.rs Normal file
View file

@ -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<Utc>,
}
/// 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<Utc>,
) -> Result<Option<PendingCliAuthorization>> {
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<Utc>) -> Result<u64> {
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<PendingCliAuthorization> {
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<Utc>) -> 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<Utc> {
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<u8> =
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,
}));
}
}

40
lib/components/fabro-store/src/auth_session_store.rs
View file

@ -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<AuthSessionRecord> {
let identity = IdpIdentity::new(
row.try_get::<String, _>("identity_issuer")?,
row.try_get::<String, _>("identity_subject")?,
)
.map_err(|err| {
Error::Other(format!(
"stored auth session has an invalid identity: {err}"
))
})?;
Ok(AuthSessionRecord {
id: parse_uuid(&row.try_get::<String, _>("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::<String, _>("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<DateTime<Utc>> {
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> {
Uuid::parse_str(value)
.map_err(|err| Error::Other(format!("stored auth session has an invalid id: {err}")))

1
lib/components/fabro-store/src/blob_store.rs
View file

@ -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)
}

14
lib/components/fabro-store/src/error.rs
View file

@ -1,4 +1,4 @@
use fabro_types::BlobHash;
use fabro_types::{BlobHash, IdpIdentityError};
pub type Result<T> = std::result::Result<T, Error>;
@ -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}")]

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

@ -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)]

4
lib/components/fabro-store/src/record/codec.rs
View file

@ -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<R>: Send + Sync + 'static {
fn decode(bytes: &[u8]) -> Result<R>;
}
#[cfg(test)]
pub(crate) struct JsonCodec;
#[cfg(test)]
impl<R> Codec<R> for JsonCodec
where
R: Serialize + DeserializeOwned,

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

@ -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<R>` directly. See `slate/auth_codes.rs`,
//! `slate/blob_store.rs`, and `slate/run_catalog_index.rs` for the intended
//! pattern.
//! rather than exposing `Repository<R>` 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;
}

55
lib/components/fabro-store/src/record/repository.rs
View file

@ -52,13 +52,12 @@
//! async fn get(&self, id: &str) -> Result<Option<Session>> {
//! self.repo.get(&id.to_string()).await
//! }
//!
//! async fn gc_expired(&self, now: DateTime<Utc>) -> Result<u64> {
//! 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<R>` 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<R: Record> {
db: Arc<Db>,
@ -94,6 +93,7 @@ impl<R: Record> Repository<R> {
}
}
#[cfg(test)]
pub(crate) async fn get(&self, id: &R::Id) -> Result<Option<R>> {
self.db
.get(key_for_id::<R>(id)?)
@ -102,6 +102,7 @@ impl<R: Record> Repository<R> {
.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<R: Record> Repository<R> {
Err(err) => Box::pin(stream::once(async move { Err(err) })),
}
}
pub(crate) async fn gc<F>(&self, predicate: F) -> Result<u64>
where
F: Fn(&R) -> bool + Send + Sync,
{
let mut iter = self.db.scan_prefix(prefix_key::<R>(&[])?).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<Box<dyn Stream<Item = Result<T>> + 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::<TestRecord>::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::<Vec<_>>().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::<TestMarker>::new(db().await);

214
lib/components/fabro-store/src/slate/auth_codes.rs
View file

@ -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<Utc>,
}
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<AuthCode>,
consume_locks: KeyedMutex<String>,
}
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<slatedb::Db>) -> 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<Option<AuthCode>> {
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<Utc>) -> Result<u64> {
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<AuthCodeStore> {
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<chrono::Utc>) -> 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());
}
}

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

@ -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<Mutex<HashMap<RunId, Arc<RunDatabaseInner>>>>,
blobs: Arc<BlobStore>,
catalog_index: Arc<OnceCell<Arc<RunCatalogIndex>>>,
auth_codes: Arc<OnceCell<Arc<AuthCodeStore>>>,
projection_cache: Arc<RunProjectionCache>,
projection_cache_warmed: Arc<OnceCell<()>>,
run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
@ -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<Arc<AuthCodeStore>> {
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<Arc<RunCatalogIndex>> {
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")

1
lib/components/fabro-store/src/slate/run_catalog_index.rs
View file

@ -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")
}

31
lib/components/fabro-store/src/sqlite_row.rs Normal file
View file

@ -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> {
IdpIdentity::new(
row.try_get::<String, _>("identity_issuer")?,
row.try_get::<String, _>("identity_subject")?,
)
.map_err(|source| Error::InvalidStoredIdentity { record, source })
}
pub(crate) fn timestamp_from_row(
row: &SqliteRow,
record: &'static str,
field: &'static str,
) -> Result<DateTime<Utc>> {
let value: i64 = row.try_get(field)?;
DateTime::from_timestamp_millis(value).ok_or(Error::InvalidStoredTimestamp {
record,
field,
value,
})
}

30
lib/components/fabro-store/src/test_support/mod.rs
View file

@ -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)
}

22
lib/foundation/fabro-db/migrations/2026082201_oauth_authorization_codes.sql Normal file
View file

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

74
lib/foundation/fabro-db/tests/sqlite.rs
View file

@ -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::<String, _>("name"), "code_hash");
assert_eq!(columns[0].get::<String, _>("type"), "BLOB");
assert_eq!(columns[0].get::<i64, _>("notnull"), 1);
assert_eq!(columns[0].get::<i64, _>("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<String> = 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,