mollusk/peerspeak
Public Access
1
0
Fork 0
Code Issues Pull Requests 3 Actions 1 Packages Projects Releases 10 Wiki Activity

style: apply cargo fmt across the crate (A20)

Browse Source 
The repo never enforced rustfmt, so formatting had drifted broadly. This is a
single mechanical `cargo fmt` pass over the whole crate (no behavioral change;
lib suite green, 493 passed). Going forward fmt should be enforced (planned CI
fmt --check step). Part of the 0.6.1 hygiene pass.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
mollusk
2026-06-29 02:11:44 -04:00
co-authored by Claude Opus 4.8
parent e0325d4590
commit d0a16cb8b9
54 changed files with 3761 additions and 1789 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/network/gossip.rs
+274 -95
View File
@@ -1,14 +1,14 @@
use crate::network::{RoomState, NetError, PeerState, RoomEvent, PeerSpeakTicket};
use crate::network::{NetError, PeerSpeakTicket, PeerState, RoomEvent, RoomState};
use async_trait::async_trait;
use iroh::{Endpoint, EndpointAddr, EndpointId, SecretKey, Signature, TransportAddr};
use iroh_gossip::net::Gossip;
use iroh_gossip::proto::TopicId;
use serde::{Deserialize, Serialize};
use std::collections::{BTreeSet, HashMap, HashSet};
use std::sync::{Arc, Mutex};
use tokio::sync::mpsc;
use tokio::sync::mpsc::Receiver;
use std::sync::{Arc, Mutex};
use std::collections::{BTreeSet, HashMap, HashSet};
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.
@@ -78,7 +78,12 @@ fn signable_bytes(topic: &[u8; 32], author: &EndpointId, ts: u64, msg: &GossipMe
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 }
GossipPayload {
author,
ts,
msg,
sig,
}
}
/// Why a received gossip payload was rejected (logging + tests).
@@ -233,7 +238,10 @@ impl ClockSkewMonitor {
}
fn drop_oldest_authors(&mut self) {
let remove_count = self.authors.len().saturating_sub(CLOCK_SKEW_AUTHORS_SOFT_CAP);
let remove_count = self
.authors
.len()
.saturating_sub(CLOCK_SKEW_AUTHORS_SOFT_CAP);
let mut by_age: Vec<_> = self
.authors
.iter()
@@ -339,7 +347,8 @@ fn verify_gossip(
return Err(GossipReject::OutOfWindow);
}
if let GossipMessage::Announce(state) = &payload.msg
&& state.addr.id != payload.author {
&& state.addr.id != payload.author
{
return Err(GossipReject::AnnounceAddressMismatch);
}
Ok(())
@@ -466,7 +475,8 @@ impl RoomState for IrohGossipState {
// 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());
self.address_lookup
.add_endpoint_info(ticket.host_addr.clone());
for addr in &extra_bootstrap {
self.address_lookup.add_endpoint_info(addr.clone());
}
@@ -478,7 +488,10 @@ impl RoomState for IrohGossipState {
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
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);
@@ -505,32 +518,44 @@ impl RoomState for IrohGossipState {
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));
crate::log_msg(&format!(
"Spawned gossip topic loop for self_id={:?}",
self_id
));
let mut state_mutations_seen = HashMap::new();
let mut clock_skew_monitor = ClockSkewMonitor::default();
// 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()),
))
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;
}
&& 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));
crate::log_msg(&format!(
"Gossip received Event::Received from delivery={:?}",
msg.delivered_from
));
// Reject oversized frames BEFORE deserializing: parsing
// allocates, so a size check has to precede `from_slice` to
// bound the memory a hostile peer can make us hold.
@@ -577,13 +602,13 @@ impl RoomState for IrohGossipState {
continue;
}
let our_id = {
self_state_clone.lock().unwrap()
.as_ref()
.map(|s| s.addr.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");
crate::log_msg(
"Gossip Event::Received from ourselves; ignoring",
);
continue;
}
@@ -635,7 +660,9 @@ impl RoomState for IrohGossipState {
// 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);
state.sharing = state
.sharing
.and_then(crate::screenshare::sanitize_ticket);
// The game-presence label is untrusted
// peer text like the name: sanitize +
// length-cap at ingest (strip bidi/control,
@@ -651,8 +678,11 @@ impl RoomState for IrohGossipState {
// bounds a future fetch (reject anything
// outside the attachment cap).
state.music = state.music.and_then(|mut m| {
let name = crate::sanitize::sanitize_game_label(&m.name);
if name.is_empty() || !crate::files::size_within_cap(m.size) {
let name =
crate::sanitize::sanitize_game_label(&m.name);
if name.is_empty()
|| !crate::files::size_within_cap(m.size)
{
return None;
}
m.name = name;
@@ -679,8 +709,10 @@ impl RoomState for IrohGossipState {
// cleared ONLY once re-admitted — clearing it before
// a possible reject would orphan its recovery state
// (Tier C F-01 audit).
let is_reconnecting =
disconnected_peers.lock().unwrap().contains(&payload.author);
let is_reconnecting = disconnected_peers
.lock()
.unwrap()
.contains(&payload.author);
let admitted = {
let mut peer_map = peers.lock().unwrap();
let is_new = !peer_map.contains_key(&payload.author);
@@ -689,11 +721,17 @@ impl RoomState for IrohGossipState {
// memory/tasks/dials without bound (Tier C
// F-01). Existing-peer updates and reconnects
// are exempt; only brand-new authors are gated.
let subject_to_cap = announce_subject_to_cap(is_new, is_reconnecting);
if !admit_into_roster(peer_map.len(), subject_to_cap, MAX_ACTIVE_PEERS) {
let subject_to_cap =
announce_subject_to_cap(is_new, is_reconnecting);
if !admit_into_roster(
peer_map.len(),
subject_to_cap,
MAX_ACTIVE_PEERS,
) {
None
} else {
let state_changed = peer_map.get(&payload.author) != Some(&state);
let state_changed =
peer_map.get(&payload.author) != Some(&state);
if is_new || state_changed {
peer_map.insert(payload.author, state.clone());
}
@@ -723,35 +761,54 @@ impl RoomState for IrohGossipState {
// address set, so leave/re-announce cycles
// can't accumulate attacker-supplied history
// (Tier C F-01).
let _ = address_lookup.set_endpoint_info(state.addr.clone());
let _ = event_tx.send(RoomEvent::PeerJoined(payload.author, state)).await;
let _ = address_lookup
.set_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;
let _ = event_tx
.send(RoomEvent::PeerUpdated(payload.author, state))
.await;
}
}
GossipMessage::Leave => {
crate::log_msg(&format!("Gossip peer leave request from author={:?}", payload.author));
crate::log_msg(&format!(
"Gossip peer leave request from author={:?}",
payload.author
));
// Drop this id's address-lookup entry so cycling
// distinct identities through Announce→Leave can't
// grow the lookup for the room's lifetime (Tier C
// F-01 audit). Re-announce re-populates it.
let _ = address_lookup.remove_endpoint_info(payload.author);
let removed = peers.lock().unwrap().remove(&payload.author).is_some();
let removed =
peers.lock().unwrap().remove(&payload.author).is_some();
let was_disconnected = disconnected_peers
.lock()
.unwrap()
.remove(&payload.author);
if removed || was_disconnected {
let _ = event_tx.send(RoomEvent::PeerLeft(payload.author)).await;
let _ = event_tx
.send(RoomEvent::PeerLeft(payload.author))
.await;
}
}
GossipMessage::Chat { name, text, ts, attachment } => {
crate::log_msg(&format!("Gossip chat from author={:?}", payload.author));
GossipMessage::Chat {
name,
text,
ts,
attachment,
} => {
crate::log_msg(&format!(
"Gossip chat from author={:?}",
payload.author
));
// Defensively normalize an untrusted attachment
// descriptor: sanitize the filename and drop it
// entirely if it declares an out-of-cap size.
@@ -762,18 +819,23 @@ impl RoomState for IrohGossipState {
a.name = crate::files::sanitize_filename(&a.name);
Some(a)
});
let _ = event_tx.send(RoomEvent::ChatMessage {
from: payload.author,
name,
text,
ts,
attachment,
}).await;
let _ = event_tx
.send(RoomEvent::ChatMessage {
from: payload.author,
name,
text,
ts,
attachment,
})
.await;
}
}
}
Err(e) => {
crate::log_msg(&format!("Gossip failed to deserialize payload: {:?}", e));
crate::log_msg(&format!(
"Gossip failed to deserialize payload: {:?}",
e
));
}
}
}
@@ -782,18 +844,24 @@ impl RoomState for IrohGossipState {
// 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()),
))
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;
}
&& 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));
@@ -807,7 +875,10 @@ impl RoomState for IrohGossipState {
let removed = peers.lock().unwrap().remove(&peer_id).is_some();
if removed {
disconnected_peers.lock().unwrap().insert(peer_id);
crate::log_msg(&format!("Peer connection lost (NeighborDown): {:?}", peer_id));
crate::log_msg(&format!(
"Peer connection lost (NeighborDown): {:?}",
peer_id
));
let _ = event_tx.send(RoomEvent::PeerConnectionLost(peer_id)).await;
}
}
@@ -844,7 +915,9 @@ impl RoomState for IrohGossipState {
);
if let Ok(bytes) = serde_json::to_vec(&payload) {
crate::log_msg("Broadcasting updated self state to gossip");
sender.broadcast(bytes.into()).await
sender
.broadcast(bytes.into())
.await
.map_err(|e| NetError::Gossip(e.to_string()))?;
}
}
@@ -909,10 +982,17 @@ impl RoomState for IrohGossipState {
&self.secret_key,
&topic,
ts,
GossipMessage::Chat { name, text, ts, attachment },
GossipMessage::Chat {
name,
text,
ts,
attachment,
},
);
if let Ok(bytes) = serde_json::to_vec(&payload) {
sender.broadcast(bytes.into()).await
sender
.broadcast(bytes.into())
.await
.map_err(|e| NetError::Gossip(e.to_string()))?;
}
}
@@ -934,12 +1014,7 @@ impl RoomState for IrohGossipState {
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,
);
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;
@@ -1060,7 +1135,9 @@ mod tests {
#[test]
fn reconnecting_and_existing_peers_are_exempt_from_the_cap() {
// A brand-new author counts against the cap...
assert!(announce_subject_to_cap(/* is_new */ true, /* is_reconnecting */ false));
assert!(announce_subject_to_cap(
/* is_new */ true, /* is_reconnecting */ false
));
// ...but an ordinary update from an in-roster peer does not...
assert!(!announce_subject_to_cap(false, false));
// ...and neither does a re-announce from a peer mid-reconnect, even
@@ -1070,7 +1147,11 @@ mod tests {
// Combined with admit_into_roster: a reconnecting author passes at a full
// roster, a brand-new one does not.
assert!(admit_into_roster(3, announce_subject_to_cap(true, true), 3));
assert!(!admit_into_roster(3, announce_subject_to_cap(true, false), 3));
assert!(!admit_into_roster(
3,
announce_subject_to_cap(true, false),
3
));
}
#[test]
@@ -1105,7 +1186,10 @@ mod tests {
assert_eq!(monitor.observe(author, -122_000, 40_000), None);
assert_eq!(
monitor.observe(author, -123_000, 69_999),
Some(ClockSkewWarning { author, skew_ms: -123_000 })
Some(ClockSkewWarning {
author,
skew_ms: -123_000
})
);
assert_eq!(monitor.observe(author, -124_000, 70_000), None);
}
@@ -1125,7 +1209,10 @@ mod tests {
assert_eq!(monitor.observe(author, 127_000, 319_999), None);
assert_eq!(
monitor.observe(author, 128_000, 320_000),
Some(ClockSkewWarning { author, skew_ms: 128_000 })
Some(ClockSkewWarning {
author,
skew_ms: 128_000
})
);
}
@@ -1141,11 +1228,17 @@ mod tests {
assert_eq!(monitor.observe(b, 121_000, 1_000), None);
assert_eq!(
monitor.observe(b, 121_000, 2_000),
Some(ClockSkewWarning { author: b, skew_ms: 121_000 })
Some(ClockSkewWarning {
author: b,
skew_ms: 121_000
})
);
assert_eq!(
monitor.observe(a, -121_000, 2_000),
Some(ClockSkewWarning { author: a, skew_ms: -121_000 })
Some(ClockSkewWarning {
author: a,
skew_ms: -121_000
})
);
}
@@ -1202,7 +1295,10 @@ mod tests {
assert!(long.as_str().len() > MAX_RELAY_URL_LEN);
let addr = EndpointAddr::from_parts(
id,
[TransportAddr::Relay(short.clone()), TransportAddr::Relay(long)],
[
TransportAddr::Relay(short.clone()),
TransportAddr::Relay(long),
],
);
let out = sanitize_endpoint_addr(&addr);
let relays: Vec<_> = out.relay_urls().cloned().collect();
@@ -1222,7 +1318,12 @@ mod tests {
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 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();
@@ -1252,7 +1353,13 @@ mod tests {
};
let serialized = serde_json::to_string(&original).unwrap();
let deserialized: GossipMessage = serde_json::from_str(&serialized).unwrap();
if let GossipMessage::Chat { name, text, ts, attachment } = deserialized {
if let GossipMessage::Chat {
name,
text,
ts,
attachment,
} = deserialized
{
assert_eq!(name, "Alice");
assert_eq!(text, "Hello");
assert_eq!(ts, 123456789);
@@ -1305,7 +1412,10 @@ mod tests {
// defaulting the attachment to None (serde(default)).
let legacy = r#"{"Chat":{"name":"Old","text":"hi","ts":7}}"#;
let parsed: GossipMessage = serde_json::from_str(legacy).unwrap();
if let GossipMessage::Chat { name, attachment, .. } = parsed {
if let GossipMessage::Chat {
name, attachment, ..
} = parsed
{
assert_eq!(name, "Old");
assert_eq!(attachment, None);
} else {
@@ -1368,7 +1478,10 @@ mod tests {
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(()));
assert_eq!(
verify_gossip(&p, &topic, 5_000, GOSSIP_FRESHNESS_MS),
Ok(())
);
}
#[test]
@@ -1392,7 +1505,12 @@ mod tests {
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, attachment: None };
p.msg = GossipMessage::Chat {
name: "x".into(),
text: "y".into(),
ts: 5_000,
attachment: None,
};
assert_eq!(
verify_gossip(&p, &topic, 5_000, GOSSIP_FRESHNESS_MS),
Err(GossipReject::BadSignature)
@@ -1418,16 +1536,34 @@ mod tests {
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),
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),
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());
assert!(
verify_gossip(
&p,
&topic,
1_000_000 + GOSSIP_FRESHNESS_MS - 1,
GOSSIP_FRESHNESS_MS
)
.is_ok()
);
}
#[test]
@@ -1449,10 +1585,30 @@ mod tests {
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));
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));
@@ -1465,13 +1621,26 @@ mod tests {
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, attachment: None };
let earlier_chat = GossipMessage::Chat { name: "A".into(), text: "earlier".into(), ts: 100, attachment: None };
let later_chat = GossipMessage::Chat {
name: "A".into(),
text: "later".into(),
ts: 200,
attachment: None,
};
let earlier_chat = GossipMessage::Chat {
name: "A".into(),
text: "earlier".into(),
ts: 100,
attachment: None,
};
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");
assert!(
seen.is_empty(),
"chat must not populate the state-mutation replay map"
);
}
#[test]
@@ -1481,8 +1650,18 @@ mod tests {
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,
&GossipMessage::Leave,
5
));
assert!(admit_state_mutation(&mut seen, author, &announce, 5));
assert!(admit_state_mutation(&mut seen, other, &GossipMessage::Leave, 5));
assert!(admit_state_mutation(
&mut seen,
other,
&GossipMessage::Leave,
5
));
}
}