diff --git a/src/api/model.rs b/src/api/model.rs index 3c7497d..127c348 100644 --- a/src/api/model.rs +++ b/src/api/model.rs @@ -124,6 +124,10 @@ pub(super) struct ZeroCoreData { pub(super) conntrack_pressure_active: bool, pub(super) conntrack_event_queue_depth: u64, pub(super) conntrack_rule_apply_ok: bool, + pub(super) conntrack_rule_reconcile_success_total: u64, + pub(super) conntrack_rule_reconcile_error_total: u64, + pub(super) conntrack_rule_rollback_success_total: u64, + pub(super) conntrack_rule_rollback_error_total: u64, pub(super) conntrack_delete_attempt_total: u64, pub(super) conntrack_delete_success_total: u64, pub(super) conntrack_delete_not_found_total: u64, diff --git a/src/api/runtime_stats.rs b/src/api/runtime_stats.rs index d9536f1..77885d0 100644 --- a/src/api/runtime_stats.rs +++ b/src/api/runtime_stats.rs @@ -61,6 +61,14 @@ pub(super) fn build_zero_all_data(stats: &Stats, configured_users: usize) -> Zer conntrack_pressure_active: stats.get_conntrack_pressure_active(), conntrack_event_queue_depth: stats.get_conntrack_event_queue_depth(), conntrack_rule_apply_ok: stats.get_conntrack_rule_apply_ok(), + conntrack_rule_reconcile_success_total: stats + .get_conntrack_rule_reconcile_success_total(), + conntrack_rule_reconcile_error_total: stats + .get_conntrack_rule_reconcile_error_total(), + conntrack_rule_rollback_success_total: stats + .get_conntrack_rule_rollback_success_total(), + conntrack_rule_rollback_error_total: stats + .get_conntrack_rule_rollback_error_total(), conntrack_delete_attempt_total: stats.get_conntrack_delete_attempt_total(), conntrack_delete_success_total: stats.get_conntrack_delete_success_total(), conntrack_delete_not_found_total: stats.get_conntrack_delete_not_found_total(), diff --git a/src/conntrack_control.rs b/src/conntrack_control.rs index 275e75d..4204df0 100644 --- a/src/conntrack_control.rs +++ b/src/conntrack_control.rs @@ -15,8 +15,8 @@ mod firewall; use firewall::{ DeleteOutcome, delete_conntrack_entry, effective_conntrack_enabled, probe_runtime_support, - reconcile_rules, }; +pub(crate) use firewall::FirewallAuthority; const CONNTRACK_EVENT_QUEUE_CAPACITY: usize = 32_768; const PRESSURE_RELEASE_TICKS: u8 = 3; @@ -115,7 +115,6 @@ async fn run_conntrack_controller_worker( runtime_support, false, ); - reconcile_rules(&cfg, runtime_support, stats.as_ref()).await; loop { tokio::select! { @@ -129,7 +128,6 @@ async fn run_conntrack_controller_worker( effective_enabled = effective_conntrack_enabled(&cfg, runtime_support); delete_budget_tokens = cfg.server.conntrack_control.delete_budget_per_sec; apply_runtime_state(stats.as_ref(), shared.as_ref(), &cfg, runtime_support, pressure_state.active); - reconcile_rules(&cfg, runtime_support, stats.as_ref()).await; } event = close_rx.recv() => { let Some(event) = event else { diff --git a/src/conntrack_control/firewall.rs b/src/conntrack_control/firewall.rs index 2fc42f7..f8a6a3a 100644 --- a/src/conntrack_control/firewall.rs +++ b/src/conntrack_control/firewall.rs @@ -1,434 +1,18 @@ -use std::collections::BTreeSet; -use std::net::IpAddr; -use std::time::Duration; +// Process-owned firewall reconciliation and privileged conntrack helpers. -use tokio::io::AsyncWriteExt; -use tokio::process::Command; -use tracing::{debug, warn}; +mod actor; +mod command; +mod iptables; +mod model; +mod nftables; +mod runtime; +mod transaction; -use crate::config::{ConntrackBackend, ConntrackMode, ProxyConfig}; -use crate::proxy::shared_state::ConntrackCloseEvent; -use crate::stats::Stats; -#[cfg(unix)] -use crate::util::trusted_command::resolve_trusted_helper; +pub(crate) use actor::FirewallAuthority; +pub(super) use runtime::{ + DeleteOutcome, delete_conntrack_entry, effective_conntrack_enabled, probe_runtime_support, +}; -use super::{ConntrackRuntimeSupport, NetfilterBackend}; - -/// Reconciles kernel NOTRACK rules with the active listener policy. -pub(super) async fn reconcile_rules( - cfg: &ProxyConfig, - runtime_support: ConntrackRuntimeSupport, - stats: &Stats, -) { - if !cfg.server.conntrack_control.inline_conntrack_control { - clear_notrack_rules_all_backends().await; - stats.set_conntrack_rule_apply_ok(true); - return; - } - - if !effective_conntrack_enabled(cfg, runtime_support) { - clear_notrack_rules_all_backends().await; - stats.set_conntrack_rule_apply_ok(false); - return; - } - - let backend = runtime_support - .netfilter_backend - .expect("netfilter backend must be available for effective conntrack control"); - let apply_result = match backend { - NetfilterBackend::Nftables => apply_nft_rules(cfg).await, - NetfilterBackend::Iptables => apply_iptables_rules(cfg).await, - }; - if let Err(error) = apply_result { - warn!(error = %error, "Failed to reconcile conntrack/notrack rules"); - stats.set_conntrack_rule_apply_ok(false); - } else { - stats.set_conntrack_rule_apply_ok(true); - } -} - -/// Probes the effective firewall backend and conntrack deletion capability. -pub(super) fn probe_runtime_support( - configured_backend: ConntrackBackend, -) -> ConntrackRuntimeSupport { - ConntrackRuntimeSupport { - netfilter_backend: pick_backend(configured_backend), - has_cap_net_admin: has_cap_net_admin(), - has_conntrack_binary: command_exists("conntrack"), - } -} - -/// Resolves whether conntrack close publication is usable for this runtime. -pub(super) fn effective_conntrack_enabled( - cfg: &ProxyConfig, - runtime_support: ConntrackRuntimeSupport, -) -> bool { - cfg.server.conntrack_control.inline_conntrack_control - && runtime_support.has_cap_net_admin - && runtime_support.netfilter_backend.is_some() - && runtime_support.has_conntrack_binary -} - -fn pick_backend(configured: ConntrackBackend) -> Option { - match configured { - ConntrackBackend::Auto => { - if command_exists("nft") { - Some(NetfilterBackend::Nftables) - } else if command_exists("iptables") { - Some(NetfilterBackend::Iptables) - } else { - None - } - } - ConntrackBackend::Nftables => command_exists("nft").then_some(NetfilterBackend::Nftables), - ConntrackBackend::Iptables => { - command_exists("iptables").then_some(NetfilterBackend::Iptables) - } - } -} - -fn command_exists(binary: &str) -> bool { - #[cfg(unix)] - { - resolve_trusted_helper(binary).is_some() - } - #[cfg(not(unix))] - { - let _ = binary; - false - } -} - -fn listener_port_set(cfg: &ProxyConfig) -> Vec { - let mut ports: BTreeSet = BTreeSet::new(); - if cfg.server.listeners.is_empty() { - ports.insert(cfg.server.port); - } else { - for listener in &cfg.server.listeners { - ports.insert(listener.port.unwrap_or(cfg.server.port)); - } - } - ports.into_iter().collect() -} - -fn notrack_targets(cfg: &ProxyConfig) -> (Vec<(Option, u16)>, Vec<(Option, u16)>) { - let mode = cfg.server.conntrack_control.mode; - let mut v4_targets: BTreeSet<(Option, u16)> = BTreeSet::new(); - let mut v6_targets: BTreeSet<(Option, u16)> = BTreeSet::new(); - - match mode { - ConntrackMode::Tracked => {} - ConntrackMode::Notrack => { - if cfg.server.listeners.is_empty() { - let port = cfg.server.port; - if let Some(ipv4) = cfg - .server - .listen_addr_ipv4 - .as_ref() - .and_then(|value| value.parse::().ok()) - { - if ipv4.is_unspecified() { - v4_targets.insert((None, port)); - } else { - v4_targets.insert((Some(ipv4), port)); - } - } - if let Some(ipv6) = cfg - .server - .listen_addr_ipv6 - .as_ref() - .and_then(|value| value.parse::().ok()) - { - if ipv6.is_unspecified() { - v6_targets.insert((None, port)); - } else { - v6_targets.insert((Some(ipv6), port)); - } - } - } else { - for listener in &cfg.server.listeners { - let port = listener.port.unwrap_or(cfg.server.port); - if listener.ip.is_ipv4() { - if listener.ip.is_unspecified() { - v4_targets.insert((None, port)); - } else { - v4_targets.insert((Some(listener.ip), port)); - } - } else if listener.ip.is_unspecified() { - v6_targets.insert((None, port)); - } else { - v6_targets.insert((Some(listener.ip), port)); - } - } - } - } - ConntrackMode::Hybrid => { - let ports = listener_port_set(cfg); - for ip in &cfg.server.conntrack_control.hybrid_listener_ips { - if ip.is_ipv4() { - for port in &ports { - v4_targets.insert((Some(*ip), *port)); - } - } else { - for port in &ports { - v6_targets.insert((Some(*ip), *port)); - } - } - } - } - } - - ( - v4_targets.into_iter().collect(), - v6_targets.into_iter().collect(), - ) -} - -async fn apply_nft_rules(cfg: &ProxyConfig) -> Result<(), String> { - let _ = run_command( - "nft", - &["delete", "table", "inet", "telemt_conntrack"], - None, - ) - .await; - if matches!(cfg.server.conntrack_control.mode, ConntrackMode::Tracked) { - return Ok(()); - } - - let (v4_targets, v6_targets) = notrack_targets(cfg); - let mut rules = Vec::new(); - for (ip, port) in v4_targets { - let rule = if let Some(ip) = ip { - format!("tcp dport {} ip daddr {} notrack", port, ip) - } else { - format!("tcp dport {} notrack", port) - }; - rules.push(rule); - } - for (ip, port) in v6_targets { - let rule = if let Some(ip) = ip { - format!("tcp dport {} ip6 daddr {} notrack", port, ip) - } else { - format!("tcp dport {} notrack", port) - }; - rules.push(rule); - } - - let rule_blob = if rules.is_empty() { - String::new() - } else { - format!(" {}\n", rules.join("\n ")) - }; - let script = format!( - "table inet telemt_conntrack {{\n chain preraw {{\n type filter hook prerouting priority raw; policy accept;\n{rule_blob} }}\n}}\n" - ); - run_command("nft", &["-f", "-"], Some(script)).await -} - -async fn apply_iptables_rules(cfg: &ProxyConfig) -> Result<(), String> { - apply_iptables_rules_for_binary("iptables", cfg, true).await?; - apply_iptables_rules_for_binary("ip6tables", cfg, false).await?; - Ok(()) -} - -async fn apply_iptables_rules_for_binary( - binary: &str, - cfg: &ProxyConfig, - ipv4: bool, -) -> Result<(), String> { - if !command_exists(binary) { - return Ok(()); - } - let chain = "TELEMT_NOTRACK"; - let _ = run_command( - binary, - &["-t", "raw", "-D", "PREROUTING", "-j", chain], - None, - ) - .await; - let _ = run_command(binary, &["-t", "raw", "-F", chain], None).await; - let _ = run_command(binary, &["-t", "raw", "-X", chain], None).await; - if matches!(cfg.server.conntrack_control.mode, ConntrackMode::Tracked) { - return Ok(()); - } - - run_command(binary, &["-t", "raw", "-N", chain], None).await?; - run_command(binary, &["-t", "raw", "-F", chain], None).await?; - if run_command( - binary, - &["-t", "raw", "-C", "PREROUTING", "-j", chain], - None, - ) - .await - .is_err() - { - run_command( - binary, - &["-t", "raw", "-I", "PREROUTING", "1", "-j", chain], - None, - ) - .await?; - } - - let (v4_targets, v6_targets) = notrack_targets(cfg); - let selected = if ipv4 { v4_targets } else { v6_targets }; - for (ip, port) in selected { - let mut args = vec![ - "-t".to_string(), - "raw".to_string(), - "-A".to_string(), - chain.to_string(), - "-p".to_string(), - "tcp".to_string(), - "--dport".to_string(), - port.to_string(), - ]; - if let Some(ip) = ip { - args.push("-d".to_string()); - args.push(ip.to_string()); - } - args.push("-j".to_string()); - args.push("CT".to_string()); - args.push("--notrack".to_string()); - let arg_refs: Vec<&str> = args.iter().map(String::as_str).collect(); - run_command(binary, &arg_refs, None).await?; - } - Ok(()) -} - -async fn clear_notrack_rules_all_backends() { - let _ = run_command( - "nft", - &["delete", "table", "inet", "telemt_conntrack"], - None, - ) - .await; - let _ = run_command( - "iptables", - &["-t", "raw", "-D", "PREROUTING", "-j", "TELEMT_NOTRACK"], - None, - ) - .await; - let _ = run_command("iptables", &["-t", "raw", "-F", "TELEMT_NOTRACK"], None).await; - let _ = run_command("iptables", &["-t", "raw", "-X", "TELEMT_NOTRACK"], None).await; - let _ = run_command( - "ip6tables", - &["-t", "raw", "-D", "PREROUTING", "-j", "TELEMT_NOTRACK"], - None, - ) - .await; - let _ = run_command("ip6tables", &["-t", "raw", "-F", "TELEMT_NOTRACK"], None).await; - let _ = run_command("ip6tables", &["-t", "raw", "-X", "TELEMT_NOTRACK"], None).await; -} - -/// Result of one best-effort kernel conntrack deletion. -pub(super) enum DeleteOutcome { - /// The kernel reported successful deletion. - Deleted, - /// No matching conntrack entry existed. - NotFound, - /// The helper was unavailable or returned an unexpected failure. - Error, -} - -/// Deletes the exact TCP tuple represented by one close event. -pub(super) async fn delete_conntrack_entry(event: ConntrackCloseEvent) -> DeleteOutcome { - if !command_exists("conntrack") { - return DeleteOutcome::Error; - } - let args = vec![ - "-D".to_string(), - "-p".to_string(), - "tcp".to_string(), - "-s".to_string(), - event.src.ip().to_string(), - "--sport".to_string(), - event.src.port().to_string(), - "-d".to_string(), - event.dst.ip().to_string(), - "--dport".to_string(), - event.dst.port().to_string(), - ]; - let arg_refs: Vec<&str> = args.iter().map(String::as_str).collect(); - match run_command("conntrack", &arg_refs, None).await { - Ok(()) => DeleteOutcome::Deleted, - Err(error) => { - if error.contains("0 flow entries have been deleted") { - DeleteOutcome::NotFound - } else { - debug!(error = %error, "conntrack delete failed"); - DeleteOutcome::Error - } - } - } -} - -async fn run_command(binary: &str, args: &[&str], stdin: Option) -> Result<(), String> { - const COMMAND_TIMEOUT: Duration = Duration::from_secs(30); - #[cfg(unix)] - let Some(command_path) = resolve_trusted_helper(binary) else { - return Err(format!("{binary} is not available")); - }; - #[cfg(not(unix))] - return Err(format!("{binary} is not available")); - #[cfg(unix)] - let mut command = Command::new(command_path); - command.args(args); - if stdin.is_some() { - command.stdin(std::process::Stdio::piped()); - } - command.stdout(std::process::Stdio::null()); - command.stderr(std::process::Stdio::piped()); - command.kill_on_drop(true); - let mut child = command - .spawn() - .map_err(|error| format!("spawn {binary} failed: {error}"))?; - let output = tokio::time::timeout(COMMAND_TIMEOUT, async move { - if let Some(blob) = stdin - && let Some(mut writer) = child.stdin.take() - { - writer - .write_all(blob.as_bytes()) - .await - .map_err(|error| format!("stdin write {binary} failed: {error}"))?; - } - child - .wait_with_output() - .await - .map_err(|error| format!("wait {binary} failed: {error}")) - }) - .await - .map_err(|_| format!("{binary} timed out after {}s", COMMAND_TIMEOUT.as_secs()))??; - if output.status.success() { - return Ok(()); - } - let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string(); - Err(if stderr.is_empty() { - format!("{binary} exited with status {}", output.status) - } else { - stderr - }) -} - -fn has_cap_net_admin() -> bool { - #[cfg(target_os = "linux")] - { - let Ok(status) = std::fs::read_to_string("/proc/self/status") else { - return false; - }; - for line in status.lines() { - if let Some(raw) = line.strip_prefix("CapEff:") { - let caps = raw.trim(); - if let Ok(bits) = u64::from_str_radix(caps, 16) { - const CAP_NET_ADMIN_BIT: u64 = 12; - return (bits & (1u64 << CAP_NET_ADMIN_BIT)) != 0; - } - } - } - false - } - #[cfg(not(target_os = "linux"))] - { - false - } -} +#[cfg(test)] +#[path = "firewall/tests.rs"] +mod tests; diff --git a/src/conntrack_control/firewall/actor.rs b/src/conntrack_control/firewall/actor.rs new file mode 100644 index 0000000..561665d --- /dev/null +++ b/src/conntrack_control/firewall/actor.rs @@ -0,0 +1,375 @@ +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; + +use tokio::sync::{Notify, watch}; +use tokio_util::sync::CancellationToken; +use tracing::warn; + +use crate::config::ProxyConfig; +use crate::maestro::control_plane::ProcessControlPlane; +use crate::stats::Stats; + +use super::command::{CommandError, FirewallCommandRunner, SystemCommandRunner}; +use super::model::{AppliedPlan, AppliedState, DesiredPolicy, DesiredState}; +use super::transaction::{InterruptibleRunner, reconcile_once, recover_to_empty}; + +const SHUTDOWN_CLEANUP_TIMEOUT: Duration = Duration::from_secs(30); +const SHUTDOWN_WAIT_TIMEOUT: Duration = Duration::from_secs(35); +const INITIAL_RECONCILE_TIMEOUT: Duration = Duration::from_secs(65); +const RETRY_DELAYS: [Duration; 6] = [ + Duration::from_secs(1), + Duration::from_secs(2), + Duration::from_secs(4), + Duration::from_secs(8), + Duration::from_secs(16), + Duration::from_secs(30), +]; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) enum ReconcileOutcome { + Applied, + Failed, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) struct ReconcileStatus { + pub(super) generation: u64, + pub(super) outcome: ReconcileOutcome, +} + +/// Process-owned publisher and shutdown owner for conntrack firewall policy. +#[derive(Clone)] +pub(crate) struct FirewallAuthority { + desired_tx: watch::Sender>, + status_rx: watch::Receiver>, + terminal: CancellationToken, + closed: Arc, + completed_flag: Arc, + cleanup_succeeded: Arc, + completed: Arc, +} + +impl FirewallAuthority { + /// Starts the single process-owned firewall reconciler. + pub(crate) fn spawn(control_plane: &ProcessControlPlane) -> Result { + let (desired_tx, desired_rx) = watch::channel(None); + let (status_tx, status_rx) = watch::channel(None); + let terminal = CancellationToken::new(); + let closed = Arc::new(AtomicBool::new(false)); + let completed_flag = Arc::new(AtomicBool::new(false)); + let cleanup_succeeded = Arc::new(AtomicBool::new(false)); + let completed = Arc::new(Notify::new()); + let actor = FirewallReconciler::new( + SystemCommandRunner, + desired_rx, + status_tx, + terminal.clone(), + Arc::clone(&closed), + Arc::clone(&completed_flag), + Arc::clone(&cleanup_succeeded), + Arc::clone(&completed), + ); + control_plane + .spawn_cooperative(move |process_cancellation| async move { + actor.run(process_cancellation).await; + }) + .map_err(|_| { + "process control-plane admission closed before conntrack firewall startup" + .to_string() + })?; + Ok(Self { + desired_tx, + status_rx, + terminal, + closed, + completed_flag, + cleanup_succeeded, + completed, + }) + } + + /// Publishes policy only after its runtime generation becomes active. + pub(crate) fn publish( + &self, + generation: u64, + config: Arc, + stats: Arc, + ) -> bool { + if self.closed.load(Ordering::Acquire) { + stats.set_conntrack_rule_apply_ok(false); + return false; + } + stats.set_conntrack_rule_apply_ok(false); + self.desired_tx.send_replace(Some(DesiredState { + generation, + policy: DesiredPolicy::from_config(config.as_ref()), + stats, + })); + true + } + + /// Publishes startup policy and waits for its first bounded attempt. + pub(crate) async fn publish_initial( + &self, + generation: u64, + config: Arc, + stats: Arc, + ) -> bool { + let mut status_rx = self.status_rx.clone(); + if !self.publish(generation, config, stats) { + return false; + } + tokio::time::timeout(INITIAL_RECONCILE_TIMEOUT, async move { + loop { + if let Some(status) = *status_rx.borrow_and_update() + && status.generation == generation + { + return status.outcome == ReconcileOutcome::Applied; + } + if status_rx.changed().await.is_err() { + return false; + } + } + }) + .await + .unwrap_or(false) + } + + /// Stops policy admission and waits for bounded terminal cleanup. + pub(crate) async fn shutdown_and_clear(&self) -> bool { + let completed = self.completed.notified(); + tokio::pin!(completed); + completed.as_mut().enable(); + if !self.closed.swap(true, Ordering::AcqRel) { + if let Some(desired) = self.desired_tx.borrow().as_ref() { + desired.stats.set_conntrack_rule_apply_ok(false); + } + self.terminal.cancel(); + } + let finished = self.completed_flag.load(Ordering::Acquire) + || tokio::time::timeout(SHUTDOWN_WAIT_TIMEOUT, completed) + .await + .is_ok(); + finished && self.cleanup_succeeded.load(Ordering::Acquire) + } +} + +struct CompletionGuard { + closed: Arc, + completed_flag: Arc, + completed: Arc, +} + +impl Drop for CompletionGuard { + fn drop(&mut self) { + self.closed.store(true, Ordering::Release); + self.completed_flag.store(true, Ordering::Release); + self.completed.notify_waiters(); + } +} + +pub(super) struct FirewallReconciler { + runner: R, + desired_rx: watch::Receiver>, + status_tx: watch::Sender>, + terminal: CancellationToken, + completion: CompletionGuard, + cleanup_succeeded: Arc, + applied: AppliedState, + last_generation: u64, + last_policy: Option, + last_stats: Option>, +} + +impl FirewallReconciler +where + R: FirewallCommandRunner + 'static, +{ + pub(super) fn new( + runner: R, + desired_rx: watch::Receiver>, + status_tx: watch::Sender>, + terminal: CancellationToken, + closed: Arc, + completed_flag: Arc, + cleanup_succeeded: Arc, + completed: Arc, + ) -> Self { + Self { + runner, + desired_rx, + status_tx, + terminal, + completion: CompletionGuard { + closed, + completed_flag, + completed, + }, + cleanup_succeeded, + applied: AppliedState::Unknown, + last_generation: 0, + last_policy: None, + last_stats: None, + } + } + + pub(super) async fn run(mut self, process_cancellation: CancellationToken) { + let mut current = None; + let mut retry_index = 0usize; + 'run: loop { + if current.is_none() { + let changed = tokio::select! { + biased; + _ = self.terminal.cancelled() => false, + _ = process_cancellation.cancelled() => false, + changed = self.desired_rx.changed() => changed.is_ok(), + }; + if !changed { + break; + } + current = self.take_latest_desired(); + retry_index = 0; + if current.is_none() { + continue; + } + } + + let desired = current.as_ref().expect("desired state is present").clone(); + let interruptible = InterruptibleRunner::new( + &self.runner, + &self.terminal, + &process_cancellation, + ); + let result = reconcile_once( + &interruptible, + &interruptible, + &mut self.applied, + &desired, + ) + .await; + match result { + Ok(()) => { + desired.stats.increment_conntrack_rule_reconcile_success_total(); + desired.stats.set_conntrack_rule_apply_ok(true); + self.status_tx.send_replace(Some(ReconcileStatus { + generation: desired.generation, + outcome: ReconcileOutcome::Applied, + })); + if self.terminal.is_cancelled() || process_cancellation.is_cancelled() { + desired.stats.set_conntrack_rule_apply_ok(false); + break; + } + current = None; + retry_index = 0; + } + Err(failure) if failure.cancelled => break, + Err(failure) => { + desired.stats.increment_conntrack_rule_reconcile_error_total(); + desired.stats.set_conntrack_rule_apply_ok(false); + if let Some(rollback_succeeded) = failure.rollback_succeeded { + if rollback_succeeded { + desired + .stats + .increment_conntrack_rule_rollback_success_total(); + } else { + desired + .stats + .increment_conntrack_rule_rollback_error_total(); + } + } + self.status_tx.send_replace(Some(ReconcileStatus { + generation: desired.generation, + outcome: ReconcileOutcome::Failed, + })); + warn!( + generation = desired.generation, + error = %failure.message, + "Failed to reconcile conntrack firewall policy" + ); + + let delay = RETRY_DELAYS[retry_index.min(RETRY_DELAYS.len() - 1)]; + retry_index = retry_index.saturating_add(1); + let retry_deadline = tokio::time::Instant::now() + delay; + loop { + tokio::select! { + biased; + _ = self.terminal.cancelled() => break 'run, + _ = process_cancellation.cancelled() => break 'run, + changed = self.desired_rx.changed() => { + if changed.is_err() { + break 'run; + } + if let Some(next) = self.take_latest_desired() { + current = Some(next); + retry_index = 0; + break; + } + } + _ = tokio::time::sleep_until(retry_deadline) => break, + } + } + } + } + + if self.terminal.is_cancelled() || process_cancellation.is_cancelled() { + break; + } + if self.desired_rx.has_changed().unwrap_or(false) { + if let Some(next) = self.take_latest_desired() { + current = Some(next); + retry_index = 0; + } + } + } + + if let Some(stats) = &self.last_stats { + stats.set_conntrack_rule_apply_ok(false); + } + if let Err(error) = tokio::time::timeout( + SHUTDOWN_CLEANUP_TIMEOUT, + recover_to_empty(&self.runner), + ) + .await + .unwrap_or_else(|_| Err(CommandError::failed("firewall shutdown cleanup timed out"))) + { + warn!(error = %error, "Failed to clear conntrack firewall policy during shutdown"); + } else { + self.applied = AppliedState::Known(AppliedPlan::Empty); + self.cleanup_succeeded.store(true, Ordering::Release); + } + let _completion = &self.completion; + } + + pub(super) fn take_latest_desired(&mut self) -> Option { + let next = self.desired_rx.borrow_and_update().clone()?; + if next.generation < self.last_generation { + warn!( + generation = next.generation, + active_generation = self.last_generation, + "Ignored stale conntrack firewall policy publication" + ); + return None; + } + if next.generation == self.last_generation { + if self.last_policy.as_ref() != Some(&next.policy) { + warn!( + generation = next.generation, + "Ignored conflicting conntrack firewall policy for active generation" + ); + return None; + } + self.last_stats = Some(next.stats.clone()); + if let AppliedState::Known(applied) = &self.applied + && applied.matches_policy(&next.policy) + { + next.stats.set_conntrack_rule_apply_ok(true); + } + return Some(next); + } + self.last_generation = next.generation; + self.last_policy = Some(next.policy.clone()); + self.last_stats = Some(next.stats.clone()); + Some(next) + } +} diff --git a/src/conntrack_control/firewall/command.rs b/src/conntrack_control/firewall/command.rs new file mode 100644 index 0000000..c9fb7d5 --- /dev/null +++ b/src/conntrack_control/firewall/command.rs @@ -0,0 +1,204 @@ +use std::time::Duration; + +#[cfg(unix)] +use tokio::io::AsyncWriteExt; +#[cfg(unix)] +use tokio::process::Command; + +#[cfg(unix)] +use crate::util::trusted_command::resolve_trusted_helper; + +const COMMAND_TIMEOUT: Duration = Duration::from_secs(30); + +#[derive(Clone, Debug, Eq, PartialEq)] +pub(super) struct CommandSpec { + pub(super) binary: &'static str, + pub(super) args: Vec, + pub(super) stdin: Option, +} + +impl CommandSpec { + pub(super) fn new(binary: &'static str, args: impl IntoIterator) -> Self { + Self { + binary, + args: args.into_iter().map(str::to_string).collect(), + stdin: None, + } + } + + pub(super) fn with_stdin( + binary: &'static str, + args: impl IntoIterator, + stdin: String, + ) -> Self { + Self { + binary, + args: args.into_iter().map(str::to_string).collect(), + stdin: Some(stdin), + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) enum CommandErrorKind { + Missing, + NotFound, + Cancelled, + Timeout, + Failed, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub(super) struct CommandError { + pub(super) kind: CommandErrorKind, + pub(super) message: String, +} + +impl CommandError { + pub(super) fn cancelled() -> Self { + Self { + kind: CommandErrorKind::Cancelled, + message: "firewall transaction cancelled".to_string(), + } + } + + pub(super) fn failed(message: impl Into) -> Self { + Self { + kind: CommandErrorKind::Failed, + message: message.into(), + } + } +} + +impl std::fmt::Display for CommandError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str(&self.message) + } +} + +pub(super) trait FirewallCommandRunner: Send + Sync { + fn available(&self, binary: &str) -> bool; + + fn has_cap_net_admin(&self) -> bool; + + async fn run(&self, spec: CommandSpec) -> Result<(), CommandError>; +} + +#[derive(Clone, Copy, Default)] +pub(super) struct SystemCommandRunner; + +impl FirewallCommandRunner for SystemCommandRunner { + fn available(&self, binary: &str) -> bool { + #[cfg(unix)] + { + resolve_trusted_helper(binary).is_some() + } + #[cfg(not(unix))] + { + let _ = binary; + false + } + } + + fn has_cap_net_admin(&self) -> bool { + #[cfg(target_os = "linux")] + { + let Ok(status) = std::fs::read_to_string("/proc/self/status") else { + return false; + }; + for line in status.lines() { + if let Some(raw) = line.strip_prefix("CapEff:") { + let caps = raw.trim(); + if let Ok(bits) = u64::from_str_radix(caps, 16) { + const CAP_NET_ADMIN_BIT: u64 = 12; + return (bits & (1u64 << CAP_NET_ADMIN_BIT)) != 0; + } + } + } + false + } + #[cfg(not(target_os = "linux"))] + { + false + } + } + + async fn run(&self, spec: CommandSpec) -> Result<(), CommandError> { + #[cfg(not(unix))] + { + Err(CommandError { + kind: CommandErrorKind::Missing, + message: format!("{} is not available", spec.binary), + }) + } + #[cfg(unix)] + { + let Some(command_path) = resolve_trusted_helper(spec.binary) else { + return Err(CommandError { + kind: CommandErrorKind::Missing, + message: format!("{} is not available", spec.binary), + }); + }; + let mut command = Command::new(command_path); + command.args(&spec.args); + command.env("LC_ALL", "C"); + if spec.stdin.is_some() { + command.stdin(std::process::Stdio::piped()); + } + command.stdout(std::process::Stdio::null()); + command.stderr(std::process::Stdio::piped()); + command.kill_on_drop(true); + let mut child = command.spawn().map_err(|error| CommandError { + kind: CommandErrorKind::Failed, + message: format!("spawn {} failed: {error}", spec.binary), + })?; + let binary = spec.binary; + let output = tokio::time::timeout(COMMAND_TIMEOUT, async move { + if let Some(blob) = spec.stdin + && let Some(mut writer) = child.stdin.take() + { + writer + .write_all(blob.as_bytes()) + .await + .map_err(|error| CommandError { + kind: CommandErrorKind::Failed, + message: format!("stdin write {binary} failed: {error}"), + })?; + } + child + .wait_with_output() + .await + .map_err(|error| CommandError { + kind: CommandErrorKind::Failed, + message: format!("wait {binary} failed: {error}"), + }) + }) + .await + .map_err(|_| CommandError { + kind: CommandErrorKind::Timeout, + message: format!("{binary} timed out after {}s", COMMAND_TIMEOUT.as_secs()), + })??; + if output.status.success() { + return Ok(()); + } + let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string(); + let message = if stderr.is_empty() { + format!("{binary} exited with status {}", output.status) + } else { + stderr + }; + let kind = if is_not_found_error(&message) { + CommandErrorKind::NotFound + } else { + CommandErrorKind::Failed + }; + Err(CommandError { kind, message }) + } + } +} + +pub(super) fn is_not_found_error(message: &str) -> bool { + message.contains("No chain/target/match by that name") + || message.contains("Bad rule (does a matching rule exist in that chain?)") + || message.contains("Could not process rule: No such file or directory") +} diff --git a/src/conntrack_control/firewall/iptables.rs b/src/conntrack_control/firewall/iptables.rs new file mode 100644 index 0000000..a570556 --- /dev/null +++ b/src/conntrack_control/firewall/iptables.rs @@ -0,0 +1,255 @@ +use super::command::{ + CommandError, CommandErrorKind, CommandSpec, FirewallCommandRunner, +}; +use super::model::{NotrackTarget, ShadowSlot}; + +const DISPATCH_CHAIN: &str = "TELEMT_NOTRACK"; +const SHADOW_CHAIN_A: &str = "TELEMT_NT_A"; +const SHADOW_CHAIN_B: &str = "TELEMT_NT_B"; +const MAX_OWNED_JUMPS: usize = 8; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) enum IpFamily { + V4, + V6, +} + +impl IpFamily { + fn command_binary(self) -> &'static str { + match self { + Self::V4 => "iptables", + Self::V6 => "ip6tables", + } + } + + fn restore_binary(self) -> &'static str { + match self { + Self::V4 => "iptables-restore", + Self::V6 => "ip6tables-restore", + } + } +} + +pub(super) fn family_available( + runner: &R, + family: IpFamily, +) -> bool { + runner.available(family.command_binary()) && runner.available(family.restore_binary()) +} + +fn shadow_chain(slot: ShadowSlot) -> &'static str { + match slot { + ShadowSlot::A => SHADOW_CHAIN_A, + ShadowSlot::B => SHADOW_CHAIN_B, + } +} + +pub(super) async fn stage_family( + runner: &R, + family: IpFamily, + slot: ShadowSlot, + targets: &[NotrackTarget], +) -> Result<(), CommandError> { + if targets.is_empty() { + return Ok(()); + } + require_family(runner, family)?; + ensure_owned_chains(runner, family).await?; + let script = render_stage_script(slot, targets); + runner + .run(CommandSpec::with_stdin( + family.restore_binary(), + ["--noflush"], + script, + )) + .await +} + +pub(super) async fn activate_family( + runner: &R, + family: IpFamily, + slot: Option, +) -> Result<(), CommandError> { + require_family(runner, family)?; + if slot.is_some() { + ensure_prerouting_jump(runner, family).await?; + } + runner + .run(CommandSpec::with_stdin( + family.restore_binary(), + ["--noflush"], + render_dispatch_script(slot), + )) + .await +} + +pub(super) async fn cleanup_all( + runner: &R, +) -> Result<(), CommandError> { + let mut errors = Vec::new(); + for family in [IpFamily::V4, IpFamily::V6] { + if !runner.available(family.command_binary()) { + continue; + } + if let Err(error) = cleanup_family(runner, family).await { + errors.push(error.message); + } + } + if errors.is_empty() { + Ok(()) + } else { + Err(CommandError::failed(errors.join("; "))) + } +} + +async fn cleanup_family( + runner: &R, + family: IpFamily, +) -> Result<(), CommandError> { + let binary = family.command_binary(); + let mut errors = Vec::new(); + for _ in 0..MAX_OWNED_JUMPS { + let result = runner + .run(CommandSpec::new( + binary, + ["-t", "raw", "-D", "PREROUTING", "-j", DISPATCH_CHAIN], + )) + .await; + match result { + Ok(()) => {} + Err(error) + if matches!(error.kind, CommandErrorKind::NotFound | CommandErrorKind::Missing) => + { + break; + } + Err(error) => { + errors.push(error.message); + break; + } + } + } + for chain in [DISPATCH_CHAIN, SHADOW_CHAIN_A, SHADOW_CHAIN_B] { + for operation in ["-F", "-X"] { + let result = runner + .run(CommandSpec::new( + binary, + ["-t", "raw", operation, chain], + )) + .await; + if let Err(error) = result + && !matches!(error.kind, CommandErrorKind::NotFound | CommandErrorKind::Missing) + { + errors.push(error.message); + } + } + } + if errors.is_empty() { + Ok(()) + } else { + Err(CommandError::failed(errors.join("; "))) + } +} + +async fn ensure_prerouting_jump( + runner: &R, + family: IpFamily, +) -> Result<(), CommandError> { + let binary = family.command_binary(); + match runner + .run(CommandSpec::new( + binary, + ["-t", "raw", "-C", "PREROUTING", "-j", DISPATCH_CHAIN], + )) + .await + { + Ok(()) => Ok(()), + Err(error) if error.kind == CommandErrorKind::NotFound => { + runner + .run(CommandSpec::new( + binary, + [ + "-t", + "raw", + "-I", + "PREROUTING", + "1", + "-j", + DISPATCH_CHAIN, + ], + )) + .await + } + Err(error) => Err(error), + } +} + +async fn ensure_owned_chains( + runner: &R, + family: IpFamily, +) -> Result<(), CommandError> { + let binary = family.command_binary(); + for chain in [DISPATCH_CHAIN, SHADOW_CHAIN_A, SHADOW_CHAIN_B] { + match runner + .run(CommandSpec::new(binary, ["-t", "raw", "-N", chain])) + .await + { + Ok(()) => {} + Err(error) if is_chain_exists_error(&error.message) => {} + Err(error) => return Err(error), + } + } + Ok(()) +} + +pub(super) fn is_chain_exists_error(message: &str) -> bool { + message.contains("Chain already exists") +} + +fn require_family( + runner: &R, + family: IpFamily, +) -> Result<(), CommandError> { + for binary in [family.command_binary(), family.restore_binary()] { + if !runner.available(binary) { + return Err(CommandError { + kind: CommandErrorKind::Missing, + message: format!("{binary} is required for conntrack firewall reconciliation"), + }); + } + } + Ok(()) +} + +pub(super) fn render_stage_script( + slot: ShadowSlot, + targets: &[NotrackTarget], +) -> String { + let chain = shadow_chain(slot); + let mut script = format!("*raw\n-F {chain}\n"); + for target in targets { + script.push_str("-A "); + script.push_str(chain); + script.push_str(" -p tcp --dport "); + script.push_str(&target.port.to_string()); + if let Some(ip) = target.ip { + script.push_str(" -d "); + script.push_str(&ip.to_string()); + } + script.push_str(" -j CT --notrack\n"); + } + script.push_str("COMMIT\n"); + script +} + +pub(super) fn render_dispatch_script(slot: Option) -> String { + let mut script = format!("*raw\n-F {DISPATCH_CHAIN}\n"); + if let Some(slot) = slot { + script.push_str("-A "); + script.push_str(DISPATCH_CHAIN); + script.push_str(" -j "); + script.push_str(shadow_chain(slot)); + script.push('\n'); + } + script.push_str("COMMIT\n"); + script +} diff --git a/src/conntrack_control/firewall/model.rs b/src/conntrack_control/firewall/model.rs new file mode 100644 index 0000000..4e97623 --- /dev/null +++ b/src/conntrack_control/firewall/model.rs @@ -0,0 +1,205 @@ +use std::collections::BTreeSet; +use std::net::IpAddr; +use std::sync::Arc; + +use crate::config::{ConntrackBackend, ConntrackMode, ProxyConfig}; +use crate::stats::Stats; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) enum ShadowSlot { + A, + B, +} + +impl ShadowSlot { + pub(super) fn other(self) -> Self { + match self { + Self::A => Self::B, + Self::B => Self::A, + } + } + +} + +#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)] +pub(super) struct NotrackTarget { + pub(super) ip: Option, + pub(super) port: u16, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub(super) enum DesiredPolicy { + Empty, + Rules { + configured_backend: ConntrackBackend, + v4: Vec, + v6: Vec, + }, +} + +impl DesiredPolicy { + pub(super) fn from_config(cfg: &ProxyConfig) -> Self { + if !cfg.server.conntrack_control.inline_conntrack_control + || matches!(cfg.server.conntrack_control.mode, ConntrackMode::Tracked) + { + return Self::Empty; + } + let (v4, v6) = notrack_targets(cfg); + if v4.is_empty() && v6.is_empty() { + Self::Empty + } else { + Self::Rules { + configured_backend: cfg.server.conntrack_control.backend, + v4, + v6, + } + } + } +} + +#[derive(Clone)] +pub(super) struct DesiredState { + pub(super) generation: u64, + pub(super) policy: DesiredPolicy, + pub(super) stats: Arc, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub(super) enum AppliedPlan { + Empty, + Iptables { + slot: ShadowSlot, + v4: Vec, + v6: Vec, + }, + Nftables { + slot: ShadowSlot, + v4: Vec, + v6: Vec, + }, +} + +impl AppliedPlan { + pub(super) fn slot(&self) -> Option { + match self { + Self::Empty => None, + Self::Iptables { slot, .. } | Self::Nftables { slot, .. } => Some(*slot), + } + } + + pub(super) fn matches_policy(&self, policy: &DesiredPolicy) -> bool { + match (self, policy) { + (Self::Empty, DesiredPolicy::Empty) => true, + ( + Self::Iptables { v4, v6, .. }, + DesiredPolicy::Rules { + configured_backend, + v4: desired_v4, + v6: desired_v6, + }, + ) => { + matches!( + configured_backend, + ConntrackBackend::Auto | ConntrackBackend::Iptables + ) && v4 == desired_v4 + && v6 == desired_v6 + } + ( + Self::Nftables { v4, v6, .. }, + DesiredPolicy::Rules { + configured_backend, + v4: desired_v4, + v6: desired_v6, + }, + ) => { + matches!( + configured_backend, + ConntrackBackend::Auto | ConntrackBackend::Nftables + ) && v4 == desired_v4 + && v6 == desired_v6 + } + _ => false, + } + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub(super) enum AppliedState { + Known(AppliedPlan), + Unknown, +} + +fn listener_port_set(cfg: &ProxyConfig) -> Vec { + let mut ports = BTreeSet::new(); + if cfg.server.listeners.is_empty() { + ports.insert(cfg.server.port); + } else { + for listener in &cfg.server.listeners { + ports.insert(listener.port.unwrap_or(cfg.server.port)); + } + } + ports.into_iter().collect() +} + +fn notrack_targets(cfg: &ProxyConfig) -> (Vec, Vec) { + let mut v4_targets = BTreeSet::new(); + let mut v6_targets = BTreeSet::new(); + match cfg.server.conntrack_control.mode { + ConntrackMode::Tracked => {} + ConntrackMode::Notrack => { + if cfg.server.listeners.is_empty() { + let port = cfg.server.port; + for raw in [ + cfg.server.listen_addr_ipv4.as_deref(), + cfg.server.listen_addr_ipv6.as_deref(), + ] + .into_iter() + .flatten() + { + if let Ok(ip) = raw.parse::() { + let target = NotrackTarget { + ip: (!ip.is_unspecified()).then_some(ip), + port, + }; + if ip.is_ipv4() { + v4_targets.insert(target); + } else { + v6_targets.insert(target); + } + } + } + } else { + for listener in &cfg.server.listeners { + let target = NotrackTarget { + ip: (!listener.ip.is_unspecified()).then_some(listener.ip), + port: listener.port.unwrap_or(cfg.server.port), + }; + if listener.ip.is_ipv4() { + v4_targets.insert(target); + } else { + v6_targets.insert(target); + } + } + } + } + ConntrackMode::Hybrid => { + for ip in &cfg.server.conntrack_control.hybrid_listener_ips { + for port in listener_port_set(cfg) { + let target = NotrackTarget { + ip: Some(*ip), + port, + }; + if ip.is_ipv4() { + v4_targets.insert(target); + } else { + v6_targets.insert(target); + } + } + } + } + } + ( + v4_targets.into_iter().collect(), + v6_targets.into_iter().collect(), + ) +} diff --git a/src/conntrack_control/firewall/nftables.rs b/src/conntrack_control/firewall/nftables.rs new file mode 100644 index 0000000..7b9f353 --- /dev/null +++ b/src/conntrack_control/firewall/nftables.rs @@ -0,0 +1,147 @@ +use super::command::{CommandError, CommandErrorKind, CommandSpec, FirewallCommandRunner}; +use super::model::{NotrackTarget, ShadowSlot}; + +const LEGACY_TABLE: &str = "telemt_conntrack"; +const TABLE_A: &str = "telemt_conntrack_a"; +const TABLE_B: &str = "telemt_conntrack_b"; + +pub(super) fn available(runner: &R) -> bool { + runner.available("nft") +} + +fn table(slot: ShadowSlot) -> &'static str { + match slot { + ShadowSlot::A => TABLE_A, + ShadowSlot::B => TABLE_B, + } +} + +pub(super) async fn stage( + runner: &R, + slot: ShadowSlot, + v4: &[NotrackTarget], + v6: &[NotrackTarget], +) -> Result<(), CommandError> { + require_nft(runner)?; + delete_table_if_present(runner, table(slot)).await?; + runner + .run(CommandSpec::with_stdin( + "nft", + ["-f", "-"], + render_stage_script(slot, v4, v6), + )) + .await +} + +pub(super) async fn activate( + runner: &R, + slot: ShadowSlot, +) -> Result<(), CommandError> { + require_nft(runner)?; + runner + .run(CommandSpec::with_stdin( + "nft", + ["-f", "-"], + render_activate_script(slot), + )) + .await +} + +pub(super) async fn deactivate( + runner: &R, + slot: ShadowSlot, +) -> Result<(), CommandError> { + delete_table_if_present(runner, table(slot)).await +} + +pub(super) async fn cleanup_all( + runner: &R, +) -> Result<(), CommandError> { + if !runner.available("nft") { + return Ok(()); + } + let mut errors = Vec::new(); + for table_name in [LEGACY_TABLE, TABLE_A, TABLE_B] { + if let Err(error) = delete_table_if_present(runner, table_name).await { + errors.push(error.message); + } + } + if errors.is_empty() { + Ok(()) + } else { + Err(CommandError::failed(errors.join("; "))) + } +} + +async fn delete_table_if_present( + runner: &R, + table_name: &'static str, +) -> Result<(), CommandError> { + match runner + .run(CommandSpec::new( + "nft", + ["delete", "table", "inet", table_name], + )) + .await + { + Ok(()) => Ok(()), + Err(error) + if matches!(error.kind, CommandErrorKind::NotFound | CommandErrorKind::Missing) => + { + Ok(()) + } + Err(error) => Err(error), + } +} + +fn require_nft(runner: &R) -> Result<(), CommandError> { + if runner.available("nft") { + Ok(()) + } else { + Err(CommandError { + kind: CommandErrorKind::Missing, + message: "nft is required for conntrack firewall reconciliation".to_string(), + }) + } +} + +pub(super) fn render_stage_script( + slot: ShadowSlot, + v4: &[NotrackTarget], + v6: &[NotrackTarget], +) -> String { + let table = table(slot); + let mut script = format!( + "add table inet {table}\nadd chain inet {table} rules\n" + ); + for target in v4 { + script.push_str("add rule inet "); + script.push_str(table); + script.push_str(" rules tcp dport "); + script.push_str(&target.port.to_string()); + if let Some(ip) = target.ip { + script.push_str(" ip daddr "); + script.push_str(&ip.to_string()); + } + script.push_str(" notrack\n"); + } + for target in v6 { + script.push_str("add rule inet "); + script.push_str(table); + script.push_str(" rules tcp dport "); + script.push_str(&target.port.to_string()); + if let Some(ip) = target.ip { + script.push_str(" ip6 daddr "); + script.push_str(&ip.to_string()); + } + script.push_str(" notrack\n"); + } + script +} + +pub(super) fn render_activate_script(slot: ShadowSlot) -> String { + let table = table(slot); + format!( + "add chain inet {table} preraw {{ type filter hook prerouting priority raw; policy accept; }}\nadd rule inet {table} preraw jump rules\n" + ) +} diff --git a/src/conntrack_control/firewall/runtime.rs b/src/conntrack_control/firewall/runtime.rs new file mode 100644 index 0000000..706e007 --- /dev/null +++ b/src/conntrack_control/firewall/runtime.rs @@ -0,0 +1,91 @@ +use tracing::debug; + +use crate::config::{ConntrackBackend, ProxyConfig}; +use crate::conntrack_control::{ConntrackRuntimeSupport, NetfilterBackend}; +use crate::proxy::shared_state::ConntrackCloseEvent; + +use super::command::{CommandSpec, FirewallCommandRunner, SystemCommandRunner}; +use super::nftables; + +/// Probes the effective firewall backend and conntrack deletion capability. +pub(in crate::conntrack_control) fn probe_runtime_support( + configured_backend: ConntrackBackend, +) -> ConntrackRuntimeSupport { + let runner = SystemCommandRunner; + let iptables_available = runner.available("iptables"); + let netfilter_backend = match configured_backend { + ConntrackBackend::Auto if nftables::available(&runner) => Some(NetfilterBackend::Nftables), + ConntrackBackend::Auto if iptables_available => Some(NetfilterBackend::Iptables), + ConntrackBackend::Nftables if nftables::available(&runner) => { + Some(NetfilterBackend::Nftables) + } + ConntrackBackend::Iptables if iptables_available => Some(NetfilterBackend::Iptables), + _ => None, + }; + ConntrackRuntimeSupport { + netfilter_backend, + has_cap_net_admin: runner.has_cap_net_admin(), + has_conntrack_binary: runner.available("conntrack"), + } +} + +/// Resolves whether conntrack close publication is usable for this runtime. +pub(in crate::conntrack_control) fn effective_conntrack_enabled( + cfg: &ProxyConfig, + runtime_support: ConntrackRuntimeSupport, +) -> bool { + cfg.server.conntrack_control.inline_conntrack_control + && runtime_support.has_cap_net_admin + && runtime_support.netfilter_backend.is_some() + && runtime_support.has_conntrack_binary +} + +/// Result of one best-effort kernel conntrack deletion. +pub(in crate::conntrack_control) enum DeleteOutcome { + /// The kernel reported successful deletion. + Deleted, + /// No matching conntrack entry existed. + NotFound, + /// The helper was unavailable or returned an unexpected failure. + Error, +} + +/// Deletes the exact TCP tuple represented by one close event. +pub(in crate::conntrack_control) async fn delete_conntrack_entry( + event: ConntrackCloseEvent, +) -> DeleteOutcome { + let runner = SystemCommandRunner; + if !runner.available("conntrack") { + return DeleteOutcome::Error; + } + let args = vec![ + "-D".to_string(), + "-p".to_string(), + "tcp".to_string(), + "-s".to_string(), + event.src.ip().to_string(), + "--sport".to_string(), + event.src.port().to_string(), + "-d".to_string(), + event.dst.ip().to_string(), + "--dport".to_string(), + event.dst.port().to_string(), + ]; + match runner + .run(CommandSpec { + binary: "conntrack", + args, + stdin: None, + }) + .await + { + Ok(()) => DeleteOutcome::Deleted, + Err(error) if error.message.contains("0 flow entries have been deleted") => { + DeleteOutcome::NotFound + } + Err(error) => { + debug!(error = %error, "conntrack delete failed"); + DeleteOutcome::Error + } + } +} diff --git a/src/conntrack_control/firewall/tests.rs b/src/conntrack_control/firewall/tests.rs new file mode 100644 index 0000000..473caef --- /dev/null +++ b/src/conntrack_control/firewall/tests.rs @@ -0,0 +1,493 @@ +use std::collections::{BTreeMap, BTreeSet}; +use std::future::pending; +use std::sync::{Arc, Mutex}; +use std::sync::atomic::Ordering; +use std::time::Duration; + +use tokio::sync::{Notify, watch}; +use tokio_util::sync::CancellationToken; + +use crate::config::ConntrackBackend; +use crate::stats::Stats; + +use super::actor::{FirewallReconciler, ReconcileOutcome}; +use super::command::{CommandError, CommandErrorKind, CommandSpec, FirewallCommandRunner}; +use super::model::{ + AppliedPlan, AppliedState, DesiredPolicy, DesiredState, NotrackTarget, ShadowSlot, +}; +use super::transaction::{InterruptibleRunner, reconcile_once}; + +#[path = "tests/model_tests.rs"] +mod model_tests; + +#[derive(Clone)] +struct FailureRule { + binary: &'static str, + occurrence: usize, +} + +#[derive(Default)] +struct FakeState { + calls: Vec, + binary_calls: BTreeMap<&'static str, usize>, + failures: Vec, +} + +#[derive(Clone)] +struct FakeRunner { + available: Arc>, + has_cap_net_admin: bool, + state: Arc>, +} + +impl FakeRunner { + fn all_available() -> Self { + Self { + available: Arc::new(BTreeSet::from([ + "conntrack", + "ip6tables", + "ip6tables-restore", + "iptables", + "iptables-restore", + "nft", + ])), + has_cap_net_admin: true, + state: Arc::new(Mutex::new(FakeState::default())), + } + } + + fn with_failure(self, binary: &'static str, occurrence: usize) -> Self { + self.state + .lock() + .unwrap() + .failures + .push(FailureRule { binary, occurrence }); + self + } + + fn calls(&self) -> Vec { + self.state.lock().unwrap().calls.clone() + } +} + +impl FirewallCommandRunner for FakeRunner { + fn available(&self, binary: &str) -> bool { + self.available.contains(binary) + } + + fn has_cap_net_admin(&self) -> bool { + self.has_cap_net_admin + } + + async fn run(&self, spec: CommandSpec) -> Result<(), CommandError> { + let mut state = self.state.lock().unwrap(); + let occurrence = { + let count = state.binary_calls.entry(spec.binary).or_default(); + *count += 1; + *count + }; + state.calls.push(spec.clone()); + if state + .failures + .iter() + .any(|failure| failure.binary == spec.binary && failure.occurrence == occurrence) + { + return Err(CommandError::failed(format!( + "injected {} failure at occurrence {}", + spec.binary, occurrence + ))); + } + drop(state); + + let operation = spec.args.get(2).map(String::as_str); + if (matches!(spec.binary, "iptables" | "ip6tables") + && matches!(operation, Some("-C" | "-D" | "-F" | "-X"))) + || (spec.binary == "nft" + && spec.args.first().map(String::as_str) == Some("delete")) + { + return Err(CommandError { + kind: CommandErrorKind::NotFound, + message: "injected object not found".to_string(), + }); + } + Ok(()) + } +} + +struct BlockingRunner { + entered: Arc, +} + +impl FirewallCommandRunner for BlockingRunner { + fn available(&self, _binary: &str) -> bool { + true + } + + fn has_cap_net_admin(&self) -> bool { + true + } + + async fn run(&self, _spec: CommandSpec) -> Result<(), CommandError> { + self.entered.notify_one(); + pending().await + } +} + +fn target(ip: Option<&str>, port: u16) -> NotrackTarget { + NotrackTarget { + ip: ip.map(|value| value.parse().unwrap()), + port, + } +} + +fn desired(generation: u64, policy: DesiredPolicy) -> DesiredState { + DesiredState { + generation, + policy, + stats: Arc::new(Stats::new()), + } +} + +fn dual_stack_policy(port: u16) -> DesiredPolicy { + DesiredPolicy::Rules { + configured_backend: ConntrackBackend::Iptables, + v4: vec![target(Some("192.0.2.10"), port)], + v6: vec![target(Some("2001:db8::10"), port)], + } +} + +fn nft_dual_stack_policy(port: u16) -> DesiredPolicy { + DesiredPolicy::Rules { + configured_backend: ConntrackBackend::Nftables, + v4: vec![target(Some("192.0.2.10"), port)], + v6: vec![target(Some("2001:db8::10"), port)], + } +} + +#[tokio::test] +async fn successful_reconcile_flips_between_shadow_slots() { + let runner = FakeRunner::all_available(); + let mut applied = AppliedState::Known(AppliedPlan::Empty); + let first = desired(1, dual_stack_policy(443)); + + reconcile_once(&runner, &runner, &mut applied, &first) + .await + .unwrap(); + assert!(matches!( + applied, + AppliedState::Known(AppliedPlan::Iptables { + slot: ShadowSlot::A, + .. + }) + )); + + let second = desired(2, dual_stack_policy(8443)); + reconcile_once(&runner, &runner, &mut applied, &second) + .await + .unwrap(); + assert!(matches!( + applied, + AppliedState::Known(AppliedPlan::Iptables { + slot: ShadowSlot::B, + .. + }) + )); + assert!(runner.calls().iter().any(|call| { + call.stdin + .as_deref() + .is_some_and(|script| script.contains("-A TELEMT_NOTRACK -j TELEMT_NT_B")) + })); +} + +#[tokio::test] +async fn identical_policy_is_command_free_after_convergence() { + let runner = FakeRunner::all_available(); + let mut applied = AppliedState::Known(AppliedPlan::Empty); + reconcile_once( + &runner, + &runner, + &mut applied, + &desired(1, dual_stack_policy(443)), + ) + .await + .unwrap(); + let calls_after_convergence = runner.calls().len(); + + reconcile_once( + &runner, + &runner, + &mut applied, + &desired(2, dual_stack_policy(443)), + ) + .await + .unwrap(); + + assert_eq!(runner.calls().len(), calls_after_convergence); +} + +#[tokio::test] +async fn backend_migration_failure_restores_previous_backend() { + let runner = FakeRunner::all_available().with_failure("ip6tables-restore", 3); + let mut applied = AppliedState::Known(AppliedPlan::Empty); + reconcile_once( + &runner, + &runner, + &mut applied, + &desired(1, dual_stack_policy(443)), + ) + .await + .unwrap(); + let previous = applied.clone(); + + let failure = reconcile_once( + &runner, + &runner, + &mut applied, + &desired(2, nft_dual_stack_policy(8443)), + ) + .await + .unwrap_err(); + + assert_eq!(failure.rollback_succeeded, Some(true)); + assert_eq!(applied, previous); +} + +#[tokio::test] +async fn unavailable_target_does_not_clear_confirmed_applied_policy() { + let runner = FakeRunner::all_available(); + let mut applied = AppliedState::Known(AppliedPlan::Empty); + reconcile_once( + &runner, + &runner, + &mut applied, + &desired(1, dual_stack_policy(443)), + ) + .await + .unwrap(); + let previous = applied.clone(); + let unavailable = FakeRunner { + available: Arc::new(BTreeSet::new()), + has_cap_net_admin: false, + state: Arc::new(Mutex::new(FakeState::default())), + }; + + let failure = reconcile_once( + &unavailable, + &unavailable, + &mut applied, + &desired(2, nft_dual_stack_policy(8443)), + ) + .await + .unwrap_err(); + + assert_eq!(failure.rollback_succeeded, None); + assert_eq!(applied, previous); + assert!(unavailable.calls().is_empty()); +} + +#[tokio::test] +async fn partial_dual_stack_failure_restores_previous_applied_plan() { + let runner = FakeRunner::all_available().with_failure("ip6tables-restore", 4); + let mut applied = AppliedState::Known(AppliedPlan::Empty); + let first = desired(1, dual_stack_policy(443)); + reconcile_once(&runner, &runner, &mut applied, &first) + .await + .unwrap(); + let previous = applied.clone(); + + let failure = reconcile_once( + &runner, + &runner, + &mut applied, + &desired(2, dual_stack_policy(8443)), + ) + .await + .unwrap_err(); + + assert_eq!(failure.rollback_succeeded, Some(true)); + assert_eq!(applied, previous); +} + +#[tokio::test] +async fn rollback_failure_marks_applied_state_unknown() { + let runner = FakeRunner::all_available() + .with_failure("ip6tables-restore", 4) + .with_failure("nft", 2); + let mut applied = AppliedState::Known(AppliedPlan::Empty); + reconcile_once( + &runner, + &runner, + &mut applied, + &desired(1, dual_stack_policy(443)), + ) + .await + .unwrap(); + + let failure = reconcile_once( + &runner, + &runner, + &mut applied, + &desired(2, dual_stack_policy(8443)), + ) + .await + .unwrap_err(); + + assert_eq!(failure.rollback_succeeded, Some(false)); + assert_eq!(applied, AppliedState::Unknown); +} + +#[tokio::test] +async fn transaction_cancellation_does_not_claim_a_new_applied_plan() { + let entered = Arc::new(Notify::new()); + let runner = BlockingRunner { + entered: entered.clone(), + }; + let terminal = CancellationToken::new(); + let process_cancellation = CancellationToken::new(); + let interruptible = InterruptibleRunner::new( + &runner, + &terminal, + &process_cancellation, + ); + let mut applied = AppliedState::Known(AppliedPlan::Empty); + let desired = desired(1, dual_stack_policy(443)); + let failure = { + let transaction = reconcile_once(&interruptible, &runner, &mut applied, &desired); + tokio::pin!(transaction); + + tokio::select! { + _ = entered.notified() => terminal.cancel(), + _ = &mut transaction => panic!("transaction completed before injected cancellation"), + } + transaction.await.unwrap_err() + }; + + assert!(failure.cancelled); + assert_eq!(applied, AppliedState::Known(AppliedPlan::Empty)); +} + +#[test] +fn desired_watch_coalesces_and_rejects_stale_or_conflicting_generations() { + let runner = FakeRunner::all_available(); + let (desired_tx, desired_rx) = watch::channel(None); + let (status_tx, _status_rx) = watch::channel(None); + let terminal = CancellationToken::new(); + let closed = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let completed = Arc::new(Notify::new()); + let mut reconciler = FirewallReconciler::new( + runner, + desired_rx, + status_tx, + terminal, + closed, + Arc::new(std::sync::atomic::AtomicBool::new(false)), + Arc::new(std::sync::atomic::AtomicBool::new(false)), + completed, + ); + + desired_tx.send_replace(Some(desired(1, dual_stack_policy(443)))); + desired_tx.send_replace(Some(desired(2, dual_stack_policy(8443)))); + assert_eq!(reconciler.take_latest_desired().unwrap().generation, 2); + + desired_tx.send_replace(Some(desired(1, dual_stack_policy(443)))); + assert!(reconciler.take_latest_desired().is_none()); + + desired_tx.send_replace(Some(desired(2, dual_stack_policy(9443)))); + assert!(reconciler.take_latest_desired().is_none()); +} + +#[tokio::test(start_paused = true)] +async fn actor_retries_with_backoff_and_cleans_owned_rules_on_shutdown() { + let runner = FakeRunner::all_available().with_failure("iptables-restore", 1); + let observed_runner = runner.clone(); + let (desired_tx, desired_rx) = watch::channel(None); + let (status_tx, mut status_rx) = watch::channel(None); + let terminal = CancellationToken::new(); + let closed = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let completed_flag = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let cleanup_succeeded = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let completed = Arc::new(Notify::new()); + let reconciler = FirewallReconciler::new( + runner, + desired_rx, + status_tx, + terminal.clone(), + closed.clone(), + completed_flag.clone(), + cleanup_succeeded.clone(), + completed, + ); + let process_cancellation = CancellationToken::new(); + let task = tokio::spawn(reconciler.run(process_cancellation)); + + let initial_desired = desired(2, dual_stack_policy(443)); + let desired_stats = initial_desired.stats.clone(); + desired_tx.send_replace(Some(initial_desired)); + status_rx.changed().await.unwrap(); + assert_eq!( + status_rx.borrow().as_ref().unwrap().outcome, + ReconcileOutcome::Failed + ); + let calls_after_failure = observed_runner.calls().len(); + + desired_tx.send_replace(Some(desired(1, dual_stack_policy(7443)))); + tokio::task::yield_now().await; + desired_tx.send_replace(Some(desired(2, dual_stack_policy(9443)))); + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_millis(999)).await; + tokio::task::yield_now().await; + assert_eq!(observed_runner.calls().len(), calls_after_failure); + + tokio::time::advance(Duration::from_millis(1)).await; + status_rx.changed().await.unwrap(); + assert_eq!( + status_rx.borrow().as_ref().unwrap().outcome, + ReconcileOutcome::Applied + ); + assert_eq!(status_rx.borrow().as_ref().unwrap().generation, 2); + assert!(observed_runner.calls().iter().all(|call| { + call.stdin + .as_ref() + .is_none_or(|script| !script.contains("7443") && !script.contains("9443")) + })); + let calls_before_shutdown = observed_runner.calls().len(); + + terminal.cancel(); + tokio::time::timeout(Duration::from_secs(1), task) + .await + .unwrap() + .unwrap(); + assert!(closed.load(Ordering::Acquire)); + assert!(completed_flag.load(Ordering::Acquire)); + assert!(cleanup_succeeded.load(Ordering::Acquire)); + assert!(!desired_stats.get_conntrack_rule_apply_ok()); + assert!(observed_runner.calls().len() > calls_before_shutdown); +} + +#[tokio::test] +async fn actor_reports_terminal_cleanup_failure_separately_from_completion() { + let runner = FakeRunner::all_available().with_failure("nft", 1); + let (_desired_tx, desired_rx) = watch::channel(None); + let (status_tx, _status_rx) = watch::channel(None); + let terminal = CancellationToken::new(); + terminal.cancel(); + let closed = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let completed_flag = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let cleanup_succeeded = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let reconciler = FirewallReconciler::new( + runner, + desired_rx, + status_tx, + terminal, + closed.clone(), + completed_flag.clone(), + cleanup_succeeded.clone(), + Arc::new(Notify::new()), + ); + + reconciler.run(CancellationToken::new()).await; + + assert!(closed.load(Ordering::Acquire)); + assert!(completed_flag.load(Ordering::Acquire)); + assert!(!cleanup_succeeded.load(Ordering::Acquire)); +} diff --git a/src/conntrack_control/firewall/tests/model_tests.rs b/src/conntrack_control/firewall/tests/model_tests.rs new file mode 100644 index 0000000..c24434d --- /dev/null +++ b/src/conntrack_control/firewall/tests/model_tests.rs @@ -0,0 +1,123 @@ +use crate::config::{ConntrackBackend, ConntrackMode, ProxyConfig}; + +use super::super::command::is_not_found_error; +use super::super::iptables::{self, is_chain_exists_error}; +use super::super::model::{DesiredPolicy, ShadowSlot}; +use super::super::nftables; +use super::target; + +#[test] +fn desired_policy_derives_exact_listener_targets() { + let mut config = ProxyConfig::default(); + config.server.port = 8443; + config.server.listen_addr_ipv4 = Some("0.0.0.0".to_string()); + config.server.listen_addr_ipv6 = Some("2001:db8::10".to_string()); + config.server.conntrack_control.inline_conntrack_control = true; + config.server.conntrack_control.mode = ConntrackMode::Notrack; + config.server.conntrack_control.backend = ConntrackBackend::Iptables; + + assert_eq!( + DesiredPolicy::from_config(&config), + DesiredPolicy::Rules { + configured_backend: ConntrackBackend::Iptables, + v4: vec![target(None, 8443)], + v6: vec![target(Some("2001:db8::10"), 8443)], + } + ); + + config.server.conntrack_control.mode = ConntrackMode::Tracked; + assert_eq!(DesiredPolicy::from_config(&config), DesiredPolicy::Empty); + config.server.conntrack_control.mode = ConntrackMode::Notrack; + config.server.conntrack_control.inline_conntrack_control = false; + assert_eq!(DesiredPolicy::from_config(&config), DesiredPolicy::Empty); +} + +#[test] +fn hybrid_policy_is_a_sorted_deduplicated_address_port_product() { + let mut config = ProxyConfig::default(); + config.server.conntrack_control.inline_conntrack_control = true; + config.server.conntrack_control.mode = ConntrackMode::Hybrid; + config.server.conntrack_control.hybrid_listener_ips = vec![ + "2001:db8::10".parse().unwrap(), + "192.0.2.10".parse().unwrap(), + "192.0.2.10".parse().unwrap(), + ]; + config.server.listeners = vec![ + serde_json::from_value(serde_json::json!({ + "ip": "0.0.0.0", + "port": 8443 + })) + .unwrap(), + serde_json::from_value(serde_json::json!({ + "ip": "::", + "port": 443 + })) + .unwrap(), + ]; + + assert_eq!( + DesiredPolicy::from_config(&config), + DesiredPolicy::Rules { + configured_backend: ConntrackBackend::Auto, + v4: vec![ + target(Some("192.0.2.10"), 443), + target(Some("192.0.2.10"), 8443), + ], + v6: vec![ + target(Some("2001:db8::10"), 443), + target(Some("2001:db8::10"), 8443), + ], + } + ); +} + +#[test] +fn restore_renderers_keep_staging_detached_from_activation() { + let stage = iptables::render_stage_script( + ShadowSlot::B, + &[target(Some("192.0.2.20"), 443)], + ); + assert!(stage.contains("-F TELEMT_NT_B\n")); + assert!(stage.contains( + "-A TELEMT_NT_B -p tcp --dport 443 -d 192.0.2.20 -j CT --notrack\n" + )); + assert!(!stage.contains("-A TELEMT_NOTRACK -j TELEMT_NT_B")); + assert!(!stage.contains(":TELEMT_")); + + let activation = iptables::render_dispatch_script(Some(ShadowSlot::B)); + assert!(activation.contains("-F TELEMT_NOTRACK\n")); + assert!(activation.contains("-A TELEMT_NOTRACK -j TELEMT_NT_B\n")); + assert!(!activation.contains(":TELEMT_")); + + let nft_stage = nftables::render_stage_script( + ShadowSlot::A, + &[target(None, 443)], + &[target(Some("2001:db8::20"), 8443)], + ); + assert!(!nft_stage.contains("hook prerouting")); + assert!(nft_stage.contains("tcp dport 443 notrack\n")); + assert!(nft_stage.contains("tcp dport 8443 ip6 daddr 2001:db8::20 notrack\n")); + assert!(nftables::render_activate_script(ShadowSlot::A).contains("hook prerouting")); +} + +#[test] +fn command_error_classification_only_accepts_absent_owned_objects() { + assert!(is_not_found_error( + "iptables: No chain/target/match by that name." + )); + assert!(is_not_found_error( + "Bad rule (does a matching rule exist in that chain?)." + )); + assert!(is_not_found_error( + "Error: Could not process rule: No such file or directory" + )); + assert!(!is_not_found_error("Permission denied")); + assert!(!is_not_found_error( + "can't initialize iptables table `raw': Table does not exist" + )); + assert!(!is_not_found_error( + "Another app is currently holding the xtables lock" + )); + assert!(is_chain_exists_error("iptables: Chain already exists.")); + assert!(!is_chain_exists_error("Permission denied")); +} diff --git a/src/conntrack_control/firewall/transaction.rs b/src/conntrack_control/firewall/transaction.rs new file mode 100644 index 0000000..3220fe2 --- /dev/null +++ b/src/conntrack_control/firewall/transaction.rs @@ -0,0 +1,333 @@ +use std::time::Duration; + +use tokio_util::sync::CancellationToken; + +use crate::config::ConntrackBackend; + +use super::command::{ + CommandError, CommandErrorKind, CommandSpec, FirewallCommandRunner, +}; +use super::iptables::{self, IpFamily}; +use super::model::{AppliedPlan, AppliedState, DesiredPolicy, DesiredState, ShadowSlot}; +use super::nftables; + +const TRANSACTION_TIMEOUT: Duration = Duration::from_secs(30); + +pub(super) struct InterruptibleRunner<'a, R> { + inner: &'a R, + terminal: &'a CancellationToken, + process_cancellation: &'a CancellationToken, +} + +impl<'a, R> InterruptibleRunner<'a, R> { + pub(super) fn new( + inner: &'a R, + terminal: &'a CancellationToken, + process_cancellation: &'a CancellationToken, + ) -> Self { + Self { + inner, + terminal, + process_cancellation, + } + } +} + +impl FirewallCommandRunner for InterruptibleRunner<'_, R> { + fn available(&self, binary: &str) -> bool { + self.inner.available(binary) + } + + fn has_cap_net_admin(&self) -> bool { + self.inner.has_cap_net_admin() + } + + async fn run(&self, spec: CommandSpec) -> Result<(), CommandError> { + tokio::select! { + biased; + _ = self.terminal.cancelled() => Err(CommandError::cancelled()), + _ = self.process_cancellation.cancelled() => Err(CommandError::cancelled()), + result = self.inner.run(spec) => result, + } + } +} + +#[derive(Debug)] +pub(super) struct ReconcileFailure { + pub(super) message: String, + pub(super) rollback_succeeded: Option, + pub(super) cancelled: bool, +} + +pub(super) async fn reconcile_once( + interruptible: &I, + recovery_runner: &R, + applied: &mut AppliedState, + desired: &DesiredState, +) -> Result<(), ReconcileFailure> +where + I: FirewallCommandRunner, + R: FirewallCommandRunner, +{ + if let AppliedState::Known(current) = applied + && current.matches_policy(&desired.policy) + { + return Ok(()); + } + + if matches!(applied, AppliedState::Unknown) { + match tokio::time::timeout(TRANSACTION_TIMEOUT, recover_to_empty(interruptible)).await { + Ok(Ok(())) => *applied = AppliedState::Known(AppliedPlan::Empty), + Ok(Err(error)) if error.kind == CommandErrorKind::Cancelled => { + return Err(cancelled_failure()); + } + Ok(Err(error)) => { + return Err(ReconcileFailure { + message: format!("startup recovery failed: {error}"), + rollback_succeeded: None, + cancelled: false, + }); + } + Err(_) => { + return Err(ReconcileFailure { + message: "startup recovery timed out".to_string(), + rollback_succeeded: None, + cancelled: false, + }); + } + } + } + + let previous = match applied { + AppliedState::Known(plan) => plan.clone(), + AppliedState::Unknown => unreachable!("unknown state was recovered above"), + }; + let target = match resolve_target(interruptible, &desired.policy, &previous) { + Ok(target) => target, + Err(error) => { + return Err(ReconcileFailure { + message: error.message, + rollback_succeeded: None, + cancelled: false, + }); + } + }; + + let transition = tokio::time::timeout( + TRANSACTION_TIMEOUT, + transition_plan(interruptible, &previous, &target), + ) + .await; + match transition { + Ok(Ok(())) => { + *applied = AppliedState::Known(target); + Ok(()) + } + Ok(Err(error)) if error.kind == CommandErrorKind::Cancelled => Err(cancelled_failure()), + Ok(Err(error)) => { + rollback_after_failure(recovery_runner, applied, previous, error.message).await + } + Err(_) => { + rollback_after_failure( + recovery_runner, + applied, + previous, + "firewall apply transaction timed out".to_string(), + ) + .await + } + } +} + +async fn rollback_after_failure( + runner: &R, + applied: &mut AppliedState, + previous: AppliedPlan, + apply_error: String, +) -> Result<(), ReconcileFailure> { + let rollback = tokio::time::timeout(TRANSACTION_TIMEOUT, restore_plan(runner, &previous)).await; + let rollback_result = match rollback { + Ok(result) => result, + Err(_) => Err(CommandError::failed("firewall rollback timed out")), + }; + match rollback_result { + Ok(()) => { + *applied = AppliedState::Known(previous); + Err(ReconcileFailure { + message: apply_error, + rollback_succeeded: Some(true), + cancelled: false, + }) + } + Err(rollback_error) => { + *applied = AppliedState::Unknown; + if rollback_error.kind == CommandErrorKind::Cancelled { + return Err(cancelled_failure()); + } + Err(ReconcileFailure { + message: format!("{apply_error}; rollback failed: {rollback_error}"), + rollback_succeeded: Some(false), + cancelled: false, + }) + } + } +} + +fn cancelled_failure() -> ReconcileFailure { + ReconcileFailure { + message: "firewall transaction cancelled for process shutdown".to_string(), + rollback_succeeded: None, + cancelled: true, + } +} + +pub(super) fn resolve_target( + runner: &R, + desired: &DesiredPolicy, + previous: &AppliedPlan, +) -> Result { + let DesiredPolicy::Rules { + configured_backend, + v4, + v6, + } = desired + else { + return Ok(AppliedPlan::Empty); + }; + if !runner.has_cap_net_admin() { + return Err(CommandError::failed( + "CAP_NET_ADMIN is required for conntrack firewall reconciliation", + )); + } + let next_slot = previous.slot().map_or(ShadowSlot::A, ShadowSlot::other); + let iptables_available = (v4.is_empty() || iptables::family_available(runner, IpFamily::V4)) + && (v6.is_empty() || iptables::family_available(runner, IpFamily::V6)); + match configured_backend { + ConntrackBackend::Nftables if nftables::available(runner) => Ok(AppliedPlan::Nftables { + slot: next_slot, + v4: v4.clone(), + v6: v6.clone(), + }), + ConntrackBackend::Iptables if iptables_available => Ok(AppliedPlan::Iptables { + slot: next_slot, + v4: v4.clone(), + v6: v6.clone(), + }), + ConntrackBackend::Auto if nftables::available(runner) => Ok(AppliedPlan::Nftables { + slot: next_slot, + v4: v4.clone(), + v6: v6.clone(), + }), + ConntrackBackend::Auto if iptables_available => Ok(AppliedPlan::Iptables { + slot: next_slot, + v4: v4.clone(), + v6: v6.clone(), + }), + backend => Err(CommandError::failed(format!( + "configured conntrack firewall backend {backend:?} is unavailable" + ))), + } +} + +pub(super) async fn transition_plan( + runner: &R, + previous: &AppliedPlan, + target: &AppliedPlan, +) -> Result<(), CommandError> { + install_plan(runner, target).await?; + match (previous, target) { + ( + AppliedPlan::Iptables { + v4: previous_v4, + v6: previous_v6, + .. + }, + AppliedPlan::Iptables { v4, v6, .. }, + ) => { + if !previous_v4.is_empty() && v4.is_empty() { + iptables::activate_family(runner, IpFamily::V4, None).await?; + } + if !previous_v6.is_empty() && v6.is_empty() { + iptables::activate_family(runner, IpFamily::V6, None).await?; + } + } + (AppliedPlan::Nftables { slot: previous_slot, .. }, AppliedPlan::Nftables { slot, .. }) + if previous_slot != slot => + { + nftables::deactivate(runner, *previous_slot).await?; + } + (_, _) if previous != target => clear_plan(runner, previous).await?, + _ => {} + } + Ok(()) +} + +async fn install_plan( + runner: &R, + plan: &AppliedPlan, +) -> Result<(), CommandError> { + match plan { + AppliedPlan::Empty => Ok(()), + AppliedPlan::Iptables { slot, v4, v6 } => { + if !v4.is_empty() { + iptables::stage_family(runner, IpFamily::V4, *slot, v4).await?; + } + if !v6.is_empty() { + iptables::stage_family(runner, IpFamily::V6, *slot, v6).await?; + } + if !v4.is_empty() { + iptables::activate_family(runner, IpFamily::V4, Some(*slot)).await?; + } + if !v6.is_empty() { + iptables::activate_family(runner, IpFamily::V6, Some(*slot)).await?; + } + Ok(()) + } + AppliedPlan::Nftables { slot, v4, v6 } => { + nftables::stage(runner, *slot, v4, v6).await?; + nftables::activate(runner, *slot).await + } + } +} + +async fn clear_plan( + runner: &R, + plan: &AppliedPlan, +) -> Result<(), CommandError> { + match plan { + AppliedPlan::Empty => Ok(()), + AppliedPlan::Iptables { v4, v6, .. } => { + if !v4.is_empty() { + iptables::activate_family(runner, IpFamily::V4, None).await?; + } + if !v6.is_empty() { + iptables::activate_family(runner, IpFamily::V6, None).await?; + } + Ok(()) + } + AppliedPlan::Nftables { slot, .. } => nftables::deactivate(runner, *slot).await, + } +} + +async fn restore_plan( + runner: &R, + plan: &AppliedPlan, +) -> Result<(), CommandError> { + recover_to_empty(runner).await?; + install_plan(runner, plan).await +} + +pub(super) async fn recover_to_empty( + runner: &R, +) -> Result<(), CommandError> { + let nft_result = nftables::cleanup_all(runner).await; + let iptables_result = iptables::cleanup_all(runner).await; + match (nft_result, iptables_result) { + (Ok(()), Ok(())) => Ok(()), + (Err(first), Ok(())) | (Ok(()), Err(first)) => Err(first), + (Err(first), Err(second)) => Err(CommandError::failed(format!( + "{}; {}", + first.message, second.message + ))), + } +} diff --git a/src/maestro/control_plane.rs b/src/maestro/control_plane.rs index c8f6f42..5d4c60d 100644 --- a/src/maestro/control_plane.rs +++ b/src/maestro/control_plane.rs @@ -133,6 +133,21 @@ impl ProcessControlPlane { Ok(()) } + /// Registers a cooperatively cancelled task whose cleanup future must finish. + pub(crate) fn spawn_cooperative(&self, spawn: S) -> Result<(), S> + where + S: FnOnce(CancellationToken) -> F, + F: Future + Send + 'static, + { + let Some(registration) = self.inner.admission.try_register() else { + return Err(spawn); + }; + let cancellation = self.inner.cancellation.clone(); + self.inner.tasks.spawn(spawn(cancellation)); + drop(registration); + Ok(()) + } + /// Closes task admission, cancels all owned work, and joins it within the deadline. pub(crate) async fn shutdown(&self, timeout: Duration) -> bool { let deadline = tokio::time::Instant::now() + timeout; @@ -254,4 +269,22 @@ mod tests { assert!(shutdown.await.unwrap()); assert!(completed.load(Ordering::Acquire)); } + + #[tokio::test] + async fn cooperative_task_observes_cancellation_and_finishes_cleanup() { + let scope = ProcessControlPlane::new(); + let completed = Arc::new(AtomicBool::new(false)); + let completed_task = Arc::clone(&completed); + assert!( + scope + .spawn_cooperative(move |cancellation| async move { + cancellation.cancelled().await; + completed_task.store(true, Ordering::Release); + }) + .is_ok() + ); + + assert!(scope.shutdown(Duration::from_secs(1)).await); + assert!(completed.load(Ordering::Acquire)); + } } diff --git a/src/maestro/orchestrator.rs b/src/maestro/orchestrator.rs index b0e832e..db7fec1 100644 --- a/src/maestro/orchestrator.rs +++ b/src/maestro/orchestrator.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use arc_swap::ArcSwap; use tokio::sync::{RwLock, Semaphore, watch}; -use tracing::{error, info}; +use tracing::{error, info, warn}; use crate::api; use crate::ip_tracker::UserIpTracker; @@ -40,7 +40,7 @@ pub(super) async fn run_telemt_core( process_started_at, process_started_at_epoch_secs, startup_tracker, - config, + mut config, config_path, has_rust_log, effective_log_level, @@ -48,6 +48,13 @@ pub(super) async fn run_telemt_core( logging_guard: _logging_guard, } = bootstrap::bootstrap(privilege_drop_requested).await?; + if privilege_drop_requested && config.server.conntrack_control.inline_conntrack_control { + warn!( + "Inline conntrack control is disabled when process privileges are dropped" + ); + config.server.conntrack_control.inline_conntrack_control = false; + } + let quota_store = Arc::new(QuotaStore::default()); let connection_authority = Arc::new(UserConnectionAuthority::default()); let stats = Arc::new(Stats::with_process_authorities( @@ -367,9 +374,26 @@ pub(super) async fn run_telemt_core( .await .map_err(std::io::Error::other)?; + #[cfg(target_os = "linux")] + let conntrack_firewall = { + let authority = crate::conntrack_control::FirewallAuthority::spawn(&process_control_plane) + .map_err(std::io::Error::other)?; + if !authority + .publish_initial(1, runtime.config.clone(), stats.clone()) + .await + { + warn!( + "Initial conntrack firewall reconciliation failed; background retries remain active" + ); + } + Some(authority) + }; + #[cfg(not(target_os = "linux"))] + let conntrack_firewall = None::; + drop_after_bind(); - runtime_tasks::spawn_metrics_if_configured( + if let Err(error) = runtime_tasks::spawn_metrics_if_configured( &runtime.config, &startup_tracker, active_runtime.clone(), @@ -377,7 +401,15 @@ pub(super) async fn run_telemt_core( tls_full_cert_budget.clone(), process_control_plane.clone(), ) - .await?; + .await + { + if let Some(conntrack_firewall) = &conntrack_firewall + && !conntrack_firewall.shutdown_and_clear().await + { + warn!("Conntrack firewall cleanup failed after metrics startup error"); + } + return Err(error.into()); + } runtime_watch_tx.send_replace(Some(active_runtime.load_full().watch_state())); active_runtime_tx.send_replace(Some(active_runtime.clone())); @@ -401,6 +433,7 @@ pub(super) async fn run_telemt_core( runtime_watch_tx, listener_manager, web_trace, + conntrack_firewall.clone(), ); shutdown::spawn_signal_handlers( @@ -413,6 +446,7 @@ pub(super) async fn run_telemt_core( active_runtime, quota_state, reload_supervisor, + conntrack_firewall, process_control_plane, ) .await; diff --git a/src/maestro/reload_supervisor.rs b/src/maestro/reload_supervisor.rs index c1e3998..22956a4 100644 --- a/src/maestro/reload_supervisor.rs +++ b/src/maestro/reload_supervisor.rs @@ -8,6 +8,7 @@ use tokio::sync::watch; use tokio_util::sync::CancellationToken; use tracing::{info, warn}; +use crate::conntrack_control::FirewallAuthority; use crate::stats::QuotaStore; use crate::tls_front::cache::TlsFullCertBudget; use crate::web::trace::WebTraceStore; @@ -33,6 +34,7 @@ pub(crate) struct ReloadSupervisor { runtime_watch_tx: watch::Sender>, listener_manager: Arc>, web_trace: Arc, + conntrack_firewall: Option, } /// Process-owned handle that quiesces reloads before shutdown snapshots the runtime. @@ -106,6 +108,7 @@ impl ReloadSupervisor { runtime_watch_tx: watch::Sender>, listener_manager: ListenerManager, web_trace: Arc, + conntrack_firewall: Option, ) -> ReloadSupervisorHandle { let listener_manager = Arc::new(Mutex::new(listener_manager)); let supervisor = Self { @@ -120,6 +123,7 @@ impl ReloadSupervisor { runtime_watch_tx, listener_manager: listener_manager.clone(), web_trace, + conntrack_firewall, }; let control = supervisor.control.clone(); let shutdown = CancellationToken::new(); @@ -339,6 +343,14 @@ impl ReloadSupervisor { old_runtime.stop_accepting_sessions(); listener_manager.activate_runtime_generation(new_runtime.clone()) }; + let conntrack_firewall_published = match &self.conntrack_firewall { + Some(conntrack_firewall) => conntrack_firewall.publish( + new_runtime.id, + new_runtime.config(), + new_runtime.stats.clone(), + ), + None => true, + }; self.web_trace .apply_policy(new_runtime.id, &new_runtime.config().web.debug); config_watcher_activation.send_replace(true); @@ -353,6 +365,12 @@ impl ReloadSupervisor { .apply_reload(&new_runtime.config().general.log_level); self.runtime_watch_tx .send_replace(Some(new_runtime.watch_state())); + if !conntrack_firewall_published { + let warning = "conntrack firewall reconciler is unavailable after runtime activation" + .to_string(); + warn!(reload_id = command.reload_id, warning = %warning); + self.control.add_warning(command.reload_id, warning).await; + } info!( reload_id = command.reload_id, diff --git a/src/maestro/reload_supervisor_tests.rs b/src/maestro/reload_supervisor_tests.rs index 57f6adf..2842409 100644 --- a/src/maestro/reload_supervisor_tests.rs +++ b/src/maestro/reload_supervisor_tests.rs @@ -61,6 +61,7 @@ async fn fixture(request: ReloadRequest) -> ReloadFixture { runtime_watch_tx, listener_manager, web_trace, + conntrack_firewall: None, }); let command = ReloadCommand { reload_id: accepted.reload_id, @@ -275,6 +276,7 @@ async fn quiesce_joins_idle_supervisor_and_rejects_later_submissions() { runtime.config().web.debug.clone(), &runtime.config().web.limits, ), + None, ); tokio::time::timeout(Duration::from_secs(1), handle.quiesce()) diff --git a/src/maestro/runtime_build.rs b/src/maestro/runtime_build.rs index ea2221a..5372012 100644 --- a/src/maestro/runtime_build.rs +++ b/src/maestro/runtime_build.rs @@ -408,6 +408,12 @@ pub(crate) fn resolve_reload_config( fields.push("server.max_connections".to_string()); effective.server.max_connections = old.server.max_connections; } + if serde_json::to_value(&old.server.conntrack_control).ok() + != serde_json::to_value(&desired.server.conntrack_control).ok() + { + fields.push("server.conntrack_control".to_string()); + effective.server.conntrack_control = old.server.conntrack_control.clone(); + } if old.general.direct_relay_buffer_budget_max_bytes != desired.general.direct_relay_buffer_budget_max_bytes { diff --git a/src/maestro/runtime_build_tests.rs b/src/maestro/runtime_build_tests.rs index 38b7f87..f309c4a 100644 --- a/src/maestro/runtime_build_tests.rs +++ b/src/maestro/runtime_build_tests.rs @@ -127,6 +127,42 @@ fn process_wide_connection_and_direct_buffer_envelopes_are_restart_only() { assert!(!resolved.runtime_changed); } +#[test] +fn conntrack_control_policy_is_restart_only_as_one_process_owned_unit() { + let old = ProxyConfig::default(); + let mut desired = old.clone(); + desired.server.conntrack_control.inline_conntrack_control = + !old.server.conntrack_control.inline_conntrack_control; + desired.server.conntrack_control.mode = crate::config::ConntrackMode::Notrack; + desired.server.conntrack_control.backend = crate::config::ConntrackBackend::Iptables; + desired.server.conntrack_control.profile = + crate::config::ConntrackPressureProfile::Aggressive; + desired.server.conntrack_control.hybrid_listener_ips = + vec!["192.0.2.10".parse().unwrap()]; + desired + .server + .conntrack_control + .pressure_high_watermark_pct = 90; + desired.server.conntrack_control.pressure_low_watermark_pct = 40; + desired.server.conntrack_control.delete_budget_per_sec = old + .server + .conntrack_control + .delete_budget_per_sec + .saturating_add(1); + + let resolved = resolve_reload_config(&old, &desired).unwrap(); + + assert_eq!( + resolved.deferred_process_fields, + vec!["server.conntrack_control".to_string()] + ); + assert_eq!( + serde_json::to_value(&resolved.effective.server.conntrack_control).unwrap(), + serde_json::to_value(&old.server.conntrack_control).unwrap() + ); + assert!(!resolved.runtime_changed); +} + #[test] fn mixed_reload_retains_process_state_and_applies_runtime_state() { let old = ProxyConfig::default(); diff --git a/src/maestro/shutdown.rs b/src/maestro/shutdown.rs index 226060e..6264d7d 100644 --- a/src/maestro/shutdown.rs +++ b/src/maestro/shutdown.rs @@ -23,6 +23,7 @@ use super::control_plane::ProcessControlPlane; use super::generation::RuntimeGeneration; use super::helpers::{format_uptime, unit_label}; use super::reload_supervisor::ReloadSupervisorHandle; +use crate::conntrack_control::FirewallAuthority; use crate::quota_state::QuotaStateOwner; use crate::stats::Stats; use crate::synlimit_control; @@ -54,6 +55,7 @@ pub(crate) async fn wait_for_shutdown( active_runtime: Arc>, quota_state: Arc, reload_supervisor: ReloadSupervisorHandle, + conntrack_firewall: Option, process_control_plane: ProcessControlPlane, ) { let signal = wait_for_shutdown_signal().await; @@ -63,6 +65,7 @@ pub(crate) async fn wait_for_shutdown( active_runtime, quota_state, reload_supervisor, + conntrack_firewall, process_control_plane, ) .await; @@ -95,6 +98,7 @@ async fn perform_shutdown( active_runtime: Arc>, quota_state: Arc, reload_supervisor: ReloadSupervisorHandle, + conntrack_firewall: Option, process_control_plane: ProcessControlPlane, ) { let shutdown_started_at = Instant::now(); @@ -126,6 +130,12 @@ async fn perform_shutdown( warn!("ME shutdown: pool lifecycle deadline expired"); } + if let Some(conntrack_firewall) = conntrack_firewall + && !conntrack_firewall.shutdown_and_clear().await + { + warn!("Conntrack firewall cleanup did not complete successfully"); + } + if let Err(error) = synlimit_control::clear_synlimit_rules_all_backends().await { warn!(error = %error, "Failed to clear SYN limiter rules during shutdown"); } diff --git a/src/metrics/render/connections.rs b/src/metrics/render/connections.rs index f7c2ae0..2285291 100644 --- a/src/metrics/render/connections.rs +++ b/src/metrics/render/connections.rs @@ -266,6 +266,53 @@ pub(super) fn render( } ); + let _ = writeln!( + out, + "# HELP telemt_conntrack_rule_reconcile_total Conntrack firewall reconciliations by result" + ); + let _ = writeln!(out, "# TYPE telemt_conntrack_rule_reconcile_total counter"); + let _ = writeln!( + out, + "telemt_conntrack_rule_reconcile_total{{result=\"success\"}} {}", + if core_enabled { + stats.get_conntrack_rule_reconcile_success_total() + } else { + 0 + } + ); + let _ = writeln!( + out, + "telemt_conntrack_rule_reconcile_total{{result=\"error\"}} {}", + if core_enabled { + stats.get_conntrack_rule_reconcile_error_total() + } else { + 0 + } + ); + let _ = writeln!( + out, + "# HELP telemt_conntrack_rule_rollback_total Conntrack firewall rollbacks by result" + ); + let _ = writeln!(out, "# TYPE telemt_conntrack_rule_rollback_total counter"); + let _ = writeln!( + out, + "telemt_conntrack_rule_rollback_total{{result=\"success\"}} {}", + if core_enabled { + stats.get_conntrack_rule_rollback_success_total() + } else { + 0 + } + ); + let _ = writeln!( + out, + "telemt_conntrack_rule_rollback_total{{result=\"error\"}} {}", + if core_enabled { + stats.get_conntrack_rule_rollback_error_total() + } else { + 0 + } + ); + let _ = writeln!( out, "# HELP telemt_conntrack_event_queue_depth Pending close events in conntrack control queue" diff --git a/src/metrics/tests.rs b/src/metrics/tests.rs index 2b57ed6..c1932b8 100644 --- a/src/metrics/tests.rs +++ b/src/metrics/tests.rs @@ -43,6 +43,10 @@ async fn test_render_metrics_format() { stats.increment_connects_bad_with_class("tls_handshake_bad_client"); stats.increment_handshake_timeouts(); stats.increment_handshake_failure_class("timeout"); + stats.increment_conntrack_rule_reconcile_success_total(); + stats.increment_conntrack_rule_reconcile_error_total(); + stats.increment_conntrack_rule_rollback_success_total(); + stats.increment_conntrack_rule_rollback_error_total(); shared_state .handshake .auth_expensive_checks_total @@ -123,6 +127,12 @@ async fn test_render_metrics_format() { ); assert!(output.contains("telemt_handshake_timeouts_total 1")); assert!(output.contains("telemt_handshake_failures_by_class_total{class=\"timeout\"} 1")); + assert!( + output.contains("telemt_conntrack_rule_reconcile_total{result=\"success\"} 1") + ); + assert!(output.contains("telemt_conntrack_rule_reconcile_total{result=\"error\"} 1")); + assert!(output.contains("telemt_conntrack_rule_rollback_total{result=\"success\"} 1")); + assert!(output.contains("telemt_conntrack_rule_rollback_total{result=\"error\"} 1")); assert!(output.contains("telemt_auth_expensive_checks_total 9")); assert!(output.contains("telemt_auth_budget_exhausted_total 2")); assert!(output.contains("telemt_upstream_connect_attempt_total 2")); diff --git a/src/stats/core_counters.rs b/src/stats/core_counters.rs index 9017270..819324a 100644 --- a/src/stats/core_counters.rs +++ b/src/stats/core_counters.rs @@ -135,6 +135,38 @@ impl Stats { .store(ok, Ordering::Relaxed); } + /// Records one successful process-owned firewall reconciliation. + pub fn increment_conntrack_rule_reconcile_success_total(&self) { + if self.telemetry_core_enabled() { + self.conntrack_rule_reconcile_success_total + .fetch_add(1, Ordering::Relaxed); + } + } + + /// Records one failed process-owned firewall reconciliation. + pub fn increment_conntrack_rule_reconcile_error_total(&self) { + if self.telemetry_core_enabled() { + self.conntrack_rule_reconcile_error_total + .fetch_add(1, Ordering::Relaxed); + } + } + + /// Records one successful firewall transaction rollback. + pub fn increment_conntrack_rule_rollback_success_total(&self) { + if self.telemetry_core_enabled() { + self.conntrack_rule_rollback_success_total + .fetch_add(1, Ordering::Relaxed); + } + } + + /// Records one rollback failure that leaves applied firewall state unknown. + pub fn increment_conntrack_rule_rollback_error_total(&self) { + if self.telemetry_core_enabled() { + self.conntrack_rule_rollback_error_total + .fetch_add(1, Ordering::Relaxed); + } + } + pub fn increment_conntrack_delete_attempt_total(&self) { if self.telemetry_core_enabled() { self.conntrack_delete_attempt_total diff --git a/src/stats/core_getters.rs b/src/stats/core_getters.rs index c1730c1..01d4437 100644 --- a/src/stats/core_getters.rs +++ b/src/stats/core_getters.rs @@ -84,6 +84,26 @@ impl Stats { pub fn get_conntrack_rule_apply_ok(&self) -> bool { self.conntrack_rule_apply_ok_gauge.load(Ordering::Relaxed) } + /// Returns successful process-owned firewall reconciliations. + pub fn get_conntrack_rule_reconcile_success_total(&self) -> u64 { + self.conntrack_rule_reconcile_success_total + .load(Ordering::Relaxed) + } + /// Returns failed process-owned firewall reconciliations. + pub fn get_conntrack_rule_reconcile_error_total(&self) -> u64 { + self.conntrack_rule_reconcile_error_total + .load(Ordering::Relaxed) + } + /// Returns successful firewall transaction rollbacks. + pub fn get_conntrack_rule_rollback_success_total(&self) -> u64 { + self.conntrack_rule_rollback_success_total + .load(Ordering::Relaxed) + } + /// Returns rollback failures that left applied firewall state unknown. + pub fn get_conntrack_rule_rollback_error_total(&self) -> u64 { + self.conntrack_rule_rollback_error_total + .load(Ordering::Relaxed) + } pub fn get_conntrack_delete_attempt_total(&self) -> u64 { self.conntrack_delete_attempt_total.load(Ordering::Relaxed) } diff --git a/src/stats/mod.rs b/src/stats/mod.rs index f751a10..6f7be67 100644 --- a/src/stats/mod.rs +++ b/src/stats/mod.rs @@ -150,6 +150,10 @@ pub struct Stats { conntrack_pressure_active_gauge: AtomicBool, conntrack_event_queue_depth_gauge: AtomicU64, conntrack_rule_apply_ok_gauge: AtomicBool, + conntrack_rule_reconcile_success_total: AtomicU64, + conntrack_rule_reconcile_error_total: AtomicU64, + conntrack_rule_rollback_success_total: AtomicU64, + conntrack_rule_rollback_error_total: AtomicU64, conntrack_delete_attempt_total: AtomicU64, conntrack_delete_success_total: AtomicU64, conntrack_delete_not_found_total: AtomicU64, diff --git a/src/util/trusted_command.rs b/src/util/trusted_command.rs index ab677be..cb40cdd 100644 --- a/src/util/trusted_command.rs +++ b/src/util/trusted_command.rs @@ -2,10 +2,12 @@ use std::os::unix::fs::{MetadataExt, PermissionsExt}; use std::path::{Path, PathBuf}; const TRUSTED_HELPER_DIRS: [&str; 4] = ["/usr/sbin", "/usr/bin", "/sbin", "/bin"]; -const TRUSTED_HELPERS: [&str; 10] = [ +const TRUSTED_HELPERS: [&str; 12] = [ "nft", "iptables", + "iptables-restore", "ip6tables", + "ip6tables-restore", "conntrack", "pfctl", "systemctl", @@ -69,6 +71,13 @@ mod tests { assert!(resolve_trusted_helper("../bin/nft").is_none()); } + #[test] + fn privileged_helper_allowlist_accepts_exact_restore_names() { + assert!(TRUSTED_HELPERS.contains(&"iptables-restore")); + assert!(TRUSTED_HELPERS.contains(&"ip6tables-restore")); + assert!(!TRUSTED_HELPERS.contains(&"iptables-restore-wrapper")); + } + #[test] fn writable_executable_is_not_trusted() { let directory = tempfile::tempdir().unwrap();