cellos_supervisor/

per_flow.rs

//! Real-time per-flow `network_flow_decision` events via nflog.
//!
//! Honest scope: this module is the *real-time* companion to
//! [`crate::nft_counters`] (FC-38 Phase 1 post-run counter scrape). When the
//! operator opts in via `CELLOS_FIRECRACKER_PER_FLOW_EBPF=1`, the supervisor:
//!
//! 1. Rewrites the cell's nft ruleset so every `accept` / `drop` verdict is
//!    *also* logged via the netfilter `log group N prefix "cellos-flow ..."`
//!    action ([`augment_ruleset_with_log_actions`]). The original verdict is
//!    preserved so the rewrite is policy-neutral.
//! 2. Spawns a dedicated OS thread inside the cell's network namespace
//!    ([`spawn_per_flow_listener_in_netns`]) that opens a `NETLINK_NETFILTER`
//!    socket, binds the configured nfnetlink_log group, and translates each
//!    `NFULNL_PACKET_HDR` + payload into a [`cellos_core::NetworkFlowDecision`]
//!    CloudEvent emitted live to the supervisor's [`cellos_core::ports::EventSink`].
//!
//! **Backend chosen**: `nflog` (`NFNL_SUBSYS_ULOG`). The specification calls
//! for an eBPF-based tracer; eBPF is **deferred** because the workspace does
//! not yet carry an eBPF crate (e.g. `aya` / `redbpf`) and Linux distros do
//! not uniformly grant `CAP_BPF` to non-root supervisors. `nflog` requires
//! only `CAP_NET_ADMIN` (already wired for nft application), works on every
//! supported Linux kernel >= 3.13, and provides identical attribution
//! granularity for accept/drop verdicts. The env var
//! `CELLOS_FIRECRACKER_PER_FLOW_BACKEND=ebpf` is recognised but currently
//! warns + falls back to nflog. When an eBPF backend lands, the
//! [`PerFlowBackend`] enum here grows a real second arm and the listener
//! spawn dispatches on it.
//!
//! ## Why a dedicated OS thread (mirrors `dns_proxy::spawn`)
//!
//! `setns(2)` mutates the *calling thread*'s netns association. Tokio's
//! `spawn_blocking` workers are pooled — polluting one would leak the cell's
//! netns into unrelated tasks. We therefore allocate a fresh `std::thread`,
//! `setns` there, run the recv loop to completion, and let the OS reclaim
//! the thread on exit.
//!
//! ## Pure parsing surface
//!
//! Everything except `spawn_per_flow_listener_in_netns` is pure: ruleset
//! augmentation, nflog datagram decoding, L3/L4 attribution, decision
//! building. The pure helpers are exercised by 18 unit tests on every
//! platform; the Linux-only spawn helper is integration-tested under
//! `tests/supervisor_per_flow_realtime.rs` (`#[ignore]`d, requires
//! `CAP_NET_ADMIN`).

#![allow(dead_code)]

use std::sync::atomic::AtomicBool;
use std::sync::Arc;

use cellos_core::{NetworkFlowDecision, NetworkFlowDecisionOutcome, NetworkFlowDirection};

// ────────────────────────────────────────────────────────────────────────
// Env-var contract
// ────────────────────────────────────────────────────────────────────────

/// D1 opt-in: when set to `1`, the supervisor augments the nft ruleset with
/// `log group N prefix "cellos-flow ..."` actions and spawns a per-flow
/// listener thread in the cell's netns.
pub const ENV_PER_FLOW_EBPF: &str = "CELLOS_FIRECRACKER_PER_FLOW_EBPF";

/// E7 / FC-38 Phase 2 alias for [`ENV_PER_FLOW_EBPF`]. The original env
/// var name retains the `_EBPF` suffix from when eBPF was the planned
/// backend; the implemented backend is nflog (eBPF deferred — see module
/// doc). Operators rolling out FC-38 Phase 2 on top of `nft` ergonomically
/// expect a backend-neutral flag name. Both env vars are honoured by
/// [`build_activation_from_env`]; setting either one to `"1"` opts in.
/// The legacy name remains stable for existing operators.
pub const ENV_PER_FLOW_REALTIME: &str = "CELLOS_PER_FLOW_REALTIME";

/// Optional override for the nfnetlink_log group bound by the listener.
/// Defaults to [`DEFAULT_NFLOG_GROUP`] when unset or unparseable.
pub const ENV_PER_FLOW_NFLOG_GROUP: &str = "CELLOS_FIRECRACKER_PER_FLOW_NFLOG_GROUP";

/// Optional backend selector (legacy name). Recognised values: `nflog`
/// (default), `ebpf` (E7-4 host-side aya loader — attempts to load a BPF
/// program inside the cell's netns; falls back to nflog if eBPF startup
/// fails for any reason).
pub const ENV_PER_FLOW_BACKEND: &str = "CELLOS_FIRECRACKER_PER_FLOW_BACKEND";

/// E7-5 alias for [`ENV_PER_FLOW_BACKEND`] — the spec-canonical name
/// without the `_FIRECRACKER_` infix. Both are honoured; the legacy name
/// retains precedence so operators that already shipped the long name
/// keep the same value.
pub const ENV_PER_FLOW_BACKEND_E7: &str = "CELLOS_PER_FLOW_BACKEND";

/// Default nflog group when the operator does not override it. Chosen
/// outside the typical 0..32 range many distros use for their stock chains
/// to minimise collisions in shared host netns observability tooling.
pub const DEFAULT_NFLOG_GROUP: u16 = 100;

/// nft `log` action prefix used to tag augmented rules. Distinguishes
/// CellOS-emitted log packets from any other unrelated logging the operator
/// may have configured on the host. The decoder filters on this prefix
/// substring before building the decision event.
pub const LOG_PREFIX_BASE: &str = "cellos-flow";

/// Backend selection result of [`select_backend_from_env`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PerFlowBackend {
    /// nfnetlink_log via `NETLINK_NETFILTER` socket. Default.
    Nflog,
    /// eBPF tracer (deferred — selecting this currently warns + falls back
    /// to nflog at runtime).
    Ebpf,
}

/// Activation context for the per-flow listener.
///
/// Constructed once per `Supervisor::run` when the env gate is on; consumed
/// by [`spawn_per_flow_listener_in_netns`] when the workload child PID is
/// known. Mirrors the shape of [`crate::dns_proxy::spawn::EventSinkEmitter`]
/// + activation pair.
pub struct PerFlowActivation {
    /// Portable cell identifier (stamped into emitted events).
    pub cell_id: String,
    /// Run UUID/identifier (stamped into emitted events).
    pub run_id: String,
    /// nfnetlink_log group bound by the listener (matches the `log group N`
    /// action in the augmented ruleset).
    pub nflog_group: u16,
    /// Backend in effect (resolves at activation time so the warning fires
    /// once per run rather than per packet).
    pub backend: PerFlowBackend,
    /// Optional sha256 digest of the policy bundle, stamped into events.
    pub policy_digest: Option<String>,
    /// Optional trust-keyset id stamped into events.
    pub keyset_id: Option<String>,
    /// Optional issuer kid stamped into events.
    pub issuer_kid: Option<String>,
}

/// Read the env vars and return `Some(PerFlowActivation)` when the gate is on.
///
/// Returns `None` when [`ENV_PER_FLOW_EBPF`] is unset / not `1`. Always
/// returns `Some` when the gate is on; backend selection / group parsing
/// degrades to defaults rather than failing.
pub fn build_activation_from_env(
    cell_id: &str,
    run_id: &str,
    policy_digest: Option<String>,
    keyset_id: Option<String>,
    issuer_kid: Option<String>,
) -> Option<PerFlowActivation> {
    // Either the legacy `_EBPF` name or the backend-neutral `_REALTIME`
    // alias opts in. We deliberately accept both rather than aliasing one
    // to the other in shell rc files: operators can pick whichever name
    // matches their telemetry convention without coordinating with the
    // supervisor binary's env-var contract.
    let ebpf_on = std::env::var(ENV_PER_FLOW_EBPF).as_deref() == Ok("1");
    let realtime_on = std::env::var(ENV_PER_FLOW_REALTIME).as_deref() == Ok("1");
    if !ebpf_on && !realtime_on {
        return None;
    }
    let nflog_group = std::env::var(ENV_PER_FLOW_NFLOG_GROUP)
        .ok()
        .and_then(|s| s.trim().parse::<u16>().ok())
        .unwrap_or(DEFAULT_NFLOG_GROUP);
    let backend = select_backend_from_env();
    Some(PerFlowActivation {
        cell_id: cell_id.to_string(),
        run_id: run_id.to_string(),
        nflog_group,
        backend,
        policy_digest,
        keyset_id,
        issuer_kid,
    })
}

