From edf031dd399cc0c65d52a4972601a21e609fb01d Mon Sep 17 00:00:00 2001 From: Aldo Borrero <82811+aldoborrero@users.noreply.github.com> Date: Tue, 6 Oct 2026 17:18:01 +0200 Subject: [PATCH 1/2] Recheck administrator authority during user changes --- ...e4a0120c8cab6a7fb09c9a5498563fb895ea7.json | 22 ++ ...8448a2090c0cc6beced66a8f615c2484aa86b.json | 35 ++ ...7d67cbe74587fa68c29659edb184d7dd5afc8.json | 16 + crates/horae/src/server_fns/users.rs | 64 +++- .../src/server_fns/users/tests/authority.rs | 325 ++++++++++++++++++ .../src/server_fns/users/tests/concurrency.rs | 23 +- 6 files changed, 459 insertions(+), 26 deletions(-) create mode 100644 .sqlx/query-2222bdda941662838ed867ac8fbe4a0120c8cab6a7fb09c9a5498563fb895ea7.json create mode 100644 .sqlx/query-449f1174e711adfebfdd1fc4b118448a2090c0cc6beced66a8f615c2484aa86b.json create mode 100644 .sqlx/query-dfe8f696b4fcf0c6f5405c8ca137d67cbe74587fa68c29659edb184d7dd5afc8.json create mode 100644 crates/horae/src/server_fns/users/tests/authority.rs diff --git a/.sqlx/query-2222bdda941662838ed867ac8fbe4a0120c8cab6a7fb09c9a5498563fb895ea7.json b/.sqlx/query-2222bdda941662838ed867ac8fbe4a0120c8cab6a7fb09c9a5498563fb895ea7.json new file mode 100644 index 00000000..23fdef1c --- /dev/null +++ b/.sqlx/query-2222bdda941662838ed867ac8fbe4a0120c8cab6a7fb09c9a5498563fb895ea7.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT COUNT(*) AS \"count!\" FROM users WHERE org_id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "count!", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + null + ] + }, + "hash": "2222bdda941662838ed867ac8fbe4a0120c8cab6a7fb09c9a5498563fb895ea7" +} diff --git a/.sqlx/query-449f1174e711adfebfdd1fc4b118448a2090c0cc6beced66a8f615c2484aa86b.json b/.sqlx/query-449f1174e711adfebfdd1fc4b118448a2090c0cc6beced66a8f615c2484aa86b.json new file mode 100644 index 00000000..144cf42a --- /dev/null +++ b/.sqlx/query-449f1174e711adfebfdd1fc4b118448a2090c0cc6beced66a8f615c2484aa86b.json @@ -0,0 +1,35 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT id FROM users WHERE id = $1 AND org_id = $2\n AND active AND org_role = $3 FOR SHARE", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Uuid", + { + "Custom": { + "name": "org_role", + "kind": { + "Enum": [ + "admin", + "manager", + "member" + ] + } + } + } + ] + }, + "nullable": [ + false + ] + }, + "hash": "449f1174e711adfebfdd1fc4b118448a2090c0cc6beced66a8f615c2484aa86b" +} diff --git a/.sqlx/query-dfe8f696b4fcf0c6f5405c8ca137d67cbe74587fa68c29659edb184d7dd5afc8.json b/.sqlx/query-dfe8f696b4fcf0c6f5405c8ca137d67cbe74587fa68c29659edb184d7dd5afc8.json new file mode 100644 index 00000000..546f1836 --- /dev/null +++ b/.sqlx/query-dfe8f696b4fcf0c6f5405c8ca137d67cbe74587fa68c29659edb184d7dd5afc8.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO users (id, org_id, email, name, org_role) VALUES ($1, $2, $3, 'Target', 'member')", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Uuid", + "Text" + ] + }, + "nullable": [] + }, + "hash": "dfe8f696b4fcf0c6f5405c8ca137d67cbe74587fa68c29659edb184d7dd5afc8" +} diff --git a/crates/horae/src/server_fns/users.rs b/crates/horae/src/server_fns/users.rs index b76ef0a6..a28c3d05 100644 --- a/crates/horae/src/server_fns/users.rs +++ b/crates/horae/src/server_fns/users.rs @@ -36,9 +36,29 @@ pub async fn list_users(include_inactive: bool) -> Result, Ser pub async fn create_user(email: String, name: String, role: String) -> Result { let admin = require_admin().await?; let state = crate::state::global_state().await; - let id = uuid::Uuid::now_v7(); let org_role: OrgRole = parse_enum(&role, "role (use admin, manager, or member)")?; + let user = insert_user(&state.db, admin.org_id, admin.id, &email, &name, org_role).await?; + state + .plugins + .dispatch(crate::plugin::AppEvent::UserCreated { + occurred_at: chrono::Utc::now(), + org_id: admin.org_id, + user: user_payload(&user), + }); + Ok(user) +} + +#[cfg(feature = "server")] +async fn insert_user( + db: &sqlx::PgPool, + org_id: uuid::Uuid, + actor_id: uuid::Uuid, + email: &str, + name: &str, + org_role: OrgRole, +) -> Result { + let mut tx = begin_user_access_change(db, org_id, actor_id).await?; let user = sqlx::query_as!( User, r#"INSERT INTO users (id, org_id, email, name, org_role) @@ -47,13 +67,13 @@ pub async fn create_user(email: String, name: String, role: String) -> Result""#, - id, - admin.org_id, + uuid::Uuid::now_v7(), + org_id, email, name, org_role as OrgRole, ) - .fetch_one(&state.db) + .fetch_one(&mut *tx) .await .map_err(|e| { if e.to_string().contains("users_email_key") { @@ -62,14 +82,7 @@ pub async fn create_user(email: String, name: String, role: String) -> Result Result Result, ServerFnError> { let mut tx = db.begin().await.map_err(server_err)?; // Both role and activation changes must serialize before counting admins; @@ -89,6 +103,19 @@ async fn begin_user_access_change( .await .map_err(server_err)? .ok_or_else(|| not_found("Organization not found"))?; + // Recheck after the organization lock: authority may have been revoked + // while this request waited. Keep the actor locked through commit. + sqlx::query_scalar!( + "SELECT id FROM users WHERE id = $1 AND org_id = $2 + AND active AND org_role = $3 FOR SHARE", + actor_id, + org_id, + OrgRole::Admin as OrgRole, + ) + .fetch_optional(&mut *tx) + .await + .map_err(server_err)? + .ok_or_else(|| forbidden("Active administrator access required"))?; Ok(tx) } @@ -150,7 +177,8 @@ pub async fn set_user_role(user_id: String, role: String) -> Result Result Result<(User, Option), ServerFnError> { - let mut tx = begin_user_access_change(db, org_id).await?; + let mut tx = begin_user_access_change(db, org_id, actor_id).await?; // Read the prior role/active so the event reports the transition (a no-op // role change emits nothing, FR-012) and so we can refuse demoting the last // active admin — including an admin dropping their own role. @@ -214,7 +243,8 @@ pub async fn set_user_active(user_id: String, active: bool) -> Result Result Result<(User, Option), ServerFnError> { - let mut tx = begin_user_access_change(db, org_id).await?; + let mut tx = begin_user_access_change(db, org_id, actor_id).await?; let current = user_active_role(&mut tx, user_id, org_id).await?; let was_active: Option = current.map(|(active, _)| active); @@ -273,5 +304,6 @@ async fn change_user_active( mod tests { use super::*; + mod authority; mod concurrency; } diff --git a/crates/horae/src/server_fns/users/tests/authority.rs b/crates/horae/src/server_fns/users/tests/authority.rs new file mode 100644 index 00000000..0b12bfb7 --- /dev/null +++ b/crates/horae/src/server_fns/users/tests/authority.rs @@ -0,0 +1,325 @@ +use super::*; +use crate::server_fns::test_seed::{seed, wait_for_blocked}; +use sqlx::PgPool; +use uuid::Uuid; + +#[derive(Clone, Copy, Debug)] +enum Change { + Create, + Role, + Active, +} + +async fn attempt( + pool: &PgPool, + org: Uuid, + actor: Uuid, + target: Uuid, + change: Change, +) -> Result<(), ServerFnError> { + match change { + Change::Create => insert_user( + pool, + org, + actor, + &format!("{}@test.com", Uuid::now_v7()), + "New member", + OrgRole::Member, + ) + .await + .map(|_| ()), + Change::Role => change_user_role(pool, org, actor, target, OrgRole::Manager) + .await + .map(|_| ()), + Change::Active => change_user_active(pool, org, actor, target, false) + .await + .map(|_| ()), + } +} + +async fn target(pool: &PgPool, org: Uuid) -> Uuid { + let id = Uuid::now_v7(); + sqlx::query!("INSERT INTO users (id, org_id, email, name, org_role) VALUES ($1, $2, $3, 'Target', 'member')", id, org, format!("{id}@test.com")) + .execute(pool).await.unwrap(); + id +} + +async fn assert_unchanged(pool: &PgPool, org: Uuid, target: Uuid) { + let count = sqlx::query_scalar!( + "SELECT COUNT(*) AS \"count!\" FROM users WHERE org_id = $1", + org + ) + .fetch_one(pool) + .await + .unwrap(); + assert_eq!(count, 2, "unauthorized request created a user"); + let state = user_active_role(&mut pool.acquire().await.unwrap(), target, org) + .await + .unwrap(); + assert_eq!( + state, + Some((true, OrgRole::Member)), + "unauthorized request modified the target" + ); +} + +#[sqlx::test(migrations = "./migrations")] +#[serial_test::serial] +async fn revoked_administrator_cannot_create_change_roles_or_deactivate(pool: PgPool) { + let ids = seed(&pool, OrgRole::Admin).await; + let target = target(&pool, ids.org_id).await; + for (role, active) in [ + (OrgRole::Member, true), + (OrgRole::Manager, true), + (OrgRole::Admin, false), + ] { + sqlx::query!( + "UPDATE users SET org_role = $2, active = $3 WHERE id = $1", + ids.user_id, + role as OrgRole, + active + ) + .execute(&pool) + .await + .unwrap(); + for change in [Change::Create, Change::Role, Change::Active] { + let result = attempt(&pool, ids.org_id, ids.user_id, target, change).await; + assert!( + matches!( + result, + Err(ServerFnError::ServerError { + code: FORBIDDEN, + .. + }) + ), + "{role:?}/{active}/{change:?}: {result:?}" + ); + assert_unchanged(&pool, ids.org_id, target).await; + } + } +} + +#[sqlx::test(migrations = "./migrations")] +#[serial_test::serial] +async fn foreign_and_unknown_actors_cannot_change_local_user_access(pool: PgPool) { + let ids = seed(&pool, OrgRole::Admin).await; + let foreign = seed(&pool, OrgRole::Admin).await; + let target = target(&pool, ids.org_id).await; + for actor in [foreign.user_id, Uuid::now_v7()] { + for change in [Change::Create, Change::Role, Change::Active] { + let result = attempt(&pool, ids.org_id, actor, target, change).await; + assert!( + matches!( + result, + Err(ServerFnError::ServerError { + code: FORBIDDEN, + .. + }) + ), + "{change:?}: {result:?}" + ); + assert_unchanged(&pool, ids.org_id, target).await; + } + } +} + +#[sqlx::test(migrations = "./migrations")] +#[serial_test::serial] +async fn user_mutations_wait_for_actor_revocation_and_recheck_after_commit(pool: PgPool) { + for change in [Change::Create, Change::Role, Change::Active] { + let ids = seed(&pool, OrgRole::Admin).await; + let target = target(&pool, ids.org_id).await; + let mut revocation = pool.begin().await.unwrap(); + let blocker = sqlx::query_scalar!("SELECT pg_backend_pid() AS \"pid!\"") + .fetch_one(&mut *revocation) + .await + .unwrap(); + sqlx::query!( + "UPDATE users SET org_role = 'member' WHERE id = $1", + ids.user_id + ) + .execute(&mut *revocation) + .await + .unwrap(); + let db = pool.clone(); + let mut request = + tokio::spawn( + async move { attempt(&db, ids.org_id, ids.user_id, target, change).await }, + ); + let result = tokio::select! { + result = &mut request => { + revocation.commit().await.unwrap(); + result.unwrap() + }, + () = wait_for_blocked(&pool, blocker) => { + revocation.commit().await.unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(10), request).await.unwrap().unwrap() + }, + }; + assert!( + matches!( + result, + Err(ServerFnError::ServerError { + code: FORBIDDEN, + .. + }) + ), + "{change:?}: {result:?}" + ); + assert_unchanged(&pool, ids.org_id, target).await; + } +} + +#[sqlx::test(migrations = "./migrations")] +#[serial_test::serial] +async fn user_mutations_recheck_revocation_after_waiting_for_organization_lock(pool: PgPool) { + for change in [Change::Create, Change::Role, Change::Active] { + let ids = seed(&pool, OrgRole::Admin).await; + let target = target(&pool, ids.org_id).await; + let mut revocation = pool.begin().await.unwrap(); + let blocker = sqlx::query_scalar!("SELECT pg_backend_pid() AS \"pid!\"") + .fetch_one(&mut *revocation) + .await + .unwrap(); + sqlx::query!( + "SELECT id FROM organizations WHERE id = $1 FOR UPDATE", + ids.org_id + ) + .fetch_one(&mut *revocation) + .await + .unwrap(); + let db = pool.clone(); + let mut request = + tokio::spawn( + async move { attempt(&db, ids.org_id, ids.user_id, target, change).await }, + ); + let early = tokio::select! { + result = &mut request => Some(result.unwrap()), + () = wait_for_blocked(&pool, blocker) => None, + }; + sqlx::query!("UPDATE users SET active = false WHERE id = $1", ids.user_id) + .execute(&mut *revocation) + .await + .unwrap(); + revocation.commit().await.unwrap(); + let result = match early { + Some(result) => result, + None => tokio::time::timeout(std::time::Duration::from_secs(10), request) + .await + .unwrap() + .unwrap(), + }; + assert!( + matches!( + result, + Err(ServerFnError::ServerError { + code: FORBIDDEN, + .. + }) + ), + "{change:?}: {result:?}" + ); + assert_unchanged(&pool, ids.org_id, target).await; + } +} + +#[sqlx::test(migrations = "./migrations")] +#[serial_test::serial] +async fn active_administrator_can_create_change_roles_and_deactivate(pool: PgPool) { + let ids = seed(&pool, OrgRole::Admin).await; + let created = insert_user( + &pool, + ids.org_id, + ids.user_id, + "new@test.com", + "New", + OrgRole::Member, + ) + .await + .unwrap(); + assert_eq!(created.org_id, ids.org_id); + assert_eq!(created.org_role, OrgRole::Member); + let (updated, previous) = + change_user_role(&pool, ids.org_id, ids.user_id, created.id, OrgRole::Manager) + .await + .unwrap(); + assert_eq!(previous, Some(OrgRole::Member)); + assert_eq!(updated.org_role, OrgRole::Manager); + let (updated, previous) = change_user_active(&pool, ids.org_id, ids.user_id, created.id, false) + .await + .unwrap(); + assert_eq!(previous, Some(true)); + assert!(!updated.active); +} + +#[sqlx::test(migrations = "./migrations")] +#[serial_test::serial] +async fn duplicate_creation_rolls_back_and_leaves_access_changes_usable(pool: PgPool) { + let ids = seed(&pool, OrgRole::Admin).await; + let target = target(&pool, ids.org_id).await; + let result = insert_user( + &pool, + ids.org_id, + ids.user_id, + &format!("{target}@test.com"), + "Duplicate", + OrgRole::Admin, + ) + .await; + assert!( + matches!( + result, + Err(ServerFnError::ServerError { code: CONFLICT, .. }) + ), + "{result:?}" + ); + assert_unchanged(&pool, ids.org_id, target).await; + let (updated, _) = tokio::time::timeout( + std::time::Duration::from_secs(10), + change_user_role(&pool, ids.org_id, ids.user_id, target, OrgRole::Manager), + ) + .await + .unwrap() + .unwrap(); + assert_eq!(updated.org_role, OrgRole::Manager); +} + +#[sqlx::test(migrations = "./migrations")] +#[serial_test::serial] +async fn access_transaction_holds_actor_authority_until_commit(pool: PgPool) { + let ids = seed(&pool, OrgRole::Admin).await; + let mut access = begin_user_access_change(&pool, ids.org_id, ids.user_id) + .await + .unwrap(); + let blocker = sqlx::query_scalar!("SELECT pg_backend_pid() AS \"pid!\"") + .fetch_one(&mut *access) + .await + .unwrap(); + let db = pool.clone(); + let mut revoke = tokio::spawn(async move { + sqlx::query!("UPDATE users SET active = false WHERE id = $1", ids.user_id) + .execute(&db) + .await + .unwrap(); + }); + tokio::select! { + result = &mut revoke => panic!("revocation completed before access transaction committed: {result:?}"), + () = wait_for_blocked(&pool, blocker) => {}, + } + access.commit().await.unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(10), revoke) + .await + .unwrap() + .unwrap(); + let result = begin_user_access_change(&pool, ids.org_id, ids.user_id).await; + assert!( + matches!( + result, + Err(ServerFnError::ServerError { + code: FORBIDDEN, + .. + }) + ), + "{result:?}" + ); +} diff --git a/crates/horae/src/server_fns/users/tests/concurrency.rs b/crates/horae/src/server_fns/users/tests/concurrency.rs index 47b381ef..f89226dc 100644 --- a/crates/horae/src/server_fns/users/tests/concurrency.rs +++ b/crates/horae/src/server_fns/users/tests/concurrency.rs @@ -13,14 +13,15 @@ enum Change { async fn remove_admin( pool: &PgPool, org: Uuid, + actor: Uuid, user: Uuid, change: Change, ) -> Result { match change { - Change::Demote => change_user_role(pool, org, user, OrgRole::Member) + Change::Demote => change_user_role(pool, org, actor, user, OrgRole::Member) .await .map(|(user, _)| user), - Change::Deactivate => change_user_active(pool, org, user, false) + Change::Deactivate => change_user_active(pool, org, actor, user, false) .await .map(|(user, _)| user), } @@ -58,7 +59,8 @@ async fn concurrent_changes_keep_an_admin(pool: &PgPool, first: Change, second: let mut changes = JoinSet::new(); let db = pool.clone(); - changes.spawn(async move { remove_admin(&db, ids.org_id, ids.user_id, first).await }); + changes + .spawn(async move { remove_admin(&db, ids.org_id, ids.user_id, ids.user_id, first).await }); timeout(std::time::Duration::from_secs(10), async { while blocked_connections(pool).await < 1 { @@ -69,7 +71,7 @@ async fn concurrent_changes_keep_an_admin(pool: &PgPool, first: Change, second: .expect("first change must reach the held user lock"); let db = pool.clone(); - changes.spawn(async move { remove_admin(&db, ids.org_id, other, second).await }); + changes.spawn(async move { remove_admin(&db, ids.org_id, other, other, second).await }); // The second request either waits behind the first transaction or commits // while the first is paused. Observe that boundary, not a timing assumption. @@ -139,7 +141,7 @@ async fn the_last_admin_cannot_be_demoted_or_deactivated(pool: PgPool) { let ids = seed(&pool, OrgRole::Admin).await; for change in [Change::Demote, Change::Deactivate] { assert!(matches!( - remove_admin(&pool, ids.org_id, ids.user_id, change).await, + remove_admin(&pool, ids.org_id, ids.user_id, ids.user_id, change).await, Err(ServerFnError::ServerError { code: CONFLICT, .. }) )); } @@ -148,11 +150,12 @@ async fn the_last_admin_cannot_be_demoted_or_deactivated(pool: PgPool) { #[sqlx::test(migrations = "./migrations")] async fn unchanged_admin_access_is_allowed(pool: PgPool) { let ids = seed(&pool, OrgRole::Admin).await; - let (user, role) = change_user_role(&pool, ids.org_id, ids.user_id, OrgRole::Admin) - .await - .unwrap(); + let (user, role) = + change_user_role(&pool, ids.org_id, ids.user_id, ids.user_id, OrgRole::Admin) + .await + .unwrap(); assert_eq!(role, Some(user.org_role)); - let (user, active) = change_user_active(&pool, ids.org_id, ids.user_id, true) + let (user, active) = change_user_active(&pool, ids.org_id, ids.user_id, ids.user_id, true) .await .unwrap(); assert_eq!(active, Some(user.active)); @@ -164,7 +167,7 @@ async fn access_changes_cannot_target_another_organization(pool: PgPool) { let foreign = seed(&pool, OrgRole::Admin).await; for change in [Change::Demote, Change::Deactivate] { assert!(matches!( - remove_admin(&pool, ids.org_id, foreign.user_id, change).await, + remove_admin(&pool, ids.org_id, ids.user_id, foreign.user_id, change).await, Err(ServerFnError::ServerError { code: NOT_FOUND, .. From 5b96629521ca31b3232740ac020db007bdbcf858 Mon Sep 17 00:00:00 2001 From: Aldo Borrero <82811+aldoborrero@users.noreply.github.com> Date: Thu, 8 Oct 2026 21:35:58 +0200 Subject: [PATCH 2/2] Keep last-admin checks current after transaction lock waits --- crates/horae/src/server_fns/users.rs | 6 ++++ .../src/server_fns/users/tests/concurrency.rs | 29 +++++++++++++++++++ 2 files changed, 35 insertions(+) diff --git a/crates/horae/src/server_fns/users.rs b/crates/horae/src/server_fns/users.rs index a28c3d05..72ffdaef 100644 --- a/crates/horae/src/server_fns/users.rs +++ b/crates/horae/src/server_fns/users.rs @@ -93,6 +93,12 @@ async fn begin_user_access_change( actor_id: uuid::Uuid, ) -> Result, ServerFnError> { let mut tx = db.begin().await.map_err(server_err)?; + // The last-admin count must include changes committed while the + // organization lock was awaited, regardless of the connection default. + sqlx::query!("SET TRANSACTION ISOLATION LEVEL READ COMMITTED") + .execute(&mut *tx) + .await + .map_err(server_err)?; // Both role and activation changes must serialize before counting admins; // locking only the target user lets two admins remove each other. sqlx::query!( diff --git a/crates/horae/src/server_fns/users/tests/concurrency.rs b/crates/horae/src/server_fns/users/tests/concurrency.rs index f89226dc..69dca6d7 100644 --- a/crates/horae/src/server_fns/users/tests/concurrency.rs +++ b/crates/horae/src/server_fns/users/tests/concurrency.rs @@ -136,6 +136,35 @@ async fn concurrent_deactivations_keep_an_active_admin(pool: PgPool) { concurrent_changes_keep_an_admin(&pool, Change::Deactivate, Change::Deactivate).await; } +#[sqlx::test(migrations = "./migrations")] +#[serial_test::serial] +async fn repeatable_read_default_cannot_remove_every_administrator(pool: PgPool) { + let requests = sqlx::postgres::PgPoolOptions::new() + .max_connections(5) + .after_connect(|connection, _| { + Box::pin(async move { + sqlx::query!( + "SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL REPEATABLE READ" + ) + .execute(connection) + .await?; + Ok(()) + }) + }) + .connect_with((*pool.connect_options()).clone()) + .await + .unwrap(); + for (first, second) in [ + (Change::Demote, Change::Demote), + (Change::Deactivate, Change::Deactivate), + (Change::Demote, Change::Deactivate), + (Change::Deactivate, Change::Demote), + ] { + concurrent_changes_keep_an_admin(&requests, first, second).await; + } + requests.close().await; +} + #[sqlx::test(migrations = "./migrations")] async fn the_last_admin_cannot_be_demoted_or_deactivated(pool: PgPool) { let ids = seed(&pool, OrgRole::Admin).await;