#!/usr/bin/env bash
# EvoFirewall Linux sync agent — nft / ipset / iptables
set -Eeuo pipefail
umask 077

CONF_FILE=/etc/evofw/agent.conf
LOG_FILE=/var/log/evofw-firewall.log
STATE_DIR=/var/lib/evofw
HASH_FILE="${STATE_DIR}/last_hash"
POLICY_FILE="${STATE_DIR}/last_policy.json"
IP_HITS_TOP=200
PORT_HITS_TOP=500

log() { echo "$(date -u +%Y-%m-%dT%H:%M:%SZ) $*" | tee -a "$LOG_FILE"; }

if [[ ! -f "$CONF_FILE" ]]; then
  log "missing $CONF_FILE"
  exit 1
fi
# shellcheck disable=SC1090
source "$CONF_FILE"

: "${EVOFW_CP_URL:?}"
: "${CLIENT_TOKEN:?}"
CLIENT_TOKEN="${CLIENT_TOKEN//$'\r'/}"
CLIENT_TOKEN="${CLIENT_TOKEN//$'\n'/}"
BACKEND="${KERNEL_BACKEND:-auto}"
SYNC_SCRIPT=/usr/local/sbin/evofw-firewall.sh

mkdir -p "$STATE_DIR"
# One sync owns downloads, kernel state and committed hashes at a time.
command -v timeout >/dev/null 2>&1 || { log "need timeout (coreutils)"; exit 1; }
command -v flock >/dev/null 2>&1 || { log "need flock (util-linux)"; exit 1; }
exec 9>"${STATE_DIR}/sync.lock"
flock -n 9 || exit 0
RUN_DIR=$(mktemp -d "${STATE_DIR}/run.XXXXXX")
trap 'rm -rf -- "$RUN_DIR"' EXIT
# Python is required: JSON and network data must never be interpreted as shell.
command -v python3 >/dev/null 2>&1 || { log "need python3"; exit 1; }
# Bound every HTTP operation, including telemetry and self-update.
curl() { command curl --connect-timeout 10 --max-time 60 "$@"; }
# Keep xtables lock waits bounded.
iptables() { command timeout 60 iptables -w 10 "$@"; }
iptables-restore() { command timeout 60 iptables-restore "$@"; }
iptables-save() { command timeout 60 iptables-save "$@"; }
ip6tables() { command timeout 60 ip6tables -w 10 "$@"; }
ip6tables-restore() { command timeout 60 ip6tables-restore "$@"; }
ip6tables-save() { command timeout 60 ip6tables-save "$@"; }
nft() { command timeout 60 nft "$@"; }
ipset() { command timeout 60 ipset "$@"; }
on_sync_error() {
  local rc=$?
  trap - ERR
  log "sync failed rc=$rc; successful hash unchanged"
  if declare -F telemetry_worker >/dev/null 2>&1; then
    telemetry_worker enqueue error "${KERNEL_METHOD:-$BACKEND}" || true
    flush_reports || true
  fi
  exit "$rc"
}
trap on_sync_error ERR
CANDIDATE_FILE="${RUN_DIR}/candidate.json"
PORT_RULES_FILE="${RUN_DIR}/port_rules.json"

# Credentials are reconciled before any policy or telemetry request.
credential_sync() {
  python3 - "$STATE_DIR" "$CONF_FILE" "$EVOFW_CP_URL" "$CLIENT_TOKEN" "${CLIENT_ID:-}" <<'PY_CREDENTIALS'
import json, os, re, secrets, sys, urllib.request, urllib.error, uuid
from pathlib import Path
state_dir, conf, origin, token, client_id = sys.argv[1:]
origin = origin.rstrip('/')
pending_path = Path(state_dir) / 'token-rotation-pending.json'
def atomic(path, value):
    tmp = Path(str(path) + '.new')
    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, 'w') as out:
        out.write(value); out.flush(); os.fsync(out.fileno())
    os.replace(tmp, path)
    fd = os.open(str(Path(path).parent), os.O_RDONLY)
    try: os.fsync(fd)
    finally: os.close(fd)
def request(path, auth, body=None):
    data = None if body is None else json.dumps(body).encode()
    req = urllib.request.Request(origin + path, data=data, headers={'Authorization': 'Bearer ' + auth, 'Content-Type': 'application/json'})
    with urllib.request.urlopen(req, timeout=60) as response:
        return json.loads(response.read(16385))
try:
    if pending_path.exists():
        pending = json.loads(pending_path.read_text())
        if pending.get('origin') != origin or pending.get('client_id') != client_id: raise ValueError('credential origin mismatch')
    else:
        try: status = request('/v1/agent/credentials', token)
        except Exception:
            print(token); sys.exit(0) # Old server/pending approval: keep active protection.
        rotation = status.get('rotation')
        if not rotation:
            print(token); sys.exit(0)
        request_id = str(uuid.UUID(rotation['request_id']))
        pending = {'request_id': request_id, 'next_token': 'evofw_' + secrets.token_urlsafe(32), 'old_token': token, 'origin': origin, 'client_id': client_id}
        atomic(pending_path, json.dumps(pending))
    if not re.fullmatch(r'evofw_[A-Za-z0-9_-]{32,128}', pending['next_token']): raise ValueError('invalid pending token')
    try:
        ack = request('/v1/agent/token-rotation', pending['old_token'], {'request_id': pending['request_id'], 'next_token': pending['next_token']})
    except urllib.error.HTTPError as error:
        if error.code != 401: raise
        # The old-token replay TTL may have elapsed during an offline restart.
        # A successful normal authenticated read proves the saved candidate is active.
        try:
            status = request('/v1/agent/credentials', pending['next_token'])
            if status.get('client_id') != client_id or (status.get('completed_rotation') or {}).get('request_id') != pending['request_id']: raise ValueError('candidate identity mismatch')
            ack = {'ok': True, 'request_id': pending['request_id'], 'client_id': client_id}
        except urllib.error.HTTPError as proof_error:
            if proof_error.code != 401: raise
            # A successful old-token read proves no exchange committed. Only an
            # explicitly absent or superseding server request can abandon this attempt.
            status = request('/v1/agent/credentials', pending['old_token'])
            active_request = status.get('rotation')
            if status.get('client_id') != client_id or (active_request and active_request.get('request_id') == pending['request_id']): raise ValueError('exchange still pending')
            pending_path.unlink()
            fd = os.open(state_dir, os.O_RDONLY)
            try: os.fsync(fd)
            finally: os.close(fd)
            print(pending['old_token']); sys.exit(0) # Next sync reconciles a newer request.

    if ack.get('ok') is not True or ack.get('request_id') != pending['request_id'] or ack.get('client_id') != client_id: raise ValueError('mismatched credential ACK')
    text = Path(conf).read_text()
    text, count = re.subn(r'^CLIENT_TOKEN=.*$', "CLIENT_TOKEN='" + pending['next_token'] + "'", text, flags=re.M)
    if count != 1: raise ValueError('invalid agent config')
    atomic(Path(conf), text)
    pending_path.unlink()
    fd = os.open(state_dir, os.O_RDONLY)
    try: os.fsync(fd)
    finally: os.close(fd)
    print(pending['next_token'])
except Exception:
    # Never print tokens, response bodies, or exception details.
    sys.exit(1)
PY_CREDENTIALS
}

file_sha256() {
  local f=$1
  if command -v sha256sum >/dev/null 2>&1; then
    sha256sum "$f" 2>/dev/null | awk '{print $1}'
  elif command -v openssl >/dev/null 2>&1; then
    openssl dgst -sha256 "$f" 2>/dev/null | awk '{print $NF}'
  else
    echo ""
  fi
}

