holger-estate-front/src/downstreams.rs
//! **The front's downstream interface** (lane HOL-WGDS, owner 2026-10-27): a WireGuard
//! interface of its own for the downstreams declared `via: Wireguard(key: ..)`, apart from
//! the estate's inner tunnel (`holger-hub`, the relay to main and the twin).
//!
//! ```text
//! downstream lab-2 ──udp/51830──► front: HubRelay (isolated) ── FilterPort ──► holger-ds 10.46.0.1/24
//! │ tcp/50070 only
//! ▼
//! the sync door (call rule: sync + reads only)
//! │ over wg0
//! ▼
//! main 10.44.0.2:8443 (HTTP) / :50051 (gRPC)
//! ```
//!
//! | piece | from |
//! |---|---|
//! | the relay, one UDP port, one key, a `/32` per downstream, refusal of an unlisted key | edda `wg-appliance-core` [`HubRelay`], isolated ([`HubRelay::isolate_peers`]): a downstream reaches the front only, never another downstream |
//! | the peers | main's live `Wireguard` downstreams (`GET /-/sync/downstreams/wg`, the console's bearer), relisted with [`HubRelay::relist_named_peer_with_psk`] and dropped with [`HubRelay::forget_peer`] at the next look after a revocation |
//! | the packet rule | holger-downstream `via::inbound` / `via::outbound` in [`FilterPort`]: TCP to the sync door and its answers, nothing else reaches the front's kernel (not the console's 443, the clock, the egress door, the installer's listener) |
//! | the call rule | holger-downstream `via::door_allows` in [`door`]: the mirror's reads, its sync report, the sync manifest and ack; never an admin, console, twin or replication call |
//!
//! The front never dials a downstream: it answers on the door, and the relay answers the
//! handshakes of the keys main lists.
use std::collections::BTreeMap;
use std::io;
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock};
use std::time::Duration;
use holger_downstream_rules::via;
use serde::Deserialize;
use wg_appliance_core::hubrelay::{HubPort, HubRelay};
/// Where the front keeps its downstream interface's key, under the data root (never the
/// estate tunnel's `wg/`).
pub const KEY_REL: &str = "wg-downstreams";
/// How often main's list of WireGuard downstreams is looked at.
pub const POLL: Duration = Duration::from_secs(5);
/// The offer keys (server-lib `sync::downstreams::OFFER_HUB_*`): main keeps them for its invites.
pub const OFFER_HUB_KEY: &str = "downstream_hub_key";
pub const OFFER_HUB_ENDPOINT: &str = "downstream_hub_endpoint";
/// One WireGuard downstream main lists (server-lib `sync::downstreams::RelayPeer`): its key,
/// its address and the link's pre-shared key (lane WG-PSK; none for an enrolment from before
/// it). Its `Debug` never prints the PSK.
#[derive(Clone, PartialEq, Eq, Deserialize)]
pub struct WgPeer {
pub name: String,
pub key: String,
pub address: Ipv4Addr,
#[serde(default)]
pub psk: Option<String>,
}
impl std::fmt::Debug for WgPeer {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("WgPeer")
.field("name", &self.name)
.field("key", &self.key)
.field("address", &self.address)
.field("psk", &self.psk.as_ref().map(|_| ".."))
.finish()
}
}
/// **The hub port, filtered** — a downstream peer's packet reaches the front's kernel only
/// when it is TCP to the sync door; the front's packet reaches a downstream only when it is
/// the door's answer. Everything else is dropped here and counted.
pub struct FilterPort {
inner: Arc<dyn HubPort>,
/// Packets from a downstream the rule refused.
pub dropped_in: AtomicU64,
/// Packets toward a downstream the rule refused.
pub dropped_out: AtomicU64,
}
impl FilterPort {
pub fn new(inner: Arc<dyn HubPort>) -> Arc<FilterPort> {
Arc::new(FilterPort {
inner,
dropped_in: AtomicU64::new(0),
dropped_out: AtomicU64::new(0),
})
}
}
impl HubPort for FilterPort {
fn deliver(&self, packet: &[u8]) {
match via::inbound(packet) {
Ok(()) => self.inner.deliver(packet),
Err(why) => {
self.dropped_in.fetch_add(1, Ordering::Relaxed);
tracing::debug!(
why,
"downstreams: a downstream's packet is dropped at the hub port"
);
}
}
}
fn next(&self, buf: &mut [u8], within: Duration) -> io::Result<Option<usize>> {
let deadline = std::time::Instant::now() + within;
loop {
let left = deadline.saturating_duration_since(std::time::Instant::now());
match self.inner.next(buf, left)? {
None => return Ok(None),
Some(n) => match via::outbound(&buf[..n]) {
Ok(()) => return Ok(Some(n)),
Err(_) => {
// IPv6 router solicitations of a fresh tun, an ICMP the kernel would
// send: nothing but the door's answer leaves toward a downstream.
self.dropped_out.fetch_add(1, Ordering::Relaxed);
if left.is_zero() {
return Ok(None);
}
}
},
}
}
}
}
/// **The front's downstream interface**: the relay and who is listed on it.
pub struct DownstreamHub {
pub relay: HubRelay,
pub port: Arc<FilterPort>,
/// Listed now: address → (name, key).
listed: RwLock<BTreeMap<Ipv4Addr, (String, String)>>,
/// The pre-shared key each address is listed with (memory only, as main's list is).
psks: RwLock<BTreeMap<Ipv4Addr, Option<String>>>,
}
impl DownstreamHub {
/// The relay on `udp` (0 = any; [`via::HUB_UDP_PORT`] on the front) with `private`, its
/// own end on `port` (the `holger-ds` tun on the front, memory in a test), isolated.
pub fn start(udp: u16, private: [u8; 32], port: Arc<dyn HubPort>) -> io::Result<DownstreamHub> {
let port = FilterPort::new(port);
let relay = HubRelay::spawn_with_port(udp, private, via::HUB_ADDR, port.clone())?;
relay.isolate_peers();
Ok(DownstreamHub {
relay,
port,
listed: RwLock::new(BTreeMap::new()),
psks: RwLock::new(BTreeMap::new()),
})
}
/// The name of the downstream at `ip`, when one is listed there.
pub fn name_at(&self, ip: Ipv4Addr) -> Option<String> {
self.listed
.read()
.unwrap_or_else(|e| e.into_inner())
.get(&ip)
.map(|(n, _)| n.clone())
}
/// ★ **List exactly `peers`**: a new or re-keyed downstream is (re)listed at its address,
/// one main no longer lists (revoked) is forgotten — its key is refused from now on and
/// nothing is routed to its address. Answers the lines to say.
pub fn apply(&self, peers: &[WgPeer]) -> Vec<String> {
let mut said = Vec::new();
let mut listed = self.listed.write().unwrap_or_else(|e| e.into_inner());
for p in peers {
if !via::on_hub_net(p.address) || p.address == via::HUB_ADDR {
said.push(format!(
"downstream '{}' at {} is not on the downstreams' subnet: not listed",
p.name, p.address
));
continue;
}
let Ok(key) = via::wg_key(&p.key) else {
said.push(format!(
"downstream '{}': its key is not a WireGuard key: not listed",
p.name
));
continue;
};
// The link's pre-shared key (lane WG-PSK), as main lists it. A blank or broken one
// is no key: the downstream is not listed (never a link without its key, a silent
// downgrade). None: an enrolment from before the lane, listed as before.
let psk_b64 = p.psk.as_deref().map(str::trim).map(str::to_string);
let psk = match psk_b64.as_deref().map(wg_appliance_core::decode_key) {
None => None,
Some(Ok(k)) => Some(k),
Some(Err(_)) => {
said.push(format!("downstream '{}' at {}: main listed a pre-shared key that is not one: not listed", p.name, p.address));
continue;
}
};
let mut psks = self.psks.write().unwrap_or_else(|e| e.into_inner());
if listed
.get(&p.address)
.is_some_and(|(n, k)| n == &p.name && k == &p.key)
&& psks.get(&p.address) == Some(&psk_b64)
{
continue;
}
match self
.relay
.relist_named_peer_with_psk(key, psk, p.address, &p.name)
{
Ok(_) => {
listed.insert(p.address, (p.name.clone(), p.key.clone()));
psks.insert(p.address, psk_b64.clone());
let with = if psk.is_some() {
" with its pre-shared key"
} else {
" WITHOUT a pre-shared key (enrolled before WG-PSK: invite it again under a new key to add one)"
};
said.push(format!(
"downstream '{}' listed at {}{with}",
p.name, p.address
));
}
Err(e) => said.push(format!("downstream '{}' at {}: {e}", p.name, p.address)),
}
}
let psks_now = self.psks.read().unwrap_or_else(|e| e.into_inner()).clone();
let gone: Vec<Ipv4Addr> = listed
.iter()
.filter(|(a, (n, k))| {
!peers.iter().any(|p| {
p.address == **a
&& &p.name == n
&& &p.key == k
&& psks_now.get(*a)
== Some(&p.psk.as_deref().map(str::trim).map(str::to_string))
})
})
.map(|(a, _)| *a)
.collect();
for a in gone {
if let Some((n, _)) = listed.remove(&a) {
self.psks
.write()
.unwrap_or_else(|e| e.into_inner())
.remove(&a);
if !listed.contains_key(&a) {
self.relay.forget_peer(a);
}
said.push(format!("downstream '{n}' at {a} is no longer listed by main (revoked): its key is refused from now on"));
}
}
said
}
}
impl DownstreamHub {
/// ★ **The relay's counters, said** (the front's estate relay says its own every few
/// minutes, lane D12): who is listed, each one's handshakes and packets, and what was
/// refused or dropped. A WireGuard downstream that never completes a handshake is read
/// here: `listed 0` (main's list never reached the relay), or `refused N` (a key the
/// relay does not list, or an initiation sealed to another key than this hub's).
pub fn counters_line(&self) -> String {
counters_line(
&self.relay.verdict(),
&self.listed.read().unwrap_or_else(|e| e.into_inner()),
&self.port,
)
}
}
/// Pure: [`DownstreamHub::counters_line`] of a verdict, the listed names and the port's drops.
pub fn counters_line(
v: &wg_appliance_core::hubrelay::RelayVerdict,
listed: &BTreeMap<Ipv4Addr, (String, String)>,
port: &FilterPort,
) -> String {
let peers: Vec<String> = v
.peers
.iter()
.filter(|(a, _, _)| listed.contains_key(a))
.map(|(a, _, c)| {
format!(
"{} {a}: handshakes {} · rx {} pkts · tx {} pkts · keepalives {}",
listed.get(a).map(|(n, _)| n.as_str()).unwrap_or("?"),
c.handshakes,
c.rx_packets,
c.tx_packets,
c.keepalives
)
})
.collect();
format!(
"listed {}{}{} · refused {} · to-door {} · dropped at the door's port {} in, {} out · isolated {} · no-endpoint {}",
listed.len(),
if peers.is_empty() { "" } else { ": " },
peers.join(" | "),
v.refused,
v.to_hub,
port.dropped_in.load(Ordering::Relaxed),
port.dropped_out.load(Ordering::Relaxed),
v.dropped_isolated,
v.dropped_no_endpoint,
)
}
/// The front's downstream key (minted once, kept under [`KEY_REL`]).
pub fn identity(root: &Path) -> Result<wg_appliance_core::ceremony::Identity, String> {
wg_appliance_core::ceremony::load_or_mint_identity(&root.join(KEY_REL))
.map_err(|e| format!("the front's downstream key: {e}"))
}
/// Pure: **what the front offers about its downstream interface**: its public key and
/// `<public address>:<port>`. Nothing without a public address.
pub fn offer(public_b64: &str, public: Option<Ipv4Addr>) -> BTreeMap<String, String> {
match public {
Some(ip) if !public_b64.is_empty() => BTreeMap::from([
(OFFER_HUB_KEY.to_string(), public_b64.to_string()),
(
OFFER_HUB_ENDPOINT.to_string(),
SocketAddrV4::new(ip, via::HUB_UDP_PORT).to_string(),
),
]),
_ => BTreeMap::new(),
}
}
/// Pure: **main's list of WireGuard downstreams**, from `GET /-/sync/downstreams/wg`.
pub fn parse_peers(body: &[u8]) -> Result<Vec<WgPeer>, String> {
serde_json::from_slice(body)
.map_err(|e| format!("main's WireGuard downstreams are not readable: {e}"))
}
/// **Open the `holger-ds` interface** ([`via::HUB_IFACE`], `10.46.0.1/24`) — needs
/// `CAP_NET_ADMIN` (the front is PID 1 of its medium).
pub fn open_tun() -> Result<Arc<dyn HubPort>, String> {
let (_, prefix) = via::HUB_NET;
let tun = wg_appliance_core::tun::TunDevice::create(via::HUB_IFACE, 1420)
.map_err(|e| format!("{}: {e} (CAP_NET_ADMIN?)", via::HUB_IFACE))?;
wg_appliance_core::addr::assign_ipv4(via::HUB_IFACE, via::HUB_ADDR, prefix)
.map_err(|e| format!("{} {}/{prefix}: {e}", via::HUB_IFACE, via::HUB_ADDR))?;
wg_appliance_core::addr::link_up(via::HUB_IFACE)
.map_err(|e| format!("{} up: {e}", via::HUB_IFACE))?;
Ok(Arc::new(wg_appliance_core::hubrelay::TunPort(tun)))
}
/// One GET of main's HTTP door with the console's bearer (HTTP/1.1, `Connection: close`).
async fn get(at: SocketAddrV4, path: &str, bearer: Option<&str>) -> Result<Vec<u8>, String> {
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
let mut s = tokio::time::timeout(Duration::from_secs(5), tokio::net::TcpStream::connect(at))
.await
.map_err(|_| format!("{at}: no answer"))?
.map_err(|e| format!("{at}: {e}"))?;
let auth = bearer
.map(|b| format!("Authorization: Bearer {b}\r\n"))
.unwrap_or_default();
s.write_all(
format!("GET {path} HTTP/1.1\r\nHost: {at}\r\n{auth}Connection: close\r\n\r\n").as_bytes(),
)
.await
.map_err(|e| e.to_string())?;
let mut raw = Vec::new();
tokio::time::timeout(Duration::from_secs(10), s.read_to_end(&mut raw))
.await
.map_err(|_| format!("{at}: the answer did not end"))?
.map_err(|e| e.to_string())?;
let split = raw
.windows(4)
.position(|w| w == b"\r\n\r\n")
.ok_or_else(|| format!("{at}: not an HTTP answer"))?;
let head = String::from_utf8_lossy(&raw[..split]).to_string();
let status = head.split_whitespace().nth(1).unwrap_or("");
let body = raw[split + 4..].to_vec();
if status != "200" {
return Err(format!(
"{at}{path}: HTTP {status}: {}",
String::from_utf8_lossy(&body).trim()
));
}
// A chunked answer (hyper's default for a computed body) is unchunked here.
if head
.to_ascii_lowercase()
.contains("transfer-encoding: chunked")
{
return unchunk(&body);
}
Ok(body)
}
fn unchunk(mut b: &[u8]) -> Result<Vec<u8>, String> {
let mut out = Vec::new();
loop {
let Some(eol) = b.windows(2).position(|w| w == b"\r\n") else {
return Err("a bad chunk".into());
};
let n = usize::from_str_radix(
String::from_utf8_lossy(&b[..eol])
.trim()
.split(';')
.next()
.unwrap_or(""),
16,
)
.map_err(|_| "a bad chunk size".to_string())?;
b = &b[eol + 2..];
if n == 0 {
return Ok(out);
}
if b.len() < n + 2 {
return Err("a short chunk".into());
}
out.extend_from_slice(&b[..n]);
b = &b[n + 2..];
}
}
/// What the look at main needs: main's HTTP door and the console's bearer, now.
pub type MainNow = Arc<dyn Fn() -> Option<(SocketAddrV4, Option<String>)> + Send + Sync>;
/// **How the downstreams' interface is opened when it is first needed** (RA5-28): `None` when it
/// could not be (said by the opener).
pub type HubStart = Arc<dyn Fn() -> Option<Arc<DownstreamHub>> + Send + Sync>;
/// ★ **Follow main's list for the life of the process**: every [`POLL`], the peers main
/// lists are the peers the relay answers. A look that fails keeps the list as it is.
pub fn follow(hub: Arc<DownstreamHub>, main: MainNow) {
follow_lazily(Arc::new(move || Some(Arc::clone(&hub))), main, POLL);
}
/// ★ **[`follow`], with the relay opened only once main lists a WireGuard downstream**
/// (RA5-28): a front whose main declares none opens no udp/51830 at all. Once opened it stays
/// open and follows the list (an emptied list revokes every peer). Every `poll`.
pub fn follow_lazily(start: HubStart, main: MainNow, poll: Duration) {
// A32-14, stays a task: the downstream interface follows main's list, for the front's life.
tokio::spawn(async move {
let mut last_err = String::new();
let mut read_once = false;
let mut hub: Option<Arc<DownstreamHub>> = None;
let mut counters = crate::role::CountersSay::new(crate::role::COUNTERS_EVERY);
loop {
if main().is_none() && !read_once {
if let Some(l) = counters.due(std::time::Instant::now(), || "main is not known yet (the front is not fed its primary): no look at its list".to_string()) {
tracing::info!("downstreams: {l}");
}
}
if let Some((at, bearer)) = main() {
if bearer.is_none() && last_err.is_empty() && !read_once {
tracing::warn!(%at, "downstreams: main's list is asked with NO console bearer (main has not offered its console key yet)");
}
match get(at, "/-/sync/downstreams/wg", bearer.as_deref())
.await
.and_then(|b| parse_peers(&b))
{
Ok(peers) => {
if !read_once || !last_err.is_empty() {
tracing::info!(%at, listed = peers.len(), "downstreams: main's list of WireGuard downstreams is read");
}
read_once = true;
last_err.clear();
if hub.is_none() && !peers.is_empty() {
hub = start();
}
if let Some(hub) = &hub {
for l in hub.apply(&peers) {
tracing::info!("downstreams: {l}");
}
}
}
Err(e) if e != last_err => {
tracing::warn!(error = %e, "downstreams: main's list of WireGuard downstreams could not be read; the relay keeps the peers it has");
last_err = e;
}
Err(_) => {}
}
}
if let (true, Some(hub)) = (read_once, &hub) {
if let Some(l) = counters.due(std::time::Instant::now(), || hub.counters_line()) {
tracing::info!("downstreams: relay counters: {l}");
}
}
tokio::time::sleep(poll).await;
}
});
}
/// Main's two doors now, `(http, grpc)` (its overlay address on an estate); `None` until fed.
pub type Targets = Arc<dyn Fn() -> Option<(SocketAddrV4, SocketAddrV4)> + Send + Sync>;
/// The door's refusal: 403 for HTTP, gRPC `PERMISSION_DENIED` (7) for gRPC.
fn refuse(grpc: bool, why: &str) -> axum::response::Response {
use axum::http::{header, HeaderValue, StatusCode};
use axum::response::IntoResponse as _;
if grpc {
let mut r = (StatusCode::OK, [(header::CONTENT_TYPE, "application/grpc")]).into_response();
r.headers_mut()
.insert("grpc-status", HeaderValue::from_static("7"));
let msg: String = why
.chars()
.map(|c| {
if c.is_ascii_alphanumeric() || "-_.[]():".contains(c) {
c.to_string()
} else {
format!("%{:02X}", c as u32 & 0xff)
}
})
.collect();
if let Ok(v) = HeaderValue::from_str(&msg) {
r.headers_mut().insert("grpc-message", v);
}
r
} else {
(StatusCode::FORBIDDEN, format!("{why}\n")).into_response()
}
}
/// ★ **The sync door's router**: every call `via::door_allows` refuses is answered here
/// (403, or gRPC `PERMISSION_DENIED`) and never reaches main; the rest is carried to main's
/// HTTP door (`http`) or gRPC door (`grpc`), named by the downstream it came in as
/// ([`via::DOWNSTREAM_HEADER`], the caller's own stripped). `name_of` names the caller by
/// its address on the downstream interface; an unknown caller is refused. Serve it with
/// `into_make_service_with_connect_info::<SocketAddr>()`.
pub fn door(
main: Targets,
name_of: Arc<dyn Fn(Ipv4Addr) -> Option<String> + Send + Sync>,
) -> axum::Router {
use axum::extract::ConnectInfo;
use axum::http::header;
axum::Router::new().fallback(move |ConnectInfo(from): ConnectInfo<SocketAddr>, req: axum::extract::Request| {
let (name_of, main) = (Arc::clone(&name_of), Arc::clone(&main));
async move {
let is_grpc = req.headers().get(header::CONTENT_TYPE).and_then(|v| v.to_str().ok()).is_some_and(|v| v.starts_with("application/grpc"));
let path = req.uri().path().to_string();
let method = req.method().as_str().to_string();
let who = match from.ip() {
std::net::IpAddr::V4(ip) => name_of(ip),
std::net::IpAddr::V6(_) => None,
};
let Some(who) = who else {
tracing::warn!(target: "holger_front::audit", %from, %path, "downstreams: a call from an address no downstream is listed at is refused");
return refuse(is_grpc, &format!("REFUSED [downstream-door] {from} is no listed downstream"));
};
let pq = req.uri().path_and_query().map(|p| p.as_str().to_string()).unwrap_or_else(|| path.clone());
if let Err(why) = via::door_allows(&method, &pq, is_grpc) {
tracing::warn!(target: "holger_front::audit", downstream = %who, %method, %path, "downstreams: refused at the sync door: {why}");
return refuse(is_grpc, &format!("REFUSED [downstream-door] {why}: a downstream reaches main's sync door only"));
}
let Some((http, grpc)) = main() else {
return refuse(is_grpc, "main is not known yet: the front has not been fed its primary");
};
let answer = carry(req, if is_grpc { grpc } else { http }, is_grpc, &who).await;
tracing::info!(target: "holger_front::audit", downstream = %who, %method, %path, status = answer.status().as_u16(), "downstreams: carried to main's sync door");
answer
}
})
}
/// **Serve the door on `bind`** ([`via::HUB_ADDR`]:[`via::DOOR_PORT`] on the front).
pub async fn serve_door(bind: SocketAddr, router: axum::Router) -> io::Result<SocketAddr> {
let l = tokio::net::TcpListener::bind(bind).await?;
let at = l.local_addr()?;
// TCP_NODELAY on every connection (HOL-PERF-3): axum 0.8 leaves Nagle on.
let l = axum::serve::ListenerExt::tap_io(l, |tcp: &mut tokio::net::TcpStream| {
let _ = tcp.set_nodelay(true);
});
// A32-14, stays a task: the sync door's accept loop, for the front's life.
tokio::spawn(async move {
if let Err(e) = axum::serve(
l,
router.into_make_service_with_connect_info::<SocketAddr>(),
)
.await
{
tracing::error!(error = %e, "downstreams: the sync door stopped");
}
});
Ok(at)
}
type Client = hyper_util::client::legacy::Client<
hyper_util::client::legacy::connect::HttpConnector,
axum::body::Body,
>;
fn clients() -> &'static (Client, Client) {
static C: std::sync::OnceLock<(Client, Client)> = std::sync::OnceLock::new();
C.get_or_init(|| {
let exec = hyper_util::rt::TokioExecutor::new();
let h1 = hyper_util::client::legacy::Client::builder(exec.clone())
.pool_idle_timeout(Duration::from_secs(30))
.pool_timer(hyper_util::rt::TokioTimer::new())
.build_http();
let h2 = hyper_util::client::legacy::Client::builder(exec)
.http2_only(true)
.timer(hyper_util::rt::TokioTimer::new())
.pool_timer(hyper_util::rt::TokioTimer::new())
.pool_idle_timeout(Duration::from_secs(30))
.http2_keep_alive_interval(Duration::from_secs(10))
.http2_keep_alive_timeout(Duration::from_secs(5))
.http2_keep_alive_while_idle(true)
.build_http();
(h1, h2)
})
}
async fn carry(
req: axum::extract::Request,
to: SocketAddrV4,
grpc: bool,
who: &str,
) -> axum::response::Response {
use axum::http::{header, HeaderValue, StatusCode};
use axum::response::IntoResponse as _;
let (mut parts, body) = req.into_parts();
let pq = parts
.uri
.path_and_query()
.map(|p| p.as_str())
.unwrap_or("/")
.to_string();
let Ok(uri) = format!("http://{to}{pq}").parse::<axum::http::Uri>() else {
return (StatusCode::BAD_REQUEST, "the path cannot be carried\n").into_response();
};
for h in [
"connection",
"keep-alive",
"proxy-connection",
"transfer-encoding",
"trailer",
"upgrade",
via::DOWNSTREAM_HEADER,
] {
parts.headers.remove(h);
}
if let Ok(v) = HeaderValue::from_str(who) {
parts.headers.insert(via::DOWNSTREAM_HEADER, v);
}
if !grpc {
parts.headers.insert(
header::HOST,
HeaderValue::from_str(&to.to_string()).unwrap_or(HeaderValue::from_static("main")),
);
}
parts.uri = uri;
parts.version = if grpc {
axum::http::Version::HTTP_2
} else {
axum::http::Version::HTTP_11
};
let (h1, h2) = clients();
let req = axum::http::Request::from_parts(parts, body);
match if grpc {
h2.request(req).await
} else {
h1.request(req).await
} {
Ok(resp) => {
let (parts, body) = resp.into_parts();
axum::response::Response::from_parts(parts, axum::body::Body::new(body))
}
Err(e) => {
tracing::warn!(%to, error = %e, "downstreams: main's sync door did not answer");
(StatusCode::BAD_GATEWAY, "main's sync door did not answer\n").into_response()
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use wg_appliance_core::hubrelay::MemPort;
/// ★ **RA5-28: the downstreams' relay opens only once main lists a WireGuard downstream.**
/// Every front opened its WireGuard relay on public udp/51830 at boot, downstream or none:
/// handshake parsing nobody asked for. Now main's list is read first; the relay is opened
/// at the first WireGuard downstream it names, once.
#[tokio::test(flavor = "multi_thread")]
async fn the_downstreams_relay_opens_only_once_main_lists_a_wireguard_downstream() {
use std::sync::atomic::AtomicUsize;
let listed = Arc::new(std::sync::Mutex::new("[]".to_string()));
let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let std::net::SocketAddr::V4(at) = l.local_addr().unwrap() else {
unreachable!()
};
let body = Arc::clone(&listed);
tokio::spawn(async move {
while let Ok((mut s, _)) = l.accept().await {
let b = body.lock().unwrap().clone();
tokio::spawn(async move {
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
let mut buf = [0u8; 2048];
let _ = s.read(&mut buf).await;
let _ = s.write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{b}", b.len()).as_bytes()).await;
});
}
});
let started = Arc::new(AtomicUsize::new(0));
let count = Arc::clone(&started);
let start: HubStart = Arc::new(move || {
count.fetch_add(1, Ordering::SeqCst);
DownstreamHub::start(0, [7; 32], MemPort::new())
.ok()
.map(Arc::new)
});
follow_lazily(
start,
Arc::new(move || Some((at, None))),
Duration::from_millis(50),
);
tokio::time::sleep(Duration::from_millis(500)).await;
assert_eq!(
started.load(Ordering::SeqCst),
0,
"main lists no WireGuard downstream: the relay stays closed"
);
*listed.lock().unwrap() = r#"[{"name":"lab","key":"AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=","address":"10.46.0.2"}]"#.to_string();
tokio::time::sleep(Duration::from_millis(500)).await;
assert_eq!(
started.load(Ordering::SeqCst),
1,
"opened once, at the first WireGuard downstream"
);
}
/// ★ **The downstreams' subnet is no other link's** (fast holger RED 2026-10-28, lane
/// OX-TRAIT2): `holger-ds` took `10.45.0.1/24`, and the installer's own WireGuard link to
/// every VM (nordisk-pairing `installer_link`) is `10.45.0.1 -> 10.45.0.2/30`. On the front
/// the installer's address was then a LOCAL one: its pairing feed never came back, the front
/// was never fed and the install timed out with "front: no pair status yet".
#[test]
fn the_downstream_hub_shares_no_subnet_with_the_installer_link_or_the_estate() {
use nordisk_pairing::installer_link::{LINK_INSTALLER, LINK_PREFIX, LINK_VM};
let net = |a: std::net::Ipv4Addr, p: u8| u32::from(a) & (u32::MAX << (32 - u32::from(p)));
let (hub, hub_prefix) = via::HUB_NET;
let overlaps = |a: std::net::Ipv4Addr, p: u8| {
let shorter = p.min(hub_prefix);
net(a, shorter) == net(hub, shorter)
};
assert!(!overlaps(LINK_INSTALLER, LINK_PREFIX) && !overlaps(LINK_VM, LINK_PREFIX), "holger-ds {hub}/{hub_prefix} overlaps the installer link {LINK_INSTALLER}/{LINK_PREFIX}");
assert!(
!overlaps(std::net::Ipv4Addr::new(10, 44, 0, 0), 24),
"holger-ds {hub}/{hub_prefix} overlaps the estate tunnel 10.44.0.0/24"
);
assert_eq!(
net(via::HUB_ADDR, hub_prefix),
net(hub, hub_prefix),
"the front's address is on its own subnet"
);
}
fn tcp(src: [u8; 4], dst: [u8; 4], sport: u16, dport: u16) -> Vec<u8> {
let mut p = vec![0u8; 40];
p[0] = 0x45;
p[9] = 6;
p[12..16].copy_from_slice(&src);
p[16..20].copy_from_slice(&dst);
p[20..22].copy_from_slice(&sport.to_be_bytes());
p[22..24].copy_from_slice(&dport.to_be_bytes());
p
}
/// ★ The filtered port lets the sync door's traffic through, both ways, and nothing else.
#[test]
fn the_filtered_port_carries_the_sync_door_and_nothing_else() {
let mem = MemPort::new();
let f = FilterPort::new(mem.clone());
let to_door = tcp([10, 46, 0, 2], [10, 46, 0, 1], 40000, via::DOOR_PORT);
f.deliver(&to_door);
assert_eq!(
mem.delivered(Duration::from_millis(200)).as_deref(),
Some(&to_door[..])
);
f.deliver(&tcp([10, 46, 0, 2], [10, 46, 0, 1], 40000, 443));
assert_eq!(
mem.delivered(Duration::from_millis(100)),
None,
"the console's 443 is never delivered"
);
assert_eq!(f.dropped_in.load(Ordering::Relaxed), 1);
mem.inject(&tcp([10, 46, 0, 1], [10, 46, 0, 2], 443, 40000));
mem.inject(&tcp([10, 46, 0, 1], [10, 46, 0, 2], via::DOOR_PORT, 40000));
let mut buf = [0u8; 128];
let n = f
.next(&mut buf, Duration::from_millis(500))
.unwrap()
.expect("the door's answer leaves");
assert_eq!(
&buf[20..22],
&via::DOOR_PORT.to_be_bytes(),
"and only it: {:?}",
&buf[..n]
);
assert_eq!(f.dropped_out.load(Ordering::Relaxed), 1);
}
/// ★ **`holger-ds` on a REAL TUN** (needs `/dev/net/tun` and `CAP_NET_ADMIN`, a privileged
/// container; skipped by name otherwise unless `HOLGER_WG_TUN_REQUIRED=1`): its own
/// interface beside the estate's, its own subnet (`10.46.0.1/24`), MTU 1420, up, and the
/// kernel routes the downstreams' subnet into it — never the estate's `10.44.0.0/24`.
#[test]
fn the_downstreams_interface_opens_on_a_real_tun_apart_from_the_estates() {
let required =
std::env::var(crate::wg_only::TUN_REQUIRED_ENV).is_ok_and(|v| v.trim() == "1");
let port = match open_tun() {
Ok(p) => p,
Err(e) if !required && crate::wg_only::is_runner_limit(&e) => {
eprintln!(
"SKIPPED (set {}=1 to require the real TUN): {e}",
crate::wg_only::TUN_REQUIRED_ENV
);
return;
}
Err(e) => panic!("open_tun on this runner: {e}"),
};
let sys = |f: &str| {
std::fs::read_to_string(format!("/sys/class/net/{}/{f}", via::HUB_IFACE))
.unwrap_or_else(|e| panic!("{f}: {e}"))
.trim()
.to_string()
};
assert_eq!(sys("mtu"), "1420");
let sock =
std::net::UdpSocket::bind((via::HUB_ADDR, 0)).expect("10.46.0.1 is on holger-ds");
let to = SocketAddrV4::new(Ipv4Addr::new(10, 46, 0, 2), 9);
let deadline = std::time::Instant::now() + Duration::from_secs(5);
let mut buf = [0u8; 2048];
let mut seen = false;
while std::time::Instant::now() < deadline && !seen {
sock.send_to(b"holger-ds probe", to).unwrap();
while let Ok(Some(n)) = port.next(&mut buf, Duration::from_millis(300)) {
if n >= 20 && buf[0] >> 4 == 4 && buf[16..20] == [10, 46, 0, 2] {
seen = true;
break;
}
}
}
assert!(seen, "the downstreams' subnet is not routed into holger-ds");
assert_ne!(
via::HUB_IFACE,
crate::wg_only::HUB_IFACE,
"never the estate's interface"
);
}
#[test]
fn the_front_offers_its_downstream_interface_only_with_a_public_address() {
assert!(offer("KEY", None).is_empty());
let o = offer("KEY", Some(Ipv4Addr::new(198, 51, 100, 4)));
assert_eq!(
(o[OFFER_HUB_KEY].as_str(), o[OFFER_HUB_ENDPOINT].as_str()),
("KEY", "198.51.100.4:51830")
);
}
/// ★ The relay's counters name who is listed and what was refused (a downstream that
/// never completes a handshake is read in the front's log, lane OX-WGDS4).
#[test]
fn the_relay_counters_name_the_listed_downstreams_and_the_refusals() {
let hub = DownstreamHub::start(0, [7u8; 32], MemPort::new()).unwrap();
assert!(
hub.counters_line().starts_with("listed 0 · refused 0"),
"{}",
hub.counters_line()
);
let (_, public) = wg_appliance_core::keys::generate_keypair();
let said = hub.apply(&[WgPeer {
name: "lab-2".into(),
key: wg_appliance_core::keys::encode_key(&public),
address: Ipv4Addr::new(10, 46, 0, 2),
psk: None,
}]);
assert_eq!(said, vec!["downstream 'lab-2' listed at 10.46.0.2 WITHOUT a pre-shared key (enrolled before WG-PSK: invite it again under a new key to add one)".to_string()]);
let line = hub.counters_line();
assert!(
line.starts_with("listed 1: lab-2 10.46.0.2: handshakes 0"),
"{line}"
);
// A datagram from nobody listed is refused, and said.
let s = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
s.send_to(&[1u8; 148], ("127.0.0.1", hub.relay.port()))
.unwrap();
let t0 = std::time::Instant::now();
while !hub.counters_line().contains("refused 1") && t0.elapsed() < Duration::from_secs(5) {
std::thread::sleep(Duration::from_millis(20));
}
assert!(
hub.counters_line().contains("refused 1"),
"{}",
hub.counters_line()
);
}
#[test]
fn mains_list_is_read_and_a_chunked_answer_unchunked() {
let body = br#"[{"name":"lab-2","key":"AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=","address":"10.46.0.2"}]"#;
let p = parse_peers(body).unwrap();
assert_eq!(
(p[0].name.as_str(), p[0].address),
("lab-2", Ipv4Addr::new(10, 46, 0, 2))
);
assert!(parse_peers(b"nope").is_err());
assert_eq!(
unchunk(b"5\r\nhello\r\n6\r\n world\r\n0\r\n\r\n").unwrap(),
b"hello world"
);
assert!(unchunk(b"zz\r\n").is_err());
}
}