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

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"

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 ! install -m 755 "$tmp" "$SYNC_SCRIPT"; then
    log "self-update: install failed — keep current"
    rm -f "$tmp"
    return 0
  fi
  rm -f "$tmp"
  rm -f "$HASH_FILE"
  log "self-update: installed script sha256=${remote_sha} — re-exec"
  exec env EVOFW_SKIP_SELF_UPDATE=1 "$SYNC_SCRIPT" || {
    log "self-update: exec failed — continue current"
    return 0
  }
}

maybe_self_update

curl_policy() {
  local dest="$1"
  local code
  code=$(curl -sS -o "$dest" -w "%{http_code}" \
    -H "Authorization: Bearer ${CLIENT_TOKEN}" \
    -H "Accept: application/json" \
    "${EVOFW_CP_URL%/}/v1/agent/policy") || return 1
  if [[ "$code" == "403" ]]; then
    log "pending approval"
    return 2
  fi
  if [[ "$code" != "200" ]]; then
    log "policy HTTP $code"
    return 1
  fi
  return 0
}

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

parse_policy() {
  local f="$1"
  PORT_RULES_FILE="${STATE_DIR}/last_port_rules.json"
  if command -v jq >/dev/null 2>&1; then
    HASH=$(jq -r '.hash // empty' "$f")
    DEFAULT_ACTION=$(jq -r '.default_action // empty' "$f")
    if [[ -z "$DEFAULT_ACTION" ]]; then
      local legacy
      legacy=$(jq -r '.policy_mode // "blacklist"' "$f")
      if [[ "$legacy" == "whitelist" ]]; then DEFAULT_ACTION=drop; else DEFAULT_ACTION=accept; fi
    fi
    mapfile -t DENY < <(jq -r '.deny_cidrs[]? // empty' "$f")
    mapfile -t ALLOW < <(jq -r '.allow_cidrs[]? // empty' "$f")
    jq -c '.port_rules // []' "$f" >"$PORT_RULES_FILE" 2>/dev/null || echo '[]' >"$PORT_RULES_FILE"
    return 0
  fi
  if command -v python3 >/dev/null 2>&1; then
    eval "$(python3 - "$f" "$PORT_RULES_FILE" <<'PY'
import json,sys
d=json.load(open(sys.argv[1],encoding="utf-8"))
print(f'HASH={d.get("hash") or ""}')
da=d.get("default_action") or ""
if not da:
  da="drop" if d.get("policy_mode")=="whitelist" else "accept"
print(f'DEFAULT_ACTION={da}')
print("DENY=("+" ".join(json.dumps(x) for x in (d.get("deny_cidrs") or []))+")")
print("ALLOW=("+" ".join(json.dumps(x) for x in (d.get("allow_cidrs") or []))+")")
open(sys.argv[2],"w",encoding="utf-8").write(json.dumps(d.get("port_rules") or []))
PY
)"
    return 0
  fi
  log "need jq or python3"
  exit 1
}

HASH=""; DEFAULT_ACTION=accept; DENY=(); ALLOW=()
PORT_RULES_FILE="${STATE_DIR}/last_port_rules.json"
parse_policy "$POLICY_FILE"
# Empty deny/allow is valid — agent may have no rule sets yet.
DENY=("${DENY[@]+"${DENY[@]}"}")
ALLOW=("${ALLOW[@]+"${ALLOW[@]}"}")
[[ -f "$PORT_RULES_FILE" ]] || echo '[]' >"$PORT_RULES_FILE"
PORT_RULES_COUNT=0
if command -v python3 >/dev/null 2>&1; then
  PORT_RULES_COUNT=$(python3 -c 'import json,sys; print(len(json.load(open(sys.argv[1]))))' "$PORT_RULES_FILE" 2>/dev/null || echo 0)
fi
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"
printf '%s' '{"rules":[],"listeners":[]}' >"$HOST_FW_FILE"
printf '%s' '[]' >"$IP_HITS_FILE"
printf '%s' '[]' >"$PORT_HITS_FILE"
# 1 when deny_port_hits dynamic set is available for this apply.
PORT_HITS_ENABLED=0

