From 35ffcb676803e2012693fb575c0d8e56ba1f88e0 Mon Sep 17 00:00:00 2001 From: nessi Date: Thu, 9 Jul 2026 21:07:03 +0200 Subject: [PATCH] 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 | 2 +- backend/app/agent_assets/nexafabric-agent.py | 119 ++++++++++++++++++- backend/app/api/v1/router.py | 33 +++-- 3 files changed, 142 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index adea9c6..48dbf4b 100644 --- a/README.md +++ b/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: diff --git a/backend/app/agent_assets/nexafabric-agent.py b/backend/app/agent_assets/nexafabric-agent.py index 5051adc..deaceaa 100644 --- a/backend/app/agent_assets/nexafabric-agent.py +++ b/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), + }, } diff --git a/backend/app/api/v1/router.py b/backend/app/api/v1/router.py index 29413a7..b2955df 100644 --- a/backend/app/api/v1/router.py +++ b/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], - "severity": "high" if port in {22, 3389, 445} else "medium", + "reason": "Blocked by firewall" if decision == "blocked" else sensitive_ports[port], + "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"