NexaFabric/backend/app/agent_assets/nexafabric-agent.py

#!/usr/bin/env python3
import argparse
import hashlib
import json
import os
import platform
import re
import select
import socket
import ssl
import struct
import subprocess
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


VERSION = "0.3.0"
VM_INTERFACE_RE = re.compile(r"(?:tap|fwbr|fwln|fwpr)(\d+)")
VM_INTERFACE_DETAIL_RE = re.compile(r"(?:tap|fwbr|fwln|fwpr)(\d+)i(\d+)")
LOG_FIELD_RE = re.compile(r"\b([A-Z]+)=([^\s]+)")
FIREWALL_LOG_TS_RE = re.compile(r"\b(\d{2}/[A-Za-z]{3}/\d{4}:\d{2}:\d{2}:\d{2} [+-]\d{4})\b")
ETH_P_IP = 0x0800
ETH_P_ALL = 0x0003
ETH_P_8021Q = 0x8100
ETH_P_8021AD = 0x88A8
IP_PROTOCOLS = {1: "icmp", 6: "tcp", 17: "udp"}


def read_text(path: str) -> str | None:
    try:
        return Path(path).read_text(encoding="utf-8").strip()
    except OSError:
        return None


def run_command(args: list[str], timeout: int = 5) -> tuple[int, str]:
    try:
        result = subprocess.run(args, capture_output=True, text=True, timeout=timeout, check=False)
        output = result.stdout.strip() or result.stderr.strip()
        return result.returncode, output
    except (OSError, subprocess.SubprocessError):
        return 127, ""


def executable_exists(path: str) -> bool:
    return Path(path).exists() and os.access(path, os.X_OK)


def flow_int(value: object, default: int = 0) -> int:
    try:
        return int(value) if value not in (None, "") else default
    except (TypeError, ValueError):
        return default


def collect_interfaces() -> list[dict[str, Any]]:
    interfaces = []
    for item in Path("/sys/class/net").iterdir() if Path("/sys/class/net").exists() else []:
        name = item.name
        if name == "lo":
            continue
        vm_match = VM_INTERFACE_RE.search(name)
        interfaces.append(
            {
                "name": name,
                "vmid": vm_match.group(1) if vm_match else None,
                "operstate": read_text(f"/sys/class/net/{name}/operstate") or "unknown",
                "rx_bytes": int(read_text(f"/sys/class/net/{name}/statistics/rx_bytes") or 0),
                "tx_bytes": int(read_text(f"/sys/class/net/{name}/statistics/tx_bytes") or 0),
                "rx_packets": int(read_text(f"/sys/class/net/{name}/statistics/rx_packets") or 0),
                "tx_packets": int(read_text(f"/sys/class/net/{name}/statistics/tx_packets") or 0),
            }
        )
    return interfaces


def interface_rank(name: str) -> int:
    if name.startswith("tap"):
        return 0
    if name.startswith("fwln"):
        return 1
    if name.startswith("fwpr"):
        return 2
    if name.startswith("fwbr"):
        return 3
    return 9


def collect_interface_traffic(interfaces: list[dict[str, Any]]) -> list[dict[str, Any]]:
    selected: dict[tuple[str, str], dict[str, Any]] = {}
    for interface in interfaces:
        vmid = interface.get("vmid")
        if not vmid:
            continue
        detail_match = VM_INTERFACE_DETAIL_RE.search(str(interface.get("name") or ""))
        nic = detail_match.group(2) if detail_match else "0"
        key = (str(vmid), nic)
        current = selected.get(key)
        if current and interface_rank(str(current.get("name") or "")) <= interface_rank(str(interface.get("name") or "")):
            continue
        selected[key] = interface

    traffic = []
    for (vmid, nic), interface in sorted(selected.items()):
        rx_bytes = int(interface.get("rx_bytes") or 0)
        tx_bytes = int(interface.get("tx_bytes") or 0)
        rx_packets = int(interface.get("rx_packets") or 0)
        tx_packets = int(interface.get("tx_packets") or 0)
        traffic.append(
            {
                "vmid": vmid,
                "nic": nic,
                "interface": interface.get("name"),
                "source": f"vm:{vmid}",
                "destination": "network",
                "protocol": "interface-counter",
                "rx_bytes": rx_bytes,
                "tx_bytes": tx_bytes,
                "bytes": rx_bytes + tx_bytes,
                "rx_packets": rx_packets,
                "tx_packets": tx_packets,
                "packets": rx_packets + tx_packets,
                "state": interface.get("operstate") or "unknown",
            }
        )
    return traffic


