Menu
Public repository / 22da10d5eeb459769d6d21108a8ba1b363d5b313
AkurAI-VPN / crates/akurai-node/src/tunnel.rs
Latest change 67ca9028c2fad783b70c10cd9fac111ce5b10de1 - release: v0.3.4 by Claude Code
//! The data plane — an encrypted overlay pump with relay fallback and direct paths.
//!
//! `run` opens the `akurai0` TUN, assigns the node's overlay IP and a
//! `100.88.0.0/16`-ONLY route (never a default route — the host\'s internet path
//! is inviolable), binds a UDP socket, and runs two blocking pumps over a shared
//! per-peer state table:
//!
//! - **TUN → UDP**: read an IPv4 packet, route it to the carrying peer, encrypt
//! with that peer\'s Noise session, and send a `Data` frame on the peer\'s
//! current path — a **direct** UDP address once one is confirmed, otherwise the
//! relay (which forwards by destination overlay IP).
//! - **UDP → TUN**: receive a frame; complete handshakes, learn/confirm direct
//! paths from `PeerAddr` hints, or decrypt a `Data` frame and write the inner
//! packet to the TUN.
//!
//! Direct paths (MVP3): handshakes go via the relay, which sends each end a
//! `PeerAddr` hint with the other\'s observed UDP address. Both nodes probe that
//! address (opening any NAT hole); a frame received straight from a peer confirms
//! the direct path and traffic leaves the hub. If no direct path forms, the relay
//! carries everything. End-to-end: the relay only ever sees ciphertext.
//! Fail-closed: a packet to an unknown destination, or from an unverified peer, is dropped.
use std::collections::{HashMap, HashSet};
use std::io;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, UdpSocket};
use std::path::PathBuf;
// Interface/route setup shells out to the OS CLI on every platform: `ip`/`sysctl`
// on Linux, `ifconfig`/`route` on macOS, `netsh` on Windows.
use std::process::Command;
use std::sync::{Arc, Mutex, RwLock};
use std::thread;
use std::time::Duration;
use akurai_common::frame::{Frame, FrameKind};
use akurai_transport::noise::{self, Initiator};
use akurai_transport::session::Session;
use akurai_transport::Keypair;
use crate::acl::Acl;
use crate::peers::PeerTable;
/// Control-plane heartbeat settings. `tunnel::run` starts the heartbeat after
/// the UDP socket is bound, so the reported endpoint uses the real tunnel port.
#[derive(Clone)]
pub struct HeartbeatConfig {
pub control_url: String,
pub cookie_jar: PathBuf,
pub pubkey_b64: String,
pub node_token: Option<String>,
}
/// Runtime configuration for the data-plane daemon.
// Several fields drive interface/route setup, which is fully exercised only on the
// Linux gateway path; the macOS client path uses a subset and the Windows stub none.
// Suppress the resulting platform-specific dead-field warnings OFF Linux only, so
// the Linux build is unaffected.
#[cfg_attr(not(target_os = "linux"), allow(dead_code))]
pub struct TunnelConfig {
pub iface: String,
pub overlay_ip: Ipv4Addr,
pub overlay_cidr: String,
pub mtu: u16,
pub relay: SocketAddr,
pub keypair: Keypair,
pub peers: PeerTable,
/// Subnets THIS node is a gateway for (MVP2). When non-empty, the node
/// enables IP forwarding so it can relay overlay traffic to the real subnet.
pub advertise: Vec<akurai_common::Cidr>,
/// Use this peer as a full-tunnel EXIT node (route 0.0.0.0/0 over the overlay
/// to it). Explicit opt-in only — `None` means normal split-tunnel.
pub exit_node: Option<Ipv4Addr>,
/// Fine-grained ACL (MVP4). `Some` ⇒ enforce a fail-closed tag policy on
/// every outbound flow (`this-node → routing-peer`); `None` ⇒ no ACL file,
/// so any peer in the map is reachable (network-membership-only).
pub acl: Option<Acl>,
/// Control-plane URL for periodic peer-map refresh (`GET /api/peermap`).
/// When `Some`, a background thread fetches every 30 s and hot-swaps the
/// peer table. `None` ⇒ no refresh (static file or one-shot fetch only).
pub control_url: Option<String>,
/// Cookie jar path passed to `curl` for cookie-authenticated control-plane
/// requests. Used only when `node_token` is `None`.
pub cookie_jar: Option<PathBuf>,
/// Durable node token for bearer-token auth (`Authorization: Bearer <t>`).
/// When `Some`, the peer-map refresh thread uses bearer auth instead of the
/// session cookie — so the refresh survives cookie expiry. `None` ⇒ fall
/// back to cookie-jar auth (requires a valid session cookie).
pub node_token: Option<String>,
/// Optional control-plane heartbeat. Started after the UDP socket is bound
/// so the advertised endpoint can carry the current direct UDP candidate.
pub heartbeat: Option<HeartbeatConfig>,
}
/// Per-peer session + path state, shared between the pump threads.
#[derive(Default)]
struct PeerState {
/// Completed Noise session (None until the handshake finishes).
session: Option<Session>,
/// In-flight handshake we initiated.
pending: Option<Initiator>,
/// A confirmed direct UDP path — outbound frames go here instead of the relay.
direct: Option<SocketAddr>,
/// A direct address learned from a relay `PeerAddr` hint, being probed but
/// not yet confirmed.
candidate: Option<SocketAddr>,
}
type Table = HashMap<Ipv4Addr, PeerState>;
/// The handles both pump loops share: the TUN, the UDP socket, the per-peer
/// state table, the peer/routing table, this node\'s keypair, its overlay IP,
/// and the relay address. Bundled (and cheaply `Clone`, since the heavy members
/// are `Arc`s) so each pump takes a single context argument.
#[derive(Clone)]
struct Pump {
tun: Arc<akurai_sys::TunDevice>,
sock: Arc<UdpSocket>,
table: Arc<Mutex<Table>>,
peers: Arc<RwLock<Arc<PeerTable>>>,
kp: Arc<Keypair>,
my_ip: Ipv4Addr,
relay: SocketAddr,
}
/// 32 bytes of OS entropy for an ephemeral handshake secret.
fn rand32() -> io::Result<[u8; 32]> {
let mut b = [0u8; 32];
crate::rng::fill_random(&mut b)?;
Ok(b)
}
/// Decode a 6-byte `PeerAddr` payload (4 octets + BE port) into a socket address.
fn parse_addr(payload: &[u8]) -> Option<SocketAddr> {
if payload.len() != 6 {
return None;
}
let ip = Ipv4Addr::new(payload[0], payload[1], payload[2], payload[3]);
let port = u16::from_be_bytes([payload[4], payload[5]]);
Some(SocketAddr::from((ip, port)))
}
/// Pick outbound addresses for first-contact handshakes. A confirmed direct path
/// is used by itself. A candidate path is tried first with relay fallback. When
/// the control plane advertises a peer endpoint, seed it as the candidate.
fn handshake_paths(
st: &mut PeerState,
relay: SocketAddr,
advertised: Option<SocketAddr>,
) -> Vec<SocketAddr> {
if let Some(direct) = st.direct {
return vec![direct];
}
if st.candidate.is_none() {
st.candidate = advertised;
}
let mut paths = Vec::new();
if let Some(candidate) = st.candidate {
if candidate != relay {
paths.push(candidate);
}
}
paths.push(relay);
paths
}
/// Bring up the overlay interface. Platform-specific: Linux uses `ip`/`sysctl`,
/// macOS uses `ifconfig`/`route`, Windows uses `netsh`. `iface` is the REAL
/// interface name (the requested name on Linux and Windows, the kernel-assigned
/// `utunN` on macOS). NO default route is ever installed.
///
/// Bring up the overlay interface on Linux: address, MTU, the overlay-only route,
/// any peer-advertised subnet routes, and IP forwarding if this node is a gateway.
/// Uses `ip`/`sysctl` (safe `std::process::Command`); adds NO default route.
#[cfg(target_os = "linux")]
fn setup_interface(cfg: &TunnelConfig, iface: &str) -> io::Result<()> {
let ip = |args: &[&str]| -> io::Result<()> {
let status = Command::new("ip").args(args).status()?;
if !status.success() {
return Err(io::Error::other(format!("`ip {}` failed", args.join(" "))));
}
Ok(())
};
let addr = format!("{}/32", cfg.overlay_ip);
let mtu = cfg.mtu.to_string();
ip(&["addr", "add", &addr, "dev", iface])?;
ip(&["link", "set", iface, "mtu", &mtu])?;
ip(&["link", "set", iface, "up"])?;
// Overlay-scoped route ONLY. Never 0.0.0.0/0.
ip(&["route", "add", &cfg.overlay_cidr, "dev", iface])?;
// IPv6 overlay address + route (fd88::/48), derived from the node\'s index
// (100.88.0.N ↔ fd88::N). Best-effort — IPv6 may be disabled on the host.
let idx6 = u32::from(cfg.overlay_ip).wrapping_sub(u32::from(akurai_common::OVERLAY_IPV4_NET));
let ip6 = akurai_common::OverlayIpv6::from_index(idx6 as u64).addr();
let _ = Command::new("ip")
.args([
"-6",
"addr",
"add",
&format!("{ip6}/{}", akurai_common::OVERLAY_IPV6_PREFIX_LEN),
"dev",
iface,
])
.status();
let _ = Command::new("ip")
.args([
"-6",
"route",
"add",
&format!(
"{}/{}",
akurai_common::OVERLAY_IPV6_NET,
akurai_common::OVERLAY_IPV6_PREFIX_LEN
),
"dev",
iface,
])
.status();
// Subnet routes: point each peer-advertised subnet at the overlay. A
// 0.0.0.0/0 advertisement is SKIPPED here — a peer can never silently capture
// all traffic; full-tunnel happens only via the explicit `--exit-node` opt-in.
for (subnet, _gw) in cfg.peers.advertised_routes() {
if subnet.prefix_len() == 0 {
continue; // never auto-install a default route
}
let cidr = subnet.to_string();
let _ = Command::new("ip")
.args(["route", "add", &cidr, "dev", iface])
.status();
}
// Exit node (full-tunnel), EXPLICIT opt-in only: route everything via the
// overlay using two /1 routes that out-specific the real default WITHOUT
// deleting it — so when the tunnel/TUN closes, normal routing is restored
// automatically. The chosen exit peer must advertise 0.0.0.0/0.
if let Some(exit) = cfg.exit_node {
eprintln!("{iface}: full-tunnel — routing 0.0.0.0/0 via exit node {exit}");
let _ = Command::new("ip")
.args(["route", "add", "0.0.0.0/1", "dev", iface])
.status();
let _ = Command::new("ip")
.args(["route", "add", "128.0.0.0/1", "dev", iface])
.status();
}
// If THIS node is a gateway, enable IP forwarding. An exit gateway (advertising
// 0.0.0.0/0) also masquerades overlay traffic out its real interface.
if !cfg.advertise.is_empty() {
let _ = Command::new("sysctl")
.args(["-qw", "net.ipv4.ip_forward=1"])
.status();
if cfg.advertise.iter().any(|c| c.prefix_len() == 0) {
let _ = Command::new("sh")
.args([
"-c",
"iptables -t nat -C POSTROUTING -s 100.88.0.0/16 ! -o akurai0 -j MASQUERADE 2>/dev/null \
|| iptables -t nat -A POSTROUTING -s 100.88.0.0/16 ! -o akurai0 -j MASQUERADE",
])
.status();
}
}
Ok(())
}
/// Bring up the overlay interface on macOS: assign the overlay address as a
/// point-to-point host address, set the MTU, bring it up, and add the overlay-only
/// route over `utunN`. The Linux-only gateway bits (`sysctl`/`iptables`, IPv6,
/// per-peer subnet routes) are intentionally omitted — this is the client path.
/// NO default route is installed.
#[cfg(target_os = "macos")]
fn setup_interface(cfg: &TunnelConfig, iface: &str) -> io::Result<()> {
let ip = cfg.overlay_ip.to_string();
let status = Command::new("ifconfig")
.args([iface, &ip, &ip, "up"])
.status()?;
if !status.success() {
return Err(io::Error::other(format!(
"`ifconfig {iface} {ip} {ip} up` failed"
)));
}
// Best-effort MTU (the macOS analogue of `ip link set <iface> mtu`).
let _ = Command::new("ifconfig")
.args([iface, "mtu", &cfg.mtu.to_string()])
.status();
// Overlay-scoped route ONLY. Never a default route.
let status = Command::new("route")
.args(["-n", "add", "-net", &cfg.overlay_cidr, "-interface", iface])
.status()?;
if !status.success() {
return Err(io::Error::other(format!(
"`route -n add -net {} -interface {iface}` failed",
cfg.overlay_cidr
)));
}
Ok(())
}
/// Bring up the overlay interface on Windows via `netsh`: assign the overlay IPv4
/// address with the `/16` overlay mask and add the overlay-only route over the
/// Wintun adapter. `iface` is the Wintun adapter name — the requested name (Wintun
/// honors it, unlike macOS's kernel-assigned `utunN`). The Linux-only gateway bits
/// are intentionally omitted — this is the client path. NO default route is
/// installed.
#[cfg(target_os = "windows")]
fn setup_interface(cfg: &TunnelConfig, iface: &str) -> io::Result<()> {
let ip = cfg.overlay_ip.to_string();
// Static overlay address with the 255.255.0.0 (/16) overlay mask.
let status = Command::new("netsh")
.args([
"interface",
"ip",
"set",
"address",
&format!("name={iface}"),
"static",
&ip,
"255.255.0.0",
])
.status()?;
if !status.success() {
return Err(io::Error::other(format!(
"`netsh interface ip set address name={iface} static {ip} 255.255.0.0` failed"
)));
}
// Overlay-scoped route ONLY over the Wintun adapter. Never a default route.
let status = Command::new("netsh")
.args([
"interface",
"ipv4",
"add",
"route",
&cfg.overlay_cidr,
iface,
])
.status()?;
if !status.success() {
return Err(io::Error::other(format!(
"`netsh interface ipv4 add route {} {iface}` failed",
cfg.overlay_cidr
)));
}
Ok(())
}
/// Run the data plane. Opens the TUN, brings up the interface, binds the socket,
/// and blocks running the pumps until the process is killed.
pub fn run(cfg: TunnelConfig) -> io::Result<()> {
let tun = Arc::new(akurai_sys::create_tun(&cfg.iface)?);
// The REAL interface name: the requested name on Linux, the kernel-assigned
// `utunN` on macOS. Interface/route setup must target this, not `cfg.iface`.
let iface = tun.name().to_string();
setup_interface(&cfg, &iface)?;
let sock = Arc::new(UdpSocket::bind(("0.0.0.0", 0))?);
let relay = cfg.relay;
let heartbeat = cfg.heartbeat.clone();
let table: Arc<Mutex<Table>> = Arc::new(Mutex::new(HashMap::new()));
let control_url = cfg.control_url;
let cookie_jar = cfg.cookie_jar;
let node_token = cfg.node_token;
let peers: Arc<RwLock<Arc<PeerTable>>> = Arc::new(RwLock::new(Arc::new(cfg.peers)));
let kp = Arc::new(cfg.keypair);
let my_ip = cfg.overlay_ip;
// ACL is consulted only on the outbound (TUN→UDP) path, which runs on this
// thread, so it need not be shared with the UDP pump.
let acl = cfg.acl;
if let Some(hb) = heartbeat {
let port = sock.local_addr()?.port();
crate::heartbeat::spawn(
hb.control_url,
hb.cookie_jar,
hb.pubkey_b64,
hb.node_token,
relay,
port,
);
}
// MagicDNS: resolve `<peer-name>.akurai` (and bare `<peer-name>`) to a peer\'s
// overlay IP, served on the node\'s own overlay IP:53.
{
let dns_peers = Arc::clone(&peers);
let bind = SocketAddr::from((my_ip, 53));
thread::spawn(move || {
let _ = akurai_dns::serve(bind, "akurai", move |label| {
// Take a short read lock, clone the inner Arc, then operate
// lock-free so DNS resolution never blocks the refresh writer.
let pt = { Arc::clone(&*dns_peers.read().unwrap()) };
pt.resolve_name(label)
});
});
}
// Keepalive to the relay (stay known) + probe direct candidates/paths to keep
// any NAT hole open.
{
let sock = Arc::clone(&sock);
let table = Arc::clone(&table);
let peers = Arc::clone(&peers);
thread::spawn(move || loop {
if let Some(f) = Frame::new(
FrameKind::Keepalive,
my_ip,
Ipv4Addr::UNSPECIFIED,
Vec::new(),
) {
let bytes = f.encode();
let _ = sock.send_to(&bytes, relay);
let addrs = {
let advertised = {
let pt = Arc::clone(&*peers.read().unwrap());
pt.direct_endpoints()
};
let mut addrs = Vec::new();
if let Ok(mut t) = table.lock() {
for (peer_ip, endpoint) in advertised {
let st = t.entry(peer_ip).or_default();
if st.direct.is_none() && st.candidate != Some(endpoint) {
st.candidate = Some(endpoint);
}
}
for st in t.values() {
if let Some(a) = st.direct.or(st.candidate) {
if a != relay {
addrs.push(a);
}
}
}
}
addrs
};
for addr in addrs {
let _ = sock.send_to(&bytes, addr);
}
}
thread::sleep(Duration::from_secs(15));
});
}
// Peer-map refresh: every 30 s, re-fetch the control-plane peer map and
// hot-swap the shared table if the result is non-empty. An empty or failed
// fetch leaves the existing table untouched — we never wipe peers on error.
// When a node token is available, use bearer-token auth so the refresh
// survives cookie expiry; otherwise fall back to the cookie jar.
if let Some(url) = control_url {
let peers_rw = Arc::clone(&peers);
thread::spawn(move || loop {
let fetched = if let Some(ref tok) = node_token {
PeerTable::fetch_with_token(&url, tok)
} else if let Some(ref jar) = cookie_jar {
PeerTable::fetch(&url, jar)
} else {
PeerTable::default()
};
let n = fetched.len();
if n > 0 {
*peers_rw.write().unwrap() = Arc::new(fetched);
eprintln!("akurai-node: peers refreshed — {n} peer(s)");
}
thread::sleep(Duration::from_secs(30));
});
}
let pump = Pump {
tun,
sock,
table,
peers,
kp,
my_ip,
relay,
};
let udp = {
let pump = pump.clone();
thread::spawn(move || udp_pump(pump))
};
tun_pump(pump, acl)?;
let _ = udp.join();
Ok(())
}
/// Diagnostic: bring up the TUN and echo ICMP echo-requests straight back as
/// replies for `secs` seconds. Pinging any overlay IP (routed to the TUN) gets a
/// reply **iff** `TunDevice::recv` and `send` move packets correctly on this
/// platform — so this runtime-verifies the platform packet path (the macOS
/// 4-byte AF header / the Windows Wintun ring) on real hardware, without needing
/// a second machine.
pub fn selftest(cfg: TunnelConfig, secs: u64) -> io::Result<()> {
let tun = akurai_sys::create_tun(&cfg.iface)?;
let iface = tun.name().to_string();
setup_interface(&cfg, &iface)?;
eprintln!(
"akurai-node: selftest TUN up — overlay {} on {iface}, echoing ICMP for {secs}s",
cfg.overlay_ip
);
let deadline = std::time::Instant::now() + Duration::from_secs(secs);
let mut buf = [0u8; 2048];
let mut echoed = 0u32;
while std::time::Instant::now() < deadline {
let n = match tun.recv(&mut buf) {
Ok(0) => continue,
Ok(n) => n,
Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue,
Err(_) => continue,
};
if let Some(reply) = icmp_echo_reply(&buf[..n]) {
if tun.send(&reply).is_ok() {
echoed += 1;
eprintln!("akurai-node: selftest echoed ICMP reply #{echoed}");
}
}
}
Ok(())
}
/// Transport-level echo peer: binds a UDP socket, connects to the relay, and
/// for `secs` seconds:
/// - responds to `HandshakeInit` from any peer in `cfg.peers`, storing a
/// `Session` keyed by the peer's overlay IP;
/// - decrypts `Data` frames, calls `icmp_echo_reply` on the inner packet, and
/// returns the ICMP reply encrypted back to the sender via the relay;
/// - sends `Keepalive` to the relay every 5 s so the relay learns our UDP endpoint.
///
/// No TUN device is opened — this is a pure-userspace overlay peer, so it
/// compiles and runs on Linux, macOS, and Windows without any kernel privileges.
pub fn echo_peer(cfg: TunnelConfig, secs: u64) -> io::Result<()> {
eprintln!(
"akurai-node: echo-peer up — overlay {}, peer-echoing for {secs}s",
cfg.overlay_ip
);
let sock = UdpSocket::bind("0.0.0.0:0")?;
// Short read timeout so the receive loop wakes up periodically to check the
// deadline, even when no frames arrive.
sock.set_read_timeout(Some(Duration::from_millis(500)))?;
let relay = cfg.relay;
let my_ip = cfg.overlay_ip;
let kp = cfg.keypair;
let peers = cfg.peers;
// Keepalive thread: 5 s cadence (faster than the tunnel's 15 s) so the relay
// records our UDP address quickly, before node A sends us a HandshakeInit.
{
let sock_ka = sock.try_clone()?;
thread::spawn(move || loop {
if let Some(f) = Frame::new(
FrameKind::Keepalive,
my_ip,
Ipv4Addr::UNSPECIFIED,
Vec::new(),
) {
let _ = sock_ka.send_to(&f.encode(), relay);
}
thread::sleep(Duration::from_secs(5));
});
}
// Per-peer state on the main thread only — no sharing required.
let mut sessions: HashMap<Ipv4Addr, Session> = HashMap::new();
// In-flight initiators for any handshake *we* started (echo-peer is normally
// a pure responder, but we store any pending state so HandshakeResp is handled
// gracefully rather than silently dropped).
let mut pending: HashMap<Ipv4Addr, Initiator> = HashMap::new();
let deadline = std::time::Instant::now() + Duration::from_secs(secs);
let mut buf = [0u8; 4096];
while std::time::Instant::now() < deadline {
let (n, from) = match sock.recv_from(&mut buf) {
Ok(v) => v,
// Timeout (WouldBlock on Unix, TimedOut on Windows) — loop to
// recheck the deadline.
Err(ref e)
if e.kind() == io::ErrorKind::WouldBlock || e.kind() == io::ErrorKind::TimedOut =>
{
continue
}
Err(ref e) if e.kind() == io::ErrorKind::Interrupted => continue,
Err(e) => return Err(e),
};
let Some(frame) = Frame::decode(&buf[..n]) else {
continue;
};
match frame.kind {
FrameKind::HandshakeInit => {
// Mirror udp_pump: respond only when the initiator's recovered
// static key matches the configured peer.
let Ok(eph) = rand32() else { continue };
if let Some((keys, msg2, init_pub)) = noise::respond(&kp, &eph, &frame.payload) {
match peers.get(&frame.src) {
Some(p) if p.public_key == init_pub => {
sessions.insert(frame.src, Session::new(keys));
if let Some(reply) =
Frame::new(FrameKind::HandshakeResp, my_ip, frame.src, msg2)
{
// Send on the arrival path (same as udp_pump).
let _ = sock.send_to(&reply.encode(), from);
}
}
_ => {} // unknown peer or mismatched key: drop
}
}
}
FrameKind::HandshakeResp => {
// Complete any initiator we started (graceful, even though we
// normally do not initiate as echo-peer).
if let Some(state) = pending.remove(&frame.src) {
if let Some(keys) = noise::finalize(state, &frame.payload) {
sessions.insert(frame.src, Session::new(keys));
}
}
}
FrameKind::Data => {
// Decrypt the inner IP packet; if it is an ICMP echo-request,
// encrypt the reply and send it back via the relay.
if let Some(sess) = sessions.get_mut(&frame.src) {
if let Some(inner) = sess.decrypt(&frame.payload) {
if let Some(reply_pkt) = icmp_echo_reply(&inner) {
let ct = sess.encrypt(&reply_pkt);
if let Some(reply_frame) =
Frame::new(FrameKind::Data, my_ip, frame.src, ct)
{
let _ = sock.send_to(&reply_frame.encode(), relay);
eprintln!("echo-peer answered ping from {}", frame.src);
}
}
}
}
}
// PeerAddr hints and keepalives require no action from a pure responder.
FrameKind::PeerAddr | FrameKind::Keepalive => {}
}
}
Ok(())
}
/// If `pkt` is an IPv4 ICMP echo *request*, build the matching echo *reply*
/// (swap src/dst, type 8→0, recompute the ICMP checksum). The IPv4 header
/// checksum is unchanged because swapping src↔dst leaves the 16-bit word sum
/// invariant. Returns `None` for anything that is not an ICMP echo request.
pub(crate) fn icmp_echo_reply(pkt: &[u8]) -> Option<Vec<u8>> {
if pkt.len() < 28 || (pkt[0] >> 4) != 4 {
return None; // need ≥20B IPv4 header + ≥8B ICMP; IPv4 only
}
let ihl = ((pkt[0] & 0x0f) as usize) * 4;
if ihl < 20 || pkt.len() < ihl + 8 || pkt[9] != 1 || pkt[ihl] != 8 {
return None; // proto must be ICMP (1) and type echo-request (8)
}
let mut out = pkt.to_vec();
for i in 0..4 {
out.swap(12 + i, 16 + i); // swap source and destination IPv4 addresses
}
out[ihl] = 0; // ICMP echo reply
out[ihl + 2] = 0;
out[ihl + 3] = 0;
let csum = checksum16(&out[ihl..]);
out[ihl + 2] = (csum >> 8) as u8;
out[ihl + 3] = (csum & 0xff) as u8;
Some(out)
}
/// Internet checksum (RFC 1071): one's-complement sum of 16-bit big-endian words.
pub(crate) fn checksum16(data: &[u8]) -> u16 {
let mut sum = 0u32;
let mut i = 0;
while i + 1 < data.len() {
sum += u16::from_be_bytes([data[i], data[i + 1]]) as u32;
i += 2;
}
if i < data.len() {
sum += (data[i] as u32) << 8;
}
while (sum >> 16) != 0 {
sum = (sum & 0xffff) + (sum >> 16);
}
!(sum as u16)
}
/// Read packets from the TUN, encrypt to the routing peer (initiating a handshake
/// via the relay on first contact), and send `Data` frames on the peer\'s current
/// path (direct if confirmed, else the relay).
fn tun_pump(pump: Pump, acl: Option<Acl>) -> io::Result<()> {
let Pump {
tun,
sock,
table,
peers,
kp,
my_ip,
relay,
} = pump;
let mut buf = [0u8; 2048];
// Peers already logged as ACL-denied, so the drop notice prints once each.
let mut acl_denied_logged: HashSet<Ipv4Addr> = HashSet::new();
loop {
let n = match tun.recv(&mut buf) {
Ok(0) => return Ok(()),
Ok(n) => n,
Err(ref e) if e.kind() == io::ErrorKind::Interrupted => continue,
Err(e) => return Err(e),
};
let pkt = &buf[..n];
// Destination overlay IP — IPv4 (header dst @16) or IPv6 (header dst @24).
let dest: IpAddr = match pkt[0] >> 4 {
4 if n >= 20 => IpAddr::V4(Ipv4Addr::new(pkt[16], pkt[17], pkt[18], pkt[19])),
6 if n >= 40 => {
let mut b = [0u8; 16];
b.copy_from_slice(&pkt[24..40]);
IpAddr::V6(Ipv6Addr::from(b))
}
_ => continue,
};
// Snapshot the current peer table for this iteration: hold the read lock
// only long enough to clone the inner Arc, then release it before any I/O.
let pt = { Arc::clone(&*peers.read().unwrap()) };
// Route to the carrying peer: an exact overlay peer (v4 or derived v6), or
// the gateway peer advertising the subnet `dest` falls in. The session is
// keyed by the PEER\'s overlay IPv4, so all of a peer\'s traffic shares it.
let Some(peer) = pt.route_to_ip(dest) else {
continue; // fail-closed: no peer/route for this destination
};
// Fine-grained ACL gate (fail-closed): if enforcement is on and this
// node is not permitted to reach the routing peer, DROP the packet —
// no handshake, no send, no per-peer state. A one-time notice per
// denied peer keeps the data path silent but observable.
if let Some(acl) = acl.as_ref() {
if !acl.permits(peer) {
if acl_denied_logged.insert(peer.overlay_ip) {
let label = if peer.name.is_empty() {
peer.overlay_ip.to_string()
} else {
peer.name.clone()
};
eprintln!("acl: denied {label}");
}
continue;
}
}
let peer_ip = peer.overlay_ip;
let peer_pub = peer.public_key;
let peer_endpoint = peer.endpoint;
// pt (and the peer borrow) are no longer needed past this point.
drop(pt);
let mut out: Vec<(Vec<u8>, SocketAddr)> = Vec::new();
{
let mut t = table.lock().unwrap();
let st = t.entry(peer_ip).or_default();
if let Some(sess) = st.session.as_mut() {
let ct = sess.encrypt(pkt);
let addr = st.direct.unwrap_or(relay);
if let Some(f) = Frame::new(FrameKind::Data, my_ip, peer_ip, ct) {
out.push((f.encode(), addr));
}
} else if st.pending.is_none() {
// First contact: initiate via the relay and drop this packet (the
// upper layer retransmits once a session exists).
if let Ok(eph) = rand32() {
let (state, msg1) = noise::initiate(&kp, &peer_pub, &eph);
st.pending = Some(state);
if let Some(f) = Frame::new(FrameKind::HandshakeInit, my_ip, peer_ip, msg1) {
let bytes = f.encode();
for addr in handshake_paths(st, relay, peer_endpoint) {
out.push((bytes.clone(), addr));
}
}
}
}
}
for (bytes, addr) in out {
let _ = sock.send_to(&bytes, addr);
}
}
}
/// Receive frames: confirm direct paths, complete handshakes, act on `PeerAddr`
/// hints, or decrypt `Data` and write the inner packet to the TUN. Survives
/// transient recv errors (logs once, sleeps 100 ms, continues).
fn udp_pump(pump: Pump) {
let Pump {
tun,
sock,
table,
peers,
kp,
my_ip,
relay,
} = pump;
let mut buf = [0u8; 4096];
loop {
let (n, from) = match sock.recv_from(&mut buf) {
Ok(v) => v,
Err(ref e) if e.kind() == io::ErrorKind::Interrupted => continue,
Err(e) => {
// Transient error (e.g. EAGAIN / ENETDOWN): log, back off briefly,
// and retry — never kill the receive path on a recoverable error.
eprintln!("akurai-node: udp recv error: {e} — retrying");
thread::sleep(Duration::from_millis(100));
continue;
}
};
let Some(frame) = Frame::decode(&buf[..n]) else {
continue;
};
// Snapshot the current peer table for this iteration: hold the read lock
// only long enough to clone the inner Arc, then release it before any I/O.
let ptab = { Arc::clone(&*peers.read().unwrap()) };
// A frame arriving straight from a known peer (not the relay) proves a
// working direct path — switch this peer\'s outbound path to it.
if from != relay && frame.kind != FrameKind::PeerAddr && ptab.get(&frame.src).is_some() {
if let Ok(mut t) = table.lock() {
let st = t.entry(frame.src).or_default();
if st.direct.is_none() {
eprintln!("akurai-node: direct path to {} via {from}", frame.src);
}
st.direct = Some(from);
}
}
match frame.kind {
FrameKind::PeerAddr => {
// The relay hints that peer `frame.src` is reachable at the payload
// address. Remember it and probe to open any NAT hole.
if let (Some(addr), true) =
(parse_addr(&frame.payload), ptab.get(&frame.src).is_some())
{
{
let mut t = table.lock().unwrap();
let st = t.entry(frame.src).or_default();
if st.direct.is_none() {
st.candidate = Some(addr);
}
}
if let Some(f) = Frame::new(
FrameKind::Keepalive,
my_ip,
Ipv4Addr::UNSPECIFIED,
Vec::new(),
) {
let _ = sock.send_to(&f.encode(), addr);
}
}
}
FrameKind::HandshakeInit => {
let Ok(eph) = rand32() else { continue };
if let Some((keys, msg2, init_pub)) = noise::respond(&kp, &eph, &frame.payload) {
match ptab.get(&frame.src) {
Some(p) if p.public_key == init_pub => {
let install = {
let mut t = table.lock().unwrap();
let st = t.entry(frame.src).or_default();
if st.session.is_none() {
st.session = Some(Session::new(keys));
true
} else {
false
}
};
if !install {
continue;
}
if let Some(reply) =
Frame::new(FrameKind::HandshakeResp, my_ip, frame.src, msg2)
{
let _ = sock.send_to(&reply.encode(), from); // reply on arrival path
}
}
_ => {} // unknown / mismatched peer: drop
}
}
}
FrameKind::HandshakeResp => {
let mut t = table.lock().unwrap();
let st = t.entry(frame.src).or_default();
if let Some(state) = st.pending.take() {
if let Some(keys) = noise::finalize(state, &frame.payload) {
st.session = Some(Session::new(keys));
}
}
}
FrameKind::Data => {
let mut t = table.lock().unwrap();
if let Some(st) = t.get_mut(&frame.src) {
if let Some(sess) = st.session.as_mut() {
if let Some(pt) = sess.decrypt(&frame.payload) {
drop(t);
let _ = tun.send(&pt);
}
}
}
}
FrameKind::Keepalive => {} // direct-path confirmation handled above
}
}
}
#[cfg(test)]
mod selftest_tests {
use super::*;
#[test]
fn echo_request_becomes_reply() {
// IPv4 (20B) + ICMP echo request (type 8). src 1.1.1.1, dst 2.2.2.2.
let mut pkt = vec![
0x45, 0, 0, 28, 0, 0, 0, 0, 64, 1, 0, 0, // IPv4 header
1, 1, 1, 1, 2, 2, 2, 2, // src, dst
8, 0, 0, 0, 0, 1, 0, 1, // ICMP echo request
];
pkt[22] = 0xf7; // any non-zero icmp checksum; the reply recomputes it
pkt[23] = 0xfd;
let reply = icmp_echo_reply(&pkt).expect("should transform");
assert_eq!(&reply[12..16], &[2, 2, 2, 2]); // src now the old dst
assert_eq!(&reply[16..20], &[1, 1, 1, 1]); // dst now the old src
assert_eq!(reply[20], 0); // ICMP type echo-reply
assert_eq!(checksum16(&reply[20..]), 0); // valid ⇒ folds to zero
}
#[test]
fn non_icmp_is_ignored() {
let tcp = vec![
0x45u8, 0, 0, 28, 0, 0, 0, 0, 64, 6, 0, 0, 1, 1, 1, 1, 2, 2, 2, 2, 0, 0, 0, 0, 0, 0, 0,
0,
];
assert!(icmp_echo_reply(&tcp).is_none());
}
}