feat: add firewall enforcement status checking to policy apply with datacenter/node/guest/interface validation and warning collection

Add cluster_firewall_options_url and node_firewall_options_url helpers to build firewall options endpoint paths, implement config_interface_status to parse network interface firewall flags from VM/LXC config with bridge/firewall/raw fields, add firewall_enforcement_status to check enable status across datacenter/node/guest levels and validate interface firewall flags with warning collection for disabled settings, extend apply_rules response with enforcement_

backend/app/services/providers/proxmox.py

@@ -152,6 +152,13 @@ class ProxmoxProvider(Provider):
         vmid = target["vmid"]
         return f"{connection.api_url.rstrip('/')}/api2/json/nodes/{node}/{kind}/{vmid}/config"
 
+    def cluster_firewall_options_url(self, connection: ProviderConnection) -> str:
+        return f"{connection.api_url.rstrip('/')}/api2/json/cluster/firewall/options"
+
+    def node_firewall_options_url(self, connection: ProviderConnection, target: dict[str, Any]) -> str:
+        node = target["node"]
+        return f"{connection.api_url.rstrip('/')}/api2/json/nodes/{node}/firewall/options"
+
     def policy_marker(self, rule: dict[str, Any]) -> str:
         return f"NexaFabric policy={rule.get('policy_id')}"
 
@@ -223,6 +230,59 @@ class ProxmoxProvider(Provider):
             update_response.raise_for_status()
         return [{"interface": key, "firewall": 1} for key in sorted(changes)]
 
+    def config_interface_status(self, config: dict[str, Any]) -> list[dict[str, Any]]:
+        interfaces = []
+        for key, value in config.items():
+            if not key.startswith("net") or not isinstance(value, str):
+                continue
+            parts = {part.split("=", 1)[0]: part.split("=", 1)[1] for part in value.split(",") if "=" in part}
+            interfaces.append(
+                {
+                    "interface": key,
+                    "bridge": parts.get("bridge"),
+                    "firewall": parts.get("firewall") == "1",
+                    "raw": value,
+                }
+            )
+        return interfaces
+
+    async def firewall_enforcement_status(
+        self,
+        client: httpx.AsyncClient,
+        headers: dict[str, str],
+        connection: ProviderConnection,
+        target: dict[str, Any],
+    ) -> dict[str, Any]:
+        status: dict[str, Any] = {"target": target, "warnings": []}
+        checks = [
+            ("datacenter", self.cluster_firewall_options_url(connection)),
+            ("node", self.node_firewall_options_url(connection, target)),
+            ("guest", self.firewall_options_url(connection, target)),
+            ("config", self.workload_config_url(connection, target)),
+        ]
+        for name, url in checks:
+            try:
+                response = await client.get(url, headers=headers)
+                response.raise_for_status()
+                data = response.json().get("data", {})
+            except Exception as exc:
+                status[name] = {"error": str(exc)}
+                status["warnings"].append(f"Unable to read {name} firewall status: {exc}")
+                continue
+            if name == "config":
+                interfaces = self.config_interface_status(data)
+                status["interfaces"] = interfaces
+                disabled = [interface["interface"] for interface in interfaces if not interface.get("firewall")]
+                if disabled:
+                    status["warnings"].append(f"VM/LXC network firewall flag is disabled on: {', '.join(disabled)}")
+                continue
+            status[name] = data
+            enabled = data.get("enable")
+            if str(enabled) not in {"1", "True", "true"}:
+                label = {"datacenter": "Datacenter", "node": "Node", "guest": "VM/LXC"}.get(name, name)
+                status["warnings"].append(f"{label} firewall enable option is not active.")
+        return status
+
     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}
@@ -246,6 +306,7 @@ class ProxmoxProvider(Provider):
         deleted_rules = []
         enabled_targets = []
         enabled_interfaces = []
+        enforcement_status = []
         audit_only_rules = []
         async with httpx.AsyncClient(verify=connection.verify_tls, timeout=20) as client:
             for target_rules in grouped.values():
@@ -277,6 +338,7 @@ class ProxmoxProvider(Provider):
                     create_response = await client.post(rules_url, headers=headers, data=provider_rule)
                     create_response.raise_for_status()
                     applied_rules.append({"target": target, "rule": provider_rule, "result": create_response.json().get("data")})
+                enforcement_status.append(await self.firewall_enforcement_status(client, headers, connection, target))
 
         return {
             "applied": True,
@@ -284,6 +346,7 @@ class ProxmoxProvider(Provider):
             "rules_deleted": len(deleted_rules),
             "firewall_enabled": enabled_targets,
             "interfaces_enabled": enabled_interfaces,
+            "enforcement_status": enforcement_status,
             "audit_only": audit_only_rules,
             "rules": applied_rules,
         }

backend/tests/test_proxmox_provider.py

@@ -34,6 +34,10 @@ class FakeAsyncClient:
     async def get(self, url: str, **__: object) -> FakeResponse:
         if url.endswith("/config"):
             return FakeResponse({"net0": "virtio=AA:BB:CC:DD:EE:FF,bridge=vmbr0,firewall=0", "name": "web"})
+        if url.endswith("/cluster/firewall/options"):
+            return FakeResponse({"enable": 1})
+        if url.endswith("/firewall/options"):
+            return FakeResponse({"enable": 1})
         return FakeResponse(
             [
                 {"pos": 0, "comment": "manual rule"},