def selected_flow_interfaces(interfaces: list[dict[str, Any]]) -> list[dict[str, Any]]:
    selected: dict[tuple[str, str], dict[str, Any]] = {}
    for interface in interfaces:
        name = str(interface.get("name") or "")
        vmid = interface.get("vmid")
        if not vmid or not name.startswith(("tap", "fwln")):
            continue
        detail_match = VM_INTERFACE_DETAIL_RE.search(name)
        nic = detail_match.group(2) if detail_match else "0"
        key = (str(vmid), nic)
        current = selected.get(key)
        if current and interface_rank(str(current.get("name") or "")) <= interface_rank(name):
            continue
        selected[key] = interface
    return list(selected.values())


def ipv4_address(raw: bytes) -> str:
    return socket.inet_ntoa(raw)


def parse_packet_flow(packet: bytes) -> dict[str, Any] | None:
    if len(packet) < 34:
        return None
    offset = 12
    eth_type = struct.unpack("!H", packet[offset:offset + 2])[0]
    offset = 14
    while eth_type in {ETH_P_8021Q, ETH_P_8021AD}:
        if len(packet) < offset + 4:
            return None
        eth_type = struct.unpack("!H", packet[offset + 2:offset + 4])[0]
        offset += 4
    if eth_type != ETH_P_IP or len(packet) < offset + 20:
        return None

    version_ihl = packet[offset]
    version = version_ihl >> 4
    ihl = (version_ihl & 0x0F) * 4
    if version != 4 or ihl < 20 or len(packet) < offset + ihl:
        return None
    total_length = struct.unpack("!H", packet[offset + 2:offset + 4])[0]
    protocol_number = packet[offset + 9]
    protocol = IP_PROTOCOLS.get(protocol_number)
    if not protocol:
        return None
    source_ip = ipv4_address(packet[offset + 12:offset + 16])
    destination_ip = ipv4_address(packet[offset + 16:offset + 20])
    transport_offset = offset + ihl
    source_port = None
    destination_port = None
    if protocol in {"tcp", "udp"}:
        if len(packet) < transport_offset + 4:
            return None
        source_port, destination_port = struct.unpack("!HH", packet[transport_offset:transport_offset + 4])
    elif protocol == "icmp" and len(packet) >= transport_offset + 2:
        source_port = packet[transport_offset]
        destination_port = packet[transport_offset + 1]

    return {
        "source_ip": source_ip,
        "destination_ip": destination_ip,
        "protocol": protocol,
        "source_port": source_port,
        "destination_port": destination_port,
        "packets": 1,
        "bytes": total_length if total_length else max(len(packet) - offset, 0),
        "state": "observed",
    }


