nessi
e6c9a92ff0
Add EBPF_HELPER_CONTRACT.md documenting helper binary invocation with --json/--limit/--duration/--interfaces parameters and expected JSON output format with flows/diagnostics, implement collect_ebpf_flows to invoke helper binary with configurable timeout/window/interfaces and normalize flow fields (source_ip/destination_ip/protocol/ports/packets/bytes/state), add executable_exists and flow_int helpers for binary validation
700 lines
26 KiB
Python
700 lines
26 KiB
Python
#!/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]+)")
|
|
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"])
|
|
continue
|
|
flow.update(
|
|
{
|
|
"vmid": vmid,
|
|
"nic": nic,
|
|
"interface": name,
|
|
"collector": "linux-af-packet",
|
|
}
|
|
)
|
|
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)
|
|
flows.append(flow)
|
|
if len(flows) >= limit:
|
|
break
|
|
return flows
|
|
|
|
|
|
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")
|
|
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:],
|
|
}
|
|
|
|
|
|
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)
|
|
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)
|
|
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())
|