From 2e7d62bab1c29e6b3838d509a0431ece3456acf2 Mon Sep 17 00:00:00 2001 From: Floating-Y <118035379+Floating-Y@users.noreply.github.com> Date: Fri, 16 Oct 2026 16:56:06 +0800 Subject: [PATCH 1/2] Keep an active system administrator when account removals overlap Signed-off-by: Floating-Y <118035379+Floating-Y@users.noreply.github.com> --- .../src/api/system_admin_tests.rs | 47 ++++++-- crates/utopia-store/src/accounts.rs | 42 ++++++- .../tests/store/a_retired_account.rs | 109 ++++++++++++++++++ .../store/an_admin_grants_system_admin.rs | 15 ++- 4 files changed, 200 insertions(+), 13 deletions(-) diff --git a/crates/utopia-server/src/api/system_admin_tests.rs b/crates/utopia-server/src/api/system_admin_tests.rs index c273f95c9..726c188ab 100644 --- a/crates/utopia-server/src/api/system_admin_tests.rs +++ b/crates/utopia-server/src/api/system_admin_tests.rs @@ -1,8 +1,8 @@ -//! 授予系统管理员走的那条接口：只有系统管理员能调，调了留审计。 +//! 管理员变更留审计，既有会话也在下一次请求重新检查权限。 use crate::state::AppState; use axum::{ body::Body, - http::{Request, StatusCode}, + http::{Method, Request, StatusCode}, }; use serde_json::json; use std::sync::Arc; @@ -10,7 +10,8 @@ use tower::ServiceExt; use uuid::Uuid; #[tokio::test] -async fn only_a_system_admin_grants_the_flag_and_the_grant_is_audited() -> anyhow::Result<()> { +async fn admin_changes_are_audited_and_existing_sessions_recheck_permissions() -> anyhow::Result<()> +{ let Some(url) = utopia_store::test_db::url() else { return Ok(()); }; @@ -39,16 +40,17 @@ async fn only_a_system_admin_grants_the_flag_and_the_grant_is_audited() -> anyho )) .execute(&pool) .await?; - let grant = |caller: Uuid| { + let admin_token = crate::auth::issue_token(&state, admin)?; + let member_token = crate::auth::issue_token(&state, member)?; + let change = |token: String, method: Method, is_admin: bool| { let state = state.clone(); async move { - let token = crate::auth::issue_token(&state, caller)?; let request = Request::builder() - .method("PUT") + .method(method) .uri(format!("/api/v1/admin/users/{member}")) .header("authorization", format!("Bearer {token}")) .header("content-type", "application/json") - .body(Body::from(json!({ "is_admin": true }).to_string()))?; + .body(Body::from(json!({ "is_admin": is_admin }).to_string()))?; let response = super::router(state, &Default::default()) .oneshot(request) .await?; @@ -56,8 +58,10 @@ async fn only_a_system_admin_grants_the_flag_and_the_grant_is_audited() -> anyho } }; - anyhow::ensure!(grant(member).await? == StatusCode::FORBIDDEN); - anyhow::ensure!(grant(admin).await? == StatusCode::OK); + anyhow::ensure!( + change(member_token.clone(), Method::PUT, true).await? == StatusCode::FORBIDDEN + ); + anyhow::ensure!(change(admin_token.clone(), Method::PUT, true).await? == StatusCode::OK); let flag: bool = sqlx::query_scalar("SELECT is_admin FROM users WHERE id = $1") .bind(member) @@ -72,6 +76,31 @@ async fn only_a_system_admin_grants_the_flag_and_the_grant_is_audited() -> anyho .fetch_one(&pool) .await?; anyhow::ensure!(flag && audited == 1, "flag={flag} audited={audited}"); + anyhow::ensure!(change(member_token.clone(), Method::PUT, true).await? == StatusCode::OK); + anyhow::ensure!(change(admin_token.clone(), Method::PUT, false).await? == StatusCode::OK); + anyhow::ensure!( + change(member_token.clone(), Method::PUT, true).await? == StatusCode::FORBIDDEN + ); + anyhow::ensure!(change(admin_token.clone(), Method::PUT, true).await? == StatusCode::OK); + anyhow::ensure!(change(admin_token, Method::DELETE, false).await? == StatusCode::OK); + anyhow::ensure!(change(member_token, Method::PUT, true).await? == StatusCode::UNAUTHORIZED); + + let changes: Vec<(String, i64)> = sqlx::query_as( + "SELECT action, count(*) FROM audit_events + WHERE actor_id = $1 AND target_id = $2 GROUP BY action ORDER BY action", + ) + .bind(admin) + .bind(member) + .fetch_all(&pool) + .await?; + assert_eq!( + changes, + vec![ + ("user.admin_granted".into(), 2), + ("user.admin_revoked".into(), 1), + ("user.deactivated".into(), 1), + ] + ); anyhow::Ok(()) } .await; diff --git a/crates/utopia-store/src/accounts.rs b/crates/utopia-store/src/accounts.rs index b665ed6c2..c02a215c2 100644 --- a/crates/utopia-store/src/accounts.rs +++ b/crates/utopia-store/src/accounts.rs @@ -292,6 +292,7 @@ pub async fn deactivate_user(pool: &PgPool, target: Uuid, actor: Uuid) -> AppRes return Err(AppError::Validation("不能停用自己的账号".into())); } let mut tx = pool.begin().await?; + lock_account_organization(&mut tx, target).await?; let victim: Option<(bool, Uuid, Option>)> = sqlx::query_as("SELECT is_admin, org_id, deactivated_at FROM users WHERE id = $1") .bind(target) @@ -360,18 +361,57 @@ pub async fn set_system_admin( if target == actor && !value { return Err(AppError::Validation("不能收回自己的系统管理员".into())); } + let mut tx = pool.begin().await?; + let org_id = lock_account_organization(&mut tx, target).await?; + let is_admin: bool = + sqlx::query_scalar("SELECT is_admin FROM users WHERE id = $1 AND deactivated_at IS NULL") + .bind(target) + .fetch_optional(&mut *tx) + .await? + .ok_or(AppError::NotFound)?; + if is_admin && !value { + let others: i64 = sqlx::query_scalar( + "SELECT count(*) FROM users + WHERE org_id = $1 AND is_admin AND deactivated_at IS NULL AND id <> $2", + ) + .bind(org_id) + .bind(target) + .fetch_one(&mut *tx) + .await?; + if others == 0 { + return Err(AppError::Validation( + "这是组织里最后一个管理员，收回权限之后没人能管成员".into(), + )); + } + } let res = sqlx::query("UPDATE users SET is_admin = $2 WHERE id = $1 AND deactivated_at IS NULL") .bind(target) .bind(value) - .execute(pool) + .execute(&mut *tx) .await?; if res.rows_affected() == 0 { return Err(AppError::NotFound); } + tx.commit().await?; Ok(()) } +async fn lock_account_organization( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + target: Uuid, +) -> AppResult { + // 撤权与停用共用组织锁；拿锁后再读账号、数管理员，不能信任请求鉴权时的快照。 + sqlx::query_scalar( + "SELECT id FROM organizations + WHERE id = (SELECT org_id FROM users WHERE id = $1) FOR UPDATE", + ) + .bind(target) + .fetch_optional(&mut **tx) + .await? + .ok_or(AppError::NotFound) +} + /// 单点登录允许自动开户时，用身份提供方核实过的 email 找账号；找不到就按注册的 /// 规则开一个（默认工作区的 viewer；部署里还没人时就是首个管理员），并把这个 /// subject 绑上。开户与绑定同一个事务：绑定撞车（`taken` / `already_linked`）时 diff --git a/crates/utopia-store/tests/store/a_retired_account.rs b/crates/utopia-store/tests/store/a_retired_account.rs index fd4374c81..daaaa28c7 100644 --- a/crates/utopia-store/tests/store/a_retired_account.rs +++ b/crates/utopia-store/tests/store/a_retired_account.rs @@ -130,3 +130,112 @@ async fn the_last_admin_and_oneself_are_protected() -> anyhow::Result<()> { .await?; run } + +async fn concurrent_admin_removal(deactivate: [bool; 2]) -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + let (org, a, b) = org_with_two_admins(&pool).await?; + let run = async { + // Both requests passed AuthUser before either changes the other's account. + for actor in [a, b] { + assert!( + utopia_store::accounts::find_user_by_id(&pool, actor) + .await? + .unwrap() + .is_admin + ); + } + let mut gate = pool.begin().await?; + sqlx::query("SELECT id FROM users WHERE org_id = $1 FOR UPDATE") + .bind(org) + .execute(&mut *gate) + .await?; + let mut tasks = tokio::task::JoinSet::new(); + let mut pids = Vec::new(); + for ((actor, target), retire) in [(a, b), (b, a)].into_iter().zip(deactivate) { + let writer = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect(&url) + .await?; + pids.push( + sqlx::query_scalar::<_, i32>("SELECT pg_backend_pid()") + .fetch_one(&writer) + .await?, + ); + tasks.spawn(async move { + if retire { + utopia_store::accounts::deactivate_user(&writer, target, actor).await + } else { + utopia_store::accounts::set_system_admin(&writer, target, actor, false).await + } + }); + } + // Observe real lock waits, not elapsed time: before the fix both writers + // have passed their checks; afterwards one waits on the organization lock. + let waiting = tokio::time::timeout(std::time::Duration::from_secs(10), async { + loop { + let blocked: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_stat_activity + WHERE pid = ANY($1) AND cardinality(pg_blocking_pids(pid)) > 0", + ) + .bind(&pids) + .fetch_one(&pool) + .await?; + if blocked == 2 { + break anyhow::Ok(()); + } + tokio::task::yield_now().await; + } + }) + .await; + gate.commit().await?; + waiting??; + let mut succeeded = 0; + while let Some(result) = + tokio::time::timeout(std::time::Duration::from_secs(10), tasks.join_next()).await? + { + match result? { + Ok(()) => succeeded += 1, + Err(utopia_core::AppError::Validation(_)) => {} + Err(error) => return Err(error.into()), + } + } + let remaining: i64 = sqlx::query_scalar( + "SELECT count(*) FROM users + WHERE org_id = $1 AND is_admin AND deactivated_at IS NULL", + ) + .bind(org) + .fetch_one(&pool) + .await?; + anyhow::Ok((succeeded, remaining)) + } + .await; + sqlx::query("DELETE FROM organizations WHERE id = $1") + .bind(org) + .execute(&pool) + .await?; + let (succeeded, remaining) = run?; + assert_eq!( + remaining, 1, + "concurrent removals must retain an active administrator" + ); + assert_eq!(succeeded, 1, "exactly one removal must succeed"); + Ok(()) +} + +#[tokio::test] +async fn concurrent_revocations_keep_an_active_admin() -> anyhow::Result<()> { + concurrent_admin_removal([false, false]).await +} + +#[tokio::test] +async fn concurrent_deactivations_keep_an_active_admin() -> anyhow::Result<()> { + concurrent_admin_removal([true, true]).await +} + +#[tokio::test] +async fn concurrent_revocation_and_deactivation_keep_an_active_admin() -> anyhow::Result<()> { + concurrent_admin_removal([false, true]).await +} diff --git a/crates/utopia-store/tests/store/an_admin_grants_system_admin.rs b/crates/utopia-store/tests/store/an_admin_grants_system_admin.rs index c1b733648..553bfe1c4 100644 --- a/crates/utopia-store/tests/store/an_admin_grants_system_admin.rs +++ b/crates/utopia-store/tests/store/an_admin_grants_system_admin.rs @@ -44,6 +44,13 @@ async fn granting_a_member_makes_them_admin() -> anyhow::Result<()> { let run = async { utopia_store::accounts::set_system_admin(&pool, member, admin, true).await?; assert!(is_admin(&pool, member).await?); + utopia_store::accounts::set_system_admin(&pool, member, admin, false).await?; + assert!(!is_admin(&pool, member).await?); + // A request admitted before the actor's revocation still cannot remove + // the last administrator. Repeating a non-admin's revocation is harmless. + let refused = utopia_store::accounts::set_system_admin(&pool, admin, member, false).await; + assert!(matches!(refused, Err(utopia_core::AppError::Validation(_)))); + utopia_store::accounts::set_system_admin(&pool, member, admin, false).await?; Ok::<_, anyhow::Error>(()) } .await; @@ -90,9 +97,11 @@ async fn granting_a_deactivated_or_unknown_user_is_not_found() -> anyhow::Result let run = async { utopia_store::accounts::deactivate_user(&pool, member, admin).await?; for target in [member, Uuid::now_v7()] { - let missing = - utopia_store::accounts::set_system_admin(&pool, target, admin, true).await; - assert!(matches!(missing, Err(utopia_core::AppError::NotFound))); + for value in [true, false] { + let missing = + utopia_store::accounts::set_system_admin(&pool, target, admin, value).await; + assert!(matches!(missing, Err(utopia_core::AppError::NotFound))); + } } assert!(!is_admin(&pool, member).await?); Ok::<_, anyhow::Error>(()) From fd78ef03ba9108379c32e5e7f9a1032b9cdd6ee8 Mon Sep 17 00:00:00 2001 From: Floating-Y <118035379+Floating-Y@users.noreply.github.com> Date: Fri, 16 Oct 2026 20:53:49 +0800 Subject: [PATCH 2/2] Serialize administrator removals without blocking foreign key checks Signed-off-by: Floating-Y <118035379+Floating-Y@users.noreply.github.com> --- .../src/api/system_admin_tests.rs | 47 ++++--------------- crates/utopia-store/src/accounts.rs | 2 +- .../tests/store/a_retired_account.rs | 9 ---- 3 files changed, 10 insertions(+), 48 deletions(-) diff --git a/crates/utopia-server/src/api/system_admin_tests.rs b/crates/utopia-server/src/api/system_admin_tests.rs index 726c188ab..c273f95c9 100644 --- a/crates/utopia-server/src/api/system_admin_tests.rs +++ b/crates/utopia-server/src/api/system_admin_tests.rs @@ -1,8 +1,8 @@ -//! 管理员变更留审计，既有会话也在下一次请求重新检查权限。 +//! 授予系统管理员走的那条接口：只有系统管理员能调，调了留审计。 use crate::state::AppState; use axum::{ body::Body, - http::{Method, Request, StatusCode}, + http::{Request, StatusCode}, }; use serde_json::json; use std::sync::Arc; @@ -10,8 +10,7 @@ use tower::ServiceExt; use uuid::Uuid; #[tokio::test] -async fn admin_changes_are_audited_and_existing_sessions_recheck_permissions() -> anyhow::Result<()> -{ +async fn only_a_system_admin_grants_the_flag_and_the_grant_is_audited() -> anyhow::Result<()> { let Some(url) = utopia_store::test_db::url() else { return Ok(()); }; @@ -40,17 +39,16 @@ async fn admin_changes_are_audited_and_existing_sessions_recheck_permissions() - )) .execute(&pool) .await?; - let admin_token = crate::auth::issue_token(&state, admin)?; - let member_token = crate::auth::issue_token(&state, member)?; - let change = |token: String, method: Method, is_admin: bool| { + let grant = |caller: Uuid| { let state = state.clone(); async move { + let token = crate::auth::issue_token(&state, caller)?; let request = Request::builder() - .method(method) + .method("PUT") .uri(format!("/api/v1/admin/users/{member}")) .header("authorization", format!("Bearer {token}")) .header("content-type", "application/json") - .body(Body::from(json!({ "is_admin": is_admin }).to_string()))?; + .body(Body::from(json!({ "is_admin": true }).to_string()))?; let response = super::router(state, &Default::default()) .oneshot(request) .await?; @@ -58,10 +56,8 @@ async fn admin_changes_are_audited_and_existing_sessions_recheck_permissions() - } }; - anyhow::ensure!( - change(member_token.clone(), Method::PUT, true).await? == StatusCode::FORBIDDEN - ); - anyhow::ensure!(change(admin_token.clone(), Method::PUT, true).await? == StatusCode::OK); + anyhow::ensure!(grant(member).await? == StatusCode::FORBIDDEN); + anyhow::ensure!(grant(admin).await? == StatusCode::OK); let flag: bool = sqlx::query_scalar("SELECT is_admin FROM users WHERE id = $1") .bind(member) @@ -76,31 +72,6 @@ async fn admin_changes_are_audited_and_existing_sessions_recheck_permissions() - .fetch_one(&pool) .await?; anyhow::ensure!(flag && audited == 1, "flag={flag} audited={audited}"); - anyhow::ensure!(change(member_token.clone(), Method::PUT, true).await? == StatusCode::OK); - anyhow::ensure!(change(admin_token.clone(), Method::PUT, false).await? == StatusCode::OK); - anyhow::ensure!( - change(member_token.clone(), Method::PUT, true).await? == StatusCode::FORBIDDEN - ); - anyhow::ensure!(change(admin_token.clone(), Method::PUT, true).await? == StatusCode::OK); - anyhow::ensure!(change(admin_token, Method::DELETE, false).await? == StatusCode::OK); - anyhow::ensure!(change(member_token, Method::PUT, true).await? == StatusCode::UNAUTHORIZED); - - let changes: Vec<(String, i64)> = sqlx::query_as( - "SELECT action, count(*) FROM audit_events - WHERE actor_id = $1 AND target_id = $2 GROUP BY action ORDER BY action", - ) - .bind(admin) - .bind(member) - .fetch_all(&pool) - .await?; - assert_eq!( - changes, - vec![ - ("user.admin_granted".into(), 2), - ("user.admin_revoked".into(), 1), - ("user.deactivated".into(), 1), - ] - ); anyhow::Ok(()) } .await; diff --git a/crates/utopia-store/src/accounts.rs b/crates/utopia-store/src/accounts.rs index c02a215c2..486d95dec 100644 --- a/crates/utopia-store/src/accounts.rs +++ b/crates/utopia-store/src/accounts.rs @@ -404,7 +404,7 @@ async fn lock_account_organization( // 撤权与停用共用组织锁；拿锁后再读账号、数管理员，不能信任请求鉴权时的快照。 sqlx::query_scalar( "SELECT id FROM organizations - WHERE id = (SELECT org_id FROM users WHERE id = $1) FOR UPDATE", + WHERE id = (SELECT org_id FROM users WHERE id = $1) FOR NO KEY UPDATE", ) .bind(target) .fetch_optional(&mut **tx) diff --git a/crates/utopia-store/tests/store/a_retired_account.rs b/crates/utopia-store/tests/store/a_retired_account.rs index daaaa28c7..ce84193ac 100644 --- a/crates/utopia-store/tests/store/a_retired_account.rs +++ b/crates/utopia-store/tests/store/a_retired_account.rs @@ -138,15 +138,6 @@ async fn concurrent_admin_removal(deactivate: [bool; 2]) -> anyhow::Result<()> { let pool = PgPool::connect(&url).await?; let (org, a, b) = org_with_two_admins(&pool).await?; let run = async { - // Both requests passed AuthUser before either changes the other's account. - for actor in [a, b] { - assert!( - utopia_store::accounts::find_user_by_id(&pool, actor) - .await? - .unwrap() - .is_admin - ); - } let mut gate = pool.begin().await?; sqlx::query("SELECT id FROM users WHERE org_id = $1 FOR UPDATE") .bind(org)