# Pull a newer sync script from CP before policy (pending agents too).
# Errors must not abort this run — keep the current script.
maybe_self_update() {
  if [[ "${EVOFW_SKIP_SELF_UPDATE:-}" == "1" ]]; then
    return 0
  fi
  if [[ ! -f "$SYNC_SCRIPT" ]]; then
    return 0
  fi
  local local_sha tmp code remote_sha
  local_sha=$(file_sha256 "$SYNC_SCRIPT")
  tmp=$(mktemp "${STATE_DIR}/sync-script.XXXXXX") || return 0
  if [[ -n "$local_sha" ]]; then
    code=$(curl -sS -o "$tmp" -w '%{http_code}' \
      -H "If-None-Match: \"${local_sha}\"" \
      "${EVOFW_CP_URL%/}/v1/agent/sync-script") || code="000"
  else
    code=$(curl -sS -o "$tmp" -w '%{http_code}' \
      "${EVOFW_CP_URL%/}/v1/agent/sync-script") || code="000"
  fi
  if [[ "$code" == "304" ]]; then
    rm -f "$tmp"
    return 0
  fi
  if [[ "$code" != "200" ]]; then
    log "self-update: sync-script HTTP ${code} — keep current"
    rm -f "$tmp"
    return 0
  fi
  if ! head -n1 "$tmp" | grep -q '^#!'; then
    log "self-update: sync-script is not a shell script — keep current"
    rm -f "$tmp"
    return 0
  fi
  remote_sha=$(file_sha256 "$tmp")
  if [[ -z "$remote_sha" ]]; then
    log "self-update: cannot hash download — keep current"
    rm -f "$tmp"
    return 0
  fi
  if [[ -n "$local_sha" && "$remote_sha" == "$local_sha" ]]; then
    rm -f "$tmp"
    return 0
  fi
  if ! bash -n "$tmp" || ! install -m 755 "$tmp" "${SYNC_SCRIPT}.new" || ! mv -f "${SYNC_SCRIPT}.new" "$SYNC_SCRIPT"; then
    log "self-update: install failed — keep current"
    rm -f "$tmp"
    return 0
  fi
  rm -f "$tmp"
  rm -f "${STATE_DIR}/kernel_state"
  log "self-update: installed script sha256=${remote_sha} — re-exec"
  rm -rf -- "$RUN_DIR"
  flock -u 9
  exec 9>&-
  exec env EVOFW_SKIP_SELF_UPDATE=1 "$SYNC_SCRIPT" || {
    log "self-update: exec failed — continue current"
    return 0
  }
}

