zc2 0.0.28

//! Dashboard round-trips for the zero-trust mesh: register this node's signing
//! pubkey, pull the authorized-node roster + the dashboard's voucher-signing
//! pubkey, and obtain a signed job voucher before dispatch.
//!
//! Contract (dashboard side, `zak-dashboard`):
//!   POST /api/broker/node/register  {node_pubkey, mesh_endpoint?, deploy_runtime?}
//!                                                        (X-Broker-Api-Key: zk_…)
//!   GET  /api/broker/node/roster    → [{node_pubkey, revoked}]
//!   GET  /api/broker/mesh/owner-ips → {ips, complete}   (zc#212)
//!   GET  /api/broker/voucher/pubkey → {pubkey}
//!   POST /api/broker/voucher        {budget_credits}     → {signed_json, sig}

use std::time::Duration;

fn base(api_url: &str) -> &str {
    api_url.trim_end_matches('/')
}

/// Build the JSON payload for `register_node`. Omits an optional key entirely
/// when its value is `None` — absent means "unchanged" server-side, so a
/// tunnel flap must not send an explicit `null` that would clear a
/// previously-stored address, and a caller that cannot answer whether this
/// host runs containers must not overwrite what the hub already recorded.
/// Split out from `register_node` so this contract has direct test coverage.
///
/// `deploy_runtime` is sent explicitly when known, `false` included: a host
/// whose container runtime stopped working is exactly the case the field
/// exists for, and omitting it there would leave the hub advertising the node
/// as still able to host deployments.
fn register_payload(
    node_pubkey: &str,
    mesh_endpoint: Option<&str>,
    deploy_runtime: Option<bool>,
) -> String {
    let mut body = serde_json::Map::new();
    body.insert("node_pubkey".to_string(), node_pubkey.into());
    if let Some(ep) = mesh_endpoint {
        body.insert("mesh_endpoint".to_string(), ep.into());
    }
    if let Some(available) = deploy_runtime {
        body.insert("deploy_runtime".to_string(), available.into());
    }
    serde_json::Value::Object(body).to_string()
}

/// Register this node's Ed25519 public key with the dashboard (idempotent).
///
/// `deploy_runtime` reports whether this host can run a deployment at all
/// (`deploy::runtime_available`). The hub decides broker eligibility from it:
/// without it a live broker with no usable container runtime is offered as a
/// deployment host and everything placed there sits `pending` forever. `None`
/// leaves whatever the hub already recorded untouched.
pub fn register_node(
    api_url: &str,
    api_key: &str,
    node_pubkey: &str,
    mesh_endpoint: Option<&str>,
    deploy_runtime: Option<bool>,
) -> Result<(), String> {
    let endpoint = format!("{}/api/broker/node/register", base(api_url));
    let payload = register_payload(node_pubkey, mesh_endpoint, deploy_runtime);
    let resp = ureq::post(&endpoint)
        .config()
        .timeout_global(Some(Duration::from_secs(10)))
        .http_status_as_error(false)
        .build()
        .header("X-Broker-Api-Key", api_key)
        .header("Content-Type", "application/json")
        .send(payload.as_str())
        .map_err(|e| e.to_string())?;
    let code = resp.status().as_u16();
    if (200..300).contains(&code) {
        Ok(())
    } else {
        Err(format!("register_node HTTP {code}"))
    }
}

/// Parse a roster response body: `[{"node_pubkey": "...", "revoked": bool}, …]`
/// → `(pubkey, revoked)` pairs. Unknown/malformed entries are skipped.
pub fn parse_roster(body: &str) -> Vec<(String, bool)> {
    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()?.to_string();
                    let revoked = e.get("revoked").and_then(|r| r.as_bool()).unwrap_or(false);
                    Some((pk, revoked))
                })
                .collect()
        })
        .unwrap_or_default()
}

