crates/installer/nordisk-narrate/src/sender.rs
//! **The box's side** (feature `dev-narrate` ONLY): answer a knock with one
//! sealed WHO, and once an AUTHENTICATED hello names an address, stream the
//! boot log there — from the line the hello asks for, then live.
//!
//! The state machine ([`Narrator`]) is pure over time and datagrams, so every
//! rule is a unit test; [`start`] is the thin thread around it.
use crate::wire::{self, BoxId, Hello, Up};
use crypto_box::{PublicKey, SecretKey};
use std::net::{SocketAddr, UdpSocket};
use std::time::{Duration, Instant};
/// **Where the lines come from**: everything from sequence number `from`, at
/// most `max` lines. Returns `(dropped, lines, next)`: `dropped` lines right
/// before the first returned one are gone for good, and `next` is where the
/// following read starts. nordisk-pairing's `BootLog` is one (feature `bootlog`).
pub trait Source: Send + Sync {
fn since(&self, from: u64, max: usize) -> (u64, Vec<String>, u64);
}
#[cfg(feature = "bootlog")]
impl Source for nordisk_pairing::bootlog::BootLog {
fn since(&self, from: u64, max: usize) -> (u64, Vec<String>, u64) {
let c = nordisk_pairing::bootlog::BootLog::since(self, from, max);
(c.dropped, c.lines, c.next)
}
}
/// The box stops streaming when the installer has said nothing this long (its
/// run ended, or it moved): nothing is sprayed at an address nobody reads.
pub const PEER_TIMEOUT: Duration = Duration::from_secs(30);
/// At most one WHO per this interval, whoever knocks.
pub const WHO_EVERY: Duration = Duration::from_millis(200);
/// Lines read from the source per datagram attempt.
const READ: usize = 64;
/// **What the box knows**: its per-boot key and id, the installer's public key,
/// and — once a hello authenticated it — where to stream and from which line.
pub struct Narrator {
eph: SecretKey,
installer: PublicKey,
id: BoxId,
peer: Option<Peer>,
cursor: u64,
last_counter: u64,
last_who: Option<Instant>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct Peer {
addr: SocketAddr,
at: Instant,
/// The `from` of the last hello.
from: u64,
}
impl Narrator {
pub fn new(installer_pub: [u8; 32], product: &str, mac: &str) -> Narrator {
Narrator {
eph: wire::generate(),
installer: PublicKey::from(installer_pub),
id: BoxId { product: product.into(), mac: mac.into(), boot: wire::random_u64() },
peer: None,
cursor: 0,
last_counter: 0,
last_who: None,
}
}
pub fn id(&self) -> &BoxId {
&self.id
}
/// Where the stream goes now, if anywhere.
pub fn peer(&self, now: Instant) -> Option<SocketAddr> {
self.peer.filter(|p| now.duration_since(p.at) < PEER_TIMEOUT).map(|p| p.addr)
}
/// The next line to send.
pub fn cursor(&self) -> u64 {
self.cursor
}
/// **One datagram arrived.** A knock is answered (the returned WHO goes back
/// to `from`), rate-limited; a hello that opens under the installer's key,
/// names this boot and is newer than every earlier one points the stream at
/// `from`. Everything else is ignored.
pub fn on_datagram(&mut self, from: SocketAddr, d: &[u8], now: Instant) -> Option<Vec<u8>> {
if let Some(nonce) = wire::parse_knock(d) {
if self.last_who.is_some_and(|t| now.duration_since(t) < WHO_EVERY) {
return None;
}
self.last_who = Some(now);
return Some(wire::seal_up(&self.eph, &self.installer, &Up::Who { id: self.id.clone(), echo: nonce }));
}
if let Ok(h) = wire::open_hello(&self.eph, &self.installer, d) {
self.hello(from, h, now);
}
None
}
fn hello(&mut self, from: SocketAddr, h: Hello, now: Instant) {
if h.boot != self.id.boot || h.counter <= self.last_counter {
return; // another boot's, or a replay
}
self.last_counter = h.counter;
match self.peer {
// A new reader (or its NAT moved): start where it asks.
Some(p) if p.addr != from => self.cursor = h.from,
None => self.cursor = h.from,
// It has more than this box thinks it sent: it skipped ahead.
Some(_) if h.from > self.cursor => self.cursor = h.from,
// No progress since its last hello while this box sent past it:
// what was sent was lost — go back.
Some(p) if h.from < self.cursor && h.from == p.from => self.cursor = h.from,
Some(_) => {}
}
self.peer = Some(Peer { addr: from, at: now, from: h.from });
}
/// **Up to `budget` datagrams** of lines from the cursor, to the peer — or
/// nothing when no authenticated hello is fresh.
pub fn pump(&mut self, source: &dyn Source, now: Instant, budget: usize) -> Vec<(SocketAddr, Vec<u8>)> {
let Some(to) = self.peer(now) else { return Vec::new() };
let mut out = Vec::new();
while out.len() < budget {
let (dropped, lines, next) = source.since(self.cursor, READ);
if lines.is_empty() {
if dropped > 0 || next > self.cursor {
out.push((to, wire::seal_up(&self.eph, &self.installer, &Up::Data { id: self.id.clone(), first: next, dropped, lines: Vec::new() })));
self.cursor = next;
}
break;
}
let first = next - lines.len() as u64;
let (up, n) = wire::pack(&self.id, first, dropped, &lines);
out.push((to, wire::seal_up(&self.eph, &self.installer, &up)));
self.cursor = first + n as u64;
}
out
}
}
/// **How a box narrates.**
#[derive(Debug, Clone)]
pub struct Config {
/// The installer's X25519 public key (the seed's `installer_pubkey`).
pub installer_pub: [u8; 32],
/// `0.0.0.0:`[`crate::PORT`] on a box (bound before any NIC has an
/// address: a hello can only arrive once one has, and then the whole
/// buffer is flushed).
pub bind: SocketAddr,
pub product: String,
/// The box's first NIC address, for the installer to tell boxes apart.
pub mac: String,
}
/// **Start narrating** on its own thread; returns the bound address. Every
/// error after the bind is swallowed: this is diagnostics, and a narration
/// that could stop a boot would be worse than none.
pub fn start(cfg: Config, source: &'static dyn Source) -> std::io::Result<SocketAddr> {
let sock = UdpSocket::bind(cfg.bind)?;
sock.set_read_timeout(Some(Duration::from_millis(40)))?;
let at = sock.local_addr()?;
let mut n = Narrator::new(cfg.installer_pub, &cfg.product, &cfg.mac);
std::thread::Builder::new().name("dev-narrate".into()).spawn(move || {
let mut buf = vec![0u8; 2048];
loop {
if let Ok((len, from)) = sock.recv_from(&mut buf) {
if let Some(who) = n.on_datagram(from, &buf[..len], Instant::now()) {
let _ = sock.send_to(&who, from);
}
}
for (to, d) in n.pump(source, Instant::now(), 32) {
let _ = sock.send_to(&d, to);
}
}
})?;
Ok(at)
}
/// **The first NIC's MAC** (`/sys/class/net/<iface>/address`), `-` when unknown.
pub fn mac_of(iface: &str) -> String {
std::fs::read_to_string(format!("/sys/class/net/{iface}/address")).map(|s| s.trim().to_string()).unwrap_or_else(|_| "-".into())
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Mutex;
struct Log(Mutex<Vec<String>>);
impl Source for Log {
fn since(&self, from: u64, max: usize) -> (u64, Vec<String>, u64) {
let l = self.0.lock().unwrap();
let f = (from as usize).min(l.len());
let v: Vec<String> = l[f..].iter().take(max).cloned().collect();
(0, v.clone(), (f + v.len()) as u64)
}
}
fn addr(p: u16) -> SocketAddr {
SocketAddr::from(([127, 0, 0, 1], p))
}
fn lines_of(installer: &SecretKey, sent: &[(SocketAddr, Vec<u8>)]) -> Vec<(u64, Vec<String>)> {
sent.iter()
.map(|(_, d)| match wire::open_up(installer, d).unwrap().1 {
Up::Data { first, lines, .. } => (first, lines),
other => panic!("{other:?}"),
})
.collect()
}
/// **A replayed hello is refused**, and a hello for another boot too: a
/// captured hello re-sent from an attacker's address cannot move the stream.
#[test]
fn a_replayed_or_other_boot_hello_does_not_move_the_stream() {
let installer = wire::generate();
let mut n = Narrator::new(*installer.public_key().as_bytes(), "gunnar-server", "-");
let t0 = Instant::now();
let eph = PublicKey::from(wire::open_up(&installer, &n.on_datagram(addr(9), &wire::knock([0; 16]), t0).unwrap()).unwrap().0);
let h1 = wire::seal_hello(&installer, &eph, &Hello { boot: n.id().boot, counter: 5, from: 0 });
assert!(n.on_datagram(addr(1000), &h1, t0).is_none());
assert_eq!(n.peer(t0), Some(addr(1000)));
n.on_datagram(addr(6666), &h1, t0); // the same hello, replayed from elsewhere
assert_eq!(n.peer(t0), Some(addr(1000)), "a replay moved the stream");
let other_boot = wire::seal_hello(&installer, &eph, &Hello { boot: n.id().boot ^ 1, counter: 9, from: 0 });
n.on_datagram(addr(6666), &other_boot, t0);
assert_eq!(n.peer(t0), Some(addr(1000)), "another boot's hello moved the stream");
assert_eq!(n.peer(t0 + PEER_TIMEOUT), None, "an installer that stopped saying hello is not streamed to");
}
/// **Nothing before a hello; everything from 0 after it; a loss is resent.**
#[test]
fn the_stream_starts_at_the_hellos_line_and_goes_back_when_the_installer_did_not_move() {
let installer = wire::generate();
let log = Log(Mutex::new((0..10).map(|i| format!("line {i}")).collect()));
let mut n = Narrator::new(*installer.public_key().as_bytes(), "gunnar-server", "-");
let t0 = Instant::now();
assert!(n.pump(&log, t0, 32).is_empty(), "no hello, no datagram");
let who = n.on_datagram(addr(9), &wire::knock([0; 16]), t0).unwrap();
let eph = PublicKey::from(wire::open_up(&installer, &who).unwrap().0);
assert!(n.on_datagram(addr(9), &wire::knock([0; 16]), t0).is_none(), "a second WHO within the interval");
let boot = n.id().boot;
let hello = |c: u64, from: u64| wire::seal_hello(&installer, &eph, &Hello { boot, counter: c, from });
let h = hello(1, 0);
n.on_datagram(addr(1000), &h, t0);
let sent = n.pump(&log, t0, 32);
let got: Vec<String> = lines_of(&installer, &sent).into_iter().flat_map(|(_, l)| l).collect();
assert_eq!(got.len(), 10);
assert_eq!(got[0], "line 0");
assert_eq!(n.cursor(), 10);
// The installer got nothing (all lost) and says from=0 again → resend.
let h = hello(2, 0);
n.on_datagram(addr(1000), &h, t0);
assert_eq!(n.cursor(), 0, "no progress between two hellos: the box goes back");
let again = lines_of(&installer, &n.pump(&log, t0, 32));
assert_eq!(again[0].0, 0);
// It moved to 4 (in flight: 4..10): nothing is resent.
let h = hello(3, 4);
n.on_datagram(addr(1000), &h, t0);
assert_eq!(n.cursor(), 10, "progress: the in-flight lines are not resent");
log.0.lock().unwrap().push("live".into());
assert_eq!(lines_of(&installer, &n.pump(&log, t0, 32)), vec![(10, vec!["live".to_string()])]);
}
}