railiance-infra/tests/test_secret_and_receipt_contracts.py

from __future__ import annotations

import copy
import contextlib
import io
import json
import sys
import tempfile
import unittest
import uuid
from pathlib import Path
from unittest.mock import patch

import yaml

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))

from check_secret_paths import is_encrypted_content, is_protected_path  # noqa: E402
from s1_receipt import ReceiptError, load_receipt, validate_receipt  # noqa: E402
from sops_rotation import rotation_plan  # noqa: E402
import sops_rotation  # noqa: E402


class SecretAndReceiptContractTests(unittest.TestCase):
    def test_inventory_secret_paths_are_protected(self) -> None:
        self.assertTrue(is_protected_path("secrets/provider.yaml"))
        self.assertTrue(is_protected_path("inventory/group_vars/secrets.sops.yaml"))
        self.assertFalse(is_protected_path("inventory/group_vars/all.yaml"))

    def test_plaintext_and_empty_files_fail(self) -> None:
        self.assertFalse(is_encrypted_content("secrets/example.yaml", "value: clear"))
        self.assertFalse(is_encrypted_content("secrets/example.age", ""))
        self.assertTrue(is_encrypted_content("secrets/example.yaml", "sops:\n  age: []\n"))

    def test_synthetic_chain_validates(self) -> None:
        load_receipt(ROOT / "docs" / "evidence" / "s1-receipts" / "synthetic-provisioning-chain.json")

    def test_incomplete_passing_chain_fails(self) -> None:
        payload = json.loads(
            (ROOT / "docs" / "evidence" / "s1-receipts" / "synthetic-provisioning-chain.json").read_text()
        )
        payload["phases"][2]["status"] = "not-run"
        with self.assertRaisesRegex(ReceiptError, "every phase"):
            validate_receipt(payload)

    def test_secret_shaped_receipt_content_fails(self) -> None:
        payload = json.loads(
            (ROOT / "docs" / "evidence" / "s1-receipts" / "synthetic-provisioning-chain.json").read_text()
        )
        payload["token"] = "not-even-a-real-value"
        with self.assertRaisesRegex(ReceiptError, "secret-shaped key"):
            validate_receipt(payload)

        payload.pop("token")
        payload["provider_access_token"] = "redacted-is-still-not-allowed"
        with self.assertRaisesRegex(ReceiptError, "secret-shaped key"):
            validate_receipt(payload)

    def test_passing_verification_without_evidence_fails(self) -> None:
        payload = {
            "schema_version": "1.0",
            "receipt_id": str(uuid.uuid4()),
            "event_type": "verification",
            "synthetic": False,
            "created_at": "2026-08-23T09:30:00Z",
            "source_revision": "3734a1c",
            "inventory_sha256": "a" * 64,
            "status": "pass",
            "hosts": ["Railiance01"],
            "profiles": {"Railiance01": "ufw-managed"},
            "observed_at": "2026-08-23T09:30:00Z",
            "fresh_until": "2026-08-24T09:30:00Z",
            "evidence": [],
        }
        with self.assertRaisesRegex(ReceiptError, "host evidence"):
            validate_receipt(payload)

    def test_current_sops_recipient_metadata_matches_policy(self) -> None:
        plan = rotation_plan(ROOT)
        self.assertTrue(plan)
        self.assertFalse(any(item["changed"] for item in plan))

    def test_rotation_approval_binds_ciphertext(self) -> None:
        plan = [{"path": "secrets/example.yaml", "sha256": "a" * 64,
                 "before_recipients": ["age1before"],
                 "after_recipients": ["age1after"], "changed": True}]
        approval = {"approved": True, "approved_by": "reviewer",
                    "approved_at": "2026-09-27T00:00:00Z",
                    "changes": [{k: v for k, v in plan[0].items() if k != "changed"}]}
        with tempfile.TemporaryDirectory() as tmp:
            path = Path(tmp) / "approval.yaml"
            path.write_text(yaml.safe_dump(approval))
            sops_rotation._load_approval(path, plan)
            plan[0]["sha256"] = "b" * 64
            with self.assertRaises(sops_rotation.RotationError):
                sops_rotation._load_approval(path, plan)

    def test_applied_rotation_keeps_reviewed_before_and_after(self) -> None:
        before = [{"path": "secrets/example.yaml", "sha256": "a" * 64,
                   "before_recipients": ["age1before"],
                   "after_recipients": ["age1after"], "changed": True}]
        after = [{"path": "secrets/example.yaml", "sha256": "b" * 64,
                  "before_recipients": ["age1after"],
                  "after_recipients": ["age1after"], "changed": False}]
        with tempfile.TemporaryDirectory() as tmp:
            receipt_path = Path(tmp) / "receipt.json"
            with patch.object(sys, "argv", ["rotation", "--apply", "--approval-file",
                                           "approval.yaml", "--receipt", str(receipt_path)]), \
                 patch.object(sops_rotation, "rotation_plan", side_effect=[before, after]), \
                 patch.object(sops_rotation, "_load_approval"), \
                 patch.object(sops_rotation, "_apply"), \
                 patch.object(sops_rotation, "_verify_decryption", return_value=True), \
                 contextlib.redirect_stdout(io.StringIO()) as output:
                self.assertEqual(sops_rotation.main(), 0)
            receipt = json.loads(receipt_path.read_text())
            self.assertEqual(receipt["before_recipients"], ["age1before"])
            self.assertEqual(receipt["after_recipients"], ["age1after"])
            self.assertEqual(receipt["file_metadata"][0]["sha256"], "a" * 64)
            self.assertEqual(receipt["file_metadata"][0]["after_sha256"], "b" * 64)
            self.assertEqual(json.loads(output.getvalue())["changes"], 1)

    def test_default_rotation_output_contains_reviewable_plan(self) -> None:
        with patch.object(sys, "argv", ["rotation"]), \
             contextlib.redirect_stdout(io.StringIO()) as output:
            self.assertEqual(sops_rotation.main(), 0)
        payload = json.loads(output.getvalue())
        self.assertEqual(payload["plan"], rotation_plan(ROOT))
        self.assertFalse(payload["decryption_verified"])


if __name__ == "__main__":
    unittest.main()