# State and immutable outbox share one fsync+rename commit. Sequence allocation
# can never get ahead of a durable report, and retries reuse the original bytes.
telemetry_worker() {
  APPLY_ERROR_CODE="${APPLY_ERROR_CODE:-apply_failed}" python3 - "$STATE_DIR" "$RUN_DIR" "$@" <<'PY'
import datetime, hashlib, ipaddress, json, os, pathlib, re, shlex, sys, time, uuid
root, run, op = pathlib.Path(sys.argv[1]), pathlib.Path(sys.argv[2]), sys.argv[3]
MAX = 9007199254740991
state_path = root / "telemetry-state.json"
def atomic(path, value):
    tmp = path.with_suffix(path.suffix + ".new")
    with open(tmp, "w", encoding="utf-8", newline="\n") as f:
        json.dump(value, f, separators=(",", ":"), sort_keys=True)
        f.flush()
        os.fsync(f.fileno())
    os.replace(tmp, path)
    if os.name != "nt":
        fd = os.open(str(path.parent), os.O_RDONLY)
        try: os.fsync(fd)
        finally: os.close(fd)
def load(path, default=None, strict=False):
    try: return json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError: return default
    except (ValueError, UnicodeError):
        if strict: raise
        return default
def epochs(): return {k: str(uuid.uuid4()) for k in ("traffic", "ip", "port")}
def current_boot():
    override = os.environ.get("EVOFW_TEST_BOOT_ID")
    if override: return override
    try: return pathlib.Path("/proc/sys/kernel/random/boot_id").read_text().strip()
    except OSError: return "unknown-" + str(uuid.uuid4())
state = load(state_path, strict=True)
if state is None:
    state = {"version": 2, "next_sequence": 0, "pending": [], "epochs": epochs(), "lost_reports": 0, "last_parts": {}}
if state.get("version") != 2 or type(state.get("next_sequence")) is not int or not isinstance(state.get("pending"), list):
    raise ValueError("invalid durable telemetry state; manual recovery required")
if op == "assert-sequence":
    if state.get("sequence_exhausted") or not 0 <= state["next_sequence"] <= MAX: raise ValueError("sequence exhausted; re-enrollment required")
elif op == "init":
    boot = current_boot()
    if state.get("boot_id") != boot:
        state.update(boot_id=boot, epochs=epochs(), last_parts={})
    atomic(state_path, state)
elif op == "cache-request":
    policy = load(root / "last_policy.json", {})
    etag, body = policy.get("_response_etag"), policy.get("_response_body")
    if not isinstance(etag, str) or not re.fullmatch(r'"[A-Za-z0-9:_-]{1,128}"', etag) or not isinstance(body, str) or len(body.encode()) > 8388608: sys.exit(1)
    wire = json.loads(body)
    if wire.get("hash") != policy.get("hash") or wire.get("generation") != policy.get("generation"): sys.exit(1)
    (run / "cache-etag").write_text(etag)
    (run / "cached-policy.json").write_text(body, encoding="utf-8", newline="\n")
elif op == "commit-policy":
    policy = load(run / "normalized.json", strict=True)
    if not isinstance(policy, dict): raise ValueError("missing verified policy")
    previous = load(root / "last_policy.json", {})
    policy.pop("_response_etag", None)
    policy.pop("_response_body", None)
    status_file = run / "policy-http-status"
    status = status_file.read_text().strip() if status_file.exists() else ""
    if status == "200":
        headers = run / "policy-headers.txt"
        etags = re.findall(r'^etag:\s*(.*?)\s*$', headers.read_text(errors="replace"), re.I | re.M) if headers.exists() else []
        body_file = run / "candidate.json"
        if etags and re.fullmatch(r'"[A-Za-z0-9:_-]{1,128}"', etags[-1]) and body_file.exists():
            body = body_file.read_text(encoding="utf-8")
            wire = json.loads(body)
            if len(body.encode()) <= 8388608 and wire.get("hash") == policy.get("hash") and wire.get("generation") == policy.get("generation"):
                policy.update(_response_etag=etags[-1], _response_body=body)
    elif previous.get("hash") == policy.get("hash") and previous.get("generation") == policy.get("generation"):
        for key in ("_response_etag", "_response_body"):
            if key in previous: policy[key] = previous[key]
    # The matching validator and exact received JSON commit together, only
    # after kernel apply/verification. A failed candidate cannot replace them.
    atomic(root / "last_policy.json", policy)
elif op == "host-due":
    f = root / "host_firewall.json"
    sys.exit(0 if not f.exists() or time.time() - f.stat().st_mtime >= 300 else 1)
elif op == "rotate":
    state.update(epochs=epochs(), last_parts={})
    if len(sys.argv) > 4: state["active_families"] = sys.argv[4].split(",")
    atomic(state_path, state)
elif op == "peek":
    if not state["pending"]: sys.exit(1)
    # The exact committed JSON payload is sent again after any network failure.
    (run / "pending.json").write_text(state["pending"][0], encoding="utf-8")
elif op == "ack":
    ack = load(run / "report-ack.json", {})
    if not state["pending"]: sys.exit(1)
    expected = json.loads(state["pending"][0])
    accepted = ack.get("accepted") is True or ack.get("duplicate") is True or ack.get("reason") == "stale_sequence"
    if ack.get("ok") is not True or not accepted or ack.get("report_id") != expected["report_id"] or ack.get("sequence") != expected["sequence"]:
        sys.exit(1)
    if ack.get("reason") == "stale_sequence":
        floor = ack.get("sequence_floor")
        if type(floor) is not int or not expected["sequence"] <= floor <= MAX: sys.exit(1)
        if floor == MAX:
            state["sequence_exhausted"] = True
            atomic(state_path, state)
            sys.exit(1)
        if floor + 1 > state["next_sequence"]:
            state.update(next_sequence=floor + 1, epochs=epochs(), last_parts={}, coverage_unknown=True)
    state["pending"].pop(0)
    atomic(state_path, state)
elif op == "capture":
    method = sys.argv[4]
    unsupported, parts = set(), {"traffic": {}, "ip": {}, "port": {}}
    out = {"kernel_method": method, "capabilities": ["telemetry-v2", "policy-ipv4"], "quality": {"partial": False, "truncated": False, "unsupported": [], "lost_reports": 0}}
    def safe(n, metric):
        if type(n) is int and 0 <= n <= MAX: return n
        unsupported.add(metric + "_counter_unsafe")
        return None
    def details(kind, values, limit):
        values.sort(key=lambda x: (-x["packets"], json.dumps(x, sort_keys=True)))
        out[kind + "_hits"] = values[:limit]
        out.setdefault("detail_counts", {}).update({kind + "_found": len(values), kind + "_sent": min(len(values), limit)})
        if len(values) > limit: out["quality"]["truncated"] = True
    dropped, accepted, traffic_seen = 0, 0, False
    traffic_bad = False
    ip_values, port_values = [], []
    if method == "nft":
        data = load(run / "metrics-nft.json")
        if not isinstance(data, dict) or not isinstance(data.get("nftables"), list):
            unsupported.add("nft_json_unavailable")
        else:
            out["capabilities"] += ["nft-atomic-apply", "policy-port-acl"]
            names = {obj.get("set", {}).get("name") for obj in data["nftables"]}
            if {"deny_v6", "allow_v6"}.issubset(names): out["capabilities"] += ["policy-ipv6", "policy-port-acl-ipv6"]
            ip_available = port_available = False
            for item in data["nftables"]:
                r = item.get("rule", {})
                if r.get("family") == "inet" and r.get("table") == "evofw":
                    comment, chain, exprs = r.get("comment", ""), r.get("chain"), r.get("expr", [])
                    if not (comment.startswith("evofw-v2/") or comment.startswith("evofw-port-")):
                        unsupported.add("legacy_rule_ids")
                        traffic_bad = True
                        continue
                    verdict = next((k for e in exprs for k in ("drop", "accept") if k in e), None)
                    # accept at prerouting is not a final input/forward decision.
                    if verdict is None or (chain == "prerouting" and verdict == "accept"): continue
                    if chain not in ("input", "forward", "prerouting"): continue
                    counter = next((e["counter"] for e in exprs if isinstance(e.get("counter"), dict)), {})
                    n = safe(counter.get("packets"), "traffic")
                    if n is None: traffic_bad = True; continue
                    traffic_seen = True
                    key = chain + "/" + comment
                    if key in parts["traffic"]: traffic_bad = True; unsupported.add("duplicate_rule_ids"); continue
                    parts["traffic"][key] = n
                    if verdict == "drop": dropped += n
                    else: accepted += n
                obj = item.get("set", item.get("element", {}))
                if obj.get("family") != "inet" or obj.get("table") != "evofw": continue
                name = obj.get("name")
                if name not in ("deny_v4", "deny_v6", "deny_port_hits", "deny_port_hits_v4", "deny_port_hits_v6"): continue
                elements = obj.get("elem", [])
                if not isinstance(elements, list): elements = [elements]
                if name.startswith("deny_port_hits"):
                    port_available = True
                    if name == "deny_port_hits_v6": out["capabilities"].append("port-counters-ipv6")
                # A set without per-element counters is valid for enforcement,
                # but it cannot claim measured per-IP packets.
                if name in ("deny_v4", "deny_v6") and obj.get("counter") is True: ip_available = True
                for raw in elements:
                    elem = raw.get("elem", raw) if isinstance(raw, dict) else {}
                    if not isinstance(elem, dict): unsupported.add("ip_counters_unavailable"); continue
                    n = safe((elem.get("counter") or {}).get("packets"), "ip" if name in ("deny_v4", "deny_v6") else "port")
                    if n is None: continue
                    val = elem.get("val")
                    if name in ("deny_v4", "deny_v6"):
                        ip_available = True
                        if name == "deny_v6" and "ip-counters-ipv6" not in out["capabilities"]: out["capabilities"].append("ip-counters-ipv6")
                        if isinstance(val, dict) and "prefix" in val:
                            prefix = val["prefix"]
                            if prefix.get("len") != (128 if name == "deny_v6" else 32):
                                unsupported.add("ip_prefix_counters"); continue
                            val = prefix.get("addr")
                        try: address = str(ipaddress.ip_address(val))
                        except (ValueError, TypeError, ipaddress.AddressValueError): unsupported.add("ip_element_format"); continue
                        parts["ip"][address] = n
                        if n: ip_values.append({"ip": address, "packets": n})
                    else:
                        concat = val.get("concat") if isinstance(val, dict) else None
                        if not isinstance(concat, list) or len(concat) != 3: unsupported.add("port_element_format"); continue
                        address, proto, port = concat
                        proto = {6: "tcp", 17: "udp", "tcp": "tcp", "udp": "udp"}.get(proto)
                        try: address = str(ipaddress.ip_address(address))
                        except (ValueError, TypeError): unsupported.add("port_element_format"); continue
                        if proto is None or type(port) is not int or not 1 <= port <= 65535: unsupported.add("port_element_format"); continue
                        key = address + ":" + proto + ":" + str(port)
                        parts["port"][key] = n
                        if n: port_values.append({"ip": address, "port": port, "protocol": proto, "packets": n})
            if ip_available: details("ip", ip_values, 200); out["capabilities"].append("ip-counters")
            else: unsupported.add("ip_counters_unavailable")
            if port_available: details("port", port_values, 500); out["capabilities"].append("port-counters")
            else: unsupported.add("port_counters_unavailable")
    elif method == "ipset":
        out["capabilities"].append("ipset-staged-apply")
        if (run / "metrics-iptables-v6.txt").exists(): out["capabilities"].append("policy-ipv6")
        expected = set(state.get("active_families", ["4"]))
        hint = run / "metrics-expected-families"
        if hint.exists(): expected.update(hint.read_text().strip().split(","))
        last_policy = load(root / "last_policy.json", {})
        if any(":" in c for c in last_policy.get("deny_cidrs", []) + last_policy.get("allow_cidrs", [])): expected.add("6")
        traffic_files = [(run / "metrics-iptables.txt", "v4"), (run / "metrics-iptables-v6.txt", "v6")]
        for path, family in traffic_files:
          if path.exists():
            for line in path.read_text().splitlines():
                m = re.match(r"^\[([0-9]+):[0-9]+\] (-A EVOFW_INPUT .*)$", line)
                if not m: continue
                tokens = shlex.split(m[2])
                if "-j" not in tokens: continue
                verdict = tokens[tokens.index("-j") + 1]
                if verdict not in ("DROP", "ACCEPT"): continue
                n = safe(int(m[1]), "traffic")
                if n is None: traffic_bad = True; continue
                parts["traffic"][family + "/" + m[2]] = n
                traffic_seen = True
                if verdict == "DROP": dropped += n
                else: accepted += n
          elif family[1:] in expected:
            traffic_bad = True
            unsupported.add("iptables_counters_unavailable" if family == "v4" else "ip6tables_counters_unavailable")
        path = run / "metrics-ipset.txt"
        ip_available = False
        if path.exists():
            for line in path.read_text().splitlines():
                tokens = shlex.split(line)
                if not tokens or tokens[0] != "add" or len(tokens) < 3: continue
                if "packets" not in tokens: continue
                ip_available = True
                n = safe(int(tokens[tokens.index("packets") + 1]), "ip")
                if n is None: continue
                net = ipaddress.ip_network(tokens[2], strict=False)
                if net.prefixlen != net.max_prefixlen: unsupported.add("ip_prefix_counters"); continue
                if net.version == 6 and "ip-counters-ipv6" not in out["capabilities"]: out["capabilities"].append("ip-counters-ipv6")
                address = str(net.network_address)
                parts["ip"][address] = n
                if n: ip_values.append({"ip": address, "packets": n})
        if ip_available: details("ip", ip_values, 200); out["capabilities"].append("ip-counters")
        else: unsupported.add("ip_counters_unavailable")
        unsupported.add("port_counters_unavailable")
    else: unsupported.add("kernel_counters_unavailable")
    if traffic_seen and not traffic_bad and safe(dropped, "traffic") is not None and safe(accepted, "traffic") is not None:
        out.update(packets_dropped=dropped, packets_accepted=accepted)
        out["capabilities"].append("traffic-counters")
    else: unsupported.add("traffic_counters_unavailable")
    if state.get("boot_id", "").startswith("unknown-"): unsupported.add("boot_id_unavailable")
    out["quality"].update(partial=bool(unsupported) or out["quality"]["truncated"], unsupported=sorted(unsupported))
    out["parts"] = parts
    atomic(run / "snapshot.json", out)
elif op == "enqueue":
    if state.get("sequence_exhausted"): raise ValueError("sequence exhausted; re-enrollment required")
    if len(state["pending"]) >= 32:
        state["lost_reports"] = min(MAX, state["lost_reports"] + 1)
        atomic(state_path, state)
        raise RuntimeError("telemetry outbox is full; apply is blocked")
    sequence = state["next_sequence"]
    if not 0 <= sequence <= MAX: raise ValueError("telemetry sequence exhausted")
    snap = load(run / "snapshot.json", {"quality": {"partial": True, "truncated": False, "unsupported": ["snapshot_unavailable"], "lost_reports": 0}, "capabilities": ["telemetry-v2"]})
    parts = snap.pop("parts", {})
    # A partial family observation is not a new aggregate baseline. Keep the
    # complete prior per-rule map so a reset hidden by other-family growth is
    # still detected when collection recovers.
    required_fields = {"traffic": ("packets_dropped", "packets_accepted"), "ip": ("ip_hits",), "port": ("port_hits",)}
    for kind, fields in required_fields.items():
        if any(field not in snap for field in fields): parts[kind] = state.get("last_parts", {}).get(kind, {})
    # Detect explicit counter resets even when the desired policy is unchanged.
    for kind, values in parts.items():
        previous = state.get("last_parts", {}).get(kind, {})
        if any(key in previous and value < previous[key] for key, value in values.items()): state["epochs"][kind] = str(uuid.uuid4())
    now = int(time.time() * 1000)
    report = dict(snap)
    report.update(telemetry_version=2, report_id=str(uuid.uuid4()), sequence=sequence,
        boot_id=state.get("boot_id", "unknown"), observed_at=datetime.datetime.fromtimestamp(now / 1000, datetime.timezone.utc).isoformat(timespec="milliseconds"),
        interval_ms=max(1, now - state.get("last_observed_ms", now - 60000)), counter_epoch=state["epochs"].copy(),
        agent_version="linux/10", status=sys.argv[4], kernel_method=sys.argv[5], source="agent")
    report["quality"]["lost_reports"] = state["lost_reports"]
    if state.get("coverage_unknown"):
        report["quality"]["partial"] = True
        report["quality"]["unsupported"].append("sequence_rollback_coverage_unknown")
    policy = load(root / "last_policy.json", {})
    h = policy.get("hash", "")
    if isinstance(h, str) and re.fullmatch(r"sha256:[0-9a-f]{64}", h): report["applied_hash"] = h
    revision = policy.get("generation")
    if type(revision) is int and 0 <= revision <= MAX: report["applied_revision"] = revision
    if policy: report["prefix_count"] = len(policy.get("deny_cidrs", [])) + len(policy.get("allow_cidrs", []))
    host = load(root / "host_firewall.json")
    if isinstance(host, dict): report["host_firewall"] = host
    else:
        report["quality"]["partial"] = True
        if "host_firewall_unavailable" not in report["quality"]["unsupported"]: report["quality"]["unsupported"].append("host_firewall_unavailable")
    if report["status"] != "ok": report["apply_error"] = {"code": os.environ.get("APPLY_ERROR_CODE", "apply_failed"), "message": "Policy sync or apply failed; last successful hash retained", "stage": "agent", "retryable": True}
    state["pending"].append(json.dumps(report, separators=(",", ":"), sort_keys=True))
    state.update(next_sequence=sequence + 1, last_observed_ms=now, last_parts=parts, lost_reports=0, coverage_unknown=False)
    atomic(state_path, state)
else: raise ValueError("unknown telemetry operation")
PY
}

