Compare commits

...
Author SHA1 Message Date
molluskandClaude Opus 4.8 1adf8a97bb S8 (Pass 2): wire grace-aware audio-membership admission
Closes S8 (High): inbound audio was authenticated by identity (remote_id) but
NOT by room membership, so a former member who knew a current member's endpoint
could reconnect on the audio ALPN and inject into / eavesdrop on the mix while
invisible in the roster. Now audio is admitted only for live gossip-roster
members (the senior+user-resolved GRACE-AWARE policy).

Transport (src/network/iroh_impl.rs):
- New per-session admitted_audio: HashSet<EndpointId> on Shared (internal state,
  no wire/serialization change). Cleared on disconnect_all.
- AudioRouter::accept consults audio_sender_admitted BEFORE ensure_supervisor —
  a non-member never gets a supervisor, sender handle, datagram reader, or
  outbound mix. Brief StdMutex check, released before the await (no RT lock).
- Pure apply_audio_admission_event(roster, peer, event) with AudioAdmissionEvent
  {RosterPresent insert, TransientDropGrace no-op, Remove}. Grace deliberately
  cannot ADD membership — it only preserves an already-admitted peer — so an
  unknown peer can't sneak in via a grace event. +3 lifecycle tests (on top of
  Pass-1's 4 predicate tests).
- admit/keep_for_reconnect_grace/remove/query methods for core to drive.

Core (src/core/mod.rs) — authority is core's VERIFIED gossip-roster events, not
transport connect/disconnect:
- PeerJoined / PeerUpdated: admit_audio_sender before connect_peer.
- PeerConnectionLost: keep_audio_sender_for_reconnect_grace (preserve through the
  existing RECONNECT_GRACE window — no audio cut on transient blips).
- gossip PeerLeft, transport ConnEvent::Left, grace-timer expiry: remove_audio_sender
  before disconnect_peer + jitter removal (removal-before-teardown bounds the
  in-flight-datagram race).
- datagram receiver: audio_sender_admitted gate before any jitter buffer (defense
  in depth against a datagram racing a removal). Mixer stays off the hot path.

Mid-join: a peer who dials audio before we've verified their signed Announce is
dropped (no "pending" admission, which would reintroduce the eavesdrop); their
reconnect loop recovers once the Announce admits them.

tests/transport_loopback.rs: admit both ends before connecting, mirroring the
production room-event order.

