"""Load-bearing vs attributive evidence classification (ยง9.6). The YAML file is the declaration. ``SHIPPED_RULES`` is the pin that makes drift fail the test. No function here grants or denies an action based on whether a local evidence record exists. """ from __future__ import annotations import os from dataclasses import dataclass from pathlib import Path from typing import Any import yaml from secrets_engine.config import repo_root from secrets_engine.errors import SecretsEngineError KIND_LOAD_BEARING = "load-bearing" KIND_ATTRIBUTIVE = "attributive" KIND_HEARTBEAT = "heartbeat" # First match wins. Destroy is always load-bearing; production control # mutations are load-bearing; everything else is attributive. SHIPPED_RULES = ( { "id": "heartbeat", "kind": KIND_HEARTBEAT, "actions": ("evidence-heartbeat",), "stages": ("build", "test", "prod", "unknown"), }, { "id": "destroy", "kind": KIND_LOAD_BEARING, "actions": ("lifecycle-destroy",), "stages": ("build", "test", "prod", "unknown"), }, { "id": "production-control-mutation", "kind": KIND_LOAD_BEARING, "actions": ( "revoke", "lifecycle-suspend", "lifecycle-deactivate", "provision", "session-revoke", "wrap", ), "stages": ("prod",), }, { "id": "default-attributive", "kind": KIND_ATTRIBUTIVE, "actions": ("*",), "stages": ("build", "test", "prod", "unknown"), }, ) class EvidenceClassificationError(SecretsEngineError): """Classification file missing or malformed.""" exit_code = 1 @dataclass(frozen=True) class EvidenceClass: kind: str rule_id: str action: str stage: str queued_locally: bool @property def completeness_claimed(self) -> bool: return False def classification_path() -> Path: override = os.environ.get("SECRETS_ENGINE_EVIDENCE_CLASSIFICATION", "") if override: return Path(override) return repo_root() / "evidence-classification.yaml" def _normalize_rules(raw: Any) -> tuple[dict[str, Any], ...]: if not isinstance(raw, list) or not raw: raise EvidenceClassificationError( "evidence-classification.yaml must list at least one rule" ) rules: list[dict[str, Any]] = [] for item in raw: if not isinstance(item, dict): raise EvidenceClassificationError("each classification rule must be a map") kind = str(item.get("kind", "")) if kind not in {KIND_LOAD_BEARING, KIND_ATTRIBUTIVE, KIND_HEARTBEAT}: raise EvidenceClassificationError(f"unknown evidence kind {kind!r}") actions = tuple(str(a) for a in item.get("actions") or ()) stages = tuple(str(s) for s in item.get("stages") or ()) if not actions or not stages: raise EvidenceClassificationError( f"rule {item.get('id')!r} needs actions and stages" ) rules.append( { "id": str(item.get("id", "")), "kind": kind, "actions": actions, "stages": stages, } ) return tuple(rules) def load_classification_rules( path: Path | None = None, ) -> tuple[dict[str, Any], ...]: target = path or classification_path() try: data = yaml.safe_load(target.read_text(encoding="utf-8")) or {} except (OSError, yaml.YAMLError) as exc: raise EvidenceClassificationError( f"unable to load evidence classification {target}: {exc}" ) from exc if data.get("completeness_claimed") is not False: raise EvidenceClassificationError( f"{target} must declare completeness_claimed: false" ) if data.get("no_control_branches_on_presence") is not True: raise EvidenceClassificationError( f"{target} must declare no_control_branches_on_presence: true" ) return _normalize_rules(data.get("rules")) def classify( action: str, stage: str, *, rules: tuple[dict[str, Any], ...] | None = None, ) -> EvidenceClass: """Return the evidence class for an action/stage. Never a permission.""" table = rules if rules is not None else load_classification_rules() stage_key = stage if stage else "unknown" for rule in table: actions = rule["actions"] stages = rule["stages"] if "*" not in actions and action not in actions: continue if stage_key not in stages and "*" not in stages: continue kind = str(rule["kind"]) return EvidenceClass( kind=kind, rule_id=str(rule["id"]), action=action, stage=stage_key, queued_locally=kind in {KIND_LOAD_BEARING, KIND_HEARTBEAT}, ) return EvidenceClass( kind=KIND_ATTRIBUTIVE, rule_id="implicit-attributive", action=action, stage=stage_key, queued_locally=False, )