fix(firewall): replace nftables tables in one transaction [P0]

Both owned tables were torn down before the replacement was applied, so a failure in
between left the host with neither.

Host firewall: apply_host_firewall ran `nft delete table` as its own command and only then
piped the new ruleset to `nft -f -`. Any failure in that second step - a syntax error, a
transient netlink error, the process dying - left the host with no INPUT policy and no
allow rules at all. That is fail-open on the host firewall, reached by a single failed
command. The code comment even claimed a rejected batch could not take the existing rules
down, which stopped being true the moment the delete moved out of the batch.

Anti-torrent was worse: delete, then four separate commands to rebuild table, counter,
chain and rule. A failure partway through left a half-built table - a table and counter
with no chain, or a chain with no rule - which the inspector reports as drift but which
provides no protection in the meantime.

Both now emit delete and rebuild as a single `nft -f` script. nft applies a script
atomically, so a rejected ruleset leaves the previous one exactly as it was. The delete
line is only included when the table actually exists, because `delete table` on a missing
table aborts the whole transaction - table_exists in nft.rs answers that, treating a
not-found error as absent and any other error as a real failure.

Rendering the anti-torrent table as a script also collapses four command invocations and
their argument-array plumbing into one string, and removes run_step, which had no other
user.

Four tests: for each table, that the script omits the delete when nothing is there and
that delete precedes the rebuild in the same script when it is; plus that the anti-torrent
script creates every object the inspector later requires, checked element by element
rather than through the listing-format contract matcher.

Not changed, deliberately: tc has no equivalent transaction, so the traffic module keeps
its rebuild-on-drift model. Losing shaping briefly is not comparable to losing the host
firewall.

cargo test --workspace: 183 + 4 passed. clippy -D warnings clean.

traffic_core/src/firewall/antitorrent.rs

@@ -14,7 +14,7 @@ use crate::{config::Config, docker, netns::NsTransport};
 
 use super::{
     ANTITORRENT_TABLE, firewall_transport,
-    nft::{is_nft_not_found, normalize_nft_text, run_allow_notfound, run_step},
+    nft::{is_nft_not_found, normalize_nft_text, run_allow_notfound, run_nft_batch, table_exists},
 };
 
 #[derive(Debug, Clone)]
@@ -44,32 +44,16 @@ pub(super) fn apply_antitorrent_firewall(cfg: &Config) -> anyhow::Result<()> {
         return Ok(());
     }
 
-    run_allow_notfound(
-        transport.as_ref(),
-        &["nft", "delete", "table", "inet", ANTITORRENT_TABLE],
-    )?;
-
-    run_step(
-        transport.as_ref(),
-        &["nft", "add", "table", "inet", ANTITORRENT_TABLE],
-    )?;
-    run_step(
-        transport.as_ref(),
-        &[
-            "nft",
-            "add",
-            "counter",
-            "inet",
-            ANTITORRENT_TABLE,
-            "antitorrent_dropped_v4",
-        ],
-    )?;
-    run_step_chain(transport.as_ref())?;
-    apply_rule(
-        transport.as_ref(),
+    // One transaction, like the host firewall: delete and rebuild together, so a rejected
+    // ruleset leaves the existing guard in place instead of removing it first and then
+    // failing to put anything back. The four separate commands this replaces could also
+    // leave a half-built table - a table and counter but no chain, or a chain with no rule.
+    let present = table_exists(transport.as_ref(), ANTITORRENT_TABLE)?;
+    run_nft_batch(&build_antitorrent_batch_script(
         &cfg.anti_torrent.peer_ipv4_cidr,
         cfg.anti_torrent.max_connections_per_peer,
-    )?;
+        present,
+    ))?;
 
     let verify = inspect_firewall(transport.as_ref(), cfg)?;
     if firewall_has_drift(&verify) {
@@ -208,63 +192,30 @@ pub fn antitorrent_drop_counter(cfg: &Config) -> anyhow::Result<Option<(u64, u64
     }
 }
 
+/// Renders the whole owned table as one nft script.
+///
 /// `ct state new` is load-bearing: without it the threshold would start dropping packets
 /// of already established flows as soon as the counter is high, instead of only refusing
 /// new ones.