313 lib / clippy --all-targets / transport_loopback 4 / reconnect_eviction 6 /
release — all re-run green by the senior. Former-member-rejection + mid-join
recovery are verifiable only in a 2-machine call (senior's to run).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-18 04:41:37 -04:00
molluskandClaude Opus 4.8 10707152a3 S8 (design-first): pure audio_sender_admitted membership seam (unwired)
Design-first checkpoint for S8 (authorize inbound audio against live room
membership). Codex's design note (in the handoff task-report.md) establishes
the authoritative roster = gossip IrohGossipState.peers, NOT the audio
transport connection list, and recommends mirroring it into an audio-admission
snapshot consulted at AudioRouter::accept + datagram ingest.

This commit lands ONLY the pure decision seam + tests; wiring is deliberately
paused for a senior decision on the reconnect-grace policy (gossip drops a peer
from the roster on transient NeighborDown, but core keeps the audio supervisor
alive for RECONNECT_GRACE — a strict roster-only gate would cut audio on blips).

- audio_sender_admitted(remote, roster) -> bool (pub(crate), #[allow(dead_code)]).
- 4 tests: member admitted, stranger rejected, former member rejected after
  roster removal, mid-join peer rejected until authenticated Announce inserts it.
- No behavior change: accept/datagram/mixer paths untouched. S8 remains OPEN.

310 lib tests / clippy --all-targets / release all green (re-run by senior).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-18 04:24:01 -04:00
mollusk f2e72624f7 Merge codex-security-s11-presence-discovery: honest presence/discovery state on apply failure (S11) 2026-06-18 03:32:42 -04:00
molluskandClaude Opus 4.8 319d0c5e29 S11: make presence/discovery state honest on apply failure
SetPresenceMode and the Discoverable time-box auto-revert both committed
the new presence_mode to local state *before* apply_discovery and only
log_msg'd on failure, so a failed off-transition could leave the n0 DNS
PkarrPublisher running while the UI showed not-discoverable (privacy /
reality mismatch — security-open-handoff S11, from the W7 P7 review).

Fix (Codex, senior-reviewed):
- discovery.rs: pure resolve_presence_transition(prev, requested, apply_ok)
  -> (mode, Option<error>) seam — on failure keep the previous (truthful)
  mode and surface a message. +4 unit tests.
- apply_discovery now builds the replacement resolver/publisher services
  BEFORE clearing the service set, so a builder failure leaves the old
  posture fully intact (no partial state) — "keep previous mode" is then
  provably truthful.
- Both SetPresenceMode and the time-box revert apply discovery first, route
  through the seam, commit only the truthful mode, and surface failures via
  the existing PresenceModeReverted (corrects the picker) + UiEvent::Error.
  No new wire/event variant.
- A failed off-transition stays Discoverable and arms a 60s retry
  (DISCOVERY_REVERT_RETRY) so the beacon never stands stuck.
- P3 notes documented: relay-resolve exposes n0 query metadata (by design);
  no explicit iroh unpublish API exists, so the bounded ~30s pkarr TTL
  linger is documented, not behavior-changed; DirectOnly stays no-n0.

306 lib tests / clippy --all-targets / release all green (re-run by senior).
Runtime publish-stop behavior still wants a 2-machine / packet-capture check.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-18 03:32:38 -04:00
mollusk d56c2c90b2 Merge codex-security-hardening: S10 log redaction + T1/T2/T5/T6/T7 trust-boundary fixes 2026-06-18 03:05:56 -04:00
6 changed files with 438 additions and 61 deletions
+7 -5
View File
@@ -955,13 +955,15 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
state.friend_presence.insert(id, presence); state.friend_presence.insert(id, presence);
} }
UiEvent::PresenceModeReverted { mode } => { UiEvent::PresenceModeReverted { mode } => {
// The Discoverable time-box elapsed; core dropped us back to // Core corrected the committed presence mode. Mirror + persist so
// `mode` (Normal) and stopped publishing. Mirror + persist so the // the picker reflects the discovery state the endpoint actually has.
// presence picker reflects it, and tell the user why it changed.
state.config.presence_mode = mode; state.config.presence_mode = mode;
state.config.save(); state.config.save();
state.status_message = state.status_message = if mode == PresenceMode::Normal {
"Discoverable timed out — back to Normal".to_string(); "Discoverable timed out — back to Normal".to_string()
} else {
format!("Presence mode stayed {mode}")
};
} }
UiEvent::ShutdownComplete => { UiEvent::ShutdownComplete => {
if state.closing { if state.closing {
+4 -5
View File
@@ -124,11 +124,10 @@ pub enum UiEvent {
/// joinable gathering (with a one-click ticket). Emitted by the outbound ping /// joinable gathering (with a one-click ticket). Emitted by the outbound ping
/// scheduler; absence of a recent event = treat as offline. /// scheduler; absence of a recent event = treat as offline.
FriendPresence { id: EndpointId, presence: FriendPresence }, FriendPresence { id: EndpointId, presence: FriendPresence },
/// The Discoverable time-box elapsed (W7 P6): the core auto-reverted our presence /// Core corrected the committed presence posture. Usually the Discoverable
/// posture to the carried `mode` (always `Normal`) and stopped publishing. The /// time-box elapsed and the core auto-reverted to `Normal`; on discovery apply
/// GUI must mirror + persist this so its presence picker stops showing /// failure, this carries the previous truthful mode. The GUI must mirror +
/// Discoverable. Distinct from a user-driven change so the GUI knows to update /// persist this so its presence picker matches the endpoint's discovery state.
/// without having issued the command itself.
PresenceModeReverted { mode: PresenceMode }, PresenceModeReverted { mode: PresenceMode },
/// Core finished orderly app shutdown and the GUI can exit. /// Core finished orderly app shutdown and the GUI can exit.
ShutdownComplete, ShutdownComplete,
+161 -37
View File
@@ -13,6 +13,7 @@ use crate::network::{
use crate::core::messages::{CoreCommand, UiEvent}; use crate::core::messages::{CoreCommand, UiEvent};
use crate::config::{NetworkMode, RecordingMode}; use crate::config::{NetworkMode, RecordingMode};
use crate::presence::PresenceMode;
use crate::audio::multitrack::MultitrackRecorder; use crate::audio::multitrack::MultitrackRecorder;
use iroh::{Endpoint, EndpointAddr, EndpointId, RelayMode, SecretKey, endpoint::presets, protocol::Router}; use iroh::{Endpoint, EndpointAddr, EndpointId, RelayMode, SecretKey, endpoint::presets, protocol::Router};
use iroh_gossip::net::Gossip; use iroh_gossip::net::Gossip;
@@ -69,10 +70,27 @@ const RECONNECT_GRACE: Duration = Duration::from_secs(45);
/// 48 kHz / 60 ms frame) while bounding malicious datagram copy/decode churn. /// 48 kHz / 60 ms frame) while bounding malicious datagram copy/decode churn.
const MAX_OPUS_PAYLOAD: usize = 4000; const MAX_OPUS_PAYLOAD: usize = 4000;
/// If the Discoverable time-box tries to revert but discovery service reconfiguration
/// fails, retry soon while keeping the UI in the still-possible publishing state.
const DISCOVERY_REVERT_RETRY: Duration = Duration::from_secs(60);
fn audio_datagram_len_ok(len: usize) -> bool { fn audio_datagram_len_ok(len: usize) -> bool {
(4..=4 + MAX_OPUS_PAYLOAD).contains(&len) (4..=4 + MAX_OPUS_PAYLOAD).contains(&len)
} }
fn arm_discovery_retry(
discovery_deadline: &mut Option<tokio::time::Instant>,
now: tokio::time::Instant,
) {
let retry_deadline = now + DISCOVERY_REVERT_RETRY;
if discovery_deadline
.map(|current| current > retry_deadline)
.unwrap_or(true)
{
*discovery_deadline = Some(retry_deadline);
}
}
/// Per-peer reconnect grace timers (see [`RECONNECT_GRACE`]). Shared between the /// Per-peer reconnect grace timers (see [`RECONNECT_GRACE`]). Shared between the
/// room-event task (which arms one on a transient drop and cancels it on a /// room-event task (which arms one on a transient drop and cancels it on a
/// gossip rejoin) and the conn-event task (which cancels it when the audio link /// gossip rejoin) and the conn-event task (which cancels it when the audio link
@@ -119,6 +137,7 @@ fn arm_grace_timer(
let handle = tokio::spawn(async move { let handle = tokio::spawn(async move {
tokio::time::sleep(grace).await; tokio::time::sleep(grace).await;
crate::log_msg(&format!("Reconnect grace expired; evicting peer {:?}", peer_id)); 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; transport_evict.disconnect_peer(peer_id).await;
jitter_evict.lock().await.remove(&peer_id); jitter_evict.lock().await.remove(&peer_id);
// Scrub our internal state *before* announcing the eviction, so anything // Scrub our internal state *before* announcing the eviction, so anything
@@ -352,6 +371,7 @@ impl ConnEventHandler {
// until the grace timer or the slow gossip Leave. // until the grace timer or the slow gossip Leave.
cancel_grace_timer(&self.grace_timers, &id); cancel_grace_timer(&self.grace_timers, &id);
self.seen_connected.lock().unwrap().remove(&id); self.seen_connected.lock().unwrap().remove(&id);
self.transport.remove_audio_sender(id);
self.transport.disconnect_peer(id).await; self.transport.disconnect_peer(id).await;
self.jitter.lock().await.remove(&id); self.jitter.lock().await.remove(&id);
let _ = self.ui_tx.send(UiEvent::PeerLeft { id }).await; let _ = self.ui_tx.send(UiEvent::PeerLeft { id }).await;
@@ -466,12 +486,11 @@ impl NetStack {
/// `DnsAddressLookup`, mirroring the `N0` preset) is added when `plan.resolver`; the /// `DnsAddressLookup`, mirroring the `N0` preset) is added when `plan.resolver`; the
/// n0 DNS *publisher* (`PkarrPublisher`) when `plan.publisher`. /// n0 DNS *publisher* (`PkarrPublisher`) when `plan.publisher`.
/// ///
/// Idempotent and reversible: it clears the whole service set and reinstalls exactly /// Idempotent and reversible: it builds the replacement services first, then clears
/// what the plan wants, so flipping `publisher` off simply drops the publisher (its /// the service set and reinstalls exactly what the plan wants. Flipping `publisher`
/// republish task ends when the last clone is dropped, and the already-published /// off drops the publisher (its republish task ends when the last clone is dropped,
/// record TTL-expires within ~30s) without an endpoint rebuild and without disturbing /// and the already-published record TTL-expires within ~30s) without an endpoint
/// resolution. The brief clear→re-add window is a few synchronous calls; presence /// rebuild and without disturbing resolution.
/// toggles are rare, so a concurrent dial racing it is not a practical concern.
fn apply_discovery( fn apply_discovery(
endpoint: &Endpoint, endpoint: &Endpoint,
memory_lookup: &iroh::address_lookup::memory::MemoryLookup, memory_lookup: &iroh::address_lookup::memory::MemoryLookup,
@@ -482,16 +501,34 @@ fn apply_discovery(
pkarr::{PkarrPublisher, PkarrResolver}, pkarr::{PkarrPublisher, PkarrResolver},
}; };
let services = endpoint.address_lookup()?; let services = endpoint.address_lookup()?;
let pkarr_resolver = if plan.resolver {
Some(PkarrResolver::n0_dns().into_address_lookup(endpoint)?)
} else {
None
};
let dns_resolver = if plan.resolver {
Some(DnsAddressLookup::n0_dns().into_address_lookup(endpoint)?)
} else {
None
};
let publisher = if plan.publisher {
Some(PkarrPublisher::n0_dns().into_address_lookup(endpoint)?)
} else {
None
};
services.clear(); services.clear();
// Always keep the local, server-free lookup (this is what ticket/gossip dialing // Always keep the local, server-free lookup (this is what ticket/gossip dialing
// depends on — it must survive every posture, including DirectOnly). // depends on — it must survive every posture, including DirectOnly).
services.add(memory_lookup.clone()); services.add(memory_lookup.clone());
if plan.resolver { if let Some(pkarr_resolver) = pkarr_resolver {
services.add(PkarrResolver::n0_dns().into_address_lookup(endpoint)?); services.add(pkarr_resolver);
services.add(DnsAddressLookup::n0_dns().into_address_lookup(endpoint)?);
} }
if plan.publisher { if let Some(dns_resolver) = dns_resolver {
services.add(PkarrPublisher::n0_dns().into_address_lookup(endpoint)?); services.add(dns_resolver);
}
if let Some(publisher) = publisher {
services.add(publisher);
} }
Ok(()) Ok(())
} }
@@ -851,22 +888,64 @@ async fn run_core_loop(
// W7 P6 time-box: Discoverable auto-reverts to Normal after DISCOVERY_TIMEBOX // W7 P6 time-box: Discoverable auto-reverts to Normal after DISCOVERY_TIMEBOX
// so a publish beacon never stands indefinitely. The branch is disabled // so a publish beacon never stands indefinitely. The branch is disabled
// (`if` guard) unless a deadline is armed; `unwrap_or_else` is unreachable // (`if` guard) unless a deadline is armed; `unwrap_or_else` is unreachable
// belt-and-braces. On fire: stop publishing, drop to Normal, tell the GUI. // belt-and-braces. On fire: stop publishing first, then commit Normal only
// if the endpoint's discovery services accepted the non-publishing plan.
_ = tokio::time::sleep_until( _ = tokio::time::sleep_until(
discovery_deadline.unwrap_or_else(tokio::time::Instant::now), discovery_deadline.unwrap_or_else(tokio::time::Instant::now),
), if discovery_deadline.is_some() => { ), if discovery_deadline.is_some() => {
discovery_deadline = None; let previous_mode = *presence_mode.lock().unwrap();
*presence_mode.lock().unwrap() = crate::presence::PresenceMode::Normal; if previous_mode != PresenceMode::Discoverable {
let plan = crate::discovery::lookup_plan(network_mode, false); discovery_deadline = None;
if let Err(e) = apply_discovery(&net.endpoint, &net.memory_lookup, plan) { continue;
crate::log_msg(&format!("discovery: time-box revert failed: {e:#}")); }
let requested_mode = PresenceMode::Normal;
let now = tokio::time::Instant::now();
let plan = crate::discovery::lookup_plan(
network_mode,
requested_mode.publishes_to_discovery(),
);
let apply_result = apply_discovery(&net.endpoint, &net.memory_lookup, plan);
let (committed_mode, transition_error) =
crate::discovery::resolve_presence_transition(
previous_mode,
requested_mode,
apply_result.is_ok(),
);
*presence_mode.lock().unwrap() = committed_mode;
discovery_deadline = if committed_mode == PresenceMode::Discoverable {
Some(now + DISCOVERY_REVERT_RETRY)
} else {
None
};
match apply_result {
Ok(()) => {
crate::log_msg(
"discovery: Discoverable time-box elapsed → reverting to Normal",
);
let _ = ui_tx
.send(UiEvent::PresenceModeReverted {
mode: PresenceMode::Normal,
})
.await;
}
Err(e) => {
crate::log_msg(&format!("discovery: time-box revert failed: {e:#}"));
if committed_mode != requested_mode {
let _ = ui_tx
.send(UiEvent::PresenceModeReverted {
mode: committed_mode,
})
.await;
}
if let Some(message) = transition_error {
let _ = ui_tx
.send(UiEvent::Error(format!("{message} ({e:#})")))
.await;
}
}
} }
crate::log_msg("discovery: Discoverable time-box elapsed → reverting to Normal");
let _ = ui_tx
.send(UiEvent::PresenceModeReverted {
mode: crate::presence::PresenceMode::Normal,
})
.await;
continue; continue;
} }
}; };
@@ -1162,6 +1241,9 @@ async fn run_core_loop(
}; };
while let Some((from_peer, bytes)) = datagram_rx.recv().await { 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()) { if !audio_datagram_len_ok(bytes.len()) {
// Malformed (< sequence header) or oversized Opus payload. // Malformed (< sequence header) or oversized Opus payload.
continue; continue;
@@ -1396,6 +1478,7 @@ async fn run_core_loop(
// A (re)join means the peer is back — cancel any // A (re)join means the peer is back — cancel any
// pending reconnect grace timer before re-adding it. // pending reconnect grace timer before re-adding it.
cancel_grace_timer(&grace_timers_events, &peer_id); cancel_grace_timer(&grace_timers_events, &peer_id);
transport_events.admit_audio_sender(peer_id);
// Establish the audio connection as soon as the peer // Establish the audio connection as soon as the peer
// is known (the transport dedupes the full-mesh race). // is known (the transport dedupes the full-mesh race).
// Hand over the full address so reconnects can dial // Hand over the full address so reconnects can dial
@@ -1447,6 +1530,7 @@ async fn run_core_loop(
{ {
peers.remove(&peer_id); peers.remove(&peer_id);
} }
transport_events.remove_audio_sender(peer_id);
transport_events.disconnect_peer(peer_id).await; transport_events.disconnect_peer(peer_id).await;
jitter_events.lock().await.remove(&peer_id); jitter_events.lock().await.remove(&peer_id);
let _ = ui_tx_events.send(UiEvent::PeerLeft { id: peer_id }).await; let _ = ui_tx_events.send(UiEvent::PeerLeft { id: peer_id }).await;
@@ -1459,6 +1543,7 @@ async fn run_core_loop(
// it. Idempotent: an ordinary mute/unmute update just // it. Idempotent: an ordinary mute/unmute update just
// re-records the same address. // re-records the same address.
cancel_grace_timer(&grace_timers_events, &peer_id); cancel_grace_timer(&grace_timers_events, &peer_id);
transport_events.admit_audio_sender(peer_id);
transport_events.connect_peer(state.addr.clone()).await; transport_events.connect_peer(state.addr.clone()).await;
// Auto-heal a friend's saved address (W7) on the // Auto-heal a friend's saved address (W7) on the
// re-announce too — this is the path that catches a // re-announce too — this is the path that catches a
@@ -1500,6 +1585,7 @@ async fn run_core_loop(
// hasn't recovered within RECONNECT_GRACE. A gossip // hasn't recovered within RECONNECT_GRACE. A gossip
// rejoin (PeerJoined/PeerUpdated) or a transport // rejoin (PeerJoined/PeerUpdated) or a transport
// reconnect (ConnEvent::Connected) cancels it first. // 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; let _ = ui_tx_events.send(UiEvent::PeerConnecting { id: peer_id }).await;
arm_grace_timer( arm_grace_timer(
&grace_timers_events, &grace_timers_events,
@@ -1798,22 +1884,60 @@ async fn run_core_loop(
} }
CoreCommand::SetPresenceMode(mode) => { CoreCommand::SetPresenceMode(mode) => {
*presence_mode.lock().unwrap() = mode; let previous_mode = *presence_mode.lock().unwrap();
// W7 P6: re-apply n0 DNS discovery for the new posture (publish on iff let now = tokio::time::Instant::now();
// Discoverable). Runtime — no endpoint rebuild; clears + reinstalls the
// address-lookup services. The resolver stays on regardless so we can if previous_mode == mode {
// still look up moved friends. // Same-mode requests are no-ops for discovery wiring, but keep the
let plan = crate::discovery::lookup_plan(network_mode, mode.publishes_to_discovery()); // existing UX: re-selecting Discoverable restarts the clock.
if let Err(e) = apply_discovery(&net.endpoint, &net.memory_lookup, plan) { discovery_deadline = if mode == PresenceMode::Discoverable {
crate::log_msg(&format!("discovery: apply failed: {e:#}")); Some(now + crate::discovery::DISCOVERY_TIMEBOX)
} else {
None
};
continue;
} }
// Arm (Discoverable) or cancel (any other posture) the auto-revert
// time-box. Re-selecting Discoverable restarts the clock. // W7 P6/S11: re-apply n0 DNS discovery for the requested posture
discovery_deadline = if mode == crate::presence::PresenceMode::Discoverable { // first, then commit the presence mode only if the endpoint accepted
Some(tokio::time::Instant::now() + crate::discovery::DISCOVERY_TIMEBOX) // that discovery plan. This keeps the UI truthful when dropping the
// publisher fails.
let plan =
crate::discovery::lookup_plan(network_mode, mode.publishes_to_discovery());
let apply_result = apply_discovery(&net.endpoint, &net.memory_lookup, plan);
let (committed_mode, transition_error) =
crate::discovery::resolve_presence_transition(
previous_mode,
mode,
apply_result.is_ok(),
);
*presence_mode.lock().unwrap() = committed_mode;
if committed_mode == PresenceMode::Discoverable {
if apply_result.is_ok() && mode == PresenceMode::Discoverable {
discovery_deadline = Some(now + crate::discovery::DISCOVERY_TIMEBOX);
} else {
arm_discovery_retry(&mut discovery_deadline, now);
}
} else { } else {
None discovery_deadline = None;
}; }
if let Err(e) = apply_result {
crate::log_msg(&format!("discovery: apply failed: {e:#}"));
if committed_mode != mode {
let _ = ui_tx
.send(UiEvent::PresenceModeReverted {
mode: committed_mode,
})
.await;
}
if let Some(message) = transition_error {
let _ = ui_tx
.send(UiEvent::Error(format!("{message} ({e:#})")))
.await;
}
}
} }
CoreCommand::SetRecordingMode(mode) => { CoreCommand::SetRecordingMode(mode) => {
+96 -10
View File
@@ -8,14 +8,19 @@
//! The model (from `docs/contacts-plan.md` P6, decided 2026-06-16): //! The model (from `docs/contacts-plan.md` P6, decided 2026-06-16):
//! - **Resolving is always allowed on relay-capable modes** — a stationary friend //! - **Resolving is always allowed on relay-capable modes** — a stationary friend
//! (typically in `Normal`) must be able to look up a friend who moved networks. A //! (typically in `Normal`) must be able to look up a friend who moved networks. A
//! resolve is a DNS query to n0 that publishes nothing; it only fires when a saved //! resolve is a DNS query to n0 that publishes nothing, but still exposes query
//! address is stale and the dial falls through to discovery. //! timing/source metadata to n0; it only fires when a saved address is stale and
//! the dial falls through to discovery.
//! - **Publishing is gated on `Discoverable`** and asymmetric: only the mover //! - **Publishing is gated on `Discoverable`** and asymmetric: only the mover
//! publishes their address to n0 DNS; everyone else just looks it up. //! publishes their address to n0 DNS; everyone else just looks it up.
//! - **Stopping publishing removes the local publisher service**; iroh does not
//! expose an explicit unpublish call here, so already-published pkarr records can
//! linger until their default ~30s TTL expires.
//! - **`DirectOnly` is the explicit no-server posture** — neither resolve nor publish //! - **`DirectOnly` is the explicit no-server posture** — neither resolve nor publish
//! ever touches n0 there, regardless of the Discoverable toggle. //! ever touches n0 there, regardless of the Discoverable toggle.
use crate::config::NetworkMode; use crate::config::NetworkMode;
use crate::presence::PresenceMode;
use std::time::Duration; use std::time::Duration;
/// How long `Discoverable` stays on before auto-reverting to `Normal`. Discovery is /// How long `Discoverable` stays on before auto-reverting to `Normal`. Discovery is
@@ -46,12 +51,43 @@ pub fn lookup_plan(network_mode: NetworkMode, want_publish: bool) -> LookupPlan
match network_mode { match network_mode {
// The explicit serverless posture: no n0 contact at all, even to resolve. // The explicit serverless posture: no n0 contact at all, even to resolve.
// A Discoverable toggle here is intentionally inert. // A Discoverable toggle here is intentionally inert.
NetworkMode::DirectOnly => LookupPlan { resolver: false, publisher: false }, NetworkMode::DirectOnly => LookupPlan {
resolver: false,
publisher: false,
},
// Relay-capable: always resolve (so a stationary friend can find a mover); // Relay-capable: always resolve (so a stationary friend can find a mover);
// publish only when the user opted into Discoverable. // publish only when the user opted into Discoverable.
NetworkMode::RelayNoDiscovery | NetworkMode::N0Full => { NetworkMode::RelayNoDiscovery | NetworkMode::N0Full => LookupPlan {
LookupPlan { resolver: true, publisher: want_publish } resolver: true,
} publisher: want_publish,
},
}
}
/// Decide which presence mode may be committed after attempting to apply discovery
/// services for `requested`.
///
/// On failure, keep the previous mode: it is the only locally truthful state because
/// the endpoint's discovery services may still reflect the old posture. Same-mode
/// requests are no-ops from a presence-truth perspective and do not surface an error.
pub fn resolve_presence_transition(
previous: PresenceMode,
requested: PresenceMode,
apply_ok: bool,
) -> (PresenceMode, Option<String>) {
if previous == requested {
return (previous, None);
}
if apply_ok {
(requested, None)
} else {
(
previous,
Some(format!(
"Couldn't update discovery mode; keeping {previous}."
)),
)
} }
} }
@@ -64,12 +100,18 @@ mod tests {
for mode in [NetworkMode::RelayNoDiscovery, NetworkMode::N0Full] { for mode in [NetworkMode::RelayNoDiscovery, NetworkMode::N0Full] {
assert_eq!( assert_eq!(
lookup_plan(mode, false), lookup_plan(mode, false),
LookupPlan { resolver: true, publisher: false }, LookupPlan {
resolver: true,
publisher: false
},
"{mode:?}: resolve always on, no publish when not Discoverable" "{mode:?}: resolve always on, no publish when not Discoverable"
); );
assert_eq!( assert_eq!(
lookup_plan(mode, true), lookup_plan(mode, true),
LookupPlan { resolver: true, publisher: true }, LookupPlan {
resolver: true,
publisher: true
},
"{mode:?}: Discoverable adds publish on top of resolve" "{mode:?}: Discoverable adds publish on top of resolve"
); );
} }
@@ -79,12 +121,18 @@ mod tests {
fn direct_only_never_touches_n0_even_when_discoverable() { fn direct_only_never_touches_n0_even_when_discoverable() {
assert_eq!( assert_eq!(
lookup_plan(NetworkMode::DirectOnly, false), lookup_plan(NetworkMode::DirectOnly, false),
LookupPlan { resolver: false, publisher: false } LookupPlan {
resolver: false,
publisher: false
}
); );
// The serverless posture overrides the Discoverable request entirely. // The serverless posture overrides the Discoverable request entirely.
assert_eq!( assert_eq!(
lookup_plan(NetworkMode::DirectOnly, true), lookup_plan(NetworkMode::DirectOnly, true),
LookupPlan { resolver: false, publisher: false } LookupPlan {
resolver: false,
publisher: false
}
); );
} }
@@ -92,4 +140,42 @@ mod tests {
fn timebox_is_thirty_minutes() { fn timebox_is_thirty_minutes() {
assert_eq!(DISCOVERY_TIMEBOX, Duration::from_secs(1800)); assert_eq!(DISCOVERY_TIMEBOX, Duration::from_secs(1800));
} }
#[test]
fn presence_transition_commits_requested_mode_after_successful_apply() {
assert_eq!(
resolve_presence_transition(PresenceMode::Normal, PresenceMode::Discoverable, true),
(PresenceMode::Discoverable, None)
);
}
#[test]
fn presence_transition_keeps_previous_mode_when_apply_fails() {
let (mode, err) =
resolve_presence_transition(PresenceMode::Normal, PresenceMode::Discoverable, false);
assert_eq!(mode, PresenceMode::Normal);
assert!(err.unwrap().contains("keeping Normal"));
}
#[test]
fn presence_transition_keeps_discoverable_when_off_transition_fails() {
let (mode, err) =
resolve_presence_transition(PresenceMode::Discoverable, PresenceMode::Normal, false);
assert_eq!(mode, PresenceMode::Discoverable);
assert!(err.unwrap().contains("keeping Discoverable"));
}
#[test]
fn presence_transition_same_mode_is_noop_without_error() {
assert_eq!(
resolve_presence_transition(
PresenceMode::Discoverable,
PresenceMode::Discoverable,
false
),
(PresenceMode::Discoverable, None)
);
}
} }
+167 -4
View File
@@ -5,7 +5,7 @@ use bytes::Bytes;
use tokio::sync::mpsc; use tokio::sync::mpsc;
use tokio::sync::mpsc::Receiver; use tokio::sync::mpsc::Receiver;
use std::sync::{Arc, Mutex as StdMutex}; use std::sync::{Arc, Mutex as StdMutex};
use std::collections::HashMap; use std::collections::{HashMap, HashSet};
use std::time::Duration; use std::time::Duration;
use async_trait::async_trait; use async_trait::async_trait;
@@ -56,6 +56,10 @@ struct Shared {
/// supervisor inserts its connection when the link comes up and removes it /// supervisor inserts its connection when the link comes up and removes it
/// when the link dies. /// when the link dies.
live_conns: StdMutex<HashMap<EndpointId, Connection>>, 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)>, incoming_tx: mpsc::Sender<(EndpointId, Bytes)>,
/// Best-effort link-state notifications for the UI (connecting / connected). /// Best-effort link-state notifications for the UI (connecting / connected).
conn_events_tx: mpsc::Sender<ConnEvent>, conn_events_tx: mpsc::Sender<ConnEvent>,
@@ -112,6 +116,16 @@ impl Shared {
crate::log_msg(&format!("Transport: stopped supervising peer {:?}", peer_id)); 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. /// 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)) 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 /// 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 — /// 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 /// 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() { if shared.self_id.to_string() < peer_id.to_string() {
return Ok(()); 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 // Route the connection to this peer's supervisor (creating it if the
// inbound link beat the gossip join event). try_send keeps the // inbound link arrives after the signed gossip Announce admitted it).
// protocol handler from ever blocking; a full queue only happens if // try_send keeps the protocol handler from ever blocking; a full queue
// links are churning, and the supervisor will get the next one. // only happens if links are churning, and the supervisor gets the next one.
let inbound_tx = shared.ensure_supervisor(peer_id).await; let inbound_tx = shared.ensure_supervisor(peer_id).await;
if inbound_tx.try_send(connection).is_err() { if inbound_tx.try_send(connection).is_err() {
crate::log_msg(&format!("Transport: dropped inbound link from {:?} (queue full)", peer_id)); crate::log_msg(&format!("Transport: dropped inbound link from {:?} (queue full)", peer_id));
@@ -369,6 +425,7 @@ impl IrohTransport {
addrs: StdMutex::new(HashMap::new()), addrs: StdMutex::new(HashMap::new()),
peers: tokio::sync::Mutex::new(HashMap::new()), peers: tokio::sync::Mutex::new(HashMap::new()),
live_conns: StdMutex::new(HashMap::new()), live_conns: StdMutex::new(HashMap::new()),
admitted_audio: StdMutex::new(HashSet::new()),
incoming_tx, incoming_tx,
conn_events_tx, conn_events_tx,
}); });
@@ -397,11 +454,35 @@ impl IrohTransport {
} }
self.shared.senders.lock().unwrap().clear(); self.shared.senders.lock().unwrap().clear();
self.shared.addrs.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 // Give the CONNECTION_CLOSE frames a moment to flush before the caller
// shuts the endpoint/router down (the `conns` clones are still alive // shuts the endpoint/router down (the `conns` clones are still alive
// here, so the endpoint can still transmit them). // here, so the endpoint can still transmit them).
tokio::time::sleep(Duration::from_millis(150)).await; 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] #[async_trait]
@@ -443,3 +524,85 @@ impl NetworkTransport for IrohTransport {
.ok_or_else(|| NetError::Other("Connection events already subscribed".to_string())) .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));
}
}
+3
View File
@@ -169,6 +169,9 @@ async fn loopback_sequenced_audio_reaches_peer_and_decodes() {
b.lookup.add_endpoint_info(a.endpoint.addr()); b.lookup.add_endpoint_info(a.endpoint.addr());
let a_id = a.endpoint.id(); 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. // Subscribe to incoming datagrams on B before any are sent.
let mut b_rx = b.transport.receive_datagrams().await.expect("subscribe B"); let mut b_rx = b.transport.receive_datagrams().await.expect("subscribe B");