feat: add active firewall rules display to workload insights with managed rule detection and enable status

Add active_firewall_rules_for_workload helper to fetch live firewall rules from provider with NexaFabric policy comment detection and enable status, implement list_firewall_rules method in ProxmoxProvider to retrieve rules via firewall API endpoint, extend WorkloadInsight schema with active_firewall_rules field, add ActiveRulesList component showing rule type/action/protocol/port with enable status and
This commit is contained in:
2026-07-09 19:42:51 +02:00
parent 1b81847fc6
commit b12ac38c6c
5 changed files with 89 additions and 2 deletions
+36 -1
View File
@@ -299,6 +299,40 @@ def flow_ip_label(owner: Workload | None, subnets: list[Subnet], value: str) ->
return f"{value} ({'internal' if subnet_label_for_ip(subnets, value) else 'external'})"
async def active_firewall_rules_for_workload(db: Session, workload: Workload) -> list[dict[str, object]]:
cluster = db.get(Cluster, workload.cluster_id)
if not cluster:
return []
resolved = workload_provider_target(db, cluster, f"workload:{workload.id}")
if not resolved:
return []
provider_target, _ = resolved
provider = get_provider(cluster.provider)
list_rules = getattr(provider, "list_firewall_rules", None)
if not list_rules:
return []
try:
rules = await list_rules(
ProviderConnection(
api_url=cluster.api_url,
token=cluster.token_ref or "",
verify_tls=cluster.verify_tls,
read_only=True,
),
provider_target,
)
except Exception as exc:
return [{"error": f"Unable to read active firewall rules: {exc}", "target": provider_target}]
return [
{
**rule,
"target": provider_target,
"managed_by_nexafabric": "NexaFabric policy=" in str(rule.get("comment") or ""),
}
for rule in rules
]
def dashboard_top_talkers(db: Session) -> list[dict[str, int | str]]:
totals: dict[str, int] = {}
workloads = db.scalars(select(Workload)).all()
@@ -994,7 +1028,7 @@ def workloads(_: CurrentUser, db: Session = Depends(get_db)) -> list[Workload]:
@api_router.get("/vms/{workload_id}/insights", response_model=WorkloadInsight)
def workload_insights(workload_id: str, _: CurrentUser, db: Session = Depends(get_db)) -> WorkloadInsight:
async def workload_insights(workload_id: str, _: CurrentUser, db: Session = Depends(get_db)) -> WorkloadInsight:
workload = db.get(Workload, workload_id)
if not workload:
raise HTTPException(status_code=404, detail="Workload not found")
@@ -1076,6 +1110,7 @@ def workload_insights(workload_id: str, _: CurrentUser, db: Session = Depends(ge
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),
matching_policies=policies,
effective_decision=decision,
audit_mode_notes=audit_mode_notes,