zc2 0.0.27

//! Mesh exposure (zc#209): makes a healthy deployment reachable from the
//! other hosts on the WireGuard mesh.
//!
//! `runtime::DockerCli::run` publishes no ports (see `runtime`'s module doc
//! comment), so the address `drive()` records, the container's own `ip:port`
//! on `deploy_network()`, is reachable only from this host and that docker
//! network. Publishing on the mesh IP instead (`-p <mesh_ip>:…`) would only
//! work for a host-native broker: a containerised broker's mesh IP lives in
//! its own netns, where the docker daemon cannot bind it. So the broker
//! forwards: it listens on its mesh IP and splices raw TCP to the container.
//! That works in both topologies, because the broker owns its mesh IP either
//! way and already reaches the container IP (it health-checks it). Raw TCP
//! rather than the `/serve` HTTP proxy, so WebSockets (Jupyter kernels) and
//! absolute redirects pass through unchanged.
//!
//! Listeners bind the mesh IP only, never `0.0.0.0`: nothing is exposed on
//! the host's LAN or WAN interfaces. They admit only the environment's owner
//! (zc#212, see `MeshAllowlist`); anyone else uses `/serve`, the billed path.
//! `ZAKURO_MESH_EXPOSE=0` turns exposure off entirely.

use std::collections::{HashMap, HashSet};
use std::io::ErrorKind;
use std::net::{IpAddr, Shutdown, SocketAddr, TcpListener, TcpStream};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, MutexGuard, OnceLock, RwLock};
use std::thread::JoinHandle;
use std::time::{Duration, Instant};

use super::reconcile::StatusReporter;
use super::state::{DeploymentRecord, LocalState};
use crate::broker::node_sync::OwnerMeshIps;

/// How often an accept loop checks whether its deployment was released.
const ACCEPT_POLL: Duration = Duration::from_millis(100);

/// How long a forwarded connection waits for the container to accept it.
const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);

/// Where `sync` makes deployments reachable. A trait so `sync` is unit tested
/// against a fake, the same way `drive()` fakes `ContainerRuntime`.
pub trait MeshExposer {
    /// Forward `listen_ip:<port>` to `target` (`container_ip:port`) for
    /// deployment `id`, trying `preferred_port` first and an OS-assigned port
    /// when that one is taken. An exposure of `id` that already listens on
    /// `listen_ip` for `target` is kept as is; one that points elsewhere is
    /// moved, keeping its port. Returns the `ip:port` now listening, or
    /// `None` when nothing could be bound.
    fn expose(
        &self,
        id: &str,
        listen_ip: &str,
        target: &str,
        preferred_port: u16,
    ) -> Option<String>;

    /// Stop listening for `id`. A no-op when `id` isn't exposed.
    fn release(&self, id: &str);

    /// Every deployment id currently exposed.
    fn exposed_ids(&self) -> Vec<String>;
}

/// One exposed deployment: the listener's address, where it forwards to, and
/// the accept loop that owns the listener.
struct Forward {
    listen: SocketAddr,
    target: String,
    stop: Arc<AtomicBool>,
    accept_loop: JoinHandle<()>,
}

impl Forward {
    /// Stop accepting and wait for the accept loop to drop the listener, so
    /// the port is free again when this returns. Connections already
    /// forwarded run until either side closes them, which the container does
    /// once it is stopped or replaced.
    fn close(self) {
        self.stop.store(true, Ordering::Relaxed);
        let _ = self.accept_loop.join();
    }
}

/// Decides whether a connection from `peer` may use a forwarder that listens
/// on `own`.
type Policy = Arc<dyn Fn(IpAddr, IpAddr) -> bool + Send + Sync>;

/// The real `MeshExposer`: one accept thread per exposed deployment and two
/// copy threads per forwarded connection. Environments see a handful of
/// interactive connections, not load, and threads keep this usable from the
/// synchronous reconciler tick without an async runtime.
pub struct TcpForwarder {
    forwards: Mutex<HashMap<String, Forward>>,
    policy: Policy,
}

impl Default for TcpForwarder {
    /// Admits only this host's own connections: the fail-closed floor. The
    /// broker's forwarder (`forwarder()`) also admits the owner's addresses.
    fn default() -> Self {
        TcpForwarder::with_policy(|peer, own| peer == own)
    }
}

impl TcpForwarder {
    /// A forwarder that lets through only the connections `policy(peer, own)`
    /// accepts, `own` being the address it listens on.
    pub fn with_policy(policy: impl Fn(IpAddr, IpAddr) -> bool + Send + Sync + 'static) -> Self {
        TcpForwarder {
            forwards: Mutex::default(),
            policy: Arc::new(policy),
        }
    }

