[PATCH proxmox-ebpf v2 06/27] agent: add userspace coordinator and stateless policy subsystem

[Hannes Laimer <h.laimer@proxmox.com>]

The agent applies the SDN running config to the kernel: BPF programs on
guest NICs stamp each packet with its source identity and enforce the
(src, dst) policy on delivery.

It runs one-shot per event (boot, SDN apply, guest NIC plug) instead of
as a resident daemon: those triggers cover every situation where kernel
state has to change, and programs and maps are pinned in bpffs, so
enforcement persists between invocations. Every run first checks that
the pinned programs match the built binary. A pure code change swaps
the programs atomically with no traffic gap. An incompatible change to
the map layout tears down and rebuilds the state, the only case with a
short enforcement gap.

Enforcement is stateless default-deny on the receiving side: a packet
is dropped unless an allow cell exists for its (source, destination)
identity pair. A guest NIC that should be enforced but cannot be fails
the plug rather than coming up open.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 include/mark.h           |  30 +++
 src/agent.rs             |  69 +++++++
 src/main.rs              |  68 +++++++
 src/policy/bpf/tap.bpf.c |  68 +++++++
 src/policy/bpf/types.h   |  23 +++
 src/policy/mod.rs        | 268 +++++++++++++++++++++++++++
 src/policy/types.rs      |  45 +++++
 src/running_config.rs    |  38 ++++
 src/state.rs             | 132 ++++++++++++++
 src/subsystem.rs         | 385 +++++++++++++++++++++++++++++++++++++++
 src/tc.rs                | 152 ++++++++++++++++
 11 files changed, 1278 insertions(+)
 create mode 100644 include/mark.h
 create mode 100644 src/agent.rs
 create mode 100644 src/main.rs
 create mode 100644 src/policy/bpf/tap.bpf.c
 create mode 100644 src/policy/bpf/types.h
 create mode 100644 src/policy/mod.rs
 create mode 100644 src/policy/types.rs
 create mode 100644 src/running_config.rs
 create mode 100644 src/state.rs
 create mode 100644 src/subsystem.rs
 create mode 100644 src/tc.rs

