diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index dfb7302abf9..e83d089b965 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -259,6 +259,103 @@ 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 5 seconds plus a stable per-socket +offset below 9 seconds, measured from the +start of that session's last successful authoritative check; repeated failures +never extend the cutoff. The offset comes from the server-generated connection +UUID, spreading outage closes across scans without a timer per socket. 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. Each query runs in a transaction used only for the read with a server-side statement +limit of at most 900 ms, below the shared 1-second client deadline; a tighter +operator limit is preserved. Rollback restores the writer session's limit. +Cancelled or failed lookups close the checked-out connection under SQLx's +bounded close path rather than waiting on its return-to-pool ping. The server +limit also stops work when PostgreSQL has not yet noticed a disconnect. +The conservative staleness bound is 5 seconds of base grace, 9 seconds of jitter, +one 5-second cadence, two 5-second scan windows, and one 1-second query deadline: +30 seconds, matching the 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. When the delegation check did not already establish the captured +owner's membership, admission makes a writer-authoritative membership +read to record whether that owner was already a relay member. This adds a writer +dependency to admission; a failed point read refuses the socket with an +authorization-unavailable AUTH response. The baseline distinguishes 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. The desktop leaves explicit +server-side policy-read failures unauthenticated and retries them with its +existing jittered exponential reconnect schedule, capped at 30 seconds. A +successful later admission restores the connection without manual preconnect; +`restricted:` and `blocked:` responses remain terminal, and invalid or unknown +AUTH evidence retains its three-rejection cap. 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. + +Targets are ordered within each scan by the oldest last-successful-check time +among their sessions, then by community, principal, and captured owner. A target +missed at a deadline therefore gets priority on the next scan while it remains +live. Under sustained writer saturation, deadlines can still leave targets +unresolved. Each unresolved socket closes at its own grace-plus-offset cutoff. +The five-second scan cadence spreads closes over two or three sweeps rather +than continuously. Compared with the former ten-second grace, a six-to-ten-second +blip can now close some sockets; the stable offsets reduce the peak reconnect +load while preserving the thirty-second bound. Confirmed revocations still +close immediately when an authoritative snapshot denies access. + ### Step 5: Cleanup On disconnect (any cause): diff --git a/Justfile b/Justfile index 4a1181a6f66..ea54d1d3f41 100644 --- a/Justfile +++ b/Justfile @@ -641,6 +641,7 @@ test-unit: + test(=state::tests::nip_fi_disconnect_does_not_close_unproven_audio_socket) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::enforce_end_is_recorded_only_by_the_winning_terminal) + + test(=state::tests::live_revalidation_does_not_record_deny_set_revocation) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=state::tests::manager_disconnect_sets_reason_enqueues_frame_then_cancels) + test(/^api::nip_fi::/)' @@ -661,6 +662,8 @@ test-unit: cargo nextest run -p buzz-relay --lib -E 'test(=state::tests::admitted_index_follows_every_key_change) + test(/^shared_reads::tests::/) + test(/^membership_cache::tests::/) + test(/^community_active::tests::/) + test(=state::tests::concurrent_open_channel_misses_share_the_in_flight_read) + test(=state::tests::concurrent_restriction_and_host_misses_share_the_in_flight_read) + test(=tenant::tests::app_state_resolver_serves_fresh_cached_hosts_only) + test(=tenant::tests::an_overlong_host_fails_closed_before_the_resolver) + test(=state::tests::disconnect_drops_the_cached_restriction_row)' # Retain partially consumed mesh receives across housekeeping ticks. cargo nextest run -p buzz-relay --lib -E 'test(=mesh_boot::tests::demo_echo_retains_pending_receive_across_housekeeping_ticks)' + # Durable session policy, fairness, timeout and scale contracts are infra-free. + cargo nextest run -p buzz-relay --lib -E 'test(/^state::live_authorization_tests::/)' # boot_lifecycle spawns the real relay binary and asserts its startup # lifecycle, including that buzz_startup_phase_* reaches /metrics. Its # non-ignored tests need no Postgres or Redis; the Postgres cases are diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 92eeb42f882..3d3b732d5aa 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -92,10 +92,11 @@ pub(crate) use runtime::{ pub use store::{ admin_moderation, allowlist, api_token, archived_identities, artifact, artifact_sync, channel, channel_members, community, community_auth, deletion, dm, entity_deletion, event, feed, - git_repo, leader_lock, media, membership_snapshot, 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, user, user_deletion, user_deletion_fences, workflow, + git_repo, leader_lock, live_authorization, media, membership_snapshot, 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, user, user_deletion, + user_deletion_fences, workflow, }; pub use allowlist::AllowlistEntry; @@ -110,6 +111,9 @@ pub use error::{DbError, Result}; pub use event::{ChannelHeadPrecondition, ChannelHeadWriteStatus, DeletionTarget}; pub use event::{EventQuery, DEFAULT_MAX_PAGE_LIMIT}; pub use leader_lock::SessionLeader; +pub use live_authorization::{ + LiveAuthorizationState, LiveAuthorizationTarget, LIVE_AUTHORIZATION_QUERY_TIMEOUT, +}; 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 00000000000..bbe0734b6db --- /dev/null +++ b/crates/buzz-db/src/store/live_authorization.rs @@ -0,0 +1,657 @@ +//! Bounded, tenant-scoped authorization snapshots for already-open sessions. + +use buzz_core::CommunityId; +use buzz_datastore_tracing::datastore_span; +use sqlx::{Connection as _, Row as _}; +use uuid::Uuid; + +use crate::{observability, Db, DbError, Result}; + +/// End-to-end deadline shared with the live-session revalidator. +pub const LIVE_AUTHORIZATION_QUERY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(1); +const STATEMENT_TIMEOUT: std::time::Duration = + LIVE_AUTHORIZATION_QUERY_TIMEOUT.saturating_sub(std::time::Duration::from_millis(100)); + +// A cancelled SQLx query must not wait for its unfinished statement in the +// pool's return-to-pool ping. SQLx bounds close_on_drop; PostgreSQL's own +// deadline below also terminates work if it has not noticed the disconnect. +struct AuthorizationConnection { + connection: sqlx::pool::PoolConnection, + reusable: bool, +} + +impl Drop for AuthorizationConnection { + fn drop(&mut self) { + if !self.reusable { + self.connection.close_on_drop(); + } + } +} + +async fn fetch_authorization_rows( + pool: &sqlx::PgPool, + query: sqlx::query::Query<'_, sqlx::Postgres, sqlx::postgres::PgArguments>, +) -> Result> { + let deadline = tokio::time::Instant::now() + LIVE_AUTHORIZATION_QUERY_TIMEOUT; + let mut guard = AuthorizationConnection { + connection: observability::acquire_writer_until( + pool, + observability::WriterOperation::Authorization, + deadline, + ) + .await?, + reusable: false, + }; + // BEGIN and tighten travel in one message. Keep a stricter operator cap; + // rollback restores the writer's session setting on successful reuse. + let begin = format!( + "BEGIN; SELECT set_config('statement_timeout', CASE \ + WHEN current_setting('statement_timeout')::interval = interval '0' \ + OR current_setting('statement_timeout')::interval > interval '{0}ms' \ + THEN '{0}ms' ELSE current_setting('statement_timeout') END, true)", + STATEMENT_TIMEOUT.as_millis(), + ); + // The only interpolation is the integer millisecond value of the + // compile-time deadline above; no request data enters this SQL. + let mut tx = guard + .connection + .begin_with(sqlx::AssertSqlSafe(begin)) + .await?; + let rows = query.fetch_all(&mut *tx).await?; + tx.rollback().await?; + guard.reusable = true; + Ok(rows) +} + +/// 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 rows = fetch_authorization_rows(pool, 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) + ).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() + } + + async fn wait_for_batch_activity(monitor: &PgPool, application: &str, active: bool) { + tokio::time::timeout(std::time::Duration::from_millis(1500), async { + loop { + let running: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_stat_activity \ + WHERE application_name = $1 AND state = 'active' AND wait_event_type = 'Lock')", + ) + .bind(application) + .fetch_one(monitor) + .await + .expect("observe batch backend"); + if running == active { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("batch backend reaches expected activity within deadline"); + } + + async fn blocked_batch_fixture() -> ( + Db, + PgPool, + sqlx::Transaction<'static, sqlx::Postgres>, + String, + ) { + use sqlx::postgres::{PgConnectOptions, PgPoolOptions}; + let monitor = setup_db().await; + let application = format!("live-auth-deadline-{}", Uuid::new_v4()); + let options: PgConnectOptions = crate::test_support::database_url().parse().unwrap(); + let pool = PgPoolOptions::new() + .max_connections(1) + .after_connect(|conn, _| { + Box::pin(async move { + sqlx::raw_sql("SET statement_timeout = '120s'; SET lock_timeout = '0'") + .execute(conn) + .await?; + Ok(()) + }) + }) + .connect_with(options.application_name(&application)) + .await + .unwrap(); + let mut lock = monitor.pool.begin().await.unwrap(); + sqlx::query("LOCK TABLE users IN ACCESS EXCLUSIVE MODE") + .execute(&mut *lock) + .await + .unwrap(); + (Db::from_pool(pool), monitor.pool, lock, application) + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn batch_deadline_preserves_tighter_cap_and_restores_writer_session() { + let (db, _monitor, lock, _application) = blocked_batch_fixture().await; + lock.rollback().await.unwrap(); + for (configured, effective, restored) in + [("120s", "900ms", "2min"), ("200ms", "200ms", "200ms")] + { + let mut conn = db.pool.acquire().await.unwrap(); + sqlx::query("SELECT set_config('statement_timeout', $1, false)") + .bind(configured) + .execute(&mut *conn) + .await + .unwrap(); + let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *conn) + .await + .unwrap(); + drop(conn); + let rows = fetch_authorization_rows( + &db.pool, + sqlx::query("SELECT current_setting('statement_timeout') AS cap"), + ) + .await + .unwrap(); + assert_eq!( + rows[0].get::("cap"), + effective, + "preserve tighter cap inside read" + ); + db.live_authorization_state_batch(&[LiveAuthorizationTarget { + community_id: CommunityId::from_uuid(Uuid::new_v4()), + pubkey: pubkey(), + session_owner_pubkey: None, + }]) + .await + .unwrap(); + let mut conn = db.pool.acquire().await.unwrap(); + let after: String = sqlx::query_scalar("SHOW statement_timeout") + .fetch_one(&mut *conn) + .await + .unwrap(); + let after_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *conn) + .await + .unwrap(); + assert_eq!( + after_pid, pid, + "successful batch reuses the same writer backend" + ); + assert_eq!( + after, restored, + "writer writes retain their session timeout" + ); + } + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn batch_statement_deadline_releases_writer_and_cancels_backend() { + let (db, monitor, lock, application) = blocked_batch_fixture().await; + let target = LiveAuthorizationTarget { + community_id: CommunityId::from_uuid(Uuid::new_v4()), + pubkey: pubkey(), + session_owner_pubkey: None, + }; + let task_db = db.clone(); + let task = tokio::spawn(async move { + tokio::time::timeout( + std::time::Duration::from_millis(1500), + task_db.live_authorization_state_batch(&[target]), + ) + .await + }); + wait_for_batch_activity(&monitor, &application, true).await; + let error = task + .await + .unwrap() + .expect("server cancels before client deadline") + .expect_err("locked batch must fail"); + let DbError::Sqlx(sqlx::Error::Database(error)) = error else { + panic!("expected PostgreSQL statement cancellation, got {error:?}"); + }; + assert_eq!(error.code().as_deref(), Some("57014")); + let _available = + tokio::time::timeout(std::time::Duration::from_millis(1500), db.pool.acquire()) + .await + .expect("writer slot released promptly") + .unwrap(); + wait_for_batch_activity(&monitor, &application, false).await; + lock.rollback().await.unwrap(); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn cancelled_batch_closes_connection_and_stops_backend_within_deadline() { + let (db, monitor, lock, application) = blocked_batch_fixture().await; + let target = LiveAuthorizationTarget { + community_id: CommunityId::from_uuid(Uuid::new_v4()), + pubkey: pubkey(), + session_owner_pubkey: None, + }; + let task_db = db.clone(); + let task = + tokio::spawn(async move { task_db.live_authorization_state_batch(&[target]).await }); + wait_for_batch_activity(&monitor, &application, true).await; + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + let _available = + tokio::time::timeout(std::time::Duration::from_millis(1500), db.pool.acquire()) + .await + .expect("cancelled checkout does not strand writer slot") + .unwrap(); + wait_for_batch_activity(&monitor, &application, false).await; + lock.rollback().await.unwrap(); + } + + #[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 relay_keypair = Keys::generate(); + 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"); + let _membership = db + .add_relay_member( + community_a, + &hex::encode(owner), + "member", + None, + &relay_keypair, + ) + .await + .expect("add owner to community A"); + let _membership = db + .add_relay_member( + community_a, + &hex::encode(current_owner), + "member", + None, + &relay_keypair, + ) + .await + .expect("add current owner to community A"); + let _membership = db + .add_relay_member( + community_b, + &hex::encode(direct), + "member", + None, + &relay_keypair, + ) + .await + .expect("add principal to community B"); + let _membership = db + .add_relay_member( + community_b, + &hex::encode(current_owner), + "member", + None, + &relay_keypair, + ) + .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 d1cb647694b..3f2e86ff6a8 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -47,6 +47,8 @@ pub mod feed; pub mod git_repo; /// Deployment-wide leader election over session advisory locks. pub mod leader_lock; +/// Batched authoritative access-state reads for live connections. +pub mod live_authorization; /// Compact viewer membership projections. pub mod membership_snapshot; /// Community moderation: reports, bans/timeouts, audit actions. diff --git a/crates/buzz-relay/src/audio/handler.rs b/crates/buzz-relay/src/audio/handler.rs index b3b09cdb414..ff8f2c8ffdf 100644 --- a/crates/buzz-relay/src/audio/handler.rs +++ b/crates/buzz-relay/src/audio/handler.rs @@ -1111,7 +1111,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 @@ -1121,27 +1121,46 @@ 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) => { + crate::handlers::auth::record_enforce_refusal( + &control, + Some(&denial), + nip_fi_assertion.as_ref(), + ); + auth_line.finish(AuthResult::from_outcome(denial.outcome)); + 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 { crate::handlers::auth::record_enforce_refusal( &control, Some(&denial), @@ -1165,6 +1184,16 @@ async fn run_active_audio_connection( check_cancel!(cancel, terminal_ctrl_rx, disconnect_reason, ws_send, cleanup: auth_line.abort_reason(abort_cause())); + control.mark_live_authorized( + pubkey.to_bytes(), + admitted_owner, + final_check.owner_was_relay_member_at_admission, + final_check.membership_via_owner, + token_owner.is_some(), + crate::nip_fi_session::NipFiWsRoute::Audio, + final_check_started_at, + ); + // Every auth and admission check passed; what follows is room joining. auth_line.finish(AuthResult::Success); @@ -4329,7 +4358,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 { @@ -15953,6 +15982,37 @@ mod tests { server.abort(); } + /// Binds durable revocation to the real audio AUTH enrollment call. + #[tokio::test] + #[ignore = "requires Postgres — runs in postgres-ci nextest lane"] + async fn durable_ban_revalidates_real_audio_admission_without_delivery() { + let state = audio_test_state_real_db() + .await + .expect("isolated PostgreSQL"); + let (tenant, channel_id, member) = seed_audio_fixture(state.db.pool()).await; + let (mut client, server) = + open_admitted_audio_socket(&state, tenant.clone(), channel_id, &member, None).await; + state + .db + .ban_community_member( + tenant.community(), + member.public_key().as_bytes(), + member.public_key().as_bytes(), + None, + None, + ) + .await + .expect("commit ban only"); + assert_eq!( + state.revalidate_live_authorizations().await, + 1, + "real audio admission must enroll the socket in durable scans" + ); + expect_policy_close(&mut client).await; + server.abort(); + let _ = server.await; + } + /// Every frame up to and including the server's close. async fn audio_frames_until_close( client: &mut tokio_tungstenite::WebSocketStream< diff --git a/crates/buzz-relay/src/handlers/auth.rs b/crates/buzz-relay/src/handlers/auth.rs index ecbc7d9e0d0..886a95925b4 100644 --- a/crates/buzz-relay/src/handlers/auth.rs +++ b/crates/buzz-relay/src/handlers/auth.rs @@ -57,6 +57,40 @@ 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, + owner_reason: None, + }), + 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, + owner_reason: None, + }), + 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. /// @@ -304,6 +338,28 @@ 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, +} + +fn admission_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", + } +} + /// Final ban and relay-membership verdict for an authenticating socket. /// /// Callers bind the socket to `pubkey` first (the registry a ban's or @@ -311,15 +367,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_verdict( state, community, @@ -331,9 +388,13 @@ pub(crate) async fn final_admission_denial( ) .await; if let Some(denial) = ban_verdict_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(), @@ -341,27 +402,55 @@ 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, - owner_reason: None, - }), - 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, - owner_reason: None, - }) - } + .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(error) => { + warn!( + community = %community, + pubkey = %pubkey.to_hex(), + error_class = admission_db_error_class(&error), + "captured-owner membership writer check 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, + owner_reason: None, + }); + 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, + owner_reason: None, + }); + return final_check; + } + }; + final_check } /// Owner to record on an admitted socket: the proven NIP-OA owner, else the @@ -889,7 +978,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 @@ -913,30 +1002,30 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: .as_ref() .map(crate::nip_fi_core::NipFiSession::from_assertion), ); - // An admitted owner (NIP-OA, a NIP-FI agent token, or a stored - // `agent_owner_pubkey`) makes this socket an agent's. - let admitted_agent = + let admitted_owner = match admitted_owner(&state, conn.tenant.community(), pubkey, nip_oa_owner).await { - Ok(Some(owner)) => { - state.conn_manager.set_admitted_owner(conn_id, owner); - true - } - Ok(None) => false, + Ok(owner) => owner, Err(denial) => { deny_admission(&conn, &attempt, &event_id_hex, denial); return; } }; - let denial = match final_admission_denial( + if let Some(owner) = admitted_owner { + state.conn_manager.set_admitted_owner(conn_id, owner); + } + let admitted_agent = admitted_owner.is_some(); + 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", @@ -1001,6 +1090,15 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: } // Only the frame that won `authenticate` reaches here, once. attempt.log(AuthResult::Success); + conn.community_control.mark_live_authorized( + pubkey.to_bytes(), + admitted_owner, + final_check.owner_was_relay_member_at_admission, + final_check.membership_via_owner, + token_owner.is_some(), + crate::nip_fi_session::NipFiWsRoute::Root, + final_check_started_at, + ); state.conn_manager.mark_admitted(conn_id); if let Some(session) = conn.community_control.nip_fi_enforce() { session.admit(pubkey); @@ -1068,7 +1166,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}; @@ -1278,7 +1376,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. @@ -2592,7 +2690,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"] @@ -2689,7 +2787,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"] @@ -3287,6 +3385,70 @@ mod tests { } } + /// Binds the durable scan to real AUTH enrollment, with Redis unavailable. + #[tokio::test] + #[ignore = "requires Postgres — runs in postgres-ci nextest lane"] + async fn durable_ban_revalidates_real_root_admission_without_delivery() { + use std::sync::Arc; + let state = auth_test_state_real_db_expect().await; + let community = seeded_community(&state).await; + let (member, owner, agent, bystander) = ( + Keys::generate(), + Keys::generate(), + Keys::generate(), + Keys::generate(), + ); + let tag = buzz_sdk::nip_oa::compute_auth_tag(&owner, &agent.public_key(), "") + .expect("owner proof"); + let tag = serde_json::from_str(&tag).expect("owner tag JSON"); + let mut connections = Vec::new(); + let mut lifecycle_guards = Vec::new(); + for (keys, tag, challenge) in [ + (&member, None, "durable-member"), + (&agent, Some(tag), "durable-agent"), + (&bystander, None, "durable-bystander"), + ] { + let (conn, _control) = registered_pending_conn(&state, community, challenge); + lifecycle_guards.push(state.community_connections.register( + conn.conn_id, + community, + conn.community_control.clone(), + )); + handle_auth( + signed_auth(keys, challenge, tag), + Arc::clone(&conn), + Arc::clone(&state), + ) + .await; + assert!(matches!( + conn.auth_state_snapshot(), + AuthState::Authenticated(_) + )); + connections.push(conn); + } + for key in [&member, &owner] { + state + .db + .ban_community_member( + community, + key.public_key().as_bytes(), + bystander.public_key().as_bytes(), + None, + None, + ) + .await + .expect("commit ban only"); + } + assert!(connections.iter().all(|conn| !conn.cancel.is_cancelled())); + assert_eq!(state.revalidate_live_authorizations().await, 2); + assert!(connections[0].cancel.is_cancelled(), "direct member closes"); + assert!( + connections[1].cancel.is_cancelled(), + "owner-linked agent closes" + ); + assert!(!connections[2].cancel.is_cancelled(), "bystander stays"); + } + /// AUTH with NIP-OA records a previously ownerless agent's owner. That /// closes the agent's earlier ownerless socket so it reconnects with /// the owner attached, but not the socket being admitted, even when @@ -4797,6 +4959,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 @@ -4888,7 +5067,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 532207a3dbc..be22bd396eb 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1357,6 +1357,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. { @@ -1962,6 +1972,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 9731c20d920..e782d8cc750 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; @@ -53,6 +55,8 @@ pub(crate) enum CommunityDisconnectReason { /// Root-socket ban or roster removal. Its `OK false` frame carries the /// reason, so the close stays bare. Removed, + /// The relay could not re-establish the socket's access decision. + AuthorizationUnavailable, } impl CommunityDisconnectReason { @@ -63,6 +67,7 @@ impl CommunityDisconnectReason { Self::CommunityDeleted => "community_deleted", Self::AuthPolicyUpgraded => "auth_policy_upgraded", Self::AuthorizationDenied => "authorization_denied", + Self::AuthorizationUnavailable => "authorization_unavailable", Self::AccessRevoked => "access_revoked", Self::Removed => "removed", } @@ -86,6 +91,12 @@ impl CommunityDisconnectReason { code: axum::extract::ws::close_code::POLICY, reason: WsUtf8Bytes::from_static("authorization denied"), })), + Self::AuthorizationUnavailable => { + WsMessage::Close(Some(axum::extract::ws::CloseFrame { + code: axum::extract::ws::close_code::POLICY, + reason: WsUtf8Bytes::from_static("authorization unavailable"), + })) + } Self::AccessRevoked => WsMessage::Close(Some(axum::extract::ws::CloseFrame { code: axum::extract::ws::close_code::POLICY, reason: WsUtf8Bytes::from_static("access revoked"), @@ -95,6 +106,51 @@ 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(5); +/// Spread outage-driven closes across scans without extending the 30s budget. +const LIVE_AUTHORIZATION_OUTAGE_JITTER: std::time::Duration = std::time::Duration::from_secs(9); +/// 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 = + buzz_db::LIVE_AUTHORIZATION_QUERY_TIMEOUT; +const LIVE_AUTHORIZATION_SCAN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); +// Conservative schedule: allow grace and jitter 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_OUTAGE_JITTER.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() +); +fn live_authorization_failure_grace(connection_id: Uuid) -> std::time::Duration { + // Connection IDs are server-generated random UUIDs. A stable offset means + // repeated failures cannot move a session's cutoff or add timers per socket. + let offset_ms = (connection_id.as_u128() % LIVE_AUTHORIZATION_OUTAGE_JITTER.as_millis()) as u64; + LIVE_AUTHORIZATION_DB_FAILURE_GRACE + std::time::Duration::from_millis(offset_ms) +} + +#[derive(Clone)] +struct LiveAuthorizationRecord { + pubkey: [u8; 32], + owner_pubkey: Option<[u8; 32]>, + owner_was_relay_member_at_admission: bool, + membership_via_owner: bool, + membership_via_token_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 { @@ -126,6 +182,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. @@ -146,6 +204,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(), nip_fi_enforce: Arc::default(), @@ -232,6 +291,53 @@ impl CommunityConnectionControl { let _ = self.owner.set(owner); } + /// Mark the socket admitted after its final authoritative policy check. + #[allow(clippy::too_many_arguments)] // Captures the distinct admission policy facts. + 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, + membership_via_token_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, + membership_via_token_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() { @@ -288,14 +394,40 @@ impl CommunityConnectionControl { route: crate::nip_fi_session::NipFiWsRoute, frame_tx: Option<&mpsc::Sender>, enforce: fn(&crate::nip_fi_shadow::EnforceSession), + ) { + self.publish_nip_fi_denial( + route, + buzz_auth::DenialClass::AuthorizationDenied, + frame_tx, + enforce, + ); + } + + fn publish_nip_fi_denial( + &self, + route: crate::nip_fi_session::NipFiWsRoute, + class: buzz_auth::DenialClass, + frame_tx: Option<&mpsc::Sender>, + enforce: fn(&crate::nip_fi_shadow::EnforceSession), ) { 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 | Some(CommunityDisconnectReason::Removed) => { - *current = Some(CommunityDisconnectReason::AuthorizationDenied); + None => { + *current = Some(reason); + true + } + Some(CommunityDisconnectReason::Removed) + if class == buzz_auth::DenialClass::AuthorizationDenied => + { + *current = Some(reason); true } Some(_) => false, @@ -309,10 +441,7 @@ impl CommunityConnectionControl { } } 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)); } } @@ -346,10 +475,24 @@ 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, + crate::nip_fi_shadow::EnforceSession::revoke, + ); + } + + fn manager_disconnect_nip_fi_with_class( + &self, + frame_tx: &mpsc::Sender, + class: buzz_auth::DenialClass, + enforce: fn(&crate::nip_fi_shadow::EnforceSession), + ) { + self.publish_nip_fi_denial( crate::nip_fi_session::NipFiWsRoute::Root, + class, Some(frame_tx), - crate::nip_fi_shadow::EnforceSession::revoke, + enforce, ); self.cancel.cancel(); } @@ -398,27 +541,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. @@ -603,6 +758,67 @@ 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, + membership_via_token_owner: record.membership_via_token_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. @@ -822,6 +1038,347 @@ 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| { + // A stale session that was omitted by a deadline keeps its previous + // successful check time. Use the oldest socket for each target so a + // newer sibling cannot keep that authorization behind the same tail. + // Identity fields make equal-age targets deterministic. + let oldest_check = sessions_by_target + .get(target) + .and_then(|sessions| sessions.iter().map(|session| session.last_checked_at).min()) + .unwrap_or(started_at); + ( + oldest_check, + 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_failure_grace(session.connection_id) + { + 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 + && (session.membership_via_token_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, @@ -1094,6 +1651,32 @@ 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**. @@ -1444,7 +2027,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, @@ -2680,6 +3263,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 @@ -3210,6 +3825,74 @@ pub(crate) mod tests { ); } + /// Live policy failures close root sockets without attributing them to + /// the NIP-FI deny set, just as the audio registry does. + #[test] + fn live_revalidation_does_not_record_deny_set_revocation() { + for class in [ + buzz_auth::DenialClass::AuthorizationDenied, + buzz_auth::DenialClass::AuthorizationUnavailable, + ] { + let (records, lines, _, ()) = + crate::nip_fi_shadow::tests::enforce_observed(async {}, |()| async move { + let manager = ConnectionManager::new(); + let community = buzz_core::CommunityId::from_uuid(Uuid::new_v4()); + let conn_id = Uuid::new_v4(); + let key = nostr::Keys::generate().public_key(); + let (tx, _rx) = mpsc::channel(8); + let (ctrl_tx, _ctrl_rx) = mpsc::channel(8); + let (terminal_tx, mut terminal_rx) = mpsc::channel(1); + let cancel = CancellationToken::new(); + let control = CommunityConnectionControl::new(cancel.clone()); + let reason = control.disconnect_reason(); + control.attach_nip_fi_enforce(crate::nip_fi_shadow::EnforceSession::new( + "ws", + crate::nip_fi_shadow::BOUND.into(), + Some(key), + )); + control.nip_fi_enforce().expect("attached").admit(key); + manager.register( + conn_id, + tx, + ctrl_tx, + terminal_tx, + None, + cancel.clone(), + community, + Arc::new(AtomicU8::new(0)), + Arc::new(Mutex::new(HashMap::new())), + 3, + control, + ); + manager.set_authenticated_identity( + conn_id, + key.to_bytes().to_vec(), + Some(NipFiSession::new("test-issuer", None)), + ); + assert!(manager.disconnect_live_authorization(community, conn_id, class)); + assert!(cancel.is_cancelled()); + assert_eq!( + *reason.borrow(), + Some( + if class == buzz_auth::DenialClass::AuthorizationUnavailable { + CommunityDisconnectReason::AuthorizationUnavailable + } else { + CommunityDisconnectReason::AuthorizationDenied + } + ), + ); + assert!(terminal_rx.try_recv().is_ok(), "terminal denial frame"); + }); + assert_eq!(records.iter().filter(|r| *r == "ws admit/admit").count(), 1); + assert!( + records + .iter() + .all(|r| !r.contains("end/") && !r.contains("deny/")), + "{class:?} must not be attributed to the deny set: {lines:?}" + ); + } + } + #[test] fn nip_fi_disconnect_audio_is_issuer_scoped() { let registry = CommunityConnectionRegistry::new(); @@ -5989,3 +6672,1656 @@ 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 { + control: CommunityConnectionControl, + 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(|| NipFiSession::new(TEST_ISSUER, owner_pubkey)), + ); + control.set_terminal_frame_sender(terminal_tx.clone()); + } + control.mark_live_authorized( + pubkey, + owner_pubkey, + owner_was_relay_member_at_admission, + membership_via_owner, + nip_fi && 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.clone(), + ); + conn_manager.set_authenticated_identity( + connection_id, + pubkey.to_vec(), + nip_fi.then(|| NipFiSession::new(TEST_ISSUER, owner_pubkey)), + ); + 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 { + control, + cancel, + terminal_rx, + reason_rx, + _guard: guard, + } + } + + fn register_audio_authorization_session_at( + registry: &CommunityConnectionRegistry, + community_id: CommunityId, + connection_id: Uuid, + pubkey: [u8; 32], + checked_at: TokioInstant, + ) -> (CommunityConnectionGuard, CancellationToken) { + let cancel = CancellationToken::new(); + let control = CommunityConnectionControl::new(cancel.clone()); + control.mark_live_authorized( + pubkey, + None, + false, + false, + false, + crate::nip_fi_session::NipFiWsRoute::Audio, + checked_at, + ); + let guard = registry.register(connection_id, community_id, control); + (guard, cancel) + } + + 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, + 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, + membership_via_token_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 outage_closes_are_spread_across_scans_within_revocation_budget() { + let registry = CommunityConnectionRegistry::new(); + let manager = ConnectionManager::new(); + let tenant = community(0x2722); + let checked_at = TokioInstant::now(); + // Same target and successful snapshot, but three different socket cutoffs. + let sessions = [0, 5000, 8999].map(|offset| { + register_audio_authorization_session_at( + ®istry, + tenant, + Uuid::from_u128(9000 + offset), + [0x49; 32], + checked_at, + ) + }); + let mut closes = Vec::new(); + for _ in 0..3 { + tokio::time::advance(LIVE_AUTHORIZATION_REVALIDATION_INTERVAL).await; + let report = revalidate_live_authorization_sessions( + ®istry, + &manager, + true, + true, + &CancellationToken::new(), + database_failure_lookup(), + ) + .await; + closes.push(report.closed_for_db_failure); + } + assert_eq!( + closes, + [1, 1, 1], + "outage does not close all sibling sockets in one sweep" + ); + assert!(sessions.iter().all(|(_, cancel)| cancel.is_cancelled())); + assert!(checked_at.elapsed() <= LIVE_AUTHORIZATION_MAX_STALENESS); + assert!( + LIVE_AUTHORIZATION_WORST_CASE_STALENESS_SECS + <= LIVE_AUTHORIZATION_MAX_STALENESS.as_secs() + ); + } + + #[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 + LIVE_AUTHORIZATION_OUTAGE_JITTER, + ) + .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(start_paused = true)] + async fn scan_timeout_keeps_stale_sessions_oldest_for_the_next_bounded_scan() { + const PEERS: usize = + LIVE_AUTHORIZATION_BATCH_SIZE * LIVE_AUTHORIZATION_QUERY_CONCURRENCY * 5; + const TARGETS: usize = PEERS + 1; + + let registry = CommunityConnectionRegistry::new(); + let manager = ConnectionManager::new(); + let tenant = community(0x2721); + let peers_checked_at = TokioInstant::now(); + let mut guards = Vec::with_capacity(TARGETS + 1); + for index in 0..PEERS { + let mut pubkey = [0; 32]; + pubkey[..8].copy_from_slice(&(index as u64).to_be_bytes()); + let (guard, _) = register_audio_authorization_session_at( + ®istry, + tenant, + Uuid::from_u128(index as u128 + 1), + pubkey, + peers_checked_at, + ); + guards.push(guard); + } + + tokio::time::advance(std::time::Duration::from_millis(500)).await; + let tail_pubkey = [0xff; 32]; + let tail_checked_at = TokioInstant::now(); + let tail_target = buzz_db::LiveAuthorizationTarget { + community_id: tenant, + pubkey: tail_pubkey, + session_owner_pubkey: None, + }; + let (tail_old_guard, tail_old_cancel) = register_audio_authorization_session_at( + ®istry, + tenant, + Uuid::from_u128(PEERS as u128 + 1), + tail_pubkey, + tail_checked_at, + ); + guards.push(tail_old_guard); + + tokio::time::advance(std::time::Duration::from_millis(500)).await; + let tail_new_checked_at = TokioInstant::now(); + let (tail_new_guard, tail_new_cancel) = register_audio_authorization_session_at( + ®istry, + tenant, + Uuid::from_u128(PEERS as u128 + 2), + tail_pubkey, + tail_new_checked_at, + ); + guards.push(tail_new_guard); + + let first_batch = Arc::new(std::sync::Mutex::new(None)); + let delayed_success: LiveAuthorizationBatchLookup = Arc::new({ + let first_batch = Arc::clone(&first_batch); + move |targets| { + let mut recorded = first_batch.lock().expect("first scan batch lock"); + if recorded.is_none() { + *recorded = Some(targets.clone()); + } + drop(recorded); + Box::pin(async move { + // Five waves finish just under the shared deadline; the + // final tail batch is still running when ScanTimeout fires. + tokio::time::sleep(std::time::Duration::from_millis(990)).await; + Ok(targets + .into_iter() + .map(|target| access_state(target, None, false, false, false)) + .collect()) + }) + } + }); + + let first_report = revalidate_live_authorization_sessions( + ®istry, + &manager, + false, + false, + &CancellationToken::new(), + delayed_success, + ) + .await; + + assert!(first_report.scan_timed_out); + assert_eq!(first_report.checked_targets, PEERS); + assert_eq!(first_report.failed_targets, 1); + assert_eq!(first_report.closed_sessions, 0); + assert!(!tail_old_cancel.is_cancelled()); + assert!(!tail_new_cancel.is_cancelled()); + let first_batch = first_batch + .lock() + .expect("first scan batch lock") + .clone() + .expect("scan should start at least one batch"); + assert_eq!(first_batch.len(), LIVE_AUTHORIZATION_BATCH_SIZE); + assert_eq!(first_batch[0].pubkey, [0; 32]); + assert_eq!(first_batch[1].pubkey[..8], 1_u64.to_be_bytes()); + let unrefreshed = registry + .live_authorization_sessions() + .into_iter() + .filter(|session| session.pubkey == tail_pubkey) + .collect::>(); + assert_eq!(unrefreshed.len(), 2); + let unrefreshed_checks = unrefreshed + .iter() + .map(|session| session.last_checked_at) + .collect::>(); + assert!(unrefreshed_checks.contains(&tail_checked_at)); + assert!(unrefreshed_checks.contains(&tail_new_checked_at)); + + let first_retry_batch = Arc::new(std::sync::Mutex::new(None)); + let retry_failure: LiveAuthorizationBatchLookup = Arc::new({ + let first_retry_batch = Arc::clone(&first_retry_batch); + move |targets| { + let mut recorded = first_retry_batch.lock().expect("retry scan batch lock"); + if recorded.is_none() { + *recorded = Some(targets.clone()); + } + drop(recorded); + Box::pin(async move { + std::future::pending::>>() + .await + }) + } + }); + let retry_report = revalidate_live_authorization_sessions( + ®istry, + &manager, + false, + false, + &CancellationToken::new(), + retry_failure, + ) + .await; + + assert!(retry_report.scan_timed_out); + assert_eq!(retry_report.failed_targets, TARGETS); + let first_retry_batch = first_retry_batch + .lock() + .expect("retry scan batch lock") + .clone() + .expect("retry scan should start at least one batch"); + assert_eq!(first_retry_batch[0], tail_target); + assert!(tail_old_cancel.is_cancelled()); + assert!(tail_new_cancel.is_cancelled()); + drop(guards); + } + + #[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( + LIVE_AUTHORIZATION_DB_FAILURE_GRACE - std::time::Duration::from_secs(1), + ) + .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(LIVE_AUTHORIZATION_OUTAGE_JITTER + 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())); + + // The same terminal chosen by revalidation must reach access logging, + // rather than falling back to the generic cancellation reason. + let state = crate::state::tests::test_state().await; + let (mut conn, _rx) = crate::connection::tests::test_conn_with_auth( + crate::connection::tests::authenticated_state(), + ); + Arc::get_mut(&mut conn) + .expect("sole fixture owner") + .community_control = sessions[0].control.clone(); + assert_eq!( + crate::connection::server_close_reason(&state, &conn), + Some("authorization_unavailable") + ); + + 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_OUTAGE_JITTER + - 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, + 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::*; + + /// A token owner's membership is independent of the NIP-OA feature flag. + /// Drive the production final admission check and scan with the same facts. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn token_owner_admission_survives_scan_with_nip_oa_disabled_until_owner_removal() { + let pool = sqlx::PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect to isolated database"); + let mut state = + (*crate::state::tests::test_state_with_database_pool(pool).await).clone(); + let mut config = (*state.config).clone(); + config.require_relay_membership = true; + config.allow_nip_oa_auth = false; + state.config = Arc::new(config); + let tenant = state + .db + .ensure_configured_community(&format!( + "token-owner-revalidation-{}.example", + Uuid::new_v4() + )) + .await + .expect("create community") + .id; + let owner = nostr::Keys::generate(); + let agent = nostr::Keys::generate(); + let owner_bytes = owner.public_key().to_bytes(); + let agent_bytes = agent.public_key().to_bytes(); + for key in [owner_bytes, agent_bytes] { + state + .db + .ensure_user_for_authorization(tenant, &key) + .await + .expect("create user"); + } + state + .db + .set_agent_owner_for_authorization(tenant, &agent_bytes, &owner_bytes) + .await + .expect("record verified owner"); + let write = state + .db + .add_relay_member( + tenant, + &owner.public_key().to_hex(), + "member", + None, + &state.relay_keypair, + ) + .await + .expect("add owner"); + assert!(write.outcome); + + let admitted = crate::handlers::auth::final_admission_check( + &state, + tenant, + agent.public_key(), + None, + None, + Some(owner.public_key()), + Some(owner_bytes), + ) + .await; + assert!(admitted.denial.is_none()); + assert!(admitted.membership_via_owner); + let without_token = crate::handlers::auth::final_admission_check( + &state, + tenant, + agent.public_key(), + None, + None, + None, + Some(owner_bytes), + ) + .await; + assert!(without_token.denial.is_some(), "NIP-OA remains disabled"); + let sessions = [ + crate::nip_fi_session::NipFiWsRoute::Root, + crate::nip_fi_session::NipFiWsRoute::Audio, + ] + .map(|route| { + register_live_connection_with_owner_membership( + &state.community_connections, + &state.conn_manager, + tenant, + agent_bytes, + Some(owner_bytes), + admitted.owner_was_relay_member_at_admission, + admitted.membership_via_owner, + route, + true, + ) + }); + assert_eq!(state.revalidate_live_authorizations().await, 0); + assert!(sessions + .iter() + .all(|session| !session.cancel.is_cancelled())); + + let removed = state + .db + .remove_relay_member(tenant, &owner.public_key().to_hex(), &state.relay_keypair) + .await + .expect("remove owner without Redis delivery"); + assert_eq!( + removed.outcome, + buzz_db::relay_members::RemoveResult::Removed + ); + assert_eq!(state.revalidate_live_authorizations().await, 2); + assert!(sessions.iter().all(|session| session.cancel.is_cancelled())); + } + + #[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 relay_keypair = nostr::Keys::generate(); + 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, &relay_keypair) + .await + .expect("add disposable roster member") + .outcome + ); + } + 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), &relay_keypair) + .await + .expect("commit direct member removal") + .outcome, + buzz_db::relay_members::RemoveResult::Removed + ); + assert_eq!( + db.remove_relay_member(tenant, &hex::encode(removed_owner), &relay_keypair) + .await + .expect("commit captured owner removal") + .outcome, + 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 owner delegation. + pub membership_via_owner: bool, + /// Whether delegation came from a verified NIP-FI token rather than NIP-OA. + pub membership_via_token_owner: bool, + /// Most recent successful authoritative access check. + pub last_checked_at: TokioInstant, +} diff --git a/crates/buzz-test-client/tests/live_revalidation.rs b/crates/buzz-test-client/tests/live_revalidation.rs new file mode 100644 index 00000000000..a018f1b9afd --- /dev/null +++ b/crates/buzz-test-client/tests/live_revalidation.rs @@ -0,0 +1,212 @@ +//! Local-only reproduction of durable revalidation against a running relay. +//! Requires a disposable database, NIP-FI off, relay membership enabled, and +//! NIP-OA enabled. It writes durable bans directly, without Redis publication. +//! Run with DATABASE_URL and RELAY_URL pointing to that isolated stack. +mod external_infra_live { + use futures_util::{SinkExt, StreamExt}; + use nostr::{EventBuilder, Keys, Kind, Tag}; + use serde_json::{json, Value}; + use sqlx::PgPool; + use std::time::{Duration, Instant}; + use tokio::net::TcpStream; + use tokio_tungstenite::{ + connect_async, + tungstenite::{client::IntoClientRequest, Message}, + MaybeTlsStream, WebSocketStream, + }; + use uuid::Uuid; + + type Socket = WebSocketStream>; + + async fn frame(socket: &mut Socket) -> Message { + tokio::time::timeout(Duration::from_secs(30), socket.next()) + .await + .expect("bounded relay frame") + .expect("socket remains open") + .expect("frame") + } + + async fn open( + base: &str, + host: &str, + channel: Option, + key: &Keys, + owner: Option<&Keys>, + ) -> Socket { + let url = channel.map_or_else(|| base.to_owned(), |id| format!("{base}/huddle/{id}/audio")); + let mut request = url.into_client_request().unwrap(); + request.headers_mut().insert("host", host.parse().unwrap()); + let (mut socket, _) = connect_async(request).await.expect("real relay connection"); + let mut challenge = None; + for _ in 0..32 { + if let Message::Text(text) = frame(&mut socket).await { + let value: Value = serde_json::from_str(&text).unwrap(); + if channel.is_some() && value["type"] == "challenge" { + challenge = Some(value["challenge"].as_str().unwrap().to_owned()); + break; + } + if channel.is_none() && value[0] == "AUTH" { + challenge = Some(value[1].as_str().unwrap().to_owned()); + break; + } + } + } + let challenge = challenge.expect("bounded authentication challenge"); + let mut event = EventBuilder::new(Kind::Authentication, "") + .tag(Tag::parse(["relay", &format!("ws://{host}")]).unwrap()) + .tag(Tag::parse(["challenge", &challenge]).unwrap()); + if let Some(owner) = owner { + let tag = buzz_sdk::nip_oa::compute_auth_tag(owner, &key.public_key(), "").unwrap(); + let tag: Vec = serde_json::from_str(&tag).unwrap(); + event = event.tag(Tag::parse(tag).unwrap()); + } + let event = event.sign_with_keys(key).unwrap(); + let auth = if channel.is_some() { + json!({"type":"auth","event":event}) + } else { + json!(["AUTH", event]) + }; + socket + .send(Message::Text(auth.to_string().into())) + .await + .unwrap(); + for _ in 0..32 { + if let Message::Text(text) = frame(&mut socket).await { + let value: Value = serde_json::from_str(&text).unwrap(); + if channel.is_some() { + assert_ne!(value["type"], "error", "{value}"); + if value["type"] == "joined" { + return socket; + } + } else if value[0] == "OK" { + assert_eq!(value[2], true, "{value}"); + return socket; + } + } + } + panic!("no admission acknowledgement"); + } + + async fn closed(mut socket: Socket, since: Instant, label: &str) -> Duration { + for _ in 0..64 { + let message = frame(&mut socket).await; + if let Message::Close(close) = message { + let close = close.unwrap_or_else(|| panic!("{label}: close without policy reason")); + assert_eq!(u16::from(close.code), 1008, "{label}: {close:?}"); + let elapsed = since.elapsed(); + assert!(elapsed < Duration::from_secs(30), "{label}: {elapsed:?}"); + println!( + "{label}: policy close in {} ms ({})", + elapsed.as_millis(), + close.reason + ); + return elapsed; + } + } + panic!("{label}: no policy close"); + } + + #[tokio::test] + #[ignore = "requires isolated relay network and PostgreSQL"] + async fn external_infra_durable_ban_and_database_failure_close_real_root_and_audio() { + let db = PgPool::connect(&std::env::var("DATABASE_URL").expect("disposable DATABASE_URL")) + .await + .unwrap(); + let base = std::env::var("RELAY_URL").expect("isolated RELAY_URL"); + assert!(base.starts_with("ws://127.0.0.1:"), "local relay only"); + let community = Uuid::new_v4(); + let host = format!( + "revalidation-{}.test:{}", + community.simple(), + url::Url::parse(&base).unwrap().port().unwrap() + ); + let channel = Uuid::new_v4(); + let (member, owner, agent, bystander) = ( + Keys::generate(), + Keys::generate(), + Keys::generate(), + Keys::generate(), + ); + sqlx::query("INSERT INTO communities(id,host) VALUES($1,$2)") + .bind(community) + .bind(&host) + .execute(&db) + .await + .unwrap(); + sqlx::query("INSERT INTO channels(community_id,id,name,channel_type,visibility,created_by) VALUES($1,$2,'local-revalidation','stream','open',$3)") + .bind(community).bind(channel).bind(bystander.public_key().to_bytes().to_vec()).execute(&db).await.unwrap(); + for key in [&member, &owner, &agent, &bystander] { + sqlx::query("INSERT INTO users(community_id,pubkey) VALUES($1,$2)") + .bind(community) + .bind(key.public_key().to_bytes().to_vec()) + .execute(&db) + .await + .unwrap(); + // Admit the agent through its verified owner on both routes. + // Direct root membership bypasses first-use owner materialization; + // later audio materialization intentionally revokes that ownerless socket. + if key.public_key() != agent.public_key() { + sqlx::query( + "INSERT INTO relay_members(community_id,pubkey,role) VALUES($1,$2,'member')", + ) + .bind(community) + .bind(key.public_key().to_hex()) + .execute(&db) + .await + .unwrap(); + } + sqlx::query("INSERT INTO channel_members(community_id,channel_id,pubkey,role) VALUES($1,$2,$3,'member')").bind(community).bind(channel).bind(key.public_key().to_bytes().to_vec()).execute(&db).await.unwrap(); + } + let root_member = open(&base, &host, None, &member, None).await; + let root_agent = open(&base, &host, None, &agent, Some(&owner)).await; + let audio_member = open(&base, &host, Some(channel), &member, None).await; + let audio_agent = open(&base, &host, Some(channel), &agent, Some(&owner)).await; + let mut control = open(&base, &host, None, &bystander, None).await; + // Keep the huddle active while its restricted participants leave. + let audio_control = open(&base, &host, Some(channel), &bystander, None).await; + // No moderation event or pub/sub operation: only authoritative DB commits. + let since = Instant::now(); + for key in [&member, &owner] { + sqlx::query("INSERT INTO community_bans(community_id,pubkey,banned,actor_pubkey) VALUES($1,$2,true,$3)") + .bind(community).bind(key.public_key().to_bytes().to_vec()).bind(bystander.public_key().to_bytes().to_vec()).execute(&db).await.unwrap(); + } + tokio::join!( + closed(root_member, since, "root member"), + closed(root_agent, since, "root owner-linked agent"), + closed(audio_member, since, "audio member"), + closed(audio_agent, since, "audio owner-linked agent") + ); + control + .send(Message::Ping(vec![1, 2, 3].into())) + .await + .unwrap(); + let mut pong = false; + for _ in 0..32 { + if let Message::Pong(data) = frame(&mut control).await { + assert_eq!(data.as_ref(), [1, 2, 3]); + pong = true; + break; + } + } + assert!(pong, "clear bystander responds within the frame bound"); + println!("clear bystander remains responsive after durable bans"); + // This test owns a disposable database. Restore the schema even on failure. + sqlx::query("ALTER TABLE community_bans RENAME TO live_unavailable_bans") + .execute(&db) + .await + .unwrap(); + let since = Instant::now(); + let closes = tokio::spawn(async move { + tokio::join!( + closed(control, since, "root database failure"), + closed(audio_control, since, "audio database failure") + ) + }); + let outcome = closes.await; + sqlx::query("ALTER TABLE live_unavailable_bans RENAME TO community_bans") + .execute(&db) + .await + .unwrap(); + outcome.expect("database failures close within the bound"); + } +} diff --git a/desktop/src/shared/api/relayAuthPolicy.ts b/desktop/src/shared/api/relayAuthPolicy.ts index 6e100fe7b46..3f0796a678a 100644 --- a/desktop/src/shared/api/relayAuthPolicy.ts +++ b/desktop/src/shared/api/relayAuthPolicy.ts @@ -9,14 +9,15 @@ * a connection that is in fact authenticated. The session is usable; * treat it as authenticated. * - `auth-required: verification failed` — covers ±60s clock-skew rejects - * and the relay's fail-closed allowlist DB lookup errors, both of which - * can clear on retry. + * and invalid auth evidence, so it stays subject to the rejection cap. * * Only `restricted:` and `blocked:` rejections (not a relay member / banned) - * are known permanent. Everything else retries with normal backoff, but - * latches terminal after `MAX_CONSECUTIVE_AUTH_REJECTIONS` consecutive - * rejections so a genuinely broken identity (e.g. persistently wrong system - * clock) still surfaces the terminal error card instead of flapping forever. + * are known permanent. The relay's exact internal policy-read error replies + * remain unauthenticated and retry without spending the evidence-rejection + * budget. Other reasons retry with normal backoff, but latch terminal after + * `MAX_CONSECUTIVE_AUTH_REJECTIONS` consecutive rejections so a genuinely + * broken identity (e.g. persistently wrong system clock) still surfaces the + * terminal error card instead of flapping forever. * * The rejection streak is preserved across environment-driven resume * attempts (focus/online/visibility); only explicit user re-engagement — @@ -26,6 +27,17 @@ export type AuthOkDecision = "authenticated" | "retry" | "terminal"; export const MAX_CONSECUTIVE_AUTH_REJECTIONS = 3; +// Exact `OK false` messages emitted by the relay's AUTH policy checks when a +// writer-backed dependency is unavailable. Keep unknown errors capped, and do +// not classify the generic verification failure because it also means invalid +// auth evidence. +const SERVER_UNAVAILABLE_AUTH_REPLIES = new Set([ + "error: internal error checking restriction state", + "error: internal error checking allowlist", + "error: internal error checking relay membership", + "error: internal error recording agent owner", +]); + export type RelayAuthRequest = { pendingEventId: string; resolve: () => void; @@ -48,7 +60,7 @@ export function armRelayAuthentication( }); } -/** Tracks consecutive AUTH rejections across reconnect attempts. */ +/** Tracks consecutive evidence and unknown AUTH rejections across reconnect attempts. */ export class AuthOkTracker { private consecutiveRejections = 0; @@ -66,14 +78,18 @@ export class AuthOkTracker { return "authenticated"; } - this.consecutiveRejections++; - if ( normalized.startsWith("restricted:") || normalized.startsWith("blocked:") ) { return "terminal"; } + + if (SERVER_UNAVAILABLE_AUTH_REPLIES.has(normalized)) { + return "retry"; + } + + this.consecutiveRejections++; if (this.consecutiveRejections >= MAX_CONSECUTIVE_AUTH_REJECTIONS) { return "terminal"; } diff --git a/desktop/src/shared/api/relayClientAuthRecovery.test.mjs b/desktop/src/shared/api/relayClientAuthRecovery.test.mjs new file mode 100644 index 00000000000..0379f2b594f --- /dev/null +++ b/desktop/src/shared/api/relayClientAuthRecovery.test.mjs @@ -0,0 +1,259 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +let fakeNow = 0; +const pendingTimers = new Map(); +let nextTimerId = 1; + +globalThis.window = { + setTimeout: (fn, delayMs) => { + const id = nextTimerId++; + pendingTimers.set(id, { fn, fireAt: fakeNow + delayMs, delayMs }); + return id; + }, + clearTimeout: (id) => pendingTimers.delete(id), + __TAURI_INTERNALS__: { + invoke: async () => undefined, + }, +}; + +const originalRandom = Math.random; +Math.random = () => 0.999; + +const { RelayClient } = await import("./relayClientSession.ts"); +const { armRelayAuthentication } = await import("./relayAuthPolicy.ts"); +const { AUTH_TIMEOUT_MS, RECONNECT_BASE_DELAY_MS, RECONNECT_MAX_DELAY_MS } = + await import("./relayClientTimings.ts"); + +function resetHarness() { + fakeNow = 0; + pendingTimers.clear(); + nextTimerId = 1; + Math.random = () => 0.999; +} + +function connectedClient() { + const client = new RelayClient(); + client.wsId = 7; + client.connectionGeneration = 1; + client.keepAliveRequested = true; + client.hasConnectedOnce = true; + client.connectionStateEmitter.set("connected"); + return client; +} + +function armAuthRequest(client, attempt) { + const authentication = armRelayAuthentication( + AUTH_TIMEOUT_MS, + (request) => { + request.pendingEventId = `auth-${attempt}`; + client.authRequest = request; + }, + (error) => client.resetConnection(error), + ); + authentication.catch(() => {}); + return authentication; +} + +function installAutomaticReconnect(client) { + let attempts = 0; + client.ensureConnected = () => { + attempts++; + client.connectionGeneration++; + client.wsId = 7 + attempts; + const authentication = armAuthRequest(client, attempts); + return authentication.then(() => { + client.connectionStateEmitter.set("connected"); + return client.connectionGeneration; + }); + }; + return () => attempts; +} + +async function deliver( + client, + frame, + generation = client.connectionGeneration, +) { + await client.handleWsMessage( + { type: "Text", data: JSON.stringify(frame) }, + generation, + ); +} + +async function flushMicrotasks() { + for (let i = 0; i < 20; i++) await Promise.resolve(); +} + +async function fireReconnect(client) { + const timerId = client.reconnectTimeout; + assert.notEqual(timerId, null, "an automatic reconnect should be scheduled"); + const timer = pendingTimers.get(timerId); + assert.ok(timer, "the scheduled reconnect should have a timer"); + pendingTimers.delete(timerId); + fakeNow = Math.max(fakeNow, timer.fireAt); + timer.fn(); + await flushMicrotasks(); + assert.ok(client.authRequest, "the automatic attempt should reach AUTH"); +} + +test("server-unavailable AUTH replies keep retrying through capped backoff and recover", async () => { + resetHarness(); + const client = connectedClient(); + const automaticReconnects = installAutomaticReconnect(client); + const unavailableReplies = [ + "error: internal error checking restriction state", + "error: internal error checking allowlist", + "error: internal error checking relay membership", + "error: internal error recording agent owner", + ]; + const delays = []; + let expectedBase = RECONNECT_BASE_DELAY_MS; + armAuthRequest(client, 0); + + for (let attempt = 0; attempt < 8; attempt++) { + const generation = client.connectionGeneration; + const eventId = client.authRequest.pendingEventId; + await deliver(client, [ + "OK", + eventId, + false, + unavailableReplies[attempt % unavailableReplies.length], + ]); + + assert.equal(client.terminal, false); + assert.equal(client.getConnectionState(), "reconnecting"); + const timer = pendingTimers.get(client.reconnectTimeout); + assert.ok(timer, "a server-unavailable AUTH reply must schedule a retry"); + const expectedDelay = Math.min( + expectedBase * (0.75 + 0.999 * 0.5), + RECONNECT_MAX_DELAY_MS, + ); + assert.equal(timer.delayMs, expectedDelay); + assert.ok(timer.delayMs >= expectedBase * 0.75); + assert.ok(timer.delayMs <= RECONNECT_MAX_DELAY_MS); + delays.push(timer.delayMs); + expectedBase = Math.min(expectedBase * 2, RECONNECT_MAX_DELAY_MS); + + if (attempt === 0) { + const scheduledTimer = client.reconnectTimeout; + await deliver( + client, + ["OK", eventId, false, "blocked: you are banned from this community"], + generation, + ); + assert.equal(client.terminal, false, "stale AUTH frames are ignored"); + assert.equal(client.reconnectTimeout, scheduledTimer); + } + + await fireReconnect(client); + } + + assert.deepEqual(delays.slice(5), [30_000, 30_000, 30_000]); + assert.equal(automaticReconnects(), 8); + await deliver(client, ["OK", client.authRequest.pendingEventId, true, ""]); + await flushMicrotasks(); + + assert.equal(client.getConnectionState(), "connected"); + assert.equal(client.terminal, false); + assert.notEqual(client.wsId, null); + assert.equal(client.reconnectTimeout, null); +}); + +test("bans and nonmembership remain terminal after a transient writer failure", async () => { + for (const message of [ + "blocked: you are banned from this community", + "restricted: not a relay member", + ]) { + resetHarness(); + const client = connectedClient(); + installAutomaticReconnect(client); + armAuthRequest(client, 0); + + await deliver(client, [ + "OK", + client.authRequest.pendingEventId, + false, + "error: internal error checking relay membership", + ]); + await fireReconnect(client); + await deliver(client, [ + "OK", + client.authRequest.pendingEventId, + false, + message, + ]); + + assert.equal(client.terminal, true); + assert.equal(client.getConnectionState(), "disconnected"); + assert.equal(client.reconnectTimeout, null); + } +}); + +test("invalid and unknown AUTH evidence still hits the three-rejection cap", async () => { + for (const messages of [ + [ + "auth-required: verification failed", + "auth-required: verification failed", + "auth-required: verification failed", + ], + ["", "unrecognized AUTH failure", "error: internal"], + ]) { + resetHarness(); + const client = connectedClient(); + installAutomaticReconnect(client); + armAuthRequest(client, 0); + + for (let index = 0; index < messages.length; index++) { + await deliver(client, [ + "OK", + client.authRequest.pendingEventId, + false, + messages[index], + ]); + assert.equal(client.terminal, index === messages.length - 1); + if (index < messages.length - 1) await fireReconnect(client); + } + assert.equal(client.reconnectTimeout, null); + } +}); + +test("unavailable replies do not spend or erase an invalid-evidence streak", async () => { + resetHarness(); + const client = connectedClient(); + installAutomaticReconnect(client); + armAuthRequest(client, 0); + + for (let index = 0; index < 2; index++) { + await deliver(client, [ + "OK", + client.authRequest.pendingEventId, + false, + "auth-required: verification failed", + ]); + assert.equal(client.terminal, false); + await fireReconnect(client); + } + + await deliver(client, [ + "OK", + client.authRequest.pendingEventId, + false, + "error: internal error checking relay membership", + ]); + assert.equal(client.terminal, false); + await fireReconnect(client); + + await deliver(client, [ + "OK", + client.authRequest.pendingEventId, + false, + "unrecognized AUTH failure", + ]); + assert.equal(client.terminal, true); + assert.equal(client.reconnectTimeout, null); +}); + +test.after(() => { + Math.random = originalRandom; +}); diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index b4084e167c7..58d4602ba23 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -520,6 +520,7 @@ run_unit_tests() { state::tests::nip_fi_disconnect_does_not_close_unproven_audio_socket state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason state::tests::enforce_end_is_recorded_only_by_the_winning_terminal + state::tests::live_revalidation_does_not_record_deny_set_revocation state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame state::tests::manager_disconnect_sets_reason_enqueues_frame_then_cancels ) @@ -550,6 +551,9 @@ run_unit_tests() { run_test_step "buzz-relay mesh receive lifetime regression" \ cargo test -p buzz-relay --lib mesh_boot::tests::demo_echo_retains_pending_receive_across_housekeeping_ticks -- --exact --nocapture + run_test_step "buzz-relay durable live authorization contracts" \ + cargo test -p buzz-relay --lib state::live_authorization_tests:: -- --nocapture + run_test_step "buzz-relay boot lifecycle tests" \ cargo test -p buzz-relay --test boot_lifecycle -- --nocapture