nordisk-pairing 0.5.29

//! **The front proves the tunnel is the only path** (feature `mesh-probe`, off by default:
//! it DIALS the estate's boxes). Gunnar's `gunnar-front` `mesh_probe` (lane W9) and Holger's
//! `holger-estate-front` copy were the same code; it is shared here since P4c (A28 M11).
//!
//! An estate box is WireGuard-only (lane E1): its estate doors bind on wg0's overlay address. The
//! front sits on the same utility network as main and the twin, so it is the one place a
//! plain-IP connection between boxes can be tried for real. Once the installer has fed the
//! boxes, the product's runner dials each box's UTILITY address on the product's plain ports —
//! every one must be refused — and each box's control plane on its OVERLAY address, which must
//! answer ([`Watch::look`]). The lines say what was seen, one per dial and one verdict line
//! ([`PASS`] / [`FAIL`]), each at its level ([`levels`]) and only when it changed ([`due`]).
//!
//! What stays the product's: its plain ports, its runner (a jera job, a thread), and the config
//! switch that turns the probe on (Holger: `wg_only`; Gunnar: always, on the wg-only medium).

use std::net::{IpAddr, SocketAddr};
use std::sync::Mutex;
use std::time::{Duration, Instant};

use crate::join::JoinReply;
use crate::Role;

/// The verdict line's words, as the kit looks for them (Gunnar's).
pub const PASS: &str = "wg-only: PASS";
pub const FAIL: &str = "wg-only: FAIL";

/// How long one dial may take before it counts as dropped (refused).
pub const DIAL: Duration = Duration::from_secs(3);
/// How often the probe looks until it has passed once; then rarely.
pub const UNTIL_PASSED: Duration = Duration::from_secs(20);
pub const AFTER_PASSED: Duration = Duration::from_secs(600);
/// A standing failure is said again only this long after it was last said.
pub const REPEAT_FAILURE: Duration = Duration::from_secs(600);

/// One box to probe: its name, its utility address and its overlay address.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Box3 {
    pub name: String,
    pub utility: IpAddr,
    pub overlay: IpAddr,
}

/// Pure: **the boxes to probe**, from the fed reply — main, and the twin, each carrying a
/// utility address. `None` until main is; with `twin_required` (Gunnar), `None` until the
/// twin is too, else (Holger) the twin is probed once it is fed.
pub fn boxes_of(reply: &JoinReply, twin_required: bool) -> Option<Vec<Box3>> {
    let of = |role: Role| reply.peers.iter().find(|p| p.role == role).and_then(|p| Some(Box3 { name: p.name.clone(), utility: IpAddr::V4(p.utility?), overlay: IpAddr::V4(p.overlay) }));
    let mut out = vec![of(Role::Primary)?];
    match of(Role::Twin) {
        Some(t) => out.push(t),
        None if twin_required => return None,
        None => {}
    }
    Some(out)
}

/// What one dial saw.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Seen {
    /// Refused (or dropped: nothing came back in time) — and why.
    Refused(String),
    /// A connection was made.
    Answered,
}

/// **Dial `at` once** (blocking: a look runs on the queue's blocking thread). Refused: the
/// kernel refused (nothing listens there), nothing came back (dropped), or the connection was
/// CLOSED AT ACCEPT with nothing said. Answered: a connection that stays open (a listener waiting
/// for its client, as gRPC does) or that speaks.
pub fn dial(at: SocketAddr, within: Duration) -> Seen {
    use std::io::Read as _;
    let mut s = match std::net::TcpStream::connect_timeout(&at, within) {
        Ok(s) => s,
        Err(e) if e.kind() == std::io::ErrorKind::TimedOut => return Seen::Refused(format!("no answer within {} s (dropped)", within.as_secs())),
        Err(e) => return Seen::Refused(e.to_string()),
    };
    let _ = s.set_read_timeout(Some(within));
    let mut b = [0u8; 1];
    match s.read(&mut b) {
        Ok(0) => Seen::Refused("closed at accept, nothing said".into()),
        Err(e) if matches!(e.kind(), std::io::ErrorKind::ConnectionReset | std::io::ErrorKind::ConnectionAborted) => Seen::Refused(format!("closed at accept, nothing said ({e})")),
        Ok(_) | Err(_) => Seen::Answered,
    }
}

