diff --git a/docs/IMPLEMENTATION_SPEC.md b/docs/IMPLEMENTATION_SPEC.md index c8a8142..30010ac 100644 --- a/docs/IMPLEMENTATION_SPEC.md +++ b/docs/IMPLEMENTATION_SPEC.md @@ -82,8 +82,6 @@ awg2-harness validate-runtime ## 3. Структура модулей -Фактическая раскладка — плоская, без вложенных каталогов модулей: - ```text traffic_core/src/ ├── main.rs точка входа и диспетчер подкоманд @@ -94,8 +92,16 @@ traffic_core/src/ ├── netns.rs транспорты выполнения команд ├── iface.rs обнаружение интерфейса ├── wg.rs обнаружение peers и разбор вывода awg/wg -├── traffic.rs tc/IFB: инспекция, diff, teardown, apply -├── firewall.rs nftables: host firewall и anti-torrent +├── traffic/ +│ ├── mod.rs общие модели и точки входа +│ ├── parse_tc.rs чистые парсеры вывода tc/ip +│ ├── inspect.rs inspect_and_diff и контракт class id +│ └── apply.rs teardown и полная пересборка топологии +├── firewall/ +│ ├── mod.rs apply_firewall, выбор namespace, общие константы +│ ├── nft.rs нормализация текста nftables и обёртки команд +│ ├── host.rs таблица inet awg2_host_firewall (host namespace) +│ └── antitorrent.rs таблица inet awg2_harness (namespace по конфигу) ├── sysctl_tuning.rs управляемый sysctl baseline ├── swap.rs управляемый swap ├── status.rs сборка и печать статуса @@ -106,7 +112,11 @@ reboot_module/src/ └── main.rs планировщик перезагрузки ``` -Разбиение `traffic` и `firewall` на подкаталоги (`model.rs`, `inspect.rs`, `apply.rs`) обсуждалось, но не выполнено. Пока эти модули остаются одиночными файлами, любые изменения должны сохранять разделение на инспекцию, diff и apply внутри файла. +`traffic` разбит по ответственности: парсеры отдельно от инспекции, инспекция отдельно от apply. Чистые парсеры собраны в `parse_tc.rs` вместе со своими тестами — это тестируемая поверхность модуля, всё остальное требует живого контейнера. + +`firewall` разбит по подсистемам, а не по фазам: host firewall и anti-torrent независимы, применяются в разных namespace и владеют разными таблицами. Общий у них только слой работы с `nft`. + +Приватные типы объявлены в соответствующем `mod.rs`. В Rust приватные элементы видны в модуле объявления и всех дочерних модулях, поэтому подмодули работают с ними без расширения видимости до `pub(crate)`. ## 4. Модель данных diff --git a/traffic_core/src/firewall.rs b/traffic_core/src/firewall.rs deleted file mode 100644 index 5c62ea1..0000000 --- a/traffic_core/src/firewall.rs +++ /dev/null @@ -1,890 +0,0 @@ -use crate::{ - config::Config, - docker, - lock::FileLock, - netns::{NsTransport, build_host_transport, build_transport}, -}; -use serde_json::Value; -use std::{ - io::Write, - process::{Command, Stdio}, - time::Duration, -}; - -const ANTITORRENT_TABLE: &str = "awg2_harness"; -const HOST_FW_TABLE: &str = "awg2_host_firewall"; - -#[derive(Debug, Clone)] -struct FirewallActual { - table_exists: bool, - 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<()> { - let _lock = - FileLock::acquire_timeout("/run/awg2-harness/firewall.lock", Duration::from_secs(10))?; - - apply_host_firewall(cfg)?; - apply_antitorrent_firewall(cfg)?; - Ok(()) -} - -fn apply_antitorrent_firewall(cfg: &Config) -> anyhow::Result<()> { - if !cfg.anti_torrent.enabled { - apply_antitorrent_disabled(cfg)?; - return Ok(()); - } - - let transport = firewall_transport(cfg)?; - - let actual = inspect_firewall(transport.as_ref(), cfg)?; - if !firewall_has_drift(&actual) { - return Ok(()); - } - - run_allow_notfound( - transport.as_ref(), - &["nft", "delete", "table", "inet", ANTITORRENT_TABLE], - )?; - - run_step( - transport.as_ref(), - &["nft", "add", "table", "inet", ANTITORRENT_TABLE], - )?; - run_step( - transport.as_ref(), - &[ - "nft", - "add", - "counter", - "inet", - ANTITORRENT_TABLE, - "antitorrent_dropped_v4", - ], - )?; - run_step_chain(transport.as_ref())?; - apply_rule( - transport.as_ref(), - &cfg.anti_torrent.peer_ipv4_cidr, - cfg.anti_torrent.max_connections_per_peer, - )?; - - let verify = inspect_firewall(transport.as_ref(), cfg)?; - if firewall_has_drift(&verify) { - anyhow::bail!( - "anti-torrent firewall verify-after-apply failed: {}", - firewall_drift_reason(&verify) - ); - } - Ok(()) -} - -fn apply_antitorrent_disabled(cfg: &Config) -> anyhow::Result<()> { - if cfg.anti_torrent.apply_in_amnezia_netns { - match docker::discover_target(cfg) { - Ok(target) => { - let transport = build_transport(target.container_pid, &target.container_name); - run_allow_notfound( - transport.as_ref(), - &["nft", "delete", "table", "inet", ANTITORRENT_TABLE], - )?; - } - Err(_) => { - // container is absent now: nothing to remove in its netns - } - } - } else { - let transport = build_host_transport(); - run_allow_notfound( - transport.as_ref(), - &["nft", "delete", "table", "inet", ANTITORRENT_TABLE], - )?; - } - Ok(()) -} - -fn apply_host_firewall(cfg: &Config) -> anyhow::Result<()> { - let transport = build_host_transport(); - if !cfg.host_firewall.enabled { - run_allow_notfound( - transport.as_ref(), - &["nft", "delete", "table", "inet", HOST_FW_TABLE], - )?; - return Ok(()); - } - - if host_firewall_drift_reason(cfg)?.is_none() { - return Ok(()); - } - - let policy = cfg - .host_firewall - .default_input_policy - .trim() - .to_ascii_lowercase(); - - run_allow_notfound( - transport.as_ref(), - &["nft", "delete", "table", "inet", HOST_FW_TABLE], - )?; - - apply_host_firewall_batch(cfg, &policy)?; - - if let Some(reason) = host_firewall_drift_reason(cfg)? { - anyhow::bail!("host firewall verify-after-apply failed: {reason}"); - } - Ok(()) -} - -fn apply_host_firewall_batch(cfg: &Config, policy: &str) -> anyhow::Result<()> { - let batch = build_host_firewall_batch_script(cfg, policy); - run_nft_batch(&batch) -} - -fn build_host_firewall_batch_script(cfg: &Config, policy: &str) -> String { - let mut batch = String::new(); - batch.push_str(&format!("add table inet {HOST_FW_TABLE}\n")); - batch.push_str(&format!( - "add chain inet {HOST_FW_TABLE} input {{ type filter hook input priority filter; policy {policy}; }}\n" - )); - batch.push_str(&format!( - "add rule inet {HOST_FW_TABLE} input iif \"lo\" accept\n" - )); - batch.push_str(&format!( - "add rule inet {HOST_FW_TABLE} input ct state established,related accept\n" - )); - batch.push_str(&format!( - "add rule inet {HOST_FW_TABLE} input tcp dport {} accept\n", - cfg.host_firewall.ssh_port - )); - for proto in &cfg.host_firewall.amnezia_protocols { - let p = proto.trim().to_ascii_lowercase(); - batch.push_str(&format!( - "add rule inet {HOST_FW_TABLE} input {p} dport {} accept\n", - cfg.host_firewall.amnezia_port - )); - } - if cfg.host_firewall.allow_icmp { - batch.push_str(&format!( - "add rule inet {HOST_FW_TABLE} input ip protocol icmp accept\n" - )); - batch.push_str(&format!( - "add rule inet {HOST_FW_TABLE} input ip6 nexthdr icmpv6 accept\n" - )); - } - batch -} - -fn run_nft_batch(script: &str) -> anyhow::Result<()> { - let mut child = Command::new("nft") - .args(["-f", "-"]) - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn()?; - - if let Some(mut stdin) = child.stdin.take() { - stdin.write_all(script.as_bytes())?; - } else { - anyhow::bail!("failed to open stdin for nft batch"); - } - - let out = child.wait_with_output()?; - if !out.status.success() { - anyhow::bail!( - "nft batch apply failed: {}", - String::from_utf8_lossy(&out.stderr) - ); - } - Ok(()) -} - -fn inspect_firewall(transport: &dyn NsTransport, cfg: &Config) -> anyhow::Result { - let table = match transport.run(&["nft", "list", "table", "inet", ANTITORRENT_TABLE]) { - 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 nt = normalize_nft_text(&table); - let table_exists = !nt.trim().is_empty(); - let counter_exists = nt.contains("counter antitorrent_dropped_v4"); - let chain_exists = nt.contains("chain forward_antitorrent"); - let hook_forward_present = nt.contains("hook forward"); - let policy_accept_present = nt.contains("policy accept"); - let ct_state_new_present = nt.contains("ct state new"); - let meter_present = connlimit_state_present(&nt); - let counter_name_used = nt.contains("counter name antitorrent_dropped_v4"); - let drop_present = nt.contains(" drop"); - - let rule_matches = firewall_rule_contract_matches( - &table, - &cfg.anti_torrent.peer_ipv4_cidr, - cfg.anti_torrent.max_connections_per_peer, - ); - - Ok(FirewallActual { - table_exists, - counter_exists, - chain_exists, - rule_matches, - hook_forward_present, - policy_accept_present, - ct_state_new_present, - meter_present, - counter_name_used, - drop_present, - }) -} - -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()) - } -} - -pub fn antitorrent_drift_reason(cfg: &Config) -> anyhow::Result> { - if !cfg.anti_torrent.enabled { - return Ok(None); - } - let transport = firewall_transport(cfg)?; - let actual = inspect_firewall(transport.as_ref(), cfg)?; - if firewall_has_drift(&actual) { - return Ok(Some(firewall_drift_reason(&actual))); - } - Ok(None) -} - -pub fn host_firewall_drift_reason(cfg: &Config) -> anyhow::Result> { - if !cfg.host_firewall.enabled { - return Ok(None); - } - let transport = build_host_transport(); - let table = match transport.run(&["nft", "list", "table", "inet", HOST_FW_TABLE]) { - Ok(out) => out, - Err(e) if is_nft_not_found(&e.to_string()) => return Ok(Some("missing nft table".into())), - Err(e) => return Err(anyhow::anyhow!("host firewall inspect failed: {e}")), - }; - let nt = normalize_nft_text(&table); - Ok(host_firewall_drift_reason_from_text(&nt, cfg)) -} - -fn host_firewall_drift_reason_from_text(nt: &str, cfg: &Config) -> Option { - let ssh = format!("tcp dport {} accept", cfg.host_firewall.ssh_port); - if !nt.contains("chain input") { - return Some("missing input chain".into()); - } - if !nt.contains("type filter hook input") { - return Some("missing input hook".into()); - } - let pol = format!( - "policy {}", - cfg.host_firewall - .default_input_policy - .trim() - .to_ascii_lowercase() - ); - if !nt.contains(&pol) { - return Some("wrong input policy".into()); - } - if !nt.contains("iif lo accept") { - return Some("missing loopback allow".into()); - } - if !has_established_related_rule(nt) { - return Some("missing established/related allow".into()); - } - if !nt.contains(&ssh) { - return Some("missing ssh allow rule".into()); - } - for proto in &cfg.host_firewall.amnezia_protocols { - let p = proto.trim().to_ascii_lowercase(); - let need = format!("{p} dport {} accept", cfg.host_firewall.amnezia_port); - if !nt.contains(&need) { - return Some(format!("missing amnezia allow rule for {p}")); - } - } - if cfg.host_firewall.allow_icmp { - if !nt.contains("ip protocol icmp accept") { - return Some("missing icmp allow rule".into()); - } - if !(nt.contains("ip6 nexthdr icmpv6 accept") - || nt.contains("ip6 nexthdr ipv6-icmp accept")) - { - return Some("missing icmpv6 allow rule".into()); - } - } - None -} - -fn has_established_related_rule(nt: &str) -> bool { - nt.contains("ct state established related accept") - || nt.contains("ct state related established accept") -} - -fn apply_rule(transport: &dyn NsTransport, cidr: &str, max_conn: u32) -> anyhow::Result<()> { - let _ = transport.run(&[ - "nft", - "add", - "rule", - "inet", - ANTITORRENT_TABLE, - "forward_antitorrent", - "ip", - "saddr", - cidr, - "ct", - "state", - "new", - "meter", - "awg2_peer_connlimit_v4", - "{", - "ip", - "saddr", - "ct", - "count", - "over", - &max_conn.to_string(), - "}", - "counter", - "name", - "antitorrent_dropped_v4", - "drop", - ])?; - Ok(()) -} - -fn run_step(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> { - let _ = transport.run(args)?; - Ok(()) -} - -fn run_step_chain(transport: &dyn NsTransport) -> anyhow::Result<()> { - run_step( - transport, - &[ - "nft", - "add", - "chain", - "inet", - ANTITORRENT_TABLE, - "forward_antitorrent", - "{", - "type", - "filter", - "hook", - "forward", - "priority", - "filter;", - "policy", - "accept;", - "}", - ], - ) -} - -fn run_allow_notfound(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> { - match transport.run(args) { - Ok(_) => Ok(()), - Err(e) => { - let msg = e.to_string().to_ascii_lowercase(); - if msg.contains("no such file") - || msg.contains("no such") - || msg.contains("not found") - || msg.contains("not exist") - || msg.contains("no such table") - || msg.contains("no such device") - || msg.contains("cannot find") - { - return Ok(()); - } - Err(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/dynamic set".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 { - let table = normalize_nft_text(table); - 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") - && (table.contains("priority filter") || table.contains("priority 0")) - && table.contains("policy accept") - && table.contains(&needle_cidr) - && table.contains("ct state new") - && connlimit_state_present(&table) - && table.contains(&needle_max) - && table.contains("counter name antitorrent_dropped_v4") - && table.contains("drop") -} - -fn connlimit_state_present(table: &str) -> bool { - table.contains("meter awg2_peer_connlimit_v4") - || (table.contains("set awg2_peer_connlimit_v4") - && table.contains("flags dynamic") - && table.contains("add @awg2_peer_connlimit_v4")) -} - -fn normalize_nft_text(input: &str) -> String { - input - .replace(['\n', '\t', '{', '}', ';', ',', '"'], " ") - .split_whitespace() - .collect::>() - .join(" ") -} - -pub fn antitorrent_drop_counter(cfg: &Config) -> anyhow::Result> { - if !cfg.anti_torrent.enabled { - return Ok(None); - } - - let transport = firewall_transport(cfg)?; - let json = transport.run(&[ - "nft", - "-j", - "list", - "counter", - "inet", - ANTITORRENT_TABLE, - "antitorrent_dropped_v4", - ]); - - match json { - Ok(raw) => { - if let Some(v) = parse_counter_json(&raw)? { - Ok(Some(v)) - } else { - 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") - { - Ok(None) - } else { - Err(anyhow::anyhow!("nft counter inspection failed: {e}")) - } - } - } -} - -fn parse_counter_json(raw: &str) -> anyhow::Result> { - let v: Value = serde_json::from_str(raw)?; - let Some(arr) = v.get("nftables").and_then(|x| x.as_array()) else { - return Ok(None); - }; - for item in arr { - if let Some(counter) = item.get("counter") { - let packets = counter.get("packets").and_then(|x| x.as_u64()); - let bytes = counter.get("bytes").and_then(|x| x.as_u64()); - if let (Some(p), Some(b)) = (packets, bytes) { - return Ok(Some((p, b))); - } - } - } - Ok(None) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn parse_nft_counter_json() { - let raw = r#"{"nftables":[{"metainfo":{"json_schema_version":1}},{"counter":{"family":"inet","table":"awg2_harness","name":"antitorrent_dropped_v4","handle":2,"packets":7,"bytes":448}}]}"#; - 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)); - } - - #[test] - fn nft_text_parser_tolerates_priority_0_and_spacing() { - let table = r#" -table inet awg2_harness { - counter antitorrent_dropped_v4 - chain forward_antitorrent { - type filter hook forward priority 0; 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 nft_rule_contract_accepts_debian13_dynamic_set_rendering() { - let table = r#" -table inet awg2_harness { - counter antitorrent_dropped_v4 { - packets 0 bytes 0 - } - set awg2_peer_connlimit_v4 { - type ipv4_addr - size 65535 - flags dynamic - } - chain forward_antitorrent { - type filter hook forward priority filter; policy accept; - ip saddr 10.8.1.0/24 ct state new add @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 host_firewall_parser_tolerates_quoted_lo_and_ipv6_icmp_alias() { - let table = r#" -table inet awg2_host_firewall { - chain input { - type filter hook input priority filter; policy drop; - iif "lo" accept - ct state established,related accept - tcp dport 2323 accept - udp dport 443 accept - ip protocol icmp accept - ip6 nexthdr ipv6-icmp accept - } -} -"#; - let cfg = Config { - amnezia: crate::config::AmneziaConfig { - container_names: vec!["amnezia-awg2".into()], - interface_candidates: vec!["awg0".into()], - interface_patterns: vec!["awg*".into()], - peer_ipv4_cidr: "10.8.1.0/24".into(), - wg_binaries: vec!["awg".into(), "wg".into()], - }, - traffic: crate::config::TrafficConfig { - enabled: true, - peer_limit: "70mbit".into(), - root_rate: "1000mbit".into(), - reconcile_interval_secs: 15, - container_ready_timeout_secs: 30, - class_start_id: 10, - ifb_name: "ifb0".into(), - }, - sysctl_tuning: crate::config::SysctlTuningConfig::default(), - host_firewall: crate::config::HostFirewallConfig::default(), - anti_torrent: crate::config::AntiTorrentConfig { - enabled: false, - max_connections_per_peer: 100, - peer_ipv4_cidr: "10.8.1.0/24".into(), - apply_in_amnezia_netns: false, - }, - reboot: crate::config::RebootConfig { - enabled: false, - window_start_utc: "01:00".into(), - window_end_utc: "02:00".into(), - }, - swap: crate::config::SwapConfig { - enabled: false, - path: "/swapfile".into(), - size_mb: 2048, - priority: 1, - swappiness: 1, - }, - }; - let nt = normalize_nft_text(table); - assert_eq!(host_firewall_drift_reason_from_text(&nt, &cfg), None); - } - - #[test] - fn normalized_established_related_rule_matches() { - let nt = normalize_nft_text("ct state established,related accept"); - assert!(nt.contains("ct state established related accept")); - assert!(has_established_related_rule(&nt)); - } - - #[test] - fn host_firewall_drift_detects_missing_established_related_rule() { - let table = r#" -table inet awg2_host_firewall { - chain input { - type filter hook input priority filter; policy drop; - iif "lo" accept - tcp dport 2323 accept - udp dport 443 accept - ip protocol icmp accept - ip6 nexthdr icmpv6 accept - } -} -"#; - let cfg = Config { - amnezia: crate::config::AmneziaConfig { - container_names: vec!["amnezia-awg2".into()], - interface_candidates: vec!["awg0".into()], - interface_patterns: vec!["awg*".into()], - peer_ipv4_cidr: "10.8.1.0/24".into(), - wg_binaries: vec!["awg".into(), "wg".into()], - }, - traffic: crate::config::TrafficConfig { - enabled: true, - peer_limit: "70mbit".into(), - root_rate: "1000mbit".into(), - reconcile_interval_secs: 15, - container_ready_timeout_secs: 30, - class_start_id: 10, - ifb_name: "ifb0".into(), - }, - sysctl_tuning: crate::config::SysctlTuningConfig::default(), - host_firewall: crate::config::HostFirewallConfig::default(), - anti_torrent: crate::config::AntiTorrentConfig { - enabled: false, - max_connections_per_peer: 100, - peer_ipv4_cidr: "10.8.1.0/24".into(), - apply_in_amnezia_netns: false, - }, - reboot: crate::config::RebootConfig { - enabled: false, - window_start_utc: "01:00".into(), - window_end_utc: "02:00".into(), - }, - swap: crate::config::SwapConfig { - enabled: false, - path: "/swapfile".into(), - size_mb: 2048, - priority: 1, - swappiness: 1, - }, - }; - let nt = normalize_nft_text(table); - assert_eq!( - host_firewall_drift_reason_from_text(&nt, &cfg), - Some("missing established/related allow".into()) - ); - } - - #[test] - fn host_firewall_batch_script_does_not_include_delete_table() { - let cfg = Config { - amnezia: crate::config::AmneziaConfig { - container_names: vec!["amnezia-awg2".into()], - interface_candidates: vec!["awg0".into()], - interface_patterns: vec!["awg*".into()], - peer_ipv4_cidr: "10.8.1.0/24".into(), - wg_binaries: vec!["awg".into(), "wg".into()], - }, - traffic: crate::config::TrafficConfig { - enabled: true, - peer_limit: "70mbit".into(), - root_rate: "1000mbit".into(), - reconcile_interval_secs: 15, - container_ready_timeout_secs: 30, - class_start_id: 10, - ifb_name: "ifb0".into(), - }, - sysctl_tuning: crate::config::SysctlTuningConfig::default(), - host_firewall: crate::config::HostFirewallConfig::default(), - anti_torrent: crate::config::AntiTorrentConfig { - enabled: false, - max_connections_per_peer: 100, - peer_ipv4_cidr: "10.8.1.0/24".into(), - apply_in_amnezia_netns: false, - }, - reboot: crate::config::RebootConfig { - enabled: false, - window_start_utc: "01:00".into(), - window_end_utc: "02:00".into(), - }, - swap: crate::config::SwapConfig { - enabled: false, - path: "/swapfile".into(), - size_mb: 2048, - priority: 1, - swappiness: 1, - }, - }; - - let script = build_host_firewall_batch_script(&cfg, "drop"); - assert!(script.contains("add table inet awg2_host_firewall")); - assert!(!script.contains("delete table inet awg2_host_firewall")); - } - - #[test] - fn host_firewall_policy_drift_check_trims_policy_value() { - let table = r#" -table inet awg2_host_firewall { - chain input { - type filter hook input priority filter; policy drop; - iif "lo" accept - ct state established,related accept - tcp dport 2323 accept - udp dport 443 accept - ip protocol icmp accept - ip6 nexthdr icmpv6 accept - } -} -"#; - let mut cfg = Config { - amnezia: crate::config::AmneziaConfig { - container_names: vec!["amnezia-awg2".into()], - interface_candidates: vec!["awg0".into()], - interface_patterns: vec!["awg*".into()], - peer_ipv4_cidr: "10.8.1.0/24".into(), - wg_binaries: vec!["awg".into(), "wg".into()], - }, - traffic: crate::config::TrafficConfig { - enabled: true, - peer_limit: "70mbit".into(), - root_rate: "1000mbit".into(), - reconcile_interval_secs: 15, - container_ready_timeout_secs: 30, - class_start_id: 10, - ifb_name: "ifb0".into(), - }, - sysctl_tuning: crate::config::SysctlTuningConfig::default(), - host_firewall: crate::config::HostFirewallConfig::default(), - anti_torrent: crate::config::AntiTorrentConfig { - enabled: false, - max_connections_per_peer: 100, - peer_ipv4_cidr: "10.8.1.0/24".into(), - apply_in_amnezia_netns: false, - }, - reboot: crate::config::RebootConfig { - enabled: false, - window_start_utc: "01:00".into(), - window_end_utc: "02:00".into(), - }, - swap: crate::config::SwapConfig { - enabled: false, - path: "/swapfile".into(), - size_mb: 2048, - priority: 1, - swappiness: 1, - }, - }; - cfg.host_firewall.default_input_policy = " drop ".into(); - let nt = normalize_nft_text(table); - assert_eq!(host_firewall_drift_reason_from_text(&nt, &cfg), None); - } -} diff --git a/traffic_core/src/firewall/antitorrent.rs b/traffic_core/src/firewall/antitorrent.rs new file mode 100644 index 0000000..3f7ee23 --- /dev/null +++ b/traffic_core/src/firewall/antitorrent.rs @@ -0,0 +1,450 @@ +//! Anti-torrent connection-count guard, owned table `inet awg2_harness`. +//! +//! Limits concurrent conntrack flows per VPN peer IP. Not torrent protocol detection: a +//! generic per-source guard that makes high-peer-count P2P unattractive while leaving +//! browsing, streaming, gaming and messaging untouched. +//! +//! Applied in the namespace selected by `anti_torrent.apply_in_amnezia_netns`. The default +//! is the Amnezia container namespace, because if Amnezia NATs inside the container then +//! the host bridge never sees the `10.8.1.x` source addresses the rule keys on. + +use serde_json::Value; + +use crate::{config::Config, docker, netns::NsTransport}; + +use super::{ + ANTITORRENT_TABLE, firewall_transport, + nft::{is_nft_not_found, normalize_nft_text, run_allow_notfound, run_step}, +}; + +#[derive(Debug, Clone)] +struct FirewallActual { + table_exists: bool, + 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(super) fn apply_antitorrent_firewall(cfg: &Config) -> anyhow::Result<()> { + if !cfg.anti_torrent.enabled { + apply_antitorrent_disabled(cfg)?; + return Ok(()); + } + + let transport = firewall_transport(cfg)?; + + let actual = inspect_firewall(transport.as_ref(), cfg)?; + if !firewall_has_drift(&actual) { + return Ok(()); + } + + run_allow_notfound( + transport.as_ref(), + &["nft", "delete", "table", "inet", ANTITORRENT_TABLE], + )?; + + run_step( + transport.as_ref(), + &["nft", "add", "table", "inet", ANTITORRENT_TABLE], + )?; + run_step( + transport.as_ref(), + &[ + "nft", + "add", + "counter", + "inet", + ANTITORRENT_TABLE, + "antitorrent_dropped_v4", + ], + )?; + run_step_chain(transport.as_ref())?; + apply_rule( + transport.as_ref(), + &cfg.anti_torrent.peer_ipv4_cidr, + cfg.anti_torrent.max_connections_per_peer, + )?; + + let verify = inspect_firewall(transport.as_ref(), cfg)?; + if firewall_has_drift(&verify) { + anyhow::bail!( + "anti-torrent firewall verify-after-apply failed: {}", + firewall_drift_reason(&verify) + ); + } + Ok(()) +} + +fn apply_antitorrent_disabled(cfg: &Config) -> anyhow::Result<()> { + if cfg.anti_torrent.apply_in_amnezia_netns { + match docker::discover_target(cfg) { + Ok(target) => { + let transport = + crate::netns::build_transport(target.container_pid, &target.container_name); + run_allow_notfound( + transport.as_ref(), + &["nft", "delete", "table", "inet", ANTITORRENT_TABLE], + )?; + } + Err(_) => { + // container is absent now: nothing to remove in its netns + } + } + } else { + let transport = crate::netns::build_host_transport(); + run_allow_notfound( + transport.as_ref(), + &["nft", "delete", "table", "inet", ANTITORRENT_TABLE], + )?; + } + Ok(()) +} + +fn inspect_firewall(transport: &dyn NsTransport, cfg: &Config) -> anyhow::Result { + let table = match transport.run(&["nft", "list", "table", "inet", ANTITORRENT_TABLE]) { + 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 nt = normalize_nft_text(&table); + let table_exists = !nt.trim().is_empty(); + let counter_exists = nt.contains("counter antitorrent_dropped_v4"); + let chain_exists = nt.contains("chain forward_antitorrent"); + let hook_forward_present = nt.contains("hook forward"); + let policy_accept_present = nt.contains("policy accept"); + let ct_state_new_present = nt.contains("ct state new"); + let meter_present = connlimit_state_present(&nt); + let counter_name_used = nt.contains("counter name antitorrent_dropped_v4"); + let drop_present = nt.contains(" drop"); + + let rule_matches = firewall_rule_contract_matches( + &table, + &cfg.anti_torrent.peer_ipv4_cidr, + cfg.anti_torrent.max_connections_per_peer, + ); + + Ok(FirewallActual { + table_exists, + counter_exists, + chain_exists, + rule_matches, + hook_forward_present, + policy_accept_present, + ct_state_new_present, + meter_present, + counter_name_used, + drop_present, + }) +} + +pub fn antitorrent_drift_reason(cfg: &Config) -> anyhow::Result> { + if !cfg.anti_torrent.enabled { + return Ok(None); + } + let transport = firewall_transport(cfg)?; + let actual = inspect_firewall(transport.as_ref(), cfg)?; + if firewall_has_drift(&actual) { + return Ok(Some(firewall_drift_reason(&actual))); + } + Ok(None) +} + +pub fn antitorrent_drop_counter(cfg: &Config) -> anyhow::Result> { + if !cfg.anti_torrent.enabled { + return Ok(None); + } + + let transport = firewall_transport(cfg)?; + let json = transport.run(&[ + "nft", + "-j", + "list", + "counter", + "inet", + ANTITORRENT_TABLE, + "antitorrent_dropped_v4", + ]); + + match json { + Ok(raw) => { + if let Some(v) = parse_counter_json(&raw)? { + Ok(Some(v)) + } else { + 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") + { + Ok(None) + } else { + Err(anyhow::anyhow!("nft counter inspection failed: {e}")) + } + } + } +} + +/// `ct state new` is load-bearing: without it the threshold would start dropping packets +/// of already established flows as soon as the counter is high, instead of only refusing +/// new ones. +fn apply_rule(transport: &dyn NsTransport, cidr: &str, max_conn: u32) -> anyhow::Result<()> { + let _ = transport.run(&[ + "nft", + "add", + "rule", + "inet", + ANTITORRENT_TABLE, + "forward_antitorrent", + "ip", + "saddr", + cidr, + "ct", + "state", + "new", + "meter", + "awg2_peer_connlimit_v4", + "{", + "ip", + "saddr", + "ct", + "count", + "over", + &max_conn.to_string(), + "}", + "counter", + "name", + "antitorrent_dropped_v4", + "drop", + ])?; + Ok(()) +} + +fn run_step_chain(transport: &dyn NsTransport) -> anyhow::Result<()> { + run_step( + transport, + &[ + "nft", + "add", + "chain", + "inet", + ANTITORRENT_TABLE, + "forward_antitorrent", + "{", + "type", + "filter", + "hook", + "forward", + "priority", + "filter;", + "policy", + "accept;", + "}", + ], + ) +} + +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) +} + +/// Ordered most structural first, so the reported reason names the outermost missing +/// piece rather than a symptom of it. +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/dynamic set".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 { + let table = normalize_nft_text(table); + 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") + && (table.contains("priority filter") || table.contains("priority 0")) + && table.contains("policy accept") + && table.contains(&needle_cidr) + && table.contains("ct state new") + && connlimit_state_present(&table) + && table.contains(&needle_max) + && table.contains("counter name antitorrent_dropped_v4") + && table.contains("drop") +} + +/// Debian 13 renders the same rule either as an inline `meter` or as a named dynamic set +/// plus an `add @set` statement. Both are the rule we applied. +fn connlimit_state_present(table: &str) -> bool { + table.contains("meter awg2_peer_connlimit_v4") + || (table.contains("set awg2_peer_connlimit_v4") + && table.contains("flags dynamic") + && table.contains("add @awg2_peer_connlimit_v4")) +} + +fn parse_counter_json(raw: &str) -> anyhow::Result> { + let v: Value = serde_json::from_str(raw)?; + let Some(arr) = v.get("nftables").and_then(|x| x.as_array()) else { + return Ok(None); + }; + for item in arr { + if let Some(counter) = item.get("counter") { + let packets = counter.get("packets").and_then(|x| x.as_u64()); + let bytes = counter.get("bytes").and_then(|x| x.as_u64()); + if let (Some(p), Some(b)) = (packets, bytes) { + return Ok(Some((p, b))); + } + } + } + Ok(None) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_nft_counter_json() { + let raw = r#"{"nftables":[{"metainfo":{"json_schema_version":1}},{"counter":{"family":"inet","table":"awg2_harness","name":"antitorrent_dropped_v4","handle":2,"packets":7,"bytes":448}}]}"#; + 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)); + } + + #[test] + fn nft_text_parser_tolerates_priority_0_and_spacing() { + let table = r#" +table inet awg2_harness { + counter antitorrent_dropped_v4 + chain forward_antitorrent { + type filter hook forward priority 0; 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 nft_rule_contract_accepts_debian13_dynamic_set_rendering() { + let table = r#" +table inet awg2_harness { + counter antitorrent_dropped_v4 { + packets 0 bytes 0 + } + set awg2_peer_connlimit_v4 { + type ipv4_addr + size 65535 + flags dynamic + } + chain forward_antitorrent { + type filter hook forward priority filter; policy accept; + ip saddr 10.8.1.0/24 ct state new add @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)); + } +} diff --git a/traffic_core/src/firewall/host.rs b/traffic_core/src/firewall/host.rs new file mode 100644 index 0000000..1c069b3 --- /dev/null +++ b/traffic_core/src/firewall/host.rs @@ -0,0 +1,242 @@ +//! Host INPUT firewall, owned table `inet awg2_host_firewall`. +//! +//! Applied in the host namespace only. The table is scoped so that disabling the module +//! removes exactly what it created and nothing that Docker, Amnezia or the operator owns. + +use crate::{config::Config, netns::build_host_transport}; + +use super::{ + HOST_FW_TABLE, + nft::{is_nft_not_found, normalize_nft_text, run_allow_notfound, run_nft_batch}, +}; + +pub(super) fn apply_host_firewall(cfg: &Config) -> anyhow::Result<()> { + let transport = build_host_transport(); + if !cfg.host_firewall.enabled { + run_allow_notfound( + transport.as_ref(), + &["nft", "delete", "table", "inet", HOST_FW_TABLE], + )?; + return Ok(()); + } + + if host_firewall_drift_reason(cfg)?.is_none() { + return Ok(()); + } + + let policy = cfg + .host_firewall + .default_input_policy + .trim() + .to_ascii_lowercase(); + + run_allow_notfound( + transport.as_ref(), + &["nft", "delete", "table", "inet", HOST_FW_TABLE], + )?; + + apply_host_firewall_batch(cfg, &policy)?; + + if let Some(reason) = host_firewall_drift_reason(cfg)? { + anyhow::bail!("host firewall verify-after-apply failed: {reason}"); + } + Ok(()) +} + +fn apply_host_firewall_batch(cfg: &Config, policy: &str) -> anyhow::Result<()> { + let batch = build_host_firewall_batch_script(cfg, policy); + run_nft_batch(&batch) +} + +/// Renders the desired INPUT chain. +/// +/// The script never contains `delete table`: deletion happens as a separate step before +/// the batch, so a batch that fails to parse cannot take the existing rules down with it. +fn build_host_firewall_batch_script(cfg: &Config, policy: &str) -> String { + let mut batch = String::new(); + batch.push_str(&format!("add table inet {HOST_FW_TABLE}\n")); + batch.push_str(&format!( + "add chain inet {HOST_FW_TABLE} input {{ type filter hook input priority filter; policy {policy}; }}\n" + )); + batch.push_str(&format!( + "add rule inet {HOST_FW_TABLE} input iif \"lo\" accept\n" + )); + batch.push_str(&format!( + "add rule inet {HOST_FW_TABLE} input ct state established,related accept\n" + )); + batch.push_str(&format!( + "add rule inet {HOST_FW_TABLE} input tcp dport {} accept\n", + cfg.host_firewall.ssh_port + )); + for proto in &cfg.host_firewall.amnezia_protocols { + let p = proto.trim().to_ascii_lowercase(); + batch.push_str(&format!( + "add rule inet {HOST_FW_TABLE} input {p} dport {} accept\n", + cfg.host_firewall.amnezia_port + )); + } + if cfg.host_firewall.allow_icmp { + batch.push_str(&format!( + "add rule inet {HOST_FW_TABLE} input ip protocol icmp accept\n" + )); + batch.push_str(&format!( + "add rule inet {HOST_FW_TABLE} input ip6 nexthdr icmpv6 accept\n" + )); + } + batch +} + +pub fn host_firewall_drift_reason(cfg: &Config) -> anyhow::Result> { + if !cfg.host_firewall.enabled { + return Ok(None); + } + let transport = build_host_transport(); + let table = match transport.run(&["nft", "list", "table", "inet", HOST_FW_TABLE]) { + Ok(out) => out, + Err(e) if is_nft_not_found(&e.to_string()) => return Ok(Some("missing nft table".into())), + Err(e) => return Err(anyhow::anyhow!("host firewall inspect failed: {e}")), + }; + let nt = normalize_nft_text(&table); + Ok(host_firewall_drift_reason_from_text(&nt, cfg)) +} + +/// Checks the normalized ruleset against config. Returns the first missing piece. +/// +/// SSH is checked before anything optional: locking the operator out is the one failure +/// mode of this module that cannot be repaired remotely. +fn host_firewall_drift_reason_from_text(nt: &str, cfg: &Config) -> Option { + let ssh = format!("tcp dport {} accept", cfg.host_firewall.ssh_port); + if !nt.contains("chain input") { + return Some("missing input chain".into()); + } + if !nt.contains("type filter hook input") { + return Some("missing input hook".into()); + } + let pol = format!( + "policy {}", + cfg.host_firewall + .default_input_policy + .trim() + .to_ascii_lowercase() + ); + if !nt.contains(&pol) { + return Some("wrong input policy".into()); + } + if !nt.contains("iif lo accept") { + return Some("missing loopback allow".into()); + } + if !has_established_related_rule(nt) { + return Some("missing established/related allow".into()); + } + if !nt.contains(&ssh) { + return Some("missing ssh allow rule".into()); + } + for proto in &cfg.host_firewall.amnezia_protocols { + let p = proto.trim().to_ascii_lowercase(); + let need = format!("{p} dport {} accept", cfg.host_firewall.amnezia_port); + if !nt.contains(&need) { + return Some(format!("missing amnezia allow rule for {p}")); + } + } + if cfg.host_firewall.allow_icmp { + if !nt.contains("ip protocol icmp accept") { + return Some("missing icmp allow rule".into()); + } + if !(nt.contains("ip6 nexthdr icmpv6 accept") + || nt.contains("ip6 nexthdr ipv6-icmp accept")) + { + return Some("missing icmpv6 allow rule".into()); + } + } + None +} + +/// `nft` renders the state list in either order depending on version. +fn has_established_related_rule(nt: &str) -> bool { + nt.contains("ct state established related accept") + || nt.contains("ct state related established accept") +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::firewall::tests::test_config; + + #[test] + fn host_firewall_parser_tolerates_quoted_lo_and_ipv6_icmp_alias() { + let table = r#" +table inet awg2_host_firewall { + chain input { + type filter hook input priority filter; policy drop; + iif "lo" accept + ct state established,related accept + tcp dport 2323 accept + udp dport 443 accept + ip protocol icmp accept + ip6 nexthdr ipv6-icmp accept + } +} +"#; + let cfg = test_config(); + let nt = normalize_nft_text(table); + assert_eq!(host_firewall_drift_reason_from_text(&nt, &cfg), None); + } + + #[test] + fn normalized_established_related_rule_matches() { + let nt = normalize_nft_text("ct state established,related accept"); + assert!(nt.contains("ct state established related accept")); + assert!(has_established_related_rule(&nt)); + } + + #[test] + fn host_firewall_drift_detects_missing_established_related_rule() { + let table = r#" +table inet awg2_host_firewall { + chain input { + type filter hook input priority filter; policy drop; + iif "lo" accept + tcp dport 2323 accept + udp dport 443 accept + ip protocol icmp accept + ip6 nexthdr icmpv6 accept + } +} +"#; + let cfg = test_config(); + let nt = normalize_nft_text(table); + assert_eq!( + host_firewall_drift_reason_from_text(&nt, &cfg), + Some("missing established/related allow".into()) + ); + } + + #[test] + fn host_firewall_batch_script_does_not_include_delete_table() { + let cfg = test_config(); + let script = build_host_firewall_batch_script(&cfg, "drop"); + assert!(script.contains("add table inet awg2_host_firewall")); + assert!(!script.contains("delete table inet awg2_host_firewall")); + } + + #[test] + fn host_firewall_policy_drift_check_trims_policy_value() { + let table = r#" +table inet awg2_host_firewall { + chain input { + type filter hook input priority filter; policy drop; + iif "lo" accept + ct state established,related accept + tcp dport 2323 accept + udp dport 443 accept + ip protocol icmp accept + ip6 nexthdr icmpv6 accept + } +} +"#; + let mut cfg = test_config(); + cfg.host_firewall.default_input_policy = " drop ".into(); + let nt = normalize_nft_text(table); + assert_eq!(host_firewall_drift_reason_from_text(&nt, &cfg), None); + } +} diff --git a/traffic_core/src/firewall/mod.rs b/traffic_core/src/firewall/mod.rs new file mode 100644 index 0000000..e3073d6 --- /dev/null +++ b/traffic_core/src/firewall/mod.rs @@ -0,0 +1,99 @@ +//! nftables state owned by AWG2 Harness. +//! +//! Two independent subsystems live here and are deliberately kept apart: +//! +//! - [`host`] owns `inet awg2_host_firewall` in the host namespace (INPUT policy); +//! - [`antitorrent`] owns `inet awg2_harness` in the namespace selected by config. +//! +//! Neither ever touches the global ruleset, host `FORWARD`, or Docker NAT chains. + +mod antitorrent; +mod host; +mod nft; + +use std::time::Duration; + +use crate::{ + config::Config, + docker, + lock::FileLock, + netns::{NsTransport, build_host_transport, build_transport}, +}; + +pub use antitorrent::{antitorrent_drift_reason, antitorrent_drop_counter}; +pub use host::host_firewall_drift_reason; + +const ANTITORRENT_TABLE: &str = "awg2_harness"; +const HOST_FW_TABLE: &str = "awg2_host_firewall"; + +pub fn apply_firewall(cfg: &Config) -> anyhow::Result<()> { + let _lock = + FileLock::acquire_timeout("/run/awg2-harness/firewall.lock", Duration::from_secs(10))?; + + host::apply_host_firewall(cfg)?; + antitorrent::apply_antitorrent_firewall(cfg)?; + Ok(()) +} + +/// Anti-torrent rules go wherever the peer source addresses are actually visible. +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()) + } +} + +#[cfg(test)] +pub(crate) mod tests { + use crate::config::{ + AmneziaConfig, AntiTorrentConfig, Config, HostFirewallConfig, RebootConfig, SwapConfig, + SysctlTuningConfig, TrafficConfig, + }; + + /// Baseline config for firewall tests. Mutate the one field a test is about. + pub(crate) fn test_config() -> Config { + Config { + amnezia: AmneziaConfig { + container_names: vec!["amnezia-awg2".into()], + interface_candidates: vec!["awg0".into()], + interface_patterns: vec!["awg*".into()], + peer_ipv4_cidr: "10.8.1.0/24".into(), + wg_binaries: vec!["awg".into(), "wg".into()], + }, + traffic: TrafficConfig { + enabled: true, + peer_limit: "70mbit".into(), + root_rate: "1000mbit".into(), + reconcile_interval_secs: 15, + container_ready_timeout_secs: 30, + class_start_id: 10, + ifb_name: "ifb0".into(), + }, + sysctl_tuning: SysctlTuningConfig::default(), + host_firewall: HostFirewallConfig::default(), + anti_torrent: AntiTorrentConfig { + enabled: false, + max_connections_per_peer: 100, + peer_ipv4_cidr: "10.8.1.0/24".into(), + apply_in_amnezia_netns: false, + }, + reboot: RebootConfig { + enabled: false, + window_start_utc: "01:00".into(), + window_end_utc: "02:00".into(), + }, + swap: SwapConfig { + enabled: false, + path: "/swapfile".into(), + size_mb: 2048, + priority: 1, + swappiness: 1, + }, + } + } +} diff --git a/traffic_core/src/firewall/nft.rs b/traffic_core/src/firewall/nft.rs new file mode 100644 index 0000000..cec69fd --- /dev/null +++ b/traffic_core/src/firewall/nft.rs @@ -0,0 +1,84 @@ +//! Shared nftables plumbing: text normalization and command wrappers. +//! +//! Ruleset text is compared after normalization because `nft list` rendering differs +//! between versions and between kernel/userspace paths - quoting, commas inside +//! `ct state established,related`, and brace/newline layout are all unstable. Matching on +//! raw output would report drift on a correct ruleset. + +use std::{ + io::Write, + process::{Command, Stdio}, +}; + +use crate::netns::NsTransport; + +pub(super) fn normalize_nft_text(input: &str) -> String { + input + .replace(['\n', '\t', '{', '}', ';', ',', '"'], " ") + .split_whitespace() + .collect::>() + .join(" ") +} + +pub(super) 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") +} + +pub(super) fn run_step(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> { + let _ = transport.run(args)?; + Ok(()) +} + +/// Deleting a table that is already absent is the expected outcome when the subsystem is +/// disabled or being rebuilt. Every other failure propagates. +pub(super) fn run_allow_notfound(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> { + match transport.run(args) { + Ok(_) => Ok(()), + Err(e) => { + let msg = e.to_string().to_ascii_lowercase(); + if msg.contains("no such file") + || msg.contains("no such") + || msg.contains("not found") + || msg.contains("not exist") + || msg.contains("no such table") + || msg.contains("no such device") + || msg.contains("cannot find") + { + return Ok(()); + } + Err(e) + } + } +} + +/// Applies a ruleset through `nft -f -`. +/// +/// A batch is atomic: either the whole script lands or nothing does. That matters for the +/// host firewall, where a partially applied INPUT chain could lock out SSH. +pub(super) fn run_nft_batch(script: &str) -> anyhow::Result<()> { + let mut child = Command::new("nft") + .args(["-f", "-"]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn()?; + + if let Some(mut stdin) = child.stdin.take() { + stdin.write_all(script.as_bytes())?; + } else { + anyhow::bail!("failed to open stdin for nft batch"); + } + + let out = child.wait_with_output()?; + if !out.status.success() { + anyhow::bail!( + "nft batch apply failed: {}", + String::from_utf8_lossy(&out.stderr) + ); + } + Ok(()) +}