From d9f140b3e0a19b124f3b46232bbf0c8e771558dd Mon Sep 17 00:00:00 2001 From: Crimson Date: Sat, 9 May 2026 23:38:53 +0500 Subject: [PATCH] =?UTF-8?q?=D0=9F=D0=BE=D0=BB=D0=BD=D0=BE=D1=81=D1=82?= =?UTF-8?q?=D1=8C=D1=8E=20=D0=B4=D0=BE=D0=B2=D0=B5=D0=B4=D0=B5=D0=BD=D0=B0?= =?UTF-8?q?=20production-=D0=BB=D0=BE=D0=B3=D0=B8=D0=BA=D0=B0=20AWG2=20har?= =?UTF-8?q?ness=20=D0=BF=D0=BE=20up2:=20strict=20execution,=20tc=20parser,?= =?UTF-8?q?=20anti-torrent,=20status=20=D0=B8=20=D0=B4=D0=BE=D0=BA=D1=83?= =?UTF-8?q?=D0=BC=D0=B5=D0=BD=D1=82=D0=B0=D1=86=D0=B8=D1=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/IMPLEMENTATION_SPEC.md | 24 ++ docs/INSTALLATION.md | 4 + tools/build.sh | 6 +- traffic_core/src/cli.rs | 1 - traffic_core/src/config.rs | 26 ++- traffic_core/src/daemon.rs | 23 +- traffic_core/src/docker.rs | 1 - traffic_core/src/error.rs | 1 - traffic_core/src/firewall.rs | 148 ++++++++++-- traffic_core/src/iface.rs | 43 +++- traffic_core/src/lock.rs | 45 +++- traffic_core/src/main.rs | 4 +- traffic_core/src/netns.rs | 3 +- traffic_core/src/reboot.rs | 13 +- traffic_core/src/status.rs | 40 ++-- traffic_core/src/swap.rs | 16 +- traffic_core/src/traffic.rs | 440 ++++++++++++++++++++++++++++++----- traffic_core/src/wg.rs | 7 +- 18 files changed, 699 insertions(+), 146 deletions(-) diff --git a/docs/IMPLEMENTATION_SPEC.md b/docs/IMPLEMENTATION_SPEC.md index 323eead..d68a45e 100644 --- a/docs/IMPLEMENTATION_SPEC.md +++ b/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: diff --git a/docs/INSTALLATION.md b/docs/INSTALLATION.md index 05466eb..2e33525 100644 --- a/docs/INSTALLATION.md +++ b/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 diff --git a/tools/build.sh b/tools/build.sh index b3cb252..e870f8b 100644 --- a/tools/build.sh +++ b/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" - diff --git a/traffic_core/src/cli.rs b/traffic_core/src/cli.rs index b908721..895b1da 100644 --- a/traffic_core/src/cli.rs +++ b/traffic_core/src/cli.rs @@ -21,4 +21,3 @@ pub enum Command { EnsureSwap, ValidateConfig, } - diff --git a/traffic_core/src/config.rs b/traffic_core/src/config.rs index e10207d..ee98384 100644 --- a/traffic_core/src/config.rs +++ b/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); } } - diff --git a/traffic_core/src/daemon.rs b/traffic_core/src/daemon.rs index 4f33f46..086f9ab 100644 --- a/traffic_core/src/daemon.rs +++ b/traffic_core/src/daemon.rs @@ -5,15 +5,7 @@ use std::{ time::{Duration, Instant}, }; -use crate::{ - config::Config, - docker, - firewall, - iface, - netns::build_transport, - reboot, - traffic, -}; +use crate::{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>) -> 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); } - diff --git a/traffic_core/src/docker.rs b/traffic_core/src/docker.rs index 69f5210..4068663 100644 --- a/traffic_core/src/docker.rs +++ b/traffic_core/src/docker.rs @@ -62,4 +62,3 @@ pub fn run_capture(args: &[&str]) -> anyhow::Result { } Ok(String::from_utf8_lossy(&out.stdout).to_string()) } - diff --git a/traffic_core/src/error.rs b/traffic_core/src/error.rs index fd398ca..f3168fb 100644 --- a/traffic_core/src/error.rs +++ b/traffic_core/src/error.rs @@ -7,4 +7,3 @@ pub enum AppError { #[error("not found: {0}")] NotFound(String), } - diff --git a/traffic_core/src/firewall.rs b/traffic_core/src/firewall.rs index 03c46e0..7def823 100644 --- a/traffic_core/src/firewall.rs +++ b/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!( - "nft 'add table inet awg2_harness' 2>/dev/null || true; \ - nft 'add chain inet awg2_harness forward_antitorrent {{ type filter hook forward priority filter; policy accept; }}' 2>/dev/null || true; \ - 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, - max = cfg.anti_torrent.max_connections_per_peer - ); - apply_rule(transport.as_ref(), &rule)?; + run_allow_exists( + transport.as_ref(), + &["nft", "add", "table", "inet", "awg2_harness"], + )?; + run_allow_exists( + transport.as_ref(), + &[ + "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 { 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<()> { - let _ = transport.run(&["sh", "-c", rule])?; +fn apply_rule(transport: &dyn NsTransport, cidr: &str, max_conn: u32) -> anyhow::Result<()> { + 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> { + 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::().ok()) { + let bytes = toks + .iter() + .position(|t| *t == "bytes") + .and_then(|i| toks.get(i + 1)) + .and_then(|s| s.parse::().ok()) + .unwrap_or(0); + return Ok(Some((v, bytes))); + } + } + } + } + Ok(None) +} diff --git a/traffic_core/src/iface.rs b/traffic_core/src/iface.rs index 75aa950..c8054cd 100644 --- a/traffic_core/src/iface.rs +++ b/traffic_core/src/iface.rs @@ -20,18 +20,18 @@ pub fn detect_interface_with_transport( transport: &dyn NsTransport, ) -> anyhow::Result { for ifname in &cfg.amnezia.interface_candidates { - if transport - .run(&["ip", "link", "show", ifname]) - .is_ok() - { + if transport.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: 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: mtu 1500").expect("ifname"); + assert_eq!(v, "veth0"); + } +} diff --git a/traffic_core/src/lock.rs b/traffic_core/src/lock.rs index 495779f..c12fbe5 100644 --- a/traffic_core/src/lock.rs +++ b/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() { - let stale = is_stale_lock(p)?; - if stale { - let _ = fs::remove_file(p); - } else { + 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); + 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 { 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()); + } +} diff --git a/traffic_core/src/main.rs b/traffic_core/src/main.rs index 63ce2fc..b155bec 100644 --- a/traffic_core/src/main.rs +++ b/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(); diff --git a/traffic_core/src/netns.rs b/traffic_core/src/netns.rs index 9ceaaf8..0ffa2e1 100644 --- a/traffic_core/src/netns.rs +++ b/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 { @@ -55,4 +55,3 @@ pub fn build_transport(pid: u32, container: &str) -> Box { }) } } - diff --git a/traffic_core/src/reboot.rs b/traffic_core/src/reboot.rs index e9c6930..b1034aa 100644 --- a/traffic_core/src/reboot.rs +++ b/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 anyhow::Result { let mut s = v.split(':'); - let h = s.next().ok_or_else(|| anyhow::anyhow!("bad time"))?.parse()?; - let m = s.next().ok_or_else(|| anyhow::anyhow!("bad time"))?.parse()?; + let h = s + .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")) } - diff --git a/traffic_core/src/status.rs b/traffic_core/src/status.rs index a20e99f..4f59709 100644 --- a/traffic_core/src/status.rs +++ b/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") - .args([ - "-t", - &target.container_pid.to_string(), - "-n", - "--", - "nft", - "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"), + match firewall::antitorrent_drop_counter(cfg) { + Ok(Some((packets, bytes))) => { + println!(" dropped_packets: {}", packets); + println!(" dropped_bytes: {}", bytes); + } + Ok(None) => println!(" dropped_packets: n/a"), + Err(e) => println!(" dropped_packets: inspect error ({e})"), } } @@ -110,4 +113,3 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> { Ok(()) } - diff --git a/traffic_core/src/swap.rs b/traffic_core/src/swap.rs index 6e3f456..a266636 100644 --- a/traffic_core/src/swap.rs +++ b/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(()) } - diff --git a/traffic_core/src/traffic.rs b/traffic_core/src/traffic.rs index a2cec8f..5e46406 100644 --- a/traffic_core/src/traffic.rs +++ b/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> { 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 { 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 - .run(&[ - "sh", - "-c", - &format!("tc filter show dev {iface} parent 1: 2>/dev/null"), - ]) + let filters_egress = transport + .run(&["tc", "filter", "show", "dev", iface, "parent", "1:"]) + .unwrap_or_default(); + let filters_ingress = transport + .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 = actual_map_egress.keys().cloned().collect(); + let actual_ips_ingress: BTreeSet = actual_map_ingress.keys().cloned().collect(); let expected_ips: BTreeSet = 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={})", - actual_ips.len(), + "egress peer filter mismatch (actual={}, expected={})", + 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 { - let mut set = BTreeSet::new(); - for line in filters.lines() { - if !line.contains("match") || !line.contains("dst") { - continue; - } - let toks: Vec<&str> = line.split_whitespace().collect(); - for (idx, tok) in toks.iter().enumerate() { - if *tok == "dst" { - if let Some(ip) = toks.get(idx + 1) { - set.insert((*ip).to_string()); - } - break; +fn extract_filter_ip_class_map(filters: &str) -> BTreeMap { + let mut out = BTreeMap::new(); + + for block in split_filter_blocks(filters) { + let mut ip: Option = None; + let mut class_id: Option = 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)); + } + if t.contains("flowid") && class_id.is_none() { + class_id = extract_flowid_class_id(t); } } + + 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<()> { - let cmd = format!( - "tc qdisc del dev {iface} root 2>/dev/null || true; \ - tc qdisc del dev {iface} ingress 2>/dev/null || true; \ - ip link del {ifb} 2>/dev/null || true" - ); - let _ = transport.run(&["sh", "-c", &cmd])?; +fn split_filter_blocks(raw: &str) -> Vec { + let mut blocks = Vec::new(); + let mut cur = Vec::new(); + + for line in raw.lines() { + let t = line.trim(); + 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 { + 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::().ok() +} + +fn expected_peer_class_map(peers: &[wg::Peer], class_start_id: u16) -> BTreeMap { + 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 { + 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 { + 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!( - "tc qdisc add dev {iface} root handle 1: htb; \ - tc class add dev {iface} parent 1: classid 1:1 htb rate {root} ceil {root}; \ - modprobe ifb 2>/dev/null || true; \ - ip link add name {ifb} type ifb 2>/dev/null || true; \ - ip link set dev {ifb} up; \ - tc qdisc add dev {iface} handle ffff: ingress; \ - tc filter add dev {iface} parent ffff: protocol all u32 match u32 0 0 action mirred egress redirect dev {ifb}; \ - tc qdisc add dev {ifb} root handle 2: htb; \ - tc class add dev {ifb} parent 2: classid 2:1 htb rate {root} ceil {root}", - root = cfg.traffic.root_rate - ); - let _ = transport.run(&["sh", "-c", &setup_cmd])?; + let root = cfg.traffic.root_rate.as_str(); + run_allow_exists(transport, &["modprobe", "ifb"])?; + run_allow_exists( + transport, + &["ip", "link", "add", "name", ifb, "type", "ifb"], + )?; + run_step(transport, &["ip", "link", "set", "dev", ifb, "up"])?; + run_allow_exists( + transport, + &[ + "tc", "qdisc", "add", "dev", iface, "root", "handle", "1:", "htb", + ], + )?; + 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 per_peer = format!( - "tc class add dev {iface} parent 1:1 classid 1:{class_id} htb rate {limit} ceil {limit}; \ - tc qdisc add dev {iface} parent 1:{class_id} fq_codel; \ - 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}; \ - tc qdisc add dev {ifb} parent 2:{class_id} fq_codel; \ - tc filter add dev {ifb} parent 2: protocol ip u32 match ip src {ip} flowid 2:{class_id}", - limit = cfg.traffic.peer_limit - ); - let _ = transport.run(&["sh", "-c", &per_peer])?; + let ip = p.ipv4.to_string(); + let class_egress = format!("1:{class_id}"); + let class_ingress = format!("2:{class_id}"); + let limit = cfg.traffic.peer_limit.as_str(); + + run_allow_exists( + transport, + &[ + "tc", + "class", + "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)); + } +} diff --git a/traffic_core/src/wg.rs b/traffic_core/src/wg.rs index 444716d..9f0c885 100644 --- a/traffic_core/src/wg.rs +++ b/traffic_core/src/wg.rs @@ -27,14 +27,12 @@ pub fn discover_peers_with_transport( peer_cidr: &str, ) -> anyhow::Result> { 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 - .run(&["sh", "-c", "wg show 2>/dev/null"]) - .unwrap_or_default(); + let fallback = transport.run(&["wg", "show"]).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"); } } -