Establishes VERSIONING.md: SemVer 0.x (MINOR = breaking wire change) for the release version, and per-plane protocol versions enforced on the wire so incompatible peers fail fast and legibly instead of via silent decode/signature errors. ⚠️ BREAKING WIRE CHANGE — all peers must run >= 0.2.0 to interoperate (ALPNs and gossip subscription topics changed). A pre-0.2.0 peer (e.g. an un-resynced dopedart) can no longer connect, by design, and now fails at the handshake. - New src/protocol.rs: single source of truth for AUDIO/FRIENDS/GOSSIP_PROTO, the derived ALPNs (peerspeak/audio/1, peerspeak/friends/1), GOSSIP_SIG_DOMAIN, and versioned_topic(). Unit tests assert ALPN/domain strings match their integer versions (no silent drift) + that topic namespacing is deterministic. - Unified ALPNs: audio was b"peerspeak-audio" (unversioned, and duplicated in iroh_impl.rs + core/mod.rs) -> peerspeak/audio/1 from protocol.rs; friends re-exports protocol::FRIENDS_ALPN (was peerspeak/friends/0 -> /1). - Gossip: subscribe to versioned_topic(ticket.topic_id) so different gossip versions never share a swarm; the raw topic_id stays the room identity and what signatures bind. GOSSIP_SIG_DOMAIN centralized into protocol.rs. - Cargo.toml 0.1.0 -> 0.2.0. 316 lib tests / clippy --all-targets clean. VERSIONING.md documents the bump rules, the "I changed X -> what do I bump" table, and a release checklist. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
896 lines
38 KiB
Rust
896 lines
38 KiB
Rust
use crate::network::{RoomState, NetError, PeerState, RoomEvent, PeerSpeakTicket};
|
|
use iroh::{Endpoint, EndpointAddr, EndpointId, SecretKey, Signature};
|
|
use iroh_gossip::net::Gossip;
|
|
use iroh_gossip::proto::TopicId;
|
|
use tokio::sync::mpsc;
|
|
use tokio::sync::mpsc::Receiver;
|
|
use std::sync::{Arc, Mutex};
|
|
use std::collections::HashMap;
|
|
use async_trait::async_trait;
|
|
use tokio_stream::StreamExt;
|
|
use serde::{Serialize, Deserialize};
|
|
|
|
/// Domain-separation tag mixed into every signed gossip payload so a signature
|
|
/// can never be lifted out of this protocol/version into another context.
|
|
use crate::protocol::GOSSIP_SIG_DOMAIN;
|
|
|
|
/// How far a payload's sender-stamped timestamp may differ from local time
|
|
/// before it's rejected as stale (replayed) or implausibly future. Bounds the
|
|
/// window in which a captured *genuine* signed payload could be replayed; within
|
|
/// it, a replay is byte-identical (the signature fixes the bytes) so the gossip
|
|
/// swarm's own in-flight de-duplication suppresses it. 2 minutes tolerates
|
|
/// reasonable cross-peer clock skew without leaving a wide replay window.
|
|
const GOSSIP_FRESHNESS_MS: u64 = 120_000;
|
|
|
|
/// A gossip message plus the authentication envelope that proves who sent it.
|
|
/// `author` is the claimed sender (an `EndpointId`, which *is* an ed25519 public
|
|
/// key); `sig` is that key's signature over [`signable_bytes`], so a forged
|
|
/// `author` can't validate (the attacker lacks the victim's secret key). `ts`
|
|
/// (sender-stamped unix-millis) is covered by the signature and gates replay
|
|
/// freshness — distinct from `GossipMessage::Chat.ts`, which is only for display.
|
|
#[derive(Serialize, Deserialize, Clone)]
|
|
pub struct GossipPayload {
|
|
pub author: EndpointId,
|
|
pub ts: u64,
|
|
pub msg: GossipMessage,
|
|
pub sig: Signature,
|
|
}
|
|
|
|
// `iroh::Signature` doesn't implement `Debug`, so derive everything else and
|
|
// hand-write a `Debug` that omits the (noisy, non-Debug) signature.
|
|
impl std::fmt::Debug for GossipPayload {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
f.debug_struct("GossipPayload")
|
|
.field("author", &self.author)
|
|
.field("ts", &self.ts)
|
|
.field("msg_kind", &gossip_message_kind(&self.msg))
|
|
.finish_non_exhaustive()
|
|
}
|
|
}
|
|
|
|
/// Current unix time in milliseconds (0 on the impossible pre-epoch error).
|
|
fn now_millis() -> u64 {
|
|
std::time::SystemTime::now()
|
|
.duration_since(std::time::UNIX_EPOCH)
|
|
.map(|d| d.as_millis() as u64)
|
|
.unwrap_or(0)
|
|
}
|
|
|
|
/// The exact bytes an ed25519 gossip signature covers: domain tag + room topic +
|
|
/// claimed author + timestamp + message. Binding the **topic** prevents replay
|
|
/// into another room; binding the **timestamp** bounds temporal replay; binding
|
|
/// the **author** means a signature only validates for the key it claims to come
|
|
/// from. Deterministic (same inputs → same bytes on both ends).
|
|
fn signable_bytes(topic: &[u8; 32], author: &EndpointId, ts: u64, msg: &GossipMessage) -> Vec<u8> {
|
|
serde_json::to_vec(&(GOSSIP_SIG_DOMAIN, topic, author, ts, msg)).unwrap_or_default()
|
|
}
|
|
|
|
/// Build a signed payload from our secret key for the given room topic.
|
|
fn sign_gossip(secret: &SecretKey, topic: &[u8; 32], ts: u64, msg: GossipMessage) -> GossipPayload {
|
|
let author = secret.public();
|
|
let sig = secret.sign(&signable_bytes(topic, &author, ts, &msg));
|
|
GossipPayload { author, ts, msg, sig }
|
|
}
|
|
|
|
/// Why a received gossip payload was rejected (logging + tests).
|
|
#[derive(Debug, PartialEq, Eq)]
|
|
enum GossipReject {
|
|
/// Signature didn't validate for the claimed `author` — a forgery/spoof.
|
|
BadSignature,
|
|
/// Timestamp outside the freshness window — stale (replay) or implausibly future.
|
|
OutOfWindow,
|
|
/// A signed Announce advertised an address for a different node id.
|
|
AnnounceAddressMismatch,
|
|
}
|
|
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
|
|
enum StateMutationKind {
|
|
Announce,
|
|
Leave,
|
|
}
|
|
|
|
fn gossip_message_kind(msg: &GossipMessage) -> &'static str {
|
|
match msg {
|
|
GossipMessage::Announce(_) => "Announce",
|
|
GossipMessage::Leave => "Leave",
|
|
GossipMessage::Chat { .. } => "Chat",
|
|
}
|
|
}
|
|
|
|
fn state_mutation_kind(msg: &GossipMessage) -> Option<StateMutationKind> {
|
|
match msg {
|
|
GossipMessage::Announce(_) => Some(StateMutationKind::Announce),
|
|
GossipMessage::Leave => Some(StateMutationKind::Leave),
|
|
GossipMessage::Chat { .. } => None,
|
|
}
|
|
}
|
|
|
|
fn admit_state_mutation(
|
|
seen: &mut HashMap<(EndpointId, StateMutationKind), u64>,
|
|
author: EndpointId,
|
|
msg: &GossipMessage,
|
|
ts: u64,
|
|
) -> bool {
|
|
let Some(kind) = state_mutation_kind(msg) else {
|
|
return true;
|
|
};
|
|
let key = (author, kind);
|
|
if seen.get(&key).is_some_and(|last_ts| ts <= *last_ts) {
|
|
return false;
|
|
}
|
|
seen.insert(key, ts);
|
|
true
|
|
}
|
|
|
|
fn peer_state_for_log(state: &PeerState) -> String {
|
|
format!(
|
|
"name={:?}, muted={}, addr_id={}, addrs={}, sharing={}",
|
|
state.name,
|
|
state.is_muted,
|
|
crate::short_id(&state.addr.id.to_string()),
|
|
state.addr.addrs.len(),
|
|
state.sharing.is_some()
|
|
)
|
|
}
|
|
|
|
/// Authenticate a received payload against the room topic and local clock. The
|
|
/// signature must validate for the claimed `author` (closing the author-spoofing
|
|
/// vector, security S2), and the timestamp must be within `window_ms` of
|
|
/// `now_ms`. Pure — no I/O — so it's directly unit-testable.
|
|
fn verify_gossip(
|
|
payload: &GossipPayload,
|
|
topic: &[u8; 32],
|
|
now_ms: u64,
|
|
window_ms: u64,
|
|
) -> Result<(), GossipReject> {
|
|
let bytes = signable_bytes(topic, &payload.author, payload.ts, &payload.msg);
|
|
payload
|
|
.author
|
|
.verify(&bytes, &payload.sig)
|
|
.map_err(|_| GossipReject::BadSignature)?;
|
|
if now_ms.abs_diff(payload.ts) > window_ms {
|
|
return Err(GossipReject::OutOfWindow);
|
|
}
|
|
if let GossipMessage::Announce(state) = &payload.msg
|
|
&& state.addr.id != payload.author {
|
|
return Err(GossipReject::AnnounceAddressMismatch);
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Compute the gossip bootstrap peer set for a (re)join: the ticket's host plus
|
|
/// every peer we already knew about (`extra`), with self removed and ids
|
|
/// de-duplicated. Without the retained `extra` peers, the ROOM CREATOR rejoining
|
|
/// their own room would have an empty list (their ticket names only themselves as
|
|
/// host) and so never re-enter the swarm — that was bug A8. Pure → unit-testable.
|
|
fn compute_bootstrap(
|
|
self_id: EndpointId,
|
|
ticket_host: EndpointId,
|
|
extra: &[EndpointAddr],
|
|
) -> Vec<EndpointId> {
|
|
let mut out: Vec<EndpointId> = Vec::new();
|
|
if ticket_host != self_id {
|
|
out.push(ticket_host);
|
|
}
|
|
for addr in extra {
|
|
if addr.id != self_id && !out.contains(&addr.id) {
|
|
out.push(addr.id);
|
|
}
|
|
}
|
|
out
|
|
}
|
|
|
|
#[derive(Serialize, Deserialize, Clone, Debug)]
|
|
pub enum GossipMessage {
|
|
Announce(PeerState),
|
|
Leave,
|
|
/// A room text-chat message: the author's display name, the text, and a
|
|
/// sender-stamped millisecond timestamp.
|
|
Chat { name: String, text: String, ts: u64 },
|
|
}
|
|
|
|
pub struct IrohGossipState {
|
|
_endpoint: Endpoint,
|
|
gossip: Gossip,
|
|
address_lookup: iroh::address_lookup::memory::MemoryLookup,
|
|
/// Our node secret key, used to sign every outgoing gossip payload so peers
|
|
/// can authenticate us (security S2). Same key the endpoint binds with.
|
|
secret_key: SecretKey,
|
|
self_state: Arc<Mutex<Option<PeerState>>>,
|
|
peers: Arc<Mutex<HashMap<EndpointId, PeerState>>>,
|
|
event_tx: mpsc::Sender<RoomEvent>,
|
|
event_rx: Mutex<Option<mpsc::Receiver<RoomEvent>>>,
|
|
active_topic: Mutex<Option<tokio::task::JoinHandle<()>>>,
|
|
active_topic_id: Mutex<Option<TopicId>>,
|
|
/// Raw 32-byte id of the joined topic, retained so the out-of-task signers
|
|
/// (`update_self_state`/`send_chat`/`leave`) can bind signatures to the room.
|
|
active_topic_bytes: Mutex<Option<[u8; 32]>>,
|
|
active_sender: Mutex<Option<iroh_gossip::api::GossipSender>>,
|
|
}
|
|
|
|
impl IrohGossipState {
|
|
pub fn new(
|
|
endpoint: Endpoint,
|
|
gossip: Gossip,
|
|
address_lookup: iroh::address_lookup::memory::MemoryLookup,
|
|
secret_key: SecretKey,
|
|
) -> Self {
|
|
let (event_tx, event_rx) = mpsc::channel(100);
|
|
Self {
|
|
_endpoint: endpoint,
|
|
gossip,
|
|
address_lookup,
|
|
secret_key,
|
|
self_state: Arc::new(Mutex::new(None)),
|
|
peers: Arc::new(Mutex::new(HashMap::new())),
|
|
event_tx,
|
|
event_rx: Mutex::new(Some(event_rx)),
|
|
active_topic: Mutex::new(None),
|
|
active_topic_id: Mutex::new(None),
|
|
active_topic_bytes: Mutex::new(None),
|
|
active_sender: Mutex::new(None),
|
|
}
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl RoomState for IrohGossipState {
|
|
async fn join(
|
|
&self,
|
|
ticket_str: &str,
|
|
self_state: PeerState,
|
|
extra_bootstrap: Vec<EndpointAddr>,
|
|
) -> Result<(), NetError> {
|
|
crate::log_msg(&format!(
|
|
"RoomState::join: self_id={}, self_name={:?}, ticket={}",
|
|
crate::short_id(&self_state.addr.id.to_string()),
|
|
self_state.name,
|
|
crate::redact_for_log(ticket_str)
|
|
));
|
|
let ticket = ticket_str.parse::<PeerSpeakTicket>()?;
|
|
// Version-namespace the subscribed topic (VERSIONING.md): peers on a
|
|
// different gossip protocol version derive a different topic from the same
|
|
// ticket and never share a swarm. The raw ticket.topic_id stays the room
|
|
// identity (and what signatures bind, below).
|
|
let topic_id = TopicId::from_bytes(crate::protocol::versioned_topic(ticket.topic_id));
|
|
|
|
crate::log_msg(&format!(
|
|
"Parsed ticket. host_id={}, host_addrs={}, topic={}",
|
|
crate::short_id(&ticket.host_addr.id.to_string()),
|
|
ticket.host_addr.addrs.len(),
|
|
crate::short_bytes_hex(&ticket.topic_id)
|
|
));
|
|
|
|
// Stop any currently running topic
|
|
let _ = self.leave().await;
|
|
|
|
// Make every dial target resolvable: the ticket host plus any retained
|
|
// peers handed in (their addresses may have aged out of a fresh endpoint's
|
|
// book even though the persistent lookup usually still holds them).
|
|
self.address_lookup.add_endpoint_info(ticket.host_addr.clone());
|
|
for addr in &extra_bootstrap {
|
|
self.address_lookup.add_endpoint_info(addr.clone());
|
|
}
|
|
|
|
// Bootstrap to the ticket host AND every retained peer (minus self). The
|
|
// retained peers are what let the room creator rejoin a room they left —
|
|
// their own ticket names only themselves as host (bug A8).
|
|
let bootstrap_peers =
|
|
compute_bootstrap(self_state.addr.id, ticket.host_addr.id, &extra_bootstrap);
|
|
crate::log_msg(&format!("Bootstrap peers for join: {:?}", bootstrap_peers));
|
|
|
|
let gossip_topic = self.gossip.subscribe(topic_id, bootstrap_peers).await
|
|
.map_err(|e| {
|
|
let err = format!("Failed to join gossip topic: {}", e);
|
|
crate::log_msg(&err);
|
|
NetError::Gossip(err)
|
|
})?;
|
|
|
|
let (gossip_sender, mut gossip_receiver) = gossip_topic.split();
|
|
|
|
*self.self_state.lock().unwrap() = Some(self_state.clone());
|
|
*self.active_topic_id.lock().unwrap() = Some(topic_id);
|
|
*self.active_topic_bytes.lock().unwrap() = Some(ticket.topic_id);
|
|
*self.active_sender.lock().unwrap() = Some(gossip_sender.clone());
|
|
|
|
let event_tx = self.event_tx.clone();
|
|
let peers = self.peers.clone();
|
|
let address_lookup = self.address_lookup.clone();
|
|
let self_state_clone = self.self_state.clone();
|
|
let gossip_sender_clone = gossip_sender.clone();
|
|
let self_id = self_state.addr.id;
|
|
// The signing key + room topic the spawned loop needs to sign our own
|
|
// broadcasts and authenticate every inbound payload (security S2).
|
|
let secret_key = self.secret_key.clone();
|
|
let topic_bytes = ticket.topic_id;
|
|
|
|
let handle = tokio::spawn(async move {
|
|
crate::log_msg(&format!("Spawned gossip topic loop for self_id={:?}", self_id));
|
|
let mut state_mutations_seen = HashMap::new();
|
|
|
|
// Broadcast initial state
|
|
let initial_payload = {
|
|
let guard = self_state_clone.lock().unwrap();
|
|
guard.as_ref().map(|s| sign_gossip(
|
|
&secret_key,
|
|
&topic_bytes,
|
|
now_millis(),
|
|
GossipMessage::Announce(s.clone()),
|
|
))
|
|
};
|
|
|
|
if let Some(payload) = initial_payload
|
|
&& let Ok(bytes) = serde_json::to_vec(&payload) {
|
|
crate::log_msg(&format!("Broadcasting initial state from self_id={:?}", self_id));
|
|
let _ = gossip_sender_clone.broadcast(bytes.into()).await;
|
|
}
|
|
|
|
// Stream topic messages
|
|
while let Some(res) = gossip_receiver.next().await {
|
|
match res {
|
|
Ok(iroh_gossip::api::Event::Received(msg)) => {
|
|
crate::log_msg(&format!("Gossip received Event::Received from delivery={:?}", msg.delivered_from));
|
|
match serde_json::from_slice::<GossipPayload>(&msg.content) {
|
|
Ok(payload) => {
|
|
// Authenticate before trusting `author` for ANY
|
|
// action: a forged/stale payload is dropped here
|
|
// so it can't impersonate, evict, or poison
|
|
// presence/address-book (security S2).
|
|
if let Err(reason) =
|
|
verify_gossip(&payload, &topic_bytes, now_millis(), GOSSIP_FRESHNESS_MS)
|
|
{
|
|
crate::log_msg(&format!(
|
|
"Gossip dropped unauthenticated/stale payload claiming author={:?}: {:?}",
|
|
payload.author, reason
|
|
));
|
|
continue;
|
|
}
|
|
let our_id = {
|
|
self_state_clone.lock().unwrap()
|
|
.as_ref()
|
|
.map(|s| s.addr.id)
|
|
};
|
|
|
|
if Some(payload.author) == our_id {
|
|
crate::log_msg("Gossip Event::Received from ourselves; ignoring");
|
|
continue;
|
|
}
|
|
|
|
if !admit_state_mutation(
|
|
&mut state_mutations_seen,
|
|
payload.author,
|
|
&payload.msg,
|
|
payload.ts,
|
|
) {
|
|
crate::log_msg(&format!(
|
|
"Gossip dropped replayed state mutation author={}, kind={}, ts={}",
|
|
crate::short_id(&payload.author.to_string()),
|
|
gossip_message_kind(&payload.msg),
|
|
payload.ts
|
|
));
|
|
continue;
|
|
}
|
|
|
|
crate::log_msg(&format!(
|
|
"Gossip Event::Received author={}, kind={}",
|
|
crate::short_id(&payload.author.to_string()),
|
|
gossip_message_kind(&payload.msg)
|
|
));
|
|
|
|
match payload.msg {
|
|
GossipMessage::Announce(mut state) => {
|
|
// Presence names are untrusted (and author-
|
|
// spoofable): sanitize at ingest so every
|
|
// consumer gets a safe value (security S4).
|
|
state.name = crate::sanitize::sanitize_name(&state.name);
|
|
// A peer's avatar is equally untrusted: a
|
|
// custom image is validated (size + safe
|
|
// decode within bounds) or downgraded to a
|
|
// monogram, so a malformed/oversized/bomb
|
|
// image can't crash or exhaust us (W4).
|
|
state.avatar = state.avatar.sanitize_incoming();
|
|
// Screen-share tickets are capabilities and
|
|
// peer-supplied: cap/validate once at ingest
|
|
// so invalid offers never render a Watch button.
|
|
state.sharing = state.sharing.and_then(crate::screenshare::sanitize_ticket);
|
|
let (is_new, state_changed) = {
|
|
let mut peer_map = peers.lock().unwrap();
|
|
let is_new = !peer_map.contains_key(&payload.author);
|
|
let state_changed = peer_map.get(&payload.author) != Some(&state);
|
|
if is_new || state_changed {
|
|
peer_map.insert(payload.author, state.clone());
|
|
}
|
|
(is_new, state_changed)
|
|
};
|
|
|
|
if is_new {
|
|
crate::log_msg(&format!(
|
|
"Gossip new peer joined: {}, state: {}",
|
|
crate::short_id(&payload.author.to_string()),
|
|
peer_state_for_log(&state)
|
|
));
|
|
address_lookup.add_endpoint_info(state.addr.clone());
|
|
let _ = event_tx.send(RoomEvent::PeerJoined(payload.author, state)).await;
|
|
} else if state_changed {
|
|
crate::log_msg(&format!(
|
|
"Gossip peer state updated: {}, state: {}",
|
|
crate::short_id(&payload.author.to_string()),
|
|
peer_state_for_log(&state)
|
|
));
|
|
let _ = event_tx.send(RoomEvent::PeerUpdated(payload.author, state)).await;
|
|
}
|
|
}
|
|
GossipMessage::Leave => {
|
|
crate::log_msg(&format!("Gossip peer leave request from author={:?}", payload.author));
|
|
let removed = peers.lock().unwrap().remove(&payload.author).is_some();
|
|
if removed {
|
|
let _ = event_tx.send(RoomEvent::PeerLeft(payload.author)).await;
|
|
}
|
|
}
|
|
GossipMessage::Chat { name, text, ts } => {
|
|
crate::log_msg(&format!("Gossip chat from author={:?}", payload.author));
|
|
let _ = event_tx.send(RoomEvent::ChatMessage {
|
|
from: payload.author,
|
|
name,
|
|
text,
|
|
ts,
|
|
}).await;
|
|
}
|
|
}
|
|
}
|
|
Err(e) => {
|
|
crate::log_msg(&format!("Gossip failed to deserialize payload: {:?}", e));
|
|
}
|
|
}
|
|
}
|
|
Ok(iroh_gossip::api::Event::NeighborUp(peer_id)) => {
|
|
crate::log_msg(&format!("Gossip event: NeighborUp={:?}", peer_id));
|
|
// Resend state on new neighbor connection to guarantee synchronization
|
|
let payload_opt = {
|
|
let guard = self_state_clone.lock().unwrap();
|
|
guard.as_ref().map(|state| sign_gossip(
|
|
&secret_key,
|
|
&topic_bytes,
|
|
now_millis(),
|
|
GossipMessage::Announce(state.clone()),
|
|
))
|
|
};
|
|
if let Some(payload) = payload_opt
|
|
&& let Ok(bytes) = serde_json::to_vec(&payload) {
|
|
crate::log_msg(&format!("Broadcasting state to new neighbor={:?}", peer_id));
|
|
let _ = gossip_sender_clone.broadcast(bytes.into()).await;
|
|
}
|
|
}
|
|
Ok(iroh_gossip::api::Event::NeighborDown(peer_id)) => {
|
|
crate::log_msg(&format!("Gossip event: NeighborDown={:?}", peer_id));
|
|
// A NeighborDown is a *transient* loss, not a graceful
|
|
// leave: emit PeerConnectionLost so the core marks the peer
|
|
// "reconnecting" and keeps its audio supervisor redialing,
|
|
// rather than tearing everything down. (Treating this as a
|
|
// PeerLeft is exactly what defeated reconnect in the field —
|
|
// it aborted the supervisor ~30s in.) We still drop our
|
|
// cached presence entry; a rejoin re-announces as new.
|
|
let removed = peers.lock().unwrap().remove(&peer_id).is_some();
|
|
if removed {
|
|
crate::log_msg(&format!("Peer connection lost (NeighborDown): {:?}", peer_id));
|
|
let _ = event_tx.send(RoomEvent::PeerConnectionLost(peer_id)).await;
|
|
}
|
|
}
|
|
Ok(other) => {
|
|
crate::log_msg(&format!("Gossip other event: {:?}", other));
|
|
}
|
|
Err(e) => {
|
|
crate::log_msg(&format!("Gossip error event: {:?}", e));
|
|
}
|
|
}
|
|
}
|
|
crate::log_msg("Gossip topic loop terminated");
|
|
});
|
|
|
|
*self.active_topic.lock().unwrap() = Some(handle);
|
|
Ok(())
|
|
}
|
|
|
|
async fn update_self_state(&self, self_state: PeerState) -> Result<(), NetError> {
|
|
crate::log_msg(&format!(
|
|
"RoomState::update_self_state: state: {}",
|
|
peer_state_for_log(&self_state)
|
|
));
|
|
*self.self_state.lock().unwrap() = Some(self_state.clone());
|
|
|
|
let sender_opt = self.active_sender.lock().unwrap().clone();
|
|
let topic_opt = *self.active_topic_bytes.lock().unwrap();
|
|
if let (Some(sender), Some(topic)) = (sender_opt, topic_opt) {
|
|
let payload = sign_gossip(
|
|
&self.secret_key,
|
|
&topic,
|
|
now_millis(),
|
|
GossipMessage::Announce(self_state),
|
|
);
|
|
if let Ok(bytes) = serde_json::to_vec(&payload) {
|
|
crate::log_msg("Broadcasting updated self state to gossip");
|
|
sender.broadcast(bytes.into()).await
|
|
.map_err(|e| NetError::Gossip(e.to_string()))?;
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
async fn send_chat(&self, text: String) -> Result<(), NetError> {
|
|
let name = {
|
|
let guard = self.self_state.lock().unwrap();
|
|
match guard.as_ref() {
|
|
Some(s) => s.name.clone(),
|
|
None => return Err(NetError::Other("Not in a room".to_string())),
|
|
}
|
|
};
|
|
let ts = now_millis();
|
|
|
|
let sender_opt = self.active_sender.lock().unwrap().clone();
|
|
let topic_opt = *self.active_topic_bytes.lock().unwrap();
|
|
if let (Some(sender), Some(topic)) = (sender_opt, topic_opt) {
|
|
let payload = sign_gossip(
|
|
&self.secret_key,
|
|
&topic,
|
|
ts,
|
|
GossipMessage::Chat { name, text, ts },
|
|
);
|
|
if let Ok(bytes) = serde_json::to_vec(&payload) {
|
|
sender.broadcast(bytes.into()).await
|
|
.map_err(|e| NetError::Gossip(e.to_string()))?;
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
async fn leave(&self) -> Result<(), NetError> {
|
|
crate::log_msg("RoomState::leave called");
|
|
{
|
|
let mut handle_guard = self.active_topic.lock().unwrap();
|
|
if let Some(handle) = handle_guard.take() {
|
|
crate::log_msg("Aborting active gossip topic background task");
|
|
handle.abort();
|
|
}
|
|
}
|
|
|
|
*self.active_topic_id.lock().unwrap() = None;
|
|
let topic_opt = self.active_topic_bytes.lock().unwrap().take();
|
|
|
|
let sender_opt = self.active_sender.lock().unwrap().take();
|
|
if let (Some(sender), Some(topic)) = (sender_opt, topic_opt) {
|
|
let payload = sign_gossip(
|
|
&self.secret_key,
|
|
&topic,
|
|
now_millis(),
|
|
GossipMessage::Leave,
|
|
);
|
|
if let Ok(bytes) = serde_json::to_vec(&payload) {
|
|
crate::log_msg("Broadcasting Leave message to gossip");
|
|
let _ = sender.broadcast(bytes.into()).await;
|
|
}
|
|
}
|
|
|
|
self.peers.lock().unwrap().clear();
|
|
Ok(())
|
|
}
|
|
|
|
fn active_peers(&self) -> Vec<(EndpointId, PeerState)> {
|
|
let guard = self.peers.lock().unwrap();
|
|
guard.iter().map(|(k, v)| (*k, v.clone())).collect()
|
|
}
|
|
|
|
async fn subscribe_events(&self) -> Result<Receiver<RoomEvent>, NetError> {
|
|
let mut rx_guard = self.event_rx.lock().unwrap();
|
|
if let Some(rx) = rx_guard.take() {
|
|
Ok(rx)
|
|
} else {
|
|
Err(NetError::Other("Events already subscribed".to_string()))
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::network::PeerState;
|
|
use iroh::SecretKey;
|
|
use std::collections::HashMap;
|
|
|
|
fn sample_peer_state_for(id: EndpointId) -> PeerState {
|
|
let addr = iroh::EndpointAddr::from(id);
|
|
PeerState {
|
|
name: "TestPeerGossip".to_string(),
|
|
is_muted: true,
|
|
addr,
|
|
sharing: None,
|
|
avatar: crate::avatar::Avatar::default(),
|
|
}
|
|
}
|
|
|
|
fn fresh_id() -> EndpointId {
|
|
SecretKey::generate().public()
|
|
}
|
|
|
|
fn addr_of(id: EndpointId) -> EndpointAddr {
|
|
EndpointAddr::from(id)
|
|
}
|
|
|
|
#[test]
|
|
fn bootstrap_client_dials_host() {
|
|
// A non-host (client) with no retained peers dials just the ticket host.
|
|
let me = fresh_id();
|
|
let host = fresh_id();
|
|
assert_eq!(compute_bootstrap(me, host, &[]), vec![host]);
|
|
}
|
|
|
|
#[test]
|
|
fn bootstrap_host_rejoin_dials_retained_peers() {
|
|
// The A8 regression: the room creator's ticket names themselves as host,
|
|
// so the ticket-host contributes nothing — the retained peer is the only
|
|
// (and essential) dial target. An empty result here would reproduce A8.
|
|
let me = fresh_id();
|
|
let peer = fresh_id();
|
|
let bootstrap = compute_bootstrap(me, /* ticket_host = */ me, &[addr_of(peer)]);
|
|
assert_eq!(bootstrap, vec![peer]);
|
|
}
|
|
|
|
#[test]
|
|
fn bootstrap_host_rejoin_with_no_retained_peers_is_empty() {
|
|
// Host rejoin before anyone was ever seen: nothing to dial (unavoidable),
|
|
// but it must not contain self.
|
|
let me = fresh_id();
|
|
assert!(compute_bootstrap(me, me, &[]).is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn bootstrap_excludes_self_and_dedups_host() {
|
|
// Self is never dialed (even if it shows up among retained peers), and a
|
|
// peer that equals the ticket host isn't listed twice.
|
|
let me = fresh_id();
|
|
let host = fresh_id();
|
|
let other = fresh_id();
|
|
let extra = [addr_of(me), addr_of(host), addr_of(other)];
|
|
let bootstrap = compute_bootstrap(me, host, &extra);
|
|
assert_eq!(bootstrap, vec![host, other]);
|
|
assert!(!bootstrap.contains(&me));
|
|
}
|
|
|
|
#[test]
|
|
fn test_gossip_message_leave_round_trip() {
|
|
let original = GossipMessage::Leave;
|
|
let serialized = serde_json::to_string(&original).unwrap();
|
|
let deserialized: GossipMessage = serde_json::from_str(&serialized).unwrap();
|
|
assert!(matches!(deserialized, GossipMessage::Leave));
|
|
}
|
|
|
|
#[test]
|
|
fn test_gossip_payload_announce_round_trip() {
|
|
let secret = SecretKey::generate();
|
|
let topic = [9u8; 32];
|
|
let peer_state = sample_peer_state_for(secret.public());
|
|
let payload = sign_gossip(&secret, &topic, 1000, GossipMessage::Announce(peer_state.clone()));
|
|
|
|
let serialized = serde_json::to_string(&payload).unwrap();
|
|
let deserialized: GossipPayload = serde_json::from_str(&serialized).unwrap();
|
|
|
|
// Author is derived from the signing key, and the signed payload still
|
|
// verifies after a serde round-trip.
|
|
assert_eq!(deserialized.author, secret.public());
|
|
assert!(verify_gossip(&deserialized, &topic, 1000, GOSSIP_FRESHNESS_MS).is_ok());
|
|
match deserialized.msg {
|
|
GossipMessage::Announce(state) => {
|
|
assert_eq!(state, peer_state);
|
|
}
|
|
GossipMessage::Leave | GossipMessage::Chat { .. } => {
|
|
panic!("Expected GossipMessage::Announce");
|
|
}
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn test_gossip_message_chat_round_trip() {
|
|
// Test normal chat message
|
|
let original = GossipMessage::Chat {
|
|
name: "Alice".to_string(),
|
|
text: "Hello".to_string(),
|
|
ts: 123456789,
|
|
};
|
|
let serialized = serde_json::to_string(&original).unwrap();
|
|
let deserialized: GossipMessage = serde_json::from_str(&serialized).unwrap();
|
|
if let GossipMessage::Chat { name, text, ts } = deserialized {
|
|
assert_eq!(name, "Alice");
|
|
assert_eq!(text, "Hello");
|
|
assert_eq!(ts, 123456789);
|
|
} else {
|
|
panic!("Expected GossipMessage::Chat");
|
|
}
|
|
|
|
// Test empty strings and large timestamp
|
|
let original_empty = GossipMessage::Chat {
|
|
name: "".to_string(),
|
|
text: "".to_string(),
|
|
ts: u64::MAX,
|
|
};
|
|
let serialized_empty = serde_json::to_string(&original_empty).unwrap();
|
|
let deserialized_empty: GossipMessage = serde_json::from_str(&serialized_empty).unwrap();
|
|
if let GossipMessage::Chat { name, text, ts } = deserialized_empty {
|
|
assert_eq!(name, "");
|
|
assert_eq!(text, "");
|
|
assert_eq!(ts, u64::MAX);
|
|
} else {
|
|
panic!("Expected GossipMessage::Chat");
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn test_gossip_payload_chat_round_trip() {
|
|
let secret = SecretKey::generate();
|
|
let topic = [3u8; 32];
|
|
let payload = sign_gossip(
|
|
&secret,
|
|
&topic,
|
|
987654321,
|
|
GossipMessage::Chat {
|
|
name: "Bob".to_string(),
|
|
text: "Hi there".to_string(),
|
|
ts: 987654321,
|
|
},
|
|
);
|
|
|
|
let serialized = serde_json::to_string(&payload).unwrap();
|
|
let deserialized: GossipPayload = serde_json::from_str(&serialized).unwrap();
|
|
|
|
assert_eq!(deserialized.author, secret.public());
|
|
if let GossipMessage::Chat { name, text, ts } = deserialized.msg {
|
|
assert_eq!(name, "Bob");
|
|
assert_eq!(text, "Hi there");
|
|
assert_eq!(ts, 987654321);
|
|
} else {
|
|
panic!("Expected GossipMessage::Chat");
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn test_gossip_chat_unicode_round_trip() {
|
|
let original = GossipMessage::Chat {
|
|
name: "🎙 User".to_string(),
|
|
text: "héllo 🎙 世界".to_string(),
|
|
ts: 1717171717,
|
|
};
|
|
let serialized = serde_json::to_string(&original).unwrap();
|
|
let deserialized: GossipMessage = serde_json::from_str(&serialized).unwrap();
|
|
if let GossipMessage::Chat { name, text, ts } = deserialized {
|
|
assert_eq!(name, "🎙 User");
|
|
assert_eq!(text, "héllo 🎙 世界");
|
|
assert_eq!(ts, 1717171717);
|
|
} else {
|
|
panic!("Expected GossipMessage::Chat");
|
|
}
|
|
}
|
|
|
|
// --- S2 authentication (verify_gossip) ---
|
|
|
|
#[test]
|
|
fn verify_accepts_a_genuine_signed_payload() {
|
|
let secret = SecretKey::generate();
|
|
let topic = [1u8; 32];
|
|
let p = sign_gossip(&secret, &topic, 5_000, GossipMessage::Leave);
|
|
assert_eq!(verify_gossip(&p, &topic, 5_000, GOSSIP_FRESHNESS_MS), Ok(()));
|
|
}
|
|
|
|
#[test]
|
|
fn verify_rejects_a_forged_author() {
|
|
// Attacker signs with their OWN key but claims to be the victim — the
|
|
// classic S2 spoof. The signature can't validate against the victim key.
|
|
let attacker = SecretKey::generate();
|
|
let victim = SecretKey::generate();
|
|
let topic = [2u8; 32];
|
|
let mut p = sign_gossip(&attacker, &topic, 5_000, GossipMessage::Leave);
|
|
p.author = victim.public(); // forge the claimed author
|
|
assert_eq!(
|
|
verify_gossip(&p, &topic, 5_000, GOSSIP_FRESHNESS_MS),
|
|
Err(GossipReject::BadSignature)
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn verify_rejects_a_tampered_message() {
|
|
// Flipping the message after signing must invalidate the signature.
|
|
let secret = SecretKey::generate();
|
|
let topic = [4u8; 32];
|
|
let mut p = sign_gossip(&secret, &topic, 5_000, GossipMessage::Leave);
|
|
p.msg = GossipMessage::Chat { name: "x".into(), text: "y".into(), ts: 5_000 };
|
|
assert_eq!(
|
|
verify_gossip(&p, &topic, 5_000, GOSSIP_FRESHNESS_MS),
|
|
Err(GossipReject::BadSignature)
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn verify_rejects_cross_room_replay() {
|
|
// A payload signed for one topic must not validate against another room
|
|
// (the topic is bound into the signed bytes).
|
|
let secret = SecretKey::generate();
|
|
let p = sign_gossip(&secret, &[7u8; 32], 5_000, GossipMessage::Leave);
|
|
assert_eq!(
|
|
verify_gossip(&p, &[8u8; 32], 5_000, GOSSIP_FRESHNESS_MS),
|
|
Err(GossipReject::BadSignature)
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn verify_rejects_stale_and_future_timestamps() {
|
|
let secret = SecretKey::generate();
|
|
let topic = [5u8; 32];
|
|
let p = sign_gossip(&secret, &topic, 1_000_000, GossipMessage::Leave);
|
|
// Far in the past relative to "now" → stale (replay).
|
|
assert_eq!(
|
|
verify_gossip(&p, &topic, 1_000_000 + GOSSIP_FRESHNESS_MS + 1, GOSSIP_FRESHNESS_MS),
|
|
Err(GossipReject::OutOfWindow)
|
|
);
|
|
// Implausibly future.
|
|
assert_eq!(
|
|
verify_gossip(&p, &topic, 1_000_000 - GOSSIP_FRESHNESS_MS - 1, GOSSIP_FRESHNESS_MS),
|
|
Err(GossipReject::OutOfWindow)
|
|
);
|
|
// Within the window (clock skew tolerance) → accepted.
|
|
assert!(verify_gossip(&p, &topic, 1_000_000 + GOSSIP_FRESHNESS_MS - 1, GOSSIP_FRESHNESS_MS).is_ok());
|
|
}
|
|
|
|
#[test]
|
|
fn verify_rejects_announce_with_address_for_another_identity() {
|
|
let signer = SecretKey::generate();
|
|
let advertised = SecretKey::generate();
|
|
let topic = [6u8; 32];
|
|
let state = sample_peer_state_for(advertised.public());
|
|
let p = sign_gossip(&signer, &topic, 5_000, GossipMessage::Announce(state));
|
|
|
|
assert_eq!(
|
|
verify_gossip(&p, &topic, 5_000, GOSSIP_FRESHNESS_MS),
|
|
Err(GossipReject::AnnounceAddressMismatch)
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn state_mutation_replay_gate_drops_replayed_leave_and_announce() {
|
|
let author = fresh_id();
|
|
let mut seen = HashMap::new();
|
|
|
|
assert!(admit_state_mutation(&mut seen, author, &GossipMessage::Leave, 10));
|
|
assert!(!admit_state_mutation(&mut seen, author, &GossipMessage::Leave, 10));
|
|
assert!(!admit_state_mutation(&mut seen, author, &GossipMessage::Leave, 9));
|
|
assert!(admit_state_mutation(&mut seen, author, &GossipMessage::Leave, 11));
|
|
|
|
let announce = GossipMessage::Announce(sample_peer_state_for(author));
|
|
assert!(admit_state_mutation(&mut seen, author, &announce, 10));
|
|
assert!(!admit_state_mutation(&mut seen, author, &announce, 10));
|
|
assert!(!admit_state_mutation(&mut seen, author, &announce, 9));
|
|
assert!(admit_state_mutation(&mut seen, author, &announce, 12));
|
|
}
|
|
|
|
#[test]
|
|
fn state_mutation_replay_gate_leaves_chat_ordering_untouched() {
|
|
let author = fresh_id();
|
|
let mut seen = HashMap::new();
|
|
let later_chat = GossipMessage::Chat { name: "A".into(), text: "later".into(), ts: 200 };
|
|
let earlier_chat = GossipMessage::Chat { name: "A".into(), text: "earlier".into(), ts: 100 };
|
|
|
|
assert!(admit_state_mutation(&mut seen, author, &later_chat, 200));
|
|
assert!(admit_state_mutation(&mut seen, author, &earlier_chat, 100));
|
|
assert!(admit_state_mutation(&mut seen, author, &later_chat, 200));
|
|
assert!(seen.is_empty(), "chat must not populate the state-mutation replay map");
|
|
}
|
|
|
|
#[test]
|
|
fn state_mutation_replay_gate_is_per_author_and_kind() {
|
|
let author = fresh_id();
|
|
let other = fresh_id();
|
|
let mut seen = HashMap::new();
|
|
let announce = GossipMessage::Announce(sample_peer_state_for(author));
|
|
|
|
assert!(admit_state_mutation(&mut seen, author, &GossipMessage::Leave, 5));
|
|
assert!(admit_state_mutation(&mut seen, author, &announce, 5));
|
|
assert!(admit_state_mutation(&mut seen, other, &GossipMessage::Leave, 5));
|
|
}
|
|
}
|