crates/installer/nordisk-twin/src/certify.rs
//! **The certification, driven by the front.** One attempt is every step, in the
//! order the source demands; `Err` names the step it stopped at, so a retrying
//! driver (and its signal) says WHERE a pairing is, not only that it is not.
//!
//! ```text
//! 1 prepare reader BecomeTwin(source) twin only; a recharge skips it
//! 2 enrol source the principal, if the product enrols (gunnar: group twin)
//! 3 request reader TwinCsr the key is born on the reader
//! 4 issue source IssueTwinCertificate the source's OWN issuer signs
//! 5 install reader InstallTwinCertificate our key, the CA signed it, pin it
//! ```
//!
//! READER is the box that will dial, SOURCE the box whose door it dials: a twin
//! and its primary, or — the same five steps, step 1 skipped — a fresh main and
//! the twin it recharges from.
use std::time::Duration;
/// Which boxes, at which control addresses.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Target {
/// The reader's control plane (`host:port`).
pub reader_ctl: String,
/// The source's control plane.
pub source_ctl: String,
/// `Some(primary as the twin is told it)` for a twin (step 1); `None` for a
/// recharge, whose reader has no role to take.
pub become_twin_of: Option<String>,
/// The principal the source issues for — the CN its door maps.
pub principal: String,
/// The reader's credential slot, as the product names it (`twin`,
/// `recharge`): which directory steps 3 and 5 use on the reader.
pub purpose: String,
}
/// What the source's issue step answers.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Issued {
pub cert_pem: String,
pub issuer_pem: String,
pub port: u16,
/// The addresses the source's door answers on, in its order of preference.
pub hosts: Vec<String>,
}
/// **The calls, one per step** — gRPC in a product, a fake in the tests.
pub trait Boxes {
/// Step 1 on the reader: take the twin role, live. Answers whatever the
/// product's step 2 needs (gunnar: the twin's ssh key line).
fn become_twin(&self, reader_ctl: &str, source: &str) -> Result<String, String>;
/// Step 2 on the source: enrol `principal`. A product whose door maps a
/// fixed CN needs nothing here.
fn enrol(&self, _source_ctl: &str, _principal: &str, _answer: &str) -> Result<(), String> {
Ok(())
}
/// Step 3 on the reader: the request for its key.
fn request(&self, reader_ctl: &str, purpose: &str) -> Result<String, String>;
/// Step 4 on the source.
fn issue(&self, source_ctl: &str, principal: &str, csr_pem: &str) -> Result<Issued, String>;
/// Step 5 on the reader: days of certificate left.
fn install(&self, reader_ctl: &str, purpose: &str, issued: &Issued) -> Result<i64, String>;
}
/// A certification that took.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Paired {
/// The first door the reader will try: `host:port`.
pub replication: String,
pub days_left: i64,
}
/// **One attempt, every step.**
pub fn once(boxes: &dyn Boxes, t: &Target) -> Result<Paired, String> {
let answer = match &t.become_twin_of {
Some(src) => boxes.become_twin(&t.reader_ctl, src).map_err(|e| format!("prepare: the reader at {} would not become a twin of {src}: {e}", t.reader_ctl))?,
None => String::new(),
};
boxes.enrol(&t.source_ctl, &t.principal, &answer).map_err(|e| format!("enrol: the source at {} would not enrol {}: {e}", t.source_ctl, t.principal))?;
let csr = boxes.request(&t.reader_ctl, &t.purpose).map_err(|e| format!("request: the reader at {} gave no signing request: {e}", t.reader_ctl))?;
let issued = boxes.issue(&t.source_ctl, &t.principal, &csr).map_err(|e| format!("issue: the source at {} would not certify {}: {e}", t.source_ctl, t.principal))?;
let days_left = boxes.install(&t.reader_ctl, &t.purpose, &issued).map_err(|e| format!("install: the reader at {} refused its certificate: {e}", t.reader_ctl))?;
let host = issued.hosts.first().cloned().unwrap_or_else(|| t.source_ctl.rsplit_once(':').map(|(h, _)| h.to_string()).unwrap_or_default());
Ok(Paired { replication: format!("{host}:{}", issued.port), days_left })
}
/// A driver's waits between attempts: 5 s doubling to 30 s. The source answers
/// Unauthenticated until its admin is in, and the reader has nothing to request
/// before it is a twin — minutes, not failures.
pub fn waits() -> impl Iterator<Item = Duration> {
std::iter::successors(Some(Duration::from_secs(5)), |d| Some((*d * 2).min(Duration::from_secs(30))))
}
/// **Attempt until it takes**, or `waits` runs out (a test's bound). Each
/// attempt's outcome goes to `said` — the driver's signal.
pub fn run(boxes: &dyn Boxes, t: &Target, waits: impl IntoIterator<Item = Duration>, said: &mut dyn FnMut(u64, &Result<Paired, String>)) -> Result<Paired, String> {
run_inner(boxes, t, waits, None, said)
}
/// **The pair the front wants certified NOW**, shared by the front with every
/// certify loop it started (Gunnar b69c4f00, `certify::Wanted`): a re-feed that
/// moves the twin or the primary hands the new pair to a new loop, and the loop
/// for the old pair stops. `None`: the front wants no pair (its feed names none).
pub type Wanted = std::sync::Arc<std::sync::Mutex<Option<Target>>>;
/// What a superseded loop's `Err` starts with ([`is_superseded`]).
pub const SUPERSEDED: &str = "superseded";
/// Did [`run_while`] stop because the front no longer wants its pair?
pub fn is_superseded(err: &str) -> bool {
err.starts_with(SUPERSEDED)
}
/// **[`run`], for as long as `wanted` names this loop's pair** (Gunnar b69c4f00).
/// A loop whose pair the front no longer wants stops, and says so in its `Err`
/// ([`SUPERSEDED`]): before an attempt, after one (its outcome then does NOT go to
/// `said`, so a product's record stays the new pair's), and before it sleeps.
pub fn run_while(boxes: &dyn Boxes, t: &Target, waits: impl IntoIterator<Item = Duration>, wanted: &std::sync::Mutex<Option<Target>>, said: &mut dyn FnMut(u64, &Result<Paired, String>)) -> Result<Paired, String> {
run_inner(boxes, t, waits, Some(wanted), said)
}
fn run_inner(boxes: &dyn Boxes, t: &Target, waits: impl IntoIterator<Item = Duration>, wanted: Option<&std::sync::Mutex<Option<Target>>>, said: &mut dyn FnMut(u64, &Result<Paired, String>)) -> Result<Paired, String> {
let still = || wanted.is_none_or(|w| w.lock().unwrap_or_else(|p| p.into_inner()).as_ref() == Some(t));
let superseded = || {
Err(format!(
"{SUPERSEDED}: the certification of the reader at {} against the source at {} stops, because the front was fed another pair (a box was laid again elsewhere)",
t.reader_ctl, t.source_ctl
))
};
let mut waits = waits.into_iter();
let mut n = 0;
loop {
if !still() {
return superseded();
}
n += 1;
let r = once(boxes, t);
if !still() {
return superseded();
}
said(n, &r);
if r.is_ok() {
return r;
}
if !still() {
return superseded();
}
match waits.next() {
Some(w) => std::thread::sleep(w),
None => return r,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::cell::RefCell;
#[derive(Default)]
struct Fake {
calls: RefCell<Vec<String>>,
refuse_issue: RefCell<u32>,
}
impl Boxes for Fake {
fn become_twin(&self, r: &str, s: &str) -> Result<String, String> {
self.calls.borrow_mut().push(format!("become {r} of {s}"));
Ok(String::new())
}
fn request(&self, r: &str, p: &str) -> Result<String, String> {
self.calls.borrow_mut().push(format!("request {r} {p}"));
Ok("CSR".into())
}
fn issue(&self, s: &str, who: &str, csr: &str) -> Result<Issued, String> {
self.calls.borrow_mut().push(format!("issue {s} {who} {csr}"));
let mut n = self.refuse_issue.borrow_mut();
if *n > 0 {
*n -= 1;
return Err("no replication door recorded".into());
}
Ok(Issued { cert_pem: "C".into(), issuer_pem: "CA".into(), port: 50052, hosts: vec!["10.13.8.2".into()] })
}
fn install(&self, r: &str, p: &str, i: &Issued) -> Result<i64, String> {
self.calls.borrow_mut().push(format!("install {r} {p} {}", i.port));
Ok(364)
}
}
#[test]
fn a_twin_is_certified_in_the_order_the_source_demands_and_a_recharge_skips_the_role() {
let fake = Fake::default();
let t = Target { reader_ctl: "10.13.8.3:50051".into(), source_ctl: "10.13.8.2:50051".into(), become_twin_of: Some("10.13.8.2:50051".into()), principal: "holger-twin".into(), purpose: "twin".into() };
let p = once(&fake, &t).unwrap();
assert_eq!(p, Paired { replication: "10.13.8.2:50052".into(), days_left: 364 });
assert_eq!(*fake.calls.borrow(), ["become 10.13.8.3:50051 of 10.13.8.2:50051", "request 10.13.8.3:50051 twin", "issue 10.13.8.2:50051 holger-twin CSR", "install 10.13.8.3:50051 twin 50052"]);
let fake = Fake::default();
let r = Target { become_twin_of: None, principal: "holger-recharge".into(), purpose: "recharge".into(), ..t.clone() };
once(&fake, &r).unwrap();
assert!(fake.calls.borrow()[0].starts_with("request"), "a recharge takes no role");
}
#[test]
fn a_refused_step_is_named_and_retried() {
let fake = Fake { refuse_issue: RefCell::new(1), ..Default::default() };
let t = Target { reader_ctl: "r:1".into(), source_ctl: "s:1".into(), become_twin_of: None, principal: "p".into(), purpose: "twin".into() };
let mut seen = Vec::new();
let p = run(&fake, &t, [Duration::ZERO], &mut |n, r| seen.push((n, r.as_ref().err().cloned()))).unwrap();
assert_eq!(p.days_left, 364);
assert!(seen[0].1.as_deref().unwrap().starts_with("issue: the source at s:1"), "{seen:?}");
assert_eq!(seen.len(), 2);
assert_eq!(waits().take(5).map(|d| d.as_secs()).collect::<Vec<_>>(), [5, 10, 20, 30, 30]);
}
/// ★ Ported from Gunnar b69c4f00 (`a_loop_whose_pair_the_front_no_longer_wants_stops_dialling_it`;
/// estate-box3 life1 R2: the loop for the DEAD twin dialled it for 10+ min between the new
/// loop's tries). The re-feed lands while the old loop's attempt is under way: that attempt's
/// outcome is not said, and no further attempt is made.
#[test]
fn a_loop_whose_pair_the_front_no_longer_wants_stops_dialling_it() {
struct DeadTwin<'a> {
calls: RefCell<Vec<String>>,
wanted: &'a std::sync::Mutex<Option<Target>>,
moved_to: Target,
}
impl Boxes for DeadTwin<'_> {
fn become_twin(&self, r: &str, _: &str) -> Result<String, String> {
self.calls.borrow_mut().push(format!("become {r}"));
// The installer re-feeds the front while this dial times out.
*self.wanted.lock().unwrap() = Some(self.moved_to.clone());
Err("could not be reached (Unavailable): tcp connect error".into())
}
fn request(&self, _: &str, _: &str) -> Result<String, String> {
unreachable!()
}
fn issue(&self, _: &str, _: &str, _: &str) -> Result<Issued, String> {
unreachable!()
}
fn install(&self, _: &str, _: &str, _: &Issued) -> Result<i64, String> {
unreachable!()
}
}
let at = |reader: &str| Target { reader_ctl: reader.into(), source_ctl: "10.6.0.14:50051".into(), become_twin_of: Some("10.6.0.14:50051".into()), principal: "holger-twin".into(), purpose: "twin".into() };
let (old, new) = (at("10.6.0.13:50051"), at("10.6.0.15:50051"));
let wanted = std::sync::Mutex::new(Some(old.clone()));
let fake = DeadTwin { calls: RefCell::new(Vec::new()), wanted: &wanted, moved_to: new };
let mut said = 0;
let e = run_while(&fake, &old, std::iter::repeat_n(Duration::ZERO, 5), &wanted, &mut |_, _| said += 1).unwrap_err();
assert!(is_superseded(&e) && e.contains("10.6.0.13:50051"), "the old pair is never certified, and says why: {e}");
assert_eq!(*fake.calls.borrow(), ["become 10.6.0.13:50051"], "one attempt at the dead twin, then the loop stops");
assert_eq!(said, 0, "the superseded attempt is not said: the record is the new pair's");
// A loop the front wants no pair for at all stops before it dials.
let none = std::sync::Mutex::new(None);
let fake = Fake::default();
assert!(is_superseded(&run_while(&fake, &old, [], &none, &mut |_, _| {}).unwrap_err()));
assert!(fake.calls.borrow().is_empty(), "nothing dialled");
// A loop whose pair is still wanted goes on as before.
let fake = Fake::default();
let wanted = std::sync::Mutex::new(Some(old.clone()));
assert_eq!(run_while(&fake, &old, [], &wanted, &mut |_, _| {}).unwrap().days_left, 364);
}
}