dora-cli 1.0.1

use eyre::{Context, bail};
use std::{
    collections::{BTreeMap, HashSet},
    net::{IpAddr, SocketAddr},
    path::Path,
};

use dora_core::topics::{
    DORA_COORDINATOR_PORT_WS_DEFAULT, DORA_ZENOH_LISTEN_PORT_DEFAULT, zenoh_endpoint,
};

#[derive(Debug, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ClusterConfig {
    pub coordinator: CoordinatorConfig,
    /// Shared Zenoh peer endpoint that all daemons use as a rendezvous
    /// for cross-daemon discovery (e.g. `tcp/192.168.1.1:5456`). When
    /// set, `dora cluster up` passes it to every daemon via
    /// `dora daemon --zenoh-peer <ep>`. The first daemon to bind the
    /// endpoint serves as the gossip hub; the rest fall through to
    /// connect-only. Set this when running on networks without working
    /// multicast (dev containers, hardened deployments, many CI
    /// runners) — otherwise daemons can't find each other.
    #[serde(default)]
    pub zenoh_peer: Option<String>,
    pub machines: Vec<MachineConfig>,
}

#[derive(Debug, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CoordinatorConfig {
    pub addr: IpAddr,
    #[serde(default = "default_port")]
    pub port: u16,
}

fn default_port() -> u16 {
    DORA_COORDINATOR_PORT_WS_DEFAULT
}

#[derive(Debug, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MachineConfig {
    pub id: String,
    pub host: String,
    #[serde(default)]
    pub user: Option<String>,
    /// Custom SSH port. Defaults to the SSH client default (22).
    #[serde(default)]
    pub port: Option<u16>,
    /// Custom daemon local-listen port. Defaults to 53291.
    /// Set when running multiple daemons on the same host
    /// (see `examples/multiple-daemons`).
    #[serde(default)]
    pub daemon_port: Option<u16>,
    /// Labels for label-based scheduling (e.g. `gpu: "true"`, `arch: arm64`).
    #[serde(default)]
    pub labels: BTreeMap<String, String>,
    /// Address the *other daemons* use to reach this machine's zenoh listener.
    ///
    /// Defaults to `host` when that is an IP address, which is the common case:
    /// on a mesh VPN the tunnel address is both how you SSH in and how peers
    /// dial. Set it when the two differ — an SSH-only jump address, a hostname
    /// rather than an IP, or a multi-homed machine whose SSH interface is not
    /// the one the other daemons can reach.
    #[serde(default)]
    pub zenoh_addr: Option<String>,
    /// Port this machine's zenoh listener binds. Defaults to 5456.
    ///
    /// Per-machine like `daemon_port`, so two daemons on one host stay
    /// expressible.
    #[serde(default)]
    pub zenoh_port: Option<u16>,
}

impl MachineConfig {
    /// The address peers dial to reach this machine, if one can be determined.
    ///
    /// A hostname in `host` yields `None`: `--zenoh-listen` needs an address to
    /// *bind*, which only the remote machine could resolve, and dora will not
    /// guess on its behalf.
    fn dialable_addr(&self) -> Option<IpAddr> {
        match &self.zenoh_addr {
            Some(addr) => addr.parse().ok(),
            None => self.host.parse().ok(),
        }
    }
}

impl ClusterConfig {
    pub fn load(path: &Path) -> eyre::Result<Self> {
        let raw = std::fs::read_to_string(path)
            .with_context(|| format!("failed to read `{}`", path.display()))?;
        let config: Self = serde_yaml::from_str(&raw)
            .with_context(|| format!("failed to parse `{}`", path.display()))?;
        config.validate()?;
        Ok(config)
    }

