use crate::{
binary::FreenetBinary,
docker::{DockerNatBackend, DockerNatConfig},
network::TestNetwork,
peer::{get_free_port, TestPeer},
process::{self, PeerProcess},
remote::{PeerLocation, RemoteMachine},
Error, Result,
};
use chrono::Utc;
use std::collections::HashMap;
use std::fs;
use std::net::Ipv4Addr;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::{Duration, SystemTime};
#[derive(Debug, Clone)]
pub enum Backend {
Local,
DockerNat(DockerNatConfig),
}
impl Default for Backend {
fn default() -> Self {
if std::env::var("FREENET_TEST_DOCKER_NAT").is_ok() {
let mut config = DockerNatConfig::default();
if let Ok(emulation) = std::env::var("FREENET_TEST_NETWORK_EMULATION") {
config.network_emulation = match emulation.to_lowercase().as_str() {
"lan" => Some(crate::docker::NetworkEmulation::lan()),
"regional" => Some(crate::docker::NetworkEmulation::regional()),
"intercontinental" => Some(crate::docker::NetworkEmulation::intercontinental()),
"high_latency" => Some(crate::docker::NetworkEmulation::high_latency()),
"challenging" => Some(crate::docker::NetworkEmulation::challenging()),
other => {
tracing::warn!(
"Unknown FREENET_TEST_NETWORK_EMULATION value '{}', ignoring. \
Valid options: lan, regional, intercontinental, high_latency, challenging",
other
);
None
}
};
}
Backend::DockerNat(config)
} else {
Backend::Local
}
}
}
struct GatewayInfo {
address: String,
public_key_path: PathBuf,
}
pub struct NetworkBuilder {
gateways: usize,
peers: usize,
binary: FreenetBinary,
min_connectivity: f64,
connectivity_timeout: Duration,
preserve_data_on_failure: bool,
preserve_data_on_success: bool,
peer_locations: HashMap<usize, PeerLocation>,
default_location: PeerLocation,
min_connections: Option<usize>,
max_connections: Option<usize>,
start_stagger: Duration,
backend: Backend,
}
impl Default for NetworkBuilder {
fn default() -> Self {
Self::new()
}
}
impl NetworkBuilder {
pub fn new() -> Self {
Self {
gateways: 1,
peers: 3,
binary: FreenetBinary::default(),
min_connectivity: 1.0, connectivity_timeout: Duration::from_secs(30),
preserve_data_on_failure: false,
preserve_data_on_success: false,
peer_locations: HashMap::new(),
default_location: PeerLocation::Local,
min_connections: None,
max_connections: None,
start_stagger: Duration::from_millis(500),
backend: Backend::default(),
}
}
pub fn gateways(mut self, n: usize) -> Self {
self.gateways = n;
self
}
pub fn peers(mut self, n: usize) -> Self {
self.peers = n;
self
}
pub fn binary(mut self, binary: FreenetBinary) -> Self {
self.binary = binary;
self
}
pub fn require_connectivity(mut self, ratio: f64) -> Self {
self.min_connectivity = ratio;
self
}
pub fn connectivity_timeout(mut self, timeout: Duration) -> Self {
self.connectivity_timeout = timeout;
self
}
pub fn min_connections(mut self, min: usize) -> Self {
self.min_connections = Some(min);
self
}
pub fn max_connections(mut self, max: usize) -> Self {
self.max_connections = Some(max);
self
}
pub fn start_stagger(mut self, delay: Duration) -> Self {
self.start_stagger = delay;
self
}
pub fn preserve_temp_dirs_on_failure(mut self, preserve: bool) -> Self {
self.preserve_data_on_failure = preserve;
self
}
pub fn preserve_temp_dirs_on_success(mut self, preserve: bool) -> Self {
self.preserve_data_on_success = preserve;
self
}
pub fn peer_location(mut self, index: usize, location: PeerLocation) -> Self {
self.peer_locations.insert(index, location);
self
}
pub fn default_location(mut self, location: PeerLocation) -> Self {
self.default_location = location;
self
}
pub fn distribute_across_remotes(mut self, machines: Vec<RemoteMachine>) -> Self {
let total_peers = self.gateways + self.peers;
for (idx, machine) in (0..total_peers).zip(machines.iter().cycle()) {
self.peer_locations
.insert(idx, PeerLocation::Remote(machine.clone()));
}
self
}
pub fn backend(mut self, backend: Backend) -> Self {
self.backend = backend;
self
}
pub async fn build(self) -> Result<TestNetwork> {
match self.backend.clone() {
Backend::Local => self.build_local().await,
Backend::DockerNat(config) => self.build_docker_nat(config).await,
}
}
async fn build_local(self) -> Result<TestNetwork> {
let binary_path = self.binary.resolve()?;
tracing::info!(
"Starting test network: {} gateways, {} peers",
self.gateways,
self.peers
);
let base_dir = resolve_base_dir();
fs::create_dir_all(&base_dir)?;
cleanup_old_runs(&base_dir, 5)?;
let run_root = create_run_directory(&base_dir)?;
let mut run_status = RunStatusGuard::new(&run_root);
let mut gateways = Vec::new();
for i in 0..self.gateways {
let peer = match self.start_peer(&binary_path, i, true, &run_root).await {
Ok(peer) => peer,
Err(err) => {
let detail = format!("failed to start gateway {i}: {err}");
run_status.mark("failure", Some(&detail));
return Err(err);
}
};
gateways.push(peer);
}
let gateway_info: Vec<_> = gateways
.iter()
.map(|gw| GatewayInfo {
address: format!("{}:{}", gw.network_address, gw.network_port),
public_key_path: gw
.public_key_path
.clone()
.expect("Gateway must have public key"),
})
.collect();
let mut peers = Vec::new();
for i in 0..self.peers {
let peer = match self
.start_peer_with_gateways(
&binary_path,
i + self.gateways,
false,
&gateway_info,
&run_root,
)
.await
{
Ok(peer) => peer,
Err(err) => {
let detail = format!("failed to start peer {}: {}", i + self.gateways, err);
run_status.mark("failure", Some(&detail));
return Err(err);
}
};
peers.push(peer);
if i + 1 < self.peers && !self.start_stagger.is_zero() {
tokio::time::sleep(self.start_stagger).await;
}
}
let network = TestNetwork::new(gateways, peers, self.min_connectivity, run_root.clone());
match network
.wait_until_ready_with_timeout(self.connectivity_timeout)
.await
{
Ok(()) => {
if self.preserve_data_on_success {
match preserve_network_state(&network) {
Ok(path) => {
println!("Network data directories preserved at {}", path.display());
}
Err(err) => {
eprintln!(
"Failed to preserve network data directories after success: {}",
err
);
}
}
}
let detail = format!("success: gateways={}, peers={}", self.gateways, self.peers);
run_status.mark("success", Some(&detail));
Ok(network)
}
Err(err) => {
if let Err(log_err) = dump_recent_logs(&network) {
eprintln!("Failed to dump logs after connectivity error: {}", log_err);
}
if self.preserve_data_on_failure {
match preserve_network_state(&network) {
Ok(path) => {
eprintln!("Network data directories preserved at {}", path.display());
}
Err(copy_err) => {
eprintln!("Failed to preserve network data directories: {}", copy_err);
}
}
}
let detail = err.to_string();
run_status.mark("failure", Some(&detail));
Err(err)
}
}
}
pub fn build_sync(self) -> Result<TestNetwork> {
tokio::runtime::Runtime::new()?.block_on(self.build())
}
async fn start_peer(
&self,
binary_path: &PathBuf,
index: usize,
is_gateway: bool,
run_root: &Path,
) -> Result<TestPeer> {
self.start_peer_with_gateways(binary_path, index, is_gateway, &[], run_root)
.await
}
async fn start_peer_with_gateways(
&self,
binary_path: &PathBuf,
index: usize,
is_gateway: bool,
gateway_info: &[GatewayInfo],
run_root: &Path,
) -> Result<TestPeer> {
let location = self
.peer_locations
.get(&index)
.cloned()
.unwrap_or_else(|| self.default_location.clone());
let id = if is_gateway {
format!("gw{}", index)
} else {
format!("peer{}", index)
};
let network_address = match &location {
PeerLocation::Local => {
let addr_index = index as u32;
let second_octet = ((addr_index / 256) % 254 + 1) as u8;
let third_octet = (addr_index % 256) as u8;
Ipv4Addr::new(127, second_octet, third_octet, 1).to_string()
}
PeerLocation::Remote(remote) => {
remote.discover_public_address()?
}
};
let (ws_port, network_port) = match &location {
PeerLocation::Local => (get_free_port()?, get_free_port()?),
PeerLocation::Remote(_) => (0, 0), };
let data_dir = create_peer_dir(run_root, &id)?;
tracing::debug!(
"Starting {} {} - ws:{} net:{}",
if is_gateway { "gateway" } else { "peer" },
id,
ws_port,
network_port
);
let keypair_path = data_dir.join("keypair.pem");
let public_key_path = data_dir.join("public_key.pem");
generate_keypair(&keypair_path, &public_key_path)?;
if let PeerLocation::Remote(remote) = &location {
let remote_data_dir = remote.remote_work_dir().join(&id);
let mkdir_cmd = format!("mkdir -p {}", remote_data_dir.display());
remote.exec(&mkdir_cmd)?;
let remote_keypair = remote_data_dir.join("keypair.pem");
let remote_pubkey = remote_data_dir.join("public_key.pem");
remote.scp_upload(&keypair_path, remote_keypair.to_str().unwrap())?;
remote.scp_upload(&public_key_path, remote_pubkey.to_str().unwrap())?;
if !is_gateway {
for gw in gateway_info {
let gw_pubkey_name = gw.public_key_path.file_name().ok_or_else(|| {
Error::PeerStartupFailed("Invalid gateway pubkey path".to_string())
})?;
let remote_gw_pubkey = remote_data_dir.join(gw_pubkey_name);
remote.scp_upload(&gw.public_key_path, remote_gw_pubkey.to_str().unwrap())?;
}
}
}
let mut args = vec![
"network".to_string(),
"--data-dir".to_string(),
match &location {
PeerLocation::Local => data_dir.to_string_lossy().to_string(),
PeerLocation::Remote(remote) => remote
.remote_work_dir()
.join(&id)
.to_string_lossy()
.to_string(),
},
"--config-dir".to_string(),
match &location {
PeerLocation::Local => data_dir.to_string_lossy().to_string(),
PeerLocation::Remote(remote) => remote
.remote_work_dir()
.join(&id)
.to_string_lossy()
.to_string(),
},
"--ws-api-port".to_string(),
ws_port.to_string(),
"--network-address".to_string(),
network_address.clone(),
"--network-port".to_string(),
network_port.to_string(),
"--public-network-address".to_string(),
network_address.clone(),
"--public-network-port".to_string(),
network_port.to_string(),
"--skip-load-from-network".to_string(),
];
if is_gateway {
args.push("--is-gateway".to_string());
}
args.push("--transport-keypair".to_string());
let keypair_arg = match &location {
PeerLocation::Local => data_dir.join("keypair.pem").to_string_lossy().to_string(),
PeerLocation::Remote(remote) => remote
.remote_work_dir()
.join(&id)
.join("keypair.pem")
.to_string_lossy()
.to_string(),
};
args.push(keypair_arg);
if !is_gateway && !gateway_info.is_empty() {
let gateways_toml = data_dir.join("gateways.toml");
let mut content = String::new();
for gw in gateway_info {
let gw_pubkey_path = match &location {
PeerLocation::Local => gw.public_key_path.clone(),
PeerLocation::Remote(remote) => {
let gw_pubkey_name = gw.public_key_path.file_name().ok_or_else(|| {
Error::PeerStartupFailed("Invalid gateway pubkey path".to_string())
})?;
remote.remote_work_dir().join(&id).join(gw_pubkey_name)
}
};
content.push_str(&format!(
"[[gateways]]\n\
address = {{ hostname = \"{}\" }}\n\
public_key = \"{}\"\n\n",
gw.address,
gw_pubkey_path.display()
));
}
std::fs::write(&gateways_toml, content)?;
if let PeerLocation::Remote(remote) = &location {
let remote_gateways_toml = remote.remote_work_dir().join(&id).join("gateways.toml");
remote.scp_upload(&gateways_toml, remote_gateways_toml.to_str().unwrap())?;
}
}
let env_vars = vec![
("NETWORK_ADDRESS".to_string(), network_address.clone()),
(
"PUBLIC_NETWORK_ADDRESS".to_string(),
network_address.clone(),
),
("PUBLIC_NETWORK_PORT".to_string(), network_port.to_string()),
(
"FREENET_TELEMETRY_ENABLED".to_string(),
"false".to_string(),
),
];
if let Some(min_conn) = self.min_connections {
args.push("--min-number-of-connections".to_string());
args.push(min_conn.to_string());
}
if let Some(max_conn) = self.max_connections {
args.push("--max-number-of-connections".to_string());
args.push(max_conn.to_string());
}
let process: Box<dyn PeerProcess + Send> = match &location {
PeerLocation::Local => Box::new(process::spawn_local_peer(
binary_path,
&args,
&data_dir,
&env_vars,
)?),
PeerLocation::Remote(remote) => {
let remote_data_dir = remote.remote_work_dir().join(&id);
let local_cache_dir = run_root.join(format!("{}-cache", id));
std::fs::create_dir_all(&local_cache_dir)?;
Box::new(
process::spawn_remote_peer(
binary_path,
&args,
remote,
&remote_data_dir,
&local_cache_dir,
&env_vars,
)
.await?,
)
}
};
tokio::time::sleep(Duration::from_millis(100)).await;
Ok(TestPeer {
id,
is_gateway,
ws_port,
network_port,
network_address,
data_dir,
process,
public_key_path: Some(public_key_path),
location,
})
}
async fn build_docker_nat(self, config: DockerNatConfig) -> Result<TestNetwork> {
let binary_path = self.binary.resolve()?;
tracing::info!(
"Starting Docker NAT test network: {} gateways, {} peers",
self.gateways,
self.peers
);
let base_dir = resolve_base_dir();
fs::create_dir_all(&base_dir)?;
cleanup_old_runs(&base_dir, 5)?;
let run_root = create_run_directory(&base_dir)?;
let mut run_status = RunStatusGuard::new(&run_root);
let mut docker_backend = DockerNatBackend::new(config).await.map_err(|e| {
run_status.mark("failure", Some(&format!("Docker init failed: {}", e)));
e
})?;
docker_backend.create_public_network().await.map_err(|e| {
run_status.mark(
"failure",
Some(&format!("Failed to create public network: {}", e)),
);
e
})?;
let ws_port: u16 = 9000;
let network_port: u16 = 31337;
let mut gateways = Vec::new();
for i in 0..self.gateways {
let data_dir = create_peer_dir(&run_root, &format!("gw{}", i))?;
let keypair_path = data_dir.join("keypair.pem");
let public_key_path = data_dir.join("public_key.pem");
generate_keypair(&keypair_path, &public_key_path)?;
let (info, process) = docker_backend
.create_gateway(
i,
&binary_path,
&keypair_path,
&public_key_path,
ws_port,
network_port,
&run_root,
)
.await
.map_err(|e| {
let detail = format!("failed to start gateway {}: {}", i, e);
run_status.mark("failure", Some(&detail));
e
})?;
let peer = TestPeer {
id: format!("gw{}", i),
is_gateway: true,
ws_port: info.host_ws_port,
network_port: info.network_port,
network_address: info.public_ip.to_string(),
data_dir,
process: Box::new(process),
public_key_path: Some(public_key_path),
location: PeerLocation::Local, };
gateways.push(peer);
}
let gateway_info: Vec<_> = gateways
.iter()
.map(|gw| GatewayInfo {
address: format!("{}:{}", gw.network_address, network_port),
public_key_path: gw
.public_key_path
.clone()
.expect("Gateway must have public key"),
})
.collect();
let mut peers = Vec::new();
for i in 0..self.peers {
let peer_index = i + self.gateways;
let data_dir = create_peer_dir(&run_root, &format!("peer{}", peer_index))?;
let keypair_path = data_dir.join("keypair.pem");
let public_key_path = data_dir.join("public_key.pem");
generate_keypair(&keypair_path, &public_key_path)?;
let gateways_toml_path = data_dir.join("gateways.toml");
let mut gateways_content = String::new();
for gw in &gateway_info {
gateways_content.push_str(&format!(
"[[gateways]]\n\
address = {{ hostname = \"{}\" }}\n\
public_key = \"/config/gw_public_key.pem\"\n\n",
gw.address,
));
}
std::fs::write(&gateways_toml_path, &gateways_content)?;
let gateway_public_key_path = gateway_info.first().map(|gw| gw.public_key_path.clone());
let (info, process) = docker_backend
.create_peer(
peer_index,
&binary_path,
&keypair_path,
&public_key_path,
&gateways_toml_path,
gateway_public_key_path.as_deref(),
ws_port,
network_port,
&run_root,
)
.await
.map_err(|e| {
let detail = format!("failed to start peer {}: {}", peer_index, e);
run_status.mark("failure", Some(&detail));
e
})?;
let peer = TestPeer {
id: format!("peer{}", peer_index),
is_gateway: false,
ws_port: info.host_ws_port,
network_port: info.network_port,
network_address: info.private_ip.to_string(),
data_dir,
process: Box::new(process),
public_key_path: Some(public_key_path),
location: PeerLocation::Local,
};
peers.push(peer);
if i + 1 < self.peers && !self.start_stagger.is_zero() {
tokio::time::sleep(self.start_stagger).await;
}
}
let network = TestNetwork::new_with_docker(
gateways,
peers,
self.min_connectivity,
run_root.clone(),
Some(docker_backend),
);
match network
.wait_until_ready_with_timeout(self.connectivity_timeout)
.await
{
Ok(()) => {
if self.preserve_data_on_success {
println!(
"Network data directories preserved at {}",
run_root.display()
);
}
let detail = format!(
"success: gateways={}, peers={} (Docker NAT)",
self.gateways, self.peers
);
run_status.mark("success", Some(&detail));
Ok(network)
}
Err(err) => {
if let Err(log_err) = dump_recent_logs(&network) {
eprintln!("Failed to dump logs after connectivity error: {}", log_err);
}
if self.preserve_data_on_failure {
eprintln!(
"Network data directories preserved at {}",
run_root.display()
);
}
let detail = err.to_string();
run_status.mark("failure", Some(&detail));
Err(err)
}
}
}
}
fn resolve_base_dir() -> PathBuf {
if let Some(path) = std::env::var_os("FREENET_TEST_NETWORK_BASE_DIR") {
PathBuf::from(path)
} else {
std::env::temp_dir().join("freenet-test-networks")
}
}
fn cleanup_old_runs(base_dir: &Path, max_runs: usize) -> Result<()> {
let mut runs: Vec<(PathBuf, SystemTime)> = fs::read_dir(base_dir)?
.filter_map(|entry| {
let entry = entry.ok()?;
let file_type = entry.file_type().ok()?;
if !file_type.is_dir() {
return None;
}
let metadata = entry.metadata().ok()?;
let modified = metadata.modified().unwrap_or(SystemTime::UNIX_EPOCH);
Some((entry.path(), modified))
})
.collect();
if runs.len() <= max_runs {
return Ok(());
}
runs.sort_by_key(|(_, modified)| *modified);
let remove_count = runs.len() - max_runs;
for (path, _) in runs.into_iter().take(remove_count) {
if let Err(err) = fs::remove_dir_all(&path) {
tracing::warn!(
?err,
path = %path.display(),
"Failed to remove old freenet test network run directory"
);
}
}
Ok(())
}
fn create_run_directory(base_dir: &Path) -> Result<PathBuf> {
let timestamp = Utc::now().format("%Y%m%d-%H%M%S").to_string();
for attempt in 0..100 {
let candidate = if attempt == 0 {
base_dir.join(×tamp)
} else {
base_dir.join(format!("{}-{}", ×tamp, attempt))
};
if !candidate.exists() {
fs::create_dir_all(&candidate)?;
return Ok(candidate);
}
}
Err(Error::Other(anyhow::anyhow!(
"Unable to allocate run directory after repeated attempts"
)))
}
fn create_peer_dir(run_root: &Path, id: &str) -> Result<PathBuf> {
let dir = run_root.join(id);
fs::create_dir_all(&dir)?;
Ok(dir)
}
struct RunStatusGuard {
status_path: PathBuf,
}
impl RunStatusGuard {
fn new(run_root: &Path) -> Self {
let status_path = run_root.join("run_status.txt");
let _ = fs::write(&status_path, b"status=initializing\n");
Self { status_path }
}
fn mark(&mut self, status: &str, detail: Option<&str>) {
let mut content = format!("status={}", status);
if let Some(detail) = detail {
content.push('\n');
content.push_str("detail=");
content.push_str(detail);
}
content.push('\n');
if let Err(err) = fs::write(&self.status_path, content) {
tracing::warn!(
?err,
path = %self.status_path.display(),
"Failed to write run status"
);
}
}
}
fn generate_keypair(
private_key_path: &std::path::Path,
public_key_path: &std::path::Path,
) -> Result<()> {
use rand::RngCore;
use x25519_dalek::{PublicKey, StaticSecret};
let mut secret_bytes = [0u8; 32];
rand::thread_rng().fill_bytes(&mut secret_bytes);
let secret = StaticSecret::from(secret_bytes);
let public = PublicKey::from(&secret);
drop(secret);
std::fs::write(private_key_path, hex::encode(secret_bytes))
.map_err(|e| Error::Other(anyhow::anyhow!("Failed to write private key: {}", e)))?;
std::fs::write(public_key_path, hex::encode(public.as_bytes()))
.map_err(|e| Error::Other(anyhow::anyhow!("Failed to write public key: {}", e)))?;
Ok(())
}
fn dump_recent_logs(network: &TestNetwork) -> Result<()> {
const MAX_LOG_LINES: usize = 200;
let mut logs = network.read_logs()?;
let total = logs.len();
if total > MAX_LOG_LINES {
logs.drain(0..(total - MAX_LOG_LINES));
}
eprintln!(
"\n--- Network connectivity check failed; showing {} of {} log entries ---",
logs.len(),
total
);
for entry in logs {
let level = entry.level.as_deref().unwrap_or("INFO");
let ts_display = entry
.timestamp_raw
.clone()
.or_else(|| entry.timestamp.map(|ts| ts.to_rfc3339()));
if let Some(ts) = ts_display {
eprintln!("[{}] [{}] {}: {}", entry.peer_id, ts, level, entry.message);
} else {
eprintln!("[{}] {}: {}", entry.peer_id, level, entry.message);
}
}
eprintln!("--- End of network logs ---\n");
Ok(())
}
fn preserve_network_state(network: &TestNetwork) -> Result<PathBuf> {
Ok(network.run_root().to_path_buf())
}