85 lines
3.3 KiB
Python
85 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Single canonical Production Baseline authority for the GitOps reader.
|
|
|
|
The API and the one-shot writer import this module. Workspace and release
|
|
copies are projections only; they are never selected as a fallback source.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
AUTHORITY_PATH = Path(os.getenv(
|
|
"AIOS_GITOPS_BASELINE_AUTHORITY",
|
|
"/var/lib/aios-gitops-runtime-anchor/baseline/current.json",
|
|
))
|
|
AUTHORITY_SHA_PATH = AUTHORITY_PATH.with_suffix(".sha256")
|
|
|
|
|
|
class BaselineAuthorityError(RuntimeError):
|
|
"""Fail-closed error raised for an unavailable or inconsistent authority."""
|
|
|
|
|
|
class BaselineAuthority:
|
|
def __init__(self, path: Path | str = AUTHORITY_PATH):
|
|
self.path = Path(path)
|
|
self.sha_path = self.path.with_suffix(".sha256")
|
|
|
|
def _raw(self) -> bytes:
|
|
try:
|
|
raw = self.path.read_bytes()
|
|
except OSError as exc:
|
|
raise BaselineAuthorityError(f"authority_unavailable:{exc}") from exc
|
|
if not raw:
|
|
raise BaselineAuthorityError("authority_empty")
|
|
return raw
|
|
|
|
def read(self) -> dict[str, Any]:
|
|
raw = self._raw()
|
|
digest = hashlib.sha256(raw).hexdigest()
|
|
try:
|
|
value = json.loads(raw.decode("utf-8"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise BaselineAuthorityError("authority_invalid_json") from exc
|
|
if not isinstance(value, dict):
|
|
raise BaselineAuthorityError("authority_not_object")
|
|
expected = str(value.get("authority_sha256", ""))
|
|
if expected and expected != digest:
|
|
raise BaselineAuthorityError("authority_embedded_digest_mismatch")
|
|
if self.sha_path.exists():
|
|
sidecar = self.sha_path.read_text(encoding="utf-8").strip().split()[0]
|
|
if sidecar != digest:
|
|
raise BaselineAuthorityError("authority_sidecar_digest_mismatch")
|
|
required = {
|
|
"current_baseline", "previous_baseline", "target_oid", "tag_oid",
|
|
"evidence_closure_oid", "approval_plan_id", "status", "generation",
|
|
"source_identity",
|
|
}
|
|
missing = sorted(required.difference(value))
|
|
if missing:
|
|
raise BaselineAuthorityError("authority_missing:" + ",".join(missing))
|
|
value = dict(value)
|
|
value["authority_sha256"] = digest
|
|
value["authority_path"] = str(self.path)
|
|
return value
|
|
|
|
def read_consistent(self, policy: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
value = self.read()
|
|
if policy:
|
|
expected = policy.get("release_binding", policy)
|
|
for key, field in (("version", "current_baseline"),
|
|
("previous_version", "previous_baseline"),
|
|
("target_oid", "target_oid"),
|
|
("tag_oid", "tag_oid"),
|
|
("evidence_closure_oid", "evidence_closure_oid"),
|
|
("approval_plan_id", "approval_plan_id")):
|
|
wanted = expected.get(key)
|
|
if wanted is not None and value.get(field) != wanted:
|
|
raise BaselineAuthorityError(f"authority_binding_mismatch:{field}")
|
|
return value
|
|
|