holger-estate-front/src/front_report.rs
//! ★ **The front tells main what it serves** (lane HOL-FLOW-FRONT, Henrik's Q1 of
//! HOL-FLOW-SMART): its doors, every name it serves and where each goes, and its console's
//! address, with a proof of its own (`holger_downstream_rules::front`). Main keeps the last one
//! and its flow unfolds the front to it; nothing the front did not report is drawn.
//!
//! | piece | what it is |
//! |---|---|
//! | [`report_of`] | pure: the report from the feed (the same `WebConfig` and `RepoRoute` the web face serves by) |
//! | [`key_for`] | the proof's key for one box: x25519 of the front's console key and the box's offered one, then `front::key_of` (its own context, never the bearer's) |
//! | [`send`] | one `POST /-/front/report` to main's HTTP door; an OLDER main answers 404, which is [`Answer::OlderMain`]: said once, never an error |
//! | [`start`] | a task for the front's life: a report when what it serves changed, and every [`RESEND`] so main sees it alive |
use std::collections::BTreeMap;
use std::net::SocketAddrV4;
use std::sync::Arc;
use std::time::Duration;
use holger_downstream_rules::front::{self, FrontDoor, FrontReport, FrontRoute, Sent};
use holger_downstream_rules::via;
/// How often the front looks whether what it serves changed.
pub const LOOK_EVERY: Duration = Duration::from_secs(30);
/// How often an unchanged report is sent again (main says "reported N ago").
pub const RESEND: Duration = Duration::from_secs(600);
/// How long the front waits before it asks an older main again.
pub const OLDER_MAIN_AGAIN: Duration = Duration::from_secs(3600);
/// Pure: **the front's report**, from its feed: its doors (`https`, `http`, and the
/// downstreams' interface when it is open), the console's name, every name the repository
/// route carries to main, monetize's names, and the downstreams' sync door when open.
pub fn report_of(
feed: &BTreeMap<String, String>,
estate: &str,
me: &str,
at_unix: i64,
downstreams_open: bool,
console: bool,
) -> FrontReport {
let web = estate_front::web::WebConfig::from(estate, feed);
let route = estate_front::route::RepoRoute::of(
feed,
&web,
crate::PRODUCT,
Some((crate::HTTP_PORT, crate::CONTROL_PORT)),
crate::CARRIED_GRPC,
);
let mut doors = vec![
FrontDoor {
name: "https".into(),
listen: format!("tcp/{}", estate_front::web::HTTPS_PORT),
what: "the web face: TLS with the estate's certificate".into(),
},
FrontDoor {
name: "http".into(),
listen: format!("tcp/{}", estate_front::web::HTTP_PORT),
what: "the certificate's challenges, and the way to https".into(),
},
];
let mut routes = Vec::new();
let console_host = web.canonical();
if console {
routes.push(FrontRoute {
host: console_host.clone(),
door: "https".into(),
to: front::TO_CONSOLE.into(),
what: "holger-ui, and its pass-through to main's two doors".into(),
});
}
let services = route
.grpc_services
.iter()
.map(|s| format!("{s}*"))
.collect::<Vec<_>>()
.join(", ");
for h in &route.hosts {
routes.push(FrontRoute {
host: h.clone(),
door: "https".into(),
to: front::TO_MAIN.into(),
what: format!(
"main http :{}, gRPC :{} ({services})",
route.http_port, route.grpc_port
),
});
}
for h in &web.callback_only {
routes.push(FrontRoute {
host: h.clone(),
door: "https".into(),
to: front::TO_MONETIZE.into(),
what: "a payment vendor's callbacks".into(),
});
}
if let Some(h) = &web.control_host {
routes.push(FrontRoute {
host: h.clone(),
door: "https".into(),
to: front::TO_MONETIZE.into(),
what: "the ONE monetize's control door".into(),
});
}
if downstreams_open {
doors.push(FrontDoor {
name: "downstreams".into(),
listen: format!("udp/{}", via::HUB_UDP_PORT),
what: format!("the WireGuard downstreams' interface ({})", via::HUB_IFACE),
});
routes.push(FrontRoute {
host: format!("{}:{}", via::HUB_ADDR, via::DOOR_PORT),
door: "downstreams".into(),
to: front::TO_MAIN.into(),
what: "the sync door: main's sync and read calls only".into(),
});
}
FrontReport {
front: me.to_string(),
at_unix,
doors,
routes,
console_url: if console {
front::console_url(&format!("https://{console_host}")).unwrap_or_default()
} else {
String::new()
},
}
}
/// **The proof's key for one box**: x25519 of the front's console key and the box's offered
/// console key (base64), then `front::key_of`.
pub fn key_for(my_private: &[u8; 32], box_public_b64: &str) -> Result<[u8; 32], String> {
let theirs = wg_appliance_core::keys::decode_key(box_public_b64.trim())
.map_err(|e| format!("the box's offered console key is not a 32-byte base64 key: {e:?}"))?;
let shared = boringtun::x25519::StaticSecret::from(*my_private)
.diffie_hellman(&boringtun::x25519::PublicKey::from(theirs));
if !shared.was_contributory() {
return Err("the box's offered console key is a low-order point".into());
}
Ok(front::key_of(shared.as_bytes()))
}
/// What main answered.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Answer {
/// Kept.
Kept,
/// An older main: it has no such door (404). Not an error: it draws the front closed.
OlderMain,
/// Refused, with its status and words.
Refused(u16, String),
}
/// **One report, sent** to main's HTTP door `at` (HTTP/1.1, `Connection: close`).
pub async fn send(at: SocketAddrV4, sent: &Sent) -> Result<Answer, String> {
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
let body = serde_json::to_vec(sent).map_err(|e| e.to_string())?;
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 head = format!(
"POST {} HTTP/1.1\r\nHost: {at}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
front::PATH,
body.len()
);
s.write_all(head.as_bytes())
.await
.map_err(|e| e.to_string())?;
s.write_all(&body).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: u16 = head
.split_whitespace()
.nth(1)
.and_then(|s| s.parse().ok())
.ok_or_else(|| format!("{at}: not an HTTP answer"))?;
Ok(match status {
200..=299 => Answer::Kept,
404 => Answer::OlderMain,
_ => Answer::Refused(
status,
String::from_utf8_lossy(&raw[split + 4..])
.trim()
.to_string(),
),
})
}
/// What the loop needs now: the feed, this front's name, main's HTTP door, main's offered
/// console key; `None` while the front is not fed its primary.
pub type Now =
Arc<dyn Fn() -> Option<(BTreeMap<String, String>, String, SocketAddrV4, String)> + Send + Sync>;
/// Is the downstreams' interface open now (lane HOL-WGDS: it opens at the first WireGuard
/// downstream main lists)?
pub type Open = Arc<dyn Fn() -> bool + Send + Sync>;
/// Set once the downstreams' interface is open (`main.rs`, `open_downstreams`).
pub static DOWNSTREAMS_OPEN: std::sync::atomic::AtomicBool =
std::sync::atomic::AtomicBool::new(false);
/// Pure: **main's offered console key**, out of the feed (`offer.<primary>.console_box_pubkey`).
pub fn main_key(reply: &nordisk_pairing::join::JoinReply) -> Option<String> {
let p = reply
.peers
.iter()
.find(|p| p.role == nordisk_pairing::Role::Primary)?;
reply
.feed
.get(&nordisk_pairing::bootstrap::fed_offer_key(
&p.name,
crate::console_key::OFFER_BOX_PUBKEY,
))
.map(|v| v.trim().to_string())
.filter(|v| !v.is_empty())
}
/// ★ **Report to main for the life of the process**: every [`LOOK_EVERY`], when what the
/// front serves changed or [`RESEND`] passed. An older main is asked again after
/// [`OLDER_MAIN_AGAIN`]; a refusal is said when it changes.
pub fn start(estate: String, now: Now, downstreams_open: Open, console: bool) {
// A32-14, stays a task: the front's report to main, for the front's life.
tokio::spawn(async move {
let mut last: Option<(FrontReport, std::time::Instant)> = None;
let mut older_until: Option<std::time::Instant> = None;
let mut said = String::new();
loop {
tokio::time::sleep(LOOK_EVERY).await;
if older_until.is_some_and(|t| std::time::Instant::now() < t) {
continue;
}
let Some((feed, me, at, box_key)) = now() else {
continue;
};
let Some(private) = crate::console_key::adopted() else {
continue;
};
let at_unix = estate_clock::now_unix_ms().div_euclid(1000);
let report = report_of(&feed, &estate, &me, at_unix, downstreams_open(), console);
let same = last.as_ref().is_some_and(|(r, when)| {
FrontReport {
at_unix: r.at_unix,
..report.clone()
} == *r
&& when.elapsed() < RESEND
});
if same {
continue;
}
let key = match key_for(&private, &box_key) {
Ok(k) => k,
Err(e) => {
if e != said {
tracing::warn!(error = %e, "front: no report to main: no key for it");
said = e;
}
continue;
}
};
let sent = Sent {
proof: front::proof(&key, &report),
report: report.clone(),
};
match send(at, &sent).await {
Ok(Answer::Kept) => {
if last.is_none() || !said.is_empty() {
tracing::info!(%at, doors = report.doors.len(), routes = report.routes.len(), "front: main keeps this front's report of its doors and routes");
}
said.clear();
older_until = None;
last = Some((report, std::time::Instant::now()));
}
Ok(Answer::OlderMain) => {
tracing::info!(%at, "front: main is an older holger: it takes no front report (its flow draws this front closed); asked again in an hour");
older_until = Some(std::time::Instant::now() + OLDER_MAIN_AGAIN);
}
Ok(Answer::Refused(status, words)) => {
let e = format!("HTTP {status}: {words}");
if e != said {
tracing::warn!(%at, refused = %e, "front: main refused this front's report");
said = e;
}
}
Err(e) => {
if e != said {
tracing::warn!(%at, error = %e, "front: the report to main could not be sent");
said = e;
}
}
}
}
});
}
#[cfg(test)]
mod tests {
use super::*;
fn feed() -> BTreeMap<String, String> {
[
("web_hosts", "acme.example,www.acme.example,repo.acme.example,pay.acme.example,monetize.acme.example"),
("repo_hosts", "repo.acme.example"),
("web_callback_only", "pay.acme.example"),
("monetize_control_host", "monetize.acme.example"),
]
.into_iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect()
}
fn route<'a>(r: &'a FrontReport, host: &str) -> &'a FrontRoute {
r.routes
.iter()
.find(|x| x.host == host)
.unwrap_or_else(|| panic!("a route for {host}: {:?}", r.routes))
}
/// ★ **The front reports what it serves, from the feed it serves by**: its two web doors,
/// the console at the canonical name, the repository's name carried to main with the
/// ports and the gRPC services the route carries, monetize's names, and the downstreams'
/// door and sync route only once that interface is open. Nothing secret.
#[test]
fn the_front_reports_its_doors_and_where_each_name_goes() {
let r = report_of(&feed(), "acme", "vm1", 42, false, true);
assert_eq!((r.front.as_str(), r.at_unix), ("vm1", 42));
assert_eq!(
r.doors
.iter()
.map(|d| (d.name.as_str(), d.listen.as_str()))
.collect::<Vec<_>>(),
vec![("https", "tcp/443"), ("http", "tcp/80")]
);
assert_eq!(route(&r, "www.acme.example").to, front::TO_CONSOLE);
let repo = route(&r, "repo.acme.example");
assert_eq!(repo.to, front::TO_MAIN);
assert!(
repo.what.contains(":8443")
&& repo.what.contains(":50051")
&& repo.what.contains("holger.*"),
"{}",
repo.what
);
assert_eq!(route(&r, "pay.acme.example").to, front::TO_MONETIZE);
assert_eq!(route(&r, "monetize.acme.example").to, front::TO_MONETIZE);
assert!(
r.routes.iter().all(|x| x.host != "acme.example"),
"the bare name is a 301 to www, no route"
);
assert_eq!(r.console_url, "https://www.acme.example");
let open = report_of(&feed(), "acme", "vm1", 42, true, true);
assert!(open
.doors
.iter()
.any(|d| d.name == "downstreams" && d.listen == "udp/51830"));
let sync = route(&open, "10.46.0.1:50070");
assert_eq!(
(sync.door.as_str(), sync.to.as_str()),
("downstreams", front::TO_MAIN)
);
let none = report_of(&feed(), "acme", "vm1", 42, false, false);
assert!(
none.routes.iter().all(|x| x.to != front::TO_CONSOLE) && none.console_url.is_empty(),
"no console built in, none reported"
);
let text = front::text(&r);
for secret in ["BEGIN", "bearer", "token", "key"] {
assert!(
!text.to_lowercase().contains(&secret.to_lowercase()),
"{secret} in {text}"
);
}
}
/// ★ **Both ends derive one key**: the front from its console key and main's offered one,
/// main from its own and the front's; a report the front proves under it holds at main.
#[test]
fn the_front_and_main_derive_the_same_key_and_the_proof_holds() {
let (front_private, main_private) = ([0x11u8; 32], [0x22u8; 32]);
let pub_of = |k: &[u8; 32]| {
wg_appliance_core::keys::encode_key(
&wg_appliance_core::keys::public_from_private_bytes(k),
)
};
let at_front = key_for(&front_private, &pub_of(&main_private)).unwrap();
let at_main = key_for(&main_private, &pub_of(&front_private)).unwrap();
assert_eq!(at_front, at_main);
let r = report_of(&feed(), "acme", "vm1", 1_000, false, true);
assert_eq!(
front::check(&at_main, &r, &front::proof(&at_front, &r), 1_010),
Ok(())
);
assert!(key_for(&front_private, "short").is_err());
}
/// One answer from a stand-in main on loopback.
async fn answering(status: &'static str) -> SocketAddrV4 {
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let at = match l.local_addr().unwrap() {
std::net::SocketAddr::V4(a) => a,
_ => unreachable!(),
};
tokio::spawn(async move {
let (mut s, _) = l.accept().await.unwrap();
let mut buf = vec![0u8; 65536];
let _ = s.read(&mut buf).await;
let _ = s
.write_all(
format!(
"HTTP/1.1 {status}\r\nContent-Length: 4\r\nConnection: close\r\n\r\nnope"
)
.as_bytes(),
)
.await;
});
at
}
/// ★ **An older main ignores the report safely**: it has no such door and answers 404,
/// which is no error (the front says so and asks again later); a refusal keeps its words.
#[tokio::test]
async fn an_older_main_answers_404_and_that_is_no_error() {
let sent = Sent::default();
assert_eq!(
send(answering("404 Not Found").await, &sent).await,
Ok(Answer::OlderMain)
);
assert_eq!(
send(answering("200 OK").await, &sent).await,
Ok(Answer::Kept)
);
assert_eq!(
send(answering("403 Forbidden").await, &sent).await,
Ok(Answer::Refused(403, "nope".into()))
);
}
}