From b82b03a143ef881894734d4ec55bf21154993052 Mon Sep 17 00:00:00 2001 From: michel Date: Sun, 19 Jul 2026 17:26:52 +0200 Subject: [PATCH] feat: firewalled peer connections via server-brokered connect MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement the FIREWALLED connection mode: when a direct outbound connection to a peer fails (peer unreachable), fall back to asking the server to broker it. - Add MessageFactory::build_connect_to_peer (server code 18). - A pre-established outbound failure now emits ClientOperation::PeerConnectFailed instead of a normal disconnect; the client registers a correlation token and sends ConnectToPeer so the peer connects back. - The listener accepts the peer's PierceFirewall (peer code 0) connect-back, matches the pending token, registers it as the P control connection, and fires PeerConnected to flush the queued download. Fix a latent aliasing bug this surfaced: a peer actor could emit two terminal notifications (a connect-failure, then a disconnect when later replaced and stopped), and the second removed the *newer* actor registered under the same username, killing the brokered connection mid-transfer. An actor now reports a terminal outcome at most once (disconnect_reported guard). Verified end to end against soulfind: the mock advertises an unlistened port so the direct connect is refused, the server brokers the connection, the mock pierces back, and the file transfers — bytes match. A companion test (soulfind_brokers_connect_to_peer_between_users) pins the server behaviour the mode relies on. --- soulseek-rs-lib/src/actor/peer_actor.rs | 36 ++- soulseek-rs-lib/src/client.rs | 62 ++++- .../src/message/server/message_factory.rs | 16 ++ soulseek-rs-lib/src/peer/listen.rs | 58 +++- soulseek-rs-lib/tests/e2e.rs | 260 ++++++++++++++++++ 5 files changed, 427 insertions(+), 5 deletions(-) diff --git a/soulseek-rs-lib/src/actor/peer_actor.rs b/soulseek-rs-lib/src/actor/peer_actor.rs index a03a79a..d902ccf 100644 --- a/soulseek-rs-lib/src/actor/peer_actor.rs +++ b/soulseek-rs-lib/src/actor/peer_actor.rs @@ -53,6 +53,14 @@ pub struct PeerActor { /// True when we initiated this connection (no stream supplied at /// construction), so we must send a `PeerInit` once connected. outbound: bool, + /// Set once the connection is up. A pre-established failure on an outbound + /// connection means the peer is unreachable (likely firewalled). + established: bool, + /// Set once this actor has reported a terminal outcome (disconnect or + /// connect-failure). Guards against a second notification — e.g. when a + /// replaced actor is later stopped — which would otherwise alias and evict + /// a newer actor registered under the same username. + disconnect_reported: bool, } impl PeerActor { @@ -83,6 +91,8 @@ impl PeerActor { queued_messages: Vec::new(), own_username, outbound, + established: false, + disconnect_reported: false, } } @@ -429,9 +439,22 @@ impl PeerActor { self.stream.take(); - if let Err(e) = self.client_channel.send( - ClientOperation::PeerDisconnected(username, Some(error.into())), - ) { + if self.disconnect_reported { + return; + } + self.disconnect_reported = true; + + // A direct outbound connection that never established means the peer is + // unreachable (likely firewalled): signal a connect failure so the + // client can fall back to server-brokered connect. Anything else is a + // normal disconnect. + let op = if self.outbound && !self.established { + ClientOperation::PeerConnectFailed(username) + } else { + ClientOperation::PeerDisconnected(username, Some(error.into())) + }; + + if let Err(e) = self.client_channel.send(op) { error!("Failed to send disconnect notification: {}", e); } } @@ -441,6 +464,11 @@ impl PeerActor { self.stream.take(); + if self.disconnect_reported { + return; + } + self.disconnect_reported = true; + if let Err(e) = self .client_channel .send(ClientOperation::PeerDisconnected(username, None)) @@ -552,6 +580,8 @@ impl PeerActor { } }; + self.established = true; + let Some(ref mut stream) = self.stream else { return; }; diff --git a/soulseek-rs-lib/src/client.rs b/soulseek-rs-lib/src/client.rs index 83099f3..3a55f03 100644 --- a/soulseek-rs-lib/src/client.rs +++ b/soulseek-rs-lib/src/client.rs @@ -18,7 +18,7 @@ use std::{ net::TcpStream, sync::{ RwLock, - atomic::{AtomicBool, Ordering}, + atomic::{AtomicBool, AtomicU32, Ordering}, mpsc::{Receiver, Sender}, }, thread::{self, sleep}, @@ -31,6 +31,13 @@ use std::{ use crate::{debug, error, info, trace, warn}; const DEFAULT_LISTEN_PORT: u16 = 2234; +/// Source of non-zero correlation tokens for server-brokered connections. +static NEXT_CONNECT_TOKEN: AtomicU32 = AtomicU32::new(1); + +fn next_connect_token() -> u32 { + NEXT_CONNECT_TOKEN.fetch_add(1, Ordering::Relaxed).max(1) +} + #[derive(Debug, Clone)] pub struct ClientSettings { pub username: String, @@ -98,6 +105,10 @@ pub enum ClientOperation { SetServerSender(Sender), PrivateMessageReceived(UserMessage), PeerConnected(String), + /// A direct outbound connection to this peer failed before it was + /// established — the peer is likely firewalled, so fall back to asking the + /// server to broker the connection. + PeerConnectFailed(String), } pub struct ClientContext { pub peer_registry: Option, @@ -106,6 +117,9 @@ pub struct ClientContext { server_sender: Option>, searches: HashMap, private_messages: Vec, + /// Correlation tokens for server-brokered (firewalled) connections, mapping + /// a token we sent in a ConnectToPeer to the peer we expect back. + pending_connect_tokens: HashMap, actor_system: Arc, } impl Default for ClientContext { @@ -333,11 +347,23 @@ impl ClientContext { server_sender: None, searches: HashMap::new(), private_messages: Vec::new(), + pending_connect_tokens: HashMap::new(), downloads: DownloadStore::new(), actor_system, } } + /// Remember that a server-brokered connection to `username` is pending under + /// `token`; the peer will quote it back in a PierceFirewall. + pub fn add_pending_connect(&mut self, token: u32, username: String) { + self.pending_connect_tokens.insert(token, username); + } + + /// Resolve and consume the peer expected for a brokered connection `token`. + pub fn take_pending_connect(&mut self, token: u32) -> Option { + self.pending_connect_tokens.remove(&token) + } + /// Record a private message received from another user. pub fn push_private_message(&mut self, message: UserMessage) { self.private_messages.push(message); @@ -1277,6 +1303,40 @@ impl Client { } } } + ClientOperation::PeerConnectFailed(username) => { + // Direct connect failed: ask the server to + // broker it. Register a correlation token, then + // send ConnectToPeer so the (firewalled) peer + // connects back to our listener quoting it. + let token = next_connect_token(); + let server_sender = match client_context + .write_safe() + { + Ok(mut ctx) => { + ctx.add_pending_connect( + token, + username.clone(), + ); + ctx.server_sender.clone() + } + Err(e) => { + error!( + "[client] PeerConnectFailed write: {}", + e + ); + continue; + } + }; + if let Some(sender) = server_sender { + let msg = crate::message::server::MessageFactory::build_connect_to_peer( + token, + &username, + ConnectionType::P, + ); + let _ = sender + .send(ServerMessage::SendMessage(msg)); + } + } } } Err(e) => { diff --git a/soulseek-rs-lib/src/message/server/message_factory.rs b/soulseek-rs-lib/src/message/server/message_factory.rs index d8748ed..1ee490d 100644 --- a/soulseek-rs-lib/src/message/server/message_factory.rs +++ b/soulseek-rs-lib/src/message/server/message_factory.rs @@ -80,6 +80,22 @@ impl MessageFactory { Message::new().write_int32(23).write_int32(id).clone() } + /// Ask the server (code 18) to broker a connection to a firewalled peer: + /// the server tells that peer to connect back to us, quoting `token`. + #[must_use] + pub fn build_connect_to_peer( + token: u32, + username: &str, + connection_type: ConnectionType, + ) -> Message { + Message::new() + .write_int32(18) + .write_int32(token) + .write_string(username) + .write_string(&connection_type.to_string()) + .clone() + } + #[must_use] pub fn build_set_status_message(status_code: u32) -> Message { Message::new() diff --git a/soulseek-rs-lib/src/peer/listen.rs b/soulseek-rs-lib/src/peer/listen.rs index 254ad95..bb0cd86 100644 --- a/soulseek-rs-lib/src/peer/listen.rs +++ b/soulseek-rs-lib/src/peer/listen.rs @@ -16,7 +16,6 @@ const PEER_INIT_MESSAGE_CODE: u8 = 1; #[derive(Clone)] struct ConnectionContext { - #[allow(dead_code)] client_sender: Sender, client_context: Arc>, own_username: String, @@ -227,6 +226,54 @@ fn handle_file_connection( } } +/// Handle a `PierceFirewall` (peer init code 0): a peer we asked the server to +/// broker (because a direct connection failed) connecting back to us. It quotes +/// the correlation token we registered; we register the stream as its P control +/// connection and tell the client the connection is live. +fn handle_pierce_firewall( + mut message: Message, + stream: TcpStream, + reader: MessageReader, + context: &ConnectionContext, + peer_ip: &str, + peer_port: u16, +) { + message.set_pointer(5); // skip length prefix (4) + int8 code (1) + let token = message.read_int32(); + + let username = match context.client_context.write_safe() { + Ok(mut ctx) => ctx.take_pending_connect(token), + Err(e) => { + error!("[listener] pierce firewall lock: {}", e); + return; + } + }; + let Some(username) = username else { + debug!( + "[listener:{peer_ip}:{peer_port}] PierceFirewall token {token} is not pending; ignoring" + ); + return; + }; + + let peer = Peer::new( + username.clone(), + ConnectionType::P, + peer_ip.to_string(), + peer_port.into(), + None, + 0, + 0, + 0, + ); + handle_peer_connection(peer, stream, reader, context, peer_ip, peer_port); + + // Inbound peers don't self-announce, so nudge the client to flush any + // downloads queued for this now-connected peer. + let _ = context + .client_sender + .send(ClientOperation::PeerConnected(username)); +} + fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) { let Ok(peer_addr) = stream.peer_addr() else { error!("[listener] failed to get peer address"); @@ -245,6 +292,15 @@ fn handle_incoming_connection(stream: TcpStream, context: ConnectionContext) { return; }; + // A firewalled peer brokered through the server connects back with a + // PierceFirewall (code 0) instead of a PeerInit (code 1). + if message.get_message_code() == 0 { + handle_pierce_firewall( + message, stream, reader, &context, &peer_ip, peer_port, + ); + return; + } + let Some(init_data) = parse_peer_init_message(message) else { error!( "[listener:{peer_ip}:{peer_port}] Invalid or unknown peer init message" diff --git a/soulseek-rs-lib/tests/e2e.rs b/soulseek-rs-lib/tests/e2e.rs index ef9c2e0..c20e847 100644 --- a/soulseek-rs-lib/tests/e2e.rs +++ b/soulseek-rs-lib/tests/e2e.rs @@ -27,6 +27,7 @@ use std::time::{Duration, Instant}; use soulseek_rs::message::Message; use soulseek_rs::message::server::MessageFactory; +use soulseek_rs::peer::ConnectionType; use soulseek_rs::{Client, ClientSettings, DownloadStatus, PeerAddress}; /// A Soulseek server to test against: either a child soulfind process we @@ -487,6 +488,60 @@ fn serve_file_over_f( Ok(()) } +/// Log a raw socket in to the server and drain up to the login response, +/// returning the still-open stream (the user stays online while it lives). +fn login_raw( + server_addr: &str, + username: &str, + password: &str, +) -> std::io::Result { + let mut srv = connect_retry(server_addr, Duration::from_secs(5))?; + srv.set_read_timeout(Some(Duration::from_secs(10)))?; + srv.write_all( + &MessageFactory::build_login_message(username, password).get_buffer(), + )?; + srv.flush()?; + loop { + let msg = read_framed(&mut srv)?; + if msg.get_message_code() == 1 { + break; + } + } + Ok(srv) +} + +/// Read framed messages from `stream` until one has `code`, or the deadline +/// passes. Returns the matching message. +fn read_until_code( + stream: &mut TcpStream, + code: u8, + timeout: Duration, +) -> Option { + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + match read_framed(stream) { + Ok(msg) if msg.get_message_code() == code => return Some(msg), + Ok(_) => {} + Err(_) => return None, + } + } + None +} + +/// Like [`read_until_code`] but turns a miss into an `io::Error` for `?`. +fn expect_code( + stream: &mut TcpStream, + code: u8, + timeout: Duration, +) -> std::io::Result { + read_until_code(stream, code, timeout).ok_or_else(|| { + std::io::Error::new( + std::io::ErrorKind::TimedOut, + format!("timed out waiting for message code {code}"), + ) + }) +} + fn unique_download_dir() -> PathBuf { let dir = std::env::temp_dir().join(format!( "soulseek-e2e-dl-{}-{:?}", @@ -584,6 +639,52 @@ fn a_file_downloads_from_a_peer_over_p_and_f_connections() { let _ = std::fs::remove_dir_all(&download_dir); } +#[test] +fn soulfind_brokers_connect_to_peer_between_users() { + // The firewalled download mode relies on the server forwarding a + // ConnectToPeer request to the target user. Confirm soulfind does this: a + // requester (with a wait port so the server knows its address) asks the + // server to broker a connection to an online target, and the target must + // receive a forwarded ConnectToPeer (server code 18) naming the requester. + let server = server_or_skip!(); + let addr = format!("{}:{}", server.host, server.port); + + let mut target = + login_raw(&addr, "e2e_broker_target", "pw").expect("target login"); + + let mut requester = + login_raw(&addr, "e2e_broker_req", "pw").expect("requester login"); + let req_port = free_port().expect("free port"); + requester + .write_all( + &MessageFactory::build_set_wait_port_message(req_port).get_buffer(), + ) + .expect("set wait port"); + requester.flush().expect("flush wait port"); + + let token = 987_654_u32; + requester + .write_all( + &MessageFactory::build_connect_to_peer( + token, + "e2e_broker_target", + ConnectionType::P, + ) + .get_buffer(), + ) + .expect("send ConnectToPeer"); + requester.flush().expect("flush ConnectToPeer"); + + let mut brokered = read_until_code(&mut target, 18, Duration::from_secs(5)) + .expect("target should receive a brokered ConnectToPeer"); + brokered.set_pointer(8); + assert_eq!( + brokered.read_string(), + "e2e_broker_req", + "the brokered message should name the requester" + ); +} + // --------------------------------------------------------------------------- // Direct-connection download: the client initiates the peer connection. // @@ -783,3 +884,162 @@ fn a_file_downloads_from_a_peer_via_direct_connection() { let _ = std::fs::remove_dir_all(&download_dir); } + +// --------------------------------------------------------------------------- +// Firewalled download: the peer is unreachable directly, so the connection is +// brokered through the server. +// +// The mock advertises a port nobody listens on, so the downloader's direct +// connection fails. The client then asks the server to broker the connection +// (ConnectToPeer); the mock, reading its server stream, sees the forwarded +// request and connects back to the downloader with a PierceFirewall. That +// pierced connection becomes the P control channel, and the file is served. +// --------------------------------------------------------------------------- + +struct MockFirewalledUpload { + server_addr: String, + username: String, + password: String, + bogus_port: u16, + downloader_listen_addr: String, + filename: String, + content: Vec, + token: u32, + ready: Sender<()>, +} + +fn run_mock_firewalled_peer(cfg: &MockFirewalledUpload) -> std::io::Result<()> { + // 1. Log in and advertise a port that nobody listens on, so the downloader's + // direct connection is refused and it falls back to server brokering. + let mut srv = login_raw(&cfg.server_addr, &cfg.username, &cfg.password)?; + srv.write_all( + &MessageFactory::build_set_wait_port_message(cfg.bogus_port) + .get_buffer(), + )?; + srv.flush()?; + let _ = cfg.ready.send(()); + + // 2. Wait for the server-brokered ConnectToPeer (server code 18) and read + // the correlation token (after username, type, ip and port). + let mut ctp = expect_code(&mut srv, 18, Duration::from_secs(15))?; + ctp.set_pointer(8); + let _who = ctp.read_string(); + let _conn_type = ctp.read_string(); + let _ip = ctp.read_int32(); + let _port = ctp.read_int32(); + let connect_token = ctp.read_int32(); + + // 3. Connect back to the downloader with a PierceFirewall (peer code 0); + // this becomes the P control connection. + let mut p = + connect_retry(&cfg.downloader_listen_addr, Duration::from_secs(5))?; + p.set_read_timeout(Some(Duration::from_secs(10)))?; + p.write_all( + &MessageFactory::build_pierce_firewall_message(connect_token) + .get_buffer(), + )?; + p.flush()?; + + // 4. Negotiate the transfer over the pierced connection. + let _queue = expect_code(&mut p, 43, Duration::from_secs(10))?; + let mut tr = Message::new(); + tr.write_int32(40) + .write_int32(1) + .write_int32(cfg.token) + .write_string(&cfg.filename) + .write_int64(cfg.content.len() as u64); + p.write_all(&tr.get_buffer())?; + p.flush()?; + let _response = expect_code(&mut p, 41, Duration::from_secs(10))?; + + // 5. Serve the bytes over an F connection to the downloader's listener. + serve_file_over_f( + &cfg.downloader_listen_addr, + &cfg.username, + cfg.token, + &cfg.content, + ) +} + +#[test] +fn a_file_downloads_from_a_firewalled_peer_via_server_broker() { + let server = server_or_skip!(); + + let client_port = free_port().expect("free client listen port"); + let mut client = Client::with_settings(server.listening_settings( + "e2e_fw_dl", + "pw", + client_port, + )); + client.connect().expect("connect"); + assert!(client.login().expect("login")); + + let bogus_port = free_port().expect("bogus port"); // advertised, unlistened + let filename = "firewalled_song.mp3"; + let content: Vec = (0..2000u32).map(|i| (i % 251) as u8).collect(); + let size = content.len() as u64; + let token = 606_060_u32; + let download_dir = unique_download_dir(); + + let (ready_tx, ready_rx) = std::sync::mpsc::channel(); + let cfg = MockFirewalledUpload { + server_addr: format!("{}:{}", server.host, server.port), + username: "e2e_fw_peer".to_string(), + password: "pw".to_string(), + bogus_port, + downloader_listen_addr: format!("127.0.0.1:{client_port}"), + filename: filename.to_string(), + content: content.clone(), + token, + ready: ready_tx, + }; + let uploader = std::thread::spawn(move || { + if let Err(e) = run_mock_firewalled_peer(&cfg) { + eprintln!("[mock firewalled peer] {e}"); + } + }); + + ready_rx + .recv_timeout(Duration::from_secs(10)) + .expect("mock firewalled peer ready"); + std::thread::sleep(Duration::from_millis(1000)); + + let (_download, status_rx) = client + .download( + filename.to_string(), + "e2e_fw_peer".to_string(), + size, + download_dir.display().to_string(), + ) + .expect("start download"); + + let mut completed = false; + let deadline = Instant::now() + Duration::from_secs(25); + while Instant::now() < deadline { + match status_rx.recv_timeout(Duration::from_millis(500)) { + Ok(DownloadStatus::Completed) => { + completed = true; + break; + } + Ok(DownloadStatus::Failed | DownloadStatus::TimedOut) => break, + _ => {} + } + if client + .get_all_downloads() + .iter() + .any(|d| matches!(d.status, DownloadStatus::Completed)) + { + completed = true; + break; + } + } + let _ = uploader.join(); + + assert!(completed, "the firewalled download should reach Completed"); + + let written = std::fs::read(download_dir.join(filename)) + .expect("downloaded file should exist"); + assert_eq!(written, content, "downloaded bytes should match the source"); + + let _ = std::fs::remove_dir_all(&download_dir); +}