geointel/scripts/build_release_package.py

#!/usr/bin/env python3
"""Build or verify a checksummed and SSH-signed GeoIntel release package."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Sequence


ROOT = Path(__file__).resolve().parents[1]
MANIFEST_NAME = "release-manifest.json"
SIGNATURE_NAME = f"{MANIFEST_NAME}.sig"
SIGNERS_NAME = "allowed_signers"
CHECKSUMS_NAME = "CHECKSUMS.sha256"
GENERATED_NAMES = {MANIFEST_NAME, SIGNATURE_NAME, SIGNERS_NAME, CHECKSUMS_NAME}
RELEASE_ID_RE = re.compile(r"^v[0-9]+\.[0-9]+\.[0-9]+(?:-[0-9A-Za-z.-]+)?$")


def run(
    command: Sequence[str],
    *,
    input_bytes: bytes | None = None,
    cwd: Path = ROOT,
) -> subprocess.CompletedProcess[bytes]:
    result = subprocess.run(
        list(command),
        cwd=cwd,
        input=input_bytes,
        capture_output=True,
        check=False,
    )
    if result.returncode != 0:
        stderr = result.stderr.decode("utf-8", errors="replace").strip()
        raise RuntimeError(f"{' '.join(command)} failed: {stderr}")
    return result


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def package_files(package_dir: Path, *, include_generated: bool) -> list[Path]:
    result: list[Path] = []
    for path in sorted(package_dir.rglob("*"), key=lambda item: item.as_posix()):
        if path.is_symlink():
            raise RuntimeError(f"Release packages may not contain symlinks: {path}")
        if not path.is_file():
            continue
        relative = path.relative_to(package_dir)
        if not include_generated and relative.as_posix() in GENERATED_NAMES:
            continue
        result.append(path)
    return result


def public_key(signing_key: Path) -> str:
    public_path = Path(f"{signing_key}.pub")
    if public_path.is_file():
        value = public_path.read_text(encoding="utf-8").strip()
    else:
        value = run(("ssh-keygen", "-y", "-f", str(signing_key))).stdout.decode(
            "utf-8"
        ).strip()
    if not value.startswith(("ssh-ed25519 ", "ssh-rsa ", "ecdsa-sha2-")):
        raise RuntimeError("Unsupported or invalid SSH public signing key")
    return " ".join(value.split()[:2])


def write_checksums(package_dir: Path) -> None:
    lines = []
    for path in package_files(package_dir, include_generated=True):
        relative = path.relative_to(package_dir).as_posix()
        if relative == CHECKSUMS_NAME:
            continue
        lines.append(f"{sha256(path)}  {relative}")
    (package_dir / CHECKSUMS_NAME).write_text(
        "\n".join(lines) + "\n",
        encoding="utf-8",
        newline="\n",
    )


def verify_checksums(package_dir: Path) -> None:
    checksum_path = package_dir / CHECKSUMS_NAME
    if not checksum_path.is_file():
        raise RuntimeError(f"Missing {CHECKSUMS_NAME}")
    expected_paths: set[str] = set()
    for line in checksum_path.read_text(encoding="utf-8").splitlines():
        digest, separator, relative = line.partition("  ")
        if not separator or not re.fullmatch(r"[0-9a-f]{64}", digest):
            raise RuntimeError(f"Invalid checksum line: {line!r}")
        candidate = (package_dir / relative).resolve()
        try:
            candidate.relative_to(package_dir)
        except ValueError as exc:
            raise RuntimeError(f"Checksum path escapes package: {relative}") from exc
        if not candidate.is_file() or candidate.is_symlink():
            raise RuntimeError(f"Checksummed release file is unavailable: {relative}")
        if sha256(candidate) != digest:
            raise RuntimeError(f"Checksum mismatch: {relative}")
        expected_paths.add(relative)
    actual_paths = {
        path.relative_to(package_dir).as_posix()
        for path in package_files(package_dir, include_generated=True)
        if path.relative_to(package_dir).as_posix() != CHECKSUMS_NAME
    }
    if expected_paths != actual_paths:
        raise RuntimeError(
            "Checksum inventory differs from package contents: "
            f"missing={sorted(actual_paths - expected_paths)}, "
            f"unexpected={sorted(expected_paths - actual_paths)}"
        )


def verify_package(package_dir: Path) -> dict[str, object]:
    package_dir = package_dir.expanduser().resolve()
    verify_checksums(package_dir)
    manifest_path = package_dir / MANIFEST_NAME
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    identity = str(manifest["signature"]["identity"])
    namespace = str(manifest["signature"]["namespace"])
    run(
        (
            "ssh-keygen",
            "-Y",
            "verify",
            "-f",
            str(package_dir / SIGNERS_NAME),
            "-I",
            identity,
            "-n",
            namespace,
            "-s",
            str(package_dir / SIGNATURE_NAME),
        ),
        input_bytes=manifest_path.read_bytes(),
    )
    if manifest.get("scope") != "Belgium and the Belgian North Sea":
        raise RuntimeError("Unexpected release scope")
    if manifest.get("release_id") != f"v{manifest.get('version')}":
        raise RuntimeError("Release id and semantic version differ")
    return manifest


def build_package(args: argparse.Namespace) -> dict[str, object]:
    package_dir = args.output_dir.expanduser().resolve()
    package_dir.mkdir(parents=True, exist_ok=True)
    version = (ROOT / "VERSION").read_text(encoding="utf-8").strip()
    release_id = args.release_id or f"v{version}"
    if not RELEASE_ID_RE.fullmatch(release_id) or release_id != f"v{version}":
        raise RuntimeError("Release id must equal v<VERSION> and be valid SemVer")

    commit = run(("git", "rev-parse", "HEAD")).stdout.decode().strip()
    dirty = run(("git", "status", "--porcelain=v1")).stdout.decode().strip()
    if dirty:
        raise RuntimeError("Release package requires a clean Git worktree")
    tag_commit = run(("git", "rev-list", "-n", "1", release_id)).stdout.decode().strip()
    if tag_commit != commit:
        raise RuntimeError(f"Tag {release_id} does not point to HEAD")
    if args.image_revision != commit:
        raise RuntimeError("Image revision must equal the tagged Git commit")

    evidence = []
    for path in package_files(package_dir, include_generated=False):
        evidence.append(
            {
                "path": path.relative_to(package_dir).as_posix(),
                "size_bytes": path.stat().st_size,
                "sha256": sha256(path),
            }
        )
    if not evidence:
        raise RuntimeError("At least one release evidence file is required")

    signing_key = args.signing_key.expanduser().resolve()
    if not signing_key.is_file():
        raise RuntimeError(f"SSH signing key is unavailable: {signing_key}")
    identity = args.identity
    namespace = "geointel-release"
    (package_dir / SIGNERS_NAME).write_text(
        f"{identity} {public_key(signing_key)}\n",
        encoding="utf-8",
        newline="\n",
    )
    manifest: dict[str, object] = {
        "schema_version": 1,
        "release_id": release_id,
        "version": version,
        "scope": "Belgium and the Belgian North Sea",
        "created_at": datetime.now(timezone.utc).isoformat(),
        "git": {
            "commit": commit,
            "tag": release_id,
            "clean": True,
        },
        "image": {
            "name": args.image_name,
            "id": args.image_id,
            "revision": args.image_revision,
        },
        "signature": {
            "algorithm": "SSH",
            "identity": identity,
            "namespace": namespace,
        },
        "evidence": evidence,
    }
    manifest_path = package_dir / MANIFEST_NAME
    manifest_path.write_text(
        json.dumps(manifest, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
        newline="\n",
    )
    signature_path = package_dir / SIGNATURE_NAME
    if signature_path.exists():
        signature_path.unlink()
    run(
        (
            "ssh-keygen",
            "-Y",
            "sign",
            "-f",
            str(signing_key),
            "-n",
            namespace,
            str(manifest_path),
        )
    )
    if not signature_path.is_file():
        raise RuntimeError("ssh-keygen did not create the detached signature")
    write_checksums(package_dir)
    return verify_package(package_dir)


def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    subparsers = parser.add_subparsers(dest="command", required=True)
    build = subparsers.add_parser("build")
    build.add_argument("--output-dir", type=Path, required=True)
    build.add_argument("--release-id")
    build.add_argument("--image-name", required=True)
    build.add_argument("--image-id", required=True)
    build.add_argument("--image-revision", required=True)
    build.add_argument(
        "--signing-key",
        type=Path,
        default=os.environ.get("GEOINTEL_RELEASE_SIGNING_KEY"),
        required=not bool(os.environ.get("GEOINTEL_RELEASE_SIGNING_KEY")),
    )
    build.add_argument("--identity", default="geointel-release")
    verify = subparsers.add_parser("verify")
    verify.add_argument("--package-dir", type=Path, required=True)
    return parser.parse_args(argv)


def main(argv: Sequence[str] | None = None) -> int:
    args = parse_args(argv)
    try:
        if args.command == "build":
            manifest = build_package(args)
            print(
                f"Built and verified signed release package "
                f"{manifest['release_id']} at {args.output_dir.resolve()}"
            )
        else:
            manifest = verify_package(args.package_dir)
            print(
                f"Verified signed release package "
                f"{manifest['release_id']} at {args.package_dir.resolve()}"
            )
    except (OSError, KeyError, ValueError, RuntimeError, json.JSONDecodeError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())