/// Resolve the backend from [`ENV_PER_FLOW_BACKEND`] (legacy) or
/// [`ENV_PER_FLOW_BACKEND_E7`] (spec-canonical name).
///
/// `ebpf` is honoured: [`crate::ebpf_flow::EbpfFlowMonitor::start`] will
/// be attempted at activation time. Runtime fallback to nflog happens
/// *inside* the activation path when the monitor returns an
/// [`crate::ebpf_flow::EbpfMonitorError`] (Linux-only build, missing BPF
/// object, attach failure, etc.) — operators see a single
/// `tracing::warn` per-cell and the cell still emits per-flow events
/// via nflog.
///
/// **Precedence**: when both env vars are set, the legacy name wins.
/// This matches the same shape as [`build_activation_from_env`] where
/// both opt-in names are accepted but the legacy one is the precedent.
/// Operators migrating to the new name should clear the legacy one to
/// avoid surprise.
///
/// Any value other than `ebpf` (including unset / empty / typos) maps
/// to [`PerFlowBackend::Nflog`] — silently, because the supervisor
/// pre-Phase-2 already defaulted to nflog and unknown values must not
/// regress the existing path.
pub fn select_backend_from_env() -> PerFlowBackend {
    let legacy = std::env::var(ENV_PER_FLOW_BACKEND).ok();
    let canonical = std::env::var(ENV_PER_FLOW_BACKEND_E7).ok();
    let resolved = legacy.as_deref().or(canonical.as_deref()).unwrap_or("");
    match resolved {
        "ebpf" => PerFlowBackend::Ebpf,
        _ => PerFlowBackend::Nflog,
    }
}

// ────────────────────────────────────────────────────────────────────────
// Pure ruleset augmentation
// ────────────────────────────────────────────────────────────────────────

/// Idempotently rewrite an nft ruleset string so every `accept` / `drop`
/// verdict is preceded by a `log group N prefix "cellos-flow {accept|drop}"`
/// action.
///
/// **Idempotent**: a second call on already-augmented output is a no-op;
/// rules whose tail already contains a `log group ` prefix are left alone.
///
/// **Skipped lines**: loopback shortcut (`oif "lo" accept`) and policy
/// declarations (`policy drop;` / `type filter hook ...`) are NOT augmented
/// — we only care about the explicit allow / deny attribution lines.
///
/// The original verdict is preserved verbatim at the end of the rewritten
/// rule so the rewrite is policy-neutral. The `log` action is a side-effect
/// in nftables; control flow continues to the trailing `accept` / `drop`.
pub fn augment_ruleset_with_log_actions(ruleset: &str, group: u16) -> String {
    let mut out = String::with_capacity(ruleset.len() + 64);
    for (idx, line) in ruleset.lines().enumerate() {
        if idx > 0 {
            out.push('\n');
        }
        let augmented = augment_line(line, group);
        out.push_str(&augmented);
    }
    // Preserve trailing newline if present so callers diffing strings byte
    // for byte don't see a phantom change.
    if ruleset.ends_with('\n') && !out.ends_with('\n') {
        out.push('\n');
    }
    out
}

fn augment_line(line: &str, group: u16) -> String {
    let trimmed = line.trim_start();
    let leading_ws_len = line.len() - trimmed.len();
    // Skip already-augmented lines (idempotency).
    if trimmed.contains("log group ") {
        return line.to_string();
    }
    // Skip the loopback shortcut — we don't audit `lo` traffic.
    if trimmed.starts_with("oif \"lo\" accept") {
        return line.to_string();
    }
    // Skip chain / policy / brace lines — no verdict to augment.
    if trimmed.starts_with("policy ")
        || trimmed.starts_with("type filter hook")
        || trimmed.starts_with("chain ")
        || trimmed.starts_with("table ")
        || trimmed == "}"
        || trimmed == "{"
        || trimmed.is_empty()
    {
        return line.to_string();
    }
    // Determine the trailing verdict (must be the last whitespace-separated
    // token). We only augment lines whose verdict is `accept` or `drop`; any
    // other terminator (return, jump, queue, ...) is left untouched.
    let last_tok = trimmed.split_whitespace().last().unwrap_or("");
    let verdict = match last_tok {
        "accept" => "accept",
        "drop" => "drop",
        _ => return line.to_string(),
    };
    // Reuse the leading whitespace from the original line so indentation is
    // preserved in the rewritten ruleset.
    let prefix = &line[..leading_ws_len];
    let body_without_verdict = trimmed
        .rsplit_once(char::is_whitespace)
        .map(|(head, _)| head)
        .unwrap_or("");
    format!(
        "{prefix}{body_without_verdict} log group {group} prefix \"{LOG_PREFIX_BASE} {verdict}\" {verdict}"
    )
}

// ────────────────────────────────────────────────────────────────────────
// Pure datagram decode (used by the Linux listener + unit tests)
// ────────────────────────────────────────────────────────────────────────

/// One decoded nflog datagram — what the listener pulls out of an
/// `NFULNL_PACKET_HDR` netlink message.
///
/// `prefix` is the operator-set prefix string from the augmented `log group`
/// action; the listener filters on `prefix.starts_with(LOG_PREFIX_BASE)` so
/// stray nflog traffic on the same group from unrelated host tooling
/// doesn't pollute the cell's audit trail.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DecodedNflog {
    /// `cellos-flow accept` or `cellos-flow drop` (the verdict suffix is the
    /// classifier's signal).
    pub prefix: String,
    /// Raw L3 packet payload from `NFULA_PAYLOAD`. Empty when copy-mode is
    /// `META` rather than `PACKET`.
    pub payload: Vec<u8>,
}

/// Errors returned by [`decode_nflog_datagram`].
#[derive(Debug, thiserror::Error)]
pub enum NflogDecodeError {
    #[error("nflog datagram too short ({0} bytes)")]
    TooShort(usize),
    #[error("nflog datagram missing required attributes")]
    MissingAttrs,
}

// nfnetlink/nflog attribute type constants (from `linux/netfilter/nfnetlink_log.h`).
// Replicated here so we don't pull in an external crate just for a handful of
// integers. The values are a stable kernel ABI.
pub(crate) const NFULA_PACKET_HDR: u16 = 1;
pub(crate) const NFULA_PAYLOAD: u16 = 9;
pub(crate) const NFULA_PREFIX: u16 = 10;

// nflog command + copy-mode constants.
pub(crate) const NFULNL_CFG_CMD_BIND: u8 = 1;
pub(crate) const NFULNL_CFG_CMD_PF_BIND: u8 = 4;
pub(crate) const NFULNL_CFG_CMD_PF_UNBIND: u8 = 5;
pub(crate) const NFULNL_COPY_PACKET: u8 = 2;

/// nfnetlink subsys id for nflog (`NFNL_SUBSYS_ULOG`).
pub(crate) const NFNL_SUBSYS_ULOG: u8 = 4;
/// `NFULNL_MSG_PACKET` — the message type carrying captured packets.
pub(crate) const NFULNL_MSG_PACKET: u8 = 0;
/// `NFULNL_MSG_CONFIG` — used for bind / copy-mode commands.
pub(crate) const NFULNL_MSG_CONFIG: u8 = 1;

/// netlink socket family literal (`AF_NETLINK`). libc carries the constant
/// but we pin it here so the pure decoder unit-tests don't need libc on
/// non-Linux hosts.
pub(crate) const AF_NETLINK_LITERAL: i32 = 16;
/// `NETLINK_NETFILTER` socket protocol literal.
pub(crate) const NETLINK_NETFILTER_LITERAL: i32 = 12;

/// Decode the *body* of an `NFULNL_MSG_PACKET` netlink message — i.e. the
/// bytes following the `nlmsghdr` + `nfgenmsg` headers — into a
/// [`DecodedNflog`]. The walking logic understands the TLV ("attribute")
/// stream nflog uses: pairs of `(nla_len: u16, nla_type: u16, value...)`
/// padded to 4-byte alignment.
///
/// The minimal set of attributes Phase 1 consumes is `NFULA_PREFIX` (the
/// `log group ... prefix "..."` operator string) and `NFULA_PAYLOAD` (the
/// captured L3 packet bytes; empty when copy-mode is `META`). Other
/// attributes — hwaddr, indev, outdev, mark, ts, uid, gid — are skipped.
pub fn decode_nflog_datagram(body: &[u8]) -> Result<DecodedNflog, NflogDecodeError> {
    if body.len() < 4 {
        return Err(NflogDecodeError::TooShort(body.len()));
    }
    let mut prefix: Option<String> = None;
    let mut payload: Option<Vec<u8>> = None;

    let mut cursor = 0usize;
    while cursor + 4 <= body.len() {
        let nla_len = u16::from_ne_bytes([body[cursor], body[cursor + 1]]) as usize;
        let nla_type = u16::from_ne_bytes([body[cursor + 2], body[cursor + 3]]);
        if nla_len < 4 || cursor + nla_len > body.len() {
            break;
        }
        let value_start = cursor + 4;
        let value_end = cursor + nla_len;
        let value = &body[value_start..value_end];
        match nla_type {
            NFULA_PREFIX => {
                // NUL-terminated string.
                let s = value
                    .iter()
                    .position(|b| *b == 0)
                    .map(|n| &value[..n])
                    .unwrap_or(value);
                prefix = Some(String::from_utf8_lossy(s).into_owned());
            }
            NFULA_PAYLOAD => {
                payload = Some(value.to_vec());
            }
            _ => {}
        }
        // Advance to the next attribute, padded to 4 bytes.
        cursor += (nla_len + 3) & !3;
    }
    let prefix = prefix.ok_or(NflogDecodeError::MissingAttrs)?;
    Ok(DecodedNflog {
        prefix,
        payload: payload.unwrap_or_default(),
    })
}

