diff --git a/README.md b/README.md index bfb69d5..a47c7e0 100644 --- a/README.md +++ b/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. diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 2ad11ae..d7dfc15 100644 --- a/docs/CONFIGURATION.md +++ b/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` — включает плановую перезагрузку. diff --git a/docs/MIGRATION.md b/docs/MIGRATION.md index d684dd0..36e3422 100644 --- a/docs/MIGRATION.md +++ b/docs/MIGRATION.md @@ -124,7 +124,7 @@ per-peer ingress filter: src peer_ip -> 2: 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, команда должна сообщить об этом явно, а не разрушать чужое состояние без подтверждения. diff --git a/traffic_core/src/firewall.rs b/traffic_core/src/firewall.rs index 7def823..d8441e7 100644 --- a/traffic_core/src/firewall.rs +++ b/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> { + 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> { 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(); diff --git a/traffic_core/src/netns.rs b/traffic_core/src/netns.rs index 0ffa2e1..ac210fe 100644 --- a/traffic_core/src/netns.rs +++ b/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 { 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 { + 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 { }) } } + +pub fn build_host_transport() -> Box { + Box::new(HostTransport) +} diff --git a/traffic_core/src/reboot.rs b/traffic_core/src/reboot.rs index b1034aa..93e32cd 100644 --- a/traffic_core/src/reboot.rs +++ b/traffic_core/src/reboot.rs @@ -66,7 +66,7 @@ fn next_reboot_time(start_hhmm: &str, end_hhmm: &str) -> anyhow::Result 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::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); + } +} diff --git a/traffic_core/src/status.rs b/traffic_core/src/status.rs index 4f59709..4063ae0 100644 --- a/traffic_core/src/status.rs +++ b/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!( diff --git a/traffic_core/src/traffic.rs b/traffic_core/src/traffic.rs index 5e46406..c89d49b 100644 --- a/traffic_core/src/traffic.rs +++ b/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, + pub egress_ceil: Option, + pub ingress_rate: Option, + pub ingress_ceil: Option, + pub drift_reason: Option, +} + pub fn inspect_drift_summary(cfg: &Config) -> anyhow::Result> { if !cfg.traffic.enabled { return Ok(None); @@ -39,6 +58,7 @@ pub fn inspect_drift_summary(cfg: &Config) -> anyhow::Result> { &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>> { + 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 { 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 { + 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")); + } }