zc2 0.0.25

//! Mesh directory: the CLI's view of "which brokers are on the mesh".
//!
//! The hub's node roster (`GET /api/broker/node/roster`) is the deployment-wide
//! list of authorized broker signing keys, each with the `mesh_endpoint`
//! (`ip:port`) the broker last advertised. A broker's public identity is its
//! key fingerprint — `zc://node-<fp>` with `fp = sha256(pubkey)[..8]` (see
//! `broker::node_identity`) — so the roster is enough to resolve a `zc://node-…`
//! target to an address, and to enumerate the mesh, WITHOUT a local broker and
//! WITHOUT the fleet's shared `ZAKURO_PEER_KEY` (only `/peer/*` needs that;
//! `/health` and `/workers` are open client surfaces).
//!
//! Sources, in order: the hub (credentials from `~/.zakuro/credentials` /
//! env), then the on-disk roster cache a local broker may have written
//! (`~/.zakuro/roster.json`) so the directory keeps working offline.
//!
//! IP-FREE RULE: this module returns endpoints to *callers that dial them*; the
//! user-facing renderers (`zc brokers`) print only `zc://node-<fp>`, the
//! broker's self-reported `node_name`, and reachability — never an address.

use std::time::Duration;

/// One authorized broker as the roster describes it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MeshBroker {
    /// 16-hex key fingerprint (bare, no `zc://node-` prefix).
    pub fp: String,
    /// `ip:port` the broker advertised, if it has a tunnel up.
    pub endpoint: Option<String>,
    pub revoked: bool,
}

impl MeshBroker {
    pub fn node_uri(&self) -> String {
        format!("zc://node-{}", self.fp)
    }
}

/// Result of probing one broker's `/health`.
#[derive(Debug, Clone)]
pub struct BrokerProbe {
    pub broker: MeshBroker,
    pub reachable: bool,
    pub node_name: Option<String>,
    pub mesh_tunnel: Option<String>,
}

/// Parse a roster body (`[{node_pubkey, revoked, mesh_endpoint}]`) into
/// fingerprinted entries. Entries whose pubkey does not decode are dropped.
pub fn parse_roster(body: &str) -> Vec<MeshBroker> {
    let v: serde_json::Value = match serde_json::from_str(body) {
        Ok(v) => v,
        Err(_) => return Vec::new(),
    };
    v.as_array()
        .map(|arr| {
            arr.iter()
                .filter_map(|e| {
                    let pk = e.get("node_pubkey")?.as_str()?;
                    let fp = crate::broker::node_identity::fingerprint_of_pubkey_b64(pk)?;
                    let revoked = e.get("revoked").and_then(|r| r.as_bool()).unwrap_or(false);
                    let endpoint = e
                        .get("mesh_endpoint")
                        .and_then(|m| m.as_str())
                        .map(str::trim)
                        .filter(|s| !s.is_empty())
                        .map(str::to_string);
                    Some(MeshBroker {
                        fp,
                        endpoint,
                        revoked,
                    })
                })
                .collect()
        })
        .unwrap_or_default()
}

/// Fetch the roster from the hub. Needs an API key (zk_ or a `zc login`
/// session token) — loaded from the saved credentials when not in the env.
pub fn fetch_from_hub() -> Result<Vec<MeshBroker>, String> {
    crate::credentials::load_into_env();
    let api_key = std::env::var("ZAKURO_API_KEY")
        .ok()
        .filter(|k| !k.trim().is_empty())
        .ok_or_else(|| "not signed in (run `zc login` or set ZAKURO_API_KEY)".to_string())?;
    let api_url = crate::credentials::default_api_url();
    let endpoint = format!("{}/api/broker/node/roster", api_url.trim_end_matches('/'));
    let resp = ureq::get(&endpoint)
        .config()
        .timeout_global(Some(Duration::from_secs(10)))
        .http_status_as_error(false)
        .build()
        .header("X-Broker-Api-Key", &api_key)
        .call()
        .map_err(|e| format!("roster request failed: {e}"))?;
    let status = resp.status().as_u16();
    if status != 200 {
        return Err(format!("roster HTTP {status} from {api_url}"));
    }
    let body = resp
        .into_body()
        .read_to_string()
        .map_err(|e| format!("roster read failed: {e}"))?;
    Ok(parse_roster(&body))
}

/// The roster a local broker cached on disk (`~/.zakuro/roster.json`), if any.
pub fn load_cached() -> Vec<MeshBroker> {
    let cache = crate::broker::roster_cache::RosterCache::load();
    cache
        .entries()
        .into_iter()
        .filter_map(|(pk, revoked)| {
            let fp = crate::broker::node_identity::fingerprint_of_pubkey_b64(&pk)?;
            let endpoint = cache.endpoint_for_fingerprint(&fp);
            Some(MeshBroker {
                fp,
                endpoint,
                revoked,
            })
        })
        .collect()
}

