diff --git a/README.md b/README.md index a47c7e0..d30e0b0 100644 --- a/README.md +++ b/README.md @@ -10,14 +10,16 @@ ```text AWG2_harness/ -├── Amnezia limits (old bash)/ # рабочая bash-реализация; источник truth для поведения -├── reboot_module/ # legacy-архив (не production path) +├── Amnezia limits (old bash)/ # historical reference only (не часть production path) +├── reboot_module/ # archived legacy module (не часть production path) ├── traffic_core/ # целевая зона для Rust traffic/orchestrator логики └── docs/ # документация по миграции, архитектуре и эксплуатации ``` Production reboot path — только встроенный scheduler в [`traffic_core/src/reboot.rs`](traffic_core/src/reboot.rs). +Важно: ничего из legacy-деревьев не устанавливается, не вызывается и не требуется для runtime пути `awg2-harness`. + Основная документация: - [`docs/MIGRATION.md`](docs/MIGRATION.md) — исправленная миграционная стратегия bash → Rust multi-orchestrator. diff --git a/config.example.toml b/config.example.toml index 0695c24..d604374 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 = "70mbit" -root_rate = "1000mbit" +peer_limit = "50mbit" +root_rate = "500mbit" reconcile_interval_secs = 15 container_ready_timeout_secs = 120 class_start_id = 10 diff --git a/docs/ACCEPTANCE_CHECKLIST.md b/docs/ACCEPTANCE_CHECKLIST.md index a667038..9c05f85 100644 --- a/docs/ACCEPTANCE_CHECKLIST.md +++ b/docs/ACCEPTANCE_CHECKLIST.md @@ -46,8 +46,12 @@ - [ ] Повторный `apply-limits` не меняет healthy state без причины. - [ ] Если удалить root qdisc на AWG/WG-интерфейсе, daemon восстанавливает topology. - [ ] Если удалить ingress qdisc, daemon восстанавливает topology. +- [ ] Если удалить ingress redirect filter (`parent ffff:`), drift детектируется и topology восстанавливается. - [ ] Если удалить `ifb0`, daemon восстанавливает topology. +- [ ] Если сделать `ifb0` DOWN, drift детектируется и topology восстанавливается. - [ ] Если удалить один peer filter, daemon восстанавливает topology. +- [ ] Если изменить `rate/ceil` peer class, drift детектируется и topology восстанавливается. +- [ ] Если добавить stale peer class/filter, drift детектируется и topology восстанавливается. - [ ] После удаления peer устаревшие class/filter исчезают через rebuild. - [ ] После добавления peer появляются новые class/filter. @@ -65,11 +69,14 @@ - [ ] `status` показывает desired peer limit из конфига, а не hardcoded 70. - [ ] `status` показывает egress state. - [ ] `status` показывает ingress/ifb state. +- [ ] `status` показывает WireGuard clients (online/offline, latest handshake, RX/TX, endpoint). - [ ] `status` показывает missing/stale peer rules. - [ ] `status` явно показывает partial/drift state. +- [ ] `status` показывает tc filter counters по peer или `filter statistics: unavailable`. - [ ] Логи доступны через `journalctl -u awg2-harness`. - [ ] Ошибка отсутствия `tc`/`ip`/`wg` логируется явно. - [ ] Последний reconcile result виден в status или journal. +- [ ] `status` показывает последние события reconcile/firewall из `/run/awg2-harness/events.log`. ## G. Reboot scheduler @@ -94,8 +101,9 @@ - [ ] Есть chain `forward_antitorrent` с hook `forward` и policy `accept`. - [ ] Правило ограничивает `ip saddr `, default `10.8.1.0/24`. - [ ] Лимит считается per source VPN IP через meter/set, а не глобально по всей подсети. -- [ ] Порог совпадает с `anti_torrent.max_connections_per_peer`, default `200`. +- [ ] Порог совпадает с `anti_torrent.max_connections_per_peer`, default `100`. - [ ] Drop counter виден в `status`. +- [ ] На first-run без table `awg2-harness apply-firewall` создаёт `table inet awg2_harness`, а не падает. - [ ] При превышении порога новые flows дропаются, established flows ниже порога не ломаются. - [ ] После restart контейнера nftables rule пересоздаётся в новом netns. - [ ] При `anti_torrent.enabled = false` удаляется/отключается только AWG2-owned nftables state; Docker/Amnezia rules не flush-ятся. diff --git a/docs/ANTI_TORRENT.md b/docs/ANTI_TORRENT.md index 785c50f..ddbeff2 100644 --- a/docs/ANTI_TORRENT.md +++ b/docs/ANTI_TORRENT.md @@ -70,7 +70,7 @@ table inet awg2_harness { ```toml [anti_torrent] enabled = true -max_connections_per_peer = 200 +max_connections_per_peer = 100 ``` Практические варианты: diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 55ecc84..221cffb 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -13,8 +13,8 @@ peer_ipv4_cidr = "10.8.1.0/24" [traffic] enabled = true -peer_limit = "70mbit" -root_rate = "1000mbit" +peer_limit = "50mbit" +root_rate = "500mbit" reconcile_interval_secs = 15 container_ready_timeout_secs = 120 class_start_id = 10 @@ -65,7 +65,7 @@ swappiness = 1 `anti_torrent.enabled` — включает nftables connection-count guard против high-peer-count P2P-нагрузки. Это новое расширение, не часть old bash. -`anti_torrent.max_connections_per_peer` — максимум одновременных conntrack flows на один VPN peer IP. Default `200`; `250` — более консервативный вариант. +`anti_torrent.max_connections_per_peer` — максимум одновременных conntrack flows на один VPN peer IP. Default `100`; `250` — более мягкий вариант. `anti_torrent.peer_ipv4_cidr` — подсеть, на которую распространяется правило. Обычно должна совпадать с `amnezia.peer_ipv4_cidr`. diff --git a/docs/INSTALLATION.md b/docs/INSTALLATION.md index 46498dc..1303eba 100644 --- a/docs/INSTALLATION.md +++ b/docs/INSTALLATION.md @@ -59,8 +59,8 @@ 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-установка **не должна** автоматически копировать [`config.example.toml`](../config.example.toml) как боевой конфиг. +Если используется [`tools/build.sh`](../tools/build.sh), режим `--install` завершится ошибкой при отсутствии `/etc/awg2-harness/config.toml`. Режим `--install` копирует только бинарь и unit, выполняет `systemctl daemon-reload`, но не делает `systemctl enable --now`. Сначала подготовьте осознанный production-конфиг вручную. @@ -103,6 +103,8 @@ cat /proc/sys/vm/swappiness `/etc/systemd/system/awg2-harness.service`: +Reference unit в репозитории: [`awg2-harness.service`](../awg2-harness.service). + ```ini [Unit] Description=AWG2 Harness multi-orchestrator diff --git a/history.txt b/history.txt deleted file mode 100644 index 0d0e633..0000000 --- a/history.txt +++ /dev/null @@ -1,26 +0,0 @@ -Проект пишется полностью на раст, модуль ребута уже есть, я его добавил в проект, -его надо просто интегрировать в общую систему, суть пакета в том, что берется сервер, я меняю данные ssh, ставлю амнезию (awg2.0) -через официальный способ, а этот пакет заменяется обвязку, которая была на корявом bash написана (папка Amnezia limits (old bash)), суть в чем -bash довольно нестабильный в плане сложно дебажить и тд, но та версия которая щас лежить в архиве, она работает, -возможно она не эффективная или где то есть мелкие баги, но основную суть это ограничение скорости она выполняет. -Пакет на расте наш общий это как бы обертка между хостом и амнезией оригинальной, который выполняет важные задачи -для сервера и пиров при этом не лезет в само ядро амнезии, которое крутит в докере. -То есть суть такая, этот скрипт мы собрали на сервере для сборки, потом на продовом сервере настроили ssh, поставили амнезию, закинули наш бинарь на расте, там его правильно подключили и тд -и все, оно все само работает, должно переживать перезагрузки и обновления со стороны амнезии (контейнер может обновляться и наша обертка не должна быть тупой в этом плане, она должна подстраиваться). -Еще момент, хост машина всегда на дебиан 13 amd, работаем на полную стабильность и производительность, поэтому берем только этот хост, чтобы все хорошо оптимизировать. - -Еще так как у нас раст, многое можно делать нативными библиотеками (если это конечно снизит нагрузку допустим, если же профита нет, то делаем как проще, без причины усложнять не надо). - -Проект надо считать как продакшен и он на сервер ставится единожды, там автообновления не требуются для пакета, так что это не продумываем, у нас максимально стабильный пакеет, который поставил и забыл. - -Также у пакета должен быть файл конфигурации (чтобы можно было настраивать перезагрузку (он уже есть), -и настраивать там скорость пиров, щас стоит 70, но вдруг я захочу 60 и тд, то есть важные моменты в конфиг -вынести надо и он должен быть единый для всего этого пакета и понятный при этом, всякий мусор который никогда -не будет перенастраиваться туда не надо вносить!) - - -Таргет хост будет содержать дебиан 13 чистую, ssh (преднастроенный заранее), амнезию awg2.0 и после доставляется наш скрипт к ним. - -Еще момент, не помню есть ли это в Amnezia limits (old bash), но добавлю отдельно, что пакет должен добавлять в системе файл подкачки на 2гб правильный (правильный в плане с правильным приоритетом, чтобы система только в экстренной ситуации туда скидывала что-то, если основная озу забилась) - -Проект сразу поставляется в боевом варианте, без example конфигов!!! \ No newline at end of file diff --git a/tools/README.md b/tools/README.md index 12fa0b2..2e8db8e 100644 --- a/tools/README.md +++ b/tools/README.md @@ -27,6 +27,7 @@ Unit: [`awg2-harness.service`](../awg2-harness.service). - `--prepare-build-env` — установить недостающие build-пакеты; - `--prepare-runtime-env` — установить недостающие runtime-пакеты; - `--bootstrap-rust` — установить rustup и pinned toolchain; +- `--build` — запустить build gate явно; - `--install` — установить бинарь + systemd unit (без enable/start); - `--enable` — `systemctl enable --now awg2-harness.service`; - `--yes` — подтверждение системных изменений (apt/rustup); @@ -46,7 +47,6 @@ Unit: [`awg2-harness.service`](../awg2-harness.service). ./tools/build.sh --prepare-runtime-env --yes install -d -m 0755 /etc/awg2-harness install -m 0644 config.toml /etc/awg2-harness/config.toml -./tools/build.sh ./tools/build.sh --install ./tools/build.sh --enable ``` diff --git a/tools/build.sh b/tools/build.sh index e92e001..6c75caa 100644 --- a/tools/build.sh +++ b/tools/build.sh @@ -10,10 +10,12 @@ RUST_TOOLCHAIN_VERSION="1.95.0" PREPARE_BUILD_ENV=false PREPARE_RUNTIME_ENV=false BOOTSTRAP_RUST=false +DO_BUILD=false INSTALL=false ENABLE=false YES=false SKIP_FMT=false +HAS_ACTION=false RUN_AS_ROOT=() @@ -43,6 +45,7 @@ Usage: ./tools/build.sh --prepare-build-env --yes ./tools/build.sh --prepare-runtime-env --yes ./tools/build.sh --bootstrap-rust --yes + ./tools/build.sh --build ./tools/build.sh --install ./tools/build.sh --enable @@ -50,6 +53,7 @@ Flags: --prepare-build-env Install missing build dependencies via apt --prepare-runtime-env Install missing runtime dependencies via apt --bootstrap-rust Install rustup/toolchain pinned by this repo + --build Run build gate (default when no action flags) --install Install binary and systemd unit (no enable/start) --enable Enable and start awg2-harness.service --yes Confirm system mutations (apt/rustup) @@ -60,11 +64,12 @@ EOF for arg in "$@"; do case "$arg" in - --prepare-build-env) PREPARE_BUILD_ENV=true ;; - --prepare-runtime-env) PREPARE_RUNTIME_ENV=true ;; - --bootstrap-rust) BOOTSTRAP_RUST=true ;; - --install) INSTALL=true ;; - --enable) ENABLE=true ;; + --prepare-build-env) PREPARE_BUILD_ENV=true; HAS_ACTION=true ;; + --prepare-runtime-env) PREPARE_RUNTIME_ENV=true; HAS_ACTION=true ;; + --bootstrap-rust) BOOTSTRAP_RUST=true; HAS_ACTION=true ;; + --build) DO_BUILD=true; HAS_ACTION=true ;; + --install) INSTALL=true; HAS_ACTION=true ;; + --enable) ENABLE=true; HAS_ACTION=true ;; --yes) YES=true ;; --skip-fmt) SKIP_FMT=true ;; --help|-h) usage; exit 0 ;; @@ -72,6 +77,10 @@ for arg in "$@"; do esac done +if [[ "$HAS_ACTION" == false ]]; then + DO_BUILD=true +fi + require_cmd() { command -v "$1" >/dev/null 2>&1 || { echo "Missing command: $1" @@ -239,7 +248,9 @@ if [[ "$BOOTSTRAP_RUST" == true ]]; then bootstrap_rust fi -prepare_build_gate +if [[ "$DO_BUILD" == true ]]; then + prepare_build_gate +fi if [[ "$INSTALL" == true ]]; then install_artifacts @@ -249,4 +260,6 @@ if [[ "$ENABLE" == true ]]; then enable_service fi -echo "Build completed: $TARGET_BIN" +if [[ "$DO_BUILD" == true ]]; then + echo "Build completed: $TARGET_BIN" +fi diff --git a/traffic_core/src/daemon.rs b/traffic_core/src/daemon.rs index 086f9ab..e952169 100644 --- a/traffic_core/src/daemon.rs +++ b/traffic_core/src/daemon.rs @@ -1,5 +1,6 @@ use std::{ fs, + io::Write, sync::{Arc, Mutex}, thread, time::{Duration, Instant}, @@ -15,6 +16,7 @@ struct LastActionState { } const LAST_STATE_FILE: &str = "/run/awg2-harness/last_action.state"; +const EVENTS_FILE: &str = "/run/awg2-harness/events.log"; pub fn run(cfg: Config) -> anyhow::Result<()> { info!("starting daemon"); @@ -22,13 +24,17 @@ pub fn run(cfg: Config) -> anyhow::Result<()> { if let Err(e) = traffic::reconcile_traffic(&cfg) { set_err(&state, format!("startup traffic reconcile: {e}")); + append_event(&format!("startup_reconcile traffic=error error=\"{e}\"")); } else { set_ok(&state, "startup traffic reconcile ok"); + append_event("startup_reconcile traffic=ok"); } if let Err(e) = firewall::apply_firewall(&cfg) { set_err(&state, format!("startup firewall reconcile: {e}")); + append_event(&format!("startup_firewall error=\"{e}\"")); } else { set_ok(&state, "startup firewall reconcile ok"); + append_event("startup_firewall ok"); } let _ = reboot::maybe_schedule_reboot(&cfg); let reboot_cfg = cfg.clone(); @@ -53,14 +59,18 @@ pub fn run(cfg: Config) -> anyhow::Result<()> { if let Err(e) = traffic::reconcile_traffic(&cfg) { error!(error = %e, "traffic reconcile error"); set_err(&state, format!("periodic traffic reconcile: {e}")); + append_event(&format!("periodic_reconcile traffic=error error=\"{e}\"")); } else { set_ok(&state, "periodic traffic reconcile ok"); + append_event("periodic_reconcile traffic=ok"); } if let Err(e) = firewall::apply_firewall(&cfg) { error!(error = %e, "firewall reconcile error"); set_err(&state, format!("periodic firewall reconcile: {e}")); + append_event(&format!("periodic_reconcile firewall=error error=\"{e}\"")); } else { set_ok(&state, "periodic firewall reconcile ok"); + append_event("periodic_reconcile firewall=ok"); } thread::sleep(Duration::from_secs( cfg.traffic.reconcile_interval_secs.max(5), @@ -94,16 +104,21 @@ fn docker_events_loop(cfg: &Config, state: &Arc>) -> anyh if cfg.amnezia.container_names.iter().any(|n| n == &name) && wait_for_interface(cfg, &name, cfg.traffic.container_ready_timeout_secs) { + append_event(&format!("docker_event container={name} action=start")); if let Err(e) = traffic::reconcile_traffic(cfg) { set_err(state, format!("event traffic reconcile: {e}")); + append_event(&format!("event_reconcile traffic=error error=\"{e}\"")); } else { set_ok(state, "event traffic reconcile ok"); + append_event("event_reconcile traffic=ok"); } if let Err(e) = firewall::apply_firewall(cfg) { set_err(state, format!("event firewall reconcile: {e}")); + append_event(&format!("event_reconcile firewall=error error=\"{e}\"")); } else { set_ok(state, "event firewall reconcile ok"); + append_event("event_reconcile firewall=ok"); } } } @@ -150,3 +165,15 @@ fn persist_state(state: &LastActionState) { ); let _ = fs::write(LAST_STATE_FILE, content); } + +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 _ = writeln!(f, "{ts} {event}"); + } +} diff --git a/traffic_core/src/firewall.rs b/traffic_core/src/firewall.rs index cb7875c..5e56fc0 100644 --- a/traffic_core/src/firewall.rs +++ b/traffic_core/src/firewall.rs @@ -11,6 +11,12 @@ struct FirewallActual { counter_exists: bool, chain_exists: bool, rule_matches: bool, + hook_forward_present: bool, + policy_accept_present: bool, + ct_state_new_present: bool, + meter_present: bool, + counter_name_used: bool, + drop_present: bool, } pub fn apply_firewall(cfg: &Config) -> anyhow::Result<()> { @@ -25,7 +31,7 @@ pub fn apply_firewall(cfg: &Config) -> anyhow::Result<()> { } let actual = inspect_firewall(transport.as_ref(), cfg)?; - if actual.table_exists && actual.counter_exists && actual.chain_exists && actual.rule_matches { + if !firewall_has_drift(&actual) { return Ok(()); } @@ -57,23 +63,44 @@ pub fn apply_firewall(cfg: &Config) -> anyhow::Result<()> { )?; 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"); + if firewall_has_drift(&verify) { + anyhow::bail!( + "firewall verify-after-apply failed: {}", + firewall_drift_reason(&verify) + ); } Ok(()) } fn inspect_firewall(transport: &dyn NsTransport, cfg: &Config) -> anyhow::Result { - let table = inspect_cmd( - transport, - &["nft", "list", "table", "inet", "awg2_harness"], - "firewall table", - )?; + let table = match transport.run(&["nft", "list", "table", "inet", "awg2_harness"]) { + Ok(out) => out, + Err(e) if is_nft_not_found(&e.to_string()) => { + return Ok(FirewallActual { + table_exists: false, + counter_exists: false, + chain_exists: false, + rule_matches: false, + hook_forward_present: false, + policy_accept_present: false, + ct_state_new_present: false, + meter_present: false, + counter_name_used: false, + drop_present: false, + }); + } + Err(e) => return Err(anyhow::anyhow!("nft table inspect failed: {e}")), + }; 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 hook_forward_present = table.contains("hook forward"); + let policy_accept_present = table.contains("policy accept"); + let ct_state_new_present = table.contains("ct state new"); + let meter_present = table.contains("meter awg2_peer_connlimit_v4"); + let counter_name_used = table.contains("counter name antitorrent_dropped_v4"); + let drop_present = table.contains(" drop") || table.contains("\ndrop"); let rule_matches = firewall_rule_contract_matches( &table, @@ -86,6 +113,12 @@ fn inspect_firewall(transport: &dyn NsTransport, cfg: &Config) -> anyhow::Result counter_exists, chain_exists, rule_matches, + hook_forward_present, + policy_accept_present, + ct_state_new_present, + meter_present, + counter_name_used, + drop_present, }) } @@ -179,10 +212,59 @@ fn run_allow_notfound(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Res } } -fn inspect_cmd(transport: &dyn NsTransport, args: &[&str], what: &str) -> anyhow::Result { - transport - .run(args) - .map_err(|e| anyhow::anyhow!("{what} inspect failed: {e}")) +fn is_nft_not_found(msg: &str) -> bool { + let s = msg.to_ascii_lowercase(); + s.contains("no such file") + || s.contains("not found") + || s.contains("not exist") + || s.contains("no such table") +} + +fn firewall_has_drift(actual: &FirewallActual) -> bool { + !(actual.table_exists + && actual.counter_exists + && actual.chain_exists + && actual.rule_matches + && actual.hook_forward_present + && actual.policy_accept_present + && actual.ct_state_new_present + && actual.meter_present + && actual.counter_name_used + && actual.drop_present) +} + +fn firewall_drift_reason(actual: &FirewallActual) -> String { + if !actual.table_exists { + return "missing nft table".to_string(); + } + if !actual.counter_exists { + return "missing nft counter".to_string(); + } + if !actual.chain_exists { + return "missing nft chain".to_string(); + } + if !actual.hook_forward_present { + return "missing nft forward hook".to_string(); + } + if !actual.policy_accept_present { + return "missing nft policy accept".to_string(); + } + if !actual.ct_state_new_present { + return "missing ct state new".to_string(); + } + if !actual.meter_present { + return "missing nft meter".to_string(); + } + if !actual.counter_name_used { + return "missing nft counter name usage".to_string(); + } + if !actual.drop_present { + return "missing drop verdict".to_string(); + } + if !actual.rule_matches { + return "wrong anti-torrent threshold".to_string(); + } + "unknown firewall drift".to_string() } fn firewall_rule_contract_matches(table: &str, cidr: &str, max_conn: u32) -> bool { diff --git a/traffic_core/src/status.rs b/traffic_core/src/status.rs index 4063ae0..4f423bc 100644 --- a/traffic_core/src/status.rs +++ b/traffic_core/src/status.rs @@ -1,6 +1,13 @@ use crate::{config::Config, docker, firewall, netns::build_transport, traffic, wg}; -use std::{fs, process::Command}; +use std::{ + collections::BTreeMap, + fs, + process::Command, + time::{SystemTime, UNIX_EPOCH}, +}; + +const ONLINE_HANDSHAKE_THRESHOLD_SECS: u64 = 180; pub fn print_status(cfg: &Config) -> anyhow::Result<()> { println!("System:"); @@ -38,6 +45,52 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> { println!(" - {}", p.ipv4); } + println!("WireGuard clients:"); + println!(" IP State Latest handshake RX TX Endpoint"); + match wg::discover_peer_runtime( + target.container_pid, + &target.container_name, + &cfg.amnezia.peer_ipv4_cidr, + ) { + Ok(runtime_peers) => { + let mut by_ip: BTreeMap = BTreeMap::new(); + for rp in &runtime_peers { + for ip in &rp.allowed_ips { + by_ip.insert(ip.clone(), rp); + } + } + + let peers_sorted = wg::discover_peers( + target.container_pid, + &target.container_name, + &cfg.amnezia.peer_ipv4_cidr, + )?; + for p in peers_sorted { + let ip = p.ipv4.to_string(); + if let Some(rp) = by_ip.get(&ip) { + let state = handshake_is_online( + rp.latest_handshake_epoch, + ONLINE_HANDSHAKE_THRESHOLD_SECS, + ); + let hs = format_handshake_age(rp.latest_handshake_epoch); + let rx = format_bytes(rp.transfer_rx_bytes.unwrap_or(0)); + let tx = format_bytes(rp.transfer_tx_bytes.unwrap_or(0)); + let ep = rp.endpoint.clone().unwrap_or_else(|| "-".to_string()); + println!( + " {:<11} {:<8} {:<19} {:<9} {:<9} {}", + ip, state, hs, rx, tx, ep + ); + } else { + println!( + " {:<11} {:<8} {:<19} {:<9} {:<9} -", + ip, "offline", "never", "0B", "0B" + ); + } + } + } + Err(e) => println!(" unavailable ({e})"), + } + println!("Traffic desired:"); println!(" peer_limit: {}", cfg.traffic.peer_limit); println!(" root_rate: {}", cfg.traffic.root_rate); @@ -90,6 +143,23 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> { Err(e) => println!(" unavailable ({e})"), } + println!("Peer traffic counters:"); + println!(" IP Egress hits Ingress hits"); + match collect_peer_filter_hits( + &target.interface_name, + &cfg.traffic.ifb_name, + target.container_pid, + &target.container_name, + ) { + Ok(map) if map.is_empty() => println!(" filter statistics: unavailable"), + Ok(map) => { + for (ip, (eg, ing)) in map { + println!(" {:<11} {:<13} {}", ip, eg, ing); + } + } + Err(_) => println!(" filter statistics: unavailable"), + } + println!("Anti-torrent:"); println!(" enabled: {}", cfg.anti_torrent.enabled); println!( @@ -134,5 +204,115 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> { } } + 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}"); + } + } + Ok(()) } + +fn handshake_is_online(epoch: Option, threshold_secs: u64) -> &'static str { + if let Some(ts) = epoch { + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + if now.saturating_sub(ts) <= threshold_secs { + return "online"; + } + } + "offline" +} + +fn format_handshake_age(epoch: Option) -> String { + let Some(ts) = epoch else { + return "never".to_string(); + }; + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + let age = now.saturating_sub(ts); + format!("{}s ago", age) +} + +fn format_bytes(v: u64) -> String { + const KB: f64 = 1024.0; + const MB: f64 = KB * 1024.0; + const GB: f64 = MB * 1024.0; + let b = v as f64; + if b >= GB { + format!("{:.1}GB", b / GB) + } else if b >= MB { + format!("{:.1}MB", b / MB) + } else if b >= KB { + format!("{:.1}KB", b / KB) + } else { + format!("{}B", v) + } +} + +fn collect_peer_filter_hits( + iface: &str, + ifb: &str, + pid: u32, + container_name: &str, +) -> anyhow::Result> { + let transport = build_transport(pid, container_name); + let egress = transport.run(&["tc", "-s", "filter", "show", "dev", iface, "parent", "1:"])?; + let ingress = transport.run(&["tc", "-s", "filter", "show", "dev", ifb, "parent", "2:"])?; + let egress_hits = parse_tc_filter_hits(&egress); + let ingress_hits = parse_tc_filter_hits(&ingress); + + let mut out: BTreeMap = BTreeMap::new(); + for (ip, c) in egress_hits { + out.entry(ip).or_insert((0, 0)).0 = c; + } + for (ip, c) in ingress_hits { + out.entry(ip).or_insert((0, 0)).1 = c; + } + Ok(out) +} + +fn parse_tc_filter_hits(raw: &str) -> BTreeMap { + let mut out = BTreeMap::new(); + let mut current_ip: Option = None; + for line in raw.lines() { + let t = line.trim(); + if t.contains("match ") { + current_ip = extract_tc_match_ip(t); + } else if t.starts_with("Sent ") { + if let Some(ip) = current_ip.clone() { + let hits = t + .split_whitespace() + .nth(4) + .and_then(|s| s.parse::().ok()) + .unwrap_or(0); + out.insert(ip, hits); + } + } + } + out +} + +fn extract_tc_match_ip(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 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(format!("{b1}.{b2}.{b3}.{b4}")) +} diff --git a/traffic_core/src/traffic.rs b/traffic_core/src/traffic.rs index 87f8217..d90baa5 100644 --- a/traffic_core/src/traffic.rs +++ b/traffic_core/src/traffic.rs @@ -278,17 +278,38 @@ fn inspect_and_diff( }); } - let ifb_link = inspect_cmd(transport, &["ip", "link", "show", ifb], "ifb link")?; + let ingress_redirect = inspect_cmd( + transport, + &["tc", "filter", "show", "dev", iface, "parent", "ffff:"], + "ingress redirect filter", + )?; + if !has_ingress_redirect_to_ifb(&ingress_redirect, ifb) { + return Ok(TrafficDrift { + has_drift: true, + reason: format!("missing ingress redirect to {ifb}"), + }); + } + + let ifb_link = match transport.run(&["ip", "link", "show", ifb]) { + Ok(out) => out, + Err(e) if is_not_found_error(&e.to_string()) => { + return Ok(TrafficDrift { + has_drift: true, + reason: "missing ifb interface".to_string(), + }); + } + Err(e) => return Err(anyhow::anyhow!("ifb link inspect failed: {e}")), + }; if ifb_link.trim().is_empty() { return Ok(TrafficDrift { has_drift: true, reason: "missing ifb interface".to_string(), }); } - if !ifb_link.contains(" UP ") && !ifb_link.contains("state UP") { + if !link_is_up(&ifb_link) { return Ok(TrafficDrift { has_drift: true, - reason: "ifb interface is not UP".to_string(), + reason: format!("{ifb} exists but is not UP"), }); } @@ -553,6 +574,27 @@ fn has_fq_codel_for_parent(raw_qdisc: &str, parent_class: &str) -> bool { }) } +fn has_ingress_redirect_to_ifb(output: &str, ifb_name: &str) -> bool { + output.lines().any(|line| { + line.contains("mirred") + && line.contains("redirect") + && line.contains(&format!("dev {ifb_name}")) + }) || output.contains(&format!("redirect dev {ifb_name}")) +} + +fn link_is_up(output: &str) -> bool { + output.contains("state UP") || (output.contains('<') && output.contains("UP")) +} + +fn is_not_found_error(stderr_or_msg: &str) -> bool { + let s = stderr_or_msg.to_ascii_lowercase(); + s.contains("does not exist") + || s.contains("not found") + || s.contains("cannot find device") + || s.contains("no such device") + || s.contains("no such file") +} + fn parse_root_class_rate_ceil(raw: &str, class_id: &str) -> Option { for line in raw.lines() { let t = line.trim(); @@ -976,6 +1018,37 @@ mod tests { assert!(!has_fq_codel_for_parent(raw, "1:11")); } + #[test] + fn detects_missing_ingress_redirect() { + let raw = "filter parent ffff: protocol all pref 1 u32"; + assert!(!has_ingress_redirect_to_ifb(raw, "ifb0")); + } + + #[test] + fn detects_ingress_redirect_to_ifb() { + let raw = "filter parent ffff: protocol all pref 1 u32\n action order 1: mirred (Egress Redirect to device ifb0) stolen"; + assert!(has_ingress_redirect_to_ifb(raw, "ifb0")); + } + + #[test] + fn detects_ifb_up_state() { + assert!(link_is_up( + "2: ifb0: mtu 1500 qdisc fq_codel state UNKNOWN" + )); + assert!(link_is_up("2: ifb0: mtu 1500 state UP")); + assert!(!link_is_up( + "2: ifb0: mtu 1500 state DOWN" + )); + } + + #[test] + fn detects_not_found_errors() { + assert!(is_not_found_error("RTNETLINK answers: No such device")); + assert!(is_not_found_error("Cannot find device \"ifb0\"")); + assert!(is_not_found_error("does not exist")); + assert!(!is_not_found_error("permission denied")); + } + #[test] fn parse_root_rate_ceil() { let raw = "class htb 1:1 root rate 1000Mbit ceil 1000Mbit"; diff --git a/traffic_core/src/wg.rs b/traffic_core/src/wg.rs index a035df9..0563927 100644 --- a/traffic_core/src/wg.rs +++ b/traffic_core/src/wg.rs @@ -10,6 +10,16 @@ pub struct Peer { pub ipv4: Ipv4Addr, } +#[derive(Debug, Clone)] +pub struct PeerRuntime { + pub public_key: String, + pub allowed_ips: Vec, + pub endpoint: Option, + pub latest_handshake_epoch: Option, + pub transfer_rx_bytes: Option, + pub transfer_tx_bytes: Option, +} + pub fn discover_peers(pid: u32, container: &str, peer_cidr: &str) -> anyhow::Result> { let boxed: Box = if nsenter_available() { Box::new(NsenterTransport { pid }) @@ -43,6 +53,31 @@ pub fn discover_peers_with_transport( Ok(peers) } +pub fn discover_peer_runtime( + 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_peer_runtime_with_transport(boxed.as_ref(), peer_cidr) +} + +pub fn discover_peer_runtime_with_transport( + transport: &dyn NsTransport, + peer_cidr: &str, +) -> anyhow::Result> { + let raw = transport + .run(&["wg", "show", "all", "dump"]) + .map_err(|e| anyhow::anyhow!("wg show all dump failed: {e}"))?; + parse_wg_show_dump_runtime_peers(&raw, peer_cidr) +} + fn parse_allowed_ips(raw: &str, peer_cidr: &str) -> anyhow::Result> { let mut out = Vec::new(); for line in raw.lines() { @@ -92,6 +127,76 @@ fn parse_fallback_wg_show(raw: &str, peer_cidr: &str) -> anyhow::Result anyhow::Result> { + let mut out = Vec::new(); + for line in raw.lines() { + let cols: Vec<&str> = line.split('\t').collect(); + if cols.len() < 8 { + continue; + } + + // peer lines include endpoint/allowed_ips/handshake/rx/tx; interface lines do not match this contract + let public_key = cols.get(1).map(|s| s.trim()).unwrap_or_default(); + let allowed = cols.get(4).map(|s| s.trim()).unwrap_or_default(); + if public_key.is_empty() || allowed.is_empty() || allowed == "(none)" { + continue; + } + + let filtered_allowed = filter_allowed_ips_in_cidr(allowed, peer_cidr)?; + if filtered_allowed.is_empty() { + continue; + } + + let endpoint = cols + .get(3) + .map(|s| s.trim()) + .filter(|s| !s.is_empty() && *s != "(none)") + .map(|s| s.to_string()); + + let latest_handshake_epoch = cols + .get(5) + .and_then(|s| s.trim().parse::().ok()) + .filter(|v| *v > 0); + let transfer_rx_bytes = cols.get(6).and_then(|s| s.trim().parse::().ok()); + let transfer_tx_bytes = cols.get(7).and_then(|s| s.trim().parse::().ok()); + + out.push(PeerRuntime { + public_key: public_key.to_string(), + allowed_ips: filtered_allowed, + endpoint, + latest_handshake_epoch, + transfer_rx_bytes, + transfer_tx_bytes, + }); + } + Ok(out) +} + +fn filter_allowed_ips_in_cidr(raw: &str, peer_cidr: &str) -> anyhow::Result> { + let mut out = Vec::new(); + for tok in raw.split(',') { + let tok = tok.trim(); + let Some((ip, pfx)) = tok.split_once('/') else { + continue; + }; + if pfx != "32" { + continue; + } + let Ok(v4) = ip.parse::() else { + continue; + }; + if config::ip_in_cidr(v4, peer_cidr)? { + out.push(v4.to_string()); + } + } + out.sort(); + out.dedup(); + Ok(out) +} + #[cfg(test)] mod tests { use super::*; @@ -180,4 +285,34 @@ mod tests { assert_eq!(peers[0].ipv4.to_string(), "10.8.1.4"); assert_eq!(peers[1].ipv4.to_string(), "10.8.1.5"); } + + #[test] + fn parses_wg_show_dump_runtime_peers() { + let raw = "wg0\tpriv\tpub\t51820\toff\nwg0\tpubkey1\t(none)\t1.2.3.4:51820\t10.8.1.10/32\t1715000000\t123\t456\t0\nwg0\tpubkey2\t(none)\t(none)\t10.8.1.27/32,192.168.1.5/32\t0\t0\t0\t0"; + let peers = parse_wg_show_dump_runtime_peers(raw, "10.8.1.0/24").expect("parse"); + assert_eq!(peers.len(), 2); + assert_eq!(peers[0].allowed_ips, vec!["10.8.1.10".to_string()]); + assert_eq!(peers[1].allowed_ips, vec!["10.8.1.27".to_string()]); + assert_eq!(peers[1].latest_handshake_epoch, None); + } + + #[test] + fn maps_allowed_ip_to_runtime_peer() { + let raw = "wg0\tpubkey1\t(none)\t1.2.3.4:51820\t10.8.1.10/32,10.8.1.11/32\t1715000000\t123\t456\t0"; + let peers = parse_wg_show_dump_runtime_peers(raw, "10.8.1.0/24").expect("parse"); + assert_eq!(peers.len(), 1); + assert_eq!(peers[0].allowed_ips.len(), 2); + } + + #[test] + fn propagates_wg_dump_command_failure() { + let transport = TestTransport::new(vec![( + "wg show all dump", + Err(anyhow::anyhow!("wg broken")), + )]); + let err = discover_peer_runtime_with_transport(&transport, "10.8.1.0/24") + .err() + .expect("err"); + assert!(err.to_string().contains("wg show all dump failed")); + } }