def collect_packet_flows(interfaces: list[dict[str, Any]], duration: int, limit: int = 500) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    flow_interfaces = selected_flow_interfaces(interfaces)
    sockets: dict[socket.socket, dict[str, Any]] = {}
    diagnostics: dict[str, Any] = {
        "collector": "linux-af-packet",
        "duration_seconds": duration,
        "interfaces_requested": [interface.get("name") for interface in flow_interfaces],
        "interfaces_opened": [],
        "errors": [],
    }
    for interface in flow_interfaces:
        name = str(interface.get("name") or "")
        try:
            packet_socket = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL))
            packet_socket.bind((name, 0))
            packet_socket.setblocking(False)
            sockets[packet_socket] = interface
            diagnostics["interfaces_opened"].append(name)
        except OSError as exc:
            diagnostics["errors"].append({"interface": name, "error": str(exc)})

    if not sockets:
        return [], diagnostics

    flows: dict[tuple[object, ...], dict[str, Any]] = {}
    deadline = time.monotonic() + max(duration, 1)
    try:
        while time.monotonic() < deadline:
            timeout = min(1.0, max(deadline - time.monotonic(), 0.0))
            readable, _, _ = select.select(list(sockets), [], [], timeout)
            for packet_socket in readable:
                interface = sockets[packet_socket]
                try:
                    packet = packet_socket.recv(65535)
                except OSError:
                    continue
                flow = parse_packet_flow(packet)
                if not flow:
                    continue
                name = str(interface.get("name") or "")
                vmid = str(interface.get("vmid") or "")
                detail_match = VM_INTERFACE_DETAIL_RE.search(name)
                nic = detail_match.group(2) if detail_match else "0"
                key = (
                    vmid,
                    nic,
                    flow["source_ip"],
                    flow["destination_ip"],
                    flow["protocol"],
                    flow.get("source_port"),
                    flow.get("destination_port"),
                )
                current = flows.get(key)
                if current:
                    current["packets"] += 1
                    current["bytes"] += int(flow["bytes"])
                    current["last_seen_at"] = datetime.now(timezone.utc).isoformat()
                    continue
                now = datetime.now(timezone.utc).isoformat()
                flow.update(
                    {
                        "vmid": vmid,
                        "nic": nic,
                        "interface": name,
                        "collector": "linux-af-packet",
                        "first_seen_at": now,
                        "last_seen_at": now,
                        "observed_at": now,
                    }
                )
                flows[key] = flow
                if len(flows) >= limit:
                    diagnostics["truncated"] = True
                    return sorted(flows.values(), key=lambda item: int(item.get("bytes") or 0), reverse=True), diagnostics
    finally:
        for packet_socket in sockets:
            packet_socket.close()

    return sorted(flows.values(), key=lambda item: int(item.get("bytes") or 0), reverse=True), diagnostics


def parse_conntrack_line(line: str) -> dict[str, Any] | None:
    parts = line.split()
    if len(parts) < 5 or parts[0] not in {"tcp", "udp", "icmp"}:
        return None
    protocol = parts[0]
    state = None
    if protocol == "tcp" and len(parts) > 3 and "=" not in parts[3]:
        state = parts[3]

    values: dict[str, list[str]] = {}
    for part in parts:
        if "=" not in part:
            continue
        key, value = part.split("=", 1)
        values.setdefault(key, []).append(value)

    src_values = values.get("src", [])
    dst_values = values.get("dst", [])
    if not src_values or not dst_values:
        return None

    packet_values = [int(value) for value in values.get("packets", []) if value.isdigit()]
    byte_values = [int(value) for value in values.get("bytes", []) if value.isdigit()]
    sport = values.get("sport", [None])[0]
    dport = values.get("dport", [None])[0]
    return {
        "source_ip": src_values[0],
        "destination_ip": dst_values[0],
        "protocol": protocol,
        "source_port": int(sport) if sport and sport.isdigit() else None,
        "destination_port": int(dport) if dport and dport.isdigit() else None,
        "packets": sum(packet_values),
        "bytes": sum(byte_values),
        "state": state,
    }


def collect_flows(limit: int = 500) -> list[dict[str, Any]]:
    code, output = run_command(["conntrack", "-L", "-o", "extended"], timeout=10)
    if code != 0 or not output:
        return []
    flows = []
    seen = set()
    for line in output.splitlines():
        flow = parse_conntrack_line(line)
        if not flow:
            continue
        key = (
            flow["source_ip"],
            flow["destination_ip"],
            flow["protocol"],
            flow.get("source_port"),
            flow.get("destination_port"),
        )
        if key in seen:
            continue
        seen.add(key)
        now = datetime.now(timezone.utc).isoformat()
        flow["first_seen_at"] = now
        flow["last_seen_at"] = now
        flow["observed_at"] = now
        flow["collector"] = flow.get("collector") or "conntrack"
        flows.append(flow)
        if len(flows) >= limit:
            break
    return flows


