diff --git a/docs/src/content/docs/config/tun.md b/docs/src/content/docs/config/tun.md index 69374615..826bd463 100644 --- a/docs/src/content/docs/config/tun.md +++ b/docs/src/content/docs/config/tun.md @@ -228,13 +228,38 @@ mark-aware reverse-path filtering. No universal VPN, firewall, resolver-manager, or distribution compatibility is promised. -Reload retains the interface, capture, and redirect listener identities. Network -topology changes require restart and are rejected before retiring live workers. +Reload retains the interface, capture, and redirect listener identities. Managed +TUN topology configuration changes require restart and are rejected before retiring live workers. Ordinary worker reloads can reset existing connections and briefly interrupt traffic, but do not disable capture. Failed candidates roll back on the same network lease. Unexpected owned routing changes or a missing private firewall table cause orderly shutdown. Resolver-setting changes do not. +After startup, changes to the underlying main-table default path, gateway, or +explicit preferred source, and loss of the selected source address, trigger recovery +in place. Ordinary IPv6 privacy-address rotation does not reset existing flows. +Both IP families are monitored. A +complete outage pauses forwarding until an underlay returns; temporary recovery +failures are retried without removing capture. Recovery rebuilds DNS and proxy +transports from the accepted in-memory configuration, not from edited files. +Explicit reloads wait while offline and can replace a configuration that prevents +workers from restarting once an underlay is available. +The TUN, REDIRECT ports, fake-IP mappings, and explicit selector choices remain +owned and retained. WireGuard peer hostnames are refreshed best-effort; a failed +refresh retains the previous endpoints rather than blocking unrelated workers. Old flows, +pending connections, and connections draining from earlier reloads are cancelled. +Applications must reconnect; requests are not replayed and existing TCP/QUIC +sessions are not migrated. TUN TCP resets are best-effort. + +Detection uses netlink notifications and periodic checks, with a short settling +delay. It observes kernel route/source selection for representative off-link +destinations, not Internet reachability or every destination-specific policy. +Packet loss, captive portals, or a Wi-Fi change that preserves the observed +routing and carrier state may not trigger recovery. Explicit `bind_interface` +and listen-address settings remain fixed; they are never silently moved to another +interface. A reset affects all proxy workers in this process, including local +listeners. + shoes exclusively owns its token-named `inet` nftables table, including all of its contents. Do not add rules, chains, or sets to that table: external changes inside it are neither audited nor preserved. Cleanup removes the exact owned table even diff --git a/docs/src/content/docs/zh-cn/config/tun.md b/docs/src/content/docs/zh-cn/config/tun.md index cd895751..151a34f0 100644 --- a/docs/src/content/docs/zh-cn/config/tun.md +++ b/docs/src/content/docs/zh-cn/config/tun.md @@ -127,7 +127,11 @@ TCP REDIRECT 要求每个加速地址族都有可用的环回网络。如果启 冲突的表、优先级（包括固定的后备优先级 `32768`）、标记保留或前面的非本地策略规则会导致启动拒绝。其他 VPN 和防火墙管理器需要明确的共存验证； shoes 不会覆盖外部防火墙的决定。 REDIRECT 还拒绝使用其虚拟对等点作为网关的本机路由和不透明的下一跳对象路由。捕获的回复在标记感知反向路径过滤之前携带捕获标记。不承诺通用 VPN、防火墙、解析器管理器或分发兼容性。 -重新加载会保留接口、捕获和重定向监听器身份。网络拓扑更改需要重新启动，并且在退休工作人员之前会被拒绝。普通工作线程重新加载可以重置现有连接并短暂中断流量，但不会禁用捕获。失败的候选者将回滚同一网络租约。意外的所属路由更改或丢失的专用防火墙表会导致有序关闭。解析器设置更改则不会。 +重新加载会保留接口、捕获和重定向监听器身份。托管 TUN 拓扑配置更改需要重新启动，并且在停止现有工作线程之前会被拒绝。普通工作线程重新加载可以重置现有连接并短暂中断流量，但不会禁用捕获。失败的候选者将回滚同一网络租约。意外的所属路由更改或丢失的专用防火墙表会导致有序关闭。解析器设置更改则不会。 + +启动后，底层主路由表的默认路径、网关或显式首选源地址发生变化，或当前选用的源地址被删除时，shoes 会在进程内恢复连接。普通 IPv6 隐私地址轮换不会重置现有连接。IPv4 和 IPv6 分别受到监视。全部底层连接中断时暂停转发，连接恢复后重新启动工作线程；临时恢复失败会自动重试，不会移除捕获规则。恢复使用已接受的内存配置重建 DNS 和代理传输，不会重新读取已编辑的配置文件。显式重载在离线时等待，底层网络可用后可替换导致工作线程无法重启的配置。TUN、REDIRECT 端口、fake-IP 映射和显式选择的代理保持不变。WireGuard 对等端主机名会尽力重新解析；失败时保留原有端点，不阻塞其他工作线程。旧连接、正在建立的连接，以及此前重新加载后仍在排空的连接都会被取消。应用程序必须重新连接；shoes 不会重放请求，也不会迁移现有 TCP/QUIC 会话。TUN TCP 重置为尽力而为。 + +检测使用 netlink 通知和周期检查，并等待短暂的稳定时间。它观察内核对代表性非直连目标的路由和源地址选择，不检测互联网可达性，也不覆盖所有按目标配置的策略路由。丢包、认证门户或保持相同路由和载波状态的 Wi-Fi 切换可能不会触发恢复。显式 `bind_interface` 和监听地址始终保持固定，不会自动移到其他接口。重置影响本进程中的所有代理工作线程，包括本地监听器。 shoes 独家拥有其名为 `inet` nftables 表的代币，包括其所有内容。不要向该表添加规则、链或集：其中的外部更改既不会被审核也不会被保留。清理会删除确切拥有的表，即使其内容已更改；其他表保持不变。连接网络集通过 nftables 自动更新并缓存在内存中，而不是记录为防火墙快照。 diff --git a/src/anytls/anytls_client_pool.rs b/src/anytls/anytls_client_pool.rs index 72e9f883..0a4aaaf6 100644 --- a/src/anytls/anytls_client_pool.rs +++ b/src/anytls/anytls_client_pool.rs @@ -80,10 +80,10 @@ impl SessionPool { ) { // Each checkout opens exactly one stream; only its idempotent terminal // transition may release it. Idle sessions have no active streams. - let Ok(runtime) = tokio::runtime::Handle::try_current() else { + if tokio::runtime::Handle::try_current().is_err() { session.close(); return; - }; + } if self.options.disable_reuse || !session.is_reusable() { session.close(); return; @@ -98,17 +98,16 @@ impl SessionPool { let weak = Arc::downgrade(self); let check_interval = interval(self.options.idle_session_check_interval_secs); state.reaper = Some( - runtime - .spawn(async move { - loop { - tokio::time::sleep(check_interval).await; - let Some(pool) = weak.upgrade() else { return }; - if !pool.reap() { - return; - } + crate::network_tasks::spawn(async move { + loop { + tokio::time::sleep(check_interval).await; + let Some(pool) = weak.upgrade() else { return }; + if !pool.reap() { + return; } - }) - .abort_handle(), + } + }) + .abort_handle(), ); } } diff --git a/src/anytls/anytls_client_session.rs b/src/anytls/anytls_client_session.rs index 4721b764..b1f5e2f6 100644 --- a/src/anytls/anytls_client_session.rs +++ b/src/anytls/anytls_client_session.rs @@ -327,7 +327,7 @@ impl ClientSessionState { W: tokio::io::AsyncWrite + Send + Unpin + 'static, { let writer_state = Arc::clone(&session); - let writer_task = tokio::spawn(async move { + let writer_task = crate::network_tasks::spawn(async move { tokio::select! { result = Self::writer_loop( Arc::downgrade(&writer_state), writer, outgoing_rx, @@ -341,8 +341,8 @@ impl ClientSessionState { } writer_state.close(); }); - let watchdog_task = tokio::spawn(Self::watchdog_loop(Arc::clone(&session))); - let reader_task = tokio::spawn(async move { + let watchdog_task = crate::network_tasks::spawn(Self::watchdog_loop(Arc::clone(&session))); + let reader_task = crate::network_tasks::spawn(async move { tokio::select! { result = Self::reader_loop( Arc::downgrade(&session), reader, session.close_notify.clone(), diff --git a/src/anytls/anytls_server_session.rs b/src/anytls/anytls_server_session.rs index 392bafc5..319090a0 100644 --- a/src/anytls/anytls_server_session.rs +++ b/src/anytls/anytls_server_session.rs @@ -439,7 +439,7 @@ impl AnyTlsSession { if let Some(stream) = stream_opt { let session = Arc::clone(self); - let handle = tokio::spawn(async move { + let handle = crate::network_tasks::spawn(async move { let _permit = permit; if let Err(e) = session.handle_new_stream(stream).await { log::debug!("AnyTLS stream {} error: {}", stream_id, e); diff --git a/src/control/probes.rs b/src/control/probes.rs index 9c71fc47..dd99f65c 100644 --- a/src/control/probes.rs +++ b/src/control/probes.rs @@ -42,11 +42,16 @@ struct FullBudgetAttempts { } pub struct ProbeState { - connectors: Mutex>>, + transports: Mutex>, history: Mutex>, full_budget_attempts: Mutex>, +} + +struct ProbeTransports { + connectors: Mutex>>, resolver: Arc, forbidden: Vec, + network: Option, } impl ProbeState { @@ -61,14 +66,27 @@ impl ProbeState { pub fn new(resolver: Arc, forbidden: Vec) -> Self { Self { - connectors: Mutex::new(HashMap::new()), + transports: Mutex::new(Arc::new(ProbeTransports { + connectors: Mutex::new(HashMap::new()), + resolver, + forbidden, + network: crate::network_tasks::NetworkTasks::current(), + })), history: Mutex::new(HashMap::new()), full_budget_attempts: Mutex::new(HashMap::new()), - resolver, - forbidden, } } + pub(crate) fn reset_network( + &self, + resolver: Arc, + forbidden: Vec, + ) { + let replacement = Self::new(resolver, forbidden); + *self.transports.lock() = replacement.transports.into_inner(); + self.full_budget_attempts.lock().clear(); + } + pub fn decorate(&self, name: &str, object: &mut Value) { let histories = self.history.lock(); let Some(history) = histories.get(name) else { @@ -166,7 +184,9 @@ impl ProbeState { } urls.insert(query.url.to_string(), tokio::time::Instant::now()); } +} +impl ProbeTransports { fn connector(&self, catalog: &Catalog, name: &str) -> io::Result> { if let Some(connector) = self.connectors.lock().get(name) { return Ok(connector.clone()); @@ -669,12 +689,29 @@ async fn request_with_tls_config( query: &ProbeQuery, tls_config: impl FnOnce() -> rustls::ClientConfig, ) -> io::Result<(u16, bool)> { - let resolver = ProbeResolver::for_query( - generation.probes.resolver.clone(), - &generation.probes.forbidden, - query, - )?; - let chain = generation.probes.connector(&generation.catalog, name)?; + let transports = generation.probes.transports.lock().clone(); + let request = + request_with_transports(&transports, &generation.catalog, name, query, tls_config); + match &transports.network { + Some(network) => tokio::select! { + biased; + _ = network.cancelled.cancelled() => Err(io::Error::new(io::ErrorKind::Interrupted, "probe network retired")), + result = network.scope(request) => result, + }, + None => request.await, + } +} + +async fn request_with_transports( + transports: &ProbeTransports, + catalog: &Catalog, + name: &str, + query: &ProbeQuery, + tls_config: impl FnOnce() -> rustls::ClientConfig, +) -> io::Result<(u16, bool)> { + let resolver = + ProbeResolver::for_query(transports.resolver.clone(), &transports.forbidden, query)?; + let chain = transports.connector(catalog, name)?; let started = Instant::now(); let host = query .url diff --git a/src/control/probes/tests.rs b/src/control/probes/tests.rs index 6dd2ea72..b1758361 100644 --- a/src/control/probes/tests.rs +++ b/src/control/probes/tests.rs @@ -12,6 +12,109 @@ async fn publish_for_api_probes(runtime: &ControlRuntime, generation: &Arc = Arc::new(PendingAnswers::default()); + let probes = old_network + .scope(async { ProbeState::new(old_resolver.clone(), vec![]) }) + .await; + probes.record("DIRECT", "http://example.test/", 12, true); + let old = probes.transports.lock().clone(); + let catalog = crate::config::create_server_configs( + serde_json::from_value(json!([ + {"client_group":"chosen", "type":"selector", "client_chains":[ + {"chain":[{"name":"direct", "protocol":{"type":"direct"}}]} + ]} + ])) + .unwrap(), + ) + .unwrap() + .catalog + .unwrap(); + let connector = old.connector(&catalog, "DIRECT").unwrap(); + old_network.cancel(); + old_network.wait().await; + let new_network = crate::network_tasks::NetworkLifetime::default(); + let new_resolver: Arc = Arc::new(PendingAnswers::default()); + new_network + .scope(async { + probes.reset_network(new_resolver.clone(), vec!["198.18.0.0/15".parse().unwrap()]); + }) + .await; + let current = probes.transports.lock().clone(); + assert!(!Arc::ptr_eq(&old, ¤t)); + assert!(Arc::ptr_eq(¤t.resolver, &new_resolver)); + assert!(old.network.as_ref().unwrap().cancelled.is_cancelled()); + assert!(!current.network.as_ref().unwrap().cancelled.is_cancelled()); + assert!(current.connectors.lock().is_empty()); + assert!(Arc::ptr_eq( + &connector, + &old.connector(&catalog, "DIRECT").unwrap() + )); + assert!(current.connectors.lock().is_empty()); + assert!(!Arc::ptr_eq( + &connector, + ¤t.connector(&catalog, "DIRECT").unwrap() + )); + assert_eq!(current.forbidden, ["198.18.0.0/15".parse().unwrap()]); + let mut object = json!({}); + probes.decorate("DIRECT", &mut object); + assert_eq!(object["history"][0]["delay"], 12); +} + +#[tokio::test] +async fn network_reset_cancels_a_connected_stalled_probe() { + tokio::time::timeout(Duration::from_secs(3), async { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let network = crate::network_tasks::NetworkLifetime::default(); + let catalog = crate::config::create_server_configs( + serde_json::from_value(json!([ + {"client_group":"path", "type":"selector", "client_chains":[{"chain":[ + {"name":"proxy", "address":listener.local_addr().unwrap().to_string(), + "protocol":{"type":"http"}} + ]}]} + ])) + .unwrap(), + ) + .unwrap() + .catalog + .unwrap(); + let generation = network + .scope(async { + generation( + catalog, + Arc::new(Answers(vec!["192.0.2.1:80".parse().unwrap()])), + ) + }) + .await; + let query = query(); + let request = request(&generation, "proxy", &query); + tokio::pin!(request); + let peer = async { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut header = Vec::new(); + while !header.ends_with(b"\r\n\r\n") { + header.push(stream.read_u8().await.unwrap()); + } + network.cancel(); + stream + }; + let stream = tokio::select! { + result = &mut request => panic!("probe completed before reset: {result:?}"), + stream = peer => stream, + }; + assert_eq!( + request.await.unwrap_err().kind(), + io::ErrorKind::Interrupted + ); + network.wait().await; + drop(stream); + }) + .await + .unwrap(); +} + #[tokio::test] async fn automatic_probes_use_declared_chains_not_callers_and_keep_group_settings_separate() { tokio::time::timeout(Duration::from_secs(5), async { @@ -1086,7 +1189,7 @@ async fn probe_wrapped_errors(h2_proxy: bool) { proxy: Box::new( crate::tcp::proxy_connector_impl::ProxyConnectorImpl::from_config( config, - generation.probes.resolver.clone(), + generation.probes.transports.lock().resolver.clone(), ) .unwrap() .unwrap(), @@ -1096,10 +1199,16 @@ async fn probe_wrapped_errors(h2_proxy: bool) { InitialHopEntry::Direct(socket) }; let chain = ClientProxyChain::new(vec![entry], vec![]); - generation.probes.connectors.lock().insert( - "member-0".into(), - Arc::new(ClientChainGroup::new(vec![chain])), - ); + generation + .probes + .transports + .lock() + .connectors + .lock() + .insert( + "member-0".into(), + Arc::new(ClientChainGroup::new(vec![chain])), + ); generation .catalog .selection @@ -2274,8 +2383,8 @@ async fn review_forbidden_literals_do_not_construct_or_cache_probe_transports() )), ); let mut generation = generation(catalog, Arc::new(Answers(Vec::new()))); - generation - .probes + Arc::get_mut(generation.probes.transports.get_mut()) + .unwrap() .forbidden .push("100.64.0.0/10".parse().unwrap()); for url in [ @@ -2288,7 +2397,16 @@ async fn review_forbidden_literals_do_not_construct_or_cache_probe_transports() let error = request(&generation, "quic", &query).await.unwrap_err(); assert_eq!(error.kind(), io::ErrorKind::PermissionDenied); assert!(local_failure(&error)); - assert!(generation.probes.connectors.lock().is_empty(), "{url}"); + assert!( + generation + .probes + .transports + .lock() + .connectors + .lock() + .is_empty(), + "{url}" + ); let error = request(&generation, "missing", &query).await.unwrap_err(); assert_eq!( error.kind(), @@ -2319,6 +2437,8 @@ async fn cached_connectors_reject_forbidden_hostname_candidates() { for name in ["DIRECT", "http"] { let cached = generation .probes + .transports + .lock() .connector(&generation.catalog, name) .unwrap(); for _ in 0..2 { @@ -2334,6 +2454,8 @@ async fn cached_connectors_reject_forbidden_hostname_candidates() { &cached, &generation .probes + .transports + .lock() .connector(&generation.catalog, name) .unwrap() )); @@ -2364,6 +2486,8 @@ async fn carrier_probe_checks_http_hostname_before_connecting() { ); let cached = generation .probes + .transports + .lock() .connector(&generation.catalog, "outbound") .unwrap(); for _ in 0..2 { @@ -2382,6 +2506,8 @@ async fn carrier_probe_checks_http_hostname_before_connecting() { &cached, &generation .probes + .transports + .lock() .connector(&generation.catalog, "outbound") .unwrap() )); diff --git a/src/desktop/linux/ingress.rs b/src/desktop/linux/ingress.rs index 9f0d2be0..2c020537 100644 --- a/src/desktop/linux/ingress.rs +++ b/src/desktop/linux/ingress.rs @@ -140,7 +140,7 @@ impl Ingress { .listeners .iter() .map(|listener| { - tokio::spawn(serve( + crate::network_tasks::spawn(serve( listener.clone(), selector.clone(), resolver.clone(), @@ -230,7 +230,7 @@ async fn serve( }; let selector = selector.clone(); let resolver = resolver.clone(); - connections.spawn(async move { + crate::network_tasks::spawn_in(&mut connections, async move { let _permit = permit; let _ = crate::tun::handle_tcp_connection( stream, diff --git a/src/desktop/linux/mod.rs b/src/desktop/linux/mod.rs index 90772e0f..4f5ed478 100644 --- a/src/desktop/linux/mod.rs +++ b/src/desktop/linux/mod.rs @@ -7,6 +7,7 @@ mod routes; pub(crate) mod spec; #[cfg(test)] mod tests; +mod underlay; use std::io; use std::os::fd::BorrowedFd; @@ -37,7 +38,9 @@ pub(crate) struct Session { _connection: AbortOnDropHandle<()>, _events: AbortOnDropHandle<()>, changes: tokio::sync::mpsc::Receiver<()>, + underlay: Arc, reconcile_after: tokio::time::Instant, + poll_at: tokio::time::Instant, } impl Session { @@ -56,8 +59,15 @@ impl Session { .map_err(io::Error::other)?; let connection = AbortOnDropHandle::new(tokio::spawn(connection)); let (changed, changes) = tokio::sync::mpsc::channel(1); + let underlay = Arc::new(underlay::Monitor::default()); + let monitor = underlay.clone(); let events = AbortOnDropHandle::new(tokio::spawn(async move { - while events.next().await.is_some() { + while let Some((message, _)) = events.next().await { + if let rtnetlink::packet_core::NetlinkPayload::InnerMessage(message) = + message.payload + { + monitor.event(&message); + } let _ = changed.try_send(()); } })); @@ -119,7 +129,9 @@ impl Session { _connection: connection, _events: events, changes, + underlay, reconcile_after: tokio::time::Instant::now(), + poll_at: tokio::time::Instant::now() + Duration::from_secs(10), }) } @@ -170,6 +182,10 @@ impl Session { self.ingress.clone() } + pub(crate) fn spec(&self) -> &Spec { + &self.record.spec + } + fn firewall_policy(&self, active: bool) -> serde_json::Value { nft::install( &self.record.spec, @@ -181,15 +197,6 @@ impl Session { ) } - pub(crate) fn validate(&self, configs: &[crate::config::ResolvedConfig]) -> io::Result<()> { - if Spec::from_resolved_configs(configs)?.as_ref() != Some(&self.record.spec) { - return Err(spec::invalid( - "managed TUN topology changes require a restart", - )); - } - Ok(()) - } - pub(crate) async fn activate(&mut self) -> io::Result<()> { routes::check( &self.handle, @@ -221,20 +228,39 @@ impl Session { } } self.active = true; + self.underlay.initialize( + underlay::snapshot( + &self.handle, + self.record.ifindex.unwrap(), + self.record.spec.linux.bypass_mark, + ) + .await?, + ); Ok(()) } pub(crate) async fn changed(&mut self) { tokio::time::sleep_until(self.reconcile_after).await; tokio::select! { - _ = self.changes.recv() => {}, - _ = tokio::time::sleep(Duration::from_secs(10)) => {}, + _ = self.changes.recv() => { + self.poll_at = tokio::time::Instant::now() + Duration::from_millis(250); + tokio::time::sleep_until(self.poll_at).await; + }, + _ = tokio::time::sleep_until(self.poll_at) => {}, } } - pub(crate) async fn reconcile(&mut self) -> io::Result<()> { + pub(crate) fn available(&self) -> bool { + self.underlay.available() + } + + pub(crate) fn refresh_bootstrap(&self) -> io::Result<()> { + self._policy.refresh_bootstrap() + } + + pub(crate) async fn reconcile(&mut self) -> io::Result { if !self.active { - return Ok(()); + return Ok(false); } while self.changes.try_recv().is_ok() {} let index = self.record.ifindex.unwrap(); @@ -259,7 +285,10 @@ impl Session { self.on_link = networks; } self.reconcile_after = tokio::time::Instant::now() + Duration::from_secs(1); - Ok(()) + self.poll_at = tokio::time::Instant::now() + Duration::from_secs(10); + Ok(self.underlay.update( + underlay::snapshot(&self.handle, index, self.record.spec.linux.bypass_mark).await?, + )) } pub(crate) async fn deactivate(&mut self) -> io::Result<()> { diff --git a/src/desktop/linux/routes.rs b/src/desktop/linux/routes.rs index 4e9aa694..1143beda 100644 --- a/src/desktop/linux/routes.rs +++ b/src/desktop/linux/routes.rs @@ -71,7 +71,7 @@ pub(super) fn owned_link(link: &LinkMessage, index: u32, name: &str, alias: &str .contains(&LinkAttribute::IfAlias(alias.to_owned())) } -fn table(route: &RouteMessage) -> u32 { +pub(super) fn table(route: &RouteMessage) -> u32 { route .attributes .iter() @@ -82,7 +82,7 @@ fn table(route: &RouteMessage) -> u32 { .unwrap_or(u32::from(route.header.table)) } -async fn dump(handle: &Handle, ipv6: bool) -> io::Result> { +pub(super) async fn dump(handle: &Handle, ipv6: bool) -> io::Result> { let family = if ipv6 { AddressFamily::Inet6 } else { @@ -231,6 +231,20 @@ pub(super) async fn preflight(handle: &Handle, spec: &Spec) -> io::Result<()> { Ok(()) } +pub(super) fn native_default(route: &RouteMessage) -> bool { + table(route) == 254 + && route.header.kind == RouteType::Unicast + && route.header.destination_prefix_length == 0 + && route.header.source_prefix_length == 0 + && route.header.tos == 0 + && !route.attributes.iter().any(|attribute| { + matches!( + attribute, + RouteAttribute::Source(_) | RouteAttribute::Iif(_) | RouteAttribute::Mark(_) + ) + }) +} + pub(super) fn check_mark_reservation(rule: &RuleMessage, spec: &Spec) -> io::Result<()> { let mark = rule .attributes diff --git a/src/desktop/linux/tests.rs b/src/desktop/linux/tests.rs index d1f9a9f5..044ed5dc 100644 --- a/src/desktop/linux/tests.rs +++ b/src/desktop/linux/tests.rs @@ -1,4 +1,5 @@ use super::*; +mod reconnect; use serde_json::json; mod fallback; @@ -430,6 +431,17 @@ fn managed_kernel_capture_and_cleanup() -> io::Result<()> { const PARENT: &str = "SHOES_MANAGED_PARENT_NETNS"; const MOUNT: &str = "SHOES_MANAGED_PARENT_MOUNTNS"; let namespace = std::fs::read_link("/proc/self/ns/net")?; + if std::env::var_os("SHOES_MANAGED_PEER_ECHO").is_some() { + assert_ne!(namespace.as_os_str(), std::env::var_os(PARENT).unwrap()); + assert_ne!( + std::fs::read_link("/proc/self/ns/mnt")?.as_os_str(), + std::env::var_os(MOUNT).unwrap() + ); + return tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()? + .block_on(reconnect::echo()); + } if std::env::var_os("SHOES_ORPHAN_HELPER").is_some() { assert_ne!(namespace.as_os_str(), std::env::var_os(PARENT).unwrap()); return tokio::runtime::Builder::new_current_thread().enable_all().build()? @@ -586,7 +598,7 @@ fn managed_kernel_capture_and_cleanup() -> io::Result<()> { } let mut daemons = resolved_fixture().await?; let result = - tokio::time::timeout(Duration::from_secs(90), kernel_capture()).await?; + tokio::time::timeout(Duration::from_secs(180), kernel_capture()).await?; for child in &mut daemons { let _ = child.kill().await; let _ = child.wait().await; @@ -620,7 +632,7 @@ fn managed_kernel_capture_and_cleanup() -> io::Result<()> { "--", "timeout", "--kill-after=5s", - "120s", + "210s", "unshare", "--net", "--mount", @@ -1032,11 +1044,15 @@ async fn kernel_capture() -> io::Result<()> { } session.reconcile().await?; let index = session.record.ifindex; - session.validate(&[crate::config::ResolvedConfig::TunServer(config.clone())])?; + let topology = crate::desktop::NetworkTopology { + spec: Some(session.spec().clone()), + ingress: session.ingress(), + }; + topology.validate(&[crate::config::ResolvedConfig::TunServer(config.clone())])?; let mut changed = config.clone(); changed.mtu += 1; assert!( - session + topology .validate(&[crate::config::ResolvedConfig::TunServer(changed)]) .is_err() ); @@ -1089,6 +1105,9 @@ async fn kernel_capture() -> io::Result<()> { "managed process recovery (redirect={redirect}): {error}" )) })?; + reconnect::handover(redirect).await.map_err(|error| { + io::Error::other(format!("handover (redirect={redirect}): {error}")) + })?; } orphan_helper_recovery().await?; preflight_failures().await?; diff --git a/src/desktop/linux/tests/reconnect.rs b/src/desktop/linux/tests/reconnect.rs new file mode 100644 index 00000000..a5889043 --- /dev/null +++ b/src/desktop/linux/tests/reconnect.rs @@ -0,0 +1,386 @@ +use super::*; +use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt}; + +pub(super) async fn echo() -> io::Result<()> { + let mut tasks = tokio::task::JoinSet::new(); + for address in ["198.51.100.2:19123", "[fd00:2::2]:19123"] { + let tcp = tokio::net::TcpListener::bind(address).await?; + let udp = tokio::net::UdpSocket::bind(address).await?; + tasks.spawn(async move { + let mut flows = tokio::task::JoinSet::new(); + loop { + let (mut stream, peer) = tokio::select! { + result = tcp.accept() => result.unwrap(), + _ = flows.join_next(), if !flows.is_empty() => continue, + }; + flows.spawn(async move { + let mut packet = [0; 4]; + while stream.read_exact(&mut packet).await.is_ok() { + if &packet != b"ping" { + break; + } + if stream + .write_all(format!("{}\n", peer.ip()).as_bytes()) + .await + .is_err() + { + break; + } + } + }); + } + }); + tasks.spawn(async move { + let mut packet = [0; 4]; + loop { + let (_, peer) = udp.recv_from(&mut packet).await.unwrap(); + udp.send_to(peer.ip().to_string().as_bytes(), peer) + .await + .unwrap(); + } + }); + } + println!("Peer echo ready"); + tasks.join_next().await; + Err(io::Error::other("echo service stopped")) +} + +async fn line( + reader: &mut (impl tokio::io::AsyncBufRead + Unpin), + expected: &str, +) -> io::Result<()> { + tokio::time::timeout(Duration::from_secs(15), async { + let mut line = String::new(); + loop { + line.clear(); + if reader.read_line(&mut line).await? == 0 { + return Err(io::Error::other(format!("child exited before {expected}"))); + } + if line.contains(expected) { + return Ok(()); + } + } + }) + .await? +} + +async fn payload(sources: [&str; 2]) -> io::Result<()> { + for (target, expected) in ["198.51.100.2:19123", "[fd00:2::2]:19123"] + .into_iter() + .zip(sources) + { + let mut stream = tokio::net::TcpStream::connect(target).await?; + stream.write_all(b"ping").await?; + let mut reader = tokio::io::BufReader::new(stream); + let mut reply = String::new(); + tokio::time::timeout(Duration::from_secs(3), reader.read_line(&mut reply)).await??; + assert_eq!(reply.trim(), expected); + let udp = tokio::net::UdpSocket::bind(if target.starts_with('[') { + "[::]:0" + } else { + "0.0.0.0:0" + }) + .await?; + udp.send_to(b"ping", target).await?; + let mut bytes = [0; 128]; + let length = tokio::time::timeout(Duration::from_secs(3), udp.recv(&mut bytes)).await??; + assert_eq!(std::str::from_utf8(&bytes[..length]).unwrap(), expected); + } + Ok(()) +} + +fn identity() -> io::Result { + let links: serde_json::Value = + serde_json::from_str(&run("ip", &["-j", "link", "show", "shoes0"])?)?; + let tables: serde_json::Value = serde_json::from_str(&run("nft", &["-j", "list", "tables"])?)?; + Ok(json!({"index":links[0]["ifindex"], "alias":links[0]["ifalias"], "tables":tables})) +} + +pub(super) async fn handover(redirect: bool) -> io::Result<()> { + run( + "ip", + &[ + "link", "add", "native1", "type", "veth", "peer", "name", "remote1", + ], + )?; + run("ip", &["link", "set", "remote1", "netns", "peer"])?; + for args in [ + vec!["link", "set", "native1", "up"], + vec!["addr", "add", "192.0.3.1/24", "dev", "native1"], + vec![ + "-6", + "addr", + "add", + "fd00:3::1/64", + "dev", + "native1", + "nodad", + ], + vec!["-n", "peer", "link", "set", "remote1", "up"], + vec![ + "-n", + "peer", + "addr", + "add", + "192.0.3.2/24", + "dev", + "remote1", + ], + vec![ + "-n", + "peer", + "-6", + "addr", + "add", + "fd00:3::2/64", + "dev", + "remote1", + "nodad", + ], + ] { + run("ip", &args)?; + } + let mut peer = tokio::process::Command::new("ip") + .args(["netns", "exec", "peer"]) + .arg(std::env::current_exe()?) + .args(KERNEL_TEST_ARGS) + .env("SHOES_MANAGED_PEER_ECHO", "1") + .stdout(std::process::Stdio::piped()) + .kill_on_drop(true) + .spawn()?; + line( + &mut tokio::io::BufReader::new(peer.stdout.take().unwrap()), + "Peer echo ready", + ) + .await?; + let path = "/run/reconnect.yaml"; + let unix_path = "/run/reconnect.sock"; + let mut config = vec![ + serde_json::to_value(configuration(redirect))?, + json!({"path":unix_path, "protocol":{"type":"socks"}}), + ]; + std::fs::write( + path, + serde_yaml::to_string(&config).map_err(io::Error::other)?, + )?; + let mut child = tokio::process::Command::new(std::env::current_exe()?) + .args(KERNEL_TEST_ARGS) + .env("SHOES_MANAGED_PROCESS_CONFIG", path) + .stdout(std::process::Stdio::piped()) + .kill_on_drop(true) + .spawn()?; + let mut stdout = tokio::io::BufReader::new(child.stdout.take().unwrap()); + line(&mut stdout, "Servers ready").await?; + let original = identity()?; + payload(["192.0.2.1", "fd00:1::1"]).await?; + let mut stale = tokio::net::TcpStream::connect("198.51.100.2:19123").await?; + stale.write_all(b"ping").await?; + let mut stale = tokio::io::BufReader::new(stale); + let mut reply = String::new(); + stale.read_line(&mut reply).await?; + run("ip", &["link", "add", "unused0", "type", "dummy"])?; + run("ip", &["link", "set", "unused0", "up"])?; + run("ip", &["addr", "add", "192.0.99.1/24", "dev", "unused0"])?; + tokio::time::sleep(Duration::from_secs(2)).await; + stale.write_all(b"ping").await?; + reply.clear(); + tokio::time::timeout(Duration::from_secs(3), stale.read_line(&mut reply)).await??; + assert_eq!( + reply.trim(), + "192.0.2.1", + "unrelated interface churn reset a live flow" + ); + run("ip", &["link", "del", "unused0"])?; + std::fs::write(path, "invalid: [")?; + run("ip", &["route", "replace", "default", "via", "192.0.3.2"])?; + run( + "ip", + &["-6", "route", "replace", "default", "via", "fd00:3::2"], + )?; + line(&mut stdout, "Network reconnected").await?; + assert_eq!(identity()?, original); + let result = tokio::time::timeout(Duration::from_secs(3), stale.read(&mut [0])).await?; + assert!(matches!(result, Ok(0)) || result.is_err(), "{result:?}"); + payload(["192.0.3.1", "fd00:3::1"]).await?; + + // Both families disappear and return unchanged; recovery must not require a new config. + run("ip", &["route", "del", "default"])?; + run("ip", &["-6", "route", "del", "default"])?; + line(&mut stdout, "Network unavailable; waiting for underlay").await?; + assert!(child.try_wait()?.is_none()); + assert_eq!(identity()?, original); + // A retryable bind failure must not remove capture or block a correcting reload. + std::fs::remove_file(unix_path)?; + std::fs::create_dir(unix_path)?; + run("ip", &["route", "add", "default", "via", "192.0.3.2"])?; + run("ip", &["-6", "route", "add", "default", "via", "fd00:3::2"])?; + assert!( + tokio::time::timeout( + Duration::from_secs(3), + line(&mut stdout, "Network reconnected") + ) + .await + .is_err() + ); + assert!(child.try_wait()?.is_none()); + assert_eq!(identity()?, original); + config.pop(); + config.push(json!({"address":"127.0.0.1:19124", "protocol":{"type":"socks"}})); + std::fs::write( + path, + serde_yaml::to_string(&config).map_err(io::Error::other)?, + )?; + assert_eq!( + unsafe { libc::kill(child.id().unwrap() as i32, libc::SIGHUP) }, + 0 + ); + line(&mut stdout, "Network reconnected").await?; + line(&mut stdout, "Servers ready").await?; + assert!(std::path::Path::new(unix_path).is_dir()); + std::fs::remove_dir(unix_path)?; + let mut socks = tokio::net::TcpStream::connect("127.0.0.1:19124").await?; + socks.write_all(&[5, 1, 0]).await?; + let mut greeting = [0; 2]; + tokio::time::timeout(Duration::from_secs(3), socks.read_exact(&mut greeting)).await??; + assert_eq!(greeting, [5, 0]); + drop(socks); + payload(["192.0.3.1", "fd00:3::1"]).await?; + + // Same-interface source replacement within the existing subnet. + run("ip", &["addr", "add", "192.0.3.9/24", "dev", "native1"])?; + run( + "ip", + &[ + "route", + "replace", + "default", + "via", + "192.0.3.2", + "src", + "192.0.3.9", + ], + )?; + run( + "ip", + &[ + "-6", + "addr", + "add", + "fd00:3::9/64", + "dev", + "native1", + "nodad", + ], + )?; + run( + "ip", + &[ + "-6", + "route", + "replace", + "default", + "via", + "fd00:3::2", + "src", + "fd00:3::9", + ], + )?; + line(&mut stdout, "Network reconnected").await?; + payload(["192.0.3.9", "fd00:3::9"]).await?; + assert_eq!(identity()?, original); + // Multipath defaults use nested nexthops even with only one eligible path. + run( + "ip", + &[ + "route", + "replace", + "default", + "src", + "192.0.3.9", + "nexthop", + "via", + "192.0.3.2", + "dev", + "native1", + ], + )?; + run( + "ip", + &[ + "-6", + "route", + "replace", + "default", + "src", + "fd00:3::9", + "nexthop", + "via", + "fd00:3::2", + "dev", + "native1", + ], + )?; + run("ip", &["link", "set", "native1", "down"])?; + run("ip", &["link", "set", "native1", "up"])?; + // Link-down can remove IPv6 addresses; restore the same lease before its default. + for address in ["fd00:3::1/64", "fd00:3::9/64"] { + run( + "ip", + &["-6", "addr", "replace", address, "dev", "native1", "nodad"], + )?; + } + run( + "ip", + &[ + "route", + "replace", + "default", + "src", + "192.0.3.9", + "nexthop", + "via", + "192.0.3.2", + "dev", + "native1", + ], + )?; + run( + "ip", + &[ + "-6", + "route", + "replace", + "default", + "src", + "fd00:3::9", + "nexthop", + "via", + "fd00:3::2", + "dev", + "native1", + ], + )?; + line(&mut stdout, "Network reconnected").await?; + payload(["192.0.3.9", "fd00:3::9"]).await?; + run("ip", &["route", "del", "default"])?; + run("ip", &["-6", "route", "del", "default"])?; + line(&mut stdout, "Network unavailable; waiting for underlay").await?; + assert_eq!( + unsafe { libc::kill(child.id().unwrap() as i32, libc::SIGTERM) }, + 0 + ); + assert!( + tokio::time::timeout(Duration::from_secs(10), child.wait()) + .await?? + .success() + ); + peer.kill().await?; + peer.wait().await?; + run("ip", &["link", "del", "native1"])?; + run("ip", &["route", "replace", "default", "via", "192.0.2.2"])?; + run( + "ip", + &["-6", "route", "replace", "default", "via", "fd00:1::2"], + )?; + Ok(()) +} diff --git a/src/desktop/linux/underlay.rs b/src/desktop/linux/underlay.rs new file mode 100644 index 00000000..75be86ea --- /dev/null +++ b/src/desktop/linux/underlay.rs @@ -0,0 +1,455 @@ +use std::collections::BTreeSet; +use std::io; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; + +use futures::TryStreamExt; +use parking_lot::Mutex; +use rtnetlink::packet_route::{ + RouteNetlinkMessage, + address::{AddressAttribute, AddressFlags, AddressHeaderFlags, AddressMessage}, + link::{LinkAttribute, LinkFlags}, + route::{RouteAddress, RouteAttribute, RouteMessage, RouteVia}, +}; +use rtnetlink::{Handle, RouteMessageBuilder}; + +use super::routes; + +#[derive(Clone, Debug, PartialEq, Eq)] +struct Path { + index: u32, + name: String, + gateway: Option, + source: Option, + preferred_source: Option, + nexthop: Option, + metric: u32, + addresses: BTreeSet, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub(super) struct Snapshot([Option; 2]); + +impl Snapshot { + pub fn available(&self) -> bool { + self.0.iter().any(Option::is_some) + } + + fn changed_from(&self, previous: &Self) -> bool { + self.0 + .iter() + .zip(&previous.0) + .any(|(new, old)| match (new, old) { + (Some(new), Some(old)) => { + new.index != old.index + || new.name != old.name + || new.gateway != old.gateway + || new.preferred_source != old.preferred_source + // A new privacy address can become preferred while the old source is valid. + || old + .source + .is_some_and(|source| !new.addresses.contains(&source)) + } + (None, None) => false, + _ => true, + }) + } +} + +#[derive(Default)] +struct State { + snapshot: Snapshot, + disrupted: bool, +} + +#[derive(Default)] +pub(super) struct Monitor(Mutex); + +impl Monitor { + pub fn initialize(&self, snapshot: Snapshot) { + *self.0.lock() = State { + snapshot, + disrupted: false, + }; + } + + pub fn update(&self, snapshot: Snapshot) -> bool { + let mut state = self.0.lock(); + let changed = snapshot.changed_from(&state.snapshot) || state.disrupted; + state.snapshot = snapshot; + state.disrupted = false; + changed + } + + pub fn available(&self) -> bool { + self.0.lock().snapshot.available() + } + + // Retains losses even when a down/up burst finishes before the next dump. + pub fn event(&self, message: &RouteNetlinkMessage) { + let mut state = self.0.lock(); + let lost = state.snapshot.0.iter().flatten().any(|path| match message { + RouteNetlinkMessage::DelLink(link) => link.header.index == path.index, + RouteNetlinkMessage::NewLink(link) => { + link.header.index == path.index + && !link + .header + .flags + .contains(LinkFlags::Up | LinkFlags::LowerUp) + } + RouteNetlinkMessage::DelAddress(address) => { + address.header.index == path.index && address.attributes.iter().any(|attribute| { + matches!(attribute, AddressAttribute::Local(ip) | AddressAttribute::Address(ip) + if Some(*ip) == path.source) + }) + } + RouteNetlinkMessage::DelRoute(route) => { + routes::native_default(route) + && matches_path(route, path.index, path.gateway, path.nexthop) + && metric(route) == path.metric + } + _ => false, + }); + state.disrupted |= lost; + } +} + +fn metric(route: &RouteMessage) -> u32 { + route + .attributes + .iter() + .find_map(|attribute| match attribute { + RouteAttribute::Priority(priority) => Some(*priority), + _ => None, + }) + .unwrap_or(0) +} + +fn route_ip(route: &RouteMessage, source: bool) -> Option { + attribute_ip(&route.attributes, source) +} + +fn attribute_ip(attributes: &[RouteAttribute], source: bool) -> Option { + attributes.iter().find_map(|attribute| { + let address = match attribute { + RouteAttribute::PrefSource(address) if source => address, + RouteAttribute::Gateway(address) if !source => address, + RouteAttribute::Via(RouteVia::Inet(ip)) if !source => return Some((*ip).into()), + RouteAttribute::Via(RouteVia::Inet6(ip)) if !source => return Some((*ip).into()), + _ => return None, + }; + match address { + RouteAddress::Inet(ip) => Some(IpAddr::V4(*ip)), + RouteAddress::Inet6(ip) => Some(IpAddr::V6(*ip)), + _ => None, + } + }) +} + +fn nexthop(route: &RouteMessage) -> Option { + route + .attributes + .iter() + .find_map(|attribute| match attribute { + RouteAttribute::NhId(id) => Some(*id), + _ => None, + }) +} + +fn matches_path( + route: &RouteMessage, + index: u32, + gateway: Option, + nh: Option, +) -> bool { + (route.attributes.contains(&RouteAttribute::Oif(index)) && route_ip(route, false) == gateway) + || nh.is_some_and(|id| nexthop(route) == Some(id)) + || route.attributes.iter().any(|attribute| match attribute { + RouteAttribute::MultiPath(hops) => hops.iter().any(|hop| { + hop.interface_index == index && attribute_ip(&hop.attributes, false) == gateway + }), + _ => false, + }) +} + +fn usable_addresses(addresses: &[AddressMessage], index: u32, ipv6: bool) -> BTreeSet { + addresses.iter().filter(|address| { + address.header.index == index + && !address.header.flags.intersects(AddressHeaderFlags::Tentative | AddressHeaderFlags::Dadfailed) + && !address.attributes.iter().any(|attribute| matches!(attribute, + AddressAttribute::Flags(flags) if flags.intersects(AddressFlags::Tentative | AddressFlags::Dadfailed))) + }).filter_map(|address| { + address.attributes.iter().find_map(|attribute| match attribute { + AddressAttribute::Local(ip) => Some(*ip), + _ => None, + }).or_else(|| address.attributes.iter().find_map(|attribute| match attribute { + AddressAttribute::Address(ip) => Some(*ip), + _ => None, + })) + }).filter(|ip| ip.is_ipv6() == ipv6 && !ip.is_loopback() && !ip.is_unspecified()).collect() +} + +pub(super) async fn snapshot(handle: &Handle, owned_index: u32, mark: u32) -> io::Result { + let links = routes::request(handle.link().get().execute().try_collect::>()).await?; + let addresses = + routes::request(handle.address().get().execute().try_collect::>()).await?; + let mut snapshot = Snapshot::default(); + for ipv6 in [false, true] { + let defaults = routes::dump(handle, ipv6).await?; + if !defaults.iter().any(routes::native_default) { + continue; + } + // Read-only lookups select the kernel's actual next hop and source. No probe is sent. + let query = if ipv6 { + RouteMessageBuilder::::new() + .destination_prefix("2606:4700:4700::1111".parse().unwrap(), 128) + .mark(mark) + .build() + } else { + RouteMessageBuilder::::new() + .destination_prefix(Ipv4Addr::new(1, 1, 1, 1), 32) + .mark(mark) + .build() + }; + let mut response = handle.route().get(query).execute(); + let lookup = response.try_next(); + let route = match tokio::time::timeout(std::time::Duration::from_secs(10), lookup).await { + Ok(Ok(Some(route))) => route, + Ok(Ok(None)) => continue, + Ok(Err(rtnetlink::Error::NetlinkError(_))) => continue, + Ok(Err(error)) => return Err(io::Error::other(error)), + Err(_) => { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "underlay lookup timed out", + )); + } + }; + let Some(index) = route + .attributes + .iter() + .find_map(|attribute| match attribute { + RouteAttribute::Oif(index) => Some(*index), + _ => None, + }) + else { + continue; + }; + if index == owned_index || routes::table(&route) != 254 { + continue; + } + let gateway = route_ip(&route, false); + let nexthop = nexthop(&route); + let Some(default) = defaults + .iter() + .filter(|route| { + routes::native_default(route) && matches_path(route, index, gateway, nexthop) + }) + .min_by_key(|route| metric(route)) + else { + continue; + }; + let Some(link) = links.iter().find(|link| link.header.index == index) else { + continue; + }; + if !link + .header + .flags + .contains(LinkFlags::Up | LinkFlags::LowerUp) + { + continue; + } + let Some(name) = link + .attributes + .iter() + .find_map(|attribute| match attribute { + LinkAttribute::IfName(name) => Some(name.clone()), + _ => None, + }) + else { + continue; + }; + let addresses = usable_addresses(&addresses, index, ipv6); + if addresses.is_empty() { + continue; + } + snapshot.0[usize::from(ipv6)] = Some(Path { + index, + name, + gateway, + source: route_ip(&route, true), + preferred_source: route_ip(default, true), + nexthop, + metric: metric(default), + addresses, + }); + } + Ok(snapshot) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn sample(ipv6: bool) -> Snapshot { + let mut snapshot = Snapshot::default(); + let source: IpAddr = if ipv6 { "2001:db8::1" } else { "192.0.2.1" } + .parse() + .unwrap(); + snapshot.0[usize::from(ipv6)] = Some(Path { + index: 2, + name: "native0".into(), + gateway: Some( + if ipv6 { "2001:db8::2" } else { "192.0.2.2" } + .parse() + .unwrap(), + ), + source: Some(source), + preferred_source: None, + nexthop: None, + metric: 100, + addresses: [source].into(), + }); + snapshot + } + + #[test] + fn identity_source_gateway_and_availability_changes_reset() { + for ipv6 in [false, true] { + let initial = sample(ipv6); + let monitor = Monitor::default(); + monitor.initialize(initial.clone()); + assert!(!monitor.update(initial.clone())); + for mutation in 0..4 { + let mut changed = initial.clone(); + let path = changed.0[usize::from(ipv6)].as_mut().unwrap(); + match mutation { + 0 => path.index += 1, + 1 => path.name = "native1".into(), + 2 => { + path.gateway = Some( + if ipv6 { "2001:db8::3" } else { "192.0.2.3" } + .parse() + .unwrap(), + ) + } + _ => { + path.source = Some( + if ipv6 { "2001:db8::4" } else { "192.0.2.4" } + .parse() + .unwrap(), + ); + path.addresses = [path.source.unwrap()].into(); + } + } + monitor.initialize(initial.clone()); + assert!(monitor.update(changed.clone())); + assert!(!monitor.update(changed)); + } + monitor.initialize(initial.clone()); + assert!(monitor.update(Snapshot::default())); + assert!(!monitor.available()); + assert!(!monitor.update(Snapshot::default())); + assert!(monitor.update(initial)); + assert!(monitor.available()); + } + } + + #[test] + fn metadata_and_unused_addresses_do_not_reset_but_source_removal_does() { + let monitor = Monitor::default(); + let initial = sample(false); + monitor.initialize(initial.clone()); + let mut changed = initial; + let path = changed.0[0].as_mut().unwrap(); + path.metric = 200; + path.addresses.insert("192.0.2.4".parse().unwrap()); + assert!(!monitor.update(changed.clone())); + changed.0[0] + .as_mut() + .unwrap() + .addresses + .remove(&"192.0.2.1".parse().unwrap()); + assert!(monitor.update(changed)); + } + + #[test] + fn privacy_rotation_keeps_old_flows_but_explicit_source_changes_reset() { + let monitor = Monitor::default(); + let initial = sample(true); + monitor.initialize(initial.clone()); + let mut rotated = initial.clone(); + let path = rotated.0[1].as_mut().unwrap(); + path.source = Some("2001:db8::9".parse().unwrap()); + path.addresses.insert(path.source.unwrap()); + assert!(!monitor.update(rotated.clone())); + rotated.0[1] + .as_mut() + .unwrap() + .addresses + .remove(&"2001:db8::1".parse().unwrap()); + assert!(!monitor.update(rotated.clone())); + rotated.0[1].as_mut().unwrap().preferred_source = Some("2001:db8::9".parse().unwrap()); + assert!(monitor.update(rotated)); + } + + #[test] + fn multipath_defaults_match_the_selected_hop_and_preserve_loss_events() { + use rtnetlink::packet_route::route::{RouteNextHop, RouteType}; + let initial = sample(false); + let path = initial.0[0].as_ref().unwrap(); + let mut route = RouteMessage::default(); + route.header.table = 254; + route.header.kind = RouteType::Unicast; + let mut hop = RouteNextHop::default(); + hop.interface_index = path.index; + hop.attributes + .push(RouteAttribute::Gateway(RouteAddress::Inet( + "192.0.2.2".parse().unwrap(), + ))); + route.attributes = vec![ + RouteAttribute::MultiPath(vec![hop]), + RouteAttribute::Priority(100), + ]; + assert!(matches_path(&route, path.index, path.gateway, None)); + assert!(!matches_path(&route, 99, path.gateway, None)); + let monitor = Monitor::default(); + monitor.initialize(initial.clone()); + monitor.event(&RouteNetlinkMessage::DelRoute(route)); + assert!(monitor.update(initial)); + } + + #[test] + fn short_outages_survive_event_coalescing_without_tracking_unrelated_links() { + let monitor = Monitor::default(); + let initial = sample(false); + monitor.initialize(initial.clone()); + let mut link = rtnetlink::packet_route::link::LinkMessage::default(); + link.header.index = 3; + monitor.event(&RouteNetlinkMessage::DelLink(link.clone())); + assert!(!monitor.update(initial.clone())); + link.header.index = 2; + monitor.event(&RouteNetlinkMessage::NewLink(link.clone())); + link.header.flags = LinkFlags::Up | LinkFlags::LowerUp; + monitor.event(&RouteNetlinkMessage::NewLink(link)); + assert!(monitor.update(initial.clone())); + assert!(!monitor.update(initial)); + } + + #[test] + fn unusable_addresses_and_lifetime_countdowns_are_ignored() { + let mut address = AddressMessage::default(); + address.header.index = 2; + address + .attributes + .push(AddressAttribute::Address("2001:db8::1".parse().unwrap())); + assert_eq!(usable_addresses(&[address.clone()], 2, true).len(), 1); + assert!(usable_addresses(&[address.clone()], 3, true).is_empty()); + address.header.flags = AddressHeaderFlags::Tentative; + assert!(usable_addresses(&[address.clone()], 2, true).is_empty()); + address.header.flags = AddressHeaderFlags::empty(); + address + .attributes + .push(AddressAttribute::Flags(AddressFlags::Dadfailed)); + assert!(usable_addresses(&[address], 2, true).is_empty()); + } +} diff --git a/src/desktop/mod.rs b/src/desktop/mod.rs index ba1ee2b8..b2d76509 100644 --- a/src/desktop/mod.rs +++ b/src/desktop/mod.rs @@ -6,19 +6,70 @@ pub(crate) const MANAGED_STATE_DIRECTORY: &str = "/run/shoes"; use std::io; +#[cfg(test)] +tokio::task_local! { + pub(crate) static TEST_NETWORK: std::cell::RefCell>; +} + +pub(crate) struct NetworkTopology { + #[cfg(target_os = "linux")] + spec: Option, + #[cfg(target_os = "linux")] + ingress: Option>, +} + +impl NetworkTopology { + pub(crate) fn validate(&self, configs: &[crate::config::ResolvedConfig]) -> io::Result<()> { + #[cfg(target_os = "linux")] + if linux::Spec::from_resolved_configs(configs)? != self.spec { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + if self.spec.is_none() { + "enabling managed TUN requires a restart" + } else { + "managed TUN topology changes require a restart" + }, + )); + } + let _ = configs; + Ok(()) + } + + #[cfg(target_os = "linux")] + pub(crate) fn ingress(&self) -> Option> { + self.ingress.clone() + } +} + #[derive(Default)] pub(crate) struct NetworkSession { + #[cfg(test)] + test_changed: bool, + #[cfg(test)] + test_available: Option, #[cfg(target_os = "linux")] session: Option, #[cfg(test)] test_events: Option<( - tokio::sync::mpsc::Receiver<()>, + tokio::sync::mpsc::Receiver>, tokio::sync::mpsc::Sender<()>, )>, } impl NetworkSession { + pub(crate) fn topology(&self) -> NetworkTopology { + NetworkTopology { + #[cfg(target_os = "linux")] + spec: self.session.as_ref().map(|session| session.spec().clone()), + #[cfg(target_os = "linux")] + ingress: self.ingress(), + } + } pub(crate) async fn prepare(configs: &[crate::config::Config]) -> io::Result { + #[cfg(test)] + if let Ok(Some(network)) = TEST_NETWORK.try_with(|network| network.borrow_mut().take()) { + return Ok(network); + } #[cfg(target_os = "linux")] { if let Some(spec) = linux::Spec::from_configs(configs)? { @@ -26,6 +77,10 @@ impl NetworkSession { session: Some(linux::Session::prepare(spec).await?), #[cfg(test)] test_events: None, + #[cfg(test)] + test_changed: false, + #[cfg(test)] + test_available: None, }); } } @@ -33,20 +88,16 @@ impl NetworkSession { Ok(Self::default()) } - pub(crate) fn validate(&self, configs: &[crate::config::ResolvedConfig]) -> io::Result<()> { - #[cfg(target_os = "linux")] - match &self.session { - Some(session) => return session.validate(configs), - None if linux::Spec::from_resolved_configs(configs)?.is_some() => { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - "enabling managed TUN requires a restart", - )); - } - _ => {} + pub(crate) fn monitors_underlay(&self) -> bool { + #[cfg(test)] + if self.test_events.is_some() { + return true; } - let _ = configs; - Ok(()) + #[cfg(target_os = "linux")] + if self.session.is_some() { + return true; + } + false } #[cfg(target_os = "linux")] @@ -73,7 +124,11 @@ impl NetworkSession { pub(crate) async fn changed(&mut self) { #[cfg(test)] if let Some((changes, _)) = &mut self.test_events { - changes.recv().await.unwrap(); + let event = changes.recv().await.unwrap(); + self.test_changed = event.is_some(); + if let Some(available) = event { + self.test_available = Some(available); + } return; } #[cfg(target_os = "linux")] @@ -83,22 +138,42 @@ impl NetworkSession { std::future::pending().await } - pub(crate) async fn reconcile(&mut self) -> io::Result<()> { + pub(crate) fn available(&self) -> bool { #[cfg(test)] - if let Some((_, reconciled)) = &self.test_events { - reconciled.send(()).await.unwrap(); - return Ok(()); + if let Some(available) = self.test_available { + return available; } #[cfg(target_os = "linux")] - if let Some(session) = &mut self.session { - session.reconcile().await?; + if let Some(session) = &self.session { + return session.available(); + } + true + } + + pub(crate) fn refresh_bootstrap(&self) -> io::Result<()> { + #[cfg(target_os = "linux")] + if let Some(session) = &self.session { + session.refresh_bootstrap()?; } Ok(()) } + pub(crate) async fn reconcile(&mut self) -> io::Result { + #[cfg(test)] + if let Some((_, reconciled)) = &self.test_events { + let _ = reconciled.try_send(()); + return Ok(std::mem::take(&mut self.test_changed)); + } + #[cfg(target_os = "linux")] + if let Some(session) = &mut self.session { + return session.reconcile().await; + } + Ok(false) + } + #[cfg(test)] pub(crate) fn with_test_events( - changes: tokio::sync::mpsc::Receiver<()>, + changes: tokio::sync::mpsc::Receiver>, reconciled: tokio::sync::mpsc::Sender<()>, ) -> Self { Self { diff --git a/src/dns/bootstrap.rs b/src/dns/bootstrap.rs index efd88d3c..e12538c0 100644 --- a/src/dns/bootstrap.rs +++ b/src/dns/bootstrap.rs @@ -46,7 +46,7 @@ pub(crate) fn controlled(endpoints: &[SocketAddr]) -> io::Result io::Result Self { + Self { + limits: self.limits, + forbidden: self.forbidden.clone(), + pools: self.pools.clone(), + ..Self::new() + } + } + #[cfg(unix)] pub(crate) fn for_rule_bootstrap( configs: &[crate::config::Config], @@ -195,7 +204,7 @@ impl DnsRegistry { /// Register a DNS group. pub fn register(&mut self, name: String, resolver: Arc) { - self.groups.insert(name, resolver); + self.groups.insert(name, super::network::scoped(resolver)); } /// Get a resolver by group name, returns None if not found. @@ -213,14 +222,14 @@ impl DnsRegistry { let forbidden = self.forbidden.clone(); self.default_resolver .get_or_insert_with(|| { - real_only( + super::network::scoped(real_only( Arc::new(SystemResolver::new( Arc::new(CachingNativeResolver::new()), forbidden.clone(), Duration::from_secs(5), )), forbidden, - ) + )) }) .clone() } diff --git a/src/dns/builder/tests.rs b/src/dns/builder/tests.rs index 0bf4bd76..69943c04 100644 --- a/src/dns/builder/tests.rs +++ b/src/dns/builder/tests.rs @@ -553,6 +553,52 @@ async fn fake_context_reload_sharing_disable_and_reenable_keep_pool_identity() { assert_eq!(context.restore(ip).unwrap().as_deref(), Some("stable.test")); } +#[tokio::test] +async fn network_rebuild_retires_resolvers_but_keeps_live_fake_ip_pools() { + let old = crate::network_tasks::NetworkLifetime::default(); + let registry = old + .scope(build_dns_registry(fake_groups(true))) + .await + .unwrap(); + let resolver = registry.get_by_name("fake").unwrap(); + let context = resolver.fake_dns_context().unwrap(); + let ip = context.allocate("network.test", false).unwrap(); + let retained = registry.retained_state(); + old.cancel(); + old.wait().await; + assert_eq!( + resolver + .resolve_location( + &crate::address::NetLocation::from_str("network.test:80", None).unwrap() + ) + .await + .unwrap_err() + .kind(), + std::io::ErrorKind::Interrupted + ); + let fresh = crate::network_tasks::NetworkLifetime::default(); + let rebuilt = fresh + .scope(build_dns_registry_reusing( + fake_groups(true), + registry.limits, + Some(&retained), + )) + .await + .unwrap(); + assert!(Arc::ptr_eq( + ®istry.pools["fake"].pool, + &rebuilt.pools["fake"].pool + )); + let next = rebuilt + .get_by_name("fake") + .unwrap() + .fake_dns_context() + .unwrap(); + assert_eq!(next.restore(ip).unwrap().as_deref(), Some("network.test")); + assert_eq!(next.allocate("network.test", false).unwrap(), ip); + assert!(next.allocate("after.test", false).is_ok()); +} + #[tokio::test] async fn equivalent_pool_ranges_reuse_live_mappings_on_reload() { let mut groups = fake_groups(true); diff --git a/src/dns/exchange.rs b/src/dns/exchange.rs index 31953e70..64517710 100644 --- a/src/dns/exchange.rs +++ b/src/dns/exchange.rs @@ -323,7 +323,7 @@ fn connection_setup( deadline: tokio::time::Instant, ) -> Connection { // A losing query must not cancel a shared handshake. Dropping its cache owner still does. - let task = tokio_util::task::AbortOnDropHandle::new(tokio::spawn(async move { + let task = tokio_util::task::AbortOnDropHandle::new(crate::network_tasks::spawn(async move { tokio::time::timeout_at(deadline, future) .await .map_err(|_| { diff --git a/src/dns/mod.rs b/src/dns/mod.rs index cf5a08d1..eb7dfa6f 100644 --- a/src/dns/mod.rs +++ b/src/dns/mod.rs @@ -20,6 +20,7 @@ pub mod fake_ip; mod guard; mod hickory_resolver; mod message; +mod network; mod parsed; mod proxy_runtime; mod stream; diff --git a/src/dns/network.rs b/src/dns/network.rs new file mode 100644 index 00000000..304291c7 --- /dev/null +++ b/src/dns/network.rs @@ -0,0 +1,62 @@ +use std::future::Future; +use std::io; +use std::pin::Pin; +use std::sync::Arc; + +use crate::network_tasks::NetworkTasks; +use crate::resolver::{DnsExchangeFuture, ResolveFuture, Resolver}; + +#[derive(Debug)] +struct NetworkResolver { + inner: Arc, + network: NetworkTasks, +} + +pub(super) fn scoped(inner: Arc) -> Arc { + match NetworkTasks::current() { + Some(network) => Arc::new(NetworkResolver { inner, network }), + None => inner, + } +} + +impl NetworkResolver { + fn run( + &self, + operation: impl FnOnce(Arc) -> Pin> + Send>> + + Send + + 'static, + ) -> Pin> + Send>> { + let inner = self.inner.clone(); + let network = self.network.clone(); + Box::pin(async move { + tokio::select! { + biased; + _ = network.cancelled.cancelled() => Err(io::Error::new(io::ErrorKind::Interrupted, "DNS network retired")), + result = network.scope(async move { operation(inner).await }) => result, + } + }) + } +} + +impl Resolver for NetworkResolver { + fn resolve_location(&self, location: &crate::address::NetLocation) -> ResolveFuture { + let location = location.clone(); + self.run(move |inner| inner.resolve_location(&location)) + } + + fn exchange(&self, request: Vec) -> DnsExchangeFuture { + self.run(move |inner| inner.exchange(request)) + } + + fn exchange_with_deadline( + &self, + request: Vec, + deadline: tokio::time::Instant, + ) -> DnsExchangeFuture { + self.run(move |inner| inner.exchange_with_deadline(request, deadline)) + } + + fn fake_dns_context(&self) -> Option> { + self.inner.fake_dns_context() + } +} diff --git a/src/dns/proxy_runtime.rs b/src/dns/proxy_runtime.rs index 0c660078..3d5af780 100644 --- a/src/dns/proxy_runtime.rs +++ b/src/dns/proxy_runtime.rs @@ -30,6 +30,7 @@ pub struct ProxyRuntimeProvider { quic_binder: ProxyQuicBinder, /// Timeout for establishing connections to DNS upstreams. connect_timeout: Duration, + network: Option, } impl ProxyRuntimeProvider { @@ -49,6 +50,7 @@ impl ProxyRuntimeProvider { bootstrap_resolver, quic_binder, connect_timeout, + network: crate::network_tasks::NetworkTasks::current(), } } @@ -65,11 +67,18 @@ impl ProxyRuntimeProvider { /// Spawn handle for tokio runtime. #[derive(Clone, Default)] -pub struct TokioSpawnHandle; +pub struct TokioSpawnHandle(Option); impl Spawn for TokioSpawnHandle { fn spawn_bg(&mut self, future: impl Future + Send + 'static) { - tokio::spawn(future); + match &self.0 { + Some(network) => { + network.spawn(future); + } + None => { + crate::network_tasks::spawn(future); + } + } } } @@ -83,7 +92,7 @@ impl RuntimeProvider for ProxyRuntimeProvider { type Tcp = ProxiedTcp; fn create_handle(&self) -> Self::Handle { - TokioSpawnHandle + TokioSpawnHandle(self.network.clone()) } fn connect_tcp( diff --git a/src/dns/proxy_runtime/tests.rs b/src/dns/proxy_runtime/tests.rs index 0f4ea283..953d6f81 100644 --- a/src/dns/proxy_runtime/tests.rs +++ b/src/dns/proxy_runtime/tests.rs @@ -122,7 +122,7 @@ fn test_provider_is_clone() { #[test] fn test_spawn_handle_is_clone() { - let handle = TokioSpawnHandle; + let handle = TokioSpawnHandle::default(); let _cloned = handle.clone(); } diff --git a/src/dns/stream.rs b/src/dns/stream.rs index 6e93b1d5..bbfb991d 100644 --- a/src/dns/stream.rs +++ b/src/dns/stream.rs @@ -162,7 +162,7 @@ pub(super) struct TcpDnsStream { impl TcpDnsStream { pub fn new(context: Arc) -> Self { let (stream, mut peer) = tokio::io::duplex(16 * 1024); - let task = tokio::spawn(async move { + let task = crate::network_tasks::spawn(async move { let _ = serve_tcp(&mut peer, context).await; }); Self { stream, task } diff --git a/src/h2mux/h2mux_client_session.rs b/src/h2mux/h2mux_client_session.rs index 9f9fdc68..3b97e5b8 100644 --- a/src/h2mux/h2mux_client_session.rs +++ b/src/h2mux/h2mux_client_session.rs @@ -125,7 +125,7 @@ impl H2MuxClientSession { let (drain_tx, drain_rx) = oneshot::channel(); // Spawn connection driver - let abort_handle = tokio::spawn(async move { + let abort_handle = crate::network_tasks::spawn(async move { // Dropping the last owner must still flush queued DATA and END_STREAM. let result = tokio::select! { result = &mut connection => Some(result), @@ -167,7 +167,7 @@ impl H2MuxClientSession { is_closed: Arc, driver: tokio::task::AbortHandle, ) -> tokio::task::AbortHandle { - tokio::spawn(async move { + crate::network_tasks::spawn(async move { let mut timer = interval(PING_INTERVAL); timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); // Skip the first tick which returns immediately diff --git a/src/h2mux/h2mux_server_session.rs b/src/h2mux/h2mux_server_session.rs index 38b7b8ae..d6702301 100644 --- a/src/h2mux/h2mux_server_session.rs +++ b/src/h2mux/h2mux_server_session.rs @@ -145,7 +145,7 @@ impl H2MuxServerSession { // Spawn acceptor task with idle timeout monitoring let is_closed_clone = Arc::clone(&is_closed); - let driver = tokio::spawn(async move { + let driver = crate::network_tasks::spawn(async move { Self::accept_loop(connection, inbound_tx, is_closed_clone, activity).await; }); @@ -188,7 +188,7 @@ impl H2MuxServerSession { Some(Ok((request, respond))) => { let inbound_tx = inbound_tx.clone(); let slots = slots.clone(); - tasks.spawn(async move { + crate::network_tasks::spawn_in(&mut tasks, async move { if let Err(e) = Self::handle_stream(request, respond, inbound_tx, slots).await { debug!("H2MuxServerSession: stream error: {}", e); } @@ -227,7 +227,7 @@ impl H2MuxServerSession { Ok((request, respond)) => { let inbound_tx = inbound_tx.clone(); let slots = slots.clone(); - tasks.spawn(async move { + crate::network_tasks::spawn_in(&mut tasks, async move { if let Err(e) = Self::handle_stream(request, respond, inbound_tx, slots).await { debug!("H2MuxServerSession: stream error during drain: {}", e); } @@ -367,7 +367,7 @@ where }; let Some(inbound) = inbound else { break }; let fut = handler(inbound); - tasks.spawn(async move { + crate::network_tasks::spawn_in(&mut tasks, async move { if let Err(e) = fut.await { debug!("H2MUX stream handler error: {}", e); } @@ -415,7 +415,7 @@ where let proxy_selector = proxy_selector.clone(); let resolver = resolver.clone(); - tasks.spawn(async move { + crate::network_tasks::spawn_in(&mut tasks, async move { if let Err(e) = handle_h2mux_stream(inbound, udp_enabled, proxy_selector, resolver).await { diff --git a/src/http2/client.rs b/src/http2/client.rs index 80f7b596..608a6313 100644 --- a/src/http2/client.rs +++ b/src/http2/client.rs @@ -100,7 +100,7 @@ impl H2ClientSession { let driver_retired = retired.clone(); let active = Arc::new(ActiveStreams::default()); let driver_active = active.clone(); - let abort_handle = tokio::spawn(async move { + let abort_handle = crate::network_tasks::spawn(async move { let completed = tokio::select! { result = &mut connection => Some(result), _ = &mut drain_rx => None, diff --git a/src/http2/server.rs b/src/http2/server.rs index 7ec44102..865802fa 100644 --- a/src/http2/server.rs +++ b/src/http2/server.rs @@ -81,7 +81,7 @@ pub(crate) async fn serve( }; let selector = selector.clone(); let resolver = resolver.clone(); - tasks.spawn(async move { + crate::network_tasks::spawn_in(&mut tasks, async move { let _permits = (local, global); if let Err(error) = tunnel(request, respond, target, selector, resolver).await { log::debug!("HTTP/2 CONNECT stream ended: {error}"); diff --git a/src/hysteria2_server.rs b/src/hysteria2_server.rs index e2ca00e1..0a7cbf38 100644 --- a/src/hysteria2_server.rs +++ b/src/hysteria2_server.rs @@ -272,7 +272,7 @@ impl UdpSession { let send_socket = client_socket.clone(); let cancel_token = session_cancel_token.clone(); - let task = tokio::spawn(async move { + let task = crate::network_tasks::spawn(async move { if let Err(e) = run_udp_remote_to_local_loop( session_id, connection, @@ -541,7 +541,7 @@ async fn run_tcp_loop( let client_proxy_selector = client_proxy_selector.clone(); let resolver = resolver.clone(); let peer = connection.remote_address(); - tasks.spawn(async move { + crate::network_tasks::spawn_in(&mut tasks, async move { let _permit = permit; if let Err(e) = process_tcp_stream( client_proxy_selector, @@ -865,7 +865,7 @@ pub async fn start_hysteria2_server( let resolver = resolver.clone(); let client_proxy_selector = client_proxy_selector.clone(); let hysteria2_password = hysteria2_password.clone(); - let join_handle = tokio::spawn(async move { + let join_handle = crate::network_tasks::spawn(async move { let mut tasks = crate::listener_tasks::ListenerTasks::immediate(); loop { let conn = tokio::select! { diff --git a/src/lib.rs b/src/lib.rs index 32cf8047..04555bc4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -60,6 +60,7 @@ mod hysteria2_server; mod listener_tasks; mod mixed_handler; mod naiveproxy; +mod network_tasks; mod ntp; mod option_util; mod port_forward_handler; diff --git a/src/listener_tasks.rs b/src/listener_tasks.rs index d47198f1..7529035a 100644 --- a/src/listener_tasks.rs +++ b/src/listener_tasks.rs @@ -118,7 +118,7 @@ impl ListenerTasks { } pub fn spawn(&mut self, future: impl Future + Send + 'static) { - self.tasks.spawn(future); + crate::network_tasks::spawn_in(&mut self.tasks, future); } pub fn is_empty(&self) -> bool { @@ -137,12 +137,12 @@ impl Drop for ListenerTasks { if self.tasks.is_empty() || self.grace.is_zero() { return; } - let Ok(runtime) = tokio::runtime::Handle::try_current() else { + if tokio::runtime::Handle::try_current().is_err() { return; - }; + } let mut tasks = std::mem::take(&mut self.tasks); let deadline = tokio::time::Instant::now() + self.grace; - runtime.spawn(async move { + crate::network_tasks::spawn(async move { let drained = tokio::time::timeout_at(deadline, async { while let Some(result) = tasks.join_next().await { if let Err(error) = result { diff --git a/src/naiveproxy/naive_client_session.rs b/src/naiveproxy/naive_client_session.rs index 7ce8f7f7..f383fedf 100644 --- a/src/naiveproxy/naive_client_session.rs +++ b/src/naiveproxy/naive_client_session.rs @@ -72,7 +72,7 @@ impl NaiveClientSession { errors.start_driver(); let (drain_tx, drain_rx) = oneshot::channel(); - let abort_handle = tokio::spawn(async move { + let abort_handle = crate::network_tasks::spawn(async move { // Retired sessions must flush queued DATA and END_STREAM before teardown. let result = tokio::select! { result = &mut connection => Some(result), diff --git a/src/naiveproxy/naive_hyper_service.rs b/src/naiveproxy/naive_hyper_service.rs index 82b4ea59..2a41398d 100644 --- a/src/naiveproxy/naive_hyper_service.rs +++ b/src/naiveproxy/naive_hyper_service.rs @@ -67,6 +67,32 @@ impl tokio::io::AsyncRead for HyperUpgradedStream { mod fallback_tests { use super::*; + #[tokio::test] + async fn executor_propagates_network_retirement_to_descendants() { + let network = crate::network_tasks::NetworkLifetime::default(); + let tasks = Arc::new(Mutex::new(JoinSet::new())); + let executor = ConnectionExecutor(Arc::downgrade(&tasks)); + let marker = Arc::new(()); + let retained = marker.clone(); + let (ready, started) = tokio::sync::oneshot::channel(); + network + .scope(async { + hyper::rt::Executor::execute(&executor, async move { + assert!(crate::network_tasks::NetworkTasks::current().is_some()); + crate::network_tasks::spawn(async move { + let _retained = retained; + ready.send(()).unwrap(); + std::future::pending::<()>().await; + }); + }); + }) + .await; + started.await.unwrap(); + network.cancel(); + network.wait().await; + assert_eq!(Arc::strong_count(&marker), 1); + } + #[tokio::test] async fn connection_executor_cancels_children_without_a_reference_cycle() { let tasks = Arc::new(Mutex::new(JoinSet::new())); @@ -185,7 +211,7 @@ where if let Some(tasks) = self.0.upgrade() { let mut tasks = tasks.lock(); while tasks.try_join_next().is_some() {} - tasks.spawn(async move { + crate::network_tasks::spawn_in(&mut tasks, async move { let _ = future.await; }); } diff --git a/src/network_tasks.rs b/src/network_tasks.rs new file mode 100644 index 00000000..081a3401 --- /dev/null +++ b/src/network_tasks.rs @@ -0,0 +1,268 @@ +use std::collections::HashMap; +use std::future::Future; +use std::sync::{Arc, Weak}; + +use parking_lot::Mutex; +use tokio::task::{AbortHandle, JoinHandle, JoinSet}; +use tokio_util::task::TaskTracker; + +tokio::task_local! { + static CURRENT: NetworkTasks; +} + +#[derive(Debug, Default)] +struct State { + cancelled: bool, + next_id: u64, + tasks: HashMap>, +} + +/// One underlay lifetime, shared across ordinary configuration reloads. +#[derive(Clone, Debug, Default)] +pub(crate) struct NetworkTasks { + untracked: bool, + state: Arc>, + tracker: TaskTracker, + pub(crate) cancelled: tokio_util::sync::CancellationToken, +} + +#[derive(Default)] +pub(crate) struct NetworkLifetime(pub NetworkTasks); + +impl NetworkLifetime { + pub fn untracked() -> Self { + Self(NetworkTasks { + untracked: true, + ..NetworkTasks::default() + }) + } +} + +impl std::ops::Deref for NetworkLifetime { + type Target = NetworkTasks; + fn deref(&self) -> &Self::Target { + &self.0 + } +} + +impl Drop for NetworkLifetime { + fn drop(&mut self) { + self.0.cancel(); + } +} + +struct Registration { + state: Weak>, + id: u64, +} + +impl Drop for Registration { + fn drop(&mut self) { + if let Some(state) = self.state.upgrade() { + state.lock().tasks.remove(&self.id); + } + } +} + +impl NetworkTasks { + pub fn current() -> Option { + CURRENT.try_with(Clone::clone).ok() + } + + pub async fn scope(&self, future: F) -> F::Output { + if self.untracked { + future.await + } else { + CURRENT.scope(self.clone(), future).await + } + } + + pub fn spawn(&self, future: F) -> JoinHandle + where + F: Future + Send + 'static, + F::Output: Send + 'static, + { + let registration = self.register(); + let id = registration.id; + let context = self.clone(); + let cancelled = self.cancelled.clone(); + let task = self.tracker.spawn(CURRENT.scope(context, async move { + let _registration = registration; + if cancelled.is_cancelled() { + std::future::pending::<()>().await; + } + future.await + })); + self.attach(id, task.abort_handle()); + task + } + + fn register(&self) -> Registration { + let mut state = self.state.lock(); + let id = state.next_id; + state.next_id += 1; + state.tasks.insert(id, None); + Registration { + state: Arc::downgrade(&self.state), + id, + } + } + + fn attach(&self, id: u64, task: AbortHandle) { + let mut state = self.state.lock(); + if state.cancelled { + task.abort(); + } + // Spawning into a stopped runtime can synchronously destroy the registration. + if let Some(slot) = state.tasks.get_mut(&id) { + *slot = Some(task); + } + } + + /// Tracks drivers that must process endpoint closure rather than being aborted. + pub fn spawn_completion(&self, future: F) + where + F: Future + Send + 'static, + { + self.tracker.spawn(CURRENT.scope(self.clone(), future)); + } + + pub fn cancel(&self) { + self.cancelled.cancel(); + let mut state = self.state.lock(); + state.cancelled = true; + for task in state.tasks.values().flatten() { + task.abort(); + } + self.tracker.close(); + } + + pub async fn wait(&self) { + self.tracker.wait().await; + } +} + +pub(crate) fn spawn(future: F) -> JoinHandle +where + F: Future + Send + 'static, + F::Output: Send + 'static, +{ + match NetworkTasks::current() { + Some(context) => context.spawn(future), + None => tokio::spawn(future), + } +} + +pub(crate) fn spawn_in(tasks: &mut JoinSet, future: F) -> AbortHandle +where + F: Future + Send + 'static, + F::Output: Send + 'static, +{ + let Some(context) = NetworkTasks::current() else { + return tasks.spawn(future); + }; + let registration = context.register(); + let id = registration.id; + let cancelled = context.cancelled.clone(); + let task = tasks.spawn(context.tracker.track_future(CURRENT.scope( + context.clone(), + async move { + let _registration = registration; + if cancelled.is_cancelled() { + std::future::pending::<()>().await; + } + future.await + }, + ))); + context.attach(id, task.clone()); + task +} + +pub(crate) fn inherit(future: F) -> impl Future { + let context = NetworkTasks::current(); + async move { + match context { + Some(context) => context.scope(future).await, + None => future.await, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn unmanaged_work_does_not_register_a_network_lifetime() { + let network = NetworkLifetime::untracked(); + let mut task = None; + network + .scope(async { + assert!(NetworkTasks::current().is_none()); + task = Some(spawn(async { NetworkTasks::current().is_none() })); + }) + .await; + network.cancel(); + assert!(task.unwrap().await.unwrap()); + assert!(network.state.lock().tasks.is_empty()); + } + + #[tokio::test] + async fn reset_cancels_descendants_and_rejects_late_spawns() { + let context = NetworkTasks::default(); + let (ready, started) = tokio::sync::oneshot::channel(); + let owned = Arc::new(()); + let retained = owned.clone(); + let parent = context.spawn(async move { + spawn(async move { + let _retained = retained; + ready.send(()).unwrap(); + std::future::pending::<()>().await; + }); + }); + parent.await.unwrap(); + started.await.unwrap(); + context.cancel(); + context.wait().await; + assert_eq!(Arc::strong_count(&owned), 1); + assert!(context.spawn(async { 1 }).await.unwrap_err().is_cancelled()); + context.wait().await; + assert!(context.state.lock().tasks.is_empty()); + assert_eq!(spawn(async { 2 }).await.unwrap(), 2); + } + + #[tokio::test] + async fn join_set_cancellation_releases_registrations_and_blocks_late_work() { + let network = NetworkLifetime::default(); + let mut tasks = JoinSet::new(); + network + .scope(async { + spawn_in(&mut tasks, std::future::pending::<()>()); + }) + .await; + network.cancel(); + network.wait().await; + assert!(tasks.join_next().await.unwrap().unwrap_err().is_cancelled()); + network + .scope(async { + spawn_in(&mut tasks, async { panic!("retired work ran") }); + }) + .await; + network.wait().await; + assert!(tasks.join_next().await.unwrap().unwrap_err().is_cancelled()); + assert!(network.state.lock().tasks.is_empty()); + } + + #[test] + fn stopped_runtime_destroys_registrations_without_locking_recursively() { + let runtime = tokio::runtime::Builder::new_current_thread() + .build() + .unwrap(); + let handle = runtime.handle().clone(); + drop(runtime); + let _entered = handle.enter(); + let network = NetworkTasks::default(); + drop(network.spawn(async {})); + assert!(network.state.lock().tasks.is_empty()); + } +} diff --git a/src/process.rs b/src/process.rs index c27476f8..6223f35f 100644 --- a/src/process.rs +++ b/src/process.rs @@ -153,7 +153,7 @@ async fn prepare(paths: &[String], previous: Option<&DnsRegistry>) -> io::Result None, &mut |_, _, _| {}, &mut None, - &crate::desktop::NetworkSession::default(), + &crate::desktop::NetworkSession::default().topology(), ) .await } @@ -165,7 +165,7 @@ async fn prepare_tracking( previous_servers: Option<&PreparedServers>, discovered: &mut DiscoverDependencies<'_>, control: &mut Option, - network: &crate::desktop::NetworkSession, + network: &crate::desktop::NetworkTopology, ) -> io::Result { let mut cache_lease = previous_servers.and_then(|previous| previous.cache_lease.clone()); let (mut validated, snapshot, memory) = load_candidate( @@ -264,7 +264,7 @@ async fn prepare_tracking( dns::check_interception(&validated.configs, &validated.dns_groups).await?; let protect_tun_endpoints = config::protects_tun_endpoints(&validated.configs); let mut dns = dns::build_dns_registry_for_runtime( - std::mem::take(&mut validated.dns_groups), + validated.dns_groups.clone(), validated.global_limits, previous, &synthetic_endpoints, @@ -366,6 +366,75 @@ struct RunningServers { } impl RunningServers { + async fn pause_network( + &mut self, + prepared: &mut PreparedServers, + control: Option<&crate::control::runtime::ControlService>, + lifetime: &crate::network_tasks::NetworkTasks, + ) -> io::Result<()> { + if let (Some(control), Some(generation)) = (control, &prepared.generation) { + control.runtime.begin_activation(generation).await?; + } + #[cfg(unix)] + self.wireguard.cancel(); + lifetime.cancel(); + self.stop().await; + lifetime.wait().await; + prepared.tcp.defer_admission(); + prepared.dns = prepared.dns.retained_state(); + self.quic_retirements.wait().await + } + + async fn reconnect( + &mut self, + prepared: &mut PreparedServers, + network: &mut crate::desktop::NetworkSession, + ) -> io::Result<()> { + self.rebuild_network(prepared, network).await?; + self.start_retry(prepared).await?; + prepared.certificates.start().await; + Ok(()) + } + + async fn rebuild_network( + &mut self, + prepared: &mut PreparedServers, + network: &mut crate::desktop::NetworkSession, + ) -> io::Result<()> { + network.refresh_bootstrap()?; + let endpoints: Vec<_> = + crate::config::tun_endpoint_ips(&prepared.validated.configs).collect(); + prepared.dns = dns::build_dns_registry_for_runtime( + prepared.validated.dns_groups.clone(), + prepared.validated.global_limits, + Some(&prepared.dns), + &endpoints, + config::protects_tun_endpoints(&prepared.validated.configs), + ) + .await?; + #[cfg(unix)] + match tokio::time::timeout(Duration::from_secs(3), prepared.wireguard.refresh_peers()).await + { + Ok(Ok(())) => {} + Ok(Err(error)) => { + log::warn!( + "WireGuard peer refresh incomplete; retaining unresolved peer endpoints: {error}" + ) + } + Err(_) => { + log::warn!("WireGuard peer refresh timed out; retaining unresolved peer endpoints") + } + } + if let Some(generation) = &prepared.generation { + generation.probes.reset_network( + prepared.dns.get_or_create_default(), + prepared.dns.forbidden_ranges().to_vec(), + ); + generation.catalog.selection.automatic.choices_changed(); + } + Ok(()) + } + async fn shutdown_network( &mut self, registries: &[&PreparedServers], @@ -829,14 +898,21 @@ async fn run_inner( _ = shutdown.recv() => return Ok(()), result = config::load_config_files(&paths) => result?, }; - *network = crate::desktop::NetworkSession::prepare(&configs).await?; + let mut lifetime = crate::network_tasks::NetworkLifetime::default(); + *network = lifetime + .scope(crate::desktop::NetworkSession::prepare(&configs)) + .await?; + if !network.monitors_underlay() { + lifetime = crate::network_tasks::NetworkLifetime::untracked(); + } let mut discovered = |paths: &[PathBuf], recovery: &[PathBuf], root: Option<&Path>| { watches.discovered(paths, recovery, root) }; + let topology = network.topology(); let mut prepared = tokio::select! { biased; _ = shutdown.recv() => return Ok(()), - result = prepare_tracking(&paths, configs, None, None, &mut discovered, control, network) => result?, + result = lifetime.scope(prepare_tracking(&paths, configs, None, None, &mut discovered, control, &topology)) => result?, }; network.prepare_network().await?; #[cfg(target_os = "linux")] @@ -857,7 +933,7 @@ async fn run_inner( tokio::select! { biased; _ = shutdown.recv() => return running.shutdown_network(&[&prepared], network).await, - result = running.start(&mut prepared) => if let Err(error) = result { + result = lifetime.scope(running.start(&mut prepared)) => if let Err(error) = result { let _ = running.shutdown_network(&[&prepared], network).await; return Err(error); }, @@ -879,11 +955,17 @@ async fn run_inner( ))); } println!("Servers ready"); + let mut reconnecting = false; + let mut pending_reload = false; + let mut force_network_reset = false; + let mut retry_at = tokio::time::Instant::now(); loop { + let can_retry = reconnecting && network.available(); enum Change { Shutdown, TunFailure(io::Error), Reload { debounce: bool }, + Reconnect, } let change = wait_with_reconciliation( async { @@ -891,21 +973,112 @@ async fn run_inner( biased; _ = shutdown.recv() => Change::Shutdown, error = running.tun_failure() => Change::TunFailure(error), + _ = std::future::ready(()), if force_network_reset => Change::Reconnect, _ = reload.recv() => Change::Reload { debounce: false }, - _ = remote_rule_change(&mut prepared) => Change::Reload { debounce: false }, + _ = lifetime.scope(remote_rule_change(&mut prepared)), if !reconnecting => Change::Reload { debounce: false }, _ = watches.changed(&mut changes) => Change::Reload { debounce: true }, + _ = tokio::time::sleep_until(retry_at), if can_retry => Change::Reconnect, + _ = std::future::ready(()), if pending_reload && !reconnecting => Change::Reload { debounce: false }, } }, network, ) .await; let debounce = match change { - Ok(Change::Shutdown) => return running.shutdown_network(&[&prepared], network).await, - Ok(Change::TunFailure(error)) => { + Ok(Some(Change::Shutdown)) => { + return running.shutdown_network(&[&prepared], network).await; + } + Ok(Some(Change::TunFailure(error))) => { let _ = running.shutdown_network(&[&prepared], network).await; return Err(error); } - Ok(Change::Reload { debounce }) => debounce, + Ok(Some(Change::Reload { debounce })) => { + if reconnecting { + pending_reload = true; + force_network_reset = network.available(); + continue; + } + pending_reload = false; + debounce + } + Ok(None) | Ok(Some(Change::Reconnect)) => { + force_network_reset = false; + if !reconnecting { + log::info!("Underlay changed; retiring network workers"); + } + // A rejected recovery reload may have rebuilt transports without starting workers. + if let Err(error) = running + .pause_network(&mut prepared, control.as_ref(), &lifetime) + .await + { + let _ = running.shutdown_network(&[&prepared], network).await; + return Err(error); + } + if !reconnecting && !network.available() { + println!("Network unavailable; waiting for underlay"); + } + reconnecting = true; + if !network.available() { + continue; + } + lifetime = crate::network_tasks::NetworkLifetime::default(); + let mut result = tokio::select! { + biased; + _ = shutdown.recv() => return running.shutdown_network(&[&prepared], network).await, + result = tokio::time::timeout(Duration::from_secs(10), lifetime.scope(async { + if pending_reload { + // Restore rollback state without requiring the old listeners to bind. + running.rebuild_network(&mut prepared, network).await + } else { + running.reconnect(&mut prepared, network).await + } + })) => result.unwrap_or_else(|_| Err(io::Error::new(io::ErrorKind::TimedOut, "network recovery timed out"))), + }; + // Ownership failures remain terminal, unlike transient worker startup errors. + match network.reconcile().await { + Err(error) => { + let _ = running.shutdown_network(&[&prepared], network).await; + return Err(error); + } + Ok(changed) if changed || !network.available() => { + result = Err(io::Error::new( + io::ErrorKind::Interrupted, + "underlay changed during recovery", + )); + } + Ok(_) => {} + } + match result { + Ok(()) if pending_reload => { + pending_reload = false; + false + } + Ok(()) => { + if let Some(control) = control.as_ref() + && let Err(error) = control.runtime.resume().await + { + let _ = running.shutdown_network(&[&prepared], network).await; + return Err(error); + } + prepared.tcp.activate(); + reconnecting = false; + println!("Network reconnected"); + continue; + } + Err(error) => { + log::warn!("Network recovery deferred: {error}"); + if let Err(error) = running + .pause_network(&mut prepared, control.as_ref(), &lifetime) + .await + { + let _ = running.shutdown_network(&[&prepared], network).await; + return Err(error); + } + retry_at = tokio::time::Instant::now() + Duration::from_secs(2); + continue; + } + } + } Err(error) => { let cleanup = running.shutdown_network(&[&prepared], network).await; return Err(io::Error::other(format!( @@ -914,15 +1087,31 @@ async fn run_inner( } }; if debounce { - tokio::select! { + let settled = tokio::select! { biased; _ = shutdown.recv() => return running.shutdown_network(&[&prepared], network).await, error = running.tun_failure() => { let _ = running.shutdown_network(&[&prepared], network).await; return Err(error); }, - _ = reload.recv() => {} - _ = watches.debounce(&mut changes) => {} + result = wait_with_reconciliation(async { + tokio::select! { + _ = reload.recv() => {}, + _ = watches.debounce(&mut changes) => {}, + } + }, network) => result, + }; + match settled { + Ok(Some(())) => {} + Ok(None) => { + pending_reload = true; + force_network_reset = true; + continue; + } + Err(error) => { + let _ = running.shutdown_network(&[&prepared], network).await; + return Err(error); + } } } while changes.try_recv().is_ok() {} @@ -934,6 +1123,7 @@ async fn run_inner( let mut discovered = |paths: &[PathBuf], recovery: &[PathBuf], root: Option<&Path>| { watches.discovered(paths, recovery, root) }; + let topology = network.topology(); let candidate = tokio::select! { biased; _ = shutdown.recv() => return running.shutdown_network(&[&prepared], network).await, @@ -941,10 +1131,22 @@ async fn run_inner( let _ = running.shutdown_network(&[&prepared], network).await; return Err(error); }, - result = async { + result = wait_with_reconciliation(async { let configs = config::load_config_files(&paths).await?; - prepare_tracking(&paths, configs, Some(&prepared.dns), Some(&prepared), &mut discovered, control, network).await - } => result, + lifetime.scope(prepare_tracking(&paths, configs, Some(&prepared.dns), Some(&prepared), &mut discovered, control, &topology)).await + }, network) => result, + }; + let candidate = match candidate { + Ok(Some(candidate)) => candidate, + Ok(None) => { + pending_reload = true; + force_network_reset = true; + continue; + } + Err(error) => { + let _ = running.shutdown_network(&[&prepared], network).await; + return Err(error); + } }; let mut candidate = match candidate { Ok(candidate) => candidate, @@ -993,15 +1195,31 @@ async fn run_inner( let result = tokio::select! { biased; _ = shutdown.recv() => return running.shutdown_network(&[&prepared, &candidate], network).await, - result = running.activate(&mut candidate, control.as_ref()) => result, + result = lifetime.scope(running.activate(&mut candidate, control.as_ref())) => result, }; if let Err(error) = result { let rollback = tokio::select! { biased; _ = shutdown.recv() => return running.shutdown_network(&[&prepared, &candidate], network).await, - result = running.rollback(&mut prepared, &mut candidate, control.as_ref()) => result, + result = lifetime.scope(running.rollback(&mut prepared, &mut candidate, control.as_ref())) => result, }; if let Err(rollback) = rollback { + if reconnecting { + eprintln!( + "Reload rejected; previous servers unavailable, recovery continues: {error}; rollback: {rollback}" + ); + if let Err(error) = running + .pause_network(&mut prepared, control.as_ref(), &lifetime) + .await + { + let _ = running + .shutdown_network(&[&prepared, &candidate], network) + .await; + return Err(error); + } + retry_at = tokio::time::Instant::now() + Duration::from_secs(2); + continue; + } let cleanup = running .shutdown_network(&[&prepared, &candidate], network) .await; @@ -1010,11 +1228,19 @@ async fn run_inner( ))); } eprintln!("Reload rejected; previous servers restored: {error}"); + if reconnecting { + reconnecting = false; + println!("Network reconnected"); + } continue; } watches.activated(&candidate.validated.reload_files); let mut retired = std::mem::replace(&mut prepared, candidate); retired.certificates.shutdown().await; + if reconnecting { + reconnecting = false; + println!("Network reconnected"); + } println!("Servers ready"); } } @@ -1022,15 +1248,17 @@ async fn run_inner( async fn wait_with_reconciliation( change: impl std::future::Future, network: &mut crate::desktop::NetworkSession, -) -> io::Result { +) -> io::Result> { tokio::pin!(change); loop { tokio::select! { biased; - result = &mut change => return Ok(result), + result = &mut change => return Ok(Some(result)), _ = network.changed() => { // Retain in-flight refreshes and finish network mutations before handling reload/shutdown. - network.reconcile().await?; + if network.reconcile().await? { + return Ok(None); + } } } } diff --git a/src/process/tests.rs b/src/process/tests.rs index 2847a9a4..9c0a97e8 100644 --- a/src/process/tests.rs +++ b/src/process/tests.rs @@ -37,7 +37,7 @@ async fn ntp_initial_sync_precedes_deferred_certificate_binds() { Some(&previous), &mut |_, _, _| {}, &mut None, - &crate::desktop::NetworkSession::default(), + &crate::desktop::NetworkSession::default().topology(), ) .await .unwrap(); @@ -98,7 +98,7 @@ async fn network_reconciliation_does_not_cancel_inflight_refresh() { request.push(stream.read_u8().await.unwrap()); } for _ in 0..3 { - events.send(()).await.unwrap(); + events.send(None).await.unwrap(); reconciled.recv().await.unwrap(); } stream.write_all(format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n", body.len()).as_bytes()).await.unwrap(); @@ -106,13 +106,363 @@ async fn network_reconciliation_does_not_cancel_inflight_refresh() { std::future::pending::<()>().await; }; tokio::select! { - result = wait_with_reconciliation(remote_rule_change(&mut prepared), &mut network) => result.unwrap(), + result = wait_with_reconciliation(remote_rule_change(&mut prepared), &mut network) => result.unwrap().unwrap(), _ = fixture => unreachable!(), } assert_eq!(std::fs::read(path).unwrap(), body); }).await.unwrap(); } +#[tokio::test] +async fn underlay_change_interrupts_the_wait_without_consuming_a_reload() { + let (events, receiver) = mpsc::channel(1); + let (ack, _reconciled) = mpsc::channel(1); + let mut network = crate::desktop::NetworkSession::with_test_events(receiver, ack); + events.send(Some(true)).await.unwrap(); + assert!( + wait_with_reconciliation(std::future::pending::<()>(), &mut network) + .await + .unwrap() + .is_none() + ); +} + +#[cfg(unix)] +#[tokio::test] +async fn reconnect_loop_defers_offline_reload_and_accepts_repairs_after_failed_start() { + const NAME: &str = "process::tests::reconnect_loop_defers_offline_reload_and_accepts_repairs_after_failed_start"; + const CHILD: &str = "SHOES_RECONNECT_LOOP_TEST"; + if std::env::var_os(CHILD).is_none() { + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", NAME, "--nocapture"]) + .env(CHILD, "1") + .output() + .unwrap(); + shoes_test_support::isolated::assert_one_test_passed(&output); + assert!( + String::from_utf8_lossy(&output.stderr) + .contains("Reload rejected; previous servers unavailable, recovery continues:") + ); + return; + } + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + async fn greet(address: std::net::SocketAddr) -> tokio::net::TcpStream { + loop { + if let Ok(mut stream) = tokio::net::TcpStream::connect(address).await { + stream.write_all(&[5, 1, 0]).await.unwrap(); + let mut reply = [0; 2]; + stream.read_exact(&mut reply).await.unwrap(); + assert_eq!(reply, [5, 0]); + return stream; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + } + + tokio::time::timeout(Duration::from_secs(20), async { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.json"); + let unix = directory.path().join("listener.sock"); + let first = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let second = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let address = first.local_addr().unwrap(); + let added = second.local_addr().unwrap(); + let mut config = vec![ + serde_json::json!({"address":address.to_string(), "protocol":{"type":"socks"}}), + serde_json::json!({"path":unix, "protocol":{"type":"socks"}}), + ]; + std::fs::write(&path, serde_json::to_vec(&config).unwrap()).unwrap(); + drop((first, second)); + let (events, receiver) = mpsc::channel(1); + let (ack, mut reconciled) = mpsc::channel(1); + let network = crate::desktop::NetworkSession::with_test_events(receiver, ack); + let process = crate::desktop::TEST_NETWORK.scope( + std::cell::RefCell::new(Some(network)), + Box::pin(run(vec![path.to_string_lossy().into_owned()], false, true)), + ); + tokio::pin!(process); + let fixture = async { + let mut live = greet(address).await; + events.send(Some(false)).await.unwrap(); + reconciled.recv().await.unwrap(); + assert_eq!(live.read(&mut [0]).await.unwrap(), 0); + config.push( + serde_json::json!({"address":added.to_string(), "protocol":{"type":"socks"}}), + ); + std::fs::write(&path, serde_json::to_vec(&config).unwrap()).unwrap(); + assert_eq!( + unsafe { libc::kill(std::process::id() as i32, libc::SIGHUP) }, + 0 + ); + tokio::time::sleep(Duration::from_millis(100)).await; + assert!(tokio::net::TcpStream::connect(added).await.is_err()); + events.send(Some(true)).await.unwrap(); + let mut live = greet(added).await; + + events.send(Some(false)).await.unwrap(); + assert_eq!(live.read(&mut [0]).await.unwrap(), 0); + std::fs::remove_file(&unix).unwrap(); + std::fs::create_dir(&unix).unwrap(); + events.send(Some(true)).await.unwrap(); + // A non-AddrInUse startup error must keep retrying while reload remains usable. + tokio::time::sleep(Duration::from_millis(2300)).await; + let mut queued = tokio::net::TcpStream::connect(address).await.unwrap(); + queued.write_all(&[5, 1, 0]).await.unwrap(); + assert!( + tokio::time::timeout(Duration::from_millis(100), queued.read_u8()) + .await + .is_err() + ); + assert_eq!( + unsafe { libc::kill(std::process::id() as i32, libc::SIGHUP) }, + 0 + ); + // An unchanged reload fails activation and rollback, but must not end recovery. + assert!( + tokio::time::timeout(Duration::from_secs(1), queued.read_u8()) + .await + .is_err() + ); + config.remove(1); + std::fs::write(&path, serde_json::to_vec(&config).unwrap()).unwrap(); + assert_eq!( + unsafe { libc::kill(std::process::id() as i32, libc::SIGHUP) }, + 0 + ); + assert_eq!(queued.read_u8().await.unwrap(), 5); + assert_eq!(queued.read_u8().await.unwrap(), 0); + assert!(unix.is_dir()); + let mut live = greet(added).await; + events.send(Some(false)).await.unwrap(); + assert_eq!(live.read(&mut [0]).await.unwrap(), 0); + assert_eq!( + unsafe { libc::kill(std::process::id() as i32, libc::SIGTERM) }, + 0 + ); + }; + tokio::select! { + result = &mut process => panic!("process exited before recovery: {result:?}"), + _ = fixture => {}, + } + process.await.unwrap(); + }) + .await + .unwrap(); +} + +#[tokio::test] +async fn reconnect_restarts_certificate_workers_retired_by_a_rejected_candidate() { + const NAME: &str = + "process::tests::reconnect_restarts_certificate_workers_retired_by_a_rejected_candidate"; + const CHILD: &str = "SHOES_RECONNECT_CERTIFICATE_TEST"; + if std::env::var_os(CHILD).is_none() { + shoes_test_support::isolated::run_isolated(NAME, CHILD, "1"); + return; + } + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + async fn challenge(address: std::net::SocketAddr) { + tokio::time::timeout(Duration::from_secs(2), async { + let mut stream = tokio::net::TcpStream::connect(address).await.unwrap(); + stream + .write_all(b"GET / HTTP/1.1\r\nHost: proxy.test\r\n\r\n") + .await + .unwrap(); + let mut response = String::new(); + stream.read_to_string(&mut response).await.unwrap(); + assert!(response.starts_with("HTTP/1.1 404 ")); + }) + .await + .unwrap(); + } + + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.json"); + let reserved = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let address = reserved.local_addr().unwrap(); + std::fs::write(&path, serde_json::json!([ + {"certificate_source":"managed", "type":"acme", "domains":["proxy.test"], + "directory_url":"http://127.0.0.1:9/directory", "state_dir":directory.path().join("state"), + "challenge":{"type":"http01", "listen":[address.to_string()]}}, + {"address":"127.0.0.1:0", "protocol":{"type":"socks"}} + ]).to_string()).unwrap(); + drop(reserved); + let old = crate::network_tasks::NetworkLifetime::default(); + let mut prepared = old + .scope(prepare(&[path.to_string_lossy().into_owned()], None)) + .await + .unwrap(); + let mut running = RunningServers::default(); + old.scope(running.activate(&mut prepared, None)) + .await + .unwrap(); + challenge(address).await; + + // Candidate retirement can stop the accepted certificate workers before rollback fails. + prepared + .certificates + .retire_unused(&Default::default()) + .await; + running + .pause_network(&mut prepared, None, &old) + .await + .unwrap(); + let fresh = crate::network_tasks::NetworkLifetime::default(); + fresh + .scope(running.reconnect( + &mut prepared, + &mut crate::desktop::NetworkSession::default(), + )) + .await + .unwrap(); + challenge(address).await; + running.shutdown(&[&prepared]).await.unwrap(); +} + +#[tokio::test] +async fn network_reset_keeps_controller_generation_and_explicit_selection() { + const CHILD: &str = "SHOES_RECONNECT_CONTROL_TEST"; + if std::env::var_os(CHILD).is_none() { + shoes_test_support::isolated::run_isolated( + "process::tests::network_reset_keeps_controller_generation_and_explicit_selection", + CHILD, + "1", + ); + return; + } + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.json"); + std::fs::write( + &path, + serde_json::json!([ + {"client_group":"chosen", "type":"selector", "client_chains":[ + {"chain":[{"name":"first", "protocol":{"type":"direct"}}]}, + {"chain":[{"name":"second", "protocol":{"type":"direct"}}]} + ]}, + {"address":"127.0.0.1:0", "protocol":{"type":"socks"}} + ]) + .to_string(), + ) + .unwrap(); + let paths = [path.to_string_lossy().into_owned()]; + let configs = config::load_config_files(&paths).await.unwrap(); + let mut control = None; + let mut network = crate::desktop::NetworkSession::default(); + let old = crate::network_tasks::NetworkLifetime::default(); + let mut prepared = old + .scope(prepare_tracking( + &paths, + configs, + None, + None, + &mut |_, _, _| {}, + &mut control, + &network.topology(), + )) + .await + .unwrap(); + let mut control = control.unwrap(); + let generation = prepared.generation.clone().unwrap(); + control.runtime.publish(generation.clone()).await.unwrap(); + control.runtime.resume().await.unwrap(); + control + .runtime + .select(generation.id, "chosen".into(), "second".into()) + .await + .unwrap(); + let selection = generation.catalog.selection.snapshot(); + let mut running = RunningServers::default(); + old.scope(running.start(&mut prepared)).await.unwrap(); + running + .pause_network(&mut prepared, Some(&control), &old) + .await + .unwrap(); + let fresh = crate::network_tasks::NetworkLifetime::default(); + fresh + .scope(running.reconnect(&mut prepared, &mut network)) + .await + .unwrap(); + control.runtime.resume().await.unwrap(); + prepared.tcp.activate(); + assert!(std::sync::Arc::ptr_eq( + &generation, + &control.runtime.generation().unwrap() + )); + assert!(std::sync::Arc::ptr_eq( + &selection, + &generation.catalog.selection.snapshot() + )); + assert_eq!(selection["chosen"], "second"); + running + .pause_network(&mut prepared, Some(&control), &fresh) + .await + .unwrap(); + running.shutdown(&[&prepared]).await.unwrap(); + control.shutdown().await; +} + +#[tokio::test] +async fn reconnect_uses_memory_and_cancels_current_and_draining_connections() { + const NAME: &str = + "process::tests::reconnect_uses_memory_and_cancels_current_and_draining_connections"; + const CHILD: &str = "SHOES_RECONNECT_PROCESS_TEST"; + if std::env::var_os(CHILD).is_none() { + shoes_test_support::isolated::run_isolated(NAME, CHILD, "1"); + return; + } + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + tokio::time::timeout(Duration::from_secs(15), async { + let echo = shoes_test_support::test_servers::start_tcp_stream_echo_server("127.0.0.1", 0).await.unwrap(); + let reserved = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let listen = reserved.local_addr().unwrap(); + let reserved_socks = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let socks = reserved_socks.local_addr().unwrap(); + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.json"); + std::fs::write(&path, serde_json::json!([ + {"global_settings": {"reload_grace_secs":300}}, + {"address":listen.to_string(), "protocol":{"type":"forward", "target":echo.local_addr().to_string()}}, + {"address":socks.to_string(), "protocol":{"type":"socks"}} + ]).to_string()).unwrap(); + drop((reserved, reserved_socks)); + let mut lifetime = crate::network_tasks::NetworkLifetime::default(); + let mut prepared = lifetime.scope(prepare(&[path.to_string_lossy().into_owned()], None)).await.unwrap(); + let mut running = RunningServers::default(); + lifetime.scope(running.start(&mut prepared)).await.unwrap(); + let mut draining = tokio::net::TcpStream::connect(listen).await.unwrap(); + draining.write_all(b"before").await.unwrap(); + let mut bytes = [0; 6]; + draining.read_exact(&mut bytes).await.unwrap(); + running.stop().await; + lifetime.scope(running.start(&mut prepared)).await.unwrap(); + draining.write_all(b"reload").await.unwrap(); + draining.read_exact(&mut bytes).await.unwrap(); + assert_eq!(&bytes, b"reload"); + let mut pending = tokio::net::TcpStream::connect(socks).await.unwrap(); + pending.write_all(&[5]).await.unwrap(); + let mut current = tokio::net::TcpStream::connect(listen).await.unwrap(); + current.write_all(b"active").await.unwrap(); + current.read_exact(&mut bytes).await.unwrap(); + std::fs::write(&path, "invalid: [").unwrap(); + running.pause_network(&mut prepared, None, &lifetime).await.unwrap(); + for stream in [&mut draining, &mut pending, &mut current] { + let result = tokio::time::timeout(Duration::from_secs(1), stream.read(&mut [0])).await + .expect("network reset waited for ordinary reload grace"); + assert!(matches!(result, Ok(0)) || result.is_err(), "{result:?}"); + } + lifetime = crate::network_tasks::NetworkLifetime::default(); + lifetime.scope(running.reconnect(&mut prepared, &mut crate::desktop::NetworkSession::default())) + .await.unwrap(); + prepared.tcp.activate(); + let mut fresh = tokio::net::TcpStream::connect(listen).await.unwrap(); + fresh.write_all(b"fresh!").await.unwrap(); + fresh.read_exact(&mut bytes).await.unwrap(); + assert_eq!(&bytes, b"fresh!"); + running.pause_network(&mut prepared, None, &lifetime).await.unwrap(); + }).await.unwrap(); +} + async fn load_validated( paths: &[String], discovered: &mut DiscoverDependencies<'_>, diff --git a/src/quic_client/hop.rs b/src/quic_client/hop.rs index 5c8c87ce..2d7ad1a3 100644 --- a/src/quic_client/hop.rs +++ b/src/quic_client/hop.rs @@ -74,16 +74,18 @@ impl HoppingSocket { task: parking_lot::Mutex::new(None), }); let weak = Arc::downgrade(&socket); - *socket.task.lock() = Some(AbortOnDropHandle::new(tokio::spawn(async move { - loop { - let seconds = rand::rng().random_range(min..=max); - tokio::time::sleep(Duration::from_secs(seconds)).await; - let Some(socket) = weak.upgrade() else { return }; - if let Err(error) = socket.hop() { - log::warn!("HY2 port hop failed; keeping current socket: {error}"); + *socket.task.lock() = Some(AbortOnDropHandle::new(crate::network_tasks::spawn( + async move { + loop { + let seconds = rand::rng().random_range(min..=max); + tokio::time::sleep(Duration::from_secs(seconds)).await; + let Some(socket) = weak.upgrade() else { return }; + if let Err(error) = socket.hop() { + log::warn!("HY2 port hop failed; keeping current socket: {error}"); + } } - } - }))); + }, + ))); Ok(socket) } diff --git a/src/quic_client/hysteria2.rs b/src/quic_client/hysteria2.rs index 222922e0..2662a3ec 100644 --- a/src/quic_client/hysteria2.rs +++ b/src/quic_client/hysteria2.rs @@ -79,7 +79,7 @@ pub(super) async fn authenticate( .await .map_err(io::Error::other)?; let hold = sender.clone(); - let task = AbortOnDropHandle::new(tokio::spawn(async move { + let task = AbortOnDropHandle::new(crate::network_tasks::spawn(async move { // Both the driver and at least one SendRequest must survive authentication. let _hold = hold; let _ = driver.wait_idle().await; diff --git a/src/quic_client/lifecycle_tests.rs b/src/quic_client/lifecycle_tests.rs index 2f92225c..d80ec7b5 100644 --- a/src/quic_client/lifecycle_tests.rs +++ b/src/quic_client/lifecycle_tests.rs @@ -84,6 +84,29 @@ fn protocol() -> Protocol { }) } +#[tokio::test] +async fn network_reset_cancels_warmed_pool_and_retained_connection_handles() { + tokio::time::timeout(Duration::from_secs(5), async { + let network = crate::network_tasks::NetworkLifetime::default(); + let (server, dialer) = network.scope(async { endpoints() }).await; + let pool = Pool::default(); + let protocol = protocol(); + let (session, peer) = + tokio::join!(network.scope(pool.get(dialer, &protocol, true)), async { + server.accept().await.unwrap().await.unwrap() + },); + let session = session.unwrap(); + let mut auth = peer.accept_uni().await.unwrap(); + assert_eq!(auth.read_to_end(50).await.unwrap().len(), 50); + network.cancel(); + network.wait().await; + session.connection().closed().await; + assert!(session.connection().open_bi().await.is_err()); + }) + .await + .unwrap(); +} + #[tokio::test] async fn exhausted_udp_ids_replace_the_connection_without_migrating_live_associations() { use crate::tcp::proxy_connector::ProxyConnector; diff --git a/src/quic_client/session.rs b/src/quic_client/session.rs index 130e266a..88abe0e8 100644 --- a/src/quic_client/session.rs +++ b/src/quic_client/session.rs @@ -71,7 +71,7 @@ impl Pool { let (sender, result) = watch::channel(None); let slots = Arc::downgrade(&self.slots); let key = dialer.route_key().clone(); - let task = tokio::spawn(async move { + let task = crate::network_tasks::spawn(async move { let result = tokio::time::timeout(SETUP_TIMEOUT, Generation::open(dialer, protocol)) .await .unwrap_or_else(|_| { @@ -162,28 +162,31 @@ impl Generation { generation.udp_enabled = config.udp_enabled; generation .tasks - .push(AbortOnDropHandle::new(tokio::spawn(tuic::heartbeat( - generation.connection().clone(), - config.heartbeat_interval_secs, - )))); + .push(AbortOnDropHandle::new(crate::network_tasks::spawn( + tuic::heartbeat( + generation.connection().clone(), + config.heartbeat_interval_secs, + ), + ))); let (sender, receiver) = tokio::sync::mpsc::channel(64); generation.dissociate = Some(sender); generation .tasks - .push(AbortOnDropHandle::new(tokio::spawn(tuic::dissociate_loop( - commands.clone(), - receiver, - )))); + .push(AbortOnDropHandle::new(crate::network_tasks::spawn( + tuic::dissociate_loop(commands.clone(), receiver), + ))); generation.commands = Some(commands); } } generation .tasks - .push(AbortOnDropHandle::new(tokio::spawn(udp::receive( - generation.connection().clone(), - generation.registry.clone(), - matches!(generation.protocol, Protocol::Tuic(_)), - )))); + .push(AbortOnDropHandle::new(crate::network_tasks::spawn( + udp::receive( + generation.connection().clone(), + generation.registry.clone(), + matches!(generation.protocol, Protocol::Tuic(_)), + ), + ))); generation.initialized = true; Ok(generation) } @@ -198,7 +201,7 @@ impl Drop for Generation { self.registry.lock().close(); if self.initialized && self.connection().close_reason().is_none() - && let Ok(runtime) = tokio::runtime::Handle::try_current() + && tokio::runtime::Handle::try_current().is_ok() { // Retiring a route must not discard FIN/data already accepted by // Quinn. One bounded drain task per generation retains its endpoint @@ -206,7 +209,7 @@ impl Drop for Generation { let connection = self.connection().clone(); let endpoint = self.carrier._endpoint.clone(); let tasks = std::mem::take(&mut self.tasks); - runtime.spawn(async move { + crate::network_tasks::spawn(async move { let _ = tokio::time::timeout(Duration::from_secs(3), connection.closed()).await; connection.close(0u32.into(), b"client retired"); drop((tasks, endpoint)); diff --git a/src/quic_client/tuic.rs b/src/quic_client/tuic.rs index 5b32db85..b2bfd81a 100644 --- a/src/quic_client/tuic.rs +++ b/src/quic_client/tuic.rs @@ -46,7 +46,7 @@ impl CommandWriter { // Dropping a partial command sends FIN, which peers treat as a fatal // protocol error. Once admitted, finish independently of caller cancellation. // The permit bounds detached writers; connection closure releases them. - tokio::spawn(async move { + crate::network_tasks::spawn(async move { let _permit = permit; stream.write_all(&bytes).await?; stream.finish()?; diff --git a/src/quic_client/udp.rs b/src/quic_client/udp.rs index 9044647b..0c3f5e7e 100644 --- a/src/quic_client/udp.rs +++ b/src/quic_client/udp.rs @@ -126,7 +126,7 @@ pub(super) async fn receive( } stream = connection.accept_uni(), if is_tuic && streams.len() < QUEUE_SIZE => { let Ok(mut stream) = stream else { break }; - streams.spawn(async move { + crate::network_tasks::spawn_in(&mut streams, async move { tokio::time::timeout( Duration::from_secs(5), stream.read_to_end(tuic::MAX_PACKET), diff --git a/src/quic_endpoint.rs b/src/quic_endpoint.rs index d9937dbd..06be0674 100644 --- a/src/quic_endpoint.rs +++ b/src/quic_endpoint.rs @@ -19,7 +19,8 @@ const LISTENER_CLOSE_GRACE: Duration = Duration::from_secs(1); #[derive(Debug)] struct ListenerRuntime { endpoint_started: AtomicBool, - retired: CancellationToken, + retired: Option, + network: Option, } impl Runtime for ListenerRuntime { @@ -31,12 +32,29 @@ impl Runtime for ListenerRuntime { // Quinn 0.11 spawns the endpoint driver synchronously before exposing the // endpoint. Later spawns are connection drivers; recheck this on upgrades. if self.endpoint_started.swap(true, Ordering::Relaxed) { - quinn::TokioRuntime.spawn(future); + match &self.network { + Some(network) => { + network.spawn_completion(future); + } + None => quinn::TokioRuntime.spawn(future), + } return; } - let mut driver = AbortOnDropHandle::new(tokio::spawn(future)); - let retired = self.retired.clone(); - tokio::spawn(async move { + if self.retired.is_none() { + match &self.network { + Some(network) => { + network.spawn(future); + } + None => quinn::TokioRuntime.spawn(future), + } + return; + } + let mut driver = AbortOnDropHandle::new(match &self.network { + Some(network) => network.spawn(future), + None => tokio::spawn(future), + }); + let retired = self.retired.clone().unwrap(); + let supervisor = async move { tokio::select! { _ = &mut driver => {}, _ = async { @@ -46,7 +64,15 @@ impl Runtime for ListenerRuntime { } // Dropping the endpoint driver closes its connection-event channels. // Connection drivers then terminate and wake application waiters. - }); + }; + match &self.network { + Some(network) => { + network.spawn(supervisor); + } + None => { + tokio::spawn(supervisor); + } + } } fn wrap_udp_socket(&self, socket: std::net::UdpSocket) -> io::Result> { @@ -116,15 +142,24 @@ impl QuicEndpoint { let retired = CancellationToken::new(); let runtime = ListenerRuntime { endpoint_started: AtomicBool::new(false), - retired: retired.clone(), + retired: Some(retired.clone()), + network: crate::network_tasks::NetworkTasks::current(), }; (Arc::new(runtime), Some(retired.drop_guard())) } else { - (Arc::new(quinn::TokioRuntime), None) + ( + Arc::new(ListenerRuntime { + endpoint_started: AtomicBool::new(false), + retired: None, + network: crate::network_tasks::NetworkTasks::current(), + }), + None, + ) }; let socket = MemorySocket { inner: socket, endpoint_memory: None, + cancelled: None, _release: release, }; quinn::Endpoint::new_with_abstract_socket( @@ -222,6 +257,9 @@ pub(crate) fn socket_with_memory( Ok(Arc::new(MemorySocket { inner: quinn::TokioRuntime.wrap_udp_socket(socket)?, endpoint_memory: Some(memory), + // Hickory owns its Quinn runtime; retire its driver through the socket instead. + cancelled: crate::network_tasks::NetworkTasks::current() + .map(|network| parking_lot::Mutex::new(Box::pin(network.cancelled.cancelled_owned()))), _release: None, })) } @@ -230,6 +268,8 @@ pub(crate) fn socket_with_memory( struct MemorySocket { inner: Arc, endpoint_memory: Option, + cancelled: + Option>>>, // Field order signals retirement only after the native socket has been dropped. _release: Option>, } @@ -265,6 +305,14 @@ impl AsyncUdpSocket for MemorySocket { bufs: &mut [IoSliceMut<'_>], meta: &mut [quinn::udp::RecvMeta], ) -> Poll> { + if let Some(cancelled) = &self.cancelled + && cancelled.lock().as_mut().poll(cx).is_ready() + { + return Poll::Ready(Err(io::Error::new( + io::ErrorKind::Interrupted, + "DNS QUIC network retired", + ))); + } self.inner.poll_recv(cx, bufs, meta) } diff --git a/src/quic_endpoint/tests.rs b/src/quic_endpoint/tests.rs index b003bcc5..973a4f38 100644 --- a/src/quic_endpoint/tests.rs +++ b/src/quic_endpoint/tests.rs @@ -11,7 +11,8 @@ async fn listener_runtime_cancels_only_the_endpoint_driver_after_grace() { let retired = CancellationToken::new(); let runtime = ListenerRuntime { endpoint_started: AtomicBool::new(false), - retired: retired.clone(), + retired: Some(retired.clone()), + network: None, }; let endpoint = Arc::new(()); let owned_endpoint = endpoint.clone(); @@ -48,7 +49,8 @@ async fn listener_runtime_reaps_a_completed_driver_and_its_watcher() { let baseline = metrics.num_alive_tasks(); let runtime = ListenerRuntime { endpoint_started: AtomicBool::new(false), - retired: CancellationToken::new(), + retired: Some(CancellationToken::new()), + network: None, }; runtime.spawn(Box::pin(async {})); for _ in 0..4 { @@ -57,6 +59,58 @@ async fn listener_runtime_reaps_a_completed_driver_and_its_watcher() { assert_eq!(metrics.num_alive_tasks(), baseline); } +#[tokio::test] +async fn network_reset_retires_external_quinn_runtime_through_dns_socket() { + timeout(Duration::from_secs(3), async { + let network = crate::network_tasks::NetworkLifetime::default(); + let (client_config, server_config) = configs(); + let server = + quinn::Endpoint::server(server_config, "127.0.0.1:0".parse().unwrap()).unwrap(); + let budget = Arc::new(Budget::new(Some(1))); + let socket = network + .scope(async { + socket_with_memory( + std::net::UdpSocket::bind("127.0.0.1:0").unwrap(), + budget.acquire(1).unwrap(), + ) + .unwrap() + }) + .await; + let mut client = quinn::Endpoint::new_with_abstract_socket( + quinn::EndpointConfig::default(), + None, + socket, + Arc::new(quinn::TokioRuntime), + ) + .unwrap(); + client.set_default_client_config(client_config); + let (connection, peer) = tokio::join!( + async { + client + .connect(server.local_addr().unwrap(), "localhost") + .unwrap() + .await + .unwrap() + }, + async { server.accept().await.unwrap().await.unwrap() }, + ); + connection + .send_datagram(Bytes::from_static(b"before")) + .unwrap(); + assert_eq!(peer.read_datagram().await.unwrap(), b"before"[..]); + network.cancel(); + connection.closed().await; + assert!( + connection + .send_datagram(Bytes::from_static(b"after")) + .is_err() + ); + drop((connection, client, peer, server)); + }) + .await + .unwrap(); +} + #[tokio::test] async fn endpoint_cancellation_wakes_retained_handles_before_releasing_memory() { let (client_config, server_config) = configs(); @@ -119,6 +173,7 @@ async fn socket_release_waits_for_delegated_poller() { let socket = Arc::new(MemorySocket { inner: quinn::TokioRuntime.wrap_udp_socket(socket).unwrap(), endpoint_memory: None, + cancelled: None, _release: Some(release), }); let budget = Arc::new(Budget::new(Some(1))); diff --git a/src/quic_server.rs b/src/quic_server.rs index bda15a52..d148087e 100644 --- a/src/quic_server.rs +++ b/src/quic_server.rs @@ -34,7 +34,7 @@ async fn start_quic_server( )? { let resolver = resolver.clone(); let server_handler = server_handler.clone(); - let join_handle = tokio::spawn(async move { + let join_handle = crate::network_tasks::spawn(async move { let mut tasks = crate::listener_tasks::ListenerTasks::immediate(); loop { let conn = tokio::select! { @@ -104,7 +104,7 @@ async fn process_connection( let cloned_resolver = resolver.clone(); let cloned_handler = server_handler.clone(); let peer = connection.remote_address(); - tasks.spawn(async move { + crate::network_tasks::spawn_in(&mut tasks, async move { if let Err(e) = process_streams(cloned_resolver, cloned_handler, stream, peer).await { error!("Failed to process streams: {e}"); } @@ -260,7 +260,7 @@ pub async fn start_quic_servers( } for endpoint in endpoints { let service = service.clone(); - handles.push(tokio::spawn(async move { + handles.push(crate::network_tasks::spawn(async move { let mut tasks = crate::listener_tasks::ListenerTasks::immediate(); loop { let incoming = tokio::select! { diff --git a/src/routing/udp_relay.rs b/src/routing/udp_relay.rs index 669d1d51..8048bc14 100644 --- a/src/routing/udp_relay.rs +++ b/src/routing/udp_relay.rs @@ -44,7 +44,7 @@ impl UdpRelay { output, peer, }; - let task = tokio::spawn(async move { + let task = crate::network_tasks::spawn(async move { let _ = run_udp_routing( ServerStream::Targeted(Box::new(stream)), selector, @@ -200,6 +200,33 @@ mod tests { } } + #[tokio::test] + async fn network_reset_retires_relay_even_while_its_handle_survives() { + let network = crate::network_tasks::NetworkLifetime::default(); + let resolver: Arc = Arc::new(NoHostnameResolver); + let selector = Arc::new(ClientProxySelector::new(vec![ConnectRule::new( + vec![crate::address::NetLocationMask::from("0.0.0.0/0").unwrap()], + ConnectAction::new_allow( + None, + crate::tcp::chain_builder::build_direct_chain_group(resolver.clone()), + ), + )])); + let relay = network + .scope(async { UdpRelay::new(selector, resolver, None).unwrap() }) + .await; + network.cancel(); + network.wait().await; + assert!(relay.recv().await.is_err()); + assert!( + relay + .send_to( + Bytes::new(), + NetLocation::from_str("192.0.2.1:53", None).unwrap() + ) + .is_err() + ); + } + #[tokio::test] async fn owned_input_preserves_allocation_and_queue_drop_policy() { let (tx, mut input) = mpsc::channel(QUEUE_SIZE); diff --git a/src/socket_policy.rs b/src/socket_policy.rs index 53530e4b..e7502ab2 100644 --- a/src/socket_policy.rs +++ b/src/socket_policy.rs @@ -22,6 +22,21 @@ pub(crate) struct SocketPolicy(()); #[cfg(target_os = "linux")] impl SocketPolicy { + pub(crate) fn refresh_bootstrap(&self) -> io::Result<()> { + let endpoints = bootstrap_endpoints() + .ok_or_else(|| io::Error::other("managed socket policy disappeared"))?; + let bootstrap = crate::dns::controlled_bootstrap(&endpoints)?; + let old = { + let mut current = POLICY.write(); + let policy = current + .as_mut() + .ok_or_else(|| io::Error::other("managed socket policy disappeared"))?; + std::mem::replace(&mut policy.bootstrap, bootstrap) + }; + drop(old); + Ok(()) + } + pub(crate) fn install( interface: String, bypass_mark: u32, diff --git a/src/tcp/tcp_server.rs b/src/tcp/tcp_server.rs index 5312ac08..62e78c0c 100644 --- a/src/tcp/tcp_server.rs +++ b/src/tcp/tcp_server.rs @@ -556,7 +556,7 @@ pub(crate) async fn start_servers_with_listeners( let task = tokio_util::task::AbortOnDropHandle::new( start_tun_server(tun_config, resolver).await?, ); - Ok(vec![tokio::spawn(async move { + Ok(vec![crate::network_tasks::spawn(async move { if let Err(error) = task .await .unwrap_or_else(|error| Err(std::io::Error::other(error))) @@ -671,7 +671,7 @@ async fn start_tcp_servers( let tcp_config = tcp_config.clone(); let resolver = resolver.clone(); let admission = sockets.admission.clone(); - handles.push(tokio::spawn(async move { + handles.push(crate::network_tasks::spawn(async move { // Uncommitted flows must not outlive rollback through graceful draining. if let Some(admission) = admission { admission.cancelled().await; @@ -699,7 +699,7 @@ async fn start_tcp_servers( create_tcp_server_handler(protocol, &client_proxy_selector, &resolver, None)? .into(); let admission = sockets.admission.clone(); - let handle = tokio::spawn(async move { + let handle = crate::network_tasks::spawn(async move { if let Some(admission) = admission { admission.cancelled().await; } diff --git a/src/tuic_server.rs b/src/tuic_server.rs index 020ba47a..96ff63bc 100644 --- a/src/tuic_server.rs +++ b/src/tuic_server.rs @@ -262,7 +262,7 @@ async fn run_bidirectional_loop( }; let client_proxy_selector = client_proxy_selector.clone(); let resolver = resolver.clone(); - tasks.spawn(async move { + crate::network_tasks::spawn_in(&mut tasks, async move { let _permit = permit; let result = async { let Some(stream) = reader.await? else { @@ -528,7 +528,7 @@ impl UdpSession { }; session.task = Some( - tokio::spawn(async move { + crate::network_tasks::spawn(async move { let result = if is_uni_stream { run_udp_remote_to_local_stream_loop( assoc_id, @@ -717,7 +717,7 @@ async fn run_unidirectional_loop( // Spawn a cleanup task for UDP sessions that terminates when connection closes let cleanup_session_map = udp_session_map.clone(); let cleanup_cancel_token = cancel_token.clone(); - tasks.spawn(async move { + crate::network_tasks::spawn_in(&mut tasks, async move { let mut interval = tokio::time::interval(Duration::from_secs(1)); loop { tokio::select! { @@ -781,7 +781,7 @@ async fn run_unidirectional_loop( let resolver = resolver.clone(); let udp_session_map = udp_session_map.clone(); let cancel_token = cancel_token.clone(); - tasks.spawn(async move { + crate::network_tasks::spawn_in(&mut tasks, async move { let _permit = permit; // Per TUIC protocol, each uni stream carries exactly ONE command. // The reference implementation (handle_stream.rs) handles one task per stream. @@ -1162,7 +1162,7 @@ pub async fn start_tuic_server( let client_proxy_selector = client_proxy_selector.clone(); let uuid = uuid.clone(); let password = password.clone(); - let join_handle = tokio::spawn(async move { + let join_handle = crate::network_tasks::spawn(async move { let mut tasks = crate::listener_tasks::ListenerTasks::immediate(); loop { let conn = tokio::select! { diff --git a/src/tun/mod.rs b/src/tun/mod.rs index a425bcc7..a367f78f 100644 --- a/src/tun/mod.rs +++ b/src/tun/mod.rs @@ -189,7 +189,7 @@ async fn run_tun_server_inner( let proxy_selector = proxy_selector.clone(); let resolver = resolver.clone(); - tasks.spawn(async move { + crate::network_tasks::spawn_in(&mut tasks, async move { info!("Starting TCP connection handler"); let mut connections = JoinSet::new(); loop { @@ -203,7 +203,7 @@ async fn run_tun_server_inner( let proxy_selector = proxy_selector.clone(); let resolver = resolver.clone(); - connections.spawn(async move { + crate::network_tasks::spawn_in(&mut connections, async move { let remote_addr = new_conn.remote_addr; let target = socket_addr_to_net_location(remote_addr); @@ -233,7 +233,7 @@ async fn run_tun_server_inner( let limits = config.resource_limits.clone(); let wake = tcp_stack.wake_handle(); - tasks.spawn(async move { + crate::network_tasks::spawn_in(&mut tasks, async move { handle_udp_packets( udp_from_stack_rx, udp_to_stack_tx, @@ -417,7 +417,7 @@ async fn start_tun( let (shutdown_tx, shutdown_rx) = oneshot::channel(); let (ready_tx, ready_rx) = oneshot::channel(); - let handle = tokio::spawn(async move { + let handle = crate::network_tasks::spawn(async move { let _keep_alive = shutdown_tx; let mut ready = Some(ready_tx); let result = run_tun_from_config_inner( diff --git a/src/tun/udp_manager.rs b/src/tun/udp_manager.rs index 790b7dd3..1999d765 100644 --- a/src/tun/udp_manager.rs +++ b/src/tun/udp_manager.rs @@ -262,7 +262,7 @@ impl TunUdpManager { let (tx, rx) = mpsc::channel(CHANNEL_SIZE); - let handle = tokio::spawn(session_task( + let handle = crate::network_tasks::spawn(session_task( peer_addr, rx, self.response_tx.clone(), @@ -389,7 +389,7 @@ async fn session_task( }; let (write_tx, write_rx) = mpsc::channel(CHANNEL_SIZE); - let handle = tokio::spawn(destination_task( + let handle = crate::network_tasks::spawn(destination_task( peer_addr, source_addr, stream, diff --git a/src/wireguard/detour.rs b/src/wireguard/detour.rs index bd1b05bb..a3450e2d 100644 --- a/src/wireguard/detour.rs +++ b/src/wireguard/detour.rs @@ -111,11 +111,14 @@ pub(super) fn start( incoming: incoming.clone(), drops: drops.clone(), }; - workers.spawn(crate::control::url_test::BACKGROUND.scope((), worker.run())); + crate::network_tasks::spawn_in( + &mut workers, + crate::control::url_test::BACKGROUND.scope((), worker.run()), + ); } drop(incoming); let shutdown = cancel.clone(); - tokio::spawn(async move { + crate::network_tasks::spawn(async move { tokio::select! { biased; _ = shutdown.cancelled() => {}, diff --git a/src/wireguard/mod.rs b/src/wireguard/mod.rs index 5a5a4b33..c337dce7 100644 --- a/src/wireguard/mod.rs +++ b/src/wireguard/mod.rs @@ -334,6 +334,22 @@ impl Prepared { } Ok(()) } + + pub(crate) async fn refresh_peers(&mut self) -> stdio::Result<()> { + let mut failure = None; + for endpoint in &mut self.endpoints { + for (peer, current) in endpoint.config.peers.iter().zip(&mut endpoint.peers) { + match PreparedPeer::resolve(&endpoint.config, peer, &endpoint.outer_resolver).await + { + Ok(refreshed) => *current = refreshed, + Err(error) => { + failure.get_or_insert(error); + } + } + } + } + failure.map_or(Ok(()), Err) + } } struct EndpointRuntime { @@ -351,10 +367,14 @@ pub(crate) struct Running { endpoints: Vec, } impl Running { - pub async fn stop(&mut self) { + pub(crate) fn cancel(&self) { for endpoint in &self.endpoints { endpoint.cancel.cancel(); } + } + + pub async fn stop(&mut self) { + self.cancel(); for mut endpoint in self.endpoints.drain(..) { let _ = (&mut endpoint.task).await; } @@ -482,7 +502,7 @@ impl PreparedEndpoint { let name = config.id.clone(); let cancel = CancellationToken::new(); let shutdown = cancel.clone(); - let task = tokio::spawn(async move { + let task = tokio::spawn(crate::network_tasks::inherit(async move { let mut tasks = JoinSet::new(); let mut inbound_udp_count = None; if let Some((selector, resolver)) = inbound { @@ -494,6 +514,7 @@ impl PreparedEndpoint { report.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); let failed = loop { tokio::select! { + biased; _ = shutdown.cancelled() => break false, _ = device.wait() => break true, _ = &mut stack.done => break true, @@ -531,7 +552,7 @@ impl PreparedEndpoint { stack.drops.load(Ordering::Relaxed) ); drop(stack); - }); + })); Ok(EndpointRuntime { cancel, task }) } } @@ -558,13 +579,13 @@ fn spawn_inbound( limits, ); let udp_count = manager.destination_counter(); - tasks.spawn(async move { + crate::network_tasks::spawn_in(tasks, async move { if let Err(error) = manager.run().await { log::warn!("WireGuard UDP forwarding stopped: {error}"); } }); let mut tcp = stack.tcp.take().unwrap(); - tasks.spawn(async move { + crate::network_tasks::spawn_in(tasks, async move { let mut flows = JoinSet::new(); loop { let connection = tokio::select! { @@ -576,7 +597,7 @@ fn spawn_inbound( }; let selector = selector.clone(); let resolver = resolver.clone(); - flows.spawn(async move { + crate::network_tasks::spawn_in(&mut flows, async move { log::debug!( "WireGuard TCP {} -> {}", connection.source, diff --git a/src/wireguard/tests.rs b/src/wireguard/tests.rs index 329d8994..7ee31276 100644 --- a/src/wireguard/tests.rs +++ b/src/wireguard/tests.rs @@ -106,6 +106,42 @@ async fn start_prepared(prepared: &Prepared) -> Running { running } +#[tokio::test] +async fn reconnect_resolves_peers_again_and_retries_without_replacing_connectors() { + #[derive(Debug)] + struct Answer(Option); + impl Resolver for Answer { + fn resolve_location(&self, _: &NetLocation) -> crate::resolver::ResolveFuture { + let address = self.0; + Box::pin(async move { + address.map(|address| vec![address]).ok_or_else(|| { + stdio::Error::new(stdio::ErrorKind::NotConnected, "network unavailable") + }) + }) + } + } + let mut configs = validate(&pair_config(51820)).unwrap(); + let mut prepared = Prepared::new(&mut configs).await.unwrap(); + let connector = prepared.endpoints[CLIENT].connector.clone(); + let original = prepared.endpoints[CLIENT].peers[0].endpoints.clone(); + prepared.endpoints[CLIENT].config.peers[0].endpoint = Some("peer.test:51820".into()); + prepared.endpoints[CLIENT].outer_resolver = Arc::new(Answer(None)); + let later = "192.0.2.10:51820".parse().unwrap(); + prepared.endpoints[SERVER].config.peers[0].endpoint = Some("later.test:51820".into()); + prepared.endpoints[SERVER].outer_resolver = Arc::new(Answer(Some(later))); + assert!(prepared.refresh_peers().await.is_err()); + assert_eq!(prepared.endpoints[CLIENT].peers[0].endpoints, original); + assert_eq!(prepared.endpoints[SERVER].peers[0].endpoints, [later]); + let replacement = "192.0.2.9:51820".parse().unwrap(); + prepared.endpoints[CLIENT].outer_resolver = Arc::new(Answer(Some(replacement))); + prepared.refresh_peers().await.unwrap(); + assert_eq!(prepared.endpoints[CLIENT].peers[0].endpoints, [replacement]); + assert!(Arc::ptr_eq( + &connector, + &prepared.endpoints[CLIENT].connector + )); +} + fn dns_answer(bytes: &[u8]) -> Vec { use hickory_resolver::proto::{ op::{Message, MessageType}, diff --git a/src/xhttp/server.rs b/src/xhttp/server.rs index 9e645d40..f5c31ae9 100644 --- a/src/xhttp/server.rs +++ b/src/xhttp/server.rs @@ -337,7 +337,7 @@ impl XhttpServer { continue; }; let state = self.state.clone(); - requests.spawn(async move { + crate::network_tasks::spawn_in(&mut requests, async move { let _request_permit = request_permit; let Ok(Ok((mut request, stream))) = tokio::time::timeout( Duration::from_secs(state.config.limits.setup_timeout_secs), @@ -1262,7 +1262,7 @@ impl State { continue; }; let state = self.clone(); - requests.spawn(async move { + crate::network_tasks::spawn_in(&mut requests, async move { let _request_permit = request_permit; let (parts, body) = request.into_parts(); let mut request = Request::from_parts(parts, ()); diff --git a/src/xhttp/stream.rs b/src/xhttp/stream.rs index e4f394db..fc3e6808 100644 --- a/src/xhttp/stream.rs +++ b/src/xhttp/stream.rs @@ -535,10 +535,12 @@ impl Tasks { tokio::select! { biased; _ = cancel.cancelled() => {}, _ = future => {} } }; if crate::control::url_test::is_background() { - self.tracker - .spawn(crate::control::url_test::BACKGROUND.scope((), task)); + crate::network_tasks::spawn( + self.tracker + .track_future(crate::control::url_test::BACKGROUND.scope((), task)), + ); } else { - self.tracker.spawn(task); + crate::network_tasks::spawn(self.tracker.track_future(task)); } } pub async fn wait(&self) { diff --git a/src/xhttp/tests.rs b/src/xhttp/tests.rs index 0a5c06c7..b4daba1e 100644 --- a/src/xhttp/tests.rs +++ b/src/xhttp/tests.rs @@ -10,6 +10,53 @@ use crate::{ tcp::{proxy_connector_impl::ProxyConnectorImpl, socket_connector_impl::SocketConnectorImpl}, }; +#[tokio::test] +async fn network_reset_retires_h2_and_h3_request_workers() { + tokio::time::timeout(Duration::from_secs(10), async { + let certificate = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap(); + let resolver: Arc = Arc::new(NativeResolver::new()); + for version in ["h2", "h3"] { + let target = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let reservation = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let address = reservation.local_addr().unwrap(); + let protocol = json!({"type":"xhttp", "path":"/tunnel", "http_version":version, + "protocol":{"type":"vless","user_id":"550e8400-e29b-41d4-a716-446655440000"}}); + let mut client_protocol = protocol.clone(); + client_protocol["mode"] = json!("stream-one"); + let (server, client) = if version == "h3" { + (json!({"address":address.to_string(), "transport":"quic", + "quic_settings":{"cert":certificate.cert.pem(),"key":certificate.signing_key.serialize_pem()}, "protocol":protocol}), + json!({"address":address.to_string(), "transport":"quic", + "quic_settings":{"verify":false,"sni_hostname":"localhost"},"protocol":client_protocol})) + } else { + (json!({"address":address.to_string(), "protocol":protocol}), + json!({"address":address.to_string(), "protocol":client_protocol})) + }; + let server = crate::config::create_server_configs(serde_json::from_value(json!([server])).unwrap()).unwrap().configs.remove(0); + let network = crate::network_tasks::NetworkLifetime::default(); + drop(reservation); + let handles = network.scope(crate::tcp::tcp_server::start_servers(server, resolver.clone())).await.unwrap(); + let config: crate::config::ResolvedClientConfig = crate::config::test_support::from_json(client).unwrap(); + let socket = SocketConnectorImpl::from_config(&config, Some(&config.address)).unwrap(); + let proxy = ProxyConnectorImpl::from_config(config, resolver.clone()).unwrap().unwrap(); + let chain = ClientProxyChain::new(vec![InitialHopEntry::Proxy { + socket: Arc::new(socket), proxy: Box::new(proxy), + }], vec![]); + let destination = crate::address::NetLocation::from_str(&target.local_addr().unwrap().to_string(), None).unwrap(); + let mut stream = chain.connect_tcp(destination.into(), &resolver).await.unwrap().client_stream; + stream.write_all(b"x").await.unwrap(); + stream.flush().await.unwrap(); + let (mut accepted, _) = target.accept().await.unwrap(); + assert_eq!(accepted.read_u8().await.unwrap(), b'x'); + network.cancel(); + network.wait().await; + let result = accepted.read(&mut [0]).await; + assert!(matches!(result, Ok(0)) || result.is_err(), "{result:?}"); + for handle in handles { let _ = handle.await; } + } + }).await.unwrap(); +} + #[tokio::test] async fn all_modes_and_carriers_exchange_data_before_upload_eof() { let certificate = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap();