govoplan/tests/test_release_base_catalog_trust.py

from __future__ import annotations

import json
from datetime import UTC, datetime
from pathlib import Path
import sys
import tempfile
import unittest

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey


META_ROOT = Path(__file__).resolve().parents[1]
RELEASE_TOOLS_ROOT = META_ROOT / "tools" / "release"
if str(RELEASE_TOOLS_ROOT) not in sys.path:
    sys.path.insert(0, str(RELEASE_TOOLS_ROOT))

from govoplan_release.catalog import canonical_hash  # noqa: E402
from govoplan_release.selective_catalog import (  # noqa: E402
    candidate_keyring_from_authenticated_base,
    load_authenticated_catalog_base,
    public_key_base64,
    signature,
)


class ReleaseBaseCatalogTrustTests(unittest.TestCase):
    def test_exact_signed_base_pair_is_loaded_once_and_carried_in_memory(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            private_key = Ed25519PrivateKey.generate()
            catalog, keyring = self._write_pair(root, private_key=private_key)

            authenticated = load_authenticated_catalog_base(
                base_catalog=catalog,
                base_keyring=keyring,
                web_root=root,
                channel="stable",
                public_base_url="https://invalid.example",
                signer_public_keys={"release-key": public_key_base64(private_key)},
            )

            keyring.unlink()

        self.assertEqual("release-key", authenticated.keyring["keys"][0]["key_id"])
        self.assertEqual(
            canonical_hash(authenticated.keyring),
            authenticated.catalog["release"]["keyring_sha256"],
        )

    def test_deliberate_old_to_new_signer_rotation_is_carried_forward(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            private_key = Ed25519PrivateKey.generate()
            new_key = Ed25519PrivateKey.generate()
            catalog, keyring = self._write_pair(root, private_key=private_key)
            configured = {
                "release-key": public_key_base64(private_key),
                "release-key-next": public_key_base64(new_key),
            }

            authenticated = load_authenticated_catalog_base(
                base_catalog=catalog,
                base_keyring=keyring,
                web_root=root,
                channel="stable",
                public_base_url="https://invalid.example",
                signer_public_keys=configured,
            )
            candidate = candidate_keyring_from_authenticated_base(
                authenticated.keyring,
                signer_public_keys=configured,
                generated_at=datetime(2026, 7, 22, 12, tzinfo=UTC),
            )

        self.assertEqual(
            {"release-key", "release-key-next"},
            {item["key_id"] for item in candidate["keys"]},
        )

    def test_injected_extra_key_without_catalog_authorization_is_rejected(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            private_key = Ed25519PrivateKey.generate()
            extra_key = Ed25519PrivateKey.generate()
            catalog, keyring = self._write_pair(root, private_key=private_key)
            payload = json.loads(keyring.read_text(encoding="utf-8"))
            payload["keys"].append(
                {
                    "key_id": "injected-key",
                    "status": "active",
                    "public_key": public_key_base64(extra_key),
                }
            )
            keyring.write_text(json.dumps(payload), encoding="utf-8")

            with self.assertRaisesRegex(ValueError, "does not pin"):
                load_authenticated_catalog_base(
                    base_catalog=catalog,
                    base_keyring=keyring,
                    web_root=root,
                    channel="stable",
                    public_base_url="https://invalid.example",
                    signer_public_keys={
                        "release-key": public_key_base64(private_key)
                    },
                )

    def test_unpinned_or_symlinked_base_keyring_fails_closed(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            private_key = Ed25519PrivateKey.generate()
            catalog, keyring = self._write_pair(root, private_key=private_key)
            payload = json.loads(keyring.read_text(encoding="utf-8"))
            payload["generated_at"] = "changed"
            keyring.write_text(json.dumps(payload), encoding="utf-8")

            with self.assertRaisesRegex(ValueError, "does not pin"):
                load_authenticated_catalog_base(
                    base_catalog=catalog,
                    base_keyring=keyring,
                    web_root=root,
                    channel="stable",
                    public_base_url="https://invalid.example",
                    signer_public_keys={
                        "release-key": public_key_base64(private_key)
                    },
                )

            keyring.unlink()
            outside = root / "outside.json"
            outside.write_text("{}", encoding="utf-8")
            keyring.symlink_to(outside)
            with self.assertRaisesRegex(ValueError, "opened safely"):
                load_authenticated_catalog_base(
                    base_catalog=catalog,
                    base_keyring=keyring,
                    web_root=root,
                    channel="stable",
                    public_base_url="https://invalid.example",
                    signer_public_keys={
                        "release-key": public_key_base64(private_key)
                    },
                )

    @staticmethod
    def _write_pair(
        root: Path, *, private_key: Ed25519PrivateKey
    ) -> tuple[Path, Path]:
        keys = [
            {
                "key_id": "release-key",
                "status": "active",
                "public_key": public_key_base64(private_key),
            }
        ]
        keyring_payload = {
            "keyring_version": "1",
            "purpose": "govoplan module package catalog signatures",
            "keys": keys,
        }
        catalog_payload = {
            "channel": "stable",
            "sequence": 1,
            "core_release": {"version": "1.0.0"},
            "modules": [],
            "release": {"keyring_sha256": canonical_hash(keyring_payload)},
        }
        catalog_payload["signatures"] = [
            signature(
                catalog_payload,
                key_id="release-key",
                private_key=private_key,
            )
        ]
        catalog = root / "stable.json"
        keyring = root / "keyring.json"
        catalog.write_text(json.dumps(catalog_payload), encoding="utf-8")
        keyring.write_text(json.dumps(keyring_payload), encoding="utf-8")
        return catalog, keyring


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