#!/usr/bin/env python3 """Resolve security-zone declarations without guessing workload identity.""" from __future__ import annotations import argparse import hashlib import json import sys from datetime import date from pathlib import Path from typing import Any, Iterable import yaml MATURITY_RANK = {"M0": 0, "M1": 1, "M2": 2, "M3": 3} CRITICALITY_FLOOR = {"low": 0, "medium": 1, "high": 2, "critical": 3} DATACLASS_FLOOR = { "synthetic": 0, "internal": 1, "confidential": 2, "restricted": 3, } ZONE_FLOOR = { "z0-experimental": 0, "z1-operational": 1, "z2-protected": 2, "z2-continuity": 2, "z3-critical": 3, } CONTROL_PROFILE = { "z0-experimental": { "flex-auth/pre-sign": ("advisory", "fail_open"), "ops-warden/agent-high-risk-read": ("enforced", "fail_closed"), "ops-warden/plan-zone-rule": ("advisory", "fail_closed"), }, "z1-operational": { "flex-auth/pre-sign": ("advisory", "fail_open"), "ops-warden/agent-high-risk-read": ("enforced", "fail_closed"), "ops-warden/plan-zone-rule": ("advisory", "fail_closed"), }, "z2-protected": { "flex-auth/pre-sign": ("enforced", "fail_open"), "ops-warden/agent-high-risk-read": ("enforced", "fail_closed"), "ops-warden/plan-zone-rule": ("enforced", "fail_closed"), }, "z2-continuity": { "flex-auth/pre-sign": ("enforced", "fail_open"), "ops-warden/agent-high-risk-read": ("enforced", "fail_closed"), "ops-warden/plan-zone-rule": ("enforced", "fail_closed"), }, "z3-critical": { "flex-auth/pre-sign": ("enforced", "fail_closed"), "ops-warden/agent-high-risk-read": ("enforced", "fail_closed"), "ops-warden/plan-zone-rule": ("enforced", "fail_closed"), }, "unknown": { "flex-auth/pre-sign": ("advisory", "fail_open"), "ops-warden/agent-high-risk-read": ("enforced", "fail_closed"), "ops-warden/plan-zone-rule": ("enforced", "fail_closed"), }, } class DeclarationError(ValueError): """A declaration violates the security-zones_v0.1 contract.""" def _required(mapping: dict[str, Any], key: str, where: str) -> Any: value = mapping.get(key) if value is None or value == "" or value == []: raise DeclarationError(f"{where}.{key} is required") return value def _parse_date(value: Any, where: str) -> date: if isinstance(value, date): return value try: return date.fromisoformat(str(value)) except ValueError as exc: raise DeclarationError(f"{where} must be an ISO date") from exc def _services(document: dict[str, Any]) -> Iterable[dict[str, Any]]: services = document.get("services") if services is not None: if not isinstance(services, list) or not services: raise DeclarationError("services must be a non-empty list") if "zones" in document or "workload_identity" in document: raise DeclarationError( "multi-service declarations keep zones and workload_identity per service" ) yield from services return yield document def _validate_identity(service: str, identity: Any) -> dict[str, Any]: if not isinstance(identity, dict): raise DeclarationError(f"{service}.workload_identity must be a mapping") name = str(_required(identity, "name", f"{service}.workload_identity")) if name != service: raise DeclarationError( f"{service}.workload_identity.name must equal service, got {name!r}" ) _required(identity, "kind", f"{service}.workload_identity") _required(identity, "responsible_repo", f"{service}.workload_identity") bindings = _required( identity, "identity_bindings", f"{service}.workload_identity" ) if not isinstance(bindings, list): raise DeclarationError( f"{service}.workload_identity.identity_bindings must be a list" ) for index, binding in enumerate(bindings): where = f"{service}.workload_identity.identity_bindings[{index}]" if not isinstance(binding, dict): raise DeclarationError(f"{where} must be a mapping") for key in ("scheme", "authority", "subject", "principal_type"): _required(binding, key, where) if binding["principal_type"] not in {"service", "agent"}: raise DeclarationError(f"{where}.principal_type must be service or agent") return identity def _admission(service: str, zones: dict[str, Any]) -> tuple[str, str]: membership = str(_required(zones, "membership", f"{service}.zones")) if membership not in ZONE_FLOOR: raise DeclarationError(f"{service}.zones.membership is unknown: {membership!r}") context = _required(zones, "context", f"{service}.zones") if not isinstance(context, dict): raise DeclarationError(f"{service}.zones.context must be a mapping") maturity = str(_required(context, "maturity", f"{service}.zones.context")) criticality = str( _required(context, "criticality", f"{service}.zones.context") ) dataclass = str( _required(context, "data_classification", f"{service}.zones.context") ) if maturity not in MATURITY_RANK: raise DeclarationError(f"{service}.zones.context.maturity is invalid") if criticality not in CRITICALITY_FLOOR: raise DeclarationError(f"{service}.zones.context.criticality is invalid") if dataclass == "public": return "unknown", "public_data_classification_floor_unresolved" if dataclass == "n/a": _required( context, "data_classification_reason", f"{service}.zones.context", ) data_floor = 0 elif dataclass in DATACLASS_FLOOR: data_floor = DATACLASS_FLOOR[dataclass] else: raise DeclarationError( f"{service}.zones.context.data_classification is invalid" ) zone_rank = ZONE_FLOOR[membership] context_rank = max(CRITICALITY_FLOOR[criticality], data_floor) if zone_rank < context_rank: return "unsatisfied", f"{membership}_below_M{context_rank}_context_floor" if MATURITY_RANK[maturity] < zone_rank: return "unsatisfied", f"{maturity}_below_M{zone_rank}_zone_floor" if membership == "z2-continuity": supported = { str(fact) for item in zones["evidence"] if isinstance(item, dict) for fact in item.get("supports", []) } required = {"continuity-dependency", "recovery"} if not required.issubset(supported): return "unsatisfied", "continuity_evidence_incomplete" return "satisfied", "admission_floor_met" def resolve_service(service_entry: dict[str, Any], source: str) -> dict[str, Any]: service = str(_required(service_entry, "service", source)) zones = service_entry.get("zones") if zones is None: return { "workload_id": service, "declared_zone": None, "admission": "unknown", "admission_reason": "zone_membership_absent", "effective_zone": "unknown", "membership_revision": None, "controls": _controls("unknown"), "source": source, } identity = _validate_identity(service, service_entry.get("workload_identity")) if not isinstance(zones, dict): raise DeclarationError(f"{service}.zones must be a mapping") if zones.get("standard") != "security-zones_v0.1": raise DeclarationError( f"{service}.zones.standard must be security-zones_v0.1" ) for key in ( "responsible_party", "justification", "evidence", "reviewed", "review_due", ): _required(zones, key, f"{service}.zones") if not isinstance(zones["evidence"], list): raise DeclarationError(f"{service}.zones.evidence must be a list") reviewed = _parse_date(zones["reviewed"], f"{service}.zones.reviewed") review_due = _parse_date(zones["review_due"], f"{service}.zones.review_due") if review_due <= reviewed: raise DeclarationError(f"{service}.zones.review_due must be after reviewed") admission, reason = _admission(service, zones) membership = str(zones["membership"]) effective = membership if admission == "satisfied" else "unknown" revision_input = json.dumps( {"workload_identity": identity, "zones": zones}, sort_keys=True, separators=(",", ":"), default=str, ).encode() revision = "sha256:" + hashlib.sha256(revision_input).hexdigest() return { "workload_id": service, "declared_zone": membership, "admission": admission, "admission_reason": reason, "effective_zone": effective, "membership_revision": revision, "guarantees": [ "authoritative-workload-identity", "explicit-zone-membership", "non-inferred-resolution", "enforcement-time-exception-expiry", ], "controls": _controls(effective), "source": source, } def _controls(zone: str) -> list[dict[str, str]]: return [ {"id": control, "stance": stance, "failure_mode": failure} for control, (stance, failure) in CONTROL_PROFILE[zone].items() ] def resolve_paths(paths: Iterable[Path]) -> dict[str, Any]: records: list[dict[str, Any]] = [] errors: list[dict[str, str]] = [] for path in paths: try: document = yaml.safe_load(path.read_text()) or {} if not isinstance(document, dict): raise DeclarationError("document must be a mapping") for entry in _services(document): if not isinstance(entry, dict): raise DeclarationError("service entry must be a mapping") records.append(resolve_service(entry, str(path))) except (OSError, yaml.YAMLError, DeclarationError) as exc: errors.append({"source": str(path), "error": str(exc)}) return {"ok": not errors, "standard": "security-zones_v0.1", "records": records, "errors": errors} def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("paths", nargs="+", type=Path) args = parser.parse_args() result = resolve_paths(args.paths) print(json.dumps(result, indent=2, sort_keys=True)) return 0 if result["ok"] else 1 if __name__ == "__main__": sys.exit(main())