mirror of
https://github.com/block/buzz.git
synced 2026-10-28 13:01:10 +00:00
fix(db): commit invitation ownership on admission
Co-authored-by: Codex <codex@openai.com> Signed-off-by: Brad Seiler <seiler@squareup.com> Signed-off-by: Local Builder <1c476cb1127ea270cb900bdc3891060140bdd1173f91498030ee9b4db01c2f20@buzz.block.builderlab.xyz>
This commit is contained in:
@@ -151,6 +151,9 @@ the same machine.
|
||||
New invitations are durable v2 bearers. A community ban permanently revokes
|
||||
outstanding v2 invitations issued by that principal or its recorded agents.
|
||||
Unban does not restore them; an unrestricted issuer can mint new invitations.
|
||||
Claim credentials are verified without user writes. Successful claims commit
|
||||
verified ownership with membership, policy evidence, and v2 consumption.
|
||||
Rejected claims create no users or owner links.
|
||||
Claim and mint restriction reads use the writer inside a transaction serialized
|
||||
with community restriction and agent-owner writes. Admission committed first
|
||||
precedes the ban; a ban committed first prevents subsequent admission. Restriction
|
||||
|
||||
@@ -41,6 +41,10 @@ pub enum DbError {
|
||||
#[error("invitation principal is restricted")]
|
||||
InviteRestricted,
|
||||
|
||||
/// A verified invitation owner conflicts with the recorded owner.
|
||||
#[error("invitation owner does not match the recorded owner")]
|
||||
InviteOwnerMismatch,
|
||||
|
||||
/// JSON serialization or deserialization failed.
|
||||
#[error("serialization error: {0}")]
|
||||
Serde(#[from] serde_json::Error),
|
||||
|
||||
@@ -29,6 +29,9 @@ use uuid::Uuid;
|
||||
use crate::error::Result;
|
||||
use crate::{CommunityId, Db};
|
||||
|
||||
mod claim_owner;
|
||||
pub use claim_owner::{claim_relay_membership_with_owner, MembershipClaimResult};
|
||||
|
||||
/// Outcome of a v2 invite claim. Expected invalid/expired/exhausted states are
|
||||
/// typed variants so the relay layer can map them to distinct HTTP responses
|
||||
/// without inspecting database errors.
|
||||
@@ -61,6 +64,26 @@ pub enum ClaimOutcome {
|
||||
Restricted,
|
||||
}
|
||||
|
||||
/// A completed claim and whether its committed transaction first recorded an owner.
|
||||
pub struct InviteClaimResult {
|
||||
/// Membership or denial outcome.
|
||||
pub outcome: ClaimOutcome,
|
||||
/// Notify ownerless sockets only after a successful commit created the link.
|
||||
pub owner_link_created: bool,
|
||||
/// User rows first created by this successful transaction.
|
||||
pub users_created: u64,
|
||||
}
|
||||
|
||||
impl From<ClaimOutcome> for InviteClaimResult {
|
||||
fn from(outcome: ClaimOutcome) -> Self {
|
||||
Self {
|
||||
outcome,
|
||||
owner_link_created: false,
|
||||
users_created: 0,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A freshly minted v2 invite, including the plaintext code and metadata.
|
||||
#[derive(Debug)]
|
||||
pub struct MintedInvite {
|
||||
@@ -304,6 +327,29 @@ pub async fn claim_relay_invite(
|
||||
claimer_pubkey: &str,
|
||||
policy_version: Option<&str>,
|
||||
) -> Result<ClaimOutcome> {
|
||||
Ok(claim_relay_invite_with_owner(
|
||||
pool,
|
||||
community,
|
||||
token_hash,
|
||||
claimer_pubkey,
|
||||
policy_version,
|
||||
None,
|
||||
)
|
||||
.await?
|
||||
.outcome)
|
||||
}
|
||||
|
||||
/// Claim an invite with an owner verified by the caller, without pre-admission writes.
|
||||
/// Restriction and first-write-wins owner checks hold the admission lock; ownership
|
||||
/// commits atomically with membership, policy evidence, and invite consumption.
|
||||
pub async fn claim_relay_invite_with_owner(
|
||||
pool: &PgPool,
|
||||
community: CommunityId,
|
||||
token_hash: &[u8; 32],
|
||||
claimer_pubkey: &str,
|
||||
policy_version: Option<&str>,
|
||||
verified_owner: Option<&str>,
|
||||
) -> Result<InviteClaimResult> {
|
||||
let mut tx = crate::begin_community_write_transaction(
|
||||
pool,
|
||||
community,
|
||||
@@ -311,9 +357,9 @@ pub async fn claim_relay_invite(
|
||||
)
|
||||
.await?;
|
||||
|
||||
lock_admission(tx.conn(), community).await?;
|
||||
if principal_restricted(tx.conn(), community, claimer_pubkey).await? {
|
||||
return Ok(ClaimOutcome::Restricted);
|
||||
lock_admission(&mut tx, community).await?;
|
||||
if claim_owner::claimant_restricted(&mut tx, community, claimer_pubkey, verified_owner).await? {
|
||||
return Ok(ClaimOutcome::Restricted.into());
|
||||
}
|
||||
|
||||
// 2. SELECT FOR UPDATE — lock the invite row for the duration of this txn.
|
||||
@@ -332,7 +378,7 @@ pub async fn claim_relay_invite(
|
||||
let Some(invite) = row else {
|
||||
tx.rollback().await?;
|
||||
log_claim_outcome(community, None, "invalid", None, None);
|
||||
return Ok(ClaimOutcome::Invalid);
|
||||
return Ok(ClaimOutcome::Invalid.into());
|
||||
};
|
||||
|
||||
let invite_id: uuid::Uuid = invite.try_get("id")?;
|
||||
@@ -352,13 +398,13 @@ pub async fn claim_relay_invite(
|
||||
max_uses,
|
||||
Some(use_count),
|
||||
);
|
||||
return Ok(ClaimOutcome::Expired);
|
||||
return Ok(ClaimOutcome::Expired.into());
|
||||
}
|
||||
|
||||
let revoked_at: Option<DateTime<Utc>> = invite.try_get("revoked_at")?;
|
||||
let issuer: String = invite.try_get("created_by")?;
|
||||
if revoked_at.is_some() || principal_restricted(tx.conn(), community, &issuer).await? {
|
||||
return Ok(ClaimOutcome::Revoked);
|
||||
if revoked_at.is_some() || principal_restricted(&mut tx, community, &issuer).await? {
|
||||
return Ok(ClaimOutcome::Revoked.into());
|
||||
}
|
||||
|
||||
let uses_remaining = || max_uses.map(|mu| mu - use_count);
|
||||
@@ -384,6 +430,9 @@ pub async fn claim_relay_invite(
|
||||
.execute(tx.conn())
|
||||
.await?;
|
||||
}
|
||||
let owner =
|
||||
claim_owner::persist_claim_owner(&mut tx, community, claimer_pubkey, verified_owner)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
log_claim_outcome(
|
||||
community,
|
||||
@@ -392,9 +441,13 @@ pub async fn claim_relay_invite(
|
||||
max_uses,
|
||||
Some(use_count),
|
||||
);
|
||||
return Ok(ClaimOutcome::AlreadyMember {
|
||||
use_count,
|
||||
uses_remaining: uses_remaining(),
|
||||
return Ok(InviteClaimResult {
|
||||
outcome: ClaimOutcome::AlreadyMember {
|
||||
use_count,
|
||||
uses_remaining: uses_remaining(),
|
||||
},
|
||||
owner_link_created: owner.link_created,
|
||||
users_created: owner.users_created,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -409,7 +462,7 @@ pub async fn claim_relay_invite(
|
||||
max_uses,
|
||||
Some(use_count),
|
||||
);
|
||||
return Ok(ClaimOutcome::Exhausted);
|
||||
return Ok(ClaimOutcome::Exhausted.into());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -443,6 +496,9 @@ pub async fn claim_relay_invite(
|
||||
}
|
||||
|
||||
if !inserted {
|
||||
let owner =
|
||||
claim_owner::persist_claim_owner(&mut tx, community, claimer_pubkey, verified_owner)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
log_claim_outcome(
|
||||
community,
|
||||
@@ -451,9 +507,13 @@ pub async fn claim_relay_invite(
|
||||
max_uses,
|
||||
Some(use_count),
|
||||
);
|
||||
return Ok(ClaimOutcome::AlreadyMember {
|
||||
use_count,
|
||||
uses_remaining: uses_remaining(),
|
||||
return Ok(InviteClaimResult {
|
||||
outcome: ClaimOutcome::AlreadyMember {
|
||||
use_count,
|
||||
uses_remaining: uses_remaining(),
|
||||
},
|
||||
owner_link_created: owner.link_created,
|
||||
users_created: owner.users_created,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -466,7 +526,10 @@ pub async fn claim_relay_invite(
|
||||
.execute(tx.conn())
|
||||
.await?;
|
||||
|
||||
// 11. Commit.
|
||||
// 11. Commit membership, evidence, consumption, and verified ownership together.
|
||||
let owner =
|
||||
claim_owner::persist_claim_owner(&mut tx, community, claimer_pubkey, verified_owner)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
|
||||
let new_uses_remaining = max_uses.map(|mu| mu - new_use_count);
|
||||
@@ -479,9 +542,13 @@ pub async fn claim_relay_invite(
|
||||
Some(new_use_count),
|
||||
);
|
||||
|
||||
Ok(ClaimOutcome::Joined {
|
||||
use_count: new_use_count,
|
||||
uses_remaining: new_uses_remaining,
|
||||
Ok(InviteClaimResult {
|
||||
outcome: ClaimOutcome::Joined {
|
||||
use_count: new_use_count,
|
||||
uses_remaining: new_uses_remaining,
|
||||
},
|
||||
owner_link_created: owner.link_created,
|
||||
users_created: owner.users_created,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -502,6 +569,27 @@ impl Db {
|
||||
mint_relay_invite(&self.pool, community, created_by, ttl_secs, max_uses).await
|
||||
}
|
||||
|
||||
/// Claim with verified owner provenance, returning committed owner-link creation.
|
||||
#[datastore_span(name = "claim_relay_invite_with_owner", system = "postgresql")]
|
||||
pub async fn claim_relay_invite_with_owner(
|
||||
&self,
|
||||
community: CommunityId,
|
||||
token_hash: &[u8; 32],
|
||||
claimer_pubkey: &str,
|
||||
policy_version: Option<&str>,
|
||||
verified_owner: Option<&str>,
|
||||
) -> Result<InviteClaimResult> {
|
||||
claim_relay_invite_with_owner(
|
||||
&self.pool,
|
||||
community,
|
||||
token_hash,
|
||||
claimer_pubkey,
|
||||
policy_version,
|
||||
verified_owner,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Delete one bounded batch of invites expired before `cutoff`.
|
||||
#[datastore_span(name = "reap_expired_relay_invites", system = "postgresql")]
|
||||
pub async fn reap_expired_relay_invites(&self, cutoff: DateTime<Utc>) -> Result<u64> {
|
||||
@@ -540,7 +628,7 @@ mod postgres_tests {
|
||||
use sqlx::PgPool;
|
||||
use uuid::Uuid;
|
||||
|
||||
async fn setup_pool() -> PgPool {
|
||||
pub(super) async fn setup_pool() -> PgPool {
|
||||
PgPool::connect(&crate::test_support::database_url())
|
||||
.await
|
||||
.expect("connect to test DB")
|
||||
@@ -575,7 +663,7 @@ mod postgres_tests {
|
||||
admin.close().await;
|
||||
}
|
||||
|
||||
async fn make_test_community(pool: &PgPool) -> CommunityId {
|
||||
pub(super) async fn make_test_community(pool: &PgPool) -> CommunityId {
|
||||
let id = Uuid::new_v4();
|
||||
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
|
||||
.bind(id)
|
||||
@@ -606,11 +694,11 @@ mod postgres_tests {
|
||||
tx.commit().await.expect("commit test cleanup");
|
||||
}
|
||||
|
||||
fn test_pubkey() -> String {
|
||||
pub(super) fn test_pubkey() -> String {
|
||||
format!("{:064x}", Uuid::new_v4().as_u128())
|
||||
}
|
||||
|
||||
async fn use_count(pool: &PgPool, community: CommunityId, invite_id: Uuid) -> i32 {
|
||||
pub(super) async fn use_count(pool: &PgPool, community: CommunityId, invite_id: Uuid) -> i32 {
|
||||
sqlx::query_scalar(
|
||||
"SELECT use_count FROM relay_invites WHERE community_id = $1 AND id = $2",
|
||||
)
|
||||
@@ -1019,3 +1107,6 @@ mod revocation_postgres_tests;
|
||||
#[cfg(test)]
|
||||
#[path = "relay_invite/upgrade_postgres_tests.rs"]
|
||||
mod upgrade_postgres_tests;
|
||||
|
||||
#[cfg(test)]
|
||||
mod owner_admission_postgres_tests;
|
||||
|
||||
145
crates/buzz-db/src/store/relay_invite/claim_owner.rs
Normal file
145
crates/buzz-db/src/store/relay_invite/claim_owner.rs
Normal file
@@ -0,0 +1,145 @@
|
||||
use crate::error::Result;
|
||||
use crate::{CommunityId, DbError};
|
||||
use sqlx::{PgPool, Postgres, Transaction};
|
||||
|
||||
/// Result of a committed legacy invitation claim.
|
||||
pub struct MembershipClaimResult {
|
||||
/// Whether this claim inserted membership.
|
||||
pub inserted: bool,
|
||||
/// Whether the same commit first recorded verified ownership.
|
||||
pub owner_link_created: bool,
|
||||
/// User rows first created by this successful transaction.
|
||||
pub users_created: u64,
|
||||
}
|
||||
|
||||
/// Check existing and presented owner identities without creating a user row.
|
||||
pub(super) async fn claimant_restricted(
|
||||
tx: &mut Transaction<'_, Postgres>,
|
||||
community: CommunityId,
|
||||
claimant: &str,
|
||||
verified_owner: Option<&str>,
|
||||
) -> Result<bool> {
|
||||
if super::principal_restricted(tx, community, claimant).await? {
|
||||
return Ok(true);
|
||||
}
|
||||
let Some(owner) = verified_owner else {
|
||||
return Ok(false);
|
||||
};
|
||||
if super::principal_restricted(tx, community, owner).await? {
|
||||
return Ok(true);
|
||||
}
|
||||
let recorded: Option<Option<Vec<u8>>> = sqlx::query_scalar(
|
||||
"SELECT agent_owner_pubkey FROM users WHERE community_id = $1 AND pubkey = $2",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(
|
||||
hex::decode(claimant)
|
||||
.map_err(|_| DbError::InvalidData("invalid invitation principal".into()))?,
|
||||
)
|
||||
.fetch_optional(&mut **tx)
|
||||
.await?;
|
||||
let owner_bytes =
|
||||
hex::decode(owner).map_err(|_| DbError::InvalidData("invalid invitation owner".into()))?;
|
||||
if recorded
|
||||
.flatten()
|
||||
.is_some_and(|existing| existing != owner_bytes)
|
||||
{
|
||||
return Err(DbError::InviteOwnerMismatch);
|
||||
}
|
||||
Ok(false)
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
pub(super) struct OwnerPersistence {
|
||||
pub(super) link_created: bool,
|
||||
pub(super) users_created: u64,
|
||||
}
|
||||
|
||||
/// Called only on an admitted claim, while the admission lock is still held.
|
||||
pub(super) async fn persist_claim_owner(
|
||||
tx: &mut Transaction<'_, Postgres>,
|
||||
community: CommunityId,
|
||||
claimant: &str,
|
||||
verified_owner: Option<&str>,
|
||||
) -> Result<OwnerPersistence> {
|
||||
let Some(owner) = verified_owner else {
|
||||
return Ok(OwnerPersistence::default());
|
||||
};
|
||||
let owner_bytes =
|
||||
hex::decode(owner).map_err(|_| DbError::InvalidData("invalid invitation owner".into()))?;
|
||||
let claimant_bytes = hex::decode(claimant)
|
||||
.map_err(|_| DbError::InvalidData("invalid invitation principal".into()))?;
|
||||
let mut users_created = 0;
|
||||
for key in [&owner_bytes, &claimant_bytes] {
|
||||
users_created += sqlx::query(
|
||||
"INSERT INTO users (community_id, pubkey) VALUES ($1, $2) ON CONFLICT DO NOTHING",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(key)
|
||||
.execute(&mut **tx)
|
||||
.await?
|
||||
.rows_affected();
|
||||
}
|
||||
let link_created = sqlx::query("UPDATE users SET agent_owner_pubkey=$3 WHERE community_id=$1 AND pubkey=$2 AND agent_owner_pubkey IS NULL")
|
||||
.bind(community.as_uuid()).bind(claimant_bytes).bind(owner_bytes).execute(&mut **tx).await?.rows_affected() > 0;
|
||||
Ok(OwnerPersistence {
|
||||
link_created,
|
||||
users_created,
|
||||
})
|
||||
}
|
||||
|
||||
/// Atomically claim membership and caller-verified ownership after policy checks.
|
||||
pub async fn claim_relay_membership_with_owner(
|
||||
pool: &PgPool,
|
||||
community: CommunityId,
|
||||
pubkey: &str,
|
||||
role: &str,
|
||||
policy_version: Option<&str>,
|
||||
verified_owner: Option<&str>,
|
||||
) -> Result<MembershipClaimResult> {
|
||||
let connection = crate::observability::acquire_writer(
|
||||
pool,
|
||||
crate::observability::WriterOperation::Authorization,
|
||||
)
|
||||
.await?;
|
||||
let mut tx = sqlx::Transaction::begin(connection, None).await?;
|
||||
crate::deletion::DeletionStore::new(pool.clone())
|
||||
.guard_transaction(&mut tx, community)
|
||||
.await?;
|
||||
super::lock_admission(&mut tx, community).await?;
|
||||
if claimant_restricted(&mut tx, community, pubkey, verified_owner).await? {
|
||||
return Err(DbError::InviteRestricted);
|
||||
}
|
||||
let inserted = sqlx::query(
|
||||
"INSERT INTO relay_members (community_id, pubkey, role, added_by) \
|
||||
VALUES ($1, $2, $3, 'invite') \
|
||||
ON CONFLICT (community_id, pubkey) DO NOTHING",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(pubkey)
|
||||
.bind(role)
|
||||
.execute(&mut *tx)
|
||||
.await?
|
||||
.rows_affected()
|
||||
> 0;
|
||||
|
||||
if let Some(version) = policy_version {
|
||||
sqlx::query(
|
||||
"INSERT INTO join_policy_acceptances (community_id, pubkey, policy_version) \
|
||||
VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(pubkey)
|
||||
.bind(version)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
}
|
||||
|
||||
let owner = persist_claim_owner(&mut tx, community, pubkey, verified_owner).await?;
|
||||
tx.commit().await?;
|
||||
Ok(MembershipClaimResult {
|
||||
inserted,
|
||||
owner_link_created: owner.link_created,
|
||||
users_created: owner.users_created,
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,112 @@
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn failed_policy_evidence_rolls_back_verified_owner_and_membership() {
|
||||
let pool = postgres_tests::setup_pool().await;
|
||||
let community = postgres_tests::make_test_community(&pool).await;
|
||||
let claimant = postgres_tests::test_pubkey();
|
||||
let owner = postgres_tests::test_pubkey();
|
||||
let invite = mint_relay_invite(
|
||||
&pool,
|
||||
community,
|
||||
&postgres_tests::test_pubkey(),
|
||||
3600,
|
||||
Some(1),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(claim_relay_invite_with_owner(
|
||||
&pool,
|
||||
community,
|
||||
&hash_v2_code(&invite.code),
|
||||
&claimant,
|
||||
Some("too-short"),
|
||||
Some(&owner)
|
||||
)
|
||||
.await
|
||||
.is_err());
|
||||
assert!(claim_relay_membership_with_owner(
|
||||
&pool,
|
||||
community,
|
||||
&claimant,
|
||||
"member",
|
||||
Some("too-short"),
|
||||
Some(&owner)
|
||||
)
|
||||
.await
|
||||
.is_err());
|
||||
let users: i64 = sqlx::query_scalar("SELECT count(*) FROM users WHERE community_id=$1")
|
||||
.bind(community.as_uuid())
|
||||
.fetch_one(&pool)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(users, 0);
|
||||
assert!(
|
||||
!crate::relay_members::is_relay_member(&pool, community, &claimant)
|
||||
.await
|
||||
.unwrap()
|
||||
);
|
||||
assert_eq!(
|
||||
postgres_tests::use_count(&pool, community, invite.invite_id).await,
|
||||
0
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn successful_claim_reports_only_committed_user_creation() {
|
||||
let pool = postgres_tests::setup_pool().await;
|
||||
let community = postgres_tests::make_test_community(&pool).await;
|
||||
let claimant = postgres_tests::test_pubkey();
|
||||
let owner = postgres_tests::test_pubkey();
|
||||
let invite = mint_relay_invite(
|
||||
&pool,
|
||||
community,
|
||||
&postgres_tests::test_pubkey(),
|
||||
3600,
|
||||
Some(1),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let hash = hash_v2_code(&invite.code);
|
||||
let first =
|
||||
claim_relay_invite_with_owner(&pool, community, &hash, &claimant, None, Some(&owner))
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(matches!(first.outcome, ClaimOutcome::Joined { .. }));
|
||||
assert!(first.owner_link_created);
|
||||
assert_eq!(first.users_created, 2);
|
||||
let retry =
|
||||
claim_relay_invite_with_owner(&pool, community, &hash, &claimant, None, Some(&owner))
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(matches!(retry.outcome, ClaimOutcome::AlreadyMember { .. }));
|
||||
assert!(!retry.owner_link_created);
|
||||
assert_eq!(retry.users_created, 0);
|
||||
let legacy_agent = postgres_tests::test_pubkey();
|
||||
let legacy = claim_relay_membership_with_owner(
|
||||
&pool,
|
||||
community,
|
||||
&legacy_agent,
|
||||
"member",
|
||||
None,
|
||||
Some(&owner),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(legacy.inserted && legacy.owner_link_created);
|
||||
assert_eq!(legacy.users_created, 1);
|
||||
let retry = claim_relay_membership_with_owner(
|
||||
&pool,
|
||||
community,
|
||||
&legacy_agent,
|
||||
"member",
|
||||
None,
|
||||
Some(&owner),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(!retry.inserted && !retry.owner_link_created);
|
||||
assert_eq!(retry.users_created, 0);
|
||||
}
|
||||
@@ -182,43 +182,16 @@ pub async fn claim_relay_membership(
|
||||
role: &str,
|
||||
policy_version: Option<&str>,
|
||||
) -> Result<bool> {
|
||||
let mut tx = crate::begin_community_write_transaction(
|
||||
Ok(super::relay_invite::claim_relay_membership_with_owner(
|
||||
pool,
|
||||
community,
|
||||
observability::WriterOperation::Authorization,
|
||||
pubkey,
|
||||
role,
|
||||
policy_version,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
super::relay_invite::lock_admission(tx.conn(), community).await?;
|
||||
if super::relay_invite::principal_restricted(tx.conn(), community, pubkey).await? {
|
||||
return Err(DbError::InviteRestricted);
|
||||
}
|
||||
let inserted = sqlx::query(
|
||||
"INSERT INTO relay_members (community_id, pubkey, role, added_by) \
|
||||
VALUES ($1, $2, $3, 'invite') \
|
||||
ON CONFLICT (community_id, pubkey) DO NOTHING",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(pubkey)
|
||||
.bind(role)
|
||||
.execute(tx.conn())
|
||||
.await?
|
||||
.rows_affected()
|
||||
> 0;
|
||||
|
||||
if let Some(version) = policy_version {
|
||||
sqlx::query(
|
||||
"INSERT INTO join_policy_acceptances (community_id, pubkey, policy_version) \
|
||||
VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
|
||||
)
|
||||
.bind(community.as_uuid())
|
||||
.bind(pubkey)
|
||||
.bind(version)
|
||||
.execute(tx.conn())
|
||||
.await?;
|
||||
}
|
||||
|
||||
tx.commit().await?;
|
||||
Ok(inserted)
|
||||
.inserted)
|
||||
}
|
||||
|
||||
/// Returns whether a member has persisted acceptance evidence for a policy version.
|
||||
@@ -1059,6 +1032,27 @@ impl Db {
|
||||
claim_relay_membership(&self.pool, community, pubkey, role, policy_version).await
|
||||
}
|
||||
|
||||
/// Claim legacy membership and verified ownership in one admission transaction.
|
||||
#[datastore_span(name = "claim_relay_membership_with_owner", system = "postgresql")]
|
||||
pub async fn claim_relay_membership_with_owner(
|
||||
&self,
|
||||
community: CommunityId,
|
||||
pubkey: &str,
|
||||
role: &str,
|
||||
policy_version: Option<&str>,
|
||||
verified_owner: Option<&str>,
|
||||
) -> Result<super::relay_invite::MembershipClaimResult> {
|
||||
super::relay_invite::claim_relay_membership_with_owner(
|
||||
&self.pool,
|
||||
community,
|
||||
pubkey,
|
||||
role,
|
||||
policy_version,
|
||||
verified_owner,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Returns whether a member has persisted acceptance evidence for a policy version.
|
||||
#[datastore_span(name = "has_join_policy_acceptance", system = "postgresql")]
|
||||
pub async fn has_join_policy_acceptance(
|
||||
|
||||
@@ -35,6 +35,8 @@ use crate::invite_token;
|
||||
use crate::state::AppState;
|
||||
|
||||
use super::{api_error, bridge, internal_error};
|
||||
mod claim_auth;
|
||||
use claim_auth::{authenticate, map_claim_error};
|
||||
|
||||
/// Fixed-window size for the per-pubkey claim rate limiter.
|
||||
pub(crate) const CLAIM_RATE_WINDOW: Duration = Duration::from_secs(60);
|
||||
@@ -226,64 +228,6 @@ pub async fn accept_policy(
|
||||
Ok(Json(serde_json::json!({ "receipt": receipt })))
|
||||
}
|
||||
|
||||
/// Shared prelude: bind the tenant from the Host header and verify the NIP-98
|
||||
/// signature + replay for `path`.
|
||||
async fn authenticate(
|
||||
state: &Arc<AppState>,
|
||||
headers: &HeaderMap,
|
||||
path: &str,
|
||||
body: &[u8],
|
||||
) -> Result<(buzz_core::TenantContext, nostr::PublicKey), (StatusCode, Json<Value>)> {
|
||||
let raw_host = headers
|
||||
.get(axum::http::header::HOST)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.unwrap_or("");
|
||||
let tenant = crate::tenant::bind_community(&state.db, raw_host)
|
||||
.await
|
||||
.map_err(|_| {
|
||||
api_error(
|
||||
StatusCode::NOT_FOUND,
|
||||
"relay: no community is configured for this host",
|
||||
)
|
||||
})?;
|
||||
|
||||
let url = bridge::nip98_expected_url(&state.config.relay_url, &tenant, path);
|
||||
let bridge::VerifiedBridgeAuth {
|
||||
pubkey,
|
||||
event_id_bytes,
|
||||
signed_created_at,
|
||||
} = bridge::verify_nip98_exempt_invite_claim(headers, "POST", &url, Some(body))?;
|
||||
bridge::check_nip98_replay(state, &tenant, event_id_bytes).await?;
|
||||
|
||||
materialize_invite_owner(state, &tenant, &pubkey, headers, signed_created_at).await?;
|
||||
Ok((tenant, pubkey))
|
||||
}
|
||||
|
||||
async fn materialize_invite_owner(
|
||||
state: &AppState,
|
||||
tenant: &buzz_core::TenantContext,
|
||||
pubkey: &nostr::PublicKey,
|
||||
headers: &HeaderMap,
|
||||
signed_created_at: Option<u64>,
|
||||
) -> Result<(), (StatusCode, Json<Value>)> {
|
||||
// The claim remains membership-exempt. Only cryptographically verified
|
||||
// owner provenance is persisted, so the transactional restriction lookup
|
||||
// also covers an agent onboarding for the first time.
|
||||
if let Some(owner) = super::relay_members::extract_nip_oa_owner(
|
||||
pubkey.as_bytes(),
|
||||
super::relay_members::extract_auth_tag_header(headers),
|
||||
signed_created_at,
|
||||
) {
|
||||
if !super::relay_members::materialize_nip_oa_owner(state, tenant, pubkey, &owner).await {
|
||||
return Err(api_error(
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
"restriction lookup unavailable",
|
||||
));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn map_mint_error(error: buzz_db::DbError) -> (StatusCode, Json<Value>) {
|
||||
match error {
|
||||
buzz_db::DbError::InviteRestricted => api_error(StatusCode::FORBIDDEN, "invite_restricted"),
|
||||
@@ -369,10 +313,18 @@ async fn mint_invite_checked(
|
||||
return e.into_response();
|
||||
}
|
||||
|
||||
if let Err(error) =
|
||||
materialize_invite_owner(&state, &tenant, &pubkey, &headers, signed_created_at).await
|
||||
{
|
||||
return error.into_response();
|
||||
if let Some(owner) = super::relay_members::extract_nip_oa_owner(
|
||||
pubkey.as_bytes(),
|
||||
super::relay_members::extract_auth_tag_header(&headers),
|
||||
signed_created_at,
|
||||
) {
|
||||
if !super::relay_members::materialize_nip_oa_owner(&state, &tenant, &pubkey, &owner).await {
|
||||
return api_error(
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
"restriction lookup unavailable",
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
}
|
||||
|
||||
match mint_invite_inner(&state, body, tenant, pubkey).await {
|
||||
@@ -462,7 +414,8 @@ pub async fn claim_invite(
|
||||
headers: HeaderMap,
|
||||
body: axum::body::Bytes,
|
||||
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
|
||||
let (tenant, pubkey) = authenticate(&state, &headers, "/api/invites/claim", &body).await?;
|
||||
let (tenant, pubkey, verified_owner) =
|
||||
authenticate(&state, &headers, "/api/invites/claim", &body).await?;
|
||||
|
||||
if claim_rate_limited(&state, tenant.community(), &pubkey) {
|
||||
return Err(api_error(
|
||||
@@ -475,6 +428,7 @@ pub async fn claim_invite(
|
||||
.map_err(|e| api_error(StatusCode::BAD_REQUEST, &format!("invalid claim JSON: {e}")))?;
|
||||
|
||||
let claimer_hex = pubkey.to_hex();
|
||||
let owner_hex = verified_owner.map(|owner| owner.to_hex());
|
||||
let key = invite_token::derive_invite_key(&state.relay_keypair);
|
||||
|
||||
// --- v2 database-backed path ---
|
||||
@@ -499,7 +453,7 @@ pub async fn claim_invite(
|
||||
let token_hash = hash_v2_code(&request.code);
|
||||
let outcome = state
|
||||
.db
|
||||
.claim_relay_invite(
|
||||
.claim_relay_invite_with_owner(
|
||||
tenant.community(),
|
||||
&token_hash,
|
||||
&claimer_hex,
|
||||
@@ -508,11 +462,16 @@ pub async fn claim_invite(
|
||||
.join_policy
|
||||
.as_ref()
|
||||
.map(|policy| policy.version.as_str()),
|
||||
owner_hex.as_deref(),
|
||||
)
|
||||
.await
|
||||
.map_err(|e| internal_error(&format!("v2 invite claim: {e}")))?;
|
||||
|
||||
return match outcome {
|
||||
.map_err(map_claim_error)?;
|
||||
metrics::counter!("buzz_users_created_total", "community" => tenant.host().to_owned())
|
||||
.increment(outcome.users_created);
|
||||
if outcome.owner_link_created {
|
||||
state.disconnect_unowned_agent_clusterwide(&tenant, pubkey.as_bytes());
|
||||
}
|
||||
return match outcome.outcome {
|
||||
buzz_db::relay_invite::ClaimOutcome::Joined { .. } => {
|
||||
tracing::info!(
|
||||
community = %tenant.community(),
|
||||
@@ -584,9 +543,9 @@ pub async fn claim_invite(
|
||||
.map_err(|_| api_error(StatusCode::FORBIDDEN, "join_policy_required"))?;
|
||||
}
|
||||
|
||||
let was_inserted = state
|
||||
let claim = state
|
||||
.db
|
||||
.claim_relay_membership(
|
||||
.claim_relay_membership_with_owner(
|
||||
tenant.community(),
|
||||
&claimer_hex,
|
||||
&payload.r,
|
||||
@@ -595,14 +554,16 @@ pub async fn claim_invite(
|
||||
.join_policy
|
||||
.as_ref()
|
||||
.map(|policy| policy.version.as_str()),
|
||||
owner_hex.as_deref(),
|
||||
)
|
||||
.await
|
||||
.map_err(|e| match e {
|
||||
buzz_db::DbError::InviteRestricted => {
|
||||
api_error(StatusCode::FORBIDDEN, "invite_restricted")
|
||||
}
|
||||
e => internal_error(&format!("invite claim insert: {e}")),
|
||||
})?;
|
||||
.map_err(map_claim_error)?;
|
||||
metrics::counter!("buzz_users_created_total", "community" => tenant.host().to_owned())
|
||||
.increment(claim.users_created);
|
||||
if claim.owner_link_created {
|
||||
state.disconnect_unowned_agent_clusterwide(&tenant, pubkey.as_bytes());
|
||||
}
|
||||
let was_inserted = claim.inserted;
|
||||
|
||||
if was_inserted {
|
||||
tracing::info!(
|
||||
@@ -654,6 +615,7 @@ fn claim_key_rate_limited(
|
||||
|
||||
#[cfg(test)]
|
||||
mod postgres_tests {
|
||||
mod owner_admission_tests;
|
||||
mod restriction_tests;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
58
crates/buzz-relay/src/api/invites/claim_auth.rs
Normal file
58
crates/buzz-relay/src/api/invites/claim_auth.rs
Normal file
@@ -0,0 +1,58 @@
|
||||
use super::{api_error, bridge, internal_error, AppState};
|
||||
use axum::{
|
||||
http::{HeaderMap, StatusCode},
|
||||
response::Json,
|
||||
};
|
||||
use serde_json::Value;
|
||||
use std::sync::Arc;
|
||||
|
||||
/// Verify tenant-scoped NIP-98 and optional NIP-OA provenance without user writes.
|
||||
pub(super) async fn authenticate(
|
||||
state: &Arc<AppState>,
|
||||
headers: &HeaderMap,
|
||||
path: &str,
|
||||
body: &[u8],
|
||||
) -> Result<
|
||||
(
|
||||
buzz_core::TenantContext,
|
||||
nostr::PublicKey,
|
||||
Option<nostr::PublicKey>,
|
||||
),
|
||||
(StatusCode, Json<Value>),
|
||||
> {
|
||||
let raw_host = headers
|
||||
.get(axum::http::header::HOST)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.unwrap_or("");
|
||||
let tenant = crate::tenant::bind_community(&state.db, raw_host)
|
||||
.await
|
||||
.map_err(|_| {
|
||||
api_error(
|
||||
StatusCode::NOT_FOUND,
|
||||
"relay: no community is configured for this host",
|
||||
)
|
||||
})?;
|
||||
let url = bridge::nip98_expected_url(&state.config.relay_url, &tenant, path);
|
||||
let bridge::VerifiedBridgeAuth {
|
||||
pubkey,
|
||||
event_id_bytes,
|
||||
signed_created_at,
|
||||
} = bridge::verify_nip98_exempt_invite_claim(headers, "POST", &url, Some(body))?;
|
||||
bridge::check_nip98_replay(state, &tenant, event_id_bytes).await?;
|
||||
let owner = super::super::relay_members::extract_nip_oa_owner(
|
||||
pubkey.as_bytes(),
|
||||
super::super::relay_members::extract_auth_tag_header(headers),
|
||||
signed_created_at,
|
||||
);
|
||||
Ok((tenant, pubkey, owner))
|
||||
}
|
||||
|
||||
pub(super) fn map_claim_error(error: buzz_db::DbError) -> (StatusCode, Json<Value>) {
|
||||
match error {
|
||||
buzz_db::DbError::InviteRestricted => api_error(StatusCode::FORBIDDEN, "invite_restricted"),
|
||||
buzz_db::DbError::InviteOwnerMismatch => {
|
||||
api_error(StatusCode::FORBIDDEN, "invite_owner_mismatch")
|
||||
}
|
||||
e => internal_error(&format!("invite claim: {e}")),
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,177 @@
|
||||
mod postgres_tests {
|
||||
use super::super::restriction_tests::postgres_tests::{fixture, post_with_owner_tag};
|
||||
use super::super::*;
|
||||
|
||||
async fn users(state: &AppState, community: buzz_core::CommunityId) -> i64 {
|
||||
sqlx::query_scalar("SELECT count(*) FROM users WHERE community_id=$1")
|
||||
.bind(community.as_uuid())
|
||||
.fetch_one(state.db.pool())
|
||||
.await
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn invalid_signed_claims_do_not_materialize_fresh_owner_identities() {
|
||||
let (state, host, community, _issuer) = fixture().await;
|
||||
let before = users(&state, community).await;
|
||||
let unknown_v2 = buzz_core::invite::encode_v2_code(&[42; 32]);
|
||||
for code in ["garbage", "v2.bad", &unknown_v2] {
|
||||
let owner = Keys::generate();
|
||||
let agent = Keys::generate();
|
||||
let tag = buzz_sdk::nip_oa::compute_auth_tag(&owner, &agent.public_key(), "").unwrap();
|
||||
let response = post_with_owner_tag(
|
||||
state.clone(),
|
||||
&host,
|
||||
"/api/invites/claim",
|
||||
&agent,
|
||||
serde_json::json!({"code":code}).to_string(),
|
||||
&tag,
|
||||
)
|
||||
.await;
|
||||
assert_eq!(response.status(), StatusCode::FORBIDDEN);
|
||||
assert_eq!(users(&state, community).await, before);
|
||||
assert!(!state
|
||||
.db
|
||||
.is_relay_member(community, &agent.public_key().to_hex())
|
||||
.await
|
||||
.unwrap());
|
||||
}
|
||||
let owner = Keys::generate();
|
||||
let agent = Keys::generate();
|
||||
let tag = buzz_sdk::nip_oa::compute_auth_tag(&owner, &agent.public_key(), "").unwrap();
|
||||
for attempt in 0..=CLAIM_RATE_LIMIT {
|
||||
let response = post_with_owner_tag(
|
||||
state.clone(),
|
||||
&host,
|
||||
"/api/invites/claim",
|
||||
&agent,
|
||||
serde_json::json!({"code":"garbage"}).to_string(),
|
||||
&tag,
|
||||
)
|
||||
.await;
|
||||
assert_eq!(
|
||||
response.status(),
|
||||
if attempt == CLAIM_RATE_LIMIT {
|
||||
StatusCode::TOO_MANY_REQUESTS
|
||||
} else {
|
||||
StatusCode::FORBIDDEN
|
||||
}
|
||||
);
|
||||
assert_eq!(users(&state, community).await, before);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn verified_owner_is_committed_only_with_successful_v1_and_v2_claims() {
|
||||
let (state, host, community, issuer) = fixture().await;
|
||||
let v2 = mint_code(
|
||||
state.clone(),
|
||||
&host,
|
||||
&issuer,
|
||||
serde_json::json!({"max_uses":1}),
|
||||
)
|
||||
.await;
|
||||
let key = derive_invite_key(&state.relay_keypair);
|
||||
let (v1, _) = crate::invite_token::mint_invite(&key, community, 3600);
|
||||
for code in [v1, v2] {
|
||||
let owner = Keys::generate();
|
||||
let agent = Keys::generate();
|
||||
let tag = buzz_sdk::nip_oa::compute_auth_tag(&owner, &agent.public_key(), "").unwrap();
|
||||
let response = post_with_owner_tag(
|
||||
state.clone(),
|
||||
&host,
|
||||
"/api/invites/claim",
|
||||
&agent,
|
||||
serde_json::json!({"code":code}).to_string(),
|
||||
&tag,
|
||||
)
|
||||
.await;
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
assert_eq!(read_json(response).await["status"], "joined");
|
||||
assert!(state
|
||||
.db
|
||||
.is_agent_owner(
|
||||
community,
|
||||
agent.public_key().as_bytes(),
|
||||
owner.public_key().as_bytes()
|
||||
)
|
||||
.await
|
||||
.unwrap());
|
||||
let before = users(&state, community).await;
|
||||
let response = post_with_owner_tag(
|
||||
state.clone(),
|
||||
&host,
|
||||
"/api/invites/claim",
|
||||
&agent,
|
||||
serde_json::json!({"code":code}).to_string(),
|
||||
&tag,
|
||||
)
|
||||
.await;
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
assert_eq!(read_json(response).await["status"], "already_member");
|
||||
assert_eq!(users(&state, community).await, before);
|
||||
let impostor_owner = Keys::generate();
|
||||
let mismatched =
|
||||
buzz_sdk::nip_oa::compute_auth_tag(&impostor_owner, &agent.public_key(), "")
|
||||
.unwrap();
|
||||
let response = post_with_owner_tag(
|
||||
state.clone(),
|
||||
&host,
|
||||
"/api/invites/claim",
|
||||
&agent,
|
||||
serde_json::json!({"code":code}).to_string(),
|
||||
&mismatched,
|
||||
)
|
||||
.await;
|
||||
assert_eq!(response.status(), StatusCode::FORBIDDEN);
|
||||
assert_eq!(read_json(response).await["error"], "invite_owner_mismatch");
|
||||
assert_eq!(users(&state, community).await, before);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn banned_verified_owner_creates_no_users_in_either_claim_format() {
|
||||
let (state, host, community, issuer) = fixture().await;
|
||||
let v2 = mint_code(
|
||||
state.clone(),
|
||||
&host,
|
||||
&issuer,
|
||||
serde_json::json!({"max_uses":1}),
|
||||
)
|
||||
.await;
|
||||
let key = derive_invite_key(&state.relay_keypair);
|
||||
let (v1, _) = crate::invite_token::mint_invite(&key, community, 3600);
|
||||
let owner = Keys::generate();
|
||||
state
|
||||
.db
|
||||
.ban_community_member(
|
||||
community,
|
||||
owner.public_key().as_bytes(),
|
||||
issuer.public_key().as_bytes(),
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let before = users(&state, community).await;
|
||||
for code in [v1, v2] {
|
||||
let agent = Keys::generate();
|
||||
let tag = buzz_sdk::nip_oa::compute_auth_tag(&owner, &agent.public_key(), "").unwrap();
|
||||
let response = post_with_owner_tag(
|
||||
state.clone(),
|
||||
&host,
|
||||
"/api/invites/claim",
|
||||
&agent,
|
||||
serde_json::json!({"code":code}).to_string(),
|
||||
&tag,
|
||||
)
|
||||
.await;
|
||||
assert_eq!(response.status(), StatusCode::FORBIDDEN);
|
||||
assert_eq!(read_json(response).await["error"], "invite_restricted");
|
||||
assert_eq!(users(&state, community).await, before);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,7 +1,8 @@
|
||||
mod postgres_tests {
|
||||
pub(super) mod postgres_tests {
|
||||
use super::super::*;
|
||||
|
||||
async fn fixture() -> (Arc<AppState>, String, buzz_core::CommunityId, Keys) {
|
||||
pub(in crate::api::invites::postgres_tests) async fn fixture(
|
||||
) -> (Arc<AppState>, String, buzz_core::CommunityId, Keys) {
|
||||
let host = format!("restricted-invite-{}.example", Uuid::new_v4().simple());
|
||||
let state = invite_test_state(&host)
|
||||
.await
|
||||
@@ -21,7 +22,7 @@ mod postgres_tests {
|
||||
(state, host, community, owner)
|
||||
}
|
||||
|
||||
async fn post_with_owner_tag(
|
||||
pub(in crate::api::invites::postgres_tests) async fn post_with_owner_tag(
|
||||
state: Arc<AppState>,
|
||||
host: &str,
|
||||
path: &str,
|
||||
@@ -138,6 +139,42 @@ mod postgres_tests {
|
||||
.is_relay_member(community, &agent.public_key().to_hex())
|
||||
.await
|
||||
.unwrap());
|
||||
for key in [&agent, &owner] {
|
||||
assert!(state
|
||||
.db
|
||||
.get_user(community, key.public_key().as_bytes())
|
||||
.await
|
||||
.unwrap()
|
||||
.is_none());
|
||||
}
|
||||
state
|
||||
.db
|
||||
.unban_community_member(
|
||||
community,
|
||||
owner.public_key().as_bytes(),
|
||||
issuer.public_key().as_bytes(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let response = post_with_owner_tag(
|
||||
state.clone(),
|
||||
&host,
|
||||
"/api/invites/claim",
|
||||
&agent,
|
||||
serde_json::json!({"code":code}).to_string(),
|
||||
&tag,
|
||||
)
|
||||
.await;
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
assert!(state
|
||||
.db
|
||||
.is_agent_owner(
|
||||
community,
|
||||
agent.public_key().as_bytes(),
|
||||
owner.public_key().as_bytes()
|
||||
)
|
||||
.await
|
||||
.unwrap());
|
||||
let other_agent = Keys::generate();
|
||||
// A credential signed for another key cannot bind this claimant's owner.
|
||||
let response = post_with_owner_tag(
|
||||
|
||||
Reference in New Issue
Block a user