//! 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)] use std::sync::Arc; 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::{StatusReport, control_socket_path}; use tsunagi::overlay::MemoryTunFactory; use tsunagi::testing::{config_with, network, wait_for_peers, wait_until}; 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 { 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 { // The interface belongs to the agent now. interface: agent .overlay() .map_or_else(String::new, |overlay| overlay.interface), mtu: agent.overlay().map_or(0, |overlay| overlay.mtu), address: view.overlay_address_v4.map(|a| a.to_string()), prefix_len: view.ipv4_range.map_or(0, |range| range.prefix_len), peers: view .peers .iter() .map(|peer| tsunagi::ipc::OverlayPeerReport { public_key: peer.public_key.to_string(), address: peer.overlay_address_v4.map(|a| a.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, dns: None, } }) }) } #[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_reconcile(Duration::from_millis(20), Duration::from_millis(250)), ) .await .unwrap(); let agent = Agent::spawn( config_with(dir_a.path(), &discovery) .with_interface(Arc::new(tuns), "tca0", 1280) .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_reconcile(Duration::from_millis(20), Duration::from_millis(250)), ) .await .unwrap(); let agent_b = Agent::spawn( config_with(dir_b.path(), &discovery) .with_interface(Arc::new(MemoryTunFactory::new()), "tcb0", 1280) .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 report carries what a reader needs, in structured form: how it is // laid out is the CLI's business and is tested there. assert_eq!(report.endpoint_id, agent.endpoint_id().to_string()); assert_eq!( overlay.peers.iter().filter(|peer| peer.is_up()).count(), 1, "the tunnel is up: {:?}", overlay.peers ); assert!(overlay.address.is_some()); 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")) ); } /// A source that can also leave, the way the binary's one does. #[derive(Debug)] struct Control(Agent); impl tsunagi::ipc::unix::ReportSource for Control { fn report(&self) -> BoxFuture<'_, StatusReport> { Box::pin(async move { StatusReport::default() }) } fn join( &self, name: String, secret: String, ) -> BoxFuture<'_, Result> { Box::pin(async move { let name = tsunagi::identity::NetworkName::new(&name).map_err(|e| e.to_string())?; let secret = tsunagi::identity::NetworkSecret::decode(&secret).map_err(|e| e.to_string())?; let already = self .0 .list_networks() .await .map_err(|err| err.to_string())? .iter() .any(|other| other.name == name); let network_id = self .0 .join_network(&name, &secret) .await .map_err(|err| err.to_string())?; Ok(tsunagi::ipc::JoinedReport { name: name.as_str().to_string(), network_id: network_id.to_string(), already_configured: already, name_shared_with: None, }) }) } fn leave(&self, network_id: String) -> BoxFuture<'_, Result> { Box::pin(async move { let wanted: tsunagi::NetworkId = network_id.parse().map_err(|_| "not an id")?; let name = self .0 .list_networks() .await .map_err(|err| err.to_string())? .into_iter() .find(|network| network.network_id == wanted) .map(|network| network.name.as_str().to_string()) .ok_or("not a network this agent is in")?; let outcome = self .0 .leave_network(wanted) .await .map_err(|err| err.to_string())?; Ok(tsunagi::ipc::LeftReport { name, announced: outcome.announced, peers_told: outcome.peers_told as u32, }) }) } } #[tokio::test] async fn a_client_can_leave_a_network_through_the_running_agent() { // The release can only be published by the agent that is running, and // only while its sessions are up, so leaving goes over this socket // rather than being done behind its back in the state store. let discovery = SharedMemoryDiscovery::new(); let (name, secret) = network("control-leave"); let dir = TempDir::new().unwrap(); let agent = Agent::spawn(config_with(dir.path(), &discovery)) .await .unwrap(); let other = TempDir::new().unwrap(); let peer = Agent::spawn(config_with(other.path(), &discovery)) .await .unwrap(); let network_id = agent.join_network(&name, &secret).await.unwrap(); peer.join_network(&name, &secret).await.unwrap(); wait_for_peers(&agent, network_id, 1).await; let socket_path = dir.path().join("control.sock"); let control = ControlSocket::bind(&socket_path, Arc::new(Control(agent.clone()))) .await .unwrap(); let report = tsunagi::ipc::unix::leave_network(&socket_path, &network_id.to_string()) .await .unwrap(); assert_eq!(report.name, name.as_str()); assert!(report.announced); assert_eq!(report.peers_told, 1); 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()) .await .unwrap_err(); assert!(err.to_string().contains("not a network"), "{err}"); control.shutdown().await; agent.shutdown().await; peer.shutdown().await; } #[tokio::test] async fn a_client_can_add_a_network_to_a_running_agent() { // One agent per identity, and it may be in several networks at once — // but the state directory belongs to that one live agent, so a second // `up` cannot add a network to it. Without this there was no way at // all: you could leave a network while running, but not join one. let discovery = SharedMemoryDiscovery::new(); let (first, first_secret) = network("control-join-one"); let (second, second_secret) = network("control-join-two"); let dir = TempDir::new().unwrap(); let agent = Agent::spawn(config_with(dir.path(), &discovery)) .await .unwrap(); agent.join_network(&first, &first_secret).await.unwrap(); let socket_path = dir.path().join("control.sock"); let control = ControlSocket::bind(&socket_path, Arc::new(Control(agent.clone()))) .await .unwrap(); let report = tsunagi::ipc::unix::join_network( &socket_path, second.as_str(), second_secret.encode().as_str(), ) .await .unwrap(); assert_eq!(report.name, second.as_str()); assert!(!report.already_configured); // Running, not merely written down: it is in the agent's own list and // answering for status straight away, on the same identity. let joined: tsunagi::NetworkId = report.network_id.parse().unwrap(); assert!(agent.is_active(joined).await); assert_eq!(agent.list_networks().await.unwrap().len(), 2); 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( &socket_path, second.as_str(), second_secret.encode().as_str(), ) .await .unwrap(); assert!(again.already_configured); 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") .await .unwrap_err(); assert!(!err.to_string().is_empty()); assert_eq!(agent.list_networks().await.unwrap().len(), 2); control.shutdown().await; agent.shutdown().await; }