fix(relay): revoke invitations durably and fence restricted claims

Keep v1 natural drain with an optional UTC cutoff. Serialize claim and mint policy reads with restriction and owner writes. Share historical revocation repair across migration and desired-state upgrades while retaining deletion fences.

Co-authored-by: Codex <codex@openai.com>
Signed-off-by: Brad Seiler <seiler@squareup.com>
Signed-off-by: Local Builder <1c476cb1127ea270cb900bdc3891060140bdd1173f91498030ee9b4db01c2f20@buzz.block.builderlab.xyz>
This commit is contained in:
Brad Seiler
2026-10-09 19:36:31 -04:00
committed by seiler
parent 6286a54321
commit d2c2a74943
23 changed files with 1253 additions and 23 deletions

View File

@@ -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

View File

@@ -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

View File

@@ -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),

View File

@@ -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);

View File

@@ -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<bool> {
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<uuid::Uuid>,
@@ -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<DateTime<Utc>> = invite.try_get("revoked_at")?;
let issuer: String = invite.try_get("created_by")?;
if revoked_at.is_some() || principal_restricted(tx.conn(), community, &issuer).await? {
return Ok(ClaimOutcome::Revoked);
}
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;

View File

@@ -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<Result<ClaimOutcome>> {
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);
}

View File

@@ -0,0 +1,120 @@
use super::*;
use uuid::Uuid;
async fn legacy_fixture() -> (PgPool, Vec<Uuid>) {
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;
}

View File

@@ -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') \

View File

@@ -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<u64>,
) -> Result<(), (StatusCode, Json<Value>)> {
// The claim remains membership-exempt. Only cryptographically verified
// owner provenance is persisted, so the transactional restriction lookup
// also covers an agent onboarding for the first time.
if let Some(owner) = super::relay_members::extract_nip_oa_owner(
pubkey.as_bytes(),
super::relay_members::extract_auth_tag_header(headers),
signed_created_at,
) {
if !super::relay_members::materialize_nip_oa_owner(state, tenant, pubkey, &owner).await {
return Err(api_error(
StatusCode::SERVICE_UNAVAILABLE,
"restriction lookup unavailable",
));
}
}
Ok(())
}
fn map_mint_error(error: buzz_db::DbError) -> (StatusCode, Json<Value>) {
match error {
buzz_db::DbError::InviteRestricted => api_error(StatusCode::FORBIDDEN, "invite_restricted"),
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;

View File

@@ -0,0 +1,254 @@
use super::*;
async fn fixture() -> (Arc<AppState>, String, buzz_core::CommunityId, Keys) {
let host = format!("restricted-invite-{}.example", Uuid::new_v4().simple());
let state = invite_test_state(&host)
.await
.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<AppState>,
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());
}

View File

@@ -379,6 +379,9 @@ pub struct Config {
/// documents or age attestation are configured.
pub join_policy: Option<JoinPolicyConfig>,
/// Absolute UTC Unix-second cutoff for legacy v1 invites; unset drains naturally.
pub invite_v1_invalid_after: Option<u64>,
/// Deployment-admin API and SPA configuration. Absent means the surface is disabled.
pub admin: Option<AdminConfig>,
@@ -410,6 +413,28 @@ fn parse_bind_addr(raw: &str) -> Result<SocketAddr, ConfigError> {
.map_err(|e| ConfigError::InvalidBindAddr(e.to_string()))
}
fn parse_invite_v1_cutoff(raw: Option<&str>) -> Result<Option<u64>, 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<Option<u64>, 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<u64, ConfigError> {
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}");
}
}
}

View File

@@ -157,6 +157,27 @@ pub fn verify_invite(
key: &[u8; 32],
community: CommunityId,
code: &str,
) -> Result<InvitePayload, InviteError> {
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<u64>,
) -> Result<InvitePayload, InviteError> {
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<u64>,
) -> Result<InvitePayload, InviteError> {
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:?}"
);
}
}
}

View File

@@ -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();

View File

@@ -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
$$;

View File

@@ -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));
$$;

View File

@@ -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
$$;

View File

@@ -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
$$;

View File

@@ -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
$$;

View File

@@ -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

View File

@@ -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),

View File

@@ -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();

View File

@@ -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();

View File

@@ -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
$$;