ops-mason/src/ops_mason/plan.py
tegwick 2b83324c01 Bind OpenBao builds to approved inputs and secure credential delivery
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e75a-fc5c-7913-9dba-9846210c766d
2026-09-28 11:55:39 +02:00

89 lines
2.9 KiB
Python

"""Parse a construction plan file's YAML frontmatter.
See docs/construction-plan-format.md. The frontmatter's status field is
the phase-4 hard gate: the build executor (executor.py) refuses to run
against anything but status: approved with both approved_by/approved_at
set.
"""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any
import yaml
class PlanError(RuntimeError):
pass
class _PlanLoader(yaml.SafeLoader):
"""Reject ambiguous approval documents rather than keeping the last key."""
def _unique_mapping(loader, node, deep=False):
result = {}
for key_node, value_node in node.value:
key = loader.construct_object(key_node, deep=deep)
if not isinstance(key, str) or key in result:
raise PlanError("plan mappings require unique string keys")
result[key] = loader.construct_object(value_node, deep=deep)
return result
_PlanLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, _unique_mapping)
@dataclass
class ConstructionPlan:
id: str
status: str
approved_by: str | None
approved_at: str | None
demand_source: str
consumer_repo: str
credential_type: str
path: Path
build_spec: dict[str, Any] | None = None
approved_spec_sha256: str | None = None
@classmethod
def load(cls, path: str | Path) -> ConstructionPlan:
path = Path(path)
text = path.read_text()
lines = text.splitlines()
if not lines or lines[0] != "---":
raise PlanError(f"{path}: missing YAML frontmatter")
try:
end = lines.index("---", 1)
except ValueError:
raise PlanError(f"{path}: missing YAML frontmatter delimiter") from None
try:
data = yaml.load("\n".join(lines[1:end]), Loader=_PlanLoader)
except yaml.YAMLError as exc:
raise PlanError(f"{path}: invalid YAML frontmatter") from exc
if not isinstance(data, dict):
raise PlanError(f"{path}: frontmatter must be a mapping")
missing = {"id", "status"} - set(data)
if missing:
raise PlanError(f"{path}: frontmatter missing field(s): {', '.join(sorted(missing))}")
return cls(
id=data["id"],
status=data["status"],
approved_by=data.get("approved_by"),
approved_at=data.get("approved_at"),
demand_source=data.get("demand_source", ""),
consumer_repo=data.get("consumer_repo", ""),
credential_type=data.get("credential_type", ""),
path=path,
build_spec=data.get("build_spec"),
approved_spec_sha256=data.get("approved_spec_sha256"),
)
def is_approved(self) -> bool:
return self.status == "approved" and all(
isinstance(value, str) and bool(value.strip())
for value in (self.approved_by, self.approved_at)
)