def firewall_log_seen_at(line: str) -> str:
    match = FIREWALL_LOG_TS_RE.search(line)
    if match:
        try:
            return datetime.strptime(match.group(1), "%d/%b/%Y:%H:%M:%S %z").astimezone(timezone.utc).isoformat()
        except ValueError:
            pass
    return datetime.now(timezone.utc).isoformat()


def parse_firewall_log_line(line: str) -> dict[str, Any] | None:
    fields = {key: value for key, value in LOG_FIELD_RE.findall(line)}
    source_ip = fields.get("SRC")
    destination_ip = fields.get("DST")
    protocol = fields.get("PROTO", "").lower()
    if not source_ip or not destination_ip or protocol not in {"tcp", "udp", "icmp"}:
        return None

    lowered = line.lower()
    decision = None
    if any(token in lowered for token in ("drop", "reject", "deny", "blocked")):
        decision = "blocked"
    elif any(token in lowered for token in ("accept", "allow")):
        decision = "allowed"
    if not decision:
        return None

    source_port = fields.get("SPT")
    destination_port = fields.get("DPT")
    length = fields.get("LEN")
    seen_at = firewall_log_seen_at(line)
    return {
        "source_ip": source_ip,
        "destination_ip": destination_ip,
        "protocol": protocol,
        "source_port": int(source_port) if source_port and source_port.isdigit() else None,
        "destination_port": int(destination_port) if destination_port and destination_port.isdigit() else None,
        "packets": 1,
        "bytes": int(length) if length and length.isdigit() else 0,
        "state": decision,
        "decision": decision,
        "collector": "firewall-log",
        "log_excerpt": line[-180:],
        "first_seen_at": seen_at,
        "last_seen_at": seen_at,
        "observed_at": seen_at,
    }


def firewall_log_lines_from_files(max_lines: int = 5000) -> tuple[list[str], list[dict[str, str]]]:
    lines: list[str] = []
    errors: list[dict[str, str]] = []
    for path in ("/var/log/pve-firewall.log", "/var/log/pve-firewall.log.1"):
        try:
            with open(path, "r", encoding="utf-8", errors="ignore") as handle:
                file_lines = handle.readlines()[-max_lines:]
                lines.extend(line.rstrip("\n") for line in file_lines)
        except OSError as exc:
            errors.append({"path": path, "error": str(exc)})
    return lines[-max_lines:], errors


def collect_firewall_log_flows(since_minutes: int = 5, limit: int = 500) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    diagnostics = {"collector": "journalctl-kernel+pve-firewall-log", "since_minutes": since_minutes, "errors": []}
    code, output = run_command(
        ["journalctl", "-k", "--since", f"-{max(since_minutes, 1)} min", "--no-pager", "-o", "cat"],
        timeout=10,
    )
    log_lines: list[str] = []
    if code == 0 and output:
        log_lines.extend(output.splitlines())
    else:
        diagnostics["errors"].append(output or "journalctl returned no firewall log output")
    file_lines, file_errors = firewall_log_lines_from_files(max(limit * 2, 1000))
    log_lines.extend(file_lines)
    diagnostics["file_errors"] = file_errors
    diagnostics["lines_scanned"] = len(log_lines)
    if not log_lines:
        return [], diagnostics

    flows: dict[tuple[object, ...], dict[str, Any]] = {}
    for line in log_lines:
        if "SRC=" not in line or "DST=" not in line:
            continue
        flow = parse_firewall_log_line(line)
        if not flow:
            continue
        key = (
            flow["source_ip"],
            flow["destination_ip"],
            flow["protocol"],
            flow.get("source_port"),
            flow.get("destination_port"),
            flow.get("decision"),
        )
        current = flows.get(key)
        if current:
            current["packets"] += 1
            current["bytes"] += int(flow.get("bytes") or 0)
            current["last_seen_at"] = flow.get("last_seen_at") or datetime.now(timezone.utc).isoformat()
            current["observed_at"] = current["last_seen_at"]
            continue
        flows[key] = flow
        if len(flows) >= limit:
            diagnostics["truncated"] = True
            break
    diagnostics["flow_count"] = len(flows)
    return sorted(flows.values(), key=lambda item: int(item.get("packets") or 0), reverse=True), diagnostics