/// L3/L4 attribution distilled from an nflog payload.
///
/// All fields are optional because nflog can be configured with copy-mode
/// `META` (no payload at all) or because the captured packet may be a
/// non-IP frame (very rare in practice — netfilter only hooks IP).
///
/// L5-15: `src_addr` / `src_port` / `protocol_byte` are populated alongside
/// the dst fields so the listener can derive a [`FlowKey`] suitable for
/// `FlowAccumulator::record()`. The wire-emitted `NetworkFlowDecision`
/// schema only carries dst attribution today, so the src fields are
/// listener-internal — they exist on this struct to keep one parsing pass
/// rather than re-walking the payload.
///
/// [`FlowKey`]: crate::ebpf_flow::connection_tracking::FlowKey
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct FlowAttribution {
    pub src_addr: Option<String>,
    pub src_port: Option<u16>,
    pub dst_addr: Option<String>,
    pub dst_port: Option<u16>,
    pub protocol: Option<String>,
    /// IANA protocol byte (6 = TCP, 17 = UDP, 1 = ICMP, 58 = ICMPv6). Kept
    /// alongside the human-readable `protocol` string so the FlowKey hash
    /// stays cheap and matches the `FlowKey::protocol: u8` field directly.
    pub protocol_byte: Option<u8>,
}

/// Decode the L3 + L4 headers of an nflog payload to extract destination
/// IP / port / transport protocol. Returns a default-empty
/// [`FlowAttribution`] for any parse failure — we deliberately degrade
/// gracefully rather than drop the entire decision event.
///
/// Supported shapes:
/// - IPv4 + TCP / UDP / ICMP
/// - IPv6 + TCP / UDP / ICMPv6
///
/// Extension headers in IPv6 are NOT walked; an exotic packet with
/// HBH/Routing/Fragment headers between the IPv6 fixed header and the
/// transport will surface with `protocol: None` and no port.
pub fn decode_l3_l4_attribution(payload: &[u8]) -> FlowAttribution {
    if payload.is_empty() {
        return FlowAttribution::default();
    }
    let version = payload[0] >> 4;
    match version {
        4 => decode_ipv4(payload),
        6 => decode_ipv6(payload),
        _ => FlowAttribution::default(),
    }
}

fn decode_ipv4(p: &[u8]) -> FlowAttribution {
    if p.len() < 20 {
        return FlowAttribution::default();
    }
    let ihl = (p[0] & 0x0f) as usize * 4;
    if ihl < 20 || p.len() < ihl {
        return FlowAttribution::default();
    }
    let proto_byte = p[9];
    let src = std::net::Ipv4Addr::new(p[12], p[13], p[14], p[15]).to_string();
    let dst = std::net::Ipv4Addr::new(p[16], p[17], p[18], p[19]).to_string();
    let mut attr = FlowAttribution {
        src_addr: Some(src),
        src_port: None,
        dst_addr: Some(dst),
        dst_port: None,
        protocol: None,
        protocol_byte: Some(proto_byte),
    };
    match proto_byte {
        6 => {
            attr.protocol = Some("tcp".to_string());
            attr.src_port = parse_src_port(p, ihl);
            attr.dst_port = parse_dst_port(p, ihl);
        }
        17 => {
            attr.protocol = Some("udp".to_string());
            attr.src_port = parse_src_port(p, ihl);
            attr.dst_port = parse_dst_port(p, ihl);
        }
        1 => attr.protocol = Some("icmp".to_string()),
        _ => {}
    }
    attr
}

fn decode_ipv6(p: &[u8]) -> FlowAttribution {
    if p.len() < 40 {
        return FlowAttribution::default();
    }
    let next_header = p[6];
    let mut src_octets = [0u8; 16];
    src_octets.copy_from_slice(&p[8..24]);
    let src = std::net::Ipv6Addr::from(src_octets).to_string();
    let mut dst_octets = [0u8; 16];
    dst_octets.copy_from_slice(&p[24..40]);
    let dst = std::net::Ipv6Addr::from(dst_octets).to_string();
    let mut attr = FlowAttribution {
        src_addr: Some(src),
        src_port: None,
        dst_addr: Some(dst),
        dst_port: None,
        protocol: None,
        protocol_byte: Some(next_header),
    };
    match next_header {
        6 => {
            attr.protocol = Some("tcp".to_string());
            attr.src_port = parse_src_port(p, 40);
            attr.dst_port = parse_dst_port(p, 40);
        }
        17 => {
            attr.protocol = Some("udp".to_string());
            attr.src_port = parse_src_port(p, 40);
            attr.dst_port = parse_dst_port(p, 40);
        }
        58 => attr.protocol = Some("icmp6".to_string()),
        _ => {}
    }
    attr
}

fn parse_src_port(p: &[u8], l4_offset: usize) -> Option<u16> {
    if p.len() < l4_offset + 4 {
        return None;
    }
    Some(u16::from_be_bytes([p[l4_offset], p[l4_offset + 1]]))
}

fn parse_dst_port(p: &[u8], l4_offset: usize) -> Option<u16> {
    if p.len() < l4_offset + 4 {
        return None;
    }
    Some(u16::from_be_bytes([p[l4_offset + 2], p[l4_offset + 3]]))
}

/// L5-15 — derive a [`FlowKey`] from a parsed [`FlowAttribution`].
///
/// Returns `None` when any required 5-tuple field is missing — non-IP
/// payloads, copy-mode `META` datagrams (no payload), or transport
/// protocols without ports (ICMP, ICMPv6) all surface as `None`. The
/// caller treats `None` as "skip the accumulator update" rather than
/// fabricating a synthetic key.
///
/// [`FlowKey`]: crate::ebpf_flow::connection_tracking::FlowKey
pub fn flow_key_from_attribution(
    attr: &FlowAttribution,
) -> Option<crate::ebpf_flow::connection_tracking::FlowKey> {
    use crate::ebpf_flow::connection_tracking::FlowKey;
    let src_addr: std::net::IpAddr = attr.src_addr.as_deref()?.parse().ok()?;
    let dst_addr: std::net::IpAddr = attr.dst_addr.as_deref()?.parse().ok()?;
    let src_port = attr.src_port?;
    let dst_port = attr.dst_port?;
    let protocol = attr.protocol_byte?;
    Some(FlowKey {
        src_addr,
        src_port,
        dst_addr,
        dst_port,
        protocol,
    })
}

/// L5-15 — record an Opened flow into the shared accumulator, swallowing
/// mutex poisoning (a poisoned accumulator is a bug elsewhere, but the
/// listener MUST NOT panic on the recv path or the cell loses every
/// subsequent per-flow event). On `PoisonError` we still walk through the
/// inner accumulator via `into_inner()`-style recovery (`get_mut` on the
/// guard) so the count remains correct even after the panic that poisoned
/// the lock.
pub fn record_opened_flow(
    accumulator: &std::sync::Arc<
        std::sync::Mutex<crate::ebpf_flow::connection_tracking::FlowAccumulator>,
    >,
    key: crate::ebpf_flow::connection_tracking::FlowKey,
) {
    use crate::ebpf_flow::connection_tracking::{FlowEvent, FlowEventKind};
    let event = FlowEvent {
        key,
        kind: FlowEventKind::Opened,
        timestamp_ns: 0,
    };
    match accumulator.lock() {
        Ok(mut guard) => guard.record(&event),
        Err(poisoned) => {
            // Lock was poisoned by an unrelated panic. Continue recording
            // rather than dropping the event — `unique_flow_count` is
            // monotonic so a partially-updated set is still a strict
            // undercount, never an overcount.
            poisoned.into_inner().record(&event);
        }
    }
}

// ────────────────────────────────────────────────────────────────────────
// Decision builder
// ────────────────────────────────────────────────────────────────────────

/// Compose a [`NetworkFlowDecision`] from a decoded nflog packet + the
/// activation context.
///
/// `prefix` carries the verdict suffix (`cellos-flow accept` or
/// `cellos-flow drop`). The `decision` enum + `reason_code` follow the same
/// convention as [`crate::nft_counters::classify_rule_repr`] but with a
/// per-flow flavour: `nft_log_accept` / `nft_log_drop` mark this as a
/// real-time event rather than a post-run scrape attribution. Operators
/// querying the audit trail can `.filter(reasonCode startsWith "nft_log_")`
/// to isolate the per-flow stream.
pub fn build_decision(
    activation: &PerFlowActivation,
    prefix: &str,
    payload: &[u8],
    observed_at: &str,
) -> NetworkFlowDecision {
    let attribution = decode_l3_l4_attribution(payload);
    let (decision, reason_code) = if prefix.contains("accept") {
        (NetworkFlowDecisionOutcome::Allow, "nft_log_accept")
    } else if prefix.contains("drop") {
        (NetworkFlowDecisionOutcome::Deny, "nft_log_drop")
    } else {
        // Unknown verdict — surface as deny default-drop so operators see
        // it rather than silently dropping the event.
        (NetworkFlowDecisionOutcome::Deny, "nft_log_unknown")
    };
    NetworkFlowDecision {
        schema_version: "1.0.0".to_string(),
        cell_id: activation.cell_id.clone(),
        run_id: activation.run_id.clone(),
        decision_id: uuid::Uuid::new_v4().to_string(),
        direction: NetworkFlowDirection::Egress,
        decision,
        reason_code: reason_code.to_string(),
        nft_rule_ref: None,
        dst_addr: attribution.dst_addr,
        dst_port: attribution.dst_port,
        protocol: attribution.protocol,
        packet_count: None,
        byte_count: None,
        policy_digest: activation.policy_digest.clone(),
        keyset_id: activation.keyset_id.clone(),
        issuer_kid: activation.issuer_kid.clone(),
        correlation_id: None,
        observed_at: observed_at.to_string(),
    }
}

