diff --git a/backend/app/agent_assets/EBPF_HELPER_CONTRACT.md b/backend/app/agent_assets/EBPF_HELPER_CONTRACT.md new file mode 100644 index 0000000..69b077a --- /dev/null +++ b/backend/app/agent_assets/EBPF_HELPER_CONTRACT.md @@ -0,0 +1,37 @@ +# NexaFabric eBPF helper contract + +Agent 0.3.0 can call an optional helper binary at `/opt/nexafabric-agent/nexafabric-ebpf`. + +The helper is invoked as: + +```sh +nexafabric-ebpf --json --limit 1500 --duration 10 --interfaces tap100i0,fwln100i0 +``` + +It must print JSON to stdout: + +```json +{ + "flows": [ + { + "source_ip": "172.16.0.10", + "destination_ip": "172.16.0.20", + "protocol": "tcp", + "source_port": 443, + "destination_port": 53020, + "packets": 10, + "bytes": 14800, + "vmid": "100", + "interface": "tap100i0", + "direction": "ingress", + "state": "observed" + } + ], + "diagnostics": { + "attach_mode": "tc", + "interfaces_attached": ["tap100i0"] + } +} +``` + +The Python agent merges these flows with firewall-log, packet, and conntrack fallback collectors. diff --git a/backend/app/agent_assets/nexafabric-agent.py b/backend/app/agent_assets/nexafabric-agent.py index 5fb3e51..e975b79 100644 --- a/backend/app/agent_assets/nexafabric-agent.py +++ b/backend/app/agent_assets/nexafabric-agent.py @@ -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), }, } diff --git a/backend/app/api/v1/router.py b/backend/app/api/v1/router.py index 29ba63e..018d6bf 100644 --- a/backend/app/api/v1/router.py +++ b/backend/app/api/v1/router.py @@ -663,6 +663,11 @@ def compact_agent_payload(payload: AgentHeartbeat) -> dict: value["flow_count"] = len(flows) value["flows"] = flows[:50] value["flows_truncated"] = len(flows) > 50 + ebpf_flows = value.get("ebpf_flows") + if isinstance(ebpf_flows, list): + value["ebpf_flow_count"] = len(ebpf_flows) + value["ebpf_flows"] = ebpf_flows[:50] + value["ebpf_flows_truncated"] = len(ebpf_flows) > 50 return value @@ -1417,6 +1422,10 @@ cat > "$CONFIG_DIR/config.json" <<'JSON' "node_name": "{node.name}", "interval_seconds": 30, "flow_limit": 1500, + "ebpf_collector": true, + "ebpf_binary": "/opt/nexafabric-agent/nexafabric-ebpf", + "ebpf_window_seconds": 10, + "ebpf_timeout_seconds": 15, "packet_flow_collector": true, "packet_flow_window_seconds": 10, "firewall_log_collector": true, diff --git a/backend/app/schemas/domain.py b/backend/app/schemas/domain.py index d81a38b..00f9ebe 100644 --- a/backend/app/schemas/domain.py +++ b/backend/app/schemas/domain.py @@ -238,6 +238,7 @@ class AgentHeartbeat(BaseModel): interfaces: list[dict[str, Any]] = [] interface_traffic: list[dict[str, Any]] = [] flows: list[dict[str, Any]] = [] + ebpf_flows: list[dict[str, Any]] = [] conntrack: dict[str, Any] = {} firewall: dict[str, Any] = {} extra: dict[str, Any] = {}