diff --git a/tools/build.sh b/tools/build.sh new file mode 100644 index 0000000..b3cb252 --- /dev/null +++ b/tools/build.sh @@ -0,0 +1,77 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +CRATE_MANIFEST="$ROOT_DIR/traffic_core/Cargo.toml" +TARGET_BIN="$ROOT_DIR/traffic_core/target/release/awg2-harness" + +AUTO_INSTALL=false +SKIP_FMT=false + +for arg in "$@"; do + case "$arg" in + --install) AUTO_INSTALL=true ;; + --skip-fmt) SKIP_FMT=true ;; + *) echo "Unknown arg: $arg"; exit 1 ;; + esac +done + +require_cmd() { + command -v "$1" >/dev/null 2>&1 || { echo "Missing command: $1"; exit 1; } +} + +if [[ -f /etc/os-release ]]; then + # shellcheck disable=SC1091 + source /etc/os-release + case "${ID:-}" in + debian|ubuntu) ;; + *) echo "Unsupported OS: ${ID:-unknown}. Only Debian/Ubuntu are supported."; exit 1 ;; + esac +fi + +ARCH="$(uname -m)" +if [[ "$ARCH" != "x86_64" && "$ARCH" != "amd64" ]]; then + echo "Unsupported arch: $ARCH. Only amd64/x86_64 is supported." + exit 1 +fi + +if ! command -v cargo >/dev/null 2>&1; then + echo "cargo not found. Install rustup first: https://rustup.rs" + exit 1 +fi + +for cmd in gcc pkg-config; do + require_cmd "$cmd" +done + +if [[ "$SKIP_FMT" != true ]]; then + cargo fmt --manifest-path "$CRATE_MANIFEST" --all --check +fi + +cargo clippy --manifest-path "$CRATE_MANIFEST" --all-targets -- -D warnings +cargo test --manifest-path "$CRATE_MANIFEST" --release +cargo build --manifest-path "$CRATE_MANIFEST" --release + +require_cmd file +require_cmd ldd + +file "$TARGET_BIN" +ldd "$TARGET_BIN" || true +"$TARGET_BIN" --help >/dev/null + +sha256sum "$TARGET_BIN" +ls -lh "$TARGET_BIN" + +if [[ "$AUTO_INSTALL" == true ]]; then + require_cmd install + 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 + 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/Cargo.lock b/traffic_core/Cargo.lock index 14ba9be..ad32840 100644 --- a/traffic_core/Cargo.lock +++ b/traffic_core/Cargo.lock @@ -2,6 +2,15 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "aho-corasick" +version = "1.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301" +dependencies = [ + "memchr", +] + [[package]] name = "android_system_properties" version = "0.1.5" @@ -84,6 +93,8 @@ dependencies = [ "serde", "thiserror", "toml", + "tracing", + "tracing-subscriber", ] [[package]] @@ -284,6 +295,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "lazy_static" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" + [[package]] name = "libc" version = "0.2.186" @@ -296,12 +313,30 @@ version = "0.4.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + [[package]] name = "memchr" version = "2.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -386,6 +421,23 @@ dependencies = [ "getrandom", ] +[[package]] +name = "regex-automata" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e1dd4122fc1595e8162618945476892eefca7b88c52820e74af6262213cae8f" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + +[[package]] +name = "regex-syntax" +version = "0.8.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a" + [[package]] name = "rustversion" version = "1.0.22" @@ -431,6 +483,15 @@ dependencies = [ "serde", ] +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "shlex" version = "1.3.0" @@ -443,6 +504,12 @@ version = "0.4.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" +[[package]] +name = "smallvec" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" + [[package]] name = "strsim" version = "0.11.1" @@ -480,6 +547,15 @@ dependencies = [ "syn", ] +[[package]] +name = "thread_local" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f60246a4944f24f6e018aa17cdeffb7818b76356965d03b07d6a9886e8962185" +dependencies = [ + "cfg-if", +] + [[package]] name = "toml" version = "0.8.23" @@ -521,6 +597,67 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801" +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "pin-project-lite", + "tracing-attributes", + "tracing-core", +] + +[[package]] +name = "tracing-attributes" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", +] + [[package]] name = "unicode-ident" version = "1.0.24" @@ -533,6 +670,12 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" diff --git a/traffic_core/Cargo.toml b/traffic_core/Cargo.toml index 94a5961..54de44e 100644 --- a/traffic_core/Cargo.toml +++ b/traffic_core/Cargo.toml @@ -11,3 +11,5 @@ rand = "0.8" serde = { version = "1.0", features = ["derive"] } thiserror = "2.0" toml = "0.8" +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/traffic_core/src/config.rs b/traffic_core/src/config.rs index 9042a14..e10207d 100644 --- a/traffic_core/src/config.rs +++ b/traffic_core/src/config.rs @@ -1,4 +1,4 @@ -use std::{fs, net::Ipv4Addr, path::Path}; +use std::{fs, net::Ipv4Addr, path::Path, process::Command}; use serde::Deserialize; @@ -72,10 +72,38 @@ impl Config { if self.swap.swappiness > 100 { anyhow::bail!("swap.swappiness must be <= 100"); } + if self.traffic.container_ready_timeout_secs == 0 { + anyhow::bail!("traffic.container_ready_timeout_secs must be > 0"); + } + if self.traffic.reconcile_interval_secs == 0 { + anyhow::bail!("traffic.reconcile_interval_secs 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"); + } parse_cidr_v4(&self.amnezia.peer_ipv4_cidr)?; parse_cidr_v4(&self.anti_torrent.peer_ipv4_cidr)?; + self.check_host_tools()?; Ok(()) } + + fn check_host_tools(&self) -> anyhow::Result<()> { + let mut required = vec!["tc", "ip", "wg", "nsenter", "docker", "modprobe"]; + if self.anti_torrent.enabled { + required.push("nft"); + } + + for tool in required { + if !tool_exists(tool) { + anyhow::bail!("required host tool not found in PATH: {tool}"); + } + } + Ok(()) + } +} + +fn tool_exists(tool: &str) -> bool { + Command::new(tool).arg("--version").output().is_ok() } pub fn parse_cidr_v4(cidr: &str) -> anyhow::Result<(Ipv4Addr, u8)> { @@ -106,3 +134,22 @@ pub fn ip_in_cidr(ip: Ipv4Addr, cidr: &str) -> anyhow::Result { Ok((ip_u & mask) == (base_u & mask)) } +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_cidr_invalid_prefix() { + let err = parse_cidr_v4("10.8.1.0/33").err(); + assert!(err.is_some()); + } + + #[test] + fn ip_membership() { + let ok = ip_in_cidr("10.8.1.10".parse().expect("ip"), "10.8.1.0/24").expect("cidr"); + assert!(ok); + let no = ip_in_cidr("10.9.1.10".parse().expect("ip"), "10.8.1.0/24").expect("cidr"); + assert!(!no); + } +} + diff --git a/traffic_core/src/daemon.rs b/traffic_core/src/daemon.rs index 01a2214..4f33f46 100644 --- a/traffic_core/src/daemon.rs +++ b/traffic_core/src/daemon.rs @@ -1,31 +1,80 @@ -use std::{thread, time::Duration}; +use std::{ + fs, + sync::{Arc, Mutex}, + thread, + time::{Duration, Instant}, +}; -use crate::{config::Config, firewall, reboot, traffic}; +use crate::{ + config::Config, + docker, + firewall, + iface, + netns::build_transport, + reboot, + traffic, +}; +use tracing::{error, info}; + +#[derive(Debug, Clone, Default)] +struct LastActionState { + last_ok: Option, + last_error: Option, +} + +const LAST_STATE_FILE: &str = "/run/awg2-harness/last_action.state"; pub fn run(cfg: Config) -> anyhow::Result<()> { - println!("starting daemon"); + info!("starting daemon"); + let state = Arc::new(Mutex::new(LastActionState::default())); - let _ = traffic::reconcile_traffic(&cfg); - let _ = firewall::apply_firewall(&cfg); + if let Err(e) = traffic::reconcile_traffic(&cfg) { + set_err(&state, format!("startup traffic reconcile: {e}")); + } else { + set_ok(&state, "startup traffic reconcile ok"); + } + if let Err(e) = firewall::apply_firewall(&cfg) { + set_err(&state, format!("startup firewall reconcile: {e}")); + } else { + set_ok(&state, "startup firewall reconcile 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 || { - let _ = docker_events_loop(events_cfg); + 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)); + } + } }); loop { if let Err(e) = traffic::reconcile_traffic(&cfg) { - eprintln!("traffic reconcile error: {e}"); + error!(error = %e, "traffic reconcile error"); + set_err(&state, format!("periodic traffic reconcile: {e}")); + } else { + set_ok(&state, "periodic traffic reconcile ok"); } if let Err(e) = firewall::apply_firewall(&cfg) { - eprintln!("firewall reconcile error: {e}"); + error!(error = %e, "firewall reconcile error"); + set_err(&state, format!("periodic firewall reconcile: {e}")); + } else { + set_ok(&state, "periodic firewall reconcile ok"); } thread::sleep(Duration::from_secs(cfg.traffic.reconcile_interval_secs.max(5))); } } -fn docker_events_loop(cfg: Config) -> anyhow::Result<()> { +fn docker_events_loop(cfg: &Config, state: &Arc>) -> anyhow::Result<()> { use std::io::{BufRead, BufReader}; use std::process::{Command, Stdio}; @@ -42,11 +91,63 @@ fn docker_events_loop(cfg: Config) -> anyhow::Result<()> { for line in reader.lines() { let name = line?.trim().to_string(); - if cfg.amnezia.container_names.iter().any(|n| n == &name) { - let _ = traffic::reconcile_traffic(&cfg); - let _ = firewall::apply_firewall(&cfg); + if cfg.amnezia.container_names.iter().any(|n| n == &name) + && wait_for_interface(cfg, &name, cfg.traffic.container_ready_timeout_secs) + { + if let Err(e) = traffic::reconcile_traffic(cfg) { + set_err(state, format!("event traffic reconcile: {e}")); + } else { + set_ok(state, "event traffic reconcile ok"); + } + + if let Err(e) = firewall::apply_firewall(cfg) { + set_err(state, format!("event firewall reconcile: {e}")); + } else { + set_ok(state, "event firewall reconcile ok"); + } } } Ok(()) } +fn wait_for_interface(cfg: &Config, container_name: &str, timeout_secs: u64) -> bool { + let start = Instant::now(); + while start.elapsed() < Duration::from_secs(timeout_secs.max(1)) { + if let Ok(target) = docker::discover_target(cfg) { + if target.container_name == container_name { + let transport = build_transport(target.container_pid, &target.container_name); + if iface::detect_interface_with_transport(cfg, transport.as_ref()).is_ok() { + return true; + } + } + } + + thread::sleep(Duration::from_secs(2)); + } + false +} + +fn set_ok(state: &Arc>, msg: &str) { + if let Ok(mut s) = state.lock() { + s.last_ok = Some(msg.to_string()); + persist_state(&s); + } +} + +fn set_err(state: &Arc>, msg: String) { + if let Ok(mut s) = state.lock() { + s.last_error = Some(msg); + persist_state(&s); + } +} + +fn persist_state(state: &LastActionState) { + let _ = fs::create_dir_all("/run/awg2-harness"); + let content = format!( + "last_ok={}\nlast_error={}\n", + state.last_ok.clone().unwrap_or_default(), + state.last_error.clone().unwrap_or_default() + ); + let _ = fs::write(LAST_STATE_FILE, content); +} + diff --git a/traffic_core/src/docker.rs b/traffic_core/src/docker.rs index 308a8f3..69f5210 100644 --- a/traffic_core/src/docker.rs +++ b/traffic_core/src/docker.rs @@ -25,7 +25,7 @@ pub fn discover_target(cfg: &Config) -> anyhow::Result { ])?; let container_pid = pid_str.trim().parse::()?; let netns_path = format!("/proc/{container_pid}/ns/net"); - let interface_name = crate::iface::detect_interface(cfg, &container_name)?; + let interface_name = crate::iface::detect_interface(cfg, container_pid, &container_name)?; Ok(RuntimeTarget { container_name, diff --git a/traffic_core/src/firewall.rs b/traffic_core/src/firewall.rs index 97cbecf..03c46e0 100644 --- a/traffic_core/src/firewall.rs +++ b/traffic_core/src/firewall.rs @@ -1,29 +1,62 @@ -use crate::{config::Config, docker}; +use crate::{ + config::Config, + docker, + netns::{build_transport, NsTransport}, +}; + +#[derive(Debug, Clone)] +struct FirewallActual { + table_exists: bool, + chain_exists: bool, + rule_matches: bool, +} pub fn apply_firewall(cfg: &Config) -> anyhow::Result<()> { let target = docker::discover_target(cfg)?; + let transport = build_transport(target.container_pid, &target.container_name); if !cfg.anti_torrent.enabled { - let _ = docker::run_capture(&[ - "docker", - "exec", - &target.container_name, - "sh", - "-c", - "nft delete table inet awg2_harness 2>/dev/null || true", - ])?; + let _ = transport.run(&["sh", "-c", "nft delete table inet awg2_harness 2>/dev/null || true"])?; + return Ok(()); + } + + let actual = inspect_firewall(transport.as_ref(), cfg)?; + if actual.table_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 'flush chain inet awg2_harness forward_antitorrent'; \ - 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'", + 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 ); - let _ = docker::run_capture(&["docker", "exec", &target.container_name, "sh", "-c", &rule])?; + apply_rule(transport.as_ref(), &rule)?; + 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"]) + .unwrap_or_default(); + + let table_exists = !table.trim().is_empty(); + 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 rule_matches = table.contains(&needle_cidr) && table.contains(&needle_max); + + Ok(FirewallActual { + table_exists, + chain_exists, + rule_matches, + }) +} + +fn apply_rule(transport: &dyn NsTransport, rule: &str) -> anyhow::Result<()> { + let _ = transport.run(&["sh", "-c", rule])?; Ok(()) } diff --git a/traffic_core/src/iface.rs b/traffic_core/src/iface.rs index 1c79a54..75aa950 100644 --- a/traffic_core/src/iface.rs +++ b/traffic_core/src/iface.rs @@ -1,23 +1,34 @@ -use crate::{config::Config, docker::run_capture}; +use crate::{ + config::Config, + netns::{DockerExecTransport, NsTransport, NsenterTransport, nsenter_available}, +}; -pub fn detect_interface(cfg: &Config, container: &str) -> anyhow::Result { +pub fn detect_interface(cfg: &Config, pid: u32, container: &str) -> anyhow::Result { + let boxed: Box = if nsenter_available() { + Box::new(NsenterTransport { pid }) + } else { + Box::new(DockerExecTransport { + container: container.to_string(), + }) + }; + + detect_interface_with_transport(cfg, boxed.as_ref()) +} + +pub fn detect_interface_with_transport( + cfg: &Config, + transport: &dyn NsTransport, +) -> anyhow::Result { for ifname in &cfg.amnezia.interface_candidates { - let ok = std::process::Command::new("docker") - .args(["exec", container, "ip", "link", "show", ifname]) - .output()?; - if ok.status.success() { + if transport + .run(&["ip", "link", "show", ifname]) + .is_ok() + { return Ok(ifname.clone()); } } - let links = run_capture(&[ - "docker", - "exec", - container, - "sh", - "-c", - "ip -o link show | awk -F': ' '{print $2}'", - ])?; + let links = transport.run(&["sh", "-c", "ip -o link show | awk -F': ' '{print $2}'"])?; for raw in links.lines() { let name = raw.trim(); diff --git a/traffic_core/src/lock.rs b/traffic_core/src/lock.rs index 7d7cb85..495779f 100644 --- a/traffic_core/src/lock.rs +++ b/traffic_core/src/lock.rs @@ -11,7 +11,12 @@ impl FileLock { fs::create_dir_all(parent)?; } if p.exists() { - anyhow::bail!("lock already exists: {path}"); + let stale = is_stale_lock(p)?; + if stale { + let _ = fs::remove_file(p); + } else { + anyhow::bail!("lock already exists: {path}"); + } } fs::write(p, std::process::id().to_string())?; Ok(Self { @@ -26,3 +31,14 @@ impl Drop for FileLock { } } +fn is_stale_lock(path: &Path) -> anyhow::Result { + let pid_raw = fs::read_to_string(path)?; + let pid = match pid_raw.trim().parse::() { + Ok(v) => v, + Err(_) => return Ok(true), + }; + + let proc_path = format!("/proc/{pid}"); + Ok(!Path::new(&proc_path).exists()) +} + diff --git a/traffic_core/src/main.rs b/traffic_core/src/main.rs index 436fe9f..63ce2fc 100644 --- a/traffic_core/src/main.rs +++ b/traffic_core/src/main.rs @@ -16,11 +16,16 @@ mod wg; use std::path::PathBuf; use clap::Parser; +use tracing_subscriber::EnvFilter; use crate::cli::{Cli, Command}; 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"))) + .init(); + let cli = Cli::parse(); let config_path = cli .config diff --git a/traffic_core/src/netns.rs b/traffic_core/src/netns.rs index a406540..9ceaaf8 100644 --- a/traffic_core/src/netns.rs +++ b/traffic_core/src/netns.rs @@ -1,16 +1,58 @@ use std::process::Command; -pub fn nsenter_run_capture(pid: u32, args: &[&str]) -> anyhow::Result { - let mut cmd = Command::new("nsenter"); - cmd.args(["-t", &pid.to_string(), "-n"]); - cmd.args(args); - let out = cmd.output()?; - if !out.status.success() { - anyhow::bail!( - "nsenter failed: {}", - String::from_utf8_lossy(&out.stderr) - ); - } - Ok(String::from_utf8_lossy(&out.stdout).to_string()) +pub trait NsTransport: Send + Sync { + fn run(&self, args: &[&str]) -> anyhow::Result; +} + +#[derive(Debug, Clone)] +pub struct NsenterTransport { + pub pid: u32, +} + +#[derive(Debug, Clone)] +pub struct DockerExecTransport { + pub container: String, +} + +impl NsTransport for NsenterTransport { + fn run(&self, args: &[&str]) -> anyhow::Result { + let mut cmd = Command::new("nsenter"); + cmd.args(["-t", &self.pid.to_string(), "-n", "--"]); + cmd.args(args); + let out = cmd.output()?; + if !out.status.success() { + anyhow::bail!("nsenter failed: {}", String::from_utf8_lossy(&out.stderr)); + } + Ok(String::from_utf8_lossy(&out.stdout).to_string()) + } +} + +impl NsTransport for DockerExecTransport { + fn run(&self, args: &[&str]) -> anyhow::Result { + let mut cmd = Command::new("docker"); + cmd.arg("exec").arg(&self.container).args(args); + let out = cmd.output()?; + if !out.status.success() { + anyhow::bail!( + "docker exec transport failed: {}", + String::from_utf8_lossy(&out.stderr) + ); + } + Ok(String::from_utf8_lossy(&out.stdout).to_string()) + } +} + +pub fn nsenter_available() -> bool { + Command::new("nsenter").arg("--version").output().is_ok() +} + +pub fn build_transport(pid: u32, container: &str) -> Box { + if nsenter_available() { + Box::new(NsenterTransport { pid }) + } else { + Box::new(DockerExecTransport { + container: container.to_string(), + }) + } } diff --git a/traffic_core/src/reboot.rs b/traffic_core/src/reboot.rs index 07c33e8..e9c6930 100644 --- a/traffic_core/src/reboot.rs +++ b/traffic_core/src/reboot.rs @@ -1,5 +1,6 @@ -use chrono::{Duration, Timelike, Utc}; +use chrono::{Datelike, Duration, NaiveDateTime, NaiveTime, Utc}; use rand::Rng; +use std::{process::Command, thread, time::Duration as StdDuration}; use crate::config::Config; @@ -8,37 +9,79 @@ pub fn maybe_schedule_reboot(cfg: &Config) -> anyhow::Result<()> { return Ok(()); } - let start = parse_hhmm(&cfg.reboot.window_start_utc)?; - let end = parse_hhmm(&cfg.reboot.window_end_utc)?; - let now = Utc::now(); - let date = now.date_naive(); + let planned = next_reboot_time(&cfg.reboot.window_start_utc, &cfg.reboot.window_end_utc)?; + println!("reboot planned at UTC {}", planned); + Ok(()) +} - let start_dt = date - .and_hms_opt(start.0, start.1, 0) - .ok_or_else(|| anyhow::anyhow!("invalid reboot start time"))?; - let end_dt = date - .and_hms_opt(end.0, end.1, 0) - .ok_or_else(|| anyhow::anyhow!("invalid reboot end time"))?; +pub fn reboot_scheduler_loop(cfg: Config) -> anyhow::Result<()> { + if !cfg.reboot.enabled { + return Ok(()); + } - let span = if end_dt > start_dt { - end_dt - start_dt + loop { + let planned = next_reboot_time(&cfg.reboot.window_start_utc, &cfg.reboot.window_end_utc)?; + println!("reboot scheduler: next reboot at UTC {}", planned); + + loop { + let now = Utc::now().naive_utc(); + if now >= planned { + break; + } + thread::sleep(StdDuration::from_secs(30)); + } + + let out = Command::new("systemctl") + .args(["--no-block", "reboot"]) + .output()?; + if out.status.success() { + println!("reboot request accepted by systemctl"); + return Ok(()); + } + + eprintln!( + "reboot request failed: {}", + String::from_utf8_lossy(&out.stderr) + ); + thread::sleep(StdDuration::from_secs(60)); + } +} + +fn next_reboot_time(start_hhmm: &str, end_hhmm: &str) -> anyhow::Result { + let start = parse_hhmm(start_hhmm)?; + let end = parse_hhmm(end_hhmm)?; + + let now = Utc::now().naive_utc(); + let today = now.date(); + let mut base_day = today; + + let now_t = NaiveTime::from_hms_opt(now.hour(), now.minute(), now.second()) + .ok_or_else(|| anyhow::anyhow!("bad current time"))?; + if now_t >= start { + base_day = today + .succ_opt() + .ok_or_else(|| anyhow::anyhow!("date overflow"))?; + } + + let span = if end > start { + end.signed_duration_since(start) } else { Duration::hours(1) }; let secs = rand::thread_rng().gen_range(0..span.num_seconds().max(1)); - let planned = start_dt + Duration::seconds(secs); - println!("reboot planned at UTC {}", planned); - Ok(()) + let base_dt = base_day + .and_hms_opt(start.hour(), start.minute(), 0) + .ok_or_else(|| anyhow::anyhow!("invalid base datetime"))?; + Ok(base_dt + Duration::seconds(secs)) } - -fn parse_hhmm(v: &str) -> anyhow::Result<(u32, u32)> { +fn parse_hhmm(v: &str) -> 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()?; if h > 23 || m > 59 { anyhow::bail!("bad time") } - Ok((h, m)) + 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 ae50f10..a20e99f 100644 --- a/traffic_core/src/status.rs +++ b/traffic_core/src/status.rs @@ -1,7 +1,18 @@ -use crate::{config::Config, docker, wg}; +use crate::{config::Config, docker, traffic, wg}; + +use std::{fs, process::Command}; pub fn print_status(cfg: &Config) -> anyhow::Result<()> { println!("System:"); + let svc = Command::new("systemctl") + .args(["is-active", "awg2-harness"]) + .output(); + match svc { + Ok(out) if out.status.success() => println!(" service: active"), + Ok(_) => println!(" service: inactive"), + Err(_) => println!(" service: unknown"), + } + let target = match docker::discover_target(cfg) { Ok(t) => t, Err(e) => { @@ -16,7 +27,11 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> { println!(" netns: {}", target.netns_path); println!(" interface: {}", target.interface_name); - let peers = wg::discover_peers(&target.container_name, &cfg.amnezia.peer_ipv4_cidr)?; + let peers = wg::discover_peers( + target.container_pid, + &target.container_name, + &cfg.amnezia.peer_ipv4_cidr, + )?; println!("Peers:"); println!(" total: {}", peers.len()); for p in peers { @@ -29,12 +44,70 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> { println!(" ifb_name: {}", cfg.traffic.ifb_name); println!(" class_start_id: {}", cfg.traffic.class_start_id); + println!("Traffic actual:"); + match traffic::inspect_drift_summary(cfg) { + Ok(None) => println!(" drift: no"), + Ok(Some(reason)) => { + println!(" drift: yes"); + println!(" reason: {}", reason); + } + Err(e) => println!(" drift: unknown ({e})"), + } + 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 { + 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"), + } + } + + 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!("Conntrack:"); + 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); + } + } + Ok(()) } diff --git a/traffic_core/src/swap.rs b/traffic_core/src/swap.rs index 5b20fba..6e3f456 100644 --- a/traffic_core/src/swap.rs +++ b/traffic_core/src/swap.rs @@ -8,7 +8,17 @@ pub fn ensure_swap(cfg: &Config) -> anyhow::Result<()> { } let path = &cfg.swap.path; - if !std::path::Path::new(path).exists() { + let expected_size = cfg.swap.size_mb * 1024 * 1024; + if std::path::Path::new(path).exists() { + let actual_size = fs::metadata(path)?.len(); + if actual_size != expected_size { + anyhow::bail!( + "existing swapfile has wrong size: {} bytes, expected {} bytes; manual intervention required", + actual_size, + expected_size + ); + } + } else { run("fallocate", &["-l", &format!("{}M", cfg.swap.size_mb), path])?; run("chmod", &["600", path])?; run("mkswap", &[path])?; @@ -22,6 +32,7 @@ pub fn ensure_swap(cfg: &Config) -> anyhow::Result<()> { ensure_fstab(path, cfg.swap.priority)?; run("sysctl", &["-w", &format!("vm.swappiness={}", cfg.swap.swappiness)])?; + ensure_swappiness_persist(cfg.swap.swappiness)?; Ok(()) } @@ -49,3 +60,13 @@ fn run(program: &str, args: &[&str]) -> anyhow::Result<()> { Ok(()) } +fn ensure_swappiness_persist(swappiness: u8) -> anyhow::Result<()> { + let conf_path = "/etc/sysctl.d/99-awg2-harness.conf"; + let expected = format!("vm.swappiness = {}\n", swappiness); + let current = fs::read_to_string(conf_path).unwrap_or_default(); + if current != expected { + fs::write(conf_path, expected)?; + } + Ok(()) +} + diff --git a/traffic_core/src/traffic.rs b/traffic_core/src/traffic.rs index 0f6aeae..a2cec8f 100644 --- a/traffic_core/src/traffic.rs +++ b/traffic_core/src/traffic.rs @@ -1,4 +1,49 @@ -use crate::{config::Config, docker, lock::FileLock, wg}; +use crate::{ + config::Config, + docker, + lock::FileLock, + netns::{NsTransport, build_transport}, + wg, +}; + +use std::collections::BTreeSet; + +#[derive(Debug, Clone)] +struct TrafficDrift { + has_drift: bool, + reason: String, +} + +pub fn inspect_drift_summary(cfg: &Config) -> anyhow::Result> { + if !cfg.traffic.enabled { + return Ok(None); + } + + let target = docker::discover_target(cfg)?; + let transport = build_transport(target.container_pid, &target.container_name); + let peers = wg::discover_peers( + target.container_pid, + &target.container_name, + &cfg.amnezia.peer_ipv4_cidr, + )?; + + if peers.is_empty() { + return Ok(Some("no peers discovered".to_string())); + } + + let drift = inspect_and_diff( + transport.as_ref(), + &target.interface_name, + &cfg.traffic.ifb_name, + &peers, + )?; + + if drift.has_drift { + Ok(Some(drift.reason)) + } else { + Ok(None) + } +} pub fn reconcile_traffic(cfg: &Config) -> anyhow::Result<()> { if !cfg.traffic.enabled { @@ -7,34 +52,142 @@ pub fn reconcile_traffic(cfg: &Config) -> anyhow::Result<()> { let _lock = FileLock::acquire("/run/awg2-harness/traffic.lock")?; let target = docker::discover_target(cfg)?; - let peers = wg::discover_peers(&target.container_name, &cfg.amnezia.peer_ipv4_cidr)?; + let transport = build_transport(target.container_pid, &target.container_name); + let peers = wg::discover_peers( + target.container_pid, + &target.container_name, + &cfg.amnezia.peer_ipv4_cidr, + )?; if peers.is_empty() { println!("no peers found, skip tc apply"); return Ok(()); } + let drift = inspect_and_diff( + transport.as_ref(), + &target.interface_name, + &cfg.traffic.ifb_name, + &peers, + )?; + + if !drift.has_drift { + println!("traffic rules up-to-date, skip apply"); + return Ok(()); + } + + println!("traffic drift detected: {}", drift.reason); + if cfg.traffic.full_rebuild_on_drift { - teardown_owned_topology(&target.container_name, &target.interface_name, &cfg.traffic.ifb_name)?; - apply_full_topology(cfg, &target.container_name, &target.interface_name, &cfg.traffic.ifb_name, &peers)?; + teardown_owned_topology(transport.as_ref(), &target.interface_name, &cfg.traffic.ifb_name)?; + apply_full_topology( + cfg, + transport.as_ref(), + &target.interface_name, + &cfg.traffic.ifb_name, + &peers, + )?; } Ok(()) } -fn teardown_owned_topology(container: &str, iface: &str, ifb: &str) -> anyhow::Result<()> { +fn inspect_and_diff( + transport: &dyn NsTransport, + iface: &str, + ifb: &str, + peers: &[wg::Peer], +) -> anyhow::Result { + let root_qdisc = transport + .run(&["sh", "-c", &format!("tc qdisc show dev {iface} 2>/dev/null")]) + .unwrap_or_default(); + if !root_qdisc.contains("htb") { + return Ok(TrafficDrift { + has_drift: true, + reason: "missing egress root htb qdisc".to_string(), + }); + } + + let ingress_qdisc = transport + .run(&["sh", "-c", &format!("tc qdisc show dev {iface} 2>/dev/null")]) + .unwrap_or_default(); + if !ingress_qdisc.contains("ingress") { + return Ok(TrafficDrift { + has_drift: true, + reason: "missing ingress qdisc".to_string(), + }); + } + + let ifb_link = transport + .run(&["sh", "-c", &format!("ip link show {ifb} 2>/dev/null")]) + .unwrap_or_default(); + if ifb_link.trim().is_empty() { + return Ok(TrafficDrift { + has_drift: true, + reason: "missing ifb interface".to_string(), + }); + } + + let filters = transport + .run(&[ + "sh", + "-c", + &format!("tc filter show dev {iface} parent 1: 2>/dev/null"), + ]) + .unwrap_or_default(); + + let actual_ips = extract_filter_ips(&filters); + let expected_ips: BTreeSet = peers.iter().map(|p| p.ipv4.to_string()).collect(); + + if actual_ips != expected_ips { + return Ok(TrafficDrift { + has_drift: true, + reason: format!( + "peer filter mismatch (actual={}, expected={})", + actual_ips.len(), + expected_ips.len() + ), + }); + } + + 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; + } + } + } + set +} + +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 _ = docker::run_capture(&["docker", "exec", container, "sh", "-c", &cmd])?; + let _ = transport.run(&["sh", "-c", &cmd])?; Ok(()) } fn apply_full_topology( cfg: &Config, - container: &str, + transport: &dyn NsTransport, iface: &str, ifb: &str, peers: &[wg::Peer], @@ -51,7 +204,7 @@ fn apply_full_topology( tc class add dev {ifb} parent 2: classid 2:1 htb rate {root} ceil {root}", root = cfg.traffic.root_rate ); - let _ = docker::run_capture(&["docker", "exec", container, "sh", "-c", &setup_cmd])?; + let _ = transport.run(&["sh", "-c", &setup_cmd])?; for (idx, p) in peers.iter().enumerate() { let class_id = cfg.traffic.class_start_id + idx as u16; @@ -65,7 +218,7 @@ fn apply_full_topology( tc filter add dev {ifb} parent 2: protocol ip u32 match ip src {ip} flowid 2:{class_id}", limit = cfg.traffic.peer_limit ); - let _ = docker::run_capture(&["docker", "exec", container, "sh", "-c", &per_peer])?; + let _ = transport.run(&["sh", "-c", &per_peer])?; } Ok(()) diff --git a/traffic_core/src/wg.rs b/traffic_core/src/wg.rs index aed613a..444716d 100644 --- a/traffic_core/src/wg.rs +++ b/traffic_core/src/wg.rs @@ -1,26 +1,39 @@ use std::net::Ipv4Addr; -use crate::{config, docker::run_capture}; +use crate::{ + config, + netns::{DockerExecTransport, NsTransport, NsenterTransport, nsenter_available}, +}; #[derive(Debug, Clone)] pub struct Peer { pub ipv4: Ipv4Addr, } -pub fn discover_peers(container: &str, peer_cidr: &str) -> anyhow::Result> { - let primary = run_capture(&[ - "docker", - "exec", - container, - "sh", - "-c", - "wg show all allowed-ips 2>/dev/null", - ]) - .unwrap_or_default(); +pub fn discover_peers(pid: u32, container: &str, peer_cidr: &str) -> anyhow::Result> { + let boxed: Box = if nsenter_available() { + Box::new(NsenterTransport { pid }) + } else { + Box::new(DockerExecTransport { + container: container.to_string(), + }) + }; + + discover_peers_with_transport(boxed.as_ref(), peer_cidr) +} + +pub fn discover_peers_with_transport( + transport: &dyn NsTransport, + peer_cidr: &str, +) -> anyhow::Result> { + let primary = transport + .run(&["sh", "-c", "wg show all allowed-ips 2>/dev/null"]) + .unwrap_or_default(); let mut peers = parse_allowed_ips(&primary, peer_cidr)?; if peers.is_empty() { - let fallback = run_capture(&["docker", "exec", container, "sh", "-c", "wg show 2>/dev/null"]) + let fallback = transport + .run(&["sh", "-c", "wg show 2>/dev/null"]) .unwrap_or_default(); peers = parse_fallback_wg_show(&fallback, peer_cidr)?; } @@ -89,5 +102,29 @@ mod tests { let peers = parse_allowed_ips(raw, "10.8.1.0/24").expect("parse"); assert_eq!(peers.len(), 2); } + + #[test] + fn parse_fallback() { + let raw = "allowed ips: 10.8.1.4/32, 10.8.1.5/32\nallowed ips: 192.168.0.1/32"; + let peers = parse_fallback_wg_show(raw, "10.8.1.0/24").expect("parse"); + assert_eq!(peers.len(), 2); + } + + #[test] + fn cidr_filtering_keeps_only_target_subnet() { + let raw = "wg0 key1 10.8.1.10/32\nwg0 key2 10.9.1.2/32"; + let peers = parse_allowed_ips(raw, "10.8.1.0/24").expect("parse"); + assert_eq!(peers.len(), 1); + assert_eq!(peers[0].ipv4.to_string(), "10.8.1.10"); + } + + #[test] + fn lexicographic_sorting_parity() { + let raw = "wg0 key1 10.8.1.2/32\nwg0 key2 10.8.1.10/32"; + let mut peers = parse_allowed_ips(raw, "10.8.1.0/24").expect("parse"); + peers.sort_by_key(|p| p.ipv4.to_string()); + assert_eq!(peers[0].ipv4.to_string(), "10.8.1.10"); + assert_eq!(peers[1].ipv4.to_string(), "10.8.1.2"); + } }