def merge_flow_sources(*sources: list[dict[str, Any]], limit: int = 500) -> list[dict[str, Any]]:
    flows: dict[tuple[object, ...], dict[str, Any]] = {}
    for source in sources:
        for flow in source:
            key = (
                flow.get("source_ip"),
                flow.get("destination_ip"),
                flow.get("protocol"),
                flow.get("source_port"),
                flow.get("destination_port"),
                flow.get("decision") or flow.get("state") or "observed",
            )
            current = flows.get(key)
            if current:
                current["packets"] = int(current.get("packets") or 0) + int(flow.get("packets") or 0)
                current["bytes"] = int(current.get("bytes") or 0) + int(flow.get("bytes") or 0)
                current_seen = str(current.get("last_seen_at") or current.get("observed_at") or "")
                flow_seen = str(flow.get("last_seen_at") or flow.get("observed_at") or "")
                if flow_seen > current_seen:
                    current["last_seen_at"] = flow_seen
                    current["observed_at"] = flow_seen
                continue
            flows[key] = dict(flow)
    return sorted(
        flows.values(),
        key=lambda item: (
            1 if str(item.get("decision") or item.get("state") or "").lower() in {"blocked", "drop", "dropped", "reject", "rejected", "deny", "denied"} else 0,
            1 if item.get("collector") == "firewall-log" else 0,
            int(item.get("bytes") or 0),
            int(item.get("packets") or 0),
        ),
        reverse=True,
    )[:limit]


def collect_ebpf_flows(config: dict[str, Any], interfaces: list[dict[str, Any]], limit: int) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    binary = str(config.get("ebpf_binary") or "/opt/nexafabric-agent/nexafabric-ebpf")
    diagnostics: dict[str, Any] = {
        "collector": "ebpf",
        "enabled": bool(config.get("ebpf_collector", False)),
        "binary": binary,
        "errors": [],
    }
    if not diagnostics["enabled"]:
        return [], diagnostics
    if not executable_exists(binary):
        diagnostics["errors"].append("eBPF helper binary is not installed or not executable")
        return [], diagnostics

    flow_interfaces = [str(interface.get("name")) for interface in selected_flow_interfaces(interfaces)]
    if not flow_interfaces:
        diagnostics["errors"].append("no VM/LXC tap or firewall-link interfaces found")
        return [], diagnostics

    args = [
        binary,
        "--json",
        "--limit",
        str(limit),
        "--duration",
        str(int(config.get("ebpf_window_seconds", config.get("packet_flow_window_seconds", 10)))),
        "--interfaces",
        ",".join(flow_interfaces),
    ]
    code, output = run_command(args, timeout=int(config.get("ebpf_timeout_seconds", 15)))
    diagnostics["exit_code"] = code
    if code != 0 or not output:
        diagnostics["errors"].append(output or "eBPF helper returned no output")
        return [], diagnostics
    try:
        payload = json.loads(output)
    except json.JSONDecodeError as exc:
        diagnostics["errors"].append(f"eBPF helper returned invalid JSON: {exc}")
        diagnostics["output_excerpt"] = output[:300]
        return [], diagnostics

    flows = payload.get("flows", [])
    if not isinstance(flows, list):
        diagnostics["errors"].append("eBPF helper JSON has no flows array")
        return [], diagnostics
    normalized = []
    for flow in flows[:limit]:
        if not isinstance(flow, dict):
            continue
        source_ip = flow.get("source_ip")
        destination_ip = flow.get("destination_ip")
        if not source_ip or not destination_ip:
            continue
        normalized.append(
            {
                **flow,
                "source_ip": str(source_ip),
                "destination_ip": str(destination_ip),
                "protocol": str(flow.get("protocol") or "unknown").lower(),
                "source_port": flow_int(flow.get("source_port"), 0) or None,
                "destination_port": flow_int(flow.get("destination_port"), 0) or None,
                "packets": flow_int(flow.get("packets")),
                "bytes": flow_int(flow.get("bytes")),
                "state": str(flow.get("state") or "observed"),
                "collector": "ebpf",
            }
        )
    diagnostics["flow_count"] = len(normalized)
    if isinstance(payload.get("diagnostics"), dict):
        diagnostics["helper"] = payload["diagnostics"]
    return normalized, diagnostics


