use dashmap::DashMap;
use parking_lot::RwLock;
use std::collections::HashMap;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, Instant};
use thiserror::Error;
use tracing::{debug, info};
#[derive(Debug, Error)]
pub enum TorError {
#[error("Tor not running")]
NotRunning,
#[error("Tor already running")]
AlreadyRunning,
#[error("Connection failed: {0}")]
ConnectionFailed(String),
#[error("Hidden service creation failed: {0}")]
HiddenServiceFailed(String),
#[error("Circuit creation failed: {0}")]
CircuitFailed(String),
#[error("Invalid onion address: {0}")]
InvalidOnionAddress(String),
#[error("SOCKS5 proxy error: {0}")]
Socks5Error(String),
#[error("Tor configuration error: {0}")]
ConfigError(String),
#[error("IO error: {0}")]
Io(#[from] std::io::Error),
}
pub type Result<T> = std::result::Result<T, TorError>;
pub type CircuitId = u32;
pub type StreamId = u64;
pub type OnionAddress = String;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CircuitState {
Building,
Ready,
Active,
Degraded,
Failed,
Closing,
}
#[derive(Debug, Clone)]
pub struct CircuitInfo {
pub id: CircuitId,
pub state: CircuitState,
pub hops: Vec<String>,
pub created_at: Instant,
pub stream_count: usize,
pub bytes_sent: u64,
pub bytes_received: u64,
}
#[derive(Debug, Clone)]
pub struct HiddenServiceConfig {
pub local_port: u16,
pub virtual_port: u16,
pub data_dir: PathBuf,
pub max_connections: usize,
pub use_v3: bool,
}
impl Default for HiddenServiceConfig {
fn default() -> Self {
Self {
local_port: 8080,
virtual_port: 8080,
data_dir: std::env::temp_dir().join("tor-hidden-service"),
max_connections: 100,
use_v3: true,
}
}
}
#[derive(Debug, Clone)]
pub struct TorConfig {
pub socks_proxy: SocketAddr,
pub control_port: SocketAddr,
pub data_dir: PathBuf,
pub stream_isolation: bool,
pub max_circuits: usize,
pub circuit_timeout: Duration,
pub enable_bandwidth_limit: bool,
pub max_bandwidth_bps: u64,
pub use_bridges: bool,
pub bridges: Vec<String>,
}
impl Default for TorConfig {
fn default() -> Self {
Self {
socks_proxy: "127.0.0.1:9050"
.parse()
.expect("static socket addr literal must parse"),
control_port: "127.0.0.1:9051"
.parse()
.expect("static socket addr literal must parse"),
data_dir: std::env::temp_dir().join("tor-data"),
stream_isolation: true,
max_circuits: 10,
circuit_timeout: Duration::from_secs(60),
enable_bandwidth_limit: false,
max_bandwidth_bps: 0,
use_bridges: false,
bridges: Vec::new(),
}
}
}
impl TorConfig {
pub fn high_privacy() -> Self {
Self {
stream_isolation: true,
max_circuits: 5,
circuit_timeout: Duration::from_secs(90),
enable_bandwidth_limit: true,
max_bandwidth_bps: 1_000_000, ..Default::default()
}
}
pub fn high_performance() -> Self {
Self {
stream_isolation: false,
max_circuits: 20,
circuit_timeout: Duration::from_secs(30),
enable_bandwidth_limit: false,
max_bandwidth_bps: 0,
..Default::default()
}
}
pub fn censorship_resistant() -> Self {
Self {
use_bridges: true,
bridges: vec![
"obfs4 192.0.2.1:443".to_string(),
"obfs4 192.0.2.2:443".to_string(),
],
stream_isolation: true,
max_circuits: 8,
circuit_timeout: Duration::from_secs(120),
..Default::default()
}
}
}
#[derive(Debug, Clone, Default)]
pub struct TorStats {
pub circuits_created: usize,
pub active_circuits: usize,
pub streams_created: usize,
pub active_streams: usize,
pub total_bytes_sent: u64,
pub total_bytes_received: u64,
pub hidden_services: usize,
pub connection_failures: usize,
pub avg_circuit_build_time_ms: f64,
}
pub struct TorManager {
config: TorConfig,
running: Arc<RwLock<bool>>,
circuits: Arc<DashMap<CircuitId, CircuitInfo>>,
next_circuit_id: Arc<RwLock<CircuitId>>,
stream_circuits: Arc<DashMap<StreamId, CircuitId>>,
next_stream_id: Arc<RwLock<StreamId>>,
hidden_services: Arc<DashMap<OnionAddress, HiddenServiceConfig>>,
stats: Arc<RwLock<TorStats>>,
circuit_build_times: Arc<RwLock<Vec<f64>>>,
}
impl TorManager {
pub async fn new(config: TorConfig) -> Result<Self> {
info!("Creating Tor manager");
if config.max_circuits == 0 {
return Err(TorError::ConfigError(
"max_circuits must be > 0".to_string(),
));
}
Ok(Self {
config,
running: Arc::new(RwLock::new(false)),
circuits: Arc::new(DashMap::new()),
next_circuit_id: Arc::new(RwLock::new(0)),
stream_circuits: Arc::new(DashMap::new()),
next_stream_id: Arc::new(RwLock::new(0)),
hidden_services: Arc::new(DashMap::new()),
stats: Arc::new(RwLock::new(TorStats::default())),
circuit_build_times: Arc::new(RwLock::new(Vec::with_capacity(100))),
})
}
pub async fn start(&mut self) -> Result<()> {
let mut running = self.running.write();
if *running {
return Err(TorError::AlreadyRunning);
}
info!("Starting Tor manager");
info!("SOCKS proxy: {}", self.config.socks_proxy);
info!("Control port: {}", self.config.control_port);
*running = true;
info!("Tor manager started");
Ok(())
}
pub async fn stop(&mut self) -> Result<()> {
let mut running = self.running.write();
if !*running {
return Err(TorError::NotRunning);
}
info!("Stopping Tor manager");
for circuit in self.circuits.iter() {
let circuit_id = circuit.key();
debug!("Closing circuit {}", circuit_id);
}
self.circuits.clear();
self.hidden_services.clear();
*running = false;
info!("Tor manager stopped");
Ok(())
}
pub fn is_running(&self) -> bool {
*self.running.read()
}
pub async fn create_circuit(&self) -> Result<CircuitId> {
if !self.is_running() {
return Err(TorError::NotRunning);
}
if self.circuits.len() >= self.config.max_circuits {
self.cleanup_circuits();
if self.circuits.len() >= self.config.max_circuits {
return Err(TorError::CircuitFailed(
"Maximum circuits reached".to_string(),
));
}
}
let circuit_id = {
let mut id = self.next_circuit_id.write();
let current_id = *id;
*id += 1;
current_id
};
let start_time = Instant::now();
let circuit = CircuitInfo {
id: circuit_id,
state: CircuitState::Ready,
hops: vec![
"GuardNode".to_string(),
"MiddleNode".to_string(),
"ExitNode".to_string(),
],
created_at: Instant::now(),
stream_count: 0,
bytes_sent: 0,
bytes_received: 0,
};
self.circuits.insert(circuit_id, circuit);
let build_time_ms = start_time.elapsed().as_secs_f64() * 1000.0;
let mut build_times = self.circuit_build_times.write();
build_times.push(build_time_ms);
if build_times.len() > 100 {
build_times.remove(0);
}
let mut stats = self.stats.write();
stats.circuits_created += 1;
stats.active_circuits = self.circuits.len();
stats.avg_circuit_build_time_ms =
build_times.iter().sum::<f64>() / build_times.len() as f64;
info!("Created circuit {} in {:.1}ms", circuit_id, build_time_ms);
Ok(circuit_id)
}
pub async fn connect(&self, address: &str) -> Result<StreamId> {
if !self.is_running() {
return Err(TorError::NotRunning);
}
let circuit_id = if self.config.stream_isolation {
self.create_circuit().await?
} else {
self.get_or_create_circuit().await?
};
let stream_id = {
let mut id = self.next_stream_id.write();
let current_id = *id;
*id += 1;
current_id
};
self.stream_circuits.insert(stream_id, circuit_id);
if let Some(mut circuit) = self.circuits.get_mut(&circuit_id) {
circuit.stream_count += 1;
circuit.state = CircuitState::Active;
}
let mut stats = self.stats.write();
stats.streams_created += 1;
stats.active_streams = self.stream_circuits.len();
debug!("Connected to {} via circuit {}", address, circuit_id);
Ok(stream_id)
}
pub async fn create_hidden_service(&self, config: HiddenServiceConfig) -> Result<OnionAddress> {
if !self.is_running() {
return Err(TorError::NotRunning);
}
let onion_addr = if config.use_v3 {
format!(
"{}.onion",
"abcdefghijklmnopqrstuvabcdefghijklmnopqrstuvabcdefghijkl"
)
} else {
format!("{}.onion", "abcdefghijklmnop")
};
self.hidden_services.insert(onion_addr.clone(), config);
let mut stats = self.stats.write();
stats.hidden_services = self.hidden_services.len();
info!("Created hidden service: {}", onion_addr);
Ok(onion_addr)
}
pub async fn remove_hidden_service(&self, onion_addr: &str) -> Result<()> {
if !self.is_running() {
return Err(TorError::NotRunning);
}
if self.hidden_services.remove(onion_addr).is_some() {
let mut stats = self.stats.write();
stats.hidden_services = self.hidden_services.len();
info!("Removed hidden service: {}", onion_addr);
Ok(())
} else {
Err(TorError::InvalidOnionAddress(onion_addr.to_string()))
}
}
async fn get_or_create_circuit(&self) -> Result<CircuitId> {
let best_circuit = self
.circuits
.iter()
.filter(|entry| {
let circuit = entry.value();
circuit.state == CircuitState::Ready && circuit.stream_count < 10
})
.min_by_key(|entry| entry.value().stream_count)
.map(|entry| *entry.key());
if let Some(circuit_id) = best_circuit {
Ok(circuit_id)
} else {
self.create_circuit().await
}
}
fn cleanup_circuits(&self) {
let now = Instant::now();
let timeout = self.config.circuit_timeout;
let old_circuits: Vec<CircuitId> = self
.circuits
.iter()
.filter(|entry| {
let circuit = entry.value();
circuit.stream_count == 0 && now.duration_since(circuit.created_at) > timeout
})
.map(|entry| *entry.key())
.collect();
for circuit_id in old_circuits {
debug!("Removing old circuit {}", circuit_id);
self.circuits.remove(&circuit_id);
}
let mut stats = self.stats.write();
stats.active_circuits = self.circuits.len();
}
pub fn close_stream(&self, stream_id: StreamId) -> Result<()> {
if let Some((_, circuit_id)) = self.stream_circuits.remove(&stream_id) {
if let Some(mut circuit) = self.circuits.get_mut(&circuit_id) {
circuit.stream_count = circuit.stream_count.saturating_sub(1);
if circuit.stream_count == 0 {
circuit.state = CircuitState::Ready;
}
}
let mut stats = self.stats.write();
stats.active_streams = self.stream_circuits.len();
debug!("Closed stream {}", stream_id);
Ok(())
} else {
Err(TorError::ConnectionFailed(format!(
"Stream {} not found",
stream_id
)))
}
}
pub fn get_circuit(&self, circuit_id: CircuitId) -> Option<CircuitInfo> {
self.circuits.get(&circuit_id).map(|e| e.value().clone())
}
pub fn get_circuits(&self) -> Vec<CircuitInfo> {
self.circuits
.iter()
.map(|entry| entry.value().clone())
.collect()
}
pub fn get_hidden_services(&self) -> HashMap<OnionAddress, HiddenServiceConfig> {
self.hidden_services
.iter()
.map(|entry| (entry.key().clone(), entry.value().clone()))
.collect()
}
pub fn stats(&self) -> TorStats {
self.stats.read().clone()
}
pub fn config(&self) -> &TorConfig {
&self.config
}
pub fn validate_onion_address(address: &str) -> bool {
if !address.ends_with(".onion") {
return false;
}
let name = address
.strip_suffix(".onion")
.expect("just confirmed ends_with('.onion')");
let is_valid_base32 = |c: char| c.is_ascii_lowercase() || ('2'..='7').contains(&c);
if name.len() == 56 {
return name.chars().all(is_valid_base32);
}
if name.len() == 16 {
return name.chars().all(is_valid_base32);
}
false
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_manager_creation() {
let config = TorConfig::default();
let manager = TorManager::new(config)
.await
.expect("test: TorManager::new should succeed");
assert!(!manager.is_running());
assert_eq!(manager.stats().circuits_created, 0);
}
#[tokio::test]
async fn test_start_stop() {
let config = TorConfig::default();
let mut manager = TorManager::new(config)
.await
.expect("test: TorManager::new should succeed");
assert!(!manager.is_running());
manager.start().await.expect("test: start should succeed");
assert!(manager.is_running());
manager.stop().await.expect("test: stop should succeed");
assert!(!manager.is_running());
}
#[tokio::test]
async fn test_create_circuit() {
let config = TorConfig::default();
let mut manager = TorManager::new(config)
.await
.expect("test: TorManager::new should succeed");
manager.start().await.expect("test: start should succeed");
let circuit_id = manager
.create_circuit()
.await
.expect("test: create_circuit should succeed");
assert_eq!(circuit_id, 0);
let circuit = manager
.get_circuit(circuit_id)
.expect("test: circuit should exist after creation");
assert_eq!(circuit.state, CircuitState::Ready);
assert_eq!(circuit.hops.len(), 3);
let stats = manager.stats();
assert_eq!(stats.circuits_created, 1);
assert_eq!(stats.active_circuits, 1);
}
#[tokio::test]
async fn test_max_circuits_limit() {
let config = TorConfig {
max_circuits: 2,
circuit_timeout: Duration::from_millis(100),
..Default::default()
};
let mut manager = TorManager::new(config)
.await
.expect("test: TorManager::new should succeed");
manager.start().await.expect("test: start should succeed");
let circuit1 = manager
.create_circuit()
.await
.expect("test: create first circuit should succeed");
let _circuit2 = manager
.create_circuit()
.await
.expect("test: create second circuit should succeed");
if let Some(mut circuit) = manager.circuits.get_mut(&circuit1) {
circuit.stream_count = 0;
}
tokio::time::sleep(Duration::from_millis(150)).await;
let result = manager.create_circuit().await;
assert!(result.is_ok());
}
#[tokio::test]
async fn test_connect() {
let config = TorConfig::default();
let mut manager = TorManager::new(config)
.await
.expect("test: TorManager::new should succeed");
manager.start().await.expect("test: start should succeed");
let stream_id = manager
.connect("example.onion:8080")
.await
.expect("test: connect should succeed");
assert_eq!(stream_id, 0);
let stats = manager.stats();
assert_eq!(stats.streams_created, 1);
assert_eq!(stats.active_streams, 1);
}
#[tokio::test]
async fn test_stream_isolation() {
let config = TorConfig {
stream_isolation: true,
..Default::default()
};
let mut manager = TorManager::new(config)
.await
.expect("test: TorManager::new should succeed");
manager.start().await.expect("test: start should succeed");
manager
.connect("example1.onion:8080")
.await
.expect("test: connect to example1 should succeed");
manager
.connect("example2.onion:8080")
.await
.expect("test: connect to example2 should succeed");
let stats = manager.stats();
assert_eq!(stats.circuits_created, 2);
assert_eq!(stats.streams_created, 2);
}
#[tokio::test]
async fn test_hidden_service() {
let config = TorConfig::default();
let mut manager = TorManager::new(config)
.await
.expect("test: TorManager::new should succeed");
manager.start().await.expect("test: start should succeed");
let hs_config = HiddenServiceConfig::default();
let onion_addr = manager
.create_hidden_service(hs_config)
.await
.expect("test: create_hidden_service should succeed");
assert!(onion_addr.ends_with(".onion"));
assert_eq!(onion_addr.len(), 62);
let stats = manager.stats();
assert_eq!(stats.hidden_services, 1);
}
#[tokio::test]
async fn test_remove_hidden_service() {
let config = TorConfig::default();
let mut manager = TorManager::new(config)
.await
.expect("test: TorManager::new should succeed");
manager.start().await.expect("test: start should succeed");
let hs_config = HiddenServiceConfig::default();
let onion_addr = manager
.create_hidden_service(hs_config)
.await
.expect("test: create_hidden_service should succeed");
manager
.remove_hidden_service(&onion_addr)
.await
.expect("test: remove_hidden_service should succeed");
let stats = manager.stats();
assert_eq!(stats.hidden_services, 0);
}
#[tokio::test]
async fn test_close_stream() {
let config = TorConfig::default();
let mut manager = TorManager::new(config)
.await
.expect("test: TorManager::new should succeed");
manager.start().await.expect("test: start should succeed");
let stream_id = manager
.connect("example.onion:8080")
.await
.expect("test: connect should succeed");
manager
.close_stream(stream_id)
.expect("test: close_stream should succeed");
let stats = manager.stats();
assert_eq!(stats.active_streams, 0);
}
#[tokio::test]
async fn test_config_presets() {
let high_privacy = TorConfig::high_privacy();
assert!(high_privacy.stream_isolation);
assert_eq!(high_privacy.max_circuits, 5);
let high_performance = TorConfig::high_performance();
assert!(!high_performance.stream_isolation);
assert_eq!(high_performance.max_circuits, 20);
let censorship = TorConfig::censorship_resistant();
assert!(censorship.use_bridges);
assert!(!censorship.bridges.is_empty());
}
#[test]
fn test_validate_onion_address() {
let v3_addr = "abcdefghijklmnopqrstuvabcdefghijklmnopqrstuvabcdefghijkl.onion";
assert!(TorManager::validate_onion_address(v3_addr));
let v2_addr = "abcdefghijklmnop.onion";
assert!(TorManager::validate_onion_address(v2_addr));
assert!(!TorManager::validate_onion_address("invalid"));
assert!(!TorManager::validate_onion_address("example.com"));
assert!(!TorManager::validate_onion_address("abc.onion"));
assert!(!TorManager::validate_onion_address(
"abcdefghijklmnopqrstuvwxyz234567abcdefghijklmnopqrstuvw.onion"
));
}
#[tokio::test]
async fn test_circuit_cleanup() {
let config = TorConfig {
circuit_timeout: Duration::from_millis(100),
..Default::default()
};
let mut manager = TorManager::new(config)
.await
.expect("test: TorManager::new should succeed");
manager.start().await.expect("test: start should succeed");
let _circuit_id = manager
.create_circuit()
.await
.expect("test: create_circuit should succeed");
tokio::time::sleep(Duration::from_millis(150)).await;
manager.cleanup_circuits();
}
#[tokio::test]
async fn test_not_running_errors() {
let config = TorConfig::default();
let manager = TorManager::new(config)
.await
.expect("test: TorManager::new should succeed");
assert!(manager.create_circuit().await.is_err());
assert!(manager.connect("example.onion:8080").await.is_err());
let hs_config = HiddenServiceConfig::default();
assert!(manager.create_hidden_service(hs_config).await.is_err());
}
}