zc2 0.0.28

//! Per-node Ed25519 identity: signs every /peer/* request; verified against the
//! dashboard-synced roster. Zero-trust replacement for the shared ZAKURO_PEER_KEY.

use base64::Engine;
use ring::signature::{self, Ed25519KeyPair, KeyPair};
use std::sync::Mutex;
use std::time::Instant;

const SKEW_SECS: u64 = 30;
const B64: base64::engine::general_purpose::GeneralPurpose =
    base64::engine::general_purpose::STANDARD;

pub struct NodeKey {
    pkcs8: Vec<u8>,
    pair: Ed25519KeyPair,
}

impl NodeKey {
    /// Generate a fresh Ed25519 node keypair.
    pub fn generate() -> NodeKey {
        let rng = ring::rand::SystemRandom::new();
        let doc = Ed25519KeyPair::generate_pkcs8(&rng).expect("pkcs8 gen");
        let pkcs8 = doc.as_ref().to_vec();
        let pair = Ed25519KeyPair::from_pkcs8(&pkcs8).expect("from pkcs8");
        NodeKey { pkcs8, pair }
    }

    /// Reconstruct from persisted PKCS#8 DER bytes.
    pub fn from_pkcs8(der: &[u8]) -> Result<NodeKey, String> {
        let pair = Ed25519KeyPair::from_pkcs8(der).map_err(|e| e.to_string())?;
        Ok(NodeKey {
            pkcs8: der.to_vec(),
            pair,
        })
    }

    pub fn pkcs8_bytes(&self) -> &[u8] {
        &self.pkcs8
    }

    pub fn public_b64(&self) -> String {
        B64.encode(self.pair.public_key().as_ref())
    }

    pub fn sign(&self, msg: &[u8]) -> String {
        B64.encode(self.pair.sign(msg).as_ref())
    }

    /// 16-char lowercase hex fingerprint (first 8 bytes of SHA-256 of the raw public key).
    pub fn fingerprint(&self) -> String {
        use ring::digest::{digest, SHA256};
        let d = digest(&SHA256, self.pair.public_key().as_ref());
        d.as_ref()[..8]
            .iter()
            .map(|b| format!("{:02x}", b))
            .collect()
    }

    /// Key-derived node identity URI.
    pub fn node_uri(&self) -> String {
        format!("zc://node-{}", self.fingerprint())
    }
}

/// Strip an optional `zc://` prefix, then an optional `node-` prefix.
/// Used to reduce a `?node=` filter/target arg down to its fingerprint form.
pub(crate) fn strip_node_arg(arg: &str) -> &str {
    let s = arg.strip_prefix("zc://").unwrap_or(arg);
    s.strip_prefix("node-").unwrap_or(s)
}

/// Same match semantics as [`node_filter_matches`], but against an already-known
/// fingerprint (e.g. a peer's, read off a worker's `node` handle) rather than
/// this process's own [`NodeKey`]. Never falls back to a hostname/label match.
pub(crate) fn fp_matches_node_arg(fp: &str, arg: &str) -> bool {
    strip_node_arg(arg) == fp
}

/// True iff `arg` (optionally `zc://`-prefixed, optionally `node-`-prefixed) reduces,
/// via that stripping, to exactly `key.fingerprint()`. Never falls back to a
/// hostname/label comparison — anything else returns false.
pub fn node_filter_matches(key: &NodeKey, arg: &str) -> bool {
    fp_matches_node_arg(&key.fingerprint(), arg)
}

impl NodeKey {
    /// Load the node key from the state dir, creating it (0600) on first run.
    ///
    /// The directory is `$ZAKURO_HOME`, else `~/.zakuro` — see
    /// [`crate::credentials::dir`] for why a container depends on the former.
    pub fn load_or_create() -> NodeKey {
        Self::load_or_create_in(crate::credentials::dir())
    }

    /// [`load_or_create`](Self::load_or_create) against an explicit directory.
    ///
    /// Exists so the persistence behaviour can be tested without redirecting
    /// `HOME`/`ZAKURO_HOME`: those are process-global, and `envs::update()`
    /// rewrites arbitrary keys from config, so a test that sets them races
    /// every other test that boots a broker.
    pub fn load_or_create_in(dir: Option<std::path::PathBuf>) -> NodeKey {
        let path = dir.map(|d| d.join("node_key"));
        if let Some(p) = &path {
            if let Ok(der) = std::fs::read(p) {
                if let Ok(k) = NodeKey::from_pkcs8(&der) {
                    return k;
                }
            }
        }
        let k = NodeKey::generate();
        if let Some(p) = &path {
            if let Some(dir) = p.parent() {
                let _ = std::fs::create_dir_all(dir);
            }
            if std::fs::write(p, k.pkcs8_bytes()).is_ok() {
                #[cfg(unix)]
                {
                    use std::os::unix::fs::PermissionsExt;
                    let _ = std::fs::set_permissions(p, std::fs::Permissions::from_mode(0o600));
                }
            }
        }
        k
    }

