diff --git a/src/bin/tsunagi.rs b/src/bin/tsunagi.rs index ce10c3d..f3993b4 100644 --- a/src/bin/tsunagi.rs +++ b/src/bin/tsunagi.rs @@ -45,6 +45,8 @@ enum Command { Id(PathArgs), /// Joins a network and runs until interrupted. Up(UpArgs), + /// Asks a running agent what it is doing. + Status(StatusArgs), /// Prints the one-time privileged setup for the overlay interface. /// /// Run its output once as root, then run `tsunagi up` as an ordinary @@ -53,6 +55,16 @@ enum Command { TunSetup(TunSetupArgs), } +#[derive(Debug, Args)] +struct StatusArgs { + #[command(flatten)] + paths: PathArgs, + + /// Control socket to talk to. Derived from the state directory by default. + #[arg(long)] + control_socket: Option, +} + #[derive(Debug, Args)] struct TunSetupArgs { #[command(flatten)] @@ -203,6 +215,10 @@ struct UpArgs { /// How often to print a status summary, in seconds. Zero disables it. #[arg(long, default_value_t = 15)] status_interval: u64, + + /// Control socket to serve. Derived from the state directory by default. + #[arg(long)] + control_socket: Option, } /// Reads the shared secret from an argument or a file. @@ -292,9 +308,36 @@ async fn run(command: Command) -> Result<(), Box> { Command::Id(paths) => show_id(paths).await, Command::Up(args) => up(args).await, Command::TunSetup(args) => tun_setup(args).await, + Command::Status(args) => status(args).await, } } +/// Path of the local control socket for a state directory. +fn control_socket(paths: &StoragePaths, override_path: Option<&PathBuf>) -> PathBuf { + match override_path { + Some(path) => path.clone(), + None => tsunagi::ipc::control_socket_path(&paths.state_dir), + } +} + +async fn status(args: StatusArgs) -> Result<(), Box> { + let paths = args.paths.resolve()?; + let socket = control_socket(&paths, args.control_socket.as_ref()); + if !socket.exists() { + return Err(format!( + "no agent is running for {} (no control socket at {})", + paths.state_dir.display(), + socket.display() + ) + .into()); + } + let report = tsunagi::ipc::unix::request_status(&socket) + .await + .map_err(|err| format!("cannot reach the agent at {}: {err}", socket.display()))?; + print!("{}", report.render()); + Ok(()) +} + /// Works out the interface name and overlay address, then prints the /// privileged commands that prepare it. /// @@ -508,6 +551,31 @@ async fn up(args: UpArgs) -> Result<(), Box> { agent.endpoint_id() ); } + // Serve `tsunagi status` for as long as this agent runs. Failing to bind + // is not fatal: the agent itself works fine without it. + let control = { + let agent = agent.clone(); + let plugin = wireguard.clone(); + let source: Arc = Arc::new( + move || -> tsunagi::BoxFuture<'static, tsunagi::ipc::StatusReport> { + let agent = agent.clone(); + let plugin = plugin.clone(); + Box::pin(async move { build_report(&agent, plugin.as_deref()).await }) + }, + ); + let path = control_socket(&paths, args.control_socket.as_ref()); + match tsunagi::ipc::unix::ControlSocket::bind(path, source).await { + Ok(socket) => { + println!(" control {}", socket.path().display()); + Some(socket) + } + Err(err) => { + eprintln!("warning: `tsunagi status` will not work: {err}"); + None + } + } + }; + println!("Press Ctrl-C to stop.\n"); let status_every = @@ -538,11 +606,114 @@ async fn up(args: UpArgs) -> Result<(), Box> { } } + if let Some(control) = control { + control.shutdown().await; + } agent.shutdown().await; println!("stopped."); Ok(()) } +/// Collects a status report from the agent and, when present, the WireGuard +/// plugin. The two are combined here because only this binary knows about +/// both. +async fn build_report( + agent: &Agent, + wireguard: Option<&WireguardPlugin>, +) -> tsunagi::ipc::StatusReport { + use tsunagi::ipc::{NetworkReport, OverlayPeerReport, OverlayReport, PeerReport, StatusReport}; + + let Ok(status) = agent.status().await else { + return StatusReport::default(); + }; + + let networks = status + .networks + .iter() + .map(|network| { + let overlay = wireguard + .and_then(|plugin| plugin.overview(network.network_id)) + .map(|view| OverlayReport { + interface: view.interface.clone(), + mtu: view.mtu, + address: view.overlay_address.to_string(), + prefix: view.overlay_prefix.to_string(), + prefix_len: view.overlay_prefix_len, + peers: view + .peers + .iter() + .map(|peer| OverlayPeerReport { + public_key: peer.public_key.to_string(), + address: peer.overlay_address.to_string(), + handshake_secs_ago: peer + .tunnel + .as_ref() + .and_then(|tunnel| tunnel.health.since_handshake) + .map(|since| since.as_secs()), + tx_packets: peer + .tunnel + .as_ref() + .map_or(0, |tunnel| tunnel.stats.tx_packets), + rx_packets: peer + .tunnel + .as_ref() + .map_or(0, |tunnel| tunnel.stats.rx_packets), + dropped: peer.tunnel.as_ref().map_or(0, |tunnel| { + tunnel.stats.dropped_wrong_source + tunnel.stats.dropped_oversize + }), + protocol_errors: peer + .tunnel + .as_ref() + .map_or(0, |tunnel| tunnel.stats.protocol_errors), + path: peer + .tunnel + .as_ref() + .map(|tunnel| tunnel.path.clone()) + .unwrap_or_else(|| "no data link".into()), + }) + .collect(), + unroutable_packets: view.unroutable_packets, + multicast_packets: view.multicast_packets, + }); + + NetworkReport { + name: network.name.to_string(), + network_id: network.network_id.to_string(), + active: matches!(network.state, tsunagi::agent::NetworkState::Active), + peers: network + .peers + .iter() + .map(|peer| PeerReport { + endpoint_id: peer.endpoint_id.to_string(), + hostname: peer.hostname.clone(), + transport: format!("{:?}", peer.transport), + rtt_ms: peer.rtt.map(|rtt| rtt.as_millis() as u64), + }) + .collect(), + dial_failures: network.metrics.dial_failures, + handshake_failures: network.metrics.handshake_failures, + control_messages: ( + network.metrics.control_messages_sent, + network.metrics.control_messages_received, + ), + overlay, + } + }) + .collect(); + + StatusReport { + endpoint_id: status.endpoint_id.to_string(), + hostname: status.hostname.clone(), + bound_sockets: status + .bound_sockets + .iter() + .map(ToString::to_string) + .collect(), + cache_healthy: status.cache_healthy, + networks, + } +} + /// Resolves when the process is asked to stop. /// /// Both Ctrl-C and `SIGTERM` are handled, so a service manager stopping the diff --git a/src/dataplane/wireguard/device.rs b/src/dataplane/wireguard/device.rs index 00f8678..a51e6fd 100644 --- a/src/dataplane/wireguard/device.rs +++ b/src/dataplane/wireguard/device.rs @@ -186,6 +186,7 @@ struct Inner { routes: RwLock>, next_index: AtomicU32, unroutable: AtomicU64, + multicast: AtomicU64, } impl std::fmt::Debug for Inner { @@ -215,6 +216,7 @@ impl WireguardDevice { routes: RwLock::new(HashMap::new()), next_index: AtomicU32::new(1), unroutable: AtomicU64::new(0), + multicast: AtomicU64::new(0), }); let reader = tokio::spawn(read_from_os(Arc::clone(&inner))); @@ -327,10 +329,22 @@ impl WireguardDevice { peers } - /// Packets the operating system sent that no peer owns the address for. + /// Unicast packets the operating system sent to an address no peer owns. + /// + /// A non-zero value means something tried to reach a host that is not in + /// the overlay. pub fn unroutable_packets(&self) -> u64 { self.inner.unroutable.load(Ordering::Relaxed) } + + /// Multicast packets dropped. + /// + /// Expected and harmless: Linux emits multicast listener and router + /// solicitation traffic on any IPv6 interface, and this overlay is + /// unicast only. Counted separately so it does not look like a fault. + pub fn multicast_packets(&self) -> u64 { + self.inner.multicast.load(Ordering::Relaxed) + } } impl Drop for WireguardDevice { @@ -407,6 +421,12 @@ async fn read_from_os(inner: Arc) { inner.unroutable.fetch_add(1, Ordering::Relaxed); continue; }; + // The kernel emits multicast on every IPv6 interface. The overlay is + // unicast only, so this is dropped, but it is not a fault. + if destination.is_multicast() { + inner.multicast.fetch_add(1, Ordering::Relaxed); + continue; + } let target = read_lock(&inner.routes).get(&destination).copied(); let Some(target) = target else { inner.unroutable.fetch_add(1, Ordering::Relaxed); diff --git a/src/dataplane/wireguard/plugin.rs b/src/dataplane/wireguard/plugin.rs index 7ddb0fb..735c11e 100644 --- a/src/dataplane/wireguard/plugin.rs +++ b/src/dataplane/wireguard/plugin.rs @@ -152,8 +152,10 @@ pub struct NetworkOverview { pub overlay_prefix_len: u8, /// Peers this agent knows about. pub peers: Vec, - /// Packets the operating system sent to an address no peer owns. + /// Unicast packets the operating system sent to an address no peer owns. pub unroutable_packets: u64, + /// Multicast packets dropped. Expected, not a fault. + pub multicast_packets: u64, } impl NetworkOverview { @@ -327,6 +329,11 @@ impl WireguardPlugin { .as_ref() .map(|device| device.unroutable_packets()) .unwrap_or(0), + multicast_packets: state + .device + .as_ref() + .map(|device| device.multicast_packets()) + .unwrap_or(0), }) } diff --git a/src/ipc/mod.rs b/src/ipc/mod.rs new file mode 100644 index 0000000..ef27bed --- /dev/null +++ b/src/ipc/mod.rs @@ -0,0 +1,271 @@ +//! The local control interface. +//! +//! This is how a command line tool asks a running agent what it is doing. It +//! is deliberately **an adapter over the public API, not part of the core**: +//! nothing in [`crate::agent`] knows this module exists, so a Windows named +//! pipe or an authenticated loopback socket can be added beside it without +//! touching anything else. +//! +//! It is also a different interface from the peer-to-peer control protocol in +//! [`crate::proto`]. That one is between machines and is authenticated by the +//! network secret; this one is between processes on one machine and is +//! authorised by filesystem permissions. +//! +//! # Access +//! +//! The socket lives inside the agent's state directory, which is owner-only, +//! and the socket itself is created with mode `0600`. There is no +//! unauthenticated listener reachable by other local users, and nothing is +//! exposed on the network. +//! +//! # Wire format +//! +//! Length-prefixed postcard, with the same frame bounds the network protocol +//! uses. The report types here are a stable data transfer format of their own +//! rather than the crate's internal structures, so internal refactors do not +//! silently change what a client sees. + +#[cfg(unix)] +pub mod unix; + +use std::path::{Path, PathBuf}; + +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; + +/// Largest accepted local control message. +pub const MAX_MESSAGE_LEN: usize = 1024 * 1024; + +/// Where the control socket for a state directory lives. +/// +/// A Unix socket path is limited to around 100 bytes, which a state directory +/// nested deeply enough will exceed. So the runtime directory is preferred +/// when the platform provides one — which is also where a runtime socket +/// belongs — with a short name derived from the state directory so that two +/// agents with different state never share a socket. The state directory +/// itself is the fallback. +/// +/// Both the agent and the client compute this the same way, so neither has to +/// be told where the other put it. +pub fn control_socket_path(state_dir: &Path) -> PathBuf { + let digest = Sha256::digest(state_dir.as_os_str().as_encoded_bytes()); + let tag = hex::encode(&digest[..8]); + + if let Some(runtime) = std::env::var_os("XDG_RUNTIME_DIR") { + let runtime = PathBuf::from(runtime); + if runtime.is_absolute() { + return runtime.join("tsunagi").join(format!("{tag}.sock")); + } + } + state_dir.join("agent.sock") +} + +/// What a client asks for. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[non_exhaustive] +pub enum Request { + /// Report what the agent is doing. + Status, +} + +/// What the agent answers. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[non_exhaustive] +pub enum Response { + /// A status report. + Status(Box), + /// The request could not be served. + Error(String), +} + +/// Everything the agent is doing, in one snapshot. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct StatusReport { + /// This device's persistent endpoint id. + pub endpoint_id: String, + /// Hostname announced to peers. + pub hostname: String, + /// Sockets the endpoint is bound to. + pub bound_sockets: Vec, + /// Whether the disposable cache is usable. + pub cache_healthy: bool, + /// One entry per configured network. + pub networks: Vec, +} + +/// One network. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct NetworkReport { + /// Network name. + pub name: String, + /// Public network identifier. + pub network_id: String, + /// Whether the network is running locally. + pub active: bool, + /// Authenticated control plane peers. + pub peers: Vec, + /// Outbound dials that failed. + pub dial_failures: u64, + /// Handshakes rejected in either direction. + pub handshake_failures: u64, + /// Control messages sent and received. + pub control_messages: (u64, u64), + /// The overlay, when an IP plugin is running one. + pub overlay: Option, +} + +/// One control plane peer. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct PeerReport { + /// The peer's endpoint id. + pub endpoint_id: String, + /// Hostname it announced, if any. + pub hostname: Option, + /// `Direct`, `Relay` or `Unknown`, as the transport reports it. + pub transport: String, + /// Round-trip time in milliseconds, when a path is selected. + pub rtt_ms: Option, +} + +/// The WireGuard overlay of one network. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct OverlayReport { + /// Packet interface name. + pub interface: String, + /// Interface MTU. + pub mtu: u32, + /// This agent's overlay address. + pub address: String, + /// The subnet every member shares. + pub prefix: String, + /// Prefix length of that subnet. + pub prefix_len: u8, + /// One entry per overlay peer. + pub peers: Vec, + /// Unicast packets sent to an address no peer owns. + pub unroutable_packets: u64, + /// Multicast packets dropped. Expected, not a fault. + pub multicast_packets: u64, +} + +/// One overlay peer. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct OverlayPeerReport { + /// The peer's WireGuard public key. + pub public_key: String, + /// Its overlay address. + pub address: String, + /// Seconds since the last WireGuard handshake. + /// + /// `None` means the tunnel has never handshaken and cannot carry traffic. + pub handshake_secs_ago: Option, + /// Packets encrypted and sent to this peer. + pub tx_packets: u64, + /// Packets decrypted from this peer. + pub rx_packets: u64, + /// Data packets dropped: wrong source address, or too large for the path. + pub dropped: u64, + /// WireGuard protocol errors. + /// + /// A few are normal while a tunnel is being set up, because both ends + /// start a handshake at once and one of the two is discarded. + pub protocol_errors: u64, + /// What the transport reports about the path in use. + pub path: String, +} + +impl OverlayPeerReport { + /// Whether the tunnel has handshaken and can carry traffic. + pub fn is_up(&self) -> bool { + self.handshake_secs_ago.is_some() + } +} + +impl StatusReport { + /// Renders the report the way the command line prints it. + pub fn render(&self) -> String { + use std::fmt::Write as _; + let mut out = String::new(); + let _ = writeln!(out, "endpoint {}", self.endpoint_id); + let _ = writeln!(out, "hostname {}", self.hostname); + let _ = writeln!(out, "bound {}", self.bound_sockets.join(", ")); + if !self.cache_healthy { + let _ = writeln!(out, "cache UNAVAILABLE"); + } + + for network in &self.networks { + let _ = writeln!( + out, + "\nnetwork {} ({}) {}", + network.name, + network.network_id, + if network.active { "active" } else { "inactive" } + ); + if network.peers.is_empty() { + let _ = writeln!(out, " no peers"); + } + for peer in &network.peers { + let _ = writeln!( + out, + " peer {} {} {}{}", + &peer.endpoint_id[..10.min(peer.endpoint_id.len())], + peer.hostname.as_deref().unwrap_or("?"), + peer.transport, + match peer.rtt_ms { + Some(rtt) => format!(" rtt {rtt}ms"), + None => String::new(), + } + ); + } + if network.dial_failures > 0 || network.handshake_failures > 0 { + let _ = writeln!( + out, + " {} dial failure(s), {} handshake failure(s)", + network.dial_failures, network.handshake_failures + ); + } + + if let Some(overlay) = &network.overlay { + let up = overlay.peers.iter().filter(|peer| peer.is_up()).count(); + let _ = writeln!( + out, + " overlay {} {}/{} mtu {} {}/{} tunnel(s) up", + overlay.interface, + overlay.address, + overlay.prefix_len, + overlay.mtu, + up, + overlay.peers.len() + ); + for peer in &overlay.peers { + let _ = writeln!( + out, + " {} {} {} tx {} rx {}{} {}", + &peer.public_key[..8.min(peer.public_key.len())], + peer.address, + match peer.handshake_secs_ago { + Some(secs) => format!("handshake {secs}s ago"), + None => "NOT HANDSHAKEN".to_string(), + }, + peer.tx_packets, + peer.rx_packets, + if peer.dropped > 0 { + format!(" DROPPED {}", peer.dropped) + } else { + String::new() + }, + peer.path + ); + } + if overlay.unroutable_packets > 0 { + let _ = writeln!( + out, + " {} packet(s) to addresses nobody owns", + overlay.unroutable_packets + ); + } + } + } + out + } +} diff --git a/src/ipc/unix.rs b/src/ipc/unix.rs new file mode 100644 index 0000000..86a27cc --- /dev/null +++ b/src/ipc/unix.rs @@ -0,0 +1,196 @@ +//! 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. + +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{UnixListener, UnixStream}; +use tokio::task::JoinHandle; + +use crate::BoxFuture; +use crate::error::{Error, Result}; + +use super::{MAX_MESSAGE_LEN, Request, Response, StatusReport}; + +/// 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>; +} + +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 { + path: PathBuf, + task: Option>, +} + +impl ControlSocket { + /// Binds the socket and starts serving. + /// + /// A socket file left behind by a crashed agent is replaced, but only + /// after checking that nothing is listening on it, so two live agents + /// never fight over one path. + pub async fn bind(path: impl AsRef, source: Arc) -> Result { + let path = path.as_ref().to_path_buf(); + + if let Some(parent) = path.parent() { + crate::storage::create_dir(parent)?; + } + + 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, + })?; + } + } + } + + let listener = UnixListener::bind(&path).map_err(|source| Error::Io { + path: path.clone(), + source, + })?; + restrict(&path)?; + + let task = tokio::spawn(serve(listener, source)); + Ok(Self { + path, + task: Some(task), + }) + } + + /// The path being served. + pub fn path(&self) -> &Path { + &self.path + } + + /// Stops serving and removes the socket file. + pub async fn shutdown(mut self) { + if let Some(task) = self.task.take() { + task.abort(); + let _ = task.await; + } + let _ = std::fs::remove_file(&self.path); + } +} + +impl Drop for ControlSocket { + fn drop(&mut self) { + if let Some(task) = self.task.take() { + task.abort(); + } + let _ = std::fs::remove_file(&self.path); + } +} + +#[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| { + Error::Io { + path: path.to_path_buf(), + source, + } + }) +} + +async fn serve(listener: UnixListener, source: Arc) { + loop { + let Ok((stream, _)) = listener.accept().await else { + continue; + }; + let source = Arc::clone(&source); + tokio::spawn(async move { + if let Err(err) = handle(stream, source).await { + tracing::debug!(%err, "local control request failed"); + } + }); + } +} + +async fn handle(mut stream: UnixStream, source: Arc) -> Result<()> { + let request: Request = read_message(&mut stream).await?; + let response = match request { + Request::Status => Response::Status(Box::new(source.report().await)), + }; + write_message(&mut stream, &response).await +} + +/// Asks a running agent for its status. +pub async fn request_status(path: impl AsRef) -> Result { + let path = path.as_ref(); + let mut stream = UnixStream::connect(path) + .await + .map_err(|source| Error::Io { + path: path.to_path_buf(), + source, + })?; + write_message(&mut stream, &Request::Status).await?; + match read_message::(&mut stream).await? { + Response::Status(report) => Ok(*report), + Response::Error(reason) => Err(Error::Storage(reason)), + } +} + +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(&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 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(""), + source, + } +} diff --git a/src/lib.rs b/src/lib.rs index 559fcd8..eb92b1b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -14,7 +14,10 @@ //! * [`proto`] — the control protocol: framing, messages, handshake. //! * [`net`] — the iroh connectivity adapter and its observability surface. //! * [`agent`] — the runtime: agent lifecycle, per-network runtimes, reconnect. -//! * [`dataplane`] — the minimal contract future IP plugins must satisfy. +//! * [`dataplane`] — the contract IP plugins satisfy, the packet transport, +//! and the WireGuard data plane. +//! * [`ipc`] — the local control interface a command line tool talks to. An +//! adapter over the public API; the core does not know it exists. //! //! # What this library deliberately does not do //! @@ -36,6 +39,7 @@ pub mod dataplane; pub mod discovery; pub mod error; pub mod identity; +pub mod ipc; pub mod net; pub mod proto; pub mod storage; diff --git a/tests/local_control.rs b/tests/local_control.rs new file mode 100644 index 0000000..2f958b5 --- /dev/null +++ b/tests/local_control.rs @@ -0,0 +1,228 @@ +//! The local control interface: a client asking a running agent for status. +//! +//! Uses a real Unix socket on a temporary path, the real agent and the real +//! WireGuard data plane, so what a `tsunagi status` client would see is what +//! is checked here. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +mod common; + +use std::sync::Arc; +use std::time::Duration; + +use common::{config_with, network, wait_for_peers, wait_until}; +use tempfile::TempDir; +use tsunagi::dataplane::IpPlugin; +use tsunagi::dataplane::wireguard::{MemoryTunFactory, WireguardConfig, WireguardPlugin}; +use tsunagi::discovery::SharedMemoryDiscovery; +use tsunagi::ipc::unix::{ControlSocket, request_status}; +use tsunagi::ipc::{StatusReport, control_socket_path}; +use tsunagi::{Agent, BoxFuture}; + +/// Builds the report the way the binary does, from the agent plus the plugin. +fn source(agent: Agent, plugin: Arc) -> Arc { + Arc::new(move || -> BoxFuture<'static, StatusReport> { + let agent = agent.clone(); + let plugin = Arc::clone(&plugin); + Box::pin(async move { + let status = agent.status().await.unwrap(); + let networks = status + .networks + .iter() + .map(|net| tsunagi::ipc::NetworkReport { + name: net.name.to_string(), + network_id: net.network_id.to_string(), + active: true, + peers: net + .peers + .iter() + .map(|peer| tsunagi::ipc::PeerReport { + endpoint_id: peer.endpoint_id.to_string(), + hostname: peer.hostname.clone(), + transport: format!("{:?}", peer.transport), + rtt_ms: peer.rtt.map(|rtt| rtt.as_millis() as u64), + }) + .collect(), + overlay: plugin.overview(net.network_id).map(|view| { + tsunagi::ipc::OverlayReport { + interface: view.interface.clone(), + mtu: view.mtu, + address: view.overlay_address.to_string(), + prefix: view.overlay_prefix.to_string(), + prefix_len: view.overlay_prefix_len, + peers: view + .peers + .iter() + .map(|peer| tsunagi::ipc::OverlayPeerReport { + public_key: peer.public_key.to_string(), + address: peer.overlay_address.to_string(), + handshake_secs_ago: peer + .tunnel + .as_ref() + .and_then(|t| t.health.since_handshake) + .map(|since| since.as_secs()), + ..Default::default() + }) + .collect(), + ..Default::default() + } + }), + ..Default::default() + }) + .collect(); + StatusReport { + endpoint_id: status.endpoint_id.to_string(), + hostname: status.hostname.clone(), + bound_sockets: status + .bound_sockets + .iter() + .map(ToString::to_string) + .collect(), + cache_healthy: status.cache_healthy, + networks, + } + }) + }) +} + +#[tokio::test] +async fn a_client_sees_the_agent_and_its_overlay() { + let discovery = SharedMemoryDiscovery::new(); + let (name, secret) = network("control-socket"); + + let dir_a = TempDir::new().unwrap(); + let tuns = MemoryTunFactory::new(); + let plugin = WireguardPlugin::open( + WireguardConfig::new(dir_a.path().join("wg")) + .with_interface_prefix("tca") + .with_reconcile(Duration::from_millis(20), Duration::from_millis(250)), + Arc::new(tuns), + ) + .await + .unwrap(); + let agent = Agent::spawn( + config_with(dir_a.path(), &discovery).with_plugin(plugin.clone() as Arc), + ) + .await + .unwrap(); + + let dir_b = TempDir::new().unwrap(); + let plugin_b = WireguardPlugin::open( + WireguardConfig::new(dir_b.path().join("wg")) + .with_interface_prefix("tcb") + .with_reconcile(Duration::from_millis(20), Duration::from_millis(250)), + Arc::new(MemoryTunFactory::new()), + ) + .await + .unwrap(); + let agent_b = Agent::spawn( + config_with(dir_b.path(), &discovery).with_plugin(plugin_b.clone() as Arc), + ) + .await + .unwrap(); + + let network_id = agent.join_network(&name, &secret).await.unwrap(); + agent_b.join_network(&name, &secret).await.unwrap(); + wait_for_peers(&agent, network_id, 1).await; + + // A short path: a Unix socket address is limited to about 100 bytes. + let socket_path = dir_a.path().join("agent.sock"); + let control = ControlSocket::bind(&socket_path, source(agent.clone(), plugin.clone())) + .await + .unwrap(); + + let report = wait_until("the overlay is reported as up", || { + let socket_path = socket_path.clone(); + async move { + let report = request_status(&socket_path).await.ok()?; + let overlay = report.networks.first()?.overlay.as_ref()?; + overlay + .peers + .iter() + .any(|peer| peer.is_up()) + .then_some(report) + } + }) + .await; + + assert_eq!(report.endpoint_id, agent.endpoint_id().to_string()); + assert_eq!(report.networks.len(), 1); + let net = &report.networks[0]; + assert_eq!(net.network_id, network_id.to_string()); + assert_eq!(net.peers.len(), 1); + assert_eq!(net.peers[0].endpoint_id, agent_b.endpoint_id().to_string()); + + let overlay = net.overlay.as_ref().unwrap(); + assert!(overlay.interface.starts_with("tca")); + assert_eq!(overlay.mtu, 1280); + assert_eq!(overlay.peers.len(), 1); + + // The rendering a user actually sees mentions the important parts. + let rendered = report.render(); + assert!(rendered.contains(&agent.endpoint_id().to_string())); + assert!(rendered.contains("1/1 tunnel(s) up"), "{rendered}"); + assert!(rendered.contains(&overlay.address)); + + control.shutdown().await; + assert!(!socket_path.exists(), "the socket is removed on shutdown"); + + // With nothing listening, a client gets an error rather than hanging. + assert!(request_status(&socket_path).await.is_err()); + + agent.shutdown().await; + agent_b.shutdown().await; +} + +#[tokio::test] +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 = + Arc::new(|| -> BoxFuture<'static, StatusReport> { + Box::pin(async { StatusReport::default() }) + }); + + // A file with nobody listening is a leftover from a crash. + std::fs::write(&path, b"stale").unwrap(); + let first = ControlSocket::bind(&path, Arc::clone(&empty)) + .await + .unwrap(); + assert!(request_status(&path).await.is_ok()); + + // A live socket is not stolen from the agent that owns it. + let second = ControlSocket::bind(&path, Arc::clone(&empty)).await; + assert!( + matches!(second, Err(tsunagi::Error::StateLocked { .. })), + "a second agent must not take over a live control socket" + ); + + first.shutdown().await; +} + +#[test] +fn the_socket_path_is_derived_and_short_enough() { + let deep = std::path::PathBuf::from( + "/home/someone/.local/share/with/a/very/deeply/nested/directory/that/goes/on/and/on/and/on/tsunagi/state", + ); + let path = control_socket_path(&deep); + + // A Unix socket address is limited to roughly 100 bytes, so a deep state + // directory must not produce a path that cannot be bound. + if std::env::var_os("XDG_RUNTIME_DIR").is_some() { + assert!( + path.as_os_str().len() < 100, + "derived path is {} bytes: {}", + path.as_os_str().len(), + path.display() + ); + } + + // Deterministic, and different state directories never share a socket. + assert_eq!(path, control_socket_path(&deep)); + assert_ne!( + path, + control_socket_path(&std::path::PathBuf::from("/somewhere/else")) + ); +}