Reach a peer through one that can reach both
Two members of a mesh could both reach a third and not each other, and that pair was simply lost to one another: a packet for a peer with no data link was counted undeliverable and dropped. Now it goes through a member that has both. What travels is not routes. Each agent says only which peers *it* has a live link with — first-hand, over the control plane, one hop, never a claim about somebody else's reachability — and everybody computes their own way through from that. The choice is local and deterministic (the lowest endpoint id among the peers that have a link to the destination), so there is nothing to agree, nothing to elect, and two agents may well route each direction differently. It is soft state: repeated while it holds, expired when it stops, so a relay that disappears stops being chosen without anybody revoking anything. The one in the middle carries bytes it cannot read. A datagram is wrapped with the peer it is for, and unwrapped on the other side into the link for the peer it came *from* — which matters, because a packet attributed to the carrier would be dropped as coming from an address the carrier does not hold. The tunnel stays end to end, and the relayed datagram goes link in, link out: it never reaches the middle's interface, so no routing, forwarding or firewall setting of that host is involved. One hop, so a loop cannot form without counting anything. A protocol is handed one link per peer that now outlives the paths under it. A direct link that dies, a hop that changes, a direct link that comes back: none of it tears down a tunnel any more, and the size a protocol may use does not change with the path. Where there was never a direct link at all, the link exists anyway as long as a hop does, so a peer reachable only through somebody still gets a tunnel. The data ALPN is `tsunagi/data/2`: every datagram now carries a tag saying whether it is direct, for somebody else, or from somebody else. The local control protocol is 13, for the relay counters — what this device carried for others is their traffic on its uplink, and that should not be invisible. `status` says `via <peer>` on a path through somebody. Fairness between the peers a relay carries for is deliberately not here yet: the queues are bounded and the counters are what a limit would be built on. Tested with fake links for the mechanics, and end to end with three real agents — two that cannot reach each other directly, a real WireGuard packet crossing through the middle. The one arrangement a single host cannot produce by itself is a pair that cannot see each other, so that is a `testing`-only switch on the agent config and exists in no release build. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
tsunagi
Claude Opus 5
@@ -43,6 +43,11 @@ Keep these separate. Crossing them is the main thing to review for.
|
||||
- **A protocol never sees the network secret.** It proves who is at the other
|
||||
end of a tunnel; proving membership of a network stays in the core, which
|
||||
is why the authenticated transport does too.
|
||||
- **A relay carries bytes, never packets.** A datagram passed on for two
|
||||
other peers goes link in, link out: it is never written to the middle's
|
||||
interface, never decrypted there, and never relayed twice. Reachability
|
||||
that decides a route is first-hand and volatile — a live announcement
|
||||
over the control plane, never a signed record and never second-hand.
|
||||
- **Plugins never learn reachability.** An `IpPlugin` is handed a `PacketLink`
|
||||
per peer and moves datagrams over it. Addresses, hole punching and relays
|
||||
belong to `crates/tsunagi/src/dataplane/transport/`. A plugin announcement says *who*, never
|
||||
|
||||
@@ -536,6 +536,27 @@ async fn main() -> Result<()> {
|
||||
reaches the internet by accident. Opt into `DirectOnly` or `N0Defaults`
|
||||
explicitly.
|
||||
|
||||
### Through somebody in the middle
|
||||
|
||||
Two members can both reach a third and not each other: a blocked path, a
|
||||
relay that is unavailable, a network only reachable from inside somebody
|
||||
else's building. When that happens the pair is routed through a member that
|
||||
has both.
|
||||
|
||||
Nothing is agreed and nothing is elected. Each member says only which peers
|
||||
*it* has a live link with, first-hand, over the control plane and one hop
|
||||
only; everyone picks their own way through from that, deterministically, and
|
||||
drops it the moment a direct link exists. There is no routing protocol, no
|
||||
second-hand claim to weigh, and a relayed datagram is never relayed again —
|
||||
so a loop cannot form.
|
||||
|
||||
The one in the middle carries **bytes it cannot read**: the tunnel stays end
|
||||
to end between the two ends, and a relayed datagram never touches the middle
|
||||
host's interface, so no forwarding, routing or firewall setting of that host
|
||||
is involved. `status` says `via <peer>` on a path that goes through
|
||||
somebody, and counts what this device has carried for others — it is their
|
||||
traffic on your uplink, and that should not be invisible.
|
||||
|
||||
## How peers find each other
|
||||
|
||||
Two different lookups are involved, and only one of them is this project's:
|
||||
|
||||
@@ -2525,6 +2525,21 @@ fn network_section(
|
||||
));
|
||||
}
|
||||
|
||||
// Only when something has: a relay that has carried nothing is not
|
||||
// worth a line, and one that has is worth knowing about — it is
|
||||
// somebody else's traffic on this device's uplink.
|
||||
let relayed = network.relay_forwarded + network.relay_sent_via + network.relay_received_via;
|
||||
if relayed > 0 {
|
||||
section.push(Row::new(
|
||||
Health::Info,
|
||||
"relay",
|
||||
format!(
|
||||
"{} carried for others · {} sent through a peer, {} arrived through one",
|
||||
network.relay_forwarded, network.relay_sent_via, network.relay_received_via
|
||||
),
|
||||
));
|
||||
}
|
||||
|
||||
let rows = member_rows(network, own_id);
|
||||
let online = rows.iter().filter(|row| row.online()).count();
|
||||
if rows.is_empty() {
|
||||
@@ -3530,6 +3545,9 @@ async fn build_report(
|
||||
network_id: network.network_id.to_string(),
|
||||
active: matches!(network.state, tsunagi::agent::NetworkState::Active),
|
||||
candidates: network.candidates.len() as u32,
|
||||
relay_forwarded: network.relay.forwarded,
|
||||
relay_sent_via: network.relay.sent_via,
|
||||
relay_received_via: network.relay.received_via,
|
||||
peers: network
|
||||
.peers
|
||||
.iter()
|
||||
@@ -3825,6 +3843,9 @@ mod status_tests {
|
||||
network_id: "xa7gyz".into(),
|
||||
active: true,
|
||||
candidates: 1,
|
||||
relay_forwarded: 0,
|
||||
relay_sent_via: 0,
|
||||
relay_received_via: 0,
|
||||
peers: vec![PeerReport {
|
||||
endpoint_id: ONLINE.into(),
|
||||
hostname: Some("music".into()),
|
||||
|
||||
@@ -162,6 +162,43 @@ impl WgAgent {
|
||||
(agent, plugin, tuns)
|
||||
}
|
||||
|
||||
/// An agent that refuses a direct data path to some peers.
|
||||
///
|
||||
/// The one arrangement a single host cannot produce by itself: two
|
||||
/// agents that both reach a third and not each other. The control
|
||||
/// plane is untouched — they are members and they talk — only the
|
||||
/// direct data link is refused, which is the real-world case of a
|
||||
/// blocked or unreachable data path.
|
||||
async fn spawn_cut_off(
|
||||
discovery: &SharedMemoryDiscovery,
|
||||
tag: &str,
|
||||
blocked: Arc<std::sync::Mutex<std::collections::HashSet<EndpointId>>>,
|
||||
) -> Self {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let tuns = MemoryTunFactory::new();
|
||||
let plugin = WireguardPlugin::open(
|
||||
WireguardConfig::new(dir.path().join("wireguard"))
|
||||
.with_reconcile(Duration::from_millis(20), Duration::from_millis(250)),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let agent = Agent::spawn(
|
||||
config_with(dir.path(), discovery)
|
||||
.with_overlay_ipv4_range(Some(tsunagi::state::DEFAULT_IPV4_RANGE))
|
||||
.with_interface(Arc::new(tuns.clone()), tag, 1280)
|
||||
.with_unreachable_data_peers(blocked)
|
||||
.with_plugin(plugin.clone() as Arc<dyn IpPlugin>),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
Self {
|
||||
dir,
|
||||
agent,
|
||||
plugin,
|
||||
tuns,
|
||||
}
|
||||
}
|
||||
|
||||
/// An agent whose interface claims to be on the host but is not.
|
||||
///
|
||||
/// Used only by the missing-address test: everywhere else the in-memory
|
||||
@@ -1306,3 +1343,78 @@ async fn an_mtu_below_what_ipv4_guarantees_is_refused() {
|
||||
.is_ok()
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_peer_with_no_direct_path_is_reached_through_one_that_has_both() {
|
||||
// A and B can each reach C and not each other. Without a way through
|
||||
// the middle they are lost to one another while sitting in the same
|
||||
// mesh; with one, C carries their datagrams without being able to read
|
||||
// a byte of them — the WireGuard tunnel is still end to end.
|
||||
let discovery = SharedMemoryDiscovery::new();
|
||||
let (name, secret) = network("wg-relay");
|
||||
|
||||
let a_blocks = Arc::new(std::sync::Mutex::new(std::collections::HashSet::new()));
|
||||
let b_blocks = Arc::new(std::sync::Mutex::new(std::collections::HashSet::new()));
|
||||
let a = WgAgent::spawn_cut_off(&discovery, "tra", Arc::clone(&a_blocks)).await;
|
||||
let b = WgAgent::spawn_cut_off(&discovery, "trb", Arc::clone(&b_blocks)).await;
|
||||
let middle = WgAgent::spawn(&discovery, "trc").await;
|
||||
a_blocks.lock().unwrap().insert(b.endpoint_id());
|
||||
b_blocks.lock().unwrap().insert(a.endpoint_id());
|
||||
|
||||
let network_id = middle.agent.join_network(&name, &secret).await.unwrap();
|
||||
a.agent.join_network(&name, &secret).await.unwrap();
|
||||
b.agent.join_network(&name, &secret).await.unwrap();
|
||||
|
||||
// All three are members and all three talk: only the data path
|
||||
// between A and B is missing.
|
||||
wait_for_peers(&a.agent, network_id, 2).await;
|
||||
wait_for_peers(&b.agent, network_id, 2).await;
|
||||
middle.wait_for_tunnels(network_id, 2).await;
|
||||
|
||||
let a_addr = a.overlay(network_id).await;
|
||||
let b_addr = b.overlay(network_id).await;
|
||||
assert_ne!(a_addr, b_addr);
|
||||
|
||||
// A's tunnel to B comes up through C.
|
||||
wait_until("a's tunnel to b is established", || async {
|
||||
let view = a.plugin.overview(network_id)?;
|
||||
view.peers
|
||||
.iter()
|
||||
.find(|peer| peer.endpoint_id == b.endpoint_id())?
|
||||
.tunnel
|
||||
.as_ref()
|
||||
.map(|tunnel| tunnel.health.since_handshake)?
|
||||
.map(|_| ())
|
||||
})
|
||||
.await;
|
||||
let path = a
|
||||
.plugin
|
||||
.overview(network_id)
|
||||
.unwrap()
|
||||
.peers
|
||||
.iter()
|
||||
.find(|peer| peer.endpoint_id == b.endpoint_id())
|
||||
.and_then(|peer| peer.tunnel.as_ref().map(|tunnel| tunnel.path.clone()))
|
||||
.unwrap_or_default();
|
||||
assert!(
|
||||
path.contains("via"),
|
||||
"the path should say it goes through somebody: {path}"
|
||||
);
|
||||
|
||||
// And a real packet crosses: A's interface to B's interface, through C.
|
||||
a.tun(network_id)
|
||||
.await
|
||||
.push_from_os(ipv4_packet(a_addr, b_addr, b"through the middle"));
|
||||
let seen = tokio::time::timeout(
|
||||
tsunagi::testing::DEADLINE,
|
||||
b.tun(network_id).await.pop_to_os(),
|
||||
)
|
||||
.await
|
||||
.expect("the packet should arrive")
|
||||
.unwrap();
|
||||
assert_eq!(&seen[20..], b"through the middle");
|
||||
|
||||
a.shutdown().await;
|
||||
b.shutdown().await;
|
||||
middle.shutdown().await;
|
||||
}
|
||||
|
||||
@@ -567,6 +567,8 @@ impl Agent {
|
||||
hostname: self.inner.read_hostname(),
|
||||
transport: self.inner.transport.get().cloned(),
|
||||
device_secret: self.inner.identity.signing_key(),
|
||||
#[cfg(feature = "testing")]
|
||||
unreachable_data_peers: Arc::clone(&self.inner.config.unreachable_data_peers),
|
||||
ipv4_range: range.propose,
|
||||
ipv4_fallback: range.fallback,
|
||||
range_conflict: range.conflict,
|
||||
@@ -771,6 +773,7 @@ impl Agent {
|
||||
range: None,
|
||||
range_conflict: None,
|
||||
metrics: NetworkMetrics::default(),
|
||||
relay: Default::default(),
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -15,7 +15,7 @@ use tokio::sync::{broadcast, mpsc, oneshot};
|
||||
use tokio::task::JoinHandle;
|
||||
|
||||
use crate::config::{Limits, ReconnectPolicy};
|
||||
use crate::dataplane::transport::{InboundLink, PacketTransport, SharedLink};
|
||||
use crate::dataplane::transport::{InboundLink, PacketLink, PacketTransport, SharedLink};
|
||||
use crate::dataplane::{PluginCapability, SharedPlugin};
|
||||
use crate::discovery::{Candidate, CandidateSource, NetworkDiscovery};
|
||||
use crate::error::{Error, Result};
|
||||
@@ -137,6 +137,9 @@ pub(crate) struct RuntimeParams {
|
||||
pub(crate) hostname: String,
|
||||
/// Signing key for this agent's own records.
|
||||
pub(crate) device_secret: iroh::SecretKey,
|
||||
/// Peers with no direct data path, for tests only.
|
||||
#[cfg(feature = "testing")]
|
||||
pub(crate) unreachable_data_peers: Arc<std::sync::Mutex<std::collections::HashSet<EndpointId>>>,
|
||||
/// The IPv4 overlay range this agent would use, if the network has not
|
||||
/// already settled on another one.
|
||||
pub(crate) ipv4_range: Option<Ipv4Range>,
|
||||
@@ -212,6 +215,14 @@ pub(crate) fn spawn(params: RuntimeParams) -> NetworkHandle {
|
||||
}
|
||||
}
|
||||
|
||||
/// How long a peer's account of what it can reach is believed.
|
||||
///
|
||||
/// Soft state: it is repeated while it holds, and simply stops being
|
||||
/// repeated when the peer goes. Nothing is revoked, so nothing can be
|
||||
/// revoked wrongly — a relay that disappears stops being chosen by
|
||||
/// itself, a few announcements later.
|
||||
const REACH_EXPIRY: Duration = Duration::from_secs(90);
|
||||
|
||||
/// How long a network waits to be told its range before proposing the one
|
||||
/// derived from its id.
|
||||
///
|
||||
@@ -236,7 +247,23 @@ struct Runtime {
|
||||
dial_results_tx: mpsc::Sender<DialOutcome>,
|
||||
dial_results_rx: mpsc::Receiver<DialOutcome>,
|
||||
/// Live data plane links, keyed by peer and plugin protocol.
|
||||
///
|
||||
/// The direct ones. What a protocol holds is the hub's link for that
|
||||
/// peer, which outlives any particular path.
|
||||
links: HashMap<(EndpointId, String), SharedLink>,
|
||||
/// Every data link of this network, and what is reached through what.
|
||||
hub: Arc<crate::dataplane::relay::RelayHub>,
|
||||
/// Which peers each peer says it has a live link with, and when it
|
||||
/// last said so.
|
||||
///
|
||||
/// First-hand only: a peer speaks for itself and never for anybody
|
||||
/// else, so there is no hearsay to weigh and nothing to poison. Soft
|
||||
/// state — an entry that stops being refreshed expires, which is how a
|
||||
/// relay that has gone stops being chosen without anybody revoking it.
|
||||
reachable: HashMap<EndpointId, (HashSet<EndpointId>, std::time::Instant)>,
|
||||
/// What this agent last told its peers it could reach, so it is only
|
||||
/// said again when it changed.
|
||||
announced_reach: HashSet<EndpointId>,
|
||||
/// Links currently being opened, so we do not start two.
|
||||
opening: HashSet<(EndpointId, String)>,
|
||||
link_results_tx: mpsc::Sender<LinkOutcome>,
|
||||
@@ -283,6 +310,9 @@ impl Runtime {
|
||||
dial_results_tx,
|
||||
dial_results_rx,
|
||||
links: HashMap::new(),
|
||||
hub: crate::dataplane::relay::RelayHub::new(network_id),
|
||||
reachable: HashMap::new(),
|
||||
announced_reach: HashSet::new(),
|
||||
opening: HashSet::new(),
|
||||
link_results_tx,
|
||||
link_results_rx,
|
||||
@@ -491,6 +521,10 @@ 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;
|
||||
// The same timer repairs the relay picture: reachability is soft
|
||||
// state, repeated while it holds and forgotten when it stops.
|
||||
self.update_paths();
|
||||
self.announce_reach(true);
|
||||
|
||||
let mut candidates: Vec<Candidate> = Vec::new();
|
||||
|
||||
@@ -1040,6 +1074,10 @@ impl Runtime {
|
||||
}
|
||||
for (peer, protocol) in dead {
|
||||
self.links.remove(&(peer, protocol.clone()));
|
||||
// Only the direct path is gone. Whether the peer is still
|
||||
// reachable through somebody is decided below; the protocol's
|
||||
// link stays either way.
|
||||
self.hub.clear_direct(peer, &protocol);
|
||||
self.emit(Event::DataLinkDown {
|
||||
network: self.network_id,
|
||||
peer,
|
||||
@@ -1076,6 +1114,9 @@ impl Runtime {
|
||||
if self.links.contains_key(&key) || self.opening.contains(&key) {
|
||||
continue;
|
||||
}
|
||||
if self.no_direct_data(peer) {
|
||||
continue;
|
||||
}
|
||||
// The smaller endpoint id dials; the other side accepts. Both
|
||||
// compute this identically, so exactly one link is created.
|
||||
if self.local_id.as_bytes() >= peer.as_bytes() {
|
||||
@@ -1106,11 +1147,142 @@ impl Runtime {
|
||||
}
|
||||
}
|
||||
|
||||
/// Tells peers which peers this agent has a live link with, when that
|
||||
/// has changed.
|
||||
///
|
||||
/// Only what it can see for itself, and only to the peers it is
|
||||
/// talking to: one hop, no flooding, nothing second-hand. A peer that
|
||||
/// hears this is the only one that can use it, because it is also the
|
||||
/// only one that could route through this agent.
|
||||
fn announce_reach(&mut self, force: bool) {
|
||||
let live: HashSet<EndpointId> = self
|
||||
.links
|
||||
.iter()
|
||||
.filter(|(_, link)| !link.is_closed())
|
||||
.map(|((peer, _), _)| *peer)
|
||||
.collect();
|
||||
if !force && live == self.announced_reach {
|
||||
return;
|
||||
}
|
||||
self.announced_reach = live.clone();
|
||||
let message = ControlMessage::Reachable {
|
||||
peers: live.iter().map(|peer| *peer.as_bytes()).collect(),
|
||||
};
|
||||
let peers: Vec<EndpointId> = self.sessions.keys().copied().collect();
|
||||
for peer in peers {
|
||||
if let Err(err) = self.send_to(peer, message.clone()) {
|
||||
tracing::debug!(%err, "could not queue a reachability announcement");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Chooses, for every peer with no direct link, somebody to go through.
|
||||
///
|
||||
/// Local and deterministic: the lowest endpoint id among the peers
|
||||
/// that both this agent has a link with and that say they have a link
|
||||
/// with the destination. Nothing is agreed with anybody — two agents
|
||||
/// may well pick differently, and traffic each way finds its own path.
|
||||
fn update_paths(&mut self) {
|
||||
self.reachable
|
||||
.retain(|_, (_, heard)| heard.elapsed() < REACH_EXPIRY);
|
||||
|
||||
let served = self.served_protocols();
|
||||
let live: HashSet<EndpointId> = self
|
||||
.links
|
||||
.iter()
|
||||
.filter(|(_, link)| !link.is_closed())
|
||||
.map(|((peer, _), _)| *peer)
|
||||
.collect();
|
||||
|
||||
// Every peer this agent would carry traffic with: a session says
|
||||
// who it is and what it speaks, which is also what a protocol
|
||||
// needs before it can have a link at all.
|
||||
let wanted: Vec<(EndpointId, String)> = self
|
||||
.sessions
|
||||
.values()
|
||||
.flat_map(|session| {
|
||||
let peer = session.peer;
|
||||
session
|
||||
.capabilities
|
||||
.iter()
|
||||
.filter(|capability| capability.enabled)
|
||||
.map(move |capability| (peer, capability.protocol.clone(), capability.version))
|
||||
})
|
||||
.filter(|(_, protocol, version)| {
|
||||
served
|
||||
.iter()
|
||||
.any(|(name, ours)| name == protocol && ours == version)
|
||||
})
|
||||
.map(|(peer, protocol, _)| (peer, protocol))
|
||||
.collect();
|
||||
|
||||
for (peer, protocol) in wanted {
|
||||
if live.contains(&peer) {
|
||||
// There is a direct link; a hop is only for when there is
|
||||
// not, and holding a stale one would keep a peer that has
|
||||
// gone looking reachable.
|
||||
if self.hub.has_link(peer, &protocol) {
|
||||
self.hub.set_hop(peer, &protocol, None);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
let hop = self
|
||||
.reachable
|
||||
.iter()
|
||||
.filter(|(middle, (reaches, _))| {
|
||||
live.contains(*middle) && reaches.contains(&peer) && **middle != peer
|
||||
})
|
||||
.map(|(middle, _)| *middle)
|
||||
.min_by(|a, b| a.as_bytes().cmp(b.as_bytes()));
|
||||
|
||||
let known = self.hub.has_link(peer, &protocol);
|
||||
if !known && hop.is_none() {
|
||||
// Nothing to reach it with and nothing to offer a
|
||||
// protocol; inventing a link object here would only look
|
||||
// like connectivity.
|
||||
continue;
|
||||
}
|
||||
self.hub.set_hop(peer, &protocol, hop);
|
||||
if known {
|
||||
continue;
|
||||
}
|
||||
// A peer reachable only through somebody still gets a link, so
|
||||
// its tunnel can be built: the protocol is not told how its
|
||||
// datagrams travel, only that this is the way to that peer.
|
||||
let Some(plugin) = self
|
||||
.params
|
||||
.plugins
|
||||
.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);
|
||||
self.emit(Event::DataLinkUp {
|
||||
network: self.network_id,
|
||||
peer,
|
||||
protocol,
|
||||
path,
|
||||
max_datagram,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
fn handle_link_result(&mut self, outcome: LinkOutcome) {
|
||||
let key = (outcome.peer, outcome.protocol.clone());
|
||||
self.opening.remove(&key);
|
||||
match outcome.result {
|
||||
Ok(link) => self.adopt_link(outcome.peer, outcome.protocol, link),
|
||||
Ok(link) => {
|
||||
self.adopt_link(outcome.peer, outcome.protocol, link);
|
||||
// A new link is both a path for us and one we can offer
|
||||
// others, so it changes what we route and what we say.
|
||||
self.update_paths();
|
||||
self.announce_reach(false);
|
||||
}
|
||||
Err(reason) => {
|
||||
// A data plane that cannot be set up is reported, never fatal.
|
||||
self.metrics.data_link_failures += 1;
|
||||
@@ -1124,8 +1296,32 @@ impl Runtime {
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether this agent refuses a direct data path to a peer.
|
||||
///
|
||||
/// Always false outside a test build: see
|
||||
/// [`crate::config::AgentConfig::with_unreachable_data_peers`].
|
||||
fn no_direct_data(&self, peer: EndpointId) -> bool {
|
||||
#[cfg(feature = "testing")]
|
||||
{
|
||||
match self.params.unreachable_data_peers.lock() {
|
||||
Ok(guard) => guard.contains(&peer),
|
||||
Err(poisoned) => poisoned.into_inner().contains(&peer),
|
||||
}
|
||||
}
|
||||
#[cfg(not(feature = "testing"))]
|
||||
{
|
||||
let _ = peer;
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
fn install_link(&mut self, inbound: InboundLink) {
|
||||
if self.no_direct_data(inbound.peer) {
|
||||
return;
|
||||
}
|
||||
self.adopt_link(inbound.peer, inbound.protocol, inbound.link);
|
||||
self.update_paths();
|
||||
self.announce_reach(false);
|
||||
}
|
||||
|
||||
/// Hands a link to the plugin that owns its protocol.
|
||||
@@ -1141,10 +1337,19 @@ impl Runtime {
|
||||
};
|
||||
|
||||
let path = link.path_description();
|
||||
let max_datagram = link.max_datagram_size();
|
||||
self.links
|
||||
.insert((peer, protocol.clone()), Arc::clone(&link));
|
||||
plugin.on_peer_link(self.network_id, peer, 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);
|
||||
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,
|
||||
@@ -1157,6 +1362,10 @@ impl Runtime {
|
||||
|
||||
/// Drops every link to a peer.
|
||||
fn drop_links_for(&mut self, peer: EndpointId) {
|
||||
// The peer itself is going, not just a path to it, so the
|
||||
// protocol's link goes with it.
|
||||
self.hub.remove_peer(peer);
|
||||
self.reachable.remove(&peer);
|
||||
let keys: Vec<(EndpointId, String)> = self
|
||||
.links
|
||||
.keys()
|
||||
@@ -1386,6 +1595,17 @@ impl Runtime {
|
||||
ControlMessage::State { records } => {
|
||||
self.pending_state.push((peer, records.clone()));
|
||||
}
|
||||
// What that peer can reach, from that peer. Kept with the time
|
||||
// it was heard, so it can go stale on its own.
|
||||
ControlMessage::Reachable { peers } => {
|
||||
let heard: HashSet<EndpointId> = peers
|
||||
.iter()
|
||||
.filter_map(|raw| EndpointId::from_bytes(raw).ok())
|
||||
.collect();
|
||||
self.reachable
|
||||
.insert(peer, (heard, std::time::Instant::now()));
|
||||
self.update_paths();
|
||||
}
|
||||
ControlMessage::Pong { .. } | ControlMessage::Bye { .. } => {}
|
||||
}
|
||||
|
||||
@@ -1532,6 +1752,7 @@ impl Runtime {
|
||||
.range_conflict
|
||||
.filter(|_| self.effective_range().is_none()),
|
||||
metrics: self.metrics.clone(),
|
||||
relay: self.hub.counters(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -176,6 +176,8 @@ pub struct NetworkStatus {
|
||||
pub range_conflict: Option<crate::state::Ipv4Range>,
|
||||
/// Per-network counters.
|
||||
pub metrics: NetworkMetrics,
|
||||
/// What has gone through a peer in the middle, in both directions.
|
||||
pub relay: crate::dataplane::relay::RelayCounters,
|
||||
}
|
||||
|
||||
impl NetworkStatus {
|
||||
|
||||
@@ -223,6 +223,10 @@ pub struct AgentConfig {
|
||||
pub bind_addrs: Vec<SocketAddr>,
|
||||
/// How much external connectivity machinery the endpoint may use.
|
||||
pub transport: TransportPolicy,
|
||||
/// Peers with no direct data path, for tests. See
|
||||
/// [`AgentConfig::with_unreachable_data_peers`].
|
||||
#[cfg(feature = "testing")]
|
||||
pub unreachable_data_peers: Arc<std::sync::Mutex<std::collections::HashSet<iroh::EndpointId>>>,
|
||||
/// Hostname announced to peers. `None` keeps whatever the state store holds,
|
||||
/// falling back to the OS hostname and finally to a short endpoint id.
|
||||
pub hostname: Option<String>,
|
||||
@@ -264,6 +268,10 @@ impl AgentConfig {
|
||||
paths,
|
||||
bind_addrs: Vec::new(),
|
||||
transport: TransportPolicy::default(),
|
||||
#[cfg(feature = "testing")]
|
||||
unreachable_data_peers: Arc::new(std::sync::Mutex::new(
|
||||
std::collections::HashSet::new(),
|
||||
)),
|
||||
hostname: None,
|
||||
discovery: None,
|
||||
discovery_interval: Duration::from_secs(5),
|
||||
@@ -292,6 +300,21 @@ impl AgentConfig {
|
||||
self
|
||||
}
|
||||
|
||||
/// Peers this agent will not open or accept a **direct** data link with.
|
||||
///
|
||||
/// Test-only, and behind the `testing` feature so a release build has
|
||||
/// no such switch. The one arrangement that cannot be made on a single
|
||||
/// host is two agents that both reach a third and not each other,
|
||||
/// which is exactly the case a relay exists for.
|
||||
#[cfg(feature = "testing")]
|
||||
pub fn with_unreachable_data_peers(
|
||||
mut self,
|
||||
peers: Arc<std::sync::Mutex<std::collections::HashSet<iroh::EndpointId>>>,
|
||||
) -> Self {
|
||||
self.unreachable_data_peers = peers;
|
||||
self
|
||||
}
|
||||
|
||||
/// Sets the transport policy.
|
||||
pub fn with_transport(mut self, transport: TransportPolicy) -> Self {
|
||||
self.transport = transport;
|
||||
|
||||
@@ -19,6 +19,7 @@
|
||||
//! A data plane failure never stops the daemon: errors returned here are
|
||||
//! recorded and surfaced, the control plane keeps running.
|
||||
|
||||
pub mod relay;
|
||||
pub mod transport;
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
@@ -0,0 +1,806 @@
|
||||
//! Carrying a peer's packets through another peer.
|
||||
//!
|
||||
//! Two members of a network can both reach a third and yet not reach each
|
||||
//! other: a blocked path, a relay that is unavailable, a network that is
|
||||
//! only reachable from inside somebody else's building. Without a way
|
||||
//! through the middle, that pair is simply lost to each other while both
|
||||
//! sit in the same mesh.
|
||||
//!
|
||||
//! # What goes through the middle, and what does not
|
||||
//!
|
||||
//! The relay moves **bytes it cannot read**. A datagram is wrapped with the
|
||||
//! peer it is for, handed to the peer in the middle, and unwrapped on the
|
||||
//! other side; the protocol's own encryption is end to end between the two
|
||||
//! ends of the tunnel, so the one in the middle carries an opaque payload.
|
||||
//!
|
||||
//! It never reaches the middle's operating system either: a relayed
|
||||
//! datagram goes link in, link out. Nothing is written to its interface,
|
||||
//! so no routing, forwarding or firewall setting of its host is involved —
|
||||
//! which is both the fast path and the only one allowed here, since an
|
||||
//! agent touches no system object it did not create.
|
||||
//!
|
||||
//! # What is deliberately not here
|
||||
//!
|
||||
//! No routing protocol and nothing second-hand. A peer says only which
|
||||
//! peers *it* has a live link with, and that is used to pick one hop —
|
||||
//! never a claim about somebody else's reachability. One hop means loops
|
||||
//! are impossible by construction rather than by a counter, and a relayed
|
||||
//! datagram is never relayed again.
|
||||
//!
|
||||
//! Fairness between the peers a relay carries for is not addressed yet: the
|
||||
//! queues are bounded and the counters say how much went through, which is
|
||||
//! what a limit would be built on.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
||||
use std::sync::{Arc, Mutex, RwLock, Weak};
|
||||
|
||||
use bytes::{BufMut, Bytes, BytesMut};
|
||||
use iroh::EndpointId;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
use crate::BoxFuture;
|
||||
use crate::identity::NetworkId;
|
||||
|
||||
use super::transport::{PacketLink, SharedLink, TransportError};
|
||||
|
||||
/// A datagram straight from the peer that sent it.
|
||||
const TAG_DIRECT: u8 = 0;
|
||||
/// A datagram for somebody else, to be passed on.
|
||||
const TAG_TO_RELAY: u8 = 1;
|
||||
/// A datagram that was passed on, naming who it came from.
|
||||
const TAG_RELAYED: u8 = 2;
|
||||
|
||||
/// The largest header any of the three shapes needs.
|
||||
///
|
||||
/// Subtracted from every link's datagram size, relayed or not, so that the
|
||||
/// size a protocol may use does not change when the path does. A tunnel
|
||||
/// that had to renegotiate its packet size every time a path changed would
|
||||
/// be worse than one that is a few bytes smaller than it could be.
|
||||
pub const RELAY_OVERHEAD: usize = 1 + 32;
|
||||
|
||||
/// How many datagrams may wait for a protocol to read them.
|
||||
///
|
||||
/// Bounded, and the oldest is dropped rather than the newest kept waiting:
|
||||
/// these are datagrams, loss is ordinary, and a queue that grows is worse
|
||||
/// than one that spills.
|
||||
const INBOX: usize = 256;
|
||||
|
||||
/// What a datagram on a data link turned out to be.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
enum Frame {
|
||||
/// From the peer at the other end of this link, for us.
|
||||
Direct(Bytes),
|
||||
/// From the peer at the other end, for somebody else.
|
||||
ToRelay { to: EndpointId, payload: Bytes },
|
||||
/// Passed on by the peer at the other end, from somebody else.
|
||||
Relayed { from: EndpointId, payload: Bytes },
|
||||
}
|
||||
|
||||
fn wrap(tag: u8, id: Option<EndpointId>, payload: &[u8]) -> Bytes {
|
||||
let mut out = BytesMut::with_capacity(1 + 32 + payload.len());
|
||||
out.put_u8(tag);
|
||||
if let Some(id) = id {
|
||||
out.put_slice(id.as_bytes());
|
||||
}
|
||||
out.put_slice(payload);
|
||||
out.freeze()
|
||||
}
|
||||
|
||||
/// Reads a datagram, or `None` if it is not one of ours.
|
||||
///
|
||||
/// Nothing here trusts a length: every field is checked before it is read,
|
||||
/// because this is bytes off the network like any other.
|
||||
fn unwrap(raw: Bytes) -> Option<Frame> {
|
||||
let tag = *raw.first()?;
|
||||
match tag {
|
||||
TAG_DIRECT => Some(Frame::Direct(raw.slice(1..))),
|
||||
TAG_TO_RELAY | TAG_RELAYED => {
|
||||
if raw.len() < 1 + 32 {
|
||||
return None;
|
||||
}
|
||||
let mut id = [0u8; 32];
|
||||
id.copy_from_slice(&raw[1..33]);
|
||||
let id = EndpointId::from_bytes(&id).ok()?;
|
||||
let payload = raw.slice(33..);
|
||||
Some(if tag == TAG_TO_RELAY {
|
||||
Frame::ToRelay { to: id, payload }
|
||||
} else {
|
||||
Frame::Relayed { from: id, payload }
|
||||
})
|
||||
}
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// What has gone through the relay, from both sides of it.
|
||||
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
|
||||
pub struct RelayCounters {
|
||||
/// Datagrams this agent sent to a peer through somebody else.
|
||||
pub sent_via: u64,
|
||||
/// Datagrams this agent passed on for two other peers.
|
||||
pub forwarded: u64,
|
||||
/// Datagrams that arrived for this agent through somebody else.
|
||||
pub received_via: u64,
|
||||
/// Datagrams that could not be passed on: no link to the destination.
|
||||
pub dropped_no_link: u64,
|
||||
/// Datagrams that arrived relayed from a peer we have no link object
|
||||
/// for, or that could not be read at all.
|
||||
pub dropped_unknown: u64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default)]
|
||||
struct Tally {
|
||||
sent_via: AtomicU64,
|
||||
forwarded: AtomicU64,
|
||||
received_via: AtomicU64,
|
||||
dropped_no_link: AtomicU64,
|
||||
dropped_unknown: AtomicU64,
|
||||
}
|
||||
|
||||
impl Tally {
|
||||
fn snapshot(&self) -> RelayCounters {
|
||||
RelayCounters {
|
||||
sent_via: self.sent_via.load(Ordering::Relaxed),
|
||||
forwarded: self.forwarded.load(Ordering::Relaxed),
|
||||
received_via: self.received_via.load(Ordering::Relaxed),
|
||||
dropped_no_link: self.dropped_no_link.load(Ordering::Relaxed),
|
||||
dropped_unknown: self.dropped_unknown.load(Ordering::Relaxed),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Every data link of one network, and what it can reach through what.
|
||||
///
|
||||
/// One per network runtime. It owns the raw links the transport produced
|
||||
/// and hands protocols a [`PeerLink`] each, which is the thing that knows
|
||||
/// whether a peer is reached directly or through somebody.
|
||||
#[derive(Debug)]
|
||||
pub struct RelayHub {
|
||||
network: NetworkId,
|
||||
/// What a protocol holds, one per peer and protocol. Kept alive here so
|
||||
/// a datagram can be injected into a link the protocol is reading.
|
||||
links: Mutex<HashMap<(EndpointId, String), Arc<PeerLink>>>,
|
||||
/// The raw links, which is what a hop is: a way to reach the peer in
|
||||
/// the middle, never wrapped again.
|
||||
raw: Mutex<HashMap<(EndpointId, String), SharedLink>>,
|
||||
tally: Arc<Tally>,
|
||||
}
|
||||
|
||||
impl RelayHub {
|
||||
/// A hub for one network.
|
||||
pub fn new(network: NetworkId) -> Arc<Self> {
|
||||
Arc::new(Self {
|
||||
network,
|
||||
links: Mutex::new(HashMap::new()),
|
||||
raw: Mutex::new(HashMap::new()),
|
||||
tally: Arc::new(Tally::default()),
|
||||
})
|
||||
}
|
||||
|
||||
/// What has gone through it.
|
||||
pub fn counters(&self) -> RelayCounters {
|
||||
self.tally.snapshot()
|
||||
}
|
||||
|
||||
fn links(&self) -> std::sync::MutexGuard<'_, HashMap<(EndpointId, String), Arc<PeerLink>>> {
|
||||
match self.links.lock() {
|
||||
Ok(guard) => guard,
|
||||
Err(poisoned) => poisoned.into_inner(),
|
||||
}
|
||||
}
|
||||
|
||||
fn raw(&self) -> std::sync::MutexGuard<'_, HashMap<(EndpointId, String), SharedLink>> {
|
||||
match self.raw.lock() {
|
||||
Ok(guard) => guard,
|
||||
Err(poisoned) => poisoned.into_inner(),
|
||||
}
|
||||
}
|
||||
|
||||
/// The link a protocol uses for a peer, created on first ask.
|
||||
///
|
||||
/// The same object for the life of the peer, whatever happens to the
|
||||
/// path underneath it: a protocol's tunnel survives a direct link
|
||||
/// coming and going, and a change of hop, without noticing.
|
||||
pub fn link(self: &Arc<Self>, peer: EndpointId, protocol: &str) -> Arc<PeerLink> {
|
||||
let key = (peer, protocol.to_string());
|
||||
let mut links = self.links();
|
||||
if let Some(existing) = links.get(&key) {
|
||||
return Arc::clone(existing);
|
||||
}
|
||||
let (inbox_tx, inbox_rx) = mpsc::channel(INBOX);
|
||||
let link = Arc::new(PeerLink {
|
||||
network: self.network,
|
||||
peer,
|
||||
protocol: protocol.to_string(),
|
||||
direct: RwLock::new(None),
|
||||
hop: RwLock::new(None),
|
||||
inbox_tx,
|
||||
inbox_rx: tokio::sync::Mutex::new(inbox_rx),
|
||||
changed: tokio::sync::Notify::new(),
|
||||
closed: AtomicBool::new(false),
|
||||
gone: tokio::sync::Notify::new(),
|
||||
hub: Arc::downgrade(self),
|
||||
tally: Arc::clone(&self.tally),
|
||||
});
|
||||
links.insert(key, Arc::clone(&link));
|
||||
link
|
||||
}
|
||||
|
||||
/// Whether a protocol already holds a link for this peer.
|
||||
///
|
||||
/// Asked before handing one out, because a protocol is given its link
|
||||
/// once and keeps it: handing the same peer a second link would leave
|
||||
/// it with two and a tunnel it is no longer reading for.
|
||||
pub fn has_link(&self, peer: EndpointId, protocol: &str) -> bool {
|
||||
self.links().contains_key(&(peer, protocol.to_string()))
|
||||
}
|
||||
|
||||
/// Records the direct link to a peer, replacing any previous one.
|
||||
pub fn set_direct(self: &Arc<Self>, peer: EndpointId, protocol: &str, raw: SharedLink) {
|
||||
self.raw().insert((peer, protocol.to_string()), raw.clone());
|
||||
let link = self.link(peer, protocol);
|
||||
link.set_direct(Some(raw));
|
||||
}
|
||||
|
||||
/// Forgets the direct link to a peer. The protocol's link stays.
|
||||
pub fn clear_direct(self: &Arc<Self>, peer: EndpointId, protocol: &str) {
|
||||
self.raw().remove(&(peer, protocol.to_string()));
|
||||
if let Some(link) = self.links().get(&(peer, protocol.to_string())) {
|
||||
link.set_direct(None);
|
||||
}
|
||||
}
|
||||
|
||||
/// Routes a peer through another peer, or stops doing so.
|
||||
///
|
||||
/// The hop must be a peer with a direct link of its own; anything else
|
||||
/// is a request to relay through somebody unreachable.
|
||||
pub fn set_hop(self: &Arc<Self>, peer: EndpointId, protocol: &str, hop: Option<EndpointId>) {
|
||||
let resolved = hop.and_then(|hop| {
|
||||
self.raw()
|
||||
.get(&(hop, protocol.to_string()))
|
||||
.map(|link| (hop, Arc::clone(link)))
|
||||
});
|
||||
let link = self.link(peer, protocol);
|
||||
link.set_hop(resolved);
|
||||
}
|
||||
|
||||
/// Closes and forgets everything held for a peer.
|
||||
pub fn remove_peer(&self, peer: EndpointId) {
|
||||
self.raw().retain(|(other, _), _| *other != peer);
|
||||
let mut links = self.links();
|
||||
links.retain(|(other, _), link| {
|
||||
if *other == peer {
|
||||
link.close();
|
||||
false
|
||||
} else {
|
||||
true
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// Closes everything. The network is going away.
|
||||
pub fn close(&self) {
|
||||
self.raw().clear();
|
||||
for link in self.links().drain() {
|
||||
link.1.close();
|
||||
}
|
||||
}
|
||||
|
||||
/// Passes a datagram on to the peer it is for.
|
||||
///
|
||||
/// Only over a direct link: a relayed datagram is never relayed again,
|
||||
/// which is what makes a loop impossible without counting hops.
|
||||
fn forward(&self, to: EndpointId, protocol: &str, from: EndpointId, payload: Bytes) {
|
||||
let link = self.raw().get(&(to, protocol.to_string())).cloned();
|
||||
let Some(link) = link else {
|
||||
self.tally.dropped_no_link.fetch_add(1, Ordering::Relaxed);
|
||||
return;
|
||||
};
|
||||
match link.send(wrap(TAG_RELAYED, Some(from), &payload)) {
|
||||
Ok(()) => {
|
||||
self.tally.forwarded.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
Err(err) => {
|
||||
tracing::debug!(%err, "cannot pass a datagram on");
|
||||
self.tally.dropped_no_link.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Hands a datagram that arrived through somebody to the peer's link.
|
||||
fn inject(&self, from: EndpointId, protocol: &str, payload: Bytes) {
|
||||
let link = self.links().get(&(from, protocol.to_string())).cloned();
|
||||
let Some(link) = link else {
|
||||
// Nothing is reading for that peer: it is not one this agent
|
||||
// carries traffic with, so the datagram has no owner.
|
||||
self.tally.dropped_unknown.fetch_add(1, Ordering::Relaxed);
|
||||
return;
|
||||
};
|
||||
match link.inbox_tx.try_send(payload) {
|
||||
Ok(()) => {
|
||||
self.tally.received_via.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
Err(_) => {
|
||||
self.tally.dropped_unknown.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A protocol's link to one peer, however that peer is reached.
|
||||
///
|
||||
/// It outlives any particular path. A direct link that dies, a hop that
|
||||
/// changes, a direct link that comes back: none of it is visible to the
|
||||
/// protocol holding this, so its tunnel is not torn down and rebuilt every
|
||||
/// time the way through changes.
|
||||
#[derive(Debug)]
|
||||
pub struct PeerLink {
|
||||
network: NetworkId,
|
||||
peer: EndpointId,
|
||||
protocol: String,
|
||||
direct: RwLock<Option<SharedLink>>,
|
||||
hop: RwLock<Option<(EndpointId, SharedLink)>>,
|
||||
inbox_tx: mpsc::Sender<Bytes>,
|
||||
inbox_rx: tokio::sync::Mutex<mpsc::Receiver<Bytes>>,
|
||||
/// Woken when the path changes, so a reader parked on the old one
|
||||
/// starts again on the new one.
|
||||
changed: tokio::sync::Notify,
|
||||
closed: AtomicBool,
|
||||
gone: tokio::sync::Notify,
|
||||
hub: Weak<RelayHub>,
|
||||
tally: Arc<Tally>,
|
||||
}
|
||||
|
||||
impl PeerLink {
|
||||
fn direct(&self) -> Option<SharedLink> {
|
||||
match self.direct.read() {
|
||||
Ok(guard) => guard.clone(),
|
||||
Err(poisoned) => poisoned.into_inner().clone(),
|
||||
}
|
||||
}
|
||||
|
||||
fn hop(&self) -> Option<(EndpointId, SharedLink)> {
|
||||
match self.hop.read() {
|
||||
Ok(guard) => guard.clone(),
|
||||
Err(poisoned) => poisoned.into_inner().clone(),
|
||||
}
|
||||
}
|
||||
|
||||
fn set_direct(&self, link: Option<SharedLink>) {
|
||||
match self.direct.write() {
|
||||
Ok(mut guard) => *guard = link,
|
||||
Err(poisoned) => *poisoned.into_inner() = link,
|
||||
}
|
||||
self.changed.notify_waiters();
|
||||
}
|
||||
|
||||
fn set_hop(&self, hop: Option<(EndpointId, SharedLink)>) {
|
||||
match self.hop.write() {
|
||||
Ok(mut guard) => *guard = hop,
|
||||
Err(poisoned) => *poisoned.into_inner() = hop,
|
||||
}
|
||||
self.changed.notify_waiters();
|
||||
}
|
||||
|
||||
fn close(&self) {
|
||||
self.closed.store(true, Ordering::Relaxed);
|
||||
self.gone.notify_waiters();
|
||||
self.changed.notify_waiters();
|
||||
}
|
||||
|
||||
/// The peer this link carries traffic for.
|
||||
pub fn peer(&self) -> EndpointId {
|
||||
self.peer
|
||||
}
|
||||
|
||||
/// Which peer it goes through, when it is not direct.
|
||||
pub fn via(&self) -> Option<EndpointId> {
|
||||
if self.direct().is_some_and(|link| !link.is_closed()) {
|
||||
return None;
|
||||
}
|
||||
self.hop().map(|(peer, _)| peer)
|
||||
}
|
||||
}
|
||||
|
||||
impl PacketLink for PeerLink {
|
||||
fn network(&self) -> NetworkId {
|
||||
self.network
|
||||
}
|
||||
|
||||
fn peer(&self) -> EndpointId {
|
||||
self.peer
|
||||
}
|
||||
|
||||
fn max_datagram_size(&self) -> usize {
|
||||
// The same size whatever the path, so a protocol never has to
|
||||
// resize because the way through changed.
|
||||
let direct = self.direct().map(|link| link.max_datagram_size());
|
||||
let hop = self.hop().map(|(_, link)| link.max_datagram_size());
|
||||
let smallest = match (direct, hop) {
|
||||
(Some(a), Some(b)) => a.min(b),
|
||||
(Some(a), None) | (None, Some(a)) => a,
|
||||
(None, None) => 0,
|
||||
};
|
||||
smallest.saturating_sub(RELAY_OVERHEAD)
|
||||
}
|
||||
|
||||
fn send(&self, payload: Bytes) -> Result<(), TransportError> {
|
||||
if self.closed.load(Ordering::Relaxed) {
|
||||
return Err(TransportError::Closed);
|
||||
}
|
||||
// Direct while there is one: a hop is what you use when there is
|
||||
// nothing better, never a preference.
|
||||
if let Some(direct) = self.direct()
|
||||
&& !direct.is_closed()
|
||||
{
|
||||
return direct.send(wrap(TAG_DIRECT, None, &payload));
|
||||
}
|
||||
if let Some((_, hop)) = self.hop() {
|
||||
let out = hop.send(wrap(TAG_TO_RELAY, Some(self.peer), &payload));
|
||||
if out.is_ok() {
|
||||
self.tally.sent_via.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
Err(TransportError::Unreachable(format!(
|
||||
"no path to {}",
|
||||
self.peer.fmt_short()
|
||||
)))
|
||||
}
|
||||
|
||||
fn recv(&self) -> BoxFuture<'_, Option<Bytes>> {
|
||||
Box::pin(async move {
|
||||
let mut inbox = self.inbox_rx.lock().await;
|
||||
loop {
|
||||
if self.closed.load(Ordering::Relaxed) {
|
||||
return None;
|
||||
}
|
||||
let direct = self.direct();
|
||||
let changed = self.changed.notified();
|
||||
let gone = self.gone.notified();
|
||||
|
||||
let raw = tokio::select! {
|
||||
// Whatever arrived through somebody else, already
|
||||
// stripped of its wrapper by the link it came in on.
|
||||
injected = inbox.recv() => return injected,
|
||||
raw = async {
|
||||
match &direct {
|
||||
Some(link) => link.recv().await,
|
||||
// Nothing direct: wait for a path or an injection.
|
||||
None => std::future::pending().await,
|
||||
}
|
||||
} => raw,
|
||||
// The path changed underneath: look again.
|
||||
() = changed => continue,
|
||||
() = gone => return None,
|
||||
};
|
||||
|
||||
let Some(raw) = raw else {
|
||||
// The direct link ended. The peer may still be
|
||||
// reachable through somebody, so this link is not over.
|
||||
self.set_direct(None);
|
||||
continue;
|
||||
};
|
||||
match unwrap(raw) {
|
||||
Some(Frame::Direct(payload)) => return Some(payload),
|
||||
// This agent is the one in the middle.
|
||||
Some(Frame::ToRelay { to, payload }) => {
|
||||
if let Some(hub) = self.hub.upgrade() {
|
||||
hub.forward(to, &self.protocol, self.peer, payload);
|
||||
}
|
||||
}
|
||||
// Somebody passed this on for a peer we talk to.
|
||||
Some(Frame::Relayed { from, payload }) => {
|
||||
if let Some(hub) = self.hub.upgrade() {
|
||||
hub.inject(from, &self.protocol, payload);
|
||||
}
|
||||
}
|
||||
None => {
|
||||
self.tally.dropped_unknown.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
fn closed(&self) -> BoxFuture<'_, ()> {
|
||||
Box::pin(async move {
|
||||
loop {
|
||||
if self.closed.load(Ordering::Relaxed) {
|
||||
return;
|
||||
}
|
||||
self.gone.notified().await;
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
fn is_closed(&self) -> bool {
|
||||
self.closed.load(Ordering::Relaxed)
|
||||
}
|
||||
|
||||
fn path_description(&self) -> String {
|
||||
if let Some(direct) = self.direct()
|
||||
&& !direct.is_closed()
|
||||
{
|
||||
return direct.path_description();
|
||||
}
|
||||
match self.hop() {
|
||||
Some((peer, link)) => {
|
||||
format!("via {} ({})", peer.fmt_short(), link.path_description())
|
||||
}
|
||||
None => "no path".to_string(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
|
||||
|
||||
use super::*;
|
||||
use crate::identity::{NetworkKeys, NetworkName, NetworkSecret};
|
||||
|
||||
fn network() -> NetworkId {
|
||||
NetworkKeys::derive(
|
||||
&NetworkName::new("relay").unwrap(),
|
||||
&NetworkSecret::from_bytes([3u8; 32]).unwrap(),
|
||||
)
|
||||
.network_id()
|
||||
}
|
||||
|
||||
fn peer(seed: u8) -> EndpointId {
|
||||
iroh::SecretKey::from_bytes(&[seed; 32]).public()
|
||||
}
|
||||
|
||||
/// A link that records what was sent and can be fed what arrives.
|
||||
#[derive(Debug)]
|
||||
struct Wire {
|
||||
network: NetworkId,
|
||||
peer: EndpointId,
|
||||
sent: Mutex<Vec<Bytes>>,
|
||||
inbound: tokio::sync::Mutex<mpsc::Receiver<Bytes>>,
|
||||
feed: mpsc::Sender<Bytes>,
|
||||
closed: AtomicBool,
|
||||
}
|
||||
|
||||
impl Wire {
|
||||
fn new(peer: EndpointId) -> Arc<Self> {
|
||||
let (feed, inbound) = mpsc::channel(32);
|
||||
Arc::new(Self {
|
||||
network: network(),
|
||||
peer,
|
||||
sent: Mutex::new(Vec::new()),
|
||||
inbound: tokio::sync::Mutex::new(inbound),
|
||||
feed,
|
||||
closed: AtomicBool::new(false),
|
||||
})
|
||||
}
|
||||
|
||||
fn sent(&self) -> Vec<Bytes> {
|
||||
self.sent.lock().unwrap().clone()
|
||||
}
|
||||
|
||||
async fn arrive(&self, raw: Bytes) {
|
||||
self.feed.send(raw).await.unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
impl PacketLink for Wire {
|
||||
fn network(&self) -> NetworkId {
|
||||
self.network
|
||||
}
|
||||
fn peer(&self) -> EndpointId {
|
||||
self.peer
|
||||
}
|
||||
fn max_datagram_size(&self) -> usize {
|
||||
1200
|
||||
}
|
||||
fn send(&self, payload: Bytes) -> Result<(), TransportError> {
|
||||
if self.closed.load(Ordering::Relaxed) {
|
||||
return Err(TransportError::Closed);
|
||||
}
|
||||
self.sent.lock().unwrap().push(payload);
|
||||
Ok(())
|
||||
}
|
||||
fn recv(&self) -> BoxFuture<'_, Option<Bytes>> {
|
||||
Box::pin(async move { self.inbound.lock().await.recv().await })
|
||||
}
|
||||
fn closed(&self) -> BoxFuture<'_, ()> {
|
||||
Box::pin(async move { std::future::pending().await })
|
||||
}
|
||||
fn is_closed(&self) -> bool {
|
||||
self.closed.load(Ordering::Relaxed)
|
||||
}
|
||||
fn path_description(&self) -> String {
|
||||
"test wire".into()
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_datagram_survives_the_wrapping_and_a_broken_one_is_refused() {
|
||||
let id = peer(1);
|
||||
let direct = wrap(TAG_DIRECT, None, b"hello");
|
||||
assert_eq!(unwrap(direct).unwrap(), Frame::Direct(Bytes::from("hello")));
|
||||
|
||||
let onward = wrap(TAG_TO_RELAY, Some(id), b"hello");
|
||||
assert_eq!(
|
||||
unwrap(onward).unwrap(),
|
||||
Frame::ToRelay {
|
||||
to: id,
|
||||
payload: Bytes::from("hello")
|
||||
}
|
||||
);
|
||||
|
||||
// Bytes off the network: nothing is read before it is checked.
|
||||
assert!(unwrap(Bytes::new()).is_none());
|
||||
assert!(unwrap(Bytes::from_static(&[TAG_TO_RELAY, 1, 2, 3])).is_none());
|
||||
assert!(unwrap(Bytes::from_static(&[200, 1, 2, 3])).is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_direct_link_is_preferred_and_a_hop_is_used_when_there_is_none() {
|
||||
let hub = RelayHub::new(network());
|
||||
let (them, middle) = (peer(1), peer(2));
|
||||
|
||||
let to_middle = Wire::new(middle);
|
||||
hub.set_direct(middle, "test-ip", to_middle.clone());
|
||||
|
||||
// No direct link to them: nothing to send on, and nothing invented.
|
||||
let link = hub.link(them, "test-ip");
|
||||
assert!(link.send(Bytes::from("x")).is_err());
|
||||
assert_eq!(link.via(), None);
|
||||
|
||||
// Routed through the middle: the datagram goes out on that link,
|
||||
// wrapped with who it is for.
|
||||
hub.set_hop(them, "test-ip", Some(middle));
|
||||
assert_eq!(link.via(), Some(middle));
|
||||
link.send(Bytes::from("through you")).unwrap();
|
||||
let sent = to_middle.sent();
|
||||
assert_eq!(sent.len(), 1);
|
||||
assert_eq!(
|
||||
unwrap(sent[0].clone()).unwrap(),
|
||||
Frame::ToRelay {
|
||||
to: them,
|
||||
payload: Bytes::from("through you")
|
||||
}
|
||||
);
|
||||
assert_eq!(hub.counters().sent_via, 1);
|
||||
|
||||
// A direct link appears: it is used, and the hop is not.
|
||||
let to_them = Wire::new(them);
|
||||
hub.set_direct(them, "test-ip", to_them.clone());
|
||||
assert_eq!(link.via(), None);
|
||||
link.send(Bytes::from("straight")).unwrap();
|
||||
assert_eq!(
|
||||
unwrap(to_them.sent()[0].clone()).unwrap(),
|
||||
Frame::Direct(Bytes::from("straight"))
|
||||
);
|
||||
assert_eq!(to_middle.sent().len(), 1, "nothing more went the long way");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn the_one_in_the_middle_passes_it_on_without_reading_it() {
|
||||
// C between A and B. A's datagram arrives wrapped for B; C puts it
|
||||
// on its link to B, saying who it came from, and never sees inside.
|
||||
let hub = RelayHub::new(network());
|
||||
let (a, b) = (peer(1), peer(2));
|
||||
let from_a = Wire::new(a);
|
||||
let to_b = Wire::new(b);
|
||||
hub.set_direct(a, "test-ip", from_a.clone());
|
||||
hub.set_direct(b, "test-ip", to_b.clone());
|
||||
|
||||
let a_link = hub.link(a, "test-ip");
|
||||
let reading = tokio::spawn(async move { a_link.recv().await });
|
||||
|
||||
from_a.arrive(wrap(TAG_TO_RELAY, Some(b), b"opaque")).await;
|
||||
// It is passed on rather than returned to the protocol here.
|
||||
tokio::time::timeout(std::time::Duration::from_millis(200), async {
|
||||
while to_b.sent().is_empty() {
|
||||
tokio::task::yield_now().await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("it was passed on");
|
||||
assert_eq!(
|
||||
unwrap(to_b.sent()[0].clone()).unwrap(),
|
||||
Frame::Relayed {
|
||||
from: a,
|
||||
payload: Bytes::from("opaque")
|
||||
}
|
||||
);
|
||||
assert_eq!(hub.counters().forwarded, 1);
|
||||
assert!(!reading.is_finished(), "the protocol was not handed it");
|
||||
|
||||
// Nowhere to put it is a drop and a count, never an error upwards.
|
||||
from_a
|
||||
.arrive(wrap(TAG_TO_RELAY, Some(peer(9)), b"nobody"))
|
||||
.await;
|
||||
tokio::time::timeout(std::time::Duration::from_millis(200), async {
|
||||
while hub.counters().dropped_no_link == 0 {
|
||||
tokio::task::yield_now().await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("counted");
|
||||
reading.abort();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn what_arrives_through_somebody_is_handed_to_the_right_peer() {
|
||||
// B's side: a datagram from A arrives on the link with C, and the
|
||||
// protocol reading A's link is the one that gets it — otherwise
|
||||
// the packet would be attributed to C and dropped as coming from
|
||||
// an address C does not hold.
|
||||
let hub = RelayHub::new(network());
|
||||
let (a, c) = (peer(1), peer(3));
|
||||
let from_c = Wire::new(c);
|
||||
hub.set_direct(c, "test-ip", from_c.clone());
|
||||
let a_link = hub.link(a, "test-ip");
|
||||
let c_link = hub.link(c, "test-ip");
|
||||
|
||||
let reading_a = tokio::spawn(async move { a_link.recv().await });
|
||||
let driving_c = tokio::spawn(async move { c_link.recv().await });
|
||||
|
||||
from_c.arrive(wrap(TAG_RELAYED, Some(a), b"from a")).await;
|
||||
let got = tokio::time::timeout(std::time::Duration::from_secs(2), reading_a)
|
||||
.await
|
||||
.expect("delivered")
|
||||
.unwrap();
|
||||
assert_eq!(got, Some(Bytes::from("from a")));
|
||||
assert_eq!(hub.counters().received_via, 1);
|
||||
driving_c.abort();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_link_outlives_the_path_under_it() {
|
||||
// The protocol's tunnel must not be torn down because a path
|
||||
// changed: the link object is the peer, not the way to it.
|
||||
let hub = RelayHub::new(network());
|
||||
let (them, middle) = (peer(1), peer(2));
|
||||
let direct = Wire::new(them);
|
||||
hub.set_direct(them, "test-ip", direct.clone());
|
||||
let link = hub.link(them, "test-ip");
|
||||
assert!(!link.is_closed());
|
||||
|
||||
hub.clear_direct(them, "test-ip");
|
||||
assert!(
|
||||
!link.is_closed(),
|
||||
"still the same peer, still the same link"
|
||||
);
|
||||
assert!(
|
||||
link.send(Bytes::from("x")).is_err(),
|
||||
"but nothing to send on"
|
||||
);
|
||||
|
||||
let hop = Wire::new(middle);
|
||||
hub.set_direct(middle, "test-ip", hop.clone());
|
||||
hub.set_hop(them, "test-ip", Some(middle));
|
||||
link.send(Bytes::from("x")).unwrap();
|
||||
assert_eq!(hop.sent().len(), 1);
|
||||
|
||||
// The peer going away is what closes it.
|
||||
hub.remove_peer(them);
|
||||
assert!(link.is_closed());
|
||||
assert!(link.send(Bytes::from("x")).is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn the_size_a_protocol_may_use_does_not_change_with_the_path() {
|
||||
let hub = RelayHub::new(network());
|
||||
let (them, middle) = (peer(1), peer(2));
|
||||
hub.set_direct(them, "test-ip", Wire::new(them));
|
||||
hub.set_direct(middle, "test-ip", Wire::new(middle));
|
||||
let link = hub.link(them, "test-ip");
|
||||
let direct_size = link.max_datagram_size();
|
||||
assert_eq!(direct_size, 1200 - RELAY_OVERHEAD);
|
||||
|
||||
hub.set_hop(them, "test-ip", Some(middle));
|
||||
hub.clear_direct(them, "test-ip");
|
||||
assert_eq!(
|
||||
link.max_datagram_size(),
|
||||
direct_size,
|
||||
"a tunnel that resized on every path change would be worse"
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -274,6 +274,12 @@ pub struct NetworkReport {
|
||||
/// has no way to reach anybody" — which look identical in a report
|
||||
/// that counts only members.
|
||||
pub candidates: u32,
|
||||
/// Datagrams this agent passed on between two other peers.
|
||||
pub relay_forwarded: u64,
|
||||
/// Datagrams this agent sent to a peer through somebody else.
|
||||
pub relay_sent_via: u64,
|
||||
/// Datagrams that reached this agent through somebody else.
|
||||
pub relay_received_via: u64,
|
||||
/// Outbound dials that failed.
|
||||
pub dial_failures: u64,
|
||||
/// Handshakes rejected in either direction.
|
||||
|
||||
@@ -367,7 +367,7 @@ pub async fn set_dns(
|
||||
///
|
||||
/// 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', 12]);
|
||||
pub const CONTROL_PROTOCOL: u32 = u32::from_be_bytes([b'T', b'S', b'N', 13]);
|
||||
|
||||
async fn write_message<T: serde::Serialize>(stream: &mut UnixStream, value: &T) -> Result<()> {
|
||||
let encoded = postcard::to_stdvec(value)
|
||||
|
||||
@@ -28,7 +28,12 @@ pub const ALPN: &[u8] = b"tsunagi/ctrl/1";
|
||||
/// They carry one IP plugin's packets for one network and nothing else, so a
|
||||
/// saturated or broken data plane cannot disturb control traffic, and the
|
||||
/// transport underneath can be replaced without touching the control protocol.
|
||||
pub const DATA_ALPN: &[u8] = b"tsunagi/data/1";
|
||||
///
|
||||
/// Version 2 puts a tag on every datagram, so one can say "this is for
|
||||
/// somebody else" and be passed on by the peer in the middle. An agent
|
||||
/// speaking version 1 simply does not form a data link with one speaking
|
||||
/// version 2, which is what the version in an ALPN is for.
|
||||
pub const DATA_ALPN: &[u8] = b"tsunagi/data/2";
|
||||
|
||||
/// Largest plugin protocol identifier accepted when opening a data channel.
|
||||
pub const MAX_DATA_PROTOCOL_LEN: usize = 32;
|
||||
@@ -124,6 +129,20 @@ pub enum ControlMessage {
|
||||
/// The records. Bounded by [`crate::config::Limits::max_state_records`].
|
||||
records: Vec<crate::state::SignedRecord>,
|
||||
},
|
||||
/// Which peers this sender has a live data link with, right now.
|
||||
///
|
||||
/// First-hand and nothing else: a sender speaks only for itself, never
|
||||
/// about what somebody else can reach. That is what makes it usable
|
||||
/// without weighing hearsay — the claim is proved or disproved by
|
||||
/// sending through it.
|
||||
///
|
||||
/// A snapshot rather than a change, because a snapshot is idempotent
|
||||
/// and repairs itself; and volatile, so it lives here and not in
|
||||
/// signed state, which is for what has to survive a member being away.
|
||||
Reachable {
|
||||
/// The peers, as raw endpoint keys.
|
||||
peers: Vec<[u8; 32]>,
|
||||
},
|
||||
/// Graceful goodbye.
|
||||
///
|
||||
/// A peer going away is not a revocation of anything.
|
||||
@@ -191,6 +210,11 @@ pub fn validate_capability(
|
||||
/// network, the other networks or the agent.
|
||||
pub fn validate(message: &ControlMessage, limits: &Limits) -> Result<(), ProtocolError> {
|
||||
match message {
|
||||
ControlMessage::Reachable { peers } => {
|
||||
// One entry per member at most; a sender claiming more than a
|
||||
// network can hold is not describing this network.
|
||||
check_len("reachable.peers", peers.len(), limits.max_state_records)?;
|
||||
}
|
||||
ControlMessage::Announce(announcement) => {
|
||||
check_len(
|
||||
"announce.hostname",
|
||||
@@ -229,6 +253,7 @@ pub fn kind(message: &ControlMessage) -> &'static str {
|
||||
ControlMessage::Ping { .. } => "ping",
|
||||
ControlMessage::Pong { .. } => "pong",
|
||||
ControlMessage::State { .. } => "state",
|
||||
ControlMessage::Reachable { .. } => "reachable",
|
||||
ControlMessage::Bye { .. } => "bye",
|
||||
}
|
||||
}
|
||||
|
||||
@@ -55,6 +55,16 @@ keeping the WireGuard identity, shutdown removing every interface, a forged
|
||||
overlay claim being rejected, and the core carrying the payload without
|
||||
interpreting it.
|
||||
|
||||
`crates/tsunagi/src/dataplane/relay.rs` has its own tests for the way
|
||||
through a peer in the middle: the wrapping and what a malformed one does, a
|
||||
direct path being preferred over a hop, the middle passing a datagram on
|
||||
without being handed it, what arrives through somebody reaching the peer it
|
||||
came from rather than the one that carried it, a link outliving the paths
|
||||
under it, and the datagram size not changing when the path does.
|
||||
`tests/wireguard.rs` proves it end to end — two agents that can each reach a
|
||||
third and not each other, with a real WireGuard packet crossing through the
|
||||
middle.
|
||||
|
||||
Unit tests in `crates/tsunagi/src/state/` cover the signed record model directly: tampering
|
||||
with any field breaks verification, a newer version wins while an older one
|
||||
never rolls back, two authors claiming one address resolve the same way no
|
||||
|
||||
Reference in New Issue
Block a user