119 lines
4.8 KiB
Python
119 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Verify a GuestOps release archive and its immutable CI release record."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
|
|
|
|
SHA = re.compile(r"[0-9a-f]{40}")
|
|
DIGEST = re.compile(r"sha256:[0-9a-f]{64}")
|
|
|
|
|
|
def require(condition: bool, message: str) -> None:
|
|
if not condition:
|
|
raise ValueError(message)
|
|
|
|
|
|
def sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def loaded_image_id(reference: str) -> str:
|
|
result = subprocess.run(
|
|
["docker", "image", "inspect", "--format", "{{.Id}}", reference],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
return result.stdout.strip()
|
|
|
|
|
|
def validate(
|
|
archive: Path,
|
|
record_path: Path,
|
|
expected_commit: str,
|
|
expected_version: str,
|
|
verify_loaded_images: bool = False,
|
|
) -> dict[str, object]:
|
|
require(archive.is_file(), "Release archive does not exist.")
|
|
require(record_path.is_file(), "Release record does not exist.")
|
|
require(SHA.fullmatch(expected_commit) is not None, "Expected commit must be a full lowercase Git SHA.")
|
|
require(re.fullmatch(r"[0-9]+\.[0-9]+\.[0-9]+", expected_version) is not None,
|
|
"Expected version must use MAJOR.MINOR.PATCH.")
|
|
|
|
record = json.loads(record_path.read_text(encoding="utf-8"))
|
|
require(isinstance(record, dict), "Release record must be a JSON object.")
|
|
require(record.get("schemaVersion") == 1, "Unsupported release-record schema.")
|
|
require(record.get("commit") == expected_commit, "Release-record commit does not match the approved candidate.")
|
|
require(record.get("version") == expected_version, "Release-record version does not match the approved version.")
|
|
|
|
artifact = record.get("artifact")
|
|
require(isinstance(artifact, dict), "Release record has no artifact object.")
|
|
require(artifact.get("name") == archive.name, "Release archive filename does not match the record.")
|
|
require(artifact.get("size") == archive.stat().st_size, "Release archive size does not match the record.")
|
|
archive_sha = sha256(archive)
|
|
require(artifact.get("sha256") == archive_sha, "Release archive SHA-256 does not match the record.")
|
|
|
|
images = record.get("images")
|
|
require(isinstance(images, dict) and set(images) == {"api", "worker"},
|
|
"Release record must contain exactly API and worker images.")
|
|
expected_references = {
|
|
"api": f"guestops-api:{expected_commit}",
|
|
"worker": f"guestops-worker:{expected_commit}",
|
|
}
|
|
verified_images: dict[str, dict[str, str]] = {}
|
|
for name, reference in expected_references.items():
|
|
image = images.get(name)
|
|
require(isinstance(image, dict), f"Release record has no {name} image object.")
|
|
require(image.get("reference") == reference, f"{name} image reference is not bound to the full candidate SHA.")
|
|
image_id = str(image.get("id", ""))
|
|
require(DIGEST.fullmatch(image_id) is not None, f"{name} image ID is not an immutable SHA-256 digest.")
|
|
if verify_loaded_images:
|
|
require(loaded_image_id(reference) == image_id, f"Loaded {name} image ID does not match the release record.")
|
|
verified_images[name] = {"reference": reference, "id": image_id}
|
|
|
|
return {
|
|
"schemaVersion": 1,
|
|
"verified": True,
|
|
"commit": expected_commit,
|
|
"version": expected_version,
|
|
"archive": {"name": archive.name, "size": archive.stat().st_size, "sha256": archive_sha},
|
|
"releaseRecord": {"name": record_path.name, "sha256": sha256(record_path)},
|
|
"images": verified_images,
|
|
"loadedImageIdsVerified": verify_loaded_images,
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--archive", required=True, type=Path)
|
|
parser.add_argument("--record", required=True, type=Path)
|
|
parser.add_argument("--commit", required=True)
|
|
parser.add_argument("--version", required=True)
|
|
parser.add_argument("--verify-loaded-images", action="store_true")
|
|
parser.add_argument("--output", type=Path, help="Optional path for the non-sensitive verification summary.")
|
|
args = parser.parse_args()
|
|
result = validate(args.archive, args.record, args.commit, args.version, args.verify_loaded_images)
|
|
rendered = json.dumps(result, indent=2, sort_keys=True) + "\n"
|
|
if args.output:
|
|
args.output.write_text(rendered, encoding="utf-8")
|
|
print(rendered, end="")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except (OSError, ValueError, json.JSONDecodeError, subprocess.SubprocessError) as error:
|
|
print(f"Release verification failed: {error}", file=__import__("sys").stderr)
|
|
raise SystemExit(1)
|