fix(identity): make private huddle admission atomic

Signed-off-by: Cea Stapleton Cordasco <261786559+cea-block@users.noreply.github.com>
This commit is contained in:
Cea Stapleton Cordasco
2026-07-31 08:26:05 -05:00
parent 8921f6d7a8
commit af657ea23b
4 changed files with 656 additions and 69 deletions

View File

@@ -9,6 +9,7 @@ use sqlx::{PgPool, Postgres, Row, Transaction};
use uuid::Uuid;
use crate::error::{DbError, Result};
use crate::identity_binding::{BindIdentityResult, IdentityBindingConflict, IdentityBindingInput};
use buzz_core::CommunityId;
// Re-export the canonical enum definitions from buzz-core.
@@ -385,12 +386,7 @@ pub async fn add_member(
role: MemberRole,
invited_by: Option<&[u8]>,
) -> Result<MemberRecord> {
if pubkey.len() != 32 {
return Err(DbError::InvalidData(format!(
"pubkey must be 32 bytes, got {}",
pubkey.len()
)));
}
validate_member_pubkey(pubkey)?;
let mut tx = pool.begin().await?;
@@ -398,7 +394,105 @@ pub async fn add_member(
// sequence against concurrent membership writes on this channel.
acquire_channel_membership_lock(&mut tx, community_id, channel_id).await?;
let channel = get_channel_tx(&mut tx, community_id, channel_id).await?;
let record = add_member_tx(&mut tx, community_id, channel_id, pubkey, role, invited_by).await?;
tx.commit().await?;
Ok(record)
}
/// Outcome of atomically adding a channel member and binding corporate identity.
#[derive(Debug, Clone)]
pub enum ChannelAdmissionOutcome {
/// Membership and any staged identity binding committed together.
Joined {
/// The committed membership row.
member: MemberRecord,
/// Binding committed in the same transaction, when one was staged.
identity_binding: Option<BindIdentityResult>,
},
/// The staged identity conflicts with an active binding.
IdentityConflict(IdentityBindingConflict),
/// The staged identity principal or key is revoked.
IdentityRevoked,
}
/// Add a channel member and optional corporate identity binding in one transaction.
pub async fn add_member_with_identity(
pool: &PgPool,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
role: MemberRole,
invited_by: Option<&[u8]>,
identity: Option<&IdentityBindingInput<'_>>,
) -> Result<ChannelAdmissionOutcome> {
validate_member_pubkey(pubkey)?;
if identity.is_some_and(|identity| identity.pubkey != pubkey) {
return Err(DbError::InvalidData(
"channel membership pubkey does not match staged identity key".to_string(),
));
}
let mut tx = pool.begin().await?;
// Keep this first: every channel membership writer shares this lock order.
acquire_channel_membership_lock(&mut tx, community_id, channel_id).await?;
let member =
match add_member_tx(&mut tx, community_id, channel_id, pubkey, role, invited_by).await {
Ok(member) => member,
Err(error) => {
tx.rollback().await?;
return Err(error);
}
};
let identity_binding = if let Some(identity) = identity {
match crate::identity_binding::bind_or_validate_identity_tx(&mut tx, community_id, identity)
.await
{
Ok(binding @ (BindIdentityResult::Created | BindIdentityResult::Matched)) => {
Some(binding)
}
Ok(BindIdentityResult::Conflict(conflict)) => {
tx.rollback().await?;
return Ok(ChannelAdmissionOutcome::IdentityConflict(conflict));
}
Ok(BindIdentityResult::Revoked) => {
tx.rollback().await?;
return Ok(ChannelAdmissionOutcome::IdentityRevoked);
}
Err(error) => {
tx.rollback().await?;
return Err(error);
}
}
} else {
None
};
tx.commit().await?;
Ok(ChannelAdmissionOutcome::Joined {
member,
identity_binding,
})
}
fn validate_member_pubkey(pubkey: &[u8]) -> Result<()> {
if pubkey.len() != 32 {
return Err(DbError::InvalidData(format!(
"pubkey must be 32 bytes, got {}",
pubkey.len()
)));
}
Ok(())
}
async fn add_member_tx(
tx: &mut Transaction<'_, Postgres>,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
role: MemberRole,
invited_by: Option<&[u8]>,
) -> Result<MemberRecord> {
let channel = get_channel_tx(tx, community_id, channel_id).await?;
let effective_role = if channel.visibility == "private" {
let inviter = invited_by.ok_or_else(|| {
@@ -409,7 +503,7 @@ pub async fn add_member(
let is_creator_bootstrap = inviter == pubkey && inviter == channel.created_by.as_slice();
if !is_creator_bootstrap {
let inviter_role_str = get_active_role_tx(&mut tx, community_id, channel_id, inviter)
let inviter_role_str = get_active_role_tx(tx, community_id, channel_id, inviter)
.await?
.ok_or_else(|| {
DbError::AccessDenied("inviter is not an active member".to_string())
@@ -433,7 +527,7 @@ pub async fn add_member(
// elevated roles. Self-join always gets Member.
if role.is_elevated() {
let granter_role = match invited_by {
Some(inv) => get_active_role_tx(&mut tx, community_id, channel_id, inv).await?,
Some(inv) => get_active_role_tx(tx, community_id, channel_id, inv).await?,
None => None,
};
match granter_role.as_deref() {
@@ -465,10 +559,10 @@ pub async fn add_member(
// current authority from a removed row would make soft-deleted ownership a
// resurrection token: an owner removed by another owner could self-rejoin
// via kind:9021 (`Member, None`) and silently regain ownership.
let current_role = get_active_role_tx(&mut tx, community_id, channel_id, pubkey).await?;
let current_role = get_active_role_tx(tx, community_id, channel_id, pubkey).await?;
if let Some(current_role) = current_role.filter(|r| r != effective_role.as_str()) {
let actor_role = match invited_by {
Some(inviter) => get_active_role_tx(&mut tx, community_id, channel_id, inviter).await?,
Some(inviter) => get_active_role_tx(tx, community_id, channel_id, inviter).await?,
None => None,
};
let actor_role: Option<MemberRole> = actor_role.and_then(|r| r.parse().ok());
@@ -488,7 +582,7 @@ pub async fn add_member(
)
.bind(community_id.as_uuid())
.bind(channel_id)
.fetch_one(&mut *tx)
.fetch_one(&mut **tx)
.await?;
let owner_count: i64 = row.try_get("cnt")?;
if owner_count <= 1 {
@@ -514,7 +608,7 @@ pub async fn add_member(
.bind(pubkey)
.bind(effective_role.as_str())
.bind(invited_by)
.execute(&mut *tx)
.execute(&mut **tx)
.await?;
let row = sqlx::query(
@@ -526,11 +620,10 @@ pub async fn add_member(
.bind(community_id.as_uuid())
.bind(channel_id)
.bind(pubkey)
.fetch_one(&mut *tx)
.fetch_one(&mut **tx)
.await?;
let record = row_to_member_record(row)?;
tx.commit().await?;
Ok(record)
}
@@ -1530,7 +1623,9 @@ mod tests {
const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz";
async fn setup_pool() -> PgPool {
PgPool::connect(TEST_DB_URL)
let database_url =
std::env::var("BUZZ_TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.to_string());
PgPool::connect(&database_url)
.await
.expect("connect to test DB")
}
@@ -1602,6 +1697,388 @@ mod tests {
get_channel(pool, CommunityId::from_uuid(community_id), id).await
}
fn identity_for<'a>(pubkey: &'a [u8], uid: &'a str) -> IdentityBindingInput<'a> {
IdentityBindingInput {
issuer: "https://idp.example",
uid,
pubkey,
display_name: Some("private@example.com"),
source: crate::identity_binding::SOURCE_JWT_NPUB,
}
}
async fn trusted_assertion_count(pool: &PgPool, community: CommunityId) -> i64 {
sqlx::query_scalar("SELECT COUNT(*) FROM events WHERE community_id = $1 AND kind = $2")
.bind(community.as_uuid())
.bind(buzz_core::kind::KIND_USER_TRUSTED_ASSERTION as i32)
.fetch_one(pool)
.await
.expect("trusted assertion count")
}
async fn active_membership_count(
pool: &PgPool,
community: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
) -> i64 {
sqlx::query_scalar(
"SELECT COUNT(*) FROM channel_members \
WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3 \
AND removed_at IS NULL",
)
.bind(community.as_uuid())
.bind(channel_id)
.bind(pubkey)
.fetch_one(pool)
.await
.expect("active membership count")
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn atomic_huddle_admission_membership_failure_leaves_no_identity_state() {
let pool = setup_pool().await;
let community_id = make_test_community(&pool).await;
let community = CommunityId::from_uuid(community_id);
let owner = random_pubkey();
let non_member_inviter = random_pubkey();
let joiner = random_pubkey();
let channel = create_test_channel(
&pool,
community_id,
"atomic-membership-failure",
ChannelType::Stream,
ChannelVisibility::Private,
None,
&owner,
Some(3600),
)
.await
.expect("create private huddle");
let identity = identity_for(&joiner, "membership-failure");
let error = add_member_with_identity(
&pool,
community,
channel.id,
&joiner,
MemberRole::Member,
Some(&non_member_inviter),
Some(&identity),
)
.await
.expect_err("non-member inviter must fail admission");
assert!(matches!(error, DbError::AccessDenied(_)), "{error:?}");
assert_eq!(
active_membership_count(&pool, community, channel.id, &joiner).await,
0
);
assert!(
crate::identity_binding::get_active_identity_binding_by_pubkey(
&pool, community, &joiner,
)
.await
.expect("binding lookup")
.is_none()
);
assert_eq!(trusted_assertion_count(&pool, community).await, 0);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn atomic_huddle_admission_identity_conflict_rolls_back_membership() {
let pool = setup_pool().await;
let community_id = make_test_community(&pool).await;
let community = CommunityId::from_uuid(community_id);
let owner = random_pubkey();
let bound_key = random_pubkey();
let joiner = random_pubkey();
let channel = create_test_channel(
&pool,
community_id,
"atomic-identity-conflict",
ChannelType::Stream,
ChannelVisibility::Private,
None,
&owner,
Some(3600),
)
.await
.expect("create private huddle");
crate::identity_binding::bind_or_validate_identity(
&pool,
community,
"https://idp.example",
"conflicting-principal",
&bound_key,
Some("bound@example.com"),
crate::identity_binding::SOURCE_JWT_NPUB,
)
.await
.expect("seed conflicting binding");
let identity = identity_for(&joiner, "conflicting-principal");
let outcome = add_member_with_identity(
&pool,
community,
channel.id,
&joiner,
MemberRole::Member,
Some(&owner),
Some(&identity),
)
.await
.expect("typed identity conflict");
assert!(matches!(
outcome,
ChannelAdmissionOutcome::IdentityConflict(_)
));
assert_eq!(
active_membership_count(&pool, community, channel.id, &joiner).await,
0
);
assert_eq!(trusted_assertion_count(&pool, community).await, 0);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn atomic_huddle_admission_identity_storage_failure_rolls_back_membership() {
let pool = setup_pool().await;
let community_id = make_test_community(&pool).await;
let community = CommunityId::from_uuid(community_id);
let owner = random_pubkey();
let joiner = random_pubkey();
let channel = create_test_channel(
&pool,
community_id,
"atomic-identity-storage-failure",
ChannelType::Stream,
ChannelVisibility::Private,
None,
&owner,
Some(3600),
)
.await
.expect("create private huddle");
let suffix = community_id.simple();
let function_name = format!("buzz_test_fail_identity_{suffix}");
let trigger_name = format!("buzz_test_fail_identity_insert_{suffix}");
// Identifiers and the literal UUID below are derived only from a generated UUID.
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE FUNCTION {function_name}() RETURNS trigger LANGUAGE plpgsql AS $$ \
BEGIN RAISE EXCEPTION 'injected identity storage failure'; END $$"
)))
.execute(&pool)
.await
.expect("create failure function");
sqlx::query(sqlx::AssertSqlSafe(format!(
"CREATE TRIGGER {trigger_name} BEFORE INSERT ON identity_bindings \
FOR EACH ROW WHEN (NEW.community_id = '{community_id}'::uuid) \
EXECUTE FUNCTION {function_name}()"
)))
.execute(&pool)
.await
.expect("create failure trigger");
let identity = identity_for(&joiner, "storage-failure");
let result = add_member_with_identity(
&pool,
community,
channel.id,
&joiner,
MemberRole::Member,
Some(&owner),
Some(&identity),
)
.await;
sqlx::query(sqlx::AssertSqlSafe(format!(
"DROP TRIGGER {trigger_name} ON identity_bindings"
)))
.execute(&pool)
.await
.expect("drop failure trigger");
sqlx::query(sqlx::AssertSqlSafe(format!(
"DROP FUNCTION {function_name}()"
)))
.execute(&pool)
.await
.expect("drop failure function");
assert!(matches!(result, Err(DbError::Sqlx(_))), "{result:?}");
assert_eq!(
active_membership_count(&pool, community, channel.id, &joiner).await,
0
);
assert!(
crate::identity_binding::get_active_identity_binding_by_pubkey(
&pool, community, &joiner,
)
.await
.expect("binding lookup")
.is_none()
);
assert_eq!(trusted_assertion_count(&pool, community).await, 0);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn atomic_huddle_admission_success_and_retry_commit_each_row_once() {
let pool = setup_pool().await;
let community_id = make_test_community(&pool).await;
let community = CommunityId::from_uuid(community_id);
let owner = random_pubkey();
let joiner = random_pubkey();
let channel = create_test_channel(
&pool,
community_id,
"atomic-success-retry",
ChannelType::Stream,
ChannelVisibility::Private,
None,
&owner,
Some(3600),
)
.await
.expect("create private huddle");
let identity = identity_for(&joiner, "successful-principal");
let first = add_member_with_identity(
&pool,
community,
channel.id,
&joiner,
MemberRole::Member,
Some(&owner),
Some(&identity),
)
.await
.expect("first admission");
assert!(matches!(
first,
ChannelAdmissionOutcome::Joined {
identity_binding: Some(BindIdentityResult::Created),
..
}
));
let retry = add_member_with_identity(
&pool,
community,
channel.id,
&joiner,
MemberRole::Member,
Some(&owner),
Some(&identity),
)
.await
.expect("idempotent retry");
assert!(matches!(
retry,
ChannelAdmissionOutcome::Joined {
identity_binding: Some(BindIdentityResult::Matched),
..
}
));
assert_eq!(
active_membership_count(&pool, community, channel.id, &joiner).await,
1
);
let binding_count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM identity_bindings \
WHERE community_id = $1 AND pubkey = $2 AND revoked_at IS NULL",
)
.bind(community.as_uuid())
.bind(&joiner)
.fetch_one(&pool)
.await
.expect("binding count");
assert_eq!(binding_count, 1);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn existing_member_and_non_corporate_paths_remain_idempotent() {
let pool = setup_pool().await;
let community_id = make_test_community(&pool).await;
let community = CommunityId::from_uuid(community_id);
let owner = random_pubkey();
let existing_member = random_pubkey();
let non_corporate_joiner = random_pubkey();
let channel = create_test_channel(
&pool,
community_id,
"unchanged-admission-paths",
ChannelType::Stream,
ChannelVisibility::Private,
None,
&owner,
Some(3600),
)
.await
.expect("create private huddle");
add_member(
&pool,
community,
channel.id,
&existing_member,
MemberRole::Member,
Some(&owner),
)
.await
.expect("existing member add");
add_member(
&pool,
community,
channel.id,
&existing_member,
MemberRole::Member,
Some(&owner),
)
.await
.expect("existing member retry");
assert_eq!(
active_membership_count(&pool, community, channel.id, &existing_member).await,
1
);
let outcome = add_member_with_identity(
&pool,
community,
channel.id,
&non_corporate_joiner,
MemberRole::Member,
Some(&owner),
None,
)
.await
.expect("non-corporate admission");
assert!(matches!(
outcome,
ChannelAdmissionOutcome::Joined {
identity_binding: None,
..
}
));
assert_eq!(
active_membership_count(&pool, community, channel.id, &non_corporate_joiner).await,
1
);
assert!(
crate::identity_binding::get_active_identity_binding_by_pubkey(
&pool,
community,
&non_corporate_joiner,
)
.await
.expect("binding lookup")
.is_none()
);
assert_eq!(trusted_assertion_count(&pool, community).await, 0);
}
async fn insert_channel_with_id(
pool: &PgPool,
community_id: Uuid,

View File

@@ -2229,6 +2229,28 @@ impl Db {
.await
}
/// Adds a channel member and optional corporate identity binding atomically.
pub async fn add_member_with_identity(
&self,
community_id: CommunityId,
channel_id: Uuid,
pubkey: &[u8],
role: channel::MemberRole,
invited_by: Option<&[u8]>,
identity: Option<&identity_binding::IdentityBindingInput<'_>>,
) -> Result<channel::ChannelAdmissionOutcome> {
channel::add_member_with_identity(
&self.pool,
community_id,
channel_id,
pubkey,
role,
invited_by,
identity,
)
.await
}
/// Removes a member from a channel.
pub async fn remove_member(
&self,

View File

@@ -329,59 +329,42 @@ async fn handle_active_audio_connection(
}
};
let identity_decision = match crate::corporate_identity::finalize_corporate_identity(
&state,
tenant.community(),
pubkey,
identity_proof,
)
.await
{
Ok(decision) => decision,
Err(e) => {
warn!(channel_id = %channel_id, pubkey = %pubkey_hex, error = %e, "audio: corporate identity finalization denied");
let _ = ws_send
.send(WsMessage::Text(
serde_json::json!({"type": "error", "message": e.public_message()})
.to_string()
.into(),
))
.await;
return;
}
};
crate::corporate_identity::spawn_session_revalidation(
Arc::clone(&state),
tenant.community(),
pubkey,
identity_decision,
cancel.clone(),
);
if let Some(added_by) = auto_add_member_by {
if let Err(e) = state
.db
.add_member(
tenant.community(),
channel_id,
&pubkey_bytes,
MemberRole::Member,
Some(&added_by),
)
.await
// Existing members and open channels retain the established identity path.
// Private-huddle auto-add is deferred until room admission succeeds, then
// membership and direct identity binding commit in one database transaction.
let deferred_private_admission = if let Some(added_by) = auto_add_member_by {
Some((added_by, identity_proof))
} else {
let identity_decision = match crate::corporate_identity::finalize_corporate_identity(
&state,
tenant.community(),
pubkey,
identity_proof,
)
.await
{
warn!(channel_id = %channel_id, pubkey = %pubkey_hex, "audio membership auto-add failed: {e}");
let _ = ws_send
.send(WsMessage::Text(
serde_json::json!({"type":"error","message":"not a member"})
.to_string()
.into(),
))
.await;
cancel.cancel();
return;
}
state.invalidate_membership(&tenant, channel_id, &pubkey_bytes);
}
Ok(decision) => decision,
Err(e) => {
warn!(channel_id = %channel_id, pubkey = %pubkey_hex, error = %e, "audio: corporate identity finalization denied");
let _ = ws_send
.send(WsMessage::Text(
serde_json::json!({"type": "error", "message": e.public_message()})
.to_string()
.into(),
))
.await;
return;
}
};
crate::corporate_identity::spawn_session_revalidation(
Arc::clone(&state),
tenant.community(),
pubkey,
identity_decision,
cancel.clone(),
);
None
};
// Huddle cross-pod routing (mesh) OR single-pod guardrail.
//
@@ -646,6 +629,103 @@ async fn handle_active_audio_connection(
}
};
if let Some((added_by, identity_proof)) = deferred_private_admission {
let identity_input =
crate::corporate_identity::binding_input_for_proof(&identity_proof, &pubkey);
let outcome = state
.db
.add_member_with_identity(
tenant.community(),
channel_id,
&pubkey_bytes,
MemberRole::Member,
Some(&added_by),
identity_input.as_ref(),
)
.await;
let committed_binding = match outcome {
Ok(buzz_db::channel::ChannelAdmissionOutcome::Joined {
identity_binding, ..
}) => identity_binding,
Ok(buzz_db::channel::ChannelAdmissionOutcome::IdentityConflict(conflict)) => Some(
buzz_db::identity_binding::BindIdentityResult::Conflict(conflict),
),
Ok(buzz_db::channel::ChannelAdmissionOutcome::IdentityRevoked) => {
Some(buzz_db::identity_binding::BindIdentityResult::Revoked)
}
Err(e) => {
warn!(channel_id = %channel_id, pubkey = %pubkey_hex, "audio membership auto-add failed: {e}");
let _ = ws_send
.send(WsMessage::Text(
serde_json::json!({"type":"error","message":"not a member"})
.to_string()
.into(),
))
.await;
if let (Some(session), Some(stream)) =
(remote_session.as_ref(), remote_stream.as_mut())
{
crate::audio::join::send_clean_close(
stream,
session.fenced(),
session.pubkey(),
)
.await;
}
room.remove_peer(peer_id);
state
.audio_rooms
.cleanup_if_empty(tenant.community(), channel_id);
return;
}
};
let identity_decision =
match crate::corporate_identity::finalize_atomic_corporate_identity_result(
&state,
tenant.community(),
pubkey,
identity_proof,
committed_binding,
)
.await
{
Ok(decision) => decision,
Err(e) => {
warn!(channel_id = %channel_id, pubkey = %pubkey_hex, error = %e, "audio: corporate identity finalization denied");
let _ = ws_send
.send(WsMessage::Text(
serde_json::json!({"type": "error", "message": e.public_message()})
.to_string()
.into(),
))
.await;
if let (Some(session), Some(stream)) =
(remote_session.as_ref(), remote_stream.as_mut())
{
crate::audio::join::send_clean_close(
stream,
session.fenced(),
session.pubkey(),
)
.await;
}
room.remove_peer(peer_id);
state
.audio_rooms
.cleanup_if_empty(tenant.community(), channel_id);
return;
}
};
crate::corporate_identity::spawn_session_revalidation(
Arc::clone(&state),
tenant.community(),
pubkey,
identity_decision,
cancel.clone(),
);
state.invalidate_membership(&tenant, channel_id, &pubkey_bytes);
}
info!(
channel_id = %channel_id,
pubkey = %pubkey_hex,

View File

@@ -656,7 +656,15 @@ pub async fn finalize_atomic_corporate_identity_result(
) -> Result<CorporateIdentityDecision, CorporateIdentityError> {
let result = match proof {
CorporateIdentityProof::NotRequired => Ok(CorporateIdentityDecision::NotRequired),
CorporateIdentityProof::Delegated { .. } => Err(CorporateIdentityError::DelegationDenied),
CorporateIdentityProof::Delegated {
owner_pubkey,
owner_issuer,
owner_uid,
} => Ok(CorporateIdentityDecision::Delegated {
owner_pubkey,
owner_issuer,
owner_uid,
}),
CorporateIdentityProof::Direct { claims, source } => {
let binding = committed_binding.ok_or_else(|| {
buzz_db::DbError::InvalidData(