def collect_conntrack() -> dict[str, Any]:
    code, output = run_command(["conntrack", "-C"])
    if code == 0 and output.isdigit():
        return {"count": int(output), "source": "conntrack"}
    for path in ("/proc/net/nf_conntrack", "/proc/net/ip_conntrack"):
        try:
            with open(path, "r", encoding="utf-8", errors="ignore") as handle:
                return {"count": sum(1 for _ in handle), "source": path}
        except OSError:
            continue
    return {"count": None, "source": "unavailable"}


def collect_flow_diagnostics(
    conntrack: dict[str, Any],
    flow_count: int,
    packet_diagnostics: dict[str, Any] | None = None,
    firewall_log_diagnostics: dict[str, Any] | None = None,
    ebpf_diagnostics: dict[str, Any] | None = None,
) -> dict[str, Any]:
    code, conntrack_path = run_command(["sh", "-c", "command -v conntrack"])
    return {
        "flow_count": flow_count,
        "ebpf_collector": ebpf_diagnostics,
        "packet_collector": packet_diagnostics,
        "firewall_log_collector": firewall_log_diagnostics,
        "conntrack_binary": conntrack_path if code == 0 else None,
        "conntrack_count": conntrack.get("count"),
        "conntrack_source": conntrack.get("source"),
        "bridge_nf_call_iptables": read_text("/proc/sys/net/bridge/bridge-nf-call-iptables"),
        "bridge_nf_call_ip6tables": read_text("/proc/sys/net/bridge/bridge-nf-call-ip6tables"),
        "nf_conntrack_max": read_text("/proc/sys/net/netfilter/nf_conntrack_max"),
        "nf_conntrack_count": read_text("/proc/sys/net/netfilter/nf_conntrack_count"),
    }


def collect_firewall() -> dict[str, Any]:
    status = {}
    code, output = run_command(["systemctl", "is-active", "pve-firewall"])
    status["pve_firewall"] = output if code == 0 else "unknown"
    code, output = run_command(["nft", "-j", "list", "ruleset"], timeout=10)
    if code == 0 and output:
        status["nft_ruleset_sha256"] = hashlib.sha256(output.encode("utf-8")).hexdigest()
    else:
        status["nft_ruleset_sha256"] = None
    return status


