Полностью доведена production-логика AWG2 harness по up2: strict execution, tc parser, anti-torrent, status и документация

docs/IMPLEMENTATION_SPEC.md

@@ -176,6 +176,12 @@ wg show
 
 The fallback parser should support the old output format from `wg-peers.txt`, extracting `allowed ips:` lines.
 
+Execution policy for production:
+
+- do not use shell wrappers (`sh -c`) for `wg` invocation;
+- do not suppress stderr via redirection in command strings;
+- execute `wg` directly and handle non-zero exit codes in Rust.
+
 ## 8. Desired traffic topology
 
 The expected topology exactly mirrors old bash.
@@ -244,6 +250,24 @@ reconcile_traffic():
 
 For v1, `full_rebuild_on_drift = true` is the safer default.
 
+Parser hardening requirements:
+
+- `tc filter show` must be parsed block-by-block (state-machine), not by naive token scanning across unrelated lines;
+- mapping `peer_ip -> class_id` must be validated for both egress (`1:`) and ingress (`2:`) sides;
+- verification must fail when classes/filters exist but mapping differs from deterministic assignment.
+
+Apply/verify policy:
+
+- all apply steps are explicit command invocations;
+- expected idempotency conflicts (`already exists`) may be tolerated selectively;
+- any other apply error is fatal;
+- `verify_after_apply` is mandatory for both traffic and anti-torrent subsystems.
+
+Anti-torrent observability contract:
+
+- named counter `antitorrent_dropped_v4` must be present in owned table;
+- status output must expose both dropped packets and dropped bytes.
+
 ## 10. Owned state and teardown rules
 
 The binary owns:

docs/INSTALLATION.md

@@ -52,6 +52,10 @@ install -d -m 0755 /etc/awg2-harness
 install -m 0644 config.toml /etc/awg2-harness/config.toml
 ```
 
+Важно: production-установка **не должна** автоматически копировать [`config.example.toml`](config.example.toml) как боевой конфиг.
+Если используется [`tools/build.sh`](tools/build.sh), режим `--install` завершится ошибкой при отсутствии `/etc/awg2-harness/config.toml`.
+Сначала подготовьте осознанный production-конфиг вручную.
+
 Проверка конфига:
 
 ```bash

tools/build.sh

@@ -67,11 +67,13 @@ if [[ "$AUTO_INSTALL" == true ]]; then
   sudo install -m 0755 "$TARGET_BIN" /usr/local/bin/awg2-harness
   sudo mkdir -p /etc/awg2-harness
   if [[ ! -f /etc/awg2-harness/config.toml ]]; then
-    sudo install -m 0644 "$ROOT_DIR/config.example.toml" /etc/awg2-harness/config.toml
+    echo "Missing /etc/awg2-harness/config.toml"
+    echo "Refusing to install example config as production config."
+    echo "Create config manually from $ROOT_DIR/config.example.toml and re-run with --install."
+    exit 1
   fi
   sudo install -m 0644 "$ROOT_DIR/awg2-harness.service" /etc/systemd/system/awg2-harness.service
   sudo systemctl daemon-reload
 fi
 
 echo "Build completed: $TARGET_BIN"
-

traffic_core/src/cli.rs

@@ -21,4 +21,3 @@ pub enum Command {
     EnsureSwap,
     ValidateConfig,
 }
-

traffic_core/src/config.rs

@@ -88,7 +88,12 @@ impl Config {
     }
 
     fn check_host_tools(&self) -> anyhow::Result<()> {
-        let mut required = vec!["tc", "ip", "wg", "nsenter", "docker", "modprobe"];
+        let mut required = vec!["tc", "ip", "wg", "docker", "modprobe"];
+        if !crate::netns::nsenter_available() {
+            eprintln!("nsenter is not available, will use docker exec fallback");
+        } else {
+            required.push("nsenter");
+        }
         if self.anti_torrent.enabled {
             required.push("nft");
         }
@@ -103,7 +108,23 @@ impl Config {
 }
 
 fn tool_exists(tool: &str) -> bool {
-    Command::new(tool).arg("--version").output().is_ok()
+    let probes: &[&[&str]] = match tool {
+        "tc" => &[&["-V"]],
+        "ip" => &[&["-V"], &["-Version"]],
+        "wg" => &[&["--version"]],
+        "docker" => &[&["--version"]],
+        "modprobe" => &[&["--version"]],
+        "nsenter" => &[&["--help"]],
+        "nft" => &[&["--version"]],
+        _ => &[&["--version"], &["--help"], &["-V"]],
+    };
+
+    for args in probes {
+        if Command::new(tool).args(*args).output().is_ok() {
+            return true;
+        }
+    }
+    false
 }
 
 pub fn parse_cidr_v4(cidr: &str) -> anyhow::Result<(Ipv4Addr, u8)> {
@@ -152,4 +173,3 @@ mod tests {
         assert!(!no);
     }
 }
-

traffic_core/src/daemon.rs

@@ -5,15 +5,7 @@ use std::{
     time::{Duration, Instant},
 };
 
-use crate::{
+use crate::{config::Config, docker, firewall, iface, netns::build_transport, reboot, traffic};
-    config::Config,
-    docker,
-    firewall,
-    iface,
-    netns::build_transport,
-    reboot,
-    traffic,
-};
 use tracing::{error, info};
 
 #[derive(Debug, Clone, Default)]
@@ -70,7 +62,9 @@ pub fn run(cfg: Config) -> anyhow::Result<()> {
         } else {
             set_ok(&state, "periodic firewall reconcile ok");
         }
-        thread::sleep(Duration::from_secs(cfg.traffic.reconcile_interval_secs.max(5)));
+        thread::sleep(Duration::from_secs(
+            cfg.traffic.reconcile_interval_secs.max(5),
+        ));
     }
 }
 