    fn forwards(&self) -> MutexGuard<'_, HashMap<String, Forward>> {
        self.forwards
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }
}

impl MeshExposer for TcpForwarder {
    fn expose(
        &self,
        id: &str,
        listen_ip: &str,
        target: &str,
        preferred_port: u16,
    ) -> Option<String> {
        let ip: IpAddr = listen_ip.parse().ok()?;
        let mut forwards = self.forwards();
        let mut port = preferred_port;
        if let Some(existing) = forwards.remove(id) {
            if existing.target == target && existing.listen.ip() == ip {
                let endpoint = existing.listen.to_string();
                forwards.insert(id.to_string(), existing);
                return Some(endpoint);
            }
            // A new container (a version bump, a restart) or a new mesh IP:
            // keep the port, so the address the hub shows stays the same.
            port = existing.listen.port();
            existing.close();
        }
        let listener = bind(ip, port)?;
        let listen = listener.local_addr().ok()?;
        let stop = Arc::new(AtomicBool::new(false));
        let accept_loop = {
            let stop = Arc::clone(&stop);
            let policy = Arc::clone(&self.policy);
            let target = target.to_string();
            std::thread::Builder::new()
                .name(format!("mesh-fwd-{id}"))
                .spawn(move || accept_loop(listener, target, stop, policy))
                .ok()?
        };
        forwards.insert(
            id.to_string(),
            Forward {
                listen,
                target: target.to_string(),
                stop,
                accept_loop,
            },
        );
        Some(listen.to_string())
    }

    fn release(&self, id: &str) {
        let released = self.forwards().remove(id);
        if let Some(forward) = released {
            forward.close();
        }
    }

    fn exposed_ids(&self) -> Vec<String> {
        let mut ids: Vec<String> = self.forwards().keys().cloned().collect();
        ids.sort();
        ids
    }
}

/// Listen on `ip:preferred_port`, or on an OS-assigned port when that one is
/// taken. Non-blocking, so the accept loop notices a release.
fn bind(ip: IpAddr, preferred_port: u16) -> Option<TcpListener> {
    let listener =
        match TcpListener::bind((ip, preferred_port)).or_else(|_| TcpListener::bind((ip, 0))) {
            Ok(listener) => listener,
            Err(e) => {
                eprintln!("  [DEPLOY] cannot listen on mesh IP {ip}: {e}");
                return None;
            }
        };
    listener.set_nonblocking(true).ok()?;
    Some(listener)
}

/// Accept connections until `stop` is set, forwarding each one `policy`
/// admits to `target` on its own thread and closing the rest at once. Owns
/// the listener, so returning frees the port.
fn accept_loop(listener: TcpListener, target: String, stop: Arc<AtomicBool>, policy: Policy) {
    let own = listener.local_addr().map(|addr| addr.ip()).ok();
    while !stop.load(Ordering::Relaxed) {
        match listener.accept() {
            Ok((client, peer)) => {
                if !own.is_some_and(|own| policy(peer.ip(), own)) {
                    log_refusal(peer.ip(), &target);
                    drop(client);
                    continue;
                }
                let conn_target = target.clone();
                let spawned = std::thread::Builder::new()
                    .name("mesh-fwd-conn".to_string())
                    .spawn(move || splice(client, &conn_target));
                if let Err(e) = spawned {
                    eprintln!(
                        "  [DEPLOY] mesh forward to {target}: no thread for a connection: {e}"
                    );
                }
            }
            Err(e) if e.kind() == ErrorKind::WouldBlock => std::thread::sleep(ACCEPT_POLL),
            Err(e) => {
                eprintln!("  [DEPLOY] mesh forward to {target}: accept failed: {e}");
                std::thread::sleep(ACCEPT_POLL);
            }
        }
    }
}

/// How often a refused source address is logged, at most.
const REFUSAL_LOG_EVERY: Duration = Duration::from_secs(60);

/// Log a refused connection, at most once a minute per source address, so a
/// peer hammering the port can't flood the broker's log.
fn log_refusal(peer: IpAddr, target: &str) {
    static LOGGED: OnceLock<Mutex<HashMap<IpAddr, Instant>>> = OnceLock::new();
    let mut logged = LOGGED
        .get_or_init(Mutex::default)
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner());
    let now = Instant::now();
    if logged
        .get(&peer)
        .is_some_and(|at| now.duration_since(*at) < REFUSAL_LOG_EVERY)
    {
        return;
    }
    logged.insert(peer, now);
    eprintln!(
        "  [DEPLOY] mesh forward to {target}: refused {peer}, not one of the owner's addresses (others use /serve)"
    );
}

