From 89e1032cebb78456e3253eecb9eb68c58321d8db Mon Sep 17 00:00:00 2001 From: Ardeshir Nahavandi Fard <68429735+Vonarian@users.noreply.github.com> Date: Mon, 20 Jul 2026 09:40:16 +0330 Subject: [PATCH 1/7] feat(cli): add -v, --verbose, and --log-level options --- aether/src/cli.rs | 66 +++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 61 insertions(+), 5 deletions(-) diff --git a/aether/src/cli.rs b/aether/src/cli.rs index bc95d68..78447ff 100644 --- a/aether/src/cli.rs +++ b/aether/src/cli.rs @@ -51,16 +51,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 +82,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 +92,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"), @@ -132,7 +152,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 +163,46 @@ 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", "debug"); + } else { + set("AETHER_LOG", "trace"); + } + } + Ok(()) } fn set(key: &str, value: &str) { std::env::set_var(key, value); } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_cli_parse_log_levels() { + std::env::remove_var("AETHER_LOG"); + parse_args(&["--verbose".to_string()]).unwrap(); + assert_eq!(std::env::var("AETHER_LOG").unwrap(), "debug"); + + std::env::remove_var("AETHER_LOG"); + parse_args(&["-v".to_string()]).unwrap(); + assert_eq!(std::env::var("AETHER_LOG").unwrap(), "debug"); + + std::env::remove_var("AETHER_LOG"); + parse_args(&["-vv".to_string()]).unwrap(); + assert_eq!(std::env::var("AETHER_LOG").unwrap(), "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"); + } +} + From 74e4a36641f9deadc10ef1cb3fd400787ed12323 Mon Sep 17 00:00:00 2001 From: Ardeshir Nahavandi Fard <68429735+Vonarian@users.noreply.github.com> Date: Mon, 20 Jul 2026 09:40:27 +0330 Subject: [PATCH 2/7] feat(main): initialize env_logger after CLI argument parsing --- aether/src/main.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/aether/src/main.rs b/aether/src/main.rs index 2f81c78..5b16999 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(); From 2dfc756cbdafd4b84ff2143775fa7818e9fe0174 Mon Sep 17 00:00:00 2001 From: Ardeshir Nahavandi Fard <68429735+Vonarian@users.noreply.github.com> Date: Mon, 20 Jul 2026 09:41:09 +0330 Subject: [PATCH 3/7] refactor(log): demote per-packet debug logs to trace level --- aether/src/masque_h2.rs | 4 ++-- aether/src/noize.rs | 16 ++++++++-------- aether/src/quic.rs | 20 ++++++++++---------- aether/src/wireguard.rs | 8 ++++---- 4 files changed, 24 insertions(+), 24 deletions(-) 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/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); } } } From 6a9a3fdd193bc23d160d2f41398eea6273b109ad Mon Sep 17 00:00:00 2001 From: Ardeshir Nahavandi Fard <68429735+Vonarian@users.noreply.github.com> Date: Mon, 20 Jul 2026 09:58:04 +0330 Subject: [PATCH 4/7] fix(cli): scope -v and -vv flags to aether crate log level --- aether/src/cli.rs | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/aether/src/cli.rs b/aether/src/cli.rs index 78447ff..87506d4 100644 --- a/aether/src/cli.rs +++ b/aether/src/cli.rs @@ -165,9 +165,9 @@ pub fn parse_args(args: &[String]) -> crate::error::Result<()> { if verbose_count > 0 && std::env::var("AETHER_LOG").is_err() { if verbose_count == 1 { - set("AETHER_LOG", "debug"); + set("AETHER_LOG", "info,aether=debug"); } else { - set("AETHER_LOG", "trace"); + set("AETHER_LOG", "info,aether=trace"); } } @@ -186,15 +186,15 @@ mod tests { fn test_cli_parse_log_levels() { std::env::remove_var("AETHER_LOG"); parse_args(&["--verbose".to_string()]).unwrap(); - assert_eq!(std::env::var("AETHER_LOG").unwrap(), "debug"); + 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(), "debug"); + 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(), "trace"); + 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(); @@ -206,3 +206,4 @@ mod tests { } } + From 468808f3b92c4be511b033d2d8e0f821a2adece7 Mon Sep 17 00:00:00 2001 From: Ardeshir Nahavandi Fard <68429735+Vonarian@users.noreply.github.com> Date: Mon, 20 Jul 2026 15:03:00 +0330 Subject: [PATCH 5/7] feat(health): add configurable tunnel health monitoring and reconnect thresholds (closes #44) --- aether/src/cli.rs | 54 ++++++++- aether/src/main.rs | 92 ++++++++++++--- aether/src/tunnelping.rs | 240 ++++++++++++++++++++++++++++++++++++++- 3 files changed, 364 insertions(+), 22 deletions(-) diff --git a/aether/src/cli.rs b/aether/src/cli.rs index 87506d4..f9ad530 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 @@ -139,11 +146,21 @@ pub fn parse_args(args: &[String]) -> 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"), @@ -178,12 +195,16 @@ 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"); @@ -204,6 +225,37 @@ mod tests { parse_args(&["--log-level".to_string(), "trace".to_string()]).unwrap(); assert_eq!(std::env::var("AETHER_LOG").unwrap(), "trace"); } + + #[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"); + } } + + + diff --git a/aether/src/main.rs b/aether/src/main.rs index 5b16999..952d673 100644 --- a/aether/src/main.rs +++ b/aether/src/main.rs @@ -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/tunnelping.rs b/aether/src/tunnelping.rs index 8e322a0..a312439 100644 --- a/aether/src/tunnelping.rs +++ b/aether/src/tunnelping.rs @@ -31,15 +31,73 @@ 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) -> ProbeTarget { + 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 default_target; + } + + let s = if let Some(stripped) = s.strip_prefix("http://") { + stripped + } else if let Some(stripped) = s.strip_prefix("https://") { + 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 default_target; + } + + 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 +118,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 +128,127 @@ 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 = parse_probe_url(&config.probe_url); + + 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 +376,52 @@ 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"); + 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"); + assert_eq!(t2.host, "1.1.1.1"); + assert_eq!(t2.port, 8080); + assert_eq!(t2.path, "/check"); + + let t3 = parse_probe_url(""); + assert_eq!(t3.host, HTTP_PROBE_HOST); + assert_eq!(t3.port, 80); + assert_eq!(t3.path, HTTP_PROBE_PATH); + } +} + + + + From f967e630e4dd9ae9d551e5a3a27b92d6433ba649 Mon Sep 17 00:00:00 2001 From: Ardeshir Nahavandi Fard <68429735+Vonarian@users.noreply.github.com> Date: Mon, 20 Jul 2026 15:04:31 +0330 Subject: [PATCH 6/7] test(health): add async test for spawn_health_monitor failure threshold trigger --- aether/src/tunnelping.rs | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/aether/src/tunnelping.rs b/aether/src/tunnelping.rs index a312439..d4c9573 100644 --- a/aether/src/tunnelping.rs +++ b/aether/src/tunnelping.rs @@ -420,8 +420,35 @@ mod tests { assert_eq!(t3.port, 80); assert_eq!(t3.path, HTTP_PROBE_PATH); } + + #[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}" + ); + } } + From 97892b14593b712de4d79e4e2bd2896173116afd Mon Sep 17 00:00:00 2001 From: Ardeshir Nahavandi Fard <68429735+Vonarian@users.noreply.github.com> Date: Mon, 20 Jul 2026 15:19:00 +0330 Subject: [PATCH 7/7] fix(health): resolve PR review comments on task cancellation, URL parsing, and test env cleanup --- aether/src/cli.rs | 9 +++++++++ aether/src/tunnelping.rs | 43 ++++++++++++++++++++++++++++++---------- 2 files changed, 41 insertions(+), 11 deletions(-) diff --git a/aether/src/cli.rs b/aether/src/cli.rs index f9ad530..4afec4f 100644 --- a/aether/src/cli.rs +++ b/aether/src/cli.rs @@ -224,6 +224,7 @@ mod tests { 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] @@ -252,6 +253,13 @@ mod tests { ); 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"); } } @@ -259,3 +267,4 @@ mod tests { + diff --git a/aether/src/tunnelping.rs b/aether/src/tunnelping.rs index d4c9573..5b28680 100644 --- a/aether/src/tunnelping.rs +++ b/aether/src/tunnelping.rs @@ -32,10 +32,11 @@ fn http_probe_port() -> u16 { } pub async fn http_probe(stack: &netstack::StackHandle) -> Result<()> { - let target = parse_probe_url(""); + let target = parse_probe_url("")?; http_probe_target(stack, &target).await.map(|_| ()) } + #[derive(Debug, Clone, PartialEq, Eq)] pub struct ProbeTarget { pub host: String, @@ -43,7 +44,7 @@ pub struct ProbeTarget { pub path: String, } -pub fn parse_probe_url(url: &str) -> ProbeTarget { +pub fn parse_probe_url(url: &str) -> Result { let default_port = http_probe_port(); let default_target = ProbeTarget { host: HTTP_PROBE_HOST.to_string(), @@ -53,13 +54,17 @@ pub fn parse_probe_url(url: &str) -> ProbeTarget { let s = url.trim(); if s.is_empty() { - return default_target; + 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 if let Some(stripped) = s.strip_prefix("https://") { - stripped } else { s }; @@ -78,12 +83,13 @@ pub fn parse_probe_url(url: &str) -> ProbeTarget { }; if host.is_empty() { - return default_target; + return Ok(default_target); } - ProbeTarget { host, port, path } + Ok(ProbeTarget { host, port, path }) } + pub async fn http_probe_target( stack: &netstack::StackHandle, target: &ProbeTarget, @@ -193,7 +199,17 @@ pub fn spawn_health_monitor( return None; } - let target = parse_probe_url(&config.probe_url); + 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={:?})", @@ -405,22 +421,27 @@ mod tests { #[test] fn test_parse_probe_url() { - let t1 = parse_probe_url("http://cp.cloudflare.com/generate_204"); + 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"); + 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(""); + 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);