flush_reports() {
  local attempt code
  for attempt in 1 2 3 4; do
    telemetry_worker peek || return 0
    code=$(curl -sS -o "$RUN_DIR/report-ack.json" -w '%{http_code}' -X POST "${EVOFW_CP_URL%/}/v1/agent/apply-report" \
      -H "Authorization: Bearer ${CLIENT_TOKEN}" -H 'Content-Type: application/json' \
      --data-binary @"$RUN_DIR/pending.json") || code=000
    if [[ "$code" != 200 ]] || ! telemetry_worker ack; then
      log "telemetry pending; ACK not confirmed http=$code"
      return 0
    fi
  done
}

telemetry_worker init
flush_reports
telemetry_worker assert-sequence
if ! ACTIVE_TOKEN=$(credential_sync); then log "credential recovery pending; defer sync"; exit 1; fi
CLIENT_TOKEN="$ACTIVE_TOKEN"
maybe_self_update

curl_policy() {
  local dest="$1"
  local code curl_rc=0 validator=()
  if telemetry_worker cache-request; then validator=(-H "If-None-Match: $(cat "$RUN_DIR/cache-etag")"); fi
  code=$(ulimit -f 8192; curl --max-filesize 8388608 -sS -D "$RUN_DIR/policy-headers.txt" "${validator[@]}" -o "$dest" -w "%{http_code}" \
    -H "Authorization: Bearer ${CLIENT_TOKEN}" \
    -H "Accept: application/json" \
    "${EVOFW_CP_URL%/}/v1/agent/policy") || curl_rc=$?
  if [[ "$curl_rc" == 63 || "$curl_rc" == 153 ]]; then
    APPLY_ERROR_CODE=POLICY_BUDGET_EXCEEDED
    log "policy download exceeds 8MiB budget"
    return 1
  fi
  [[ "$curl_rc" == 0 ]] || return 1
  if [[ "$code" == "304" ]]; then
    [[ -f "$RUN_DIR/cached-policy.json" && -f "$RUN_DIR/cache-etag" ]] || { log "policy304 without validated cache"; return 1; }
    cp "$RUN_DIR/cached-policy.json" "$dest"
    printf '%s' 304 >"$RUN_DIR/policy-http-status"
    return 0
  fi
  if [[ "$code" == "403" ]]; then
    log "pending approval"
    return 2
  fi
  if [[ "$code" != "200" ]]; then
    local error_code
    error_code=$(python3 - "$dest" <<'PY'
import json, pathlib, sys
f = pathlib.Path(sys.argv[1])
try:
    body = json.loads(f.read_text()) if f.exists() and f.stat().st_size <= 65536 else {}
    code = (body.get("error") or {}).get("code")
    if code in ("POLICY_BUDGET_EXCEEDED", "UNSUPPORTED_CAPABILITY"): print(code)
except (ValueError, TypeError, AttributeError): pass
PY
) || error_code=""
    if [[ -n "$error_code" ]]; then APPLY_ERROR_CODE=$error_code; fi
    log "policy HTTP $code"
    return 1
  fi
  printf '%s' 200 >"$RUN_DIR/policy-http-status"
  return 0
}

# Do not use `if ! cmd; rc=$?` — after `!`, $? is 0, not the real status.
policy_rc=0
curl_policy "$CANDIDATE_FILE" || policy_rc=$?
if [[ "$policy_rc" -eq 2 ]]; then
  exit 0
fi
if [[ "$policy_rc" -ne 0 ]]; then
  false
fi

