From d4cd6a4e43455aba146606b3c4cd4f0dba6e9fdd Mon Sep 17 00:00:00 2001 From: Crimson Date: Sun, 10 May 2026 21:29:52 +0500 Subject: [PATCH] =?UTF-8?q?traffic=5Fcore:=20=D0=BF=D1=80=D0=BE=D0=B4?= =?UTF-8?q?=D0=B0=D0=BA=D1=88=D0=BD-=D1=84=D0=B8=D0=BA=D1=81=D1=8B=20runti?= =?UTF-8?q?me=20=D0=B8=20=D0=B2=D0=B0=D0=BB=D0=B8=D0=B4=D0=B0=D1=86=D0=B8?= =?UTF-8?q?=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config.example.toml | 4 +- traffic_core/src/cli.rs | 3 +- traffic_core/src/config.rs | 102 ++++++++++++++++++++++++++++++++- traffic_core/src/daemon.rs | 69 ++++++++++++++++------ traffic_core/src/firewall.rs | 42 +++++++++++--- traffic_core/src/main.rs | 18 +++--- traffic_core/src/status.rs | 108 +++++++++++++++++++++++++++++++++++ traffic_core/src/traffic.rs | 39 ++++++++++++- traffic_core/src/wg.rs | 20 +++---- 9 files changed, 353 insertions(+), 52 deletions(-) diff --git a/config.example.toml b/config.example.toml index d604374..0695c24 100644 --- a/config.example.toml +++ b/config.example.toml @@ -6,8 +6,8 @@ peer_ipv4_cidr = "10.8.1.0/24" [traffic] enabled = true -peer_limit = "50mbit" -root_rate = "500mbit" +peer_limit = "70mbit" +root_rate = "1000mbit" reconcile_interval_secs = 15 container_ready_timeout_secs = 120 class_start_id = 10 diff --git a/traffic_core/src/cli.rs b/traffic_core/src/cli.rs index 895b1da..5cd0483 100644 --- a/traffic_core/src/cli.rs +++ b/traffic_core/src/cli.rs @@ -5,7 +5,7 @@ use clap::{Parser, Subcommand}; #[derive(Debug, Parser)] #[command(name = "awg2-harness")] pub struct Cli { - #[arg(long)] + #[arg(long, global = true)] pub config: Option, #[command(subcommand)] @@ -20,4 +20,5 @@ pub enum Command { Status, EnsureSwap, ValidateConfig, + ValidateRuntime, } diff --git a/traffic_core/src/config.rs b/traffic_core/src/config.rs index d2bfa66..920dc85 100644 --- a/traffic_core/src/config.rs +++ b/traffic_core/src/config.rs @@ -61,13 +61,30 @@ impl Config { Ok(cfg) } - pub fn validate(&self) -> anyhow::Result<()> { + pub fn validate_config(&self) -> anyhow::Result<()> { if self.amnezia.container_names.is_empty() { anyhow::bail!("amnezia.container_names must not be empty"); } + if self.amnezia.interface_candidates.is_empty() { + anyhow::bail!("amnezia.interface_candidates must not be empty"); + } + if self.amnezia.interface_patterns.is_empty() { + anyhow::bail!("amnezia.interface_patterns must not be empty"); + } if self.traffic.class_start_id < 10 { anyhow::bail!("traffic.class_start_id must be >= 10"); } + if self.traffic.ifb_name.trim().is_empty() { + anyhow::bail!("traffic.ifb_name must not be empty"); + } + if self.traffic.peer_limit.trim().is_empty() { + anyhow::bail!("traffic.peer_limit must not be empty"); + } + if self.traffic.root_rate.trim().is_empty() { + anyhow::bail!("traffic.root_rate must not be empty"); + } + parse_tc_rate_bps(&self.traffic.peer_limit)?; + parse_tc_rate_bps(&self.traffic.root_rate)?; if self.swap.swappiness > 100 { anyhow::bail!("swap.swappiness must be <= 100"); } @@ -77,12 +94,36 @@ impl Config { if self.traffic.reconcile_interval_secs == 0 { anyhow::bail!("traffic.reconcile_interval_secs must be > 0"); } + if self.swap.enabled { + if self.swap.size_mb == 0 { + anyhow::bail!("swap.size_mb must be > 0 when swap.enabled=true"); + } + if !self.swap.path.starts_with('/') { + anyhow::bail!("swap.path must be absolute"); + } + } + if self.anti_torrent.enabled && self.anti_torrent.max_connections_per_peer == 0 { + anyhow::bail!("anti_torrent.max_connections_per_peer must be > 0"); + } if self.reboot.enabled && self.reboot.window_start_utc == self.reboot.window_end_utc { anyhow::bail!("reboot window start and end must differ"); } + if self.reboot.enabled { + parse_hhmm(&self.reboot.window_start_utc)?; + parse_hhmm(&self.reboot.window_end_utc)?; + } parse_cidr_v4(&self.amnezia.peer_ipv4_cidr)?; parse_cidr_v4(&self.anti_torrent.peer_ipv4_cidr)?; - self.check_host_tools()?; + Ok(()) + } + + pub fn validate_runtime(&self) -> anyhow::Result<()> { + self.check_host_tools() + } + + pub fn validate_all(&self) -> anyhow::Result<()> { + self.validate_config()?; + self.validate_runtime()?; Ok(()) } @@ -106,6 +147,49 @@ impl Config { } } +fn parse_hhmm(v: &str) -> anyhow::Result<()> { + let mut s = v.split(':'); + let h = s + .next() + .ok_or_else(|| anyhow::anyhow!("invalid time: {v}"))? + .parse::()?; + let m = s + .next() + .ok_or_else(|| anyhow::anyhow!("invalid time: {v}"))? + .parse::()?; + if s.next().is_some() { + anyhow::bail!("invalid time: {v}"); + } + if h > 23 || m > 59 { + anyhow::bail!("invalid time: {v}"); + } + Ok(()) +} + +fn parse_tc_rate_bps(v: &str) -> anyhow::Result { + let raw = v.trim().to_ascii_lowercase().replace(' ', ""); + if raw.is_empty() { + anyhow::bail!("invalid tc rate: empty"); + } + let (num, mul) = if let Some(n) = raw.strip_suffix("gbit") { + (n, 1_000_000_000u64) + } else if let Some(n) = raw.strip_suffix("mbit") { + (n, 1_000_000u64) + } else if let Some(n) = raw.strip_suffix("kbit") { + (n, 1_000u64) + } else if let Some(n) = raw.strip_suffix("bit") { + (n, 1u64) + } else { + anyhow::bail!("invalid tc rate unit: {v}"); + }; + + let base = num + .parse::() + .map_err(|_| anyhow::anyhow!("invalid tc rate number: {v}"))?; + base.checked_mul(mul) + .ok_or_else(|| anyhow::anyhow!("tc rate overflow: {v}")) +} + fn tool_exists(tool: &str) -> bool { let probes: &[&[&str]] = match tool { "tc" => &[&["-V"]], @@ -171,4 +255,18 @@ mod tests { let no = ip_in_cidr("10.9.1.10".parse().expect("ip"), "10.8.1.0/24").expect("cidr"); assert!(!no); } + + #[test] + fn parse_hhmm_works() { + assert!(parse_hhmm("01:00").is_ok()); + assert!(parse_hhmm("23:59").is_ok()); + assert!(parse_hhmm("24:00").is_err()); + } + + #[test] + fn parse_tc_rate_works() { + assert_eq!(parse_tc_rate_bps("1gbit").expect("bps"), 1_000_000_000); + assert_eq!(parse_tc_rate_bps("1000mbit").expect("bps"), 1_000_000_000); + assert!(parse_tc_rate_bps("foo").is_err()); + } } diff --git a/traffic_core/src/daemon.rs b/traffic_core/src/daemon.rs index e952169..25bb1c9 100644 --- a/traffic_core/src/daemon.rs +++ b/traffic_core/src/daemon.rs @@ -6,17 +6,21 @@ 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, traffic}; use tracing::{error, info}; #[derive(Debug, Clone, Default)] struct LastActionState { + last_ok_at: Option, last_ok: Option, + last_error_at: Option, last_error: Option, } const LAST_STATE_FILE: &str = "/run/awg2-harness/last_action.state"; const EVENTS_FILE: &str = "/run/awg2-harness/events.log"; +const EVENTS_MAX_LINES: usize = 1000; +const EVENTS_MAX_BYTES: usize = 256 * 1024; pub fn run(cfg: Config) -> anyhow::Result<()> { info!("starting daemon"); @@ -36,22 +40,22 @@ pub fn run(cfg: Config) -> anyhow::Result<()> { set_ok(&state, "startup firewall reconcile ok"); append_event("startup_firewall ok"); } - let _ = reboot::maybe_schedule_reboot(&cfg); - let reboot_cfg = cfg.clone(); - thread::spawn(move || { - if let Err(e) = reboot::reboot_scheduler_loop(reboot_cfg) { - error!(error = %e, "reboot scheduler loop error"); - } - }); - let events_cfg = cfg.clone(); let events_state = state.clone(); thread::spawn(move || { loop { - if let Err(e) = docker_events_loop(&events_cfg, &events_state) { - set_err(&events_state, format!("docker events loop crashed: {e}")); - thread::sleep(Duration::from_secs(10)); + match docker_events_loop(&events_cfg, &events_state) { + Ok(()) => { + set_err( + &events_state, + "docker events stream ended, restarting".to_string(), + ); + } + Err(e) => { + set_err(&events_state, format!("docker events loop crashed: {e}")); + } } + thread::sleep(Duration::from_secs(10)); } }); @@ -144,13 +148,16 @@ fn wait_for_interface(cfg: &Config, container_name: &str, timeout_secs: u64) -> fn set_ok(state: &Arc>, msg: &str) { if let Ok(mut s) = state.lock() { + s.last_ok_at = Some(now_rfc3339()); s.last_ok = Some(msg.to_string()); + s.last_error = None; persist_state(&s); } } fn set_err(state: &Arc>, msg: String) { if let Ok(mut s) = state.lock() { + s.last_error_at = Some(now_rfc3339()); s.last_error = Some(msg); persist_state(&s); } @@ -159,8 +166,10 @@ fn set_err(state: &Arc>, msg: String) { fn persist_state(state: &LastActionState) { let _ = fs::create_dir_all("/run/awg2-harness"); let content = format!( - "last_ok={}\nlast_error={}\n", + "last_ok_at={}\nlast_ok={}\nlast_error_at={}\nlast_error={}\n", + state.last_ok_at.clone().unwrap_or_default(), state.last_ok.clone().unwrap_or_default(), + state.last_error_at.clone().unwrap_or_default(), state.last_error.clone().unwrap_or_default() ); let _ = fs::write(LAST_STATE_FILE, content); @@ -168,12 +177,34 @@ fn persist_state(state: &LastActionState) { fn append_event(event: &str) { let _ = fs::create_dir_all("/run/awg2-harness"); - let ts = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); - if let Ok(mut f) = fs::OpenOptions::new() - .create(true) - .append(true) - .open(EVENTS_FILE) + let ts = now_rfc3339(); + let mut lines: Vec = fs::read_to_string(EVENTS_FILE) + .ok() + .map(|s| s.lines().map(|v| v.to_string()).collect()) + .unwrap_or_default(); + lines.push(format!("{ts} {event}")); + while lines.len() > EVENTS_MAX_LINES { + lines.remove(0); + } + + let mut content = lines.join("\n"); + if !content.is_empty() { + content.push('\n'); + } + while content.len() > EVENTS_MAX_BYTES { + if let Some(pos) = content.find('\n') { + content.drain(..=pos); + } else { + break; + } + } + + if let Ok(mut f) = fs::OpenOptions::new().create(true).write(true).truncate(true).open(EVENTS_FILE) { - let _ = writeln!(f, "{ts} {event}"); + let _ = f.write_all(content.as_bytes()); } } + +fn now_rfc3339() -> String { + chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true) +} diff --git a/traffic_core/src/firewall.rs b/traffic_core/src/firewall.rs index 5e56fc0..6e95251 100644 --- a/traffic_core/src/firewall.rs +++ b/traffic_core/src/firewall.rs @@ -20,16 +20,13 @@ struct FirewallActual { } pub fn apply_firewall(cfg: &Config) -> anyhow::Result<()> { - let transport = firewall_transport(cfg)?; - if !cfg.anti_torrent.enabled { - run_allow_notfound( - transport.as_ref(), - &["nft", "delete", "table", "inet", "awg2_harness"], - )?; + apply_firewall_disabled(cfg)?; return Ok(()); } + let transport = firewall_transport(cfg)?; + let actual = inspect_firewall(transport.as_ref(), cfg)?; if !firewall_has_drift(&actual) { return Ok(()); @@ -72,6 +69,30 @@ pub fn apply_firewall(cfg: &Config) -> anyhow::Result<()> { Ok(()) } +fn apply_firewall_disabled(cfg: &Config) -> anyhow::Result<()> { + if cfg.anti_torrent.apply_in_amnezia_netns { + match docker::discover_target(cfg) { + Ok(target) => { + let transport = build_transport(target.container_pid, &target.container_name); + run_allow_notfound( + transport.as_ref(), + &["nft", "delete", "table", "inet", "awg2_harness"], + )?; + } + Err(_) => { + // container is absent now: nothing to remove in its netns + } + } + } else { + let transport = build_host_transport(); + run_allow_notfound( + transport.as_ref(), + &["nft", "delete", "table", "inet", "awg2_harness"], + )?; + } + Ok(()) +} + fn inspect_firewall(transport: &dyn NsTransport, cfg: &Config) -> anyhow::Result { let table = match transport.run(&["nft", "list", "table", "inet", "awg2_harness"]) { Ok(out) => out, @@ -199,11 +220,14 @@ fn run_allow_notfound(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Res match transport.run(args) { Ok(_) => Ok(()), Err(e) => { - let msg = e.to_string(); - if msg.contains("No such file") - || msg.contains("No such") + let msg = e.to_string().to_ascii_lowercase(); + if msg.contains("no such file") + || msg.contains("no such") || msg.contains("not found") || msg.contains("not exist") + || msg.contains("no such table") + || msg.contains("no such device") + || msg.contains("cannot find") { return Ok(()); } diff --git a/traffic_core/src/main.rs b/traffic_core/src/main.rs index b155bec..baa9460 100644 --- a/traffic_core/src/main.rs +++ b/traffic_core/src/main.rs @@ -7,7 +7,6 @@ mod firewall; mod iface; mod lock; mod netns; -mod reboot; mod status; mod swap; mod traffic; @@ -36,35 +35,40 @@ fn main() -> anyhow::Result<()> { match cli.command { Command::ValidateConfig => { let cfg = Config::load(&config_path)?; - cfg.validate()?; + cfg.validate_config()?; println!("config valid: {}", config_path.display()); } + Command::ValidateRuntime => { + let cfg = Config::load(&config_path)?; + cfg.validate_runtime()?; + println!("runtime valid"); + } Command::ApplyLimits => { let cfg = Config::load(&config_path)?; - cfg.validate()?; + cfg.validate_all()?; traffic::reconcile_traffic(&cfg)?; println!("traffic limits reconciled"); } Command::ApplyFirewall => { let cfg = Config::load(&config_path)?; - cfg.validate()?; + cfg.validate_all()?; firewall::apply_firewall(&cfg)?; println!("firewall reconciled"); } Command::Status => { let cfg = Config::load(&config_path)?; - cfg.validate()?; + cfg.validate_config()?; status::print_status(&cfg)?; } Command::EnsureSwap => { let cfg = Config::load(&config_path)?; - cfg.validate()?; + cfg.validate_all()?; swap::ensure_swap(&cfg)?; println!("swap ensured"); } Command::Daemon => { let cfg = Config::load(&config_path)?; - cfg.validate()?; + cfg.validate_all()?; daemon::run(cfg)?; } } diff --git a/traffic_core/src/status.rs b/traffic_core/src/status.rs index c33b648..de8826f 100644 --- a/traffic_core/src/status.rs +++ b/traffic_core/src/status.rs @@ -24,6 +24,11 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> { Ok(t) => t, Err(e) => { println!(" container: not available ({e})"); + println!(" container_id: n/a"); + println!(" pid: n/a"); + println!(" netns: n/a"); + println!(" interface: n/a"); + print_degraded_sections(cfg)?; return Ok(()); } }; @@ -215,9 +220,112 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> { } } + print_reboot_section(); + Ok(()) } +fn print_degraded_sections(cfg: &Config) -> anyhow::Result<()> { + println!("Peers:"); + println!(" unavailable (container/netns not discovered)"); + + println!("WireGuard clients:"); + println!(" unavailable (container/netns not discovered)"); + + println!("Traffic desired:"); + println!(" peer_limit: {}", cfg.traffic.peer_limit); + println!(" root_rate: {}", cfg.traffic.root_rate); + println!(" ifb_name: {}", cfg.traffic.ifb_name); + println!(" class_start_id: {}", cfg.traffic.class_start_id); + + println!("Traffic actual:"); + println!(" drift: unknown (container/netns not discovered)"); + + println!("Anti-torrent:"); + println!(" enabled: {}", cfg.anti_torrent.enabled); + println!( + " max_connections_per_peer: {}", + cfg.anti_torrent.max_connections_per_peer + ); + if cfg.anti_torrent.enabled { + 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})"), + } + } + + println!("Swap:"); + println!(" enabled: {}", cfg.swap.enabled); + println!(" path: {}", cfg.swap.path); + println!(" size_mb: {}", cfg.swap.size_mb); + println!(" priority: {}", cfg.swap.priority); + + println!("Conntrack:"); + let ct_count = Command::new("sysctl") + .args(["-n", "net.netfilter.nf_conntrack_count"]) + .output() + .ok() + .map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string()) + .unwrap_or_else(|| "n/a".to_string()); + let ct_max = Command::new("sysctl") + .args(["-n", "net.netfilter.nf_conntrack_max"]) + .output() + .ok() + .map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string()) + .unwrap_or_else(|| "n/a".to_string()); + println!(" nf_conntrack_count: {}", ct_count); + println!(" nf_conntrack_max: {}", ct_max); + + println!("Last actions:"); + let last = fs::read_to_string("/run/awg2-harness/last_action.state").unwrap_or_default(); + if last.trim().is_empty() { + println!(" no persisted state"); + } else { + for line in last.lines() { + println!(" {}", line); + } + } + + println!("Recent events:"); + let events = fs::read_to_string("/run/awg2-harness/events.log").unwrap_or_default(); + let lines: Vec<&str> = events.lines().collect(); + if lines.is_empty() { + println!(" no recent events"); + } else { + for l in lines.iter().rev().take(10).rev() { + println!(" {l}"); + } + } + + print_reboot_section(); + Ok(()) +} + +fn print_reboot_section() { + println!("Reboot scheduler:"); + let svc = Command::new("systemctl") + .args(["is-active", "awg2-reboot-scheduler"]) + .output(); + match svc { + Ok(out) if out.status.success() => println!(" service: active"), + Ok(_) => println!(" service: inactive"), + Err(_) => println!(" service: unknown"), + } + + let plan = fs::read_to_string("/run/awg2-harness/reboot.plan").unwrap_or_default(); + if plan.trim().is_empty() { + println!(" planned: n/a"); + } else { + for line in plan.lines() { + println!(" {}", line); + } + } +} + fn handshake_is_online(epoch: Option, threshold_secs: u64) -> &'static str { if let Some(ts) = epoch { let now = SystemTime::now() diff --git a/traffic_core/src/traffic.rs b/traffic_core/src/traffic.rs index f52b020..4086064 100644 --- a/traffic_core/src/traffic.rs +++ b/traffic_core/src/traffic.rs @@ -972,13 +972,40 @@ fn parse_class_rate_ceil_map(raw: &str, prefix: &str) -> BTreeMap bool { - normalize_tc_rate(&v.rate) == limit && normalize_tc_rate(&v.ceil) == limit + let Ok(limit_bps) = parse_tc_rate_bps(limit) else { + return false; + }; + let Ok(rate_bps) = parse_tc_rate_bps(&v.rate) else { + return false; + }; + let Ok(ceil_bps) = parse_tc_rate_bps(&v.ceil) else { + return false; + }; + rate_bps == limit_bps && ceil_bps == limit_bps } fn normalize_tc_rate(v: &str) -> String { v.trim().to_ascii_lowercase().replace(' ', "") } +fn parse_tc_rate_bps(v: &str) -> anyhow::Result { + let n = normalize_tc_rate(v); + let (num, mul) = if let Some(x) = n.strip_suffix("gbit") { + (x, 1_000_000_000u64) + } else if let Some(x) = n.strip_suffix("mbit") { + (x, 1_000_000u64) + } else if let Some(x) = n.strip_suffix("kbit") { + (x, 1_000u64) + } else if let Some(x) = n.strip_suffix("bit") { + (x, 1u64) + } else { + anyhow::bail!("unknown tc rate unit: {v}") + }; + let base: u64 = num.parse().map_err(|_| anyhow::anyhow!("invalid tc rate: {v}"))?; + base.checked_mul(mul) + .ok_or_else(|| anyhow::anyhow!("tc rate overflow: {v}")) +} + #[cfg(test)] mod tests { use super::*; @@ -1094,4 +1121,14 @@ mod tests { assert!(rate_ceil_matches_limit(&rc, "70mbit")); assert!(!rate_ceil_matches_limit(&rc, "50mbit")); } + + #[test] + fn tc_rate_numeric_equivalence() { + let rc = ClassRateCeil { + rate: "1Gbit".to_string(), + ceil: "1000Mbit".to_string(), + }; + assert!(rate_ceil_matches_limit(&rc, "1000mbit")); + assert!(rate_ceil_matches_limit(&rc, "1gbit")); + } } diff --git a/traffic_core/src/wg.rs b/traffic_core/src/wg.rs index 0563927..4ebc16d 100644 --- a/traffic_core/src/wg.rs +++ b/traffic_core/src/wg.rs @@ -36,11 +36,10 @@ pub fn discover_peers_with_transport( transport: &dyn NsTransport, peer_cidr: &str, ) -> anyhow::Result> { - let primary = transport - .run(&["wg", "show", "all", "allowed-ips"]) - .map_err(|e| anyhow::anyhow!("wg show all allowed-ips failed: {e}"))?; - - let mut peers = parse_allowed_ips(&primary, peer_cidr)?; + let mut peers = match transport.run(&["wg", "show", "all", "allowed-ips"]) { + Ok(primary) => parse_allowed_ips(&primary, peer_cidr)?, + Err(_) => Vec::new(), + }; if peers.is_empty() { let fallback = transport .run(&["wg", "show"]) @@ -261,16 +260,15 @@ mod tests { } #[test] - fn command_failure_is_propagated() { + fn primary_failure_falls_back_to_wg_show() { let transport = TestTransport::new(vec![( "wg show all allowed-ips", Err(anyhow::anyhow!("wg broken")), - )]); + ), ("wg show", Ok("allowed ips: 10.8.1.4/32"))]); - let err = discover_peers_with_transport(&transport, "10.8.1.0/24").err(); - assert!(err.is_some()); - let msg = err.expect("err").to_string(); - assert!(msg.contains("wg show all allowed-ips failed")); + let peers = discover_peers_with_transport(&transport, "10.8.1.0/24").expect("peers"); + assert_eq!(peers.len(), 1); + assert_eq!(peers[0].ipv4.to_string(), "10.8.1.4"); } #[test]