nessi
3cd2c0a2f1
Add setup wizard with status tracking via SystemSetting model, implement /setup/status and /setup/complete endpoints to create initial admin user and optional cluster configuration, add workload insights endpoint with traffic analysis and policy matching including audit mode detection, implement enforcement_mode property on Policy model with audit/enforced states, add Modal component for dialogs, create SetupWizard page with multi
38 lines
1.5 KiB
Python
38 lines
1.5 KiB
Python
from typing import Any
|
|
|
|
from app.models.domain import Policy
|
|
|
|
|
|
class PolicyEngine:
|
|
def compile(self, policy: Policy) -> dict[str, Any]:
|
|
definition = policy.definition or {}
|
|
source = definition.get("source", "any")
|
|
destination = definition.get("destination", "any")
|
|
service = definition.get("service", {"protocol": "any", "ports": "any"})
|
|
action = definition.get("action", "allow")
|
|
direction = definition.get("direction", "ingress")
|
|
enforcement_mode = definition.get("enforcement_mode", "enforced")
|
|
|
|
generated_rule = {
|
|
"policy_id": policy.id,
|
|
"policy_version": policy.version,
|
|
"source": source,
|
|
"destination": destination,
|
|
"protocol": service.get("protocol", "any"),
|
|
"ports": service.get("ports", "any"),
|
|
"direction": direction,
|
|
"action": action,
|
|
"enforcement_mode": enforcement_mode,
|
|
"audit_only": enforcement_mode == "audit",
|
|
"logging": bool(definition.get("logging", False)),
|
|
"description": definition.get("description", policy.name),
|
|
}
|
|
|
|
warnings = []
|
|
if source == "any" and destination == "any":
|
|
warnings.append("Policy targets all sources and destinations.")
|
|
if action == "allow" and service.get("ports") == "any":
|
|
warnings.append("Broad allow policy uses all ports.")
|
|
|
|
return {"rules": [generated_rule], "warnings": warnings, "conflicts": []}
|