diff --git a/Cargo.lock b/Cargo.lock index 65a5dd1..43b4d07 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3832,6 +3832,7 @@ dependencies = [ "hkdf", "hmac 0.13.0", "iroh", + "netdev 0.45.1", "netwatch", "postcard", "rand 0.10.3", diff --git a/crates/tsunagi-cli/src/main.rs b/crates/tsunagi-cli/src/main.rs index d343af7..c60d5cc 100644 --- a/crates/tsunagi-cli/src/main.rs +++ b/crates/tsunagi-cli/src/main.rs @@ -712,14 +712,14 @@ enum Observed { /// Asks the agent, and falls back to the state store. async fn observe(paths: &StoragePaths, socket: &std::path::Path) -> Observed { - let socket_present = socket.exists(); + let socket_present = tsunagi::ipc::is_serving(socket).await; let why = if socket_present { - match tsunagi::ipc::unix::request_status(socket).await { + match tsunagi::ipc::request_status(socket).await { Ok(report) => return Observed::Agent(Box::new(report)), Err(err) => format!("{err}"), } } else { - "no control socket for this state directory".to_string() + "no agent is serving this state directory".to_string() }; // Read-only, and deliberately tolerant: a state directory that has never @@ -876,7 +876,11 @@ fn dns_publisher() -> Arc { { Arc::new(tsunagi::dns::publish::ResolvedPublisher::new()) } - #[cfg(not(target_os = "linux"))] + #[cfg(target_os = "windows")] + { + Arc::new(tsunagi::dns::publish::NrptPublisher::new()) + } + #[cfg(not(any(target_os = "linux", target_os = "windows")))] { Arc::new(tsunagi::dns::publish::UnsupportedPublisher::new()) } @@ -1212,7 +1216,7 @@ struct AgentControl { paths: StoragePaths, } -impl tsunagi::ipc::unix::ReportSource for AgentControl { +impl tsunagi::ipc::ReportSource for AgentControl { fn report(&self) -> tsunagi::BoxFuture<'_, tsunagi::ipc::StatusReport> { Box::pin(async move { let dns = self.dns_state().await; @@ -1531,9 +1535,9 @@ async fn join_network( // The running agent, because a second `up` cannot have the directory // and because this way the network starts at once instead of at the // next restart. - if socket.exists() { + if tsunagi::ipc::is_serving(socket).await { let report = - tsunagi::ipc::unix::join_network(socket, name.as_str(), secret.encode().as_str()) + tsunagi::ipc::join_network(socket, name.as_str(), secret.encode().as_str()) .await?; // The id in full either way: it is what every other command takes, // and the shortened form in a report is for reading, not copying. @@ -1618,7 +1622,7 @@ async fn join_network( /// comes from the running agent, because without a peer to contact the /// other side has nothing to go on. async fn invite(socket: &std::path::Path, name: &NetworkName, secret: &NetworkSecret) { - let endpoint = tsunagi::ipc::unix::request_status(socket) + let endpoint = tsunagi::ipc::request_status(socket) .await .ok() .map(|report| report.endpoint_id) @@ -1726,8 +1730,8 @@ async fn set_active( let id = network.network_id.to_string(); let name = network.name.clone(); - if socket.exists() { - let report = tsunagi::ipc::unix::set_active(socket, &id, active).await?; + if tsunagi::ipc::is_serving(socket).await { + let report = tsunagi::ipc::set_active(socket, &id, active).await?; match (report.active, report.changed) { (false, true) => println!( "stopped `{}` ({}); everything it has is kept", @@ -1777,8 +1781,8 @@ async fn leave_network( // The running agent does it, because only it can publish the release // while its sessions are still up. - if socket.exists() { - let report = tsunagi::ipc::unix::leave_network(socket, &id).await?; + if tsunagi::ipc::is_serving(socket).await { + let report = tsunagi::ipc::leave_network(socket, &id).await?; println!("left `{}` ({})", report.name, short(&id, 10)); match (report.announced, report.peers_told) { (true, 0) => eprintln!( @@ -1846,7 +1850,7 @@ fn forget_protocol_state(paths: &StoragePaths, network: tsunagi::NetworkId) { async fn wipe(args: WipeArgs) -> Result<(), Box> { let paths = args.paths.resolve()?; let socket = control_socket(&paths, args.control_socket.as_ref()); - if socket.exists() { + if tsunagi::ipc::is_serving(&socket).await { return Err( "stop the agent first: a wipe removes the state it is using, and leaving a \ network properly needs it running anyway" @@ -1883,8 +1887,9 @@ async fn wipe(args: WipeArgs) -> Result<(), Box> { } let removed = tsunagi::storage::wipe(&paths)?; - // A socket file with nothing behind it is a leftover of the same kind. - if tokio::net::UnixStream::connect(&socket).await.is_err() { + // A socket file with nothing behind it is a leftover of the same kind. On + // Windows there is no socket file, so this only ever tidies a Unix one. + if !tsunagi::ipc::is_serving(&socket).await { let _ = std::fs::remove_file(&socket); } println!("removed {} item(s):", removed.entries().count()); @@ -1918,8 +1923,8 @@ async fn dns_command(args: DnsArgs) -> Result<(), Box> { // The running agent, so it takes effect now; it stores the setting too, // so the two can never say different things. - if socket.exists() { - let report = tsunagi::ipc::unix::set_dns(&socket, enable, port).await?; + if tsunagi::ipc::is_serving(&socket).await { + let report = tsunagi::ipc::set_dns(&socket, enable, port).await?; match report { Some(report) => { println!( @@ -2083,8 +2088,8 @@ async fn set_hostname( socket: &std::path::Path, name: &str, ) -> Result<(), Box> { - if socket.exists() { - return match tsunagi::ipc::unix::set_hostname(socket, name).await { + if tsunagi::ipc::is_serving(socket).await { + return match tsunagi::ipc::set_hostname(socket, name).await { Ok(accepted) => { println!("{accepted}"); Ok(()) @@ -2128,7 +2133,7 @@ async fn rotate_key( // The rotation writes, so it needs the directory to itself. Refused up // front rather than after the lock fails, because what a lock failure // says does not tell the reader what to do about it. - if socket.exists() { + if tsunagi::ipc::is_serving(socket).await { return Err( "stop the agent first: replacing the signing key rewrites state it is using".into(), ); @@ -2741,7 +2746,12 @@ fn host_section() -> report::Section { use tsunagi::overlay::{Privilege, probe_net_admin}; match probe_net_admin() { Privilege::Available => { - host.push(Row::new(Health::Good, "privileges", "CAP_NET_ADMIN held")); + let held = if cfg!(target_os = "windows") { + "assumes an elevated process; creation reports if not" + } else { + "CAP_NET_ADMIN held" + }; + host.push(Row::new(Health::Good, "privileges", held)); host.push(Row::new( Health::Good, "interface", @@ -3339,7 +3349,7 @@ async fn up(args: UpArgs) -> Result<(), Box> { // answer to a question nobody asked. Err(tsunagi::Error::StateLocked { path }) => { let socket = control_socket(&paths, args.control_socket.as_ref()); - if socket.exists() { + if tsunagi::ipc::is_serving(&socket).await { return Err(format!( "an agent is already running for {}, and one state directory is one \ agent — it is the device, not a network.\n\n\ @@ -3394,14 +3404,14 @@ async fn up(args: UpArgs) -> Result<(), Box> { let agent = agent.clone(); let plugin = wireguard.clone(); let dns = Arc::clone(&dns); - let source: Arc = Arc::new(AgentControl { + let source: Arc = Arc::new(AgentControl { agent, plugin, dns, paths: paths.clone(), }); let path = control_socket(&paths, args.control_socket.as_ref()); - match tsunagi::ipc::unix::ControlSocket::bind(path, source).await { + match tsunagi::ipc::ControlSocket::bind(path, source).await { Ok(socket) => { println!(" control {}", socket.path().display()); Some(socket) @@ -3643,11 +3653,20 @@ fn system_tun_factory() -> Result, Box Result, Box> { + use tsunagi::overlay::{ManagedTunFactory, WintunProvisioner}; + let provisioner = WintunProvisioner::new()?; + Ok(Arc::new(ManagedTunFactory::new(Arc::new(provisioner)))) +} + /// There is no provisioner for this platform yet. /// /// Refused here rather than at the first packet, and with the one thing that /// does work on every platform named. -#[cfg(not(target_os = "linux"))] +#[cfg(not(any(target_os = "linux", target_os = "windows")))] fn system_tun_factory() -> Result, Box> { Err(format!( "managing the overlay interface is not implemented on {} yet. \ diff --git a/crates/tsunagi-wg-quic/tests/local_control.rs b/crates/tsunagi-wg-quic/tests/local_control.rs index 63f485f..5ca3df4 100644 --- a/crates/tsunagi-wg-quic/tests/local_control.rs +++ b/crates/tsunagi-wg-quic/tests/local_control.rs @@ -12,7 +12,7 @@ use std::time::Duration; use tempfile::TempDir; use tsunagi::dataplane::IpPlugin; use tsunagi::discovery::SharedMemoryDiscovery; -use tsunagi::ipc::unix::{ControlSocket, request_status}; +use tsunagi::ipc::{ControlSocket, request_status}; use tsunagi::ipc::{StatusReport, control_socket_path}; use tsunagi::overlay::MemoryTunFactory; use tsunagi::testing::{config_with, network, wait_for_peers, wait_until}; @@ -20,7 +20,7 @@ use tsunagi::{Agent, BoxFuture}; use tsunagi_wg_quic::{WireguardConfig, WireguardPlugin}; /// Builds the report the way the binary does, from the agent plus the plugin. -fn source(agent: Agent, plugin: Arc) -> Arc { +fn source(agent: Agent, plugin: Arc) -> Arc { Arc::new(move || -> BoxFuture<'static, StatusReport> { let agent = agent.clone(); let plugin = Arc::clone(&plugin); @@ -186,7 +186,7 @@ async fn a_leftover_socket_file_is_replaced_but_a_live_one_is_not() { let dir = TempDir::new().unwrap(); let path = dir.path().join("agent.sock"); - let empty: Arc = + let empty: Arc = Arc::new(|| -> BoxFuture<'static, StatusReport> { Box::pin(async { StatusReport::default() }) }); @@ -238,7 +238,7 @@ fn the_socket_path_is_derived_and_short_enough() { #[derive(Debug)] struct Control(Agent); -impl tsunagi::ipc::unix::ReportSource for Control { +impl tsunagi::ipc::ReportSource for Control { fn report(&self) -> BoxFuture<'_, StatusReport> { Box::pin(async move { StatusReport::default() }) } @@ -324,7 +324,7 @@ async fn a_client_can_leave_a_network_through_the_running_agent() { .await .unwrap(); - let report = tsunagi::ipc::unix::leave_network(&socket_path, &network_id.to_string()) + let report = tsunagi::ipc::leave_network(&socket_path, &network_id.to_string()) .await .unwrap(); assert_eq!(report.name, name.as_str()); @@ -333,7 +333,7 @@ async fn a_client_can_leave_a_network_through_the_running_agent() { assert!(agent.list_networks().await.unwrap().is_empty()); // Asking again names the state it is in rather than failing obscurely. - let err = tsunagi::ipc::unix::leave_network(&socket_path, &network_id.to_string()) + let err = tsunagi::ipc::leave_network(&socket_path, &network_id.to_string()) .await .unwrap_err(); assert!(err.to_string().contains("not a network"), "{err}"); @@ -364,7 +364,7 @@ async fn a_client_can_add_a_network_to_a_running_agent() { .await .unwrap(); - let report = tsunagi::ipc::unix::join_network( + let report = tsunagi::ipc::join_network( &socket_path, second.as_str(), second_secret.encode().as_str(), @@ -382,7 +382,7 @@ async fn a_client_can_add_a_network_to_a_running_agent() { assert!(agent.network_status(joined).await.is_ok()); // Joining the same one again is not an error, and says which it was. - let again = tsunagi::ipc::unix::join_network( + let again = tsunagi::ipc::join_network( &socket_path, second.as_str(), second_secret.encode().as_str(), @@ -393,7 +393,7 @@ async fn a_client_can_add_a_network_to_a_running_agent() { assert_eq!(again.network_id, report.network_id); // A secret that is not one is refused rather than stored. - let err = tsunagi::ipc::unix::join_network(&socket_path, "rubbish", "not-a-secret") + let err = tsunagi::ipc::join_network(&socket_path, "rubbish", "not-a-secret") .await .unwrap_err(); assert!(!err.to_string().is_empty()); diff --git a/crates/tsunagi/Cargo.toml b/crates/tsunagi/Cargo.toml index 21e3ad4..a3775f7 100644 --- a/crates/tsunagi/Cargo.toml +++ b/crates/tsunagi/Cargo.toml @@ -12,11 +12,21 @@ repository.workspace = true [features] default = [] -# A real TUN device, so a plugin can carry actual IP traffic. Needs -# CAP_NET_ADMIN at run time; without it the in-memory device serves tests. -tun-device = ["dep:tun", "dep:rtnetlink", "dep:caps", "dep:futures-util"] -# Telling the operating system where to send its DNS questions. Linux only -# for now; the zone and the server work without it. +# A real TUN device, so a plugin can carry actual IP traffic. On Linux it +# needs CAP_NET_ADMIN; on Windows the process must run elevated and +# `wintun.dll` must be reachable. Without either the in-memory device serves +# tests. The platform-specific dependencies are declared per target, so +# enabling this only pulls what the target it is built for actually uses. +tun-device = [ + "dep:tun", + "dep:rtnetlink", + "dep:caps", + "dep:futures-util", + "dep:netdev", +] +# Telling the operating system where to send its DNS questions. systemd-resolved +# on Linux, the Name Resolution Policy Table on Windows; the zone and the +# server work without it either way. dns-publish = ["dep:zbus"] # The test harness in `testing`, for this crate's own tests and for a # protocol crate's. Not on by default: it is only of use to a test. @@ -24,7 +34,9 @@ testing = ["dep:tempfile", "dep:tracing-subscriber"] [dependencies] iroh.workspace = true -tokio.workspace = true +# `net` for the local control interface: a Unix socket on Unix, a named pipe +# on Windows. Both live behind the same `net` feature. +tokio = { workspace = true, features = ["net"] } serde.workspace = true postcard.workspace = true bytes.workspace = true @@ -69,6 +81,16 @@ zbus = { version = "5.19", default-features = false, features = ["tokio"], optio caps = { version = "0.5", optional = true } futures-util = { version = "0.3", default-features = false, optional = true } +# Windows-only interface provisioning. The Wintun adapter itself is created +# and driven through the `tun` dependency, whose safe wrapper is the only way +# in — loading `wintun.dll` is an `unsafe` call this crate forbids. `netdev` +# (already in the tree through `netwatch`) reads the interface table through a +# safe wrapper, which is how the provisioner observes an adapter by name +# before deciding what to change. Creating an adapter needs the process to be +# elevated and `wintun.dll` to be reachable at run time. +[target.'cfg(target_os = "windows")'.dependencies] +netdev = { version = "0.45", optional = true } + [dev-dependencies] # Its own tests use the harness it publishes. tsunagi = { path = ".", features = ["testing"] } diff --git a/crates/tsunagi/src/dns/publish/mod.rs b/crates/tsunagi/src/dns/publish/mod.rs index 77002be..ca46f15 100644 --- a/crates/tsunagi/src/dns/publish/mod.rs +++ b/crates/tsunagi/src/dns/publish/mod.rs @@ -23,6 +23,11 @@ mod resolved; #[cfg(all(feature = "dns-publish", target_os = "linux"))] pub use resolved::ResolvedPublisher; +#[cfg(all(feature = "dns-publish", target_os = "windows"))] +mod windows; +#[cfg(all(feature = "dns-publish", target_os = "windows"))] +pub use windows::NrptPublisher; + mod unsupported; pub use unsupported::UnsupportedPublisher; @@ -143,6 +148,10 @@ pub fn interface_index(name: &str) -> Option { .ok() } +/// The kernel's index for an interface. +/// +/// Only the systemd-resolved publisher needs one, so off Linux there is +/// nothing to look up and this is always `None`. #[cfg(not(target_os = "linux"))] pub fn interface_index(_name: &str) -> Option { None diff --git a/crates/tsunagi/src/dns/publish/windows.rs b/crates/tsunagi/src/dns/publish/windows.rs new file mode 100644 index 0000000..ed31871 --- /dev/null +++ b/crates/tsunagi/src/dns/publish/windows.rs @@ -0,0 +1,296 @@ +//! Telling the Windows DNS client to send some questions here, through the +//! Name Resolution Policy Table. +//! +//! The NRPT is how Windows does split DNS: a rule says "names under this +//! suffix are resolved by these servers", and names outside every rule are +//! resolved the ordinary way. That is exactly the contract of [`Published`] — +//! route these suffixes here and claim nothing else — so one NRPT rule per +//! suffix is the whole of it. The rules are keyed by namespace, not by +//! interface, so the interface name in [`Published`] is not needed here. +//! +//! # Why PowerShell +//! +//! The NRPT lives in the registry, but a rule only takes effect once the DNS +//! client is told to reload its policy, and that notification is an RPC with +//! no safe wrapper this crate may call — it forbids `unsafe`. The +//! `DnsClient` PowerShell module does the write *and* the reload, so it is +//! used by absolute path from `%SystemRoot%`. Every value passed to it is one +//! this agent produced: the suffixes are validated zone names (letters, +//! digits, `-`, `_`, `.` only) and the servers are addresses it is listening +//! on. They are still single-quoted and escaped when built into the script, +//! so nothing could be read as PowerShell rather than as data. +//! +//! # The port +//! +//! The Windows DNS client always asks on port 53; it cannot be told another. +//! So a server on a different port cannot be reached this way, and rather than +//! configure a rule that points at nothing, this says so — the same honest +//! failure the systemd-resolved path gives for an old resolver with no port. +//! +//! # Privilege +//! +//! Writing an NRPT rule needs an elevated process. An ordinary user is +//! refused, which is reported as its own kind of error because the answer to +//! it — run elevated, or as a service — differs from "this is not Windows". + +use std::collections::BTreeSet; +use std::net::IpAddr; +use std::sync::Mutex; + +use crate::BoxFuture; + +use super::{DnsPublisher, PublishError, Published}; + +/// The port the Windows DNS client is fixed to. +const DNS_PORT: u16 = 53; + +/// Configures the Name Resolution Policy Table through PowerShell. +#[derive(Debug, Default)] +pub struct NrptPublisher { + /// The namespaces last configured, so shutdown knows what to undo. + applied: Mutex>, +} + +impl NrptPublisher { + /// Creates the publisher. Nothing is contacted until [`Self::apply`]. + pub fn new() -> Self { + Self::default() + } +} + +/// A suffix as the NRPT wants it: a leading dot means "this and everything +/// under it". +fn namespace(domain: &str) -> String { + let trimmed = domain.trim_matches('.'); + format!(".{trimmed}") +} + +/// A PowerShell single-quoted literal, with any embedded quote doubled. +fn ps_literal(value: &str) -> String { + format!("'{}'", value.replace('\'', "''")) +} + +/// A PowerShell array literal, `@('a','b')`, or `@()` when empty. +fn ps_array(values: &[String]) -> String { + let items: Vec = values.iter().map(|value| ps_literal(value)).collect(); + format!("@({})", items.join(",")) +} + +/// The script that clears our rules for these namespaces and then adds them. +fn apply_script(namespaces: &[String], servers: &[String]) -> String { + format!( + "$ErrorActionPreference = 'Stop'\n\ + $ns = {ns}\n\ + $srv = {srv}\n\ + foreach ($n in $ns) {{ Get-DnsClientNrptRule | Where-Object {{ $_.Namespace -eq $n }} | \ + ForEach-Object {{ Remove-DnsClientNrptRule -Name $_.Name -Force }} }}\n\ + foreach ($n in $ns) {{ Add-DnsClientNrptRule -Namespace $n -NameServers $srv }}\n", + ns = ps_array(namespaces), + srv = ps_array(servers), + ) +} + +/// The script that clears our rules for these namespaces. +fn revert_script(namespaces: &[String]) -> String { + format!( + "$ErrorActionPreference = 'Stop'\n\ + $ns = {ns}\n\ + foreach ($n in $ns) {{ Get-DnsClientNrptRule | Where-Object {{ $_.Namespace -eq $n }} | \ + ForEach-Object {{ Remove-DnsClientNrptRule -Name $_.Name -Force }} }}\n", + ns = ps_array(namespaces), + ) +} + +/// Turns what PowerShell said on failure into the kind of failure it is. +fn classify(text: &str) -> PublishError { + let lower = text.to_ascii_lowercase(); + if lower.contains("access is denied") + || lower.contains("requires elevation") + || lower.contains("run as administrator") + || lower.contains("administrator privilege") + || lower.contains("permissiondenied") + { + PublishError::Refused(format!("the Windows DNS client refused the change: {text}")) + } else if lower.contains("is not recognized") + || lower.contains("commandnotfoundexception") + || lower.contains("not recognized as the name of a cmdlet") + { + PublishError::Unavailable(format!( + "the DnsClient PowerShell module is not available: {text}" + )) + } else { + PublishError::Failed(format!("the Windows DNS client failed the change: {text}")) + } +} + +/// Runs a PowerShell script from `%SystemRoot%` and maps a failure to a +/// [`PublishError`]. +async fn powershell(script: String) -> Result<(), PublishError> { + let program = powershell_path(); + let output = tokio::task::spawn_blocking(move || { + std::process::Command::new(program) + .args([ + "-NoProfile", + "-NonInteractive", + "-ExecutionPolicy", + "Bypass", + "-Command", + &script, + ]) + .output() + }) + .await + .map_err(|err| PublishError::Failed(format!("could not run PowerShell: {err}")))? + .map_err(|err| PublishError::Unavailable(format!("could not run PowerShell: {err}")))?; + + if output.status.success() { + return Ok(()); + } + let stderr = String::from_utf8_lossy(&output.stderr); + let text = if stderr.trim().is_empty() { + String::from_utf8_lossy(&output.stdout).trim().to_string() + } else { + stderr.trim().to_string() + }; + Err(classify(if text.is_empty() { "no output" } else { &text })) +} + +/// The absolute path to Windows PowerShell, so nothing on `PATH` can stand in +/// for it. +fn powershell_path() -> std::path::PathBuf { + let root = std::env::var_os("SystemRoot").unwrap_or_else(|| r"C:\Windows".into()); + std::path::Path::new(&root) + .join("System32") + .join("WindowsPowerShell") + .join("v1.0") + .join("powershell.exe") +} + +impl DnsPublisher for NrptPublisher { + fn name(&self) -> &str { + "windows-nrpt" + } + + fn apply<'a>(&'a self, published: &'a Published) -> BoxFuture<'a, Result<(), PublishError>> { + Box::pin(async move { + if published.domains.is_empty() { + return Err(PublishError::Unavailable( + "there are no suffixes to route, so there is no rule to write".to_string(), + )); + } + + // Windows can only ask on port 53, so a server anywhere else is + // unreachable this way. Keep the ones it can use. + let servers: BTreeSet = published + .servers + .iter() + .filter(|server| server.port() == DNS_PORT) + .map(|server| server.ip()) + .collect(); + if servers.is_empty() { + let elsewhere = published + .servers + .iter() + .map(|server| server.to_string()) + .collect::>() + .join(", "); + return Err(PublishError::Unavailable(format!( + "the Windows DNS client only asks on port {DNS_PORT}, and the server is on \ + {elsewhere}. Run the server on port {DNS_PORT}, or point your resolver at it \ + yourself." + ))); + } + + let namespaces: Vec = + published.domains.iter().map(|d| namespace(d)).collect(); + let servers: Vec = servers.iter().map(|ip| ip.to_string()).collect(); + + powershell(apply_script(&namespaces, &servers)).await?; + + match self.applied.lock() { + Ok(mut guard) => *guard = namespaces, + Err(poisoned) => *poisoned.into_inner() = namespaces, + } + Ok(()) + }) + } + + fn revert(&self) -> BoxFuture<'_, Result<(), PublishError>> { + Box::pin(async move { + let namespaces = match self.applied.lock() { + Ok(mut guard) => std::mem::take(&mut *guard), + Err(poisoned) => std::mem::take(&mut *poisoned.into_inner()), + }; + if namespaces.is_empty() { + return Ok(()); + } + powershell(revert_script(&namespaces)).await + }) + } +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + + use super::*; + use std::net::SocketAddr; + + #[test] + fn a_suffix_gets_a_leading_dot() { + assert_eq!(namespace("lab"), ".lab"); + assert_eq!(namespace(".lab"), ".lab"); + assert_eq!(namespace("a.b"), ".a.b"); + } + + #[test] + fn a_quote_in_a_value_cannot_escape_the_literal() { + // Zone names never contain a quote, but the escaping is what makes + // that a guarantee rather than a hope. + assert_eq!(ps_literal("a'b"), "'a''b'"); + assert_eq!(ps_array(&["x".into(), "y".into()]), "@('x','y')"); + assert_eq!(ps_array(&[]), "@()"); + } + + #[test] + fn the_apply_script_removes_before_it_adds() { + let script = apply_script(&[".lab".into()], &["10.13.37.69".into()]); + let remove = script.find("Remove-DnsClientNrptRule").unwrap(); + let add = script.find("Add-DnsClientNrptRule").unwrap(); + assert!(remove < add, "a stale rule must go before the new one:\n{script}"); + assert!(script.contains("@('.lab')")); + assert!(script.contains("@('10.13.37.69')")); + } + + #[test] + fn an_access_denied_is_a_refusal_a_missing_cmdlet_is_unavailable() { + assert!(matches!( + classify("Access is denied"), + PublishError::Refused(_) + )); + assert!(matches!( + classify("The term 'Add-DnsClientNrptRule' is not recognized"), + PublishError::Unavailable(_) + )); + assert!(matches!(classify("something else"), PublishError::Failed(_))); + } + + #[tokio::test] + async fn a_server_only_on_a_nonstandard_port_is_unavailable_not_a_failure() { + // Nothing is run: Windows cannot ask there, and saying so is the + // honest answer without touching the DNS client. + let publisher = NrptPublisher::new(); + let published = Published { + interface: "tsun0".into(), + servers: vec![SocketAddr::from(([10, 13, 37, 69], 5354))], + domains: vec!["lab".into()], + }; + let err = publisher.apply(&published).await.unwrap_err(); + assert!(matches!(err, PublishError::Unavailable(_)), "{err}"); + } + + #[tokio::test] + async fn reverting_without_having_applied_does_nothing_and_succeeds() { + NrptPublisher::new().revert().await.unwrap(); + } +} diff --git a/crates/tsunagi/src/ipc/mod.rs b/crates/tsunagi/src/ipc/mod.rs index cfc8097..f4abb33 100644 --- a/crates/tsunagi/src/ipc/mod.rs +++ b/crates/tsunagi/src/ipc/mod.rs @@ -27,11 +27,33 @@ #[cfg(unix)] pub mod unix; +#[cfg(windows)] +pub mod windows; + +// The one transport this build serves on. The two adapters expose the same +// items, so the rest of the crate — and the client wrappers below — name the +// transport through this alias and never a platform directly. +#[cfg(unix)] +use unix as transport; +#[cfg(windows)] +use windows as transport; + +/// Serves the local control interface on this platform's transport. +/// +/// A Unix socket on Unix, a named pipe on Windows; the same API either way. +#[cfg(any(unix, windows))] +pub use transport::ControlSocket; use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; + +use crate::BoxFuture; +use crate::error::{Error, Result}; /// Largest accepted local control message. pub const MAX_MESSAGE_LEN: usize = 1024 * 1024; @@ -382,3 +404,337 @@ impl OverlayPeerReport { self.handshake_secs_ago.is_some() } } + +// --------------------------------------------------------------------------- +// The transport-agnostic half of the local control interface. +// +// Everything below is the same on every platform: the request/response +// framing, the dispatch of a request onto a [`ReportSource`], and the client +// wrappers that ask one question and read one answer. Only the listener and +// the stream differ, and those live in the per-platform `unix` and `windows` +// adapters. Each adapter drives the shared [`serve_connection`] for its +// accepted streams and offers a `connect`/`is_serving` for the client side, +// so the logic here is written once. +// --------------------------------------------------------------------------- + +/// Builds the report that answers a status request, and applies the changes a +/// client asks for. +/// +/// Supplied by the caller, because only the caller knows which plugins are +/// running and what they can report or change. That is what keeps this module +/// free of any knowledge of them. +pub trait ReportSource: Send + Sync + 'static { + /// Produces a fresh report. + fn report(&self) -> BoxFuture<'_, StatusReport>; + + /// Changes the name the agent answers to, returning the accepted form. + /// + /// Defaulted to a refusal so that a source which only reports — the + /// closure impl below, and every test that uses it — stays valid and + /// says plainly that it cannot do this, rather than appearing to. + fn set_hostname( + &self, + _hostname: String, + ) -> BoxFuture<'_, std::result::Result> { + Box::pin(async move { Err("this agent cannot change its hostname".to_string()) }) + } + + /// Leaves a network, publishing a release first. + /// + /// Defaulted to a refusal for the same reason as the above: a source + /// that only reports says so plainly rather than appearing to do it. + fn leave(&self, _network_id: String) -> BoxFuture<'_, std::result::Result> { + Box::pin(async move { Err("this agent cannot leave a network".to_string()) }) + } + + /// Joins a network, or starts one that is configured and not running. + /// + /// Defaulted to a refusal, like the others: a source that only reports + /// says so rather than appearing to have done it. + fn join( + &self, + _name: String, + _secret: String, + ) -> BoxFuture<'_, std::result::Result> { + Box::pin(async move { Err("this agent cannot join a network".to_string()) }) + } + + /// Stops serving a network, or starts serving it again. + /// + /// Defaulted to a refusal, like the others. + fn set_active( + &self, + _network_id: String, + _active: bool, + ) -> BoxFuture<'_, std::result::Result> { + Box::pin(async move { Err("this agent cannot stop or start a network".to_string()) }) + } + + /// Turns the local resolver on or off while the agent runs. + /// + /// Defaulted to a refusal, like the others. + fn set_dns( + &self, + _enable: bool, + _port: Option, + ) -> BoxFuture<'_, std::result::Result, String>> { + Box::pin(async move { Err("this agent cannot serve DNS".to_string()) }) + } +} + +impl ReportSource for F +where + F: Fn() -> BoxFuture<'static, StatusReport> + Send + Sync + 'static, +{ + fn report(&self) -> BoxFuture<'_, StatusReport> { + (self)() + } +} + +/// How long either end waits for the other. +/// +/// A local answer comes from memory, so anything this slow means the agent is +/// wedged rather than busy. Saying so beats waiting: unbounded, one wedged +/// runtime leaves `tsunagi status` hanging with nothing on screen and no way +/// out but Ctrl-C. +pub const EXCHANGE_TIMEOUT: Duration = Duration::from_secs(5); + +/// Marks the wire format of the local control interface. +/// +/// `b"TSN"` followed by the version, so a mismatch is recognised as one +/// instead of being read as a length. The encoding is postcard, which is not +/// self-describing: adding a field to a report changes how the bytes parse, +/// and without this a client one build ahead of its agent reports something +/// like "Found an Option discriminant that wasn't 0 or 1" — which says +/// nothing about the actual problem, that the two are different builds. +/// +/// Bump it whenever [`Request`], [`Response`] or anything they contain +/// changes shape. +pub const CONTROL_PROTOCOL: u32 = u32::from_be_bytes([b'T', b'S', b'N', 13]); + +/// Reads one request off an accepted stream, answers it, writes the response. +/// +/// The per-platform adapters call this for every connection they accept, so +/// the request handling is identical on every transport. +pub(crate) async fn serve_connection(mut stream: S, source: Arc) -> Result<()> +where + S: AsyncRead + AsyncWrite + Unpin, +{ + // Bounded, so a connection that sends nothing cannot hold a task open. + // Only the wait for the request: building the report afterwards takes as + // long as it takes, and cutting it off would answer a live client with a + // closed socket. + let request: Request = + match tokio::time::timeout(EXCHANGE_TIMEOUT, read_message(&mut stream)).await { + Ok(request) => request?, + Err(_) => { + return Err(Error::Timeout { + what: "a local control connection".to_string(), + }); + } + }; + let response = match request { + Request::Status => Response::Status(Box::new(source.report().await)), + Request::SetHostname(hostname) => match source.set_hostname(hostname).await { + Ok(accepted) => Response::Hostname(accepted), + Err(reason) => Response::Error(reason), + }, + Request::Leave(network_id) => match source.leave(network_id).await { + Ok(report) => Response::Left(report), + Err(reason) => Response::Error(reason), + }, + Request::Join { name, secret } => match source.join(name, secret).await { + Ok(report) => Response::Joined(report), + Err(reason) => Response::Error(reason), + }, + Request::SetActive { network_id, active } => { + match source.set_active(network_id, active).await { + Ok(report) => Response::Active(report), + Err(reason) => Response::Error(reason), + } + } + Request::Dns { enable, port } => match source.set_dns(enable, port).await { + Ok(report) => Response::Dns(report), + Err(reason) => Response::Error(reason), + }, + }; + write_message(&mut stream, &response).await +} + +/// Connects, sends one request and reads the answer, all within `within`. +/// +/// The bound covers the whole exchange rather than each read: an agent that +/// answers the header and then stops is as stuck as one that never answers. +async fn exchange(path: &Path, request: &Request, within: Duration) -> Result { + let attempt = async { + let mut stream = transport::connect(path).await?; + write_message(&mut stream, request).await?; + read_message::(&mut stream).await + }; + + match tokio::time::timeout(within, attempt).await { + Ok(result) => result, + Err(_) => Err(Error::Timeout { + what: format!("the agent at {}", path.display()), + }), + } +} + +/// Whether an agent is serving the control interface at `path`. +/// +/// Used to tell a running agent from a leftover: on Unix a socket file with +/// nothing behind it, on Windows nothing at all. +pub async fn is_serving(path: impl AsRef) -> bool { + transport::probe(path.as_ref()).await +} + +/// Asks a running agent for its status. +pub async fn request_status(path: impl AsRef) -> Result { + let path = path.as_ref(); + match exchange(path, &Request::Status, EXCHANGE_TIMEOUT).await? { + Response::Status(report) => Ok(*report), + Response::Error(reason) => Err(Error::Storage(reason)), + other => Err(Error::Storage(format!("unexpected answer: {other:?}"))), + } +} + +/// Asks a running agent to answer to a different name. +/// +/// Returns the name it accepted, which is the canonical form of what was +/// asked for and may differ from it. +pub async fn set_hostname(path: impl AsRef, hostname: &str) -> Result { + let path = path.as_ref(); + let request = Request::SetHostname(hostname.to_string()); + match exchange(path, &request, EXCHANGE_TIMEOUT).await? { + Response::Hostname(accepted) => Ok(accepted), + Response::Error(reason) => Err(Error::Storage(reason)), + other => Err(Error::Storage(format!("unexpected answer: {other:?}"))), + } +} + +/// Asks a running agent to leave a network. +/// +/// The agent publishes the release and removes the network; this only +/// carries the request and the outcome. +pub async fn leave_network(path: impl AsRef, network_id: &str) -> Result { + let path = path.as_ref(); + let request = Request::Leave(network_id.to_string()); + match exchange(path, &request, EXCHANGE_TIMEOUT).await? { + Response::Left(report) => Ok(report), + Response::Error(reason) => Err(Error::Storage(reason)), + other => Err(Error::Storage(format!("unexpected answer: {other:?}"))), + } +} + +/// Asks a running agent to join a network. +/// +/// The one way to add a network to an agent that is already up: the state +/// directory belongs to one live agent, so a second `up` cannot. +pub async fn join_network( + path: impl AsRef, + name: &str, + secret: &str, +) -> Result { + let path = path.as_ref(); + let request = Request::Join { + name: name.to_string(), + secret: secret.to_string(), + }; + match exchange(path, &request, EXCHANGE_TIMEOUT).await? { + Response::Joined(report) => Ok(report), + Response::Error(reason) => Err(Error::Storage(reason)), + other => Err(Error::Storage(format!("unexpected answer: {other:?}"))), + } +} + +/// Asks a running agent to stop serving a network, or to serve it again. +pub async fn set_active( + path: impl AsRef, + network_id: &str, + active: bool, +) -> Result { + let path = path.as_ref(); + let request = Request::SetActive { + network_id: network_id.to_string(), + active, + }; + match exchange(path, &request, EXCHANGE_TIMEOUT).await? { + Response::Active(report) => Ok(report), + Response::Error(reason) => Err(Error::Storage(reason)), + other => Err(Error::Storage(format!("unexpected answer: {other:?}"))), + } +} + +/// Turns the running agent's local resolver on or off. +pub async fn set_dns( + path: impl AsRef, + enable: bool, + port: Option, +) -> Result> { + let path = path.as_ref(); + match exchange(path, &Request::Dns { enable, port }, EXCHANGE_TIMEOUT).await? { + Response::Dns(report) => Ok(report), + Response::Error(reason) => Err(Error::Storage(reason)), + other => Err(Error::Storage(format!("unexpected answer: {other:?}"))), + } +} + +pub(crate) async fn write_message(stream: &mut S, value: &T) -> Result<()> +where + S: AsyncWrite + Unpin, + T: serde::Serialize, +{ + let encoded = postcard::to_stdvec(value) + .map_err(|err| Error::Storage(format!("cannot encode a control message: {err}")))?; + if encoded.len() > MAX_MESSAGE_LEN { + return Err(Error::Storage("control message is too large".into())); + } + let len = encoded.len() as u32; + stream + .write_all(&CONTROL_PROTOCOL.to_be_bytes()) + .await + .map_err(io_error)?; + stream + .write_all(&len.to_be_bytes()) + .await + .map_err(io_error)?; + stream.write_all(&encoded).await.map_err(io_error)?; + stream.flush().await.map_err(io_error) +} + +pub(crate) async fn read_message(stream: &mut S) -> Result +where + T: for<'de> serde::Deserialize<'de>, + S: AsyncRead + Unpin, +{ + let mut header = [0u8; 4]; + stream.read_exact(&mut header).await.map_err(io_error)?; + let version = u32::from_be_bytes(header); + if version != CONTROL_PROTOCOL { + return Err(Error::Storage(format!( + "the other end speaks control protocol {version:#010x} and this build speaks \ + {CONTROL_PROTOCOL:#010x}; they are different builds of tsunagi, so restart the \ + agent with the binary you are running now" + ))); + } + + stream.read_exact(&mut header).await.map_err(io_error)?; + let len = u32::from_be_bytes(header) as usize; + // Checked before allocating, exactly as on the network. + if len > MAX_MESSAGE_LEN { + return Err(Error::Storage(format!( + "control message of {len} bytes exceeds the {MAX_MESSAGE_LEN} byte limit" + ))); + } + let mut payload = vec![0u8; len]; + stream.read_exact(&mut payload).await.map_err(io_error)?; + postcard::from_bytes(&payload) + .map_err(|err| Error::Storage(format!("cannot decode a control message: {err}"))) +} + +pub(crate) fn io_error(source: std::io::Error) -> Error { + Error::Io { + path: PathBuf::from(""), + source, + } +} diff --git a/crates/tsunagi/src/ipc/unix.rs b/crates/tsunagi/src/ipc/unix.rs index e385adf..d8922a9 100644 --- a/crates/tsunagi/src/ipc/unix.rs +++ b/crates/tsunagi/src/ipc/unix.rs @@ -1,99 +1,31 @@ //! A Unix socket adapter for the local control interface. //! -//! One of possibly several adapters; see [`super`]. It serves exactly the -//! requests in [`Request`] and nothing else, and it is reachable only by a -//! process that can open a file inside the agent's owner-only state -//! directory. +//! One of the per-platform adapters; see [`super`]. It provides only what is +//! particular to a Unix socket — binding a listener, connecting a client, and +//! the owner-only permissions — and hands every accepted connection to the +//! shared [`serve_connection`](super::serve_connection). The request framing, +//! the dispatch and the client wrappers are the same on every platform and +//! live in [`super`], re-exported here so `ipc::unix::request_status` and its +//! siblings keep resolving. +//! +//! It is reachable only by a process that can open a file inside the agent's +//! owner-only state directory, and the socket itself is created mode `0600`. use std::path::{Path, PathBuf}; use std::sync::Arc; -use std::time::Duration; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::{UnixListener, UnixStream}; use tokio::task::JoinHandle; -use crate::BoxFuture; use crate::error::{Error, Result}; -use super::{ - ActiveReport, DnsReport, JoinedReport, LeftReport, MAX_MESSAGE_LEN, Request, Response, - StatusReport, +// The transport-agnostic surface, re-exported so this module is a complete +// view of the local control interface on its own. +pub use super::{ + CONTROL_PROTOCOL, EXCHANGE_TIMEOUT, ReportSource, is_serving, join_network, leave_network, + request_status, set_active, set_dns, set_hostname, }; -/// Builds the report that answers a status request. -/// -/// Supplied by the caller, because only the caller knows which plugins are -/// running and what they can report. That is what keeps this module free of -/// any knowledge of them. -pub trait ReportSource: Send + Sync + 'static { - /// Produces a fresh report. - fn report(&self) -> BoxFuture<'_, StatusReport>; - - /// Changes the name the agent answers to, returning the accepted form. - /// - /// Defaulted to a refusal so that a source which only reports — the - /// closure impl below, and every test that uses it — stays valid and - /// says plainly that it cannot do this, rather than appearing to. - fn set_hostname( - &self, - _hostname: String, - ) -> BoxFuture<'_, std::result::Result> { - Box::pin(async move { Err("this agent cannot change its hostname".to_string()) }) - } - - /// Leaves a network, publishing a release first. - /// - /// Defaulted to a refusal for the same reason as the above: a source - /// that only reports says so plainly rather than appearing to do it. - fn leave(&self, _network_id: String) -> BoxFuture<'_, std::result::Result> { - Box::pin(async move { Err("this agent cannot leave a network".to_string()) }) - } - - /// Joins a network, or starts one that is configured and not running. - /// - /// Defaulted to a refusal, like the others: a source that only reports - /// says so rather than appearing to have done it. - fn join( - &self, - _name: String, - _secret: String, - ) -> BoxFuture<'_, std::result::Result> { - Box::pin(async move { Err("this agent cannot join a network".to_string()) }) - } - - /// Stops serving a network, or starts serving it again. - /// - /// Defaulted to a refusal, like the others. - fn set_active( - &self, - _network_id: String, - _active: bool, - ) -> BoxFuture<'_, std::result::Result> { - Box::pin(async move { Err("this agent cannot stop or start a network".to_string()) }) - } - - /// Turns the local resolver on or off while the agent runs. - /// - /// Defaulted to a refusal, like the others. - fn set_dns( - &self, - _enable: bool, - _port: Option, - ) -> BoxFuture<'_, std::result::Result, String>> { - Box::pin(async move { Err("this agent cannot serve DNS".to_string()) }) - } -} - -impl ReportSource for F -where - F: Fn() -> BoxFuture<'static, StatusReport> + Send + Sync + 'static, -{ - fn report(&self) -> BoxFuture<'_, StatusReport> { - (self)() - } -} - /// Serves the local control interface on a Unix socket. #[derive(Debug)] pub struct ControlSocket { @@ -115,18 +47,14 @@ impl ControlSocket { } if path.exists() { - match UnixStream::connect(&path).await { - Ok(_) => { - return Err(Error::StateLocked { path: path.clone() }); - } - // Nothing is listening, so the file is a leftover. - Err(_) => { - std::fs::remove_file(&path).map_err(|source| Error::Io { - path: path.clone(), - source, - })?; - } + if is_serving(&path).await { + return Err(Error::StateLocked { path: path.clone() }); } + // Nothing is listening, so the file is a leftover. + std::fs::remove_file(&path).map_err(|source| Error::Io { + path: path.clone(), + source, + })?; } let listener = UnixListener::bind(&path).map_err(|source| Error::Io { @@ -166,7 +94,6 @@ impl Drop for ControlSocket { } } -#[cfg(unix)] fn restrict(path: &Path) -> Result<()> { use std::os::unix::fs::PermissionsExt; std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)).map_err(|source| { @@ -184,248 +111,35 @@ async fn serve(listener: UnixListener, source: Arc) { }; let source = Arc::clone(&source); tokio::spawn(async move { - if let Err(err) = handle(stream, source).await { + if let Err(err) = super::serve_connection(stream, source).await { tracing::debug!(%err, "local control request failed"); } }); } } -async fn handle(mut stream: UnixStream, source: Arc) -> Result<()> { - // Bounded, so a connection that sends nothing cannot hold a task open. - // Only the wait for the request: building the report afterwards takes as - // long as it takes, and cutting it off would answer a live client with a - // closed socket. - let request: Request = - match tokio::time::timeout(EXCHANGE_TIMEOUT, read_message(&mut stream)).await { - Ok(request) => request?, - Err(_) => { - return Err(Error::Timeout { - what: "a local control connection".to_string(), - }); - } - }; - let response = match request { - Request::Status => Response::Status(Box::new(source.report().await)), - Request::SetHostname(hostname) => match source.set_hostname(hostname).await { - Ok(accepted) => Response::Hostname(accepted), - Err(reason) => Response::Error(reason), - }, - Request::Leave(network_id) => match source.leave(network_id).await { - Ok(report) => Response::Left(report), - Err(reason) => Response::Error(reason), - }, - Request::Join { name, secret } => match source.join(name, secret).await { - Ok(report) => Response::Joined(report), - Err(reason) => Response::Error(reason), - }, - Request::SetActive { network_id, active } => { - match source.set_active(network_id, active).await { - Ok(report) => Response::Active(report), - Err(reason) => Response::Error(reason), - } - } - Request::Dns { enable, port } => match source.set_dns(enable, port).await { - Ok(report) => Response::Dns(report), - Err(reason) => Response::Error(reason), - }, - }; - write_message(&mut stream, &response).await -} - -/// How long either end waits for the other. -/// -/// A local answer comes from memory, so anything this slow means the agent is -/// wedged rather than busy. Saying so beats waiting: unbounded, one wedged -/// runtime leaves `tsunagi status` hanging with nothing on screen and no way -/// out but Ctrl-C. -const EXCHANGE_TIMEOUT: Duration = Duration::from_secs(5); - -/// Asks a running agent for its status. -pub async fn request_status(path: impl AsRef) -> Result { - let path = path.as_ref(); - match exchange(path, &Request::Status, EXCHANGE_TIMEOUT).await? { - Response::Status(report) => Ok(*report), - Response::Error(reason) => Err(Error::Storage(reason)), - other => Err(Error::Storage(format!("unexpected answer: {other:?}"))), - } -} - -/// Asks a running agent to answer to a different name. -/// -/// Returns the name it accepted, which is the canonical form of what was -/// asked for and may differ from it. -pub async fn set_hostname(path: impl AsRef, hostname: &str) -> Result { - let path = path.as_ref(); - let request = Request::SetHostname(hostname.to_string()); - match exchange(path, &request, EXCHANGE_TIMEOUT).await? { - Response::Hostname(accepted) => Ok(accepted), - Response::Error(reason) => Err(Error::Storage(reason)), - other => Err(Error::Storage(format!("unexpected answer: {other:?}"))), - } -} - -/// Connects, sends one request and reads the answer, all within `within`. -/// -/// The bound covers the whole exchange rather than each read: an agent that -/// answers the header and then stops is as stuck as one that never answers. -async fn exchange(path: &Path, request: &Request, within: Duration) -> Result { - let attempt = async { - let mut stream = UnixStream::connect(path) - .await - .map_err(|source| Error::Io { - path: path.to_path_buf(), - source, - })?; - write_message(&mut stream, request).await?; - read_message::(&mut stream).await - }; - - match tokio::time::timeout(within, attempt).await { - Ok(result) => result, - Err(_) => Err(Error::Timeout { - what: format!("the agent at {}", path.display()), - }), - } -} - -/// Asks a running agent to leave a network. -/// -/// The agent publishes the release and removes the network; this only -/// carries the request and the outcome. -pub async fn leave_network(path: impl AsRef, network_id: &str) -> Result { - let path = path.as_ref(); - let request = Request::Leave(network_id.to_string()); - match exchange(path, &request, EXCHANGE_TIMEOUT).await? { - Response::Left(report) => Ok(report), - Response::Error(reason) => Err(Error::Storage(reason)), - other => Err(Error::Storage(format!("unexpected answer: {other:?}"))), - } -} - -/// Asks a running agent to join a network. -/// -/// The one way to add a network to an agent that is already up: the state -/// directory belongs to one live agent, so a second `up` cannot. -pub async fn join_network( - path: impl AsRef, - name: &str, - secret: &str, -) -> Result { - let path = path.as_ref(); - let request = Request::Join { - name: name.to_string(), - secret: secret.to_string(), - }; - match exchange(path, &request, EXCHANGE_TIMEOUT).await? { - Response::Joined(report) => Ok(report), - Response::Error(reason) => Err(Error::Storage(reason)), - other => Err(Error::Storage(format!("unexpected answer: {other:?}"))), - } -} - -/// Asks a running agent to stop serving a network, or to serve it again. -pub async fn set_active( - path: impl AsRef, - network_id: &str, - active: bool, -) -> Result { - let path = path.as_ref(); - let request = Request::SetActive { - network_id: network_id.to_string(), - active, - }; - match exchange(path, &request, EXCHANGE_TIMEOUT).await? { - Response::Active(report) => Ok(report), - Response::Error(reason) => Err(Error::Storage(reason)), - other => Err(Error::Storage(format!("unexpected answer: {other:?}"))), - } -} - -/// Turns the running agent's local resolver on or off. -pub async fn set_dns( - path: impl AsRef, - enable: bool, - port: Option, -) -> Result> { - let path = path.as_ref(); - match exchange(path, &Request::Dns { enable, port }, EXCHANGE_TIMEOUT).await? { - Response::Dns(report) => Ok(report), - Response::Error(reason) => Err(Error::Storage(reason)), - other => Err(Error::Storage(format!("unexpected answer: {other:?}"))), - } -} - -/// Marks the wire format of the local control socket. -/// -/// `b"TSN"` followed by the version, so a mismatch is recognised as one -/// instead of being read as a length. The encoding is postcard, which is not -/// self-describing: adding a field to a report changes how the bytes parse, -/// and without this a client one build ahead of its agent reports something -/// like "Found an Option discriminant that wasn't 0 or 1" — which says -/// nothing about the actual problem, that the two are different builds. -/// -/// Bump it whenever [`Request`], [`Response`] or anything they contain -/// changes shape. -pub const CONTROL_PROTOCOL: u32 = u32::from_be_bytes([b'T', b'S', b'N', 13]); - -async fn write_message(stream: &mut UnixStream, value: &T) -> Result<()> { - let encoded = postcard::to_stdvec(value) - .map_err(|err| Error::Storage(format!("cannot encode a control message: {err}")))?; - if encoded.len() > MAX_MESSAGE_LEN { - return Err(Error::Storage("control message is too large".into())); - } - let len = encoded.len() as u32; - stream - .write_all(&CONTROL_PROTOCOL.to_be_bytes()) - .await - .map_err(io_error)?; - stream - .write_all(&len.to_be_bytes()) - .await - .map_err(io_error)?; - stream.write_all(&encoded).await.map_err(io_error)?; - stream.flush().await.map_err(io_error) -} - -async fn read_message serde::Deserialize<'de>>(stream: &mut UnixStream) -> Result { - let mut header = [0u8; 4]; - stream.read_exact(&mut header).await.map_err(io_error)?; - let version = u32::from_be_bytes(header); - if version != CONTROL_PROTOCOL { - return Err(Error::Storage(format!( - "the other end speaks control protocol {version:#010x} and this build speaks \ - {CONTROL_PROTOCOL:#010x}; they are different builds of tsunagi, so restart the \ - agent with the binary you are running now" - ))); - } - - stream.read_exact(&mut header).await.map_err(io_error)?; - let len = u32::from_be_bytes(header) as usize; - // Checked before allocating, exactly as on the network. - if len > MAX_MESSAGE_LEN { - return Err(Error::Storage(format!( - "control message of {len} bytes exceeds the {MAX_MESSAGE_LEN} byte limit" - ))); - } - let mut payload = vec![0u8; len]; - stream.read_exact(&mut payload).await.map_err(io_error)?; - postcard::from_bytes(&payload) - .map_err(|err| Error::Storage(format!("cannot decode a control message: {err}"))) -} - -fn io_error(source: std::io::Error) -> Error { - Error::Io { - path: PathBuf::from(""), +/// Connects a client to the socket at `path`. +pub(crate) async fn connect(path: &Path) -> Result { + UnixStream::connect(path).await.map_err(|source| Error::Io { + path: path.to_path_buf(), source, - } + }) +} + +/// Whether an agent is listening on the socket at `path`. +pub(crate) async fn probe(path: &Path) -> bool { + UnixStream::connect(path).await.is_ok() } #[cfg(test)] mod tests { #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + use std::time::Duration; + + use super::super::{Request, Response, StatusReport, exchange}; use super::*; + use crate::BoxFuture; #[tokio::test] async fn a_silent_agent_is_reported_rather_than_waited_out() { diff --git a/crates/tsunagi/src/ipc/windows.rs b/crates/tsunagi/src/ipc/windows.rs new file mode 100644 index 0000000..06fb75a --- /dev/null +++ b/crates/tsunagi/src/ipc/windows.rs @@ -0,0 +1,284 @@ +//! A named-pipe adapter for the local control interface. +//! +//! The Windows counterpart of [`super::unix`]. It provides only what is +//! particular to a named pipe — creating the server instance, opening a client +//! and turning the state directory path into a pipe name — and hands every +//! accepted connection to the shared [`serve_connection`](super::serve_connection). +//! The framing, the dispatch and the client wrappers are the same on every +//! platform and live in [`super`], re-exported here so `ipc::windows::request_status` +//! and its siblings resolve just as their Unix equivalents do. +//! +//! # From a path to a pipe +//! +//! The rest of the agent addresses the control interface by a filesystem path, +//! the same one a Unix socket would live at. A named pipe has no filesystem +//! path, so the path is hashed into a stable name under `\\.\pipe\`. The agent +//! and a client derive it the same way from the same path, so neither has to +//! be told where the other put it. +//! +//! # Who may connect +//! +//! The pipe rejects clients from other machines, and takes the process's +//! default security, under which the creating user has full access. Narrowing +//! that further with an explicit ACL needs a raw security-descriptor call this +//! crate forbids, so it is not attempted; on a single-user machine, and against +//! other unprivileged users, the default is the practical equivalent of the +//! owner-only Unix socket. There is no leftover to clean up: a pipe exists only +//! while its server does. + +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; + +use sha2::{Digest, Sha256}; +use tokio::net::windows::named_pipe::{ClientOptions, NamedPipeClient, ServerOptions}; +use tokio::task::JoinHandle; + +use crate::error::{Error, Result}; + +pub use super::{ + CONTROL_PROTOCOL, EXCHANGE_TIMEOUT, ReportSource, is_serving, join_network, leave_network, + request_status, set_active, set_dns, set_hostname, +}; + +/// `ERROR_ACCESS_DENIED`: what creating the first pipe instance returns when +/// one already exists, so another agent already owns the name. +const ERROR_ACCESS_DENIED: i32 = 5; +/// `ERROR_PIPE_BUSY`: every instance is serving a client right now. A server +/// is there; the client only has to wait for a free instance. +const ERROR_PIPE_BUSY: i32 = 231; + +/// Serves the local control interface on a named pipe. +#[derive(Debug)] +pub struct ControlSocket { + path: PathBuf, + task: Option>, +} + +impl ControlSocket { + /// Binds the pipe and starts serving. + /// + /// Claiming the first instance of the name is what detects a second agent: + /// if the name already exists, the create is refused and that is reported + /// as the state being locked, so two live agents never share one pipe. + pub async fn bind(path: impl AsRef, source: Arc) -> Result { + let path = path.as_ref().to_path_buf(); + let name = pipe_name(&path); + + let server = match ServerOptions::new() + .first_pipe_instance(true) + .reject_remote_clients(true) + .create(&name) + { + Ok(server) => server, + Err(err) if err.raw_os_error() == Some(ERROR_ACCESS_DENIED) => { + return Err(Error::StateLocked { path }); + } + Err(source) => return Err(Error::Io { path, source }), + }; + + let task = tokio::spawn(serve(name, server, source)); + Ok(Self { + path, + task: Some(task), + }) + } + + /// The path the pipe name was derived from. + pub fn path(&self) -> &Path { + &self.path + } + + /// Stops serving. The pipe goes with the server, so there is nothing to + /// remove. + pub async fn shutdown(mut self) { + if let Some(task) = self.task.take() { + task.abort(); + let _ = task.await; + } + } +} + +impl Drop for ControlSocket { + fn drop(&mut self) { + if let Some(task) = self.task.take() { + task.abort(); + } + } +} + +/// Accepts one client at a time, keeping a fresh instance ready for the next. +/// +/// A named pipe server instance serves a single client, so a new instance is +/// created as soon as one is taken — otherwise a second `tsunagi status` while +/// the first is mid-flight would find nothing listening. +async fn serve(name: String, first: tokio::net::windows::named_pipe::NamedPipeServer, source: Arc) { + let mut server = first; + loop { + if server.connect().await.is_err() { + match next_instance(&name) { + Some(next) => { + server = next; + continue; + } + None => return, + } + } + let connected = server; + match next_instance(&name) { + Some(next) => server = next, + None => { + // Nothing left to accept the next client on, but the one in + // hand is still answered before the loop ends. + let source = Arc::clone(&source); + tokio::spawn(async move { + let _ = super::serve_connection(connected, source).await; + }); + return; + } + } + let source = Arc::clone(&source); + tokio::spawn(async move { + if let Err(err) = super::serve_connection(connected, source).await { + tracing::debug!(%err, "local control request failed"); + } + }); + } +} + +/// Creates the next pipe instance, or `None` if the name can no longer be +/// served. +fn next_instance(name: &str) -> Option { + match ServerOptions::new().reject_remote_clients(true).create(name) { + Ok(server) => Some(server), + Err(err) => { + tracing::debug!(%err, "cannot create the next control pipe instance"); + None + } + } +} + +/// Connects a client to the pipe for `path`, waiting out a busy pipe. +/// +/// The wait is bounded by the caller's exchange timeout, so a pipe that is +/// busy forever is given up on rather than spun on. +pub(crate) async fn connect(path: &Path) -> Result { + let name = pipe_name(path); + loop { + match ClientOptions::new().open(&name) { + Ok(client) => return Ok(client), + Err(err) if err.raw_os_error() == Some(ERROR_PIPE_BUSY) => { + tokio::time::sleep(Duration::from_millis(50)).await; + } + Err(source) => { + return Err(Error::Io { + path: path.to_path_buf(), + source, + }); + } + } + } +} + +/// Whether an agent is serving the pipe for `path`. +/// +/// A busy pipe still means a server is there; only a name nothing has created +/// counts as not serving. +pub(crate) async fn probe(path: &Path) -> bool { + let name = pipe_name(path); + match ClientOptions::new().open(&name) { + Ok(_) => true, + Err(err) => err.raw_os_error() == Some(ERROR_PIPE_BUSY), + } +} + +/// The pipe name for a control-socket path. +/// +/// A hash rather than the path itself, because a pipe name may not contain the +/// separators and drive letters a path does, and because two agents with +/// different state directories must never collide. +fn pipe_name(path: &Path) -> String { + let digest = Sha256::digest(path.as_os_str().as_encoded_bytes()); + format!(r"\\.\pipe\tsunagi-{}", hex::encode(&digest[..16])) +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + + use std::time::Duration; + + use super::super::{Request, Response, StatusReport, exchange}; + use super::*; + use crate::BoxFuture; + + #[test] + fn one_path_maps_to_one_name_and_two_paths_do_not_collide() { + let a = pipe_name(Path::new(r"C:\a\agent.sock")); + let b = pipe_name(Path::new(r"C:\b\agent.sock")); + assert!(a.starts_with(r"\\.\pipe\tsunagi-"), "{a}"); + assert_eq!(a, pipe_name(Path::new(r"C:\a\agent.sock"))); + assert_ne!(a, b); + } + + #[tokio::test] + async fn a_request_makes_the_round_trip_over_a_real_pipe() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("agent.sock"); + let source: Arc = Arc::new(|| -> BoxFuture<'static, StatusReport> { + Box::pin(async { StatusReport::default() }) + }); + + assert!(!is_serving(&path).await, "nothing serves the pipe yet"); + let control = ControlSocket::bind(&path, source).await.unwrap(); + assert!(is_serving(&path).await, "the agent serves it now"); + + let answer = exchange(&path, &Request::Status, EXCHANGE_TIMEOUT) + .await + .unwrap(); + assert!(matches!(answer, Response::Status(_))); + + control.shutdown().await; + } + + #[tokio::test] + async fn a_silent_agent_is_reported_rather_than_waited_out() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("agent.sock"); + // A server instance that accepts and then never answers, like a + // wedged agent. Unbounded, the exchange below would never return. + let name = pipe_name(&path); + let server = ServerOptions::new() + .first_pipe_instance(true) + .create(&name) + .unwrap(); + let _accept = tokio::spawn(async move { + let _ = server.connect().await; + // Hold the connection open, answering nothing. + tokio::time::sleep(Duration::from_secs(30)).await; + }); + + let error = exchange(&path, &Request::Status, Duration::from_millis(200)) + .await + .expect_err("a silent agent cannot be reported as healthy"); + assert!(matches!(error, Error::Timeout { .. }), "{error:?}"); + } + + #[tokio::test] + async fn a_second_agent_on_the_same_pipe_is_refused() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("agent.sock"); + let source: Arc = Arc::new(|| -> BoxFuture<'static, StatusReport> { + Box::pin(async { StatusReport::default() }) + }); + + let first = ControlSocket::bind(&path, Arc::clone(&source)).await.unwrap(); + let second = ControlSocket::bind(&path, source).await; + assert!( + matches!(second, Err(Error::StateLocked { .. })), + "{second:?}" + ); + + first.shutdown().await; + } +} diff --git a/crates/tsunagi/src/overlay/mod.rs b/crates/tsunagi/src/overlay/mod.rs index c053a58..09abd15 100644 --- a/crates/tsunagi/src/overlay/mod.rs +++ b/crates/tsunagi/src/overlay/mod.rs @@ -56,3 +56,6 @@ pub use tun::{MemoryTun, MemoryTunFactory, TunDevice, TunFactory, TunRequest, ad #[cfg(all(feature = "tun-device", target_os = "linux"))] pub use provision::NetlinkProvisioner; + +#[cfg(all(feature = "tun-device", target_os = "windows"))] +pub use provision::WintunProvisioner; diff --git a/crates/tsunagi/src/overlay/provision/mod.rs b/crates/tsunagi/src/overlay/provision/mod.rs index 5e8b83a..1cd0769 100644 --- a/crates/tsunagi/src/overlay/provision/mod.rs +++ b/crates/tsunagi/src/overlay/provision/mod.rs @@ -20,9 +20,11 @@ //! pure and tested on every platform, and only the execution is behind //! [`InterfaceProvisioner`]. //! -//! Three implementations: +//! Four implementations: //! //! * `NetlinkProvisioner` on Linux, which needs `CAP_NET_ADMIN`. +//! * `WintunProvisioner` on Windows, a Wintun adapter configured with `netsh`, +//! which needs an elevated process. //! * [`MockProvisioner`], an in-memory host used by the tests. //! * [`UnsupportedProvisioner`] elsewhere, which fails with an explanation //! and a pointer at the manual route rather than pretending to work. @@ -51,6 +53,11 @@ mod linux; #[cfg(all(feature = "tun-device", target_os = "linux"))] pub use linux::NetlinkProvisioner; +#[cfg(all(feature = "tun-device", target_os = "windows"))] +mod windows; +#[cfg(all(feature = "tun-device", target_os = "windows"))] +pub use windows::WintunProvisioner; + mod privilege; pub use privilege::{Privilege, probe_net_admin}; diff --git a/crates/tsunagi/src/overlay/provision/privilege.rs b/crates/tsunagi/src/overlay/provision/privilege.rs index 033819f..2f31e39 100644 --- a/crates/tsunagi/src/overlay/provision/privilege.rs +++ b/crates/tsunagi/src/overlay/provision/privilege.rs @@ -60,10 +60,19 @@ impl Privilege { #[cfg(all(feature = "tun-device", target_os = "linux"))] pub use linux_impl::{NetAdmin, probe_net_admin}; -#[cfg(not(all(feature = "tun-device", target_os = "linux")))] +#[cfg(all(feature = "tun-device", target_os = "windows"))] +pub use windows_impl::probe_net_admin; + +#[cfg(not(any( + all(feature = "tun-device", target_os = "linux"), + all(feature = "tun-device", target_os = "windows") +)))] pub use other_impl::probe_net_admin; -#[cfg(not(all(feature = "tun-device", target_os = "linux")))] +#[cfg(not(any( + all(feature = "tun-device", target_os = "linux"), + all(feature = "tun-device", target_os = "windows") +)))] mod other_impl { use super::Privilege; @@ -73,6 +82,25 @@ mod other_impl { } } +#[cfg(all(feature = "tun-device", target_os = "windows"))] +mod windows_impl { + use super::Privilege; + + /// Whether this process can configure interfaces. + /// + /// Windows has no capability to hold and lower the way Linux does: + /// creating an adapter simply needs the process to be elevated. Reading + /// whether it *is* elevated means inspecting the process token through a + /// raw call this crate forbids, so the probe is deliberately optimistic — + /// it reports that the platform can manage interfaces — and the real + /// check is left to Wintun's adapter creation, which fails with a precise + /// message when the process is not elevated. This matches how the Linux + /// path treats the open itself as the honest answer. + pub fn probe_net_admin() -> Privilege { + Privilege::Available + } +} + #[cfg(all(feature = "tun-device", target_os = "linux"))] mod linux_impl { use caps::{CapSet, Capability}; diff --git a/crates/tsunagi/src/overlay/provision/windows.rs b/crates/tsunagi/src/overlay/provision/windows.rs new file mode 100644 index 0000000..811eba5 --- /dev/null +++ b/crates/tsunagi/src/overlay/provision/windows.rs @@ -0,0 +1,502 @@ +//! Managing the overlay interface on Windows, through Wintun and `netsh`. +//! +//! The shape is the same as the Linux provisioner: observe what is there, +//! [`plan_changes`](super::plan_changes) to decide what to do, apply the +//! difference. Only the two mechanisms differ. +//! +//! # Creating the interface +//! +//! The interface is a Wintun adapter, created through the `tun` dependency's +//! safe wrapper — [`open_tun`](super::super::tun) — because loading +//! `wintun.dll` is an `unsafe` call and this crate forbids `unsafe`. The +//! adapter is created, not made persistent: closing the last handle to it +//! removes it, so it goes away when the agent does, however the agent ends. +//! The reader and writer inside the device hold that handle, so the adapter +//! lives exactly as long as the device does. +//! +//! # Configuring it +//! +//! Addresses and MTU are applied with `netsh`. On Linux the same work is done +//! in process over a netlink socket; here there is no equivalent that this +//! crate may call, because the in-process route is the IP Helper API and that +//! is `unsafe`. `netsh` is the supported tool for the job, it is invoked by +//! absolute path from `%SystemRoot%` so nothing on `PATH` can stand in for it, +//! and every value handed to it is one this agent derived — an interface name +//! from the network id, an address it allocated — never anything a peer said. +//! +//! # Observing it +//! +//! The interface table is read through `netdev`, whose enumeration is a safe +//! wrapper over the same IP Helper API. An adapter is matched by the friendly +//! name it was created with. A match that carries a default gateway is treated +//! as foreign and left alone: a tsunagi overlay interface never has one, so a +//! gateway is the mark of a real device that happens to share the name. +//! +//! # What Windows cannot do that Linux can +//! +//! Wintun exposes no way to delete an adapter this process did not create, so +//! a leftover from a run that was killed is *reused* rather than replaced: its +//! addresses are flushed and the planned ones applied, which reaches the same +//! end state. A reused adapter is not removed on exit, only closed — that is a +//! Wintun limitation, and the common case of a clean start and a clean stop is +//! unaffected. + +use std::net::{IpAddr, Ipv4Addr}; +use std::sync::{Arc, Mutex}; + +use crate::BoxFuture; +use crate::overlay::OverlayError; + +use super::super::config::Cidr; +use super::super::tun::{TunDevice, TunRequest, open_tun}; +use super::privilege::{Privilege, probe_net_admin}; +use super::{ + InterfacePlan, InterfaceProvisioner, InterfaceState, LinkKind, Provisioned, plan_changes, +}; + +/// Manages the overlay interface with Wintun and `netsh`. +#[derive(Debug)] +pub struct WintunProvisioner { + /// Interfaces this process created, so they are adjusted rather than + /// replaced. See [`plan_changes`]. + ours: Mutex>, + /// Devices kept alive for as long as the interface should exist. Dropping + /// one closes the adapter, which removes it if this process created it. + held: Mutex)>>, +} + +impl WintunProvisioner { + /// Creates the provisioner, after checking the platform supports one. + /// + /// Whether the process is actually elevated is not checked here — that + /// cannot be read without an `unsafe` call — so a missing privilege + /// surfaces at adapter creation with a precise message, the same way the + /// Linux path treats the open itself as the honest answer. + pub fn new() -> Result { + match probe_net_admin() { + Privilege::Available => {} + Privilege::Missing(reason) => return Err(OverlayError::Unavailable(reason)), + Privilege::Unsupported => { + return Err(OverlayError::Unavailable( + "interface management is not compiled in".to_string(), + )); + } + } + Ok(Self { + ours: Mutex::new(Vec::new()), + held: Mutex::new(Vec::new()), + }) + } + + fn is_ours(&self, name: &str) -> bool { + lock(&self.ours) + .iter() + .any(|owned| owned.eq_ignore_ascii_case(name)) + } + + /// Opens the Wintun adapter, which is what creates it. + fn create_device(&self, plan: &InterfacePlan) -> Result, OverlayError> { + let request = TunRequest::bare(plan.name.clone(), plan.mtu); + open_tun(&request) + } +} + +impl InterfaceProvisioner for WintunProvisioner { + fn name(&self) -> &str { + "wintun" + } + + fn reconcile<'a>( + &'a self, + plan: &'a InterfacePlan, + ) -> BoxFuture<'a, Result> { + Box::pin(async move { + let current = observe(&plan.name).await; + let changes = plan_changes(¤t, plan, self.is_ours(&plan.name))?; + + // Wintun cannot delete an adapter this process did not create, so a + // "delete then create" from `plan_changes` becomes "flush then + // reuse": clear the stale addresses off the interface that is + // there, and let the create step below reopen the same adapter. + // The end state — an adapter carrying exactly the plan's addresses + // — is identical. + let flush: Vec = if changes.delete_link { + tracing::info!( + interface = %plan.name, + "reusing an abandoned interface left by an earlier run" + ); + current.addresses.clone() + } else { + Vec::new() + }; + + for cidr in flush.iter().chain(changes.remove.iter()) { + // Best effort: an address that is already gone is the outcome + // wanted, and refusing to start over one is not worth it. + if let Err(err) = address_del(&plan.name, *cidr).await { + tracing::debug!(interface = %plan.name, %cidr, %err, "could not remove address"); + } + } + + let device = if changes.create_link { + let device = self.create_device(plan)?; + lock(&self.ours).push(plan.name.clone()); + lock(&self.held).push((plan.name.clone(), Arc::clone(&device))); + Some(device) + } else { + None + }; + + // A failure past this point leaves an interface that exists but + // cannot carry traffic. Undo the bookkeeping for one we just + // created so it is not mistaken for a working interface, and drop + // the device so the adapter closes. + let configured = configure(plan, &changes).await; + if let Err(err) = configured { + if changes.create_link { + lock(&self.held).retain(|(held, _)| held != &plan.name); + lock(&self.ours).retain(|owned| owned != &plan.name); + } + return Err(err); + } + + Ok(Provisioned { changes, device }) + }) + } + + fn remove<'a>(&'a self, name: &'a str) -> BoxFuture<'a, Result<(), OverlayError>> { + Box::pin(async move { + // Dropping the device closes the adapter, which removes it if this + // process created it. A reused adapter is only closed, not removed; + // flushing its addresses first keeps no stale configuration behind. + let current = observe(name).await; + for cidr in ¤t.addresses { + if let Err(err) = address_del(name, *cidr).await { + tracing::debug!(interface = %name, %cidr, %err, "could not remove address"); + } + } + lock(&self.held).retain(|(held, _)| !held.eq_ignore_ascii_case(name)); + lock(&self.ours).retain(|owned| !owned.eq_ignore_ascii_case(name)); + Ok(()) + }) + } +} + +fn lock(mutex: &Mutex) -> std::sync::MutexGuard<'_, T> { + match mutex.lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + } +} + +/// `fe80::/10` and `169.254.0.0/16`, which Windows assigns on its own. +fn is_link_local(addr: IpAddr) -> bool { + match addr { + IpAddr::V4(addr) => addr.is_link_local(), + IpAddr::V6(addr) => (addr.segments()[0] & 0xffc0) == 0xfe80, + } +} + +/// What the interface named `name` currently looks like. +/// +/// Never fails: a name nothing is using reads as [`InterfaceState::absent`], +/// which is what the plan then acts on. +async fn observe(name: &str) -> InterfaceState { + let name = name.to_string(); + let found = tokio::task::spawn_blocking(move || { + netdev::get_interfaces() + .into_iter() + .find(|iface| match &iface.friendly_name { + Some(friendly) => friendly.eq_ignore_ascii_case(&name), + None => false, + }) + }) + .await + .ok() + .flatten(); + + let Some(iface) = found else { + return InterfaceState::absent(); + }; + + let mut addresses: Vec = Vec::new(); + for net in &iface.ipv4 { + if let Ok(cidr) = Cidr::new(IpAddr::V4(net.addr()), net.prefix_len()) + && !is_link_local(cidr.addr) + { + addresses.push(cidr); + } + } + for net in &iface.ipv6 { + if let Ok(cidr) = Cidr::new(IpAddr::V6(net.addr()), net.prefix_len()) + && !is_link_local(cidr.addr) + { + addresses.push(cidr); + } + } + addresses.sort(); + + // A tsunagi overlay interface only ever carries the addresses it was given + // and no gateway. One that has a gateway is a real device sharing the name, + // so it is foreign and never touched. + let kind = if iface.gateway.is_some() { + LinkKind::Foreign("gatewayed".to_string()) + } else { + LinkKind::Tun + }; + + InterfaceState { + kind, + // Wintun reports an adapter with a live session as up and an abandoned + // one as down, so "up" doubles as "some process is holding it". + attached: iface.is_up(), + up: iface.is_up(), + mtu: iface.mtu.unwrap_or(0), + addresses, + } +} + +/// Applies MTU, link state and addresses. +async fn configure(plan: &InterfacePlan, changes: &super::Changes) -> Result<(), OverlayError> { + if let Some(mtu) = changes.set_mtu { + set_mtu(&plan.name, mtu).await?; + } + for cidr in &changes.add { + address_add(&plan.name, *cidr).await?; + } + if changes.bring_up { + // Best effort: a Wintun adapter with a session is up already, and a + // failure to nudge it is not a reason to fail the whole reconcile. + if let Err(err) = set_up(&plan.name).await { + tracing::debug!(interface = %plan.name, %err, "could not enable the interface"); + } + } + Ok(()) +} + +/// The dotted netmask for an IPv4 prefix length. +fn ipv4_mask(prefix_len: u8) -> Ipv4Addr { + if prefix_len == 0 { + Ipv4Addr::UNSPECIFIED + } else { + Ipv4Addr::from(u32::MAX << (32 - u32::from(prefix_len))) + } +} + +/// The `netsh` arguments that add an address to an interface. +fn address_add_args(name: &str, cidr: Cidr) -> Vec { + match cidr.addr { + IpAddr::V4(addr) => vec![ + "interface".into(), + "ipv4".into(), + "add".into(), + "address".into(), + format!("name={name}"), + format!("address={addr}"), + format!("mask={}", ipv4_mask(cidr.prefix_len)), + "store=active".into(), + ], + IpAddr::V6(addr) => vec![ + "interface".into(), + "ipv6".into(), + "add".into(), + "address".into(), + format!("interface={name}"), + format!("address={addr}/{}", cidr.prefix_len), + "store=active".into(), + ], + } +} + +/// The `netsh` arguments that remove an address from an interface. +fn address_del_args(name: &str, cidr: Cidr) -> Vec { + match cidr.addr { + IpAddr::V4(addr) => vec![ + "interface".into(), + "ipv4".into(), + "delete".into(), + "address".into(), + format!("name={name}"), + format!("address={addr}"), + "store=active".into(), + ], + IpAddr::V6(addr) => vec![ + "interface".into(), + "ipv6".into(), + "delete".into(), + "address".into(), + format!("interface={name}"), + format!("address={addr}"), + "store=active".into(), + ], + } +} + +/// The `netsh` arguments that set an interface's MTU, one call per family. +fn mtu_args(name: &str, family: &str, mtu: u32) -> Vec { + vec![ + "interface".into(), + family.into(), + "set".into(), + "subinterface".into(), + name.into(), + format!("mtu={mtu}"), + "store=active".into(), + ] +} + +async fn address_add(name: &str, cidr: Cidr) -> Result<(), OverlayError> { + netsh(address_add_args(name, cidr)).await +} + +async fn address_del(name: &str, cidr: Cidr) -> Result<(), OverlayError> { + netsh(address_del_args(name, cidr)).await +} + +/// Sets the MTU on both families. +/// +/// A family that is turned off on the adapter cannot take an MTU, so this +/// fails only when *neither* would — one family carrying the plan's MTU is +/// enough, and refusing the whole interface because the other is disabled +/// would be wrong. +async fn set_mtu(name: &str, mtu: u32) -> Result<(), OverlayError> { + let v4 = netsh(mtu_args(name, "ipv4", mtu)).await; + let v6 = netsh(mtu_args(name, "ipv6", mtu)).await; + match (&v4, &v6) { + (Err(_), Err(_)) => v4, + _ => { + if let Err(err) = &v4 { + tracing::debug!(interface = %name, %err, "could not set the IPv4 MTU"); + } + if let Err(err) = &v6 { + tracing::debug!(interface = %name, %err, "could not set the IPv6 MTU"); + } + Ok(()) + } + } +} + +async fn set_up(name: &str) -> Result<(), OverlayError> { + netsh(vec![ + "interface".into(), + "set".into(), + "interface".into(), + format!("name={name}"), + "admin=enabled".into(), + ]) + .await +} + +/// Runs `netsh` from `%SystemRoot%\System32`, so nothing on `PATH` can stand +/// in for it, and turns a non-zero exit into an error carrying what it said. +async fn netsh(args: Vec) -> Result<(), OverlayError> { + let program = system32("netsh.exe"); + let display = format!("netsh {}", args.join(" ")); + let output = tokio::task::spawn_blocking(move || { + std::process::Command::new(program).args(&args).output() + }) + .await + .map_err(|err| OverlayError::Other(format!("could not run {display}: {err}")))? + .map_err(|err| OverlayError::Unavailable(format!("could not run {display}: {err}")))?; + + if output.status.success() { + return Ok(()); + } + let message = { + let stderr = String::from_utf8_lossy(&output.stderr); + let text = if stderr.trim().is_empty() { + String::from_utf8_lossy(&output.stdout).trim().to_string() + } else { + stderr.trim().to_string() + }; + if text.is_empty() { + format!("exit code {}", output.status) + } else { + text + } + }; + Err(OverlayError::Unavailable(format!("{display} failed: {message}"))) +} + +/// The absolute path to a program in `System32`. +fn system32(exe: &str) -> std::path::PathBuf { + let root = std::env::var_os("SystemRoot").unwrap_or_else(|| r"C:\Windows".into()); + std::path::Path::new(&root).join("System32").join(exe) +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + + use super::*; + use std::net::Ipv6Addr; + + fn v4(last: u8, prefix: u8) -> Cidr { + Cidr::new(IpAddr::V4(Ipv4Addr::new(10, 13, 37, last)), prefix).unwrap() + } + + fn v6(last: u16) -> Cidr { + Cidr::new(IpAddr::V6(Ipv6Addr::new(0xfd00, 0, 0, 0, 0, 0, 0, last)), 64).unwrap() + } + + #[test] + fn a_prefix_length_becomes_the_right_dotted_mask() { + assert_eq!(ipv4_mask(24), Ipv4Addr::new(255, 255, 255, 0)); + assert_eq!(ipv4_mask(16), Ipv4Addr::new(255, 255, 0, 0)); + assert_eq!(ipv4_mask(32), Ipv4Addr::new(255, 255, 255, 255)); + assert_eq!(ipv4_mask(0), Ipv4Addr::UNSPECIFIED); + } + + #[test] + fn a_v4_address_is_added_by_name_with_a_dotted_mask() { + let args = address_add_args("tsun0", v4(69, 24)); + assert!(args.contains(&"ipv4".to_string())); + assert!(args.contains(&"add".to_string())); + assert!(args.contains(&"name=tsun0".to_string())); + assert!(args.contains(&"address=10.13.37.69".to_string())); + assert!(args.contains(&"mask=255.255.255.0".to_string())); + } + + #[test] + fn a_v6_address_is_added_by_interface_with_a_prefix() { + let args = address_add_args("tsun0", v6(1)); + assert!(args.contains(&"ipv6".to_string())); + assert!(args.contains(&"interface=tsun0".to_string())); + assert!(args.contains(&"address=fd00::1/64".to_string())); + } + + #[test] + fn deleting_an_address_names_no_mask() { + let args = address_del_args("tsun0", v4(69, 24)); + assert!(args.contains(&"delete".to_string())); + assert!(args.contains(&"address=10.13.37.69".to_string())); + assert!(!args.iter().any(|arg| arg.starts_with("mask="))); + } + + #[test] + fn the_mtu_is_set_per_family_on_the_subinterface() { + let args = mtu_args("tsun0", "ipv4", 1280); + assert!(args.contains(&"subinterface".to_string())); + assert!(args.contains(&"tsun0".to_string())); + assert!(args.contains(&"mtu=1280".to_string())); + } + + #[test] + fn link_local_addresses_are_not_ours_to_manage() { + assert!(is_link_local(IpAddr::V6(Ipv6Addr::new( + 0xfe80, 0, 0, 0, 0, 0, 0, 1 + )))); + assert!(is_link_local(IpAddr::V4(Ipv4Addr::new(169, 254, 1, 1)))); + assert!(!is_link_local(IpAddr::V4(Ipv4Addr::new(10, 13, 37, 1)))); + assert!(!is_link_local(IpAddr::V6(Ipv6Addr::new( + 0xfd00, 0, 0, 0, 0, 0, 0, 1 + )))); + } + + #[test] + fn system32_is_absolute_and_ends_with_the_program() { + let path = system32("netsh.exe"); + assert!(path.is_absolute(), "{path:?}"); + assert!(path.ends_with("netsh.exe"), "{path:?}"); + } +} diff --git a/crates/tsunagi/src/overlay/tun.rs b/crates/tsunagi/src/overlay/tun.rs index 3abace7..f2cf472 100644 --- a/crates/tsunagi/src/overlay/tun.rs +++ b/crates/tsunagi/src/overlay/tun.rs @@ -312,13 +312,17 @@ mod system { /// A real TUN interface. /// - /// Created by opening `/dev/net/tun`, which needs `CAP_NET_ADMIN` and is - /// why [`open_tun`] is only ever called from - /// [`provision`](super::super::provision), where that capability is - /// raised for the length of the call and no longer. + /// On Linux it is created by opening `/dev/net/tun`, which needs + /// `CAP_NET_ADMIN`; on Windows it is a Wintun adapter, which needs the + /// process to be elevated and `wintun.dll` to be reachable. Either way + /// [`open_tun`] is only ever called from + /// [`provision`](super::super::provision), which holds whatever privilege + /// the platform requires for the length of the call and no longer. /// - /// It is deliberately **not** made persistent, so the kernel removes the - /// interface when this value is dropped — however the process ends. + /// It is deliberately **not** made persistent, so the operating system + /// removes the interface when this value is dropped — however the process + /// ends. (A Wintun adapter created this way is likewise torn down when the + /// last handle to it closes, which the reader and writer below hold.) pub struct SystemTun { name: String, mtu: u32, @@ -383,6 +387,7 @@ mod system { pub(crate) fn open_tun(request: &TunRequest) -> Result, OverlayError> { let mut config = tun::Configuration::default(); config.tun_name(&request.name); + #[cfg(target_os = "linux")] config.platform_config(|platform| { // The crate's own root check is not the check we want: this holds // CAP_NET_ADMIN without being root. Whether the open succeeds is @@ -391,14 +396,16 @@ mod system { }); // Packet information stays off, so reads and writes are raw IP // packets. `ip tuntap add ... mode tun` also defaults to no packet - // information, so the flags match when attaching to one. + // information, so the flags match when attaching to one. The address + // and MTU are left off the configuration on purpose: the provisioner + // applies those to the interface separately, the same way on every + // platform, so the plan stays the single source of what it carries. let device = tun::create_as_async(&config).map_err(|err| { OverlayError::Unavailable(format!( - "cannot create the TUN interface `{}`: {err}. Creating one needs \ - CAP_NET_ADMIN; grant it with `setcap cap_net_admin+p`, or run with \ - `--no-tun` to keep the tunnels off the operating system.", - request.name + "cannot create the TUN interface `{}`: {err}. {}", + request.name, + open_hint() )) })?; @@ -410,4 +417,24 @@ mod system { writer: Mutex::new(writer), }) as Arc) } + + /// The platform-appropriate tail of a "cannot create the interface" error: + /// what privilege it needs and the fallback that always works. + fn open_hint() -> &'static str { + #[cfg(target_os = "linux")] + { + "Creating one needs CAP_NET_ADMIN; grant it with `setcap cap_net_admin+p`, \ + or run with `--no-tun` to keep the tunnels off the operating system." + } + #[cfg(target_os = "windows")] + { + "Creating one needs an elevated process and `wintun.dll` on the search path; \ + run as Administrator with the DLL beside the executable, or run with `--no-tun` \ + to keep the tunnels off the operating system." + } + #[cfg(not(any(target_os = "linux", target_os = "windows")))] + { + "Run with `--no-tun` to keep the tunnels off the operating system." + } + } }