feat: add kernel firewall log parsing to agent with blocked traffic detection and dashboard suspicious traffic enhancement

Add parse_firewall_log_line to extract SRC/DST/PROTO/SPT/DPT/LEN from kernel log lines with drop/reject/accept decision classification, implement collect_firewall_log_flows to parse journalctl -k output from last 5 minutes with flow aggregation by 5-tuple+decision, add merge_flow_sources to combine packet flows and firewall log flows with deduplication, extend agent config with

README.md

@@ -192,7 +192,7 @@ journalctl -u nexafabric-agent -f
 systemctl restart nexafabric-agent
 ```
 
-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.
+Agent version `0.2.1` 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, recent kernel firewall log drops/rejects, 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 and blocked traffic once guest IPs have been discovered. Blocked traffic visibility depends on Proxmox/kernel firewall logging being enabled for the rule or default drop that rejected the packet. 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:
 

backend/app/agent_assets/nexafabric-agent.py

@@ -18,9 +18,10 @@ from pathlib import Path
 from typing import Any
 
 
-VERSION = "0.2.0"
+VERSION = "0.2.1"
 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]+)")
 ETH_P_IP = 0x0800
 ETH_P_ALL = 0x0003
 ETH_P_8021Q = 0x8100
@@ -326,6 +327,100 @@ def collect_flows(limit: int = 500) -> list[dict[str, Any]]:
     return flows
 
 
+def parse_firewall_log_line(line: str) -> dict[str, Any] | None:
+    fields = {key: value for key, value in LOG_FIELD_RE.findall(line)}
+    source_ip = fields.get("SRC")
+    destination_ip = fields.get("DST")
+    protocol = fields.get("PROTO", "").lower()
+    if not source_ip or not destination_ip or protocol not in {"tcp", "udp", "icmp"}:
+        return None
+
+    lowered = line.lower()
+    decision = None
+    if any(token in lowered for token in ("drop", "reject", "deny", "blocked")):
+        decision = "blocked"
+    elif any(token in lowered for token in ("accept", "allow")):
+        decision = "allowed"
+    if not decision:
+        return None
+
+    source_port = fields.get("SPT")
+    destination_port = fields.get("DPT")
+    length = fields.get("LEN")
+    return {
+        "source_ip": source_ip,
+        "destination_ip": destination_ip,
+        "protocol": protocol,
+        "source_port": int(source_port) if source_port and source_port.isdigit() else None,
+        "destination_port": int(destination_port) if destination_port and destination_port.isdigit() else None,
+        "packets": 1,
+        "bytes": int(length) if length and length.isdigit() else 0,
+        "state": decision,
+        "decision": decision,
+        "collector": "firewall-log",
+        "log_excerpt": line[-500:],
+    }
+
+
+def collect_firewall_log_flows(since_minutes: int = 5, limit: int = 500) -> tuple[list[dict[str, Any]], dict[str, Any]]:
+    diagnostics = {"collector": "journalctl-kernel", "since_minutes": since_minutes, "errors": []}
+    code, output = run_command(
+        ["journalctl", "-k", "--since", f"-{max(since_minutes, 1)} min", "--no-pager", "-o", "cat"],
+        timeout=10,
+    )
+    if code != 0 or not output:
+        diagnostics["errors"].append(output or "journalctl returned no firewall log output")
+        return [], diagnostics
+
+    flows: dict[tuple[object, ...], dict[str, Any]] = {}
+    for line in output.splitlines():
+        if "SRC=" not in line or "DST=" not in line:
+            continue
+        flow = parse_firewall_log_line(line)
+        if not flow:
+            continue
+        key = (
+            flow["source_ip"],
+            flow["destination_ip"],
+            flow["protocol"],
+            flow.get("source_port"),
+            flow.get("destination_port"),
+            flow.get("decision"),
+        )
+        current = flows.get(key)
+        if current:
+            current["packets"] += 1
+            current["bytes"] += int(flow.get("bytes") or 0)
+            continue
+        flows[key] = flow
+        if len(flows) >= limit:
+            diagnostics["truncated"] = True
+            break
+    diagnostics["flow_count"] = len(flows)
+    return sorted(flows.values(), key=lambda item: int(item.get("packets") or 0), reverse=True), diagnostics
+
+
+def merge_flow_sources(*sources: list[dict[str, Any]], limit: int = 500) -> list[dict[str, Any]]:
+    flows: dict[tuple[object, ...], dict[str, Any]] = {}
+    for source in sources:
+        for flow in source:
+            key = (
+                flow.get("source_ip"),
+                flow.get("destination_ip"),
+                flow.get("protocol"),
+                flow.get("source_port"),
+                flow.get("destination_port"),
+                flow.get("decision") or flow.get("state") or "observed",
+            )
+            current = flows.get(key)
+            if current:
+                current["packets"] = int(current.get("packets") or 0) + int(flow.get("packets") or 0)
+                current["bytes"] = int(current.get("bytes") or 0) + int(flow.get("bytes") or 0)
+                continue
+            flows[key] = dict(flow)
+    return sorted(flows.values(), key=lambda item: int(item.get("bytes") or 0), reverse=True)[:limit]
+
+
 def collect_conntrack() -> dict[str, Any]:
     code, output = run_command(["conntrack", "-C"])
     if code == 0 and output.isdigit():
@@ -339,11 +434,17 @@ def collect_conntrack() -> dict[str, Any]:
     return {"count": None, "source": "unavailable"}
 
 
-def collect_flow_diagnostics(conntrack: dict[str, Any], flow_count: int, packet_diagnostics: dict[str, Any] | None = None) -> dict[str, Any]:
+def collect_flow_diagnostics(
+    conntrack: dict[str, Any],
+    flow_count: int,
+    packet_diagnostics: dict[str, Any] | None = None,
+    firewall_log_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,
+        "firewall_log_collector": firewall_log_diagnostics,
         "conntrack_binary": conntrack_path if code == 0 else None,
         "conntrack_count": conntrack.get("count"),
         "conntrack_source": conntrack.get("source"),
@@ -372,13 +473,20 @@ def collect_payload(config: dict[str, Any]) -> dict[str, Any]:
     flow_limit = int(config.get("flow_limit", 500))
     packet_flows: list[dict[str, Any]] = []
     packet_diagnostics: dict[str, Any] | None = None
+    firewall_log_flows: list[dict[str, Any]] = []
+    firewall_log_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)
+    if bool(config.get("firewall_log_collector", True)):
+        firewall_log_flows, firewall_log_diagnostics = collect_firewall_log_flows(
+            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)
     conntrack = collect_conntrack()
     return {
         "version": VERSION,
@@ -393,7 +501,10 @@ 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), packet_diagnostics)},
+        "extra": {
+            "platform": platform.platform(),
+            "flow_diagnostics": collect_flow_diagnostics(conntrack, len(flows), packet_diagnostics, firewall_log_diagnostics),
+        },
     }
 
 

backend/app/api/v1/router.py

@@ -526,6 +526,16 @@ def flow_policy_decision(active_matches: list[dict[str, object]], policy_matches
     return "observed"
 
 
+def raw_flow_decision(flow: TrafficFlow) -> str | None:
+    raw = flow.raw if isinstance(flow.raw, dict) else {}
+    decision = str(raw.get("decision") or flow.state or "").lower()
+    if decision in {"blocked", "drop", "dropped", "reject", "rejected", "deny", "denied"}:
+        return "blocked"
+    if decision in {"allowed", "accept", "accepted", "allow"}:
+        return "allowed"
+    return None
+
+
 def policy_read_payload(policy: Policy, deployment_status: dict[str, object] | None = None) -> dict[str, object]:
     return {
         "id": policy.id,
@@ -739,11 +749,14 @@ def dashboard_suspicious_traffic(db: Session) -> list[dict[str, int | str]]:
     events: dict[tuple[str, str, int], dict[str, int | str]] = {}
     for flow in db.scalars(select(TrafficFlow).order_by(TrafficFlow.updated_at.desc()).limit(500)).all():
         port = flow.destination_port or 0
-        if port not in sensitive_ports:
-            continue
         source_internal = bool(subnet_label_for_ip(subnets, flow.source_ip))
         destination_internal = bool(subnet_label_for_ip(subnets, flow.destination_ip)) or flow.destination_ip in workload_ips
-        if source_internal or not destination_internal:
+        if not destination_internal:
+            continue
+        decision = raw_flow_decision(flow)
+        if source_internal and decision != "blocked":
+            continue
+        if port not in sensitive_ports and decision != "blocked":
             continue
         key = (flow.source_ip, flow.destination_ip, port)
         event = events.setdefault(
@@ -754,8 +767,9 @@ def dashboard_suspicious_traffic(db: Session) -> list[dict[str, int | str]]:
                 "protocol": flow.protocol,
                 "port": port,
                 "bytes": 0,
-                "reason": sensitive_ports[port],
+                "reason": "Blocked by firewall" if decision == "blocked" else sensitive_ports[port],
-                "severity": "high" if port in {22, 3389, 445} else "medium",
+                "severity": "high" if port in {22, 3389, 445} or decision == "blocked" else "medium",
+                "decision": decision or "observed",
             },
         )
         event["bytes"] = int(event["bytes"]) + int(flow.bytes or 0)
@@ -1227,6 +1241,8 @@ cat > "$CONFIG_DIR/config.json" <<'JSON'
   "flow_limit": 500,
   "packet_flow_collector": true,
   "packet_flow_window_seconds": 10,
+  "firewall_log_collector": true,
+  "firewall_log_window_minutes": 5,
   "verify_tls": true
 }}
 JSON
@@ -1359,7 +1375,7 @@ def agent_heartbeat(payload: AgentHeartbeat, authorization: str | None = Header(
                 destination_port=flow_int(raw_flow.get("destination_port"), 0) or None,
                 bytes=flow_int(raw_flow.get("bytes")),
                 packets=flow_int(raw_flow.get("packets")),
-                state=str(raw_flow.get("state") or "") or None,
+                state=str(raw_flow.get("decision") or raw_flow.get("state") or "") or None,
                 observed_at=observed_at.replace(tzinfo=None) if observed_at.tzinfo else observed_at,
                 raw=raw_flow,
             )
@@ -1416,6 +1432,8 @@ async def workload_insights(workload_id: str, _: CurrentUser, db: Session = Depe
                 for policy in policies
                 if policy_matches_flow(db, policy, flow, ip_owners, workload_ips_by_id, workload.cluster_id)
             ]
+            raw_decision = raw_flow_decision(flow)
+            raw_payload = flow.raw if isinstance(flow.raw, dict) else {}
             traffic.append(
                 {
                     "source": flow_endpoint_label(source_owner, known_subnets, flow.source_ip),
@@ -1430,7 +1448,8 @@ async def workload_insights(workload_id: str, _: CurrentUser, db: Session = Depe
                     "bytes": flow.bytes,
                     "packets": flow.packets,
                     "state": flow.state,
-                    "decision": flow_policy_decision(matching_firewall_rules, matching_policies),
+                    "decision": raw_decision or flow_policy_decision(matching_firewall_rules, matching_policies),
+                    "collector": raw_payload.get("collector"),
                     "matching_firewall_rules": matching_firewall_rules,
                     "matching_audit_policies": [
                         policy for policy in matching_policies if str(policy.get("enforcement_mode")) == "audit"