Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
97377e3703 |
No files matched your search
@@ -59,8 +59,10 @@ Keep these separate. Crossing them is the main thing to review for.
|
||||
other peers goes link in, link out: it is never written to the middle's
|
||||
interface or decrypted there. Multihop routing has a bounded hop limit;
|
||||
equal-cost next hops are chosen per flow. Reachability is first-hand and
|
||||
volatile: each authenticated member advertises its own protocol-specific
|
||||
links. Build routing tables on topology changes, never per packet. Transit
|
||||
volatile: each authenticated member advertises its own direct links, of
|
||||
whatever kind. A link is a way to reach a neighbour and a protocol is an
|
||||
end-to-end tunnel to a peer; the two meet only in the envelope, so a
|
||||
tunnel never depends on which transports carry it. Build routing tables on topology changes, never per packet. Transit
|
||||
must not acquire a routing mutex or wait for a protocol/TUN reader.
|
||||
- **Broadcast fanout belongs at local IP ingress.** Participation is local to
|
||||
each network, enabled by default, persisted and advertised to authenticated
|
||||
|
||||
@@ -49,9 +49,9 @@ with relay fallback), and keep the network state among themselves.
|
||||
|
||||
- **Two things to know:** a network name and a secret. A secret is generated for you when you make a network.
|
||||
- **Direct, encrypted links** between members (WireGuard cryptography), with relay fallback behind strict NATs.
|
||||
- **Names for machines:** `server.home` just works, even while `server` is switched off.
|
||||
- **Names for machines:** `server.home` just works, even while `server` is switched off. A name belongs to whoever claimed it first; a later device with the same name keeps working by address and is told so.
|
||||
- **Exit nodes:** send all your internet traffic through any member that offers one.
|
||||
- **Multi-hop:** members that can't reach each other directly are routed through a member that can.
|
||||
- **Multi-hop:** members that can't reach each other directly are routed through a member that can, over whatever kind of link each hop has. A member reached that way is online, and shown as relayed; traffic moves to a direct link by itself the moment one comes up.
|
||||
- **LAN games:** UDP broadcasts are relayed, so "LAN" lobbies show up across the mesh.
|
||||
- **Several networks at once** on one device.
|
||||
- **Tray app** for Windows, macOS and Linux; a command line for everything.
|
||||
|
||||
@@ -129,6 +129,19 @@ pub enum Event {
|
||||
/// Why it is not up.
|
||||
reason: String,
|
||||
},
|
||||
/// The name this agent claimed was claimed first by another member.
|
||||
///
|
||||
/// Nothing stops working: this agent keeps its address and every
|
||||
/// connection. It just is not found by that name until it is renamed or
|
||||
/// the member that holds the name lets it go.
|
||||
HostnameTaken {
|
||||
/// The network.
|
||||
network: NetworkId,
|
||||
/// The name that is taken.
|
||||
name: String,
|
||||
/// The member that holds it.
|
||||
holder: EndpointId,
|
||||
},
|
||||
/// An IP plugin reported an error. Never fatal.
|
||||
PluginError {
|
||||
/// The network the call was scoped to.
|
||||
|
||||
@@ -26,8 +26,8 @@ mod status;
|
||||
|
||||
pub use events::Event;
|
||||
pub use status::{
|
||||
AgentStatus, CandidateStatus, MemberStatus, NetworkMetrics, NetworkState, NetworkStatus,
|
||||
OverlayStatus, PeerStatus,
|
||||
AgentStatus, CandidateStatus, DataPath, HostnameConflict, MemberStatus, NetworkMetrics,
|
||||
NetworkState, NetworkStatus, OverlayStatus, PeerStatus,
|
||||
};
|
||||
|
||||
use std::collections::HashMap;
|
||||
@@ -981,6 +981,7 @@ impl Agent {
|
||||
// An inactive network has no runtime to ask; the roster and
|
||||
// the range come from one. Empty, not invented.
|
||||
members: Vec::new(),
|
||||
hostname_conflict: None,
|
||||
range: None,
|
||||
range_conflict: None,
|
||||
metrics: NetworkMetrics::default(),
|
||||
|
||||
@@ -32,7 +32,8 @@ use super::events::Event;
|
||||
use super::session::{self, Session, SessionEvent};
|
||||
use super::shutdown::Shutdown;
|
||||
use super::status::{
|
||||
CandidateStatus, MemberStatus, NetworkMetrics, NetworkState, NetworkStatus, PeerStatus,
|
||||
CandidateStatus, DataPath, HostnameConflict, MemberStatus, NetworkMetrics, NetworkState,
|
||||
NetworkStatus, PeerStatus,
|
||||
};
|
||||
|
||||
/// An inbound connection that already passed the handshake.
|
||||
@@ -320,6 +321,10 @@ struct Runtime {
|
||||
std::time::Instant,
|
||||
),
|
||||
>,
|
||||
/// The protocols this agent last announced, in the order it announced
|
||||
/// them. The order is its preference, and the lower of two ids settles
|
||||
/// which protocol a pair runs.
|
||||
local_capabilities: Vec<PluginCapability>,
|
||||
/// What this agent last told its peers it could reach, so it is only
|
||||
/// said again when it changed.
|
||||
announced_reach: HashSet<crate::proto::message::ReachableLink>,
|
||||
@@ -338,6 +343,9 @@ struct Runtime {
|
||||
/// publish a claim afterwards — the periodic check, a peer's records
|
||||
/// arriving — would silently take it back.
|
||||
released: bool,
|
||||
/// The name this agent last said was taken, so it is said once per
|
||||
/// change rather than on every round.
|
||||
reported_name_taken: Option<HostnameConflict>,
|
||||
/// The address last reported as absent from the interface, so it is said
|
||||
/// once rather than for ever.
|
||||
reported_missing: Option<std::net::Ipv4Addr>,
|
||||
@@ -366,6 +374,16 @@ impl Runtime {
|
||||
let (link_results_tx, link_results_rx) = mpsc::channel(64);
|
||||
let (candidate_tx, discovery_candidates) =
|
||||
mpsc::channel(params.limits.max_discovery_candidates.max(1));
|
||||
let hub = crate::dataplane::relay::RelayHub::new(network_id, local_id);
|
||||
// The order the protocols were configured in is also how links are
|
||||
// preferred when a neighbour can be reached over more than one.
|
||||
hub.set_priority(
|
||||
params
|
||||
.plugins
|
||||
.iter()
|
||||
.map(|plugin| plugin.protocol_id().to_string())
|
||||
.collect(),
|
||||
);
|
||||
let discovery_worker = params.discovery.as_ref().map(|backend| {
|
||||
DiscoveryWorker::spawn(
|
||||
backend.clone(),
|
||||
@@ -393,8 +411,9 @@ impl Runtime {
|
||||
dial_results_tx,
|
||||
dial_results_rx,
|
||||
links: HashMap::new(),
|
||||
hub: crate::dataplane::relay::RelayHub::new(network_id, local_id),
|
||||
hub,
|
||||
reachable: HashMap::new(),
|
||||
local_capabilities: Vec::new(),
|
||||
announced_reach: HashSet::new(),
|
||||
opening: HashSet::new(),
|
||||
link_results_tx,
|
||||
@@ -403,6 +422,7 @@ impl Runtime {
|
||||
reported_mismatch: HashSet::new(),
|
||||
released: false,
|
||||
reported_missing: None,
|
||||
reported_name_taken: None,
|
||||
state: StateSet::new(),
|
||||
pending_state: Vec::new(),
|
||||
own_version: 0,
|
||||
@@ -617,6 +637,8 @@ impl Runtime {
|
||||
tracing::debug!(%err, "could not queue re-announcement");
|
||||
}
|
||||
}
|
||||
// What was announced decides which protocol each pair runs.
|
||||
self.update_paths();
|
||||
}
|
||||
|
||||
/// Queues a message without blocking the runtime loop.
|
||||
@@ -669,6 +691,7 @@ impl Runtime {
|
||||
// alone, so something has to look again; this runs on a timer and
|
||||
// the check is a comparison when nothing has changed.
|
||||
self.ensure_own_claim().await;
|
||||
self.report_hostname_taken();
|
||||
// The same timer repairs the relay picture: reachability is soft
|
||||
// state, repeated while it holds and forgotten when it stops.
|
||||
self.update_paths();
|
||||
@@ -1031,12 +1054,25 @@ impl Runtime {
|
||||
}
|
||||
};
|
||||
|
||||
// A name keeps the time it was first claimed for as long as this
|
||||
// member keeps the name, so a later claimant cannot overtake it by
|
||||
// the passage of time, and a new or changed name starts the clock
|
||||
// afresh.
|
||||
let hostname_since = wanted_hostname.as_ref().map(|name| {
|
||||
let kept = self
|
||||
.state
|
||||
.get(&self.local_id)
|
||||
.filter(|record| record.body.hostname() == Some(name.as_str()))
|
||||
.and_then(|record| record.body.hostname_since());
|
||||
kept.unwrap_or_else(unix_millis)
|
||||
});
|
||||
let body = RecordBody::Claim {
|
||||
address: wanted_address,
|
||||
// The range only travels with an address, so a member of an
|
||||
// IPv6-only network does not assert one.
|
||||
range: wanted_address.and(range),
|
||||
hostname: wanted_hostname,
|
||||
hostname_since,
|
||||
};
|
||||
|
||||
// Nothing to say is not the same as saying nothing changed: a member
|
||||
@@ -1100,6 +1136,7 @@ impl Runtime {
|
||||
for (peer, records) in std::mem::take(&mut self.pending_state) {
|
||||
self.receive_state(peer, records).await;
|
||||
}
|
||||
self.report_hostname_taken();
|
||||
}
|
||||
|
||||
/// Sends everything we know to every peer.
|
||||
@@ -1658,68 +1695,93 @@ impl Runtime {
|
||||
}
|
||||
}
|
||||
|
||||
/// The protocol this agent and a member run their tunnel in, if they have
|
||||
/// one in common.
|
||||
///
|
||||
/// Both ends must name the same one without asking each other, and what
|
||||
/// each side was configured to prefer can differ. So the member with the
|
||||
/// lower id decides, by the order it announces its protocols in, and the
|
||||
/// other side reads that order from the announcement it already has.
|
||||
fn tunnel_protocol(&self, session: &Session) -> Option<String> {
|
||||
let usable = |ours: &[PluginCapability], theirs: &[PluginCapability], name: &str| {
|
||||
ours.iter().any(|capability| {
|
||||
capability.enabled
|
||||
&& capability.protocol == name
|
||||
&& theirs.iter().any(|other| {
|
||||
other.enabled
|
||||
&& other.protocol == name
|
||||
&& other.version == capability.version
|
||||
})
|
||||
})
|
||||
};
|
||||
let (decides, other) = if self.local_id.as_bytes() < session.peer.as_bytes() {
|
||||
(&self.local_capabilities, &session.capabilities)
|
||||
} else {
|
||||
(&session.capabilities, &self.local_capabilities)
|
||||
};
|
||||
decides
|
||||
.iter()
|
||||
.find(|capability| usable(decides, other, &capability.protocol))
|
||||
.map(|capability| capability.protocol.clone())
|
||||
}
|
||||
|
||||
/// Builds the graph off the packet path. Every remote row comes from that
|
||||
/// member's authenticated control session; stale or incompatible rows never
|
||||
/// enter the graph. The hub replaces an immutable table only on changes.
|
||||
/// member's authenticated control session, and a row says only who that
|
||||
/// member has a link to, never what kind of link: a member in the middle
|
||||
/// needs no protocol in common with either end. Stale rows never enter the
|
||||
/// graph. The hub replaces an immutable table only on changes.
|
||||
fn update_paths(&mut self) {
|
||||
self.reachable.retain(|peer, (_, heard)| {
|
||||
heard.elapsed() < REACH_EXPIRY && self.sessions.contains_key(peer)
|
||||
});
|
||||
for (protocol, version) in self.served_protocols() {
|
||||
let peers: Vec<_> = self
|
||||
.sessions
|
||||
.values()
|
||||
.filter(|session| {
|
||||
session.capabilities.iter().any(|capability| {
|
||||
capability.enabled
|
||||
&& capability.protocol == protocol
|
||||
&& capability.version == version
|
||||
})
|
||||
})
|
||||
.map(|session| session.peer)
|
||||
.collect();
|
||||
let mut members: HashSet<_> = peers.iter().map(|peer| *peer.as_bytes()).collect();
|
||||
members.insert(*self.local_id.as_bytes());
|
||||
let graph = self
|
||||
.reachable
|
||||
.iter()
|
||||
.filter(|(peer, _)| members.contains(peer.as_bytes()))
|
||||
.map(|(peer, (links, _))| {
|
||||
let mut neighbors: Vec<_> = links
|
||||
.iter()
|
||||
.filter(|link| {
|
||||
link.protocol == protocol
|
||||
&& members.contains(&link.peer)
|
||||
&& link.peer != *peer.as_bytes()
|
||||
})
|
||||
.map(|link| link.peer)
|
||||
.collect();
|
||||
neighbors.sort_unstable();
|
||||
(*peer.as_bytes(), neighbors)
|
||||
})
|
||||
.collect();
|
||||
self.hub.set_topology(&protocol, graph, members);
|
||||
for peer in peers {
|
||||
if self.hub.has_link(peer, &protocol) || !self.hub.reachable(peer, &protocol) {
|
||||
continue;
|
||||
}
|
||||
let Some(plugin) = self
|
||||
.params
|
||||
.plugins
|
||||
let mut members: HashSet<_> = self.sessions.keys().map(|peer| *peer.as_bytes()).collect();
|
||||
members.insert(*self.local_id.as_bytes());
|
||||
let graph = self
|
||||
.reachable
|
||||
.iter()
|
||||
.filter(|(peer, _)| members.contains(peer.as_bytes()))
|
||||
.map(|(peer, (links, _))| {
|
||||
let mut neighbors: Vec<_> = links
|
||||
.iter()
|
||||
.find(|plugin| plugin.protocol_id() == protocol)
|
||||
.cloned()
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
let link = self.hub.link(peer, &protocol);
|
||||
let path = link.path_description();
|
||||
let max_datagram = link.max_datagram_size();
|
||||
plugin.on_peer_link(self.network_id, peer, link);
|
||||
.filter(|link| members.contains(&link.peer) && link.peer != *peer.as_bytes())
|
||||
.map(|link| link.peer)
|
||||
.collect();
|
||||
neighbors.sort_unstable();
|
||||
neighbors.dedup();
|
||||
(*peer.as_bytes(), neighbors)
|
||||
})
|
||||
.collect();
|
||||
self.hub.set_topology(graph, members);
|
||||
|
||||
let wanted: Vec<(EndpointId, String)> = self
|
||||
.sessions
|
||||
.values()
|
||||
.filter_map(|session| Some((session.peer, self.tunnel_protocol(session)?)))
|
||||
.collect();
|
||||
for (peer, protocol) in wanted {
|
||||
if self.hub.has_link(peer, &protocol) || !self.hub.reachable(peer) {
|
||||
continue;
|
||||
}
|
||||
let Some(plugin) = self
|
||||
.params
|
||||
.plugins
|
||||
.iter()
|
||||
.find(|plugin| plugin.protocol_id() == protocol)
|
||||
.cloned()
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
let link = self.hub.link(peer, &protocol);
|
||||
let relayed = self.hub.route(peer).is_some_and(|route| route.hops > 1);
|
||||
let path = link.path_description();
|
||||
let max_datagram = link.max_datagram_size();
|
||||
plugin.on_peer_link(self.network_id, peer, link);
|
||||
// A direct link has already been reported as one.
|
||||
if relayed {
|
||||
self.emit(Event::DataLinkUp {
|
||||
network: self.network_id,
|
||||
peer,
|
||||
protocol: protocol.clone(),
|
||||
protocol,
|
||||
path,
|
||||
max_datagram,
|
||||
});
|
||||
@@ -1786,30 +1848,25 @@ impl Runtime {
|
||||
|
||||
/// Hands a link to the plugin that owns its protocol.
|
||||
fn adopt_link(&mut self, peer: EndpointId, protocol: String, link: SharedLink) {
|
||||
let Some(plugin) = self
|
||||
if !self
|
||||
.params
|
||||
.plugins
|
||||
.iter()
|
||||
.find(|plugin| plugin.protocol_id() == protocol)
|
||||
.cloned()
|
||||
else {
|
||||
.any(|plugin| plugin.protocol_id() == protocol)
|
||||
{
|
||||
return;
|
||||
};
|
||||
}
|
||||
|
||||
let path = link.path_description();
|
||||
let max_datagram = link.max_datagram_size();
|
||||
self.links
|
||||
.insert((peer, protocol.clone()), Arc::clone(&link));
|
||||
// The protocol is handed the hub's link, not this one: the same
|
||||
// object for as long as the peer is a peer, whatever happens to
|
||||
// the path under it. A tunnel is not torn down because a direct
|
||||
// link came or went.
|
||||
let first_time = !self.hub.has_link(peer, &protocol);
|
||||
// What this is, is a way to reach the peer. Which protocol's
|
||||
// tunnel runs over it is decided in `update_paths`, and that tunnel
|
||||
// is handed the hub's link, not this one: the same object for as
|
||||
// long as the peer is a peer, whatever happens to the path under
|
||||
// it. A tunnel is not torn down because a direct link came or went.
|
||||
self.hub.set_direct(peer, &protocol, link);
|
||||
let shared = self.hub.link(peer, &protocol);
|
||||
let max_datagram = shared.max_datagram_size();
|
||||
if first_time {
|
||||
plugin.on_peer_link(self.network_id, peer, shared);
|
||||
}
|
||||
self.metrics.data_links_established += 1;
|
||||
self.emit(Event::DataLinkUp {
|
||||
network: self.network_id,
|
||||
@@ -1941,6 +1998,13 @@ impl Runtime {
|
||||
});
|
||||
}
|
||||
capabilities.truncate(self.params.limits.max_capabilities);
|
||||
self.local_capabilities = capabilities
|
||||
.iter()
|
||||
.map(|capability| PluginCapability {
|
||||
data: Vec::new(),
|
||||
..capability.clone()
|
||||
})
|
||||
.collect();
|
||||
let exit_node = self.exit_ready();
|
||||
self.announced_exit = exit_node;
|
||||
Announcement {
|
||||
@@ -2175,6 +2239,69 @@ impl Runtime {
|
||||
|
||||
// ----------------------------------------------------------------- status
|
||||
|
||||
/// Whether a member can be reached now, by any path.
|
||||
///
|
||||
/// The single definition every client reports: a live control session,
|
||||
/// or a data path to it through other members. Being reached over a long
|
||||
/// path is being online; how well is `data_path`'s to say.
|
||||
fn is_online(&self, peer: EndpointId) -> bool {
|
||||
peer == self.local_id || self.sessions.contains_key(&peer) || self.hub.reachable(peer)
|
||||
}
|
||||
|
||||
/// How packets reach a member, which says nothing about whether it is
|
||||
/// there.
|
||||
fn data_path(&self, peer: EndpointId) -> DataPath {
|
||||
match self.hub.route(peer) {
|
||||
None => DataPath::None,
|
||||
Some(route) if route.hops <= 1 => DataPath::Direct {
|
||||
transport: self
|
||||
.hub
|
||||
.direct_kinds(peer)
|
||||
.into_iter()
|
||||
.next()
|
||||
.unwrap_or_default(),
|
||||
},
|
||||
Some(route) => DataPath::Relayed {
|
||||
hops: route.hops,
|
||||
via: route.via,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/// Says so when this agent's own name is held by someone else, and
|
||||
/// again if that changes. Never anything but a statement: the agent keeps
|
||||
/// working by address either way.
|
||||
fn report_hostname_taken(&mut self) {
|
||||
let taken = self.hostname_conflict_of(&self.local_id);
|
||||
if taken == self.reported_name_taken {
|
||||
return;
|
||||
}
|
||||
if let Some(conflict) = &taken {
|
||||
tracing::warn!(
|
||||
network = %self.network_id,
|
||||
name = %conflict.name,
|
||||
holder = %conflict.holder.fmt_short(),
|
||||
"this name was claimed first by another member; it stays reachable by \
|
||||
address but is not found by this name until it is renamed"
|
||||
);
|
||||
self.emit(Event::HostnameTaken {
|
||||
network: self.network_id,
|
||||
name: conflict.name.clone(),
|
||||
holder: conflict.holder,
|
||||
});
|
||||
}
|
||||
self.reported_name_taken = taken;
|
||||
}
|
||||
|
||||
fn hostname_conflict_of(&self, author: &EndpointId) -> Option<HostnameConflict> {
|
||||
self.state
|
||||
.hostname_taken(author)
|
||||
.map(|(name, holder)| HostnameConflict {
|
||||
name: name.to_string(),
|
||||
holder,
|
||||
})
|
||||
}
|
||||
|
||||
fn status(&self) -> NetworkStatus {
|
||||
let served = self.served_protocols();
|
||||
let mut peers: Vec<PeerStatus> = self
|
||||
@@ -2212,6 +2339,7 @@ impl Runtime {
|
||||
control_messages_received: session.messages_received,
|
||||
control_bytes_sent: session.bytes_sent,
|
||||
control_bytes_received: session.bytes_received,
|
||||
data_path: self.data_path(session.peer),
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
@@ -2249,6 +2377,9 @@ impl Runtime {
|
||||
overlay_address_v4: self.state.address_of(&endpoint_id),
|
||||
hostname: self.state.hostname_of(&endpoint_id).map(str::to_string),
|
||||
last_hostname: self.known_hostnames.get(&endpoint_id).cloned(),
|
||||
hostname_conflict: self.hostname_conflict_of(&endpoint_id),
|
||||
online: self.is_online(endpoint_id),
|
||||
data_path: self.data_path(endpoint_id),
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
@@ -2276,6 +2407,7 @@ impl Runtime {
|
||||
peers,
|
||||
candidates,
|
||||
members,
|
||||
hostname_conflict: self.hostname_conflict_of(&self.local_id),
|
||||
range: self.effective_range(),
|
||||
// Only worth reporting while it is actually stuck: with a
|
||||
// fallback in play the configured range being another
|
||||
@@ -2290,6 +2422,15 @@ impl Runtime {
|
||||
}
|
||||
}
|
||||
|
||||
/// Milliseconds since the Unix epoch, for the time a name was first claimed.
|
||||
fn unix_millis() -> u64 {
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map_or(u64::MAX, |elapsed| {
|
||||
u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX)
|
||||
})
|
||||
}
|
||||
|
||||
/// Which side dialled, given a role and the two identities.
|
||||
fn initiator_of(role: Role, local: EndpointId, peer: EndpointId) -> EndpointId {
|
||||
match role {
|
||||
|
||||
@@ -59,6 +59,43 @@ pub struct OverlayStatus {
|
||||
pub broadcast_rules: Option<crate::overlay::BroadcastRulesReport>,
|
||||
}
|
||||
|
||||
/// How packets reach a member right now.
|
||||
///
|
||||
/// Being reachable is one thing and being reachable well is another, and
|
||||
/// the second is never a reason to call a member away: a member that can only
|
||||
/// be reached through somebody else is online, over a longer path.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum DataPath {
|
||||
/// Nothing carries packets to it yet. The member is connected for
|
||||
/// control only, which is the state of a data path being set up.
|
||||
None,
|
||||
/// A direct link, of the named kind of transport.
|
||||
Direct {
|
||||
/// The kind of transport carrying it, e.g. `wg`.
|
||||
transport: String,
|
||||
},
|
||||
/// Through other members.
|
||||
Relayed {
|
||||
/// Number of links on the way, at least two.
|
||||
hops: u8,
|
||||
/// The member the first of them goes to.
|
||||
via: EndpointId,
|
||||
},
|
||||
}
|
||||
|
||||
/// A name that a member claimed and somebody else holds.
|
||||
///
|
||||
/// A name belongs to whoever claimed it first. The member that claimed it
|
||||
/// later is still a member and still reachable by its address; it is just
|
||||
/// not found by that name, on any member of the network.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct HostnameConflict {
|
||||
/// The name that is taken.
|
||||
pub name: String,
|
||||
/// The member that holds it.
|
||||
pub holder: EndpointId,
|
||||
}
|
||||
|
||||
/// A member the signed state knows about, connected or not.
|
||||
///
|
||||
/// This is the durable roster: it comes from signed records, so a member that
|
||||
@@ -85,6 +122,15 @@ pub struct MemberStatus {
|
||||
/// Not signed, so it is a convenience for naming a member that is away
|
||||
/// and never a claim: the zone is served from `hostname`.
|
||||
pub last_hostname: Option<String>,
|
||||
/// The name it claimed, when somebody else claimed it first.
|
||||
pub hostname_conflict: Option<HostnameConflict>,
|
||||
/// Whether this agent can reach it now, by any path: a live control
|
||||
/// session, or a data path through other members. The one answer to "is
|
||||
/// it online" for every client, so that none of them has to work it out
|
||||
/// from a different part of the picture. This agent counts as online.
|
||||
pub online: bool,
|
||||
/// How packets reach it. [`DataPath::None`] while it is offline.
|
||||
pub data_path: DataPath,
|
||||
}
|
||||
|
||||
/// Status of one authenticated session.
|
||||
@@ -128,6 +174,8 @@ pub struct PeerStatus {
|
||||
pub control_bytes_sent: u64,
|
||||
/// Control payload bytes read from this session.
|
||||
pub control_bytes_received: u64,
|
||||
/// How packets reach this peer right now.
|
||||
pub data_path: DataPath,
|
||||
}
|
||||
|
||||
/// Counters scoped to one logical network.
|
||||
@@ -188,6 +236,10 @@ pub struct NetworkStatus {
|
||||
pub candidates: Vec<CandidateStatus>,
|
||||
/// Members the signed state knows about, whether connected or not.
|
||||
pub members: Vec<MemberStatus>,
|
||||
/// This agent's own name, when another member claimed it first and so
|
||||
/// holds it. This agent keeps working by address; nobody finds it by
|
||||
/// this name until it is renamed or the other member lets it go.
|
||||
pub hostname_conflict: Option<HostnameConflict>,
|
||||
/// The overlay range this network uses, once it has one.
|
||||
pub range: Option<crate::state::Ipv4Range>,
|
||||
/// The range it could not have, because another network on this agent
|
||||
|
||||
@@ -67,10 +67,41 @@ impl Tally {
|
||||
}
|
||||
}
|
||||
|
||||
/// The links to one neighbour, best first. Which transport carries a frame
|
||||
/// is decided here and nowhere else: the end-to-end protocol is in the
|
||||
/// envelope and never depends on it.
|
||||
#[derive(Debug)]
|
||||
struct NextHop {
|
||||
peer: PeerId,
|
||||
link: SharedLink,
|
||||
links: Box<[SharedLink]>,
|
||||
}
|
||||
|
||||
impl NextHop {
|
||||
fn usable(&self) -> bool {
|
||||
self.links.iter().any(|link| !link.is_closed())
|
||||
}
|
||||
|
||||
fn first_open(&self) -> Option<&SharedLink> {
|
||||
self.links.iter().find(|link| !link.is_closed())
|
||||
}
|
||||
|
||||
/// Hands a frame to the best open link that accepts it. A link that
|
||||
/// cannot carry the frame (too large for its path, closed a moment ago)
|
||||
/// leaves it to the next one; the frame is only copied when there is
|
||||
/// another to try.
|
||||
fn send(&self, frame: Bytes) -> Result<(), TransportError> {
|
||||
let mut open = self.links.iter().filter(|link| !link.is_closed());
|
||||
let Some(mut current) = open.next() else {
|
||||
return Err(TransportError::Closed);
|
||||
};
|
||||
for next in open {
|
||||
if current.send(frame.clone()).is_ok() {
|
||||
return Ok(());
|
||||
}
|
||||
current = next;
|
||||
}
|
||||
current.send(frame)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
@@ -82,21 +113,22 @@ struct ForwardRoute {
|
||||
impl ForwardRoute {
|
||||
fn select(&self, source: &PeerId, flow: FlowId) -> Option<&NextHop> {
|
||||
if self.next.len() == 1 {
|
||||
return self.next.first().filter(|hop| !hop.link.is_closed());
|
||||
return self.next.first().filter(|hop| hop.usable());
|
||||
}
|
||||
let seed = u64::from_le_bytes(source[..8].try_into().ok()?);
|
||||
let start = mix64(flow ^ seed) as usize % self.next.len();
|
||||
// Stable fallback if a link closed just before its table was replaced.
|
||||
(0..self.next.len())
|
||||
.map(|i| &self.next[(start + i) % self.next.len()])
|
||||
.find(|hop| !hop.link.is_closed())
|
||||
.find(|hop| hop.usable())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Default)]
|
||||
struct ForwardingTable {
|
||||
routes: HashMap<PeerId, ForwardRoute>,
|
||||
inboxes: HashMap<PeerId, mpsc::Sender<Bytes>>,
|
||||
/// Local delivery, by who sent it and which end-to-end protocol it is.
|
||||
inboxes: HashMap<(PeerId, u64), mpsc::Sender<Bytes>>,
|
||||
members: HashSet<PeerId>,
|
||||
}
|
||||
|
||||
@@ -121,7 +153,7 @@ impl Plane {
|
||||
return;
|
||||
}
|
||||
if header.destination == self.local {
|
||||
let Some(inbox) = table.inboxes.get(&header.source) else {
|
||||
let Some(inbox) = table.inboxes.get(&(header.source, header.protocol)) else {
|
||||
self.tally.dropped_unknown.fetch_add(1, Ordering::Relaxed);
|
||||
return;
|
||||
};
|
||||
@@ -144,7 +176,7 @@ impl Plane {
|
||||
self.tally.dropped_no_link.fetch_add(1, Ordering::Relaxed);
|
||||
return;
|
||||
};
|
||||
if hop.link.send(envelope::decrement(frame)).is_ok() {
|
||||
if hop.send(envelope::decrement(frame)).is_ok() {
|
||||
self.tally.forwarded.fetch_add(1, Ordering::Relaxed);
|
||||
} else {
|
||||
self.tally.dropped_no_link.fetch_add(1, Ordering::Relaxed);
|
||||
@@ -152,27 +184,65 @@ impl Plane {
|
||||
}
|
||||
}
|
||||
|
||||
/// One physical link to a neighbour, of whatever kind of transport.
|
||||
#[derive(Debug)]
|
||||
struct Protocol {
|
||||
plane: Arc<Plane>,
|
||||
graph: HashMap<PeerId, Vec<PeerId>>,
|
||||
members: HashSet<PeerId>,
|
||||
raw: HashMap<PeerId, SharedLink>,
|
||||
readers: HashMap<PeerId, JoinHandle<()>>,
|
||||
links: HashMap<PeerId, Arc<PeerLink>>,
|
||||
struct RawLink {
|
||||
kind: String,
|
||||
link: SharedLink,
|
||||
reader: JoinHandle<()>,
|
||||
}
|
||||
|
||||
impl Protocol {
|
||||
/// How a destination is reached right now.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub struct RouteSummary {
|
||||
/// Number of transport links to it. One is a direct link.
|
||||
pub hops: u8,
|
||||
/// The neighbour the first link goes to. For a direct link that is the
|
||||
/// destination itself.
|
||||
pub via: EndpointId,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct State {
|
||||
plane: Arc<Plane>,
|
||||
/// Transport kinds, best first. A neighbour with several links is
|
||||
/// reached over the first of these it has.
|
||||
priority: Vec<String>,
|
||||
graph: HashMap<PeerId, Vec<PeerId>>,
|
||||
members: HashSet<PeerId>,
|
||||
raw: HashMap<PeerId, Vec<RawLink>>,
|
||||
links: HashMap<(PeerId, String), Arc<PeerLink>>,
|
||||
}
|
||||
|
||||
impl State {
|
||||
fn rank(&self, kind: &str) -> usize {
|
||||
self.priority
|
||||
.iter()
|
||||
.position(|known| known == kind)
|
||||
.unwrap_or(usize::MAX)
|
||||
}
|
||||
|
||||
/// A neighbour's links, best kind first.
|
||||
fn ordered(&self, peer: &PeerId) -> Vec<&RawLink> {
|
||||
let mut links: Vec<&RawLink> = self
|
||||
.raw
|
||||
.get(peer)
|
||||
.map(|links| links.iter().collect())
|
||||
.unwrap_or_default();
|
||||
links.sort_by_key(|raw| (self.rank(&raw.kind), raw.kind.clone()));
|
||||
links
|
||||
}
|
||||
|
||||
fn publish(&self) {
|
||||
let mut graph = self.graph.clone();
|
||||
graph.insert(
|
||||
self.plane.local,
|
||||
self.raw
|
||||
.iter()
|
||||
.filter(|(_, link)| !link.is_closed())
|
||||
.map(|(peer, _)| *peer)
|
||||
.collect(),
|
||||
);
|
||||
let mut neighbours: Vec<PeerId> = self
|
||||
.raw
|
||||
.iter()
|
||||
.filter(|(_, links)| links.iter().any(|raw| !raw.link.is_closed()))
|
||||
.map(|(peer, _)| *peer)
|
||||
.collect();
|
||||
neighbours.sort_unstable();
|
||||
graph.insert(self.plane.local, neighbours);
|
||||
let routing = RoutingTable::build(self.plane.local, &graph, ROUTING_HOP_LIMIT);
|
||||
let routes = routing
|
||||
.iter()
|
||||
@@ -181,10 +251,12 @@ impl Protocol {
|
||||
.next_hops
|
||||
.iter()
|
||||
.filter_map(|id| {
|
||||
self.raw.get(id).map(|link| NextHop {
|
||||
peer: *id,
|
||||
link: link.clone(),
|
||||
})
|
||||
let links: Box<[_]> = self
|
||||
.ordered(id)
|
||||
.into_iter()
|
||||
.map(|raw| raw.link.clone())
|
||||
.collect();
|
||||
(!links.is_empty()).then_some(NextHop { peer: *id, links })
|
||||
})
|
||||
.collect();
|
||||
(!next.is_empty()).then_some((
|
||||
@@ -204,17 +276,21 @@ impl Protocol {
|
||||
inboxes: self
|
||||
.links
|
||||
.iter()
|
||||
.map(|(peer, link)| (*peer, link.inbox_tx.clone()))
|
||||
.map(|((peer, _), link)| ((*peer, link.tag), link.inbox_tx.clone()))
|
||||
.collect(),
|
||||
}));
|
||||
}
|
||||
|
||||
fn stop_readers(&mut self) {
|
||||
for raw in self.raw.values().flatten() {
|
||||
raw.reader.abort();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Protocol {
|
||||
impl Drop for State {
|
||||
fn drop(&mut self) {
|
||||
for task in self.readers.values() {
|
||||
task.abort();
|
||||
}
|
||||
self.stop_readers();
|
||||
for link in self.links.values() {
|
||||
link.close();
|
||||
}
|
||||
@@ -222,12 +298,17 @@ impl Drop for Protocol {
|
||||
}
|
||||
}
|
||||
|
||||
/// One network's transport links and independently published protocol tables.
|
||||
/// One network's transport links and its published forwarding table.
|
||||
///
|
||||
/// There is one table for the whole network, whatever protocols run over it.
|
||||
/// A link is a way to reach a neighbour; a protocol is an end-to-end channel
|
||||
/// to a peer. The two meet only in the envelope, which names the protocol, so
|
||||
/// a tunnel keeps working over any mix of transports and a middle member
|
||||
/// needs no protocol in common with either end.
|
||||
#[derive(Debug)]
|
||||
pub struct RelayHub {
|
||||
network: NetworkId,
|
||||
local: PeerId,
|
||||
protocols: Mutex<HashMap<String, Protocol>>,
|
||||
state: Mutex<State>,
|
||||
tally: Arc<Tally>,
|
||||
changed: Arc<Notify>,
|
||||
}
|
||||
@@ -235,40 +316,38 @@ pub struct RelayHub {
|
||||
impl RelayHub {
|
||||
/// Creates the router at this network's local endpoint.
|
||||
pub fn new(network: NetworkId, local: EndpointId) -> Arc<Self> {
|
||||
let tally = Arc::new(Tally::default());
|
||||
Arc::new(Self {
|
||||
network,
|
||||
local: *local.as_bytes(),
|
||||
protocols: Mutex::default(),
|
||||
tally: Arc::default(),
|
||||
state: Mutex::new(State {
|
||||
plane: Arc::new(Plane {
|
||||
local: *local.as_bytes(),
|
||||
table: ArcSwap::from_pointee(ForwardingTable::default()),
|
||||
tally: tally.clone(),
|
||||
}),
|
||||
priority: Vec::new(),
|
||||
graph: HashMap::new(),
|
||||
members: HashSet::new(),
|
||||
raw: HashMap::new(),
|
||||
links: HashMap::new(),
|
||||
}),
|
||||
tally,
|
||||
changed: Arc::default(),
|
||||
})
|
||||
}
|
||||
|
||||
fn protocols(&self) -> std::sync::MutexGuard<'_, HashMap<String, Protocol>> {
|
||||
self.protocols
|
||||
.lock()
|
||||
.unwrap_or_else(|error| error.into_inner())
|
||||
fn state(&self) -> std::sync::MutexGuard<'_, State> {
|
||||
self.state.lock().unwrap_or_else(|error| error.into_inner())
|
||||
}
|
||||
|
||||
fn protocol<'a>(
|
||||
&self,
|
||||
protocols: &'a mut HashMap<String, Protocol>,
|
||||
name: &str,
|
||||
) -> &'a mut Protocol {
|
||||
protocols
|
||||
.entry(name.to_owned())
|
||||
.or_insert_with(|| Protocol {
|
||||
plane: Arc::new(Plane {
|
||||
local: self.local,
|
||||
table: ArcSwap::from_pointee(ForwardingTable::default()),
|
||||
tally: self.tally.clone(),
|
||||
}),
|
||||
graph: HashMap::new(),
|
||||
members: HashSet::new(),
|
||||
raw: HashMap::new(),
|
||||
readers: HashMap::new(),
|
||||
links: HashMap::new(),
|
||||
})
|
||||
/// Sets which kinds of transport are preferred, best first, when a
|
||||
/// neighbour can be reached over more than one.
|
||||
pub fn set_priority(&self, kinds: Vec<String>) {
|
||||
let mut state = self.state();
|
||||
if state.priority != kinds {
|
||||
state.priority = kinds;
|
||||
state.publish();
|
||||
}
|
||||
}
|
||||
|
||||
/// Wakes the control loop promptly when a transport reader ends.
|
||||
@@ -282,14 +361,11 @@ impl RelayHub {
|
||||
}
|
||||
|
||||
/// Publishes authenticated topology. Refreshing identical state is cheap.
|
||||
pub fn set_topology(
|
||||
&self,
|
||||
protocol: &str,
|
||||
graph: HashMap<PeerId, Vec<PeerId>>,
|
||||
members: HashSet<PeerId>,
|
||||
) {
|
||||
let mut protocols = self.protocols();
|
||||
let state = self.protocol(&mut protocols, protocol);
|
||||
///
|
||||
/// Every row of `graph` is one member's own account of who it has a link
|
||||
/// to; what kind of link it is does not matter here.
|
||||
pub fn set_topology(&self, graph: HashMap<PeerId, Vec<PeerId>>, members: HashSet<PeerId>) {
|
||||
let mut state = self.state();
|
||||
if state.graph != graph || state.members != members {
|
||||
state.graph = graph;
|
||||
state.members = members;
|
||||
@@ -298,22 +374,44 @@ impl RelayHub {
|
||||
}
|
||||
|
||||
/// Whether the current table has a path to this destination.
|
||||
pub fn reachable(&self, peer: EndpointId, protocol: &str) -> bool {
|
||||
self.protocols().get(protocol).is_some_and(|state| {
|
||||
state
|
||||
.plane
|
||||
.table
|
||||
.load()
|
||||
.routes
|
||||
.contains_key(peer.as_bytes())
|
||||
pub fn reachable(&self, peer: EndpointId) -> bool {
|
||||
self.state()
|
||||
.plane
|
||||
.table
|
||||
.load()
|
||||
.routes
|
||||
.contains_key(peer.as_bytes())
|
||||
}
|
||||
|
||||
/// How a destination is reached right now, if it is.
|
||||
pub fn route(&self, peer: EndpointId) -> Option<RouteSummary> {
|
||||
let table = self.state().plane.table.load_full();
|
||||
let route = table.routes.get(peer.as_bytes())?;
|
||||
let via = EndpointId::from_bytes(&route.next.first()?.peer).ok()?;
|
||||
Some(RouteSummary {
|
||||
hops: route.hops,
|
||||
via,
|
||||
})
|
||||
}
|
||||
|
||||
/// Stable logical link, preserved across topology changes.
|
||||
/// The kinds of transport with an open direct link to a peer, best
|
||||
/// first.
|
||||
pub fn direct_kinds(&self, peer: EndpointId) -> Vec<String> {
|
||||
let state = self.state();
|
||||
state
|
||||
.ordered(peer.as_bytes())
|
||||
.into_iter()
|
||||
.filter(|raw| !raw.link.is_closed())
|
||||
.map(|raw| raw.kind.clone())
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Stable end-to-end channel for a protocol, preserved across topology
|
||||
/// and transport changes.
|
||||
pub fn link(&self, peer: EndpointId, protocol: &str) -> Arc<PeerLink> {
|
||||
let mut protocols = self.protocols();
|
||||
let state = self.protocol(&mut protocols, protocol);
|
||||
if let Some(link) = state.links.get(peer.as_bytes()) {
|
||||
let mut state = self.state();
|
||||
let key = (*peer.as_bytes(), protocol.to_owned());
|
||||
if let Some(link) = state.links.get(&key) {
|
||||
return link.clone();
|
||||
}
|
||||
let (inbox_tx, inbox_rx) = mpsc::channel(INBOX);
|
||||
@@ -321,88 +419,118 @@ impl RelayHub {
|
||||
let link = Arc::new(PeerLink {
|
||||
network: self.network,
|
||||
peer,
|
||||
tag: envelope::protocol_tag(protocol),
|
||||
plane: state.plane.clone(),
|
||||
inbox_tx,
|
||||
inbox_rx: tokio::sync::Mutex::new(inbox_rx),
|
||||
closed,
|
||||
});
|
||||
state.links.insert(*peer.as_bytes(), link.clone());
|
||||
state.links.insert(key, link.clone());
|
||||
state.publish();
|
||||
link
|
||||
}
|
||||
|
||||
/// Whether the protocol already owns its logical link.
|
||||
/// Whether the protocol already owns its end-to-end channel to a peer.
|
||||
pub fn has_link(&self, peer: EndpointId, protocol: &str) -> bool {
|
||||
self.protocols()
|
||||
.get(protocol)
|
||||
.is_some_and(|state| state.links.contains_key(peer.as_bytes()))
|
||||
self.state()
|
||||
.links
|
||||
.contains_key(&(*peer.as_bytes(), protocol.to_owned()))
|
||||
}
|
||||
|
||||
/// Installs a raw transport link and starts its independent ingress reader.
|
||||
pub fn set_direct(&self, peer: EndpointId, protocol: &str, raw: SharedLink) {
|
||||
///
|
||||
/// A neighbour may have one link per kind of transport; installing a
|
||||
/// second of the same kind replaces the first.
|
||||
pub fn set_direct(&self, peer: EndpointId, kind: &str, raw: SharedLink) {
|
||||
if raw.network() != self.network || raw.peer() != peer {
|
||||
return;
|
||||
}
|
||||
let mut protocols = self.protocols();
|
||||
let state = self.protocol(&mut protocols, protocol);
|
||||
let mut state = self.state();
|
||||
let id = *peer.as_bytes();
|
||||
if let Some(old) = state.readers.remove(&id) {
|
||||
old.abort();
|
||||
}
|
||||
state.raw.insert(id, raw.clone());
|
||||
state.publish();
|
||||
let plane = state.plane.clone();
|
||||
let changed = self.changed.clone();
|
||||
state.readers.insert(
|
||||
id,
|
||||
tokio::spawn(async move {
|
||||
let mut batch = 0;
|
||||
while let Some(frame) = raw.recv().await {
|
||||
plane.receive(id, frame);
|
||||
batch += 1;
|
||||
if batch == 64 {
|
||||
tokio::task::yield_now().await;
|
||||
batch = 0;
|
||||
}
|
||||
let reader_link = raw.clone();
|
||||
let reader = tokio::spawn(async move {
|
||||
let mut batch = 0;
|
||||
while let Some(frame) = reader_link.recv().await {
|
||||
plane.receive(id, frame);
|
||||
batch += 1;
|
||||
if batch == 64 {
|
||||
tokio::task::yield_now().await;
|
||||
batch = 0;
|
||||
}
|
||||
changed.notify_one();
|
||||
}),
|
||||
);
|
||||
}
|
||||
changed.notify_one();
|
||||
});
|
||||
let links = state.raw.entry(id).or_default();
|
||||
if let Some(position) = links.iter().position(|known| known.kind == kind) {
|
||||
links.remove(position).reader.abort();
|
||||
}
|
||||
links.push(RawLink {
|
||||
kind: kind.to_owned(),
|
||||
link: raw,
|
||||
reader,
|
||||
});
|
||||
state.publish();
|
||||
}
|
||||
|
||||
/// Removes a physical link without tearing down end-to-end protocol state.
|
||||
pub fn clear_direct(&self, peer: EndpointId, protocol: &str) {
|
||||
if let Some(state) = self.protocols().get_mut(protocol) {
|
||||
state.raw.remove(peer.as_bytes());
|
||||
if let Some(task) = state.readers.remove(peer.as_bytes()) {
|
||||
task.abort();
|
||||
/// Removes one physical link without tearing down end-to-end protocol state.
|
||||
pub fn clear_direct(&self, peer: EndpointId, kind: &str) {
|
||||
let mut state = self.state();
|
||||
let mut changed = false;
|
||||
if let Some(links) = state.raw.get_mut(peer.as_bytes()) {
|
||||
if let Some(position) = links.iter().position(|known| known.kind == kind) {
|
||||
links.remove(position).reader.abort();
|
||||
changed = true;
|
||||
}
|
||||
if links.is_empty() {
|
||||
state.raw.remove(peer.as_bytes());
|
||||
}
|
||||
}
|
||||
if changed {
|
||||
state.publish();
|
||||
}
|
||||
}
|
||||
|
||||
/// Revokes all links and topology involving a departed authenticated peer.
|
||||
pub fn remove_peer(&self, peer: EndpointId) {
|
||||
for state in self.protocols().values_mut() {
|
||||
state.raw.remove(peer.as_bytes());
|
||||
if let Some(task) = state.readers.remove(peer.as_bytes()) {
|
||||
task.abort();
|
||||
let mut state = self.state();
|
||||
if let Some(links) = state.raw.remove(peer.as_bytes()) {
|
||||
for raw in links {
|
||||
raw.reader.abort();
|
||||
}
|
||||
if let Some(link) = state.links.remove(peer.as_bytes()) {
|
||||
}
|
||||
let gone: Vec<_> = state
|
||||
.links
|
||||
.keys()
|
||||
.filter(|(id, _)| id == peer.as_bytes())
|
||||
.cloned()
|
||||
.collect();
|
||||
for key in gone {
|
||||
if let Some(link) = state.links.remove(&key) {
|
||||
link.close();
|
||||
}
|
||||
state.members.remove(peer.as_bytes());
|
||||
state.graph.remove(peer.as_bytes());
|
||||
for neighbors in state.graph.values_mut() {
|
||||
neighbors.retain(|id| id != peer.as_bytes());
|
||||
}
|
||||
state.publish();
|
||||
}
|
||||
state.members.remove(peer.as_bytes());
|
||||
state.graph.remove(peer.as_bytes());
|
||||
for neighbors in state.graph.values_mut() {
|
||||
neighbors.retain(|id| id != peer.as_bytes());
|
||||
}
|
||||
state.publish();
|
||||
}
|
||||
|
||||
/// Stops every reader and releases all transport handles.
|
||||
pub fn close(&self) {
|
||||
self.protocols().clear();
|
||||
let mut state = self.state();
|
||||
state.stop_readers();
|
||||
state.raw.clear();
|
||||
for link in state.links.values() {
|
||||
link.close();
|
||||
}
|
||||
state.links.clear();
|
||||
state.graph.clear();
|
||||
state.members.clear();
|
||||
state.plane.table.store(Arc::default());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -411,6 +539,7 @@ impl RelayHub {
|
||||
pub struct PeerLink {
|
||||
network: NetworkId,
|
||||
peer: EndpointId,
|
||||
tag: u64,
|
||||
plane: Arc<Plane>,
|
||||
inbox_tx: mpsc::Sender<Bytes>,
|
||||
inbox_rx: tokio::sync::Mutex<mpsc::Receiver<Bytes>>,
|
||||
@@ -454,10 +583,11 @@ impl PacketLink for PeerLink {
|
||||
let hop = route
|
||||
.select(&self.plane.local, flow)
|
||||
.ok_or(TransportError::Closed)?;
|
||||
hop.link.send(envelope::encode(
|
||||
hop.send(envelope::encode(
|
||||
self.plane.local,
|
||||
*self.peer.as_bytes(),
|
||||
flow,
|
||||
self.tag,
|
||||
&payload,
|
||||
))?;
|
||||
if route.hops > 1 {
|
||||
@@ -489,7 +619,9 @@ impl PacketLink for PeerLink {
|
||||
fn path_description(&self) -> String {
|
||||
let table = self.plane.table.load();
|
||||
match table.routes.get(self.peer.as_bytes()) {
|
||||
Some(route) if route.hops == 1 => route.next[0].link.path_description(),
|
||||
Some(route) if route.hops == 1 => route.next[0]
|
||||
.first_open()
|
||||
.map_or_else(|| "unreachable".into(), |link| link.path_description()),
|
||||
Some(route) => format!(
|
||||
"relay {} hops via {} ({} equal paths)",
|
||||
route.hops,
|
||||
|
||||
@@ -16,6 +16,9 @@ fn peer(seed: u8) -> EndpointId {
|
||||
fn id(seed: u8) -> PeerId {
|
||||
*peer(seed).as_bytes()
|
||||
}
|
||||
fn tag() -> u64 {
|
||||
envelope::protocol_tag("ip")
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct Wire {
|
||||
@@ -78,9 +81,8 @@ impl PacketLink for Wire {
|
||||
}
|
||||
}
|
||||
|
||||
fn topology(hub: &RelayHub, protocol: &str, edges: &[(u8, &[u8])], members: &[u8]) {
|
||||
fn topology(hub: &RelayHub, edges: &[(u8, &[u8])], members: &[u8]) {
|
||||
hub.set_topology(
|
||||
protocol,
|
||||
edges
|
||||
.iter()
|
||||
.map(|(a, bs)| (id(*a), bs.iter().map(|b| id(*b)).collect()))
|
||||
@@ -96,7 +98,7 @@ async fn ecmp_is_per_flow_direct_wins_and_existing_tunnel_survives_changes() {
|
||||
let c = Wire::new(3, true);
|
||||
hub.set_direct(peer(2), "ip", b.clone());
|
||||
hub.set_direct(peer(3), "ip", c.clone());
|
||||
topology(&hub, "ip", &[(2, &[4]), (3, &[4])], &[1, 2, 3, 4]);
|
||||
topology(&hub, &[(2, &[4]), (3, &[4])], &[1, 2, 3, 4]);
|
||||
let link = hub.link(peer(4), "ip");
|
||||
let mut chosen = HashMap::new();
|
||||
for _ in 0..8 {
|
||||
@@ -132,7 +134,7 @@ async fn ecmp_is_per_flow_direct_wins_and_existing_tunnel_survives_changes() {
|
||||
}
|
||||
assert_eq!(c.take().len(), 100);
|
||||
hub.clear_direct(peer(2), "ip");
|
||||
topology(&hub, "ip", &[(3, &[])], &[1, 2, 3, 4]);
|
||||
topology(&hub, &[(3, &[])], &[1, 2, 3, 4]);
|
||||
assert!(link.send(Bytes::new()).is_err());
|
||||
assert!(!link.is_closed());
|
||||
assert!(Arc::ptr_eq(&link, &hub.link(peer(4), "ip")));
|
||||
@@ -152,8 +154,8 @@ async fn transport_reader_forwards_without_any_plugin_reader_and_isolates_protoc
|
||||
let c = Wire::new(3, true);
|
||||
hub.set_direct(peer(1), "ip", a.clone());
|
||||
hub.set_direct(peer(3), "ip", c.clone());
|
||||
topology(&hub, "ip", &[(3, &[4])], &[1, 2, 3, 4]);
|
||||
let frame = envelope::encode(id(1), id(4), 91, b"end-to-end encrypted");
|
||||
topology(&hub, &[(3, &[4])], &[1, 2, 3, 4]);
|
||||
let frame = envelope::encode(id(1), id(4), 91, tag(), b"end-to-end encrypted");
|
||||
let pointer = frame.as_ptr();
|
||||
a.feed.send(frame).await.unwrap();
|
||||
tokio::time::timeout(std::time::Duration::from_secs(1), async {
|
||||
@@ -170,12 +172,136 @@ async fn transport_reader_forwards_without_any_plugin_reader_and_isolates_protoc
|
||||
ROUTING_HOP_LIMIT - 1
|
||||
);
|
||||
assert_eq!(&sent[0][RELAY_OVERHEAD..], b"end-to-end encrypted");
|
||||
let isolated = hub.link(peer(4), "other-ip");
|
||||
assert!(isolated.send(Bytes::new()).is_err());
|
||||
assert_eq!(hub.counters().forwarded, 1);
|
||||
hub.close();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_frame_is_delivered_only_to_the_protocol_it_names() {
|
||||
let hub = RelayHub::new(network(), peer(2));
|
||||
hub.set_direct(peer(1), "ip", Wire::new(1, true));
|
||||
let ip = hub.link(peer(1), "ip");
|
||||
let other = hub.link(peer(1), "other-ip");
|
||||
let plane = ip.plane.clone();
|
||||
plane.receive(id(1), envelope::encode(id(1), id(2), 0, tag(), b"for ip"));
|
||||
plane.receive(
|
||||
id(1),
|
||||
envelope::encode(
|
||||
id(1),
|
||||
id(2),
|
||||
0,
|
||||
envelope::protocol_tag("other-ip"),
|
||||
b"for other",
|
||||
),
|
||||
);
|
||||
plane.receive(
|
||||
id(1),
|
||||
envelope::encode(
|
||||
id(1),
|
||||
id(2),
|
||||
0,
|
||||
envelope::protocol_tag("nobody-runs-this"),
|
||||
b"for nobody",
|
||||
),
|
||||
);
|
||||
assert_eq!(ip.recv().await.unwrap(), Bytes::from_static(b"for ip"));
|
||||
assert_eq!(
|
||||
other.recv().await.unwrap(),
|
||||
Bytes::from_static(b"for other")
|
||||
);
|
||||
assert_eq!(hub.counters().dropped_unknown, 1);
|
||||
hub.close();
|
||||
}
|
||||
|
||||
/// A middle member joins two neighbours over different transports, and the
|
||||
/// two ends share a protocol the middle has never heard of.
|
||||
#[tokio::test]
|
||||
async fn a_tunnel_crosses_a_middle_member_whose_links_are_of_different_kinds() {
|
||||
let a = RelayHub::new(network(), peer(1));
|
||||
let c = RelayHub::new(network(), peer(3));
|
||||
let b = RelayHub::new(network(), peer(2));
|
||||
let a_to_c = Wire::new(3, true);
|
||||
let c_to_a = Wire::new(1, true);
|
||||
let c_to_b = Wire::new(2, true);
|
||||
let b_to_c = Wire::new(3, true);
|
||||
a.set_direct(peer(3), "wg", a_to_c.clone());
|
||||
c.set_direct(peer(1), "wg", c_to_a);
|
||||
c.set_direct(peer(2), "wg-quic", c_to_b.clone());
|
||||
b.set_direct(peer(3), "wg-quic", b_to_c.clone());
|
||||
topology(&a, &[(3, &[1, 2])], &[1, 2, 3]);
|
||||
topology(&b, &[(3, &[1, 2])], &[1, 2, 3]);
|
||||
topology(&c, &[(1, &[3]), (2, &[3])], &[1, 2, 3]);
|
||||
|
||||
let outgoing = a.link(peer(2), "tunnel");
|
||||
let incoming = b.link(peer(1), "tunnel");
|
||||
outgoing
|
||||
.send_flow(Bytes::from_static(b"sealed"), 5)
|
||||
.unwrap();
|
||||
let frame = a_to_c.take().remove(0);
|
||||
|
||||
// The middle forwards it onto its other kind of link without looking
|
||||
// inside.
|
||||
c.state().plane.receive(id(1), frame);
|
||||
let frame = c_to_b.take().remove(0);
|
||||
assert_eq!(c.counters().forwarded, 1);
|
||||
|
||||
b.state().plane.receive(id(3), frame);
|
||||
assert_eq!(
|
||||
incoming.recv().await.unwrap(),
|
||||
Bytes::from_static(b"sealed")
|
||||
);
|
||||
assert_eq!(b.counters().received_via, 1);
|
||||
assert_eq!(
|
||||
b.route(peer(1)),
|
||||
Some(RouteSummary {
|
||||
hops: 2,
|
||||
via: peer(3)
|
||||
})
|
||||
);
|
||||
a.close();
|
||||
b.close();
|
||||
c.close();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_neighbour_is_reached_over_the_preferred_kind_and_falls_back_when_it_is_lost() {
|
||||
let hub = RelayHub::new(network(), peer(1));
|
||||
hub.set_priority(vec!["wg".into(), "tcp-tls".into(), "wg-quic".into()]);
|
||||
let slow = Wire::new(2, true);
|
||||
let fast = Wire::new(2, true);
|
||||
hub.set_direct(peer(2), "wg-quic", slow.clone());
|
||||
let link = hub.link(peer(2), "tunnel");
|
||||
link.send(Bytes::from_static(b"one")).unwrap();
|
||||
assert_eq!(slow.take().len(), 1);
|
||||
assert_eq!(hub.direct_kinds(peer(2)), ["wg-quic"]);
|
||||
|
||||
// A better kind appears: traffic moves to it with nothing to redo.
|
||||
hub.set_direct(peer(2), "wg", fast.clone());
|
||||
assert_eq!(hub.direct_kinds(peer(2)), ["wg", "wg-quic"]);
|
||||
link.send(Bytes::from_static(b"two")).unwrap();
|
||||
assert_eq!(fast.take().len(), 1);
|
||||
assert!(slow.take().is_empty());
|
||||
|
||||
// It dies: the worse one carries on, and the logical link is untouched.
|
||||
fast.dead.store(true, Ordering::Relaxed);
|
||||
link.send(Bytes::from_static(b"three")).unwrap();
|
||||
assert_eq!(slow.take().len(), 1);
|
||||
assert!(!link.is_closed());
|
||||
assert_eq!(
|
||||
hub.route(peer(2)),
|
||||
Some(RouteSummary {
|
||||
hops: 1,
|
||||
via: peer(2)
|
||||
})
|
||||
);
|
||||
|
||||
hub.clear_direct(peer(2), "wg-quic");
|
||||
hub.clear_direct(peer(2), "wg");
|
||||
assert!(!hub.reachable(peer(2)));
|
||||
assert!(Arc::ptr_eq(&link, &hub.link(peer(2), "tunnel")));
|
||||
hub.close();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn local_delivery_is_bounded_and_malformed_or_unknown_sources_are_dropped() {
|
||||
let hub = RelayHub::new(network(), peer(2));
|
||||
@@ -184,18 +310,21 @@ async fn local_delivery_is_bounded_and_malformed_or_unknown_sources_are_dropped(
|
||||
let link = hub.link(peer(1), "ip");
|
||||
let plane = link.plane.clone();
|
||||
for _ in 0..INBOX + 1 {
|
||||
plane.receive(id(1), envelope::encode(id(1), id(2), 0, b"ip"));
|
||||
plane.receive(id(1), envelope::encode(id(1), id(2), 0, tag(), b"ip"));
|
||||
}
|
||||
assert_eq!(hub.counters().dropped_congested, 1);
|
||||
for _ in 0..INBOX {
|
||||
assert_eq!(link.recv().await.unwrap(), Bytes::from_static(b"ip"));
|
||||
}
|
||||
plane.receive(id(1), Bytes::from_static(b"broken"));
|
||||
plane.receive(id(1), envelope::encode(id(9), id(2), 0, b"unknown member"));
|
||||
plane.receive(
|
||||
id(1),
|
||||
envelope::encode(id(9), id(2), 0, tag(), b"unknown member"),
|
||||
);
|
||||
assert_eq!(hub.counters().dropped_unknown, 2);
|
||||
plane.receive(
|
||||
id(1),
|
||||
envelope::encode(id(1), id(9), 0, b"unknown destination"),
|
||||
envelope::encode(id(1), id(9), 0, tag(), b"unknown destination"),
|
||||
);
|
||||
assert_eq!(hub.counters().dropped_no_link, 1);
|
||||
hub.close();
|
||||
@@ -210,11 +339,11 @@ async fn inconsistent_tables_cannot_loop_beyond_the_hop_limit() {
|
||||
a.set_direct(peer(2), "ip", ab.clone());
|
||||
b.set_direct(peer(1), "ip", ba.clone());
|
||||
// Deliberately inconsistent snapshots while updates are in flight.
|
||||
topology(&a, "ip", &[(2, &[4])], &[1, 2, 3, 4]);
|
||||
topology(&b, "ip", &[(1, &[4])], &[1, 2, 3, 4]);
|
||||
topology(&a, &[(2, &[4])], &[1, 2, 3, 4]);
|
||||
topology(&b, &[(1, &[4])], &[1, 2, 3, 4]);
|
||||
let ap = a.link(peer(4), "ip").plane.clone();
|
||||
let bp = b.link(peer(4), "ip").plane.clone();
|
||||
let mut frame = envelope::encode(id(3), id(4), 7, b"transit");
|
||||
let mut frame = envelope::encode(id(3), id(4), 7, tag(), b"transit");
|
||||
for turn in 0..ROUTING_HOP_LIMIT {
|
||||
let (plane, incoming, wire) = if turn % 2 == 0 {
|
||||
(&ap, id(2), &ab)
|
||||
@@ -253,7 +382,7 @@ async fn forwarding_benchmark() {
|
||||
hub.set_direct(peer(3), "ip", wire.clone());
|
||||
let alternate = Wire::new(5, false);
|
||||
hub.set_direct(peer(5), "ip", alternate.clone());
|
||||
topology(&hub, "ip", &[(3, &[4]), (5, &[4])], &[1, 2, 3, 4, 5]);
|
||||
topology(&hub, &[(3, &[4]), (5, &[4])], &[1, 2, 3, 4, 5]);
|
||||
let plane = hub.link(peer(4), "ip").plane.clone();
|
||||
let source = id(1);
|
||||
let destination = id(4);
|
||||
@@ -264,7 +393,9 @@ async fn forwarding_benchmark() {
|
||||
const BATCHES: usize = 100;
|
||||
for _ in 0..BATCHES {
|
||||
let frames: Vec<_> = (0..COUNT)
|
||||
.map(|flow| envelope::encode(source, destination, flow as u64, &vec![42; size]))
|
||||
.map(|flow| {
|
||||
envelope::encode(source, destination, flow as u64, tag(), &vec![42; size])
|
||||
})
|
||||
.collect();
|
||||
let start = std::time::Instant::now();
|
||||
for frame in frames {
|
||||
|
||||
@@ -3,8 +3,8 @@ use super::{FlowId, PeerId};
|
||||
use crate::config::{MAX_DATA_DATAGRAM, ROUTING_HOP_LIMIT};
|
||||
use bytes::{BufMut, Bytes, BytesMut};
|
||||
|
||||
pub(crate) const HEADER: usize = 2 + 32 + 32 + 8;
|
||||
const VERSION: u8 = 1;
|
||||
pub(crate) const HEADER: usize = 2 + 32 + 32 + 8 + 8;
|
||||
const VERSION: u8 = 2;
|
||||
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
pub(crate) struct Header {
|
||||
@@ -12,6 +12,15 @@ pub(crate) struct Header {
|
||||
pub destination: PeerId,
|
||||
pub remaining: u8,
|
||||
pub flow: FlowId,
|
||||
/// Which end-to-end protocol the payload belongs to, whatever links carry it.
|
||||
pub protocol: u64,
|
||||
}
|
||||
|
||||
/// A stable 64-bit tag for a protocol id (FNV-1a), the same on every member.
|
||||
pub(crate) fn protocol_tag(name: &str) -> u64 {
|
||||
name.bytes().fold(0xcbf2_9ce4_8422_2325, |hash, byte| {
|
||||
(hash ^ u64::from(byte)).wrapping_mul(0x0000_0100_0000_01b3)
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn decode(bytes: &[u8]) -> Option<Header> {
|
||||
@@ -28,16 +37,24 @@ pub(crate) fn decode(bytes: &[u8]) -> Option<Header> {
|
||||
destination: bytes[34..66].try_into().ok()?,
|
||||
remaining: bytes[1],
|
||||
flow: u64::from_be_bytes(bytes[66..74].try_into().ok()?),
|
||||
protocol: u64::from_be_bytes(bytes[74..82].try_into().ok()?),
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn encode(source: PeerId, destination: PeerId, flow: FlowId, payload: &[u8]) -> Bytes {
|
||||
pub(crate) fn encode(
|
||||
source: PeerId,
|
||||
destination: PeerId,
|
||||
flow: FlowId,
|
||||
protocol: u64,
|
||||
payload: &[u8],
|
||||
) -> Bytes {
|
||||
let mut bytes = BytesMut::with_capacity(HEADER + payload.len());
|
||||
bytes.put_u8(VERSION);
|
||||
bytes.put_u8(ROUTING_HOP_LIMIT);
|
||||
bytes.extend_from_slice(&source);
|
||||
bytes.extend_from_slice(&destination);
|
||||
bytes.put_u64(flow);
|
||||
bytes.put_u64(protocol);
|
||||
bytes.extend_from_slice(payload);
|
||||
bytes.freeze()
|
||||
}
|
||||
@@ -58,7 +75,7 @@ mod tests {
|
||||
use super::*;
|
||||
#[test]
|
||||
fn transit_reuses_the_buffer_and_rejects_invalid_headers() {
|
||||
let frame = encode([1; 32], [2; 32], 17, &[42; 1280]);
|
||||
let frame = encode([1; 32], [2; 32], 17, protocol_tag("wg"), &[42; 1280]);
|
||||
let address = frame.as_ptr();
|
||||
let frame = decrement(frame);
|
||||
assert_eq!(
|
||||
@@ -66,7 +83,10 @@ mod tests {
|
||||
address,
|
||||
"no payload copy for an owned buffer"
|
||||
);
|
||||
assert_eq!(decode(&frame).unwrap().remaining, ROUTING_HOP_LIMIT - 1);
|
||||
let header = decode(&frame).unwrap();
|
||||
assert_eq!(header.remaining, ROUTING_HOP_LIMIT - 1);
|
||||
assert_eq!(header.protocol, protocol_tag("wg"));
|
||||
assert_ne!(protocol_tag("wg"), protocol_tag("wg-quic"));
|
||||
assert_eq!(&frame[HEADER..], &[42; 1280]);
|
||||
for length in 0..HEADER {
|
||||
assert!(decode(&frame[..length]).is_none());
|
||||
|
||||
@@ -484,11 +484,43 @@ pub struct NetworkReport {
|
||||
pub range_conflict: Option<String>,
|
||||
/// The overlay, when a protocol is running one.
|
||||
pub overlay: Option<OverlayReport>,
|
||||
/// This agent's own name, when another member claimed it first.
|
||||
pub hostname_conflict: Option<HostnameConflictReport>,
|
||||
/// Exit-node settings: whether this agent offers one, and whether it
|
||||
/// uses one.
|
||||
pub exit: ExitReport,
|
||||
}
|
||||
|
||||
/// How packets reach a member right now.
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub enum DataPathReport {
|
||||
/// Nothing carries packets to it: it is away, or only connected for
|
||||
/// control while a data path is being set up.
|
||||
#[default]
|
||||
None,
|
||||
/// A direct link.
|
||||
Direct {
|
||||
/// The kind of transport carrying it, e.g. `wg`.
|
||||
transport: String,
|
||||
},
|
||||
/// Through other members.
|
||||
Relayed {
|
||||
/// Number of links on the way, at least two.
|
||||
hops: u8,
|
||||
/// The member the first of them goes to, as an endpoint id.
|
||||
via: String,
|
||||
},
|
||||
}
|
||||
|
||||
/// A name somebody else claimed first.
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct HostnameConflictReport {
|
||||
/// The name that is taken.
|
||||
pub name: String,
|
||||
/// The member that holds it, as an endpoint id.
|
||||
pub holder: String,
|
||||
}
|
||||
|
||||
/// One member of the network, from signed state.
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct MemberReport {
|
||||
@@ -500,6 +532,15 @@ pub struct MemberReport {
|
||||
pub failed_dials: u32,
|
||||
/// The last hostname this agent saw it announce, kept while it is away.
|
||||
pub hostname: Option<String>,
|
||||
/// Whether this agent can reach it now, by any path. The one answer
|
||||
/// every client gives to "is it online": a member reached through
|
||||
/// others is online, over a longer path.
|
||||
pub online: bool,
|
||||
/// How packets reach it.
|
||||
pub path: DataPathReport,
|
||||
/// The name it claimed, when somebody else claimed it first. It still
|
||||
/// works by address; it is not found by that name.
|
||||
pub hostname_conflict: Option<HostnameConflictReport>,
|
||||
}
|
||||
|
||||
/// One control plane peer.
|
||||
@@ -516,6 +557,8 @@ pub struct PeerReport {
|
||||
pub rtt_ms: Option<u64>,
|
||||
/// The peer offers itself as an exit node in this network.
|
||||
pub exit_node: bool,
|
||||
/// How packets reach it.
|
||||
pub path: DataPathReport,
|
||||
}
|
||||
|
||||
/// The WireGuard overlay of one network.
|
||||
@@ -779,7 +822,7 @@ pub const EXCHANGE_TIMEOUT: Duration = Duration::from_secs(5);
|
||||
///
|
||||
/// Bump it whenever [`Request`], [`Response`] or anything they contain
|
||||
/// changes shape.
|
||||
pub const CONTROL_PROTOCOL: u32 = u32::from_be_bytes([b'T', b'S', b'N', 17]);
|
||||
pub const CONTROL_PROTOCOL: u32 = u32::from_be_bytes([b'T', b'S', b'N', 18]);
|
||||
|
||||
/// Reads one request off an accepted stream, answers it, writes the response.
|
||||
///
|
||||
|
||||
@@ -20,7 +20,7 @@ use crate::error::ProtocolError;
|
||||
/// The version in the ALPN is the wire-compatibility version of the control
|
||||
/// protocol. It is independent of the network identity scheme version, so
|
||||
/// bumping it must not change any existing [`crate::NetworkId`].
|
||||
pub const ALPN: &[u8] = b"tsunagi/ctrl/4";
|
||||
pub const ALPN: &[u8] = b"tsunagi/ctrl/5";
|
||||
|
||||
/// ALPN of the tsunagi data plane.
|
||||
///
|
||||
@@ -30,9 +30,11 @@ pub const ALPN: &[u8] = b"tsunagi/ctrl/4";
|
||||
/// transport underneath can be replaced without touching the control protocol.
|
||||
///
|
||||
/// Version 4 adds source, destination, hop limit and stable flow id above
|
||||
/// transport fragmentation. All members must upgrade together. Persistent
|
||||
/// identities and network configuration do not change.
|
||||
pub const DATA_ALPN: &[u8] = b"tsunagi/data/4";
|
||||
/// transport fragmentation. Version 5 adds the end-to-end protocol to that
|
||||
/// envelope, so a tunnel no longer depends on the kind of link under it, and
|
||||
/// a first-claimed time to a signed hostname claim. All members must upgrade
|
||||
/// together. Persistent identities and network configuration do not change.
|
||||
pub const DATA_ALPN: &[u8] = b"tsunagi/data/5";
|
||||
|
||||
/// Largest plugin protocol identifier accepted when opening a data channel.
|
||||
pub const MAX_DATA_PROTOCOL_LEN: usize = 32;
|
||||
@@ -41,7 +43,7 @@ pub const MAX_DATA_PROTOCOL_LEN: usize = 32;
|
||||
pub const MAX_SIGNATURE_LEN: usize = 64;
|
||||
|
||||
/// Control protocol version carried inside the handshake.
|
||||
pub const PROTOCOL_VERSION: u16 = 4;
|
||||
pub const PROTOCOL_VERSION: u16 = 5;
|
||||
|
||||
/// First message of the handshake, sent by the initiator.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
|
||||
@@ -150,7 +150,7 @@ pub fn derived_ipv4_range(network: NetworkId) -> Ipv4Range {
|
||||
}
|
||||
|
||||
/// Frozen domain separator for the bytes a record signature covers.
|
||||
pub const RECORD_DOMAIN: &str = "tsunagi-signed-record-v2";
|
||||
pub const RECORD_DOMAIN: &str = "tsunagi-signed-record-v3";
|
||||
|
||||
/// Longest hostname a record may carry.
|
||||
///
|
||||
@@ -207,6 +207,18 @@ pub enum RecordBody {
|
||||
range: Option<Ipv4Range>,
|
||||
/// The name this author answers to.
|
||||
hostname: Option<String>,
|
||||
/// When this author first claimed that name, in milliseconds since
|
||||
/// the Unix epoch, kept across every later version that keeps the
|
||||
/// name. It is what makes a name belong to whoever claimed it first
|
||||
/// rather than to whoever happens to sort lowest, so a member that
|
||||
/// arrives later with the same name cannot take it over.
|
||||
///
|
||||
/// The author's own word: a member with a wrong clock can claim an
|
||||
/// earlier time. That costs it nothing it could not do by other
|
||||
/// means, since anybody who knows the network secret is a member,
|
||||
/// and it cannot take an address or reach anything it was not
|
||||
/// already entitled to.
|
||||
hostname_since: Option<u64>,
|
||||
},
|
||||
/// This author gives up everything it claimed.
|
||||
///
|
||||
@@ -243,6 +255,14 @@ impl RecordBody {
|
||||
}
|
||||
}
|
||||
|
||||
/// When the author first claimed the hostname, if it has one.
|
||||
pub fn hostname_since(&self) -> Option<u64> {
|
||||
match self {
|
||||
RecordBody::Claim { hostname_since, .. } => *hostname_since,
|
||||
RecordBody::Release => None,
|
||||
}
|
||||
}
|
||||
|
||||
fn canonical(&self) -> Vec<u8> {
|
||||
let mut out = Vec::with_capacity(96);
|
||||
match self {
|
||||
@@ -250,6 +270,7 @@ impl RecordBody {
|
||||
address,
|
||||
range,
|
||||
hostname,
|
||||
hostname_since,
|
||||
} => {
|
||||
push_lp(&mut out, b"claim");
|
||||
push_opt(
|
||||
@@ -270,6 +291,13 @@ impl RecordBody {
|
||||
.as_deref(),
|
||||
);
|
||||
push_opt(&mut out, hostname.as_deref().map(str::as_bytes));
|
||||
push_opt(
|
||||
&mut out,
|
||||
hostname_since
|
||||
.map(|since| since.to_be_bytes())
|
||||
.as_ref()
|
||||
.map(|bytes| &bytes[..]),
|
||||
);
|
||||
}
|
||||
RecordBody::Release => {
|
||||
push_lp(&mut out, b"release");
|
||||
@@ -391,8 +419,12 @@ impl SignedRecord {
|
||||
address,
|
||||
range,
|
||||
hostname,
|
||||
hostname_since,
|
||||
} = &self.body
|
||||
{
|
||||
if hostname.is_none() && hostname_since.is_some() {
|
||||
return Err(StateError::Malformed("a time was claimed for no hostname"));
|
||||
}
|
||||
// Bounds before anything is believed, because all of this came
|
||||
// off the network.
|
||||
if let Some(range) = range {
|
||||
@@ -585,25 +617,33 @@ impl StateSet {
|
||||
|
||||
/// Who currently holds each claimed hostname.
|
||||
///
|
||||
/// The same rule as addresses, for the same reason: a name is owned, two
|
||||
/// members may claim one, and every replica has to reach the same answer
|
||||
/// about who has it without asking anybody.
|
||||
/// A name belongs to the member that claimed it first, and the others
|
||||
/// that claim it keep working by address but are not found by that name.
|
||||
/// "First" is the time each author signed into its own claim, with the
|
||||
/// lower id deciding a tie and a claim with no time counting as last, so
|
||||
/// every replica reaches the same answer about who has it without
|
||||
/// asking anybody. It is a pure function of the records: nothing depends
|
||||
/// on the order they arrived in, so there is no race to lose.
|
||||
pub fn hostname_holders(&self) -> HashMap<String, EndpointId> {
|
||||
let mut holders: HashMap<String, EndpointId> = HashMap::new();
|
||||
let mut holders: HashMap<String, (u64, EndpointId)> = HashMap::new();
|
||||
for (author, record) in &self.records {
|
||||
let Some(hostname) = record.body.hostname() else {
|
||||
continue;
|
||||
};
|
||||
let claim = (record.body.hostname_since().unwrap_or(u64::MAX), *author);
|
||||
holders
|
||||
.entry(hostname.to_string())
|
||||
.and_modify(|held| {
|
||||
if author.as_bytes() < held.as_bytes() {
|
||||
*held = *author;
|
||||
if (claim.0, claim.1.as_bytes()) < (held.0, held.1.as_bytes()) {
|
||||
*held = claim;
|
||||
}
|
||||
})
|
||||
.or_insert(*author);
|
||||
.or_insert(claim);
|
||||
}
|
||||
holders
|
||||
.into_iter()
|
||||
.map(|(hostname, (_, author))| (hostname, author))
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// The hostname an author holds, if it holds one uncontested.
|
||||
@@ -612,6 +652,19 @@ impl StateSet {
|
||||
(self.hostname_holders().get(hostname) == Some(author)).then_some(hostname)
|
||||
}
|
||||
|
||||
/// The name an author claimed and the member that holds it instead, when
|
||||
/// somebody who claimed it earlier does.
|
||||
pub fn hostname_taken(&self, author: &EndpointId) -> Option<(&str, EndpointId)> {
|
||||
let hostname = self.records.get(author)?.body.hostname()?;
|
||||
let holder = *self.hostname_holders().get(hostname)?;
|
||||
(holder != *author).then_some((hostname, holder))
|
||||
}
|
||||
|
||||
/// When an author first claimed the name it holds now, if it has one.
|
||||
pub fn hostname_since_of(&self, author: &EndpointId) -> Option<u64> {
|
||||
self.records.get(author)?.body.hostname_since()
|
||||
}
|
||||
|
||||
/// The address an author holds, if it holds one uncontested.
|
||||
pub fn address_of(&self, author: &EndpointId) -> Option<Ipv4Addr> {
|
||||
let address = self.records.get(author)?.body.claimed_address()?;
|
||||
@@ -661,6 +714,7 @@ mod tests {
|
||||
address: Some(address.parse().unwrap()),
|
||||
range: Some(range()),
|
||||
hostname: None,
|
||||
hostname_since: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -720,6 +774,7 @@ mod tests {
|
||||
address: Some("10.99.0.1".parse().unwrap()),
|
||||
range: Some(range()),
|
||||
hostname: None,
|
||||
hostname_since: None,
|
||||
},
|
||||
);
|
||||
assert!(matches!(outside.verify(id), Err(StateError::Malformed(_))));
|
||||
@@ -735,6 +790,7 @@ mod tests {
|
||||
prefix_len: 31,
|
||||
}),
|
||||
hostname: None,
|
||||
hostname_since: None,
|
||||
},
|
||||
);
|
||||
assert!(matches!(no_hosts.verify(id), Err(StateError::Malformed(_))));
|
||||
@@ -776,6 +832,7 @@ mod tests {
|
||||
address: None,
|
||||
range: None,
|
||||
hostname: Some(bad.to_string()),
|
||||
hostname_since: None,
|
||||
},
|
||||
);
|
||||
assert!(
|
||||
@@ -797,6 +854,7 @@ mod tests {
|
||||
address: Some("10.13.37.5".parse().unwrap()),
|
||||
range: None,
|
||||
hostname: None,
|
||||
hostname_since: None,
|
||||
},
|
||||
);
|
||||
assert!(matches!(record.verify(id), Err(StateError::Malformed(_))));
|
||||
@@ -818,6 +876,7 @@ mod tests {
|
||||
address: None,
|
||||
range: None,
|
||||
hostname: Some("music".into()),
|
||||
hostname_since: None,
|
||||
},
|
||||
)
|
||||
};
|
||||
@@ -860,6 +919,7 @@ mod tests {
|
||||
address: None,
|
||||
range: None,
|
||||
hostname: Some(name.to_string()),
|
||||
hostname_since: None,
|
||||
},
|
||||
)
|
||||
};
|
||||
@@ -879,6 +939,109 @@ mod tests {
|
||||
assert_eq!(set.hostname_of(&secret.public()), Some("new"));
|
||||
}
|
||||
|
||||
fn named_at(
|
||||
secret: &SecretKey,
|
||||
id: NetworkId,
|
||||
version: u64,
|
||||
name: &str,
|
||||
since: u64,
|
||||
) -> SignedRecord {
|
||||
SignedRecord::sign(
|
||||
secret,
|
||||
id,
|
||||
version,
|
||||
RecordBody::Claim {
|
||||
address: None,
|
||||
range: None,
|
||||
hostname: Some(name.into()),
|
||||
hostname_since: Some(since),
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_name_belongs_to_whoever_claimed_it_first_whatever_their_ids_or_arrival_order() {
|
||||
let id = network("naming");
|
||||
let mut keys = [SecretKey::generate(), SecretKey::generate()];
|
||||
keys.sort_by_key(|key| *key.public().as_bytes());
|
||||
let [low, high] = keys;
|
||||
|
||||
// The member with the higher id claimed first, so it holds the name,
|
||||
// and the lower id does not take it back by sorting first.
|
||||
let early = named_at(&high, id, 1, "music", 1_000);
|
||||
let late = named_at(&low, id, 1, "music", 2_000);
|
||||
for order in [[&early, &late], [&late, &early]] {
|
||||
let mut set = StateSet::new();
|
||||
for record in order {
|
||||
set.merge(id, (*record).clone()).unwrap();
|
||||
}
|
||||
assert_eq!(set.hostname_of(&high.public()), Some("music"));
|
||||
assert_eq!(set.hostname_of(&low.public()), None);
|
||||
assert_eq!(
|
||||
set.hostname_taken(&low.public()),
|
||||
Some(("music", high.public()))
|
||||
);
|
||||
assert_eq!(set.hostname_taken(&high.public()), None);
|
||||
}
|
||||
|
||||
// At the same instant the lower id decides, the same way everywhere.
|
||||
let mut set = StateSet::new();
|
||||
set.merge(id, named_at(&high, id, 1, "music", 5)).unwrap();
|
||||
set.merge(id, named_at(&low, id, 1, "music", 5)).unwrap();
|
||||
assert_eq!(set.hostname_of(&low.public()), Some("music"));
|
||||
assert_eq!(set.hostname_of(&high.public()), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_name_that_loses_is_found_again_once_the_holder_lets_it_go() {
|
||||
let id = network("naming");
|
||||
let first = SecretKey::generate();
|
||||
let second = SecretKey::generate();
|
||||
let mut set = StateSet::new();
|
||||
set.merge(id, named_at(&first, id, 1, "music", 1)).unwrap();
|
||||
set.merge(id, named_at(&second, id, 1, "music", 2)).unwrap();
|
||||
assert_eq!(set.hostname_of(&second.public()), None);
|
||||
|
||||
// Renaming starts the clock afresh, so it cannot be used to jump the
|
||||
// queue for a name somebody else already holds.
|
||||
set.merge(id, named_at(&first, id, 2, "other", 3)).unwrap();
|
||||
assert_eq!(set.hostname_of(&second.public()), Some("music"));
|
||||
set.merge(id, named_at(&first, id, 3, "music", 4)).unwrap();
|
||||
assert_eq!(set.hostname_of(&second.public()), Some("music"));
|
||||
assert_eq!(set.hostname_of(&first.public()), None);
|
||||
|
||||
// And a release hands it on.
|
||||
set.merge(id, SignedRecord::sign(&second, id, 2, RecordBody::Release))
|
||||
.unwrap();
|
||||
assert_eq!(set.hostname_of(&first.public()), Some("music"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_time_with_no_name_is_not_a_valid_claim() {
|
||||
let id = network("naming");
|
||||
let secret = SecretKey::generate();
|
||||
let record = SignedRecord::sign(
|
||||
&secret,
|
||||
id,
|
||||
1,
|
||||
RecordBody::Claim {
|
||||
address: None,
|
||||
range: None,
|
||||
hostname: None,
|
||||
hostname_since: Some(1),
|
||||
},
|
||||
);
|
||||
assert!(matches!(record.verify(id), Err(StateError::Malformed(_))));
|
||||
|
||||
// The time is signed: moving it breaks the signature, so nobody can
|
||||
// improve another member's place in the queue.
|
||||
let mut forged = named_at(&secret, id, 1, "music", 9_000);
|
||||
if let RecordBody::Claim { hostname_since, .. } = &mut forged.body {
|
||||
*hostname_since = Some(1);
|
||||
}
|
||||
assert_eq!(forged.verify(id).unwrap_err(), StateError::BadSignature);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_release_gives_up_the_name_as_well_as_the_address() {
|
||||
let secret = SecretKey::generate();
|
||||
@@ -894,6 +1057,7 @@ mod tests {
|
||||
address: Some("10.13.37.5".parse().unwrap()),
|
||||
range: Some(range()),
|
||||
hostname: Some("music".into()),
|
||||
hostname_since: None,
|
||||
},
|
||||
),
|
||||
)
|
||||
@@ -1057,6 +1221,7 @@ mod tests {
|
||||
address: Some("10.99.0.7".parse().unwrap()),
|
||||
range: Some(custom),
|
||||
hostname: None,
|
||||
hostname_since: None,
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
@@ -21,7 +21,7 @@ use crate::identity::{DeviceIdentity, NetworkId, NetworkName, NetworkSecret};
|
||||
use crate::state::{RecordBody, SignedRecord};
|
||||
|
||||
/// Schema version written by this build.
|
||||
pub const SCHEMA_VERSION: i64 = 6;
|
||||
pub const SCHEMA_VERSION: i64 = 7;
|
||||
|
||||
/// Key of the stored hostname setting.
|
||||
const SETTING_HOSTNAME: &str = "hostname";
|
||||
@@ -221,6 +221,23 @@ impl StateStore {
|
||||
)
|
||||
.map_err(|err| self.corrupt(format!("cannot migrate schema to 6: {err}")))?;
|
||||
}
|
||||
// Migration 6 -> 7: a claim gained the time its hostname was first
|
||||
// claimed, which changed the signing domain. Records signed before
|
||||
// no longer verify, so they are discarded rather than kept as rows
|
||||
// every read has to reject; each member re-publishes its own claim on
|
||||
// its next run and the others arrive again from their authors. The
|
||||
// version counter is kept, so nothing is ever published twice at one
|
||||
// version.
|
||||
if found < 7 {
|
||||
self.conn
|
||||
.execute_batch(
|
||||
"BEGIN;
|
||||
DELETE FROM signed_records;
|
||||
PRAGMA user_version = 7;
|
||||
COMMIT;",
|
||||
)
|
||||
.map_err(|err| self.corrupt(format!("cannot migrate schema to 7: {err}")))?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,308 @@
|
||||
//! Connectivity does not depend on what kind of link each hop happens to be,
|
||||
//! and a member is online however far away it is.
|
||||
|
||||
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
|
||||
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use bytes::Bytes;
|
||||
use iroh::EndpointId;
|
||||
use tsunagi::NetworkId;
|
||||
use tsunagi::agent::DataPath;
|
||||
use tsunagi::dataplane::transport::SharedLink;
|
||||
use tsunagi::dataplane::{IpPlugin, PluginCapability, PluginError};
|
||||
use tsunagi::discovery::SharedMemoryDiscovery;
|
||||
use tsunagi::testing::{TestAgent, network, wait_for_peers, wait_until};
|
||||
|
||||
/// A protocol that moves bytes between two members unchanged and keeps the
|
||||
/// link it was handed, so a test can use the link directly.
|
||||
#[derive(Debug)]
|
||||
struct Carrier {
|
||||
protocol: String,
|
||||
links: Mutex<HashMap<EndpointId, SharedLink>>,
|
||||
}
|
||||
|
||||
impl Carrier {
|
||||
fn new(protocol: &str) -> Arc<Self> {
|
||||
Arc::new(Self {
|
||||
protocol: protocol.into(),
|
||||
links: Mutex::default(),
|
||||
})
|
||||
}
|
||||
|
||||
fn link(&self, peer: EndpointId) -> Option<SharedLink> {
|
||||
self.links.lock().unwrap().get(&peer).cloned()
|
||||
}
|
||||
}
|
||||
|
||||
impl IpPlugin for Carrier {
|
||||
fn protocol_id(&self) -> &str {
|
||||
&self.protocol
|
||||
}
|
||||
|
||||
fn protocol_version(&self) -> u16 {
|
||||
1
|
||||
}
|
||||
|
||||
fn local_capability(
|
||||
&self,
|
||||
_network: NetworkId,
|
||||
) -> Result<Option<PluginCapability>, PluginError> {
|
||||
Ok(Some(PluginCapability {
|
||||
protocol: self.protocol.clone(),
|
||||
version: 1,
|
||||
enabled: true,
|
||||
data: Vec::new(),
|
||||
}))
|
||||
}
|
||||
|
||||
fn on_peer_capability(
|
||||
&self,
|
||||
_network: NetworkId,
|
||||
_peer: EndpointId,
|
||||
_capability: &PluginCapability,
|
||||
) -> Result<(), PluginError> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn on_peer_link(&self, _network: NetworkId, peer: EndpointId, link: SharedLink) {
|
||||
self.links.lock().unwrap().insert(peer, link);
|
||||
}
|
||||
|
||||
fn on_peer_gone(&self, _network: NetworkId, peer: EndpointId) {
|
||||
self.links.lock().unwrap().remove(&peer);
|
||||
}
|
||||
|
||||
fn on_network_deactivated(&self, _network: NetworkId) {}
|
||||
}
|
||||
|
||||
type Blocked = Arc<Mutex<HashSet<EndpointId>>>;
|
||||
|
||||
async fn member(
|
||||
discovery: &SharedMemoryDiscovery,
|
||||
plugins: &[Arc<Carrier>],
|
||||
blocked: &Blocked,
|
||||
) -> TestAgent {
|
||||
let plugins: Vec<_> = plugins.to_vec();
|
||||
let blocked = Arc::clone(blocked);
|
||||
TestAgent::spawn_with(
|
||||
move |config| {
|
||||
let config = config.with_unreachable_data_peers(blocked);
|
||||
plugins
|
||||
.into_iter()
|
||||
.fold(config, |config, plugin| config.with_plugin(plugin))
|
||||
},
|
||||
discovery,
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
async fn exchange(from: &SharedLink, to: &SharedLink, payload: &'static [u8]) {
|
||||
from.send(Bytes::from_static(payload)).unwrap();
|
||||
let got = tokio::time::timeout(tsunagi::testing::DEADLINE, to.recv())
|
||||
.await
|
||||
.expect("the datagram should arrive")
|
||||
.expect("the link stays open");
|
||||
assert_eq!(got, Bytes::from_static(payload));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_tunnel_crosses_a_member_that_shares_no_protocol_with_either_end_and_moves_to_a_direct_link_by_itself()
|
||||
{
|
||||
// A and B share one protocol and nothing else. C shares a different
|
||||
// one with each, so every link here is of a different kind and the
|
||||
// middle has never heard of what A and B speak. A and B cannot reach
|
||||
// each other directly, which is the case a blocked UDP path produces.
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("mixed-transports");
|
||||
|
||||
let blocked_a: Blocked = Arc::default();
|
||||
let blocked_b: Blocked = Arc::default();
|
||||
let a_shared = Carrier::new("shared");
|
||||
let b_shared = Carrier::new("shared");
|
||||
let a = member(
|
||||
&discovery,
|
||||
&[Carrier::new("a-and-c"), Arc::clone(&a_shared)],
|
||||
&blocked_a,
|
||||
)
|
||||
.await;
|
||||
let b = member(
|
||||
&discovery,
|
||||
&[Carrier::new("c-and-b"), Arc::clone(&b_shared)],
|
||||
&blocked_b,
|
||||
)
|
||||
.await;
|
||||
let c = member(
|
||||
&discovery,
|
||||
&[Carrier::new("a-and-c"), Carrier::new("c-and-b")],
|
||||
&Blocked::default(),
|
||||
)
|
||||
.await;
|
||||
let (a_id, b_id, c_id) = (
|
||||
a.agent.endpoint_id(),
|
||||
b.agent.endpoint_id(),
|
||||
c.agent.endpoint_id(),
|
||||
);
|
||||
blocked_a.lock().unwrap().insert(b_id);
|
||||
blocked_b.lock().unwrap().insert(a_id);
|
||||
|
||||
let network_id = a.agent.join_network(&name, &secret).await.unwrap();
|
||||
b.agent.join_network(&name, &secret).await.unwrap();
|
||||
c.agent.join_network(&name, &secret).await.unwrap();
|
||||
for agent in [&a, &b, &c] {
|
||||
wait_for_peers(&agent.agent, network_id, 2).await;
|
||||
}
|
||||
|
||||
// The tunnel comes up through C, and data goes both ways.
|
||||
let to_b = wait_until("a has a link to b", || async { a_shared.link(b_id) }).await;
|
||||
let to_a = wait_until("b has a link to a", || async { b_shared.link(a_id) }).await;
|
||||
exchange(&to_b, &to_a, b"a to b, sealed").await;
|
||||
exchange(&to_a, &to_b, b"b to a, sealed").await;
|
||||
|
||||
// The status says what is true: online, and through whom.
|
||||
let status = a.agent.network_status(network_id).await.unwrap();
|
||||
let b_peer = status.peers.iter().find(|p| p.endpoint_id == b_id).unwrap();
|
||||
assert_eq!(b_peer.data_path, DataPath::Relayed { hops: 2, via: c_id });
|
||||
let c_peer = status.peers.iter().find(|p| p.endpoint_id == c_id).unwrap();
|
||||
assert!(matches!(&c_peer.data_path, DataPath::Direct { transport } if transport == "a-and-c"));
|
||||
let b_member = wait_until("a lists b as a member", || async {
|
||||
let status = a.agent.network_status(network_id).await.ok()?;
|
||||
status.members.into_iter().find(|m| m.endpoint_id == b_id)
|
||||
})
|
||||
.await;
|
||||
assert!(b_member.online);
|
||||
let carried = c.agent.network_status(network_id).await.unwrap().relay;
|
||||
assert!(carried.forwarded >= 2, "{carried:?}");
|
||||
|
||||
// The direct path becomes possible: the same link carries on, now
|
||||
// without C, and nobody has to do anything about it.
|
||||
blocked_a.lock().unwrap().clear();
|
||||
blocked_b.lock().unwrap().clear();
|
||||
wait_until("a reaches b directly", || async {
|
||||
let status = a.agent.network_status(network_id).await.ok()?;
|
||||
let peer = status.peers.iter().find(|p| p.endpoint_id == b_id)?;
|
||||
matches!(&peer.data_path, DataPath::Direct { transport } if transport == "shared")
|
||||
.then_some(())
|
||||
})
|
||||
.await;
|
||||
let forwarded = c
|
||||
.agent
|
||||
.network_status(network_id)
|
||||
.await
|
||||
.unwrap()
|
||||
.relay
|
||||
.forwarded;
|
||||
exchange(&to_b, &to_a, b"direct now").await;
|
||||
assert!(
|
||||
Arc::ptr_eq(&to_b, &a_shared.link(b_id).unwrap()),
|
||||
"the tunnel's link is the one it always had"
|
||||
);
|
||||
assert_eq!(
|
||||
c.agent
|
||||
.network_status(network_id)
|
||||
.await
|
||||
.unwrap()
|
||||
.relay
|
||||
.forwarded,
|
||||
forwarded,
|
||||
"nothing goes through C any more"
|
||||
);
|
||||
|
||||
a.stop().await;
|
||||
b.stop().await;
|
||||
c.stop().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_name_belongs_to_the_member_that_claimed_it_first_on_every_member() {
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("first-claim");
|
||||
let spawn = |hostname: &'static str| {
|
||||
let discovery = discovery.clone();
|
||||
async move {
|
||||
TestAgent::spawn_with(move |config| config.with_hostname(hostname), &discovery)
|
||||
.await
|
||||
.unwrap()
|
||||
}
|
||||
};
|
||||
|
||||
let first = spawn("music").await;
|
||||
let network_id = first.agent.join_network(&name, &secret).await.unwrap();
|
||||
let first_id = first.agent.endpoint_id();
|
||||
wait_until("the first claim is signed", || async {
|
||||
let status = first.agent.network_status(network_id).await.ok()?;
|
||||
status
|
||||
.members
|
||||
.iter()
|
||||
.any(|m| m.endpoint_id == first_id && m.hostname.as_deref() == Some("music"))
|
||||
.then_some(())
|
||||
})
|
||||
.await;
|
||||
|
||||
// Whatever the ids turn out to be, the later one does not take it.
|
||||
let second = spawn("music").await;
|
||||
let third = spawn("studio").await;
|
||||
second.agent.join_network(&name, &secret).await.unwrap();
|
||||
third.agent.join_network(&name, &secret).await.unwrap();
|
||||
let second_id = second.agent.endpoint_id();
|
||||
for agent in [&first, &second, &third] {
|
||||
wait_for_peers(&agent.agent, network_id, 2).await;
|
||||
}
|
||||
|
||||
for (viewer, label) in [(&first, "first"), (&second, "second"), (&third, "third")] {
|
||||
let status = wait_until(&format!("{label} sees both claims"), || async {
|
||||
let status = viewer.agent.network_status(network_id).await.ok()?;
|
||||
// The roster is the authors of signed records, so a member
|
||||
// being in it means its claim has arrived.
|
||||
let both = [first_id, second_id]
|
||||
.iter()
|
||||
.all(|id| status.members.iter().any(|m| m.endpoint_id == *id));
|
||||
both.then_some(status)
|
||||
})
|
||||
.await;
|
||||
let holder = status
|
||||
.members
|
||||
.iter()
|
||||
.find(|m| m.endpoint_id == first_id)
|
||||
.unwrap();
|
||||
let loser = status
|
||||
.members
|
||||
.iter()
|
||||
.find(|m| m.endpoint_id == second_id)
|
||||
.unwrap();
|
||||
assert_eq!(holder.hostname.as_deref(), Some("music"), "{label}");
|
||||
assert_eq!(loser.hostname, None, "{label}: no name for the later one");
|
||||
let conflict = loser.hostname_conflict.as_ref().expect(label);
|
||||
assert_eq!(conflict.name, "music");
|
||||
assert_eq!(conflict.holder, first_id);
|
||||
assert!(holder.hostname_conflict.is_none(), "{label}");
|
||||
// Nothing about being second makes it less of a member.
|
||||
assert!(
|
||||
loser.online || viewer.agent.endpoint_id() == second_id,
|
||||
"{label}"
|
||||
);
|
||||
}
|
||||
|
||||
// The later one is told so itself, and keeps working.
|
||||
let own = second.agent.network_status(network_id).await.unwrap();
|
||||
let conflict = own.hostname_conflict.expect("told its name is taken");
|
||||
assert_eq!(
|
||||
(conflict.name.as_str(), conflict.holder),
|
||||
("music", first_id)
|
||||
);
|
||||
assert!(
|
||||
first
|
||||
.agent
|
||||
.network_status(network_id)
|
||||
.await
|
||||
.unwrap()
|
||||
.hostname_conflict
|
||||
.is_none()
|
||||
);
|
||||
|
||||
first.stop().await;
|
||||
second.stop().await;
|
||||
third.stop().await;
|
||||
}
|
||||
@@ -9,13 +9,17 @@ use std::collections::HashMap;
|
||||
|
||||
use eframe::egui;
|
||||
|
||||
use tsunagi::ipc::{NetworkReport, OverlayPeerReport};
|
||||
use tsunagi::ipc::{DataPathReport, NetworkReport, OverlayPeerReport};
|
||||
|
||||
use crate::agent::{AgentClient, Command};
|
||||
use crate::exit;
|
||||
use crate::format;
|
||||
use crate::stats::{Traffic, Unit};
|
||||
|
||||
/// A device that works but not as well as it could: reached through others,
|
||||
/// or with a name somebody else holds.
|
||||
const AMBER: egui::Color32 = egui::Color32::from_rgb(0xd9, 0xa4, 0x2c);
|
||||
|
||||
/// A readable title for a network's window.
|
||||
pub(crate) fn title(network: &NetworkReport) -> String {
|
||||
format!("tsunagi · {}", network.name)
|
||||
@@ -23,7 +27,15 @@ pub(crate) fn title(network: &NetworkReport) -> String {
|
||||
|
||||
/// One row of the device table, built from a tunnel and/or a signed member.
|
||||
struct Device<'a> {
|
||||
/// The agent's own answer to whether it can be reached by any path.
|
||||
online: bool,
|
||||
/// How packets reach it.
|
||||
path: Option<&'a DataPathReport>,
|
||||
/// What to call the member its traffic goes through, when it does.
|
||||
via_name: Option<String>,
|
||||
/// The name it claimed that another member holds, and what to call that
|
||||
/// member.
|
||||
name_taken: Option<(&'a str, String)>,
|
||||
/// The last hostname seen for it, kept while it is offline.
|
||||
hostname: Option<&'a str>,
|
||||
/// What clicking the hostname copies: `hostname.network`, so it resolves.
|
||||
@@ -93,6 +105,13 @@ fn name_cell(ui: &mut egui::Ui, device: &Device<'_>) {
|
||||
}
|
||||
_ => format::copy_field(ui, &short_id(device.id), device.id),
|
||||
}
|
||||
if let Some((name, holder)) = &device.name_taken {
|
||||
ui.label(egui::RichText::new(egui_phosphor::regular::WARNING).color(AMBER))
|
||||
.on_hover_text(format!(
|
||||
"`{name}` was claimed first by {holder}. This device works by address, \
|
||||
but nobody finds it by that name."
|
||||
));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@@ -276,6 +295,11 @@ fn known_ids<'a>(network: &'a NetworkReport, own_id: &str) -> Vec<&'a str> {
|
||||
ids
|
||||
}
|
||||
|
||||
/// How many other devices can be reached now, by any path.
|
||||
pub(crate) fn online_count(network: &NetworkReport, own_id: &str) -> usize {
|
||||
devices(network, own_id).iter().filter(|d| d.online).count()
|
||||
}
|
||||
|
||||
/// How many other devices this network knows of, connected or not.
|
||||
pub(crate) fn known_count(network: &NetworkReport, own_id: &str) -> usize {
|
||||
known_ids(network, own_id).len()
|
||||
@@ -310,10 +334,29 @@ fn devices<'a>(network: &'a NetworkReport, own_id: &str) -> Vec<Device<'a>> {
|
||||
.map(|id| {
|
||||
let overlay = overlay_peers.iter().find(|p| p.endpoint_id == id);
|
||||
let member = network.members.iter().find(|m| m.endpoint_id == id);
|
||||
let online = overlay.is_some_and(|p| p.handshake_secs_ago.is_some());
|
||||
let online = member.is_some_and(|m| m.online)
|
||||
|| network.peers.iter().any(|p| p.endpoint_id == id);
|
||||
let path = network
|
||||
.peers
|
||||
.iter()
|
||||
.find(|p| p.endpoint_id == id)
|
||||
.map(|p| &p.path)
|
||||
.or_else(|| member.map(|m| &m.path));
|
||||
let via_name = match path {
|
||||
Some(DataPathReport::Relayed { via, .. }) => Some(exit::name_of(network, via)),
|
||||
_ => None,
|
||||
};
|
||||
let name_taken = member.and_then(|m| {
|
||||
m.hostname_conflict
|
||||
.as_ref()
|
||||
.map(|c| (c.name.as_str(), exit::name_of(network, &c.holder)))
|
||||
});
|
||||
let hostname = hostnames.get(id).copied();
|
||||
Device {
|
||||
online,
|
||||
path,
|
||||
via_name,
|
||||
name_taken,
|
||||
hostname,
|
||||
copy: hostname.map(|h| format!("{h}.{}", network.name)),
|
||||
id,
|
||||
@@ -463,15 +506,7 @@ fn device_table(
|
||||
.spacing([10.0, 4.0])
|
||||
.show(ui, |ui| {
|
||||
for (column, header) in [
|
||||
"device",
|
||||
"exit",
|
||||
"proto",
|
||||
"address",
|
||||
"handshake",
|
||||
"pkts",
|
||||
"bytes",
|
||||
"rate",
|
||||
"plot",
|
||||
"device", "exit", "proto", "address", "link", "pkts", "bytes", "rate", "plot",
|
||||
]
|
||||
.into_iter()
|
||||
.enumerate()
|
||||
@@ -496,13 +531,7 @@ fn device_table(
|
||||
ui.weak("—");
|
||||
}
|
||||
});
|
||||
cell(ui, 4, &mut widths, |ui| {
|
||||
ui.label(
|
||||
device
|
||||
.handshake_secs
|
||||
.map_or_else(|| "offline".to_string(), format::age),
|
||||
);
|
||||
});
|
||||
cell(ui, 4, &mut widths, |ui| link_cell(ui, device));
|
||||
|
||||
match device.overlay.filter(|_| device.online) {
|
||||
Some(peer) => {
|
||||
@@ -580,6 +609,121 @@ fn device_table(
|
||||
ui.memory_mut(|m| m.data.insert_temp(widths_id, widths));
|
||||
}
|
||||
|
||||
/// How a device is reached, and how long its tunnel has been up.
|
||||
///
|
||||
/// A device reached through others is online, over a longer path, and says
|
||||
/// so: the colour is the only difference in how it is treated.
|
||||
fn link_cell(ui: &mut egui::Ui, device: &Device<'_>) {
|
||||
if !device.online {
|
||||
ui.label("offline");
|
||||
return;
|
||||
}
|
||||
let (path, good) = match device.path {
|
||||
Some(DataPathReport::Direct { transport }) => (format!("direct · {transport}"), true),
|
||||
Some(DataPathReport::Relayed { hops, .. }) => (
|
||||
format!(
|
||||
"via {} · {hops} hops",
|
||||
device.via_name.as_deref().unwrap_or("another device")
|
||||
),
|
||||
false,
|
||||
),
|
||||
Some(DataPathReport::None) | None => ("no data path".to_string(), false),
|
||||
};
|
||||
let age = device.handshake_secs.map(format::age);
|
||||
let text = match age {
|
||||
Some(age) => format!("{path} · {age}"),
|
||||
None => path,
|
||||
};
|
||||
let response = if good {
|
||||
ui.label(text)
|
||||
} else {
|
||||
ui.label(egui::RichText::new(text).color(AMBER))
|
||||
};
|
||||
if !good {
|
||||
response.on_hover_text(match device.path {
|
||||
Some(DataPathReport::Relayed { .. }) => {
|
||||
"no direct link yet: traffic goes through another device, and moves to a \
|
||||
direct link by itself as soon as one comes up"
|
||||
}
|
||||
_ => "connected, but no path carries packets yet",
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
fn yes_no(value: bool) -> &'static str {
|
||||
if value { "yes" } else { "no" }
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
|
||||
use super::*;
|
||||
use tsunagi::ipc::{HostnameConflictReport, MemberReport};
|
||||
|
||||
fn member(id: &str, online: bool, path: DataPathReport) -> MemberReport {
|
||||
MemberReport {
|
||||
endpoint_id: id.into(),
|
||||
online,
|
||||
path,
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_device_reached_through_another_is_online_whatever_its_tunnel_says() {
|
||||
// The tray used to call a device online only while its tunnel had
|
||||
// handshaken, which disagreed with the command line about the same
|
||||
// device. Both now read the agent's answer.
|
||||
let network = NetworkReport {
|
||||
members: vec![
|
||||
member("own", true, DataPathReport::None),
|
||||
member(
|
||||
"far",
|
||||
true,
|
||||
DataPathReport::Relayed {
|
||||
hops: 2,
|
||||
via: "near".into(),
|
||||
},
|
||||
),
|
||||
member("away", false, DataPathReport::None),
|
||||
],
|
||||
..Default::default()
|
||||
};
|
||||
let all = devices(&network, "own");
|
||||
let far = all.iter().find(|d| d.id == "far").unwrap();
|
||||
assert!(far.online);
|
||||
assert!(matches!(
|
||||
far.path,
|
||||
Some(DataPathReport::Relayed { hops: 2, .. })
|
||||
));
|
||||
assert!(!all.iter().find(|d| d.id == "away").unwrap().online);
|
||||
assert_eq!(online_count(&network, "own"), 1);
|
||||
assert_eq!(known_count(&network, "own"), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_name_somebody_else_holds_is_carried_to_the_row() {
|
||||
let mut taken = member("late", true, DataPathReport::None);
|
||||
taken.hostname_conflict = Some(HostnameConflictReport {
|
||||
name: "music".into(),
|
||||
holder: "early".into(),
|
||||
});
|
||||
let network = NetworkReport {
|
||||
members: vec![member("early", true, DataPathReport::None), taken],
|
||||
..Default::default()
|
||||
};
|
||||
let all = devices(&network, "own");
|
||||
let late = all.iter().find(|d| d.id == "late").unwrap();
|
||||
assert_eq!(
|
||||
late.name_taken.as_ref().map(|(name, _)| *name),
|
||||
Some("music")
|
||||
);
|
||||
assert!(
|
||||
all.iter()
|
||||
.find(|d| d.id == "early")
|
||||
.unwrap()
|
||||
.name_taken
|
||||
.is_none()
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -243,8 +243,8 @@ fn draw_network(
|
||||
}
|
||||
ui.weak("peers");
|
||||
ui.label(format!(
|
||||
"{} connected · {} known",
|
||||
network.peers.len(),
|
||||
"{} online · {} known",
|
||||
devices::online_count(network, own_id),
|
||||
devices::known_count(network, own_id)
|
||||
));
|
||||
ui.end_row();
|
||||
|
||||
@@ -42,6 +42,7 @@ fn source(agent: Agent, plugin: Arc<WireguardPlugin>) -> Arc<dyn tsunagi::ipc::R
|
||||
transport: format!("{:?}", peer.transport),
|
||||
rtt_ms: peer.rtt.map(|rtt| rtt.as_millis() as u64),
|
||||
exit_node: peer.exit_node,
|
||||
path: Default::default(),
|
||||
})
|
||||
.collect(),
|
||||
overlay: plugin.overview(net.network_id).map(|view| {
|
||||
|
||||
+180
-7
@@ -3156,7 +3156,18 @@ fn dns_absent_section() -> report::Section {
|
||||
struct MemberRow<'a> {
|
||||
endpoint_id: &'a str,
|
||||
hostname: Option<&'a str>,
|
||||
/// `Some` exactly when there is an authenticated session right now.
|
||||
/// Whether it can be reached now, by any path: the agent's own answer,
|
||||
/// the same one every other client gives.
|
||||
online: bool,
|
||||
/// How packets reach it.
|
||||
path: Option<&'a tsunagi::ipc::DataPathReport>,
|
||||
/// What to call the member the first of those links goes to.
|
||||
via_label: Option<String>,
|
||||
/// The name it claimed that another member holds, and what to call the
|
||||
/// holder.
|
||||
name_taken: Option<(&'a str, String)>,
|
||||
/// `Some` exactly when there is an authenticated session right now: how
|
||||
/// that control connection reaches the peer.
|
||||
transport: Option<&'a str>,
|
||||
rtt_ms: Option<u64>,
|
||||
overlay_address_v4: Option<&'a str>,
|
||||
@@ -3170,7 +3181,7 @@ struct MemberRow<'a> {
|
||||
|
||||
impl MemberRow<'_> {
|
||||
fn online(&self) -> bool {
|
||||
self.transport.is_some()
|
||||
self.online
|
||||
}
|
||||
|
||||
/// What to call it: the name it announced — the last one, when it is
|
||||
@@ -3194,6 +3205,10 @@ fn member_rows<'a>(network: &'a tsunagi::ipc::NetworkReport, own_id: &str) -> Ve
|
||||
rows.entry(id).or_insert_with(|| MemberRow {
|
||||
endpoint_id: id,
|
||||
hostname: None,
|
||||
online: false,
|
||||
path: None,
|
||||
via_label: None,
|
||||
name_taken: None,
|
||||
transport: None,
|
||||
rtt_ms: None,
|
||||
overlay_address_v4: None,
|
||||
@@ -3210,11 +3225,19 @@ fn member_rows<'a>(network: &'a tsunagi::ipc::NetworkReport, own_id: &str) -> Ve
|
||||
row.overlay_address_v4 = member.overlay_address_v4.as_deref();
|
||||
row.failed_dials = member.failed_dials;
|
||||
row.hostname = member.hostname.as_deref();
|
||||
row.online = member.online;
|
||||
row.path = Some(&member.path);
|
||||
row.name_taken = member
|
||||
.hostname_conflict
|
||||
.as_ref()
|
||||
.map(|conflict| (conflict.name.as_str(), exit_name(network, &conflict.holder)));
|
||||
}
|
||||
for peer in &network.peers {
|
||||
let row = entry(&mut rows, &peer.endpoint_id);
|
||||
row.hostname = peer.hostname.as_deref().or(row.hostname);
|
||||
row.exit_node = peer.exit_node;
|
||||
row.online = true;
|
||||
row.path = Some(&peer.path);
|
||||
row.transport = Some(&peer.transport);
|
||||
row.rtt_ms = peer.rtt_ms;
|
||||
}
|
||||
@@ -3230,6 +3253,9 @@ fn member_rows<'a>(network: &'a tsunagi::ipc::NetworkReport, own_id: &str) -> Ve
|
||||
|
||||
for row in rows.values_mut() {
|
||||
row.exit_in_use = network.exit.via.as_deref() == Some(row.endpoint_id);
|
||||
if let Some(tsunagi::ipc::DataPathReport::Relayed { via, .. }) = row.path {
|
||||
row.via_label = Some(exit_name(network, via));
|
||||
}
|
||||
}
|
||||
|
||||
// This agent is in the roster too — it signs claims like everyone else —
|
||||
@@ -3285,6 +3311,23 @@ fn network_section(
|
||||
),
|
||||
);
|
||||
}
|
||||
if let Some(taken) = &network.hostname_conflict {
|
||||
section.push(
|
||||
Row::new(
|
||||
Health::Degraded,
|
||||
"hostname",
|
||||
format!(
|
||||
"`{}` is held by {}, which claimed it first",
|
||||
taken.name,
|
||||
exit_name(network, &taken.holder)
|
||||
),
|
||||
)
|
||||
.with_note(
|
||||
"this device works by address, but nobody finds it by that name. \
|
||||
Pick another with `tsng id hostname <name>`.",
|
||||
),
|
||||
);
|
||||
}
|
||||
if let Some(conflict) = &network.range_conflict {
|
||||
section.push(
|
||||
Row::new(
|
||||
@@ -3547,7 +3590,10 @@ fn exit_name(network: &tsunagi::ipc::NetworkReport, id: &str) -> String {
|
||||
fn member_row(row: &MemberRow<'_>) -> report::Row {
|
||||
use report::{Health, Row};
|
||||
|
||||
let Some(transport) = row.transport else {
|
||||
let taken = row.name_taken.as_ref().map(|(name, holder)| {
|
||||
format!(" · name `{name}` is held by {holder}: reachable by address only")
|
||||
});
|
||||
if !row.online {
|
||||
// Away. Not a fault of this agent, and in a mesh of laptops it is the
|
||||
// ordinary condition, so it is stated rather than flagged.
|
||||
let mut detail = "offline".to_string();
|
||||
@@ -3557,6 +3603,7 @@ fn member_row(row: &MemberRow<'_>) -> report::Row {
|
||||
if row.exit_in_use {
|
||||
detail.push_str(" · your exit node");
|
||||
}
|
||||
detail.extend(taken);
|
||||
let out = Row::new(Health::Info, row.label(), detail);
|
||||
return if row.failed_dials > 0 {
|
||||
// Attributed to the member it concerns, rather than left as a
|
||||
@@ -3568,17 +3615,36 @@ fn member_row(row: &MemberRow<'_>) -> report::Row {
|
||||
} else {
|
||||
out
|
||||
};
|
||||
};
|
||||
}
|
||||
|
||||
let direct = transport.eq_ignore_ascii_case("direct");
|
||||
use tsunagi::ipc::DataPathReport;
|
||||
let direct_control = row
|
||||
.transport
|
||||
.is_none_or(|transport| transport.eq_ignore_ascii_case("direct"));
|
||||
let tunnel_up = row.tunnel.is_some_and(|tunnel| tunnel.is_up());
|
||||
let has_overlay = row.tunnel.is_some();
|
||||
// Online is online: a member reached through others is as much here as
|
||||
// one reached directly, and is said to be reached that way.
|
||||
let (path, data_path_good) = match row.path {
|
||||
Some(DataPathReport::Direct { transport }) => (format!("direct {transport}"), true),
|
||||
Some(DataPathReport::Relayed { hops, .. }) => (
|
||||
format!(
|
||||
"relayed via {} ({hops} hops)",
|
||||
row.via_label.as_deref().unwrap_or("another member")
|
||||
),
|
||||
false,
|
||||
),
|
||||
Some(DataPathReport::None) | None => ("connected, no data path yet".to_string(), false),
|
||||
};
|
||||
|
||||
let mut detail = String::new();
|
||||
if let Some(v4) = row.overlay_address_v4 {
|
||||
detail.push_str(&format!("{v4:<15} "));
|
||||
}
|
||||
detail.push_str(&transport.to_lowercase());
|
||||
detail.push_str(&path);
|
||||
if let Some(transport) = row.transport {
|
||||
detail.push_str(&format!(" · control {}", transport.to_lowercase()));
|
||||
}
|
||||
if let Some(rtt) = row.rtt_ms {
|
||||
detail.push_str(&format!(" rtt {rtt}ms"));
|
||||
}
|
||||
@@ -3587,8 +3653,9 @@ fn member_row(row: &MemberRow<'_>) -> report::Row {
|
||||
} else if row.exit_node {
|
||||
detail.push_str(" · exit node");
|
||||
}
|
||||
detail.extend(taken);
|
||||
|
||||
let health = if !direct || (has_overlay && !tunnel_up) {
|
||||
let health = if !data_path_good || !direct_control || (has_overlay && !tunnel_up) {
|
||||
Health::Degraded
|
||||
} else {
|
||||
Health::Good
|
||||
@@ -4890,6 +4957,7 @@ async fn build_report(
|
||||
transport: peer.transport.to_string(),
|
||||
rtt_ms: peer.rtt.map(|rtt| rtt.as_millis() as u64),
|
||||
exit_node: peer.exit_node,
|
||||
path: data_path_report(&peer.data_path),
|
||||
})
|
||||
.collect(),
|
||||
members: network
|
||||
@@ -4911,6 +4979,9 @@ async fn build_report(
|
||||
.last_hostname
|
||||
.clone()
|
||||
.or_else(|| member.hostname.clone()),
|
||||
online: member.online,
|
||||
path: data_path_report(&member.data_path),
|
||||
hostname_conflict: member.hostname_conflict.as_ref().map(conflict_report),
|
||||
})
|
||||
.collect(),
|
||||
range: network.range.map(|range| range.to_string()),
|
||||
@@ -4922,6 +4993,7 @@ async fn build_report(
|
||||
network.metrics.control_messages_received,
|
||||
),
|
||||
overlay,
|
||||
hostname_conflict: network.hostname_conflict.as_ref().map(conflict_report),
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
@@ -4955,6 +5027,28 @@ async fn build_report(
|
||||
}
|
||||
}
|
||||
|
||||
fn data_path_report(path: &tsunagi::agent::DataPath) -> tsunagi::ipc::DataPathReport {
|
||||
match path {
|
||||
tsunagi::agent::DataPath::None => tsunagi::ipc::DataPathReport::None,
|
||||
tsunagi::agent::DataPath::Direct { transport } => tsunagi::ipc::DataPathReport::Direct {
|
||||
transport: transport.clone(),
|
||||
},
|
||||
tsunagi::agent::DataPath::Relayed { hops, via } => tsunagi::ipc::DataPathReport::Relayed {
|
||||
hops: *hops,
|
||||
via: via.to_string(),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
fn conflict_report(
|
||||
conflict: &tsunagi::agent::HostnameConflict,
|
||||
) -> tsunagi::ipc::HostnameConflictReport {
|
||||
tsunagi::ipc::HostnameConflictReport {
|
||||
name: conflict.name.clone(),
|
||||
holder: conflict.holder.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
/// One network's exit-node settings, as the control interface reports them.
|
||||
fn exit_report(network: &tsunagi::agent::NetworkStatus) -> tsunagi::ipc::ExitReport {
|
||||
let rules = |report: &Option<tsunagi::overlay::RuleSetReport>| {
|
||||
@@ -5260,6 +5354,7 @@ mod status_tests {
|
||||
fn network_after_a_peer_returned() -> NetworkReport {
|
||||
NetworkReport {
|
||||
exit: Default::default(),
|
||||
hostname_conflict: None,
|
||||
broadcast: true,
|
||||
name: "LAB".into(),
|
||||
network_id: "xa7gyz".into(),
|
||||
@@ -5274,6 +5369,9 @@ mod status_tests {
|
||||
transport: "direct".into(),
|
||||
rtt_ms: Some(24),
|
||||
exit_node: false,
|
||||
path: tsunagi::ipc::DataPathReport::Direct {
|
||||
transport: "wg".into(),
|
||||
},
|
||||
}],
|
||||
members: vec![
|
||||
MemberReport {
|
||||
@@ -5281,12 +5379,18 @@ mod status_tests {
|
||||
overlay_address_v4: Some("10.13.37.69".into()),
|
||||
failed_dials: 0,
|
||||
hostname: None,
|
||||
..Default::default()
|
||||
},
|
||||
MemberReport {
|
||||
endpoint_id: ONLINE.into(),
|
||||
overlay_address_v4: Some("10.13.37.237".into()),
|
||||
failed_dials: 0,
|
||||
hostname: None,
|
||||
online: true,
|
||||
path: tsunagi::ipc::DataPathReport::Direct {
|
||||
transport: "wg".into(),
|
||||
},
|
||||
..Default::default()
|
||||
},
|
||||
],
|
||||
range: Some("10.13.37.0/24".into()),
|
||||
@@ -5528,6 +5632,7 @@ mod status_tests {
|
||||
overlay_address_v4: Some("10.13.37.99".into()),
|
||||
failed_dials: 9,
|
||||
hostname: None,
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
let rows = member_rows(&network, OWN);
|
||||
@@ -5648,6 +5753,7 @@ mod status_tests {
|
||||
overlay_address_v4: Some("10.13.37.99".into()),
|
||||
failed_dials: 0,
|
||||
hostname: Some("laptop".into()),
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
let rows = member_rows(&network, OWN);
|
||||
@@ -5668,6 +5774,7 @@ mod status_tests {
|
||||
overlay_address_v4: Some("10.13.37.99".into()),
|
||||
failed_dials: 9,
|
||||
hostname: None,
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
let mut out = report::Report::new();
|
||||
@@ -5681,6 +5788,69 @@ mod status_tests {
|
||||
assert!(text.contains("1 of 2 online"), "{text}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_member_reached_through_another_is_online_and_says_through_whom() {
|
||||
// No direct link, a path through somebody else: online all the same,
|
||||
// and said to be, rather than being shown as away or as a fault.
|
||||
let mut network = network_after_a_peer_returned();
|
||||
network.members.push(MemberReport {
|
||||
endpoint_id: AWAY.into(),
|
||||
overlay_address_v4: Some("10.13.37.99".into()),
|
||||
hostname: Some("laptop".into()),
|
||||
online: true,
|
||||
path: tsunagi::ipc::DataPathReport::Relayed {
|
||||
hops: 2,
|
||||
via: ONLINE.into(),
|
||||
},
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
let rows = member_rows(&network, OWN);
|
||||
let away = rows.iter().find(|row| row.endpoint_id == AWAY).unwrap();
|
||||
assert!(away.online());
|
||||
|
||||
let mut out = report::Report::new();
|
||||
out.push(network_section(&network, OWN, false));
|
||||
let text = out.render(false);
|
||||
assert!(text.contains("2 of 2 online"), "{text}");
|
||||
assert!(
|
||||
text.contains("relayed via music (2 hops)"),
|
||||
"names the member it goes through: {text}"
|
||||
);
|
||||
assert!(!text.contains("offline"), "{text}");
|
||||
assert_eq!(out.worst(), Health::Degraded, "{text}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_name_somebody_else_holds_is_said_on_both_the_member_and_this_device() {
|
||||
let mut network = network_after_a_peer_returned();
|
||||
network.members[1].hostname_conflict = Some(tsunagi::ipc::HostnameConflictReport {
|
||||
name: "music".into(),
|
||||
holder: OWN.into(),
|
||||
});
|
||||
let text = {
|
||||
let mut out = report::Report::new();
|
||||
out.push(network_section(&network, OWN, false));
|
||||
out.render(false)
|
||||
};
|
||||
assert!(
|
||||
text.contains("name `music` is held by") && text.contains("reachable by address only"),
|
||||
"{text}"
|
||||
);
|
||||
|
||||
let mut network = network_after_a_peer_returned();
|
||||
network.hostname_conflict = Some(tsunagi::ipc::HostnameConflictReport {
|
||||
name: "studio".into(),
|
||||
holder: ONLINE.into(),
|
||||
});
|
||||
let mut out = report::Report::new();
|
||||
out.push(network_section(&network, OWN, false));
|
||||
let text = out.render(false);
|
||||
assert!(text.contains("`studio` is held by music"), "{text}");
|
||||
assert!(text.contains("tsng id hostname"), "{text}");
|
||||
assert_eq!(out.worst(), Health::Degraded, "{text}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_relayed_peer_is_graded_as_degraded_quality() {
|
||||
let mut network = network_after_a_peer_returned();
|
||||
@@ -5710,6 +5880,9 @@ mod status_tests {
|
||||
// With a peer connected the same counter is just history.
|
||||
let mut network = network_after_a_peer_returned();
|
||||
network.peers.clear();
|
||||
for member in &mut network.members {
|
||||
member.online = false;
|
||||
}
|
||||
network.overlay = Some(overlay(Vec::new()));
|
||||
network.handshake_failures = 4;
|
||||
|
||||
|
||||
+29
-15
@@ -9,7 +9,7 @@ Two versions exist and are independent:
|
||||
|
||||
- **Identity scheme**, `tsunagi-network-id-v1`. Frozen. Changing it creates a
|
||||
different network space for the same name and secret.
|
||||
- **Control protocol**, ALPN `tsunagi/ctrl/4`, `PROTOCOL_VERSION = 4`.
|
||||
- **Control protocol**, ALPN `tsunagi/ctrl/5`, `PROTOCOL_VERSION = 5`.
|
||||
|
||||
Upgrading the crate or bumping the control protocol must never change an
|
||||
existing `NetworkId`.
|
||||
@@ -135,7 +135,7 @@ event does not report a network id.
|
||||
## The data plane protocol
|
||||
|
||||
IP plugin packets never travel on a control connection. They use their own
|
||||
ALPN, `tsunagi/data/4`, on their own iroh connection:
|
||||
ALPN, `tsunagi/data/5`, on their own iroh connection:
|
||||
|
||||
```text
|
||||
initiator -> responder : (the same membership handshake as above)
|
||||
@@ -172,32 +172,46 @@ holding up subsequent packets. Packet IDs are scoped to a QUIC connection;
|
||||
reassembly is created only after the membership handshake. All limits live in
|
||||
`config.rs`. There is no transport-layer retransmission added by this framing.
|
||||
|
||||
Version 4 adds a routing envelope inside the transport fragmentation framing:
|
||||
Version 4 added a routing envelope inside the transport fragmentation framing,
|
||||
and version 5 added the end-to-end protocol to it:
|
||||
|
||||
| field | bytes | encoding |
|
||||
|---|---:|---|
|
||||
| envelope version | 1 | `1` |
|
||||
| envelope version | 1 | `2` |
|
||||
| remaining hops | 1 | initially `16`; transit decrements, drops at `1` |
|
||||
| source peer | 32 | original endpoint public key |
|
||||
| destination peer | 32 | final endpoint public key |
|
||||
| flow id | 8 | big endian opaque stable hash supplied by the plugin |
|
||||
| protocol | 8 | big endian FNV-1a of the end-to-end protocol id (`wg`, `tcp-tls`, ...) |
|
||||
| payload | remaining | end-to-end encrypted plugin bytes |
|
||||
|
||||
The fixed header is 74 bytes. Zero/oversized hop limits, unknown versions,
|
||||
The fixed header is 82 bytes. Zero/oversized hop limits, unknown versions,
|
||||
short frames, oversized datagrams and unknown sources are rejected. A frame
|
||||
addressed here is delivered to its source's protocol inbox; transit goes straight
|
||||
to its next transport. The network and protocol are bound to the authenticated
|
||||
transport connection, not supplied by the frame. Intermediate nodes cannot
|
||||
decrypt or authenticate the inner WireGuard payload; the destination does that.
|
||||
Flow ids are routing hints, not authorization proofs.
|
||||
addressed here is delivered to the inbox of its source and protocol; transit
|
||||
goes straight to its next transport, whatever kind of transport that is. The
|
||||
network is bound to the authenticated transport connection, not supplied by
|
||||
the frame. The protocol names which end-to-end tunnel the payload belongs to
|
||||
and is independent of the links it crosses: a datagram of one protocol can
|
||||
travel over a link opened for another, and a middle member needs no protocol
|
||||
in common with either end. Intermediate nodes cannot decrypt or authenticate
|
||||
the inner payload; the destination's tunnel does that, so a frame naming the
|
||||
wrong protocol or a forged source is simply discarded there. Flow ids are
|
||||
routing hints, not authorization proofs.
|
||||
|
||||
Control ALPN 3 carries `Reachable { links: [{ peer, protocol }] }`. Each row
|
||||
belongs to the authenticated sender and is replaced atomically, expires after
|
||||
90 seconds, and is withdrawn on session closure. Only compatible authenticated
|
||||
members enter a protocol's graph; local edges always come from actual links.
|
||||
`Reachable { links: [{ peer, protocol }] }` lists the sender's direct links;
|
||||
`protocol` is the kind of transport the link is, for diagnostics, and routing
|
||||
ignores it. Each row belongs to the authenticated sender and is replaced
|
||||
atomically, expires after 90 seconds, and is withdrawn on session closure. Only
|
||||
authenticated members enter the graph; local edges always come from actual
|
||||
links.
|
||||
Announcements refresh on the existing control maintenance interval and are
|
||||
sent immediately on link changes. Topology stays volatile, outside signed state.
|
||||
|
||||
Version 5 also adds `hostname_since` to a signed hostname claim (signing domain
|
||||
`tsunagi-signed-record-v3`): a name belongs to whoever claimed it first. Stored
|
||||
records signed before are discarded on upgrade and re-published by their
|
||||
authors.
|
||||
|
||||
Upgrade both endpoints and intermediate peers together. Older control and data
|
||||
ALPNs cannot interoperate. The network secret, device identity, identity
|
||||
derivation, transcript encoding and stored network configuration are unchanged.
|
||||
@@ -220,7 +234,7 @@ not affect other networks.
|
||||
| `Pong { seq, payload }` | the echoed reply |
|
||||
| `State { records }` | a snapshot of signed records, merged into what the receiver holds |
|
||||
| `Peers { peers }` | unverified member address hints, requiring their own handshake |
|
||||
| `Reachable { links }` | sender's current direct data links, scoped by protocol |
|
||||
| `Reachable { links }` | sender's current direct data links, by kind of transport |
|
||||
| `Bye { reason }` | graceful goodbye; not a revocation of anything |
|
||||
|
||||
`broadcast` is a per-network local opt-in, enabled by default and treated as
|
||||
|
||||
+39
-9
@@ -15,12 +15,27 @@ the default 1280-byte interface MTU also works on small QUIC paths.
|
||||
## Graph and lifecycle
|
||||
|
||||
Each authenticated control session advertises that peer's own direct data
|
||||
links, including the plugin protocol. Tables are separate for each network
|
||||
and protocol; only members offering a matching enabled protocol version enter
|
||||
the graph. A snapshot refreshes on the maintenance interval, expires after
|
||||
90 seconds, and is withdrawn when the session ends. Link arrival, closure and
|
||||
changed announcements rebuild routes immediately. Unchanged announcements
|
||||
refresh their age without republishing the forwarding snapshot.
|
||||
links. There is one table for each network, whatever protocols run over it: a
|
||||
link is a way to reach a neighbour, and a protocol is an end-to-end tunnel to a
|
||||
peer. The envelope names the protocol, so the two are independent. A neighbour
|
||||
reached over several kinds of transport (UDP, TCP, QUIC datagrams) is sent to
|
||||
over the best open one, in the order the protocols were configured, and over
|
||||
the next if that one fails or cannot carry the frame. Every authenticated
|
||||
member enters the graph; a member in the middle needs no protocol in common
|
||||
with either end, only a link to each. A snapshot refreshes on the maintenance
|
||||
interval, expires after 90 seconds, and is withdrawn when the session ends.
|
||||
Link arrival, closure and changed announcements rebuild routes immediately.
|
||||
Unchanged announcements refresh their age without republishing the forwarding
|
||||
snapshot.
|
||||
|
||||
Which protocol a pair of members runs their tunnel in is settled without
|
||||
asking: the member with the lower endpoint id decides, by the order it
|
||||
announces its protocols in, among those both have enabled at the same wire
|
||||
version. Both ends read the same answer from what they already hold. The
|
||||
tunnel is created as soon as the peer is reachable by any path, and it does not
|
||||
change when a direct link comes or goes underneath it: as soon as a direct
|
||||
link comes up it is shorter than the path through others, and traffic moves to
|
||||
it with nothing to redo. The user sees only a lower round-trip time.
|
||||
|
||||
Breadth-first search computes the shortest directed paths and every equal-cost
|
||||
first hop in sorted order. An actual direct link always wins. The first row is
|
||||
@@ -30,14 +45,29 @@ loops if different nodes have not yet received the same topology update.
|
||||
|
||||
The existing control plane still forms authenticated pairwise sessions.
|
||||
Routing provides multihop **data** paths among these members; it does not add
|
||||
control-plane flooding or carry control sessions through the data plane.
|
||||
control-plane flooding or carry control sessions through the data plane, so a
|
||||
member that has no control session with another has no data path to it either.
|
||||
Announcements are volatile, not durable membership or availability guarantees.
|
||||
|
||||
## What "online" means
|
||||
|
||||
One definition, computed by the agent and reported identically by `tsng status`
|
||||
and the tray: a member is online when this agent can reach it by any path — a
|
||||
live authenticated control session (QUIC keep-alive every 5 seconds, so a
|
||||
vanished peer is noticed within about 15), or a data path to it through other
|
||||
members. It is never inferred from ping, which a firewall may block, or from a
|
||||
tunnel's last handshake, which says only when traffic last flowed.
|
||||
|
||||
How it is reached is reported separately: `direct` over a named transport,
|
||||
`relayed` through a named member over N hops, or no data path yet while one is
|
||||
being set up. Being relayed is not being degraded in any way but distance, and
|
||||
is shown so it is not a surprise.
|
||||
|
||||
## Packet path
|
||||
|
||||
The control loop binds next hops to transport handles in an immutable table
|
||||
and publishes it through `ArcSwap`. Readers do not acquire the topology mutex.
|
||||
The transit routine validates the fixed 74-byte envelope, looks up the source
|
||||
The transit routine validates the fixed 82-byte envelope, looks up the source
|
||||
and destination, selects a cached next hop, decrements one byte, and calls the
|
||||
transport's synchronous datagram send. It never searches the graph, formats an
|
||||
endpoint string, validates an Ed25519 key, decrypts, or touches TUN.
|
||||
@@ -86,5 +116,5 @@ counts sends without storing frames. Results are CPU forwarding cost, **not**
|
||||
end-to-end network latency or VPN throughput; encryption, fragmentation,
|
||||
sockets, congestion and scheduling contribute separately.
|
||||
|
||||
Wire compatibility: control ALPN `tsunagi/ctrl/4`, data ALPN `tsunagi/data/4`.
|
||||
Wire compatibility: control ALPN `tsunagi/ctrl/5`, data ALPN `tsunagi/data/5`.
|
||||
Upgrade every participant together; saved identities and network state persist.
|
||||
+19
-1
@@ -43,7 +43,9 @@ document trusted merely because the neighbour who forwarded it signed it.
|
||||
- Duplicates do not change the result.
|
||||
- Two conflicting signed records at the same version from the same author need
|
||||
explicit handling; they are not resolved by luck.
|
||||
- Neither arrival order nor system clocks decide a winner.
|
||||
- Neither arrival order nor the receiving replica's clock decides a winner. The
|
||||
one thing an author's own clock is used for is the time it first claimed a
|
||||
hostname, signed into its record (see below).
|
||||
- Compaction must not drop what is needed to stop revoked records being
|
||||
resurrected.
|
||||
|
||||
@@ -53,6 +55,22 @@ A hostname is a mutable binding to a persistent author, not an identity. A
|
||||
rename must be a signed record that revokes the specific old binding and
|
||||
announces the new one, ideally atomically in one record.
|
||||
|
||||
A hostname belongs to the member that claimed it **first**. Each claim carries
|
||||
`hostname_since`, the author's own time in milliseconds when it first claimed
|
||||
that exact name, signed with the rest of the record and kept across every later
|
||||
version that keeps the name; renaming starts it afresh. Replicas order claims
|
||||
by `(hostname_since, author id)`, a pure function of the records, so every
|
||||
replica names the same holder whatever order the records arrived in and there
|
||||
is no race to lose. A member that arrives later with a name already held keeps
|
||||
working by address and keeps its claim, but is not found by that name on any
|
||||
member; it is told so in its status and in `tsng status`, and finds the name
|
||||
again if the holder renames or releases it.
|
||||
|
||||
It is the author's word, so a member with a clock set far back can claim an
|
||||
older time. That can reorder who holds a *name* and nothing else — addresses,
|
||||
keys and everything the data plane trusts are decided without it — and anybody
|
||||
who knows the secret could take a name by other means anyway.
|
||||
|
||||
Turning a computer off is not a revocation of its hostname and does not remove
|
||||
the participant. Revocations are not dropped merely because they are old, and
|
||||
no acknowledgement from offline peers is required to keep working.
|
||||
|
||||
+1
-1
@@ -184,7 +184,7 @@ QUIC DATAGRAM frames themselves cannot fragment.
|
||||
Reassembly is bounded and expires incomplete packets; one lost fragment loses
|
||||
one packet, without blocking unrelated traffic. The logical data payload limit
|
||||
is 64 KiB, with the relay envelope subtracted before it reaches the plugin.
|
||||
The new framing uses data ALPN `tsunagi/data/4`; both ends and intermediate
|
||||
The new framing uses data ALPN `tsunagi/data/5`; both ends and intermediate
|
||||
peers need the updated binary. Network identities and saved state do not change.
|
||||
See [protocol.md](protocol.md#the-data-plane-protocol) for the wire format.
|
||||
|
||||
|
||||
Reference in new issue
Block a user