// ────────────────────────────────────────────────────────────────────────
// Linux-only listener spawn
// ────────────────────────────────────────────────────────────────────────

/// Handle returned by [`spawn_per_flow_listener_in_netns`].
pub struct PerFlowListenerHandle {
    /// Set to `true` to ask the listener to exit at the next iteration
    /// (within `LISTENER_RECV_TIMEOUT_MS` worst case).
    pub shutdown: Arc<AtomicBool>,
    /// Join handle for the listener OS thread. `Some` until [`Self::join`]
    /// consumes it.
    #[cfg(target_os = "linux")]
    pub thread: Option<std::thread::JoinHandle<PerFlowListenerStats>>,
}

/// Aggregate stats returned when the listener thread exits.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct PerFlowListenerStats {
    /// Total nflog datagrams the listener observed.
    pub datagrams_total: u64,
    /// Datagrams whose prefix matched [`LOG_PREFIX_BASE`] (i.e. originated
    /// from the augmented ruleset).
    pub datagrams_matched: u64,
    /// Datagrams that failed [`decode_nflog_datagram`].
    pub datagrams_decode_failed: u64,
    /// Events emitted to the supervisor's sink.
    pub events_emitted: u64,
}

impl PerFlowListenerHandle {
    /// Join the listener thread and return its cumulative stats.
    #[cfg(target_os = "linux")]
    pub fn join(&mut self) -> Option<PerFlowListenerStats> {
        let handle = self.thread.take()?;
        match handle.join() {
            Ok(s) => Some(s),
            Err(_) => {
                tracing::warn!(
                    target: "cellos.supervisor.per_flow",
                    "per-flow listener thread panicked on join"
                );
                None
            }
        }
    }

    /// Stub join on non-Linux platforms.
    #[cfg(not(target_os = "linux"))]
    pub fn join(&mut self) -> Option<PerFlowListenerStats> {
        None
    }
}

/// Worst-case shutdown latency: the listener's `SO_RCVTIMEO` is set to
/// 100ms so the recv loop checks the shutdown flag at that cadence.
#[cfg(target_os = "linux")]
const LISTENER_RECV_TIMEOUT_MS: i64 = 100;

/// Linux-only — spawn a dedicated OS thread, `setns(2)` into
/// `/proc/<child_pid>/ns/net`, open a `NETLINK_NETFILTER` socket, bind the
/// nflog group, and emit one [`NetworkFlowDecision`] CloudEvent per
/// matching datagram until `shutdown` is set.
///
/// **Capability requirement**: the supervisor must hold `CAP_NET_ADMIN` to
/// open `NETLINK_NETFILTER`. If the syscall fails with `EPERM` the listener
/// thread emits a single `tracing::warn` and exits cleanly — the cell still
/// runs, just without per-flow events.
#[cfg(target_os = "linux")]
pub fn spawn_per_flow_listener_in_netns(
    child_pid: u32,
    activation: PerFlowActivation,
    sink: Arc<dyn cellos_core::ports::EventSink>,
    shutdown: Arc<AtomicBool>,
    accumulator: Option<
        Arc<std::sync::Mutex<crate::ebpf_flow::connection_tracking::FlowAccumulator>>,
    >,
) -> std::io::Result<PerFlowListenerHandle> {
    use std::fs::File;
    use std::os::unix::io::AsRawFd;

    let netns_path = format!("/proc/{child_pid}/ns/net");
    let netns_file = File::open(&netns_path)
        .map_err(|e| std::io::Error::new(e.kind(), format!("open netns at {netns_path}: {e}")))?;

    let runtime_handle = tokio::runtime::Handle::try_current().ok();
    let shutdown_for_thread = shutdown.clone();

    let thread = std::thread::Builder::new()
        .name(format!("cellos-per-flow-{child_pid}"))
        .spawn(move || {
            // SAFETY: setns is the documented Linux syscall for moving the
            // calling thread into the namespace referenced by `fd`. The
            // listener thread never returns to a tokio worker, so polluting
            // its netns is intentional and isolated. `netns_file` is held
            // for the lifetime of the thread.
            let setns_rc = unsafe { libc::setns(netns_file.as_raw_fd(), libc::CLONE_NEWNET) };
            if setns_rc != 0 {
                let err = std::io::Error::last_os_error();
                tracing::warn!(
                    target: "cellos.supervisor.per_flow",
                    error = %err,
                    child_pid = child_pid,
                    "setns(CLONE_NEWNET) failed — per-flow listener bailing"
                );
                return PerFlowListenerStats::default();
            }
            run_listener_loop(
                activation,
                sink,
                shutdown_for_thread,
                runtime_handle,
                accumulator,
            )
        })?;

    Ok(PerFlowListenerHandle {
        shutdown,
        thread: Some(thread),
    })
}

/// Stub on non-Linux platforms: per-flow events require netfilter / nflog
/// which are Linux-only.
#[cfg(not(target_os = "linux"))]
pub fn spawn_per_flow_listener_in_netns(
    _child_pid: u32,
    _activation: PerFlowActivation,
    _sink: Arc<dyn cellos_core::ports::EventSink>,
    shutdown: Arc<AtomicBool>,
    _accumulator: Option<
        Arc<std::sync::Mutex<crate::ebpf_flow::connection_tracking::FlowAccumulator>>,
    >,
) -> std::io::Result<PerFlowListenerHandle> {
    Ok(PerFlowListenerHandle { shutdown })
}