    /// Build the signed headers for an outbound `/peer/*` request.
    pub fn sign_headers(&self, method: &str, path: &str, body: &[u8]) -> Vec<(String, String)> {
        let nonce = gen_nonce();
        let ts = now_secs();
        let sig = self.sign(canonical(method, path, body, &nonce, ts).as_bytes());
        vec![
            ("X-Node-Id".to_string(), self.public_b64()),
            ("X-Node-Sig".to_string(), sig),
            ("X-Node-Nonce".to_string(), nonce),
            ("X-Node-Ts".to_string(), ts.to_string()),
        ]
    }
}

/// Fingerprint of a base64-encoded Ed25519 public key: SHA-256 of the decoded
/// raw pubkey bytes, first 8 bytes as lowercase hex. Matches `NodeKey::fingerprint()`
/// for the corresponding key. `None` if `pubkey_b64` doesn't decode.
pub fn fingerprint_of_pubkey_b64(pubkey_b64: &str) -> Option<String> {
    use ring::digest::{digest, SHA256};
    let raw = B64.decode(pubkey_b64).ok()?;
    let d = digest(&SHA256, &raw);
    Some(
        d.as_ref()[..8]
            .iter()
            .map(|b| format!("{:02x}", b))
            .collect(),
    )
}

/// Current unix time in seconds (0 if the clock is before the epoch).
pub fn now_secs() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

/// 128-bit random nonce as lowercase hex.
pub fn gen_nonce() -> String {
    use ring::rand::SecureRandom;
    let mut buf = [0u8; 16];
    ring::rand::SystemRandom::new()
        .fill(&mut buf)
        .expect("rng fill");
    buf.iter().map(|b| format!("{:02x}", b)).collect()
}

/// Verify a signed `/peer/*` request. `is_rostered` gates the signer pubkey against
/// the synced roster; `header` fetches a request header by name. Returns the verified
/// signer pubkey (b64) on success.
#[allow(clippy::too_many_arguments)]
pub fn verify_request(
    is_rostered: &dyn Fn(&str) -> bool,
    guard: &ReplayGuard,
    method: &str,
    path: &str,
    body: &[u8],
    header: &dyn Fn(&str) -> Option<String>,
    now: u64,
) -> Result<String, String> {
    let node_id = header("X-Node-Id").ok_or("missing X-Node-Id")?;
    let sig = header("X-Node-Sig").ok_or("missing X-Node-Sig")?;
    let nonce = header("X-Node-Nonce").ok_or("missing X-Node-Nonce")?;
    let ts: u64 = header("X-Node-Ts")
        .and_then(|s| s.parse().ok())
        .ok_or("missing/invalid X-Node-Ts")?;
    if !is_rostered(&node_id) {
        return Err("node not in roster (or revoked)".into());
    }
    if !guard.fresh(&nonce, ts, now) {
        return Err("stale timestamp or replayed nonce".into());
    }
    if !verify_sig(
        &node_id,
        canonical(method, path, body, &nonce, ts).as_bytes(),
        &sig,
    ) {
        return Err("bad signature".into());
    }
    Ok(node_id)
}

/// Lowercase hex SHA-256 of a request body.
pub fn sha256_hex(body: &[u8]) -> String {
    let d = ring::digest::digest(&ring::digest::SHA256, body);
    d.as_ref().iter().map(|b| format!("{:02x}", b)).collect()
}

/// Canonical string a node signs for a `/peer/*` request.
pub fn canonical(method: &str, path: &str, body: &[u8], nonce: &str, ts: u64) -> String {
    format!("{method}\n{path}\n{}\n{nonce}\n{ts}", sha256_hex(body))
}

/// Verify an Ed25519 signature (all b64) over `msg`.
pub fn verify_sig(pubkey_b64: &str, msg: &[u8], sig_b64: &str) -> bool {
    let pk = match B64.decode(pubkey_b64) {
        Ok(v) => v,
        Err(_) => return false,
    };
    let sig = match B64.decode(sig_b64) {
        Ok(v) => v,
        Err(_) => return false,
    };
    let upk = signature::UnparsedPublicKey::new(&signature::ED25519, pk);
    upk.verify(msg, &sig).is_ok()
}