nft_join() {
  local out="" p
  for p in "$@"; do
    [[ -n "$out" ]] && out+=", "
    out+="$p"
  done
  printf '%s' "$out"
}

nft_add_chunk() {
  local table=$1 name=$2 setname=$3
  shift 3
  local joined; joined=$(nft_join "$@")
  nft add element "$table" "$name" "$setname" "{ ${joined} }" 2>>"$LOG_FILE" || {
    for p in "$@"; do nft add element "$table" "$name" "$setname" "{ $p }" 2>>"$LOG_FILE" || true; done
  }
}

# Ensure inet set exists. Prefer per-element counters; many kernels reject
# `counter` on interval sets — fall back to plain interval (no per-IP hits).
# Caller must delete referencing chains before recreating a set.
ensure_nft_set() {
  local table=$1 name=$2 setname=$3
  local def
  def=$(nft list set "$table" "$name" "$setname" 2>/dev/null || true)
  if [[ -n "$def" ]] && [[ "$def" == *"counter"* ]]; then
    return 0
  fi
  if [[ -n "$def" ]]; then
    # Upgrade path: drop old set without counters (chain must already be gone).
    nft delete set "$table" "$name" "$setname" 2>>"$LOG_FILE" || true
  fi
  if nft add set "$table" "$name" "$setname" '{ type ipv4_addr; flags interval; counter; }' 2>>"$LOG_FILE"; then
    return 0
  fi
  # Set may still exist if delete failed — try plain create only if missing.
  if nft list set "$table" "$name" "$setname" >/dev/null 2>&1; then
    log "nft: keep existing set $setname (no per-element counter)"
    return 0
  fi
  if nft add set "$table" "$name" "$setname" '{ type ipv4_addr; flags interval; }' 2>>"$LOG_FILE"; then
    log "nft: set $setname without counter (interval+counter unsupported)"
    return 0
  fi
  log "nft: failed to create set $setname"
  return 1
}

collect_nft_stats() {
  PACKETS_DROPPED=0; PACKETS_ACCEPTED=0
  local line n
  while IFS= read -r line; do
    [[ "$line" =~ packets[[:space:]]+([0-9]+) ]] || continue
    n="${BASH_REMATCH[1]}"
    # Policy set hits only (ignore lo / established noise)
    if [[ "$line" == *@deny_v4* ]]; then
      PACKETS_DROPPED=$((PACKETS_DROPPED + n))
    elif [[ "$line" == *@allow_v4* ]]; then
      PACKETS_ACCEPTED=$((PACKETS_ACCEPTED + n))
    elif [[ "$line" == *" counter drop"* && "$line" != *@* ]]; then
      PACKETS_DROPPED=$((PACKETS_DROPPED + n))
    elif [[ "$line" == *" counter accept"* && "$line" != *@* && "$line" != *established* && "$line" != *"iif \"lo\""* && "$line" != *"iif lo"* ]]; then
      PACKETS_ACCEPTED=$((PACKETS_ACCEPTED + n))
    fi
  done < <(nft list chain inet evofw input 2>/dev/null || true)
}

# Parse nft/ipset listing from file → IP_HITS_FILE (top-N JSON).
build_ip_hits_from_file() {
  local src="$1"
  printf '%s' '[]' >"$IP_HITS_FILE"
  [[ -f "$src" ]] || return 0
  if ! command -v python3 >/dev/null 2>&1; then
    return 0
  fi
  IP_HITS_TOP="$IP_HITS_TOP" IN_F="$src" OUT_F="$IP_HITS_FILE" python3 - <<'PY' 2>>"$LOG_FILE" || printf '%s' '[]' >"$IP_HITS_FILE"
import json, os, re
path = os.environ["IN_F"]
top = int(os.environ.get("IP_HITS_TOP", "200"))
try:
    with open(path, "r", encoding="utf-8", errors="replace") as f:
        text = f.read()
except OSError:
    text = ""
hits = {}
for m in re.finditer(r"([0-9]{1,3}(?:\.[0-9]{1,3}){3}(?:/[0-9]{1,2})?)\s+(?:counter\s+)?packets\s+(\d+)", text):
    ip, pkts = m.group(1), int(m.group(2))
    if pkts > 0:
        hits[ip] = max(hits.get(ip, 0), pkts)
for m in re.finditer(r"^([0-9]{1,3}(?:\.[0-9]{1,3}){3}(?:/[0-9]{1,2})?)\s+packets\s+(\d+)", text, re.M):
    ip, pkts = m.group(1), int(m.group(2))
    if pkts > 0:
        hits[ip] = max(hits.get(ip, 0), pkts)
items = [{"ip": k, "packets": v} for k, v in hits.items()]
items.sort(key=lambda x: x["packets"], reverse=True)
with open(os.environ["OUT_F"], "w", encoding="utf-8") as f:
    json.dump(items[:top], f, separators=(",", ":"))
print(len(items[:top]))
PY
}