/// Copy bytes both ways until each side has finished sending, half-closing
/// the other side as each direction ends, so a client that shuts its write
/// side after a request still receives the whole response.
fn splice(client: TcpStream, target: &str) {
    // An accepted socket inherits the listener's non-blocking flag on BSDs.
    if client.set_nonblocking(false).is_err() {
        return;
    }
    let Ok(addr) = target.parse::<SocketAddr>() else {
        return;
    };
    let upstream = match TcpStream::connect_timeout(&addr, CONNECT_TIMEOUT) {
        Ok(upstream) => upstream,
        Err(e) => {
            eprintln!("  [DEPLOY] mesh forward to {target} failed: {e}");
            return;
        }
    };
    let (Ok(mut from_client), Ok(mut from_upstream)) = (client.try_clone(), upstream.try_clone())
    else {
        return;
    };
    let (mut to_client, mut to_upstream) = (client, upstream);
    let downstream = std::thread::spawn(move || {
        let _ = std::io::copy(&mut from_upstream, &mut to_client);
        let _ = to_client.shutdown(Shutdown::Write);
    });
    let _ = std::io::copy(&mut from_client, &mut to_upstream);
    let _ = to_upstream.shutdown(Shutdown::Write);
    let _ = downstream.join();
}

/// The broker's one `TcpForwarder`, admitting the owner's addresses (see
/// `allowlist()`). Exposures have to outlive a reconcile tick (`deploy::tick`
/// loads `LocalState` afresh each time), so they live here rather than in the
/// tick.
pub fn forwarder() -> &'static TcpForwarder {
    static FORWARDER: OnceLock<TcpForwarder> = OnceLock::new();
    FORWARDER.get_or_init(|| TcpForwarder::with_policy(|peer, own| allowlist().allows(peer, own)))
}

/// The address to expose deployments on: this host's mesh IP, unless the
/// operator set `ZAKURO_MESH_EXPOSE` to `0`, `false`, `off` or `no`.
pub fn exposure_ip(mesh_ip: Option<String>, mesh_expose: Option<&str>) -> Option<String> {
    let disabled = mesh_expose.is_some_and(|value| {
        matches!(
            value.trim().to_ascii_lowercase().as_str(),
            "0" | "false" | "off" | "no"
        )
    });
    mesh_ip.filter(|_| !disabled)
}

/// Who may connect through the mesh forwarder (zc#212): the environment's
/// owner only. Deployments are owner-scoped, so a broker runs only its
/// owner's environments, and the owner's addresses are what the hub reports
/// on `/api/broker/mesh/owner-ips`: their own broker nodes plus their VPN
/// peer. This broker's own address is always allowed. Everyone else uses the
/// billed `/serve` path.
///
/// Fails closed: until an answer has been learned, from the hub or from the
/// copy an earlier run cached in `mesh_allowlist.json`, only this broker's own
/// address may connect.
#[derive(Debug, Default)]
pub struct MeshAllowlist {
    ips: RwLock<Option<HashSet<IpAddr>>>,
}

impl MeshAllowlist {
    /// Whether `peer` may use a forwarder listening on `own`.
    pub fn allows(&self, peer: IpAddr, own: IpAddr) -> bool {
        peer == own || self.read().as_ref().is_some_and(|ips| ips.contains(&peer))
    }

    /// Take the hub's answer; returns whether the allowlist changed. A
    /// complete answer replaces it. An incomplete one (the hub could not reach
    /// the VPN manager, so it lists broker nodes only) is taken only while
    /// nothing is known: dropping the owner's VPN peer during a manager outage
    /// would lock them out of their own environments.
    pub fn apply(&self, answer: &OwnerMeshIps) -> bool {
        let mut ips = self.write();
        if !answer.complete && ips.is_some() {
            return false;
        }
        let fresh: HashSet<IpAddr> = answer.ips.iter().copied().collect();
        if ips.as_ref() == Some(&fresh) {
            return false;
        }
        *ips = Some(fresh);
        true
    }

    /// Best-effort write to `{credentials::dir()}/mesh_allowlist.json`, so a
    /// restarted broker admits the owner before it next reaches the hub.
    /// Nothing is written while nothing is known.
    pub fn persist(&self) {
        let Some(ips) = self.read().clone() else {
            return;
        };
        let Some(path) = Self::path() else {
            return;
        };
        let mut list: Vec<String> = ips.iter().map(IpAddr::to_string).collect();
        list.sort();
        let Ok(json) = serde_json::to_string(&serde_json::json!({ "ips": list })) else {
            return;
        };
        if let Some(dir) = path.parent() {
            let _ = std::fs::create_dir_all(dir);
        }
        let _ = std::fs::write(&path, json);
    }

