GuestOps/deploy/verify_source_package.py

105 lines
3.9 KiB
Python

#!/usr/bin/env python3
"""Verify a GuestOps source archive against its deterministic source record."""
from __future__ import annotations
import argparse
import hashlib
import json
from pathlib import Path
import re
import sys
FULL_SHA = re.compile(r"[0-9a-f]{40}")
SHA256 = re.compile(r"[0-9a-f]{64}")
SEMVER = re.compile(r"[0-9]+\.[0-9]+\.[0-9]+")
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 validate(
archive: Path,
record_path: Path,
expected_commit: str | None = None,
expected_version: str | None = None,
) -> dict[str, object]:
require(archive.is_file(), "Source archive does not exist.")
require(record_path.is_file(), "Source record does not exist.")
record = json.loads(record_path.read_text(encoding="utf-8"))
require(isinstance(record, dict), "Source record must be a JSON object.")
require(record.get("schemaVersion") == 1, "Unsupported source-record schema.")
commit = record.get("commit")
version = record.get("version")
require(isinstance(commit, str) and FULL_SHA.fullmatch(commit) is not None,
"Source record commit must be a full lowercase Git SHA.")
require(isinstance(version, str) and SEMVER.fullmatch(version) is not None,
"Source record version must use MAJOR.MINOR.PATCH.")
if expected_commit is not None:
require(FULL_SHA.fullmatch(expected_commit) is not None,
"Expected commit must be a full lowercase Git SHA.")
require(commit == expected_commit, "Source-record commit does not match the selected release.")
if expected_version is not None:
require(SEMVER.fullmatch(expected_version) is not None,
"Expected version must use MAJOR.MINOR.PATCH.")
require(version == expected_version, "Source-record version does not match the selected release.")
artifact = record.get("artifact")
require(isinstance(artifact, dict), "Source record has no artifact object.")
require(artifact.get("name") == archive.name, "Source archive filename does not match the record.")
require(artifact.get("size") == archive.stat().st_size, "Source archive size does not match the record.")
recorded_sha = artifact.get("sha256")
require(isinstance(recorded_sha, str) and SHA256.fullmatch(recorded_sha) is not None,
"Source archive record must contain a lowercase SHA-256 digest.")
actual_sha = sha256(archive)
require(recorded_sha == actual_sha, "Source archive SHA-256 does not match the record.")
return {
"artifact": {"name": archive.name, "sha256": actual_sha, "size": archive.stat().st_size},
"commit": commit,
"schemaVersion": 1,
"sourceRecord": {"name": record_path.name, "sha256": sha256(record_path)},
"verified": True,
"version": version,
}
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("--expected-commit")
parser.add_argument("--expected-version")
parser.add_argument("--output", type=Path)
args = parser.parse_args()
try:
result = validate(
args.archive,
args.record,
args.expected_commit,
args.expected_version,
)
except (OSError, ValueError, json.JSONDecodeError) as error:
print(f"Source package verification failed: {error}", file=sys.stderr)
raise SystemExit(1)
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__":
main()