//! Subject-scoped ephemeral-state purge for `user disable`. //! //! [`purge_subject`] deletes every Redis record whose `subject` field names //! the user (canonical email or user-id string): login sessions (plus their //! `:meta` sidecars), the lockout counters (the account key plus every //! per-IP composite key), authorization codes, device records //! (plus `:cnf`/`:poll` sidecars and `user_code` mappings), and token grants //! (cascade-revoked through [`grants::revoke`] so member token keys, the //! member set, and the barrier stay correct, plus standalone access/refresh //! records and `:act` sidecars). //! //! Records without a `subject` field are never touched: `par:`, `pending:`, //! `dpop_jti:`, `client_assertion_jti:`, `rotation:`, `totp_replay:`, and the //! `{rl}` rate-limit counters carry no subject, so they cannot name a user. //! //! Like the DCR `DELETE` revocation this scans the keyspaces with `SCAN MATCH //! * COUNT 100` and filters on the raw `subject` hash field. `user //! disable` is a rare administrative call, so a bounded linear scan is //! proportionate; on Cluster topologies the scan covers the contacted node //! only (documented limit — Single and Sentinel scan fully). use std::collections::{HashMap, HashSet}; use crate::db::user_repo::normalize_email; use crate::error::AppError; use crate::state::RedisConn; use crate::store::{Store, StoreError, grants, sha256_hex}; /// How many records one subject purge removed, per key kind. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct PurgedSubject { /// Login session records deleted (each also removed its `:meta` sidecar). pub sessions: u64, /// Lockout counters deleted (the account key plus every per-IP composite). pub lockouts: u64, /// Authorization-code records deleted. pub codes: u64, /// Device-code records deleted (each also removed `:cnf`/`:poll` sidecars). pub devices: u64, /// `user_code` mappings deleted. pub user_codes: u64, /// Grant families cascade-revoked. pub grants: u64, /// Standalone access-token records deleted (each also removed `:act`). pub access: u64, /// Standalone refresh-token records deleted. pub refresh: u64, } /// Delete every Redis record whose `subject` is `email` or `user_id`. /// /// `email` must be the canonical (lowercase) address from /// [`user_repo::find_by_email`](crate::db::user_repo::find_by_email); the /// lockout key is derived from it with the same normalization the login flow /// uses. Refresh-owned families cascade through [`grants::revoke`]; the /// grant's member `:act` sidecars are removed alongside because the revoke /// script deletes member keys only. /// /// # Errors /// /// Returns a fail-closed store failure when Redis is unavailable; callers /// (notably `user disable`) must leave the Postgres `disabled` flag untouched /// on error. pub async fn purge_subject( conn: &mut RedisConn, store: &Store, email: &str, user_id: &str, ) -> Result { let mut summary = PurgedSubject::default(); let session_prefix = store.session_key_prefix(); for key in scan_keys(conn, &session_prefix).await? { if key.ends_with(":meta") { continue; } if is_subject(&read_fields(conn, &key).await?, email, user_id) { del_key(conn, &key).await?; del_key(conn, &format!("{key}:meta")).await?; summary.sessions += 1; } } let fail_key = store.login_fail_key(&sha256_hex(normalize_email(email).as_bytes())); let removed: i64 = redis::cmd("DEL") .arg(&fail_key) .query_async(conn) .await .map_err(store_mapped)?; if removed > 0 { summary.lockouts += 1; } let account_hash = sha256_hex(normalize_email(email).as_bytes()); let composite_pattern = format!("{}*:{account_hash}", store.login_fail_ip_prefix()); let mut cursor = "0".to_owned(); loop { let (next, batch): (String, Vec) = redis::cmd("SCAN") .arg(&cursor) .arg("MATCH") .arg(&composite_pattern) .arg("COUNT") .arg(100) .query_async(conn) .await .map_err(store_mapped)?; for key in &batch { del_key(conn, key).await?; summary.lockouts += 1; } if next == "0" { break; } cursor = next; } for key in scan_keys(conn, &store.code_key_prefix()).await? { if is_subject(&read_fields(conn, &key).await?, email, user_id) { del_key(conn, &key).await?; summary.codes += 1; } } let device_prefix = store.device_key_prefix(); let mut device_hashes: HashSet = HashSet::new(); for key in scan_keys(conn, &device_prefix).await? { if key.ends_with(":cnf") || key.ends_with(":poll") { continue; } if is_subject(&read_fields(conn, &key).await?, email, user_id) { del_key(conn, &key).await?; del_key(conn, &format!("{key}:cnf")).await?; del_key(conn, &format!("{key}:poll")).await?; if let Some(hash) = key.strip_prefix(&device_prefix) { device_hashes.insert(hash.to_owned()); } summary.devices += 1; } } if !device_hashes.is_empty() { for key in scan_keys(conn, &store.user_code_key_prefix()).await? { let target: Option = redis::cmd("GET") .arg(&key) .query_async(conn) .await .map_err(store_mapped)?; if target.is_some_and(|hash| device_hashes.contains(&hash)) { del_key(conn, &key).await?; summary.user_codes += 1; } } } let grant_prefix = store.grant_key_prefix(); let access_prefix = store.access_key_prefix(); let mut grant_ids: Vec = Vec::new(); for key in scan_keys(conn, &grant_prefix).await? { if key.ends_with(":tokens") { continue; } if is_subject(&read_fields(conn, &key).await?, email, user_id) && let Some(id) = key.strip_prefix(&grant_prefix) { grant_ids.push(id.to_owned()); } } for grant_id in &grant_ids { let members: Vec = redis::cmd("SMEMBERS") .arg(store.grant_tokens_key(grant_id)) .query_async(conn) .await .map_err(store_mapped)?; grants::revoke(conn, store, grant_id) .await .map_err(AppError::from)?; for member in &members { del_key(conn, &format!("{access_prefix}{member}:act")).await?; } summary.grants += 1; } for key in scan_keys(conn, &access_prefix).await? { if key.ends_with(":act") { continue; } if is_subject(&read_fields(conn, &key).await?, email, user_id) { del_key(conn, &key).await?; del_key(conn, &format!("{key}:act")).await?; summary.access += 1; } } for key in scan_keys(conn, &store.refresh_key_prefix()).await? { if is_subject(&read_fields(conn, &key).await?, email, user_id) { del_key(conn, &key).await?; summary.refresh += 1; } } Ok(summary) } /// Delete every login session (plus `:meta` sidecar) whose `subject` is /// `email` or `user_id`, except the session named by `keep_session_id`. /// /// The credential-change path (`change_password` / `confirm_totp` in /// `src/ui/login.rs`) calls this before the database update and before /// re-issuing the current session, so a password or TOTP change revokes the /// subject's other sessions while the session in hand survives. The sweep /// runs first so a Redis outage fails before any credential is written (the /// password stays unchanged); a later database failure then costs only /// re-authentication. Tokens, codes, /// and grants are deliberately untouched: ending the login sessions forces /// re-authentication, while live OAuth tokens keep working until they expire /// or the account is disabled (which runs the full [`purge_subject`]). /// /// # Errors /// /// Returns a fail-closed store failure when Redis is unavailable; callers /// must surface the error instead of reporting success with siblings intact. pub async fn purge_sibling_sessions( conn: &mut RedisConn, store: &Store, email: &str, user_id: &str, keep_session_id: &str, ) -> Result { let keep_key = store.session_key(keep_session_id); let mut revoked = 0; let session_prefix = store.session_key_prefix(); for key in scan_keys(conn, &session_prefix).await? { if key.ends_with(":meta") || key == keep_key { continue; } if is_subject(&read_fields(conn, &key).await?, email, user_id) { del_key(conn, &key).await?; del_key(conn, &format!("{key}:meta")).await?; revoked += 1; } } Ok(revoked) } /// Whether a record's `subject` field names this user by email or id. fn is_subject(fields: &HashMap, email: &str, user_id: &str) -> bool { fields .get("subject") .is_some_and(|subject| subject == email || subject == user_id) } /// List every key under `prefix` with a bounded `SCAN` loop. async fn scan_keys(conn: &mut RedisConn, prefix: &str) -> Result, AppError> { let pattern = format!("{prefix}*"); let mut cursor = "0".to_owned(); let mut out = Vec::new(); loop { let (next, batch): (String, Vec) = redis::cmd("SCAN") .arg(&cursor) .arg("MATCH") .arg(&pattern) .arg("COUNT") .arg(100) .query_async(conn) .await .map_err(store_mapped)?; out.extend(batch); if next == "0" { break; } cursor = next; } Ok(out) } /// Read one record's hash fields; a missing key reads as empty. async fn read_fields(conn: &mut RedisConn, key: &str) -> Result, AppError> { redis::cmd("HGETALL") .arg(key) .query_async(conn) .await .map_err(store_mapped) } /// Delete one key. async fn del_key(conn: &mut RedisConn, key: &str) -> Result<(), AppError> { let _: () = redis::cmd("DEL") .arg(key) .query_async(conn) .await .map_err(store_mapped)?; Ok(()) } /// Map a Redis failure onto the fail-closed store policy (transient outages /// render `503`, everything else is masked). fn store_mapped(error: redis::RedisError) -> AppError { AppError::from(StoreError::Redis(error)) } #[cfg(test)] mod tests { use std::time::Duration; use super::{PurgedSubject, purge_sibling_sessions, purge_subject}; use crate::{ config::DurabilityMode, error::{AppError, OAuthErrorCode}, state::RedisConn, store::Store, }; #[derive(Debug, thiserror::Error)] enum TestError { #[error("redis error: {0}")] Redis(#[from] redis::RedisError), #[error("store error: {0}")] Store(#[from] crate::store::StoreError), #[error("app error: {0}")] App(#[from] crate::error::AppError), #[error("unexpected test value: {0}")] Unexpected(String), } type TestResult = Result<(), TestError>; /// Shared compose Redis plus a per-test uuid prefix pair. /// /// Unique prefixes are mandatory: `purge_subject` `SCAN`s its prefix, and a /// per-process counter collides across parallel nextest processes. async fn harness() -> Result<(RedisConn, Store), TestError> { let manager = crate::test_support::redis_manager() .await .map_err(|error| TestError::Unexpected(format!("shared redis: {error}")))?; let (state_prefix, rl_prefix) = crate::test_support::unique_prefixes(); let store = Store::from_parts(state_prefix, rl_prefix, DurabilityMode::WaitAof); Ok((RedisConn::Single(manager), store)) } async fn hset(conn: &mut RedisConn, key: &str, pairs: &[(&str, &str)]) -> TestResult { let mut cmd = redis::cmd("HSET"); cmd.arg(key); for (field, value) in pairs { cmd.arg(field).arg(value); } cmd.query_async::<()>(conn).await?; redis::cmd("PEXPIRE") .arg(key) .arg(60_000) .query_async::<()>(conn) .await?; Ok(()) } async fn set(conn: &mut RedisConn, key: &str, value: &str) -> TestResult { redis::cmd("SET") .arg(key) .arg(value) .arg("PX") .arg(60_000) .query_async::<()>(conn) .await?; Ok(()) } async fn exists(conn: &mut RedisConn, key: &str) -> Result { let count: i64 = redis::cmd("EXISTS").arg(key).query_async(conn).await?; Ok(count > 0) } async fn require_gone(conn: &mut RedisConn, key: &str) -> TestResult { if exists(conn, key).await? { return Err(TestError::Unexpected(format!("key survived purge: {key}"))); } Ok(()) } async fn require_present(conn: &mut RedisConn, key: &str) -> TestResult { if !exists(conn, key).await? { return Err(TestError::Unexpected(format!("key missing: {key}"))); } Ok(()) } #[tokio::test] async fn purge_subject_deletes_all_subject_state() -> TestResult { let (mut conn, store) = harness().await?; let email = "alice@example.test"; let user_id = uuid::Uuid::new_v4().to_string(); let decoy = "bob@example.test"; let session = "sess-alice"; let session_key = store.session_key(session); hset( &mut conn, &session_key, &[ ("subject", email), ("mfa_verified", "true"), ("restricted", "false"), ], ) .await?; hset( &mut conn, &format!("{session_key}:meta"), &[ ("auth_time", "1700000000"), ("csrf", "csrf-secret"), ("request_id", "req-1"), ], ) .await?; let fail_key = store.login_fail_key(&crate::store::sha256_hex(email.as_bytes())); redis::cmd("INCR") .arg(&fail_key) .query_async::<()>(&mut conn) .await?; let code_key = store.code_key("codehashalice"); hset( &mut conn, &code_key, &[ ("client_id", "client-a"), ("redirect_uri", "https://client.example.test/cb"), ("scope", "orders:read"), ( "code_challenge", "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM", ), ("subject", email), ("grant_id", "grant-code"), ("cnf", ""), ], ) .await?; let device_hash = "devicehashalice"; let device_key = store.device_key(device_hash); hset( &mut conn, &device_key, &[ ("client_id", "client-a"), ("scope", "orders:read"), ("status", "approved"), ("subject", user_id.as_str()), ("grant_id", ""), ], ) .await?; set(&mut conn, &format!("{device_key}:cnf"), "binding").await?; set(&mut conn, &format!("{device_key}:poll"), "1700000000").await?; let user_code_hash = "usercodehashalice"; let user_code_key = store.user_code_key(user_code_hash); set(&mut conn, &user_code_key, device_hash).await?; let grant_id = "grant-alice"; let grant_key = store.grant_key(grant_id); hset( &mut conn, &grant_key, &[ ("grant_id", grant_id), ("client_id", "client-a"), ("subject", email), ("scope", "orders:read"), ], ) .await?; let access_hash = "accesshashalice"; let refresh_hash = "refreshhashalice"; let access_key = store.access_key(access_hash); let refresh_key = store.refresh_key(refresh_hash); for (key, hash) in [(&access_key, access_hash), (&refresh_key, refresh_hash)] { hset( &mut conn, key, &[ ("client_id", "client-a"), ("grant_id", grant_id), ("scope", "orders:read"), ("subject", email), ("cnf", ""), ], ) .await?; redis::cmd("SADD") .arg(store.grant_tokens_key(grant_id)) .arg(hash) .query_async::<()>(&mut conn) .await?; } set(&mut conn, &format!("{access_key}:act"), "jkt").await?; let lone_access_key = store.access_key("loneaccessalice"); hset( &mut conn, &lone_access_key, &[ ("client_id", "client-a"), ("grant_id", "orphan"), ("scope", "orders:read"), ("subject", email), ("cnf", ""), ], ) .await?; set(&mut conn, &format!("{lone_access_key}:act"), "jkt").await?; let lone_refresh_key = store.refresh_key("lonerefreshalice"); hset( &mut conn, &lone_refresh_key, &[ ("client_id", "client-a"), ("grant_id", "orphan"), ("scope", "orders:read"), ("subject", email), ("cnf", ""), ], ) .await?; let decoy_session_key = store.session_key("sess-bob"); hset( &mut conn, &decoy_session_key, &[ ("subject", decoy), ("mfa_verified", "true"), ("restricted", "false"), ], ) .await?; let decoy_access_key = store.access_key("accesshashbob"); hset( &mut conn, &decoy_access_key, &[ ("client_id", "client-b"), ("grant_id", "grant-bob"), ("scope", "orders:read"), ("subject", decoy), ("cnf", ""), ], ) .await?; let summary = purge_subject(&mut conn, &store, email, &user_id).await?; let expected = PurgedSubject { sessions: 1, lockouts: 1, codes: 1, devices: 1, user_codes: 1, grants: 1, access: 1, refresh: 1, }; if summary != expected { return Err(TestError::Unexpected(format!( "wrong purge summary: {summary:?} (expected {expected:?})" ))); } for key in [ session_key.clone(), format!("{session_key}:meta"), fail_key, code_key, device_key.clone(), format!("{device_key}:cnf"), format!("{device_key}:poll"), user_code_key, grant_key, store.grant_tokens_key(grant_id), access_key.clone(), format!("{access_key}:act"), refresh_key, lone_access_key.clone(), format!("{lone_access_key}:act"), lone_refresh_key, ] { require_gone(&mut conn, &key).await?; } require_present(&mut conn, &decoy_session_key).await?; require_present(&mut conn, &decoy_access_key).await?; Ok(()) } #[tokio::test] async fn purge_subject_with_no_matching_state_returns_zero() -> TestResult { let (mut conn, store) = harness().await?; let summary = purge_subject(&mut conn, &store, "nobody@example.test", "no-id").await?; if summary != PurgedSubject::default() { return Err(TestError::Unexpected(format!( "expected zero summary, got {summary:?}" ))); } Ok(()) } async fn dead_conn() -> Result { let client = redis::Client::open("redis://127.0.0.1:1")?; let manager = redis::aio::ConnectionManager::new_lazy_with_config( client, redis::aio::ConnectionManagerConfig::new() .set_number_of_retries(0) .set_response_timeout(Some(Duration::from_millis(200))) .set_connection_timeout(Some(Duration::from_millis(200))), )?; Ok(RedisConn::Single(manager)) } fn require_unavailable(error: AppError) -> TestResult { if error.code() != OAuthErrorCode::TemporarilyUnavailable { return Err(TestError::Unexpected(format!( "a store outage must be temporarily_unavailable, got {:?}", error.code() ))); } if error.status().as_u16() != 503 { return Err(TestError::Unexpected(format!( "a store outage must render 503, got {}", error.status() ))); } Ok(()) } #[tokio::test] async fn purge_sibling_sessions_deletes_meta_sidecar_and_keeps_current() -> TestResult { let (mut conn, store) = harness().await?; let email = "sibling-meta@example.test"; let user_id = uuid::Uuid::new_v4().to_string(); let current_key = store.session_key("sess-current"); hset( &mut conn, ¤t_key, &[("subject", email), ("mfa_verified", "true")], ) .await?; hset(&mut conn, &format!("{current_key}:meta"), &[("csrf", "a")]).await?; let sibling_key = store.session_key("sess-sibling"); hset( &mut conn, &sibling_key, &[("subject", email), ("mfa_verified", "true")], ) .await?; hset(&mut conn, &format!("{sibling_key}:meta"), &[("csrf", "b")]).await?; let revoked = purge_sibling_sessions(&mut conn, &store, email, &user_id, "sess-current").await?; if revoked != 1 { return Err(TestError::Unexpected(format!( "exactly one sibling must be revoked, got {revoked}" ))); } require_present(&mut conn, ¤t_key).await?; require_present(&mut conn, &format!("{current_key}:meta")).await?; require_gone(&mut conn, &sibling_key).await?; require_gone(&mut conn, &format!("{sibling_key}:meta")).await?; Ok(()) } #[tokio::test] async fn purge_sibling_sessions_leaves_other_subjects_tokens_and_grants() -> TestResult { let (mut conn, store) = harness().await?; let email = "sibling-scope@example.test"; let user_id = uuid::Uuid::new_v4().to_string(); let other = "stranger@example.test"; let victim_key = store.session_key("sess-victim"); hset(&mut conn, &victim_key, &[("subject", email)]).await?; let other_key = store.session_key("sess-other"); hset(&mut conn, &other_key, &[("subject", other)]).await?; hset(&mut conn, &format!("{other_key}:meta"), &[("csrf", "c")]).await?; let grant_id = "grant-survivor"; hset( &mut conn, &store.grant_key(grant_id), &[("subject", email), ("scope", "orders:read")], ) .await?; redis::cmd("SADD") .arg(store.grant_tokens_key(grant_id)) .arg("memberhash") .query_async::<()>(&mut conn) .await?; let access_key = store.access_key("survivoraccess"); hset(&mut conn, &access_key, &[("subject", email)]).await?; set(&mut conn, &format!("{access_key}:act"), "jkt").await?; let refresh_key = store.refresh_key("survivorrefresh"); hset(&mut conn, &refresh_key, &[("subject", email)]).await?; let code_key = store.code_key("survivorcode"); hset(&mut conn, &code_key, &[("subject", email)]).await?; let revoked = purge_sibling_sessions(&mut conn, &store, email, &user_id, "sess-victim").await?; if revoked != 0 { return Err(TestError::Unexpected(format!( "keeping the only own session must revoke nothing, got {revoked}" ))); } for key in [ victim_key, other_key.clone(), format!("{other_key}:meta"), store.grant_key(grant_id), store.grant_tokens_key(grant_id), access_key.clone(), format!("{access_key}:act"), refresh_key, code_key, ] { require_present(&mut conn, &key).await?; } Ok(()) } #[tokio::test] async fn purge_sibling_sessions_matches_subject_by_user_id() -> TestResult { let (mut conn, store) = harness().await?; let email = "sibling-id@example.test"; let user_id = uuid::Uuid::new_v4().to_string(); let current_key = store.session_key("sess-id-current"); hset(&mut conn, ¤t_key, &[("subject", user_id.as_str())]).await?; let sibling_key = store.session_key("sess-id-sibling"); hset(&mut conn, &sibling_key, &[("subject", user_id.as_str())]).await?; let revoked = purge_sibling_sessions(&mut conn, &store, email, &user_id, "sess-id-current").await?; if revoked != 1 { return Err(TestError::Unexpected(format!( "a user-id-keyed sibling must be revoked, got {revoked}" ))); } require_present(&mut conn, ¤t_key).await?; require_gone(&mut conn, &sibling_key).await?; Ok(()) } #[tokio::test] async fn purge_subject_fails_closed_on_store_outage() -> TestResult { let mut conn = dead_conn().await?; let store = Store::from_parts( "purge-outage:{state}:".to_owned(), "oauth:{rl}:".to_owned(), DurabilityMode::WaitAof, ); match purge_subject(&mut conn, &store, "outage@example.test", "outage-id").await { Err(error) => require_unavailable(error), Ok(_) => Err(TestError::Unexpected( "purge_subject on a dead store must fail, never report success".to_owned(), )), } } #[tokio::test] async fn purge_sibling_sessions_fails_closed_on_store_outage() -> TestResult { let mut conn = dead_conn().await?; let store = Store::from_parts( "purge-outage:{state}:".to_owned(), "oauth:{rl}:".to_owned(), DurabilityMode::WaitAof, ); match purge_sibling_sessions( &mut conn, &store, "outage@example.test", "outage-id", "sess-keep", ) .await { Err(error) => require_unavailable(error), Ok(_) => Err(TestError::Unexpected( "purge_sibling_sessions on a dead store must fail, never report success".to_owned(), )), } } }