1use crate::{logs::LogEntry, process::PeerProcess, Error, Result};
8use bollard::{
9 container::{
10 Config, CreateContainerOptions, LogOutput, LogsOptions, RemoveContainerOptions,
11 StartContainerOptions, StopContainerOptions, UploadToContainerOptions,
12 },
13 exec::{CreateExecOptions, StartExecResults},
14 image::BuildImageOptions,
15 network::CreateNetworkOptions,
16 secret::{ContainerStateStatusEnum, HostConfig, Ipam, IpamConfig, PortBinding},
17 Docker,
18};
19use futures::StreamExt;
20use ipnetwork::Ipv4Network;
21use rand::Rng;
22use std::{
23 collections::HashMap,
24 net::Ipv4Addr,
25 path::{Path, PathBuf},
26 time::Duration,
27};
28
29#[derive(Debug, Clone)]
31pub struct DockerNatConfig {
32 pub topology: NatTopology,
34 pub public_subnet: Ipv4Network,
36 pub private_subnet_base: Ipv4Addr,
38 pub cleanup_on_drop: bool,
40 pub name_prefix: String,
42 pub network_emulation: Option<NetworkEmulation>,
45}
46
47#[derive(Debug, Clone)]
66pub struct NetworkEmulation {
67 pub delay_ms: u32,
69 pub jitter_ms: u32,
71 pub loss_percent: f64,
73 pub loss_correlation: f64,
75}
76
77impl Default for NetworkEmulation {
78 fn default() -> Self {
79 Self {
80 delay_ms: 0,
81 jitter_ms: 0,
82 loss_percent: 0.0,
83 loss_correlation: 25.0, }
85 }
86}
87
88impl NetworkEmulation {
89 pub fn lan() -> Self {
91 Self {
92 delay_ms: 1,
93 jitter_ms: 1,
94 loss_percent: 0.0,
95 ..Default::default()
96 }
97 }
98
99 pub fn regional() -> Self {
101 Self {
102 delay_ms: 40,
103 jitter_ms: 10,
104 loss_percent: 0.1,
105 ..Default::default()
106 }
107 }
108
109 pub fn intercontinental() -> Self {
111 Self {
112 delay_ms: 125,
113 jitter_ms: 25,
114 loss_percent: 0.5,
115 ..Default::default()
116 }
117 }
118
119 pub fn high_latency() -> Self {
122 Self {
123 delay_ms: 200,
124 jitter_ms: 30,
125 loss_percent: 1.0,
126 ..Default::default()
127 }
128 }
129
130 pub fn challenging() -> Self {
132 Self {
133 delay_ms: 150,
134 jitter_ms: 50,
135 loss_percent: 3.0,
136 loss_correlation: 50.0,
137 }
138 }
139}
140
141impl Default for DockerNatConfig {
142 fn default() -> Self {
143 let timestamp = chrono::Utc::now().format("%Y%m%d-%H%M%S").to_string();
145 let random_id = rand::thread_rng().gen::<u16>();
146 let name_prefix = format!("freenet-nat-{}-{}", timestamp, random_id);
147
148 let second_octet = rand::thread_rng().gen_range(16..=31);
154 let public_subnet = format!("172.{}.0.0/16", second_octet).parse().unwrap();
155
156 let private_first_octet = rand::thread_rng().gen_range(1..=250);
159
160 let network_emulation = if let Ok(emulation) =
162 std::env::var("FREENET_TEST_NETWORK_EMULATION")
163 {
164 match emulation.to_lowercase().as_str() {
165 "lan" => Some(NetworkEmulation::lan()),
166 "regional" => Some(NetworkEmulation::regional()),
167 "intercontinental" => Some(NetworkEmulation::intercontinental()),
168 "high_latency" => Some(NetworkEmulation::high_latency()),
169 "challenging" => Some(NetworkEmulation::challenging()),
170 other => {
171 tracing::warn!(
172 "Unknown FREENET_TEST_NETWORK_EMULATION value '{}', ignoring. \
173 Valid options: lan, regional, intercontinental, high_latency, challenging",
174 other
175 );
176 None
177 }
178 }
179 } else {
180 None
181 };
182
183 Self {
184 topology: NatTopology::OnePerNat,
185 public_subnet,
186 private_subnet_base: Ipv4Addr::new(10, private_first_octet, 0, 0),
187 cleanup_on_drop: true,
188 name_prefix,
189 network_emulation,
190 }
191 }
192}
193
194#[derive(Debug, Clone)]
196pub enum NatTopology {
197 OnePerNat,
199 Custom(Vec<NatNetwork>),
201}
202
203#[derive(Debug, Clone)]
205pub struct NatNetwork {
206 pub name: String,
207 pub peer_indices: Vec<usize>,
208 pub nat_type: NatType,
209}
210
211#[derive(Debug, Clone, Default)]
213pub enum NatType {
214 #[default]
216 RestrictedCone,
217 FullCone { forwarded_ports: Option<Vec<u16>> },
219}
220
221pub struct DockerNatBackend {
223 docker: Docker,
224 config: DockerNatConfig,
225 networks: Vec<String>,
227 containers: Vec<String>,
229 peer_containers: HashMap<usize, DockerPeerInfo>,
231 public_network_id: Option<String>,
233}
234
235#[derive(Debug, Clone)]
237pub struct DockerPeerInfo {
238 pub container_id: String,
239 pub container_name: String,
240 pub private_ip: Ipv4Addr,
242 pub public_ip: Ipv4Addr,
244 pub host_ws_port: u16,
246 pub network_port: u16,
248 pub is_gateway: bool,
250 pub nat_router_id: Option<String>,
252}
253
254pub struct DockerProcess {
256 docker: Docker,
257 container_id: String,
258 container_name: String,
259 local_log_cache: PathBuf,
260}
261
262impl PeerProcess for DockerProcess {
263 fn is_running(&self) -> bool {
264 let docker = self.docker.clone();
266 let id = self.container_id.clone();
267
268 tokio::task::block_in_place(|| {
269 tokio::runtime::Handle::current().block_on(async {
270 match docker.inspect_container(&id, None).await {
271 Ok(info) => info
272 .state
273 .and_then(|s| s.status)
274 .map(|s| s == ContainerStateStatusEnum::RUNNING)
275 .unwrap_or(false),
276 Err(_) => false,
277 }
278 })
279 })
280 }
281
282 fn kill(&mut self) -> Result<()> {
283 let docker = self.docker.clone();
284 let id = self.container_id.clone();
285
286 tokio::task::block_in_place(|| {
287 tokio::runtime::Handle::current().block_on(async {
288 let _ = docker
290 .stop_container(&id, Some(StopContainerOptions { t: 5 }))
291 .await;
292 Ok(())
293 })
294 })
295 }
296
297 fn log_path(&self) -> PathBuf {
298 self.local_log_cache.clone()
299 }
300
301 fn read_logs(&self) -> Result<Vec<LogEntry>> {
302 let docker = self.docker.clone();
303 let id = self.container_id.clone();
304 let cache_path = self.local_log_cache.clone();
305
306 tokio::task::block_in_place(|| {
307 tokio::runtime::Handle::current().block_on(async {
308 let options = LogsOptions::<String> {
310 stdout: true,
311 stderr: true,
312 timestamps: true,
313 ..Default::default()
314 };
315
316 let mut logs = docker.logs(&id, Some(options));
317 let mut log_content = String::new();
318
319 while let Some(log_result) = logs.next().await {
320 match log_result {
321 Ok(LogOutput::StdOut { message }) | Ok(LogOutput::StdErr { message }) => {
322 log_content.push_str(&String::from_utf8_lossy(&message));
323 }
324 _ => {}
325 }
326 }
327
328 if let Some(parent) = cache_path.parent() {
330 std::fs::create_dir_all(parent)?;
331 }
332 std::fs::write(&cache_path, &log_content)?;
333
334 crate::logs::read_log_file(&cache_path)
336 })
337 })
338 }
339}
340
341impl Drop for DockerProcess {
342 fn drop(&mut self) {
343 let _ = self.kill();
344 }
345}
346
347impl DockerNatBackend {
348 pub async fn new(config: DockerNatConfig) -> Result<Self> {
350 let docker = Docker::connect_with_local_defaults()
351 .map_err(|e| Error::Other(anyhow::anyhow!("Failed to connect to Docker: {}", e)))?;
352
353 docker
355 .ping()
356 .await
357 .map_err(|e| Error::Other(anyhow::anyhow!("Docker ping failed: {}", e)))?;
358
359 Self::cleanup_stale_resources(&docker, Duration::from_secs(10)).await?;
365
366 Ok(Self {
367 docker,
368 config,
369 networks: Vec::new(),
370 containers: Vec::new(),
371 peer_containers: HashMap::new(),
372 public_network_id: None,
373 })
374 }
375
376 async fn cleanup_stale_resources(docker: &Docker, max_age: Duration) -> Result<()> {
382 use bollard::container::ListContainersOptions;
383 use bollard::network::ListNetworksOptions;
384
385 let now = std::time::SystemTime::now();
386 let now_secs = now.duration_since(std::time::UNIX_EPOCH).unwrap().as_secs() as i64;
387 let cutoff = if max_age.is_zero() {
389 i64::MAX } else {
391 now_secs - max_age.as_secs() as i64
392 };
393
394 if max_age.is_zero() {
395 tracing::debug!("Cleaning up ALL freenet-nat resources");
396 } else {
397 tracing::debug!(
398 "Cleaning up freenet-nat resources older than {} seconds",
399 max_age.as_secs()
400 );
401 }
402
403 let mut filters = HashMap::new();
405 filters.insert("name".to_string(), vec!["freenet-nat-".to_string()]);
406
407 let options = ListContainersOptions {
408 all: true,
409 filters,
410 ..Default::default()
411 };
412
413 match docker.list_containers(Some(options)).await {
414 Ok(containers) => {
415 let mut removed_count = 0;
416 for container in containers {
417 if let Some(name) = container.names.and_then(|n| n.first().cloned()) {
419 if let Some(created) = container.created {
420 if created < cutoff {
421 if let Some(id) = container.id {
422 tracing::info!(
423 "Removing stale container: {} (age: {}s)",
424 name,
425 now.duration_since(std::time::UNIX_EPOCH).unwrap().as_secs()
426 as i64
427 - created
428 );
429 let _ = docker
430 .stop_container(&id, Some(StopContainerOptions { t: 2 }))
431 .await;
432 let _ = docker
433 .remove_container(
434 &id,
435 Some(RemoveContainerOptions {
436 force: true,
437 ..Default::default()
438 }),
439 )
440 .await;
441 removed_count += 1;
442 }
443 }
444 }
445 }
446 }
447 if removed_count > 0 {
448 tracing::info!("Removed {} stale container(s)", removed_count);
449 }
450 }
451 Err(e) => {
452 tracing::warn!("Failed to list containers for cleanup: {}", e);
453 }
454 }
455
456 let mut filters = HashMap::new();
458 filters.insert("name".to_string(), vec!["freenet-nat-".to_string()]);
459
460 let options = ListNetworksOptions { filters };
461
462 match docker.list_networks(Some(options)).await {
463 Ok(networks) => {
464 let mut removed_count = 0;
465 for network in networks {
466 if let Some(name) = &network.name {
467 if name.starts_with("freenet-nat-") {
468 if let Some(timestamp_str) = name.strip_prefix("freenet-nat-") {
470 let parts: Vec<&str> = timestamp_str.split('-').collect();
472 if parts.len() >= 2 {
473 let date_time = format!("{}-{}", parts[0], parts[1]);
474 if let Ok(created_time) = chrono::NaiveDateTime::parse_from_str(
475 &date_time,
476 "%Y%m%d-%H%M%S",
477 ) {
478 let created_timestamp = created_time.and_utc().timestamp();
479 if created_timestamp < cutoff {
480 if let Some(id) = &network.id {
481 tracing::info!(
482 "Removing stale network: {} (age: {}s)",
483 name,
484 now.duration_since(std::time::UNIX_EPOCH)
485 .unwrap()
486 .as_secs()
487 as i64
488 - created_timestamp
489 );
490 let _ = docker.remove_network(id).await;
491 removed_count += 1;
492 }
493 }
494 }
495 }
496 }
497 }
498 }
499 }
500 if removed_count > 0 {
501 tracing::info!("Removed {} stale network(s)", removed_count);
502 }
503 }
504 Err(e) => {
505 tracing::warn!("Failed to list networks for cleanup: {}", e);
506 }
507 }
508
509 Ok(())
510 }
511
512 pub async fn create_public_network(&mut self) -> Result<String> {
517 const MAX_SUBNET_RETRIES: usize = 10;
518
519 for attempt in 0..MAX_SUBNET_RETRIES {
520 let network_name = format!("{}-public", self.config.name_prefix);
521
522 let options = CreateNetworkOptions {
523 name: network_name.clone(),
524 driver: "bridge".to_string(),
525 ipam: Ipam {
526 config: Some(vec![IpamConfig {
527 subnet: Some(self.config.public_subnet.to_string()),
528 ..Default::default()
529 }]),
530 ..Default::default()
531 },
532 ..Default::default()
533 };
534
535 match self.docker.create_network(options).await {
536 Ok(response) => {
537 let network_id = response.id;
538 self.networks.push(network_id.clone());
539 self.public_network_id = Some(network_id.clone());
540 tracing::info!(
541 "Created public network: {} ({}) with subnet {}",
542 network_name,
543 network_id,
544 self.config.public_subnet
545 );
546 return Ok(network_id);
547 }
548 Err(e) => {
549 let error_msg = e.to_string();
550 if error_msg.contains("Pool overlaps") {
551 let old_subnet = self.config.public_subnet;
553 let new_second_octet = rand::thread_rng().gen_range(16..=31);
554 self.config.public_subnet =
555 format!("172.{}.0.0/16", new_second_octet).parse().unwrap();
556 tracing::warn!(
557 "Subnet {} conflicts with existing network, retrying with {} (attempt {}/{})",
558 old_subnet,
559 self.config.public_subnet,
560 attempt + 1,
561 MAX_SUBNET_RETRIES
562 );
563 continue;
564 }
565 return Err(Error::Other(anyhow::anyhow!(
566 "Failed to create public network: {}",
567 e
568 )));
569 }
570 }
571 }
572
573 Err(Error::Other(anyhow::anyhow!(
574 "Failed to create public network after {} attempts due to subnet conflicts. \
575 This may indicate stale Docker networks. Try running: \
576 docker network ls | grep freenet-nat | awk '{{print $1}}' | xargs -r docker network rm",
577 MAX_SUBNET_RETRIES
578 )))
579 }
580
581 pub async fn create_nat_network(
583 &mut self,
584 peer_index: usize,
585 ) -> Result<(String, String, Ipv4Addr)> {
586 let network_name = format!("{}-nat-{}", self.config.name_prefix, peer_index);
589 let base = self.config.private_subnet_base.octets();
590 let subnet = Ipv4Network::new(
591 Ipv4Addr::new(base[0], base[1].wrapping_add(peer_index as u8), 0, 0),
592 24,
593 )
594 .map_err(|e| Error::Other(anyhow::anyhow!("Invalid subnet: {}", e)))?;
595
596 let options = CreateNetworkOptions {
597 name: network_name.clone(),
598 driver: "bridge".to_string(),
599 internal: true, ipam: Ipam {
601 config: Some(vec![IpamConfig {
602 subnet: Some(subnet.to_string()),
603 ..Default::default()
604 }]),
605 ..Default::default()
606 },
607 ..Default::default()
608 };
609
610 let response =
611 self.docker.create_network(options).await.map_err(|e| {
612 Error::Other(anyhow::anyhow!("Failed to create NAT network: {}", e))
613 })?;
614
615 let network_id = response.id;
616 self.networks.push(network_id.clone());
617
618 let router_name = format!("{}-router-{}", self.config.name_prefix, peer_index);
620 let public_network_id = self
621 .public_network_id
622 .as_ref()
623 .ok_or_else(|| Error::Other(anyhow::anyhow!("Public network not created yet")))?;
624
625 let router_public_ip = Ipv4Addr::new(
630 self.config.public_subnet.ip().octets()[0],
631 self.config.public_subnet.ip().octets()[1],
632 peer_index as u8, 100, );
635 let router_private_ip =
637 Ipv4Addr::new(base[0], base[1].wrapping_add(peer_index as u8), 0, 254);
638
639 let public_octets = self.config.public_subnet.ip().octets();
643 let public_pattern = format!("172\\.{}\\.", public_octets[1]);
644 let private_pattern = format!(" {}\\.", base[0]);
645 let peer_private_ip = Ipv4Addr::new(base[0], base[1].wrapping_add(peer_index as u8), 0, 2);
647
648 let dnat_rules = if std::env::var("FREENET_TEST_FULL_CONE_NAT").is_ok() {
664 format!(
667 "iptables -t nat -A PREROUTING -i $PUBLIC_IF -p udp --dport 31337 -j DNAT --to-destination {}:31337 && \
668 echo 'Full Cone NAT: DNAT rule added for port 31337 -> {}:31337' && ",
669 peer_private_ip, peer_private_ip
670 )
671 } else if std::env::var("FREENET_TEST_SYMMETRIC_NAT").is_ok() {
672 format!(
675 "iptables -t nat -A POSTROUTING -o $PUBLIC_IF -p udp -j MASQUERADE --random && \
676 echo 'Symmetric NAT: Random port mapping enabled (hole punching will fail)' && "
677 )
678 } else {
679 format!(
695 "echo 'Port-Restricted Cone NAT: EIM + port-restricted filtering' && \
696 iptables -t nat -A PREROUTING -i $PUBLIC_IF -p udp --dport 31337 -j DNAT --to-destination {}:31337 && \
697 iptables -t nat -A POSTROUTING -o $PUBLIC_IF -p udp --sport 31337 -j SNAT --to-source $PUBLIC_IP:31337 && ",
698 peer_private_ip
699 )
700 };
701
702 let router_config = Config {
703 image: Some("alpine:latest".to_string()),
704 hostname: Some(router_name.clone()),
705 cmd: Some(vec![
706 "sh".to_string(),
707 "-c".to_string(),
708 format!(
713 "apk add --no-cache iptables iproute2 > /dev/null 2>&1 && \
714 PUBLIC_IF=$(ip -o addr show | grep '{}' | awk '{{print $2}}') && \
715 PRIVATE_IF=$(ip -o addr show | grep '{}' | awk '{{print $2}}') && \
716 PUBLIC_IP=$(ip -o addr show dev $PUBLIC_IF | awk '/inet / {{split($4,a,\"/\"); print a[1]}}') && \
717 echo \"Public interface: $PUBLIC_IF ($PUBLIC_IP), Private interface: $PRIVATE_IF\" && \
718 {}iptables -t nat -A POSTROUTING -o $PUBLIC_IF -j MASQUERADE && \
719 iptables -A FORWARD -i $PRIVATE_IF -o $PUBLIC_IF -j ACCEPT && \
720 iptables -A FORWARD -i $PUBLIC_IF -o $PRIVATE_IF -j ACCEPT && \
721 echo 'NAT router ready' && \
722 tail -f /dev/null",
723 public_pattern, private_pattern, dnat_rules
724 ),
725 ]),
726 host_config: Some(HostConfig {
727 cap_add: Some(vec!["NET_ADMIN".to_string()]),
728 sysctls: Some(HashMap::from([
729 ("net.ipv4.ip_forward".to_string(), "1".to_string()),
730 ])),
731 ..Default::default()
732 }),
733 ..Default::default()
734 };
735
736 let router_id = self
737 .docker
738 .create_container(
739 Some(CreateContainerOptions {
740 name: router_name.clone(),
741 ..Default::default()
742 }),
743 router_config,
744 )
745 .await
746 .map_err(|e| Error::Other(anyhow::anyhow!("Failed to create NAT router: {}", e)))?
747 .id;
748
749 self.containers.push(router_id.clone());
750
751 let _ = self
753 .docker
754 .disconnect_network(
755 "bridge",
756 bollard::network::DisconnectNetworkOptions {
757 container: router_id.clone(),
758 force: true,
759 },
760 )
761 .await;
762
763 self.docker
765 .connect_network(
766 public_network_id,
767 bollard::network::ConnectNetworkOptions {
768 container: router_id.clone(),
769 endpoint_config: bollard::secret::EndpointSettings {
770 ipam_config: Some(bollard::secret::EndpointIpamConfig {
771 ipv4_address: Some(router_public_ip.to_string()),
772 ..Default::default()
773 }),
774 ..Default::default()
775 },
776 },
777 )
778 .await
779 .map_err(|e| {
780 Error::Other(anyhow::anyhow!(
781 "Failed to connect router to public network: {}",
782 e
783 ))
784 })?;
785
786 self.docker
788 .connect_network(
789 &network_id,
790 bollard::network::ConnectNetworkOptions {
791 container: router_id.clone(),
792 endpoint_config: bollard::secret::EndpointSettings {
793 ipam_config: Some(bollard::secret::EndpointIpamConfig {
794 ipv4_address: Some(router_private_ip.to_string()),
795 ..Default::default()
796 }),
797 ..Default::default()
798 },
799 },
800 )
801 .await
802 .map_err(|e| {
803 Error::Other(anyhow::anyhow!(
804 "Failed to connect router to private network: {}",
805 e
806 ))
807 })?;
808
809 self.docker
811 .start_container(&router_id, None::<StartContainerOptions<String>>)
812 .await
813 .map_err(|e| Error::Other(anyhow::anyhow!("Failed to start NAT router: {}", e)))?;
814
815 tokio::time::sleep(Duration::from_secs(2)).await;
817
818 tracing::info!(
819 "Created NAT network {} with router {} (public: {}, private: {})",
820 network_name,
821 router_name,
822 router_public_ip,
823 router_private_ip
824 );
825
826 Ok((network_id, router_id, router_public_ip))
827 }
828
829 pub async fn ensure_base_image(&self) -> Result<String> {
831 let image_name = "freenet-test-peer:latest";
832
833 if self.docker.inspect_image(image_name).await.is_ok() {
835 tracing::debug!("Base image {} already exists", image_name);
836 return Ok(image_name.to_string());
837 }
838
839 tracing::info!("Building base image {}...", image_name);
840
841 let dockerfile = r#"
843FROM ubuntu:24.04
844RUN apt-get update && \
845 apt-get install -y --no-install-recommends \
846 libssl3 \
847 ca-certificates \
848 iproute2 \
849 && rm -rf /var/lib/apt/lists/*
850RUN mkdir -p /data /config
851WORKDIR /app
852"#;
853
854 let mut tar_builder = tar::Builder::new(Vec::new());
856 let mut header = tar::Header::new_gnu();
857 header.set_path("Dockerfile")?;
858 header.set_size(dockerfile.len() as u64);
859 header.set_mode(0o644);
860 header.set_cksum();
861 tar_builder.append(&header, dockerfile.as_bytes())?;
862 let tar_data = tar_builder.into_inner()?;
863
864 let options = BuildImageOptions {
866 dockerfile: "Dockerfile",
867 t: image_name,
868 rm: true,
869 ..Default::default()
870 };
871
872 let mut build_stream = self
873 .docker
874 .build_image(options, None, Some(tar_data.into()));
875
876 while let Some(result) = build_stream.next().await {
877 match result {
878 Ok(info) => {
879 if let Some(stream) = info.stream {
880 tracing::debug!("Build: {}", stream.trim());
881 }
882 if let Some(error) = info.error {
883 return Err(Error::Other(anyhow::anyhow!(
884 "Image build error: {}",
885 error
886 )));
887 }
888 }
889 Err(e) => {
890 return Err(Error::Other(anyhow::anyhow!("Image build failed: {}", e)));
891 }
892 }
893 }
894
895 tracing::info!("Built base image {}", image_name);
896 Ok(image_name.to_string())
897 }
898
899 pub async fn copy_binary_to_container(
901 &self,
902 container_id: &str,
903 binary_path: &Path,
904 ) -> Result<()> {
905 let binary_data = std::fs::read(binary_path)?;
907
908 let mut tar_builder = tar::Builder::new(Vec::new());
910 let mut header = tar::Header::new_gnu();
911 header.set_path("freenet")?;
912 header.set_size(binary_data.len() as u64);
913 header.set_mode(0o755);
914 header.set_cksum();
915 tar_builder.append(&header, binary_data.as_slice())?;
916 let tar_data = tar_builder.into_inner()?;
917
918 self.docker
920 .upload_to_container(
921 container_id,
922 Some(UploadToContainerOptions {
923 path: "/app",
924 ..Default::default()
925 }),
926 tar_data.into(),
927 )
928 .await
929 .map_err(|e| Error::Other(anyhow::anyhow!("Failed to copy binary: {}", e)))?;
930
931 Ok(())
932 }
933
934 pub async fn create_gateway(
936 &mut self,
937 index: usize,
938 binary_path: &Path,
939 keypair_path: &Path,
940 public_key_path: &Path,
941 ws_port: u16,
942 network_port: u16,
943 run_root: &Path,
944 ) -> Result<(DockerPeerInfo, DockerProcess)> {
945 let container_name = format!("{}-gw-{}", self.config.name_prefix, index);
946 let image = self.ensure_base_image().await?;
947
948 let public_network_id = self
949 .public_network_id
950 .as_ref()
951 .ok_or_else(|| Error::Other(anyhow::anyhow!("Public network not created yet")))?;
952
953 let gateway_ip = Ipv4Addr::new(
955 self.config.public_subnet.ip().octets()[0],
956 self.config.public_subnet.ip().octets()[1],
957 0,
958 10 + index as u8,
959 );
960
961 let config = Config {
963 image: Some(image),
964 hostname: Some(container_name.clone()),
965 exposed_ports: Some(HashMap::from([(
966 format!("{}/tcp", ws_port),
967 HashMap::new(),
968 )])),
969 host_config: Some(HostConfig {
970 port_bindings: Some(HashMap::from([(
971 format!("{}/tcp", ws_port),
972 Some(vec![PortBinding {
973 host_ip: Some("0.0.0.0".to_string()),
974 host_port: None, }]),
976 )])),
977 cap_add: Some(vec!["NET_ADMIN".to_string()]),
978 ..Default::default()
979 }),
980 env: Some(vec![
981 format!(
982 "RUST_LOG={}",
983 std::env::var("RUST_LOG").unwrap_or_else(|_| "info".to_string())
984 ),
985 "RUST_BACKTRACE=1".to_string(),
986 "FREENET_TELEMETRY_ENABLED=false".to_string(),
993 ]),
994 cmd: Some(vec![
995 "/app/freenet".to_string(),
996 "network".to_string(),
997 "--data-dir".to_string(),
998 "/data".to_string(),
999 "--config-dir".to_string(),
1000 "/config".to_string(),
1001 "--ws-api-address".to_string(),
1002 "0.0.0.0".to_string(),
1003 "--ws-api-port".to_string(),
1004 ws_port.to_string(),
1005 "--network-address".to_string(),
1006 "0.0.0.0".to_string(),
1007 "--network-port".to_string(),
1008 network_port.to_string(),
1009 "--public-network-address".to_string(),
1010 gateway_ip.to_string(),
1011 "--public-network-port".to_string(),
1012 network_port.to_string(),
1013 "--is-gateway".to_string(),
1014 "--skip-load-from-network".to_string(),
1015 "--transport-keypair".to_string(),
1016 "/config/keypair.pem".to_string(),
1017 ]),
1018 ..Default::default()
1019 };
1020
1021 let container_id = self
1022 .docker
1023 .create_container(
1024 Some(CreateContainerOptions {
1025 name: container_name.clone(),
1026 ..Default::default()
1027 }),
1028 config,
1029 )
1030 .await
1031 .map_err(|e| {
1032 Error::Other(anyhow::anyhow!("Failed to create gateway container: {}", e))
1033 })?
1034 .id;
1035
1036 self.containers.push(container_id.clone());
1037
1038 self.docker
1040 .connect_network(
1041 public_network_id,
1042 bollard::network::ConnectNetworkOptions {
1043 container: container_id.clone(),
1044 endpoint_config: bollard::secret::EndpointSettings {
1045 ipam_config: Some(bollard::secret::EndpointIpamConfig {
1046 ipv4_address: Some(gateway_ip.to_string()),
1047 ..Default::default()
1048 }),
1049 ..Default::default()
1050 },
1051 },
1052 )
1053 .await
1054 .map_err(|e| {
1055 Error::Other(anyhow::anyhow!(
1056 "Failed to connect gateway to network: {}",
1057 e
1058 ))
1059 })?;
1060
1061 self.copy_binary_to_container(&container_id, binary_path)
1063 .await?;
1064 self.copy_file_to_container(&container_id, keypair_path, "/config/keypair.pem")
1065 .await?;
1066 self.copy_file_to_container(&container_id, public_key_path, "/config/public_key.pem")
1067 .await?;
1068
1069 self.docker
1071 .start_container(&container_id, None::<StartContainerOptions<String>>)
1072 .await
1073 .map_err(|e| Error::Other(anyhow::anyhow!("Failed to start gateway: {}", e)))?;
1074
1075 self.apply_network_emulation(&container_id, &container_name)
1077 .await?;
1078
1079 let host_ws_port = self.get_container_host_port(&container_id, ws_port).await?;
1081
1082 let info = DockerPeerInfo {
1083 container_id: container_id.clone(),
1084 container_name: container_name.clone(),
1085 private_ip: gateway_ip, public_ip: gateway_ip,
1087 host_ws_port,
1088 network_port,
1089 is_gateway: true,
1090 nat_router_id: None,
1091 };
1092
1093 self.peer_containers.insert(index, info.clone());
1094
1095 let local_log_cache = run_root.join(format!("gw{}", index)).join("peer.log");
1096
1097 tracing::info!(
1098 "Created gateway {} at {} (ws: localhost:{})",
1099 container_name,
1100 gateway_ip,
1101 host_ws_port
1102 );
1103
1104 Ok((
1105 info,
1106 DockerProcess {
1107 docker: self.docker.clone(),
1108 container_id,
1109 container_name,
1110 local_log_cache,
1111 },
1112 ))
1113 }
1114
1115 pub async fn create_peer(
1117 &mut self,
1118 index: usize,
1119 binary_path: &Path,
1120 keypair_path: &Path,
1121 public_key_path: &Path,
1122 gateways_toml_path: &Path,
1123 gateway_public_key_path: Option<&Path>,
1124 ws_port: u16,
1125 network_port: u16,
1126 run_root: &Path,
1127 ) -> Result<(DockerPeerInfo, DockerProcess)> {
1128 let container_name = format!("{}-peer-{}", self.config.name_prefix, index);
1129 let image = self.ensure_base_image().await?;
1130
1131 let (nat_network_id, router_id, router_public_ip) = self.create_nat_network(index).await?;
1133
1134 let base = self.config.private_subnet_base.octets();
1136 let private_ip = Ipv4Addr::new(base[0], base[1].wrapping_add(index as u8), 0, 2);
1137
1138 let config = Config {
1140 image: Some(image),
1141 hostname: Some(container_name.clone()),
1142 exposed_ports: Some(HashMap::from([(
1143 format!("{}/tcp", ws_port),
1144 HashMap::new(),
1145 )])),
1146 host_config: Some(HostConfig {
1147 port_bindings: Some(HashMap::from([(
1148 format!("{}/tcp", ws_port),
1149 Some(vec![PortBinding {
1150 host_ip: Some("0.0.0.0".to_string()),
1151 host_port: None, }]),
1153 )])),
1154 cap_add: Some(vec!["NET_ADMIN".to_string()]),
1155 ..Default::default()
1156 }),
1157 env: Some(vec![
1158 format!(
1159 "RUST_LOG={}",
1160 std::env::var("RUST_LOG").unwrap_or_else(|_| "info".to_string())
1161 ),
1162 "RUST_BACKTRACE=1".to_string(),
1163 "FREENET_TELEMETRY_ENABLED=false".to_string(),
1170 ]),
1171 cmd: Some(vec![
1172 "/app/freenet".to_string(),
1173 "network".to_string(),
1174 "--data-dir".to_string(),
1175 "/data".to_string(),
1176 "--config-dir".to_string(),
1177 "/config".to_string(),
1178 "--ws-api-address".to_string(),
1179 "0.0.0.0".to_string(),
1180 "--ws-api-port".to_string(),
1181 ws_port.to_string(),
1182 "--network-address".to_string(),
1183 "0.0.0.0".to_string(),
1184 "--network-port".to_string(),
1185 network_port.to_string(),
1186 "--skip-load-from-network".to_string(),
1188 "--transport-keypair".to_string(),
1189 "/config/keypair.pem".to_string(),
1190 ]),
1191 ..Default::default()
1192 };
1193
1194 let container_id = self
1195 .docker
1196 .create_container(
1197 Some(CreateContainerOptions {
1198 name: container_name.clone(),
1199 ..Default::default()
1200 }),
1201 config,
1202 )
1203 .await
1204 .map_err(|e| Error::Other(anyhow::anyhow!("Failed to create peer container: {}", e)))?
1205 .id;
1206
1207 self.containers.push(container_id.clone());
1208
1209 self.docker
1212 .connect_network(
1213 &nat_network_id,
1214 bollard::network::ConnectNetworkOptions {
1215 container: container_id.clone(),
1216 endpoint_config: bollard::secret::EndpointSettings {
1217 ipam_config: Some(bollard::secret::EndpointIpamConfig {
1218 ipv4_address: Some(private_ip.to_string()),
1219 ..Default::default()
1220 }),
1221 gateway: Some(
1222 Ipv4Addr::new(base[0], base[1].wrapping_add(index as u8), 0, 1)
1223 .to_string(),
1224 ),
1225 ..Default::default()
1226 },
1227 },
1228 )
1229 .await
1230 .map_err(|e| {
1231 Error::Other(anyhow::anyhow!(
1232 "Failed to connect peer to NAT network: {}",
1233 e
1234 ))
1235 })?;
1236
1237 self.copy_binary_to_container(&container_id, binary_path)
1239 .await?;
1240 self.copy_file_to_container(&container_id, keypair_path, "/config/keypair.pem")
1241 .await?;
1242 self.copy_file_to_container(&container_id, public_key_path, "/config/public_key.pem")
1243 .await?;
1244 self.copy_file_to_container(&container_id, gateways_toml_path, "/config/gateways.toml")
1245 .await?;
1246
1247 if let Some(gw_pubkey_path) = gateway_public_key_path {
1249 self.copy_file_to_container(&container_id, gw_pubkey_path, "/config/gw_public_key.pem")
1250 .await?;
1251 }
1252
1253 self.docker
1255 .start_container(&container_id, None::<StartContainerOptions<String>>)
1256 .await
1257 .map_err(|e| Error::Other(anyhow::anyhow!("Failed to start peer: {}", e)))?;
1258
1259 self.apply_network_emulation(&container_id, &container_name)
1261 .await?;
1262
1263 let host_ws_port = self.get_container_host_port(&container_id, ws_port).await?;
1265
1266 let router_gateway = Ipv4Addr::new(base[0], base[1].wrapping_add(index as u8), 0, 254);
1269 let public_subnet = self.config.public_subnet;
1270 self.exec_in_container(
1271 &container_id,
1272 &[
1273 "sh",
1274 "-c",
1275 &format!("ip route add {} via {}", public_subnet, router_gateway),
1276 ],
1277 )
1278 .await?;
1279
1280 let info = DockerPeerInfo {
1281 container_id: container_id.clone(),
1282 container_name: container_name.clone(),
1283 private_ip,
1284 public_ip: router_public_ip,
1285 host_ws_port,
1286 network_port,
1287 is_gateway: false,
1288 nat_router_id: Some(router_id),
1289 };
1290
1291 self.peer_containers.insert(index, info.clone());
1292
1293 let local_log_cache = run_root.join(format!("peer{}", index)).join("peer.log");
1294
1295 tracing::info!(
1296 "Created peer {} at {} behind NAT {} (ws: localhost:{})",
1297 container_name,
1298 private_ip,
1299 router_public_ip,
1300 host_ws_port
1301 );
1302
1303 Ok((
1304 info,
1305 DockerProcess {
1306 docker: self.docker.clone(),
1307 container_id,
1308 container_name,
1309 local_log_cache,
1310 },
1311 ))
1312 }
1313
1314 pub async fn copy_file_to_container_pub(
1316 &self,
1317 container_id: &str,
1318 local_path: &Path,
1319 container_path: &str,
1320 ) -> Result<()> {
1321 self.copy_file_to_container(container_id, local_path, container_path)
1322 .await
1323 }
1324
1325 async fn copy_file_to_container(
1327 &self,
1328 container_id: &str,
1329 local_path: &Path,
1330 container_path: &str,
1331 ) -> Result<()> {
1332 let file_data = std::fs::read(local_path)?;
1333 let file_name = Path::new(container_path)
1334 .file_name()
1335 .ok_or_else(|| Error::Other(anyhow::anyhow!("Invalid container path")))?
1336 .to_str()
1337 .ok_or_else(|| Error::Other(anyhow::anyhow!("Invalid file name")))?;
1338
1339 let dir_path = Path::new(container_path)
1340 .parent()
1341 .ok_or_else(|| Error::Other(anyhow::anyhow!("Invalid container path")))?
1342 .to_str()
1343 .ok_or_else(|| Error::Other(anyhow::anyhow!("Invalid directory path")))?;
1344
1345 let mut tar_builder = tar::Builder::new(Vec::new());
1347 let mut header = tar::Header::new_gnu();
1348 header.set_path(file_name)?;
1349 header.set_size(file_data.len() as u64);
1350 header.set_mode(0o644);
1351 header.set_cksum();
1352 tar_builder.append(&header, file_data.as_slice())?;
1353 let tar_data = tar_builder.into_inner()?;
1354
1355 self.docker
1356 .upload_to_container(
1357 container_id,
1358 Some(UploadToContainerOptions {
1359 path: dir_path,
1360 ..Default::default()
1361 }),
1362 tar_data.into(),
1363 )
1364 .await
1365 .map_err(|e| Error::Other(anyhow::anyhow!("Failed to copy file: {}", e)))?;
1366
1367 Ok(())
1368 }
1369
1370 async fn exec_in_container(&self, container_id: &str, cmd: &[&str]) -> Result<String> {
1372 let exec = self
1373 .docker
1374 .create_exec(
1375 container_id,
1376 CreateExecOptions {
1377 cmd: Some(cmd.iter().map(|s| s.to_string()).collect()),
1378 attach_stdout: Some(true),
1379 attach_stderr: Some(true),
1380 ..Default::default()
1381 },
1382 )
1383 .await
1384 .map_err(|e| Error::Other(anyhow::anyhow!("Failed to create exec: {}", e)))?;
1385
1386 let output = self
1387 .docker
1388 .start_exec(&exec.id, None)
1389 .await
1390 .map_err(|e| Error::Other(anyhow::anyhow!("Failed to start exec: {}", e)))?;
1391
1392 let mut result = String::new();
1393 if let StartExecResults::Attached { mut output, .. } = output {
1394 while let Some(Ok(msg)) = output.next().await {
1395 match msg {
1396 LogOutput::StdOut { message } | LogOutput::StdErr { message } => {
1397 result.push_str(&String::from_utf8_lossy(&message));
1398 }
1399 _ => {}
1400 }
1401 }
1402 }
1403
1404 Ok(result)
1405 }
1406
1407 async fn apply_network_emulation(
1412 &self,
1413 container_id: &str,
1414 container_name: &str,
1415 ) -> Result<()> {
1416 let Some(ref emulation) = self.config.network_emulation else {
1417 return Ok(());
1418 };
1419
1420 if emulation.delay_ms == 0 && emulation.loss_percent == 0.0 {
1422 return Ok(());
1423 }
1424
1425 let mut tc_args = vec!["tc", "qdisc", "add", "dev", "eth0", "root", "netem"];
1428
1429 let delay_str;
1430 let jitter_str;
1431 let loss_str;
1432 let correlation_str;
1433
1434 if emulation.delay_ms > 0 {
1436 delay_str = format!("{}ms", emulation.delay_ms);
1437 tc_args.push("delay");
1438 tc_args.push(&delay_str);
1439
1440 if emulation.jitter_ms > 0 {
1441 jitter_str = format!("{}ms", emulation.jitter_ms);
1442 tc_args.push(&jitter_str);
1443 }
1444 }
1445
1446 if emulation.loss_percent > 0.0 {
1448 loss_str = format!("{:.2}%", emulation.loss_percent);
1449 tc_args.push("loss");
1450 tc_args.push(&loss_str);
1451
1452 if emulation.loss_correlation > 0.0 {
1453 correlation_str = format!("{:.0}%", emulation.loss_correlation);
1454 tc_args.push(&correlation_str);
1455 }
1456 }
1457
1458 tracing::info!(
1459 "Applying network emulation to {}: delay={}ms±{}ms, loss={:.2}%",
1460 container_name,
1461 emulation.delay_ms,
1462 emulation.jitter_ms,
1463 emulation.loss_percent
1464 );
1465
1466 let output = self.exec_in_container(container_id, &tc_args).await?;
1467
1468 if !output.is_empty() && output.contains("Error") {
1469 tracing::warn!(
1470 "Network emulation may have failed for {}: {}",
1471 container_name,
1472 output.trim()
1473 );
1474 } else {
1475 tracing::debug!(
1476 "Network emulation applied to {}: {:?}",
1477 container_name,
1478 tc_args
1479 );
1480 }
1481
1482 Ok(())
1483 }
1484
1485 pub async fn cleanup(&mut self) -> Result<()> {
1487 tracing::info!("Cleaning up Docker NAT resources...");
1488
1489 for container_id in self.containers.drain(..) {
1491 let _ = self
1492 .docker
1493 .stop_container(&container_id, Some(StopContainerOptions { t: 2 }))
1494 .await;
1495 let _ = self
1496 .docker
1497 .remove_container(
1498 &container_id,
1499 Some(RemoveContainerOptions {
1500 force: true,
1501 ..Default::default()
1502 }),
1503 )
1504 .await;
1505 }
1506
1507 for network_id in self.networks.drain(..) {
1509 let _ = self.docker.remove_network(&network_id).await;
1510 }
1511
1512 self.peer_containers.clear();
1513 self.public_network_id = None;
1514
1515 Ok(())
1516 }
1517
1518 pub fn get_peer_info(&self, index: usize) -> Option<&DockerPeerInfo> {
1520 self.peer_containers.get(&index)
1521 }
1522
1523 pub async fn dump_iptables_counters(&self) -> Result<std::collections::HashMap<usize, String>> {
1529 let mut results = std::collections::HashMap::new();
1530
1531 for (&peer_index, peer_info) in &self.peer_containers {
1532 if let Some(router_id) = &peer_info.nat_router_id {
1533 let mut output = String::new();
1534
1535 output.push_str("=== NAT table ===\n");
1537 match self
1538 .exec_in_container(router_id, &["iptables", "-t", "nat", "-nvL"])
1539 .await
1540 {
1541 Ok(s) => output.push_str(&s),
1542 Err(e) => output.push_str(&format!("Error: {}\n", e)),
1543 }
1544
1545 output.push_str("\n=== FORWARD chain ===\n");
1547 match self
1548 .exec_in_container(router_id, &["iptables", "-nvL", "FORWARD"])
1549 .await
1550 {
1551 Ok(s) => output.push_str(&s),
1552 Err(e) => output.push_str(&format!("Error: {}\n", e)),
1553 }
1554
1555 results.insert(peer_index, output);
1556 }
1557 }
1558
1559 Ok(results)
1560 }
1561
1562 pub async fn dump_conntrack_table(&self) -> Result<std::collections::HashMap<usize, String>> {
1567 let mut results = std::collections::HashMap::new();
1568
1569 for (&peer_index, peer_info) in &self.peer_containers {
1570 if let Some(router_id) = &peer_info.nat_router_id {
1571 let _ = self
1573 .exec_in_container(router_id, &["apk", "add", "--no-cache", "conntrack-tools"])
1574 .await;
1575
1576 match self
1578 .exec_in_container(router_id, &["conntrack", "-L", "-p", "udp"])
1579 .await
1580 {
1581 Ok(s) if s.trim().is_empty() => {
1582 results.insert(peer_index, "(no UDP conntrack entries)".to_string());
1583 }
1584 Ok(s) => {
1585 results.insert(peer_index, s);
1586 }
1587 Err(e) => {
1588 results.insert(peer_index, format!("Error: {}", e));
1589 }
1590 }
1591 }
1592 }
1593
1594 Ok(results)
1595 }
1596
1597 pub async fn dump_peer_routes(&self) -> Result<std::collections::HashMap<usize, String>> {
1601 let mut results = std::collections::HashMap::new();
1602
1603 for (&peer_index, peer_info) in &self.peer_containers {
1604 if peer_info.nat_router_id.is_some() {
1605 match self
1607 .exec_in_container(&peer_info.container_id, &["ip", "route"])
1608 .await
1609 {
1610 Ok(s) => {
1611 results.insert(peer_index, s);
1612 }
1613 Err(e) => {
1614 results.insert(peer_index, format!("Error: {}", e));
1615 }
1616 }
1617 }
1618 }
1619
1620 Ok(results)
1621 }
1622
1623 async fn get_container_host_port(
1630 &self,
1631 container_id: &str,
1632 container_port: u16,
1633 ) -> Result<u16> {
1634 let info = self
1635 .docker
1636 .inspect_container(container_id, None)
1637 .await
1638 .map_err(|e| {
1639 Error::Other(anyhow::anyhow!(
1640 "Failed to inspect container for port allocation: {}",
1641 e
1642 ))
1643 })?;
1644
1645 let port_key = format!("{}/tcp", container_port);
1646
1647 let host_port = info
1648 .network_settings
1649 .and_then(|ns| ns.ports)
1650 .and_then(|ports| ports.get(&port_key).cloned())
1651 .flatten()
1652 .and_then(|bindings| bindings.first().cloned())
1653 .and_then(|binding| binding.host_port)
1654 .and_then(|port_str| port_str.parse::<u16>().ok())
1655 .ok_or_else(|| {
1656 Error::Other(anyhow::anyhow!(
1657 "Failed to get allocated host port for container {} port {}",
1658 container_id,
1659 container_port
1660 ))
1661 })?;
1662
1663 Ok(host_port)
1664 }
1665}
1666
1667impl Drop for DockerNatBackend {
1668 fn drop(&mut self) {
1669 if self.config.cleanup_on_drop {
1670 tracing::info!("Cleaning up Docker NAT backend resources...");
1671
1672 let docker = self.docker.clone();
1674 let containers = std::mem::take(&mut self.containers);
1675 let networks = std::mem::take(&mut self.networks);
1676
1677 let cleanup = async {
1681 let container_futures = containers.into_iter().map(|container_id| {
1683 let docker = docker.clone();
1684 async move {
1685 if let Err(e) = docker
1686 .stop_container(&container_id, Some(StopContainerOptions { t: 2 }))
1687 .await
1688 {
1689 tracing::debug!("Failed to stop container {}: {}", container_id, e);
1690 }
1691 if let Err(e) = docker
1692 .remove_container(
1693 &container_id,
1694 Some(RemoveContainerOptions {
1695 force: true,
1696 ..Default::default()
1697 }),
1698 )
1699 .await
1700 {
1701 tracing::debug!("Failed to remove container {}: {}", container_id, e);
1702 }
1703 }
1704 });
1705
1706 futures::future::join_all(container_futures).await;
1708
1709 for network_id in networks {
1711 if let Err(e) = docker.remove_network(&network_id).await {
1712 tracing::debug!("Failed to remove network {}: {}", network_id, e);
1713 }
1714 }
1715
1716 tracing::info!("Docker NAT backend cleanup complete");
1717 };
1718
1719 if let Ok(handle) = tokio::runtime::Handle::try_current() {
1722 tokio::task::block_in_place(|| {
1723 handle.block_on(cleanup);
1724 });
1725 } else if let Ok(rt) = tokio::runtime::Runtime::new() {
1726 rt.block_on(cleanup);
1727 } else {
1728 tracing::error!("Failed to create runtime for cleanup");
1729 }
1730 }
1731 }
1732}