diff --git a/.github/workflows/release-ci.yml b/.github/workflows/release-ci.yml index 7d8e6fc..5becd05 100644 --- a/.github/workflows/release-ci.yml +++ b/.github/workflows/release-ci.yml @@ -75,6 +75,16 @@ jobs: cargo test --lib -p soth-extensions cargo test --lib -p soth-parse cargo test --lib -p soth-code + # soth-proxy and soth-cli carry Windows-divergent runtime code the + # ubuntu PR lanes never compile: the named drain event + # (drain_signal), the ipconfig-based network watcher, and the + # sequenced rotation in the supervisor. Test them natively so + # cfg(windows) API mistakes surface before a release. + - name: proxy + cli tests on Windows + shell: bash + run: | + cargo test --lib -p soth-proxy + cargo test --bins -p soth-cli # Phase 4b sidecar updater — Windows cross-compile validation. # The crate is windows-only at runtime; we cross-compile from @@ -90,7 +100,9 @@ jobs: - shell: bash run: | cargo build -p soth-cli-update-sidecar --bin soth-update --release - cargo test --lib -p soth-cli-update-sidecar + # The sidecar is a binary-only crate: `cargo test --lib` errors + # with "no library targets found". Test the bin target instead. + cargo test --bins -p soth-cli-update-sidecar # FFI conformance is also a per-PR lane (paths-gated on # sdk-core / detect / classify / bindings / fixtures), but we diff --git a/Cargo.lock b/Cargo.lock index 71e510d..0e5fd36 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4093,6 +4093,7 @@ dependencies = [ "dirs", "ed25519-dalek", "hex", + "ipconfig", "libc", "owo-colors", "rand 0.8.5", @@ -4406,6 +4407,7 @@ dependencies = [ "tracing-opentelemetry", "tracing-subscriber", "uuid", + "windows-sys 0.59.0", "zeroize", ] diff --git a/crates/soth-cli/Cargo.toml b/crates/soth-cli/Cargo.toml index 6c7a32d..2523711 100644 --- a/crates/soth-cli/Cargo.toml +++ b/crates/soth-cli/Cargo.toml @@ -59,7 +59,9 @@ comfy-table = "7.0" windows-sys = { version = "0.59", features = [ "Win32_Foundation", "Win32_Networking_WinInet", + "Win32_System_Threading", ] } +ipconfig = "0.3" [dev-dependencies] criterion = { version = "0.5", features = ["async_tokio"] } diff --git a/crates/soth-cli/src/command_graph.rs b/crates/soth-cli/src/command_graph.rs index 9be8908..14d1e3c 100644 --- a/crates/soth-cli/src/command_graph.rs +++ b/crates/soth-cli/src/command_graph.rs @@ -1510,9 +1510,14 @@ mod tests { std::fs::create_dir_all(&bundle_dir).expect("create bundle dir"); std::fs::write(bundle_dir.join("manifest.json"), "{}").expect("write bundle marker"); let old_home = env::var_os("HOME"); + // dirs::home_dir() reads USERPROFILE on Windows, not HOME — without + // overriding both, `~` expansion escapes the temp sandbox into the + // real user profile and the test fails on missing bundle state. + let old_userprofile = env::var_os("USERPROFILE"); let old_soth_home = env::var_os("SOTH_HOME_DIR"); unsafe { env::set_var("HOME", temp.path()); + env::set_var("USERPROFILE", temp.path()); env::set_var("SOTH_HOME_DIR", &soth_home); } let result = std::panic::catch_unwind(|| f(&temp)); @@ -1520,6 +1525,10 @@ mod tests { Some(value) => unsafe { env::set_var("HOME", value) }, None => unsafe { env::remove_var("HOME") }, } + match old_userprofile { + Some(value) => unsafe { env::set_var("USERPROFILE", value) }, + None => unsafe { env::remove_var("USERPROFILE") }, + } match old_soth_home { Some(value) => unsafe { env::set_var("SOTH_HOME_DIR", value) }, None => unsafe { env::remove_var("SOTH_HOME_DIR") }, diff --git a/crates/soth-cli/src/commands/proxy/network_watcher.rs b/crates/soth-cli/src/commands/proxy/network_watcher.rs index 33d12a9..d5959c8 100644 --- a/crates/soth-cli/src/commands/proxy/network_watcher.rs +++ b/crates/soth-cli/src/commands/proxy/network_watcher.rs @@ -1,46 +1,108 @@ //! Detects primary-network changes (wifi switch, captive-portal address -//! reassignment, VPN up/down) and signals the supervisor to graceful-rotate -//! the proxy worker. Without this, after a network change `soth-mitm`'s -//! upstream connection pool keeps trying to reuse TCP sockets bound to the -//! old gateway — surfacing as "tunnel connection failed" until `soth up`. +//! reassignment, VPN up/down, hotspot toggles) and signals the supervisor to +//! graceful-rotate the proxy worker. Without this, after a network change +//! `soth-mitm`'s DNS resolver and upstream sockets keep pointing at the old +//! gateway — surfacing as "tunnel connection failed" until `soth up`. //! -//! Implementation: poll `getifaddrs(3)` every 5s and compare the set of -//! non-loopback IPv4 addresses bound to UP interfaces. Polling (vs. -//! event-driven `SCNetworkConfiguration` / netlink / `NotifyIpInterfaceChange`) -//! costs us a few seconds of latency but avoids three platform-specific code -//! paths and a new dependency. The symptom we're fixing already takes -//! seconds to surface, so the latency floor is acceptable. +//! Implementation: poll a [`NetworkSignature`] every 5s and compare. The +//! signature covers what actually changes across hotspot/AP transitions: +//! +//! - non-loopback IPv4 **and** IPv6 addresses on UP interfaces (v6-only and +//! v6-primary tethering used to be invisible), +//! - the system DNS server list (a same-subnet network switch can change +//! resolvers without changing our addresses), +//! - an offline marker, so *drop → recover on the same address* (hotspot +//! toggled off/on, AP roam) fires a rotation even though the signature is +//! unchanged — previously the empty set was skipped and the recovery was +//! silent. +//! +//! Fires are debounced: at most one signal per [`DEBOUNCE`] window, with the +//! trailing change delivered on the first tick after the window closes, so a +//! flapping hotspot causes one rotation per window instead of a storm. +//! +//! Polling (vs. event-driven `SCNetworkConfiguration` / netlink / +//! `NotifyIpInterfaceChange`) costs a few seconds of latency but avoids +//! three platform-specific code paths. The symptom we're fixing already +//! takes seconds to surface, so the latency floor is acceptable. use std::collections::BTreeSet; -use std::net::Ipv4Addr; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; use std::time::Duration; use tokio::sync::watch; -use tracing::{debug, info, warn}; +use tracing::{debug, info}; const POLL_INTERVAL: Duration = Duration::from_secs(5); +/// Minimum spacing between emitted change signals. Rotation itself takes a +/// few seconds; anything faster just kills connections repeatedly. +const DEBOUNCE: Duration = Duration::from_secs(10); + +/// Point-in-time view of the network facts the proxy worker depends on. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +struct NetworkSignature { + v4: BTreeSet, + /// Global + ULA only. Link-local (fe80::/10) churns on every interface + /// event without affecting routing and is excluded. + v6: BTreeSet, + /// Ordered system DNS server list. + dns: Vec, +} + +impl NetworkSignature { + /// No usable addresses — treat as offline regardless of DNS remnants. + fn is_addrless(&self) -> bool { + self.v4.is_empty() && self.v6.is_empty() + } +} /// Spawn the watcher. Returns a `watch::Receiver` whose value increments on /// every detected change. The supervisor awaits `changed()` on it. pub fn spawn() -> watch::Receiver { + spawn_with_source(read_network_signature) +} + +/// Watcher core with an injectable signature source (tests script network +/// transitions through it; production passes [`read_network_signature`]). +fn spawn_with_source( + mut source: impl FnMut() -> NetworkSignature + Send + 'static, +) -> watch::Receiver { let (tx, rx) = watch::channel(0u64); tokio::spawn(async move { - let mut last = local_v4_addrs(); - debug!(addrs = ?last, "network watcher: initial address set"); + let mut last = source(); + debug!(signature = ?last, "network watcher: initial signature"); + let mut was_offline = false; + let mut pending_signal = false; + let mut last_fire: Option = None; let mut tick = tokio::time::interval(POLL_INTERVAL); tick.tick().await; loop { tick.tick().await; - let current = local_v4_addrs(); - if current.is_empty() { + let current = source(); + if current.is_addrless() { + // Offline. Remember it so recovery fires even when the + // network comes back with an identical signature. + if !was_offline { + debug!("network watcher: all addresses gone (offline)"); + } + was_offline = true; continue; } - if current != last { + if current != last || was_offline { info!( previous = ?last, current = ?current, - "primary network changed — signalling supervisor for graceful proxy rotation" + recovered_from_offline = was_offline, + "primary network changed — scheduling supervisor signal for graceful proxy rotation" ); last = current; + was_offline = false; + pending_signal = true; + } + let debounce_open = last_fire + .map(|fired| fired.elapsed() >= DEBOUNCE) + .unwrap_or(true); + if pending_signal && debounce_open { + pending_signal = false; + last_fire = Some(tokio::time::Instant::now()); let next = tx.borrow().wrapping_add(1); if tx.send(next).is_err() { return; @@ -51,15 +113,38 @@ pub fn spawn() -> watch::Receiver { rx } +fn read_network_signature() -> NetworkSignature { + let (v4, v6) = local_addrs(); + NetworkSignature { + v4, + v6, + dns: system_dns_servers(), + } +} + +/// True for IPv6 addresses that identify a network rather than interface +/// noise: global unicast and ULA, excluding loopback/link-local/unspecified. +fn is_meaningful_v6(addr: &Ipv6Addr) -> bool { + if addr.is_loopback() || addr.is_unspecified() { + return false; + } + // Link-local fe80::/10. + (addr.segments()[0] & 0xffc0) != 0xfe80 +} + #[cfg(unix)] -fn local_v4_addrs() -> BTreeSet { - use libc::{freeifaddrs, getifaddrs, ifaddrs, sockaddr_in, AF_INET, IFF_LOOPBACK, IFF_UP}; - let mut set = BTreeSet::new(); +fn local_addrs() -> (BTreeSet, BTreeSet) { + use libc::{ + freeifaddrs, getifaddrs, ifaddrs, sockaddr_in, sockaddr_in6, AF_INET, AF_INET6, + IFF_LOOPBACK, IFF_UP, + }; + let mut v4 = BTreeSet::new(); + let mut v6 = BTreeSet::new(); unsafe { let mut head: *mut ifaddrs = std::ptr::null_mut(); if getifaddrs(&mut head) != 0 || head.is_null() { - warn!("getifaddrs failed; skipping network change check this tick"); - return set; + tracing::warn!("getifaddrs failed; skipping network change check this tick"); + return (v4, v6); } let mut cur = head; while !cur.is_null() { @@ -72,71 +157,118 @@ fn local_v4_addrs() -> BTreeSet { if flags & IFF_LOOPBACK != 0 || flags & IFF_UP == 0 { continue; } - if (*entry.ifa_addr).sa_family as i32 != AF_INET { - continue; + match (*entry.ifa_addr).sa_family as i32 { + family if family == AF_INET => { + let sin = &*(entry.ifa_addr as *const sockaddr_in); + v4.insert(Ipv4Addr::from(u32::from_be(sin.sin_addr.s_addr))); + } + family if family == AF_INET6 => { + let sin6 = &*(entry.ifa_addr as *const sockaddr_in6); + let addr = Ipv6Addr::from(sin6.sin6_addr.s6_addr); + if is_meaningful_v6(&addr) { + v6.insert(addr); + } + } + _ => {} } - let sin = &*(entry.ifa_addr as *const sockaddr_in); - let addr = Ipv4Addr::from(u32::from_be(sin.sin_addr.s_addr)); - set.insert(addr); } freeifaddrs(head); } - set + (v4, v6) } -/// Windows path: ask the kernel which local IPv4 it would route to a -/// public destination via a UDP "connect" trick. Connecting a UDP -/// socket sends no packets — it only updates the kernel's route -/// resolution for that socket — but `local_addr()` then returns the -/// IP that the default route would use. That IP is exactly what -/// flips on a wifi switch / interface change, so polling it -/// detects the change without any FFI into `GetAdaptersAddresses`. -/// -/// Trade-off vs `getifaddrs` on Unix: this captures only the -/// default-route IP, not the full set of bound IPv4s. On Windows -/// that's enough for the watcher's purpose — the supervisor reload -/// is triggered by *any* change, and the default-route IP changes -/// on every realistic network event (wifi network switch, VPN -/// up/down, dock/undock with USB-Ethernet, captive-portal IP -/// reassignment). It does NOT catch "added a secondary adapter -/// while the existing default route is still active" but neither -/// did the previous empty stub, and the upstream connection pool -/// only cares about routes that actually flip. +/// Windows: enumerate adapters via `ipconfig` (the same crate +/// hickory-resolver uses for its Windows DNS discovery). This replaces the +/// old UDP-connect trick, which captured only the default-route IPv4 — it +/// missed secondary adapters, every IPv6 transition, and DNS-only changes. #[cfg(windows)] -fn local_v4_addrs() -> BTreeSet { - use std::net::{IpAddr, UdpSocket}; +fn local_addrs() -> (BTreeSet, BTreeSet) { + let mut v4 = BTreeSet::new(); + let mut v6 = BTreeSet::new(); + let Ok(adapters) = ipconfig::get_adapters() else { + tracing::warn!("ipconfig::get_adapters failed; skipping network change check this tick"); + return (v4, v6); + }; + for adapter in adapters + .iter() + .filter(|adapter| adapter.oper_status() == ipconfig::OperStatus::IfOperStatusUp) + { + for addr in adapter.ip_addresses() { + match addr { + IpAddr::V4(value) => { + if !value.is_loopback() && !value.is_unspecified() { + v4.insert(*value); + } + } + IpAddr::V6(value) => { + if is_meaningful_v6(value) { + v6.insert(*value); + } + } + } + } + } + (v4, v6) +} + +/// Catch-all for non-Unix, non-Windows targets (BSDs we don't officially +/// ship to, WASM, etc.). Empty sets mean the watcher polls but never fires. +#[cfg(not(any(unix, windows)))] +fn local_addrs() -> (BTreeSet, BTreeSet) { + (BTreeSet::new(), BTreeSet::new()) +} - let mut set = BTreeSet::new(); - let Ok(sock) = UdpSocket::bind("0.0.0.0:0") else { - return set; +/// System DNS servers, best-effort. An empty list disables DNS-change +/// detection for the tick but never blocks address-based detection. +/// +/// Unix: parse `/etc/resolv.conf` — written by `configd` on macOS and by +/// the resolver stack on Linux. (On systemd-resolved hosts this is the +/// constant 127.0.0.53 stub; DNS-change detection is then inert there, +/// which is harmless — address changes still fire.) Zone-scoped +/// nameservers (`fe80::1%en0`) don't parse as bare `IpAddr` and are +/// skipped. +#[cfg(unix)] +fn system_dns_servers() -> Vec { + let Ok(contents) = std::fs::read_to_string("/etc/resolv.conf") else { + return Vec::new(); }; - // 8.8.8.8 is a stable, well-known target. UDP connect doesn't - // emit packets — it only sets the destination so the kernel - // resolves a route. If the host is fully offline the connect - // can fail; fall through to an empty set, which matches the - // "no networks" baseline and won't spuriously trigger reloads. - if sock.connect("8.8.8.8:80").is_err() { - return set; - } - let Ok(addr) = sock.local_addr() else { - return set; + parse_resolv_conf_nameservers(&contents) +} + +#[cfg(unix)] +fn parse_resolv_conf_nameservers(contents: &str) -> Vec { + contents + .lines() + .filter_map(|line| { + let trimmed = line.trim(); + let value = trimmed.strip_prefix("nameserver")?.trim(); + value.split_whitespace().next()?.parse::().ok() + }) + .collect() +} + +#[cfg(windows)] +fn system_dns_servers() -> Vec { + let Ok(adapters) = ipconfig::get_adapters() else { + return Vec::new(); }; - if let IpAddr::V4(v4) = addr.ip() { - if !v4.is_loopback() && !v4.is_unspecified() { - set.insert(v4); + let mut servers = Vec::new(); + for adapter in adapters + .iter() + .filter(|adapter| adapter.oper_status() == ipconfig::OperStatus::IfOperStatusUp) + { + for server in adapter.dns_servers() { + if !servers.contains(server) { + servers.push(*server); + } } } - set + servers } -/// Catch-all for non-Unix, non-Windows targets (BSDs we don't -/// officially ship to, WASM, etc.). Empty set means the watcher -/// polls but never fires — same effective behaviour as the -/// pre-fix Windows path, just scoped to platforms we don't -/// actively support. #[cfg(not(any(unix, windows)))] -fn local_v4_addrs() -> BTreeSet { - BTreeSet::new() +fn system_dns_servers() -> Vec { + Vec::new() } #[cfg(all(test, any(unix, windows)))] @@ -144,26 +276,168 @@ mod tests { use super::*; #[test] - fn local_v4_addrs_includes_loopback_when_loopback_filter_disabled() { - // Sanity check: getifaddrs returns *something* on a normal host, and - // our filter excludes loopback. We can't assert specific addresses - // (CI / dev hosts vary), but the iteration should not panic and - // should not return loopback (127.0.0.1) since it's excluded. - let set = local_v4_addrs(); + fn local_addrs_exclude_loopback() { + let (v4, v6) = local_addrs(); + assert!( + !v4.contains(&Ipv4Addr::LOCALHOST), + "v4 loopback should be filtered out" + ); assert!( - !set.contains(&Ipv4Addr::LOCALHOST), - "loopback should be filtered out" + !v6.contains(&Ipv6Addr::LOCALHOST), + "v6 loopback should be filtered out" ); } + #[test] + fn meaningful_v6_excludes_link_local_and_loopback() { + assert!(!is_meaningful_v6(&Ipv6Addr::LOCALHOST)); + assert!(!is_meaningful_v6(&Ipv6Addr::UNSPECIFIED)); + assert!(!is_meaningful_v6(&"fe80::1".parse().unwrap())); + assert!(is_meaningful_v6(&"2001:db8::1".parse().unwrap())); + // ULA stays in — it identifies the local network. + assert!(is_meaningful_v6(&"fd00::1".parse().unwrap())); + } + + #[cfg(unix)] + #[test] + fn resolv_conf_parser_extracts_nameservers() { + let sample = "# comment\nnameserver 192.168.1.1\nnameserver 2606:4700::1111\nsearch example.com\nnameserver fe80::1%en0\nnameserver not-an-ip\n"; + let servers = parse_resolv_conf_nameservers(sample); + assert_eq!( + servers, + vec![ + "192.168.1.1".parse::().unwrap(), + "2606:4700::1111".parse::().unwrap(), + ] + ); + } + + #[test] + fn addrless_signature_reads_as_offline() { + let empty = NetworkSignature::default(); + assert!(empty.is_addrless()); + let with_dns_only = NetworkSignature { + dns: vec!["1.1.1.1".parse().unwrap()], + ..NetworkSignature::default() + }; + assert!(with_dns_only.is_addrless()); + let online = NetworkSignature { + v4: [Ipv4Addr::new(192, 168, 1, 2)].into_iter().collect(), + ..NetworkSignature::default() + }; + assert!(!online.is_addrless()); + } + #[tokio::test(flavor = "current_thread", start_paused = true)] async fn watcher_emits_no_events_when_addresses_stable() { let mut rx = spawn(); // Advance past several poll intervals; nothing should fire because - // the address set hasn't changed. + // the signature hasn't changed. tokio::time::advance(POLL_INTERVAL * 3).await; - // changed() should not have fired. let changed = tokio::time::timeout(Duration::from_millis(10), rx.changed()).await; assert!(changed.is_err(), "watcher should not emit when stable"); } + + fn sig_with_v4(last_octet: u8) -> NetworkSignature { + NetworkSignature { + v4: [Ipv4Addr::new(192, 168, 1, last_octet)] + .into_iter() + .collect(), + ..NetworkSignature::default() + } + } + + /// Drive a scripted sequence of signatures through the watcher; the + /// final entry repeats forever once the script is exhausted. + fn spawn_scripted(script: Vec) -> watch::Receiver { + let mut queue = script.into_iter(); + let mut current = NetworkSignature::default(); + spawn_with_source(move || { + if let Some(next) = queue.next() { + current = next; + } + current.clone() + }) + } + + /// Let the spawned watcher task run (paused-clock tests advance time + /// manually; the task still needs scheduler turns to consume ticks). + async fn drain_scheduler() { + for _ in 0..8 { + tokio::task::yield_now().await; + } + } + + #[tokio::test(flavor = "current_thread", start_paused = true)] + async fn watcher_fires_on_offline_recovery_with_same_signature() { + // initial read, then: online → offline → online with the SAME + // signature. The old watcher skipped empty sets entirely and never + // fired on same-address recovery (hotspot toggled off/on). + let rx = spawn_scripted(vec![ + sig_with_v4(2), // initial read + sig_with_v4(2), // tick 1: stable + NetworkSignature::default(), // tick 2: offline + sig_with_v4(2), // tick 3: recovered, same address + ]); + drain_scheduler().await; + tokio::time::advance(POLL_INTERVAL * 3).await; + drain_scheduler().await; + assert!( + rx.has_changed().expect("watcher alive"), + "recovery with unchanged signature must fire a rotation signal" + ); + } + + #[tokio::test(flavor = "current_thread", start_paused = true)] + async fn watcher_fires_on_dns_only_change() { + let base = sig_with_v4(2); + let mut with_new_dns = base.clone(); + with_new_dns.dns = vec!["9.9.9.9".parse().unwrap()]; + let rx = spawn_scripted(vec![base.clone(), base, with_new_dns]); + drain_scheduler().await; + tokio::time::advance(POLL_INTERVAL * 2).await; + drain_scheduler().await; + assert!( + rx.has_changed().expect("watcher alive"), + "DNS-only change must fire" + ); + } + + #[tokio::test(flavor = "current_thread", start_paused = true)] + async fn watcher_debounces_rapid_flapping() { + // Flap on every tick: the first change fires immediately, changes + // inside the debounce window are deferred, and the trailing change + // is delivered once the window closes. + let mut rx = spawn_scripted(vec![ + sig_with_v4(2), // initial read + sig_with_v4(3), // tick 1: change → fires + sig_with_v4(2), // tick 2: change inside window → deferred + sig_with_v4(3), // tick 3: change inside window → still deferred + ]); + drain_scheduler().await; + + tokio::time::advance(POLL_INTERVAL).await; // tick 1 + drain_scheduler().await; + assert!( + rx.has_changed().expect("watcher alive"), + "first change must fire immediately" + ); + rx.borrow_and_update(); + + tokio::time::advance(POLL_INTERVAL).await; // tick 2, inside window + drain_scheduler().await; + assert!( + !rx.has_changed().expect("watcher alive"), + "change inside the debounce window must be deferred" + ); + + // Advance past the debounce window; the deferred change fires on the + // next tick. + tokio::time::advance(DEBOUNCE + POLL_INTERVAL).await; + drain_scheduler().await; + assert!( + rx.has_changed().expect("watcher alive"), + "deferred change must fire once the debounce window closes" + ); + } } diff --git a/crates/soth-cli/src/commands/proxy/start.rs b/crates/soth-cli/src/commands/proxy/start.rs index c2dbf9a..0bd99f5 100644 --- a/crates/soth-cli/src/commands/proxy/start.rs +++ b/crates/soth-cli/src/commands/proxy/start.rs @@ -170,10 +170,11 @@ pub async fn run( #[cfg(not(unix))] let listener_fd: Option = None; - let mut child = spawn_proxy_process(generated_path.as_path(), listener_fd) + let (mut child, ready_probe) = spawn_proxy_process(generated_path.as_path(), listener_fd) .await .context("spawn soth-proxy process")?; - wait_for_listener_start(&mut child, expected_port).await?; + wait_for_listener_start(&mut child, expected_port, &ready_probe).await?; + drop(ready_probe); // Historian sibling process. Spawned only when both `enabled` and // run_mode == Subprocess. Watched in its own background task so its @@ -613,9 +614,42 @@ async fn spawn_classify_daemon_process(config_path: &Path) -> Result { .map_err(|error| anyhow::anyhow!("failed launching classify daemon worker: {error}")) } -async fn spawn_proxy_process(config_path: &Path, listener_fd: Option) -> Result { +/// Supervisor-side handle to the worker readiness handshake: the path the +/// spawned child will write its ready-file to (see +/// `soth_proxy::drain_signal::WORKER_READY_FILE_ENV`). +struct WorkerReadyProbe { + path: PathBuf, +} + +impl WorkerReadyProbe { + fn new() -> Self { + let path = soth_home_dir() + .join("run") + .join(format!("worker_ready_{}.json", uuid::Uuid::new_v4())); + Self { path } + } + + fn is_ready(&self) -> bool { + self.path.exists() + } +} + +impl Drop for WorkerReadyProbe { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.path); + } +} + +async fn spawn_proxy_process( + config_path: &Path, + listener_fd: Option, +) -> Result<(Child, WorkerReadyProbe)> { let current_exe = std::env::current_exe().context("resolve current executable for proxy worker")?; + let ready_probe = WorkerReadyProbe::new(); + if let Some(parent) = ready_probe.path.parent() { + let _ = std::fs::create_dir_all(parent); + } let mut cmd = Command::new(current_exe); // `start --daemon-child` + SOTH_PROXY_WORKER=1 selects the in-process MITM // runtime path in `run()` below, replacing the historical `soth-proxy` @@ -623,6 +657,10 @@ async fn spawn_proxy_process(config_path: &Path, listener_fd: Option) -> Re cmd.arg("start").arg("--daemon-child"); cmd.env(PROXY_WORKER_ENV, "1"); cmd.env("SOTH_PROXY_CONFIG", config_path); + cmd.env( + soth_proxy::drain_signal::WORKER_READY_FILE_ENV, + &ready_probe.path, + ); if let Some(fd) = listener_fd { cmd.env("SOTH_LISTENER_FD", fd.to_string()); } @@ -643,8 +681,10 @@ async fn spawn_proxy_process(config_path: &Path, listener_fd: Option) -> Re const CREATE_NEW_PROCESS_GROUP: u32 = 0x0000_0200; cmd.creation_flags(CREATE_NO_WINDOW | CREATE_NEW_PROCESS_GROUP); } - cmd.spawn() - .map_err(|error| anyhow::anyhow!("failed launching proxy worker: {error}")) + let child = cmd + .spawn() + .map_err(|error| anyhow::anyhow!("failed launching proxy worker: {error}"))?; + Ok((child, ready_probe)) } /// Env var toggle that re-executed child processes use to enter in-process @@ -790,16 +830,56 @@ async fn supervise_proxy( } ProxyExit::Reload => { info!("reload requested (SIGHUP or network change) — performing graceful child rotation"); - let mut new_child = spawn_proxy_process(config_path, listener_fd) + // Windows: the port cannot be shared (no FD passing; the + // worker binds with SO_EXCLUSIVEADDRUSE), so rotation is + // sequenced — signal the drain FIRST. The old child closes + // its listener within milliseconds (freeing the port for + // the new child's bind-retry loop) and keeps draining + // in-flight flows in the background while the new child + // starts. On Unix the inherited-listener overlap flow is + // kept: new child up first, then drain the old. + #[cfg(windows)] + let old_child_draining = { + let signaled = child.id().map(signal_windows_drain_event).unwrap_or(false); + if !signaled { + warn!("could not signal drain event on old proxy child; hard-killing before respawn"); + let _ = terminate_child(child).await; + } + signaled + }; + + let (mut new_child, ready_probe) = spawn_proxy_process(config_path, listener_fd) .await .context("spawn new soth-proxy for graceful rotation")?; - if let Err(error) = wait_for_listener_start(&mut new_child, expected_port).await { + if let Err(error) = + wait_for_listener_start(&mut new_child, expected_port, &ready_probe).await + { + // Unix: the old child is untouched, keep serving on it. + // Windows: the old child is already draining; fall + // through — the supervisor loop observes its exit and + // respawns via the normal restart path. warn!(error = %error, "new proxy child failed to start; keeping old child"); let _ = terminate_child(&mut new_child).await; continue; } info!("new proxy child healthy — draining old child"); + #[cfg(unix)] graceful_stop_child(child).await?; + #[cfg(windows)] + if old_child_draining { + match tokio::time::timeout(Duration::from_secs(30), child.wait()).await { + Ok(Ok(status)) => { + info!(status = %status, "old proxy child exited after drain"); + } + Ok(Err(error)) => { + warn!(error = %error, "error waiting for old proxy child"); + } + Err(_) => { + warn!("old proxy child did not exit within 30s drain window; killing"); + let _ = terminate_child(child).await; + } + } + } *child = new_child; last_healthy = Instant::now(); consecutive_failures = 0; @@ -865,11 +945,17 @@ async fn supervise_proxy( ); tokio::time::sleep(Duration::from_millis(backoff_ms)).await; - *child = spawn_proxy_process(config_path, listener_fd) + let (respawned_child, ready_probe) = spawn_proxy_process(config_path, listener_fd) .await .context("respawn soth-proxy process")?; - if let Err(error) = - wait_for_listener_start_with_timeout(child, expected_port, next_startup_timeout).await + *child = respawned_child; + if let Err(error) = wait_for_listener_start_with_timeout( + child, + expected_port, + next_startup_timeout, + &ready_probe, + ) + .await { warn!( timeout_secs = next_startup_timeout.as_secs(), @@ -981,23 +1067,97 @@ async fn graceful_stop_child(child: &mut Child) -> Result<()> { Ok(()) } -#[cfg(not(unix))] +/// Windows equivalent of the SIGUSR1 drain: sets the named kernel event the +/// worker created at startup (`Local\soth-worker-drain-{pid}`, see +/// `soth_proxy::drain_signal`), then waits up to 30 seconds for exit. +/// Falls back to a hard kill if the event can't be opened (worker predates +/// this protocol, or died) or the drain window elapses. +#[cfg(windows)] +async fn graceful_stop_child(child: &mut Child) -> Result<()> { + let signaled = child.id().map(signal_windows_drain_event).unwrap_or(false); + if !signaled { + warn!("could not signal drain event on old proxy child; falling back to hard kill"); + return terminate_child(child).await; + } + match tokio::time::timeout(Duration::from_secs(30), child.wait()).await { + Ok(Ok(status)) => { + info!(status = %status, "old proxy child exited after drain"); + } + Ok(Err(error)) => { + warn!(error = %error, "error waiting for old proxy child"); + } + Err(_) => { + warn!("old proxy child did not exit within 30s drain window; killing"); + let _ = terminate_child(child).await; + } + } + Ok(()) +} + +/// Open the worker's named drain event and set it. Returns `false` when the +/// event doesn't exist or can't be signaled — callers fall back to a hard +/// kill, matching the pre-drain behavior. +#[cfg(windows)] +fn signal_windows_drain_event(pid: u32) -> bool { + use windows_sys::Win32::Foundation::CloseHandle; + use windows_sys::Win32::System::Threading::{OpenEventW, SetEvent, EVENT_MODIFY_STATE}; + + let name: Vec = soth_proxy::drain_signal::drain_event_name(pid) + .encode_utf16() + .chain(std::iter::once(0)) + .collect(); + unsafe { + let handle = OpenEventW(EVENT_MODIFY_STATE, 0, name.as_ptr()); + if handle.is_null() { + return false; + } + let signaled = SetEvent(handle) != 0; + CloseHandle(handle); + signaled + } +} + +#[cfg(not(any(unix, windows)))] async fn graceful_stop_child(child: &mut Child) -> Result<()> { terminate_child(child).await } -async fn wait_for_listener_start(child: &mut Child, port: u16) -> Result<()> { - wait_for_listener_start_with_timeout(child, port, listener_startup_timeout()).await +async fn wait_for_listener_start( + child: &mut Child, + port: u16, + ready_probe: &WorkerReadyProbe, +) -> Result<()> { + wait_for_listener_start_with_timeout(child, port, listener_startup_timeout(), ready_probe).await } +/// Wait until the spawned worker reports ready. +/// +/// Primary signal: the worker's ready-file (written by that exact child once +/// its listener is live). The old TCP probe on `127.0.0.1:` could not +/// distinguish the new child's listener from the old child's during rotation +/// — and on Unix it didn't even prove the worker was up, since the +/// supervisor itself holds the inherited listener. The probe is kept only as +/// a late fallback for version-skewed workers that predate the ready-file +/// protocol (possible across hot-updates): it is consulted only after half +/// the startup budget has elapsed with no ready-file. async fn wait_for_listener_start_with_timeout( child: &mut Child, port: u16, timeout: Duration, + ready_probe: &WorkerReadyProbe, ) -> Result<()> { - let deadline = Instant::now() + timeout; + let start = Instant::now(); + let deadline = start + timeout; + let probe_fallback_after = start + timeout / 2; loop { - if is_local_listener_ready(port) { + if ready_probe.is_ready() { + return Ok(()); + } + if Instant::now() >= probe_fallback_after && is_local_listener_ready(port) { + warn!( + port, + "worker ready-file not seen but port answers; accepting via legacy probe (version-skewed worker?)" + ); return Ok(()); } if let Some(status) = child @@ -1009,7 +1169,7 @@ async fn wait_for_listener_start_with_timeout( if Instant::now() >= deadline { let _ = terminate_child(child).await; anyhow::bail!( - "soth-proxy did not open 127.0.0.1:{} within {}s startup timeout", + "soth-proxy did not report ready on 127.0.0.1:{} within {}s startup timeout", port, timeout.as_secs() ); @@ -1778,14 +1938,30 @@ async fn run_classify_daemon_worker() -> Result<()> { mod tests { use super::*; + #[test] + fn worker_ready_probe_removes_file_on_drop() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("worker_ready_test.json"); + let probe = WorkerReadyProbe { path: path.clone() }; + assert!(!probe.is_ready()); + std::fs::write(&path, b"{\"pid\": 1}\n").expect("write ready file"); + assert!(probe.is_ready()); + drop(probe); + assert!(!path.exists(), "probe drop must clean up the ready file"); + } + fn with_temp_home(f: impl FnOnce(std::path::PathBuf) -> T + std::panic::UnwindSafe) -> T { let guard = crate::commands::proxy::lock_test_env(); let temp = tempfile::tempdir().expect("tempdir"); let old_home = std::env::var_os("HOME"); + // dirs::home_dir() reads USERPROFILE on Windows, not HOME — override + // both so `~` expansion can't escape the temp sandbox. + let old_userprofile = std::env::var_os("USERPROFILE"); let old_soth_home = std::env::var_os("SOTH_HOME_DIR"); let soth_home = temp.path().join(".soth"); unsafe { std::env::set_var("HOME", temp.path()); + std::env::set_var("USERPROFILE", temp.path()); std::env::set_var("SOTH_HOME_DIR", &soth_home); } @@ -1799,6 +1975,14 @@ mod tests { std::env::remove_var("HOME"); }, } + match old_userprofile { + Some(value) => unsafe { + std::env::set_var("USERPROFILE", value); + }, + None => unsafe { + std::env::remove_var("USERPROFILE"); + }, + } match old_soth_home { Some(value) => unsafe { std::env::set_var("SOTH_HOME_DIR", value); diff --git a/crates/soth-cli/src/commands/proxy/system.rs b/crates/soth-cli/src/commands/proxy/system.rs index 316ac94..f30716b 100644 --- a/crates/soth-cli/src/commands/proxy/system.rs +++ b/crates/soth-cli/src/commands/proxy/system.rs @@ -1695,10 +1695,14 @@ mod tests { let guard = crate::commands::proxy::lock_test_env(); let temp = tempfile::tempdir().expect("tempdir"); let old_home = env::var_os("HOME"); + // dirs::home_dir() reads USERPROFILE on Windows, not HOME — override + // both so `~` expansion can't escape the temp sandbox. + let old_userprofile = env::var_os("USERPROFILE"); let old_soth_home = env::var_os("SOTH_HOME_DIR"); let soth_home = temp.path().join(".soth"); unsafe { env::set_var("HOME", temp.path()); + env::set_var("USERPROFILE", temp.path()); env::set_var("SOTH_HOME_DIR", &soth_home); } let result = f(); @@ -1710,6 +1714,14 @@ mod tests { env::remove_var("HOME"); }, } + match old_userprofile { + Some(value) => unsafe { + env::set_var("USERPROFILE", value); + }, + None => unsafe { + env::remove_var("USERPROFILE"); + }, + } match old_soth_home { Some(value) => unsafe { env::set_var("SOTH_HOME_DIR", value); diff --git a/crates/soth-proxy/Cargo.toml b/crates/soth-proxy/Cargo.toml index ca46e0d..9e119ad 100644 --- a/crates/soth-proxy/Cargo.toml +++ b/crates/soth-proxy/Cargo.toml @@ -64,6 +64,13 @@ soth-sync = { workspace = true } soth-extensions = { workspace = true, optional = true } soth-mitm = "0.3.3" +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.59", features = [ + "Win32_Foundation", + "Win32_Security", + "Win32_System_Threading", +] } + [dev-dependencies] tower = { version = "0.5", features = ["util"] } tempfile = { workspace = true } diff --git a/crates/soth-proxy/src/drain_signal.rs b/crates/soth-proxy/src/drain_signal.rs new file mode 100644 index 0000000..effe61b --- /dev/null +++ b/crates/soth-proxy/src/drain_signal.rs @@ -0,0 +1,176 @@ +//! Cross-platform graceful-drain signal for the proxy worker. +//! +//! Unix: SIGUSR1, sent by the supervisor during graceful child rotation. +//! Windows: a named kernel event `Local\soth-worker-drain-{pid}` that the +//! supervisor opens and sets. `GenerateConsoleCtrlEvent` is not usable here — +//! the worker is spawned with `CREATE_NO_WINDOW` so it does not share a +//! console with the supervisor, and console ctrl events cannot cross that +//! boundary. A named event has no such restriction. +//! +//! Without this, Windows rotation had no drain path at all: the supervisor +//! hard-killed the worker (`TerminateProcess`), resetting every in-flight +//! connection on each network change. + +use tracing::info; + +/// Env var carrying the supervisor-chosen path of the readiness file. The +/// worker writes `{"pid": }` there once its listener is live; the +/// supervisor waits for the file instead of TCP-probing the port. +/// +/// The old probe (`TcpStream::connect(127.0.0.1:)`) could not tell +/// the new child's listener from the old child's during rotation — on Unix +/// it didn't even prove the worker was up, since the supervisor itself +/// holds the inherited listener and the kernel completes connects from the +/// backlog. The ready-file is written by the worker process itself, so it +/// is unambiguous. +pub const WORKER_READY_FILE_ENV: &str = "SOTH_WORKER_READY_FILE"; + +/// Write the readiness file if the supervisor requested one (best-effort: +/// standalone/foreground runs have no supervisor and skip this). Written +/// atomically via temp-file + rename so the supervisor never reads a +/// partial file. +pub(crate) fn write_worker_ready_file() { + let Ok(path) = std::env::var(WORKER_READY_FILE_ENV) else { + return; + }; + let path = std::path::PathBuf::from(path); + let body = format!("{{\"pid\": {}}}\n", std::process::id()); + let temp = path.with_extension("tmp"); + let written = + std::fs::write(&temp, body.as_bytes()).and_then(|()| std::fs::rename(&temp, &path)); + match written { + Ok(()) => info!(path = %path.display(), "worker readiness file written"), + Err(error) => tracing::warn!( + path = %path.display(), + %error, + "failed writing worker readiness file; supervisor will fall back to port probe" + ), + } +} + +/// The name of the drain event for a worker process id. Kept in sync with +/// the supervisor side (`soth-cli/src/commands/proxy/start.rs`). +#[cfg(windows)] +pub fn drain_event_name(pid: u32) -> String { + format!("Local\\soth-worker-drain-{pid}") +} + +/// Wait until a shutdown is requested: Ctrl+C on all platforms, plus +/// SIGUSR1 (Unix) or the named drain event (Windows). +pub(crate) async fn wait_for_shutdown_signal() { + #[cfg(unix)] + { + use tokio::signal::unix::{signal, SignalKind}; + let mut usr1 = signal(SignalKind::user_defined1()).expect("listen for SIGUSR1"); + tokio::select! { + _ = tokio::signal::ctrl_c() => { + info!("shutdown requested (Ctrl+C)"); + } + _ = usr1.recv() => { + info!("graceful drain requested (SIGUSR1)"); + } + } + } + #[cfg(windows)] + { + let (tx, rx) = tokio::sync::oneshot::channel::<()>(); + // A plain std thread (not spawn_blocking): the tokio runtime waits + // for blocking tasks on drop, and this thread parks forever when + // shutdown comes via Ctrl+C instead of the event. + let spawned = std::thread::Builder::new() + .name("soth-drain-event".to_string()) + .spawn(move || { + if wait_for_drain_event(std::process::id()) { + let _ = tx.send(()); + } + // On failure the sender drops; the receiver arm below + // falls back to Ctrl+C-only. + }) + .is_ok(); + if !spawned { + tracing::warn!("failed to spawn drain-event thread; Ctrl+C only"); + } + tokio::select! { + _ = tokio::signal::ctrl_c() => { + info!("shutdown requested (Ctrl+C)"); + } + result = rx => match result { + Ok(()) => { + info!("graceful drain requested (drain event)"); + } + Err(_) => { + tracing::warn!( + "drain-event listener unavailable; waiting on Ctrl+C only" + ); + let _ = tokio::signal::ctrl_c().await; + info!("shutdown requested (Ctrl+C)"); + } + }, + } + } + #[cfg(not(any(unix, windows)))] + { + let _ = tokio::signal::ctrl_c().await; + info!("shutdown requested (Ctrl+C)"); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ready_file_written_atomically_when_env_set() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("worker_ready_test.json"); + // Env var is process-global; nothing else in this test binary reads + // it, and we clean up immediately after. + std::env::set_var(WORKER_READY_FILE_ENV, &path); + write_worker_ready_file(); + std::env::remove_var(WORKER_READY_FILE_ENV); + + let body = std::fs::read_to_string(&path).expect("ready file written"); + assert!(body.contains(&format!("\"pid\": {}", std::process::id()))); + assert!( + !path.with_extension("tmp").exists(), + "temp file must be renamed away" + ); + } + + #[test] + fn ready_file_skipped_without_env() { + // Must not panic or create anything when unsupervised. + std::env::remove_var(WORKER_READY_FILE_ENV); + write_worker_ready_file(); + } +} + +/// Create the named drain event and block until the supervisor sets it. +/// Returns `true` when the event fired, `false` on any setup/wait failure. +// unsafe_code is denied crate-wide; this function is one of the two FFI +// exceptions (with sqlite_vec) — kernel event syscalls have no safe wrapper. +#[allow(unsafe_code)] +#[cfg(windows)] +fn wait_for_drain_event(pid: u32) -> bool { + use windows_sys::Win32::Foundation::{CloseHandle, WAIT_OBJECT_0}; + use windows_sys::Win32::System::Threading::{CreateEventW, WaitForSingleObject, INFINITE}; + + let name: Vec = drain_event_name(pid) + .encode_utf16() + .chain(std::iter::once(0)) + .collect(); + // Manual-reset, initially unsignaled. Created by the worker (not the + // supervisor) so it exists for the worker's whole lifetime; the + // supervisor opens it by name only when it wants to drain. + let handle = unsafe { CreateEventW(std::ptr::null(), 1, 0, name.as_ptr()) }; + if handle.is_null() { + tracing::warn!( + error = %std::io::Error::last_os_error(), + "failed to create drain event; graceful drain unavailable" + ); + return false; + } + let wait_result = unsafe { WaitForSingleObject(handle, INFINITE) }; + unsafe { CloseHandle(handle) }; + wait_result == WAIT_OBJECT_0 +} diff --git a/crates/soth-proxy/src/lib.rs b/crates/soth-proxy/src/lib.rs index a19a8ca..69d8a4f 100644 --- a/crates/soth-proxy/src/lib.rs +++ b/crates/soth-proxy/src/lib.rs @@ -1,6 +1,7 @@ -// unsafe_code is denied crate-wide; the single exception is sqlite_vec, which -// must call sqlite3_auto_extension via FFI. deny (not forbid) is used so that -// the per-module allow override on sqlite_vec is permitted. +// unsafe_code is denied crate-wide; the exceptions are sqlite_vec (must call +// sqlite3_auto_extension via FFI) and drain_signal's Windows kernel-event +// syscalls. deny (not forbid) is used so that per-item allow overrides are +// permitted. #![deny(unsafe_code)] #![allow(clippy::too_many_arguments)] @@ -14,6 +15,7 @@ pub mod classify_task; pub mod config; pub mod db; +pub mod drain_signal; pub mod error; pub mod gating; pub mod runtime; diff --git a/crates/soth-proxy/src/runtime.rs b/crates/soth-proxy/src/runtime.rs index cc830d2..7e256a6 100644 --- a/crates/soth-proxy/src/runtime.rs +++ b/crates/soth-proxy/src/runtime.rs @@ -2,7 +2,8 @@ //! //! `run(registry)` drives the full proxy lifecycle: loads config, opens the DB, //! downloads classify models if needed, spins up MITM + ops + sync + telemetry, -//! then waits for Ctrl+C or SIGUSR1 (graceful drain) and shuts everything down. +//! then waits for Ctrl+C or a graceful-drain signal (SIGUSR1 on Unix, the +//! named drain event on Windows — see `drain_signal`) and shuts everything down. //! //! Extension registration is the caller's responsibility: the `soth` CLI //! constructs an `ExtensionRegistry` (registering e.g. `HistorianExtension`) @@ -336,26 +337,10 @@ async fn run_inner(ext_registry: Option) -> Result<()> { None => proxy.start().await.context("start mitm proxy")?, }; + crate::drain_signal::write_worker_ready_file(); info!("soth-proxy started; press Ctrl+C to stop"); - #[cfg(unix)] - { - use tokio::signal::unix::{signal, SignalKind}; - let mut usr1 = signal(SignalKind::user_defined1()).expect("listen for SIGUSR1"); - tokio::select! { - _ = tokio::signal::ctrl_c() => { - info!("shutdown requested (Ctrl+C)"); - } - _ = usr1.recv() => { - info!("graceful drain requested (SIGUSR1)"); - } - } - } - #[cfg(not(unix))] - { - tokio::signal::ctrl_c().await.context("wait for Ctrl+C")?; - info!("shutdown requested"); - } + crate::drain_signal::wait_for_shutdown_signal().await; proxy_handle .shutdown(Duration::from_secs(30)) .await @@ -647,26 +632,10 @@ async fn run_inner() -> Result<()> { None => proxy.start().await.context("start mitm proxy")?, }; + crate::drain_signal::write_worker_ready_file(); info!("soth-proxy started; press Ctrl+C to stop"); - #[cfg(unix)] - { - use tokio::signal::unix::{signal, SignalKind}; - let mut usr1 = signal(SignalKind::user_defined1()).expect("listen for SIGUSR1"); - tokio::select! { - _ = tokio::signal::ctrl_c() => { - info!("shutdown requested (Ctrl+C)"); - } - _ = usr1.recv() => { - info!("graceful drain requested (SIGUSR1)"); - } - } - } - #[cfg(not(unix))] - { - tokio::signal::ctrl_c().await.context("wait for Ctrl+C")?; - info!("shutdown requested"); - } + crate::drain_signal::wait_for_shutdown_signal().await; proxy_handle .shutdown(Duration::from_secs(30)) .await