ab/tsunagi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Make the protocol a crate of its own

Browse Source 
tsunagi-wg-quic. The line between a protocol and the system level is now
drawn by the compiler: nothing in it can reach into tsunagi beyond what
tsunagi makes public, and it carries its own version — which is not the
version peers compare.

Two things the compiler found the moment the boundary was real. The key
store was reaching into the core's `pub(crate)` file-permission helpers;
those are a legitimate service of the system level, because a protocol
keeping keys on disk has the same obligation the agent does, so they are
public now with that said. And the test harness was about to be copied
into a second crate, which is how two copies start to drift; it is a
`testing` feature of the core instead, which is also what anybody writing
a protocol would need.

The bridges put up while things were moving are gone: the error
conversion between the two levels, and the re-exports of the system
level's types from the protocol crate. Imports now say which level they
come from, which is the point.

One deliberate deviation, stated rather than hidden. The authenticated
transport stayed in the core. Moving it would have meant handing a
protocol the network's keys so it could prove membership itself, and a
plugin that can authenticate on the control plane is a worse trade than
a module boundary is worth. So the core proves who is at the other end
and the protocol owns what is said over it — the same separation, without
the secret crossing.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
tsunagi
2026-09-21 19:55:30 +01:00
co-authored by Claude Opus 5
parent ff7e235414
commit 142fdf995c
31 changed files with 289 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/tsunagi/tests/authentication.rs
+1 -3
View File
@@ -8,9 +8,6 @@
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod common;
use common::{TestAgent, network, settle, wait_event};
use iroh::endpoint::{Connection, PortmapperConfig, RecvStream, SendStream, presets};
use iroh::{Endpoint, EndpointAddr, RelayMode};
use tsunagi::agent::Event;
@@ -22,6 +19,7 @@ use tsunagi::proto::message::{
};
use tsunagi::proto::{read_frame, write_frame};
use tsunagi::test_support;
use tsunagi::testing::{TestAgent, network, settle, wait_event};
/// A bare iroh endpoint with no tsunagi agent behind it.
async fn raw_endpoint() -> Endpoint {
crates/tsunagi/tests/cache_and_state.rs
+1 -3
View File
@@ -2,14 +2,12 @@
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod common;
use std::io::Write;
use common::{TestAgent, config_with, local_config, network, wait_for_peers};
use tsunagi::config::StoragePaths;
use tsunagi::discovery::SharedMemoryDiscovery;
use tsunagi::storage::CacheOutcome;
use tsunagi::testing::{TestAgent, config_with, local_config, network, wait_for_peers};
use tsunagi::{Agent, Error};
/// Overwrites a file with bytes that are definitely not a SQLite database.
crates/tsunagi/tests/common/mod.rs
-185
View File
@@ -1,185 +0,0 @@
//! Shared helpers for the integration tests.
//!
//! Every test uses real iroh endpoints on loopback, its own temporary SQLite
//! files and independent agent instances. Only discovery is substituted; iroh,
//! the handshake, message passing and persistent storage are not.
//!
//! Synchronisation is always "wait for a specific event or condition, under one
//! overall deadline", never a fixed multi-second sleep.
#![allow(dead_code, clippy::unwrap_used, clippy::expect_used, clippy::panic)]
use std::future::Future;
use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tempfile::TempDir;
use tokio::sync::broadcast::error::RecvError;
use tsunagi::agent::Event;
use tsunagi::config::{AgentConfig, StoragePaths, TransportPolicy};
use tsunagi::discovery::SharedMemoryDiscovery;
use tsunagi::identity::{NetworkName, NetworkSecret};
use tsunagi::{Agent, Result};
/// Installs a tracing subscriber when `TSUNAGI_TEST_LOG` is set.
///
/// The library never installs a global subscriber itself; tests opt in.
pub fn init_tracing() {
use std::sync::Once;
static ONCE: Once = Once::new();
ONCE.call_once(|| {
if let Ok(filter) = std::env::var("TSUNAGI_TEST_LOG") {
let _ = tracing_subscriber::fmt()
.with_env_filter(tracing_subscriber::EnvFilter::new(filter))
.with_writer(std::io::stderr)
.try_init();
}
});
}
/// Overall deadline for anything a test waits on.
pub const DEADLINE: Duration = Duration::from_secs(30);
/// How often condition polling re-checks. Never used as the primary sync.
const POLL_INTERVAL: Duration = Duration::from_millis(25);
/// Builds a fully local configuration rooted at `dir`.
///
/// Loopback binding with a dynamic port, no relay, no address lookup and no
/// port mapping, so the suite needs neither the internet nor privileges.
/// Timeouts are shortened so that failure paths finish quickly.
pub fn local_config(dir: &Path) -> AgentConfig {
let limits = tsunagi::Limits {
dial_timeout: Duration::from_millis(1500),
handshake_timeout: Duration::from_secs(5),
..Default::default()
};
AgentConfig::new(StoragePaths::under(dir))
.with_transport(TransportPolicy::LocalOnly)
.with_loopback_bind()
.with_discovery_interval(Duration::from_millis(150))
.with_limits(limits)
}
/// Builds a local configuration wired to a shared in-memory discovery table.
pub fn config_with(dir: &Path, discovery: &SharedMemoryDiscovery) -> AgentConfig {
local_config(dir).with_discovery(Arc::new(discovery.clone()))
}
/// A temporary directory plus the agent running on it.
pub struct TestAgent {
pub dir: TempDir,
pub agent: Agent,
}
impl TestAgent {
/// Starts an agent on a fresh temporary directory.
pub async fn spawn(discovery: &SharedMemoryDiscovery) -> Result<Self> {
init_tracing();
let dir = TempDir::new().expect("temp dir");
let agent = Agent::spawn(config_with(dir.path(), discovery)).await?;
Ok(Self { dir, agent })
}
/// Starts an agent with a caller-supplied configuration on a fresh dir.
pub async fn spawn_with(
build: impl FnOnce(AgentConfig) -> AgentConfig,
discovery: &SharedMemoryDiscovery,
) -> Result<Self> {
init_tracing();
let dir = TempDir::new().expect("temp dir");
let agent = Agent::spawn(build(config_with(dir.path(), discovery))).await?;
Ok(Self { dir, agent })
}
/// Stops the agent and returns the directory so it can be reopened.
pub async fn stop(self) -> TempDir {
self.agent.shutdown().await;
self.dir
}
}
/// A network name and a fresh high-entropy secret.
pub fn network(name: &str) -> (NetworkName, NetworkSecret) {
(
NetworkName::new(name).expect("valid network name"),
NetworkSecret::generate(),
)
}
/// Waits for an event matching `predicate`, under the global deadline.
pub async fn wait_event<T>(
rx: &mut tokio::sync::broadcast::Receiver<Event>,
predicate: impl Fn(&Event) -> Option<T>,
) -> T {
let deadline = Instant::now() + DEADLINE;
let mut seen: Vec<String> = Vec::new();
loop {
let remaining = deadline
.checked_duration_since(Instant::now())
.unwrap_or_default();
assert!(!remaining.is_zero(), "timed out waiting for an event");
match tokio::time::timeout(remaining, rx.recv()).await {
Ok(Ok(event)) => {
if let Some(value) = predicate(&event) {
return value;
}
if seen.len() < 12 {
let label = format!("{event:?}");
seen.push(label.chars().take(70).collect());
}
}
Ok(Err(RecvError::Lagged(skipped))) => {
panic!("event subscriber lagged, missed {skipped} events; first seen: {seen:#?}");
}
Ok(Err(RecvError::Closed)) => panic!("event channel closed while waiting"),
Err(_) => panic!("timed out waiting for an event"),
}
}
}
/// Polls an async condition until it returns `Some`, under the global deadline.
pub async fn wait_until<T, F, Fut>(what: &str, mut probe: F) -> T
where
F: FnMut() -> Fut,
Fut: Future<Output = Option<T>>,
{
let deadline = Instant::now() + DEADLINE;
loop {
if let Some(value) = probe().await {
return value;
}
assert!(
Instant::now() < deadline,
"timed out waiting for condition: {what}"
);
tokio::time::sleep(POLL_INTERVAL).await;
}
}
/// Lets a few discovery rounds pass.
///
/// Only ever used before asserting that something did **not** happen; waiting
/// for success always goes through [`wait_event`] or [`wait_until`].
pub async fn settle() {
tokio::time::sleep(Duration::from_millis(900)).await;
}
/// Waits until `agent` has `count` authenticated peers in `network`.
pub async fn wait_for_peers(
agent: &Agent,
network: tsunagi::NetworkId,
count: usize,
) -> Vec<iroh::EndpointId> {
wait_until(&format!("{count} peers in {network}"), || async move {
let status = agent.network_status(network).await.ok()?;
if status.peers.len() >= count {
Some(status.connected_peers())
} else {
None
}
})
.await
}
crates/tsunagi/tests/device_identity.rs
+1 -3
View File
@@ -8,9 +8,6 @@
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod common;
use common::{config_with, network, wait_until};
use tempfile::TempDir;
use tsunagi::Agent;
use tsunagi::config::StoragePaths;
@@ -18,6 +15,7 @@ use tsunagi::discovery::SharedMemoryDiscovery;
use tsunagi::identity::{NetworkKeys, NetworkName, NetworkSecret};
use tsunagi::state::{RecordBody, StateSet};
use tsunagi::storage::StateStore;
use tsunagi::testing::{config_with, network, wait_until};
#[tokio::test]
async fn a_changed_name_reaches_peers_and_replaces_the_old_claim() {
crates/tsunagi/tests/discovery.rs
+1 -3
View File
@@ -3,16 +3,14 @@
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod common;
use std::sync::Arc;
use common::{TestAgent, local_config, network, settle, wait_for_peers};
use tsunagi::Agent;
use tsunagi::discovery::{
CandidateSource, CompositeDiscovery, NetworkDiscovery, SharedMemoryDiscovery, StaticBootstrap,
};
use tsunagi::identity::NetworkKeys;
use tsunagi::testing::{TestAgent, local_config, network, settle, wait_for_peers};
#[tokio::test]
async fn a_static_bootstrap_candidate_is_enough_to_join() {
crates/tsunagi/tests/end_to_end.rs
+1 -3
View File
@@ -3,12 +3,10 @@
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod common;
use common::{TestAgent, network, wait_event, wait_for_peers};
use tsunagi::agent::Event;
use tsunagi::discovery::SharedMemoryDiscovery;
use tsunagi::proto::ControlMessage;
use tsunagi::testing::{TestAgent, network, wait_event, wait_for_peers};
#[tokio::test]
async fn two_agents_authenticate_and_exchange_messages() {
crates/tsunagi/tests/identity.rs
+1 -3
View File
@@ -5,12 +5,10 @@
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod common;
use common::{TestAgent, config_with, network};
use tsunagi::Agent;
use tsunagi::discovery::SharedMemoryDiscovery;
use tsunagi::identity::{NetworkKeys, NetworkName, NetworkSecret};
use tsunagi::testing::{TestAgent, config_with, network};
#[test]
fn derivation_is_a_pure_function_of_name_and_secret() {
crates/tsunagi/tests/interface_provisioning.rs
-253
View File
@@ -1,253 +0,0 @@
//! The agent managing its own overlay interface.
//!
//! Everything here is real except the host: real agents, real control plane,
//! real iroh links, the real plugin lifecycle and the real reconciliation
//! rules. The host itself is a [`MockHost`], so what the agent would have
//! done to a machine's interfaces is asserted instead of done — which is how
//! this runs with no privileges and without touching the machine it is on.
//!
//! What the real provisioner adds on top of this is the netlink calls, and
//! only those.
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod common;
use std::net::{IpAddr, Ipv4Addr};
use std::sync::Arc;
use std::time::Duration;
use common::{config_with, network, wait_until};
use tempfile::TempDir;
use tsunagi::dataplane::IpPlugin;
use tsunagi::dataplane::wireguard::{
Cidr, InterfaceState, LinkKind, ManagedTunFactory, MockHost, MockProvisioner, WireguardConfig,
WireguardPlugin,
};
use tsunagi::discovery::SharedMemoryDiscovery;
use tsunagi::identity::NetworkId;
use tsunagi::state::DEFAULT_IPV4_RANGE;
use tsunagi::{Agent, NetworkStatus};
/// An agent whose overlay interface is applied to a pretend host.
struct HostedAgent {
_dir: TempDir,
agent: Agent,
_plugin: Arc<WireguardPlugin>,
host: MockHost,
}
impl HostedAgent {
async fn spawn(discovery: &SharedMemoryDiscovery, tag: &str, host: MockHost) -> Self {
let dir = TempDir::new().unwrap();
let provisioner = Arc::new(MockProvisioner::new(host.clone()));
let factory = Arc::new(ManagedTunFactory::new(provisioner));
let config = WireguardConfig::new(dir.path().join("wireguard"))
.with_reconcile(Duration::from_millis(20), Duration::from_millis(100));
let plugin = WireguardPlugin::open(config).await.unwrap();
let agent = Agent::spawn(
config_with(dir.path(), discovery)
.with_overlay_ipv4_range(Some(DEFAULT_IPV4_RANGE))
.with_interface(factory, tag, 1280)
.with_plugin(plugin.clone() as Arc<dyn IpPlugin>),
)
.await
.unwrap();
Self {
_dir: dir,
agent,
_plugin: plugin,
host,
}
}
/// The interface name the plugin settled on for a network.
/// The one interface this agent owns. Not per network.
async fn interface(&self, _network: NetworkId) -> String {
wait_until("the agent named its interface", || async {
self.agent
.overlay()
.map(|overlay| overlay.interface)
.filter(|name| !name.is_empty())
})
.await
}
/// Waits until the pretend host shows an interface in the given state.
async fn wait_for_host<T>(
&self,
what: &str,
name: &str,
probe: impl Fn(Option<InterfaceState>) -> Option<T>,
) -> T {
wait_until(what, || async { probe(self.host.get(name)) }).await
}
}
fn v4(state: &InterfaceState) -> Vec<Ipv4Addr> {
state
.addresses
.iter()
.filter_map(|cidr| match cidr.addr {
IpAddr::V4(addr) => Some(addr),
IpAddr::V6(_) => None,
})
.collect()
}
/// Whether the overlay address has been put on the interface yet.
///
/// It is allocated and signed at the system level, so it arrives on a later
/// reconciliation than the interface itself rather than with it.
fn addressed(state: &InterfaceState) -> bool {
!state.addresses.is_empty()
}
#[tokio::test]
async fn an_agent_creates_and_configures_its_own_overlay_interface() {
let discovery = SharedMemoryDiscovery::new();
let (name, secret) = network("provision-create");
let agent = HostedAgent::spawn(&discovery, "tsunp", MockHost::new()).await;
let network_id = agent.agent.join_network(&name, &secret).await.unwrap();
let interface = agent.interface(network_id).await;
let state = agent
.wait_for_host("the interface to be created", &interface, |state| state)
.await;
assert_eq!(state.kind, LinkKind::Tun);
assert!(state.up, "the agent brought the link up itself");
assert_eq!(state.mtu, 1280);
let addresses = agent
.wait_for_host("the allocated IPv4 address", &interface, |state| {
state.map(|state| v4(&state)).filter(|v4| !v4.is_empty())
})
.await;
assert_eq!(addresses.len(), 1);
assert!(
DEFAULT_IPV4_RANGE.contains(addresses[0]),
"{:?} is outside {DEFAULT_IPV4_RANGE}",
addresses[0]
);
agent.agent.shutdown().await;
}
#[tokio::test]
async fn an_interface_left_by_a_crashed_run_is_replaced_rather_than_tripped_over() {
let discovery = SharedMemoryDiscovery::new();
let (name, secret) = network("provision-crash");
// Work out the name this agent will use, then seed the host with what a
// run that died would have left there: the interface, still carrying an
// address from an allocation that no longer applies, with nothing holding
// it open.
let probe = HostedAgent::spawn(&discovery, "tsunc", MockHost::new()).await;
let network_id = probe.agent.join_network(&name, &secret).await.unwrap();
let interface = probe.interface(network_id).await;
probe.agent.shutdown().await;
let host = MockHost::new();
let stale = Cidr::new(IpAddr::V4(Ipv4Addr::new(10, 13, 37, 178)), 24).unwrap();
host.insert_stale_tun(&interface, vec![stale]);
let agent = HostedAgent::spawn(&discovery, "tsunc", host).await;
let network_id = agent.agent.join_network(&name, &secret).await.unwrap();
assert_eq!(agent.interface(network_id).await, interface);
// Waited on directly rather than on "it has some address": the address
// this agent was allocated arrives a reconciliation after the interface
// does, and in between the leftover is still the only one there.
let state = agent
.wait_for_host("the stale address to be replaced", &interface, |state| {
state.filter(|state| {
state.attached && addressed(state) && !state.addresses.contains(&stale)
})
})
.await;
assert!(
!state.addresses.contains(&stale),
"the stale address is gone: {:?}",
state.addresses
);
agent.agent.shutdown().await;
}
#[tokio::test]
async fn an_interface_belonging_to_something_else_is_left_alone() {
let discovery = SharedMemoryDiscovery::new();
let (name, secret) = network("provision-foreign");
let probe = HostedAgent::spawn(&discovery, "tsunf", MockHost::new()).await;
let network_id = probe.agent.join_network(&name, &secret).await.unwrap();
let interface = probe.interface(network_id).await;
probe.agent.shutdown().await;
// Somebody else's bridge happens to hold the name.
let host = MockHost::new();
let theirs = InterfaceState {
kind: LinkKind::Foreign("bridge".into()),
attached: true,
up: true,
mtu: 1500,
addresses: vec![Cidr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 9, 1)), 24).unwrap()],
};
host.insert(&interface, theirs.clone());
let agent = HostedAgent::spawn(&discovery, "tsunf", host).await;
agent.agent.join_network(&name, &secret).await.unwrap();
// Give the plugin several reconciliation rounds to do the wrong thing.
tokio::time::sleep(Duration::from_millis(300)).await;
assert_eq!(
agent.host.get(&interface),
Some(theirs),
"the foreign interface must be untouched"
);
// The control plane is unaffected by the data plane refusing.
assert!(matches!(
agent.agent.network_status(network_id).await,
Ok(NetworkStatus { .. })
));
agent.agent.shutdown().await;
}
#[tokio::test]
async fn leaving_a_network_takes_its_address_off_the_interface_but_not_the_interface() {
// The interface belongs to the agent, so it outlives any one network:
// another may still be using it. What a network takes with it is its own
// address.
let discovery = SharedMemoryDiscovery::new();
let (name, secret) = network("provision-cleanup");
let agent = HostedAgent::spawn(&discovery, "tsunx", MockHost::new()).await;
let network_id = agent.agent.join_network(&name, &secret).await.unwrap();
let interface = agent.interface(network_id).await;
agent
.wait_for_host("the address to be assigned", &interface, |state| {
state.filter(addressed)
})
.await;
agent.agent.deactivate_network(network_id).await.unwrap();
let state = agent
.wait_for_host("the address to be withdrawn", &interface, |state| {
state.filter(|state| !addressed(state))
})
.await;
assert_eq!(state.kind, LinkKind::Tun, "the interface is still there");
// And it goes when the agent does.
agent.agent.shutdown().await;
assert!(
agent.host.names().is_empty(),
"nothing is left behind: {:?}",
agent.host.names()
);
}
crates/tsunagi/tests/local_control.rs
-236
View File
@@ -1,236 +0,0 @@
//! 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<WireguardPlugin>) -> Arc<dyn tsunagi::ipc::unix::ReportSource> {
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<dyn IpPlugin>),
)
.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<dyn IpPlugin>),
)
.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<dyn tsunagi::ipc::unix::ReportSource> =
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"))
);
}
crates/tsunagi/tests/multi_peer.rs
+1 -3
View File
@@ -3,16 +3,14 @@
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod common;
use std::collections::HashSet;
use std::sync::Arc;
use common::{TestAgent, network, wait_event, wait_for_peers, wait_until};
use tsunagi::agent::Event;
use tsunagi::dataplane::TestCapabilityPlugin;
use tsunagi::discovery::SharedMemoryDiscovery;
use tsunagi::proto::ControlMessage;
use tsunagi::testing::{TestAgent, network, wait_event, wait_for_peers, wait_until};
#[tokio::test]
async fn four_agents_form_a_mesh_and_exchange_distinguishable_messages() {
crates/tsunagi/tests/network_isolation.rs
+1 -3
View File
@@ -2,9 +2,6 @@
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod common;
use common::{TestAgent, network, settle, wait_event, wait_for_peers};
use iroh::endpoint::{PortmapperConfig, presets};
use iroh::{Endpoint, RelayMode};
use tsunagi::agent::Event;
@@ -16,6 +13,7 @@ use tsunagi::proto::message::{
};
use tsunagi::proto::{read_frame, write_frame};
use tsunagi::test_support;
use tsunagi::testing::{TestAgent, network, settle, wait_event, wait_for_peers};
const LIMIT: usize = 64 * 1024;
crates/tsunagi/tests/resilience.rs
+3 -3
View File
@@ -3,18 +3,18 @@
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod common;
use std::net::SocketAddr;
use std::time::Duration;
use common::{TestAgent, config_with, local_config, network, settle, wait_event, wait_for_peers};
use iroh::{EndpointAddr, SecretKey};
use tsunagi::agent::Event;
use tsunagi::config::ReconnectPolicy;
use tsunagi::discovery::SharedMemoryDiscovery;
use tsunagi::identity::NetworkKeys;
use tsunagi::proto::ControlMessage;
use tsunagi::testing::{
TestAgent, config_with, local_config, network, settle, wait_event, wait_for_peers,
};
use tsunagi::{Agent, Error};
/// An endpoint id nobody is listening for, at an address nothing answers on.
crates/tsunagi/tests/restart.rs
+1 -3
View File
@@ -2,13 +2,11 @@
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod common;
use common::{TestAgent, config_with, network, settle, wait_event, wait_for_peers};
use tsunagi::agent::Event;
use tsunagi::discovery::SharedMemoryDiscovery;
use tsunagi::identity::{NetworkName, NetworkSecret};
use tsunagi::proto::ControlMessage;
use tsunagi::testing::{TestAgent, config_with, network, settle, wait_event, wait_for_peers};
use tsunagi::{Agent, Error};
#[tokio::test]
crates/tsunagi/tests/wireguard.rs
-1170
View File
File diff suppressed because it is too large Load Diff