/// Pure: **the lines and the verdict** for what was seen — `(box, at, plain?, seen)`. PASS when
/// every plain-IP dial was refused and every tunnel dial answered.
pub fn verdict(seen: &[(String, SocketAddr, bool, Seen)]) -> (Vec<String>, bool) {
    let mut lines = Vec::new();
    let mut ok = true;
    for (name, at, plain, s) in seen {
        match (plain, s) {
            (true, Seen::Refused(why)) => lines.push(format!("wg-only: {name} {at} plain IP refused ({why})")),
            (true, Seen::Answered) => {
                ok = false;
                lines.push(format!("wg-only: {name} {at} ANSWERED over plain IP"));
            }
            (false, Seen::Answered) => lines.push(format!("wg-only: {name} {at} answers over the tunnel")),
            (false, Seen::Refused(why)) => {
                ok = false;
                lines.push(format!("wg-only: {name} {at} does NOT answer over the tunnel ({why})"));
            }
        }
    }
    let plain = seen.iter().filter(|s| s.2).count();
    let tunnel = seen.len() - plain;
    lines.push(if ok {
        format!("{PASS} — {plain} plain-IP connection(s) to the boxes refused, {tunnel} over the tunnel answered")
    } else {
        format!("{FAIL} — a plain-IP connection answered, or the tunnel did not")
    });
    (lines, ok)
}

/// **Probe the boxes once**: every plain-IP port, then each control plane over the tunnel.
pub fn probe(boxes: &[Box3], ports: &[u16], control: u16, within: Duration) -> Vec<(String, SocketAddr, bool, Seen)> {
    let mut seen = Vec::new();
    for b in boxes {
        for p in ports {
            let at = SocketAddr::new(b.utility, *p);
            seen.push((b.name.clone(), at, true, dial(at, within)));
        }
        let at = SocketAddr::new(b.overlay, control);
        seen.push((b.name.clone(), at, false, dial(at, within)));
    }
    seen
}

/// **Each line and whether it is a defect (WARN) or a fact (INFO)**: a plain-IP connection that
/// ANSWERED, a tunnel that does NOT answer and the FAIL verdict are defects.
pub fn levels(lines: &[String]) -> Vec<(bool, String)> {
    lines.iter().map(|l| (l.contains("ANSWERED over plain IP") || l.contains("does NOT answer over the tunnel") || l.starts_with(FAIL), l.clone())).collect()
}

/// **Whether to say the probe's lines now**: the first time, whenever they or the verdict
/// changed, and a standing failure again only after [`REPEAT_FAILURE`] since it was last said.
pub fn due(last: Option<&(Vec<String>, bool)>, lines: &[String], ok: bool, said_at: Instant, now: Instant) -> bool {
    match last {
        None => true,
        Some((l, k)) if l.as_slice() != lines || *k != ok => true,
        Some(_) => !ok && now.duration_since(said_at) >= REPEAT_FAILURE,
    }
}

/// **The probe's memory between looks**: the boxes last probed and whether they passed, and
/// the lines last said and when.
#[derive(Default)]
pub struct Watch {
    last: Mutex<Option<(Vec<Box3>, bool)>>,
    said: Mutex<Option<(Vec<String>, bool, Instant)>>,
}

impl Watch {
    /// Whether the probe has passed for the boxes it last saw (the ticker then looks rarely).
    pub fn passed(&self) -> bool {
        self.last.lock().map(|l| l.as_ref().is_some_and(|l| l.1)).unwrap_or(false)
    }

