laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-07-30 19:46:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'refactor-transport' into refactor-next

Browse Source
This commit is contained in:
Johnathan Corgan
2026-07-11 18:46:00 +00:00
parent 14edddc14a 32475d859e
commit 775c7ae4c0
11 changed files with 901 additions and 829 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/transport/mod.rs
+1
View File
@@ -7,6 +7,7 @@
#[cfg(test)]
pub mod loopback;
pub mod nym;
pub mod socks5;
pub mod tcp;
pub mod tor;
pub mod udp;
src/transport/nym/mod.rs
+63 -215
View File
@@ -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<Mutex<OwnedWriteHalf>>,
/// 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<Mutex<HashMap<TransportAddr, NymConnection>>>;
/// A pending background connection attempt.
struct ConnectingEntry {
/// Background task performing SOCKS5 connect + socket configuration.
task: JoinHandle<Result<(TcpStream, u16), TransportError>>,
}
/// Map of addresses with background connection attempts in progress.
type ConnectingPool = Arc<Mutex<HashMap<TransportAddr, ConnectingEntry>>>;
// ============================================================================
// 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<TargetAddr, TransportError> {
/// Parse a TransportAddr string into a shared SOCKS5 target address.
fn parse_target_addr(addr: &TransportAddr) -> Result<SocksTarget, TransportError> {
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::<SocketAddr>() {
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<TargetAddr, TransportError>
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<TargetAddr, TransportError>
// ============================================================================
/// 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<NymStats>,
) {
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::<SocketAddr>().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::<SocketAddr>().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 for the XX handshake: 4 prefix + 37 payload = 41 bytes.
const MSG1_WIRE_SIZE: usize = 41;
src/transport/nym/stats.rs
+34 -41
View File
@@ -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 {
src/transport/socks5/dialer.rs
+238
View File
@@ -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<TcpStream, DialError> {
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)));
}
}
src/transport/nym/mock_socks5.rs → src/transport/socks5/mock.rs
+62 -17
View File
@@ -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<TcpListener>,
}
@@ -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> {
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<Self> {
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
src/transport/socks5/mod.rs
+20
View File
@@ -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;
src/transport/socks5/pool.rs
+211
View File
@@ -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<M> {
/// Write half of the split stream.
pub writer: Arc<Mutex<OwnedWriteHalf>>,
/// 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<M> = Arc<Mutex<HashMap<TransportAddr, ProxiedConnection<M>>>>;
/// 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<Result<(TcpStream, u16), TransportError>>,
}
/// Map of addresses with background connection attempts in progress.
pub(crate) type ConnectingPool = Arc<Mutex<HashMap<TransportAddr, ConnectingEntry>>>;
/// 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<M>(
pool: &ProxiedPool<M>,
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<S: ProxiedStats, M>(
mut reader: OwnedReadHalf,
transport_id: TransportId,
remote_addr: TransportAddr,
packet_tx: PacketTx,
pool: ProxiedPool<M>,
mtu: u16,
stats: Arc<S>,
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);
}
}
src/transport/socks5/stats.rs
+90
View File
@@ -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()
}
}
src/transport/tor/mock_socks5.rs
-225
View File
@@ -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<TcpListener>,
}
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> {
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<Self> {
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;
})
}
}
src/transport/tor/mod.rs
+150 -291
View File
@@ -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<TorAddr, TransportError> {
/// 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<SocksTarget, TransportError> {
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<TorAddr, TransportError> {
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::<SocketAddr>() {
// 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<TorAddr, TransportError> {
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<Mutex<OwnedWriteHalf>>,
/// 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<Mutex<HashMap<TransportAddr, TorConnection>>>;
/// 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<Result<(TcpStream, u16), TransportError>>,
}
/// Map of addresses with background connection attempts in progress.
type ConnectingPool = Arc<Mutex<HashMap<TransportAddr, ConnectingEntry>>>;
// ============================================================================
// Tor Transport
// ============================================================================
@@ -173,7 +132,7 @@ pub struct TorTransport {
/// Current state.
state: TransportState,
/// Connection pool: addr -> per-connection state.
pool: ConnectionPool,
pool: ProxiedPool<Direction>,
/// 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<Direction>,
mtu: u16,
stats: Arc<TorStats>,
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<Direction>,
mtu: u16,
max_inbound: usize,
stats: Arc<TorStats>,
@@ -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::<SocketAddr>().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 = 41;
@@ -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
// ========================================================================
src/transport/tor/stats.rs
+32 -40
View File
@@ -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 {