Add pure WireGuard direct UDP data plane plugin and transport

Peers communicating on uncensored networks can benefit from raw WireGuard
performance over direct UDP without QUIC framing or double-encryption
overhead. This provides Tailscale-like direct tunnel performance while
preserving Tsunagi's decentralized mesh architecture.

Key changes:
- Create `tsunagi-wireguard` crate implementing the `IpPlugin` and
  `PacketTransport` contracts:
  - `WG_PROTOCOL = "wg"` with wire version 1.
  - Bounded `WgAnnouncement` carrying WireGuard public keys, listening UDP
    ports, and candidate IP addresses (filtering out overlay interfaces).
  - `WgKeyStore` managing persistent per-network WireGuard X25519 keys
    in SQLite with zeroization on drop.
  - `WireguardTransport` providing direct UDP packet transmission, candidate
    address tracking, inbound link delivery, and roaming support.
  - `WireguardDevice` running userspace WireGuard encryption via `boringtun`,
    timer maintenance, and AllowedIPs source enforcement.
- Integrate `wg` into `tsunagi-cli`:
  - Set default protocols to `"wg,tcp-tls,wg-quic"` so all protocols are
    enabled by default in order of throughput and performance.
  - Support `"wireguard"` as a convenience alias for `"wg"` in `--protocol`.
  - Bind `PureWgTransport` and launch the inbound datagram accept loop on `up`.
  - Pass `pure_wg` plugin to `AgentControl` and update `build_report` so
    active WireGuard tunnels and overlay metrics are reported in `tsunagi status`.
  - Clean up WireGuard keys on network removal in `forget_protocol_state`.
- Comprehensive test coverage in `tsunagi-wireguard`:
  - `announcement_tests`: round-trip serialization and verification of malformed,
    mismatched network, zero port, or self-claiming announcements.
  - `transport_tests`: loopback UDP link delivery between transports.
  - `plugin_tests`: end-to-end two-peer handshake and bidirectional IP packet
    transit over direct UDP.

Validation on Linux:
- cargo fmt --all -- --check passed.
- cargo clippy --locked --workspace --all-targets -- -D warnings passed.
- cargo test --locked --workspace --all-targets passed.
This commit is contained in:
Aleksandr Bogomiakov committed 2026-10-01 01:24:25 +01:00
1 parent 46575d0899
commit 7070a16104
14 files changed
+2906 -90

No files matched your search

