From f54f8f9b82bea2887d0c0b41c4c78ddd6afe7568 Mon Sep 17 00:00:00 2001 From: Michel de Graaf Date: Mon, 27 Jul 2026 10:05:25 +0200 Subject: [PATCH] feat(daemon): resident session, auth-gated control socket, and event fan-out The daemon drains the library's destructive take_* accessors once and fans events out per connection, which is what lets more than one client share a session. A Unix socket at 0600 is the local transport and its permissions are the authentication; TCP is opt-in and needs the generated token. Rooms, transfers, and the conversation are daemon-owned and restored on restart. --- Cargo.lock | 1 + docs/openrpc.json | 29 +- soulseek-rs/Cargo.toml | 1 + soulseek-rs/src/daemon/hub.rs | 449 ++++++++++++++ soulseek-rs/src/daemon/mod.rs | 425 +++++++++++++ soulseek-rs/src/daemon/proto.rs | 16 +- soulseek-rs/src/daemon/server.rs | 993 +++++++++++++++++++++++++++++++ soulseek-rs/src/lib.rs | 4 + soulseek-rs/src/remote/mod.rs | 1 + 9 files changed, 1917 insertions(+), 2 deletions(-) create mode 100644 soulseek-rs/src/daemon/hub.rs create mode 100644 soulseek-rs/src/daemon/server.rs create mode 100644 soulseek-rs/src/remote/mod.rs diff --git a/Cargo.lock b/Cargo.lock index 041d018..bf03565 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2553,6 +2553,7 @@ dependencies = [ "color-eyre", "directories", "dotenv", + "getrandom 0.4.3", "igd-next", "keyring", "ratatui", diff --git a/docs/openrpc.json b/docs/openrpc.json index cbda3bf..9c34c0a 100644 --- a/docs/openrpc.json +++ b/docs/openrpc.json @@ -72,6 +72,33 @@ ], "type": "object" }, + "ChatMessageDto": { + "description": "One line of chat history. Unlike a live [`UserMessageDto`] this carries a\ndirection, because a conversation the daemon collected includes what this\naccount sent as well as what it received.", + "properties": { + "at": { + "description": "Unix seconds.", + "format": "int64", + "type": "integer" + }, + "outgoing": { + "type": "boolean" + }, + "peer": { + "description": "The other party, whichever way the message went.", + "type": "string" + }, + "text": { + "type": "string" + } + }, + "required": [ + "peer", + "outgoing", + "text", + "at" + ], + "type": "object" + }, "DaemonStatus": { "description": "What the daemon is and what its session is doing, from `daemon.status`.", "properties": { @@ -509,7 +536,7 @@ "properties": { "messages": { "items": { - "$ref": "#/components/schemas/UserMessageDto" + "$ref": "#/components/schemas/ChatMessageDto" }, "type": "array" } diff --git a/soulseek-rs/Cargo.toml b/soulseek-rs/Cargo.toml index d1d6c54..73856c5 100644 --- a/soulseek-rs/Cargo.toml +++ b/soulseek-rs/Cargo.toml @@ -42,6 +42,7 @@ directories = "6" # system libdbus and breaks builds without libdbus-1-dev. keyring = { version = "3", features = ["apple-native", "windows-native", "async-secret-service", "async-io", "crypto-rust"] } chrono = { version = "0.4.45", default-features = false, features = ["clock", "serde"] } +getrandom = "0.4.3" [dev-dependencies] schemars = "1.2.1" diff --git a/soulseek-rs/src/daemon/hub.rs b/soulseek-rs/src/daemon/hub.rs new file mode 100644 index 0000000..1acee85 --- /dev/null +++ b/soulseek-rs/src/daemon/hub.rs @@ -0,0 +1,449 @@ +//! The single drainer, and the fan-out that makes multiple clients possible. +//! +//! The library's `take_*` accessors are destructive — they hand over their +//! buffer and empty it. That is fine for one process, but a daemon serving two +//! clients cannot let either of them drain: whoever asked first would silently +//! steal the other's events. So the daemon drains once, on its own cadence, +//! and every attached connection gets its own copy. +//! +//! A subscriber that stops reading must not be able to stall the drainer, so +//! the queues are bounded and a full one costs that client its connection +//! rather than costing everyone else their events. + +use super::proto::{ + BrowseEvent, ChatMessageDto, DownloadStatusEvent, Event, Response, + SessionLossEvent, UploadInfoDto, UserMessageDto, +}; +use crate::api::SessionApi; +use soulseek_rs::DownloadStatus; +use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::mpsc::{Receiver, SyncSender, TrySendError, sync_channel}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +/// How often the drainer samples the session. The TUI already redraws on this +/// cadence, so events reach a remote UI as promptly as a local one. +pub const POLL: Duration = Duration::from_millis(100); + +/// How many events may back up for one client before it is disconnected. +/// Generous enough that an ordinary consumer never notices, small enough that +/// a dead one is dropped promptly instead of growing without bound. +const QUEUE_DEPTH: usize = 1024; + +/// How many private messages the daemon keeps for `message.history`, so a +/// client that attaches later still sees what arrived while nobody was +/// listening. +const HISTORY_LIMIT: usize = 500; + +struct Subscriber { + id: u64, + sender: SyncSender, +} + +/// Fans daemon-side events out to every attached connection. +pub struct Hub { + subscribers: Mutex>, + next_id: AtomicU64, + /// The conversation so far, newest last — both directions, so a client + /// that attaches later sees a whole conversation rather than half of one. + history: Mutex>, +} + +impl Hub { + #[must_use] + pub const fn new() -> Self { + Self { + subscribers: Mutex::new(Vec::new()), + next_id: AtomicU64::new(0), + history: Mutex::new(Vec::new()), + } + } + + /// Attach a connection. The returned receiver yields notifications until + /// [`Hub::unsubscribe`] or a full queue drops it. + pub fn subscribe(&self) -> (u64, Receiver) { + let id = self.next_id.fetch_add(1, Ordering::Relaxed); + let (sender, receiver) = sync_channel(QUEUE_DEPTH); + if let Ok(mut subscribers) = self.subscribers.lock() { + subscribers.push(Subscriber { id, sender }); + } + (id, receiver) + } + + pub fn unsubscribe(&self, id: u64) { + if let Ok(mut subscribers) = self.subscribers.lock() { + subscribers.retain(|subscriber| subscriber.id != id); + } + } + + pub fn subscribers(&self) -> usize { + self.subscribers.lock().map_or(0, |subscribers| subscribers.len()) + } + + /// Send one event to everyone, dropping any client that has stopped + /// keeping up. The drainer must never block on a subscriber. + fn publish(&self, event: Event, params: serde_json::Value) { + let Ok(mut subscribers) = self.subscribers.lock() else { + return; + }; + let notification = Response::notification(event.wire_name(), params); + subscribers.retain(|subscriber| { + match subscriber.sender.try_send(notification.clone()) { + Ok(()) => true, + Err(TrySendError::Full(_)) => { + soulseek_rs::warn!( + "control client {} is not reading its events; dropping it", + subscriber.id + ); + false + } + Err(TrySendError::Disconnected(_)) => false, + } + }); + } + + /// Add one line to the conversation, oldest lines falling off the front. + pub fn remember(&self, message: ChatMessageDto) { + let Ok(mut history) = self.history.lock() else { + return; + }; + history.push(message); + let overflow = history.len().saturating_sub(HISTORY_LIMIT); + if overflow > 0 { + history.drain(0..overflow); + } + } + + /// Seed the conversation from disk at startup. + pub fn restore(&self, messages: Vec) { + if let Ok(mut history) = self.history.lock() { + *history = messages; + } + } + + #[must_use] + pub fn history(&self) -> Vec { + self.history + .lock() + .map_or_else(|_| Vec::new(), |history| history.clone()) + } +} + +impl Default for Hub { + fn default() -> Self { + Self::new() + } +} + +/// A transfer the daemon started, whose status channel it pumps into events. +pub struct PendingDownload { + pub username: String, + pub filename: String, + pub updates: Receiver, +} + +/// Everything the drainer owns between ticks. +/// +/// Kept apart from [`Hub`] because it is single-threaded by construction: only +/// the drainer touches it, so none of it needs a lock. +#[derive(Default)] +struct DrainState { + /// Last upload state reported per (user, file), so an upload that has not + /// moved is not re-announced every tick. + uploads: HashMap<(String, String), UploadInfoDto>, + /// Whether the session-loss event has already gone out. It is terminal, so + /// it is worth saying exactly once. + reported_loss: bool, +} + +/// Peers we have asked for a listing and are still waiting on. `take_browse_result` +/// is per-user and destructive, so the daemon has to know who to ask about. +#[derive(Default)] +pub struct PendingBrowses(Mutex>); + +impl PendingBrowses { + pub fn expect(&self, username: &str) { + if let Ok(mut pending) = self.0.lock() + && !pending.iter().any(|name| name == username) + { + pending.push(username.to_string()); + } + } + + fn snapshot(&self) -> Vec { + self.0.lock().map_or_else(|_| Vec::new(), |pending| pending.clone()) + } + + fn settled(&self, username: &str) { + self.forget(username); + } + + /// Stop waiting on a peer whose request never reached the wire. + pub fn forget(&self, username: &str) { + if let Ok(mut pending) = self.0.lock() { + pending.retain(|name| name != username); + } + } +} + +/// Drain everything the session has buffered and fan it out, forever. +/// +/// This is the only caller of the `take_*` family in daemon mode; anything else +/// draining them would take events out of this loop's hands. +pub fn run( + session: Arc, + hub: Arc, + browses: Arc, + downloads: Arc>>, + stop: Arc, +) { + let mut state = DrainState::default(); + while !stop.load(Ordering::Relaxed) { + tick(&*session, &hub, &browses, &downloads, &mut state); + std::thread::sleep(POLL); + } +} + +fn tick( + session: &dyn SessionApi, + hub: &Hub, + browses: &PendingBrowses, + downloads: &Mutex>, + state: &mut DrainState, +) { + for event in session.take_room_events() { + hub.publish( + Event::Room, + serde_json::to_value(super::proto::RoomEventDto::from(&event)) + .unwrap_or(serde_json::Value::Null), + ); + } + + for message in session.take_private_messages() { + let dto = UserMessageDto::from(&message); + hub.remember(ChatMessageDto { + peer: dto.username.clone(), + outgoing: false, + text: dto.message.clone(), + at: i64::from(dto.timestamp), + }); + hub.publish( + Event::Message, + serde_json::to_value(&dto).unwrap_or(serde_json::Value::Null), + ); + } + + publish_uploads(session, hub, state); + publish_browses(session, hub, browses); + publish_downloads(hub, downloads); + + if !state.reported_loss + && let Some(loss) = session.session_loss() + { + state.reported_loss = true; + hub.publish( + Event::SessionLoss, + serde_json::to_value(SessionLossEvent { loss: loss.into() }) + .unwrap_or(serde_json::Value::Null), + ); + } +} + +/// Report an upload whenever it differs from what was last announced. +/// +/// Both sources matter: the drained events carry states that came and went +/// between two ticks (a small file served straight out of the queue), and the +/// live table carries progress. +fn publish_uploads(session: &dyn SessionApi, hub: &Hub, state: &mut DrainState) { + let sampled = session + .take_upload_events() + .iter() + .chain(session.uploads().iter()) + .map(UploadInfoDto::from) + .collect::>(); + + for upload in sampled { + let key = (upload.username.clone(), upload.filename.clone()); + if state.uploads.get(&key) == Some(&upload) { + continue; + } + hub.publish( + Event::Upload, + serde_json::to_value(&upload).unwrap_or(serde_json::Value::Null), + ); + state.uploads.insert(key, upload); + } +} + +fn publish_browses( + session: &dyn SessionApi, + hub: &Hub, + browses: &PendingBrowses, +) { + for username in browses.snapshot() { + let Some(directories) = session.take_browse_result(&username) else { + continue; + }; + browses.settled(&username); + hub.publish( + Event::Browse, + serde_json::to_value(BrowseEvent { + username, + directories: directories.iter().map(Into::into).collect(), + }) + .unwrap_or(serde_json::Value::Null), + ); + } +} + +/// Pump each transfer's status channel, dropping it once it can say no more. +fn publish_downloads(hub: &Hub, downloads: &Mutex>) { + let Ok(mut downloads) = downloads.lock() else { + return; + }; + downloads.retain(|pending| { + loop { + match pending.updates.try_recv() { + Ok(status) => hub.publish( + Event::DownloadStatus, + serde_json::to_value(DownloadStatusEvent { + username: pending.username.clone(), + filename: pending.filename.clone(), + status: (&status).into(), + }) + .unwrap_or(serde_json::Value::Null), + ), + Err(std::sync::mpsc::TryRecvError::Empty) => return true, + Err(std::sync::mpsc::TryRecvError::Disconnected) => { + return false; + } + } + } + }); +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn drain(receiver: &Receiver) -> Vec { + let mut seen = Vec::new(); + while let Ok(response) = receiver.try_recv() { + seen.push(response.method.unwrap_or_default()); + } + seen + } + + #[test] + fn every_subscriber_receives_the_same_event() { + // The whole reason the hub exists: two clients must both see an event + // that only exists once in the library's buffer. + let hub = Hub::new(); + let (_, first) = hub.subscribe(); + let (_, second) = hub.subscribe(); + + hub.publish(Event::Room, json!({ "event": "left", "room": "lobby" })); + + assert_eq!(drain(&first), ["event.room"]); + assert_eq!(drain(&second), ["event.room"]); + } + + #[test] + fn unsubscribing_stops_delivery() { + let hub = Hub::new(); + let (id, receiver) = hub.subscribe(); + hub.unsubscribe(id); + hub.publish(Event::Upload, json!({})); + assert!(drain(&receiver).is_empty()); + assert_eq!(hub.subscribers(), 0); + } + + #[test] + fn a_client_that_stops_reading_is_dropped_and_the_others_keep_up() { + let hub = Hub::new(); + let (_, slow) = hub.subscribe(); + let (_, healthy) = hub.subscribe(); + + // Overrun the slow client's queue while the healthy one keeps reading, + // which is exactly the asymmetry the bound exists to survive. + let mut received = 0; + for _ in 0..(QUEUE_DEPTH + 10) { + hub.publish(Event::Upload, json!({})); + received += drain(&healthy).len(); + } + + assert_eq!( + hub.subscribers(), + 1, + "a client that never reads must lose its connection, not stall the \ + drainer" + ); + assert_eq!( + received, + QUEUE_DEPTH + 10, + "the healthy client must not lose events to the slow one" + ); + drop(slow); + } + + #[test] + fn a_dropped_receiver_removes_its_subscriber() { + let hub = Hub::new(); + let (_, receiver) = hub.subscribe(); + drop(receiver); + hub.publish(Event::Message, json!({})); + assert_eq!(hub.subscribers(), 0); + } + + #[test] + fn history_keeps_the_newest_messages_within_the_cap() { + let hub = Hub::new(); + for at in 0..(HISTORY_LIMIT as i64 + 5) { + hub.remember(ChatMessageDto { + peer: "bob".into(), + outgoing: false, + text: format!("m{at}"), + at, + }); + } + let history = hub.history(); + assert_eq!(history.len(), HISTORY_LIMIT); + assert_eq!(history[0].at, 5, "the oldest messages are the ones dropped"); + assert_eq!(history[HISTORY_LIMIT - 1].at, HISTORY_LIMIT as i64 + 4); + } + + #[test] + fn history_carries_both_directions_of_a_conversation() { + // Half a conversation is worse than none: a remote client must see + // what this account said as well as what it was told. + let hub = Hub::new(); + hub.remember(ChatMessageDto { + peer: "bob".into(), + outgoing: false, + text: "hi".into(), + at: 1, + }); + hub.remember(ChatMessageDto { + peer: "bob".into(), + outgoing: true, + text: "hello".into(), + at: 2, + }); + let history = hub.history(); + assert_eq!(history.len(), 2); + assert!(!history[0].outgoing); + assert!(history[1].outgoing); + } + + #[test] + fn a_browse_is_only_awaited_once() { + let pending = PendingBrowses::default(); + pending.expect("bob"); + pending.expect("bob"); + assert_eq!(pending.snapshot(), ["bob"]); + pending.settled("bob"); + assert!(pending.snapshot().is_empty()); + } +} diff --git a/soulseek-rs/src/daemon/mod.rs b/soulseek-rs/src/daemon/mod.rs index 3e9c555..02e4527 100644 --- a/soulseek-rs/src/daemon/mod.rs +++ b/soulseek-rs/src/daemon/mod.rs @@ -1,7 +1,432 @@ //! The resident service: one logged-in session, controlled over a socket. +//! +//! Where a one-shot command logs in, does its job and exits, the daemon logs in +//! once and stays. That is what makes cross-invocation work possible — a queue +//! that survives the command that filled it, a chat window that was open while +//! nobody was watching — and what lets several clients share one session +//! instead of fighting the server's one-login-per-account rule. +//! +//! Two transports, two answers to "who is this": +//! +//! * A Unix socket at `/daemon.sock`, mode 0600. The socket's own +//! permissions are the authentication, the same bargain Docker and ssh-agent +//! make, so a local client — an agent especially — needs no flags and no +//! token. +//! * TCP, opt-in, always requiring the token in `/daemon.token`. +pub mod hub; /// Generates `docs/openrpc.json` from the wire types and checks it is current. /// Test-only: the published document is served at runtime by `include_str!`. #[cfg(test)] mod openrpc; pub mod proto; +pub mod server; + +use crate::api::SessionApi; +use crate::commands::{Ctx, Session}; +use crate::output::{CliError, CliResult}; +use crate::persist::state::{PersistedDownload, PersistedMessage, StateStore}; +use hub::{Hub, PendingBrowses, PendingDownload}; +use proto::ChatMessageDto; +use server::Daemon; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +/// The control socket, where a local client looks without being told to. +pub const SOCKET_NAME: &str = "daemon.sock"; +/// The shared secret for TCP clients — the cookie-file convention, so a local +/// client can pick it up without the operator handling it at all. +pub const TOKEN_NAME: &str = "daemon.token"; + +/// How often daemon-owned state is written back. Frequent enough that a kill +/// loses little, rare enough not to churn the disk on a busy queue. +const SAVE_INTERVAL: Duration = Duration::from_secs(5); + +#[must_use] +pub fn socket_path() -> Option { + crate::persist::paths::state_dir().map(|dir| dir.join(SOCKET_NAME)) +} + +#[must_use] +pub fn token_path() -> Option { + crate::persist::paths::state_dir().map(|dir| dir.join(TOKEN_NAME)) +} + +/// The token a local client should present, when there is one to find. +#[must_use] +pub fn stored_token() -> Option { + let token = std::fs::read_to_string(token_path()?).ok()?; + let token = token.trim().to_string(); + (!token.is_empty()).then_some(token) +} + +/// Read the token, or mint one on first start. +/// +/// Written 0600 and never printed to stdout, so `daemon` output stays usable in +/// a pipe while the secret is still one `daemon token` away. +pub fn load_or_create_token(path: &Path) -> std::io::Result { + if let Some(existing) = std::fs::read_to_string(path) + .ok() + .map(|token| token.trim().to_string()) + .filter(|token| !token.is_empty()) + { + return Ok(existing); + } + + let token = mint_token(); + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent)?; + } + std::fs::write(path, format!("{token}\n"))?; + restrict(path)?; + Ok(token) +} + +/// 256 bits from the OS, hex-encoded. Long enough that the one-second penalty +/// on a wrong answer is the only rate limit the daemon needs. +fn mint_token() -> String { + use std::fmt::Write; + + let mut bytes = [0u8; 32]; + getrandom::fill(&mut bytes).expect("the OS always has randomness"); + bytes + .iter() + .fold(String::with_capacity(64), |mut token, byte| { + let _ = write!(token, "{byte:02x}"); + token + }) +} + +/// Make a secret readable by nobody else. A no-op where the platform has no +/// such notion, which is why TCP is opt-in rather than the default. +#[cfg(unix)] +fn restrict(path: &Path) -> std::io::Result<()> { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)) +} + +#[cfg(not(unix))] +const fn restrict(_path: &Path) -> std::io::Result<()> { + Ok(()) +} + +/// Whether a daemon is listening on the local socket right now. +/// +/// This is what makes daemon mode seamless: a client that finds a live socket +/// uses it, and one that does not behaves exactly as it always has. A socket +/// file left by a crash answers `None`, so a stale file cannot strand a client. +#[must_use] +pub fn local_daemon() -> Option { + let path = socket_path()?; + #[cfg(unix)] + { + std::os::unix::net::UnixStream::connect(&path) + .ok() + .map(|_| path) + } + #[cfg(not(unix))] + { + let _ = path; + None + } +} + +/// Run the daemon until it is asked to stop. +pub fn run( + ctx: &Ctx, + bind: Option<&str>, + upload_slots: Option, +) -> CliResult { + let state_dir = crate::persist::paths::state_dir().ok_or_else(|| { + CliError::usage("no state directory: set SOULSEEK_STATE_DIR") + })?; + std::fs::create_dir_all(&state_dir).map_err(|e| { + CliError::usage(format!( + "cannot use state directory {}: {e}", + state_dir.display() + )) + })?; + + let token = load_or_create_token(&state_dir.join(TOKEN_NAME)) + .map_err(|e| CliError::usage(format!("cannot read the token: {e}")))?; + + // Bind before logging in: a second daemon should fail in a millisecond, + // not after a full login handshake against the server. + let mut listeners = Vec::new(); + #[cfg(unix)] + { + let path = state_dir.join(SOCKET_NAME); + listeners.push(server::bind_unix(&path).map_err(|e| { + CliError::usage(format!("cannot listen on {}: {e}", path.display())) + })?); + ctx.out.status(&format!("listening on {}", path.display())); + } + if let Some(address) = bind { + listeners.push(server::bind_tcp(address).map_err(|e| { + CliError::connection(format!("cannot listen on {address}: {e}")) + })?); + ctx.out.status(&format!( + "listening on {address} (remote clients need the token from \ + `soulseek-rs daemon token`)" + )); + } + if listeners.is_empty() { + return Err(CliError::usage( + "nothing to listen on: pass --bind on a platform without Unix \ + sockets", + )); + } + + let session = Session::open(ctx)?; + if let Some(slots) = upload_slots { + session.client.set_upload_slots(usize::from(slots)); + } + + let store = StateStore::new(state_dir); + let hub = Arc::new(Hub::new()); + let browses = Arc::new(PendingBrowses::default()); + let downloads = Arc::new(Mutex::new(Vec::new())); + let stop = Arc::new(AtomicBool::new(false)); + + restore(ctx, &store, session.client.as_ref(), &hub, &downloads); + + let daemon = Arc::new(Daemon { + session: Arc::clone(&session.client), + hub: Arc::clone(&hub), + browses: Arc::clone(&browses), + downloads: Arc::clone(&downloads), + download_dir: ctx.download_dir.clone(), + server: ctx.settings.server_address.to_string(), + token, + started: Instant::now(), + stop: Arc::clone(&stop), + rooms: Mutex::new(store.load_rooms()), + }); + + let drainer = { + let session = Arc::clone(&session.client); + let hub = Arc::clone(&hub); + let stop = Arc::clone(&stop); + std::thread::spawn(move || { + hub::run(session, hub, browses, downloads, stop); + }) + }; + + for listener in listeners { + let daemon = Arc::clone(&daemon); + std::thread::spawn(move || listener.serve(daemon)); + } + + let (folders, files) = session.client.shared_counts(); + ctx.out.status(&format!( + "logged in as {}, sharing {files} files in {folders} folders", + session.client.username() + )); + ctx.out.status("ready"); + + // The listeners run on their own threads; this one only has to notice a + // `daemon.stop` and write state back on the way past. + let mut last_save = Instant::now(); + while !stop.load(Ordering::Relaxed) { + std::thread::sleep(hub::POLL); + if last_save.elapsed() >= SAVE_INTERVAL { + last_save = Instant::now(); + save(&store, session.client.as_ref(), &hub, &daemon); + } + } + + save(&store, session.client.as_ref(), &hub, &daemon); + ctx.out.status("stopping"); + // The drainer watches the same flag; joining keeps a half-written state + // file from racing the process exit. + let _ = drainer.join(); + cleanup(); + Ok(()) +} + +/// Remove the socket file so the next start does not have to reason about +/// whether it is stale. +fn cleanup() { + #[cfg(unix)] + if let Some(path) = socket_path() { + let _ = std::fs::remove_file(path); + } +} + +/// Bring back what the last run left: re-queue unfinished transfers, rejoin +/// rooms, and reload the conversation. +fn restore( + ctx: &Ctx, + store: &StateStore, + session: &dyn SessionApi, + hub: &Hub, + downloads: &Mutex>, +) { + hub.restore( + store + .load_messages() + .into_iter() + .map(|message| ChatMessageDto { + peer: message.peer, + outgoing: message.outgoing, + text: message.text, + at: message.at.timestamp(), + }) + .collect(), + ); + + for room in store.load_rooms() { + if let Err(e) = session.join_room(&room) { + ctx.out.warn(&format!("cannot rejoin {room}: {e}")); + } + } + + let unfinished: Vec = store + .load_downloads() + .into_iter() + .filter(|download| !download.completed) + .collect(); + if unfinished.is_empty() { + return; + } + ctx.out + .status(&format!("resuming {} transfers", unfinished.len())); + for download in unfinished { + match session.download( + download.filename.clone(), + download.username.clone(), + download.size, + download.download_directory.clone(), + ) { + Ok((_, updates)) => { + if let Ok(mut pending) = downloads.lock() { + pending.push(PendingDownload { + username: download.username, + filename: download.filename, + updates, + }); + } + } + Err(e) => ctx + .out + .warn(&format!("cannot resume {}: {e}", download.filename)), + } + } +} + +/// Write daemon-owned state back. Best-effort: a state file that cannot be +/// written is not a reason to drop a working session. +fn save( + store: &StateStore, + session: &dyn SessionApi, + hub: &Hub, + daemon: &Daemon, +) { + let downloads: Vec = session + .get_all_downloads() + .iter() + .map(|download| PersistedDownload { + username: download.username.clone(), + filename: download.filename.clone(), + size: download.size, + download_directory: download.download_directory.clone(), + completed: matches!( + download.status, + soulseek_rs::DownloadStatus::Completed + ), + }) + .collect(); + let _ = store.save_downloads(&downloads); + + let messages: Vec = hub + .history() + .into_iter() + .map(|message| PersistedMessage { + peer: message.peer, + outgoing: message.outgoing, + text: message.text, + at: chrono::DateTime::from_timestamp(message.at, 0) + .unwrap_or_default() + .into(), + }) + .collect(); + let _ = store.save_messages(&messages); + let _ = store.save_rooms(&daemon.rooms()); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_token_is_minted_once_and_then_reused() { + let dir = tempfile::tempdir().expect("a temp dir"); + let path = dir.path().join(TOKEN_NAME); + + let first = load_or_create_token(&path).expect("a token is minted"); + let second = load_or_create_token(&path).expect("the token is reused"); + + assert_eq!( + first, second, + "restarting must not lock existing clients out" + ); + assert_eq!(first.len(), 64, "256 bits, hex encoded"); + assert!(first.chars().all(|c| c.is_ascii_hexdigit())); + } + + #[test] + fn two_tokens_are_not_the_same() { + let dir = tempfile::tempdir().expect("a temp dir"); + let first = load_or_create_token(&dir.path().join("a")).unwrap(); + let second = load_or_create_token(&dir.path().join("b")).unwrap(); + assert_ne!(first, second); + } + + #[cfg(unix)] + #[test] + fn a_token_file_is_readable_by_nobody_else() { + use std::os::unix::fs::PermissionsExt; + let dir = tempfile::tempdir().expect("a temp dir"); + let path = dir.path().join(TOKEN_NAME); + load_or_create_token(&path).expect("a token is minted"); + let mode = std::fs::metadata(&path).unwrap().permissions().mode(); + assert_eq!(mode & 0o777, 0o600, "the token is a secret on disk"); + } + + #[test] + fn a_blank_token_file_is_replaced_rather_than_trusted() { + let dir = tempfile::tempdir().expect("a temp dir"); + let path = dir.path().join(TOKEN_NAME); + std::fs::write(&path, " \n").unwrap(); + let token = load_or_create_token(&path).expect("a token is minted"); + assert_eq!(token.len(), 64, "an empty secret must never authenticate"); + } + + #[cfg(unix)] + #[test] + fn a_socket_left_by_a_dead_daemon_is_replaced_not_refused() { + let dir = tempfile::tempdir().expect("a temp dir"); + let path = dir.path().join(SOCKET_NAME); + // A plain file standing in for what a crash left behind: nothing is + // listening on it, so binding must succeed. + std::fs::write(&path, "").unwrap(); + assert!( + server::bind_unix(&path).is_ok(), + "a stale socket must not strand the daemon" + ); + } + + #[cfg(unix)] + #[test] + fn a_second_daemon_refuses_to_steal_a_live_socket() { + let dir = tempfile::tempdir().expect("a temp dir"); + let path = dir.path().join(SOCKET_NAME); + let _first = server::bind_unix(&path).expect("the first daemon binds"); + assert!( + server::bind_unix(&path).is_err(), + "taking over a live socket would silently steal its clients" + ); + } +} diff --git a/soulseek-rs/src/daemon/proto.rs b/soulseek-rs/src/daemon/proto.rs index 9f46e1d..299097a 100644 --- a/soulseek-rs/src/daemon/proto.rs +++ b/soulseek-rs/src/daemon/proto.rs @@ -881,10 +881,24 @@ pub struct Members { pub users: Vec, } +/// One line of chat history. Unlike a live [`UserMessageDto`] this carries a +/// direction, because a conversation the daemon collected includes what this +/// account sent as well as what it received. +#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)] +#[cfg_attr(test, derive(schemars::JsonSchema))] +pub struct ChatMessageDto { + /// The other party, whichever way the message went. + pub peer: String, + pub outgoing: bool, + pub text: String, + /// Unix seconds. + pub at: i64, +} + #[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)] #[cfg_attr(test, derive(schemars::JsonSchema))] pub struct Messages { - pub messages: Vec, + pub messages: Vec, } #[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)] diff --git a/soulseek-rs/src/daemon/server.rs b/soulseek-rs/src/daemon/server.rs new file mode 100644 index 0000000..05178da --- /dev/null +++ b/soulseek-rs/src/daemon/server.rs @@ -0,0 +1,993 @@ +//! The control listener: who may connect, and what they may ask for. +//! +//! One thread per connection, which is noise next to the thousand-odd peer +//! threads the client already runs. Each connection owns a writer thread fed by +//! a channel, so request replies and pushed events interleave onto the socket +//! without a lock and without either being able to block the other. + +use super::hub::{Hub, PendingBrowses, PendingDownload}; +use super::proto::{ + Ack, AuthParams, AuthResult, CODE_APPLICATION, CODE_INVALID_PARAMS, + ChatMessageDto, + CODE_INVALID_REQUEST, CODE_METHOD_NOT_FOUND, CODE_PARSE, DaemonStatus, + DirectoriesParams, DownloadDto, DownloadStartParams, DownloadStarted, + Downloads, IntervalSeconds, MessageParams, Members, Messages, Method, + OPENRPC, PROTOCOL_VERSION, QueryParams, Request, Response, RoomRef, + RpcError, SayParams, SearchResultDto, SearchResults, Seconds, SharesStatus, + SlotsParams, TransferRef, UploadInfoDto, Uploads, UserInfoDto, UserRef, + UserResult, +}; +use crate::api::SessionApi; +use crate::output::Exit; +use serde::de::DeserializeOwned; +use serde_json::Value; +use std::io::{BufRead, BufReader, Write}; +use std::net::{TcpListener, TcpStream}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc::{Receiver, SyncSender, sync_channel}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +/// How long a failed authentication costs the caller. Not a serious defence — +/// the token is 256 bits — but it keeps a wrong-token loop off the CPU. +const AUTH_FAILURE_DELAY: Duration = Duration::from_secs(1); + +/// Replies waiting to go out on one connection. Bounded, so a client that has +/// stopped reading cannot make the daemon buffer without limit. +const WRITE_QUEUE_DEPTH: usize = 1024; + +/// Everything a connection needs to answer a request. +pub struct Daemon { + pub session: Arc, + pub hub: Arc, + pub browses: Arc, + pub downloads: Arc>>, + /// Where transfers land, on *this* machine. A remote caller cannot see + /// this filesystem, so it does not get to choose. + pub download_dir: String, + pub server: String, + pub token: String, + pub started: Instant, + pub stop: Arc, + /// Rooms this session has joined, so a restart can rejoin them. + pub rooms: Mutex>, +} + +impl Daemon { + /// Answer one request. + /// + /// The match is exhaustive over [`Method`], so a new method cannot be added + /// to the protocol without being implemented here. + #[allow(clippy::too_many_lines)] + pub fn call(&self, method: Method, params: Value) -> Result { + match method { + // Handled during the handshake; reaching here means a second auth. + Method::Auth => ok(Ack::OK), + Method::DaemonStatus => ok(self.status()), + Method::DaemonStop => { + self.stop.store(true, Ordering::Relaxed); + ok(Ack::OK) + } + Method::RpcDiscover => serde_json::from_str(OPENRPC) + .map_err(|e| internal(format!("the contract is unreadable: {e}"))), + + Method::SearchStart => { + let query: QueryParams = parse(params)?; + // Zero window: put the search on the wire and return. The + // caller decides how long to let answers accumulate, which is + // what keeps one client's search from blocking the daemon. + self.session + .search_with_cancel(&query.query, Duration::ZERO, None) + .map_err(|e| failed(Exit::Connection, format!("search failed: {e}")))?; + ok(Ack::OK) + } + Method::SearchResults => { + let query: QueryParams = parse(params)?; + ok(SearchResults { + results: self + .session + .get_search_results(&query.query) + .iter() + .map(SearchResultDto::from) + .collect(), + }) + } + Method::SearchWishlist => { + let query: QueryParams = parse(params)?; + self.session + .start_wishlist_search(&query.query) + .map_err(|e| failed(Exit::Connection, e.to_string()))?; + ok(Ack::OK) + } + Method::SearchWishlistInterval => ok(IntervalSeconds { + seconds: self.session.wishlist_interval().as_secs(), + }), + + Method::DownloadStart => self.start_download(parse(params)?), + Method::DownloadList => ok(Downloads { + downloads: self + .session + .get_all_downloads() + .iter() + .map(DownloadDto::from) + .collect(), + }), + Method::DownloadPause => { + let transfer: TransferRef = parse(params)?; + ok(Ack { + ok: self + .session + .pause_download(&transfer.username, &transfer.filename), + }) + } + Method::DownloadResume => { + let transfer: TransferRef = parse(params)?; + ok(Ack { + ok: self + .session + .resume_download(&transfer.username, &transfer.filename), + }) + } + Method::DownloadRemove => { + let transfer: TransferRef = parse(params)?; + ok(Ack { + ok: self + .session + .remove_download(&transfer.username, &transfer.filename), + }) + } + Method::DownloadRemoveQueued => { + let transfer: TransferRef = parse(params)?; + ok(Ack { + ok: self.session.remove_queued_download( + &transfer.username, + &transfer.filename, + ), + }) + } + + Method::UploadList => ok(Uploads { + uploads: self + .session + .uploads() + .iter() + .map(UploadInfoDto::from) + .collect(), + }), + Method::UploadCancel => { + let transfer: TransferRef = parse(params)?; + ok(Ack { + ok: self + .session + .cancel_upload(&transfer.username, &transfer.filename), + }) + } + Method::UploadSlots => { + let slots: SlotsParams = parse(params)?; + self.session.set_upload_slots(slots.slots); + ok(Ack::OK) + } + + Method::PrivilegesCheck => { + self.session + .check_privileges() + .map_err(|e| failed(Exit::Connection, e.to_string()))?; + ok(Ack::OK) + } + Method::PrivilegesOwn => ok(Seconds { + seconds: self.session.own_privilege_seconds(), + }), + + Method::RoomListRequest => { + self.session + .request_room_list() + .map_err(|e| failed(Exit::Connection, e.to_string()))?; + ok(Ack::OK) + } + Method::RoomJoin => { + let room: RoomRef = parse(params)?; + self.session + .join_room(&room.room) + .map_err(|e| failed(Exit::Connection, e.to_string()))?; + self.joined(&room.room, true); + ok(Ack::OK) + } + Method::RoomLeave => { + let room: RoomRef = parse(params)?; + self.session + .leave_room(&room.room) + .map_err(|e| failed(Exit::Connection, e.to_string()))?; + self.joined(&room.room, false); + ok(Ack::OK) + } + Method::RoomSay => { + let say: SayParams = parse(params)?; + self.session + .say_in_room(&say.room, &say.message) + .map_err(|e| failed(Exit::Connection, e.to_string()))?; + ok(Ack::OK) + } + Method::RoomMembers => { + let room: RoomRef = parse(params)?; + ok(Members { + users: self.session.room_members(&room.room), + }) + } + + Method::MessageSend => { + let message: MessageParams = parse(params)?; + self.session + .send_private_message(&message.username, &message.message) + .map_err(|e| failed(Exit::Connection, e.to_string()))?; + // Record our own half of the conversation: the drainer only + // ever sees incoming messages, so without this a client that + // attaches later reads a one-sided chat. + self.hub.remember(ChatMessageDto { + peer: message.username, + outgoing: true, + text: message.message, + at: now_seconds(), + }); + ok(Ack::OK) + } + Method::MessageHistory => ok(Messages { + messages: self.hub.history(), + }), + + Method::BrowseUser => { + let user: UserRef = parse(params)?; + // Register before asking: the answer can land on the very next + // drainer tick, and an unexpected listing is discarded. + self.browses.expect(&user.username); + if let Err(e) = self.session.browse_user(&user.username) { + self.browses.forget(&user.username); + return Err(failed(Exit::Connection, e.to_string())); + } + ok(Ack::OK) + } + Method::UserRequest => { + let user: UserRef = parse(params)?; + self.session + .request_user_info(&user.username) + .map_err(|e| failed(Exit::Connection, e.to_string()))?; + ok(Ack::OK) + } + Method::UserInfoOf => { + let user: UserRef = parse(params)?; + ok(UserResult { + user: self + .session + .user_info(&user.username) + .as_ref() + .map(UserInfoDto::from), + }) + } + + Method::SharesStatusOf => ok(self.shares()), + Method::SharesSet => { + let directories: DirectoriesParams = parse(params)?; + self.session + .set_shared_directories(directories.directories) + .map_err(|e| failed(Exit::Usage, e.to_string()))?; + ok(self.shares()) + } + Method::SharesReindex => { + self.session + .set_shared_directories(self.session.shared_directories()) + .map_err(|e| failed(Exit::Usage, e.to_string()))?; + ok(self.shares()) + } + } + } + + /// Track membership so a restart can rejoin what was open. The library + /// has no "which rooms am I in" accessor, so the daemon keeps the list. + fn joined(&self, room: &str, member: bool) { + let Ok(mut rooms) = self.rooms.lock() else { + return; + }; + rooms.retain(|name| name != room); + if member { + rooms.push(room.to_string()); + } + } + + #[must_use] + pub fn rooms(&self) -> Vec { + self.rooms + .lock() + .map_or_else(|_| Vec::new(), |rooms| rooms.clone()) + } + + fn start_download( + &self, + params: DownloadStartParams, + ) -> Result { + let directory = params + .download_directory + .unwrap_or_else(|| self.download_dir.clone()); + let (download, updates) = self + .session + .download_with_metadata( + params.filename.clone(), + params.username.clone(), + params.size, + directory, + params.metadata.into(), + ) + .map_err(|e| failed(Exit::Transfer, format!("cannot start: {e}")))?; + + if let Ok(mut pending) = self.downloads.lock() { + pending.push(PendingDownload { + username: params.username, + filename: params.filename, + updates, + }); + } + ok(DownloadStarted { + download: DownloadDto::from(&download), + }) + } + + fn shares(&self) -> SharesStatus { + let (folders, files) = self.session.shared_counts(); + SharesStatus { + folders, + files, + directories: self.session.shared_directories(), + } + } + + fn status(&self) -> DaemonStatus { + let (folders, files) = self.session.shared_counts(); + DaemonStatus { + username: self.session.username(), + server: self.server.clone(), + daemon_version: env!("CARGO_PKG_VERSION").to_string(), + protocol: PROTOCOL_VERSION, + listen_port: self.session.listen_port(), + shared_folders: folders, + shared_files: files, + download_dir: self.download_dir.clone(), + session_loss: self.session.session_loss().map(Into::into), + clients: self.hub.subscribers(), + uptime_secs: self.started.elapsed().as_secs(), + } + } +} + +fn ok(value: T) -> Result { + serde_json::to_value(value) + .map_err(|e| internal(format!("cannot serialize the reply: {e}"))) +} + +fn parse(params: Value) -> Result { + serde_json::from_value(params) + .map_err(|e| RpcError::new(CODE_INVALID_PARAMS, e.to_string())) +} + +fn failed(exit: Exit, message: impl Into) -> RpcError { + RpcError::application(exit, message) +} + +fn internal(message: impl Into) -> RpcError { + RpcError::new(CODE_APPLICATION, message) +} + +/// Wall-clock seconds, to stamp a message the way the server stamps the ones +/// it delivers. A clock before the epoch reads as zero rather than panicking. +fn now_seconds() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |since| since.as_secs() as i64) +} + +// --------------------------------------------------------------------------- +// Transports +// --------------------------------------------------------------------------- + +/// Where control connections arrive. +/// +/// The Unix socket is the local path — its file permissions are the +/// authentication, the same bargain Docker and ssh-agent make. TCP is opt-in +/// and always needs the token. +pub enum Listener { + #[cfg(unix)] + Unix(std::os::unix::net::UnixListener), + Tcp(TcpListener), +} + +pub enum Stream { + #[cfg(unix)] + Unix(std::os::unix::net::UnixStream), + Tcp(TcpStream), +} + +impl Stream { + fn try_clone(&self) -> std::io::Result { + match self { + #[cfg(unix)] + Self::Unix(stream) => stream.try_clone().map(Self::Unix), + Self::Tcp(stream) => stream.try_clone().map(Self::Tcp), + } + } + + /// Whether the transport itself proves who is calling. + const fn trusted_by_transport(&self) -> bool { + match self { + #[cfg(unix)] + Self::Unix(_) => true, + Self::Tcp(_) => false, + } + } +} + +impl std::io::Read for Stream { + fn read(&mut self, buffer: &mut [u8]) -> std::io::Result { + match self { + #[cfg(unix)] + Self::Unix(stream) => stream.read(buffer), + Self::Tcp(stream) => stream.read(buffer), + } + } +} + +impl Write for Stream { + fn write(&mut self, buffer: &[u8]) -> std::io::Result { + match self { + #[cfg(unix)] + Self::Unix(stream) => stream.write(buffer), + Self::Tcp(stream) => stream.write(buffer), + } + } + + fn flush(&mut self) -> std::io::Result<()> { + match self { + #[cfg(unix)] + Self::Unix(stream) => stream.flush(), + Self::Tcp(stream) => stream.flush(), + } + } +} + +impl Listener { + /// Accept forever, handing each connection to its own thread. + pub fn serve(self, daemon: Arc) { + match self { + #[cfg(unix)] + Self::Unix(listener) => { + for stream in listener.incoming() { + match stream { + Ok(stream) => spawn_connection( + Stream::Unix(stream), + Arc::clone(&daemon), + ), + Err(e) => { + soulseek_rs::warn!("control accept failed: {e}"); + } + } + } + } + Self::Tcp(listener) => { + for stream in listener.incoming() { + match stream { + Ok(stream) => spawn_connection( + Stream::Tcp(stream), + Arc::clone(&daemon), + ), + Err(e) => { + soulseek_rs::warn!("control accept failed: {e}"); + } + } + } + } + } + } +} + +fn spawn_connection(stream: Stream, daemon: Arc) { + std::thread::spawn(move || { + if let Err(e) = handle(stream, daemon) { + soulseek_rs::debug!("control connection ended: {e}"); + } + }); +} + +/// Serve one connection: authenticate, then answer requests until it closes. +fn handle(stream: Stream, daemon: Arc) -> std::io::Result<()> { + let reader = BufReader::new(stream.try_clone()?); + let trusted = stream.trusted_by_transport(); + + let (out, outbox) = sync_channel::(WRITE_QUEUE_DEPTH); + let writer = std::thread::spawn(move || write_loop(stream, &outbox)); + + let mut lines = reader.lines(); + let authenticated = authenticate(&mut lines, &daemon, trusted, &out); + if !authenticated { + drop(out); + let _ = writer.join(); + return Ok(()); + } + + // Only now does this connection get to see events: an unauthenticated + // socket must learn nothing at all. + let (subscription, events) = daemon.hub.subscribe(); + let pump = { + let out = out.clone(); + std::thread::spawn(move || { + for event in events { + if out.send(event).is_err() { + break; + } + } + }) + }; + + for line in lines { + let line = line?; + if line.trim().is_empty() { + continue; + } + let daemon = Arc::clone(&daemon); + let out = out.clone(); + // One thread per request: a slow method must not hold up the ones + // behind it, and the ids let a client match replies out of order. + std::thread::spawn(move || { + if let Some(response) = answer(&line, &daemon) { + let _ = out.send(response); + } + }); + } + + daemon.hub.unsubscribe(subscription); + drop(out); + let _ = pump.join(); + let _ = writer.join(); + Ok(()) +} + +/// The first request has to be `auth`, and has to be good. +fn authenticate( + lines: &mut std::io::Lines>, + daemon: &Daemon, + trusted: bool, + out: &SyncSender, +) -> bool { + let Some(Ok(line)) = lines.next() else { + return false; + }; + let request: Request = match serde_json::from_str(&line) { + Ok(request) => request, + Err(e) => { + let _ = out.send(Response::failure( + None, + RpcError::new(CODE_PARSE, e.to_string()), + )); + return false; + } + }; + + if Method::from_wire(&request.method) != Some(Method::Auth) { + let _ = out.send(Response::failure( + request.id, + RpcError::new( + CODE_INVALID_REQUEST, + "the first request on a connection must be `auth`", + ), + )); + return false; + } + + let params: AuthParams = + serde_json::from_value(request.params).unwrap_or_default(); + if !accepts(daemon, trusted, params.token.as_deref()) { + std::thread::sleep(AUTH_FAILURE_DELAY); + let _ = out.send(Response::failure( + request.id, + RpcError::application(Exit::Connection, "authentication failed"), + )); + return false; + } + + let result = serde_json::to_value(AuthResult { + protocol: PROTOCOL_VERSION, + daemon_version: env!("CARGO_PKG_VERSION").to_string(), + username: daemon.session.username(), + }) + .unwrap_or(Value::Null); + let _ = out.send(Response::result(request.id, result)); + true +} + +/// A Unix socket has already proved the caller is this user, so a token is +/// optional there and mandatory everywhere else. A token that *is* offered is +/// always checked, so a wrong one never passes. +fn accepts(daemon: &Daemon, trusted: bool, token: Option<&str>) -> bool { + match token { + Some(token) => constant_time_eq(token, &daemon.token), + None => trusted, + } +} + +/// Compare without leaking the position of the first difference through timing. +fn constant_time_eq(left: &str, right: &str) -> bool { + if left.len() != right.len() { + return false; + } + left.bytes() + .zip(right.bytes()) + .fold(0u8, |difference, (a, b)| difference | (a ^ b)) + == 0 +} + +/// Turn one request line into the reply it deserves, or `None` for a +/// notification, which by JSON-RPC rule is answered with silence. +fn answer(line: &str, daemon: &Daemon) -> Option { + let request: Request = match serde_json::from_str(line) { + Ok(request) => request, + Err(e) => { + return Some(Response::failure( + None, + RpcError::new(CODE_PARSE, e.to_string()), + )); + } + }; + let id = request.id.clone(); + let Some(method) = Method::from_wire(&request.method) else { + return id.is_some().then(|| { + Response::failure( + request.id, + RpcError::new( + CODE_METHOD_NOT_FOUND, + format!("no such method: {}", request.method), + ), + ) + }); + }; + + let outcome = daemon.call(method, request.params); + id.map(|id| match outcome { + Ok(result) => Response::result(Some(id), result), + Err(error) => Response::failure(Some(id), error), + }) +} + +fn write_loop(stream: Stream, outbox: &Receiver) { + let mut stream = stream; + for response in outbox { + let Ok(line) = serde_json::to_string(&response) else { + continue; + }; + if writeln!(stream, "{line}").is_err() || stream.flush().is_err() { + return; + } + } +} + +/// Bind the TCP control port, refusing to come up rather than silently +/// listening somewhere unintended. +pub fn bind_tcp(address: &str) -> std::io::Result { + TcpListener::bind(address).map(Listener::Tcp) +} + +/// Bind the Unix control socket with permissions that make it the local +/// authentication: only this user can open it. +#[cfg(unix)] +pub fn bind_unix(path: &std::path::Path) -> std::io::Result { + use std::os::unix::fs::PermissionsExt; + + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent)?; + } + // A socket file outlives the process that made it, so a previous crash + // would otherwise make the daemon unstartable. + if path.exists() { + remove_stale_socket(path)?; + } + let listener = std::os::unix::net::UnixListener::bind(path)?; + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))?; + Ok(Listener::Unix(listener)) +} + +/// Remove a socket file left by a dead daemon — but never one a live daemon is +/// still serving, which would silently steal its clients. +#[cfg(unix)] +fn remove_stale_socket(path: &std::path::Path) -> std::io::Result<()> { + match std::os::unix::net::UnixStream::connect(path) { + Ok(_) => Err(std::io::Error::new( + std::io::ErrorKind::AddrInUse, + format!("a daemon is already listening on {}", path.display()), + )), + Err(_) => std::fs::remove_file(path), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn a_unix_socket_needs_no_token_but_tcp_does() { + let daemon = daemon_with_token("secret"); + assert!(accepts(&daemon, true, None), "socket permissions are the auth"); + assert!(!accepts(&daemon, false, None), "TCP must demand a token"); + } + + #[test] + fn a_wrong_token_is_refused_on_every_transport() { + let daemon = daemon_with_token("secret"); + assert!(!accepts(&daemon, false, Some("guess"))); + assert!( + !accepts(&daemon, true, Some("guess")), + "a token that is offered is always checked, even locally" + ); + assert!(accepts(&daemon, false, Some("secret"))); + } + + #[test] + fn token_comparison_does_not_short_circuit_on_length_or_content() { + assert!(constant_time_eq("abc", "abc")); + assert!(!constant_time_eq("abc", "abd")); + assert!(!constant_time_eq("abc", "abcd")); + assert!(!constant_time_eq("", "a")); + assert!(constant_time_eq("", "")); + } + + #[test] + fn an_unknown_method_is_reported_rather_than_ignored() { + let daemon = daemon_with_token("t"); + let response = answer( + r#"{"jsonrpc":"2.0","id":1,"method":"search.nope"}"#, + &daemon, + ) + .expect("a request with an id is always answered"); + let error = response.error.expect("an unknown method is an error"); + assert_eq!(error.code, CODE_METHOD_NOT_FOUND); + } + + #[test] + fn a_notification_gets_no_reply_even_when_it_fails() { + let daemon = daemon_with_token("t"); + assert!( + answer(r#"{"jsonrpc":"2.0","method":"search.nope"}"#, &daemon) + .is_none(), + "a request without an id must be answered with silence" + ); + } + + #[test] + fn unparseable_input_is_a_parse_error_not_a_dropped_connection() { + let daemon = daemon_with_token("t"); + let response = + answer("this is not json", &daemon).expect("garbage is answered"); + assert_eq!(response.error.expect("an error").code, CODE_PARSE); + } + + #[test] + fn bad_params_are_reported_as_such() { + let daemon = daemon_with_token("t"); + let error = daemon + .call(Method::RoomJoin, json!({ "not_a_room": 1 })) + .expect_err("the params do not match"); + assert_eq!(error.code, CODE_INVALID_PARAMS); + } + + #[test] + fn discover_serves_the_published_contract() { + let daemon = daemon_with_token("t"); + let document = daemon + .call(Method::RpcDiscover, Value::Null) + .expect("the contract is compiled in"); + assert_eq!(document["openrpc"], "1.3.2"); + assert!( + document["methods"] + .as_array() + .is_some_and(|methods| methods.len() == Method::ALL.len() + + super::super::proto::Event::ALL.len()), + "discover must describe every method and event" + ); + } + + #[test] + fn stop_is_recorded_so_the_daemon_can_wind_down() { + let daemon = daemon_with_token("t"); + assert!(!daemon.stop.load(Ordering::Relaxed)); + daemon + .call(Method::DaemonStop, Value::Null) + .expect("stop is always accepted"); + assert!(daemon.stop.load(Ordering::Relaxed)); + } + + // A session that answers nothing: enough to exercise dispatch, parameter + // validation, and the auth gate without a Soulseek server. + struct SilentSession; + + fn daemon_with_token(token: &str) -> Daemon { + Daemon { + session: Arc::new(SilentSession), + hub: Arc::new(Hub::new()), + browses: Arc::new(PendingBrowses::default()), + downloads: Arc::new(Mutex::new(Vec::new())), + download_dir: "/tmp".to_string(), + server: "server.invalid:2416".to_string(), + token: token.to_string(), + started: Instant::now(), + stop: Arc::new(AtomicBool::new(false)), + rooms: Mutex::new(Vec::new()), + } + } + + impl SessionApi for SilentSession { + fn username(&self) -> String { + "tester".to_string() + } + fn listen_port(&self) -> Option { + None + } + fn session_loss(&self) -> Option { + None + } + fn search( + &self, + _query: &str, + _timeout: Duration, + ) -> soulseek_rs::Result> { + Ok(Vec::new()) + } + fn search_with_cancel( + &self, + _query: &str, + _timeout: Duration, + _cancel: Option>, + ) -> soulseek_rs::Result> { + Ok(Vec::new()) + } + fn get_search_results( + &self, + _key: &str, + ) -> Vec { + Vec::new() + } + fn get_search_results_count(&self, _key: &str) -> usize { + 0 + } + fn try_get_search_results( + &self, + _key: &str, + ) -> Option> { + None + } + fn start_wishlist_search(&self, _query: &str) -> soulseek_rs::Result<()> { + Ok(()) + } + fn wishlist_interval(&self) -> Duration { + Duration::from_mins(12) + } + fn download( + &self, + filename: String, + username: String, + size: u64, + directory: String, + ) -> soulseek_rs::Result<( + soulseek_rs::types::Download, + Receiver, + )> { + self.download_with_metadata( + filename, + username, + size, + directory, + soulseek_rs::types::DownloadMetadata::default(), + ) + } + fn download_with_metadata( + &self, + _filename: String, + _username: String, + _size: u64, + _directory: String, + _metadata: soulseek_rs::types::DownloadMetadata, + ) -> soulseek_rs::Result<( + soulseek_rs::types::Download, + Receiver, + )> { + Err(soulseek_rs::SoulseekRs::NotConnected) + } + fn get_all_downloads(&self) -> Vec { + Vec::new() + } + fn pause_download(&self, _username: &str, _filename: &str) -> bool { + false + } + fn resume_download(&self, _username: &str, _filename: &str) -> bool { + false + } + fn remove_queued_download( + &self, + _username: &str, + _filename: &str, + ) -> bool { + false + } + fn remove_download(&self, _username: &str, _filename: &str) -> bool { + false + } + fn uploads(&self) -> Vec { + Vec::new() + } + fn take_upload_events(&self) -> Vec { + Vec::new() + } + fn cancel_upload(&self, _username: &str, _filename: &str) -> bool { + false + } + fn set_upload_slots(&self, _slots: usize) {} + fn check_privileges(&self) -> soulseek_rs::Result<()> { + Ok(()) + } + fn own_privilege_seconds(&self) -> Option { + None + } + fn request_room_list(&self) -> soulseek_rs::Result<()> { + Ok(()) + } + fn join_room(&self, _room: &str) -> soulseek_rs::Result<()> { + Ok(()) + } + fn leave_room(&self, _room: &str) -> soulseek_rs::Result<()> { + Ok(()) + } + fn say_in_room( + &self, + _room: &str, + _message: &str, + ) -> soulseek_rs::Result<()> { + Ok(()) + } + fn room_members(&self, _room: &str) -> Vec { + Vec::new() + } + fn take_room_events(&self) -> Vec { + Vec::new() + } + fn send_private_message( + &self, + _username: &str, + _message: &str, + ) -> soulseek_rs::Result<()> { + Ok(()) + } + fn take_private_messages(&self) -> Vec { + Vec::new() + } + fn browse_user(&self, _username: &str) -> soulseek_rs::Result<()> { + Ok(()) + } + fn take_browse_result( + &self, + _username: &str, + ) -> Option> { + None + } + fn request_user_info(&self, _username: &str) -> soulseek_rs::Result<()> { + Ok(()) + } + fn user_info(&self, _username: &str) -> Option { + None + } + fn shared_counts(&self) -> (u32, u32) { + (0, 0) + } + fn shared_directories(&self) -> Vec { + Vec::new() + } + fn set_shared_directories( + &self, + _directories: Vec, + ) -> soulseek_rs::Result<()> { + Ok(()) + } + } +} diff --git a/soulseek-rs/src/lib.rs b/soulseek-rs/src/lib.rs index db24d15..fab2657 100644 --- a/soulseek-rs/src/lib.rs +++ b/soulseek-rs/src/lib.rs @@ -3,7 +3,11 @@ pub mod api; pub mod cli; +pub mod commands; pub mod daemon; +pub mod directories; pub mod models; pub mod output; pub mod persist; +pub mod port_mapping; +pub mod remote; diff --git a/soulseek-rs/src/remote/mod.rs b/soulseek-rs/src/remote/mod.rs new file mode 100644 index 0000000..179adb7 --- /dev/null +++ b/soulseek-rs/src/remote/mod.rs @@ -0,0 +1 @@ +//! placeholder