/// Parse `mesh_endpoint`s from a roster body → `(pubkey, "ip:port")` pairs.
/// Entries with a null or absent endpoint are skipped: they are authorized
/// nodes with no tunnel up, which is normal, not malformed.
pub fn parse_endpoints(body: &str) -> Vec<(String, String)> {
    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()?.to_string();
                    let ep = e.get("mesh_endpoint")?.as_str()?.to_string();
                    Some((pk, ep))
                })
                .collect()
        })
        .unwrap_or_default()
}

/// Request the roster and read the response body (shared by `fetch_roster`
/// and `fetch_roster_full` so the HTTP round-trip is implemented once).
fn fetch_roster_body(api_url: &str, api_key: &str) -> Result<String, String> {
    let endpoint = format!("{}/api/broker/node/roster", base(api_url));
    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| e.to_string())?;
    if resp.status().as_u16() != 200 {
        return Err(format!("fetch_roster HTTP {}", resp.status().as_u16()));
    }
    resp.into_body().read_to_string().map_err(|e| e.to_string())
}

/// Fetch the authorized-node roster from the dashboard.
pub fn fetch_roster(api_url: &str, api_key: &str) -> Result<Vec<(String, bool)>, String> {
    Ok(parse_roster(&fetch_roster_body(api_url, api_key)?))
}

/// `(pubkey_b64, revoked)` authorization pairs, as `parse_roster` returns them.
pub type RosterEntries = Vec<(String, bool)>;

/// `(pubkey_b64, "ip:port")` mesh endpoints, as `parse_endpoints` returns them.
pub type RosterEndpoints = Vec<(String, String)>;

/// Fetch the roster once and return both authorization pairs and endpoints.
///
/// `fetch_roster` reads and discards the body, so a caller that also wants
/// endpoints would have to poll again. This is the tick-loop path, polled by
/// every broker, so it stays one request.
pub fn fetch_roster_full(
    api_url: &str,
    api_key: &str,
) -> Result<(RosterEntries, RosterEndpoints), String> {
    let body = fetch_roster_body(api_url, api_key)?;
    Ok((parse_roster(&body), parse_endpoints(&body)))
}

/// The mesh addresses that belong to the calling broker's owner, as the hub
/// reports them on `GET /api/broker/mesh/owner-ips` (zc#212): the owner's own
/// broker nodes plus their VPN peer. `complete` is false when the hub could
/// not look the VPN peer up; `ips` then holds the broker nodes only.
#[derive(Debug, Clone, PartialEq)]
pub struct OwnerMeshIps {
    pub ips: Vec<std::net::IpAddr>,
    pub complete: bool,
}

/// Parse `{"ips": ["10.13.13.22", …], "complete": bool}`. Entries that are not
/// IP addresses are skipped, and a missing `complete` reads as false. A body
/// without an `ips` array is `None`, never "the owner has no addresses",
/// which would lock the owner out of their own environments.
pub fn parse_owner_ips(body: &str) -> Option<OwnerMeshIps> {
    let v: serde_json::Value = serde_json::from_str(body).ok()?;
    let ips = v
        .get("ips")?
        .as_array()?
        .iter()
        .filter_map(|ip| ip.as_str()?.parse().ok())
        .collect();
    let complete = v.get("complete").and_then(|c| c.as_bool()).unwrap_or(false);
    Some(OwnerMeshIps { ips, complete })
}

/// `GET {api_url}/api/broker/mesh/owner-ips` for the owner of `api_key`.
pub fn fetch_owner_ips(api_url: &str, api_key: &str) -> Result<OwnerMeshIps, String> {
    let endpoint = format!("{}/api/broker/mesh/owner-ips", base(api_url));
    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| e.to_string())?;
    let code = resp.status().as_u16();
    if code != 200 {
        return Err(format!("fetch_owner_ips HTTP {code}"));
    }
    let body = resp
        .into_body()
        .read_to_string()
        .map_err(|e| e.to_string())?;
    parse_owner_ips(&body).ok_or_else(|| "fetch_owner_ips: unexpected response body".to_string())
}

