[PATCH proxmox-ebpf 1/3] add the shared tc subsystem code

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

A subsystem owns tc classifier programs and their maps, attached per
interface and pinned in bpffs together with an object fingerprint and
a schema version, so state survives between the one-shot calls from
the consumer driving it and re-running never interrupts traffic.
Apply paths coordinate through a shared flock, loading takes it
exclusively.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 src/lib.rs       |   3 +
 src/subsystem.rs | 392 +++++++++++++++++++++++++++++++++++++++++++++++
 src/tc.rs        | 151 ++++++++++++++++++
 3 files changed, 546 insertions(+)
 create mode 100644 src/subsystem.rs
 create mode 100644 src/tc.rs

diff --git a/src/lib.rs b/src/lib.rs
index 244f44c..572d861 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,2 +1,5 @@
 //! eBPF subsystems, one per concern, each behind a cargo feature of its name. Nothing here runs
 //! on its own, consumers pull in only the subsystem they drive.
+
+pub mod subsystem;
+pub mod tc;
diff --git a/src/subsystem.rs b/src/subsystem.rs
new file mode 100644
index 0000000..466e5cb
--- /dev/null
+++ b/src/subsystem.rs
@@ -0,0 +1,392 @@
+//! The shared subsystem code. A subsystem's tc programs, one per direction it declares, 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. Each subsystem owns one [`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, 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,
+    directions: &'static [Direction],
+    schema_version: Option<u32>,
+}
+
+impl TcPrograms {
+    pub fn new(
+        name: &'static str,
+        obj: &'static [u8],
+        fingerprint: u64,
+        prog_name: fn(Direction) -> &'static str,
+        directions: &'static [Direction],
+        schema_version: Option<u32>,
+    ) -> Self {
+        Self {
+            name,
+            obj,
+            fingerprint,
+            prog_name,
+            directions,
+            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 {
+        self.directions
+            .iter()
+            .all(|&dir| self.prog_pin_path(dir).exists())
+    }
+
+    fn verify(&self) -> anyhow::Result<()> {
+        let names: Vec<&str> = self
+            .directions
+            .iter()
+            .map(|&d| (self.prog_name)(d))
+            .collect();
+        tc::verify(&[self.obj], &names)
+    }
+
+    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 self.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 self.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 self.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 nothing 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..205df48
--- /dev/null
+++ b/src/tc.rs
@@ -0,0 +1,151 @@
+//! TC link plumbing shared by all 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