/// Hub first, cached roster as the offline fallback. `Err` only when neither
/// source yields anything; the message names the hub failure.
pub fn directory() -> Result<Vec<MeshBroker>, String> {
    match fetch_from_hub() {
        Ok(list) => Ok(list),
        Err(e) => {
            let cached = load_cached();
            if cached.is_empty() {
                Err(e)
            } else {
                crate::vpn::vlog(&format!(
                    "hub roster unavailable ({e}); using cached roster"
                ));
                Ok(cached)
            }
        }
    }
}

/// Resolve a bare fingerprint (or `zc://node-<fp>` / `node-<fp>`) to the
/// broker's advertised `ip:port`. `None` when unknown, revoked, or off-mesh.
pub fn endpoint_for(arg: &str) -> Option<String> {
    let fp = crate::broker::node_identity::strip_node_arg(arg);
    directory()
        .ok()?
        .into_iter()
        .find(|b| b.fp == fp && !b.revoked)
        .and_then(|b| b.endpoint)
}

/// True when `s` has the shape of a key fingerprint: 16 lowercase hex chars.
pub fn looks_like_fingerprint(s: &str) -> bool {
    s.len() == 16
        && s.bytes()
            .all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase())
}

/// Probe every broker that advertises an endpoint, concurrently, through the
/// mesh-aware HTTP agent (sidecar proxy when the host is not routable).
/// Brokers without an endpoint are returned unreachable without a probe.
pub fn probe_all(brokers: Vec<MeshBroker>, timeout: Duration) -> Vec<BrokerProbe> {
    let handles: Vec<_> = brokers
        .into_iter()
        .map(|b| {
            std::thread::spawn(move || {
                let Some(ep) = b.endpoint.clone() else {
                    return BrokerProbe {
                        broker: b,
                        reachable: false,
                        node_name: None,
                        mesh_tunnel: None,
                    };
                };
                let agent = crate::vpn::mesh_agent(timeout);
                let url = format!("http://{}/health", ep);
                let json: Option<serde_json::Value> = agent
                    .get(&url)
                    .call()
                    .ok()
                    .and_then(|r| r.into_body().read_to_string().ok())
                    .and_then(|t| serde_json::from_str(&t).ok());
                let is_broker = json
                    .as_ref()
                    .and_then(|j| j.get("service"))
                    .and_then(|s| s.as_str())
                    == Some("zakuro-broker");
                BrokerProbe {
                    node_name: json
                        .as_ref()
                        .and_then(|j| j.get("node_name"))
                        .and_then(|s| s.as_str())
                        .map(str::to_string),
                    mesh_tunnel: json
                        .as_ref()
                        .and_then(|j| j.get("mesh_tunnel"))
                        .and_then(|s| s.as_str())
                        .map(str::to_string),
                    reachable: is_broker,
                    broker: b,
                }
            })
        })
        .collect();
    let mut out: Vec<BrokerProbe> = handles.into_iter().filter_map(|h| h.join().ok()).collect();
    out.sort_by(|a, b| {
        b.reachable
            .cmp(&a.reachable)
            .then_with(|| a.broker.fp.cmp(&b.broker.fp))
    });
    out
}

