From 32475d859e14478a75df006993eaec8766bc2939 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sat, 11 Jul 2026 18:45:41 +0000 Subject: [PATCH] transport: consolidate tor and nym SOCKS5 dialing into a shared module The Nym transport was a near-clone of the Tor transport's outbound SOCKS5 path. Extract the shared logic into src/transport/socks5/ so both transports drive one implementation instead of two maintained copies: - share one SOCKS5 mock server between the tor and nym tests - extract the common send/receive/connect counters into a shared ProxiedStatsBase (snapshot structs and emitted metrics unchanged) - add a shared Socks5Dialer that collapses all six connect variants; the sole dialing difference (tor's per-destination circuit-isolation auth vs nym's no-auth) is an enum on the dialer - route both tor and nym dialing through the shared dialer - share the proxied connection pool, generic over a per-connection meta type that carries tor's inbound/outbound direction counting (nym uses the unit type) - share the proxied receive loop Behavior-neutral: no wire-format, config-key, emitted-metric, log, or error-variant change. Tor keeps its control-port, inbound/onion, and directory surface and its own address validation. --- src/transport/mod.rs | 1 + src/transport/nym/mod.rs | 278 +++-------- src/transport/nym/stats.rs | 75 ++- src/transport/socks5/dialer.rs | 238 ++++++++++ .../{nym/mock_socks5.rs => socks5/mock.rs} | 79 +++- src/transport/socks5/mod.rs | 20 + src/transport/socks5/pool.rs | 211 +++++++++ src/transport/socks5/stats.rs | 90 ++++ src/transport/tor/mock_socks5.rs | 225 --------- src/transport/tor/mod.rs | 441 ++++++------------ src/transport/tor/stats.rs | 72 ++- 11 files changed, 901 insertions(+), 829 deletions(-) create mode 100644 src/transport/socks5/dialer.rs rename src/transport/{nym/mock_socks5.rs => socks5/mock.rs} (66%) create mode 100644 src/transport/socks5/mod.rs create mode 100644 src/transport/socks5/pool.rs create mode 100644 src/transport/socks5/stats.rs delete mode 100644 src/transport/tor/mock_socks5.rs diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 1176215..31df5f1 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -7,6 +7,7 @@ #[cfg(test)] pub mod loopback; pub mod nym; +pub mod socks5; pub mod tcp; pub mod tor; pub mod udp; diff --git a/src/transport/nym/mod.rs b/src/transport/nym/mod.rs index 053677f..67dd79a 100644 --- a/src/transport/nym/mod.rs +++ b/src/transport/nym/mod.rs @@ -14,19 +14,17 @@ pub mod stats; -#[cfg(test)] -mod mock_socks5; - use super::{ - ConnectionState, DiscoveredPeer, PacketTx, ReceivedPacket, Transport, TransportAddr, - TransportError, TransportId, TransportState, TransportType, + ConnectionState, DiscoveredPeer, PacketTx, Transport, TransportAddr, TransportError, + TransportId, TransportState, TransportType, }; use crate::config::NymConfig; -use crate::transport::framing::read_fmp_packet; +use crate::transport::socks5::{ + ConnectingEntry, ConnectingPool, DialError, ProxiedConnection, ProxiedPool, Socks5Auth, + Socks5Dialer, SocksTarget, poll_connecting, proxied_receive_loop, +}; use stats::NymStats; -use futures::FutureExt; -use socket2::TcpKeepalive; use std::collections::HashMap; use std::net::SocketAddr; use std::sync::Arc; @@ -35,41 +33,9 @@ use tokio::io::AsyncWriteExt; use tokio::net::TcpStream; use tokio::net::tcp::OwnedWriteHalf; use tokio::sync::Mutex; -use tokio::task::JoinHandle; use tokio::time::Instant; -use tokio_socks::tcp::Socks5Stream; use tracing::{debug, info, trace, warn}; -// ============================================================================ -// Connection Pool -// ============================================================================ - -/// State for a single Nym connection to a peer. -struct NymConnection { - /// Write half of the split stream. - writer: Arc>, - /// Receive task for this connection. - recv_task: JoinHandle<()>, - /// MTU for this connection. - #[allow(dead_code)] - mtu: u16, - /// When the connection was established. - #[allow(dead_code)] - established_at: Instant, -} - -/// Shared connection pool. -type ConnectionPool = Arc>>; - -/// A pending background connection attempt. -struct ConnectingEntry { - /// Background task performing SOCKS5 connect + socket configuration. - task: JoinHandle>, -} - -/// Map of addresses with background connection attempts in progress. -type ConnectingPool = Arc>>; - // ============================================================================ // Nym Transport // ============================================================================ @@ -89,7 +55,7 @@ pub struct NymTransport { /// Current state. state: TransportState, /// Connection pool: addr -> per-connection state. - pool: ConnectionPool, + pool: ProxiedPool<()>, /// Pending connection attempts: addr -> background connect task. connecting: ConnectingPool, /// Channel for delivering received packets to Node. @@ -347,20 +313,16 @@ impl NymTransport { "Connecting via Nym mixnet SOCKS5 proxy" ); - let connect_start = Instant::now(); - let socks_result = tokio::time::timeout(Duration::from_millis(timeout_ms), async { - match target_addr { - TargetAddr::Ip(socket_addr) => Socks5Stream::connect(proxy_addr, socket_addr).await, - TargetAddr::Hostname(host, port) => { - Socks5Stream::connect(proxy_addr, (host.as_str(), port)).await - } - } - }) - .await; + let dialer = Socks5Dialer { + proxy_addr: proxy_addr.to_string(), + connect_timeout: Duration::from_millis(timeout_ms), + auth: Socks5Auth::None, + }; - let stream = match socks_result { - Ok(Ok(socks_stream)) => socks_stream.into_inner(), - Ok(Err(e)) => { + let connect_start = Instant::now(); + let stream = match dialer.dial(&target_addr).await { + Ok(stream) => stream, + Err(DialError::Socks(e)) => { self.stats.record_socks5_error(); warn!( transport_id = %self.transport_id, @@ -371,7 +333,7 @@ impl NymTransport { ); return Err(TransportError::ConnectionRefused); } - Err(_) => { + Err(DialError::Timeout) => { self.stats.record_connect_timeout(); warn!( transport_id = %self.transport_id, @@ -381,18 +343,9 @@ impl NymTransport { ); return Err(TransportError::Timeout); } + Err(DialError::Setup(e)) => return Err(e), }; - // Configure socket options via socket2 - let std_stream = stream - .into_std() - .map_err(|e| TransportError::StartFailed(format!("into_std: {}", e)))?; - configure_socket(&std_stream)?; - - // Convert back to tokio - let stream = TcpStream::from_std(std_stream) - .map_err(|e| TransportError::StartFailed(format!("from_std: {}", e)))?; - // Split and spawn receive task let (read_half, write_half) = stream.into_split(); let writer = Arc::new(Mutex::new(write_half)); @@ -417,11 +370,12 @@ impl NymTransport { .await; }); - let conn = NymConnection { + let conn = ProxiedConnection { writer: writer.clone(), recv_task, mtu, established_at: Instant::now(), + meta: (), }; let mut pool = self.pool.lock().await; @@ -485,29 +439,23 @@ impl NymTransport { "Nym SOCKS5 CONNECT starting (this may take several minutes through mixnet)" ); - let socks_result = tokio::time::timeout(Duration::from_millis(timeout_ms), async { - match target_addr { - TargetAddr::Ip(socket_addr) => { - Socks5Stream::connect(proxy_addr.as_str(), socket_addr).await - } - TargetAddr::Hostname(host, port) => { - Socks5Stream::connect(proxy_addr.as_str(), (host.as_str(), port)).await - } - } - }) - .await; + let dialer = Socks5Dialer { + proxy_addr, + connect_timeout: Duration::from_millis(timeout_ms), + auth: Socks5Auth::None, + }; - let stream = match socks_result { - Ok(Ok(socks_stream)) => { + let stream = match dialer.dial(&target_addr).await { + Ok(stream) => { debug!( transport_id = %transport_id, remote_addr = %remote_addr, elapsed_secs = connect_start.elapsed().as_secs(), "Nym SOCKS5 CONNECT succeeded" ); - socks_stream.into_inner() + stream } - Ok(Err(e)) => { + Err(DialError::Socks(e)) => { warn!( transport_id = %transport_id, remote_addr = %remote_addr, @@ -517,7 +465,7 @@ impl NymTransport { ); return Err(TransportError::ConnectionRefused); } - Err(_) => { + Err(DialError::Timeout) => { warn!( transport_id = %transport_id, remote_addr = %remote_addr, @@ -528,20 +476,11 @@ impl NymTransport { ); return Err(TransportError::Timeout); } + Err(DialError::Setup(e)) => return Err(e), }; - // Configure socket options via socket2 - let std_stream = stream - .into_std() - .map_err(|e| TransportError::StartFailed(format!("into_std: {}", e)))?; - configure_socket(&std_stream)?; - let mtu = config.mtu(); - // Convert back to tokio - let stream = TcpStream::from_std(std_stream) - .map_err(|e| TransportError::StartFailed(format!("from_std: {}", e)))?; - Ok((stream, mtu)) }); @@ -553,43 +492,9 @@ impl NymTransport { /// Query the state of a connection to a remote address. pub fn connection_state_sync(&self, addr: &TransportAddr) -> ConnectionState { - // Check established pool first - if let Ok(pool) = self.pool.try_lock() { - if pool.contains_key(addr) { - return ConnectionState::Connected; - } - } else { - return ConnectionState::Connecting; - } - - // Check connecting pool - let mut connecting = match self.connecting.try_lock() { - Ok(c) => c, - Err(_) => return ConnectionState::Connecting, - }; - - let entry = match connecting.get_mut(addr) { - Some(e) => e, - None => return ConnectionState::None, - }; - - if !entry.task.is_finished() { - return ConnectionState::Connecting; - } - - // Task is done — take the result - let addr_clone = addr.clone(); - let task = connecting.remove(&addr_clone).unwrap().task; - - match task.now_or_never() { - Some(Ok(Ok((stream, mtu)))) => { - self.promote_connection(addr, stream, mtu); - ConnectionState::Connected - } - Some(Ok(Err(e))) => ConnectionState::Failed(format!("{}", e)), - Some(Err(e)) => ConnectionState::Failed(format!("task failed: {}", e)), - None => ConnectionState::Connecting, - } + poll_connecting(&self.pool, &self.connecting, addr, |stream, mtu| { + self.promote_connection(addr, stream, mtu) + }) } /// Promote a completed background connection to the established pool. @@ -616,11 +521,12 @@ impl NymTransport { .await; }); - let conn = NymConnection { + let conn = ProxiedConnection { writer, recv_task, mtu, established_at: Instant::now(), + meta: (), }; if let Ok(mut pool) = self.pool.try_lock() { @@ -707,23 +613,14 @@ impl Transport for NymTransport { // Address Parsing // ============================================================================ -/// Target address for the SOCKS5 CONNECT request. -#[derive(Clone, Debug)] -enum TargetAddr { - /// Numeric IP:port. - Ip(SocketAddr), - /// Hostname:port (DNS resolved by the exit node). - Hostname(String, u16), -} - -/// Parse a TransportAddr string into a target address. -fn parse_target_addr(addr: &TransportAddr) -> Result { +/// Parse a TransportAddr string into a shared SOCKS5 target address. +fn parse_target_addr(addr: &TransportAddr) -> Result { let s = addr.as_str().ok_or_else(|| { TransportError::InvalidAddress("Nym address must be a valid UTF-8 string".into()) })?; if let Ok(socket_addr) = s.parse::() { - Ok(TargetAddr::Ip(socket_addr)) + Ok(SocksTarget::Ip(socket_addr)) } else { let (host, port_str) = s.rsplit_once(':').ok_or_else(|| { TransportError::InvalidAddress(format!("invalid address (expected host:port): {}", s)) @@ -731,7 +628,7 @@ fn parse_target_addr(addr: &TransportAddr) -> Result let port: u16 = port_str .parse() .map_err(|_| TransportError::InvalidAddress(format!("invalid port: {}", s)))?; - Ok(TargetAddr::Hostname(host.to_string(), port)) + Ok(SocksTarget::Hostname(host.to_string(), port)) } } @@ -740,59 +637,32 @@ fn parse_target_addr(addr: &TransportAddr) -> Result // ============================================================================ /// Per-connection Nym receive loop. +/// +/// Thin wrapper over the shared `proxied_receive_loop`: nym has no pool +/// counters, so its teardown hook is a no-op. Emits the terminal +/// "receive loop stopped" debug (without a `direction` field) that the +/// shared loop deliberately leaves to each transport. async fn nym_receive_loop( - mut reader: tokio::net::tcp::OwnedReadHalf, + reader: tokio::net::tcp::OwnedReadHalf, transport_id: TransportId, remote_addr: TransportAddr, packet_tx: PacketTx, - pool: ConnectionPool, + pool: ProxiedPool<()>, mtu: u16, stats: Arc, ) { - debug!( - transport_id = %transport_id, - remote_addr = %remote_addr, - "Nym receive loop starting" - ); - - loop { - match read_fmp_packet(&mut reader, mtu).await { - Ok(data) => { - stats.record_recv(data.len()); - - trace!( - transport_id = %transport_id, - remote_addr = %remote_addr, - bytes = data.len(), - "Nym packet received" - ); - - let packet = ReceivedPacket::new(transport_id, remote_addr.clone(), data); - - if packet_tx.send(packet).await.is_err() { - debug!( - transport_id = %transport_id, - "Packet channel closed, stopping Nym receive loop" - ); - break; - } - } - Err(e) => { - stats.record_recv_error(); - debug!( - transport_id = %transport_id, - remote_addr = %remote_addr, - error = %e, - "Nym receive error, removing connection" - ); - break; - } - } - } - - // Clean up: remove ourselves from the pool - let mut pool_guard = pool.lock().await; - pool_guard.remove(&remote_addr); + proxied_receive_loop( + reader, + transport_id, + remote_addr.clone(), + packet_tx, + pool, + mtu, + stats, + "Nym", + |_stats, _meta| {}, + ) + .await; debug!( transport_id = %transport_id, @@ -801,28 +671,6 @@ async fn nym_receive_loop( ); } -// ============================================================================ -// Socket Configuration -// ============================================================================ - -/// Configure socket options on a SOCKS5-connected stream. -fn configure_socket(stream: &std::net::TcpStream) -> Result<(), TransportError> { - let socket = socket2::SockRef::from(stream); - - // TCP_NODELAY — always enable for FIPS (latency-sensitive protocol messages) - socket - .set_tcp_nodelay(true) - .map_err(|e| TransportError::StartFailed(format!("set nodelay: {}", e)))?; - - // TCP keepalive (30s, matching TCP transport) - let keepalive = TcpKeepalive::new().with_time(Duration::from_secs(30)); - socket - .set_tcp_keepalive(&keepalive) - .map_err(|e| TransportError::StartFailed(format!("set keepalive: {}", e)))?; - - Ok(()) -} - // ============================================================================ // Address Validation // ============================================================================ @@ -869,7 +717,7 @@ mod tests { fn test_parse_target_addr_ipv4() { let addr = TransportAddr::from_string("192.0.2.10:2121"); match parse_target_addr(&addr).unwrap() { - TargetAddr::Ip(socket_addr) => { + SocksTarget::Ip(socket_addr) => { assert_eq!( socket_addr, "192.0.2.10:2121".parse::().unwrap() @@ -886,7 +734,7 @@ mod tests { // correctly (this is the path that actually dials peers). let addr = TransportAddr::from_string("[2001:db8::1]:443"); match parse_target_addr(&addr).unwrap() { - TargetAddr::Ip(socket_addr) => { + SocksTarget::Ip(socket_addr) => { assert_eq!( socket_addr, "[2001:db8::1]:443".parse::().unwrap() @@ -900,7 +748,7 @@ mod tests { fn test_parse_target_addr_hostname() { let addr = TransportAddr::from_string("peer.example.com:8443"); match parse_target_addr(&addr).unwrap() { - TargetAddr::Hostname(host, port) => { + SocksTarget::Hostname(host, port) => { assert_eq!(host, "peer.example.com"); assert_eq!(port, 8443); } @@ -1074,8 +922,8 @@ mod tests { // ======================================================================== use crate::config::TcpConfig; + use crate::transport::socks5::mock::MockSocks5Server; use crate::transport::tcp::TcpTransport; - use mock_socks5::MockSocks5Server; /// msg1 wire size: 4 prefix + 4 sender_idx + 106 noise_msg1 = 114 bytes. const MSG1_WIRE_SIZE: usize = 114; diff --git a/src/transport/nym/stats.rs b/src/transport/nym/stats.rs index ec6f287..f349c93 100644 --- a/src/transport/nym/stats.rs +++ b/src/transport/nym/stats.rs @@ -1,98 +1,81 @@ //! Nym transport statistics. -use portable_atomic::{AtomicU64, Ordering}; +use portable_atomic::Ordering; use serde::Serialize; +use crate::transport::socks5::{ProxiedStats, ProxiedStatsBase}; + /// Statistics for a Nym transport instance. /// /// Uses atomic counters for lock-free updates from per-connection /// receive loops and the send path concurrently. pub struct NymStats { - pub packets_sent: AtomicU64, - pub bytes_sent: AtomicU64, - pub packets_recv: AtomicU64, - pub bytes_recv: AtomicU64, - pub send_errors: AtomicU64, - pub recv_errors: AtomicU64, - pub mtu_exceeded: AtomicU64, - pub connections_established: AtomicU64, - pub connect_timeouts: AtomicU64, - pub socks5_errors: AtomicU64, + /// Shared send/receive/connect counters common to proxied transports. + base: ProxiedStatsBase, } impl NymStats { /// Create a new stats instance with all counters at zero. pub fn new() -> Self { Self { - packets_sent: AtomicU64::new(0), - bytes_sent: AtomicU64::new(0), - packets_recv: AtomicU64::new(0), - bytes_recv: AtomicU64::new(0), - send_errors: AtomicU64::new(0), - recv_errors: AtomicU64::new(0), - mtu_exceeded: AtomicU64::new(0), - connections_established: AtomicU64::new(0), - connect_timeouts: AtomicU64::new(0), - socks5_errors: AtomicU64::new(0), + base: ProxiedStatsBase::new(), } } /// Record a successful send. pub fn record_send(&self, bytes: usize) { - self.packets_sent.fetch_add(1, Ordering::Relaxed); - self.bytes_sent.fetch_add(bytes as u64, Ordering::Relaxed); + self.base.record_send(bytes); } /// Record a successful receive. pub fn record_recv(&self, bytes: usize) { - self.packets_recv.fetch_add(1, Ordering::Relaxed); - self.bytes_recv.fetch_add(bytes as u64, Ordering::Relaxed); + self.base.record_recv(bytes); } /// Record a send error. pub fn record_send_error(&self) { - self.send_errors.fetch_add(1, Ordering::Relaxed); + self.base.record_send_error(); } /// Record a receive error. pub fn record_recv_error(&self) { - self.recv_errors.fetch_add(1, Ordering::Relaxed); + self.base.record_recv_error(); } /// Record an MTU exceeded rejection. pub fn record_mtu_exceeded(&self) { - self.mtu_exceeded.fetch_add(1, Ordering::Relaxed); + self.base.record_mtu_exceeded(); } /// Record a successful outbound connection. pub fn record_connection_established(&self) { - self.connections_established.fetch_add(1, Ordering::Relaxed); + self.base.record_connection_established(); } /// Record a connect timeout. pub fn record_connect_timeout(&self) { - self.connect_timeouts.fetch_add(1, Ordering::Relaxed); + self.base.record_connect_timeout(); } /// Record a SOCKS5 protocol error. pub fn record_socks5_error(&self) { - self.socks5_errors.fetch_add(1, Ordering::Relaxed); + self.base.record_socks5_error(); } /// Take a snapshot of all counters. pub fn snapshot(&self) -> NymStatsSnapshot { NymStatsSnapshot { - packets_sent: self.packets_sent.load(Ordering::Relaxed), - bytes_sent: self.bytes_sent.load(Ordering::Relaxed), - packets_recv: self.packets_recv.load(Ordering::Relaxed), - bytes_recv: self.bytes_recv.load(Ordering::Relaxed), - send_errors: self.send_errors.load(Ordering::Relaxed), - recv_errors: self.recv_errors.load(Ordering::Relaxed), - mtu_exceeded: self.mtu_exceeded.load(Ordering::Relaxed), - connections_established: self.connections_established.load(Ordering::Relaxed), - connect_timeouts: self.connect_timeouts.load(Ordering::Relaxed), - socks5_errors: self.socks5_errors.load(Ordering::Relaxed), + packets_sent: self.base.packets_sent.load(Ordering::Relaxed), + bytes_sent: self.base.bytes_sent.load(Ordering::Relaxed), + packets_recv: self.base.packets_recv.load(Ordering::Relaxed), + bytes_recv: self.base.bytes_recv.load(Ordering::Relaxed), + send_errors: self.base.send_errors.load(Ordering::Relaxed), + recv_errors: self.base.recv_errors.load(Ordering::Relaxed), + mtu_exceeded: self.base.mtu_exceeded.load(Ordering::Relaxed), + connections_established: self.base.connections_established.load(Ordering::Relaxed), + connect_timeouts: self.base.connect_timeouts.load(Ordering::Relaxed), + socks5_errors: self.base.socks5_errors.load(Ordering::Relaxed), } } } @@ -103,6 +86,16 @@ impl Default for NymStats { } } +impl ProxiedStats for NymStats { + fn record_recv(&self, bytes: usize) { + self.base.record_recv(bytes); + } + + fn record_recv_error(&self) { + self.base.record_recv_error(); + } +} + /// Point-in-time snapshot of Nym stats (non-atomic, copyable). #[derive(Clone, Debug, Default, Serialize)] pub struct NymStatsSnapshot { diff --git a/src/transport/socks5/dialer.rs b/src/transport/socks5/dialer.rs new file mode 100644 index 0000000..9faff1d --- /dev/null +++ b/src/transport/socks5/dialer.rs @@ -0,0 +1,238 @@ +//! Shared SOCKS5 dialer for the proxied (Tor / Nym) transports. +//! +//! Collapses the six near-identical dial cores (nym×3 + tor×3) into one +//! timeout-wrapped, auth-branched SOCKS5 CONNECT followed by socket +//! configuration. The dialer records **no** stats and emits **no** logs — +//! callers classify the returned [`DialError`] (e.g. tor's refused-vs-error +//! split) and do their own logging. + +use std::net::SocketAddr; +use std::time::Duration; + +use socket2::TcpKeepalive; +use tokio::net::TcpStream; +use tokio_socks::tcp::Socks5Stream; + +use crate::transport::TransportError; + +/// Target of a SOCKS5 CONNECT request. +/// +/// A tor `.onion` host and a clearnet hostname both map to `Hostname` (the +/// proxy / exit resolves it); numeric `IP:port` maps to `Ip`. +#[derive(Clone, Debug)] +pub enum SocksTarget { + /// Numeric IP:port — dialed as a `SocketAddr`. + Ip(SocketAddr), + /// Hostname:port — resolved by the SOCKS5 proxy / exit node. + Hostname(String, u16), +} + +/// SOCKS5 authentication mode for a dial. +#[derive(Clone, Debug)] +pub enum Socks5Auth { + /// No authentication (nym). + None, + /// Username/password auth (tor: `("fips", isolation_key)` for + /// per-destination circuit isolation via `IsolateSOCKSAuth`). + Password { username: String, password: String }, +} + +/// Failure classification returned by [`Socks5Dialer::dial`]. +/// +/// Carries the raw `tokio_socks::Error` so callers can classify it (tor's +/// `matches!(e, ConnectionRefused)` split) without the dialer recording stats. +pub enum DialError { + /// The SOCKS5 CONNECT did not complete within the connect timeout. + Timeout, + /// The SOCKS5 CONNECT returned an error (refusal, ruleset, I/O, protocol). + Socks(tokio_socks::Error), + /// Post-connect socket setup (`into_std` / `configure_socket` / `from_std`) + /// failed. + Setup(TransportError), +} + +/// Timeout-wrapped, auth-branched SOCKS5 dialer. +/// +/// Carries no stats and does no logging — callers classify/record/log. +pub struct Socks5Dialer { + /// SOCKS5 proxy address (`host:port`). + pub proxy_addr: String, + /// Timeout applied to the `Socks5Stream::connect*` call only. + pub connect_timeout: Duration, + /// Authentication mode. + pub auth: Socks5Auth, +} + +impl Socks5Dialer { + /// Perform a timeout-wrapped SOCKS5 CONNECT and configure the socket. + /// + /// Only the `Socks5Stream::connect*` call is wrapped in + /// `tokio::time::timeout`; `into_inner → into_std → configure_socket → + /// from_std` run **outside** the timeout. Elapsed → [`DialError::Timeout`]; + /// an inner `Err(e)` → [`DialError::Socks`]; setup failures → + /// [`DialError::Setup`]. + pub async fn dial(&self, target: &SocksTarget) -> Result { + let socks_result = tokio::time::timeout(self.connect_timeout, async { + match (&self.auth, target) { + (Socks5Auth::None, SocksTarget::Ip(socket_addr)) => { + Socks5Stream::connect(self.proxy_addr.as_str(), *socket_addr).await + } + (Socks5Auth::None, SocksTarget::Hostname(host, port)) => { + Socks5Stream::connect(self.proxy_addr.as_str(), (host.as_str(), *port)).await + } + (Socks5Auth::Password { username, password }, SocksTarget::Ip(socket_addr)) => { + Socks5Stream::connect_with_password( + self.proxy_addr.as_str(), + *socket_addr, + username.as_str(), + password.as_str(), + ) + .await + } + ( + Socks5Auth::Password { username, password }, + SocksTarget::Hostname(host, port), + ) => { + Socks5Stream::connect_with_password( + self.proxy_addr.as_str(), + (host.as_str(), *port), + username.as_str(), + password.as_str(), + ) + .await + } + } + }) + .await; + + let stream = match socks_result { + Ok(Ok(socks_stream)) => socks_stream.into_inner(), + Ok(Err(e)) => return Err(DialError::Socks(e)), + Err(_) => return Err(DialError::Timeout), + }; + + // Configure socket options via socket2 — OUTSIDE the timeout. + let std_stream = stream.into_std().map_err(|e| { + DialError::Setup(TransportError::StartFailed(format!("into_std: {}", e))) + })?; + configure_socket(&std_stream).map_err(DialError::Setup)?; + + // Convert back to tokio. + let stream = TcpStream::from_std(std_stream).map_err(|e| { + DialError::Setup(TransportError::StartFailed(format!("from_std: {}", e))) + })?; + + Ok(stream) + } +} + +/// Configure socket options on a SOCKS5-connected stream. +/// +/// Sets TCP_NODELAY and a 30 s keepalive on the underlying TCP connection. +fn configure_socket(stream: &std::net::TcpStream) -> Result<(), TransportError> { + let socket = socket2::SockRef::from(stream); + + // TCP_NODELAY — always enable for FIPS (latency-sensitive protocol messages) + socket + .set_tcp_nodelay(true) + .map_err(|e| TransportError::StartFailed(format!("set nodelay: {}", e)))?; + + // TCP keepalive (30s, matching TCP transport) + let keepalive = TcpKeepalive::new().with_time(Duration::from_secs(30)); + socket + .set_tcp_keepalive(&keepalive) + .map_err(|e| TransportError::StartFailed(format!("set keepalive: {}", e)))?; + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::transport::socks5::mock::MockSocks5Server; + + async fn dest_listener() -> (tokio::net::TcpListener, SocketAddr) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + (listener, addr) + } + + #[tokio::test] + async fn test_dial_no_auth_success() { + let (dest, dest_addr) = dest_listener().await; + let mock = MockSocks5Server::new(dest_addr).await.unwrap(); + let proxy_addr = mock.addr(); + let _proxy = mock.spawn(); + + let dialer = Socks5Dialer { + proxy_addr: proxy_addr.to_string(), + connect_timeout: Duration::from_secs(5), + auth: Socks5Auth::None, + }; + + let result = dialer.dial(&SocksTarget::Ip(dest_addr)).await; + assert!(result.is_ok(), "no-auth dial should succeed"); + drop(dest); + } + + #[tokio::test] + async fn test_dial_password_success() { + let (dest, dest_addr) = dest_listener().await; + let mock = MockSocks5Server::new(dest_addr).await.unwrap(); + let proxy_addr = mock.addr(); + let _proxy = mock.spawn(); + + let dialer = Socks5Dialer { + proxy_addr: proxy_addr.to_string(), + connect_timeout: Duration::from_secs(5), + auth: Socks5Auth::Password { + username: "fips".to_string(), + password: "isolation-key".to_string(), + }, + }; + + // Hostname target exercises the domain ATYP path through the mock. + let result = dialer + .dial(&SocksTarget::Hostname("example.com".to_string(), 2121)) + .await; + assert!(result.is_ok(), "password-auth dial should succeed"); + drop(dest); + } + + #[tokio::test] + async fn test_dial_refused_maps_to_socks_error() { + let dummy: SocketAddr = "127.0.0.1:1".parse().unwrap(); + let mock = MockSocks5Server::with_reply_code(dummy, 0x05) + .await + .unwrap(); + let proxy_addr = mock.addr(); + let _proxy = mock.spawn(); + + let dialer = Socks5Dialer { + proxy_addr: proxy_addr.to_string(), + connect_timeout: Duration::from_secs(5), + auth: Socks5Auth::None, + }; + + let result = dialer.dial(&SocksTarget::Ip(dummy)).await; + assert!( + matches!(result, Err(DialError::Socks(_))), + "refused CONNECT should surface as DialError::Socks carrying the raw error" + ); + } + + #[tokio::test] + async fn test_dial_timeout() { + // 192.0.2.1 is TEST-NET-1: non-routable, so the connect stalls until + // the short timeout elapses. + let dialer = Socks5Dialer { + proxy_addr: "192.0.2.1:9050".to_string(), + connect_timeout: Duration::from_millis(300), + auth: Socks5Auth::None, + }; + + let target: SocketAddr = "10.0.0.1:2121".parse().unwrap(); + let result = dialer.dial(&SocksTarget::Ip(target)).await; + assert!(matches!(result, Err(DialError::Timeout))); + } +} diff --git a/src/transport/nym/mock_socks5.rs b/src/transport/socks5/mock.rs similarity index 66% rename from src/transport/nym/mock_socks5.rs rename to src/transport/socks5/mock.rs index 2eff6fc..5c1b1fb 100644 --- a/src/transport/nym/mock_socks5.rs +++ b/src/transport/socks5/mock.rs @@ -1,15 +1,21 @@ -//! Mock SOCKS5 server for testing the Nym transport's connect path. +//! Mock SOCKS5 server for testing the proxied (Tor / Nym) transports. //! -//! A copy of the Tor transport's mock, implementing just enough of the -//! SOCKS5 protocol (RFC 1928) to support the no-auth (and, defensively, -//! username/password) CONNECT flow, then proxying bytes bidirectionally to a -//! fixed target. +//! Implements just enough of the SOCKS5 protocol (RFC 1928) to support the +//! no-auth and username/password (RFC 1929) CONNECT flows, then proxies bytes +//! bidirectionally to a fixed target. //! -//! Difference from the Tor mock: this one accepts connections in a loop and -//! handles each on its own task. `NymTransport::start_async` first probes the -//! proxy port for readiness (opening and immediately dropping a connection); -//! looping lets the mock shrug that probe off — its handler returns on the -//! short first read — and still serve the real data connection that follows. +//! Combines the supersets of the two former per-transport mocks: +//! +//! - **Accept loop with per-connection spawn** (from the Nym mock). +//! `NymTransport::start_async` first probes the proxy port for readiness +//! (opening and immediately dropping a connection); handling each connection +//! on its own task lets the mock shrug that probe off — the handler returns +//! on the short first read — and still serve the real data connection that +//! follows. +//! - **`with_reply_code` / `reply_code` injection** (from the Tor mock). A +//! non-success REP code (e.g. `0x05` Connection refused) is sent as an error +//! reply and the connection is closed without proxying, exercising the +//! client's connect-error path. use std::net::SocketAddr; use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -31,14 +37,19 @@ const AUTH_SUBNEG_SUCCESS: u8 = 0x00; /// A minimal mock SOCKS5 proxy server for testing. /// -/// Accepts connections in a loop, performs the SOCKS5 handshake (supporting -/// both no-auth and username/password auth), then connects to a fixed target -/// address and proxies bytes bidirectionally. +/// Accepts connections in a loop, handling each on its own task: performs the +/// SOCKS5 handshake (supporting both no-auth and username/password auth), then +/// connects to a fixed target address and proxies bytes bidirectionally. pub struct MockSocks5Server { /// Address the mock proxy is listening on. addr: SocketAddr, /// The real target address to connect to (ignores SOCKS5 requested target). target_addr: SocketAddr, + /// SOCKS5 reply code (REP field) to send in the CONNECT reply. When this + /// is `REP_SUCCESS` (0x00) the server proxies to `target_addr`; any other + /// value is sent as an error reply and the connection is closed without + /// proxying. Use `0x05` (Connection refused) to drive the refused path. + reply_code: u8, /// Listener handle. listener: Option, } @@ -46,18 +57,28 @@ pub struct MockSocks5Server { impl MockSocks5Server { /// Create a new mock SOCKS5 server that forwards to the given target. /// - /// Binds to `127.0.0.1:0` (OS-assigned port). + /// Binds to `127.0.0.1:0` (OS-assigned port). Replies with success and + /// proxies bytes bidirectionally. pub async fn new(target_addr: SocketAddr) -> std::io::Result { + Self::with_reply_code(target_addr, REP_SUCCESS).await + } + + /// Create a mock SOCKS5 server that replies to the CONNECT request with + /// the given REP code. A non-success code (e.g. `0x05` Connection refused) + /// causes the server to send the error reply and close the connection + /// without proxying, exercising the client's connect-error path. + pub async fn with_reply_code(target_addr: SocketAddr, reply_code: u8) -> std::io::Result { let listener = TcpListener::bind("127.0.0.1:0").await?; let addr = listener.local_addr()?; Ok(Self { addr, target_addr, + reply_code, listener: Some(listener), }) } - /// Get the proxy's listen address (for `NymConfig.socks5_addr`). + /// Get the proxy's listen address (for `socks5_addr` config). pub fn addr(&self) -> SocketAddr { self.addr } @@ -68,6 +89,7 @@ impl MockSocks5Server { pub fn spawn(mut self) -> JoinHandle<()> { let listener = self.listener.take().expect("listener already consumed"); let target_addr = self.target_addr; + let reply_code = self.reply_code; tokio::spawn(async move { loop { @@ -78,14 +100,14 @@ impl MockSocks5Server { // Handle each connection independently so the readiness probe // (which opens and drops a connection) cannot block the real // data connection behind it. - tokio::spawn(handle_conn(client, target_addr)); + tokio::spawn(handle_conn(client, target_addr, reply_code)); } }) } } /// Handle a single accepted connection: SOCKS5 handshake then byte proxy. -async fn handle_conn(mut client: tokio::net::TcpStream, target_addr: SocketAddr) { +async fn handle_conn(mut client: tokio::net::TcpStream, target_addr: SocketAddr, reply_code: u8) { // === Method negotiation === // Client sends: [version, nmethods, methods...] let mut ver_nmethods = [0u8; 2]; @@ -137,6 +159,7 @@ async fn handle_conn(mut client: tokio::net::TcpStream, target_addr: SocketAddr) let mut passwd = vec![0u8; plen]; client.read_exact(&mut passwd).await.expect("read password"); + // Always accept (Tor uses these as isolation keys, not real auth). client .write_all(&[AUTH_SUBNEG_VERSION, AUTH_SUBNEG_SUCCESS]) .await @@ -178,6 +201,28 @@ async fn handle_conn(mut client: tokio::net::TcpStream, target_addr: SocketAddr) other => panic!("unsupported ATYP: {}", other), } + // Error path: reply with the configured REP code and close without + // proxying (the client maps the REP code to a tokio_socks::Error). + if reply_code != REP_SUCCESS { + let reply = [ + SOCKS_VERSION, + reply_code, + 0x00, // RSV + ATYP_IPV4, + 0, + 0, + 0, + 0, // bind addr + 0, + 0, // bind port + ]; + client + .write_all(&reply) + .await + .expect("write error connect reply"); + return; + } + // Connect to the real target. let mut target = tokio::net::TcpStream::connect(target_addr) .await diff --git a/src/transport/socks5/mod.rs b/src/transport/socks5/mod.rs new file mode 100644 index 0000000..eb62883 --- /dev/null +++ b/src/transport/socks5/mod.rs @@ -0,0 +1,20 @@ +//! Shared SOCKS5 / proxied-transport building blocks. +//! +//! The Tor and Nym transports both dial outbound through a local SOCKS5 +//! proxy and share the same connection-pool, receive-loop, and dialing +//! machinery. That shared surface lives here so it is written once rather +//! than copied into each transport. + +mod dialer; +mod pool; +mod stats; + +pub use dialer::{DialError, Socks5Auth, Socks5Dialer, SocksTarget}; +pub(crate) use pool::{ + ConnectingEntry, ConnectingPool, ProxiedConnection, ProxiedPool, ProxiedStats, poll_connecting, + proxied_receive_loop, +}; +pub(crate) use stats::ProxiedStatsBase; + +#[cfg(test)] +pub(crate) mod mock; diff --git a/src/transport/socks5/pool.rs b/src/transport/socks5/pool.rs new file mode 100644 index 0000000..62ca332 --- /dev/null +++ b/src/transport/socks5/pool.rs @@ -0,0 +1,211 @@ +//! Shared connection pool for the proxied (Tor / Nym) transports. +//! +//! Both transports keep the same two maps — an established-connection pool and +//! a pending-connection ("connecting") pool — and poll a completed background +//! connect the same way. The only per-transport difference is the metadata +//! carried on each pooled connection (`Direction` for tor's inbound/outbound +//! pool accounting, `()` for nym), captured by the generic `M` type parameter. + +use std::collections::HashMap; +use std::sync::Arc; + +use futures::FutureExt; +use tokio::net::TcpStream; +use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; +use tokio::sync::Mutex; +use tokio::task::JoinHandle; +use tokio::time::Instant; +use tracing::{debug, trace}; + +use crate::transport::framing::read_fmp_packet; +use crate::transport::{ + ConnectionState, PacketTx, ReceivedPacket, TransportAddr, TransportError, TransportId, +}; + +/// State for a single pooled connection to a peer. +/// +/// `M` is per-transport metadata: `Direction` for tor (drives +/// inbound/outbound pool accounting), `()` for nym. +pub(crate) struct ProxiedConnection { + /// Write half of the split stream. + pub writer: Arc>, + /// Receive task for this connection. + pub recv_task: JoinHandle<()>, + /// MTU for this connection. + #[allow(dead_code)] + pub mtu: u16, + /// When the connection was established. + #[allow(dead_code)] + pub established_at: Instant, + /// Per-transport metadata (tor: `Direction`; nym: `()`). + pub meta: M, +} + +/// Shared connection pool: addr -> per-connection state. +pub(crate) type ProxiedPool = Arc>>>; + +/// A pending background connection attempt. +/// +/// Holds the JoinHandle for a spawned SOCKS5 connect task. The task +/// produces a configured `TcpStream` and MTU on success. +pub(crate) struct ConnectingEntry { + /// Background task performing SOCKS5 connect + socket configuration. + pub task: JoinHandle>, +} + +/// Map of addresses with background connection attempts in progress. +pub(crate) type ConnectingPool = Arc>>; + +/// Poll the state of a connection to a remote address. +/// +/// Checks both established and connecting pools. If a background connect task +/// has completed successfully, invokes `promote` (which spawns a receive loop +/// and inserts into the established pool) and reports `Connected`; on failure +/// reports it. Synchronous — uses `try_lock` internally and returns +/// `ConnectionState::Connecting` if a lock can't be acquired. +/// +/// This is the byte-for-byte former `connection_state_sync` body, with the +/// per-transport `promote_connection` call abstracted behind `promote`. +pub(crate) fn poll_connecting( + pool: &ProxiedPool, + connecting: &ConnectingPool, + addr: &TransportAddr, + promote: impl FnOnce(TcpStream, u16), +) -> ConnectionState { + // Check established pool first + if let Ok(pool) = pool.try_lock() { + if pool.contains_key(addr) { + return ConnectionState::Connected; + } + } else { + return ConnectionState::Connecting; // can't tell, assume still going + } + + // Check connecting pool + let mut connecting = match connecting.try_lock() { + Ok(c) => c, + Err(_) => return ConnectionState::Connecting, + }; + + let entry = match connecting.get_mut(addr) { + Some(e) => e, + None => return ConnectionState::None, + }; + + // Check if the background task has completed + if !entry.task.is_finished() { + return ConnectionState::Connecting; + } + + // Task is done — take the result and remove from connecting pool. + let addr_clone = addr.clone(); + let task = connecting.remove(&addr_clone).unwrap().task; + + // Since the task is finished, we can safely poll it with now_or_never. + match task.now_or_never() { + Some(Ok(Ok((stream, mtu)))) => { + promote(stream, mtu); + ConnectionState::Connected + } + Some(Ok(Err(e))) => ConnectionState::Failed(format!("{}", e)), + Some(Err(e)) => ConnectionState::Failed(format!("task failed: {}", e)), + None => ConnectionState::Connecting, + } +} + +/// Minimal stats surface the shared receive loop needs. +/// +/// The per-transport stats structs implement this by delegating to their +/// shared counter base; the loop records received bytes and receive errors +/// without knowing the concrete transport. +pub(crate) trait ProxiedStats: Send + Sync + 'static { + /// Record a successful receive of `bytes` bytes. + fn record_recv(&self, bytes: usize); + /// Record a receive error. + fn record_recv_error(&self); +} + +/// Shared per-connection receive loop for the proxied transports. +/// +/// Reads complete FMP packets, delivers them to the node, and on error/EOF +/// removes the connection from the pool and runs `on_remove` for any +/// per-transport teardown accounting. The `label` is the in-loop log word +/// ("Nym" / "Tor"). +/// +/// Teardown/cleanup contract (reproduced exactly to stay behavior-neutral): +/// the pool entry is removed, and `on_remove` fires **only** when the removal +/// returned `Some`, taking the metadata from the removed entry, and after the +/// pool guard is dropped. Firing on `Some` only means a concurrent +/// `close`/`stop` teardown of the same address can never double-count. +/// +/// The terminal "receive loop stopped" log is **not** emitted here — it is +/// hoisted into each per-transport wrapper (tor carries a `direction` field +/// nym lacks), so this loop is silent on exit. +#[allow(clippy::too_many_arguments)] +pub(crate) async fn proxied_receive_loop( + mut reader: OwnedReadHalf, + transport_id: TransportId, + remote_addr: TransportAddr, + packet_tx: PacketTx, + pool: ProxiedPool, + mtu: u16, + stats: Arc, + label: &'static str, + on_remove: impl Fn(&S, &M), +) { + debug!( + transport_id = %transport_id, + remote_addr = %remote_addr, + "{} receive loop starting", + label + ); + + loop { + match read_fmp_packet(&mut reader, mtu).await { + Ok(data) => { + stats.record_recv(data.len()); + + trace!( + transport_id = %transport_id, + remote_addr = %remote_addr, + bytes = data.len(), + "{} packet received", + label + ); + + let packet = ReceivedPacket::new(transport_id, remote_addr.clone(), data); + + if packet_tx.send(packet).await.is_err() { + debug!( + transport_id = %transport_id, + "Packet channel closed, stopping {} receive loop", + label + ); + break; + } + } + Err(e) => { + stats.record_recv_error(); + debug!( + transport_id = %transport_id, + remote_addr = %remote_addr, + error = %e, + "{} receive error, removing connection", + label + ); + break; + } + } + } + + // Clean up: remove ourselves from the pool, then run per-transport + // teardown accounting. The teardown fires only when this loop actually + // removed the entry, using the metadata from the removed entry, so a + // concurrent close/stop teardown of the same address can never + // double-count. + let mut pool_guard = pool.lock().await; + if let Some(removed) = pool_guard.remove(&remote_addr) { + drop(pool_guard); + on_remove(&*stats, &removed.meta); + } +} diff --git a/src/transport/socks5/stats.rs b/src/transport/socks5/stats.rs new file mode 100644 index 0000000..c4e7cbb --- /dev/null +++ b/src/transport/socks5/stats.rs @@ -0,0 +1,90 @@ +//! Shared atomic counter base for the proxied (Tor / Nym) transports. +//! +//! The two SOCKS5-proxied transports share an identical set of ten +//! send/receive/connect counters. That counter logic is defined once here and +//! embedded as `base` in each transport's stats struct; the transport-specific +//! counters (tor's refused/accepted/rejected/control + pool occupancy) stay in +//! the per-transport struct. The flat per-transport snapshot structs are left +//! unchanged so the emitted metric JSON is byte-identical. + +use portable_atomic::{AtomicU64, Ordering}; + +/// Atomic counters common to every SOCKS5-proxied transport. +pub(crate) struct ProxiedStatsBase { + pub packets_sent: AtomicU64, + pub bytes_sent: AtomicU64, + pub packets_recv: AtomicU64, + pub bytes_recv: AtomicU64, + pub send_errors: AtomicU64, + pub recv_errors: AtomicU64, + pub mtu_exceeded: AtomicU64, + pub connections_established: AtomicU64, + pub connect_timeouts: AtomicU64, + pub socks5_errors: AtomicU64, +} + +impl ProxiedStatsBase { + /// Create a base with all counters at zero. + pub fn new() -> Self { + Self { + packets_sent: AtomicU64::new(0), + bytes_sent: AtomicU64::new(0), + packets_recv: AtomicU64::new(0), + bytes_recv: AtomicU64::new(0), + send_errors: AtomicU64::new(0), + recv_errors: AtomicU64::new(0), + mtu_exceeded: AtomicU64::new(0), + connections_established: AtomicU64::new(0), + connect_timeouts: AtomicU64::new(0), + socks5_errors: AtomicU64::new(0), + } + } + + /// Record a successful send. + pub fn record_send(&self, bytes: usize) { + self.packets_sent.fetch_add(1, Ordering::Relaxed); + self.bytes_sent.fetch_add(bytes as u64, Ordering::Relaxed); + } + + /// Record a successful receive. + pub fn record_recv(&self, bytes: usize) { + self.packets_recv.fetch_add(1, Ordering::Relaxed); + self.bytes_recv.fetch_add(bytes as u64, Ordering::Relaxed); + } + + /// Record a send error. + pub fn record_send_error(&self) { + self.send_errors.fetch_add(1, Ordering::Relaxed); + } + + /// Record a receive error. + pub fn record_recv_error(&self) { + self.recv_errors.fetch_add(1, Ordering::Relaxed); + } + + /// Record an MTU exceeded rejection. + pub fn record_mtu_exceeded(&self) { + self.mtu_exceeded.fetch_add(1, Ordering::Relaxed); + } + + /// Record a successful outbound connection. + pub fn record_connection_established(&self) { + self.connections_established.fetch_add(1, Ordering::Relaxed); + } + + /// Record a connect timeout. + pub fn record_connect_timeout(&self) { + self.connect_timeouts.fetch_add(1, Ordering::Relaxed); + } + + /// Record a SOCKS5 protocol error. + pub fn record_socks5_error(&self) { + self.socks5_errors.fetch_add(1, Ordering::Relaxed); + } +} + +impl Default for ProxiedStatsBase { + fn default() -> Self { + Self::new() + } +} diff --git a/src/transport/tor/mock_socks5.rs b/src/transport/tor/mock_socks5.rs deleted file mode 100644 index 91c1e3e..0000000 --- a/src/transport/tor/mock_socks5.rs +++ /dev/null @@ -1,225 +0,0 @@ -//! Mock SOCKS5 server for testing. -//! -//! Implements just enough of the SOCKS5 protocol (RFC 1928) to support -//! the username/password auth + CONNECT flow used by TorTransport. -//! Proxies bytes bidirectionally between the SOCKS5 client and a real -//! TCP target. - -use std::net::SocketAddr; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use tokio::net::TcpListener; -use tokio::task::JoinHandle; - -/// SOCKS5 protocol constants. -const SOCKS_VERSION: u8 = 0x05; -const AUTH_NONE: u8 = 0x00; -const AUTH_PASSWORD: u8 = 0x02; -const CMD_CONNECT: u8 = 0x01; -const ATYP_IPV4: u8 = 0x01; -const ATYP_DOMAIN: u8 = 0x03; -const REP_SUCCESS: u8 = 0x00; - -/// Username/password auth sub-negotiation version (RFC 1929). -const AUTH_SUBNEG_VERSION: u8 = 0x01; -const AUTH_SUBNEG_SUCCESS: u8 = 0x00; - -/// A minimal mock SOCKS5 proxy server for testing. -/// -/// Accepts a single connection, performs the SOCKS5 handshake (supporting -/// both no-auth and username/password auth), then connects to a fixed -/// target address and proxies bytes bidirectionally. -pub struct MockSocks5Server { - /// Address the mock proxy is listening on. - addr: SocketAddr, - /// The real target address to connect to (ignores SOCKS5 requested target). - target_addr: SocketAddr, - /// SOCKS5 reply code (REP field) to send in the CONNECT reply. When this - /// is `REP_SUCCESS` (0x00) the server proxies to `target_addr`; any other - /// value is sent as an error reply and the connection is closed without - /// proxying. Use `0x05` (Connection refused) to drive the refused path. - reply_code: u8, - /// Listener handle. - listener: Option, -} - -impl MockSocks5Server { - /// Create a new mock SOCKS5 server that forwards to the given target. - /// - /// Binds to `127.0.0.1:0` (OS-assigned port). Replies with success and - /// proxies bytes bidirectionally. - pub async fn new(target_addr: SocketAddr) -> std::io::Result { - Self::with_reply_code(target_addr, REP_SUCCESS).await - } - - /// Create a mock SOCKS5 server that replies to the CONNECT request with - /// the given REP code. A non-success code (e.g. `0x05` Connection refused) - /// causes the server to send the error reply and close the connection - /// without proxying, exercising the client's connect-error path. - pub async fn with_reply_code(target_addr: SocketAddr, reply_code: u8) -> std::io::Result { - let listener = TcpListener::bind("127.0.0.1:0").await?; - let addr = listener.local_addr()?; - Ok(Self { - addr, - target_addr, - reply_code, - listener: Some(listener), - }) - } - - /// Get the proxy's listen address (for TorConfig.socks5_addr). - pub fn addr(&self) -> SocketAddr { - self.addr - } - - /// Run the proxy, accepting one connection and proxying it. - /// - /// Returns a JoinHandle that completes when the proxied connection ends. - pub fn spawn(mut self) -> JoinHandle<()> { - let listener = self.listener.take().expect("listener already consumed"); - let target_addr = self.target_addr; - let reply_code = self.reply_code; - - tokio::spawn(async move { - // Accept one SOCKS5 client - let (mut client, _) = listener.accept().await.expect("accept failed"); - - // === Method negotiation === - // Client sends: [version, nmethods, methods...] - let mut ver_nmethods = [0u8; 2]; - client - .read_exact(&mut ver_nmethods) - .await - .expect("read version+nmethods"); - assert_eq!(ver_nmethods[0], SOCKS_VERSION, "expected SOCKS5"); - let nmethods = ver_nmethods[1] as usize; - - let mut methods = vec![0u8; nmethods]; - client.read_exact(&mut methods).await.expect("read methods"); - - // Prefer username/password auth if offered, fall back to no-auth - let selected = if methods.contains(&AUTH_PASSWORD) { - AUTH_PASSWORD - } else if methods.contains(&AUTH_NONE) { - AUTH_NONE - } else { - panic!("no supported auth method offered"); - }; - - // Reply: [version, selected_method] - client - .write_all(&[SOCKS_VERSION, selected]) - .await - .expect("write method reply"); - - // === Username/password sub-negotiation (RFC 1929) === - if selected == AUTH_PASSWORD { - // Client sends: [ver(1), ulen(1), uname(ulen), plen(1), passwd(plen)] - let mut subneg_header = [0u8; 2]; - client - .read_exact(&mut subneg_header) - .await - .expect("read subneg header"); - assert_eq!( - subneg_header[0], AUTH_SUBNEG_VERSION, - "expected auth subneg v1" - ); - - let ulen = subneg_header[1] as usize; - let mut uname = vec![0u8; ulen]; - client.read_exact(&mut uname).await.expect("read username"); - - let mut plen_buf = [0u8; 1]; - client.read_exact(&mut plen_buf).await.expect("read plen"); - let plen = plen_buf[0] as usize; - let mut passwd = vec![0u8; plen]; - client.read_exact(&mut passwd).await.expect("read password"); - - // Always accept (Tor uses these as isolation keys, not real auth) - client - .write_all(&[AUTH_SUBNEG_VERSION, AUTH_SUBNEG_SUCCESS]) - .await - .expect("write subneg reply"); - } - - // === Connect request === - // Client sends: [version, cmd, rsv, atyp, addr..., port] - let mut header = [0u8; 4]; - client - .read_exact(&mut header) - .await - .expect("read connect header"); - assert_eq!(header[0], SOCKS_VERSION); - assert_eq!(header[1], CMD_CONNECT); - - // Read and skip the address (we connect to target_addr regardless) - match header[3] { - ATYP_IPV4 => { - let mut addr_port = [0u8; 6]; // 4 IP + 2 port - client - .read_exact(&mut addr_port) - .await - .expect("read IPv4 addr"); - } - ATYP_DOMAIN => { - let mut len_buf = [0u8; 1]; - client - .read_exact(&mut len_buf) - .await - .expect("read domain len"); - let domain_len = len_buf[0] as usize; - let mut domain_port = vec![0u8; domain_len + 2]; // domain + 2 port - client - .read_exact(&mut domain_port) - .await - .expect("read domain addr"); - } - other => panic!("unsupported ATYP: {}", other), - } - - // Error path: reply with the configured REP code and close without - // proxying (the client maps the REP code to a tokio_socks::Error). - if reply_code != REP_SUCCESS { - let reply = [ - SOCKS_VERSION, - reply_code, - 0x00, // RSV - ATYP_IPV4, - 0, - 0, - 0, - 0, // bind addr - 0, - 0, // bind port - ]; - client - .write_all(&reply) - .await - .expect("write error connect reply"); - return; - } - - // Connect to the real target - let mut target = tokio::net::TcpStream::connect(target_addr) - .await - .expect("connect to target"); - - // Reply: success, bind addr = 0.0.0.0:0 - let reply = [ - SOCKS_VERSION, - REP_SUCCESS, - 0x00, // RSV - ATYP_IPV4, - 0, - 0, - 0, - 0, // bind addr - 0, - 0, // bind port - ]; - client.write_all(&reply).await.expect("write connect reply"); - - // Proxy bytes bidirectionally - let _ = tokio::io::copy_bidirectional(&mut client, &mut target).await; - }) - } -} diff --git a/src/transport/tor/mod.rs b/src/transport/tor/mod.rs index e183632..d7bcab5 100644 --- a/src/transport/tor/mod.rs +++ b/src/transport/tor/mod.rs @@ -23,19 +23,19 @@ pub mod stats; #[cfg(test)] mod mock_control; -#[cfg(test)] -mod mock_socks5; use super::{ - ConnectionState, DiscoveredPeer, PacketTx, ReceivedPacket, Transport, TransportAddr, - TransportError, TransportId, TransportState, TransportType, + ConnectionState, DiscoveredPeer, PacketTx, Transport, TransportAddr, TransportError, + TransportId, TransportState, TransportType, }; use crate::config::TorConfig; -use crate::transport::framing::read_fmp_packet; +use crate::transport::socks5::{ + ConnectingEntry, ConnectingPool, DialError, ProxiedConnection, ProxiedPool, Socks5Auth, + Socks5Dialer, SocksTarget, poll_connecting, proxied_receive_loop, +}; use control::{ControlAuth, TorControlClient, TorMonitoringInfo}; use stats::TorStats; -use futures::FutureExt; use socket2::TcpKeepalive; use std::collections::HashMap; use std::net::SocketAddr; @@ -47,32 +47,22 @@ use tokio::net::{TcpListener, TcpStream}; use tokio::sync::Mutex; use tokio::task::JoinHandle; use tokio::time::Instant; -use tokio_socks::tcp::Socks5Stream; use tracing::{debug, info, trace, warn}; // ============================================================================ -// Tor Address Types +// Address Parsing // ============================================================================ -/// Tor-specific address type for SOCKS5 CONNECT. -#[derive(Clone, Debug)] -pub enum TorAddr { - /// .onion hidden service address (hostname, port). - Onion(String, u16), - /// Clearnet address routed through Tor (IP, port). - Clearnet(SocketAddr), - /// Clearnet hostname routed through Tor (hostname, port). - /// Passed as hostname to SOCKS5 so Tor resolves it — avoids local - /// DNS leaks and is compatible with SafeSocks 1. - ClearnetHostname(String, u16), -} - -/// Parse a TransportAddr string into a TorAddr. +/// Parse a TransportAddr string into a shared SOCKS5 target address. /// -/// If the address contains ".onion:", parse as an onion address. -/// If it parses as a numeric IP:port, use Clearnet. +/// If the address contains ".onion:", parse as an onion hostname. +/// If it parses as a numeric IP:port, use an IP target. /// Otherwise, treat as a clearnet hostname:port for Tor-side DNS resolution. -fn parse_tor_addr(addr: &TransportAddr) -> Result { +/// Onion and clearnet hostnames both become `SocksTarget::Hostname` — they +/// were already dialed identically (`connect_with_password` with a domain +/// target) — while the per-transport parse rules (onion detection, the +/// fully-qualified-hostname requirement) are preserved here. +fn parse_tor_addr(addr: &TransportAddr) -> Result { let s = addr.as_str().ok_or_else(|| { TransportError::InvalidAddress("Tor address must be a valid UTF-8 string".into()) })?; @@ -85,10 +75,10 @@ fn parse_tor_addr(addr: &TransportAddr) -> Result { let port: u16 = port_str.parse().map_err(|_| { TransportError::InvalidAddress(format!("invalid port in onion address: {}", s)) })?; - Ok(TorAddr::Onion(host.to_string(), port)) + Ok(SocksTarget::Hostname(host.to_string(), port)) } else if let Ok(socket_addr) = s.parse::() { // Numeric IP:port - Ok(TorAddr::Clearnet(socket_addr)) + Ok(SocksTarget::Ip(socket_addr)) } else { // Hostname:port — pass through SOCKS5 for Tor-side DNS resolution let (host, port_str) = s.rsplit_once(':').ok_or_else(|| { @@ -103,7 +93,7 @@ fn parse_tor_addr(addr: &TransportAddr) -> Result { host ))); } - Ok(TorAddr::ClearnetHostname(host.to_string(), port)) + Ok(SocksTarget::Hostname(host.to_string(), port)) } } @@ -122,37 +112,6 @@ enum Direction { Outbound, } -/// State for a single Tor connection to a peer. -struct TorConnection { - /// Write half of the split stream. - writer: Arc>, - /// Receive task for this connection. - recv_task: JoinHandle<()>, - /// MTU for this connection. - #[allow(dead_code)] - mtu: u16, - /// When the connection was established. - #[allow(dead_code)] - established_at: Instant, - /// Direction of the connection — drives pool-inbound/outbound accounting. - direction: Direction, -} - -/// Shared connection pool. -type ConnectionPool = Arc>>; - -/// A pending background connection attempt. -/// -/// Holds the JoinHandle for a spawned SOCKS5 connect task. The task -/// produces a configured `TcpStream` and MTU on success. -struct ConnectingEntry { - /// Background task performing SOCKS5 connect + socket configuration. - task: JoinHandle>, -} - -/// Map of addresses with background connection attempts in progress. -type ConnectingPool = Arc>>; - // ============================================================================ // Tor Transport // ============================================================================ @@ -173,7 +132,7 @@ pub struct TorTransport { /// Current state. state: TransportState, /// Connection pool: addr -> per-connection state. - pool: ConnectionPool, + pool: ProxiedPool, /// Pending connection attempts: addr -> background connect task. connecting: ConnectingPool, /// Channel for delivering received packets to Node. @@ -537,14 +496,14 @@ impl TorTransport { for (addr, conn) in pool.drain() { conn.recv_task.abort(); let _ = conn.recv_task.await; - match conn.direction { + match conn.meta { Direction::Inbound => self.stats.record_pool_inbound_removed(), Direction::Outbound => self.stats.record_pool_outbound_removed(), } debug!( transport_id = %self.transport_id, remote_addr = %addr, - direction = ?conn.direction, + direction = ?conn.meta, "Tor connection closed (transport stopping)" ); } @@ -711,7 +670,7 @@ impl TorTransport { let mut pool = self.pool.lock().await; if let Some(conn) = pool.remove(addr) { conn.recv_task.abort(); - match conn.direction { + match conn.meta { Direction::Inbound => self.stats.record_pool_inbound_removed(), Direction::Outbound => self.stats.record_pool_outbound_removed(), } @@ -746,35 +705,19 @@ impl TorTransport { // Uses username/password auth for stream isolation: each destination // gets its own Tor circuit via IsolateSOCKSAuth. The credentials are // not verified by Tor — they serve purely as circuit isolation keys. - let isolation_key = addr.to_string(); - let connect_start = Instant::now(); - let socks_result = tokio::time::timeout(Duration::from_millis(timeout_ms), async { - match &tor_addr { - TorAddr::Onion(host, port) | TorAddr::ClearnetHostname(host, port) => { - Socks5Stream::connect_with_password( - proxy_addr, - (host.as_str(), *port), - "fips", - &isolation_key, - ) - .await - } - TorAddr::Clearnet(socket_addr) => { - Socks5Stream::connect_with_password( - proxy_addr, - *socket_addr, - "fips", - &isolation_key, - ) - .await - } - } - }) - .await; + let dialer = Socks5Dialer { + proxy_addr: proxy_addr.to_string(), + connect_timeout: Duration::from_millis(timeout_ms), + auth: Socks5Auth::Password { + username: "fips".to_string(), + password: addr.to_string(), + }, + }; - let stream = match socks_result { - Ok(Ok(socks_stream)) => socks_stream.into_inner(), - Ok(Err(e)) => { + let connect_start = Instant::now(); + let stream = match dialer.dial(&tor_addr).await { + Ok(stream) => stream, + Err(DialError::Socks(e)) => { // A SOCKS5 REP=0x05 reply is a genuine connection refusal; // every other failure mode (unreachable proxy, ruleset, // protocol/parse, I/O) is a generic SOCKS5 error. @@ -792,7 +735,7 @@ impl TorTransport { ); return Err(TransportError::ConnectionRefused); } - Err(_) => { + Err(DialError::Timeout) => { self.stats.record_connect_timeout(); warn!( transport_id = %self.transport_id, @@ -802,18 +745,9 @@ impl TorTransport { ); return Err(TransportError::Timeout); } + Err(DialError::Setup(e)) => return Err(e), }; - // Configure socket options via socket2 - let std_stream = stream - .into_std() - .map_err(|e| TransportError::StartFailed(format!("into_std: {}", e)))?; - configure_socket(&std_stream, &self.config)?; - - // Convert back to tokio - let stream = TcpStream::from_std(std_stream) - .map_err(|e| TransportError::StartFailed(format!("from_std: {}", e)))?; - // Split and spawn receive task let (read_half, write_half) = stream.into_split(); let writer = Arc::new(Mutex::new(write_half)); @@ -839,12 +773,12 @@ impl TorTransport { .await; }); - let conn = TorConnection { + let conn = ProxiedConnection { writer: writer.clone(), recv_task, mtu, established_at: Instant::now(), - direction: Direction::Outbound, + meta: Direction::Outbound, }; let mut pool = self.pool.lock().await; @@ -913,33 +847,18 @@ impl TorTransport { let task = tokio::spawn(async move { // SOCKS5 CONNECT through proxy with timeout. // Uses username/password auth for stream isolation (see connect()). - let socks_result = tokio::time::timeout(Duration::from_millis(timeout_ms), async { - match &tor_addr { - TorAddr::Onion(host, port) | TorAddr::ClearnetHostname(host, port) => { - Socks5Stream::connect_with_password( - proxy_addr.as_str(), - (host.as_str(), *port), - "fips", - &isolation_key, - ) - .await - } - TorAddr::Clearnet(socket_addr) => { - Socks5Stream::connect_with_password( - proxy_addr.as_str(), - *socket_addr, - "fips", - &isolation_key, - ) - .await - } - } - }) - .await; + let dialer = Socks5Dialer { + proxy_addr, + connect_timeout: Duration::from_millis(timeout_ms), + auth: Socks5Auth::Password { + username: "fips".to_string(), + password: isolation_key, + }, + }; - let stream = match socks_result { - Ok(Ok(socks_stream)) => socks_stream.into_inner(), - Ok(Err(e)) => { + let stream = match dialer.dial(&tor_addr).await { + Ok(stream) => stream, + Err(DialError::Socks(e)) => { // Mirror the synchronous path: count a genuine REP=0x05 // refusal precisely, everything else as a SOCKS5 error. if matches!(e, tokio_socks::Error::ConnectionRefused) { @@ -955,7 +874,7 @@ impl TorTransport { ); return Err(TransportError::ConnectionRefused); } - Err(_) => { + Err(DialError::Timeout) => { debug!( transport_id = %transport_id, remote_addr = %remote_addr, @@ -963,20 +882,11 @@ impl TorTransport { ); return Err(TransportError::Timeout); } + Err(DialError::Setup(e)) => return Err(e), }; - // Configure socket options via socket2 - let std_stream = stream - .into_std() - .map_err(|e| TransportError::StartFailed(format!("into_std: {}", e)))?; - configure_socket(&std_stream, &config)?; - let mtu = config.mtu(); - // Convert back to tokio - let stream = TcpStream::from_std(std_stream) - .map_err(|e| TransportError::StartFailed(format!("from_std: {}", e)))?; - Ok((stream, mtu)) }); @@ -995,52 +905,9 @@ impl TorTransport { /// This method is synchronous but uses `try_lock` internally. /// Returns `ConnectionState::Connecting` if locks can't be acquired. pub fn connection_state_sync(&self, addr: &TransportAddr) -> ConnectionState { - // Check established pool first - if let Ok(pool) = self.pool.try_lock() { - if pool.contains_key(addr) { - return ConnectionState::Connected; - } - } else { - return ConnectionState::Connecting; // can't tell, assume still going - } - - // Check connecting pool - let mut connecting = match self.connecting.try_lock() { - Ok(c) => c, - Err(_) => return ConnectionState::Connecting, - }; - - let entry = match connecting.get_mut(addr) { - Some(e) => e, - None => return ConnectionState::None, - }; - - // Check if the background task has completed - if !entry.task.is_finished() { - return ConnectionState::Connecting; - } - - // Task is done — take the result and remove from connecting pool. - let addr_clone = addr.clone(); - let task = connecting.remove(&addr_clone).unwrap().task; - - // Since the task is finished, we can safely poll it with now_or_never. - match task.now_or_never() { - Some(Ok(Ok((stream, mtu)))) => { - // Promote to established pool - self.promote_connection(addr, stream, mtu); - ConnectionState::Connected - } - Some(Ok(Err(e))) => ConnectionState::Failed(format!("{}", e)), - Some(Err(e)) => { - // JoinError (panic or cancel) - ConnectionState::Failed(format!("task failed: {}", e)) - } - None => { - // Shouldn't happen since is_finished() was true - ConnectionState::Connecting - } - } + poll_connecting(&self.pool, &self.connecting, addr, |stream, mtu| { + self.promote_connection(addr, stream, mtu) + }) } /// Promote a completed background connection to the established pool. @@ -1071,12 +938,12 @@ impl TorTransport { .await; }); - let conn = TorConnection { + let conn = ProxiedConnection { writer, recv_task, mtu, established_at: Instant::now(), - direction: Direction::Outbound, + meta: Direction::Outbound, }; // Use try_lock since we're in a sync context and the pool @@ -1106,7 +973,7 @@ impl TorTransport { let mut pool = self.pool.lock().await; if let Some(conn) = pool.remove(addr) { conn.recv_task.abort(); - match conn.direction { + match conn.meta { Direction::Inbound => self.stats.record_pool_inbound_removed(), Direction::Outbound => self.stats.record_pool_outbound_removed(), } @@ -1173,76 +1040,38 @@ impl Transport for TorTransport { /// Per-connection Tor receive loop. /// -/// Reads complete FMP packets using the stream reader, delivers them to -/// the node via the packet channel. On error or EOF, removes the -/// connection from the pool and exits. `direction` is captured so the -/// cleanup path can decrement the correct `pool_inbound` / -/// `pool_outbound` counter. +/// Thin wrapper over the shared `proxied_receive_loop`. The teardown hook +/// decrements the direction-specific `pool_inbound` / `pool_outbound` counter +/// using the metadata on the removed pool entry, conditional on this loop +/// actually removing it (so a concurrent close/stop never drives the counter +/// below zero). `direction` is retained for the terminal "receive loop +/// stopped" debug field the shared loop deliberately leaves to each transport. #[allow(clippy::too_many_arguments)] async fn tor_receive_loop( - mut reader: tokio::net::tcp::OwnedReadHalf, + reader: tokio::net::tcp::OwnedReadHalf, transport_id: TransportId, remote_addr: TransportAddr, packet_tx: PacketTx, - pool: ConnectionPool, + pool: ProxiedPool, mtu: u16, stats: Arc, direction: Direction, ) { - debug!( - transport_id = %transport_id, - remote_addr = %remote_addr, - "Tor receive loop starting" - ); - - loop { - match read_fmp_packet(&mut reader, mtu).await { - Ok(data) => { - stats.record_recv(data.len()); - - trace!( - transport_id = %transport_id, - remote_addr = %remote_addr, - bytes = data.len(), - "Tor packet received" - ); - - let packet = ReceivedPacket::new(transport_id, remote_addr.clone(), data); - - if packet_tx.send(packet).await.is_err() { - debug!( - transport_id = %transport_id, - "Packet channel closed, stopping Tor receive loop" - ); - break; - } - } - Err(e) => { - stats.record_recv_error(); - debug!( - transport_id = %transport_id, - remote_addr = %remote_addr, - error = %e, - "Tor receive error, removing connection" - ); - break; - } - } - } - - // Clean up: remove ourselves from the pool, then decrement the - // direction-specific pool counter. Decrement is conditional on the - // entry actually being removed so a double-cleanup never drives - // the counter below zero. - let mut pool_guard = pool.lock().await; - let removed = pool_guard.remove(&remote_addr).is_some(); - drop(pool_guard); - if removed { - match direction { + proxied_receive_loop( + reader, + transport_id, + remote_addr.clone(), + packet_tx, + pool, + mtu, + stats, + "Tor", + |stats, meta| match meta { Direction::Inbound => stats.record_pool_inbound_removed(), Direction::Outbound => stats.record_pool_outbound_removed(), - } - } + }, + ) + .await; debug!( transport_id = %transport_id, @@ -1252,36 +1081,6 @@ async fn tor_receive_loop( ); } -// ============================================================================ -// Socket Configuration -// ============================================================================ - -/// Configure socket options on a SOCKS5-connected stream. -/// -/// Sets TCP_NODELAY and keepalive on the underlying TCP connection. -fn configure_socket( - stream: &std::net::TcpStream, - _config: &TorConfig, -) -> Result<(), TransportError> { - let socket = socket2::SockRef::from(stream); - - // TCP_NODELAY — always enable for FIPS (latency-sensitive protocol messages) - socket - .set_tcp_nodelay(true) - .map_err(|e| TransportError::StartFailed(format!("set nodelay: {}", e)))?; - - // TCP keepalive (30s default, matching TCP transport) - let keepalive_secs = 30u64; - if keepalive_secs > 0 { - let keepalive = TcpKeepalive::new().with_time(Duration::from_secs(keepalive_secs)); - socket - .set_tcp_keepalive(&keepalive) - .map_err(|e| TransportError::StartFailed(format!("set keepalive: {}", e)))?; - } - - Ok(()) -} - // ============================================================================ // Accept Loop (onion service inbound) // ============================================================================ @@ -1296,7 +1095,7 @@ async fn tor_accept_loop( listener: TcpListener, transport_id: TransportId, packet_tx: PacketTx, - pool: ConnectionPool, + pool: ProxiedPool, mtu: u16, max_inbound: usize, stats: Arc, @@ -1390,12 +1189,12 @@ async fn tor_accept_loop( .await; }); - let conn = TorConnection { + let conn = ProxiedConnection { writer, recv_task, mtu, established_at: Instant::now(), - direction: Direction::Inbound, + meta: Direction::Inbound, }; { @@ -1455,11 +1254,11 @@ mod tests { let addr = TransportAddr::from_string("abcdef1234567890.onion:2121"); let tor_addr = parse_tor_addr(&addr).unwrap(); match tor_addr { - TorAddr::Onion(host, port) => { + SocksTarget::Hostname(host, port) => { assert_eq!(host, "abcdef1234567890.onion"); assert_eq!(port, 2121); } - _ => panic!("expected Onion variant"), + _ => panic!("expected Hostname variant"), } } @@ -1468,13 +1267,13 @@ mod tests { let addr = TransportAddr::from_string("192.168.1.1:8080"); let tor_addr = parse_tor_addr(&addr).unwrap(); match tor_addr { - TorAddr::Clearnet(socket_addr) => { + SocksTarget::Ip(socket_addr) => { assert_eq!( socket_addr, "192.168.1.1:8080".parse::().unwrap() ); } - _ => panic!("expected Clearnet variant"), + _ => panic!("expected Ip variant"), } } @@ -1483,11 +1282,11 @@ mod tests { let addr = TransportAddr::from_string("peer1.example.com:2121"); let tor_addr = parse_tor_addr(&addr).unwrap(); match tor_addr { - TorAddr::ClearnetHostname(host, port) => { + SocksTarget::Hostname(host, port) => { assert_eq!(host, "peer1.example.com"); assert_eq!(port, 2121); } - _ => panic!("expected ClearnetHostname variant"), + _ => panic!("expected Hostname variant"), } } @@ -1537,11 +1336,11 @@ mod tests { let parsed = parse_tor_addr(&TransportAddr::from_string(&advertised)).unwrap(); match parsed { - TorAddr::Onion(host, port) => { + SocksTarget::Hostname(host, port) => { assert_eq!(host, onion); assert_eq!(port, 443); } - other => panic!("expected Onion variant, got {:?}", other), + other => panic!("expected Hostname variant, got {:?}", other), } // Sanity-check the inverse: the bare-onion form (the bug) must @@ -1655,8 +1454,8 @@ mod tests { // ======================================================================== use crate::config::TcpConfig; + use crate::transport::socks5::mock::MockSocks5Server; use crate::transport::tcp::TcpTransport; - use mock_socks5::MockSocks5Server; /// msg1 wire size: 4 prefix + 4 sender_idx + 106 noise_msg1 = 114 bytes. const MSG1_WIRE_SIZE: usize = 114; @@ -1860,6 +1659,66 @@ mod tests { dest.stop_async().await.unwrap(); } + /// A receive-loop exit and a `close_connection_async` on the same address + /// must decrement the outbound pool counter exactly once. The recv-loop + /// teardown and `close_connection_async` both remove-then-decrement only + /// when their own `remove()` returned the entry, so whichever runs second + /// finds nothing and does not double-decrement (which would wrap the + /// counter far below zero). + #[tokio::test] + async fn test_recv_exit_and_close_decrement_pool_once() { + let (dest_tx, _dest_rx) = packet_channel(32); + let dest_config = TcpConfig { + bind_addr: Some("127.0.0.1:0".to_string()), + ..Default::default() + }; + let mut dest = TcpTransport::new(TransportId::new(100), None, dest_config, dest_tx); + dest.start_async().await.unwrap(); + let dest_addr = dest.local_addr().unwrap(); + + let mock = MockSocks5Server::new(dest_addr).await.unwrap(); + let proxy_addr = mock.addr(); + let _proxy_handle = mock.spawn(); + + let (tor_tx, _tor_rx) = packet_channel(32); + let tor_config = TorConfig { + socks5_addr: Some(proxy_addr.to_string()), + ..Default::default() + }; + let mut tor = TorTransport::new(TransportId::new(200), None, tor_config, tor_tx); + tor.start_async().await.unwrap(); + + // Establish an outbound connection: pool_outbound == 1. + let target = TransportAddr::from_string(&dest_addr.to_string()); + tor.send_async(&target, &build_msg1_frame()).await.unwrap(); + assert_eq!(tor.stats().snapshot().pool_outbound, 1); + + // Drop the destination so the recv loop hits EOF and runs its + // teardown, removing the entry and decrementing exactly once. + dest.stop_async().await.unwrap(); + + let mut waited = 0; + while tor.stats().snapshot().pool_outbound != 0 && waited < 100 { + tokio::time::sleep(Duration::from_millis(20)).await; + waited += 1; + } + assert_eq!( + tor.stats().snapshot().pool_outbound, + 0, + "recv-loop teardown should decrement the outbound counter once" + ); + + // A close on the now-absent address must not decrement again. + tor.close_connection_async(&target).await; + assert_eq!( + tor.stats().snapshot().pool_outbound, + 0, + "the second remover must not decrement (no underflow below zero)" + ); + + tor.stop_async().await.unwrap(); + } + // ======================================================================== // Control port mode tests // ======================================================================== diff --git a/src/transport/tor/stats.rs b/src/transport/tor/stats.rs index 87a1790..cf0d38c 100644 --- a/src/transport/tor/stats.rs +++ b/src/transport/tor/stats.rs @@ -5,23 +5,16 @@ use portable_atomic::{AtomicU64, Ordering}; use serde::Serialize; use crate::transport::PoolCounters; +use crate::transport::socks5::{ProxiedStats, ProxiedStatsBase}; /// Statistics for a Tor transport instance. /// /// Uses atomic counters for lock-free updates from per-connection /// receive loops and the send path concurrently. pub struct TorStats { - pub packets_sent: AtomicU64, - pub bytes_sent: AtomicU64, - pub packets_recv: AtomicU64, - pub bytes_recv: AtomicU64, - pub send_errors: AtomicU64, - pub recv_errors: AtomicU64, - pub mtu_exceeded: AtomicU64, - pub connections_established: AtomicU64, - pub connect_timeouts: AtomicU64, + /// Shared send/receive/connect counters common to proxied transports. + base: ProxiedStatsBase, pub connect_refused: AtomicU64, - pub socks5_errors: AtomicU64, pub connections_accepted: AtomicU64, pub connections_rejected: AtomicU64, pub control_errors: AtomicU64, @@ -35,17 +28,8 @@ impl TorStats { /// Create a new stats instance with all counters at zero. pub fn new() -> Self { Self { - packets_sent: AtomicU64::new(0), - bytes_sent: AtomicU64::new(0), - packets_recv: AtomicU64::new(0), - bytes_recv: AtomicU64::new(0), - send_errors: AtomicU64::new(0), - recv_errors: AtomicU64::new(0), - mtu_exceeded: AtomicU64::new(0), - connections_established: AtomicU64::new(0), - connect_timeouts: AtomicU64::new(0), + base: ProxiedStatsBase::new(), connect_refused: AtomicU64::new(0), - socks5_errors: AtomicU64::new(0), connections_accepted: AtomicU64::new(0), connections_rejected: AtomicU64::new(0), control_errors: AtomicU64::new(0), @@ -55,39 +39,37 @@ impl TorStats { /// Record a successful send. pub fn record_send(&self, bytes: usize) { - self.packets_sent.fetch_add(1, Ordering::Relaxed); - self.bytes_sent.fetch_add(bytes as u64, Ordering::Relaxed); + self.base.record_send(bytes); } /// Record a successful receive. pub fn record_recv(&self, bytes: usize) { - self.packets_recv.fetch_add(1, Ordering::Relaxed); - self.bytes_recv.fetch_add(bytes as u64, Ordering::Relaxed); + self.base.record_recv(bytes); } /// Record a send error. pub fn record_send_error(&self) { - self.send_errors.fetch_add(1, Ordering::Relaxed); + self.base.record_send_error(); } /// Record a receive error. pub fn record_recv_error(&self) { - self.recv_errors.fetch_add(1, Ordering::Relaxed); + self.base.record_recv_error(); } /// Record an MTU exceeded rejection. pub fn record_mtu_exceeded(&self) { - self.mtu_exceeded.fetch_add(1, Ordering::Relaxed); + self.base.record_mtu_exceeded(); } /// Record a successful outbound connection. pub fn record_connection_established(&self) { - self.connections_established.fetch_add(1, Ordering::Relaxed); + self.base.record_connection_established(); } /// Record a connect timeout. pub fn record_connect_timeout(&self) { - self.connect_timeouts.fetch_add(1, Ordering::Relaxed); + self.base.record_connect_timeout(); } /// Record a connection refused. @@ -97,7 +79,7 @@ impl TorStats { /// Record a SOCKS5 protocol error. pub fn record_socks5_error(&self) { - self.socks5_errors.fetch_add(1, Ordering::Relaxed); + self.base.record_socks5_error(); } /// Record a successful inbound connection via onion service. @@ -143,17 +125,17 @@ impl TorStats { /// Take a snapshot of all counters. pub fn snapshot(&self) -> TorStatsSnapshot { TorStatsSnapshot { - packets_sent: self.packets_sent.load(Ordering::Relaxed), - bytes_sent: self.bytes_sent.load(Ordering::Relaxed), - packets_recv: self.packets_recv.load(Ordering::Relaxed), - bytes_recv: self.bytes_recv.load(Ordering::Relaxed), - send_errors: self.send_errors.load(Ordering::Relaxed), - recv_errors: self.recv_errors.load(Ordering::Relaxed), - mtu_exceeded: self.mtu_exceeded.load(Ordering::Relaxed), - connections_established: self.connections_established.load(Ordering::Relaxed), - connect_timeouts: self.connect_timeouts.load(Ordering::Relaxed), + packets_sent: self.base.packets_sent.load(Ordering::Relaxed), + bytes_sent: self.base.bytes_sent.load(Ordering::Relaxed), + packets_recv: self.base.packets_recv.load(Ordering::Relaxed), + bytes_recv: self.base.bytes_recv.load(Ordering::Relaxed), + send_errors: self.base.send_errors.load(Ordering::Relaxed), + recv_errors: self.base.recv_errors.load(Ordering::Relaxed), + mtu_exceeded: self.base.mtu_exceeded.load(Ordering::Relaxed), + connections_established: self.base.connections_established.load(Ordering::Relaxed), + connect_timeouts: self.base.connect_timeouts.load(Ordering::Relaxed), connect_refused: self.connect_refused.load(Ordering::Relaxed), - socks5_errors: self.socks5_errors.load(Ordering::Relaxed), + socks5_errors: self.base.socks5_errors.load(Ordering::Relaxed), connections_accepted: self.connections_accepted.load(Ordering::Relaxed), connections_rejected: self.connections_rejected.load(Ordering::Relaxed), control_errors: self.control_errors.load(Ordering::Relaxed), @@ -169,6 +151,16 @@ impl Default for TorStats { } } +impl ProxiedStats for TorStats { + fn record_recv(&self, bytes: usize) { + self.base.record_recv(bytes); + } + + fn record_recv_error(&self) { + self.base.record_recv_error(); + } +} + /// Point-in-time snapshot of Tor stats (non-atomic, copyable). #[derive(Clone, Debug, Default, Serialize)] pub struct TorStatsSnapshot {