Generated
+25
View File
@@ -4253,6 +4253,7 @@ dependencies = [
"tsunagi",
"tsunagi-tcp-tls",
"tsunagi-wg-quic",
"tsunagi-wireguard",
]
[[package]]
@@ -4309,6 +4310,30 @@ dependencies = [
"zeroize",
]
[[package]]
name = "tsunagi-wireguard"
version = "0.1.0"
dependencies = [
"boringtun",
"bytes",
"data-encoding",
"hex",
"iroh",
"netdev 0.46.3",
"postcard",
"rand 0.10.3",
"rusqlite",
"serde",
"subtle",
"tempfile",
"thiserror 2.0.20",
"tokio",
"tracing",
"tracing-subscriber",
"tsunagi",
"zeroize",
]
[[package]]
name = "tun"
version = "0.8.14"
+1
View File
@@ -17,6 +17,7 @@ doc = false
[dependencies]
tsunagi = { path = "../tsunagi", version = "0.1.0", features = ["tun-device", "dns-publish"] }
tsunagi-wireguard = { path = "../tsunagi-wireguard", version = "0.1.0" }
tsunagi-wg-quic = { path = "../tsunagi-wg-quic", version = "0.1.0" }
tsunagi-tcp-tls = { path = "../tsunagi-tcp-tls", version = "0.1.0" }
iroh.workspace = true
+210 -90
View File
@@ -27,6 +27,10 @@ use tsunagi_tcp_tls::{
TCP_TLS_PROTOCOL, TCP_TLS_VERSION, TcpTlsCodec, TcpTlsConfig, TcpTlsPlugin, TcpTlsTransport,
};
use tsunagi_wg_quic::{WireguardConfig, WireguardPlugin};
use tsunagi_wireguard::{
WG_PROTOCOL, WG_WIRE_VERSION, WireguardConfig as PureWgConfig, WireguardPlugin as PureWgPlugin,
WireguardTransport as PureWgTransport,
};
/// A small agent for private mesh networks.
#[derive(Debug, Parser)]
@@ -421,7 +425,7 @@ struct UpArgs {
long = "protocol",
value_name = "LIST",
value_delimiter = ',',
default_value = "tcp-tls,wg-quic",
default_value = "wg,tcp-tls,wg-quic",
help_heading = "Transport"
)]
protocols: Vec<String>,
@@ -1190,6 +1194,12 @@ struct ProtocolSpec {
/// Every protocol this build has.
const PROTOCOLS: &[ProtocolSpec] = &[
ProtocolSpec {
name: WG_PROTOCOL,
version: WG_WIRE_VERSION,
summary: "Pure WireGuard direct UDP transport for uncensored networks and maximum performance",
options: PureWgPlugin::OPTIONS,
},
ProtocolSpec {
name: TCP_TLS_PROTOCOL,
version: TCP_TLS_VERSION,
@@ -1291,6 +1301,7 @@ fn show_protocols() -> Result<(), Box<dyn std::error::Error>> {
/// published.
struct AgentControl {
agent: Agent,
pure_wg: Option<Arc<PureWgPlugin>>,
plugin: Option<Arc<WireguardPlugin>>,
tcp_tls: Option<Arc<TcpTlsPlugin>>,
/// The resolver, which this owns so it can be switched while running.
@@ -1305,6 +1316,7 @@ impl tsunagi::ipc::ReportSource for AgentControl {
let dns = self.dns_state().await;
build_report(
&self.agent,
self.pure_wg.as_deref(),
self.plugin.as_deref(),
self.tcp_tls.as_deref(),
dns,
@@ -2005,6 +2017,19 @@ fn forget_protocol_state(paths: &StoragePaths, network: tsunagi::NetworkId) {
}
Err(err) => eprintln!("warning: the wg-quic key store could not be opened: {err}"),
}
let pure_store = PureWgConfig::new(paths.state_dir.join("wg"));
match tsunagi_wireguard::WgKeyStore::open(pure_store.key_store_path()) {
Ok(store) => {
if let Err(err) = store.forget(network) {
eprintln!("warning: the wg key for it could not be removed: {err}");
}
}
Err(err) if !pure_store.key_store_path().exists() => {
let _ = err;
}
Err(err) => eprintln!("warning: the wg key store could not be opened: {err}"),
}
}
/// `tsunagi wipe`: back to a device that has never joined anything.
@@ -2871,7 +2896,7 @@ fn member_row(row: &MemberRow<'_>) -> report::Row {
format!("{stats:<48} · {}", tunnel.path)
}
None => {
if proto == "wg-quic" {
if proto == "wg-quic" || proto == "wg" {
"no WireGuard handshake yet; the tunnel cannot carry traffic".to_string()
} else {
format!("no {proto} handshake yet; the tunnel cannot carry traffic")
@@ -3489,7 +3514,13 @@ async fn up(args: UpArgs) -> Result<(), Box<dyn std::error::Error>> {
.iter()
.filter(|name| !name.eq_ignore_ascii_case("none"))
{
let Some(spec) = PROTOCOLS.iter().find(|spec| spec.name == name.as_str()) else {
// Accept "wireguard" as a convenience alias for the canonical "wg".
let canonical = if name.eq_ignore_ascii_case("wireguard") {
WG_PROTOCOL
} else {
name.as_str()
};
let Some(spec) = PROTOCOLS.iter().find(|spec| spec.name == canonical) else {
let known: Vec<&str> = PROTOCOLS.iter().map(|spec| spec.name).collect();
return Err(format!(
"this build has no protocol called `{name}`; it has {}. \
@@ -3524,6 +3555,8 @@ async fn up(args: UpArgs) -> Result<(), Box<dyn std::error::Error>> {
let mut wireguard = None;
let mut tcp_tls_transport = None;
let mut tcp_tls_plugin = None;
let mut pure_wg_plugin: Option<Arc<PureWgPlugin>> = None;
let mut pure_wg_transport: Option<Arc<PureWgTransport>> = None;
if !wanted.is_empty() {
let tun_factory: Arc<dyn TunFactory> = if args.no_tun {
Arc::new(MemoryTunFactory::new())
@@ -3574,6 +3607,35 @@ async fn up(args: UpArgs) -> Result<(), Box<dyn std::error::Error>> {
tcp_tls_transport = Some(transport);
tcp_tls_plugin = Some(plugin);
}
WG_PROTOCOL => {
let mut wg_cfg = PureWgConfig::new(paths.state_dir.join("wg"));
if let Some(mtu) = args.mtu {
wg_cfg = wg_cfg.with_mtu(mtu);
}
let wg_cfg = PureWgPlugin::configure(wg_cfg, &options)?;
let identity = tsunagi::storage::StateStore::open(paths.state_db())?
.load_or_create_device_identity()?;
let transport = Arc::new(
PureWgTransport::bind(&identity, wg_cfg.port)
.await
.map_err(|err| err.to_string())?,
);
for ep_addr in &bootstrap {
for addr in ep_addr.ip_addrs() {
transport.set_peer_addr(ep_addr.id, *addr);
}
}
let plugin = PureWgPlugin::open(wg_cfg, Arc::clone(&transport)).await?;
config = config
.with_plugin(plugin.clone() as Arc<dyn IpPlugin>)
.with_custom_transport(
WG_PROTOCOL,
Arc::clone(&transport)
as Arc<dyn tsunagi::dataplane::transport::PacketTransport>,
);
pure_wg_transport = Some(transport);
pure_wg_plugin = Some(plugin);
}
other => return Err(format!("`{other}` is listed but not built in").into()),
}
}
@@ -3626,6 +3688,21 @@ async fn up(args: UpArgs) -> Result<(), Box<dyn std::error::Error>> {
.await;
});
}
if let Some(transport) = pure_wg_transport {
let agent_inbound = agent.clone();
tokio::spawn(async move {
transport
.accept_loop(move |inbound| {
let agent = agent_inbound.clone();
tokio::spawn(async move {
if let Err(err) = agent.install_inbound_link(inbound).await {
tracing::debug!(%err, "failed to install inbound wg link");
}
});
})
.await;
});
}
let protocol_names: Vec<&str> = wanted.iter().map(|p| p.name).collect();
@@ -3662,11 +3739,13 @@ async fn up(args: UpArgs) -> Result<(), Box<dyn std::error::Error>> {
// is not fatal: the agent itself works fine without it.
let control = {
let agent = agent.clone();
let pure_wg = pure_wg_plugin.clone();
let plugin = wireguard.clone();
let tcp_tls = tcp_tls_plugin.clone();
let dns = Arc::clone(&dns);
let source: Arc<dyn tsunagi::ipc::ReportSource> = Arc::new(AgentControl {
agent,
pure_wg,
plugin,
tcp_tls,
dns,
@@ -3741,6 +3820,7 @@ async fn up(args: UpArgs) -> Result<(), Box<dyn std::error::Error>> {
/// both.
async fn build_report(
agent: &Agent,
pure_wg: Option<&PureWgPlugin>,
wireguard: Option<&WireguardPlugin>,
tcp_tls: Option<&TcpTlsPlugin>,
dns: Option<DnsState>,
@@ -3761,57 +3841,126 @@ async fn build_report(
.iter()
.map(|network| {
let wg_view = wireguard.and_then(|plugin| plugin.overview(network.network_id));
let pure_wg_view = pure_wg.and_then(|plugin| plugin.overview(network.network_id));
let tcp_tls_reports = tcp_tls
.map(|p| p.active_peers(network.network_id))
.unwrap_or_default();
let overlay = if let Some(view) = wg_view {
let mut peers = Vec::new();
for peer in &view.peers {
let tcp_tls_match = tcp_tls_reports.iter().find(|r| r.peer == peer.endpoint_id);
// Use whichever WireGuard view is available for the overlay
// address and range — wg-quic and pure wg both derive them
// identically from the control plane's allocations.
let wg_overlay_addr = wg_view
.as_ref()
.and_then(|v| v.overlay_address_v4)
.or_else(|| pure_wg_view.as_ref().and_then(|v| v.overlay_address_v4));
let wg_prefix_len = wg_view
.as_ref()
.and_then(|v| v.ipv4_range)
.or_else(|| pure_wg_view.as_ref().and_then(|v| v.ipv4_range))
.map_or(0, |r| r.prefix_len);
let has_any_plugin =
wg_view.is_some() || pure_wg_view.is_some() || !tcp_tls_reports.is_empty();
if let Some(tcp_tls) = tcp_tls_match {
peers.push(OverlayPeerReport {
endpoint_id: peer.endpoint_id.to_string(),
protocol: tcp_tls.protocol.clone(),
public_key: peer.public_key.to_string(),
address: peer.overlay_address_v4.map(|addr| addr.to_string()),
handshake_secs_ago: Some(tcp_tls.uptime_secs),
tx_packets: tcp_tls.tx_packets,
rx_packets: tcp_tls.rx_packets,
dropped: tcp_tls.dropped,
protocol_errors: 0,
path: tcp_tls.path.clone(),
});
} else if let Some(tunnel) = &peer.tunnel {
peers.push(OverlayPeerReport {
endpoint_id: peer.endpoint_id.to_string(),
protocol: "wg-quic".into(),
public_key: peer.public_key.to_string(),
address: peer.overlay_address_v4.map(|addr| addr.to_string()),
handshake_secs_ago: tunnel.health.since_handshake.map(|s| s.as_secs()),
tx_packets: tunnel.stats.tx_packets,
rx_packets: tunnel.stats.rx_packets,
dropped: tunnel.stats.dropped_wrong_source
+ tunnel.stats.dropped_oversize,
protocol_errors: tunnel.stats.protocol_errors,
path: tunnel.path.clone(),
});
} else {
peers.push(OverlayPeerReport {
endpoint_id: peer.endpoint_id.to_string(),
protocol: "wg-quic".into(),
public_key: peer.public_key.to_string(),
address: peer.overlay_address_v4.map(|addr| addr.to_string()),
handshake_secs_ago: None,
tx_packets: 0,
rx_packets: 0,
dropped: 0,
protocol_errors: 0,
path: "no data link".into(),
});
let overlay = if has_any_plugin && overlay.is_some() {
let mut peers: Vec<OverlayPeerReport> = Vec::new();
// 1. Peers known to the wg-quic plugin.
if let Some(ref view) = wg_view {
for peer in &view.peers {
let tcp_tls_match =
tcp_tls_reports.iter().find(|r| r.peer == peer.endpoint_id);
if let Some(tcp_tls) = tcp_tls_match {
peers.push(OverlayPeerReport {
endpoint_id: peer.endpoint_id.to_string(),
protocol: tcp_tls.protocol.clone(),
public_key: peer.public_key.to_string(),
address: peer.overlay_address_v4.map(|addr| addr.to_string()),
handshake_secs_ago: Some(tcp_tls.uptime_secs),
tx_packets: tcp_tls.tx_packets,
rx_packets: tcp_tls.rx_packets,
dropped: tcp_tls.dropped,
protocol_errors: 0,
path: tcp_tls.path.clone(),
});
} else if let Some(tunnel) = &peer.tunnel {
peers.push(OverlayPeerReport {
endpoint_id: peer.endpoint_id.to_string(),
protocol: "wg-quic".into(),
public_key: peer.public_key.to_string(),
address: peer.overlay_address_v4.map(|addr| addr.to_string()),
handshake_secs_ago: tunnel
.health
.since_handshake
.map(|s| s.as_secs()),
tx_packets: tunnel.stats.tx_packets,
rx_packets: tunnel.stats.rx_packets,
dropped: tunnel.stats.dropped_wrong_source
+ tunnel.stats.dropped_oversize,
protocol_errors: tunnel.stats.protocol_errors,
path: tunnel.path.clone(),
});
} else {
peers.push(OverlayPeerReport {
endpoint_id: peer.endpoint_id.to_string(),
protocol: "wg-quic".into(),
public_key: peer.public_key.to_string(),
address: peer.overlay_address_v4.map(|addr| addr.to_string()),
handshake_secs_ago: None,
tx_packets: 0,
rx_packets: 0,
dropped: 0,
protocol_errors: 0,
path: "no data link".into(),
});
}
}
}
// 2. Peers known to the pure WG plugin but not yet covered.
if let Some(ref view) = pure_wg_view {
for peer in &view.peers {
if peers
.iter()
.any(|p| p.endpoint_id == peer.endpoint_id.to_string())
{
continue;
}
if let Some(tunnel) = &peer.tunnel {
peers.push(OverlayPeerReport {
endpoint_id: peer.endpoint_id.to_string(),
protocol: "wg".into(),
public_key: peer.public_key.to_string(),
address: peer.overlay_address_v4.map(|addr| addr.to_string()),
handshake_secs_ago: tunnel
.health
.since_handshake
.map(|s| s.as_secs()),
tx_packets: tunnel.stats.tx_packets,
rx_packets: tunnel.stats.rx_packets,
dropped: tunnel.stats.dropped_wrong_source
+ tunnel.stats.dropped_oversize,
protocol_errors: tunnel.stats.protocol_errors,
path: tunnel.path.clone(),
});
} else {
peers.push(OverlayPeerReport {
endpoint_id: peer.endpoint_id.to_string(),
protocol: "wg".into(),
public_key: peer.public_key.to_string(),
address: peer.overlay_address_v4.map(|addr| addr.to_string()),
handshake_secs_ago: None,
tx_packets: 0,
rx_packets: 0,
dropped: 0,
protocol_errors: 0,
path: "no data link".into(),
});
}
}
}
// 3. tcp-tls peers not yet covered by either WG view.
for report in &tcp_tls_reports {
if !peers
.iter()
@@ -3837,58 +3986,29 @@ async fn build_report(
});
}
}
Some(OverlayReport {
interface: overlay
.as_ref()
.map_or_else(String::new, |o| o.interface.clone()),
on_host: overlay.as_ref().is_some_and(|o| o.on_host),
mtu: overlay.as_ref().map_or(0, |o| o.mtu),
address: view.overlay_address_v4.map(|addr| addr.to_string()),
prefix_len: view.ipv4_range.map_or(0, |r| r.prefix_len),
peers,
unroutable_packets: overlay.as_ref().map_or(0, |o| o.counters.unroutable),
multicast_packets: overlay.as_ref().map_or(0, |o| o.counters.multicast),
unroutable_sample: overlay
.as_ref()
.and_then(|o| o.counters.unroutable_sample.map(|a| a.to_string())),
})
} else if overlay.is_some() && !tcp_tls_reports.is_empty() {
let peers = tcp_tls_reports
.into_iter()
.map(|report| {
let address = network
let (address, prefix_len) = if wg_overlay_addr.is_some() {
(wg_overlay_addr.map(|addr| addr.to_string()), wg_prefix_len)
} else {
(
network
.members
.iter()
.find(|m| m.endpoint_id == report.peer)
.find(|m| m.endpoint_id == status.endpoint_id)
.and_then(|m| m.overlay_address_v4)
.map(|a| a.to_string());
OverlayPeerReport {
endpoint_id: report.peer.to_string(),
protocol: report.protocol,
public_key: String::new(),
address,
handshake_secs_ago: Some(report.uptime_secs),
tx_packets: report.tx_packets,
rx_packets: report.rx_packets,
dropped: report.dropped,
protocol_errors: 0,
path: report.path,
}
})
.collect();
.map(|a| a.to_string()),
network.range.map_or(0, |r| r.prefix_len),
)
};
Some(OverlayReport {
interface: overlay
.as_ref()
.map_or_else(String::new, |o| o.interface.clone()),
on_host: overlay.as_ref().is_some_and(|o| o.on_host),
mtu: overlay.as_ref().map_or(0, |o| o.mtu),
address: network
.members
.iter()
.find(|m| m.endpoint_id == status.endpoint_id)
.and_then(|m| m.overlay_address_v4)
.map(|a| a.to_string()),
prefix_len: network.range.map_or(0, |r| r.prefix_len),
address,
prefix_len,
peers,
unroutable_packets: overlay.as_ref().map_or(0, |o| o.counters.unroutable),
multicast_packets: overlay.as_ref().map_or(0, |o| o.counters.multicast),
+35
View File
@@ -0,0 +1,35 @@
[package]
name = "tsunagi-wireguard"
version = "0.1.0"
description = "Pure WireGuard direct UDP transport and protocol plugin for tsunagi mesh networks."
edition.workspace = true
rust-version.workspace = true
license.workspace = true
repository.workspace = true
[dependencies]
tsunagi = { path = "../tsunagi", version = "0.1.0" }
iroh.workspace = true
tokio = { workspace = true, features = ["net", "io-util", "time", "sync", "macros"] }
serde.workspace = true
postcard.workspace = true
bytes.workspace = true
thiserror.workspace = true
tracing.workspace = true
data-encoding.workspace = true
hex.workspace = true
rusqlite = { version = "0.40", features = ["bundled"] }
subtle = "2.6"
zeroize = { version = "1.9", features = ["derive"] }
rand = "0.10"
boringtun = { version = "0.7.1", default-features = false }
netdev = "0.46"
[dev-dependencies]
tsunagi = { path = "../tsunagi", features = ["testing"] }
tokio = { version = "1.53", features = ["rt", "rt-multi-thread", "sync", "time", "macros"] }
tempfile.workspace = true
tracing-subscriber.workspace = true
[lints]
workspace = true
@@ -0,0 +1,142 @@
//! What a pure WireGuard peer tells the network about itself.
//!
//! Exchanged via [`tsunagi::dataplane::PluginCapability`]. The announcement
//! carries the WireGuard public key, the listening UDP port, and candidate IP addresses
//! for direct UDP packet transport.
use serde::{Deserialize, Serialize};
use tsunagi::dataplane::PluginError;
use tsunagi::identity::NetworkId;
use crate::keys::WgPublicKey;
/// The wire protocol identifier for pure WireGuard.
pub const WG_PROTOCOL: &str = "wg";
/// Wire version of the `wg` announcement format.
pub const ANNOUNCEMENT_VERSION: u16 = 1;
/// Wire version peers compare during capability exchange.
///
/// This is the same as [`ANNOUNCEMENT_VERSION`]; keeping a separate name
/// makes the intent explicit in the CLI protocol registry.
pub const WG_WIRE_VERSION: u16 = ANNOUNCEMENT_VERSION;
/// Maximum number of IP candidates allowed in an announcement.
pub const MAX_CANDIDATE_ADDRS: usize = 32;
/// What one participant advertises for the pure WireGuard UDP data plane.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct WgAnnouncement {
/// Announcement format version.
pub version: u16,
/// The peer's WireGuard public key.
pub public_key: [u8; 32],
/// The network this key is for.
pub network: [u8; 32],
/// The UDP port the peer is listening on for WireGuard packets.
pub port: u16,
/// Candidate IP addresses where this peer can be reached on `port`.
#[serde(default)]
pub addrs: Vec<std::net::IpAddr>,
}
/// A peer announcement that has been validated against a specific network.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ValidatedAnnouncement {
/// The peer's WireGuard public key.
pub public_key: WgPublicKey,
/// The UDP port the peer listens on.
pub port: u16,
/// Validated candidate IP addresses.
pub addrs: Vec<std::net::IpAddr>,
}
impl WgAnnouncement {
/// Builds this agent's announcement.
pub fn new(
network: NetworkId,
public_key: &WgPublicKey,
port: u16,
addrs: Vec<std::net::IpAddr>,
) -> Self {
Self {
version: ANNOUNCEMENT_VERSION,
public_key: *public_key.as_bytes(),
network: *network.as_bytes(),
port,
addrs,
}
}
/// Encodes the announcement into the opaque capability payload.
pub fn encode(&self) -> Result<Vec<u8>, PluginError> {
postcard::to_allocvec(self)
.map_err(|err| PluginError::Other(format!("cannot encode announcement: {err}")))
}
/// Decodes and validates a payload received from a peer.
pub fn decode_and_validate(
payload: &[u8],
network: NetworkId,
local_key: &WgPublicKey,
) -> Result<ValidatedAnnouncement, PluginError> {
let announcement: Self = postcard::from_bytes(payload)
.map_err(|_| PluginError::Rejected("malformed WireGuard announcement".into()))?;
announcement.validate(network, local_key)
}
fn validate(
self,
network: NetworkId,
local_key: &WgPublicKey,
) -> Result<ValidatedAnnouncement, PluginError> {
if self.version != ANNOUNCEMENT_VERSION {
return Err(PluginError::Rejected(format!(
"peer speaks WireGuard announcement version {} but this build speaks \
{ANNOUNCEMENT_VERSION}; one of the two needs updating",
self.version
)));
}
let public_key = WgPublicKey::from_bytes(self.public_key);
if public_key.is_zero() {
return Err(PluginError::Rejected(
"WireGuard public key is all zeroes".into(),
));
}
if &public_key == local_key {
return Err(PluginError::Rejected(
"peer announced this agent's own WireGuard key".into(),
));
}
if self.network != *network.as_bytes() {
return Err(PluginError::Rejected(
"announcement is for a different network".into(),
));
}
if self.port == 0 {
return Err(PluginError::Rejected("WireGuard port cannot be 0".into()));
}
if self.addrs.len() > MAX_CANDIDATE_ADDRS {
return Err(PluginError::Rejected(format!(
"too many candidate addresses: {} > {MAX_CANDIDATE_ADDRS}",
self.addrs.len()
)));
}
let mut valid_addrs = Vec::new();
for addr in self.addrs {
if !addr.is_unspecified() && !valid_addrs.contains(&addr) {
valid_addrs.push(addr);
}
}
Ok(ValidatedAnnouncement {
public_key,
port: self.port,
addrs: valid_addrs,
})
}
}
+527
View File
@@ -0,0 +1,527 @@
//! Userspace WireGuard device implementation over direct UDP links.
//!
//! Uses [`boringtun::noise::Tunn`] state machine to encrypt and decrypt
//! standard WireGuard protocol packets directly over a [`SharedLink`].
use std::collections::{HashMap, VecDeque};
use std::net::Ipv4Addr;
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, RwLock};
use std::time::Duration;
use boringtun::noise::{Tunn, TunnResult};
use bytes::Bytes;
use iroh::EndpointId;
use tokio::task::JoinHandle;
use tsunagi::dataplane::PacketSink;
use tsunagi::dataplane::PluginError;
use tsunagi::dataplane::routing::{FlowId, flow::ip_flow};
use tsunagi::dataplane::transport::{SharedLink, TransportError};
use tsunagi::identity::NetworkId;
use crate::keys::{WgPublicKey, WgSecretKey};
const TIMER_INTERVAL: Duration = Duration::from_millis(250);
const SCRATCH: usize = 4096;
struct FlowTunnel {
tunn: Tunn,
queued: VecDeque<FlowId>,
}
fn is_data(result: &TunnResult<'_>) -> bool {
matches!(result, TunnResult::WriteToNetwork(bytes) if bytes.starts_with(&[4, 0, 0, 0]))
}
impl FlowTunnel {
fn encapsulate<'a>(
&mut self,
packet: &[u8],
scratch: &'a mut [u8],
flow: FlowId,
) -> (TunnResult<'a>, FlowId) {
let result = self.tunn.encapsulate(packet, scratch);
if is_data(&result) {
return (result, flow);
}
if self.queued.len() < 256 {
self.queued.push_back(flow);
}
(result, 0)
}
fn decapsulate<'a>(
&mut self,
packet: &[u8],
scratch: &'a mut [u8],
) -> (TunnResult<'a>, FlowId) {
let result = self.tunn.decapsulate(None, packet, scratch);
let mut flow = 0;
if packet.is_empty() {
if is_data(&result) {
flow = self.queued.pop_front().unwrap_or(0);
} else if let Some(pending) = self.queued.pop_front() {
self.queued.push_back(pending);
}
}
(result, flow)
}
fn update_timers<'a>(&mut self, scratch: &'a mut [u8]) -> TunnResult<'a> {
let result = self.tunn.update_timers(scratch);
if matches!(
result,
TunnResult::Err(boringtun::noise::errors::WireGuardError::ConnectionExpired)
) {
self.queued.clear();
}
result
}
}
#[derive(Debug, Default)]
struct PeerCounters {
tx_packets: AtomicU64,
tx_bytes: AtomicU64,
rx_packets: AtomicU64,
rx_bytes: AtomicU64,
dropped_wrong_source: AtomicU64,
dropped_oversize: AtomicU64,
dropped_no_session: AtomicU64,
protocol_errors: AtomicU64,
}
/// A snapshot of one peer's tunnel counters.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct PeerStats {
/// Plaintext packets encrypted and sent to this peer.
pub tx_packets: u64,
/// Plaintext bytes encrypted and sent to this peer.
pub tx_bytes: u64,
/// Plaintext packets decrypted from this peer and given to the OS.
pub rx_packets: u64,
/// Plaintext bytes decrypted from this peer and given to the OS.
pub rx_bytes: u64,
/// Packets dropped because their source was not this peer's address.
pub dropped_wrong_source: u64,
/// Packets dropped because they did not fit in one link datagram.
pub dropped_oversize: u64,
/// Packets dropped because there was no session to encrypt with yet.
pub dropped_no_session: u64,
/// WireGuard protocol errors, including packets that failed to decrypt.
pub protocol_errors: u64,
}
/// Whether a peer's tunnel has completed a handshake.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PeerHealth {
/// Time since the last successful WireGuard handshake.
pub since_handshake: Option<Duration>,
}
impl PeerHealth {
/// Whether the tunnel has ever completed a handshake.
pub fn is_up(&self) -> bool {
self.since_handshake.is_some()
}
}
struct Peer {
endpoint_id: EndpointId,
public_key: WgPublicKey,
overlay_v4: Mutex<Option<Ipv4Addr>>,
tunn: Mutex<FlowTunnel>,
link: SharedLink,
counters: Arc<PeerCounters>,
task: Mutex<Option<JoinHandle<()>>>,
}
impl std::fmt::Debug for Peer {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Peer")
.field("peer", &self.endpoint_id.fmt_short().to_string())
.field("public_key", &self.public_key)
.finish()
}
}
impl Drop for Peer {
fn drop(&mut self) {
if let Ok(mut guard) = self.task.lock()
&& let Some(task) = guard.take()
{
task.abort();
}
}
}
impl Peer {
fn stats(&self) -> PeerStats {
PeerStats {
tx_packets: self.counters.tx_packets.load(Ordering::Relaxed),
tx_bytes: self.counters.tx_bytes.load(Ordering::Relaxed),
rx_packets: self.counters.rx_packets.load(Ordering::Relaxed),
rx_bytes: self.counters.rx_bytes.load(Ordering::Relaxed),
dropped_wrong_source: self.counters.dropped_wrong_source.load(Ordering::Relaxed),
dropped_oversize: self.counters.dropped_oversize.load(Ordering::Relaxed),
dropped_no_session: self.counters.dropped_no_session.load(Ordering::Relaxed),
protocol_errors: self.counters.protocol_errors.load(Ordering::Relaxed),
}
}
fn health(&self) -> PeerHealth {
let guard = match self.tunn.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
PeerHealth {
since_handshake: guard.tunn.time_since_last_handshake(),
}
}
}
/// What a peer's tunnel looks like from outside.
#[derive(Debug, Clone)]
pub struct PeerSummary {
/// The peer's control plane identity.
pub endpoint_id: EndpointId,
/// The peer's WireGuard public key.
pub public_key: WgPublicKey,
/// The overlay address it holds, once agreed.
pub overlay_address_v4: Option<Ipv4Addr>,
/// Whether the tunnel has handshaken.
pub health: PeerHealth,
/// Traffic counters.
pub stats: PeerStats,
/// What the transport reports about the path carrying this tunnel.
pub path: String,
/// Largest datagram the link accepts.
pub max_datagram: usize,
}
struct Inner {
network: NetworkId,
private_key: WgSecretKey,
peers: RwLock<HashMap<WgPublicKey, Arc<Peer>>>,
sink: Arc<dyn PacketSink>,
next_index: AtomicU32,
}
impl std::fmt::Debug for Inner {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Inner")
.field("network", &self.network.fmt_short())
.finish()
}
}
/// The userspace WireGuard tunnels of one network.
#[derive(Debug)]
pub struct WireguardDevice {
inner: Arc<Inner>,
tasks: Vec<JoinHandle<()>>,
}
impl WireguardDevice {
/// Starts the WireGuard tunnels for one network.
pub fn start(network: NetworkId, private_key: WgSecretKey, sink: Arc<dyn PacketSink>) -> Self {
let inner = Arc::new(Inner {
network,
private_key,
peers: RwLock::new(HashMap::new()),
sink,
next_index: AtomicU32::new(1),
});
let timers = tokio::spawn(drive_timers(Arc::clone(&inner)));
Self {
inner,
tasks: vec![timers],
}
}
/// Encrypts an IP packet and sends it to a peer over its direct link.
pub fn carry(&self, peer: EndpointId, packet: &[u8]) -> bool {
let found = read_lock(&self.inner.peers)
.values()
.find(|candidate| candidate.endpoint_id == peer)
.cloned();
let Some(peer) = found else {
return false;
};
let mut scratch = [0u8; SCRATCH];
let flow = ip_flow(packet);
let result = {
let mut tunn = match peer.tunn.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
let (result, selected_flow) = tunn.encapsulate(packet, &mut scratch, flow);
match result {
TunnResult::WriteToNetwork(out) => Some((out.len(), selected_flow)),
_ => None,
}
};
let Some((len, selected_flow)) = result else {
peer.counters
.dropped_no_session
.fetch_add(1, Ordering::Relaxed);
return false;
};
send_to_peer_flow(&peer, &scratch[..len], selected_flow);
peer.counters.tx_packets.fetch_add(1, Ordering::Relaxed);
peer.counters
.tx_bytes
.fetch_add(packet.len() as u64, Ordering::Relaxed);
true
}
/// Adds or replaces a peer and starts its tunnel.
pub fn add_peer(
&self,
endpoint_id: EndpointId,
public_key: WgPublicKey,
overlay_v4: Option<Ipv4Addr>,
link: SharedLink,
keepalive: Option<u16>,
) -> Result<(), PluginError> {
if public_key == self.inner.private_key.public() {
return Err(PluginError::Rejected(
"refusing to add ourselves as a WireGuard peer".into(),
));
}
let index = self.inner.next_index.fetch_add(1, Ordering::Relaxed);
let tunn = Tunn::new(
self.inner.private_key.to_static_secret(),
public_key.into_x25519(),
None,
keepalive,
index,
None,
);
let peer = Arc::new(Peer {
endpoint_id,
public_key,
overlay_v4: Mutex::new(overlay_v4),
tunn: Mutex::new(FlowTunnel {
tunn,
queued: VecDeque::new(),
}),
link,
counters: Arc::new(PeerCounters::default()),
task: Mutex::new(None),
});
let task = tokio::spawn(read_from_link(Arc::clone(&self.inner), Arc::clone(&peer)));
if let Ok(mut guard) = peer.task.lock() {
*guard = Some(task);
}
write_lock(&self.inner.peers).insert(public_key, Arc::clone(&peer));
kick_handshake(&peer);
Ok(())
}
/// Removes a peer and stops its tunnel.
pub fn remove_peer(&self, public_key: &WgPublicKey) {
write_lock(&self.inner.peers).remove(public_key);
}
/// Removes every peer whose key is not in `keep`.
pub fn retain_peers(&self, keep: &[WgPublicKey]) {
let stale: Vec<WgPublicKey> = read_lock(&self.inner.peers)
.keys()
.filter(|key| !keep.contains(key))
.copied()
.collect();
for key in stale {
self.remove_peer(&key);
}
}
/// Whether a peer's tunnel exists.
pub fn has_peer(&self, public_key: &WgPublicKey) -> bool {
read_lock(&self.inner.peers).contains_key(public_key)
}
/// A snapshot of every peer.
pub fn peers(&self) -> Vec<PeerSummary> {
let mut peers: Vec<PeerSummary> = read_lock(&self.inner.peers)
.values()
.map(|peer| {
let overlay_v4 = match peer.overlay_v4.lock() {
Ok(guard) => *guard,
Err(poisoned) => *poisoned.into_inner(),
};
PeerSummary {
endpoint_id: peer.endpoint_id,
public_key: peer.public_key,
overlay_address_v4: overlay_v4,
health: peer.health(),
stats: peer.stats(),
path: peer.link.path_description(),
max_datagram: peer.link.max_datagram_size(),
}
})
.collect();
peers.sort_by_key(|peer| peer.public_key);
peers
}
}
impl Drop for WireguardDevice {
fn drop(&mut self) {
for task in &self.tasks {
task.abort();
}
}
}
fn read_lock<T>(lock: &RwLock<T>) -> std::sync::RwLockReadGuard<'_, T> {
match lock.read() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
}
}
fn write_lock<T>(lock: &RwLock<T>) -> std::sync::RwLockWriteGuard<'_, T> {
match lock.write() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
}
}
fn kick_handshake(peer: &Peer) {
let mut scratch = vec![0u8; SCRATCH];
let len = {
let mut tunn = match peer.tunn.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
match tunn.tunn.format_handshake_initiation(&mut scratch, false) {
TunnResult::WriteToNetwork(out) => Some(out.len()),
_ => None,
}
};
if let Some(len) = len {
send_to_peer(peer, &scratch[..len]);
}
}
fn send_to_peer(peer: &Peer, payload: &[u8]) {
send_to_peer_flow(peer, payload, 0);
}
fn send_to_peer_flow(peer: &Peer, payload: &[u8], flow: FlowId) {
match peer.link.send_flow(Bytes::copy_from_slice(payload), flow) {
Ok(()) => {}
Err(TransportError::TooLarge { .. }) => {
peer.counters
.dropped_oversize
.fetch_add(1, Ordering::Relaxed);
}
Err(TransportError::Closed) => {}
Err(err) => {
tracing::trace!(%err, "dropping a WireGuard packet the link refused");
}
}
}
async fn read_from_link(inner: Arc<Inner>, peer: Arc<Peer>) {
let mut scratch = vec![0u8; SCRATCH];
loop {
let Some(datagram) = peer.link.recv().await else {
return;
};
let mut input: Option<&[u8]> = Some(&datagram);
loop {
let outcome = {
let mut tunn = match peer.tunn.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
let (result, flow) = tunn.decapsulate(input.unwrap_or(&[]), &mut scratch);
match result {
TunnResult::WriteToNetwork(out) => Outcome::ToNetwork(out.len(), flow),
TunnResult::WriteToTunnelV6(out, _) => Outcome::ToTunnel(out.len()),
TunnResult::WriteToTunnelV4(out, _) => Outcome::ToTunnel(out.len()),
TunnResult::Done => Outcome::Done,
TunnResult::Err(err) => {
tracing::trace!(?err, "WireGuard decapsulation failed");
Outcome::Failed
}
}
};
match outcome {
Outcome::ToNetwork(len, flow) => {
send_to_peer_flow(&peer, &scratch[..len], flow);
input = None;
continue;
}
Outcome::ToTunnel(len) => {
let payload = Bytes::copy_from_slice(&scratch[..len]);
inner
.sink
.deliver(inner.network, peer.endpoint_id, payload)
.await;
peer.counters.rx_packets.fetch_add(1, Ordering::Relaxed);
peer.counters
.rx_bytes
.fetch_add(len as u64, Ordering::Relaxed);
break;
}
Outcome::Failed => {
peer.counters
.protocol_errors
.fetch_add(1, Ordering::Relaxed);
break;
}
Outcome::Done => break,
}
}
}
}
enum Outcome {
ToNetwork(usize, FlowId),
ToTunnel(usize),
Done,
Failed,
}
async fn drive_timers(inner: Arc<Inner>) {
let mut ticker = tokio::time::interval(TIMER_INTERVAL);
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
loop {
ticker.tick().await;
let peers: Vec<Arc<Peer>> = read_lock(&inner.peers).values().cloned().collect();
for peer in peers {
let mut scratch = vec![0u8; SCRATCH];
let len = {
let mut tunn = match peer.tunn.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
match tunn.update_timers(&mut scratch) {
TunnResult::WriteToNetwork(out) => Some(out.len()),
TunnResult::Err(err) => {
tracing::trace!(?err, "WireGuard timer error");
None
}
_ => None,
}
};
if let Some(len) = len {
send_to_peer(&peer, &scratch[..len]);
}
}
}
}
+160
View File
@@ -0,0 +1,160 @@
//! WireGuard key material for pure WireGuard transport.
//!
//! Keys belong to this protocol plugin alone. They are not derived from
//! the iroh device key and not derived from the network secret, so compromising
//! or rotating one does not affect the others.
//!
//! Keys are X25519, encoded the way WireGuard encodes them: standard base64
//! with padding, 44 characters.
use boringtun::x25519;
use data_encoding::BASE64;
use zeroize::{Zeroize, Zeroizing};
use tsunagi::dataplane::PluginError;
/// Length of a raw WireGuard key in bytes.
pub const KEY_LEN: usize = 32;
/// Length of the base64 text form of a key.
pub const KEY_TEXT_LEN: usize = 44;
/// Applies the X25519 clamping WireGuard applies to private keys.
fn clamp(bytes: &mut [u8; KEY_LEN]) {
bytes[0] &= 248;
bytes[31] &= 127;
bytes[31] |= 64;
}
/// A WireGuard public key.
///
/// Public, safe to log, and the identity a peer is known by inside the overlay.
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct WgPublicKey([u8; KEY_LEN]);
impl WgPublicKey {
/// Wraps raw key bytes.
pub fn from_bytes(bytes: [u8; KEY_LEN]) -> Self {
Self(bytes)
}
/// The raw key bytes.
pub fn as_bytes(&self) -> &[u8; KEY_LEN] {
&self.0
}
/// The key in the form the WireGuard implementation expects.
pub(crate) fn into_x25519(self) -> x25519::PublicKey {
x25519::PublicKey::from(self.0)
}
/// Whether this is the all-zero key, which is never a valid peer.
pub fn is_zero(&self) -> bool {
self.0 == [0u8; KEY_LEN]
}
/// The base64 form WireGuard uses.
pub fn encode(&self) -> String {
BASE64.encode(&self.0)
}
/// Parses the base64 form WireGuard uses.
pub fn decode(text: &str) -> Result<Self, PluginError> {
if text.len() != KEY_TEXT_LEN {
return Err(PluginError::Rejected(format!(
"a WireGuard key is {KEY_TEXT_LEN} base64 characters, got {}",
text.len()
)));
}
let raw = BASE64
.decode(text.as_bytes())
.map_err(|_| PluginError::Rejected("key is not valid base64".into()))?;
let bytes = <[u8; KEY_LEN]>::try_from(raw.as_slice())
.map_err(|_| PluginError::Rejected("key is not 32 bytes".into()))?;
Ok(Self(bytes))
}
/// A short prefix for logs and diagnostics.
pub fn fmt_short(&self) -> String {
self.encode().chars().take(8).collect()
}
}
impl std::fmt::Display for WgPublicKey {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.encode())
}
}
impl std::fmt::Debug for WgPublicKey {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "WgPublicKey({})", self.fmt_short())
}
}
/// A WireGuard private key.
///
/// Clamped on creation so it matches what `wg genkey` produces. Zeroized on
/// drop, and prints as `[redacted]` in Debug so it cannot leak into logs.
#[derive(Clone)]
pub struct WgSecretKey {
bytes: Zeroizing<[u8; KEY_LEN]>,
public: WgPublicKey,
}
impl WgSecretKey {
/// Generates a fresh clamped random private key.
pub fn generate() -> Self {
let mut bytes = [0u8; KEY_LEN];
rand::fill(&mut bytes);
clamp(&mut bytes);
let secret = x25519::StaticSecret::from(bytes);
let public = WgPublicKey(*x25519::PublicKey::from(&secret).as_bytes());
Self {
bytes: Zeroizing::new(bytes),
public,
}
}
/// Wraps existing key bytes, clamping them to a valid X25519 private key.
pub fn from_bytes(raw: &[u8; KEY_LEN]) -> Self {
let mut bytes = *raw;
clamp(&mut bytes);
let secret = x25519::StaticSecret::from(bytes);
let public = WgPublicKey(*x25519::PublicKey::from(&secret).as_bytes());
Self {
bytes: Zeroizing::new(bytes),
public,
}
}
/// The matching public key.
pub fn public(&self) -> WgPublicKey {
self.public
}
/// The key in the form the WireGuard implementation expects.
pub(crate) fn to_static_secret(&self) -> x25519::StaticSecret {
x25519::StaticSecret::from(*self.bytes)
}
/// The raw key bytes.
pub fn as_bytes(&self) -> &[u8; KEY_LEN] {
&self.bytes
}
}
impl Drop for WgSecretKey {
fn drop(&mut self) {
self.bytes.zeroize();
}
}
impl std::fmt::Debug for WgSecretKey {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("WgSecretKey")
.field("bytes", &"[redacted]")
.field("public", &self.public)
.finish()
}
}
+28
View File
@@ -0,0 +1,28 @@
//! The pure WireGuard protocol plugin for Tsunagi: WireGuard cryptography over direct UDP.
//!
//! Deliberately a separate crate from the core, ensuring clear boundaries:
//! - Direct UDP datagram transport without QUIC encapsulation or double-encryption.
//! - Standard WireGuard cryptography via `boringtun`.
//! - Peer discovery and key exchange managed through the Tsunagi control plane.
#![deny(unsafe_code)]
#![warn(missing_docs)]
pub mod announcement;
pub mod device;
pub mod keys;
pub mod plugin;
pub mod store;
pub mod transport;
pub use announcement::{
ANNOUNCEMENT_VERSION, ValidatedAnnouncement, WG_PROTOCOL, WG_WIRE_VERSION, WgAnnouncement,
};
pub use device::{PeerHealth, PeerStats, PeerSummary, WireguardDevice};
pub use keys::{WgPublicKey, WgSecretKey};
pub use plugin::{
DEFAULT_MTU, MIN_MTU, NetworkOverview, PeerOverview, WIREGUARD_OVERHEAD, WireguardConfig,
WireguardPlugin, local_ip_candidates,
};
pub use store::WgKeyStore;
pub use transport::{DEFAULT_PORT, MAX_UDP_DATAGRAM_SIZE, WireguardLink, WireguardTransport};
+742
View File
@@ -0,0 +1,742 @@
//! The pure WireGuard IP plugin implementation.
//!
//! Provides [`WireguardPlugin`] implementing [`IpPlugin`].
use std::collections::{BTreeSet, HashMap};
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::path::PathBuf;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;
use iroh::EndpointId;
use tokio::sync::{mpsc, oneshot};
use tokio::task::JoinHandle;
use tsunagi::BoxFuture;
use tsunagi::dataplane::PacketSink;
use tsunagi::dataplane::transport::SharedLink;
use tsunagi::dataplane::{IpPlugin, PluginCapability, PluginContext, PluginError, ProtocolOption};
use tsunagi::identity::NetworkId;
use tsunagi::state::Ipv4Range;
use crate::announcement::{
ANNOUNCEMENT_VERSION, ValidatedAnnouncement, WG_PROTOCOL, WgAnnouncement,
};
use crate::device::{PeerSummary, WireguardDevice};
use crate::keys::{WgPublicKey, WgSecretKey};
use crate::store::WgKeyStore;
use crate::transport::WireguardTransport;
/// Smallest interface MTU the overlay accepts.
pub const MIN_MTU: u32 = 576;
/// Default interface MTU.
pub const DEFAULT_MTU: u32 = 1280;
/// WireGuard packet overhead.
pub const WIREGUARD_OVERHEAD: u32 = 32;
/// Discovers local network interface IPs to announce to peers, excluding loopback and overlay interfaces.
pub fn local_ip_candidates() -> Vec<IpAddr> {
let mut ips = Vec::new();
for iface in netdev::get_interfaces() {
if iface.is_up() && !iface.is_loopback() {
let name_lower = iface.name.to_lowercase();
if name_lower.starts_with("tsun") || name_lower.contains("tsunagi") {
continue;
}
for ip in iface.ipv4 {
let addr = ip.addr();
if !addr.is_loopback() && !addr.is_unspecified() {
ips.push(IpAddr::V4(addr));
}
}
for ip in iface.ipv6 {
let addr = ip.addr();
if !addr.is_loopback() && !addr.is_unspecified() {
ips.push(IpAddr::V6(addr));
}
}
}
}
ips
}
/// Configuration of the pure WireGuard plugin.
#[derive(Debug, Clone)]
pub struct WireguardConfig {
/// Directory for the plugin's own key store.
pub state_dir: PathBuf,
/// WireGuard keepalive in seconds.
pub keepalive: Option<u16>,
/// The largest packet a tunnel will carry.
pub mtu: u32,
/// UDP port to listen on.
pub port: Option<u16>,
/// How long to coalesce changes before reconciling.
pub reconcile_debounce: Duration,
/// Periodic reconcile interval.
pub reconcile_interval: Duration,
}
impl WireguardConfig {
/// Creates a configuration rooted at `state_dir`.
pub fn new(state_dir: impl Into<PathBuf>) -> Self {
Self {
state_dir: state_dir.into(),
keepalive: Some(25),
mtu: DEFAULT_MTU,
port: None,
reconcile_debounce: Duration::from_millis(200),
reconcile_interval: Duration::from_secs(15),
}
}
/// Sets the interface MTU.
pub fn with_mtu(mut self, mtu: u32) -> Self {
self.mtu = mtu;
self
}
/// Sets the UDP listen port.
pub fn with_port(mut self, port: u16) -> Self {
self.port = Some(port);
self
}
/// Path of the plugin's key store.
pub fn key_store_path(&self) -> PathBuf {
self.state_dir.join("wireguard.sqlite")
}
}
/// What this agent has set up for one network.
#[derive(Debug, Clone)]
pub struct NetworkOverview {
/// The network.
pub network: NetworkId,
/// The largest packet a tunnel in this network will carry.
pub mtu: u32,
/// This agent's WireGuard public key in this network.
pub public_key: WgPublicKey,
/// This agent's overlay address, once agreed.
pub overlay_address_v4: Option<Ipv4Addr>,
/// The IPv4 overlay range in use.
pub ipv4_range: Option<Ipv4Range>,
/// Peers this agent knows about.
pub peers: Vec<PeerOverview>,
}
/// One peer of the overlay.
#[derive(Debug, Clone)]
pub struct PeerOverview {
/// The peer's control plane identity.
pub endpoint_id: EndpointId,
/// The peer's WireGuard public key.
pub public_key: WgPublicKey,
/// The overlay address the network agreed it holds.
pub overlay_address_v4: Option<Ipv4Addr>,
/// Whether a data plane link to it exists.
pub has_link: bool,
/// The running tunnel, once there is a link.
pub tunnel: Option<PeerSummary>,
}
impl PeerOverview {
/// Whether the tunnel to this peer has handshaken and can carry traffic.
pub fn is_up(&self) -> bool {
self.tunnel
.as_ref()
.is_some_and(|tunnel| tunnel.health.is_up())
}
}
#[derive(Debug)]
struct NetworkState {
key: WgSecretKey,
device: Option<Arc<WireguardDevice>>,
announcements: HashMap<EndpointId, ValidatedAnnouncement>,
links: HashMap<EndpointId, SharedLink>,
allocations: HashMap<EndpointId, Ipv4Addr>,
ipv4_range: Option<Ipv4Range>,
}
#[derive(Debug, Default)]
struct Shared {
networks: HashMap<NetworkId, NetworkState>,
}
#[derive(Debug)]
enum Command {
Prepare(NetworkId),
Sync(NetworkId),
Link {
network: NetworkId,
peer: EndpointId,
link: SharedLink,
},
Teardown(NetworkId),
Stop(oneshot::Sender<()>),
}
struct Worker {
config: WireguardConfig,
transport: Arc<WireguardTransport>,
local_id: OnceLock<EndpointId>,
store: WgKeyStore,
shared: Mutex<Shared>,
context: OnceLock<PluginContext>,
sink: OnceLock<Arc<dyn PacketSink>>,
}
impl std::fmt::Debug for Worker {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Worker")
.field("store", &self.store.path())
.field("port", &self.transport.bound_port())
.finish()
}
}
/// The pure WireGuard data plane plugin.
#[derive(Debug)]
pub struct WireguardPlugin {
worker: Arc<Worker>,
commands: mpsc::Sender<Command>,
task: Mutex<Option<JoinHandle<()>>>,
}
impl WireguardPlugin {
/// Supported options for pure WireGuard.
pub const OPTIONS: &'static [ProtocolOption] = &[
ProtocolOption {
key: "port",
value: "PORT",
help: "UDP port to listen on for WireGuard packets",
default: Some("51820"),
},
ProtocolOption {
key: "keepalive",
value: "SECONDS",
help: "keeps a tunnel and NAT mapping warm; 0 turns it off",
default: Some("25"),
},
ProtocolOption {
key: "mtu",
value: "BYTES",
help: "largest packet a tunnel will carry, at least 576",
default: Some("1280"),
},
];
/// Applies `key=value` settings to a configuration.
pub fn configure(
mut config: WireguardConfig,
options: &[(String, String)],
) -> Result<WireguardConfig, PluginError> {
for (key, value) in options {
match key.as_str() {
"port" => {
let port: u16 = value.parse().map_err(|_| {
PluginError::Other(format!("port={value} is not a valid port number"))
})?;
config = config.with_port(port);
}
"keepalive" => {
let seconds: u16 = value.parse().map_err(|_| {
PluginError::Other(format!("keepalive={value} is not a number of seconds"))
})?;
config.keepalive = (seconds > 0).then_some(seconds);
}
"mtu" => {
let mtu: u32 = value.parse().map_err(|_| {
PluginError::Other(format!("mtu={value} is not a number of bytes"))
})?;
config = config.with_mtu(mtu);
}
other => {
let known: Vec<&str> = Self::OPTIONS.iter().map(|spec| spec.key).collect();
return Err(PluginError::Other(format!(
"`{other}` is not a setting of {WG_PROTOCOL}; it takes {}",
known.join(", ")
)));
}
}
}
Ok(config)
}
/// Opens the plugin's key store, binds the UDP transport and starts the reconciliation worker.
pub async fn open(
config: WireguardConfig,
transport: Arc<WireguardTransport>,
) -> Result<Arc<Self>, PluginError> {
if config.mtu < MIN_MTU {
return Err(PluginError::Other(format!(
"an MTU of {} is below the {MIN_MTU} bytes every IPv4 host must be able to reassemble",
config.mtu
)));
}
let path = config.key_store_path();
let store = tokio::task::spawn_blocking(move || WgKeyStore::open(path))
.await
.map_err(|err| PluginError::Other(format!("key store task failed: {err}")))??;
let worker = Arc::new(Worker {
config,
transport,
local_id: OnceLock::new(),
store,
shared: Mutex::new(Shared::default()),
context: OnceLock::new(),
sink: OnceLock::new(),
});
let (commands, receiver) = mpsc::channel(64);
let task = tokio::spawn(run(Arc::clone(&worker), receiver));
Ok(Arc::new(Self {
worker,
commands,
task: Mutex::new(Some(task)),
}))
}
/// What this agent has set up for a network.
pub fn overview(&self, network: NetworkId) -> Option<NetworkOverview> {
let shared = self.worker.lock_shared();
let state = shared.networks.get(&network)?;
let tunnels: HashMap<WgPublicKey, PeerSummary> = state
.device
.as_ref()
.map(|device| {
device
.peers()
.into_iter()
.map(|summary| (summary.public_key, summary))
.collect()
})
.unwrap_or_default();
let mut peers: Vec<PeerOverview> = state
.announcements
.iter()
.map(|(endpoint_id, announcement)| PeerOverview {
endpoint_id: *endpoint_id,
public_key: announcement.public_key,
overlay_address_v4: state.allocations.get(endpoint_id).copied(),
has_link: state.links.contains_key(endpoint_id),
tunnel: tunnels.get(&announcement.public_key).cloned(),
})
.collect();
peers.sort_by_key(|peer| peer.public_key);
Some(NetworkOverview {
network,
mtu: self.worker.config.mtu,
public_key: state.key.public(),
overlay_address_v4: state.allocations.get(&self.worker.local_id()).copied(),
ipv4_range: state.ipv4_range,
peers,
})
}
fn nudge(&self, command: Command) {
if let Err(err) = self.commands.try_send(command) {
tracing::debug!(%err, "WireGuard command queue is busy");
}
}
}
impl Worker {
fn local_id(&self) -> EndpointId {
self.local_id.get().copied().unwrap_or_else(|| {
EndpointId::from_bytes(&[1u8; 32]).unwrap_or_else(|_| unreachable!("a fixed valid key"))
})
}
fn lock_shared(&self) -> std::sync::MutexGuard<'_, Shared> {
match self.shared.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
}
}
fn report(&self, network: NetworkId, reason: impl std::fmt::Display) {
tracing::warn!(network = %network.fmt_short(), %reason, "pure WireGuard plugin error");
if let Some(context) = self.context.get() {
context.report_error(network, WG_PROTOCOL, reason.to_string());
}
}
fn request_reannounce(&self, network: NetworkId) {
if let Some(context) = self.context.get() {
context.request_reannounce(network);
}
}
async fn prepare(self: &Arc<Self>, network: NetworkId) -> Result<bool, PluginError> {
let existing = {
let shared = self.lock_shared();
shared
.networks
.get(&network)
.map(|state| state.device.is_some())
};
if let Some(has_device) = existing {
if has_device {
return Ok(false);
}
self.ensure_device(network).await?;
return Ok(false);
}
let worker = Arc::clone(self);
let key = tokio::task::spawn_blocking(move || worker.store.load_or_create(network))
.await
.map_err(|err| PluginError::Other(format!("key store task failed: {err}")))??;
{
let mut shared = self.lock_shared();
shared.networks.entry(network).or_insert(NetworkState {
key,
device: None,
announcements: HashMap::new(),
links: HashMap::new(),
allocations: HashMap::new(),
ipv4_range: None,
});
}
let device = self.ensure_device(network).await;
if let Err(err) = device {
self.report(network, err);
}
Ok(true)
}
async fn ensure_device(&self, network: NetworkId) -> Result<(), PluginError> {
let key = {
let shared = self.lock_shared();
match shared.networks.get(&network) {
Some(state) if state.device.is_none() => state.key.clone(),
_ => return Ok(()),
}
};
let sink = match self.sink.get() {
Some(sink) => Arc::clone(sink),
None => Arc::new(tsunagi::dataplane::DiscardPackets) as Arc<dyn PacketSink>,
};
let device = Arc::new(WireguardDevice::start(network, key, sink));
let mut shared = self.lock_shared();
if let Some(state) = shared.networks.get_mut(&network) {
state.device = Some(device);
}
Ok(())
}
fn sync(&self, network: NetworkId) {
let mut shared = self.lock_shared();
let Some(state) = shared.networks.get_mut(&network) else {
return;
};
let Some(device) = state.device.clone() else {
return;
};
let allocations = state.allocations.clone();
let mut wanted: Vec<WgPublicKey> = Vec::new();
for (endpoint_id, announcement) in &state.announcements {
let Some(link) = state.links.get(endpoint_id) else {
continue;
};
if link.is_closed() {
continue;
}
wanted.push(announcement.public_key);
if device.has_peer(&announcement.public_key) {
continue;
}
let peer_v4 = allocations.get(endpoint_id).copied();
if let Err(err) = device.add_peer(
*endpoint_id,
announcement.public_key,
peer_v4,
Arc::clone(link),
self.config.keepalive,
) {
tracing::debug!(%err, "cannot start WireGuard tunnel");
}
}
device.retain_peers(&wanted);
}
async fn teardown(&self, network: NetworkId) {
self.lock_shared().networks.remove(&network);
}
fn known_networks(&self) -> Vec<NetworkId> {
self.lock_shared().networks.keys().copied().collect()
}
}
async fn run(worker: Arc<Worker>, mut commands: mpsc::Receiver<Command>) {
let mut pending: BTreeSet<NetworkId> = BTreeSet::new();
let mut deadline: Option<tokio::time::Instant> = None;
let mut ticker = tokio::time::interval(worker.config.reconcile_interval);
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
ticker.tick().await;
loop {
let wait_until = deadline;
tokio::select! {
biased;
command = commands.recv() => {
let Some(command) = command else { break };
match command {
Command::Prepare(network) => {
match worker.prepare(network).await {
Ok(true) => worker.request_reannounce(network),
Ok(false) => {}
Err(err) => worker.report(network, err),
}
pending.insert(network);
}
Command::Sync(network) => {
pending.insert(network);
}
Command::Link { network, peer, link } => {
{
let mut shared = worker.lock_shared();
if let Some(state) = shared.networks.get_mut(&network) {
state.links.insert(peer, link);
}
}
pending.insert(network);
}
Command::Teardown(network) => {
pending.remove(&network);
worker.teardown(network).await;
continue;
}
Command::Stop(reply) => {
for network in worker.known_networks() {
worker.teardown(network).await;
}
let _ = reply.send(());
return;
}
}
deadline = Some(tokio::time::Instant::now() + worker.config.reconcile_debounce);
}
_ = async {
match wait_until {
Some(at) => tokio::time::sleep_until(at).await,
None => std::future::pending::<()>().await,
}
}, if wait_until.is_some() => {
deadline = None;
for network in std::mem::take(&mut pending) {
worker.sync(network);
}
}
_ = ticker.tick() => {
for network in worker.known_networks() {
if let Err(err) = worker.ensure_device(network).await {
tracing::debug!(%err, "WireGuard device still unavailable");
}
worker.sync(network);
}
}
}
}
}
impl IpPlugin for WireguardPlugin {
fn protocol_id(&self) -> &str {
WG_PROTOCOL
}
fn protocol_version(&self) -> u16 {
ANNOUNCEMENT_VERSION
}
fn options(&self) -> &'static [ProtocolOption] {
Self::OPTIONS
}
fn attach(&self, context: PluginContext) {
if let Some(local) = context.local_endpoint_id() {
let _ = self.worker.local_id.set(local);
}
let _ = self.worker.sink.set(context.packet_sink());
let _ = self.worker.context.set(context);
}
fn on_network_activated(&self, network: NetworkId) {
self.nudge(Command::Prepare(network));
}
fn local_capability(
&self,
network: NetworkId,
) -> Result<Option<PluginCapability>, PluginError> {
let shared = self.worker.lock_shared();
let Some(state) = shared.networks.get(&network) else {
drop(shared);
self.nudge(Command::Prepare(network));
return Ok(None);
};
let bound_port = self.worker.transport.bound_port();
let addrs = local_ip_candidates();
let announcement = WgAnnouncement::new(network, &state.key.public(), bound_port, addrs);
Ok(Some(PluginCapability {
protocol: WG_PROTOCOL.to_string(),
version: ANNOUNCEMENT_VERSION,
enabled: true,
data: announcement.encode()?,
}))
}
fn on_peer_capability(
&self,
network: NetworkId,
peer: EndpointId,
capability: &PluginCapability,
) -> Result<(), PluginError> {
let local_key = {
let shared = self.worker.lock_shared();
match shared.networks.get(&network) {
Some(state) => state.key.public(),
None => {
drop(shared);
self.nudge(Command::Prepare(network));
return Err(PluginError::Unavailable(
"WireGuard is not ready for this network yet".into(),
));
}
}
};
let validated = WgAnnouncement::decode_and_validate(&capability.data, network, &local_key)?;
let socket_addrs: Vec<SocketAddr> = validated
.addrs
.iter()
.map(|ip| SocketAddr::new(*ip, validated.port))
.collect();
// Record candidate addresses in transport and optionally notify if we are the acceptor
let local_id = self.worker.local_id();
let we_accept = local_id.as_bytes() > peer.as_bytes();
self.worker
.transport
.notify_peer_capability(network, peer, &socket_addrs, we_accept);
let changed = {
let mut shared = self.worker.lock_shared();
match shared.networks.get_mut(&network) {
Some(state) => {
state.announcements.insert(peer, validated.clone()) != Some(validated)
}
None => false,
}
};
if changed {
self.nudge(Command::Sync(network));
}
Ok(())
}
fn on_address_allocation(
&self,
network: NetworkId,
range: Ipv4Range,
allocations: &[(EndpointId, Ipv4Addr)],
) {
let changed = {
let mut shared = self.worker.lock_shared();
match shared.networks.get_mut(&network) {
Some(state) => {
let fresh: HashMap<EndpointId, Ipv4Addr> =
allocations.iter().copied().collect();
let changed = state.allocations != fresh || state.ipv4_range != Some(range);
state.allocations = fresh;
state.ipv4_range = Some(range);
changed
}
None => false,
}
};
if changed {
self.nudge(Command::Sync(network));
}
}
fn carry(&self, network: NetworkId, peer: EndpointId, packet: bytes::Bytes) -> bool {
let shared = self.worker.lock_shared();
shared
.networks
.get(&network)
.and_then(|state| state.device.as_ref())
.is_some_and(|device| device.carry(peer, &packet))
}
fn on_peer_link(&self, network: NetworkId, peer: EndpointId, link: SharedLink) {
self.nudge(Command::Link {
network,
peer,
link,
});
}
fn on_peer_gone(&self, network: NetworkId, peer: EndpointId) {
let removed = {
let mut shared = self.worker.lock_shared();
match shared.networks.get_mut(&network) {
Some(state) => {
let had_link = state.links.remove(&peer).is_some();
state.announcements.remove(&peer).is_some() || had_link
}
None => false,
}
};
if removed {
self.nudge(Command::Sync(network));
}
}
fn on_network_deactivated(&self, network: NetworkId) {
self.nudge(Command::Teardown(network));
}
fn on_network_forgotten(&self, network: NetworkId) {
if let Err(err) = self.worker.store.forget(network) {
tracing::warn!(%err, "could not forget WireGuard key for network");
}
}
fn shutdown<'a>(&'a self) -> BoxFuture<'a, ()> {
Box::pin(async move {
let (tx, rx) = oneshot::channel();
if self.commands.send(Command::Stop(tx)).await.is_ok() {
let _ = rx.await;
}
let task = match self.task.lock() {
Ok(mut guard) => guard.take(),
Err(poisoned) => poisoned.into_inner().take(),
};
if let Some(task) = task {
let _ = task.await;
}
})
}
}
+160
View File
@@ -0,0 +1,160 @@
//! The plugin's own key store.
//!
//! Deliberately a separate SQLite file from the agent's `state.sqlite`: plugin
//! keys are not the iroh identity and not the network secret, and their
//! lifecycle is the plugin's business alone.
//!
//! One key per network, so a participant presents a different WireGuard
//! identity in each network it belongs to.
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use rusqlite::{Connection, OptionalExtension, params};
use tsunagi::dataplane::PluginError;
use tsunagi::identity::NetworkId;
use crate::keys::{KEY_LEN, WgSecretKey};
/// Schema version written by this build.
pub const SCHEMA_VERSION: i64 = 1;
/// Per-network WireGuard private keys.
#[derive(Debug)]
pub struct WgKeyStore {
conn: Mutex<Connection>,
path: PathBuf,
}
impl WgKeyStore {
/// Opens, creating the file and its directory if needed.
pub fn open(path: impl AsRef<Path>) -> Result<Self, PluginError> {
let path = path.as_ref().to_path_buf();
if let Some(parent) = path.parent() {
tsunagi::storage::create_dir(parent)
.map_err(|err| PluginError::Other(format!("cannot create {parent:?}: {err}")))?;
}
let existed = path.exists();
let conn = Connection::open(&path).map_err(|err| {
PluginError::Other(format!("cannot open the WireGuard key store: {err}"))
})?;
tsunagi::storage::restrict_path_permissions(&path)
.map_err(|err| PluginError::Other(format!("cannot secure the key store: {err}")))?;
conn.busy_timeout(std::time::Duration::from_secs(5))
.and_then(|()| conn.pragma_update(None, "journal_mode", "WAL"))
.and_then(|()| conn.pragma_update(None, "synchronous", "NORMAL"))
.map_err(|err| PluginError::Other(format!("cannot configure the key store: {err}")))?;
if existed {
let integrity: String = conn
.query_row("PRAGMA integrity_check", [], |row| row.get(0))
.map_err(|err| {
PluginError::Other(format!("WireGuard key store is unusable: {err}"))
})?;
if integrity != "ok" {
return Err(PluginError::Other(format!(
"WireGuard key store at {} is corrupt and will not be recreated: {integrity}",
path.display()
)));
}
}
let found: i64 = conn
.query_row("PRAGMA user_version", [], |row| row.get(0))
.map_err(|err| PluginError::Other(format!("cannot read the schema version: {err}")))?;
if found > SCHEMA_VERSION {
return Err(PluginError::Other(format!(
"WireGuard key store schema {found} is newer than {SCHEMA_VERSION}"
)));
}
if found < SCHEMA_VERSION {
conn.execute_batch(
"BEGIN;
CREATE TABLE IF NOT EXISTS network_keys (
network_id BLOB PRIMARY KEY,
secret BLOB NOT NULL,
created_at INTEGER NOT NULL
);
PRAGMA user_version = 1;
COMMIT;",
)
.map_err(|err| PluginError::Other(format!("cannot create the schema: {err}")))?;
}
Ok(Self {
conn: Mutex::new(conn),
path,
})
}
/// Path of the underlying file.
pub fn path(&self) -> &Path {
&self.path
}
/// Loads the stored key for `network`, or generates, stores and returns a fresh one.
pub fn load_or_create(&self, network: NetworkId) -> Result<WgSecretKey, PluginError> {
let guard = self
.conn
.lock()
.map_err(|_| PluginError::Other("key store lock is poisoned".into()))?;
let existing: Option<Vec<u8>> = guard
.query_row(
"SELECT secret FROM network_keys WHERE network_id = ?1",
params![network.as_bytes().as_slice()],
|row| row.get(0),
)
.optional()
.map_err(|err| PluginError::Other(format!("cannot read the key store: {err}")))?;
if let Some(secret) = existing {
let bytes = <[u8; KEY_LEN]>::try_from(secret.as_slice()).map_err(|_| {
PluginError::Other(format!(
"stored WireGuard key for {} is {} bytes, expected {KEY_LEN}",
network.fmt_short(),
secret.len()
))
})?;
return Ok(WgSecretKey::from_bytes(&bytes));
}
let fresh = WgSecretKey::generate();
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_err(|err| PluginError::Other(format!("system clock is broken: {err}")))?
.as_secs() as i64;
guard
.execute(
"INSERT INTO network_keys (network_id, secret, created_at) VALUES (?1, ?2, ?3)",
params![
network.as_bytes().as_slice(),
fresh.as_bytes().as_slice(),
now
],
)
.map_err(|err| PluginError::Other(format!("cannot write the key store: {err}")))?;
Ok(fresh)
}
/// Removes the stored key for `network`.
pub fn forget(&self, network: NetworkId) -> Result<(), PluginError> {
let guard = self
.conn
.lock()
.map_err(|_| PluginError::Other("key store lock is poisoned".into()))?;
guard
.execute(
"DELETE FROM network_keys WHERE network_id = ?1",
params![network.as_bytes().as_slice()],
)
.map_err(|err| {
PluginError::Other(format!("cannot delete from the key store: {err}"))
})?;
Ok(())
}
}
+472
View File
@@ -0,0 +1,472 @@
//! Pure WireGuard UDP packet transport and link implementation.
//!
//! Provides [`WireguardTransport`] implementing [`PacketTransport`]
//! and [`WireguardLink`] implementing [`PacketLink`].
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, RwLock};
use bytes::Bytes;
use iroh::EndpointId;
use tokio::net::UdpSocket;
use tokio::sync::{Notify, mpsc};
use tsunagi::BoxFuture;
use tsunagi::dataplane::transport::{
InboundLink, PacketLink, PacketTransport, SharedLink, TransportError,
};
use tsunagi::identity::{DeviceIdentity, NetworkId};
/// Default UDP port to bind and listen on for pure WireGuard.
pub const DEFAULT_PORT: u16 = 51820;
/// Maximum payload length for a single framed datagram over UDP.
pub const MAX_UDP_DATAGRAM_SIZE: usize = 65507;
/// An authenticated datagram link carried over direct UDP.
#[derive(Debug)]
pub struct WireguardLink {
network: NetworkId,
peer: EndpointId,
target_addr: Arc<RwLock<SocketAddr>>,
tx: mpsc::Sender<Bytes>,
inbound_tx: mpsc::Sender<Bytes>,
rx: tokio::sync::Mutex<mpsc::Receiver<Bytes>>,
closed_notify: Arc<Notify>,
is_closed: Arc<AtomicBool>,
}
impl WireguardLink {
/// Creates a new `WireguardLink` targeting `target_addr` over `socket`.
pub fn new(
network: NetworkId,
peer: EndpointId,
target_addr: SocketAddr,
socket: Arc<UdpSocket>,
) -> Self {
let (tx_out, mut rx_out) = mpsc::channel::<Bytes>(1024);
let (tx_in, rx_in) = mpsc::channel::<Bytes>(1024);
let is_closed = Arc::new(AtomicBool::new(false));
let closed_notify = Arc::new(Notify::new());
let target_holder = Arc::new(RwLock::new(target_addr));
let socket_w = Arc::clone(&socket);
let target_r = Arc::clone(&target_holder);
let is_closed_w = Arc::clone(&is_closed);
let closed_notify_w = Arc::clone(&closed_notify);
tokio::spawn(async move {
while let Some(msg) = rx_out.recv().await {
if msg.len() > MAX_UDP_DATAGRAM_SIZE {
continue;
}
let target = match target_r.read() {
Ok(g) => *g,
Err(p) => *p.into_inner(),
};
if let Err(err) = socket_w.send_to(&msg, target).await {
tracing::trace!(%err, %target, "failed to send WireGuard UDP datagram");
}
}
is_closed_w.store(true, Ordering::Release);
closed_notify_w.notify_waiters();
});
Self {
network,
peer,
target_addr: target_holder,
tx: tx_out,
inbound_tx: tx_in,
rx: tokio::sync::Mutex::new(rx_in),
closed_notify,
is_closed,
}
}
/// Delivers an inbound packet received from UDP into this link's receive queue.
pub fn deliver_inbound(&self, payload: Bytes) -> bool {
if self.is_closed.load(Ordering::Acquire) {
return false;
}
// If queue is full, dropping is normal for an unreliable datagram link
self.inbound_tx.try_send(payload).is_ok()
}
/// Updates the target socket address (roaming / NAT hole punch update).
pub fn update_target(&self, new_target: SocketAddr) {
if let Ok(mut guard) = self.target_addr.write()
&& *guard != new_target
{
tracing::debug!(peer = %self.peer.fmt_short(), old = %*guard, new = %new_target, "WireGuard endpoint roamed");
*guard = new_target;
}
}
/// Returns the current target address.
pub fn target_addr(&self) -> SocketAddr {
match self.target_addr.read() {
Ok(g) => *g,
Err(p) => *p.into_inner(),
}
}
}
impl PacketLink for WireguardLink {
fn network(&self) -> NetworkId {
self.network
}
fn peer(&self) -> EndpointId {
self.peer
}
fn max_datagram_size(&self) -> usize {
MAX_UDP_DATAGRAM_SIZE
}
fn send(&self, payload: Bytes) -> Result<(), TransportError> {
if payload.len() > MAX_UDP_DATAGRAM_SIZE {
return Err(TransportError::TooLarge {
size: payload.len(),
limit: MAX_UDP_DATAGRAM_SIZE,
});
}
if self.is_closed.load(Ordering::Acquire) {
return Err(TransportError::Closed);
}
self.tx
.try_send(payload)
.map_err(|_| TransportError::Closed)
}
fn recv(&self) -> BoxFuture<'_, Option<Bytes>> {
Box::pin(async move {
let mut guard = self.rx.lock().await;
guard.recv().await
})
}
fn closed(&self) -> BoxFuture<'_, ()> {
Box::pin(async move {
if self.is_closed.load(Ordering::Acquire) {
return;
}
self.closed_notify.notified().await;
})
}
fn is_closed(&self) -> bool {
self.is_closed.load(Ordering::Acquire)
}
fn path_description(&self) -> String {
format!("wireguard via {}", self.target_addr())
}
}
type InboundCallback = Box<dyn Fn(InboundLink) + Send + Sync + 'static>;
/// Direct UDP WireGuard packet transport.
pub struct WireguardTransport {
local_id: EndpointId,
listen_port: u16,
socket: Arc<UdpSocket>,
peer_addresses: RwLock<HashMap<EndpointId, Vec<SocketAddr>>>,
peer_networks: RwLock<HashMap<EndpointId, NetworkId>>,
links: RwLock<HashMap<(NetworkId, EndpointId), Arc<WireguardLink>>>,
inbound_cb: Arc<Mutex<Option<InboundCallback>>>,
}
impl std::fmt::Debug for WireguardTransport {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("WireguardTransport")
.field("local_id", &self.local_id.fmt_short().to_string())
.field("listen_port", &self.listen_port)
.finish()
}
}
impl WireguardTransport {
/// Binds a UDP socket for pure WireGuard transport.
///
/// Attempts `requested_port` (default 51820). If that fails, falls back to an ephemeral port.
pub async fn bind(
identity: &DeviceIdentity,
requested_port: Option<u16>,
) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
let target_port = requested_port.unwrap_or(DEFAULT_PORT);
let bind_addr = SocketAddr::from(([0, 0, 0, 0], target_port));
let socket = match UdpSocket::bind(bind_addr).await {
Ok(s) => s,
Err(err) if target_port > 0 => {
tracing::warn!(%err, port = target_port, "failed to bind WireGuard UDP port, falling back to ephemeral port");
UdpSocket::bind(SocketAddr::from(([0, 0, 0, 0], 0))).await?
}
Err(err) => return Err(Box::new(err)),
};
let bound_port = socket.local_addr()?.port();
tracing::info!(port = bound_port, "WireGuard UDP transport bound");
Ok(Self {
local_id: identity.endpoint_id(),
listen_port: bound_port,
socket: Arc::new(socket),
peer_addresses: RwLock::new(HashMap::new()),
peer_networks: RwLock::new(HashMap::new()),
links: RwLock::new(HashMap::new()),
inbound_cb: Arc::new(Mutex::new(None)),
})
}
/// The local port this transport is listening on.
pub fn bound_port(&self) -> u16 {
self.listen_port
}
/// Associates a peer with a network ID.
pub fn set_peer_network(&self, peer: EndpointId, network: NetworkId) {
if let Ok(mut guard) = self.peer_networks.write() {
guard.insert(peer, network);
}
}
/// Records an observed UDP target address for a peer.
pub fn set_peer_addr(&self, peer: EndpointId, addr: SocketAddr) {
if let Ok(mut guard) = self.peer_addresses.write() {
let addrs = guard.entry(peer).or_default();
addrs.retain(|a| *a != addr);
addrs.insert(0, addr);
}
}
/// Records multiple observed UDP target addresses for a peer.
pub fn set_peer_addrs(&self, peer: EndpointId, addrs: impl IntoIterator<Item = SocketAddr>) {
if let Ok(mut guard) = self.peer_addresses.write() {
let entry = guard.entry(peer).or_default();
for addr in addrs {
if !entry.contains(&addr) {
entry.push(addr);
}
}
}
}
/// Returns candidate socket addresses known for a peer.
pub fn peer_addrs(&self, peer: EndpointId) -> Vec<SocketAddr> {
let guard = match self.peer_addresses.read() {
Ok(g) => g,
Err(p) => p.into_inner(),
};
guard.get(&peer).cloned().unwrap_or_default()
}
/// Notifies the transport that a peer's capability was announced.
///
/// When `we_accept` is true (i.e. `local_id > peer`), the transport can proactively
/// create and install an inbound link so both ends are ready and can punch NAT.
pub fn notify_peer_capability(
&self,
network: NetworkId,
peer: EndpointId,
addrs: &[SocketAddr],
we_accept: bool,
) {
if let Ok(mut guard) = self.peer_networks.write() {
guard.insert(peer, network);
}
self.set_peer_addrs(peer, addrs.iter().copied());
if we_accept && !addrs.is_empty() {
let mut guard = match self.links.write() {
Ok(g) => g,
Err(p) => p.into_inner(),
};
if let std::collections::hash_map::Entry::Vacant(entry) = guard.entry((network, peer)) {
let link = Arc::new(WireguardLink::new(
network,
peer,
addrs[0],
Arc::clone(&self.socket),
));
entry.insert(Arc::clone(&link));
drop(guard);
let cb_guard = match self.inbound_cb.lock() {
Ok(g) => g,
Err(p) => p.into_inner(),
};
if let Some(ref cb) = *cb_guard {
cb(InboundLink {
network,
peer,
protocol: crate::announcement::WG_PROTOCOL.to_string(),
link,
});
}
}
}
}
/// Runs the receive loop for incoming UDP datagrams.
pub async fn accept_loop<F>(&self, on_inbound: F)
where
F: Fn(InboundLink) + Send + Sync + 'static,
{
{
let mut guard = match self.inbound_cb.lock() {
Ok(g) => g,
Err(p) => p.into_inner(),
};
*guard = Some(Box::new(on_inbound));
}
let socket = Arc::clone(&self.socket);
let mut buf = [0u8; MAX_UDP_DATAGRAM_SIZE];
while let Ok((len, src_addr)) = socket.recv_from(&mut buf).await {
let datagram = Bytes::copy_from_slice(&buf[..len]);
// 1. Check if an active link exists matching this peer / target
let matched_link = {
let guard = match self.links.read() {
Ok(g) => g,
Err(p) => p.into_inner(),
};
guard
.values()
.find(|link| {
link.target_addr() == src_addr || link.target_addr().ip() == src_addr.ip()
})
.cloned()
};
if let Some(link) = matched_link {
link.update_target(src_addr);
link.deliver_inbound(datagram);
continue;
}
// 2. Check if a peer has candidate addresses matching src_addr
let matched_peer = {
let guard = match self.peer_addresses.read() {
Ok(g) => g,
Err(p) => p.into_inner(),
};
guard
.iter()
.find(|(_, addrs)| {
addrs
.iter()
.any(|a| *a == src_addr || a.ip() == src_addr.ip())
})
.map(|(peer, _)| *peer)
};
if let Some(peer) = matched_peer {
let network = {
let guard = match self.peer_networks.read() {
Ok(g) => g,
Err(p) => p.into_inner(),
};
guard.get(&peer).copied().or_else(|| {
if guard.len() == 1 {
guard.values().next().copied()
} else {
match self.links.read() {
Ok(g) => g.keys().next().map(|(net, _)| *net),
Err(p) => p.into_inner().keys().next().map(|(net, _)| *net),
}
}
})
};
if let Some(network) = network {
let link = Arc::new(WireguardLink::new(
network,
peer,
src_addr,
Arc::clone(&self.socket),
));
link.deliver_inbound(datagram);
{
let mut guard = match self.links.write() {
Ok(g) => g,
Err(p) => p.into_inner(),
};
guard.insert((network, peer), Arc::clone(&link));
}
let cb_guard = match self.inbound_cb.lock() {
Ok(g) => g,
Err(p) => p.into_inner(),
};
if let Some(ref cb) = *cb_guard {
cb(InboundLink {
network,
peer,
protocol: crate::announcement::WG_PROTOCOL.to_string(),
link,
});
}
}
}
}
}
}
impl PacketTransport for WireguardTransport {
fn name(&self) -> &str {
"wireguard"
}
fn as_any(&self) -> &dyn std::any::Any {
self
}
fn open<'a>(
&'a self,
network: NetworkId,
peer: EndpointId,
_protocol: &'a str,
) -> BoxFuture<'a, Result<SharedLink, TransportError>> {
Box::pin(async move {
let addrs = self.peer_addrs(peer);
let Some(target) = addrs.first().copied() else {
return Err(TransportError::Unreachable(format!(
"no candidate UDP address known for peer {}",
peer.fmt_short()
)));
};
let link = {
let mut guard = match self.links.write() {
Ok(g) => g,
Err(p) => p.into_inner(),
};
if let Some(existing) = guard.get(&(network, peer))
&& !existing.is_closed()
{
return Ok(Arc::clone(existing) as SharedLink);
}
let link = Arc::new(WireguardLink::new(
network,
peer,
target,
Arc::clone(&self.socket),
));
guard.insert((network, peer), Arc::clone(&link));
link
};
Ok(link as SharedLink)
})
}
}
@@ -0,0 +1,86 @@
#![allow(missing_docs)]
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
use std::net::IpAddr;
use tsunagi::identity::NetworkId;
use tsunagi_wireguard::{ANNOUNCEMENT_VERSION, WgAnnouncement, WgPublicKey, WgSecretKey};
#[test]
fn announcement_round_trip() {
let network = NetworkId::from_bytes([42u8; 32]);
let key = WgSecretKey::generate().public();
let local = WgSecretKey::generate().public();
let addrs: Vec<IpAddr> = vec![
"192.168.1.100".parse().unwrap(),
"10.0.0.5".parse().unwrap(),
];
let announcement = WgAnnouncement::new(network, &key, 51820, addrs.clone());
let encoded = announcement.encode().expect("encode");
let validated = WgAnnouncement::decode_and_validate(&encoded, network, &local).expect("decode");
assert_eq!(validated.public_key, key);
assert_eq!(validated.port, 51820);
assert_eq!(validated.addrs, addrs);
}
#[test]
fn announcement_rejects_zero_key() {
let network = NetworkId::from_bytes([42u8; 32]);
let local = WgSecretKey::generate().public();
let zero = WgPublicKey::from_bytes([0u8; 32]);
let announcement = WgAnnouncement::new(network, &zero, 51820, vec![]);
let encoded = announcement.encode().unwrap();
let err = WgAnnouncement::decode_and_validate(&encoded, network, &local).unwrap_err();
assert!(err.to_string().contains("all zeroes"));
}
#[test]
fn announcement_rejects_own_key() {
let network = NetworkId::from_bytes([42u8; 32]);
let local = WgSecretKey::generate().public();
let announcement = WgAnnouncement::new(network, &local, 51820, vec![]);
let encoded = announcement.encode().unwrap();
let err = WgAnnouncement::decode_and_validate(&encoded, network, &local).unwrap_err();
assert!(err.to_string().contains("own WireGuard key"));
}
#[test]
fn announcement_rejects_wrong_network() {
let network_a = NetworkId::from_bytes([1u8; 32]);
let network_b = NetworkId::from_bytes([2u8; 32]);
let key = WgSecretKey::generate().public();
let local = WgSecretKey::generate().public();
let announcement = WgAnnouncement::new(network_a, &key, 51820, vec![]);
let encoded = announcement.encode().unwrap();
let err = WgAnnouncement::decode_and_validate(&encoded, network_b, &local).unwrap_err();
assert!(err.to_string().contains("different network"));
}
#[test]
fn announcement_rejects_zero_port() {
let network = NetworkId::from_bytes([42u8; 32]);
let key = WgSecretKey::generate().public();
let local = WgSecretKey::generate().public();
let announcement = WgAnnouncement {
version: ANNOUNCEMENT_VERSION,
public_key: *key.as_bytes(),
network: *network.as_bytes(),
port: 0,
addrs: vec![],
};
let encoded = announcement.encode().unwrap();
let err = WgAnnouncement::decode_and_validate(&encoded, network, &local).unwrap_err();
assert!(err.to_string().contains("cannot be 0"));
}
@@ -0,0 +1,206 @@
#![allow(missing_docs)]
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use bytes::Bytes;
use tokio::sync::mpsc;
use tsunagi::dataplane::transport::{InboundLink, PacketTransport};
use tsunagi::dataplane::{IpPlugin, PacketSink, PluginContext};
use tsunagi::identity::{DeviceIdentity, NetworkId};
use tsunagi_wireguard::{WireguardConfig, WireguardPlugin, WireguardTransport};
struct TestSink {
delivered: mpsc::Sender<(NetworkId, iroh::EndpointId, Bytes)>,
}
impl PacketSink for TestSink {
fn deliver(
&self,
network: NetworkId,
from: iroh::EndpointId,
packet: Bytes,
) -> tsunagi::BoxFuture<'_, ()> {
let delivered = self.delivered.clone();
Box::pin(async move {
let _ = delivered.send((network, from, packet)).await;
})
}
}
fn sample_ipv4_packet() -> Bytes {
let mut packet = vec![0u8; 40];
packet[0] = 0x45; // IPv4, IHL 5
packet[2..4].copy_from_slice(&40u16.to_be_bytes()); // total len 40
packet[8] = 64; // TTL
packet[9] = 17; // UDP
packet[12..16].copy_from_slice(&[10, 0, 0, 1]); // src 10.0.0.1
packet[16..20].copy_from_slice(&[10, 0, 0, 2]); // dst 10.0.0.2
Bytes::from(packet)
}
#[tokio::test]
async fn test_wireguard_plugin_end_to_end() {
let _ = tracing_subscriber::fmt()
.with_env_filter("trace")
.with_test_writer()
.try_init();
let identity_a = DeviceIdentity::generate();
let peer_a = identity_a.endpoint_id();
let identity_b = DeviceIdentity::generate();
let peer_b = identity_b.endpoint_id();
let network = NetworkId::from_bytes([77u8; 32]);
let tmp_a = tempfile::tempdir().expect("tmp A");
let tmp_b = tempfile::tempdir().expect("tmp B");
// Transports
let transport_b = Arc::new(
WireguardTransport::bind(&identity_b, Some(0))
.await
.expect("bind B"),
);
let port_b = transport_b.bound_port();
let transport_a = Arc::new(
WireguardTransport::bind(&identity_a, Some(0))
.await
.expect("bind A"),
);
let port_a = transport_a.bound_port();
// Plugins
let cfg_a = WireguardConfig::new(tmp_a.path());
let plugin_a = WireguardPlugin::open(cfg_a, Arc::clone(&transport_a))
.await
.expect("open plugin A");
let cfg_b = WireguardConfig::new(tmp_b.path());
let plugin_b = WireguardPlugin::open(cfg_b, Arc::clone(&transport_b))
.await
.expect("open plugin B");
// Attach sinks
let (sink_a_tx, mut sink_a_rx) = mpsc::channel(10);
plugin_a.attach(PluginContext::with_sink(Arc::new(TestSink {
delivered: sink_a_tx,
})));
let (sink_b_tx, mut sink_b_rx) = mpsc::channel(10);
plugin_b.attach(PluginContext::with_sink(Arc::new(TestSink {
delivered: sink_b_tx,
})));
// Capability exchange
plugin_a.on_network_activated(network);
plugin_b.on_network_activated(network);
let cap_a = loop {
if let Some(cap) = plugin_a.local_capability(network).expect("cap A") {
break cap;
}
tokio::time::sleep(Duration::from_millis(50)).await;
};
let cap_b = loop {
if let Some(cap) = plugin_b.local_capability(network).expect("cap B") {
break cap;
}
tokio::time::sleep(Duration::from_millis(50)).await;
};
plugin_a
.on_peer_capability(network, peer_b, &cap_b)
.expect("A on cap B");
plugin_b
.on_peer_capability(network, peer_a, &cap_a)
.expect("B on cap A");
// Register loopback address candidates
let addr_b: SocketAddr = format!("127.0.0.1:{port_b}").parse().unwrap();
transport_a.set_peer_addr(peer_b, addr_b);
let addr_a: SocketAddr = format!("127.0.0.1:{port_a}").parse().unwrap();
transport_b.set_peer_addr(peer_a, addr_a);
// Spawn accept loop on A
let transport_a_clone = Arc::clone(&transport_a);
tokio::spawn(async move {
transport_a_clone.accept_loop(|_| {}).await;
});
// Spawn accept loop on B
let (inbound_tx, mut inbound_rx) = mpsc::channel::<InboundLink>(1);
let transport_b_clone = Arc::clone(&transport_b);
tokio::spawn(async move {
transport_b_clone
.accept_loop(move |inbound| {
let _ = inbound_tx.try_send(inbound);
})
.await;
});
// Open link from A to B
let link_a = tokio::time::timeout(
Duration::from_secs(5),
transport_a.open(network, peer_b, "wg"),
)
.await
.expect("open should not timeout")
.expect("open should succeed");
plugin_a.on_peer_link(network, peer_b, Arc::clone(&link_a));
// Wait for inbound link on B and hand to plugin B
let inbound_b = tokio::time::timeout(Duration::from_secs(5), inbound_rx.recv())
.await
.expect("accept should not timeout")
.expect("inbound link should arrive");
plugin_b.on_peer_link(network, peer_a, inbound_b.link);
// Wait briefly for WireGuard handshake to complete
let mut up = false;
for _ in 0..50 {
if let Some(overview_a) = plugin_a.overview(network)
&& overview_a.peers.iter().any(|p| p.is_up())
{
up = true;
break;
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
assert!(up, "WireGuard handshake should complete between peers");
// Send IP packet from A to B
let packet_a_to_b = sample_ipv4_packet();
let sent = plugin_a.carry(network, peer_b, packet_a_to_b.clone());
assert!(sent, "plugin A carry should succeed");
let received_b = tokio::time::timeout(Duration::from_secs(5), sink_b_rx.recv())
.await
.expect("recv B timeout")
.expect("recv B option");
assert_eq!(received_b.0, network);
assert_eq!(received_b.1, peer_a);
assert_eq!(received_b.2, packet_a_to_b);
// Send IP packet from B to A
let packet_b_to_a = sample_ipv4_packet();
let sent_reply = plugin_b.carry(network, peer_a, packet_b_to_a.clone());
assert!(sent_reply, "plugin B carry should succeed");
let received_a = tokio::time::timeout(Duration::from_secs(5), sink_a_rx.recv())
.await
.expect("recv A timeout")
.expect("recv A option");
assert_eq!(received_a.0, network);
assert_eq!(received_a.1, peer_b);
assert_eq!(received_a.2, packet_b_to_a);
}
@@ -0,0 +1,112 @@
#![allow(missing_docs)]
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use bytes::Bytes;
use tokio::sync::mpsc;
use tsunagi::dataplane::transport::{InboundLink, PacketTransport};
use tsunagi::identity::{DeviceIdentity, NetworkId};
use tsunagi_wireguard::WireguardTransport;
#[tokio::test]
async fn test_wireguard_transport_loopback() {
let identity_a = DeviceIdentity::generate();
let peer_a = identity_a.endpoint_id();
let identity_b = DeviceIdentity::generate();
let peer_b = identity_b.endpoint_id();
let network = NetworkId::from_bytes([42u8; 32]);
// Bind transport B on port 0 (ephemeral)
let transport_b = Arc::new(
WireguardTransport::bind(&identity_b, Some(0))
.await
.expect("bind server should succeed"),
);
let port_b = transport_b.bound_port();
assert!(port_b > 0);
// Bind transport A on port 0 (ephemeral)
let transport_a = Arc::new(
WireguardTransport::bind(&identity_a, Some(0))
.await
.expect("bind client should succeed"),
);
let port_a = transport_a.bound_port();
assert!(port_a > 0);
// Register B's address and network on transport A
let addr_b: SocketAddr = format!("127.0.0.1:{port_b}").parse().unwrap();
transport_a.set_peer_addr(peer_b, addr_b);
transport_a.set_peer_network(peer_b, network);
// Register A's address and network on transport B
let addr_a: SocketAddr = format!("127.0.0.1:{port_a}").parse().unwrap();
transport_b.set_peer_addr(peer_a, addr_a);
transport_b.set_peer_network(peer_a, network);
// Spawn accept loop on A
let transport_a_clone = Arc::clone(&transport_a);
tokio::spawn(async move {
transport_a_clone.accept_loop(|_| {}).await;
});
// Spawn accept loop on B
let (inbound_tx, mut inbound_rx) = mpsc::channel::<InboundLink>(1);
let transport_b_clone = Arc::clone(&transport_b);
tokio::spawn(async move {
transport_b_clone
.accept_loop(move |inbound| {
let _ = inbound_tx.try_send(inbound);
})
.await;
});
// Open link from A to B
let link_a = tokio::time::timeout(
Duration::from_secs(5),
transport_a.open(network, peer_b, "wg"),
)
.await
.expect("open should not timeout")
.expect("open should succeed");
// Send from A to B to trigger inbound link discovery on B
let data_a_to_b = Bytes::from_static(b"wireguard packet from A to B");
link_a.send(data_a_to_b.clone()).expect("send A->B");
// Receive inbound link on B
let inbound_b = tokio::time::timeout(Duration::from_secs(5), inbound_rx.recv())
.await
.expect("accept should not timeout")
.expect("inbound link should arrive");
assert_eq!(inbound_b.network, network);
assert_eq!(inbound_b.peer, peer_a);
assert_eq!(inbound_b.protocol, "wg");
let link_b = inbound_b.link;
// Check packet received on B
let received_on_b = tokio::time::timeout(Duration::from_secs(2), link_b.recv())
.await
.expect("recv B timeout")
.expect("recv B option");
assert_eq!(received_on_b, data_a_to_b);
// Send from B to A
let data_b_to_a = Bytes::from_static(b"wireguard packet from B to A");
link_b.send(data_b_to_a.clone()).expect("send B->A");
let received_on_a = tokio::time::timeout(Duration::from_secs(2), link_a.recv())
.await
.expect("recv A timeout")
.expect("recv A option");
assert_eq!(received_on_a, data_b_to_a);
// Path descriptions
assert!(link_a.path_description().contains("wireguard via"));
assert!(link_b.path_description().contains("wireguard via"));
}