From 05d7eca1ae1068f9476a1bc87361b221648f9982 Mon Sep 17 00:00:00 2001 From: Duncan Date: Mon, 17 Aug 2026 23:21:22 -0400 Subject: [PATCH] feat(egress): durable owner-identity capabilities + huddle session (C2a) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add OwnerIdentityCapability

over the existing egress registry: a generation-stamped, registry-tracked handle for authority that outlives the bounded lease that derived it. Two policies land: Session (authenticated connections — the huddle audio socket, later the frontend relay WS) and Bearer (pre-minted Blossom headers, threaded in a follow-up). Each capability is registered with its revocation handle (the session's cancellation token; the bearer's registry id) so the C5 coordinator barrier only invokes what C2 registered — it never retrofits the registry schema. admit_exercise() validates BOTH current egress admission AND capability_generation == current identity-persistence generation immediately before each transmission, so a stale capability sends zero bytes. Issuance runs under a bounded lease (the signing that derives the capability is an ordinary leased operation). The huddle audio socket is threaded: the NIP-42 auth signs under a bounded lease (dropped before the joined-await), the session capability is registered with the connection's cancel token, and the send task validates it before every frame batch — a frame cannot ride the established peer after an identity transition supersedes it. C2 builds substrate only: the coordinator revocation barrier (revoke_durable_capabilities_before) and drain wiring defer to C5 with the egress drain, gated behind the same generation bump C5 introduces. Per-item allow(dead_code) with the C5-consumer comment; C5's zero-allow confirmation extends to these. generation never bumps until C5, so this is behavior-preserving. 8 new unit tests (2438 lib pass): generation-stamp, exercise admits when live+current, stale-capability zero-bytes controls (generation bump, drain, latch) for both kinds, barrier revokes old-generation only, registration- completeness + deregister-on-drop, and a no-transition control. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- desktop/src-tauri/src/huddle/relay_api.rs | 40 ++ .../src-tauri/src/owner_identity_egress.rs | 409 ++++++++++++++++++ 2 files changed, 449 insertions(+) diff --git a/desktop/src-tauri/src/huddle/relay_api.rs b/desktop/src-tauri/src/huddle/relay_api.rs index 9baccf8ec..fd7215b5a 100644 --- a/desktop/src-tauri/src/huddle/relay_api.rs +++ b/desktop/src-tauri/src/huddle/relay_api.rs @@ -98,6 +98,13 @@ pub(crate) async fn connect_audio_relay( nostr::Tag::parse(["relay", &relay_url]).map_err(|e| format!("tag relay: {e}"))?, nostr::Tag::parse(["challenge", &challenge]).map_err(|e| format!("tag challenge: {e}"))?, ]; + // Issuance runs under a bounded egress lease: the NIP-42 auth that + // establishes this session's connection-lifetime authority is an ordinary + // leased sign→send operation (spec L4569-4570). The lease is held only + // across sign → send-auth and dropped before we await `joined`; the + // durable session capability registered below carries the authority + // forward for the unsigned frames the send task emits. + let auth_lease = crate::owner_identity_egress::try_admit_owner_identity_egress().await?; let event = nostr::EventBuilder::new(nostr::Kind::Custom(22242), "") .tags(tags) .sign_with_keys(&keys) @@ -120,6 +127,10 @@ pub(crate) async fn connect_audio_relay( .send(WsMsg::Text(auth_msg.to_string().into())) .await .map_err(|e| format!("send auth: {e}"))?; + // Sign → auth-send complete: the bounded lease has served its purpose. + // Drop it before awaiting `joined` so it does not span the wait (spec + // L4505-4508); the session capability below carries authority forward. + drop(auth_lease); let initial_peers: Vec<(u8, String)> = tokio::time::timeout(HANDSHAKE_TIMEOUT, async { loop { @@ -162,6 +173,12 @@ pub(crate) async fn connect_audio_relay( let cancel = CancellationToken::new(); let cancel_clone = cancel.clone(); + // Register the durable session capability, stamped with the current + // identity-persistence generation and carrying THIS connection's + // cancellation token as its revocation handle. The send task validates it + // before every frame; the C5 coordinator barrier invokes the registered + // token to tear the socket down when an identity transition supersedes it. + let session = crate::owner_identity_egress::register_owner_session(cancel.clone()); let (pcm_tx, pcm_rx) = tokio::sync::mpsc::channel::>(50); let output_device_name = state .huddle_audio @@ -181,6 +198,7 @@ pub(crate) async fn connect_audio_relay( tts_cancel, tts_active, output_device_name, + session, }) .await { @@ -215,6 +233,12 @@ struct AudioRelayPipelineArgs { tts_cancel: Arc, tts_active: Arc, output_device_name: Option, + /// The durable owner-identity session capability for this connection. + /// Validated before every frame send so a frame cannot ride the + /// established peer after an identity transition has superseded it. + session: crate::owner_identity_egress::OwnerIdentityCapability< + crate::owner_identity_egress::SessionPolicy, + >, } async fn audio_relay_pipeline(args: AudioRelayPipelineArgs) -> Result<(), String> { @@ -228,6 +252,7 @@ async fn audio_relay_pipeline(args: AudioRelayPipelineArgs) -> Result<(), String tts_cancel, tts_active, output_device_name, + session, } = args; let mut encoder = opus::Encoder::new(48000, opus::Channels::Mono, opus::Application::Voip) @@ -247,6 +272,10 @@ async fn audio_relay_pipeline(args: AudioRelayPipelineArgs) -> Result<(), String let cancel_send = cancel.clone(); let send_task = tokio::spawn(async move { + // The session capability moves into the send task: it lives exactly as + // long as the frames it authorizes, and dropping it (task exit) + // deregisters it from the egress registry. + let session = session; use super::wire::{audio_level_dbov, FrameHeader, V2_HEADER_LEN}; let mut encoder = encoder; // Move encoder into task. const FRAME_SAMPLES: usize = 960; @@ -272,6 +301,17 @@ async fn audio_relay_pipeline(args: AudioRelayPipelineArgs) -> Result<(), String if pcm_bytes.len() % 4 != 0 { continue; // Malformed batch. } + + // Validate the session capability immediately before transmitting + // this batch: a cheap generation compare, not a re-auth. Once an + // identity transition supersedes the generation this session was + // authenticated under, exercise is refused and the connection tears + // down — no frame rides the old peer under an unresolved identity. + if let Err(e) = session.admit_exercise() { + eprintln!("buzz-desktop: huddle audio session revoked: {e}"); + break; + } + let samples: Vec = pcm_bytes .chunks_exact(4) .map(|b| f32::from_le_bytes([b[0], b[1], b[2], b[3]])) diff --git a/desktop/src-tauri/src/owner_identity_egress.rs b/desktop/src-tauri/src/owner_identity_egress.rs index 104edcb6f..320ef53eb 100644 --- a/desktop/src-tauri/src/owner_identity_egress.rs +++ b/desktop/src-tauri/src/owner_identity_egress.rs @@ -82,10 +82,12 @@ //! `ManagedAgentKeyed` sites). //! 5. the four `project_git_workflow` PR-status/merge sends (items 5–8). +use std::collections::HashMap; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{LazyLock, Mutex}; use tokio::sync::Notify; +use tokio_util::sync::CancellationToken; /// Admission state of owner-identity egress — the third recovery state /// (`Indeterminate`) extends the existing `AppState::identity_lost` / @@ -462,6 +464,262 @@ fn admit_managed_agent_after_wait() -> Result { } } +// --------------------------------------------------------------------------- +// Durable owner-identity capabilities (C2 / P30-C1) +// --------------------------------------------------------------------------- + +/// A generation-stamped, registry-tracked owner-identity capability whose +/// authority OUTLIVES the bounded lease that derived it. +/// +/// The bounded [`OwnerIdentityEgressLease`] above covers authority derived and +/// consumed inside one sign → auth → transmit window. But two owner-key +/// operations mint authority that a LATER, separate operation exercises: +/// +/// - **Sessions** ([`SessionPolicy`]) — the huddle audio socket authenticates +/// ONCE with a NIP-42 event, then a long-lived task emits unsigned frames +/// over the established peer indefinitely; the frontend relay WS is the same +/// shape (`create_auth_event` signs the handshake, later frames ride the +/// connection). Cloning the owner keys before the transition cannot express +/// that a later unsigned frame inherits pre-transition authority. +/// - **Bearers** ([`BearerPolicy`]) — `mint_media_get_auth` / +/// `sign_blossom_upload_auth` sign a server-scoped Blossom header that +/// callers attach at LATER HTTP transmissions, up to ten minutes after +/// issuance. The bearer TTL is wider than a transition, so TTL is not a +/// substitute for invalidation. +/// +/// Every durable capability is REGISTERED in this same egress registry and +/// STAMPED with the identity-persistence generation current at issuance +/// (issuance itself runs under a bounded lease — the signing that derives the +/// capability is an ordinary leased operation). Every transmission over it +/// validates, immediately before the irreversible boundary (frame send / +/// header attach), BOTH current egress admission AND +/// `capability_generation == current identity-persistence generation` +/// ([`OwnerIdentityCapability::admit_exercise`]) — a stale capability is +/// refused with zero bytes sent. The type is the only carrier of that stamped +/// generation, so no exercise site can skip the check (there is no raw handle +/// to exercise). +/// +/// The registry additionally holds each capability's REVOCATION HANDLE (the +/// session cancellation token; the bearer's registry id for invalidation), so +/// the C5 coordinator barrier only *invokes* what C2 already registered — it +/// never retrofits the registry schema. Registering the teardown authority is +/// substrate (C2); invoking it at the transition barrier is C5. +#[derive(Debug)] +#[must_use = "a durable owner-identity capability must be validated \ + (admit_exercise) immediately before every transmission over it"] +pub struct OwnerIdentityCapability { + /// Registry id — the key under which this capability's revocation handle + /// lives, so the C5 barrier can invalidate it by generation. + id: u64, + /// The identity-persistence generation current when this capability was + /// issued. Exercise is refused once the registry generation advances past + /// it. + generation: u64, + _policy: std::marker::PhantomData

