use crate::bootlog::{BootLog, MAX_READ};
use crate::door::{parse_request, Request, Response};
pub const MAX_FEED_BYTES: usize = 16 * 1024 * 1024;
use crate::join::{JoinReply, JoinRequest};
use crate::mtls;
use crate::signal::Signal;
use std::collections::BTreeMap;
use std::io::{Read, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::sync::Arc;
use std::time::Duration;
pub const PORT: u16 = 8474;
pub const HELLO_PATH: &str = "/install/hello";
pub const LOG_PATH: &str = "/install/log";
pub const FEED_PATH: &str = "/install/feed";
pub const STATUS_PATH: &str = "/install/status";
pub const CLOSE_PATH: &str = "/install/close";
pub const OFFER_PATH: &str = "/install/offer";
pub const SECRET_OFFER_PATH: &str = "/install/secret-offer";
pub const REFEED_PATH: &str = "/install/refeed";
pub const CERT_PATH: &str = "/install/cert";
pub const RENEW_PATH: &str = "/install/renew";
pub const BOOTSTRAP_PATH: &str = "/install/bootstrap";
pub fn refeed_refusal(kept: Option<&JoinReply>, new: &JoinReply) -> Option<String> {
let Some(kept) = kept else {
return Some("REFUSED [refeed-never-fed] this VM was never fed; a re-feed changes what an install fed, it is not an install".into());
};
let why = if new.estate != kept.estate {
format!("the re-feed is for estate `{}`, this VM is of `{}`", new.estate, kept.estate)
} else if new.me.name != kept.me.name {
format!("the re-feed names this VM `{}`, the install named it `{}`", new.me.name, kept.me.name)
} else if new.me.overlay != kept.me.overlay {
format!("the re-feed moves `{}` to overlay {}, the install gave it {}", kept.me.name, new.me.overlay, kept.me.overlay)
} else if new.role != kept.role {
format!("the re-feed says role {:?}, the install elected {:?} — a role is a bootstrap fact", new.role, kept.role)
} else if new.cert_pem.is_some() {
"the re-feed carries a certificate; a closed VM keeps the one its install signed".to_string()
} else {
return None;
};
Some(format!("REFUSED [refeed-changes-the-vm] {why}"))
}
#[derive(Clone, Default, PartialEq, Eq)]
pub struct SecretOffer(BTreeMap<String, String>);
impl SecretOffer {
pub fn new(values: BTreeMap<String, String>) -> SecretOffer {
SecretOffer(values)
}
pub fn is_empty(&self) -> bool {
self.0.is_empty()
}
pub fn len(&self) -> usize {
self.0.len()
}
pub fn keys(&self) -> Vec<String> {
self.0.keys().cloned().collect()
}
pub fn expose(&self) -> &BTreeMap<String, String> {
&self.0
}
pub fn into_inner(self) -> BTreeMap<String, String> {
self.0
}
pub fn render(&self) -> String {
toml::to_string(&self.0).unwrap_or_default()
}
pub fn parse(text: &str) -> Result<SecretOffer, String> {
toml::from_str(text).map(SecretOffer).map_err(|_| "not a secret offer (a TOML table of strings); the body is not quoted".to_string())
}
}
impl std::fmt::Debug for SecretOffer {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "SecretOffer({:?} — values withheld)", self.keys())
}
}
pub fn fed_offer_key(vm: &str, key: &str) -> String {
format!("offer.{vm}.{key}")
}
pub fn offers_in(feed: &BTreeMap<String, String>, vm: &str) -> BTreeMap<String, String> {
let prefix = format!("offer.{vm}.");
feed.iter().filter_map(|(k, v)| k.strip_prefix(&prefix).map(|k| (k.to_string(), v.clone()))).collect()
}
pub fn render_offer(offer: &BTreeMap<String, String>) -> String {
toml::to_string(offer).unwrap_or_default()
}
pub fn parse_offer(text: &str) -> Result<BTreeMap<String, String>, String> {
toml::from_str(text).map_err(|e| format!("not an offer: {e}"))
}
pub trait Vm: Send + Sync {
fn hello(&self) -> Result<JoinRequest, String>;
fn fed(&self, reply: &JoinReply) -> Result<(), String>;
fn status(&self) -> Signal;
fn close(&self) -> Result<(), String>;
fn identity(&self) -> mtls::Issued;
fn offer(&self) -> BTreeMap<String, String> {
BTreeMap::new()
}
fn secret_offer(&self) -> SecretOffer {
SecretOffer::default()
}
fn refed(&self, _reply: &JoinReply) -> Result<(), String> {
Err("REFUSED [refeed-unsupported] this VM takes no re-feed after its close".into())
}
fn renewed(&self, _cert_pem: &str) -> Result<(), String> {
Err("REFUSED [renew-not-kept] this VM keeps no renewed certificate".into())
}
fn rebootstrapped(&self, _boot: &mtls::Issued) -> Result<(), String> {
Err("REFUSED [rebootstrap-not-kept] this VM keeps no fresh bootstrap".into())
}
fn live_running(&self) -> Result<crate::live::Running, String> {
Err("REFUSED [live-unsupported] this VM takes no live update".into())
}
fn live_stage(&self, _stage: &crate::live::Stage) -> Result<String, String> {
Err("REFUSED [live-unsupported] this VM takes no live update".into())
}
fn live_stage_declared(&self, stage: &crate::live::Stage, declared: &crate::live::Declared) -> Result<String, String> {
if declared.is_empty() {
self.live_stage(stage)
} else {
Err(crate::live::declared_unsupported())
}
}
}
pub fn installer_only(cn: &str) -> String {
format!(
"REFUSED [installer-only] this is a VM's bootstrap listener; it serves the installer's own certificate (CN `{}`) and nothing else — `{cn}` was presented",
mtls::INSTALLER_CN
)
}
fn query_from(path: &str) -> u64 {
path.split_once('?')
.map(|(_, q)| q)
.unwrap_or("")
.split('&')
.find_map(|kv| kv.strip_prefix("from="))
.and_then(|v| v.parse().ok())
.unwrap_or(0)
}
pub fn respond(vm: &dyn Vm, log: &BootLog, req: &Request) -> (Response, bool) {
let path = req.path.split('?').next().unwrap_or("");
let ok = |body: String| Response { status: 200, body };
match (req.method.as_str(), path) {
("GET", HELLO_PATH) => match vm.hello().and_then(|h| h.render().map_err(|e| e.to_string())) {
Ok(t) => (ok(t), false),
Err(e) => (Response { status: 503, body: format!("{e}\n") }, false),
},
("GET", LOG_PATH) => (ok(log.since(query_from(&req.path), MAX_READ).render()), false),
("POST", FEED_PATH) => match JoinReply::parse(&String::from_utf8_lossy(&req.body)) {
Err(e) => (Response { status: 400, body: format!("{e}\n") }, false),
Ok(reply) => match vm.fed(&reply) {
Ok(()) => (ok(format!("fed `{}`\n", reply.me.name)), false),
Err(e) => (Response { status: 503, body: format!("{e}\n") }, false),
},
},
("GET", STATUS_PATH) => (ok(vm.status().render().unwrap_or_default()), false),
("GET", OFFER_PATH) => (ok(render_offer(&vm.offer())), false),
("GET", SECRET_OFFER_PATH) => (ok(vm.secret_offer().render()), false),
("POST", CLOSE_PATH) => match vm.close() {
Ok(()) => {
log.seal_head();
(ok("closed: the bootstrap certificate is dead on this VM\n".into()), true)
}
Err(e) => (Response { status: 503, body: format!("{e}\n") }, false),
},
(_, HELLO_PATH | LOG_PATH | FEED_PATH | STATUS_PATH | OFFER_PATH | SECRET_OFFER_PATH | CLOSE_PATH) => (Response { status: 405, body: "REFUSED [install-method] wrong method for this path\n".into() }, false),
_ => (Response { status: 404, body: format!("REFUSED [install-path] this listener has {HELLO_PATH}, {LOG_PATH}, {FEED_PATH}, {STATUS_PATH}, {OFFER_PATH}, {SECRET_OFFER_PATH} and {CLOSE_PATH}\n") }, false),
}
}
#[derive(Debug, Default)]
pub struct StepsSaid(std::sync::Mutex<std::collections::BTreeSet<&'static str>>);
impl StepsSaid {
pub fn step(&self, method: &str, path: &str, status: u16) -> Option<&'static str> {
if status != 200 {
return None;
}
let path = path.split('?').next().unwrap_or("");
let line = match (method, path) {
("GET", HELLO_PATH) => "hello — this VM presented itself to the installer (its kind, tunnel key, capacity and a signing request for its own key)",
("GET", SECRET_OFFER_PATH) => "successor-offer — this VM handed the installer what its successor takes (the values are never logged)",
_ => return None,
};
self.0.lock().unwrap_or_else(|p| p.into_inner()).insert(line).then_some(line)
}
}
fn respond_saying(vm: &dyn Vm, log: &BootLog, req: &Request, said: &StepsSaid, say: &dyn Fn(Tone, &str)) -> (Response, bool) {
let (r, closed) = respond(vm, log, req);
if let Some(l) = said.step(&req.method, &req.path, r.status) {
say(Tone::Info, l);
}
(r, closed)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Tone {
Info,
Warn,
}
#[cfg(not(test))]
pub const REQUEST_READ_TIMEOUT: Duration = Duration::from_secs(10);
#[cfg(test)]
pub const REQUEST_READ_TIMEOUT: Duration = Duration::from_millis(300);
pub const REQUEST_BUDGET: Duration = Duration::from_secs(60);
#[derive(Debug)]
enum ConnError {
Gone(String),
Refused(String),
}
impl ConnError {
fn io(from: &str, e: std::io::Error) -> ConnError {
use std::io::ErrorKind::*;
let line = format!("{from}: {e}");
match e.kind() {
ConnectionReset | ConnectionAborted | BrokenPipe | UnexpectedEof | NotConnected | WouldBlock | TimedOut => ConnError::Gone(line),
_ => ConnError::Refused(line),
}
}
fn line(self) -> String {
match self {
ConnError::Gone(l) | ConnError::Refused(l) => l,
}
}
}
pub const REACHED: &str = "the installer reached this VM over its link (mutual TLS, the installer's own certificate)";
pub fn serve_one(vm: &dyn Vm, log: &BootLog, cfg: Arc<rustls::ServerConfig>, stream: TcpStream) -> Result<bool, String> {
serve_one_with(cfg, stream, &|req| respond(vm, log, req)).map_err(ConnError::line)
}
fn serve_one_with(cfg: Arc<rustls::ServerConfig>, stream: TcpStream, answer: &dyn Fn(&Request) -> (Response, bool)) -> Result<bool, ConnError> {
let from = stream.peer_addr().map(|a| a.to_string()).unwrap_or_else(|_| "?".into());
let _ = stream.set_read_timeout(Some(REQUEST_READ_TIMEOUT));
let _ = stream.set_write_timeout(Some(Duration::from_secs(20)));
let conn = rustls::ServerConnection::new(cfg).map_err(|e| ConnError::Refused(e.to_string()))?;
let mut tls = rustls::StreamOwned::new(conn, stream);
let mut bytes = Vec::new();
let mut chunk = [0u8; 4096];
let started = std::time::Instant::now();
let req = loop {
let n = match tls.read(&mut chunk) {
Ok(n) => n,
Err(e) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut | std::io::ErrorKind::Interrupted) && started.elapsed() < REQUEST_BUDGET => continue,
Err(e) => return Err(ConnError::io(&from, e)),
};
if n == 0 {
return Err(ConnError::Gone(format!("{from}: closed before a request")));
}
bytes.extend_from_slice(&chunk[..n]);
let bound = if bytes.starts_with(format!("POST {} ", crate::live::STAGE_PATH).as_bytes()) { crate::live::MAX_STAGE_BYTES } else { MAX_FEED_BYTES };
if bytes.len() > bound {
return Err(ConnError::Refused(format!("{from}: request too large")));
}
if let Some(r) = parse_request(&bytes) {
break r.map_err(ConnError::Refused)?;
}
};
let cn = mtls::peer_cn(&tls.conn).unwrap_or_default();
let (resp, closed) = if cn == mtls::INSTALLER_CN {
answer(&req)
} else {
(Response { status: 403, body: format!("{}\n", installer_only(&cn)) }, false)
};
tls.write_all(&resp.bytes()).map_err(|e| ConnError::io(&from, e))?;
tls.conn.send_close_notify();
let _ = tls.flush();
if cn != mtls::INSTALLER_CN {
return Err(ConnError::Refused(format!("{from}: {}", installer_only(&cn))));
}
Ok(closed)
}
pub fn respond_closed(log: &BootLog, req: &Request) -> Response {
let path = req.path.split('?').next().unwrap_or("");
match (req.method.as_str(), path) {
("GET", LOG_PATH) => Response { status: 200, body: log.since(query_from(&req.path), MAX_READ).render() },
(_, LOG_PATH) => Response { status: 405, body: "REFUSED [install-method] wrong method for this path\n".into() },
(_, HELLO_PATH | FEED_PATH | STATUS_PATH | OFFER_PATH | SECRET_OFFER_PATH | CLOSE_PATH) => {
Response { status: 410, body: format!("REFUSED [install-closed] this VM's install is closed; only {LOG_PATH} is left\n") }
}
_ => Response { status: 404, body: format!("REFUSED [install-path] this closed listener has {LOG_PATH} only\n") },
}
}
pub fn respond_closed_to(vm: &dyn Vm, log: &BootLog, req: &Request) -> Response {
let path = req.path.split('?').next().unwrap_or("");
match (req.method.as_str(), path) {
("POST", REFEED_PATH) => match JoinReply::parse(&String::from_utf8_lossy(&req.body)) {
Err(e) => Response { status: 400, body: format!("{e}\n") },
Ok(reply) => match vm.refed(&reply) {
Ok(()) => Response { status: 200, body: format!("re-fed `{}`\n", reply.me.name) },
Err(e) => Response { status: 409, body: format!("{e}\n") },
},
},
(_, REFEED_PATH) => Response { status: 405, body: "REFUSED [install-method] wrong method for this path\n".into() },
("GET", CERT_PATH) => Response { status: 200, body: vm.identity().cert_pem },
("POST", RENEW_PATH) => {
let pem = String::from_utf8_lossy(&req.body).to_string();
if let Some(why) = mtls::renewal_refusal(&vm.identity().cert_pem, &pem) {
return Response { status: 409, body: format!("{why}\n") };
}
match vm.renewed(&pem) {
Ok(()) => Response { status: 200, body: "renewed\n".into() },
Err(e) => Response { status: 409, body: format!("{e}\n") },
}
}
("POST", BOOTSTRAP_PATH) => {
#[derive(serde::Deserialize)]
struct Pair {
cert_pem: String,
key_pem: String,
}
let boot = match toml::from_str::<Pair>(&String::from_utf8_lossy(&req.body)) {
Ok(p) => mtls::Issued { cert_pem: p.cert_pem, key_pem: p.key_pem },
Err(e) => return Response { status: 400, body: format!("REFUSED [rebootstrap-unreadable] not a bootstrap pair: {e}\n") },
};
if let Some(why) = mtls::bootstrap_refusal(&boot) {
return Response { status: 409, body: format!("{why}\n") };
}
match vm.rebootstrapped(&boot) {
Ok(()) => Response { status: 200, body: "bootstrap kept\n".into() },
Err(e) => Response { status: 409, body: format!("{e}\n") },
}
}
("GET", crate::live::RUNNING_PATH) => match vm.live_running() {
Ok(r) => Response { status: 200, body: r.render() },
Err(e) => Response { status: 409, body: format!("{e}\n") },
},
("POST", crate::live::STAGE_PATH) => match crate::live::Stage::decode_declared(&String::from_utf8_lossy(&req.body)) {
Err(e) => Response { status: 400, body: format!("REFUSED {e}\n") },
Ok((stage, declared)) => match vm.live_stage_declared(&stage, &declared) {
Ok(said) => Response { status: 200, body: format!("{said}\n") },
Err(e) => Response { status: 409, body: format!("{e}\n") },
},
},
(_, crate::live::RUNNING_PATH | crate::live::STAGE_PATH) => Response { status: 405, body: "REFUSED [install-method] wrong method for this path\n".into() },
(_, CERT_PATH | RENEW_PATH | BOOTSTRAP_PATH) => Response { status: 405, body: "REFUSED [install-method] wrong method for this path\n".into() },
_ => respond_closed(log, req),
}
}
fn config_for(vm: &dyn Vm, ca_pem: &str, current: &mut Option<(String, Arc<rustls::ServerConfig>)>, say: &dyn Fn(&str)) -> Option<Arc<rustls::ServerConfig>> {
let id = vm.identity();
match current {
Some((pem, cfg)) if *pem == id.cert_pem => Some(cfg.clone()),
_ => match mtls::server_config(ca_pem, &id) {
Ok(cfg) => {
*current = Some((id.cert_pem.clone(), cfg.clone()));
Some(cfg)
}
Err(e) => {
say(&format!("the bootstrap listener cannot show its certificate: {e}"));
None
}
},
}
}
pub fn serve(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: TcpListener, say: &dyn Fn(&str)) {
serve_until_closed(vm, log, ca_pem, &listener, &|_, l| say(l))
}
fn serve_until_closed(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: &TcpListener, say: &dyn Fn(Tone, &str)) {
let mut current = None;
let said = StepsSaid::default();
let mut reached = false;
for stream in listener.incoming() {
let Ok(s) = stream else { continue };
let Some(cfg) = config_for(vm, ca_pem, &mut current, &|l| say(Tone::Warn, l)) else { continue };
let served = serve_one_with(cfg, s, &|req| respond_saying(vm, log, req, &said, say));
if served.is_ok() && !reached {
reached = true;
say(Tone::Info, REACHED);
}
match served {
Ok(true) => {
say(Tone::Info, "the installer closed the bootstrap listener: the bootstrap certificate is dead on this VM");
return;
}
Ok(false) => {}
Err(ConnError::Gone(e)) => say(Tone::Info, &format!("a reader of the bootstrap listener went away before its answer: {e}")),
Err(ConnError::Refused(e)) => say(Tone::Warn, &format!("a connection to the bootstrap listener was refused: {e}")),
}
}
}
pub fn serve_log_door(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: &TcpListener, say: &dyn Fn(&str)) {
serve_log_door_at(vm, log, ca_pem, listener, &|_, l| say(l))
}
pub fn serve_log_door_at(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: &TcpListener, say: &dyn Fn(Tone, &str)) {
serve_log_door_said(vm, log, ca_pem, listener, say, false)
}
fn serve_log_door_said(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: &TcpListener, say: &dyn Fn(Tone, &str), mut reached: bool) {
let mut current = None;
let mut gone_said = false;
for stream in listener.incoming() {
let Ok(s) = stream else { continue };
let Some(cfg) = config_for(vm, ca_pem, &mut current, &|l| say(Tone::Warn, l)) else { continue };
let served = serve_one_with(cfg, s, &|req| (respond_closed_to(vm, log, req), false));
if served.is_ok() && !reached {
reached = true;
say(Tone::Info, REACHED);
}
match served {
Ok(_) => {}
Err(ConnError::Gone(e)) => {
if !gone_said {
gone_said = true;
say(Tone::Info, &format!("a reader of the log door went away before its answer ({e}); the installer reads again, and this is said once"));
}
}
Err(ConnError::Refused(e)) => say(Tone::Warn, &format!("a connection to the log door was refused: {e}")),
}
}
}
pub fn serve_for_life(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: TcpListener, say: &dyn Fn(&str)) {
serve_for_life_at(vm, log, ca_pem, listener, &|_, l| say(l))
}
pub fn serve_for_life_at(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: TcpListener, say: &dyn Fn(Tone, &str)) {
serve_until_closed(vm, log, ca_pem, &listener, say);
say(Tone::Info, "the log door is open: the installer reads this VM's log over its link, for life");
serve_log_door_said(vm, log, ca_pem, &listener, say, true);
}
pub fn serve_after_close(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: TcpListener, say: &dyn Fn(&str)) {
serve_after_close_at(vm, log, ca_pem, listener, &|_, l| say(l))
}
pub fn serve_after_close_at(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: TcpListener, say: &dyn Fn(Tone, &str)) {
log.seal_head();
say(Tone::Info, "the install closed on an earlier start: only the log door is open, the installer reads this VM's log over its link");
serve_log_door_at(vm, log, ca_pem, &listener, say);
}
pub fn serve_for_life_or_after_close(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: TcpListener, closed: bool, say: &dyn Fn(&str)) {
serve_for_life_or_after_close_at(vm, log, ca_pem, listener, closed, &|_, l| say(l))
}
pub fn serve_for_life_or_after_close_at(vm: &dyn Vm, log: &BootLog, ca_pem: &str, listener: TcpListener, closed: bool, say: &dyn Fn(Tone, &str)) {
if closed {
serve_after_close_at(vm, log, ca_pem, listener, say)
} else {
serve_for_life_at(vm, log, ca_pem, listener, say)
}
}
pub fn bind(port: u16) -> std::io::Result<TcpListener> {
TcpListener::bind(SocketAddr::from(([0, 0, 0, 0], port)))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::elect::Capacity;
use crate::seed::{Kind, TunnelKind};
use std::sync::Mutex;
struct FakeVm {
fed: Mutex<Vec<JoinReply>>,
closed: Mutex<bool>,
}
impl Vm for FakeVm {
fn hello(&self) -> Result<JoinRequest, String> {
Ok(JoinRequest { estate: "e".into(), kind: Kind::Gunnar, tunnel_pubkey: "K".into(), capacity: Some(Capacity { cores: 2, ram_bytes: 1, data_bytes: 1, disk_mib_s: 1 }), csr_pem: String::new() })
}
fn fed(&self, reply: &JoinReply) -> Result<(), String> {
self.fed.lock().unwrap().push(reply.clone());
Ok(())
}
fn status(&self) -> Signal {
Signal::of("main", TunnelKind::Wireguard, &crate::pair::Machine::default())
}
fn close(&self) -> Result<(), String> {
*self.closed.lock().unwrap() = true;
Ok(())
}
fn identity(&self) -> crate::mtls::Issued {
crate::mtls::Issued { cert_pem: String::new(), key_pem: String::new() }
}
fn offer(&self) -> BTreeMap<String, String> {
BTreeMap::from([("ticket_pubkey".to_string(), "ab12".to_string())])
}
}
fn get(path: &str) -> Request {
Request { method: "GET".into(), path: path.into(), body: Vec::new() }
}
#[test]
fn the_listener_answers_hello_log_feed_status_and_close_and_nothing_else() {
let vm = FakeVm { fed: Mutex::new(Vec::new()), closed: Mutex::new(false) };
let log = BootLog::new(10);
log.push("korp-installer: pass began\ngunnar: listening");
let (r, c) = respond(&vm, &log, &get(HELLO_PATH));
assert_eq!((r.status, c), (200, false));
assert_eq!(JoinRequest::parse(&r.body).unwrap().tunnel_pubkey, "K");
let (r, _) = respond(&vm, &log, &get(&format!("{LOG_PATH}?from=1")));
let chunk = crate::bootlog::Chunk::parse(&r.body).unwrap();
assert_eq!((chunk.lines, chunk.next), (vec!["gunnar: listening".to_string()], 2));
let reply = JoinReply { estate: "e".into(), me: crate::join::Assigned { name: "main".into(), overlay: crate::OVERLAY_BOX_A }, cert_pem: None, role: None, primary_overlay: None, primary_utility: None, peers: vec![], feed: Default::default() };
let (r, _) = respond(&vm, &log, &Request { method: "POST".into(), path: FEED_PATH.into(), body: reply.render().unwrap().into_bytes() });
assert_eq!(r.status, 200, "{}", r.body);
assert_eq!(vm.fed.lock().unwrap().len(), 1);
let (r, _) = respond(&vm, &log, &get(STATUS_PATH));
assert!(Signal::parse(&r.body).is_ok(), "{}", r.body);
let (r, _) = respond(&vm, &log, &get(OFFER_PATH));
assert_eq!(parse_offer(&r.body).unwrap()["ticket_pubkey"], "ab12");
assert_eq!(parse_offer("").unwrap(), BTreeMap::new(), "an empty body is an empty offer");
let feed = BTreeMap::from([(fed_offer_key("main", "ticket_pubkey"), "ab12".to_string()), (fed_offer_key("twin", "twin_pubkey"), "ssh-ed25519 X".to_string()), ("monetize_pubkey".to_string(), "m".to_string())]);
assert_eq!(offers_in(&feed, "main"), BTreeMap::from([("ticket_pubkey".to_string(), "ab12".to_string())]));
assert_eq!(offers_in(&feed, "twin")["twin_pubkey"], "ssh-ed25519 X");
assert_eq!(respond(&vm, &log, &get(CLOSE_PATH)).0.status, 405);
assert_eq!(respond(&vm, &log, &get("/pair/join")).0.status, 404);
let (r, closed) = respond(&vm, &log, &Request { method: "POST".into(), path: CLOSE_PATH.into(), body: Vec::new() });
assert!(r.status == 200 && closed && *vm.closed.lock().unwrap());
}
#[test]
fn the_listener_says_the_hello_and_the_secret_offer_once() {
let said = StepsSaid::default();
let mut lines = Vec::new();
for (m, p, st) in [("GET", HELLO_PATH, 200), ("GET", HELLO_PATH, 200), ("GET", LOG_PATH, 200), ("GET", HELLO_PATH, 503), ("GET", SECRET_OFFER_PATH, 200), ("GET", SECRET_OFFER_PATH, 200), ("POST", FEED_PATH, 200)] {
if let Some(l) = said.step(m, p, st) {
lines.push(l);
}
}
assert_eq!(lines.len(), 2, "{lines:?}");
assert!(lines[0].starts_with("hello — "), "{lines:?}");
assert!(lines[1].starts_with("successor-offer — "), "{lines:?}");
for l in &lines {
assert_eq!(crate::redact::secret_class(l), None, "a step line the redaction withholds whole: {l}");
}
}
#[test]
fn the_secret_offer_is_its_own_path_and_never_shows_a_value() {
struct Front;
impl Vm for Front {
fn hello(&self) -> Result<JoinRequest, String> {
Err("n/a".into())
}
fn fed(&self, _r: &JoinReply) -> Result<(), String> {
Ok(())
}
fn status(&self) -> Signal {
Signal::of("front", TunnelKind::Wireguard, &crate::pair::Machine::default())
}
fn close(&self) -> Result<(), String> {
Ok(())
}
fn identity(&self) -> crate::mtls::Issued {
crate::mtls::Issued { cert_pem: String::new(), key_pem: String::new() }
}
fn secret_offer(&self) -> SecretOffer {
SecretOffer::new(BTreeMap::from([("tls.gunnar.rs.key".to_string(), "-----BEGIN PRIVATE KEY-----\nK\n".to_string())]))
}
}
let log = BootLog::new(4);
let (r, _) = respond(&Front, &log, &get(SECRET_OFFER_PATH));
assert_eq!(r.status, 200);
let s = SecretOffer::parse(&r.body).unwrap();
assert_eq!(s.expose()["tls.gunnar.rs.key"], "-----BEGIN PRIVATE KEY-----\nK\n");
assert_eq!(format!("{s:?}"), "SecretOffer([\"tls.gunnar.rs.key\"] — values withheld)");
assert!(parse_offer(&respond(&Front, &log, &get(OFFER_PATH)).0.body).unwrap().is_empty(), "the public offer does not carry it");
let box_vm = FakeVm { fed: Mutex::new(Vec::new()), closed: Mutex::new(false) };
assert!(SecretOffer::parse(&respond(&box_vm, &log, &get(SECRET_OFFER_PATH)).0.body).unwrap().is_empty(), "a box hands over nothing");
assert!(!SecretOffer::parse("= nope").unwrap_err().contains("nope"), "the error never quotes the body");
}
#[test]
fn a_closed_listener_answers_the_log_and_refuses_the_install_by_name() {
let log = BootLog::new(10);
log.push("a\nb\nc");
let r = respond_closed(&log, &get(&format!("{LOG_PATH}?from=1")));
assert_eq!(r.status, 200);
let chunk = crate::bootlog::Chunk::parse(&r.body).unwrap();
assert_eq!((chunk.lines, chunk.next), (vec!["b".to_string(), "c".to_string()], 3));
for p in [HELLO_PATH, STATUS_PATH, OFFER_PATH, SECRET_OFFER_PATH] {
let r = respond_closed(&log, &get(p));
assert!(r.status == 410 && r.body.starts_with("REFUSED [install-closed]"), "{p}: {} {}", r.status, r.body);
}
for p in [FEED_PATH, CLOSE_PATH] {
assert_eq!(respond_closed(&log, &Request { method: "POST".into(), path: p.into(), body: Vec::new() }).status, 410, "{p}");
}
assert_eq!(respond_closed(&log, &Request { method: "POST".into(), path: LOG_PATH.into(), body: Vec::new() }).status, 405);
assert_eq!(respond_closed(&log, &get("/pair/join")).status, 404);
}
fn reply_as(name: &str, role: Option<crate::Role>) -> JoinReply {
JoinReply { estate: "e".into(), me: crate::join::Assigned { name: name.into(), overlay: crate::OVERLAY_BOX_A }, cert_pem: None, role, primary_overlay: None, primary_utility: None, peers: vec![], feed: Default::default() }
}
#[test]
fn a_closed_vm_takes_a_refeed_through_its_log_door_and_refuses_it_by_default() {
let log = BootLog::new(10);
let post = |p: &str, r: &JoinReply| Request { method: "POST".into(), path: p.into(), body: r.render().unwrap().into_bytes() };
let vm = FakeVm { fed: Mutex::new(Vec::new()), closed: Mutex::new(true) };
let r = respond_closed_to(&vm, &log, &post(REFEED_PATH, &reply_as("main", Some(crate::Role::Primary))));
assert!(r.status == 409 && r.body.contains("[refeed-unsupported]"), "{} {}", r.status, r.body);
assert!(vm.fed.lock().unwrap().is_empty());
assert_eq!(respond_closed_to(&vm, &log, &get(REFEED_PATH)).status, 405);
assert_eq!(respond_closed_to(&vm, &log, &Request { method: "POST".into(), path: REFEED_PATH.into(), body: b"not toml [".to_vec() }).status, 400);
assert_eq!(respond_closed_to(&vm, &log, &post(FEED_PATH, &reply_as("main", None))).status, 410, "the install's feed stays closed");
assert_eq!(respond_closed_to(&vm, &log, &get(LOG_PATH)).status, 200, "the log door is unchanged");
}
#[test]
fn a_refeed_never_changes_what_the_vm_is() {
let kept = reply_as("main", Some(crate::Role::Primary));
assert!(refeed_refusal(None, &kept).unwrap().contains("[refeed-never-fed]"));
assert_eq!(refeed_refusal(Some(&kept), &JoinReply { primary_utility: Some([10, 0, 0, 9].into()), ..kept.clone() }), None, "addresses around it may change");
for (bad, says) in [
(JoinReply { estate: "other".into(), ..kept.clone() }, "estate"),
(reply_as("twin", Some(crate::Role::Primary)), "names this VM"),
(JoinReply { me: crate::join::Assigned { name: "main".into(), overlay: crate::OVERLAY_BOX_B }, ..kept.clone() }, "overlay"),
(reply_as("main", Some(crate::Role::Twin)), "bootstrap fact"),
(JoinReply { cert_pem: Some("C".into()), ..kept.clone() }, "certificate"),
] {
let e = refeed_refusal(Some(&kept), &bad).expect(says);
assert!(e.starts_with("REFUSED [refeed-changes-the-vm]") && e.contains(says), "{e}");
}
}
#[test]
fn the_from_query_is_read_and_a_missing_one_is_the_start() {
assert_eq!(query_from("/install/log?from=42"), 42);
assert_eq!(query_from("/install/log?x=1&from=7"), 7);
assert_eq!(query_from("/install/log"), 0);
assert!(installer_only("bootstrap").starts_with("REFUSED [installer-only]"));
}
struct QuietPath {
at: std::net::SocketAddrV4,
quiet: Duration,
}
impl crate::installer_link::Path for QuietPath {
fn connect(&mut self, timeout: Duration) -> Result<Box<dyn crate::installer_link::Wire + '_>, String> {
let s = TcpStream::connect_timeout(&SocketAddr::V4(self.at), timeout).map_err(|e| e.to_string())?;
let _ = s.set_read_timeout(Some(timeout));
std::thread::sleep(self.quiet);
Ok(Box::new(s))
}
fn gone(&mut self, _wait: Duration) -> Result<bool, String> {
Ok(false)
}
fn describe(&self) -> String {
format!("{} (quiet)", self.at)
}
}
struct DoorVm(crate::mtls::Issued);
impl Vm for DoorVm {
fn hello(&self) -> Result<JoinRequest, String> {
Err("closed".into())
}
fn fed(&self, _r: &JoinReply) -> Result<(), String> {
Ok(())
}
fn status(&self) -> Signal {
Signal::of("main", TunnelKind::Wireguard, &crate::pair::Machine::default())
}
fn close(&self) -> Result<(), String> {
Ok(())
}
fn identity(&self) -> crate::mtls::Issued {
self.0.clone()
}
}
#[test]
fn the_log_door_serves_a_slow_reader_and_says_a_reader_going_away_once_at_info() {
let ca = crate::mtls::InstallerCa::mint("gunnar.rs", Duration::from_secs(3600)).unwrap();
let boot = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
let me = ca.issue_installer(Duration::from_secs(3600)).unwrap();
let log = Arc::new(BootLog::new(10));
log.push("gunnar: serving");
let l = TcpListener::bind("127.0.0.1:0").unwrap();
let std::net::SocketAddr::V4(at) = l.local_addr().unwrap() else { panic!() };
let said: Arc<std::sync::Mutex<Vec<(Tone, String)>>> = Arc::default();
let (s2, log2, ca_pem) = (said.clone(), log.clone(), ca.cert_pem.clone());
std::thread::spawn(move || serve_after_close_at(&DoorVm(boot), &log2, &ca_pem, l, &|t, line| s2.lock().unwrap().push((t, line.to_string()))));
for _ in 0..3 {
drop(TcpStream::connect(at).unwrap());
}
let quiet = QuietPath { at, quiet: REQUEST_READ_TIMEOUT * 2 };
let chunk = crate::installer::Client::new(Box::new(quiet), &ca.cert_pem, &me, Duration::from_secs(20)).unwrap().log(0).expect("a slow reader is served, not refused");
assert_eq!(chunk.lines, vec!["gunnar: serving".to_string()]);
let stranger = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
assert!(crate::installer::Client::new(Box::new(crate::installer_link::DirectPath { at }), &ca.cert_pem, &stranger, Duration::from_secs(5)).unwrap().log(1).is_err());
std::thread::sleep(Duration::from_millis(300));
let said = said.lock().unwrap().clone();
let warns: Vec<_> = said.iter().filter(|(t, _)| *t == Tone::Warn).collect();
assert_eq!(warns.len(), 1, "only the stranger is a WARN refusal: {said:?}");
assert!(warns[0].1.contains("installer-only"), "{said:?}");
let gone: Vec<_> = said.iter().filter(|(_, l)| l.contains("went away")).collect();
assert_eq!(gone.len(), 1, "a reader going away is said once: {said:?}");
assert_eq!(gone[0].0, Tone::Info, "{said:?}");
assert!(said.iter().filter(|(_, l)| l.contains("only the log door is open")).all(|(t, _)| *t == Tone::Info), "{said:?}");
}
#[test]
fn the_listener_says_once_that_the_installer_reached_it() {
let ca = crate::mtls::InstallerCa::mint("gunnar.rs", Duration::from_secs(3600)).unwrap();
let boot = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
let me = ca.issue_installer(Duration::from_secs(3600)).unwrap();
let log = Arc::new(BootLog::new(10));
log.push("gunnar: listening");
let l = TcpListener::bind("127.0.0.1:0").unwrap();
let std::net::SocketAddr::V4(at) = l.local_addr().unwrap() else { panic!() };
let said: Arc<std::sync::Mutex<Vec<(Tone, String)>>> = Arc::default();
let (s2, log2, ca_pem) = (said.clone(), log.clone(), ca.cert_pem.clone());
std::thread::spawn(move || serve_for_life_at(&DoorVm(boot), &log2, &ca_pem, l, &|t, line| s2.lock().unwrap().push((t, line.to_string()))));
let stranger = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
assert!(crate::installer::Client::new(Box::new(crate::installer_link::DirectPath { at }), &ca.cert_pem, &stranger, Duration::from_secs(5)).unwrap().log(0).is_err());
std::thread::sleep(Duration::from_millis(200));
assert!(!said.lock().unwrap().iter().any(|(_, l)| l == REACHED), "a stranger is not the installer: {:?}", said.lock().unwrap());
for _ in 0..3 {
crate::installer::Client::new(Box::new(crate::installer_link::DirectPath { at }), &ca.cert_pem, &me, Duration::from_secs(5)).unwrap().log(0).unwrap();
}
std::thread::sleep(Duration::from_millis(200));
let said = said.lock().unwrap().clone();
let reached: Vec<_> = said.iter().filter(|(_, l)| l == REACHED).collect();
assert_eq!(reached.len(), 1, "said once: {said:?}");
assert_eq!(reached[0].0, Tone::Info);
}
#[test]
fn the_installers_close_seals_the_head_of_the_log() {
let ca = crate::mtls::InstallerCa::mint("gunnar.rs", Duration::from_secs(3600)).unwrap();
let boot = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
let me = ca.issue_installer(Duration::from_secs(3600)).unwrap();
let log = Arc::new(BootLog::with_head(4, 10_000, 50, 10_000));
log.push("install: laid\npairing: PAIRED");
let l = TcpListener::bind("127.0.0.1:0").unwrap();
let std::net::SocketAddr::V4(at) = l.local_addr().unwrap() else { panic!() };
let (log2, ca_pem) = (log.clone(), ca.cert_pem.clone());
std::thread::spawn(move || serve_for_life_at(&DoorVm(boot), &log2, &ca_pem, l, &|_, _| {}));
let client = || crate::installer::Client::new(Box::new(crate::installer_link::DirectPath { at }), &ca.cert_pem, &me, Duration::from_secs(10)).unwrap();
client().close().expect("the installer closes the install");
for i in 0..20 {
log.push(&format!("runtime {i}"));
}
assert_eq!(log.head_len(), 2, "the close sealed the head: the runtime went to the tail");
let c = client().log(0).unwrap();
assert_eq!(&c.lines[..2], &["install: laid", "pairing: PAIRED"]);
}
#[test]
fn a_restart_after_the_close_seals_the_head_before_the_log_door_opens() {
let ca = crate::mtls::InstallerCa::mint("gunnar.rs", Duration::from_secs(3600)).unwrap();
let boot = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
let log = Arc::new(BootLog::with_head(4, 10_000, 50, 10_000));
log.push("korp-start: booted");
let l = TcpListener::bind("127.0.0.1:0").unwrap();
let (log2, ca_pem) = (log.clone(), ca.cert_pem.clone());
let (tx, rx) = std::sync::mpsc::channel();
std::thread::spawn(move || serve_after_close_at(&DoorVm(boot), &log2, &ca_pem, l, &|_, _| { let _ = tx.send(()); }));
rx.recv_timeout(Duration::from_secs(10)).expect("the log door opened");
for i in 0..20 {
log.push(&format!("runtime {i}"));
}
assert_eq!(log.head_len(), 1);
}
struct LiveVm(crate::mtls::Issued, std::sync::Mutex<Vec<String>>);
impl Vm for LiveVm {
fn hello(&self) -> Result<JoinRequest, String> {
Err("closed".into())
}
fn fed(&self, _r: &JoinReply) -> Result<(), String> {
Ok(())
}
fn status(&self) -> Signal {
Signal::of("main", TunnelKind::Wireguard, &crate::pair::Machine::default())
}
fn close(&self) -> Result<(), String> {
Ok(())
}
fn identity(&self) -> crate::mtls::Issued {
self.0.clone()
}
fn live_running(&self) -> Result<crate::live::Running, String> {
Ok(crate::live::Running { base: "b".repeat(64), app: "a".repeat(64), slot: "a".into(), live_update: true })
}
fn live_stage(&self, stage: &crate::live::Stage) -> Result<String, String> {
if stage.base != "b".repeat(64) {
return Err("REFUSED [live-base-mismatch] other base".into());
}
self.1.lock().unwrap().push(stage.pin_hex.clone());
Ok(format!("staged into slot b: {} bytes", stage.archive.len()))
}
fn live_stage_declared(&self, stage: &crate::live::Stage, d: &crate::live::Declared) -> Result<String, String> {
let said = self.live_stage(stage)?;
let files = d.release.as_ref().map(|(b, _)| b.len()).unwrap_or(0);
Ok(format!("{said}; declaration {files} bytes, env {:?}", d.env))
}
}
#[test]
fn the_log_door_takes_a_live_update_from_the_installer_only() {
let ca = crate::mtls::InstallerCa::mint("gunnar.rs", Duration::from_secs(3600)).unwrap();
let boot = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
let me = ca.issue_installer(Duration::from_secs(3600)).unwrap();
let log = Arc::new(BootLog::new(10));
let l = TcpListener::bind("127.0.0.1:0").unwrap();
let std::net::SocketAddr::V4(at) = l.local_addr().unwrap() else { panic!() };
let vm = Arc::new(LiveVm(boot.clone(), Default::default()));
let (vm2, log2, ca_pem) = (vm.clone(), log.clone(), ca.cert_pem.clone());
std::thread::spawn(move || serve_after_close_at(&*vm2, &log2, &ca_pem, l, &|_, _| {}));
let client = || crate::installer::Client::new(Box::new(crate::installer_link::DirectPath { at }), &ca.cert_pem, &me, Duration::from_secs(60)).unwrap();
let r = client().live_running().unwrap();
assert_eq!((r.slot.as_str(), r.live_update), ("a", true));
let big = crate::live::Stage { pin_hex: "c".repeat(64), signature: vec![9; 72], base: "b".repeat(64), archive: vec![7u8; MAX_FEED_BYTES + 1024] };
let said = client().live_stage(&big).expect("a payload larger than a feed is taken");
assert!(said.contains("staged into slot b"), "{said}");
assert_eq!(vm.1.lock().unwrap().as_slice(), &["c".repeat(64)]);
let other = crate::live::Stage { base: "f".repeat(64), archive: vec![1; 10], ..big.clone() };
assert!(client().live_stage(&other).unwrap_err().contains("[live-base-mismatch]"));
let stranger = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
assert!(crate::installer::Client::new(Box::new(crate::installer_link::DirectPath { at }), &ca.cert_pem, &stranger, Duration::from_secs(5)).unwrap().live_running().is_err());
let l = TcpListener::bind("127.0.0.1:0").unwrap();
let std::net::SocketAddr::V4(at2) = l.local_addr().unwrap() else { panic!() };
let (log3, ca_pem) = (log.clone(), ca.cert_pem.clone());
std::thread::spawn(move || serve_after_close_at(&DoorVm(boot), &log3, &ca_pem, l, &|_, _| {}));
let mut plain = crate::installer::Client::new(Box::new(crate::installer_link::DirectPath { at: at2 }), &ca.cert_pem, &me, Duration::from_secs(10)).unwrap();
assert!(plain.live_running().unwrap_err().contains("[live-unsupported]"));
}
#[test]
fn the_log_door_hands_the_declaration_to_the_vm() {
let ca = crate::mtls::InstallerCa::mint("gunnar.rs", Duration::from_secs(3600)).unwrap();
let boot = ca.issue_bootstrap(Duration::from_secs(3600)).unwrap();
let me = ca.issue_installer(Duration::from_secs(3600)).unwrap();
let log = Arc::new(BootLog::new(10));
let l = TcpListener::bind("127.0.0.1:0").unwrap();
let std::net::SocketAddr::V4(at) = l.local_addr().unwrap() else { panic!() };
let vm = Arc::new(LiveVm(boot.clone(), Default::default()));
let (vm2, log2, ca_pem) = (vm.clone(), log.clone(), ca.cert_pem.clone());
std::thread::spawn(move || serve_after_close_at(&*vm2, &log2, &ca_pem, l, &|_, _| {}));
let mut client = crate::installer::Client::new(Box::new(crate::installer_link::DirectPath { at }), &ca.cert_pem, &me, Duration::from_secs(60)).unwrap();
let st = crate::live::Stage { pin_hex: "c".repeat(64), signature: vec![9; 72], base: "b".repeat(64), archive: vec![7u8; 1000] };
let d = crate::live::Declared { release: Some((vec![1; 33], vec![2; 72])), env: Some(vec!["korp.env.A=1".into()]) };
let said = client.live_stage_declared(&st, &d).unwrap();
assert!(said.contains("declaration 33 bytes") && said.contains("korp.env.A=1"), "{said}");
struct PlainVm(LiveVm);
impl Vm for PlainVm {
fn hello(&self) -> Result<JoinRequest, String> { self.0.hello() }
fn fed(&self, r: &JoinReply) -> Result<(), String> { self.0.fed(r) }
fn status(&self) -> Signal { self.0.status() }
fn close(&self) -> Result<(), String> { self.0.close() }
fn identity(&self) -> crate::mtls::Issued { self.0.identity() }
fn live_stage(&self, stage: &crate::live::Stage) -> Result<String, String> { self.0.live_stage(stage) }
}
let l = TcpListener::bind("127.0.0.1:0").unwrap();
let std::net::SocketAddr::V4(at2) = l.local_addr().unwrap() else { panic!() };
let (log3, ca_pem) = (log.clone(), ca.cert_pem.clone());
std::thread::spawn(move || serve_after_close_at(&PlainVm(LiveVm(boot, Default::default())), &log3, &ca_pem, l, &|_, _| {}));
let mut plain = crate::installer::Client::new(Box::new(crate::installer_link::DirectPath { at: at2 }), &ca.cert_pem, &me, Duration::from_secs(10)).unwrap();
let e = plain.live_stage_declared(&st, &d).unwrap_err();
assert!(e.contains("[live-declared-unsupported]") && crate::live::falls_to_keep(&e), "{e}");
assert!(plain.live_stage_declared(&st, &crate::live::Declared::default()).unwrap().contains("staged into slot b"));
}
}