broker/scripts/build_cli.py

#!/usr/bin/env python3
"""Build the coreci-broker CLI zipapp artifact (stdlib only).

Usage: python3 scripts/build_cli.py <version>

Stages an explicit allowlist into a temp dir, packs it with
``python3 -m zipapp ... --python "/usr/bin/env python3" --compress``, then
runs a normalize post-pass (G-009) so same-environment rebuilds are
byte-identical: entries sorted by name, ``date_time=(1980,1,1,0,0,0)``,
``ZIP_DEFLATED``, ``external_attr`` set to constants (files
``0o100644 << 16``, dirs ``0o40755 << 16`` — never preserved/umask-derived),
shebang re-prepended, exec bit preserved. Prints the artifact sha256.
"""

import argparse
import hashlib
import io
import os
import shutil
import stat
import subprocess
import sys
import tempfile
import zipfile

ROOT_SOURCES = ("coreci_broker.py", "coreci_broker_client.py")
UI_SOURCE_DIR = "coreci_broker_ui"
SHIM_SOURCE = os.path.join("cli", "__main__.py")
STAGED_SHIM = "__main__.py"
VERSION_MODULE = "_cli_version.py"
FIXED_DATE_TIME = (1980, 1, 1, 0, 0, 0)
FILE_EXTERNAL_ATTR = 0o100644 << 16
DIR_EXTERNAL_ATTR = 0o40755 << 16


def repo_root():
    return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


def artifact_name(version):
    tag = version[1:] if version.startswith("v") else version
    return "coreci-broker-v{}.pyz".format(tag)


def stage_sources(repo, staging):
    for name in ROOT_SOURCES:
        shutil.copyfile(os.path.join(repo, name), os.path.join(staging, name))
    ui_root = os.path.join(repo, UI_SOURCE_DIR)
    staged_ui = os.path.join(staging, UI_SOURCE_DIR)
    for current, _dirs, files in os.walk(ui_root):
        relative = os.path.relpath(current, ui_root)
        target_dir = staged_ui if relative == "." else os.path.join(staged_ui, relative)
        os.makedirs(target_dir, exist_ok=True)
        for name in sorted(files):
            shutil.copyfile(os.path.join(current, name), os.path.join(target_dir, name))
    shutil.copyfile(os.path.join(repo, SHIM_SOURCE), os.path.join(staging, STAGED_SHIM))


def write_version_module(staging, version):
    with open(os.path.join(staging, VERSION_MODULE), "w", encoding="utf-8") as handle:
        handle.write('VERSION = "{}"\n'.format(version))


def run_zipapp(staging, artifact):
    command = [
        sys.executable, "-m", "zipapp", staging,
        "-o", artifact,
        "--python", "/usr/bin/env python3",
        "--compress",
    ]
    completed = subprocess.run(command, capture_output=True, text=True)
    if completed.returncode != 0:
        detail = (completed.stderr or completed.stdout).strip()
        raise RuntimeError(
            "zipapp failed with code {}: {}".format(completed.returncode, detail)
        )


def normalize_archive(artifact):
    with open(artifact, "rb") as handle:
        raw = handle.read()
    if not raw.startswith(b"#!"):
        raise RuntimeError("zipapp artifact is missing its shebang prefix")
    shebang = raw.split(b"\n", 1)[0] + b"\n"
    entries = []
    with zipfile.ZipFile(io.BytesIO(raw)) as archive:
        for info in archive.infolist():
            payload = b"" if info.is_dir() else archive.read(info)
            entries.append((info.filename, payload))
    entries.sort(key=lambda entry: entry[0])
    body = io.BytesIO()
    with zipfile.ZipFile(body, "w", compression=zipfile.ZIP_DEFLATED) as archive:
        for name, payload in entries:
            info = zipfile.ZipInfo(name, date_time=FIXED_DATE_TIME)
            info.compress_type = zipfile.ZIP_DEFLATED
            info.external_attr = DIR_EXTERNAL_ATTR if name.endswith("/") else FILE_EXTERNAL_ATTR
            archive.writestr(info, payload)
    mode = os.stat(artifact).st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH
    temp_path = artifact + ".tmp"
    with open(temp_path, "wb") as handle:
        handle.write(shebang)
        handle.write(body.getvalue())
    os.replace(temp_path, artifact)
    os.chmod(artifact, mode)


def sha256_file(path):
    digest = hashlib.sha256()
    with open(path, "rb") as handle:
        for chunk in iter(lambda: handle.read(65536), b""):
            digest.update(chunk)
    return digest.hexdigest()


def build(version, repo=None, dist_dir=None):
    repo = os.path.abspath(repo or repo_root())
    dist_dir = dist_dir or os.path.join(repo, "dist")
    os.makedirs(dist_dir, exist_ok=True)
    artifact = os.path.join(dist_dir, artifact_name(version))
    with tempfile.TemporaryDirectory(prefix="coreci-broker-cli-") as staging:
        stage_sources(repo, staging)
        write_version_module(staging, version)
        run_zipapp(staging, artifact)
    normalize_archive(artifact)
    return artifact


def main(argv=None):
    parser = argparse.ArgumentParser(
        description="Build the coreci-broker CLI zipapp artifact."
    )
    parser.add_argument("version", help="release version of the artifact, e.g. v0.2.1")
    args = parser.parse_args(argv)
    artifact = build(args.version)
    print(artifact)
    print("sha256: {}".format(sha256_file(artifact)))
    return 0


if __name__ == "__main__":
    sys.exit(main())