diff --git a/src/handlers/http/modal/query/querier_rbac.rs b/src/handlers/http/modal/query/querier_rbac.rs index 4bcc5def2..3bd829827 100644 --- a/src/handlers/http/modal/query/querier_rbac.rs +++ b/src/handlers/http/modal/query/querier_rbac.rs @@ -32,7 +32,7 @@ use crate::{ parseable::{DEFAULT_TENANT, PARSEABLE}, rbac::{ Users, - map::{roles, users, write_user_groups}, + map::{read_user_groups, roles, users, write_user_groups}, user::{self, UserType}, }, tenants::TENANT_METADATA, @@ -128,27 +128,7 @@ pub async fn delete_user( metadata.users.retain(|user| user.userid() != userid); // also delete from user groups - let user_groups = Users.get_user_groups(&userid, &tenant_id); - let mut groups_to_update = Vec::new(); - - for user_group in user_groups { - if let Some(groups) = write_user_groups().get_mut(tenant) - && let Some(ug) = groups.get_mut(&user_group) - && let Some(users) = users().get(tenant) - && let Some(user) = users.get(&userid) - { - let userid = match &user.ty { - UserType::Native(basic) => basic.username.clone(), - UserType::OAuth(oauth) => oauth.userid.clone(), - UserType::ApiKey(api_key) => api_key.userid.clone(), - }; - ug.remove_users_by_user_ids(HashSet::from_iter([userid]))?; - groups_to_update.push(ug.clone()); - } else { - // User not found, skip or log as needed - continue; - } - } + let groups_to_update = remove_user_from_groups(&userid, &tenant_id)?; // For each updated group, replace in place if found; otherwise push for updated_group in &groups_to_update { @@ -163,6 +143,7 @@ pub async fn delete_user( } } put_metadata(&metadata, &tenant_id).await?; + apply_user_removal_to_groups(&userid, &tenant_id, &groups_to_update); if let Err(e) = sync_user_deletion_with_ingestors(&req, &userid, &tenant_id, &caller_userid).await @@ -175,6 +156,57 @@ pub async fn delete_user( Ok(HttpResponse::Ok().json(format!("deleted user: {username}"))) } +/// Returns copies of the user's groups with the member removed, without storing +/// them. Caller holds `UPDATE_LOCK` and persists them before applying the removal. +pub(crate) fn remove_user_from_groups( + userid: &str, + tenant_id: &Option, +) -> Result, RBACError> { + let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); + let Some(member_id) = Users + .get_user(userid, tenant_id) + .map(|user| match &user.ty { + UserType::Native(basic) => basic.username.clone(), + UserType::OAuth(oauth) => oauth.userid.clone(), + UserType::ApiKey(api_key) => api_key.userid.clone(), + }) + else { + return Ok(Vec::new()); + }; + let mut groups_to_update = Vec::new(); + + for user_group in Users.get_user_groups(userid, tenant_id) { + // Removing members refreshes sessions, which comes first in the lock + // order (see `rbac::map`): update a copy without holding USER_GROUPS. + let Some(mut ug) = read_user_groups() + .get(tenant) + .and_then(|groups| groups.get(&user_group)) + .cloned() + else { + continue; + }; + ug.remove_users_by_user_ids(HashSet::from_iter([member_id.clone()]))?; + groups_to_update.push(ug); + } + Ok(groups_to_update) +} + +pub(crate) fn apply_user_removal_to_groups( + userid: &str, + tenant_id: &Option, + groups_to_update: &[user::UserGroup], +) { + let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); + if let Some(groups) = write_user_groups().get_mut(tenant) { + for updated_group in groups_to_update { + if let Some(stored) = groups.get_mut(&updated_group.name) { + // Preserve any other membership or role changes since the copy was made. + stored.users.retain(|member| member.userid() != userid); + } + } + } +} + // Handler PATCH /user/{userid}/role/add => Add roles to a user pub async fn add_roles_to_user( req: HttpRequest, diff --git a/src/handlers/http/oidc.rs b/src/handlers/http/oidc.rs index 134fd145e..617c4b4dc 100644 --- a/src/handlers/http/oidc.rs +++ b/src/handlers/http/oidc.rs @@ -36,7 +36,7 @@ static COOKIE_REQUIRE_CROSS_SITE: AtomicBool = AtomicBool::new(false); pub fn set_cookie_cross_site(enabled: bool) { COOKIE_REQUIRE_CROSS_SITE.store(enabled, Ordering::Relaxed); } -use chrono::TimeDelta; +use chrono::{TimeDelta, Utc}; use openid::Bearer; use serde::Deserialize; use tokio::sync::RwLock; @@ -124,29 +124,20 @@ pub async fn login( let tenant_id = get_tenant_id_from_key(&session_key); match session_key { // We can exchange basic auth for session cookie - SessionKey::BasicAuth { username, password } => match Users.get_user(&username, &tenant_id) - { - Some( - ref user @ User { - ty: UserType::Native(ref basic), - .. - }, - ) if basic.verify_password(&password) => { - let user_cookie = cookie_username(&username); - let user_id_cookie = cookie_userid(&username); - let session_cookie = exchange_basic_for_cookie( - user, - SessionKey::BasicAuth { username, password }, - EXPIRY_DURATION, - ); - - Ok(redirect_to_client( - query.redirect.as_str(), - [user_cookie, user_id_cookie, session_cookie], - )) + SessionKey::BasicAuth { username, password } => { + match exchange_basic_for_cookie(&username, &password, &tenant_id, EXPIRY_DURATION) { + Some(session_cookie) => { + let user_cookie = cookie_username(&username); + let user_id_cookie = cookie_userid(&username); + + Ok(redirect_to_client( + query.redirect.as_str(), + [user_cookie, user_id_cookie, session_cookie], + )) + } + None => Err(OIDCError::BadRequest("Bad Request".to_string())), } - _ => Err(OIDCError::BadRequest("Bad Request".to_string())), - }, + } // if it's a valid active session, just redirect back key @ SessionKey::SessionId(_) => { let resp = if Users.session_exists(&key) { @@ -510,15 +501,28 @@ fn callback_base_url(req: &HttpRequest) -> String { format!("{}://{host}/", PARSEABLE.options.get_scheme()) } -fn exchange_basic_for_cookie( - user: &User, - key: SessionKey, +/// Re-verifies the Basic credentials and swaps their session for a cookie one. +/// `new_basic_session` re-reads the user under SESSIONS, so an admin change +/// racing the login cannot leave a cookie session with stale credentials. +pub(crate) fn exchange_basic_for_cookie( + username: &str, + password: &str, + tenant_id: &Option, expires_in: TimeDelta, -) -> Cookie<'static> { +) -> Option> { let id = Ulid::new(); - Users.remove_session(&key); - Users.new_session(user, SessionKey::SessionId(id), expires_in); - cookie_session(id) + let mut sessions = Users.new_basic_session( + username, + password, + tenant_id.as_deref().unwrap_or(DEFAULT_TENANT), + SessionKey::SessionId(id), + Utc::now() + expires_in, + )?; + sessions.remove_session(&SessionKey::BasicAuth { + username: username.to_owned(), + password: password.to_owned(), + }); + Some(cookie_session(id)) } fn redirect_to_oidc_logout(mut logout_endpoint: Url, redirect: &Url) -> HttpResponse { diff --git a/src/rbac/map.rs b/src/rbac/map.rs index c1617c058..bbb2a3e20 100644 --- a/src/rbac/map.rs +++ b/src/rbac/map.rs @@ -40,6 +40,12 @@ use serde::{Deserialize, Serialize}; pub type Roles = HashMap>; +// Lock order: SESSIONS -> USERS -> USER_GROUPS -> ROLES -> DEFAULT_ROLE. +// Never acquire a map while holding one later in this order, and never take a +// second guard on a map this thread already holds (even two reads): these +// locks are task-fair, so a queued writer blocks every new reader. +// `Sessions::check_auth` reads USERS, USER_GROUPS and ROLES, so its caller +// must hold no guard other than the SESSIONS one it is called on. pub static USERS: OnceCell> = OnceCell::new(); pub static ROLES: OnceCell> = OnceCell::new(); pub static DEFAULT_ROLE: Lazy>>> = diff --git a/src/rbac/mod.rs b/src/rbac/mod.rs index 95ad9a52d..bc7f0d592 100644 --- a/src/rbac/mod.rs +++ b/src/rbac/mod.rs @@ -27,7 +27,8 @@ use actix_web::dev::ServiceRequest; use actix_web::http::header::{HeaderName, HeaderValue}; use chrono::{DateTime, Duration, TimeDelta, Utc}; use itertools::Itertools; -use rayon::iter::{IntoParallelRefIterator, ParallelBridge, ParallelIterator}; +use parking_lot::RwLockWriteGuard; +use rayon::iter::{IntoParallelRefIterator, ParallelIterator}; use serde::Serialize; use url::Url; @@ -46,6 +47,10 @@ use self::user::UserType; pub const EXPIRY_DURATION: Duration = Duration::hours(1); +// A Basic login re-verifies when the user changes mid-login; bound the +// retries so a stream of password changes cannot spin it. +const BASIC_LOGIN_ATTEMPTS: usize = 3; + #[derive(PartialEq, Debug)] pub enum Response { Authorized, @@ -79,7 +84,10 @@ impl Users { pub fn put_user(&self, user: User) { let tenant_id = user.tenant.as_deref().unwrap_or(DEFAULT_TENANT); - mut_sessions().remove_user(user.userid(), tenant_id); + // Holding SESSIONS across both steps keeps a concurrent Basic login + // (`new_basic_session`) from tracking a session for the user being replaced. + let mut sessions = mut_sessions(); + sessions.remove_user(user.userid(), tenant_id); mut_users().insert(user); } @@ -168,6 +176,8 @@ impl Users { // caller ensures that this operation is valid for the user pub fn change_password_hash(&self, userid: &str, hash: &String, tenant_id: &Option) { let tenant_id = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); + // SESSIONS before USERS; see the lock order in `map`. + let mut sessions = mut_sessions(); if let Some(users) = mut_users().get_mut(tenant_id) && let Some(User { ty: UserType::Native(user), @@ -175,12 +185,14 @@ impl Users { }) = users.get_mut(userid) { user.password_hash.clone_from(hash); - mut_sessions().remove_user(userid, tenant_id); + sessions.remove_user(userid, tenant_id); }; } pub fn add_roles(&self, userid: &str, roles: HashSet, tenant_id: &Option) { let tenant_id = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); + // SESSIONS before USERS; see the lock order in `map`. + let mut sessions = mut_sessions(); if let Some(users) = mut_users().get_mut(tenant_id) && let Some(user) = users.get_mut(userid) { @@ -188,19 +200,21 @@ impl Users { manual.extend(roles); user.set_manual_roles(manual); let new_perms = roles_to_permission(user.roles(), tenant_id); - mut_sessions().refresh_user_permissions(userid, tenant_id, &new_perms); + sessions.refresh_user_permissions(userid, tenant_id, &new_perms); }; } pub fn remove_roles(&self, userid: &str, roles: HashSet, tenant_id: &Option) { let tenant_id = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); + // SESSIONS before USERS; see the lock order in `map`. + let mut sessions = mut_sessions(); if let Some(users) = mut_users().get_mut(tenant_id) && let Some(user) = users.get_mut(userid) { let manual = user.manual_roles(); user.set_manual_roles(HashSet::from_iter(manual.difference(&roles).cloned())); let new_perms = roles_to_permission(user.roles(), tenant_id); - mut_sessions().refresh_user_permissions(userid, tenant_id, &new_perms); + sessions.refresh_user_permissions(userid, tenant_id, &new_perms); }; } @@ -283,32 +297,102 @@ impl Users { get_tenant_id_from_key(&key) }; let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); - if let Some(users) = users().get(tenant) - && let Some( - user @ User { - ty: UserType::Native(basic_user), - .. - }, - ) = users.get(username) - { - // if user exists and password matches - // add this user to auth map - if basic_user.verify_password(password) { - let mut sessions = mut_sessions(); - sessions.track_new( - username.clone(), - key.clone(), - DateTime::::MAX_UTC, - roles_to_permission(user.roles(), tenant), - &user.tenant, - ); - return sessions - .check_auth(&key, action, context_stream, context_user) - .expect("entry for this key just added"); - } + // add this user to auth map + let Some(sessions) = self.new_basic_session( + username, + password, + tenant, + key.clone(), + DateTime::::MAX_UTC, + ) else { + return Response::UnAuthorized; + }; + sessions + .check_auth(&key, action, context_stream, context_user) + .expect("entry for this key just added") + } + + /// Verifies a native user's password and tracks `key` as their session. + /// + /// The slow verification runs without holding SESSIONS; the user is then + /// re-read under it (lock order SESSIONS -> USERS, see `map`), so a password + /// change, role change or deletion racing the verification cannot leave a + /// stale session. Returns the SESSIONS guard so the caller can use the new + /// session before any later write changes it. + pub fn new_basic_session( + &self, + username: &str, + password: &str, + tenant: &str, + key: SessionKey, + expiry: DateTime, + ) -> Option> { + for _ in 0..BASIC_LOGIN_ATTEMPTS { + let verified_hash = self.verified_password_hash(username, password, tenant)?; + let mut sessions = mut_sessions(); + let current = users() + .get(tenant) + .and_then(|users| users.get(username)) + .filter(|user| { + matches!(&user.ty, UserType::Native(basic) if basic.password_hash == verified_hash) + }) + .map(|user| { + ( + roles_to_permission(user.roles(), tenant), + user.tenant.clone(), + ) + }); + // The user changed after verification; verify against the new state. + let Some((permissions, user_tenant)) = current else { + continue; + }; + sessions.track_new(username.to_owned(), key, expiry, permissions, &user_tenant); + return Some(sessions); } - Response::UnAuthorized + None + } + + /// Returns the stored hash if `password` matches the native user's. The + /// slow hash check runs without holding any RBAC lock. + fn verified_password_hash( + &self, + username: &str, + password: &str, + tenant: &str, + ) -> Option { + let basic = match users().get(tenant).and_then(|users| users.get(username)) { + Some(User { + ty: UserType::Native(basic), + .. + }) => basic.clone(), + _ => return None, + }; + basic + .verify_password(password) + .then_some(basic.password_hash) + } + + /// Copies out the native users named `username`, with the tenant each is + /// stored under, so callers can run the slow password check after dropping + /// USERS. Optionally limited to one tenant. + fn native_users_named( + &self, + username: &str, + tenant_id: Option<&str>, + ) -> Vec<(String, Option, user::Basic)> { + users() + .iter() + .filter(|(tenant, _)| tenant_id.is_none_or(|tenant_id| tenant_id == tenant.as_str())) + .flat_map(|(tenant, usermap)| { + usermap.values().filter_map(move |user| match &user.ty { + UserType::Native(basic) if basic.username.eq(username) => { + Some((tenant.clone(), user.tenant.clone(), basic.clone())) + } + _ => None, + }) + }) + .collect() } #[inline(always)] @@ -318,17 +402,9 @@ impl Users { password: &str, tenant_id: &str, ) -> bool { - users() - .get(tenant_id) - .is_some_and(|tenant_users| { - tenant_users - .values() - .par_bridge() - .find_any(|user| { - matches!(&user.ty, UserType::Native(basic) if basic.username.eq(username) && basic.verify_password(password)) - }) - .is_some() - }) + self.native_users_named(username, Some(tenant_id)) + .par_iter() + .any(|(_, _, basic)| basic.verify_password(password)) } pub fn mutate_request_with_basic_user( @@ -337,14 +413,9 @@ impl Users { password: &str, req: &mut ServiceRequest, ) { - if let Some((tenant, _)) = users().par_iter().find_any(|(_, usermap)| { - usermap - .par_iter() - .find_any(|(_, user)| { - matches!(&user.ty, UserType::Native(basic) if basic.username.eq(username) && basic.verify_password(password)) && user.tenant.is_some() - }) - .is_some() - }) { + if let Some((tenant, _, _)) = self.native_users_named(username, None).par_iter().find_any( + |(_, user_tenant, basic)| user_tenant.is_some() && basic.verify_password(password), + ) { req.headers_mut().insert( HeaderName::from_static(TENANT_ID), HeaderValue::from_bytes(tenant.as_bytes()).unwrap(), @@ -357,18 +428,12 @@ impl Users { } pub fn get_user_tenant_from_basic(&self, username: &str, password: &str) -> Option { - for usermap in users().values() { - for user in usermap.values() { - if let UserType::Native(basic) = &user.ty - && basic.username.eq(username) - && basic.verify_password(password) - && let Some(tenant) = &user.tenant - { - return Some(tenant.clone()); - } - } - } - None + // A user without a tenant never matches, so skip its password check. + self.native_users_named(username, None) + .into_iter() + .find_map(|(_, user_tenant, basic)| { + user_tenant.filter(|_| basic.verify_password(password)) + }) } } @@ -412,3 +477,830 @@ pub fn roles_to_permission(roles: Vec, tenant_id: &str) -> Vec = Mutex::new(()); + + // `map::init` needs `PARSEABLE`, so tests create the maps directly. They hold + // the shared locks on purpose, so they run one at a time. + fn auth_state() -> MutexGuard<'static, ()> { + let serial = SERIAL + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + USERS.get_or_init(|| RwLock::new(map::Users::default())); + SESSIONS.get_or_init(|| RwLock::new(Sessions::default())); + USER_GROUPS.get_or_init(|| RwLock::new(UserGroups::default())); + ROLES.get_or_init(|| RwLock::new(HashMap::new())); + serial + } + + /// A fresh tenant with an admin role, so tests never see each other's users. + fn tenant() -> String { + let tenant = format!("lock-order-{}", Ulid::new()); + mut_roles().entry(tenant.clone()).or_default().insert( + ROLE.to_owned(), + Role::create_internal_role(vec![DefaultPrivilege::Admin]), + ); + tenant + } + + // Default Argon2 parameters take seconds per hash in debug builds. + fn cheap_hash(password: &str) -> String { + let salt = SaltString::encode_b64(&Ulid::new().to_bytes()).unwrap(); + let params = Params::new(Params::MIN_M_COST, 1, 1, None).unwrap(); + Argon2::new(Algorithm::Argon2id, Version::V0x13, params) + .hash_password(password.as_bytes(), &salt) + .unwrap() + .to_string() + } + + /// A native user (not yet stored) and its Basic-auth key. + fn basic_user(tenant: &str, roles: &[&str]) -> (User, SessionKey) { + let username = format!("user-{}", Ulid::new()); + let password = format!("password-{}", Ulid::new()); + let user = User { + ty: UserType::Native(Basic { + username: username.clone(), + password_hash: cheap_hash(&password), + email: None, + }), + roles: roles.iter().map(|role| role.to_string()).collect(), + user_groups: HashSet::new(), + tenant: Some(tenant.to_owned()), + protected: false, + }; + (user, SessionKey::BasicAuth { username, password }) + } + + fn native_user(tenant: &str, roles: &[&str]) -> (User, SessionKey) { + let (user, key) = basic_user(tenant, roles); + mut_users().insert(user.clone()); + (user, key) + } + + /// Tracks a browser-style session for `user`; its permissions come from roles. + fn session_for(user: &User) -> SessionKey { + let key = SessionKey::SessionId(Ulid::new()); + Users.new_session(user, key.clone(), TimeDelta::minutes(5)); + key + } + + fn spawn(f: impl FnOnce() -> T + Send + 'static) -> mpsc::Receiver { + let (tx, rx) = mpsc::channel(); + thread::spawn(move || { + let _ = tx.send(f()); + }); + rx + } + + // A deadlocked thread holds the process-wide maps forever and would hang + // every later test, so abort the run instead of panicking. The abort may + // also write a core dump, which can add several seconds to the failure. + fn deadlock(what: &str) -> ! { + // Bypass libtest's output capture, which an abort would discard. libtest + // prints no result after an abort, so name the test; every caller runs + // on the test's own thread, which libtest names after it. + let current = thread::current(); + let test = current.name().unwrap_or("?"); + let _ = writeln!( + std::io::stderr(), + "{test}: {what} within {DEADLINE:?}: lock-order deadlock, aborting" + ); + std::process::abort() + } + + fn finish(rx: &mpsc::Receiver, what: &str) -> T { + match rx.recv_timeout(DEADLINE) { + Ok(value) => value, + Err(mpsc::RecvTimeoutError::Timeout) => deadlock(&format!("{what} did not finish")), + Err(mpsc::RecvTimeoutError::Disconnected) => panic!("{what} panicked"), + } + } + + fn wait_until(what: &str, done: impl Fn() -> bool) { + let start = Instant::now(); + while !done() { + if start.elapsed() >= DEADLINE { + deadlock(&format!("timed out waiting for {what}")); + } + thread::sleep(Duration::from_millis(1)); + } + } + + // parking_lot's RwLock is task-fair: a queued writer sets the exclusive bit + // and blocks every new reader until it has run. + fn writer_queued(lock: &RwLock) -> bool { + lock.is_locked_exclusive() + } + + fn users_lock() -> &'static RwLock { + USERS.get().unwrap() + } + + fn sessions_lock() -> &'static RwLock { + SESSIONS.get().unwrap() + } + + /// Spawns `write` and waits until it has finished or is queued for USERS. + fn queue_user_writer(write: impl FnOnce() + Send + 'static) -> mpsc::Receiver<()> { + let done = Arc::new(AtomicBool::new(false)); + let rx = spawn({ + let done = done.clone(); + move || { + write(); + done.store(true, Ordering::SeqCst); + } + }); + wait_until("the user writer to queue or finish", || { + done.load(Ordering::SeqCst) || writer_queued(users_lock()) + }); + rx + } + + /// Stands in for a session-authorized request inside `Sessions::check_auth`: + /// it takes SESSIONS(read), then waits for `go` before it reads the user maps. + fn paused_session_check( + key: SessionKey, + ) -> (mpsc::Sender<()>, mpsc::Receiver>) { + let (held_tx, held_rx) = mpsc::channel(); + let (go_tx, go_rx) = mpsc::channel::<()>(); + let result = spawn(move || { + let sessions = sessions(); + held_tx.send(()).unwrap(); + go_rx.recv().unwrap(); + sessions.check_auth(&key, Action::Query, None, None) + }); + finish(&held_rx, "the session check to take SESSIONS"); + (go_tx, result) + } + + fn first_login(key: &SessionKey) -> mpsc::Receiver { + let key = key.clone(); + spawn(move || Users.authorize(key, Action::Query, None, None)) + } + + // Reproduction: a first Basic login kept its USERS read guard while it took + // SESSIONS(write) and then re-read USERS inside `check_auth`. A user write + // queued in between waited for the outer guard and blocked the inner read. + #[test] + fn first_basic_login_survives_a_queued_user_write() { + let _serial = auth_state(); + let tenant = tenant(); + let (_, login) = native_user(&tenant, &[ROLE]); + let (newcomer, _) = basic_user(&tenant, &[ROLE]); + + // Another request reading sessions holds the login back at SESSIONS. + let reader = sessions(); + let login_rx = first_login(&login); + wait_until("the login to queue for SESSIONS", || { + writer_queued(sessions_lock()) + }); + // For example an OIDC login storing its user. + let writer = queue_user_writer(move || mut_users().insert(newcomer)); + drop(reader); + + assert_eq!(finish(&login_rx, "first Basic login"), Response::Authorized); + finish(&writer, "user write"); + assert!(Users.session_exists(&login)); + } + + // Reproduction of the reported three-way cycle: login holds USERS and waits + // for SESSIONS, a session check holds SESSIONS and waits for USERS, and a + // queued user write blocks that read. + #[test] + fn first_login_session_check_and_user_write_do_not_cycle() { + let _serial = auth_state(); + let tenant = tenant(); + let (_, login) = native_user(&tenant, &[ROLE]); + let (member, _) = native_user(&tenant, &[ROLE]); + let member_session = session_for(&member); + let (newcomer, _) = basic_user(&tenant, &[ROLE]); + + let (go, check) = paused_session_check(member_session); + let login_rx = first_login(&login); + wait_until("the login to queue for SESSIONS", || { + writer_queued(sessions_lock()) + }); + let writer = queue_user_writer(move || mut_users().insert(newcomer)); + go.send(()).unwrap(); + + assert_eq!(finish(&check, "session check"), Some(Response::Authorized)); + assert_eq!(finish(&login_rx, "first Basic login"), Response::Authorized); + finish(&writer, "user write"); + } + + // Reproduction: role and password changes took USERS(write) and then + // SESSIONS(write), the reverse of `check_auth`. + fn assert_admin_write_survives_session_check(write: fn(&User)) { + let _serial = auth_state(); + let tenant = tenant(); + let (target, _) = native_user(&tenant, &[]); + let (member, _) = native_user(&tenant, &[ROLE]); + let member_session = session_for(&member); + + let (go, check) = paused_session_check(member_session); + let writer = spawn(move || write(&target)); + wait_until("the admin write to queue for SESSIONS", || { + writer_queued(sessions_lock()) + }); + go.send(()).unwrap(); + + assert_eq!(finish(&check, "session check"), Some(Response::Authorized)); + finish(&writer, "admin write"); + } + + #[test] + fn adding_roles_does_not_deadlock_with_a_session_check() { + assert_admin_write_survives_session_check(|user| { + Users.add_roles( + user.userid(), + HashSet::from([ROLE.to_owned()]), + &user.tenant, + ) + }); + } + + #[test] + fn removing_roles_does_not_deadlock_with_a_session_check() { + assert_admin_write_survives_session_check(|user| { + Users.remove_roles( + user.userid(), + HashSet::from([ROLE.to_owned()]), + &user.tenant, + ) + }); + } + + #[test] + fn password_change_does_not_deadlock_with_a_session_check() { + assert_admin_write_survives_session_check(|user| { + Users.change_password_hash(user.userid(), &cheap_hash("changed"), &user.tenant) + }); + } + + // Not a reproduction: `put_user` used to release SESSIONS before writing + // USERS. This guards the lock order now that it holds both. + #[test] + fn replacing_a_user_does_not_deadlock_with_a_session_check() { + assert_admin_write_survives_session_check(|user| Users.put_user(user.clone())); + } + + // If `put_user` released SESSIONS between removing the user's sessions and + // storing the replacement, a first login that had verified the old password + // could track a session in that gap, and nothing would remove it. The gap + // has no lock acquisition a test could stop a thread at, so this checks + // that SESSIONS stays held across it, then that a racing login gets nothing. + #[test] + fn replacing_a_user_during_first_login_leaves_no_stale_session() { + let _serial = auth_state(); + let tenant = tenant(); + let (user, login) = native_user(&tenant, &[ROLE]); + let mut replacement = user; + if let UserType::Native(basic) = &mut replacement.ty { + basic.password_hash = cheap_hash("changed"); + } + + // Stops `put_user` after it removed the sessions, before it stores the user. + let reader = users(); + let writer = spawn(move || Users.put_user(replacement)); + wait_until("the replacement to queue for USERS", || { + writer_queued(users_lock()) + }); + assert!( + sessions_lock().is_locked_exclusive(), + "put_user released SESSIONS before storing the replacement" + ); + let login_rx = first_login(&login); + drop(reader); + + finish(&writer, "user replacement"); + assert_eq!(finish(&login_rx, "login"), Response::UnAuthorized); + assert!(!Users.session_exists(&login)); + } + + // Reproduction: deleting a user on the querier held USER_GROUPS(write) (and + // USERS) while refreshing sessions, the reverse of `check_auth`. + #[test] + fn group_removal_does_not_deadlock_with_a_session_check() { + let _serial = auth_state(); + let tenant = tenant(); + let group_name = format!("group-{}", Ulid::new()); + let (mut leaving, _) = basic_user(&tenant, &[]); + let (mut member, _) = basic_user(&tenant, &[ROLE]); + for user in [&mut leaving, &mut member] { + user.user_groups.insert(group_name.clone()); + mut_users().insert(user.clone()); + } + let group = UserGroup::new( + group_name.clone(), + HashSet::from([ROLE.to_owned()]), + HashSet::from([ + GroupUser::from_user(&leaving), + GroupUser::from_user(&member), + ]), + ); + write_user_groups().insert(group, &tenant); + let member_session = session_for(&member); + + let (go, check) = paused_session_check(member_session); + let removal = spawn({ + let userid = leaving.userid().to_owned(); + let tenant = leaving.tenant.clone(); + move || remove_user_from_groups(&userid, &tenant) + }); + wait_until("the group removal to queue for SESSIONS", || { + writer_queued(sessions_lock()) + }); + go.send(()).unwrap(); + + assert_eq!(finish(&check, "session check"), Some(Response::Authorized)); + let updated_groups = finish(&removal, "group removal").unwrap(); + assert_eq!(updated_groups.len(), 1); + let leaving_member = GroupUser::from_user(&leaving); + assert!(!updated_groups[0].users.contains(&leaving_member)); + assert!( + read_user_groups()[&tenant][&group_name] + .users + .contains(&leaving_member) + ); + crate::handlers::http::modal::query::querier_rbac::apply_user_removal_to_groups( + leaving.userid(), + &leaving.tenant, + &updated_groups, + ); + let groups = read_user_groups(); + let users = &groups[&tenant][&group_name].users; + assert!(!users.contains(&GroupUser::from_user(&leaving))); + assert!(users.contains(&GroupUser::from_user(&member))); + } + + #[test] + fn basic_login_authorizes_once_then_reuses_the_session() { + let _serial = auth_state(); + let tenant = tenant(); + let (_, login) = native_user(&tenant, &[ROLE]); + + assert_eq!(finish(&first_login(&login), "login"), Response::Authorized); + assert!(Users.session_exists(&login)); + assert_eq!(finish(&first_login(&login), "login"), Response::Authorized); + + let SessionKey::BasicAuth { username, .. } = &login else { + unreachable!() + }; + let wrong = SessionKey::BasicAuth { + username: username.clone(), + password: "wrong".into(), + }; + assert_eq!( + finish(&first_login(&wrong), "login"), + Response::UnAuthorized + ); + assert!(!Users.session_exists(&wrong)); + let unknown = SessionKey::BasicAuth { + username: format!("missing-{}", Ulid::new()), + password: "whatever".into(), + }; + assert_eq!( + finish(&first_login(&unknown), "login"), + Response::UnAuthorized + ); + assert_eq!( + Users.authorize( + SessionKey::SessionId(Ulid::new()), + Action::Query, + None, + None + ), + Response::ReloadRequired + ); + } + + #[test] + fn basic_login_without_roles_is_tracked_but_unauthorized() { + let _serial = auth_state(); + let tenant = tenant(); + let (_, login) = native_user(&tenant, &[]); + + assert_eq!( + finish(&first_login(&login), "login"), + Response::UnAuthorized + ); + assert!(Users.session_exists(&login)); + } + + #[test] + fn session_check_includes_group_permissions() { + let _serial = auth_state(); + let tenant = tenant(); + let group_name = format!("group-{}", Ulid::new()); + let (mut grouped, _) = basic_user(&tenant, &[]); + grouped.user_groups.insert(group_name.clone()); + mut_users().insert(grouped.clone()); + let (outsider, _) = native_user(&tenant, &[]); + write_user_groups().insert( + UserGroup::new( + group_name, + HashSet::from([ROLE.to_owned()]), + HashSet::from([GroupUser::from_user(&grouped)]), + ), + &tenant, + ); + + let grouped_session = session_for(&grouped); + let outsider_session = session_for(&outsider); + assert_eq!( + Users.authorize(grouped_session, Action::Query, None, None), + Response::Authorized + ); + assert_eq!( + Users.authorize(outsider_session, Action::Query, None, None), + Response::UnAuthorized + ); + } + + // A password change that lands while a first login is verifying the old + // password must not leave a session keyed by the old password. + #[test] + fn password_change_during_first_login_leaves_no_stale_session() { + let _serial = auth_state(); + let tenant = tenant(); + let (user, login) = native_user(&tenant, &[ROLE]); + + let reader = sessions(); + let login_rx = first_login(&login); + wait_until("the login to queue for SESSIONS", || { + writer_queued(sessions_lock()) + }); + let new_hash = cheap_hash("changed"); + let writer = queue_user_writer(move || { + if let Some(users) = mut_users().get_mut(user.tenant.as_deref().unwrap()) + && let Some(User { + ty: UserType::Native(basic), + .. + }) = users.get_mut(user.userid()) + { + basic.password_hash = new_hash; + } + }); + finish(&writer, "password change"); + drop(reader); + + assert_eq!(finish(&login_rx, "login"), Response::UnAuthorized); + assert!(!Users.session_exists(&login)); + } + + /// Runs `/login`'s exchange of a Basic key for a session cookie. + fn cookie_login(key: &SessionKey, tenant: &str) -> mpsc::Receiver>> { + let (key, tenant) = (key.clone(), Some(tenant.to_owned())); + spawn(move || { + let (username, password) = credentials(&key); + exchange_basic_for_cookie(username, password, &tenant, EXPIRY_DURATION) + }) + } + + /// The session a `/login` cookie names, checked to belong to `login`'s user + /// and tenant, with the Basic session `/login` authorized through removed. + fn issued_session( + cookie: Option>, + login: &SessionKey, + tenant: &str, + ) -> SessionKey { + let cookie = cookie.expect("the cookie login to issue a cookie"); + let session = SessionKey::SessionId(cookie.value().parse().unwrap()); + assert_eq!( + Users.get_userid_from_session(&session), + Some((credentials(login).0.to_owned(), tenant.to_owned())) + ); + assert!(!Users.session_exists(login)); + session + } + + /// How many sessions, of any kind, are tracked for `login`'s user. + fn tracked_sessions(login: &SessionKey, tenant: &str) -> usize { + let username = credentials(login).0; + sessions() + .get_active_sessions() + .into_iter() + .filter(|(_, userid, tenant_id)| userid == username && tenant_id == tenant) + .count() + } + + // `/login` verified the password and then tracked its cookie session from a + // copy of the user, so a password change in between left a session that + // outlived the reset. + #[test] + fn password_change_during_cookie_login_leaves_no_session() { + let _serial = auth_state(); + let tenant = tenant(); + let (user, login) = native_user(&tenant, &[ROLE]); + + let reader = sessions(); + let login_rx = cookie_login(&login, &tenant); + wait_until("the cookie login to queue for SESSIONS", || { + writer_queued(sessions_lock()) + }); + let new_hash = cheap_hash("changed"); + let writer = queue_user_writer(move || { + if let Some(users) = mut_users().get_mut(user.tenant.as_deref().unwrap()) + && let Some(User { + ty: UserType::Native(basic), + .. + }) = users.get_mut(user.userid()) + { + basic.password_hash = new_hash; + } + }); + finish(&writer, "password change"); + drop(reader); + + assert!(finish(&login_rx, "cookie login").is_none()); + assert_eq!(tracked_sessions(&login, &tenant), 0); + } + + #[test] + fn cookie_login_tracks_a_session_with_the_users_roles_and_tenant() { + let _serial = auth_state(); + let tenant = tenant(); + let (_, login) = native_user(&tenant, &[ROLE]); + let (_, roleless) = native_user(&tenant, &[]); + + // `/login` authorizes the Basic key, which tracks it, before exchanging it. + finish(&first_login(&login), "login"); + assert!(Users.session_exists(&login)); + let cookie = finish(&cookie_login(&login, &tenant), "cookie login"); + let session = issued_session(cookie, &login, &tenant); + assert_eq!( + Users.authorize(session, Action::Query, None, None), + Response::Authorized + ); + let cookie = finish(&cookie_login(&roleless, &tenant), "cookie login"); + let session = issued_session(cookie, &roleless, &tenant); + assert_eq!( + Users.authorize(session, Action::Query, None, None), + Response::UnAuthorized + ); + + let (username, _) = credentials(&login); + let wrong = SessionKey::BasicAuth { + username: username.to_owned(), + password: "wrong".into(), + }; + let before = tracked_sessions(&login, &tenant); + assert!(finish(&cookie_login(&wrong, &tenant), "cookie login").is_none()); + assert_eq!(tracked_sessions(&login, &tenant), before); + } + + // The Basic lookups verify passwords after dropping USERS; they must still + // pick the user's tenant and reject a wrong password or tenant. + #[test] + fn basic_tenant_lookups_match_the_users_tenant() { + let _serial = auth_state(); + let tenant = tenant(); + let (_, login) = native_user(&tenant, &[ROLE]); + let (username, password) = credentials(&login); + + assert_eq!( + Users.get_user_tenant_from_basic(username, password), + Some(tenant.clone()) + ); + assert_eq!(Users.get_user_tenant_from_basic(username, "wrong"), None); + assert!(Users.validate_basic_user_tenant_id(username, password, &tenant)); + assert!(!Users.validate_basic_user_tenant_id(username, "wrong", &tenant)); + assert!(!Users.validate_basic_user_tenant_id(username, password, "elsewhere")); + + let mut req = actix_web::test::TestRequest::default().to_srv_request(); + Users.mutate_request_with_basic_user(username, password, &mut req); + assert_eq!(req.headers().get(TENANT_ID).unwrap(), tenant.as_str()); + let mut req = actix_web::test::TestRequest::default().to_srv_request(); + Users.mutate_request_with_basic_user(username, "wrong", &mut req); + assert!(req.headers().get(TENANT_ID).is_none()); + } + + // A role change that lands while a first login is verifying must apply to + // the session that login creates. + #[test] + fn role_change_during_first_login_applies_to_the_session() { + let _serial = auth_state(); + let tenant = tenant(); + let (user, login) = native_user(&tenant, &[ROLE]); + + let reader = sessions(); + let login_rx = first_login(&login); + wait_until("the login to queue for SESSIONS", || { + writer_queued(sessions_lock()) + }); + let writer = queue_user_writer(move || { + let mut demoted = user; + demoted.roles.clear(); + mut_users().insert(demoted); + }); + finish(&writer, "role change"); + drop(reader); + + assert_eq!(finish(&login_rx, "login"), Response::UnAuthorized); + assert_eq!(sessions().get(&login), Some(&Vec::new())); + } + + // Resetting the password to the same value mid-login changes the stored + // hash; the login verifies again and still succeeds. + #[test] + fn same_password_reset_during_first_login_retries_and_succeeds() { + let _serial = auth_state(); + let tenant = tenant(); + let (user, login) = native_user(&tenant, &[ROLE]); + + let reader = sessions(); + let login_rx = first_login(&login); + wait_until("the login to queue for SESSIONS", || { + writer_queued(sessions_lock()) + }); + let new_hash = cheap_hash(credentials(&login).1); + let writer = queue_user_writer(move || { + if let Some(users) = mut_users().get_mut(user.tenant.as_deref().unwrap()) + && let Some(User { + ty: UserType::Native(basic), + .. + }) = users.get_mut(user.userid()) + { + basic.password_hash = new_hash; + } + }); + finish(&writer, "password reset"); + drop(reader); + + assert_eq!(finish(&login_rx, "login"), Response::Authorized); + assert!(Users.session_exists(&login)); + } + + // Reproduction: `GET /users` rendered each user while holding USERS(read), + // and rendering read USERS again for the user's roles and groups. + #[test] + fn rendering_users_survives_a_queued_user_write() { + let _serial = auth_state(); + let tenant = tenant(); + let group_name = format!("group-{}", Ulid::new()); + let (mut user, _) = basic_user(&tenant, &[ROLE]); + user.user_groups.insert(group_name.clone()); + mut_users().insert(user.clone()); + write_user_groups().insert( + UserGroup::new( + group_name.clone(), + HashSet::from([ROLE.to_owned()]), + HashSet::from([GroupUser::from_user(&user)]), + ), + &tenant, + ); + let (newcomer, _) = basic_user(&tenant, &[ROLE]); + + let (held_tx, held_rx) = mpsc::channel(); + let (go_tx, go_rx) = mpsc::channel::<()>(); + let render = spawn({ + let tenant = tenant.clone(); + let userid = user.userid().to_owned(); + move || { + // Like `list_users_prism`, which renders inside `users().get(..)`. + let users = users(); + held_tx.send(()).unwrap(); + go_rx.recv().unwrap(); + utils::prism_roles(&users[&tenant][&userid]) + } + }); + finish(&held_rx, "the render to take USERS"); + let writer = queue_user_writer(move || mut_users().insert(newcomer)); + go_tx.send(()).unwrap(); + + let (direct, groups, group_names) = finish(&render, "user render"); + finish(&writer, "user write"); + assert_eq!(direct.into_keys().collect::>(), [ROLE]); + assert_eq!(groups[&group_name].keys().collect::>(), [ROLE]); + assert_eq!(group_names, HashSet::from([group_name])); + } + + #[test] + fn deletion_during_first_login_leaves_no_session() { + let _serial = auth_state(); + let tenant = tenant(); + let (user, login) = native_user(&tenant, &[ROLE]); + + let reader = sessions(); + let login_rx = first_login(&login); + wait_until("the login to queue for SESSIONS", || { + writer_queued(sessions_lock()) + }); + let deletion = spawn(move || { + let mut users = Users; + users.delete_user(user.userid(), &user.tenant) + }); + wait_until("the user to be removed", || { + users_lock() + .try_read() + .is_some_and(|users| !users[&tenant].contains_key(credentials(&login).0)) + }); + drop(reader); + + assert_eq!(finish(&login_rx, "login"), Response::UnAuthorized); + finish(&deletion, "deletion"); + assert!(!Users.session_exists(&login)); + } + + fn credentials(key: &SessionKey) -> (&str, &str) { + match key { + SessionKey::BasicAuth { username, password } => (username, password), + SessionKey::SessionId(_) => unreachable!(), + } + } + + // Logins, session checks and user writes racing freely must all finish. + #[test] + fn concurrent_logins_session_checks_and_user_writes_finish() { + const ROUNDS: usize = 40; + let _serial = auth_state(); + let tenant = tenant(); + // Set everything up first: a deadlocked worker would block the test thread. + let logins: Vec<_> = (0..4).map(|_| native_user(&tenant, &[ROLE]).1).collect(); + let sessions: Vec<_> = (0..4) + .map(|_| session_for(&native_user(&tenant, &[ROLE]).0)) + .collect(); + let targets: Vec<_> = (0..2).map(|_| native_user(&tenant, &[]).0).collect(); + let newcomers: Vec> = (0..2) + .map(|_| { + (0..ROUNDS) + .map(|_| basic_user(&tenant, &[ROLE]).0) + .collect() + }) + .collect(); + let new_hash = cheap_hash("changed"); + + let mut workers = Vec::new(); + for login in logins { + workers.push(spawn(move || { + for _ in 0..ROUNDS { + // Forget the session so every round is a first login. + Users.remove_session(&login); + assert_eq!( + Users.authorize(login.clone(), Action::Query, None, None), + Response::Authorized + ); + } + })); + } + for session in sessions { + workers.push(spawn(move || { + for _ in 0..ROUNDS { + assert_eq!( + Users.authorize(session.clone(), Action::Query, None, None), + Response::Authorized + ); + } + })); + } + for (target, newcomers) in targets.into_iter().zip(newcomers) { + let new_hash = new_hash.clone(); + workers.push(spawn(move || { + let roles = HashSet::from([ROLE.to_owned()]); + for newcomer in newcomers { + Users.add_roles(target.userid(), roles.clone(), &target.tenant); + Users.remove_roles(target.userid(), roles.clone(), &target.tenant); + Users.change_password_hash(target.userid(), &new_hash, &target.tenant); + mut_users().insert(newcomer); + Users.put_user(target.clone()); + } + })); + } + for (index, worker) in workers.iter().enumerate() { + finish(worker, &format!("worker {index}")); + } + } +} diff --git a/src/rbac/utils.rs b/src/rbac/utils.rs index 435fc4b1e..e4785713a 100644 --- a/src/rbac/utils.rs +++ b/src/rbac/utils.rs @@ -25,13 +25,12 @@ use crate::{ }; use super::{ - Users, UsersPrism, + UsersPrism, map::roles, user::{User, UserType}, }; pub fn to_prism_user(user: &User) -> UsersPrism { - let tenant_id = user.tenant.as_deref().unwrap_or(DEFAULT_TENANT); let (id, username, method, email, picture) = match &user.ty { UserType::Native(basic) => ( user.userid(), @@ -59,8 +58,29 @@ pub fn to_prism_user(user: &User) -> UsersPrism { None, ), }; - let direct_roles: HashMap = Users - .get_role(id, &user.tenant) + let (direct_roles, group_roles, user_groups) = prism_roles(user); + + UsersPrism { + id: id.into(), + username: username.into(), + method: method.into(), + email: mask_pii_string(email), + picture: mask_pii_url(picture), + roles: direct_roles, + group_roles, + user_groups, + } +} + +type GroupRoles = HashMap>; + +/// The user's direct roles, roles inherited per group, and group names. +pub(super) fn prism_roles(user: &User) -> (HashMap, GroupRoles, HashSet) { + let tenant_id = user.tenant.as_deref().unwrap_or(DEFAULT_TENANT); + // Read roles and groups from `user` itself: callers render while holding + // USERS(read), and reading it again deadlocks behind a queued writer. + let direct_roles: HashMap = user + .roles .iter() .filter_map(|role_name| { roles() @@ -79,12 +99,12 @@ pub fn to_prism_user(user: &User) -> UsersPrism { }) .collect(); - let mut group_roles: HashMap> = HashMap::new(); + let mut group_roles: GroupRoles = HashMap::new(); let mut user_groups = HashSet::new(); // user might be part of some user groups, fetch the roles from there as well - for user_group in Users.get_user_groups(user.userid(), &user.tenant) { + for user_group in &user.user_groups { if let Some(groups) = read_user_groups().get(tenant_id) - && let Some(group) = groups.get(&user_group) + && let Some(group) = groups.get(user_group) { let ug_roles: HashMap = group .roles @@ -100,20 +120,11 @@ pub fn to_prism_user(user: &User) -> UsersPrism { }) .collect(); group_roles.insert(group.name.clone(), ug_roles); - user_groups.insert(user_group); + user_groups.insert(user_group.clone()); } } - UsersPrism { - id: id.into(), - username: username.into(), - method: method.into(), - email: mask_pii_string(email), - picture: mask_pii_url(picture), - roles: direct_roles, - group_roles, - user_groups, - } + (direct_roles, group_roles, user_groups) } //mask PII string if the P_MASK_PII is set