/// Fetch the dashboard's voucher-signing public key (b64).
pub fn fetch_voucher_pubkey(api_url: &str) -> Result<String, String> {
    let endpoint = format!("{}/api/broker/voucher/pubkey", base(api_url));
    let resp = ureq::get(&endpoint)
        .config()
        .timeout_global(Some(Duration::from_secs(10)))
        .http_status_as_error(false)
        .build()
        .call()
        .map_err(|e| e.to_string())?;
    if resp.status().as_u16() != 200 {
        return Err(format!(
            "fetch_voucher_pubkey HTTP {}",
            resp.status().as_u16()
        ));
    }
    let text = resp
        .into_body()
        .read_to_string()
        .map_err(|e| e.to_string())?;
    let v: serde_json::Value = serde_json::from_str(&text).map_err(|e| e.to_string())?;
    v.get("pubkey")
        .and_then(|p| p.as_str())
        .map(|s| s.to_string())
        .ok_or_else(|| "voucher pubkey response missing `pubkey`".into())
}

/// Obtain a dashboard-signed job voucher. Returns `(signed_json, sig)` — the
/// exact bytes the dashboard signed plus its signature (see `voucher.rs`).
pub fn obtain_voucher(
    api_url: &str,
    api_key: &str,
    budget_credits: f64,
) -> Result<(String, String), String> {
    let endpoint = format!("{}/api/broker/voucher", base(api_url));
    let payload = serde_json::json!({ "budget_credits": budget_credits }).to_string();
    let resp = ureq::post(&endpoint)
        .config()
        .timeout_global(Some(Duration::from_secs(10)))
        .http_status_as_error(false)
        .build()
        .header("X-Broker-Api-Key", api_key)
        .header("Content-Type", "application/json")
        .send(payload.as_str())
        .map_err(|e| e.to_string())?;
    if resp.status().as_u16() != 200 {
        return Err(format!("obtain_voucher HTTP {}", resp.status().as_u16()));
    }
    let text = resp
        .into_body()
        .read_to_string()
        .map_err(|e| e.to_string())?;
    let v: serde_json::Value = serde_json::from_str(&text).map_err(|e| e.to_string())?;
    let signed_json = v
        .get("signed_json")
        .and_then(|s| s.as_str())
        .ok_or("voucher response missing `signed_json`")?
        .to_string();
    let sig = v
        .get("sig")
        .and_then(|s| s.as_str())
        .ok_or("voucher response missing `sig`")?
        .to_string();
    Ok((signed_json, sig))
}

