diff --git a/Cargo.lock b/Cargo.lock index 22d8705a0..c033bcd24 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1808,6 +1808,7 @@ dependencies = [ "rustls", "serde", "serde_json", + "sha2 0.10.9", "syslog-tracing", "time", "tokio", diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index 00814e1a2..716cf07a7 100755 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -37,6 +37,7 @@ bcrypt = { workspace = true } unicode-properties = "0.1.4" base64 = { workspace = true } time = { workspace = true } +sha2 = { workspace = true } rand = { workspace = true } axum-server = { workspace = true } rustls = { workspace = true } diff --git a/crates/server/src/console/pages/auth_pages.rs b/crates/server/src/console/pages/auth_pages.rs index 22bba844f..7021544d6 100755 --- a/crates/server/src/console/pages/auth_pages.rs +++ b/crates/server/src/console/pages/auth_pages.rs @@ -9,11 +9,12 @@ use axum::Form; use axum::extract::{ConnectInfo, State}; use axum::http::{HeaderMap, StatusCode}; use axum::response::{Html, IntoResponse, Response}; -use extenddb_storage::management_store::RateLimitStore; +use extenddb_storage::management_store::{PasswordHashStore, RateLimitStore}; use serde::Deserialize; use crate::console::ConsoleState; use crate::console::html; +use crate::console::session::CredentialStamp; use crate::management::CallerIdentity; use crate::management::auth::log_password_check_error; use crate::rate_limit::{self, LoginAttempt, LoginRefusal}; @@ -47,15 +48,16 @@ pub async fn login_submit( // Try admin auth first, then IAM user auth. A storage error (including a // timed-out wait for a bcrypt slot) is answered 500 and recorded as - // nothing: no password was checked. + // nothing: no password was checked. A match carries the stamp of the + // password hash it was checked against, and the session is bound to it. let checked = check_credentials(&form.username, &form.password, &*state.catalog_store).await; - let identity = + let (identity, stamp) = match settle_attempt(attempt, checked, &*state.catalog_store, &state.listen_url).await { - Ok(identity) => identity, + Ok(login) => login, Err(page) => return page, }; - let (token, _csrf) = state.sessions.create(identity).await; + let (token, _csrf) = state.sessions.create(identity, stamp).await; let cookie = session_cookie(&token, 28800, state.secure_cookies); ( @@ -126,7 +128,7 @@ IAM users: enter account_id/user_name. /// Result of an admin login attempt — three-state to distinguish "not found" /// (fall through to IAM) from "wrong password" (reject immediately). enum AdminLoginResult { - Authenticated(CallerIdentity), + Authenticated(CallerIdentity, CredentialStamp), WrongPassword, NotFound, } @@ -166,16 +168,18 @@ fn refusal_page(refusal: LoginRefusal, listen_url: &str) -> Response { } } -/// Check the form's credentials, admin first and then IAM user. `Ok(None)` -/// is a wrong password or an unknown principal; `Err(())` is a storage -/// error, including a timed-out wait for a bcrypt slot. +/// Check the form's credentials, admin first and then IAM user. A match +/// returns the identity with the stamp of the password hash it was checked +/// against. `Ok(None)` is a wrong password or an unknown principal; +/// `Err(())` is a storage error, including a timed-out wait for a bcrypt +/// slot. async fn check_credentials( username: &str, password: &str, store: &dyn extenddb_storage::CatalogStore, -) -> Result, ()> { +) -> Result, ()> { match try_admin_login(username, password, store).await? { - AdminLoginResult::Authenticated(id) => Ok(Some(id)), + AdminLoginResult::Authenticated(id, stamp) => Ok(Some((id, stamp))), // Admin exists, wrong password: do NOT fall through. AdminLoginResult::WrongPassword => Ok(None), // Admin not found: try IAM user auth. @@ -189,12 +193,12 @@ async fn check_credentials( /// the attempt releases its reservation. async fn settle_attempt( attempt: LoginAttempt, - checked: Result, ()>, + checked: Result, ()>, rate_limiter: &dyn RateLimitStore, listen_url: &str, -) -> Result { +) -> Result<(CallerIdentity, CredentialStamp), Response> { match checked { - Ok(Some(identity)) => Ok(identity), + Ok(Some(login)) => Ok(login), Ok(None) => { attempt.record_failure(rate_limiter).await; Err(invalid_credentials(listen_url)) @@ -207,16 +211,20 @@ async fn settle_attempt( async fn try_admin_login( username: &str, password: &str, - store: &dyn extenddb_storage::management_store::AdminStore, + store: &dyn PasswordHashStore, ) -> Result { - let result = store - .verify_admin_password(username, password) + let hash = store + .get_admin_password_hash(username) .await .map_err(|e| log_password_check_error("Console login: DB error during admin auth", &e))?; - Ok(match result { - Some(true) => AdminLoginResult::Authenticated(CallerIdentity::Admin(username.to_owned())), - Some(false) => AdminLoginResult::WrongPassword, - None => AdminLoginResult::NotFound, + let Some(hash) = hash else { + return Ok(AdminLoginResult::NotFound); + }; + Ok(match verify_login(password, hash).await? { + Some(stamp) => { + AdminLoginResult::Authenticated(CallerIdentity::Admin(username.to_owned()), stamp) + } + None => AdminLoginResult::WrongPassword, }) } @@ -225,34 +233,90 @@ async fn try_admin_login( async fn try_iam_login( username: &str, password: &str, - store: &dyn extenddb_storage::management_store::ManagementStore, -) -> Result, ()> { + store: &dyn PasswordHashStore, +) -> Result, ()> { let Some((acct_id, uname)) = username.split_once('/') else { return Ok(None); }; - let ok = store - .verify_iam_user_password(acct_id, uname, password) + // `None` when the user does not exist or has no console password. + let hash = store + .get_user_password_hash(acct_id, uname) .await .map_err(|e| { log_password_check_error("Console login: DB error during IAM user auth", &e) })?; - Ok(ok.then(|| CallerIdentity::IamUser { + let Some(hash) = hash else { + return Ok(None); + }; + let Some(stamp) = verify_login(password, hash).await? else { + return Ok(None); + }; + let identity = CallerIdentity::IamUser { account_id: acct_id.to_owned(), user_name: uname.to_owned(), - })) + }; + Ok(Some((identity, stamp))) +} + +/// Verify a login password against the hash read from the catalog. On a +/// match, returns the stamp of that same hash, which the new session is bound +/// to. The hash is read once and used for both, so a password change that +/// lands during the login either fails the login or ends the new session at +/// its first request. +/// +/// All console login verification goes through this function, and through +/// the bcrypt slots of [`extenddb_storage::password::verify_password`]. +/// `Ok(None)` is a wrong password. `Err(())` is a verification that did not +/// run (no bcrypt slot within the wait, or a failed blocking task); the +/// caller answers 500 and records nothing. +async fn verify_login(password: &str, hash: String) -> Result, ()> { + let stamp = CredentialStamp::of(&hash); + let matched = extenddb_storage::password::verify_password(password.to_owned(), hash) + .await + .map_err(|e| log_password_check_error("Console login: password verification failed", &e))?; + Ok(matched.then_some(stamp)) } #[cfg(test)] mod tests { + use extenddb_storage::management_store::{OpError, OpResult}; use extenddb_storage::password::VERIFY_OVERLOADED; + use futures::future::BoxFuture; use super::*; - use crate::management::auth::tests::FailingAdmins; use crate::rate_limit::PRINCIPAL_LIMITED; use crate::rate_limit::tests::{MemoryStore, capture_logs, peer}; const LISTEN_URL: &str = "https://127.0.0.1:8443"; + /// A password hash store whose reads fail with `OpError::Internal` + /// carrying `message`. Logins read nothing else from it. + struct FailingHashes { + message: &'static str, + } + + impl PasswordHashStore for FailingHashes { + fn get_admin_password_hash(&self, _: &str) -> BoxFuture<'_, OpResult>> { + let error = OpError::Internal(self.message.to_owned()); + Box::pin(async move { Err(error) }) + } + + fn get_user_password_hash( + &self, + _: &str, + _: &str, + ) -> BoxFuture<'_, OpResult>> { + let error = OpError::Internal(self.message.to_owned()); + Box::pin(async move { Err(error) }) + } + } + + /// A password made at run time. The store fails before any check, so its + /// value does not matter, and no literal reaches the password check. + fn any_password() -> String { + format!("pw-{}", std::process::id()) + } + async fn body_text(response: Response) -> String { let bytes = axum::body::to_bytes(response.into_body(), usize::MAX) .await @@ -278,11 +342,12 @@ mod tests { #[tokio::test] async fn a_store_error_in_the_admin_check_is_an_error_not_an_unknown_admin() { - let admins = FailingAdmins { + let hashes = FailingHashes { message: "store is down", }; + let password = any_password(); let (logs, _guard) = capture_logs(); - assert!(try_admin_login("admin", "pw", &admins).await.is_err()); + assert!(try_admin_login("admin", &password, &hashes).await.is_err()); let text = logs.text(); assert!( text.contains("ERROR") && text.contains("Console login: DB error during admin auth"), @@ -292,11 +357,12 @@ mod tests { #[tokio::test] async fn a_bcrypt_slot_timeout_is_an_error_and_is_not_logged_as_a_store_error() { - let admins = FailingAdmins { + let hashes = FailingHashes { message: VERIFY_OVERLOADED, }; + let password = any_password(); let (logs, _guard) = capture_logs(); - assert!(try_admin_login("admin", "pw", &admins).await.is_err()); + assert!(try_admin_login("admin", &password, &hashes).await.is_err()); let text = logs.text(); assert!(!text.contains("DB error"), "{text}"); } @@ -310,9 +376,9 @@ mod tests { let attempt = rate_limit::begin_login(&attempts, "admin", peer("198.51.100.90")) .await .unwrap(); - let page = settle_attempt(attempt, Err(()), &attempts, LISTEN_URL) - .await - .unwrap_err(); + let Err(page) = settle_attempt(attempt, Err(()), &attempts, LISTEN_URL).await else { + panic!("a storage error was settled as a login"); + }; assert_eq!(page.status(), StatusCode::INTERNAL_SERVER_ERROR); assert_eq!(attempts.row_count(), 0); } @@ -323,9 +389,9 @@ mod tests { let attempt = rate_limit::begin_login(&attempts, "admin", peer("198.51.100.91")) .await .unwrap(); - let page = settle_attempt(attempt, Ok(None), &attempts, LISTEN_URL) - .await - .unwrap_err(); + let Err(page) = settle_attempt(attempt, Ok(None), &attempts, LISTEN_URL).await else { + panic!("a wrong password was settled as a login"); + }; assert_eq!(page.status(), StatusCode::OK); assert!(body_text(page).await.contains("Invalid credentials")); assert_eq!(attempts.rows_for("admin", "198.51.100.91"), 1); diff --git a/crates/server/src/console/pages/mod.rs b/crates/server/src/console/pages/mod.rs index a6aee44cd..0db74123a 100755 --- a/crates/server/src/console/pages/mod.rs +++ b/crates/server/src/console/pages/mod.rs @@ -4,7 +4,8 @@ //! Console page handlers. //! //! Each handler queries the catalog DB directly and renders server-side HTML. -//! Authentication is via session cookies validated against the in-memory store. +//! Authentication is via session cookies validated against the in-memory store +//! and, on every request, against the principal's password hash in the catalog. //! All state-changing POST forms include a CSRF token validated on submission. mod access_key_pages; @@ -42,11 +43,12 @@ pub use user_pages::{create_user, delete_user, new_user_form, user_detail}; use std::sync::Arc; -use axum::http::HeaderMap; +use axum::http::{HeaderMap, StatusCode}; use axum::response::{IntoResponse, Redirect, Response}; use crate::console::ConsoleState; use crate::console::html; +use crate::console::session::SessionCheck; use crate::management::CallerIdentity; /// Session data returned by `require_session`: identity + CSRF token. @@ -55,24 +57,39 @@ pub struct SessionData { pub csrf_token: String, } -/// Extract session identity and CSRF token from the cookie header. Returns -/// the session data or a redirect to the login page. +/// Extract the session from the cookie header and check it against the +/// catalog. Returns the session data; a redirect to the login page when there +/// is no session, it expired, or its principal was deleted or had its +/// password changed; or 503 when the catalog cannot be read. #[allow(clippy::result_large_err)] async fn require_session( headers: &HeaderMap, state: &Arc, ) -> Result { - let token = extract_session_token(headers); - let result = match token { - Some(t) => state.sessions.get(&t).await, - None => None, + let Some(token) = extract_session_token(headers) else { + return Err(Redirect::to("/console/login").into_response()); }; - match result { - Some((identity, csrf_token)) => Ok(SessionData { + session_response(state.sessions.validate(&token, &*state.catalog_store).await) +} + +/// The session data for a session that passed its check, or the response a +/// console request gets when it did not. +#[allow(clippy::result_large_err)] +fn session_response(check: SessionCheck) -> Result { + match check { + SessionCheck::Valid { + identity, + csrf_token, + } => Ok(SessionData { identity, csrf_token, }), - None => Err(Redirect::to("/console/login").into_response()), + SessionCheck::Invalid => Err(Redirect::to("/console/login").into_response()), + SessionCheck::Unavailable => Err(( + StatusCode::SERVICE_UNAVAILABLE, + "The catalog is unavailable. Try again.", + ) + .into_response()), } } @@ -168,3 +185,43 @@ fn op_error_message(e: extenddb_storage::management_store::OpError) -> String { OpError::Database { .. } => "Database error".to_owned(), } } + +#[cfg(test)] +mod tests { + use axum::http::header::LOCATION; + + use super::*; + + #[test] + fn a_passed_check_returns_the_session() { + let Ok(session) = session_response(SessionCheck::Valid { + identity: CallerIdentity::Admin("alice".to_owned()), + csrf_token: "csrf".to_owned(), + }) else { + panic!("a session that passed its check was refused"); + }; + assert!(matches!(session.identity, CallerIdentity::Admin(ref n) if n == "alice")); + assert_eq!(session.csrf_token, "csrf"); + } + + #[test] + fn an_ended_session_gets_the_login_redirect() { + let Err(response) = session_response(SessionCheck::Invalid) else { + panic!("an ended session was accepted"); + }; + assert_eq!(response.status(), StatusCode::SEE_OTHER); + assert_eq!( + response.headers().get(LOCATION).map(|v| v.as_bytes()), + Some(&b"/console/login"[..]) + ); + } + + #[test] + fn an_unreadable_catalog_gets_503() { + let Err(response) = session_response(SessionCheck::Unavailable) else { + panic!("a request was accepted while the catalog could not be read"); + }; + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + assert!(response.headers().get(LOCATION).is_none()); + } +} diff --git a/crates/server/src/console/session.rs b/crates/server/src/console/session.rs index a6aa11ae5..cc71ff2b2 100755 --- a/crates/server/src/console/session.rs +++ b/crates/server/src/console/session.rs @@ -6,11 +6,21 @@ //! Sessions are random 32-byte hex tokens stored in a cookie. They map to a //! `CallerIdentity` (admin or IAM user) and a CSRF token for form protection. //! Sessions expire after 8 hours of inactivity. A background reaper is not -//! needed — expired sessions are pruned lazily on lookup. +//! needed: expired sessions are pruned lazily on lookup. +//! +//! Each session also holds a [`CredentialStamp`] of the password hash its +//! principal logged in with. [`SessionStore::validate`] reads the principal's +//! current hash from the catalog on every request, and ends the session when +//! the principal no longer exists or its hash has changed. Every server reads +//! the same catalog, so deleting a principal or changing its password through +//! any server ends its sessions on all of them, each at the next request it +//! serves for that session. use std::collections::HashMap; use std::time::{Duration, Instant}; +use extenddb_storage::management_store::{OpResult, PasswordHashStore}; +use sha2::{Digest, Sha256}; use tokio::sync::Mutex; use crate::management::CallerIdentity; @@ -21,12 +31,46 @@ const SESSION_TTL: Duration = Duration::from_secs(8 * 3600); /// Maximum sessions before lazy pruning triggers on insert. const MAX_SESSIONS: usize = 10_000; +/// SHA-256 digest of the password hash a session was created against. +/// +/// bcrypt draws a new salt for every hash, so each password change stores a +/// different hash, even when the new password equals the old one. The +/// session keeps this digest rather than the hash itself. +#[derive(Clone, PartialEq, Eq)] +pub struct CredentialStamp([u8; 32]); + +impl CredentialStamp { + /// The stamp of a stored password hash. + #[must_use] + pub fn of(password_hash: &str) -> Self { + Self(Sha256::digest(password_hash.as_bytes()).into()) + } +} + struct SessionEntry { identity: CallerIdentity, csrf_token: String, + stamp: CredentialStamp, last_active: Instant, } +/// Result of [`SessionStore::validate`]. +pub enum SessionCheck { + /// The session is live and its principal still has the password hash the + /// session was created against. + Valid { + identity: CallerIdentity, + csrf_token: String, + }, + /// There is no such session, it expired, or its principal was deleted or + /// had its password changed. In the last two cases the session has been + /// removed from the store. + Invalid, + /// The catalog could not be read. The session is kept, and the request + /// must be refused. + Unavailable, +} + /// Thread-safe in-memory session store. /// /// Uses `tokio::sync::Mutex` because this store is accessed from async @@ -34,6 +78,9 @@ struct SessionEntry { /// points and poisons on panic. pub struct SessionStore { sessions: Mutex>, + /// How long a session lasts without activity: [`SESSION_TTL`], except in + /// tests. + ttl: Duration, } impl Default for SessionStore { @@ -48,39 +95,97 @@ impl SessionStore { pub fn new() -> Self { Self { sessions: Mutex::new(HashMap::new()), + ttl: SESSION_TTL, } } - /// Create a session for the given identity. Returns `(session_token, csrf_token)`. - pub async fn create(&self, identity: CallerIdentity) -> (String, String) { + /// Create an empty session store whose sessions last `ttl` without + /// activity. + #[cfg(test)] + fn with_ttl(ttl: Duration) -> Self { + Self { + sessions: Mutex::new(HashMap::new()), + ttl, + } + } + + /// Create a session for the given identity, bound to the stamp of the + /// password hash the login verified. Returns `(session_token, csrf_token)`. + pub async fn create( + &self, + identity: CallerIdentity, + stamp: CredentialStamp, + ) -> (String, String) { let token = generate_token(); let csrf = generate_token(); let entry = SessionEntry { identity, csrf_token: csrf.clone(), + stamp, last_active: Instant::now(), }; let mut map = self.sessions.lock().await; // Lazy prune if too many sessions. if map.len() >= MAX_SESSIONS { let now = Instant::now(); - map.retain(|_, e| now.duration_since(e.last_active) < SESSION_TTL); + let ttl = self.ttl; + map.retain(|_, e| now.duration_since(e.last_active) < ttl); } map.insert(token.clone(), entry); (token, csrf) } - /// Look up a session by token. Returns `(identity, csrf_token)` if valid, - /// refreshing the last-active timestamp. Returns `None` if expired or not found. - pub async fn get(&self, token: &str) -> Option<(CallerIdentity, String)> { + /// Check a session token for a console request. + /// + /// Looks the session up, then reads the principal's current password hash + /// from `catalog`. Only a request that passes the check refreshes the + /// session's last-active time, so requests refused while the catalog + /// cannot be read do not keep an idle session alive. The store lock is + /// not held during the read. + pub async fn validate(&self, token: &str, catalog: &dyn PasswordHashStore) -> SessionCheck { + let Some((identity, csrf_token, stamp)) = self.lookup(token).await else { + return SessionCheck::Invalid; + }; + match current_password_hash(catalog, &identity).await { + Ok(Some(hash)) if CredentialStamp::of(&hash) == stamp => { + self.touch(token).await; + SessionCheck::Valid { + identity, + csrf_token, + } + } + Ok(_) => { + self.remove(token).await; + SessionCheck::Invalid + } + Err(e) => { + tracing::warn!("console session check could not read the catalog: {e:?}"); + SessionCheck::Unavailable + } + } + } + + /// Look up a session by token. Returns `(identity, csrf_token, stamp)` if + /// it has not expired. Does not refresh the last-active timestamp. + async fn lookup(&self, token: &str) -> Option<(CallerIdentity, String, CredentialStamp)> { let mut map = self.sessions.lock().await; - let entry = map.get_mut(token)?; - if entry.last_active.elapsed() > SESSION_TTL { + let entry = map.get(token)?; + if entry.last_active.elapsed() > self.ttl { map.remove(token); return None; } - entry.last_active = Instant::now(); - Some((entry.identity.clone(), entry.csrf_token.clone())) + Some(( + entry.identity.clone(), + entry.csrf_token.clone(), + entry.stamp.clone(), + )) + } + + /// Record activity on a session that passed its check. + async fn touch(&self, token: &str) { + if let Some(entry) = self.sessions.lock().await.get_mut(token) { + entry.last_active = Instant::now(); + } } /// Remove a session (logout). @@ -90,6 +195,21 @@ impl SessionStore { } } +/// The principal's stored password hash: `None` if the principal does not +/// exist or has no console password. +async fn current_password_hash( + catalog: &dyn PasswordHashStore, + identity: &CallerIdentity, +) -> OpResult> { + match identity { + CallerIdentity::Admin(name) => catalog.get_admin_password_hash(name).await, + CallerIdentity::IamUser { + account_id, + user_name, + } => catalog.get_user_password_hash(account_id, user_name).await, + } +} + /// Generate a cryptographically random 32-byte hex token. fn generate_token() -> String { use rand::Rng; @@ -106,3 +226,7 @@ fn hex_encode(bytes: &[u8]) -> String { } s } + +#[cfg(test)] +#[path = "session_tests.rs"] +mod tests; diff --git a/crates/server/src/console/session_tests.rs b/crates/server/src/console/session_tests.rs new file mode 100644 index 000000000..2dce1eddc --- /dev/null +++ b/crates/server/src/console/session_tests.rs @@ -0,0 +1,281 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Tests for console session validation against the catalog. + +use std::collections::HashMap; +use std::sync::Mutex as StdMutex; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::{Duration, Instant}; + +use extenddb_storage::management_store::{OpError, OpResult, PasswordHashStore}; +use futures::future::BoxFuture; + +use super::{CredentialStamp, SESSION_TTL, SessionCheck, SessionStore}; +use crate::management::CallerIdentity; + +const ACCOUNT: &str = "111111111111"; + +/// Catalog holding password hashes for admins and IAM users. With `failing` +/// set, every read returns an error. +#[derive(Default)] +struct FakeCatalog { + admins: StdMutex>, + users: StdMutex>, + failing: AtomicBool, +} + +impl FakeCatalog { + fn set_admin(&self, name: &str, hash: &str) { + self.admins + .lock() + .unwrap() + .insert(name.to_owned(), hash.to_owned()); + } + + fn delete_admin(&self, name: &str) { + self.admins.lock().unwrap().remove(name); + } + + fn set_user(&self, user: &str, hash: &str) { + self.users + .lock() + .unwrap() + .insert((ACCOUNT.to_owned(), user.to_owned()), hash.to_owned()); + } + + fn delete_user(&self, user: &str) { + self.users + .lock() + .unwrap() + .remove(&(ACCOUNT.to_owned(), user.to_owned())); + } + + fn answer(&self, hash: Option) -> OpResult> { + if self.failing.load(Ordering::SeqCst) { + Err(OpError::Internal("Database error".to_owned())) + } else { + Ok(hash) + } + } +} + +impl PasswordHashStore for FakeCatalog { + fn get_admin_password_hash(&self, admin_name: &str) -> BoxFuture<'_, OpResult>> { + let hash = self.admins.lock().unwrap().get(admin_name).cloned(); + let answer = self.answer(hash); + Box::pin(async move { answer }) + } + + fn get_user_password_hash( + &self, + account_id: &str, + user_name: &str, + ) -> BoxFuture<'_, OpResult>> { + let key = (account_id.to_owned(), user_name.to_owned()); + let hash = self.users.lock().unwrap().get(&key).cloned(); + let answer = self.answer(hash); + Box::pin(async move { answer }) + } +} + +/// Log an admin in: record its hash in the catalog and create a session +/// bound to that hash. +async fn admin_session( + store: &SessionStore, + catalog: &FakeCatalog, + name: &str, + hash: &str, +) -> String { + catalog.set_admin(name, hash); + let identity = CallerIdentity::Admin(name.to_owned()); + store.create(identity, CredentialStamp::of(hash)).await.0 +} + +async fn user_session( + store: &SessionStore, + catalog: &FakeCatalog, + user: &str, + hash: &str, +) -> String { + catalog.set_user(user, hash); + let identity = CallerIdentity::IamUser { + account_id: ACCOUNT.to_owned(), + user_name: user.to_owned(), + }; + store.create(identity, CredentialStamp::of(hash)).await.0 +} + +async fn admin_name_of(store: &SessionStore, catalog: &FakeCatalog, token: &str) -> Option { + match store.validate(token, catalog).await { + SessionCheck::Valid { + identity: CallerIdentity::Admin(name), + .. + } => Some(name), + _ => None, + } +} + +async fn is_invalid(store: &SessionStore, catalog: &FakeCatalog, token: &str) -> bool { + matches!(store.validate(token, catalog).await, SessionCheck::Invalid) +} + +async fn is_valid(store: &SessionStore, catalog: &FakeCatalog, token: &str) -> bool { + matches!( + store.validate(token, catalog).await, + SessionCheck::Valid { .. } + ) +} + +#[tokio::test] +async fn deleted_admin_session_is_refused_and_other_admins_keep_theirs() { + let store = SessionStore::new(); + let catalog = FakeCatalog::default(); + let alice = admin_session(&store, &catalog, "alice", "hash-alice").await; + let bob = admin_session(&store, &catalog, "bob", "hash-bob").await; + assert_eq!( + admin_name_of(&store, &catalog, &alice).await.as_deref(), + Some("alice") + ); + + catalog.delete_admin("alice"); + + assert!(is_invalid(&store, &catalog, &alice).await); + assert_eq!( + admin_name_of(&store, &catalog, &bob).await.as_deref(), + Some("bob") + ); +} + +#[tokio::test] +async fn ended_session_is_removed_from_the_store() { + let store = SessionStore::new(); + let catalog = FakeCatalog::default(); + let alice = admin_session(&store, &catalog, "alice", "hash-alice").await; + + catalog.delete_admin("alice"); + assert!(is_invalid(&store, &catalog, &alice).await); + + // Restoring the same hash does not bring the session back. + catalog.set_admin("alice", "hash-alice"); + assert!(is_invalid(&store, &catalog, &alice).await); +} + +#[tokio::test] +async fn admin_password_change_ends_the_old_session_only() { + let store = SessionStore::new(); + let catalog = FakeCatalog::default(); + let old = admin_session(&store, &catalog, "alice", "hash-1").await; + + catalog.set_admin("alice", "hash-2"); + assert!(is_invalid(&store, &catalog, &old).await); + + let new = store + .create( + CallerIdentity::Admin("alice".to_owned()), + CredentialStamp::of("hash-2"), + ) + .await + .0; + assert!(is_valid(&store, &catalog, &new).await); +} + +#[tokio::test] +async fn iam_user_session_ends_on_password_change_and_delete() { + let store = SessionStore::new(); + let catalog = FakeCatalog::default(); + let first = user_session(&store, &catalog, "carol", "hash-1").await; + assert!(is_valid(&store, &catalog, &first).await); + + catalog.set_user("carol", "hash-2"); + assert!(is_invalid(&store, &catalog, &first).await); + + let second = user_session(&store, &catalog, "carol", "hash-2").await; + assert!(is_valid(&store, &catalog, &second).await); + catalog.delete_user("carol"); + assert!(is_invalid(&store, &catalog, &second).await); +} + +#[tokio::test] +async fn catalog_error_refuses_the_request_and_keeps_the_session() { + let store = SessionStore::new(); + let catalog = FakeCatalog::default(); + let alice = admin_session(&store, &catalog, "alice", "hash-alice").await; + + catalog.failing.store(true, Ordering::SeqCst); + assert!(matches!( + store.validate(&alice, &catalog).await, + SessionCheck::Unavailable + )); + + catalog.failing.store(false, Ordering::SeqCst); + assert!(is_valid(&store, &catalog, &alice).await); +} + +/// Move a session's last-active time `by` into the past. +async fn backdate(store: &SessionStore, token: &str, by: Duration) -> Instant { + let then = Instant::now() + .checked_sub(by) + .expect("the monotonic clock is past the backdate"); + store + .sessions + .lock() + .await + .get_mut(token) + .expect("session exists") + .last_active = then; + then +} + +async fn last_active(store: &SessionStore, token: &str) -> Instant { + store.sessions.lock().await[token].last_active +} + +#[tokio::test] +async fn only_a_request_that_passes_the_check_counts_as_activity() { + let store = SessionStore::new(); + let catalog = FakeCatalog::default(); + let alice = admin_session(&store, &catalog, "alice", "hash-alice").await; + let then = backdate(&store, &alice, Duration::from_secs(60)).await; + + catalog.failing.store(true, Ordering::SeqCst); + assert!(matches!( + store.validate(&alice, &catalog).await, + SessionCheck::Unavailable + )); + assert_eq!(last_active(&store, &alice).await, then); + + catalog.failing.store(false, Ordering::SeqCst); + assert!(is_valid(&store, &catalog, &alice).await); + assert!(last_active(&store, &alice).await > then + Duration::from_secs(59)); +} + +#[test] +fn sessions_last_eight_hours_without_activity() { + assert_eq!(SESSION_TTL, Duration::from_secs(8 * 3600)); + assert_eq!(SessionStore::new().ttl, SESSION_TTL); +} + +#[tokio::test] +async fn a_session_idle_longer_than_the_ttl_ends_and_stays_ended() { + let store = SessionStore::with_ttl(Duration::from_secs(60)); + let catalog = FakeCatalog::default(); + let alice = admin_session(&store, &catalog, "alice", "hash-alice").await; + let bob = admin_session(&store, &catalog, "bob", "hash-bob").await; + backdate(&store, &alice, Duration::from_secs(61)).await; + backdate(&store, &bob, Duration::from_secs(50)).await; + + assert!(is_invalid(&store, &catalog, &alice).await); + assert!(!store.sessions.lock().await.contains_key(&alice)); + // A later request with the same token does not bring the session back. + assert!(is_invalid(&store, &catalog, &alice).await); + + assert!(is_valid(&store, &catalog, &bob).await); +} + +#[tokio::test] +async fn unknown_token_is_invalid() { + let store = SessionStore::new(); + let catalog = FakeCatalog::default(); + assert!(is_invalid(&store, &catalog, "no-such-token").await); +} diff --git a/crates/storage-cassandra/src/admin_store.rs b/crates/storage-cassandra/src/admin_store.rs index a7e2d599a..2189233b9 100755 --- a/crates/storage-cassandra/src/admin_store.rs +++ b/crates/storage-cassandra/src/admin_store.rs @@ -117,8 +117,9 @@ impl extenddb_storage::management_store::AdminStore for CassandraCatalogStore { return Err(OpError::NotFound("Admin user not found".to_owned())); } + // Quorum, so the console's quorum session check sees the deletion. let query = format!("DELETE FROM {catalog_keyspace}.admin_users WHERE admin_name = ?"); - crate::cassandra_util::execute( + crate::cassandra_util::execute_quorum( &session, &query, cdrs_tokio::query_values!(admin_name.as_str()), @@ -142,10 +143,11 @@ impl extenddb_storage::management_store::AdminStore for CassandraCatalogStore { return Err(OpError::NotFound("Admin user not found".to_owned())); } + // Quorum, so the console's quorum session check sees the new hash. let query = format!( "UPDATE {catalog_keyspace}.admin_users SET password_hash = ? WHERE admin_name = ?" ); - crate::cassandra_util::execute( + crate::cassandra_util::execute_quorum( &session, &query, cdrs_tokio::query_values!(password_hash.as_str(), admin_name.as_str()), diff --git a/crates/storage-cassandra/src/cassandra_util.rs b/crates/storage-cassandra/src/cassandra_util.rs index a4ebf5c0c..22c70a02a 100644 --- a/crates/storage-cassandra/src/cassandra_util.rs +++ b/crates/storage-cassandra/src/cassandra_util.rs @@ -355,8 +355,6 @@ pub async fn apply_lwt( values: QueryValues, context: &str, ) -> Result { - use cdrs_tokio::types::IntoRustByName; - let result = session .query_with_values(query, values) .await @@ -364,6 +362,48 @@ pub async fn apply_lwt( tracing::error!("{context} lwt failed: {e}"); E::db_error("Database error".to_owned()) })?; + lwt_applied_flag(&result, context) +} + +/// [`apply_lwt`] with the commit phase at `LOCAL_QUORUM`. +/// +/// [`apply_lwt`] commits at the driver's default consistency, `ONE`, so a +/// `LOCAL_QUORUM` read issued after it returns can still miss the write. Use +/// this where a later quorum read must see the change. The serial phase keeps +/// the default, `SERIAL`, which every other LWT on these rows also uses. +/// +/// # Errors +/// Returns an error if Cassandra does not acknowledge the statement at quorum +/// or the response cannot be parsed. +pub async fn apply_lwt_quorum( + session: &Arc, + query: &str, + values: QueryValues, + context: &str, +) -> Result { + use cdrs_tokio::consistency::Consistency; + use cdrs_tokio::statement::StatementParamsBuilder; + + let params = StatementParamsBuilder::new() + .with_consistency(Consistency::LocalQuorum) + .with_values(values) + .build(); + let result = session + .query_with_params(query, params) + .await + .map_err(|e| { + tracing::error!("{context} quorum lwt failed: {e}"); + E::db_error("Database error".to_owned()) + })?; + lwt_applied_flag(&result, context) +} + +/// The `[applied]` flag of an LWT response, `false` when the response has no row. +fn lwt_applied_flag( + result: &cdrs_tokio::frame::Envelope, + context: &str, +) -> Result { + use cdrs_tokio::types::IntoRustByName; let body = result.response_body().map_err(|e| { tracing::error!("{context} lwt response_body failed: {e}"); diff --git a/crates/storage-cassandra/src/lib.rs b/crates/storage-cassandra/src/lib.rs index f374e77cd..3f0489402 100644 --- a/crates/storage-cassandra/src/lib.rs +++ b/crates/storage-cassandra/src/lib.rs @@ -30,6 +30,7 @@ mod management_store; mod metadata_engine; pub mod migrations; pub mod operations; +mod password_hash_store; pub(crate) mod propagation_hold; mod snapshot_engine; mod stream_engine; diff --git a/crates/storage-cassandra/src/management_store/users.rs b/crates/storage-cassandra/src/management_store/users.rs index d076aef40..ee55cab6d 100644 --- a/crates/storage-cassandra/src/management_store/users.rs +++ b/crates/storage-cassandra/src/management_store/users.rs @@ -197,8 +197,9 @@ impl CassandraCatalogStore { "delete_user iam_user_tags", ) .await?; - // iam_users last: user is only gone once all dependent rows are cleaned up - crate::cassandra_util::apply_lwt( + // iam_users last: user is only gone once all dependent rows are cleaned up. + // Quorum commit, so the console's quorum session check sees the deletion. + crate::cassandra_util::apply_lwt_quorum( &session, &format!("DELETE FROM {ks}.iam_users WHERE account_id = ? AND user_name = ? IF EXISTS"), cdrs_tokio::query_values!(account_id, user_name), @@ -421,7 +422,8 @@ impl CassandraCatalogStore { WHERE account_id = ? AND user_name = ? IF EXISTS" ); - crate::cassandra_util::apply_lwt( + // Quorum commit, so the console's quorum session check sees the new hash. + crate::cassandra_util::apply_lwt_quorum( self.session(), &update_query, cdrs_tokio::query_values!(password_hash, account_id, user_name), diff --git a/crates/storage-cassandra/src/password_hash_store.rs b/crates/storage-cassandra/src/password_hash_store.rs new file mode 100644 index 000000000..7038b2617 --- /dev/null +++ b/crates/storage-cassandra/src/password_hash_store.rs @@ -0,0 +1,74 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! `PasswordHashStore` implementation for `CassandraCatalogStore`. +//! +//! Both reads run at `LOCAL_QUORUM`. The writes that change or remove these +//! hashes (`delete_admin`, `change_admin_password`, `delete_user`, +//! `change_user_password`) also commit at `LOCAL_QUORUM`, so a read issued +//! after one of them returns sees its result in the same datacenter. + +use cdrs_tokio::types::IntoRustByName; +use cdrs_tokio::types::rows::Row; +use extenddb_storage::management_store::{OpError, OpResult, PasswordHashStore}; +use futures::future::BoxFuture; + +use crate::catalog_store::CassandraCatalogStore; + +impl PasswordHashStore for CassandraCatalogStore { + fn get_admin_password_hash(&self, admin_name: &str) -> BoxFuture<'_, OpResult>> { + let admin_name = admin_name.to_owned(); + let session = self.session().clone(); + let catalog_keyspace = self.catalog_keyspace(); + Box::pin(async move { + let query = format!( + "SELECT password_hash FROM {catalog_keyspace}.admin_users WHERE admin_name = ?" + ); + let rows = crate::cassandra_util::query_rows_quorum( + &session, + &query, + cdrs_tokio::query_values!(admin_name.as_str()), + "get_admin_password_hash", + ) + .await?; + password_hash_of(rows, "get_admin_password_hash") + }) + } + + fn get_user_password_hash( + &self, + account_id: &str, + user_name: &str, + ) -> BoxFuture<'_, OpResult>> { + let account_id = account_id.to_owned(); + let user_name = user_name.to_owned(); + let session = self.session().clone(); + let catalog_keyspace = self.catalog_keyspace(); + Box::pin(async move { + let query = format!( + "SELECT password_hash FROM {catalog_keyspace}.iam_users \ + WHERE account_id = ? AND user_name = ?" + ); + let rows = crate::cassandra_util::query_rows_quorum( + &session, + &query, + cdrs_tokio::query_values!(account_id.as_str(), user_name.as_str()), + "get_user_password_hash", + ) + .await?; + password_hash_of(rows, "get_user_password_hash") + }) + } +} + +/// The `password_hash` column of the first row: `None` when there is no row +/// or the column is null. +fn password_hash_of(rows: Vec, context: &str) -> OpResult> { + let Some(row) = rows.into_iter().next() else { + return Ok(None); + }; + row.get_by_name("password_hash").map_err(|e| { + tracing::error!("{context} parse password_hash: {e}"); + OpError::Internal("Database error".to_owned()) + }) +} diff --git a/crates/storage-mongodb/src/lib.rs b/crates/storage-mongodb/src/lib.rs index 67a9b2e7b..3580b5fb5 100644 --- a/crates/storage-mongodb/src/lib.rs +++ b/crates/storage-mongodb/src/lib.rs @@ -21,6 +21,7 @@ mod management_store; mod metadata_engine; pub mod migrations; mod operations; +mod password_hash_store; pub mod pushdown; mod snapshot_engine; mod stream_engine; diff --git a/crates/storage-mongodb/src/password_hash_store.rs b/crates/storage-mongodb/src/password_hash_store.rs new file mode 100644 index 000000000..be35a9504 --- /dev/null +++ b/crates/storage-mongodb/src/password_hash_store.rs @@ -0,0 +1,50 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! `PasswordHashStore` implementation for `MongoCatalogStore`. + +use extenddb_storage::management_store::{OpError, OpResult, PasswordHashStore}; +use futures::future::BoxFuture; +use mongodb::bson::{Document, doc}; + +use crate::catalog_store::MongoCatalogStore; + +impl PasswordHashStore for MongoCatalogStore { + fn get_admin_password_hash(&self, admin_name: &str) -> BoxFuture<'_, OpResult>> { + let filter = doc! { "_id": admin_name }; + Box::pin(async move { + find_password_hash(self, "admin_users", filter, "get_admin_password_hash").await + }) + } + + fn get_user_password_hash( + &self, + account_id: &str, + user_name: &str, + ) -> BoxFuture<'_, OpResult>> { + let filter = doc! { "account_id": account_id, "user_name": user_name }; + Box::pin(async move { + find_password_hash(self, "iam_users", filter, "get_user_password_hash").await + }) + } +} + +/// The string `password_hash` field of the document matching `filter`: +/// `None` when no document matches or the field is absent or null. +async fn find_password_hash( + store: &MongoCatalogStore, + collection: &str, + filter: Document, + context: &str, +) -> OpResult> { + let doc = store + .catalog_db() + .collection::(collection) + .find_one(filter) + .await + .map_err(|e| { + tracing::error!("{context}: {e}"); + OpError::Internal("Database error".to_owned()) + })?; + Ok(doc.and_then(|d| d.get_str("password_hash").ok().map(str::to_owned))) +} diff --git a/crates/storage-postgres/src/lib.rs b/crates/storage-postgres/src/lib.rs index 2e6d81bcb..7d0f9d28c 100755 --- a/crates/storage-postgres/src/lib.rs +++ b/crates/storage-postgres/src/lib.rs @@ -24,6 +24,7 @@ mod management_store; mod metadata_engine; pub mod migrations; mod operations; +mod password_hash_store; mod pg_util; mod snapshot_engine; #[doc(hidden)] diff --git a/crates/storage-postgres/src/password_hash_store.rs b/crates/storage-postgres/src/password_hash_store.rs new file mode 100644 index 000000000..8c54fa58f --- /dev/null +++ b/crates/storage-postgres/src/password_hash_store.rs @@ -0,0 +1,50 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! `PasswordHashStore` implementation for `PostgresCatalogStore`. + +use extenddb_storage::management_store::{OpError, OpResult, PasswordHashStore}; +use futures::future::BoxFuture; + +use crate::catalog_store::PostgresCatalogStore; + +impl PasswordHashStore for PostgresCatalogStore { + fn get_admin_password_hash(&self, admin_name: &str) -> BoxFuture<'_, OpResult>> { + let admin_name = admin_name.to_owned(); + Box::pin(async move { + let row: Option<(String,)> = + sqlx::query_as("SELECT password_hash FROM admin_users WHERE admin_name = $1") + .bind(&admin_name) + .fetch_optional(self.pool()) + .await + .map_err(|e| { + tracing::error!("get_admin_password_hash: {e}"); + OpError::Internal("Database error".to_owned()) + })?; + Ok(row.map(|(hash,)| hash)) + }) + } + + fn get_user_password_hash( + &self, + account_id: &str, + user_name: &str, + ) -> BoxFuture<'_, OpResult>> { + let account_id = account_id.to_owned(); + let user_name = user_name.to_owned(); + Box::pin(async move { + let row: Option<(Option,)> = sqlx::query_as( + "SELECT password_hash FROM iam_users WHERE account_id = $1 AND user_name = $2", + ) + .bind(&account_id) + .bind(&user_name) + .fetch_optional(self.pool()) + .await + .map_err(|e| { + tracing::error!("get_user_password_hash: {e}"); + OpError::Internal("Database error".to_owned()) + })?; + Ok(row.and_then(|(hash,)| hash)) + }) + } +} diff --git a/crates/storage-sqlite/src/lib.rs b/crates/storage-sqlite/src/lib.rs index 5b65e8f8e..2e1401039 100644 --- a/crates/storage-sqlite/src/lib.rs +++ b/crates/storage-sqlite/src/lib.rs @@ -37,6 +37,7 @@ mod metadata; pub mod migrations; mod number_key; mod operations; +mod password_hash_store; mod schema; mod serve_lock; mod snapshot_engine; diff --git a/crates/storage-sqlite/src/password_hash_store.rs b/crates/storage-sqlite/src/password_hash_store.rs new file mode 100644 index 000000000..16ce6759f --- /dev/null +++ b/crates/storage-sqlite/src/password_hash_store.rs @@ -0,0 +1,123 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! `PasswordHashStore` implementation for `SqliteCatalogStore`. + +use extenddb_storage::management_store::{OpError, OpResult, PasswordHashStore}; +use futures::future::BoxFuture; + +use crate::catalog_store::SqliteCatalogStore; + +impl PasswordHashStore for SqliteCatalogStore { + fn get_admin_password_hash(&self, admin_name: &str) -> BoxFuture<'_, OpResult>> { + let admin_name = admin_name.to_owned(); + Box::pin(async move { + let row: Option<(String,)> = + sqlx::query_as("SELECT password_hash FROM admin_users WHERE admin_name = ?") + .bind(&admin_name) + .fetch_optional(self.pool()) + .await + .map_err(|e| { + tracing::error!("get_admin_password_hash: {e}"); + OpError::Internal("Database error".to_owned()) + })?; + Ok(row.map(|(hash,)| hash)) + }) + } + + fn get_user_password_hash( + &self, + account_id: &str, + user_name: &str, + ) -> BoxFuture<'_, OpResult>> { + let account_id = account_id.to_owned(); + let user_name = user_name.to_owned(); + Box::pin(async move { + let row: Option<(Option,)> = sqlx::query_as( + "SELECT password_hash FROM iam_users WHERE account_id = ? AND user_name = ?", + ) + .bind(&account_id) + .bind(&user_name) + .fetch_optional(self.pool()) + .await + .map_err(|e| { + tracing::error!("get_user_password_hash: {e}"); + OpError::Internal("Database error".to_owned()) + })?; + Ok(row.and_then(|(hash,)| hash)) + }) + } +} + +#[cfg(test)] +mod tests { + use extenddb_storage::management_store::{AdminStore, ManagementStore, PasswordHashStore}; + use sqlx::sqlite::SqlitePoolOptions; + + use crate::catalog_store::SqliteCatalogStore; + + const ACCOUNT: &str = "111111111111"; + + async fn store() -> SqliteCatalogStore { + // One connection keeps the in-memory database alive for the test. + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect("sqlite::memory:") + .await + .expect("open in-memory sqlite"); + sqlx::query("PRAGMA foreign_keys = ON") + .execute(&pool) + .await + .unwrap(); + crate::migrations::apply_for_tests(&pool).await; + SqliteCatalogStore::new(pool) + } + + #[tokio::test] + async fn admin_hash_follows_password_change_and_delete() { + let s = store().await; + assert_eq!(s.get_admin_password_hash("ops").await.unwrap(), None); + s.create_admin("ops", "hash-1").await.unwrap(); + let hash = s.get_admin_password_hash("ops").await.unwrap(); + assert_eq!(hash.as_deref(), Some("hash-1")); + s.change_admin_password("ops", "hash-2").await.unwrap(); + let hash = s.get_admin_password_hash("ops").await.unwrap(); + assert_eq!(hash.as_deref(), Some("hash-2")); + s.delete_admin("ops").await.unwrap(); + assert_eq!(s.get_admin_password_hash("ops").await.unwrap(), None); + } + + #[tokio::test] + async fn user_hash_follows_password_change_and_delete() { + let s = store().await; + s.create_account(ACCOUNT, "acct").await.unwrap(); + s.create_user(ACCOUNT, "bob", None).await.unwrap(); + // A user without a console password has no hash. + assert_eq!( + s.get_user_password_hash(ACCOUNT, "bob").await.unwrap(), + None + ); + s.change_user_password(ACCOUNT, "bob", "hash-1") + .await + .unwrap(); + let hash = s.get_user_password_hash(ACCOUNT, "bob").await.unwrap(); + assert_eq!(hash.as_deref(), Some("hash-1")); + s.delete_user(ACCOUNT, "bob").await.unwrap(); + assert_eq!( + s.get_user_password_hash(ACCOUNT, "bob").await.unwrap(), + None + ); + } + + #[tokio::test] + async fn user_hash_is_gone_after_account_delete() { + let s = store().await; + s.create_account(ACCOUNT, "acct").await.unwrap(); + s.create_user(ACCOUNT, "bob", Some("hash-1")).await.unwrap(); + s.delete_account(ACCOUNT).await.unwrap(); + assert_eq!( + s.get_user_password_hash(ACCOUNT, "bob").await.unwrap(), + None + ); + } +} diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs index 38ad349e1..7e1036021 100755 --- a/crates/storage/src/lib.rs +++ b/crates/storage/src/lib.rs @@ -754,6 +754,7 @@ impl StorageEngine for T where pub trait CatalogStore: management_store::ManagementStore + management_store::AdminStore + + management_store::PasswordHashStore + management_store::SettingsStore + management_store::MetricsStore + management_store::RateLimitStore diff --git a/crates/storage/src/management_store/mod.rs b/crates/storage/src/management_store/mod.rs index a9feb55a3..e2574fc83 100755 --- a/crates/storage/src/management_store/mod.rs +++ b/crates/storage/src/management_store/mod.rs @@ -145,6 +145,29 @@ pub trait AdminStore: Send + Sync { ) -> BoxFuture<'_, OpResult>>; } +// ── Password hashes ──────────────────────────────────────────────────── + +/// Reads the stored password hash of a console principal. +/// +/// The console reads the hash once at login, verifies the password against +/// it, and binds the session to that hash. Every later console request reads +/// the hash again and ends the session when the principal is gone or its hash +/// differs. Implementations must return the latest acknowledged value: the +/// read must see any password change or deletion that has already returned +/// to its caller, on any server sharing the catalog. +pub trait PasswordHashStore: Send + Sync { + /// The stored password hash of an admin, or `None` if the admin does not exist. + fn get_admin_password_hash(&self, admin_name: &str) -> BoxFuture<'_, OpResult>>; + + /// The stored console password hash of an IAM user, or `None` if the user + /// does not exist or has no console password. + fn get_user_password_hash( + &self, + account_id: &str, + user_name: &str, + ) -> BoxFuture<'_, OpResult>>; +} + // ── Management store (IAM) ───────────────────────────────────────────── /// IAM entity CRUD: accounts, users, groups, roles, policies, access keys. diff --git a/docs/getting-started.md b/docs/getting-started.md index d74557338..c1c27db98 100755 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -1281,7 +1281,7 @@ Navigate to `https://127.0.0.1:18443/console/` in your browser (adjust the host Admin users have full access to all management operations. IAM users can view their own details and manage their own access keys. -Sessions expire after 8 hours of inactivity. Click "Logout" to end a session immediately. +Sessions expire after 8 hours of inactivity. Click "Logout" to end a session immediately. Deleting a user or changing its password ends that user's sessions at their next request. ## 9. Audit logging diff --git a/docs/manuals/01-architecture-guide.md b/docs/manuals/01-architecture-guide.md index 492b8470b..36ccd41c7 100755 --- a/docs/manuals/01-architecture-guide.md +++ b/docs/manuals/01-architecture-guide.md @@ -247,7 +247,7 @@ TLS is supported via rustls. `extenddb init` generates a self-signed certificate - CSRF tokens on all state-changing POST handlers - HttpOnly session cookies - Security headers: X-Content-Type-Options, X-Frame-Options, Referrer-Policy -- 8-hour session expiry +- Sessions expire after 8 hours of inactivity ### Input Validation @@ -273,12 +273,13 @@ The management web console is a server-rendered HTML interface mounted at `/cons ### Architecture -The console uses the `ManagementStore` trait for all data access — it never queries the database directly. Write operations route through shared functions in `management::ops`, ensuring validation and business logic are defined once. This means every action available in the console is also available through the management REST API. +The console reads all data through four of the traits that `CatalogStore` combines (`ManagementStore`, `PasswordHashStore`, `SettingsStore` and `RateLimitStore`) and never queries the database directly. Write operations route through shared functions in `management::ops`, ensuring validation and business logic are defined once. This means every action available in the console is also available through the management REST API. ### Authentication Model -- Session-based authentication with 8-hour expiry +- Session-based authentication; sessions expire after 8 hours of inactivity - Sessions stored in-memory with random tokens in HttpOnly cookies (`SameSite=Strict`, `Path=/console`) +- Each request reads the principal's password hash from the catalog and ends the session if the principal was deleted or its password changed - CSRF tokens on all state-changing POST handlers - Security headers: X-Content-Type-Options (nosniff), X-Frame-Options (DENY), Referrer-Policy (strict-origin-when-cross-origin) - Login rate limiting per user and source address, shared with the management API (see the security model manual) diff --git a/docs/manuals/05-admin-guide.md b/docs/manuals/05-admin-guide.md index 6637cbfaa..aa2b9b10b 100755 --- a/docs/manuals/05-admin-guide.md +++ b/docs/manuals/05-admin-guide.md @@ -829,7 +829,7 @@ The management web console is served at `/console/` on the same port as the Dyna - Admin users: enter username and password - IAM users: enter `account_id/user_name` as username, console password as password -Sessions expire after 8 hours. Click "Logout" to end immediately. +Sessions expire after 8 hours of inactivity. Click "Logout" to end immediately. Deleting an admin or IAM user, or changing its password, ends its console sessions on every server at their next request. ## Monitoring diff --git a/docs/manuals/10-security-model.md b/docs/manuals/10-security-model.md index 18e8409d2..544830cc5 100755 --- a/docs/manuals/10-security-model.md +++ b/docs/manuals/10-security-model.md @@ -227,7 +227,8 @@ An admin session can read and change every account. Every page and form under `/ The management web console (`/console/*`) implements: - **CSRF protection**: Unique tokens generated per session, injected into all forms via JavaScript, validated on every POST handler -- **Session management**: Server-side sessions with 8-hour expiry, HttpOnly and `Secure` cookies with `SameSite=Strict` and `Path=/console` (a dev-mode build serves plain HTTP and leaves `Secure` off) +- **Session management**: Server-side sessions that expire after 8 hours of inactivity, HttpOnly and `Secure` cookies with `SameSite=Strict` and `Path=/console` (a dev-mode build serves plain HTTP and leaves `Secure` off) +- **Session revocation**: A session is bound to the password hash its admin or IAM user logged in with. Every console request reads that hash from the catalog again. When the principal has been deleted, or its password has been changed, the session is removed and the request is redirected to the login page. This happens at the session's next request on every server that shares the catalog, because no server keeps a copy of the hash between requests. On Cassandra the read and the writes that change the hash use `LOCAL_QUORUM`, so a server in another datacenter sees the change once it has replicated there. If the catalog cannot be read, the request gets HTTP 503 and the session is kept. A refused request does not count as activity, so the 8 hour idle timeout keeps running during a catalog outage. - **Security headers**: X-Content-Type-Options (nosniff), X-Frame-Options (DENY), Referrer-Policy (strict-origin-when-cross-origin) - **HSTS**: Sent automatically (TLS is always enabled) - **Login rate limiting**: Failed login attempts are counted per user and source address, and per source address, as described under [Login Rate Limiting](#login-rate-limiting). diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index f22a2c4d1..81fbbc7b3 100755 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -848,9 +848,15 @@ The error log line names the exact relations for one leaked table and gives this ### Console redirects to login page on every request -**Cause:** The session cookie has expired (sessions last 8 hours) or cookies are blocked by the browser. +**Cause:** The session cookie has expired (sessions last 8 hours of inactivity), the user's password was changed or the user was deleted after the session began, or cookies are blocked by the browser. -**Fix:** Log in again. Ensure cookies are enabled for the extenddb server address. If using a reverse proxy, ensure it forwards the `Cookie` and `Set-Cookie` headers. +**Fix:** Log in again, with the new password if it was changed. Ensure cookies are enabled for the extenddb server address. If using a reverse proxy, ensure it forwards the `Cookie` and `Set-Cookie` headers. + +### Console pages return HTTP 503 "The catalog is unavailable" + +**Cause:** The console checks every session against the catalog database, and that read failed. The session is kept. + +**Fix:** Check that the catalog database is reachable from the server, then reload the page. After a PostgreSQL restart the first reloads can still answer 503: each pooled catalog connection that the restart closed fails once and is then replaced. ### Console shows "Encryption key not found" when creating access keys diff --git a/tests/test_console_session_revocation.py b/tests/test_console_session_revocation.py new file mode 100644 index 000000000..96bba24f5 --- /dev/null +++ b/tests/test_console_session_revocation.py @@ -0,0 +1,223 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""Console sessions end when their principal is deleted or its password changes. + +A console session is bound to the password hash its principal logged in with. +Every console request reads the principal's hash from the catalog again, so a +session stops working at its next request after the principal is deleted or +its password is changed, on any server that shares the catalog. + +Prerequisites match tests/test_console_integration.py: + - extenddb running with `auth.provider = "builtin"` on EXTENDDB_TEST_ENDPOINT + - Admin credentials in EXTENDDB_ADMIN_USER / EXTENDDB_ADMIN_PASSWORD + +The cross-server test also needs EXTENDDB_TEST_PEER_ENDPOINT, a second server +on the same catalog. It is skipped when that variable is unset. +""" + +from __future__ import annotations + +import os +import uuid + +import pytest +import requests + +from management_helpers import ManagementClient + + +def _require_auth_env() -> tuple[str, str, str]: + endpoint = os.environ.get("EXTENDDB_TEST_ENDPOINT", "").strip() + admin_user = os.environ.get("EXTENDDB_ADMIN_USER", "").strip() + admin_pass = os.environ.get("EXTENDDB_ADMIN_PASSWORD", "").strip() + if not endpoint or not admin_user or not admin_pass: + pytest.fail( + "MISCONFIGURED: console session tests require EXTENDDB_TEST_ENDPOINT, " + "EXTENDDB_ADMIN_USER, and EXTENDDB_ADMIN_PASSWORD. " + "These must be set by devtools/run-tests before test execution." + ) + return endpoint, admin_user, admin_pass + + +def _unique(prefix: str) -> str: + return f"{prefix}-{uuid.uuid4().hex[:10]}" + + +def _password() -> str: + return uuid.uuid4().hex + + +class Console: + """One browser: a cookie jar talking to one server's console.""" + + def __init__(self, base_url: str) -> None: + self.base_url = base_url.rstrip("/") + self.http = requests.Session() + # Self-signed certs from ``extenddb init``. Passed on each request: + # a session-level ``verify = False`` loses to REQUESTS_CA_BUNDLE. + self.verify = not self.base_url.startswith("https://") + + def login(self, username: str, password: str) -> requests.Response: + return self.http.post( + f"{self.base_url}/console/login", + data={"username": username, "password": password}, + allow_redirects=False, + timeout=30, + verify=self.verify, + ) + + def dashboard(self) -> requests.Response: + return self.http.get( + f"{self.base_url}/console", + allow_redirects=False, + timeout=30, + verify=self.verify, + ) + + +def logged_in(base_url: str, username: str, password: str) -> Console: + console = Console(base_url) + resp = console.login(username, password) + assert resp.status_code == 303, f"login as {username}: {resp.status_code} {resp.text[:200]}" + assert_reaches_console(console) + return console + + +def assert_reaches_console(console: Console) -> None: + resp = console.dashboard() + assert resp.status_code == 200, f"expected the dashboard, got {resp.status_code}" + assert "Dashboard" in resp.text + + +def assert_sent_to_login(console: Console) -> None: + resp = console.dashboard() + assert resp.status_code == 303, f"expected the login redirect, got {resp.status_code}" + assert resp.headers.get("location") == "/console/login" + + +class Admins: + """Creates admins through the management API and deletes leftovers.""" + + def __init__(self, mgmt: ManagementClient) -> None: + self.mgmt = mgmt + self.created: list[str] = [] + + def _request(self, method: str, path: str, body: dict | None = None) -> requests.Response: + return requests.request( + method, + f"{self.mgmt.base_url}/admins{path}", + json=body, + auth=self.mgmt.admin_auth, + timeout=self.mgmt.timeout, + verify=self.mgmt.verify, + ) + + def create(self) -> tuple[str, str]: + name, password = _unique("sess-rev"), _password() + resp = self._request("POST", "", {"admin_name": name, "password": password}) + assert resp.status_code == 201, resp.text + self.created.append(name) + return name, password + + def delete(self, name: str) -> None: + resp = self._request("DELETE", f"/{name}") + assert resp.status_code == 204, resp.text + + def change_password(self, name: str, password: str) -> None: + resp = self._request("PUT", f"/{name}/password", {"password": password}) + assert resp.status_code == 204, resp.text + + def cleanup(self) -> None: + for name in self.created: + self._request("DELETE", f"/{name}") + + +@pytest.fixture(scope="module") +def auth_env() -> tuple[str, str, str]: + return _require_auth_env() + + +@pytest.fixture(scope="module") +def endpoint(auth_env) -> str: + return auth_env[0] + + +@pytest.fixture(scope="module") +def mgmt(auth_env) -> ManagementClient: + return ManagementClient(auth_env[0], auth_env[1], auth_env[2]) + + +@pytest.fixture() +def admins(mgmt): + registry = Admins(mgmt) + yield registry + registry.cleanup() + + +class TestConsoleSessionRevocation: + def test_deleted_admin_is_sent_to_login_and_other_admin_keeps_access( + self, endpoint, admins + ): + alice, alice_pw = admins.create() + bob, bob_pw = admins.create() + alice_console = logged_in(endpoint, alice, alice_pw) + bob_console = logged_in(endpoint, bob, bob_pw) + + admins.delete(alice) + + assert_sent_to_login(alice_console) + assert_reaches_console(bob_console) + + def test_admin_password_change_ends_session_and_new_password_logs_in( + self, endpoint, admins + ): + name, old_pw = admins.create() + old_console = logged_in(endpoint, name, old_pw) + + new_pw = _password() + admins.change_password(name, new_pw) + + assert_sent_to_login(old_console) + logged_in(endpoint, name, new_pw) + + def test_iam_user_session_ends_on_password_change_and_delete(self, endpoint, mgmt): + account_id = f"{uuid.uuid4().int % 10**12:012d}" + user, first_pw = _unique("sess-rev"), _password() + resp = mgmt.create_account(account_id, _unique("sess-rev")) + assert resp.status_code == 201, resp.text + try: + resp = mgmt.create_user(account_id, user, first_pw) + assert resp.status_code == 201, resp.text + first_console = logged_in(endpoint, f"{account_id}/{user}", first_pw) + + second_pw = _password() + resp = requests.put( + f"{mgmt.base_url}/accounts/{account_id}/users/{user}/password", + json={"password": second_pw}, + auth=mgmt.admin_auth, + timeout=mgmt.timeout, + verify=mgmt.verify, + ) + assert resp.status_code == 204, resp.text + assert_sent_to_login(first_console) + + second_console = logged_in(endpoint, f"{account_id}/{user}", second_pw) + resp = mgmt.delete_user(account_id, user) + assert resp.status_code == 204, resp.text + assert_sent_to_login(second_console) + finally: + mgmt.delete_user(account_id, user) + mgmt.delete_account(account_id) + + def test_deletion_through_one_server_ends_session_on_another(self, admins): + peer = os.environ.get("EXTENDDB_TEST_PEER_ENDPOINT", "").strip() + if not peer: + pytest.skip("EXTENDDB_TEST_PEER_ENDPOINT is not set") + name, password = admins.create() + peer_console = logged_in(peer, name, password) + + # Deleted through EXTENDDB_TEST_ENDPOINT, checked on the peer at once. + admins.delete(name) + + assert_sent_to_login(peer_console)