Split the system level and the command line into a workspace
First step of separating the layers. The library and the binary are now crates/tsunagi and crates/tsunagi-cli, which means the plugin crate to come can be told apart from the core by the compiler rather than by discipline. Falls out of it immediately: the CLI's dependencies stop being features of the library. clap, anstream and tracing-subscriber were optional dependencies behind a `cli` feature that every library user had to remember to turn off; now they belong to the crate that uses them, and the library defaults to no features at all. The one test that drives the binary moved beside it — a library cannot depend on a binary built from a crate that depends on the library — and was rewritten against the public API instead of the test harness. AGENTS.md said to prefer one crate. It now says the system level and its plugins are separate crates, for the reason above, and that everything else stays one crate. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
tsunagi
Claude Opus 5
@@ -0,0 +1,425 @@
|
||||
//! Scenarios 3 and 9: an attacker without the secret, and the protocol's
|
||||
//! boundaries.
|
||||
//!
|
||||
//! These tests speak the wire protocol directly against a real agent, because
|
||||
//! that is the only way to present a *correct* public network id with a wrong
|
||||
//! proof, replay a captured proof on a second connection, or send a message
|
||||
//! before authenticating.
|
||||
|
||||
#![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;
|
||||
use tsunagi::discovery::SharedMemoryDiscovery;
|
||||
use tsunagi::identity::{NetworkId, NetworkKeys, NetworkName, NetworkSecret};
|
||||
use tsunagi::proto::handshake::{ROLE_INITIATOR, ROLE_RESPONDER};
|
||||
use tsunagi::proto::message::{
|
||||
ALPN, AuthProof, ControlMessage, Envelope, Hello, HelloAck, PROTOCOL_VERSION, decode, encode,
|
||||
};
|
||||
use tsunagi::proto::{read_frame, write_frame};
|
||||
use tsunagi::test_support;
|
||||
|
||||
/// A bare iroh endpoint with no tsunagi agent behind it.
|
||||
async fn raw_endpoint() -> Endpoint {
|
||||
Endpoint::builder(presets::Minimal)
|
||||
.alpns(vec![ALPN.to_vec()])
|
||||
.relay_mode(RelayMode::Disabled)
|
||||
.clear_address_lookup()
|
||||
.portmapper_config(PortmapperConfig::Disabled)
|
||||
.clear_ip_transports()
|
||||
.bind_addr("127.0.0.1:0")
|
||||
.unwrap()
|
||||
.bind()
|
||||
.await
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
async fn open_control_stream(
|
||||
endpoint: &Endpoint,
|
||||
target: EndpointAddr,
|
||||
) -> (Connection, SendStream, RecvStream) {
|
||||
let conn = endpoint.connect(target, ALPN).await.unwrap();
|
||||
let (send, recv) = conn.open_bi().await.unwrap();
|
||||
(conn, send, recv)
|
||||
}
|
||||
|
||||
const LIMIT: usize = 64 * 1024;
|
||||
|
||||
/// Sends `Hello` and reads the responder's `HelloAck`.
|
||||
async fn exchange_hellos(
|
||||
send: &mut SendStream,
|
||||
recv: &mut RecvStream,
|
||||
version: u16,
|
||||
network_id: NetworkId,
|
||||
nonce: [u8; 16],
|
||||
) -> HelloAck {
|
||||
let hello = Hello {
|
||||
version,
|
||||
network_id: *network_id.as_bytes(),
|
||||
nonce,
|
||||
};
|
||||
write_frame(send, &encode(&hello).unwrap(), LIMIT)
|
||||
.await
|
||||
.unwrap();
|
||||
decode(&read_frame(recv, LIMIT).await.unwrap()).unwrap()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_attacker_who_knows_the_public_network_id_is_still_rejected() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("closed-network");
|
||||
|
||||
let victim = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let mut events = victim.agent.subscribe();
|
||||
let network_id = victim.agent.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
// The attacker knows the address and the correct *public* network id. It
|
||||
// does not know the secret, so it cannot compute a valid proof.
|
||||
let attacker = raw_endpoint().await;
|
||||
let (conn, mut send, mut recv) =
|
||||
open_control_stream(&attacker, victim.agent.local_addr()).await;
|
||||
|
||||
let _ack = exchange_hellos(
|
||||
&mut send,
|
||||
&mut recv,
|
||||
PROTOCOL_VERSION,
|
||||
network_id,
|
||||
[9u8; 16],
|
||||
)
|
||||
.await;
|
||||
let bogus = AuthProof { proof: [0xAB; 32] };
|
||||
write_frame(&mut send, &encode(&bogus).unwrap(), LIMIT)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// The responder must not reveal a proof of its own and must reject us.
|
||||
let reply = read_frame(&mut recv, LIMIT).await;
|
||||
assert!(
|
||||
reply.is_err(),
|
||||
"the agent must not answer an invalid proof, got {reply:?}"
|
||||
);
|
||||
|
||||
let reason = wait_event(&mut events, |event| match event {
|
||||
Event::HandshakeRejected { reason, .. } => Some(reason.clone()),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
assert!(
|
||||
reason.contains("authentication failed"),
|
||||
"unexpected reason: {reason}"
|
||||
);
|
||||
|
||||
// The victim is unharmed: no session, and the network is still running.
|
||||
let status = victim.agent.network_status(network_id).await.unwrap();
|
||||
assert!(status.peers.is_empty());
|
||||
assert_eq!(status.metrics.sessions_established, 0);
|
||||
|
||||
conn.close(0u32.into(), b"done");
|
||||
attacker.close().await;
|
||||
victim.agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_wrong_secret_under_the_same_name_lands_in_a_different_space() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let name = NetworkName::new("same-name").unwrap();
|
||||
|
||||
let good = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let bad = TestAgent::spawn(&discovery).await.unwrap();
|
||||
|
||||
let good_id = good
|
||||
.agent
|
||||
.join_network(&name, &NetworkSecret::generate())
|
||||
.await
|
||||
.unwrap();
|
||||
let bad_id = bad
|
||||
.agent
|
||||
.join_network(&name, &NetworkSecret::generate())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_ne!(good_id, bad_id);
|
||||
|
||||
// Even with discovery wired together, the two never form a session.
|
||||
//
|
||||
// This test only shows that discovery separated them; the authoritative
|
||||
// check that the secret itself gates membership is
|
||||
// `an_attacker_who_knows_the_public_network_id_is_still_rejected`.
|
||||
settle().await;
|
||||
let status = good.agent.network_status(good_id).await.unwrap();
|
||||
assert!(status.peers.is_empty());
|
||||
|
||||
good.agent.shutdown().await;
|
||||
bad.agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_unsupported_protocol_version_is_rejected() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("versioned");
|
||||
|
||||
let victim = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let mut events = victim.agent.subscribe();
|
||||
let network_id = victim.agent.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
let attacker = raw_endpoint().await;
|
||||
let (conn, mut send, mut recv) =
|
||||
open_control_stream(&attacker, victim.agent.local_addr()).await;
|
||||
|
||||
let hello = Hello {
|
||||
version: PROTOCOL_VERSION + 7,
|
||||
network_id: *network_id.as_bytes(),
|
||||
nonce: [1u8; 16],
|
||||
};
|
||||
write_frame(&mut send, &encode(&hello).unwrap(), LIMIT)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(read_frame(&mut recv, LIMIT).await.is_err());
|
||||
|
||||
let reason = wait_event(&mut events, |event| match event {
|
||||
Event::HandshakeRejected { reason, .. } => Some(reason.clone()),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
assert!(reason.contains("version"), "unexpected reason: {reason}");
|
||||
|
||||
// The agent is still alive and usable afterwards.
|
||||
assert!(victim.agent.network_status(network_id).await.is_ok());
|
||||
|
||||
conn.close(0u32.into(), b"done");
|
||||
attacker.close().await;
|
||||
victim.agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_control_message_before_authentication_is_rejected() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("no-early-messages");
|
||||
|
||||
let victim = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let mut events = victim.agent.subscribe();
|
||||
let network_id = victim.agent.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
let attacker = raw_endpoint().await;
|
||||
let (conn, mut send, mut recv) =
|
||||
open_control_stream(&attacker, victim.agent.local_addr()).await;
|
||||
|
||||
// A perfectly well formed control message, sent where a Hello belongs.
|
||||
let envelope = Envelope {
|
||||
network_id: *network_id.as_bytes(),
|
||||
message: ControlMessage::Ping {
|
||||
seq: 1,
|
||||
payload: b"too early".to_vec(),
|
||||
},
|
||||
};
|
||||
write_frame(&mut send, &encode(&envelope).unwrap(), LIMIT)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(read_frame(&mut recv, LIMIT).await.is_err());
|
||||
|
||||
wait_event(&mut events, |event| match event {
|
||||
Event::HandshakeRejected { .. } => Some(()),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
|
||||
let status = victim.agent.network_status(network_id).await.unwrap();
|
||||
assert!(status.peers.is_empty());
|
||||
|
||||
conn.close(0u32.into(), b"done");
|
||||
attacker.close().await;
|
||||
victim.agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_oversized_frame_is_rejected_before_it_is_allocated() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("bounded-frames");
|
||||
|
||||
let victim = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let mut events = victim.agent.subscribe();
|
||||
let network_id = victim.agent.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
let attacker = raw_endpoint().await;
|
||||
let (conn, mut send, mut recv) =
|
||||
open_control_stream(&attacker, victim.agent.local_addr()).await;
|
||||
|
||||
// Announce four gigabytes and then send nothing. The agent must reject the
|
||||
// header instead of allocating the buffer.
|
||||
send.write_all(&u32::MAX.to_be_bytes()).await.unwrap();
|
||||
assert!(read_frame(&mut recv, LIMIT).await.is_err());
|
||||
|
||||
wait_event(&mut events, |event| match event {
|
||||
Event::HandshakeRejected { reason, .. } if reason.contains("exceeds") => Some(()),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
|
||||
// Still serving other work.
|
||||
assert!(victim.agent.network_status(network_id).await.is_ok());
|
||||
|
||||
conn.close(0u32.into(), b"done");
|
||||
attacker.close().await;
|
||||
victim.agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_proof_cannot_be_replayed_on_another_connection() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("channel-bound");
|
||||
|
||||
let victim = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let mut events = victim.agent.subscribe();
|
||||
let network_id = victim.agent.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
// A legitimate member: it knows the secret and can compute real proofs.
|
||||
let keys = NetworkKeys::derive(&name, &secret);
|
||||
let auth_key = test_support::auth_key(&keys);
|
||||
let member = raw_endpoint().await;
|
||||
let local = *member.id().as_bytes();
|
||||
let remote = *victim.agent.endpoint_id().as_bytes();
|
||||
let network_bytes = *network_id.as_bytes();
|
||||
|
||||
// Connection one: a genuine, successful handshake.
|
||||
let (conn1, mut send1, mut recv1) =
|
||||
open_control_stream(&member, victim.agent.local_addr()).await;
|
||||
let nonce_i = [11u8; 16];
|
||||
let ack1 = exchange_hellos(
|
||||
&mut send1,
|
||||
&mut recv1,
|
||||
PROTOCOL_VERSION,
|
||||
network_id,
|
||||
nonce_i,
|
||||
)
|
||||
.await;
|
||||
let cb1 = test_support::channel_binding(&conn1, &network_bytes).unwrap();
|
||||
let genuine = test_support::compute_proof(
|
||||
&auth_key,
|
||||
ROLE_INITIATOR,
|
||||
PROTOCOL_VERSION,
|
||||
&network_bytes,
|
||||
&local,
|
||||
&remote,
|
||||
&cb1,
|
||||
&nonce_i,
|
||||
&ack1.nonce,
|
||||
);
|
||||
write_frame(
|
||||
&mut send1,
|
||||
&encode(&AuthProof { proof: genuine }).unwrap(),
|
||||
LIMIT,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let their_proof: AuthProof = decode(&read_frame(&mut recv1, LIMIT).await.unwrap()).unwrap();
|
||||
|
||||
// The responder proved membership too, and it used the responder role.
|
||||
let expected = test_support::compute_proof(
|
||||
&auth_key,
|
||||
ROLE_RESPONDER,
|
||||
PROTOCOL_VERSION,
|
||||
&network_bytes,
|
||||
&local,
|
||||
&remote,
|
||||
&cb1,
|
||||
&nonce_i,
|
||||
&ack1.nonce,
|
||||
);
|
||||
assert_eq!(their_proof.proof, expected, "responder proof must verify");
|
||||
assert_ne!(
|
||||
their_proof.proof, genuine,
|
||||
"roles must not produce the same proof, or it could be reflected"
|
||||
);
|
||||
|
||||
// Connection two: replay the captured proof verbatim. The TLS exporter
|
||||
// differs per connection, so the proof no longer matches.
|
||||
let (conn2, mut send2, mut recv2) =
|
||||
open_control_stream(&member, victim.agent.local_addr()).await;
|
||||
let ack2 = exchange_hellos(
|
||||
&mut send2,
|
||||
&mut recv2,
|
||||
PROTOCOL_VERSION,
|
||||
network_id,
|
||||
nonce_i,
|
||||
)
|
||||
.await;
|
||||
let cb2 = test_support::channel_binding(&conn2, &network_bytes).unwrap();
|
||||
assert_ne!(cb1, cb2, "channel binding must differ between connections");
|
||||
let _ = ack2;
|
||||
|
||||
write_frame(
|
||||
&mut send2,
|
||||
&encode(&AuthProof { proof: genuine }).unwrap(),
|
||||
LIMIT,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(
|
||||
read_frame(&mut recv2, LIMIT).await.is_err(),
|
||||
"a replayed proof must be rejected"
|
||||
);
|
||||
|
||||
let reason = wait_event(&mut events, |event| match event {
|
||||
Event::HandshakeRejected { reason, .. } if reason.contains("authentication failed") => {
|
||||
Some(reason.clone())
|
||||
}
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
assert!(reason.contains("authentication failed"));
|
||||
|
||||
conn1.close(0u32.into(), b"done");
|
||||
conn2.close(0u32.into(), b"done");
|
||||
member.close().await;
|
||||
victim.agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_hello_for_an_inactive_network_is_rejected() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("active-only");
|
||||
|
||||
let victim = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let mut events = victim.agent.subscribe();
|
||||
victim.agent.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
// A different, perfectly valid network space the victim is not in.
|
||||
let other = NetworkKeys::derive(
|
||||
&NetworkName::new("somewhere-else").unwrap(),
|
||||
&NetworkSecret::generate(),
|
||||
);
|
||||
|
||||
let attacker = raw_endpoint().await;
|
||||
let (conn, mut send, mut recv) =
|
||||
open_control_stream(&attacker, victim.agent.local_addr()).await;
|
||||
let hello = Hello {
|
||||
version: PROTOCOL_VERSION,
|
||||
network_id: *other.network_id().as_bytes(),
|
||||
nonce: [3u8; 16],
|
||||
};
|
||||
write_frame(&mut send, &encode(&hello).unwrap(), LIMIT)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(read_frame(&mut recv, LIMIT).await.is_err());
|
||||
|
||||
let reason = wait_event(&mut events, |event| match event {
|
||||
Event::HandshakeRejected {
|
||||
network, reason, ..
|
||||
} => {
|
||||
// Before a successful handshake the claimed network is unverified,
|
||||
// so it must not be reported as fact.
|
||||
assert!(network.is_none());
|
||||
Some(reason.clone())
|
||||
}
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
assert!(reason.contains("unknown"), "unexpected reason: {reason}");
|
||||
|
||||
conn.close(0u32.into(), b"done");
|
||||
attacker.close().await;
|
||||
victim.agent.shutdown().await;
|
||||
}
|
||||
@@ -0,0 +1,286 @@
|
||||
//! Scenario 7: the disposable cache and the mandatory state store.
|
||||
|
||||
#![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::{Agent, Error};
|
||||
|
||||
/// Overwrites a file with bytes that are definitely not a SQLite database.
|
||||
fn corrupt(path: &std::path::Path) {
|
||||
let mut file = std::fs::File::create(path).unwrap();
|
||||
file.write_all(&[0x7f; 8192]).unwrap();
|
||||
file.sync_all().unwrap();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_missing_cache_is_recreated_and_does_not_block_connecting() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("missing-cache");
|
||||
|
||||
let peer = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let subject = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let network_id = peer.agent.join_network(&name, &secret).await.unwrap();
|
||||
subject.agent.join_network(&name, &secret).await.unwrap();
|
||||
wait_for_peers(&peer.agent, network_id, 1).await;
|
||||
|
||||
let dir = subject.stop().await;
|
||||
let paths = StoragePaths::under(dir.path());
|
||||
std::fs::remove_dir_all(&paths.cache_dir).unwrap();
|
||||
assert!(!paths.cache_db().exists());
|
||||
|
||||
let restarted = Agent::spawn(config_with(dir.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
restarted.status().await.unwrap().cache_outcome,
|
||||
CacheOutcome::Created
|
||||
);
|
||||
wait_for_peers(&restarted, network_id, 1).await;
|
||||
|
||||
restarted.shutdown().await;
|
||||
peer.agent.shutdown().await;
|
||||
drop(restarted);
|
||||
drop(dir);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_corrupt_cache_is_discarded_and_does_not_block_connecting() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("corrupt-cache");
|
||||
|
||||
let peer = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let subject = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let network_id = peer.agent.join_network(&name, &secret).await.unwrap();
|
||||
subject.agent.join_network(&name, &secret).await.unwrap();
|
||||
wait_for_peers(&peer.agent, network_id, 1).await;
|
||||
|
||||
let device_id = subject.agent.endpoint_id();
|
||||
let dir = subject.stop().await;
|
||||
let paths = StoragePaths::under(dir.path());
|
||||
corrupt(&paths.cache_db());
|
||||
|
||||
let restarted = Agent::spawn(config_with(dir.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
let status = restarted.status().await.unwrap();
|
||||
assert!(
|
||||
matches!(status.cache_outcome, CacheOutcome::Reset(_)),
|
||||
"expected the cache to be discarded, got {:?}",
|
||||
status.cache_outcome
|
||||
);
|
||||
assert!(status.cache_healthy);
|
||||
assert_eq!(
|
||||
restarted.endpoint_id(),
|
||||
device_id,
|
||||
"a bad cache must not touch the identity"
|
||||
);
|
||||
wait_for_peers(&restarted, network_id, 1).await;
|
||||
|
||||
restarted.shutdown().await;
|
||||
peer.agent.shutdown().await;
|
||||
drop(restarted);
|
||||
drop(dir);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_stale_cache_does_not_prevent_connecting_through_discovery() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("stale-cache");
|
||||
|
||||
let peer = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let subject = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let network_id = peer.agent.join_network(&name, &secret).await.unwrap();
|
||||
subject.agent.join_network(&name, &secret).await.unwrap();
|
||||
wait_for_peers(&peer.agent, network_id, 1).await;
|
||||
|
||||
// Stop both. When they come back they bind new ports, so every cached
|
||||
// address hint is stale, and only fresh discovery can bridge the gap.
|
||||
let peer_dir = peer.stop().await;
|
||||
let subject_dir = subject.stop().await;
|
||||
assert!(
|
||||
StoragePaths::under(subject_dir.path()).cache_db().exists(),
|
||||
"hints were written, so the cache is genuinely stale now"
|
||||
);
|
||||
|
||||
let peer_again = Agent::spawn(config_with(peer_dir.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
let subject_again = Agent::spawn(config_with(subject_dir.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
wait_for_peers(&subject_again, network_id, 1).await;
|
||||
wait_for_peers(&peer_again, network_id, 1).await;
|
||||
|
||||
peer_again.shutdown().await;
|
||||
subject_again.shutdown().await;
|
||||
drop(peer_again);
|
||||
drop(subject_again);
|
||||
drop(peer_dir);
|
||||
drop(subject_dir);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_corrupt_state_store_is_an_error_and_never_a_fresh_identity() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("corrupt-state");
|
||||
|
||||
let subject = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let device_id = subject.agent.endpoint_id();
|
||||
subject.agent.join_network(&name, &secret).await.unwrap();
|
||||
let dir = subject.stop().await;
|
||||
|
||||
let paths = StoragePaths::under(dir.path());
|
||||
corrupt(&paths.state_db());
|
||||
|
||||
let result = Agent::spawn(config_with(dir.path(), &discovery)).await;
|
||||
match result {
|
||||
Err(Error::StateCorrupted { path, reason }) => {
|
||||
assert_eq!(path, paths.state_db());
|
||||
assert!(!reason.is_empty());
|
||||
}
|
||||
Err(other) => panic!("expected StateCorrupted, got {other:?}"),
|
||||
Ok(agent) => {
|
||||
let new_id = agent.endpoint_id();
|
||||
agent.shutdown().await;
|
||||
panic!(
|
||||
"a corrupt state store must not yield a working agent (id {new_id}) — the previous identity was {device_id}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
drop(dir);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_state_store_from_a_newer_build_is_refused() {
|
||||
let dir = tempfile::TempDir::new().unwrap();
|
||||
let paths = StoragePaths::under(dir.path());
|
||||
std::fs::create_dir_all(&paths.state_dir).unwrap();
|
||||
|
||||
{
|
||||
let conn = rusqlite::Connection::open(paths.state_db()).unwrap();
|
||||
conn.pragma_update(None, "user_version", 999i64).unwrap();
|
||||
}
|
||||
|
||||
let result = Agent::spawn(local_config(dir.path())).await;
|
||||
assert!(
|
||||
matches!(result, Err(Error::UnsupportedSchema { found: 999, .. })),
|
||||
"expected UnsupportedSchema"
|
||||
);
|
||||
}
|
||||
|
||||
/// A store written by the previous schema must come up, not break.
|
||||
///
|
||||
/// The record body gained a hostname, which changed the signing domain, so
|
||||
/// records written before it can never verify again. Leaving them in place
|
||||
/// would mean every read rejecting rows that look exactly like corruption.
|
||||
#[tokio::test]
|
||||
async fn a_state_store_from_the_previous_schema_is_migrated_and_stays_usable() {
|
||||
let dir = tempfile::TempDir::new().unwrap();
|
||||
let paths = StoragePaths::under(dir.path());
|
||||
std::fs::create_dir_all(&paths.state_dir).unwrap();
|
||||
|
||||
// Build a schema-2 store by hand, with a record in it.
|
||||
{
|
||||
let conn = rusqlite::Connection::open(paths.state_db()).unwrap();
|
||||
conn.execute_batch(
|
||||
"BEGIN;
|
||||
CREATE TABLE device_identity (
|
||||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||||
secret_key BLOB NOT NULL,
|
||||
created_at INTEGER NOT NULL
|
||||
);
|
||||
CREATE TABLE networks (
|
||||
network_id BLOB PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
secret BLOB NOT NULL,
|
||||
auto_start INTEGER NOT NULL DEFAULT 1,
|
||||
created_at INTEGER NOT NULL
|
||||
);
|
||||
CREATE TABLE settings (key TEXT PRIMARY KEY, value TEXT NOT NULL);
|
||||
CREATE TABLE signed_records (
|
||||
network_id BLOB NOT NULL,
|
||||
author BLOB NOT NULL,
|
||||
version INTEGER NOT NULL,
|
||||
body BLOB NOT NULL,
|
||||
signature BLOB NOT NULL,
|
||||
PRIMARY KEY (network_id, author)
|
||||
);
|
||||
CREATE TABLE own_record_version (
|
||||
network_id BLOB PRIMARY KEY,
|
||||
version INTEGER NOT NULL
|
||||
);
|
||||
INSERT INTO signed_records VALUES (x'00', x'11', 7, x'2222', x'3333');
|
||||
INSERT INTO own_record_version VALUES (x'00', 7);
|
||||
PRAGMA user_version = 2;
|
||||
COMMIT;",
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
// An agent comes up on it, which is the whole point.
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let agent = Agent::spawn(config_with(dir.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
let (name, secret) = network("migrated");
|
||||
let network_id = agent.join_network(&name, &secret).await.unwrap();
|
||||
assert!(agent.network_status(network_id).await.is_ok());
|
||||
agent.shutdown().await;
|
||||
|
||||
let conn = rusqlite::Connection::open(paths.state_db()).unwrap();
|
||||
let version: i64 = conn
|
||||
.query_row("PRAGMA user_version", [], |row| row.get(0))
|
||||
.unwrap();
|
||||
assert_eq!(version, tsunagi::storage::SCHEMA_VERSION);
|
||||
|
||||
// The unverifiable record is gone rather than left to be rejected for
|
||||
// ever, and the counter is keyed by author now.
|
||||
let stale: i64 = conn
|
||||
.query_row(
|
||||
"SELECT count(*) FROM signed_records WHERE author = x'11'",
|
||||
[],
|
||||
|row| row.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(stale, 0, "records from the old signing domain are dropped");
|
||||
conn.query_row(
|
||||
"SELECT count(*) FROM own_record_version WHERE author IS NOT NULL",
|
||||
[],
|
||||
|row| row.get::<_, i64>(0),
|
||||
)
|
||||
.expect("the counter is keyed by author");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn secrets_never_appear_in_status_or_debug_output() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("no-leaks");
|
||||
|
||||
let agent = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let network_id = agent.agent.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
let status = agent.agent.status().await.unwrap();
|
||||
let rendered = format!("{status:?}");
|
||||
let encoded = secret.encode();
|
||||
assert!(!rendered.contains(encoded.as_str()));
|
||||
assert!(rendered.contains(&network_id.to_string()) || rendered.contains("NetworkId"));
|
||||
|
||||
let network_status = agent.agent.network_status(network_id).await.unwrap();
|
||||
assert!(!format!("{network_status:?}").contains(encoded.as_str()));
|
||||
|
||||
let keys = tsunagi::identity::NetworkKeys::derive(&name, &secret);
|
||||
let keys_debug = format!("{keys:?}");
|
||||
assert!(keys_debug.contains("<redacted>"));
|
||||
assert!(!keys_debug.contains(encoded.as_str()));
|
||||
|
||||
agent.agent.shutdown().await;
|
||||
}
|
||||
@@ -0,0 +1,185 @@
|
||||
//! 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
|
||||
}
|
||||
@@ -0,0 +1,244 @@
|
||||
//! The device's own identity: the name it answers to and the key it signs
|
||||
//! with, and what happens when either is changed.
|
||||
//!
|
||||
//! Both are things a user may reasonably change on a machine they own, and
|
||||
//! neither may leave the state store in a shape the next start cannot use.
|
||||
//! That is what these check: not that changing them is prevented, but that
|
||||
//! the store survives it and says something true afterwards.
|
||||
|
||||
#![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;
|
||||
use tsunagi::discovery::SharedMemoryDiscovery;
|
||||
use tsunagi::identity::{NetworkKeys, NetworkName, NetworkSecret};
|
||||
use tsunagi::state::{RecordBody, StateSet};
|
||||
use tsunagi::storage::StateStore;
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_changed_name_reaches_peers_and_replaces_the_old_claim() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("renaming");
|
||||
|
||||
let a = TempDir::new().unwrap();
|
||||
let b = TempDir::new().unwrap();
|
||||
let agent_a = Agent::spawn(config_with(a.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
let agent_b = Agent::spawn(config_with(b.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
let network_id = agent_a.join_network(&name, &secret).await.unwrap();
|
||||
agent_b.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
let accepted = agent_a.set_hostname("Renamed Host").await.unwrap();
|
||||
assert_eq!(accepted, "renamedhost", "reduced to a canonical form");
|
||||
assert_eq!(agent_a.hostname(), "renamedhost");
|
||||
|
||||
// The peer is told, rather than finding out on its next restart.
|
||||
wait_until("the peer learns the new name", || async {
|
||||
let status = agent_b.network_status(network_id).await.ok()?;
|
||||
status
|
||||
.peers
|
||||
.iter()
|
||||
.any(|peer| peer.hostname.as_deref() == Some("renamedhost"))
|
||||
.then_some(())
|
||||
})
|
||||
.await;
|
||||
|
||||
// And the signed claim says it, so the name outlives the session.
|
||||
wait_until("the claim carries the new name", || async {
|
||||
let status = agent_b.network_status(network_id).await.ok()?;
|
||||
status
|
||||
.members
|
||||
.iter()
|
||||
.any(|member| member.hostname.as_deref() == Some("renamedhost"))
|
||||
.then_some(())
|
||||
})
|
||||
.await;
|
||||
|
||||
agent_a.shutdown().await;
|
||||
agent_b.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_name_that_reduces_to_nothing_is_refused_rather_than_stored() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let dir = TempDir::new().unwrap();
|
||||
let agent = Agent::spawn(config_with(dir.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let before = agent.hostname();
|
||||
assert!(agent.set_hostname("---").await.is_err());
|
||||
assert!(agent.set_hostname("").await.is_err());
|
||||
assert_eq!(agent.hostname(), before, "the old name still stands");
|
||||
|
||||
agent.shutdown().await;
|
||||
}
|
||||
|
||||
/// Replacing the signing key must leave a store the next run can use.
|
||||
///
|
||||
/// The user is entitled to do this on a machine they own, and they lose the
|
||||
/// address and name the old key held — there is no way to sign on a dead
|
||||
/// key's behalf, and nothing here may overrule an author. What must not
|
||||
/// happen is that the store is left in a shape that breaks.
|
||||
#[tokio::test]
|
||||
async fn rotating_the_signing_key_releases_what_it_held_and_leaves_a_usable_store() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let paths = StoragePaths::under(dir.path());
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("rotation");
|
||||
let network_id = NetworkKeys::derive(&name, &secret).network_id();
|
||||
|
||||
// Run once so there is an identity, a network, and a claim to give up.
|
||||
let agent = Agent::spawn(config_with(dir.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
agent.join_network(&name, &secret).await.unwrap();
|
||||
let before = agent.endpoint_id();
|
||||
wait_until("the agent claims an address", || async {
|
||||
let status = agent.network_status(network_id).await.ok()?;
|
||||
status
|
||||
.members
|
||||
.iter()
|
||||
.any(|member| member.endpoint_id == before && member.overlay_address_v4.is_some())
|
||||
.then_some(())
|
||||
})
|
||||
.await;
|
||||
let claimed = agent
|
||||
.network_status(network_id)
|
||||
.await
|
||||
.unwrap()
|
||||
.members
|
||||
.iter()
|
||||
.find(|member| member.endpoint_id == before)
|
||||
.and_then(|member| member.overlay_address_v4)
|
||||
.expect("an address was claimed");
|
||||
agent.shutdown().await;
|
||||
|
||||
let (replacement, released) = {
|
||||
let store = StateStore::open(paths.state_db()).unwrap();
|
||||
store.rotate_device_identity().unwrap()
|
||||
};
|
||||
assert_ne!(replacement.endpoint_id(), before, "a different author");
|
||||
assert_eq!(released, vec![network_id]);
|
||||
|
||||
// The outgoing key signed a release, and it still verifies: a record
|
||||
// whose author no longer runs is not thereby invalid.
|
||||
{
|
||||
let store = StateStore::open(paths.state_db()).unwrap();
|
||||
assert_eq!(
|
||||
store.device_identity().unwrap().map(|id| id.endpoint_id()),
|
||||
Some(replacement.endpoint_id())
|
||||
);
|
||||
|
||||
let mut set = StateSet::new();
|
||||
for record in store.signed_records(network_id).unwrap() {
|
||||
set.merge(network_id, record)
|
||||
.expect("every record verifies");
|
||||
}
|
||||
let old = set.get(&before).expect("the old author is still on record");
|
||||
assert!(matches!(old.body, RecordBody::Release));
|
||||
assert_eq!(
|
||||
set.address_of(&before),
|
||||
None,
|
||||
"the address it held is free again"
|
||||
);
|
||||
assert!(
|
||||
!set.address_holders().contains_key(&claimed),
|
||||
"{claimed} is no longer reserved"
|
||||
);
|
||||
}
|
||||
|
||||
// The whole point: the next run comes up on it.
|
||||
let agent = Agent::spawn(config_with(dir.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(agent.endpoint_id(), replacement.endpoint_id());
|
||||
let network_id = agent.join_network(&name, &secret).await.unwrap();
|
||||
wait_until("the new identity claims an address of its own", || async {
|
||||
let status = agent.network_status(network_id).await.ok()?;
|
||||
status
|
||||
.members
|
||||
.iter()
|
||||
.any(|member| {
|
||||
member.endpoint_id == replacement.endpoint_id()
|
||||
&& member.overlay_address_v4.is_some()
|
||||
})
|
||||
.then_some(())
|
||||
})
|
||||
.await;
|
||||
agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn rotating_a_store_that_has_never_run_just_creates_an_identity() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let paths = StoragePaths::under(dir.path());
|
||||
|
||||
let store = StateStore::open(paths.state_db()).unwrap();
|
||||
assert!(store.device_identity().unwrap().is_none());
|
||||
|
||||
let (identity, released) = store.rotate_device_identity().unwrap();
|
||||
assert!(
|
||||
released.is_empty(),
|
||||
"nothing was held, so nothing is given up"
|
||||
);
|
||||
assert_eq!(
|
||||
store.device_identity().unwrap().map(|id| id.endpoint_id()),
|
||||
Some(identity.endpoint_id())
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn the_hostname_defaults_to_the_machines_own_name() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let dir = TempDir::new().unwrap();
|
||||
let agent = Agent::spawn(config_with(dir.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Whatever this machine is called, the agent uses it rather than
|
||||
// inventing a name from the key — unless the host has no usable one.
|
||||
match tsunagi::agent::system_hostname() {
|
||||
Some(system) => assert_eq!(agent.hostname(), system),
|
||||
None => assert!(agent.hostname().starts_with("tsunagi-")),
|
||||
}
|
||||
|
||||
agent.shutdown().await;
|
||||
}
|
||||
|
||||
/// A secret must not reach the control socket, whatever else `id` prints.
|
||||
#[tokio::test]
|
||||
async fn secrets_stay_out_of_the_status_report() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let dir = TempDir::new().unwrap();
|
||||
let (name, secret) = network("no-secrets-on-the-wire");
|
||||
|
||||
let agent = Agent::spawn(config_with(dir.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
agent.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
let status = agent.status().await.unwrap();
|
||||
let rendered = format!("{status:?}");
|
||||
assert!(!rendered.contains(secret.encode().as_str()));
|
||||
|
||||
agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_network_name_and_secret_are_unaffected_by_the_device_key() {
|
||||
// Network identity is derived from the name and secret only. Replacing
|
||||
// the device key must not move the network the device belongs to.
|
||||
let name = NetworkName::new("stable").unwrap();
|
||||
let secret = NetworkSecret::from_bytes(vec![9u8; 32]).unwrap();
|
||||
let first = NetworkKeys::derive(&name, &secret).network_id();
|
||||
let second = NetworkKeys::derive(&name, &secret).network_id();
|
||||
assert_eq!(first, second);
|
||||
}
|
||||
@@ -0,0 +1,141 @@
|
||||
//! Discovery backends: static bootstrap candidates, composition, and the fact
|
||||
//! that discovery only ever supplies *candidates*.
|
||||
|
||||
#![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;
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_static_bootstrap_candidate_is_enough_to_join() {
|
||||
let (name, secret) = network("bootstrap-only");
|
||||
|
||||
// The listener runs with no discovery at all: it only accepts.
|
||||
let listener_dir = tempfile::TempDir::new().unwrap();
|
||||
let listener = Agent::spawn(local_config(listener_dir.path()))
|
||||
.await
|
||||
.unwrap();
|
||||
let network_id = listener.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
// The joiner is handed the listener's iroh id and addresses up front, which
|
||||
// is exactly what a static bootstrap entry is.
|
||||
let bootstrap = Arc::new(StaticBootstrap::new([listener.local_addr()]));
|
||||
let joiner_dir = tempfile::TempDir::new().unwrap();
|
||||
let joiner = Agent::spawn(local_config(joiner_dir.path()).with_discovery(bootstrap))
|
||||
.await
|
||||
.unwrap();
|
||||
joiner.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
wait_for_peers(&joiner, network_id, 1).await;
|
||||
wait_for_peers(&listener, network_id, 1).await;
|
||||
|
||||
let status = joiner.network_status(network_id).await.unwrap();
|
||||
assert_eq!(
|
||||
status
|
||||
.candidates
|
||||
.iter()
|
||||
.find(|candidate| candidate.endpoint_id == listener.endpoint_id())
|
||||
.map(|candidate| candidate.source),
|
||||
Some(CandidateSource::Bootstrap)
|
||||
);
|
||||
|
||||
joiner.shutdown().await;
|
||||
listener.shutdown().await;
|
||||
drop(joiner);
|
||||
drop(listener);
|
||||
drop(joiner_dir);
|
||||
drop(listener_dir);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_composite_backend_merges_its_sources() {
|
||||
let (name, secret) = network("composite");
|
||||
let keys = NetworkKeys::derive(&name, &secret);
|
||||
|
||||
let shared = SharedMemoryDiscovery::new();
|
||||
let via_shared = TestAgent::spawn(&shared).await.unwrap();
|
||||
let network_id = via_shared.agent.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
let bootstrap_dir = tempfile::TempDir::new().unwrap();
|
||||
let via_bootstrap = Agent::spawn(local_config(bootstrap_dir.path()))
|
||||
.await
|
||||
.unwrap();
|
||||
via_bootstrap.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
// One backend knows the bootstrap peer, the other knows the shared-table
|
||||
// peer. Composed, the joiner reaches both.
|
||||
let composite = Arc::new(CompositeDiscovery::new([
|
||||
Arc::new(StaticBootstrap::new([via_bootstrap.local_addr()])) as Arc<dyn NetworkDiscovery>,
|
||||
Arc::new(shared.clone()) as Arc<dyn NetworkDiscovery>,
|
||||
]));
|
||||
let joiner_dir = tempfile::TempDir::new().unwrap();
|
||||
let joiner = Agent::spawn(local_config(joiner_dir.path()).with_discovery(composite))
|
||||
.await
|
||||
.unwrap();
|
||||
joiner.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
wait_for_peers(&joiner, network_id, 2).await;
|
||||
|
||||
// Publishing went to the shared backend, so the other agent finds us too.
|
||||
assert!(shared.len(&keys.discovery_key()) >= 2);
|
||||
|
||||
joiner.shutdown().await;
|
||||
via_bootstrap.shutdown().await;
|
||||
via_shared.agent.shutdown().await;
|
||||
drop(joiner);
|
||||
drop(via_bootstrap);
|
||||
drop(joiner_dir);
|
||||
drop(bootstrap_dir);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn discovery_entries_are_withdrawn_when_a_network_stops() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("withdrawn");
|
||||
let keys = NetworkKeys::derive(&name, &secret);
|
||||
|
||||
let agent = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let network_id = agent.agent.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
settle().await;
|
||||
assert_eq!(discovery.len(&keys.discovery_key()), 1);
|
||||
|
||||
agent.agent.deactivate_network(network_id).await.unwrap();
|
||||
assert!(discovery.is_empty(&keys.discovery_key()));
|
||||
|
||||
agent.agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn forgetting_a_network_removes_it_from_the_state_store() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("forgettable");
|
||||
|
||||
let agent = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let network_id = agent.agent.join_network(&name, &secret).await.unwrap();
|
||||
assert_eq!(agent.agent.list_networks().await.unwrap().len(), 1);
|
||||
|
||||
agent.agent.forget_network(network_id).await.unwrap();
|
||||
assert!(agent.agent.list_networks().await.unwrap().is_empty());
|
||||
assert!(!agent.agent.is_active(network_id).await);
|
||||
|
||||
let dir = agent.stop().await;
|
||||
let restarted =
|
||||
Agent::spawn(local_config(dir.path()).with_discovery(Arc::new(discovery.clone())))
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(restarted.list_networks().await.unwrap().is_empty());
|
||||
assert_eq!(restarted.status().await.unwrap().networks.len(), 0);
|
||||
|
||||
restarted.shutdown().await;
|
||||
drop(restarted);
|
||||
drop(dir);
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
//! The full vertical slice: persistent identity, network space, discovery,
|
||||
//! real iroh connections, authentication and a control message exchange.
|
||||
|
||||
#![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;
|
||||
|
||||
#[tokio::test]
|
||||
async fn two_agents_authenticate_and_exchange_messages() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("vertical-slice");
|
||||
|
||||
let a = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let b = TestAgent::spawn(&discovery).await.unwrap();
|
||||
|
||||
let mut events_a = a.agent.subscribe();
|
||||
let network_id = a.agent.join_network(&name, &secret).await.unwrap();
|
||||
let same_id = b.agent.join_network(&name, &secret).await.unwrap();
|
||||
assert_eq!(
|
||||
network_id, same_id,
|
||||
"the same name and secret must produce the same network space"
|
||||
);
|
||||
|
||||
let peers = wait_for_peers(&a.agent, network_id, 1).await;
|
||||
assert_eq!(peers, vec![b.agent.endpoint_id()]);
|
||||
|
||||
// The exchange itself: a ping must come back as a matching pong.
|
||||
a.agent
|
||||
.send(
|
||||
network_id,
|
||||
b.agent.endpoint_id(),
|
||||
ControlMessage::Ping {
|
||||
seq: 42,
|
||||
payload: b"vertical".to_vec(),
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let payload = wait_event(&mut events_a, |event| match event {
|
||||
Event::MessageReceived {
|
||||
network,
|
||||
peer,
|
||||
message: ControlMessage::Pong { seq: 42, payload },
|
||||
} if *network == network_id && *peer == b.agent.endpoint_id() => Some(payload.clone()),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
assert_eq!(payload, b"vertical".to_vec());
|
||||
|
||||
// Both sides announce a hostname over the authenticated session.
|
||||
let status = a.agent.network_status(network_id).await.unwrap();
|
||||
let peer = &status.peers[0];
|
||||
assert_eq!(peer.hostname.as_deref(), Some(b.agent.hostname().as_str()));
|
||||
assert!(peer.transport != tsunagi::net::TransportKind::Unknown);
|
||||
assert!(peer.rtt.is_some(), "a verified path must report an RTT");
|
||||
|
||||
a.agent.shutdown().await;
|
||||
b.agent.shutdown().await;
|
||||
}
|
||||
@@ -0,0 +1,175 @@
|
||||
//! Scenario 1: deterministic network identity.
|
||||
//!
|
||||
//! The same name and secret must yield the same network space on different
|
||||
//! devices, and nothing else — hostname, device key, restart — may change it.
|
||||
|
||||
#![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};
|
||||
|
||||
#[test]
|
||||
fn derivation_is_a_pure_function_of_name_and_secret() {
|
||||
let name = NetworkName::new("home").unwrap();
|
||||
let other = NetworkName::new("Home").unwrap();
|
||||
let secret = NetworkSecret::generate();
|
||||
let other_secret = NetworkSecret::generate();
|
||||
|
||||
let a = NetworkKeys::derive(&name, &secret);
|
||||
let b = NetworkKeys::derive(&name, &secret);
|
||||
assert_eq!(a.network_id(), b.network_id());
|
||||
assert_eq!(a.discovery_key(), b.discovery_key());
|
||||
assert_eq!(a.descriptor(), b.descriptor());
|
||||
|
||||
// A different name is a different space. Names are used verbatim, so case
|
||||
// matters.
|
||||
assert_ne!(
|
||||
a.network_id(),
|
||||
NetworkKeys::derive(&other, &secret).network_id()
|
||||
);
|
||||
// A different secret is a different space.
|
||||
assert_ne!(
|
||||
a.network_id(),
|
||||
NetworkKeys::derive(&name, &other_secret).network_id()
|
||||
);
|
||||
// Separated key material: the discovery key is not the network id.
|
||||
assert_ne!(a.network_id().as_bytes(), a.discovery_key().as_bytes());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn descriptor_carries_no_creator_time_or_secret() {
|
||||
let name = NetworkName::new("shared").unwrap();
|
||||
let secret = NetworkSecret::generate();
|
||||
let first = NetworkKeys::derive(&name, &secret).descriptor();
|
||||
let second = NetworkKeys::derive(&name, &secret).descriptor();
|
||||
|
||||
// Two independently built descriptors are byte identical: no random
|
||||
// creator id, no creation timestamp, no owner signature.
|
||||
assert_eq!(first.to_canonical_bytes(), second.to_canonical_bytes());
|
||||
|
||||
let encoded = first.to_canonical_bytes();
|
||||
let secret_bytes = secret.encode();
|
||||
assert!(
|
||||
!encoded
|
||||
.windows(secret_bytes.len())
|
||||
.any(|window| window == secret_bytes.as_bytes()),
|
||||
"the secret must never appear in the public descriptor"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn names_are_validated_not_silently_normalised() {
|
||||
assert!(NetworkName::new("").is_err());
|
||||
assert!(NetworkName::new(" home").is_err(), "must not be trimmed");
|
||||
assert!(NetworkName::new("home ").is_err(), "must not be trimmed");
|
||||
assert!(NetworkName::new("ho\nme").is_err());
|
||||
assert!(NetworkName::new("a".repeat(65)).is_err());
|
||||
assert_eq!(NetworkName::new("home").unwrap().as_str(), "home");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn secrets_are_not_truncated_or_normalised() {
|
||||
let secret = NetworkSecret::generate();
|
||||
let text = secret.encode();
|
||||
let round_tripped = NetworkSecret::decode(&text).unwrap();
|
||||
assert_eq!(secret, round_tripped);
|
||||
|
||||
// Short secrets are rejected rather than stretched.
|
||||
assert!(NetworkSecret::from_bytes(vec![7u8; 15]).is_err());
|
||||
assert!(NetworkSecret::from_bytes(vec![7u8; 16]).is_ok());
|
||||
|
||||
// Debug output must not leak the secret.
|
||||
let rendered = format!("{secret:?}");
|
||||
assert_eq!(rendered, "NetworkSecret(<redacted>)");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn different_devices_and_hostnames_agree_on_the_network_id() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("agreement");
|
||||
|
||||
let a = TestAgent::spawn_with(|cfg| cfg.with_hostname("alpha"), &discovery)
|
||||
.await
|
||||
.unwrap();
|
||||
let b = TestAgent::spawn_with(|cfg| cfg.with_hostname("beta"), &discovery)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_ne!(
|
||||
a.agent.endpoint_id(),
|
||||
b.agent.endpoint_id(),
|
||||
"different devices must have different endpoint ids"
|
||||
);
|
||||
assert_eq!(a.agent.hostname(), "alpha");
|
||||
assert_eq!(b.agent.hostname(), "beta");
|
||||
|
||||
let id_a = a.agent.join_network(&name, &secret).await.unwrap();
|
||||
let id_b = b.agent.join_network(&name, &secret).await.unwrap();
|
||||
assert_eq!(id_a, id_b);
|
||||
|
||||
a.agent.shutdown().await;
|
||||
b.agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn restart_keeps_the_device_and_network_identity() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("stable");
|
||||
|
||||
let first = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let device_id = first.agent.endpoint_id();
|
||||
let network_id = first.agent.join_network(&name, &secret).await.unwrap();
|
||||
let dir = first.stop().await;
|
||||
|
||||
let reopened = Agent::spawn(config_with(dir.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
reopened.endpoint_id(),
|
||||
device_id,
|
||||
"restarting must not mint a new peer"
|
||||
);
|
||||
let networks = reopened.list_networks().await.unwrap();
|
||||
assert_eq!(networks.len(), 1);
|
||||
assert_eq!(networks[0].network_id, network_id);
|
||||
assert!(networks[0].active, "auto-start networks come back up");
|
||||
|
||||
reopened.shutdown().await;
|
||||
drop(reopened);
|
||||
drop(dir);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn changing_the_secret_keeps_the_device_identity() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let dir = tempfile::TempDir::new().unwrap();
|
||||
let agent = Agent::spawn(config_with(dir.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
let device_id = agent.endpoint_id();
|
||||
|
||||
let name = NetworkName::new("rotating").unwrap();
|
||||
let old = agent
|
||||
.join_network(&name, &NetworkSecret::generate())
|
||||
.await
|
||||
.unwrap();
|
||||
let new = agent
|
||||
.join_network(&name, &NetworkSecret::generate())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_ne!(old, new, "a new secret is a new network space");
|
||||
assert_eq!(
|
||||
agent.endpoint_id(),
|
||||
device_id,
|
||||
"rotating the network secret must not change the persistent iroh id"
|
||||
);
|
||||
|
||||
agent.shutdown().await;
|
||||
drop(agent);
|
||||
drop(dir);
|
||||
}
|
||||
@@ -0,0 +1,244 @@
|
||||
//! 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_interface_prefix(tag)
|
||||
.with_reconcile(Duration::from_millis(20), Duration::from_millis(100));
|
||||
let plugin = WireguardPlugin::open(config, factory).await.unwrap();
|
||||
let agent = Agent::spawn(
|
||||
config_with(dir.path(), discovery)
|
||||
.with_overlay_ipv4_range(Some(DEFAULT_IPV4_RANGE))
|
||||
.with_plugin(plugin.clone() as Arc<dyn IpPlugin>),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
Self {
|
||||
_dir: dir,
|
||||
agent,
|
||||
plugin,
|
||||
host,
|
||||
}
|
||||
}
|
||||
|
||||
/// The interface name the plugin settled on for a network.
|
||||
async fn interface(&self, network: NetworkId) -> String {
|
||||
wait_until("the plugin named its interface", || async {
|
||||
self.plugin
|
||||
.overview(network)
|
||||
.map(|view| view.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()
|
||||
}
|
||||
|
||||
fn has_v6(state: &InterfaceState) -> bool {
|
||||
state
|
||||
.addresses
|
||||
.iter()
|
||||
.any(|cidr| matches!(cidr.addr, IpAddr::V6(_)))
|
||||
}
|
||||
|
||||
#[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.filter(|state| !state.addresses.is_empty())
|
||||
})
|
||||
.await;
|
||||
|
||||
assert_eq!(state.kind, LinkKind::Tun);
|
||||
assert!(state.up, "the agent brought the link up itself");
|
||||
assert_eq!(state.mtu, 1280);
|
||||
assert!(has_v6(&state), "the derived overlay address is assigned");
|
||||
|
||||
// The IPv4 address is allocated at run time, so it arrives on a later
|
||||
// reconciliation than the interface itself.
|
||||
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);
|
||||
|
||||
let state = agent
|
||||
.wait_for_host("the interface to be rebuilt", &interface, |state| {
|
||||
state.filter(|state| state.attached && has_v6(state))
|
||||
})
|
||||
.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_removes_the_interface_from_the_host() {
|
||||
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 interface to exist", &interface, |state| state)
|
||||
.await;
|
||||
|
||||
agent.agent.deactivate_network(network_id).await.unwrap();
|
||||
|
||||
agent
|
||||
.wait_for_host("the interface to be removed", &interface, |state| {
|
||||
state.is_none().then_some(())
|
||||
})
|
||||
.await;
|
||||
assert!(
|
||||
agent.host.names().is_empty(),
|
||||
"nothing is left behind: {:?}",
|
||||
agent.host.names()
|
||||
);
|
||||
|
||||
agent.agent.shutdown().await;
|
||||
}
|
||||
@@ -0,0 +1,234 @@
|
||||
//! 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 {
|
||||
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,
|
||||
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_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<dyn IpPlugin>),
|
||||
)
|
||||
.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<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_empty());
|
||||
|
||||
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"))
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,179 @@
|
||||
//! Scenario 2: several agents find each other, authenticate for real and
|
||||
//! exchange distinguishable messages.
|
||||
|
||||
#![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;
|
||||
|
||||
#[tokio::test]
|
||||
async fn four_agents_form_a_mesh_and_exchange_distinguishable_messages() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("mesh-of-four");
|
||||
|
||||
let mut agents = Vec::new();
|
||||
for index in 0..4 {
|
||||
let hostname = format!("host-{index}");
|
||||
agents.push(
|
||||
TestAgent::spawn_with(move |cfg| cfg.with_hostname(hostname), &discovery)
|
||||
.await
|
||||
.unwrap(),
|
||||
);
|
||||
}
|
||||
|
||||
let mut network_ids = HashSet::new();
|
||||
for agent in &agents {
|
||||
network_ids.insert(agent.agent.join_network(&name, &secret).await.unwrap());
|
||||
}
|
||||
assert_eq!(network_ids.len(), 1, "one deterministic network space");
|
||||
let network_id = network_ids.into_iter().next().unwrap();
|
||||
|
||||
// Full mesh: every agent must end up with the other three.
|
||||
for agent in &agents {
|
||||
wait_for_peers(&agent.agent, network_id, 3).await;
|
||||
}
|
||||
|
||||
// Each peer announced its own hostname, so sessions are distinguishable.
|
||||
// A peer counts as connected as soon as its session is authenticated, which
|
||||
// can be a round before its announcement carrying the hostname arrives, so
|
||||
// this waits for the hostnames rather than reading them straight away.
|
||||
let hostnames: HashSet<String> = wait_until("three distinct peer hostnames", || async {
|
||||
let status = agents[0].agent.network_status(network_id).await.ok()?;
|
||||
let hostnames: HashSet<String> = status
|
||||
.peers
|
||||
.iter()
|
||||
.filter_map(|peer| peer.hostname.clone())
|
||||
.collect();
|
||||
(hostnames.len() >= 3).then_some(hostnames)
|
||||
})
|
||||
.await;
|
||||
assert_eq!(
|
||||
hostnames.len(),
|
||||
3,
|
||||
"three distinct hostnames: {hostnames:?}"
|
||||
);
|
||||
|
||||
// Distinguishable request/response: each peer echoes its own sequence.
|
||||
let mut events = agents[0].agent.subscribe();
|
||||
for (index, peer) in agents.iter().skip(1).enumerate() {
|
||||
agents[0]
|
||||
.agent
|
||||
.send(
|
||||
network_id,
|
||||
peer.agent.endpoint_id(),
|
||||
ControlMessage::Ping {
|
||||
seq: index as u64 + 1,
|
||||
payload: format!("to-{index}").into_bytes(),
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
let mut seen = HashSet::new();
|
||||
while seen.len() < 3 {
|
||||
let (peer, seq, payload) = wait_event(&mut events, |event| match event {
|
||||
Event::MessageReceived {
|
||||
network,
|
||||
peer,
|
||||
message: ControlMessage::Pong { seq, payload },
|
||||
} if *network == network_id => Some((*peer, *seq, payload.clone())),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
assert_eq!(payload, format!("to-{}", seq - 1).into_bytes());
|
||||
seen.insert(peer);
|
||||
}
|
||||
assert_eq!(seen.len(), 3);
|
||||
|
||||
for agent in agents {
|
||||
agent.agent.shutdown().await;
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_late_joiner_is_picked_up_by_the_existing_members() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("late-joiner");
|
||||
|
||||
let first = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let network_id = first.agent.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
// Nobody else is there yet. That is "nobody found so far", not proof that
|
||||
// the network is empty, and the agent is ready regardless.
|
||||
let status = first.agent.network_status(network_id).await.unwrap();
|
||||
assert!(status.peers.is_empty());
|
||||
|
||||
let second = TestAgent::spawn(&discovery).await.unwrap();
|
||||
second.agent.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
wait_for_peers(&first.agent, network_id, 1).await;
|
||||
wait_for_peers(&second.agent, network_id, 1).await;
|
||||
|
||||
first.agent.shutdown().await;
|
||||
second.agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn opaque_plugin_capabilities_cross_the_control_plane() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("capabilities");
|
||||
|
||||
// An explicitly test-only capability: nothing here advertises WireGuard.
|
||||
let plugin_a = Arc::new(TestCapabilityPlugin::new("test-ip", b"payload-a".to_vec()));
|
||||
let plugin_b = Arc::new(TestCapabilityPlugin::new("test-ip", b"payload-b".to_vec()));
|
||||
|
||||
let a = TestAgent::spawn_with(
|
||||
{
|
||||
let plugin = Arc::clone(&plugin_a);
|
||||
move |cfg| cfg.with_plugin(plugin)
|
||||
},
|
||||
&discovery,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let b = TestAgent::spawn_with(
|
||||
{
|
||||
let plugin = Arc::clone(&plugin_b);
|
||||
move |cfg| cfg.with_plugin(plugin)
|
||||
},
|
||||
&discovery,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let network_id = a.agent.join_network(&name, &secret).await.unwrap();
|
||||
b.agent.join_network(&name, &secret).await.unwrap();
|
||||
wait_for_peers(&a.agent, network_id, 1).await;
|
||||
|
||||
let observed = wait_until("plugin a sees b's capability", || {
|
||||
let plugin = Arc::clone(&plugin_a);
|
||||
async move {
|
||||
let seen = plugin.observed();
|
||||
if seen.is_empty() { None } else { Some(seen) }
|
||||
}
|
||||
})
|
||||
.await;
|
||||
|
||||
let (seen_network, seen_peer, capability) = &observed[0];
|
||||
assert_eq!(*seen_network, network_id);
|
||||
assert_eq!(*seen_peer, b.agent.endpoint_id());
|
||||
assert_eq!(capability.protocol, "test-ip");
|
||||
assert_eq!(capability.data, b"payload-b".to_vec());
|
||||
assert!(capability.enabled);
|
||||
|
||||
// The core carried the payload without interpreting it.
|
||||
let status = a.agent.network_status(network_id).await.unwrap();
|
||||
assert_eq!(status.peers[0].capabilities[0].data, b"payload-b".to_vec());
|
||||
|
||||
a.agent.shutdown().await;
|
||||
b.agent.shutdown().await;
|
||||
}
|
||||
@@ -0,0 +1,266 @@
|
||||
//! Scenario 4: one agent in two networks at once, with no bleed between them.
|
||||
|
||||
#![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;
|
||||
use tsunagi::discovery::SharedMemoryDiscovery;
|
||||
use tsunagi::identity::NetworkKeys;
|
||||
use tsunagi::proto::handshake::ROLE_INITIATOR;
|
||||
use tsunagi::proto::message::{
|
||||
ALPN, AuthProof, ControlMessage, Envelope, Hello, HelloAck, PROTOCOL_VERSION, decode, encode,
|
||||
};
|
||||
use tsunagi::proto::{read_frame, write_frame};
|
||||
use tsunagi::test_support;
|
||||
|
||||
const LIMIT: usize = 64 * 1024;
|
||||
|
||||
#[tokio::test]
|
||||
async fn one_agent_in_two_networks_keeps_them_apart() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name_a, secret_a) = network("alpha-net");
|
||||
let (name_b, secret_b) = network("beta-net");
|
||||
|
||||
let hub = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let alpha_peer = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let beta_peer = TestAgent::spawn(&discovery).await.unwrap();
|
||||
|
||||
let alpha = hub.agent.join_network(&name_a, &secret_a).await.unwrap();
|
||||
let beta = hub.agent.join_network(&name_b, &secret_b).await.unwrap();
|
||||
assert_ne!(alpha, beta);
|
||||
|
||||
alpha_peer
|
||||
.agent
|
||||
.join_network(&name_a, &secret_a)
|
||||
.await
|
||||
.unwrap();
|
||||
beta_peer
|
||||
.agent
|
||||
.join_network(&name_b, &secret_b)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
wait_for_peers(&hub.agent, alpha, 1).await;
|
||||
wait_for_peers(&hub.agent, beta, 1).await;
|
||||
|
||||
// Statuses do not mix: each network sees exactly its own peer.
|
||||
let status_alpha = hub.agent.network_status(alpha).await.unwrap();
|
||||
let status_beta = hub.agent.network_status(beta).await.unwrap();
|
||||
assert_eq!(
|
||||
status_alpha.connected_peers(),
|
||||
vec![alpha_peer.agent.endpoint_id()]
|
||||
);
|
||||
assert_eq!(
|
||||
status_beta.connected_peers(),
|
||||
vec![beta_peer.agent.endpoint_id()]
|
||||
);
|
||||
|
||||
// Addressing a peer of one network through the other is refused locally.
|
||||
let wrong = hub
|
||||
.agent
|
||||
.send(
|
||||
alpha,
|
||||
beta_peer.agent.endpoint_id(),
|
||||
ControlMessage::Ping {
|
||||
seq: 1,
|
||||
payload: Vec::new(),
|
||||
},
|
||||
)
|
||||
.await;
|
||||
assert!(matches!(wrong, Err(tsunagi::Error::NoSuchPeer { .. })));
|
||||
|
||||
// Messages stay in their own network.
|
||||
let mut events = hub.agent.subscribe();
|
||||
hub.agent
|
||||
.broadcast(
|
||||
alpha,
|
||||
ControlMessage::Ping {
|
||||
seq: 7,
|
||||
payload: b"alpha-only".to_vec(),
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let from = wait_event(&mut events, |event| match event {
|
||||
Event::MessageReceived {
|
||||
network,
|
||||
peer,
|
||||
message: ControlMessage::Pong { seq: 7, payload },
|
||||
} if payload == b"alpha-only" => Some((*network, *peer)),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
assert_eq!(from, (alpha, alpha_peer.agent.endpoint_id()));
|
||||
|
||||
let beta_status = hub.agent.network_status(beta).await.unwrap();
|
||||
assert_eq!(
|
||||
beta_status.metrics.control_messages_received,
|
||||
beta_status.peers[0].control_messages_received,
|
||||
"beta's counters are its own"
|
||||
);
|
||||
assert!(
|
||||
beta_status
|
||||
.peers
|
||||
.iter()
|
||||
.all(|peer| peer.endpoint_id != alpha_peer.agent.endpoint_id())
|
||||
);
|
||||
|
||||
// Deactivating one network must not disturb the other.
|
||||
hub.agent.deactivate_network(alpha).await.unwrap();
|
||||
assert!(hub.agent.network_status(alpha).await.is_err());
|
||||
wait_for_peers(&hub.agent, beta, 1).await;
|
||||
hub.agent
|
||||
.send(
|
||||
beta,
|
||||
beta_peer.agent.endpoint_id(),
|
||||
ControlMessage::Ping {
|
||||
seq: 8,
|
||||
payload: b"still-here".to_vec(),
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
wait_event(&mut events, |event| match event {
|
||||
Event::MessageReceived {
|
||||
network,
|
||||
message: ControlMessage::Pong { seq: 8, .. },
|
||||
..
|
||||
} if *network == beta => Some(()),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
|
||||
hub.agent.shutdown().await;
|
||||
alpha_peer.agent.shutdown().await;
|
||||
beta_peer.agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_authenticated_session_cannot_speak_for_another_network() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name_a, secret_a) = network("session-scope-a");
|
||||
let (name_b, secret_b) = network("session-scope-b");
|
||||
|
||||
let hub = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let mut events = hub.agent.subscribe();
|
||||
let alpha = hub.agent.join_network(&name_a, &secret_a).await.unwrap();
|
||||
let beta = hub.agent.join_network(&name_b, &secret_b).await.unwrap();
|
||||
|
||||
// A genuine member of `alpha`, driven by hand so it can misbehave.
|
||||
let keys = NetworkKeys::derive(&name_a, &secret_a);
|
||||
let auth_key = test_support::auth_key(&keys);
|
||||
let member = Endpoint::builder(presets::Minimal)
|
||||
.alpns(vec![ALPN.to_vec()])
|
||||
.relay_mode(RelayMode::Disabled)
|
||||
.clear_address_lookup()
|
||||
.portmapper_config(PortmapperConfig::Disabled)
|
||||
.clear_ip_transports()
|
||||
.bind_addr("127.0.0.1:0")
|
||||
.unwrap()
|
||||
.bind()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let conn = member.connect(hub.agent.local_addr(), ALPN).await.unwrap();
|
||||
let (mut send, mut recv) = conn.open_bi().await.unwrap();
|
||||
|
||||
let alpha_bytes = *alpha.as_bytes();
|
||||
let nonce_i = [5u8; 16];
|
||||
let hello = Hello {
|
||||
version: PROTOCOL_VERSION,
|
||||
network_id: alpha_bytes,
|
||||
nonce: nonce_i,
|
||||
};
|
||||
write_frame(&mut send, &encode(&hello).unwrap(), LIMIT)
|
||||
.await
|
||||
.unwrap();
|
||||
let ack: HelloAck = decode(&read_frame(&mut recv, LIMIT).await.unwrap()).unwrap();
|
||||
|
||||
let cb = test_support::channel_binding(&conn, &alpha_bytes).unwrap();
|
||||
let proof = test_support::compute_proof(
|
||||
&auth_key,
|
||||
ROLE_INITIATOR,
|
||||
PROTOCOL_VERSION,
|
||||
&alpha_bytes,
|
||||
member.id().as_bytes(),
|
||||
hub.agent.endpoint_id().as_bytes(),
|
||||
&cb,
|
||||
&nonce_i,
|
||||
&ack.nonce,
|
||||
);
|
||||
write_frame(&mut send, &encode(&AuthProof { proof }).unwrap(), LIMIT)
|
||||
.await
|
||||
.unwrap();
|
||||
let _responder_proof: AuthProof = decode(&read_frame(&mut recv, LIMIT).await.unwrap()).unwrap();
|
||||
|
||||
// Authenticated for alpha. Now try to speak for beta on the same session.
|
||||
let smuggled = Envelope {
|
||||
network_id: *beta.as_bytes(),
|
||||
message: ControlMessage::Ping {
|
||||
seq: 99,
|
||||
payload: b"wrong network".to_vec(),
|
||||
},
|
||||
};
|
||||
write_frame(&mut send, &encode(&smuggled).unwrap(), LIMIT)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let reason = wait_event(&mut events, |event| match event {
|
||||
Event::ProtocolViolation {
|
||||
network, reason, ..
|
||||
} if *network == Some(alpha) => Some(reason.clone()),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
assert!(
|
||||
reason.contains("network id does not match"),
|
||||
"unexpected reason: {reason}"
|
||||
);
|
||||
|
||||
// Beta saw nothing at all, and both networks keep running.
|
||||
let beta_status = hub.agent.network_status(beta).await.unwrap();
|
||||
assert_eq!(beta_status.metrics.control_messages_received, 0);
|
||||
assert!(beta_status.peers.is_empty());
|
||||
assert!(hub.agent.network_status(alpha).await.is_ok());
|
||||
|
||||
conn.close(0u32.into(), b"done");
|
||||
member.close().await;
|
||||
hub.agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn deactivating_one_network_leaves_the_agent_and_others_running() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name_a, secret_a) = network("keep-a");
|
||||
let (name_b, secret_b) = network("keep-b");
|
||||
|
||||
let agent = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let alpha = agent.agent.join_network(&name_a, &secret_a).await.unwrap();
|
||||
let beta = agent.agent.join_network(&name_b, &secret_b).await.unwrap();
|
||||
|
||||
agent.agent.deactivate_network(alpha).await.unwrap();
|
||||
settle().await;
|
||||
|
||||
let status = agent.agent.status().await.unwrap();
|
||||
assert_eq!(status.networks.len(), 2, "both stay configured");
|
||||
assert_eq!(
|
||||
status.network(&alpha).map(|net| net.state),
|
||||
Some(tsunagi::agent::NetworkState::Inactive)
|
||||
);
|
||||
assert_eq!(
|
||||
status.network(&beta).map(|net| net.state),
|
||||
Some(tsunagi::agent::NetworkState::Active)
|
||||
);
|
||||
|
||||
// Deactivating twice is an error, not a crash.
|
||||
assert!(agent.agent.deactivate_network(alpha).await.is_err());
|
||||
// And it can be brought back.
|
||||
agent.agent.activate_network(alpha).await.unwrap();
|
||||
assert!(agent.agent.network_status(alpha).await.is_ok());
|
||||
|
||||
agent.agent.shutdown().await;
|
||||
}
|
||||
@@ -0,0 +1,345 @@
|
||||
//! Scenarios 8 and 10: unreachable participants, bounded retries, and the
|
||||
//! agent lifecycle including state directory ownership.
|
||||
|
||||
#![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::{Agent, Error};
|
||||
|
||||
/// An endpoint id nobody is listening for, at an address nothing answers on.
|
||||
fn dead_candidate() -> EndpointAddr {
|
||||
let unreachable: SocketAddr = "127.0.0.1:1".parse().unwrap();
|
||||
EndpointAddr::new(SecretKey::generate().public()).with_ip_addr(unreachable)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_dead_candidate_does_not_hold_up_the_reachable_ones() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("dead-candidate");
|
||||
let keys = NetworkKeys::derive(&name, &secret);
|
||||
|
||||
// Poison the rendezvous table before anybody real shows up.
|
||||
let dead = dead_candidate();
|
||||
discovery.insert_raw(keys.discovery_key(), dead.clone());
|
||||
|
||||
let a = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let b = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let mut events = a.agent.subscribe();
|
||||
|
||||
let network_id = a.agent.join_network(&name, &secret).await.unwrap();
|
||||
b.agent.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
// The reachable peer still connects.
|
||||
let peers = wait_for_peers(&a.agent, network_id, 1).await;
|
||||
assert_eq!(peers, vec![b.agent.endpoint_id()]);
|
||||
|
||||
// And the dead candidate is reported as a failure, not silently forgotten.
|
||||
wait_event(&mut events, |event| match event {
|
||||
Event::DialFailed { peer, .. } if *peer == dead.id => Some(()),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
|
||||
let status = a.agent.network_status(network_id).await.unwrap();
|
||||
assert!(
|
||||
status
|
||||
.candidates
|
||||
.iter()
|
||||
.any(|candidate| candidate.endpoint_id == dead.id
|
||||
&& candidate.consecutive_failures > 0),
|
||||
"a failing candidate must stay visible as an unverified candidate"
|
||||
);
|
||||
assert!(
|
||||
status.peers.iter().all(|peer| peer.endpoint_id != dead.id),
|
||||
"a candidate must never be reported as a peer"
|
||||
);
|
||||
|
||||
a.agent.shutdown().await;
|
||||
b.agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_vanished_peer_is_retried_with_backoff_and_others_keep_working() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("vanishing-peer");
|
||||
|
||||
let watcher = TestAgent::spawn_with(
|
||||
|cfg| {
|
||||
cfg.with_reconnect(ReconnectPolicy {
|
||||
initial_delay: Duration::from_millis(50),
|
||||
max_delay: Duration::from_millis(200),
|
||||
factor: 1.5,
|
||||
jitter: 0.2,
|
||||
max_consecutive_failures: None,
|
||||
})
|
||||
},
|
||||
&discovery,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let stayer = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let leaver = TestAgent::spawn(&discovery).await.unwrap();
|
||||
|
||||
let network_id = watcher.agent.join_network(&name, &secret).await.unwrap();
|
||||
stayer.agent.join_network(&name, &secret).await.unwrap();
|
||||
leaver.agent.join_network(&name, &secret).await.unwrap();
|
||||
wait_for_peers(&watcher.agent, network_id, 2).await;
|
||||
|
||||
let leaver_id = leaver.agent.endpoint_id();
|
||||
let mut events = watcher.agent.subscribe();
|
||||
leaver.agent.shutdown().await;
|
||||
drop(leaver);
|
||||
|
||||
wait_event(&mut events, |event| match event {
|
||||
Event::PeerDisconnected { peer, .. } if *peer == leaver_id => Some(()),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
|
||||
// The other peer is untouched and still answers.
|
||||
watcher
|
||||
.agent
|
||||
.send(
|
||||
network_id,
|
||||
stayer.agent.endpoint_id(),
|
||||
ControlMessage::Ping {
|
||||
seq: 3,
|
||||
payload: b"still here".to_vec(),
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
wait_event(&mut events, |event| match event {
|
||||
Event::MessageReceived {
|
||||
peer,
|
||||
message: ControlMessage::Pong { seq: 3, .. },
|
||||
..
|
||||
} if *peer == stayer.agent.endpoint_id() => Some(()),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
|
||||
// The watcher does retry the peer that went away.
|
||||
wait_event(&mut events, |event| match event {
|
||||
Event::DialFailed { peer, .. } if *peer == leaver_id => Some(()),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
|
||||
// Retries are bounded by the backoff rather than spinning.
|
||||
settle().await;
|
||||
let status = watcher.agent.network_status(network_id).await.unwrap();
|
||||
let failures = status
|
||||
.candidates
|
||||
.iter()
|
||||
.find(|candidate| candidate.endpoint_id == leaver_id)
|
||||
.map(|candidate| candidate.consecutive_failures)
|
||||
.unwrap_or(0);
|
||||
assert!(
|
||||
(1..=40).contains(&failures),
|
||||
"expected bounded backed-off retries, got {failures}"
|
||||
);
|
||||
assert_eq!(
|
||||
status.connected_peers(),
|
||||
vec![stayer.agent.endpoint_id()],
|
||||
"the surviving peer keeps its session"
|
||||
);
|
||||
|
||||
watcher.agent.shutdown().await;
|
||||
stayer.agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn retries_stop_when_the_network_is_deactivated() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("stop-retrying");
|
||||
let keys = NetworkKeys::derive(&name, &secret);
|
||||
discovery.insert_raw(keys.discovery_key(), dead_candidate());
|
||||
|
||||
let agent = TestAgent::spawn_with(
|
||||
|cfg| {
|
||||
cfg.with_discovery_interval(Duration::from_millis(80))
|
||||
.with_reconnect(ReconnectPolicy {
|
||||
initial_delay: Duration::from_millis(20),
|
||||
max_delay: Duration::from_millis(60),
|
||||
factor: 1.2,
|
||||
jitter: 0.1,
|
||||
max_consecutive_failures: None,
|
||||
})
|
||||
},
|
||||
&discovery,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let mut events = agent.agent.subscribe();
|
||||
let network_id = agent.agent.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
// Retries are definitely happening.
|
||||
wait_event(&mut events, |event| match event {
|
||||
Event::DialFailed { .. } => Some(()),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
|
||||
agent.agent.deactivate_network(network_id).await.unwrap();
|
||||
|
||||
// Drain whatever was already queued, then require silence.
|
||||
while events.try_recv().is_ok() {}
|
||||
settle().await;
|
||||
let mut stragglers = 0;
|
||||
while let Ok(event) = events.try_recv() {
|
||||
if matches!(event, Event::DialFailed { .. }) {
|
||||
stragglers += 1;
|
||||
}
|
||||
}
|
||||
assert_eq!(
|
||||
stragglers, 0,
|
||||
"a deactivated network must stop dialling entirely"
|
||||
);
|
||||
|
||||
agent.agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_second_agent_on_the_same_state_directory_is_refused() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let dir = tempfile::TempDir::new().unwrap();
|
||||
|
||||
let first = Agent::spawn(config_with(dir.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let second = Agent::spawn(config_with(dir.path(), &discovery)).await;
|
||||
match second {
|
||||
Err(Error::StateLocked { path }) => {
|
||||
assert!(path.starts_with(dir.path()));
|
||||
}
|
||||
Err(other) => panic!("expected StateLocked, got {other:?}"),
|
||||
Ok(agent) => {
|
||||
agent.shutdown().await;
|
||||
panic!("two live agents must not share one state directory");
|
||||
}
|
||||
}
|
||||
|
||||
// After a clean stop the directory is immediately claimable again.
|
||||
let device_id = first.endpoint_id();
|
||||
first.shutdown().await;
|
||||
drop(first);
|
||||
|
||||
let third = Agent::spawn(config_with(dir.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(third.endpoint_id(), device_id);
|
||||
third.shutdown().await;
|
||||
drop(third);
|
||||
drop(dir);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn shutdown_releases_resources_and_rejects_further_work() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("clean-stop");
|
||||
|
||||
let a = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let b = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let network_id = a.agent.join_network(&name, &secret).await.unwrap();
|
||||
b.agent.join_network(&name, &secret).await.unwrap();
|
||||
wait_for_peers(&a.agent, network_id, 1).await;
|
||||
|
||||
a.agent.shutdown().await;
|
||||
|
||||
// The endpoint is closed and the networks are gone.
|
||||
assert!(a.agent.endpoint().is_closed());
|
||||
assert!(matches!(
|
||||
a.agent.network_status(network_id).await,
|
||||
Err(Error::NetworkNotActive(_))
|
||||
));
|
||||
assert!(
|
||||
a.agent
|
||||
.send(
|
||||
network_id,
|
||||
b.agent.endpoint_id(),
|
||||
ControlMessage::Ping {
|
||||
seq: 1,
|
||||
payload: Vec::new()
|
||||
}
|
||||
)
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
|
||||
// Shutting down twice is harmless.
|
||||
a.agent.shutdown().await;
|
||||
|
||||
// The peer notices and carries on.
|
||||
let status = b.agent.network_status(network_id).await.unwrap();
|
||||
assert_eq!(status.state, tsunagi::agent::NetworkState::Active);
|
||||
|
||||
b.agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn several_independent_agents_coexist_in_one_process() {
|
||||
// No global state: two completely separate rendezvous tables, two networks
|
||||
// with the same name but different secrets, four agents, one process.
|
||||
let left = SharedMemoryDiscovery::new();
|
||||
let right = SharedMemoryDiscovery::new();
|
||||
let (name, left_secret) = network("same-name-different-world");
|
||||
let (_, right_secret) = network("ignored");
|
||||
|
||||
let l1 = TestAgent::spawn(&left).await.unwrap();
|
||||
let l2 = TestAgent::spawn(&left).await.unwrap();
|
||||
let r1 = TestAgent::spawn(&right).await.unwrap();
|
||||
let r2 = TestAgent::spawn(&right).await.unwrap();
|
||||
|
||||
let left_id = l1.agent.join_network(&name, &left_secret).await.unwrap();
|
||||
l2.agent.join_network(&name, &left_secret).await.unwrap();
|
||||
let right_id = r1.agent.join_network(&name, &right_secret).await.unwrap();
|
||||
r2.agent.join_network(&name, &right_secret).await.unwrap();
|
||||
assert_ne!(left_id, right_id);
|
||||
|
||||
wait_for_peers(&l1.agent, left_id, 1).await;
|
||||
wait_for_peers(&r1.agent, right_id, 1).await;
|
||||
assert_eq!(
|
||||
l1.agent
|
||||
.network_status(left_id)
|
||||
.await
|
||||
.unwrap()
|
||||
.connected_peers(),
|
||||
vec![l2.agent.endpoint_id()]
|
||||
);
|
||||
|
||||
for agent in [l1, l2, r1, r2] {
|
||||
agent.agent.shutdown().await;
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_agent_without_discovery_still_starts_and_serves_status() {
|
||||
let dir = tempfile::TempDir::new().unwrap();
|
||||
let agent = Agent::spawn(local_config(dir.path())).await.unwrap();
|
||||
let (name, secret) = network("no-discovery");
|
||||
let network_id = agent.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
let status = agent.network_status(network_id).await.unwrap();
|
||||
assert!(status.peers.is_empty());
|
||||
assert!(status.candidates.is_empty());
|
||||
agent.recheck().await;
|
||||
assert!(agent.recheck_network(network_id).await.is_ok());
|
||||
|
||||
agent.shutdown().await;
|
||||
drop(agent);
|
||||
drop(dir);
|
||||
}
|
||||
@@ -0,0 +1,219 @@
|
||||
//! Scenarios 5 and 6: restart recovery and rotating the network secret.
|
||||
|
||||
#![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::{Agent, Error};
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_restarted_agent_keeps_its_identity_and_reconnects() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("survives-restart");
|
||||
|
||||
let peer = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let restarting = TestAgent::spawn(&discovery).await.unwrap();
|
||||
|
||||
let network_id = peer.agent.join_network(&name, &secret).await.unwrap();
|
||||
restarting.agent.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
wait_for_peers(&peer.agent, network_id, 1).await;
|
||||
let device_id = restarting.agent.endpoint_id();
|
||||
let old_sockets = restarting.agent.status().await.unwrap().bound_sockets;
|
||||
|
||||
let mut peer_events = peer.agent.subscribe();
|
||||
let dir = restarting.stop().await;
|
||||
|
||||
// The surviving peer notices the session ending.
|
||||
wait_event(&mut peer_events, |event| match event {
|
||||
Event::PeerDisconnected { peer, .. } if *peer == device_id => Some(()),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
|
||||
// Restart from the same state directory. Binding to port zero again means a
|
||||
// different local UDP port, which the refreshed discovery entry covers.
|
||||
let restarted = Agent::spawn(config_with(dir.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(restarted.endpoint_id(), device_id);
|
||||
let new_sockets = restarted.status().await.unwrap().bound_sockets;
|
||||
assert_ne!(old_sockets, new_sockets, "a fresh local port is expected");
|
||||
|
||||
// The configured network came back up on its own and both sides reconnect.
|
||||
assert!(restarted.is_active(network_id).await);
|
||||
wait_for_peers(&restarted, network_id, 1).await;
|
||||
wait_for_peers(&peer.agent, network_id, 1).await;
|
||||
|
||||
// And the restored session really works.
|
||||
restarted
|
||||
.send(
|
||||
network_id,
|
||||
peer.agent.endpoint_id(),
|
||||
ControlMessage::Ping {
|
||||
seq: 5,
|
||||
payload: b"back".to_vec(),
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let mut events = restarted.subscribe();
|
||||
wait_event(&mut events, |event| match event {
|
||||
Event::MessageReceived {
|
||||
message: ControlMessage::Pong { seq: 5, payload },
|
||||
..
|
||||
} if payload == b"back" => Some(()),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
|
||||
restarted.shutdown().await;
|
||||
peer.agent.shutdown().await;
|
||||
drop(restarted);
|
||||
drop(dir);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn joining_a_network_twice_is_not_an_error() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("idempotent-join");
|
||||
|
||||
let agent = TestAgent::spawn(&discovery).await.unwrap();
|
||||
|
||||
// Joining is declarative: saying it twice in one run must be fine.
|
||||
let first = agent.agent.join_network(&name, &secret).await.unwrap();
|
||||
let again = agent.agent.join_network(&name, &secret).await.unwrap();
|
||||
assert_eq!(first, again);
|
||||
assert_eq!(agent.agent.list_networks().await.unwrap().len(), 1);
|
||||
|
||||
// Activating explicitly is the strict version and does report it.
|
||||
assert!(matches!(
|
||||
agent.agent.activate_network(first).await,
|
||||
Err(Error::NetworkAlreadyActive(_))
|
||||
));
|
||||
|
||||
// And after a restart, where the network came back up on its own, the
|
||||
// same command must still succeed. This is what running the CLI twice
|
||||
// does.
|
||||
let dir = agent.stop().await;
|
||||
let restarted = Agent::spawn(config_with(dir.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(restarted.is_active(first).await, "auto-start brought it up");
|
||||
let rejoined = restarted.join_network(&name, &secret).await.unwrap();
|
||||
assert_eq!(rejoined, first);
|
||||
assert!(restarted.network_status(first).await.is_ok());
|
||||
|
||||
restarted.shutdown().await;
|
||||
drop(restarted);
|
||||
drop(dir);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn readiness_does_not_wait_for_anyone_else() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("lonely");
|
||||
|
||||
// No peers exist and no relay is reachable. The agent must still come up.
|
||||
let alone = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let network_id = alone.agent.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
let status = alone.agent.network_status(network_id).await.unwrap();
|
||||
assert!(status.peers.is_empty());
|
||||
assert_eq!(status.state, tsunagi::agent::NetworkState::Active);
|
||||
|
||||
alone.agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn rotating_the_secret_moves_everyone_to_a_new_space() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let name = NetworkName::new("rotate-me").unwrap();
|
||||
let old_secret = NetworkSecret::generate();
|
||||
let new_secret = NetworkSecret::generate();
|
||||
|
||||
let a = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let b = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let a_id = a.agent.endpoint_id();
|
||||
|
||||
let old = a.agent.join_network(&name, &old_secret).await.unwrap();
|
||||
b.agent.join_network(&name, &old_secret).await.unwrap();
|
||||
wait_for_peers(&a.agent, old, 1).await;
|
||||
|
||||
// Rotation through the public API: deactivate the old space, join the new.
|
||||
// No dedicated command is needed for this.
|
||||
a.agent.deactivate_network(old).await.unwrap();
|
||||
let new = a.agent.join_network(&name, &new_secret).await.unwrap();
|
||||
assert_ne!(old, new);
|
||||
assert_eq!(a.agent.endpoint_id(), a_id, "device identity is untouched");
|
||||
|
||||
// B still holds the old secret, so it must not reach the new space.
|
||||
settle().await;
|
||||
let status = a.agent.network_status(new).await.unwrap();
|
||||
assert!(
|
||||
status.peers.is_empty(),
|
||||
"the old secret must not open the new space"
|
||||
);
|
||||
assert!(
|
||||
a.agent
|
||||
.send(
|
||||
old,
|
||||
b.agent.endpoint_id(),
|
||||
ControlMessage::Ping {
|
||||
seq: 1,
|
||||
payload: Vec::new()
|
||||
}
|
||||
)
|
||||
.await
|
||||
.is_err(),
|
||||
"the deactivated network cannot be used any more"
|
||||
);
|
||||
|
||||
// Once B rotates too, they meet again in the new space.
|
||||
b.agent.deactivate_network(old).await.unwrap();
|
||||
let b_new = b.agent.join_network(&name, &new_secret).await.unwrap();
|
||||
assert_eq!(b_new, new);
|
||||
wait_for_peers(&a.agent, new, 1).await;
|
||||
|
||||
a.agent.shutdown().await;
|
||||
b.agent.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_rotated_out_network_does_not_come_back_after_a_restart() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let name = NetworkName::new("no-resurrection").unwrap();
|
||||
|
||||
let agent = TestAgent::spawn(&discovery).await.unwrap();
|
||||
let old = agent
|
||||
.agent
|
||||
.join_network(&name, &NetworkSecret::generate())
|
||||
.await
|
||||
.unwrap();
|
||||
let new = agent
|
||||
.agent
|
||||
.join_network(&name, &NetworkSecret::generate())
|
||||
.await
|
||||
.unwrap();
|
||||
agent.agent.deactivate_network(old).await.unwrap();
|
||||
let dir = agent.stop().await;
|
||||
|
||||
let restarted = Agent::spawn(config_with(dir.path(), &discovery))
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(!restarted.is_active(old).await);
|
||||
assert!(restarted.is_active(new).await);
|
||||
assert!(matches!(
|
||||
restarted.network_status(old).await,
|
||||
Err(Error::NetworkNotActive(_))
|
||||
));
|
||||
|
||||
restarted.shutdown().await;
|
||||
drop(restarted);
|
||||
drop(dir);
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user