deusdata--codebase-memory-mcp/scripts/ci/verify-release-selection.py

#!/usr/bin/env python3
"""Reconcile every canonical public release container to selected bytes."""

from __future__ import annotations

import argparse
import csv
import hashlib
import os
import pathlib
import re
import stat
import sys
import tarfile
import zipfile
from collections.abc import Sequence
from typing import BinaryIO, Optional


class ContractError(Exception):
    """The release-container namespace or selected-byte binding is invalid."""


TARGETS = (
    "linux-amd64",
    "linux-arm64",
    "linux-amd64-portable",
    "linux-arm64-portable",
    "darwin-amd64",
    "darwin-arm64",
    "windows-amd64",
    "windows-arm64",
)
VARIANTS = ("unstripped", "debug-stripped", "stripped")
FIELD_KEY = {variant: variant.replace("-", "_") for variant in VARIANTS}
SELECTION_FIELDS = (
    "target",
    "selected_variant",
    "selected_path",
    "selected_sha256",
    "selected_size",
    "decision",
    "unstripped_sha256",
    "unstripped_scan_path",
    "unstripped_classification",
    "unstripped_analysis_id",
    "unstripped_virustotal_url",
    "debug_stripped_sha256",
    "debug_stripped_scan_path",
    "debug_stripped_classification",
    "debug_stripped_analysis_id",
    "debug_stripped_virustotal_url",
    "stripped_sha256",
    "stripped_scan_path",
    "stripped_classification",
    "stripped_analysis_id",
    "stripped_virustotal_url",
)
SHA256_RE = re.compile(r"\A[0-9a-f]{64}\Z")
ANALYSIS_ID_RE = re.compile(r"\A[A-Za-z0-9_+/-]+={0,2}\Z")
MAX_MANIFEST_BYTES = 16 * 1024 * 1024
MAX_ARCHIVE_BYTES = 1024 * 1024 * 1024
MAX_MEMBER_BYTES = 512 * 1024 * 1024
MAX_TOTAL_MEMBER_BYTES = 2 * 1024 * 1024 * 1024
MAX_MEMBERS = 128


def archive_specs() -> dict[str, tuple[str, str]]:
    specs: dict[str, tuple[str, str]] = {}
    for target in TARGETS:
        binary = "codebase-memory-mcp.exe" if target.startswith("windows-") else "codebase-memory-mcp"
        suffix = ".zip" if target.startswith("windows-") else ".tar.gz"
        specs[f"codebase-memory-mcp-{target}{suffix}"] = (target, binary)
        if (
            target.startswith(("darwin-", "windows-"))
            or target.endswith("-portable")
        ):
            specs[f"codebase-memory-mcp-{target}.mcpb"] = (
                target,
                f"server/{binary}",
            )
    return specs


ARCHIVES = archive_specs()