    /// **One look** over `boxes` on `ports` / `control`: the lines to say now (empty when they
    /// were said already) and the verdict.
    pub fn look(&self, boxes: &[Box3], ports: &[u16], control: u16, within: Duration) -> (Vec<(bool, String)>, bool) {
        let seen = probe(boxes, ports, control, within);
        let (lines, ok) = verdict(&seen);
        let now = Instant::now();
        let mut last = self.last.lock().unwrap_or_else(|e| e.into_inner());
        let mut said = self.said.lock().unwrap_or_else(|e| e.into_inner());
        // A changed estate (other boxes fed) is a changed verdict too.
        let changed_boxes = last.as_ref().map(|l| l.0.as_slice() != boxes).unwrap_or(true);
        let key = said.as_ref().map(|(l, k, _)| (l.clone(), *k));
        let at = said.as_ref().map(|s| s.2).unwrap_or(now);
        let say = changed_boxes || due(key.as_ref(), &lines, ok, at, now);
        *last = Some((boxes.to_vec(), ok));
        if !say {
            return (Vec::new(), ok);
        }
        *said = Some((lines.clone(), ok, now));
        (levels(&lines), ok)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::join::{Assigned, Peer};
    use std::collections::BTreeMap;
    use std::net::Ipv4Addr;

    fn peer(name: &str, role: Role, last: u8, utility: bool) -> Peer {
        Peer { name: name.into(), role, tunnel_pubkey: "K".into(), overlay: Ipv4Addr::new(10, 44, 0, last), endpoint: None, utility: utility.then(|| Ipv4Addr::new(10, 13, 8, last)) }
    }

    fn reply(peers: Vec<Peer>) -> JoinReply {
        JoinReply { estate: "holger.test".into(), me: Assigned { name: "vm1".into(), overlay: crate::OVERLAY_FRONT }, cert_pem: None, role: Some(Role::Front), primary_overlay: None, primary_utility: None, peers, feed: BTreeMap::new() }
    }

    /// A listener that holds its connections, as a gRPC server waits for its client.
    fn holding(at: &str) -> SocketAddr {
        let l = std::net::TcpListener::bind(at).unwrap();
        let a = l.local_addr().unwrap();
        std::thread::spawn(move || {
            let mut held = Vec::new();
            for s in l.incoming().flatten() {
                held.push(s);
            }
        });
        a
    }

    /// Main once it is fed with a utility address, and the twin once it is too; with the twin
    /// required (Gunnar), nothing until both are.
    #[test]
    fn the_boxes_are_main_and_the_twin_once_fed() {
        assert_eq!(boxes_of(&reply(vec![]), false), None, "nothing fed");
        assert_eq!(boxes_of(&reply(vec![peer("vm2", Role::Primary, 2, false)]), false), None, "main with no utility address");
        assert_eq!(boxes_of(&reply(vec![peer("vm2", Role::Primary, 2, true), peer("vm3", Role::Twin, 3, false)]), false).unwrap().len(), 1, "a twin with no utility address is not probed yet");
        assert_eq!(boxes_of(&reply(vec![peer("vm2", Role::Primary, 2, true), peer("vm3", Role::Twin, 3, false)]), true), None, "Gunnar: both or nothing");
        for twin_required in [false, true] {
            let b = boxes_of(&reply(vec![peer("vm3", Role::Twin, 3, true), peer("vm2", Role::Primary, 2, true)]), twin_required).unwrap();
            assert_eq!(b.iter().map(|b| (b.name.as_str(), b.utility.to_string(), b.overlay.to_string())).collect::<Vec<_>>(), vec![("vm2", "10.13.8.2".into(), "10.44.0.2".into()), ("vm3", "10.13.8.3".into(), "10.44.0.3".into())]);
        }
    }

    /// **A port that accepts and closes at once, nothing said, is refused**; a port that accepts
    /// and waits, or speaks, ANSWERED.
    #[test]
    fn closed_at_accept_is_refused_and_a_listener_that_waits_or_speaks_answers() {
        use std::io::Write as _;
        let closer = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let c = closer.local_addr().unwrap();
        std::thread::spawn(move || {
            for s in closer.incoming().flatten() {
                drop(s);
            }
        });
        let speaker = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let sp = speaker.local_addr().unwrap();
        std::thread::spawn(move || {
            let mut held = Vec::new();
            for mut s in speaker.incoming().flatten() {
                let _ = s.write_all(b"SSH-2.0-x\r\n");
                held.push(s);
            }
        });
        let w = holding("127.0.0.1:0");
        assert!(matches!(dial(c, Duration::from_secs(2)), Seen::Refused(why) if why.contains("closed at accept")), "accepted and closed, nothing said: refused");
        assert_eq!(dial(w, Duration::from_secs(1)), Seen::Answered, "a listener that waits for the client is there");
        assert_eq!(dial(sp, Duration::from_secs(2)), Seen::Answered, "a listener that speaks is there");
    }

    /// ★ **Refused on the utility address, answered on the tunnel**: PASS. A plain-IP
    /// connection that answers — or a tunnel that does not — is FAIL, named. Here the utility
    /// address is 127.0.0.1 and the tunnel 127.0.0.2.
    #[test]
    fn plain_ip_refused_and_the_tunnel_answering_is_a_pass_and_the_rest_a_named_fail() {
        let tunnel = holding("127.0.0.2:0");
        let port = tunnel.port();
        let b = vec![Box3 { name: "main".into(), utility: "127.0.0.1".parse().unwrap(), overlay: "127.0.0.2".parse().unwrap() }];
        let (lines, ok) = verdict(&probe(&b, &[port], port, Duration::from_secs(1)));
        assert!(ok, "{lines:#?}");
        assert!(lines[0].contains("plain IP refused") && lines[1].contains("answers over the tunnel") && lines[2].starts_with(PASS), "{lines:#?}");

        // A box that answers over plain IP fails, by name.
        let plain = holding("127.0.0.1:0");
        let (lines, ok) = verdict(&probe(&b, &[plain.port()], port, Duration::from_secs(1)));
        assert!(!ok);
        assert!(lines.iter().any(|l| l.contains("ANSWERED over plain IP")) && lines.last().unwrap().starts_with(FAIL), "{lines:#?}");
        // And a tunnel that does not answer fails too.
        let gone = std::net::TcpListener::bind("127.0.0.2:0").unwrap().local_addr().unwrap().port();
        let (lines, ok) = verdict(&probe(&b, &[], gone, Duration::from_secs(1)));
        assert!(!ok && lines[0].contains("does NOT answer over the tunnel"), "{lines:#?}");
    }

    /// Said once, at the level it deserves: the refusal and the answering tunnel are INFO, a
    /// defect WARN; the same lines a second time are not said again, a changed verdict is.
    #[test]
    fn a_look_says_its_lines_once_and_again_when_they_change() {
        let tunnel = holding("127.0.0.2:0");
        let port = tunnel.port();
        let b = vec![Box3 { name: "main".into(), utility: "127.0.0.1".parse().unwrap(), overlay: "127.0.0.2".parse().unwrap() }];
        let w = Watch::default();
        let (said, ok) = w.look(&b, &[port], port, Duration::from_secs(1));
        assert!(ok && w.passed());
        assert_eq!(said.len(), 3);
        assert!(said.iter().all(|(warn, _)| !warn), "a healthy estate says only facts: {said:?}");
        let (again, ok) = w.look(&b, &[port], port, Duration::from_secs(1));
        assert!(ok && again.is_empty(), "the same verdict is not said again: {again:?}");
        let plain = holding("127.0.0.1:0");
        let (changed, ok) = w.look(&b, &[plain.port()], port, Duration::from_secs(1));
        assert!(!ok && !w.passed());
        assert!(changed.iter().any(|(warn, l)| *warn && l.starts_with(FAIL)), "{changed:?}");
        let t0 = Instant::now();
        let key = (vec![format!("{FAIL} — x")], false);
        assert!(!due(Some(&key), &key.0, false, t0, t0 + Duration::from_secs(20)), "a standing failure 20 s later: not again");
        assert!(due(Some(&key), &key.0, false, t0, t0 + REPEAT_FAILURE), "again after the repeat");
    }
}