Authenticate selective catalog base pairs
This commit is contained in:
184
tests/test_release_base_catalog_trust.py
Normal file
184
tests/test_release_base_catalog_trust.py
Normal file
@@ -0,0 +1,184 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user