govoplan/tests/test_runtime_distribution.py

from __future__ import annotations

import base64
from datetime import UTC, datetime, timedelta
import hashlib
from pathlib import Path
import sys
import tempfile
import unittest

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


META_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(META_ROOT / "tools" / "deployment"))
sys.path.insert(0, str(META_ROOT / "tools" / "release"))

from govoplan_deploy.distribution import (  # noqa: E402
    DistributionError,
    canonical_signed_payload,
    verify_manifest,
    verify_manifest_binding,
    verify_offline_image_index,
)


class RuntimeDistributionTests(unittest.TestCase):
    def setUp(self) -> None:
        self.now = datetime(2026, 8, 3, tzinfo=UTC)
        self.private = Ed25519PrivateKey.generate()
        public = self.private.public_key().public_bytes(
            serialization.Encoding.PEM,
            serialization.PublicFormat.SubjectPublicKeyInfo,
        ).decode("ascii")
        self.keyring = {
            "schema_version": "1",
            "purpose": "govoplan-runtime-distribution",
            "keys": [
                {
                    "key_id": "release-1",
                    "algorithm": "ed25519",
                    "status": "active",
                    "public_key_pem": public,
                    "not_before": (self.now - timedelta(days=1)).isoformat(),
                    "expires_at": (self.now + timedelta(days=365)).isoformat(),
                }
            ],
        }

    def test_verifies_signature_and_exact_runtime_binding(self) -> None:
        payload = self._manifest()

        key_id = verify_manifest(
            payload,
            self.keyring,
            expected_channel="stable",
            now=self.now,
        )
        verify_manifest_binding(
            payload,
            channel="stable",
            version="1.2.3",
            api_image=payload["images"]["api"]["index"],
            web_image=payload["images"]["web"]["index"],
            enabled_modules=("access", "files"),
            composition_sha256="c" * 64,
            dependencies=payload["dependencies"],
        )

        self.assertEqual("release-1", key_id)

    def test_tamper_expiry_revocation_and_unknown_key_fail_closed(self) -> None:
        payload = self._manifest()
        payload["composition"]["module_ids"].append("mail")
        with self.assertRaisesRegex(DistributionError, "signature verification"):
            verify_manifest(payload, self.keyring, now=self.now)

        expired = self._manifest()
        expired["expires_at"] = (self.now - timedelta(seconds=1)).isoformat()
        expired["signatures"] = [self._signature(expired)]
        with self.assertRaisesRegex(DistributionError, "expired"):
            verify_manifest(expired, self.keyring, now=self.now)

        revoked = self._manifest()
        revoked["revoked"] = True
        revoked["signatures"] = [self._signature(revoked)]
        with self.assertRaisesRegex(DistributionError, "revoked"):
            verify_manifest(revoked, self.keyring, now=self.now)

        unknown = self._manifest()
        unknown["signatures"][0]["key_id"] = "other-key"
        with self.assertRaisesRegex(DistributionError, "active trusted key"):
            verify_manifest(unknown, self.keyring, now=self.now)

        with self.assertRaisesRegex(DistributionError, "expected 'candidate'"):
            verify_manifest(
                self._manifest(),
                self.keyring,
                expected_channel="candidate",
                now=self.now,
            )

    def test_offline_image_index_is_complete_and_digest_bound(self) -> None:
        with tempfile.TemporaryDirectory(prefix="govoplan-offline-images-") as value:
            root = Path(value)
            api = root / "api.oci.tar"
            web = root / "web.oci.tar"
            api.write_bytes(b"api archive")
            web.write_bytes(b"web archive")
            api_ref = "registry.example/govoplan/api@sha256:" + "a" * 64
            web_ref = "registry.example/govoplan/web@sha256:" + "b" * 64
            index = {
                "schema_version": "1",
                "images": [
                    {
                        "reference": api_ref,
                        "archive": api.name,
                        "sha256": hashlib.sha256(api.read_bytes()).hexdigest(),
                    },
                    {
                        "reference": web_ref,
                        "archive": web.name,
                        "sha256": hashlib.sha256(web.read_bytes()).hexdigest(),
                    },
                ],
            }

            paths = verify_offline_image_index(
                index,
                root=root,
                expected_references=(api_ref, web_ref),
            )
            self.assertEqual((api, web), paths)

            index["images"][1]["sha256"] = "0" * 64
            with self.assertRaisesRegex(DistributionError, "digest mismatch"):
                verify_offline_image_index(
                    index,
                    root=root,
                    expected_references=(api_ref, web_ref),
                )

    def _manifest(self) -> dict[str, object]:
        artifact = {"url": "https://downloads.example.test/artifact.json", "sha256": "d" * 64}
        manifest: dict[str, object] = {
            "schema_version": "1",
            "channel": "stable",
            "sequence": 1,
            "version": "1.2.3",
            "issued_at": (self.now - timedelta(minutes=1)).isoformat(),
            "expires_at": (self.now + timedelta(days=30)).isoformat(),
            "revoked": False,
            "deployer": {
                "url": "https://downloads.example.test/govoplan-deploy.pyz",
                "sha256": "e" * 64,
            },
            "images": {
                "api": {
                    "index": "registry.example/govoplan/api@sha256:" + "a" * 64,
                    "platforms": {
                        "linux/amd64": "registry.example/govoplan/api@sha256:" + "1" * 64,
                        "linux/arm64": "registry.example/govoplan/api@sha256:" + "2" * 64,
                    },
                    "sbom": dict(artifact),
                    "provenance": dict(artifact),
                },
                "web": {
                    "index": "registry.example/govoplan/web@sha256:" + "b" * 64,
                    "platforms": {
                        "linux/amd64": "registry.example/govoplan/web@sha256:" + "3" * 64,
                        "linux/arm64": "registry.example/govoplan/web@sha256:" + "4" * 64,
                    },
                    "sbom": dict(artifact),
                    "provenance": dict(artifact),
                },
            },
            "dependencies": {
                "postgres": "docker.io/library/postgres@sha256:" + "5" * 64,
                "redis": "docker.io/library/redis@sha256:" + "6" * 64,
                "load_balancer": "docker.io/library/haproxy@sha256:" + "7" * 64,
            },
            "composition": {
                "sha256": "c" * 64,
                "module_ids": ["access", "files"],
                "packages": [
                    {
                        "name": "govoplan-core",
                        "version": "1.2.3",
                        "wheel_sha256": "8" * 64,
                    }
                ],
            },
        }
        manifest["signatures"] = [self._signature(manifest)]
        return manifest

    def _signature(self, payload: dict[str, object]) -> dict[str, str]:
        return {
            "key_id": "release-1",
            "algorithm": "ed25519",
            "value": base64.b64encode(
                self.private.sign(canonical_signed_payload(payload))
            ).decode("ascii"),
        }


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