diff --git a/.env.example b/.env.example index 6674deec40..7123f0353b 100644 --- a/.env.example +++ b/.env.example @@ -372,3 +372,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/TESTING.md b/TESTING.md index d939265c41..f4a4bde5e6 100644 --- a/TESTING.md +++ b/TESTING.md @@ -146,6 +146,31 @@ 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. +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 e27b319bed..e0f3f3e27b 100644 --- a/crates/buzz-db/src/error.rs +++ b/crates/buzz-db/src/error.rs @@ -37,6 +37,10 @@ pub enum DbError { #[error("access denied: {0}")] AccessDenied(String), + /// Invitation admission or minting was denied by authoritative restrictions. + #[error("invitation principal is restricted")] + InviteRestricted, + /// 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 7cb0879ff8..7f16dd339d 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -709,7 +709,7 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 66); + assert_eq!(migrations.len(), 67); assert_eq!(migrations[64].version, 65); let retire_ttl = strip_sql_comments(migrations[64].sql.as_str()); for statement in [ @@ -722,6 +722,14 @@ mod postgres_tests { let desired_schema = crate::test_support::desired_state_schema_sql(); assert!(!desired_schema.contains("events_refresh_channel_ttl")); assert!(!desired_schema.contains("refresh_channel_ttl_after_event_insert")); + assert_eq!(migrations[65].version, 66); + assert!(migrations[65] + .sql + .as_str() + .contains("CREATE TABLE community_nip_fi_issuers")); + assert!(migrations + .iter() + .any(|migration| migration.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/relay_invite.rs b/crates/buzz-db/src/store/relay_invite.rs index b6d76fb734..c8fb4339a1 100644 --- a/crates/buzz-db/src/store/relay_invite.rs +++ b/crates/buzz-db/src/store/relay_invite.rs @@ -55,6 +55,10 @@ pub enum ClaimOutcome { Exhausted, /// No invite row matches `(community_id, token_hash)`. Invalid, + /// The issuer's invitation was permanently revoked. + Revoked, + /// The claimant or its verified owner is banned or timed out. + Restricted, } /// A freshly minted v2 invite, including the plaintext code and metadata. @@ -123,6 +127,10 @@ pub async fn mint_relay_invite( crate::observability::WriterOperation::Authorization, ) .await?; + lock_admission(tx.conn(), community).await?; + if principal_restricted(tx.conn(), 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) \ @@ -148,6 +156,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::PgConnection, + 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::PgConnection, + 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, @@ -263,9 +311,14 @@ pub async fn claim_relay_invite( ) .await?; + lock_admission(tx.conn(), community).await?; + if principal_restricted(tx.conn(), community, claimer_pubkey).await? { + return Ok(ClaimOutcome::Restricted); + } + // 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", @@ -302,6 +355,12 @@ pub async fn claim_relay_invite( return Ok(ClaimOutcome::Expired); } + let revoked_at: Option> = invite.try_get("revoked_at")?; + let issuer: String = invite.try_get("created_by")?; + if revoked_at.is_some() || principal_restricted(tx.conn(), community, &issuer).await? { + return Ok(ClaimOutcome::Revoked); + } + let uses_remaining = || max_uses.map(|mu| mu - use_count); // 5. Check existing membership. @@ -647,7 +706,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(_))); @@ -672,7 +731,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); @@ -718,7 +777,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); @@ -762,7 +821,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); @@ -799,10 +858,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); @@ -904,7 +963,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); @@ -930,7 +989,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); @@ -953,3 +1012,10 @@ 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; 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 0000000000..a472cab39c --- /dev/null +++ b/crates/buzz-db/src/store/relay_invite/revocation_postgres_tests.rs @@ -0,0 +1,332 @@ +use super::*; +use crate::moderation::{ban_member, unban_member}; +use uuid::Uuid; + +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 + ); +} + +async fn queued_claim( + pool: &PgPool, + community: CommunityId, + code: &str, + claimant: String, +) -> tokio::task::JoinHandle> { + let name = format!("invite-race-{}", Uuid::new_v4()); + let connection_name = name.clone(); + let claim_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(); + let hash = hash_v2_code(code); + let claim = tokio::spawn(async move { + claim_relay_invite(&claim_pool, community, &hash, &claimant, None).await + }); + 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='advisory' AND NOT l.granted)") + .bind(&name).fetch_one(pool).await.unwrap(); + if waiting {break;} + assert!(!claim.is_finished(),"claim bypassed the production admission lock"); + tokio::task::yield_now().await; + } + }).await.expect("claim must observably wait on the admission lock"); + 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(); + let mut restriction = pool.begin().await.unwrap(); + sqlx::query("INSERT INTO community_bans(community_id,pubkey,banned,muted_until,actor_pubkey) VALUES($1,$2,$3,CASE WHEN $3 THEN NULL ELSE clock_timestamp()+interval '1 hour' END,$4)") + .bind(community.as_uuid()).bind(vec![2u8;32]).bind(!timeout).bind(vec![3u8;32]) + .execute(&mut *restriction).await.unwrap(); + let claim = queued_claim(&pool, community, &invite.code, claimant.clone()).await; + restriction.commit().await.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 owner = pool.begin().await.unwrap(); + sqlx::query("UPDATE users SET agent_owner_pubkey=$3 WHERE community_id=$1 AND pubkey=$2") + .bind(community.as_uuid()) + .bind(vec![2u8; 32]) + .bind(vec![4u8; 32]) + .execute(&mut *owner) + .await + .unwrap(); + let claim = queued_claim(&pool, community, &invite.code, agent).await; + owner.commit().await.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 0000000000..0bc964b7f0 --- /dev/null +++ b/crates/buzz-db/src/store/relay_invite/upgrade_postgres_tests.rs @@ -0,0 +1,120 @@ +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 { + 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(&pool) + .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, 5); +} + +#[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/lock_invite_restrictions.sql"), + include_str!("../../../../../schema/functions/revoke_banned_issuer_invites.sql"), + include_str!("../../../../../schema/functions/revoke_restricted_agent_invites.sql"), + include_str!("../../../../../schema/functions/backfill_invite_revocations.sql"), + ] { + sqlx::raw_sql(sql).execute(&pool).await.unwrap(); + } + sqlx::raw_sql("CREATE TRIGGER invite_restriction_lock BEFORE INSERT OR UPDATE ON community_bans FOR EACH ROW EXECUTE FUNCTION lock_invite_restrictions(); CREATE TRIGGER invite_issuer_revocation AFTER INSERT OR UPDATE ON community_bans FOR EACH ROW EXECUTE FUNCTION revoke_banned_issuer_invites(); CREATE TRIGGER invite_owner_lock BEFORE INSERT OR UPDATE OF agent_owner_pubkey ON users FOR EACH ROW EXECUTE FUNCTION lock_invite_restrictions(); CREATE TRIGGER invite_owner_revocation AFTER INSERT OR UPDATE OF agent_owner_pubkey ON users FOR EACH ROW EXECUTE FUNCTION revoke_restricted_agent_invites();") + .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_members.rs b/crates/buzz-db/src/store/relay_members.rs index 6b21c7ce01..66d3785d59 100644 --- a/crates/buzz-db/src/store/relay_members.rs +++ b/crates/buzz-db/src/store/relay_members.rs @@ -188,6 +188,10 @@ pub async fn claim_relay_membership( observability::WriterOperation::Authorization, ) .await?; + super::relay_invite::lock_admission(tx.conn(), community).await?; + if super::relay_invite::principal_restricted(tx.conn(), community, pubkey).await? { + return Err(DbError::InviteRestricted); + } let inserted = sqlx::query( "INSERT INTO relay_members (community_id, pubkey, role, added_by) \ VALUES ($1, $2, $3, 'invite') \ diff --git a/crates/buzz-relay/src/api/invites.rs b/crates/buzz-relay/src/api/invites.rs index bff1df0fcc..d06ca664a6 100644 --- a/crates/buzz-relay/src/api/invites.rs +++ b/crates/buzz-relay/src/api/invites.rs @@ -251,15 +251,42 @@ async fn authenticate( let bridge::VerifiedBridgeAuth { pubkey, event_id_bytes, - .. + signed_created_at, } = bridge::verify_nip98_exempt_invite_claim(headers, "POST", &url, Some(body))?; bridge::check_nip98_replay(state, &tenant, event_id_bytes).await?; + materialize_invite_owner(state, &tenant, &pubkey, headers, signed_created_at).await?; Ok((tenant, pubkey)) } +async fn materialize_invite_owner( + state: &AppState, + tenant: &buzz_core::TenantContext, + pubkey: &nostr::PublicKey, + headers: &HeaderMap, + signed_created_at: Option, +) -> Result<(), (StatusCode, Json)> { + // The claim remains membership-exempt. Only cryptographically verified + // owner provenance is persisted, so the transactional restriction lookup + // also covers an agent onboarding for the first time. + if let Some(owner) = super::relay_members::extract_nip_oa_owner( + pubkey.as_bytes(), + super::relay_members::extract_auth_tag_header(headers), + signed_created_at, + ) { + if !super::relay_members::materialize_nip_oa_owner(state, tenant, pubkey, &owner).await { + return Err(api_error( + StatusCode::SERVICE_UNAVAILABLE, + "restriction lookup unavailable", + )); + } + } + Ok(()) +} + fn map_mint_error(error: buzz_db::DbError) -> (StatusCode, Json) { 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 +369,12 @@ async fn mint_invite_checked( return e.into_response(); } + if let Err(error) = + materialize_invite_owner(&state, &tenant, &pubkey, &headers, signed_created_at).await + { + return error.into_response(); + } + match mint_invite_inner(&state, body, tenant, pubkey).await { Ok(json) => json.into_response(), Err(e) => e.into_response(), @@ -516,6 +549,12 @@ pub async fn claim_invite( buzz_db::relay_invite::ClaimOutcome::Exhausted => { Err(api_error(StatusCode::FORBIDDEN, "invite_exhausted")) } + buzz_db::relay_invite::ClaimOutcome::Revoked => { + 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 +562,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 @@ -556,7 +597,12 @@ pub async fn claim_invite( .map(|policy| policy.version.as_str()), ) .await - .map_err(|e| internal_error(&format!("invite claim insert: {e}")))?; + .map_err(|e| match e { + buzz_db::DbError::InviteRestricted => { + api_error(StatusCode::FORBIDDEN, "invite_restricted") + } + e => internal_error(&format!("invite claim insert: {e}")), + })?; if was_inserted { tracing::info!( @@ -608,6 +654,7 @@ fn claim_key_rate_limited( #[cfg(test)] mod postgres_tests { + mod restriction_tests; use std::sync::Arc; use std::time::Duration; 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 0000000000..0f1a69d9b2 --- /dev/null +++ b/crates/buzz-relay/src/api/invites/postgres_tests/restriction_tests.rs @@ -0,0 +1,254 @@ +use super::*; + +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) +} + +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()); + 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 (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 mut state = Arc::try_unwrap(state) + .ok() + .expect("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; + assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR); + assert!(!state + .db + .is_relay_member(community, &claimant.public_key().to_hex()) + .await + .unwrap()); +} diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 129d2792ed..49be08ede0 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, @@ -410,6 +413,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 @@ -1439,6 +1464,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, @@ -2891,3 +2917,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/invite_token.rs b/crates/buzz-relay/src/invite_token.rs index e271f40f0b..9886b78153 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/migrations/0061_invite_revocation.sql b/migrations/0061_invite_revocation.sql new file mode 100644 index 0000000000..6555200622 --- /dev/null +++ b/migrations/0061_invite_revocation.sql @@ -0,0 +1,121 @@ +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)); +$$; + +CREATE FUNCTION lock_invite_restrictions() RETURNS TRIGGER +LANGUAGE plpgsql AS $$ +BEGIN + PERFORM invite_admission_lock(NEW.community_id); + RETURN NEW; +END +$$; + +-- Revocation is durable: neither unban nor expiry restores an issued bearer. +CREATE FUNCTION revoke_banned_issuer_invites() RETURNS TRIGGER +LANGUAGE plpgsql AS $$ +BEGIN + IF NEW.banned AND (NEW.ban_expires_at IS NULL OR NEW.ban_expires_at > clock_timestamp()) THEN + UPDATE relay_invites SET revoked_at = clock_timestamp() + WHERE community_id = NEW.community_id AND revoked_at IS NULL + AND (created_by = encode(NEW.pubkey, 'hex') OR created_by IN ( + SELECT encode(pubkey, 'hex') FROM users + WHERE community_id = NEW.community_id AND agent_owner_pubkey = NEW.pubkey + )); + END IF; + RETURN NEW; +END +$$; + +-- Learning a verified owner must not revive invites issued by its restricted agent. +CREATE FUNCTION revoke_restricted_agent_invites() RETURNS TRIGGER +LANGUAGE plpgsql AS $$ +BEGIN + IF NEW.agent_owner_pubkey IS NOT NULL AND EXISTS ( + SELECT 1 FROM community_bans WHERE community_id = NEW.community_id + AND pubkey = NEW.agent_owner_pubkey AND banned + AND (ban_expires_at IS NULL OR ban_expires_at > clock_timestamp()) + ) THEN + UPDATE relay_invites SET revoked_at = clock_timestamp() + WHERE community_id = NEW.community_id AND created_by = encode(NEW.pubkey, 'hex') + AND revoked_at IS NULL; + END IF; + RETURN NEW; +END +$$; + +CREATE TRIGGER invite_restriction_lock BEFORE INSERT OR UPDATE ON community_bans +FOR EACH ROW EXECUTE FUNCTION lock_invite_restrictions(); +CREATE TRIGGER invite_issuer_revocation AFTER INSERT OR UPDATE ON community_bans +FOR EACH ROW EXECUTE FUNCTION revoke_banned_issuer_invites(); + +CREATE TRIGGER invite_owner_lock BEFORE INSERT OR UPDATE OF agent_owner_pubkey ON users +FOR EACH ROW EXECUTE FUNCTION lock_invite_restrictions(); +CREATE TRIGGER invite_owner_revocation AFTER INSERT OR UPDATE OF agent_owner_pubkey ON users +FOR EACH ROW EXECUTE FUNCTION revoke_restricted_agent_invites(); + +-- 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 0000000000..abe8118b0b --- /dev/null +++ b/schema/functions/backfill_invite_revocations.sql @@ -0,0 +1,60 @@ +-- 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 0000000000..c5d7098d3f --- /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/functions/lock_invite_restrictions.sql b/schema/functions/lock_invite_restrictions.sql new file mode 100644 index 0000000000..c8aec1f2c6 --- /dev/null +++ b/schema/functions/lock_invite_restrictions.sql @@ -0,0 +1,7 @@ +CREATE FUNCTION lock_invite_restrictions() RETURNS TRIGGER +LANGUAGE plpgsql AS $$ +BEGIN + PERFORM invite_admission_lock(NEW.community_id); + RETURN NEW; +END +$$; diff --git a/schema/functions/revoke_banned_issuer_invites.sql b/schema/functions/revoke_banned_issuer_invites.sql new file mode 100644 index 0000000000..beb40f1ca4 --- /dev/null +++ b/schema/functions/revoke_banned_issuer_invites.sql @@ -0,0 +1,15 @@ +-- Revocation is durable: neither unban nor expiry restores an issued bearer. +CREATE FUNCTION revoke_banned_issuer_invites() RETURNS TRIGGER +LANGUAGE plpgsql AS $$ +BEGIN + IF NEW.banned AND (NEW.ban_expires_at IS NULL OR NEW.ban_expires_at > clock_timestamp()) THEN + UPDATE relay_invites SET revoked_at = clock_timestamp() + WHERE community_id = NEW.community_id AND revoked_at IS NULL + AND (created_by = encode(NEW.pubkey, 'hex') OR created_by IN ( + SELECT encode(pubkey, 'hex') FROM users + WHERE community_id = NEW.community_id AND agent_owner_pubkey = NEW.pubkey + )); + END IF; + RETURN NEW; +END +$$; diff --git a/schema/functions/revoke_restricted_agent_invites.sql b/schema/functions/revoke_restricted_agent_invites.sql new file mode 100644 index 0000000000..3a6d2359b7 --- /dev/null +++ b/schema/functions/revoke_restricted_agent_invites.sql @@ -0,0 +1,16 @@ +-- Learning a verified owner must not revive invites issued by its restricted agent. +CREATE FUNCTION revoke_restricted_agent_invites() RETURNS TRIGGER +LANGUAGE plpgsql AS $$ +BEGIN + IF NEW.agent_owner_pubkey IS NOT NULL AND EXISTS ( + SELECT 1 FROM community_bans WHERE community_id = NEW.community_id + AND pubkey = NEW.agent_owner_pubkey AND banned + AND (ban_expires_at IS NULL OR ban_expires_at > clock_timestamp()) + ) THEN + UPDATE relay_invites SET revoked_at = clock_timestamp() + WHERE community_id = NEW.community_id AND created_by = encode(NEW.pubkey, 'hex') + AND revoked_at IS NULL; + END IF; + RETURN NEW; +END +$$; diff --git a/schema/schema.sql b/schema/schema.sql index d252cbdb7e..d8ed4f9019 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -78,6 +78,12 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; \i functions/attach_community_write_fence.sql \i functions/profile_name_fields.sql +\i functions/invite_admission_lock.sql +\i functions/lock_invite_restrictions.sql +\i functions/revoke_banned_issuer_invites.sql +\i functions/revoke_restricted_agent_invites.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 4e0f2eab0a..65a7a9d63d 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/schema/triggers/community_bans.sql b/schema/triggers/community_bans.sql index 063e3ab456..27c66ee1a2 100644 --- a/schema/triggers/community_bans.sql +++ b/schema/triggers/community_bans.sql @@ -1 +1,6 @@ SELECT attach_community_write_fence('community_bans'); + +CREATE TRIGGER invite_restriction_lock BEFORE INSERT OR UPDATE ON community_bans +FOR EACH ROW EXECUTE FUNCTION lock_invite_restrictions(); +CREATE TRIGGER invite_issuer_revocation AFTER INSERT OR UPDATE ON community_bans +FOR EACH ROW EXECUTE FUNCTION revoke_banned_issuer_invites(); diff --git a/schema/triggers/users.sql b/schema/triggers/users.sql index 5f65f3c763..60bd2c5c76 100644 --- a/schema/triggers/users.sql +++ b/schema/triggers/users.sql @@ -1 +1,6 @@ SELECT attach_community_write_fence('users'); + +CREATE TRIGGER invite_owner_lock BEFORE INSERT OR UPDATE OF agent_owner_pubkey ON users +FOR EACH ROW EXECUTE FUNCTION lock_invite_restrictions(); +CREATE TRIGGER invite_owner_revocation AFTER INSERT OR UPDATE OF agent_owner_pubkey ON users +FOR EACH ROW EXECUTE FUNCTION revoke_restricted_agent_invites(); diff --git a/scripts/reconcile-schema-after-pgschema.sql b/scripts/reconcile-schema-after-pgschema.sql index 06d6258643..c673c1a779 100644 --- a/scripts/reconcile-schema-after-pgschema.sql +++ b/scripts/reconcile-schema-after-pgschema.sql @@ -232,3 +232,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')) <> 4 THEN + RAISE EXCEPTION 'invitation revocation triggers missing from live catalog'; + END IF; +END +$$;