collect_nft_ip_hits() {
  local dump="${STATE_DIR}/nft_deny_v4.txt"
  nft list set inet evofw deny_v4 >"$dump" 2>/dev/null || : >"$dump"
  build_ip_hits_from_file "$dump"
}

collect_ipset_ip_hits() {
  local dump="${STATE_DIR}/ipset_deny_v4.txt"
  ipset list evofw_deny_v4 >"$dump" 2>/dev/null || : >"$dump"
  build_ip_hits_from_file "$dump"
}

collect_ip_hits() {
  printf '%s' '[]' >"$IP_HITS_FILE"
  if [[ "${KERNEL_METHOD:-}" == "nft" ]] || { [[ -z "${KERNEL_METHOD:-}" || "${KERNEL_METHOD:-}" == "auto" ]] && command -v nft >/dev/null 2>&1 && nft list set inet evofw deny_v4 >/dev/null 2>&1; }; then
    collect_nft_ip_hits
    return
  fi
  if command -v ipset >/dev/null 2>&1 && ipset list evofw_deny_v4 >/dev/null 2>&1; then
    collect_ipset_ip_hits
  fi
}

# Parse nft dynamic concat set from file → PORT_HITS_FILE.
build_port_hits_from_file() {
  local src="$1"
  printf '%s' '[]' >"$PORT_HITS_FILE"
  [[ -f "$src" && -s "$src" ]] || return 0
  if ! command -v python3 >/dev/null 2>&1; then
    return 0
  fi
  PORT_HITS_TOP="$PORT_HITS_TOP" IN_F="$src" OUT_F="$PORT_HITS_FILE" python3 - <<'PY' 2>>"$LOG_FILE" || printf '%s' '[]' >"$PORT_HITS_FILE"
import json, os, re
path = os.environ["IN_F"]
top = int(os.environ.get("PORT_HITS_TOP", "500"))
try:
    with open(path, "r", encoding="utf-8", errors="replace") as f:
        text = f.read()
except OSError:
    text = ""
hits = {}
proto_map = {"6": "tcp", "17": "udp", "tcp": "tcp", "udp": "udp"}
pat = re.compile(
    r"([0-9]{1,3}(?:\.[0-9]{1,3}){3})\s*\.\s*([A-Za-z0-9]+)\s*\.\s*(\d+)\s+"
    r"(?:timeout\s+\S+\s+)?(?:counter\s+)?packets\s+(\d+)",
    re.I,
)
for m in pat.finditer(text):
    ip, raw_proto, port_s, pkts_s = m.group(1), m.group(2).lower(), m.group(3), m.group(4)
    proto = proto_map.get(raw_proto)
    if not proto:
        continue
    pkts = int(pkts_s)
    if pkts <= 0:
        continue
    port = int(port_s)
    if port < 1 or port > 65535:
        continue
    key = (ip, port, proto)
    hits[key] = max(hits.get(key, 0), pkts)
items = [
    {"ip": ip, "port": port, "protocol": proto, "packets": pkts}
    for (ip, port, proto), pkts in hits.items()
]
items.sort(key=lambda x: x["packets"], reverse=True)
with open(os.environ["OUT_F"], "w", encoding="utf-8") as f:
    json.dump(items[:top], f, separators=(",", ":"))
print(len(items[:top]))
PY
}

