cellos_supervisor/

nft_counters.rs

//! Pure-data parsing + classification of `nft list ruleset --json` output for
//! FC-38 Phase 1 per-flow `network_flow_decision` events.
//!
//! Honest scope: this module operates on the structured JSON nftables prints
//! when invoked with `--json`. It does **not** invoke `nft` itself, does no
//! I/O, and does not attempt real-time per-packet observation. Phase 2
//! (future) is eBPF / nflog real-time per-flow events; this Phase 1 layer
//! emits "which rule fired with what counter delta" attribution after the
//! workload exits, scraped via `nsenter ... nft list ruleset --json` from the
//! supervisor's hot path.
//!
//! The functions in this module are deliberately pure so they can be unit
//! tested on any platform — including macOS dev hosts where `nft(8)` is not
//! available — by feeding synthesized JSON shaped like the real binary
//! produces. The callers that DO invoke `nft` live behind
//! `#[cfg(target_os = "linux")]` in `supervisor.rs`.
//!
//! Two responsibilities:
//!
//! 1. [`parse_nft_list_ruleset_json`] turns the JSON document into a flat
//!    [`Vec<NftCounterRow>`]. One row per rule that nft printed, carrying
//!    its `family` / `table` / `chain` / `handle`, the `packets` / `bytes`
//!    counter from the rule's expression list when present, and a
//!    text representation of the rule's expression that the classifier can
//!    work on.
//!
//! 2. [`classify_rule_repr`] takes the synthesized text representation and
//!    decides which of the four FC-38 reason-codes Phase 1 attributes
//!    today, plus the destination address / port / protocol when the rule
//!    expresses them.
//!
//! See `docs/trust-plane-runtime.md` §Per-flow enforcement events (FC-38)
//! for the operator-side documentation.
//!
//! `parse_nft_list_ruleset_json` (and its private helpers) is only consumed
//! by the `#[cfg(target_os = "linux")]` `scan_nft_counters_in_ns` path in
//! `supervisor.rs`. On non-Linux dev builds the binary's hot path doesn't
//! reach the parser, so we suppress the `dead_code` lint at module level
//! rather than chase the symbol through cfg gates — the parser is still
//! exercised on every platform via `cargo test -p cellos-supervisor --lib
//! nft_counters`.

#![allow(dead_code)]

use serde::{Deserialize, Serialize};

/// One counter row distilled from `nft list ruleset --json` output.
///
/// `rule_repr` is a synthesized text shape (e.g. `"ip daddr 10.0.0.1 tcp dport 443 accept"`)
/// rebuilt from the JSON expression tree so the classifier can run pure
/// string matching against the same shape `generate_nft_ruleset` emits in
/// `supervisor.rs`. We do NOT try to round-trip the original text — the
/// repr is a normalized debug form purely for classification + audit
/// surfaces.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct NftCounterRow {
    /// nft family, e.g. `"inet"`, `"ip"`, `"ip6"`.
    pub family: String,
    /// Table name, e.g. `"cellos_<safe_id>"`.
    pub table: String,
    /// Chain name, e.g. `"output"`.
    pub chain: String,
    /// nft handle for the rule (stable within a single ruleset version).
    pub handle: u64,
    /// Packet counter from the rule's `counter` expression, or `0` when
    /// no counter was present. Phase 1 emits per-rule events whether the
    /// counter is present or zero — operators want to know "this deny rule
    /// applied, no traffic hit it" as much as "this deny rule fired N times".
    pub packet_count: u64,
    /// Byte counter from the rule's `counter` expression, or `0`.
    pub byte_count: u64,
    /// Synthesized text shape for the classifier (see struct doc).
    pub rule_repr: String,
}

