202 lines
7.5 KiB
Python
202 lines
7.5 KiB
Python
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)
|
|
|
|
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()
|