feat: add network: endpoint resolver, VM.Config.Options privilege requirement, top talkers dashboard widget, and automatic guest firewall enablement

Add network: prefix support in endpoint_values to resolve network names to IPAM subnet CIDRs for policy matching, extend workload_provider_target to accept vmid: prefix and bare workload names as fallback resolution methods, implement dashboard_top_talkers to aggregate traffic flows by workload with interface_traffic fallback when flows unavailable, add

backend/app/api/v1/router.py

@@ -202,10 +202,15 @@ def import_discovered_ips(db: Session, cluster_id: str, workload: Workload, addr
 
 
 def workload_provider_target(db: Session, cluster: Cluster, ref: str) -> tuple[dict, Workload] | None:
-    if not ref.startswith("workload:"):
-        return None
-    workload_id = ref.removeprefix("workload:")
-    workload = db.get(Workload, workload_id)
+    workload: Workload | None = None
+    if ref.startswith("workload:"):
+        workload_ref = ref.removeprefix("workload:")
+        workload = db.get(Workload, workload_ref)
+    elif ref.startswith("vmid:"):
+        workload_ref = ref.removeprefix("vmid:")
+        workload = db.scalar(select(Workload).where(Workload.cluster_id == cluster.id, Workload.external_id == workload_ref))
+    else:
+        workload = db.scalar(select(Workload).where(Workload.cluster_id == cluster.id, Workload.name == ref))
     if not workload or workload.cluster_id != cluster.id:
         return None
     node = db.get(Node, workload.node_id)
@@ -237,6 +242,15 @@ def endpoint_values(db: Session, cluster: Cluster, ref: str) -> tuple[list[str |
         return [], [f"Workload {workload.name} has no assigned IP address for provider-side source/destination matching."]
     if is_ip_or_cidr(ref):
         return [ref], []
+    if ref.startswith("network:"):
+        network_name = ref.removeprefix("network:")
+        network = db.scalar(select(Network).where(Network.cluster_id == cluster.id, Network.name == network_name))
+        if not network:
+            return [], [f"Network {network_name} was not found in this cluster."]
+        values = [subnet.cidr for subnet in db.scalars(select(Subnet).where(Subnet.network_id == network.id).order_by(Subnet.cidr)).all()]
+        if values:
+            return values, []
+        return [], [f"Network {network_name} has no IPAM subnets to use as provider-side matcher."]
     return [], [f"Endpoint {ref} is not yet resolvable to a Proxmox firewall matcher."]
 
 
@@ -272,6 +286,48 @@ def flow_endpoint_label(owner: Workload | None, subnets: list[Subnet], value: st
     return subnet_label_for_ip(subnets, value) or "external"
 
 
+def dashboard_top_talkers(db: Session) -> list[dict[str, int | str]]:
+    totals: dict[str, int] = {}
+    workloads = db.scalars(select(Workload)).all()
+    workloads_by_node_vmid = {(workload.node_id, workload.external_id): workload for workload in workloads}
+    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()
+    }
+
+    flows = db.scalars(select(TrafficFlow)).all()
+    for flow in flows:
+        raw = flow.raw if isinstance(flow.raw, dict) else {}
+        candidates: list[Workload] = []
+        raw_vmid = str(raw.get("vmid") or "")
+        if raw_vmid:
+            workload = workloads_by_node_vmid.get((flow.node_id, raw_vmid))
+            if workload:
+                candidates.append(workload)
+        for owner in (ip_owners.get(flow.source_ip), ip_owners.get(flow.destination_ip)):
+            if owner and owner not in candidates:
+                candidates.append(owner)
+        for workload in candidates:
+            totals[workload.name] = totals.get(workload.name, 0) + int(flow.bytes or 0)
+
+    if not totals:
+        agents = db.scalars(select(NodeAgent)).all()
+        for agent in agents:
+            payload = agent.last_payload if isinstance(agent.last_payload, dict) else {}
+            for item in payload.get("interface_traffic", []):
+                if not isinstance(item, dict):
+                    continue
+                workload = workloads_by_node_vmid.get((agent.node_id, str(item.get("vmid") or "")))
+                if not workload:
+                    continue
+                totals[workload.name] = totals.get(workload.name, 0) + int(item.get("bytes") or 0)
+
+    return [
+        {"name": name, "bytes": bytes_value}
+        for name, bytes_value in sorted(totals.items(), key=lambda item: item[1], reverse=True)[:5]
+    ]
+
+
 def proxmox_action(action: str) -> str:
     return {"allow": "ACCEPT", "deny": "DROP", "reject": "REJECT"}.get(action, "ACCEPT")
 
@@ -414,7 +470,7 @@ def dashboard(_: CurrentUser, db: Session = Depends(get_db)) -> dict:
             {"id": node.id, "name": node.name, "status": node.status, "cluster_id": node.cluster_id}
             for node in faulty_nodes
         ],
-        "top_talkers": [],
+        "top_talkers": dashboard_top_talkers(db),
     }
 
 

backend/app/services/providers/proxmox.py

@@ -133,6 +133,12 @@ class ProxmoxProvider(Provider):
         vmid = target["vmid"]
         return f"{connection.api_url.rstrip('/')}/api2/json/nodes/{node}/{kind}/{vmid}/firewall/rules"
 
+    def firewall_options_url(self, connection: ProviderConnection, target: dict[str, Any]) -> str:
+        kind = "lxc" if target.get("kind") == "lxc" else "qemu"
+        node = target["node"]
+        vmid = target["vmid"]
+        return f"{connection.api_url.rstrip('/')}/api2/json/nodes/{node}/{kind}/{vmid}/firewall/options"
+
     def policy_marker(self, rule: dict[str, Any]) -> str:
         return f"NexaFabric policy={rule.get('policy_id')}"
 
@@ -156,6 +162,17 @@ class ProxmoxProvider(Provider):
                 deletions.append({"pos": pos, "comment": comment})
         return deletions
 
+    async def enable_guest_firewall(
+        self,
+        client: httpx.AsyncClient,
+        headers: dict[str, str],
+        connection: ProviderConnection,
+        target: dict[str, Any],
+    ) -> dict[str, Any]:
+        response = await client.put(self.firewall_options_url(connection, target), headers=headers, data={"enable": 1})
+        response.raise_for_status()
+        return response.json().get("data")
+
     async def apply_rules(self, connection: ProviderConnection, rules: list[dict[str, Any]]) -> dict[str, Any]:
         if connection.read_only:
             return {"applied": False, "reason": "Cluster is read-only", "rules": rules}
@@ -177,6 +194,7 @@ class ProxmoxProvider(Provider):
 
         applied_rules = []
         deleted_rules = []
+        enabled_targets = []
         audit_only_rules = []
         async with httpx.AsyncClient(verify=connection.verify_tls, timeout=20) as client:
             for target_rules in grouped.values():
@@ -184,6 +202,8 @@ class ProxmoxProvider(Provider):
                 rules_url = self.firewall_rules_url(connection, target)
                 marker = self.policy_marker(target_rules[0])
                 deleted_rules.extend(await self.delete_existing_policy_rules(client, headers, rules_url, marker))
+                if any(not rule.get("audit_only") for rule in target_rules):
+                    enabled_targets.append({"target": target, "result": await self.enable_guest_firewall(client, headers, connection, target)})
 
                 for rule in target_rules:
                     if rule.get("audit_only"):
@@ -210,6 +230,7 @@ class ProxmoxProvider(Provider):
             "applied": True,
             "rules_written": len(applied_rules),
             "rules_deleted": len(deleted_rules),
+            "firewall_enabled": enabled_targets,
             "audit_only": audit_only_rules,
             "rules": applied_rules,
         }

