diff --git a/.env.example b/.env.example index 8e1fe159391..94f074f48ea 100644 --- a/.env.example +++ b/.env.example @@ -369,3 +369,7 @@ RUST_LOG=buzz_relay=debug,buzz_datastore=info,buzz_db=debug,buzz_auth=debug,buzz # BUZZ_TERMS_OF_SERVICE_MARKDOWN="# Terms of Service\n\nFull terms here." # BUZZ_PRIVACY_POLICY_MARKDOWN="# Privacy Policy\n\nFull policy here." # BUZZ_AGE_ATTESTATION_REQUIRED=true + +# Optional absolute legacy-v1 invite cutoff, UTC RFC3339 whole seconds. +# Unset preserves natural token expiry; never changes v2 invitations. +# BUZZ_INVITE_V1_INVALID_AFTER=2026-12-01T00:00:00Z diff --git a/Justfile b/Justfile index 04c6e46e700..a03a8fd600a 100644 --- a/Justfile +++ b/Justfile @@ -591,6 +591,10 @@ test-unit: + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=state::tests::manager_disconnect_sets_reason_enqueues_frame_then_cancels) + test(/^api::nip_fi::/)' + # Invitation drain/cutoff boundary contracts are infra-free. + cargo nextest run -p buzz-relay --lib -E 'test(/^invite_token::/) + test(/^config::invite_cutoff_tests::/)' + # Retain partially consumed mesh receives across housekeeping ticks. + cargo nextest run -p buzz-relay --lib -E 'test(=mesh_boot::tests::demo_echo_retains_pending_receive_across_housekeeping_ticks)' # boot_lifecycle spawns the real relay binary and asserts its startup # lifecycle, including that buzz_startup_phase_* reaches /metrics. Its # non-ignored tests need no Postgres or Redis; the Postgres cases are diff --git a/TESTING.md b/TESTING.md index d939265c414..719bb774e3e 100644 --- a/TESTING.md +++ b/TESTING.md @@ -146,6 +146,36 @@ backgrounded or you lost the terminal: `pkill -f buzz-relay`. Leaving it running will collide with the next reviewer who follows this doc on the same machine. +### Invitation compatibility and revocation + +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. +An issuer or verified owner timeout also blocks claims on existing invitations +while it lasts, without durable revocation. Claims resume after timeout expiry. +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 +lookup failures deny admission, and onboarding does not require prior membership. + +Legacy v1 bearers contain no issuer identity. Unset `BUZZ_INVITE_V1_INVALID_AFTER` +preserves their natural expiry and claimant restriction checks, but cannot revoke +an individual legacy issuer's bearers. Drain for at most 30 days from the last +possible v1 issuance across **all serving and rollback versions**; do not use merge +time as the start. Full issuer-revocation coverage requires completing that drain +or an explicitly approved compatibility scope decision. + +Operators may set `BUZZ_INVITE_V1_INVALID_AFTER` to an absolute UTC RFC3339 +whole-second timestamp (for example `2026-12-01T00:00:00Z`). v1 verification +rejects at and after that instant, never extends bearer expiry, and leaves v2 +unchanged. Invalid or empty values fail configuration. Apply the same setting to +all serving pods and retain it on compatible rollbacks; rolling back to a version +without this setting's enforcement restores legacy acceptance until natural expiry. +This change does not choose or deploy a cutoff. + ### 4. Smoke test the CLI against the relay End-to-end: generate an identity, create a channel, post a message, read it diff --git a/crates/buzz-db/src/error.rs b/crates/buzz-db/src/error.rs index 8c051a7a793..aa3882e0042 100644 --- a/crates/buzz-db/src/error.rs +++ b/crates/buzz-db/src/error.rs @@ -37,6 +37,14 @@ pub enum DbError { #[error("access denied: {0}")] AccessDenied(String), + /// Invitation admission or minting was denied by authoritative restrictions. + #[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), diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index b6ead54e06f..66638ed7eda 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -705,7 +705,12 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 60); + assert_eq!(migrations.len(), 61); + assert_eq!(migrations[60].version, 61); + assert!(migrations[60] + .sql + .as_str() + .contains("ADD COLUMN revoked_at")); assert_eq!(migrations[58].version, 59); assert!(migrations[58].sql.as_str().contains("buzz.push_enabled")); assert_eq!(migrations[59].version, 60); diff --git a/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs b/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs index b32b8bfbd44..107bbb38d5a 100644 --- a/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs +++ b/crates/buzz-db/src/store/event_follow_up_postgres_tests.rs @@ -1018,13 +1018,30 @@ async fn assert_async_isolation(db: &Db, community: CommunityId, keys: &Keys) { endpoint_grant, max_class, subscriptions, expires_at, updated_at) \ SELECT $1, author, installation_id, source_event_id, source_created_at, generation, \ active, endpoint_enabled, endpoint_hash, endpoint_grant, max_class, \ - subscriptions, expires_at, clock_timestamp() FROM push_leases WHERE community_id=$2", + subscriptions, expires_at, \ + (SELECT received_at + interval '1 second' FROM events WHERE community_id=$1 AND id=$3) \ + FROM push_leases WHERE community_id=$2", ) .bind(later.as_uuid()) .bind(community.as_uuid()) + .bind(before_enrollment.id.as_bytes().as_slice()) .execute(&mut *activation) .await .unwrap(); + let leases_strictly_newer: bool = sqlx::query_scalar( + "SELECT count(*) > 0 AND bool_and(l.updated_at > e.received_at) \ + FROM push_leases l JOIN events e ON e.community_id=l.community_id \ + WHERE l.community_id=$1 AND e.id=$2", + ) + .bind(later.as_uuid()) + .bind(before_enrollment.id.as_bytes().as_slice()) + .fetch_one(&mut *activation) + .await + .unwrap(); + assert!( + leases_strictly_newer, + "fixture leases must follow recorded receipt" + ); activation.commit().await.unwrap(); assert_eq!(match_count(&isolated, later, &before_enrollment).await, 0); assert_eq!( diff --git a/crates/buzz-db/src/store/moderation.rs b/crates/buzz-db/src/store/moderation.rs index cdcc339fe5b..ae9e30b152f 100644 --- a/crates/buzz-db/src/store/moderation.rs +++ b/crates/buzz-db/src/store/moderation.rs @@ -340,6 +340,11 @@ pub async fn ban_member( reason: Option<&str>, expires_at: Option>, ) -> Result<()> { + let mut tx = pool.begin().await?; + crate::deletion::DeletionStore::new(pool.clone()) + .guard_transaction(&mut tx, community) + .await?; + crate::relay_invite::lock_admission(&mut tx, community).await?; sqlx::query( r#" INSERT INTO community_bans ( @@ -358,9 +363,11 @@ pub async fn ban_member( .bind(expires_at) .bind(reason) .bind(actor) - .execute(pool) + .execute(&mut *tx) .await?; + crate::relay_invite::revoke_banned_issuer_invites(&mut tx, community, pubkey).await?; + tx.commit().await?; Ok(()) } @@ -504,6 +511,11 @@ pub async fn timeout_member( muted_until: DateTime, reason: Option<&str>, ) -> Result<()> { + let mut tx = pool.begin().await?; + crate::deletion::DeletionStore::new(pool.clone()) + .guard_transaction(&mut tx, community) + .await?; + crate::relay_invite::lock_admission(&mut tx, community).await?; sqlx::query( r#" INSERT INTO community_bans ( @@ -521,9 +533,11 @@ pub async fn timeout_member( .bind(muted_until) .bind(reason) .bind(actor) - .execute(pool) + .execute(&mut *tx) .await?; + crate::relay_invite::revoke_banned_issuer_invites(&mut tx, community, pubkey).await?; + tx.commit().await?; Ok(()) } diff --git a/crates/buzz-db/src/store/relay_admin_actions.rs b/crates/buzz-db/src/store/relay_admin_actions.rs index 0a876beb2ae..8ce80a471fd 100644 --- a/crates/buzz-db/src/store/relay_admin_actions.rs +++ b/crates/buzz-db/src/store/relay_admin_actions.rs @@ -458,8 +458,12 @@ pub async fn execute_ban_with_marker( reason: Option<&str>, ) -> Result { let mut tx = pool.begin().await?; + crate::deletion::DeletionStore::new(pool.clone()) + .guard_transaction(&mut tx, community_id) + .await?; + crate::relay_invite::lock_admission(&mut tx, community_id).await?; - // Verify lease ownership first — abort without touching domain rows if the + // Verify lease ownership before domain writes — abort if the // lease is already gone. This prevents the commit entirely on a stale worker. let owned: bool = sqlx::query_scalar( r#" @@ -508,6 +512,8 @@ pub async fn execute_ban_with_marker( .execute(&mut *tx) .await?; + crate::relay_invite::revoke_banned_issuer_invites(&mut tx, community_id, target_pubkey).await?; + let marker = sqlx::query( r#" UPDATE relay_admin_actions @@ -549,6 +555,10 @@ pub async fn execute_timeout_with_marker( reason: Option<&str>, ) -> Result { let mut tx = pool.begin().await?; + crate::deletion::DeletionStore::new(pool.clone()) + .guard_transaction(&mut tx, community_id) + .await?; + crate::relay_invite::lock_admission(&mut tx, community_id).await?; let owned: bool = sqlx::query_scalar( r#" @@ -596,6 +606,8 @@ pub async fn execute_timeout_with_marker( .execute(&mut *tx) .await?; + crate::relay_invite::revoke_banned_issuer_invites(&mut tx, community_id, target_pubkey).await?; + let marker = sqlx::query( r#" UPDATE relay_admin_actions diff --git a/crates/buzz-db/src/store/relay_invite.rs b/crates/buzz-db/src/store/relay_invite.rs index 4bb48e121ca..c864597cbb3 100644 --- a/crates/buzz-db/src/store/relay_invite.rs +++ b/crates/buzz-db/src/store/relay_invite.rs @@ -28,6 +28,11 @@ use sqlx::{PgPool, Row as _}; use crate::error::Result; use crate::{CommunityId, Db}; +mod claim_owner; +mod revocation; +pub use claim_owner::{claim_relay_membership_with_owner, MembershipClaimResult}; +pub(crate) use revocation::{revoke_banned_issuer_invites, revoke_restricted_agent_invites}; + /// 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. @@ -54,6 +59,32 @@ pub enum ClaimOutcome { Exhausted, /// No invite row matches `(community_id, token_hash)`. Invalid, + /// The issuer's invitation was permanently revoked. + Revoked, + /// The issuer or its owner is currently banned or timed out. + IssuerRestricted, + /// The claimant or its verified owner is banned or timed out. + 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 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. @@ -125,6 +156,10 @@ pub async fn mint_relay_invite( crate::deletion::DeletionStore::new(pool.clone()) .guard_transaction(&mut tx, community) .await?; + lock_admission(&mut tx, community).await?; + if principal_restricted(&mut tx, community, created_by).await? { + return Err(crate::DbError::InviteRestricted); + } let row = sqlx::query( "INSERT INTO relay_invites (community_id, token_hash, max_uses, expires_at, created_by) \ VALUES ($1, $2, $3, $4, $5) \ @@ -150,6 +185,46 @@ pub async fn mint_relay_invite( }) } +/// Lock invitation admission against community restriction and owner writes. +/// Call before reading restrictions or locking an invite row, and hold to commit. +pub(crate) async fn lock_admission( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community: CommunityId, +) -> Result<()> { + sqlx::query("SELECT invite_admission_lock($1)") + .bind(community.as_uuid()) + .execute(&mut **tx) + .await?; + Ok(()) +} + +/// Authoritative, transaction-local lookup; nonmembers need no user row. +pub(crate) async fn principal_restricted( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + community: CommunityId, + pubkey: &str, +) -> Result { + let bytes = hex::decode(pubkey) + .map_err(|_| crate::DbError::InvalidData("invalid invitation principal".into()))?; + if bytes.len() != 32 { + return Err(crate::DbError::InvalidData( + "invalid invitation principal".into(), + )); + } + Ok(sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM community_bans \ + WHERE community_id = $1 AND pubkey IN (\ + SELECT $2::bytea UNION ALL \ + SELECT agent_owner_pubkey FROM users WHERE community_id = $1 AND pubkey = $2\ + ) AND ((banned AND (ban_expires_at IS NULL OR ban_expires_at > clock_timestamp())) \ + OR muted_until > clock_timestamp()))", + ) + .bind(community.as_uuid()) + .bind(bytes) + .fetch_one(&mut **tx) + .await?) +} + fn log_claim_outcome( community: CommunityId, invite_id: Option, @@ -226,6 +301,29 @@ pub async fn claim_relay_invite( claimer_pubkey: &str, policy_version: Option<&str>, ) -> Result { + 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 { let connection = crate::observability::acquire_writer( pool, crate::observability::WriterOperation::Authorization, @@ -233,9 +331,19 @@ pub async fn claim_relay_invite( .await?; let mut tx = sqlx::Transaction::begin(connection, None).await?; + crate::deletion::DeletionStore::new(pool.clone()) + .guard_transaction(&mut tx, community) + .await?; + lock_admission(&mut tx, community).await?; + if claim_owner::claimant_restricted(&mut tx, community, claimer_pubkey, verified_owner).await? { + tx.rollback().await?; + log_claim_outcome(community, None, "restricted", None, None); + return Ok(ClaimOutcome::Restricted.into()); + } + // 2. SELECT FOR UPDATE — lock the invite row for the duration of this txn. let row = sqlx::query( - "SELECT id, max_uses, use_count, expires_at \ + "SELECT id, max_uses, use_count, expires_at, created_by, revoked_at \ FROM relay_invites \ WHERE community_id = $1 AND token_hash = $2 \ FOR UPDATE", @@ -249,7 +357,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")?; @@ -269,7 +377,22 @@ pub async fn claim_relay_invite( max_uses, Some(use_count), ); - return Ok(ClaimOutcome::Expired); + return Ok(ClaimOutcome::Expired.into()); + } + + let revoked_at: Option> = invite.try_get("revoked_at")?; + let issuer: String = invite.try_get("created_by")?; + let denial = if revoked_at.is_some() { + Some((ClaimOutcome::Revoked, "revoked")) + } else if principal_restricted(&mut tx, community, &issuer).await? { + Some((ClaimOutcome::IssuerRestricted, "issuer_restricted")) + } else { + None + }; + if let Some((outcome, label)) = denial { + tx.rollback().await?; + log_claim_outcome(community, Some(invite_id), label, max_uses, Some(use_count)); + return Ok(outcome.into()); } let uses_remaining = || max_uses.map(|mu| mu - use_count); @@ -295,6 +418,9 @@ pub async fn claim_relay_invite( .execute(&mut *tx) .await?; } + let owner = + claim_owner::persist_claim_owner(&mut tx, community, claimer_pubkey, verified_owner) + .await?; tx.commit().await?; log_claim_outcome( community, @@ -303,9 +429,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, }); } @@ -320,7 +450,7 @@ pub async fn claim_relay_invite( max_uses, Some(use_count), ); - return Ok(ClaimOutcome::Exhausted); + return Ok(ClaimOutcome::Exhausted.into()); } } @@ -354,6 +484,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, @@ -362,9 +495,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, }); } @@ -377,7 +514,10 @@ pub async fn claim_relay_invite( .execute(&mut *tx) .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); @@ -390,9 +530,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, }) } @@ -413,6 +557,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 { + 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) -> Result { @@ -451,7 +616,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") @@ -486,7 +651,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) @@ -517,11 +682,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", ) @@ -617,7 +782,7 @@ mod postgres_tests { .await .expect("begin quiescing"); - let error = mint_relay_invite(&pool, community, "owner", 3600, Some(1)) + let error = mint_relay_invite(&pool, community, &test_pubkey(), 3600, Some(1)) .await .expect_err("quiescing must reject invite minting"); assert!(matches!(error, crate::error::DbError::AccessDenied(_))); @@ -642,7 +807,7 @@ mod postgres_tests { let community = make_test_community(&pool).await; let first = test_pubkey(); let second = test_pubkey(); - let invite = mint_relay_invite(&pool, community, "owner", 3600, Some(1)) + let invite = mint_relay_invite(&pool, community, &test_pubkey(), 3600, Some(1)) .await .expect("mint bounded invite"); let hash = hash_v2_code(&invite.code); @@ -688,7 +853,7 @@ mod postgres_tests { let community = make_test_community(&pool).await; let first = test_pubkey(); let second = test_pubkey(); - let invite = mint_relay_invite(&pool, community, "owner", 3600, Some(1)) + let invite = mint_relay_invite(&pool, community, &test_pubkey(), 3600, Some(1)) .await .expect("mint bounded invite"); let hash = hash_v2_code(&invite.code); @@ -732,7 +897,7 @@ mod postgres_tests { let pool = setup_pool().await; let community_a = make_test_community(&pool).await; let community_b = make_test_community(&pool).await; - let invite = mint_relay_invite(&pool, community_a, "owner", 3600, Some(2)) + let invite = mint_relay_invite(&pool, community_a, &test_pubkey(), 3600, Some(2)) .await .expect("mint invite"); let hash = hash_v2_code(&invite.code); @@ -769,10 +934,10 @@ mod postgres_tests { async fn retention_sweep_deletes_only_invites_older_than_cutoff() { let pool = setup_pool().await; let community = make_test_community(&pool).await; - let old = mint_relay_invite(&pool, community, "owner", 3600, Some(1)) + let old = mint_relay_invite(&pool, community, &test_pubkey(), 3600, Some(1)) .await .expect("mint old invite"); - let recent = mint_relay_invite(&pool, community, "owner", 3600, Some(1)) + let recent = mint_relay_invite(&pool, community, &test_pubkey(), 3600, Some(1)) .await .expect("mint recent invite"); let cutoff = Utc::now() - chrono::Duration::days(30); @@ -874,7 +1039,7 @@ mod postgres_tests { async fn unlimited_invites_count_each_new_member() { let pool = setup_pool().await; let community = make_test_community(&pool).await; - let invite = mint_relay_invite(&pool, community, "owner", 3600, None) + let invite = mint_relay_invite(&pool, community, &test_pubkey(), 3600, None) .await .expect("mint unlimited invite"); let hash = hash_v2_code(&invite.code); @@ -900,7 +1065,7 @@ mod postgres_tests { let pool = setup_pool().await; let community = make_test_community(&pool).await; let pubkey = test_pubkey(); - let invite = mint_relay_invite(&pool, community, "owner", 3600, Some(1)) + let invite = mint_relay_invite(&pool, community, &test_pubkey(), 3600, Some(1)) .await .expect("mint bounded invite"); let hash = hash_v2_code(&invite.code); @@ -923,3 +1088,16 @@ mod postgres_tests { delete_test_community(&pool, community).await; } } + +#[cfg(test)] +mod revocation_postgres_tests; + +#[cfg(test)] +#[path = "relay_invite/upgrade_postgres_tests.rs"] +mod upgrade_postgres_tests; + +#[cfg(test)] +mod owner_admission_postgres_tests; + +#[cfg(test)] +mod writer_protocol_postgres_tests; diff --git a/crates/buzz-db/src/store/relay_invite/claim_owner.rs b/crates/buzz-db/src/store/relay_invite/claim_owner.rs new file mode 100644 index 00000000000..e7f96a48f91 --- /dev/null +++ b/crates/buzz-db/src/store/relay_invite/claim_owner.rs @@ -0,0 +1,150 @@ +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 { + 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>> = 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 { + 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; + // Defensive parity with other owner writers; admission already rejects banned owners. + if link_created { + super::revoke_restricted_agent_invites(tx, community, &claimant_bytes, &owner_bytes) + .await?; + } + 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 { + 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, + }) +} diff --git a/crates/buzz-db/src/store/relay_invite/owner_admission_postgres_tests.rs b/crates/buzz-db/src/store/relay_invite/owner_admission_postgres_tests.rs new file mode 100644 index 00000000000..9c9fe6de5af --- /dev/null +++ b/crates/buzz-db/src/store/relay_invite/owner_admission_postgres_tests.rs @@ -0,0 +1,160 @@ +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); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn issuer_timeout_denies_temporarily_without_revoking_invite() { + let pool = postgres_tests::setup_pool().await; + let community = postgres_tests::make_test_community(&pool).await; + let issuer = postgres_tests::test_pubkey(); + let claimant = postgres_tests::test_pubkey(); + let invite = mint_relay_invite(&pool, community, &issuer, 3600, Some(1)) + .await + .unwrap(); + sqlx::query("INSERT INTO community_bans (community_id,pubkey,muted_until,actor_pubkey) VALUES ($1,$2,clock_timestamp()+interval '1 hour',$2)") + .bind(community.as_uuid()).bind(hex::decode(&issuer).unwrap()).execute(&pool).await.unwrap(); + assert_eq!( + claim_relay_invite( + &pool, + community, + &hash_v2_code(&invite.code), + &claimant, + None + ) + .await + .unwrap(), + ClaimOutcome::IssuerRestricted + ); + let revoked: Option> = + sqlx::query_scalar("SELECT revoked_at FROM relay_invites WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(invite.invite_id) + .fetch_one(&pool) + .await + .unwrap(); + assert!(revoked.is_none()); + sqlx::query("UPDATE community_bans SET muted_until=clock_timestamp()-interval '1 second' WHERE community_id=$1") + .bind(community.as_uuid()).execute(&pool).await.unwrap(); + assert!(matches!( + claim_relay_invite( + &pool, + community, + &hash_v2_code(&invite.code), + &claimant, + None + ) + .await + .unwrap(), + ClaimOutcome::Joined { use_count: 1, .. } + )); +} diff --git a/crates/buzz-db/src/store/relay_invite/revocation.rs b/crates/buzz-db/src/store/relay_invite/revocation.rs new file mode 100644 index 00000000000..adf0846db05 --- /dev/null +++ b/crates/buzz-db/src/store/relay_invite/revocation.rs @@ -0,0 +1,52 @@ +//! Application-owned revocation, under the admission lock and in the writer's commit. +use crate::{error::Result, CommunityId}; +use sqlx::{Postgres, Transaction}; + +/// Durably revoke an active banned principal's invites and its verified agents' invites. +/// The caller holds the admission lock before writing restriction or invite rows. +pub(crate) async fn revoke_banned_issuer_invites( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + pubkey: &[u8], +) -> Result<()> { + sqlx::query( + "UPDATE relay_invites SET revoked_at = clock_timestamp() \ + WHERE community_id = $1 AND revoked_at IS NULL \ + AND (created_by = encode($2::bytea, 'hex') OR created_by IN (\ + SELECT encode(pubkey, 'hex') FROM users \ + WHERE community_id = $1 AND agent_owner_pubkey = $2\ + )) AND EXISTS (\ + SELECT 1 FROM community_bans WHERE community_id = $1 AND pubkey = $2 \ + AND banned AND (ban_expires_at IS NULL OR ban_expires_at > clock_timestamp())\ + )", + ) + .bind(community.as_uuid()) + .bind(pubkey) + .execute(&mut **tx) + .await?; + Ok(()) +} + +/// Durably revoke an agent's invites when its newly verified owner is actively banned. +/// Timeouts block admission temporarily but never durably revoke invites. +pub(crate) async fn revoke_restricted_agent_invites( + tx: &mut Transaction<'_, Postgres>, + community: CommunityId, + agent_pubkey: &[u8], + owner_pubkey: &[u8], +) -> Result<()> { + sqlx::query( + "UPDATE relay_invites SET revoked_at = clock_timestamp() \ + WHERE community_id = $1 AND created_by = encode($2::bytea, 'hex') \ + AND revoked_at IS NULL AND EXISTS (\ + SELECT 1 FROM community_bans WHERE community_id = $1 AND pubkey = $3 \ + AND banned AND (ban_expires_at IS NULL OR ban_expires_at > clock_timestamp())\ + )", + ) + .bind(community.as_uuid()) + .bind(agent_pubkey) + .bind(owner_pubkey) + .execute(&mut **tx) + .await?; + Ok(()) +} diff --git a/crates/buzz-db/src/store/relay_invite/revocation_postgres_tests.rs b/crates/buzz-db/src/store/relay_invite/revocation_postgres_tests.rs new file mode 100644 index 00000000000..03bfb5311f0 --- /dev/null +++ b/crates/buzz-db/src/store/relay_invite/revocation_postgres_tests.rs @@ -0,0 +1,390 @@ +use super::*; +use crate::moderation::{ban_member, unban_member}; +use uuid::Uuid; + +pub(super) async fn fixture() -> (PgPool, CommunityId, String, String) { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + let id = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(id) + .bind(format!("invite-{}.example", id.simple())) + .execute(&pool) + .await + .unwrap(); + ( + pool, + CommunityId::from_uuid(id), + hex::encode([1; 32]), + hex::encode([2; 32]), + ) +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn ban_permanently_revokes_issued_invites_but_not_another_tenant() { + let (pool, community, issuer, claimant) = fixture().await; + let other_id = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(other_id) + .bind(format!("other-{}.example", other_id.simple())) + .execute(&pool) + .await + .unwrap(); + let other = CommunityId::from_uuid(other_id); + let invite = mint_relay_invite(&pool, community, &issuer, 3600, None) + .await + .unwrap(); + let unaffected = mint_relay_invite(&pool, other, &issuer, 3600, None) + .await + .unwrap(); + ban_member(&pool, community, &[1; 32], &[3; 32], None, None) + .await + .unwrap(); + assert!(matches!( + mint_relay_invite(&pool, community, &issuer, 3600, None).await, + Err(crate::DbError::InviteRestricted) + )); + assert!(unban_member(&pool, community, &[1; 32], &[3; 32]) + .await + .unwrap()); + assert_eq!( + claim_relay_invite( + &pool, + community, + &hash_v2_code(&invite.code), + &claimant, + None + ) + .await + .unwrap(), + ClaimOutcome::Revoked + ); + assert!(matches!( + claim_relay_invite( + &pool, + other, + &hash_v2_code(&unaffected.code), + &claimant, + None + ) + .await + .unwrap(), + ClaimOutcome::Joined { .. } + )); + let renewed = mint_relay_invite(&pool, community, &issuer, 3600, None) + .await + .unwrap(); + assert!(matches!( + claim_relay_invite( + &pool, + community, + &hash_v2_code(&renewed.code), + &claimant, + None + ) + .await + .unwrap(), + ClaimOutcome::Joined { .. } + )); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn nonmember_restrictions_cover_both_claim_formats_and_owner() { + let (pool, community, issuer, claimant) = fixture().await; + let invite = mint_relay_invite(&pool, community, &issuer, 3600, Some(1)) + .await + .unwrap(); + ban_member(&pool, community, &[2; 32], &[3; 32], None, None) + .await + .unwrap(); + assert_eq!( + claim_relay_invite( + &pool, + community, + &hash_v2_code(&invite.code), + &claimant, + None + ) + .await + .unwrap(), + ClaimOutcome::Restricted + ); + assert!(matches!( + crate::relay_members::claim_relay_membership(&pool, community, &claimant, "member", None) + .await, + Err(crate::DbError::InviteRestricted) + )); + let agent = hex::encode([4; 32]); + sqlx::query("INSERT INTO users (community_id,pubkey) VALUES ($1,$2),($1,$3)") + .bind(community.as_uuid()) + .bind(&[2u8; 32][..]) + .bind(&[4u8; 32][..]) + .execute(&pool) + .await + .unwrap(); + crate::user::set_agent_owner(&pool, community, &[4; 32], &[2; 32]) + .await + .unwrap(); + assert_eq!( + claim_relay_invite(&pool, community, &hash_v2_code(&invite.code), &agent, None) + .await + .unwrap(), + ClaimOutcome::Restricted + ); + assert!( + !crate::relay_members::is_relay_member(&pool, community, &claimant) + .await + .unwrap() + ); + let count: i32 = + sqlx::query_scalar("SELECT use_count FROM relay_invites WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(invite.invite_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(count, 0); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn issuer_agent_revocation_survives_owner_unban() { + let (pool, community, issuer, claimant) = fixture().await; + sqlx::query("INSERT INTO users (community_id,pubkey) VALUES ($1,$2),($1,$3)") + .bind(community.as_uuid()) + .bind(&[1u8; 32][..]) + .bind(&[4u8; 32][..]) + .execute(&pool) + .await + .unwrap(); + crate::user::set_agent_owner(&pool, community, &[1; 32], &[4; 32]) + .await + .unwrap(); + let invite = mint_relay_invite(&pool, community, &issuer, 3600, None) + .await + .unwrap(); + ban_member(&pool, community, &[4; 32], &[3; 32], None, None) + .await + .unwrap(); + unban_member(&pool, community, &[4; 32], &[3; 32]) + .await + .unwrap(); + assert_eq!( + claim_relay_invite( + &pool, + community, + &hash_v2_code(&invite.code), + &claimant, + None + ) + .await + .unwrap(), + ClaimOutcome::Revoked + ); +} + +pub(super) async fn named_pool() -> (PgPool, String) { + let name = format!("invite-race-{}", Uuid::new_v4()); + let connection_name = name.clone(); + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .after_connect(move |conn, _| { + let name = connection_name.clone(); + Box::pin(async move { + sqlx::query("SELECT set_config('application_name',$1,false)") + .bind(name) + .execute(conn) + .await?; + Ok(()) + }) + }) + .connect(&crate::test_support::database_url()) + .await + .unwrap(); + (pool, name) +} + +pub(super) async fn wait_for_lock( + pool: &PgPool, + name: &str, + kind: &str, + task: &tokio::task::JoinHandle, +) { + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM pg_locks l JOIN pg_stat_activity a ON a.pid=l.pid \ + WHERE a.application_name=$1 AND l.locktype=$2 AND NOT l.granted)", + ) + .bind(name) + .bind(kind) + .fetch_one(pool) + .await + .unwrap(); + if waiting { + break; + } + assert!(!task.is_finished(), "writer bypassed expected {kind} lock"); + tokio::task::yield_now().await; + } + }) + .await + .expect("writer must observably wait on the expected lock"); +} + +async fn queued_claim( + pool: &PgPool, + community: CommunityId, + code: &str, + claimant: String, +) -> tokio::task::JoinHandle> { + let (claim_pool, name) = named_pool().await; + let hash = hash_v2_code(code); + let claim = tokio::spawn(async move { + claim_relay_invite(&claim_pool, community, &hash, &claimant, None).await + }); + wait_for_lock(pool, &name, "advisory", &claim).await; + claim +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn claimant_ban_and_timeout_commit_before_queued_claim_without_consuming_a_use() { + for timeout in [false, true] { + let (pool, community, issuer, claimant) = fixture().await; + let invite = mint_relay_invite(&pool, community, &issuer, 3600, Some(1)) + .await + .unwrap(); + // Hold the existing restriction row so the production writer stops only + // after acquiring admission. The claim must queue behind that writer. + ban_member(&pool, community, &[2; 32], &[3; 32], None, None) + .await + .unwrap(); + unban_member(&pool, community, &[2; 32], &[3; 32]) + .await + .unwrap(); + let mut row_lock = pool.begin().await.unwrap(); + sqlx::query("SELECT pubkey FROM community_bans WHERE community_id=$1 FOR UPDATE") + .bind(community.as_uuid()) + .fetch_one(&mut *row_lock) + .await + .unwrap(); + let (writer_pool, name) = named_pool().await; + let restriction = tokio::spawn(async move { + if timeout { + crate::moderation::timeout_member( + &writer_pool, + community, + &[2; 32], + &[3; 32], + Utc::now() + chrono::Duration::hours(1), + None, + ) + .await + } else { + ban_member(&writer_pool, community, &[2; 32], &[3; 32], None, None).await + } + }); + wait_for_lock(&pool, &name, "transactionid", &restriction).await; + let claim = queued_claim(&pool, community, &invite.code, claimant.clone()).await; + row_lock.commit().await.unwrap(); + restriction.await.unwrap().unwrap(); + assert_eq!(claim.await.unwrap().unwrap(), ClaimOutcome::Restricted); + let uses: i32 = sqlx::query_scalar( + "SELECT use_count FROM relay_invites WHERE community_id=$1 AND id=$2", + ) + .bind(community.as_uuid()) + .bind(invite.invite_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(uses, 0); + assert!( + !crate::relay_members::is_relay_member(&pool, community, &claimant) + .await + .unwrap() + ); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn owner_materialization_commits_before_queued_agent_claim() { + let (pool, community, issuer, agent) = fixture().await; + let invite = mint_relay_invite(&pool, community, &issuer, 3600, None) + .await + .unwrap(); + sqlx::query("INSERT INTO users(community_id,pubkey) VALUES($1,$2),($1,$3)") + .bind(community.as_uuid()) + .bind(vec![2u8; 32]) + .bind(vec![4u8; 32]) + .execute(&pool) + .await + .unwrap(); + ban_member(&pool, community, &[4; 32], &[3; 32], None, None) + .await + .unwrap(); + let mut row_lock = pool.begin().await.unwrap(); + sqlx::query("SELECT pubkey FROM users WHERE community_id=$1 AND pubkey=$2 FOR UPDATE") + .bind(community.as_uuid()) + .bind(vec![2u8; 32]) + .fetch_one(&mut *row_lock) + .await + .unwrap(); + let (writer_pool, name) = named_pool().await; + let owner = tokio::spawn(async move { + crate::user::set_agent_owner(&writer_pool, community, &[2; 32], &[4; 32]).await + }); + wait_for_lock(&pool, &name, "transactionid", &owner).await; + let claim = queued_claim(&pool, community, &invite.code, agent).await; + row_lock.commit().await.unwrap(); + assert!(owner.await.unwrap().unwrap()); + assert_eq!(claim.await.unwrap().unwrap(), ClaimOutcome::Restricted); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn admission_first_then_ban_preserves_the_committed_use_and_denies_retries() { + let (pool, community, issuer, claimant) = fixture().await; + let invite = mint_relay_invite(&pool, community, &issuer, 3600, Some(2)) + .await + .unwrap(); + assert!(matches!( + claim_relay_invite( + &pool, + community, + &hash_v2_code(&invite.code), + &claimant, + None + ) + .await + .unwrap(), + ClaimOutcome::Joined { use_count: 1, .. } + )); + ban_member(&pool, community, &[2; 32], &[3; 32], None, None) + .await + .unwrap(); + assert_eq!( + claim_relay_invite( + &pool, + community, + &hash_v2_code(&invite.code), + &claimant, + None + ) + .await + .unwrap(), + ClaimOutcome::Restricted + ); + let uses: i32 = + sqlx::query_scalar("SELECT use_count FROM relay_invites WHERE community_id=$1 AND id=$2") + .bind(community.as_uuid()) + .bind(invite.invite_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(uses, 1); +} diff --git a/crates/buzz-db/src/store/relay_invite/upgrade_postgres_tests.rs b/crates/buzz-db/src/store/relay_invite/upgrade_postgres_tests.rs new file mode 100644 index 00000000000..74ad7988651 --- /dev/null +++ b/crates/buzz-db/src/store/relay_invite/upgrade_postgres_tests.rs @@ -0,0 +1,119 @@ +use super::*; +use uuid::Uuid; + +async fn legacy_fixture() -> (PgPool, Vec) { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + crate::migration::run_migrations_through(&pool, 60) + .await + .unwrap(); + let mut communities = Vec::new(); + for index in 0..4 { + let community = Uuid::new_v4(); + sqlx::query("INSERT INTO communities(id,host) VALUES($1,$2)") + .bind(community) + .bind(format!("upgrade-{index}-{community}.example")) + .execute(&pool) + .await + .unwrap(); + sqlx::query("INSERT INTO relay_invites(community_id,token_hash,created_by,expires_at,created_at) VALUES($1,$2,$3,now()+interval '1 day',now()-interval '1 hour')") + .bind(community).bind(vec![index as u8;32]).bind(hex::encode([1;32])) + .execute(&pool).await.unwrap(); + if index != 3 { + // Historical accepted ban after issuance, already lifted by upgrade. + sqlx::query("INSERT INTO moderation_actions(community_id,actor_pubkey,action,target_pubkey) VALUES($1,$2,'ban',$3)") + .bind(community).bind(vec![2u8;32]).bind(vec![1u8;32]) + .execute(&pool).await.unwrap(); + } + if index == 1 || index == 2 { + let mut tx = pool.begin().await.unwrap(); + sqlx::query("SELECT set_config('buzz.deletion_executor_community',$1,true),set_config('buzz.deletion_fence_generation','7',true)") + .bind(community.to_string()).execute(&mut *tx).await.unwrap(); + sqlx::query( + "UPDATE communities SET deletion_state=$2,deletion_fence_generation=7 WHERE id=$1", + ) + .bind(community) + .bind(if index == 1 { "quiescing" } else { "tombstone" }) + .execute(&mut *tx) + .await + .unwrap(); + tx.commit().await.unwrap(); + } + communities.push(community); + } + (pool, communities) +} + +async fn assert_repaired_and_fenced(pool: &PgPool, communities: &[Uuid]) { + for (index, community) in communities.iter().enumerate() { + let revoked: bool = sqlx::query_scalar( + "SELECT revoked_at IS NOT NULL FROM relay_invites WHERE community_id=$1", + ) + .bind(community) + .fetch_one(pool) + .await + .unwrap(); + assert_eq!(revoked, index != 3, "tenant {index}"); + } + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT backfill_invite_revocations()") + .fetch_one(pool) + .await + .unwrap(), + 0, + "repair is idempotent" + ); + // The repair restores context; ordinary writes stay fenced afterward. + let error = sqlx::query("UPDATE relay_invites SET use_count=use_count+1 WHERE community_id=$1") + .bind(communities[1]) + .execute(pool) + .await + .unwrap_err(); + assert!( + error.to_string().contains("community write fenced"), + "{error}" + ); + let triggers: i64=sqlx::query_scalar("SELECT count(*) FROM pg_trigger WHERE NOT tgisinternal AND tgname IN ('invite_restriction_lock','invite_issuer_revocation','invite_owner_lock','invite_owner_revocation','community_write_fence_relay_invites')") + .fetch_one(pool).await.unwrap(); + assert_eq!(triggers, 1, "only the community deletion fence remains"); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_backfill_repairs_history_in_active_and_deletion_fenced_tenants() { + let (pool, communities) = legacy_fixture().await; + crate::migration::run_migrations(&pool).await.unwrap(); + assert_repaired_and_fenced(&pool, &communities).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_desired_upgrade_reconciles_the_same_populated_history() { + let (pool, communities) = legacy_fixture().await; + // Apply the actual desired-state function/trigger sources to the populated + // old schema, then the actual pgschema reconciliation entry point. + sqlx::raw_sql("ALTER TABLE relay_invites ADD COLUMN revoked_at TIMESTAMPTZ") + .execute(&pool) + .await + .unwrap(); + for sql in [ + include_str!("../../../../../schema/functions/invite_admission_lock.sql"), + include_str!("../../../../../schema/functions/backfill_invite_revocations.sql"), + ] { + sqlx::raw_sql(sql).execute(&pool).await.unwrap(); + } + let still_unrepaired: i64 = + sqlx::query_scalar("SELECT count(*) FROM relay_invites WHERE revoked_at IS NULL") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(still_unrepaired, 4, "DDL alone cannot repair history"); + sqlx::raw_sql(include_str!( + "../../../../../scripts/reconcile-schema-after-pgschema.sql" + )) + .execute(&pool) + .await + .unwrap(); + assert_repaired_and_fenced(&pool, &communities).await; +} diff --git a/crates/buzz-db/src/store/relay_invite/writer_protocol_postgres_tests.rs b/crates/buzz-db/src/store/relay_invite/writer_protocol_postgres_tests.rs new file mode 100644 index 00000000000..608ac81bb92 --- /dev/null +++ b/crates/buzz-db/src/store/relay_invite/writer_protocol_postgres_tests.rs @@ -0,0 +1,357 @@ +use super::revocation_postgres_tests::{fixture, named_pool, wait_for_lock}; +use super::*; + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn authorization_owner_writer_waits_before_claim_owner_row_and_links_once() { + let (pool, community, _, claimant) = fixture().await; + let owner = hex::encode([4; 32]); + for key in [[2; 32], [4; 32]] { + crate::user::ensure_user(&pool, community, &key) + .await + .unwrap(); + } + let mut claim = pool.begin().await.unwrap(); + lock_admission(&mut claim, community).await.unwrap(); + let (writer_pool, name) = named_pool().await; + let setter = tokio::spawn(async move { + Db::from_pool(writer_pool) + .set_agent_owner_for_authorization(community, &[2; 32], &[4; 32]) + .await + }); + wait_for_lock(&pool, &name, "advisory", &setter).await; + // Exercise the production claim persistence while the setter is queued. + // If the setter locks the user first, this bounded operation cannot finish. + let written = tokio::time::timeout( + std::time::Duration::from_secs(5), + super::claim_owner::persist_claim_owner(&mut claim, community, &claimant, Some(&owner)), + ) + .await + .unwrap() + .unwrap(); + assert!(written.link_created); + claim.commit().await.unwrap(); + assert!( + !tokio::time::timeout(std::time::Duration::from_secs(5), setter) + .await + .unwrap() + .unwrap() + .unwrap() + ); + assert_eq!( + crate::user::get_agent_channel_policy(&pool, community, &[2; 32]) + .await + .unwrap() + .unwrap() + .1, + Some(vec![4; 32]) + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn ensure_user_insert_and_conflict_do_not_take_admission_lock() { + let (pool, community, _, _) = fixture().await; + let mut claim = pool.begin().await.unwrap(); + lock_admission(&mut claim, community).await.unwrap(); + let (writer_pool, name) = named_pool().await; + for expected in [true, false] { + let mut table_lock = pool.begin().await.unwrap(); + sqlx::query("LOCK TABLE users IN ACCESS EXCLUSIVE MODE") + .execute(&mut *table_lock) + .await + .unwrap(); + let insert_pool = writer_pool.clone(); + let insert = tokio::spawn(async move { + crate::user::ensure_user(&insert_pool, community, &[5; 32]).await + }); + // Inspect the writer while its INSERT is in flight, not after its + // autocommit releases locks. It must reach the table behind admission. + wait_for_lock(&pool, &name, "relation", &insert).await; + let locks: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_locks l JOIN pg_stat_activity a ON a.pid=l.pid \ + WHERE a.application_name=$1 AND l.locktype='advisory'", + ) + .bind(&name) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!( + locks, 0, + "user insert must not hold or wait on advisory locks" + ); + table_lock.commit().await.unwrap(); + assert_eq!( + tokio::time::timeout(std::time::Duration::from_secs(5), insert) + .await + .unwrap() + .unwrap() + .unwrap(), + expected + ); + } + claim.rollback().await.unwrap(); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn linking_banned_owner_revokes_agent_invites_but_timeout_does_not() { + for banned in [true, false] { + let (pool, community, issuer, claimant) = fixture().await; + for key in [[1; 32], [4; 32]] { + crate::user::ensure_user(&pool, community, &key) + .await + .unwrap(); + } + let invite = mint_relay_invite(&pool, community, &issuer, 3600, Some(1)) + .await + .unwrap(); + if banned { + crate::moderation::ban_member(&pool, community, &[4; 32], &[3; 32], None, None) + .await + .unwrap(); + } else { + crate::moderation::timeout_member( + &pool, + community, + &[4; 32], + &[3; 32], + Utc::now() + chrono::Duration::hours(1), + None, + ) + .await + .unwrap(); + } + assert!( + crate::user::set_agent_owner(&pool, community, &[1; 32], &[4; 32]) + .await + .unwrap() + ); + let revoked: bool = + sqlx::query_scalar("SELECT revoked_at IS NOT NULL FROM relay_invites WHERE id=$1") + .bind(invite.invite_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(revoked, banned); + if banned { + crate::moderation::unban_member(&pool, community, &[4; 32], &[3; 32]) + .await + .unwrap(); + assert_eq!( + claim_relay_invite( + &pool, + community, + &hash_v2_code(&invite.code), + &claimant, + None + ) + .await + .unwrap(), + ClaimOutcome::Revoked + ); + } else { + assert_eq!( + claim_relay_invite( + &pool, + community, + &hash_v2_code(&invite.code), + &claimant, + None + ) + .await + .unwrap(), + ClaimOutcome::IssuerRestricted + ); + crate::moderation::untimeout_member(&pool, community, &[4; 32], &[3; 32]) + .await + .unwrap(); + assert!(matches!( + claim_relay_invite( + &pool, + community, + &hash_v2_code(&invite.code), + &claimant, + None + ) + .await + .unwrap(), + ClaimOutcome::Joined { .. } + )); + } + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn marker_writers_serialize_before_rows_and_commit_revocation_with_marker() { + for timeout in [false, true] { + let (pool, community, issuer, _) = fixture().await; + let invite = mint_relay_invite(&pool, community, &issuer, 3600, None) + .await + .unwrap(); + for key in [[1; 32], [4; 32]] { + crate::user::ensure_user(&pool, community, &key) + .await + .unwrap(); + } + crate::user::set_agent_owner(&pool, community, &[4; 32], &[1; 32]) + .await + .unwrap(); + let agent_invite = mint_relay_invite(&pool, community, &hex::encode([4; 32]), 3600, None) + .await + .unwrap(); + if timeout { + // A timeout must also repair revocation for an existing active ban, + // rather than revoking invites for timeout-only principals. + crate::moderation::ban_member(&pool, community, &[1; 32], &[3; 32], None, None) + .await + .unwrap(); + sqlx::query("UPDATE relay_invites SET revoked_at=NULL WHERE id IN ($1,$2)") + .bind(invite.invite_id) + .bind(agent_invite.invite_id) + .execute(&pool) + .await + .unwrap(); + } + let action = uuid::Uuid::new_v4(); + let lease = uuid::Uuid::new_v4(); + sqlx::query("INSERT INTO relay_admin_actions \ + (id, report_community_id, request_id, actor_pubkey, actor_role, action, state, \ + enforcement_target_pubkey, timeout_secs, timeout_until, action_lease_token, action_lease_expires_at) \ + VALUES ($1,$2,$1,$3,'operator',$4,'enforcing',$5, \ + CASE WHEN $6 THEN 3600 END, CASE WHEN $6 THEN clock_timestamp()+interval '1 hour' END, \ + $7,clock_timestamp()+interval '1 hour')") + .bind(action).bind(community.as_uuid()).bind(vec![3u8;32]) + .bind(if timeout {"timeout"} else {"ban"}).bind(vec![1u8;32]).bind(timeout).bind(lease) + .execute(&pool).await.unwrap(); + let mut admission = pool.begin().await.unwrap(); + lock_admission(&mut admission, community).await.unwrap(); + let (writer_pool, name) = named_pool().await; + let writer = tokio::spawn(async move { + if timeout { + crate::relay_admin_actions::execute_timeout_with_marker( + &writer_pool, + action, + lease, + community, + &[1; 32], + &[3; 32], + Utc::now() + chrono::Duration::hours(1), + None, + ) + .await + } else { + crate::relay_admin_actions::execute_ban_with_marker( + &writer_pool, + action, + lease, + community, + &[1; 32], + &[3; 32], + None, + ) + .await + } + }); + wait_for_lock(&pool, &name, "advisory", &writer).await; + // The action row must still be free while admission is held. This also + // prevents future inversion with action-row locking. + sqlx::query("SELECT id FROM relay_admin_actions WHERE id=$1 FOR UPDATE NOWAIT") + .bind(action) + .fetch_one(&mut *admission) + .await + .unwrap(); + admission.commit().await.unwrap(); + assert!(writer.await.unwrap().unwrap()); + let state: (bool, Option) = sqlx::query_as( + "SELECT i.revoked_at IS NOT NULL, a.step_marker FROM relay_invites i \ + JOIN relay_admin_actions a ON a.id=$2 WHERE i.id=$1", + ) + .bind(invite.invite_id) + .bind(action) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(state, (true, Some("mutation_committed".into()))); + let agent_revoked: bool = + sqlx::query_scalar("SELECT revoked_at IS NOT NULL FROM relay_invites WHERE id=$1") + .bind(agent_invite.invite_id) + .fetch_one(&pool) + .await + .unwrap(); + assert!(agent_revoked, "marker ban also revokes owned-agent invites"); + // Already marked: revocation and domain mutations must roll back too. + sqlx::query("UPDATE relay_invites SET revoked_at=NULL WHERE id IN ($1,$2)") + .bind(invite.invite_id) + .bind(agent_invite.invite_id) + .execute(&pool) + .await + .unwrap(); + let repeated = if timeout { + crate::relay_admin_actions::execute_timeout_with_marker( + &pool, + action, + lease, + community, + &[1; 32], + &[3; 32], + Utc::now() + chrono::Duration::hours(1), + None, + ) + .await + } else { + crate::relay_admin_actions::execute_ban_with_marker( + &pool, action, lease, community, &[1; 32], &[3; 32], None, + ) + .await + } + .unwrap(); + assert!(!repeated); + let revoked: bool = + sqlx::query_scalar("SELECT revoked_at IS NOT NULL FROM relay_invites WHERE id=$1") + .bind(invite.invite_id) + .fetch_one(&pool) + .await + .unwrap(); + assert!(!revoked, "failed marker must not commit revocation"); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn linked_authorization_owner_and_missing_agent_do_not_wait_on_admission() { + let (pool, community, _, _) = fixture().await; + for key in [[2; 32], [4; 32]] { + crate::user::ensure_user(&pool, community, &key) + .await + .unwrap(); + } + assert!( + crate::user::set_agent_owner(&pool, community, &[2; 32], &[4; 32]) + .await + .unwrap() + ); + let mut claim = pool.begin().await.unwrap(); + lock_admission(&mut claim, community).await.unwrap(); + let db = Db::from_pool(pool.clone()); + for owner in [[4; 32], [6; 32]] { + assert!(!tokio::time::timeout( + std::time::Duration::from_secs(5), + db.set_agent_owner_for_authorization(community, &[2; 32], &owner) + ) + .await + .unwrap() + .unwrap()); + } + assert!(matches!( + tokio::time::timeout( + std::time::Duration::from_secs(5), + db.set_agent_owner_for_authorization(community, &[7; 32], &[4; 32]) + ) + .await + .unwrap(), + Err(crate::DbError::NotFound(_)) + )); + claim.rollback().await.unwrap(); +} diff --git a/crates/buzz-db/src/store/relay_members.rs b/crates/buzz-db/src/store/relay_members.rs index 8db20820146..f273a702d6d 100644 --- a/crates/buzz-db/src/store/relay_members.rs +++ b/crates/buzz-db/src/store/relay_members.rs @@ -177,36 +177,16 @@ pub async fn claim_relay_membership( role: &str, policy_version: Option<&str>, ) -> Result { - let connection = - observability::acquire_writer(pool, observability::WriterOperation::Authorization).await?; - let mut tx = sqlx::Transaction::begin(connection, None).await?; - 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", + Ok(super::relay_invite::claim_relay_membership_with_owner( + pool, + community, + pubkey, + role, + policy_version, + None, ) - .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?; - } - - tx.commit().await?; - Ok(inserted) + .inserted) } /// Returns whether a member has persisted acceptance evidence for a policy version. @@ -943,6 +923,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::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( diff --git a/crates/buzz-db/src/store/user.rs b/crates/buzz-db/src/store/user.rs index ec4b897a617..58818ef3da8 100644 --- a/crates/buzz-db/src/store/user.rs +++ b/crates/buzz-db/src/store/user.rs @@ -330,6 +330,30 @@ async fn set_agent_owner_with_operation( operation: crate::observability::WriterOperation, ) -> Result { let mut connection = crate::observability::acquire_writer(pool, operation).await?; + // Ownership is first-write-wins and is never cleared by supported writers. + // Already-linked agents are the common authorization path: avoid queuing + // those read-only calls behind a community's claim/restriction transaction. + let recorded: Option>> = sqlx::query_scalar( + "SELECT agent_owner_pubkey FROM users WHERE community_id = $1 AND pubkey = $2", + ) + .bind(community_id.as_uuid()) + .bind(agent_pubkey) + .fetch_optional(&mut *connection) + .await?; + match recorded { + None => { + return Err(crate::error::DbError::NotFound( + "agent pubkey not found in users table".into(), + )) + } + Some(Some(_)) => return Ok(false), + Some(None) => {} + } + let mut tx = sqlx::Transaction::begin(connection, None).await?; + crate::deletion::DeletionStore::new(pool.clone()) + .guard_transaction(&mut tx, community_id) + .await?; + crate::relay_invite::lock_admission(&mut tx, community_id).await?; // Conditional UPDATE: only set owner if currently NULL. This makes // "first mint wins" atomic — no TOCTOU race between concurrent mints. let result = sqlx::query( @@ -338,7 +362,7 @@ async fn set_agent_owner_with_operation( .bind(owner_pubkey) .bind(community_id.as_uuid()) .bind(agent_pubkey) - .execute(&mut *connection) + .execute(&mut *tx) .await?; if result.rows_affected() == 0 { @@ -347,7 +371,7 @@ async fn set_agent_owner_with_operation( let exists = sqlx::query(r#"SELECT 1 FROM users WHERE community_id = $1 AND pubkey = $2"#) .bind(community_id.as_uuid()) .bind(agent_pubkey) - .fetch_optional(&mut *connection) + .fetch_optional(&mut *tx) .await?; if exists.is_none() { return Err(crate::error::DbError::NotFound( @@ -355,8 +379,17 @@ async fn set_agent_owner_with_operation( )); } // Row exists but owner already set — return false (not an error). + tx.commit().await?; return Ok(false); } + crate::relay_invite::revoke_restricted_agent_invites( + &mut tx, + community_id, + agent_pubkey, + owner_pubkey, + ) + .await?; + tx.commit().await?; Ok(true) } diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index d631df75003..69f31408bc6 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -8396,7 +8396,9 @@ pub(crate) mod postgres_tests { "CREATE SCHEMA {schema}; \ CREATE TABLE {schema}.users (LIKE public.users INCLUDING ALL); \ CREATE TABLE {schema}.community_bans (LIKE public.community_bans INCLUDING ALL); \ - CREATE TABLE {schema}.relay_members (LIKE public.relay_members INCLUDING ALL);" + CREATE TABLE {schema}.relay_members (LIKE public.relay_members INCLUDING ALL); \ + CREATE TABLE {schema}.communities (LIKE public.communities INCLUDING ALL); \ + CREATE TABLE {schema}.relay_invites (LIKE public.relay_invites INCLUDING ALL);" ))) .execute(state.db.pool()) .await @@ -8410,10 +8412,24 @@ pub(crate) mod postgres_tests { ) .await .expect("schema pool"); + for sql in [ + include_str!("../../../../schema/functions/community_deletion_lock_key.sql"), + include_str!("../../../../schema/functions/invite_admission_lock.sql"), + ] { + sqlx::raw_sql(sql) + .execute(&pool) + .await + .expect("writer lock functions"); + } Arc::get_mut(&mut state).expect("unique state").db = buzz_db::Db::from_pool(pool); let tenant = fresh_tenant("late-owner.test"); let other = fresh_tenant("late-owner-other.test"); + sqlx::query("INSERT INTO communities(id, host) VALUES($1,'late-owner.test')") + .bind(tenant.community().as_uuid()) + .execute(state.db.pool()) + .await + .expect("writer community"); let (owner, agent) = (Keys::generate(), Keys::generate()); let agent_bytes = agent.public_key().to_bytes(); @@ -8484,12 +8500,6 @@ pub(crate) mod postgres_tests { "{action}: the HTTP request recorded the owner" ); - sqlx::raw_sql(sqlx::AssertSqlSafe(format!( - "ALTER TABLE {schema}.users RENAME TO users_unavailable" - ))) - .execute(&admin) - .await - .expect("break the owner-to-agent lookup"); let owner_hex = owner.public_key().to_hex(); match action { "ban" => state @@ -8522,6 +8532,12 @@ pub(crate) mod postgres_tests { .expect("removal commits"); } } + sqlx::raw_sql(sqlx::AssertSqlSafe(format!( + "ALTER TABLE {schema}.users RENAME TO users_unavailable" + ))) + .execute(&admin) + .await + .expect("break the owner-to-agent lookup"); let revoked = state .revoke_live_access( &tenant, diff --git a/crates/buzz-relay/src/api/invites.rs b/crates/buzz-relay/src/api/invites.rs index bff1df0fcc8..3384f6023ec 100644 --- a/crates/buzz-relay/src/api/invites.rs +++ b/crates/buzz-relay/src/api/invites.rs @@ -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,40 +228,9 @@ 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, - headers: &HeaderMap, - path: &str, - body: &[u8], -) -> Result<(buzz_core::TenantContext, nostr::PublicKey), (StatusCode, Json)> { - 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, - .. - } = bridge::verify_nip98_exempt_invite_claim(headers, "POST", &url, Some(body))?; - bridge::check_nip98_replay(state, &tenant, event_id_bytes).await?; - - Ok((tenant, pubkey)) -} - fn map_mint_error(error: buzz_db::DbError) -> (StatusCode, Json) { match error { + buzz_db::DbError::InviteRestricted => api_error(StatusCode::FORBIDDEN, "invite_restricted"), buzz_db::DbError::InvalidData(message) | buzz_db::DbError::DeletionSafety(message) => { api_error(StatusCode::BAD_REQUEST, &message) } @@ -342,6 +313,20 @@ async fn mint_invite_checked( return e.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 { Ok(json) => json.into_response(), Err(e) => e.into_response(), @@ -429,7 +414,8 @@ pub async fn claim_invite( headers: HeaderMap, body: axum::body::Bytes, ) -> Result, (StatusCode, Json)> { - 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( @@ -442,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 --- @@ -466,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, @@ -475,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(), @@ -516,6 +508,13 @@ pub async fn claim_invite( buzz_db::relay_invite::ClaimOutcome::Exhausted => { Err(api_error(StatusCode::FORBIDDEN, "invite_exhausted")) } + buzz_db::relay_invite::ClaimOutcome::Revoked + | buzz_db::relay_invite::ClaimOutcome::IssuerRestricted => { + Err(api_error(StatusCode::FORBIDDEN, "invite_invalid")) + } + buzz_db::relay_invite::ClaimOutcome::Restricted => { + Err(api_error(StatusCode::FORBIDDEN, "invite_restricted")) + } buzz_db::relay_invite::ClaimOutcome::Invalid => { Err(api_error(StatusCode::FORBIDDEN, "invite_invalid")) } @@ -523,16 +522,18 @@ pub async fn claim_invite( } // --- v1 HMAC path (stateless tokens, drain window) --- - let payload = invite_token::verify_invite(&key, tenant.community(), &request.code).map_err( - |e| match e { - // Expired is post-MAC: revealing it helps the UX without helping a forger. - invite_token::InviteError::Expired => { - api_error(StatusCode::FORBIDDEN, "invite_expired") - } - // Everything else stays coarse so the endpoint is a poor oracle. - _ => api_error(StatusCode::FORBIDDEN, "invite_invalid"), - }, - )?; + let payload = invite_token::verify_invite_with_cutoff( + &key, + tenant.community(), + &request.code, + state.config.invite_v1_invalid_after, + ) + .map_err(|e| match e { + // Expired is post-MAC: revealing it helps the UX without helping a forger. + invite_token::InviteError::Expired => api_error(StatusCode::FORBIDDEN, "invite_expired"), + // Everything else stays coarse so the endpoint is a poor oracle. + _ => api_error(StatusCode::FORBIDDEN, "invite_invalid"), + })?; if let Some(policy) = &state.config.join_policy { let receipt = request @@ -543,9 +544,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, @@ -554,9 +555,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!("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!( @@ -608,6 +616,8 @@ 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; diff --git a/crates/buzz-relay/src/api/invites/claim_auth.rs b/crates/buzz-relay/src/api/invites/claim_auth.rs new file mode 100644 index 00000000000..756b2ba848a --- /dev/null +++ b/crates/buzz-relay/src/api/invites/claim_auth.rs @@ -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, + headers: &HeaderMap, + path: &str, + body: &[u8], +) -> Result< + ( + buzz_core::TenantContext, + nostr::PublicKey, + Option, + ), + (StatusCode, Json), +> { + 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) { + 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}")), + } +} diff --git a/crates/buzz-relay/src/api/invites/postgres_tests/owner_admission_tests.rs b/crates/buzz-relay/src/api/invites/postgres_tests/owner_admission_tests.rs new file mode 100644 index 00000000000..5288accb9d7 --- /dev/null +++ b/crates/buzz-relay/src/api/invites/postgres_tests/owner_admission_tests.rs @@ -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); + } + } +} diff --git a/crates/buzz-relay/src/api/invites/postgres_tests/restriction_tests.rs b/crates/buzz-relay/src/api/invites/postgres_tests/restriction_tests.rs new file mode 100644 index 00000000000..7f4f61baf2b --- /dev/null +++ b/crates/buzz-relay/src/api/invites/postgres_tests/restriction_tests.rs @@ -0,0 +1,337 @@ +pub(super) mod postgres_tests { + use super::super::*; + + pub(in crate::api::invites::postgres_tests) async fn fixture( + ) -> (Arc, String, buzz_core::CommunityId, Keys) { + let host = format!("restricted-invite-{}.example", Uuid::new_v4().simple()); + let state = invite_test_state(&host) + .await + .expect("isolated PostgreSQL state"); + let community = state + .db + .ensure_configured_community(&host) + .await + .unwrap() + .id; + let owner = Keys::generate(); + state + .db + .add_relay_member(community, &owner.public_key().to_hex(), "owner", None) + .await + .unwrap(); + (state, host, community, owner) + } + + pub(in crate::api::invites::postgres_tests) async fn post_with_owner_tag( + state: Arc, + host: &str, + path: &str, + agent: &Keys, + body: String, + auth_tag: &str, + ) -> axum::response::Response { + let auth = nip98_auth_header(agent, &format!("https://{host}{path}"), body.as_bytes()); + build_router(state) + .oneshot( + Request::builder() + .method("POST") + .uri(path) + .header(header::HOST, host) + .header(header::AUTHORIZATION, auth) + .header("x-auth-tag", auth_tag) + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(body)) + .unwrap(), + ) + .await + .unwrap() + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn signed_nonmember_claims_reject_restrictions_without_consuming_invite() { + let (state, host, community, issuer) = fixture().await; + let code = mint_code( + state.clone(), + &host, + &issuer, + serde_json::json!({"max_uses":1}), + ) + .await; + let claimant = Keys::generate(); + state + .db + .ban_community_member( + community, + claimant.public_key().as_bytes(), + issuer.public_key().as_bytes(), + None, + None, + ) + .await + .unwrap(); + let key = derive_invite_key(&state.relay_keypair); + let (legacy, _) = crate::invite_token::mint_invite(&key, community, 3600); + for bearer in [&code, &legacy] { + let response = post_json( + state.clone(), + &host, + "/api/invites/claim", + &claimant, + serde_json::json!({"code":bearer}).to_string(), + ) + .await; + assert_eq!(response.status(), StatusCode::FORBIDDEN); + assert_eq!(read_json(response).await["error"], "invite_restricted"); + } + assert!(!state + .db + .is_relay_member(community, &claimant.public_key().to_hex()) + .await + .unwrap()); + let unrestricted = Keys::generate(); + assert_eq!( + post_json( + state, + &host, + "/api/invites/claim", + &unrestricted, + serde_json::json!({"code":code}).to_string() + ) + .await + .status(), + StatusCode::OK + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn verified_owner_blocks_first_agent_claim_but_forged_owner_does_not_bind() { + let (state, host, community, issuer) = fixture().await; + let code = mint_code(state.clone(), &host, &issuer, serde_json::json!({})).await; + 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 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!(!state + .db + .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( + state.clone(), + &host, + "/api/invites/claim", + &other_agent, + serde_json::json!({"code":code}).to_string(), + &tag, + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + assert!(!state + .db + .is_agent_owner( + community, + other_agent.public_key().as_bytes(), + owner.public_key().as_bytes() + ) + .await + .unwrap()); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn committed_issuer_ban_survives_unban_through_http() { + let (state, host, community, issuer) = fixture().await; + let code = mint_code(state.clone(), &host, &issuer, serde_json::json!({})).await; + state + .db + .ban_community_member( + community, + issuer.public_key().as_bytes(), + issuer.public_key().as_bytes(), + None, + None, + ) + .await + .unwrap(); + state + .db + .unban_community_member( + community, + issuer.public_key().as_bytes(), + issuer.public_key().as_bytes(), + ) + .await + .unwrap(); + let response = post_json( + state.clone(), + &host, + "/api/invites/claim", + &Keys::generate(), + serde_json::json!({"code":code}).to_string(), + ) + .await; + assert_eq!(response.status(), StatusCode::FORBIDDEN); + assert_eq!(read_json(response).await["error"], "invite_invalid"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn cutoff_rejects_only_v1_and_restriction_lookup_failure_never_admits() { + let Some(identity) = [ + "NEXTEST_RUN_ID", + "NEXTEST_BINARY_ID", + "NEXTEST_TEST_NAME", + "NEXTEST_ATTEMPT_ID", + ] + .iter() + .map(|name| std::env::var(name).ok()) + .collect::>>() else { + eprintln!("skipping destructive fixture outside postgres-isolation wrapper"); + return; + }; + let url = std::env::var("BUZZ_TEST_DATABASE_URL").unwrap_or_default(); + assert!( + wrapper_owns_database(&url, &identity.join(":")), + "refusing destructive DDL on a shared database" + ); + let (state, host, community, issuer) = fixture().await; + let code = mint_code(state.clone(), &host, &issuer, serde_json::json!({})).await; + let key = derive_invite_key(&state.relay_keypair); + let (legacy, _) = crate::invite_token::mint_invite(&key, community, 3600); + let Ok(mut state) = Arc::try_unwrap(state) else { + panic!("exclusive fixture state"); + }; + Arc::make_mut(&mut state.config).invite_v1_invalid_after = Some(0); + let state = Arc::new(state); + let response = post_json( + state.clone(), + &host, + "/api/invites/claim", + &Keys::generate(), + serde_json::json!({"code":legacy}).to_string(), + ) + .await; + assert_eq!(response.status(), StatusCode::FORBIDDEN); + assert_eq!(read_json(response).await["error"], "invite_expired"); + assert_eq!( + post_json( + state.clone(), + &host, + "/api/invites/claim", + &Keys::generate(), + serde_json::json!({"code":code}).to_string() + ) + .await + .status(), + StatusCode::OK + ); + // This test owns its per-process database; never alters a shared fixture. + sqlx::query("ALTER TABLE community_bans RENAME TO unavailable_restrictions") + .execute(state.db.pool()) + .await + .unwrap(); + let claimant = Keys::generate(); + let response = post_json( + state.clone(), + &host, + "/api/invites/claim", + &claimant, + serde_json::json!({"code":code}).to_string(), + ) + .await; + let status = response.status(); + let membership = state + .db + .is_relay_member(community, &claimant.public_key().to_hex()) + .await; + sqlx::query("ALTER TABLE unavailable_restrictions RENAME TO community_bans") + .execute(state.db.pool()) + .await + .unwrap(); + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR); + assert!(!membership.unwrap()); + } + + fn wrapper_owns_database(url: &str, identity: &str) -> bool { + let hash = hex::encode(Sha256::digest(identity.as_bytes())); + url.rsplit('/').next() == Some(format!("buzz_nt_{}", &hash[..24]).as_str()) + } + + #[test] + fn destructive_fixture_rejects_shared_database_urls() { + let identity = "run:binary:test:attempt"; + let hash = hex::encode(Sha256::digest(identity.as_bytes())); + assert!(wrapper_owns_database( + &format!("postgres://localhost/buzz_nt_{}", &hash[..24]), + identity + )); + for url in [ + "postgres://localhost/buzz", + "postgres://localhost/buzz_nt_other", + "", + ] { + assert!(!wrapper_owns_database(url, identity)); + } + } +} diff --git a/crates/buzz-relay/src/api/mesh_demo.rs b/crates/buzz-relay/src/api/mesh_demo.rs index 8649b97671a..77dc9aaed84 100644 --- a/crates/buzz-relay/src/api/mesh_demo.rs +++ b/crates/buzz-relay/src/api/mesh_demo.rs @@ -336,8 +336,9 @@ mod tests { }, ) .await; - assert_eq!(resp.status(), StatusCode::OK); + let status = resp.status(); let body = body_json(resp).await; + assert_eq!(status, StatusCode::OK, "forwarded response: {body}"); assert_eq!(body["outcome"], "forwarded"); assert_eq!(body["echoed_payload"], "mesh echo evidence"); owner_task.abort(); diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 44662bb9e10..049b104fe24 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -379,6 +379,9 @@ pub struct Config { /// documents or age attestation are configured. pub join_policy: Option, + /// Absolute UTC Unix-second cutoff for legacy v1 invites; unset drains naturally. + pub invite_v1_invalid_after: Option, + /// Deployment-admin API and SPA configuration. Absent means the surface is disabled. pub admin: Option, @@ -405,6 +408,28 @@ fn parse_bind_addr(raw: &str) -> Result { .map_err(|e| ConfigError::InvalidBindAddr(e.to_string())) } +fn parse_invite_v1_cutoff(raw: Option<&str>) -> Result, ConfigError> { + raw.map(|value| { + chrono::DateTime::parse_from_rfc3339(value) + .ok() + .filter(|time| time.offset().local_minus_utc() == 0 && time.timestamp_subsec_nanos() == 0) + .and_then(|time| u64::try_from(time.timestamp()).ok()) + .ok_or_else(|| ConfigError::InvalidValue( + "BUZZ_INVITE_V1_INVALID_AFTER must be a UTC RFC3339 timestamp with whole seconds".into(), + )) + }).transpose() +} + +fn invite_v1_cutoff_from_env() -> Result, ConfigError> { + match std::env::var("BUZZ_INVITE_V1_INVALID_AFTER") { + Ok(raw) => parse_invite_v1_cutoff(Some(&raw)), + Err(std::env::VarError::NotPresent) => Ok(None), + Err(std::env::VarError::NotUnicode(_)) => Err(ConfigError::InvalidValue( + "BUZZ_INVITE_V1_INVALID_AFTER must be valid Unicode".into(), + )), + } +} + fn positive_u64_from_env(name: &str, default: u64) -> Result { match std::env::var(name) { Ok(raw) => raw @@ -1433,6 +1458,7 @@ impl Config { push_gateway_timeout, operator_listener_timeout, join_policy, + invite_v1_invalid_after: invite_v1_cutoff_from_env()?, admin, web_dir, serve_git_web_gui, @@ -2871,3 +2897,27 @@ mod tests { ); } } + +#[cfg(test)] +mod invite_cutoff_tests { + use super::*; + + #[test] + fn cutoff_configuration_is_optional_and_strict() { + assert_eq!(parse_invite_v1_cutoff(None).unwrap(), None); + assert_eq!( + parse_invite_v1_cutoff(Some("1970-01-01T00:01:40Z")).unwrap(), + Some(100) + ); + for raw in [ + "", + "tomorrow", + "100", + "2026-10-10T00:00:00+01:00", + "1969-12-31T23:59:59Z", + "2026-10-10T00:00:00.1Z", + ] { + assert!(parse_invite_v1_cutoff(Some(raw)).is_err(), "{raw}"); + } + } +} diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 564ad3d00dd..fed8e848071 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -6960,8 +6960,8 @@ mod postgres_tests { use sqlx::postgres::PgConnectOptions; use tokio_util::sync::CancellationToken; - // A schema with bans but no `users` table: the ban commits, and only - // the owned-agent lookup fails. + // A supported ban writer commits before the users table is taken away: + // only the subsequent owned-agent lookup fails. let db_url = crate::test_support::database_url(); let admin = sqlx::PgPool::connect(&db_url) .await @@ -6969,7 +6969,10 @@ mod postgres_tests { let schema = format!("revoke_owner_{}", Uuid::new_v4().simple()); sqlx::raw_sql(sqlx::AssertSqlSafe(format!( "CREATE SCHEMA {schema}; \ - CREATE TABLE {schema}.community_bans (LIKE public.community_bans INCLUDING ALL);" + CREATE TABLE {schema}.community_bans (LIKE public.community_bans INCLUDING ALL); \ + CREATE TABLE {schema}.users (LIKE public.users INCLUDING ALL); \ + CREATE TABLE {schema}.communities (LIKE public.communities INCLUDING ALL); \ + CREATE TABLE {schema}.relay_invites (LIKE public.relay_invites INCLUDING ALL);" ))) .execute(&admin) .await @@ -6981,8 +6984,22 @@ mod postgres_tests { let pool = sqlx::PgPool::connect_with(options) .await .expect("schema pool"); + for sql in [ + include_str!("../../../../schema/functions/community_deletion_lock_key.sql"), + include_str!("../../../../schema/functions/invite_admission_lock.sql"), + ] { + sqlx::raw_sql(sql) + .execute(&pool) + .await + .expect("writer lock functions"); + } let state = build_canvas_ingest_state(&db_url, &pool).await; let community = buzz_core::tenant::CommunityId::from_uuid(Uuid::new_v4()); + sqlx::query("INSERT INTO communities(id, host) VALUES($1,'revoke-owner.test')") + .bind(community.as_uuid()) + .execute(&pool) + .await + .expect("writer community"); let tenant = TenantContext::resolved(community, "revoke-owner.test".to_string()); let (owner, agent, bystander) = ( nostr::Keys::generate(), @@ -7040,6 +7057,10 @@ mod postgres_tests { ) .await .expect("ban commits"); + sqlx::raw_sql(sqlx::AssertSqlSafe(format!("DROP TABLE {schema}.users"))) + .execute(&admin) + .await + .expect("break only the post-commit owner lookup"); let revoked = state .revoke_live_access( &tenant, diff --git a/crates/buzz-relay/src/invite_token.rs b/crates/buzz-relay/src/invite_token.rs index e271f40f0b5..9886b781530 100644 --- a/crates/buzz-relay/src/invite_token.rs +++ b/crates/buzz-relay/src/invite_token.rs @@ -157,6 +157,27 @@ pub fn verify_invite( key: &[u8; 32], community: CommunityId, code: &str, +) -> Result { + verify_invite_at(key, community, code, now_unix(), None) +} + +/// Verify a legacy bearer with the configured absolute invalid-after time. +/// v2 admission never calls this verifier. +pub fn verify_invite_with_cutoff( + key: &[u8; 32], + community: CommunityId, + code: &str, + invalid_after: Option, +) -> Result { + verify_invite_at(key, community, code, now_unix(), invalid_after) +} + +fn verify_invite_at( + key: &[u8; 32], + community: CommunityId, + code: &str, + now: u64, + invalid_after: Option, ) -> Result { if code.len() > MAX_CODE_LEN { return Err(InviteError::Malformed); @@ -178,7 +199,7 @@ pub fn verify_invite( let payload: InvitePayload = serde_json::from_slice(&payload_bytes).map_err(|_| InviteError::Malformed)?; - if payload.e < now_unix() { + if payload.e < now || invalid_after.is_some_and(|cutoff| now >= cutoff) { return Err(InviteError::Expired); } if payload.c != community.as_uuid().to_string() { @@ -404,3 +425,41 @@ mod policy_acceptance_tests { assert!(verify_policy_acceptance(&[8_u8; 32], &receipt, "invite-a", "v1").is_err()); } } + +#[cfg(test)] +mod cutoff_tests { + use super::*; + + #[test] + fn cutoff_never_extends_expiry_and_rejects_at_boundary() { + let key = [42; 32]; + let community = CommunityId::from_uuid(uuid::Uuid::new_v4()); + let payload = InvitePayload { + c: community.to_string(), + r: "member".into(), + e: 200, + n: "fixture".into(), + }; + let bytes = serde_json::to_vec(&payload).unwrap(); + let code = format!( + "{}.{}", + URL_SAFE_NO_PAD.encode(&bytes), + URL_SAFE_NO_PAD.encode(sign_payload(&key, &bytes)) + ); + for (now, cutoff, allowed) in [ + (100, None, true), + (199, None, true), + (201, None, false), + (99, Some(100), true), + (100, Some(100), false), + (101, Some(100), false), + (201, Some(300), false), + ] { + assert_eq!( + verify_invite_at(&key, community, &code, now, cutoff).is_ok(), + allowed, + "now={now} cutoff={cutoff:?}" + ); + } + } +} diff --git a/crates/buzz-relay/src/mesh_boot.rs b/crates/buzz-relay/src/mesh_boot.rs index 63a346258f5..d18d7ff8ca0 100644 --- a/crates/buzz-relay/src/mesh_boot.rs +++ b/crates/buzz-relay/src/mesh_boot.rs @@ -421,24 +421,39 @@ pub(crate) async fn run_demo_echo( tracing::info!(%session_id, %peer, "mesh demo echo: session open"); let mut drain_tick = tokio::time::interval(std::time::Duration::from_millis(100)); loop { - let frame = tokio::select! { - _ = drain_tick.tick() => { - if shutting_down.load(Ordering::Relaxed) { - if let Some(community_id) = stream.community_id() { - if let Err(e) = stream.send_goodbye(community_id, GoodbyeReason::Draining).await { - tracing::warn!(%session_id, "mesh demo echo: draining goodbye failed: {e}"); - } else { - tracing::info!(%session_id, "mesh demo echo: sent draining goodbye"); + let frame = { + // A receive can consume bytes before waiting on Redis validation. + // Keep it alive across housekeeping ticks; only shutdown may cancel it. + let recv = stream.recv_validated(&directory); + tokio::pin!(recv); + loop { + tokio::select! { + _ = drain_tick.tick() => { + if shutting_down.load(Ordering::Relaxed) { + break None; } - } else { - let _ = stream.finish(); - tracing::info!(%session_id, "mesh demo echo: drain before community latch — closing"); } - return; + frame = &mut recv => break Some(frame), } - continue; } - frame = stream.recv_validated(&directory) => frame, + }; + // The pending receive and its mutable stream borrow have ended before + // the drain response or next echo uses the stream. + let Some(frame) = frame else { + if let Some(community_id) = stream.community_id() { + if let Err(e) = stream + .send_goodbye(community_id, GoodbyeReason::Draining) + .await + { + tracing::warn!(%session_id, "mesh demo echo: draining goodbye failed: {e}"); + } else { + tracing::info!(%session_id, "mesh demo echo: sent draining goodbye"); + } + } else { + let _ = stream.finish(); + tracing::info!(%session_id, "mesh demo echo: drain before community latch — closing"); + } + return; }; match frame { Ok(Some(ReliableFrame::Data(payload))) => { @@ -696,6 +711,72 @@ mod tests { MeshStream::new(Box::new(StubSend), Box::new(StubRecv)) } + struct PausedRecv { + started: Option>, + resume: Option>, + calls: Arc, + } + + impl StreamRecvHalf for PausedRecv { + fn recv_frame(&mut self) -> BoxFuture<'_, Result, MeshError>> { + self.calls.fetch_add(1, Ordering::Relaxed); + let started = self.started.take(); + let resume = self.resume.take(); + Box::pin(async move { + if let Some(started) = started { + let _ = started.send(()); + } + if let Some(resume) = resume { + resume.await.expect("test resumes the in-flight receive"); + } + Ok(None) + }) + } + } + + /// Housekeeping must retain a receive that already consumed transport state. + /// Returning to recv_frame after a tick loses that state and closes the stream. + #[tokio::test(start_paused = true)] + async fn demo_echo_retains_pending_receive_across_housekeeping_ticks() { + let pool = deadpool_redis::Config::from_url("redis://127.0.0.1:1") + .create_pool(Some(deadpool_redis::Runtime::Tokio1)) + .expect("test pool"); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (resume_tx, resume_rx) = tokio::sync::oneshot::channel(); + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let fenced = FencedHeader { + session_id: uuid::Uuid::new_v4(), + generation: 1, + owner_runtime_id: rid(1), + }; + let stream = MeshStream::new( + Box::new(StubSend), + Box::new(PausedRecv { + started: Some(started_tx), + resume: Some(resume_rx), + calls: Arc::clone(&calls), + }), + ); + let worker = tokio::spawn(run_demo_echo( + SessionDirectory::new(pool), + ReliableInbound { + fenced, + from: rid(2), + stream: crate::tunnel::reliable::ReliableMeshStream::new_inbound(fenced, stream), + }, + Arc::new(AtomicBool::new(false)), + )); + started_rx + .await + .expect("production consumer started receiving"); + tokio::time::advance(std::time::Duration::from_millis(300)).await; + tokio::task::yield_now().await; + assert_eq!(calls.load(Ordering::Relaxed), 1, "tick restarted a receive"); + assert!(!worker.is_finished(), "tick abandoned the pending receive"); + resume_tx.send(()).expect("receive remains alive"); + worker.await.expect("consumer completed after stream EOF"); + } + fn rid(byte: u8) -> RuntimeId { RuntimeId([byte; 32]) } diff --git a/crates/buzz-relay/src/tunnel/reliable.rs b/crates/buzz-relay/src/tunnel/reliable.rs index 5153e104df6..8238d410830 100644 --- a/crates/buzz-relay/src/tunnel/reliable.rs +++ b/crates/buzz-relay/src/tunnel/reliable.rs @@ -329,6 +329,9 @@ impl ReliableMeshStream { /// pinned fenced tuple and the Redis directory. This is the reliable-stream /// equivalent of Dawn's hot-path media floor, but authoritative: stale or /// mismatched frames fail the session rather than being dropped silently. + /// This receive is not cancellation-safe: it can consume frame bytes before + /// awaiting directory validation. Preserve an in-flight receive across + /// housekeeping ticks; cancellation is only safe when terminating the stream. pub async fn recv_validated( &mut self, directory: &SessionDirectory, diff --git a/crates/buzz-relay/tests/boot_lifecycle.rs b/crates/buzz-relay/tests/boot_lifecycle.rs index 027fc4d1c26..ad6f20f17bd 100644 --- a/crates/buzz-relay/tests/boot_lifecycle.rs +++ b/crates/buzz-relay/tests/boot_lifecycle.rs @@ -168,20 +168,38 @@ fn wait_for_relay_metrics(process: &mut RelayProcess, port: u16) -> String { /// [`METRICS_SCRAPE_DEADLINE`]. A relay that exits first is a failure, not a /// timeout, so the panic names the real cause. fn wait_for_scraped_metric(process: &mut RelayProcess, port: u16, needle: &str) -> String { + wait_for_scraped_metrics(process, port, &[needle]) +} + +/// Waits for one scrape to contain every requested sample, bounded by +/// [`METRICS_SCRAPE_DEADLINE`]. A scrape can overlap the sampler's sequential +/// metric writes, so related series may appear in different responses. +fn wait_for_scraped_metrics(process: &mut RelayProcess, port: u16, needles: &[&str]) -> String { + let requested = needles.join(", "); + let mut last_scrape = String::new(); + let mut last_error: Option; let deadline = Instant::now() + METRICS_SCRAPE_DEADLINE; loop { assert!( process.try_wait().is_none(), - "relay exited before exporting {needle}" + "relay exited before exporting {requested}" ); - if let Ok(response) = scrape_metrics(port) { - if response.contains(needle) { - return response; + match scrape_metrics(port) { + Ok(response) => { + if needles + .iter() + .all(|needle| response.lines().any(|line| line.starts_with(needle))) + { + return response; + } + last_scrape = response; + last_error = None; } + Err(error) => last_error = Some(error.to_string()), } assert!( Instant::now() < deadline, - "relay did not export {needle} within {METRICS_SCRAPE_DEADLINE:?}" + "relay did not export {requested} within {METRICS_SCRAPE_DEADLINE:?}; last error: {last_error:?}; last scrape: {last_scrape}" ); thread::sleep(Duration::from_millis(20)); } @@ -814,10 +832,10 @@ mod postgres_tests { /// fires immediately, so the wait is bounded by /// [`METRICS_SCRAPE_DEADLINE`] and never a fixed sleep. /// - /// The completion-timestamp gauge is the needle because `sample` writes it - /// last, after the cache already serves the report: observing it proves the - /// whole cycle ran, with no window where the counters have landed but the - /// timestamp has not. + /// The sampler writes counters and histograms before the completion gauge, + /// but a concurrent `/metrics` scrape is not an atomic snapshot. Wait for + /// the actual timestamp, counter, and duration samples together so a scrape + /// that overlaps those sequential writes cannot report a false failure. #[test] #[ignore = "requires PostgreSQL"] fn startup_owns_the_dependency_sampler() { @@ -839,10 +857,14 @@ mod postgres_tests { ("REDIS_URL", &redis_url), ("BUZZ_GIT_CONFORMANCE_PROBE", "false"), ]); - let scrape = wait_for_scraped_metric( + let scrape = wait_for_scraped_metrics( &mut process, metrics_port, - "buzz_readiness_dependency_sample_completed_timestamp_seconds", + &[ + "buzz_readiness_dependency_sample_completed_timestamp_seconds ", + "buzz_readiness_dependency_checks_total{", + "buzz_readiness_check_duration_seconds_count{check=\"overall\"}", + ], ); let output = process.terminate(); let logs = format!( @@ -851,14 +873,6 @@ mod postgres_tests { String::from_utf8_lossy(&output.stderr) ); - assert!( - scrape.contains("buzz_readiness_dependency_checks_total{"), - "the same cycle must publish its per-dependency outcomes: {scrape}" - ); - assert!( - scrape.contains("buzz_readiness_check_duration_seconds"), - "a completed evaluation must publish its latency too: {scrape}" - ); assert!( !scrape.contains("buzz_readiness_checks_total{"), "no readiness probe was sent, so the sampler alone produced this: {scrape}" diff --git a/migrations/0061_invite_revocation.sql b/migrations/0061_invite_revocation.sql new file mode 100644 index 00000000000..41eb0702f87 --- /dev/null +++ b/migrations/0061_invite_revocation.sql @@ -0,0 +1,75 @@ +ALTER TABLE relay_invites ADD COLUMN revoked_at TIMESTAMPTZ; + +-- Serialize invitation admission with restriction and ownership changes per tenant. +CREATE FUNCTION invite_admission_lock(community UUID) RETURNS VOID +LANGUAGE sql AS $$ + SELECT pg_advisory_xact_lock(hashtextextended('invite-admission:' || community::text, 0)); +$$; + +-- Live revocation is application-owned: supported restriction and owner writers +-- take invite_admission_lock before row locks and revoke in the same commit. +-- Rollout gate: drain all old writer pods before lifting bans or reopening +-- invitation admission, then rerun backfill_invite_revocations() under the +-- schema-destruction maintenance lock to repair bans committed during rollout. +-- Include active bans and recorded prior bans, so lifting a ban before this +-- migration cannot restore an invite issued before that ban. Scope every owner +-- and audit lookup to the invitation's community. +-- Schema migrations hold the exclusive schema-destruction lock. Retain the +-- universal tenant fence too: acquire its shared lock, then read the current +-- generation before applying this narrowly scoped historical security repair. +-- A quiescing tenant may return to active, so skipping it could revive bearers. +CREATE FUNCTION backfill_invite_revocations() RETURNS BIGINT +LANGUAGE plpgsql AS $$ +DECLARE + target RECORD; + affected BIGINT; + total BIGINT := 0; + current_generation BIGINT; + prior_community TEXT := current_setting('buzz.deletion_executor_community', true); + prior_generation TEXT := current_setting('buzz.deletion_fence_generation', true); +BEGIN + FOR target IN SELECT id FROM communities + WHERE EXISTS (SELECT 1 FROM relay_invites WHERE community_id = communities.id) + ORDER BY id + LOOP + PERFORM pg_advisory_xact_lock_shared(community_deletion_lock_key(target.id)); + SELECT deletion_fence_generation INTO STRICT current_generation + FROM communities WHERE id = target.id; + PERFORM set_config('buzz.deletion_executor_community', target.id::text, true); + PERFORM set_config('buzz.deletion_fence_generation', current_generation::text, true); + UPDATE relay_invites i SET revoked_at = clock_timestamp() + WHERE i.community_id = target.id AND i.revoked_at IS NULL AND EXISTS ( + SELECT 1 FROM ( + SELECT CASE WHEN i.created_by ~ '^[0-9a-f]{64}$' + THEN decode(i.created_by, 'hex') END AS pubkey + UNION ALL + SELECT agent_owner_pubkey FROM users + WHERE community_id = i.community_id AND encode(pubkey, 'hex') = i.created_by + AND agent_owner_pubkey IS NOT NULL + ) principal + WHERE EXISTS ( + SELECT 1 FROM community_bans b WHERE b.community_id = i.community_id + AND b.pubkey = principal.pubkey AND b.banned + AND (b.ban_expires_at IS NULL OR b.ban_expires_at > clock_timestamp()) + ) OR EXISTS ( + SELECT 1 FROM moderation_actions a WHERE a.community_id = i.community_id + AND a.target_pubkey = principal.pubkey AND a.action IN ('ban', 'resolve:ban') + AND a.created_at >= i.created_at + ) OR EXISTS ( + SELECT 1 FROM relay_admin_actions a WHERE a.report_community_id = i.community_id + AND a.enforcement_target_pubkey = principal.pubkey AND a.action = 'ban' + AND (a.step_marker = 'mutation_committed' OR a.state = 'succeeded') + AND a.updated_at >= i.created_at + ) + ) + ; + GET DIAGNOSTICS affected = ROW_COUNT; + total := total + affected; + PERFORM set_config('buzz.deletion_executor_community', COALESCE(prior_community, ''), true); + PERFORM set_config('buzz.deletion_fence_generation', COALESCE(prior_generation, ''), true); + END LOOP; + RETURN total; +END +$$; + +SELECT backfill_invite_revocations(); diff --git a/schema/functions/backfill_invite_revocations.sql b/schema/functions/backfill_invite_revocations.sql new file mode 100644 index 00000000000..e434de59905 --- /dev/null +++ b/schema/functions/backfill_invite_revocations.sql @@ -0,0 +1,65 @@ +-- Live revocation is application-owned: supported restriction and owner writers +-- take invite_admission_lock before row locks and revoke in the same commit. +-- Rollout gate: drain all old writer pods before lifting bans or reopening +-- invitation admission, then rerun backfill_invite_revocations() under the +-- schema-destruction maintenance lock to repair bans committed during rollout. +-- Include active bans and recorded prior bans, so lifting a ban before this +-- migration cannot restore an invite issued before that ban. Scope every owner +-- and audit lookup to the invitation's community. +-- Schema migrations hold the exclusive schema-destruction lock. Retain the +-- universal tenant fence too: acquire its shared lock, then read the current +-- generation before applying this narrowly scoped historical security repair. +-- A quiescing tenant may return to active, so skipping it could revive bearers. +CREATE FUNCTION backfill_invite_revocations() RETURNS BIGINT +LANGUAGE plpgsql AS $$ +DECLARE + target RECORD; + affected BIGINT; + total BIGINT := 0; + current_generation BIGINT; + prior_community TEXT := current_setting('buzz.deletion_executor_community', true); + prior_generation TEXT := current_setting('buzz.deletion_fence_generation', true); +BEGIN + FOR target IN SELECT id FROM communities + WHERE EXISTS (SELECT 1 FROM relay_invites WHERE community_id = communities.id) + ORDER BY id + LOOP + PERFORM pg_advisory_xact_lock_shared(community_deletion_lock_key(target.id)); + SELECT deletion_fence_generation INTO STRICT current_generation + FROM communities WHERE id = target.id; + PERFORM set_config('buzz.deletion_executor_community', target.id::text, true); + PERFORM set_config('buzz.deletion_fence_generation', current_generation::text, true); + UPDATE relay_invites i SET revoked_at = clock_timestamp() + WHERE i.community_id = target.id AND i.revoked_at IS NULL AND EXISTS ( + SELECT 1 FROM ( + SELECT CASE WHEN i.created_by ~ '^[0-9a-f]{64}$' + THEN decode(i.created_by, 'hex') END AS pubkey + UNION ALL + SELECT agent_owner_pubkey FROM users + WHERE community_id = i.community_id AND encode(pubkey, 'hex') = i.created_by + AND agent_owner_pubkey IS NOT NULL + ) principal + WHERE EXISTS ( + SELECT 1 FROM community_bans b WHERE b.community_id = i.community_id + AND b.pubkey = principal.pubkey AND b.banned + AND (b.ban_expires_at IS NULL OR b.ban_expires_at > clock_timestamp()) + ) OR EXISTS ( + SELECT 1 FROM moderation_actions a WHERE a.community_id = i.community_id + AND a.target_pubkey = principal.pubkey AND a.action IN ('ban', 'resolve:ban') + AND a.created_at >= i.created_at + ) OR EXISTS ( + SELECT 1 FROM relay_admin_actions a WHERE a.report_community_id = i.community_id + AND a.enforcement_target_pubkey = principal.pubkey AND a.action = 'ban' + AND (a.step_marker = 'mutation_committed' OR a.state = 'succeeded') + AND a.updated_at >= i.created_at + ) + ) + ; + GET DIAGNOSTICS affected = ROW_COUNT; + total := total + affected; + PERFORM set_config('buzz.deletion_executor_community', COALESCE(prior_community, ''), true); + PERFORM set_config('buzz.deletion_fence_generation', COALESCE(prior_generation, ''), true); + END LOOP; + RETURN total; +END +$$; diff --git a/schema/functions/invite_admission_lock.sql b/schema/functions/invite_admission_lock.sql new file mode 100644 index 00000000000..c5d7098d3f7 --- /dev/null +++ b/schema/functions/invite_admission_lock.sql @@ -0,0 +1,5 @@ +-- Serialize invitation admission with restriction and ownership changes per tenant. +CREATE FUNCTION invite_admission_lock(community UUID) RETURNS VOID +LANGUAGE sql AS $$ + SELECT pg_advisory_xact_lock(hashtextextended('invite-admission:' || community::text, 0)); +$$; diff --git a/schema/schema.sql b/schema/schema.sql index 5ec8bd5829a..61cf2ecb440 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -79,6 +79,9 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i functions/enforce_community_tombstone.sql \i functions/attach_community_write_fence.sql +\i functions/invite_admission_lock.sql +\i functions/backfill_invite_revocations.sql + -- tables \i tables/public/communities.sql \i tables/public/channels.sql diff --git a/schema/tables/public/relay_invites.sql b/schema/tables/public/relay_invites.sql index 4e0f2eab0a3..65a7a9d63de 100644 --- a/schema/tables/public/relay_invites.sql +++ b/schema/tables/public/relay_invites.sql @@ -11,6 +11,7 @@ CREATE TABLE relay_invites ( max_uses INTEGER CHECK (max_uses BETWEEN 1 AND 10000), use_count INTEGER NOT NULL DEFAULT 0 CHECK (use_count >= 0), expires_at TIMESTAMPTZ NOT NULL, + revoked_at TIMESTAMPTZ, created_by TEXT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), PRIMARY KEY (community_id, id), diff --git a/scripts/reconcile-schema-after-pgschema.sql b/scripts/reconcile-schema-after-pgschema.sql index 7eef8c4bbe5..9b96615ecdd 100644 --- a/scripts/reconcile-schema-after-pgschema.sql +++ b/scripts/reconcile-schema-after-pgschema.sql @@ -248,3 +248,19 @@ BEGIN RAISE EXCEPTION 'replica_heartbeat must contain its singleton row after pgschema apply'; END IF; END $$; + +-- Invitation revocation is historical DML; pgschema does not execute it. +-- Use the same tenant-fenced, idempotent repair as migration0061. +SELECT backfill_invite_revocations(); +DO $$ +BEGIN + IF backfill_invite_revocations() <> 0 THEN + RAISE EXCEPTION 'invitation revocation backfill did not converge'; + END IF; + IF (SELECT count(*) FROM pg_trigger WHERE NOT tgisinternal + AND tgname IN ('invite_restriction_lock', 'invite_issuer_revocation', + 'invite_owner_lock', 'invite_owner_revocation')) <> 0 THEN + RAISE EXCEPTION 'invitation revocation must use supported application writers, not triggers'; + END IF; +END +$$; diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index 4f5ed92ba36..818e2ccd53f 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -339,6 +339,14 @@ run_unit_tests() { run_test_step "buzz-relay binary tests" \ cargo test -p buzz-relay --bin buzz-relay -- --nocapture + run_test_step "buzz-relay invitation token and cutoff contracts" \ + cargo test -p buzz-relay --lib invite_token:: -- --nocapture + run_test_step "buzz-relay invitation cutoff configuration" \ + cargo test -p buzz-relay --lib config::invite_cutoff_tests:: -- --nocapture + + run_test_step "buzz-relay mesh receive lifetime regression" \ + cargo test -p buzz-relay --lib mesh_boot::tests::demo_echo_retains_pending_receive_across_housekeeping_ticks -- --exact --nocapture + run_test_step "buzz-relay boot lifecycle tests" \ cargo test -p buzz-relay --test boot_lifecycle -- --nocapture }