Watch
0
0
Fork
You've already forked hyperhive
0
hyperhive/hive-priv/src/egress.rs
atlas 5189e8fb4b egress: refresh a leased pair on dhcp renewal and default anti-spoof to log
dnsmasq runs the dhcp-script on a renewal only with script-on-renewal, so a pair loaded at the first lease timed out while the lease kept renewing and the bridge table dropped the container's traffic. Set script-on-renewal, time pairs out remaining + 300s, default antiSpoof to log, and run the vm test through renewals with drop.
2026-11-13 23:55:12 +01:00

1339 lines
48 KiB
Rust

//! The per-port bindings behind the `bridge hive_egress` anti-spoofing table
//! (`nix/host-modules/hive-egress.nix`), as four sub-commands run by systemd
//! units, by dnsmasq and by squid, never through the socket.
//!
//! - `egress-load <ruleset-file>` replaces the tables with the ruleset, the
//! current bindings and the leases of dnsmasq's lease file, in one nft
//! transaction. The bindings are looked up first, while the old tables still
//! filter. A lease load that fails is a warning; the ruleset then loads with
//! the bindings alone.
//! - `egress-pin [<port>]` binds each port of the agent bridge to the MAC its
//! container uses. 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. udev runs it per new port, a timer for the ports
//! still unbound.
//! - `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.
//! - `egress-whois` is squid's external ACL helper: it answers a source address
//! with the agent that holds it ([`agent_of`]).
//!
//! A port's owner is found from the host side: the port's peer index is the
//! index of one link in a registered container's network namespace (the leader
//! comes from `/run/systemd/machines`), and that link's peer index is the port's
//! own index. A veth pair the container makes inside its namespace can't claim
//! another port.
//!
//! The bound MAC is the one the link has when the port is first looked at. The
//! binding stays for as long as the port exists (`/run/hive-egress-pins`), and
//! a kept binding is never given up: a port that claims a MAC a kept binding
//! holds, or that another new claim shares, stays unbound and is looked at
//! again on the next pass.
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};
use crate::egress_ech::same_host;
/// The sub-command squid runs to learn which agent holds a source address.
pub(crate) const WHOIS: &str = "egress-whois";
/// 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 sub-command that loads the tables.
pub(crate) const LOAD: &str = "egress-load";
/// 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;
/// Seconds a leased pair outlives the lease it was loaded for. A renewal whose
/// script run comes a little late then finds the pair still in the set.
const LEASE_MARGIN_SECS: u64 = 300;
/// 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 file of the unrestricted agents' names, one per line, that
/// `HIVE_EGRESS_UNRESTRICTED` names. Unset or unreadable lists no agent.
const UNRESTRICTED_ENV: &str = "HIVE_EGRESS_UNRESTRICTED";
/// The prefix of an agent container's machine name.
const AGENT_PREFIX: &str = "h-";
/// The nft table the commands write into.
const TABLE: &str = "bridge hive_egress";
/// The table whose `direct_mac` set the commands fill.
const INET_TABLE: &str = "inet 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, and no other port takes its MAC. A
/// kept binding is never given up, so a port can only get a binding for a MAC
/// nobody holds. A port no machine owns, an unusable MAC, a MAC a kept binding
/// holds and a MAC two new claims share all leave the port unbound.
pub(crate) fn resolve(
ports: &[HostPort],
machines: &[(String, Vec<NsLink>)],
earlier: &[Pin],
) -> Resolution {
let mut kept: Vec<Pin> = Vec::new();
let mut claims: Vec<Pin> = Vec::new();
let mut out = Resolution::default();
for port in ports {
if let Some(pin) = earlier
.iter()
.find(|p| p.port == port.name && p.ifindex == port.ifindex)
{
kept.push(pin.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)) => claims.push(Pin {
port: port.name.clone(),
ifindex: port.ifindex,
mac,
machine: machine.to_string(),
}),
}
}
for claim in &claims {
let holders: Vec<&str> = kept
.iter()
.chain(claims.iter())
.filter(|p| p.mac == claim.mac && p.port != claim.port)
.map(|p| p.port.as_str())
.collect();
if holders.is_empty() {
out.pins.push(claim.clone());
} else {
out.unbound.push((
claim.port.clone(),
format!(
"{} has {}, which {} also has",
claim.machine,
claim.mac,
holders.join(", ")
),
));
}
}
kept.append(&mut out.pins);
out.pins = kept;
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 MACs of the ports squid does not intercept: those of containers that are
/// not agents (`h-<name>`), and of the agents `unrestricted` names. A MAC in no
/// binding is in none of them, so a flow from an unknown source is intercepted
/// and squid refuses it.
pub(crate) fn direct_macs(pins: &[Pin], unrestricted: &[String]) -> Vec<Mac> {
let mut macs: Vec<Mac> = pins
.iter()
.filter(|p| match p.machine.strip_prefix(AGENT_PREFIX) {
Some(agent) => unrestricted.iter().any(|u| u == agent),
None => true,
})
.map(|p| p.mac)
.collect();
macs.sort_by_key(|m| m.0);
macs.dedup();
macs
}
/// The nft input that adds the bindings `pins` to the two sets of the bridge
/// table and the [`direct_macs`] to the exemption set of the inet table.
pub(crate) fn pin_elements(pins: &[Pin], unrestricted: &[String]) -> String {
let mut script = String::new();
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()
);
}
for mac in direct_macs(pins, unrestricted) {
let _ = writeln!(script, "add element {INET_TABLE} direct_mac {{ {mac} }}");
}
script
}
/// The nft input that makes the bindings and the exemptions exactly those of
/// `pins` and `unrestricted`, in one transaction: a reader never sees a set
/// empty in between.
pub(crate) fn pin_script(pins: &[Pin], unrestricted: &[String]) -> String {
format!(
"flush set {TABLE} port_mac\nflush set {TABLE} port_chaddr\n\
flush set {INET_TABLE} direct_mac\n{}",
pin_elements(pins, unrestricted)
)
}
/// The names in the unrestricted file: trimmed, without blank lines and
/// comments. A file that cannot be read lists no one, so every agent is then
/// intercepted.
fn unrestricted_names() -> Vec<String> {
let Some(path) = std::env::var_os(UNRESTRICTED_ENV).filter(|p| !p.is_empty()) else {
return Vec::new();
};
std::fs::read_to_string(path)
.unwrap_or_default()
.lines()
.map(str::trim)
.filter(|l| !l.is_empty() && !l.starts_with('#'))
.map(str::to_owned)
.collect()
}
/// The agent whose container holds the IPv4 address `ip`, as dnsmasq's lease
/// file and the bindings of the ports say: the lease names a MAC, the MAC is
/// bound to one port, and that port's machine is `h-<agent>`. An address in the
/// pool of the ECH relay (`egress_ech`) stands for the leased address it was
/// derived from. `None` when any
/// step is missing or ambiguous (two leases for the address under different
/// MACs, a MAC bound twice, a machine that is no agent, a name that is no
/// identifier): a flow whose agent cannot be told is refused, never guessed.
pub(crate) fn agent_of(ip: &str, leases: &str, pins: &[Pin], now: u64) -> Option<String> {
let ip: std::net::Ipv4Addr = ip.parse().ok()?;
let mut macs = leases.lines().filter_map(|line| {
let mut cols = line.split_whitespace();
let (expiry, mac, addr) = (cols.next()?, cols.next()?, cols.next()?);
let live = match expiry.parse::<u64>().ok()? {
0 => true,
t => t > now,
};
(live && same_host(ip, addr.parse::<std::net::Ipv4Addr>().ok()?)).then(|| Mac::parse(mac))
});
let mac = macs.next()??;
if macs.any(|other| other != Some(mac)) {
return None;
}
let mut machines = pins.iter().filter(|p| p.mac == mac).map(|p| &p.machine);
let machine = machines.next()?;
if machines.any(|other| other != machine) {
return None;
}
let agent = machine.strip_prefix(AGENT_PREFIX)?;
hive_types::Ident::parse(agent).ok()?;
Some(agent.to_owned())
}
/// [`agent_of`] on the files as they are now, which are read for every call.
pub(crate) fn agent_of_now(ip: &str) -> Option<String> {
let leasefile = std::env::var("HIVE_EGRESS_LEASEFILE").unwrap_or_else(|_| LEASEFILE.into());
let pins = parse_state(&std::fs::read_to_string(state_path()).unwrap_or_default());
let leases = std::fs::read_to_string(leasefile).unwrap_or_default();
agent_of(ip, &leases, &pins, now_secs())
}
/// `egress-whois`: squid's external ACL helper. Each line on stdin starts with
/// a source address; the answer is `OK user=<agent>` or `ERR`. The two files
/// are read for every line, since squid caches the answers itself.
fn whois() -> Result<()> {
use std::io::{BufRead as _, Write as _};
let stdin = std::io::stdin();
let mut stdout = std::io::stdout().lock();
for line in stdin.lock().lines() {
let line = line.context("read stdin")?;
let ip = line.split_whitespace().next().unwrap_or_default();
match agent_of_now(ip) {
Some(agent) => writeln!(stdout, "OK user={agent}"),
None => writeln!(stdout, "ERR"),
}
.context("write stdout")?;
stdout.flush().context("flush stdout")?;
}
Ok(())
}
/// 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 `remaining_secs`
/// plus [`LEASE_MARGIN_SECS`] from now, so every renewal moves the timeout
/// forward; `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}");
let timeout = remaining_secs.saturating_add(LEASE_MARGIN_SECS);
match action {
"add" | "old" => Some(format!(
"add element {set} {{ {pair} }}\n\
delete element {set} {{ {pair} }}\n\
add element {set} {{ {pair} timeout {timeout}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
}
/// The ports no kept binding covers: the ones a pass has to look up.
pub(crate) fn ports_to_resolve<'a>(ports: &'a [HostPort], earlier: &[Pin]) -> Vec<&'a HostPort> {
ports
.iter()
.filter(|p| {
!earlier
.iter()
.any(|k| k.port == p.name && k.ifindex == p.ifindex)
})
.collect()
}
/// What a pass found out.
struct Survey {
earlier: Vec<Pin>,
resolution: Resolution,
}
fn state_path() -> String {
std::env::var("HIVE_EGRESS_STATE").unwrap_or_else(|_| STATE.into())
}
/// Look at the bridge's ports, the earlier bindings and, when a port has none,
/// the containers. Changes nothing.
fn survey(bridge: &str) -> Result<Survey> {
let ports = parse_ports(&output_of("ip", &["-j", "link", "show", "master", bridge])?)?;
let earlier = parse_state(&std::fs::read_to_string(state_path()).unwrap_or_default());
let machines = if ports_to_resolve(&ports, &earlier).is_empty() {
Vec::new()
} else {
machine_links(Path::new(MACHINES_DIR))
};
let resolution = resolve(&ports, &machines, &earlier);
Ok(Survey {
earlier,
resolution,
})
}
/// Keep the bindings for the next pass and print the new ones.
fn commit(survey: &Survey) -> Result<()> {
let path = state_path();
std::fs::write(&path, state_text(&survey.resolution.pins))
.with_context(|| format!("write {path}"))?;
for pin in survey
.resolution
.pins
.iter()
.filter(|p| !survey.earlier.contains(p))
{
eprintln!("{PIN}: {} bound to {} ({})", pin.port, pin.mac, pin.machine);
}
Ok(())
}
/// 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())
}
/// Whose "stays unbound" lines a pass prints.
#[derive(Clone, Copy)]
enum Report<'a> {
All,
Port(&'a str),
Nothing,
}
fn report(resolution: &Resolution, which: Report<'_>) {
for (name, why) in &resolution.unbound {
let show = match which {
Report::All => true,
Report::Port(port) => port == name,
Report::Nothing => false,
};
if show {
eprintln!("{PIN}: {name} stays unbound: {why}");
}
}
}
/// `egress-pin [<port>]`: one pass, or with `<port>` passes until that port is
/// bound or gone. Without `<port>` it prints only what it binds: the timer
/// runs it again and again for the ports still unbound.
fn pin(args: &[String]) -> Result<()> {
let bridge = std::env::var(BRIDGE_ENV).with_context(|| format!("{BRIDGE_ENV} is not set"))?;
let wait_for = match args {
[] => None,
[port] if is_plain_name(port) => Some(port.as_str()),
_ => bail!("usage: hive-priv {PIN} [<port>]"),
};
let deadline = Instant::now() + WAIT;
loop {
let survey = {
let _lock = lock()?;
let survey = survey(&bridge)?;
nft(&pin_script(&survey.resolution.pins, &unrestricted_names()))?;
commit(&survey)?;
survey
};
let Some(port) = wait_for else {
report(&survey.resolution, Report::Nothing);
return Ok(());
};
let bound = survey.resolution.pins.iter().any(|p| p.port == port);
let present = port_exists(port);
if bound || !present || Instant::now() >= deadline {
report(&survey.resolution, Report::Port(port));
if present && !bound {
bail!("port {port} stays unbound");
}
return Ok(());
}
std::thread::sleep(POLL);
}
}
/// Whether an interface of this name exists, on any bridge or none: a port is
/// created before it joins the bridge.
fn port_exists(name: &str) -> bool {
Command::new("ip")
.args(["link", "show", "dev", name])
.output()
.is_ok_and(|o| o.status.success())
}
/// Load `ruleset` with the bindings and the leases in the same transaction, so
/// no frame meets the new table with empty sets. A lease load that `run`
/// refuses is a warning: the ruleset loads again with the bindings alone.
pub(crate) fn load_ruleset(
ruleset: &str,
pins: &str,
leases: &str,
run: impl Fn(&str) -> Result<()>,
) -> Result<()> {
match run(&format!("{ruleset}\n{pins}{leases}")) {
Err(e) if !leases.is_empty() => {
eprintln!("{LOAD}: loading the leases failed, loading without them: {e:#}");
run(&format!("{ruleset}\n{pins}")).context("load the ruleset")
}
other => other,
}
}
/// `egress-load <ruleset-file>`: replace the tables with the ruleset and the
/// current bindings and leases. The bindings are looked up first, while the old
/// tables still filter. A lookup that fails loads the ruleset with none: the
/// ports stay unbound until the timer or udev binds them.
fn load(args: &[String]) -> Result<()> {
let bridge = std::env::var(BRIDGE_ENV).with_context(|| format!("{BRIDGE_ENV} is not set"))?;
let [file] = args else {
bail!("usage: hive-priv {LOAD} <ruleset-file>");
};
let ruleset = std::fs::read_to_string(file).with_context(|| format!("read {file}"))?;
let path = std::env::var("HIVE_EGRESS_LEASEFILE").unwrap_or_else(|_| LEASEFILE.into());
let leases = seed_script(
&std::fs::read_to_string(&path).unwrap_or_default(),
now_secs(),
);
let _lock = lock()?;
let found = survey(&bridge);
let pins = match &found {
Ok(found) => pin_elements(&found.resolution.pins, &unrestricted_names()),
Err(e) => {
eprintln!("{LOAD}: binding the ports failed, loading without bindings: {e:#}");
String::new()
}
};
load_ruleset(&ruleset, &pins, &leases, nft)?;
if let Ok(found) = &found {
commit(found)?;
report(&found.resolution, Report::All);
}
Ok(())
}
/// `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),
WHOIS => whois(),
LOAD => load(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 a_kept_binding_is_never_given_up_to_a_new_claim_on_its_mac() {
let ports = [port("vb-a", 10, 2), port("vb-b", 11, 2)];
let earlier = [pin("vb-a", 10, 1, "a")];
let machines = [
("a".to_string(), vec![link(2, Some(10), mac(1))]),
("b".to_string(), vec![link(2, Some(11), mac(1))]),
];
let got = resolve(&ports, &machines, &earlier);
assert_eq!(got.pins, vec![pin("vb-a", 10, 1, "a")]);
assert_eq!(got.unbound.len(), 1);
assert_eq!(got.unbound[0].0, "vb-b");
}
#[test]
fn a_new_claim_on_a_free_mac_binds_next_to_a_kept_binding() {
let ports = [port("vb-a", 10, 2), port("vb-b", 11, 2)];
let earlier = [pin("vb-a", 10, 1, "a")];
let machines = [("b".to_string(), vec![link(2, Some(11), mac(2))])];
let got = resolve(&ports, &machines, &earlier);
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_port_refused_for_a_taken_mac_binds_on_a_later_pass_once_it_has_its_own() {
let ports = [port("vb-a", 10, 2), port("vb-b", 11, 2)];
let earlier = [pin("vb-a", 10, 1, "a")];
let rival = [("b".to_string(), vec![link(2, Some(11), mac(1))])];
assert!(resolve(&ports, &rival, &earlier).pins.len() == 1);
// The rival restores its own MAC; the next pass reads its link again.
let restored = [("b".to_string(), vec![link(2, Some(11), mac(2))])];
let got = resolve(&ports, &restored, &earlier);
assert_eq!(
got.pins,
vec![pin("vb-a", 10, 1, "a"), pin("vb-b", 11, 2, "b")]
);
}
#[test]
fn only_ports_without_a_kept_binding_are_looked_up() {
let ports = [port("vb-a", 10, 2), port("vb-b", 11, 2)];
let a = pin("vb-a", 10, 1, "a");
assert_eq!(ports_to_resolve(&ports, &[]).len(), 2);
assert_eq!(
ports_to_resolve(&ports, std::slice::from_ref(&a))[0].name,
"vb-b"
);
let b = pin("vb-b", 11, 2, "b");
assert!(ports_to_resolve(&ports, &[a, b]).is_empty());
// The same name with another index is a new port.
assert_eq!(
ports_to_resolve(&[port("vb-a", 12, 2)], &[pin("vb-a", 10, 1, "a")]).len(),
1
);
}
#[test]
fn the_ruleset_loads_with_the_bindings_and_leases_in_one_script() {
let calls = std::cell::RefCell::new(Vec::new());
let pins = pin_elements(&[pin("vb-a", 10, 10, "a")], &[]);
let leases = lease_script("add", "02:00:00:00:00:0a", "10.42.0.10", 90).unwrap();
load_ruleset("table bridge hive_egress { }", &pins, &leases, |s| {
calls.borrow_mut().push(s.to_string());
Ok(())
})
.unwrap();
let calls = calls.into_inner();
assert_eq!(calls.len(), 1, "one transaction");
let script = &calls[0];
assert!(script.starts_with("table bridge hive_egress { }"));
assert!(script.contains("port_mac { \"vb-a\" . 02:00:00:00:00:0a }"));
assert!(script.contains("lease_v4 { 02:00:00:00:00:0a . 10.42.0.10 timeout 390s }"));
assert!(!script.contains("flush set"), "the sets are new");
}
#[test]
fn a_lease_load_that_fails_is_a_warning_and_the_bindings_still_load() {
let calls = std::cell::RefCell::new(Vec::new());
let pins = pin_elements(&[pin("vb-a", 10, 10, "a")], &[]);
let leases = lease_script("add", "02:00:00:00:00:0a", "10.42.0.10", 90).unwrap();
let result = load_ruleset("RULESET", &pins, &leases, |s| {
calls.borrow_mut().push(s.to_string());
if s.contains("lease_v4") {
anyhow::bail!("lease_v4 refused")
}
Ok(())
});
assert!(result.is_ok());
let calls = calls.into_inner();
assert_eq!(calls.len(), 2);
assert!(calls[1].contains("port_mac") && !calls[1].contains("lease_v4"));
}
#[test]
fn a_ruleset_that_fails_without_leases_is_an_error() {
let fails = |_: &str| -> Result<()> { anyhow::bail!("refused") };
assert!(load_ruleset("RULESET", "", "", fails).is_err());
assert!(load_ruleset("RULESET", "", "x", fails).is_err());
}
#[test]
fn pin_script_replaces_all_three_sets_in_one_script() {
let script = pin_script(&[pin("vb-h-a", 10, 10, "h-a")], &[]);
assert_eq!(
script,
"flush set bridge hive_egress port_mac\n\
flush set bridge hive_egress port_chaddr\n\
flush set inet hive_egress direct_mac\n\
add element bridge hive_egress port_mac { \"vb-h-a\" . 02:00:00:00:00:0a }\n\
add element bridge hive_egress port_chaddr { \"vb-h-a\" . 0x02000000000a }\n"
);
assert!(!pin_script(&[], &[]).contains("add element"));
}
/// Service containers and the agents the file lists are exempt; an agent the
/// file does not list, and one it lists under another name, are not.
#[test]
fn only_service_containers_and_unrestricted_agents_are_exempt() {
let pins = [
pin("vb-h-atlas", 10, 1, "h-atlas"),
pin("vb-h-uma", 11, 2, "h-uma"),
pin("vb-matrix", 12, 3, "hive-matrix"),
pin("vb-h-umax", 13, 4, "h-umax"),
];
let listed = ["uma".to_string()];
assert_eq!(direct_macs(&pins, &listed), vec![mac(2), mac(3)]);
// Control: with nothing listed, only the service container stays.
assert_eq!(direct_macs(&pins, &[]), vec![mac(3)]);
let script = pin_elements(&pins, &listed);
assert!(script.contains("add element inet hive_egress direct_mac { 02:00:00:00:00:02 }"));
assert!(script.contains("add element inet hive_egress direct_mac { 02:00:00:00:00:03 }"));
assert!(!script.contains("direct_mac { 02:00:00:00:00:01 }"));
assert!(!script.contains("direct_mac { 02:00:00:00:00:04 }"));
}
const NOW: u64 = 1_000;
fn lease_line(expiry: u64, last: u8, ip: &str) -> String {
format!("{expiry} {} {ip} * *\n", mac(last))
}
#[test]
fn an_address_belongs_to_the_agent_of_the_leased_mac() {
let pins = [
pin("vb-h-atlas", 10, 1, "h-atlas"),
pin("vb-h-uma", 11, 2, "h-uma"),
];
let leases = lease_line(NOW + 60, 1, "10.42.0.5") + &lease_line(NOW + 60, 2, "10.42.0.6");
assert_eq!(
agent_of("10.42.0.5", &leases, &pins, NOW).as_deref(),
Some("atlas")
);
assert_eq!(
agent_of("10.42.0.6", &leases, &pins, NOW).as_deref(),
Some("uma")
);
// A lease with no end counts.
let forever = lease_line(0, 1, "10.42.0.7");
assert_eq!(
agent_of("10.42.0.7", &forever, &pins, NOW).as_deref(),
Some("atlas")
);
}
#[test]
fn a_relay_pool_address_names_the_agent_it_was_derived_from() {
let pins = [
pin("vb-h-atlas", 10, 1, "h-atlas"),
pin("vb-h-uma", 11, 2, "h-uma"),
];
let leases = lease_line(NOW + 60, 1, "10.42.0.5") + &lease_line(NOW + 60, 2, "10.42.0.6");
assert_eq!(
agent_of("198.18.0.5", &leases, &pins, NOW).as_deref(),
Some("atlas")
);
assert_eq!(
agent_of("198.18.0.6", &leases, &pins, NOW).as_deref(),
Some("uma")
);
// No lease behind the pool address, and an address of the pool that
// is itself leased: neither names anyone.
assert_eq!(agent_of("198.18.0.7", &leases, &pins, NOW), None);
let leased_pool = lease_line(NOW + 60, 1, "198.18.0.9");
assert_eq!(agent_of("198.18.0.9", &leased_pool, &pins, NOW), None);
// Two leases that share the pool address are ambiguous.
let twins = lease_line(NOW + 60, 1, "10.42.0.5") + &lease_line(NOW + 60, 2, "10.44.0.5");
assert_eq!(agent_of("198.18.0.5", &twins, &pins, NOW), None);
}
#[test]
fn an_address_nobody_can_be_told_to_hold_has_no_agent() {
let pins = [
pin("vb-h-atlas", 10, 1, "h-atlas"),
pin("vb-matrix", 11, 2, "hive-matrix"),
pin("vb-h-x", 12, 3, "h-x y"),
];
let leases = lease_line(NOW + 60, 1, "10.42.0.5");
// Control for every refusal below: the same lease and pins name an agent.
assert!(agent_of("10.42.0.5", &leases, &pins, NOW).is_some());
// No lease for the address, an expired one, and text that is no address.
assert_eq!(agent_of("10.42.0.9", &leases, &pins, NOW), None);
assert_eq!(
agent_of("10.42.0.5", &lease_line(NOW, 1, "10.42.0.5"), &pins, NOW),
None
);
assert_eq!(
agent_of(
"10.42.0.5",
&lease_line(NOW - 1, 1, "10.42.0.5"),
&pins,
NOW
),
None
);
for bad in ["", "-", "10.42.0.5.1", "fe80::1", "10.42.0.5/32"] {
assert_eq!(agent_of(bad, &leases, &pins, NOW), None, "{bad:?}");
}
// A leased MAC no port is bound to.
assert_eq!(
agent_of(
"10.42.0.8",
&lease_line(NOW + 60, 7, "10.42.0.8"),
&pins,
NOW
),
None
);
// A service container is no agent, and a machine name that is no identifier names none.
assert_eq!(
agent_of(
"10.42.0.8",
&lease_line(NOW + 60, 2, "10.42.0.8"),
&pins,
NOW
),
None
);
assert_eq!(
agent_of(
"10.42.0.8",
&lease_line(NOW + 60, 3, "10.42.0.8"),
&pins,
NOW
),
None
);
}
#[test]
fn two_macs_leased_the_same_address_name_no_agent() {
let pins = [
pin("vb-h-atlas", 10, 1, "h-atlas"),
pin("vb-h-uma", 11, 2, "h-uma"),
];
let leases = lease_line(NOW + 60, 1, "10.42.0.5") + &lease_line(NOW + 60, 2, "10.42.0.5");
assert_eq!(agent_of("10.42.0.5", &leases, &pins, NOW), None);
// Control: an expired second lease is no rival.
let leases = lease_line(NOW + 60, 1, "10.42.0.5") + &lease_line(NOW - 5, 2, "10.42.0.5");
assert_eq!(
agent_of("10.42.0.5", &leases, &pins, NOW).as_deref(),
Some("atlas")
);
}
#[test]
fn a_mac_bound_to_two_machines_names_no_agent() {
let pins = [
pin("vb-h-atlas", 10, 1, "h-atlas"),
pin("vb-h-uma", 11, 1, "h-uma"),
];
let leases = lease_line(NOW + 60, 1, "10.42.0.5");
assert_eq!(agent_of("10.42.0.5", &leases, &pins, NOW), None);
}
#[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 390s }\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 a_renewal_moves_the_timeout_to_the_new_remaining_time_plus_the_margin() {
let mac = "02:00:00:00:00:0a";
let first = lease_script("add", mac, "10.42.0.10", 120).unwrap();
let renewed = lease_script("old", mac, "10.42.0.10", 120).unwrap();
let later = lease_script("old", mac, "10.42.0.10", 3600).unwrap();
assert!(first.ends_with("10.42.0.10 timeout 420s }\n"));
assert_eq!(renewed, first, "old and add load the same pair");
assert!(later.ends_with("10.42.0.10 timeout 3900s }\n"));
assert!(
later.starts_with("add element bridge hive_egress lease_v4 { 02:00:00:00:00:0a . 10.42.0.10 }\ndelete element"),
"the old timeout is deleted first, so the new one replaces it"
);
}
#[test]
fn a_remaining_time_that_overflows_with_the_margin_does_not_wrap() {
let script = lease_script("old", "02:00:00:00:00:0a", "10.42.0.10", u64::MAX).unwrap();
assert!(script.contains(&format!("timeout {}s", u64::MAX)));
}
#[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 1300s"));
assert!(!script.contains("10.42.0.11"));
assert!(script.contains("10.42.0.12 timeout 3900s"));
}
#[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);
}
}