backend/tests/test_proxmox_provider.py

@@ -19,6 +19,8 @@ class FakeResponse:
 class FakeAsyncClient:
     deleted_urls: list[str] = []
     posted_payloads: list[dict] = []
+    put_urls: list[str] = []
+    put_payloads: list[dict] = []
 
     def __init__(self, **_: object) -> None:
         return None
@@ -41,6 +43,11 @@ class FakeAsyncClient:
         self.deleted_urls.append(url)
         return FakeResponse(None)
 
+    async def put(self, url: str, data: dict, **__: object) -> FakeResponse:
+        self.put_urls.append(url)
+        self.put_payloads.append(data)
+        return FakeResponse(None)
+
     async def post(self, _: str, data: dict, **__: object) -> FakeResponse:
         self.posted_payloads.append(data)
         return FakeResponse({"pos": 1})
@@ -50,6 +57,8 @@ class FakeAsyncClient:
 async def test_apply_rules_replaces_only_marked_nexafabric_rules(monkeypatch: pytest.MonkeyPatch) -> None:
     FakeAsyncClient.deleted_urls = []
     FakeAsyncClient.posted_payloads = []
+    FakeAsyncClient.put_urls = []
+    FakeAsyncClient.put_payloads = []
     monkeypatch.setattr(proxmox.httpx, "AsyncClient", FakeAsyncClient)
 
     result = await ProxmoxProvider().apply_rules(
@@ -74,6 +83,8 @@ async def test_apply_rules_replaces_only_marked_nexafabric_rules(monkeypatch: py
     assert result["applied"] is True
     assert result["rules_deleted"] == 1
     assert FakeAsyncClient.deleted_urls == ["https://pve.example:8006/api2/json/nodes/pve1/qemu/100/firewall/rules/1"]
+    assert FakeAsyncClient.put_urls == ["https://pve.example:8006/api2/json/nodes/pve1/qemu/100/firewall/options"]
+    assert FakeAsyncClient.put_payloads == [{"enable": 1}]
     assert FakeAsyncClient.posted_payloads == [
         {
             "type": "in",