From d3293885ac657e8ce96f85801bd14be7961900fc Mon Sep 17 00:00:00 2001 From: Cea Stapleton Cordasco <261786559+cea-block@users.noreply.github.com> Date: Tue, 4 Aug 2026 11:43:41 -0500 Subject: [PATCH] feat(auth): install typed provider provenance Canonical O4 slice: 09_provider_installation Final-tree source: e45553ea65b66631771c16908513f983d3c0b881 Signed-off-by: Cea Stapleton Cordasco <261786559+cea-block@users.noreply.github.com> --- crates/buzz-auth/src/lib.rs | 36 +- crates/buzz-auth/src/provider/mod.rs | 177 +- crates/buzz-auth/src/provider/tests.rs | 15 +- .../src/authorization_runtime/ephemeral.rs | 406 +++++ .../src/authorization_runtime/production.rs | 1439 +++++++++++++++++ crates/buzz-relay/src/state.rs | 649 +++++++- 6 files changed, 2697 insertions(+), 25 deletions(-) create mode 100644 crates/buzz-relay/src/authorization_runtime/ephemeral.rs create mode 100644 crates/buzz-relay/src/authorization_runtime/production.rs diff --git a/crates/buzz-auth/src/lib.rs b/crates/buzz-auth/src/lib.rs index 1699555831..c70c7ec0e5 100644 --- a/crates/buzz-auth/src/lib.rs +++ b/crates/buzz-auth/src/lib.rs @@ -17,10 +17,18 @@ /// Channel access checking trait and helpers. pub mod access; +/// Complete Blossom operation authentication verification. +pub mod blossom; /// Versioned, transport-neutral authorization context. pub mod context; /// Authentication error types. pub mod error; +/// Trusted-workspace adapter for sealed verifier and binding evidence. +pub mod evidence_adapter; +/// Federated provider-evidence finalization. +pub mod finalization; +/// Bounded, versioned authorization leases. +pub mod lease; /// NIP-42 challenge–response authentication. pub mod nip42; /// NIP-98 HTTP Auth verification (kind:27235). @@ -44,10 +52,25 @@ pub use context::{ DelegationCapability, DelegationExpiry, DirectBindingResolutionSink, EnrollmentMode, ExistingBindingResolutionSink, FederatedAuthorityAdapter, FederatedAuthorization, FederatedIdentityRequirement, FederatedPrincipal, NostrAuthority, ResolvedFederatedPolicy, - VerifiedFederatedAssertion, VerifiedKeyAttestation, VerifiedNostrProof, VerifiedOwnerAdmission, + VerifiedFederatedAssertion, VerifiedKeyAttestation, VerifiedNostrProof, + VerifiedOperationBinding, VerifiedOperationBindingKind, VerifiedOwnerAdmission, VerifiedTransportDelegation, VersionedBindingRef, }; pub use error::AuthError; +pub use evidence_adapter::{ + ActiveBindingResolution, EvidenceAdapterError, VerifiedDelegationOutput, + VerifiedEvidenceAdapter, +}; +pub use finalization::{AuthorizationFinalizer, FinalizationError, VerificationOnlyDisposition}; +pub use lease::{ + AccessLeasePolicy, ApplicationLeaseLimit, AuthorizationClock, AuthorizationClockError, + AuthorizationClockSkew, AuthorizationLease, AuthorizationLeaseValidator, + AuthorizationOperationGuard, AuthorizationTime, BindingLeaseBound, LeaseIssueError, + LeasePolicyError, LeaseRenewalAction, LeaseRenewalLeadTime, LeaseRenewalSchedule, + LeaseUseRequirement, LeaseValidationError, LeaseVersion, SharedAuthorizationClock, + SystemAuthorizationClock, VerificationStatusPolicy, MAX_APPLICATION_LEASE_SECONDS, + MAX_AUTHORIZATION_CLOCK_SKEW_SECONDS, MAX_LEASE_RENEWAL_LEAD_SECONDS, +}; pub use nip42::{generate_challenge, verify_nip42_event}; pub use nip98::verify_nip98_event; pub use nip98_replay::{ @@ -55,13 +78,14 @@ pub use nip98_replay::{ MAX_REPLAY_TTL_SECS, }; pub use provider::{ - AuthorizationAuthority, AuthorizationCapability, AuthorizationClock, AuthorizationDenial, + resolve_authorization, AuthorizationAuthority, AuthorizationCapability, + AuthorizationClock as ProviderAuthorizationClock, AuthorizationDenial, AuthorizationDenialReason, AuthorizationOutcome, AuthorizationProfileId, AuthorizationProvider, AuthorizationProviderFuture, AuthorizationRequest, AuthorizationRuntime, CapabilitySet, - CapabilitySnapshot, DecisionSource, PolicyVersion, ProviderAllow, ProviderAllowReason, - ProviderAuthorizationError, ProviderContractError, ProviderDecision, ProviderTimeout, - ProviderUnavailable, ProviderUnavailableReason, RetryAfter, MAX_PROVIDER_FRESHNESS_SECONDS, - MAX_PROVIDER_TIMEOUT, + CapabilitySnapshot, DecisionSource, OwnerAdmissionError, PolicyVersion, ProviderAllow, + ProviderAllowReason, ProviderAuthorizationError, ProviderContractError, ProviderDecision, + ProviderTimeout, ProviderUnavailable, ProviderUnavailableReason, RetryAfter, + MAX_PROVIDER_FRESHNESS_SECONDS, MAX_PROVIDER_TIMEOUT, }; pub use rate_limit::{ ip_rate_limit_key, rate_limit_key, LimitType, RateLimitConfig, RateLimitResult, RateLimiter, diff --git a/crates/buzz-auth/src/provider/mod.rs b/crates/buzz-auth/src/provider/mod.rs index 7b3e032962..77fc26dc9f 100644 --- a/crates/buzz-auth/src/provider/mod.rs +++ b/crates/buzz-auth/src/provider/mod.rs @@ -15,10 +15,10 @@ use crate::context::{ authority::{resolve_direct_binding, resolve_existing_binding}, resolve_current_federated_policy, AdmissionExpiry, AssertionTransport, AuthContext, AuthContextError, AuthContextInput, AuthMethod, AuthTransport, AuthoritativeBindingResolution, - AuthoritativeFederatedResolution, AuthorityAdapterError, BindingVersion, + AuthoritativeFederatedResolution, AuthorityAdapterError, AuthorizationReason, BindingVersion, CapabilityFinalizationSeal, FederatedAuthorityAdapter, FederatedPolicyStamp, FederatedPrincipal, ResolvedFederatedPolicy, VerifiedFederatedAssertion, VerifiedNostrProof, - VerifiedOwnerAdmission, + VerifiedOwnerAdmission, VersionedBindingRef, }; const MAX_OPAQUE_ID_BYTES: usize = 256; @@ -68,6 +68,11 @@ impl fmt::Debug for AuthorizationCapability { pub struct CapabilitySet(Vec); impl CapabilitySet { + /// Build the exact one-capability request used by on-demand authorization. + pub fn single(capability: AuthorizationCapability) -> Self { + Self(vec![capability]) + } + /// Build a non-empty set, sorting and removing duplicate capabilities. pub fn new( mut capabilities: Vec, @@ -91,6 +96,11 @@ impl CapabilitySet { .iter() .all(|capability| self.0.binary_search(capability).is_ok()) } + + /// Return whether the set includes one exact portable capability. + pub fn contains(&self, capability: AuthorizationCapability) -> bool { + self.0.binary_search(&capability).is_ok() + } } impl fmt::Debug for CapabilitySet { @@ -124,6 +134,10 @@ impl AuthorizationProfileId { } Ok(Self(value)) } + #[cfg(test)] + pub(crate) fn new(value: impl Into) -> Result { + Self::from_server_configuration(value) + } /// Exact profile identifier for provider routing. pub fn as_str(&self) -> &str { &self.0 @@ -354,6 +368,90 @@ impl AuthorizationRequest { let Some(delegation) = proof.verified_delegation() else { return Err(ProviderContractError::DelegationRequired); }; + if !delegation.capability().is_transport_wide() { + return Err(ProviderContractError::UnsupportedDelegationScope); + } + if delegation.owner_pubkey() != owner.bound_pubkey() { + return Err(ProviderContractError::DelegatedOwnerMismatch); + } + if delegation + .expires_at() + .is_some_and(|bound| bound.is_expired_at(now_unix_seconds)) + { + return Err(ProviderContractError::DelegationExpired); + } + if owner + .expires_at() + .is_some_and(|bound| bound.is_expired_at(now_unix_seconds)) + { + return Err(ProviderContractError::BindingExpired); + } + let evidence_valid_from = federated_policy.stamp().effective_from(); + let mut evidence_valid_until = federated_policy.stamp().effective_until(); + if let Some(delegation) = delegation.expires_at() { + evidence_valid_until = evidence_valid_until.min(delegation.unix_seconds()); + } + if let Some(binding) = owner.expires_at() { + evidence_valid_until = evidence_valid_until.min(binding.unix_seconds()); + } + Ok(Self { + authorization_domain: proof.authorization_domain(), + transport: proof.authorized_transport(), + actor_pubkey: proof.actor_pubkey(), + proof_method: proof.proof_method(), + authority: AuthorizationAuthority::Delegated { + owner_pubkey: owner.bound_pubkey(), + binding_id: owner.binding_id(), + binding_version: owner.binding_version(), + }, + principal: owner.principal().clone(), + key_attested: false, + assertion_transport: None, + assertion_not_before: None, + assertion_expires_at: None, + federated_policy: federated_policy.into_stamp(), + requested_capabilities, + correlation_id, + decision_source: DecisionSource::DelegatedOwnerBinding, + evidence_valid_from, + evidence_valid_until, + }) + } + + /// Build a delegated request from a sealed active binding-store record. + /// + /// The evidence adapter can create this record only from typed current + /// storage output. Enrolled-in-request bindings are rejected so delegated + /// authorization retains O3's existing-active requirement. + pub fn delegated_from_active_binding( + proof: &VerifiedNostrProof, + owner: &VersionedBindingRef, + federated_policy: ResolvedFederatedPolicy, + requested_capabilities: CapabilitySet, + correlation_id: Uuid, + now_unix_seconds: u64, + ) -> Result { + if correlation_id.is_nil() { + return Err(ProviderContractError::InvalidCorrelationId); + } + validate_federated_policy( + &federated_policy, + proof.authorization_domain(), + correlation_id, + now_unix_seconds, + )?; + if proof.authorization_domain() != owner.authorization_domain() { + return Err(ProviderContractError::AuthorizationDomainMismatch); + } + if owner.authorization_reason() != AuthorizationReason::ExistingBinding { + return Err(ProviderContractError::DelegatedBindingNotExistingActive); + } + let Some(delegation) = proof.verified_delegation() else { + return Err(ProviderContractError::DelegationRequired); + }; + if !delegation.capability().is_transport_wide() { + return Err(ProviderContractError::UnsupportedDelegationScope); + } if delegation.owner_pubkey() != owner.bound_pubkey() { return Err(ProviderContractError::DelegatedOwnerMismatch); } @@ -1133,7 +1231,6 @@ impl CapabilitySnapshot { pub const fn reason(&self) -> ProviderAllowReason { self.reason } - /// Consume a direct capability decision and finalize authoritative context. /// /// The current enrollment policy is reread after provider I/O, then the @@ -1241,7 +1338,7 @@ impl CapabilitySnapshot { { return Err(ProviderContractError::CapabilityBindingChanged.into()); } - let admission = VerifiedOwnerAdmission::new( + let admission = VerifiedOwnerAdmission::from_capability_snapshot( self.authorization_domain, self.principal, AdmissionExpiry::new(self.effective_until)?, @@ -1363,6 +1460,38 @@ impl CapabilitySnapshot { } Ok(()) } + + /// Derive current delegated-owner admission for one exact active binding. + pub fn verified_owner_admission( + &self, + owner: &VersionedBindingRef, + ) -> Result { + if self.decision_source != DecisionSource::DelegatedOwnerBinding { + return Err(OwnerAdmissionError::NotDelegatedSnapshot); + } + if self.authorization_domain != owner.authorization_domain() { + return Err(OwnerAdmissionError::AuthorizationDomainMismatch); + } + if self.owner_pubkey != Some(owner.bound_pubkey()) { + return Err(OwnerAdmissionError::OwnerKeyMismatch); + } + if self.principal != *owner.principal() { + return Err(OwnerAdmissionError::PrincipalMismatch); + } + if self.binding_id != Some(owner.binding_id()) { + return Err(OwnerAdmissionError::BindingIdMismatch); + } + if self.binding_version != Some(owner.binding_version()) { + return Err(OwnerAdmissionError::BindingVersionMismatch); + } + let fresh_until = AdmissionExpiry::new(self.fresh_until) + .map_err(|_| OwnerAdmissionError::InvalidFreshnessBound)?; + Ok(VerifiedOwnerAdmission::from_capability_snapshot( + self.authorization_domain, + self.principal.clone(), + fresh_until, + )) + } } fn finalization_time( @@ -1435,6 +1564,40 @@ where } } +/// Failure to derive delegated-owner admission from a capability snapshot. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)] +#[non_exhaustive] +pub enum OwnerAdmissionError { + #[error("capability snapshot does not carry delegated owner authority")] + NotDelegatedSnapshot, + #[error("capability snapshot and owner binding domains do not match")] + AuthorizationDomainMismatch, + #[error("capability snapshot and owner binding keys do not match")] + OwnerKeyMismatch, + #[error("capability snapshot and owner binding principals do not match")] + PrincipalMismatch, + #[error("capability snapshot and owner binding identifiers do not match")] + BindingIdMismatch, + #[error("capability snapshot and owner binding versions do not match")] + BindingVersionMismatch, + #[error("capability snapshot has an invalid admission freshness bound")] + InvalidFreshnessBound, +} + +impl OwnerAdmissionError { + pub const fn code(self) -> &'static str { + match self { + Self::NotDelegatedSnapshot => "authorization_owner_admission_001", + Self::AuthorizationDomainMismatch => "authorization_owner_admission_002", + Self::OwnerKeyMismatch => "authorization_owner_admission_003", + Self::PrincipalMismatch => "authorization_owner_admission_004", + Self::BindingIdMismatch => "authorization_owner_admission_005", + Self::BindingVersionMismatch => "authorization_owner_admission_006", + Self::InvalidFreshnessBound => "authorization_owner_admission_007", + } + } +} + impl fmt::Debug for CapabilitySnapshot { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { formatter @@ -1496,7 +1659,7 @@ impl fmt::Debug for AuthorizationOutcome { /// completes, an allowed decision is checked against exactly one fresh sample. /// Provider freshness and all effective evidence bounds use that same value; /// callers must not precompute and pass a decision-start timestamp. -async fn resolve_authorization( +pub async fn resolve_authorization( provider: &dyn AuthorizationProvider, request: &AuthorizationRequest, clock: &dyn AuthorizationClock, @@ -1716,6 +1879,9 @@ pub enum ProviderContractError { /// A capability snapshot was presented to a different configured runtime. #[error("provider capability snapshot does not belong to this authorization runtime")] AuthorizationRuntimeMismatch, + /// A narrower operation-bound delegation reached a transport-wide request path. + #[error("delegation scope is not valid for transport-wide authorization")] + UnsupportedDelegationScope, } impl ProviderContractError { @@ -1756,6 +1922,7 @@ impl ProviderContractError { Self::CapabilityBindingChanged => "authorization_provider_contract_032", Self::FederatedPolicyChanged => "authorization_provider_contract_033", Self::AuthorizationRuntimeMismatch => "authorization_provider_contract_034", + Self::UnsupportedDelegationScope => "authorization_provider_contract_035", } } } diff --git a/crates/buzz-auth/src/provider/tests.rs b/crates/buzz-auth/src/provider/tests.rs index 190cdf7da6..f346f31741 100644 --- a/crates/buzz-auth/src/provider/tests.rs +++ b/crates/buzz-auth/src/provider/tests.rs @@ -336,15 +336,13 @@ fn capability_coverage_is_exhaustive(capability: AuthorizationCapability) { fn proof_method_for_transport(transport: AuthTransport) -> AuthMethod { match transport { AuthTransport::RelayWebSocket => AuthMethod::Nip42, - AuthTransport::HttpBridge | AuthTransport::Git | AuthTransport::MediaDownload => { - AuthMethod::Nip98 - } - AuthTransport::MediaUpload => AuthMethod::Blossom, + AuthTransport::HttpBridge | AuthTransport::Git => AuthMethod::Nip98, + AuthTransport::MediaUpload | AuthTransport::MediaDownload => AuthMethod::Blossom, AuthTransport::Audio => AuthMethod::Nip42, } } -fn all_contract_errors() -> [ProviderContractError; 34] { +fn all_contract_errors() -> [ProviderContractError; 35] { [ ProviderContractError::EmptyCapabilitySet, ProviderContractError::EmptyProfileId, @@ -380,6 +378,7 @@ fn all_contract_errors() -> [ProviderContractError; 34] { ProviderContractError::CapabilityBindingChanged, ProviderContractError::FederatedPolicyChanged, ProviderContractError::AuthorizationRuntimeMismatch, + ProviderContractError::UnsupportedDelegationScope, ] } @@ -1837,6 +1836,12 @@ async fn delegated_owner_admission_does_not_require_owner_assertion() { assert_eq!(snapshot.binding_id(), Some(Uuid::from_u128(10))); assert_eq!(snapshot.binding_version(), Some(BindingVersion::INITIAL)); assert_eq!(snapshot.transport(), AuthTransport::RelayWebSocket); + let admission = snapshot + .verified_owner_admission(&existing_binding(&owner)) + .expect("delegated snapshot matches the exact owner binding"); + assert_eq!(admission.authorization_domain(), domain(1)); + assert_eq!(admission.principal(), request.principal()); + assert_eq!(admission.fresh_until().unix_seconds(), 180); } #[tokio::test] diff --git a/crates/buzz-relay/src/authorization_runtime/ephemeral.rs b/crates/buzz-relay/src/authorization_runtime/ephemeral.rs new file mode 100644 index 0000000000..a557c1fc70 --- /dev/null +++ b/crates/buzz-relay/src/authorization_runtime/ephemeral.rs @@ -0,0 +1,406 @@ +//! Relay-authenticated authority carried only across ephemeral Redis fan-out. + +use std::{fmt, sync::Arc}; + +use async_trait::async_trait; +use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _}; +use buzz_core::CommunityId; +use hmac::{Hmac, KeyInit, Mac}; +use sha2::{Digest, Sha256}; +use thiserror::Error; + +use super::executor::{revalidate_ephemeral_claim, EphemeralAuthorityClaim}; +use super::transport::ProtectedAuthorization; +use crate::connection::QueuedOutboundReleaseFence; +use crate::state::AppState; + +const DOMAIN_SEPARATOR: &[u8] = b"buzz-ephemeral-redis-authority-v1"; +type HmacSha256 = Hmac; +const PRESENCE_PREFIX: &str = "pa1:"; + +#[derive(Clone, serde::Deserialize, serde::Serialize)] +pub(crate) struct ProtectedPresenceValue { + pub(crate) status: String, + pub(crate) context_id: [u8; 32], + pub(crate) authority: String, +} + +impl fmt::Debug for ProtectedPresenceValue { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("ProtectedPresenceValue") + .field("status", &"[redacted]") + .field("context_id", &"[redacted]") + .field("authority", &"[redacted]") + .finish() + } +} + +pub(crate) fn encode_presence( + status: String, + context_id: [u8; 32], + authority: String, +) -> Result { + let value = serde_json::to_vec(&ProtectedPresenceValue { + status, + context_id, + authority, + })?; + Ok(format!( + "{PRESENCE_PREFIX}{}", + URL_SAFE_NO_PAD.encode(value) + )) +} + +pub(crate) fn decode_presence( + value: &str, +) -> Result, EphemeralAuthorityError> { + let Some(value) = value.strip_prefix(PRESENCE_PREFIX) else { + return Ok(None); + }; + let value = URL_SAFE_NO_PAD + .decode(value) + .map_err(|_| EphemeralAuthorityError::InvalidEnvelope)?; + Ok(Some(serde_json::from_slice(&value)?)) +} + +/// Seal one provider-neutral sender authority for trusted relay Redis fan-out. +pub(crate) fn seal( + state: &AppState, + authority: &ProtectedAuthorization, + event: &nostr::Event, +) -> Result { + seal_context(state, authority, event.id.to_bytes()) +} + +/// Seal enforcing authority for a non-event ephemeral effect boundary. +pub(crate) fn seal_context( + state: &AppState, + authority: &ProtectedAuthorization, + context_id: [u8; 32], +) -> Result { + let claim = authority + .seal_ephemeral_delivery(context_id)? + .ok_or(EphemeralAuthorityError::AuthorityRequired)?; + seal_claim(&signing_key(state), &claim) +} + +fn seal_claim( + signing_key: &[u8; 32], + claim: &EphemeralAuthorityClaim, +) -> Result { + let payload = serde_json::to_vec(claim)?; + let mut mac = ::new_from_slice(signing_key) + .map_err(|_| EphemeralAuthorityError::InvalidSignature)?; + mac.update(DOMAIN_SEPARATOR); + mac.update(&(payload.len() as u64).to_be_bytes()); + mac.update(&payload); + let signature = mac.finalize().into_bytes(); + Ok(format!( + "{}.{}", + URL_SAFE_NO_PAD.encode(payload), + URL_SAFE_NO_PAD.encode(signature) + )) +} + +/// Verify and preflight a remote sender authority before any local fan-out. +pub(crate) async fn verify( + state: &AppState, + community_id: CommunityId, + event: &nostr::Event, + token: &str, +) -> Result, EphemeralAuthorityError> { + let claim = verify_claim(&signing_key(state), token)?; + if claim.community_id != *community_id.as_uuid() + || claim.event_id != event.id.to_bytes() + || claim.actor_pubkey != event.pubkey.to_bytes() + || claim.capability != "community_write" + { + return Err(EphemeralAuthorityError::ContextMismatch); + } + let authority = Arc::new(RemoteEphemeralAuthority { + db: state.db.clone(), + claim, + }); + if !authority.release().await { + return Err(EphemeralAuthorityError::ExpiredOrInvalidated); + } + Ok(authority) +} + +/// Database/signing material retained by an internal cross-node effect owner. +#[derive(Clone)] +pub(crate) enum AuthorityTokenVerifier { + Database { + db: buzz_db::Db, + signing_key: [u8; 32], + }, + #[cfg(test)] + TestAllow, + #[cfg(test)] + TestDeny, + #[cfg(test)] + TestConditional(Arc), +} + +/// Authority retained by an ephemeral effect owner through cleanup. The +/// absolute lease deadline is available to synchronous realtime boundaries; +/// invalidation and binding state remain asynchronously revalidated. +#[derive(Clone)] +pub(crate) struct RetainedEphemeralAuthority { + fence: Arc, + expires_at: u64, +} + +impl RetainedEphemeralAuthority { + pub(crate) fn expires_at(&self) -> u64 { + self.expires_at + } + + pub(crate) async fn release(&self) -> bool { + self.is_time_valid() && self.fence.release().await + } + + pub(crate) fn is_time_valid(&self) -> bool { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .is_ok_and(|duration| duration.as_secs() < self.expires_at) + } +} + +impl AuthorityTokenVerifier { + /// Build from the shared relay signing secret without retaining it. + pub(crate) fn new(db: buzz_db::Db, relay_secret: &[u8]) -> Self { + Self::Database { + db, + signing_key: signing_key_from_secret(relay_secret), + } + } + + #[cfg(test)] + pub(crate) fn allow_for_test() -> Self { + Self::TestAllow + } + + #[cfg(test)] + pub(crate) fn deny_for_test() -> Self { + Self::TestDeny + } + + #[cfg(test)] + pub(crate) fn conditional_for_test(gate: Arc) -> Self { + Self::TestConditional(gate) + } + + /// Verify exact context and current database authority at the effect owner. + pub(crate) async fn verify_context( + &self, + community_id: CommunityId, + context_id: [u8; 32], + token: &str, + ) -> Result { + match self { + Self::Database { db, signing_key } => { + let claim = verify_claim(signing_key, token)?; + if claim.community_id != *community_id.as_uuid() + || claim.event_id != context_id + || claim.capability != "audio_join" + { + return Err(EphemeralAuthorityError::ContextMismatch); + } + let expires_at = claim.expires_at; + let authority = RetainedEphemeralAuthority { + fence: Arc::new(RemoteEphemeralAuthority { + db: db.clone(), + claim, + }), + expires_at, + }; + if !authority.release().await { + return Err(EphemeralAuthorityError::ExpiredOrInvalidated); + } + Ok(authority) + } + #[cfg(test)] + Self::TestAllow => Ok(RetainedEphemeralAuthority { + fence: Arc::new(TestConditionalAuthority { + gate: Arc::new(std::sync::atomic::AtomicBool::new(true)), + }), + expires_at: test_authority_expiry(), + }), + #[cfg(test)] + Self::TestDeny => Err(EphemeralAuthorityError::ExpiredOrInvalidated), + #[cfg(test)] + Self::TestConditional(gate) => { + if !gate.load(std::sync::atomic::Ordering::SeqCst) { + return Err(EphemeralAuthorityError::ExpiredOrInvalidated); + } + Ok(RetainedEphemeralAuthority { + fence: Arc::new(TestConditionalAuthority { + gate: Arc::clone(gate), + }), + expires_at: test_authority_expiry(), + }) + } + } + } + + /// Verify exact context, actor, and current database authority. + pub(crate) async fn verify_actor_context( + &self, + community_id: CommunityId, + context_id: [u8; 32], + actor_pubkey: [u8; 32], + token: &str, + ) -> Result<(), EphemeralAuthorityError> { + match self { + Self::Database { db, signing_key } => { + let claim = verify_claim(signing_key, token)?; + if claim.community_id != *community_id.as_uuid() + || claim.event_id != context_id + || claim.actor_pubkey != actor_pubkey + || claim.capability != "community_write" + { + return Err(EphemeralAuthorityError::ContextMismatch); + } + revalidate_ephemeral_claim(db, &claim) + .await + .map_err(|_| EphemeralAuthorityError::ExpiredOrInvalidated) + } + #[cfg(test)] + Self::TestAllow => Ok(()), + #[cfg(test)] + Self::TestDeny => Err(EphemeralAuthorityError::ExpiredOrInvalidated), + #[cfg(test)] + Self::TestConditional(gate) => gate + .load(std::sync::atomic::Ordering::SeqCst) + .then_some(()) + .ok_or(EphemeralAuthorityError::ExpiredOrInvalidated), + } + } +} + +#[cfg(test)] +fn test_authority_expiry() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("test wall clock after Unix epoch") + .as_secs() + + 3_600 +} + +#[cfg(test)] +struct TestConditionalAuthority { + gate: Arc, +} + +#[cfg(test)] +#[async_trait] +impl QueuedOutboundReleaseFence for TestConditionalAuthority { + async fn release(&self) -> bool { + self.gate.load(std::sync::atomic::Ordering::SeqCst) + } +} + +fn verify_claim( + signing_key: &[u8; 32], + token: &str, +) -> Result { + let (payload, signature) = token + .split_once('.') + .ok_or(EphemeralAuthorityError::InvalidEnvelope)?; + let payload = URL_SAFE_NO_PAD + .decode(payload) + .map_err(|_| EphemeralAuthorityError::InvalidEnvelope)?; + let signature = URL_SAFE_NO_PAD + .decode(signature) + .map_err(|_| EphemeralAuthorityError::InvalidEnvelope)?; + let mut mac = ::new_from_slice(signing_key) + .map_err(|_| EphemeralAuthorityError::InvalidSignature)?; + mac.update(DOMAIN_SEPARATOR); + mac.update(&(payload.len() as u64).to_be_bytes()); + mac.update(&payload); + mac.verify_slice(&signature) + .map_err(|_| EphemeralAuthorityError::InvalidSignature)?; + Ok(serde_json::from_slice(&payload)?) +} + +fn signing_key(state: &AppState) -> [u8; 32] { + signing_key_from_secret(state.relay_keypair.secret_key().as_secret_bytes()) +} + +fn signing_key_from_secret(secret: &[u8]) -> [u8; 32] { + let mut digest = Sha256::new(); + digest.update(DOMAIN_SEPARATOR); + digest.update(secret); + digest.finalize().into() +} + +struct RemoteEphemeralAuthority { + db: buzz_db::Db, + claim: EphemeralAuthorityClaim, +} + +#[async_trait] +impl QueuedOutboundReleaseFence for RemoteEphemeralAuthority { + async fn release(&self) -> bool { + revalidate_ephemeral_claim(&self.db, &self.claim) + .await + .is_ok() + } +} + +#[derive(Debug, Error)] +pub(crate) enum EphemeralAuthorityError { + #[error("ephemeral sender authority is required")] + AuthorityRequired, + #[error("ephemeral sender authority envelope is invalid")] + InvalidEnvelope, + #[error("ephemeral sender authority signature is invalid")] + InvalidSignature, + #[error("ephemeral sender authority context does not match")] + ContextMismatch, + #[error("ephemeral sender authority is expired or invalidated")] + ExpiredOrInvalidated, + #[error("ephemeral sender authority could not be sealed")] + Transport(#[from] super::transport::ProtectedTransportError), + #[error("ephemeral sender authority serialization failed")] + Serialization(#[from] serde_json::Error), +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn protected_presence_round_trips_exact_authority_context() { + let context_id = [0x5a; 32]; + let encoded = encode_presence("online".into(), context_id, "sealed-authority".into()) + .expect("encode presence"); + let decoded = decode_presence(&encoded) + .expect("decode presence") + .expect("protected value"); + assert_eq!(decoded.status, "online"); + assert_eq!(decoded.context_id, context_id); + assert_eq!(decoded.authority, "sealed-authority"); + let debug = format!("{decoded:?}"); + assert!(!debug.contains("online")); + assert!(!debug.contains("sealed-authority")); + assert!(!debug.contains("5a5a")); + } + + #[test] + fn legacy_presence_is_distinct_from_protected_presence() { + assert!(decode_presence("online") + .expect("legacy presence is valid") + .is_none()); + } + + #[test] + fn malformed_protected_presence_fails_closed() { + assert!(matches!( + decode_presence("pa1:not-base64***"), + Err(EphemeralAuthorityError::InvalidEnvelope) + )); + } +} diff --git a/crates/buzz-relay/src/authorization_runtime/production.rs b/crates/buzz-relay/src/authorization_runtime/production.rs new file mode 100644 index 0000000000..e0bf5f48a9 --- /dev/null +++ b/crates/buzz-relay/src/authorization_runtime/production.rs @@ -0,0 +1,1439 @@ +//! Disabled-by-default production construction for protected authorization. +//! +//! Configuration names exact authorization domains. There is no default +//! domain, provider fallback, request-selected profile, or implicit Enforce. + +use std::{collections::HashMap, env, sync::Arc, time::Duration}; + +use async_trait::async_trait; +use buzz_auth::{ + resolve_current_federated_policy, AccessLeasePolicy, ActiveBindingResolution, + ApplicationLeaseLimit, AuthContextInput, AuthorizationClockSkew, AuthorizationOutcome, + AuthorizationProvider, BindingLeaseBound, BindingSource, CapabilitySet, EnrollmentMode, + FederatedAuthorization, LeaseVersion, ProviderTimeout, ResolvedFederatedPolicy, Scope, + SharedAuthorizationClock, SystemAuthorizationClock, VerificationStatusPolicy, + VerifiedEvidenceAdapter, +}; +use buzz_core::{CommunityId, TenantContext}; +use sha2::{Digest, Sha256}; +use thiserror::Error; + +use super::{ + finalization::{ + AuthorizationMode, DomainAuthorizationPolicy, DomainProviderSelector, + RelayAuthorizationFinalizer, RuntimeAuthorizationDisposition, + }, + invalidation::{ + AuthorizationDependencies, AuthorizationInvalidationConfig, + AuthorizationInvalidationRuntime, AuthorizationObserver, + }, + transport::{ + DomainTransportPolicy, LeaseCurrentState, LeaseCurrentStateError, + LeaseCurrentStateObserver, ProtectedAuthorizationResolver, ProtectedOperationRequest, + ProtectedResolution, ProtectedResolutionError, ProtectedTransportRuntime, + }, +}; + +const DOMAINS_ENV: &str = "BUZZ_PROTECTED_AUTHORIZATION_DOMAINS"; +const PROFILE_ENV: &str = "BUZZ_PROTECTED_AUTHORIZATION_PROFILE"; +const LEASE_SECONDS_ENV: &str = "BUZZ_PROTECTED_AUTHORIZATION_LEASE_SECONDS"; +const RESTORE_BOOTSTRAPS_ENV: &str = "BUZZ_PROTECTED_AUTHORIZATION_RESTORE_BOOTSTRAPS"; +const MAX_AUDIO_RECONCILIATION_SWEEPS: usize = 8; + +/// Runtime plus its durable invalidation worker. +pub struct InstalledProtectedRuntime { + /// Exact-domain transport runtime. + pub transport: Arc, + /// Durable invalidation runtime initialized before transport installation. + pub invalidation: AuthorizationInvalidationRuntime, + /// Independent high-water witness verified before transport installation. + pub restore: Arc, + /// Whether any exact domain is authoritative Enforce. + pub enforce_enabled: bool, + /// Exact authoritative domains whose crash remnants may be reconciled. + pub enforcing_domains: Vec, + /// Enforce domains whose optional public projection requires reconciliation. + pub projection_domains: Vec, +} + +/// Result of the single production installation boundary. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ProtectedRuntimeInstallation { + /// No exact authorization domains were configured; legacy behavior remains. + Disabled, + /// Exact-domain transport, restore, and invalidation runtimes were installed. + Installed, +} + +/// Exact-domain provider registry supplied by the deployment adapter. +/// +/// The OSS runtime never invents a current-admission decision. Enforce +/// construction fails when a configured domain has no injected O2 provider. +#[derive(Default)] +pub struct ProductionProviderRegistry { + providers: HashMap>, +} + +impl ProductionProviderRegistry { + /// Build an exact registry, rejecting duplicate domain mappings. + pub fn new( + providers: impl IntoIterator)>, + ) -> Result { + let mut exact = HashMap::new(); + for (domain, provider) in providers { + if exact.insert(domain, provider).is_some() { + return Err(ProductionRuntimeError::InvalidConfiguration); + } + } + Ok(Self { providers: exact }) + } + + fn provider_for( + &self, + domain: CommunityId, + ) -> Result, ProductionRuntimeError> { + self.providers + .get(&domain) + .cloned() + .ok_or(ProductionRuntimeError::ProviderMissing) + } +} + +struct ProductionLeaseObserver { + invalidation: AuthorizationObserver, + current: LeaseCurrentState, +} + +impl LeaseCurrentStateObserver for ProductionLeaseObserver { + fn observe_current(&self) -> Result { + self.invalidation + .recheck() + .map_err(|_| LeaseCurrentStateError::Stale)?; + Ok(self.current.clone()) + } + + fn observe_commit_fence( + &self, + ) -> Result { + self.invalidation + .commit_fence() + .map_err(|_| LeaseCurrentStateError::Unavailable) + } +} + +struct ProductionResolver { + db: buzz_db::Db, + tenants: HashMap, + finalizer: RelayAuthorizationFinalizer, + invalidation: AuthorizationInvalidationRuntime, + clock: SharedAuthorizationClock, +} + +impl ProductionResolver { + fn tenant(&self, domain: CommunityId) -> Result<&TenantContext, ProtectedResolutionError> { + self.tenants + .get(&domain) + .ok_or(ProtectedResolutionError::new("configured_domain_missing")) + } + + async fn current_policy( + &self, + domain: CommunityId, + correlation_id: uuid::Uuid, + ) -> Result { + let now = self + .clock + .now() + .map_err(|_| ProtectedResolutionError::new("authorization_clock"))? + .unix_seconds(); + resolve_current_federated_policy( + &self.db.federated_authority_adapter(), + domain, + correlation_id, + now, + ) + .await + .map_err(|_| ProtectedResolutionError::new("federated_policy_unavailable")) + } + + async fn active_binding( + &self, + domain: CommunityId, + pubkey: nostr::PublicKey, + assertion: Option<&buzz_auth::VerifiedFederatedAssertion>, + ) -> Result { + let binding = self + .db + .get_active_identity_binding_by_pubkey(domain, pubkey.as_bytes()) + .await + .map_err(|_| ProtectedResolutionError::new("binding_unavailable"))? + .ok_or(ProtectedResolutionError::new("active_binding_required"))?; + let source = match binding.binding_provenance { + buzz_db::identity_binding::BindingProvenance::AttestedKey => BindingSource::AttestedKey, + buzz_db::identity_binding::BindingProvenance::Provisioned => BindingSource::Provisioned, + buzz_db::identity_binding::BindingProvenance::Tofu => BindingSource::Tofu, + }; + let expires_at = binding + .expires_at + .as_ref() + .map(|value| u64::try_from(value.timestamp())) + .transpose() + .map_err(|_| ProtectedResolutionError::new("binding_expiry_invalid"))?; + VerifiedEvidenceAdapter::new() + .active_binding_from_store( + domain, + domain, + binding.binding_id, + &binding.issuer, + &binding.uid, + pubkey, + binding.binding_version, + expires_at, + source, + ActiveBindingResolution::Existing, + assertion, + ) + .map_err(|_| ProtectedResolutionError::new("binding_evidence_mismatch")) + } + + async fn require_membership( + &self, + request: &ProtectedOperationRequest, + ) -> Result<(), ProtectedResolutionError> { + let member = request + .owner_pubkey() + .unwrap_or_else(|| request.actor_pubkey()); + match self + .db + .is_relay_member(request.authorization_domain(), &member.to_hex()) + .await + { + Ok(true) => Ok(()), + Ok(false) => Err(ProtectedResolutionError::new("membership_required")), + Err(_) => Err(ProtectedResolutionError::new("membership_unavailable")), + } + } + + fn reseal_assertion( + &self, + assertion: &buzz_auth::VerifiedFederatedAssertion, + ) -> Result { + let now = self + .clock + .now() + .map_err(|_| ProtectedResolutionError::new("authorization_clock"))? + .unix_seconds(); + VerifiedEvidenceAdapter::new() + .federated_assertion_from_validated_claims( + assertion.authorization_domain(), + assertion.authorized_transport(), + assertion.principal().issuer(), + assertion.principal().subject(), + assertion.key_attestation().map(|key| key.pubkey()), + assertion.transport(), + assertion.not_before().map(|bound| bound.unix_seconds()), + assertion.expires_at().unix_seconds(), + now, + ) + .map_err(|_| ProtectedResolutionError::new("assertion_stale")) + } +} + +#[async_trait] +impl ProtectedAuthorizationResolver for ProductionResolver { + async fn observe( + &self, + request: &ProtectedOperationRequest, + ) -> Result<(), ProtectedResolutionError> { + let tenant = self.tenant(request.authorization_domain())?; + let fence = self + .invalidation + .capture_before_evaluation(request.authorization_domain()) + .await + .map_err(|_| ProtectedResolutionError::new("invalidation_unavailable"))?; + self.require_membership(request).await?; + let capabilities = CapabilitySet::single(request.capability()); + let federated_policy = self + .current_policy(request.authorization_domain(), request.correlation_id()) + .await?; + let outcome = if let Some(owner) = request.owner_pubkey() { + let binding = self + .active_binding(request.authorization_domain(), owner, None) + .await?; + self.finalizer + .evaluate_delegated( + tenant, + request.verified_proof(), + &binding, + federated_policy, + capabilities, + request.correlation_id(), + ) + .await + } else { + let assertion = request + .verified_assertion() + .ok_or(ProtectedResolutionError::new("direct_assertion_required"))?; + let _binding = self + .active_binding( + request.authorization_domain(), + request.actor_pubkey(), + Some(assertion), + ) + .await?; + self.finalizer + .evaluate_direct( + tenant, + request.verified_proof(), + assertion, + federated_policy, + capabilities, + request.correlation_id(), + ) + .await + } + .map_err(|_| ProtectedResolutionError::new("provider_request_invalid"))?; + if !matches!(outcome, AuthorizationOutcome::Allow(_)) { + return Err(ProtectedResolutionError::new("provider_denied")); + } + self.invalidation + .recheck_evaluation(fence) + .map_err(|_| ProtectedResolutionError::new("invalidation_race")) + } + + async fn present( + &self, + request: &ProtectedOperationRequest, + ) -> Result { + if request.owner_pubkey().is_some() { + return Err(ProtectedResolutionError::new( + "client_status_requires_direct_binding", + )); + } + let tenant = self.tenant(request.authorization_domain())?; + let fence = self + .invalidation + .capture_before_evaluation(request.authorization_domain()) + .await + .map_err(|_| ProtectedResolutionError::new("invalidation_unavailable"))?; + self.require_membership(request).await?; + let assertion = request + .verified_assertion() + .ok_or(ProtectedResolutionError::new("direct_assertion_required"))?; + let binding = self + .active_binding( + request.authorization_domain(), + request.actor_pubkey(), + Some(assertion), + ) + .await?; + let evaluation_policy = self + .current_policy(request.authorization_domain(), request.correlation_id()) + .await?; + let snapshot = match self + .finalizer + .evaluate_direct( + tenant, + request.verified_proof(), + assertion, + evaluation_policy, + CapabilitySet::single(request.capability()), + request.correlation_id(), + ) + .await + .map_err(|_| ProtectedResolutionError::new("provider_request_invalid"))? + { + AuthorizationOutcome::Allow(snapshot) => snapshot, + _ => return Err(ProtectedResolutionError::new("provider_denied")), + }; + let binding_bound = BindingLeaseBound::new(&binding, snapshot.effective_until()) + .map_err(|_| ProtectedResolutionError::new("binding_bound_invalid"))?; + let authorization = FederatedAuthorization::Direct { + binding, + assertion: self.reseal_assertion(assertion)?, + }; + let access = VerifiedEvidenceAdapter::new() + .community_access_from_policy( + tenant, + request.authorization_domain(), + Scope::all_known(), + None, + ) + .map_err(|_| ProtectedResolutionError::new("community_access_invalid"))?; + let input = AuthContextInput::new( + tenant.clone(), + request.correlation_id(), + Arc::clone(request.verified_proof()), + access, + ); + let finalization_policy = self + .current_policy(request.authorization_domain(), request.correlation_id()) + .await?; + let disposition = self + .finalizer + .finalize_client_status( + input, + finalization_policy, + authorization, + snapshot, + binding_bound, + ) + .map_err(|_| ProtectedResolutionError::new("status_finalization"))?; + let dependencies = AuthorizationDependencies::from_verification_only( + request + .session_id() + .unwrap_or_else(|| request.correlation_id()), + &disposition, + ) + .map_err(|_| ProtectedResolutionError::new("invalidation_dependencies"))?; + let observer = self + .invalidation + .observe_authority(fence, dependencies, request.cancellation()) + .map_err(|_| ProtectedResolutionError::new("invalidation_race"))?; + let current = LeaseCurrentState::from_trusted_runtime( + disposition.binding_version(), + disposition.profile_id().clone(), + disposition.policy_version().clone(), + ); + Ok(super::transport::ProtectedStatusResolution::new( + disposition, + Arc::new(ProductionLeaseObserver { + invalidation: observer, + current, + }), + fence.generation(), + )) + } + + async fn resolve( + &self, + request: &ProtectedOperationRequest, + ) -> Result { + let tenant = self.tenant(request.authorization_domain())?; + let fence = self + .invalidation + .capture_before_evaluation(request.authorization_domain()) + .await + .map_err(|_| ProtectedResolutionError::new("invalidation_unavailable"))?; + let capabilities = CapabilitySet::single(request.capability()); + + if let Some(assertion) = request.enrollment_assertion() { + let evaluation_policy = self + .current_policy(request.authorization_domain(), request.correlation_id()) + .await?; + let outcome = self + .finalizer + .evaluate_direct( + tenant, + request.verified_proof(), + assertion, + evaluation_policy, + capabilities, + request.correlation_id(), + ) + .await + .map_err(|_| ProtectedResolutionError::new("provider_request_invalid"))?; + let AuthorizationOutcome::Allow(snapshot) = outcome else { + return Err(ProtectedResolutionError::new("provider_denied")); + }; + let finalization_policy = self + .current_policy(request.authorization_domain(), request.correlation_id()) + .await?; + let disposition = self + .finalizer + .finalize_enrollment( + tenant, + request.verified_proof(), + assertion, + finalization_policy, + snapshot, + request.correlation_id(), + ) + .map_err(|_| ProtectedResolutionError::new("enrollment_finalization"))?; + let dependencies = AuthorizationDependencies::from_enrollment( + request + .session_id() + .unwrap_or_else(|| request.correlation_id()), + &disposition, + ) + .map_err(|_| ProtectedResolutionError::new("invalidation_dependencies"))?; + let observer = self + .invalidation + .observe_authority(fence, dependencies, request.cancellation()) + .map_err(|_| ProtectedResolutionError::new("invalidation_race"))?; + let current = LeaseCurrentState::from_trusted_runtime( + buzz_auth::BindingVersion::INITIAL, + disposition.profile_id().clone(), + disposition.policy_version().clone(), + ); + return Ok(ProtectedResolution::enrollment( + disposition, + Arc::new(ProductionLeaseObserver { + invalidation: observer, + current, + }), + )); + } + + self.require_membership(request).await?; + let adapter = VerifiedEvidenceAdapter::new(); + let evaluation_policy = self + .current_policy(request.authorization_domain(), request.correlation_id()) + .await?; + let (authorization, snapshot) = if let Some(owner) = request.owner_pubkey() { + let binding = self + .active_binding(request.authorization_domain(), owner, None) + .await?; + let outcome = self + .finalizer + .evaluate_delegated( + tenant, + request.verified_proof(), + &binding, + evaluation_policy, + capabilities, + request.correlation_id(), + ) + .await + .map_err(|_| ProtectedResolutionError::new("provider_request_invalid"))?; + let AuthorizationOutcome::Allow(snapshot) = outcome else { + return Err(ProtectedResolutionError::new("provider_denied")); + }; + let admission = snapshot + .verified_owner_admission(&binding) + .map_err(|_| ProtectedResolutionError::new("owner_admission_mismatch"))?; + ( + FederatedAuthorization::Delegated { + owner: binding, + admission, + }, + snapshot, + ) + } else { + let assertion = request + .verified_assertion() + .ok_or(ProtectedResolutionError::new("direct_assertion_required"))?; + let binding = self + .active_binding( + request.authorization_domain(), + request.actor_pubkey(), + Some(assertion), + ) + .await?; + let outcome = self + .finalizer + .evaluate_direct( + tenant, + request.verified_proof(), + assertion, + evaluation_policy, + capabilities, + request.correlation_id(), + ) + .await + .map_err(|_| ProtectedResolutionError::new("provider_request_invalid"))?; + let AuthorizationOutcome::Allow(snapshot) = outcome else { + return Err(ProtectedResolutionError::new("provider_denied")); + }; + let owned_assertion = self.reseal_assertion(assertion)?; + ( + FederatedAuthorization::Direct { + binding, + assertion: owned_assertion, + }, + snapshot, + ) + }; + let binding = match &authorization { + FederatedAuthorization::Direct { binding, .. } => binding, + FederatedAuthorization::Delegated { owner, .. } => owner, + FederatedAuthorization::NotRequired => { + unreachable!("protected resolver requires identity") + } + }; + let binding_bound = BindingLeaseBound::new(binding, snapshot.effective_until()) + .map_err(|_| ProtectedResolutionError::new("binding_bound_invalid"))?; + let access = adapter + .community_access_from_policy( + tenant, + request.authorization_domain(), + Scope::all_known(), + None, + ) + .map_err(|_| ProtectedResolutionError::new("community_access_invalid"))?; + let input = AuthContextInput::new( + tenant.clone(), + request.correlation_id(), + Arc::clone(request.verified_proof()), + access, + ); + let finalization_policy = self + .current_policy(request.authorization_domain(), request.correlation_id()) + .await?; + let disposition = self + .finalizer + .finalize_allowed( + input, + finalization_policy, + authorization, + snapshot, + binding_bound, + LeaseVersion::INITIAL, + ) + .map_err(|_| ProtectedResolutionError::new("authorization_finalization"))?; + let RuntimeAuthorizationDisposition::Access(context) = disposition else { + return Err(ProtectedResolutionError::new( + "non_authoritative_disposition", + )); + }; + let dependencies = AuthorizationDependencies::from_context( + request + .session_id() + .unwrap_or_else(|| request.correlation_id()), + &context, + ) + .map_err(|_| ProtectedResolutionError::new("invalidation_dependencies"))?; + let observer = self + .invalidation + .observe_authority(fence, dependencies, request.cancellation()) + .map_err(|_| ProtectedResolutionError::new("invalidation_race"))?; + let lease = context + .authorization_lease() + .ok_or(ProtectedResolutionError::new("access_lease_missing"))?; + let current = LeaseCurrentState::from_trusted_runtime( + lease.binding_version(), + lease.profile_id().clone(), + lease.policy_version().clone(), + ); + Ok(ProtectedResolution::access( + context, + Arc::new(ProductionLeaseObserver { + invalidation: observer, + current, + }), + )) + } +} + +/// Build the optional exact-domain runtime from provider-neutral environment +/// configuration. An absent or blank domain list installs nothing. +pub async fn build_from_environment( + state: &crate::state::AppState, +) -> Result, ProductionRuntimeError> { + build_from_environment_with_providers(state, ProductionProviderRegistry::default()).await +} + +/// Build and install the disabled-by-default stock runtime. +/// +/// The stock OSS binary has no admission provider and therefore fails closed +/// if a non-Off domain is configured. A deployment composition root with an +/// exact O2 provider calls [`install_from_environment_with_providers`] instead. +pub async fn install_from_environment( + state: &Arc, +) -> Result { + install_from_environment_with_providers(state, ProductionProviderRegistry::default()).await +} + +/// Build and install one exact provider-neutral production runtime. +/// +/// This is the sole production composition seam for externally supplied O2 +/// providers. Restore and invalidation state is initialized before either is +/// made reachable from `AppState`; missing providers and partial installation +/// fail startup rather than falling back to legacy authorization. +pub async fn install_from_environment_with_providers( + state: &Arc, + providers: ProductionProviderRegistry, +) -> Result { + if state.protected_transport().is_some() || state.restore_protection().is_some() { + return Err(ProductionRuntimeError::AlreadyInstalled); + } + let Some(installed) = build_from_environment_with_providers(state, providers).await? else { + return Ok(ProtectedRuntimeInstallation::Disabled); + }; + let InstalledProtectedRuntime { + transport, + invalidation, + restore, + enforce_enabled, + enforcing_domains, + projection_domains, + } = installed; + let audio_restore = Arc::clone(&restore); + state + .install_restore_protection(restore) + .map_err(|_| ProductionRuntimeError::AlreadyInstalled)?; + state + .install_protected_transport(transport) + .map_err(|_| ProductionRuntimeError::AlreadyInstalled)?; + + let invalidation_worker = invalidation.clone(); + let invalidation_failure = invalidation.clone(); + let authorization_hint_subscriber = Arc::clone(&state.pubsub); + let mut workers = tokio::task::JoinSet::new(); + workers.spawn(async move { invalidation_worker.run().await }); + workers.spawn(async move { + authorization_hint_subscriber + .run_authorization_invalidation_subscriber() + .await; + }); + if enforce_enabled { + workers.spawn(run_audio_reconciliation( + state.db.clone(), + audio_restore, + enforcing_domains, + )); + } + workers.spawn( + crate::corporate_identity::run_public_projection_retirement_reconciliation( + Arc::clone(state), + projection_domains, + ), + ); + // Any required worker exit is a runtime-health failure. A detached bare + // worker could otherwise panic while Enforce kept serving from stale + // authority. The supervisor first invalidates every domain, then aborts + // the remaining workers; all protected observations fail closed. + tokio::spawn(async move { + let completion = workers.join_next().await; + invalidation_failure.fail_closed(); + workers.abort_all(); + while workers.join_next().await.is_some() {} + match completion { + Some(Ok(())) => { + tracing::error!( + "required protected authorization worker exited; runtime failed closed" + ) + } + Some(Err(error)) => tracing::error!( + %error, + "required protected authorization worker failed; runtime failed closed" + ), + None => tracing::error!( + "protected authorization worker set was empty; runtime failed closed" + ), + } + }); + Ok(ProtectedRuntimeInstallation::Installed) +} + +/// Build the disabled-by-default runtime with exact provider implementations +/// supplied by the deployment adapter. The stock OSS binary supplies an empty +/// registry, so any configured non-Off domain fails closed instead of using a +/// permissive fallback. +pub async fn build_from_environment_with_providers( + state: &crate::state::AppState, + providers: ProductionProviderRegistry, +) -> Result, ProductionRuntimeError> { + let raw = env::var(DOMAINS_ENV).unwrap_or_default(); + let configured = parse_domains(&raw)?; + let activated = state.db.activated_authorization_domains().await?; + validate_activated_domain_configuration(&configured, activated.iter().copied())?; + if configured.is_empty() { + return Ok(None); + } + validate_provider_coverage(&configured, &providers)?; + if configured + .values() + .any(|mode| *mode != AuthorizationMode::Off) + && state.identity_assertion_provenance().is_none() + { + return Err(ProductionRuntimeError::AssertionProvenanceMissing); + } + let enforcing_domains = authoritative_domains(&configured); + let projection_domains = projection_reconciliation_domains(&configured); + if !enforcing_domains.is_empty() && state.corporate_identity.is_none() { + return Err(ProductionRuntimeError::VerifierMissing); + } + let clock: SharedAuthorizationClock = Arc::new(SystemAuthorizationClock); + let restore_bootstraps = parse_restore_bootstraps( + &env::var(RESTORE_BOOTSTRAPS_ENV).unwrap_or_default(), + &enforcing_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)?; + let lease_limit = ApplicationLeaseLimit::from_seconds(lease_seconds)?; + let status_limit = ApplicationLeaseLimit::from_seconds(lease_seconds.min(60))?; + let skew = AuthorizationClockSkew::from_seconds(0)?; + 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 { + continue; + } else { + providers.provider_for(*domain)? + }; + policies.push(DomainAuthorizationPolicy::from_server_configuration( + *domain, + profile.clone(), + provider, + EnrollmentMode::AttestedKey, + *mode, + ProviderTimeout::new(Duration::from_secs(2))?, + 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 + .into_iter() + .map(|entry| { + ( + CommunityId::from_uuid(entry.id), + TenantContext::resolved(CommunityId::from_uuid(entry.id), entry.host), + ) + }) + .collect::>(); + for domain in configured.keys() { + if !host_map.contains_key(domain) { + return Err(ProductionRuntimeError::ConfiguredDomainMissing); + } + } + let restore = super::restore::RestoreProtectionRuntime::initialize( + state.db.clone(), + state.git_store.clone(), + restore_bootstraps, + ) + .await?; + activate_enforcing_domains(&state.db, &restore, enforcing_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(), + Arc::clone(&state.pubsub), + AuthorizationInvalidationConfig::default(), + Arc::clone(&restore), + ); + invalidation + .initialize_domains(enforcing_domains.iter().copied()) + .await?; + crate::corporate_identity::reconcile_public_projection_retirements_startup( + state, + &projection_domains, + ) + .await + .map_err(|_| ProductionRuntimeError::PublicProjection)?; + let finalizer = RelayAuthorizationFinalizer::new( + DomainProviderSelector::new(policies)?, + Arc::clone(&clock), + ); + let resolver: Arc = Arc::new(ProductionResolver { + db: state.db.clone(), + tenants: host_map, + finalizer, + invalidation: invalidation.clone(), + clock: Arc::clone(&clock), + }); + let transport = Arc::new(ProtectedTransportRuntime::new(transports, resolver, clock)?); + Ok(Some(InstalledProtectedRuntime { + transport, + invalidation, + restore, + enforce_enabled: !enforcing_domains.is_empty(), + enforcing_domains, + projection_domains, + })) +} + +async fn activate_enforcing_domains( + db: &buzz_db::Db, + restore: &Arc, + domains: impl IntoIterator, +) -> Result<(), ProductionRuntimeError> { + for domain in domains { + let mut digest = Sha256::new(); + digest.update(b"buzz-protected-domain-activation-v1"); + digest.update(domain.as_uuid().as_bytes()); + let fingerprint: [u8; 32] = digest.finalize().into(); + let operation_id = super::executor::ProtectedOperationId::derive( + domain, + "runtime.domain.activate.v1", + &fingerprint, + )? + .as_uuid(); + let witness = restore.begin(domain, operation_id, fingerprint).await?; + match db + .activate_authorization_domain(domain, operation_id, fingerprint) + .await + { + Ok(()) => witness.commit().await?, + Err(error) => { + let committed = db + .authorization_operation_receipt_fingerprint(domain, operation_id) + .await?; + if committed == Some(fingerprint) { + witness.commit().await?; + } else { + witness.abort().await?; + return Err(error.into()); + } + } + } + } + Ok(()) +} + +fn validate_provider_coverage( + configured: &HashMap, + providers: &ProductionProviderRegistry, +) -> Result<(), ProductionRuntimeError> { + for (domain, mode) in configured { + if *mode != AuthorizationMode::Off { + providers.provider_for(*domain)?; + } + } + Ok(()) +} + +fn validate_activated_domain_configuration( + configured: &HashMap, + activated: impl IntoIterator, +) -> Result<(), ProductionRuntimeError> { + if activated + .into_iter() + .any(|domain| configured.get(&domain) != Some(&AuthorizationMode::Enforce)) + { + return Err(ProductionRuntimeError::ActivatedDomainDowngrade); + } + Ok(()) +} + +/// Reconcile expired durable audio attempts. Discovery is read-only; an +/// unexpired claimant remains exclusively owned by its healthy replica and +/// every orphan transition is independently witnessed. +pub async fn run_audio_reconciliation( + db: buzz_db::Db, + restore: Arc, + enforcing_domains: Vec, +) { + let mut interval = tokio::time::interval(Duration::from_secs(30)); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + interval.tick().await; + if let Err(error) = + reconcile_audio_admissions_once(&db, &restore, enforcing_domains.iter().copied()).await + { + tracing::warn!(%error, "durable audio admission reconciliation failed"); + } + } +} + +async fn reconcile_audio_admissions_once( + db: &buzz_db::Db, + restore: &Arc, + domains: impl IntoIterator, +) -> Result { + let mut reconciled = 0_u64; + let mut first_error: Option = None; + for domain in domains { + let mut domain_failed = false; + for sweep in 0..MAX_AUDIO_RECONCILIATION_SWEEPS { + let mut cursor = None; + loop { + let discovered = buzz_db::audio_admission::reconcilable_audio_admissions_after( + db, domain, cursor, + ) + .await; + let candidates = match discovered { + Ok(candidates) => candidates, + Err(error) => { + tracing::warn!(%error, "durable audio admission discovery failed for one domain"); + first_error.get_or_insert_with(|| error.into()); + domain_failed = true; + break; + } + }; + let Some(last) = candidates.last().map(|candidate| candidate.admission_id) else { + break; + }; + cursor = Some(last); + for candidate in candidates { + match reconcile_audio_admission(db, restore, domain, candidate).await { + Ok(true) => reconciled = reconciled.saturating_add(1), + Ok(false) => {} + Err(error) => { + tracing::warn!( + %error, + "durable audio admission candidate failed without starving later cleanup" + ); + first_error.get_or_insert(error); + domain_failed = true; + } + } + } + } + if domain_failed { + break; + } + let remaining = + buzz_db::audio_admission::reconcilable_audio_admissions(db, domain).await; + match remaining { + Ok(remaining) if remaining.is_empty() => break, + Ok(_) if sweep + 1 < MAX_AUDIO_RECONCILIATION_SWEEPS => continue, + Ok(_) => { + first_error + .get_or_insert(ProductionRuntimeError::AudioReconciliationIncomplete); + break; + } + Err(error) => { + first_error.get_or_insert_with(|| error.into()); + break; + } + } + } + } + match first_error { + Some(error) => Err(error), + None => Ok(reconciled), + } +} + +async fn reconcile_audio_admission( + db: &buzz_db::Db, + restore: &Arc, + domain: CommunityId, + candidate: buzz_db::audio_admission::AudioAdmissionReconciliationCandidate, +) -> Result { + use super::executor::ProtectedOperationId; + + // Reconciliation never proves a graceful disconnect. Even a formerly + // visible orphan is conservatively compensated as aborted after its + // durable ownership deadline and grace period. + let finished = false; + let terminal = b"aborted".as_slice(); + let mut stable = Sha256::new(); + stable.update(b"buzz-audio-admission-reconciliation-v1"); + stable.update(candidate.admission_id.as_bytes()); + stable.update(candidate.claimant_id.as_bytes()); + stable.update(candidate.source_state.as_str().as_bytes()); + stable.update(terminal); + stable.update(candidate.state_version.to_be_bytes()); + let stable: [u8; 32] = stable.finalize().into(); + let operation_id = + ProtectedOperationId::derive(domain, "audio.admission.reconcile.v1", &stable) + .map_err(|_| ProductionRuntimeError::InvalidConfiguration)?; + let mut request = Sha256::new(); + request.update(b"buzz-audio-admission-reconciliation-request-v1"); + request.update(stable); + let request: [u8; 32] = request.finalize().into(); + let witness = restore + .begin(domain, operation_id.as_uuid(), request) + .await?; + match buzz_db::audio_admission::reconcile_claimed_audio_admission_with_receipt( + db, + domain, + candidate, + finished, + Some("orphaned_attachment"), + operation_id.as_uuid(), + request, + ) + .await + { + Ok(true) => witness.commit().await?, + Ok(false) => { + witness.abort().await?; + return Ok(false); + } + Err(error) => { + witness.abort().await?; + return Err(error.into()); + } + } + Ok(true) +} + +fn parse_positive_seconds(name: &'static str, default: u64) -> Result { + match env::var(name) { + Ok(value) => value + .parse::() + .ok() + .filter(|value| *value > 0) + .ok_or(ProductionRuntimeError::InvalidConfiguration), + Err(env::VarError::NotPresent) => Ok(default), + Err(env::VarError::NotUnicode(_)) => Err(ProductionRuntimeError::InvalidConfiguration), + } +} + +fn parse_domains( + raw: &str, +) -> Result, ProductionRuntimeError> { + let mut domains = HashMap::new(); + for item in raw + .split(',') + .map(str::trim) + .filter(|item| !item.is_empty()) + { + let (id, mode) = item + .split_once(':') + .ok_or(ProductionRuntimeError::InvalidConfiguration)?; + let domain = CommunityId::from_uuid( + uuid::Uuid::parse_str(id).map_err(|_| ProductionRuntimeError::InvalidConfiguration)?, + ); + let mode = match mode.trim().to_ascii_lowercase().as_str() { + "off" => AuthorizationMode::Off, + "shadow" => AuthorizationMode::Shadow, + "verify_only" => AuthorizationMode::VerifyOnly, + "enforce" => AuthorizationMode::Enforce, + _ => return Err(ProductionRuntimeError::InvalidConfiguration), + }; + if domains.insert(domain, mode).is_some() { + return Err(ProductionRuntimeError::InvalidConfiguration); + } + } + Ok(domains) +} + +fn authoritative_domains(configured: &HashMap) -> Vec { + configured + .iter() + .filter_map(|(domain, mode)| (*mode == AuthorizationMode::Enforce).then_some(*domain)) + .collect() +} + +fn projection_reconciliation_domains( + configured: &HashMap, +) -> Vec { + // Public projection retirement belongs to authoritative Enforce runtime. + // Observational modes must not queue or publish projection changes. + configured + .iter() + .filter_map(|(domain, mode)| (*mode == AuthorizationMode::Enforce).then_some(*domain)) + .collect() +} + +fn parse_restore_bootstraps( + raw: &str, + enforcing_domains: &[CommunityId], +) -> Result, ProductionRuntimeError> { + let mut anchors = HashMap::new(); + for item in raw + .split(',') + .map(str::trim) + .filter(|item| !item.is_empty()) + { + let (domain, bootstrap) = item + .split_once('=') + .ok_or(ProductionRuntimeError::InvalidConfiguration)?; + let domain = CommunityId::from_uuid( + uuid::Uuid::parse_str(domain.trim()) + .map_err(|_| ProductionRuntimeError::InvalidConfiguration)?, + ); + let bootstrap = uuid::Uuid::parse_str(bootstrap.trim()) + .map_err(|_| ProductionRuntimeError::InvalidConfiguration)?; + if bootstrap.is_nil() || anchors.insert(domain, bootstrap).is_some() { + return Err(ProductionRuntimeError::InvalidConfiguration); + } + } + let mut result = Vec::with_capacity(enforcing_domains.len()); + for domain in enforcing_domains { + let bootstrap = anchors + .remove(domain) + .ok_or(ProductionRuntimeError::RestoreBootstrapMissing)?; + result.push((*domain, bootstrap)); + } + if !anchors.is_empty() { + return Err(ProductionRuntimeError::InvalidConfiguration); + } + Ok(result) +} + +/// Fail-closed production construction error. +#[derive(Debug, Error)] +pub enum ProductionRuntimeError { + /// Configuration was malformed or ambiguous. + #[error("protected authorization configuration is invalid")] + InvalidConfiguration, + /// An exact configured domain has no durable host mapping. + #[error("protected authorization domain is not present")] + ConfiguredDomainMissing, + /// Typed provider or finalization configuration was rejected. + #[error(transparent)] + Provider(#[from] buzz_auth::ProviderContractError), + /// A configured authoritative or observational domain has no exact O2 provider. + #[error("protected authorization provider is not configured for this domain")] + ProviderMissing, + /// Protected production state was already partially or fully installed. + #[error("protected authorization runtime is already installed")] + AlreadyInstalled, + /// Enforce was requested without a complete domain-usable assertion verifier. + #[error("protected authorization assertion verifier is not configured")] + VerifierMissing, + /// A protected domain was configured without deployment-verified ingress + /// provenance for its direct identity assertion. + #[error("protected authorization assertion provenance is not configured")] + AssertionProvenanceMissing, + /// A previously activated domain was omitted or configured non-authoritatively. + #[error("protected authorization domain cannot be downgraded after activation")] + ActivatedDomainDowngrade, + /// An Enforce domain has no exact externally provisioned restore anchor. + #[error("protected authorization restore bootstrap is not configured")] + RestoreBootstrapMissing, + /// Lease bounds were rejected. + #[error(transparent)] + Lease(#[from] buzz_auth::LeasePolicyError), + /// Durable domain construction failed. + #[error(transparent)] + Database(#[from] buzz_db::DbError), + /// Invalidation initialization failed. + #[error(transparent)] + Invalidation(#[from] super::invalidation::AuthorizationInvalidationRuntimeError), + /// Exact-domain policy construction failed. + #[error(transparent)] + DomainPolicy(#[from] super::finalization::DomainPolicyError), + /// Protected transport construction failed. + #[error(transparent)] + Transport(#[from] super::transport::ProtectedTransportError), + /// Independent restore witness initialization failed. + #[error(transparent)] + Restore(#[from] super::restore::RestoreProtectionError), + /// Stable operation construction failed before production activation. + #[error(transparent)] + Execution(#[from] super::executor::AuthorizationExecutionError), + /// Public projection startup reconciliation failed. + #[error("public identity projection reconciliation failed")] + PublicProjection, + /// Startup could not prove that all durable audio remnants were reconciled. + #[error("protected audio reconciliation did not reach a complete fixed point")] + AudioReconciliationIncomplete, +} + +#[cfg(test)] +mod tests { + use buzz_auth::{ + AuthorizationProviderFuture, AuthorizationRequest, ProviderDecision, ProviderUnavailable, + ProviderUnavailableReason, + }; + + use super::*; + + struct SyntheticUnavailableProvider; + + impl AuthorizationProvider for SyntheticUnavailableProvider { + fn authorize<'a>( + &'a self, + _request: &'a AuthorizationRequest, + ) -> AuthorizationProviderFuture<'a> { + Box::pin(async { + ProviderDecision::Unavailable(ProviderUnavailable::new( + ProviderUnavailableReason::DependencyUnavailable, + None, + )) + }) + } + } + + #[test] + fn absent_configuration_is_disabled() { + assert!(parse_domains("").expect("empty is disabled").is_empty()); + } + + #[test] + 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"); + assert_eq!( + parsed.get(&CommunityId::from_uuid(first)), + Some(&AuthorizationMode::Enforce) + ); + assert_eq!( + parsed.get(&CommunityId::from_uuid(second)), + Some(&AuthorizationMode::VerifyOnly) + ); + } + + #[test] + fn duplicate_or_unknown_domains_fail_closed() { + let id = uuid::Uuid::new_v4(); + assert!(parse_domains(&format!("{id}:enforce,{id}:shadow")).is_err()); + assert!(parse_domains(&format!("{id}:automatic")).is_err()); + } + + #[test] + fn observational_modes_have_no_durable_runtime_domains() { + let off = uuid::Uuid::new_v4(); + let shadow = uuid::Uuid::new_v4(); + let verify = uuid::Uuid::new_v4(); + let enforce = uuid::Uuid::new_v4(); + let parsed = parse_domains(&format!( + "{off}:off,{shadow}:shadow,{verify}:verify_only,{enforce}:enforce" + )) + .expect("valid exact modes"); + assert_eq!( + authoritative_domains(&parsed), + vec![CommunityId::from_uuid(enforce)] + ); + let projection_domains = projection_reconciliation_domains(&parsed) + .into_iter() + .collect::>(); + assert_eq!(projection_domains.len(), 1); + assert!(!projection_domains.contains(&CommunityId::from_uuid(off))); + assert!(!projection_domains.contains(&CommunityId::from_uuid(shadow))); + assert!(!projection_domains.contains(&CommunityId::from_uuid(verify))); + assert!(projection_domains.contains(&CommunityId::from_uuid(enforce))); + } + + #[test] + fn activated_domain_cannot_be_omitted_or_downgraded() { + let activated = CommunityId::from_uuid(uuid::Uuid::new_v4()); + assert!(matches!( + validate_activated_domain_configuration(&HashMap::new(), [activated]), + Err(ProductionRuntimeError::ActivatedDomainDowngrade) + )); + for mode in [ + AuthorizationMode::Off, + AuthorizationMode::Shadow, + AuthorizationMode::VerifyOnly, + ] { + let configured = HashMap::from([(activated, mode)]); + assert!(matches!( + validate_activated_domain_configuration(&configured, [activated]), + Err(ProductionRuntimeError::ActivatedDomainDowngrade) + )); + } + let configured = HashMap::from([(activated, AuthorizationMode::Enforce)]); + validate_activated_domain_configuration(&configured, [activated]) + .expect("exact Enforce configuration preserves one-way activation"); + } + + #[test] + fn exact_provider_registry_has_no_fallback() { + let configured = CommunityId::from_uuid(uuid::Uuid::new_v4()); + let absent = CommunityId::from_uuid(uuid::Uuid::new_v4()); + let provider: Arc = Arc::new(SyntheticUnavailableProvider); + let registry = ProductionProviderRegistry::new([(configured, provider)]) + .expect("exact provider registry"); + assert!(registry.provider_for(configured).is_ok()); + assert!(matches!( + registry.provider_for(absent), + Err(ProductionRuntimeError::ProviderMissing) + )); + } + + #[test] + 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"); + assert!(matches!( + validate_provider_coverage(&configured, &ProductionProviderRegistry::default()), + Err(ProductionRuntimeError::ProviderMissing) + )); + let provider: Arc = Arc::new(SyntheticUnavailableProvider); + let providers = ProductionProviderRegistry::new([(enforce, provider)]) + .expect("exact provider registry"); + validate_provider_coverage(&configured, &providers) + .expect("an exact Enforce provider reaches production construction"); + } + + #[test] + fn stock_binary_uses_the_single_production_installation_boundary() { + let main = include_str!("../main.rs"); + assert!(main.contains("production::install_from_environment(&state)")); + assert!(!main.contains("production::build_from_environment(&state)")); + assert!(!main.contains("migration::prepare_postgres_authority(&state")); + let install = main + .find("production::install_from_environment(&state)") + .expect("single production installation"); + let cutover_verify = main + .find("migration::require_reconciled_authority(&state") + .expect("read-only cutover verification"); + assert!(install < cutover_verify); + } + + #[test] + fn enforce_domain_activation_precedes_snapshot_and_transport_reachability() { + let source = include_str!("production.rs"); + let activation = source + .find("activate_enforcing_domains(&state.db") + .expect("durable activation is part of construction"); + let snapshot = source + .find(".initialize_domains(enforcing_domains.iter().copied())") + .expect("invalidation snapshot is initialized"); + let transport = source + .find("ProtectedTransportRuntime::new(transports, resolver, clock)") + .expect("transport is constructed"); + assert!(activation < snapshot); + assert!(snapshot < transport); + } + + #[test] + fn production_supervises_cross_replica_authorization_hints() { + let source = include_str!("production.rs"); + let worker_set = source + .find("let mut workers = tokio::task::JoinSet::new()") + .expect("protected worker supervisor"); + let invalidation_runtime = source + .find("invalidation_worker.run().await") + .expect("durable invalidation runtime worker"); + let redis_subscriber = source + .find("run_authorization_invalidation_subscriber()") + .expect("cross-replica authorization hint subscriber"); + let supervisor = source + .find("let completion = workers.join_next().await") + .expect("fail-closed worker supervisor"); + + assert!(worker_set < invalidation_runtime); + assert!(worker_set < redis_subscriber); + assert!(invalidation_runtime < supervisor); + assert!(redis_subscriber < supervisor); + } + + #[test] + fn production_reconciles_projection_before_reachability_and_supervises_retry_worker() { + let source = include_str!("production.rs"); + let projection_domains = source + .find("let projection_domains = projection_reconciliation_domains(&configured)") + .expect("exact projection domain selection"); + let startup = source + .find("reconcile_public_projection_retirements_startup(") + .expect("startup reconciliation"); + let transport = source + .find("ProtectedTransportRuntime::new(transports, resolver, clock)") + .expect("transport construction"); + let worker_set = source + .find("let mut workers = tokio::task::JoinSet::new()") + .expect("protected worker supervisor"); + let retry_worker = source + .find("run_public_projection_retirement_reconciliation(") + .expect("continuous projection retry worker"); + let supervisor = source + .find("let completion = workers.join_next().await") + .expect("fail-closed worker supervisor"); + + assert!(projection_domains < startup); + assert!(startup < transport); + assert!(worker_set < retry_worker); + assert!(retry_worker < supervisor); + } + + #[test] + fn restore_bootstraps_are_exact_and_non_nil() { + let domain = CommunityId::from_uuid(uuid::Uuid::new_v4()); + let bootstrap = uuid::Uuid::new_v4(); + assert_eq!( + parse_restore_bootstraps(&format!("{domain}={bootstrap}"), &[domain]) + .expect("exact bootstrap"), + vec![(domain, bootstrap)] + ); + assert!(matches!( + parse_restore_bootstraps("", &[domain]), + Err(ProductionRuntimeError::RestoreBootstrapMissing) + )); + assert!( + parse_restore_bootstraps(&format!("{domain}={}", uuid::Uuid::nil()), &[domain]) + .is_err() + ); + } +} diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 6271d6cdec..8d4baa3a99 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -16,7 +16,7 @@ use tokio_util::sync::CancellationToken; use uuid::Uuid; use buzz_audit::AuditService; -use buzz_auth::{AuthService, Nip98ReplayGuard}; +use buzz_auth::{AuthService, Nip98ReplayGuard, VerifiedFederatedAssertion, VerifiedNostrProof}; use buzz_core::tenant::TenantContext; use buzz_core::CommunityId; use buzz_db::Db; @@ -32,6 +32,7 @@ use deadpool_redis; use crate::audio::AudioRoomManager; use crate::config::Config; use crate::connection::ConnectionSubscriptions; +use crate::connection::OutboundData; use crate::corporate_identity::CorporateIdentityService; use crate::subscription::SubscriptionRegistry; @@ -41,7 +42,7 @@ type ScopedRateLimiter = DashMap; /// Per-connection entry in the connection manager. struct ConnEntry { - tx: mpsc::Sender, + tx: mpsc::Sender, /// Control-frame sender, drained ahead of data and before cancel wins in /// the send loop. Used to deliver a ban-disconnect frame that must reach /// the client before the socket is closed (see [`ConnectionManager::disconnect_pubkey`]). @@ -55,9 +56,57 @@ struct ConnEntry { backpressure_count: Arc, subscriptions: ConnectionSubscriptions, authenticated_pubkey: Arc>>>, + authenticated_owner_pubkey: Arc>>>, + /// Sealed NIP-42 proof retained for every protected operation on this + /// connection. Legacy test registrations may intentionally leave it empty. + verified_nostr_proof: Arc>>>, + /// Current direct federated evidence sealed during authentication. + verified_federated_assertion: Arc>>>, + /// The enforcing authority and its hard-expiry task share one lock so + /// concurrent operations can only tighten, never extend, the session. + protected_session: Arc>, grace_limit: u8, } +struct ProtectedSessionExpiryTask { + deadline: u64, + cancel: CancellationToken, +} + +#[derive(Default)] +struct ProtectedSessionState { + /// Retaining this value keeps its invalidation observer registered until + /// tighter authority replaces it or the connection is removed. + authority: Option>, + expiry: Option, +} + +fn should_replace_protected_session_deadline(current: Option, candidate: u64) -> bool { + current.is_none_or(|current| candidate < current) +} + +fn protected_session_wake_at_from_samples( + deadline: u64, + monotonic_anchor: tokio::time::Instant, + wall_now: std::time::Duration, + coarse_delay: std::time::Duration, +) -> Option { + let wall_remaining = std::time::Duration::from_secs(deadline).checked_sub(wall_now)?; + let conservative_coarse = coarse_delay.saturating_sub(std::time::Duration::from_secs(1)); + monotonic_anchor.checked_add(wall_remaining.min(conservative_coarse)) +} + +fn protected_session_wake_at( + deadline: u64, + monotonic_anchor: tokio::time::Instant, + coarse_delay: std::time::Duration, +) -> Option { + let wall_now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .ok()?; + protected_session_wake_at_from_samples(deadline, monotonic_anchor, wall_now, coarse_delay) +} + /// Community-scoped lifecycle registry shared by every long-lived socket type. /// /// A handler registers before durable active-state revalidation. Archival after @@ -206,7 +255,7 @@ impl ConnectionManager { pub fn register( &self, conn_id: Uuid, - tx: mpsc::Sender, + tx: mpsc::Sender, ctrl_tx: mpsc::Sender, cancel: CancellationToken, community_id: CommunityId, @@ -226,6 +275,12 @@ impl ConnectionManager { backpressure_count, subscriptions, authenticated_pubkey: Arc::new(std::sync::RwLock::new(None)), + authenticated_owner_pubkey: Arc::new(std::sync::RwLock::new(None)), + verified_nostr_proof: Arc::new(std::sync::RwLock::new(None)), + verified_federated_assertion: Arc::new(std::sync::RwLock::new(None)), + protected_session: Arc::new( + std::sync::Mutex::new(ProtectedSessionState::default()), + ), grace_limit, }, ); @@ -241,7 +296,64 @@ impl ConnectionManager { /// Removes a connection from the registry. pub fn deregister(&self, conn_id: Uuid) { - self.connections.remove(&conn_id); + if let Some((_, entry)) = self.connections.remove(&conn_id) { + Self::clear_protected_session(&entry); + } + } + + fn clear_protected_session(entry: &ConnEntry) { + if let Ok(mut session) = entry.protected_session.lock() { + if let Some(task) = session.expiry.take() { + task.cancel.cancel(); + } + session.authority = None; + } else { + entry.cancel.cancel(); + } + } + + /// Atomically retain only authority with an earlier hard deadline. + /// + /// The comparison, task replacement, and authority replacement must stay + /// in this critical section: WebSocket handlers for one connection run + /// concurrently and a split read/install would permit lease extension. + fn retain_earlier_protected_session( + entry: &ConnEntry, + deadline: u64, + wake_at: tokio::time::Instant, + authority: Option>, + after_read_hook: Option<&dyn Fn()>, + ) -> bool { + if let Ok(mut session) = entry.protected_session.lock() { + let current = session.expiry.as_ref().map(|task| task.deadline); + if let Some(hook) = after_read_hook { + hook(); + } + if !should_replace_protected_session_deadline(current, deadline) { + return false; + } + if let Some(previous) = session.expiry.take() { + previous.cancel.cancel(); + } + let expiry_task = CancellationToken::new(); + let expiry_cancel = expiry_task.clone(); + let connection_cancel = entry.cancel.clone(); + tokio::spawn(async move { + tokio::select! { + _ = expiry_cancel.cancelled() => {} + _ = tokio::time::sleep_until(wake_at) => connection_cancel.cancel(), + } + }); + session.expiry = Some(ProtectedSessionExpiryTask { + deadline, + cancel: expiry_task, + }); + session.authority = authority; + true + } else { + entry.cancel.cancel(); + false + } } /// Record the authenticated pubkey for a connection after NIP-42 succeeds. @@ -250,6 +362,81 @@ impl ConnectionManager { if let Ok(mut slot) = entry.authenticated_pubkey.write() { *slot = Some(pubkey_bytes); } + if let Ok(mut slot) = entry.authenticated_owner_pubkey.write() { + *slot = None; + } + if let Ok(mut slot) = entry.verified_nostr_proof.write() { + *slot = None; + } + if let Ok(mut slot) = entry.verified_federated_assertion.write() { + *slot = None; + } + Self::clear_protected_session(&entry); + } + } + + /// Record sealed connection evidence and derive actor/owner indexes from it. + pub fn set_authenticated_authority( + &self, + conn_id: Uuid, + proof: Arc, + assertion: Option>, + ) { + if let Some(entry) = self.connections.get(&conn_id) { + if let Ok(mut slot) = entry.authenticated_pubkey.write() { + *slot = Some(proof.actor_pubkey().to_bytes().to_vec()); + } + if let Ok(mut slot) = entry.authenticated_owner_pubkey.write() { + *slot = proof + .verified_delegation() + .map(|delegation| delegation.owner_pubkey().to_bytes().to_vec()); + } + if let Ok(mut slot) = entry.verified_nostr_proof.write() { + *slot = Some(proof); + } + if let Ok(mut slot) = entry.verified_federated_assertion.write() { + *slot = assertion; + } + Self::clear_protected_session(&entry); + } + } + + /// Retain the latest enforcing authority for an established connection. + /// Legacy and observational results clear any older authority without + /// creating a new invalidation registration. + pub fn retain_protected_session_authority( + &self, + conn_id: Uuid, + authority: &crate::authorization_runtime::transport::ProtectedAuthorization, + ) { + if let Some(entry) = self.connections.get(&conn_id) { + if authority.is_enforcing() { + let candidate = authority.expires_at().unwrap_or_default(); + // Anchor monotonic time before consulting the injected + // whole-second authorization clock. Time spent sampling or + // installing the task must consume authority, never extend it. + let monotonic_anchor = tokio::time::Instant::now(); + match authority.expiry_delay() { + Ok(Some(delay)) => { + if let Some(wake_at) = + protected_session_wake_at(candidate, monotonic_anchor, delay) + { + Self::retain_earlier_protected_session( + &entry, + candidate, + wake_at, + Some(Arc::new(authority.clone())), + None, + ); + } else { + entry.cancel.cancel(); + } + } + Ok(None) | Err(_) => entry.cancel.cancel(), + } + } else { + Self::clear_protected_session(&entry); + } } } @@ -290,6 +477,37 @@ impl ConnectionManager { .and_then(|entry| entry.authenticated_pubkey.read().ok()?.clone()) } + /// Return the sealed verifier evidence recorded for a connection. + pub fn authority_for_conn(&self, conn_id: Uuid) -> Option> { + self.connections + .get(&conn_id) + .and_then(|entry| entry.verified_nostr_proof.read().ok()?.clone()) + } + + /// Return current direct federated evidence recorded for a connection. + pub fn federated_assertion_for_conn( + &self, + conn_id: Uuid, + ) -> Option> { + self.connections + .get(&conn_id) + .and_then(|entry| entry.verified_federated_assertion.read().ok()?.clone()) + } + + /// Return the server-owned cancellation token for a live connection. + pub fn cancellation_for_conn(&self, conn_id: Uuid) -> Option { + self.connections + .get(&conn_id) + .map(|entry| entry.cancel.clone()) + } + + /// Cancel one connection after protected session authority expires. + pub fn cancel_connection(&self, conn_id: Uuid) { + if let Some(entry) = self.connections.get(&conn_id) { + entry.cancel.cancel(); + } + } + /// Disconnect every live connection authenticated as `pubkey` **in /// `community`**, delivering a final `OK false` frame carrying `reason` /// before closing. @@ -435,7 +653,7 @@ impl ConnectionManager { /// On sustained backpressure (>grace_limit consecutive full buffers), /// cancels the connection. Transient stalls get a warning only. pub fn send_to(&self, conn_id: Uuid, msg: String) -> bool { - self.try_send_ws_message(conn_id, WsMessage::Text(msg.into())) + self.try_send_outbound(conn_id, OutboundData::plain(WsMessage::Text(msg.into()))) } /// Sends an already-serialized UTF-8 text payload to the given connection. @@ -445,10 +663,73 @@ impl ConnectionManager { pub fn send_to_text_bytes(&self, conn_id: Uuid, msg: Arc) -> bool { let text = WsUtf8Bytes::try_from(Bytes::clone(msg.as_ref())) .expect("relay fan-out frames are serialized UTF-8 JSON"); - self.try_send_ws_message(conn_id, WsMessage::Text(text)) + self.try_send_outbound(conn_id, OutboundData::plain(WsMessage::Text(text))) } - fn try_send_ws_message(&self, conn_id: Uuid, msg: WsMessage) -> bool { + /// Queue protected text and retain its exact authority until socket drain. + pub fn send_to_text_bytes_protected( + &self, + conn_id: Uuid, + msg: Arc, + authority: Arc, + ) -> bool { + let text = WsUtf8Bytes::try_from(Bytes::clone(msg.as_ref())) + .expect("relay fan-out frames are serialized UTF-8 JSON"); + self.try_send_outbound( + conn_id, + OutboundData::protected(WsMessage::Text(text), authority), + ) + } + + /// Queue output guarded by both the emitting operation and the recipient's + /// current read authority until socket drain. + pub fn send_to_text_bytes_protected_pair( + &self, + conn_id: Uuid, + msg: Arc, + sender: Arc, + recipient: Arc, + ) -> bool { + let text = WsUtf8Bytes::try_from(Bytes::clone(msg.as_ref())) + .expect("relay fan-out frames are serialized UTF-8 JSON"); + self.try_send_outbound( + conn_id, + OutboundData::protected_pair(WsMessage::Text(text), sender, recipient), + ) + } + + /// Queue output behind an arbitrary asynchronous sender fence. + pub(crate) fn send_to_text_bytes_guarded( + &self, + conn_id: Uuid, + msg: Arc, + authority: Arc, + ) -> bool { + let text = WsUtf8Bytes::try_from(Bytes::clone(msg.as_ref())) + .expect("relay fan-out frames are serialized UTF-8 JSON"); + self.try_send_outbound( + conn_id, + OutboundData::guarded(WsMessage::Text(text), authority), + ) + } + + /// Queue output behind a remote sender fence and local recipient fence. + pub(crate) fn send_to_text_bytes_guarded_pair( + &self, + conn_id: Uuid, + msg: Arc, + sender: Arc, + recipient: Arc, + ) -> bool { + let text = WsUtf8Bytes::try_from(Bytes::clone(msg.as_ref())) + .expect("relay fan-out frames are serialized UTF-8 JSON"); + self.try_send_outbound( + conn_id, + OutboundData::guarded_pair(WsMessage::Text(text), sender, recipient), + ) + } + + fn try_send_outbound(&self, conn_id: Uuid, msg: OutboundData) -> bool { if let Some(entry) = self.connections.get(&conn_id) { let conn = entry.value(); match conn.tx.try_send(msg) { @@ -628,6 +909,36 @@ pub struct AppState { /// byte-identically to a relay without the mesh. Access via /// [`AppState::mesh`]. pub mesh: Arc>, + /// Optional exact-domain protected-transport runtime. + /// + /// Unset preserves legacy behavior. Once installed it is immutable for the + /// process lifetime so request data cannot switch provider policy. + pub protected_transport: Arc< + std::sync::OnceLock< + Arc, + >, + >, + /// Deployment-verified ingress provenance for direct identity assertions. + /// + /// Unset by the stock binary. Header presence alone is never trusted. + pub identity_assertion_provenance: Arc< + std::sync::OnceLock< + Arc, + >, + >, + /// Complete-stack-gated NIP-FI discovery. The stock binary leaves this + /// unset; only an immutable reviewed conformance input can construct it. + pub nip_fi_discovery: Arc>, + /// Optional RFC-gated, dedicated client-status presentation runtime. + pub client_status_runtime: Arc< + std::sync::OnceLock< + Arc, + >, + >, + /// Independent version witness installed with protected authorization. + pub restore_protection: Arc< + std::sync::OnceLock>, + >, } impl AppState { @@ -804,6 +1115,11 @@ impl AppState { // `crates/buzz-test-client` once those land). tracer: Arc::new(crate::conformance::NoopTracer), mesh: Arc::new(std::sync::OnceLock::new()), + protected_transport: Arc::new(std::sync::OnceLock::new()), + identity_assertion_provenance: Arc::new(std::sync::OnceLock::new()), + nip_fi_discovery: Arc::new(std::sync::OnceLock::new()), + client_status_runtime: Arc::new(std::sync::OnceLock::new()), + restore_protection: Arc::new(std::sync::OnceLock::new()), }; ( state, @@ -820,6 +1136,99 @@ impl AppState { self.mesh.get() } + /// Install immutable protected-transport policy exactly once. + pub fn install_protected_transport( + &self, + runtime: Arc, + ) -> Result<(), Arc> { + self.protected_transport.set(runtime) + } + + /// Current protected-transport runtime, if configured. + pub fn protected_transport( + &self, + ) -> Option<&Arc> { + self.protected_transport.get() + } + + /// Install the immutable deployment-owned ingress provenance adapter. + pub fn install_identity_assertion_provenance( + &self, + verifier: Arc, + ) -> Result<(), Arc> { + self.identity_assertion_provenance.set(verifier) + } + + /// Return the installed ingress provenance adapter, if any. + pub fn identity_assertion_provenance( + &self, + ) -> Option<&Arc> { + self.identity_assertion_provenance.get() + } + + /// Install exact-revision NIP-FI discovery readiness once. + pub fn install_nip_fi_discovery( + &self, + ready: crate::nip11::ConformanceReadyNipFiDiscovery, + ) -> Result<(), crate::nip11::ConformanceReadyNipFiDiscovery> { + self.nip_fi_discovery.set(ready) + } + + /// Return complete-stack-gated NIP-FI discovery, if installed. + pub fn nip_fi_discovery(&self) -> Option<&crate::nip11::ConformanceReadyNipFiDiscovery> { + self.nip_fi_discovery.get() + } + + /// Install an externally approved dedicated client-status runtime once. + pub fn install_client_status_runtime( + &self, + runtime: Arc, + ) -> Result<(), Arc> { + self.client_status_runtime.set(runtime) + } + + /// Return the approved client-status runtime, if installed. + pub fn client_status_runtime( + &self, + ) -> Option<&Arc> { + self.client_status_runtime.get() + } + + /// Install the restore-independent version witness exactly once. + pub fn install_restore_protection( + &self, + runtime: Arc, + ) -> Result<(), Arc> { + self.restore_protection.set(runtime) + } + + /// Current restore-independent version witness, if protected mode exists. + pub fn restore_protection( + &self, + ) -> Option<&Arc> { + self.restore_protection.get() + } + + /// Database UUIDs for exact domains where autonomous effects must not run. + pub fn enforcing_protected_domain_ids(&self) -> Vec { + self.protected_transport() + .map(|runtime| { + runtime + .enforcing_domains() + .into_iter() + .map(|domain| *domain.as_uuid()) + .collect() + }) + .unwrap_or_default() + } + + /// Whether one exact server-resolved domain is in authoritative Enforce. + pub fn is_protected_enforcing(&self, domain: CommunityId) -> bool { + self.protected_transport() + .and_then(|runtime| runtime.mode_for_domain(domain)) + == Some(crate::authorization_runtime::finalization::AuthorizationMode::Enforce) + } + /// Record an event ID as locally-published for dedup, scoped to the /// community it was fanned out in. Called before Redis publish so the /// multi-node consumer can skip the echo for *this* community only — a @@ -1237,7 +1646,7 @@ mod tests { ) -> ( ConnectionManager, Uuid, - mpsc::Receiver, + mpsc::Receiver, mpsc::Receiver, CancellationToken, Arc, @@ -1261,6 +1670,228 @@ mod tests { (mgr, conn_id, rx, ctrl_rx, cancel, bp) } + #[tokio::test(start_paused = true)] + async fn protected_session_deadline_is_conservative_and_anchor_bound() { + let anchor = tokio::time::Instant::now(); + let wake_at = protected_session_wake_at_from_samples( + 102, + anchor, + std::time::Duration::from_millis(100_900), + std::time::Duration::from_secs(2), + ) + .expect("future deadline"); + assert_eq!( + wake_at.duration_since(anchor), + std::time::Duration::from_secs(1), + "whole-second authority is shortened conservatively instead of rounded late" + ); + + // Simulate work between authority-clock sampling and task install. The + // absolute monotonic wake stays tied to the earlier anchor. + tokio::time::advance(std::time::Duration::from_millis(750)).await; + let (mgr, conn_id, _rx, _ctrl_rx, cancel, _bp) = setup_conn(8); + let entry = mgr + .connections + .get(&conn_id) + .expect("registered connection"); + ConnectionManager::retain_earlier_protected_session(&entry, 102, wake_at, None, None); + drop(entry); + + tokio::time::advance(std::time::Duration::from_millis(249)).await; + tokio::task::yield_now().await; + assert!(!cancel.is_cancelled()); + tokio::time::advance(std::time::Duration::from_millis(1)).await; + tokio::task::yield_now().await; + assert!(cancel.is_cancelled()); + } + + #[tokio::test(start_paused = true)] + async fn protected_session_expiry_never_extends_and_disconnect_cleans_task() { + let (mgr, conn_id, _rx, _ctrl_rx, cancel, _bp) = setup_conn(8); + { + let entry = mgr + .connections + .get(&conn_id) + .expect("registered connection"); + ConnectionManager::retain_earlier_protected_session( + &entry, + 10, + tokio::time::Instant::now() + std::time::Duration::from_secs(10), + None, + None, + ); + } + tokio::task::yield_now().await; + tokio::time::advance(std::time::Duration::from_secs(9)).await; + assert!(!cancel.is_cancelled()); + + assert!(!should_replace_protected_session_deadline(Some(10), 20)); + assert!(!should_replace_protected_session_deadline(Some(10), 10)); + tokio::time::advance(std::time::Duration::from_secs(1)).await; + tokio::task::yield_now().await; + assert!( + cancel.is_cancelled(), + "ordinary traffic cannot extend the first hard session deadline" + ); + + let (mgr, conn_id, _rx, _ctrl_rx, cancel, _bp) = setup_conn(8); + { + let entry = mgr + .connections + .get(&conn_id) + .expect("registered connection"); + ConnectionManager::retain_earlier_protected_session( + &entry, + 20, + tokio::time::Instant::now() + std::time::Duration::from_secs(20), + None, + None, + ); + } + tokio::task::yield_now().await; + tokio::time::advance(std::time::Duration::from_secs(5)).await; + assert!(should_replace_protected_session_deadline(Some(20), 10)); + { + let entry = mgr + .connections + .get(&conn_id) + .expect("registered connection"); + ConnectionManager::retain_earlier_protected_session( + &entry, + 10, + tokio::time::Instant::now() + std::time::Duration::from_secs(5), + None, + None, + ); + } + tokio::task::yield_now().await; + tokio::time::advance(std::time::Duration::from_secs(5)).await; + tokio::task::yield_now().await; + assert!( + cancel.is_cancelled(), + "a shorter authority tightens the hard deadline" + ); + + let (mgr, conn_id, _rx, _ctrl_rx, cancel, _bp) = setup_conn(8); + { + let entry = mgr + .connections + .get(&conn_id) + .expect("registered connection"); + ConnectionManager::retain_earlier_protected_session( + &entry, + 5, + tokio::time::Instant::now() + std::time::Duration::from_secs(5), + None, + None, + ); + } + tokio::task::yield_now().await; + mgr.deregister(conn_id); + tokio::time::advance(std::time::Duration::from_secs(5)).await; + assert!( + !cancel.is_cancelled(), + "disconnect removes the obsolete expiry task" + ); + } + + #[tokio::test] + async fn concurrent_protected_operations_cannot_replace_an_earlier_deadline() { + let (mgr, conn_id, _rx, _ctrl_rx, _cancel, _bp) = setup_conn(8); + let mgr = Arc::new(mgr); + let entry = mgr + .connections + .get(&conn_id) + .expect("registered connection"); + let protected_session = Arc::clone(&entry.protected_session); + drop(entry); + + let runtime = tokio::runtime::Handle::current(); + let short_authority = + Arc::new(crate::authorization_runtime::transport::ProtectedAuthorization::Legacy); + let long_authority = + Arc::new(crate::authorization_runtime::transport::ProtectedAuthorization::Legacy); + let (short_locked, wait_for_short_lock) = std::sync::mpsc::channel(); + let (release_short, wait_for_release) = std::sync::mpsc::channel(); + let short_manager = Arc::clone(&mgr); + let retained_short_authority = Arc::clone(&short_authority); + let short_runtime = runtime.clone(); + let short = std::thread::spawn(move || { + let _runtime = short_runtime.enter(); + let entry = short_manager + .connections + .get(&conn_id) + .expect("registered connection"); + let hook = || { + short_locked.send(()).expect("test still waiting"); + wait_for_release.recv().expect("test releases first lock"); + }; + ConnectionManager::retain_earlier_protected_session( + &entry, + 10, + tokio::time::Instant::now() + std::time::Duration::from_secs(10), + Some(retained_short_authority), + Some(&hook), + ); + }); + wait_for_short_lock + .recv_timeout(std::time::Duration::from_secs(1)) + .expect("short contender holds the comparison/install lock"); + + let (long_attempting, wait_for_long_attempt) = std::sync::mpsc::channel(); + let (long_read, wait_for_long_read) = std::sync::mpsc::channel(); + let long_manager = Arc::clone(&mgr); + let long_runtime = runtime.clone(); + let long = std::thread::spawn(move || { + let _runtime = long_runtime.enter(); + long_attempting.send(()).expect("test still waiting"); + let entry = long_manager + .connections + .get(&conn_id) + .expect("registered connection"); + let hook = || { + long_read.send(()).expect("test still waiting"); + }; + ConnectionManager::retain_earlier_protected_session( + &entry, + 20, + tokio::time::Instant::now() + std::time::Duration::from_secs(20), + Some(long_authority), + Some(&hook), + ); + }); + wait_for_long_attempt + .recv_timeout(std::time::Duration::from_secs(1)) + .expect("long contender reached the atomic helper"); + assert!( + matches!( + wait_for_long_read.recv_timeout(std::time::Duration::from_millis(50)), + Err(std::sync::mpsc::RecvTimeoutError::Timeout) + ), + "a second contender cannot read the deadline before the first install completes" + ); + release_short.send(()).expect("short contender is waiting"); + short.join().expect("short retention thread"); + wait_for_long_read + .recv_timeout(std::time::Duration::from_secs(1)) + .expect("long contender reads only after the short install"); + long.join().expect("long retention thread"); + + let session = protected_session.lock().expect("protected session lock"); + assert_eq!( + session.expiry.as_ref().map(|task| task.deadline), + Some(10), + "the minimum concurrent deadline is retained regardless of completion order" + ); + assert!( + session + .authority + .as_ref() + .is_some_and(|authority| Arc::ptr_eq(authority, &short_authority)), + "the retained authority belongs to the minimum-deadline contender" + ); + } + async fn test_state() -> Arc { let mut config = crate::config::Config::from_env().expect("default config loads"); config.require_relay_membership = false; @@ -1363,7 +1994,7 @@ mod tests { "test.local".to_string(), ), remote_addr: "127.0.0.1:1234".parse().unwrap(), - corporate_identity_jwt: None, + corporate_identity_assertion: None, auth_state: RwLock::new(AuthState::Failed), subscriptions: Arc::new(Mutex::new(HashMap::new())), send_tx: tx.clone(),