diff --git a/src/core/mod.rs b/src/core/mod.rs index becde25..fc5ef44 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -8,7 +8,7 @@ use crate::codec::{AudioEncoder, opus_impl::OpusEncoder}; use crate::core::jitter::{JitterBuffer, FRAME_SAMPLES}; use crate::network::{ NetworkTransport, RoomState, PeerState, RoomEvent, ConnEvent, PeerSpeakTicket, - iroh_impl::{IrohTransport, AudioRouter}, + iroh_impl::{IrohTransport, AudioRouter, FileRouter}, gossip::IrohGossipState, }; use crate::core::messages::{CoreCommand, UiEvent}; @@ -578,6 +578,9 @@ struct NetStack { /// The persistent inbound-audio handler on `router`; per-join we bind the /// active session's transport into it, and clear it on leave. audio_router: AudioRouter, + /// The persistent chat-file-transfer handler on `router`; bound/cleared in + /// lock-step with `audio_router` (same session lifecycle). + file_router: FileRouter, /// In-memory address book (ticket + gossip fed), shared with every session. memory_lookup: iroh::address_lookup::memory::MemoryLookup, } @@ -689,6 +692,7 @@ async fn build_net_stack( .spawn(endpoint.clone()); let audio_router = AudioRouter::new(); + let file_router = FileRouter::new(); // The friends presence listener (W7 B2) rides this same persistent router as a // third ALPN — it MUST be a handler here, not a standalone accept loop, since // the router owns endpoint.accept(). Policy (who we answer / what room we @@ -696,6 +700,7 @@ async fn build_net_stack( let router = Router::builder(endpoint.clone()) .accept(iroh_gossip::net::GOSSIP_ALPN, gossip.clone()) .accept(crate::protocol::AUDIO_ALPN, audio_router.clone()) + .accept(crate::protocol::FILES_ALPN, file_router.clone()) .accept( crate::presence_net::FRIENDS_ALPN, crate::presence_net::FriendsProtocol::new(friends_handler), @@ -707,6 +712,7 @@ async fn build_net_stack( gossip, router, audio_router, + file_router, memory_lookup, }) } @@ -1071,6 +1077,7 @@ async fn run_core_loop( if let Some(session) = active_session.take() { session.shutdown(audio_backend.clone()).await; net.audio_router.clear(); + net.file_router.clear(); } *current_room.lock().unwrap() = None; @@ -1093,6 +1100,7 @@ async fn run_core_loop( crate::log_msg("Shutting down existing active session"); session.shutdown(audio_backend.clone()).await; net.audio_router.clear(); + net.file_router.clear(); } // If a network-mode / identity change was deferred while a call was @@ -1149,6 +1157,7 @@ async fn run_core_loop( // NetStack; the session just subscribes its topic below. let transport = Arc::new(IrohTransport::new(endpoint.clone())); net.audio_router.bind(&transport); + net.file_router.bind(&transport); let room_state = Arc::new(IrohGossipState::new( endpoint.clone(), @@ -1193,6 +1202,7 @@ async fn run_core_loop( crate::log_msg(&format!("Error room_state.join failed: {:?}", e)); let _ = ui_tx.send(UiEvent::Error(format!("Failed to join room: {}", e))).await; net.audio_router.clear(); + net.file_router.clear(); continue; } crate::log_msg("Joined room successfully via room_state"); @@ -1245,6 +1255,7 @@ async fn run_core_loop( let _ = ui_tx.send(UiEvent::Error(format!("Failed to start capture: {}", e))).await; let _ = room_state.leave().await; net.audio_router.clear(); + net.file_router.clear(); continue; } @@ -1257,6 +1268,7 @@ async fn run_core_loop( let _ = audio_backend.stop(); let _ = room_state.leave().await; net.audio_router.clear(); + net.file_router.clear(); continue; } @@ -1829,6 +1841,7 @@ async fn run_core_loop( session.shutdown(audio_backend.clone()).await; // Stop routing inbound audio links — the endpoint/router stay up. net.audio_router.clear(); + net.file_router.clear(); // No longer in a gathering — friends who ping see us as just online. *current_room.lock().unwrap() = None; let _ = ui_tx.send(UiEvent::RoomLeft).await; diff --git a/src/network/iroh_impl.rs b/src/network/iroh_impl.rs index 80dc871..ba7e273 100644 --- a/src/network/iroh_impl.rs +++ b/src/network/iroh_impl.rs @@ -9,7 +9,8 @@ use std::collections::{HashMap, HashSet}; use std::time::Duration; use async_trait::async_trait; -use crate::protocol::AUDIO_ALPN; +use crate::protocol::{AUDIO_ALPN, FILES_ALPN}; +use crate::files::{AttachmentId, ChatAttachment}; /// Per-peer datagram send queue depth. Audio is real-time, so a backlog is /// useless latency — keep it shallow and drop the oldest frame when full. @@ -31,6 +32,10 @@ const MAX_BACKOFF: Duration = Duration::from_secs(5); /// already means "the peer closed this on purpose." const GOODBYE_CODE: u32 = 1; +/// Bound on each phase (connect, read) of a chat-attachment fetch, so a slow or +/// stalled sender can't hang the fetch indefinitely. +const FILE_FETCH_TIMEOUT: Duration = Duration::from_secs(30); + /// State shared between the transport, its protocol handler, and every per-peer /// supervisor task. One supervisor owns a peer's whole connection lifecycle. struct Shared { @@ -60,6 +65,11 @@ struct Shared { /// verified gossip roster plus peers still inside reconnect grace; transport /// connections alone never mutate this set. admitted_audio: StdMutex>, + /// Chat file attachments we're serving to room members this session, keyed by + /// the random attachment id. Populated when we send a chat file; read by the + /// file protocol handler to answer a member's fetch. Cleared on leave. Each + /// blob is already byte-capped at send time. + served_files: StdMutex>>>, incoming_tx: mpsc::Sender<(EndpointId, Bytes)>, /// Best-effort link-state notifications for the UI (connecting / connected). conn_events_tx: mpsc::Sender, @@ -402,6 +412,88 @@ impl iroh::protocol::ProtocolHandler for AudioRouter { } } +/// Protocol handler for the file-transfer plane (`FILES_ALPN`). Mirrors +/// [`AudioRouter`]: it's persistent on the router and bound to the active +/// session's [`Shared`] on join. On an inbound stream it authenticates the peer +/// (iroh ALPN handshake gives us `remote_id`), gates on **live room membership** +/// (same invariant as audio admission, so a former member can't pull files), +/// reads a single 32-byte attachment id, and streams back the matching blob from +/// the session serve store — or nothing if the id is unknown. +#[derive(Clone, Default)] +pub struct FileRouter { + current: Arc>>>, +} + +impl std::fmt::Debug for FileRouter { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("FileRouter").finish_non_exhaustive() + } +} + +impl FileRouter { + pub fn new() -> Self { + Self::default() + } + + /// Route inbound file fetches to `transport`'s session (called on join). + pub fn bind(&self, transport: &IrohTransport) { + *self.current.lock().unwrap() = Some(transport.shared.clone()); + } + + /// Stop serving files until the next [`bind`](Self::bind) (called on leave). + pub fn clear(&self) { + *self.current.lock().unwrap() = None; + } +} + +/// Max bytes we'll read for a fetch *request* frame. A request is exactly one +/// 32-byte id; this small ceiling rejects a peer trying to stream us a huge +/// "request" as a cheap DoS. +const FILE_REQUEST_MAX: usize = 64; + +impl iroh::protocol::ProtocolHandler for FileRouter { + fn accept( + &self, + connection: Connection, + ) -> impl std::future::Future> + Send { + let peer_id = connection.remote_id(); + let shared = self.current.lock().unwrap().clone(); + async move { + // No active call → nothing to serve. + let Some(shared) = shared else { + return Ok(()); + }; + // Member gating: only current room members may fetch our files. Reuses + // the audio admission roster (the authoritative room membership set). + if !shared.audio_sender_admitted(peer_id) { + crate::log_msg(&format!( + "Transport: rejected file fetch from non-member {}", + crate::short_id(&peer_id.to_string()) + )); + return Ok(()); + } + // Accept one bidirectional stream: read the id, write the bytes. + let Ok((mut send, mut recv)) = connection.accept_bi().await else { + return Ok(()); + }; + let Ok(req) = recv.read_to_end(FILE_REQUEST_MAX).await else { + return Ok(()); + }; + let Some(id) = crate::files::parse_request(&req) else { + return Ok(()); + }; + let blob = shared.served_files.lock().unwrap().get(&id).cloned(); + if let Some(blob) = blob { + let _ = send.write_all(&blob).await; + } + // Finish either way: an unknown id closes with an empty body, which + // the fetcher reads as a zero-length result and treats as "gone". + let _ = send.finish(); + Ok(()) + } + } +} + pub struct IrohTransport { shared: Arc, incoming_rx: tokio::sync::Mutex>>, @@ -426,6 +518,7 @@ impl IrohTransport { peers: tokio::sync::Mutex::new(HashMap::new()), live_conns: StdMutex::new(HashMap::new()), admitted_audio: StdMutex::new(HashSet::new()), + served_files: StdMutex::new(HashMap::new()), incoming_tx, conn_events_tx, }); @@ -455,6 +548,7 @@ impl IrohTransport { self.shared.senders.lock().unwrap().clear(); self.shared.addrs.lock().unwrap().clear(); self.shared.admitted_audio.lock().unwrap().clear(); + self.shared.served_files.lock().unwrap().clear(); // Give the CONNECTION_CLOSE frames a moment to flush before the caller // shuts the endpoint/router down (the `conns` clones are still alive // here, so the endpoint can still transmit them). @@ -483,6 +577,59 @@ impl IrohTransport { pub fn audio_sender_admitted(&self, peer_id: EndpointId) -> bool { self.shared.audio_sender_admitted(peer_id) } + + /// Make `bytes` available to room members under `id` for the rest of this + /// session (served by the [`FileRouter`] handler). Called by core when we + /// send a chat file. The blob is cleared on leave. + pub fn serve_attachment(&self, id: AttachmentId, bytes: Arc>) { + self.shared.served_files.lock().unwrap().insert(id, bytes); + } + + /// Fetch a chat attachment's bytes from its sender over the file plane. Dials + /// the sender on `FILES_ALPN` (preferring a known full address), writes the + /// 32-byte id, and reads the response bounded by the descriptor's declared + /// size (which the caller has already validated against the global cap). The + /// read limit means a malicious sender can't stream us more than advertised. + pub async fn fetch_attachment( + &self, + from: EndpointId, + att: &ChatAttachment, + ) -> Result, NetError> { + if !crate::files::size_within_cap(att.size) { + return Err(NetError::Other("attachment size out of range".to_string())); + } + let addr = self.shared.addrs.lock().unwrap().get(&from).cloned(); + let connect = async { + match addr { + Some(addr) => self.shared.endpoint.connect(addr, FILES_ALPN).await, + None => self.shared.endpoint.connect(from, FILES_ALPN).await, + } + }; + let conn = tokio::time::timeout(FILE_FETCH_TIMEOUT, connect) + .await + .map_err(|_| NetError::Other("file fetch: connect timed out".to_string()))? + .map_err(|e| NetError::Other(format!("file fetch: connect failed: {e}")))?; + + let (mut send, mut recv) = conn + .open_bi() + .await + .map_err(|e| NetError::Other(format!("file fetch: open stream failed: {e}")))?; + send.write_all(&att.id) + .await + .map_err(|e| NetError::Other(format!("file fetch: request write failed: {e}")))?; + send.finish() + .map_err(|e| NetError::Other(format!("file fetch: request finish failed: {e}")))?; + + let read = recv.read_to_end(att.size as usize); + let bytes = tokio::time::timeout(FILE_FETCH_TIMEOUT, read) + .await + .map_err(|_| NetError::Other("file fetch: read timed out".to_string()))? + .map_err(|e| NetError::Other(format!("file fetch: read failed: {e}")))?; + if bytes.is_empty() { + return Err(NetError::Other("file fetch: sender no longer has the file".to_string())); + } + Ok(bytes) + } } #[async_trait]