Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 41 additions & 1 deletion crates/utopia-store/src/accounts.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<chrono::DateTime<chrono::Utc>>)> =
sqlx::query_as("SELECT is_admin, org_id, deactivated_at FROM users WHERE id = $1")
.bind(target)
Expand Down Expand Up @@ -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<Uuid> {
// 撤权与停用共用组织锁;拿锁后再读账号、数管理员,不能信任请求鉴权时的快照。
sqlx::query_scalar(
"SELECT id FROM organizations
WHERE id = (SELECT org_id FROM users WHERE id = $1) FOR NO KEY UPDATE",
)
.bind(target)
.fetch_optional(&mut **tx)
.await?
.ok_or(AppError::NotFound)
}

/// 单点登录允许自动开户时,用身份提供方核实过的 email 找账号;找不到就按注册的
/// 规则开一个(默认工作区的 viewer;部署里还没人时就是首个管理员),并把这个
/// subject 绑上。开户与绑定同一个事务:绑定撞车(`taken` / `already_linked`)时
Expand Down
100 changes: 100 additions & 0 deletions crates/utopia-store/tests/store/a_retired_account.rs
Original file line number Diff line number Diff line change
Expand Up @@ -130,3 +130,103 @@ 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 {
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
}
15 changes: 12 additions & 3 deletions crates/utopia-store/tests/store/an_admin_grants_system_admin.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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>(())
Expand Down
Loading