parse_policy() {
  local f="$1"
  # NUL-free validated values go to files. No eval, process substitution errors,
  # or partial shell assignments can turn a rejected policy into an apply.
  if python3 - "$f" "$RUN_DIR" <<'PY'
import ipaddress, json, pathlib, re, sys
out = pathlib.Path(sys.argv[2])
def budget(message):
    (out / "apply-error-code").write_text("POLICY_BUDGET_EXCEEDED")
    raise ValueError(message)
if pathlib.Path(sys.argv[1]).stat().st_size > 8388608: budget("policy exceeds 8MiB")
p = json.load(open(sys.argv[1], encoding="utf-8"))
if not isinstance(p, dict):
    raise ValueError("policy must be an object")
revision = p.get("generation")
if revision is not None and (type(revision) is not int or not 0 <= revision <= 9007199254740991): raise ValueError("invalid generation")
raw_rules = p.get("port_rules", [])
if not isinstance(raw_rules, list): raise ValueError("port_rules must be an array")
if len(raw_rules) > 1000: budget("more than 1000 serialized port rules")
raw_bags = [p.get("deny_cidrs", []), p.get("allow_cidrs", [])]
for rule in raw_rules:
    if not isinstance(rule, dict): raise ValueError("invalid port rule")
    raw_bags.append(rule.get("src_cidrs", []))
raw_count = 0
for bag in raw_bags:
    if not isinstance(bag, list): raise ValueError("CIDRs must be an array")
    raw_count += len(bag)
    if raw_count > 200000: budget("more than 200000 raw network memberships")
h = p.get("hash")
if not isinstance(h, str) or not re.fullmatch(r"sha256:[0-9a-f]{64}", h):
    raise ValueError("invalid policy hash")
da = p.get("default_action")
if da is None:
    if p.get("policy_mode", "blacklist") not in ("blacklist", "whitelist"):
        raise ValueError("invalid legacy policy mode")
    da = "drop" if p.get("policy_mode") == "whitelist" else "accept"
if da not in ("accept", "drop"):
    raise ValueError("invalid default action")
effective_count = 0
def networks(items):
    global effective_count
    if not isinstance(items, list):
        raise ValueError("CIDRs must be an array")
    values = []
    for value in items:
        if not isinstance(value, str) or len(value) > 128 or "%" in value:
            raise ValueError("CIDR must be unscoped IP text of at most 128 characters")
        net = ipaddress.ip_network(value, strict=False)
        values.append(net)
    values = set(values)
    # Match the server budget before optional kernel compression: adjacent or
    # overlapping distinct CIDRs remain distinct memberships in this budget.
    effective_count += len(values)
    if effective_count > 50000: budget("more than 50000 effective network memberships")
    return [str(n) for family in (4, 6) for n in ipaddress.collapse_addresses(v for v in values if v.version == family)]
p["deny_cidrs"] = networks(p.get("deny_cidrs", []))
p["allow_cidrs"] = networks(p.get("allow_cidrs", []))
rules = p.get("port_rules", [])
if not isinstance(rules, list):
    raise ValueError("port_rules must be an array")
ids = set()
for r in rules:
    if not isinstance(r, dict) or r.get("action") not in ("open", "close") or r.get("protocol") not in ("tcp", "udp"):
        raise ValueError("invalid port rule")
    if not isinstance(r.get("id"), str) or not r["id"] or len(r["id"]) > 128 or (r["id"], r["protocol"]) in ids:
        raise ValueError("invalid or duplicate port rule id")
    ids.add((r["id"], r["protocol"]))
    start, end = r.get("port_start"), r.get("port_end", r.get("port_start"))
    if type(start) is not int or type(end) is not int or not 1 <= start <= end <= 65535:
        raise ValueError("invalid port range")
    r["port_end"] = end
    r["src_cidrs"] = networks(r.get("src_cidrs", []))
p["default_action"] = da
# Write only after the entire document has passed validation.
(out / "hash").write_text(h + "\n", newline="\n")
(out / "default_action").write_text(da + "\n", newline="\n")
for kind in ("deny", "allow"):
    (out / kind).write_text("".join(c + "\n" for c in p[kind + "_cidrs"]), newline="\n")
(out / "port_rules.json").write_text(json.dumps(rules))
(out / "ipv6_required").write_text("1" if any(":" in c for c in p["deny_cidrs"] + p["allow_cidrs"] + [c for r in rules for c in r["src_cidrs"]]) else "0")
(out / "normalized.json").write_text(json.dumps(p))
PY
  then :; else
    if [[ -f "$RUN_DIR/apply-error-code" ]]; then APPLY_ERROR_CODE=$(cat "$RUN_DIR/apply-error-code"); fi
    return 1
  fi
  HASH=$(cat "$RUN_DIR/hash")
  DEFAULT_ACTION=$(cat "$RUN_DIR/default_action")
  mapfile -t DENY <"$RUN_DIR/deny"
  mapfile -t ALLOW <"$RUN_DIR/allow"
}

HASH=""; DEFAULT_ACTION=accept; DENY=(); ALLOW=()
parse_policy "$CANDIDATE_FILE"
PORT_RULES_COUNT=$(python3 -c 'import json,sys; print(len(json.load(open(sys.argv[1]))))' "$PORT_RULES_FILE")
log "default_action=$DEFAULT_ACTION deny=${#DENY[@]} allow=${#ALLOW[@]} port_rules=$PORT_RULES_COUNT hash=$HASH"

PACKETS_DROPPED=0
PACKETS_ACCEPTED=0
KERNEL_METHOD=""
APPLIED=0
# Snapshot / hits live on disk — never via bash ${var:-{...}} (} truncates expansion).
HOST_FW_FILE="${STATE_DIR}/host_firewall.json"
IP_HITS_FILE="${STATE_DIR}/ip_hits.json"
PORT_HITS_FILE="${STATE_DIR}/port_hits.json"
# 1 when deny_port_hits dynamic set is available for this apply.
PORT_HITS_ENABLED=0

# Emit nft language, never shell commands. The entire owned table is replaced
# in one checked kernel transaction; no live delete/flush precedes validation.
render_nft_batch() {
  local counters=$1 hits=$2 exists=$3
  python3 - "$RUN_DIR/normalized.json" "$counters" "$hits" "$exists" <<'PY'
import hashlib, ipaddress, json, sys
p = json.load(open(sys.argv[1]))
counters, hits, exists = (v == "1" for v in sys.argv[2:])
if exists: print("delete table inet evofw")
print("add table inet evofw")
def family(cidrs, version): return [c for c in cidrs if ipaddress.ip_network(c).version == version]
def emit_set(name, cidrs, version):
    suffix = " counter;" if counters else ""
    print(f"add set inet evofw {name} {{ type ipv{version}_addr; flags interval;{suffix} }}")
    for i in range(0, len(cidrs), 64): print(f"add element inet evofw {name} {{ {', '.join(cidrs[i:i+64])} }}")
for version in (4, 6):
    emit_set(f"deny_v{version}", family(p["deny_cidrs"], version), version)
    emit_set(f"allow_v{version}", family(p["allow_cidrs"], version), version)
    if hits: print(f"add set inet evofw deny_port_hits_v{version} {{ type ipv{version}_addr . inet_proto . inet_service; flags dynamic,timeout; timeout 1h; size 65536; counter; }}")
rules = []
for i, r in enumerate(p["port_rules"]):
    for version in (4, 6):
        sources = family(r["src_cidrs"], version)
        if not sources: continue
        all_sources = ("0.0.0.0/0" if version == 4 else "::/0") in sources
        rules.append((i, r, version, all_sources))
        if not all_sources: emit_set(f"port_src_{i}_v{version}", sources, version)
for chain in ("input", "forward", "prerouting"):
    priority = "mangle" if chain == "prerouting" else "0"
    policy = "accept" if chain == "prerouting" else p["default_action"]
    print(f"add chain inet evofw {chain} {{ type filter hook {chain} priority {priority}; policy {policy}; }}")
    serial = 0
    def rule(expression):
        global serial
        serial += 1
        if " comment " not in expression: expression += f' comment "evofw-v2/{chain}/{serial}"'
        print(f"add rule inet evofw {chain} {expression}")
    rule("ct state established,related counter accept")
    if chain == "input": rule('iif "lo" counter accept')
    for version in (4, 6):
        selector = "ip" if version == 4 else "ip6"
        if hits: rule(f"meta l4proto {{ tcp, udp }} {selector} saddr @deny_v{version} update @deny_port_hits_v{version} {{ {selector} saddr . meta l4proto . th dport }} counter drop")
        rule(f"{selector} saddr @deny_v{version} counter drop")
    scope = "fib daddr type local " if chain == "prerouting" else ("ct status dnat " if chain == "forward" else "")
    ordered = [r for r in rules if r[1]["action"] == "close"] + [r for r in rules if r[1]["action"] == "open"]
    for i, r, version, all_sources in ordered:
        selector = "ip" if version == 4 else "ip6"
        source = f"meta nfproto ipv{version} " if all_sources else f"{selector} saddr @port_src_{i}_v{version} "
        port = str(r["port_start"]) if r["port_start"] == r["port_end"] else f"{r['port_start']}-{r['port_end']}"
        verdict = "drop" if r["action"] == "close" else "accept"
        comment = "evofw-port-" + hashlib.sha256(r["id"].encode()).hexdigest() + f"-{r['protocol']}-v{version}"
        rule(f'{scope}{source}{r["protocol"]} dport {port} counter {verdict} comment "{comment}"')
    seen = set()
    for r in p["port_rules"]:
        if r["action"] != "open" or not r["src_cidrs"]: continue
        key = (r["protocol"], r["port_start"], r["port_end"])
        if key in seen: continue
        seen.add(key)
        port = str(r["port_start"]) if r["port_start"] == r["port_end"] else f"{r['port_start']}-{r['port_end']}"
        # A source-restricted open applies a default close to both IP families.
        rule(f'{scope}{r["protocol"]} dport {port} counter drop comment "evofw-port-implicit-{r["protocol"]}-{port}"')
    if chain != "prerouting":
        rule("ip saddr @allow_v4 counter accept")
        rule("ip6 saddr @allow_v6 counter accept")
        rule("counter " + p["default_action"])

PY
}