def parse_arguments(argv: Sequence[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Bind all fourteen canonical release containers to release-selection.tsv."
    )
    parser.add_argument("--selection", required=True, type=pathlib.Path)
    parser.add_argument("--archive-dir", required=True, type=pathlib.Path)
    parser.add_argument(
        "--require-policy",
        choices=("virustotal-v2", "unscanned-dry-run"),
        help="reject a selection produced under any other policy",
    )
    return parser.parse_args(argv[1:])


def regular_status(path: pathlib.Path, *, ceiling: int, label: str) -> os.stat_result:
    try:
        status = path.lstat()
    except FileNotFoundError as error:
        raise ContractError(f"missing {label}: {path}") from error
    if not stat.S_ISREG(status.st_mode):
        raise ContractError(f"{label} is not a regular non-symlink file: {path}")
    if status.st_size > ceiling:
        raise ContractError(f"{label} exceeds the {ceiling}-byte ceiling: {path}")
    return status


def load_selection(
    path: pathlib.Path, *, require_policy: Optional[str] = None
) -> dict[str, dict[str, str]]:
    regular_status(path, ceiling=MAX_MANIFEST_BYTES, label="release selection")
    try:
        text = path.read_text(encoding="utf-8")
    except UnicodeError as error:
        raise ContractError("release selection is not UTF-8") from error
    if "\x00" in text or "\r" in text:
        raise ContractError("release selection contains forbidden control bytes")
    lines = text.splitlines()
    if not lines or lines[0] != "# cbm-release-selection-v1":
        raise ContractError("release selection marker is missing")
    metadata: dict[str, str] = {}
    cursor = 1
    while cursor < len(lines) and lines[cursor].startswith("# "):
        raw = lines[cursor][2:]
        key, separator, value = raw.partition("=")
        if not separator or not key or not value or key in metadata:
            raise ContractError(f"malformed or duplicate release-selection metadata: {raw!r}")
        metadata[key] = value
        cursor += 1
    if set(metadata) != {"policy", "targets", "candidates"}:
        raise ContractError("release selection metadata fields are incomplete or unexpected")
    if metadata["policy"] not in {"virustotal-v2", "unscanned-dry-run"}:
        raise ContractError("release selection has an invalid policy")
    if require_policy is not None and metadata["policy"] != require_policy:
        raise ContractError(
            f"release selection policy is {metadata['policy']!r}; required {require_policy!r}"
        )
    if metadata["targets"] != str(len(TARGETS)) or metadata["candidates"] != str(
        len(TARGETS) * len(VARIANTS)
    ):
        raise ContractError("release selection metadata does not bind the canonical matrix")
    reader = csv.DictReader(lines[cursor:], delimiter="\t")
    if tuple(reader.fieldnames or ()) != SELECTION_FIELDS:
        raise ContractError("release selection header is malformed or unexpected")
    rows = list(reader)
    if any(
        None in row
        or any(row.get(field) is None for field in SELECTION_FIELDS)
        or any(
            any(ord(character) < 32 for character in value)
            for value in row.values()
            if value
        )
        for row in rows
    ):
        raise ContractError("release selection has missing or surplus cells")
    if [row["target"] for row in rows] != list(TARGETS):
        raise ContractError("release selection is not the exact canonical target matrix")

    selections: dict[str, dict[str, str]] = {}
    analysis_ids: set[str] = set()
    for row in rows:
        target = row["target"]
        binary = "codebase-memory-mcp.exe" if target.startswith("windows-") else "codebase-memory-mcp"
        classifications = {
            variant: row[f"{FIELD_KEY[variant]}_classification"] for variant in VARIANTS
        }
        if metadata["policy"] == "virustotal-v2":
            if any(
                classification not in {"clean", "microsoft-ml"}
                for classification in classifications.values()
            ):
                raise ContractError(f"release selection contains a hard classification: {target}")
            for variant in VARIANTS:
                sha256 = row[f"{FIELD_KEY[variant]}_sha256"]
                analysis_id = row[f"{FIELD_KEY[variant]}_analysis_id"]
                expected_url = f"https://www.virustotal.com/gui/file/{sha256}/detection"
                if ANALYSIS_ID_RE.fullmatch(analysis_id) is None:
                    raise ContractError(f"release selection lacks VT analysis evidence: {target}")
                if analysis_id in analysis_ids:
                    raise ContractError("release selection reuses one VT analysis id")
                analysis_ids.add(analysis_id)
                if row[f"{FIELD_KEY[variant]}_virustotal_url"] != expected_url:
                    raise ContractError(f"release selection has unbound VT URL: {target}")
            # Re-derived here independently of the selector: smallest artifact
            # first, first CLEAN one wins, and only if every candidate drew the
            # tolerated verdict do we ship the smallest flagged one.
            order = ("stripped", "debug-stripped", "unstripped")
            clean = [v for v in order if classifications[v] == "clean"]
            if clean:
                expected_variant = clean[0]
                if expected_variant == "stripped":
                    expected_decision = "stripped-preferred"
                else:
                    others = "-".join(
                        f"{v}:{classifications[v]}" for v in order if v != expected_variant
                    )
                    expected_decision = f"{expected_variant}-clean-after-{others}"
            else:
                expected_variant = "stripped"
                expected_decision = "stripped-all-candidates-microsoft-ml"
        else:
            if any(
                classification != "unscanned-dry-run"
                for classification in classifications.values()
            ):
                raise ContractError(f"dry-run selection disguises unscanned evidence: {target}")
            if any(
                row[f"{FIELD_KEY[variant]}_{suffix}"]
                for variant in VARIANTS
                for suffix in ("analysis_id", "virustotal_url")
            ):
                raise ContractError(f"dry-run selection contains invented VT evidence: {target}")
            expected_variant = "stripped"
            expected_decision = "stripped-unscanned-dry-run"
        if (
            row["selected_variant"] != expected_variant
            or row["decision"] != expected_decision
        ):
            raise ContractError(f"release selection contradicts the policy truth table: {target}")
        selected_variant = expected_variant
        if row["selected_path"] != f"selected/{target}/{binary}":
            raise ContractError(f"release selection has unsafe selected path: {target}")
        if (
            SHA256_RE.fullmatch(row["selected_sha256"]) is None
            or row["selected_sha256"] != row[f"{FIELD_KEY[selected_variant]}_sha256"]
            or not row["selected_size"].isdigit()
            or not 0 < int(row["selected_size"]) <= MAX_MEMBER_BYTES
        ):
            raise ContractError(f"release selection has invalid selected content: {target}")
        for variant in VARIANTS:
            sha256 = row[f"{FIELD_KEY[variant]}_sha256"]
            scan_path = pathlib.PurePosixPath(row[f"{FIELD_KEY[variant]}_scan_path"])
            if (
                SHA256_RE.fullmatch(sha256) is None
                or scan_path.as_posix() != f"objects/{sha256}"
            ):
                raise ContractError(f"release selection has invalid sibling binding: {target}")
        if row["unstripped_sha256"] == row["stripped_sha256"]:
            raise ContractError(f"release selection siblings are not byte-distinct: {target}")
        selections[target] = row
    return selections


def safe_member_name(name: str) -> bool:
    if not name or "\\" in name or "\x00" in name or name.startswith("/"):
        return False
    pure = pathlib.PurePosixPath(name)
    return (
        not pure.is_absolute()
        and pure.as_posix() == name
        and all(part not in ("", ".", "..") for part in pure.parts)
    )


def hash_stream(stream: BinaryIO, *, declared_size: int, label: str) -> tuple[str, int]:
    if declared_size < 0 or declared_size > MAX_MEMBER_BYTES:
        raise ContractError(f"archive member has an invalid size: {label}")
    digest = hashlib.sha256()
    total = 0
    while True:
        chunk = stream.read(min(1024 * 1024, declared_size - total + 1))
        if not chunk:
            break
        total += len(chunk)
        if total > declared_size or total > MAX_MEMBER_BYTES:
            raise ContractError(f"archive member exceeded its declared size: {label}")
        digest.update(chunk)
    if total != declared_size:
        raise ContractError(f"archive member size does not match metadata: {label}")
    return digest.hexdigest(), total


def verify_tar(
    path: pathlib.Path, *, required_member: str, expected_sha256: str,
    expected_size: int, require_executable_mode: bool
) -> None:
    matches = 0
    try:
        with tarfile.open(path, mode="r:gz") as archive:
            members = archive.getmembers()
            if not members or len(members) > MAX_MEMBERS:
                raise ContractError(f"tar member count is invalid: {path.name}")
            if sum(member.size for member in members) > MAX_TOTAL_MEMBER_BYTES:
                raise ContractError(f"tar exceeds total member-size ceiling: {path.name}")
            seen: set[str] = set()
            for member in members:
                if not safe_member_name(member.name) or member.name in seen:
                    raise ContractError(f"tar has unsafe or duplicate member: {path.name}:{member.name}")
                seen.add(member.name)
                if member.issym() or member.islnk() or member.isdev() or member.isfifo():
                    raise ContractError(f"tar contains a special member: {path.name}:{member.name}")
                if member.size > MAX_MEMBER_BYTES:
                    raise ContractError(f"tar member exceeds size ceiling: {path.name}:{member.name}")
                if member.name != required_member:
                    continue
                if not member.isfile():
                    raise ContractError(f"required tar executable is not regular: {path.name}")
                if require_executable_mode and stat.S_IMODE(member.mode) != 0o755:
                    raise ContractError(
                        f"required tar executable has wrong mode: {path.name}:{member.name}"
                    )
                matches += 1
                handle = archive.extractfile(member)
                if handle is None:
                    raise ContractError(f"required tar executable is unreadable: {path.name}")
                with handle:
                    actual_sha256, actual_size = hash_stream(
                        handle, declared_size=member.size, label=f"{path.name}:{member.name}"
                    )
                if actual_sha256 != expected_sha256 or actual_size != expected_size:
                    raise ContractError(f"tar executable does not match selection: {path.name}")
    except (tarfile.TarError, EOFError) as error:
        raise ContractError(f"malformed tar archive: {path.name}") from error
    if matches != 1:
        raise ContractError(f"tar must contain exactly one required executable: {path.name}")


def zip_entry_kind(info: zipfile.ZipInfo) -> str:
    mode = (info.external_attr >> 16) & 0xFFFF
    if mode and stat.S_IFMT(mode) not in (0, stat.S_IFREG, stat.S_IFDIR):
        return "special"
    return "directory" if info.is_dir() else "regular"


def verify_zip(
    path: pathlib.Path, *, required_member: str, expected_sha256: str,
    expected_size: int, require_executable_mode: bool
) -> None:
    matches = 0
    try:
        with zipfile.ZipFile(path, mode="r") as archive:
            members = archive.infolist()
            if not members or len(members) > MAX_MEMBERS:
                raise ContractError(f"zip member count is invalid: {path.name}")
            if sum(member.file_size for member in members) > MAX_TOTAL_MEMBER_BYTES:
                raise ContractError(f"zip exceeds total member-size ceiling: {path.name}")
            seen: set[str] = set()
            for member in members:
                name = member.filename
                if not safe_member_name(name) or name in seen:
                    raise ContractError(f"zip has unsafe or duplicate member: {path.name}:{name}")
                seen.add(name)
                if zip_entry_kind(member) == "special":
                    raise ContractError(f"zip contains a special member: {path.name}:{name}")
                if member.file_size > MAX_MEMBER_BYTES:
                    raise ContractError(f"zip member exceeds size ceiling: {path.name}:{name}")
                if name != required_member:
                    continue
                if zip_entry_kind(member) != "regular":
                    raise ContractError(f"required zip executable is not regular: {path.name}")
                mode = (member.external_attr >> 16) & 0xFFFF
                if require_executable_mode and stat.S_IMODE(mode) != 0o755:
                    raise ContractError(
                        f"required zip executable has wrong mode: {path.name}:{name}"
                    )
                matches += 1
                with archive.open(member, mode="r") as handle:
                    actual_sha256, actual_size = hash_stream(
                        handle, declared_size=member.file_size, label=f"{path.name}:{name}"
                    )
                if actual_sha256 != expected_sha256 or actual_size != expected_size:
                    raise ContractError(f"zip executable does not match selection: {path.name}")
    except (zipfile.BadZipFile, EOFError, RuntimeError) as error:
        raise ContractError(f"malformed zip archive: {path.name}") from error
    if matches != 1:
        raise ContractError(f"zip must contain exactly one required executable: {path.name}")


def discover_archives(root: pathlib.Path) -> dict[str, pathlib.Path]:
    try:
        root_status = root.lstat()
    except FileNotFoundError as error:
        raise ContractError(f"archive directory does not exist: {root}") from error
    if not stat.S_ISDIR(root_status.st_mode):
        raise ContractError(f"archive path is not a regular directory: {root}")

    discovered: dict[str, pathlib.Path] = {}
    archive_like: list[pathlib.Path] = []
    for directory, names, files in os.walk(root, followlinks=False):
        parent = pathlib.Path(directory)
        for name in names:
            child = parent / name
            if child.is_symlink():
                raise ContractError(f"symlinked directory in archive tree: {child}")
        for name in files:
            child = parent / name
            if child.is_symlink():
                raise ContractError(f"symlinked file in archive tree: {child}")
            if name.endswith((".tar.gz", ".zip", ".mcpb")):
                archive_like.append(child)
                if name in discovered:
                    raise ContractError(f"duplicate archive basename: {name}")
                discovered[name] = child
    actual_names = set(discovered)
    expected_names = set(ARCHIVES)
    if actual_names != expected_names or len(archive_like) != len(expected_names):
        missing = sorted(expected_names - actual_names)
        surplus = sorted(actual_names - expected_names)
        raise ContractError(
            f"archive namespace is not the exact canonical set; missing={missing}, surplus={surplus}"
        )
    return discovered


def main(argv: Sequence[str]) -> None:
    args = parse_arguments(argv)
    selections = load_selection(args.selection, require_policy=args.require_policy)
    archives = discover_archives(args.archive_dir)
    for name, (target, member) in ARCHIVES.items():
        path = archives[name]
        regular_status(path, ceiling=MAX_ARCHIVE_BYTES, label="release archive")
        selection = selections[target]
        verify = verify_tar if name.endswith(".tar.gz") else verify_zip
        verify(
            path,
            required_member=member,
            expected_sha256=selection["selected_sha256"],
            expected_size=int(selection["selected_size"]),
            require_executable_mode=not target.startswith("windows-"),
        )
    print(f"verified all {len(ARCHIVES)} canonical containers against selected bytes")


try:
    main(sys.argv)
except (ContractError, OSError, UnicodeError, ValueError) as error:
    print(f"verify-release-selection: {error}", file=sys.stderr)
    raise SystemExit(1)