diff --git a/easytier-core/src/gateway/proxy/icmp_proxy_service.rs b/easytier-core/src/gateway/proxy/icmp_proxy_service.rs index 14f4e48e..d6d4ab46 100644 --- a/easytier-core/src/gateway/proxy/icmp_proxy_service.rs +++ b/easytier-core/src/gateway/proxy/icmp_proxy_service.rs @@ -11,6 +11,7 @@ use tokio::{ sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel}, task::JoinSet, }; +use tokio_util::sync::CancellationToken; use crate::{ packet::ZCPacket, @@ -26,23 +27,6 @@ use super::{ traits::{IcmpProxyRuntime, IcmpProxySocket, ProxyRuntimeError}, }; -async fn start_icmp_runtime( - runtime: &R, - no_tun: bool, -) -> Result>, ProxyRuntimeError> { - match runtime.start_icmp().await { - Ok(socket) => Ok(Some(socket)), - Err(err) => { - runtime.stop_icmp(); - if !no_tun { - return Err(err); - } - tracing::warn!(?err, "start ICMP runtime failed without TUN"); - Ok(None) - } - } -} - pub struct IcmpProxyService { peer_manager: Arc, runtime: Arc, @@ -52,6 +36,8 @@ pub struct IcmpProxyService { pipeline_guard: std::sync::Mutex>, tasks: std::sync::Mutex>, socket: std::sync::Mutex>>, + socket_init: tokio::sync::Mutex<()>, + stopped: CancellationToken, runtime_started: AtomicBool, started: AtomicBool, } @@ -73,6 +59,8 @@ impl IcmpProxyService { pipeline_guard: std::sync::Mutex::new(None), tasks: std::sync::Mutex::new(JoinSet::new()), socket: std::sync::Mutex::new(None), + socket_init: tokio::sync::Mutex::new(()), + stopped: CancellationToken::new(), runtime_started: AtomicBool::new(false), started: AtomicBool::new(false), }) @@ -84,17 +72,6 @@ impl IcmpProxyService { } let snapshot = self.runtime.proxy_runtime_snapshot(); - let socket = match start_icmp_runtime(self.runtime.as_ref(), snapshot.no_tun).await { - Ok(socket) => socket, - Err(err) => { - self.started.store(false, Ordering::Release); - return Err(err); - } - }; - let runtime_started = socket.is_some(); - self.runtime_started - .store(runtime_started, Ordering::Release); - self.socket.lock().unwrap().clone_from(&socket); if let Some(mut response_rx) = self.response_rx.lock().unwrap().take() { let peer_manager = self.peer_manager.clone(); @@ -113,33 +90,6 @@ impl IcmpProxyService { }); } - if let Some(socket) = socket { - let service = Arc::downgrade(self); - self.tasks.lock().unwrap().spawn(async move { - loop { - let recv_result = socket.recv().await; - let (peer_ip, mut packet) = match recv_result { - Ok(packet) => packet, - Err(err) => { - tracing::error!(?err, "receive ICMP packet failed"); - continue; - } - }; - if packet.is_empty() { - tracing::error!("received empty ICMP packet"); - break; - } - let IpAddr::V4(peer_ip) = peer_ip else { - continue; - }; - let Some(service) = service.upgrade() else { - break; - }; - service.handle_socket_response(peer_ip, &mut packet); - } - }); - } - let service = Arc::downgrade(self); self.tasks.lock().unwrap().spawn(async move { loop { @@ -176,9 +126,13 @@ impl IcmpProxyService { } pub fn stop(&self) { + // Serialize shutdown with publication of a newly opened socket and its + // receive task. Opening itself must not hold this synchronous lock. + let mut socket = self.socket.lock().unwrap(); if !self.started.swap(false, Ordering::AcqRel) { return; } + self.stopped.cancel(); if let Some(guard) = self.pipeline_guard.lock().unwrap().take() { guard.close(); } @@ -186,7 +140,61 @@ impl IcmpProxyService { if self.runtime_started.swap(false, Ordering::AcqRel) { self.runtime.stop_icmp(); } - self.socket.lock().unwrap().take(); + socket.take(); + } + + async fn socket(self: &Arc) -> Result, ProxyRuntimeError> { + let _init = self.socket_init.lock().await; + if !self.started.load(Ordering::Acquire) || self.stopped.is_cancelled() { + return Err(std::io::Error::from(std::io::ErrorKind::NotConnected).into()); + } + if let Some(socket) = self.socket.lock().unwrap().clone() { + return Ok(socket); + } + + // Roll back partial resources on errors, caller cancellation or stop. + let rollback = guarden::guard! { self.runtime.stop_icmp(); }; + let socket = tokio::select! { + biased; + _ = self.stopped.cancelled() => { + return Err(std::io::Error::from(std::io::ErrorKind::NotConnected).into()); + } + socket = self.runtime.start_icmp() => socket?, + }; + let mut current = self.socket.lock().unwrap(); + if !self.started.load(Ordering::Acquire) { + return Err(std::io::Error::from(std::io::ErrorKind::NotConnected).into()); + } + + let service = Arc::downgrade(self); + let receiver = socket.clone(); + self.tasks.lock().unwrap().spawn(async move { + loop { + let recv_result = receiver.recv().await; + let (peer_ip, mut packet) = match recv_result { + Ok(packet) => packet, + Err(err) => { + tracing::error!(?err, "receive ICMP packet failed"); + continue; + } + }; + if packet.is_empty() { + tracing::error!("received empty ICMP packet"); + break; + } + let IpAddr::V4(peer_ip) = peer_ip else { + continue; + }; + let Some(service) = service.upgrade() else { + break; + }; + service.handle_socket_response(peer_ip, &mut packet); + } + }); + *current = Some(socket.clone()); + self.runtime_started.store(true, Ordering::Release); + rollback.defuse(); + Ok(socket) } async fn handle_peer_packet(self: Arc, packet: ZCPacket) -> Option { @@ -204,14 +212,13 @@ impl IcmpProxyService { destination, packet: request, } => { - let socket = self.socket.lock().unwrap().clone(); - match socket { - Some(socket) => { + match self.socket().await { + Ok(socket) => { if let Err(err) = socket.send(destination, &request).await { tracing::error!(?err, "send ICMP packet through runtime failed"); } } - None => tracing::error!("send ICMP packet without a runtime socket"), + Err(err) => tracing::error!(?err, "start ICMP runtime failed"), } None } @@ -259,25 +266,59 @@ impl PeerPacketFilter for IcmpProxyServiceFilter< #[cfg(test)] mod tests { - use std::{ - net::{IpAddr, Ipv4Addr}, - sync::{ - Arc, - atomic::{AtomicUsize, Ordering}, - }, - }; + use std::sync::atomic::AtomicUsize; + + use smoltcp::wire::{Icmpv4Message, Icmpv4Packet, IpProtocol, Ipv4Packet}; + use tokio::sync::Semaphore; use super::*; - use crate::gateway::proxy::traits::{IcmpProxySocket, ProxyRuntimeInfo, ProxyRuntimeSnapshot}; + use crate::{ + config::peers::PeerRuntimeSnapshot, + gateway::proxy::{ + cidr_table::{ProxyCidrRule, ProxyCidrSnapshot}, + traits::{ProxyRuntimeInfo, ProxyRuntimeSnapshot}, + }, + packet::PacketType, + peers::peer_manager::PortablePeerManagerConfig, + }; #[derive(Default)] - struct PartialStartRuntime { - stop_count: AtomicUsize, + struct TestSocket { + sent: std::sync::Mutex>, + receives: AtomicUsize, } - impl ProxyRuntimeInfo for PartialStartRuntime { + #[async_trait::async_trait] + impl IcmpProxySocket for TestSocket { + async fn send( + &self, + destination: Ipv4Addr, + _packet: &[u8], + ) -> Result<(), ProxyRuntimeError> { + self.sent.lock().unwrap().push(destination); + Ok(()) + } + + async fn recv(&self) -> Result<(IpAddr, Vec), ProxyRuntimeError> { + self.receives.fetch_add(1, Ordering::AcqRel); + std::future::pending().await + } + } + + #[derive(Default)] + struct TestRuntime { + snapshot: ProxyRuntimeSnapshot, + socket: Arc, + starts: AtomicUsize, + stops: AtomicUsize, + active: AtomicBool, + fail: AtomicBool, + start_gate: Option, + } + + impl ProxyRuntimeInfo for TestRuntime { fn proxy_runtime_snapshot(&self) -> ProxyRuntimeSnapshot { - ProxyRuntimeSnapshot::default() + self.snapshot } fn is_ip_local_virtual_ip(&self, _ip: &IpAddr) -> bool { @@ -285,53 +326,252 @@ mod tests { } } - struct NoopIcmpSocket; - #[async_trait::async_trait] - impl IcmpProxySocket for NoopIcmpSocket { - async fn send( - &self, - _destination: Ipv4Addr, - _packet: &[u8], - ) -> Result<(), ProxyRuntimeError> { - Ok(()) - } - - async fn recv(&self) -> Result<(IpAddr, Vec), ProxyRuntimeError> { - Err(std::io::Error::other("unused test socket").into()) - } - } - - #[async_trait::async_trait] - impl IcmpProxyRuntime for PartialStartRuntime { - type Socket = NoopIcmpSocket; + impl IcmpProxyRuntime for TestRuntime { + type Socket = TestSocket; async fn start_icmp(&self) -> Result, ProxyRuntimeError> { - Err(std::io::Error::other("partial start").into()) + self.starts.fetch_add(1, Ordering::AcqRel); + self.active.store(true, Ordering::Release); + if let Some(gate) = &self.start_gate { + gate.acquire().await.unwrap().forget(); + } + tokio::task::yield_now().await; + if self.fail.load(Ordering::Acquire) { + return Err(std::io::Error::other("partial start").into()); + } + Ok(self.socket.clone()) } fn stop_icmp(&self) { - self.stop_count.fetch_add(1, Ordering::AcqRel); + self.stops.fetch_add(1, Ordering::AcqRel); + self.active.store(false, Ordering::Release); } } - #[tokio::test] - async fn failed_runtime_start_rolls_back_partial_resources() { - let runtime = PartialStartRuntime::default(); + fn service( + runtime: Arc, + table: Arc, + ) -> Arc> { + let (packet_tx, _packet_rx) = crate::host::packet::host_packet_channel(); + let peer_manager = Arc::new( + PeerManagerCore::new_portable_for_test( + PortablePeerManagerConfig::new(PeerRuntimeSnapshot::default().runtime), + packet_tx, + ) + .unwrap(), + ); + IcmpProxyService::new(peer_manager, runtime, table, Duration::from_secs(10)) + } - assert!(start_icmp_runtime(&runtime, false).await.is_err()); - assert_eq!(runtime.stop_count.load(Ordering::Acquire), 1); + fn runtime(no_tun: bool, enable_exit_node: bool) -> Arc { + Arc::new(TestRuntime { + snapshot: ProxyRuntimeSnapshot { + virtual_ipv4: Some("10.0.0.1".parse().unwrap()), + no_tun, + enable_exit_node, + ..Default::default() + }, + ..Default::default() + }) + } + + fn echo_request(destination: &str) -> ZCPacket { + let mut bytes = [0_u8; 28]; + let mut request = Icmpv4Packet::new_unchecked(&mut bytes[20..]); + request.set_msg_type(Icmpv4Message::EchoRequest); + request.set_echo_ident(7); + request.set_echo_seq_no(11); + request.fill_checksum(); + let mut ipv4 = Ipv4Packet::new_unchecked(&mut bytes[..]); + ipv4.set_version(4); + ipv4.set_header_len(20); + ipv4.set_total_len(28); + ipv4.set_hop_limit(64); + ipv4.set_next_header(IpProtocol::Icmp); + ipv4.set_src_addr("10.0.0.2".parse().unwrap()); + ipv4.set_dst_addr(destination.parse().unwrap()); + ipv4.fill_checksum(); + let mut packet = ZCPacket::new_with_payload(&bytes); + packet.fill_peer_manager_hdr(101, 202, PacketType::Data as u8); + packet } #[tokio::test] - async fn no_tun_suppresses_start_error_after_rollback() { - let runtime = PartialStartRuntime::default(); + async fn no_tun_local_echo_does_not_open_socket() { + let runtime = runtime(true, false); + let service = service(runtime.clone(), Arc::new(ProxyCidrTable::new())); + let mut responses = service.response_rx.lock().unwrap().take().unwrap(); + service.start().await.unwrap(); + assert!( + service + .clone() + .handle_peer_packet(echo_request("10.0.0.1")) + .await + .is_none() + ); + let response = responses.try_recv().unwrap(); + let ipv4 = Ipv4Packet::new_checked(response.payload()).unwrap(); + let icmp = Icmpv4Packet::new_checked(ipv4.payload()).unwrap(); + assert_eq!(icmp.msg_type(), Icmpv4Message::EchoReply); + assert_eq!(runtime.starts.load(Ordering::Acquire), 0); + assert_eq!(runtime.socket.receives.load(Ordering::Acquire), 0); + service.stop(); + assert_eq!(runtime.stops.load(Ordering::Acquire), 0); + } - let socket = start_icmp_runtime(&runtime, true).await.unwrap(); - assert!(socket.is_none()); - if socket.is_some() { - runtime.stop_icmp(); - } - assert_eq!(runtime.stop_count.load(Ordering::Acquire), 1); + #[tokio::test] + async fn dynamic_proxy_networks_open_once_and_follow_updated_rules() { + let runtime = runtime(false, false); + let table = Arc::new(ProxyCidrTable::new()); + let service = service(runtime.clone(), table.clone()); + service.start().await.unwrap(); + assert!( + service + .clone() + .handle_peer_packet(echo_request("192.0.2.42")) + .await + .is_some() + ); + assert_eq!(runtime.starts.load(Ordering::Acquire), 0); + + table.update_snapshot(ProxyCidrSnapshot { + rules: vec![ProxyCidrRule { + cidr: "127.0.0.0/24".parse().unwrap(), + mapped_cidr: Some("192.0.2.0/24".parse().unwrap()), + }], + }); + // The first concurrent requests must share both the socket and receiver. + futures::future::join_all((0..16).map(|_| { + service + .clone() + .handle_peer_packet(echo_request("192.0.2.42")) + })) + .await; + tokio::task::yield_now().await; + assert_eq!(runtime.starts.load(Ordering::Acquire), 1); + assert_eq!(runtime.socket.receives.load(Ordering::Acquire), 1); + assert_eq!( + *runtime.socket.sent.lock().unwrap(), + vec!["127.0.0.42".parse::().unwrap(); 16] + ); + + table.update_snapshot(ProxyCidrSnapshot::default()); + assert!( + service + .clone() + .handle_peer_packet(echo_request("192.0.2.42")) + .await + .is_some() + ); + assert_eq!(runtime.socket.sent.lock().unwrap().len(), 16); + // Keep the socket for in-flight replies and subsequent network additions. + assert_eq!(runtime.stops.load(Ordering::Acquire), 0); + table.update_snapshot(ProxyCidrSnapshot { + rules: vec![ProxyCidrRule { + cidr: "192.0.2.0/24".parse().unwrap(), + mapped_cidr: None, + }], + }); + assert!( + service + .clone() + .handle_peer_packet(echo_request("192.0.2.42")) + .await + .is_none() + ); + assert_eq!(runtime.starts.load(Ordering::Acquire), 1); + assert_eq!( + runtime.socket.sent.lock().unwrap().last(), + Some(&"192.0.2.42".parse().unwrap()) + ); + service.stop(); + service.stop(); + assert_eq!(runtime.stops.load(Ordering::Acquire), 1); + assert!(!runtime.active.load(Ordering::Acquire)); + assert!(service.socket().await.is_err()); + } + + #[tokio::test] + async fn exit_node_opens_socket_without_proxy_networks() { + let runtime = runtime(false, true); + let service = service(runtime.clone(), Arc::new(ProxyCidrTable::new())); + service.start().await.unwrap(); + assert_eq!(runtime.starts.load(Ordering::Acquire), 0); + let mut packet = echo_request("192.0.2.42"); + packet + .mut_peer_manager_header() + .unwrap() + .set_exit_node(true); + assert!(service.clone().handle_peer_packet(packet).await.is_none()); + assert_eq!(runtime.starts.load(Ordering::Acquire), 1); + assert_eq!( + *runtime.socket.sent.lock().unwrap(), + ["192.0.2.42".parse::().unwrap()] + ); + } + + #[tokio::test] + async fn failed_runtime_start_rolls_back_and_can_retry() { + let runtime = runtime(true, false); + runtime.fail.store(true, Ordering::Release); + let service = service(runtime.clone(), Arc::new(ProxyCidrTable::new())); + let mut responses = service.response_rx.lock().unwrap().take().unwrap(); + service.start().await.unwrap(); + assert!(service.socket().await.is_err()); + assert_eq!(runtime.stops.load(Ordering::Acquire), 1); + assert!(!runtime.active.load(Ordering::Acquire)); + assert!( + service + .clone() + .handle_peer_packet(echo_request("10.0.0.1")) + .await + .is_none() + ); + assert!(responses.try_recv().is_ok()); + assert_eq!(runtime.starts.load(Ordering::Acquire), 1); + runtime.fail.store(false, Ordering::Release); + assert!(service.socket().await.is_ok()); + service.stop(); + assert_eq!(runtime.stops.load(Ordering::Acquire), 2); + } + + #[tokio::test] + async fn stop_cancels_pending_socket_start() { + let runtime = Arc::new(TestRuntime { + start_gate: Some(Semaphore::new(0)), + ..Default::default() + }); + let service = service(runtime.clone(), Arc::new(ProxyCidrTable::new())); + service.start().await.unwrap(); + let mut opening = Box::pin(service.socket()); + assert!(futures::poll!(&mut opening).is_pending()); + assert!(runtime.active.load(Ordering::Acquire)); + service.stop(); + assert!(opening.await.is_err()); + assert!(!runtime.active.load(Ordering::Acquire)); + assert_eq!(runtime.stops.load(Ordering::Acquire), 1); + assert_eq!(runtime.socket.receives.load(Ordering::Acquire), 0); + assert!(service.socket().await.is_err()); + assert_eq!(runtime.starts.load(Ordering::Acquire), 1); + } + + #[tokio::test] + async fn cancelled_caller_rolls_back_socket_start() { + let runtime = Arc::new(TestRuntime { + start_gate: Some(Semaphore::new(0)), + ..Default::default() + }); + let service = service(runtime.clone(), Arc::new(ProxyCidrTable::new())); + service.start().await.unwrap(); + let mut opening = Box::pin(service.socket()); + assert!(futures::poll!(&mut opening).is_pending()); + drop(opening); + assert!(!runtime.active.load(Ordering::Acquire)); + assert_eq!(runtime.stops.load(Ordering::Acquire), 1); + runtime.start_gate.as_ref().unwrap().add_permits(1); + assert!(service.socket().await.is_ok()); + service.stop(); + assert_eq!(runtime.stops.load(Ordering::Acquire), 2); } } diff --git a/easytier-core/src/gateway/proxy/service.rs b/easytier-core/src/gateway/proxy/service.rs index 126cf7d8..582a7991 100644 --- a/easytier-core/src/gateway/proxy/service.rs +++ b/easytier-core/src/gateway/proxy/service.rs @@ -379,12 +379,12 @@ where return Ok(()); } + // Runtime updates can be cancelled while a host operation is pending. + // Roll back the startup flags and resources so a later update can retry. + let rollback = guarden::guard! { self.stop_started(); }; self.runtime.latch_smoltcp(); self.tcp_started.store(true, Ordering::Release); - if let Err(error) = self.tcp.start(true).await { - self.stop_started(); - return Err(error); - } + self.tcp.start(true).await?; if let Some(icmp) = &self.icmp { self.icmp_started.store(true, Ordering::Release); @@ -398,7 +398,6 @@ where .proxy .icmp_failure_is_fatal { - self.stop_started(); return Err(error); } tracing::warn!(?error, "optional ICMP proxy runtime failed to start"); @@ -407,6 +406,7 @@ where self.udp_started.store(true, Ordering::Release); self.udp.start().await; + rollback.defuse(); Ok(()) } diff --git a/easytier-core/src/host/testkit.rs b/easytier-core/src/host/testkit.rs index 1853dd1c..8f718d4c 100644 --- a/easytier-core/src/host/testkit.rs +++ b/easytier-core/src/host/testkit.rs @@ -133,6 +133,8 @@ pub struct TestHost { Option>, pub tcp_listeners: Option>>, pub reject_tcp_bind: Mutex>, + pub tcp_bind_gate: Option>, + pub tcp_bind_runtime: Mutex>, pub reject_socks5_listener: bool, } @@ -179,6 +181,10 @@ impl VirtualTcpListenerFactory for TestHost { async fn bind_tcp(&self, options: TcpListenOptions) -> anyhow::Result> { self.tcp_binds.fetch_add(1, Ordering::Relaxed); + *self.tcp_bind_runtime.lock().unwrap() = Some(tokio::runtime::Handle::current().id()); + if let Some(gate) = &self.tcp_bind_gate { + gate.acquire().await.unwrap().forget(); + } if self.reject_socks5_listener && options.purpose == TcpListenPurpose::Socks5 { anyhow::bail!("test host rejected SOCKS5 listener"); } diff --git a/easytier-core/src/instance/lifecycle.rs b/easytier-core/src/instance/lifecycle.rs index d90133db..34c3fe04 100644 --- a/easytier-core/src/instance/lifecycle.rs +++ b/easytier-core/src/instance/lifecycle.rs @@ -82,14 +82,23 @@ where } #[cfg(feature = "proxy-packet")] - async fn start_packet_proxy(&self) -> anyhow::Result<()> { + pub(super) async fn start_packet_proxy(&self) -> anyhow::Result<()> { let config = self.runtime_config.snapshot(); let has_proxy_networks = !config.peer.runtime.core.routes.proxy_networks.is_empty(); if config.services.proxy.should_start(has_proxy_networks) { - self.packet_proxy - .start() + if let Some(runtime) = self.runtime_handle() { + let proxy = self.packet_proxy.clone(); + tokio_util::task::AbortOnDropHandle::new( + runtime.spawn(async move { proxy.start().await }), + ) .await - .map_err(anyhow::Error::new)?; + .map_err(|error| anyhow::anyhow!("packet proxy start task failed: {error}"))??; + } else { + self.packet_proxy + .start() + .await + .map_err(anyhow::Error::new)?; + } } Ok(()) } diff --git a/easytier-core/src/instance/mod.rs b/easytier-core/src/instance/mod.rs index f080ef44..8c779863 100644 --- a/easytier-core/src/instance/mod.rs +++ b/easytier-core/src/instance/mod.rs @@ -988,6 +988,12 @@ where if let Some(cleanup) = cleanup { let _ = cleanup.await; } + #[cfg(feature = "proxy-packet")] + if self.state() == CoreInstanceState::Running && self.startup_plan.packet_proxy { + // A running instance may gain its first proxy network here. Start + // from the published configuration, just as during initial startup. + self.start_packet_proxy().await?; + } Ok(()) } diff --git a/easytier-core/src/instance/tests.rs b/easytier-core/src/instance/tests.rs index 49abe15b..3205a1d8 100644 --- a/easytier-core/src/instance/tests.rs +++ b/easytier-core/src/instance/tests.rs @@ -3487,6 +3487,353 @@ virtual_ip = "10.82.0.2/24" instance.stop().await; } + #[cfg(feature = "proxy-packet")] + mod dynamic_packet_proxy { + use std::net::Ipv4Addr; + + use smoltcp::wire::{Icmpv4Message, Icmpv4Packet, IpProtocol, Ipv4Packet}; + + use super::*; + use crate::{ + gateway::proxy::icmp_host::{IcmpProxyHost, IcmpProxySocket, ProxyRuntimeError}, + packet::{PacketType, ZCPacket}, + }; + + struct RecordingIcmpHost { + opens: AtomicUsize, + requests: tokio::sync::mpsc::UnboundedSender, + } + + #[async_trait] + impl IcmpProxyHost for RecordingIcmpHost { + async fn open_icmp_v4( + &self, + _context: SocketContext, + ) -> Result, ProxyRuntimeError> { + self.opens.fetch_add(1, Ordering::Relaxed); + Ok(Arc::new(self.requests.clone())) + } + } + + #[async_trait] + impl IcmpProxySocket for tokio::sync::mpsc::UnboundedSender { + async fn send( + &self, + destination: Ipv4Addr, + _packet: &[u8], + ) -> Result<(), ProxyRuntimeError> { + self.send(destination).unwrap(); + Ok(()) + } + + async fn recv(&self) -> Result<(IpAddr, Vec), ProxyRuntimeError> { + std::future::pending().await + } + } + + fn echo_request(source_peer: u32, destination_peer: u32) -> ZCPacket { + let mut bytes = [0_u8; 28]; + let mut icmp = Icmpv4Packet::new_unchecked(&mut bytes[20..]); + icmp.set_msg_type(Icmpv4Message::EchoRequest); + icmp.set_echo_ident(7); + icmp.set_echo_seq_no(1); + icmp.fill_checksum(); + let mut ipv4 = Ipv4Packet::new_unchecked(&mut bytes[..]); + ipv4.set_version(4); + ipv4.set_header_len(20); + ipv4.set_total_len(28); + ipv4.set_hop_limit(64); + ipv4.set_next_header(IpProtocol::Icmp); + ipv4.set_src_addr("10.0.0.2".parse().unwrap()); + ipv4.set_dst_addr("198.51.100.42".parse().unwrap()); + ipv4.fill_checksum(); + let mut packet = ZCPacket::new_with_payload(&bytes); + packet.fill_peer_manager_hdr(source_peer, destination_peer, PacketType::Data as u8); + packet + } + + #[tokio::test] + async fn first_dynamic_proxy_network_starts_forwarding_and_reuses_resources() { + let mut config = test_config("dynamic-packet-proxy"); + config.peer.snapshot.runtime.core.routes.ipv4 = Some(IpPrefix { + address: "10.0.0.1".parse().unwrap(), + prefix_len: 24, + }); + let mut updated = runtime_snapshot(&config); + let host = Arc::new(TestHost::default()); + let (packet_sink, mut packets) = tokio::sync::mpsc::channel(16); + let (requests, mut destinations) = tokio::sync::mpsc::unbounded_channel(); + let icmp_host = Arc::new(RecordingIcmpHost { + opens: AtomicUsize::new(0), + requests, + }); + let mut adapters = adapters_with_host(host.clone(), None, Arc::new(packet_sink)); + adapters.icmp_proxy_host = Some(icmp_host.clone()); + let instance = CoreInstance::new(config.clone(), adapters).unwrap(); + config + .peer + .snapshot + .runtime + .core + .routes + .ipv4 + .as_mut() + .unwrap() + .address = "10.0.0.2".parse().unwrap(); + let source = build_instance(config).unwrap(); + instance.start().await.unwrap(); + source.start().await.unwrap(); + assert_eq!(host.tcp_binds.load(Ordering::Relaxed), 0); + assert_eq!(icmp_host.opens.load(Ordering::Relaxed), 0); + + let (client, server) = crate::tunnel::ring::create_ring_tunnel_pair(); + let (client_result, server_result) = tokio::join!( + source.peer_manager.add_client_tunnel(client), + instance.peer_manager.add_tunnel_as_server(server), + ); + client_result.unwrap(); + server_result.unwrap(); + + let send = || async { + source + .peer_manager + .send_msg_for_proxy( + echo_request(source.peer_id(), instance.peer_id()), + instance.peer_id(), + ) + .await + .unwrap(); + }; + send().await; + tokio::time::timeout(Duration::from_secs(5), packets.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(icmp_host.opens.load(Ordering::Relaxed), 0); + + Arc::make_mut(&mut updated.peer) + .runtime + .core + .routes + .proxy_networks = vec![proxy_network("192.0.2.0/24", Some("198.51.100.0/24"))]; + instance + .update_runtime_config(updated.clone()) + .await + .unwrap(); + assert_eq!(host.tcp_binds.load(Ordering::Relaxed), 1); + assert_eq!(icmp_host.opens.load(Ordering::Relaxed), 0); + send().await; + assert_eq!( + tokio::time::timeout(Duration::from_secs(5), destinations.recv()) + .await + .unwrap(), + Some("192.0.2.42".parse().unwrap()), + ); + + let mut removed = updated.clone(); + Arc::make_mut(&mut removed.peer) + .runtime + .core + .routes + .proxy_networks + .clear(); + instance.update_runtime_config(removed).await.unwrap(); + send().await; + tokio::time::timeout(Duration::from_secs(5), packets.recv()) + .await + .unwrap() + .unwrap(); + assert!(destinations.try_recv().is_err()); + + instance.update_runtime_config(updated).await.unwrap(); + send().await; + assert_eq!( + tokio::time::timeout(Duration::from_secs(5), destinations.recv()) + .await + .unwrap(), + Some("192.0.2.42".parse().unwrap()), + ); + assert_eq!(host.tcp_binds.load(Ordering::Relaxed), 1); + assert_eq!(icmp_host.opens.load(Ordering::Relaxed), 1); + source.stop().await; + instance.stop().await; + assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 0); + } + + #[tokio::test] + async fn dynamic_proxy_start_respects_instance_state_and_system_forwarding() { + for forward_by_system in [false, true] { + let mut config = test_config("dynamic-proxy-start-policy"); + config.connectivity.runtime.proxy.forward_by_system = forward_by_system; + let mut updated = runtime_snapshot(&config); + let host = Arc::new(TestHost::default()); + let (packet_sink, _packets) = tokio::sync::mpsc::channel(16); + let instance = CoreInstance::new( + config, + adapters_with_host(host.clone(), None, Arc::new(packet_sink)), + ) + .unwrap(); + Arc::make_mut(&mut updated.peer) + .runtime + .core + .routes + .proxy_networks = vec![proxy_network("192.0.2.0/24", None)]; + instance + .update_runtime_config(updated.clone()) + .await + .unwrap(); + assert_eq!(host.tcp_binds.load(Ordering::Relaxed), 0); + instance.start().await.unwrap(); + instance.update_runtime_config(updated).await.unwrap(); + assert_eq!( + host.tcp_binds.load(Ordering::Relaxed), + usize::from(!forward_by_system), + ); + instance.stop().await; + } + } + + #[tokio::test] + async fn dynamic_proxy_start_reports_bind_failure_and_allows_retry() { + let config = test_config("dynamic-proxy-start-failure"); + let mut updated = runtime_snapshot(&config); + let host = Arc::new(TestHost::default()); + *host.reject_tcp_bind.lock().unwrap() = Some("0.0.0.0:20000".parse().unwrap()); + let (packet_sink, _packets) = tokio::sync::mpsc::channel(16); + let instance = CoreInstance::new( + config, + adapters_with_host(host.clone(), None, Arc::new(packet_sink)), + ) + .unwrap(); + instance.start().await.unwrap(); + Arc::make_mut(&mut updated.peer) + .runtime + .core + .routes + .proxy_networks = vec![proxy_network("192.0.2.0/24", None)]; + let error = instance + .update_runtime_config(updated.clone()) + .await + .unwrap_err(); + assert!(error.to_string().contains("test host rejected TCP bind")); + assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 0); + // Configuration publication precedes resource startup. Retrying the + // same update must activate the already published proxy network. + assert_eq!(instance.node_snapshot().await.proxy_networks.len(), 1); + *host.reject_tcp_bind.lock().unwrap() = None; + instance.update_runtime_config(updated).await.unwrap(); + assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 1); + instance.stop().await; + assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 0); + } + + #[tokio::test] + async fn cancelled_dynamic_proxy_start_allows_retry() { + let config = test_config("dynamic-proxy-start-cancelled"); + let mut updated = runtime_snapshot(&config); + let gate = Arc::new(tokio::sync::Semaphore::new(0)); + let host = Arc::new(TestHost { + tcp_bind_gate: Some(gate.clone()), + ..Default::default() + }); + let (packet_sink, _packets) = tokio::sync::mpsc::channel(16); + let instance = CoreInstance::new( + config, + adapters_with_host(host.clone(), None, Arc::new(packet_sink)), + ) + .unwrap(); + instance.start().await.unwrap(); + Arc::make_mut(&mut updated.peer) + .runtime + .core + .routes + .proxy_networks = vec![proxy_network("192.0.2.0/24", None)]; + let update = tokio::spawn({ + let instance = instance.clone(); + let updated = updated.clone(); + async move { instance.update_runtime_config(updated).await } + }); + tokio::time::timeout(Duration::from_secs(5), async { + while host.tcp_binds.load(Ordering::Relaxed) == 0 { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + update.abort(); + assert!(update.await.unwrap_err().is_cancelled()); + assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 0); + + gate.add_permits(1); + instance.update_runtime_config(updated).await.unwrap(); + assert_eq!(host.tcp_binds.load(Ordering::Relaxed), 2); + assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 1); + instance.stop().await; + assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 0); + } + + #[test] + fn dynamic_proxy_start_uses_instance_executor_after_caller_runtime_exits() { + struct ExecutorRuntimeHost(tokio::runtime::Handle); + + #[async_trait] + impl InstanceRuntimeHost for ExecutorRuntimeHost { + fn runtime_handle(&self) -> Option { + Some(self.0.clone()) + } + + async fn prepare( + &self, + _packet_plane: Arc, + ) -> anyhow::Result>> { + Ok(None) + } + + async fn shutdown(&self) {} + } + + let executor = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .unwrap(); + let caller = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let host = Arc::new(TestHost::default()); + let config = test_config("dynamic-proxy-executor"); + let mut updated = runtime_snapshot(&config); + let instance = { + let _runtime = executor.enter(); + let (packet_sink, _packets) = tokio::sync::mpsc::channel(16); + let mut adapters = adapters_with_host(host.clone(), None, Arc::new(packet_sink)); + adapters.instance_runtime = + Arc::new(ExecutorRuntimeHost(executor.handle().clone())); + CoreInstance::new(config, adapters).unwrap() + }; + caller.block_on(async { + instance.start().await.unwrap(); + assert_eq!(host.tcp_binds.load(Ordering::Relaxed), 0); + Arc::make_mut(&mut updated.peer) + .runtime + .core + .routes + .proxy_networks = vec![proxy_network("192.0.2.0/24", None)]; + instance.update_runtime_config(updated).await.unwrap(); + }); + drop(caller); + + assert_eq!( + *host.tcp_bind_runtime.lock().unwrap(), + Some(executor.handle().id()), + ); + assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 1); + executor.block_on(instance.stop()); + assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 0); + } + } + #[cfg(feature = "dhcp-ipv4")] #[tokio::test] async fn runtime_core_accepts_explicit_dhcp_runtime_snapshot() {