diff --git a/config.example.toml b/config.example.toml index acd527f..0695c24 100644 --- a/config.example.toml +++ b/config.example.toml @@ -8,15 +8,14 @@ peer_ipv4_cidr = "10.8.1.0/24" enabled = true peer_limit = "70mbit" root_rate = "1000mbit" -reconcile_interval_secs = 60 -container_ready_timeout_secs = 30 +reconcile_interval_secs = 15 +container_ready_timeout_secs = 120 class_start_id = 10 ifb_name = "ifb0" -full_rebuild_on_drift = true [anti_torrent] enabled = true -max_connections_per_peer = 200 +max_connections_per_peer = 100 peer_ipv4_cidr = "10.8.1.0/24" apply_in_amnezia_netns = true diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index d7dfc15..55ecc84 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -15,15 +15,14 @@ peer_ipv4_cidr = "10.8.1.0/24" enabled = true peer_limit = "70mbit" root_rate = "1000mbit" -reconcile_interval_secs = 60 -container_ready_timeout_secs = 30 +reconcile_interval_secs = 15 +container_ready_timeout_secs = 120 class_start_id = 10 ifb_name = "ifb0" -full_rebuild_on_drift = true [anti_torrent] enabled = true -max_connections_per_peer = 200 +max_connections_per_peer = 100 peer_ipv4_cidr = "10.8.1.0/24" apply_in_amnezia_netns = true @@ -62,7 +61,7 @@ swappiness = 1 `traffic.ifb_name` — имя IFB-интерфейса для ingress shaping. В старой системе это `ifb0`. Этот параметр можно оставить в конфиге, потому что бинарь реально создаёт/удаляет этот интерфейс. -`traffic.full_rebuild_on_drift` — безопасная v1-модель: при drift удалить owned topology и собрать заново. +При drift orchestrator всегда делает controlled rebuild owned topology. `anti_torrent.enabled` — включает nftables connection-count guard против high-peer-count P2P-нагрузки. Это новое расширение, не часть old bash. diff --git a/docs/IMPLEMENTATION_SPEC.md b/docs/IMPLEMENTATION_SPEC.md index d68a45e..b0c7180 100644 --- a/docs/IMPLEMENTATION_SPEC.md +++ b/docs/IMPLEMENTATION_SPEC.md @@ -236,11 +236,8 @@ reconcile_traffic(): return NoChange if drift.exists: - if config.traffic.full_rebuild_on_drift: - teardown_owned_topology(target, desired) - apply_full_topology(target, desired) - else: - apply_incremental_fix(drift) + teardown_owned_topology(target, desired) + apply_full_topology(target, desired) verify_after_apply(target, desired) if firewall_drift.exists: @@ -248,7 +245,7 @@ reconcile_traffic(): verify_firewall_after_apply(target, firewall_desired) ``` -For v1, `full_rebuild_on_drift = true` is the safer default. +For production v1, controlled full rebuild on drift is the only supported path. Parser hardening requirements: diff --git a/docs/INSTALLATION.md b/docs/INSTALLATION.md index 207af05..46498dc 100644 --- a/docs/INSTALLATION.md +++ b/docs/INSTALLATION.md @@ -31,6 +31,7 @@ nft Сборку лучше выполнять на отдельной build-машине с тем же целевым ABI. ```bash +./tools/build.sh --prepare-build-env --bootstrap-rust --yes ./tools/build.sh ``` @@ -150,15 +151,4 @@ awg2-harness status --config /etc/awg2-harness/config.toml ## 9. Rollback -Если нужно временно вернуться к old bash: - -```bash -systemctl stop awg2-harness.service -systemctl disable awg2-harness.service -# вернуть /opt/amnezia/limit_peers.sh и /opt/amnezia/docker_watcher.sh -# вернуть amnezia-tc.service и amnezia-tc-watcher.service -systemctl daemon-reload -systemctl enable --now amnezia-tc.service amnezia-tc-watcher.service -``` - -Rollback должен быть проверен до первой production-выкладки. +Production-путь AWG2 Harness не включает возврат к old bash в штатный процесс выкладки. diff --git a/docs/MIGRATION.md b/docs/MIGRATION.md index 36e3422..0aa02a1 100644 --- a/docs/MIGRATION.md +++ b/docs/MIGRATION.md @@ -183,15 +183,14 @@ peer_ipv4_cidr = "10.8.1.0/24" enabled = true peer_limit = "70mbit" root_rate = "1000mbit" -reconcile_interval_secs = 60 -container_ready_timeout_secs = 30 +reconcile_interval_secs = 15 +container_ready_timeout_secs = 120 class_start_id = 10 ifb_name = "ifb0" -full_rebuild_on_drift = true [anti_torrent] enabled = true -max_connections_per_peer = 200 +max_connections_per_peer = 100 peer_ipv4_cidr = "10.8.1.0/24" apply_in_amnezia_netns = true diff --git a/rust-toolchain.toml b/rust-toolchain.toml index b475f2f..12dd525 100644 --- a/rust-toolchain.toml +++ b/rust-toolchain.toml @@ -1,3 +1,4 @@ [toolchain] -channel = "1.85.0" +channel = "1.95.0" +profile = "minimal" components = ["rustfmt", "clippy"] diff --git a/tools/README.md b/tools/README.md index 923a9d4..12fa0b2 100644 --- a/tools/README.md +++ b/tools/README.md @@ -1,54 +1,59 @@ -# Build tooling +# AWG2 Harness builder -[`build.sh`](tools/build.sh) — production build gate для [`traffic_core`](traffic_core/Cargo.toml). +Основной скрипт: [`build.sh`](build.sh). +Целевой crate: [`traffic_core`](../traffic_core/Cargo.toml). +Unit: [`awg2-harness.service`](../awg2-harness.service). +Пример конфига: [`config.example.toml`](../config.example.toml). -## Что делает +## Supported host -Скрипт: +- Debian/Ubuntu (apt-based) +- amd64/x86_64 -1. Проверяет OS и архитектуру (Debian/Ubuntu, amd64/x86_64). -2. Проверяет наличие инструментов: `cargo`, `rustc`, `gcc`, `pkg-config`, `file`, `ldd`. -3. Проверяет минимальную версию Rust (`rustc >= 1.85.0`). -4. Запускает quality gate: - - `cargo fmt --check` (если не указан `--skip-fmt`), - - `cargo clippy --locked --all-targets -- -D warnings`, - - `cargo test --locked --release`, - - `cargo build --locked --release`. -5. Проверяет собранный бинарь: `file`, `ldd`, `--help`, `sha256sum`. - -## Что НЕ делает - -- Не устанавливает Rust/toolchain автоматически. -- Не устанавливает системные зависимости через apt. -- Не создаёт production-конфиг автоматически. - -## Использование - -Из корня репозитория: +## Default mode ```bash ./tools/build.sh ``` -Опции: +Дефолтный запуск **не изменяет систему**. Он: -- `--skip-fmt` — пропустить шаг `cargo fmt --check`. -- `--install` — после успешной сборки установить бинарь и systemd unit. +- проверяет prereqs; +- запускает `fmt/clippy/test/build` в locked-режиме; +- печатает `file`, `ldd`, `sha256` для собранного бинаря. -## Режим `--install` +## Flags -`--install`: +- `--prepare-build-env` — установить недостающие build-пакеты; +- `--prepare-runtime-env` — установить недостающие runtime-пакеты; +- `--bootstrap-rust` — установить rustup и pinned toolchain; +- `--install` — установить бинарь + systemd unit (без enable/start); +- `--enable` — `systemctl enable --now awg2-harness.service`; +- `--yes` — подтверждение системных изменений (apt/rustup); +- `--skip-fmt` — пропустить `cargo fmt --check`; +- `--help` — справка. -1. Копирует бинарь в `/usr/local/bin/awg2-harness`. -2. Проверяет, что `/etc/awg2-harness/config.toml` уже существует. -3. Копирует [`awg2-harness.service`](awg2-harness.service). -4. Выполняет `systemd-analyze verify` (если доступно). -5. Выполняет `systemctl daemon-reload`. +## Typical build machine flow -Важно: +```bash +./tools/build.sh --prepare-build-env --bootstrap-rust --yes +./tools/build.sh +``` -- Если скрипт запущен от root, команды установки выполняются напрямую. -- Если не от root — используется `sudo` (и проверяется его наличие). -- Скрипт **не** выполняет `systemctl enable --now` автоматически. +## Typical production host flow -Это intentional: включение сервиса и запуск остаются осознанным эксплуатационным шагом. +```bash +./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 +``` + +## What the script intentionally does not do + +- не устанавливает Docker автоматически; +- не генерирует production config из example; +- не включает сервис без явного `--enable`; +- не управляет установкой AmneziaWG. diff --git a/tools/build.sh b/tools/build.sh index 1612439..e92e001 100644 --- a/tools/build.sh +++ b/tools/build.sh @@ -4,103 +4,249 @@ 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" -MIN_RUSTC="1.85.0" +RUST_TOOLCHAIN_FILE="$ROOT_DIR/rust-toolchain.toml" +RUST_TOOLCHAIN_VERSION="1.95.0" -AUTO_INSTALL=false +PREPARE_BUILD_ENV=false +PREPARE_RUNTIME_ENV=false +BOOTSTRAP_RUST=false +INSTALL=false +ENABLE=false +YES=false SKIP_FMT=false +RUN_AS_ROOT=() + +BUILD_PACKAGES=( + ca-certificates + curl + build-essential + pkg-config + file + shellcheck +) + +RUNTIME_PACKAGES=( + iproute2 + wireguard-tools + nftables + kmod + util-linux + procps + systemd +) + +usage() { + cat <<'EOF' +Usage: + ./tools/build.sh + ./tools/build.sh --prepare-build-env --yes + ./tools/build.sh --prepare-runtime-env --yes + ./tools/build.sh --bootstrap-rust --yes + ./tools/build.sh --install + ./tools/build.sh --enable + +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 + --install Install binary and systemd unit (no enable/start) + --enable Enable and start awg2-harness.service + --yes Confirm system mutations (apt/rustup) + --skip-fmt Skip cargo fmt --check + --help Show this help +EOF +} + for arg in "$@"; do case "$arg" in - --install) AUTO_INSTALL=true ;; + --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 ;; + --yes) YES=true ;; --skip-fmt) SKIP_FMT=true ;; - *) echo "Unknown arg: $arg"; exit 1 ;; + --help|-h) usage; exit 0 ;; + *) echo "Unknown arg: $arg"; usage; exit 1 ;; esac done require_cmd() { - command -v "$1" >/dev/null 2>&1 || { echo "Missing command: $1"; exit 1; } + command -v "$1" >/dev/null 2>&1 || { + echo "Missing command: $1" + exit 1 + } } -version_ge() { - local have="$1" - local need="$2" - [[ "$(printf '%s\n' "$need" "$have" | sort -V | head -n1)" == "$need" ]] -} - -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 - -require_cmd rustc - -RUSTC_VERSION="$(rustc --version | awk '{print $2}')" -if ! version_ge "$RUSTC_VERSION" "$MIN_RUSTC"; then - echo "rustc $RUSTC_VERSION is too old. Require >= $MIN_RUSTC" - 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 --locked --manifest-path "$CRATE_MANIFEST" --all-targets -- -D warnings -cargo test --locked --manifest-path "$CRATE_MANIFEST" --release -cargo build --locked --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 - - RUN_AS_ROOT=( ) +setup_root_runner() { if [[ "$(id -u)" -eq 0 ]]; then - RUN_AS_ROOT=( ) + RUN_AS_ROOT=() else require_cmd sudo RUN_AS_ROOT=(sudo) fi +} - "${RUN_AS_ROOT[@]}" install -m 0755 "$TARGET_BIN" /usr/local/bin/awg2-harness - "${RUN_AS_ROOT[@]}" mkdir -p /etc/awg2-harness - if [[ ! -f /etc/awg2-harness/config.toml ]]; then - echo "Missing /etc/awg2-harness/config.toml" - echo "Refusing to install example config as production config." - echo "Create config manually from $ROOT_DIR/config.example.toml and re-run with --install." +ensure_supported_host() { + 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 + + local arch + arch="$(uname -m)" + if [[ "$arch" != "x86_64" && "$arch" != "amd64" ]]; then + echo "Unsupported arch: $arch. Only amd64/x86_64 is supported." exit 1 fi +} + +apt_install_missing() { + local packages=("$@") + local missing=() + + require_cmd dpkg-query + require_cmd apt-get + + for pkg in "${packages[@]}"; do + if ! dpkg-query -W -f='${Status}' "$pkg" 2>/dev/null | grep -q "install ok installed"; then + missing+=("$pkg") + fi + done + + if [[ "${#missing[@]}" -eq 0 ]]; then + echo "All required packages already installed" + return 0 + fi + + echo "Missing packages: ${missing[*]}" + if [[ "$YES" != true ]]; then + echo "Re-run with --yes to install them" + exit 1 + fi + + setup_root_runner + "${RUN_AS_ROOT[@]}" apt-get update + "${RUN_AS_ROOT[@]}" apt-get install -y --no-install-recommends "${missing[@]}" +} + +bootstrap_rust() { + require_cmd curl + + if [[ "$YES" != true ]]; then + echo "This will install rustup and Rust toolchain $RUST_TOOLCHAIN_VERSION" + echo "Re-run with --yes to proceed" + exit 1 + fi + + if ! command -v rustup >/dev/null 2>&1; then + local rustup_script + rustup_script="$(mktemp)" + curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs -o "$rustup_script" + sh "$rustup_script" -y --profile minimal --default-toolchain "$RUST_TOOLCHAIN_VERSION" + rm -f "$rustup_script" + fi + + export PATH="$HOME/.cargo/bin:$PATH" + require_cmd rustup + rustup toolchain install "$RUST_TOOLCHAIN_VERSION" --profile minimal --component rustfmt --component clippy + rustup component add rustfmt clippy --toolchain "$RUST_TOOLCHAIN_VERSION" + rustc --version +} + +prepare_build_gate() { + require_cmd cargo + require_cmd rustc + require_cmd gcc + require_cmd pkg-config + require_cmd file + require_cmd ldd + + if [[ ! -f "$RUST_TOOLCHAIN_FILE" ]]; then + echo "Missing rust toolchain pin file: $RUST_TOOLCHAIN_FILE" + exit 1 + fi + + if [[ ! -f "$ROOT_DIR/traffic_core/Cargo.lock" ]]; then + echo "Missing lockfile: $ROOT_DIR/traffic_core/Cargo.lock" + exit 1 + fi + + if [[ "$SKIP_FMT" != true ]]; then + cargo fmt --manifest-path "$CRATE_MANIFEST" --all --check + fi + + cargo clippy --locked --manifest-path "$CRATE_MANIFEST" --all-targets -- -D warnings + cargo test --locked --manifest-path "$CRATE_MANIFEST" --release + cargo build --locked --manifest-path "$CRATE_MANIFEST" --release + + file "$TARGET_BIN" + ldd "$TARGET_BIN" || true + "$TARGET_BIN" --help >/dev/null + sha256sum "$TARGET_BIN" + ls -lh "$TARGET_BIN" +} + +install_artifacts() { + require_cmd install + require_cmd systemctl + + if [[ ! -f "$TARGET_BIN" ]]; then + echo "Binary not found: $TARGET_BIN" + echo "Run build first: ./tools/build.sh" + exit 1 + fi + + if [[ ! -f /etc/awg2-harness/config.toml ]]; then + echo "Missing /etc/awg2-harness/config.toml" + echo "Refusing to install example config as production config" + echo "Create config manually from $ROOT_DIR/config.example.toml and re-run with --install" + exit 1 + fi + + setup_root_runner + "${RUN_AS_ROOT[@]}" install -m 0755 "$TARGET_BIN" /usr/local/bin/awg2-harness "${RUN_AS_ROOT[@]}" install -m 0644 "$ROOT_DIR/awg2-harness.service" /etc/systemd/system/awg2-harness.service + if command -v systemd-analyze >/dev/null 2>&1; then "${RUN_AS_ROOT[@]}" systemd-analyze verify /etc/systemd/system/awg2-harness.service fi "${RUN_AS_ROOT[@]}" systemctl daemon-reload +} + +enable_service() { + require_cmd systemctl + setup_root_runner + "${RUN_AS_ROOT[@]}" systemctl enable --now awg2-harness.service +} + +ensure_supported_host + +if [[ "$PREPARE_BUILD_ENV" == true ]]; then + apt_install_missing "${BUILD_PACKAGES[@]}" +fi + +if [[ "$PREPARE_RUNTIME_ENV" == true ]]; then + apt_install_missing "${RUNTIME_PACKAGES[@]}" +fi + +if [[ "$BOOTSTRAP_RUST" == true ]]; then + bootstrap_rust +fi + +prepare_build_gate + +if [[ "$INSTALL" == true ]]; then + install_artifacts +fi + +if [[ "$ENABLE" == true ]]; then + enable_service fi echo "Build completed: $TARGET_BIN" diff --git a/traffic_core/src/config.rs b/traffic_core/src/config.rs index ee98384..d2bfa66 100644 --- a/traffic_core/src/config.rs +++ b/traffic_core/src/config.rs @@ -28,7 +28,6 @@ pub struct TrafficConfig { pub container_ready_timeout_secs: u64, pub class_start_id: u16, pub ifb_name: String, - pub full_rebuild_on_drift: bool, } #[derive(Debug, Clone, Deserialize)] diff --git a/traffic_core/src/firewall.rs b/traffic_core/src/firewall.rs index 7fa5119..cb7875c 100644 --- a/traffic_core/src/firewall.rs +++ b/traffic_core/src/firewall.rs @@ -65,27 +65,21 @@ pub fn apply_firewall(cfg: &Config) -> anyhow::Result<()> { } fn inspect_firewall(transport: &dyn NsTransport, cfg: &Config) -> anyhow::Result { - let table = transport - .run(&["nft", "list", "table", "inet", "awg2_harness"]) - .unwrap_or_default(); + let table = inspect_cmd( + transport, + &["nft", "list", "table", "inet", "awg2_harness"], + "firewall table", + )?; let table_exists = !table.trim().is_empty(); let counter_exists = table.contains("counter antitorrent_dropped_v4"); let chain_exists = table.contains("chain forward_antitorrent"); - let needle_cidr = format!("ip saddr {}", cfg.anti_torrent.peer_ipv4_cidr); - let needle_max = format!( - "ct count over {}", - cfg.anti_torrent.max_connections_per_peer + let rule_matches = firewall_rule_contract_matches( + &table, + &cfg.anti_torrent.peer_ipv4_cidr, + cfg.anti_torrent.max_connections_per_peer, ); - let rule_matches = table.contains(&needle_cidr) - && table.contains(&needle_max) - && table.contains("ct state new") - && table.contains("meter awg2_peer_connlimit_v4") - && table.contains("counter name antitorrent_dropped_v4") - && table.contains("drop") - && table.contains("hook forward") - && table.contains("policy accept"); Ok(FirewallActual { table_exists, @@ -185,6 +179,27 @@ 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 firewall_rule_contract_matches(table: &str, cidr: &str, max_conn: u32) -> bool { + let needle_cidr = format!("ip saddr {cidr}"); + let needle_max = format!("ct count over {max_conn}"); + table.contains("table inet awg2_harness") + && table.contains("counter antitorrent_dropped_v4") + && table.contains("chain forward_antitorrent") + && table.contains("type filter hook forward priority filter; policy accept") + && table.contains(&needle_cidr) + && table.contains("ct state new") + && table.contains("meter awg2_peer_connlimit_v4") + && table.contains(&needle_max) + && table.contains("counter name antitorrent_dropped_v4") + && table.contains("drop") +} + pub fn antitorrent_drop_counter(cfg: &Config) -> anyhow::Result> { if !cfg.anti_torrent.enabled { return Ok(None); @@ -201,38 +216,25 @@ pub fn antitorrent_drop_counter(cfg: &Config) -> anyhow::Result { + if let Some(v) = parse_counter_json(&raw)? { + return Ok(Some(v)); + } + return Ok(None); + } + Err(e) => { + let msg = e.to_string(); + if msg.contains("No such file") + || msg.contains("No such") + || msg.contains("not found") + || msg.contains("not exist") + { + return Ok(None); + } + return Err(anyhow::anyhow!("nft counter inspection failed: {e}")); } } - - let table = transport - .run(&["nft", "list", "table", "inet", "awg2_harness"]) - .unwrap_or_default(); - for line in table.lines() { - let t = line.trim(); - if !t.contains("packets") { - continue; - } - let toks: Vec<&str> = t.split_whitespace().collect(); - let packets = toks - .iter() - .position(|x| *x == "packets") - .and_then(|i| toks.get(i + 1)) - .and_then(|s| s.parse::().ok()); - let bytes = toks - .iter() - .position(|x| *x == "bytes") - .and_then(|i| toks.get(i + 1)) - .and_then(|s| s.parse::().ok()) - .unwrap_or(0); - if let Some(p) = packets { - return Ok(Some((p, bytes))); - } - } - - Ok(None) } fn parse_counter_json(raw: &str) -> anyhow::Result> { @@ -262,4 +264,46 @@ mod tests { let got = parse_counter_json(raw).expect("json parse"); assert_eq!(got, Some((7, 448))); } + + #[test] + fn nft_rule_contract_parser_accepts_expected_rule() { + let table = r#" +table inet awg2_harness { + counter antitorrent_dropped_v4 + chain forward_antitorrent { + type filter hook forward priority filter; policy accept; + ip saddr 10.8.1.0/24 ct state new meter awg2_peer_connlimit_v4 { ip saddr ct count over 100 } counter name antitorrent_dropped_v4 drop + } +} +"#; + assert!(firewall_rule_contract_matches(table, "10.8.1.0/24", 100)); + } + + #[test] + fn missing_counter_name_causes_drift() { + let table = r#" +table inet awg2_harness { + counter antitorrent_dropped_v4 + chain forward_antitorrent { + type filter hook forward priority filter; policy accept; + ip saddr 10.8.1.0/24 ct state new meter awg2_peer_connlimit_v4 { ip saddr ct count over 100 } drop + } +} +"#; + assert!(!firewall_rule_contract_matches(table, "10.8.1.0/24", 100)); + } + + #[test] + fn wrong_max_connection_threshold_causes_drift() { + let table = r#" +table inet awg2_harness { + counter antitorrent_dropped_v4 + chain forward_antitorrent { + type filter hook forward priority filter; policy accept; + ip saddr 10.8.1.0/24 ct state new meter awg2_peer_connlimit_v4 { ip saddr ct count over 250 } counter name antitorrent_dropped_v4 drop + } +} +"#; + assert!(!firewall_rule_contract_matches(table, "10.8.1.0/24", 100)); + } } diff --git a/traffic_core/src/traffic.rs b/traffic_core/src/traffic.rs index 67d2170..87f8217 100644 --- a/traffic_core/src/traffic.rs +++ b/traffic_core/src/traffic.rs @@ -17,6 +17,13 @@ struct TrafficDrift { reason: String, } +#[derive(Debug, Clone)] +struct FilterInspect { + map: BTreeMap, + duplicate_ips: BTreeSet, + duplicate_classes: BTreeSet, +} + #[derive(Debug, Clone, PartialEq, Eq)] struct ClassRateCeil { rate: String, @@ -106,13 +113,6 @@ pub fn reconcile_traffic(cfg: &Config) -> anyhow::Result<()> { println!("traffic drift detected: {}", drift.reason); - if !cfg.traffic.full_rebuild_on_drift { - anyhow::bail!( - "traffic drift detected but full_rebuild_on_drift=false: {}", - drift.reason - ); - } - teardown_owned_topology( transport.as_ref(), &target.interface_name, @@ -174,8 +174,10 @@ pub fn peer_status_table(cfg: &Config) -> anyhow::Result anyhow::Result anyhow::Result { - let root_qdisc = transport - .run(&["tc", "qdisc", "show", "dev", iface]) - .unwrap_or_default(); + let root_qdisc = inspect_cmd( + transport, + &["tc", "qdisc", "show", "dev", iface], + "egress qdisc", + )?; if !root_qdisc.contains("htb") { return Ok(TrafficDrift { has_drift: true, @@ -262,9 +266,11 @@ fn inspect_and_diff( }); } - let ingress_qdisc = transport - .run(&["tc", "qdisc", "show", "dev", iface]) - .unwrap_or_default(); + let ingress_qdisc = inspect_cmd( + transport, + &["tc", "qdisc", "show", "dev", iface], + "ingress qdisc", + )?; if !ingress_qdisc.contains("ingress") { return Ok(TrafficDrift { has_drift: true, @@ -272,43 +278,65 @@ fn inspect_and_diff( }); } - let ifb_link = transport - .run(&["ip", "link", "show", ifb]) - .unwrap_or_default(); + let ifb_link = inspect_cmd(transport, &["ip", "link", "show", ifb], "ifb link")?; 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") { + return Ok(TrafficDrift { + has_drift: true, + reason: "ifb interface is not UP".to_string(), + }); + } - let filters_egress = transport - .run(&["tc", "filter", "show", "dev", iface, "parent", "1:"]) - .unwrap_or_default(); - let filters_ingress = transport - .run(&["tc", "filter", "show", "dev", ifb, "parent", "2:"]) - .unwrap_or_default(); + let filters_egress = inspect_cmd( + transport, + &["tc", "filter", "show", "dev", iface, "parent", "1:"], + "egress filters", + )?; + let filters_ingress = inspect_cmd( + transport, + &["tc", "filter", "show", "dev", ifb, "parent", "2:"], + "ingress filters", + )?; - let (actual_map_egress, dup_egress) = extract_filter_ip_class_map(&filters_egress); - let (actual_map_ingress, dup_ingress) = extract_filter_ip_class_map(&filters_ingress); - if !dup_egress.is_empty() { + let inspect_egress = extract_filter_ip_class_map(&filters_egress); + let inspect_ingress = extract_filter_ip_class_map(&filters_ingress); + if !inspect_egress.duplicate_ips.is_empty() { return Ok(TrafficDrift { has_drift: true, reason: format!( "duplicate egress filters detected for {} peer(s)", - dup_egress.len() + inspect_egress.duplicate_ips.len() ), }); } - if !dup_ingress.is_empty() { + if !inspect_egress.duplicate_classes.is_empty() { + return Ok(TrafficDrift { + has_drift: true, + reason: "duplicate egress class flowid mapping detected".to_string(), + }); + } + if !inspect_ingress.duplicate_ips.is_empty() { return Ok(TrafficDrift { has_drift: true, reason: format!( "duplicate ingress filters detected for {} peer(s)", - dup_ingress.len() + inspect_ingress.duplicate_ips.len() ), }); } + if !inspect_ingress.duplicate_classes.is_empty() { + return Ok(TrafficDrift { + has_drift: true, + reason: "duplicate ingress class flowid mapping detected".to_string(), + }); + } + let actual_map_egress = inspect_egress.map; + let actual_map_ingress = inspect_ingress.map; let actual_ips_egress: BTreeSet = actual_map_egress.keys().cloned().collect(); let actual_ips_ingress: BTreeSet = actual_map_ingress.keys().cloned().collect(); let expected_ips: BTreeSet = peers.iter().map(|p| p.ipv4.to_string()).collect(); @@ -358,12 +386,16 @@ fn inspect_and_diff( } let expected_limit = normalize_tc_rate(peer_limit); - let classes_egress_full = transport - .run(&["tc", "class", "show", "dev", iface]) - .unwrap_or_default(); - let classes_ingress_full = transport - .run(&["tc", "class", "show", "dev", ifb]) - .unwrap_or_default(); + let classes_egress_full = inspect_cmd( + transport, + &["tc", "class", "show", "dev", iface], + "egress classes", + )?; + let classes_ingress_full = inspect_cmd( + transport, + &["tc", "class", "show", "dev", ifb], + "ingress classes", + )?; let expected_root = normalize_tc_rate(root_rate); let egress_rates = parse_class_rate_ceil_map(&classes_egress_full, "1:"); let ingress_rates = parse_class_rate_ceil_map(&classes_ingress_full, "2:"); @@ -414,32 +446,40 @@ fn inspect_and_diff( } } - let classes_egress = transport - .run(&["tc", "class", "show", "dev", iface, "parent", "1:1"]) - .unwrap_or_default(); - let classes_ingress = transport - .run(&["tc", "class", "show", "dev", ifb, "parent", "2:1"]) - .unwrap_or_default(); - let expected_class_count = peers.len(); - if count_leaf_classes(&classes_egress) != expected_class_count { + let classes_egress = inspect_cmd( + transport, + &["tc", "class", "show", "dev", iface, "parent", "1:1"], + "egress leaf classes", + )?; + let classes_ingress = inspect_cmd( + transport, + &["tc", "class", "show", "dev", ifb, "parent", "2:1"], + "ingress leaf classes", + )?; + let expected_set: BTreeSet = expected_map.values().copied().collect(); + if collect_leaf_class_ids(&classes_egress) != expected_set { return Ok(TrafficDrift { has_drift: true, - reason: "egress per-peer class count mismatch".to_string(), + reason: "egress per-peer class set mismatch".to_string(), }); } - if count_leaf_classes(&classes_ingress) != expected_class_count { + if collect_leaf_class_ids(&classes_ingress) != expected_set { return Ok(TrafficDrift { has_drift: true, - reason: "ingress per-peer class count mismatch".to_string(), + reason: "ingress per-peer class set mismatch".to_string(), }); } - let qdisc_egress = transport - .run(&["tc", "qdisc", "show", "dev", iface]) - .unwrap_or_default(); - let qdisc_ingress = transport - .run(&["tc", "qdisc", "show", "dev", ifb]) - .unwrap_or_default(); + let qdisc_egress = inspect_cmd( + transport, + &["tc", "qdisc", "show", "dev", iface], + "egress leaf qdisc", + )?; + let qdisc_ingress = inspect_cmd( + transport, + &["tc", "qdisc", "show", "dev", ifb], + "ingress leaf qdisc", + )?; for class_id in expected_map.values() { let class_egress = format!("1:{class_id}"); let class_ingress = format!("2:{class_id}"); @@ -463,9 +503,17 @@ fn inspect_and_diff( }) } -fn extract_filter_ip_class_map(filters: &str) -> (BTreeMap, BTreeSet) { +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 extract_filter_ip_class_map(filters: &str) -> FilterInspect { let mut out = BTreeMap::new(); - let mut duplicates = BTreeSet::new(); + let mut duplicate_ips = BTreeSet::new(); + let mut class_to_ip = BTreeMap::new(); + let mut duplicate_classes = BTreeSet::new(); for block in split_filter_blocks(filters) { let mut ip: Option = None; @@ -483,13 +531,20 @@ fn extract_filter_ip_class_map(filters: &str) -> (BTreeMap, BTreeSe if let (Some(ip), Some(class_id)) = (ip, class_id) { if out.contains_key(&ip) { - duplicates.insert(ip.clone()); + duplicate_ips.insert(ip.clone()); + } + if class_to_ip.insert(class_id, ip.clone()).is_some() { + duplicate_classes.insert(class_id); } out.insert(ip, class_id); } } - (out, duplicates) + FilterInspect { + map: out, + duplicate_ips, + duplicate_classes, + } } fn has_fq_codel_for_parent(raw_qdisc: &str, parent_class: &str) -> bool { @@ -585,16 +640,24 @@ fn hex_to_ipv4_string(hex: &str) -> Option { Some(Ipv4Addr::new(b1, b2, b3, b4).to_string()) } -fn count_leaf_classes(raw: &str) -> usize { - raw.lines() - .filter_map(|l| { - let line = l.trim(); - let pos = line.find("class htb ")?; - let class = &line[(pos + "class htb ".len())..]; - let head = class.split_whitespace().next()?; - if head.ends_with(":1") { None } else { Some(()) } - }) - .count() +fn collect_leaf_class_ids(raw: &str) -> BTreeSet { + let mut out = BTreeSet::new(); + for l in raw.lines() { + let line = l.trim(); + let Some(pos) = line.find("class htb ") else { + continue; + }; + let class = &line[(pos + "class htb ".len())..]; + let Some(head) = class.split_whitespace().next() else { + continue; + }; + if let Some(class_id) = head.split(':').nth(1).and_then(|v| v.parse::().ok()) + && class_id != 1 + { + out.insert(class_id); + } + } + out } fn teardown_owned_topology( @@ -876,25 +939,34 @@ mod tests { #[test] fn parse_tc_u32_hex_ipv4() { let raw = " match 0a080105/ffffffff at 16\n flowid 1:42"; - let (m, dup) = extract_filter_ip_class_map(raw); - assert!(dup.is_empty()); - assert_eq!(m.get("10.8.1.5").copied(), Some(42)); + let inspect = extract_filter_ip_class_map(raw); + assert!(inspect.duplicate_ips.is_empty()); + assert!(inspect.duplicate_classes.is_empty()); + assert_eq!(inspect.map.get("10.8.1.5").copied(), Some(42)); } #[test] fn parse_block_based_filters() { let raw = "filter parent 1: protocol ip pref 1 u32\n match 0a080102/ffffffff at 16\n flowid 1:10\nfilter parent 1: protocol ip pref 2 u32\n match 0a080103/ffffffff at 16\n flowid 1:11"; - let (m, dup) = extract_filter_ip_class_map(raw); - assert!(dup.is_empty()); - assert_eq!(m.get("10.8.1.2").copied(), Some(10)); - assert_eq!(m.get("10.8.1.3").copied(), Some(11)); + let inspect = extract_filter_ip_class_map(raw); + assert!(inspect.duplicate_ips.is_empty()); + assert!(inspect.duplicate_classes.is_empty()); + assert_eq!(inspect.map.get("10.8.1.2").copied(), Some(10)); + assert_eq!(inspect.map.get("10.8.1.3").copied(), Some(11)); } #[test] fn detect_duplicate_filter_ip() { let raw = "filter parent 1: protocol ip pref 1 u32\n match 0a080102/ffffffff at 16\n flowid 1:10\nfilter parent 1: protocol ip pref 2 u32\n match 0a080102/ffffffff at 16\n flowid 1:10"; - let (_m, dup) = extract_filter_ip_class_map(raw); - assert!(dup.contains("10.8.1.2")); + let inspect = extract_filter_ip_class_map(raw); + assert!(inspect.duplicate_ips.contains("10.8.1.2")); + } + + #[test] + fn detect_duplicate_filter_class_flowid() { + let raw = "filter parent 1: protocol ip pref 1 u32\n match 0a080102/ffffffff at 16\n flowid 1:10\nfilter parent 1: protocol ip pref 2 u32\n match 0a080103/ffffffff at 16\n flowid 1:10"; + let inspect = extract_filter_ip_class_map(raw); + assert!(inspect.duplicate_classes.contains(&10)); } #[test] diff --git a/traffic_core/src/wg.rs b/traffic_core/src/wg.rs index 8354cf4..a035df9 100644 --- a/traffic_core/src/wg.rs +++ b/traffic_core/src/wg.rs @@ -95,6 +95,34 @@ fn parse_fallback_wg_show(raw: &str, peer_cidr: &str) -> anyhow::Result>>, + } + + impl TestTransport { + fn new(entries: Vec<(&str, anyhow::Result<&str>)>) -> Self { + let mut replies = BTreeMap::new(); + for (cmd, out) in entries { + let mapped = out.map(|s| s.to_string()); + replies.insert(cmd.to_string(), mapped); + } + Self { + replies: Mutex::new(replies), + } + } + } + + impl crate::netns::NsTransport for TestTransport { + fn run(&self, args: &[&str]) -> anyhow::Result { + let key = args.join(" "); + let mut lock = self.replies.lock().expect("lock"); + lock.remove(&key) + .unwrap_or_else(|| Err(anyhow::anyhow!("missing stub for: {key}"))) + } + } #[test] fn parse_primary() { @@ -126,4 +154,30 @@ mod tests { assert_eq!(peers[0].ipv4.to_string(), "10.8.1.10"); assert_eq!(peers[1].ipv4.to_string(), "10.8.1.2"); } + + #[test] + fn command_failure_is_propagated() { + let transport = TestTransport::new(vec![( + "wg show all allowed-ips", + Err(anyhow::anyhow!("wg broken")), + )]); + + 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")); + } + + #[test] + fn empty_primary_output_falls_back_to_wg_show() { + let transport = TestTransport::new(vec![ + ("wg show all allowed-ips", Ok("")), + ("wg show", Ok("allowed ips: 10.8.1.4/32, 10.8.1.5/32")), + ]); + + let peers = discover_peers_with_transport(&transport, "10.8.1.0/24").expect("peers"); + assert_eq!(peers.len(), 2); + assert_eq!(peers[0].ipv4.to_string(), "10.8.1.4"); + assert_eq!(peers[1].ipv4.to_string(), "10.8.1.5"); + } }