    /// The cached allowlist, or an unknown (fail-closed) one when there is no
    /// readable cache.
    pub fn load() -> MeshAllowlist {
        let ips = Self::path()
            .and_then(|path| std::fs::read_to_string(path).ok())
            .and_then(|text| serde_json::from_str::<serde_json::Value>(&text).ok())
            .and_then(|v| {
                let ips = v.get("ips")?.as_array()?;
                Some(
                    ips.iter()
                        .filter_map(|ip| ip.as_str()?.parse().ok())
                        .collect::<HashSet<IpAddr>>(),
                )
            });
        MeshAllowlist {
            ips: RwLock::new(ips),
        }
    }

    fn path() -> Option<std::path::PathBuf> {
        crate::credentials::dir().map(|dir| dir.join("mesh_allowlist.json"))
    }

    fn read(&self) -> std::sync::RwLockReadGuard<'_, Option<HashSet<IpAddr>>> {
        self.ips
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }

    fn write(&self) -> std::sync::RwLockWriteGuard<'_, Option<HashSet<IpAddr>>> {
        self.ips
            .write()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }
}

/// The broker's one allowlist, loaded from its cache on first use.
pub fn allowlist() -> &'static MeshAllowlist {
    static ALLOWLIST: OnceLock<MeshAllowlist> = OnceLock::new();
    ALLOWLIST.get_or_init(MeshAllowlist::load)
}

/// How often the allowlist is refreshed from the hub, at most.
const ALLOWLIST_REFRESH_EVERY: Duration = Duration::from_secs(60);

/// Refresh the allowlist from the hub, at most once a minute, and cache it. A
/// failed refresh keeps what is known, so the forwarder fails closed only when
/// nothing ever was. Called from `deploy::tick`.
pub fn refresh_allowlist(api_url: &str, api_key: &str) {
    static LAST: Mutex<Option<Instant>> = Mutex::new(None);
    {
        let mut last = LAST.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
        if last.is_some_and(|at| at.elapsed() < ALLOWLIST_REFRESH_EVERY) {
            return;
        }
        *last = Some(Instant::now());
    }
    match crate::broker::node_sync::fetch_owner_ips(api_url, api_key) {
        Ok(answer) => {
            let list = allowlist();
            if list.apply(&answer) {
                list.persist();
            }
        }
        Err(e) => eprintln!("  [DEPLOY] could not refresh the mesh owner allowlist: {e}"),
    }
}

/// Bring mesh exposure in line with `local`. Every `healthy` record with a
/// resolved container `ip`/`port` is exposed on `mesh_ip` (the same
/// `phase == "healthy"` gate `/serve` applies); every other exposure is
/// released, and `mesh_ip: None` releases them all. When the endpoint a
/// record ends up with differs from what the hub last received
/// (`DeploymentRecord::mesh_endpoint`), the record is updated and, for a
/// healthy deployment, re-reported, so the console shows the address that
/// is actually listening. Returns whether `local` changed.
pub fn sync(
    local: &mut LocalState,
    exposer: &dyn MeshExposer,
    mesh_ip: Option<&str>,
    reporter: &dyn StatusReporter,
) -> bool {
    for id in exposer.exposed_ids() {
        let keep = mesh_ip.is_some() && local.deployments.get(&id).and_then(target_of).is_some();
        if !keep {
            exposer.release(&id);
        }
    }

    let mut ids: Vec<String> = local.deployments.keys().cloned().collect();
    ids.sort();
    let mut changed = false;
    for id in ids {
        let Some(rec) = local.deployments.get_mut(&id) else {
            continue;
        };
        let exposed = match (mesh_ip, target_of(rec)) {
            (Some(ip), Some((target, container_port))) => {
                exposer.expose(&id, ip, &target, rec.mesh_port.unwrap_or(container_port))
            }
            _ => None,
        };
        if exposed == rec.mesh_endpoint {
            continue;
        }
        if let Some(port) = exposed.as_deref().and_then(port_of) {
            rec.mesh_port = Some(port);
        }
        rec.mesh_endpoint = exposed;
        changed = true;
        if rec.phase == "healthy" {
            reporter.report(
                &id,
                rec.version,
                "healthy",
                "",
                1,
                rec.container_id.as_deref(),
                rec.endpoint.as_deref(),
                rec.mesh_endpoint.as_deref(),
            );
        }
    }
    changed
}

