From baa0d24eb43c7af03a58fd37928cbb45ca6da26f Mon Sep 17 00:00:00 2001 From: nessi Date: Thu, 9 Jul 2026 15:36:42 +0200 Subject: [PATCH] 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 --- README.md | 1 + backend/app/api/v1/router.py | 66 +++++++++++++++++++++-- backend/app/services/providers/proxmox.py | 21 ++++++++ backend/tests/test_proxmox_provider.py | 11 ++++ 4 files changed, 94 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 9937715..70d9bf2 100644 --- a/README.md +++ b/README.md @@ -77,6 +77,7 @@ For write-enabled firewall orchestration, create a separate token or role and do Minimum practical write privileges for VM/LXC-level firewall rules: - `VM.Audit` so NexaFabric can resolve guests and inspect existing rules. +- `VM.Config.Options` so NexaFabric can enable the guest firewall option before writing rules. - `VM.Config.Network` on `/vms` or on the narrow VM/LXC paths you want NexaFabric to manage. NexaFabric writes only rules that carry a `NexaFabric policy=...` comment marker. During apply it removes and replaces its own marked rules for the selected policy, leaving manually created Proxmox firewall rules untouched. diff --git a/backend/app/api/v1/router.py b/backend/app/api/v1/router.py index 3bee678..58979d7 100644 --- a/backend/app/api/v1/router.py +++ b/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), } diff --git a/backend/app/services/providers/proxmox.py b/backend/app/services/providers/proxmox.py index 6a181c9..b9ad556 100644 --- a/backend/app/services/providers/proxmox.py +++ b/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, } diff --git a/backend/tests/test_proxmox_provider.py b/backend/tests/test_proxmox_provider.py index 467abbb..0666b25 100644 --- a/backend/tests/test_proxmox_provider.py +++ b/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",