Полная реализация production-фиксов из up3: anti-torrent rebuild, peer-status, drift rate/ceil, reboot window

README.md

@@ -11,11 +11,13 @@
 ```text
 AWG2_harness/
 ├── Amnezia limits (old bash)/   # рабочая bash-реализация; источник truth для поведения
-├── reboot_module/               # готовый Rust-модуль плановой перезагрузки
+├── reboot_module/               # legacy-архив (не production path)
 ├── traffic_core/                # целевая зона для Rust traffic/orchestrator логики
 └── docs/                        # документация по миграции, архитектуре и эксплуатации
 ```
 
+Production reboot path — только встроенный scheduler в [`traffic_core/src/reboot.rs`](traffic_core/src/reboot.rs).
+
 Основная документация:
 
 - [`docs/MIGRATION.md`](docs/MIGRATION.md) — исправленная миграционная стратегия bash → Rust multi-orchestrator.

docs/CONFIGURATION.md

@@ -70,7 +70,12 @@ swappiness = 1
 
 `anti_torrent.peer_ipv4_cidr` — подсеть, на которую распространяется правило. Обычно должна совпадать с `amnezia.peer_ipv4_cidr`.
 
-`anti_torrent.apply_in_amnezia_netns` — правило применяется в network namespace контейнера Amnezia. Это default, потому что именно там видны исходные VPN peer IP до возможного NAT.
+`anti_torrent.apply_in_amnezia_netns` — выбор namespace для применения anti-torrent правил:
+
+- `true` (default): применять в network namespace контейнера Amnezia;
+- `false`: применять в host namespace.
+
+Для production рекомендуется `true`, если peer source IP (`10.8.1.x`) видны именно внутри netns контейнера.
 
 `reboot.enabled` — включает плановую перезагрузку.
 

docs/MIGRATION.md

@@ -124,7 +124,7 @@ per-peer ingress filter: src peer_ip -> 2:<id>
 4. `validate-config` — проверка конфига без изменения системы.
 5. `apply-firewall` — ручной reconcile AWG2-owned nftables state, если anti-torrent включён.
 6. `ensure-swap` — идемпотентный setup swap-файла 2 ГБ.
-7. Встроенный reboot scheduler из `reboot_module`.
+7. Встроенный reboot scheduler в `traffic_core` (единый production path, без отдельного runtime-модуля).
 8. Locking, чтобы одновременно не шли два apply/reconcile.
 9. Docker event watcher.
 10. Периодический reconciliation loop, потому что Docker events могут быть пропущены.
@@ -248,7 +248,7 @@ WantedBy=multi-user.target
 
 ## 12. Reboot и swap
 
-`reboot_module` интегрируется внутрь общего daemon. Окно перезагрузки должно переехать из констант в конфиг. Механика `systemctl --no-block reboot` сохраняется.
+Логика reboot работает только через встроенный scheduler `traffic_core`. Окно перезагрузки задаётся в конфиге, включая корректные окна через полночь. Механика `systemctl --no-block reboot` сохраняется.
 
 `ensure-swap` должен создать 2 ГБ swap-файл с правами `0600`, добавить одну запись в `/etc/fstab`, включить swap и выставить низкую склонность к использованию (`vm.swappiness = 1`, низкий priority). Если на сервере уже есть несовместимый swap, команда должна сообщить об этом явно, а не разрушать чужое состояние без подтверждения.
 

traffic_core/src/firewall.rs

@@ -1,7 +1,7 @@
 use crate::{
     config::Config,
     docker,
-    netns::{NsTransport, build_transport},
+    netns::{NsTransport, build_host_transport, build_transport},
 };
 
 #[derive(Debug, Clone)]
