diff --git a/aether/src/cli.rs b/aether/src/cli.rs index bc95d68..4afec4f 100644 --- a/aether/src/cli.rs +++ b/aether/src/cli.rs @@ -40,6 +40,13 @@ MASQUE transport: --fragment-size fragment chunk size in bytes (default 16-32) --fragment-delay delay between fragments in ms (default 2-10) +Health monitoring: + --health-interval background tunnel health check interval in seconds (default 20, 0 to disable) + --health-fails consecutive failed health checks before reconnect (default 2) + --health-timeout timeout for each health check probe in seconds (default 5) + --health-url custom probe URL for health check (default http://www.gstatic.com/generate_204) + + WireGuard: --keepalive persistent keepalive interval in seconds (default 5) --no-profile-retry don't retry other obfuscation profiles during scan @@ -51,16 +58,23 @@ Config files: Advanced: --tls-groups TLS key share groups, e.g. \"P-256:X25519:P-384\" - --verbose detailed debug logs: tunnel stages, validation, reconnects, retries - (equivalent to RUST_LOG=info,aether=debug; RUST_LOG overrides this) - -v, --version show version and exit +Logging: + -v, --verbose detailed debug logs (use -vv for trace) + -l, --log-level set log level (error, warn, info, debug, trace) + + -V, --version show version and exit -h, --help show this help and exit "; pub fn parse_and_apply() -> crate::error::Result<()> { let args: Vec = env::args().skip(1).collect(); + parse_args(&args) +} + +pub fn parse_args(args: &[String]) -> crate::error::Result<()> { let mut i = 0; + let mut verbose_count = 0; while i < args.len() { let arg = args[i].as_str(); @@ -75,7 +89,7 @@ pub fn parse_and_apply() -> crate::error::Result<()> { } match arg { - "-v" | "--version" => { + "-V" | "--version" => { println!("aether {}", env!("CARGO_PKG_VERSION")); std::process::exit(0); } @@ -85,6 +99,19 @@ pub fn parse_and_apply() -> crate::error::Result<()> { std::process::exit(0); } + "-v" | "--verbose" => { + verbose_count += 1; + } + "-vv" => { + verbose_count += 2; + } + "-vvv" => { + verbose_count += 3; + } + "-l" | "--log-level" => { + set("AETHER_LOG", next_value!()); + } + "--bind" => set("AETHER_SOCKS", next_value!()), "--quick-reconnect" => set("AETHER_QUICK_RECONNECT", "1"), "--no-quick-reconnect" => set("AETHER_QUICK_RECONNECT", "0"), @@ -119,11 +146,21 @@ pub fn parse_and_apply() -> crate::error::Result<()> { set("AETHER_WG_NO_DATA_CHECK", "1"); } "--validate-secs" => set("AETHER_MASQUE_VALIDATE_SECS", next_value!()), - "--reconnect-secs" => set("AETHER_MASQUE_RECONNECT_SECS", next_value!()), + "--reconnect-secs" => { + let val = next_value!(); + set("AETHER_MASQUE_RECONNECT_SECS", val); + set("AETHER_WG_RECONNECT_SECS", val); + } "--fragment" => set("AETHER_MASQUE_H2_FRAGMENT", "1"), "--fragment-size" => set("AETHER_MASQUE_H2_FRAGMENT_SIZE", next_value!()), "--fragment-delay" => set("AETHER_MASQUE_H2_FRAGMENT_DELAY", next_value!()), + "--health-interval" => set("AETHER_HEALTH_INTERVAL", next_value!()), + "--health-fails" => set("AETHER_HEALTH_MAX_FAILS", next_value!()), + "--health-timeout" => set("AETHER_HEALTH_TIMEOUT", next_value!()), + "--health-url" => set("AETHER_HEALTH_PROBE_URL", next_value!()), + + "--keepalive" => set("AETHER_WG_KEEPALIVE", next_value!()), "--no-profile-retry" => set("AETHER_WG_NO_PROFILE_RETRY", "1"), @@ -132,7 +169,6 @@ pub fn parse_and_apply() -> crate::error::Result<()> { "--masque-config" => set("AETHER_MASQUE_CONFIG", next_value!()), "--tls-groups" => set("AETHER_TLS_GROUPS", next_value!()), - "--verbose" => set("AETHER_VERBOSE", "1"), other => { return Err(crate::error::AetherError::Other(format!( @@ -144,9 +180,91 @@ pub fn parse_and_apply() -> crate::error::Result<()> { i += 1; } + if verbose_count > 0 && std::env::var("AETHER_LOG").is_err() { + if verbose_count == 1 { + set("AETHER_LOG", "info,aether=debug"); + } else { + set("AETHER_LOG", "info,aether=trace"); + } + } + Ok(()) } fn set(key: &str, value: &str) { std::env::set_var(key, value); } + +#[cfg(test)] +pub static ENV_MUTEX: std::sync::Mutex<()> = std::sync::Mutex::new(()); + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_cli_parse_log_levels() { + let _guard = ENV_MUTEX.lock().unwrap(); + std::env::remove_var("AETHER_LOG"); + parse_args(&["--verbose".to_string()]).unwrap(); + assert_eq!(std::env::var("AETHER_LOG").unwrap(), "info,aether=debug"); + + std::env::remove_var("AETHER_LOG"); + parse_args(&["-v".to_string()]).unwrap(); + assert_eq!(std::env::var("AETHER_LOG").unwrap(), "info,aether=debug"); + + std::env::remove_var("AETHER_LOG"); + parse_args(&["-vv".to_string()]).unwrap(); + assert_eq!(std::env::var("AETHER_LOG").unwrap(), "info,aether=trace"); + + std::env::remove_var("AETHER_LOG"); + parse_args(&["-l".to_string(), "warn".to_string()]).unwrap(); + assert_eq!(std::env::var("AETHER_LOG").unwrap(), "warn"); + + std::env::remove_var("AETHER_LOG"); + parse_args(&["--log-level".to_string(), "trace".to_string()]).unwrap(); + assert_eq!(std::env::var("AETHER_LOG").unwrap(), "trace"); + std::env::remove_var("AETHER_LOG"); + } + + #[test] + fn test_cli_parse_health_options() { + let _guard = ENV_MUTEX.lock().unwrap(); + parse_args(&[ + "--health-interval".to_string(), + "10".to_string(), + "--health-fails".to_string(), + "4".to_string(), + "--health-timeout".to_string(), + "3".to_string(), + "--health-url".to_string(), + "http://cp.cloudflare.com/generate_204".to_string(), + "--reconnect-secs".to_string(), + "5".to_string(), + ]) + .unwrap(); + + assert_eq!(std::env::var("AETHER_HEALTH_INTERVAL").unwrap(), "10"); + assert_eq!(std::env::var("AETHER_HEALTH_MAX_FAILS").unwrap(), "4"); + assert_eq!(std::env::var("AETHER_HEALTH_TIMEOUT").unwrap(), "3"); + assert_eq!( + std::env::var("AETHER_HEALTH_PROBE_URL").unwrap(), + "http://cp.cloudflare.com/generate_204" + ); + assert_eq!(std::env::var("AETHER_MASQUE_RECONNECT_SECS").unwrap(), "5"); + assert_eq!(std::env::var("AETHER_WG_RECONNECT_SECS").unwrap(), "5"); + + std::env::remove_var("AETHER_HEALTH_INTERVAL"); + std::env::remove_var("AETHER_HEALTH_MAX_FAILS"); + std::env::remove_var("AETHER_HEALTH_TIMEOUT"); + std::env::remove_var("AETHER_HEALTH_PROBE_URL"); + std::env::remove_var("AETHER_MASQUE_RECONNECT_SECS"); + std::env::remove_var("AETHER_WG_RECONNECT_SECS"); + } +} + + + + + + diff --git a/aether/src/main.rs b/aether/src/main.rs index 2f81c78..952d673 100644 --- a/aether/src/main.rs +++ b/aether/src/main.rs @@ -41,12 +41,12 @@ const DEFAULT_CONFIG: &str = "aether.toml"; async fn main() -> Result<()> { cli::parse_and_apply()?; - let default_filter = if std::env::var("AETHER_VERBOSE").is_ok() { - "info,aether=debug" + let log_env = if std::env::var("AETHER_LOG").is_ok() { + env_logger::Env::default().filter("AETHER_LOG") } else { - "info" + env_logger::Env::default().default_filter_or("info") }; - env_logger::Builder::from_env(env_logger::Env::default().default_filter_or(default_filter)) + env_logger::Builder::from_env(log_env) .format_timestamp_millis() .init(); @@ -557,22 +557,39 @@ async fn run_masque_tunnel( } } + let health_cfg = tunnelping::HealthConfig::from_env(); + let health_task = tunnelping::spawn_health_monitor(stack.clone(), health_cfg); + let socks_stack = stack.clone(); let socks_task = tokio::spawn(async move { log::info!("[+] socks5 server listening on {listen}"); socks::serve(listen, socks_stack).await }); - let tunnel_result = tunnel_task.await; - socks_task.abort(); + let tunnel_result = tokio::select! { + res = tunnel_task => match res { + Ok(Ok(())) => Ok(()), + Ok(Err(e)) => Err(AetherError::Other(format!("tunnel exited: {e}"))), + Err(e) => Err(AetherError::Other(format!("tunnel task join error: {e}"))), + }, + health_res = async { + if let Some(h) = health_task { + h.await + } else { + std::future::pending().await + } + } => match health_res { + Ok(Err(e)) => Err(AetherError::Other(format!("health check failure: {e}"))), + Ok(Ok(())) => Err(AetherError::Other("health monitor exited unexpectedly".into())), + Err(e) => Err(AetherError::Other(format!("health monitor join error: {e}"))), + } + }; - match tunnel_result { - Ok(Ok(())) => Ok(()), - Ok(Err(e)) => Err(AetherError::Other(format!("tunnel exited: {e}"))), - Err(e) => Err(AetherError::Other(format!("tunnel task join error: {e}"))), - } + socks_task.abort(); + tunnel_result } + fn wg_keepalive_secs() -> u16 { std::env::var("AETHER_WG_KEEPALIVE") .ok() @@ -851,21 +868,41 @@ async fn run_wireguard_tunnel( let stack = netstack::spawn(&identity.ipv4, &identity.ipv6, TUNNEL_MTU, inbound_rx, outbound_tx)?; + let health_cfg = tunnelping::HealthConfig::from_env(); + let health_task = tunnelping::spawn_health_monitor(stack.clone(), health_cfg); + let socks_stack = stack.clone(); let socks_task = tokio::spawn(async move { log::info!("[+] socks5 server listening on {listen}"); socks::serve(listen, socks_stack).await }); - let tunnel_result = tunnel.run(outbound_rx).await; - socks_task.abort(); + let tunnel_task = tokio::spawn(tunnel.run(outbound_rx)); - match tunnel_result { - Ok(()) => Ok(()), - Err(e) => Err(AetherError::Other(format!("wireguard tunnel exited: {e}"))), - } + let res = tokio::select! { + tunnel_res = tunnel_task => match tunnel_res { + Ok(Ok(())) => Ok(()), + Ok(Err(e)) => Err(AetherError::Other(format!("wireguard tunnel exited: {e}"))), + Err(e) => Err(AetherError::Other(format!("wireguard tunnel task join error: {e}"))), + }, + health_res = async { + if let Some(h) = health_task { + h.await + } else { + std::future::pending().await + } + } => match health_res { + Ok(Err(e)) => Err(AetherError::Other(format!("health check failure: {e}"))), + Ok(Ok(())) => Err(AetherError::Other("health monitor exited unexpectedly".into())), + Err(e) => Err(AetherError::Other(format!("health monitor join error: {e}"))), + } + }; + + socks_task.abort(); + res } + async fn establish_wg( identity: &account::Identity, peer: SocketAddr, @@ -979,11 +1016,36 @@ async fn run_warp_in_warp( log::info!("[*] establishing inner WARP tunnel (warp-in-warp)..."); let inner_stack = establish_wg(&secondary, forwarder, INNER_MTU, false, 20, "inner").await?; + let health_cfg = tunnelping::HealthConfig::from_env(); + let health_task = tunnelping::spawn_health_monitor(inner_stack.clone(), health_cfg); + + let socks_task = tokio::spawn(async move { + log::info!("[+] socks5 server listening on {listen}"); + socks::serve(listen, inner_stack).await + }); - log::info!("[+] socks5 server listening on {listen}"); - socks::serve(listen, inner_stack).await + let res = tokio::select! { + socks_res = socks_task => match socks_res { + Ok(res) => res, + Err(e) => Err(AetherError::Other(format!("socks server join error: {e}"))), + }, + health_res = async { + if let Some(h) = health_task { + h.await + } else { + std::future::pending().await + } + } => match health_res { + Ok(Err(e)) => Err(AetherError::Other(format!("health check failure: {e}"))), + Ok(Ok(())) => Err(AetherError::Other("health monitor exited unexpectedly".into())), + Err(e) => Err(AetherError::Other(format!("health monitor join error: {e}"))), + } + }; + + res } + async fn prompt_line(prompt: &str) -> Option { use std::io::IsTerminal; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; diff --git a/aether/src/masque_h2.rs b/aether/src/masque_h2.rs index a1dc258..051708a 100644 --- a/aether/src/masque_h2.rs +++ b/aether/src/masque_h2.rs @@ -434,7 +434,7 @@ pub async fn run( Some(ip_packet) => { let framed = masque::encode_datagram_capsule(&ip_packet); if let Err(e) = send_capsule(&mut send_stream, Bytes::from(framed)).await { - log::debug!("[h2] send: {e}"); + log::trace!("[h2] send: {e}"); return Err(e); } } @@ -545,7 +545,7 @@ async fn drain_capsules( Ok(Some(_)) => {} Ok(None) => break, Err(e) => { - log::debug!("[h2] capsule parse: {e}"); + log::trace!("[h2] capsule parse: {e}"); break; } } diff --git a/aether/src/noize.rs b/aether/src/noize.rs index 8593a68..0c5b953 100644 --- a/aether/src/noize.rs +++ b/aether/src/noize.rs @@ -141,8 +141,8 @@ pub async fn pre_handshake(sock: &UdpSocket, peer: SocketAddr, cfg: &NoizeConfig for i in 0..cfg.jc_before_hs { let pkt = junk_packet(cfg); match sock.send_to(&pkt, peer).await { - Ok(n) => log::debug!("junk[{i}] sent {n} bytes"), - Err(e) => log::debug!("junk[{i}] send failed: {e}"), + Ok(n) => log::trace!("junk[{i}] sent {n} bytes"), + Err(e) => log::trace!("junk[{i}] send failed: {e}"), } if !cfg.junk_interval.is_zero() { tokio::time::sleep(cfg.junk_interval).await; @@ -153,8 +153,8 @@ pub async fn pre_handshake(sock: &UdpSocket, peer: SocketAddr, cfg: &NoizeConfig let pkt = parse_cps(i1); if !pkt.is_empty() { match sock.send_to(&pkt, peer).await { - Ok(n) => log::debug!("signature i1 sent {n} bytes"), - Err(e) => log::debug!("signature i1 send failed: {e}"), + Ok(n) => log::trace!("signature i1 sent {n} bytes"), + Err(e) => log::trace!("signature i1 send failed: {e}"), } tokio::time::sleep(Duration::from_millis(2)).await; } @@ -163,8 +163,8 @@ pub async fn pre_handshake(sock: &UdpSocket, peer: SocketAddr, cfg: &NoizeConfig for i in 0..cfg.jc_after_i1 { let pkt = junk_packet(cfg); match sock.send_to(&pkt, peer).await { - Ok(n) => log::debug!("junk_after[{i}] sent {n} bytes"), - Err(e) => log::debug!("junk_after[{i}] send failed: {e}"), + Ok(n) => log::trace!("junk_after[{i}] sent {n} bytes"), + Err(e) => log::trace!("junk_after[{i}] send failed: {e}"), } if !cfg.junk_interval.is_zero() { tokio::time::sleep(cfg.junk_interval).await; @@ -175,8 +175,8 @@ pub async fn pre_handshake(sock: &UdpSocket, peer: SocketAddr, cfg: &NoizeConfig let pkt = parse_cps(i2); if !pkt.is_empty() { match sock.send_to(&pkt, peer).await { - Ok(n) => log::debug!("signature i2 sent {n} bytes"), - Err(e) => log::debug!("signature i2 send failed: {e}"), + Ok(n) => log::trace!("signature i2 sent {n} bytes"), + Err(e) => log::trace!("signature i2 send failed: {e}"), } } } diff --git a/aether/src/quic.rs b/aether/src/quic.rs index 70c7349..0b1a8cf 100644 --- a/aether/src/quic.rs +++ b/aether/src/quic.rs @@ -137,13 +137,13 @@ fn spawn_reader(sock: Arc, local: SocketAddr, tx: mpsc::Sender { - log::debug!("recv {n} bytes from {from}"); + log::trace!("recv {n} bytes from {from}"); if tx.send((local, from, buf[..n].to_vec())).await.is_err() { break; } }, Err(e) => { - log::debug!("recv error: {e}"); + log::trace!("recv error: {e}"); break; } } @@ -258,11 +258,11 @@ pub async fn run( Some((to_local, from, mut data)) = net_rx.recv() => { let mut hdr_buf = data.clone(); if let Ok(hdr) = quiche::Header::from_slice(&mut hdr_buf, quiche::MAX_CONN_ID_LEN) { - log::debug!("recv {} bytes type={:?} version=0x{:x} from {}", data.len(), hdr.ty, hdr.version, from); + log::trace!("recv {} bytes type={:?} version=0x{:x} from {}", data.len(), hdr.ty, hdr.version, from); } let info = quiche::RecvInfo { from, to: to_local }; if let Err(e) = conn.recv(&mut data, info) { - log::debug!("recv error: {e}"); + log::trace!("recv error: {e}"); } } @@ -286,10 +286,10 @@ pub async fn run( match masque::encode_ip_datagram(sid, &ip_packet) { Ok(framed) => { if let Err(e) = conn.dgram_send(&framed) { - log::debug!("dgram_send: {e}"); + log::trace!("dgram_send: {e}"); } } - Err(e) => log::debug!("encap: {e}"), + Err(e) => log::trace!("encap: {e}"), } } } @@ -525,11 +525,11 @@ async fn drain_datagrams( } } Ok(None) => {} - Err(e) => log::debug!("decap: {e}"), + Err(e) => log::trace!("decap: {e}"), }, Err(quiche::Error::Done) => break, Err(e) => { - log::debug!("dgram_recv: {e}"); + log::trace!("dgram_recv: {e}"); break; } } @@ -671,11 +671,11 @@ pub async fn verify_masque(p: &VerifyParams) -> Result { Ok(n) => { let mut hdr_buf = buf[..n].to_vec(); if let Ok(hdr) = quiche::Header::from_slice(&mut hdr_buf, quiche::MAX_CONN_ID_LEN) { - log::debug!("verify recv {} bytes type={:?} version=0x{:x} from {}", n, hdr.ty, hdr.version, p.peer); + log::trace!("verify recv {} bytes type={:?} version=0x{:x} from {}", n, hdr.ty, hdr.version, p.peer); } let info = quiche::RecvInfo { from: p.peer, to: local }; if let Err(e) = conn.recv(&mut buf[..n], info) { - log::debug!("verify recv error from {}: {e}", p.peer); + log::trace!("verify recv error from {}: {e}", p.peer); } } Err(e) => return Err(AetherError::Io(e)), diff --git a/aether/src/tunnelping.rs b/aether/src/tunnelping.rs index 8e322a0..5b28680 100644 --- a/aether/src/tunnelping.rs +++ b/aether/src/tunnelping.rs @@ -31,15 +31,79 @@ fn http_probe_port() -> u16 { .unwrap_or(80) } -async fn http_probe(stack: &netstack::StackHandle) -> Result<()> { - let ip = socks::dns_resolve(stack, HTTP_PROBE_HOST).await?; - let dst = SocketAddr::new(ip, http_probe_port()); +pub async fn http_probe(stack: &netstack::StackHandle) -> Result<()> { + let target = parse_probe_url("")?; + http_probe_target(stack, &target).await.map(|_| ()) +} + + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ProbeTarget { + pub host: String, + pub port: u16, + pub path: String, +} + +pub fn parse_probe_url(url: &str) -> Result { + let default_port = http_probe_port(); + let default_target = ProbeTarget { + host: HTTP_PROBE_HOST.to_string(), + port: default_port, + path: HTTP_PROBE_PATH.to_string(), + }; + + let s = url.trim(); + if s.is_empty() { + return Ok(default_target); + } + + if s.to_lowercase().starts_with("https://") { + return Err(AetherError::Other( + "https:// health probe URLs are not supported; please use an http:// URL".into(), + )); + } + + let s = if let Some(stripped) = s.strip_prefix("http://") { + stripped + } else { + s + }; + + let (host_port, path) = match s.split_once('/') { + Some((hp, p)) => (hp, format!("/{p}")), + None => (s, "/".to_string()), + }; + + let (host, port) = match host_port.rsplit_once(':') { + Some((h, p)) => match p.parse::() { + Ok(parsed_port) => (h.to_string(), parsed_port), + Err(_) => (host_port.to_string(), default_port), + }, + None => (host_port.to_string(), default_port), + }; + + if host.is_empty() { + return Ok(default_target); + } + + Ok(ProbeTarget { host, port, path }) +} + + +pub async fn http_probe_target( + stack: &netstack::StackHandle, + target: &ProbeTarget, +) -> Result { + let start = Instant::now(); + let ip = socks::dns_resolve(stack, &target.host).await?; + let dst = SocketAddr::new(ip, target.port); let conn = stack.open_tcp(dst).await?; let (sender, mut from_stack) = conn.into_split(); let request = format!( - "GET {HTTP_PROBE_PATH} HTTP/1.1\r\nHost: {HTTP_PROBE_HOST}\r\nConnection: close\r\nUser-Agent: aether-ironclad\r\n\r\n" + "GET {} HTTP/1.1\r\nHost: {}\r\nConnection: close\r\nUser-Agent: aether-ironclad\r\n\r\n", + target.path, target.host ); sender.send(request.into_bytes()).await?; @@ -60,8 +124,8 @@ async fn http_probe(stack: &netstack::StackHandle) -> Result<()> { sender.close().await; let status_line = String::from_utf8_lossy(&buf); - if status_line.contains("204") { - Ok(()) + if status_line.contains("204") || status_line.contains("200") { + Ok(start.elapsed()) } else { let first_line = status_line.lines().next().unwrap_or("").trim(); Err(AetherError::Other(format!( @@ -70,6 +134,137 @@ async fn http_probe(stack: &netstack::StackHandle) -> Result<()> { } } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct HealthConfig { + pub interval: Duration, + pub max_fails: u32, + pub timeout: Duration, + pub probe_url: String, +} + +impl Default for HealthConfig { + fn default() -> Self { + Self::from_env() + } +} + +impl HealthConfig { + pub fn from_env() -> Self { + let interval_secs = std::env::var("AETHER_HEALTH_INTERVAL") + .ok() + .and_then(|v| v.parse::().ok()) + .unwrap_or(20); + + let max_fails = std::env::var("AETHER_HEALTH_MAX_FAILS") + .ok() + .and_then(|v| v.parse::().ok()) + .filter(|&v| v > 0) + .unwrap_or(2); + + let timeout_secs = std::env::var("AETHER_HEALTH_TIMEOUT") + .ok() + .and_then(|v| v.parse::().ok()) + .filter(|&v| v > 0) + .unwrap_or(5); + + let probe_url = std::env::var("AETHER_HEALTH_PROBE_URL") + .unwrap_or_else(|_| format!("http://{HTTP_PROBE_HOST}{HTTP_PROBE_PATH}")); + + Self { + interval: Duration::from_secs(interval_secs), + max_fails, + timeout: Duration::from_secs(timeout_secs), + probe_url, + } + } +} + +pub async fn check_health( + stack: &netstack::StackHandle, + target: &ProbeTarget, + timeout: Duration, +) -> Result { + match tokio::time::timeout(timeout, http_probe_target(stack, target)).await { + Ok(res) => res, + Err(_) => Err(AetherError::Other("tunnel health check timeout".into())), + } +} + +pub fn spawn_health_monitor( + stack: netstack::StackHandle, + config: HealthConfig, +) -> Option>> { + if config.interval.is_zero() { + log::info!("[*] continuous tunnel health monitoring disabled (interval=0)"); + return None; + } + + let target = match parse_probe_url(&config.probe_url) { + Ok(t) => t, + Err(e) => { + log::warn!( + "[-] invalid health probe URL ('{}'): {e}; disabling continuous health monitor", + config.probe_url + ); + return None; + } + }; + + + log::info!( + "[+] starting background tunnel health monitoring (url='http://{}:{}{}', interval={:?}, max_fails={}, timeout={:?})", + target.host, + target.port, + target.path, + config.interval, + config.max_fails, + config.timeout + ); + + Some(tokio::spawn(async move { + let mut interval = tokio::time::interval(config.interval); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + interval.tick().await; + + let mut consecutive_fails: u32 = 0; + + loop { + interval.tick().await; + match check_health(&stack, &target, config.timeout).await { + Ok(rtt) => { + log::debug!("[+] tunnel health check passed (rtt {:?})", rtt); + if consecutive_fails > 0 { + log::info!( + "[+] tunnel health check passed (rtt {:?}); consecutive failure count reset to 0", + rtt + ); + } + consecutive_fails = 0; + } + Err(e) => { + consecutive_fails += 1; + log::warn!( + "[-] tunnel health check failed ({consecutive_fails}/{}): {e}", + config.max_fails + ); + if consecutive_fails >= config.max_fails { + log::error!( + "[-] tunnel health check failed {} consecutive times; declaring tunnel dead", + consecutive_fails + ); + return Err(AetherError::Other(format!( + "tunnel health check failed {} consecutive times", + consecutive_fails + ))); + } + } + } + } + })) +} + + + pub struct MasquePingParams { pub peer: SocketAddr, pub sni: String, @@ -197,3 +392,84 @@ pub async fn wg_http_ping_established( Err(_) => Err(AetherError::Other("ironclad http probe timeout".into())), } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_health_config_from_env() { + let _guard = crate::cli::ENV_MUTEX.lock().unwrap(); + std::env::set_var("AETHER_HEALTH_INTERVAL", "15"); + std::env::set_var("AETHER_HEALTH_MAX_FAILS", "5"); + std::env::set_var("AETHER_HEALTH_TIMEOUT", "10"); + + let cfg = HealthConfig::from_env(); + assert_eq!(cfg.interval, Duration::from_secs(15)); + assert_eq!(cfg.max_fails, 5); + assert_eq!(cfg.timeout, Duration::from_secs(10)); + + std::env::remove_var("AETHER_HEALTH_INTERVAL"); + std::env::remove_var("AETHER_HEALTH_MAX_FAILS"); + std::env::remove_var("AETHER_HEALTH_TIMEOUT"); + + let default_cfg = HealthConfig::from_env(); + assert_eq!(default_cfg.interval, Duration::from_secs(20)); + assert_eq!(default_cfg.max_fails, 2); + assert_eq!(default_cfg.timeout, Duration::from_secs(5)); + } + + #[test] + fn test_parse_probe_url() { + let t1 = parse_probe_url("http://cp.cloudflare.com/generate_204").unwrap(); + assert_eq!(t1.host, "cp.cloudflare.com"); + assert_eq!(t1.port, 80); + assert_eq!(t1.path, "/generate_204"); + + let t2 = parse_probe_url("http://1.1.1.1:8080/check").unwrap(); + assert_eq!(t2.host, "1.1.1.1"); + assert_eq!(t2.port, 8080); + assert_eq!(t2.path, "/check"); + + let t3 = parse_probe_url("").unwrap(); + assert_eq!(t3.host, HTTP_PROBE_HOST); + assert_eq!(t3.port, 80); + assert_eq!(t3.path, HTTP_PROBE_PATH); + + let err = parse_probe_url("https://cp.cloudflare.com/generate_204"); + assert!(err.is_err()); + assert!(err.unwrap_err().to_string().contains("https:// health probe URLs are not supported")); + } + + + #[tokio::test] + async fn test_spawn_health_monitor_failure_trigger() { + let (outbound_tx, _outbound_rx) = tokio::sync::mpsc::channel(128); + let (_inbound_tx, inbound_rx) = tokio::sync::mpsc::channel(128); + + let stack = netstack::spawn("172.16.0.2", "fd00::2", 1500, inbound_rx, outbound_tx) + .expect("spawn netstack"); + + let config = HealthConfig { + interval: Duration::from_millis(20), + max_fails: 2, + timeout: Duration::from_millis(50), + probe_url: "http://1.1.1.1/generate_204".to_string(), + }; + + let handle = spawn_health_monitor(stack, config).expect("spawn health monitor"); + let result = handle.await.expect("join handle"); + + assert!(result.is_err(), "expected health monitor to fail"); + let err_msg = result.unwrap_err().to_string(); + assert!( + err_msg.contains("failed 2 consecutive times"), + "unexpected error message: {err_msg}" + ); + } +} + + + + + diff --git a/aether/src/wireguard.rs b/aether/src/wireguard.rs index 66bc8d8..a7b33fb 100644 --- a/aether/src/wireguard.rs +++ b/aether/src/wireguard.rs @@ -137,7 +137,7 @@ impl WgTunnel { match tunn.decapsulate(None, &buf[..n], &mut tmp) { TunnResult::Done => {} TunnResult::Err(e) => { - log::debug!("decapsulate error: {e:?}"); + log::trace!("decapsulate error: {e:?}"); } TunnResult::WriteToNetwork(pkt) => { let mut pkt_vec = pkt.to_vec(); @@ -168,7 +168,7 @@ impl WgTunnel { match tunn.encapsulate(&ip_packet, &mut out_buf) { TunnResult::Done => {} TunnResult::Err(e) => { - log::debug!("encapsulate error: {e:?}"); + log::trace!("encapsulate error: {e:?}"); } TunnResult::WriteToNetwork(pkt) => { let mut pkt_vec = pkt.to_vec(); @@ -516,10 +516,10 @@ pub async fn verify_endpoint_keep_session( })); } TunnResult::Err(e) => { - log::debug!("[wg] decap error: {:?}", e); + log::trace!("[wg] decap error: {:?}", e); } other => { - log::debug!("[wg] unexpected decap: {:?}", other); + log::trace!("[wg] unexpected decap: {:?}", other); } } }