#[cfg(target_os = "linux")]
fn run_listener_loop(
    activation: PerFlowActivation,
    sink: Arc<dyn cellos_core::ports::EventSink>,
    shutdown: Arc<AtomicBool>,
    runtime_handle: Option<tokio::runtime::Handle>,
    accumulator: Option<
        Arc<std::sync::Mutex<crate::ebpf_flow::connection_tracking::FlowAccumulator>>,
    >,
) -> PerFlowListenerStats {
    use std::os::unix::io::FromRawFd;
    use std::os::unix::io::OwnedFd;

    let mut stats = PerFlowListenerStats::default();

    // Open AF_NETLINK / NETLINK_NETFILTER socket.
    let sock_fd =
        unsafe { libc::socket(libc::AF_NETLINK, libc::SOCK_RAW, NETLINK_NETFILTER_LITERAL) };
    if sock_fd < 0 {
        let err = std::io::Error::last_os_error();
        tracing::warn!(
            target: "cellos.supervisor.per_flow",
            error = %err,
            "socket(AF_NETLINK, NETLINK_NETFILTER) failed — per-flow listener bailing"
        );
        return stats;
    }
    // SAFETY: sock_fd is a fresh kernel-owned fd we just opened; OwnedFd
    // takes responsibility for closing on drop, including on early returns
    // from this function. The variable is bound (rather than `_`) so its
    // Drop impl runs at function exit.
    let _sock_guard = unsafe { OwnedFd::from_raw_fd(sock_fd) };

    // Bind to PID 0 (kernel-assigned) — nflog only needs a valid sockaddr_nl.
    let mut sa: libc::sockaddr_nl = unsafe { std::mem::zeroed() };
    sa.nl_family = libc::AF_NETLINK as u16;
    sa.nl_pid = 0;
    sa.nl_groups = 0;
    let bind_rc = unsafe {
        libc::bind(
            sock_fd,
            &sa as *const _ as *const libc::sockaddr,
            std::mem::size_of::<libc::sockaddr_nl>() as u32,
        )
    };
    if bind_rc != 0 {
        let err = std::io::Error::last_os_error();
        tracing::warn!(
            target: "cellos.supervisor.per_flow",
            error = %err,
            "bind() on netlink socket failed — per-flow listener bailing"
        );
        return stats;
    }

    // Configure SO_RCVTIMEO so the recv loop wakes every 100ms to check
    // the shutdown flag.
    let tv = libc::timeval {
        tv_sec: 0,
        tv_usec: (LISTENER_RECV_TIMEOUT_MS * 1000) as libc::suseconds_t,
    };
    let _ = unsafe {
        libc::setsockopt(
            sock_fd,
            libc::SOL_SOCKET,
            libc::SO_RCVTIMEO,
            &tv as *const _ as *const libc::c_void,
            std::mem::size_of::<libc::timeval>() as u32,
        )
    };

    // PF_BIND for AF_INET + AF_INET6, then BIND group N, then COPY_PACKET mode.
    if let Err(e) = send_nflog_cfg_pf_bind(sock_fd, libc::AF_INET as u16) {
        tracing::warn!(target: "cellos.supervisor.per_flow", error=%e, "PF_BIND v4 failed");
        return stats;
    }
    if let Err(e) = send_nflog_cfg_pf_bind(sock_fd, libc::AF_INET6 as u16) {
        tracing::warn!(target: "cellos.supervisor.per_flow", error=%e, "PF_BIND v6 failed");
        return stats;
    }
    if let Err(e) = send_nflog_cfg_bind_group(sock_fd, activation.nflog_group) {
        tracing::warn!(target: "cellos.supervisor.per_flow", error=%e, "BIND group failed");
        return stats;
    }
    if let Err(e) = send_nflog_cfg_copy_packet(sock_fd, activation.nflog_group) {
        tracing::warn!(target: "cellos.supervisor.per_flow", error=%e, "COPY_PACKET failed");
        return stats;
    }

    // Recv loop.
    let mut buf = vec![0u8; 65536];
    use std::sync::atomic::Ordering;
    while !shutdown.load(Ordering::SeqCst) {
        let n = unsafe { libc::recv(sock_fd, buf.as_mut_ptr() as *mut libc::c_void, buf.len(), 0) };
        if n < 0 {
            let err = std::io::Error::last_os_error();
            // EAGAIN / EWOULDBLOCK is the timeout firing — loop & re-check shutdown.
            // EWOULDBLOCK == EAGAIN on Linux glibc/musl, so one arm covers both.
            if matches!(err.raw_os_error(), Some(libc::EAGAIN)) {
                continue;
            }
            tracing::debug!(
                target: "cellos.supervisor.per_flow",
                error = %err,
                "recv() error — exiting listener loop"
            );
            break;
        }
        let received = &buf[..n as usize];
        stats.datagrams_total += 1;
        let mut offset = 0usize;
        while offset < received.len() {
            // Parse nlmsghdr (16 bytes on 64-bit).
            if received.len() - offset < 16 {
                break;
            }
            let nlmsg_len = u32::from_ne_bytes([
                received[offset],
                received[offset + 1],
                received[offset + 2],
                received[offset + 3],
            ]) as usize;
            let nlmsg_type = u16::from_ne_bytes([received[offset + 4], received[offset + 5]]);
            if nlmsg_len < 16 || offset + nlmsg_len > received.len() {
                break;
            }
            let subsys = (nlmsg_type >> 8) as u8;
            let msg_kind = (nlmsg_type & 0xff) as u8;
            // Skip headers: nlmsghdr (16) + nfgenmsg (4) = 20 bytes.
            let body_start = offset + 16 + 4;
            if subsys == NFNL_SUBSYS_ULOG
                && msg_kind == NFULNL_MSG_PACKET
                && body_start <= offset + nlmsg_len
            {
                let body = &received[body_start..offset + nlmsg_len];
                match decode_nflog_datagram(body) {
                    Ok(decoded) => {
                        if decoded.prefix.starts_with(LOG_PREFIX_BASE) {
                            stats.datagrams_matched += 1;
                            // L5-15 — populate the FlowAccumulator from the
                            // nflog event path so `exercised_egress_connections`
                            // reports the real count even when eBPF isn't wired.
                            // We only count `accept` verdicts: a dropped packet
                            // means the workload TRIED to open a flow that
                            // policy denied; the homeostasis signal counts
                            // ACTUALLY EXERCISED connections (a denied syn is
                            // not an exercised connection).
                            if decoded.prefix.contains("accept") {
                                if let Some(acc) = accumulator.as_ref() {
                                    let attribution = decode_l3_l4_attribution(&decoded.payload);
                                    if let Some(key) = flow_key_from_attribution(&attribution) {
                                        record_opened_flow(acc, key);
                                    }
                                }
                            }
                            let now = chrono::Utc::now().to_rfc3339();
                            let decision = build_decision(
                                &activation,
                                &decoded.prefix,
                                &decoded.payload,
                                &now,
                            );
                            match cloud_event_v1_network_flow_decision(
                                "cellos-supervisor",
                                &now,
                                &decision,
                            ) {
                                Ok(event) => {
                                    if let Some(rt) = runtime_handle.as_ref() {
                                        let sink = sink.clone();
                                        rt.spawn(async move {
                                            if let Err(e) = sink.emit(&event).await {
                                                tracing::warn!(
                                                    target: "cellos.supervisor.per_flow",
                                                    error = %e,
                                                    "sink emit failed for network_flow_decision event"
                                                );
                                            }
                                        });
                                        stats.events_emitted += 1;
                                    }
                                }
                                Err(e) => {
                                    tracing::debug!(
                                        target: "cellos.supervisor.per_flow",
                                        error = %e,
                                        "network_flow_decision envelope build failed"
                                    );
                                }
                            }
                        }
                    }
                    Err(_) => {
                        stats.datagrams_decode_failed += 1;
                    }
                }
            }
            // 4-byte align next message.
            offset += (nlmsg_len + 3) & !3;
        }
    }
    stats
}

#[cfg(target_os = "linux")]
fn send_nflog_cfg_pf_bind(sock_fd: i32, pf: u16) -> std::io::Result<()> {
    // NFULA_CFG_CMD attribute carries one byte (the cmd id).
    let payload = build_nfnl_cfg_msg(0, NFULNL_CFG_CMD_PF_BIND, pf);
    send_netlink(sock_fd, &payload)
}

#[cfg(target_os = "linux")]
fn send_nflog_cfg_bind_group(sock_fd: i32, group: u16) -> std::io::Result<()> {
    let payload = build_nfnl_cfg_msg(group, NFULNL_CFG_CMD_BIND, 0);
    send_netlink(sock_fd, &payload)
}

#[cfg(target_os = "linux")]
fn send_nflog_cfg_copy_packet(sock_fd: i32, group: u16) -> std::io::Result<()> {
    // NFULA_CFG_MODE: u32 copy_range (0 = full) + u8 copy_mode + 3 bytes pad.
    let mut buf = Vec::new();
    // nlmsghdr: len=24 + attr=12 = 36 (filled in by build_nfnl_cfg_envelope).
    let attr_type: u16 = 2; // NFULA_CFG_MODE
    let nla_len: u16 = 4 + 4 + 4; // header(4) + copy_range(4) + copy_mode+pad(4)
    buf.extend_from_slice(&nla_len.to_ne_bytes());
    buf.extend_from_slice(&attr_type.to_ne_bytes());
    buf.extend_from_slice(&0u32.to_be_bytes()); // copy_range = 0xffff (we send 0 → full pkt up to MTU)
    buf.push(NFULNL_COPY_PACKET);
    buf.extend_from_slice(&[0u8; 3]); // pad
    let envelope = build_nfnl_cfg_envelope(group, &buf);
    send_netlink(sock_fd, &envelope)
}

/// Build a CFG message carrying a single `NFULA_CFG_CMD` attribute.
#[cfg(target_os = "linux")]
fn build_nfnl_cfg_msg(group: u16, cmd: u8, pf: u16) -> Vec<u8> {
    // attribute: nla_len=8 (4 header + 4 padded body), nla_type=1 (NFULA_CFG_CMD)
    let mut attr = Vec::new();
    let nla_len: u16 = 4 + 1 + 3; // header(4) + 1 byte cmd + 3 byte pad → align to 4
    attr.extend_from_slice(&nla_len.to_ne_bytes());
    attr.extend_from_slice(&1u16.to_ne_bytes()); // NFULA_CFG_CMD
    attr.push(cmd);
    attr.extend_from_slice(&[0u8; 3]); // pad
    let _ = pf; // currently embedded in nfgenmsg.res_id (caller sets it via group)
    build_nfnl_cfg_envelope(group, &attr)
}

/// Build the full netlink envelope: nlmsghdr (16) + nfgenmsg (4) + attrs.
#[cfg(target_os = "linux")]
fn build_nfnl_cfg_envelope(group: u16, attrs: &[u8]) -> Vec<u8> {
    let total_len: u32 = 16 + 4 + attrs.len() as u32;
    let mut out = Vec::with_capacity(total_len as usize);
    // nlmsghdr.
    out.extend_from_slice(&total_len.to_ne_bytes());
    // type = NFNL_SUBSYS_ULOG << 8 | NFULNL_MSG_CONFIG
    let nlmsg_type: u16 = ((NFNL_SUBSYS_ULOG as u16) << 8) | NFULNL_MSG_CONFIG as u16;
    out.extend_from_slice(&nlmsg_type.to_ne_bytes());
    // flags = NLM_F_REQUEST | NLM_F_ACK
    let flags: u16 = (libc::NLM_F_REQUEST | libc::NLM_F_ACK) as u16;
    out.extend_from_slice(&flags.to_ne_bytes());
    // seq = 0, pid = 0
    out.extend_from_slice(&0u32.to_ne_bytes());
    out.extend_from_slice(&0u32.to_ne_bytes());
    // nfgenmsg: family(u8) + version(u8) + res_id(u16 BE = nflog group)
    out.push(0); // family unspec
    out.push(0); // version 0
    out.extend_from_slice(&group.to_be_bytes()); // res_id is BE in netfilter
    out.extend_from_slice(attrs);
    out
}

#[cfg(target_os = "linux")]
fn send_netlink(sock_fd: i32, payload: &[u8]) -> std::io::Result<()> {
    let mut sa: libc::sockaddr_nl = unsafe { std::mem::zeroed() };
    sa.nl_family = libc::AF_NETLINK as u16;
    let n = unsafe {
        libc::sendto(
            sock_fd,
            payload.as_ptr() as *const libc::c_void,
            payload.len(),
            0,
            &sa as *const _ as *const libc::sockaddr,
            std::mem::size_of::<libc::sockaddr_nl>() as u32,
        )
    };
    if n < 0 {
        return Err(std::io::Error::last_os_error());
    }
    Ok(())
}

