60 lines
2.4 KiB
Python
60 lines
2.4 KiB
Python
"""Central fail-closed policy seam for governed task and continuation requests."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import re
|
|
from typing import Any, Mapping
|
|
|
|
OID_RE = re.compile(r"^[0-9a-f]{40}$")
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class GovernedPolicyInput:
|
|
identity: str
|
|
task_id: str
|
|
repository: str
|
|
operation: str
|
|
base_oid: str
|
|
expected_branch_oid: str
|
|
target_branch: str
|
|
generation: int
|
|
risk: str = "engineering"
|
|
requested_capabilities: tuple[str, ...] = ()
|
|
previous_generation: int | None = None
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PolicyDecision:
|
|
decision: str
|
|
reason: str
|
|
|
|
def as_dict(self) -> dict[str, str]:
|
|
return {"decision": self.decision, "reason": self.reason}
|
|
|
|
|
|
def evaluate_policy(value: GovernedPolicyInput | Mapping[str, Any]) -> PolicyDecision:
|
|
if isinstance(value, Mapping):
|
|
value = GovernedPolicyInput(**value)
|
|
if value.repository != "aios/aios":
|
|
return PolicyDecision("DENY", "REPOSITORY_IDENTITY_MISMATCH")
|
|
if not OID_RE.fullmatch(value.base_oid) or not OID_RE.fullmatch(value.expected_branch_oid):
|
|
return PolicyDecision("DENY", "OID_INVALID")
|
|
if not value.target_branch.startswith("agent/") or value.target_branch in {"main", "master"}:
|
|
return PolicyDecision("DENY", "BRANCH_SCOPE_DENIED")
|
|
if "force_push" in value.requested_capabilities or "force" in value.operation.lower():
|
|
return PolicyDecision("DENY", "FORCE_PUSH_DENIED")
|
|
if {"main_write", "tag_write", "baseline_write", "production_write", "production_deploy", "tag_mutation", "baseline_mutation"}.intersection(value.requested_capabilities):
|
|
return PolicyDecision("DENY", "MUTATION_SCOPE_DENIED")
|
|
if not isinstance(value.generation, int) or isinstance(value.generation, bool) or value.generation < 1:
|
|
return PolicyDecision("DENY", "GENERATION_INVALID")
|
|
if value.previous_generation is not None and value.generation <= value.previous_generation:
|
|
return PolicyDecision("MANUAL_OWNER_REVIEW_REQUIRED", "GENERATION_NOT_MONOTONIC")
|
|
if not value.task_id:
|
|
return PolicyDecision("DENY", "TASK_ID_INVALID")
|
|
return PolicyDecision("ALLOW", "GOVERNED_ENGINEERING_SCOPE")
|
|
|
|
|
|
class GovernedPolicy:
|
|
def decide(self, request: GovernedPolicyInput | Mapping[str, Any]) -> PolicyDecision:
|
|
return evaluate_policy(request)
|