nessi/NexaFabric
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add eBPF flow collector with helper binary contract and agent integration

Browse Source 
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
This commit is contained in:
nessi
2026-07-10 14:15:15 +02:00
parent af68949e07
commit e6c9a92ff0
4 changed files with 148 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/agent_assets/nexafabric-agent.py
+101 -3
View File
@@ -18,7 +18,7 @@ from pathlib import Path
from typing import Any
VERSION = "0.2.3"
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]+)")
@@ -45,6 +45,17 @@ def run_command(args: list[str], timeout: int = 5) -> tuple[int, str]:
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 []:
@@ -451,6 +462,79 @@ def merge_flow_sources(*sources: list[dict[str, Any]], limit: int = 500) -> list
)[: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():
@@ -469,10 +553,12 @@ def collect_flow_diagnostics(
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,
@@ -503,8 +589,11 @@ def collect_payload(config: dict[str, Any]) -> dict[str, Any]:
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,
@@ -516,7 +605,8 @@ def collect_payload(config: dict[str, Any]) -> dict[str, Any]:
int(config.get("firewall_log_window_minutes", 5)),
flow_limit,
)
flows = merge_flow_sources(packet_flows or collect_flows(flow_limit), firewall_log_flows, limit=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,
@@ -529,11 +619,19 @@ def collect_payload(config: dict[str, Any]) -> dict[str, Any]:
"interfaces": interfaces,
"interface_traffic": collect_interface_traffic(interfaces),
"flows": flows,
"ebpf_flows": ebpf_flows,
"conntrack": conntrack,
"firewall": collect_firewall(),
"extra": {
"platform": platform.platform(),
"flow_diagnostics": collect_flow_diagnostics(conntrack, len(flows), packet_diagnostics, firewall_log_diagnostics),
"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),
},
}