/// Errors returned by [`parse_nft_list_ruleset_json`].
#[derive(Debug, thiserror::Error)]
pub enum NftCountersError {
    /// The input wasn't valid JSON.
    #[error("nft counter JSON parse failure: {0}")]
    Json(#[from] serde_json::Error),
    /// The JSON parsed but didn't contain the expected `nftables` array shape.
    #[error("nft counter JSON missing top-level `nftables` array")]
    MissingNftablesArray,
}

/// Parse the JSON document produced by `nft list ruleset --json` into a flat
/// list of counter rows.
///
/// nft's JSON shape is `{"nftables": [ {<obj>}, {<obj>}, ... ]}` where each
/// object is one of: `{"metainfo": ...}`, `{"table": ...}`, `{"chain": ...}`,
/// or `{"rule": {...}}`. Phase 1 cares only about the `rule` entries.
///
/// Each `rule` entry carries:
/// - `family`, `table`, `chain`, `handle`
/// - `expr`: a list of expression objects describing the rule's match +
///   verdict, e.g. `{"match": {"left": ..., "right": ..., "op": "=="}},
///   {"counter": {"packets": N, "bytes": N}}, {"accept": null}`.
///
/// We build the synthesized `rule_repr` by walking `expr` and emitting a
/// minimal text form sufficient for [`classify_rule_repr`]. Counter
/// expressions are surfaced into [`NftCounterRow::packet_count`] /
/// [`NftCounterRow::byte_count`].
///
/// Unknown / unsupported expression kinds are skipped silently so we degrade
/// gracefully across nftables minor versions; a rule whose entire `expr` is
/// unrecognized still surfaces as a row with an empty `rule_repr` (the
/// classifier returns a default-deny-style attribution for that case).
pub fn parse_nft_list_ruleset_json(raw: &str) -> Result<Vec<NftCounterRow>, NftCountersError> {
    let value: serde_json::Value = serde_json::from_str(raw)?;
    let arr = value
        .get("nftables")
        .and_then(|v| v.as_array())
        .ok_or(NftCountersError::MissingNftablesArray)?;

    let mut rows = Vec::new();
    for item in arr {
        let Some(rule) = item.get("rule").and_then(|v| v.as_object()) else {
            continue;
        };
        let family = rule
            .get("family")
            .and_then(|v| v.as_str())
            .unwrap_or_default()
            .to_string();
        let table = rule
            .get("table")
            .and_then(|v| v.as_str())
            .unwrap_or_default()
            .to_string();
        let chain = rule
            .get("chain")
            .and_then(|v| v.as_str())
            .unwrap_or_default()
            .to_string();
        let handle = rule.get("handle").and_then(|v| v.as_u64()).unwrap_or(0);
        let expr = rule.get("expr").and_then(|v| v.as_array());

        let mut packet_count: u64 = 0;
        let mut byte_count: u64 = 0;
        let mut repr_parts: Vec<String> = Vec::new();
        if let Some(exprs) = expr {
            for e in exprs {
                emit_expr(e, &mut repr_parts, &mut packet_count, &mut byte_count);
            }
        }
        let rule_repr = repr_parts.join(" ");
        rows.push(NftCounterRow {
            family,
            table,
            chain,
            handle,
            packet_count,
            byte_count,
            rule_repr,
        });
    }
    Ok(rows)
}

/// Walk a single expression object and append its rendered form (if any)
/// to `repr_parts`. Counter expressions are extracted into the running
/// counters rather than rendered to text.
fn emit_expr(
    expr: &serde_json::Value,
    repr_parts: &mut Vec<String>,
    packet_count: &mut u64,
    byte_count: &mut u64,
) {
    let Some(obj) = expr.as_object() else {
        return;
    };
    for (kind, body) in obj {
        match kind.as_str() {
            "counter" => {
                if let Some(c) = body.as_object() {
                    if let Some(p) = c.get("packets").and_then(|v| v.as_u64()) {
                        // Multiple counter expressions on a single rule are
                        // unusual but tolerated — sum across them so a rule
                        // with split packet/byte counters surfaces the
                        // largest observed value rather than the last one.
                        *packet_count = (*packet_count).max(p);
                    }
                    if let Some(b) = c.get("bytes").and_then(|v| v.as_u64()) {
                        *byte_count = (*byte_count).max(b);
                    }
                }
            }
            "match" => {
                if let Some(m) = body.as_object() {
                    let left = m.get("left");
                    let right = m.get("right");
                    let op = m.get("op").and_then(|v| v.as_str()).unwrap_or("==");
                    if op == "==" {
                        if let Some(rendered) = render_match(left, right) {
                            repr_parts.push(rendered);
                        }
                    }
                }
            }
            "accept" => repr_parts.push("accept".to_string()),
            "drop" => repr_parts.push("drop".to_string()),
            "policy" => {
                if let Some(s) = body.as_str() {
                    repr_parts.push(format!("policy {s}"));
                }
            }
            // ignore other expression kinds (log, jump, return, ...)
            _ => {}
        }
    }
}

/// Render an nft `match` expression to the text shape `classify_rule_repr`
/// recognizes. Supports the limited family used by `generate_nft_ruleset`:
/// `ip daddr <ip>`, `ip6 daddr <ip>`, and `<proto> dport <n>`.
fn render_match(
    left: Option<&serde_json::Value>,
    right: Option<&serde_json::Value>,
) -> Option<String> {
    let left = left?.as_object()?;
    let payload = left.get("payload")?.as_object()?;
    let proto = payload.get("protocol").and_then(|v| v.as_str())?;
    let field = payload.get("field").and_then(|v| v.as_str())?;
    let right_value = right?;
    let rendered_right = match right_value {
        serde_json::Value::String(s) => s.clone(),
        serde_json::Value::Number(n) => n.to_string(),
        _ => return None,
    };
    match (proto, field) {
        ("ip", "daddr") => Some(format!("ip daddr {rendered_right}")),
        ("ip6", "daddr") => Some(format!("ip6 daddr {rendered_right}")),
        ("udp", "dport") => Some(format!("udp dport {rendered_right}")),
        ("tcp", "dport") => Some(format!("tcp dport {rendered_right}")),
        _ => None,
    }
}

/// Result of classifying a single rule's text repr.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClassifiedRule {
    /// `allow` for `accept`, `deny` for `drop` (or default-drop policy).
    pub decision: NftDecision,
    /// FC-38 Phase 1 reason code; one of the four documented classes or
    /// the catch-all `nft_default_drop` for unrecognized drop rules.
    pub reason_code: &'static str,
    /// Optional reference back to the rule's declaration site in the spec.
    pub nft_rule_ref: Option<String>,
    /// Optional destination IP (when the rule has `ip/ip6 daddr <x>`).
    pub dst_addr: Option<String>,
    /// Optional destination port (when the rule has `<proto> dport <n>`).
    pub dst_port: Option<u16>,
    /// Optional transport protocol (`udp` / `tcp`).
    pub protocol: Option<String>,
}