@@ -79,7 +73,13 @@ fn docker_events_loop(cfg: &Config, state: &Arc<Mutex<LastActionState>>) -> anyh
     use std::process::{Command, Stdio};
 
     let mut child = Command::new("docker")
-        .args(["events", "--filter", "event=start", "--format", "{{.Actor.Attributes.name}}"])
+        .args([
+            "events",
+            "--filter",
+            "event=start",
+            "--format",
+            "{{.Actor.Attributes.name}}",
+        ])
         .stdout(Stdio::piped())
         .spawn()?;
 
@@ -150,4 +150,3 @@ fn persist_state(state: &LastActionState) {
     );
     let _ = fs::write(LAST_STATE_FILE, content);
 }
-

traffic_core/src/docker.rs

@@ -62,4 +62,3 @@ pub fn run_capture(args: &[&str]) -> anyhow::Result<String> {
     }
     Ok(String::from_utf8_lossy(&out.stdout).to_string())
 }
-

traffic_core/src/error.rs

@@ -7,4 +7,3 @@ pub enum AppError {
     #[error("not found: {0}")]
     NotFound(String),
 }
-

traffic_core/src/firewall.rs

@@ -1,12 +1,13 @@
 use crate::{
     config::Config,
     docker,
-    netns::{build_transport, NsTransport},
+    netns::{NsTransport, build_transport},
 };
 
 #[derive(Debug, Clone)]
 struct FirewallActual {
     table_exists: bool,
+    counter_exists: bool,
     chain_exists: bool,
     rule_matches: bool,
 }