collect_nft_port_hits() {
  local dump="${STATE_DIR}/nft_deny_port_hits.txt"
  printf '%s' '[]' >"$PORT_HITS_FILE"
  nft list set inet evofw deny_port_hits >"$dump" 2>/dev/null || : >"$dump"
  [[ -s "$dump" ]] || return 0
  build_port_hits_from_file "$dump"
}

collect_port_hits() {
  printf '%s' '[]' >"$PORT_HITS_FILE"
  if [[ "${KERNEL_METHOD:-}" == "nft" ]] || { [[ -z "${KERNEL_METHOD:-}" || "${KERNEL_METHOD:-}" == "auto" ]] && command -v nft >/dev/null 2>&1 && nft list set inet evofw deny_port_hits >/dev/null 2>&1; }; then
    collect_nft_port_hits
  fi
}

# Dynamic concat set for per-(ip, proto, dport) deny hits. Returns 0 if usable.
ensure_nft_port_hits_set() {
  local table=$1 name=$2
  local setname=deny_port_hits
  local def
  def=$(nft list set "$table" "$name" "$setname" 2>/dev/null || true)
  if [[ -n "$def" ]] && { [[ "$def" == *"dynamic"* ]] || [[ "$def" == *"timeout"* ]]; }; then
    # Keep existing; elements age out via timeout — do not flush on every apply
    # (counters survive policy CIDR refresh when chain is recreated).
    return 0
  fi
  if [[ -n "$def" ]]; then
    nft delete set "$table" "$name" "$setname" 2>>"$LOG_FILE" || true
  fi
  if nft add set "$table" "$name" "$setname" \
    '{ type ipv4_addr . inet_proto . inet_service; flags dynamic,timeout; timeout 1h; counter; }' \
    2>>"$LOG_FILE"; then
    return 0
  fi
  # Older kernels may need slightly different flag spelling.
  if nft add set "$table" "$name" "$setname" \
    '{ type ipv4_addr . inet_proto . inet_service; flags dynamic; timeout 1h; counter; }' \
    2>>"$LOG_FILE"; then
    return 0
  fi
  log "nft: deny_port_hits unsupported — port hits disabled"
  return 1
}

