From 060dcd303a7c3658139f0064a2751692ab3c79ec Mon Sep 17 00:00:00 2001 From: Abolfazl Akbarzadeh Date: Tue, 28 Jul 2026 21:30:23 +0330 Subject: [PATCH 1/5] fix(netstack): stop socket and pending-buffer memory growth Close TCP/UDP flows from Drop so DNS lookups and aborted SOCKS clients release smoltcp sockets instead of leaking buffers forever. Cap per-connection TCP pending with deferral/backpressure, free established sockets on TimeWait, and remove UDP entries when the app channel is closed. --- aether/src/netstack.rs | 133 +++++++++++++++++++++++++++++++++++++---- 1 file changed, 120 insertions(+), 13 deletions(-) diff --git a/aether/src/netstack.rs b/aether/src/netstack.rs index 37c7303..4dcf543 100644 --- a/aether/src/netstack.rs +++ b/aether/src/netstack.rs @@ -34,6 +34,10 @@ fn app_queue() -> usize { const MAX_INGEST_PER_TICK: usize = 512; const MAX_RECV_CHUNKS: usize = 128; +fn max_tcp_pending() -> usize { + tcp_buf().saturating_mul(2).max(64 * 1024) +} + type OpenTcpResp = oneshot::Sender>; type OpenUdpResp = oneshot::Sender>; @@ -115,6 +119,7 @@ pub struct TcpConn { pub id: usize, pub from_stack: mpsc::Receiver>, data_in: mpsc::Sender, + split: bool, } impl TcpConn { @@ -129,17 +134,32 @@ impl TcpConn { let _ = self.data_in.send(DataIn::TcpClose(self.id)).await; } - pub fn into_split(self) -> (TcpSender, mpsc::Receiver>) { + pub fn into_split(mut self) -> (TcpSender, mpsc::Receiver>) { + self.split = true; ( TcpSender { id: self.id, - data_in: self.data_in, + data_in: self.data_in.clone(), }, - self.from_stack, + std::mem::replace( + &mut self.from_stack, + { + let (_tx, rx) = mpsc::channel(1); + rx + }, + ), ) } } +impl Drop for TcpConn { + fn drop(&mut self) { + if !self.split { + let _ = self.data_in.try_send(DataIn::TcpClose(self.id)); + } + } +} + pub struct TcpSender { id: usize, data_in: mpsc::Sender, @@ -158,10 +178,17 @@ impl TcpSender { } } +impl Drop for TcpSender { + fn drop(&mut self) { + let _ = self.data_in.try_send(DataIn::TcpClose(self.id)); + } +} + pub struct UdpConn { pub id: usize, pub from_stack: mpsc::Receiver<(SocketAddr, Vec)>, data_in: mpsc::Sender, + split: bool, } impl UdpConn { @@ -176,17 +203,32 @@ impl UdpConn { let _ = self.data_in.send(DataIn::UdpClose(self.id)).await; } - pub fn into_split(self) -> (UdpSender, mpsc::Receiver<(SocketAddr, Vec)>) { + pub fn into_split(mut self) -> (UdpSender, mpsc::Receiver<(SocketAddr, Vec)>) { + self.split = true; ( UdpSender { id: self.id, - data_in: self.data_in, + data_in: self.data_in.clone(), }, - self.from_stack, + std::mem::replace( + &mut self.from_stack, + { + let (_tx, rx) = mpsc::channel(1); + rx + }, + ), ) } } +impl Drop for UdpConn { + fn drop(&mut self) { + if !self.split { + let _ = self.data_in.try_send(DataIn::UdpClose(self.id)); + } + } +} + pub struct UdpSender { id: usize, data_in: mpsc::Sender, @@ -205,6 +247,12 @@ impl UdpSender { } } +impl Drop for UdpSender { + fn drop(&mut self) { + let _ = self.data_in.try_send(DataIn::UdpClose(self.id)); + } +} + #[derive(Clone)] pub struct StackHandle { cmd_tx: mpsc::Sender, @@ -423,6 +471,8 @@ async fn run( mut inbound_rx: mpsc::Receiver>, outbound_tx: mpsc::Sender>, ) -> Result<()> { + let mut deferred: VecDeque = VecDeque::new(); + loop { let now = Instant::now(); let poll_outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { @@ -436,10 +486,22 @@ async fn run( service_udp(&mut s).await; flush_tx(&mut s, &outbound_tx).await; + while let Some(d) = deferred.pop_front() { + if let Some(back) = try_handle_data(&mut s, d) { + deferred.push_front(back); + break; + } + } + let delay = s .iface .poll_delay(Instant::now(), &s.sockets) .map(|d| std::time::Duration::from_micros(d.total_micros())); + let delay = if deferred.is_empty() { + delay + } else { + Some(delay.unwrap_or(std::time::Duration::from_millis(5))) + }; tokio::select! { biased; @@ -467,11 +529,21 @@ async fn run( } } - maybe = data_in_rx.recv() => { + maybe = data_in_rx.recv(), if deferred.is_empty() => { if let Some(d) = maybe { - handle_data(&mut s, d); - while let Ok(d2) = data_in_rx.try_recv() { - handle_data(&mut s, d2); + if let Some(back) = try_handle_data(&mut s, d) { + deferred.push_back(back); + } else { + while deferred.is_empty() { + match data_in_rx.try_recv() { + Ok(d2) => { + if let Some(back) = try_handle_data(&mut s, d2) { + deferred.push_back(back); + } + } + Err(_) => break, + } + } } } } @@ -547,6 +619,7 @@ fn handle_cmd(s: &mut NetStack, cmd: Cmd) { id, from_stack: to_app_rx, data_in: s.data_in_tx.clone(), + split: false, }; let _ = resp.send(Ok(conn)); } @@ -557,28 +630,43 @@ fn handle_cmd(s: &mut NetStack, cmd: Cmd) { } } -fn handle_data(s: &mut NetStack, d: DataIn) { +/// Returns `Some(d)` when the datagram must be deferred (TCP pending full). +fn try_handle_data(s: &mut NetStack, d: DataIn) -> Option { match d { DataIn::Tcp(id, data) => { if let Some(st) = s.tcp_conns.get_mut(&id) { - st.pending.extend_from_slice(&data); + let max = max_tcp_pending(); + if st.pending.len() >= max { + return Some(DataIn::Tcp(id, data)); + } + let space = max - st.pending.len(); + if data.len() <= space { + st.pending.extend_from_slice(&data); + } else { + st.pending.extend_from_slice(&data[..space]); + return Some(DataIn::Tcp(id, data[space..].to_vec())); + } } + None } DataIn::TcpClose(id) => { if let Some(st) = s.tcp_conns.get_mut(&id) { st.half_closed = true; } + None } DataIn::Udp(id, dst, data) => { if let Some(st) = s.udp_conns.get(&id) { let sock = s.sockets.get_mut::(st.handle); let _ = sock.send_slice(&data, to_ip_endpoint(dst)); } + None } DataIn::UdpClose(id) => { if let Some(st) = s.udp_conns.remove(&id) { s.sockets.remove(st.handle); } + None } } } @@ -603,6 +691,7 @@ async fn service_tcp(s: &mut NetStack) { id, from_stack: rx, data_in: data_in_tx.clone(), + split: false, }; let _ = resp.send(Ok(conn)); } @@ -630,6 +719,9 @@ async fn service_tcp(s: &mut NetStack) { let sent = socket.send_slice(&st.pending).unwrap_or(0); if sent > 0 { st.pending.drain(0..sent); + if st.pending.len() * 4 < st.pending.capacity() { + st.pending.shrink_to(max_tcp_pending().min(st.pending.capacity())); + } } } } @@ -673,7 +765,15 @@ async fn service_tcp(s: &mut NetStack) { if matches!(st_state, tcp::State::CloseWait) { s.sockets.get_mut::(handle).close(); } - if matches!(st_state, tcp::State::Closed) && s.tcp_conns[&id].established { + if matches!(st_state, tcp::State::TimeWait) { + if let Some(st) = s.tcp_conns.get_mut(&id) { + st.pending.clear(); + st.pending.shrink_to_fit(); + } + } + if matches!(st_state, tcp::State::Closed | tcp::State::TimeWait) + && s.tcp_conns[&id].established + { s.sockets.remove(handle); s.tcp_conns.remove(&id); } @@ -703,11 +803,18 @@ async fn service_udp(s: &mut NetStack) { } let to_app = s.udp_conns[&id].to_app.clone(); + let mut app_gone = to_app.is_closed(); for p in packets { if to_app.send(p).await.is_err() { + app_gone = true; break; } } + if app_gone { + if let Some(st) = s.udp_conns.remove(&id) { + s.sockets.remove(st.handle); + } + } } } From 62d8185bca8b0da3193ee66ca9e9fca1b84c90a8 Mon Sep 17 00:00:00 2001 From: Abolfazl Akbarzadeh Date: Tue, 28 Jul 2026 21:30:30 +0330 Subject: [PATCH 2/5] fix(wireguard): prevent obfuscation task and alloc storms Send post-handshake junk once after the first encapsulated packet instead of spawning a task on every WriteToNetwork under traffic. Run keepalive junk inline on the timer task, and reuse the encrypt and timer output buffers across iterations to cut RSS churn. --- aether/src/wireguard.rs | 25 +++++++++---------------- 1 file changed, 9 insertions(+), 16 deletions(-) diff --git a/aether/src/wireguard.rs b/aether/src/wireguard.rs index 4380531..0b402fd 100644 --- a/aether/src/wireguard.rs +++ b/aether/src/wireguard.rs @@ -177,9 +177,10 @@ impl WgTunnel { }); let send_task = tokio::spawn(async move { + let mut out_buf = vec![0u8; MAX_PACKET]; + let mut post_hs_junk_sent = false; while let Some(ip_packet) = outbound_rx.recv().await { let mut tunn = tunn_w.lock().await; - let mut out_buf = vec![0u8; MAX_PACKET]; match tunn.encapsulate(&ip_packet, &mut out_buf) { TunnResult::Done => {} @@ -202,12 +203,10 @@ impl WgTunnel { let _ = sock_w.send(&pkt_vec).await; - if aethernoize.jc_after_hs > 0 { - let sock_clone = sock_w.clone(); - let cfg_clone = aethernoize.clone(); - tokio::spawn(async move { - aethernoize::send_post_handshake_junk(&sock_clone, peer, &cfg_clone).await; - }); + // Post-handshake junk once only — not on every data packet. + if aethernoize.jc_after_hs > 0 && !post_hs_junk_sent { + post_hs_junk_sent = true; + aethernoize::send_post_handshake_junk(&sock_w, peer, &aethernoize).await; } } TunnResult::WriteToTunnelV4(_, _) | TunnResult::WriteToTunnelV6(_, _) => {} @@ -217,25 +216,19 @@ impl WgTunnel { let timer_task = tokio::spawn(async move { let mut interval = tokio::time::interval(TIMER_TICK); + let mut tmp = vec![0u8; MAX_PACKET]; loop { interval.tick().await; let mut tunn = tunn_t.lock().await; - let mut tmp = vec![0u8; MAX_PACKET]; if let TunnResult::WriteToNetwork(pkt) = tunn.update_timers(&mut tmp) { let mut pkt_vec = pkt.to_vec(); inject_client_id(&mut pkt_vec, &client_id); drop(tunn); if aethernoize_t.is_enabled() { - let sock_j = sock_t.clone(); - let cfg_j = aethernoize_t.clone(); - tokio::spawn(async move { - aethernoize::send_keepalive_junk(&sock_j, &cfg_j).await; - let _ = sock_j.send(&pkt_vec).await; - }); - } else { - let _ = sock_t.send(&pkt_vec).await; + aethernoize::send_keepalive_junk(&sock_t, &aethernoize_t).await; } + let _ = sock_t.send(&pkt_vec).await; } } }); From f6b57833771893adb9d35cdd0804498271bf92a5 Mon Sep 17 00:00:00 2001 From: Abolfazl Akbarzadeh Date: Tue, 28 Jul 2026 21:30:30 +0330 Subject: [PATCH 3/5] fix(socks): abort client tasks when the proxy server stops Track accepted SOCKS clients in a JoinSet so aborting or dropping the serve task tears down child handlers instead of leaving orphaned relays that keep netstack connections alive across reconnects. --- aether/src/socks.rs | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/aether/src/socks.rs b/aether/src/socks.rs index d263cbe..e21ceb5 100644 --- a/aether/src/socks.rs +++ b/aether/src/socks.rs @@ -27,14 +27,21 @@ pub async fn serve(listen: SocketAddr, stack: StackHandle) -> Result<()> { log::info!("socks5 listening on {listen}"); let bind_ip = listen.ip(); + let mut clients = tokio::task::JoinSet::new(); loop { - let (sock, peer) = listener.accept().await?; - let stack = stack.clone(); - tokio::spawn(async move { - if let Err(e) = handle_client(sock, stack, bind_ip).await { - log::debug!("socks client {peer} ended: {e}"); + tokio::select! { + accept = listener.accept() => { + let (sock, peer) = accept?; + let stack = stack.clone(); + clients.spawn(async move { + if let Err(e) = handle_client(sock, stack, bind_ip).await { + log::debug!("socks client {peer} ended: {e}"); + } + }); } - }); + // Reap finished clients so JoinSet does not grow without bound. + Some(_) = clients.join_next(), if !clients.is_empty() => {} + } } } From 2aadef26331d5526227593aabd2cb8469cdb6032 Mon Sep 17 00:00:00 2001 From: Abolfazl Akbarzadeh Date: Tue, 28 Jul 2026 21:30:30 +0330 Subject: [PATCH 4/5] fix(masque): bound CapsuleParser reassembly buffer Reset the capsule byte buffer if it would exceed 256 KiB so a malformed or stalled stream cannot grow memory without limit. --- aether/src/masque.rs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/aether/src/masque.rs b/aether/src/masque.rs index dcde5bc..f7b58b0 100644 --- a/aether/src/masque.rs +++ b/aether/src/masque.rs @@ -120,12 +120,19 @@ pub struct CapsuleParser { buf: Vec, } +const MAX_CAPSULE_BUF: usize = 256 * 1024; + impl CapsuleParser { pub fn new() -> Self { Self { buf: Vec::new() } } pub fn push(&mut self, data: &[u8]) { + if self.buf.len().saturating_add(data.len()) > MAX_CAPSULE_BUF { + log::warn!("capsule parser buffer exceeded {MAX_CAPSULE_BUF} bytes; resetting"); + self.buf.clear(); + return; + } self.buf.extend_from_slice(data); } From 903af2eb402be000aeac3a0e221452fa8ffe56b1 Mon Sep 17 00:00:00 2001 From: Abolfazl Akbarzadeh Date: Tue, 28 Jul 2026 21:30:30 +0330 Subject: [PATCH 5/5] fix(gool): abort nested tunnels and forwarders on teardown Return AbortOnDrop guards from establish_wg and spawn_udp_forwarder so warp-in-warp reconnect drops outer/inner WireGuard tasks and the UDP forwarder pair instead of leaking them across session loops. --- aether/src/main.rs | 29 +++++++++++++++++++---------- 1 file changed, 19 insertions(+), 10 deletions(-) diff --git a/aether/src/main.rs b/aether/src/main.rs index 4307d39..847e539 100644 --- a/aether/src/main.rs +++ b/aether/src/main.rs @@ -964,7 +964,7 @@ async fn establish_wg( obfuscate: bool, keepalive: u16, label: &'static str, -) -> Result { +) -> Result<(netstack::StackHandle, AbortOnDrop)> { let private_key = identity.private_key_bytes()?; let peer_public = identity.peer_public_key_bytes()?; @@ -1000,19 +1000,27 @@ async fn establish_wg( let tunnel = wireguard::WgTunnel::from_established(session, std::sync::Arc::new(profile), inbound_tx, ipv4); let stack = netstack::spawn(&identity.ipv4, &identity.ipv6, mtu, inbound_rx, outbound_tx)?; - tokio::spawn(async move { + let handle = tokio::spawn(async move { if let Err(e) = tunnel.run(outbound_rx).await { log::error!("[{label}] wireguard tunnel exited: {e}"); } }); - Ok(stack) + Ok((stack, AbortOnDrop(handle))) +} + +struct AbortOnDrop(tokio::task::JoinHandle<()>); + +impl Drop for AbortOnDrop { + fn drop(&mut self) { + self.0.abort(); + } } async fn spawn_udp_forwarder( outer: &netstack::StackHandle, remote: SocketAddr, -) -> Result { +) -> Result<(SocketAddr, AbortOnDrop, AbortOnDrop)> { let sock = std::sync::Arc::new(tokio::net::UdpSocket::bind("127.0.0.1:0").await?); let local = sock.local_addr()?; @@ -1024,7 +1032,7 @@ async fn spawn_udp_forwarder( let up_sock = sock.clone(); let up_peer = inner_peer.clone(); - tokio::spawn(async move { + let up = tokio::spawn(async move { let mut buf = vec![0u8; 65536]; loop { match up_sock.recv_from(&mut buf).await { @@ -1041,7 +1049,7 @@ async fn spawn_udp_forwarder( let down_sock = sock.clone(); let down_peer = inner_peer.clone(); - tokio::spawn(async move { + let down = tokio::spawn(async move { while let Some((_src, data)) = udp_rx.recv().await { let dst = *down_peer.lock().await; if let Some(dst) = dst { @@ -1050,7 +1058,7 @@ async fn spawn_udp_forwarder( } }); - Ok(local) + Ok((local, AbortOnDrop(up), AbortOnDrop(down))) } async fn run_warp_in_warp( @@ -1060,15 +1068,16 @@ async fn run_warp_in_warp( listen: SocketAddr, ) -> Result<()> { log::info!("[*] establishing outer WARP tunnel to {peer}..."); - let outer_stack = establish_wg(&primary, peer, TUNNEL_MTU, true, 5, "outer").await?; + let (outer_stack, _outer_tunnel) = establish_wg(&primary, peer, TUNNEL_MTU, true, 5, "outer").await?; - let forwarder = spawn_udp_forwarder(&outer_stack, peer).await?; + let (forwarder, _fwd_up, _fwd_down) = spawn_udp_forwarder(&outer_stack, peer).await?; log::info!("[+] inner endpoint tunneled through outer warp via {forwarder}"); log::info!("[*] establishing inner WARP tunnel (warp-in-warp)..."); - let inner_stack = establish_wg(&secondary, forwarder, INNER_MTU, false, 20, "inner").await?; + let (inner_stack, _inner_tunnel) = establish_wg(&secondary, forwarder, INNER_MTU, false, 20, "inner").await?; log::info!("[+] socks5 server listening on {listen}"); + // Guards above abort outer/inner tunnels + forwarders when this returns. socks::serve(listen, inner_stack).await }