scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-08 03:10:26 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

Move pending CLI authorizations to SQLite

Browse source
This commit is contained in:
Scott Werner 2026-08-24 11:31:43 -04:00
parent 79168d3a27
commit 05999036aa
20 changed files with 743 additions and 305 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
.authorization_codes
.consume(code, chrono::Utc::now())
.await
{
Ok(entry) => entry,
Err(err) => {
warn!(error = %err, "Failed to consume auth code");
@ -1117,8 +1111,7 @@ async fn issue_auth_code_response(
let Some(redirect_uri) = canonical_loopback_redirect_uri(redirect_uri) else {
return static_error_page(INVALID_REDIRECT_URI);
};
let entry = AuthCode {
code: code.clone(),
let entry = PendingCliAuthorization {
identity,
login: session.login.clone(),
name: session.name.clone(),
@ -1129,20 +1122,7 @@ async fn issue_auth_code_response(
expires_at: chrono::Utc::now() + chrono::Duration::seconds(60),
};
let store = match state.store_ref().auth_codes().await {
Ok(store) => store,
Err(err) => {
warn!(error = %err, "Failed to open auth code store");
return redirect_with_error(
&redirect_uri,
state_token,
"server_error",
"Could not complete GitHub sign-in",
);
}
};
if let Err(err) = store.insert(entry).await {
if let Err(err) = state.stores.authorization_codes.issue(&code, &entry).await {
warn!(error = %err, "Failed to persist auth code");
return redirect_with_error(
&redirect_uri,
@ -1185,7 +1165,9 @@ mod tests {
CliFlowCookie, DEV_TOKEN_LOGIN_INSTRUCTIONS, add_cli_flow_cookie, read_private_cli_flow,
user_agent_fingerprint, web_routes,
};
use crate::auth::{self, AuthCode, AuthErrorCode, AuthSessionRecord, InitialRefreshToken};
use crate::auth::{
self, AuthErrorCode, AuthSessionRecord, InitialRefreshToken, PendingCliAuthorization,
};
use crate::jwt_auth::{AuthMode, ConfiguredAuth};
use crate::principal_middleware::{AuthStatus, RequestAuthContext};
use crate::server::AppState;
@ -1329,10 +1311,10 @@ client_id = "github-client-id"
}
async fn insert_auth_code(state: &crate::server::AppState, code: &str, verifier: &str) {
let auth_codes = state.store_ref().auth_codes().await.unwrap();
auth_codes
.insert(AuthCode {
code: code.to_string(),
state
.stores
.authorization_codes
.issue(code, &PendingCliAuthorization {
identity: fabro_types::IdpIdentity::new("https://github.com", "12345")
.expect("identity should be valid"),
login: "octocat".to_string(),
@ -1721,9 +1703,10 @@ client_id = "github-client-id"
.nth(1)
.and_then(|segment| segment.split('&').next())
.expect("auth code should be present");
let auth_codes = state.store_ref().auth_codes().await.unwrap();
let entry = auth_codes
.consume(code)
let entry = state
.stores
.authorization_codes
.consume(code, chrono::Utc::now())
.await
.unwrap()
.expect("code should exist");
@ -2078,6 +2061,49 @@ client_id = "github-client-id"
assert_eq!(body["error"], "invalid_code");
}
#[tokio::test]
async fn token_storage_failure_returns_safe_oauth_error() {
let (app, state) = test_router(github_settings("https://fabro.example"));
state.stores.authorization_codes.test_close().await;
let raw_code = "raw-code-that-must-not-escape";
let raw_verifier = "raw-verifier-that-must-not-escape";
let redirect_uri = "http://127.0.0.1:4444/callback";
let response = app
.oneshot(
Request::builder()
.method("POST")
.uri("/auth/cli/token")
.header(header::CONTENT_TYPE, "application/json")
.body(Body::from(
json!({
"grant_type": "authorization_code",
"code": raw_code,
"code_verifier": raw_verifier,
"redirect_uri": redirect_uri
})
.to_string(),
))
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
let rendered = String::from_utf8(bytes.to_vec()).unwrap();
assert_eq!(
serde_json::from_str::<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,
};

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

@ -783,7 +783,13 @@ where
)
.await
.context("activating SQLite blob storage")?;
let auth_code_store = store.auth_codes().await?;
let retired_authorization_codes = retire_authorization_code_records(&store).await?;
if retired_authorization_codes > 0 {
info!(
removed = retired_authorization_codes,
"Removed retired SlateDB authorization code records"
);
}
// Refresh tokens now live in SQLite. Nothing reads the old records and no
// reaper collects them any more, so clear them out once rather than
// leaving them in the object store forever.
@ -857,7 +863,7 @@ where
.await?;
spawn_auth_store_reapers(
Arc::clone(&auth_code_store),
Arc::clone(&state.stores.authorization_codes),
Arc::clone(&state.stores.auth_sessions),
shutdown.clone(),
);
@ -1127,7 +1133,7 @@ async fn shutdown_signal() {
}
fn spawn_auth_store_reapers(
auth_codes: Arc<fabro_store::AuthCodeStore>,
auth_codes: Arc<fabro_store::AuthorizationCodeStore>,
auth_sessions: Arc<fabro_store::AuthSessionStore>,
shutdown: CancellationToken,
) {
@ -1135,8 +1141,15 @@ fn spawn_auth_store_reapers(
spawn_refresh_token_reaper(auth_sessions, shutdown);
}
async fn retire_authorization_code_records(store: &fabro_store::Database) -> anyhow::Result<u64> {
store
.retire_authorization_code_keyspace()
.await
.context("retiring SlateDB authorization code records")
}
fn spawn_auth_code_reaper(
auth_codes: Arc<fabro_store::AuthCodeStore>,
auth_codes: Arc<fabro_store::AuthorizationCodeStore>,
shutdown: CancellationToken,
) {
tokio::spawn(async move {
@ -1287,6 +1300,24 @@ mod tests {
}
}
#[tokio::test]
async fn authorization_code_retirement_failure_is_fatal_before_startup() {
let object_store: Arc<dyn object_store::ObjectStore> =
Arc::new(object_store::memory::InMemory::new());
let store = fabro_store::Database::new(object_store, "", Duration::from_millis(1), None);
store.test_close_slate().await.unwrap();
let err = super::retire_authorization_code_records(&store)
.await
.expect_err("retirement failure must abort startup");
let chain: Vec<String> = err.chain().map(ToString::to_string).collect();
assert_eq!(chain[0], "retiring SlateDB authorization code records");
assert!(
chain.len() > 1,
"retirement error should preserve its source chain: {chain:?}"
);
}
#[tokio::test(start_paused = true)]
async fn force_exit_after_shutdown_does_not_resolve_before_cancellation() {
let token = CancellationToken::new();

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

@ -85,9 +85,9 @@ use fabro_slack::threads::ThreadRegistry;
use fabro_slack::{blocks as slack_blocks, connection as slack_connection};
use fabro_static::EnvVars;
use fabro_store::{
ArtifactKey, ArtifactStore, AuthSessionStore, CachedRunProjection, Database, EventEnvelope,
EventPayload, KeyedMutex, NodeArtifact, PendingInterviewRecord, RunSummaryStore,
StageArtifactEntry, StageId,
ArtifactKey, ArtifactStore, AuthSessionStore, AuthorizationCodeStore, CachedRunProjection,
Database, EventEnvelope, EventPayload, KeyedMutex, NodeArtifact, PendingInterviewRecord,
RunSummaryStore, StageArtifactEntry, StageId,
};
#[cfg(test)]
use fabro_types::BlockedReason;
@ -1152,14 +1152,15 @@ pub struct AppState {
}
pub(crate) struct AppStores {
pub(crate) runs: Arc<Database>,
pub(crate) run_summaries: Arc<RunSummaryStore>,
pub(crate) auth_sessions: Arc<AuthSessionStore>,
pub(crate) automations: Arc<AutomationStore>,
pub(crate) environments: Arc<EnvironmentStore>,
pub(crate) mcp_servers: Arc<McpServerStore>,
pub(crate) vault: Arc<SecretStore>,
pub(crate) variables: Arc<VariableStore>,
pub(crate) runs: Arc<Database>,
pub(crate) run_summaries: Arc<RunSummaryStore>,
pub(crate) authorization_codes: Arc<AuthorizationCodeStore>,
pub(crate) auth_sessions: Arc<AuthSessionStore>,
pub(crate) automations: Arc<AutomationStore>,
pub(crate) environments: Arc<EnvironmentStore>,
pub(crate) mcp_servers: Arc<McpServerStore>,
pub(crate) vault: Arc<SecretStore>,
pub(crate) variables: Arc<VariableStore>,
}
#[cfg(any(test, feature = "test-support"))]
@ -1170,6 +1171,12 @@ impl AppState {
pub fn test_auth_session_store(&self) -> &Arc<AuthSessionStore> {
&self.stores.auth_sessions
}
/// Access the authorization-code store used by this router.
#[must_use]
pub fn test_authorization_code_store(&self) -> &Arc<AuthorizationCodeStore> {
&self.stores.authorization_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 authorization_codes = Arc::new(AuthorizationCodeStore::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,
authorization_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, Database, 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_authorization_code_store()
.issue("integration-code", &PendingCliAuthorization {
identity: fabro_types::IdpIdentity::new("https://github.com", "12345").unwrap(),
login: "octocat".to_string(),
name: "The Octocat".to_string(),
@ -122,7 +121,7 @@ client_id = "Iv1.test"
#[tokio::test]
async fn cli_auth_refresh_replay_revokes_chain_over_public_router() {
let (app, _store, state) = test_app(
let (app, state) = test_app(
r#"
_version = 1

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

399
lib/components/fabro-store/src/authorization_code_store.rs Normal file
View file

@ -0,0 +1,399 @@
//! SQLite-backed storage for pending CLI authorizations.
//!
//! The raw authorization code is a one-time bearer credential. It remains at
//! the HTTP boundary and is hashed before every database operation; the
//! stored domain type owns only the approved authorization the code unlocks.
use chrono::{DateTime, Utc};
use fabro_types::IdpIdentity;
use sha2::{Digest as _, Sha256};
use sqlx::sqlite::SqliteRow;
use sqlx::{Row as _, SqlitePool};
use crate::{Error, Result};
const RECORD_NAME: &str = "pending CLI authorization";
/// Approved identity and OAuth context waiting for a CLI code exchange.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingCliAuthorization {
pub identity: IdpIdentity,
pub login: String,
pub name: String,
pub email: String,
pub avatar_url: String,
pub code_challenge: String,
pub redirect_uri: String,
pub expires_at: DateTime<Utc>,
}
/// Issues, consumes, and expires pending CLI authorizations in SQLite.
pub struct AuthorizationCodeStore {
pool: SqlitePool,
}
impl std::fmt::Debug for AuthorizationCodeStore {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("AuthorizationCodeStore")
.finish_non_exhaustive()
}
}
impl AuthorizationCodeStore {
#[must_use]
pub fn new(pool: SqlitePool) -> Self {
Self { pool }
}
/// Persist a pending authorization under the SHA-256 digest of `code`.
pub async fn issue(&self, code: &str, pending: &PendingCliAuthorization) -> Result<()> {
let code_hash = hash_code(code);
sqlx::query(
r"
INSERT INTO oauth_authorization_codes (
code_hash, identity_issuer, identity_subject, login, name, email,
avatar_url, code_challenge, redirect_uri, expires_at_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
",
)
.bind(code_hash.as_slice())
.bind(pending.identity.issuer())
.bind(pending.identity.subject())
.bind(&pending.login)
.bind(&pending.name)
.bind(&pending.email)
.bind(&pending.avatar_url)
.bind(&pending.code_challenge)
.bind(&pending.redirect_uri)
.bind(pending.expires_at.timestamp_millis())
.execute(&self.pool)
.await?;
Ok(())
}
/// Atomically remove the authorization for `code` and return it if live.
///
/// Expiry is checked after deletion so every exchange attempt burns a
/// matching code, including an expired one.
pub async fn consume(
&self,
code: &str,
now: DateTime<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> {
let identity = IdpIdentity::new(
row.try_get::<String, _>("identity_issuer")?,
row.try_get::<String, _>("identity_subject")?,
)
.map_err(|source| Error::InvalidStoredIdentity {
record: RECORD_NAME,
source,
})?;
Ok(PendingCliAuthorization {
identity,
login: row.try_get("login")?,
name: row.try_get("name")?,
email: row.try_get("email")?,
avatar_url: row.try_get("avatar_url")?,
code_challenge: row.try_get("code_challenge")?,
redirect_uri: row.try_get("redirect_uri")?,
expires_at: timestamp_from_row(row, "expires_at_ms")?,
})
}
fn timestamp_from_row(row: &SqliteRow, field: &'static str) -> Result<DateTime<Utc>> {
let value: i64 = row.try_get(field)?;
DateTime::from_timestamp_millis(value).ok_or(Error::InvalidStoredTimestamp {
record: RECORD_NAME,
field,
value,
})
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use chrono::{Duration, Utc};
use fabro_types::IdpIdentity;
use sha2::{Digest as _, Sha256};
use tokio::fs;
use tokio::task::JoinSet;
use super::{AuthorizationCodeStore, PendingCliAuthorization};
use crate::{Error, test_support};
fn pending(expires_at: chrono::DateTime<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 authorization_code_issue_and_consume_round_trips_once() {
let (_directory, store) = test_support::sqlite_authorization_code_store().await;
let now = now();
let expected = pending(now + Duration::seconds(60));
store.issue("one-time-code", &expected).await.unwrap();
assert_eq!(
store.consume("one-time-code", now).await.unwrap(),
Some(expected)
);
assert!(store.consume("one-time-code", now).await.unwrap().is_none());
}
#[tokio::test]
async fn authorization_code_concurrent_consume_has_one_winner_across_store_instances() {
let (_directory, store) = test_support::sqlite_authorization_code_store().await;
let now = now();
store
.issue("contended-code", &pending(now + Duration::seconds(60)))
.await
.unwrap();
let stores = [
Arc::new(AuthorizationCodeStore::new(store.pool.clone())),
Arc::new(AuthorizationCodeStore::new(store.pool.clone())),
];
let mut tasks = JoinSet::new();
for index in 0..16 {
let store = Arc::clone(&stores[index % stores.len()]);
tasks.spawn(async move {
store
.consume("contended-code", now)
.await
.unwrap()
.is_some()
});
}
let mut winners = 0;
while let Some(result) = tasks.join_next().await {
if result.unwrap() {
winners += 1;
}
}
assert_eq!(winners, 1);
}
#[tokio::test]
async fn authorization_code_expired_consume_deletes_the_row() {
let (_directory, store) = test_support::sqlite_authorization_code_store().await;
let now = now();
store
.issue("expired-code", &pending(now - Duration::seconds(1)))
.await
.unwrap();
assert!(store.consume("expired-code", now).await.unwrap().is_none());
let rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM oauth_authorization_codes")
.fetch_one(&store.pool)
.await
.unwrap();
assert_eq!(rows, 0);
}
#[tokio::test]
async fn authorization_code_gc_removes_only_rows_at_or_before_cutoff() {
let (_directory, store) = test_support::sqlite_authorization_code_store().await;
let now = now();
for (code, expiry) in [
("before", now - Duration::seconds(1)),
("at", now),
("after", now + Duration::seconds(1)),
] {
store.issue(code, &pending(expiry)).await.unwrap();
}
assert_eq!(store.gc_expired(now).await.unwrap(), 2);
assert!(store.consume("before", now).await.unwrap().is_none());
assert!(store.consume("at", now).await.unwrap().is_none());
assert!(store.consume("after", now).await.unwrap().is_some());
}
#[tokio::test]
async fn authorization_code_survives_reopening_the_sqlite_pool() {
let (directory, store) = test_support::sqlite_authorization_code_store().await;
let now = now();
let expected = pending(now + Duration::seconds(60));
store.issue("durable-code", &expected).await.unwrap();
store.pool.close().await;
let database = fabro_db::Database::connect(directory.path().join("fabro.sqlite3"))
.await
.unwrap();
database.migrate().await.unwrap();
let reopened = AuthorizationCodeStore::new(database.clone_pool());
assert_eq!(
reopened.consume("durable-code", now).await.unwrap(),
Some(expected)
);
}
#[tokio::test]
async fn authorization_code_duplicate_hash_fails_without_overwriting() {
let (_directory, store) = test_support::sqlite_authorization_code_store().await;
let now = now();
let first = pending(now + Duration::seconds(60));
let mut second = pending(now + Duration::seconds(120));
second.login = "different-login".to_string();
store.issue("duplicate-code", &first).await.unwrap();
assert!(store.issue("duplicate-code", &second).await.is_err());
assert_eq!(
store.consume("duplicate-code", now).await.unwrap(),
Some(first)
);
}
#[tokio::test]
async fn authorization_code_errors_do_not_expose_sensitive_fields() {
let (_directory, store) = test_support::sqlite_authorization_code_store().await;
let raw_code = "raw-authorization-code";
let entry = pending(now() + Duration::seconds(60));
store.issue(raw_code, &entry).await.unwrap();
let err = store.issue(raw_code, &entry).await.unwrap_err();
let rendered = err.to_string();
let code_hash = hex::encode(super::hash_code(raw_code));
for sensitive in [
raw_code,
code_hash.as_str(),
entry.code_challenge.as_str(),
entry.redirect_uri.as_str(),
entry.login.as_str(),
entry.name.as_str(),
entry.email.as_str(),
entry.avatar_url.as_str(),
] {
assert!(
!rendered.contains(sensitive),
"storage error exposed sensitive field {sensitive:?}: {rendered}"
);
}
}
#[tokio::test]
async fn authorization_code_persistence_contains_hash_but_not_raw_code() {
let (directory, store) = test_support::sqlite_authorization_code_store().await;
let raw_code = "raw-authorization-code-that-must-never-be-persisted";
store
.issue(raw_code, &pending(Utc::now() + Duration::seconds(60)))
.await
.unwrap();
let persisted_hash: Vec<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 authorization_code_invalid_stored_timestamp_is_typed() {
let (_directory, store) = test_support::sqlite_authorization_code_store().await;
let pending = pending(Utc::now() + Duration::seconds(60));
let code_hash = super::hash_code("invalid-timestamp-code");
sqlx::query(
r"
INSERT INTO oauth_authorization_codes (
code_hash, identity_issuer, identity_subject, login, name, email,
avatar_url, code_challenge, redirect_uri, expires_at_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
",
)
.bind(code_hash.as_slice())
.bind(pending.identity.issuer())
.bind(pending.identity.subject())
.bind(&pending.login)
.bind(&pending.name)
.bind(&pending.email)
.bind(&pending.avatar_url)
.bind(&pending.code_challenge)
.bind(&pending.redirect_uri)
.bind(i64::MAX)
.execute(&store.pool)
.await
.unwrap();
let err = store
.consume("invalid-timestamp-code", Utc::now())
.await
.unwrap_err();
assert!(matches!(err, Error::InvalidStoredTimestamp {
record: "pending CLI authorization",
field: "expires_at_ms",
value: i64::MAX,
}));
}
}

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

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

@ -2,6 +2,7 @@ use chrono::{DateTime, Utc};
mod artifact_store;
pub mod auth_session_store;
mod authorization_code_store;
mod blob_store;
mod error;
mod keyed_mutex;
@ -24,6 +25,7 @@ pub use artifact_store::{
pub use auth_session_store::{
ActiveCliSession, AuthSessionRecord, AuthSessionStore, InitialRefreshToken, RotateOutcome,
};
pub use authorization_code_store::{AuthorizationCodeStore, PendingCliAuthorization};
pub use blob_store::{Blob, BlobStore};
pub use error::{Error, Result};
pub use fabro_types::{
@ -44,10 +46,7 @@ pub use run_summary_store::{
RunSummarySortDirection, RunSummaryStore, RunSummaryVisibility,
};
pub use serializable_projection::SerializableProjection;
pub use slate::{
AuthCode, AuthCodeStore, CachedRunProjection, Database, RunCatalogIndex, RunDatabase, Runs,
UnreadableRun,
};
pub use slate::{CachedRunProjection, Database, RunCatalogIndex, RunDatabase, Runs, UnreadableRun};
pub use types::EventPayload;
#[derive(Debug, Default, Clone, PartialEq, Eq)]

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,

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

@ -7,15 +7,16 @@
//! - [`Repository`]: performs the generic get/put/delete/scan/gc operations.
//!
//! Production callers should add a named domain store on top of this layer
//! rather than exposing `Repository<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;

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

@ -69,7 +69,9 @@ use std::sync::Arc;
use futures::stream::{self};
use futures::{Stream, StreamExt};
use slatedb::{Db, KeyValue, WriteBatch};
#[cfg(test)]
use slatedb::WriteBatch;
use slatedb::{Db, KeyValue};
use super::{Codec, Record, RecordId};
use crate::{Error, Result, keys};
@ -78,7 +80,7 @@ use crate::{Error, Result, keys};
/// stores.
///
/// This type is intentionally `pub(crate)`: callers should interact through a
/// named store such as `AuthCodeStore` or `BlobStore`, which can add
/// named store such as `RunCatalogIndex` or `BlobStore`, which can add
/// domain-specific behavior on top of the generic storage primitives here.
pub(crate) struct Repository<R: Record> {
db: Arc<Db>,
@ -161,6 +163,7 @@ impl<R: Record> Repository<R> {
}
}
#[cfg(test)]
pub(crate) async fn gc<F>(&self, predicate: F) -> Result<u64>
where
F: Fn(&R) -> bool + Send + Sync,

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());
}
}

81
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
@ -467,6 +452,36 @@ impl Database {
Ok(deletes)
}
/// Delete every record under the retired `auth/code` prefix.
///
/// Authorization codes move to SQLite without an import. Their short
/// lifetime makes them safe to discard, while deletion prevents an older
/// binary from accepting a code issued before the storage cutover.
/// Returns the number of records deleted; later boots are no-ops.
pub async fn retire_authorization_code_keyspace(&self) -> Result<u64> {
let db = self.open_db().await?;
let mut iter = db
.scan_prefix(keys::SlateKey::new("auth").with("code").into_prefix())
.await?;
let mut batch = slatedb::WriteBatch::new();
let mut deletes = 0_u64;
while let Some(entry) = iter.next().await? {
batch.delete(entry.key);
deletes += 1;
}
if deletes > 0 {
db.write(batch).await?;
}
Ok(deletes)
}
/// Close the shared SlateDB handle to exercise storage-failure paths.
#[cfg(any(test, feature = "test-support"))]
pub async fn test_close_slate(&self) -> Result<()> {
self.open_db().await?.close().await?;
Ok(())
}
#[must_use]
pub fn runs(&self) -> Runs {
Runs { db: self.clone() }
@ -611,6 +626,42 @@ mod tests {
);
}
#[tokio::test]
async fn retire_authorization_code_keyspace_clears_only_its_prefix_and_is_idempotent() {
let (_object_store, store) = make_store();
let db = store.open_db().await.unwrap();
let authorization_code_keys = ["aaa", "bbb"].map(|id| {
keys::SlateKey::new("auth")
.with("code")
.with(id)
.as_ref()
.to_vec()
});
let neighboring_key = keys::SlateKey::new("auth")
.with("refresh")
.with("keep")
.as_ref()
.to_vec();
let mut batch = slatedb::WriteBatch::new();
for key in &authorization_code_keys {
batch.put(key.as_slice(), b"{}".as_slice());
}
batch.put(neighboring_key.as_slice(), b"{}".as_slice());
db.write(batch).await.unwrap();
assert_eq!(store.retire_authorization_code_keyspace().await.unwrap(), 2);
assert_eq!(store.retire_authorization_code_keyspace().await.unwrap(), 0);
for key in &authorization_code_keys {
assert!(db.get(key.as_slice()).await.unwrap().is_none());
}
assert!(
db.get(neighboring_key.as_slice()).await.unwrap().is_some(),
"retiring authorization codes must not touch neighboring auth prefixes"
);
}
async fn make_summary_store() -> (tempfile::TempDir, Arc<RunSummaryStore>) {
let (directory, store) = store_test_support::sqlite_summary_store().await;
(directory, Arc::new(store))

16
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::{AuthSessionStore, AuthorizationCodeStore, RunSummaryStore};
use crate::{BlobStore, Database, Result};
/// Returns an isolated SQLite blob authority backed by its own in-memory
@ -156,6 +156,20 @@ pub(crate) async fn sqlite_auth_session_store() -> (tempfile::TempDir, AuthSessi
(directory, AuthSessionStore::new(database.clone_pool()))
}
#[cfg(test)]
pub(crate) async fn sqlite_authorization_code_store() -> (tempfile::TempDir, AuthorizationCodeStore)
{
let directory = tempfile::tempdir().unwrap();
let database = fabro_db::Database::connect(directory.path().join("fabro.sqlite3"))
.await
.unwrap();
database.migrate().await.unwrap();
(
directory,
AuthorizationCodeStore::new(database.clone_pool()),
)
}
#[cfg(test)]
pub(crate) async fn sqlite_summary_store() -> (tempfile::TempDir, RunSummaryStore) {
let directory = tempfile::tempdir().unwrap();

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

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

@ -90,6 +90,14 @@ async fn connect_creates_parent_directory_and_migrate_is_idempotent() -> anyhow:
assert_eq!(count, 1, "{table} table should exist");
}
let authorization_code_table_count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM sqlite_master \
WHERE type = 'table' AND name = 'oauth_authorization_codes'",
)
.fetch_one(database.pool())
.await?;
assert_eq!(authorization_code_table_count, 1);
let legacy_import_table_count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'legacy_imports'",
)
@ -583,6 +591,74 @@ async fn auth_sessions_schema_rejects_invalid_rows() -> anyhow::Result<()> {
Ok(())
}
#[tokio::test]
async fn authorization_code_schema_enforces_hash_identity_and_expiry_index() -> anyhow::Result<()> {
let dir = tempfile::tempdir()?;
let database = fabro_db::Database::connect(dir.path().join("fabro.sqlite3")).await?;
database.migrate().await?;
let columns = sqlx::query("PRAGMA table_info(oauth_authorization_codes)")
.fetch_all(database.pool())
.await?;
assert_eq!(columns.len(), 10);
assert_eq!(columns[0].get::<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,