Merge codex-security-s8-audio-membership: gate inbound audio by live room membership (S8)

src/core/mod.rs

@@ -137,6 +137,7 @@ fn arm_grace_timer(
     let handle = tokio::spawn(async move {
         tokio::time::sleep(grace).await;
         crate::log_msg(&format!("Reconnect grace expired; evicting peer {:?}", peer_id));
+        transport_evict.remove_audio_sender(peer_id);
         transport_evict.disconnect_peer(peer_id).await;
         jitter_evict.lock().await.remove(&peer_id);
         // Scrub our internal state *before* announcing the eviction, so anything
@@ -370,6 +371,7 @@ impl ConnEventHandler {
                 // until the grace timer or the slow gossip Leave.
                 cancel_grace_timer(&self.grace_timers, &id);
                 self.seen_connected.lock().unwrap().remove(&id);
+                self.transport.remove_audio_sender(id);
                 self.transport.disconnect_peer(id).await;
                 self.jitter.lock().await.remove(&id);
                 let _ = self.ui_tx.send(UiEvent::PeerLeft { id }).await;
@@ -1239,6 +1241,9 @@ async fn run_core_loop(
                     };
 
                     while let Some((from_peer, bytes)) = datagram_rx.recv().await {
+                        if !transport_recv.audio_sender_admitted(from_peer) {
+                            continue;
+                        }
                         if !audio_datagram_len_ok(bytes.len()) {
                             // Malformed (< sequence header) or oversized Opus payload.
                             continue;
@@ -1473,6 +1478,7 @@ async fn run_core_loop(
                                 // A (re)join means the peer is back — cancel any
                                 // pending reconnect grace timer before re-adding it.
                                 cancel_grace_timer(&grace_timers_events, &peer_id);
+                                transport_events.admit_audio_sender(peer_id);
                                 // Establish the audio connection as soon as the peer
                                 // is known (the transport dedupes the full-mesh race).
                                 // Hand over the full address so reconnects can dial
@@ -1524,6 +1530,7 @@ async fn run_core_loop(
                                 {
                                     peers.remove(&peer_id);
                                 }
+                                transport_events.remove_audio_sender(peer_id);
                                 transport_events.disconnect_peer(peer_id).await;
                                 jitter_events.lock().await.remove(&peer_id);
                                 let _ = ui_tx_events.send(UiEvent::PeerLeft { id: peer_id }).await;
@@ -1536,6 +1543,7 @@ async fn run_core_loop(
                                 // it. Idempotent: an ordinary mute/unmute update just
                                 // re-records the same address.
                                 cancel_grace_timer(&grace_timers_events, &peer_id);
+                                transport_events.admit_audio_sender(peer_id);
                                 transport_events.connect_peer(state.addr.clone()).await;
                                 // Auto-heal a friend's saved address (W7) on the
                                 // re-announce too — this is the path that catches a
@@ -1577,6 +1585,7 @@ async fn run_core_loop(
                                 // hasn't recovered within RECONNECT_GRACE. A gossip
                                 // rejoin (PeerJoined/PeerUpdated) or a transport
                                 // reconnect (ConnEvent::Connected) cancels it first.
+                                transport_events.keep_audio_sender_for_reconnect_grace(peer_id);
                                 let _ = ui_tx_events.send(UiEvent::PeerConnecting { id: peer_id }).await;
                                 arm_grace_timer(
                                     &grace_timers_events,

src/network/iroh_impl.rs

@@ -5,7 +5,7 @@ use bytes::Bytes;
 use tokio::sync::mpsc;
 use tokio::sync::mpsc::Receiver;
 use std::sync::{Arc, Mutex as StdMutex};
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::time::Duration;
 use async_trait::async_trait;
 
@@ -56,6 +56,10 @@ struct Shared {
     /// supervisor inserts its connection when the link comes up and removes it
     /// when the link dies.
     live_conns: StdMutex<HashMap<EndpointId, Connection>>,
+    /// Core-owned audio admission snapshot for this room session. It mirrors the
+    /// verified gossip roster plus peers still inside reconnect grace; transport
+    /// connections alone never mutate this set.
+    admitted_audio: StdMutex<HashSet<EndpointId>>,
     incoming_tx: mpsc::Sender<(EndpointId, Bytes)>,
     /// Best-effort link-state notifications for the UI (connecting / connected).
     conn_events_tx: mpsc::Sender<ConnEvent>,
@@ -112,6 +116,16 @@ impl Shared {
             crate::log_msg(&format!("Transport: stopped supervising peer {:?}", peer_id));
         }
     }
+
+    fn audio_sender_admitted(&self, peer_id: EndpointId) -> bool {
+        let roster = self.admitted_audio.lock().unwrap();
+        audio_sender_admitted(peer_id, &roster)
+    }
+
+    fn apply_audio_admission(&self, peer_id: EndpointId, event: AudioAdmissionEvent) {
+        let mut roster = self.admitted_audio.lock().unwrap();
+        apply_audio_admission_event(&mut roster, peer_id, event);
+    }
 }
 
 /// Why a peer's live-link wait woke up.
@@ -135,6 +149,41 @@ fn is_graceful_leave(err: &ConnectionError) -> bool {
     matches!(err, ConnectionError::ApplicationClosed(frame) if frame.error_code == VarInt::from_u32(GOODBYE_CODE))
 }
 
+/// Pure S8 membership decision: iroh already authenticated `remote` as the
+/// connection's endpoint id, so audio admission is exactly live roster membership.
+pub(crate) fn audio_sender_admitted(remote: EndpointId, roster: &HashSet<EndpointId>) -> bool {
+    roster.contains(&remote)
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum AudioAdmissionEvent {
+    /// A signed gossip Announce/Update says the peer is in the live room roster.
+    RosterPresent,
+    /// Gossip reported a transient drop; keep admission during reconnect grace.
+    TransientDropGrace,
+    /// Graceful leave, transport Left eviction, or reconnect-grace expiry.
+    Remove,
+}
+
+pub(crate) fn apply_audio_admission_event(
+    roster: &mut HashSet<EndpointId>,
+    peer_id: EndpointId,
+    event: AudioAdmissionEvent,
+) {
+    match event {
+        AudioAdmissionEvent::RosterPresent => {
+            roster.insert(peer_id);
+        }
+        AudioAdmissionEvent::TransientDropGrace => {
+            // Grace is not an authority to add membership; it only preserves an
+            // already-admitted peer until either rejoin or grace expiry.
+        }
+        AudioAdmissionEvent::Remove => {
+            roster.remove(&peer_id);
+        }
+    }
+}
+
 /// Owns a single peer's connection lifecycle for as long as the peer is in the
 /// room: obtain a link, run the send/read loops, and on loss obtain a new one —
 /// with capped backoff on the dialing side. The deterministic-initiator rule
@@ -333,10 +382,17 @@ impl iroh::protocol::ProtocolHandler for AudioRouter {
             if shared.self_id.to_string() < peer_id.to_string() {
                 return Ok(());
             }
+            if !shared.audio_sender_admitted(peer_id) {
+                crate::log_msg(&format!(
+                    "Transport: rejected inbound audio from non-member {}",
+                    crate::short_id(&peer_id.to_string())
+                ));
+                return Ok(());
+            }
             // Route the connection to this peer's supervisor (creating it if the
-            // inbound link beat the gossip join event). try_send keeps the
-            // protocol handler from ever blocking; a full queue only happens if
-            // links are churning, and the supervisor will get the next one.
+            // inbound link arrives after the signed gossip Announce admitted it).
+            // try_send keeps the protocol handler from ever blocking; a full queue
+            // only happens if links are churning, and the supervisor gets the next one.
             let inbound_tx = shared.ensure_supervisor(peer_id).await;
             if inbound_tx.try_send(connection).is_err() {
                 crate::log_msg(&format!("Transport: dropped inbound link from {:?} (queue full)", peer_id));
@@ -369,6 +425,7 @@ impl IrohTransport {
             addrs: StdMutex::new(HashMap::new()),
             peers: tokio::sync::Mutex::new(HashMap::new()),
             live_conns: StdMutex::new(HashMap::new()),
+            admitted_audio: StdMutex::new(HashSet::new()),
             incoming_tx,
             conn_events_tx,
         });
@@ -397,11 +454,35 @@ impl IrohTransport {
         }
         self.shared.senders.lock().unwrap().clear();
         self.shared.addrs.lock().unwrap().clear();
+        self.shared.admitted_audio.lock().unwrap().clear();
         // Give the CONNECTION_CLOSE frames a moment to flush before the caller
         // shuts the endpoint/router down (the `conns` clones are still alive
         // here, so the endpoint can still transmit them).
         tokio::time::sleep(Duration::from_millis(150)).await;
     }
+
+    /// Admit a peer to this session's audio plane. Core calls this from verified
+    /// gossip roster events; the transport never derives membership on its own.
+    pub fn admit_audio_sender(&self, peer_id: EndpointId) {
+        self.shared
+            .apply_audio_admission(peer_id, AudioAdmissionEvent::RosterPresent);
+    }
+
+    /// Preserve an already-admitted peer through the reconnect grace window.
+    pub fn keep_audio_sender_for_reconnect_grace(&self, peer_id: EndpointId) {
+        self.shared
+            .apply_audio_admission(peer_id, AudioAdmissionEvent::TransientDropGrace);
+    }
+
+    /// Remove a peer from audio admission before tearing down transport/jitter state.
+    pub fn remove_audio_sender(&self, peer_id: EndpointId) {
+        self.shared
+            .apply_audio_admission(peer_id, AudioAdmissionEvent::Remove);
+    }
+
+    pub fn audio_sender_admitted(&self, peer_id: EndpointId) -> bool {
+        self.shared.audio_sender_admitted(peer_id)
+    }
 }
 
 #[async_trait]
@@ -443,3 +524,85 @@ impl NetworkTransport for IrohTransport {
             .ok_or_else(|| NetError::Other("Connection events already subscribed".to_string()))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use iroh::SecretKey;
+
+    fn endpoint_id() -> EndpointId {
+        SecretKey::generate().public()
+    }
+
+    #[test]
+    fn audio_sender_admission_accepts_roster_member() {
+        let member = endpoint_id();
+        let roster = HashSet::from([member]);
+
+        assert!(audio_sender_admitted(member, &roster));
+    }
+
+    #[test]
+    fn audio_sender_admission_rejects_unknown_sender() {
+        let member = endpoint_id();
+        let stranger = endpoint_id();
+        let roster = HashSet::from([member]);
+
+        assert!(!audio_sender_admitted(stranger, &roster));
+    }
+
+    #[test]
+    fn audio_sender_admission_rejects_former_member_after_roster_removal() {
+        let former = endpoint_id();
+        let mut roster = HashSet::from([former]);
+        assert!(audio_sender_admitted(former, &roster));
+
+        roster.remove(&former);
+
+        assert!(!audio_sender_admitted(former, &roster));
+    }
+
+    #[test]
+    fn audio_sender_admission_waits_for_mid_join_announce() {
+        let joining_peer = endpoint_id();
+        let mut roster = HashSet::new();
+
+        assert!(!audio_sender_admitted(joining_peer, &roster));
+
+        roster.insert(joining_peer);
+
+        assert!(audio_sender_admitted(joining_peer, &roster));
+    }
+
+    #[test]
+    fn audio_admission_lifecycle_keeps_peer_through_transient_grace() {
+        let peer = endpoint_id();
+        let mut roster = HashSet::new();
+
+        apply_audio_admission_event(&mut roster, peer, AudioAdmissionEvent::RosterPresent);
+        assert!(audio_sender_admitted(peer, &roster));
+
+        apply_audio_admission_event(&mut roster, peer, AudioAdmissionEvent::TransientDropGrace);
+        assert!(audio_sender_admitted(peer, &roster));
+    }
+
+    #[test]
+    fn audio_admission_lifecycle_does_not_add_unknown_peer_on_grace_event() {
+        let peer = endpoint_id();
+        let mut roster = HashSet::new();
+
+        apply_audio_admission_event(&mut roster, peer, AudioAdmissionEvent::TransientDropGrace);
+
+        assert!(!audio_sender_admitted(peer, &roster));
+    }
+
+    #[test]
+    fn audio_admission_lifecycle_removes_peer_on_leave_or_grace_expiry() {
+        let peer = endpoint_id();
+        let mut roster = HashSet::from([peer]);
+
+        apply_audio_admission_event(&mut roster, peer, AudioAdmissionEvent::Remove);
+
+        assert!(!audio_sender_admitted(peer, &roster));
+    }
+}

tests/transport_loopback.rs

@@ -169,6 +169,9 @@ async fn loopback_sequenced_audio_reaches_peer_and_decodes() {
     b.lookup.add_endpoint_info(a.endpoint.addr());
 
     let a_id = a.endpoint.id();
+    let b_id = b.endpoint.id();
+    a.transport.admit_audio_sender(b_id);
+    b.transport.admit_audio_sender(a_id);
 
     // Subscribe to incoming datagrams on B before any are sent.
     let mut b_rx = b.transport.receive_datagrams().await.expect("subscribe B");