/// Where a record's container can be reached, when it should be exposed at
/// all: only a `healthy` record with a resolved container `ip` and `port`
/// qualifies, the same gate `/serve` applies.
fn target_of(rec: &DeploymentRecord) -> Option<(String, u16)> {
    if rec.phase != "healthy" {
        return None;
    }
    let ip: IpAddr = rec.ip.as_deref()?.parse().ok()?;
    let port = rec.port?;
    Some((SocketAddr::new(ip, port).to_string(), port))
}

/// The port half of an `ip:port` endpoint.
fn port_of(endpoint: &str) -> Option<u16> {
    endpoint.parse::<SocketAddr>().ok().map(|addr| addr.port())
}

/// Test double for `MeshExposer`, shared with `reconcile`'s tests. Mirrors
/// `TcpForwarder`'s contract: a re-pointed exposure keeps its port.
#[cfg(test)]
#[derive(Default)]
pub(crate) struct FakeExposer {
    /// id → (listen_ip, target, port)
    exposed: Mutex<HashMap<String, (String, String, u16)>>,
}

#[cfg(test)]
impl FakeExposer {
    pub(crate) fn target_of(&self, id: &str) -> Option<String> {
        self.exposed
            .lock()
            .unwrap()
            .get(id)
            .map(|(_, t, _)| t.clone())
    }

    pub(crate) fn port_of(&self, id: &str) -> Option<u16> {
        self.exposed.lock().unwrap().get(id).map(|(_, _, p)| *p)
    }
}

#[cfg(test)]
impl MeshExposer for FakeExposer {
    fn expose(
        &self,
        id: &str,
        listen_ip: &str,
        target: &str,
        preferred_port: u16,
    ) -> Option<String> {
        let mut exposed = self.exposed.lock().unwrap();
        let port = exposed.get(id).map_or(preferred_port, |(_, _, p)| *p);
        exposed.insert(
            id.to_string(),
            (listen_ip.to_string(), target.to_string(), port),
        );
        Some(format!("{listen_ip}:{port}"))
    }

    fn release(&self, id: &str) {
        self.exposed.lock().unwrap().remove(id);
    }

    fn exposed_ids(&self) -> Vec<String> {
        let mut ids: Vec<String> = self.exposed.lock().unwrap().keys().cloned().collect();
        ids.sort();
        ids
    }
}

/// Points `ZAKURO_HOME` at a throwaway directory for the life of a test, so
/// a test that saves `LocalState` never overwrites the real
/// `~/.zakuro/deployments.json` of whoever runs the suite (on a broker host,
/// the live broker's). Holds `HOME_ENV_LOCK` like `state`'s tests do.
#[cfg(test)]
pub(crate) struct TempZakuroHome {
    dir: std::path::PathBuf,
    prev: Option<std::ffi::OsString>,
    _lock: MutexGuard<'static, ()>,
}

#[cfg(test)]
impl TempZakuroHome {
    pub(crate) fn new() -> Self {
        let lock = crate::credentials::HOME_ENV_LOCK
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let dir = std::env::temp_dir().join(format!("zc-expose-test-{}", uuid::Uuid::new_v4()));
        let prev = std::env::var_os("ZAKURO_HOME");
        std::env::set_var("ZAKURO_HOME", &dir);
        TempZakuroHome {
            dir,
            prev,
            _lock: lock,
        }
    }
}

