Merge codex-security-hardening: S10 log redaction + T1/T2/T5/T6/T7 trust-boundary fixes
This commit is contained in:
+45
-8
@@ -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::<Vec<_>>()));
|
||||
let extra_bootstrap_ids = extra_bootstrap
|
||||
.iter()
|
||||
.map(|a| crate::short_id(&a.id.to_string()))
|
||||
.collect::<Vec<_>>();
|
||||
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();
|
||||
|
||||
+118
-6
@@ -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() {
|
||||
"<redacted:empty>".to_string()
|
||||
} else {
|
||||
format!("<redacted:{}...>", short_id(value))
|
||||
}
|
||||
}
|
||||
|
||||
pub fn short_bytes_hex(bytes: &[u8]) -> String {
|
||||
bytes.iter()
|
||||
.take(6)
|
||||
.map(|b| format!("{b:02x}"))
|
||||
.collect::<Vec<_>>()
|
||||
.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<File> {
|
||||
prepare_log_file_with_limit(path, LOG_MAX_BYTES)
|
||||
}
|
||||
|
||||
fn prepare_log_file_with_limit(path: &Path, max_bytes: u64) -> std::io::Result<File> {
|
||||
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(" "), "<redacted:empty>");
|
||||
}
|
||||
|
||||
#[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);
|
||||
}
|
||||
}
|
||||
|
||||
+145
-7
@@ -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<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
|
||||
@@ -191,11 +241,21 @@ impl RoomState for IrohGossipState {
|
||||
self_state: PeerState,
|
||||
extra_bootstrap: Vec<EndpointAddr>,
|
||||
) -> 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::<PeerSpeakTicket>()?;
|
||||
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));
|
||||
}
|
||||
}
|
||||
|
||||
+40
-21
@@ -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<FriendPresence> {
|
||||
/// 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<FriendPresence> {
|
||||
match msg {
|
||||
ControlMsg::Ping => None,
|
||||
ControlMsg::Pong { room: None } => Some(FriendPresence::Online),
|
||||
ControlMsg::Pong { room: Some(r) } => {
|
||||
if r.ticket.parse::<crate::network::PeerSpeakTicket>().is_ok() {
|
||||
Some(FriendPresence::InRoom {
|
||||
name: crate::sanitize::sanitize_name(&r.name),
|
||||
ticket: r.ticket.clone(),
|
||||
})
|
||||
} else {
|
||||
let Ok(ticket) = r.ticket.parse::<crate::network::PeerSpeakTicket>() 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");
|
||||
|
||||
+17
-15
@@ -49,16 +49,17 @@ fn decode(bytes: &[u8]) -> Result<ControlMsg> {
|
||||
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<EndpointAddr>) -> Result<ControlMsg> {
|
||||
/// 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<EndpointAddr>) -> 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<EndpointAddr>) -> 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:?}"),
|
||||
|
||||
+56
-1
@@ -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<String> {
|
||||
]
|
||||
}
|
||||
|
||||
/// 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<String> {
|
||||
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!(
|
||||
|
||||
Reference in New Issue
Block a user