    fn validate(&self) -> eyre::Result<()> {
        if self.machines.is_empty() {
            bail!("cluster config must define at least one machine");
        }
        // `zenoh_peer` is interpolated verbatim into the same remote command
        // (`dora daemon --zenoh-peer {ep}`), so it needs the same guard as the
        // id/labels below — but with a charset wide enough for endpoint syntax
        // (`tcp/1.2.3.4:7447`, IPv6 literals), which the id charset would reject.
        if let Some(ep) = &self.zenoh_peer {
            validate_endpoint_shell_safe(&format!("zenoh_peer `{ep}`"), ep)?;
        }
        let mut seen = HashSet::new();
        for m in &self.machines {
            if m.id.is_empty() {
                bail!("machine id must not be empty");
            }
            // `machine.id` and every label key/value are interpolated verbatim
            // into the shell command that `dora cluster up` runs over SSH
            // (`dora daemon --machine-id {id} --labels {k}={v},... >
            // /tmp/dora-daemon-{id}.log`). A value containing whitespace or shell
            // metacharacters would corrupt that command (or, for the id, the log
            // path), so restrict these fields to the same safe identifier charset
            // dora uses for node/data ids (`libraries/message/src/id.rs`).
            // `host`/`user` are deliberately left unrestricted: they are network
            // addresses that legitimately carry characters outside this set
            // (e.g. IPv6 literals), and are folded into the ssh *target* rather
            // than the remote command string. `zenoh_peer` *does* land in the
            // remote command, so it is validated (with a wider charset) above.
            validate_shell_safe(&format!("machine id `{}`", m.id), &m.id, true)?;
            for (k, v) in &m.labels {
                validate_shell_safe(&format!("label key `{k}` on machine `{}`", m.id), k, true)?;
                validate_shell_safe(
                    &format!("label value `{v}` on machine `{}`", m.id),
                    v,
                    false,
                )?;
            }
            if !seen.insert(&m.id) {
                bail!("duplicate machine id: `{}`", m.id);
            }
            if m.host.is_empty() {
                bail!("machine `{}` host must not be empty", m.id);
            }
            // `host`/`user` are folded into the ssh *target* (`{user}@{host}`)
            // and passed to the local `ssh` binary. They are deliberately not
            // run through `validate_shell_safe` (a target legitimately carries
            // `:`/`[`/`]` for IPv6 literals, `@`, etc.), but a value that begins
            // with `-` is parsed by `ssh`'s own argument parser as an *option*
            // rather than a hostname — e.g. `-oProxyCommand=...` executes an
            // arbitrary command on the local machine before connecting (the
            // classic ssh/git argument-injection class, cf. CVE-2017-1000117).
            // Reject a leading dash on both fields; `run_ssh` additionally
            // passes `--` before the target as defense-in-depth.
            validate_no_leading_dash(&format!("machine `{}` host", m.id), &m.host)?;
            if let Some(user) = &m.user {
                validate_no_leading_dash(&format!("machine `{}` user", m.id), user)?;
            }
            // Reject daemon_port = 0: the daemon would bind an ephemeral port
            // but exports DORA_DAEMON_LOCAL_LISTEN_PORT=0 to spawned nodes
            // (binaries/cli/src/command/daemon.rs), so they fail to connect.
            if m.daemon_port == Some(0) {
                bail!("machine `{}` daemon_port must not be 0", m.id);
            }
            // Same reasoning as `zenoh_peer` above: both fields are
            // interpolated verbatim into the remote command
            // (`--zenoh-listen {addr}:{port} --zenoh-connect ...`).
            if let Some(addr) = &m.zenoh_addr {
                validate_endpoint_shell_safe(
                    &format!("machine `{}` zenoh_addr `{addr}`", m.id),
                    addr,
                )?;
                if addr.parse::<IpAddr>().is_err() {
                    bail!(
                        "machine `{}` zenoh_addr `{addr}` is not an IP address. \
                         Peers dial the address the daemon binds, and only an \
                         address can be bound — a hostname would have to be \
                         resolved on the remote machine, which dora cannot do \
                         from here.",
                        m.id
                    );
                }
            }
            // Port 0 would bind an ephemeral port, leaving peers dialing `:0`.
            if m.zenoh_port == Some(0) {
                bail!("machine `{}` zenoh_port must not be 0", m.id);
            }
        }
        // The mesh and the rendezvous are two answers to the same question, and
        // mixing them wires every daemon to dial both. Rather than silently
        // pick one, say which to drop.
        if self.zenoh_peer.is_some()
            && self
                .machines
                .iter()
                .any(|m| m.zenoh_addr.is_some() || m.zenoh_port.is_some())
        {
            bail!(
                "cluster config sets both `zenoh_peer` and per-machine zenoh \
                 addresses. `zenoh_peer` is a single shared rendezvous the \
                 daemons discover each other through; per-machine addresses \
                 wire them into an explicit mesh instead. Keep one: drop \
                 `zenoh_peer` for the mesh (recommended — it needs no hub and \
                 no gossip), or drop the per-machine fields to keep the \
                 rendezvous."
            );
        }
        Ok(())
    }

