diff --git a/backend/app/agent_assets/nexafabric-agent.py b/backend/app/agent_assets/nexafabric-agent.py index b9e0485..9db16cb 100644 --- a/backend/app/agent_assets/nexafabric-agent.py +++ b/backend/app/agent_assets/nexafabric-agent.py @@ -18,6 +18,7 @@ from typing import Any VERSION = "0.1.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+)") def read_text(path: str) -> str | None: @@ -30,7 +31,8 @@ def read_text(path: str) -> str | 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) - return result.returncode, result.stdout.strip() + output = result.stdout.strip() or result.stderr.strip() + return result.returncode, output except (OSError, subprocess.SubprocessError): return 127, "" @@ -56,6 +58,58 @@ def collect_interfaces() -> list[dict[str, Any]]: 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 parse_conntrack_line(line: str) -> dict[str, Any] | None: parts = line.split() if len(parts) < 5 or parts[0] not in {"tcp", "udp", "icmp"}: @@ -132,6 +186,20 @@ 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]: + code, conntrack_path = run_command(["sh", "-c", "command -v conntrack"]) + return { + "flow_count": flow_count, + "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"]) @@ -146,6 +214,9 @@ 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))) + conntrack = collect_conntrack() return { "version": VERSION, "node_name": config.get("node_name"), @@ -154,11 +225,12 @@ def collect_payload(config: dict[str, Any]) -> dict[str, Any]: "kernel": platform.release(), "uptime_seconds": float(uptime.split()[0]) if uptime else None, "loadavg": list(os.getloadavg()) if hasattr(os, "getloadavg") else [], - "interfaces": collect_interfaces(), - "flows": collect_flows(int(config.get("flow_limit", 500))), - "conntrack": collect_conntrack(), + "interfaces": interfaces, + "interface_traffic": collect_interface_traffic(interfaces), + "flows": flows, + "conntrack": conntrack, "firewall": collect_firewall(), - "extra": {"platform": platform.platform()}, + "extra": {"platform": platform.platform(), "flow_diagnostics": collect_flow_diagnostics(conntrack, len(flows))}, } diff --git a/backend/app/api/v1/router.py b/backend/app/api/v1/router.py index 538ddc7..befa5a7 100644 --- a/backend/app/api/v1/router.py +++ b/backend/app/api/v1/router.py @@ -894,6 +894,32 @@ def workload_insights(workload_id: str, _: CurrentUser, db: Session = Depends(ge "ip_addresses": [flow.source_ip, flow.destination_ip], } ) + if not traffic: + agent = db.get(NodeAgent, workload.node_id) + payload = agent.last_payload if agent and isinstance(agent.last_payload, dict) else {} + for item in payload.get("interface_traffic", []): + if not isinstance(item, dict) or str(item.get("vmid")) != str(workload.external_id): + continue + traffic.append( + { + "source": workload.name, + "destination": "network", + "interface": item.get("interface"), + "protocol": item.get("protocol") or "interface-counter", + "port": None, + "bytes": item.get("bytes") or 0, + "packets": item.get("packets") or 0, + "rx_bytes": item.get("rx_bytes") or 0, + "tx_bytes": item.get("tx_bytes") or 0, + "rx_packets": item.get("rx_packets") or 0, + "tx_packets": item.get("tx_packets") or 0, + "state": item.get("state") or "unknown", + "decision": "observed", + "observed_at": payload.get("collected_at"), + "ip_addresses": workload_ips, + "note": "Interface counter fallback. No host conntrack flows were available.", + } + ) audit_mode_notes = [ f"{policy.name} is in audit mode; matching traffic is logged without enforcement." for policy in policies diff --git a/backend/app/schemas/domain.py b/backend/app/schemas/domain.py index 34c2a22..5876780 100644 --- a/backend/app/schemas/domain.py +++ b/backend/app/schemas/domain.py @@ -203,6 +203,7 @@ class AgentHeartbeat(BaseModel): uptime_seconds: float | None = None loadavg: list[float] = [] interfaces: list[dict[str, Any]] = [] + interface_traffic: list[dict[str, Any]] = [] flows: list[dict[str, Any]] = [] conntrack: dict[str, Any] = {} firewall: dict[str, Any] = {} diff --git a/frontend/src/pages/Nodes.tsx b/frontend/src/pages/Nodes.tsx index bda8135..8ddb692 100644 --- a/frontend/src/pages/Nodes.tsx +++ b/frontend/src/pages/Nodes.tsx @@ -32,7 +32,11 @@ export function Nodes() { function agentFlows(node: Node) { const flows = node.agent?.last_payload?.flows; - return Array.isArray(flows) ? flows : []; + const interfaceTraffic = node.agent?.last_payload?.interface_traffic; + if (Array.isArray(flows) && flows.length) { + return flows; + } + return Array.isArray(interfaceTraffic) ? interfaceTraffic : []; } function agentInterfaces(node: Node) { @@ -166,13 +170,21 @@ export function Nodes() {
{agentFlows(detailNode).length ? agentFlows(detailNode).slice(0, 50).map((item, index) => { const flow = item as Record; + const isInterfaceCounter = flow.protocol === "interface-counter"; return (
-
{String(flow.source_ip)}:{String(flow.source_port ?? "")} {"->"} {String(flow.destination_ip)}:{String(flow.destination_port ?? "")}
-
{String(flow.protocol ?? "unknown")} · {String(flow.bytes ?? 0)} bytes · {String(flow.packets ?? 0)} packets · {String(flow.state ?? "unknown")}
+
+ {isInterfaceCounter + ? `VMID ${String(flow.vmid)} ${String(flow.interface ?? "")}` + : `${String(flow.source_ip)}:${String(flow.source_port ?? "")} -> ${String(flow.destination_ip)}:${String(flow.destination_port ?? "")}`} +
+
+ {String(flow.protocol ?? "unknown")} · {String(flow.bytes ?? 0)} bytes · {String(flow.packets ?? 0)} packets · {String(flow.state ?? "unknown")} +
+ {isInterfaceCounter ?
rx {String(flow.rx_bytes ?? 0)} · tx {String(flow.tx_bytes ?? 0)}
: null}
); - }) :
No conntrack flows were reported in the last heartbeat.
} + }) :
No conntrack flows or interface counters were reported in the last heartbeat.
}
diff --git a/frontend/src/pages/Workloads.tsx b/frontend/src/pages/Workloads.tsx index 6f4bf9f..f7ee2c3 100644 --- a/frontend/src/pages/Workloads.tsx +++ b/frontend/src/pages/Workloads.tsx @@ -63,7 +63,9 @@ export function Workloads() {
{String(flow.source)} → {String(flow.destination)}
{String(flow.protocol)}:{String(flow.port)} · {String(flow.bytes)} bytes · {String(flow.decision)}
+ {flow.protocol === "interface-counter" ?
Interface: {String(flow.interface ?? "unknown")} · rx {String(flow.rx_bytes ?? 0)} · tx {String(flow.tx_bytes ?? 0)}
: null} {Array.isArray(flow.ip_addresses) ?
IPs: {flow.ip_addresses.join(", ")}
: null} + {flow.note ?
{String(flow.note)}
: null}
)) :
No real traffic telemetry has been collected yet. Install the node agent or enable a flow source to populate this section.
}