crates/installer/nordisk-pairing/src/installer_link.rs
//! **How the installer reaches a VM during bootstrap** (0.3.0, owner
//! 2026-09-24). The protocol is always the same — the installer's mutual TLS
//! ([`crate::bootstrap`]) — and only the PATH differs, chosen by the seed's
//! tunnel kind ([`reach`], which every [`crate::tunnel::Tunnel`] driver
//! answers through [`crate::tunnel::Tunnel::installer_reach`]):
//!
//! | tunnel | path |
//! |---|---|
//! | `wireguard` | the installer is a TEMPORARY PEER of the tunnel: every VM raises a WireGuard link from its first boot ([`vm_link`]) with the installer as its one peer, and binds the bootstrap listener on the link address ONLY — nothing of it on the public interface. The installer (behind NAT) initiates ([`WgPath`]) with a keepalive. |
//! | `mullvad` | the same: the estate mesh stays WireGuard, Mullvad is only each VM's exit. |
//! | `tailscale` | the installer would join with a one-time key and leave at the end — refused by name like the driver itself (`[tunnel-tailscale-unimplemented]`). |
//! | `none` | **REFUSED** `[installer-link-none]` (owner 2026-09-25: WireGuard + mutual TLS ALWAYS). [`DirectPath`] is a loopback harness for tests only. |
//!
//! When the installation is complete the installer tells each VM to close:
//! the VM answers and the install is gone ([`crate::installer::Client::install_gone`]).
//! Since 0.5.4 the link and its installer peer are KEPT for the VM's life —
//! the same listener is the log door ([`crate::bootstrap::serve_for_life`]).
//! A kept peer that has left goes QUIET on the VM: the installer only ever
//! dials the VM (no endpoint, no keepalive on this end), so the dataplane
//! (edda wg-appliance-core, `dataplane::GONE_AFTER`, a0e100a) drops its
//! session once the VM has sent to it and heard nothing for 14 s — before
//! boringtun would begin handshakes nobody answers (`HANDSHAKE(REKEY_TIMEOUT)`
//! every 5 s, then `CONNECTION_EXPIRED`) — and the installer's next handshake
//! reopens it. Proven on the real link: tests/installer_link_goes_quiet.rs.
//!
//! **The link's addresses are the link's own**, the same on every VM:
//! [`LINK_INSTALLER`] and [`LINK_VM`] on a /30. That is not a clash — each VM
//! is its own WireGuard session on the installer's side (one session per VM,
//! in user space), so two VMs with the same link address are two different
//! wires.
//!
//! **The installer's side runs in user space** (feature `installer-wireguard`):
//! the operator's machine holds no TUN device and no root, so the WireGuard
//! session is boringtun's (the suite's one engine) and the TCP inside it is
//! smoltcp's. The VM's side is a real interface ([`vm_link`], feature
//! `wireguard`): the VM is PID 1 and owns its network.
use crate::seed::TunnelKind;
use crate::tunnel::TunnelError;
use std::io::{Read, Write};
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4, TcpStream};
use std::time::Duration;
/// The installer's address on the link.
pub const LINK_INSTALLER: Ipv4Addr = Ipv4Addr::new(10, 45, 0, 1);
/// Every VM's address on the link.
pub const LINK_VM: Ipv4Addr = Ipv4Addr::new(10, 45, 0, 2);
pub const LINK_PREFIX: u8 = 30;
/// The UDP port each VM's end of the link listens on (the mesh's relay on
/// the front is 51820).
pub const LINK_PORT: u16 = 51821;
/// The VM's link interface.
pub const LINK_IFACE: &str = "wg-installer";
/// The MTU inside the link.
pub const LINK_MTU: usize = 1380;
/// **The path the installer takes to a VM** — since 0.4.0 exactly one: the
/// installer's WireGuard link, under mutual TLS (owner 2026-09-25).
/// `tunnel = none` is refused by name (`[installer-link-none]`).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Reach {
/// Through the installer's WireGuard link (`wireguard`, `mullvad`).
Tunnel,
}
/// **The one decision**, shared by the VM (where to bind, whether to raise a
/// link) and the installer (how to dial).
pub fn reach(kind: TunnelKind) -> Result<Reach, TunnelError> {
match kind {
TunnelKind::None => Err(TunnelError::InstallerLinkNone),
TunnelKind::Wireguard | TunnelKind::Mullvad => Ok(Reach::Tunnel),
TunnelKind::Tailscale => Err(TunnelError::Unimplemented {
kind: TunnelKind::Tailscale,
missing: "a one-time installer node on the tailnet (a pure-Rust Tailscale coordination client does not exist in the suite)",
}),
}
}
/// Where a VM binds its bootstrap listener: the link address ONLY — never
/// the public interface.
pub fn vm_bind(r: Reach) -> Ipv4Addr {
match r {
Reach::Tunnel => LINK_VM,
}
}
/// A byte stream the installer speaks TLS over.
pub trait Wire: Read + Write {}
impl<T: Read + Write> Wire for T {}
/// **The installer's path to ONE VM.**
pub trait Path: Send {
/// One connection to the VM's bootstrap listener.
fn connect(&mut self, timeout: Duration) -> Result<Box<dyn Wire + '_>, String>;
/// After the close: does the VM still answer on this path? `Ok(true)` =
/// nothing answers within `wait` (the listener is gone, the link dropped).
fn gone(&mut self, wait: Duration) -> Result<bool, String>;
fn describe(&self) -> String;
}
/// **A loopback harness for tests** (0.4.0): TCP straight to an address.
/// [`reach`] never chooses it — production is always the link.
pub struct DirectPath {
pub at: SocketAddrV4,
}
impl Path for DirectPath {
fn connect(&mut self, timeout: Duration) -> Result<Box<dyn Wire + '_>, String> {
let s = TcpStream::connect_timeout(&SocketAddr::V4(self.at), timeout).map_err(|e| format!("could not be reached: {}: {e}", self.at))?;
let _ = s.set_read_timeout(Some(timeout));
let _ = s.set_write_timeout(Some(timeout));
Ok(Box::new(s))
}
/// Asked until the port refuses (the VM answers its close first and
/// drops the listener a moment after).
fn gone(&mut self, wait: Duration) -> Result<bool, String> {
let t0 = std::time::Instant::now();
while t0.elapsed() < wait {
if TcpStream::connect_timeout(&SocketAddr::V4(self.at), Duration::from_secs(2)).is_err() {
return Ok(true);
}
std::thread::sleep(Duration::from_millis(250));
}
Ok(false)
}
fn describe(&self) -> String {
format!("{} (direct mutual TLS)", self.at)
}
}
/// The installer's keys for the link, kept on the operator's machine: its own
/// private key, and the public half of the medium's VM key.
#[derive(Clone, PartialEq, Eq)]
pub struct LinkKeys {
pub installer_private: [u8; 32],
pub vm_public: [u8; 32],
/// The link's pre-shared key (0.5.29), the one its medium's seed carries; `None` for a
/// link from before 0.5.29.
pub psk: Option<[u8; 32]>,
}
impl std::fmt::Debug for LinkKeys {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("LinkKeys { .. }")
}
}
#[cfg(feature = "installer-wireguard")]
pub use wg::WgPath;
#[cfg(feature = "installer-wireguard")]
mod wg {
use super::*;
use boringtun::noise::{Tunn, TunnResult};
use boringtun::x25519::{PublicKey, StaticSecret};
use smoltcp::iface::{Config, Interface, SocketHandle, SocketSet};
use smoltcp::phy::{Device, DeviceCapabilities, Medium};
use smoltcp::socket::tcp;
use smoltcp::wire::{HardwareAddress, IpAddress, IpCidr};
use std::collections::VecDeque;
use std::net::UdpSocket;
use std::time::Instant;
/// The in-memory "device" smoltcp reads and writes: IP packets out of the
/// tunnel (rx) and into it (tx).
#[derive(Default)]
struct Queues {
rx: VecDeque<Vec<u8>>,
tx: VecDeque<Vec<u8>>,
}
struct Rx(Vec<u8>);
struct Tx<'a>(&'a mut VecDeque<Vec<u8>>);
impl smoltcp::phy::RxToken for Rx {
fn consume<R, F: FnOnce(&[u8]) -> R>(self, f: F) -> R {
f(&self.0)
}
}
impl smoltcp::phy::TxToken for Tx<'_> {
fn consume<R, F: FnOnce(&mut [u8]) -> R>(self, len: usize, f: F) -> R {
let mut buf = vec![0u8; len];
let r = f(&mut buf);
self.0.push_back(buf);
r
}
}
impl Device for Queues {
type RxToken<'a> = Rx;
type TxToken<'a> = Tx<'a>;
fn receive(&mut self, _t: smoltcp::time::Instant) -> Option<(Rx, Tx<'_>)> {
let p = self.rx.pop_front()?;
Some((Rx(p), Tx(&mut self.tx)))
}
fn transmit(&mut self, _t: smoltcp::time::Instant) -> Option<Tx<'_>> {
Some(Tx(&mut self.tx))
}
fn capabilities(&self) -> DeviceCapabilities {
let mut c = DeviceCapabilities::default();
c.medium = Medium::Ip;
c.max_transmission_unit = LINK_MTU;
c
}
}
/// **One VM's link, the engine** — a WireGuard session (boringtun) over
/// one UDP socket, and a TCP/IP stack (smoltcp) inside it, all in user
/// space. ONE per VM per process ([`LINKS`]): the VM keeps one endpoint
/// for its one peer (the installer) and moves it to the newest datagram's
/// source, so a second socket under the same key would steal the first
/// one's answers (estate-box3 two-link-readers).
struct Engine {
endpoint: SocketAddrV4,
udp: UdpSocket,
tunn: Tunn,
iface: Interface,
dev: Queues,
sockets: SocketSet<'static>,
epoch: Instant,
next_port: u16,
net: Vec<u8>,
}
type Shared = std::sync::Arc<std::sync::Mutex<Engine>>;
/// Every link this process holds, by (VM endpoint, installer key, VM key):
/// its engine, and the turn at the VM's door its readers take.
static LINKS: std::sync::Mutex<Vec<Link>> = std::sync::Mutex::new(Vec::new());
/// One entry of [`LINKS`]: (VM endpoint, installer key, VM key, the link's PSK), the
/// engine, the turn.
type Link = ((SocketAddrV4, [u8; 32], [u8; 32], Option<[u8; 32]>), Shared, std::sync::Arc<Turn>);
/// **The turn at the VM's door.** The box's bootstrap listener and log
/// door (`bootstrap::serve_*`) serve ONE connection at a time, while its
/// kernel establishes every connection at once: a second reader's
/// connect succeeds, its request waits in the box's queue, and its read
/// timeout runs the whole time another reader is being served — so a
/// slow exchange of one reader cost the next its answer (estate-box W6,
/// `W6-two-readers-one-link`: "reader A #0: the request: nothing from
/// the VM over the link" while reader B got all four, release gate red 2
/// of 3). The readers of one VM in this process therefore take turns
/// HERE, in the order they asked: a connection is opened when its reader
/// has the turn, and a reader's timeout is for its own exchange.
#[derive(Default)]
struct Turn {
q: std::sync::Mutex<TurnQueue>,
moved: std::sync::Condvar,
}
/// A ticket queue: the order readers asked in, the ticket served, and
/// tickets whose readers stopped waiting.
#[derive(Default)]
struct TurnQueue {
next: u64,
serving: u64,
given_up: std::collections::BTreeSet<u64>,
}
/// The turn, held: given back (to the next ticket still waiting) when dropped.
struct Held(std::sync::Arc<Turn>);
impl Drop for Held {
fn drop(&mut self) {
let mut q = self.0.q.lock().unwrap_or_else(|p| p.into_inner());
q.serving += 1;
loop {
let at = q.serving;
if !q.given_up.remove(&at) {
break;
}
q.serving += 1;
}
self.0.moved.notify_all();
}
}
/// **How long a reader waits for the turn** before it goes ahead without
/// it (as every reader did before turns). The box gives one connection at
/// most `bootstrap::REQUEST_BUDGET` to ask; a turn held longer is a
/// reader that stopped (a process that died mid-answer, a stream leaked),
/// and the others must not wait on it for ever.
const TURN_WAIT: Duration = crate::bootstrap::REQUEST_BUDGET;
/// Take the turn: `Some` once it is this reader's, `None` after `within`.
fn take_turn(turn: &std::sync::Arc<Turn>, within: Duration) -> Option<Held> {
let mut q = turn.q.lock().unwrap_or_else(|p| p.into_inner());
let mine = q.next;
q.next += 1;
let until = Instant::now() + within;
while q.serving != mine {
let left = until.saturating_duration_since(Instant::now());
if left.is_zero() {
q.given_up.insert(mine);
return None;
}
q = turn.moved.wait_timeout(q, left).unwrap_or_else(|p| p.into_inner()).0;
}
Some(Held(turn.clone()))
}
fn lock(e: &Shared) -> std::sync::MutexGuard<'_, Engine> {
e.lock().unwrap_or_else(|p| p.into_inner())
}
/// **The installer as a temporary peer of one VM's link** — a handle on
/// the process's one [`Engine`] for that VM: every reader in this process
/// (the install, the capture follower, a probe, a re-admission) shares
/// its socket and session, each with its own TCP connection inside.
pub struct WgPath {
endpoint: SocketAddrV4,
keys: LinkKeys,
engine: Shared,
turn: std::sync::Arc<Turn>,
}
fn now(epoch: Instant) -> smoltcp::time::Instant {
smoltcp::time::Instant::from_millis(epoch.elapsed().as_millis() as i64)
}
fn tunn_of(keys: &LinkKeys, index: u32) -> Tunn {
Tunn::new(StaticSecret::from(keys.installer_private), PublicKey::from(keys.vm_public), keys.psk, Some(15), index, None)
}
impl Engine {
fn new(endpoint: SocketAddrV4, keys: &LinkKeys) -> Result<Engine, String> {
let udp = UdpSocket::bind("0.0.0.0:0").map_err(|e| format!("the installer's link socket: {e}"))?;
udp.connect(endpoint).map_err(|e| format!("the installer's link to {endpoint}: {e}"))?;
udp.set_read_timeout(Some(Duration::from_millis(5))).map_err(|e| e.to_string())?;
let epoch = Instant::now();
let mut dev = Queues::default();
let mut cfg = Config::new(HardwareAddress::Ip);
cfg.random_seed = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_nanos() as u64).unwrap_or(7);
let mut iface = Interface::new(cfg, &mut dev, now(epoch));
iface.update_ip_addrs(|a| {
let _ = a.push(IpCidr::new(IpAddress::Ipv4(LINK_INSTALLER), LINK_PREFIX));
});
let seed_port = 40000 + (cfg_port_seed() % 20000) as u16;
Ok(Engine { endpoint, udp, tunn: tunn_of(keys, 1), iface, dev, sockets: SocketSet::new(Vec::new()), epoch, next_port: seed_port, net: vec![0u8; 65536] })
}
/// One turn of the engine: timers, datagrams in, the stack, packets out.
fn pump(&mut self) {
if let TunnResult::WriteToNetwork(p) = self.tunn.update_timers(&mut self.net) {
let _ = self.udp.send(p);
}
let mut dgram = [0u8; 2048];
for _ in 0..64 {
let n = match self.udp.recv(&mut dgram) {
Ok(n) => n,
Err(_) => break,
};
let mut first = true;
loop {
let r = self.tunn.decapsulate(if first { Some(std::net::IpAddr::V4(*self.endpoint.ip())) } else { None }, if first { &dgram[..n] } else { &[] }, &mut self.net);
first = false;
match r {
TunnResult::WriteToNetwork(p) => {
let _ = self.udp.send(p);
}
TunnResult::WriteToTunnelV4(p, src) => {
if src == LINK_VM {
self.dev.rx.push_back(p.to_vec());
}
break;
}
_ => break,
}
}
}
let t = now(self.epoch);
self.iface.poll(t, &mut self.dev, &mut self.sockets);
while let Some(p) = self.dev.tx.pop_front() {
if let TunnResult::WriteToNetwork(out) = self.tunn.encapsulate(&p, &mut self.net) {
let _ = self.udp.send(out);
}
}
}
}
impl WgPath {
/// The link to the VM whose end listens at `endpoint` (its public
/// address and [`LINK_PORT`]) — this process's one engine for it,
/// made on first use.
pub fn new(endpoint: SocketAddrV4, keys: LinkKeys) -> Result<WgPath, String> {
let id = (endpoint, keys.installer_private, keys.vm_public, keys.psk);
let mut links = LINKS.lock().unwrap_or_else(|p| p.into_inner());
let (engine, turn) = match links.iter().find(|(k, _, _)| *k == id) {
Some((_, e, t)) => (e.clone(), t.clone()),
None => {
let e: Shared = std::sync::Arc::new(std::sync::Mutex::new(Engine::new(endpoint, &keys)?));
let t = std::sync::Arc::new(Turn::default());
links.push((id, e.clone(), t.clone()));
(e, t)
}
};
Ok(WgPath { endpoint, keys, engine, turn })
}
/// One TCP connection to the VM's listener inside the link, within `timeout`.
fn try_connect(&mut self, timeout: Duration) -> Result<SocketHandle, String> {
let sock = tcp::Socket::new(tcp::SocketBuffer::new(vec![0u8; 65536]), tcp::SocketBuffer::new(vec![0u8; 65536]));
let h = {
let mut e = lock(&self.engine);
let h = e.sockets.add(sock);
e.next_port = if e.next_port >= 60000 { 40000 } else { e.next_port + 1 };
let port = e.next_port;
let Engine { iface, sockets, .. } = &mut *e;
let cx = iface.context();
let s = sockets.get_mut::<tcp::Socket>(h);
if let Err(err) = s.connect(cx, (IpAddress::Ipv4(LINK_VM), crate::bootstrap::PORT), port) {
sockets.remove(h);
return Err(format!("the link's TCP to {LINK_VM}: {err:?}"));
}
h
};
let t0 = Instant::now();
loop {
{
let mut e = lock(&self.engine);
e.pump();
let s = e.sockets.get_mut::<tcp::Socket>(h);
match s.state() {
tcp::State::Established => return Ok(h),
tcp::State::Closed | tcp::State::TimeWait => {
e.sockets.remove(h);
return Err(format!("could not be reached: the VM at {} refused the bootstrap listener inside the link", self.endpoint));
}
_ => {}
}
if t0.elapsed() > timeout {
e.sockets.get_mut::<tcp::Socket>(h).abort();
e.pump();
e.sockets.remove(h);
let hs = match e.tunn.stats().0 {
Some(ago) => format!("the link handshook {} s ago", ago.as_secs()),
None => "no WireGuard handshake yet".to_string(),
};
return Err(format!("could not be reached: {} over the installer's link ({hs})", self.endpoint));
}
}
std::thread::yield_now();
}
}
/// A connect that found no VM: the session starts over (a VM that
/// restarted has forgotten it; waiting out boringtun's re-key would
/// cost a reader its whole timeout) — as a NEW HANDSHAKE on the link's
/// one `Tunn`, sent now, never a new `Tunn`.
///
/// The session is every reader's ([`LINKS`]): a new `Tunn` threw away
/// the one another reader's open connection was using, so whatever the
/// VM had already sealed under it — that reader's answer, its TCP's
/// retransmissions — could no longer be opened, and the reader waited
/// out the box's TCP back-off on top of the stall that made this
/// connect slow (estate-box W6, `W6-two-readers-one-link`: a box busy
/// for a few seconds became a read lost past its 20 s). A handshake on
/// the same `Tunn` keeps the old session in its ring until the new one
/// takes over, and a VM that did forget it answers the handshake.
fn fresh_session(&self) {
let mut e = lock(&self.engine);
let Engine { tunn, udp, net, .. } = &mut *e;
if let TunnResult::WriteToNetwork(p) = tunn.format_handshake_initiation(net, true) {
let _ = udp.send(p);
}
}
}
fn cfg_port_seed() -> u64 {
std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.subsec_nanos() as u64).unwrap_or(1)
}
/// One TCP connection inside the link — and, while it lives, the turn at
/// the VM's door ([`Turn`]; given back after the connection is closed).
struct LinkStream {
engine: Shared,
h: SocketHandle,
timeout: Duration,
_turn: Option<Held>,
}
impl Read for LinkStream {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
let t0 = Instant::now();
loop {
{
let mut e = lock(&self.engine);
e.pump();
let s = e.sockets.get_mut::<tcp::Socket>(self.h);
if s.can_recv() {
let n = s.recv_slice(buf).map_err(|e| std::io::Error::other(format!("{e:?}")))?;
if n > 0 {
return Ok(n);
}
} else if !s.may_recv() {
return Ok(0);
}
}
if t0.elapsed() > self.timeout {
return Err(std::io::Error::new(std::io::ErrorKind::TimedOut, "nothing from the VM over the link"));
}
std::thread::yield_now();
}
}
}
impl Write for LinkStream {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
let t0 = Instant::now();
loop {
{
let mut e = lock(&self.engine);
e.pump();
let s = e.sockets.get_mut::<tcp::Socket>(self.h);
if s.can_send() {
let n = s.send_slice(buf).map_err(|e| std::io::Error::other(format!("{e:?}")))?;
if n > 0 {
return Ok(n);
}
} else if !s.may_send() {
return Err(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "the VM closed the connection"));
}
}
if t0.elapsed() > self.timeout {
return Err(std::io::Error::new(std::io::ErrorKind::TimedOut, "the link would not take the bytes"));
}
std::thread::yield_now();
}
}
fn flush(&mut self) -> std::io::Result<()> {
let t0 = Instant::now();
loop {
{
let mut e = lock(&self.engine);
e.pump();
let s = e.sockets.get_mut::<tcp::Socket>(self.h);
if s.send_queue() == 0 || !s.may_send() {
return Ok(());
}
}
if t0.elapsed() > self.timeout {
return Err(std::io::Error::new(std::io::ErrorKind::TimedOut, "the link would not drain"));
}
std::thread::yield_now();
}
}
}
impl Drop for LinkStream {
fn drop(&mut self) {
let mut e = lock(&self.engine);
e.sockets.get_mut::<tcp::Socket>(self.h).close();
for _ in 0..20 {
e.pump();
if !e.sockets.get_mut::<tcp::Socket>(self.h).is_open() {
break;
}
}
e.sockets.remove(self.h);
}
}
impl Path for WgPath {
/// The turn at the VM's door first ([`Turn`]), then the connection;
/// `timeout` counts from the turn, never from the wait for it.
fn connect(&mut self, timeout: Duration) -> Result<Box<dyn Wire + '_>, String> {
let turn = take_turn(&self.turn, TURN_WAIT);
let t0 = Instant::now();
let mut fresh = false;
loop {
match self.try_connect(timeout.saturating_sub(t0.elapsed()).min(if fresh { timeout } else { Duration::from_secs(4) })) {
Ok(h) => return Ok(Box::new(LinkStream { engine: self.engine.clone(), h, timeout, _turn: turn })),
Err(e) if fresh || t0.elapsed() >= timeout => return Err(e),
// No answer within a few seconds on the shared session: the
// VM may have restarted and forgotten it — start over once.
Err(e) if e.starts_with("could not be reached: the VM at") => return Err(e),
Err(_) => {
fresh = true;
self.fresh_session();
}
}
}
}
/// A FRESH handshake each round (a new session, so nothing cached
/// answers): the link is gone once a round of three seconds brings no
/// answer. The VM answers its close first and drops the link a moment
/// after, so an answer early in `wait` is asked again.
fn gone(&mut self, wait: Duration) -> Result<bool, String> {
let t0 = Instant::now();
let mut index = 2u32;
let mut dgram = [0u8; 2048];
let mut net = vec![0u8; 2048];
// Its OWN socket: the probe must not move the VM's endpoint off
// the shared engine every other reader in this process uses.
let udp = UdpSocket::bind("0.0.0.0:0").map_err(|e| format!("the installer's probe socket: {e}"))?;
udp.connect(self.endpoint).map_err(|e| format!("the installer's probe to {}: {e}", self.endpoint))?;
udp.set_read_timeout(Some(Duration::from_millis(5))).map_err(|e| e.to_string())?;
while t0.elapsed() < wait {
index += 1;
let mut t = tunn_of(&self.keys, index);
let mut buf = [0u8; 256];
let init = match t.format_handshake_initiation(&mut buf, true) {
TunnResult::WriteToNetwork(p) => p.to_vec(),
other => return Err(format!("boringtun: {other:?}")),
};
let _ = udp.send(&init);
let round = Instant::now();
let mut answered = false;
while round.elapsed() < Duration::from_secs(3) {
if let Ok(n) = udp.recv(&mut dgram) {
if matches!(t.decapsulate(Some(std::net::IpAddr::V4(*self.endpoint.ip())), &dgram[..n], &mut net), TunnResult::WriteToNetwork(_) | TunnResult::Done) {
answered = true;
break;
}
}
}
if !answered {
return Ok(true);
}
std::thread::sleep(Duration::from_millis(500));
}
Ok(false)
}
fn describe(&self) -> String {
format!("{} (the installer's WireGuard link, {LINK_INSTALLER} → {LINK_VM})", self.endpoint)
}
}
/// ★ estate-box W6 (`W6-two-readers-one-link`, release gate red 2 of 3 on
/// byte-identical binaries, oden 2026-10-04: "reader A #0: the request:
/// nothing from the VM over the link", rounds of 8-38 s): every reader in
/// the process shares THIS engine's one session ([`LINKS`]), so the
/// session belongs to all of them. A connect that hears nothing for 4 s —
/// a box busy or stalled for those seconds, while another reader's
/// connection is open on the link — must not throw that session away:
/// what the VM already sealed under it (an answer, its TCP's
/// retransmissions) would be unreadable, and the other reader waits out
/// the box's TCP back-off on top of the stall.
#[cfg(test)]
mod start_over_tests {
use super::*;
fn installer_public(k: &LinkKeys) -> [u8; 32] {
PublicKey::from(&StaticSecret::from(k.installer_private)).to_bytes()
}
/// The smallest IPv4 packet the VM's end could seal for the installer.
fn packet_for_the_installer() -> Vec<u8> {
let mut p = vec![0u8; 20];
p[0] = 0x45;
p[3] = 20;
p[8] = 64;
p[9] = 6;
p[12..16].copy_from_slice(&LINK_VM.octets());
p[16..20].copy_from_slice(&LINK_INSTALLER.octets());
p
}
/// The VM's end, by hand: answer every datagram the engine sent it.
fn vm_answers(vm_udp: &UdpSocket, vm: &mut Tunn) -> usize {
let (mut dgram, mut out) = ([0u8; 2048], vec![0u8; 2048]);
let mut handled = 0;
while let Ok((n, from)) = vm_udp.recv_from(&mut dgram) {
handled += 1;
let mut first = true;
loop {
let r = vm.decapsulate(None, if first { &dgram[..n] } else { &[] }, &mut out);
first = false;
match r {
TunnResult::WriteToNetwork(p) => {
let _ = vm_udp.send_to(p, from);
}
_ => break,
}
}
}
handled
}
/// The link up: the engine handshakes, the VM answers, the engine confirms.
fn handshake(path: &WgPath, vm_udp: &UdpSocket, vm: &mut Tunn) {
{
let mut e = lock(&path.engine);
let Engine { tunn, udp, net, .. } = &mut *e;
if let TunnResult::WriteToNetwork(p) = tunn.format_handshake_initiation(net, false) {
udp.send(p).unwrap();
}
}
let t0 = Instant::now();
while lock(&path.engine).tunn.stats().0.is_none() && t0.elapsed() < Duration::from_secs(5) {
vm_answers(vm_udp, vm);
lock(&path.engine).pump();
}
assert!(lock(&path.engine).tunn.stats().0.is_some(), "the link never handshook");
vm_answers(vm_udp, vm); // the engine's confirming keepalive
}
/// Does the engine open what the VM sealed under its current session?
fn engine_opens(path: &WgPath, vm: &mut Tunn) -> bool {
let mut sealed = vec![0u8; 2048];
let sealed = match vm.encapsulate(&packet_for_the_installer(), &mut sealed) {
TunnResult::WriteToNetwork(p) => p.to_vec(),
other => panic!("the VM could not seal: {other:?}"),
};
let mut e = lock(&path.engine);
let Engine { tunn, net, endpoint, .. } = &mut *e;
matches!(tunn.decapsulate(Some(std::net::IpAddr::V4(*endpoint.ip())), &sealed, net), TunnResult::WriteToTunnelV4(..))
}
/// The engine's handshake toward a VM end keyed with `vm_psk`: true when the link is up.
fn link_comes_up(installer_psk: Option<[u8; 32]>, vm_psk: Option<[u8; 32]>, index: u32) -> bool {
let vm_private = [4u8; 32];
let k = LinkKeys { installer_private: [10u8; 32], vm_public: PublicKey::from(&StaticSecret::from(vm_private)).to_bytes(), psk: installer_psk };
let vm_udp = UdpSocket::bind("127.0.0.1:0").unwrap();
vm_udp.set_read_timeout(Some(Duration::from_millis(50))).unwrap();
let SocketAddr::V4(at) = vm_udp.local_addr().unwrap() else { unreachable!() };
let mut vm = Tunn::new(StaticSecret::from(vm_private), PublicKey::from(installer_public(&k)), vm_psk, None, index, None);
let path = WgPath::new(at, k).unwrap();
{
let mut e = lock(&path.engine);
let Engine { tunn, udp, net, .. } = &mut *e;
if let TunnResult::WriteToNetwork(p) = tunn.format_handshake_initiation(net, false) {
udp.send(p).unwrap();
}
}
let t0 = Instant::now();
while t0.elapsed() < Duration::from_secs(3) {
vm_answers(&vm_udp, &mut vm);
lock(&path.engine).pump();
if lock(&path.engine).tunn.stats().0.is_some() {
return true;
}
}
false
}
/// ★ **The installer's link handshakes only when both ends hold the same pre-shared
/// key** (0.5.29, lane WG-PSK-3), on the real engine and a boringtun VM end: the same
/// PSK on both ends is up; a missing or a wrong one on either end is refused; a link
/// from before 0.5.29 (no PSK on either end) is up as before.
#[test]
fn the_installers_link_handshakes_only_when_both_ends_hold_the_same_preshared_key() {
let (p, q) = ([21u8; 32], [22u8; 32]);
assert!(link_comes_up(Some(p), Some(p), 101), "the same PSK on both ends: up");
assert!(!link_comes_up(Some(p), None, 102), "the installer has a PSK, the VM none: refused");
assert!(!link_comes_up(None, Some(p), 103), "the VM has a PSK, the installer none: refused");
assert!(!link_comes_up(Some(p), Some(q), 104), "two different PSKs: refused");
assert!(link_comes_up(None, None, 105), "an old seed and an old link file: up as before");
}
#[test]
fn a_reader_that_starts_the_session_over_keeps_the_session_the_other_readers_use() {
let (vm_private, k) = ([3u8; 32], LinkKeys { installer_private: [9u8; 32], vm_public: PublicKey::from(&StaticSecret::from([3u8; 32])).to_bytes(), psk: None });
let vm_udp = UdpSocket::bind("127.0.0.1:0").unwrap();
vm_udp.set_read_timeout(Some(Duration::from_millis(50))).unwrap();
let at = match vm_udp.local_addr().unwrap() {
SocketAddr::V4(a) => a,
_ => unreachable!(),
};
let mut vm = Tunn::new(StaticSecret::from(vm_private), PublicKey::from(installer_public(&k)), None, None, 7, None);
let reader = WgPath::new(at, k.clone()).unwrap();
handshake(&reader, &vm_udp, &mut vm);
assert!(engine_opens(&reader, &mut vm), "the link is up: the engine opens what the VM seals");
// Another reader's connect heard nothing for 4 s (the box was busy)
// and starts the session over, while this reader's connection is open.
let other = WgPath::new(at, k.clone()).unwrap();
other.fresh_session();
// The VM has not heard of any of that yet: it seals the open
// connection's answer under the session it has.
assert!(engine_opens(&reader, &mut vm), "a reader starting the session over threw away the session another reader's open connection uses: what the VM sealed under it is unreadable now");
// …and a VM that DID forget it (it restarted) is reached again.
vm = Tunn::new(StaticSecret::from(vm_private), PublicKey::from(installer_public(&k)), None, None, 11, None);
other.fresh_session();
let t0 = Instant::now();
while t0.elapsed() < Duration::from_secs(5) {
vm_answers(&vm_udp, &mut vm);
lock(&other.engine).pump();
if vm.stats().0.is_some() && vm_answers(&vm_udp, &mut vm) == 0 && engine_opens(&other, &mut vm) {
return;
}
}
panic!("after starting over, the engine never reached the VM that had forgotten the session");
}
}
}
/// **The VM's end of the link** (feature `wireguard`): a real interface
/// [`LINK_IFACE`] at [`LINK_VM`], the medium's VM key, the installer as its
/// ONE peer (endpoint learned from the installer's handshake), listening on
/// the seed's port. `drop_installer` takes the whole link down.
#[cfg(feature = "wireguard")]
pub struct VmLink {
handle: Option<wg_appliance_core::TunnelHandle>,
pub bind: Ipv4Addr,
}
#[cfg(feature = "wireguard")]
impl VmLink {
/// Whether the installer's link is up (and whether it has handshaken).
pub fn up(&self) -> bool {
self.handle.is_some()
}
pub fn handshaken(&self) -> bool {
self.handle.as_ref().map(|h| h.handshaken()).unwrap_or(false)
}
/// **Drop the installer as a peer**: the link goes down whole.
pub fn drop_installer(&mut self) {
if let Some(mut h) = self.handle.take() {
h.shutdown();
}
}
}
/// Raise the VM's end of the link for this seed. `tunnel = none` is refused
/// (`[installer-link-none]`), and the bootstrap listener then never opens.
/// **The VM's end of the link, as a spec** (pure, 0.5.29): the medium's VM key, the installer
/// as its ONE peer with the link's pre-shared key ([`crate::seed::LinkSeed::psk`]; none for a
/// seed from before 0.5.29), the seed's port.
#[cfg(feature = "wireguard")]
pub fn vm_link_spec(l: &crate::seed::LinkSeed) -> Result<wg_appliance_core::TunnelSpec, TunnelError> {
use wg_appliance_core::{decode_key, PeerSpec, TunnelSpec};
let private = decode_key(&l.vm_private).map_err(|e| TunnelError::Failed(format!("installer_link.vm_private: {}", e.0)))?;
let installer = decode_key(&l.installer_pubkey).map_err(|e| TunnelError::Failed(format!("installer_link.installer_pubkey: {}", e.0)))?;
let preshared_key = l.psk.as_deref().map(decode_key).transpose().map_err(|e| TunnelError::Failed(format!("installer_link.psk: {}", e.0)))?;
Ok(TunnelSpec { private_key: private, listen_port: Some(l.port), peers: vec![PeerSpec { public_key: installer, preshared_key, endpoint: None, allowed_ips: vec![(LINK_INSTALLER, 32)], persistent_keepalive: None }] })
}
#[cfg(feature = "wireguard")]
pub fn vm_link(seed: &crate::seed::Seed) -> Result<VmLink, TunnelError> {
use wg_appliance_core::{addr, dataplane, TunDevice};
match reach(seed.tunnel)? {
Reach::Tunnel => {
let l = seed.installer_link.as_ref().ok_or_else(|| TunnelError::NotConfigured { kind: seed.tunnel, what: "the seed carries no [installer_link]".into() })?;
let spec = vm_link_spec(l)?;
let tun = TunDevice::create(LINK_IFACE, LINK_MTU as u32 + 40).map_err(|e| TunnelError::Failed(format!("{LINK_IFACE}: {e}")))?;
addr::assign_ipv4(LINK_IFACE, LINK_VM, LINK_PREFIX).map_err(|e| TunnelError::Failed(format!("{LINK_IFACE} {LINK_VM}/{LINK_PREFIX}: {e}")))?;
addr::link_up(LINK_IFACE).map_err(|e| TunnelError::Failed(format!("{LINK_IFACE} up: {e}")))?;
// Named: its timer lines say `wg{peer=wg-installer:installer:<fp>}`.
let handle = dataplane::spawn_named(tun, &spec, &["installer"]).map_err(|e| TunnelError::Failed(format!("{LINK_IFACE}: {e}")))?;
Ok(VmLink { handle: Some(handle), bind: LINK_VM })
}
}
}
#[cfg(test)]
mod tests {
use super::*;
/// ★ **The VM's end of the link is keyed with the seed's pre-shared key** (0.5.29, lane
/// WG-PSK-3): the installer, its one peer, carries the PSK; a seed from before 0.5.29 (no
/// PSK) raises the link as before; no Debug prints it.
#[cfg(feature = "wireguard")]
#[test]
fn the_vms_end_of_the_link_carries_the_seeds_preshared_key_and_an_old_seed_none() {
let (vm_private, _) = wg_appliance_core::keys::generate_keypair_b64();
let (_, installer_public) = wg_appliance_core::keys::generate_keypair_b64();
let mut l = crate::seed::LinkSeed { installer_pubkey: installer_public, vm_private, port: LINK_PORT, psk: None };
let psk = crate::installer::mint_wg_psk().unwrap();
l.psk = Some(psk.clone());
let spec = vm_link_spec(&l).unwrap();
assert_eq!(spec.peers.len(), 1);
assert_eq!(spec.peers[0].preshared_key, Some(wg_appliance_core::decode_key(&psk).unwrap()), "the installer peer carries the link's PSK");
assert!(!format!("{spec:?}").contains(&psk), "no Debug prints it");
l.psk = None;
assert_eq!(vm_link_spec(&l).unwrap().peers[0].preshared_key, None, "a seed from before 0.5.29: as before");
}
#[test]
fn the_path_is_the_tunnel_kinds_and_tailscale_is_refused_by_name() {
assert!(reach(TunnelKind::None).unwrap_err().to_string().starts_with("REFUSED [installer-link-none]"), "WireGuard + mutual TLS always");
assert_eq!(reach(TunnelKind::Wireguard).unwrap(), Reach::Tunnel);
assert_eq!(reach(TunnelKind::Mullvad).unwrap(), Reach::Tunnel, "mullvad is only the exit; the mesh is WireGuard");
let e = reach(TunnelKind::Tailscale).unwrap_err().to_string();
assert!(e.starts_with("REFUSED [tunnel-tailscale-unimplemented]"), "{e}");
assert_eq!(vm_bind(Reach::Tunnel), LINK_VM, "never the public interface in tunnel mode");
assert!(!crate::in_overlay(LINK_VM) && !crate::in_overlay(LINK_INSTALLER), "the link is not the mesh");
assert!(format!("{:?}", LinkKeys { installer_private: [7; 32], vm_public: [1; 32], psk: None }).contains(".."), "no key in Debug");
}
}
/// ★ estate-box3 (two-link-readers, MEASURED plain3/plain5: P5 "a hello
/// after the close: peer closed connection", T2 "the installer's own knock
/// failed", W1 "the VM closed the connection" — every one while the capture
/// follower polled the same box): a VM's link keeps ONE endpoint for its one
/// peer, the installer, and moves it to wherever the newest authenticated
/// datagram came from. Every `WgPath::new` bound a NEW UDP socket and a new
/// session under the one installer key, so two readers in one process (the
/// capture and a check, or a re-admission) stole each other's answers.
#[cfg(all(test, feature = "installer-wireguard"))]
mod shared_link_tests {
use super::*;
use boringtun::noise::{Tunn, TunnResult};
use boringtun::x25519::{PublicKey, StaticSecret};
use smoltcp::iface::{Config, Interface, SocketSet};
use smoltcp::phy::{Device, DeviceCapabilities, Medium};
use smoltcp::socket::tcp;
use smoltcp::wire::{HardwareAddress, IpAddress, IpCidr};
use std::collections::VecDeque;
use std::net::UdpSocket;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Instant;
#[derive(Default)]
struct Q {
rx: VecDeque<Vec<u8>>,
tx: VecDeque<Vec<u8>>,
}
struct Rx(Vec<u8>);
struct Tx<'a>(&'a mut VecDeque<Vec<u8>>);
impl smoltcp::phy::RxToken for Rx {
fn consume<R, F: FnOnce(&[u8]) -> R>(self, f: F) -> R {
f(&self.0)
}
}
impl smoltcp::phy::TxToken for Tx<'_> {
fn consume<R, F: FnOnce(&mut [u8]) -> R>(self, len: usize, f: F) -> R {
let mut b = vec![0u8; len];
let r = f(&mut b);
self.0.push_back(b);
r
}
}
impl Device for Q {
type RxToken<'a> = Rx;
type TxToken<'a> = Tx<'a>;
fn receive(&mut self, _t: smoltcp::time::Instant) -> Option<(Rx, Tx<'_>)> {
let p = self.rx.pop_front()?;
Some((Rx(p), Tx(&mut self.tx)))
}
fn transmit(&mut self, _t: smoltcp::time::Instant) -> Option<Tx<'_>> {
Some(Tx(&mut self.tx))
}
fn capabilities(&self) -> DeviceCapabilities {
let mut c = DeviceCapabilities::default();
c.medium = Medium::Ip;
c.max_transmission_unit = LINK_MTU;
c
}
}
/// A VM's end of the link, in user space, as WireGuard keeps it: ONE
/// peer, whose endpoint is where its newest authenticated datagram came
/// from. Its listener answers every line `x` with `ok x`.
fn fake_vm(vm: [u8; 32], installer_public: [u8; 32], stop: Arc<AtomicBool>) -> SocketAddrV4 {
let udp = UdpSocket::bind("127.0.0.1:0").unwrap();
udp.set_read_timeout(Some(Duration::from_millis(2))).unwrap();
let at = match udp.local_addr().unwrap() {
SocketAddr::V4(a) => a,
_ => unreachable!(),
};
std::thread::spawn(move || {
let mut tunn = Tunn::new(StaticSecret::from(vm), PublicKey::from(installer_public), None, None, 7, None);
let epoch = Instant::now();
let now = || smoltcp::time::Instant::from_millis(epoch.elapsed().as_millis() as i64);
let mut dev = Q::default();
let mut iface = Interface::new(Config::new(HardwareAddress::Ip), &mut dev, now());
iface.update_ip_addrs(|a| {
let _ = a.push(IpCidr::new(IpAddress::Ipv4(LINK_VM), LINK_PREFIX));
});
let mut sockets = SocketSet::new(Vec::new());
let mut listeners = Vec::new();
for _ in 0..4 {
let mut s = tcp::Socket::new(tcp::SocketBuffer::new(vec![0u8; 8192]), tcp::SocketBuffer::new(vec![0u8; 8192]));
s.listen(crate::bootstrap::PORT).unwrap();
listeners.push((sockets.add(s), Vec::<u8>::new()));
}
let mut endpoint: Option<SocketAddr> = None;
let (mut net, mut dgram) = (vec![0u8; 65536], [0u8; 2048]);
while !stop.load(Ordering::SeqCst) {
if let TunnResult::WriteToNetwork(p) = tunn.update_timers(&mut net) {
if let Some(e) = endpoint {
let _ = udp.send_to(p, e);
}
}
while let Ok((n, from)) = udp.recv_from(&mut dgram) {
let mut first = true;
loop {
let r = tunn.decapsulate(None, if first { &dgram[..n] } else { &[] }, &mut net);
first = false;
match r {
TunnResult::WriteToNetwork(p) => {
endpoint = Some(from);
let _ = udp.send_to(p, from);
}
TunnResult::WriteToTunnelV4(p, _) => {
endpoint = Some(from);
dev.rx.push_back(p.to_vec());
break;
}
TunnResult::Done => {
endpoint = Some(from);
break;
}
_ => break,
}
}
}
iface.poll(now(), &mut dev, &mut sockets);
for (h, got) in listeners.iter_mut() {
let s = sockets.get_mut::<tcp::Socket>(*h);
if s.can_recv() {
let _ = s.recv(|b| {
got.extend_from_slice(b);
(b.len(), ())
});
}
if let Some(i) = got.iter().position(|b| *b == b'\n') {
if s.can_send() {
let line: Vec<u8> = got.drain(..=i).collect();
let mut out = b"ok ".to_vec();
out.extend_from_slice(&line);
let _ = s.send_slice(&out);
s.close();
}
}
if !s.is_open() {
got.clear();
let _ = s.listen(crate::bootstrap::PORT);
}
}
iface.poll(now(), &mut dev, &mut sockets);
while let Some(p) = dev.tx.pop_front() {
if let (TunnResult::WriteToNetwork(out), Some(e)) = (tunn.encapsulate(&p, &mut net), endpoint) {
let _ = udp.send_to(out, e);
}
}
std::thread::sleep(Duration::from_millis(1));
}
});
at
}
/// The VM's end as the real box's door behaves (bootstrap::serve_*): its
/// kernel ESTABLISHES every connection at once (here: four listeners, the
/// backlog), but the door SERVES them one at a time, in order, and each
/// answer takes `serve_for` (a box that is busy). `x\n` → `ok x\n`.
fn fake_door_vm(vm: [u8; 32], installer_public: [u8; 32], stop: Arc<AtomicBool>, serve_for: Duration) -> SocketAddrV4 {
let udp = UdpSocket::bind("127.0.0.1:0").unwrap();
udp.set_read_timeout(Some(Duration::from_millis(2))).unwrap();
let at = match udp.local_addr().unwrap() {
SocketAddr::V4(a) => a,
_ => unreachable!(),
};
std::thread::spawn(move || {
let mut tunn = Tunn::new(StaticSecret::from(vm), PublicKey::from(installer_public), None, None, 9, None);
let epoch = Instant::now();
let now = || smoltcp::time::Instant::from_millis(epoch.elapsed().as_millis() as i64);
let mut dev = Q::default();
let mut iface = Interface::new(Config::new(HardwareAddress::Ip), &mut dev, now());
iface.update_ip_addrs(|a| {
let _ = a.push(IpCidr::new(IpAddress::Ipv4(LINK_VM), LINK_PREFIX));
});
let mut sockets = SocketSet::new(Vec::new());
let mut listeners = Vec::new();
for _ in 0..4 {
let mut s = tcp::Socket::new(tcp::SocketBuffer::new(vec![0u8; 8192]), tcp::SocketBuffer::new(vec![0u8; 8192]));
s.listen(crate::bootstrap::PORT).unwrap();
listeners.push(sockets.add(s));
}
// The door's queue (the order the kernel established them) and
// the one connection it is serving: what it read, when it answers.
let mut queue: VecDeque<smoltcp::iface::SocketHandle> = VecDeque::new();
let mut got: Vec<u8> = Vec::new();
let mut answer_at: Option<Instant> = None;
let mut endpoint: Option<SocketAddr> = None;
let (mut net, mut dgram) = (vec![0u8; 65536], [0u8; 2048]);
while !stop.load(Ordering::SeqCst) {
if let TunnResult::WriteToNetwork(p) = tunn.update_timers(&mut net) {
if let Some(e) = endpoint {
let _ = udp.send_to(p, e);
}
}
while let Ok((n, from)) = udp.recv_from(&mut dgram) {
let mut first = true;
loop {
let r = tunn.decapsulate(None, if first { &dgram[..n] } else { &[] }, &mut net);
first = false;
match r {
TunnResult::WriteToNetwork(p) => {
endpoint = Some(from);
let _ = udp.send_to(p, from);
}
TunnResult::WriteToTunnelV4(p, _) => {
endpoint = Some(from);
dev.rx.push_back(p.to_vec());
break;
}
TunnResult::Done => {
endpoint = Some(from);
break;
}
_ => break,
}
}
}
iface.poll(now(), &mut dev, &mut sockets);
for h in &listeners {
let s = sockets.get_mut::<tcp::Socket>(*h);
if s.state() == tcp::State::Established && !queue.contains(h) {
queue.push_back(*h);
}
}
if let Some(&h) = queue.front() {
let s = sockets.get_mut::<tcp::Socket>(h);
if s.can_recv() {
let _ = s.recv(|b| {
got.extend_from_slice(b);
(b.len(), ())
});
}
if answer_at.is_none() && got.contains(&b'\n') {
answer_at = Some(Instant::now() + serve_for);
}
if answer_at.is_some_and(|t| Instant::now() >= t) && s.can_send() {
let i = got.iter().position(|b| *b == b'\n').unwrap();
let mut out = b"ok ".to_vec();
out.extend(got.drain(..=i));
let _ = s.send_slice(&out);
s.close();
answer_at = None;
}
}
// A connection that ended (answered, or its reader went away)
// leaves the queue, and its listener listens again.
for h in &listeners {
let s = sockets.get_mut::<tcp::Socket>(*h);
if !s.is_open() {
if queue.front() == Some(h) {
got.clear();
answer_at = None;
}
queue.retain(|q| q != h);
let _ = s.listen(crate::bootstrap::PORT);
}
}
iface.poll(now(), &mut dev, &mut sockets);
while let Some(p) = dev.tx.pop_front() {
if let (TunnResult::WriteToNetwork(out), Some(e)) = (tunn.encapsulate(&p, &mut net), endpoint) {
let _ = udp.send_to(out, e);
}
}
std::thread::sleep(Duration::from_millis(1));
}
});
at
}
/// ★ estate-box W6 (`W6-two-readers-one-link`, the release gate red 2 of 3
/// on byte-identical binaries): "reader A #0: the request: nothing from
/// the VM over the link" while reader B got all four answers, both rounds
/// ending together. The box's door serves ONE connection at a time; A's
/// connection was established by the box's kernel at once and then waited
/// in its queue while the door served B — and A's read timeout was
/// running all that time. A reader's timeout is for ITS exchange: waiting
/// for the door while another reader of this process uses it is not.
#[test]
fn a_reader_queued_at_the_one_door_is_not_charged_for_another_readers_exchange() {
let (vm, k) = keys();
let installer_public = PublicKey::from(&StaticSecret::from(k.installer_private)).to_bytes();
let stop = Arc::new(AtomicBool::new(false));
// Each answer takes 3 s; each reader may wait 4.5 s for its own.
let at = fake_door_vm(vm, installer_public, stop.clone(), Duration::from_secs(3));
let timeout = Duration::from_millis(4500);
let exchange = |who: &str, s: &mut Box<dyn Wire + '_>| -> Result<String, String> {
s.write_all(format!("{who}\n").as_bytes()).map_err(|e| format!("{who}'s request: {e}"))?;
s.flush().map_err(|e| format!("{who}'s request: {e}"))?;
let mut got = Vec::new();
s.read_to_end(&mut got).map_err(|e| format!("{who}'s answer: {e}"))?;
Ok(String::from_utf8_lossy(&got).to_string())
};
// The link up first: its handshake is not what is measured.
{
let mut path = WgPath::new(at, k.clone()).unwrap();
let mut s = path.connect(timeout).expect("the first connect");
assert_eq!(exchange("warm", &mut s).as_deref(), Ok("ok warm\n"));
}
// B is connected first, so the door serves B first; A connects while B
// is still open and asks right after B (if it may connect at all
// before B is done, B asks once it has).
let (b, a) = std::thread::scope(|sc| {
let (b_up, b_is_up) = std::sync::mpsc::channel();
let (a_up, a_is_up) = std::sync::mpsc::channel::<()>();
let (kb, ex) = (&k, &exchange);
let hb = sc.spawn(move || {
let mut path = WgPath::new(at, kb.clone())?;
let mut s = path.connect(timeout)?;
let _ = b_up.send(());
let _ = a_is_up.recv_timeout(Duration::from_millis(500));
ex("b", &mut s)
});
let _ = b_is_up.recv();
let a = (|| {
let mut path = WgPath::new(at, k.clone())?;
let mut s = path.connect(timeout)?;
let _ = a_up.send(());
std::thread::sleep(Duration::from_millis(100));
exchange("a", &mut s)
})();
(hb.join().unwrap(), a)
});
stop.store(true, Ordering::SeqCst);
assert_eq!(b.as_deref(), Ok("ok b\n"), "reader B, served first");
assert_eq!(a.as_deref(), Ok("ok a\n"), "reader A lost its answer: its timeout ran while the door served reader B");
}
fn keys() -> ([u8; 32], LinkKeys) {
let (vm, installer) = ([3u8; 32], [5u8; 32]);
let vm_public = PublicKey::from(&StaticSecret::from(vm)).to_bytes();
let installer_public = PublicKey::from(&StaticSecret::from(installer)).to_bytes();
let _ = installer_public;
(vm, LinkKeys { installer_private: installer, vm_public, psk: None })
}
#[test]
fn two_readers_of_one_vm_in_one_process_both_get_their_answers() {
let (vm, k) = keys();
let installer_public = PublicKey::from(&StaticSecret::from(k.installer_private)).to_bytes();
let stop = Arc::new(AtomicBool::new(false));
let at = fake_vm(vm, installer_public, stop.clone());
let mut a = WgPath::new(at, k.clone()).unwrap();
let mut sa = a.connect(Duration::from_secs(10)).expect("reader A connects");
sa.write_all(b"reader-").unwrap();
sa.flush().unwrap();
let (got_a, got_b) = std::thread::scope(|sc| {
// Reader B — the capture's next read, a check, a re-admission — on
// its own thread, while A is open: it waits for A's turn at the
// door (the box serves one connection at a time), then gets ITS answer.
let kb = &k;
let hb = sc.spawn(move || {
let mut b = WgPath::new(at, kb.clone()).unwrap();
let mut sb = b.connect(Duration::from_secs(10)).expect("reader B connects once A is done");
sb.write_all(b"b\n").unwrap();
sb.flush().unwrap();
let mut got = Vec::new();
let _ = sb.read_to_end(&mut got);
String::from_utf8_lossy(&got).to_string()
});
std::thread::sleep(Duration::from_millis(300));
assert!(!hb.is_finished(), "reader B waits for A's turn at the door");
sa.write_all(b"a\n").unwrap();
sa.flush().unwrap();
let mut got = Vec::new();
let r = sa.read_to_end(&mut got);
drop(sa);
((String::from_utf8_lossy(&got).to_string(), r), hb.join().unwrap())
});
stop.store(true, Ordering::SeqCst);
assert_eq!(got_a.0, "ok reader-a\n", "reader A lost its answer to reader B's session: {:?}", got_a.1);
assert_eq!(got_b, "ok b\n", "reader B's answer");
}
}