77 lines
3.0 KiB
Python
77 lines
3.0 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from dataclasses import dataclass, field
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from skylattice.core.decisions import active_decisions, get_decision
|
||
|
|
from skylattice.core.intent import validate_intent_schema
|
||
|
|
from skylattice.core.paths import blueprint_dir, load_yaml
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class PolicyResult:
|
||
|
|
errors: list[str] = field(default_factory=list)
|
||
|
|
warnings: list[str] = field(default_factory=list)
|
||
|
|
|
||
|
|
@property
|
||
|
|
def ok(self) -> bool:
|
||
|
|
return not self.errors
|
||
|
|
|
||
|
|
|
||
|
|
def run_policies(customer_root, intent: dict[str, Any]) -> PolicyResult:
|
||
|
|
result = PolicyResult()
|
||
|
|
result.errors.extend(validate_intent_schema(intent))
|
||
|
|
|
||
|
|
rules = load_yaml(blueprint_dir() / "policies" / "rules.yaml").get("rules", [])
|
||
|
|
checks = {r["check"]: r for r in rules}
|
||
|
|
|
||
|
|
if "domains_non_empty" in checks and not intent.get("domains"):
|
||
|
|
result.errors.append(f"[{checks['domains_non_empty']['id']}] At least one domain is required.")
|
||
|
|
|
||
|
|
if "blueprint_pinned" in checks and not intent.get("blueprint"):
|
||
|
|
result.errors.append(f"[{checks['blueprint_pinned']['id']}] intent.blueprint must be set.")
|
||
|
|
|
||
|
|
if "overrides_have_decisions" in checks:
|
||
|
|
for ov in intent.get("overrides") or []:
|
||
|
|
did = ov.get("decision_id")
|
||
|
|
if not did:
|
||
|
|
result.errors.append(
|
||
|
|
f"[override_requires_decision] Override {ov.get('path')} missing decision_id."
|
||
|
|
)
|
||
|
|
continue
|
||
|
|
dec = get_decision(customer_root, did)
|
||
|
|
if not dec:
|
||
|
|
result.errors.append(
|
||
|
|
f"[override_requires_decision] Decision '{did}' not found for override {ov.get('path')}."
|
||
|
|
)
|
||
|
|
elif dec.get("status") != "active":
|
||
|
|
result.errors.append(
|
||
|
|
f"[override_requires_decision] Decision '{did}' is not active."
|
||
|
|
)
|
||
|
|
|
||
|
|
if "unmanaged_have_decisions" in checks:
|
||
|
|
for um in intent.get("unmanaged") or []:
|
||
|
|
if not um.get("decision_id"):
|
||
|
|
result.warnings.append(
|
||
|
|
f"[unmanaged_requires_decision] Unmanaged {um.get('kind')}:{um.get('name')} has no decision_id."
|
||
|
|
)
|
||
|
|
|
||
|
|
if "restricted_prod_curated_guard" in checks:
|
||
|
|
restricted = {
|
||
|
|
d["name"] for d in intent.get("domains") or [] if d.get("access_profile") == "restricted"
|
||
|
|
}
|
||
|
|
for ov in intent.get("overrides") or []:
|
||
|
|
path = ov.get("path") or ""
|
||
|
|
if "prod_curated_read" in path:
|
||
|
|
# path like domains.finance.prod_curated_read
|
||
|
|
parts = path.split(".")
|
||
|
|
domain = parts[1] if len(parts) > 1 else ""
|
||
|
|
if domain in restricted and not ov.get("decision_id"):
|
||
|
|
result.errors.append(
|
||
|
|
f"[restricted_prod_curated_read] {path} requires a decision."
|
||
|
|
)
|
||
|
|
|
||
|
|
# Ensure referenced active decisions exist for explainability coverage
|
||
|
|
_ = active_decisions(customer_root)
|
||
|
|
return result
|