apply_nft() {
  local batch="$RUN_DIR/nft.batch" exists=0 counters hits accepted=0
  nft list table inet evofw >/dev/null 2>&1 && exists=1
  # Optional telemetry features are tested against the complete candidate.
  # A failure never weakens enforcement: retry only without optional counters.
  for mode in '1 1' '1 0' '0 0'; do
    read -r counters hits <<<"$mode"
    render_nft_batch "$counters" "$hits" "$exists" >"$batch" || return 1
    if nft --check --file "$batch" 2>>"$LOG_FILE"; then accepted=1; break; fi
  done
  [[ "$accepted" == 1 ]] || { log "nft candidate validation failed; keep active policy"; return 1; }
  nft --file "$batch" 2>>"$LOG_FILE" || return 1
  PORT_HITS_ENABLED=$hits
  APPLIED_FAMILIES="4,6"
  KERNEL_METHOD=nft
  APPLIED=$((${#DENY[@]} + ${#ALLOW[@]}))
}

# Collect observed host firewall rules + listeners (best-effort).
collect_host_firewall() {
  printf '%s' '{"rules":[],"listeners":[]}' >"$HOST_FW_FILE"
  if ! command -v python3 >/dev/null 2>&1; then
    log "host_firewall: python3 missing — empty snapshot"
    return 0
  fi
  local tmpdir nft_f ipt_f ufw_f fwd_f ss_f out_f firewall_ok=0 listeners_ok=0
  tmpdir=$(mktemp -d "${STATE_DIR}/hostfw.XXXXXX") || return 0
  nft_f="$tmpdir/nft.txt"
  ipt_f="$tmpdir/ipt.txt"
  ufw_f="$tmpdir/ufw.txt"
  fwd_f="$tmpdir/fwd.txt"
  ss_f="$tmpdir/ss.txt"
  out_f="$tmpdir/out.json"
  # Dump to files — env vars blow ARG_MAX on large nft rulesets.
  nft list ruleset >"$nft_f" 2>/dev/null && firewall_ok=1 || true
  iptables-save >"$ipt_f" 2>/dev/null && firewall_ok=1 || true
  if command -v ufw >/dev/null 2>&1; then
    command timeout 30 ufw status verbose >"$ufw_f" 2>/dev/null || true
  else
    : >"$ufw_f"
  fi
  if command -v firewall-cmd >/dev/null 2>&1; then
    command timeout 30 firewall-cmd --list-all >"$fwd_f" 2>/dev/null || true
  else
    : >"$fwd_f"
  fi
  command timeout 30 ss -lntu >"$ss_f" 2>/dev/null && listeners_ok=1 || true
  if [[ "$firewall_ok" != 1 || "$listeners_ok" != 1 ]]; then
    # Absence of a successful observation is unknown, never an empty firewall.
    rm -f "$HOST_FW_FILE"
    rm -rf -- "$tmpdir"
    return 0
  fi

  if NFT_F="$nft_f" IPT_F="$ipt_f" UFW_F="$ufw_f" FWD_F="$fwd_f" SS_F="$ss_f" OUT_F="$out_f" python3 - <<'PY' >"$RUN_DIR/hostfw-counts.txt" 2>>"$LOG_FILE"
import json, os, re

def read(path: str) -> str:
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as f:
            return f.read()
    except OSError:
        return ""

def ownership_of(text: str) -> str:
    t = text.lower()
    if "evofw" in t:
        return "evofw"
    return "foreign"

def rule(**kwargs):
    # Omit null/empty optional fields — Zod optional rejects JSON null.
    out = {
        "ownership": kwargs["ownership"],
        "backend": kwargs["backend"],
        "raw": kwargs["raw"][:512],
    }
    for k in ("table", "chain", "action", "protocol", "dport", "sport", "saddr", "daddr", "comment"):
        v = kwargs.get(k)
        if v is not None and v != "":
            out[k] = v if not isinstance(v, str) else v[:128 if k in ("table", "chain", "saddr", "daddr") else (64 if k in ("action", "dport", "sport", "protocol") else 256)]
    return out

rules = []
listeners = []

nft = read(os.environ["NFT_F"])
cur_table = ""
cur_chain = ""
for line in nft.splitlines():
    ls = line.strip()
    if ls.startswith("table "):
        cur_table = ls
        cur_chain = ""
        continue
    m = re.match(r"chain\s+(\S+)", ls)
    if m:
        cur_chain = m.group(1)
        continue
    if not ls or ls.startswith("type ") or ls.startswith("policy ") or ls.startswith("set ") or ls.startswith("map "):
        continue
    if "accept" in ls or "drop" in ls or "reject" in ls or "jump " in ls or "goto " in ls:
        act = "accept" if " accept" in f" {ls}" or ls.endswith("accept") else (
            "drop" if " drop" in f" {ls}" or ls.endswith("drop") else (
                "reject" if "reject" in ls else "other"
            )
        )
        proto = ""
        if " tcp " in f" {ls}" or ls.startswith("tcp "):
            proto = "tcp"
        elif " udp " in f" {ls}" or ls.startswith("udp "):
            proto = "udp"
        dport = ""
        m = re.search(r"dport\s+(\S+)", ls)
        if m:
            dport = m.group(1)
        saddr = ""
        m = re.search(r"saddr\s+(\S+)", ls)
        if m:
            saddr = m.group(1).lstrip("@")
        rules.append(rule(
            ownership=ownership_of(cur_table + " " + cur_chain + " " + ls),
            backend="nft",
            table=cur_table[:120],
            chain=cur_chain[:120],
            action=act,
            protocol=proto or None,
            dport=dport or None,
            saddr=saddr or None,
            raw=ls[:500],
        ))

ipt = read(os.environ["IPT_F"])
cur_chain = ""
for line in ipt.splitlines():
    if line.startswith(":"):
        cur_chain = line[1:].split()[0] if line[1:] else ""
        continue
    if not line.startswith("-A "):
        continue
    parts = line.split(None, 2)
    chain = parts[1] if len(parts) > 1 else ""
    act = "DROP" if " -j DROP" in line else (
        "ACCEPT" if " -j ACCEPT" in line else (
            "REJECT" if " -j REJECT" in line else "other"
        )
    )
    proto = ""
    m = re.search(r"-p\s+(\w+)", line)
    if m:
        proto = m.group(1)
    dport = ""
    m = re.search(r"--dport(?:s)?\s+(\S+)", line)
    if m:
        dport = m.group(1)
    saddr = ""
    m = re.search(r"-s\s+(\S+)", line)
    if m:
        saddr = m.group(1)
    rules.append(rule(
        ownership=ownership_of(line),
        backend="iptables",
        chain=chain[:120],
        action=act.lower() if isinstance(act, str) else act,
        protocol=proto or None,
        dport=dport or None,
        saddr=saddr or None,
        raw=line[:500],
    ))

ufw = read(os.environ["UFW_F"])
for line in ufw.splitlines():
    ls = line.strip()
    if not ls or ls.startswith("Status") or ls.startswith("Logging") or ls.startswith("Default") or ls.startswith("To") or ls.startswith("--"):
        continue
    if "ALLOW" in ls or "DENY" in ls or "REJECT" in ls:
        rules.append(rule(
            ownership=ownership_of(ls),
            backend="ufw",
            action="allow" if "ALLOW" in ls else ("deny" if "DENY" in ls else "reject"),
            raw=ls[:500],
        ))

fwd = read(os.environ["FWD_F"])
for line in fwd.splitlines():
    ls = line.strip()
    if not ls:
        continue
    if ls.startswith("ports:") or ls.startswith("services:") or ":" in ls:
        rules.append(rule(
            ownership="foreign",
            backend="firewalld",
            raw=ls[:500],
        ))

ss = read(os.environ["SS_F"])
for line in ss.splitlines()[1:]:
    parts = line.split()
    if len(parts) < 5:
        continue
    proto = parts[0]
    local = parts[4]
    m = re.search(r"([^:]+):(\d+)$", local)
    if not m:
        m = re.search(r"\[([^\]]+)\]:(\d+)$", local)
        if not m:
            continue
        addr, port_s = m.group(1), m.group(2)
    else:
        addr, port_s = m.group(1), m.group(2)
    try:
        port = int(port_s)
    except ValueError:
        continue
    listeners.append({
        "protocol": "tcp" if proto.startswith("tcp") else ("udp" if proto.startswith("udp") else proto),
        "port": port,
        "address": addr,
    })

rules = rules[:500]
listeners = listeners[:200]
with open(os.environ["OUT_F"], "w", encoding="utf-8") as f:
    json.dump({"rules": rules, "listeners": listeners}, f, separators=(",", ":"))
print(f"{len(rules)} {len(listeners)}")
PY
  then
    if [[ -f "$out_f" ]]; then
      cp "$out_f" "$HOST_FW_FILE"
      log "host_firewall collected $(tr -d '\r\n' <"$RUN_DIR/hostfw-counts.txt" 2>/dev/null || echo '?')"
    fi
  else
    log "host_firewall: collect failed — empty snapshot"
    rm -f "$HOST_FW_FILE"
  fi
  rm -rf "$tmpdir" 2>/dev/null || true
}

# ipset cannot express Port ACL or Docker DNAT semantics. Refuse such a
# candidate instead of acknowledging partially enforced policy.
prepare_ipset_rules() {
  local family=$1
  python3 - "$RUN_DIR" "$family" "$DEFAULT_ACTION" <<'PY'
import pathlib, re, sys
root, family, default = pathlib.Path(sys.argv[1]), sys.argv[2], sys.argv[3]
lines = (root / ("iptables-before-v" + family + ".txt")).read_text().splitlines()
inputs = [line for line in lines if line.startswith("-A INPUT ")]
owned = [line for line in lines if line.startswith("-A EVOFW_INPUT ")]
jump = "-A INPUT -j EVOFW_INPUT"
existed = any(line.startswith(":EVOFW_INPUT ") for line in lines)
candidate = ["*filter", ":EVOFW_INPUT - [0:0]", "-F EVOFW_INPUT",
    f"-A EVOFW_INPUT -m set --match-set evofw_deny_v{family} src -j DROP",
    f"-A EVOFW_INPUT -m set --match-set evofw_allow_v{family} src -j ACCEPT"]
if default == "drop": candidate.append("-A EVOFW_INPUT -j DROP")
candidate += ["-D INPUT -j EVOFW_INPUT" for line in inputs if line == jump]
candidate += ["-I INPUT 1 -j EVOFW_INPUT", "COMMIT"]
rollback = ["*filter", ":EVOFW_INPUT - [0:0]", "-F EVOFW_INPUT", "-D INPUT -j EVOFW_INPUT"]
rollback += owned
# Restore original owned jump positions, keeping all foreign rules intact.
rollback += [f"-I INPUT {i + 1} -j EVOFW_INPUT" for i, line in enumerate(inputs) if line == jump]
if not existed: rollback.append("-X EVOFW_INPUT")
rollback.append("COMMIT")
(root / ("iptables-v" + family + ".restore")).write_text("\n".join(candidate) + "\n", newline="\n")
(root / ("rollback-v" + family + ".restore")).write_text("\n".join(rollback) + "\n", newline="\n")
PY
}

apply_ipset() {
  [[ "$PORT_RULES_COUNT" == 0 ]] || { APPLY_ERROR_CODE=UNSUPPORTED_CAPABILITY; log "ipset backend does not support Port ACL"; return 1; }
  local families=(4) version kind name stage create p tool failed=0 rollback_failed=0
  local swapped=() restored=() restore="$RUN_DIR/ipset.restore"
  if type -P ip6tables >/dev/null 2>&1 && type -P ip6tables-save >/dev/null 2>&1 && type -P ip6tables-restore >/dev/null 2>&1; then
    families+=(6)
  elif [[ "$(cat "$RUN_DIR/ipv6_required")" == 1 || "$DEFAULT_ACTION" == drop ]]; then
    APPLY_ERROR_CODE=UNSUPPORTED_CAPABILITY
    log "IPv6 enforcement needs ip6tables, ip6tables-save and ip6tables-restore; keep active policy"
    return 1
  fi
  : >"$restore"
  # Prepare every set and both rule-family candidates before any active swap.
  for version in "${families[@]}"; do
    for kind in deny allow; do
      name="evofw_${kind}_v${version}"; stage="evofw_${kind}_stage_v${version}"
      if ! ipset list "$name" >/dev/null 2>&1; then
        if [[ "$version" == 6 ]]; then tool=inet6; else tool=inet; fi
        ipset create "$name" hash:net family "$tool" counters || return 1
      fi
      ipset destroy "$stage" 2>/dev/null || true
      create=$(ipset save "$name" | sed -n '1p') || return 1
      [[ "$create" == "create $name hash:net "* ]] || { log "incompatible ipset $name"; return 1; }
      printf '%s\n' "${create/create $name /create $stage }" >>"$restore"
      if [[ "$kind" == deny ]]; then
        for p in "${DENY[@]}"; do
          [[ "$version" == 6 && "$p" != *:* || "$version" == 4 && "$p" == *:* ]] && continue
          # hash:net does not accept /0; two /1 networks enforce the same set.
          if [[ "$p" == 0.0.0.0/0 ]]; then printf 'add %s 0.0.0.0/1\nadd %s 128.0.0.0/1\n' "$stage" "$stage" >>"$restore"
          elif [[ "$p" == ::/0 ]]; then printf 'add %s ::/1\nadd %s 8000::/1\n' "$stage" "$stage" >>"$restore"
          else printf 'add %s %s\n' "$stage" "$p" >>"$restore"; fi
        done
      else
        for p in "${ALLOW[@]}"; do
          [[ "$version" == 6 && "$p" != *:* || "$version" == 4 && "$p" == *:* ]] && continue
          if [[ "$p" == 0.0.0.0/0 ]]; then printf 'add %s 0.0.0.0/1\nadd %s 128.0.0.0/1\n' "$stage" "$stage" >>"$restore"
          elif [[ "$p" == ::/0 ]]; then printf 'add %s ::/1\nadd %s 8000::/1\n' "$stage" "$stage" >>"$restore"
          else printf 'add %s %s\n' "$stage" "$p" >>"$restore"; fi
        done
      fi
    done
    if [[ "$version" == 6 ]]; then tool=ip6tables; else tool=iptables; fi
    "${tool}-save" -t filter >"$RUN_DIR/iptables-before-v${version}.txt" || return 1
    prepare_ipset_rules "$version" || return 1
  done
  ipset restore <"$restore" || { log "ipset stage failed; keep active policy"; return 1; }
  for version in "${families[@]}"; do
    if [[ "$version" == 6 ]]; then tool=ip6tables; else tool=iptables; fi
    "${tool}-restore" --wait 10 --test --noflush <"$RUN_DIR/iptables-v${version}.restore" || return 1
  done
  for version in "${families[@]}"; do
    for kind in deny allow; do
      stage="evofw_${kind}_stage_v${version}"; name="evofw_${kind}_v${version}"
      if ipset swap "$stage" "$name"; then swapped+=("$kind:$version"); else failed=1; break 2; fi
    done
  done
  if [[ "$failed" == 0 ]]; then
    for version in "${families[@]}"; do
      if [[ "$version" == 6 ]]; then tool=ip6tables; else tool=iptables; fi
      if "${tool}-restore" --wait 10 --noflush <"$RUN_DIR/iptables-v${version}.restore"; then restored+=("$version"); else failed=1; break; fi
    done
  fi
  if [[ "$failed" == 1 ]]; then
    # Restore successful first-family rule transactions as well as every set.
    for version in "${restored[@]}"; do
      if [[ "$version" == 6 ]]; then tool=ip6tables; else tool=iptables; fi
      "${tool}-restore" --wait 10 --noflush <"$RUN_DIR/rollback-v${version}.restore" || rollback_failed=1
    done
    for p in "${swapped[@]}"; do
      kind=${p%%:*}; version=${p##*:}
      ipset swap "evofw_${kind}_stage_v${version}" "evofw_${kind}_v${version}" || rollback_failed=1
    done
    if [[ "$rollback_failed" == 1 ]]; then APPLY_ERROR_CODE=rollback_failed; log "CRITICAL: cross-family rollback failed"; fi
    log "ipset apply failed; both families rollback attempted; hash not committed"
    return 1
  fi
  for p in "${swapped[@]}"; do kind=${p%%:*}; version=${p##*:}; ipset destroy "evofw_${kind}_stage_v${version}" || return 1; done
  APPLIED_FAMILIES=$(IFS=,; echo "${families[*]}")
  KERNEL_METHOD=ipset
  APPLIED=$((${#DENY[@]} + ${#ALLOW[@]}))
}

collect_telemetry() {
  local method=${KERNEL_METHOD:-$BACKEND}
  rm -f "$RUN_DIR/metrics-expected-families"
  rm -f "$RUN_DIR/metrics-nft.json" "$RUN_DIR/metrics-iptables.txt" "$RUN_DIR/metrics-ipset.txt" "$RUN_DIR/metrics-iptables-v6.txt"
  if [[ "$method" == nft || "$method" == auto ]] && type -P nft >/dev/null 2>&1; then
    nft --json list table inet evofw >"$RUN_DIR/metrics-nft.json" 2>>"$LOG_FILE" || rm -f "$RUN_DIR/metrics-nft.json"
    method=nft
  elif [[ "$method" == ipset ]] && type -P ipset >/dev/null 2>&1; then
    iptables-save -c -t filter >"$RUN_DIR/metrics-iptables.txt" 2>>"$LOG_FILE" || rm -f "$RUN_DIR/metrics-iptables.txt"
    ipset save evofw_deny_v4 >"$RUN_DIR/metrics-ipset.txt" 2>>"$LOG_FILE" || rm -f "$RUN_DIR/metrics-ipset.txt"
    if ipset list evofw_deny_v6 >/dev/null 2>&1; then
      printf '%s' '4,6' >"$RUN_DIR/metrics-expected-families"
      ip6tables-save -c -t filter >"$RUN_DIR/metrics-iptables-v6.txt" 2>>"$LOG_FILE" || rm -f "$RUN_DIR/metrics-iptables-v6.txt"
      ipset save evofw_deny_v6 >>"$RUN_DIR/metrics-ipset.txt" 2>>"$LOG_FILE" || rm -f "$RUN_DIR/metrics-ipset.txt"
    fi
  fi
  telemetry_worker capture "$method"
}

send_report() {
  # Snapshot the observed host firewall at most every five minutes, or after
  # apply. Telemetry from cached snapshots is labelled with its real contents.
  if [[ "${POLICY_CHANGED:-0}" == 1 ]] || telemetry_worker host-due; then
    collect_host_firewall
  fi
  collect_telemetry
  telemetry_worker enqueue "${REPORT_STATUS:-ok}" "${KERNEL_METHOD:-$BACKEND}"
  flush_reports
}

on_apply_error() {
  local rc=$?
  trap - ERR
  REPORT_STATUS=error
  log "apply failed rc=$rc; successful hash unchanged"
  send_report || true
  exit "$rc"
}
trap on_apply_error ERR

# Compare owned kernel structure as well as the desired hash. Packet counters,
# handles and expiring hit elements are excluded, so ordinary traffic never
# forces a replacement. Missing rules/sets or changed enforcement trigger repair.
kernel_digest() {
  local method=$1
  if [[ "$method" == nft ]]; then
    nft --json list table inet evofw >"$RUN_DIR/live-kernel.json" 2>/dev/null || return 1
  else
    ipset save evofw_deny_v4 >"$RUN_DIR/live-ipset.txt" 2>/dev/null || return 1
    ipset save evofw_allow_v4 >>"$RUN_DIR/live-ipset.txt" 2>/dev/null || return 1
    iptables-save -t filter >"$RUN_DIR/live-iptables.txt" 2>/dev/null || return 1
    rm -f "$RUN_DIR/live-iptables-v6.txt"
    if ipset list evofw_deny_v6 >/dev/null 2>&1; then
      ipset save evofw_deny_v6 >>"$RUN_DIR/live-ipset.txt" 2>/dev/null || return 1
      ipset save evofw_allow_v6 >>"$RUN_DIR/live-ipset.txt" 2>/dev/null || return 1
      ip6tables-save -t filter >"$RUN_DIR/live-iptables-v6.txt" 2>/dev/null || return 1
    fi
  fi
  python3 - "$method" "$RUN_DIR" <<'PY'
import hashlib, json, pathlib, re, sys
root = pathlib.Path(sys.argv[2])
if sys.argv[1] == "nft":
    data = json.loads((root / "live-kernel.json").read_text())
    def clean(value):
        if isinstance(value, dict):
            return {k: clean(v) for k, v in value.items() if k not in ("handle", "packets", "bytes", "expires") and not (k == "elem" and value.get("name", "").startswith("deny_port_hits"))}
        if isinstance(value, list):
            return [clean(v) for v in value if not (isinstance(v, dict) and "metainfo" in v)]
        return value
    canonical = json.dumps(clean(data), sort_keys=True, separators=(",", ":"))
else:
    sets = (root / "live-ipset.txt").read_text()
    sets = re.sub(r" (?:packets|bytes) [0-9]+", "", sets)
    v6 = root / "live-iptables-v6.txt"
    all_rules = []
    for rule_file in [root / "live-iptables.txt"] + ([v6] if v6.exists() else []):
      rules = rule_file.read_text().splitlines()
      rules = [re.sub(r"\[[0-9]+:[0-9]+\]", "[0:0]", r) for r in rules if r.startswith(":EVOFW_INPUT ") or r.startswith("-A EVOFW_INPUT ") or r.startswith("-A INPUT ")]
      input_rules = [r for r in rules if r.startswith("-A INPUT ")]
      if not input_rules or input_rules[0] != "-A INPUT -j EVOFW_INPUT":
          raise ValueError("owned INPUT jump is missing or preempted")
      all_rules += [rule_file.name] + rules
    canonical = "\n".join(sorted(sets.splitlines())) + "\n" + "\n".join(all_rules)
print(hashlib.sha256(canonical.encode()).hexdigest())
PY
}

KERNEL_STATE_FILE="$STATE_DIR/kernel_state"
if [[ -f "$HASH_FILE" && "$(tr -d '\r\n' <"$HASH_FILE")" == "$HASH" && -f "$KERNEL_STATE_FILE" ]]; then
  read -r previous_method previous_digest <"$KERNEL_STATE_FILE" || true
  if [[ "$previous_method" == nft || "$previous_method" == ipset ]]; then
    if [[ "$BACKEND" == auto || "$BACKEND" == "$previous_method" ]]; then
      live_digest=$(kernel_digest "$previous_method") || live_digest=""
      if [[ -n "$live_digest" && "$live_digest" == "$previous_digest" ]]; then
        KERNEL_METHOD=$previous_method
        APPLIED=$((${#DENY[@]} + ${#ALLOW[@]}))
        # A new generation may intentionally reuse the same content hash.
        # Commit its acknowledgement only after proving live enforcement.
        telemetry_worker commit-policy
        log "unchanged hash $HASH; kernel verified; revision acknowledged; skip apply"
        send_report
        exit 0
      fi
    fi
  fi
  log "owned kernel state drifted; reconcile policy"
fi

# Durably queue the final old-epoch snapshot BEFORE any counter-resetting apply.
# A full/failed outbox blocks apply rather than discarding unacknowledged data.
if [[ -f "$HASH_FILE" ]]; then
  if [[ -f "$KERNEL_STATE_FILE" ]]; then read -r KERNEL_METHOD _ <"$KERNEL_STATE_FILE" || true; fi
  collect_telemetry
  telemetry_worker enqueue ok "${KERNEL_METHOD:-$BACKEND}"
  flush_reports
fi

case "$BACKEND" in
  nft|auto)
    if type -P nft >/dev/null 2>&1; then apply_nft
    elif type -P ipset >/dev/null 2>&1; then apply_ipset
    else log "no backend"; exit 1; fi
    ;;
  ipset) apply_ipset ;;
  *) log "invalid backend $BACKEND"; exit 1 ;;
esac

# Store a verified structural digest for the no-change path. If inspection
# fails, enforcement succeeded but the next poll must reconcile conservatively.
live_digest=$(kernel_digest "$KERNEL_METHOD") || live_digest=""
if [[ "$KERNEL_METHOD" == ipset && -z "$live_digest" ]]; then
  log "ipset verification failed after apply; hash not acknowledged"
  false
fi
telemetry_worker rotate "${APPLIED_FAMILIES:-4}"
POLICY_CHANGED=1
printf '%s %s\n' "$KERNEL_METHOD" "$live_digest" >"${KERNEL_STATE_FILE}.new"
mv -f "${KERNEL_STATE_FILE}.new" "$KERNEL_STATE_FILE"
# Commit local success only after every required kernel operation succeeded.
telemetry_worker commit-policy
printf '%s\n' "$HASH" >"${HASH_FILE}.new"
mv -f "${HASH_FILE}.new" "$HASH_FILE"
log "applied default_action=$DEFAULT_ACTION count=$APPLIED method=$KERNEL_METHOD"
send_report