hive-priv: egress-pin and egress-lease sub-commands that bind bridge ports to MACs and load leases

hive-priv/README.md

@@ -37,7 +37,10 @@ not the whole daemon-shared crate.
 `container` (lifecycle), `container_config` (nspawn flags, systemd drop-ins),
 `storage` (btrfs subvolumes, quotas, snapshots, disk measurement), `files`
 (pause marker, socket dir, legacy tmpfiles) and `infra` (gateway nginx, forge
-admin, infra containers). Helpers more than one family calls live in `shared`.
+admin, infra containers). `egress` holds two sub-commands that never
+touch the socket, `egress-pin` and `egress-lease`; the `hive-egress` unit, udev
+and dnsmasq run them to fill the sets of the anti-spoofing table (see
+`docs/4-implementation/networking/network.md`). Helpers more than one family calls live in `shared`.
 Each module's `//!` header lists its ops.
 
 ## Implementation notes

hive-priv/src/egress.rs

@@ -0,0 +1,812 @@
+//! The per-port bindings behind the `bridge hive_egress` anti-spoofing table
+//! (`nix/host-modules/hive-egress.nix`), as two sub-commands run by systemd
+//! units and by dnsmasq, never through the socket.
+//!
+//! - `egress-pin [--seed] [<port>]` binds each port of the agent bridge to the
+//!   MAC its container uses. A port is bound only when a registered container
+//!   owns the other end of its veth pair; a port nobody owns stays unbound, and
+//!   the table drops everything it sends. With `<port>` the command first waits
+//!   for that port to be bound or to vanish. `--seed` also loads the leases in
+//!   dnsmasq's lease file.
+//! - `egress-lease <action> [<mac> <ip> [<host>]]` is dnsmasq's `dhcp-script`:
+//!   it keeps the table's set of leased MAC and IPv4 pairs in step with the
+//!   leases.
+//!
+//! The owner of a port is found from the host side: the port's peer index is
+//! the index of one link in the container's network namespace, and that link's
+//! peer index is the port's own index. Both halves have to hold, so a veth
+//! pair the container makes inside its namespace cannot claim another port.
+//! The container's leader process comes from the machine state systemd-machined
+//! keeps under `/run/systemd/machines`.
+//!
+//! The bound MAC is the one the link has when the port is first looked at, and
+//! the binding then stays for as long as the port exists: later passes keep it
+//! in `/run/hive-egress-pins` and do not read the link again, so root in the
+//! container cannot move its port to another MAC afterwards. The unit starts
+//! the command when the veth appears, while only the container's nix-built
+//! early init runs. A MAC that two ports claim binds neither.
+//!
+//! The entry points are [`run`] (the sub-commands), [`resolve`] (which port has
+//! which MAC) and [`pin_script`] / [`lease_script`] (the `nft` input).
+
+use std::fmt::Write as _;
+use std::fs::{File, OpenOptions};
+use std::io::Write as _;
+use std::path::Path;
+use std::process::{Command, Stdio};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use anyhow::{Context as _, Result, bail};
+
+/// The sub-command that binds ports.
+pub(crate) const PIN: &str = "egress-pin";
+
+/// The sub-command dnsmasq's `dhcp-script` runs.
+pub(crate) const LEASE: &str = "egress-lease";
+
+/// The bridge the table guards, set by the unit that runs the command.
+const BRIDGE_ENV: &str = "HIVE_EGRESS_BRIDGE";
+
+/// Where systemd-machined keeps one file per registered machine.
+const MACHINES_DIR: &str = "/run/systemd/machines";
+
+/// dnsmasq's lease file, when `HIVE_EGRESS_LEASEFILE` does not name another.
+const LEASEFILE: &str = "/var/lib/dnsmasq/dnsmasq.leases";
+
+/// How long `egress-pin <port>` waits for a registered container to own the
+/// port: machined learns of the machine after the veth pair exists.
+const WAIT: Duration = Duration::from_secs(30);
+
+/// The pause between two looks while waiting.
+const POLL: Duration = Duration::from_millis(250);
+
+/// A lease timeout when dnsmasq names no remaining time.
+const DEFAULT_LEASE_SECS: u64 = 3600;
+
+/// Where the bindings are kept between passes, when `HIVE_EGRESS_STATE` names
+/// no other file. It is on tmpfs: a reboot starts every port afresh.
+const STATE: &str = "/run/hive-egress-pins";
+
+/// The lock that makes one pass at a time read and replace the bindings, when
+/// `HIVE_EGRESS_LOCK` names no other file.
+const LOCK: &str = "/run/lock/hive-egress-pin.lock";
+
+/// The nft table the commands write into.
+const TABLE: &str = "bridge hive_egress";
+
+/// An Ethernet address.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) struct Mac([u8; 6]);
+
+impl Mac {
+    /// Parse `aa:bb:cc:dd:ee:ff`, either case.
+    pub(crate) fn parse(text: &str) -> Option<Self> {
+        let mut bytes = [0_u8; 6];
+        let mut parts = text.split(':');
+        for byte in &mut bytes {
+            let part = parts.next()?;
+            if part.len() != 2 {
+                return None;
+            }
+            *byte = u8::from_str_radix(part, 16).ok()?;
+        }
+        parts.next().is_none().then_some(Self(bytes))
+    }
+
+    /// A unicast address that is not all zero: the only kind a port may use.
+    fn is_usable(self) -> bool {
+        self.0[0] & 1 == 0 && self.0 != [0; 6]
+    }
+
+    /// `0x…` with twelve digits: how the table's `chaddr` set spells the
+    /// address, as the 48 bits it compares in the DHCP payload.
+    fn as_integer(self) -> String {
+        let mut text = String::from("0x");
+        for byte in self.0 {
+            let _ = write!(text, "{byte:02x}");
+        }
+        text
+    }
+}
+
+impl std::fmt::Display for Mac {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let b = self.0;
+        write!(
+            f,
+            "{:02x}:{:02x}:{:02x}:{:02x}:{:02x}:{:02x}",
+            b[0], b[1], b[2], b[3], b[4], b[5]
+        )
+    }
+}
+
+/// A port of the agent bridge, as the host sees it.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(crate) struct HostPort {
+    pub(crate) name: String,
+    pub(crate) ifindex: u64,
+    /// The index of the other end of the veth pair, in its own namespace.
+    pub(crate) peer_index: u64,
+}
+
+/// A link in a container's network namespace.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(crate) struct NsLink {
+    pub(crate) ifindex: u64,
+    pub(crate) peer_index: Option<u64>,
+    pub(crate) mac: Mac,
+}
+
+/// A port bound to a MAC.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(crate) struct Pin {
+    pub(crate) port: String,
+    /// The port's index when it was bound: a port made again under the same
+    /// name has another.
+    pub(crate) ifindex: u64,
+    pub(crate) mac: Mac,
+    /// The machine that owned the port when it was bound.
+    pub(crate) machine: String,
+}
+
+/// What [`resolve`] decided about the ports.
+#[derive(Debug, Default, PartialEq, Eq)]
+pub(crate) struct Resolution {
+    pub(crate) pins: Vec<Pin>,
+    /// Port name and why it stays unbound.
+    pub(crate) unbound: Vec<(String, String)>,
+}
+
+/// Bind each port to the MAC of the link at the other end of its veth pair.
+///
+/// `machines` holds each registered container's links. `earlier` holds the
+/// bindings of a previous pass: a port that has the same name and index keeps
+/// its binding whatever its link says now, so a MAC changed after the port was
+/// bound is never bound in its place. A port no machine owns, an unusable MAC
+/// and a MAC two ports claim all leave the port unbound.
+pub(crate) fn resolve(
+    ports: &[HostPort],
+    machines: &[(String, Vec<NsLink>)],
+    earlier: &[Pin],
+) -> Resolution {
+    let mut found: Vec<Pin> = Vec::new();
+    let mut out = Resolution::default();
+    for port in ports {
+        if let Some(kept) = earlier
+            .iter()
+            .find(|p| p.port == port.name && p.ifindex == port.ifindex)
+        {
+            found.push(kept.clone());
+            continue;
+        }
+        let owner = machines.iter().find_map(|(machine, links)| {
+            links
+                .iter()
+                .find(|l| l.ifindex == port.peer_index && l.peer_index == Some(port.ifindex))
+                .map(|l| (machine.as_str(), l.mac))
+        });
+        match owner {
+            None => out
+                .unbound
+                .push((port.name.clone(), "no registered container owns it".into())),
+            Some((machine, mac)) if !mac.is_usable() => out.unbound.push((
+                port.name.clone(),
+                format!("{machine} has the unusable address {mac}"),
+            )),
+            Some((machine, mac)) => found.push(Pin {
+                port: port.name.clone(),
+                ifindex: port.ifindex,
+                mac,
+                machine: machine.to_string(),
+            }),
+        }
+    }
+    for pin in &found {
+        let rivals: Vec<&str> = found
+            .iter()
+            .filter(|p| p.mac == pin.mac && p.port != pin.port)
+            .map(|p| p.port.as_str())
+            .collect();
+        if rivals.is_empty() {
+            out.pins.push(pin.clone());
+        } else {
+            out.unbound.push((
+                pin.port.clone(),
+                format!(
+                    "{} has {}, which {} also has",
+                    pin.machine,
+                    pin.mac,
+                    rivals.join(", ")
+                ),
+            ));
+        }
+    }
+    out
+}
+
+/// The bindings as the state file spells them: `<port> <ifindex> <mac>
+/// <machine>` per line.
+pub(crate) fn state_text(pins: &[Pin]) -> String {
+    let mut text = String::new();
+    for p in pins {
+        let _ = writeln!(text, "{} {} {} {}", p.port, p.ifindex, p.mac, p.machine);
+    }
+    text
+}
+
+/// The bindings in a state file. A line that does not parse is left out.
+pub(crate) fn parse_state(text: &str) -> Vec<Pin> {
+    text.lines()
+        .filter_map(|line| {
+            let mut cols = line.split_whitespace();
+            Some(Pin {
+                port: cols.next()?.to_string(),
+                ifindex: cols.next()?.parse().ok()?,
+                mac: Mac::parse(cols.next()?)?,
+                machine: cols.next()?.to_string(),
+            })
+        })
+        .collect()
+}
+
+/// Whether a name is safe to put into an nft script as a quoted string.
+fn is_plain_name(name: &str) -> bool {
+    !name.is_empty()
+        && name.len() < 16
+        && name
+            .bytes()
+            .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b'-'))
+}
+
+/// The nft input that makes the table's bindings exactly `pins`, in one
+/// transaction: a reader never sees the set empty in between.
+pub(crate) fn pin_script(pins: &[Pin]) -> String {
+    let mut script = format!("flush set {TABLE} port_mac\nflush set {TABLE} port_chaddr\n");
+    for Pin { port, mac, .. } in pins {
+        let _ = writeln!(
+            script,
+            "add element {TABLE} port_mac {{ \"{port}\" . {mac} }}\n\
+             add element {TABLE} port_chaddr {{ \"{port}\" . {} }}",
+            mac.as_integer()
+        );
+    }
+    script
+}
+
+/// The nft input for one lease event, or `None` for an event that changes
+/// nothing: `init`, an IPv6 address, an address or MAC that does not parse.
+///
+/// `add` and `old` replace the pair with one that times out after
+/// `remaining_secs`; `del` removes it. The first `add element` makes the
+/// `delete element` that follows succeed whether or not the pair was there.
+pub(crate) fn lease_script(
+    action: &str,
+    mac: &str,
+    ip: &str,
+    remaining_secs: u64,
+) -> Option<String> {
+    let mac = Mac::parse(mac)?;
+    let ip: std::net::Ipv4Addr = ip.parse().ok()?;
+    let set = format!("{TABLE} lease_v4");
+    let pair = format!("{mac} . {ip}");
+    match action {
+        "add" | "old" => Some(format!(
+            "add element {set} {{ {pair} }}\n\
+             delete element {set} {{ {pair} }}\n\
+             add element {set} {{ {pair} timeout {remaining_secs}s }}\n"
+        )),
+        "del" => Some(format!(
+            "add element {set} {{ {pair} }}\ndelete element {set} {{ {pair} }}\n"
+        )),
+        _ => None,
+    }
+}
+
+/// The nft input that loads the leases of a dnsmasq lease file that have not
+/// run out at `now` (seconds since the epoch). A line is `<expiry> <mac> <ip>
+/// <host> <client-id>`; expiry 0 is a lease without end.
+pub(crate) fn seed_script(leasefile: &str, now: u64) -> String {
+    let mut script = String::new();
+    for line in leasefile.lines() {
+        let mut cols = line.split_whitespace();
+        let (Some(expiry), Some(mac), Some(ip)) = (cols.next(), cols.next(), cols.next()) else {
+            continue;
+        };
+        let Ok(expiry) = expiry.parse::<u64>() else {
+            continue;
+        };
+        let remaining = if expiry == 0 {
+            DEFAULT_LEASE_SECS
+        } else if expiry > now {
+            expiry - now
+        } else {
+            continue;
+        };
+        if let Some(text) = lease_script("add", mac, ip, remaining) {
+            script.push_str(&text);
+        }
+    }
+    script
+}
+
+/// The links of one `ip -j link` listing.
+pub(crate) fn parse_links(json: &str) -> Result<Vec<NsLink>> {
+    let value: serde_json::Value = serde_json::from_str(json).context("parse ip -j link")?;
+    let Some(items) = value.as_array() else {
+        bail!("ip -j link did not print a list");
+    };
+    Ok(items
+        .iter()
+        .filter_map(|item| {
+            Some(NsLink {
+                ifindex: item.get("ifindex")?.as_u64()?,
+                peer_index: item.get("link_index").and_then(serde_json::Value::as_u64),
+                mac: Mac::parse(item.get("address")?.as_str()?)?,
+            })
+        })
+        .collect())
+}
+
+/// The bridge's ports from `ip -j link show master <bridge>`. A port whose
+/// name is not plain is left out: it never gets a binding.
+pub(crate) fn parse_ports(json: &str) -> Result<Vec<HostPort>> {
+    let value: serde_json::Value = serde_json::from_str(json).context("parse ip -j link")?;
+    let Some(items) = value.as_array() else {
+        bail!("ip -j link did not print a list");
+    };
+    Ok(items
+        .iter()
+        .filter_map(|item| {
+            let name = item.get("ifname")?.as_str()?;
+            if !is_plain_name(name) {
+                return None;
+            }
+            Some(HostPort {
+                name: name.to_string(),
+                ifindex: item.get("ifindex")?.as_u64()?,
+                peer_index: item.get("link_index")?.as_u64()?,
+            })
+        })
+        .collect())
+}
+
+/// The leader of a machine's state file, or `None` when it is not a container.
+pub(crate) fn container_leader(state: &str) -> Option<u32> {
+    let mut leader = None;
+    let mut container = false;
+    for line in state.lines() {
+        match line.split_once('=') {
+            Some(("CLASS", "container")) => container = true,
+            Some(("LEADER", pid)) => leader = pid.parse().ok(),
+            _ => {}
+        }
+    }
+    leader.filter(|_| container)
+}
+
+/// Run `program` with `args` and return its stdout.
+fn output_of(program: &str, args: &[&str]) -> Result<String> {
+    let out = Command::new(program)
+        .args(args)
+        .output()
+        .with_context(|| format!("run {program}"))?;
+    if !out.status.success() {
+        bail!(
+            "{program} {} exited {}: {}",
+            args.join(" "),
+            out.status,
+            String::from_utf8_lossy(&out.stderr).trim()
+        );
+    }
+    Ok(String::from_utf8_lossy(&out.stdout).into_owned())
+}
+
+/// Feed `script` to `nft -f -`.
+fn nft(script: &str) -> Result<()> {
+    let mut child = Command::new("nft")
+        .args(["-f", "-"])
+        .stdin(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .context("run nft")?;
+    child
+        .stdin
+        .take()
+        .context("nft stdin")?
+        .write_all(script.as_bytes())
+        .context("write to nft")?;
+    let out = child.wait_with_output().context("wait for nft")?;
+    if !out.status.success() {
+        bail!(
+            "nft exited {}: {}",
+            out.status,
+            String::from_utf8_lossy(&out.stderr).trim()
+        );
+    }
+    Ok(())
+}
+
+/// The links of every registered container, by machine name. A machine whose
+/// namespace cannot be read is reported and skipped.
+fn machine_links(dir: &Path) -> Vec<(String, Vec<NsLink>)> {
+    let Ok(entries) = std::fs::read_dir(dir) else {
+        return Vec::new();
+    };
+    let mut machines = Vec::new();
+    for entry in entries.flatten() {
+        let Some(name) = entry.file_name().to_str().map(str::to_string) else {
+            continue;
+        };
+        let Ok(state) = std::fs::read_to_string(entry.path()) else {
+            continue;
+        };
+        let Some(leader) = container_leader(&state) else {
+            continue;
+        };
+        let links = output_of(
+            "nsenter",
+            &[
+                "--target",
+                &leader.to_string(),
+                "--net",
+                "ip",
+                "-j",
+                "link",
+                "show",
+            ],
+        )
+        .and_then(|json| parse_links(&json));
+        match links {
+            Ok(links) => machines.push((name, links)),
+            Err(e) => eprintln!("{PIN}: machine {name}: {e:#}"),
+        }
+    }
+    machines
+}
+
+/// One pass: look at the bridge's ports and the containers, and make the
+/// table's bindings match. Returns the ports of the bridge and the resolution.
+fn pin_once(bridge: &str) -> Result<(Vec<HostPort>, Resolution)> {
+    let ports = parse_ports(&output_of("ip", &["-j", "link", "show", "master", bridge])?)?;
+    let machines = machine_links(Path::new(MACHINES_DIR));
+    let state_path = std::env::var("HIVE_EGRESS_STATE").unwrap_or_else(|_| STATE.into());
+    let earlier = parse_state(&std::fs::read_to_string(&state_path).unwrap_or_default());
+    let resolution = resolve(&ports, &machines, &earlier);
+    nft(&pin_script(&resolution.pins))?;
+    std::fs::write(&state_path, state_text(&resolution.pins))
+        .with_context(|| format!("write {state_path}"))?;
+    for pin in resolution.pins.iter().filter(|p| !earlier.contains(p)) {
+        eprintln!("{PIN}: {} bound to {} ({})", pin.port, pin.mac, pin.machine);
+    }
+    Ok((ports, resolution))
+}
+
+/// Wait for the lock of the bindings. It is released when the file is dropped.
+fn lock() -> Result<File> {
+    let path = std::env::var("HIVE_EGRESS_LOCK").unwrap_or_else(|_| LOCK.into());
+    let file = OpenOptions::new()
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .open(&path)
+        .with_context(|| format!("open {path}"))?;
+    file.lock().with_context(|| format!("lock {path}"))?;
+    Ok(file)
+}
+
+fn now_secs() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_or(0, |d| d.as_secs())
+}
+
+/// `egress-pin [--seed] [<port>]`.
+fn pin(args: &[String]) -> Result<()> {
+    let bridge = std::env::var(BRIDGE_ENV).with_context(|| format!("{BRIDGE_ENV} is not set"))?;
+    let mut seed = false;
+    let mut wait_for = None;
+    for arg in args {
+        match arg.as_str() {
+            "--seed" => seed = true,
+            port if wait_for.is_none() && is_plain_name(port) => wait_for = Some(port.to_string()),
+            other => bail!("usage: hive-priv {PIN} [--seed] [<port>] (got {other:?})"),
+        }
+    }
+    if seed {
+        let path = std::env::var("HIVE_EGRESS_LEASEFILE").unwrap_or_else(|_| LEASEFILE.into());
+        let leases = std::fs::read_to_string(&path).unwrap_or_default();
+        let script = seed_script(&leases, now_secs());
+        if !script.is_empty() {
+            let _lock = lock()?;
+            nft(&script).context("load leases")?;
+        }
+    }
+    let deadline = Instant::now() + WAIT;
+    loop {
+        let (ports, resolution) = {
+            let _lock = lock()?;
+            pin_once(&bridge)?
+        };
+        let Some(port) = &wait_for else {
+            report(&resolution, None);
+            return Ok(());
+        };
+        let bound = resolution.pins.iter().any(|p| &p.port == port);
+        let present = ports.iter().any(|p| &p.name == port);
+        if bound || !present || Instant::now() >= deadline {
+            report(&resolution, Some(port));
+            if present && !bound {
+                bail!("port {port} stays unbound");
+            }
+            return Ok(());
+        }
+        std::thread::sleep(POLL);
+    }
+}
+
+/// Print why ports stay unbound; with `port`, only that one.
+fn report(resolution: &Resolution, port: Option<&str>) {
+    for (name, why) in &resolution.unbound {
+        if port.is_none_or(|p| p == name) {
+            eprintln!("{PIN}: {name} stays unbound: {why}");
+        }
+    }
+}
+
+/// `egress-lease <action> [<mac> <ip> [<host>]]`. Nothing is loaded for an event
+/// that changes no lease.
+fn lease(args: &[String]) -> Result<()> {
+    let [action, mac, ip, ..] = args else {
+        return Ok(());
+    };
+    let remaining = std::env::var("DNSMASQ_TIME_REMAINING")
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(DEFAULT_LEASE_SECS);
+    match lease_script(action, mac, ip, remaining) {
+        Some(script) => nft(&script),
+        None => Ok(()),
+    }
+}
+
+/// Run `egress-pin` or `egress-lease` when `args` name one; `None` otherwise.
+/// Returns the exit code.
+pub(crate) fn run(args: &[String]) -> Option<i32> {
+    let (name, rest) = args.split_first()?;
+    let result = match name.as_str() {
+        PIN => pin(rest),
+        LEASE => lease(rest),
+        _ => return None,
+    };
+    Some(match result {
+        Ok(()) => 0,
+        Err(e) => {
+            eprintln!("{name}: {e:#}");
+            1
+        }
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn mac(last: u8) -> Mac {
+        Mac([0x02, 0, 0, 0, 0, last])
+    }
+
+    fn port(name: &str, ifindex: u64, peer_index: u64) -> HostPort {
+        HostPort {
+            name: name.into(),
+            ifindex,
+            peer_index,
+        }
+    }
+
+    fn pin(port: &str, ifindex: u64, last: u8, machine: &str) -> Pin {
+        Pin {
+            port: port.into(),
+            ifindex,
+            mac: mac(last),
+            machine: machine.into(),
+        }
+    }
+
+    fn link(ifindex: u64, peer_index: Option<u64>, m: Mac) -> NsLink {
+        NsLink {
+            ifindex,
+            peer_index,
+            mac: m,
+        }
+    }
+
+    #[test]
+    fn mac_round_trip_and_rejects() {
+        assert_eq!(Mac::parse("02:00:00:00:00:0A"), Some(mac(10)));
+        assert_eq!(mac(10).to_string(), "02:00:00:00:00:0a");
+        assert_eq!(mac(10).as_integer(), "0x02000000000a");
+        for bad in [
+            "",
+            "02:00:00:00:00",
+            "02:00:00:00:00:0a:01",
+            "2:0:0:0:0:a",
+            "zz:00:00:00:00:00",
+        ] {
+            assert_eq!(Mac::parse(bad), None, "{bad}");
+        }
+    }
+
+    #[test]
+    fn a_port_is_bound_to_the_link_at_the_other_end_of_its_pair() {
+        let ports = [port("vb-a", 10, 2), port("vb-b", 11, 2)];
+        let machines = [
+            ("a".to_string(), vec![link(2, Some(10), mac(1))]),
+            ("b".to_string(), vec![link(2, Some(11), mac(2))]),
+        ];
+        let got = resolve(&ports, &machines, &[]);
+        assert_eq!(
+            got.pins,
+            vec![pin("vb-a", 10, 1, "a"), pin("vb-b", 11, 2, "b")]
+        );
+        assert!(got.unbound.is_empty());
+    }
+
+    #[test]
+    fn a_link_a_container_made_in_its_own_namespace_claims_nothing() {
+        // The container `evil` makes a veth pair whose peer index is 10, the
+        // index of `vb-a` on the host; its link's own index is not the one
+        // `vb-a` points at, so it does not match.
+        let ports = [port("vb-a", 10, 2)];
+        let machines = [(
+            "evil".to_string(),
+            vec![link(7, Some(10), mac(9)), link(3, Some(11), mac(1))],
+        )];
+        let got = resolve(&ports, &machines, &[]);
+        assert!(got.pins.is_empty());
+        assert_eq!(got.unbound.len(), 1);
+    }
+
+    #[test]
+    fn a_bound_port_keeps_its_mac_when_the_link_changes() {
+        let ports = [port("vb-a", 10, 2)];
+        let earlier = [pin("vb-a", 10, 1, "a")];
+        let changed = [("a".to_string(), vec![link(2, Some(10), mac(9))])];
+        let kept = resolve(&ports, &changed, &earlier);
+        assert_eq!(kept.pins, earlier);
+        // The same name with another index is a new port: it is read again.
+        let remade = resolve(
+            &[port("vb-a", 12, 2)],
+            &[("a".to_string(), vec![link(2, Some(12), mac(9))])],
+            &earlier,
+        );
+        assert_eq!(remade.pins, vec![pin("vb-a", 12, 9, "a")]);
+    }
+
+    #[test]
+    fn the_state_file_round_trips() {
+        let pins = vec![pin("vb-a", 10, 1, "a"), pin("vb-b", 11, 2, "b")];
+        assert_eq!(parse_state(&state_text(&pins)), pins);
+        assert_eq!(parse_state("junk\nvb-a x 02:00:00:00:00:01 a\n"), vec![]);
+    }
+
+    #[test]
+    fn a_port_nobody_owns_stays_unbound() {
+        let got = resolve(&[port("vb-a", 10, 2)], &[], &[]);
+        assert!(got.pins.is_empty());
+        assert_eq!(got.unbound[0].0, "vb-a");
+    }
+
+    #[test]
+    fn a_mac_two_ports_have_binds_neither() {
+        let ports = [
+            port("vb-a", 10, 2),
+            port("vb-b", 11, 2),
+            port("vb-c", 12, 2),
+        ];
+        let machines = [
+            ("a".to_string(), vec![link(2, Some(10), mac(1))]),
+            ("b".to_string(), vec![link(2, Some(11), mac(1))]),
+            ("c".to_string(), vec![link(2, Some(12), mac(3))]),
+        ];
+        let got = resolve(&ports, &machines, &[]);
+        assert_eq!(got.pins, vec![pin("vb-c", 12, 3, "c")]);
+        assert_eq!(got.unbound.len(), 2);
+    }
+
+    #[test]
+    fn a_multicast_or_zero_mac_is_not_bound() {
+        let ports = [port("vb-a", 10, 2), port("vb-b", 11, 2)];
+        let machines = [
+            ("a".to_string(), vec![link(2, Some(10), Mac([0xff; 6]))]),
+            ("b".to_string(), vec![link(2, Some(11), Mac([0; 6]))]),
+        ];
+        assert!(resolve(&ports, &machines, &[]).pins.is_empty());
+    }
+
+    #[test]
+    fn pin_script_replaces_both_sets_in_one_script() {
+        let script = pin_script(&[pin("vb-a", 10, 10, "a")]);
+        assert_eq!(
+            script,
+            "flush set bridge hive_egress port_mac\n\
+             flush set bridge hive_egress port_chaddr\n\
+             add element bridge hive_egress port_mac { \"vb-a\" . 02:00:00:00:00:0a }\n\
+             add element bridge hive_egress port_chaddr { \"vb-a\" . 0x02000000000a }\n"
+        );
+        assert!(!pin_script(&[]).contains("add element"));
+    }
+
+    #[test]
+    fn lease_events_become_nft_input() {
+        let add = lease_script("add", "02:00:00:00:00:0a", "10.42.0.10", 90).unwrap();
+        assert!(add.ends_with("add element bridge hive_egress lease_v4 { 02:00:00:00:00:0a . 10.42.0.10 timeout 90s }\n"));
+        assert!(add.contains("delete element"));
+        assert_eq!(
+            lease_script("old", "02:00:00:00:00:0a", "10.42.0.10", 90),
+            Some(add)
+        );
+        let del = lease_script("del", "02:00:00:00:00:0a", "10.42.0.10", 90).unwrap();
+        assert!(del.contains("delete element") && !del.contains("timeout"));
+        assert_eq!(
+            lease_script("init", "02:00:00:00:00:0a", "10.42.0.10", 90),
+            None
+        );
+        assert_eq!(
+            lease_script("add", "02:00:00:00:00:0a", "fe80::1", 90),
+            None
+        );
+        assert_eq!(lease_script("add", "nonsense", "10.42.0.10", 90), None);
+    }
+
+    #[test]
+    fn seeding_skips_expired_and_malformed_leases() {
+        let leases = "2000 02:00:00:00:00:0a 10.42.0.10 a *\n\
+                      900 02:00:00:00:00:0b 10.42.0.11 b *\n\
+                      0 02:00:00:00:00:0c 10.42.0.12 c *\n\
+                      garbage\n";
+        let script = seed_script(leases, 1000);
+        assert!(script.contains("02:00:00:00:00:0a . 10.42.0.10 timeout 1000s"));
+        assert!(!script.contains("10.42.0.11"));
+        assert!(script.contains("10.42.0.12 timeout 3600s"));
+    }
+
+    #[test]
+    fn ip_json_is_read_for_links_and_ports() {
+        let json = r#"[{"ifindex":1,"ifname":"lo","address":"00:00:00:00:00:00"},
+            {"ifindex":2,"link_index":10,"ifname":"eth0","address":"7a:38:b5:d1:ff:84","link_netnsid":0}]"#;
+        let links = parse_links(json).unwrap();
+        assert_eq!(links[1].ifindex, 2);
+        assert_eq!(links[1].peer_index, Some(10));
+        assert_eq!(links[0].peer_index, None);
+        let ports = parse_ports(
+            r#"[{"ifindex":10,"link_index":2,"ifname":"vb-a"},{"ifindex":11,"link_index":2,"ifname":"bad\"name"}]"#,
+        )
+        .unwrap();
+        assert_eq!(ports, vec![port("vb-a", 10, 2)]);
+        assert!(parse_links("{}").is_err());
+    }
+
+    #[test]
+    fn only_a_container_with_a_leader_is_read() {
+        assert_eq!(
+            container_leader("CLASS=container\nLEADER=1234\n"),
+            Some(1234)
+        );
+        assert_eq!(container_leader("CLASS=vm\nLEADER=1234\n"), None);
+        assert_eq!(container_leader("CLASS=container\n"), None);
+    }
+
+    #[test]
+    fn plain_names_only() {
+        assert!(is_plain_name("vb-h-atlas"));
+        for bad in ["", "a b", "a\"b", "0123456789abcdef"] {
+            assert!(!is_plain_name(bad), "{bad:?}");
+        }
+    }
+
+    #[test]
+    fn other_sub_commands_are_not_ours() {
+        assert_eq!(run(&["check-rootfs".into()]), None);
+        assert_eq!(run(&[]), None);
+    }
+}

hive-priv/src/main.rs

@@ -22,7 +22,8 @@
 //! systemd drop-ins), [`rootfs`] (the salvage of a container's root before it
 //! turns ephemeral, with [`salvage_io`] for its space, reflink and lock calls),
 //! [`prestart`] (the check of a container's root that runs before each start,
-//! as a sub-command), [`storage`] (btrfs subvolumes, quotas, snapshots, disk
+//! as a sub-command), [`egress`] (the bridge anti-spoofing bindings, also
+//! sub-commands), [`storage`] (btrfs subvolumes, quotas, snapshots, disk
 //! measurement), [`files`] (pause marker, socket dir, legacy tmpfiles),
 //! [`infra`] (gateway nginx, infra containers) and [`closure`] (the swarm-signature
 //! check before an apply, and the cache-facts refresh). Helpers more than one family uses are in [`shared`].
@@ -30,6 +31,7 @@
 mod closure;
 mod container;
 mod container_config;
+mod egress;
 mod files;
 mod infra;
 mod prestart;
@@ -53,6 +55,11 @@ async fn main() -> Result<()> {
     if let Some(code) = prestart::run(&args) {
         std::process::exit(code);
     }
+    // `egress-pin` and `egress-lease` are run by the hive-egress unit, its udev
+    // trigger and dnsmasq: sub-commands like the one above, not requests.
+    if let Some(code) = egress::run(&args) {
+        std::process::exit(code);
+    }
 
     tracing_subscriber::fmt()
         .with_env_filter(