def collect_payload(config: dict[str, Any]) -> dict[str, Any]:
    uptime = read_text("/proc/uptime")
    interfaces = collect_interfaces()
    flow_limit = int(config.get("flow_limit", 1500))
    packet_flows: list[dict[str, Any]] = []
    packet_diagnostics: dict[str, Any] | None = None
    ebpf_flows: list[dict[str, Any]] = []
    ebpf_diagnostics: dict[str, Any] | None = None
    firewall_log_flows: list[dict[str, Any]] = []
    firewall_log_diagnostics: dict[str, Any] | None = None
    ebpf_flows, ebpf_diagnostics = collect_ebpf_flows(config, interfaces, flow_limit)
    if bool(config.get("packet_flow_collector", True)):
        packet_flows, packet_diagnostics = collect_packet_flows(
            interfaces,
            int(config.get("packet_flow_window_seconds", 10)),
            flow_limit,
        )
    if bool(config.get("firewall_log_collector", True)):
        firewall_log_flows, firewall_log_diagnostics = collect_firewall_log_flows(
            int(config.get("firewall_log_window_minutes", 5)),
            flow_limit,
        )
    fallback_flows = packet_flows or collect_flows(flow_limit)
    flows = merge_flow_sources(ebpf_flows, fallback_flows, firewall_log_flows, limit=flow_limit)
    conntrack = collect_conntrack()
    return {
        "version": VERSION,
        "node_name": config.get("node_name"),
        "collected_at": datetime.now(timezone.utc).isoformat(),
        "hostname": socket.gethostname(),
        "kernel": platform.release(),
        "uptime_seconds": float(uptime.split()[0]) if uptime else None,
        "loadavg": list(os.getloadavg()) if hasattr(os, "getloadavg") else [],
        "interfaces": interfaces,
        "interface_traffic": collect_interface_traffic(interfaces),
        "flows": flows,
        "ebpf_flows": ebpf_flows,
        "conntrack": conntrack,
        "firewall": collect_firewall(),
        "extra": {
            "platform": platform.platform(),
            "flow_sources": {
                "ebpf": len(ebpf_flows),
                "packet": len(packet_flows),
                "conntrack_fallback": 0 if packet_flows else len(fallback_flows),
                "firewall_log": len(firewall_log_flows),
                "merged": len(flows),
            },
            "flow_diagnostics": collect_flow_diagnostics(conntrack, len(flows), packet_diagnostics, firewall_log_diagnostics, ebpf_diagnostics),
        },
    }


def normalized_api_url(config: dict[str, Any]) -> str:
    api_url = str(config["api_url"]).rstrip("/")
    while api_url.endswith("/api/v1/api/v1"):
        api_url = api_url.removesuffix("/api/v1")
    if not api_url.endswith("/api/v1"):
        api_url = f"{api_url}/api/v1"
    return api_url


def post_heartbeat(config: dict[str, Any], payload: dict[str, Any]) -> None:
    api_url = normalized_api_url(config)
    heartbeat_url = f"{api_url}/agents/heartbeat"
    data = json.dumps(payload).encode("utf-8")
    request = urllib.request.Request(
        heartbeat_url,
        data=data,
        headers={
            "Authorization": f"Bearer {config['token']}",
            "Content-Type": "application/json",
            "User-Agent": f"nexafabric-agent/{VERSION}",
        },
        method="POST",
    )
    context = None
    if not bool(config.get("verify_tls", True)):
        context = ssl._create_unverified_context()
    with urllib.request.urlopen(request, timeout=15, context=context) as response:
        response.read()


def load_config(path: str) -> dict[str, Any]:
    return json.loads(Path(path).read_text(encoding="utf-8"))


def main() -> int:
    parser = argparse.ArgumentParser(description="NexaFabric Proxmox node telemetry agent")
    parser.add_argument("--config", default="/etc/nexafabric-agent/config.json")
    parser.add_argument("--once", action="store_true")
    args = parser.parse_args()
    config = load_config(args.config)
    interval = int(config.get("interval_seconds", 30))

    while True:
        started_at = time.monotonic()
        payload = collect_payload(config)
        try:
            post_heartbeat(config, payload)
            print(f"heartbeat ok: {payload['collected_at']}", flush=True)
        except urllib.error.HTTPError as exc:
            body = exc.read().decode("utf-8", errors="replace")[:500]
            print(f"heartbeat failed: HTTP {exc.code} {exc.reason} url={exc.url} body={body}", flush=True)
        except (OSError, urllib.error.URLError) as exc:
            print(f"heartbeat failed: {exc} api_url={normalized_api_url(config)}", flush=True)
        if args.once:
            return 0
        elapsed = time.monotonic() - started_at
        time.sleep(max(interval - elapsed, 1))


if __name__ == "__main__":
    raise SystemExit(main())