, +} + +/// A policy distinguishing the durable capability kinds. Zero-sized markers — +/// the shared behavior (registration, generation-stamp, exercise validation, +/// revocation-handle storage) lives on [`OwnerIdentityCapability`]; the policy +/// only names the kind so the constructor inventory and the registry's +/// per-kind revocation are type-directed. +pub trait CapabilityPolicy: std::fmt::Debug + private::Sealed { + /// A human label for diagnostics and the closed-world inventory. + const KIND: &'static str; +} + +/// Authenticated-connection authority (huddle audio socket, frontend relay +/// WS). The registered revocation handle is a [`CancellationToken`] whose +/// cancellation tears the connection down. +#[derive(Debug)] +pub struct SessionPolicy; +/// Pre-minted bearer authority (Blossom `t=get`/`t=upload` headers). The +/// registered revocation handle is the registry id; invalidation removes the +/// entry so a later attach fails admission. +#[derive(Debug)] +pub struct BearerPolicy; + +impl CapabilityPolicy for SessionPolicy { + const KIND: &'static str = "session"; +} +impl CapabilityPolicy for BearerPolicy { + const KIND: &'static str = "bearer"; +} + +mod private { + pub trait Sealed {} + impl Sealed for super::SessionPolicy {} + impl Sealed for super::BearerPolicy {} +} + +/// The revocation authority for one registered durable capability, invoked by +/// the C5 coordinator barrier to tear down old-generation authority before the +/// journal write + durable dispatch. +enum RevocationHandle { + /// Cancel the connection (huddle socket / frontend WS teardown path). + Session(CancellationToken), + /// Bearer invalidation is by-entry: removing the registry entry is the + /// invalidation, so no side-effecting handle is needed. The variant exists + /// so the registry records the capability's kind for the per-kind barrier. + Bearer, +} + +/// Registry of live durable capabilities, keyed by capability id. Guarded by +/// the same lock as the bounded-lease state so registration, exercise +/// validation, and the C5 barrier all linearize against generation bumps. +#[derive(Default)] +struct DurableRegistry { + /// Next capability id. Monotonic; ids are never reused. + next_id: u64, + /// Live capabilities: id → (issued generation, revocation handle). + entries: HashMap, +} + +static DURABLE: LazyLock> = + LazyLock::new(|| Mutex::new(DurableRegistry::default())); + +fn lock_durable() -> std::sync::MutexGuard<'static, DurableRegistry> { + DURABLE.lock().unwrap_or_else(|p| p.into_inner()) +} + +impl OwnerIdentityCapability

