diff --git a/README.md b/README.md index 0922dd7..9937715 100644 --- a/README.md +++ b/README.md @@ -156,16 +156,15 @@ Proxmox inventory and guest agent data are enough for: - Static LXC IP discovery. - Policy matching and firewall previews. -Actual traffic flow visibility, top talkers, byte counters, and per-workload traffic history require an additional telemetry source. Proxmox VE does not provide full flow telemetry for every VM through the basic inventory API. +Actual traffic flow visibility, top talkers, byte counters, and per-workload traffic history require the NexaFabric node agent or another telemetry source. Proxmox VE does not provide full flow telemetry for every VM through the basic inventory API. -Supported or planned options: +Supported options: -- NexaFabric node agent on Proxmox nodes to read host interface counters, VM/LXC interface hints, conntrack flows, nftables ruleset state, and pve-firewall status. +- NexaFabric node agent on Proxmox nodes to read VM/LXC interface hints, host interface counters, real IPv4 TCP/UDP/ICMP flows from Linux VM interfaces, conntrack flows when available, nftables ruleset state, and pve-firewall status. - Open vSwitch with sFlow/NetFlow/IPFIX exported to a collector. - Router/firewall flow exports from pfSense, OPNsense, FRR/BGP edge devices, or physical switches. -- eBPF or host-level telemetry in future agent builds. -Until such a source is configured, NexaFabric will show `No flow telemetry collected yet` instead of fake traffic. +Until such a source is configured, NexaFabric will show `No flow telemetry collected yet` instead of fake traffic. If the node agent can see VM interface counters but no packet flows, NexaFabric displays the counters as an explicitly marked fallback. ### 8. Install The Node Agent @@ -192,7 +191,19 @@ journalctl -u nexafabric-agent -f systemctl restart nexafabric-agent ``` -The agent reports host/interface counters, VMID hints from Proxmox interface names, conntrack flow records, pve-firewall status, and an nftables ruleset hash. NexaFabric maps flow source/destination IPs back to workloads through IPAM, so VM/LXC details can show observed traffic once guest IPs have been discovered. The agent does not enforce policies itself; Proxmox firewall rule apply remains API-driven through NexaFabric. +Agent version `0.2.0` reports host/interface counters, VMID hints from Proxmox interface names, real packet-derived IPv4 TCP/UDP/ICMP flows from VM interfaces, conntrack flow records when available, pve-firewall status, and an nftables ruleset hash. NexaFabric maps flow source/destination IPs back to workloads through IPAM, so VM/LXC details can show observed traffic once guest IPs have been discovered. The agent does not enforce policies itself; Proxmox firewall rule apply remains API-driven through NexaFabric. + +The default agent config enables the packet flow collector: + +```json +{ + "packet_flow_collector": true, + "packet_flow_window_seconds": 10, + "flow_limit": 500 +} +``` + +The collector runs as root through the Linux `AF_PACKET` interface and attaches to Proxmox VM interfaces such as `tap100i0` and `fwln100i0`. It aggregates locally before sending data to NexaFabric; packet payloads are not stored or uploaded. ### 9. Troubleshooting Proxmox Integration diff --git a/backend/app/agent_assets/nexafabric-agent.py b/backend/app/agent_assets/nexafabric-agent.py index 9db16cb..5051adc 100644 --- a/backend/app/agent_assets/nexafabric-agent.py +++ b/backend/app/agent_assets/nexafabric-agent.py @@ -5,8 +5,10 @@ import json import os import platform import re +import select import socket import ssl +import struct import subprocess import time import urllib.error @@ -16,9 +18,14 @@ from pathlib import Path from typing import Any -VERSION = "0.1.0" +VERSION = "0.2.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+)") +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: @@ -110,6 +117,152 @@ def collect_interface_traffic(interfaces: list[dict[str, Any]]) -> list[dict[str 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"}: @@ -186,10 +339,11 @@ def collect_conntrack() -> dict[str, Any]: return {"count": None, "source": "unavailable"} -def collect_flow_diagnostics(conntrack: dict[str, Any], flow_count: int) -> dict[str, Any]: +def collect_flow_diagnostics(conntrack: dict[str, Any], flow_count: int, packet_diagnostics: dict[str, Any] | None = None) -> dict[str, Any]: code, conntrack_path = run_command(["sh", "-c", "command -v conntrack"]) return { "flow_count": flow_count, + "packet_collector": packet_diagnostics, "conntrack_binary": conntrack_path if code == 0 else None, "conntrack_count": conntrack.get("count"), "conntrack_source": conntrack.get("source"), @@ -215,7 +369,16 @@ def collect_firewall() -> dict[str, Any]: def collect_payload(config: dict[str, Any]) -> dict[str, Any]: uptime = read_text("/proc/uptime") interfaces = collect_interfaces() - flows = collect_flows(int(config.get("flow_limit", 500))) + flow_limit = int(config.get("flow_limit", 500)) + packet_flows: list[dict[str, Any]] = [] + packet_diagnostics: dict[str, Any] | None = None + 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, + ) + flows = packet_flows or collect_flows(flow_limit) conntrack = collect_conntrack() return { "version": VERSION, @@ -230,7 +393,7 @@ def collect_payload(config: dict[str, Any]) -> dict[str, Any]: "flows": flows, "conntrack": conntrack, "firewall": collect_firewall(), - "extra": {"platform": platform.platform(), "flow_diagnostics": collect_flow_diagnostics(conntrack, len(flows))}, + "extra": {"platform": platform.platform(), "flow_diagnostics": collect_flow_diagnostics(conntrack, len(flows), packet_diagnostics)}, } @@ -277,6 +440,7 @@ def main() -> int: interval = int(config.get("interval_seconds", 30)) while True: + started_at = time.monotonic() payload = collect_payload(config) try: post_heartbeat(config, payload) @@ -288,7 +452,8 @@ def main() -> int: print(f"heartbeat failed: {exc} api_url={normalized_api_url(config)}", flush=True) if args.once: return 0 - time.sleep(interval) + elapsed = time.monotonic() - started_at + time.sleep(max(interval - elapsed, 1)) if __name__ == "__main__": diff --git a/backend/app/api/v1/router.py b/backend/app/api/v1/router.py index befa5a7..2c23a08 100644 --- a/backend/app/api/v1/router.py +++ b/backend/app/api/v1/router.py @@ -706,6 +706,8 @@ cat > "$CONFIG_DIR/config.json" <<'JSON' "node_name": "{node.name}", "interval_seconds": 30, "flow_limit": 500, + "packet_flow_collector": true, + "packet_flow_window_seconds": 10, "verify_tls": true }} JSON