/// Allow / deny outcome a classifier returns. Mirrors
/// [`cellos_core::NetworkFlowDecisionOutcome`] but lives in this module to
/// keep the parser independent of the public types crate's serde contract.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NftDecision {
    Allow,
    Deny,
}

/// Classify a single nft rule's text shape into a Phase 1 reason code.
///
/// The recognized shapes match the output of `generate_nft_ruleset` in
/// `supervisor.rs`:
///
/// | shape                                                   | decision | reason_code                  |
/// |---------------------------------------------------------|----------|------------------------------|
/// | `policy drop`                                           | deny     | `nft_default_drop`           |
/// | `udp dport 53 drop` or `tcp dport 53 drop`              | deny     | `nft_workload_dns_block`     |
/// | `udp dport 853 drop` (SEC-22 Phase 3d DoQ block)        | deny     | `nft_doq_blocked`            |
/// | `udp dport 443 drop` (SEC-22 Phase 3d HTTP/3 block)     | deny     | `nft_http3_blocked`          |
/// | `ip daddr <ip> udp dport 53 accept` (resolver allow)    | allow    | `nft_resolver_allowlist_match`|
/// | `ip daddr <ip> tcp dport 53 accept` (resolver allow)    | allow    | `nft_resolver_allowlist_match`|
/// | `ip daddr <ip> udp dport 853 accept` (Phase 3d allow)   | allow    | `nft_resolver_allowlist_match`|
/// | `ip daddr <ip> udp dport 443 accept` (Phase 3d allow)   | allow    | `nft_resolver_allowlist_match`|
/// | `ip[6] daddr <ip> <proto> dport <port> accept` (egress) | allow    | `nft_egress_rule_match`      |
/// | (anything else with `accept`)                           | allow    | `nft_egress_rule_match`      |
/// | (anything else with `drop`)                             | deny     | `nft_default_drop`           |
/// | (empty / unrecognized)                                  | deny     | `nft_default_drop`           |
///
/// `nft_resolver_allowlist_match` distinguishes a port-53 / port-853 / port-443
/// accept (a declared `dnsAuthority.resolvers[]` carve-out under the SEC-22
/// backstop) from an ordinary egress accept (`nft_egress_rule_match`). Both
/// are `decision: allow`; the reason code lets operators tell at a glance
/// whether traffic is hitting the resolver allowlist or a generic
/// `authority.egressRules[]` entry. Workload-DNS denials still surface as
/// `nft_workload_dns_block` from the udp/tcp 53 drop rules.
///
/// SEC-22 Phase 3d adds two new deny reason codes — `nft_doq_blocked` for
/// the `udp dport 853 drop` rule and `nft_http3_blocked` for the
/// `udp dport 443 drop` rule. These fire only when the operator has set the
/// corresponding `blockUdpDoq` / `blockUdpHttp3` opt-in. The
/// resolver-allowlist match is also extended to recognise the parallel
/// accept-leg shape on UDP/853 and UDP/443 (same `ip daddr <x> udp dport <p>
/// accept` shape, just with a different dport).
///
/// `nft_rule_ref` is populated when the rule's shape unambiguously points at
/// a spec source. For the policy-default-drop and the workload-dns-block
/// rules, we emit the synthetic refs `default-drop` and `dnsAuthority.workloadDnsBlock`
/// respectively — they don't index into a list. For Phase 3d, the synthetic
/// refs are `dnsAuthority.blockUdpDoq` and `dnsAuthority.blockUdpHttp3`. For
/// ordinary egress rules we cannot reverse the index without the full
/// ruleset context, so the caller (supervisor) is responsible for setting
/// the spec-relative ref; this function leaves it as `None` for those rows.
pub fn classify_rule_repr(rule_repr: &str) -> ClassifiedRule {
    let trimmed = rule_repr.trim();

    // Default-drop policy line ("policy drop"): lone keyword.
    if trimmed == "policy drop" {
        return ClassifiedRule {
            decision: NftDecision::Deny,
            reason_code: "nft_default_drop",
            nft_rule_ref: Some("default-drop".to_string()),
            dst_addr: None,
            dst_port: None,
            protocol: None,
        };
    }

    let has_accept = trimmed.split_whitespace().any(|t| t == "accept");
    let has_drop = trimmed.split_whitespace().any(|t| t == "drop");

    // SEC-22 workload-dns-block: bare `udp/tcp dport 53 drop` with no daddr.
    let dst_addr = extract_daddr(trimmed);
    let (proto, dport) = extract_dport(trimmed);

    if has_drop && dst_addr.is_none() && dport == Some(53) {
        let proto_string = proto.map(|p| p.to_string());
        return ClassifiedRule {
            decision: NftDecision::Deny,
            reason_code: "nft_workload_dns_block",
            nft_rule_ref: Some("dnsAuthority.workloadDnsBlock".to_string()),
            dst_addr: None,
            dst_port: Some(53),
            protocol: proto_string,
        };
    }

    // SEC-22 Phase 3d — bare `udp dport 853 drop` with no daddr (DoQ block).
    if has_drop && dst_addr.is_none() && dport == Some(853) && proto == Some("udp") {
        return ClassifiedRule {
            decision: NftDecision::Deny,
            reason_code: "nft_doq_blocked",
            nft_rule_ref: Some("dnsAuthority.blockUdpDoq".to_string()),
            dst_addr: None,
            dst_port: Some(853),
            protocol: Some("udp".to_string()),
        };
    }

    // SEC-22 Phase 3d — bare `udp dport 443 drop` with no daddr (HTTP/3 block).
    if has_drop && dst_addr.is_none() && dport == Some(443) && proto == Some("udp") {
        return ClassifiedRule {
            decision: NftDecision::Deny,
            reason_code: "nft_http3_blocked",
            nft_rule_ref: Some("dnsAuthority.blockUdpHttp3".to_string()),
            dst_addr: None,
            dst_port: Some(443),
            protocol: Some("udp".to_string()),
        };
    }

    if has_accept {
        // Resolver allowlist rule: `ip[6] daddr <ip> <proto> dport <p> accept`
        // where p is 53 (legacy SEC-22), or — under SEC-22 Phase 3d — also
        // 853 (DoQ) / 443 (HTTP/3) for UDP transport. The reason code
        // reflects "this allow exists because a SEC-22 resolver carve-out
        // is active" — see fn doc.
        let is_phase3d_udp_resolver_port =
            proto == Some("udp") && (dport == Some(853) || dport == Some(443));
        if dport == Some(53) || is_phase3d_udp_resolver_port {
            let proto_string = proto.map(|p| p.to_string());
            return ClassifiedRule {
                decision: NftDecision::Allow,
                reason_code: "nft_resolver_allowlist_match",
                nft_rule_ref: None,
                dst_addr,
                dst_port: dport,
                protocol: proto_string,
            };
        }
        let proto_string = proto.map(|p| p.to_string());
        return ClassifiedRule {
            decision: NftDecision::Allow,
            reason_code: "nft_egress_rule_match",
            nft_rule_ref: None,
            dst_addr,
            dst_port: dport,
            protocol: proto_string,
        };
    }

    if has_drop {
        let proto_string = proto.map(|p| p.to_string());
        return ClassifiedRule {
            decision: NftDecision::Deny,
            reason_code: "nft_default_drop",
            nft_rule_ref: None,
            dst_addr,
            dst_port: dport,
            protocol: proto_string,
        };
    }

    // Empty or unrecognized — treat as deny default. Honest fallback so a
    // rule we don't understand surfaces as an event the operator can see
    // rather than getting silently dropped.
    ClassifiedRule {
        decision: NftDecision::Deny,
        reason_code: "nft_default_drop",
        nft_rule_ref: None,
        dst_addr: None,
        dst_port: None,
        protocol: None,
    }
}