{ + /// The identity-persistence generation this capability was issued under. + // Consumed by C5 (barrier introspection) and the C2 tests; remove allow + // when a production caller reads it. + #[allow(dead_code)] + pub fn generation(&self) -> u64 { + self.generation + } + + /// Validate this capability immediately before an irreversible + /// transmission over it (frame send / header attach). Succeeds only when + /// egress admission is `Live` AND the stamped generation still equals the + /// current identity-persistence generation. A stale or drained capability + /// is refused so the caller sends zero bytes. + /// + /// This is a cheap compare, not a re-authentication: the durable + /// capability already proved identity at issuance; exercise only confirms + /// that proof has not been superseded by a transition. + pub fn admit_exercise(&self) -> Result<(), String> { + let inner = lock_inner(); + if inner.state != IdentityPersistenceState::Live { + return Err(format!( + "owner-identity {} capability cannot transmit: egress is {:?}", + P::KIND, + inner.state + )); + } + let current = REGISTRY.generation.load(Ordering::Acquire); + if self.generation != current { + return Err(format!( + "owner-identity {} capability is stale (issued under generation \ + {}, current {}); the identity transitioned and this authority \ + was revoked", + P::KIND, + self.generation, + current + )); + } + Ok(()) + } +} + +impl Drop for OwnerIdentityCapability

{ + fn drop(&mut self) { + // Dropping the capability handle deregisters it — a session whose task + // has ended or a bearer no longer attachable must not linger as a + // revocation target. Barrier invalidation (C5) also removes entries; + // the id is unique and never reused, so a double-remove is a no-op. + lock_durable().entries.remove(&self.id); + } +} + +/// Issue a durable owner-identity session capability, registering its +/// cancellation token so the C5 barrier can tear the connection down. +/// +/// Call this AFTER the authenticating sign/auth has completed under a bounded +/// lease (issuance is an ordinary leased operation, spec L4569–4570) and the +/// connection is established. The returned capability stamps the current +/// generation; the connection's send task validates it via +/// [`OwnerIdentityCapability::admit_exercise`] before each frame. +pub fn register_owner_session(cancel: CancellationToken) -> OwnerIdentityCapability { + let generation = REGISTRY.generation.load(Ordering::Acquire); + let mut durable = lock_durable(); + let id = durable.next_id; + durable.next_id += 1; + durable + .entries + .insert(id, (generation, RevocationHandle::Session(cancel))); + OwnerIdentityCapability { + id, + generation, + _policy: std::marker::PhantomData, + } +} + +/// Issue a durable owner-identity bearer capability, registering it so the C5 +/// barrier can invalidate it by generation. +/// +/// Call this immediately after minting the bearer header under a bounded lease. +/// The returned capability stamps the current generation; every attach site +/// validates it via [`OwnerIdentityCapability::admit_exercise`] before the HTTP +/// dispatch. +// Consumed by the Blossom attach sites in C2b; remove allow when they land. +#[allow(dead_code)] +pub fn register_owner_bearer() -> OwnerIdentityCapability { + let generation = REGISTRY.generation.load(Ordering::Acquire); + let mut durable = lock_durable(); + let id = durable.next_id; + durable.next_id += 1; + durable + .entries + .insert(id, (generation, RevocationHandle::Bearer)); + OwnerIdentityCapability { + id, + generation, + _policy: std::marker::PhantomData, + } +} + +/// Revoke every durable capability issued under a generation older than +/// `winning_generation`: cancel old sessions (invoking each registered +/// [`CancellationToken`]) and drop old bearers (removing their entries so a +/// later attach fails [`OwnerIdentityCapability::admit_exercise`]). Returns the +/// number of capabilities revoked. +/// +/// The C5 coordinator barrier calls this after +/// [`begin_egress_drain`]/[`await_egress_drain`] and BEFORE the journal write + +/// durable B dispatch, so no old-generation session or bearer can transmit +/// across the durable boundary. It only *invokes* the handles C2 registered. +// Consumed by C5 (P25/P28 coordinator barrier); remove allow when C5 lands. +#[allow(dead_code)] +pub fn revoke_durable_capabilities_before(winning_generation: u64) -> usize { + let mut durable = lock_durable(); + let stale: Vec = durable + .entries + .iter() + .filter(|(_, (gen, _))| *gen < winning_generation) + .map(|(id, _)| *id) + .collect(); + for id in &stale { + if let Some((_, RevocationHandle::Session(cancel))) = durable.entries.remove(id) { + cancel.cancel(); + } + } + stale.len() +} + +/// The number of live durable capabilities registered. Used by the +/// registration-completeness assertion (C2) to prove every issued +/// session/bearer is present with a revocation handle the C5 barrier can +/// invoke. +// Consumed by C5 and the C2 registration-completeness tests; remove allow when +// a production caller lands. +#[allow(dead_code)] +pub fn live_durable_capability_count() -> usize { + lock_durable().entries.len() +} + /// Process-global mutex serializing tests that mutate the registry's /// process-global state. Any test that admits a lease, drives a drain, or /// latches must hold this guard for its whole duration and reset via @@ -477,6 +735,11 @@ pub(crate) fn reset_registry_for_test() { let mut inner = lock_inner(); inner.state = IdentityPersistenceState::Live; inner.in_flight = 0; + drop(inner); + let mut durable = lock_durable(); + durable.entries.clear(); + // next_id is monotonic and never resets, mirroring the generation: tests + // assert on presence/count and relative generation, never absolute ids. } /// Mint an owner-identity [`EgressLease`] for tests that exercise a funnel's @@ -740,4 +1003,150 @@ mod tests { ); crate::relay_admission::reset_rate_limit_gate(); } + + // ----------------------------------------------------------------------- + // Durable capability substrate (C2 / P30-C1) + // ----------------------------------------------------------------------- + + #[test] + fn durable_capabilities_stamp_the_current_generation() { + let _g = guard(); + let session = register_owner_session(CancellationToken::new()); + let bearer = register_owner_bearer(); + let current = current_identity_persistence_generation(); + assert_eq!(session.generation(), current); + assert_eq!(bearer.generation(), current); + } + + #[test] + fn durable_capability_exercise_admits_when_live_and_current() { + let _g = guard(); + let session = register_owner_session(CancellationToken::new()); + let bearer = register_owner_bearer(); + assert!(session.admit_exercise().is_ok()); + assert!(bearer.admit_exercise().is_ok()); + } + + // §7 revocation schedule (session): a session issued under A cannot + // transmit after the generation advances to B — the frame send is refused + // with zero bytes. Drives the registry generation directly (C1 drain-test + // pattern); no production transition driver exists until C5. + #[test] + fn stale_session_exercise_refuses_after_a_generation_bump() { + let _g = guard(); + let session = register_owner_session(CancellationToken::new()); + // A transition drains then resumes at the winning generation. + begin_egress_drain().unwrap(); + resume_egress_live(); + assert!( + session.admit_exercise().is_err(), + "a session stamped under A must refuse to transmit after B wins" + ); + } + + // §7 revocation schedule (bearer): same stale-generation control for the + // pre-minted bearer attach path. + #[test] + fn stale_bearer_exercise_refuses_after_a_generation_bump() { + let _g = guard(); + let bearer = register_owner_bearer(); + begin_egress_drain().unwrap(); + resume_egress_live(); + assert!( + bearer.admit_exercise().is_err(), + "a bearer minted under A must refuse to attach after B wins" + ); + } + + // Stale-capability ZERO-BYTES control (Paul's condition a): exercise + // validation fails BEFORE any transmission, for both durable kinds, under + // both a generation mismatch and the drain/latch state — so no site can + // send a byte on stale authority. + #[test] + fn durable_exercise_refuses_while_draining_and_while_latched() { + let _g = guard(); + let session = register_owner_session(CancellationToken::new()); + let bearer = register_owner_bearer(); + begin_egress_drain().unwrap(); + assert!( + session.admit_exercise().is_err(), + "draining refuses session" + ); + assert!(bearer.admit_exercise().is_err(), "draining refuses bearer"); + latch_identity_indeterminate(); + assert!(session.admit_exercise().is_err(), "latch refuses session"); + assert!(bearer.admit_exercise().is_err(), "latch refuses bearer"); + } + + // The C5 barrier cancels old-generation sessions and drops old bearers, + // and does NOT touch capabilities issued at the winning generation. + #[test] + fn barrier_revokes_old_generation_capabilities_only() { + let _g = guard(); + let old_session_cancel = CancellationToken::new(); + let old_session = register_owner_session(old_session_cancel.clone()); + let _old_bearer = register_owner_bearer(); + assert_eq!(live_durable_capability_count(), 2); + + // Transition to B. + let winning = begin_egress_drain().unwrap(); + resume_egress_live(); + // A capability issued at the winning generation survives the barrier. + let new_session = register_owner_session(CancellationToken::new()); + + let revoked = revoke_durable_capabilities_before(winning); + assert_eq!(revoked, 2, "both old-generation capabilities revoked"); + assert!( + old_session_cancel.is_cancelled(), + "the old session's registered token was invoked" + ); + assert_eq!( + live_durable_capability_count(), + 1, + "only the winning-generation capability remains" + ); + assert!( + new_session.admit_exercise().is_ok(), + "the winning-generation session still transmits" + ); + // old_session is now deregistered by the barrier; dropping it is a + // no-op remove. + drop(old_session); + } + + // Registration-completeness assertion (Paul's condition b): every issued + // durable capability is present in the registry with a revocation handle + // the C5 barrier can invoke, and dropping the handle deregisters it so a + // dead session/bearer is never a stale revocation target. + #[test] + fn issued_capabilities_are_registered_and_deregister_on_drop() { + let _g = guard(); + assert_eq!(live_durable_capability_count(), 0); + let session = register_owner_session(CancellationToken::new()); + assert_eq!(live_durable_capability_count(), 1); + let bearer = register_owner_bearer(); + assert_eq!(live_durable_capability_count(), 2); + drop(session); + assert_eq!(live_durable_capability_count(), 1, "session deregistered"); + drop(bearer); + assert_eq!(live_durable_capability_count(), 0, "bearer deregistered"); + } + + // A no-transition control: with no generation bump, a registered + // capability keeps transmitting and the barrier revokes nothing. + #[test] + fn no_transition_leaves_durable_capabilities_intact() { + let _g = guard(); + let session = register_owner_session(CancellationToken::new()); + let current = current_identity_persistence_generation(); + assert_eq!( + revoke_durable_capabilities_before(current), + 0, + "nothing older than the current generation to revoke" + ); + assert!( + session.admit_exercise().is_ok(), + "with no transition the session still transmits" + ); + } }