/// Anti-replay: rejects reused nonces and timestamps outside a ±30s window.
pub struct ReplayGuard {
    seen: Mutex<std::collections::HashMap<String, Instant>>,
}

impl Default for ReplayGuard {
    fn default() -> Self {
        Self::new()
    }
}

impl ReplayGuard {
    pub fn new() -> Self {
        ReplayGuard {
            seen: Mutex::new(std::collections::HashMap::new()),
        }
    }

    /// True iff `ts` is within ±30s of `now` AND `nonce` was not seen before.
    pub fn fresh(&self, nonce: &str, ts: u64, now: u64) -> bool {
        if now.abs_diff(ts) > SKEW_SECS {
            return false;
        }
        let mut m = self.seen.lock().unwrap();
        m.retain(|_, t| t.elapsed().as_secs() < 60);
        if m.contains_key(nonce) {
            return false;
        }
        m.insert(nonce.to_string(), Instant::now());
        true
    }
}

#[cfg(test)]
mod fp_tests {
    use super::*;
    #[test]
    fn fingerprint_is_stable_and_unique() {
        let a = NodeKey::generate();
        let der = a.pkcs8_bytes().to_vec();
        let a2 = NodeKey::from_pkcs8(&der).unwrap();
        assert_eq!(a.fingerprint(), a2.fingerprint()); // stable across reload
        assert_eq!(a.fingerprint().len(), 16);
        assert!(a
            .fingerprint()
            .chars()
            .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase()));
        let b = NodeKey::generate();
        assert_ne!(a.fingerprint(), b.fingerprint()); // unique per key
        assert_eq!(a.node_uri(), format!("zc://node-{}", a.fingerprint()));
    }

    #[test]
    fn node_filter_matches_fingerprint_not_hostname() {
        let key = NodeKey::generate();
        let fp = key.fingerprint();
        assert!(super::node_filter_matches(&key, &format!("zc://node-{fp}")));
        assert!(super::node_filter_matches(&key, &format!("node-{fp}"))); // bare accepted
        assert!(!super::node_filter_matches(&key, "node-lxd")); // hostname rejected
        assert!(!super::node_filter_matches(&key, "zc://node-lxd"));
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn sign_verify_roundtrip() {
        let k = NodeKey::generate();
        let msg = b"hello mesh";
        let sig = k.sign(msg);
        assert!(verify_sig(&k.public_b64(), msg, &sig));
        assert!(!verify_sig(&k.public_b64(), b"tampered", &sig));
    }

    #[test]
    fn persist_roundtrip() {
        let k = NodeKey::generate();
        let der = k.pkcs8_bytes().to_vec();
        let k2 = NodeKey::from_pkcs8(&der).unwrap();
        assert_eq!(k.public_b64(), k2.public_b64());
    }

    #[test]
    fn canonical_is_stable() {
        let c = canonical("POST", "/peer/tasks/offer", b"{}", "abc", 1000);
        assert_eq!(
            c,
            "POST\n/peer/tasks/offer\n44136fa355b3678a1146ad16f7e8649e94fb4fc21fe77e8310c060f61caaff8a\nabc\n1000"
        );
    }

    /// Was written against `HOME`; now drives the directory directly. The key
    /// facts here -- same identity on reload, and 0600 on the file -- are
    /// properties of the persistence, not of how the path was resolved, and
    /// resolving it from the environment only made the test racy. Precedence
    /// is covered purely by `zakuro_home_outranks_home_and_empty_does_not_count`.
    #[test]
    fn load_or_create_persists() {
        let tmp =
            std::env::temp_dir().join(format!("zc-nodekey-{}-{}", std::process::id(), line!()));
        std::fs::create_dir_all(&tmp).unwrap();

        let k1 = NodeKey::load_or_create_in(Some(tmp.clone()));
        let k2 = NodeKey::load_or_create_in(Some(tmp.clone()));
        assert_eq!(k1.public_b64(), k2.public_b64());
        // file exists and is 0600
        let p = tmp.join("node_key");
        assert!(p.exists());
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let mode = std::fs::metadata(&p).unwrap().permissions().mode();
            assert_eq!(mode & 0o777, 0o600);
        }

        let _ = std::fs::remove_dir_all(&tmp);
    }

    /// A containerised broker has no durable `$HOME`. Keeping the identity on
    /// a mounted volume is how it survives a restart -- and for weeks it did
    /// not, because `credentials::dir()` read only `HOME`, so the volume
    /// stayed empty and every restart registered a brand-new node.
    ///
    /// Deliberately drives `load_or_create_in` with explicit paths rather than
    /// redirecting `ZAKURO_HOME`. An earlier version of this test set the
    /// variable and failed on CI: `envs::update()` rewrites arbitrary keys
    /// from config, so a concurrent test that boots a broker can move it
    /// mid-assertion. Locking cannot fix that, because those writers do not
    /// take the lock.
    #[test]
    fn a_mounted_volume_keeps_a_node_identity_across_restarts() {
        let base = std::env::temp_dir().join(format!(
            "zc-nodekey-volume-{}-{}",
            std::process::id(),
            line!()
        ));
        let mounted = base.join("volume"); // stands in for the PVC
        std::fs::create_dir_all(&mounted).unwrap();

        let k1 = NodeKey::load_or_create_in(Some(mounted.clone()));
        assert!(
            mounted.join("node_key").exists(),
            "key must land on the volume"
        );

        // Restart with the volume still attached: same identity.
        let k2 = NodeKey::load_or_create_in(Some(mounted.clone()));
        assert_eq!(k1.public_b64(), k2.public_b64());

        // A different (empty) directory is a different node -- which is
        // exactly what an unmounted volume produced on every restart.
        let elsewhere = base.join("ephemeral");
        std::fs::create_dir_all(&elsewhere).unwrap();
        let k3 = NodeKey::load_or_create_in(Some(elsewhere));
        assert_ne!(k1.public_b64(), k3.public_b64());

        let _ = std::fs::remove_dir_all(&base);
    }

    /// The precedence rule, with no environment mutation anywhere.
    #[test]
    fn zakuro_home_outranks_home_and_empty_does_not_count() {
        use crate::credentials::dir_from;
        use std::ffi::OsString;
        let v = |s: &str| Some(OsString::from(s));

        // A container points ZAKURO_HOME at its mounted volume.
        assert_eq!(
            dir_from(v("/var/lib/zakuro"), v("/root")),
            Some(std::path::PathBuf::from("/var/lib/zakuro"))
        );
        // Unset falls back to $HOME/.zakuro -- every non-container install.
        assert_eq!(
            dir_from(None, v("/root")),
            Some(std::path::PathBuf::from("/root/.zakuro"))
        );
        // Empty is not a directory. Treating it as one would resolve the node
        // key to a bare "node_key" in the process CWD.
        assert_eq!(
            dir_from(v(""), v("/root")),
            Some(std::path::PathBuf::from("/root/.zakuro"))
        );
        // Neither set: no state dir, and the caller must cope rather than
        // silently write somewhere arbitrary.
        assert_eq!(dir_from(None, None), None);
    }

    #[test]
    fn sign_headers_verify_request_roundtrip() {
        let k = NodeKey::generate();
        let body = br#"{"fn":"greet"}"#;
        let hdrs = k.sign_headers("POST", "/peer/tasks/offer", body);
        let get = |name: &str| -> Option<String> {
            hdrs.iter().find(|(h, _)| h == name).map(|(_, v)| v.clone())
        };
        let guard = ReplayGuard::new();
        let signer = k.public_b64();
        let rostered = |id: &str| id == signer;
        let now = now_secs();
        // happy path returns the signer pubkey
        let got = verify_request(
            &rostered,
            &guard,
            "POST",
            "/peer/tasks/offer",
            body,
            &get,
            now,
        )
        .unwrap();
        assert_eq!(got, signer);

        // unknown roster id → reject (fresh guard to avoid replay masking)
        let g2 = ReplayGuard::new();
        assert!(verify_request(
            &|_| false,
            &g2,
            "POST",
            "/peer/tasks/offer",
            body,
            &get,
            now
        )
        .is_err());

        // tampered body → bad signature
        let g3 = ReplayGuard::new();
        assert!(verify_request(
            &rostered,
            &g3,
            "POST",
            "/peer/tasks/offer",
            b"{}",
            &get,
            now
        )
        .is_err());

        // replay (same guard, same nonce) → reject
        assert!(verify_request(
            &rostered,
            &guard,
            "POST",
            "/peer/tasks/offer",
            body,
            &get,
            now
        )
        .is_err());
    }

    #[test]
    fn nonce_is_random_and_hex() {
        let a = gen_nonce();
        let b = gen_nonce();
        assert_eq!(a.len(), 32);
        assert_ne!(a, b);
        assert!(a.chars().all(|c| c.is_ascii_hexdigit()));
    }

    #[test]
    fn replay_and_skew() {
        let g = ReplayGuard::new();
        assert!(g.fresh("n1", 1000, 1010)); // 10s skew ok, first use
        assert!(!g.fresh("n1", 1000, 1010)); // replay
        assert!(!g.fresh("n2", 1000, 1040)); // 40s skew rejected
        assert!(!g.fresh("n3", 1040, 1000)); // future skew rejected
    }
}