1use dashmap::DashMap;
49use parking_lot::RwLock;
50use std::collections::HashMap;
51use std::net::SocketAddr;
52use std::path::PathBuf;
53use std::sync::Arc;
54use std::time::{Duration, Instant};
55use thiserror::Error;
56use tracing::{debug, info};
57
58#[derive(Debug, Error)]
60pub enum TorError {
61 #[error("Tor not running")]
62 NotRunning,
63
64 #[error("Tor already running")]
65 AlreadyRunning,
66
67 #[error("Connection failed: {0}")]
68 ConnectionFailed(String),
69
70 #[error("Hidden service creation failed: {0}")]
71 HiddenServiceFailed(String),
72
73 #[error("Circuit creation failed: {0}")]
74 CircuitFailed(String),
75
76 #[error("Invalid onion address: {0}")]
77 InvalidOnionAddress(String),
78
79 #[error("SOCKS5 proxy error: {0}")]
80 Socks5Error(String),
81
82 #[error("Tor configuration error: {0}")]
83 ConfigError(String),
84
85 #[error("IO error: {0}")]
86 Io(#[from] std::io::Error),
87}
88
89pub type Result<T> = std::result::Result<T, TorError>;
91
92pub type CircuitId = u32;
94
95pub type StreamId = u64;
97
98pub type OnionAddress = String;
100
101#[derive(Debug, Clone, Copy, PartialEq, Eq)]
103pub enum CircuitState {
104 Building,
106
107 Ready,
109
110 Active,
112
113 Degraded,
115
116 Failed,
118
119 Closing,
121}
122
123#[derive(Debug, Clone)]
125pub struct CircuitInfo {
126 pub id: CircuitId,
128
129 pub state: CircuitState,
131
132 pub hops: Vec<String>,
134
135 pub created_at: Instant,
137
138 pub stream_count: usize,
140
141 pub bytes_sent: u64,
143
144 pub bytes_received: u64,
146}
147
148#[derive(Debug, Clone)]
150pub struct HiddenServiceConfig {
151 pub local_port: u16,
153
154 pub virtual_port: u16,
156
157 pub data_dir: PathBuf,
159
160 pub max_connections: usize,
162
163 pub use_v3: bool,
165}
166
167impl Default for HiddenServiceConfig {
168 fn default() -> Self {
169 Self {
170 local_port: 8080,
171 virtual_port: 8080,
172 data_dir: std::env::temp_dir().join("tor-hidden-service"),
173 max_connections: 100,
174 use_v3: true,
175 }
176 }
177}
178
179#[derive(Debug, Clone)]
181pub struct TorConfig {
182 pub socks_proxy: SocketAddr,
184
185 pub control_port: SocketAddr,
187
188 pub data_dir: PathBuf,
190
191 pub stream_isolation: bool,
193
194 pub max_circuits: usize,
196
197 pub circuit_timeout: Duration,
199
200 pub enable_bandwidth_limit: bool,
202
203 pub max_bandwidth_bps: u64,
205
206 pub use_bridges: bool,
208
209 pub bridges: Vec<String>,
211}
212
213impl Default for TorConfig {
214 fn default() -> Self {
215 Self {
216 socks_proxy: "127.0.0.1:9050"
217 .parse()
218 .expect("static socket addr literal must parse"),
219 control_port: "127.0.0.1:9051"
220 .parse()
221 .expect("static socket addr literal must parse"),
222 data_dir: std::env::temp_dir().join("tor-data"),
223 stream_isolation: true,
224 max_circuits: 10,
225 circuit_timeout: Duration::from_secs(60),
226 enable_bandwidth_limit: false,
227 max_bandwidth_bps: 0,
228 use_bridges: false,
229 bridges: Vec::new(),
230 }
231 }
232}
233
234impl TorConfig {
235 pub fn high_privacy() -> Self {
237 Self {
238 stream_isolation: true,
239 max_circuits: 5,
240 circuit_timeout: Duration::from_secs(90),
241 enable_bandwidth_limit: true,
242 max_bandwidth_bps: 1_000_000, ..Default::default()
244 }
245 }
246
247 pub fn high_performance() -> Self {
249 Self {
250 stream_isolation: false,
251 max_circuits: 20,
252 circuit_timeout: Duration::from_secs(30),
253 enable_bandwidth_limit: false,
254 max_bandwidth_bps: 0,
255 ..Default::default()
256 }
257 }
258
259 pub fn censorship_resistant() -> Self {
261 Self {
262 use_bridges: true,
263 bridges: vec![
264 "obfs4 192.0.2.1:443".to_string(),
266 "obfs4 192.0.2.2:443".to_string(),
267 ],
268 stream_isolation: true,
269 max_circuits: 8,
270 circuit_timeout: Duration::from_secs(120),
271 ..Default::default()
272 }
273 }
274}
275
276#[derive(Debug, Clone, Default)]
278pub struct TorStats {
279 pub circuits_created: usize,
281
282 pub active_circuits: usize,
284
285 pub streams_created: usize,
287
288 pub active_streams: usize,
290
291 pub total_bytes_sent: u64,
293
294 pub total_bytes_received: u64,
296
297 pub hidden_services: usize,
299
300 pub connection_failures: usize,
302
303 pub avg_circuit_build_time_ms: f64,
305}
306
307pub struct TorManager {
309 config: TorConfig,
311
312 running: Arc<RwLock<bool>>,
314
315 circuits: Arc<DashMap<CircuitId, CircuitInfo>>,
317
318 next_circuit_id: Arc<RwLock<CircuitId>>,
320
321 stream_circuits: Arc<DashMap<StreamId, CircuitId>>,
323
324 next_stream_id: Arc<RwLock<StreamId>>,
326
327 hidden_services: Arc<DashMap<OnionAddress, HiddenServiceConfig>>,
329
330 stats: Arc<RwLock<TorStats>>,
332
333 circuit_build_times: Arc<RwLock<Vec<f64>>>,
335}
336
337impl TorManager {
338 pub async fn new(config: TorConfig) -> Result<Self> {
340 info!("Creating Tor manager");
341
342 if config.max_circuits == 0 {
344 return Err(TorError::ConfigError(
345 "max_circuits must be > 0".to_string(),
346 ));
347 }
348
349 Ok(Self {
350 config,
351 running: Arc::new(RwLock::new(false)),
352 circuits: Arc::new(DashMap::new()),
353 next_circuit_id: Arc::new(RwLock::new(0)),
354 stream_circuits: Arc::new(DashMap::new()),
355 next_stream_id: Arc::new(RwLock::new(0)),
356 hidden_services: Arc::new(DashMap::new()),
357 stats: Arc::new(RwLock::new(TorStats::default())),
358 circuit_build_times: Arc::new(RwLock::new(Vec::with_capacity(100))),
359 })
360 }
361
362 pub async fn start(&mut self) -> Result<()> {
364 let mut running = self.running.write();
365 if *running {
366 return Err(TorError::AlreadyRunning);
367 }
368
369 info!("Starting Tor manager");
370 info!("SOCKS proxy: {}", self.config.socks_proxy);
371 info!("Control port: {}", self.config.control_port);
372
373 *running = true;
380 info!("Tor manager started");
381
382 Ok(())
383 }
384
385 pub async fn stop(&mut self) -> Result<()> {
387 let mut running = self.running.write();
388 if !*running {
389 return Err(TorError::NotRunning);
390 }
391
392 info!("Stopping Tor manager");
393
394 for circuit in self.circuits.iter() {
396 let circuit_id = circuit.key();
397 debug!("Closing circuit {}", circuit_id);
398 }
399 self.circuits.clear();
400
401 self.hidden_services.clear();
403
404 *running = false;
405 info!("Tor manager stopped");
406
407 Ok(())
408 }
409
410 pub fn is_running(&self) -> bool {
412 *self.running.read()
413 }
414
415 pub async fn create_circuit(&self) -> Result<CircuitId> {
417 if !self.is_running() {
418 return Err(TorError::NotRunning);
419 }
420
421 if self.circuits.len() >= self.config.max_circuits {
422 self.cleanup_circuits();
424
425 if self.circuits.len() >= self.config.max_circuits {
426 return Err(TorError::CircuitFailed(
427 "Maximum circuits reached".to_string(),
428 ));
429 }
430 }
431
432 let circuit_id = {
433 let mut id = self.next_circuit_id.write();
434 let current_id = *id;
435 *id += 1;
436 current_id
437 };
438
439 let start_time = Instant::now();
440
441 let circuit = CircuitInfo {
448 id: circuit_id,
449 state: CircuitState::Ready,
450 hops: vec![
451 "GuardNode".to_string(),
452 "MiddleNode".to_string(),
453 "ExitNode".to_string(),
454 ],
455 created_at: Instant::now(),
456 stream_count: 0,
457 bytes_sent: 0,
458 bytes_received: 0,
459 };
460
461 self.circuits.insert(circuit_id, circuit);
462
463 let build_time_ms = start_time.elapsed().as_secs_f64() * 1000.0;
465 let mut build_times = self.circuit_build_times.write();
466 build_times.push(build_time_ms);
467 if build_times.len() > 100 {
468 build_times.remove(0);
469 }
470
471 let mut stats = self.stats.write();
472 stats.circuits_created += 1;
473 stats.active_circuits = self.circuits.len();
474 stats.avg_circuit_build_time_ms =
475 build_times.iter().sum::<f64>() / build_times.len() as f64;
476
477 info!("Created circuit {} in {:.1}ms", circuit_id, build_time_ms);
478
479 Ok(circuit_id)
480 }
481
482 pub async fn connect(&self, address: &str) -> Result<StreamId> {
484 if !self.is_running() {
485 return Err(TorError::NotRunning);
486 }
487
488 let circuit_id = if self.config.stream_isolation {
490 self.create_circuit().await?
492 } else {
493 self.get_or_create_circuit().await?
495 };
496
497 let stream_id = {
498 let mut id = self.next_stream_id.write();
499 let current_id = *id;
500 *id += 1;
501 current_id
502 };
503
504 self.stream_circuits.insert(stream_id, circuit_id);
512
513 if let Some(mut circuit) = self.circuits.get_mut(&circuit_id) {
515 circuit.stream_count += 1;
516 circuit.state = CircuitState::Active;
517 }
518
519 let mut stats = self.stats.write();
520 stats.streams_created += 1;
521 stats.active_streams = self.stream_circuits.len();
522
523 debug!("Connected to {} via circuit {}", address, circuit_id);
524
525 Ok(stream_id)
526 }
527
528 pub async fn create_hidden_service(&self, config: HiddenServiceConfig) -> Result<OnionAddress> {
530 if !self.is_running() {
531 return Err(TorError::NotRunning);
532 }
533
534 let onion_addr = if config.use_v3 {
542 format!(
543 "{}.onion",
544 "abcdefghijklmnopqrstuvabcdefghijklmnopqrstuvabcdefghijkl"
545 )
546 } else {
547 format!("{}.onion", "abcdefghijklmnop")
549 };
550
551 self.hidden_services.insert(onion_addr.clone(), config);
552
553 let mut stats = self.stats.write();
554 stats.hidden_services = self.hidden_services.len();
555
556 info!("Created hidden service: {}", onion_addr);
557
558 Ok(onion_addr)
559 }
560
561 pub async fn remove_hidden_service(&self, onion_addr: &str) -> Result<()> {
563 if !self.is_running() {
564 return Err(TorError::NotRunning);
565 }
566
567 if self.hidden_services.remove(onion_addr).is_some() {
568 let mut stats = self.stats.write();
569 stats.hidden_services = self.hidden_services.len();
570
571 info!("Removed hidden service: {}", onion_addr);
572 Ok(())
573 } else {
574 Err(TorError::InvalidOnionAddress(onion_addr.to_string()))
575 }
576 }
577
578 async fn get_or_create_circuit(&self) -> Result<CircuitId> {
580 let best_circuit = self
582 .circuits
583 .iter()
584 .filter(|entry| {
585 let circuit = entry.value();
586 circuit.state == CircuitState::Ready && circuit.stream_count < 10
587 })
588 .min_by_key(|entry| entry.value().stream_count)
589 .map(|entry| *entry.key());
590
591 if let Some(circuit_id) = best_circuit {
592 Ok(circuit_id)
593 } else {
594 self.create_circuit().await
595 }
596 }
597
598 fn cleanup_circuits(&self) {
600 let now = Instant::now();
601 let timeout = self.config.circuit_timeout;
602
603 let old_circuits: Vec<CircuitId> = self
604 .circuits
605 .iter()
606 .filter(|entry| {
607 let circuit = entry.value();
608 circuit.stream_count == 0 && now.duration_since(circuit.created_at) > timeout
609 })
610 .map(|entry| *entry.key())
611 .collect();
612
613 for circuit_id in old_circuits {
614 debug!("Removing old circuit {}", circuit_id);
615 self.circuits.remove(&circuit_id);
616 }
617
618 let mut stats = self.stats.write();
619 stats.active_circuits = self.circuits.len();
620 }
621
622 pub fn close_stream(&self, stream_id: StreamId) -> Result<()> {
624 if let Some((_, circuit_id)) = self.stream_circuits.remove(&stream_id) {
625 if let Some(mut circuit) = self.circuits.get_mut(&circuit_id) {
627 circuit.stream_count = circuit.stream_count.saturating_sub(1);
628
629 if circuit.stream_count == 0 {
630 circuit.state = CircuitState::Ready;
631 }
632 }
633
634 let mut stats = self.stats.write();
635 stats.active_streams = self.stream_circuits.len();
636
637 debug!("Closed stream {}", stream_id);
638 Ok(())
639 } else {
640 Err(TorError::ConnectionFailed(format!(
641 "Stream {} not found",
642 stream_id
643 )))
644 }
645 }
646
647 pub fn get_circuit(&self, circuit_id: CircuitId) -> Option<CircuitInfo> {
649 self.circuits.get(&circuit_id).map(|e| e.value().clone())
650 }
651
652 pub fn get_circuits(&self) -> Vec<CircuitInfo> {
654 self.circuits
655 .iter()
656 .map(|entry| entry.value().clone())
657 .collect()
658 }
659
660 pub fn get_hidden_services(&self) -> HashMap<OnionAddress, HiddenServiceConfig> {
662 self.hidden_services
663 .iter()
664 .map(|entry| (entry.key().clone(), entry.value().clone()))
665 .collect()
666 }
667
668 pub fn stats(&self) -> TorStats {
670 self.stats.read().clone()
671 }
672
673 pub fn config(&self) -> &TorConfig {
675 &self.config
676 }
677
678 pub fn validate_onion_address(address: &str) -> bool {
680 if !address.ends_with(".onion") {
684 return false;
685 }
686
687 let name = address
688 .strip_suffix(".onion")
689 .expect("just confirmed ends_with('.onion')");
690
691 let is_valid_base32 = |c: char| c.is_ascii_lowercase() || ('2'..='7').contains(&c);
693
694 if name.len() == 56 {
696 return name.chars().all(is_valid_base32);
697 }
698
699 if name.len() == 16 {
701 return name.chars().all(is_valid_base32);
702 }
703
704 false
705 }
706}
707
708#[cfg(test)]
709mod tests {
710 use super::*;
711
712 #[tokio::test]
713 async fn test_manager_creation() {
714 let config = TorConfig::default();
715 let manager = TorManager::new(config)
716 .await
717 .expect("test: TorManager::new should succeed");
718
719 assert!(!manager.is_running());
720 assert_eq!(manager.stats().circuits_created, 0);
721 }
722
723 #[tokio::test]
724 async fn test_start_stop() {
725 let config = TorConfig::default();
726 let mut manager = TorManager::new(config)
727 .await
728 .expect("test: TorManager::new should succeed");
729
730 assert!(!manager.is_running());
731
732 manager.start().await.expect("test: start should succeed");
733 assert!(manager.is_running());
734
735 manager.stop().await.expect("test: stop should succeed");
736 assert!(!manager.is_running());
737 }
738
739 #[tokio::test]
740 async fn test_create_circuit() {
741 let config = TorConfig::default();
742 let mut manager = TorManager::new(config)
743 .await
744 .expect("test: TorManager::new should succeed");
745
746 manager.start().await.expect("test: start should succeed");
747
748 let circuit_id = manager
749 .create_circuit()
750 .await
751 .expect("test: create_circuit should succeed");
752 assert_eq!(circuit_id, 0);
753
754 let circuit = manager
755 .get_circuit(circuit_id)
756 .expect("test: circuit should exist after creation");
757 assert_eq!(circuit.state, CircuitState::Ready);
758 assert_eq!(circuit.hops.len(), 3);
759
760 let stats = manager.stats();
761 assert_eq!(stats.circuits_created, 1);
762 assert_eq!(stats.active_circuits, 1);
763 }
764
765 #[tokio::test]
766 async fn test_max_circuits_limit() {
767 let config = TorConfig {
768 max_circuits: 2,
769 circuit_timeout: Duration::from_millis(100),
770 ..Default::default()
771 };
772 let mut manager = TorManager::new(config)
773 .await
774 .expect("test: TorManager::new should succeed");
775
776 manager.start().await.expect("test: start should succeed");
777
778 let circuit1 = manager
780 .create_circuit()
781 .await
782 .expect("test: create first circuit should succeed");
783 let _circuit2 = manager
784 .create_circuit()
785 .await
786 .expect("test: create second circuit should succeed");
787
788 if let Some(mut circuit) = manager.circuits.get_mut(&circuit1) {
790 circuit.stream_count = 0;
791 }
792
793 tokio::time::sleep(Duration::from_millis(150)).await;
795
796 let result = manager.create_circuit().await;
798 assert!(result.is_ok());
799 }
800
801 #[tokio::test]
802 async fn test_connect() {
803 let config = TorConfig::default();
804 let mut manager = TorManager::new(config)
805 .await
806 .expect("test: TorManager::new should succeed");
807
808 manager.start().await.expect("test: start should succeed");
809
810 let stream_id = manager
811 .connect("example.onion:8080")
812 .await
813 .expect("test: connect should succeed");
814 assert_eq!(stream_id, 0);
815
816 let stats = manager.stats();
817 assert_eq!(stats.streams_created, 1);
818 assert_eq!(stats.active_streams, 1);
819 }
820
821 #[tokio::test]
822 async fn test_stream_isolation() {
823 let config = TorConfig {
824 stream_isolation: true,
825 ..Default::default()
826 };
827 let mut manager = TorManager::new(config)
828 .await
829 .expect("test: TorManager::new should succeed");
830
831 manager.start().await.expect("test: start should succeed");
832
833 manager
835 .connect("example1.onion:8080")
836 .await
837 .expect("test: connect to example1 should succeed");
838 manager
839 .connect("example2.onion:8080")
840 .await
841 .expect("test: connect to example2 should succeed");
842
843 let stats = manager.stats();
844 assert_eq!(stats.circuits_created, 2);
845 assert_eq!(stats.streams_created, 2);
846 }
847
848 #[tokio::test]
849 async fn test_hidden_service() {
850 let config = TorConfig::default();
851 let mut manager = TorManager::new(config)
852 .await
853 .expect("test: TorManager::new should succeed");
854
855 manager.start().await.expect("test: start should succeed");
856
857 let hs_config = HiddenServiceConfig::default();
858 let onion_addr = manager
859 .create_hidden_service(hs_config)
860 .await
861 .expect("test: create_hidden_service should succeed");
862
863 assert!(onion_addr.ends_with(".onion"));
864 assert_eq!(onion_addr.len(), 62); let stats = manager.stats();
867 assert_eq!(stats.hidden_services, 1);
868 }
869
870 #[tokio::test]
871 async fn test_remove_hidden_service() {
872 let config = TorConfig::default();
873 let mut manager = TorManager::new(config)
874 .await
875 .expect("test: TorManager::new should succeed");
876
877 manager.start().await.expect("test: start should succeed");
878
879 let hs_config = HiddenServiceConfig::default();
880 let onion_addr = manager
881 .create_hidden_service(hs_config)
882 .await
883 .expect("test: create_hidden_service should succeed");
884
885 manager
886 .remove_hidden_service(&onion_addr)
887 .await
888 .expect("test: remove_hidden_service should succeed");
889
890 let stats = manager.stats();
891 assert_eq!(stats.hidden_services, 0);
892 }
893
894 #[tokio::test]
895 async fn test_close_stream() {
896 let config = TorConfig::default();
897 let mut manager = TorManager::new(config)
898 .await
899 .expect("test: TorManager::new should succeed");
900
901 manager.start().await.expect("test: start should succeed");
902
903 let stream_id = manager
904 .connect("example.onion:8080")
905 .await
906 .expect("test: connect should succeed");
907 manager
908 .close_stream(stream_id)
909 .expect("test: close_stream should succeed");
910
911 let stats = manager.stats();
912 assert_eq!(stats.active_streams, 0);
913 }
914
915 #[tokio::test]
916 async fn test_config_presets() {
917 let high_privacy = TorConfig::high_privacy();
918 assert!(high_privacy.stream_isolation);
919 assert_eq!(high_privacy.max_circuits, 5);
920
921 let high_performance = TorConfig::high_performance();
922 assert!(!high_performance.stream_isolation);
923 assert_eq!(high_performance.max_circuits, 20);
924
925 let censorship = TorConfig::censorship_resistant();
926 assert!(censorship.use_bridges);
927 assert!(!censorship.bridges.is_empty());
928 }
929
930 #[test]
931 fn test_validate_onion_address() {
932 let v3_addr = "abcdefghijklmnopqrstuvabcdefghijklmnopqrstuvabcdefghijkl.onion";
934 assert!(TorManager::validate_onion_address(v3_addr));
935
936 let v2_addr = "abcdefghijklmnop.onion";
938 assert!(TorManager::validate_onion_address(v2_addr));
939
940 assert!(!TorManager::validate_onion_address("invalid"));
942 assert!(!TorManager::validate_onion_address("example.com"));
943 assert!(!TorManager::validate_onion_address("abc.onion")); assert!(!TorManager::validate_onion_address(
947 "abcdefghijklmnopqrstuvwxyz234567abcdefghijklmnopqrstuvw.onion"
948 ));
949 }
950
951 #[tokio::test]
952 async fn test_circuit_cleanup() {
953 let config = TorConfig {
954 circuit_timeout: Duration::from_millis(100),
955 ..Default::default()
956 };
957 let mut manager = TorManager::new(config)
958 .await
959 .expect("test: TorManager::new should succeed");
960
961 manager.start().await.expect("test: start should succeed");
962
963 let _circuit_id = manager
965 .create_circuit()
966 .await
967 .expect("test: create_circuit should succeed");
968
969 tokio::time::sleep(Duration::from_millis(150)).await;
971
972 manager.cleanup_circuits();
974
975 }
978
979 #[tokio::test]
980 async fn test_not_running_errors() {
981 let config = TorConfig::default();
982 let manager = TorManager::new(config)
983 .await
984 .expect("test: TorManager::new should succeed");
985
986 assert!(manager.create_circuit().await.is_err());
988 assert!(manager.connect("example.onion:8080").await.is_err());
989
990 let hs_config = HiddenServiceConfig::default();
991 assert!(manager.create_hidden_service(hs_config).await.is_err());
992 }
993}