From 6d5dc310dfb51b65f2754b1522022c2973fa8afd Mon Sep 17 00:00:00 2001 From: nessi Date: Thu, 9 Jul 2026 19:47:39 +0200 Subject: [PATCH] feat: add flow-level firewall rule and policy matching with decision classification and audit mode visualization Add firewall_rule_matches_flow to check if active rules match traffic flows using protocol/port/IP/direction matching with enable status validation, implement policy_matches_flow to evaluate policy definitions against flows with workload/network endpoint resolution and protocol/port matching, add flow_policy_decision to determine final decision from active rules and policies with audit --- backend/app/api/v1/router.py | 224 ++++++++++++++++++++++++++++++- frontend/src/pages/Workloads.tsx | 91 +++++++++++++ 2 files changed, 312 insertions(+), 3 deletions(-) diff --git a/backend/app/api/v1/router.py b/backend/app/api/v1/router.py index 9cd3ff4..941aa28 100644 --- a/backend/app/api/v1/router.py +++ b/backend/app/api/v1/router.py @@ -333,6 +333,199 @@ async def active_firewall_rules_for_workload(db: Session, workload: Workload) -> ] +def firewall_rule_decision(action: object) -> str: + normalized = str(action or "").lower() + if normalized in {"accept", "allow"}: + return "allowed" + if normalized in {"drop", "reject", "deny"}: + return "blocked" + return "observed" + + +def port_matches(rule_value: object, flow_port: int | None) -> bool: + if rule_value in (None, "", "any"): + return True + if flow_port is None: + return False + for raw_part in str(rule_value).split(","): + part = raw_part.strip() + if not part: + continue + separator = ":" if ":" in part else "-" if "-" in part else "" + if separator: + start, end = part.split(separator, 1) + try: + if int(start) <= flow_port <= int(end): + return True + except ValueError: + continue + continue + try: + if int(part) == flow_port: + return True + except ValueError: + continue + return False + + +def ip_value_matches(rule_value: object, flow_ip: str) -> bool: + if rule_value in (None, "", "any"): + return True + try: + address = ip_address(flow_ip) + except ValueError: + return False + for raw_part in str(rule_value).split(","): + part = raw_part.strip() + if not part: + continue + try: + if "/" in part: + if address in ip_network(part, strict=False): + return True + elif address == ip_address(part): + return True + except ValueError: + if part == flow_ip: + return True + return False + + +def firewall_rule_matches_flow(rule: dict[str, object], flow: TrafficFlow, workload_ips: set[str]) -> bool: + enabled = str(rule.get("enable", "1")).lower() + if enabled in {"0", "false", "no"}: + return False + rule_type = str(rule.get("type") or "").lower() + if rule_type == "in" and flow.destination_ip not in workload_ips: + return False + if rule_type == "out" and flow.source_ip not in workload_ips: + return False + proto = str(rule.get("proto") or "any").lower() + if proto not in {"", "any"} and proto != str(flow.protocol or "").lower(): + return False + if not ip_value_matches(rule.get("source"), flow.source_ip): + return False + if not ip_value_matches(rule.get("dest"), flow.destination_ip): + return False + if not port_matches(rule.get("sport"), flow.source_port): + return False + if not port_matches(rule.get("dport"), flow.destination_port): + return False + return True + + +def firewall_rule_flow_payload(rule: dict[str, object]) -> dict[str, object]: + return { + "pos": rule.get("pos"), + "type": rule.get("type"), + "action": rule.get("action"), + "proto": rule.get("proto"), + "source": rule.get("source"), + "dest": rule.get("dest"), + "sport": rule.get("sport"), + "dport": rule.get("dport"), + "comment": rule.get("comment"), + "decision": firewall_rule_decision(rule.get("action")), + "managed_by_nexafabric": bool(rule.get("managed_by_nexafabric")), + } + + +def endpoint_ref_matches_flow_side( + db: Session, + ref: object, + flow_ip: str, + side_workload: Workload | None, + side_workload_ips: set[str], + cluster_id: str, +) -> bool: + value = str(ref or "any") + if value == "any": + return True + if value.startswith("workload:"): + workload_id = value.removeprefix("workload:") + return bool(side_workload and side_workload.id == workload_id and flow_ip in side_workload_ips) + if value.startswith("network:"): + network_name = value.removeprefix("network:") + network = db.scalar(select(Network).where(Network.cluster_id == cluster_id, Network.name == network_name)) + if not network: + return False + subnets = db.scalars(select(Subnet).where(Subnet.network_id == network.id)).all() + return any(ip_value_matches(subnet.cidr, flow_ip) for subnet in subnets) + if is_ip_or_cidr(value): + return ip_value_matches(value, flow_ip) + return False + + +def policy_matches_flow( + db: Session, + policy: Policy, + flow: TrafficFlow, + ip_owners: dict[str, Workload | None], + workload_ips_by_id: dict[str, set[str]], + cluster_id: str, +) -> bool: + if not policy.enabled: + return False + definition = normalized_policy_definition(policy.definition or {}) + policy_protocol = str(definition.get("protocol") or "any").lower() + flow_protocol = str(flow.protocol or "").lower() + if policy_protocol not in {"any", flow_protocol} and not ( + policy_protocol in {"tcp/udp", "tcp & udp", "tcp_udp"} and flow_protocol in {"tcp", "udp"} + ): + return False + if not port_matches(definition.get("ports") or definition.get("port"), flow.destination_port): + return False + source_workload = ip_owners.get(flow.source_ip) + destination_workload = ip_owners.get(flow.destination_ip) + if not endpoint_ref_matches_flow_side( + db, + definition.get("source"), + flow.source_ip, + source_workload, + workload_ips_by_id.get(source_workload.id, set()) if source_workload else set(), + cluster_id, + ): + return False + return endpoint_ref_matches_flow_side( + db, + definition.get("destination"), + flow.destination_ip, + destination_workload, + workload_ips_by_id.get(destination_workload.id, set()) if destination_workload else set(), + cluster_id, + ) + + +def policy_flow_payload(policy: Policy) -> dict[str, object]: + definition = normalized_policy_definition(policy.definition or {}) + action = str(definition.get("action") or "allow").lower() + mode = str(definition.get("enforcement_mode") or policy.enforcement_mode or "enforced").lower() + block = action in {"deny", "drop", "reject", "block"} + return { + "id": policy.id, + "name": policy.name, + "version": policy.version, + "enforcement_mode": mode, + "action": action, + "protocol": definition.get("protocol") or "any", + "ports": definition.get("ports") or definition.get("port"), + "description": definition.get("description"), + "decision": ("would_block" if block else "would_allow") if mode == "audit" else ("blocked" if block else "allowed"), + } + + +def flow_policy_decision(active_matches: list[dict[str, object]], policy_matches: list[dict[str, object]]) -> str: + for match in active_matches: + decision = str(match.get("decision") or "") + if decision in {"blocked", "allowed"}: + return decision + for match in policy_matches: + decision = str(match.get("decision") or "") + if decision in {"blocked", "allowed", "would_block", "would_allow"}: + return decision + return "observed" + + def dashboard_top_talkers(db: Session) -> list[dict[str, int | str]]: totals: dict[str, int] = {} workloads = db.scalars(select(Workload)).all() @@ -1037,13 +1230,20 @@ async def workload_insights(workload_id: str, _: CurrentUser, db: Session = Depe ).all() assigned_ips = db.scalars(select(IpAddress).where(IpAddress.workload_id == workload.id).order_by(IpAddress.address)).all() workload_ips = [address.address for address in assigned_ips] + all_assigned_ips = db.scalars(select(IpAddress).where(IpAddress.workload_id.is_not(None))).all() ip_owners = { address.address: db.get(Workload, address.workload_id) - for address in db.scalars(select(IpAddress).where(IpAddress.workload_id.is_not(None))).all() + for address in all_assigned_ips } + workload_ips_by_id: dict[str, set[str]] = {} + for address in all_assigned_ips: + if address.workload_id: + workload_ips_by_id.setdefault(address.workload_id, set()).add(address.address) known_subnets = db.scalars(select(Subnet).order_by(Subnet.cidr)).all() + active_firewall_rules = await active_firewall_rules_for_workload(db, workload) traffic = [] if workload_ips: + workload_ip_set = set(workload_ips) flows = db.scalars( select(TrafficFlow) .where((TrafficFlow.source_ip.in_(workload_ips)) | (TrafficFlow.destination_ip.in_(workload_ips))) @@ -1053,6 +1253,16 @@ async def workload_insights(workload_id: str, _: CurrentUser, db: Session = Depe for flow in flows: source_owner = ip_owners.get(flow.source_ip) destination_owner = ip_owners.get(flow.destination_ip) + matching_firewall_rules = [ + firewall_rule_flow_payload(rule) + for rule in active_firewall_rules + if "error" not in rule and firewall_rule_matches_flow(rule, flow, workload_ip_set) + ] + matching_policies = [ + policy_flow_payload(policy) + for policy in policies + if policy_matches_flow(db, policy, flow, ip_owners, workload_ips_by_id, workload.cluster_id) + ] traffic.append( { "source": flow_endpoint_label(source_owner, known_subnets, flow.source_ip), @@ -1067,7 +1277,12 @@ async def workload_insights(workload_id: str, _: CurrentUser, db: Session = Depe "bytes": flow.bytes, "packets": flow.packets, "state": flow.state, - "decision": "observed", + "decision": flow_policy_decision(matching_firewall_rules, matching_policies), + "matching_firewall_rules": matching_firewall_rules, + "matching_audit_policies": [ + policy for policy in matching_policies if str(policy.get("enforcement_mode")) == "audit" + ], + "matching_policies": matching_policies, "observed_at": flow.observed_at.isoformat() if flow.observed_at else None, "ip_addresses": [flow.source_ip, flow.destination_ip], } @@ -1095,6 +1310,9 @@ async def workload_insights(workload_id: str, _: CurrentUser, db: Session = Depe "tx_packets": item.get("tx_packets") or 0, "state": item.get("state") or "unknown", "decision": "observed", + "matching_firewall_rules": [], + "matching_audit_policies": [], + "matching_policies": [], "observed_at": payload.get("collected_at"), "ip_addresses": workload_ips, "note": "Interface counter fallback. No host conntrack flows were available.", @@ -1110,7 +1328,7 @@ async def workload_insights(workload_id: str, _: CurrentUser, db: Session = Depe workload=workload, assigned_ips=[ip_address_payload(db, address) for address in assigned_ips], traffic=traffic, - active_firewall_rules=await active_firewall_rules_for_workload(db, workload), + active_firewall_rules=active_firewall_rules, matching_policies=policies, effective_decision=decision, audit_mode_notes=audit_mode_notes, diff --git a/frontend/src/pages/Workloads.tsx b/frontend/src/pages/Workloads.tsx index c0baaa0..02ba0db 100644 --- a/frontend/src/pages/Workloads.tsx +++ b/frontend/src/pages/Workloads.tsx @@ -23,8 +23,29 @@ type TrafficSummary = { interfaceName: string; note: string; ipAddresses: string[]; + matchingFirewallRules: Array>; + matchingAuditPolicies: Array>; + matchingPolicies: Array>; }; +function records(value: unknown) { + return Array.isArray(value) ? (value.filter((item) => item && typeof item === "object") as Array>) : []; +} + +function mergeRecords(left: Array>, right: Array>) { + const seen = new Set(); + const merged: Array> = []; + for (const item of [...left, ...right]) { + const key = String(item.id ?? item.pos ?? item.comment ?? JSON.stringify(item)); + if (seen.has(key)) { + continue; + } + seen.add(key); + merged.push(item); + } + return merged; +} + function formatBytes(value: unknown) { const bytes = Number(value ?? 0); if (!Number.isFinite(bytes) || bytes <= 0) { @@ -49,11 +70,20 @@ function summarizeTraffic(traffic: Array>) { const bytes = Number(flow.bytes ?? 0); const packets = Number(flow.packets ?? 0); const ipAddresses = Array.isArray(flow.ip_addresses) ? flow.ip_addresses.map(String) : []; + const matchingFirewallRules = records(flow.matching_firewall_rules); + const matchingAuditPolicies = records(flow.matching_audit_policies); + const matchingPolicies = records(flow.matching_policies); if (existing) { existing.bytes += Number.isFinite(bytes) ? bytes : 0; existing.packets += Number.isFinite(packets) ? packets : 0; existing.count += 1; existing.ipAddresses = Array.from(new Set([...existing.ipAddresses, ...ipAddresses])); + existing.matchingFirewallRules = mergeRecords(existing.matchingFirewallRules, matchingFirewallRules); + existing.matchingAuditPolicies = mergeRecords(existing.matchingAuditPolicies, matchingAuditPolicies); + existing.matchingPolicies = mergeRecords(existing.matchingPolicies, matchingPolicies); + if (existing.decision === "observed" && String(flow.decision ?? "observed") !== "observed") { + existing.decision = String(flow.decision ?? "observed"); + } continue; } summaries.set(key, { @@ -71,6 +101,9 @@ function summarizeTraffic(traffic: Array>) { interfaceName: String(flow.interface ?? ""), note: String(flow.note ?? ""), ipAddresses, + matchingFirewallRules, + matchingAuditPolicies, + matchingPolicies, }); } return Array.from(summaries.values()).sort((left, right) => right.bytes - left.bytes); @@ -154,6 +187,59 @@ function ruleLabel(rule: Record) { return `${type} ${action} ${proto}${port}`; } +function policyLabel(policy: Record) { + const name = String(policy.name ?? "Policy"); + const mode = String(policy.enforcement_mode ?? "enforced"); + const decision = String(policy.decision ?? "observed").replace("_", " "); + const protocol = String(policy.protocol ?? "any"); + const ports = policy.ports ? `:${String(policy.ports)}` : ""; + return `${name} · ${mode} · ${decision} · ${protocol}${ports}`; +} + +function decisionClass(decision: string) { + if (decision.includes("would")) { + return "border-amber-400/40 bg-amber-400/10 text-amber-300"; + } + if (decision.includes("block")) { + return "border-danger/40 bg-danger/10 text-danger"; + } + if (decision.includes("allow")) { + return "border-accent/40 bg-accent/10 text-accent"; + } + return "border-border bg-canvas text-slate-500"; +} + +function FlowRuleContext({ flow }: { flow: TrafficSummary }) { + const activeRules = flow.matchingFirewallRules.slice(0, 2); + const auditPolicies = flow.matchingAuditPolicies.slice(0, 2); + const hasContext = activeRules.length || auditPolicies.length; + if (!hasContext) { + return
No matching active or audit rule.
; + } + + return ( +
+ {activeRules.map((rule, index) => ( +
+ Rule: {ruleLabel(rule)} + {String(rule.decision ?? "observed")} +
+ ))} + {auditPolicies.map((policy, index) => ( +
+ Audit: {policyLabel(policy)} +
+ ))} + {flow.matchingFirewallRules.length + flow.matchingAuditPolicies.length > activeRules.length + auditPolicies.length ? ( +
+ {flow.matchingFirewallRules.length + flow.matchingAuditPolicies.length - activeRules.length - auditPolicies.length} more match + {flow.matchingFirewallRules.length + flow.matchingAuditPolicies.length - activeRules.length - auditPolicies.length === 1 ? "" : "es"} +
+ ) : null} +
+ ); +} + function ActiveRulesList({ rules, compact = false }: { rules: Array>; compact?: boolean }) { const visibleRules = compact ? rules.slice(0, 3) : rules; if (!rules.length) { @@ -241,6 +327,7 @@ function TrafficTable({ traffic }: { traffic: TrafficSummary[] }) { Flow Protocol + Decision Traffic Packets Seen @@ -252,8 +339,12 @@ function TrafficTable({ traffic }: { traffic: TrafficSummary[] }) {
{endpointText(flow)}
{flow.ipAddresses.join(", ") || flow.interfaceName || "no endpoint metadata"}
+ {flow.protocol}{flow.port ? `:${flow.port}` : ""} + + {flow.decision.replace("_", " ")} + {formatBytes(flow.bytes)} {flow.packets} {flow.count} sample{flow.count === 1 ? "" : "s"}