    /// Per-machine `--zenoh-listen`/`--zenoh-connect` arguments wiring every
    /// daemon to dial every other one.
    ///
    /// Since zenoh 1.9, peers do not relay for each other: two daemons that
    /// never form a direct link exchange nothing, with no fallback to wait
    /// for. Naming every peer establishes that clique by construction, which is
    /// also what makes a cluster work on a network without multicast (a mesh
    /// VPN carries none).
    ///
    /// Nothing is derived unless *every* machine gets an endpoint. A half-mesh
    /// is the worst of both: the unmeshed daemons still need discovery, but the
    /// meshed ones disabled multicast by having explicit connect endpoints.
    pub fn zenoh_mesh_args(&self) -> ZenohMesh<'_> {
        if self.zenoh_peer.is_some() || self.machines.len() < 2 {
            // A rendezvous was chosen instead, or there is nobody to dial.
            return ZenohMesh::NotNeeded;
        }
        let mut endpoints: Vec<(&str, IpAddr, u16)> = Vec::new();
        for machine in &self.machines {
            let Some(addr) = machine.dialable_addr() else {
                return ZenohMesh::Unavailable(format!(
                    "machine `{}` has no dialable zenoh address (`host` is not an \
                     IP address and `zenoh_addr` is unset)",
                    machine.id
                ));
            };
            let port = machine.zenoh_port.unwrap_or(DORA_ZENOH_LISTEN_PORT_DEFAULT);
            // Two daemons on one host is a supported shape (`daemon_port`
            // exists for it), and there they share a default zenoh port too.
            // Both would bind the same endpoint: the second daemon's bind is
            // fatal — it was named explicitly — and it dies in the background
            // where `dora cluster up` cannot see it. Say so instead.
            if let Some((other, _, _)) = endpoints
                .iter()
                .find(|(_, other_addr, other_port)| (*other_addr, *other_port) == (addr, port))
            {
                return ZenohMesh::Unavailable(format!(
                    "machines `{other}` and `{}` would both bind zenoh endpoint {}; \
                     give one of them a distinct `zenoh_port`",
                    machine.id,
                    SocketAddr::new(addr, port)
                ));
            }
            endpoints.push((machine.id.as_str(), addr, port));
        }

        ZenohMesh::Derived(
            endpoints
                .iter()
                .map(|(id, addr, port)| {
                    let peers: Vec<String> = endpoints
                        .iter()
                        .filter(|(peer_id, _, _)| peer_id != id)
                        .map(|(_, peer_addr, peer_port)| zenoh_endpoint(*peer_addr, *peer_port))
                        .collect();
                    // A dial is one-directional but the transport it opens is
                    // not, so the full mesh is |machines| * (|machines| - 1)
                    // dials for |machines| * (|machines| - 1) / 2 links. The
                    // duplication is deliberate: whichever daemon starts first
                    // reaches the other as soon as it comes up, in either
                    // order, and zenoh drops the redundant attempt.
                    (
                        *id,
                        format!(
                            " --zenoh-listen {} --zenoh-connect {}",
                            SocketAddr::new(*addr, *port),
                            peers.join(",")
                        ),
                    )
                })
                .collect(),
        )
    }
}

/// Whether `dora cluster up` can wire the daemons into an explicit zenoh mesh.
#[derive(Debug)]
pub enum ZenohMesh<'a> {
    /// Per-machine daemon arguments: each listens on its own endpoint and dials
    /// every other machine.
    Derived(BTreeMap<&'a str, String>),
    /// Deliberately not derived — a `zenoh_peer` rendezvous is configured, or
    /// there is only one machine.
    NotNeeded,
    /// Could not be derived. The message says why and which field fixes it;
    /// the cluster falls back to its previous discovery behaviour.
    Unavailable(String),
}

/// Reject a value that will be interpolated into the remote SSH command unless
/// it consists only of the safe identifier charset `[a-zA-Z0-9_.-]`. `what`
/// names the field for the error message.
///
/// When `at_token_start` is set, a leading `-` is also rejected. A value that
/// *begins* a shell token on the remote command line — the id in
/// `--machine-id {id}`, or the first `{k}=` of `--labels {k}=v,...` — would
/// otherwise be parsed by clap as an option flag rather than the argument to
/// the preceding option, so the daemon never starts and the failure surfaces
/// only later as a confusing "did not register" timeout. A label *value* sits
/// after `{k}=` inside the token, so a leading `-` there (e.g. `priority: "-1"`)
/// is harmless and must stay allowed. (Sibling of dora-rs/dora#3135, which
/// guards the `host`/`user` fields on the ssh-target side.)
fn validate_shell_safe(what: &str, value: &str, at_token_start: bool) -> eyre::Result<()> {
    if let Some(ch) = value
        .chars()
        .find(|c| !c.is_ascii_alphanumeric() && *c != '_' && *c != '-' && *c != '.')
    {
        bail!("{what} contains invalid character `{ch}` -- only [a-zA-Z0-9_.-] are allowed");
    }
    if at_token_start && value.starts_with('-') {
        bail!("{what} must not start with `-`");
    }
    Ok(())
}

