From 9d6a5aaa98147bcb8b11b7664de3306419dd18b3 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Tue, 8 Sep 2026 01:06:00 +0200 Subject: [PATCH] fix(lib): harden the distributed leaf against hostile and slow parents Review of the leaf found the seams a stranger on the other end of a D connection could pull at, and a few that a slow one would. - A code-93 envelope is unwrapped once, as Nicotine+ does; a frame made of nothing but envelopes used to recurse once per byte. - A branch level is validated before it is used: negative levels leave the candidate unadoptable and the largest one no longer overflows. - The link reads through the crate's MessageReader, so a frame split across a read timeout is no longer torn or mistaken for a closed link. - Every event a link raises carries its link number, so a stale thread cannot adopt, close or move a newer link to the same user. - Later branch facts from the parent move our own place and are told to the server; a fresh login resets the leaf, whose stance is now the one place the four announcement messages come from. - Candidates from a later batch join the ones in flight instead of cancelling them, capped at ten; searches from the parent are budgeted at fifty a second; a reply queued for a searcher the server cannot place expires after a minute. - Truncated frames and half-read PossibleParents entries are dropped instead of yielding an empty root, a level of zero or a phantom candidate. The e2e now has a searcher of its own, so the answer must go to whoever asked and not to the parent that relayed it. --- soulseek-rs-lib/src/actor/server_actor.rs | 18 +- .../src/actor/server_actor/tests.rs | 18 +- soulseek-rs-lib/src/client/distributed.rs | 471 +++++++++++++----- soulseek-rs-lib/src/client/mod.rs | 40 +- soulseek-rs-lib/src/client/operations.rs | 58 ++- soulseek-rs-lib/src/client/tests.rs | 15 + soulseek-rs-lib/src/message/distributed.rs | 85 +++- .../src/message/server/distributed.rs | 59 ++- soulseek-rs-lib/tests/e2e.rs | 32 +- 9 files changed, 589 insertions(+), 207 deletions(-) diff --git a/soulseek-rs-lib/src/actor/server_actor.rs b/soulseek-rs-lib/src/actor/server_actor.rs index 74cdce1..ecab497 100644 --- a/soulseek-rs-lib/src/actor/server_actor.rs +++ b/soulseek-rs-lib/src/actor/server_actor.rs @@ -183,12 +183,10 @@ pub struct ServerActor { shared_folder_count: u32, shared_file_count: u32, session: SessionWatch, - /// The name we logged in as, once a login has been asked for. - username: String, } /// The messages a client sends right after a successful login: its shared-file -/// counts, its distributed-network stance, online status, and (when listening) +/// counts, online status, and (when listening) /// the port peers should connect to. Kept as a free function so it can be /// tested without a live connection. fn post_login_messages( @@ -196,7 +194,6 @@ fn post_login_messages( listen_port: u16, shared_folders: u32, shared_files: u32, - own_username: &str, ) -> Vec { let mut messages = vec![ MessageFactory::build_shared_folders_message( @@ -205,19 +202,12 @@ fn post_login_messages( ), MessageFactory::build_set_status_message(2), ]; - messages.splice(1..1, parentless_stance(own_username)); if enable_listen { messages.push(MessageFactory::build_set_wait_port_message(listen_port)); } messages } -/// What a leaf tells the server while it has no parent: it is its own branch -/// root at level 0. -fn parentless_stance(own_username: &str) -> Vec { - crate::message::distributed::stance(own_username, 0, false) -} - impl ServerActor { #[must_use] pub fn new( @@ -245,7 +235,6 @@ impl ServerActor { shared_folder_count, shared_file_count, session: SessionWatch::default(), - username: String::new(), } } @@ -579,11 +568,13 @@ impl ServerActor { self.listen_port, self.shared_folder_count, self.shared_file_count, - &self.username, ) { self.send_message(msg); } self.session.clear(); + // The distributed stance is the leaf's to announce, and a new + // session starts without a parent. + self.forward_to_client(ClientOperation::ResetDistributed); } match self.context.write_safe() { Ok(mut ctx) => ctx.logged_in = Some(message), @@ -651,7 +642,6 @@ impl ServerActor { version: ClientVersion, response: std::sync::mpsc::Sender>, ) { - self.username.clone_from(&username); if self.stream.is_none() && !self.initiate_connection() { let _ = response.send(Err(SoulseekRs::NotConnected)); return; diff --git a/soulseek-rs-lib/src/actor/server_actor/tests.rs b/soulseek-rs-lib/src/actor/server_actor/tests.rs index e19ce1f..aa99422 100644 --- a/soulseek-rs-lib/src/actor/server_actor/tests.rs +++ b/soulseek-rs-lib/src/actor/server_actor/tests.rs @@ -193,18 +193,10 @@ fn a_successful_login_marks_the_session_live_again() { #[test] fn post_login_messages_carry_counts_and_conditional_wait_port() { - let messages = post_login_messages(true, 4321, 3, 7, "me"); + let messages = post_login_messages(true, 4321, 3, 7); let codes: Vec = messages.iter().map(code_of).collect(); - // SharedFolders; the distributed stance HaveNoParent, BranchRoot, - // BranchLevel, AcceptChildren; SetStatus; SetWaitPort. - assert_eq!(codes, vec![35, 71, 127, 126, 100, 28, 2]); - // A leaf with no parent is its own branch root at level 0, and it - // declines children. - let mut root = messages[2].clone(); - root.set_pointer(4); - assert_eq!(root.read_string(), "me"); - assert_eq!(&messages[3].get_data()[4..8], &0u32.to_le_bytes()); - assert_eq!(messages[4].get_data()[4], 0); + // SharedFolders; SetStatus; SetWaitPort. + assert_eq!(codes, vec![35, 28, 2]); // The SharedFolders message (code 35) carries the real counts. let shared = messages[0].get_data(); @@ -212,7 +204,7 @@ fn post_login_messages_carry_counts_and_conditional_wait_port() { assert_eq!(u32::from_le_bytes(shared[8..12].try_into().unwrap()), 7); // Not listening omits SetWaitPort (code 2). - let no_listen = post_login_messages(false, 4321, 3, 7, "me"); + let no_listen = post_login_messages(false, 4321, 3, 7); let codes: Vec = no_listen.iter().map(code_of).collect(); - assert_eq!(codes, vec![35, 71, 127, 126, 100, 28]); + assert_eq!(codes, vec![35, 28]); } diff --git a/soulseek-rs-lib/src/client/distributed.rs b/soulseek-rs-lib/src/client/distributed.rs index 8181050..b936d40 100644 --- a/soulseek-rs-lib/src/client/distributed.rs +++ b/soulseek-rs-lib/src/client/distributed.rs @@ -3,18 +3,18 @@ //! server when that changes. use super::{Client, ClientOperation, Result, ServerMessage}; -use crate::message::Message; +use crate::message::MessageReader; use crate::message::distributed::{self, Distributed}; use crate::message::server::MessageFactory; use crate::peer::ConnectionType; use crate::utils::lock::RwLockExt; use std::collections::HashMap; -use std::io::{self, Read, Write}; +use std::io::{self, Write}; use std::net::{TcpStream, ToSocketAddrs}; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::mpsc::Sender; -use std::time::Duration; +use std::time::{Duration, Instant}; /// A candidate that has not answered the dial in this long is not a parent. const LINK_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); @@ -23,6 +23,13 @@ const CANCEL_POLL: Duration = Duration::from_secs(1); /// A distributed frame is a search or a branch fact; anything bigger is not /// one of ours. const MAX_FRAME: usize = 64 * 1024; +/// The server offers at most ten parents at a time; more candidates than +/// that in flight is a server that keeps offering, and each one is a thread. +const MAX_CANDIDATES: usize = 10; +/// Searches admitted from the parent per second. Every search costs a scan +/// of everything we share, and the tree carries the whole network's stream. +// ponytail: a flat budget; a word index over the shares is the upgrade path. +const SEARCH_BUDGET: u32 = 50; /// Where we sit in the tree, as told to the server. #[derive(Debug, Clone, PartialEq, Eq)] @@ -46,15 +53,17 @@ impl Branch { } } -#[derive(Default)] struct Candidate { - level: Option, + link: u64, + level: Option, root: Option, cancel: Arc, } -/// A candidate to dial, and the flag that tells its link to stop. +/// A candidate to dial: the link number that tags everything it sends, and +/// the flag that tells its thread to stop. pub struct Dial { + pub link: u64, pub username: String, pub host: String, pub port: u16, @@ -63,12 +72,16 @@ pub struct Dial { /// The leaf's view: at most one parent, adopted from the candidates the /// server offered, the way Nicotine+ does it. Every method is a pure state -/// change so the rule is testable without a socket. +/// change so the rule is testable without a socket. Events carry the link +/// number they came from, so a stale thread cannot speak for a newer link +/// to the same user. pub struct Leaf { own: String, parent: Option, branch: Branch, candidates: HashMap, + next_link: u64, + searches: (Instant, u32), } impl Leaf { @@ -79,6 +92,8 @@ impl Leaf { parent: None, branch: Branch::parentless(own), candidates: HashMap::new(), + next_link: 1, + searches: (Instant::now(), 0), } } @@ -87,13 +102,15 @@ impl Leaf { self.branch.clone() } + /// Whether `link` is the live link to our parent. #[must_use] - pub fn is_parent(&self, username: &str) -> bool { + pub fn is_parent(&self, username: &str, link: u64) -> bool { self.parent.as_deref() == Some(username) + && self.has_link(username, link) } - /// New candidates: dial them all, unless we already hang from a parent. - /// Candidates still being sounded out from an earlier batch are dropped. + /// New candidates to dial: those we are not already sounding out, up to + /// the cap, and none at all while we hang from a parent. pub fn consider( &mut self, candidates: Vec<(String, String, u16)>, @@ -101,50 +118,74 @@ impl Leaf { if self.parent.is_some() { return Vec::new(); } - self.cancel_candidates(); - candidates - .into_iter() - .map(|(username, host, port)| { - let cancel = Arc::new(AtomicBool::new(false)); - self.candidates.insert( - username.clone(), - Candidate { - cancel: cancel.clone(), - ..Candidate::default() - }, - ); - Dial { - username, - host, - port, - cancel, - } - }) - .collect() + let mut dials = Vec::new(); + for (username, host, port) in candidates { + if self.candidates.contains_key(&username) + || self.candidates.len() >= MAX_CANDIDATES + { + continue; + } + let link = self.next_link; + self.next_link += 1; + let cancel = Arc::new(AtomicBool::new(false)); + self.candidates.insert( + username.clone(), + Candidate { + link, + level: None, + root: None, + cancel: cancel.clone(), + }, + ); + dials.push(Dial { + link, + username, + host, + port, + cancel, + }); + } + dials } - /// A branch root does not always say so separately: level 0 is its own - /// root. - pub fn branch_level(&mut self, username: &str, level: i32) { - let Some(candidate) = self.candidates.get_mut(username) else { - return; - }; - candidate.level = Some(level); + /// A candidate said how deep it sits. A branch root does not always say + /// so separately: level 0 is its own root. A negative level is nonsense + /// and leaves the candidate unadoptable. When it is the parent talking, + /// our own place moves with it and the new stance comes back. + pub fn branch_level( + &mut self, + username: &str, + link: u64, + level: i32, + ) -> Option { + if !self.has_link(username, link) { + return None; + } + let candidate = self.candidates.get_mut(username)?; + candidate.level = u32::try_from(level).ok(); if level == 0 && candidate.root.is_none() { candidate.root = Some(username.to_string()); } + self.moved_with_parent(username) } - pub fn branch_root(&mut self, username: &str, root: &str) { - if let Some(candidate) = self.candidates.get_mut(username) { - candidate.root = Some(root.to_string()); + pub fn branch_root( + &mut self, + username: &str, + link: u64, + root: &str, + ) -> Option { + if !self.has_link(username, link) { + return None; } + self.candidates.get_mut(username)?.root = Some(root.to_string()); + self.moved_with_parent(username) } - /// A search came down from `username`. The first one from a candidate that - /// has told us its place adopts it; the stance to announce comes back. - pub fn search_from(&mut self, username: &str) -> Option { - if self.parent.is_some() { + /// A search came down from `username`. The first one from a candidate + /// that has told us its place adopts it; the stance to announce comes back. + pub fn search_from(&mut self, username: &str, link: u64) -> Option { + if self.parent.is_some() || !self.has_link(username, link) { return None; } let candidate = self.candidates.get(username)?; @@ -157,16 +198,28 @@ impl Leaf { self.cancel_candidates(); self.candidates.insert(username.to_string(), parent); self.parent = Some(username.to_string()); - self.branch = - Branch::under(&root, u32::try_from(level + 1).unwrap_or(1)); + self.branch = Branch::under(&root, level.saturating_add(1)); Some(self.branch.clone()) } + /// Whether to answer one more search from the parent now: at most + /// `SEARCH_BUDGET` a second, the rest dropped. + pub fn admit_search(&mut self, now: Instant) -> bool { + if now.duration_since(self.searches.0) >= Duration::from_secs(1) { + self.searches = (now, 0); + } + self.searches.1 += 1; + self.searches.1 <= SEARCH_BUDGET + } + /// The link to `username` closed. True when that was the parent, which /// puts us back to parentless. - pub fn closed(&mut self, username: &str) -> bool { + pub fn closed(&mut self, username: &str, link: u64) -> bool { + if !self.has_link(username, link) { + return false; + } self.candidates.remove(username); - if !self.is_parent(username) { + if self.parent.as_deref() != Some(username) { return false; } self.parent = None; @@ -180,6 +233,28 @@ impl Leaf { self.branch = Branch::parentless(&self.own); } + fn has_link(&self, username: &str, link: u64) -> bool { + self.candidates + .get(username) + .is_some_and(|c| c.link == link) + } + + fn moved_with_parent(&mut self, username: &str) -> Option { + if self.parent.as_deref() != Some(username) { + return None; + } + let candidate = self.candidates.get(username)?; + let branch = Branch::under( + candidate.root.as_deref()?, + candidate.level?.saturating_add(1), + ); + if branch == self.branch { + return None; + } + self.branch = branch.clone(); + Some(branch) + } + fn cancel_candidates(&mut self) { for candidate in self.candidates.values() { candidate.cancel.store(true, Ordering::Relaxed); @@ -205,17 +280,31 @@ pub fn announce( /// Dial a parent candidate on its own thread: a `D` PeerInit, then whatever it /// sends comes back as operations until the link closes or is cancelled. pub fn spawn_link(dial: Dial, own: String, ops: Sender) { - let _ = std::thread::Builder::new() + let closed = ClientOperation::ParentClosed { + parent: dial.username.clone(), + link: dial.link, + }; + let thread_ops = ops.clone(); + let spawned = std::thread::Builder::new() .name("soulseek-parent".to_string()) .spawn(move || { - let parent = dial.username.clone(); - if let Err(e) = link(&dial, &own, &ops) + if let Err(e) = link(&dial, &own, &thread_ops) && !dial.cancel.load(Ordering::Relaxed) { - crate::debug!("[distributed] link to {} ended: {}", parent, e); - let _ = ops.send(ClientOperation::ParentClosed { parent }); + crate::debug!( + "[distributed] link to {} ended: {}", + dial.username, + e + ); + let _ = thread_ops.send(ClientOperation::ParentClosed { + parent: dial.username, + link: dial.link, + }); } }); + if spawned.is_err() { + let _ = ops.send(closed); + } } fn link( @@ -230,70 +319,75 @@ fn link( io::Error::new(io::ErrorKind::InvalidInput, "no address") })?; let mut stream = TcpStream::connect_timeout(&addr, LINK_CONNECT_TIMEOUT)?; + if dial.cancel.load(Ordering::Relaxed) { + return Ok(()); + } stream.set_read_timeout(Some(CANCEL_POLL))?; stream.write_all( &MessageFactory::build_peer_init_message(own, ConnectionType::D, 0) .get_buffer(), )?; + let mut reader = MessageReader::new(); loop { if dial.cancel.load(Ordering::Relaxed) { return Ok(()); } - let Some(mut frame) = read_frame(&mut stream)? else { - continue; - }; - let parent = dial.username.clone(); - let operation = match distributed::parse(&mut frame) { - Some(Distributed::BranchLevel(level)) => { - ClientOperation::ParentBranchLevel { parent, level } + match reader.read_from_socket(&mut stream) { + Ok(()) => {} + Err(e) + if matches!( + e.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ) => + { + continue; } - Some(Distributed::BranchRoot(root)) => { - ClientOperation::ParentBranchRoot { parent, root } + Err(e) => return Err(e), + } + while let Some(mut frame) = reader.extract_message()? { + let Some(operation) = operation_of(dial, &mut frame) else { + continue; + }; + if ops.send(operation).is_err() { + return Ok(()); } - Some(Distributed::Search { - username, - token, - query, - }) => ClientOperation::ParentSearch { - parent, - username, - token, - query, - }, - None => continue, - }; - if ops.send(operation).is_err() { - return Ok(()); + } + if reader.buffer_len() > MAX_FRAME { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "oversized frame", + )); } } } -/// One framed message, or `None` when the read timed out with nothing sent. -fn read_frame(stream: &mut TcpStream) -> io::Result> { - let mut len = [0u8; 4]; - match stream.read_exact(&mut len) { - Ok(()) => {} - Err(e) - if matches!( - e.kind(), - io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut - ) => - { - return Ok(None); +fn operation_of( + dial: &Dial, + frame: &mut crate::message::Message, +) -> Option { + let parent = dial.username.clone(); + let link = dial.link; + Some(match distributed::parse(frame)? { + Distributed::BranchLevel(level) => ClientOperation::ParentBranchLevel { + parent, + link, + level, + }, + Distributed::BranchRoot(root) => { + ClientOperation::ParentBranchRoot { parent, link, root } } - Err(e) => return Err(e), - } - let size = u32::from_le_bytes(len) as usize; - if size > MAX_FRAME { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - "oversized frame", - )); - } - let mut data = len.to_vec(); - data.resize(4 + size, 0); - stream.read_exact(&mut data[4..])?; - Ok(Some(Message::new_with_data(data))) + Distributed::Search { + username, + token, + query, + } => ClientOperation::ParentSearch { + parent, + link, + username, + token, + query, + }, + }) } impl Client { @@ -339,6 +433,18 @@ mod tests { Leaf::new("me") } + fn candidate(name: &str) -> (String, String, u16) { + (name.to_string(), "127.0.0.1".to_string(), 2234) + } + + /// Dial the candidates and return their link numbers by name. + fn dial(leaf: &mut Leaf, names: &[&str]) -> HashMap { + leaf.consider(names.iter().map(|n| candidate(n)).collect()) + .into_iter() + .map(|d| (d.username.clone(), d)) + .collect() + } + #[test] fn a_leaf_without_a_parent_is_its_own_root_at_level_zero() { assert_eq!(leaf().branch(), Branch::parentless("me")); @@ -347,12 +453,13 @@ mod tests { #[test] fn candidates_are_dialled_only_while_there_is_no_parent() { let mut leaf = leaf(); - let dial = leaf.consider(vec![candidate("alice"), candidate("bob")]); - assert_eq!(dial.len(), 2); - leaf.branch_level("alice", 1); - leaf.branch_root("alice", "rooty"); + let dials = dial(&mut leaf, &["alice", "bob"]); + assert_eq!(dials.len(), 2); + let alice = dials["alice"].link; + leaf.branch_level("alice", alice, 1); + leaf.branch_root("alice", alice, "rooty"); assert!( - leaf.search_from("alice").is_some(), + leaf.search_from("alice", alice).is_some(), "alice becomes the parent" ); assert!(leaf.consider(vec![candidate("carol")]).is_empty()); @@ -361,48 +468,158 @@ mod tests { #[test] fn the_first_search_from_a_candidate_with_branch_info_adopts_it() { let mut leaf = leaf(); - leaf.consider(vec![candidate("alice"), candidate("bob")]); - assert_eq!(leaf.search_from("alice"), None, "no branch info yet"); - leaf.branch_level("alice", 3); - assert_eq!(leaf.search_from("alice"), None, "level without root"); - leaf.branch_root("alice", "rooty"); + let dials = dial(&mut leaf, &["alice", "bob"]); + let (alice, bob) = (dials["alice"].link, dials["bob"].link); + assert_eq!(leaf.search_from("alice", alice), None, "no branch yet"); + leaf.branch_level("alice", alice, 3); + assert_eq!(leaf.search_from("alice", alice), None, "level, no root"); + leaf.branch_root("alice", alice, "rooty"); assert_eq!( - leaf.search_from("alice"), + leaf.search_from("alice", alice), Some(Branch::under("rooty", 4)), "level 3 parent puts us at level 4" ); assert_eq!(leaf.branch(), Branch::under("rooty", 4)); - assert_eq!(leaf.search_from("alice"), None, "already the parent"); - assert_eq!(leaf.search_from("bob"), None, "a loser is ignored"); + assert!(leaf.is_parent("alice", alice)); + assert_eq!(leaf.search_from("alice", alice), None, "already parent"); + assert_eq!(leaf.search_from("bob", bob), None, "a loser is ignored"); + assert!( + dials["bob"].cancel.load(Ordering::Relaxed), + "the losing link is told to stop" + ); + assert!(!dials["alice"].cancel.load(Ordering::Relaxed)); } #[test] fn a_level_zero_candidate_is_its_own_root() { let mut leaf = leaf(); - leaf.consider(vec![candidate("alice")]); - leaf.branch_level("alice", 0); - assert_eq!(leaf.search_from("alice"), Some(Branch::under("alice", 1))); + let alice = dial(&mut leaf, &["alice"])["alice"].link; + leaf.branch_level("alice", alice, 0); + assert_eq!( + leaf.search_from("alice", alice), + Some(Branch::under("alice", 1)) + ); + } + + #[test] + fn a_hostile_branch_level_neither_panics_nor_adopts() { + let mut leaf = leaf(); + let alice = dial(&mut leaf, &["alice"])["alice"].link; + leaf.branch_root("alice", alice, "rooty"); + leaf.branch_level("alice", alice, -1); + assert_eq!(leaf.search_from("alice", alice), None, "negative level"); + leaf.branch_level("alice", alice, i32::MAX); + assert_eq!( + leaf.search_from("alice", alice), + Some(Branch::under("rooty", 1 << 31)) + ); + } + + #[test] + fn the_parents_later_moves_move_us_too() { + let mut leaf = leaf(); + let alice = dial(&mut leaf, &["alice"])["alice"].link; + leaf.branch_level("alice", alice, 0); + leaf.search_from("alice", alice); + assert_eq!(leaf.branch_level("alice", alice, 0), None, "unchanged"); + assert_eq!( + leaf.branch_root("alice", alice, "grand"), + Some(Branch::under("grand", 1)) + ); + assert_eq!( + leaf.branch_level("alice", alice, 2), + Some(Branch::under("grand", 3)) + ); + assert_eq!(leaf.branch(), Branch::under("grand", 3)); + } + + #[test] + fn a_stale_link_cannot_speak_for_a_newer_one() { + let mut leaf = leaf(); + let old = dial(&mut leaf, &["alice"])["alice"].link; + assert!(!leaf.closed("alice", old + 7), "unknown link is ignored"); + leaf.closed("alice", old); + let new = dial(&mut leaf, &["alice"])["alice"].link; + assert_ne!(old, new); + leaf.branch_level("alice", old, 0); + assert_eq!(leaf.search_from("alice", old), None, "old link ignored"); + assert!( + !leaf.closed("alice", old), + "a late close leaves the new one" + ); + leaf.branch_level("alice", new, 0); + assert!(leaf.search_from("alice", new).is_some()); + } + + #[test] + fn a_later_batch_keeps_the_candidates_in_flight_and_stays_under_the_cap() { + let mut leaf = leaf(); + let first = dial(&mut leaf, &["alice"])["alice"].link; + let again = dial(&mut leaf, &["alice", "bob"]); + assert!( + !again.contains_key("alice"), + "alice is already being dialled" + ); + assert!(again.contains_key("bob")); + assert!(leaf.has_link("alice", first)); + let names: Vec = (0..20).map(|i| format!("p{i}")).collect(); + let more = leaf + .consider(names.iter().map(|n| candidate(n)).collect::>()); + assert_eq!(more.len(), MAX_CANDIDATES - 2); } #[test] fn losing_the_parent_or_a_reset_returns_to_parentless() { let mut leaf = leaf(); - leaf.consider(vec![candidate("alice")]); - leaf.branch_level("alice", 0); - leaf.search_from("alice"); - assert!(!leaf.closed("bob"), "a stranger closing changes nothing"); - assert!(leaf.closed("alice")); + let dials = dial(&mut leaf, &["alice", "bob"]); + let alice = dials["alice"].link; + leaf.branch_level("alice", alice, 0); + leaf.search_from("alice", alice); + assert!(!leaf.closed("bob", dials["bob"].link), "not the parent"); + assert!(leaf.closed("alice", alice)); assert_eq!(leaf.branch(), Branch::parentless("me")); - leaf.consider(vec![candidate("alice")]); - leaf.branch_level("alice", 0); - leaf.search_from("alice"); + let dials = dial(&mut leaf, &["alice", "carol"]); + let alice = dials["alice"].link; + leaf.branch_level("alice", alice, 0); + leaf.search_from("alice", alice); leaf.reset(); assert_eq!(leaf.branch(), Branch::parentless("me")); + assert!(dials["alice"].cancel.load(Ordering::Relaxed)); + assert!(dials["carol"].cancel.load(Ordering::Relaxed)); assert!(!leaf.consider(vec![candidate("bob")]).is_empty()); } - fn candidate(name: &str) -> (String, String, u16) { - (name.to_string(), "127.0.0.1".to_string(), 2234) + #[test] + fn searches_beyond_the_budget_are_dropped_until_the_next_second() { + let mut leaf = leaf(); + let now = Instant::now(); + for _ in 0..SEARCH_BUDGET { + assert!(leaf.admit_search(now)); + } + assert!(!leaf.admit_search(now)); + assert!(leaf.admit_search(now + Duration::from_secs(1))); + } + + #[test] + fn announcing_tells_the_server_the_root_the_level_and_that_we_have_a_parent() + { + let (tx, rx) = std::sync::mpsc::channel(); + announce(Some(&tx), &Branch::under("rooty", 4), true); + let mut sent: Vec = Vec::new(); + while let Ok(ServerMessage::SendMessage(m)) = rx.try_recv() { + sent.push(m); + } + let codes: Vec = sent + .iter() + .map(|m| u32::from_le_bytes(m.get_data()[..4].try_into().unwrap())) + .collect(); + assert_eq!(codes, vec![71, 127, 126, 100]); + assert_eq!(sent[0].get_data()[4], 0, "HaveNoParent is false"); + let mut root = sent[1].clone(); + root.set_pointer(4); + assert_eq!(root.read_string(), "rooty"); + assert_eq!(&sent[2].get_data()[4..8], &4u32.to_le_bytes()); + assert_eq!(sent[3].get_data()[4], 0, "AcceptChildren is false"); } } diff --git a/soulseek-rs-lib/src/client/mod.rs b/soulseek-rs-lib/src/client/mod.rs index 9b80e54..160e7a6 100644 --- a/soulseek-rs-lib/src/client/mod.rs +++ b/soulseek-rs-lib/src/client/mod.rs @@ -60,6 +60,8 @@ pub const DEFAULT_UPLOAD_SLOTS: usize = 10; const BROKER_CONNECT_TIMEOUT: Duration = Duration::from_secs(20); const DEFAULT_MAX_PEERS: usize = 512; +/// How long a message waits for a control connection to its peer. +const PENDING_PEER_TTL: Duration = Duration::from_mins(1); const BROWSE_PROTECT_WINDOW: Duration = Duration::from_mins(5); @@ -338,19 +340,23 @@ pub enum ClientOperation { PossibleParents(Vec<(String, String, u16)>), /// The server told us to drop our parent. ResetDistributed, - /// A parent candidate told us how deep it sits. + /// A parent candidate told us how deep it sits. `link` says which dial + /// to that user is talking. ParentBranchLevel { parent: String, + link: u64, level: i32, }, /// A parent candidate told us whose branch it is on. ParentBranchRoot { parent: String, + link: u64, root: String, }, /// A search came down the tree from `parent`. ParentSearch { parent: String, + link: u64, username: String, token: u32, query: String, @@ -358,6 +364,7 @@ pub enum ClientOperation { /// The link to a parent or candidate is gone. ParentClosed { parent: String, + link: u64, }, } pub struct ClientContext { @@ -380,8 +387,10 @@ pub struct ClientContext { pub shared_directories: Vec, /// Peer listen addresses learned from GetPeerAddress responses. peer_addresses: HashMap, - /// Peer messages waiting for a control connection to that peer. - pending_peer_messages: HashMap>, + /// Peer messages waiting for a control connection to that peer, and + /// since when: a peer that never answers must not keep them forever. + pending_peer_messages: + HashMap)>, /// Uploads we have offered, keyed by our transfer token. uploads: HashMap, active_uploads: HashMap, @@ -474,6 +483,15 @@ impl ClientContext { } impl ClientContext { + /// A context for the client that logs in as `username`. + #[must_use] + pub fn for_user(username: &str) -> Self { + Self { + leaf: distributed::Leaf::new(username), + ..Self::new() + } + } + #[must_use] pub fn new() -> Self { let actor_system = Arc::new(ActorSystem::new()); @@ -725,7 +743,8 @@ impl ClientContext { ) { self.pending_peer_messages .entry(username.to_string()) - .or_default() + .or_insert_with(|| (Instant::now(), Vec::new())) + .1 .push(message); } @@ -736,9 +755,18 @@ impl ClientContext { ) -> Vec { self.pending_peer_messages .remove(username) + .map(|(_, messages)| messages) .unwrap_or_default() } + /// Drop peer messages nobody could deliver within + /// [`PENDING_PEER_TTL`]: the searcher went offline, or never existed. + pub(crate) fn expire_pending_peer_messages(&mut self, now: Instant) { + self.pending_peer_messages.retain(|_, (since, _)| { + now.duration_since(*since) < PENDING_PEER_TTL + }); + } + /// Store a shared-file listing received from browsing `username`. pub fn store_browse_result( &mut self, @@ -865,11 +893,13 @@ impl Client { listen_port: settings.listen_port, bound_port: None, address: settings.server_address, + context: Arc::new(RwLock::new(ClientContext::for_user( + &settings.username, + ))), username: settings.username, password: settings.password, version: settings.version, shared_directories: settings.shared_directories, - context: Arc::new(RwLock::new(ClientContext::new())), server_handle: None, session: SessionWatch::default(), } diff --git a/soulseek-rs-lib/src/client/operations.rs b/soulseek-rs-lib/src/client/operations.rs index da84f75..6d47d09 100644 --- a/soulseek-rs-lib/src/client/operations.rs +++ b/soulseek-rs-lib/src/client/operations.rs @@ -17,9 +17,6 @@ impl Client { own_username: String, ) { thread::spawn(move || { - if let Ok(mut ctx) = client_context.write_safe() { - ctx.leaf = super::distributed::Leaf::new(&own_username); - } let mut last_sweep = Instant::now(); loop { let next = reader.recv_timeout(CONNECT_SWEEP_INTERVAL); @@ -27,6 +24,9 @@ impl Client { last_sweep = Instant::now(); Self::sweep_expired_connects(&client_context); Self::sweep_stale_offers(&client_context); + if let Ok(mut ctx) = client_context.write_safe() { + ctx.expire_pending_peer_messages(Instant::now()); + } } let operation = match next { Ok(operation) => operation, @@ -716,18 +716,45 @@ impl Client { ); } } - ClientOperation::ParentBranchLevel { parent, level } => { - if let Ok(mut ctx) = client_context.write_safe() { - ctx.leaf.branch_level(&parent, level); + ClientOperation::ParentBranchLevel { + parent, + link, + level, + } => { + let Ok(mut ctx) = client_context.write_safe() else { + continue; + }; + if let Some(branch) = + ctx.leaf.branch_level(&parent, link, level) + { + super::distributed::announce( + ctx.server_sender.as_ref(), + &branch, + true, + ); } } - ClientOperation::ParentBranchRoot { parent, root } => { - if let Ok(mut ctx) = client_context.write_safe() { - ctx.leaf.branch_root(&parent, &root); + ClientOperation::ParentBranchRoot { + parent, + link, + root, + } => { + let Ok(mut ctx) = client_context.write_safe() else { + continue; + }; + if let Some(branch) = + ctx.leaf.branch_root(&parent, link, &root) + { + super::distributed::announce( + ctx.server_sender.as_ref(), + &branch, + true, + ); } } ClientOperation::ParentSearch { parent, + link, username, token, query, @@ -735,14 +762,17 @@ impl Client { let Ok(mut ctx) = client_context.write_safe() else { continue; }; - if let Some(branch) = ctx.leaf.search_from(&parent) { + if let Some(branch) = + ctx.leaf.search_from(&parent, link) + { super::distributed::announce( ctx.server_sender.as_ref(), &branch, true, ); } - if ctx.leaf.is_parent(&parent) + if ctx.leaf.is_parent(&parent, link) + && ctx.leaf.admit_search(Instant::now()) && let Some(ops) = &ctx.operations { let _ = ops.send(ClientOperation::IncomingSearch { @@ -752,11 +782,11 @@ impl Client { }); } } - ClientOperation::ParentClosed { parent } => { + ClientOperation::ParentClosed { parent, link } => { let Ok(mut ctx) = client_context.write_safe() else { continue; }; - if ctx.leaf.closed(&parent) { + if ctx.leaf.closed(&parent, link) { super::distributed::announce( ctx.server_sender.as_ref(), &ctx.leaf.branch(), @@ -764,6 +794,8 @@ impl Client { ); } } + // Also what a fresh login means: whatever tree we hung + // from belongs to the old session. ClientOperation::ResetDistributed => { let Ok(mut ctx) = client_context.write_safe() else { continue; diff --git a/soulseek-rs-lib/src/client/tests.rs b/soulseek-rs-lib/src/client/tests.rs index 031edb2..1c16b42 100644 --- a/soulseek-rs-lib/src/client/tests.rs +++ b/soulseek-rs-lib/src/client/tests.rs @@ -821,3 +821,18 @@ fn watched_users_are_listed_in_a_stable_order() { context.add_watched_user("bob"); assert_eq!(context.watched_users(), vec!["alice", "bob", "carol"]); } + +// A reply queued for a searcher the server cannot place waits a minute, +// not forever: every such search would otherwise pin a map entry and a +// protected slot in the peer registry. +#[test] +fn a_peer_message_nobody_could_deliver_expires() { + let mut ctx = ClientContext::for_user("me"); + ctx.queue_peer_message("ghost", crate::message::Message::new()); + ctx.expire_pending_peer_messages(Instant::now()); + assert_eq!(ctx.take_peer_messages("ghost").len(), 1, "still fresh"); + + ctx.queue_peer_message("ghost", crate::message::Message::new()); + ctx.expire_pending_peer_messages(Instant::now() + PENDING_PEER_TTL); + assert!(ctx.take_peer_messages("ghost").is_empty(), "expired"); +} diff --git a/soulseek-rs-lib/src/message/distributed.rs b/soulseek-rs-lib/src/message/distributed.rs index 756219e..2ae015c 100644 --- a/soulseek-rs-lib/src/message/distributed.rs +++ b/soulseek-rs-lib/src/message/distributed.rs @@ -34,15 +34,27 @@ pub fn parse(message: &mut Message) -> Option { parse_at(message, message.get_init_code()) } -/// Decode the body that follows a one-byte `code`, the pointer already past it. +/// Decode the body that follows a one-byte `code`, the pointer already past +/// it. A code-93 envelope is unwrapped exactly once, as Nicotine+ does: an +/// envelope inside an envelope is nobody's search. #[must_use] pub fn parse_at(message: &mut Message, code: u8) -> Option { + let code = if code == 93 { + message.read_int8() + } else { + code + }; match code { 3 => { - if message.read_int32() != SEARCH_IDENTIFIER { + if remaining(message) < 4 + || message.read_int32() != SEARCH_IDENTIFIER + { return None; } let username = message.read_string(); + if remaining(message) < 4 { + return None; + } let token = message.read_int32(); let query = message.read_string(); (!username.is_empty() && !query.is_empty()).then_some( @@ -53,17 +65,20 @@ pub fn parse_at(message: &mut Message, code: u8) -> Option { }, ) } - 4 => Some(Distributed::BranchLevel(message.read_int32() as i32)), - 5 => Some(Distributed::BranchRoot(message.read_string())), - // Older branch roots wrap a search in a code-93 envelope. - 93 => { - let inner = message.read_int8(); - parse_at(message, inner) + 4 => (remaining(message) >= 4) + .then(|| Distributed::BranchLevel(message.read_int32() as i32)), + 5 => { + let root = message.read_string(); + (!root.is_empty()).then_some(Distributed::BranchRoot(root)) } _ => None, } } +const fn remaining(message: &mut Message) -> usize { + message.get_size().saturating_sub(message.get_pointer()) +} + /// A distributed frame with the one-byte `code`: what a parent (or a test /// standing in for one) sends down the tree. fn framed(code: u8, body: impl FnOnce(&mut Message)) -> Message { @@ -171,5 +186,59 @@ mod tests { assert_eq!(parse(&mut ping), None); let mut short = Message::new_with_data(vec![1, 0, 0, 0, 3]); assert_eq!(parse(&mut short), None); + let mut level = Message::new_with_data(vec![3, 0, 0, 0, 4, 1, 0]); + assert_eq!(parse(&mut level), None, "a level needs four bytes"); + let mut root = + Message::new_with_data(build_branch_root("").get_buffer()); + assert_eq!(parse(&mut root), None, "an empty root is no root"); + let mut blank = + Message::new_with_data(build_search("seeker", 1, "").get_buffer()); + assert_eq!(parse(&mut blank), None, "an empty query is no search"); + } + + // An envelope inside an envelope is unwrapped once and found empty; a + // frame that is nothing but envelopes must not recurse its way off the + // stack. + #[test] + fn a_nested_envelope_is_not_unwrapped_twice() { + let mut nested = Message::new(); + nested.write_int8(93).write_raw_bytes( + Message::new() + .write_int8(93) + .write_raw_bytes(build_search("seeker", 5, "q").get_data()) + .get_data(), + ); + let mut message = Message::new_with_data(nested.get_buffer()); + assert_eq!(parse(&mut message), None); + + let mut onions = Message::new_with_data( + Message::new() + .write_raw_bytes(vec![93u8; 64 * 1024]) + .get_buffer(), + ); + assert_eq!(parse(&mut onions), None); + } + + // The frame as the protocol documents it, byte for byte: code 3, the + // constant 49, then user, token and query. + #[test] + fn a_search_frame_matches_the_documented_layout() { + let mut bytes = vec![3u8, 49, 0, 0, 0]; + bytes.extend(2u32.to_le_bytes()); + bytes.extend(b"me"); + bytes.extend(9u32.to_le_bytes()); + bytes.extend(1u32.to_le_bytes()); + bytes.extend(b"q"); + let mut framed = (bytes.len() as u32).to_le_bytes().to_vec(); + framed.extend(&bytes); + assert_eq!(build_search("me", 9, "q").get_buffer(), framed); + assert_eq!( + parse(&mut Message::new_with_data(framed)), + Some(Distributed::Search { + username: "me".to_string(), + token: 9, + query: "q".to_string() + }) + ); } } diff --git a/soulseek-rs-lib/src/message/server/distributed.rs b/soulseek-rs-lib/src/message/server/distributed.rs index fed5da9..8dacc78 100644 --- a/soulseek-rs-lib/src/message/server/distributed.rs +++ b/soulseek-rs-lib/src/message/server/distributed.rs @@ -4,6 +4,7 @@ use crate::actor::server_actor::ServerMessage; use crate::message::distributed::{Distributed, parse_at}; use crate::message::{Message, MessageHandler}; +use std::net::Ipv4Addr; use std::sync::mpsc::Sender; pub struct PossibleParentsHandler; @@ -13,22 +14,29 @@ impl MessageHandler for PossibleParentsHandler { } fn handle(&self, message: &mut Message, sender: Sender) { let count = message.read_int32(); - let mut candidates = Vec::new(); - for _ in 0..count.min(10) { - if message.get_pointer() >= message.get_size() { - break; - } - let username = message.read_string(); - // The address travels as four bytes in network order. - let ip: Vec = (0..4).map(|_| message.read_int8()).collect(); - let host = format!("{}.{}.{}.{}", ip[3], ip[2], ip[1], ip[0]); - let port = message.read_int32() as u16; - candidates.push((username, host, port)); - } + let candidates = (0..count.min(10)) + .map_while(|_| entry(message)) + .filter_map(|(username, host, port)| { + Some((username, host, u16::try_from(port).ok()?)) + }) + .collect(); let _ = sender.send(ServerMessage::PossibleParents(candidates)); } } +/// One whole `(username, host, port)` entry, or `None` where the message +/// runs out: a truncated entry is not a parent to dial. +fn entry(message: &mut Message) -> Option<(String, String, u32)> { + let username = message.read_string(); + if username.is_empty() + || message.get_size().saturating_sub(message.get_pointer()) < 8 + { + return None; + } + let host = Ipv4Addr::from(message.read_int32()).to_string(); + Some((username, host, message.read_int32())) +} + pub struct ResetDistributedHandler; impl MessageHandler for ResetDistributedHandler { fn get_code(&self) -> u32 { @@ -95,6 +103,33 @@ mod tests { ); } + // The count says two, the bytes hold one and a half: the whole one is + // a candidate, the stub is not. A port that is not a port is skipped too. + #[test] + fn a_truncated_or_absurd_entry_is_not_a_candidate() { + let (sender, received) = channel(); + let mut message = crate::message::framed(|m| { + m.write_int32(3) + .write_string("alice") + .write_int32(u32::from_le_bytes([1, 0, 0, 127])) + .write_int32(70_000) + .write_string("bob") + .write_int32(u32::from_le_bytes([8, 8, 8, 8])) + .write_int32(2235) + .write_string("carol"); + }); + PossibleParentsHandler.handle(&mut message, sender); + let ServerMessage::PossibleParents(candidates) = + received.try_recv().unwrap() + else { + panic!("expected PossibleParents"); + }; + assert_eq!( + candidates, + vec![("bob".to_string(), "8.8.8.8".to_string(), 2235)] + ); + } + #[test] fn a_reset_and_an_embedded_search_become_client_events() { let (sender, received) = channel(); diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index b0d2fe8..f4e7725 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -3669,20 +3669,21 @@ fn a_leaf_adopts_a_parent_and_answers_the_searches_it_passes_down() { std::fs::write(share_dir.join("album").join("treesearch.mp3"), b"xxxx") .unwrap(); - // The parent doubles as the searcher, so the leaf's answer comes back to - // the same listener. Bind all interfaces: soulfind reports the LAN address. - let parent_port = free_port().expect("parent port"); - let listener = - std::net::TcpListener::bind(("0.0.0.0", parent_port)).unwrap(); + // The parent and the searcher are two logged-in users with a listener + // each. Bind all interfaces: soulfind reports the LAN address. let server_addr = format!("{}:{}", server.host, server.port); - let mut parent_srv = - login_raw(&server_addr, "e2e_parent", "pw").expect("parent logs in"); - parent_srv - .write_all( - &MessageFactory::build_set_wait_port_message(parent_port) - .get_buffer(), + let raw_user = |name: &str| { + let port = free_port().expect("a port for the raw user"); + let listener = std::net::TcpListener::bind(("0.0.0.0", port)).unwrap(); + let mut srv = login_raw(&server_addr, name, "pw").expect("raw login"); + srv.write_all( + &MessageFactory::build_set_wait_port_message(port).get_buffer(), ) .unwrap(); + (listener, port, srv) + }; + let (listener, parent_port, _parent_srv) = raw_user("e2e_parent"); + let (searcher, _, _searcher_srv) = raw_user("e2e_searcher"); let mut leaf = Client::with_settings(ClientSettings { shared_directories: vec![share_dir.display().to_string()], @@ -3721,7 +3722,7 @@ fn a_leaf_adopts_a_parent_and_answers_the_searches_it_passes_down() { link.write_all(&distributed::build_branch_root("e2e_parent").get_buffer()) .unwrap(); link.write_all( - &distributed::build_search("e2e_parent", 4242, "treesearch") + &distributed::build_search("e2e_searcher", 4242, "treesearch") .get_buffer(), ) .unwrap(); @@ -3736,9 +3737,10 @@ fn a_leaf_adopts_a_parent_and_answers_the_searches_it_passes_down() { "the first search from a candidate with a branch adopts it" ); - // The answer arrives the ordinary way: a P connection to the searcher - // carrying a FileSearchResponse for our token. - let mut p = accept_within(&listener, Duration::from_secs(20)) + // The answer goes to whoever asked, not to the parent that relayed it: + // a P connection to the searcher carrying a FileSearchResponse for the + // token. + let mut p = accept_within(&searcher, Duration::from_secs(20)) .expect("the leaf dials the searcher to answer"); assert_eq!( peer_init_of(&mut p).unwrap(),