/// Render the directory for `zc brokers`. IP-free by construction: only the
/// key-derived id, the broker's own display name and liveness are printed.
pub fn render(
    probes: &[BrokerProbe],
    self_fp: Option<&str>,
    offline_without_endpoint: usize,
) -> String {
    use colored::Colorize;
    let mut out = String::new();
    let live: Vec<&BrokerProbe> = probes
        .iter()
        .filter(|p| p.broker.endpoint.is_some())
        .collect();
    if live.is_empty() {
        out.push_str("  No brokers currently advertise a mesh endpoint.\n");
    } else {
        out.push_str(&format!(
            "  {} broker(s) on the mesh:\n",
            live.iter().filter(|p| p.reachable).count()
        ));
        for p in live {
            let mark = if p.reachable {
                "✓".green().to_string()
            } else {
                "✗".red().to_string()
            };
            let me = if self_fp == Some(p.broker.fp.as_str()) {
                "  (this machine)".dimmed().to_string()
            } else {
                String::new()
            };
            let name = p.node_name.as_deref().unwrap_or("-");
            let tunnel = match p.mesh_tunnel.as_deref() {
                Some("up") => "tunnel up".to_string(),
                Some("down") => "tunnel down".yellow().to_string(),
                _ if p.reachable => String::new(),
                _ => "unreachable".dimmed().to_string(),
            };
            out.push_str(&format!(
                "    {} {}  {:<24} {}{}\n",
                mark,
                p.broker.node_uri().bold(),
                name,
                tunnel,
                me
            ));
        }
    }
    if offline_without_endpoint > 0 {
        out.push_str(&format!(
            "  {} registered node(s) without a mesh endpoint (offline).\n",
            offline_without_endpoint
        ));
    }
    out
}

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

    // A valid 32-byte Ed25519 public key, base64 (any 32 bytes decode fine for
    // fingerprinting; the value is arbitrary).
    const PK_A: &str = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=";
    const PK_B: &str = "//////////////////////////////////////////8=";

    #[test]
    fn parses_roster_rows_into_fingerprints_and_endpoints() {
        let body = format!(
            r#"[{{"node_pubkey":"{PK_A}","revoked":false,"mesh_endpoint":"10.13.13.33:9000"}},
                {{"node_pubkey":"{PK_B}","revoked":true,"mesh_endpoint":null}},
                {{"node_pubkey":"not base64!!","revoked":false}}]"#
        );
        let list = parse_roster(&body);
        assert_eq!(list.len(), 2, "undecodable pubkeys are dropped");
        let a = &list[0];
        assert!(looks_like_fingerprint(&a.fp), "fp={}", a.fp);
        assert_eq!(a.endpoint.as_deref(), Some("10.13.13.33:9000"));
        assert!(!a.revoked);
        assert_eq!(
            a.fp,
            crate::broker::node_identity::fingerprint_of_pubkey_b64(PK_A).unwrap()
        );
        assert!(list[1].revoked);
        assert!(list[1].endpoint.is_none());
        assert_eq!(a.node_uri(), format!("zc://node-{}", a.fp));
    }

    #[test]
    fn fingerprint_shape() {
        assert!(looks_like_fingerprint("13a9551e881f7f8c"));
        assert!(!looks_like_fingerprint("13A9551E881F7F8C"));
        assert!(!looks_like_fingerprint("13a9551e881f7f8"));
        assert!(!looks_like_fingerprint("localhost"));
        assert!(!looks_like_fingerprint("zc-broker-1"));
    }

    #[test]
    fn probe_marks_endpointless_nodes_unreachable_without_dialing() {
        let list = vec![MeshBroker {
            fp: "0123456789abcdef".into(),
            endpoint: None,
            revoked: false,
        }];
        let probes = probe_all(list, Duration::from_millis(50));
        assert_eq!(probes.len(), 1);
        assert!(!probes[0].reachable);
        assert!(probes[0].node_name.is_none());
    }

    #[test]
    fn probe_accepts_only_a_zakuro_broker_health_body() {
        use std::io::{Read, Write};
        use std::net::TcpListener;
        // Fake "broker" that answers /health with the right service tag.
        let good = TcpListener::bind("127.0.0.1:0").unwrap();
        let good_addr = good.local_addr().unwrap();
        std::thread::spawn(move || {
            if let Ok((mut s, _)) = good.accept() {
                let mut buf = [0u8; 1024];
                let _ = s.read(&mut buf);
                let body = r#"{"status":"healthy","service":"zakuro-broker","node_name":"fake-1","mesh_tunnel":"up"}"#;
                let _ = write!(
                    s,
                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                    body.len(),
                    body
                );
            }
        });
        // Something else (a MinIO-console-shaped HTML 200) must NOT count.
        let bad = TcpListener::bind("127.0.0.1:0").unwrap();
        let bad_addr = bad.local_addr().unwrap();
        std::thread::spawn(move || {
            if let Ok((mut s, _)) = bad.accept() {
                let mut buf = [0u8; 1024];
                let _ = s.read(&mut buf);
                let body = "<!doctype html><html></html>";
                let _ = write!(
                    s,
                    "HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                    body.len(),
                    body
                );
            }
        });
        // Not a mesh IP, so the mesh agent dials directly (no proxy involved).
        let list = vec![
            MeshBroker {
                fp: "aaaaaaaaaaaaaaaa".into(),
                endpoint: Some(good_addr.to_string()),
                revoked: false,
            },
            MeshBroker {
                fp: "bbbbbbbbbbbbbbbb".into(),
                endpoint: Some(bad_addr.to_string()),
                revoked: false,
            },
        ];
        let probes = probe_all(list, Duration::from_secs(3));
        let good_p = probes
            .iter()
            .find(|p| p.broker.fp == "aaaaaaaaaaaaaaaa")
            .unwrap();
        let bad_p = probes
            .iter()
            .find(|p| p.broker.fp == "bbbbbbbbbbbbbbbb")
            .unwrap();
        assert!(good_p.reachable);
        assert_eq!(good_p.node_name.as_deref(), Some("fake-1"));
        assert_eq!(good_p.mesh_tunnel.as_deref(), Some("up"));
        assert!(!bad_p.reachable, "an HTML 200 is not a broker");
        // reachable rows sort first
        assert_eq!(probes[0].broker.fp, "aaaaaaaaaaaaaaaa");
    }

    #[test]
    fn render_is_ip_free() {
        let probes = vec![BrokerProbe {
            broker: MeshBroker {
                fp: "0123456789abcdef".into(),
                endpoint: Some("10.13.13.33:9000".into()),
                revoked: false,
            },
            reachable: true,
            node_name: Some("zc-broker-1".into()),
            mesh_tunnel: Some("up".into()),
        }];
        let text = render(&probes, Some("0123456789abcdef"), 3);
        assert!(text.contains("zc://node-0123456789abcdef"));
        assert!(text.contains("zc-broker-1"));
        assert!(text.contains("(this machine)"));
        assert!(text.contains("3 registered node(s) without a mesh endpoint"));
        assert!(
            !text.contains("10.13.13"),
            "must never print an address: {text}"
        );
    }
}