diff --git a/include/mark.h b/include/mark.h
new file mode 100644
index 0000000..e3c0577
--- /dev/null
+++ b/include/mark.h
@@ -0,0 +1,30 @@
+#ifndef PROXMOX_EBPF_MARK_H
+#define PROXMOX_EBPF_MARK_H
+
+// Include after vmlinux.h and <bpf/bpf_helpers.h>, which provide struct
+// __sk_buff, the __uNN types, __always_inline, and barrier_var().
+//
+// skb->mark is shared with the rest of the host (firewall fwmark, routing,
+// ...). Microseg owns only the low 16 bits, which hold the group id and are
+// also what travels on the wire. The helpers below touch only that range, so
+// any bits other subsystems set are preserved.
+#define MICROSEG_MARK_MASK 0xffffu
+
+// Replace microseg's bits in skb->mark with the group, leaving the rest untouched.
+static __always_inline void microseg_mark_set(struct __sk_buff *skb, __u16 group) {
+    __u32 mark = skb->mark;
+    mark = (mark & ~MICROSEG_MARK_MASK) | ((__u32)group & MICROSEG_MARK_MASK);
+    // skb->mark is BPF context, which the verifier only allows at its full
+    // 32-bit width. Without the barrier clang narrows the write to a 16-bit
+    // store of the low half, a valid little-endian read-modify-write that the
+    // verifier then rejects as an invalid context access.
+    barrier_var(mark);
+    skb->mark = mark;
+}
+
+// The group currently stamped in skb->mark.
+static __always_inline __u16 microseg_mark_get(struct __sk_buff *skb) {
+    return (__u16)(skb->mark & MICROSEG_MARK_MASK);
+}
+
+#endif
diff --git a/src/agent.rs b/src/agent.rs
new file mode 100644
index 0000000..63f5f14
--- /dev/null
+++ b/src/agent.rs
@@ -0,0 +1,69 @@
+//! Agent orchestrator. Builds the per-host [`DesiredState`] from the SDN running-config and applies
+//! it via the [`policy`](crate::policy) subsystem. The tap_plug path
+//! ([`apply_guest_iface_policy`](Agent::apply_guest_iface_policy)) programs a single interface.
+//!
+//! The binary runs one-shot per event (boot, an SDN apply, a tap_plug), not as a resident daemon.
+//! Every invocation first makes sure the programs are loaded (installing them on the first run or a
+//! version change), then applies. Maps and links are pinned in bpffs, so they persist across
+//! invocations and re-running never interrupts traffic.
+
+use crate::{policy::PolicySubsystem, running_config, state::DesiredState};
+
+pub struct Agent {
+    policy: PolicySubsystem,
+}
+
+impl Agent {
+    pub fn new() -> Self {
+        Self {
+            policy: PolicySubsystem::new(),
+        }
+    }
+
+    /// Full pass over the host.
+    pub fn apply(&mut self) -> anyhow::Result<()> {
+        let Some(state) = build_state()? else {
+            return Ok(());
+        };
+        log::debug!(
+            "applying: {} rule cells, {} assignments",
+            state.rules.len(),
+            state.assignments.len(),
+        );
+        self.policy.apply(&state)
+    }
+
+    /// Fast path for a guest NIC that just appeared (a tap_plug).
+    ///
+    /// Forwards only a genuine enforcement failure for an assigned NIC, so the agent exits non-zero
+    /// and the caller does not bring the NIC up unenforced. A running config we cannot read or
+    /// build is logged and treated as success, so one bad config does not block every guest start.
+    pub fn apply_guest_iface_policy(&mut self, iface: &str) -> anyhow::Result<()> {
+        let state = match build_state() {
+            Ok(Some(state)) => state,
+            Ok(None) => return Ok(()),
+            Err(e) => {
+                log::error!("apply {iface}: load running config: {e:#}");
+                return Ok(());
+            }
+        };
+        self.policy.apply_guest_iface(&state, iface)
+    }
+
+    /// Detach everything and drop all pinned/run state, for package removal.
+    pub fn clear(&self) -> anyhow::Result<()> {
+        self.policy.clear()
+    }
+}
+
+fn build_state() -> anyhow::Result<Option<DesiredState>> {
+    let microseg = running_config::load_microseg()?;
+    match DesiredState::build(&microseg) {
+        Ok(s) => Ok(Some(s)),
+        Err(e) => {
+            log::error!("desired state: {e:#}");
+            Ok(None)
+        }
+    }
+}
+
diff --git a/src/main.rs b/src/main.rs
new file mode 100644
index 0000000..6b3c16c
--- /dev/null
+++ b/src/main.rs
@@ -0,0 +1,68 @@
+mod agent;
+mod policy;
+mod running_config;
+mod state;
+mod subsystem;
+mod tc;
+
+use agent::Agent;
+use anyhow::{anyhow, bail};
+use proxmox_log::{LevelFilter, Logger};
+
+const HELP: &str = "\
+Usage: proxmox-ebpf <command> [<interface>]
+
+Commands:
+  apply [<iface>]   apply the SDN running-config to BPF state. With <iface>,
+                    only that interface.
+  clear             detach all programs and drop pinned state (used on removal)
+  help              show this help
+";
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Command {
+    Apply,
+    Clear,
+    Help,
+}
+
+impl std::str::FromStr for Command {
+    type Err = anyhow::Error;
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        Ok(match s {
+            "apply" => Command::Apply,
+            "clear" => Command::Clear,
+            "help" => Command::Help,
+            other => bail!("{other:?} is not a valid command"),
+        })
+    }
+}
+
+fn init_logger() -> anyhow::Result<()> {
+    Logger::from_env("PROXMOX_EBPF_LOG", LevelFilter::INFO)
+        .stderr_pve()
+        .init()?;
+    Ok(())
+}
+
+fn main() -> anyhow::Result<()> {
+    let mut args = pico_args::Arguments::from_env();
+    let cmd: Command = args
+        .subcommand()?
+        .ok_or_else(|| anyhow!("no command specified\n\n{HELP}"))?
+        .parse()?;
+
+    init_logger()?;
+
+    match cmd {
+        Command::Help => {
+            println!("{HELP}");
+            Ok(())
+        }
+        Command::Apply => match args.opt_free_from_str::<String>()? {
+            None => Agent::new().apply(),
+            Some(iface) => Agent::new().apply_guest_iface_policy(&iface),
+        },
+        Command::Clear => Agent::new().clear(),
+    }
+}
diff --git a/src/policy/bpf/tap.bpf.c b/src/policy/bpf/tap.bpf.c
new file mode 100644
index 0000000..305144f
--- /dev/null
+++ b/src/policy/bpf/tap.bpf.c
@@ -0,0 +1,68 @@
+#include "vmlinux.h"
+#include <bpf/bpf_helpers.h>
+#include "mark.h"
+#include "types.h"
+#include "bpf_debug.h"
+
+char LICENSE[] SEC("license") = "GPL";
+
+#define TC_ACT_OK 0
+#define TC_ACT_SHOT 2
+
+struct {
+    __uint(type, BPF_MAP_TYPE_HASH);
+    __type(key, __u32); // ifindex
+    __type(value, struct guest_group);
+    __uint(max_entries, 65560); // max # of tap interfaces
+    // allocate entries on demand, so memory tracks live taps rather than the ceiling
+    __uint(map_flags, BPF_F_NO_PREALLOC);
+    __uint(pinning, LIBBPF_PIN_BY_NAME);
+} tap_to_group SEC(".maps");
+
+struct {
+    __uint(type, BPF_MAP_TYPE_HASH);
+    __type(key, struct rule_key);
+    __type(value, struct rule_value);
+    __uint(max_entries, 1048576);
+    // allocate entries on demand, so memory tracks live rules rather than the ceiling
+    __uint(map_flags, BPF_F_NO_PREALLOC);
+    __uint(pinning, LIBBPF_PIN_BY_NAME);
+} rules SEC(".maps");
+
+// Stamp this tap's group into skb->mark. The destination tap's egress reads it as src_group.
+SEC("classifier")
+int tc_tap_ingress(struct __sk_buff *skb) {
+    __u32 ifidx = skb->ifindex;
+    struct guest_group *g = bpf_map_lookup_elem(&tap_to_group, &ifidx);
+    if (!g) return TC_ACT_OK;
+    DBG("mark: src_grp=%u ifidx=%u", g->group, ifidx);
+    microseg_mark_set(skb, g->group);
+    return TC_ACT_OK;
+}
+
+
+// Enforce on egress of the destination tap, where both ends are known: dst is
+// this tap's group, src is skb->mark. Look up (src_group, dst_group) in `rules`.
+// A missing entry is default-deny, and unstamped packets (src_group=0) need an
+// explicit (0, dst_group) allow.
+SEC("classifier")
+int tc_tap_egress(struct __sk_buff *skb) {
+    __u32 ifidx = skb->ifindex;
+    __u16 src_group = microseg_mark_get(skb);
+
+    struct guest_group *g = bpf_map_lookup_elem(&tap_to_group, &ifidx);
+    if (!g) return TC_ACT_OK;
+
+    struct rule_key k = { .src_group = src_group, .dst_group = g->group };
+    struct rule_value *r = bpf_map_lookup_elem(&rules, &k);
+    if (!r) {
+        DBG("deny (no rule): src_grp=%u dst_grp=%u", src_group, g->group);
+        return TC_ACT_SHOT;
+    }
+    if (!r->allow) {
+        DBG("deny (explicit): src_grp=%u dst_grp=%u", src_group, g->group);
+        return TC_ACT_SHOT;
+    }
+    return TC_ACT_OK;
+}
+
diff --git a/src/policy/bpf/types.h b/src/policy/bpf/types.h
new file mode 100644
index 0000000..67a86a9
--- /dev/null
+++ b/src/policy/bpf/types.h
@@ -0,0 +1,23 @@
+#ifndef PROXMOX_EBPF_POLICY_TYPES_H
+#define PROXMOX_EBPF_POLICY_TYPES_H
+
+// Keep in sync with ../types.rs.
+// __u8/__u16/__u32 are provided by vmlinux.h
+
+struct guest_group {
+    __u16 group;
+    __u16 _pad;
+};
+
+struct rule_key {
+    __u16 src_group;
+    __u16 dst_group;
+};
+
+struct rule_value {
+    __u8 allow;
+    __u8 _pad[3];
+};
+
+#endif
+
diff --git a/src/policy/mod.rs b/src/policy/mod.rs
new file mode 100644
index 0000000..ff0c0ad
--- /dev/null
+++ b/src/policy/mod.rs
@@ -0,0 +1,268 @@
+//! Tap-side enforcement. Attaches BPF programs to per-VM tap/veth interfaces that (a) stamp
+//! `skb->mark` with the source group on ingress, and (b) drop packets on egress if no `(src_group,
+//! dst_group) -> allow` rule exists for the pair. Driven by the [`DesiredState`] derived from
+//! `/etc/pve/sdn/.running-config`, resolved per host via `veth{vmid}i{iface}` /
+//! `tap{vmid}i{iface}`.
+//!
+//! Two maps hold the state, `tap_to_group` (ifindex -> group) and `rules` ((src, dst) -> allow).
+
+mod types;
+
+use std::collections::{HashMap, HashSet};
+
+use anyhow::Context;
+use aya::include_bytes_aligned;
+
+use self::types::*;
+use crate::state::{DesiredState, ResolvedAssignment, ResolvedRule};
+use crate::subsystem::TcPrograms;
+use crate::tc::Direction;
+
+const NAME: &str = "policy";
+
+const POLICY_OBJ: &[u8] = include_bytes_aligned!(concat!(env!("OUT_DIR"), "/tap.bpf.o"));
+const POLICY_FINGERPRINT: u64 = TcPrograms::obj_fingerprint(POLICY_OBJ);
+
+// BUMP THIS when a semantically-incompatible change to a map definition in tap.bpf.c is made
+const SCHEMA_VERSION: u32 = 1;
+
+fn program_name(dir: Direction) -> &'static str {
+    match dir {
+        Direction::Ingress => "tc_tap_ingress",
+        Direction::Egress => "tc_tap_egress",
+    }
+}
+
+pub struct PolicySubsystem {
+    programs: TcPrograms,
+}
+
+impl PolicySubsystem {
+    pub fn new() -> Self {
+        Self {
+            programs: TcPrograms::new(
+                NAME,
+                POLICY_OBJ,
+                POLICY_FINGERPRINT,
+                program_name,
+                Some(SCHEMA_VERSION),
+            ),
+        }
+    }
+
+    /// Full pass over the host (boot, after an SDN apply). Holds the apply lock exclusively, so its
+    /// enumerate/detach/attach cannot interleave with a concurrent single-interface apply.
+    pub fn apply(&mut self, state: &DesiredState) -> anyhow::Result<()> {
+        let _lock = self.programs.lock_exclusive()?;
+        self.programs.ensure_loaded()?;
+        self.apply_full(state)
+    }
+
+    /// Detach all tap programs and drop pinned/run state, for package removal.
+    pub fn clear(&self) -> anyhow::Result<()> {
+        self.programs.clear()
+    }
+
+    /// Program a single guest NIC that just appeared (a tap_plug).
+    ///
+    /// An unassigned NIC has nothing to enforce and returns `Ok`. For an assigned NIC a failure to
+    /// install enforcement propagates, so the agent exits non-zero and the plug does not bring the
+    /// NIC up unenforced.
+    pub fn apply_guest_iface(&mut self, state: &DesiredState, iface: &str) -> anyhow::Result<()> {
+        let Some((vmid, idx)) = parse_tap_name(iface) else {
+            log::warn!("apply_guest_iface: '{iface}' is not a guest NIC name, skipping");
+            return Ok(());
+        };
+
+        let Some(id) = state
+            .assignments
+            .iter()
+            .find(|a| a.vmid == vmid && a.iface == idx)
+            .map(|a| a.id)
+        else {
+            log::debug!(
+                "apply_guest_iface: {iface} (vmid={vmid} iface={idx}) is unassigned, nothing to attach"
+            );
+            return Ok(());
+        };
+
+        // If the programs are already loaded, take the lock shared so concurrent guest plugs don't
+        // serialize. A needed (re)install rebuilds shared state, so take it exclusively and do a
+        // full apply, which covers this NIC too.
+        if self.programs.is_current() {
+            let _lock = self.programs.lock_shared()?;
+            self.apply_one(iface, id)
+        } else {
+            let _lock = self.programs.lock_exclusive()?;
+            self.programs.ensure_loaded()?;
+            self.apply_full(state)
+        }
+        .with_context(|| format!("enforcement for assigned NIC {iface}"))
+    }
+
+    fn apply_full(&mut self, state: &DesiredState) -> anyhow::Result<()> {
+        log::trace!("policy full apply");
+        let assignments = resolve_local_assignments(&state.assignments)?;
+        self.sync_tap_to_group(&assignments)?;
+        let desired: HashSet<u32> = assignments.keys().copied().collect();
+        self.programs.reconcile(&desired)?;
+        self.sync_rules(&state.rules)?;
+        Ok(())
+    }
+
+    /// Additive single-interface programming. Sets this interface's `tap_to_group` entry and
+    /// attaches its links, propagating failure so the plug does not bring the NIC up unenforced.
+    /// The global `rules` map and the other interfaces are not touched.
+    fn apply_one(&mut self, iface: &str, id: u16) -> anyhow::Result<()> {
+        let Ok(ifindex) = nix::net::if_::if_nametoindex(iface) else {
+            log::info!("apply_guest_iface: {iface} is gone, nothing to do");
+            return Ok(());
+        };
+
+        {
+            let mut tap_to_group = self.programs.hash_map::<u32, GuestGroup>("tap_to_group")?;
+            tap_to_group.insert(ifindex, GuestGroup { group: id, _pad: 0 }, 0)?;
+        }
+        log::info!("apply_guest_iface: {iface} (ifindex {ifindex}) -> id {id}");
+
+        self.programs.attach_iface(ifindex)?;
+        Ok(())
+    }
+
+    fn sync_tap_to_group(&mut self, desired: &HashMap<u32, u16>) -> anyhow::Result<()> {
+        let mut tap_to_group = self.programs.hash_map::<u32, GuestGroup>("tap_to_group")?;
+        let live: HashMap<u32, u16> = tap_to_group
+            .iter()
+            .filter_map(|r| r.ok().map(|(k, v)| (k, v.group)))
+            .collect();
+        let mut written = 0usize;
+        for (&ifidx, &id) in desired {
+            if live.get(&ifidx) != Some(&id) {
+                tap_to_group.insert(ifidx, GuestGroup { group: id, _pad: 0 }, 0)?;
+                written += 1;
+            }
+        }
+        let mut removed = 0usize;
+        for &ifidx in live.keys().filter(|i| !desired.contains_key(i)) {
+            log::debug!("drop: group mapping for {ifidx}");
+            let _ = tap_to_group.remove(&ifidx);
+            removed += 1;
+        }
+        if written + removed > 0 {
+            log::info!("tap_to_group: {written} written, {removed} removed");
+        }
+        Ok(())
+    }
+
+    fn sync_rules(&mut self, rules: &[ResolvedRule]) -> anyhow::Result<()> {
+        let mut rules_map = self.programs.hash_map::<RuleKey, RuleValue>("rules")?;
+        let live: HashMap<RuleKey, u8> = rules_map
+            .iter()
+            .filter_map(|r| r.ok().map(|(k, v)| (k, v.allow)))
+            .collect();
+        let mut desired_keys = HashSet::new();
+        let mut written = 0usize;
+        for r in rules {
+            let k = RuleKey {
+                src_group: r.src_id,
+                dst_group: r.dst_id,
+            };
+            desired_keys.insert(k);
+            let want_allow = r.allow as u8;
+            if live.get(&k) != Some(&want_allow) {
+                rules_map.insert(
+                    k,
+                    RuleValue {
+                        allow: want_allow,
+                        _pad: [0; 3],
+                    },
+                    0,
+                )?;
+                written += 1;
+                log::trace!(
+                    "rule: write src={} dst={} allow={}",
+                    k.src_group,
+                    k.dst_group,
+                    r.allow
+                );
+            }
+        }
+        let stale: Vec<_> = live
+            .keys()
+            .filter(|k| !desired_keys.contains(k))
+            .copied()
+            .collect();
+        for k in &stale {
+            let _ = rules_map.remove(k);
+        }
+        if written + stale.len() > 0 {
+            log::info!("rules: {written} written, {} removed", stale.len());
+        } else {
+            log::debug!("rules: {} entries, no changes", desired_keys.len());
+        }
+        Ok(())
+    }
+}
+
+/// Parse a guest NIC name (`tap<vmid>i<idx>` or `veth<vmid>i<idx>`) into its (vmid, iface) pair.
+fn parse_tap_name(name: &str) -> Option<(u32, u32)> {
+    let rest = name
+        .strip_prefix("tap")
+        .or_else(|| name.strip_prefix("veth"))?;
+    let (vmid, idx) = rest.split_once('i')?;
+    Some((vmid.parse().ok()?, idx.parse().ok()?))
+}
+
+fn resolve_local_assignments(
+    assignments: &[ResolvedAssignment],
+) -> anyhow::Result<HashMap<u32, u16>> {
+    // A failure here must propagate, not fall back to an empty map: a full apply treats the result
+    // as the complete desired set, so an empty one would tear down every tap's enforcement. Failing
+    // the apply keeps the last-good state in place instead.
+    let ifaces = nix::net::if_::if_nameindex().context("if_nameindex")?;
+    let by_name: HashMap<String, u32> = ifaces
+        .iter()
+        .filter_map(|i| i.name().to_str().ok().map(|n| (n.to_owned(), i.index())))
+        .collect();
+
+    let mut out = HashMap::new();
+    for a in assignments {
+        let veth = format!("veth{}i{}", a.vmid, a.iface);
+        let tap = format!("tap{}i{}", a.vmid, a.iface);
+        if let Some(&ifidx) = by_name.get(&veth).or_else(|| by_name.get(&tap)) {
+            log::debug!(
+                "resolved vmid={} iface={} -> ifindex {ifidx} id {}",
+                a.vmid,
+                a.iface,
+                a.id,
+            );
+            out.insert(ifidx, a.id);
+        } else {
+            log::trace!("skipping vmid={} iface={}: no local tap", a.vmid, a.iface);
+        }
+    }
+    log::debug!("{} managed taps on this host", out.len());
+    Ok(out)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_tap_name_handles_qemu_and_lxc() {
+        assert_eq!(parse_tap_name("tap101i0"), Some((101, 0)));
+        assert_eq!(parse_tap_name("veth200i3"), Some((200, 3)));
+        assert_eq!(parse_tap_name("tap1000i12"), Some((1000, 12)));
+    }
+
+    #[test]
+    fn parse_tap_name_rejects_other_interfaces() {
+        assert_eq!(parse_tap_name("vmbr0"), None);
+        assert_eq!(parse_tap_name("fwln101i0"), None);
+        assert_eq!(parse_tap_name("veth200i-3"), None);
+        assert_eq!(parse_tap_name("eth0"), None);
+        assert_eq!(parse_tap_name("tap101"), None);
+        assert_eq!(parse_tap_name("tapfooibar"), None);
+    }
+}
diff --git a/src/policy/types.rs b/src/policy/types.rs
new file mode 100644
index 0000000..174aecc
--- /dev/null
+++ b/src/policy/types.rs
@@ -0,0 +1,45 @@
+//! Keep in sync with `bpf/types.h`.
+
+#[repr(C)]
+#[derive(Copy, Clone)]
+pub struct GuestGroup {
+    pub group: u16,
+    pub _pad: u16,
+}
+
+#[repr(C)]
+#[derive(Copy, Clone, Hash, PartialEq, Eq)]
+pub struct RuleKey {
+    pub src_group: u16,
+    pub dst_group: u16,
+}
+
+#[repr(C)]
+#[derive(Copy, Clone)]
+pub struct RuleValue {
+    pub allow: u8,
+    pub _pad: [u8; 3],
+}
+
+unsafe impl aya::Pod for GuestGroup {}
+unsafe impl aya::Pod for RuleKey {}
+unsafe impl aya::Pod for RuleValue {}
+
+#[cfg(test)]
+mod layout {
+    use super::*;
+    use core::mem::{offset_of, size_of};
+
+    #[test]
+    fn sizes() {
+        assert_eq!(size_of::<GuestGroup>(), 4);
+        assert_eq!(size_of::<RuleKey>(), 4);
+        assert_eq!(size_of::<RuleValue>(), 4);
+    }
+
+    #[test]
+    fn offsets() {
+        assert_eq!(offset_of!(RuleKey, dst_group), 2);
+        assert_eq!(offset_of!(GuestGroup, _pad), 2);
+    }
+}
diff --git a/src/running_config.rs b/src/running_config.rs
new file mode 100644
index 0000000..fe48802
--- /dev/null
+++ b/src/running_config.rs
@@ -0,0 +1,38 @@
+//! Loader for `/etc/pve/sdn/.running-config`, the JSON snapshot perl writes on every SDN commit.
+//! The deserialize types live in [`proxmox_ve_config::sdn`]. This module is just the agent-side
+//! I/O wrapper around them.
+
+use std::path::Path;
+
+use anyhow::Context;
+use proxmox_ve_config::sdn::config::RunningConfig;
+use proxmox_ve_config::sdn::microseg::MicrosegRunningConfig;
+
+pub const PATH: &str = "/etc/pve/sdn/.running-config";
+
+/// Read and parse the full SDN running config.
+///
+/// Returns `Ok(None)` if the file does not exist yet. This is the legitimate state of a node where
+/// SDN has never been committed. Any other I/O or parse error propagates.
+pub fn load() -> anyhow::Result<Option<RunningConfig>> {
+    load_from(Path::new(PATH))
+}
+
+/// Read and project just the microseg block, returning `default()` if the file or the `microseg`
+/// key is absent. Most agent code only cares about microseg, so this is the convenient entry
+/// point.
+pub fn load_microseg() -> anyhow::Result<MicrosegRunningConfig> {
+    Ok(load()?
+        .and_then(|c| c.microseg().cloned())
+        .unwrap_or_default())
+}
+
+fn load_from(path: &Path) -> anyhow::Result<Option<RunningConfig>> {
+    let raw = match std::fs::read_to_string(path) {
+        Ok(s) => s,
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+        Err(e) => return Err(e).with_context(|| format!("read {}", path.display())),
+    };
+    let cfg = serde_json::from_str(&raw).with_context(|| format!("parse {}", path.display()))?;
+    Ok(Some(cfg))
+}
diff --git a/src/state.rs b/src/state.rs
new file mode 100644
index 0000000..fc67818
--- /dev/null
+++ b/src/state.rs
@@ -0,0 +1,132 @@
+//! Resolved desired state derived from the SDN running config.
+//!
+//! The running config carries the compiled identity registry (id to representative set) and the
+//! admin's groups, rules and assignments. This module turns that into what the data plane needs:
+//! the wire id for each managed interface and the folded `(src_id, dst_id) -> allow` rule cells.
+//! The hard, must-be-cluster-consistent work (allocation)
+//! already happened in the render. Here the agent only resolves and folds, reading nothing but the
+//! replicated config.
+
+use anyhow::Context;
+
+use proxmox_ve_config::sdn::microseg::{self, MicrosegRunningConfig};
+
+#[derive(Debug, Default)]
+pub struct DesiredState {
+    /// The permitted cells of the rule map. Absence of a cell is a deny (the data-plane default), so
+    /// only allow cells are carried.
+    pub rules: Vec<ResolvedRule>,
+    /// The wire id each managed guest NIC should stamp.
+    pub assignments: Vec<ResolvedAssignment>,
+}
+
+#[derive(Debug)]
+pub struct ResolvedRule {
+    pub src_id: u16,
+    pub dst_id: u16,
+    pub allow: bool,
+}
+
+#[derive(Debug)]
+pub struct ResolvedAssignment {
+    pub vmid: u32,
+    pub iface: u32,
+    /// The interface's wire id (the identity its group set resolves to).
+    pub id: u16,
+}
+
+impl DesiredState {
+    pub fn build(cfg: &MicrosegRunningConfig) -> anyhow::Result<Self> {
+        let entries = cfg.ids();
+        let registry = cfg.identities();
+        let policies = microseg::build_policies(entries).context("build policies")?;
+        let rules = microseg::resolved_rules(entries).context("resolve rules")?;
+
+        // Resolve each realized assignment (static plus selector-expanded) to its wire id.
+        let mut assignments = Vec::new();
+        for assignment in cfg.realized() {
+            let set = microseg::group_set(entries, &assignment.groups).with_context(|| {
+                format!("assignment vm{} net{}", assignment.vmid, assignment.iface)
+            })?;
+            let wire_id = registry
+                .id_of(&set, &policies)
+                .unwrap_or(microseg::identity::UNTAGGED_ID);
+            assignments.push(ResolvedAssignment {
+                vmid: assignment.vmid,
+                iface: assignment.iface,
+                id: wire_id,
+            });
+        }
+
+        // Fold the rule map over the allocated classes. The destination is always a real, enforced
+        // class. The source side additionally includes the untagged id (id 0, the no-group
+        // identity) so an `untagged -> X` rule admits unstamped traffic: gateway / DHCP / ARP
+        // replies, and cross-node frames that arrive without a GBP tag. An untagged *destination*
+        // would be a no-op (no program enforces there), so it is not iterated, we only emit allow
+        // cells.
+        let classes = registry.classes();
+        // its representative is the built-in untagged mark, so a predicate naming the 'untagged'
+        // group fires on the untagged id.
+        let untagged_rep: microseg::identity::GroupSet =
+            std::iter::once(microseg::UNTAGGED_MARK).collect();
+        let sources = classes
+            .iter()
+            .map(|(&id, rep)| (id, rep))
+            .chain(std::iter::once((
+                microseg::identity::UNTAGGED_ID,
+                &untagged_rep,
+            )));
+        let mut allow_cells = Vec::new();
+        for (src_id, src_rep) in sources {
+            for (&dst_id, dst_rep) in classes {
+                if microseg::verdict(&rules, src_rep, dst_rep) {
+                    allow_cells.push(ResolvedRule {
+                        src_id,
+                        dst_id,
+                        allow: true,
+                    });
+                }
+            }
+        }
+
+        Ok(Self {
+            rules: allow_cells,
+            assignments,
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn build_resolves_ids_and_folds_rules() {
+        // web=2 -> class 1, db=3 -> class 2, one rule web -> db allow.
+        let json = r#"{
+            "ids": {
+                "web": {"type":"group","id":"web","mark":2},
+                "db": {"type":"group","id":"db","mark":3},
+                "r1": {"type":"rule","id":"r1","src":["web"],"dst":["db"],"allow":1},
+                "vm100i0": {"type":"guestassignment","id":"vm100i0","vmid":100,"iface":0,"groups":["web"]}
+            },
+            "identities": {"next":3,"classes":{"1":[2],"2":[3]}},
+            "realized": [{"vmid":100,"iface":0,"groups":["web"]}]
+        }"#;
+        let cfg: MicrosegRunningConfig = serde_json::from_str(json).expect("parse running config");
+        let state = DesiredState::build(&cfg).expect("build state");
+
+        // the web interface resolves to its class id 1
+        let assignment = state
+            .assignments
+            .iter()
+            .find(|a| a.vmid == 100 && a.iface == 0)
+            .expect("assignment present");
+        assert_eq!(assignment.id, 1);
+
+        // the only permitted cell is web(1) -> db(2)
+        assert_eq!(state.rules.len(), 1);
+        let rule = &state.rules[0];
+        assert_eq!((rule.src_id, rule.dst_id, rule.allow), (1, 2, true));
+    }
+}
diff --git a/src/subsystem.rs b/src/subsystem.rs
new file mode 100644
index 0000000..0d7b62c
--- /dev/null
+++ b/src/subsystem.rs
@@ -0,0 +1,385 @@
+//! The shared subsystem mechanism. An ingress and egress tc program with their maps and links,
+//! pinned under `/sys/fs/bpf/proxmox-ebpf/<name>/`. The loaded BPF stays in the kernel between
+//! invocations, so [`TcPrograms::ensure_loaded`] loads and verifies only on the first run and on a
+//! version change. Everything else attaches links and syncs maps against what is already there.
+//! The [`policy`](crate::policy) and [`bridge`](crate::bridge) subsystems each own a
+//! [`TcPrograms`].
+
+use std::{collections::HashSet, fs::File, io::ErrorKind, path::PathBuf};
+
+use anyhow::Context;
+use aya::{EbpfLoader, programs::SchedClassifier};
+use nix::fcntl::{Flock, FlockArg};
+
+use crate::tc::{self, DIRECTIONS, Direction};
+
+pub const VERIFY_ROOT: &str = "/sys/fs/bpf/proxmox-ebpf-test";
+
+const PIN_ROOT: &str = "/sys/fs/bpf/proxmox-ebpf";
+const RUN_ROOT: &str = "/run/proxmox-ebpf";
+
+fn pin_root_for(name: &str) -> PathBuf {
+    PathBuf::from(PIN_ROOT).join(name)
+}
+
+// Small persisted state under /run/proxmox-ebpf/<name>/<key>, used to decide on each run whether
+// to tear down (schema changed) and whether to refresh existing links (the object changed).
+fn read_state(name: &str, key: &str) -> Option<u64> {
+    std::fs::read_to_string(PathBuf::from(RUN_ROOT).join(name).join(key))
+        .ok()
+        .and_then(|s| s.trim().parse().ok())
+}
+
+fn write_state(name: &str, key: &str, value: u64) -> anyhow::Result<()> {
+    let path = PathBuf::from(RUN_ROOT).join(name).join(key);
+    if let Some(parent) = path.parent() {
+        std::fs::create_dir_all(parent)?;
+    }
+    std::fs::write(path, format!("{value}\n"))?;
+    Ok(())
+}
+
+/// The pinned tc programs for one subsystem.
+///
+/// Holds only the immutable description. The loaded BPF lives in the kernel, pinned, and is
+/// reached back through those pins, so normal operation runs no verifier.
+pub struct TcPrograms {
+    name: &'static str,
+    obj: &'static [u8],
+    fingerprint: u64,
+    prog_name: fn(Direction) -> &'static str,
+    schema_version: Option<u32>,
+}
+
+impl TcPrograms {
+    pub fn new(
+        name: &'static str,
+        obj: &'static [u8],
+        fingerprint: u64,
+        prog_name: fn(Direction) -> &'static str,
+        schema_version: Option<u32>,
+    ) -> Self {
+        Self {
+            name,
+            obj,
+            fingerprint,
+            prog_name,
+            schema_version,
+        }
+    }
+
+    fn pin_root(&self) -> PathBuf {
+        pin_root_for(self.name)
+    }
+    fn links_dir(&self) -> PathBuf {
+        self.pin_root().join("links")
+    }
+    fn link_pin_path(&self, ifindex: u32, dir: Direction) -> PathBuf {
+        self.links_dir().join(tc::pin_filename(ifindex, dir))
+    }
+    fn prog_dir(&self) -> PathBuf {
+        self.pin_root().join("prog")
+    }
+    fn prog_pin_path(&self, dir: Direction) -> PathBuf {
+        self.prog_dir().join(dir.as_str())
+    }
+
+    /// Open a pinned BPF hash map by name, for the owning subsystem to sync. Valid once
+    /// [`ensure_loaded`](Self::ensure_loaded) has run, which is every caller's first step.
+    pub fn hash_map<K: aya::Pod, V: aya::Pod>(
+        &self,
+        name: &str,
+    ) -> anyhow::Result<aya::maps::HashMap<aya::maps::MapData, K, V>> {
+        let map = aya::maps::MapData::from_pin(self.pin_root().join(name))
+            .with_context(|| format!("{}: open pinned map {name}", self.name))?;
+        Ok(aya::maps::HashMap::try_from(aya::maps::Map::HashMap(map))?)
+    }
+
+    /// Open one direction's pinned program. A plain `BPF_OBJ_GET`, no verifier, since the program
+    /// was checked once when [`ensure_loaded`](Self::ensure_loaded) installed it.
+    fn program(&self, dir: Direction) -> anyhow::Result<SchedClassifier> {
+        SchedClassifier::from_pin(self.prog_pin_path(dir))
+            .with_context(|| format!("{}: open pinned program {}", self.name, dir.as_str()))
+    }
+
+    /// FNV-1a over the embedded object. A `const fn`, so each subsystem folds it into a `const` at
+    /// compile time and the per-invocation path never rehashes a constant.
+    pub const fn obj_fingerprint(obj: &[u8]) -> u64 {
+        let mut hash = 0xcbf29ce484222325u64;
+        let mut i = 0;
+        while i < obj.len() {
+            hash ^= obj[i] as u64;
+            hash = hash.wrapping_mul(0x100000001b3);
+            i += 1;
+        }
+        hash
+    }
+
+    /// The per-subsystem apply lock under `/run`, one file taken in two modes. A full apply (and
+    /// any install/teardown) takes it [exclusively](Self::lock_exclusive) so its enumerate, detach
+    /// and attach run as one unit. An additive single-interface apply takes it
+    /// [shared](Self::lock_shared) so guest plugs run concurrently and only block while a full
+    /// apply holds it. The kernel drops the lock if the process dies.
+    fn lock(&self, arg: FlockArg) -> anyhow::Result<Flock<File>> {
+        let dir = PathBuf::from(RUN_ROOT).join(self.name);
+        std::fs::create_dir_all(&dir)?;
+        let file = File::create(dir.join("lock"))?;
+        Flock::lock(file, arg).map_err(|(_, e)| anyhow::Error::new(e))
+    }
+
+    /// Take the apply lock in shared mode, for an additive single-interface apply.
+    pub fn lock_shared(&self) -> anyhow::Result<Flock<File>> {
+        self.lock(FlockArg::LockShared)
+    }
+
+    /// Take the apply lock exclusively, for a full apply or an install/teardown.
+    pub fn lock_exclusive(&self) -> anyhow::Result<Flock<File>> {
+        self.lock(FlockArg::LockExclusive)
+    }
+
+    /// Make sure the programs are loaded and pinned. The load runs only when there is no current
+    /// pin, when the schema version changed (a rebuild), or when the object changed (a refresh).
+    /// Otherwise the programs and maps pinned by an earlier run are reused untouched. Returns
+    /// whether a (re)install happened.
+    ///
+    /// Takes no lock of its own: the caller holds the apply lock exclusively whenever an install
+    /// may be needed (it gates on [`is_current`](Self::is_current) first), so the install path
+    /// here is already serialized. A subsystem with no maps passes no schema version and so never
+    /// verifies or tears down.
+    pub fn ensure_loaded(&self) -> anyhow::Result<bool> {
+        if self.is_current() {
+            return Ok(false);
+        }
+
+        // rebuild when the pinned programs are of an unknown or different schema: a different
+        // recorded schema_version, or none recorded while programs are still pinned (the /run
+        // markers were lost but the bpffs pins survived, e.g. across a service restart). The
+        // loaded map layout is then unknown, so tear down rather than bind new code to old maps.
+        let schema_changed = self.programs_pinned()
+            && match self.schema_version {
+                Some(v) => read_state(self.name, "schema_version") != Some(v as u64),
+                None => false,
+            };
+        let code_changed = read_state(self.name, "fingerprint") != Some(self.fingerprint);
+
+        if schema_changed {
+            log::warn!("{}: schema changed or unknown, rebuilding", self.name);
+            // verify the new code loads against throw-away state before tearing the old down, so a
+            // verifier rejection can't leave us with the old state wiped and nothing to replace it
+            self.verify().with_context(|| {
+                format!(
+                    "{}: new BPF code does not load against fresh state",
+                    self.name
+                )
+            })?;
+            self.tear_down().context("tear_down")?;
+        }
+
+        self.load_and_pin().context("load_and_pin")?;
+
+        // refresh links pinned by a previous run onto the new code. no-op on a fresh install or
+        // right after a teardown, where there are no links yet
+        if code_changed {
+            self.swap_existing_links();
+        }
+
+        if let Some(v) = self.schema_version
+            && let Err(e) = write_state(self.name, "schema_version", v as u64)
+        {
+            log::warn!("{}: failed to persist schema_version: {e:#}", self.name);
+        }
+        if let Err(e) = write_state(self.name, "fingerprint", self.fingerprint) {
+            log::warn!("{}: failed to persist fingerprint: {e:#}", self.name);
+        }
+        Ok(true)
+    }
+
+    /// True when the pinned programs already match this build, the `/run` fingerprint and schema
+    /// version agree with the embedded object and both programs are pinned. Lock-free. Callers use
+    /// it to decide whether a run needs the exclusive lock (an install) or only the shared one.
+    pub fn is_current(&self) -> bool {
+        let schema_changed = match self.schema_version {
+            Some(v) => read_state(self.name, "schema_version").is_some_and(|s| s != v as u64),
+            None => false,
+        };
+        let code_changed = read_state(self.name, "fingerprint") != Some(self.fingerprint);
+        !schema_changed && !code_changed && self.programs_pinned()
+    }
+
+    fn programs_pinned(&self) -> bool {
+        DIRECTIONS
+            .iter()
+            .all(|&dir| self.prog_pin_path(dir).exists())
+    }
+
+    fn verify(&self) -> anyhow::Result<()> {
+        tc::verify(&[self.obj], &DIRECTIONS.map(self.prog_name))
+    }
+
+    fn tear_down(&self) -> anyhow::Result<()> {
+        match std::fs::remove_dir_all(self.pin_root()) {
+            Ok(()) => {}
+            Err(e) if e.kind() == ErrorKind::NotFound => {}
+            Err(e) => return Err(e.into()),
+        }
+        std::fs::create_dir_all(self.links_dir())?;
+        Ok(())
+    }
+
+    /// Load and verify the object, then pin its programs. The verifier runs here and in the
+    /// throwaway [`verify`](Self::verify), nowhere else. Steady-state runs reuse the pins. The
+    /// `bpf` handle is dropped at the end, the pinned programs and maps stay resident in the
+    /// kernel.
+    fn load_and_pin(&self) -> anyhow::Result<()> {
+        std::fs::create_dir_all(self.links_dir())?;
+        std::fs::create_dir_all(self.prog_dir())?;
+        let mut bpf = EbpfLoader::new()
+            .map_pin_path(self.pin_root())
+            .load(self.obj)?;
+        for dir in DIRECTIONS {
+            let prog: &mut SchedClassifier =
+                bpf.program_mut((self.prog_name)(dir)).unwrap().try_into()?;
+            prog.load()?;
+            let path = self.prog_pin_path(dir);
+            // refreshed program is a new kernel object but the old pin file still exists, so
+            // remove it before re-pinning or pin() hits EEXIST
+            match std::fs::remove_file(&path) {
+                Ok(()) => {}
+                Err(e) if e.kind() == ErrorKind::NotFound => {}
+                Err(e) => return Err(e).context("remove stale program pin"),
+            }
+            prog.pin(&path)?;
+        }
+        Ok(())
+    }
+
+    fn live_links(&self) -> anyhow::Result<Vec<(u32, Direction)>> {
+        tc::read_pinned_links(&self.links_dir())
+    }
+
+    fn swap_existing_links(&self) {
+        let live = match self.live_links() {
+            Ok(l) => l,
+            Err(e) => {
+                log::error!("{}: read pinned links: {e:#}", self.name);
+                return;
+            }
+        };
+        for (ifindex, dir) in live {
+            if let Err(e) = self.swap_pinned_link(ifindex, dir) {
+                log::error!("{}: swap link {ifindex}-{}: {e:#}", self.name, dir.as_str());
+            }
+        }
+    }
+
+    /// Make the attached set match `desired`. Detach interfaces no longer wanted, attach the ones
+    /// missing. Refreshing existing links onto new code is done by
+    /// [`ensure_loaded`](Self::ensure_loaded). Runs under the caller's exclusive apply lock, so the
+    /// live set it samples cannot change under it.
+    ///
+    /// Returns an error if any interface failed to attach (after attempting all of them): a NIC
+    /// left without its program would pass traffic unenforced, so that surfaces as a failed apply
+    /// rather than a silent fail-open. A failed detach only leaves over-enforcement and is logged.
+    pub fn reconcile(&self, desired: &HashSet<u32>) -> anyhow::Result<()> {
+        let live: HashSet<(u32, Direction)> = self.live_links()?.into_iter().collect();
+
+        for &(ifidx, dir) in &live {
+            if desired.contains(&ifidx) {
+                continue;
+            }
+            log::debug!("{}: detach {ifidx}-{}", self.name, dir.as_str());
+            if let Err(e) = tc::detach_pinned_link(&self.link_pin_path(ifidx, dir)) {
+                log::error!("{}: detach {ifidx}-{}: {e:#}", self.name, dir.as_str());
+            } else {
+                log::info!("{}: detached {ifidx}-{}", self.name, dir.as_str());
+            }
+        }
+
+        let mut failed = 0usize;
+        for &ifidx in desired {
+            for dir in DIRECTIONS {
+                if live.contains(&(ifidx, dir)) {
+                    continue;
+                }
+                if let Err(e) = self.attach_and_pin(ifidx, dir) {
+                    log::error!("{}: attach {ifidx}-{}: {e:#}", self.name, dir.as_str());
+                    failed += 1;
+                }
+            }
+        }
+        if failed > 0 {
+            anyhow::bail!("{}: {failed} interface(s) failed to attach", self.name);
+        }
+        Ok(())
+    }
+
+    /// Attach the programs to a single interface, additively, without touching others.
+    ///
+    /// If a pin already exists, swap the program in place with no traffic gap. The swap only
+    /// succeeds while the link is still on the live netdev at this ifindex. A recycled ifindex
+    /// leaves a defunct pin, so the swap fails and we reclaim it and attach fresh. A failed attach
+    /// propagates so the caller can refuse to bring the NIC up unenforced.
+    pub fn attach_iface(&self, ifindex: u32) -> anyhow::Result<()> {
+        for dir in DIRECTIONS {
+            let path = self.link_pin_path(ifindex, dir);
+            if path.exists() {
+                match self.swap_pinned_link(ifindex, dir) {
+                    Ok(()) => continue,
+                    Err(e) => {
+                        log::debug!(
+                            "{}: {ifindex}-{} swap failed ({e:#}), rebuilding",
+                            self.name,
+                            dir.as_str()
+                        );
+                        if let Err(e) = tc::detach_pinned_link(&path) {
+                            log::warn!(
+                                "{}: reclaim {ifindex}-{}: unpin stale link: {e:#}",
+                                self.name,
+                                dir.as_str()
+                            );
+                            let _ = std::fs::remove_file(&path);
+                        }
+                    }
+                }
+            }
+            self.attach_and_pin(ifindex, dir)
+                .with_context(|| format!("{}: attach {ifindex}-{}", self.name, dir.as_str()))?;
+        }
+        Ok(())
+    }
+
+    fn swap_pinned_link(&self, ifindex: u32, dir: Direction) -> anyhow::Result<()> {
+        let path = self.link_pin_path(ifindex, dir);
+        let mut prog = self.program(dir)?;
+        tc::swap_pinned_link(&mut prog, &path)?;
+        Ok(())
+    }
+
+    fn attach_and_pin(&self, ifindex: u32, dir: Direction) -> anyhow::Result<()> {
+        let path = self.link_pin_path(ifindex, dir);
+        let mut prog = self.program(dir)?;
+        tc::attach_and_pin(&mut prog, ifindex, dir, &path)?;
+        Ok(())
+    }
+
+    /// Detach every pinned link and drop all pinned and `/run` state for this subsystem, under the
+    /// exclusive lock. For package removal: leaving links attached would keep interfaces enforcing
+    /// the last-applied policy with no agent left to update them. Best-effort past the lock: logs
+    /// and continues so one failure does not strand the rest.
+    pub fn clear(&self) -> anyhow::Result<()> {
+        let _lock = self.lock_exclusive()?;
+        for (ifindex, dir) in self.live_links().unwrap_or_default() {
+            if let Err(e) = tc::detach_pinned_link(&self.link_pin_path(ifindex, dir)) {
+                log::warn!("{}: detach {ifindex}-{}: {e:#}", self.name, dir.as_str());
+            }
+        }
+        for path in [self.pin_root(), PathBuf::from(RUN_ROOT).join(self.name)] {
+            if let Err(e) = std::fs::remove_dir_all(&path)
+                && e.kind() != ErrorKind::NotFound
+            {
+                log::warn!("{}: remove {}: {e:#}", self.name, path.display());
+            }
+        }
+        Ok(())
+    }
+}
diff --git a/src/tc.rs b/src/tc.rs
new file mode 100644
index 0000000..f8a9e86
--- /dev/null
+++ b/src/tc.rs
@@ -0,0 +1,152 @@
+//! TC link plumbing shared by the [`policy`](crate::policy) and [`bridge`](crate::bridge)
+//! subsystems. A `Direction` enum, attach/swap/detach free functions, and a uniform pin-filename
+//! layout `{ifindex}-{direction}`.
+
+use std::{io::ErrorKind, path::Path, str::FromStr};
+
+use anyhow::Context;
+use aya::{
+    EbpfLoader,
+    programs::{
+        SchedClassifier, TcAttachType,
+        links::{FdLink, PinnedLink},
+        tc,
+    },
+};
+
+use crate::subsystem::VERIFY_ROOT;
+
+#[derive(Copy, Clone, Hash, PartialEq, Eq, Debug)]
+pub enum Direction {
+    Ingress,
+    Egress,
+}
+
+pub const DIRECTIONS: [Direction; 2] = [Direction::Ingress, Direction::Egress];
+
+impl Direction {
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Self::Ingress => "ingress",
+            Self::Egress => "egress",
+        }
+    }
+    pub fn aya_type(self) -> TcAttachType {
+        match self {
+            Self::Ingress => TcAttachType::Ingress,
+            Self::Egress => TcAttachType::Egress,
+        }
+    }
+}
+
+impl FromStr for Direction {
+    type Err = ();
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "ingress" => Ok(Self::Ingress),
+            "egress" => Ok(Self::Egress),
+            _ => Err(()),
+        }
+    }
+}
+
+/// RAII handle for the shared verify root. Constructor wipes any stale contents and creates the
+/// dir. `Drop` removes it. Cleanup happens even if the verify body panics or returns Err.
+struct VerifyRoot;
+
+impl VerifyRoot {
+    fn new() -> anyhow::Result<Self> {
+        let _ = std::fs::remove_dir_all(VERIFY_ROOT);
+        std::fs::create_dir_all(VERIFY_ROOT)
+            .with_context(|| format!("create verify root {VERIFY_ROOT}"))?;
+        Ok(Self)
+    }
+    fn path(&self) -> &Path {
+        Path::new(VERIFY_ROOT)
+    }
+}
+
+impl Drop for VerifyRoot {
+    fn drop(&mut self) {
+        if let Err(e) = std::fs::remove_dir_all(VERIFY_ROOT) {
+            log::warn!("failed to clean up verify root {VERIFY_ROOT}: {e:#}");
+        }
+    }
+}
+
+/// Loads every named program in each object against a throwaway pin root, catching verifier
+/// regressions before any real state is touched.
+pub fn verify(objects: &[&[u8]], program_names: &[&str]) -> anyhow::Result<()> {
+    let root = VerifyRoot::new()?;
+    for &obj in objects {
+        let mut bpf = EbpfLoader::new().map_pin_path(root.path()).load(obj)?;
+        for &name in program_names {
+            let p: &mut SchedClassifier = bpf.program_mut(name).unwrap().try_into()?;
+            p.load()?;
+        }
+    }
+    Ok(())
+}
+
+pub fn pin_filename(ifindex: u32, dir: Direction) -> String {
+    format!("{ifindex}-{}", dir.as_str())
+}
+
+/// Ensure clsact qdisc on the iface, attach `prog` in `dir`, pin the link.
+pub fn attach_and_pin(
+    prog: &mut SchedClassifier,
+    ifindex: u32,
+    dir: Direction,
+    pin_path: &Path,
+) -> anyhow::Result<()> {
+    let name = nix::net::if_::if_indextoname(ifindex)?;
+    let name = name.to_str()?;
+    let _ = tc::qdisc_add_clsact(name);
+    let link_id = prog.attach(name, dir.aya_type())?;
+    let link = prog.take_link(link_id)?;
+    let fd_link: FdLink = link.try_into()?;
+    fd_link.pin(pin_path)?;
+    Ok(())
+}
+
+/// Rebind a pinned link to `prog` via LINK_UPDATE. Atomic, traffic sees no detach/reattach gap.
+pub fn swap_pinned_link(prog: &mut SchedClassifier, pin_path: &Path) -> anyhow::Result<()> {
+    let pinned = PinnedLink::from_pin(pin_path)?;
+    let fd_link: FdLink = pinned.into();
+    let link = fd_link.try_into()?;
+    let new_id = prog.attach_to_link(link)?;
+    // take the handle out of aya's internal tracking, we have the pin
+    let _ = prog.take_link(new_id)?;
+    Ok(())
+}
+
+pub fn detach_pinned_link(pin_path: &Path) -> anyhow::Result<()> {
+    let pinned = PinnedLink::from_pin(pin_path)?;
+    let _fd_link = pinned.unpin()?;
+    Ok(())
+}
+
+/// Read and parse every pin file in `links_dir` as `{ifindex}-{direction}`. Unrecognized names are
+/// skipped with a warning.
+pub fn read_pinned_links(links_dir: &Path) -> anyhow::Result<Vec<(u32, Direction)>> {
+    let mut out = Vec::new();
+    let dir = match std::fs::read_dir(links_dir) {
+        Ok(d) => d,
+        Err(e) if e.kind() == ErrorKind::NotFound => return Ok(out),
+        Err(e) => return Err(e.into()),
+    };
+    for entry in dir {
+        let entry = entry?;
+        let name = entry.file_name();
+        let name = name.to_string_lossy();
+        let Some((ifidx_str, dir_str)) = name.split_once('-') else {
+            log::warn!("unrecognized pin file in links dir: {name}");
+            continue;
+        };
+        match (ifidx_str.parse::<u32>(), dir_str.parse::<Direction>()) {
+            (Ok(ifidx), Ok(d)) => out.push((ifidx, d)),
+            _ => log::warn!("unrecognized pin file in links dir: {name}"),
+        }
+    }
+    Ok(out)
+}
-- 
2.47.3