feat: add AF_PACKET flow collector to agent for real VM traffic visibility with IPv4 TCP/UDP/ICMP flow extraction

Add packet flow collector in agent v0.2.0 using Linux AF_PACKET sockets to capture and aggregate IPv4 TCP/UDP/ICMP flows from VM interfaces (tap/fwln) with configurable window/limit, implement parse_packet_flow to extract 5-tuple from raw Ethernet frames with VLAN tag handling, add selected_flow_interfaces to choose best interface per VM NIC for packet capture, include packet collector
2026-07-09 15:18:35 +02:00
parent 3bfd77a74a
commit 4ceb4489c5
3 changed files with 189 additions and 11 deletions
@@ -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__":