// ────────────────────────────────────────────────────────────────────────
// Tests (18 unit tests — pure parsing + classification + decision builder)
// ────────────────────────────────────────────────────────────────────────

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

    /// Tests in this module that mutate any of the per-flow env vars
    /// (`CELLOS_FIRECRACKER_PER_FLOW_EBPF`, `CELLOS_PER_FLOW_REALTIME`,
    /// `CELLOS_FIRECRACKER_PER_FLOW_BACKEND`, `CELLOS_PER_FLOW_BACKEND`) must
    /// hold this lock for their entire duration. cargo runs tests in parallel
    /// by default and the env namespace is process-global — without
    /// serialisation, the activation/selector tests race each other and the
    /// "unset means None" contract flakes when a sibling test has just
    /// written a value between the `remove_var` and the `build_activation_from_env`
    /// call. Same pattern as `dns_proxy::upstream::tests::ENV_LOCK` and
    /// `linux_seccomp::tests::ENV_LOCK`.
    static ENV_LOCK: Mutex<()> = Mutex::new(());

    fn sample_activation() -> PerFlowActivation {
        PerFlowActivation {
            cell_id: "cell-test".to_string(),
            run_id: "run-uuid".to_string(),
            nflog_group: 100,
            backend: PerFlowBackend::Nflog,
            policy_digest: Some("sha256:deadbeef".to_string()),
            keyset_id: Some("keyset-1".to_string()),
            issuer_kid: Some("kid-1".to_string()),
        }
    }

    // ── env-driven activation ─────────────────────────────────────────
    #[test]
    fn build_activation_returns_none_when_env_unset() {
        // ENV_LOCK serialises this test against the realtime/legacy
        // activation-on tests below. Without the lock, a sibling test
        // setting `CELLOS_PER_FLOW_REALTIME=1` between our `remove_var`
        // and the `build_activation_from_env` call would return Some and
        // flake this test. We also clear both opt-in vars — the original
        // version only cleared `_EBPF`, missing the `_REALTIME` alias.
        let _g = ENV_LOCK.lock().unwrap_or_else(|p| p.into_inner());
        let prev_ebpf = std::env::var(ENV_PER_FLOW_EBPF).ok();
        let prev_rt = std::env::var(ENV_PER_FLOW_REALTIME).ok();
        // SAFETY: env mutation in tests is serialised by ENV_LOCK above.
        unsafe {
            std::env::remove_var(ENV_PER_FLOW_EBPF);
            std::env::remove_var(ENV_PER_FLOW_REALTIME);
        }
        let act = build_activation_from_env("c", "r", None, None, None);
        assert!(act.is_none());
        // Restore prior values so we don't leak state to subsequent tests
        // that might run while still holding the lock-released window.
        unsafe {
            match prev_ebpf {
                Some(v) => std::env::set_var(ENV_PER_FLOW_EBPF, v),
                None => std::env::remove_var(ENV_PER_FLOW_EBPF),
            }
            match prev_rt {
                Some(v) => std::env::set_var(ENV_PER_FLOW_REALTIME, v),
                None => std::env::remove_var(ENV_PER_FLOW_REALTIME),
            }
        }
    }

    #[test]
    fn select_backend_defaults_to_nflog() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|p| p.into_inner());
        let prev_legacy = std::env::var(ENV_PER_FLOW_BACKEND).ok();
        let prev_canonical = std::env::var(ENV_PER_FLOW_BACKEND_E7).ok();
        unsafe {
            std::env::remove_var(ENV_PER_FLOW_BACKEND);
            std::env::remove_var(ENV_PER_FLOW_BACKEND_E7);
        }
        assert_eq!(select_backend_from_env(), PerFlowBackend::Nflog);
        unsafe {
            if let Some(v) = prev_legacy {
                std::env::set_var(ENV_PER_FLOW_BACKEND, v);
            }
            if let Some(v) = prev_canonical {
                std::env::set_var(ENV_PER_FLOW_BACKEND_E7, v);
            }
        }
    }

    #[test]
    fn select_backend_ebpf_returns_ebpf_variant() {
        // E7-5: the selector now returns `PerFlowBackend::Ebpf` when
        // operators opt in. Runtime fallback to nflog (on BPF object
        // missing, attach failure, non-Linux host) happens *inside*
        // the activation path, not here. This change is observable:
        // callers can now log which backend was selected vs which one
        // actually attached.
        let _g = ENV_LOCK.lock().unwrap_or_else(|p| p.into_inner());
        let prev_legacy = std::env::var(ENV_PER_FLOW_BACKEND).ok();
        let prev_canonical = std::env::var(ENV_PER_FLOW_BACKEND_E7).ok();
        unsafe {
            std::env::remove_var(ENV_PER_FLOW_BACKEND_E7);
            std::env::set_var(ENV_PER_FLOW_BACKEND, "ebpf");
        }
        assert_eq!(select_backend_from_env(), PerFlowBackend::Ebpf);
        unsafe {
            match prev_legacy {
                Some(v) => std::env::set_var(ENV_PER_FLOW_BACKEND, v),
                None => std::env::remove_var(ENV_PER_FLOW_BACKEND),
            }
            match prev_canonical {
                Some(v) => std::env::set_var(ENV_PER_FLOW_BACKEND_E7, v),
                None => std::env::remove_var(ENV_PER_FLOW_BACKEND_E7),
            }
        }
    }

    #[test]
    fn select_backend_honours_e7_canonical_env_name() {
        // E7-5: the spec-canonical env name is CELLOS_PER_FLOW_BACKEND
        // (no _FIRECRACKER_ infix). Setting only that var must opt in
        // to eBPF.
        let _g = ENV_LOCK.lock().unwrap_or_else(|p| p.into_inner());
        let prev_legacy = std::env::var(ENV_PER_FLOW_BACKEND).ok();
        let prev_canonical = std::env::var(ENV_PER_FLOW_BACKEND_E7).ok();
        unsafe {
            std::env::remove_var(ENV_PER_FLOW_BACKEND);
            std::env::set_var(ENV_PER_FLOW_BACKEND_E7, "ebpf");
        }
        assert_eq!(select_backend_from_env(), PerFlowBackend::Ebpf);
        unsafe {
            match prev_legacy {
                Some(v) => std::env::set_var(ENV_PER_FLOW_BACKEND, v),
                None => std::env::remove_var(ENV_PER_FLOW_BACKEND),
            }
            match prev_canonical {
                Some(v) => std::env::set_var(ENV_PER_FLOW_BACKEND_E7, v),
                None => std::env::remove_var(ENV_PER_FLOW_BACKEND_E7),
            }
        }
    }

    #[test]
    fn select_backend_legacy_name_wins_over_canonical_when_both_set() {
        // E7-5 precedence: legacy var wins on conflict.
        let _g = ENV_LOCK.lock().unwrap_or_else(|p| p.into_inner());
        let prev_legacy = std::env::var(ENV_PER_FLOW_BACKEND).ok();
        let prev_canonical = std::env::var(ENV_PER_FLOW_BACKEND_E7).ok();
        unsafe {
            std::env::set_var(ENV_PER_FLOW_BACKEND, "nflog");
            std::env::set_var(ENV_PER_FLOW_BACKEND_E7, "ebpf");
        }
        assert_eq!(
            select_backend_from_env(),
            PerFlowBackend::Nflog,
            "legacy var must win when both set"
        );
        unsafe {
            match prev_legacy {
                Some(v) => std::env::set_var(ENV_PER_FLOW_BACKEND, v),
                None => std::env::remove_var(ENV_PER_FLOW_BACKEND),
            }
            match prev_canonical {
                Some(v) => std::env::set_var(ENV_PER_FLOW_BACKEND_E7, v),
                None => std::env::remove_var(ENV_PER_FLOW_BACKEND_E7),
            }
        }
    }

    #[test]
    fn select_backend_unknown_value_defaults_to_nflog() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|p| p.into_inner());
        let prev_legacy = std::env::var(ENV_PER_FLOW_BACKEND).ok();
        let prev_canonical = std::env::var(ENV_PER_FLOW_BACKEND_E7).ok();
        unsafe {
            std::env::remove_var(ENV_PER_FLOW_BACKEND_E7);
            std::env::set_var(ENV_PER_FLOW_BACKEND, "xdp"); // typo / unknown
        }
        assert_eq!(select_backend_from_env(), PerFlowBackend::Nflog);
        unsafe {
            match prev_legacy {
                Some(v) => std::env::set_var(ENV_PER_FLOW_BACKEND, v),
                None => std::env::remove_var(ENV_PER_FLOW_BACKEND),
            }
            match prev_canonical {
                Some(v) => std::env::set_var(ENV_PER_FLOW_BACKEND_E7, v),
                None => std::env::remove_var(ENV_PER_FLOW_BACKEND_E7),
            }
        }
    }

    // ── ruleset augmentation ──────────────────────────────────────────
    #[test]
    fn augment_rewrites_accept_verdict() {
        let input = "    ip daddr 10.0.0.1 tcp dport 443 accept";
        let out = augment_ruleset_with_log_actions(input, 100);
        assert!(out.contains("log group 100 prefix \"cellos-flow accept\""));
        assert!(out.ends_with("accept"));
    }

    #[test]
    fn augment_rewrites_drop_verdict() {
        let input = "    udp dport 53 drop";
        let out = augment_ruleset_with_log_actions(input, 200);
        assert!(out.contains("log group 200 prefix \"cellos-flow drop\""));
        assert!(out.ends_with("drop"));
    }

    #[test]
    fn augment_skips_loopback_shortcut() {
        let input = "    oif \"lo\" accept";
        let out = augment_ruleset_with_log_actions(input, 100);
        assert_eq!(out, input);
    }

    #[test]
    fn augment_skips_policy_lines() {
        let input = "    type filter hook output priority 0; policy drop;";
        let out = augment_ruleset_with_log_actions(input, 100);
        assert_eq!(out, input);
    }

    #[test]
    fn augment_is_idempotent() {
        let input = "    ip daddr 10.0.0.1 tcp dport 443 accept";
        let once = augment_ruleset_with_log_actions(input, 100);
        let twice = augment_ruleset_with_log_actions(&once, 100);
        assert_eq!(once, twice);
    }

    #[test]
    fn augment_preserves_chain_structure() {
        let input = "table inet cellos_test {\n  chain output {\n    type filter hook output priority 0; policy drop;\n    oif \"lo\" accept\n    udp dport 53 drop\n  }\n}";
        let out = augment_ruleset_with_log_actions(input, 100);
        assert!(out.contains("table inet cellos_test {"));
        assert!(out.contains("chain output {"));
        assert!(out.contains("policy drop;"));
        assert!(out.contains("oif \"lo\" accept"));
        assert!(out.contains("log group 100 prefix \"cellos-flow drop\" drop"));
        assert!(out.ends_with("}"));
    }

    // ── L3/L4 attribution ─────────────────────────────────────────────
    #[test]
    fn decode_l4_ipv4_tcp_extracts_dst() {
        // Minimal IPv4 + TCP header. ihl=5 (20 bytes), proto=6, dst=10.0.0.1,
        // tcp dport=443.
        let mut p = vec![0u8; 40];
        p[0] = 0x45;
        p[9] = 6; // TCP
        p[16] = 10;
        p[17] = 0;
        p[18] = 0;
        p[19] = 1;
        p[22] = 0x01; // dst port high byte
        p[23] = 0xbb; // dst port low byte (443)
        let attr = decode_l3_l4_attribution(&p);
        assert_eq!(attr.dst_addr.as_deref(), Some("10.0.0.1"));
        assert_eq!(attr.dst_port, Some(443));
        assert_eq!(attr.protocol.as_deref(), Some("tcp"));
    }

    #[test]
    fn decode_l4_ipv4_udp_extracts_dst() {
        let mut p = vec![0u8; 40];
        p[0] = 0x45;
        p[9] = 17; // UDP
        p[16] = 1;
        p[17] = 1;
        p[18] = 1;
        p[19] = 1;
        p[22] = 0x00; // dport high
        p[23] = 0x35; // dport low (53)
        let attr = decode_l3_l4_attribution(&p);
        assert_eq!(attr.dst_addr.as_deref(), Some("1.1.1.1"));
        assert_eq!(attr.dst_port, Some(53));
        assert_eq!(attr.protocol.as_deref(), Some("udp"));
    }

    #[test]
    fn decode_l4_ipv4_icmp() {
        let mut p = vec![0u8; 28];
        p[0] = 0x45;
        p[9] = 1; // ICMP
        p[16] = 8;
        p[17] = 8;
        p[18] = 8;
        p[19] = 8;
        let attr = decode_l3_l4_attribution(&p);
        assert_eq!(attr.dst_addr.as_deref(), Some("8.8.8.8"));
        assert_eq!(attr.protocol.as_deref(), Some("icmp"));
        assert_eq!(attr.dst_port, None);
    }

    #[test]
    fn decode_l4_ipv6_tcp_extracts_dst() {
        let mut p = vec![0u8; 60];
        p[0] = 0x60; // version 6
        p[6] = 6; // next header = TCP
                  // dst = 2001:db8::1
        p[24] = 0x20;
        p[25] = 0x01;
        p[26] = 0x0d;
        p[27] = 0xb8;
        p[39] = 0x01;
        p[42] = 0x01;
        p[43] = 0xbb; // dport 443
        let attr = decode_l3_l4_attribution(&p);
        assert_eq!(attr.dst_addr.as_deref(), Some("2001:db8::1"));
        assert_eq!(attr.dst_port, Some(443));
        assert_eq!(attr.protocol.as_deref(), Some("tcp"));
    }

    #[test]
    fn decode_l4_ipv6_icmp6() {
        let mut p = vec![0u8; 48];
        p[0] = 0x60;
        p[6] = 58; // ICMPv6
        let attr = decode_l3_l4_attribution(&p);
        assert_eq!(attr.protocol.as_deref(), Some("icmp6"));
        assert_eq!(attr.dst_port, None);
    }

    #[test]
    fn decode_l4_handles_empty_payload() {
        let attr = decode_l3_l4_attribution(&[]);
        assert!(attr.dst_addr.is_none());
        assert!(attr.protocol.is_none());
    }

    // ── nflog datagram decode ─────────────────────────────────────────
    #[test]
    fn decode_nflog_extracts_prefix_and_payload() {
        // Synthesize a TLV stream with NFULA_PREFIX = "cellos-flow accept\0"
        // followed by NFULA_PAYLOAD = some bytes.
        let mut buf = Vec::new();
        let prefix_str = b"cellos-flow accept\0";
        let nla_len: u16 = 4 + prefix_str.len() as u16;
        buf.extend_from_slice(&nla_len.to_ne_bytes());
        buf.extend_from_slice(&NFULA_PREFIX.to_ne_bytes());
        buf.extend_from_slice(prefix_str);
        // align to 4
        while buf.len() % 4 != 0 {
            buf.push(0);
        }
        // payload TLV
        let pkt = b"abcdef";
        let nla_len2: u16 = 4 + pkt.len() as u16;
        buf.extend_from_slice(&nla_len2.to_ne_bytes());
        buf.extend_from_slice(&NFULA_PAYLOAD.to_ne_bytes());
        buf.extend_from_slice(pkt);
        while buf.len() % 4 != 0 {
            buf.push(0);
        }
        let decoded = decode_nflog_datagram(&buf).expect("decode ok");
        assert_eq!(decoded.prefix, "cellos-flow accept");
        assert_eq!(decoded.payload, pkt);
    }

    #[test]
    fn decode_nflog_round_trip_with_packet_attribution() {
        // Synthetic round-trip: build a TLV stream containing an IPv4+TCP
        // payload, decode, attribute, build a decision. End-to-end pure
        // exercise of the decoder.
        let prefix_str = b"cellos-flow drop\0";
        let mut packet = vec![0u8; 40];
        packet[0] = 0x45;
        packet[9] = 6; // TCP
        packet[16] = 192;
        packet[17] = 0;
        packet[18] = 2;
        packet[19] = 1;
        packet[22] = 0x01;
        packet[23] = 0xbb;
        let mut buf = Vec::new();
        let nla_len: u16 = 4 + prefix_str.len() as u16;
        buf.extend_from_slice(&nla_len.to_ne_bytes());
        buf.extend_from_slice(&NFULA_PREFIX.to_ne_bytes());
        buf.extend_from_slice(prefix_str);
        while buf.len() % 4 != 0 {
            buf.push(0);
        }
        let nla_len2: u16 = 4 + packet.len() as u16;
        buf.extend_from_slice(&nla_len2.to_ne_bytes());
        buf.extend_from_slice(&NFULA_PAYLOAD.to_ne_bytes());
        buf.extend_from_slice(&packet);
        while buf.len() % 4 != 0 {
            buf.push(0);
        }
        let decoded = decode_nflog_datagram(&buf).expect("decode ok");
        assert_eq!(decoded.prefix, "cellos-flow drop");
        let activation = sample_activation();
        let decision = build_decision(
            &activation,
            &decoded.prefix,
            &decoded.payload,
            "2026-01-01T00:00:00Z",
        );
        assert_eq!(decision.decision, NetworkFlowDecisionOutcome::Deny);
        assert_eq!(decision.reason_code, "nft_log_drop");
        assert_eq!(decision.dst_addr.as_deref(), Some("192.0.2.1"));
        assert_eq!(decision.dst_port, Some(443));
        assert_eq!(decision.protocol.as_deref(), Some("tcp"));
        assert_eq!(decision.cell_id, "cell-test");
        assert_eq!(decision.run_id, "run-uuid");
    }

    #[test]
    fn decode_nflog_errors_on_short_input() {
        let err = decode_nflog_datagram(&[0u8; 2]).expect_err("must error");
        assert!(matches!(err, NflogDecodeError::TooShort(_)));
    }

    #[test]
    fn decode_nflog_errors_on_missing_prefix() {
        // Only payload, no prefix attribute → MissingAttrs
        let pkt = vec![0u8; 8];
        let mut buf = Vec::new();
        let nla_len: u16 = 4 + pkt.len() as u16;
        buf.extend_from_slice(&nla_len.to_ne_bytes());
        buf.extend_from_slice(&NFULA_PAYLOAD.to_ne_bytes());
        buf.extend_from_slice(&pkt);
        let err = decode_nflog_datagram(&buf).expect_err("must error");
        assert!(matches!(err, NflogDecodeError::MissingAttrs));
    }

    // ── decision builder ──────────────────────────────────────────────
    #[test]
    fn build_decision_for_accept_yields_allow() {
        let activation = sample_activation();
        let payload = vec![0u8; 0];
        let d = build_decision(&activation, "cellos-flow accept", &payload, "now");
        assert_eq!(d.decision, NetworkFlowDecisionOutcome::Allow);
        assert_eq!(d.reason_code, "nft_log_accept");
    }

    #[test]
    fn build_decision_for_drop_yields_deny() {
        let activation = sample_activation();
        let payload = vec![0u8; 0];
        let d = build_decision(&activation, "cellos-flow drop", &payload, "now");
        assert_eq!(d.decision, NetworkFlowDecisionOutcome::Deny);
        assert_eq!(d.reason_code, "nft_log_drop");
    }

    #[test]
    fn build_decision_unknown_prefix_falls_back_to_deny() {
        let activation = sample_activation();
        let d = build_decision(&activation, "cellos-flow weirdo", &[], "now");
        assert_eq!(d.decision, NetworkFlowDecisionOutcome::Deny);
        assert_eq!(d.reason_code, "nft_log_unknown");
    }

    // ── env-alias activation (E7 / FC-38 Phase 2) ─────────────────────
    //
    // Both `CELLOS_FIRECRACKER_PER_FLOW_EBPF=1` (legacy) and
    // `CELLOS_PER_FLOW_REALTIME=1` (backend-neutral) opt the cell in. The
    // mutation guards below restore the prior ambient values so the tests
    // are order-independent — std's test harness runs unit tests in
    // parallel by default and unrelated tests in this module also touch
    // these vars.

    fn save_realtime_env() -> (Option<String>, Option<String>) {
        let prev_ebpf = std::env::var(ENV_PER_FLOW_EBPF).ok();
        let prev_rt = std::env::var(ENV_PER_FLOW_REALTIME).ok();
        // SAFETY: env mutation in tests is gated by harness isolation
        // discipline shared across this module.
        unsafe {
            std::env::remove_var(ENV_PER_FLOW_EBPF);
            std::env::remove_var(ENV_PER_FLOW_REALTIME);
        }
        (prev_ebpf, prev_rt)
    }

    fn restore_realtime_env(prev: (Option<String>, Option<String>)) {
        unsafe {
            match prev.0 {
                Some(v) => std::env::set_var(ENV_PER_FLOW_EBPF, v),
                None => std::env::remove_var(ENV_PER_FLOW_EBPF),
            }
            match prev.1 {
                Some(v) => std::env::set_var(ENV_PER_FLOW_REALTIME, v),
                None => std::env::remove_var(ENV_PER_FLOW_REALTIME),
            }
        }
    }

    #[test]
    fn build_activation_off_when_neither_env_var_set() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|p| p.into_inner());
        let prev = save_realtime_env();
        let act = build_activation_from_env("c", "r", None, None, None);
        assert!(act.is_none(), "no env vars → no activation");
        restore_realtime_env(prev);
    }

    #[test]
    fn build_activation_on_via_legacy_ebpf_env() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|p| p.into_inner());
        let prev = save_realtime_env();
        unsafe {
            std::env::set_var(ENV_PER_FLOW_EBPF, "1");
        }
        let act = build_activation_from_env("cell-x", "run-x", None, None, None);
        assert!(act.is_some(), "legacy _EBPF=1 must still opt in");
        restore_realtime_env(prev);
    }

    #[test]
    fn build_activation_on_via_realtime_alias() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|p| p.into_inner());
        let prev = save_realtime_env();
        unsafe {
            std::env::set_var(ENV_PER_FLOW_REALTIME, "1");
        }
        let act = build_activation_from_env("cell-x", "run-x", None, None, None);
        assert!(act.is_some(), "CELLOS_PER_FLOW_REALTIME=1 must opt in");
        let act = act.unwrap();
        assert_eq!(act.cell_id, "cell-x");
        assert_eq!(act.run_id, "run-x");
        assert_eq!(act.nflog_group, DEFAULT_NFLOG_GROUP);
        restore_realtime_env(prev);
    }

    #[test]
    fn build_activation_rejects_truthy_lookalikes_for_realtime() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|p| p.into_inner());
        let prev = save_realtime_env();
        for bad in ["true", "yes", "on", "TRUE", "0", "", "2"] {
            unsafe {
                std::env::set_var(ENV_PER_FLOW_REALTIME, bad);
            }
            let act = build_activation_from_env("c", "r", None, None, None);
            assert!(
                act.is_none(),
                "value {bad:?} must not enable per-flow realtime"
            );
        }
        restore_realtime_env(prev);
    }

    #[test]
    fn flow_event_payload_serialises_as_cloud_event_schema_compatible_json() {
        // E7 unit test: a FlowEvent / NetworkFlowDecision built via
        // build_decision serialises to JSON whose top-level fields match
        // the `network_flow_decision_data_v1` schema (cellId, runId,
        // decisionId, direction, decision, reasonCode, observedAt).
        let activation = sample_activation();
        let payload = vec![0u8; 0];
        let decision = build_decision(
            &activation,
            "cellos-flow accept",
            &payload,
            "2026-05-16T00:00:00Z",
        );
        let json = serde_json::to_value(&decision).expect("serialise");
        let obj = json.as_object().expect("object");
        for required in [
            "cellId",
            "runId",
            "decisionId",
            "direction",
            "decision",
            "reasonCode",
            "observedAt",
        ] {
            assert!(
                obj.contains_key(required),
                "missing required field {required}; payload={json}"
            );
        }
        assert_eq!(obj["direction"], "egress");
        assert_eq!(obj["decision"], "allow");
        assert_eq!(obj["reasonCode"], "nft_log_accept");
    }

    // ── L5-15 nflog → FlowAccumulator wiring ──────────────────────────────
    //
    // These tests exercise the *pure* parsing + record path so they run on
    // every platform (the listener `setns(2)` spawn is Linux-only and
    // covered by `tests/supervisor_per_flow_realtime.rs`). The contract
    // under test: when the nflog listener decodes a `cellos-flow accept`
    // datagram with a valid 5-tuple, the shared accumulator's
    // `unique_flow_count()` increments by one. Duplicate observations of
    // the same 5-tuple do NOT inflate the count — that's the homeostasis
    // semantics: "distinct connections," not "packet observations."

    /// Construct a minimal IPv4+TCP payload with caller-controlled 5-tuple
    /// bytes. Mirrors the synthetic packets used by the existing
    /// `decode_l4_ipv4_tcp_extracts_dst` test but adds src bytes so a
    /// FlowKey can be derived end-to-end.
    fn ipv4_tcp_payload(src: [u8; 4], src_port: u16, dst: [u8; 4], dst_port: u16) -> Vec<u8> {
        let mut p = vec![0u8; 40];
        p[0] = 0x45; // IPv4, ihl=5
        p[9] = 6; // TCP
        p[12..16].copy_from_slice(&src);
        p[16..20].copy_from_slice(&dst);
        // TCP header starts at byte 20 (ihl=20). Source port = bytes 20-21,
        // dest port = bytes 22-23, both network-byte-order.
        p[20..22].copy_from_slice(&src_port.to_be_bytes());
        p[22..24].copy_from_slice(&dst_port.to_be_bytes());
        p
    }

    #[test]
    fn nflog_populates_flow_accumulator() {
        // Parse a mock nflog accept event for a single 5-tuple and verify
        // the accumulator reports `unique_flow_count() == 1`.
        let payload = ipv4_tcp_payload([10, 0, 0, 5], 40000, [192, 0, 2, 1], 443);

        let attribution = decode_l3_l4_attribution(&payload);
        let key = flow_key_from_attribution(&attribution).expect("5-tuple fully populated");

        let accumulator = std::sync::Arc::new(std::sync::Mutex::new(
            crate::ebpf_flow::connection_tracking::FlowAccumulator::new(),
        ));
        record_opened_flow(&accumulator, key);

        let count = accumulator
            .lock()
            .expect("acquire lock")
            .unique_flow_count();
        assert_eq!(
            count, 1,
            "single nflog-derived 5-tuple must yield unique_flow_count() == 1"
        );
    }

    #[test]
    fn duplicate_nflog_entries_are_deduplicated() {
        // The same 5-tuple observed via two distinct nflog datagrams (e.g.
        // SYN + a retransmit, or an accept rule that fires twice for the
        // same connection's first and second packet) MUST count as one
        // exercised connection. The accumulator's dedup is keyed on
        // FlowKey, so the same parsed key inserted twice still yields a
        // count of 1.
        let payload = ipv4_tcp_payload([10, 0, 0, 5], 40000, [192, 0, 2, 1], 443);

        let attribution_a = decode_l3_l4_attribution(&payload);
        let key_a = flow_key_from_attribution(&attribution_a).expect("attribution a");

        // Re-parse to simulate a second datagram landing on the listener
        // thread — separate FlowKey value, same logical 5-tuple.
        let attribution_b = decode_l3_l4_attribution(&payload);
        let key_b = flow_key_from_attribution(&attribution_b).expect("attribution b");

        let accumulator = std::sync::Arc::new(std::sync::Mutex::new(
            crate::ebpf_flow::connection_tracking::FlowAccumulator::new(),
        ));
        record_opened_flow(&accumulator, key_a);
        record_opened_flow(&accumulator, key_b);

        let count = accumulator
            .lock()
            .expect("acquire lock")
            .unique_flow_count();
        assert_eq!(
            count, 1,
            "two nflog datagrams for the same 5-tuple must dedup to one connection"
        );
    }
}