From 7070a16104ddac469641b37e05e33f97336089e8 Mon Sep 17 00:00:00 2001 From: Aleksandr Bogomiakov Date: Thu, 1 Oct 2026 01:24:25 +0100 Subject: [PATCH] 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. --- Cargo.lock | 25 + crates/tsunagi-cli/Cargo.toml | 1 + crates/tsunagi-cli/src/main.rs | 300 ++++--- crates/tsunagi-wireguard/Cargo.toml | 35 + crates/tsunagi-wireguard/src/announcement.rs | 142 ++++ crates/tsunagi-wireguard/src/device.rs | 527 +++++++++++++ crates/tsunagi-wireguard/src/keys.rs | 160 ++++ crates/tsunagi-wireguard/src/lib.rs | 28 + crates/tsunagi-wireguard/src/plugin.rs | 742 ++++++++++++++++++ crates/tsunagi-wireguard/src/store.rs | 160 ++++ crates/tsunagi-wireguard/src/transport.rs | 472 +++++++++++ .../tests/announcement_tests.rs | 86 ++ .../tsunagi-wireguard/tests/plugin_tests.rs | 206 +++++ .../tests/transport_tests.rs | 112 +++ 14 files changed, 2906 insertions(+), 90 deletions(-) create mode 100644 crates/tsunagi-wireguard/Cargo.toml create mode 100644 crates/tsunagi-wireguard/src/announcement.rs create mode 100644 crates/tsunagi-wireguard/src/device.rs create mode 100644 crates/tsunagi-wireguard/src/keys.rs create mode 100644 crates/tsunagi-wireguard/src/lib.rs create mode 100644 crates/tsunagi-wireguard/src/plugin.rs create mode 100644 crates/tsunagi-wireguard/src/store.rs create mode 100644 crates/tsunagi-wireguard/src/transport.rs create mode 100644 crates/tsunagi-wireguard/tests/announcement_tests.rs create mode 100644 crates/tsunagi-wireguard/tests/plugin_tests.rs create mode 100644 crates/tsunagi-wireguard/tests/transport_tests.rs diff --git a/Cargo.lock b/Cargo.lock index 013fbea..6229c7e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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" diff --git a/crates/tsunagi-cli/Cargo.toml b/crates/tsunagi-cli/Cargo.toml index 9f1a60f..2e86f75 100644 --- a/crates/tsunagi-cli/Cargo.toml +++ b/crates/tsunagi-cli/Cargo.toml @@ -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 diff --git a/crates/tsunagi-cli/src/main.rs b/crates/tsunagi-cli/src/main.rs index 091a9b3..21fcee7 100644 --- a/crates/tsunagi-cli/src/main.rs +++ b/crates/tsunagi-cli/src/main.rs @@ -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, @@ -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> { /// published. struct AgentControl { agent: Agent, + pure_wg: Option>, plugin: Option>, tcp_tls: Option>, /// 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> { .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> { let mut wireguard = None; let mut tcp_tls_transport = None; let mut tcp_tls_plugin = None; + let mut pure_wg_plugin: Option> = None; + let mut pure_wg_transport: Option> = None; if !wanted.is_empty() { let tun_factory: Arc = if args.no_tun { Arc::new(MemoryTunFactory::new()) @@ -3574,6 +3607,35 @@ async fn up(args: UpArgs) -> Result<(), Box> { 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) + .with_custom_transport( + WG_PROTOCOL, + Arc::clone(&transport) + as Arc, + ); + 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> { .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> { // 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 = Arc::new(AgentControl { agent, + pure_wg, plugin, tcp_tls, dns, @@ -3741,6 +3820,7 @@ async fn up(args: UpArgs) -> Result<(), Box> { /// both. async fn build_report( agent: &Agent, + pure_wg: Option<&PureWgPlugin>, wireguard: Option<&WireguardPlugin>, tcp_tls: Option<&TcpTlsPlugin>, dns: Option, @@ -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 = 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), diff --git a/crates/tsunagi-wireguard/Cargo.toml b/crates/tsunagi-wireguard/Cargo.toml new file mode 100644 index 0000000..2681bf9 --- /dev/null +++ b/crates/tsunagi-wireguard/Cargo.toml @@ -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 diff --git a/crates/tsunagi-wireguard/src/announcement.rs b/crates/tsunagi-wireguard/src/announcement.rs new file mode 100644 index 0000000..ba30acb --- /dev/null +++ b/crates/tsunagi-wireguard/src/announcement.rs @@ -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, +} + +/// 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, +} + +impl WgAnnouncement { + /// Builds this agent's announcement. + pub fn new( + network: NetworkId, + public_key: &WgPublicKey, + port: u16, + addrs: Vec, + ) -> 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, 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 { + 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 { + 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, + }) + } +} diff --git a/crates/tsunagi-wireguard/src/device.rs b/crates/tsunagi-wireguard/src/device.rs new file mode 100644 index 0000000..7f5157f --- /dev/null +++ b/crates/tsunagi-wireguard/src/device.rs @@ -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, +} + +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, +} + +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>, + tunn: Mutex, + link: SharedLink, + counters: Arc, + task: Mutex>>, +} + +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, + /// 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>>, + sink: Arc, + 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, + tasks: Vec>, +} + +impl WireguardDevice { + /// Starts the WireGuard tunnels for one network. + pub fn start(network: NetworkId, private_key: WgSecretKey, sink: Arc) -> 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, + link: SharedLink, + keepalive: Option, + ) -> 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 = 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 { + let mut peers: Vec = 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(lock: &RwLock) -> std::sync::RwLockReadGuard<'_, T> { + match lock.read() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + } +} + +fn write_lock(lock: &RwLock) -> 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, peer: Arc) { + 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) { + let mut ticker = tokio::time::interval(TIMER_INTERVAL); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + ticker.tick().await; + let peers: Vec> = 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]); + } + } + } +} diff --git a/crates/tsunagi-wireguard/src/keys.rs b/crates/tsunagi-wireguard/src/keys.rs new file mode 100644 index 0000000..71a8d82 --- /dev/null +++ b/crates/tsunagi-wireguard/src/keys.rs @@ -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 { + 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() + } +} diff --git a/crates/tsunagi-wireguard/src/lib.rs b/crates/tsunagi-wireguard/src/lib.rs new file mode 100644 index 0000000..1b24f9e --- /dev/null +++ b/crates/tsunagi-wireguard/src/lib.rs @@ -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}; diff --git a/crates/tsunagi-wireguard/src/plugin.rs b/crates/tsunagi-wireguard/src/plugin.rs new file mode 100644 index 0000000..f9c68be --- /dev/null +++ b/crates/tsunagi-wireguard/src/plugin.rs @@ -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 { + 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, + /// The largest packet a tunnel will carry. + pub mtu: u32, + /// UDP port to listen on. + pub port: Option, + /// 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) -> 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, + /// The IPv4 overlay range in use. + pub ipv4_range: Option, + /// Peers this agent knows about. + pub peers: Vec, +} + +/// 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, + /// Whether a data plane link to it exists. + pub has_link: bool, + /// The running tunnel, once there is a link. + pub tunnel: Option, +} + +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>, + announcements: HashMap, + links: HashMap, + allocations: HashMap, + ipv4_range: Option, +} + +#[derive(Debug, Default)] +struct Shared { + networks: HashMap, +} + +#[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, + local_id: OnceLock, + store: WgKeyStore, + shared: Mutex, + context: OnceLock, + sink: OnceLock>, +} + +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, + commands: mpsc::Sender, + task: Mutex>>, +} + +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 { + 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, + ) -> Result, 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 { + let shared = self.worker.lock_shared(); + let state = shared.networks.get(&network)?; + + let tunnels: HashMap = state + .device + .as_ref() + .map(|device| { + device + .peers() + .into_iter() + .map(|summary| (summary.public_key, summary)) + .collect() + }) + .unwrap_or_default(); + + let mut peers: Vec = 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, network: NetworkId) -> Result { + 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, + }; + 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 = 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 { + self.lock_shared().networks.keys().copied().collect() + } +} + +async fn run(worker: Arc, mut commands: mpsc::Receiver) { + let mut pending: BTreeSet = BTreeSet::new(); + let mut deadline: Option = 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, 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 = 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 = + 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; + } + }) + } +} diff --git a/crates/tsunagi-wireguard/src/store.rs b/crates/tsunagi-wireguard/src/store.rs new file mode 100644 index 0000000..57ac898 --- /dev/null +++ b/crates/tsunagi-wireguard/src/store.rs @@ -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, + path: PathBuf, +} + +impl WgKeyStore { + /// Opens, creating the file and its directory if needed. + pub fn open(path: impl AsRef) -> Result { + 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 { + let guard = self + .conn + .lock() + .map_err(|_| PluginError::Other("key store lock is poisoned".into()))?; + + let existing: Option> = 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(()) + } +} diff --git a/crates/tsunagi-wireguard/src/transport.rs b/crates/tsunagi-wireguard/src/transport.rs new file mode 100644 index 0000000..ca9c66a --- /dev/null +++ b/crates/tsunagi-wireguard/src/transport.rs @@ -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>, + tx: mpsc::Sender, + inbound_tx: mpsc::Sender, + rx: tokio::sync::Mutex>, + closed_notify: Arc, + is_closed: Arc, +} + +impl WireguardLink { + /// Creates a new `WireguardLink` targeting `target_addr` over `socket`. + pub fn new( + network: NetworkId, + peer: EndpointId, + target_addr: SocketAddr, + socket: Arc, + ) -> Self { + let (tx_out, mut rx_out) = mpsc::channel::(1024); + let (tx_in, rx_in) = mpsc::channel::(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> { + 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; + +/// Direct UDP WireGuard packet transport. +pub struct WireguardTransport { + local_id: EndpointId, + listen_port: u16, + socket: Arc, + peer_addresses: RwLock>>, + peer_networks: RwLock>, + links: RwLock>>, + inbound_cb: Arc>>, +} + +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, + ) -> Result> { + 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) { + 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 { + 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(&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> { + 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) + }) + } +} diff --git a/crates/tsunagi-wireguard/tests/announcement_tests.rs b/crates/tsunagi-wireguard/tests/announcement_tests.rs new file mode 100644 index 0000000..945c044 --- /dev/null +++ b/crates/tsunagi-wireguard/tests/announcement_tests.rs @@ -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 = 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")); +} diff --git a/crates/tsunagi-wireguard/tests/plugin_tests.rs b/crates/tsunagi-wireguard/tests/plugin_tests.rs new file mode 100644 index 0000000..b9acdd6 --- /dev/null +++ b/crates/tsunagi-wireguard/tests/plugin_tests.rs @@ -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::(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); +} diff --git a/crates/tsunagi-wireguard/tests/transport_tests.rs b/crates/tsunagi-wireguard/tests/transport_tests.rs new file mode 100644 index 0000000..9cf1618 --- /dev/null +++ b/crates/tsunagi-wireguard/tests/transport_tests.rs @@ -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::(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")); +}