88 lines
3.9 KiB
Python
88 lines
3.9 KiB
Python
import hashlib
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
spec = importlib.util.spec_from_file_location("verify_release", ROOT / "deploy" / "verify_release.py")
|
|
verify_release = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(verify_release)
|
|
|
|
|
|
COMMIT = "a" * 40
|
|
API_ID = "sha256:" + "b" * 64
|
|
WORKER_ID = "sha256:" + "c" * 64
|
|
|
|
|
|
class VerifyReleaseTests(unittest.TestCase):
|
|
def fixture(self, directory: str):
|
|
root = Path(directory)
|
|
archive = root / "guestops-images.tar.gz"
|
|
record = root / "release-record.json"
|
|
archive.write_bytes(b"reviewed image archive")
|
|
release = {
|
|
"schemaVersion": 1,
|
|
"version": "0.2.0",
|
|
"commit": COMMIT,
|
|
"artifact": {
|
|
"name": archive.name,
|
|
"size": archive.stat().st_size,
|
|
"sha256": hashlib.sha256(archive.read_bytes()).hexdigest(),
|
|
},
|
|
"images": {
|
|
"api": {"reference": f"guestops-api:{COMMIT}", "id": API_ID},
|
|
"worker": {"reference": f"guestops-worker:{COMMIT}", "id": WORKER_ID},
|
|
},
|
|
}
|
|
record.write_text(json.dumps(release), encoding="utf-8")
|
|
return archive, record, release
|
|
|
|
def test_verifies_candidate_archive_record_and_record_checksum(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
archive, record, _ = self.fixture(directory)
|
|
result = verify_release.validate(archive, record, COMMIT, "0.2.0")
|
|
self.assertTrue(result["verified"])
|
|
self.assertEqual(result["archive"]["sha256"], hashlib.sha256(archive.read_bytes()).hexdigest())
|
|
self.assertEqual(result["releaseRecord"]["sha256"], hashlib.sha256(record.read_bytes()).hexdigest())
|
|
self.assertFalse(result["loadedImageIdsVerified"])
|
|
|
|
def test_rejects_changed_archive(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
archive, record, _ = self.fixture(directory)
|
|
archive.write_bytes(b"changed")
|
|
with self.assertRaisesRegex(ValueError, "size does not match"):
|
|
verify_release.validate(archive, record, COMMIT, "0.2.0")
|
|
|
|
def test_rejects_wrong_commit_version_reference_and_mutable_id(self):
|
|
cases = [
|
|
(lambda value: value.update(commit="d" * 40), "commit does not match"),
|
|
(lambda value: value.update(version="0.3.0"), "version does not match"),
|
|
(lambda value: value["images"]["api"].update(reference="guestops-api:latest"), "full candidate SHA"),
|
|
(lambda value: value["images"]["worker"].update(id="worker-image"), "immutable SHA-256"),
|
|
]
|
|
for mutate, message in cases:
|
|
with self.subTest(message=message), tempfile.TemporaryDirectory() as directory:
|
|
archive, record, release = self.fixture(directory)
|
|
mutate(release)
|
|
record.write_text(json.dumps(release), encoding="utf-8")
|
|
with self.assertRaisesRegex(ValueError, message):
|
|
verify_release.validate(archive, record, COMMIT, "0.2.0")
|
|
|
|
def test_loaded_image_ids_must_match(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
archive, record, _ = self.fixture(directory)
|
|
with patch.object(verify_release, "loaded_image_id", side_effect=[API_ID, WORKER_ID]):
|
|
result = verify_release.validate(archive, record, COMMIT, "0.2.0", True)
|
|
self.assertTrue(result["loadedImageIdsVerified"])
|
|
with patch.object(verify_release, "loaded_image_id", return_value="sha256:" + "d" * 64):
|
|
with self.assertRaisesRegex(ValueError, "Loaded api image ID"):
|
|
verify_release.validate(archive, record, COMMIT, "0.2.0", True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|