@@ -13,11 +13,13 @@ struct FirewallActual {
 }
 
 pub fn apply_firewall(cfg: &Config) -> anyhow::Result<()> {
-    let target = docker::discover_target(cfg)?;
-    let transport = build_transport(target.container_pid, &target.container_name);
+    let transport = firewall_transport(cfg)?;
 
     if !cfg.anti_torrent.enabled {
-        let _ = transport.run(&["nft", "delete", "table", "inet", "awg2_harness"]);
+        run_allow_notfound(
+            transport.as_ref(),
+            &["nft", "delete", "table", "inet", "awg2_harness"],
+        )?;
         return Ok(());
     }
 
@@ -26,11 +28,16 @@ pub fn apply_firewall(cfg: &Config) -> anyhow::Result<()> {
         return Ok(());
     }
 
-    run_allow_exists(
+    run_allow_notfound(
+        transport.as_ref(),
+        &["nft", "delete", "table", "inet", "awg2_harness"],
+    )?;
+
+    run_step(
         transport.as_ref(),
         &["nft", "add", "table", "inet", "awg2_harness"],
     )?;
-    run_allow_exists(
+    run_step(
         transport.as_ref(),
         &[
             "nft",
@@ -41,7 +48,7 @@ pub fn apply_firewall(cfg: &Config) -> anyhow::Result<()> {
             "antitorrent_dropped_v4",
         ],
     )?;
-    run_allow_exists_chain(transport.as_ref())?;
+    run_step_chain(transport.as_ref())?;
     apply_rule(
         transport.as_ref(),
         &cfg.anti_torrent.peer_ipv4_cidr,
@@ -80,6 +87,15 @@ fn inspect_firewall(transport: &dyn NsTransport, cfg: &Config) -> anyhow::Result
     })
 }
 
+fn firewall_transport(cfg: &Config) -> anyhow::Result<Box<dyn NsTransport>> {
+    if cfg.anti_torrent.apply_in_amnezia_netns {
+        let target = docker::discover_target(cfg)?;
+        Ok(build_transport(target.container_pid, &target.container_name))
+    } else {
+        Ok(build_host_transport())
+    }
+}
+
 fn apply_rule(transport: &dyn NsTransport, cidr: &str, max_conn: u32) -> anyhow::Result<()> {
     let _ = transport.run(&[
         "nft",
@@ -112,21 +128,13 @@ fn apply_rule(transport: &dyn NsTransport, cidr: &str, max_conn: u32) -> anyhow:
     Ok(())
 }
 
-fn run_allow_exists(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> {
-    match transport.run(args) {
-        Ok(_) => Ok(()),
-        Err(e) => {
-            let msg = e.to_string();
-            if msg.contains("File exists") || msg.contains("exists") {
-                return Ok(());
-            }
-            Err(e)
-        }
-    }
+fn run_step(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> {
+    let _ = transport.run(args)?;
+    Ok(())
 }
 
-fn run_allow_exists_chain(transport: &dyn NsTransport) -> anyhow::Result<()> {
-    run_allow_exists(
+fn run_step_chain(transport: &dyn NsTransport) -> anyhow::Result<()> {
+    run_step(
         transport,
         &[
             "nft",
@@ -149,13 +157,29 @@ fn run_allow_exists_chain(transport: &dyn NsTransport) -> anyhow::Result<()> {
     )
 }
 
+fn run_allow_notfound(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> {
+    match transport.run(args) {
+        Ok(_) => Ok(()),
+        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(());
+            }
+            Err(e)
+        }
+    }
+}
+
 pub fn antitorrent_drop_counter(cfg: &Config) -> anyhow::Result<Option<(u64, u64)>> {
     if !cfg.anti_torrent.enabled {
         return Ok(None);
     }
 
-    let target = docker::discover_target(cfg)?;
-    let transport = build_transport(target.container_pid, &target.container_name);
+    let transport = firewall_transport(cfg)?;
     let table = transport
         .run(&["nft", "list", "table", "inet", "awg2_harness"])
         .unwrap_or_default();

traffic_core/src/netns.rs

@@ -14,6 +14,9 @@ pub struct DockerExecTransport {
     pub container: String,
 }
 
+#[derive(Debug, Clone)]
+pub struct HostTransport;
+
 impl NsTransport for NsenterTransport {
     fn run(&self, args: &[&str]) -> anyhow::Result<String> {
         let mut cmd = Command::new("nsenter");
@@ -42,6 +45,18 @@ impl NsTransport for DockerExecTransport {
     }
 }
 
+impl NsTransport for HostTransport {
+    fn run(&self, args: &[&str]) -> anyhow::Result<String> {
+        let mut cmd = Command::new(args.first().ok_or_else(|| anyhow::anyhow!("empty command"))?);
+        cmd.args(&args[1..]);
+        let out = cmd.output()?;
+        if !out.status.success() {
+            anyhow::bail!("host 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("--help").output().is_ok()
 }
@@ -55,3 +70,7 @@ pub fn build_transport(pid: u32, container: &str) -> Box<dyn NsTransport> {
         })
     }
 }
+
+pub fn build_host_transport() -> Box<dyn NsTransport> {
+    Box::new(HostTransport)
+}

traffic_core/src/reboot.rs

@@ -66,7 +66,7 @@ fn next_reboot_time(start_hhmm: &str, end_hhmm: &str) -> anyhow::Result<NaiveDat
     let span = if end > start {
         end.signed_duration_since(start)
     } else {
-        Duration::hours(1)
+        Duration::hours(24) - start.signed_duration_since(end)
     };
 
     let secs = rand::thread_rng().gen_range(0..span.num_seconds().max(1));
@@ -90,3 +90,27 @@ fn parse_hhmm(v: &str) -> anyhow::Result<NaiveTime> {
     }
     NaiveTime::from_hms_opt(h, m, 0).ok_or_else(|| anyhow::anyhow!("bad time"))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_time_rejects_invalid() {
+        assert!(parse_hhmm("24:00").is_err());
+        assert!(parse_hhmm("10:60").is_err());
+        assert!(parse_hhmm("ab:cd").is_err());
+    }
+
+    #[test]
+    fn cross_midnight_window_has_expected_span() {
+        let start = parse_hhmm("23:00").expect("start");
+        let end = parse_hhmm("02:00").expect("end");
+        let span = if end > start {
+            end.signed_duration_since(start)
+        } else {
+            Duration::hours(24) - start.signed_duration_since(end)
+        };
+        assert_eq!(span.num_hours(), 3);
+    }
+}

traffic_core/src/status.rs

@@ -67,6 +67,29 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> {
     println!("    ingress_present: {}", ingress_qdisc.contains("ingress"));
     println!("    ifb_htb_present: {}", ifb_qdisc.contains("htb"));
 
+    println!("  peer_table:");
+    match traffic::peer_status_table(cfg) {
+        Ok(Some(rows)) if rows.is_empty() => println!("    no peers discovered"),
+        Ok(Some(rows)) => {
+            for r in rows {
+                println!(
+                    "    ip={} class_id={} egress_filter={} ingress_filter={} egress={}/{} ingress={}/{} drift={}",
+                    r.ip,
+                    r.class_id,
+                    r.egress_filter_present,
+                    r.ingress_filter_present,
+                    r.egress_rate.unwrap_or_else(|| "n/a".to_string()),
+                    r.egress_ceil.unwrap_or_else(|| "n/a".to_string()),
+                    r.ingress_rate.unwrap_or_else(|| "n/a".to_string()),
+                    r.ingress_ceil.unwrap_or_else(|| "n/a".to_string()),
+                    r.drift_reason.unwrap_or_else(|| "no".to_string())
+                );
+            }
+        }
+        Ok(None) => println!("    traffic disabled"),
+        Err(e) => println!("    unavailable ({e})"),
+    }
+
     println!("Anti-torrent:");
     println!("  enabled: {}", cfg.anti_torrent.enabled);
     println!(

traffic_core/src/traffic.rs

@@ -17,6 +17,25 @@ struct TrafficDrift {
     reason: String,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct ClassRateCeil {
+    rate: String,
+    ceil: String,
+}
+
+#[derive(Debug, Clone)]
+pub struct PeerTrafficStatus {
+    pub ip: String,
+    pub class_id: u16,
+    pub egress_filter_present: bool,
+    pub ingress_filter_present: bool,
+    pub egress_rate: Option<String>,
+    pub egress_ceil: Option<String>,
+    pub ingress_rate: Option<String>,
+    pub ingress_ceil: Option<String>,
+    pub drift_reason: Option<String>,
+}
+
 pub fn inspect_drift_summary(cfg: &Config) -> anyhow::Result<Option<String>> {
     if !cfg.traffic.enabled {
         return Ok(None);
@@ -39,6 +58,7 @@ pub fn inspect_drift_summary(cfg: &Config) -> anyhow::Result<Option<String>> {
         &target.interface_name,
         &cfg.traffic.ifb_name,
         cfg.traffic.class_start_id,
+        &cfg.traffic.peer_limit,
         &peers,
     )?;
 
@@ -73,6 +93,7 @@ pub fn reconcile_traffic(cfg: &Config) -> anyhow::Result<()> {
         &target.interface_name,
         &cfg.traffic.ifb_name,
         cfg.traffic.class_start_id,
+        &cfg.traffic.peer_limit,
         &peers,
     )?;
 
@@ -101,11 +122,104 @@ pub fn reconcile_traffic(cfg: &Config) -> anyhow::Result<()> {
     Ok(())
 }
 
+pub fn peer_status_table(cfg: &Config) -> anyhow::Result<Option<Vec<PeerTrafficStatus>>> {
+    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(Vec::new()));
+    }
+
+    let filters_egress = transport
+        .run(&["tc", "filter", "show", "dev", &target.interface_name, "parent", "1:"])
+        .unwrap_or_default();
+    let filters_ingress = transport
+        .run(&["tc", "filter", "show", "dev", &cfg.traffic.ifb_name, "parent", "2:"])
+        .unwrap_or_default();
+    let classes_egress_full = transport
+        .run(&["tc", "class", "show", "dev", &target.interface_name])
+        .unwrap_or_default();
+    let classes_ingress_full = transport
+        .run(&["tc", "class", "show", "dev", &cfg.traffic.ifb_name])
+        .unwrap_or_default();
+
+    let actual_map_egress = extract_filter_ip_class_map(&filters_egress);
+    let actual_map_ingress = extract_filter_ip_class_map(&filters_ingress);
+    let egress_rates = parse_class_rate_ceil_map(&classes_egress_full, "1:");
+    let ingress_rates = parse_class_rate_ceil_map(&classes_ingress_full, "2:");
+    let expected_map = expected_peer_class_map(&peers, cfg.traffic.class_start_id);
+    let expected_limit = normalize_tc_rate(&cfg.traffic.peer_limit);
+
+    let mut out = Vec::with_capacity(peers.len());
+    for p in peers {
+        let ip = p.ipv4.to_string();
+        let class_id = *expected_map.get(&ip).unwrap_or(&cfg.traffic.class_start_id);
+        let class_egress = format!("1:{class_id}");
+        let class_ingress = format!("2:{class_id}");
+
+        let egress_filter_present = actual_map_egress.get(&ip).copied() == Some(class_id);
+        let ingress_filter_present = actual_map_ingress.get(&ip).copied() == Some(class_id);
+        let e = egress_rates.get(&class_egress);
+        let i = ingress_rates.get(&class_ingress);
+
+        let mut reasons = Vec::new();
+        if !egress_filter_present {
+            reasons.push("missing/mismatched egress filter".to_string());
+        }
+        if !ingress_filter_present {
+            reasons.push("missing/mismatched ingress filter".to_string());
+        }
+        if e.is_none() {
+            reasons.push("missing egress class".to_string());
+        }
+        if i.is_none() {
+            reasons.push("missing ingress class".to_string());
+        }
+        if let Some(v) = e
+            && !rate_ceil_matches_limit(v, &expected_limit)
+        {
+            reasons.push("egress rate/ceil mismatch".to_string());
+        }
+        if let Some(v) = i
+            && !rate_ceil_matches_limit(v, &expected_limit)
+        {
+            reasons.push("ingress rate/ceil mismatch".to_string());
+        }
+
+        out.push(PeerTrafficStatus {
+            ip,
+            class_id,
+            egress_filter_present,
+            ingress_filter_present,
+            egress_rate: e.map(|v| v.rate.clone()),
+            egress_ceil: e.map(|v| v.ceil.clone()),
+            ingress_rate: i.map(|v| v.rate.clone()),
+            ingress_ceil: i.map(|v| v.ceil.clone()),
+            drift_reason: if reasons.is_empty() {
+                None
+            } else {
+                Some(reasons.join(", "))
+            },
+        });
+    }
+
+    Ok(Some(out))
+}
+
 fn inspect_and_diff(
     transport: &dyn NsTransport,
     iface: &str,
     ifb: &str,
     class_start_id: u16,
+    peer_limit: &str,
     peers: &[wg::Peer],
 ) -> anyhow::Result<TrafficDrift> {
     let root_qdisc = transport
@@ -195,6 +309,37 @@ 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 egress_rates = parse_class_rate_ceil_map(&classes_egress_full, "1:");
+    let ingress_rates = parse_class_rate_ceil_map(&classes_ingress_full, "2:");
+
+    for class_id in expected_map.values() {
+        let class_egress = format!("1:{class_id}");
+        let class_ingress = format!("2:{class_id}");
+        let e = egress_rates.get(&class_egress);
+        let i = ingress_rates.get(&class_ingress);
+        if e.is_none() || i.is_none() {
+            return Ok(TrafficDrift {
+                has_drift: true,
+                reason: format!("missing rate/ceil class details for peer class_id={class_id}"),
+            });
+        }
+        if e.map(|v| !rate_ceil_matches_limit(v, &expected_limit)).unwrap_or(true)
+            || i.map(|v| !rate_ceil_matches_limit(v, &expected_limit)).unwrap_or(true)
+        {
+            return Ok(TrafficDrift {
+                has_drift: true,
+                reason: format!("rate/ceil mismatch for peer class_id={class_id}"),
+            });
+        }
+    }
+
     let classes_egress = transport
         .run(&["tc", "class", "show", "dev", iface, "parent", "1:1"])
         .unwrap_or_default();
@@ -324,9 +469,9 @@ fn teardown_owned_topology(
     iface: &str,
     ifb: &str,
 ) -> anyhow::Result<()> {
-    let _ = transport.run(&["tc", "qdisc", "del", "dev", iface, "root"]);
-    let _ = transport.run(&["tc", "qdisc", "del", "dev", iface, "ingress"]);
-    let _ = transport.run(&["ip", "link", "del", ifb]);
+    run_allow_notfound(transport, &["tc", "qdisc", "del", "dev", iface, "root"])?;
+    run_allow_notfound(transport, &["tc", "qdisc", "del", "dev", iface, "ingress"])?;
+    run_allow_notfound(transport, &["ip", "link", "del", ifb])?;
     Ok(())
 }
 
@@ -499,7 +644,14 @@ fn apply_full_topology(
         )?;
     }
 
-    let verify = inspect_and_diff(transport, iface, ifb, cfg.traffic.class_start_id, peers)?;
+    let verify = inspect_and_diff(
+        transport,
+        iface,
+        ifb,
+        cfg.traffic.class_start_id,
+        &cfg.traffic.peer_limit,
+        peers,
+    )?;
     if verify.has_drift {
         anyhow::bail!("verify-after-apply failed: {}", verify.reason);
     }
@@ -517,7 +669,7 @@ fn run_allow_exists(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Resul
         Ok(_) => Ok(()),
         Err(e) => {
             let msg = e.to_string();
-            if msg.contains("File exists") || msg.contains("exists") || msg.contains("EEXIST") {
+            if msg.contains("File exists") || msg.contains("EEXIST") {
                 return Ok(());
             }
             Err(e)
@@ -525,6 +677,64 @@ fn run_allow_exists(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Resul
     }
 }
 
+fn run_allow_notfound(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> {
+    match transport.run(args) {
+        Ok(_) => Ok(()),
+        Err(e) => {
+            let msg = e.to_string();
+            if msg.contains("No such file")
+                || msg.contains("No such")
+                || msg.contains("not found")
+                || msg.contains("Cannot find")
+            {
+                return Ok(());
+            }
+            Err(e)
+        }
+    }
+}
+
+fn parse_class_rate_ceil_map(raw: &str, prefix: &str) -> BTreeMap<String, ClassRateCeil> {
+    let mut out = BTreeMap::new();
+    for line in raw.lines() {
+        let t = line.trim();
+        if !t.contains("class htb") || !t.contains(" rate ") || !t.contains(" ceil ") {
+            continue;
+        }
+        let toks: Vec<&str> = t.split_whitespace().collect();
+        let class = toks
+            .iter()
+            .position(|v| *v == "htb")
+            .and_then(|i| toks.get(i + 1))
+            .map(|s| s.to_string());
+        let rate = toks
+            .iter()
+            .position(|v| *v == "rate")
+            .and_then(|i| toks.get(i + 1))
+            .map(|s| s.to_string());
+        let ceil = toks
+            .iter()
+            .position(|v| *v == "ceil")
+            .and_then(|i| toks.get(i + 1))
+            .map(|s| s.to_string());
+        if let (Some(class), Some(rate), Some(ceil)) = (class, rate, ceil)
+            && class.starts_with(prefix)
+            && !class.ends_with(":1")
+        {
+            out.insert(class, ClassRateCeil { rate, ceil });
+        }
+    }
+    out
+}
+
+fn rate_ceil_matches_limit(v: &ClassRateCeil, limit: &str) -> bool {
+    normalize_tc_rate(&v.rate) == limit && normalize_tc_rate(&v.ceil) == limit
+}
+
+fn normalize_tc_rate(v: &str) -> String {
+    v.trim().to_ascii_lowercase().replace(' ', "")
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -543,4 +753,23 @@ mod tests {
         assert_eq!(m.get("10.8.1.2").copied(), Some(10));
         assert_eq!(m.get("10.8.1.3").copied(), Some(11));
     }
+
+    #[test]
+    fn parse_class_rate_ceil() {
+        let raw = "class htb 1:10 parent 1:1 rate 70Mbit ceil 70Mbit burst 1599b cburst 1599b\nclass htb 1:1 root rate 1000Mbit ceil 1000Mbit";
+        let m = parse_class_rate_ceil_map(raw, "1:");
+        let rc = m.get("1:10").expect("class exists");
+        assert_eq!(normalize_tc_rate(&rc.rate), "70mbit");
+        assert_eq!(normalize_tc_rate(&rc.ceil), "70mbit");
+    }
+
+    #[test]
+    fn rate_limit_match_normalized() {
+        let rc = ClassRateCeil {
+            rate: "70Mbit".to_string(),
+            ceil: "70mbit".to_string(),
+        };
+        assert!(rate_ceil_matches_limit(&rc, "70mbit"));
+        assert!(!rate_ceil_matches_limit(&rc, "50mbit"));
+    }
 }