/// Redeem a voucher's nonce at the dashboard (double-spend guard). Returns
/// `Ok(true)` on first redemption, `Ok(false)` if already spent. Best-effort:
/// the caller logs failures and continues (settlement rides the existing path).
pub fn redeem_voucher(
    api_url: &str,
    api_key: &str,
    task_nonce: &str,
    actual_cost: f64,
) -> Result<bool, String> {
    let endpoint = format!("{}/api/broker/voucher/redeem", base(api_url));
    let payload =
        serde_json::json!({ "task_nonce": task_nonce, "actual_cost": actual_cost }).to_string();
    let resp = ureq::post(&endpoint)
        .config()
        .timeout_global(Some(Duration::from_secs(10)))
        .http_status_as_error(false)
        .build()
        .header("X-Broker-Api-Key", api_key)
        .header("Content-Type", "application/json")
        .send(payload.as_str())
        .map_err(|e| e.to_string())?;
    if resp.status().as_u16() != 200 {
        return Err(format!("redeem_voucher HTTP {}", resp.status().as_u16()));
    }
    let text = resp
        .into_body()
        .read_to_string()
        .map_err(|e| e.to_string())?;
    let v: serde_json::Value = serde_json::from_str(&text).map_err(|e| e.to_string())?;
    Ok(v.get("redeemed").and_then(|r| r.as_bool()).unwrap_or(false))
}

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

    #[test]
    fn register_payload_omits_mesh_endpoint_key_when_none() {
        let payload = super::register_payload("PUBKEY", None, None);
        let v: serde_json::Value = serde_json::from_str(&payload).unwrap();
        assert_eq!(v["node_pubkey"], "PUBKEY");
        assert!(
            v.get("mesh_endpoint").is_none(),
            "mesh_endpoint key must be absent, not null: {payload}"
        );
        assert!(
            !payload.contains("mesh_endpoint"),
            "payload must not mention mesh_endpoint at all: {payload}"
        );
    }

    #[test]
    fn register_payload_includes_exact_mesh_endpoint_when_some() {
        let payload = super::register_payload("PUBKEY", Some("10.13.13.6:9000"), None);
        let v: serde_json::Value = serde_json::from_str(&payload).unwrap();
        assert_eq!(v["node_pubkey"], "PUBKEY");
        assert_eq!(v["mesh_endpoint"], "10.13.13.6:9000");
    }

    #[test]
    fn register_payload_reports_a_usable_deploy_runtime() {
        let payload = super::register_payload("PUBKEY", Some("10.13.13.6:9000"), Some(true));
        let v: serde_json::Value = serde_json::from_str(&payload).unwrap();
        assert_eq!(v["node_pubkey"], "PUBKEY");
        assert_eq!(v["mesh_endpoint"], "10.13.13.6:9000");
        assert_eq!(v["deploy_runtime"], true);
    }

    #[test]
    fn register_payload_reports_a_missing_deploy_runtime_as_explicit_false() {
        // The case the field exists for: a live broker with no docker must
        // say so, not stay silent and keep being offered as a deploy host.
        let payload = super::register_payload("PUBKEY", None, Some(false));
        let v: serde_json::Value = serde_json::from_str(&payload).unwrap();
        assert_eq!(
            v.get("deploy_runtime"),
            Some(&serde_json::Value::Bool(false)),
            "deploy_runtime=false must be sent explicitly: {payload}"
        );
    }

    #[test]
    fn register_payload_omits_deploy_runtime_when_unknown() {
        // Absent means "unchanged" server-side, same rule `mesh_endpoint`
        // follows — an unknown value must not clear a stored one.
        let payload = super::register_payload("PUBKEY", None, None);
        assert!(
            !payload.contains("deploy_runtime"),
            "payload must not mention deploy_runtime at all: {payload}"
        );
    }

    #[test]
    fn parse_roster_reads_pairs_and_defaults_revoked() {
        let body = r#"[
            {"node_pubkey":"AAA","revoked":true},
            {"node_pubkey":"BBB"},
            {"nope":1}
        ]"#;
        let r = parse_roster(body);
        assert_eq!(r.len(), 2);
        assert_eq!(r[0], ("AAA".to_string(), true));
        assert_eq!(r[1], ("BBB".to_string(), false));
    }

    #[test]
    fn parse_roster_handles_garbage() {
        assert!(parse_roster("not json").is_empty());
        assert!(parse_roster("{}").is_empty());
    }

    #[test]
    fn parses_endpoints_skipping_absent_and_null() {
        let body = r#"[
            {"node_pubkey":"A","revoked":false,"mesh_endpoint":"10.13.13.6:9000"},
            {"node_pubkey":"B","revoked":false,"mesh_endpoint":null},
            {"node_pubkey":"C","revoked":false}
        ]"#;
        assert_eq!(
            super::parse_endpoints(body),
            vec![("A".to_string(), "10.13.13.6:9000".to_string())]
        );
    }

    #[test]
    fn parse_roster_still_reads_bodies_without_endpoints() {
        // An older dashboard omits the field entirely; authorization must
        // keep working.
        let body = r#"[{"node_pubkey":"A","revoked":false}]"#;
        assert_eq!(super::parse_roster(body), vec![("A".to_string(), false)]);
        assert!(super::parse_endpoints(body).is_empty());
    }

    #[test]
    fn parse_owner_ips_reads_addresses_and_completeness() {
        let got =
            parse_owner_ips(r#"{"ips":["10.13.13.22","10.13.13.5"],"complete":true}"#).unwrap();
        let want: Vec<std::net::IpAddr> = vec![
            "10.13.13.22".parse().unwrap(),
            "10.13.13.5".parse().unwrap(),
        ];
        assert_eq!(got.ips, want);
        assert!(got.complete);
    }

    #[test]
    fn parse_owner_ips_skips_entries_that_are_not_addresses() {
        let got = parse_owner_ips(r#"{"ips":["10.13.13.22","zakuro0",7,null],"complete":false}"#)
            .unwrap();
        assert_eq!(got.ips.len(), 1);
        assert!(!got.complete);
    }

    #[test]
    fn parse_owner_ips_reads_a_missing_complete_as_incomplete() {
        assert!(!parse_owner_ips(r#"{"ips":[]}"#).unwrap().complete);
    }

    #[test]
    fn parse_owner_ips_rejects_bodies_without_an_ips_array() {
        for body in ["not json", "{}", r#"{"ips":"10.13.13.22"}"#, "[]"] {
            assert_eq!(parse_owner_ips(body), None, "{body}");
        }
    }

    use std::io::{Read, Write};
    use std::net::TcpListener;

    /// Minimal mock: serve a fixed JSON body for one request, capture the request line.
    fn mock_once(status: u16, body: &'static str) -> (std::thread::JoinHandle<String>, u16) {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        let h = std::thread::spawn(move || {
            let (mut s, _) = listener.accept().unwrap();
            // read until headers + full Content-Length body are in (a single read
            // can return before the body bytes arrive)
            let mut raw = Vec::new();
            let mut buf = [0u8; 2048];
            loop {
                let n = s.read(&mut buf).unwrap_or(0);
                if n == 0 {
                    break;
                }
                raw.extend_from_slice(&buf[..n]);
                let text = String::from_utf8_lossy(&raw);
                if let Some(hdr_end) = text.find("\r\n\r\n") {
                    let want: usize = text
                        .lines()
                        .find_map(|l| {
                            l.to_ascii_lowercase()
                                .strip_prefix("content-length:")
                                .map(|v| v.trim().parse().unwrap_or(0))
                        })
                        .unwrap_or(0);
                    if raw.len() >= hdr_end + 4 + want {
                        break;
                    }
                }
            }
            let req = String::from_utf8_lossy(&raw).to_string();
            let resp = format!(
                "HTTP/1.1 {status} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
                body.len()
            );
            let _ = s.write_all(resp.as_bytes());
            req
        });
        (h, port)
    }

    // Network round-trip tests: reliable in isolation, but the blocking mock
    // server can starve under the full suite's fd/CPU pressure. Run explicitly
    // with `cargo test -- --ignored`. The parse logic above is always covered.
    #[test]
    #[ignore = "network mock; run with --ignored"]
    fn fetch_roster_parses_live_response() {
        let (h, port) = mock_once(
            200,
            r#"[{"node_pubkey":"AAA","revoked":false},{"node_pubkey":"BBB","revoked":true}]"#,
        );
        let url = format!("http://127.0.0.1:{port}");
        let roster = fetch_roster(&url, "zk_1_x").unwrap();
        let req = h.join().unwrap();
        assert!(req.starts_with("GET /api/broker/node/roster"));
        assert!(req
            .to_ascii_lowercase()
            .contains("x-broker-api-key: zk_1_x"));
        assert_eq!(roster.len(), 2);
        assert_eq!(roster[1], ("BBB".to_string(), true));
    }

    #[test]
    #[ignore = "network mock; run with --ignored"]
    fn obtain_voucher_extracts_signed_json_and_sig() {
        let (h, port) = mock_once(200, r#"{"signed_json":"{\"v\":1}","sig":"deadbeef"}"#);
        let url = format!("http://127.0.0.1:{port}");
        let (signed, sig) = obtain_voucher(&url, "zk_1_x", 5.0).unwrap();
        let req = h.join().unwrap();
        assert!(req.starts_with("POST /api/broker/voucher"));
        assert!(req.contains("budget_credits"));
        assert_eq!(signed, "{\"v\":1}");
        assert_eq!(sig, "deadbeef");
    }
}