crates/gunnar-front/src/plugin_door.rs
//! **The operator door** — vm1's mTLS door for the operator's kit, after the
//! install, when the installer is gone.
//!
//! * **mTLS only**, on vm1's public address, tcp/[`PORT`]. The door serves the
//! VM's own certificate (the one the installer's CA issued it, server name
//! `bootstrap.installer`, as the bootstrap listener does) and REQUIRES a
//! client certificate from the estate's installer CA (the seed's
//! `installer_ca_pem`) whose CN is `installer` — the operator's certificate,
//! minted with the CA and kept beside it on the operator's machine. A box's
//! certificate or the medium's bootstrap certificate is from the same CA
//! and is refused by name ([`NOT_OPERATOR`]); a certificate from any other
//! CA does not complete the handshake.
//! * **`GET /handover`**: the certificates this front serves now, for its
//! successor; and the operator door's routes ([`crate::operator_door`]):
//! the off-site backup stream, the twin's health, a fed credential's rotation.
//!
//! **No monetize behind it** (owner 13:35Z 2026-10-10, ONE monetize): the
//! plugin, vendor, cloud and signup inserts, a product server's
//! self-registration and the attach enrolment fed a monetize this front ran.
//! No estate runs one; plugins, vendors and estates are the ONE monetize's
//! (monetize.ignalina.dk), and the estate's boxes call its Fleet door.
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use axum::extract::{DefaultBodyLimit, State};
use axum::http::StatusCode;
use axum::routing::get;
use axum::{Json, Router};
use nordisk_pairing::mtls::{self, Issued};
/// The door's port, on vm1's public address (the bootstrap listener is 8474,
/// on the installer's link only).
pub const PORT: u16 = 8475;
/// The refusal a client certificate that is not the operator's gets.
pub const NOT_OPERATOR: &str = "plugin-door-not-operator";
/// The largest request taken.
pub const BODY_LIMIT: usize = 64 << 20;
/// What the door needs from the running front: the data root, and — each time
/// — the installer's last feed and the primary it names.
pub type NowFn = Arc<
dyn Fn() -> (
std::collections::BTreeMap<String, String>,
Option<crate::console::Target>,
) + Send
+ Sync,
>;
/// [`Ctx::rotate`]'s shape.
pub type RotateFn = Arc<dyn Fn(&str, &str) -> Result<bool, String> + Send + Sync>;
/// The door's context.
pub struct Ctx {
pub root: PathBuf,
pub now: NowFn,
/// **Replace one fed value** in the front's kept feed (h-admin-token): `Ok(true)`
/// when it changed. The operator door's `POST /feed/<key>`.
pub rotate: RotateFn,
}
#[derive(Clone)]
pub(crate) struct Conn {
pub(crate) ctx: Arc<Ctx>,
/// The operator's certificate (`installer`), or who else knocked (the CN
/// a refusal names). The only identity the door reads since K3: no
/// health-reader lane, no plugin registration (ONE monetize).
pub(crate) operator: Result<(), String>,
}
/// **`GET /handover` — the certificates this front serves NOW, for its
/// successor** (operator only). The installer asks the same map once, at its
/// close ([`crate::web::certs::handover`]); a certificate the front got LATER
/// (an ACME order after a web host was added, a renewal) was never handed
/// over, so the next rebuild carried the old one and asked Let's Encrypt
/// again — five certificates per week for the same names (MEASURED 2026-09-27:
/// the 3-name certificate came at 11:57, the hand-over at the close held the
/// 2-name one). The operator's kit reads this before a rebuild and carries
/// what the serving front holds. Answers `{hosts, feed}`; the key never leaves
/// except to the operator's certificate.
async fn get_handover(State(c): State<Conn>) -> (StatusCode, Json<serde_json::Value>) {
if let Err(cn) = &c.operator {
return (
StatusCode::FORBIDDEN,
Json(
serde_json::json!({"outcome": format!("REFUSED [{NOT_OPERATOR}]"), "line": format!("the client certificate is `{cn}`, not the operator's (`{}`): the hand-over goes to the operator only", mtls::INSTALLER_CN)}),
),
);
}
let (feed, hosts) = crate::web::certs::handover(&c.ctx.root, std::time::SystemTime::now());
tracing::info!(hosts = ?hosts, "plugin door: the operator read the certificate hand-over");
(
StatusCode::OK,
Json(serde_json::json!({"hosts": hosts, "feed": feed})),
)
}
/// **The operator's side of [`get_handover`]**: the status and the JSON.
pub async fn handover_remote(
addr: SocketAddr,
ca_pem: &str,
client: &Issued,
) -> Result<(u16, serde_json::Value), String> {
use tokio::io::AsyncWriteExt as _;
let cfg = mtls::client_config(ca_pem, client).map_err(|e| e.to_string())?;
let tcp = tokio::net::TcpStream::connect(addr)
.await
.map_err(|e| format!("{addr}: {e}"))?;
let mut tls = tokio_rustls::TlsConnector::from(cfg)
.connect(mtls::vm_name(), tcp)
.await
.map_err(|e| format!("{addr}: TLS: {e}"))?;
tls.write_all(
format!(
"GET /handover HTTP/1.1\r\nHost: {}\r\nConnection: close\r\n\r\n",
mtls::VM_SERVER_NAME
)
.as_bytes(),
)
.await
.map_err(|e| e.to_string())?;
tls.flush().await.map_err(|e| e.to_string())?;
read_answer(&mut tls, MAX_ANSWER, ANSWER_WAIT).await
}
/// Why the door gave a connection no answer.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Unanswered {
/// The TLS handshake did not finish within 10 s.
HandshakeTimeout,
/// The handshake was refused (no client certificate from the estate's installer CA).
Refused(String),
/// The handshake passed and serving the connection failed.
Serve(String),
}
/// **The line for a connection the door did not answer** — pure; every case is
/// said (o-ops mock 2026-09-29: a timeout and a failed serve were silent, and
/// main's self-registration saw only "no HTTP answer").
pub fn unanswered(peer: SocketAddr, why: &Unanswered) -> String {
match why {
Unanswered::HandshakeTimeout => format!("plugin door: {peer}: no TLS handshake within 10 s — the connection is dropped"),
Unanswered::Refused(e) => format!("plugin door: {peer}: handshake refused (no client certificate from the estate's installer CA): {e}"),
Unanswered::Serve(e) => format!("plugin door: {peer}: the connection ended without an answer: {e}"),
}
}
/// **Serve the door** on `listener` for the life of the process: each
/// connection is TLS 1.3 with a client certificate from `ca_pem` REQUIRED,
/// answered with `identity()` (read per connection: the VM's own certificate
/// once it has one).
pub async fn serve(
listener: tokio::net::TcpListener,
ca_pem: String,
identity: Arc<dyn Fn() -> Issued + Send + Sync>,
ctx: Arc<Ctx>,
) {
serve_refusing(listener, ca_pem, identity, ctx, None).await
}
/// [`serve`], **refusing a connection that arrives at `refused`** — the
/// front's utility address (WireGuard-only: a box reaches the door over the
/// tunnel; the operator, on the public address).
pub async fn serve_refusing(
listener: tokio::net::TcpListener,
ca_pem: String,
identity: Arc<dyn Fn() -> Issued + Send + Sync>,
ctx: Arc<Ctx>,
refused: Option<std::net::Ipv4Addr>,
) {
loop {
let (tcp, peer) = match listener.accept().await {
Ok(x) => x,
Err(e) => {
tracing::warn!(error = %e, "plugin door: accept");
tokio::time::sleep(Duration::from_millis(100)).await;
continue;
}
};
if let (Some(r), Ok(SocketAddr::V4(local))) = (refused, tcp.local_addr()) {
if *local.ip() == r {
tracing::debug!(%peer, %local, "plugin door: a connection at the utility address is refused (boxes speak over the tunnel)");
drop(tcp);
continue;
}
}
let cfg = match mtls::server_config(&ca_pem, &identity()) {
Ok(c) => c,
Err(e) => {
tracing::error!(error = %e, "plugin door: no TLS configuration");
continue;
}
};
let ctx = Arc::clone(&ctx);
tracing::debug!(%peer, "plugin door: accepted");
// A32-14, stays a task: one task per plugin-door connection.
tokio::spawn(async move {
let t0 = std::time::Instant::now();
let tls = match tokio::time::timeout(
Duration::from_secs(10),
tokio_rustls::TlsAcceptor::from(cfg).accept(tcp),
)
.await
{
Ok(Ok(s)) => s,
Ok(Err(e)) => {
tracing::info!("{}", unanswered(peer, &Unanswered::Refused(e.to_string())));
return;
}
Err(_) => {
tracing::info!("{}", unanswered(peer, &Unanswered::HandshakeTimeout));
return;
}
};
tracing::debug!(%peer, ms = t0.elapsed().as_millis() as u64, "plugin door: handshake done");
let cn = mtls::peer_cn(tls.get_ref().1).unwrap_or_default();
let operator = if cn == mtls::INSTALLER_CN {
Ok(())
} else {
Err(cn)
};
let app = Router::new()
.route("/handover", get(get_handover))
.merge(crate::operator_door::routes())
.layer(DefaultBodyLimit::max(BODY_LIMIT))
.with_state(Conn { ctx, operator });
let svc = hyper_util::service::TowerToHyperService::new(app);
let counted = Counted {
inner: tls,
read: Arc::new(std::sync::atomic::AtomicU64::new(0)),
wrote: Arc::new(std::sync::atomic::AtomicU64::new(0)),
};
let (read, wrote) = (Arc::clone(&counted.read), Arc::clone(&counted.wrote));
let builder =
hyper_util::server::conn::auto::Builder::new(hyper_util::rt::TokioExecutor::new());
let serving = builder.serve_connection(hyper_util::rt::TokioIo::new(counted), svc);
tokio::pin!(serving);
let mut beat = tokio::time::interval(Duration::from_secs(10));
beat.tick().await;
let served = loop {
tokio::select! {
r = &mut serving => break r,
_ = beat.tick() => tracing::debug!(%peer, ms = t0.elapsed().as_millis() as u64, read = read.load(std::sync::atomic::Ordering::Relaxed), wrote = wrote.load(std::sync::atomic::Ordering::Relaxed), "plugin door: connection still open"),
}
};
tracing::debug!(%peer, ms = t0.elapsed().as_millis() as u64, ok = served.is_ok(), read = read.load(std::sync::atomic::Ordering::Relaxed), wrote = wrote.load(std::sync::atomic::Ordering::Relaxed), "plugin door: connection ended");
if let Err(e) = &served {
tracing::info!("{}", unanswered(peer, &Unanswered::Serve(e.to_string())));
}
});
}
}
/// The door's TLS stream with its plaintext counted both ways (R1 evidence).
struct Counted<T> {
inner: T,
read: Arc<std::sync::atomic::AtomicU64>,
wrote: Arc<std::sync::atomic::AtomicU64>,
}
impl<T: tokio::io::AsyncRead + Unpin> tokio::io::AsyncRead for Counted<T> {
fn poll_read(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> std::task::Poll<std::io::Result<()>> {
let before = buf.filled().len();
let r = std::pin::Pin::new(&mut self.inner).poll_read(cx, buf);
self.read.fetch_add(
(buf.filled().len() - before) as u64,
std::sync::atomic::Ordering::Relaxed,
);
r
}
}
impl<T: tokio::io::AsyncWrite + Unpin> tokio::io::AsyncWrite for Counted<T> {
fn poll_write(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &[u8],
) -> std::task::Poll<std::io::Result<usize>> {
let r = std::pin::Pin::new(&mut self.inner).poll_write(cx, buf);
if let std::task::Poll::Ready(Ok(n)) = &r {
self.wrote
.fetch_add(*n as u64, std::sync::atomic::Ordering::Relaxed);
}
r
}
fn poll_flush(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::io::Result<()>> {
std::pin::Pin::new(&mut self.inner).poll_flush(cx)
}
fn poll_shutdown(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::io::Result<()>> {
std::pin::Pin::new(&mut self.inner).poll_shutdown(cx)
}
}
/// **Start the door** on `0.0.0.0:`[`PORT`] (vm1's public address).
pub async fn start(
vm: Arc<crate::vm::FrontVm>,
ca_pem: String,
utility: Option<std::net::Ipv4Addr>,
) {
let addr = SocketAddr::from((std::net::Ipv4Addr::UNSPECIFIED, PORT));
let listener = match tokio::net::TcpListener::bind(addr).await {
Ok(l) => l,
Err(e) => {
tracing::error!(error = %e, "plugin door: cannot listen on {addr} — the operator has no hand-over, backup stream or rotation");
return;
}
};
let id_vm = Arc::clone(&vm);
let identity: Arc<dyn Fn() -> Issued + Send + Sync> =
Arc::new(move || nordisk_pairing::bootstrap::Vm::identity(&*id_vm));
let now_vm = Arc::clone(&vm);
let rotate_vm = Arc::clone(&vm);
let ctx = Arc::new(Ctx {
root: vm.root().to_path_buf(),
now: Arc::new(move || {
let r = now_vm.latest();
(
r.as_ref().map(|r| r.feed.clone()).unwrap_or_default(),
r.as_ref().and_then(crate::console::target),
)
}),
rotate: Arc::new(move |k: &str, v: &str| rotate_vm.rotate(k, v)),
});
tracing::info!(%addr, "plugin door: mTLS, the operator's certificate only — GET /handover, the backup stream, the twin's health, POST /feed/<key>");
// A32-14, stays a task: the plugin door's accept loop, on the front's runtime.
tokio::spawn(serve_refusing(listener, ca_pem, identity, ctx, utility));
}
/// The most an operator-side call reads of the door's answer.
pub const MAX_ANSWER: u64 = 4 << 20;
/// How long an operator-side call waits for the door's whole answer.
pub const ANSWER_WAIT: std::time::Duration = std::time::Duration::from_secs(300);
/// **The door's HTTP answer**: its status and its JSON body — read as HTTP/1.1
/// (stinky row 16, 2026-10-10): a status line that is `HTTP/1.x NNN`, a body
/// framed by its `Content-Length`, by `chunked` coding (to its last chunk), or
/// else by the close; never more than `max` bytes of body, and all of it within
/// `wait`. A cut answer, a door that keeps talking, and a door gone silent are
/// each an error — not a guess between the outer braces, and not a hung call.
pub(crate) async fn read_answer<T: serde::de::DeserializeOwned, R: tokio::io::AsyncRead + Unpin>(
r: &mut R,
max: u64,
wait: std::time::Duration,
) -> Result<(u16, T), String> {
let (status, body) = tokio::time::timeout(wait, read_http(r, max))
.await
.map_err(|_| {
format!(
"the door gave no whole answer within {} s",
wait.as_secs_f32()
)
})??;
if body.iter().all(u8::is_ascii_whitespace) {
return Err(format!("the door answered {status} with no body"));
}
let json = serde_json::from_slice(&body).map_err(|e| {
format!(
"the door's answer does not parse: {e}: {:?}",
String::from_utf8_lossy(&body[..body.len().min(512)])
)
})?;
Ok((status, json))
}
/// The most of an answer's head (status line and headers) that is read.
const MAX_ANSWER_HEAD: usize = 64 << 10;
/// One line of the answer, at most [`MAX_ANSWER_HEAD`] bytes of it.
async fn next_line<R: tokio::io::AsyncBufRead + Unpin>(
r: &mut R,
line: &mut Vec<u8>,
) -> Result<usize, String> {
use tokio::io::{AsyncBufReadExt as _, AsyncReadExt as _};
line.clear();
r.take((MAX_ANSWER_HEAD + 1) as u64)
.read_until(b'\n', line)
.await
.map_err(|e| format!("reading the door's answer: {e}"))
}
async fn read_http<R: tokio::io::AsyncRead + Unpin>(
r: &mut R,
max: u64,
) -> Result<(u16, Vec<u8>), String> {
use tokio::io::AsyncReadExt as _;
let mut r = tokio::io::BufReader::new(r);
let mut head_len = 0usize;
let mut line = Vec::new();
let n = next_line(&mut r, &mut line).await?;
head_len += n;
let status_line = String::from_utf8_lossy(&line).trim_end().to_string();
let mut words = status_line.split(' ');
let status = match (words.next(), words.next()) {
(Some(v), Some(code)) if v.starts_with("HTTP/1.") && code.len() == 3 => {
code.parse::<u16>().ok()
}
_ => None,
}
.ok_or_else(|| {
format!(
"no HTTP answer: {:?}",
&status_line[..status_line.len().min(200)]
)
})?;
let (mut length, mut chunked) = (None::<u64>, false);
loop {
let n = next_line(&mut r, &mut line).await?;
head_len += n;
if head_len > MAX_ANSWER_HEAD {
return Err(format!(
"the door's answer head runs past {MAX_ANSWER_HEAD} bytes"
));
}
if n == 0 || !line.ends_with(b"\n") {
return Err("the door's answer was cut inside its head".into());
}
let text = String::from_utf8_lossy(&line);
let text = text.trim_end();
if text.is_empty() {
break;
}
let Some((name, value)) = text.split_once(':') else {
return Err(format!("not an HTTP header in the door's answer: {text:?}"));
};
let value = value.trim();
if name.eq_ignore_ascii_case("content-length") {
length =
Some(value.parse().map_err(|_| {
format!("the door's Content-Length is not a length: {value:?}")
})?);
} else if name.eq_ignore_ascii_case("transfer-encoding")
&& value
.to_ascii_lowercase()
.split(',')
.any(|c| c.trim() == "chunked")
{
chunked = true;
}
}
let mut body = Vec::new();
if chunked {
loop {
let n = next_line(&mut r, &mut line).await?;
if n == 0 || !line.ends_with(b"\n") {
return Err("the door's chunked answer was cut before its last chunk".into());
}
let size_text = String::from_utf8_lossy(&line);
let size_text = size_text
.trim()
.split(';')
.next()
.unwrap_or("")
.trim()
.to_string();
let size = u64::from_str_radix(&size_text, 16)
.map_err(|_| format!("not a chunk size in the door's answer: {size_text:?}"))?;
if size == 0 {
break;
}
if body.len() as u64 + size > max {
return Err(format!("the door's answer runs past {max} bytes"));
}
let at = body.len();
body.resize(at + size as usize, 0);
r.read_exact(&mut body[at..])
.await
.map_err(|e| format!("the door's chunked answer was cut inside a chunk: {e}"))?;
let mut crlf = [0u8; 2];
r.read_exact(&mut crlf)
.await
.map_err(|e| format!("the door's chunked answer was cut after a chunk: {e}"))?;
if &crlf != b"\r\n" {
return Err("a chunk of the door's answer is not ended by CRLF".into());
}
}
} else if let Some(len) = length {
if len > max {
return Err(format!("the door's answer is {len} bytes, past {max}"));
}
body.resize(len as usize, 0);
r.read_exact(&mut body)
.await
.map_err(|e| format!("the door's answer was cut short of its {len} bytes: {e}"))?;
} else {
(&mut r)
.take(max + 1)
.read_to_end(&mut body)
.await
.map_err(|e| format!("reading the door's answer: {e}"))?;
if body.len() as u64 > max {
return Err(format!("the door's answer runs past {max} bytes"));
}
}
Ok((status, body))
}
#[cfg(test)]
mod tests;