railiance-infra/scripts/sops_rotation.py

#!/usr/bin/env python3
"""Check or execute a bounded, metadata-only SOPS recipient rotation."""

from __future__ import annotations

import argparse
import hashlib
import json
import re
import shutil
import subprocess
import sys
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import yaml

from check_secret_paths import is_protected_path
from s1_receipt import validate_receipt


ROOT = Path(__file__).resolve().parents[1]


class RotationError(ValueError):
    """Rotation inputs, metadata, or approval are unsafe or incomplete."""


def _sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def _git(*args: str) -> str:
    return subprocess.check_output(["git", *args], cwd=ROOT, text=True).strip()


def protected_files(root: Path = ROOT) -> list[Path]:
    candidates = list((root / "secrets").rglob("*")) if (root / "secrets").exists() else []
    inventory = list((root / "inventory").rglob("secrets*"))
    return sorted(
        path
        for path in candidates + inventory
        if path.is_file() and is_protected_path(str(path.relative_to(root)))
    )


def load_policy(path: Path) -> list[dict[str, Any]]:
    try:
        payload = yaml.safe_load(path.read_text(encoding="utf-8"))
        rules = payload["creation_rules"]
    except (OSError, yaml.YAMLError, KeyError, TypeError) as exc:
        raise RotationError(f"cannot read SOPS policy {path}: {exc}") from exc
    if not isinstance(rules, list) or not rules:
        raise RotationError("SOPS policy has no creation_rules")
    return rules


def expected_recipients(rules: list[dict[str, Any]], relative: str) -> list[str]:
    for rule in rules:
        pattern = rule.get("path_regex")
        if not isinstance(pattern, str) or re.fullmatch(pattern, relative) is None:
            continue
        recipients = []
        for group in rule.get("key_groups", []):
            recipients.extend(group.get("age", []))
        recipients = sorted(set(recipients))
        if not recipients:
            raise RotationError(f"{relative}: matching policy has no age recipients")
        return recipients
    raise RotationError(f"{relative}: no .sops.yaml creation rule matches")


def actual_recipients(path: Path) -> list[str]:
    try:
        payload = yaml.safe_load(path.read_text(encoding="utf-8"))
        age_entries = payload["sops"]["age"]
        recipients = sorted({entry["recipient"] for entry in age_entries})
    except (OSError, yaml.YAMLError, KeyError, TypeError) as exc:
        raise RotationError(f"{path}: missing readable SOPS age metadata") from exc
    if not recipients:
        raise RotationError(f"{path}: SOPS metadata has no age recipients")
    return recipients


def rotation_plan(root: Path = ROOT) -> list[dict[str, Any]]:
    rules = load_policy(root / ".sops.yaml")
    plan = []
    for path in protected_files(root):
        relative = str(path.relative_to(root))
        before = actual_recipients(path)
        after = expected_recipients(rules, relative)
        plan.append(
            {
                "path": relative,
                "sha256": _sha256(path),
                "before_recipients": before,
                "after_recipients": after,
                "changed": before != after,
            }
        )
    if not plan:
        raise RotationError("no protected SOPS files found")
    return plan


def _load_approval(path: Path, plan: list[dict[str, Any]]) -> None:
    try:
        payload = yaml.safe_load(path.read_text(encoding="utf-8"))
    except (OSError, yaml.YAMLError) as exc:
        raise RotationError(f"cannot read approval file: {exc}") from exc
    if not isinstance(payload, dict) or payload.get("approved") is not True:
        raise RotationError("approval file must contain approved: true")
    if not payload.get("approved_by") or not payload.get("approved_at"):
        raise RotationError("approval file requires approved_by and approved_at")
    expected = [
        {
            "path": item["path"],
            "sha256": item["sha256"],
            "before_recipients": item["before_recipients"],
            "after_recipients": item["after_recipients"],
        }
        for item in plan
        if item["changed"]
    ]
    if payload.get("changes") != expected:
        raise RotationError("approval changes do not exactly match the current rotation plan")


def _verify_decryption(paths: list[Path]) -> bool:
    if shutil.which("sops") is None:
        raise RotationError("sops is required for non-printing decryption verification")
    for path in paths:
        completed = subprocess.run(
            ["sops", "--decrypt", str(path)],
            cwd=ROOT,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            check=False,
        )
        if completed.returncode != 0:
            raise RotationError(f"decryption verification failed for {path.relative_to(ROOT)}")
    return True


def _apply(plan: list[dict[str, Any]]) -> None:
    if shutil.which("sops") is None:
        raise RotationError("sops is required for rotation")
    for item in plan:
        if not item["changed"]:
            continue
        completed = subprocess.run(
            ["sops", "updatekeys", "--yes", item["path"]],
            cwd=ROOT,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            check=False,
        )
        if completed.returncode != 0:
            raise RotationError(f"sops updatekeys failed for {item['path']}")


def build_receipt(plan: list[dict[str, Any]], verified: bool, applied: bool) -> dict[str, Any]:
    all_before = sorted({r for item in plan for r in item["before_recipients"]})
    all_after = sorted({r for item in plan for r in item["after_recipients"]})
    receipt = {
        "schema_version": "1.0",
        "receipt_id": str(uuid.uuid4()),
        "event_type": "rotation",
        "synthetic": False,
        "created_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "source_revision": _git("rev-parse", "HEAD"),
        "inventory_sha256": _sha256(ROOT / "inventory" / "servers.yaml"),
        "status": "pass" if verified else "not-run",
        "applied": applied,
        "decryption_verified": verified,
        "files": [item["path"] for item in plan],
        "before_recipients": all_before,
        "after_recipients": all_after,
        "file_metadata": plan,
    }
    validate_receipt(receipt)
    return receipt


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--check", action="store_true", help="fail on recipient drift")
    parser.add_argument("--verify-decryption", action="store_true")
    parser.add_argument("--apply", action="store_true")
    parser.add_argument("--approval-file", type=Path)
    parser.add_argument("--receipt", type=Path)
    args = parser.parse_args()
    try:
        plan = rotation_plan()
        if args.check and any(item["changed"] for item in plan):
            raise RotationError("recipient drift detected")
        if args.apply:
            if args.approval_file is None:
                raise RotationError("--apply requires --approval-file")
            if not any(item["changed"] for item in plan):
                raise RotationError("--apply requires at least one recipient change")
            _load_approval(args.approval_file, plan)
            _apply(plan)
            after_plan = rotation_plan()
            if any(item["changed"] for item in after_plan):
                raise RotationError("recipient drift remains after rotation")
            if [item["path"] for item in after_plan] != [item["path"] for item in plan]:
                raise RotationError("protected file inventory changed during rotation")
            for before, after in zip(plan, after_plan):
                before["after_sha256"] = after["sha256"]
        verified = _verify_decryption(protected_files()) if args.verify_decryption or args.apply else False
        receipt = build_receipt(plan, verified, args.apply)
        if args.receipt:
            destination = args.receipt if args.receipt.is_absolute() else ROOT / args.receipt
            destination.parent.mkdir(parents=True, exist_ok=True)
            destination.write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8")
        print(
            json.dumps(
                {
                    "ok": True,
                    "files": len(plan),
                    "changes": sum(1 for item in plan if item["changed"]),
                    "decryption_verified": verified,
                    "applied": args.apply,
                    "plan": plan,
                },
                sort_keys=True,
            )
        )
        return 0
    except RotationError as exc:
        print(f"rotation failed closed: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())