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); } }