-fn apply_rule(transport: &dyn NsTransport, cidr: &str, max_conn: u32) -> anyhow::Result<()> {
-    let _ = transport.run(&[
-        "nft",
-        "add",
-        "rule",
-        "inet",
-        ANTITORRENT_TABLE,
-        "forward_antitorrent",
-        "ip",
-        "saddr",
-        cidr,
-        "ct",
-        "state",
-        "new",
-        "meter",
-        "awg2_peer_connlimit_v4",
-        "{",
-        "ip",
-        "saddr",
-        "ct",
-        "count",
-        "over",
-        &max_conn.to_string(),
-        "}",
-        "counter",
-        "name",
-        "antitorrent_dropped_v4",
-        "drop",
-    ])?;
-    Ok(())
+fn build_antitorrent_batch_script(cidr: &str, max_conn: u32, table_present: bool) -> String {
+    let mut batch = String::new();
+    if table_present {
+        batch.push_str(&format!("delete table inet {ANTITORRENT_TABLE}\n"));
     }
-
-fn run_step_chain(transport: &dyn NsTransport) -> anyhow::Result<()> {
-    run_step(
-        transport,
-        &[
-            "nft",
-            "add",
-            "chain",
-            "inet",
-            ANTITORRENT_TABLE,
-            "forward_antitorrent",
-            "{",
-            "type",
-            "filter",
-            "hook",
-            "forward",
-            "priority",
-            "filter;",
-            "policy",
-            "accept;",
-            "}",
-        ],
-    )
+    batch.push_str(&format!("add table inet {ANTITORRENT_TABLE}\n"));
+    batch.push_str(&format!(
+        "add counter inet {ANTITORRENT_TABLE} antitorrent_dropped_v4\n"
+    ));
+    batch.push_str(&format!(
+        "add chain inet {ANTITORRENT_TABLE} forward_antitorrent \
+         {{ type filter hook forward priority filter; policy accept; }}\n"
+    ));
+    batch.push_str(&format!(
+        "add rule inet {ANTITORRENT_TABLE} forward_antitorrent ip saddr {cidr} ct state new \
+         meter awg2_peer_connlimit_v4 {{ ip saddr ct count over {max_conn} }} \
+         counter name antitorrent_dropped_v4 drop\n"
+    ));
+    batch
 }
 
 fn firewall_has_drift(actual: &FirewallActual) -> bool {
@@ -371,6 +322,41 @@ mod tests {
         assert_eq!(got, Some((7, 448)));
     }
 
+    /// The script builds every object the inspector later looks for. It is checked
+    /// element by element rather than through `firewall_rule_contract_matches`, which
+    /// reads `nft list` output and not the script that produces it.
+    #[test]
+    fn rendered_batch_creates_every_object_the_inspector_requires() {
+        let script = build_antitorrent_batch_script("10.8.1.0/24", 150, false);
+        assert!(!script.contains("delete table"));
+        for needle in [
+            "add table inet awg2_harness",
+            "add counter inet awg2_harness antitorrent_dropped_v4",
+            "add chain inet awg2_harness forward_antitorrent",
+            "type filter hook forward priority filter; policy accept;",
+            "ip saddr 10.8.1.0/24",
+            "ct state new",
+            "meter awg2_peer_connlimit_v4",
+            "ct count over 150",
+            "counter name antitorrent_dropped_v4",
+            "drop",
+        ] {
+            assert!(script.contains(needle), "batch must contain {needle:?}");
+        }
+    }
+
+    #[test]
+    fn rendered_batch_deletes_and_recreates_in_one_transaction() {
+        let script = build_antitorrent_batch_script("10.8.1.0/24", 150, true);
+        let delete_at = script
+            .find("delete table inet awg2_harness")
+            .expect("delete present");
+        let add_at = script
+            .find("add table inet awg2_harness")
+            .expect("add present");
+        assert!(delete_at < add_at, "delete must precede the rebuild");
+    }
+
     #[test]
     fn nft_rule_contract_parser_accepts_expected_rule() {
         let table = r#"

traffic_core/src/firewall/host.rs

@@ -7,7 +7,7 @@ use crate::{config::Config, netns::build_host_transport};
 
 use super::{
     HOST_FW_TABLE,
-    nft::{is_nft_not_found, normalize_nft_text, run_allow_notfound, run_nft_batch},
+    nft::{is_nft_not_found, normalize_nft_text, run_allow_notfound, run_nft_batch, table_exists},
 };
 
 pub(super) fn apply_host_firewall(cfg: &Config) -> anyhow::Result<()> {
@@ -30,30 +30,43 @@ pub(super) fn apply_host_firewall(cfg: &Config) -> anyhow::Result<()> {
         .trim()
         .to_ascii_lowercase();
 
-    run_allow_notfound(
-        transport.as_ref(),
-        &["nft", "delete", "table", "inet", HOST_FW_TABLE],
+    apply_host_firewall_batch(
+        cfg,
+        &policy,
+        table_exists(transport.as_ref(), HOST_FW_TABLE)?,
     )?;
 
-    apply_host_firewall_batch(cfg, &policy)?;
-
     if let Some(reason) = host_firewall_drift_reason(cfg)? {
         anyhow::bail!("host firewall verify-after-apply failed: {reason}");
     }
     Ok(())
 }
 
-fn apply_host_firewall_batch(cfg: &Config, policy: &str) -> anyhow::Result<()> {
-    let batch = build_host_firewall_batch_script(cfg, policy);
+fn apply_host_firewall_batch(
+    cfg: &Config,
+    policy: &str,
+    table_present: bool,
+) -> anyhow::Result<()> {
+    let batch = build_host_firewall_batch_script(cfg, policy, table_present);
     run_nft_batch(&batch)
 }
 
-/// Renders the desired INPUT chain.
+/// Renders the replacement as a single nft transaction.
 ///
-/// The script never contains `delete table`: deletion happens as a separate step before
-/// the batch, so a batch that fails to parse cannot take the existing rules down with it.
-fn build_host_firewall_batch_script(cfg: &Config, policy: &str) -> String {
+/// The delete belongs *inside* the batch. `nft -f` applies a script atomically: if any
+/// line fails to parse or the kernel rejects the ruleset, nothing is committed. Deleting
+/// the table as a separate command first, as this used to, meant a failure in the
+/// following batch left the host with no INPUT policy and no allow rules at all - a
+/// fail-open window on the firewall, opened by a syntax error or a transient netlink
+/// failure.
+///
+/// The delete is only emitted when the table actually exists, because `delete table` on a
+/// missing table aborts the whole transaction.
+fn build_host_firewall_batch_script(cfg: &Config, policy: &str, table_present: bool) -> String {
     let mut batch = String::new();
+    if table_present {
+        batch.push_str(&format!("delete table inet {HOST_FW_TABLE}\n"));
+    }
     batch.push_str(&format!("add table inet {HOST_FW_TABLE}\n"));
     batch.push_str(&format!(
         "add chain inet {HOST_FW_TABLE} input {{ type filter hook input priority filter; policy {policy}; }}\n"
@@ -211,14 +224,33 @@ table inet awg2_host_firewall {
         );
     }
 
+    /// On a host with no managed table there is nothing to delete, and emitting the delete
+    /// anyway would abort the whole transaction.
     #[test]
-    fn host_firewall_batch_script_does_not_include_delete_table() {
+    fn batch_script_omits_delete_when_the_table_is_absent() {
         let cfg = test_config();
-        let script = build_host_firewall_batch_script(&cfg, "drop");
+        let script = build_host_firewall_batch_script(&cfg, "drop", false);
         assert!(script.contains("add table inet awg2_host_firewall"));
         assert!(!script.contains("delete table inet awg2_host_firewall"));
     }
 
+    /// Replacement is one transaction: the delete of the old table is part of the same
+    /// script, so a rejected ruleset leaves the previous one in place instead of having
+    /// already removed it and then failing.
+    #[test]
+    fn batch_script_deletes_and_recreates_in_one_transaction() {
+        let cfg = test_config();
+        let script = build_host_firewall_batch_script(&cfg, "drop", true);
+        let delete_at = script
+            .find("delete table inet awg2_host_firewall")
+            .expect("delete present");
+        let add_at = script
+            .find("add table inet awg2_host_firewall")
+            .expect("add present");
+        assert!(delete_at < add_at, "delete must precede the rebuild");
+        assert!(script.contains("tcp dport 2323 accept"));
+    }
+
     #[test]
     fn host_firewall_policy_drift_check_trims_policy_value() {
         let table = r#"

traffic_core/src/firewall/nft.rs

@@ -28,11 +28,6 @@ pub(super) fn is_nft_not_found(msg: &str) -> bool {
         || s.contains("no such table")
 }
 
-pub(super) fn run_step(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> {
-    let _ = transport.run(args)?;
-    Ok(())
-}
-
 /// Deleting a table that is already absent is the expected outcome when the subsystem is
 /// disabled or being rebuilt. Every other failure propagates.
 pub(super) fn run_allow_notfound(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> {
@@ -55,10 +50,24 @@ pub(super) fn run_allow_notfound(transport: &dyn NsTransport, args: &[&str]) ->
     }
 }
 
+/// Reports whether a table currently exists.
+///
+/// Needed because `delete table` on a missing table aborts the whole `nft -f` transaction,
+/// so the delete line can only be emitted when there is something to delete.
+pub(super) fn table_exists(transport: &dyn NsTransport, table: &str) -> anyhow::Result<bool> {
+    match transport.run(&["nft", "list", "table", "inet", table]) {
+        Ok(_) => Ok(true),
+        Err(e) if is_nft_not_found(&e.to_string()) => Ok(false),
+        Err(e) => Err(anyhow::anyhow!("failed to inspect table {table}: {e}")),
+    }
+}
+
 /// Applies a ruleset through `nft -f -`.
 ///
-/// A batch is atomic: either the whole script lands or nothing does. That matters for the
-/// host firewall, where a partially applied INPUT chain could lock out SSH.
+/// A batch is atomic: either the whole script lands or nothing does. That is why the
+/// delete of the old table belongs inside the script rather than in a preceding command -
+/// a rejected batch then leaves the previous ruleset exactly as it was, instead of having
+/// already removed it.
 pub(super) fn run_nft_batch(script: &str) -> anyhow::Result<()> {
     let mut child = Command::new("nft")
         .args(["-f", "-"])