From 6298645ed892e1444062e00d38ed0ea4cee77329 Mon Sep 17 00:00:00 2001 From: Cea Stapleton Cordasco <261786559+cea-block@users.noreply.github.com> Date: Tue, 4 Aug 2026 12:18:58 -0500 Subject: [PATCH] fix(auth): implement deny-protected fail-safe mode Signed-off-by: Cea Stapleton Cordasco <261786559+cea-block@users.noreply.github.com> --- crates/buzz-relay/src/api/invites.rs | 25 +- .../src/authorization_runtime/executor.rs | 1276 +++++++++++++++++ .../src/authorization_runtime/finalization.rs | 632 ++++++++ .../src/authorization_runtime/production.rs | 99 +- .../src/authorization_runtime/transport.rs | 26 +- crates/buzz-relay/src/connection.rs | 657 ++++++++- crates/buzz-relay/src/handlers/auth.rs | 22 + crates/buzz-relay/src/protected_surface.rs | 6 +- crates/buzz-workflow/src/lib.rs | 61 + crates/buzz-workflow/src/mutation_gate.rs | 77 + 10 files changed, 2808 insertions(+), 73 deletions(-) create mode 100644 crates/buzz-relay/src/authorization_runtime/executor.rs create mode 100644 crates/buzz-relay/src/authorization_runtime/finalization.rs create mode 100644 crates/buzz-workflow/src/mutation_gate.rs diff --git a/crates/buzz-relay/src/api/invites.rs b/crates/buzz-relay/src/api/invites.rs index b78c4b81a..24f012967 100644 --- a/crates/buzz-relay/src/api/invites.rs +++ b/crates/buzz-relay/src/api/invites.rs @@ -284,11 +284,16 @@ async fn authenticate( true, // POST bodies must be covered by a payload tag tenant.community(), )?; - if state + let authorization_mode = state .protected_transport() - .and_then(|runtime| runtime.mode_for_domain(tenant.community())) - != Some(AuthorizationMode::Enforce) - { + .and_then(|runtime| runtime.mode_for_domain(tenant.community())); + if authorization_mode == Some(AuthorizationMode::DenyProtected) { + return Err(api_error( + StatusCode::FORBIDDEN, + "protected authorization denied", + )); + } + if authorization_mode != Some(AuthorizationMode::Enforce) { bridge::check_nip98_replay(state, &tenant, event_id_bytes).await?; } @@ -556,10 +561,16 @@ pub async fn claim_invite( let (tenant, pubkey, identity_proof, verified_proof, enrollment_assertion) = authenticate(&state, &headers, "/api/invites/claim", &body).await?; - let enforcing = state + let authorization_mode = state .protected_transport() - .and_then(|runtime| runtime.mode_for_domain(tenant.community())) - == Some(AuthorizationMode::Enforce); + .and_then(|runtime| runtime.mode_for_domain(tenant.community())); + if authorization_mode == Some(AuthorizationMode::DenyProtected) { + return Err(api_error( + StatusCode::FORBIDDEN, + "protected authorization denied", + )); + } + let enforcing = authorization_mode == Some(AuthorizationMode::Enforce); if claim_rate_limited(&state, tenant.community(), &pubkey) { return Err(api_error( diff --git a/crates/buzz-relay/src/authorization_runtime/executor.rs b/crates/buzz-relay/src/authorization_runtime/executor.rs new file mode 100644 index 000000000..e57573512 --- /dev/null +++ b/crates/buzz-relay/src/authorization_runtime/executor.rs @@ -0,0 +1,1276 @@ +//! Transaction-owned protected mutation execution. +//! +//! A sealed permit is derived only from finalized enforcing authority. The +//! executor locks the durable invalidation and binding authorities, validates +//! expiry with the database clock, and commits the mutation and idempotency +//! receipt in one PostgreSQL transaction. + +use std::{fmt, sync::Arc}; + +use buzz_auth::{AuthorizationCapability, FederatedAuthorization}; +use buzz_core::CommunityId; +use buzz_db::authorization_invalidation::AuthorizationSelectorKind; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use sqlx::{Postgres, Row, Transaction}; +use thiserror::Error; +use uuid::Uuid; + +use super::transport::{ + LeaseCurrentStateError, ProtectedOperationAuthority, ProtectedTransportError, +}; + +const MAX_RESULT_BYTES: usize = 65_536; +const RESULT_VERSION: i16 = 1; + +/// One exact durable invalidation dependency carried to the commit boundary. +#[derive(Clone, PartialEq, Eq)] +pub struct CommitDependency { + kind: AuthorizationSelectorKind, + fingerprint: [u8; 32], + binding_version: Option, +} + +impl CommitDependency { + /// Preserve a dependency produced by the trusted invalidation runtime. + pub fn from_trusted_runtime( + kind: AuthorizationSelectorKind, + fingerprint: [u8; 32], + binding_version: Option, + ) -> Result { + if (kind == AuthorizationSelectorKind::Binding) != binding_version.is_some() + || binding_version == Some(0) + { + return Err(AuthorizationExecutionError::InvalidCommitFence); + } + Ok(Self { + kind, + fingerprint, + binding_version, + }) + } +} + +impl fmt::Debug for CommitDependency { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("CommitDependency") + .field("kind", &self.kind) + .field("fingerprint", &"[redacted]") + .field("binding_version", &"[redacted]") + .finish() + } +} + +/// Captured generation and complete selector set for one finalized authority. +#[derive(Clone, PartialEq, Eq)] +pub struct AuthorizationCommitFence { + evaluation_generation: u64, + dependencies: Vec, +} + +impl AuthorizationCommitFence { + /// Construct a commit fence from trusted finalization/invalidation state. + pub fn from_trusted_runtime( + evaluation_generation: u64, + mut dependencies: Vec, + ) -> Result { + if dependencies.is_empty() { + return Err(AuthorizationExecutionError::InvalidCommitFence); + } + dependencies.sort_by_key(|dependency| (dependency.kind, dependency.fingerprint)); + dependencies.dedup(); + let has_domain = dependencies + .iter() + .any(|dependency| dependency.kind == AuthorizationSelectorKind::Domain); + let has_binding = dependencies + .iter() + .any(|dependency| dependency.kind == AuthorizationSelectorKind::Binding); + let has_policy = dependencies + .iter() + .any(|dependency| dependency.kind == AuthorizationSelectorKind::PolicyVersion); + if !(has_domain && has_binding && has_policy) { + return Err(AuthorizationExecutionError::InvalidCommitFence); + } + Ok(Self { + evaluation_generation, + dependencies, + }) + } + + /// Construct a binding-independent fence for direct first enrollment. + pub fn from_trusted_enrollment_runtime( + evaluation_generation: u64, + mut dependencies: Vec, + ) -> Result { + if dependencies.is_empty() { + return Err(AuthorizationExecutionError::InvalidCommitFence); + } + dependencies.sort_by_key(|dependency| (dependency.kind, dependency.fingerprint)); + dependencies.dedup(); + let has = |kind| { + dependencies + .iter() + .any(|dependency| dependency.kind == kind) + }; + if !has(AuthorizationSelectorKind::Domain) + || !has(AuthorizationSelectorKind::PrincipalFingerprint) + || !has(AuthorizationSelectorKind::NostrKey) + || !has(AuthorizationSelectorKind::PolicyVersion) + || has(AuthorizationSelectorKind::Binding) + { + return Err(AuthorizationExecutionError::InvalidCommitFence); + } + Ok(Self { + evaluation_generation, + dependencies, + }) + } +} + +impl fmt::Debug for AuthorizationCommitFence { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("AuthorizationCommitFence") + .field("evaluation_generation", &self.evaluation_generation) + .field("dependency_count", &self.dependencies.len()) + .finish() + } +} + +/// Signed Redis-safe snapshot of one finalized ephemeral sender authority. +/// +/// It contains only opaque fingerprints and stable binding identifiers. Raw +/// issuer, subject, profile, and policy values never enter the fan-out wire. +#[derive(Clone, Serialize, Deserialize)] +pub(crate) struct EphemeralAuthorityClaim { + pub(crate) community_id: Uuid, + pub(crate) event_id: [u8; 32], + pub(crate) actor_pubkey: [u8; 32], + pub(crate) bound_pubkey: [u8; 32], + pub(crate) binding_id: Uuid, + pub(crate) binding_version: u64, + pub(crate) expires_at: u64, + pub(crate) capability: String, + evaluation_generation: u64, + dependencies: Vec, +} + +#[derive(Clone, Serialize, Deserialize)] +struct EphemeralCommitDependency { + kind: String, + fingerprint: [u8; 32], + binding_version: Option, +} + +impl EphemeralAuthorityClaim { + pub(super) fn from_authority( + authority: &ProtectedOperationAuthority, + event_id: [u8; 32], + ) -> Result { + authority.revalidate()?; + let capability = ephemeral_capability_label(authority.capability()) + .ok_or(ProtectedTransportError::SurfaceCapabilityMismatch)?; + let context = authority.context(); + let lease = context + .authorization_lease() + .ok_or(ProtectedTransportError::MissingAccessLease)?; + let binding = match context.federated_authorization() { + FederatedAuthorization::Direct { binding, .. } => binding, + FederatedAuthorization::Delegated { owner, .. } => owner, + FederatedAuthorization::NotRequired => { + return Err(ProtectedTransportError::MissingAccessLease) + } + }; + let fence = authority + .observer() + .observe_commit_fence() + .map_err(ProtectedTransportError::CurrentState)?; + Ok(Self { + community_id: *context.tenant().community().as_uuid(), + event_id, + actor_pubkey: context.pubkey().to_bytes(), + bound_pubkey: binding.bound_pubkey().to_bytes(), + binding_id: binding.binding_id(), + binding_version: binding.binding_version().get(), + expires_at: lease.expires_at(), + capability: capability.to_owned(), + evaluation_generation: fence.evaluation_generation, + dependencies: fence + .dependencies + .iter() + .map(|dependency| EphemeralCommitDependency { + kind: dependency.kind.as_str().to_owned(), + fingerprint: dependency.fingerprint, + binding_version: dependency.binding_version, + }) + .collect(), + }) + } +} + +/// Recheck a signed ephemeral claim against the writer database immediately +/// before a remote node releases it to a socket. +pub(crate) async fn revalidate_ephemeral_claim( + db: &buzz_db::Db, + claim: &EphemeralAuthorityClaim, +) -> Result<(), AuthorizationExecutionError> { + if claim.binding_version == 0 + || claim.dependencies.is_empty() + || !matches!(claim.capability.as_str(), "community_write" | "audio_join") + { + return Err(AuthorizationExecutionError::InvalidCommitFence); + } + let mut transaction = db.begin_transaction().await?; + let generation: Option = sqlx::query_scalar( + "SELECT generation FROM authorization_invalidation_domains \ + WHERE community_id = $1 FOR SHARE", + ) + .bind(claim.community_id) + .fetch_optional(&mut *transaction) + .await?; + let generation = generation.ok_or(AuthorizationExecutionError::Invalidated)?; + if generation < 0 || (generation as u64) < claim.evaluation_generation { + return Err(AuthorizationExecutionError::Invalidated); + } + for dependency in &claim.dependencies { + let row = sqlx::query( + "SELECT generation, sticky_deny, binding_version_floor \ + FROM authorization_invalidation_floors \ + WHERE community_id = $1 AND selector_kind = $2 \ + AND selector_fingerprint = $3 FOR SHARE", + ) + .bind(claim.community_id) + .bind(&dependency.kind) + .bind(dependency.fingerprint.as_slice()) + .fetch_optional(&mut *transaction) + .await?; + let Some(row) = row else { continue }; + let floor_generation: i64 = row.try_get("generation")?; + let sticky_deny: bool = row.try_get("sticky_deny")?; + let binding_floor: Option = row.try_get("binding_version_floor")?; + let fenced_after_evaluation = + floor_generation < 0 || floor_generation as u64 > claim.evaluation_generation; + let binding_denied = match (binding_floor, dependency.binding_version) { + (Some(floor), Some(version)) => floor < 0 || version <= floor as u64, + _ => false, + }; + if sticky_deny || fenced_after_evaluation || binding_denied { + return Err(AuthorizationExecutionError::Invalidated); + } + } + let version = i64::try_from(claim.binding_version) + .map_err(|_| AuthorizationExecutionError::InvalidBinding)?; + let active: Option = sqlx::query_scalar( + "SELECT 1 FROM identity_bindings \ + WHERE community_id = $1 AND binding_id = $2 AND pubkey = $3 \ + AND binding_version = $4 AND binding_state = 'active' FOR SHARE", + ) + .bind(claim.community_id) + .bind(claim.binding_id) + .bind(claim.bound_pubkey.as_slice()) + .bind(version) + .fetch_optional(&mut *transaction) + .await?; + if active.is_none() { + return Err(AuthorizationExecutionError::InvalidBinding); + } + ensure_not_expired(&mut transaction, claim.expires_at).await?; + transaction.commit().await?; + Ok(()) +} + +/// Stable retry identity for one protected operation. +#[derive(Clone, Copy, PartialEq, Eq, Hash)] +pub struct ProtectedOperationId(Uuid); + +impl ProtectedOperationId { + /// Derive a deterministic UUID from a domain-separated stable operation key. + pub fn derive( + authorization_domain: CommunityId, + operation_kind: &'static str, + stable_key: &[u8], + ) -> Result { + if operation_kind.is_empty() || operation_kind.len() > 128 || stable_key.is_empty() { + return Err(AuthorizationExecutionError::InvalidOperationIdentity); + } + let mut digest = Sha256::new(); + digest.update(b"buzz-protected-operation-id-v1"); + digest.update(authorization_domain.as_uuid().as_bytes()); + digest.update((operation_kind.len() as u64).to_be_bytes()); + digest.update(operation_kind.as_bytes()); + digest.update((stable_key.len() as u64).to_be_bytes()); + digest.update(stable_key); + let digest: [u8; 32] = digest.finalize().into(); + let mut bytes = [0u8; 16]; + bytes.copy_from_slice(&digest[..16]); + // Mark the digest-derived value as an RFC 4122 variant/version-5 UUID. + bytes[6] = (bytes[6] & 0x0f) | 0x50; + bytes[8] = (bytes[8] & 0x3f) | 0x80; + Ok(Self(Uuid::from_bytes(bytes))) + } + + pub(crate) const fn as_uuid(self) -> Uuid { + self.0 + } +} + +impl fmt::Debug for ProtectedOperationId { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("ProtectedOperationId([redacted])") + } +} + +/// Permit sealed from a finalized enforcing authorization context. +pub struct SealedOperationPermit { + community_id: CommunityId, + operation_id: ProtectedOperationId, + operation_kind: &'static str, + request_fingerprint: [u8; 32], + actor_pubkey: [u8; 32], + bound_pubkey: [u8; 32], + binding_id: Uuid, + binding_version: u64, + issuer: String, + subject: String, + expires_at: u64, + fence: AuthorizationCommitFence, +} + +impl SealedOperationPermit { + pub(super) fn from_authority( + authority: &ProtectedOperationAuthority, + operation_id: ProtectedOperationId, + operation_kind: &'static str, + request_fingerprint: [u8; 32], + ) -> Result { + authority.revalidate()?; + let context = authority.context(); + let lease = context + .authorization_lease() + .ok_or(ProtectedTransportError::MissingAccessLease)?; + let binding = match context.federated_authorization() { + FederatedAuthorization::Direct { binding, .. } => binding, + FederatedAuthorization::Delegated { owner, .. } => owner, + FederatedAuthorization::NotRequired => { + return Err(ProtectedTransportError::MissingAccessLease) + } + }; + let fence = authority + .observer() + .observe_commit_fence() + .map_err(ProtectedTransportError::CurrentState)?; + + let mut digest = Sha256::new(); + digest.update(b"buzz-protected-operation-request-v1"); + digest.update(context.tenant().community().as_uuid().as_bytes()); + digest.update(operation_id.as_uuid().as_bytes()); + digest.update((operation_kind.len() as u64).to_be_bytes()); + digest.update(operation_kind.as_bytes()); + let capability = capability_label(authority.capability()) + .ok_or(ProtectedTransportError::SurfaceCapabilityMismatch)?; + digest.update(capability.as_bytes()); + digest.update(context.pubkey().to_bytes()); + if let Some(owner) = context.agent_owner_pubkey() { + digest.update(owner.to_bytes()); + } + digest.update(lease.binding_id().as_bytes()); + digest.update(lease.binding_version().get().to_be_bytes()); + digest.update(lease.profile_id().as_str().as_bytes()); + digest.update(lease.policy_version().as_str().as_bytes()); + digest.update(lease.lease_version().get().to_be_bytes()); + digest.update(lease.expires_at().to_be_bytes()); + digest.update(request_fingerprint); + + Ok(Self { + community_id: context.tenant().community(), + operation_id, + operation_kind, + request_fingerprint: digest.finalize().into(), + actor_pubkey: context.pubkey().to_bytes(), + bound_pubkey: binding.bound_pubkey().to_bytes(), + binding_id: binding.binding_id(), + binding_version: binding.binding_version().get(), + issuer: binding.principal().issuer().to_owned(), + subject: binding.principal().subject().to_owned(), + expires_at: lease.expires_at(), + fence, + }) + } +} + +impl fmt::Debug for SealedOperationPermit { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("SealedOperationPermit") + .field("community_id", &"[redacted]") + .field("operation_id", &"[redacted]") + .field("operation_kind", &self.operation_kind) + .field("authority", &"[redacted]") + .finish() + } +} + +/// Binding-independent permit for one atomic direct first enrollment. +pub struct SealedEnrollmentPermit { + community_id: CommunityId, + operation_id: ProtectedOperationId, + operation_kind: &'static str, + request_fingerprint: [u8; 32], + actor_pubkey: [u8; 32], + issuer: String, + subject: String, + expires_at: u64, + fence: AuthorizationCommitFence, +} + +impl SealedEnrollmentPermit { + pub(super) fn from_authority( + authority: &super::transport::ProtectedEnrollmentAuthority, + operation_id: ProtectedOperationId, + operation_kind: &'static str, + request_fingerprint: [u8; 32], + ) -> Result { + authority.revalidate()?; + let disposition = authority.disposition(); + let fence = authority + .observer() + .observe_commit_fence() + .map_err(ProtectedTransportError::CurrentState)?; + let mut digest = Sha256::new(); + digest.update(b"buzz-protected-enrollment-request-v1"); + digest.update(disposition.authorization_domain().as_uuid().as_bytes()); + digest.update(operation_id.as_uuid().as_bytes()); + digest.update(operation_kind.as_bytes()); + digest.update(disposition.actor_pubkey().to_bytes()); + digest.update(disposition.principal().issuer().as_bytes()); + digest.update(disposition.principal().subject().as_bytes()); + digest.update(disposition.profile_id().as_str().as_bytes()); + digest.update(disposition.policy_version().as_str().as_bytes()); + digest.update(disposition.expires_at().to_be_bytes()); + digest.update(request_fingerprint); + Ok(Self { + community_id: disposition.authorization_domain(), + operation_id, + operation_kind, + request_fingerprint: digest.finalize().into(), + actor_pubkey: disposition.actor_pubkey().to_bytes(), + issuer: disposition.principal().issuer().to_owned(), + subject: disposition.principal().subject().to_owned(), + expires_at: disposition.expires_at(), + fence, + }) + } +} + +/// Start result for an idempotent protected operation. +pub enum AuthorizedOperationStart { + /// The same operation already committed; return this original typed payload. + Replay(Vec), + /// The caller owns the only transaction in which effects may be written. + Execute(Box), +} + +/// Start result for an atomic direct first enrollment. +pub enum AuthorizedEnrollmentStart { + /// The exact enrollment already committed. + Replay(Vec), + /// The caller owns the sole enrollment transaction. + Execute(Box), +} + +/// Open binding-independent enrollment transaction. +pub struct AuthorizedEnrollmentOperation { + transaction: Transaction<'static, Postgres>, + permit: SealedEnrollmentPermit, + restore: Arc, +} + +impl AuthorizedEnrollmentOperation { + /// Transaction used to bind identity, consume the invite, and add membership. + pub fn transaction(&mut self) -> &mut Transaction<'static, Postgres> { + &mut self.transaction + } + + /// Direct actor key bound by the staged assertion and Nostr proof. + pub const fn actor_pubkey(&self) -> &[u8; 32] { + &self.permit.actor_pubkey + } + + /// Literal validated issuer. + pub fn issuer(&self) -> &str { + &self.permit.issuer + } + + /// Literal validated subject. + pub fn subject(&self) -> &str { + &self.permit.subject + } + + /// Commit the enrollment and its bounded idempotency result together. + pub async fn commit( + mut self, + result_payload: &[u8], + ) -> Result, AuthorizationExecutionError> { + if result_payload.len() > MAX_RESULT_BYTES { + return Err(AuthorizationExecutionError::ResultTooLarge); + } + ensure_not_expired(&mut self.transaction, self.permit.expires_at).await?; + insert_receipt( + &mut self.transaction, + self.permit.community_id, + self.permit.operation_id, + self.permit.operation_kind, + &self.permit.request_fingerprint, + self.permit.expires_at, + result_payload, + ) + .await?; + let witness = self + .restore + .begin( + self.permit.community_id, + self.permit.operation_id.as_uuid(), + self.permit.request_fingerprint, + ) + .await?; + let commit_result = self.transaction.commit().await; + let witness_result = witness.commit().await; + if let Err(error) = witness_result { + return Err(error.into()); + } + if let Err(error) = commit_result { + // RestoreMutationGuard::commit returned success only after reading + // the exact durable operation receipt back from PostgreSQL. Never + // turn a bare failed commit acknowledgement into success. + tracing::warn!(%error, "protected enrollment commit acknowledgement was ambiguous; exact durable receipt verified"); + } + Ok(result_payload.to_vec()) + } +} + +/// Open transaction that has validated and locked its authorization authority. +pub struct AuthorizedOperation { + transaction: Transaction<'static, Postgres>, + permit: SealedOperationPermit, + restore: Option>, +} + +impl AuthorizedOperation { + /// Transaction to pass to transaction-aware mutation APIs. + pub fn transaction(&mut self) -> &mut Transaction<'static, Postgres> { + &mut self.transaction + } + + /// Atomically commit a bounded replay result with all transaction effects. + pub async fn commit( + mut self, + result_payload: &[u8], + ) -> Result, AuthorizationExecutionError> { + if result_payload.len() > MAX_RESULT_BYTES { + return Err(AuthorizationExecutionError::ResultTooLarge); + } + ensure_not_expired(&mut self.transaction, self.permit.expires_at).await?; + insert_receipt( + &mut self.transaction, + self.permit.community_id, + self.permit.operation_id, + self.permit.operation_kind, + &self.permit.request_fingerprint, + self.permit.expires_at, + result_payload, + ) + .await?; + let witness = match &self.restore { + Some(restore) => Some( + restore + .begin( + self.permit.community_id, + self.permit.operation_id.as_uuid(), + self.permit.request_fingerprint, + ) + .await?, + ), + #[cfg(test)] + None => None, + #[cfg(not(test))] + None => return Err(AuthorizationExecutionError::RestoreUnavailable), + }; + let commit_result = self.transaction.commit().await; + if let Some(witness) = witness { + witness.commit().await?; + if let Err(error) = commit_result { + // A successful witness commit proves that the exact receipt is + // durable. Without that proof the error above is returned. + tracing::warn!(%error, "protected operation commit acknowledgement was ambiguous; exact durable receipt verified"); + } + } else { + // Only test-only construction can omit restore protection. Keep + // even that path honest so a failed database commit is never + // reported as a successful protected effect. + commit_result?; + } + Ok(result_payload.to_vec()) + } +} + +#[allow(clippy::too_many_arguments)] +async fn insert_receipt( + transaction: &mut Transaction<'static, Postgres>, + community_id: CommunityId, + operation_id: ProtectedOperationId, + operation_kind: &'static str, + request_fingerprint: &[u8; 32], + expires_at: u64, + result_payload: &[u8], +) -> Result<(), AuthorizationExecutionError> { + sqlx::query( + "INSERT INTO authorization_operation_receipts \ + (community_id, operation_id, operation_kind, request_fingerprint, \ + result_version, result_payload, lease_expires_at) \ + VALUES ($1, $2, $3, $4, $5, $6, to_timestamp($7::double precision))", + ) + .bind(community_id.as_uuid()) + .bind(operation_id.as_uuid()) + .bind(operation_kind) + .bind(request_fingerprint.as_slice()) + .bind(RESULT_VERSION) + .bind(result_payload) + .bind(expires_at as f64) + .execute(&mut **transaction) + .await?; + Ok(()) +} + +/// Begin a transaction-owned protected mutation or replay its original result. +pub async fn begin_authorized_operation( + state: &crate::state::AppState, + permit: SealedOperationPermit, +) -> Result { + let restore = state + .restore_protection() + .ok_or(AuthorizationExecutionError::RestoreUnavailable)?; + begin_authorized_operation_inner(&state.db, Some(Arc::clone(restore)), permit).await +} + +async fn begin_authorized_operation_inner( + db: &buzz_db::Db, + restore: Option>, + permit: SealedOperationPermit, +) -> Result { + let mut transaction = db.begin_transaction().await?; + + // Serialize identical retries before reading the receipt. This avoids two + // first attempts executing the mutation body concurrently. + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") + .bind(permit.operation_id.as_uuid().to_string()) + .execute(&mut *transaction) + .await?; + + let committed_receipt = if let Some(row) = sqlx::query( + "SELECT operation_kind, request_fingerprint, result_version, result_payload \ + FROM authorization_operation_receipts \ + WHERE community_id = $1 AND operation_id = $2 FOR SHARE", + ) + .bind(permit.community_id.as_uuid()) + .bind(permit.operation_id.as_uuid()) + .fetch_optional(&mut *transaction) + .await? + { + let operation_kind: String = row.try_get("operation_kind")?; + let request_fingerprint: Vec = row.try_get("request_fingerprint")?; + let result_version: i16 = row.try_get("result_version")?; + let result_payload: Vec = row.try_get("result_payload")?; + if operation_kind != permit.operation_kind + || request_fingerprint.as_slice() != permit.request_fingerprint + || result_version != RESULT_VERSION + { + return Err(AuthorizationExecutionError::ConflictingRetry); + } + Some(result_payload) + } else { + None + }; + + sqlx::query( + "INSERT INTO authorization_invalidation_domains (community_id) \ + VALUES ($1) ON CONFLICT (community_id) DO NOTHING", + ) + .bind(permit.community_id.as_uuid()) + .execute(&mut *transaction) + .await?; + let generation: i64 = sqlx::query_scalar( + "SELECT generation FROM authorization_invalidation_domains \ + WHERE community_id = $1 FOR SHARE", + ) + .bind(permit.community_id.as_uuid()) + .fetch_one(&mut *transaction) + .await?; + if generation < 0 || (generation as u64) < permit.fence.evaluation_generation { + return Err(AuthorizationExecutionError::Invalidated); + } + + validate_dependency_floors(&mut transaction, permit.community_id, &permit.fence).await?; + validate_active_binding(&mut transaction, &permit).await?; + ensure_not_expired(&mut transaction, permit.expires_at).await?; + + if let Some(result_payload) = committed_receipt { + transaction.commit().await?; + return Ok(AuthorizedOperationStart::Replay(result_payload)); + } + + Ok(AuthorizedOperationStart::Execute(Box::new( + AuthorizedOperation { + transaction, + permit, + restore, + }, + ))) +} + +#[cfg(test)] +async fn begin_authorized_operation_for_test( + db: &buzz_db::Db, + permit: SealedOperationPermit, +) -> Result { + begin_authorized_operation_inner(db, None, permit).await +} + +/// Begin a binding-independent transaction that atomically creates first enrollment. +pub async fn begin_authorized_enrollment( + state: &crate::state::AppState, + permit: SealedEnrollmentPermit, +) -> Result { + let restore = state + .restore_protection() + .ok_or(AuthorizationExecutionError::RestoreUnavailable)?; + let mut transaction = state.db.begin_transaction().await?; + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") + .bind(permit.operation_id.as_uuid().to_string()) + .execute(&mut *transaction) + .await?; + + let committed_receipt = if let Some(row) = sqlx::query( + "SELECT operation_kind, request_fingerprint, result_version, result_payload \ + FROM authorization_operation_receipts \ + WHERE community_id = $1 AND operation_id = $2 FOR SHARE", + ) + .bind(permit.community_id.as_uuid()) + .bind(permit.operation_id.as_uuid()) + .fetch_optional(&mut *transaction) + .await? + { + let operation_kind: String = row.try_get("operation_kind")?; + let request_fingerprint: Vec = row.try_get("request_fingerprint")?; + let result_version: i16 = row.try_get("result_version")?; + let result_payload: Vec = row.try_get("result_payload")?; + if operation_kind != permit.operation_kind + || request_fingerprint.as_slice() != permit.request_fingerprint + || result_version != RESULT_VERSION + { + return Err(AuthorizationExecutionError::ConflictingRetry); + } + Some(result_payload) + } else { + None + }; + + sqlx::query( + "INSERT INTO authorization_invalidation_domains (community_id) \ + VALUES ($1) ON CONFLICT (community_id) DO NOTHING", + ) + .bind(permit.community_id.as_uuid()) + .execute(&mut *transaction) + .await?; + let generation: i64 = sqlx::query_scalar( + "SELECT generation FROM authorization_invalidation_domains \ + WHERE community_id = $1 FOR SHARE", + ) + .bind(permit.community_id.as_uuid()) + .fetch_one(&mut *transaction) + .await?; + if generation < 0 || (generation as u64) < permit.fence.evaluation_generation { + return Err(AuthorizationExecutionError::Invalidated); + } + validate_dependency_floors(&mut transaction, permit.community_id, &permit.fence).await?; + ensure_not_expired(&mut transaction, permit.expires_at).await?; + + if let Some(result_payload) = committed_receipt { + transaction.commit().await?; + return Ok(AuthorizedEnrollmentStart::Replay(result_payload)); + } + Ok(AuthorizedEnrollmentStart::Execute(Box::new( + AuthorizedEnrollmentOperation { + transaction, + permit, + restore: Arc::clone(restore), + }, + ))) +} + +async fn validate_dependency_floors( + transaction: &mut Transaction<'static, Postgres>, + community_id: CommunityId, + fence: &AuthorizationCommitFence, +) -> Result<(), AuthorizationExecutionError> { + for dependency in &fence.dependencies { + let row = sqlx::query( + "SELECT generation, sticky_deny, binding_version_floor \ + FROM authorization_invalidation_floors \ + WHERE community_id = $1 AND selector_kind = $2 \ + AND selector_fingerprint = $3", + ) + .bind(community_id.as_uuid()) + .bind(dependency.kind.as_str()) + .bind(dependency.fingerprint.as_slice()) + .fetch_optional(&mut **transaction) + .await?; + let Some(row) = row else { continue }; + let generation: i64 = row.try_get("generation")?; + let sticky_deny: bool = row.try_get("sticky_deny")?; + let binding_floor: Option = row.try_get("binding_version_floor")?; + let fenced_after_evaluation = + generation < 0 || generation as u64 > fence.evaluation_generation; + let binding_denied = match (binding_floor, dependency.binding_version) { + (Some(floor), Some(version)) => floor < 0 || version <= floor as u64, + _ => false, + }; + if sticky_deny || fenced_after_evaluation || binding_denied { + return Err(AuthorizationExecutionError::Invalidated); + } + } + Ok(()) +} + +async fn validate_active_binding( + transaction: &mut Transaction<'static, Postgres>, + permit: &SealedOperationPermit, +) -> Result<(), AuthorizationExecutionError> { + let version = i64::try_from(permit.binding_version) + .map_err(|_| AuthorizationExecutionError::InvalidBinding)?; + let active: Option = sqlx::query_scalar( + "SELECT 1 FROM identity_bindings \ + WHERE community_id = $1 AND binding_id = $2 \ + AND issuer = $3 AND uid = $4 AND pubkey = $5 \ + AND binding_version = $6 AND binding_state = 'active' \ + FOR SHARE", + ) + .bind(permit.community_id.as_uuid()) + .bind(permit.binding_id) + .bind(&permit.issuer) + .bind(&permit.subject) + .bind(permit.bound_pubkey.as_slice()) + .bind(version) + .fetch_optional(&mut **transaction) + .await?; + if active.is_none() { + return Err(AuthorizationExecutionError::InvalidBinding); + } + // The actor is included in the sealed fingerprint even when delegated; + // direct/owner agreement was already enforced by AuthContext finalization. + let _ = permit.actor_pubkey; + Ok(()) +} + +async fn ensure_not_expired( + transaction: &mut Transaction<'static, Postgres>, + expires_at: u64, +) -> Result<(), AuthorizationExecutionError> { + let current: f64 = + sqlx::query_scalar("SELECT EXTRACT(EPOCH FROM clock_timestamp())::double precision") + .fetch_one(&mut **transaction) + .await?; + if !current.is_finite() || current >= expires_at as f64 { + return Err(AuthorizationExecutionError::Expired); + } + Ok(()) +} + +const fn capability_label(capability: AuthorizationCapability) -> Option<&'static str> { + match capability { + AuthorizationCapability::CommunityRead => Some("community_read"), + AuthorizationCapability::CommunityWrite => Some("community_write"), + AuthorizationCapability::Moderate => Some("moderate"), + AuthorizationCapability::InviteMint => Some("invite_mint"), + AuthorizationCapability::InviteClaim => Some("invite_claim"), + AuthorizationCapability::MediaRead => Some("media_read"), + AuthorizationCapability::MediaWrite => Some("media_write"), + AuthorizationCapability::GitRead => Some("git_read"), + AuthorizationCapability::GitWrite => Some("git_write"), + AuthorizationCapability::AudioJoin => Some("audio_join"), + _ => None, + } +} + +const fn ephemeral_capability_label(capability: AuthorizationCapability) -> Option<&'static str> { + match capability { + AuthorizationCapability::CommunityWrite => Some("community_write"), + AuthorizationCapability::AudioJoin => Some("audio_join"), + _ => None, + } +} + +/// Fail-closed protected mutation execution error. +#[derive(Debug, Error)] +pub enum AuthorizationExecutionError { + /// Database operation failed. + #[error("protected operation database transaction failed")] + Database(#[from] sqlx::Error), + /// Database wrapper failed before the transaction began. + #[error("protected operation database transaction failed")] + Db(#[from] buzz_db::DbError), + /// Independent restore witness was unavailable or rejected reconciliation. + #[error("protected operation restore witness failed")] + Restore(#[from] super::restore::RestoreProtectionError), + /// Production execution was attempted without the mandatory witness. + #[error("protected operation restore witness is unavailable")] + RestoreUnavailable, + /// Captured generation or dependency set was incomplete. + #[error("protected operation commit fence is invalid")] + InvalidCommitFence, + /// Stable operation identity was empty or malformed. + #[error("protected operation identity is invalid")] + InvalidOperationIdentity, + /// The stable operation ID was reused for different input. + #[error("protected operation retry conflicts with the committed request")] + ConflictingRetry, + /// Current durable invalidation state denies the operation. + #[error("protected operation authority was invalidated")] + Invalidated, + /// The exact active binding changed or disappeared. + #[error("protected operation binding is no longer active")] + InvalidBinding, + /// The lease expired before the commit boundary. + #[error("protected operation authorization expired before commit")] + Expired, + /// Replay result exceeded the bounded receipt payload. + #[error("protected operation result is too large")] + ResultTooLarge, +} + +impl From for LeaseCurrentStateError { + fn from(_error: AuthorizationExecutionError) -> Self { + LeaseCurrentStateError::Unavailable + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; + const TEST_ISSUER: &str = "https://idp.example"; + const TEST_SUBJECT: &str = "executor-subject"; + + fn dependency( + kind: AuthorizationSelectorKind, + marker: u8, + binding_version: Option, + ) -> CommitDependency { + CommitDependency::from_trusted_runtime(kind, [marker; 32], binding_version) + .expect("valid dependency") + } + + fn operation_fence() -> AuthorizationCommitFence { + AuthorizationCommitFence::from_trusted_runtime( + 0, + vec![ + dependency(AuthorizationSelectorKind::Domain, 1, None), + dependency(AuthorizationSelectorKind::Binding, 2, Some(1)), + dependency(AuthorizationSelectorKind::PolicyVersion, 3, None), + ], + ) + .expect("complete operation fence") + } + + fn epoch_after(seconds: u64) -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system time") + .as_secs() + + seconds + } + + fn permit( + community_id: CommunityId, + binding_id: Uuid, + operation_id: ProtectedOperationId, + request_marker: u8, + expires_at: u64, + ) -> SealedOperationPermit { + SealedOperationPermit { + community_id, + operation_id, + operation_kind: "executor.test.v1", + request_fingerprint: [request_marker; 32], + actor_pubkey: [7; 32], + bound_pubkey: [7; 32], + binding_id, + binding_version: 1, + issuer: TEST_ISSUER.to_owned(), + subject: TEST_SUBJECT.to_owned(), + expires_at, + fence: operation_fence(), + } + } + + async fn integration_setup() -> (buzz_db::Db, CommunityId, Uuid) { + let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap_or_else(|_| TEST_DB_URL.to_owned()); + let pool = sqlx::PgPool::connect(&database_url) + .await + .expect("test database"); + buzz_db::migration::run_migrations(&pool) + .await + .expect("test migrations"); + let community_id = CommunityId::from_uuid(Uuid::new_v4()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_id.as_uuid()) + .bind(format!( + "executor-{}.example", + community_id.as_uuid().simple() + )) + .execute(&pool) + .await + .expect("test community"); + let binding_id = Uuid::new_v4(); + sqlx::query( + "INSERT INTO identity_bindings \ + (community_id, issuer, uid, pubkey, source, binding_id, \ + binding_version, binding_state, binding_provenance) \ + VALUES ($1, $2, $3, $4, 'jwt_npub', $5, 1, 'active', 'attested_key')", + ) + .bind(community_id.as_uuid()) + .bind(TEST_ISSUER) + .bind(TEST_SUBJECT) + .bind([7_u8; 32].as_slice()) + .bind(binding_id) + .execute(&pool) + .await + .expect("active test binding"); + (buzz_db::Db::from_pool(pool), community_id, binding_id) + } + + #[test] + fn stable_operation_identity_is_domain_and_kind_separated() { + let first_domain = CommunityId::from_uuid(Uuid::from_u128(1)); + let second_domain = CommunityId::from_uuid(Uuid::from_u128(2)); + let first = ProtectedOperationId::derive(first_domain, "event.ingest.v1", b"event") + .expect("operation id"); + assert_eq!( + first, + ProtectedOperationId::derive(first_domain, "event.ingest.v1", b"event") + .expect("stable operation id") + ); + assert_ne!( + first, + ProtectedOperationId::derive(second_domain, "event.ingest.v1", b"event") + .expect("domain-separated operation id") + ); + assert_ne!( + first, + ProtectedOperationId::derive(first_domain, "invite.mint.v1", b"event") + .expect("kind-separated operation id") + ); + } + + #[test] + fn operation_and_enrollment_fences_require_distinct_authority_sets() { + assert!(AuthorizationCommitFence::from_trusted_runtime( + 0, + vec![ + dependency(AuthorizationSelectorKind::Domain, 1, None), + dependency(AuthorizationSelectorKind::PolicyVersion, 3, None), + ], + ) + .is_err()); + assert!(AuthorizationCommitFence::from_trusted_enrollment_runtime( + 0, + vec![ + dependency(AuthorizationSelectorKind::Domain, 1, None), + dependency(AuthorizationSelectorKind::PrincipalFingerprint, 2, None,), + dependency(AuthorizationSelectorKind::NostrKey, 3, None), + dependency(AuthorizationSelectorKind::PolicyVersion, 4, None), + dependency(AuthorizationSelectorKind::Binding, 5, Some(1)), + ], + ) + .is_err()); + assert!(AuthorizationCommitFence::from_trusted_enrollment_runtime( + 0, + vec![ + dependency(AuthorizationSelectorKind::Domain, 1, None), + dependency(AuthorizationSelectorKind::PrincipalFingerprint, 2, None,), + dependency(AuthorizationSelectorKind::NostrKey, 3, None), + dependency(AuthorizationSelectorKind::PolicyVersion, 4, None), + ], + ) + .is_ok()); + let _ = operation_fence(); + } + + #[test] + fn capability_receipt_labels_are_stable() { + assert_eq!( + capability_label(AuthorizationCapability::CommunityWrite), + Some("community_write") + ); + assert_eq!( + capability_label(AuthorizationCapability::AudioJoin), + Some("audio_join") + ); + assert_eq!( + ephemeral_capability_label(AuthorizationCapability::CommunityWrite), + Some("community_write") + ); + assert_eq!( + ephemeral_capability_label(AuthorizationCapability::AudioJoin), + Some("audio_join") + ); + assert_eq!( + ephemeral_capability_label(AuthorizationCapability::MediaWrite), + None + ); + } + + #[tokio::test] + #[ignore = "requires migrated Postgres"] + async fn postgres_executor_serializes_retries_rolls_back_and_checks_expiry() { + let (db, community_id, binding_id) = integration_setup().await; + let concurrent_id = + ProtectedOperationId::derive(community_id, "executor.test.v1", b"concurrent") + .expect("operation id"); + let run = |permit| async { + match begin_authorized_operation_for_test(&db, permit) + .await + .expect("authorized operation") + { + AuthorizedOperationStart::Execute(operation) => { + operation.commit(b"committed-once").await.expect("commit"); + true + } + AuthorizedOperationStart::Replay(payload) => { + assert_eq!(payload, b"committed-once"); + false + } + } + }; + let (first, second) = tokio::join!( + run(permit( + community_id, + binding_id, + concurrent_id, + 1, + epoch_after(60), + )), + run(permit( + community_id, + binding_id, + concurrent_id, + 1, + epoch_after(60), + )), + ); + assert_ne!(first, second, "exactly one concurrent retry executes"); + + assert!(matches!( + begin_authorized_operation_for_test( + &db, + permit(community_id, binding_id, concurrent_id, 2, epoch_after(60),), + ) + .await, + Err(AuthorizationExecutionError::ConflictingRetry) + )); + + let rollback_id = + ProtectedOperationId::derive(community_id, "executor.test.v1", b"rollback") + .expect("operation id"); + let AuthorizedOperationStart::Execute(rolled_back) = begin_authorized_operation_for_test( + &db, + permit(community_id, binding_id, rollback_id, 3, epoch_after(60)), + ) + .await + .expect("open rollback operation") else { + panic!("uncommitted operation cannot replay") + }; + rolled_back + .transaction + .rollback() + .await + .expect("explicit rollback"); + assert!(matches!( + begin_authorized_operation_for_test( + &db, + permit(community_id, binding_id, rollback_id, 3, epoch_after(60),), + ) + .await + .expect("retry after rollback"), + AuthorizedOperationStart::Execute(_) + )); + + let expired_id = ProtectedOperationId::derive(community_id, "executor.test.v1", b"expired") + .expect("operation id"); + assert!(matches!( + begin_authorized_operation_for_test( + &db, + permit( + community_id, + binding_id, + expired_id, + 4, + epoch_after(0).saturating_sub(1), + ), + ) + .await, + Err(AuthorizationExecutionError::Expired) + )); + + let invalidation_id = Uuid::new_v4(); + let mut invalidation = db.begin_transaction().await.expect("invalidation tx"); + let invalidation_generation: i64 = sqlx::query_scalar( + "UPDATE authorization_invalidation_domains \ + SET generation = generation + 1 WHERE community_id = $1 \ + RETURNING generation", + ) + .bind(community_id.as_uuid()) + .fetch_one(&mut *invalidation) + .await + .expect("advance invalidation generation"); + sqlx::query( + "INSERT INTO authorization_invalidation_receipts \ + (community_id, event_id, generation, request_fingerprint) \ + VALUES ($1, $2, $3, $4)", + ) + .bind(community_id.as_uuid()) + .bind(invalidation_id) + .bind(invalidation_generation) + .bind([9_u8; 32].as_slice()) + .execute(&mut *invalidation) + .await + .expect("invalidation receipt"); + sqlx::query( + "INSERT INTO authorization_invalidation_floors \ + (community_id, selector_kind, selector_fingerprint, generation, sticky_deny) \ + VALUES ($1, 'domain', $2, $3, true)", + ) + .bind(community_id.as_uuid()) + .bind([1_u8; 32].as_slice()) + .bind(invalidation_generation) + .execute(&mut *invalidation) + .await + .expect("domain deny floor"); + invalidation.commit().await.expect("commit invalidation"); + + let invalidated_id = + ProtectedOperationId::derive(community_id, "executor.test.v1", b"invalidated") + .expect("operation id"); + assert!(matches!( + begin_authorized_operation_for_test( + &db, + permit(community_id, binding_id, invalidated_id, 5, epoch_after(60),), + ) + .await, + Err(AuthorizationExecutionError::Invalidated) + )); + } +} diff --git a/crates/buzz-relay/src/authorization_runtime/finalization.rs b/crates/buzz-relay/src/authorization_runtime/finalization.rs new file mode 100644 index 000000000..1b82752c6 --- /dev/null +++ b/crates/buzz-relay/src/authorization_runtime/finalization.rs @@ -0,0 +1,632 @@ +//! Relay adapter for exact-domain authorization policy and finalization. +//! +//! A request never selects its provider profile or activation mode. The relay +//! resolves one immutable policy from the row-zero [`TenantContext`] domain. +//! Missing and duplicate domain configuration fail closed without a global or +//! Nostr-only fallback. + +use std::{collections::HashMap, fmt, sync::Arc}; + +use buzz_auth::{ + resolve_authorization, AccessLeasePolicy, AuthorizationFinalizer, AuthorizationOutcome, + AuthorizationProfileId, AuthorizationProvider, AuthorizationRequest, BindingLeaseBound, + CapabilitySet, DecisionSource, EnrollmentMode, FederatedAuthorization, FederatedPrincipal, + FinalizationError, LeaseVersion, PolicyVersion, ProviderAuthorizationClock, + ProviderContractError, ProviderTimeout, ResolvedFederatedPolicy, SharedAuthorizationClock, + VerificationOnlyDisposition, VerificationStatusPolicy, VerifiedFederatedAssertion, + VerifiedNostrProof, VersionedBindingRef, +}; +use buzz_core::{tenant::TenantContext, CommunityId}; +use thiserror::Error; +use uuid::Uuid; + +/// Server-owned activation mode for one exact authorization domain. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AuthorizationMode { + /// Do not evaluate federated identity or provider policy. + Off, + /// Evaluate read-only provider policy without binding or authority changes. + Shadow, + /// Produce a short-lived display-only result after full direct finalization. + VerifyOnly, + /// Issue bounded access leases after full direct or delegated finalization. + Enforce, + /// Keep every protected surface active while denying all protected access. + DenyProtected, +} + +impl AuthorizationMode { + /// Whether this mode may evaluate the configured admission provider. + pub const fn evaluates_provider(self) -> bool { + matches!(self, Self::Shadow | Self::VerifyOnly | Self::Enforce) + } + + /// Whether this mode keeps the protected-surface inventory authoritative. + pub const fn protects_surfaces(self) -> bool { + matches!(self, Self::Enforce | Self::DenyProtected) + } +} + +/// Immutable server configuration for one exact authorization domain. +#[derive(Clone)] +pub struct DomainAuthorizationPolicy { + authorization_domain: CommunityId, + profile_id: AuthorizationProfileId, + provider: Arc, + enrollment_mode: EnrollmentMode, + mode: AuthorizationMode, + provider_timeout: ProviderTimeout, + access_lease_policy: AccessLeasePolicy, + verification_status_policy: VerificationStatusPolicy, +} + +impl DomainAuthorizationPolicy { + /// Build policy exclusively from trusted server configuration. + #[allow(clippy::too_many_arguments)] + pub fn from_server_configuration( + authorization_domain: CommunityId, + profile_id: impl Into, + provider: Arc, + enrollment_mode: EnrollmentMode, + mode: AuthorizationMode, + provider_timeout: ProviderTimeout, + access_lease_policy: AccessLeasePolicy, + verification_status_policy: VerificationStatusPolicy, + ) -> Result { + Ok(Self { + authorization_domain, + profile_id: AuthorizationProfileId::from_server_configuration(profile_id)?, + provider, + enrollment_mode, + mode, + provider_timeout, + access_lease_policy, + verification_status_policy, + }) + } + + /// Exact server-owned authorization domain. + pub const fn authorization_domain(&self) -> CommunityId { + self.authorization_domain + } + + /// Exact server-owned activation mode. + pub const fn mode(&self) -> AuthorizationMode { + self.mode + } + + /// Server-resolved provider profile. + pub const fn profile_id(&self) -> &AuthorizationProfileId { + &self.profile_id + } + + /// Server-resolved binding enrollment mode. + pub const fn enrollment_mode(&self) -> EnrollmentMode { + self.enrollment_mode + } +} + +impl fmt::Debug for DomainAuthorizationPolicy { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("DomainAuthorizationPolicy") + .field("authorization_domain", &"[redacted]") + .field("profile_id", &"[redacted]") + .field("provider", &"[configured]") + .field("enrollment_mode", &"[redacted]") + .field("mode", &self.mode) + .field("provider_timeout", &"[redacted]") + .field("access_lease_policy", &"[redacted]") + .field("verification_status_policy", &"[redacted]") + .finish() + } +} + +/// Immutable exact-domain provider selector. +#[derive(Clone)] +pub struct DomainProviderSelector { + policies: HashMap, +} + +impl DomainProviderSelector { + /// Build an exact-domain selector, rejecting every ambiguous duplicate. + pub fn new( + policies: impl IntoIterator, + ) -> Result { + let mut by_domain = HashMap::new(); + for policy in policies { + let domain = policy.authorization_domain; + if by_domain.insert(domain, policy).is_some() { + return Err(DomainPolicyError::AmbiguousDomainPolicy); + } + } + Ok(Self { + policies: by_domain, + }) + } + + /// Resolve policy only from the row-zero server tenant. + /// + /// No default provider exists. A federated authorization attempt for an + /// unconfigured domain is denied as missing policy. + pub fn resolve( + &self, + tenant: &TenantContext, + ) -> Result<&DomainAuthorizationPolicy, DomainPolicyError> { + self.policies + .get(&tenant.community()) + .ok_or(DomainPolicyError::MissingDomainPolicy) + } +} + +impl fmt::Debug for DomainProviderSelector { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("DomainProviderSelector") + .field("policies", &"[redacted]") + .finish() + } +} + +/// Fail-closed exact-domain policy resolution error. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)] +pub enum DomainPolicyError { + /// More than one authorization policy named the same exact domain. + #[error("federated authorization policy is ambiguous for this domain")] + AmbiguousDomainPolicy, + /// No federated policy was configured for this exact domain. + #[error("federated authorization policy is missing for this domain")] + MissingDomainPolicy, +} + +/// Provider-neutral relay finalizer with immutable policy and injected time. +#[derive(Clone)] +pub struct RelayAuthorizationFinalizer { + selector: DomainProviderSelector, + finalizer: AuthorizationFinalizer, + clock: SharedAuthorizationClock, + runtime_binding: Uuid, +} + +struct RelayProviderClock<'a>(&'a dyn buzz_auth::AuthorizationClock); + +impl ProviderAuthorizationClock for RelayProviderClock<'_> { + fn now_unix_seconds(&self) -> Option { + self.0.now().ok().map(|value| value.unix_seconds()) + } +} + +impl RelayAuthorizationFinalizer { + /// Build a runtime that shares one central clock across evaluation and finalization. + pub fn new(selector: DomainProviderSelector, clock: SharedAuthorizationClock) -> Self { + Self { + selector, + finalizer: AuthorizationFinalizer::new(Arc::clone(&clock)), + clock, + runtime_binding: Uuid::new_v4(), + } + } + + /// Evaluate current direct provider admission for a server-resolved domain. + /// + /// Off mode performs no provider call. All other modes preserve provider + /// deny and unavailable outcomes without falling back to another policy. + #[allow(clippy::too_many_arguments)] + pub async fn evaluate_direct( + &self, + tenant: &TenantContext, + proof: &VerifiedNostrProof, + assertion: &VerifiedFederatedAssertion, + federated_policy: ResolvedFederatedPolicy, + requested_capabilities: CapabilitySet, + correlation_id: Uuid, + ) -> Result { + let policy = self.selector.resolve(tenant)?; + if !policy.mode.evaluates_provider() { + return Err(RelayFinalizationError::ModeDoesNotEvaluate); + } + let now = self.finalizer.now()?; + let request = AuthorizationRequest::direct( + proof, + assertion, + federated_policy, + requested_capabilities, + correlation_id, + now.unix_seconds(), + )?; + Ok(resolve_authorization( + policy.provider.as_ref(), + &request, + &RelayProviderClock(self.clock.as_ref()), + policy.provider_timeout, + self.runtime_binding, + ) + .await) + } + + /// Evaluate current delegated-owner provider admission for an exact domain. + pub async fn evaluate_delegated( + &self, + tenant: &TenantContext, + proof: &VerifiedNostrProof, + owner: &VersionedBindingRef, + federated_policy: ResolvedFederatedPolicy, + requested_capabilities: CapabilitySet, + correlation_id: Uuid, + ) -> Result { + let policy = self.selector.resolve(tenant)?; + if !policy.mode.evaluates_provider() { + return Err(RelayFinalizationError::ModeDoesNotEvaluate); + } + let now = self.finalizer.now()?; + let request = AuthorizationRequest::delegated_from_active_binding( + proof, + owner, + federated_policy, + requested_capabilities, + correlation_id, + now.unix_seconds(), + )?; + Ok(resolve_authorization( + policy.provider.as_ref(), + &request, + &RelayProviderClock(self.clock.as_ref()), + policy.provider_timeout, + self.runtime_binding, + ) + .await) + } + + /// Finalize one validated allow snapshot according to server-owned mode. + /// + /// Off and shadow modes cannot finalize a binding, status, or access + /// context. Verify-only returns a distinct display type; enforce is the + /// only branch capable of returning an access context with a lease. + pub fn finalize_allowed( + &self, + input: buzz_auth::AuthContextInput, + federated_policy: ResolvedFederatedPolicy, + authorization: FederatedAuthorization, + snapshot: Box, + binding_bound: BindingLeaseBound, + lease_version: LeaseVersion, + ) -> Result { + let policy = self.selector.resolve(input.tenant())?; + match policy.mode { + AuthorizationMode::Off + | AuthorizationMode::Shadow + | AuthorizationMode::DenyProtected => Err(RelayFinalizationError::ModeDoesNotFinalize), + AuthorizationMode::VerifyOnly => self + .finalizer + .finalize_verification_only( + input, + federated_policy, + authorization, + snapshot, + &policy.profile_id, + binding_bound, + policy.verification_status_policy, + ) + .map(RuntimeAuthorizationDisposition::VerificationOnly) + .map_err(Into::into), + AuthorizationMode::Enforce => self + .finalizer + .finalize_access( + input, + federated_policy, + authorization, + snapshot, + &policy.profile_id, + binding_bound, + policy.access_lease_policy, + lease_version, + ) + .map(|context| RuntimeAuthorizationDisposition::Access(Box::new(context))) + .map_err(Into::into), + } + } + + /// Finalize the same current direct evidence into a short-lived, + /// display-only status. The caller separately proves the presentation + /// gate; this method cannot issue access or a lease and performs no writes. + pub fn finalize_client_status( + &self, + input: buzz_auth::AuthContextInput, + federated_policy: ResolvedFederatedPolicy, + authorization: FederatedAuthorization, + snapshot: Box, + binding_bound: BindingLeaseBound, + ) -> Result { + let policy = self.selector.resolve(input.tenant())?; + if !matches!( + policy.mode, + AuthorizationMode::VerifyOnly | AuthorizationMode::Enforce + ) { + return Err(RelayFinalizationError::ModeDoesNotFinalize); + } + self.finalizer + .finalize_verification_only( + input, + federated_policy, + authorization, + snapshot, + &policy.profile_id, + binding_bound, + policy.verification_status_policy, + ) + .map_err(Into::into) + } + + /// Finalize direct first-enrollment evidence without creating a binding. + pub fn finalize_enrollment( + &self, + tenant: &TenantContext, + proof: &VerifiedNostrProof, + assertion: &VerifiedFederatedAssertion, + federated_policy: ResolvedFederatedPolicy, + snapshot: Box, + correlation_id: Uuid, + ) -> Result { + let policy = self.selector.resolve(tenant)?; + if policy.mode != AuthorizationMode::Enforce { + return Err(RelayFinalizationError::ModeDoesNotFinalize); + } + if policy.enrollment_mode != EnrollmentMode::AttestedKey { + return Err(RelayFinalizationError::EnrollmentModeUnsupported); + } + let now = self.finalizer.now()?; + let key = assertion + .key_attestation() + .ok_or(RelayFinalizationError::EnrollmentEvidenceMismatch)?; + if proof.verified_delegation().is_some() + || proof.authorization_domain() != tenant.community() + || federated_policy.authorization_domain() != tenant.community() + || !snapshot.is_bound_to_federated_policy(&federated_policy) + || assertion.authorization_domain() != tenant.community() + || snapshot.authorization_domain() != tenant.community() + || proof.authorized_transport() != assertion.authorized_transport() + || snapshot.transport() != proof.authorized_transport() + || snapshot.actor_pubkey() != proof.actor_pubkey() + || key.pubkey() != proof.actor_pubkey() + || snapshot.owner_pubkey().is_some() + || snapshot.binding_id().is_some() + || snapshot.binding_version().is_some() + || snapshot.proof_method() != proof.proof_method() + || snapshot.principal() != assertion.principal() + || snapshot.profile_id() != &policy.profile_id + || snapshot.decision_source() != DecisionSource::DirectAssertion + || snapshot.correlation_id() != correlation_id + || !snapshot + .capabilities() + .contains(buzz_auth::AuthorizationCapability::InviteClaim) + || snapshot.issued_at() > now.unix_seconds() + || snapshot.fresh_until() <= now.unix_seconds() + || snapshot.effective_until() <= now.unix_seconds() + || assertion + .not_before() + .is_some_and(|bound| bound.is_not_yet_valid_at(now.unix_seconds())) + || assertion.expires_at().is_expired_at(now.unix_seconds()) + { + return Err(RelayFinalizationError::EnrollmentEvidenceMismatch); + } + let application_until = now + .unix_seconds() + .checked_add(policy.access_lease_policy.application_limit().seconds()) + .ok_or(RelayFinalizationError::EnrollmentEvidenceMismatch)?; + let expires_at = snapshot + .effective_until() + .min(assertion.expires_at().unix_seconds()) + .min(application_until) + .saturating_sub(policy.access_lease_policy.clock_skew().seconds()); + if expires_at <= now.unix_seconds() { + return Err(RelayFinalizationError::EnrollmentEvidenceMismatch); + } + Ok(EnrollmentDisposition { + authorization_domain: tenant.community(), + actor_pubkey: proof.actor_pubkey(), + principal: assertion.principal().clone(), + profile_id: policy.profile_id.clone(), + policy_version: snapshot.policy_version().clone(), + correlation_id, + expires_at, + }) + } +} + +/// Direct provider decision sealed for atomic first enrollment. +#[must_use] +pub struct EnrollmentDisposition { + authorization_domain: CommunityId, + actor_pubkey: nostr::PublicKey, + principal: FederatedPrincipal, + profile_id: AuthorizationProfileId, + policy_version: PolicyVersion, + correlation_id: Uuid, + expires_at: u64, +} + +impl EnrollmentDisposition { + /// Exact server-resolved authorization domain. + pub const fn authorization_domain(&self) -> CommunityId { + self.authorization_domain + } + /// Direct actor whose key is attested by the assertion. + pub const fn actor_pubkey(&self) -> nostr::PublicKey { + self.actor_pubkey + } + /// Literal issuer-qualified principal staged for enrollment. + pub const fn principal(&self) -> &FederatedPrincipal { + &self.principal + } + /// Server-selected authorization profile. + pub const fn profile_id(&self) -> &AuthorizationProfileId { + &self.profile_id + } + /// Provider policy version that authorized the enrollment. + pub const fn policy_version(&self) -> &PolicyVersion { + &self.policy_version + } + /// Exact decision correlation identifier. + pub const fn correlation_id(&self) -> Uuid { + self.correlation_id + } + /// Earliest authoritative expiry bound for the enrollment transaction. + pub const fn expires_at(&self) -> u64 { + self.expires_at + } +} + +impl fmt::Debug for EnrollmentDisposition { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("EnrollmentDisposition") + .field("evidence", &"[redacted]") + .finish() + } +} + +impl fmt::Debug for RelayAuthorizationFinalizer { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("RelayAuthorizationFinalizer") + .field("selector", &self.selector) + .field("finalizer", &self.finalizer) + .finish() + } +} + +/// Typed result of server-mode finalization. +#[must_use] +pub enum RuntimeAuthorizationDisposition { + /// Enforcing authority carrying a bounded access lease. + Access(Box), + /// Display-only verification carrying no access authority. + VerificationOnly(VerificationOnlyDisposition), +} + +impl fmt::Debug for RuntimeAuthorizationDisposition { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Access(_) => formatter + .debug_tuple("Access") + .field(&"[redacted]") + .finish(), + Self::VerificationOnly(_) => formatter + .debug_tuple("VerificationOnly") + .field(&"[redacted]") + .finish(), + } + } +} + +/// Fail-closed relay finalization adapter error. +#[derive(Debug, Error)] +pub enum RelayFinalizationError { + /// Exact-domain server policy could not be resolved. + #[error(transparent)] + DomainPolicy(#[from] DomainPolicyError), + /// Central authorization time was unavailable. + #[error(transparent)] + Clock(#[from] buzz_auth::AuthorizationClockError), + /// Provider request evidence was inconsistent. + #[error(transparent)] + ProviderContract(#[from] ProviderContractError), + /// Full provider/binding finalization failed. + #[error(transparent)] + Finalization(#[from] FinalizationError), + /// A non-evaluating mode was asked to evaluate federated policy. + #[error("server-resolved authorization mode does not evaluate federated policy")] + ModeDoesNotEvaluate, + /// A non-finalizing mode was asked to create status or authority. + #[error("server-resolved authorization mode does not permit finalization")] + ModeDoesNotFinalize, + /// First enrollment was configured for a non-attested mode. + #[error("server-resolved authorization policy does not permit direct enrollment")] + EnrollmentModeUnsupported, + /// Direct assertion, provider, proof, or expiry evidence did not match. + #[error("direct enrollment evidence is inconsistent or stale")] + EnrollmentEvidenceMismatch, +} + +#[cfg(test)] +mod tests { + use std::{future::ready, time::Duration}; + + use buzz_auth::{ + ApplicationLeaseLimit, AuthorizationClockSkew, AuthorizationDenial, + AuthorizationDenialReason, AuthorizationProviderFuture, ProviderDecision, + }; + use uuid::Uuid; + + use super::*; + + struct DenyProvider; + + impl AuthorizationProvider for DenyProvider { + fn authorize<'a>( + &'a self, + _request: &'a AuthorizationRequest, + ) -> AuthorizationProviderFuture<'a> { + Box::pin(ready(ProviderDecision::Deny(AuthorizationDenial::new( + AuthorizationDenialReason::ProviderDenied, + )))) + } + } + + fn domain(value: u128) -> CommunityId { + CommunityId::from_uuid(Uuid::from_u128(value)) + } + + fn policy(domain: CommunityId) -> DomainAuthorizationPolicy { + let application_limit = + ApplicationLeaseLimit::from_seconds(300).expect("synthetic limit is valid"); + let skew = AuthorizationClockSkew::from_seconds(5).expect("synthetic skew is valid"); + DomainAuthorizationPolicy::from_server_configuration( + domain, + "synthetic-provider.example", + Arc::new(DenyProvider), + EnrollmentMode::Provisioned, + AuthorizationMode::Enforce, + ProviderTimeout::new(Duration::from_secs(1)).expect("synthetic timeout is valid"), + AccessLeasePolicy::new(application_limit, skew), + VerificationStatusPolicy::new(application_limit, skew), + ) + .expect("synthetic policy is valid") + } + + #[test] + fn duplicate_domain_policy_is_rejected_as_ambiguous() { + let exact_domain = domain(1); + let result = DomainProviderSelector::new([policy(exact_domain), policy(exact_domain)]); + assert!(matches!( + result, + Err(DomainPolicyError::AmbiguousDomainPolicy) + )); + } + + #[test] + fn missing_domain_has_no_default_provider_fallback() { + let configured = domain(1); + let missing = domain(2); + let selector = DomainProviderSelector::new([policy(configured)]) + .expect("one exact policy is unambiguous"); + let tenant = TenantContext::resolved(missing, "missing.authorization.example"); + assert!(matches!( + selector.resolve(&tenant), + Err(DomainPolicyError::MissingDomainPolicy) + )); + } + + #[test] + fn exact_server_tenant_selects_its_policy() { + let configured = domain(1); + let selector = DomainProviderSelector::new([policy(configured)]) + .expect("one exact policy is unambiguous"); + let tenant = TenantContext::resolved(configured, "configured.authorization.example"); + let resolved = selector + .resolve(&tenant) + .expect("exact domain is configured"); + assert_eq!(resolved.authorization_domain(), configured); + assert_eq!(resolved.mode(), AuthorizationMode::Enforce); + } +} diff --git a/crates/buzz-relay/src/authorization_runtime/production.rs b/crates/buzz-relay/src/authorization_runtime/production.rs index e0bf5f48a..41cc34c6d 100644 --- a/crates/buzz-relay/src/authorization_runtime/production.rs +++ b/crates/buzz-relay/src/authorization_runtime/production.rs @@ -732,14 +732,13 @@ pub async fn build_from_environment_with_providers( return Ok(None); } validate_provider_coverage(&configured, &providers)?; - if configured - .values() - .any(|mode| *mode != AuthorizationMode::Off) + if configured.values().any(|mode| mode.evaluates_provider()) && state.identity_assertion_provenance().is_none() { return Err(ProductionRuntimeError::AssertionProvenanceMissing); } - let enforcing_domains = authoritative_domains(&configured); + let protected_domains = protected_domains(&configured); + let enforcing_domains = enforcing_domains(&configured); let projection_domains = projection_reconciliation_domains(&configured); if !enforcing_domains.is_empty() && state.corporate_identity.is_none() { return Err(ProductionRuntimeError::VerifierMissing); @@ -747,7 +746,7 @@ pub async fn build_from_environment_with_providers( let clock: SharedAuthorizationClock = Arc::new(SystemAuthorizationClock); let restore_bootstraps = parse_restore_bootstraps( &env::var(RESTORE_BOOTSTRAPS_ENV).unwrap_or_default(), - &enforcing_domains, + &protected_domains, )?; let profile = env::var(PROFILE_ENV).unwrap_or_else(|_| "current-membership-v1".to_owned()); let lease_seconds = parse_positive_seconds(LEASE_SECONDS_ENV, 300)?; @@ -757,11 +756,13 @@ pub async fn build_from_environment_with_providers( let mut policies = Vec::with_capacity(configured.len()); let mut transports = Vec::with_capacity(configured.len()); for (domain, mode) in &configured { - let provider = if *mode == AuthorizationMode::Off { + transports.push(DomainTransportPolicy::from_server_configuration( + *domain, *mode, + )); + if !mode.evaluates_provider() { continue; - } else { - providers.provider_for(*domain)? - }; + } + let provider = providers.provider_for(*domain)?; policies.push(DomainAuthorizationPolicy::from_server_configuration( *domain, profile.clone(), @@ -772,16 +773,6 @@ pub async fn build_from_environment_with_providers( AccessLeasePolicy::new(lease_limit, skew), VerificationStatusPolicy::new(status_limit, skew), )?); - transports.push(DomainTransportPolicy::from_server_configuration( - *domain, *mode, - )); - } - for (domain, mode) in &configured { - if *mode == AuthorizationMode::Off { - transports.push(DomainTransportPolicy::from_server_configuration( - *domain, *mode, - )); - } } let hosts = state.db.usage_community_hosts().await?; let host_map = hosts @@ -804,7 +795,7 @@ pub async fn build_from_environment_with_providers( restore_bootstraps, ) .await?; - activate_enforcing_domains(&state.db, &restore, enforcing_domains.iter().copied()).await?; + activate_protected_domains(&state.db, &restore, protected_domains.iter().copied()).await?; reconcile_audio_admissions_once(&state.db, &restore, enforcing_domains.iter().copied()).await?; let invalidation = AuthorizationInvalidationRuntime::new_with_restore( state.db.clone(), @@ -813,7 +804,7 @@ pub async fn build_from_environment_with_providers( Arc::clone(&restore), ); invalidation - .initialize_domains(enforcing_domains.iter().copied()) + .initialize_domains(protected_domains.iter().copied()) .await?; crate::corporate_identity::reconcile_public_projection_retirements_startup( state, @@ -843,7 +834,7 @@ pub async fn build_from_environment_with_providers( })) } -async fn activate_enforcing_domains( +async fn activate_protected_domains( db: &buzz_db::Db, restore: &Arc, domains: impl IntoIterator, @@ -886,7 +877,7 @@ fn validate_provider_coverage( providers: &ProductionProviderRegistry, ) -> Result<(), ProductionRuntimeError> { for (domain, mode) in configured { - if *mode != AuthorizationMode::Off { + if mode.evaluates_provider() { providers.provider_for(*domain)?; } } @@ -897,10 +888,11 @@ fn validate_activated_domain_configuration( configured: &HashMap, activated: impl IntoIterator, ) -> Result<(), ProductionRuntimeError> { - if activated - .into_iter() - .any(|domain| configured.get(&domain) != Some(&AuthorizationMode::Enforce)) - { + if activated.into_iter().any(|domain| { + !configured + .get(&domain) + .is_some_and(|mode| mode.protects_surfaces()) + }) { return Err(ProductionRuntimeError::ActivatedDomainDowngrade); } Ok(()) @@ -1083,6 +1075,7 @@ fn parse_domains( "shadow" => AuthorizationMode::Shadow, "verify_only" => AuthorizationMode::VerifyOnly, "enforce" => AuthorizationMode::Enforce, + "deny_protected" => AuthorizationMode::DenyProtected, _ => return Err(ProductionRuntimeError::InvalidConfiguration), }; if domains.insert(domain, mode).is_some() { @@ -1092,7 +1085,14 @@ fn parse_domains( Ok(domains) } -fn authoritative_domains(configured: &HashMap) -> Vec { +fn protected_domains(configured: &HashMap) -> Vec { + configured + .iter() + .filter_map(|(domain, mode)| mode.protects_surfaces().then_some(*domain)) + .collect() +} + +fn enforcing_domains(configured: &HashMap) -> Vec { configured .iter() .filter_map(|(domain, mode)| (*mode == AuthorizationMode::Enforce).then_some(*domain)) @@ -1112,7 +1112,7 @@ fn projection_reconciliation_domains( fn parse_restore_bootstraps( raw: &str, - enforcing_domains: &[CommunityId], + protected_domains: &[CommunityId], ) -> Result, ProductionRuntimeError> { let mut anchors = HashMap::new(); for item in raw @@ -1133,8 +1133,8 @@ fn parse_restore_bootstraps( return Err(ProductionRuntimeError::InvalidConfiguration); } } - let mut result = Vec::with_capacity(enforcing_domains.len()); - for domain in enforcing_domains { + let mut result = Vec::with_capacity(protected_domains.len()); + for domain in protected_domains { let bootstrap = anchors .remove(domain) .ok_or(ProductionRuntimeError::RestoreBootstrapMissing)?; @@ -1240,8 +1240,11 @@ mod tests { fn exact_modes_parse_without_a_default() { let first = uuid::Uuid::new_v4(); let second = uuid::Uuid::new_v4(); - let parsed = parse_domains(&format!("{first}:enforce,{second}:verify_only")) - .expect("valid exact domains"); + let third = uuid::Uuid::new_v4(); + let parsed = parse_domains(&format!( + "{first}:enforce,{second}:verify_only,{third}:deny_protected" + )) + .expect("valid exact domains"); assert_eq!( parsed.get(&CommunityId::from_uuid(first)), Some(&AuthorizationMode::Enforce) @@ -1250,6 +1253,10 @@ mod tests { parsed.get(&CommunityId::from_uuid(second)), Some(&AuthorizationMode::VerifyOnly) ); + assert_eq!( + parsed.get(&CommunityId::from_uuid(third)), + Some(&AuthorizationMode::DenyProtected) + ); } #[test] @@ -1265,12 +1272,19 @@ mod tests { let shadow = uuid::Uuid::new_v4(); let verify = uuid::Uuid::new_v4(); let enforce = uuid::Uuid::new_v4(); + let deny = uuid::Uuid::new_v4(); let parsed = parse_domains(&format!( - "{off}:off,{shadow}:shadow,{verify}:verify_only,{enforce}:enforce" + "{off}:off,{shadow}:shadow,{verify}:verify_only,{enforce}:enforce,{deny}:deny_protected" )) .expect("valid exact modes"); + let protected = protected_domains(&parsed) + .into_iter() + .collect::>(); + assert_eq!(protected.len(), 2); + assert!(protected.contains(&CommunityId::from_uuid(enforce))); + assert!(protected.contains(&CommunityId::from_uuid(deny))); assert_eq!( - authoritative_domains(&parsed), + enforcing_domains(&parsed), vec![CommunityId::from_uuid(enforce)] ); let projection_domains = projection_reconciliation_domains(&parsed) @@ -1281,6 +1295,7 @@ mod tests { assert!(!projection_domains.contains(&CommunityId::from_uuid(shadow))); assert!(!projection_domains.contains(&CommunityId::from_uuid(verify))); assert!(projection_domains.contains(&CommunityId::from_uuid(enforce))); + assert!(!projection_domains.contains(&CommunityId::from_uuid(deny))); } #[test] @@ -1304,6 +1319,9 @@ mod tests { let configured = HashMap::from([(activated, AuthorizationMode::Enforce)]); validate_activated_domain_configuration(&configured, [activated]) .expect("exact Enforce configuration preserves one-way activation"); + let configured = HashMap::from([(activated, AuthorizationMode::DenyProtected)]); + validate_activated_domain_configuration(&configured, [activated]) + .expect("deny-protected preserves the protected inventory after activation"); } #[test] @@ -1324,8 +1342,11 @@ mod tests { fn exact_provider_coverage_makes_enforce_constructible_without_fallback() { let enforce = CommunityId::from_uuid(uuid::Uuid::new_v4()); let off = CommunityId::from_uuid(uuid::Uuid::new_v4()); - let configured = parse_domains(&format!("{enforce}:enforce,{off}:off")) - .expect("exact production configuration"); + let deny = CommunityId::from_uuid(uuid::Uuid::new_v4()); + let configured = parse_domains(&format!( + "{enforce}:enforce,{off}:off,{deny}:deny_protected" + )) + .expect("exact production configuration"); assert!(matches!( validate_provider_coverage(&configured, &ProductionProviderRegistry::default()), Err(ProductionRuntimeError::ProviderMissing) @@ -1356,10 +1377,10 @@ mod tests { fn enforce_domain_activation_precedes_snapshot_and_transport_reachability() { let source = include_str!("production.rs"); let activation = source - .find("activate_enforcing_domains(&state.db") + .find("activate_protected_domains(&state.db") .expect("durable activation is part of construction"); let snapshot = source - .find(".initialize_domains(enforcing_domains.iter().copied())") + .find(".initialize_domains(protected_domains.iter().copied())") .expect("invalidation snapshot is initialized"); let transport = source .find("ProtectedTransportRuntime::new(transports, resolver, clock)") diff --git a/crates/buzz-relay/src/authorization_runtime/transport.rs b/crates/buzz-relay/src/authorization_runtime/transport.rs index d497b4fb4..d99661567 100644 --- a/crates/buzz-relay/src/authorization_runtime/transport.rs +++ b/crates/buzz-relay/src/authorization_runtime/transport.rs @@ -33,8 +33,8 @@ pub enum LegacyIdentityLane { /// immutable. This lane can never enroll, reactivate, strengthen, update, /// or retire a binding. ObserveOnly, - /// Enforce: the protected resolver owns admission/finalization and legacy - /// projection must not run. + /// Enforce or DenyProtected: the protected runtime owns the surface and + /// legacy projection must not run. ProtectedEnforce, } @@ -57,7 +57,9 @@ pub const fn legacy_identity_lane_for_mode(mode: Option) -> L Some(AuthorizationMode::Shadow) | Some(AuthorizationMode::VerifyOnly) => { LegacyIdentityLane::ObserveOnly } - Some(AuthorizationMode::Enforce) => LegacyIdentityLane::ProtectedEnforce, + Some(AuthorizationMode::Enforce | AuthorizationMode::DenyProtected) => { + LegacyIdentityLane::ProtectedEnforce + } } } @@ -532,7 +534,7 @@ impl ProtectedTransportRuntime { pub fn enforcing_domains(&self) -> Vec { self.domains .iter() - .filter_map(|(domain, mode)| (*mode == AuthorizationMode::Enforce).then_some(*domain)) + .filter_map(|(domain, mode)| mode.protects_surfaces().then_some(*domain)) .collect() } @@ -555,6 +557,7 @@ impl ProtectedTransportRuntime { let _ = self.resolver.observe(request).await; Ok(ProtectedAuthorization::Legacy) } + AuthorizationMode::DenyProtected => deny_protected_request(request), AuthorizationMode::Enforce => { let resolution = self .resolver @@ -597,6 +600,7 @@ impl ProtectedTransportRuntime { .await .map(Some) .map_err(ProtectedTransportError::Resolution), + Some(AuthorizationMode::DenyProtected) => Err(ProtectedTransportError::DenyProtected), None | Some(AuthorizationMode::Off | AuthorizationMode::Shadow) => Ok(None), } } @@ -613,6 +617,7 @@ impl ProtectedTransportRuntime { AuthorizationMode::Off | AuthorizationMode::Shadow | AuthorizationMode::VerifyOnly => { Ok(ProtectedEnrollmentAuthorization::Legacy) } + AuthorizationMode::DenyProtected => Err(ProtectedTransportError::DenyProtected), AuthorizationMode::Enforce => { if request.capability() != AuthorizationCapability::InviteClaim || request.enrollment_assertion().is_none() @@ -761,9 +766,19 @@ fn authorize_unwired_for_mode( | Some(AuthorizationMode::Shadow) | Some(AuthorizationMode::VerifyOnly) => Ok(ProtectedAuthorization::Legacy), Some(AuthorizationMode::Enforce) => Err(ProtectedTransportError::MissingVerifiedProof), + Some(AuthorizationMode::DenyProtected) => Err(ProtectedTransportError::DenyProtected), } } +fn deny_protected_request( + request: &ProtectedOperationRequest, +) -> Result { + if let Some(cancellation) = request.cancellation() { + cancellation.cancel(); + } + Err(ProtectedTransportError::DenyProtected) +} + impl fmt::Debug for ProtectedTransportRuntime { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { formatter @@ -1093,6 +1108,9 @@ pub enum ProtectedTransportError { /// An enforcing surface did not retain sealed verifier evidence. #[error("protected authorization requires verified transport evidence")] MissingVerifiedProof, + /// The exact domain is in the explicit fail-safe protected-denial mode. + #[error("protected authorization is unavailable in deny-protected mode")] + DenyProtected, /// Resolver denied or could not evaluate current policy. #[error(transparent)] Resolution(#[from] ProtectedResolutionError), diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index 55de214e7..f9e308160 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -6,6 +6,7 @@ use std::sync::atomic::{AtomicU8, Ordering}; use std::sync::Arc; use std::time::Duration; +use async_trait::async_trait; use axum::extract::ws::{Message as WsMessage, WebSocket}; use futures_util::{Sink, SinkExt, StreamExt}; use tokio::sync::{mpsc, Mutex, RwLock}; @@ -23,6 +24,229 @@ use crate::protocol::{ClientMessage, RelayMessage}; use crate::state::{run_registered_community_connection, AppState}; use buzz_pubsub::EventTopic; +/// Fail-closed release check evaluated at the socket acceptance boundary. +pub(crate) trait OutboundReleaseFence: Send + Sync { + fn release(&self) -> bool; +} + +/// Potentially asynchronous release fence retained until socket drain. +#[async_trait] +pub(crate) trait QueuedOutboundReleaseFence: Send + Sync { + async fn release(&self) -> bool; +} + +struct SyncQueuedReleaseFence { + authority: Arc, +} + +#[async_trait] +impl QueuedOutboundReleaseFence for SyncQueuedReleaseFence { + async fn release(&self) -> bool { + self.authority.release() + } +} + +pub(crate) fn queued_local_authority( + authority: Arc, +) -> Arc { + Arc::new(SyncQueuedReleaseFence { authority }) +} + +#[async_trait] +trait ChannelReadAuthoritySource: Send + Sync { + async fn channel_set_read_authorized( + &self, + community_id: buzz_core::tenant::CommunityId, + channel_ids: &[Uuid], + actor: &[u8], + ) -> bool; +} + +#[async_trait] +impl ChannelReadAuthoritySource for buzz_db::Db { + async fn channel_set_read_authorized( + &self, + community_id: buzz_core::tenant::CommunityId, + channel_ids: &[Uuid], + actor: &[u8], + ) -> bool { + buzz_db::Db::channel_set_read_authorized(self, community_id, channel_ids, actor) + .await + .unwrap_or(false) + } +} + +struct ChannelReadReleaseFence { + source: Arc, + community_id: buzz_core::tenant::CommunityId, + channel_ids: Vec, + actor: Vec, + protected: Option>, +} + +#[async_trait] +impl QueuedOutboundReleaseFence for ChannelReadReleaseFence { + async fn release(&self) -> bool { + if self + .protected + .as_ref() + .is_some_and(|authority| authority.revalidate().is_err()) + { + return false; + } + if !self + .source + .channel_set_read_authorized(self.community_id, &self.channel_ids, &self.actor) + .await + { + return false; + } + self.protected + .as_ref() + .is_none_or(|authority| authority.revalidate().is_ok()) + } +} + +/// Retain uncached channel access, plus optional protected identity authority, +/// until the socket writer accepts the queued frame. +pub(crate) fn queued_channel_read_authority( + db: buzz_db::Db, + community_id: buzz_core::tenant::CommunityId, + channel_id: Uuid, + actor: Vec, + protected: Option>, +) -> Arc { + queued_channel_set_read_authority(db, community_id, vec![channel_id], actor, protected) +} + +/// Retain uncached access to every channel that can contribute to one +/// aggregate response until the response is released. +pub(crate) fn queued_channel_set_read_authority( + db: buzz_db::Db, + community_id: buzz_core::tenant::CommunityId, + mut channel_ids: Vec, + actor: Vec, + protected: Option>, +) -> Arc { + channel_ids.sort_unstable(); + channel_ids.dedup(); + Arc::new(ChannelReadReleaseFence { + source: Arc::new(db), + community_id, + channel_ids, + actor, + protected, + }) +} + +/// Evaluate the aggregate read fence synchronously with an HTTP response +/// release. WebSocket callers retain the same fence in their outbound queue. +pub(crate) async fn release_channel_set_read_authority( + db: buzz_db::Db, + community_id: buzz_core::tenant::CommunityId, + channel_ids: Vec, + actor: Vec, + protected: Option>, +) -> bool { + queued_channel_set_read_authority(db, community_id, channel_ids, actor, protected) + .release() + .await +} + +impl OutboundReleaseFence for crate::authorization_runtime::transport::ProtectedAuthorization { + fn release(&self) -> bool { + self.release_fetched(()).is_ok() + } +} + +struct CombinedReleaseFence { + sender: Arc, + recipient: Arc, +} + +#[async_trait] +impl QueuedOutboundReleaseFence for CombinedReleaseFence { + async fn release(&self) -> bool { + self.sender.release().await + && self.recipient.release().await + && self.sender.release().await + && self.recipient.release().await + } +} + +/// One queued data frame with optional authority retained until socket drain. +pub struct OutboundData { + pub(crate) message: WsMessage, + authority: Option>, +} + +impl OutboundData { + pub(crate) fn plain(message: WsMessage) -> Self { + Self { + message, + authority: None, + } + } + + pub(crate) fn protected( + message: WsMessage, + authority: Arc, + ) -> Self { + Self { + message, + authority: Some(queued_local_authority(authority)), + } + } + + pub(crate) fn protected_pair( + message: WsMessage, + sender: Arc, + recipient: Arc, + ) -> Self { + Self { + message, + authority: Some(Arc::new(CombinedReleaseFence { + sender: queued_local_authority(sender), + recipient: queued_local_authority(recipient), + })), + } + } + + pub(crate) fn guarded( + message: WsMessage, + authority: Arc, + ) -> Self { + Self { + message, + authority: Some(authority), + } + } + + pub(crate) fn guarded_pair( + message: WsMessage, + sender: Arc, + recipient: Arc, + ) -> Self { + Self { + message, + authority: Some(Arc::new(CombinedReleaseFence { sender, recipient })), + } + } + + #[cfg(test)] + fn guarded_for_test(message: WsMessage, authority: Arc) -> Self { + Self::guarded(message, Arc::new(SyncQueuedReleaseFence { authority })) + } + + async fn release(self) -> Option { + match self.authority { + Some(authority) if authority.release().await => Some(self.message), + Some(_) => None, + None => Some(self.message), + } + } +} + /// Maximum time a new socket may hold a connection slot without completing NIP-42 auth. const AUTH_TIMEOUT: Duration = Duration::from_secs(5); @@ -59,14 +283,14 @@ pub struct ConnectionState { pub tenant: TenantContext, /// Remote socket address of the client. pub remote_addr: SocketAddr, - /// Optional corporate identity JWT captured from the WebSocket upgrade request. - pub corporate_identity_jwt: Option, + /// Optional direct identity assertion captured with verified provenance. + pub corporate_identity_assertion: Option, /// Current NIP-42 authentication state. pub auth_state: RwLock, /// Active subscriptions keyed by subscription ID. pub subscriptions: ConnectionSubscriptions, /// Sender for outbound data messages (EVENT, NOTICE, OK, etc.). - pub send_tx: mpsc::Sender, + pub send_tx: mpsc::Sender, /// Sender for outbound control frames (Pong, Close). /// Separate channel with priority drain — if this channel fills too, /// the connection is closed (writer is completely stalled). @@ -88,7 +312,43 @@ impl ConnectionState { /// `grace_limit` occurrences log a warning; sustained backpressure /// cancels the connection to prevent unbounded memory growth. pub fn send(&self, msg: String) -> bool { - match self.send_tx.try_send(WsMessage::Text(msg.into())) { + self.send_data(OutboundData::plain(WsMessage::Text(msg.into()))) + } + + /// Queue a terminal text frame on the priority control channel. + /// + /// Callers may cancel immediately after this returns: the send loop drains + /// control frames before emitting the WebSocket close frame. + pub(crate) fn send_terminal(&self, msg: String) -> bool { + self.ctrl_tx.try_send(WsMessage::Text(msg.into())).is_ok() + } + + /// Queue protected output while retaining its guard through socket drain. + pub fn send_protected( + &self, + msg: String, + authority: Arc, + ) -> bool { + self.send_data(OutboundData::protected( + WsMessage::Text(msg.into()), + authority, + )) + } + + /// Queue output behind an arbitrary asynchronous release fence. + pub(crate) fn send_guarded( + &self, + msg: String, + authority: Arc, + ) -> bool { + self.send_data(OutboundData::guarded( + WsMessage::Text(msg.into()), + authority, + )) + } + + fn send_data(&self, msg: OutboundData) -> bool { + match self.send_tx.try_send(msg) { Ok(_) => { // Successful send resets the grace counter. self.backpressure_count.store(0, Ordering::Relaxed); @@ -122,7 +382,7 @@ pub async fn handle_connection( state: Arc, addr: SocketAddr, tenant: TenantContext, - corporate_identity_jwt: Option, + corporate_identity_assertion: Option, ) { let conn_id = Uuid::new_v4(); let cancel = CancellationToken::new(); @@ -144,7 +404,7 @@ pub async fn handle_connection( tenant, conn_id, cancel, - corporate_identity_jwt, + corporate_identity_assertion, ) }, ) @@ -158,7 +418,7 @@ async fn handle_active_connection( tenant: TenantContext, conn_id: Uuid, cancel: CancellationToken, - corporate_identity_jwt: Option, + corporate_identity_assertion: Option, ) { let permit = match state.conn_semaphore.clone().try_acquire_owned() { Ok(p) => p, @@ -170,7 +430,7 @@ async fn handle_active_connection( let challenge = generate_challenge(); - let (tx, rx) = mpsc::channel::(state.config.send_buffer_size); + let (tx, rx) = mpsc::channel::(state.config.send_buffer_size); // Control channel for Pong/Close — small capacity, guaranteed delivery // even when the data buffer is full. let (ctrl_tx, ctrl_rx) = mpsc::channel::(8); @@ -182,7 +442,7 @@ async fn handle_active_connection( conn_id, tenant, remote_addr: addr, - corporate_identity_jwt, + corporate_identity_assertion, auth_state: RwLock::new(AuthState::Pending { challenge: challenge.clone(), }), @@ -197,13 +457,13 @@ async fn handle_active_connection( info!(conn_id = %conn_id, addr = %addr, "WebSocket connection established"); metrics::counter!( "buzz_ws_connections_total", - "community" => conn.tenant.host().to_owned() + "community" => crate::metrics::community_label(conn.tenant.community()) ) .increment(1); let challenge_msg = RelayMessage::auth_challenge(&challenge); if tx - .send(WsMessage::Text(challenge_msg.into())) + .send(OutboundData::plain(WsMessage::Text(challenge_msg.into()))) .await .is_err() { @@ -310,7 +570,7 @@ async fn handle_active_connection( /// treat a full control channel as terminal (Bug 7 fix). async fn send_loop( ws_send: futures_util::stream::SplitSink, - data_rx: mpsc::Receiver, + data_rx: mpsc::Receiver, ctrl_rx: mpsc::Receiver, cancel: CancellationToken, ) { @@ -319,7 +579,7 @@ async fn send_loop( async fn send_loop_inner( mut ws_send: S, - mut data_rx: mpsc::Receiver, + mut data_rx: mpsc::Receiver, mut ctrl_rx: mpsc::Receiver, cancel: CancellationToken, ) where @@ -357,16 +617,30 @@ async fn send_loop_inner( break; } } - Some(msg) = data_rx.recv() => { + Some(queued) = data_rx.recv() => { let mut batched = 1usize; - if ws_send.feed(msg).await.is_err() { + if !sink_ready_before_cancellation(&mut ws_send, &cancel).await { + break; + } + let Some(msg) = queued.release().await else { + cancel.cancel(); + break; + }; + if std::pin::Pin::new(&mut ws_send).start_send(msg).is_err() { break; } while batched < MAX_WS_SEND_BATCH { match data_rx.try_recv() { Ok(next) => { - if ws_send.feed(next).await.is_err() { + if !sink_ready_before_cancellation(&mut ws_send, &cancel).await { + return; + } + let Some(next) = next.release().await else { + cancel.cancel(); + return; + }; + if std::pin::Pin::new(&mut ws_send).start_send(next).is_err() { return; } batched += 1; @@ -385,6 +659,19 @@ async fn send_loop_inner( } } +async fn sink_ready_before_cancellation(ws_send: &mut S, cancel: &CancellationToken) -> bool +where + S: Sink + Unpin, +{ + tokio::select! { + biased; + _ = cancel.cancelled() => false, + result = std::future::poll_fn(|cx| std::pin::Pin::new(&mut *ws_send).poll_ready(cx)) => { + result.is_ok() + } + } +} + /// 3 missed pongs → disconnect. /// /// Sends Ping through the control channel so it isn't blocked by a full @@ -703,6 +990,7 @@ fn topic_for_subscription(channel_id: Option) -> EventTopic { #[cfg(test)] mod tests { use super::*; + use std::sync::atomic::{AtomicBool, AtomicUsize}; use std::sync::{Arc, Mutex}; #[derive(Debug, Default)] @@ -777,6 +1065,106 @@ mod tests { } } + struct ScriptedFence(AtomicBool); + + impl OutboundReleaseFence for ScriptedFence { + fn release(&self) -> bool { + self.0.load(Ordering::SeqCst) + } + } + + struct CountingQueuedFence(AtomicUsize); + + #[async_trait] + impl QueuedOutboundReleaseFence for CountingQueuedFence { + async fn release(&self) -> bool { + self.0.fetch_add(1, Ordering::SeqCst); + true + } + } + + #[tokio::test] + async fn combined_release_rechecks_both_sides_after_async_boundaries() { + let sender = Arc::new(CountingQueuedFence(AtomicUsize::new(0))); + let recipient = Arc::new(CountingQueuedFence(AtomicUsize::new(0))); + let fence = CombinedReleaseFence { + sender: sender.clone(), + recipient: recipient.clone(), + }; + + assert!(fence.release().await); + assert_eq!(sender.0.load(Ordering::SeqCst), 2); + assert_eq!(recipient.0.load(Ordering::SeqCst), 2); + } + + struct ScriptedChannelAuthority { + allowed: AtomicBool, + checked: Mutex>, + } + + #[async_trait] + impl ChannelReadAuthoritySource for ScriptedChannelAuthority { + async fn channel_set_read_authorized( + &self, + _community_id: buzz_core::tenant::CommunityId, + channel_ids: &[Uuid], + _actor: &[u8], + ) -> bool { + self.checked + .lock() + .expect("scripted channel checks poisoned") + .extend_from_slice(channel_ids); + self.allowed.load(Ordering::SeqCst) + } + } + + struct ReadinessBarrierSink { + ready: Arc, + polled: Arc, + waker: Arc>>, + state: Arc>, + } + + impl Sink for ReadinessBarrierSink { + type Error = std::io::Error; + + fn poll_ready( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + if self.ready.load(Ordering::SeqCst) { + std::task::Poll::Ready(Ok(())) + } else { + *self.waker.lock().expect("barrier waker poisoned") = Some(cx.waker().clone()); + self.polled.notify_one(); + std::task::Poll::Pending + } + } + + fn start_send(self: std::pin::Pin<&mut Self>, item: WsMessage) -> Result<(), Self::Error> { + self.state + .lock() + .expect("barrier sink poisoned") + .messages + .push(item); + Ok(()) + } + + fn poll_flush( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(())) + } + + fn poll_close( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + self.poll_flush(cx) + } + } + fn text_payloads(messages: &[WsMessage]) -> Vec { messages .iter() @@ -806,7 +1194,9 @@ mod tests { let (_ctrl_tx, ctrl_rx) = mpsc::channel(1); for i in 0..5 { data_tx - .send(WsMessage::Text(format!("data-{i}").into())) + .send(OutboundData::plain(WsMessage::Text( + format!("data-{i}").into(), + ))) .await .expect("queue data frame"); } @@ -822,12 +1212,208 @@ mod tests { ); } + #[tokio::test] + async fn protected_frame_revalidates_after_sink_readiness() { + let ready = Arc::new(AtomicBool::new(false)); + let polled = Arc::new(tokio::sync::Notify::new()); + let waker = Arc::new(Mutex::new(None)); + let state = Arc::new(Mutex::new(MockSinkState::default())); + let sink = ReadinessBarrierSink { + ready: Arc::clone(&ready), + polled: Arc::clone(&polled), + waker: Arc::clone(&waker), + state: Arc::clone(&state), + }; + let fence = Arc::new(ScriptedFence(AtomicBool::new(true))); + let (data_tx, data_rx) = mpsc::channel(1); + let (_ctrl_tx, ctrl_rx) = mpsc::channel(1); + let cancel = CancellationToken::new(); + data_tx + .send(OutboundData::guarded_for_test( + WsMessage::Text("protected".into()), + fence.clone(), + )) + .await + .expect("queue protected frame"); + drop(data_tx); + + let task = tokio::spawn(send_loop_inner(sink, data_rx, ctrl_rx, cancel.clone())); + polled.notified().await; + fence.0.store(false, Ordering::SeqCst); + ready.store(true, Ordering::SeqCst); + waker + .lock() + .expect("barrier waker poisoned") + .take() + .expect("poll_ready registered a waker") + .wake(); + task.await.expect("send loop joins"); + + assert!(cancel.is_cancelled()); + assert!( + state + .lock() + .expect("barrier sink poisoned") + .messages + .is_empty(), + "authority loss while readiness is pending must prevent start_send" + ); + } + + #[tokio::test] + async fn count_release_rechecks_every_contributing_channel() { + let first = Uuid::new_v4(); + let second = Uuid::new_v4(); + let source = Arc::new(ScriptedChannelAuthority { + allowed: AtomicBool::new(true), + checked: Mutex::new(Vec::new()), + }); + let fence = ChannelReadReleaseFence { + source: source.clone(), + community_id: buzz_core::tenant::CommunityId::from_uuid(Uuid::new_v4()), + channel_ids: vec![first, second], + actor: vec![7; 32], + protected: None, + }; + + assert!(fence.release().await); + assert_eq!( + *source + .checked + .lock() + .expect("scripted channel checks poisoned"), + vec![first, second] + ); + } + + #[tokio::test] + async fn net_http_004_count_release_denies_authority_loss_after_fetch() { + let source = Arc::new(ScriptedChannelAuthority { + allowed: AtomicBool::new(true), + checked: Mutex::new(Vec::new()), + }); + let fence = ChannelReadReleaseFence { + source: source.clone(), + community_id: buzz_core::tenant::CommunityId::from_uuid(Uuid::new_v4()), + channel_ids: vec![Uuid::new_v4()], + actor: vec![9; 32], + protected: None, + }; + + // The query has completed. A membership removal or an open-to-private + // transition now makes the authoritative DB check return false. + source.allowed.store(false, Ordering::SeqCst); + assert!(!fence.release().await); + } + + /// NET-WS-009: a COUNT queued while access is valid must not become + /// visible if membership or channel visibility changes while the socket + /// is waiting for sink readiness. + #[tokio::test] + async fn net_ws_009_count_release_denies_authority_loss_before_socket_acceptance() { + let ready = Arc::new(AtomicBool::new(false)); + let polled = Arc::new(tokio::sync::Notify::new()); + let waker = Arc::new(Mutex::new(None)); + let state = Arc::new(Mutex::new(MockSinkState::default())); + let sink = ReadinessBarrierSink { + ready: Arc::clone(&ready), + polled: Arc::clone(&polled), + waker: Arc::clone(&waker), + state: Arc::clone(&state), + }; + let source = Arc::new(ScriptedChannelAuthority { + allowed: AtomicBool::new(true), + checked: Mutex::new(Vec::new()), + }); + let release = Arc::new(ChannelReadReleaseFence { + source: source.clone(), + community_id: buzz_core::tenant::CommunityId::from_uuid(Uuid::new_v4()), + channel_ids: vec![Uuid::new_v4()], + actor: vec![10; 32], + protected: None, + }); + let (data_tx, data_rx) = mpsc::channel(1); + let (_ctrl_tx, ctrl_rx) = mpsc::channel(1); + let cancel = CancellationToken::new(); + data_tx + .send(OutboundData::guarded( + WsMessage::Text(RelayMessage::count("count-race", 1).into()), + release, + )) + .await + .expect("queue protected COUNT"); + drop(data_tx); + + let task = tokio::spawn(send_loop_inner(sink, data_rx, ctrl_rx, cancel.clone())); + polled.notified().await; + source.allowed.store(false, Ordering::SeqCst); + ready.store(true, Ordering::SeqCst); + waker + .lock() + .expect("barrier waker poisoned") + .take() + .expect("poll_ready registered a waker") + .wake(); + task.await.expect("send loop joins"); + + assert!(cancel.is_cancelled()); + assert!( + state + .lock() + .expect("barrier sink poisoned") + .messages + .is_empty(), + "COUNT must not reach start_send after its channel authority is lost" + ); + } + + /// O4-EXP-SESSION-001: session expiry wins over a COUNT that has been + /// computed and queued but has not yet crossed the socket boundary. + #[tokio::test(start_paused = true)] + async fn o4_exp_session_001_count_is_suppressed_when_deadline_precedes_emission() { + let ready = Arc::new(AtomicBool::new(false)); + let polled = Arc::new(tokio::sync::Notify::new()); + let waker = Arc::new(Mutex::new(None)); + let state = Arc::new(Mutex::new(MockSinkState::default())); + let sink = ReadinessBarrierSink { + ready: Arc::clone(&ready), + polled: Arc::clone(&polled), + waker: Arc::clone(&waker), + state: Arc::clone(&state), + }; + let (data_tx, data_rx) = mpsc::channel(1); + let (_ctrl_tx, ctrl_rx) = mpsc::channel(1); + let cancel = CancellationToken::new(); + data_tx + .send(OutboundData::plain(WsMessage::Text( + RelayMessage::count("expiry-race", 1).into(), + ))) + .await + .expect("queue COUNT before expiry"); + drop(data_tx); + + let task = tokio::spawn(send_loop_inner(sink, data_rx, ctrl_rx, cancel.clone())); + polled.notified().await; + tokio::time::advance(Duration::from_millis(500)).await; + cancel.cancel(); + task.await.expect("send loop joins"); + + assert!(cancel.is_cancelled()); + let messages = &state.lock().expect("barrier sink poisoned").messages; + assert!( + messages + .iter() + .all(|message| !matches!(message, WsMessage::Text(_))), + "expired session must not emit its queued COUNT" + ); + } + #[tokio::test] async fn send_loop_batch_one_preserves_single_frame_flush_behavior() { let (data_tx, data_rx) = mpsc::channel(1); let (_ctrl_tx, ctrl_rx) = mpsc::channel(1); data_tx - .send(WsMessage::Text("single".into())) + .send(OutboundData::plain(WsMessage::Text("single".into()))) .await .expect("queue data frame"); @@ -844,11 +1430,11 @@ mod tests { let (data_tx, data_rx) = mpsc::channel(MAX_WS_SEND_BATCH); let (ctrl_tx, ctrl_rx) = mpsc::channel(1); data_tx - .send(WsMessage::Text("data-0".into())) + .send(OutboundData::plain(WsMessage::Text("data-0".into()))) .await .expect("queue data frame"); data_tx - .send(WsMessage::Text("data-1".into())) + .send(OutboundData::plain(WsMessage::Text("data-1".into()))) .await .expect("queue data frame"); ctrl_tx @@ -905,4 +1491,35 @@ mod tests { "Close is sent only after the reason frame is flushed" ); } + + #[tokio::test] + async fn protected_count_denial_is_visible_before_terminal_close() { + let (_data_tx, data_rx) = mpsc::channel(1); + let (ctrl_tx, ctrl_rx) = mpsc::channel(1); + ctrl_tx + .send(WsMessage::Text( + RelayMessage::closed( + "deny-count", + "auth-required: protected authorization denied", + ) + .into(), + )) + .await + .expect("queue protected COUNT denial"); + + let cancel = CancellationToken::new(); + cancel.cancel(); + let (sink, state) = MockSink::new(None); + send_loop_inner(sink, data_rx, ctrl_rx, cancel).await; + + let state = state.lock().expect("mock sink poisoned"); + assert_eq!(state.messages.len(), 2); + assert!(matches!( + &state.messages[0], + WsMessage::Text(text) + if text.as_str().contains("deny-count") + && text.as_str().contains("protected authorization denied") + )); + assert!(matches!(state.messages[1], WsMessage::Close(_))); + } } diff --git a/crates/buzz-relay/src/handlers/auth.rs b/crates/buzz-relay/src/handlers/auth.rs index 1aa79aafe..10cb7ad6d 100644 --- a/crates/buzz-relay/src/handlers/auth.rs +++ b/crates/buzz-relay/src/handlers/auth.rs @@ -91,6 +91,28 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: Ok(mut auth_ctx) => { let pubkey = auth_ctx.pubkey; + if state + .protected_transport() + .and_then(|runtime| runtime.mode_for_domain(conn.tenant.community())) + == Some( + crate::authorization_runtime::finalization::AuthorizationMode::DenyProtected, + ) + { + metrics::counter!( + "buzz_auth_failures_total", + "reason" => "deny_protected" + ) + .increment(1); + *conn.auth_state.write().await = AuthState::Failed; + conn.send(RelayMessage::ok( + &event_id_hex, + false, + "auth-required: protected authorization unavailable", + )); + conn.cancel.cancel(); + return; + } + // Community ban gate (NIP-42 seam). Runs immediately after auth // verification succeeds and before the allowlist and relay-membership // gates, per COMMUNITY_MODERATION_PLAN.md §0 decision 4 and the diff --git a/crates/buzz-relay/src/protected_surface.rs b/crates/buzz-relay/src/protected_surface.rs index 6b840f38e..4223d934e 100644 --- a/crates/buzz-relay/src/protected_surface.rs +++ b/crates/buzz-relay/src/protected_surface.rs @@ -755,8 +755,8 @@ pub enum EffectPermitError { /// The closed identifier has no registry row. #[error("protected effect is not classified")] Unclassified, - /// The registered effect is deliberately unavailable in Enforce. - #[error("protected effect is unavailable in enforce mode")] + /// The registered effect is deliberately unavailable in protected mode. + #[error("protected effect is unavailable in protected mode")] Unavailable(UnavailableReason), } @@ -769,7 +769,7 @@ pub fn require_effect_permit( .iter() .find(|surface| surface.id == id) .ok_or(EffectPermitError::Unclassified)?; - if mode == Some(AuthorizationMode::Enforce) { + if mode.is_some_and(AuthorizationMode::protects_surfaces) { if let EnforceDisposition::DenyBeforeEffect(reason) = surface.enforce { return Err(EffectPermitError::Unavailable(reason)); } diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index 7aaa3d170..d17e7c3f0 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -33,11 +33,13 @@ pub mod action_sink; pub mod error; pub mod executor; +pub mod mutation_gate; pub mod schema; pub use action_sink::{ActionSink, ActionSinkError}; pub use error::{PartialProgress, WorkflowError}; pub use executor::ExecutionResult; +pub use mutation_gate::MutationGate; pub use schema::{ActionDef, Step, TriggerDef, WorkflowDef}; use std::collections::HashMap; @@ -87,6 +89,9 @@ pub struct WorkflowEngine { /// Action sink for executing side-effects (SendMessage, etc.). /// Late-initialized via [`set_action_sink`] after `AppState` construction. pub(crate) action_sink: OnceLock>, + /// Provider-neutral gate evaluated before every mutation or external effect. + /// Late-initialized by the embedding relay after `AppState` construction. + pub(crate) mutation_gate: OnceLock>, /// Short-TTL cache for the per-event enabled-workflow lookup, keyed /// `(community_id, channel_id)`. Most channels have no workflows, so this /// removes one SELECT from nearly every ingested event. @@ -115,6 +120,7 @@ impl WorkflowEngine { run_semaphore, last_fired: DashMap::new(), action_sink: OnceLock::new(), + mutation_gate: OnceLock::new(), workflow_cache: moka::sync::Cache::builder() .max_capacity(10_000) .time_to_live(std::time::Duration::from_secs(10)) @@ -180,6 +186,38 @@ impl WorkflowEngine { } } + /// Set the workflow mutation gate. Called once by an embedding relay. + /// + /// # Panics + /// Panics if called more than once. + pub fn set_mutation_gate(&self, gate: Arc) { + if self.mutation_gate.set(gate).is_err() { + panic!("mutation_gate already initialized"); + } + } + + /// Require current authority before a workflow mutation or external effect. + /// + /// A standalone engine with no installed gate preserves legacy behavior. + /// Once an embedding relay installs a gate, every engine-owned mutation + /// door calls this method before touching durable or external state. + pub(crate) fn require_mutation(&self, community_id: CommunityId) -> Result<(), WorkflowError> { + mutation_gate::require_configured_mutation( + self.mutation_gate.get().map(AsRef::as_ref), + community_id, + ) + } + + pub(crate) fn require_outbound_webhook( + &self, + community_id: CommunityId, + ) -> Result<(), WorkflowError> { + mutation_gate::require_configured_outbound_webhook( + self.mutation_gate.get().map(AsRef::as_ref), + community_id, + ) + } + /// Get the action sink reference. /// /// Returns `Err(WorkflowError)` if the sink has not been initialized via @@ -217,6 +255,13 @@ impl WorkflowEngine { result: Result, existing_trace: Option>, ) { + if let Err(error) = self.require_mutation(community_id) { + tracing::warn!( + run_id = %run_id, + "Skipping workflow finalization because mutation authority is unavailable: {error}" + ); + return; + } let prefix = existing_trace.unwrap_or_default(); match result { @@ -395,6 +440,14 @@ impl WorkflowEngine { continue; } + if let Err(error) = self.require_mutation(community_id) { + tracing::warn!( + workflow_id = %workflow.id, + "Skipping workflow because mutation authority is unavailable: {error}" + ); + continue; + } + let trigger_event_id_bytes = event.event.id.as_bytes().to_vec(); let run_id = match self .db @@ -608,6 +661,14 @@ impl WorkflowEngine { continue; } + if let Err(error) = self.require_mutation(community_id) { + tracing::warn!( + workflow_id = %workflow.id, + "Cron tick: skipping workflow because mutation authority is unavailable: {error}" + ); + continue; + } + // Durable at-most-once claim — the cross-pod fire boundary. // The loser receives `None` and skips BEFORE any run creation or // side effect. `community_id` is the workflow row's own diff --git a/crates/buzz-workflow/src/mutation_gate.rs b/crates/buzz-workflow/src/mutation_gate.rs new file mode 100644 index 000000000..bed866614 --- /dev/null +++ b/crates/buzz-workflow/src/mutation_gate.rs @@ -0,0 +1,77 @@ +//! Provider-neutral admission gate for workflow mutations and external effects. + +use buzz_core::tenant::CommunityId; + +use crate::WorkflowError; + +/// Server-owned gate evaluated before every workflow mutation or external effect. +/// +/// The workflow engine deliberately knows nothing about identity providers, +/// leases, or deployment configuration. A relay can install a gate that denies +/// an authorization domain until it has a transaction-owning executor. When no +/// gate is installed, the standalone engine preserves its legacy behavior. +pub trait MutationGate: Send + Sync { + /// Require current authority for one server-resolved authorization domain. + fn require_mutation(&self, community_id: CommunityId) -> Result<(), WorkflowError>; + + /// Require authority for an outbound network effect. + fn require_outbound_webhook(&self, community_id: CommunityId) -> Result<(), WorkflowError> { + self.require_mutation(community_id) + } +} + +pub(crate) fn require_configured_mutation( + gate: Option<&dyn MutationGate>, + community_id: CommunityId, +) -> Result<(), WorkflowError> { + match gate { + Some(gate) => gate.require_mutation(community_id), + None => Ok(()), + } +} + +pub(crate) fn require_configured_outbound_webhook( + gate: Option<&dyn MutationGate>, + community_id: CommunityId, +) -> Result<(), WorkflowError> { + match gate { + Some(gate) => gate.require_outbound_webhook(community_id), + None => Ok(()), + } +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + + use super::*; + + struct DenyGate(AtomicUsize); + + impl MutationGate for DenyGate { + fn require_mutation(&self, _community_id: CommunityId) -> Result<(), WorkflowError> { + self.0.fetch_add(1, Ordering::SeqCst); + Err(WorkflowError::Unauthorized( + "synthetic mutation denial".into(), + )) + } + } + + #[test] + fn absent_gate_preserves_legacy_and_configured_denial_is_authoritative() { + let community_id = CommunityId::from_uuid(uuid::Uuid::from_u128(1)); + assert!(require_configured_mutation(None, community_id).is_ok()); + + let gate = DenyGate(AtomicUsize::new(0)); + assert!(matches!( + require_configured_mutation(Some(&gate), community_id), + Err(WorkflowError::Unauthorized(_)) + )); + assert_eq!(gate.0.load(Ordering::SeqCst), 1); + assert!(matches!( + require_configured_outbound_webhook(Some(&gate), community_id), + Err(WorkflowError::Unauthorized(_)) + )); + assert_eq!(gate.0.load(Ordering::SeqCst), 2); + } +}