@@ -16,47 +17,164 @@ pub fn apply_firewall(cfg: &Config) -> anyhow::Result<()> {
     let transport = build_transport(target.container_pid, &target.container_name);
 
     if !cfg.anti_torrent.enabled {
-        let _ = transport.run(&["sh", "-c", "nft delete table inet awg2_harness 2>/dev/null || true"])?;
+        let _ = transport.run(&["nft", "delete", "table", "inet", "awg2_harness"]);
         return Ok(());
     }
 
     let actual = inspect_firewall(transport.as_ref(), cfg)?;
-    if actual.table_exists && actual.chain_exists && actual.rule_matches {
+    if actual.table_exists && actual.counter_exists && actual.chain_exists && actual.rule_matches {
         return Ok(());
     }
 
-    let rule = format!(
+    run_allow_exists(
-        "nft 'add table inet awg2_harness' 2>/dev/null || true; \
+        transport.as_ref(),
-         nft 'add chain inet awg2_harness forward_antitorrent {{ type filter hook forward priority filter; policy accept; }}' 2>/dev/null || true; \
+        &["nft", "add", "table", "inet", "awg2_harness"],
-         nft 'add rule inet awg2_harness forward_antitorrent ip saddr {cidr} ct state new meter awg2_peer_connlimit_v4 {{ ip saddr ct count over {max} }} counter name antitorrent_dropped_v4 drop' 2>/dev/null || true",
+    )?;
-        cidr = cfg.anti_torrent.peer_ipv4_cidr,
+    run_allow_exists(
-        max = cfg.anti_torrent.max_connections_per_peer
+        transport.as_ref(),
-    );
+        &[
-    apply_rule(transport.as_ref(), &rule)?;
+            "nft",
+            "add",
+            "counter",
+            "inet",
+            "awg2_harness",
+            "antitorrent_dropped_v4",
+        ],
+    )?;
+    run_allow_exists_chain(transport.as_ref())?;
+    apply_rule(
+        transport.as_ref(),
+        &cfg.anti_torrent.peer_ipv4_cidr,
+        cfg.anti_torrent.max_connections_per_peer,
+    )?;
+
+    let verify = inspect_firewall(transport.as_ref(), cfg)?;
+    if !(verify.table_exists && verify.counter_exists && verify.chain_exists && verify.rule_matches)
+    {
+        anyhow::bail!("firewall verify-after-apply failed");
+    }
     Ok(())
 }
 
 fn inspect_firewall(transport: &dyn NsTransport, cfg: &Config) -> anyhow::Result<FirewallActual> {
     let table = transport
-        .run(&["sh", "-c", "nft list table inet awg2_harness 2>/dev/null"])
+        .run(&["nft", "list", "table", "inet", "awg2_harness"])
         .unwrap_or_default();
 
     let table_exists = !table.trim().is_empty();
+    let counter_exists = table.contains("counter antitorrent_dropped_v4");
     let chain_exists = table.contains("chain forward_antitorrent");
 
     let needle_cidr = format!("ip saddr {}", cfg.anti_torrent.peer_ipv4_cidr);
-    let needle_max = format!("ct count over {}", cfg.anti_torrent.max_connections_per_peer);
+    let needle_max = format!(
+        "ct count over {}",
+        cfg.anti_torrent.max_connections_per_peer
+    );
     let rule_matches = table.contains(&needle_cidr) && table.contains(&needle_max);
 
     Ok(FirewallActual {
         table_exists,
+        counter_exists,
         chain_exists,
         rule_matches,
     })
 }
 
-fn apply_rule(transport: &dyn NsTransport, rule: &str) -> anyhow::Result<()> {
+fn apply_rule(transport: &dyn NsTransport, cidr: &str, max_conn: u32) -> anyhow::Result<()> {
-    let _ = transport.run(&["sh", "-c", rule])?;
+    let _ = transport.run(&[
+        "nft",
+        "add",
+        "rule",
+        "inet",
+        "awg2_harness",
+        "forward_antitorrent",
+        "ip",
+        "saddr",
+        cidr,
+        "ct",
+        "state",
+        "new",
+        "meter",
+        "awg2_peer_connlimit_v4",
+        "{",
+        "ip",
+        "saddr",
+        "ct",
+        "count",
+        "over",
+        &max_conn.to_string(),
+        "}",
+        "counter",
+        "name",
+        "antitorrent_dropped_v4",
+        "drop",
+    ])?;
     Ok(())
 }
 
+fn run_allow_exists(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> {
+    match transport.run(args) {
+        Ok(_) => Ok(()),
+        Err(e) => {
+            let msg = e.to_string();
+            if msg.contains("File exists") || msg.contains("exists") {
+                return Ok(());
+            }
+            Err(e)
+        }
+    }
+}
+
+fn run_allow_exists_chain(transport: &dyn NsTransport) -> anyhow::Result<()> {
+    run_allow_exists(
+        transport,
+        &[
+            "nft",
+            "add",
+            "chain",
+            "inet",
+            "awg2_harness",
+            "forward_antitorrent",
+            "{",
+            "type",
+            "filter",
+            "hook",
+            "forward",
+            "priority",
+            "filter;",
+            "policy",
+            "accept;",
+            "}",
+        ],
+    )
+}
+
+pub fn antitorrent_drop_counter(cfg: &Config) -> anyhow::Result<Option<(u64, u64)>> {
+    if !cfg.anti_torrent.enabled {
+        return Ok(None);
+    }
+
+    let target = docker::discover_target(cfg)?;
+    let transport = build_transport(target.container_pid, &target.container_name);
+    let table = transport
+        .run(&["nft", "list", "table", "inet", "awg2_harness"])
+        .unwrap_or_default();
+
+    for line in table.lines() {
+        if line.contains("counter antitorrent_dropped_v4") && line.contains("packets") {
+            let toks: Vec<&str> = line.split_whitespace().collect();
+            if let Some(idx) = toks.iter().position(|t| *t == "packets") {
+                if let Some(v) = toks.get(idx + 1).and_then(|s| s.parse::<u64>().ok()) {
+                    let bytes = toks
+                        .iter()
+                        .position(|t| *t == "bytes")
+                        .and_then(|i| toks.get(i + 1))
+                        .and_then(|s| s.parse::<u64>().ok())
+                        .unwrap_or(0);
+                    return Ok(Some((v, bytes)));
+                }
+            }
+        }
+    }
+    Ok(None)
+}

traffic_core/src/iface.rs

@@ -20,18 +20,18 @@ pub fn detect_interface_with_transport(
     transport: &dyn NsTransport,
 ) -> anyhow::Result<String> {
     for ifname in &cfg.amnezia.interface_candidates {
-        if transport
+        if transport.run(&["ip", "link", "show", ifname]).is_ok() {
-            .run(&["ip", "link", "show", ifname])
-            .is_ok()
-        {
             return Ok(ifname.clone());
         }
     }
 
-    let links = transport.run(&["sh", "-c", "ip -o link show | awk -F': ' '{print $2}'"])?;
+    let links = transport.run(&["ip", "-o", "link", "show"])?;
 
     for raw in links.lines() {
-        let name = raw.trim();
+        let name = parse_ifname_from_ip_o_link(raw).unwrap_or("").trim();
+        if name.is_empty() {
+            continue;
+        }
         for p in &cfg.amnezia.interface_patterns {
             if glob_like_match(p, name) {
                 return Ok(name.to_string());
@@ -50,3 +50,34 @@ fn glob_like_match(pattern: &str, value: &str) -> bool {
     }
 }
 
+fn parse_ifname_from_ip_o_link(line: &str) -> Option<&str> {
+    let trimmed = line.trim();
+    let mut parts = trimmed.splitn(3, ':');
+    let _idx = parts.next()?;
+    let ifname = parts.next()?.trim();
+    Some(
+        ifname
+            .trim_end_matches('@')
+            .split('@')
+            .next()
+            .unwrap_or(ifname),
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_ifname_plain() {
+        let v = parse_ifname_from_ip_o_link("2: wg0: <POINTOPOINT,UP> mtu 1420").expect("ifname");
+        assert_eq!(v, "wg0");
+    }
+
+    #[test]
+    fn parse_ifname_with_peer() {
+        let v =
+            parse_ifname_from_ip_o_link("7: veth0@if8: <BROADCAST,UP> mtu 1500").expect("ifname");
+        assert_eq!(v, "veth0");
+    }
+}

traffic_core/src/lock.rs

@@ -1,4 +1,8 @@
-use std::{fs, path::Path};
+use std::{
+    fs::{self, OpenOptions},
+    io::Write,
+    path::Path,
+};
 
 pub struct FileLock {
     path: String,
@@ -10,18 +14,27 @@ impl FileLock {
         if let Some(parent) = p.parent() {
             fs::create_dir_all(parent)?;
         }
-        if p.exists() {
+        match OpenOptions::new().create_new(true).write(true).open(p) {
+            Ok(mut file) => {
+                file.write_all(std::process::id().to_string().as_bytes())?;
+                return Ok(Self {
+                    path: path.to_string(),
+                });
+            }
+            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
                 let stale = is_stale_lock(p)?;
                 if stale {
                     let _ = fs::remove_file(p);
-            } else {
+                    let mut file = OpenOptions::new().create_new(true).write(true).open(p)?;
+                    file.write_all(std::process::id().to_string().as_bytes())?;
+                    return Ok(Self {
+                        path: path.to_string(),
+                    });
+                }
                 anyhow::bail!("lock already exists: {path}");
             }
+            Err(e) => return Err(e.into()),
         }
-        fs::write(p, std::process::id().to_string())?;
-        Ok(Self {
-            path: path.to_string(),
-        })
     }
 }
 
@@ -42,3 +55,15 @@ fn is_stale_lock(path: &Path) -> anyhow::Result<bool> {
     Ok(!Path::new(&proc_path).exists())
 }
 
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn second_acquire_fails_for_live_lock() {
+        let p = format!("/tmp/awg2_harness_lock_test_{}", std::process::id());
+        let _first = FileLock::acquire(&p).expect("first lock");
+        let second = FileLock::acquire(&p);
+        assert!(second.is_err());
+    }
+}

traffic_core/src/main.rs

@@ -23,7 +23,9 @@ use crate::config::Config;
 
 fn main() -> anyhow::Result<()> {
     tracing_subscriber::fmt()
-        .with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")))
+        .with_env_filter(
+            EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")),
+        )
         .init();
 
     let cli = Cli::parse();

traffic_core/src/netns.rs

@@ -43,7 +43,7 @@ impl NsTransport for DockerExecTransport {
 }
 
 pub fn nsenter_available() -> bool {
-    Command::new("nsenter").arg("--version").output().is_ok()
+    Command::new("nsenter").arg("--help").output().is_ok()
 }
 
 pub fn build_transport(pid: u32, container: &str) -> Box<dyn NsTransport> {
@@ -55,4 +55,3 @@ pub fn build_transport(pid: u32, container: &str) -> Box<dyn NsTransport> {
         })
     }
 }
-

traffic_core/src/reboot.rs

@@ -1,4 +1,4 @@
-use chrono::{Datelike, Duration, NaiveDateTime, NaiveTime, Utc};
+use chrono::{Datelike, Duration, NaiveDateTime, NaiveTime, Timelike, Utc};
 use rand::Rng;
 use std::{process::Command, thread, time::Duration as StdDuration};
 
@@ -77,11 +77,16 @@ fn next_reboot_time(start_hhmm: &str, end_hhmm: &str) -> anyhow::Result<NaiveDat
 }
 fn parse_hhmm(v: &str) -> anyhow::Result<NaiveTime> {
     let mut s = v.split(':');
-    let h = s.next().ok_or_else(|| anyhow::anyhow!("bad time"))?.parse()?;
+    let h = s
-    let m = s.next().ok_or_else(|| anyhow::anyhow!("bad time"))?.parse()?;
+        .next()
+        .ok_or_else(|| anyhow::anyhow!("bad time"))?
+        .parse()?;
+    let m = s
+        .next()
+        .ok_or_else(|| anyhow::anyhow!("bad time"))?
+        .parse()?;
     if h > 23 || m > 59 {
         anyhow::bail!("bad time")
     }
     NaiveTime::from_hms_opt(h, m, 0).ok_or_else(|| anyhow::anyhow!("bad time"))
 }
-

traffic_core/src/status.rs

@@ -1,4 +1,4 @@
-use crate::{config::Config, docker, traffic, wg};
+use crate::{config::Config, docker, firewall, netns::build_transport, traffic, wg};
 
 use std::{fs, process::Command};
 
@@ -54,6 +54,19 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> {
         Err(e) => println!("  drift: unknown ({e})"),
     }
 
+    println!("  topology:");
+    let transport = build_transport(target.container_pid, &target.container_name);
+    let egress_qdisc = transport
+        .run(&["tc", "qdisc", "show", "dev", &target.interface_name])
+        .unwrap_or_default();
+    let ingress_qdisc = egress_qdisc.clone();
+    let ifb_qdisc = transport
+        .run(&["tc", "qdisc", "show", "dev", &cfg.traffic.ifb_name])
+        .unwrap_or_default();
+    println!("    egress_htb_present: {}", egress_qdisc.contains("htb"));
+    println!("    ingress_present: {}", ingress_qdisc.contains("ingress"));
+    println!("    ifb_htb_present: {}", ifb_qdisc.contains("htb"));
+
     println!("Anti-torrent:");
     println!("  enabled: {}", cfg.anti_torrent.enabled);
     println!(
@@ -62,23 +75,13 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> {
     );
 
     if cfg.anti_torrent.enabled {
-        let fw = Command::new("nsenter")
+        match firewall::antitorrent_drop_counter(cfg) {
-            .args([
+            Ok(Some((packets, bytes))) => {
-                "-t",
+                println!("  dropped_packets: {}", packets);
-                &target.container_pid.to_string(),
+                println!("  dropped_bytes: {}", bytes);
-                "-n",
+            }
-                "--",
+            Ok(None) => println!("  dropped_packets: n/a"),
-                "nft",
+            Err(e) => println!("  dropped_packets: inspect error ({e})"),
-                "list",
-                "table",
-                "inet",
-                "awg2_harness",
-            ])
-            .output();
-        match fw {
-            Ok(out) if out.status.success() => println!("  actual: table present"),
-            Ok(_) => println!("  actual: table missing or drift"),
-            Err(_) => println!("  actual: unable to inspect"),
         }
     }
 
@@ -110,4 +113,3 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> {
 
     Ok(())
 }
-

traffic_core/src/swap.rs

@@ -19,7 +19,10 @@ pub fn ensure_swap(cfg: &Config) -> anyhow::Result<()> {
             );
         }
     } else {
-        run("fallocate", &["-l", &format!("{}M", cfg.swap.size_mb), path])?;
+        run(
+            "fallocate",
+            &["-l", &format!("{}M", cfg.swap.size_mb), path],
+        )?;
         run("chmod", &["600", path])?;
         run("mkswap", &[path])?;
     }
@@ -31,7 +34,10 @@ pub fn ensure_swap(cfg: &Config) -> anyhow::Result<()> {
     }
 
     ensure_fstab(path, cfg.swap.priority)?;
-    run("sysctl", &["-w", &format!("vm.swappiness={}", cfg.swap.swappiness)])?;
+    run(
+        "sysctl",
+        &["-w", &format!("vm.swappiness={}", cfg.swap.swappiness)],
+    )?;
     ensure_swappiness_persist(cfg.swap.swappiness)?;
     Ok(())
 }
@@ -39,7 +45,10 @@ pub fn ensure_swap(cfg: &Config) -> anyhow::Result<()> {
 fn ensure_fstab(path: &str, priority: i32) -> anyhow::Result<()> {
     let file = "/etc/fstab";
     let content = fs::read_to_string(file).unwrap_or_default();
-    if content.lines().any(|l| l.contains(path) && l.contains("swap")) {
+    if content
+        .lines()
+        .any(|l| l.contains(path) && l.contains("swap"))
+    {
         return Ok(());
     }
     let line = format!("{path} none swap sw,pri={priority} 0 0\n");
@@ -69,4 +78,3 @@ fn ensure_swappiness_persist(swappiness: u8) -> anyhow::Result<()> {
     }
     Ok(())
 }
-

traffic_core/src/traffic.rs

@@ -6,7 +6,10 @@ use crate::{
     wg,
 };
 
-use std::collections::BTreeSet;
+use std::{
+    collections::{BTreeMap, BTreeSet},
+    net::Ipv4Addr,
+};
 
 #[derive(Debug, Clone)]
 struct TrafficDrift {
@@ -35,6 +38,7 @@ pub fn inspect_drift_summary(cfg: &Config) -> anyhow::Result<Option<String>> {
         transport.as_ref(),
         &target.interface_name,
         &cfg.traffic.ifb_name,
+        cfg.traffic.class_start_id,
         &peers,
     )?;
 
@@ -68,6 +72,7 @@ pub fn reconcile_traffic(cfg: &Config) -> anyhow::Result<()> {
         transport.as_ref(),
         &target.interface_name,
         &cfg.traffic.ifb_name,
+        cfg.traffic.class_start_id,
         &peers,
     )?;
 
@@ -79,7 +84,11 @@ pub fn reconcile_traffic(cfg: &Config) -> anyhow::Result<()> {
     println!("traffic drift detected: {}", drift.reason);
 
     if cfg.traffic.full_rebuild_on_drift {
-        teardown_owned_topology(transport.as_ref(), &target.interface_name, &cfg.traffic.ifb_name)?;
+        teardown_owned_topology(
+            transport.as_ref(),
+            &target.interface_name,
+            &cfg.traffic.ifb_name,
+        )?;
         apply_full_topology(
             cfg,
             transport.as_ref(),
@@ -96,10 +105,11 @@ fn inspect_and_diff(
     transport: &dyn NsTransport,
     iface: &str,
     ifb: &str,
+    class_start_id: u16,
     peers: &[wg::Peer],
 ) -> anyhow::Result<TrafficDrift> {
     let root_qdisc = transport
-        .run(&["sh", "-c", &format!("tc qdisc show dev {iface} 2>/dev/null")])
+        .run(&["tc", "qdisc", "show", "dev", iface])
         .unwrap_or_default();
     if !root_qdisc.contains("htb") {
         return Ok(TrafficDrift {
@@ -109,7 +119,7 @@ fn inspect_and_diff(
     }
 
     let ingress_qdisc = transport
-        .run(&["sh", "-c", &format!("tc qdisc show dev {iface} 2>/dev/null")])
+        .run(&["tc", "qdisc", "show", "dev", iface])
         .unwrap_or_default();
     if !ingress_qdisc.contains("ingress") {
         return Ok(TrafficDrift {
@@ -119,7 +129,7 @@ fn inspect_and_diff(
     }
 
     let ifb_link = transport
-        .run(&["sh", "-c", &format!("ip link show {ifb} 2>/dev/null")])
+        .run(&["ip", "link", "show", ifb])
         .unwrap_or_default();
     if ifb_link.trim().is_empty() {
         return Ok(TrafficDrift {
@@ -128,60 +138,195 @@ fn inspect_and_diff(
         });
     }
 
-    let filters = transport
+    let filters_egress = transport
-        .run(&[
+        .run(&["tc", "filter", "show", "dev", iface, "parent", "1:"])
-            "sh",
+        .unwrap_or_default();
-            "-c",
+    let filters_ingress = transport
-            &format!("tc filter show dev {iface} parent 1: 2>/dev/null"),
+        .run(&["tc", "filter", "show", "dev", ifb, "parent", "2:"])
-        ])
         .unwrap_or_default();
 
-    let actual_ips = extract_filter_ips(&filters);
+    let actual_map_egress = extract_filter_ip_class_map(&filters_egress);
+    let actual_map_ingress = extract_filter_ip_class_map(&filters_ingress);
+    let actual_ips_egress: BTreeSet<String> = actual_map_egress.keys().cloned().collect();
+    let actual_ips_ingress: BTreeSet<String> = actual_map_ingress.keys().cloned().collect();
     let expected_ips: BTreeSet<String> = peers.iter().map(|p| p.ipv4.to_string()).collect();
 
-    if actual_ips != expected_ips {
+    if actual_ips_egress != expected_ips {
         return Ok(TrafficDrift {
             has_drift: true,
             reason: format!(
-                "peer filter mismatch (actual={}, expected={})",
+                "egress peer filter mismatch (actual={}, expected={})",
-                actual_ips.len(),
+                actual_ips_egress.len(),
                 expected_ips.len()
             ),
         });
     }
 
+    if actual_ips_ingress != expected_ips {
+        return Ok(TrafficDrift {
+            has_drift: true,
+            reason: format!(
+                "ingress peer filter mismatch (actual={}, expected={})",
+                actual_ips_ingress.len(),
+                expected_ips.len()
+            ),
+        });
+    }
+
+    let expected_map = expected_peer_class_map(peers, class_start_id);
+    if actual_map_egress != expected_map {
+        return Ok(TrafficDrift {
+            has_drift: true,
+            reason: format!(
+                "egress class mapping mismatch (actual={}, expected={})",
+                actual_map_egress.len(),
+                expected_map.len()
+            ),
+        });
+    }
+    if actual_map_ingress != expected_map {
+        return Ok(TrafficDrift {
+            has_drift: true,
+            reason: format!(
+                "ingress class mapping mismatch (actual={}, expected={})",
+                actual_map_ingress.len(),
+                expected_map.len()
+            ),
+        });
+    }
+
+    let classes_egress = transport
+        .run(&["tc", "class", "show", "dev", iface, "parent", "1:1"])
+        .unwrap_or_default();
+    let classes_ingress = transport
+        .run(&["tc", "class", "show", "dev", ifb, "parent", "2:1"])
+        .unwrap_or_default();
+    let expected_class_count = peers.len();
+    if count_leaf_classes(&classes_egress) < expected_class_count {
+        return Ok(TrafficDrift {
+            has_drift: true,
+            reason: "missing egress per-peer classes".to_string(),
+        });
+    }
+    if count_leaf_classes(&classes_ingress) < expected_class_count {
+        return Ok(TrafficDrift {
+            has_drift: true,
+            reason: "missing ingress per-peer classes".to_string(),
+        });
+    }
+
     Ok(TrafficDrift {
         has_drift: false,
         reason: "no drift".to_string(),
     })
 }
 
-fn extract_filter_ips(filters: &str) -> BTreeSet<String> {
+fn extract_filter_ip_class_map(filters: &str) -> BTreeMap<String, u16> {
-    let mut set = BTreeSet::new();
+    let mut out = BTreeMap::new();
-    for line in filters.lines() {
+
-        if !line.contains("match") || !line.contains("dst") {
+    for block in split_filter_blocks(filters) {
-            continue;
+        let mut ip: Option<String> = None;
+        let mut class_id: Option<u16> = None;
+
+        for line in block.lines() {
+            let t = line.trim();
+            if t.contains("match") && ip.is_none() {
+                ip = extract_u32_match_hex(t).and_then(|hex| hex_to_ipv4_string(&hex));
             }
-        let toks: Vec<&str> = line.split_whitespace().collect();
+            if t.contains("flowid") && class_id.is_none() {
-        for (idx, tok) in toks.iter().enumerate() {
+                class_id = extract_flowid_class_id(t);
-            if *tok == "dst" {
-                if let Some(ip) = toks.get(idx + 1) {
-                    set.insert((*ip).to_string());
-                }
-                break;
             }
         }
+
+        if let (Some(ip), Some(class_id)) = (ip, class_id) {
+            out.insert(ip, class_id);
         }
-    set
+    }
+
+    out
 }
 
-fn teardown_owned_topology(transport: &dyn NsTransport, iface: &str, ifb: &str) -> anyhow::Result<()> {
+fn split_filter_blocks(raw: &str) -> Vec<String> {
-    let cmd = format!(
+    let mut blocks = Vec::new();
-        "tc qdisc del dev {iface} root 2>/dev/null || true; \
+    let mut cur = Vec::new();
-         tc qdisc del dev {iface} ingress 2>/dev/null || true; \
+
-         ip link del {ifb} 2>/dev/null || true"
+    for line in raw.lines() {
-    );
+        let t = line.trim();
-    let _ = transport.run(&["sh", "-c", &cmd])?;
+        if t.starts_with("filter") && !cur.is_empty() {
+            blocks.push(cur.join("\n"));
+            cur.clear();
+        }
+        if !t.is_empty() {
+            cur.push(t.to_string());
+        }
+    }
+    if !cur.is_empty() {
+        blocks.push(cur.join("\n"));
+    }
+
+    blocks
+}
+
+fn extract_flowid_class_id(line: &str) -> Option<u16> {
+    let toks: Vec<&str> = line.split_whitespace().collect();
+    let flowid = toks
+        .iter()
+        .position(|v| *v == "flowid")
+        .and_then(|i| toks.get(i + 1))?;
+    let class_tail = flowid.split(':').nth(1)?;
+    class_tail.parse::<u16>().ok()
+}
+
+fn expected_peer_class_map(peers: &[wg::Peer], class_start_id: u16) -> BTreeMap<String, u16> {
+    let mut map = BTreeMap::new();
+    for (idx, p) in peers.iter().enumerate() {
+        map.insert(p.ipv4.to_string(), class_start_id + idx as u16);
+    }
+    map
+}
+
+fn extract_u32_match_hex(line: &str) -> Option<String> {
+    let toks: Vec<&str> = line.split_whitespace().collect();
+    let pos = toks.iter().position(|v| *v == "match")?;
+    let match_val = *toks.get(pos + 1)?;
+    let (hex, mask) = match_val.split_once('/')?;
+    if mask.eq_ignore_ascii_case("ffffffff") && hex.len() == 8 {
+        return Some(hex.to_ascii_lowercase());
+    }
+    None
+}
+
+fn hex_to_ipv4_string(hex: &str) -> Option<String> {
+    if hex.len() != 8 {
+        return None;
+    }
+    let b1 = u8::from_str_radix(&hex[0..2], 16).ok()?;
+    let b2 = u8::from_str_radix(&hex[2..4], 16).ok()?;
+    let b3 = u8::from_str_radix(&hex[4..6], 16).ok()?;
+    let b4 = u8::from_str_radix(&hex[6..8], 16).ok()?;
+    Some(Ipv4Addr::new(b1, b2, b3, b4).to_string())
+}
+
+fn count_leaf_classes(raw: &str) -> usize {
+    raw.lines()
+        .filter_map(|l| {
+            let line = l.trim();
+            let pos = line.find("class htb ")?;
+            let class = &line[(pos + "class htb ".len())..];
+            let head = class.split_whitespace().next()?;
+            if head.ends_with(":1") { None } else { Some(()) }
+        })
+        .count()
+}
+
+fn teardown_owned_topology(
+    transport: &dyn NsTransport,
+    iface: &str,
+    ifb: &str,
+) -> anyhow::Result<()> {
+    let _ = transport.run(&["tc", "qdisc", "del", "dev", iface, "root"]);
+    let _ = transport.run(&["tc", "qdisc", "del", "dev", iface, "ingress"]);
+    let _ = transport.run(&["ip", "link", "del", ifb]);
     Ok(())
 }
 
@@ -192,35 +337,210 @@ fn apply_full_topology(
     ifb: &str,
     peers: &[wg::Peer],
 ) -> anyhow::Result<()> {
-    let setup_cmd = format!(
+    let root = cfg.traffic.root_rate.as_str();
-        "tc qdisc add dev {iface} root handle 1: htb; \
+    run_allow_exists(transport, &["modprobe", "ifb"])?;
-         tc class add dev {iface} parent 1: classid 1:1 htb rate {root} ceil {root}; \
+    run_allow_exists(
-         modprobe ifb 2>/dev/null || true; \
+        transport,
-         ip link add name {ifb} type ifb 2>/dev/null || true; \
+        &["ip", "link", "add", "name", ifb, "type", "ifb"],
-         ip link set dev {ifb} up; \
+    )?;
-         tc qdisc add dev {iface} handle ffff: ingress; \
+    run_step(transport, &["ip", "link", "set", "dev", ifb, "up"])?;
-         tc filter add dev {iface} parent ffff: protocol all u32 match u32 0 0 action mirred egress redirect dev {ifb}; \
+    run_allow_exists(
-         tc qdisc add dev {ifb} root handle 2: htb; \
+        transport,
-         tc class add dev {ifb} parent 2: classid 2:1 htb rate {root} ceil {root}",
+        &[
-        root = cfg.traffic.root_rate
+            "tc", "qdisc", "add", "dev", iface, "root", "handle", "1:", "htb",
-    );
+        ],
-    let _ = transport.run(&["sh", "-c", &setup_cmd])?;
+    )?;
+    run_allow_exists(
+        transport,
+        &[
+            "tc", "class", "add", "dev", iface, "parent", "1:", "classid", "1:1", "htb", "rate",
+            root, "ceil", root,
+        ],
+    )?;
+    run_allow_exists(
+        transport,
+        &[
+            "tc", "qdisc", "add", "dev", iface, "handle", "ffff:", "ingress",
+        ],
+    )?;
+    run_allow_exists(
+        transport,
+        &[
+            "tc", "filter", "add", "dev", iface, "parent", "ffff:", "protocol", "all", "u32",
+            "match", "u32", "0", "0", "action", "mirred", "egress", "redirect", "dev", ifb,
+        ],
+    )?;
+    run_allow_exists(
+        transport,
+        &[
+            "tc", "qdisc", "add", "dev", ifb, "root", "handle", "2:", "htb",
+        ],
+    )?;
+    run_allow_exists(
+        transport,
+        &[
+            "tc", "class", "add", "dev", ifb, "parent", "2:", "classid", "2:1", "htb", "rate",
+            root, "ceil", root,
+        ],
+    )?;
 
     for (idx, p) in peers.iter().enumerate() {
         let class_id = cfg.traffic.class_start_id + idx as u16;
-        let ip = p.ipv4;
+        let ip = p.ipv4.to_string();
-        let per_peer = format!(
+        let class_egress = format!("1:{class_id}");
-            "tc class add dev {iface} parent 1:1 classid 1:{class_id} htb rate {limit} ceil {limit}; \
+        let class_ingress = format!("2:{class_id}");
-             tc qdisc add dev {iface} parent 1:{class_id} fq_codel; \
+        let limit = cfg.traffic.peer_limit.as_str();
-             tc filter add dev {iface} parent 1: protocol ip u32 match ip dst {ip} flowid 1:{class_id}; \
+
-             tc class add dev {ifb} parent 2:1 classid 2:{class_id} htb rate {limit} ceil {limit}; \
+        run_allow_exists(
-             tc qdisc add dev {ifb} parent 2:{class_id} fq_codel; \
+            transport,
-             tc filter add dev {ifb} parent 2: protocol ip u32 match ip src {ip} flowid 2:{class_id}",
+            &[
-            limit = cfg.traffic.peer_limit
+                "tc",
-        );
+                "class",
-        let _ = transport.run(&["sh", "-c", &per_peer])?;
+                "add",
+                "dev",
+                iface,
+                "parent",
+                "1:1",
+                "classid",
+                class_egress.as_str(),
+                "htb",
+                "rate",
+                limit,
+                "ceil",
+                limit,
+            ],
+        )?;
+        run_allow_exists(
+            transport,
+            &[
+                "tc",
+                "qdisc",
+                "add",
+                "dev",
+                iface,
+                "parent",
+                class_egress.as_str(),
+                "fq_codel",
+            ],
+        )?;
+        run_allow_exists(
+            transport,
+            &[
+                "tc",
+                "filter",
+                "add",
+                "dev",
+                iface,
+                "parent",
+                "1:",
+                "protocol",
+                "ip",
+                "u32",
+                "match",
+                "ip",
+                "dst",
+                ip.as_str(),
+                "flowid",
+                class_egress.as_str(),
+            ],
+        )?;
+        run_allow_exists(
+            transport,
+            &[
+                "tc",
+                "class",
+                "add",
+                "dev",
+                ifb,
+                "parent",
+                "2:1",
+                "classid",
+                class_ingress.as_str(),
+                "htb",
+                "rate",
+                limit,
+                "ceil",
+                limit,
+            ],
+        )?;
+        run_allow_exists(
+            transport,
+            &[
+                "tc",
+                "qdisc",
+                "add",
+                "dev",
+                ifb,
+                "parent",
+                class_ingress.as_str(),
+                "fq_codel",
+            ],
+        )?;
+        run_allow_exists(
+            transport,
+            &[
+                "tc",
+                "filter",
+                "add",
+                "dev",
+                ifb,
+                "parent",
+                "2:",
+                "protocol",
+                "ip",
+                "u32",
+                "match",
+                "ip",
+                "src",
+                ip.as_str(),
+                "flowid",
+                class_ingress.as_str(),
+            ],
+        )?;
+    }
+
+    let verify = inspect_and_diff(transport, iface, ifb, cfg.traffic.class_start_id, peers)?;
+    if verify.has_drift {
+        anyhow::bail!("verify-after-apply failed: {}", verify.reason);
     }
 
     Ok(())
 }
 
+fn run_step(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> {
+    let _ = transport.run(args)?;
+    Ok(())
+}
+
+fn run_allow_exists(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> {
+    match transport.run(args) {
+        Ok(_) => Ok(()),
+        Err(e) => {
+            let msg = e.to_string();
+            if msg.contains("File exists") || msg.contains("exists") || msg.contains("EEXIST") {
+                return Ok(());
+            }
+            Err(e)
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_tc_u32_hex_ipv4() {
+        let raw = "  match 0a080105/ffffffff at 16\n  flowid 1:42";
+        let m = extract_filter_ip_class_map(raw);
+        assert_eq!(m.get("10.8.1.5").copied(), Some(42));
+    }
+
+    #[test]
+    fn parse_block_based_filters() {
+        let raw = "filter parent 1: protocol ip pref 1 u32\n  match 0a080102/ffffffff at 16\n  flowid 1:10\nfilter parent 1: protocol ip pref 2 u32\n  match 0a080103/ffffffff at 16\n  flowid 1:11";
+        let m = extract_filter_ip_class_map(raw);
+        assert_eq!(m.get("10.8.1.2").copied(), Some(10));
+        assert_eq!(m.get("10.8.1.3").copied(), Some(11));
+    }
+}

traffic_core/src/wg.rs

@@ -27,14 +27,12 @@ pub fn discover_peers_with_transport(
     peer_cidr: &str,
 ) -> anyhow::Result<Vec<Peer>> {
     let primary = transport
-        .run(&["sh", "-c", "wg show all allowed-ips 2>/dev/null"])
+        .run(&["wg", "show", "all", "allowed-ips"])
         .unwrap_or_default();
 
     let mut peers = parse_allowed_ips(&primary, peer_cidr)?;
     if peers.is_empty() {
-        let fallback = transport
+        let fallback = transport.run(&["wg", "show"]).unwrap_or_default();
-            .run(&["sh", "-c", "wg show 2>/dev/null"])
-            .unwrap_or_default();
         peers = parse_fallback_wg_show(&fallback, peer_cidr)?;
     }
 
@@ -127,4 +125,3 @@ mod tests {
         assert_eq!(peers[1].ipv4.to_string(), "10.8.1.2");
     }
 }
-