#[cfg(test)]
impl Drop for TempZakuroHome {
    fn drop(&mut self) {
        match self.prev.take() {
            Some(v) => std::env::set_var("ZAKURO_HOME", v),
            None => std::env::remove_var("ZAKURO_HOME"),
        }
        let _ = std::fs::remove_dir_all(&self.dir);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::{Read, Write};

    // --- TcpForwarder: real sockets on 127.0.0.1 ----------------------------

    /// A target that answers each connection with `prefix` followed by
    /// everything the client sent, once the client has finished sending.
    fn echo_server(prefix: &'static str) -> String {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        std::thread::spawn(move || {
            for conn in listener.incoming() {
                let Ok(mut conn) = conn else { continue };
                std::thread::spawn(move || {
                    let mut got = Vec::new();
                    let _ = conn.read_to_end(&mut got);
                    let _ = conn.write_all(prefix.as_bytes());
                    let _ = conn.write_all(&got);
                });
            }
        });
        addr.to_string()
    }

    /// Send `msg` to `endpoint`, finish sending, and return the whole reply.
    fn roundtrip(endpoint: &str, msg: &[u8]) -> Vec<u8> {
        let mut conn = TcpStream::connect(endpoint).unwrap();
        conn.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
        conn.write_all(msg).unwrap();
        conn.shutdown(Shutdown::Write).unwrap();
        let mut reply = Vec::new();
        conn.read_to_end(&mut reply).unwrap();
        reply
    }

    fn free_port() -> u16 {
        TcpListener::bind("127.0.0.1:0")
            .unwrap()
            .local_addr()
            .unwrap()
            .port()
    }

    #[test]
    fn forwarder_splices_both_directions_to_the_target() {
        let fwd = TcpForwarder::default();
        let endpoint = fwd
            .expose("dep_1", "127.0.0.1", &echo_server("echo:"), free_port())
            .expect("exposed");
        assert!(endpoint.starts_with("127.0.0.1:"), "{endpoint}");
        assert_eq!(roundtrip(&endpoint, b"hello"), b"echo:hello");
        fwd.release("dep_1");
    }

    #[test]
    fn forwarder_listens_on_the_preferred_port_when_it_is_free() {
        let fwd = TcpForwarder::default();
        let port = free_port();
        let endpoint = fwd.expose("dep_1", "127.0.0.1", &echo_server(""), port);
        assert_eq!(endpoint, Some(format!("127.0.0.1:{port}")));
        fwd.release("dep_1");
    }

    #[test]
    fn forwarder_falls_back_to_another_port_when_the_preferred_one_is_taken() {
        let taken = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = taken.local_addr().unwrap().port();
        let fwd = TcpForwarder::default();
        let endpoint = fwd
            .expose("dep_1", "127.0.0.1", &echo_server("echo:"), port)
            .expect("exposed");
        assert_ne!(endpoint, format!("127.0.0.1:{port}"));
        assert_eq!(roundtrip(&endpoint, b"x"), b"echo:x");
        fwd.release("dep_1");
    }

    #[test]
    fn forwarder_keeps_an_unchanged_exposure() {
        let fwd = TcpForwarder::default();
        let target = echo_server("");
        let first = fwd.expose("dep_1", "127.0.0.1", &target, free_port());
        let second = fwd.expose("dep_1", "127.0.0.1", &target, free_port());
        assert!(first.is_some());
        assert_eq!(first, second);
        assert_eq!(fwd.exposed_ids(), vec!["dep_1".to_string()]);
        fwd.release("dep_1");
    }

    #[test]
    fn forwarder_repointed_at_a_new_target_keeps_its_port() {
        let fwd = TcpForwarder::default();
        let first = fwd
            .expose("dep_1", "127.0.0.1", &echo_server("old:"), free_port())
            .expect("exposed");
        let second = fwd
            .expose("dep_1", "127.0.0.1", &echo_server("new:"), free_port())
            .expect("re-exposed");
        assert_eq!(first, second);
        assert_eq!(roundtrip(&second, b"x"), b"new:x");
        fwd.release("dep_1");
    }

    #[test]
    fn forwarder_release_frees_the_port() {
        let fwd = TcpForwarder::default();
        let endpoint = fwd
            .expose("dep_1", "127.0.0.1", &echo_server(""), free_port())
            .expect("exposed");
        fwd.release("dep_1");
        assert!(fwd.exposed_ids().is_empty());
        assert!(
            TcpStream::connect(&endpoint).is_err(),
            "still listening on {endpoint}"
        );
    }

    #[test]
    fn forwarder_refuses_a_listen_address_that_is_not_an_ip() {
        let fwd = TcpForwarder::default();
        assert_eq!(
            fwd.expose("dep_1", "zakuro0", &echo_server(""), free_port()),
            None
        );
        assert!(fwd.exposed_ids().is_empty());
    }

    // --- sync: a FakeExposer and a reporter that records mesh endpoints -----

    /// (id, phase, mesh_endpoint) per report, in order.
    #[derive(Default)]
    struct Reports(Mutex<Vec<(String, String, Option<String>)>>);

    impl Reports {
        fn taken(&self) -> Vec<(String, String, Option<String>)> {
            std::mem::take(&mut *self.0.lock().unwrap())
        }
    }

    impl StatusReporter for Reports {
        fn report(
            &self,
            id: &str,
            _version: u64,
            phase: &str,
            _message: &str,
            _attempt: u32,
            _container_id: Option<&str>,
            _endpoint: Option<&str>,
            mesh_endpoint: Option<&str>,
        ) {
            self.0.lock().unwrap().push((
                id.to_string(),
                phase.to_string(),
                mesh_endpoint.map(str::to_string),
            ));
        }
    }

    fn report(
        id: &str,
        phase: &str,
        mesh_endpoint: Option<&str>,
    ) -> (String, String, Option<String>) {
        (
            id.to_string(),
            phase.to_string(),
            mesh_endpoint.map(str::to_string),
        )
    }

    fn healthy(ip: &str, port: u16) -> DeploymentRecord {
        DeploymentRecord {
            version: 1,
            container_id: Some("c1".to_string()),
            phase: "healthy".to_string(),
            ip: Some(ip.to_string()),
            port: Some(port),
            endpoint: Some(format!("{ip}:{port}")),
            ..Default::default()
        }
    }

    #[test]
    fn sync_exposes_a_healthy_deployment_and_reports_where() {
        let exposer = FakeExposer::default();
        let reports = Reports::default();
        let mut local = LocalState::default();
        local
            .deployments
            .insert("dep_1".into(), healthy("172.17.0.3", 8888));

        assert!(sync(&mut local, &exposer, Some("10.13.13.22"), &reports));

        assert_eq!(
            exposer.target_of("dep_1").as_deref(),
            Some("172.17.0.3:8888")
        );
        let rec = &local.deployments["dep_1"];
        assert_eq!(rec.mesh_endpoint.as_deref(), Some("10.13.13.22:8888"));
        assert_eq!(rec.mesh_port, Some(8888));
        assert_eq!(
            reports.taken(),
            vec![report("dep_1", "healthy", Some("10.13.13.22:8888"))]
        );
    }

    #[test]
    fn sync_reports_nothing_when_nothing_changed() {
        let exposer = FakeExposer::default();
        let reports = Reports::default();
        let mut local = LocalState::default();
        local
            .deployments
            .insert("dep_1".into(), healthy("172.17.0.3", 8888));
        sync(&mut local, &exposer, Some("10.13.13.22"), &reports);
        reports.taken();

        assert!(!sync(&mut local, &exposer, Some("10.13.13.22"), &reports));
        assert!(reports.taken().is_empty());
    }

    #[test]
    fn sync_reuses_the_port_a_previous_exposure_listened_on() {
        let exposer = FakeExposer::default();
        let mut rec = healthy("172.17.0.3", 8888);
        rec.mesh_port = Some(40123);
        let mut local = LocalState::default();
        local.deployments.insert("dep_1".into(), rec);

        sync(
            &mut local,
            &exposer,
            Some("10.13.13.22"),
            &Reports::default(),
        );

        assert_eq!(exposer.port_of("dep_1"), Some(40123));
        assert_eq!(
            local.deployments["dep_1"].mesh_endpoint.as_deref(),
            Some("10.13.13.22:40123")
        );
    }

    #[test]
    fn sync_releases_a_deployment_that_is_no_longer_healthy() {
        let exposer = FakeExposer::default();
        exposer.expose("dep_1", "10.13.13.22", "172.17.0.3:8888", 8888);
        let reports = Reports::default();
        // drive() already reported `stopped`, which carried no mesh endpoint.
        let mut rec = healthy("172.17.0.3", 8888);
        rec.phase = "stopped".to_string();
        let mut local = LocalState::default();
        local.deployments.insert("dep_1".into(), rec);

        sync(&mut local, &exposer, Some("10.13.13.22"), &reports);

        assert!(exposer.exposed_ids().is_empty());
        assert!(reports.taken().is_empty());
    }

    #[test]
    fn sync_releases_deployments_this_node_no_longer_tracks() {
        let exposer = FakeExposer::default();
        exposer.expose("dep_gone", "10.13.13.22", "172.17.0.3:8888", 8888);

        sync(
            &mut LocalState::default(),
            &exposer,
            Some("10.13.13.22"),
            &Reports::default(),
        );

        assert!(exposer.exposed_ids().is_empty());
    }

    #[test]
    fn sync_without_a_mesh_ip_withdraws_the_address_from_the_hub() {
        let exposer = FakeExposer::default();
        exposer.expose("dep_1", "10.13.13.22", "172.17.0.3:8888", 8888);
        let reports = Reports::default();
        let mut rec = healthy("172.17.0.3", 8888);
        rec.mesh_endpoint = Some("10.13.13.22:8888".to_string());
        rec.mesh_port = Some(8888);
        let mut local = LocalState::default();
        local.deployments.insert("dep_1".into(), rec);

        assert!(sync(&mut local, &exposer, None, &reports));

        assert!(exposer.exposed_ids().is_empty());
        let rec = &local.deployments["dep_1"];
        assert_eq!(rec.mesh_endpoint, None);
        // Kept, so the address comes back on the same port with the tunnel.
        assert_eq!(rec.mesh_port, Some(8888));
        assert_eq!(reports.taken(), vec![report("dep_1", "healthy", None)]);
    }

    #[test]
    fn exposure_is_on_unless_the_operator_turns_it_off() {
        let ip = || Some("10.13.13.22".to_string());
        assert_eq!(exposure_ip(ip(), None), ip());
        assert_eq!(exposure_ip(ip(), Some("1")), ip());
        for off in ["0", "false", "OFF", " no "] {
            assert_eq!(exposure_ip(ip(), Some(off)), None, "{off:?}");
        }
        assert_eq!(exposure_ip(None, None), None);
    }

    // --- owner-only access (zc#212) -----------------------------------------

    fn ip(s: &str) -> IpAddr {
        s.parse().unwrap()
    }

    fn answer(ips: &[&str], complete: bool) -> OwnerMeshIps {
        OwnerMeshIps {
            ips: ips.iter().map(|s| ip(s)).collect(),
            complete,
        }
    }

    const OWN: &str = "10.13.13.22";

    /// Like `roundtrip`, but tolerates the connection being cut: returns
    /// whatever arrived, which is nothing when the forwarder refused it.
    fn reply_or_nothing(endpoint: &str, msg: &[u8]) -> Vec<u8> {
        let Ok(mut conn) = TcpStream::connect(endpoint) else {
            return Vec::new();
        };
        let _ = conn.set_read_timeout(Some(Duration::from_secs(5)));
        let _ = conn.write_all(msg);
        let _ = conn.shutdown(Shutdown::Write);
        let mut reply = Vec::new();
        let _ = conn.read_to_end(&mut reply);
        reply
    }

    #[test]
    fn allowlist_always_lets_this_brokers_own_address_through() {
        assert!(MeshAllowlist::default().allows(ip(OWN), ip(OWN)));
    }

    #[test]
    fn allowlist_fails_closed_until_the_hub_has_answered() {
        assert!(!MeshAllowlist::default().allows(ip("10.13.13.5"), ip(OWN)));
    }

    #[test]
    fn allowlist_lets_exactly_the_owners_addresses_through() {
        let list = MeshAllowlist::default();
        assert!(list.apply(&answer(&["10.13.13.5", "10.13.13.9"], true)));
        assert!(list.allows(ip("10.13.13.5"), ip(OWN)));
        assert!(list.allows(ip("10.13.13.9"), ip(OWN)));
        assert!(!list.allows(ip("10.13.13.7"), ip(OWN)));
        assert!(list.allows(ip(OWN), ip(OWN)));
    }

    #[test]
    fn a_complete_answer_replaces_the_allowlist() {
        // e.g. the owner's other broker was revoked or moved to another account.
        let list = MeshAllowlist::default();
        list.apply(&answer(&["10.13.13.5", "10.13.13.9"], true));
        assert!(list.apply(&answer(&["10.13.13.5"], true)));
        assert!(!list.allows(ip("10.13.13.9"), ip(OWN)));
        assert!(
            !list.apply(&answer(&["10.13.13.5"], true)),
            "the same answer again is no change"
        );
    }

    #[test]
    fn an_incomplete_answer_is_used_only_when_nothing_is_known() {
        let list = MeshAllowlist::default();
        // The hub could not reach the VPN manager, so it lists broker nodes only.
        assert!(list.apply(&answer(&["10.13.13.9"], false)));
        assert!(list.allows(ip("10.13.13.9"), ip(OWN)));
        // A complete answer adds the owner's VPN peer...
        list.apply(&answer(&["10.13.13.9", "10.13.13.5"], true));
        // ...and a later incomplete one must not drop it again.
        assert!(!list.apply(&answer(&["10.13.13.9"], false)));
        assert!(list.allows(ip("10.13.13.5"), ip(OWN)));
    }

    #[test]
    fn allowlist_survives_a_restart_through_its_cache_file() {
        let _home = TempZakuroHome::new();
        let list = MeshAllowlist::default();
        list.apply(&answer(&["10.13.13.5"], true));
        list.persist();
        let reloaded = MeshAllowlist::load();
        assert!(reloaded.allows(ip("10.13.13.5"), ip(OWN)));
        assert!(!reloaded.allows(ip("10.13.13.7"), ip(OWN)));
    }

    #[test]
    fn a_missing_cache_file_loads_fail_closed() {
        let _home = TempZakuroHome::new();
        let list = MeshAllowlist::load();
        assert!(!list.allows(ip("10.13.13.5"), ip(OWN)));
        assert!(list.allows(ip(OWN), ip(OWN)));
    }

    #[test]
    fn forwarder_drops_connections_its_policy_refuses() {
        let fwd = TcpForwarder::with_policy(|_, _| false);
        let endpoint = fwd
            .expose("dep_1", "127.0.0.1", &echo_server("echo:"), free_port())
            .expect("exposed");
        // The kernel still completes the TCP handshake, but the forwarder
        // closes the connection without ever dialing the container.
        assert!(reply_or_nothing(&endpoint, b"hello").is_empty());
        fwd.release("dep_1");
    }
}