diff --git a/src/core/mod.rs b/src/core/mod.rs index 2bf208b..433e2b0 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -64,6 +64,15 @@ impl CoreController { /// clears from the room promptly. const RECONNECT_GRACE: Duration = Duration::from_secs(45); +/// Opus frames sent by our encoder are one 20 ms mono frame, normally far below +/// this. 4000 bytes still leaves room for large valid Opus packets (well above a +/// 48 kHz / 60 ms frame) while bounding malicious datagram copy/decode churn. +const MAX_OPUS_PAYLOAD: usize = 4000; + +fn audio_datagram_len_ok(len: usize) -> bool { + (4..=4 + MAX_OPUS_PAYLOAD).contains(&len) +} + /// 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 /// gossip rejoin) and the conn-event task (which cancels it when the audio link @@ -634,7 +643,7 @@ async fn probe_friends_once( let ep = endpoint.clone(); set.spawn(async move { match crate::presence_net::probe(&ep, addr).await { - Ok(reply) => crate::presence::interpret_pong(&reply).map(|p| (id, p)), + Ok((from, reply)) => crate::presence::interpret_pong(&reply, from).map(|p| (id, p)), Err(_) => None, } }); @@ -923,7 +932,12 @@ async fn run_core_loop( let ticket_str = if ticket.trim().is_empty() || ticket == "create" { let topic_id: [u8; 32] = rand::random(); let host_addr = endpoint.addr(); - crate::log_msg(&format!("Creating room. host_addr={:?}, topic_id={:?}", host_addr, topic_id)); + crate::log_msg(&format!( + "Creating room. host_id={}, host_addrs={}, topic={}", + crate::short_id(&host_addr.id.to_string()), + host_addr.addrs.len(), + crate::short_bytes_hex(&topic_id) + )); // The creator's chosen cosmetic label rides in the ticket so // every joiner inherits it; sanitize it before it leaves here. let label = crate::sanitize::sanitize_name(&room_name); @@ -931,7 +945,10 @@ async fn run_core_loop( ticket.to_string() } else { let ticket_str = ticket.trim().to_string(); - crate::log_msg(&format!("Joining room with existing ticket={}", ticket_str)); + crate::log_msg(&format!( + "Joining room with existing ticket={}", + crate::redact_for_log(&ticket_str) + )); ticket_str }; @@ -970,7 +987,17 @@ async fn run_core_loop( .map(|peers| peers.values().cloned().collect()) .unwrap_or_default(); - crate::log_msg(&format!("Attempting room_state.join with self_state={:?}, extra_bootstrap={:?}", self_state, extra_bootstrap.iter().map(|a| a.id).collect::>())); + let extra_bootstrap_ids = extra_bootstrap + .iter() + .map(|a| crate::short_id(&a.id.to_string())) + .collect::>(); + crate::log_msg(&format!( + "Attempting room_state.join self_id={}, self_name={:?}, sharing={}, extra_bootstrap={:?}", + crate::short_id(&self_state.addr.id.to_string()), + self_state.name, + self_state.sharing.is_some(), + extra_bootstrap_ids + )); if let Err(e) = room_state.join(&ticket_str, self_state.clone(), extra_bootstrap).await { crate::log_msg(&format!("Error room_state.join failed: {:?}", e)); let _ = ui_tx.send(UiEvent::Error(format!("Failed to join room: {}", e))).await; @@ -1135,8 +1162,9 @@ async fn run_core_loop( }; while let Some((from_peer, bytes)) = datagram_rx.recv().await { - if bytes.len() < 4 { - continue; // malformed: missing sequence header + if !audio_datagram_len_ok(bytes.len()) { + // Malformed (< sequence header) or oversized Opus payload. + continue; } let seq = u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]); let payload = bytes[4..].to_vec(); @@ -1987,8 +2015,8 @@ async fn run_core_loop( #[cfg(test)] mod tests { use super::{ - apply_volume, frame_level, mix_frames, mix_stereo_frames, stereo_to_mono, MicLevelMeter, - MIC_LEVEL_REPORT_SAMPLES, + apply_volume, audio_datagram_len_ok, frame_level, mix_frames, mix_stereo_frames, + stereo_to_mono, MicLevelMeter, MAX_OPUS_PAYLOAD, MIC_LEVEL_REPORT_SAMPLES, }; /// A frame of constant amplitude with the given sample count. @@ -2005,6 +2033,15 @@ mod tests { assert!(m.push(&frame(1000, MIC_LEVEL_REPORT_SAMPLES)).is_some()); } + #[test] + fn audio_datagram_length_gate_preserves_header_and_caps_payload() { + assert!(!audio_datagram_len_ok(0)); + assert!(!audio_datagram_len_ok(3)); + assert!(audio_datagram_len_ok(4)); + assert!(audio_datagram_len_ok(4 + MAX_OPUS_PAYLOAD)); + assert!(!audio_datagram_len_ok(5 + MAX_OPUS_PAYLOAD)); + } + #[test] fn mic_meter_holds_the_peak_across_the_window() { let mut m = MicLevelMeter::new(); diff --git a/src/lib.rs b/src/lib.rs index 877fdbd..64474ec 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -18,9 +18,13 @@ pub mod recents; pub mod discovery; pub mod hotkeys; -use std::path::PathBuf; +use std::fs::File; +use std::path::{Path, PathBuf}; use std::sync::OnceLock; +const LOG_MAX_BYTES: u64 = 5 * 1024 * 1024; +const LOG_MODE: u32 = 0o600; + /// Resolves the log file path once: `$XDG_STATE_HOME/peerspeak/peerspeak.log` /// (via `dirs::state_dir`), falling back to the system temp dir. Computed lazily /// so we never hardcode a per-user path. @@ -43,6 +47,65 @@ pub fn log_file_path() -> PathBuf { log_path().clone() } +/// Short, human-matchable id prefix for diagnostics. Never use this where the +/// full value is needed for protocol behavior. +pub fn short_id(id: &str) -> String { + id.chars().take(8).collect() +} + +/// Redact a capability-bearing value for logs while keeping a tiny prefix for +/// support correlation. Tickets and endpoint addresses are bearer capabilities: +/// logging the full string is equivalent to leaking the room/share. +pub fn redact_for_log(value: &str) -> String { + let value = value.trim(); + if value.is_empty() { + "".to_string() + } else { + format!("", short_id(value)) + } +} + +pub fn short_bytes_hex(bytes: &[u8]) -> String { + bytes.iter() + .take(6) + .map(|b| format!("{b:02x}")) + .collect::>() + .join("") +} + +fn rotated_log_path(path: &Path) -> PathBuf { + let file_name = path.file_name().and_then(|n| n.to_str()).unwrap_or("peerspeak.log"); + path.with_file_name(format!("{file_name}.1")) +} + +fn prepare_log_file(path: &Path) -> std::io::Result { + prepare_log_file_with_limit(path, LOG_MAX_BYTES) +} + +fn prepare_log_file_with_limit(path: &Path, max_bytes: u64) -> std::io::Result { + use std::os::unix::fs::{OpenOptionsExt, PermissionsExt}; + + if let Some(parent) = path.parent() { + let _ = std::fs::create_dir_all(parent); + } + + if std::fs::metadata(path).is_ok_and(|m| m.len() > max_bytes) { + let rotated = rotated_log_path(path); + let _ = std::fs::remove_file(&rotated); + if std::fs::rename(path, &rotated).is_err() { + let _ = std::fs::OpenOptions::new().write(true).truncate(true).open(path); + } + } + + let file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .mode(LOG_MODE) + .open(path)?; + let _ = std::fs::set_permissions(path, std::fs::Permissions::from_mode(LOG_MODE)); + Ok(file) +} + pub fn log_msg(msg: &str) { // Format the whole line into one buffer first, then emit it with a single // `write_all`. The file is opened with `O_APPEND`, so a lone `write()` is @@ -52,12 +115,61 @@ pub fn log_msg(msg: &str) { Ok(time) => format!("[{}.{:03}] {}\n", time.as_secs(), time.subsec_millis(), msg), Err(_) => format!("{}\n", msg), }; - if let Ok(mut file) = std::fs::OpenOptions::new() - .create(true) - .append(true) - .open(log_path()) - { + if let Ok(mut file) = prepare_log_file(log_path()) { use std::io::Write; let _ = file.write_all(line.as_bytes()); } } + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write; + use std::os::unix::fs::PermissionsExt; + + fn temp_log_dir() -> PathBuf { + let stamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + std::env::temp_dir().join(format!("peerspeak-log-test-{}-{stamp}", std::process::id())) + } + + #[test] + fn redaction_keeps_only_a_short_prefix() { + let secret = "abcdefghijklmnopqrstuvwxyz"; + let redacted = redact_for_log(secret); + assert!(redacted.contains("abcdefgh")); + assert!(!redacted.contains("ijklmnopqrstuvwxyz")); + assert_eq!(redact_for_log(" "), ""); + } + + #[test] + fn log_file_is_created_private() { + let dir = temp_log_dir(); + let path = dir.join("peerspeak.log"); + let _file = prepare_log_file(&path).unwrap(); + + let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777; + assert_eq!(mode, LOG_MODE); + let _ = std::fs::remove_dir_all(dir); + } + + #[test] + fn oversized_log_is_rotated_on_open() { + let dir = temp_log_dir(); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("peerspeak.log"); + { + let mut file = std::fs::File::create(&path).unwrap(); + file.write_all(b"oversized").unwrap(); + } + + let _file = prepare_log_file_with_limit(&path, 4).unwrap(); + let rotated = rotated_log_path(&path); + + assert_eq!(std::fs::read_to_string(rotated).unwrap(), "oversized"); + assert_eq!(std::fs::metadata(&path).unwrap().len(), 0); + let _ = std::fs::remove_dir_all(dir); + } +} diff --git a/src/network/gossip.rs b/src/network/gossip.rs index c233e20..6811970 100644 --- a/src/network/gossip.rs +++ b/src/network/gossip.rs @@ -43,7 +43,7 @@ impl std::fmt::Debug for GossipPayload { f.debug_struct("GossipPayload") .field("author", &self.author) .field("ts", &self.ts) - .field("msg", &self.msg) + .field("msg_kind", &gossip_message_kind(&self.msg)) .finish_non_exhaustive() } } @@ -83,6 +83,56 @@ enum GossipReject { 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 { + 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 @@ -191,11 +241,21 @@ impl RoomState for IrohGossipState { self_state: PeerState, extra_bootstrap: Vec, ) -> Result<(), NetError> { - crate::log_msg(&format!("RoomState::join: self_id={:?}, self_name={:?}, ticket={}", self_state.addr.id, self_state.name, ticket_str)); + 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::()?; let topic_id = TopicId::from_bytes(ticket.topic_id); - crate::log_msg(&format!("Parsed ticket. host_id={:?}, host_addrs={:?}, topic={:?}", ticket.host_addr.id, ticket.host_addr.addrs, 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; @@ -242,6 +302,7 @@ impl RoomState for IrohGossipState { 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 = { @@ -291,7 +352,26 @@ impl RoomState for IrohGossipState { continue; } - crate::log_msg(&format!("Gossip Event::Received from author={:?}, payload={:?}", payload.author, payload.msg)); + 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) => { @@ -305,6 +385,10 @@ impl RoomState for IrohGossipState { // 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); @@ -316,11 +400,19 @@ impl RoomState for IrohGossipState { }; if is_new { - crate::log_msg(&format!("Gossip new peer joined: {:?}, state: {:?}", payload.author, state)); + 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: {:?}", payload.author, state)); + 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; } } @@ -396,7 +488,10 @@ impl RoomState for IrohGossipState { } async fn update_self_state(&self, self_state: PeerState) -> Result<(), NetError> { - crate::log_msg(&format!("RoomState::update_self_state: state={:?}", self_state)); + 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(); @@ -495,6 +590,7 @@ 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); @@ -750,4 +846,46 @@ mod tests { 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)); + } } diff --git a/src/presence.rs b/src/presence.rs index 28b8251..69ec24c 100644 --- a/src/presence.rs +++ b/src/presence.rs @@ -110,26 +110,32 @@ pub enum FriendPresence { InRoom { name: String, ticket: String }, } -/// Interpret a peer's reply defensively. Only a `Pong` is a reply (a `Ping` is -/// not, so it yields `None`). When the peer reports a room, we **sanitize the -/// peer-supplied name** and **only surface it as joinable if the ticket actually -/// parses** as a [`crate::network::PeerSpeakTicket`] — a garbage or hostile -/// ticket downgrades the friend to plain `Online` rather than offering a dead / -/// dangerous Join button. (We still never auto-join; the user clicks.) -pub fn interpret_pong(msg: &ControlMsg) -> Option { +/// Interpret a peer's reply defensively. `from` must be the connection's +/// authenticated remote id, not any value carried in the payload. Only a `Pong` +/// is a reply (a `Ping` is not, so it yields `None`). When the peer reports a +/// room, we **sanitize the peer-supplied name** and **only surface it as joinable +/// if the ticket actually parses** as a [`crate::network::PeerSpeakTicket`] and +/// points back at the replying friend. A garbage/redirect ticket downgrades the +/// friend to plain `Online` rather than offering a dead or attacker-controlled +/// Join button. (We still never auto-join; the user clicks.) +pub fn interpret_pong(msg: &ControlMsg, from: EndpointId) -> Option { match msg { ControlMsg::Ping => None, ControlMsg::Pong { room: None } => Some(FriendPresence::Online), ControlMsg::Pong { room: Some(r) } => { - if r.ticket.parse::().is_ok() { - Some(FriendPresence::InRoom { - name: crate::sanitize::sanitize_name(&r.name), - ticket: r.ticket.clone(), - }) - } else { + let Ok(ticket) = r.ticket.parse::() else { // Online, but the advertised room is unusable — don't offer Join. - Some(FriendPresence::Online) + return Some(FriendPresence::Online); + }; + if ticket.host_addr.id != from { + // Online, but the advertised room redirects away from the friend + // who authenticated this Pong — don't offer a phishing Join. + return Some(FriendPresence::Online); } + Some(FriendPresence::InRoom { + name: crate::sanitize::sanitize_name(&r.name), + ticket: r.ticket.clone(), + }) } } } @@ -206,21 +212,22 @@ mod tests { #[test] fn interpret_ping_is_not_a_reply() { - assert_eq!(interpret_pong(&ControlMsg::Ping), None); + assert_eq!(interpret_pong(&ControlMsg::Ping, id()), None); } #[test] fn interpret_pong_online_and_inroom() { + let friend = id(); // No room -> Online. assert_eq!( - interpret_pong(&ControlMsg::Pong { room: None }), + interpret_pong(&ControlMsg::Pong { room: None }, friend), Some(FriendPresence::Online) ); // Valid ticket -> InRoom with a sanitized name. - let t = valid_ticket(id()); + let t = valid_ticket(friend); let got = interpret_pong(&ControlMsg::Pong { room: Some(RoomPresence { name: "HangOut".into(), ticket: t.clone() }), - }); + }, friend); assert_eq!(got, Some(FriendPresence::InRoom { name: "HangOut".into(), ticket: t })); } @@ -230,17 +237,29 @@ mod tests { // Online — no dead/hostile Join button is surfaced. let got = interpret_pong(&ControlMsg::Pong { room: Some(RoomPresence { name: "Trap".into(), ticket: "not-a-ticket".into() }), - }); + }, id()); + assert_eq!(got, Some(FriendPresence::Online)); + } + + #[test] + fn interpret_pong_rejects_ticket_for_a_different_host() { + let friend = id(); + let attacker = id(); + let t = valid_ticket(attacker); + let got = interpret_pong(&ControlMsg::Pong { + room: Some(RoomPresence { name: "Redirect".into(), ticket: t }), + }, friend); assert_eq!(got, Some(FriendPresence::Online)); } #[test] fn interpret_pong_sanitizes_a_hostile_room_name() { // Control/bidi characters in a peer-supplied name are stripped. - let t = valid_ticket(id()); + let friend = id(); + let t = valid_ticket(friend); let got = interpret_pong(&ControlMsg::Pong { room: Some(RoomPresence { name: "Hang\u{202e}Out\u{0007}".into(), ticket: t.clone() }), - }); + }, friend); match got { Some(FriendPresence::InRoom { name, .. }) => { assert!(!name.contains('\u{202e}'), "bidi override must be stripped"); diff --git a/src/presence_net.rs b/src/presence_net.rs index c383c3f..c30f8d8 100644 --- a/src/presence_net.rs +++ b/src/presence_net.rs @@ -49,16 +49,17 @@ fn decode(bytes: &[u8]) -> Result { serde_json::from_slice(bytes).context("failed to decode control message") } -/// Probe `peer` for presence: send a `Ping`, return their `Pong`. An error means -/// no usable reply (offline / unreachable / refused / malformed) — the caller -/// treats that as "appears offline". `peer` is usually a bare [`EndpointId`] -/// (friends store the stable id); a full [`EndpointAddr`] is also accepted (and -/// used by hermetic tests). -pub async fn probe(endpoint: &Endpoint, peer: impl Into) -> Result { +/// Probe `peer` for presence: send a `Ping`, return their authenticated id and +/// `Pong`. An error means no usable reply (offline / unreachable / refused / +/// malformed) — the caller treats that as "appears offline". `peer` is usually a +/// bare [`EndpointId`] (friends store the stable id); a full [`EndpointAddr`] is +/// also accepted (and used by hermetic tests). +pub async fn probe(endpoint: &Endpoint, peer: impl Into) -> Result<(EndpointId, ControlMsg)> { let conn = tokio::time::timeout(IO_TIMEOUT, endpoint.connect(peer, FRIENDS_ALPN)) .await .context("timed out connecting to peer")? .context("failed to connect to peer")?; + let from = conn.remote_id(); let io = async { let (mut send, mut recv) = conn.open_bi().await.context("failed to open control stream")?; @@ -77,7 +78,7 @@ pub async fn probe(endpoint: &Endpoint, peer: impl Into) -> Result .await .context("timed out awaiting pong")?; conn.close(VarInt::from_u32(0), b"done"); - result + result.map(|msg| (from, msg)) } /// A reply policy: given the *authenticated* remote id, decide whether and how to @@ -110,6 +111,10 @@ async fn handle(incoming: Incoming, handler: Handler) -> Result<()> { async fn exchange(conn: &iroh::endpoint::Connection, handler: &Handler) -> Result<()> { // The authenticated remote id — NOT anything the peer puts in the payload. let from = conn.remote_id(); + let Some(reply) = handler(from) else { + conn.close(VarInt::from_u32(0), b"not authorized"); + return Ok(()); + }; let io = async { let (mut send, mut recv) = conn.accept_bi().await.context("failed to accept stream")?; @@ -118,13 +123,9 @@ async fn exchange(conn: &iroh::endpoint::Connection, handler: &Handler) -> Resul ControlMsg::Ping => {} other => bail!("expected a ping, got {other:?}"), } - // Ask the policy what to send. None -> answer nothing (stranger / invisible): - // finish the stream with no bytes so the prober sees an empty (unusable) reply. - if let Some(reply) = handler(from) { - send.write_all(&encode(&reply)?) - .await - .context("failed to write pong")?; - } + send.write_all(&encode(&reply)?) + .await + .context("failed to write pong")?; send.finish().context("failed to finish reply stream")?; Ok::<_, anyhow::Error>(()) }; @@ -220,10 +221,11 @@ mod tests { let serve_task = tokio::spawn(async move { serve(server_ep, handler).await }); // The allowed prober gets a Pong with the room. - let pong = tokio::time::timeout(Duration::from_secs(15), probe(&prober, server_addr.clone())) + let (from, pong) = tokio::time::timeout(Duration::from_secs(15), probe(&prober, server_addr.clone())) .await .expect("probe timed out") .expect("probe failed"); + assert_eq!(from, server_addr.id); match pong { ControlMsg::Pong { room: Some(r) } => assert_eq!(r.name, "HangOut"), other => panic!("expected Pong with a room, got {other:?}"), diff --git a/src/screenshare/mod.rs b/src/screenshare/mod.rs index 96db489..2029220 100644 --- a/src/screenshare/mod.rs +++ b/src/screenshare/mod.rs @@ -25,6 +25,10 @@ use tokio::process::{Child, Command}; /// points elsewhere. const PIXELPASS_BIN: &str = "pixelpass"; +/// Pixelpass endpoint tickets are normally ~140 chars. Leave headroom for format +/// growth, but reject unbounded gossip payloads before the UI offers "Watch". +const MAX_TICKET_LEN: usize = 512; + /// How long to wait for the host to emit its ticket / the viewer to connect /// before giving up and killing the child. Startup is normally sub-second; this /// is only a safety net so a hung pixelpass can't wedge the caller forever. @@ -108,6 +112,19 @@ pub fn viewer_args(ticket: &str) -> Vec { ] } +/// Sanitize a peer-advertised pixelpass ticket at the gossip boundary. Peerspeak +/// intentionally does not depend on pixelpass/iroh-tickets, so this validates the +/// stable CLI ticket envelope we consume: bounded ASCII endpoint tickets beginning +/// with `endpoint`. Invalid input becomes `None`, which removes the Watch button. +pub fn sanitize_ticket(ticket: String) -> Option { + let ticket = ticket.trim(); + let valid_len = !ticket.is_empty() && ticket.len() <= MAX_TICKET_LEN; + let valid_shape = ticket.starts_with("endpoint") + && ticket.len() > "endpoint".len() + && ticket.bytes().all(|b| b.is_ascii_alphanumeric()); + (valid_len && valid_shape).then(|| ticket.to_string()) +} + /// Resolve the pixelpass binary: an explicit config override (used only if it /// points at an existing file), otherwise the first `pixelpass` found on /// `$PATH`. `None` means it isn't installed — a normal, handled state. An @@ -267,12 +284,29 @@ where tokio::spawn(async move { while let Ok(Some(line)) = lines.next_line().await { if let Some(ev) = parse_pixelpass_event(&line) { - crate::log_msg(&format!("pixelpass {role}: {ev:?}")); + crate::log_msg(&format!("pixelpass {role}: {}", event_for_log(&ev))); } } }); } +fn event_for_log(ev: &PixelpassEvent) -> String { + match ev { + PixelpassEvent::Ticket(ticket) => format!("ticket {}", crate::redact_for_log(ticket)), + PixelpassEvent::Connected(_) => "connected".to_string(), + PixelpassEvent::ViewerJoined { active, max } => { + format!("viewer_joined active={active} max={max}") + } + PixelpassEvent::ViewerLeft { active, max } => { + format!("viewer_left active={active} max={max}") + } + PixelpassEvent::Refused(reason) => format!("viewer_refused reason={reason:?}"), + PixelpassEvent::CaptureStarted => "capture_started".to_string(), + PixelpassEvent::CaptureStopped => "capture_stopped".to_string(), + PixelpassEvent::Other => "other".to_string(), + } +} + /// Open the viewer stream URL in a media player. Mirrors pixelpass's own /// low-latency mpv invocation; falls back to vlc. The player is reaped in a /// background task so it doesn't linger as a zombie when its window closes. @@ -340,6 +374,27 @@ mod tests { ); } + #[test] + fn sanitize_ticket_accepts_pixelpass_endpoint_ticket_shape() { + let ticket = "endpointaabwxjexzensznfvuudiapn5tyzws3angd2merarm"; + assert_eq!(sanitize_ticket(format!(" {ticket}\n")), Some(ticket.to_string())); + } + + #[test] + fn sanitize_ticket_rejects_oversized_or_garbage_ticket() { + assert_eq!(sanitize_ticket("not-a-ticket".into()), None); + assert_eq!(sanitize_ticket(format!("endpoint{}", "a".repeat(MAX_TICKET_LEN))), None); + assert_eq!(sanitize_ticket("endpointabc-def".into()), None); + } + + #[test] + fn event_log_redacts_ticket_values() { + let ticket = "endpointaabwxjexzensznfvuudiapn5tyzws3angd2merarm".to_string(); + let log = event_for_log(&PixelpassEvent::Ticket(ticket.clone())); + assert!(log.contains("endpoint")); + assert!(!log.contains(&ticket["endpoint".len() + 8..])); + } + #[test] fn parses_ticket() { assert_eq!(