168 lines
7.5 KiB
Python
168 lines
7.5 KiB
Python
"""Platform tenant envelopes (B10), tenant governance (B11) and their invariants (C14, G7).
|
|
|
|
The platform envelope is the outer bound. Tenant governance can only narrow it:
|
|
effective envelope = platform tenant envelope ∩ tenant governance envelope.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any, Mapping
|
|
|
|
from emergent_security.constitution import PLATFORM_TENANT_BOUND, TENANT_GOVERNANCE
|
|
from emergent_security.delegation import Grant
|
|
from emergent_security.model import Outcome, Request, frozen_mapping, frozen_tuple
|
|
from emergent_security.scales import RiskScales
|
|
|
|
GOVERNANCE_MODES = ("auto", "self_governed")
|
|
DEADLINE_ACTIONS = ("escalate_to_next_admin", "escalate_to_platform") # never auto-approve
|
|
GOVERNANCE_FIELDS = frozenset({
|
|
"apiVersion", "kind", "tenant", "mode", "bounded_by", "established_by", "restrictions",
|
|
"human_decision_classes", "escalation_target", "deadline_default", "on_deadline",
|
|
})
|
|
RESTRICTION_FIELDS = frozenset({"prohibitions", "ceilings"})
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PlatformTenantEnvelope:
|
|
id: str
|
|
tenant: str
|
|
capabilities: tuple[str, ...]
|
|
delegated: Mapping[str, Mapping[str, Any]]
|
|
ceilings: Mapping[str, Any]
|
|
prohibitions: tuple[str, ...]
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Mapping[str, Any]) -> PlatformTenantEnvelope:
|
|
return cls(
|
|
id=str(data["id"]),
|
|
tenant=str(data["tenant"]),
|
|
capabilities=frozen_tuple(data.get("capabilities")),
|
|
delegated=frozen_mapping({k: frozen_mapping(v) for k, v in (data.get("delegated") or {}).items()}),
|
|
ceilings=frozen_mapping(data.get("ceilings")),
|
|
prohibitions=frozen_tuple(data.get("prohibitions")),
|
|
)
|
|
|
|
def grant_for(self, class_id: str | None) -> Grant | None:
|
|
spec = self.delegated.get(class_id) if class_id else None
|
|
outcome = Outcome.from_mode(spec.get("mode")) if spec else None
|
|
if outcome is None or not outcome.grants:
|
|
return None
|
|
return Grant(
|
|
source=self.id,
|
|
outcome=outcome,
|
|
obligations=frozen_tuple(spec.get("obligations")),
|
|
evidence_required=frozen_tuple(spec.get("evidence_required")),
|
|
risk_envelope=self.ceilings,
|
|
)
|
|
|
|
def validate(self, scales: RiskScales) -> list[str]:
|
|
errors = [f"delegated class outside capabilities: {c}" for c in self.delegated if c not in self.capabilities]
|
|
errors += [f"both delegated and prohibited: {c}" for c in self.delegated if c in self.prohibitions]
|
|
errors += [
|
|
f"{c}: delegated mode must grant: {spec.get('mode')}"
|
|
for c, spec in self.delegated.items()
|
|
if not (Outcome.from_mode(spec.get("mode")) or Outcome.DENY).grants
|
|
]
|
|
return errors + scales.check(self.ceilings, f"{self.id} ceilings")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TenantGovernance:
|
|
tenant: str
|
|
mode: str
|
|
bounded_by: str
|
|
prohibitions: tuple[str, ...]
|
|
ceilings: Mapping[str, Any]
|
|
human_decision_classes: tuple[str, ...]
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Mapping[str, Any]) -> TenantGovernance:
|
|
restrictions = data.get("restrictions") or {}
|
|
return cls(
|
|
tenant=str(data.get("tenant")),
|
|
mode=str(data.get("mode", "auto")),
|
|
bounded_by=str(data.get("bounded_by")),
|
|
prohibitions=frozen_tuple(restrictions.get("prohibitions")),
|
|
ceilings=frozen_mapping(restrictions.get("ceilings")),
|
|
human_decision_classes=frozen_tuple(data.get("human_decision_classes")),
|
|
)
|
|
|
|
@property
|
|
def self_governed(self) -> bool:
|
|
return self.mode == "self_governed"
|
|
|
|
|
|
def validate_tenant_governance(
|
|
data: Mapping[str, Any], platform: PlatformTenantEnvelope, scales: RiskScales
|
|
) -> list[str]:
|
|
"""All reasons a tenant governance document would widen, or fail to bound, the platform envelope.
|
|
|
|
Works on the raw document so that unknown (widening) fields are caught before parsing.
|
|
A non-empty result means the document is rejected as a whole.
|
|
"""
|
|
errors = [f"field not allowed: {k} (tenant governance may only narrow)" for k in sorted(set(data) - GOVERNANCE_FIELDS)]
|
|
mode = data.get("mode")
|
|
restrictions = data.get("restrictions") or {}
|
|
human_classes = data.get("human_decision_classes") or []
|
|
|
|
if mode not in GOVERNANCE_MODES:
|
|
errors.append(f"mode must be one of {', '.join(GOVERNANCE_MODES)}: {mode}")
|
|
if data.get("tenant") != platform.tenant:
|
|
errors.append(f"tenant does not match platform envelope: {data.get('tenant')}")
|
|
if data.get("bounded_by") != platform.id:
|
|
errors.append(f"bounded_by does not match platform envelope {platform.id}: {data.get('bounded_by')}")
|
|
if not str(data.get("established_by", "")).startswith("human://"):
|
|
errors.append(f"established_by must be a human principal: {data.get('established_by')}")
|
|
if data.get("on_deadline") not in DEADLINE_ACTIONS:
|
|
errors.append(f"on_deadline must be one of {', '.join(DEADLINE_ACTIONS)}: {data.get('on_deadline')}")
|
|
if mode == "auto" and (restrictions or human_classes):
|
|
errors.append("auto mode carries no restrictions; use self_governed")
|
|
|
|
errors += [f"restriction not allowed: {k}" for k in sorted(set(restrictions) - RESTRICTION_FIELDS)]
|
|
for dimension, value in (restrictions.get("ceilings") or {}).items():
|
|
if scales.rank(dimension, value) is None:
|
|
errors.append(f"unknown value for {dimension}: {value}")
|
|
elif dimension in platform.ceilings and not scales.within(dimension, value, platform.ceilings[dimension]):
|
|
errors.append(f"ceiling {dimension}={value} exceeds platform ceiling {platform.ceilings[dimension]}")
|
|
errors += [f"human decision class outside platform capabilities: {c}" for c in human_classes if c not in platform.capabilities]
|
|
return errors
|
|
|
|
|
|
def effective_ceilings(
|
|
platform: PlatformTenantEnvelope, governance: TenantGovernance | None, scales: RiskScales
|
|
) -> dict[str, Any]:
|
|
"""Recomputed on every call, so platform narrowing propagates and tenant restrictions persist."""
|
|
if governance is None or not governance.self_governed:
|
|
return dict(platform.ceilings)
|
|
return scales.intersect(platform.ceilings, governance.ceilings)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TenantContext:
|
|
platform: PlatformTenantEnvelope
|
|
governance: TenantGovernance | None = None
|
|
|
|
@property
|
|
def self_governed(self) -> bool:
|
|
return self.governance is not None and self.governance.self_governed
|
|
|
|
def prohibition(self, class_id: str | None) -> tuple[str, ...]:
|
|
"""Rule ids of an explicit prohibition on this class, or () if none."""
|
|
if class_id in self.platform.prohibitions:
|
|
return (PLATFORM_TENANT_BOUND,)
|
|
if self.self_governed and class_id in self.governance.prohibitions:
|
|
return (TENANT_GOVERNANCE,)
|
|
return ()
|
|
|
|
def outside(self, request: Request, scales: RiskScales) -> tuple[str, ...]:
|
|
"""What places the request outside the platform tenant envelope (A3 step 4); () if inside."""
|
|
if request.decision_class not in self.platform.capabilities:
|
|
return (f"capability:{request.decision_class}",)
|
|
return scales.exceeded(request.risk, self.platform.ceilings, declared_only=True)
|
|
|
|
def ceilings(self, scales: RiskScales) -> dict[str, Any]:
|
|
return effective_ceilings(self.platform, self.governance, scales)
|
|
|
|
def requires_human(self, class_id: str | None) -> bool:
|
|
return self.self_governed and class_id in self.governance.human_decision_classes
|