apply_nft() {
  local table=inet name=evofw
  local deny_v4=() allow_v4=() p
  PORT_HITS_ENABLED=0
  for p in "${DENY[@]+"${DENY[@]}"}"; do [[ "$p" == *:* ]] && continue; deny_v4+=("$p"); done
  for p in "${ALLOW[@]+"${ALLOW[@]}"}"; do [[ "$p" == *:* ]] && continue; allow_v4+=("$p"); done

  nft list table "$table" "$name" >/dev/null 2>&1 || nft add table "$table" "$name"
  # Drop chain first so sets can be deleted/recreated (upgrade to counters / port hits).
  # Stats were already captured by the caller before apply_nft.
  nft delete chain "$table" "$name" input 2>/dev/null || true
  ensure_nft_set "$table" "$name" deny_v4
  ensure_nft_set "$table" "$name" allow_v4
  if ensure_nft_port_hits_set "$table" "$name"; then
    PORT_HITS_ENABLED=1
  fi
  nft flush set "$table" "$name" deny_v4 2>>"$LOG_FILE" || true
  nft flush set "$table" "$name" allow_v4 2>>"$LOG_FILE" || true

  local batch=() chunk=64
  for p in "${deny_v4[@]}"; do
    batch+=("$p")
    if ((${#batch[@]} >= chunk)); then nft_add_chunk "$table" "$name" deny_v4 "${batch[@]}"; batch=(); fi
  done
  ((${#batch[@]})) && nft_add_chunk "$table" "$name" deny_v4 "${batch[@]}"
  batch=()
  for p in "${allow_v4[@]}"; do
    batch+=("$p")
    if ((${#batch[@]} >= chunk)); then nft_add_chunk "$table" "$name" allow_v4 "${batch[@]}"; batch=(); fi
  done
  ((${#batch[@]})) && nft_add_chunk "$table" "$name" allow_v4 "${batch[@]}"

  # Unified chain: deny → Port ACL (close/open/implicit) → allow → default_action
  if [[ "$DEFAULT_ACTION" == "drop" ]]; then
    nft add chain "$table" "$name" input '{ type filter hook input priority 0; policy drop; }'
  else
    nft add chain "$table" "$name" input '{ type filter hook input priority 0; policy accept; }'
  fi
  nft add rule "$table" "$name" input ct state established,related counter accept
  nft add rule "$table" "$name" input iif lo counter accept
  if [[ "$PORT_HITS_ENABLED" -eq 1 ]]; then
    # TCP/UDP: learn (ip, proto, dport) then drop; other L4: plain drop.
    if ! nft add rule "$table" "$name" input \
      ip saddr @deny_v4 meta l4proto '{ tcp, udp }' \
      update @deny_port_hits '{ ip saddr . meta l4proto . th dport }' \
      counter drop 2>>"$LOG_FILE"; then
      log "nft: port-hit deny rule failed — fallback to plain deny drop"
      PORT_HITS_ENABLED=0
      nft add rule "$table" "$name" input ip saddr @deny_v4 counter drop
    else
      nft add rule "$table" "$name" input ip saddr @deny_v4 counter drop
    fi
  else
    nft add rule "$table" "$name" input ip saddr @deny_v4 counter drop
  fi
  # Port ACL before L3 allow so open+list is exclusive (implicit drop per open port).
  apply_nft_port_acl "$table" "$name"
  nft add rule "$table" "$name" input ip saddr @allow_v4 counter accept
  if [[ "$DEFAULT_ACTION" == "drop" ]]; then
    nft add rule "$table" "$name" input counter drop
  else
    nft add rule "$table" "$name" input counter accept
  fi
  KERNEL_METHOD=nft
  APPLIED=$((${#deny_v4[@]} + ${#allow_v4[@]}))
}

# Apply desired L4 port open/close rules from PORT_RULES_FILE (apply_version 3).
apply_nft_port_acl() {
  local table=$1 name=$2
  [[ -f "$PORT_RULES_FILE" ]] || return 0
  if ! command -v python3 >/dev/null 2>&1; then
    log "nft port ACL skipped — need python3"
    return 0
  fi
  # Delete prior per-rule src sets (name prefix port_src_)
  local setline setname
  while IFS= read -r setline; do
    setname=$(echo "$setline" | sed -n 's/.*set \(port_src_[a-zA-Z0-9_-]*\).*/\1/p')
    [[ -n "$setname" ]] || continue
    nft delete set "$table" "$name" "$setname" 2>/dev/null || true
  done < <(nft list table "$table" "$name" 2>/dev/null | grep -E 'set port_src_' || true)

  local cmds_file
  cmds_file=$(mktemp)
  python3 - "$PORT_RULES_FILE" >"$cmds_file" <<'PY'
import json, sys, re
path = sys.argv[1]
try:
    rules = json.load(open(path, encoding="utf-8"))
except Exception:
    rules = []
safe_id = re.compile(r"[^a-zA-Z0-9_]")

def parse_rule(r):
    rid = safe_id.sub("_", str(r.get("id") or "x"))[:40]
    action = r.get("action") or "open"
    proto = r.get("protocol") or "tcp"
    if proto not in ("tcp", "udp"):
        return None
    ps = int(r.get("port_start") or 0)
    pe = int(r.get("port_end") or ps)
    if ps < 1 or pe > 65535 or pe < ps:
        return None
    cidrs = [c for c in (r.get("src_cidrs") or []) if c and ":" not in c]
    if not cidrs:
        return None
    dport = f"{ps}" if ps == pe else f"{ps}-{pe}"
    return {
        "rid": rid,
        "action": action,
        "proto": proto,
        "dport": dport,
        "cidrs": cidrs,
        "is_all": any(c in ("0.0.0.0/0", "0.0.0.0") for c in cidrs),
    }

def emit(p):
    verdict = "drop" if p["action"] == "close" else "accept"
    comment = f"evofw-port-{p['rid']}"
    proto, dport = p["proto"], p["dport"]
    if p["is_all"]:
        print(f'nft add rule inet evofw input {proto} dport {dport} counter {verdict} comment "{comment}"')
        return
    setname = f"port_src_{p['rid']}_{p['proto']}"
    print(f"nft add set inet evofw {setname} '{{ type ipv4_addr; flags interval; }}'")
    chunk = []
    for c in p["cidrs"]:
        chunk.append(c)
        if len(chunk) >= 32:
            print(f"nft add element inet evofw {setname} '{{ {', '.join(chunk)} }}'")
            chunk = []
    if chunk:
        print(f"nft add element inet evofw {setname} '{{ {', '.join(chunk)} }}'")
    print(
        f'nft add rule inet evofw input ip saddr @{setname} {proto} dport {dport} counter {verdict} comment "{comment}"'
    )

parsed = [p for p in (parse_rule(r) for r in rules) if p]
closes = [p for p in parsed if p["action"] == "close"]
opens = [p for p in parsed if p["action"] != "close"]
for p in closes:
    emit(p)
for p in opens:
    emit(p)
seen = set()
for p in opens:
    key = (p["proto"], p["dport"])
    if key in seen:
        continue
    seen.add(key)
    comment = f"evofw-port-implicit-{p['proto']}-{p['dport']}"
    print(
        f'nft add rule inet evofw input {p["proto"]} dport {p["dport"]} counter drop comment "{comment}"'
    )
PY
  local cmd
  while IFS= read -r cmd; do
    [[ -n "$cmd" ]] || continue
    # shellcheck disable=SC2086
    eval "$cmd" 2>>"$LOG_FILE" || log "nft port ACL cmd failed: $cmd"
  done <"$cmds_file"
  rm -f "$cmds_file"
}

# 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
  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 || true
  iptables-save >"$ipt_f" 2>/dev/null || true
  if command -v ufw >/dev/null 2>&1; then
    ufw status verbose >"$ufw_f" 2>/dev/null || true
  else
    : >"$ufw_f"
  fi
  if command -v firewall-cmd >/dev/null 2>&1; then
    firewall-cmd --list-all >"$fwd_f" 2>/dev/null || true
  else
    : >"$fwd_f"
  fi
  ss -lntu >"$ss_f" 2>/dev/null || true

  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' >/tmp/evofw-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' </tmp/evofw-hostfw-counts.txt 2>/dev/null || echo '?')"
    fi
  else
    log "host_firewall: collect failed — empty snapshot"
    printf '%s' '{"rules":[],"listeners":[]}' >"$HOST_FW_FILE"
  fi
  rm -rf "$tmpdir" 2>/dev/null || true
}

ensure_ipset_counters() {
  local name=$1
  if ! ipset list "$name" >/dev/null 2>&1; then
    if ipset create "$name" hash:net family inet counters 2>>"$LOG_FILE"; then
      return 0
    fi
    ipset create "$name" hash:net family inet 2>>"$LOG_FILE" || {
      log "ipset: failed to create $name"
      return 1
    }
    return 0
  fi
  # Recreate once if set has no packet counters (Header lacks "counters").
  local header
  header=$(ipset list "$name" 2>/dev/null | head -n 5 || true)
  if [[ "$header" == *"counters"* ]]; then
    return 0
  fi
  # Cannot safely destroy while iptables may reference the set — leave as-is.
  log "ipset: $name has no counters (leave existing; per-IP hits unavailable)"
}

apply_ipset() {
  local dset=evofw_deny_v4 aset=evofw_allow_v4
  ensure_ipset_counters "$dset"
  ensure_ipset_counters "$aset"
  ipset flush "$dset"; ipset flush "$aset"
  local p n=0
  for p in "${DENY[@]+"${DENY[@]}"}"; do [[ "$p" == *:* ]] && continue; ipset add "$dset" "$p" -exist; n=$((n+1)); done
  for p in "${ALLOW[@]+"${ALLOW[@]}"}"; do [[ "$p" == *:* ]] && continue; ipset add "$aset" "$p" -exist; n=$((n+1)); done
  iptables -D INPUT -m set --match-set "$dset" src -j DROP 2>/dev/null || true
  iptables -D INPUT -m set --match-set "$aset" src -j ACCEPT 2>/dev/null || true
  iptables -D INPUT -j DROP 2>/dev/null || true
  # Unified: deny first, then allow, then optional default drop
  iptables -I INPUT -m set --match-set "$dset" src -j DROP
  iptables -I INPUT 2 -m set --match-set "$aset" src -j ACCEPT
  if [[ "$DEFAULT_ACTION" == "drop" ]]; then
    iptables -A INPUT -j DROP 2>/dev/null || true
  fi
  KERNEL_METHOD=ipset
  APPLIED=$n
}

send_report() {
  # If caller already collected (pre-apply), keep those values.
  if [[ -z "${STATS_CAPTURED:-}" ]]; then
    if [[ "$KERNEL_METHOD" == "nft" ]] || command -v nft >/dev/null 2>&1; then
      collect_nft_stats
    fi
  fi
  if [[ -z "${IP_HITS_CAPTURED:-}" ]]; then
    collect_ip_hits
  fi
  if [[ -z "${PORT_HITS_CAPTURED:-}" ]]; then
    collect_port_hits
  fi
  if [[ -z "${HOST_FW_CAPTURED:-}" ]]; then
    collect_host_firewall
  fi
  local report_file http_code counts n_rules n_listen n_ip n_port
  report_file=$(mktemp "${STATE_DIR}/report.XXXXXX")
  # Compose report via files only — never bash ${var:-{...}} (} closes expansion).
  if command -v python3 >/dev/null 2>&1; then
    [[ -f "$HOST_FW_FILE" ]] || printf '%s' '{"rules":[],"listeners":[]}' >"$HOST_FW_FILE"
    [[ -f "$IP_HITS_FILE" ]] || printf '%s' '[]' >"$IP_HITS_FILE"
    [[ -f "$PORT_HITS_FILE" ]] || printf '%s' '[]' >"$PORT_HITS_FILE"
    counts=$(APPLIED="${APPLIED:-0}" DROPPED="${PACKETS_DROPPED:-0}" ACCEPTED="${PACKETS_ACCEPTED:-0}" \
      METHOD="${KERNEL_METHOD:-$BACKEND}" HOST_FW_F="$HOST_FW_FILE" IP_HITS_F="$IP_HITS_FILE" \
      PORT_HITS_F="$PORT_HITS_FILE" OUT_F="$report_file" python3 - <<'PY' 2>>"$LOG_FILE"
import json, os

def load(path, default):
    try:
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception as e:
        print(f"load_error {path}: {e}", file=__import__("sys").stderr)
        return default

payload = {
  "status": "ok",
  "prefix_count": int(os.environ.get("APPLIED") or 0),
  "packets_dropped": int(os.environ.get("DROPPED") or 0),
  "packets_accepted": int(os.environ.get("ACCEPTED") or 0),
  "kernel_method": os.environ.get("METHOD") or "auto",
  "source": "agent",
  "ip_hits": load(os.environ["IP_HITS_F"], []),
  "port_hits": load(os.environ["PORT_HITS_F"], []),
  "host_firewall": load(os.environ["HOST_FW_F"], {"rules": [], "listeners": []}),
}
with open(os.environ["OUT_F"], "w", encoding="utf-8") as f:
    json.dump(payload, f, separators=(",", ":"))
n_rules = len(payload["host_firewall"].get("rules") or [])
n_listen = len(payload["host_firewall"].get("listeners") or [])
n_ip = len(payload["ip_hits"] or [])
n_port = len(payload["port_hits"] or [])
print(f"{n_rules} {n_listen} {n_ip} {n_port}")
PY
) || counts="0 0 0 0"
    # shellcheck disable=SC2086
    set -- $counts
    n_rules=${1:-0}
    n_listen=${2:-0}
    n_ip=${3:-0}
    n_port=${4:-0}
  else
    printf '{"status":"ok","prefix_count":%s,"packets_dropped":%s,"packets_accepted":%s,"kernel_method":"%s","source":"agent","ip_hits":[],"port_hits":[]}' \
      "${APPLIED:-0}" "${PACKETS_DROPPED:-0}" "${PACKETS_ACCEPTED:-0}" "${KERNEL_METHOD:-$BACKEND}" \
      >"$report_file"
    n_rules=0
    n_listen=0
    n_ip=0
    n_port=0
  fi
  http_code=$(curl -sS -o /tmp/evofw-apply-report.out -w '%{http_code}' -X POST "${EVOFW_CP_URL%/}/v1/agent/apply-report" \
    -H "Authorization: Bearer ${CLIENT_TOKEN}" \
    -H "Content-Type: application/json" \
    --data-binary @"$report_file" 2>/dev/null || echo "000")
  log "apply-report http=${http_code} host_fw=${n_rules}/${n_listen} ip_hits=${n_ip} port_hits=${n_port}"
  if [[ "$http_code" != "200" ]]; then
    log "apply-report failed body=$(head -c 200 /tmp/evofw-apply-report.out 2>/dev/null || true)"
  fi
  rm -f "$report_file" 2>/dev/null || true
  curl -fsS -X POST "${EVOFW_CP_URL%/}/v1/agent/heartbeat" \
    -H "Authorization: Bearer ${CLIENT_TOKEN}" \
    -H "Content-Type: application/json" \
    -d '{"source":"agent"}' >/dev/null 2>&1 || true
}

if [[ -f "$HASH_FILE" && "$(tr -d '\r\n' <"$HASH_FILE")" == "$HASH" && -n "$HASH" ]]; then
  log "unchanged hash $HASH — skip apply"
  if command -v nft >/dev/null 2>&1 && nft list table inet evofw >/dev/null 2>&1; then
    KERNEL_METHOD=nft
  elif command -v ipset >/dev/null 2>&1 && ipset list evofw_deny_v4 >/dev/null 2>&1; then
    KERNEL_METHOD=ipset
  else
    KERNEL_METHOD="${BACKEND}"
  fi
  # Count applied prefixes from live sets when skipping apply.
  if [[ "$KERNEL_METHOD" == "nft" ]]; then
    APPLIED=$(nft list set inet evofw deny_v4 2>/dev/null | grep -cE '[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+' || true)
    local_allow=$(nft list set inet evofw allow_v4 2>/dev/null | grep -cE '[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+' || true)
    APPLIED=$((${APPLIED:-0} + ${local_allow:-0}))
  elif [[ "$KERNEL_METHOD" == "ipset" ]]; then
    APPLIED=$(ipset list evofw_deny_v4 2>/dev/null | awk '/^[0-9]/{c++} END{print c+0}')
    local_allow=$(ipset list evofw_allow_v4 2>/dev/null | awk '/^[0-9]/{c++} END{print c+0}')
    APPLIED=$((${APPLIED:-0} + ${local_allow:-0}))
  fi
  collect_host_firewall
  HOST_FW_CAPTURED=1
  send_report
  exit 0
fi

# Capture counters BEFORE recreate (nft delete chain / flush set zeroes them).
if command -v nft >/dev/null 2>&1 && nft list table inet evofw >/dev/null 2>&1; then
  collect_nft_stats
  collect_nft_ip_hits
  collect_nft_port_hits
  STATS_CAPTURED=1
  IP_HITS_CAPTURED=1
  PORT_HITS_CAPTURED=1
elif command -v ipset >/dev/null 2>&1 && ipset list evofw_deny_v4 >/dev/null 2>&1; then
  collect_ipset_ip_hits
  IP_HITS_CAPTURED=1
  printf '%s' '[]' >"$PORT_HITS_FILE"
  PORT_HITS_CAPTURED=1
fi
collect_host_firewall
HOST_FW_CAPTURED=1

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

echo "$HASH" >"$HASH_FILE"
log "applied default_action=$DEFAULT_ACTION count=$APPLIED method=$KERNEL_METHOD"
send_report