/// Reject a value that begins with `-`, which the local `ssh` binary would
/// otherwise parse as an option rather than a hostname/user (argument
/// injection). Unlike [`validate_shell_safe`] this places no charset
/// restriction, so IPv6 literals and normal `user@host` values stay valid.
fn validate_no_leading_dash(what: &str, value: &str) -> eyre::Result<()> {
    if value.starts_with('-') {
        bail!("{what} must not start with `-` (would be parsed as an ssh option)");
    }
    Ok(())
}

/// Like [`validate_shell_safe`], but also permits the extra characters a Zenoh
/// endpoint string carries — `/` and `:` in `tcp/1.2.3.4:7447`, and the
/// `[`/`]` around an IPv6 literal like `tcp/[::1]:7447`. It still rejects
/// whitespace and shell metacharacters (`;`, `|`, `$`, backtick, ...), so the
/// value cannot break out of the remote command.
///
/// It also rejects a leading `-`, the same guard `validate_shell_safe` applies
/// at a token start: `zenoh_peer` is interpolated as the whole argument of
/// `dora daemon --zenoh-peer {ep}`, so a value like `-tcp/1.2.3.4:5456` is
/// parsed by the remote daemon's clap as an option flag instead of the
/// endpoint, and the daemon never starts — surfacing only later as a confusing
/// "daemon(s) did not register" timeout. No legitimate Zenoh endpoint begins
/// with `-`.
fn validate_endpoint_shell_safe(what: &str, value: &str) -> eyre::Result<()> {
    if let Some(ch) = value.chars().find(|c| {
        !c.is_ascii_alphanumeric() && !matches!(c, '_' | '-' | '.' | '/' | ':' | '[' | ']')
    }) {
        bail!(
            "{what} contains invalid character `{ch}` -- only [a-zA-Z0-9_.:/] and `[`/`]` are allowed"
        );
    }
    if value.starts_with('-') {
        bail!("{what} must not start with `-`");
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Write;
    use tempfile::NamedTempFile;

    fn write_yaml(content: &str) -> NamedTempFile {
        let mut f = NamedTempFile::new().unwrap();
        f.write_all(content.as_bytes()).unwrap();
        f
    }

    #[test]
    fn parse_minimal() {
        let f = write_yaml(
            "coordinator:\n  addr: 192.168.1.100\nmachines:\n  - id: arm\n    host: 192.168.1.101\n",
        );
        let cfg = ClusterConfig::load(f.path()).unwrap();
        assert_eq!(cfg.coordinator.port, DORA_COORDINATOR_PORT_WS_DEFAULT);
        assert_eq!(cfg.machines.len(), 1);
        assert_eq!(cfg.machines[0].id, "arm");
        assert!(cfg.machines[0].user.is_none());
        assert!(cfg.machines[0].port.is_none());
        assert!(cfg.machines[0].daemon_port.is_none());
        assert!(cfg.machines[0].labels.is_empty());
    }

    #[test]
    fn parse_with_daemon_port() {
        let f = write_yaml(
            "coordinator:\n  addr: 10.0.0.1\nmachines:\n  - id: a\n    host: 10.0.0.2\n    daemon_port: 53292\n",
        );
        let cfg = ClusterConfig::load(f.path()).unwrap();
        assert_eq!(cfg.machines[0].daemon_port, Some(53292));
    }

    #[test]
    fn reject_invalid_daemon_port() {
        let f = write_yaml(
            "coordinator:\n  addr: 10.0.0.1\nmachines:\n  - id: a\n    host: 10.0.0.2\n    daemon_port: 99999\n",
        );
        assert!(ClusterConfig::load(f.path()).is_err());
    }

    #[test]
    fn reject_zero_daemon_port() {
        let f = write_yaml(
            "coordinator:\n  addr: 10.0.0.1\nmachines:\n  - id: a\n    host: 10.0.0.2\n    daemon_port: 0\n",
        );
        let err = ClusterConfig::load(f.path()).unwrap_err();
        assert!(err.to_string().contains("daemon_port must not be 0"));
    }

    #[test]
    fn parse_with_zenoh_peer() {
        let f = write_yaml(
            "coordinator:\n  addr: 10.0.0.1\nzenoh_peer: tcp/10.0.0.1:5456\nmachines:\n  - id: a\n    host: 10.0.0.2\n",
        );
        let cfg = ClusterConfig::load(f.path()).unwrap();
        assert_eq!(cfg.zenoh_peer.as_deref(), Some("tcp/10.0.0.1:5456"));
    }

    #[test]
    fn parse_without_zenoh_peer_defaults_to_none() {
        let f = write_yaml(
            "coordinator:\n  addr: 10.0.0.1\nmachines:\n  - id: a\n    host: 10.0.0.2\n",
        );
        let cfg = ClusterConfig::load(f.path()).unwrap();
        assert!(cfg.zenoh_peer.is_none());
    }

    #[test]
    fn parse_with_port() {
        let f = write_yaml(
            "coordinator:\n  addr: 10.0.0.1\nmachines:\n  - id: a\n    host: 10.0.0.2\n    port: 2222\n",
        );
        let cfg = ClusterConfig::load(f.path()).unwrap();
        assert_eq!(cfg.machines[0].port, Some(2222));
    }

    #[test]
    fn reject_invalid_port() {
        let f = write_yaml(
            "coordinator:\n  addr: 10.0.0.1\nmachines:\n  - id: a\n    host: 10.0.0.2\n    port: 99999\n",
        );
        assert!(ClusterConfig::load(f.path()).is_err());
    }

    #[test]
    fn parse_full() {
        let f = write_yaml(
            "coordinator:\n  addr: 10.0.0.1\n  port: 7000\nmachines:\n  - id: a\n    host: 10.0.0.2\n    user: bob\n    labels:\n      gpu: \"true\"\n      arch: arm64\n  - id: b\n    host: 10.0.0.3\n",
        );
        let cfg = ClusterConfig::load(f.path()).unwrap();
        assert_eq!(cfg.coordinator.port, 7000);
        assert_eq!(cfg.machines.len(), 2);
        assert_eq!(cfg.machines[0].user.as_deref(), Some("bob"));
        assert_eq!(
            cfg.machines[0].labels.get("gpu").map(|s| s.as_str()),
            Some("true")
        );
        assert_eq!(
            cfg.machines[0].labels.get("arch").map(|s| s.as_str()),
            Some("arm64")
        );
    }

    #[test]
    fn reject_duplicate_ids() {
        let f = write_yaml(
            "coordinator:\n  addr: 10.0.0.1\nmachines:\n  - id: dup\n    host: a\n  - id: dup\n    host: b\n",
        );
        let err = ClusterConfig::load(f.path()).unwrap_err();
        assert!(err.to_string().contains("duplicate machine id"));
    }

    #[test]
    fn reject_empty_machines() {
        let f = write_yaml("coordinator:\n  addr: 10.0.0.1\nmachines: []\n");
        let err = ClusterConfig::load(f.path()).unwrap_err();
        assert!(err.to_string().contains("at least one machine"));
    }

    #[test]
    fn reject_empty_id() {
        let f =
            write_yaml("coordinator:\n  addr: 10.0.0.1\nmachines:\n  - id: \"\"\n    host: a\n");
        let err = ClusterConfig::load(f.path()).unwrap_err();
        assert!(err.to_string().contains("must not be empty"));
    }

    #[test]
    fn reject_id_with_shell_metacharacters() {
        // An id with a space or shell metacharacter would corrupt the remote
        // `dora daemon --machine-id {id} ...` command run over SSH.
        for bad in ["a b", "a;rm -rf /", "a$(whoami)", "a/b", "a`id`"] {
            let f = write_yaml(&format!(
                "coordinator:\n  addr: 10.0.0.1\nmachines:\n  - id: \"{bad}\"\n    host: h\n"
            ));
            let err = ClusterConfig::load(f.path())
                .unwrap_err()
                .to_string()
                .to_lowercase();
            assert!(
                err.contains("invalid character"),
                "id `{bad}` should be rejected, got: {err}"
            );
        }
    }

    #[test]
    fn reject_id_with_leading_dash() {
        // A leading `-` passes the charset check but makes the remote
        // `dora daemon --machine-id {id}` parse the id as an option flag, so the
        // daemon silently fails to start. Reject it up front.
        for bad in ["-x", "--help", "-"] {
            let f = write_yaml(&format!(
                "coordinator:\n  addr: 10.0.0.1\nmachines:\n  - id: \"{bad}\"\n    host: h\n"
            ));
            let err = ClusterConfig::load(f.path()).unwrap_err().to_string();
            assert!(
                err.contains("must not start with `-`"),
                "id `{bad}` should be rejected, got: {err}"
            );
        }
    }

    #[test]
    fn reject_label_key_with_leading_dash() {
        // A label *key* can begin the `--labels {k}=...` token, so a leading `-`
        // there is the same option-flag hazard as the machine id.
        let f = write_yaml(
            "coordinator:\n  addr: 10.0.0.1\nmachines:\n  - id: a\n    host: h\n    labels:\n      \"-gpu\": \"v\"\n",
        );
        let err = ClusterConfig::load(f.path()).unwrap_err().to_string();
        assert!(
            err.contains("must not start with `-`") && err.contains("label key"),
            "label key with leading dash should be rejected, got: {err}"
        );
    }

    #[test]
    fn accept_label_value_with_leading_dash() {
        // A label *value* sits after `{k}=` in the `--labels {k}={v}` token, so a
        // leading `-` there is never parsed as an option flag and must stay
        // valid (e.g. a numeric `-1` priority or a `-rc1` version tag).
        for v in ["-1", "-rc1"] {
            let f = write_yaml(&format!(
                "coordinator:\n  addr: 10.0.0.1\nmachines:\n  - id: a\n    host: h\n    labels:\n      priority: \"{v}\"\n"
            ));
            assert!(
                ClusterConfig::load(f.path()).is_ok(),
                "label value `{v}` should be accepted"
            );
        }
    }

    #[test]
    fn reject_label_with_shell_metacharacters() {
        // Label keys/values are interpolated into `--labels k=v` on the remote
        // shell, so a metacharacter there is the same injection vector as the id.
        let f = write_yaml(
            "coordinator:\n  addr: 10.0.0.1\nmachines:\n  - id: a\n    host: h\n    labels:\n      gpu: \"x;rm -rf /\"\n",
        );
        let err = ClusterConfig::load(f.path())
            .unwrap_err()
            .to_string()
            .to_lowercase();
        assert!(
            err.contains("invalid character") && err.contains("label value"),
            "malicious label value should be rejected, got: {err}"
        );
    }

    #[test]
    fn reject_zenoh_peer_with_shell_metacharacters() {
        // `zenoh_peer` is interpolated into `dora daemon --zenoh-peer {ep}` on
        // the remote shell, so a metacharacter there is an injection vector too.
        let f = write_yaml(
            "coordinator:\n  addr: 10.0.0.1\nzenoh_peer: \"tcp/1.2.3.4:7447;rm -rf /\"\nmachines:\n  - id: a\n    host: h\n",
        );
        let err = ClusterConfig::load(f.path())
            .unwrap_err()
            .to_string()
            .to_lowercase();
        assert!(
            err.contains("invalid character") && err.contains("zenoh_peer"),
            "malicious zenoh_peer should be rejected, got: {err}"
        );
    }

    #[test]
    fn reject_zenoh_peer_with_leading_dash() {
        // A leading `-` passes the endpoint charset check but makes the remote
        // `dora daemon --zenoh-peer {ep}` parse the endpoint as an option flag,
        // so the daemon silently fails to start. Reject it up front, matching
        // the guard on the machine id / label key.
        for bad in ["-tcp/1.2.3.4:5456", "--zenoh-peer", "-"] {
            let f = write_yaml(&format!(
                "coordinator:\n  addr: 10.0.0.1\nzenoh_peer: \"{bad}\"\nmachines:\n  - id: a\n    host: h\n"
            ));
            let err = ClusterConfig::load(f.path()).unwrap_err().to_string();
            assert!(
                err.contains("must not start with `-`") && err.contains("zenoh_peer"),
                "zenoh_peer `{bad}` should be rejected, got: {err}"
            );
        }
    }

    #[test]
    fn accept_ipv6_zenoh_peer() {
        // The endpoint charset must still permit a legitimate IPv6 literal
        // endpoint, whose `[`/`]`/`:` the id charset would reject.
        let f = write_yaml(
            "coordinator:\n  addr: 10.0.0.1\nzenoh_peer: \"tcp/[::1]:7447\"\nmachines:\n  - id: a\n    host: h\n",
        );
        let cfg = ClusterConfig::load(f.path()).unwrap();
        assert_eq!(cfg.zenoh_peer.as_deref(), Some("tcp/[::1]:7447"));
    }

    #[test]
    fn reject_host_with_leading_dash() {
        // A `host` beginning with `-` is parsed by the local `ssh` binary as an
        // option (e.g. `-oProxyCommand=touch /tmp/pwned;false`), giving arbitrary
        // local command execution before any network connection is made.
        let f = write_yaml(
            "coordinator:\n  addr: 10.0.0.1\nmachines:\n  - id: a\n    host: \"-oProxyCommand=touch /tmp/pwned;false\"\n",
        );
        let err = ClusterConfig::load(f.path()).unwrap_err().to_string();
        assert!(
            err.contains("must not start with `-`") && err.contains("host"),
            "host with leading dash should be rejected, got: {err}"
        );
    }

    #[test]
    fn reject_user_with_leading_dash() {
        // `ssh_target` yields `{user}@{host}`, so a `user` beginning with `-`
        // produces a target string that also starts with `-` — same vector.
        let f = write_yaml(
            "coordinator:\n  addr: 10.0.0.1\nmachines:\n  - id: a\n    host: h\n    user: \"-oProxyCommand=x\"\n",
        );
        let err = ClusterConfig::load(f.path()).unwrap_err().to_string();
        assert!(
            err.contains("must not start with `-`") && err.contains("user"),
            "user with leading dash should be rejected, got: {err}"
        );
    }

    #[test]
    fn accept_ipv6_host() {
        // An IPv6 literal host carries `:` (and may be bracketed) — the id
        // charset would reject it, but host/user are only checked for a leading
        // dash, so it must stay valid.
        for host in ["::1", "2001:db8::1", "192.168.1.10"] {
            let f = write_yaml(&format!(
                "coordinator:\n  addr: 10.0.0.1\nmachines:\n  - id: a\n    host: \"{host}\"\n"
            ));
            assert!(
                ClusterConfig::load(f.path()).is_ok(),
                "host `{host}` should be accepted"
            );
        }
    }

    #[test]
    fn accept_conventional_ids() {
        // Hostname-style ids (alphanumerics plus `.`, `_`, `-`) stay valid.
        for good in ["arm", "jetson-01", "node_2", "gpu.host-1"] {
            let f = write_yaml(&format!(
                "coordinator:\n  addr: 10.0.0.1\nmachines:\n  - id: {good}\n    host: h\n"
            ));
            assert!(
                ClusterConfig::load(f.path()).is_ok(),
                "id `{good}` should be accepted"
            );
        }
    }

    #[test]
    fn reject_unknown_field() {
        let f = write_yaml(
            "coordinator:\n  addr: 10.0.0.1\n  bogus: true\nmachines:\n  - id: a\n    host: b\n",
        );
        assert!(ClusterConfig::load(f.path()).is_err());
    }

    const TWO_IP_MACHINES: &str = "coordinator:\n  addr: 100.64.0.1\nmachines:\n  \
                                   - id: a\n    host: 100.64.0.2\n  \
                                   - id: b\n    host: 100.64.0.3\n";

    // Every daemon listens on its own endpoint and dials every other one: the
    // clique zenoh 1.9 requires, established without multicast or gossip.
    #[test]
    fn mesh_wires_every_machine_to_every_other() {
        let f = write_yaml(TWO_IP_MACHINES);
        let config = ClusterConfig::load(f.path()).unwrap();
        let ZenohMesh::Derived(args) = config.zenoh_mesh_args() else {
            panic!("both hosts are IPs, so the mesh must derive");
        };
        assert_eq!(
            args.get("a").map(String::as_str),
            Some(" --zenoh-listen 100.64.0.2:5456 --zenoh-connect tcp/100.64.0.3:5456")
        );
        assert_eq!(
            args.get("b").map(String::as_str),
            Some(" --zenoh-listen 100.64.0.3:5456 --zenoh-connect tcp/100.64.0.2:5456")
        );
    }

    // `zenoh_addr` overrides `host` — the SSH address and the address peers
    // dial are not always the same interface — and `zenoh_port` keeps two
    // daemons on one host expressible.
    #[test]
    fn mesh_honors_explicit_address_and_port() {
        let f = write_yaml(
            "coordinator:\n  addr: 100.64.0.1\nmachines:\n  \
             - id: a\n    host: jump.example.com\n    zenoh_addr: 100.64.0.2\n  \
             - id: b\n    host: 100.64.0.2\n    zenoh_port: 5457\n",
        );
        let config = ClusterConfig::load(f.path()).unwrap();
        let ZenohMesh::Derived(args) = config.zenoh_mesh_args() else {
            panic!("both machines have addresses, so the mesh must derive");
        };
        assert!(args["a"].contains("--zenoh-listen 100.64.0.2:5456"));
        assert!(args["a"].contains("--zenoh-connect tcp/100.64.0.2:5457"));
        assert!(args["b"].contains("--zenoh-listen 100.64.0.2:5457"));
    }

    // A machine dora cannot address leaves the cluster unmeshed rather than
    // half-meshed: explicit connect endpoints disable multicast scouting for
    // the daemons that have them, so a partial mesh partitions the rest.
    #[test]
    fn mesh_is_all_or_nothing() {
        let f = write_yaml(
            "coordinator:\n  addr: 100.64.0.1\nmachines:\n  \
             - id: a\n    host: 100.64.0.2\n  \
             - id: b\n    host: jetson.local\n",
        );
        let config = ClusterConfig::load(f.path()).unwrap();
        let ZenohMesh::Unavailable(reason) = config.zenoh_mesh_args() else {
            panic!("a machine with no dialable address must block the mesh");
        };
        assert!(
            reason.contains("`b`"),
            "reason must name the machine: {reason}"
        );
        assert!(
            reason.contains("zenoh_addr"),
            "reason must name the fix: {reason}"
        );

        // A single-machine cluster has nobody to dial.
        let f = write_yaml(
            "coordinator:\n  addr: 100.64.0.1\nmachines:\n  - id: a\n    host: 100.64.0.2\n",
        );
        let config = ClusterConfig::load(f.path()).unwrap();
        assert!(matches!(config.zenoh_mesh_args(), ZenohMesh::NotNeeded));
    }

    /// Two daemons on one host is a supported shape (`daemon_port` exists for
    /// it), and there both would default to the same zenoh port. The second
    /// daemon's bind is fatal — it was named explicitly — and it dies in the
    /// background where `dora cluster up` never sees it, so the collision has
    /// to be caught here.
    #[test]
    fn colliding_endpoints_block_the_mesh() {
        let f = write_yaml(
            "coordinator:\n  addr: 100.64.0.1\nmachines:\n  \
             - id: a\n    host: 100.64.0.2\n    daemon_port: 53291\n  \
             - id: b\n    host: 100.64.0.2\n    daemon_port: 53292\n",
        );
        let config = ClusterConfig::load(f.path()).unwrap();
        let ZenohMesh::Unavailable(reason) = config.zenoh_mesh_args() else {
            panic!("two daemons sharing an endpoint must block the mesh");
        };
        assert!(
            reason.contains("100.64.0.2:5456"),
            "unexpected reason: {reason}"
        );
        assert!(
            reason.contains("zenoh_port"),
            "reason must name the fix: {reason}"
        );

        // Distinct ports make the same pair meshable.
        let f = write_yaml(
            "coordinator:\n  addr: 100.64.0.1\nmachines:\n  \
             - id: a\n    host: 100.64.0.2\n    daemon_port: 53291\n  \
             - id: b\n    host: 100.64.0.2\n    daemon_port: 53292\n    zenoh_port: 5457\n",
        );
        let config = ClusterConfig::load(f.path()).unwrap();
        let ZenohMesh::Derived(args) = config.zenoh_mesh_args() else {
            panic!("distinct ports must derive");
        };
        assert!(args["a"].contains("--zenoh-listen 100.64.0.2:5456"));
        assert!(args["a"].contains("--zenoh-connect tcp/100.64.0.2:5457"));
        assert!(args["b"].contains("--zenoh-listen 100.64.0.2:5457"));
        assert!(args["b"].contains("--zenoh-connect tcp/100.64.0.2:5456"));
    }

    // The rendezvous and the mesh answer the same question two ways; setting
    // both would have every daemon dial a hub *and* its peers.
    #[test]
    fn rendezvous_and_mesh_are_mutually_exclusive() {
        let f = write_yaml(
            "coordinator:\n  addr: 100.64.0.1\nzenoh_peer: tcp/100.64.0.1:5456\nmachines:\n  \
             - id: a\n    host: 100.64.0.2\n    zenoh_addr: 100.64.0.2\n  \
             - id: b\n    host: 100.64.0.3\n",
        );
        let err = ClusterConfig::load(f.path()).unwrap_err().to_string();
        assert!(err.contains("zenoh_peer"), "unexpected error: {err}");

        // Without per-machine fields, `zenoh_peer` still means "rendezvous, not
        // mesh" — the daemons keep discovering each other through the hub.
        let f = write_yaml(
            "coordinator:\n  addr: 100.64.0.1\nzenoh_peer: tcp/100.64.0.1:5456\nmachines:\n  \
             - id: a\n    host: 100.64.0.2\n  - id: b\n    host: 100.64.0.3\n",
        );
        let config = ClusterConfig::load(f.path()).unwrap();
        assert!(matches!(config.zenoh_mesh_args(), ZenohMesh::NotNeeded));
    }

    #[test]
    fn reject_unusable_zenoh_fields() {
        for (yaml, expected) in [
            ("    zenoh_addr: jetson.local\n", "not an IP address"),
            ("    zenoh_port: 0\n", "must not be 0"),
            (
                "    zenoh_addr: \"1.2.3.4; rm -rf /\"\n",
                "invalid character",
            ),
        ] {
            let f = write_yaml(&format!(
                "coordinator:\n  addr: 100.64.0.1\nmachines:\n  - id: a\n    host: 100.64.0.2\n{yaml}"
            ));
            let err = ClusterConfig::load(f.path())
                .expect_err("must be rejected")
                .to_string();
            assert!(err.contains(expected), "unexpected error: {err}");
        }
    }
}