/// Extract the destination IP from a rule repr like
/// `"ip daddr 10.0.0.1 tcp dport 443 accept"`. Supports both `ip` and `ip6`
/// daddrs.
fn extract_daddr(repr: &str) -> Option<String> {
    let tokens: Vec<&str> = repr.split_whitespace().collect();
    for win in tokens.windows(3) {
        if (win[0] == "ip" || win[0] == "ip6") && win[1] == "daddr" {
            return Some(win[2].to_string());
        }
    }
    None
}

/// Extract `(protocol, port)` from a rule repr containing
/// `"<proto> dport <n>"`. Returns `(None, None)` when absent.
fn extract_dport(repr: &str) -> (Option<&'static str>, Option<u16>) {
    let tokens: Vec<&str> = repr.split_whitespace().collect();
    for win in tokens.windows(3) {
        let proto = match win[0] {
            "udp" => Some("udp"),
            "tcp" => Some("tcp"),
            _ => None,
        };
        if proto.is_some() && win[1] == "dport" {
            if let Ok(p) = win[2].parse::<u16>() {
                return (proto, Some(p));
            }
        }
    }
    (None, None)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn classify_recognizes_default_drop_policy() {
        let c = classify_rule_repr("policy drop");
        assert_eq!(c.decision, NftDecision::Deny);
        assert_eq!(c.reason_code, "nft_default_drop");
        assert_eq!(c.nft_rule_ref.as_deref(), Some("default-drop"));
        assert!(c.dst_addr.is_none());
        assert!(c.dst_port.is_none());
        assert!(c.protocol.is_none());
    }

    #[test]
    fn classify_recognizes_workload_dns_block_udp() {
        let c = classify_rule_repr("udp dport 53 drop");
        assert_eq!(c.decision, NftDecision::Deny);
        assert_eq!(c.reason_code, "nft_workload_dns_block");
        assert_eq!(
            c.nft_rule_ref.as_deref(),
            Some("dnsAuthority.workloadDnsBlock")
        );
        assert_eq!(c.dst_port, Some(53));
        assert_eq!(c.protocol.as_deref(), Some("udp"));
        assert!(c.dst_addr.is_none());
    }

    #[test]
    fn classify_recognizes_workload_dns_block_tcp() {
        let c = classify_rule_repr("tcp dport 53 drop");
        assert_eq!(c.decision, NftDecision::Deny);
        assert_eq!(c.reason_code, "nft_workload_dns_block");
        assert_eq!(c.dst_port, Some(53));
        assert_eq!(c.protocol.as_deref(), Some("tcp"));
    }

    #[test]
    fn classify_recognizes_resolver_allowlist_udp() {
        let c = classify_rule_repr("ip daddr 1.1.1.1 udp dport 53 accept");
        assert_eq!(c.decision, NftDecision::Allow);
        assert_eq!(c.reason_code, "nft_resolver_allowlist_match");
        assert_eq!(c.dst_addr.as_deref(), Some("1.1.1.1"));
        assert_eq!(c.dst_port, Some(53));
        assert_eq!(c.protocol.as_deref(), Some("udp"));
    }

    #[test]
    fn classify_recognizes_resolver_allowlist_tcp() {
        let c = classify_rule_repr("ip daddr 1.1.1.1 tcp dport 53 accept");
        assert_eq!(c.decision, NftDecision::Allow);
        assert_eq!(c.reason_code, "nft_resolver_allowlist_match");
        assert_eq!(c.dst_port, Some(53));
        assert_eq!(c.protocol.as_deref(), Some("tcp"));
    }

    #[test]
    fn classify_recognizes_egress_rule_accept_v4() {
        let c = classify_rule_repr("ip daddr 10.0.0.1 tcp dport 443 accept");
        assert_eq!(c.decision, NftDecision::Allow);
        assert_eq!(c.reason_code, "nft_egress_rule_match");
        assert_eq!(c.dst_addr.as_deref(), Some("10.0.0.1"));
        assert_eq!(c.dst_port, Some(443));
        assert_eq!(c.protocol.as_deref(), Some("tcp"));
    }

    #[test]
    fn classify_recognizes_egress_rule_accept_v6() {
        let c = classify_rule_repr("ip6 daddr 2001:db8::1 tcp dport 443 accept");
        assert_eq!(c.decision, NftDecision::Allow);
        assert_eq!(c.reason_code, "nft_egress_rule_match");
        assert_eq!(c.dst_addr.as_deref(), Some("2001:db8::1"));
        assert_eq!(c.dst_port, Some(443));
        assert_eq!(c.protocol.as_deref(), Some("tcp"));
    }

    // ── SEC-22 Phase 3d classifier tests ────────────────────────────────
    //
    // Pin the new reason codes (`nft_doq_blocked`, `nft_http3_blocked`) and
    // the resolver-allowlist match extension to UDP/853 + UDP/443. These
    // mirror the existing port-53 classifier tests above.

    #[test]
    fn classify_recognises_nft_doq_blocked() {
        let c = classify_rule_repr("udp dport 853 drop");
        assert_eq!(c.decision, NftDecision::Deny);
        assert_eq!(c.reason_code, "nft_doq_blocked");
        assert_eq!(c.nft_rule_ref.as_deref(), Some("dnsAuthority.blockUdpDoq"));
        assert_eq!(c.dst_port, Some(853));
        assert_eq!(c.protocol.as_deref(), Some("udp"));
        assert!(c.dst_addr.is_none());
    }

    #[test]
    fn classify_recognises_nft_http3_blocked() {
        let c = classify_rule_repr("udp dport 443 drop");
        assert_eq!(c.decision, NftDecision::Deny);
        assert_eq!(c.reason_code, "nft_http3_blocked");
        assert_eq!(
            c.nft_rule_ref.as_deref(),
            Some("dnsAuthority.blockUdpHttp3")
        );
        assert_eq!(c.dst_port, Some(443));
        assert_eq!(c.protocol.as_deref(), Some("udp"));
        assert!(c.dst_addr.is_none());
    }

    #[test]
    fn classify_recognises_resolver_allowlist_match_for_udp_443() {
        let c = classify_rule_repr("ip daddr 1.1.1.1 udp dport 443 accept");
        assert_eq!(c.decision, NftDecision::Allow);
        assert_eq!(c.reason_code, "nft_resolver_allowlist_match");
        assert_eq!(c.dst_addr.as_deref(), Some("1.1.1.1"));
        assert_eq!(c.dst_port, Some(443));
        assert_eq!(c.protocol.as_deref(), Some("udp"));
    }

    #[test]
    fn classify_recognises_resolver_allowlist_match_for_udp_853() {
        let c = classify_rule_repr("ip daddr 1.1.1.1 udp dport 853 accept");
        assert_eq!(c.decision, NftDecision::Allow);
        assert_eq!(c.reason_code, "nft_resolver_allowlist_match");
        assert_eq!(c.dst_addr.as_deref(), Some("1.1.1.1"));
        assert_eq!(c.dst_port, Some(853));
        assert_eq!(c.protocol.as_deref(), Some("udp"));
    }

    #[test]
    fn classify_does_not_misroute_tcp_443_to_http3_blocked() {
        // TCP/443 drop is NOT an HTTP/3 block — falls through to the
        // generic default-drop classifier so we don't accidentally label
        // ordinary TCP 443 traffic as Phase-3d HTTP/3.
        let c = classify_rule_repr("tcp dport 443 drop");
        assert_eq!(c.decision, NftDecision::Deny);
        assert_eq!(c.reason_code, "nft_default_drop");
    }

    #[test]
    fn classify_egress_rule_match_still_recognised_for_non_resolver_udp_dport() {
        // A spec-declared egress on UDP/4242 (not a resolver port) must
        // still classify as `nft_egress_rule_match`, NOT as a resolver
        // allowlist match.
        let c = classify_rule_repr("ip daddr 10.0.0.7 udp dport 4242 accept");
        assert_eq!(c.decision, NftDecision::Allow);
        assert_eq!(c.reason_code, "nft_egress_rule_match");
        assert_eq!(c.dst_port, Some(4242));
    }

    #[test]
    fn classify_handles_malformed_input_gracefully() {
        let c = classify_rule_repr("");
        assert_eq!(c.decision, NftDecision::Deny);
        assert_eq!(c.reason_code, "nft_default_drop");
        assert!(c.dst_addr.is_none());
        assert!(c.dst_port.is_none());
    }

    #[test]
    fn classify_unrecognized_drop_falls_back_to_default_drop() {
        let c = classify_rule_repr("ct state invalid drop");
        assert_eq!(c.decision, NftDecision::Deny);
        assert_eq!(c.reason_code, "nft_default_drop");
    }

    #[test]
    fn extract_dport_ignores_wrong_proto() {
        let (p, port) = extract_dport("icmp echo-request accept");
        assert!(p.is_none());
        assert!(port.is_none());
    }

    #[test]
    fn parse_extracts_one_egress_accept_row() {
        // Synthesized snippet shaped like real `nft list ruleset --json`
        // output for a single egress accept rule with a counter.
        let json = r#"{
          "nftables": [
            {"metainfo": {"version": "1.0.6"}},
            {"table": {"family": "inet", "name": "cellos_test"}},
            {"chain": {"family": "inet", "table": "cellos_test", "name": "output"}},
            {"rule": {
              "family": "inet",
              "table": "cellos_test",
              "chain": "output",
              "handle": 7,
              "expr": [
                {"match": {
                  "left": {"payload": {"protocol": "ip", "field": "daddr"}},
                  "right": "10.0.0.1",
                  "op": "=="
                }},
                {"match": {
                  "left": {"payload": {"protocol": "tcp", "field": "dport"}},
                  "right": 443,
                  "op": "=="
                }},
                {"counter": {"packets": 5, "bytes": 320}},
                {"accept": null}
              ]
            }}
          ]
        }"#;
        let rows = parse_nft_list_ruleset_json(json).expect("parse ok");
        assert_eq!(rows.len(), 1);
        let row = &rows[0];
        assert_eq!(row.family, "inet");
        assert_eq!(row.table, "cellos_test");
        assert_eq!(row.chain, "output");
        assert_eq!(row.handle, 7);
        assert_eq!(row.packet_count, 5);
        assert_eq!(row.byte_count, 320);
        assert_eq!(row.rule_repr, "ip daddr 10.0.0.1 tcp dport 443 accept");

        let classified = classify_rule_repr(&row.rule_repr);
        assert_eq!(classified.decision, NftDecision::Allow);
        assert_eq!(classified.reason_code, "nft_egress_rule_match");
        assert_eq!(classified.dst_addr.as_deref(), Some("10.0.0.1"));
        assert_eq!(classified.dst_port, Some(443));
    }

    #[test]
    fn parse_extracts_workload_dns_block_drop_row() {
        let json = r#"{
          "nftables": [
            {"rule": {
              "family": "inet",
              "table": "cellos_test",
              "chain": "output",
              "handle": 4,
              "expr": [
                {"match": {
                  "left": {"payload": {"protocol": "udp", "field": "dport"}},
                  "right": 53,
                  "op": "=="
                }},
                {"counter": {"packets": 2, "bytes": 80}},
                {"drop": null}
              ]
            }}
          ]
        }"#;
        let rows = parse_nft_list_ruleset_json(json).expect("parse ok");
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].packet_count, 2);
        assert_eq!(rows[0].rule_repr, "udp dport 53 drop");
        let c = classify_rule_repr(&rows[0].rule_repr);
        assert_eq!(c.reason_code, "nft_workload_dns_block");
        assert_eq!(c.decision, NftDecision::Deny);
    }

    #[test]
    fn parse_returns_empty_for_empty_nftables() {
        let json = r#"{"nftables": []}"#;
        let rows = parse_nft_list_ruleset_json(json).expect("parse ok");
        assert!(rows.is_empty());
    }

    #[test]
    fn parse_errors_when_nftables_array_missing() {
        let json = r#"{"not_nftables": []}"#;
        let err = parse_nft_list_ruleset_json(json).expect_err("must error");
        assert!(matches!(err, NftCountersError::MissingNftablesArray));
    }

    #[test]
    fn parse_errors_on_malformed_json() {
        let err = parse_nft_list_ruleset_json("not json").expect_err("must error");
        assert!(matches!(err, NftCountersError::Json(_)));
    }
}