#!/usr/bin/env python3 """Semantic validation for managed-infrastructure portfolio records.""" from __future__ import annotations import json import re import sys from datetime import date from pathlib import Path from entities import association_ok RESOURCE_ID = re.compile(r"^resource:[a-z0-9][a-z0-9:_-]+$") INLINE_ENDPOINT = re.compile( r"(https?://\S*s3\S*|https?://\S+\.scw\.cloud|\bs3://)", re.IGNORECASE, ) INLINE_SECRET = re.compile( r"(BEGIN [A-Z ]*PRIVATE KEY|AGE-SECRET-KEY-|AKIA[0-9A-Z]{16}|sk_live_|SCW[A-Z0-9]{17})" ) V03_FACETS = ( "description", "decision", "operational_refs", "credential_handles", "consumers", ) # Evidence and decision refs may cite provider HTTPS docs. Those are not # operating attributes and must not be treated as inline endpoints. SKIP_INLINE_PATHS = { ("evidence", "ref"), ("decision", "ref"), ("requirements", "ref"), ("cost", "price_evidence"), } STATUSES = { "proposed", "ordered", "commissioning", "active", "suspended", "retiring", "retired", "rejected", } TRANSITIONS = { "proposed": {"ordered", "rejected"}, "ordered": {"commissioning", "rejected"}, "commissioning": {"active", "rejected"}, "active": {"suspended", "retiring"}, "suspended": {"active", "retiring"}, "retiring": {"retired", "active"}, "retired": set(), "rejected": {"proposed"}, } def _day(value: str | None) -> date | None: return date.fromisoformat(value) if value else None def validate_transition(current: str, target: str) -> None: if current not in STATUSES or target not in STATUSES: raise ValueError("unknown lifecycle status") if target not in TRANSITIONS[current]: raise ValueError(f"invalid lifecycle transition {current} -> {target}") def _walk_strings(node, path=()): if isinstance(node, dict): for key, value in node.items(): yield from _walk_strings(value, path + (key,)) elif isinstance(node, list): for value in node: yield from _walk_strings(value, path) elif isinstance(node, str): yield path, node def _skip_inline(path: tuple) -> bool: if path and path[0] in {"operational_refs", "credential_handles"}: return True return len(path) >= 2 and (path[0], path[-1]) in SKIP_INLINE_PATHS def validate_record(record: dict) -> None: version = record.get("schema_version") if version not in {"0.2", "0.3"}: raise ValueError("portfolio records must use schema_version 0.2 or 0.3") if not RESOURCE_ID.fullmatch(record.get("id", "")): raise ValueError("invalid resource id") if record.get("status") not in STATUSES: raise ValueError("unknown lifecycle status") if record.get("record_scope") not in {"inventory", "example"}: raise ValueError("record_scope must be inventory or example") association_ok(record) if version == "0.3": missing = [name for name in V03_FACETS if name not in record] if missing: raise ValueError("schema 0.3 requires five facets: " + ",".join(missing)) consumers = record.get("consumers") or {} if "potential" not in consumers or "actual" not in consumers: raise ValueError("schema 0.3 consumers must list potential and actual") decision = record.get("decision") if record.get("status") in {"ordered", "commissioning"} and ( not decision or decision.get("status") != "approved" ): raise ValueError("ordered or commissioning resources require an approved decision") for ref in record.get("operational_refs") or []: if not str(ref).startswith("reef:"): raise ValueError(f"operational_refs must be reef: references: {ref}") for ref in record.get("credential_handles") or []: if not str(ref).startswith("secret:"): raise ValueError(f"credential_handles must be secret: references: {ref}") for path, value in _walk_strings(record): if _skip_inline(path): continue if INLINE_ENDPOINT.search(value): raise ValueError( "inline provider endpoint must be a reef: reference, not " + ".".join(str(part) for part in path) ) if INLINE_SECRET.search(value): raise ValueError( "inline secret material is forbidden; use a secret: handle at " + ".".join(str(part) for part in path) ) allocation = record["ownership"]["allocation"] if allocation["mode"] == "unattributed": if allocation["cost_attribution_key"] is not None: raise ValueError("unattributed resources cannot have an attribution key") elif not allocation["cost_attribution_key"]: raise ValueError("dedicated/shared resources require an attribution key") if allocation["mode"] == "shared" and ( not allocation["driver"] or not allocation["method_version"] ): raise ValueError("shared resources require a driver and method version") dimensions = [(item["metric"], item["kind"]) for item in record["capacity"]] if len(dimensions) != len(set(dimensions)): raise ValueError("capacity metric/kind pairs must be unique") if any(rel["resource_id"] == record["id"] for rel in record["relationships"]): raise ValueError("a resource cannot relate to itself") lifecycle = record["lifecycle"] proposed = _day(lifecycle["proposed_on"]) ordered = _day(lifecycle["ordered_on"]) commissioned = _day(lifecycle["commissioned_on"]) retired = _day(lifecycle["retired_on"]) dated = [value for value in (proposed, ordered, commissioned, retired) if value] if dated != sorted(dated): raise ValueError("lifecycle dates are out of order") if record["status"] == "retired" and retired is None: raise ValueError("retired resources require retired_on") # Discovery must preserve unknown commercial and commissioning dates as # null instead of manufacturing precision. Date ordering is enforced when # the authoritative repositories or provider evidence supply the values. if record["record_scope"] == "example" and not any( item["kind"] == "example" for item in record["evidence"] ): raise ValueError("examples require explicit example evidence") def main() -> int: if len(sys.argv) < 2: print(f"usage: {sys.argv[0]} RECORD.json ...", file=sys.stderr) return 2 for name in sys.argv[1:]: validate_record(json.loads(Path(name).read_text())) print(f"portfolio records valid: {len(sys.argv) - 1}") return 0 if __name__ == "__main__": raise SystemExit(main())