From b4b10690e43d138b9f8552ba4b3f6f5566ff9edb Mon Sep 17 00:00:00 2001 From: seiler Date: Mon, 5 Oct 2026 21:52:10 +0000 Subject: [PATCH] Add durable live member authorization checks Signed-off-by: seiler Co-authored-by: Codex (cherry picked from commit 92f13229ed3038d63f2a2d9c889bfd4c212e38e2) Signed-off-by: Brad Seiler --- ARCHITECTURE.md | 77 + crates/buzz-db/src/lib.rs | 3 +- .../buzz-db/src/store/live_authorization.rs | 406 ++++ crates/buzz-db/src/store/mod.rs | 2 + crates/buzz-relay/src/audio/handler.rs | 57 +- crates/buzz-relay/src/handlers/auth.rs | 180 +- crates/buzz-relay/src/main.rs | 21 + crates/buzz-relay/src/state.rs | 1930 ++++++++++++++++- 8 files changed, 2607 insertions(+), 69 deletions(-) create mode 100644 crates/buzz-db/src/store/live_authorization.rs diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 23961d940c..3d5ce0d968 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -259,6 +259,83 @@ A `CancellationToken` coordinates shutdown across all three loops. Slow clients: `ConnectionState::send()` uses `try_send` — if the send buffer is full, a grace counter increments. After `SLOW_CLIENT_GRACE_LIMIT` (3) consecutive full-buffer events, the connection is cancelled. A successful send resets the counter. +### Live Community Authorization Revalidation + +Authenticated root and huddle/audio sockets retain their server-resolved +community, principal, admission-captured owner, admission route, whether that +owner was a relay member at admission, and whether closed-relay membership was +granted through that owner. A relay-local worker checks +these records against batched writer-authoritative snapshots every 5 seconds. It +keeps the immediate local disconnect and Redis fan-out paths; this scan is the +durable backstop when a pod misses fan-out. + +The maximum authorization-staleness budget is 30 seconds. A database failure +may preserve the last successful decision for 10 seconds, measured from the +start of that session's last successful authoritative check; repeated failures +never extend the grace. The worker handles up to 500 distinct +`(community, principal, captured owner)` +targets per query, runs at most 4 queries at a time, gives each query 1 second, +and starts one shared 5-second lookup deadline before it snapshots and groups +the live-session registry. No database batch starts after that deadline; +unresolved targets are treated as lookup failures. Closing decisions are then applied in one +bounded sweep over the connection snapshot. At the default 10,000-connection +limit, the in-memory regression harness processed a full 10,000-session scan in +70 ms in the latest measured workstation run. A disposable PostgreSQL run +fetched all 10,000 authorization targets in 129 ms through the same 500-row, +four-query batching shape. Database lookups are bounded by the shared 5-second +deadline. The conservative staleness bound is 10 seconds of database grace, one +5-second cadence, two 5-second scan windows, and one 1-second query deadline: +26 seconds, below the 30-second budget. The relay's shared +`BUZZ_MAX_CONNECTIONS` semaphore caps root and audio sockets together at +10,000 by default; that is at most 20 batches and five query waves. The worker +creates no per-socket timers or per-connection revalidation queries. A scan +that exceeds its deadline treats unreturned targets as lookup failures and +closes their sessions when their existing grace has expired. If a result batch +is incomplete or malformed, every target in +that batch is treated as a lookup failure; no partial rows refresh +authorization. + +Membership is required only when `require_relay_membership` is enabled. An +open-relay nonmember remains allowed unless directly or through its current +recorded owner it has an active community ban. An owner-derived member remains +allowed only while the relay, owner-delegation setting, and current owner +membership still permit that path. Removal of the admission-captured owner +revokes its session even if the agent's stored owner changed; captured-owner +membership never grants access to a directly admitted principal. The first +stored owner link also invalidates sessions admitted without one, matching the +existing reconnect path when NIP-OA materializes ownership; the durable check +detects that transition if its disconnect fan-out is missed. A directly +admitted member with a stable owner link still relies on its own membership, +and a change between already-recorded owners alone does not create a new +membership requirement for that principal. An active ban on the captured owner +also revokes that session. On open relays, an owner who was already a nonmember +at admission does not become a membership requirement. Moderation timeouts are +not read by this worker because they restrict writes, not an established +socket. All membership, ownership, and ban predicates include the bound +community. Admission reads the captured owner's membership from the writer +when the delegation check did not already establish it; that bounded point +lookup records the baseline needed to distinguish a later membership removal +from pre-existing open-relay nonmembership. + +Each batch snapshot is the revalidation ordering point. A removal or ban +committed before that snapshot closes the session; a re-add or unban committed +before it keeps the session authorized. A later concurrent change is observed +on the next scan, and a close based on an earlier snapshot can be recovered by +reconnecting against current state. Admission binds the authenticated identity +before its final writer read and only publishes an admitted live record after +that check, so admission racing a removal is either cancelled by the immediate +path or rejected by the final read / next durable scan. + +If the writer remains unavailable past the grace, the relay fails closed and +cancels the socket. NIP-FI root and audio sockets receive their existing +route-specific `authorization_denied` or `authorization_unavailable` terminal +response; other sockets receive the normal access-revoked or unavailable close +reason. This trades connection availability during a prolonged database +outage for a bounded authorization window. Reconnecting after the database +recovers performs the ordinary admission checks. The worker shares graceful +shutdown cancellation and has no migration or rollout flag; all relay pods +must run this code for the bound to hold deployment-wide. + ### Step 5: Cleanup On disconnect (any cause): diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9205bdc899..522f445d1d 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -66,7 +66,7 @@ pub(crate) use runtime::{ }; pub use store::{ admin_moderation, allowlist, api_token, archived_identities, artifact, channel, - channel_members, community, deletion, dm, entity_deletion, event, feed, git_repo, moderation, + channel_members, community, deletion, dm, entity_deletion, event, feed, git_repo, live_authorization, moderation, nip_fi_binding, operator_listener, partition, personal_read, product_feedback, push, reaction, read_state, relay_admin_actions, relay_invite, relay_members, relay_operators, reminder, replaceable, storage_accounting, thread, thread_window, usage, user, user_deletion, @@ -83,6 +83,7 @@ pub use community::{ pub use error::{DbError, Result}; pub use event::{ChannelHeadPrecondition, ChannelHeadWriteStatus}; pub use event::{EventQuery, DEFAULT_MAX_PAGE_LIMIT}; +pub use live_authorization::{LiveAuthorizationState, LiveAuthorizationTarget}; pub use nip_fi_binding::{NipFiBindingRecord, NipFiUnbind, RowCapExceeded}; pub use reaction::ReactionEventInsertOutcome; pub use reminder::DueReminder; diff --git a/crates/buzz-db/src/store/live_authorization.rs b/crates/buzz-db/src/store/live_authorization.rs new file mode 100644 index 0000000000..2dda37ce5c --- /dev/null +++ b/crates/buzz-db/src/store/live_authorization.rs @@ -0,0 +1,406 @@ +//! Bounded, tenant-scoped authorization snapshots for already-open sessions. + +use buzz_core::CommunityId; +use buzz_datastore_tracing::datastore_span; +use sqlx::Row as _; +use uuid::Uuid; + +use crate::{observability, Db, DbError, Result}; + +/// One admitted principal and captured owner whose live authorization should be revalidated. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct LiveAuthorizationTarget { + /// The community resolved for the connection. + pub community_id: CommunityId, + /// The NIP-42-authenticated principal. + pub pubkey: [u8; 32], + /// The owner captured when this socket was admitted, if any. + pub session_owner_pubkey: Option<[u8; 32]>, +} + +/// Authoritative access state for one live principal/owner pair in one community. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct LiveAuthorizationState { + /// The community used to scope every part of this lookup. + pub community_id: CommunityId, + /// The NIP-42-authenticated principal. + pub pubkey: [u8; 32], + /// The owner captured when this socket was admitted, if any. + pub session_owner_pubkey: Option<[u8; 32]>, + /// Whether that captured owner currently belongs to the community roster. + pub session_owner_is_relay_member: bool, + /// The principal's currently recorded agent owner, if any. + pub agent_owner_pubkey: Option<[u8; 32]>, + /// Whether the principal is currently in the community relay roster. + pub relay_member: bool, + /// Whether its current owner is currently in the community relay roster. + pub owner_is_relay_member: bool, + /// Whether the principal, captured owner, or current owner has an active community ban. + pub banned: bool, +} + +/// Return one writer-authoritative access snapshot for each requested target. +/// +/// Membership, current agent ownership, and active bans are read by one SQL +/// statement so a relay can revalidate many live sessions without a query per +/// connection. Every input is an explicit `(community_id, pubkey, captured +/// owner)` tuple; the lookup never expands an owner or ban across tenant +/// boundaries. Timeouts are intentionally omitted because they restrict writes +/// rather than live access. +async fn live_authorization_state_batch( + pool: &sqlx::PgPool, + targets: &[LiveAuthorizationTarget], +) -> Result> { + if targets.is_empty() { + return Ok(Vec::new()); + } + + let community_ids = targets + .iter() + .map(|target| *target.community_id.as_uuid()) + .collect::>(); + let pubkeys = targets + .iter() + .map(|target| target.pubkey.to_vec()) + .collect::>(); + let session_owner_pubkeys = targets + .iter() + .map(|target| target.session_owner_pubkey.map(|owner| owner.to_vec())) + .collect::>(); + let mut connection = + observability::acquire_writer(pool, observability::WriterOperation::Authorization).await?; + let rows = sqlx::query( + r#" + WITH requested AS ( + SELECT DISTINCT community_id, pubkey, session_owner_pubkey + FROM unnest($1::uuid[], $2::bytea[], $3::bytea[]) + AS target(community_id, pubkey, session_owner_pubkey) + ) + SELECT requested.community_id, + requested.pubkey, + requested.session_owner_pubkey, + EXISTS ( + SELECT 1 + FROM relay_members + WHERE relay_members.community_id = requested.community_id + AND relay_members.pubkey = encode(requested.session_owner_pubkey, 'hex') + ) AS session_owner_is_relay_member, + users.agent_owner_pubkey, + EXISTS ( + SELECT 1 + FROM relay_members + WHERE relay_members.community_id = requested.community_id + AND relay_members.pubkey = encode(requested.pubkey, 'hex') + ) AS relay_member, + EXISTS ( + SELECT 1 + FROM relay_members + WHERE relay_members.community_id = requested.community_id + AND relay_members.pubkey = encode(users.agent_owner_pubkey, 'hex') + ) AS owner_is_relay_member, + EXISTS ( + SELECT 1 + FROM community_bans + WHERE community_bans.community_id = requested.community_id + AND community_bans.pubkey IN + (requested.pubkey, requested.session_owner_pubkey, users.agent_owner_pubkey) + AND community_bans.banned + AND (community_bans.ban_expires_at IS NULL + OR community_bans.ban_expires_at > now()) + ) AS banned + FROM requested + LEFT JOIN users + ON users.community_id = requested.community_id + AND users.pubkey = requested.pubkey + "#, + ) + .bind(community_ids) + .bind(pubkeys) + .bind(session_owner_pubkeys) + .fetch_all(&mut *connection) + .await?; + + rows.into_iter() + .map(|row| { + let community_id: Uuid = row.try_get("community_id")?; + let pubkey = decode_pubkey(row.try_get("pubkey")?, "requested pubkey")?; + let session_owner_pubkey = row + .try_get::>, _>("session_owner_pubkey")? + .map(|owner| decode_pubkey(owner, "session owner pubkey")) + .transpose()?; + let agent_owner_pubkey = row + .try_get::>, _>("agent_owner_pubkey")? + .map(|owner| decode_pubkey(owner, "agent owner pubkey")) + .transpose()?; + + Ok(LiveAuthorizationState { + community_id: CommunityId::from_uuid(community_id), + pubkey, + session_owner_pubkey, + session_owner_is_relay_member: row.try_get("session_owner_is_relay_member")?, + agent_owner_pubkey, + relay_member: row.try_get("relay_member")?, + owner_is_relay_member: row.try_get("owner_is_relay_member")?, + banned: row.try_get("banned")?, + }) + }) + .collect() +} + +fn decode_pubkey(bytes: Vec, field: &'static str) -> Result<[u8; 32]> { + bytes + .try_into() + .map_err(|_| DbError::InvalidData(format!("{field} must contain exactly 32 bytes"))) +} + +impl Db { + /// Read current membership, agent ownership, and ban state for a bounded + /// batch of live connections from the authoritative writer. + /// + /// The caller is responsible for bounding the batch size and total work. + /// Each result is scoped to the exact community/principal/captured-owner tuple supplied. + #[datastore_span(name = "live_authorization_state_batch", system = "postgresql")] + pub async fn live_authorization_state_batch( + &self, + targets: &[LiveAuthorizationTarget], + ) -> Result> { + live_authorization_state_batch(&self.pool, targets).await + } +} + +#[cfg(test)] +mod postgres_tests { + use super::*; + use chrono::{Duration, Utc}; + use nostr::Keys; + use sqlx::PgPool; + + async fn setup_db() -> Db { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect to disposable test database"); + Db::from_pool(pool) + } + + async fn make_community(db: &Db) -> CommunityId { + let id = Uuid::new_v4(); + let host = format!("live-auth-test-{}.example", id.simple()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(id) + .bind(host) + .execute(&db.pool) + .await + .expect("insert test community"); + CommunityId::from_uuid(id) + } + + fn pubkey() -> [u8; 32] { + Keys::generate().public_key().to_bytes() + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn batch_is_tenant_scoped_reads_owner_bans_ignores_timeouts_and_scales_to_10k() { + let db = setup_db().await; + let community_a = make_community(&db).await; + let community_b = make_community(&db).await; + let owner = pubkey(); + let current_owner = pubkey(); + let agent = pubkey(); + let late_link_agent = pubkey(); + let direct = pubkey(); + let timed_out = pubkey(); + let actor = pubkey(); + + for key in [ + owner, + current_owner, + agent, + late_link_agent, + direct, + timed_out, + actor, + ] { + db.ensure_user_for_authorization(community_a, &key) + .await + .expect("create community A user"); + } + db.ensure_user_for_authorization(community_b, &direct) + .await + .expect("create community B user"); + db.ensure_user_for_authorization(community_b, &late_link_agent) + .await + .expect("create community B agent"); + db.ensure_user_for_authorization(community_b, ¤t_owner) + .await + .expect("create community B owner"); + db.set_agent_owner_for_authorization(community_a, &agent, ¤t_owner) + .await + .expect("record agent owner"); + db.add_relay_member(community_a, &hex::encode(owner), "member", None) + .await + .expect("add owner to community A"); + db.add_relay_member(community_a, &hex::encode(current_owner), "member", None) + .await + .expect("add current owner to community A"); + db.add_relay_member(community_b, &hex::encode(direct), "member", None) + .await + .expect("add principal to community B"); + db.add_relay_member(community_b, &hex::encode(current_owner), "member", None) + .await + .expect("add current owner to community B"); + // Capture the session as ownerless, then commit its first owner link. + // This is the supported None -> Some ownership transition. + db.set_agent_owner_for_authorization(community_a, &late_link_agent, ¤t_owner) + .await + .expect("record late owner link in community A"); + db.set_agent_owner_for_authorization(community_b, &late_link_agent, ¤t_owner) + .await + .expect("record same owner link in community B"); + db.ban_community_member(community_a, &owner, &actor, None, None) + .await + .expect("ban agent owner in community A"); + db.ban_community_member(community_a, ¤t_owner, &actor, None, None) + .await + .expect("ban current agent owner in community A"); + db.timeout_community_member( + community_a, + &timed_out, + &actor, + Utc::now() + Duration::minutes(5), + None, + ) + .await + .expect("time out principal in community A"); + + let states = db + .live_authorization_state_batch(&[ + LiveAuthorizationTarget { + community_id: community_a, + pubkey: agent, + session_owner_pubkey: Some(owner), + }, + LiveAuthorizationTarget { + community_id: community_a, + pubkey: late_link_agent, + session_owner_pubkey: None, + }, + LiveAuthorizationTarget { + community_id: community_b, + pubkey: late_link_agent, + session_owner_pubkey: None, + }, + LiveAuthorizationTarget { + community_id: community_b, + pubkey: agent, + session_owner_pubkey: Some(owner), + }, + LiveAuthorizationTarget { + community_id: community_a, + pubkey: direct, + session_owner_pubkey: None, + }, + LiveAuthorizationTarget { + community_id: community_b, + pubkey: direct, + session_owner_pubkey: None, + }, + LiveAuthorizationTarget { + community_id: community_a, + pubkey: timed_out, + session_owner_pubkey: None, + }, + ]) + .await + .expect("batch live authorization state"); + let state = |community_id, pubkey| { + states + .iter() + .find(|state| state.community_id == community_id && state.pubkey == pubkey) + .expect("state for requested tenant/principal") + }; + + let agent_state = state(community_a, agent); + assert_eq!(agent_state.agent_owner_pubkey, Some(current_owner)); + assert_eq!(agent_state.session_owner_pubkey, Some(owner)); + assert!(agent_state.session_owner_is_relay_member); + assert!(!agent_state.relay_member); + assert!( + agent_state.owner_is_relay_member, + "active bans are reported independently of roster membership" + ); + assert!( + agent_state.banned, + "captured owner ban applies after owner change" + ); + let late_link_state = state(community_a, late_link_agent); + assert_eq!( + late_link_state.agent_owner_pubkey, + Some(current_owner), + "the lookup reads a first owner link added after admission" + ); + assert!( + late_link_state.banned, + "a ban on a newly recorded current owner is returned for an ownerless session" + ); + assert!( + !state(community_b, late_link_agent).banned, + "a ban on the current owner does not cross communities" + ); + assert!( + !state(community_b, agent).banned, + "a ban on the same captured owner does not cross communities" + ); + assert!( + !state(community_b, agent).session_owner_is_relay_member, + "captured-owner membership is community scoped" + ); + + assert!(!state(community_a, direct).relay_member); + assert!(state(community_b, direct).relay_member); + let timeout_state = state(community_a, timed_out); + assert!(!timeout_state.banned, "timeouts do not revoke live access"); + + // Exercise the relay's actual 500-principal batches and four-query + // concurrency at the default 10,000-socket limit. This is a disposable + // local fixture; no user or moderation rows are needed for absent keys. + const SUPPORTED_PRINCIPALS: usize = 10_000; + const BATCH_SIZE: usize = 500; + const QUERY_CONCURRENCY: usize = 4; + let targets = (0..SUPPORTED_PRINCIPALS) + .map(|index| { + let mut pubkey = [0; 32]; + pubkey[..8].copy_from_slice(&(index as u64).to_be_bytes()); + LiveAuthorizationTarget { + community_id: community_a, + pubkey, + session_owner_pubkey: None, + } + }) + .collect::>(); + let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(QUERY_CONCURRENCY)); + let mut queries = tokio::task::JoinSet::new(); + let started_at = std::time::Instant::now(); + for batch in targets.chunks(BATCH_SIZE).map(|batch| batch.to_vec()) { + let db = db.clone(); + let semaphore = std::sync::Arc::clone(&semaphore); + queries.spawn(async move { + let _permit = semaphore.acquire_owned().await.expect("query permit"); + db.live_authorization_state_batch(&batch) + .await + .map(|states| states.len()) + }); + } + let mut rows = 0; + while let Some(result) = queries.join_next().await { + rows += result.expect("batch task").expect("batch query"); + } + let elapsed = started_at.elapsed(); + assert_eq!(rows, SUPPORTED_PRINCIPALS); + eprintln!( + "10k-principal PostgreSQL scan: {rows} rows, batches of {BATCH_SIZE}, \ + concurrency {QUERY_CONCURRENCY}, {} ms", + elapsed.as_millis(), + ); + } +} diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index d36fa4d089..7cfe8a68d5 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -38,6 +38,8 @@ mod event_follow_up_postgres_tests; pub mod feed; /// Git repository name registry (NIP-34 kind:30617). pub mod git_repo; +/// Batched authoritative access-state reads for live connections. +pub mod live_authorization; /// Community moderation: reports, bans/timeouts, audit actions. pub mod moderation; /// Deployment-global operator-listener mention registrations and delivery queues. diff --git a/crates/buzz-relay/src/audio/handler.rs b/crates/buzz-relay/src/audio/handler.rs index 21a5412069..f20b5d7efe 100644 --- a/crates/buzz-relay/src/audio/handler.rs +++ b/crates/buzz-relay/src/audio/handler.rs @@ -997,7 +997,7 @@ async fn run_active_audio_connection( } // Bind, then take the final ban/membership decision the root socket - // applies (see `final_admission_denial`), before any huddle lease. A ban + // applies (see `final_admission_check`), before any huddle lease. A ban // or removal whose disconnect ran before the bind is seen by these fresh // reads; one that runs after it cancels this socket (`check_cancel!`). // Same order as root AUTH: a proven owner before the pubkey; otherwise @@ -1007,27 +1007,40 @@ async fn run_active_audio_connection( control.bind_owner(owner.to_bytes()); } control.bind_pubkey(pubkey.to_bytes()); - let owner = + let admitted_owner = crate::handlers::auth::admitted_owner(&state, tenant.community(), pubkey, nip_oa_owner) .await; - if let Ok(Some(owner)) = owner { - control.bind_owner(owner); - } - let denial = match owner { - Err(denial) => Some(denial), - Ok(_) => { - crate::handlers::auth::final_admission_denial( - &state, - tenant.community(), - pubkey, - auth_tag_json.as_deref(), - Some(signed_auth_created_at), - token_owner, + let admitted_owner = match admitted_owner { + Ok(owner) => owner, + Err(denial) => { + shadow_attempt.refused(); + exit_authorization_refusal( + &mut ws_send, + &control, + &mut terminal_ctrl_rx, + nip_fi_assertion.is_some(), + denial.class, + serde_json::json!({"type": "error", "message": denial.reason}), ) - .await + .await; + return; } }; - if let Some(denial) = denial { + if let Some(owner) = admitted_owner { + control.bind_owner(owner); + } + let final_check_started_at = tokio::time::Instant::now(); + let final_check = crate::handlers::auth::final_admission_check( + &state, + tenant.community(), + pubkey, + auth_tag_json.as_deref(), + Some(signed_auth_created_at), + token_owner, + admitted_owner, + ) + .await; + if let Some(denial) = final_check.denial { shadow_attempt.refused(); let (class, message) = (denial.class, denial.reason); warn!(channel_id = %channel_id, pubkey = %pubkey_hex, reason = message, "audio: denied at final admission check"); @@ -1043,6 +1056,14 @@ async fn run_active_audio_connection( return; } check_cancel!(cancel, terminal_ctrl_rx, disconnect_reason, ws_send); + control.mark_live_authorized( + pubkey.to_bytes(), + admitted_owner, + final_check.owner_was_relay_member_at_admission, + final_check.membership_via_owner, + crate::nip_fi_session::NipFiWsRoute::Audio, + final_check_started_at, + ); // Huddle cross-pod routing (mesh) OR single-pod guardrail. // @@ -4188,7 +4209,7 @@ mod tests { // Fix 5: use Config::for_test() which holds NIP_FI_ENV_LOCK internally. [FI-TRACE-ENV-RACE] let mut config = crate::config::Config::for_test(); config.require_relay_membership = require_relay_membership; - config.database_url = "postgres://buzz:buzz_dev@127.0.0.1:1/buzz".to_string(); // sadscan:disable np.postgres.1 -- unreachable test-only URL + config.database_url = "postgres://127.0.0.1:1/buzz".to_string(); config.redis_url = "redis://127.0.0.1:1".to_string(); let mut pool_options = sqlx::postgres::PgPoolOptions::new(); if let Some(timeout) = acquire_timeout { diff --git a/crates/buzz-relay/src/handlers/auth.rs b/crates/buzz-relay/src/handlers/auth.rs index 9ec99a5633..adf9dd7fd6 100644 --- a/crates/buzz-relay/src/handlers/auth.rs +++ b/crates/buzz-relay/src/handlers/auth.rs @@ -56,6 +56,38 @@ fn classify_relay_membership( } } +fn final_admission_from_membership( + membership: PolicyCheck>, +) -> FinalAdmissionCheck { + match membership { + PolicyCheck::Allowed(owner) => FinalAdmissionCheck { + denial: None, + membership_via_owner: owner.is_some(), + owner_was_relay_member_at_admission: false, + }, + PolicyCheck::Denied => FinalAdmissionCheck { + denial: Some(AdmissionDenial { + metric: "not_relay_member", + reason: "restricted: not a relay member", + outcome: AuthOutcome::NotRelayMember, + class: buzz_auth::DenialClass::AuthorizationDenied, + }), + membership_via_owner: false, + owner_was_relay_member_at_admission: false, + }, + PolicyCheck::DependencyError => FinalAdmissionCheck { + denial: Some(AdmissionDenial { + metric: "relay_membership_check_error", + reason: "error: internal error checking relay membership", + outcome: AuthOutcome::RelayMembershipCheckError, + class: buzz_auth::DenialClass::AuthorizationUnavailable, + }), + membership_via_owner: false, + owner_was_relay_member_at_admission: false, + }, + } +} + /// Community-ban verdict for an authenticating principal: shared by every /// socket auth seam (root and audio) so they cannot drift. /// @@ -189,6 +221,18 @@ pub(crate) fn ban_admission_denial(ban: BanOutcome) -> Option { }) } +/// Final admission result, including whether relay membership depended on the +/// agent's owner. Live revalidation must apply the same policy path that +/// admitted the socket. +pub(crate) struct FinalAdmissionCheck { + pub(crate) denial: Option, + pub(crate) membership_via_owner: bool, + /// Whether the socket's captured owner was a relay member at admission. + /// This detects later owner removal without granting direct sessions the + /// owner's membership. + pub(crate) owner_was_relay_member_at_admission: bool, +} + /// Final ban and relay-membership verdict for an authenticating socket. /// /// Callers bind the socket to `pubkey` first (the registry a ban's or @@ -196,15 +240,16 @@ pub(crate) fn ban_admission_denial(ban: BanOutcome) -> Option { /// runs after the bind and cancels the socket, or it ran before, so its /// committed ban or removal is visible to these fresh reads. Checking before /// binding leaves a gap where both are missed. Callers must also refuse a -/// socket whose cancellation token fired. Both reads fail closed. -pub(crate) async fn final_admission_denial( +/// socket whose cancellation token fired. Every writer read fails closed. +pub(crate) async fn final_admission_check( state: &AppState, community: buzz_core::CommunityId, pubkey: nostr::PublicKey, auth_tag_json: Option<&str>, signed_auth_created_at: Option, token_owner: Option, -) -> Option { + admitted_owner: Option<[u8; 32]>, +) -> FinalAdmissionCheck { let ban = community_ban_outcome( state, community, @@ -216,9 +261,13 @@ pub(crate) async fn final_admission_denial( ) .await; if let Some(denial) = ban_admission_denial(ban) { - return Some(denial); + return FinalAdmissionCheck { + denial: Some(denial), + membership_via_owner: false, + owner_was_relay_member_at_admission: false, + }; } - match crate::api::relay_members::check_relay_membership_authoritative( + let membership = crate::api::relay_members::check_relay_membership_authoritative( state, community, pubkey.as_bytes(), @@ -226,25 +275,48 @@ pub(crate) async fn final_admission_denial( signed_auth_created_at, token_owner, ) - .await - { - Ok(crate::api::relay_members::MembershipDecision::Denied) => Some(AdmissionDenial { - metric: "not_relay_member", - reason: "restricted: not a relay member", - outcome: AuthOutcome::NotRelayMember, - class: buzz_auth::DenialClass::AuthorizationDenied, - }), - Ok(_) => None, - Err(e) => { - warn!(pubkey = %pubkey.to_hex(), error = %e, "relay membership recheck failed, denying (fail-closed)"); - Some(AdmissionDenial { - metric: "relay_membership_check_error", - reason: "error: internal error checking relay membership", - outcome: AuthOutcome::RelayMembershipCheckError, - class: buzz_auth::DenialClass::AuthorizationUnavailable, - }) - } + .await; + if let Err(error) = &membership { + warn!(pubkey = %pubkey.to_hex(), error, "relay membership recheck failed, denying (fail-closed)"); } + let mut final_check = final_admission_from_membership(classify_relay_membership(membership)); + if final_check.denial.is_some() { + return final_check; + } + + final_check.owner_was_relay_member_at_admission = + match (admitted_owner, final_check.membership_via_owner) { + (None, false) => false, + (Some(_), true) => true, + (Some(owner), false) => match state + .db + .is_relay_member_writer(community, &hex::encode(owner)) + .await + { + Ok(is_member) => is_member, + Err(_) => { + warn!("captured-owner membership snapshot failed at admission; denying"); + final_check.denial = Some(AdmissionDenial { + metric: "owner_membership_check_error", + reason: "error: internal error checking relay membership", + outcome: AuthOutcome::RelayMembershipCheckError, + class: buzz_auth::DenialClass::AuthorizationUnavailable, + }); + return final_check; + } + }, + (None, true) => { + warn!("owner-derived membership admission has no captured owner; denying"); + final_check.denial = Some(AdmissionDenial { + metric: "owner_membership_check_error", + reason: "error: internal error checking relay membership", + outcome: AuthOutcome::RelayMembershipCheckError, + class: buzz_auth::DenialClass::AuthorizationUnavailable, + }); + return final_check; + } + }; + final_check } /// Owner to record on an admitted socket: the proven NIP-OA owner, else the @@ -726,7 +798,7 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: } // Bind, then take the final ban/membership decision (see - // `final_admission_denial`), then refuse if a disconnect already + // `final_admission_check`), then refuse if a disconnect already // cancelled this socket. No await separates the last check from // `authenticate`. // A proven owner is recorded before the pubkey, so this socket's @@ -750,24 +822,29 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: .as_ref() .map(crate::nip_fi_core::NipFiSession::from_assertion), ); - match admitted_owner(&state, conn.tenant.community(), pubkey, nip_oa_owner).await { - Ok(Some(owner)) => state.conn_manager.set_admitted_owner(conn_id, owner), - Ok(None) => {} - Err(denial) => { - deny_admission(&conn, &event_id_hex, denial); - return; - } + let admitted_owner = + match admitted_owner(&state, conn.tenant.community(), pubkey, nip_oa_owner).await { + Ok(owner) => owner, + Err(denial) => { + deny_admission(&conn, &event_id_hex, denial); + return; + } + }; + if let Some(owner) = admitted_owner { + state.conn_manager.set_admitted_owner(conn_id, owner); } - let denial = match final_admission_denial( + let final_check_started_at = tokio::time::Instant::now(); + let final_check = final_admission_check( &state, conn.tenant.community(), pubkey, auth_tag_json.as_deref(), Some(signed_auth_created_at), token_owner, + admitted_owner, ) - .await - { + .await; + let denial = match final_check.denial { None if conn.cancel.is_cancelled() => Some(AdmissionDenial { metric: "revoked_during_auth", reason: "blocked: access revoked", @@ -823,6 +900,14 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: if conn.cancel.is_cancelled() { return; } + conn.community_control.mark_live_authorized( + pubkey.to_bytes(), + admitted_owner, + final_check.owner_was_relay_member_at_admission, + final_check.membership_via_owner, + crate::nip_fi_session::NipFiWsRoute::Root, + final_check_started_at, + ); state.conn_manager.mark_admitted(conn_id); conn.send(RelayMessage::ok(&event_id_hex, true, "")); // _auth_permit drops here — expiry's write guard may proceed. @@ -850,7 +935,7 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: mod tests { use super::{ ban_denial, classify_allowlist, classify_relay_membership, extract_auth_tag_json, - handle_auth, nip42_denial_class, BanOutcome, PolicyCheck, + final_admission_from_membership, handle_auth, nip42_denial_class, BanOutcome, PolicyCheck, }; use crate::api::relay_members::MembershipDecision; use crate::connection::{tests::test_conn_with_auth, AuthState}; @@ -966,7 +1051,7 @@ mod tests { // Fix 5: use Config::for_test() which holds NIP_FI_ENV_LOCK internally. [FI-TRACE-ENV-RACE] let mut config = crate::config::Config::for_test(); config.require_relay_membership = false; - config.database_url = "postgres://buzz:buzz_dev@127.0.0.1:1/buzz".to_string(); // sadscan:disable np.postgres.1 + config.database_url = "postgres://127.0.0.1:1/buzz".to_string(); config.redis_url = "redis://127.0.0.1:1".to_string(); // 100ms acquire timeout: a request that falls through to the stub // pool still waits, but for 100ms instead of sqlx's 30s default. @@ -1966,7 +2051,7 @@ mod tests { /// after the bind. A user in another community, paused at the same /// point, is still admitted. /// - /// Mutation: remove the `final_admission_denial` call from + /// Mutation: remove the `final_admission_check` call from /// `handle_auth` → the banned socket is admitted → RED. #[tokio::test] #[ignore = "requires Postgres — runs in postgres-ci nextest lane"] @@ -2063,7 +2148,7 @@ mod tests { /// member: the final check reads membership from the writer. A user /// in another community, paused at the same point, is still admitted. /// - /// Mutation: make `final_admission_denial` call the replica-routed + /// Mutation: make `final_admission_check` call the replica-routed /// `check_relay_membership` → the removed socket is admitted → RED. #[tokio::test] #[ignore = "requires Postgres — runs in postgres-ci nextest lane"] @@ -3849,6 +3934,23 @@ mod tests { classify_relay_membership(Err("database unavailable".to_owned())), PolicyCheck::DependencyError ); + + let open = final_admission_from_membership(PolicyCheck::Allowed(None)); + assert!(open.denial.is_none()); + assert!(!open.membership_via_owner); + let via_owner = final_admission_from_membership(PolicyCheck::Allowed(Some(owner))); + assert!(via_owner.denial.is_none()); + assert!(via_owner.membership_via_owner); + let denied = final_admission_from_membership(PolicyCheck::Denied); + assert_eq!( + denied.denial.as_ref().map(|denial| denial.class), + Some(buzz_auth::DenialClass::AuthorizationDenied) + ); + let unavailable = final_admission_from_membership(PolicyCheck::DependencyError); + assert_eq!( + unavailable.denial.as_ref().map(|denial| denial.class), + Some(buzz_auth::DenialClass::AuthorizationUnavailable) + ); } /// The handler owns retry classification, so drive its real terminal-state @@ -3940,7 +4042,7 @@ mod tests { let snapshotter = recorder.snapshotter(); let _recorder_guard = metrics::set_default_local_recorder(&recorder); let state = crate::state::tests::test_state_with_unreachable_database_url( - "postgres://buzz:buzz_dev@127.0.0.1:1/buzz", + "postgres://127.0.0.1:1/buzz", ) .await; diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 44d385a8d3..f5334d3a1e 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1539,6 +1539,16 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> { )); } + // Durable member-access backstop: unlike lossy Redis disconnect fan-out, + // this fixed cadence rechecks only admitted local sockets against the + // authoritative writer. Its bounded scan and DB-error grace establish the + // live authorization staleness budget independently of archive maintenance. + { + let access_state = Arc::clone(&state); + let cancel = access_state.community_revalidator_cancel.clone(); + tokio::spawn(run_live_authorization_revalidator(access_state, cancel)); + } + // Per-pod dependency diagnostics runtime: one seam starts the dependency // sampler and its independent completion-epoch republisher together. { @@ -2151,6 +2161,17 @@ async fn run_community_revalidator( .await; } +async fn run_live_authorization_revalidator(state: Arc, cancel: CancellationToken) { + run_periodic_until_cancelled( + buzz_relay::state::LIVE_AUTHORIZATION_REVALIDATION_INTERVAL, + cancel, + || async { + state.revalidate_live_authorizations().await; + }, + ) + .await; +} + async fn run_periodic_until_cancelled( period: std::time::Duration, cancel: CancellationToken, diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 7b8b717491..a8879a71df 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -9,9 +9,11 @@ use std::time::Instant; use axum::body::Bytes; use axum::extract::ws::{Message as WsMessage, Utf8Bytes as WsUtf8Bytes}; use dashmap::DashMap; -use futures_util::future::join_all; +use futures_util::future::{join_all, BoxFuture}; +use futures_util::stream::{self, StreamExt}; use tokio::sync::{mpsc, watch, Semaphore}; use tokio::task::JoinHandle; +use tokio::time::Instant as TokioInstant; use tokio_util::sync::CancellationToken; use uuid::Uuid; @@ -75,6 +77,39 @@ impl CommunityDisconnectReason { } } +/// Maximum time between authoritative rechecks for live member authorization. +pub const LIVE_AUTHORIZATION_REVALIDATION_INTERVAL: std::time::Duration = + std::time::Duration::from_secs(5); +/// A short database outage may preserve the most recent decision for this long. +pub(crate) const LIVE_AUTHORIZATION_DB_FAILURE_GRACE: std::time::Duration = + std::time::Duration::from_secs(10); +/// A live session must not keep a stale authorization decision beyond this budget. +pub(crate) const LIVE_AUTHORIZATION_MAX_STALENESS: std::time::Duration = + std::time::Duration::from_secs(30); +const LIVE_AUTHORIZATION_BATCH_SIZE: usize = 500; +const LIVE_AUTHORIZATION_QUERY_CONCURRENCY: usize = 4; +const LIVE_AUTHORIZATION_QUERY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(1); +const LIVE_AUTHORIZATION_SCAN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); +// Conservative schedule: allow the grace plus a full current/next scan, the +// cadence, and one query deadline. Keep the documented session budget above it. +const LIVE_AUTHORIZATION_WORST_CASE_STALENESS_SECS: u64 = LIVE_AUTHORIZATION_DB_FAILURE_GRACE + .as_secs() + + LIVE_AUTHORIZATION_REVALIDATION_INTERVAL.as_secs() + + 2 * LIVE_AUTHORIZATION_SCAN_TIMEOUT.as_secs() + + LIVE_AUTHORIZATION_QUERY_TIMEOUT.as_secs(); +const _: () = assert!( + LIVE_AUTHORIZATION_WORST_CASE_STALENESS_SECS <= LIVE_AUTHORIZATION_MAX_STALENESS.as_secs() +); +#[derive(Clone)] +struct LiveAuthorizationRecord { + pubkey: [u8; 32], + owner_pubkey: Option<[u8; 32]>, + owner_was_relay_member_at_admission: bool, + membership_via_owner: bool, + route: crate::nip_fi_session::NipFiWsRoute, + last_checked_at: TokioInstant, +} + /// NIP-42-proven key and the NIP-FI session a socket was admitted under. #[derive(Clone)] struct ProvenIdentity { @@ -106,6 +141,8 @@ pub(crate) struct CommunityConnectionControl { pubkey: Arc>, /// Owner of an admitted agent; revoking the owner closes this socket. owner: Arc>, + /// Authoritative policy facts for durable per-member revalidation. + live_authorization: Arc>>, /// Shadow-mode observation of this socket; no enforce path reads it. nip_fi_shadow: Arc>>, /// A shadow upgrade would-deny's line, owned by the connection task. @@ -122,6 +159,7 @@ impl CommunityConnectionControl { terminal_frame_tx: Arc::new(std::sync::Mutex::new(None)), pubkey: Arc::default(), owner: Arc::default(), + live_authorization: Arc::default(), nip_fi_shadow: Arc::default(), nip_fi_line: Arc::default(), } @@ -180,6 +218,50 @@ impl CommunityConnectionControl { let _ = self.owner.set(owner); } + /// Mark the socket admitted after its final authoritative policy check. + pub(crate) fn mark_live_authorized( + &self, + pubkey: [u8; 32], + owner_pubkey: Option<[u8; 32]>, + owner_was_relay_member_at_admission: bool, + membership_via_owner: bool, + route: crate::nip_fi_session::NipFiWsRoute, + checked_at: TokioInstant, + ) { + let mut slot = self + .live_authorization + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + *slot = Some(LiveAuthorizationRecord { + pubkey, + owner_pubkey, + owner_was_relay_member_at_admission, + membership_via_owner, + route, + last_checked_at: checked_at, + }); + } + + fn live_authorization(&self) -> Option { + self.live_authorization + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() + } + + fn record_live_authorization_check(&self, checked_at: TokioInstant) { + if self.cancel.is_cancelled() { + return; + } + let mut slot = self + .live_authorization + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(record) = slot.as_mut() { + record.last_checked_at = checked_at; + } + } + fn matches_revocation(&self, pubkey: &[u8], unowned_only: bool) -> bool { let principal = self.pubkey.get().map(|k| &k[..]) == Some(pubkey); match self.owner.get() { @@ -231,14 +313,28 @@ impl CommunityConnectionControl { &self, route: crate::nip_fi_session::NipFiWsRoute, frame_tx: Option<&mpsc::Sender>, + ) { + self.publish_nip_fi_denial(route, buzz_auth::DenialClass::AuthorizationDenied, frame_tx); + } + + fn publish_nip_fi_denial( + &self, + route: crate::nip_fi_session::NipFiWsRoute, + class: buzz_auth::DenialClass, + frame_tx: Option<&mpsc::Sender>, ) { let slot = self .terminal_frame_tx .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); + let reason = if class == buzz_auth::DenialClass::AuthorizationUnavailable { + CommunityDisconnectReason::AuthorizationUnavailable + } else { + CommunityDisconnectReason::AuthorizationDenied + }; let won = self.reason_tx.send_if_modified(|current| match current { None => { - *current = Some(CommunityDisconnectReason::AuthorizationDenied); + *current = Some(reason); true } Some(_) => false, @@ -247,10 +343,7 @@ impl CommunityConnectionControl { return; } if let Some(tx) = frame_tx.or(slot.as_ref()) { - let _ = tx.try_send(crate::nip_fi_session::denial_frame( - route, - buzz_auth::DenialClass::AuthorizationDenied, - )); + let _ = tx.try_send(crate::nip_fi_session::denial_frame(route, class)); } } @@ -270,8 +363,20 @@ impl CommunityConnectionControl { /// transition on the root connection's `terminal_ctrl_tx` (drained first /// by `send_loop` on cancel), then cancels. pub(crate) fn manager_disconnect_nip_fi(&self, frame_tx: &mpsc::Sender) { - self.publish_authorization_denied( + self.manager_disconnect_nip_fi_with_class( + frame_tx, + buzz_auth::DenialClass::AuthorizationDenied, + ); + } + + fn manager_disconnect_nip_fi_with_class( + &self, + frame_tx: &mpsc::Sender, + class: buzz_auth::DenialClass, + ) { + self.publish_nip_fi_denial( crate::nip_fi_session::NipFiWsRoute::Root, + class, Some(frame_tx), ); self.cancel.cancel(); @@ -294,23 +399,39 @@ impl CommunityConnectionControl { /// `authorization_denied` transition, so its client sees the same denial /// as any other NIP-FI refusal; an Off-mode socket closes `AccessRevoked`. /// Neither overwrites a terminal response already chosen. - fn revoke_access(&self) { + fn disconnect_live_authorization( + &self, + route: crate::nip_fi_session::NipFiWsRoute, + class: buzz_auth::DenialClass, + ) { let nip_fi = self .proven_identity .read() .is_ok_and(|id| id.as_ref().is_some_and(|id| id.nip_fi.is_some())); if nip_fi { - self.publish_authorization_denied(crate::nip_fi_session::NipFiWsRoute::Audio, None); + self.publish_nip_fi_denial(route, class, None); } else { + let reason = if class == buzz_auth::DenialClass::AuthorizationUnavailable { + CommunityDisconnectReason::AuthorizationUnavailable + } else { + CommunityDisconnectReason::AccessRevoked + }; self.reason_tx.send_if_modified(|current| { current.is_none() && { - *current = Some(CommunityDisconnectReason::AccessRevoked); + *current = Some(reason); true } }); } self.cancel.cancel(); } + + fn revoke_access(&self) { + self.disconnect_live_authorization( + crate::nip_fi_session::NipFiWsRoute::Audio, + buzz_auth::DenialClass::AuthorizationDenied, + ); + } } /// Leaves headroom under the process-wide drain deadline for a stalled writer. @@ -477,6 +598,66 @@ impl CommunityConnectionRegistry { .map(|entry| entry.value().0) .collect() } + + fn live_authorization_sessions(&self) -> Vec { + self.connections + .iter() + .filter_map(|entry| { + let (community_id, control) = entry.value(); + if control.cancel.is_cancelled() { + return None; + } + let record = control.live_authorization()?; + Some(LiveAuthorizationSession { + connection_id: *entry.key(), + community_id: *community_id, + route: record.route, + pubkey: record.pubkey, + owner_pubkey: record.owner_pubkey, + owner_was_relay_member_at_admission: record.owner_was_relay_member_at_admission, + membership_via_owner: record.membership_via_owner, + last_checked_at: record.last_checked_at, + }) + }) + .collect() + } + + fn record_live_authorization_check( + &self, + community_id: CommunityId, + connection_id: Uuid, + checked_at: TokioInstant, + ) { + if let Some(entry) = self.connections.get(&connection_id) { + let (registered_community, control) = entry.value(); + if *registered_community == community_id { + control.record_live_authorization_check(checked_at); + } + } + } + + fn disconnect_live_authorization( + &self, + community_id: CommunityId, + connection_id: Uuid, + class: buzz_auth::DenialClass, + ) -> bool { + let Some(entry) = self.connections.get(&connection_id) else { + return false; + }; + let (registered_community, control) = entry.value(); + if *registered_community != community_id || control.cancel.is_cancelled() { + return false; + } + let Some(record) = control.live_authorization() else { + return false; + }; + if record.route != crate::nip_fi_session::NipFiWsRoute::Audio { + return false; + } + control.disconnect_live_authorization(record.route, class); + true + } } /// Removes a socket lifecycle registration on every handler exit path. @@ -690,6 +871,338 @@ where (closed, failures) } +type LiveAuthorizationBatchLookup = Arc< + dyn Fn( + Vec, + ) -> BoxFuture<'static, buzz_db::Result>> + + Send + + Sync, +>; + +#[derive(Debug, Clone, Copy)] +enum LiveAuthorizationFailure { + Database, + QueryTimeout, + IncompleteResult, + ScanTimeout, +} + +impl LiveAuthorizationFailure { + const fn label(self) -> &'static str { + match self { + Self::Database => "database_error", + Self::QueryTimeout => "query_timeout", + Self::IncompleteResult => "incomplete_result", + Self::ScanTimeout => "scan_timeout", + } + } +} + +#[derive(Debug, Default, Clone, Copy)] +struct LiveAuthorizationRevalidationReport { + checked_targets: usize, + failed_targets: usize, + closed_sessions: usize, + closed_for_db_failure: usize, + scan_timed_out: bool, +} + +async fn revalidate_live_authorization_sessions( + registry: &CommunityConnectionRegistry, + conn_manager: &ConnectionManager, + require_relay_membership: bool, + allow_nip_oa_auth: bool, + cancel: &CancellationToken, + lookup: LiveAuthorizationBatchLookup, +) -> LiveAuthorizationRevalidationReport { + let started_at = TokioInstant::now(); + let scan_deadline = started_at + LIVE_AUTHORIZATION_SCAN_TIMEOUT; + let sessions = registry.live_authorization_sessions(); + metrics::gauge!("buzz_live_authorization_sessions").set(sessions.len() as f64); + if sessions.is_empty() { + return LiveAuthorizationRevalidationReport::default(); + } + + let mut sessions_by_target: HashMap<_, Vec<_>> = HashMap::new(); + for session in sessions { + sessions_by_target + .entry(buzz_db::LiveAuthorizationTarget { + community_id: session.community_id, + pubkey: session.pubkey, + session_owner_pubkey: session.owner_pubkey, + }) + .or_default() + .push(session); + } + let mut targets = sessions_by_target.keys().copied().collect::>(); + targets.sort_by_key(|target| { + ( + target.community_id, + target.pubkey, + target.session_owner_pubkey, + ) + }); + let batches = targets + .chunks(LIVE_AUTHORIZATION_BATCH_SIZE) + .map(|batch| batch.to_vec()) + .collect::>(); + + let mut outcomes = HashMap::with_capacity(targets.len()); + let mut checked_at_by_target = HashMap::with_capacity(targets.len()); + let batch_queries = stream::iter(batches.into_iter().map(|batch| { + let lookup = Arc::clone(&lookup); + async move { + let checked_at = TokioInstant::now(); + let query = lookup(batch.clone()); + let result = tokio::time::timeout(LIVE_AUTHORIZATION_QUERY_TIMEOUT, query).await; + (batch, checked_at, result) + } + })) + .buffer_unordered(LIVE_AUTHORIZATION_QUERY_CONCURRENCY); + tokio::pin!(batch_queries); + + let scan_result = tokio::select! { + biased; + _ = cancel.cancelled() => return LiveAuthorizationRevalidationReport::default(), + result = tokio::time::timeout_at(scan_deadline, async { + while let Some((batch, checked_at, result)) = batch_queries.next().await { + for target in &batch { + checked_at_by_target.insert(*target, checked_at); + } + match result { + Err(_) => { + metrics::counter!( + "buzz_live_authorization_batches_total", + "outcome" => "query_timeout", + ) + .increment(1); + tracing::warn!( + batch_size = batch.len(), + query_timeout_ms = LIVE_AUTHORIZATION_QUERY_TIMEOUT.as_millis() as u64, + "live authorization revalidation database lookup timed out" + ); + for target in batch { + outcomes.insert(target, Err(LiveAuthorizationFailure::QueryTimeout)); + } + } + Ok(Err(error)) => { + let class = live_authorization_db_error_class(&error); + metrics::counter!( + "buzz_live_authorization_batches_total", + "outcome" => class, + ) + .increment(1); + tracing::warn!( + batch_size = batch.len(), + error_class = class, + "live authorization revalidation database lookup failed" + ); + for target in batch { + outcomes.insert(target, Err(LiveAuthorizationFailure::Database)); + } + } + Ok(Ok(rows)) => { + let requested = batch.iter().copied().collect::>(); + let mut states = HashMap::with_capacity(rows.len()); + let mut malformed = false; + for row in rows { + let target = buzz_db::LiveAuthorizationTarget { + community_id: row.community_id, + pubkey: row.pubkey, + session_owner_pubkey: row.session_owner_pubkey, + }; + if !requested.contains(&target) + || states.insert(target, row).is_some() + { + malformed = true; + break; + } + } + let complete = !malformed && states.len() == requested.len(); + for target in batch { + let outcome = if complete { + states + .remove(&target) + .map(Ok) + .unwrap_or(Err(LiveAuthorizationFailure::IncompleteResult)) + } else { + Err(LiveAuthorizationFailure::IncompleteResult) + }; + outcomes.insert(target, outcome); + } + if !complete { + metrics::counter!( + "buzz_live_authorization_batches_total", + "outcome" => "incomplete_result", + ) + .increment(1); + tracing::error!( + batch_size = requested.len(), + "live authorization revalidation returned an incomplete or invalid batch" + ); + } + } + } + } + }) => result, + }; + + let mut report = LiveAuthorizationRevalidationReport { + scan_timed_out: scan_result.is_err(), + ..LiveAuthorizationRevalidationReport::default() + }; + if report.scan_timed_out { + metrics::counter!( + "buzz_live_authorization_batches_total", + "outcome" => "scan_timeout", + ) + .increment(1); + tracing::warn!( + target_count = targets.len(), + scan_budget_ms = LIVE_AUTHORIZATION_SCAN_TIMEOUT.as_millis() as u64, + "live authorization revalidation exceeded its scan budget" + ); + } + for target in targets { + let Some(sessions) = sessions_by_target.get(&target) else { + continue; + }; + match outcomes + .remove(&target) + .unwrap_or(Err(LiveAuthorizationFailure::ScanTimeout)) + { + Ok(access) => { + report.checked_targets += 1; + let checked_at = checked_at_by_target + .get(&target) + .copied() + .unwrap_or(started_at); + let mut target_revoked = false; + for session in sessions { + if live_authorization_allowed( + session, + &access, + require_relay_membership, + allow_nip_oa_auth, + ) { + registry.record_live_authorization_check( + session.community_id, + session.connection_id, + checked_at, + ); + } else { + let closed = disconnect_live_authorization_session( + registry, + conn_manager, + session, + buzz_auth::DenialClass::AuthorizationDenied, + ); + report.closed_sessions += usize::from(closed); + target_revoked |= closed; + } + } + metrics::counter!( + "buzz_live_authorization_targets_total", + "outcome" => if target_revoked { "revoked" } else { "allowed" }, + ) + .increment(1); + } + Err(failure) => { + report.failed_targets += 1; + metrics::counter!( + "buzz_live_authorization_targets_total", + "outcome" => failure.label(), + ) + .increment(1); + let now = TokioInstant::now(); + for session in sessions { + if now.saturating_duration_since(session.last_checked_at) + < LIVE_AUTHORIZATION_DB_FAILURE_GRACE + { + continue; + } + let closed = disconnect_live_authorization_session( + registry, + conn_manager, + session, + buzz_auth::DenialClass::AuthorizationUnavailable, + ); + report.closed_sessions += usize::from(closed); + report.closed_for_db_failure += usize::from(closed); + } + } + } + } + metrics::histogram!("buzz_live_authorization_scan_duration_seconds") + .record(started_at.elapsed().as_secs_f64()); + if report.closed_for_db_failure > 0 { + metrics::counter!("buzz_live_authorization_fail_closed_sessions_total") + .increment(report.closed_for_db_failure as u64); + } + report +} + +fn live_authorization_allowed( + session: &LiveAuthorizationSession, + access: &buzz_db::LiveAuthorizationState, + require_relay_membership: bool, + allow_nip_oa_auth: bool, +) -> bool { + // A direct member stays authorized through its own membership. The + // captured owner's prior membership is only a revocation signal for a + // later roster removal, never a source of membership for the principal. + // The first stored owner link also invalidates sockets admitted before + // that link existed, matching materialize_nip_oa_owner's immediate + // ownerless-session disconnect when a pod misses its fan-out. + if session.community_id != access.community_id + || session.pubkey != access.pubkey + || session.owner_pubkey != access.session_owner_pubkey + || (session.owner_pubkey.is_none() && access.agent_owner_pubkey.is_some()) + || (session.membership_via_owner && session.owner_pubkey != access.agent_owner_pubkey) + || access.banned + || (session.owner_was_relay_member_at_admission && !access.session_owner_is_relay_member) + { + return false; + } + if !require_relay_membership || access.relay_member { + return true; + } + session.membership_via_owner + && allow_nip_oa_auth + && session.owner_pubkey.is_some() + && access.owner_is_relay_member +} + +fn disconnect_live_authorization_session( + registry: &CommunityConnectionRegistry, + conn_manager: &ConnectionManager, + session: &LiveAuthorizationSession, + class: buzz_auth::DenialClass, +) -> bool { + match session.route { + crate::nip_fi_session::NipFiWsRoute::Root => conn_manager.disconnect_live_authorization( + session.community_id, + session.connection_id, + class, + ), + crate::nip_fi_session::NipFiWsRoute::Audio => registry.disconnect_live_authorization( + session.community_id, + session.connection_id, + class, + ), + } +} + +fn live_authorization_db_error_class(error: &buzz_db::DbError) -> &'static str { + match error { + buzz_db::DbError::Sqlx(sqlx::Error::PoolTimedOut) => "pool_timeout", + buzz_db::DbError::Sqlx(sqlx::Error::PoolClosed) => "pool_closed", + buzz_db::DbError::Sqlx(sqlx::Error::Io(_)) => "io_error", + error if error.is_statement_cancelled() => "statement_cancelled", + _ => "database_error", + } +} + /// Tracks active Nostr WebSocket connections and provides message routing by connection ID. pub struct ConnectionManager { connections: DashMap, @@ -921,6 +1434,35 @@ impl ConnectionManager { closed } + fn disconnect_live_authorization( + &self, + community: CommunityId, + conn_id: Uuid, + class: buzz_auth::DenialClass, + ) -> bool { + let Some(entry) = self.connections.get(&conn_id) else { + return false; + }; + if entry.community_id != community || entry.cancel.is_cancelled() { + return false; + } + + let nip_fi = entry + .nip_fi + .read() + .is_ok_and(|issuer| issuer.is_some()); + if nip_fi { + entry + .community_control + .manager_disconnect_nip_fi_with_class(&entry.terminal_ctrl_tx, class); + } else { + entry + .community_control + .disconnect_live_authorization(crate::nip_fi_session::NipFiWsRoute::Root, class); + } + true + } + /// Close all live connections admitted under NIP-FI `issuer` whose proven /// pubkey or recorded agent owner equals `pubkey`, **across all /// communities**. @@ -1246,7 +1788,7 @@ pub struct AppState { pub conn_manager: Arc, /// Lifecycle cancellation for every long-lived socket, including huddle audio. pub community_connections: Arc, - /// Stops only the periodic lifecycle revalidator during graceful shutdown. + /// Stops periodic community and live-authorization revalidation at shutdown. pub community_revalidator_cancel: CancellationToken, /// Cancels push claims and in-flight delivery on process shutdown. pub push_cancel: CancellationToken, @@ -2327,6 +2869,38 @@ impl AppState { closed } + /// Revalidate every authenticated root and audio socket against current + /// community membership, agent ownership, and ban state. + /// + /// The loop bounds a scan to four concurrent writer queries, 500 + /// authorization targets per query, and five seconds total. It fails closed + /// when an authoritative answer remains unavailable for ten seconds. + pub async fn revalidate_live_authorizations(&self) -> usize { + let db = self.db.clone(); + let lookup: LiveAuthorizationBatchLookup = Arc::new(move |targets| { + let db = db.clone(); + Box::pin(async move { db.live_authorization_state_batch(&targets).await }) + }); + let report = revalidate_live_authorization_sessions( + &self.community_connections, + &self.conn_manager, + self.config.require_relay_membership, + self.config.allow_nip_oa_auth, + &self.community_revalidator_cancel, + lookup, + ) + .await; + if report.closed_sessions > 0 { + tracing::info!( + closed = report.closed_sessions, + checked_targets = report.checked_targets, + failed_targets = report.failed_targets, + "closed live sessions whose current authorization could not be retained" + ); + } + report.closed_sessions + } + /// Get accessible channel IDs with a 10-second cache. Falls back to DB on miss. /// /// The open half is shared by every reader in the community and its @@ -4919,3 +5493,1337 @@ pub(crate) mod tests { } } } + +#[cfg(test)] +mod live_authorization_tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::sync::{Barrier, Mutex}; + + const TEST_ISSUER: &str = "https://issuer.example"; + + struct LiveConnectionFixture { + cancel: CancellationToken, + terminal_rx: mpsc::Receiver, + reason_rx: watch::Receiver>, + _guard: CommunityConnectionGuard, + } + + fn community(value: u128) -> CommunityId { + CommunityId::from_uuid(Uuid::from_u128(value)) + } + + fn access_state( + target: buzz_db::LiveAuthorizationTarget, + owner: Option<[u8; 32]>, + relay_member: bool, + owner_is_relay_member: bool, + banned: bool, + ) -> buzz_db::LiveAuthorizationState { + buzz_db::LiveAuthorizationState { + community_id: target.community_id, + pubkey: target.pubkey, + session_owner_pubkey: target.session_owner_pubkey, + session_owner_is_relay_member: target.session_owner_pubkey == owner + && owner_is_relay_member, + agent_owner_pubkey: owner, + relay_member, + owner_is_relay_member, + banned, + } + } + + #[allow(clippy::too_many_arguments)] // Mirrors the distinct socket admission facts. + fn register_live_connection( + registry: &CommunityConnectionRegistry, + conn_manager: &ConnectionManager, + community_id: CommunityId, + pubkey: [u8; 32], + owner_pubkey: Option<[u8; 32]>, + membership_via_owner: bool, + route: crate::nip_fi_session::NipFiWsRoute, + nip_fi: bool, + ) -> LiveConnectionFixture { + register_live_connection_with_owner_membership( + registry, + conn_manager, + community_id, + pubkey, + owner_pubkey, + membership_via_owner, + membership_via_owner, + route, + nip_fi, + ) + } + + #[allow(clippy::too_many_arguments)] // Mirrors the distinct socket admission facts. + fn register_live_connection_with_owner_membership( + registry: &CommunityConnectionRegistry, + conn_manager: &ConnectionManager, + community_id: CommunityId, + pubkey: [u8; 32], + owner_pubkey: Option<[u8; 32]>, + owner_was_relay_member_at_admission: bool, + membership_via_owner: bool, + route: crate::nip_fi_session::NipFiWsRoute, + nip_fi: bool, + ) -> LiveConnectionFixture { + let connection_id = Uuid::new_v4(); + let cancel = CancellationToken::new(); + let control = CommunityConnectionControl::new(cancel.clone()); + control.bind_pubkey(pubkey); + if let Some(owner) = owner_pubkey { + control.bind_owner(owner); + } + let reason_rx = control.disconnect_reason(); + let (terminal_tx, terminal_rx) = mpsc::channel(1); + let (ctrl_tx, _ctrl_rx) = mpsc::channel(8); + + if route == crate::nip_fi_session::NipFiWsRoute::Audio { + control.set_proven_identity(pubkey.to_vec(), nip_fi.then(|| TEST_ISSUER.to_owned())); + control.set_terminal_frame_sender(terminal_tx.clone()); + } + control.mark_live_authorized( + pubkey, + owner_pubkey, + owner_was_relay_member_at_admission, + membership_via_owner, + route, + TokioInstant::now(), + ); + let guard = registry.register(connection_id, community_id, control.clone()); + + if route == crate::nip_fi_session::NipFiWsRoute::Root { + let (tx, _rx) = mpsc::channel(8); + conn_manager.register( + connection_id, + tx, + ctrl_tx, + terminal_tx, + None, + cancel.clone(), + community_id, + Arc::new(AtomicU8::new(0)), + Arc::new(Mutex::new(HashMap::new())), + 3, + control, + ); + conn_manager.set_authenticated_identity( + connection_id, + pubkey.to_vec(), + nip_fi.then(|| TEST_ISSUER.to_owned()), + ); + if let Some(owner) = owner_pubkey { + conn_manager.set_admitted_owner(connection_id, owner); + } + conn_manager.mark_admitted(connection_id); + } else { + drop(ctrl_tx); + } + + LiveConnectionFixture { + cancel, + terminal_rx, + reason_rx, + _guard: guard, + } + } + + fn database_failure_lookup() -> LiveAuthorizationBatchLookup { + Arc::new(|_| Box::pin(async { Err(buzz_db::DbError::Sqlx(sqlx::Error::PoolClosed)) })) + } + + #[test] + fn poisoned_authorization_record_remains_visible_to_the_revalidator() { + let registry = CommunityConnectionRegistry::new(); + let tenant = community(0x270f); + let control = CommunityConnectionControl::new(CancellationToken::new()); + let poison = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let _slot = control.live_authorization.lock().expect("initial lock"); + panic!("simulate a panic while the authorization record is locked"); + })); + assert!(poison.is_err()); + + control.mark_live_authorized( + [0x10; 32], + None, + false, + false, + crate::nip_fi_session::NipFiWsRoute::Root, + TokioInstant::now(), + ); + let _guard = registry.register(Uuid::new_v4(), tenant, control); + assert_eq!(registry.live_authorization_sessions().len(), 1); + } + + #[test] + fn policy_preserves_direct_access_and_fences_delegation_or_stale_owner_links() { + let tenant = community(0x2710); + let principal = [0x11; 32]; + let old_owner = [0x22; 32]; + let new_owner = [0x33; 32]; + let target = buzz_db::LiveAuthorizationTarget { + community_id: tenant, + pubkey: principal, + session_owner_pubkey: Some(old_owner), + }; + let direct = LiveAuthorizationSession { + connection_id: Uuid::new_v4(), + community_id: tenant, + route: crate::nip_fi_session::NipFiWsRoute::Root, + pubkey: principal, + owner_pubkey: Some(old_owner), + owner_was_relay_member_at_admission: false, + membership_via_owner: false, + last_checked_at: TokioInstant::now(), + }; + let owner_derived = LiveAuthorizationSession { + membership_via_owner: true, + ..direct.clone() + }; + + let direct_member = access_state(target, Some(old_owner), true, false, false); + assert!(live_authorization_allowed( + &direct, + &direct_member, + true, + true, + )); + assert!(!live_authorization_allowed( + &owner_derived, + &access_state(target, Some(new_owner), true, false, false), + true, + true, + )); + + let ownerless_at_admission = LiveAuthorizationSession { + owner_pubkey: None, + owner_was_relay_member_at_admission: false, + membership_via_owner: false, + ..direct.clone() + }; + assert!(!live_authorization_allowed( + &ownerless_at_admission, + &access_state( + buzz_db::LiveAuthorizationTarget { + session_owner_pubkey: None, + ..target + }, + Some(new_owner), + true, + false, + false, + ), + true, + true, + )); + + let open_nonmember = access_state( + buzz_db::LiveAuthorizationTarget { + session_owner_pubkey: None, + ..target + }, + None, + false, + false, + false, + ); + assert!(live_authorization_allowed( + &ownerless_at_admission, + &open_nonmember, + false, + false, + )); + assert!(!live_authorization_allowed( + &owner_derived, + &access_state(target, Some(old_owner), false, true, true), + true, + true, + )); + } + + #[tokio::test] + async fn owner_derived_sessions_revalidate_owner_membership_and_identity() { + let registry = CommunityConnectionRegistry::new(); + let manager = ConnectionManager::new(); + let tenant = community(0x2717); + let old_owner = [0x81; 32]; + let new_owner = [0x82; 32]; + let lost_owner_member = [0x71; 32]; + let changed_owner = [0x72; 32]; + let direct_member = [0x73; 32]; + let owner_root = register_live_connection( + ®istry, + &manager, + tenant, + lost_owner_member, + Some(old_owner), + true, + crate::nip_fi_session::NipFiWsRoute::Root, + false, + ); + let owner_audio = register_live_connection( + ®istry, + &manager, + tenant, + lost_owner_member, + Some(old_owner), + true, + crate::nip_fi_session::NipFiWsRoute::Audio, + false, + ); + let changed_owner_audio = register_live_connection( + ®istry, + &manager, + tenant, + changed_owner, + Some(old_owner), + true, + crate::nip_fi_session::NipFiWsRoute::Audio, + false, + ); + let _direct_member = register_live_connection( + ®istry, + &manager, + tenant, + direct_member, + Some(old_owner), + false, + crate::nip_fi_session::NipFiWsRoute::Root, + false, + ); + + let lookup: LiveAuthorizationBatchLookup = Arc::new(move |targets| { + Box::pin(async move { + Ok(targets + .into_iter() + .map(|target| match target.pubkey { + key if key == lost_owner_member => { + access_state(target, Some(old_owner), false, false, false) + } + key if key == changed_owner => { + access_state(target, Some(new_owner), false, false, false) + } + _ => access_state(target, Some(new_owner), true, false, false), + }) + .collect()) + }) + }); + let report = revalidate_live_authorization_sessions( + ®istry, + &manager, + true, + true, + &CancellationToken::new(), + lookup, + ) + .await; + + assert_eq!(report.checked_targets, 3); + assert_eq!(report.closed_sessions, 3); + assert!(owner_root.cancel.is_cancelled()); + assert!(owner_audio.cancel.is_cancelled()); + assert!(changed_owner_audio.cancel.is_cancelled()); + assert!(!_direct_member.cancel.is_cancelled()); + } + + #[tokio::test] + async fn captured_owner_removal_and_ban_revoke_only_the_owned_sessions() { + let registry = CommunityConnectionRegistry::new(); + let manager = ConnectionManager::new(); + let tenant = community(0x2719); + let old_owner = [0x91; 32]; + let new_owner = [0x92; 32]; + let removed_agent = [0x93; 32]; + let banned_agent = [0x94; 32]; + + let removed_session = register_live_connection_with_owner_membership( + ®istry, + &manager, + tenant, + removed_agent, + Some(old_owner), + true, + false, + crate::nip_fi_session::NipFiWsRoute::Root, + false, + ); + let removed_agent_direct = register_live_connection( + ®istry, + &manager, + tenant, + removed_agent, + Some(new_owner), + false, + crate::nip_fi_session::NipFiWsRoute::Audio, + false, + ); + let banned_session = register_live_connection_with_owner_membership( + ®istry, + &manager, + tenant, + banned_agent, + Some(old_owner), + true, + false, + crate::nip_fi_session::NipFiWsRoute::Audio, + false, + ); + let banned_agent_direct = register_live_connection( + ®istry, + &manager, + tenant, + banned_agent, + Some(new_owner), + false, + crate::nip_fi_session::NipFiWsRoute::Root, + false, + ); + + let lookup: LiveAuthorizationBatchLookup = Arc::new(move |targets| { + Box::pin(async move { + Ok(targets + .into_iter() + .map(|target| { + access_state( + target, + Some(new_owner), + true, + false, + target.pubkey == banned_agent + && target.session_owner_pubkey == Some(old_owner), + ) + }) + .collect()) + }) + }); + let report = revalidate_live_authorization_sessions( + ®istry, + &manager, + true, + true, + &CancellationToken::new(), + lookup, + ) + .await; + + assert_eq!(report.closed_sessions, 2); + assert!(removed_session.cancel.is_cancelled()); + assert!(banned_session.cancel.is_cancelled()); + assert!( + !removed_agent_direct.cancel.is_cancelled(), + "the directly admitted principal remains a member after owner removal" + ); + assert!( + !banned_agent_direct.cancel.is_cancelled(), + "an owner ban does not ban the direct principal" + ); + } + + #[tokio::test] + async fn revalidation_is_tenant_scoped_for_root_and_audio_sessions() { + let registry = CommunityConnectionRegistry::new(); + let manager = ConnectionManager::new(); + let community_a = community(0x2711); + let community_b = community(0x2712); + let pubkey = [0x44; 32]; + let removed_root = register_live_connection( + ®istry, + &manager, + community_a, + pubkey, + None, + false, + crate::nip_fi_session::NipFiWsRoute::Root, + false, + ); + let _member_audio = register_live_connection( + ®istry, + &manager, + community_b, + pubkey, + None, + false, + crate::nip_fi_session::NipFiWsRoute::Audio, + false, + ); + + let lookup: LiveAuthorizationBatchLookup = Arc::new(move |targets| { + Box::pin(async move { + Ok(targets + .into_iter() + .map(|target| { + access_state( + target, + None, + target.community_id == community_b, + false, + false, + ) + }) + .collect()) + }) + }); + let report = revalidate_live_authorization_sessions( + ®istry, + &manager, + true, + true, + &CancellationToken::new(), + lookup, + ) + .await; + + assert_eq!(report.checked_targets, 2); + assert_eq!(report.closed_sessions, 1); + assert!(removed_root.cancel.is_cancelled()); + assert_eq!( + *removed_root.reason_rx.borrow(), + Some(CommunityDisconnectReason::AccessRevoked) + ); + assert_eq!( + *_member_audio.reason_rx.borrow(), + None, + "same key remains authorized in its own community" + ); + + let ban_lookup: LiveAuthorizationBatchLookup = Arc::new(move |targets| { + Box::pin(async move { + Ok(targets + .into_iter() + .map(|target| { + access_state( + target, + None, + true, + false, + target.community_id == community_b, + ) + }) + .collect()) + }) + }); + let report = revalidate_live_authorization_sessions( + ®istry, + &manager, + true, + true, + &CancellationToken::new(), + ban_lookup, + ) + .await; + assert_eq!(report.closed_sessions, 1); + assert!( + _member_audio.cancel.is_cancelled(), + "a community ban applies only in community B" + ); + } + + #[tokio::test(start_paused = true)] + async fn incomplete_batch_does_not_refresh_any_target_authorization() { + let registry = CommunityConnectionRegistry::new(); + let manager = ConnectionManager::new(); + let tenant = community(0x2720); + let first = register_live_connection( + ®istry, + &manager, + tenant, + [0x45; 32], + None, + false, + crate::nip_fi_session::NipFiWsRoute::Root, + false, + ); + let second = register_live_connection( + ®istry, + &manager, + tenant, + [0x46; 32], + None, + false, + crate::nip_fi_session::NipFiWsRoute::Audio, + false, + ); + let partial_lookup: LiveAuthorizationBatchLookup = Arc::new(|mut targets| { + Box::pin(async move { + Ok(targets + .drain(..1) + .map(|target| access_state(target, None, true, false, false)) + .collect()) + }) + }); + + let grace = revalidate_live_authorization_sessions( + ®istry, + &manager, + true, + true, + &CancellationToken::new(), + Arc::clone(&partial_lookup), + ) + .await; + assert_eq!(grace.checked_targets, 0); + assert_eq!(grace.failed_targets, 2); + assert_eq!(grace.closed_sessions, 0); + assert!(!first.cancel.is_cancelled()); + assert!(!second.cancel.is_cancelled()); + + tokio::time::advance(LIVE_AUTHORIZATION_DB_FAILURE_GRACE).await; + let expired = revalidate_live_authorization_sessions( + ®istry, + &manager, + true, + true, + &CancellationToken::new(), + partial_lookup, + ) + .await; + assert_eq!(expired.closed_for_db_failure, 2); + assert!(first.cancel.is_cancelled()); + assert!(second.cancel.is_cancelled()); + } + + #[tokio::test] + async fn revalidation_preserves_denial_semantics_for_root_audio_and_nip_fi() { + let registry = CommunityConnectionRegistry::new(); + let manager = ConnectionManager::new(); + let tenant = community(0x2713); + let cases = [ + (crate::nip_fi_session::NipFiWsRoute::Root, false), + (crate::nip_fi_session::NipFiWsRoute::Root, true), + (crate::nip_fi_session::NipFiWsRoute::Audio, false), + (crate::nip_fi_session::NipFiWsRoute::Audio, true), + ]; + let mut sessions = cases + .iter() + .enumerate() + .map(|(index, (route, nip_fi))| { + register_live_connection( + ®istry, + &manager, + tenant, + [index as u8 + 1; 32], + None, + false, + *route, + *nip_fi, + ) + }) + .collect::>(); + + let lookup: LiveAuthorizationBatchLookup = Arc::new(|targets| { + Box::pin(async move { + Ok(targets + .into_iter() + .map(|target| access_state(target, None, false, false, false)) + .collect()) + }) + }); + let report = revalidate_live_authorization_sessions( + ®istry, + &manager, + true, + false, + &CancellationToken::new(), + lookup, + ) + .await; + + assert_eq!(report.closed_sessions, 4); + for session in &sessions { + assert!(session.cancel.is_cancelled()); + } + assert_eq!( + *sessions[0].reason_rx.borrow(), + Some(CommunityDisconnectReason::AccessRevoked) + ); + let root_fi = sessions[1] + .terminal_rx + .try_recv() + .expect("NIP-FI root terminal denial"); + let WsMessage::Text(root_fi) = root_fi else { + panic!("expected NIP-FI root denial frame"); + }; + assert!(root_fi.contains("authorization denied")); + assert_eq!( + *sessions[2].reason_rx.borrow(), + Some(CommunityDisconnectReason::AccessRevoked) + ); + let audio_fi = sessions[3] + .terminal_rx + .try_recv() + .expect("NIP-FI audio terminal denial"); + let WsMessage::Text(audio_fi) = audio_fi else { + panic!("expected NIP-FI audio denial frame"); + }; + assert!(audio_fi.contains("authorization denied")); + } + + #[tokio::test(start_paused = true)] + async fn database_failure_grace_is_bounded_and_success_resets_it() { + let registry = CommunityConnectionRegistry::new(); + let manager = ConnectionManager::new(); + let tenant = community(0x2714); + let pubkey = [0x55; 32]; + let mut sessions = [ + register_live_connection( + ®istry, + &manager, + tenant, + pubkey, + None, + false, + crate::nip_fi_session::NipFiWsRoute::Root, + false, + ), + register_live_connection( + ®istry, + &manager, + tenant, + pubkey, + None, + false, + crate::nip_fi_session::NipFiWsRoute::Root, + true, + ), + register_live_connection( + ®istry, + &manager, + tenant, + pubkey, + None, + false, + crate::nip_fi_session::NipFiWsRoute::Audio, + false, + ), + register_live_connection( + ®istry, + &manager, + tenant, + pubkey, + None, + false, + crate::nip_fi_session::NipFiWsRoute::Audio, + true, + ), + ]; + let failure = database_failure_lookup(); + let first = revalidate_live_authorization_sessions( + ®istry, + &manager, + true, + true, + &CancellationToken::new(), + Arc::clone(&failure), + ) + .await; + assert_eq!(first.closed_sessions, 0); + assert_eq!(first.failed_targets, 1); + + tokio::time::advance( + LIVE_AUTHORIZATION_DB_FAILURE_GRACE - std::time::Duration::from_secs(1), + ) + .await; + let recovered: LiveAuthorizationBatchLookup = Arc::new(|targets| { + Box::pin(async move { + Ok(targets + .into_iter() + .map(|target| access_state(target, None, true, false, false)) + .collect()) + }) + }); + let success = revalidate_live_authorization_sessions( + ®istry, + &manager, + true, + true, + &CancellationToken::new(), + recovered, + ) + .await; + assert_eq!(success.checked_targets, 1); + assert!(sessions + .iter() + .all(|session| !session.cancel.is_cancelled())); + + tokio::time::advance(std::time::Duration::from_secs(9)).await; + let repeated_failure = revalidate_live_authorization_sessions( + ®istry, + &manager, + true, + true, + &CancellationToken::new(), + Arc::clone(&failure), + ) + .await; + assert_eq!(repeated_failure.closed_sessions, 0); + tokio::time::advance(std::time::Duration::from_secs(1)).await; + let expired_grace = revalidate_live_authorization_sessions( + ®istry, + &manager, + true, + true, + &CancellationToken::new(), + failure, + ) + .await; + assert_eq!(expired_grace.closed_for_db_failure, 4); + assert!(sessions.iter().all(|session| session.cancel.is_cancelled())); + + assert_eq!( + *sessions[0].reason_rx.borrow(), + Some(CommunityDisconnectReason::AuthorizationUnavailable) + ); + for index in [1, 3] { + let frame = sessions[index] + .terminal_rx + .try_recv() + .expect("NIP-FI unavailable terminal frame"); + let WsMessage::Text(frame) = frame else { + panic!("expected NIP-FI unavailable frame"); + }; + assert!(frame.contains("authorization unavailable")); + } + assert_eq!( + *sessions[2].reason_rx.borrow(), + Some(CommunityDisconnectReason::AuthorizationUnavailable) + ); + } + + #[tokio::test(start_paused = true)] + async fn query_timeout_does_not_extend_the_authorization_grace() { + let registry = CommunityConnectionRegistry::new(); + let manager = ConnectionManager::new(); + let session = register_live_connection( + ®istry, + &manager, + community(0x2718), + [0x56; 32], + None, + false, + crate::nip_fi_session::NipFiWsRoute::Audio, + false, + ); + let timeout_lookup: LiveAuthorizationBatchLookup = Arc::new(|_| { + Box::pin(async { + std::future::pending::>>() + .await + }) + }); + + let first = revalidate_live_authorization_sessions( + ®istry, + &manager, + true, + true, + &CancellationToken::new(), + Arc::clone(&timeout_lookup), + ) + .await; + assert_eq!(first.failed_targets, 1); + assert_eq!(first.closed_sessions, 0); + tokio::time::advance( + LIVE_AUTHORIZATION_DB_FAILURE_GRACE + - LIVE_AUTHORIZATION_QUERY_TIMEOUT.saturating_mul(2), + ) + .await; + let expired = revalidate_live_authorization_sessions( + ®istry, + &manager, + true, + true, + &CancellationToken::new(), + timeout_lookup, + ) + .await; + + assert_eq!(expired.closed_for_db_failure, 1); + assert_eq!( + *session.reason_rx.borrow(), + Some(CommunityDisconnectReason::AuthorizationUnavailable) + ); + assert!(session.cancel.is_cancelled()); + } + + #[tokio::test] + async fn admission_added_during_a_scan_is_checked_on_the_next_scan() { + let registry = Arc::new(CommunityConnectionRegistry::new()); + let manager = Arc::new(ConnectionManager::new()); + let tenant = community(0x2715); + let _first = register_live_connection( + ®istry, + &manager, + tenant, + [0x61; 32], + None, + false, + crate::nip_fi_session::NipFiWsRoute::Audio, + false, + ); + let started = Arc::new(Barrier::new(2)); + let release = Arc::new(Barrier::new(2)); + let lookup: LiveAuthorizationBatchLookup = Arc::new({ + let started = Arc::clone(&started); + let release = Arc::clone(&release); + move |targets| { + let started = Arc::clone(&started); + let release = Arc::clone(&release); + Box::pin(async move { + started.wait().await; + release.wait().await; + Ok(targets + .into_iter() + .map(|target| access_state(target, None, true, false, false)) + .collect()) + }) + } + }); + let scan_registry = Arc::clone(®istry); + let scan_manager = Arc::clone(&manager); + let scan = tokio::spawn(async move { + revalidate_live_authorization_sessions( + &scan_registry, + &scan_manager, + true, + true, + &CancellationToken::new(), + lookup, + ) + .await + }); + + started.wait().await; + let admitted_during_scan = register_live_connection( + ®istry, + &manager, + tenant, + [0x62; 32], + None, + false, + crate::nip_fi_session::NipFiWsRoute::Audio, + false, + ); + release.wait().await; + let first_report = scan.await.expect("first revalidation task"); + assert_eq!(first_report.checked_targets, 1); + assert!(!admitted_during_scan.cancel.is_cancelled()); + + let next_lookup: LiveAuthorizationBatchLookup = Arc::new(|targets| { + Box::pin(async move { + Ok(targets + .into_iter() + .map(|target| { + access_state(target, None, target.pubkey == [0x61; 32], false, false) + }) + .collect()) + }) + }); + let next_report = revalidate_live_authorization_sessions( + ®istry, + &manager, + true, + true, + &CancellationToken::new(), + next_lookup, + ) + .await; + assert_eq!(next_report.closed_sessions, 1); + assert!(admitted_during_scan.cancel.is_cancelled()); + } + + struct InFlightQuery(Arc); + + impl Drop for InFlightQuery { + fn drop(&mut self) { + self.0.fetch_sub(1, Ordering::SeqCst); + } + } + + #[tokio::test] + async fn ten_thousand_sessions_use_bounded_batches_and_query_concurrency() { + const PRINCIPALS: usize = 10_000; + let registry = CommunityConnectionRegistry::new(); + let manager = ConnectionManager::new(); + let tenant = community(0x2716); + let mut guards = Vec::with_capacity(PRINCIPALS); + for index in 0..PRINCIPALS { + let mut pubkey = [0; 32]; + pubkey[..8].copy_from_slice(&(index as u64).to_be_bytes()); + let cancel = CancellationToken::new(); + let control = CommunityConnectionControl::new(cancel.clone()); + control.mark_live_authorized( + pubkey, + None, + false, + false, + crate::nip_fi_session::NipFiWsRoute::Audio, + TokioInstant::now(), + ); + guards.push(registry.register(Uuid::new_v4(), tenant, control)); + } + + let in_flight = Arc::new(AtomicUsize::new(0)); + let max_in_flight = Arc::new(AtomicUsize::new(0)); + let max_batch = Arc::new(AtomicUsize::new(0)); + let total_targets = Arc::new(AtomicUsize::new(0)); + let lookup: LiveAuthorizationBatchLookup = Arc::new({ + let in_flight = Arc::clone(&in_flight); + let max_in_flight = Arc::clone(&max_in_flight); + let max_batch = Arc::clone(&max_batch); + let total_targets = Arc::clone(&total_targets); + move |targets| { + let in_flight = Arc::clone(&in_flight); + let max_in_flight = Arc::clone(&max_in_flight); + let max_batch = Arc::clone(&max_batch); + let total_targets = Arc::clone(&total_targets); + Box::pin(async move { + let active = in_flight.fetch_add(1, Ordering::SeqCst) + 1; + max_in_flight.fetch_max(active, Ordering::SeqCst); + max_batch.fetch_max(targets.len(), Ordering::SeqCst); + total_targets.fetch_add(targets.len(), Ordering::SeqCst); + let _in_flight = InFlightQuery(in_flight); + tokio::task::yield_now().await; + Ok(targets + .into_iter() + .map(|target| access_state(target, None, false, false, false)) + .collect()) + }) + } + }); + + let started_at = std::time::Instant::now(); + let report = revalidate_live_authorization_sessions( + ®istry, + &manager, + false, + false, + &CancellationToken::new(), + lookup, + ) + .await; + let elapsed = started_at.elapsed(); + + assert_eq!(report.checked_targets, PRINCIPALS); + assert_eq!(report.closed_sessions, 0); + assert_eq!( + max_batch.load(Ordering::SeqCst), + LIVE_AUTHORIZATION_BATCH_SIZE + ); + assert_eq!(total_targets.load(Ordering::SeqCst), PRINCIPALS); + assert!(max_in_flight.load(Ordering::SeqCst) <= LIVE_AUTHORIZATION_QUERY_CONCURRENCY); + assert!(max_in_flight.load(Ordering::SeqCst) > 1); + eprintln!( + "10k-session scan: {PRINCIPALS} targets, batch {}, max {} concurrent lookups, {} ms", + max_batch.load(Ordering::SeqCst), + max_in_flight.load(Ordering::SeqCst), + elapsed.as_millis(), + ); + drop(guards); + } + + #[cfg(test)] + mod postgres_tests { + use super::*; + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn missed_member_owner_ban_and_first_owner_link_revoke_root_audio_sessions() { + let pool = sqlx::PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect to disposable PostgreSQL test database"); + let db = Db::from_pool(pool); + let host = format!("live-authorization-{}.example", Uuid::new_v4().simple()); + let tenant = db + .ensure_configured_community(&host) + .await + .expect("create test community") + .id; + let direct_member = nostr::Keys::generate().public_key().to_bytes(); + let removed_agent = nostr::Keys::generate().public_key().to_bytes(); + let banned_agent = nostr::Keys::generate().public_key().to_bytes(); + let removed_owner = nostr::Keys::generate().public_key().to_bytes(); + let banned_owner = nostr::Keys::generate().public_key().to_bytes(); + let current_owner = nostr::Keys::generate().public_key().to_bytes(); + let late_link_agent = nostr::Keys::generate().public_key().to_bytes(); + let stable_agent = nostr::Keys::generate().public_key().to_bytes(); + let actor = nostr::Keys::generate().public_key().to_bytes(); + for key in [ + direct_member, + removed_agent, + banned_agent, + removed_owner, + banned_owner, + current_owner, + late_link_agent, + stable_agent, + actor, + ] { + db.ensure_user_for_authorization(tenant, &key) + .await + .expect("create disposable authorization user"); + } + for key in [ + direct_member, + removed_agent, + banned_agent, + removed_owner, + banned_owner, + current_owner, + late_link_agent, + stable_agent, + ] { + assert!(db + .add_relay_member(tenant, &hex::encode(key), "member", None) + .await + .expect("add disposable roster member")); + } + db.set_agent_owner_for_authorization(tenant, &removed_agent, &removed_owner) + .await + .expect("record removed agent owner before admission"); + db.set_agent_owner_for_authorization(tenant, &banned_agent, &banned_owner) + .await + .expect("record banned agent owner before admission"); + + let registry = CommunityConnectionRegistry::new(); + let manager = ConnectionManager::new(); + let root = register_live_connection( + ®istry, + &manager, + tenant, + direct_member, + None, + false, + crate::nip_fi_session::NipFiWsRoute::Root, + false, + ); + let audio = register_live_connection( + ®istry, + &manager, + tenant, + direct_member, + None, + false, + crate::nip_fi_session::NipFiWsRoute::Audio, + false, + ); + let removed_owner_root = register_live_connection_with_owner_membership( + ®istry, + &manager, + tenant, + removed_agent, + Some(removed_owner), + true, + false, + crate::nip_fi_session::NipFiWsRoute::Root, + false, + ); + let removed_owner_audio = register_live_connection_with_owner_membership( + ®istry, + &manager, + tenant, + removed_agent, + Some(removed_owner), + true, + false, + crate::nip_fi_session::NipFiWsRoute::Audio, + false, + ); + let banned_owner_root = register_live_connection_with_owner_membership( + ®istry, + &manager, + tenant, + banned_agent, + Some(banned_owner), + true, + false, + crate::nip_fi_session::NipFiWsRoute::Root, + false, + ); + let banned_owner_audio = register_live_connection_with_owner_membership( + ®istry, + &manager, + tenant, + banned_agent, + Some(banned_owner), + true, + false, + crate::nip_fi_session::NipFiWsRoute::Audio, + false, + ); + let late_link_root = register_live_connection( + ®istry, + &manager, + tenant, + late_link_agent, + None, + false, + crate::nip_fi_session::NipFiWsRoute::Root, + false, + ); + let late_link_audio = register_live_connection( + ®istry, + &manager, + tenant, + late_link_agent, + None, + false, + crate::nip_fi_session::NipFiWsRoute::Audio, + false, + ); + let stable_member_root = register_live_connection( + ®istry, + &manager, + tenant, + stable_agent, + None, + false, + crate::nip_fi_session::NipFiWsRoute::Root, + false, + ); + let stable_member_audio = register_live_connection( + ®istry, + &manager, + tenant, + stable_agent, + None, + false, + crate::nip_fi_session::NipFiWsRoute::Audio, + false, + ); + + db.set_agent_owner_for_authorization(tenant, &late_link_agent, ¤t_owner) + .await + .expect("commit first stored owner link after admission"); + assert_eq!( + db.remove_relay_member(tenant, &hex::encode(direct_member)) + .await + .expect("commit direct member removal"), + buzz_db::relay_members::RemoveResult::Removed + ); + assert_eq!( + db.remove_relay_member(tenant, &hex::encode(removed_owner)) + .await + .expect("commit captured owner removal"), + buzz_db::relay_members::RemoveResult::Removed + ); + db.ban_community_member(tenant, &banned_owner, &actor, None, None) + .await + .expect("commit captured owner ban"); + // Deliberately send no registry disconnect or Redis command. The + // sockets remain live until the production scan reads the writer. + assert!(!root.cancel.is_cancelled()); + assert!(!audio.cancel.is_cancelled()); + assert!(!removed_owner_root.cancel.is_cancelled()); + assert!(!removed_owner_audio.cancel.is_cancelled()); + assert!(!banned_owner_root.cancel.is_cancelled()); + assert!(!banned_owner_audio.cancel.is_cancelled()); + assert!(!late_link_root.cancel.is_cancelled()); + assert!(!late_link_audio.cancel.is_cancelled()); + let committed_at = std::time::Instant::now(); + + let lookup: LiveAuthorizationBatchLookup = Arc::new({ + let db = db.clone(); + move |targets| { + let db = db.clone(); + Box::pin(async move { db.live_authorization_state_batch(&targets).await }) + } + }); + let report = revalidate_live_authorization_sessions( + ®istry, + &manager, + true, + true, + &CancellationToken::new(), + lookup, + ) + .await; + let elapsed = committed_at.elapsed(); + + assert_eq!(report.checked_targets, 5); + assert_eq!(report.closed_sessions, 8); + assert!(root.cancel.is_cancelled()); + assert!(audio.cancel.is_cancelled()); + assert!(removed_owner_root.cancel.is_cancelled()); + assert!(removed_owner_audio.cancel.is_cancelled()); + assert!(banned_owner_root.cancel.is_cancelled()); + assert!(banned_owner_audio.cancel.is_cancelled()); + assert!( + late_link_root.cancel.is_cancelled() && late_link_audio.cancel.is_cancelled(), + "sessions admitted ownerless are revoked when a stored owner link is later added" + ); + assert!( + !stable_member_root.cancel.is_cancelled() + && !stable_member_audio.cancel.is_cancelled(), + "a stable direct member with no owner link remains authorized" + ); + assert!(elapsed < LIVE_AUTHORIZATION_MAX_STALENESS); + eprintln!( + "observed committed-removal, ban, or owner-link change to root/audio-revocation latency: {} ms", + elapsed.as_millis(), + ); + } + } +} + +/// One admitted socket and the policy facts captured when it was authorized. +#[derive(Debug, Clone)] +pub(crate) struct LiveAuthorizationSession { + /// Connection identifier used to close only this socket. + pub connection_id: Uuid, + /// Host-resolved tenant for the socket. + pub community_id: CommunityId, + /// Route-specific terminal behavior. + pub route: crate::nip_fi_session::NipFiWsRoute, + /// NIP-42-authenticated principal. + pub pubkey: [u8; 32], + /// Owner recorded when the session was admitted. + pub owner_pubkey: Option<[u8; 32]>, + /// Whether the captured owner was a relay member when this socket was admitted. + pub owner_was_relay_member_at_admission: bool, + /// Whether closed-relay membership was granted through NIP-OA delegation. + pub membership_via_owner: bool, + /// Most recent successful authoritative access check. + pub last_checked_at: TokioInstant, +} + +/// Why a community-bound socket is being asked to stop. +/// +/// Only deletion is externally attributed today. Ordinary lifecycle exits keep +/// using cancellation alone and therefore retain the existing bare-close +/// behavior. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum CommunityDisconnectReason { + CommunityDeleted, + /// NIP-FI: the connection's proven pubkey was added to the deny set. + AuthorizationDenied, + /// The authenticated pubkey lost access to the community (e.g. a ban). + AccessRevoked, + /// The relay could not re-establish the socket's access decision. + AuthorizationUnavailable, +} + +impl CommunityDisconnectReason { + pub(crate) fn close_message(self) -> WsMessage { + match self { + Self::CommunityDeleted => WsMessage::Close(Some(axum::extract::ws::CloseFrame { + code: axum::extract::ws::close_code::POLICY, + reason: WsUtf8Bytes::from_static("community deleted"), + })), + Self::AuthorizationDenied => WsMessage::Close(Some(axum::extract::ws::CloseFrame { + code: axum::extract::ws::close_code::POLICY, + reason: WsUtf8Bytes::from_static("authorization denied"), + })), + Self::AccessRevoked => WsMessage::Close(Some(axum::extract::ws::CloseFrame { + code: axum::extract::ws::close_code::POLICY, + reason: WsUtf8Bytes::from_static("access revoked"), + })), + Self::AuthorizationUnavailable => { + WsMessage::Close(Some(axum::extract::ws::CloseFrame { + code: axum::extract::ws::close_code::POLICY, + reason: WsUtf8Bytes::from_static("authorization unavailable"), + })) + } + } + } +}