diff --git a/changelog.d/2026-07-17-build-archive-bundle-source.md b/changelog.d/2026-07-17-build-archive-bundle-source.md new file mode 100644 index 0000000..a750b4d --- /dev/null +++ b/changelog.d/2026-07-17-build-archive-bundle-source.md @@ -0,0 +1,3 @@ +### Changed +- `build_plugin_archive.py` now sources the activation runtime bundle from an out-of-tree signed handoff via `--bundle-source ` instead of the git checkout — the signed runtime ships as a release asset and is never committed. The committed `runtime-manifest.json` alone decides the mode (design decision D1): an activation manifest without `--bundle-source` is a hard error (never a silent policy-only downgrade), a policy-only manifest forbids the flag, and an in-tree `runtime/` conflicts with an external source. The bundle source is treated as hostile input (symlink argument rejected before resolution; bounded non-following enumeration; member-set equality; per-member O_NOFOLLOW regular-file/uid/nlink/mode/size/sha256 checks with the strict 0o500 install mode preserved), and its payloads are re-read + re-digested through fresh descriptors at pack time. +- `verify_archive` is rearchitected to **regenerate the canonical tar from trusted inputs and byte-compare** rather than parse a possibly-hostile archive: it never hands candidate tar bytes to `tarfile`. A shared deterministic `USTAR` serializer (used by both build and verify) reproduces the canonical tar with runtime payloads zero-filled; the candidate is only bounded-inflated and then sliced at offsets the verifier computed. Every structural property — headers, exact typeflags, ordering, padding, EOF, dialect, and the absence of hidden PAX/GNU/sparse records — is bound by the byte-compare outside the runtime ranges; each runtime payload is bound to the frozen-manifest digest; the archived `runtime-manifest.json` bytes are bound to the frozen snapshot. Reading the candidate is fd-only (`O_NOFOLLOW`/`O_NONBLOCK` + `fstat`, no path-swap window) under hard compressed- and decompressed-size caps, with a canonical gzip-header check and single-stream (no trailing/concatenated data) validation. The archive is built into an exclusive temp that is verified before an atomic rename, so a failed build never leaves a publishable artifact. Hardened through a 2-round up-front adversarial design review of the hostile-archive-reading threat model (the regenerate-and-compare pivot) before implementation. diff --git a/scripts/build_plugin_archive.py b/scripts/build_plugin_archive.py index a13c0b2..7201c9d 100644 --- a/scripts/build_plugin_archive.py +++ b/scripts/build_plugin_archive.py @@ -7,13 +7,14 @@ import gzip import hashlib import importlib.util +import io import json import os import re import stat import tarfile +import zlib from pathlib import Path, PurePosixPath -from typing import Iterable try: from scripts.skill_structure import expected_skill_relpaths, skill_tree_differences @@ -174,14 +175,50 @@ def _safe_source(path: Path) -> os.stat_result: return info -def _load_manifest(plugin_path: Path) -> dict[str, object]: +def _read_manifest_bytes(plugin_path: Path) -> bytes: + """Read the committed manifest exactly once for a frozen-snapshot check. + + The read goes through a single O_NOFOLLOW descriptor whose fstat is + checked, so the snapshot is one coherent read of one regular file — a + concurrent swap yields either the old bytes or the new bytes, never a + mixture, and everything downstream derives from the returned snapshot. + """ + manifest_path = plugin_path / "runtime-manifest.json" info = _safe_source(manifest_path) if not stat.S_ISREG(info.st_mode): raise ValueError("runtime-manifest.json is not a regular file") + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0) + try: + descriptor = os.open(manifest_path, flags) + except OSError as exc: + raise ValueError("runtime manifest is unreadable") from exc try: - manifest = json.loads(manifest_path.read_text(encoding="utf-8")) - except (OSError, UnicodeError, ValueError, RecursionError) as exc: + described = os.fstat(descriptor) + if not stat.S_ISREG(described.st_mode): + raise ValueError("runtime-manifest.json is not a regular file") + chunks = [] + while True: + chunk = os.read(descriptor, 1024 * 1024) + if not chunk: + break + chunks.append(chunk) + if sum(len(part) for part in chunks) > MAX_ARTIFACT_BYTES: + raise ValueError("runtime manifest is unreasonably large") + return b"".join(chunks) + except OSError as exc: + raise ValueError("runtime manifest is unreadable") from exc + finally: + try: + os.close(descriptor) + except OSError: + pass + + +def _parse_manifest(data: bytes) -> dict[str, object]: + try: + manifest = json.loads(data.decode("utf-8")) + except (UnicodeError, ValueError, RecursionError) as exc: raise ValueError("runtime manifest is unreadable") from exc if ( not isinstance(manifest, dict) @@ -208,7 +245,20 @@ def _load_manifest(plugin_path: Path) -> dict[str, object]: return manifest -def _validate_activation(plugin_path: Path, item: object) -> None: +def _load_manifest(plugin_path: Path) -> dict[str, object]: + return _parse_manifest(_read_manifest_bytes(plugin_path)) + + +def _validate_activation_manifest(item: object) -> tuple[dict[str, object], ...]: + """Validate the activation artifact SHAPE from the committed manifest only. + + Deliberately touches no filesystem bundle state: with the signed runtime + shipped as a release asset (never committed), the committed manifest is the + activation marker (design decision D1) and the physical bundle bytes are + validated separately against wherever they are sourced from + (`_validate_activation_bundle_tree`). + """ + fields = { "platform", "arch", @@ -289,57 +339,132 @@ def _validate_activation(plugin_path: Path, item: object) -> None: or len(contracts_value) != len(contracts) ): raise ValueError("activation runtime manifest fields are invalid") + return records - runtime_root = plugin_path / "runtime" - allowed = { - Path("runtime"), - Path("runtime/darwin-arm64"), - RUNTIME_BUNDLE_REL, - *(RUNTIME_BUNDLE_REL / record["path"] for record in records), - } - observed = {path.relative_to(plugin_path) for path in runtime_root.rglob("*")} - observed.add(Path("runtime")) - if observed != allowed: - raise ValueError("activation package runtime bundle membership is not exact") - bundle = plugin_path / RUNTIME_BUNDLE_REL - bundle_info = _safe_source(bundle) + +def _validate_activation_bundle_tree( + bundle_leaf: Path, records: tuple[dict[str, object], ...] +) -> None: + """Validate the physical runtime bundle leaf directory against the manifest. + + The leaf is HOSTILE input (an operator-supplied out-of-tree handoff): + enumerate it bounded and non-following, require member-set EQUALITY with + the manifest records, and admit only regular files owned by the invoking + user with nlink 1, the exact install mode, exact size, and exact sha256. + Devices, FIFOs, sockets, and symlinks all fail the S_ISREG/O_NOFOLLOW + checks. `validate_file_records` has already rejected traversal, separator + confusables, and Unicode-normalization/case-fold collisions in the paths. + """ + + try: + leaf_info = bundle_leaf.lstat() + except OSError as exc: + raise ValueError("runtime bundle source is missing") from exc if ( - not stat.S_ISDIR(bundle_info.st_mode) - or bundle_info.st_uid != os.getuid() - or stat.S_IMODE(bundle_info.st_mode) != runtime_bundle.INSTALL_MODE + stat.S_ISLNK(leaf_info.st_mode) + or not stat.S_ISDIR(leaf_info.st_mode) + or leaf_info.st_uid != os.getuid() + or stat.S_IMODE(leaf_info.st_mode) != runtime_bundle.INSTALL_MODE ): - raise ValueError("activation runtime bundle root identity is invalid") + raise ValueError("runtime bundle source root identity is invalid") + expected_names = {record["path"] for record in records} + observed_names: set[str] = set() + try: + with os.scandir(bundle_leaf) as entries: + for entry in entries: + observed_names.add(entry.name) + if len(observed_names) > len(expected_names): + raise ValueError( + "runtime bundle source membership is not exact" + ) + except OSError as exc: + raise ValueError("runtime bundle source could not be enumerated") from exc + if observed_names != expected_names: + raise ValueError("runtime bundle source membership is not exact") + member_flags = ( + os.O_RDONLY + | getattr(os, "O_NOFOLLOW", 0) + | getattr(os, "O_CLOEXEC", 0) + | getattr(os, "O_NONBLOCK", 0) # a swapped FIFO must not block the open + ) for record in records: - member = bundle / record["path"] - info = _safe_source(member) - if ( - not stat.S_ISREG(info.st_mode) - or info.st_uid != os.getuid() - or info.st_nlink != 1 - or stat.S_IMODE(info.st_mode) != record["install_mode"] - or info.st_size != record["size"] - or _sha256(member) != record["sha256"] - ): - raise ValueError("activation runtime bundle member identity is invalid") + member = bundle_leaf / record["path"] + try: + lexical = member.lstat() + descriptor = os.open(member, member_flags) + except OSError as exc: + raise ValueError("runtime bundle source member is unsafe") from exc + try: + info = os.fstat(descriptor) + if ( + stat.S_ISLNK(lexical.st_mode) + or not stat.S_ISREG(info.st_mode) + or info.st_uid != os.getuid() + or info.st_nlink != 1 + or stat.S_IMODE(info.st_mode) != record["install_mode"] + or info.st_size != record["size"] + ): + raise ValueError( + "runtime bundle source member identity is invalid" + ) + digest = hashlib.sha256() + read_total = 0 + while read_total <= record["size"]: + # Cap the read at the fstat'd size: a file that grows after the + # fstat cannot cause unbounded reading. + chunk = os.read(descriptor, 1024 * 1024) + if not chunk: + break + read_total += len(chunk) + digest.update(chunk) + if read_total != record["size"] or digest.hexdigest() != record["sha256"]: + raise ValueError("runtime bundle source member digest is invalid") + finally: + try: + os.close(descriptor) + except OSError: + pass -def classify_package(plugin_path: Path) -> str: - plugin_path = plugin_path.resolve(strict=True) +def _classify_from_manifest( + plugin_path: Path, manifest_bytes: bytes +) -> tuple[str, tuple[dict[str, object], ...]]: + """Classify + fully validate the package from ONE manifest byte snapshot. + + The mode decision, the full activation-manifest shape validation, and the + runtime records ALL derive from the same ``manifest_bytes`` — so a caller + that freezes the manifest once and passes those exact bytes here cannot be + raced (an A→B→A manifest swap between classification and packing is + impossible; the frozen bytes are the single source). + """ + policy = plugin_path / "signing_policy.py" policy_info = _safe_source(policy) if not stat.S_ISREG(policy_info.st_mode): raise ValueError("signing_policy.py must be a regular file") - manifest = _load_manifest(plugin_path) + manifest = _parse_manifest(manifest_bytes) artifacts = manifest["artifacts"] runtime_root = plugin_path / "runtime" if not artifacts: if runtime_root.exists() or runtime_root.is_symlink(): raise ValueError("policy-only package contains an unadvertised runtime") - return "policy-only" + return "policy-only", () if len(artifacts) != 1: raise ValueError("activation package requires exactly one runtime artifact") - _validate_activation(plugin_path, artifacts[0]) - return "activation" + # D1: the committed manifest is the activation marker. The physical bundle + # is NOT read here — it ships as a release asset (never committed) and is + # validated against its actual source by `_validate_activation_bundle_tree` + # (build) and against the frozen manifest records (verify_archive). + records = _validate_activation_manifest(artifacts[0]) + return "activation", records + + +def classify_package(plugin_path: Path) -> str: + plugin_path = plugin_path.resolve(strict=True) + mode, _records = _classify_from_manifest( + plugin_path, _read_manifest_bytes(plugin_path) + ) + return mode def _require_exact_manifest_trees(plugin_path: Path) -> None: @@ -402,9 +527,37 @@ def _require_exact_third_party_notice_tree(plugin_path: Path) -> None: raise ValueError("activation third-party notice content digest is invalid") -def _member_paths(plugin_path: Path, *, mode: str) -> list[Path]: +# Runtime DIRECTORY members are packaging scaffolding, synthesized with fixed +# canonical metadata — never statted from a mutable source tree. The two +# traversal parents are plain 0o755 directories; the bundle leaf itself keeps +# the sealed install mode. +_RUNTIME_DIR_MODES: dict[str, int] = { + "runtime": 0o755, + "runtime/darwin-arm64": 0o755, + RUNTIME_BUNDLE_REL.as_posix(): runtime_bundle.INSTALL_MODE, +} + + +def _member_plan( + plugin_path: Path, + *, + mode: str, + records: tuple[dict[str, object], ...] = (), +) -> list[tuple[str, Path | None]]: + """Return the canonical (archive name, source path) plan, sorted by name. + + Runtime members carry a ``None`` source: their archive metadata derives + entirely from the frozen manifest ``records`` (and `_RUNTIME_DIR_MODES`), + never from a stat of a mutable tree. Every other member is sourced from + the git-tracked plugin tree. + """ + if mode not in {"policy-only", "activation"}: raise ValueError("unknown archive mode") + if mode == "activation" and not records: + raise ValueError("activation member plan requires frozen manifest records") + if mode == "policy-only" and records: + raise ValueError("policy-only member plan forbids runtime records") _require_exact_manifest_trees(plugin_path) differences = skill_tree_differences(plugin_path / "skills", SPECS_DIR) if differences: @@ -415,110 +568,485 @@ def _member_paths(plugin_path: Path, *, mode: str) -> list[Path]: Path("skills"), *(Path("skills") / relative for relative in expected_skill_relpaths(SPECS_DIR)), ] + members: dict[str, Path | None] = {} if mode == "activation": _require_exact_third_party_notice_tree(plugin_path) - manifest = _load_manifest(plugin_path) - records = runtime_bundle.validate_file_records( - manifest["artifacts"][0]["files"] - ) - relatives.extend( - ( - Path("runtime"), - Path("runtime/darwin-arm64"), - RUNTIME_BUNDLE_REL, - *(RUNTIME_BUNDLE_REL / record["path"] for record in records), - *ACTIVATION_THIRD_PARTY_MEMBERS, - ) - ) - members: dict[str, Path] = {} + relatives.extend(ACTIVATION_THIRD_PARTY_MEMBERS) + for name in _RUNTIME_DIR_MODES: + members[name] = None + for record in records: + members[(RUNTIME_BUNDLE_REL / record["path"]).as_posix()] = None for relative in relatives: source = plugin_path / relative _safe_source(source) members[relative.as_posix()] = source - return [members[name] for name in sorted(members)] + return [(name, members[name]) for name in sorted(members)] -def _canonical_names(paths: Iterable[Path], plugin_path: Path) -> list[str]: - return [path.relative_to(plugin_path).as_posix() for path in paths] +# Hard resource bounds for reading a possibly-hostile archive. The verifier +# NEVER parses hostile tar/PAX bytes: it regenerates the ONE canonical tar from +# trusted inputs and byte-compares, so a candidate archive is only ever sliced +# by offsets WE computed. Only the bounded gzip inflater touches candidate +# bytes, under hard compressed- and decompressed-size caps. +# +# The runtime payload alone may be as large as MAX_ARTIFACT_BYTES (64 MiB); the +# archive additionally holds the manifest, skills, licenses, plugin metadata, +# tar headers/padding, and gzip framing, and a poorly-compressible (already +# compressed/encrypted) runtime near the limit barely shrinks. So BOTH the +# compressed and decompressed caps carry headroom above the payload cap +# (2x = 128 MiB) — otherwise the builder could reject its OWN legitimate output +# — while the decompressed cap still bounds a gzip bomb well below memory +# exhaustion. +_MAX_COMPRESSED_ARCHIVE_BYTES = 2 * MAX_ARTIFACT_BYTES +_MAX_DECOMPRESSED_ARCHIVE_BYTES = 2 * MAX_ARTIFACT_BYTES +_USTAR_BLOCK = 512 +# The complete fixed canonical 10-byte gzip header the builder emits: magic +# (1f 8b), DEFLATE method (08), flags 0 (rejects FTEXT/FHCRC/FEXTRA/FNAME/ +# FCOMMENT — the malleability/amplification vectors), mtime 0, XFL 02 +# (best-compression, GzipFile default level 9), OS ff (unknown). CPython's gzip +# writes XFL and OS deterministically (hardcoded ff for OS; 02 for level 9), so +# binding all ten is byte-reproducible across the operator's build Python and +# CI's 3.10/3.12/3.14 — and the build→verify round-trip tests fail closed on +# any version that diverges rather than silently accepting a variant. +_CANONICAL_GZIP_HEADER = b"\x1f\x8b\x08\x00\x00\x00\x00\x00\x02\xff" -def verify_archive(plugin_path: Path, archive_path: Path, *, mode: str) -> None: - plugin_path = plugin_path.resolve(strict=True) - if classify_package(plugin_path) != mode: - raise ValueError("archive mode no longer matches the source package") - expected_paths = _member_paths(plugin_path, mode=mode) - expected_names = _canonical_names(expected_paths, plugin_path) +def _synthesized_tarinfo( + name: str, *, mode: int, size: int = 0, directory: bool = False +) -> tarfile.TarInfo: + info = tarfile.TarInfo(name) + info.type = tarfile.DIRTYPE if directory else tarfile.REGTYPE + info.mode = mode + info.size = size + info.uid = 0 + info.gid = 0 + info.uname = "" + info.gname = "" + info.mtime = 0 + return info + + +def _finalize_tarinfo(info: tarfile.TarInfo) -> tarfile.TarInfo: + info.uid = 0 + info.gid = 0 + info.uname = "" + info.gname = "" + info.mtime = 0 + if info.pax_headers: + # USTAR_FORMAT would already have raised for an unrepresentable field; + # this belt-and-suspenders makes it impossible to emit an extension + # record so build and verify are byte-reproducible across environments. + raise ValueError(f"archive member is not USTAR-representable: {info.name}") + return info + + +def _emit_canonical_tar( + plan: list[tuple[str, Path | None]], + *, + plugin_path: Path, + frozen_manifest: bytes | None, + record_by_name: dict[str, dict[str, object]], + runtime_payloads: dict[str, bytes], +) -> tuple[bytes, dict[str, tuple[int, int]]]: + """Serialize the ONE canonical uncompressed tar deterministically. + + USTAR_FORMAT (tarfile raises on anything needing a PAX/GNU extension + record), zeroed ownership/mtime, exact ``_member_plan`` order. Non-runtime + payloads come from the trusted ``plugin_path`` tree (the manifest member + from ``frozen_manifest``); runtime FILE payloads come from + ``runtime_payloads`` (real bytes at build, zero-fill at verify). Both build + and verify call this, so they produce byte-identical structure. Returns + ``(tar_bytes, runtime_ranges)`` mapping each runtime file member name to its + ``(data_offset, length)`` within the tar. + """ + + buffer = io.BytesIO() + ranges: dict[str, tuple[int, int]] = {} + with tarfile.open( + fileobj=buffer, mode="w", format=tarfile.USTAR_FORMAT + ) as tar: + for name, source in plan: + if source is None and name in _RUNTIME_DIR_MODES: + tar.addfile( + _finalize_tarinfo( + _synthesized_tarinfo( + name, mode=_RUNTIME_DIR_MODES[name], directory=True + ) + ) + ) + continue + if source is None: + record = record_by_name[name] + payload = runtime_payloads[name] + if len(payload) != record["size"]: + raise ValueError("runtime payload size mismatch during emit") + info = _finalize_tarinfo( + _synthesized_tarinfo( + name, mode=record["install_mode"], size=record["size"] + ) + ) + header_offset = buffer.tell() + tar.addfile(info, io.BytesIO(payload)) + ranges[name] = (header_offset + _USTAR_BLOCK, record["size"]) + continue + info = tar.gettarinfo(str(source), arcname=name) + # git tracks ONLY the exec bit, but a working-tree checkout applies + # the local umask to the rest — so normalize non-runtime modes to + # 0o755/0o644 by the exec bit alone. Without this the operator's + # build host and CI (different umask, different checkouts) could + # emit different mode fields and the verify byte-compare would + # false-reject a legitimate archive. + info.mode = 0o755 if (info.isdir() or (info.mode & 0o111)) else 0o644 + info = _finalize_tarinfo(info) + if name == "runtime-manifest.json" and frozen_manifest is not None: + info.size = len(frozen_manifest) + tar.addfile(info, io.BytesIO(frozen_manifest)) + elif info.isfile(): + with source.open("rb") as stream: + tar.addfile(info, stream) + else: + tar.addfile(info) + return buffer.getvalue(), ranges + + +def _read_bounded_compressed(archive_path: Path) -> bytes: + """Read the compressed archive fd-only under a hard size cap. + + ``O_NOFOLLOW`` refuses a symlinked final component; ``O_NONBLOCK`` keeps a + substituted FIFO from blocking the open before the ``fstat`` regular-file + check; the cap is enforced on the bytes ACTUALLY read from the descriptor, + so a path swap or post-fstat growth cannot smuggle input past it. There is + no ``lstat``→``open`` window. + """ + + flags = ( + os.O_RDONLY + | getattr(os, "O_NOFOLLOW", 0) + | getattr(os, "O_CLOEXEC", 0) + | getattr(os, "O_NONBLOCK", 0) + ) + try: + descriptor = os.open(archive_path, flags) + except OSError as exc: + raise ValueError("plugin archive is unreadable") from exc + chunks: list[bytes] = [] + total = 0 + try: + described = os.fstat(descriptor) + if not stat.S_ISREG(described.st_mode): + raise ValueError("plugin archive is unreadable") + if described.st_size > _MAX_COMPRESSED_ARCHIVE_BYTES: + raise ValueError("plugin archive exceeds the canonical size bound") + while True: + chunk = os.read(descriptor, 1024 * 1024) + if not chunk: + break + total += len(chunk) + if total > _MAX_COMPRESSED_ARCHIVE_BYTES: + raise ValueError("plugin archive exceeds the canonical size bound") + chunks.append(chunk) + except OSError as exc: + raise ValueError("plugin archive is unreadable") from exc + finally: + try: + os.close(descriptor) + except OSError: + pass + return b"".join(chunks) + + +def _inflate_bounded(compressed: bytes) -> bytes: + """Inflate a single gzip stream under a hard output cap. + + Binds the fixed canonical gzip header, caps decompressed OUTPUT per call + (so a bomb trips before materializing the expansion), validates CRC/ISIZE + (zlib raises on mismatch at end-of-stream), and requires the stream to be + complete with NO trailing/concatenated data. + """ + + if compressed[:10] != _CANONICAL_GZIP_HEADER: + raise ValueError("plugin archive gzip header is not canonical") + decompressor = zlib.decompressobj(wbits=16 + zlib.MAX_WBITS) + chunk_size = 1024 * 1024 + chunks: list[bytes] = [] + total = 0 + pending = compressed try: - bundle = tarfile.open(archive_path, "r:gz") - except (OSError, tarfile.TarError) as exc: + while not decompressor.eof: + before = len(pending) + piece = decompressor.decompress(pending, chunk_size) + pending = decompressor.unconsumed_tail + if piece: + total += len(piece) + if total > _MAX_DECOMPRESSED_ARCHIVE_BYTES: + raise ValueError( + "plugin archive decompression exceeds the canonical bound" + ) + chunks.append(piece) + elif len(pending) >= before: + # No output produced and no input consumed: the stream cannot + # make progress (truncated/malformed). Stop; the eof check below + # rejects it. + break + except zlib.error as exc: raise ValueError("plugin archive is unreadable") from exc - with bundle: - members = bundle.getmembers() - names = [member.name for member in members] - if len(names) != len(set(names)): - raise ValueError("archive contains duplicate members") - for name in names: - pure = PurePosixPath(name) - if pure.is_absolute() or ".." in pure.parts or "\\" in name: - raise ValueError("archive contains an unsafe member path") - if names != expected_names: - raise ValueError("archive member list is not canonical") - for source, member in zip(expected_paths, members, strict=True): - source_info = _safe_source(source) - source_is_file = stat.S_ISREG(source_info.st_mode) - if source_is_file != member.isfile() or ( - not source_is_file and not member.isdir() + if not decompressor.eof: + raise ValueError("plugin archive gzip stream is incomplete") + if decompressor.unused_data: + raise ValueError("plugin archive has trailing data after the gzip stream") + return b"".join(chunks) + + +def _assert_runtime_range_map( + ranges: dict[str, tuple[int, int]], + record_by_name: dict[str, dict[str, object]], + total_length: int, +) -> None: + """Assert the regenerated runtime range map is well-formed (design item 6). + + Complete + one-to-one with the manifest runtime entries, each range + exact-length, in-bounds, and pairwise disjoint. The archived + ``runtime-manifest.json`` member is non-runtime by construction and thus + lies OUTSIDE every range (bound by the structural byte-compare). + """ + + if set(ranges) != set(record_by_name): + raise ValueError("archive runtime range map is not one-to-one with the manifest") + for name, (offset, length) in ranges.items(): + if length != record_by_name[name]["size"]: + raise ValueError("archive runtime range length does not match the manifest") + previous_end = 0 + for offset, length in sorted(ranges.values()): + if offset < previous_end or offset < 0 or offset + length > total_length: + raise ValueError("archive runtime range map is out of bounds or overlapping") + previous_end = offset + length + + +def _bytes_equal_outside_ranges( + candidate: bytes, canonical: bytes, ranges: list[tuple[int, int]] +) -> bool: + """True iff candidate == canonical at every byte NOT inside a runtime range.""" + + position = 0 + for offset, length in sorted(ranges): + if candidate[position:offset] != canonical[position:offset]: + return False + position = offset + length + return candidate[position:] == canonical[position:] + + +def verify_archive( + plugin_path: Path, + archive_path: Path, + *, + mode: str, + frozen_manifest: bytes | None = None, +) -> None: + """Verify a built archive by REGENERATING the canonical tar and comparing. + + The candidate archive's tar bytes are never handed to ``tarfile``. Instead + the ONE canonical tar is regenerated from trusted inputs (plugin tree + + frozen manifest) with runtime payloads zero-filled; the candidate is only + bounded-inflated and then sliced by offsets WE computed. Every structural + property (headers, typeflags, ordering, padding, EOF, dialect, no hidden + records) is bound by the byte-compare OUTSIDE the runtime ranges; each + runtime payload is bound to the frozen-manifest digest. ``frozen_manifest`` + lets ``build_archive`` share the exact snapshot it packed; CI omits it and + reads the snapshot once from the signed-tag checkout. + """ + + plugin_path = plugin_path.resolve(strict=True) + # Freeze ONE manifest snapshot: mode classification, full shape validation, + # runtime records, and the embedded manifest member all derive from it (a + # caller-supplied snapshot from build_archive, or a single read here in CI). + if frozen_manifest is None: + frozen_manifest = _read_manifest_bytes(plugin_path) + resolved_mode, records = _classify_from_manifest(plugin_path, frozen_manifest) + if resolved_mode != mode: + raise ValueError("archive mode no longer matches the source package") + plan = _member_plan(plugin_path, mode=mode, records=records) + record_by_name = { + (RUNTIME_BUNDLE_REL / record["path"]).as_posix(): record for record in records + } + zero_payloads = { + name: b"\x00" * record["size"] for name, record in record_by_name.items() + } + canonical, ranges = _emit_canonical_tar( + plan, + plugin_path=plugin_path, + frozen_manifest=frozen_manifest, + record_by_name=record_by_name, + runtime_payloads=zero_payloads, + ) + candidate = _inflate_bounded(_read_bounded_compressed(archive_path)) + if len(candidate) != len(canonical): + raise ValueError("archive does not match the canonical layout") + _assert_runtime_range_map(ranges, record_by_name, len(canonical)) + for name, (offset, length) in ranges.items(): + record = record_by_name[name] + if ( + length != record["size"] + or hashlib.sha256(candidate[offset : offset + length]).hexdigest() + != record["sha256"] + ): + raise ValueError(f"archive runtime member digest failed: {name}") + if not _bytes_equal_outside_ranges(candidate, canonical, list(ranges.values())): + raise ValueError("archive structure does not match the canonical layout") + + +def _read_runtime_payloads( + bundle_leaf: Path, records: tuple[dict[str, object], ...] +) -> dict[str, bytes]: + """Read + re-validate + re-digest each runtime payload into memory. + + Reading through a fresh ``O_NOFOLLOW``/``O_NONBLOCK`` descriptor and + re-checking identity here binds the EXACT bytes that will be emitted into + the archive (closing any window between the earlier tree validation and the + pack) — and the bytes are additionally re-bound to the manifest digest by + ``verify_archive`` on the temp before publish. + """ + + flags = ( + os.O_RDONLY + | getattr(os, "O_NOFOLLOW", 0) + | getattr(os, "O_CLOEXEC", 0) + | getattr(os, "O_NONBLOCK", 0) + ) + payloads: dict[str, bytes] = {} + for record in records: + name = (RUNTIME_BUNDLE_REL / record["path"]).as_posix() + try: + descriptor = os.open(bundle_leaf / record["path"], flags) + except OSError as exc: + raise ValueError("runtime bundle source member is unsafe") from exc + try: + info = os.fstat(descriptor) + if ( + not stat.S_ISREG(info.st_mode) + or info.st_uid != os.getuid() + or info.st_nlink != 1 + or stat.S_IMODE(info.st_mode) != record["install_mode"] + or info.st_size != record["size"] ): - raise ValueError(f"archive member type parity failed: {member.name}") - if stat.S_IMODE(member.mode) != stat.S_IMODE(source_info.st_mode): - raise ValueError(f"archive member mode parity failed: {member.name}") - if source_is_file: - archived = bundle.extractfile(member) - if archived is None or archived.read() != source.read_bytes(): - raise ValueError(f"archive member byte parity failed: {member.name}") - runtime_files = [ - member.name - for member in members - if member.isfile() and member.name.startswith("runtime/") - ] - expected_runtime = ( - [ - (RUNTIME_BUNDLE_REL / record["path"]).as_posix() - for record in runtime_bundle.validate_file_records( - _load_manifest(plugin_path)["artifacts"][0]["files"] - ) - ] - if mode == "activation" - else [] - ) - if runtime_files != expected_runtime: - raise ValueError("archive runtime membership does not match release mode") + raise ValueError("runtime bundle source member changed during packing") + parts: list[bytes] = [] + read_total = 0 + while read_total <= record["size"]: + chunk = os.read(descriptor, 1024 * 1024) + if not chunk: + break + parts.append(chunk) + read_total += len(chunk) + data = b"".join(parts) + if ( + len(data) != record["size"] + or hashlib.sha256(data).hexdigest() != record["sha256"] + ): + raise ValueError("runtime bundle source member changed during packing") + payloads[name] = data + finally: + try: + os.close(descriptor) + except OSError: + pass + return payloads -def build_archive(root: Path, *, plugin: str, output: Path) -> str: +def build_archive( + root: Path, + *, + plugin: str, + output: Path, + bundle_source: Path | None = None, +) -> str: if plugin != PLUGIN_NAME: raise ValueError("agent-collab is the only releaseable plugin") plugin_path = (root / "plugins" / plugin).resolve(strict=True) - mode = classify_package(plugin_path) - members = _member_paths(plugin_path, mode=mode) + # Freeze the manifest ONCE: mode classification, full shape validation, + # runtime records, the bundle-tree validation, the embedded manifest member, + # and verify_archive ALL derive from this single snapshot, so a mid-build + # A→B→A manifest swap on disk cannot change what gets packaged. + frozen_manifest: bytes | None = _read_manifest_bytes(plugin_path) + mode, records = _classify_from_manifest(plugin_path, frozen_manifest) + bundle_leaf: Path | None = None + runtime_payloads: dict[str, bytes] = {} + if mode == "policy-only": + # The committed manifest — never the flag — decides the mode. + if bundle_source is not None: + raise ValueError("policy-only manifest forbids --bundle-source") + else: + # Fail closed: an activation manifest without a matching signed bundle + # source is a hard error, never a silent policy-only downgrade. + if bundle_source is None: + raise ValueError("activation manifest requires --bundle-source") + runtime_root = plugin_path / "runtime" + if runtime_root.exists() or runtime_root.is_symlink(): + raise ValueError( + "in-tree runtime conflicts with --bundle-source; remove one source" + ) + try: + raw_source = bundle_source.lstat() + except OSError as exc: + raise ValueError("runtime bundle source is missing") from exc + # lstat the raw CLI argument BEFORE resolving — resolve() would erase + # the evidence that the argument itself was a symlink. + if stat.S_ISLNK(raw_source.st_mode) or not stat.S_ISDIR(raw_source.st_mode): + raise ValueError("runtime bundle source is unsafe") + bundle_leaf = bundle_source.resolve(strict=True) + _validate_activation_bundle_tree(bundle_leaf, records) + runtime_payloads = _read_runtime_payloads(bundle_leaf, records) + plan = _member_plan(plugin_path, mode=mode, records=records) + record_by_name = { + (RUNTIME_BUNDLE_REL / record["path"]).as_posix(): record for record in records + } + + def _reject_source_alias(candidate: Path) -> None: + for forbidden in (plugin_path, bundle_leaf): + if forbidden is not None and candidate.is_relative_to(forbidden): + raise ValueError("archive output must not alias a source tree") + + # Reject a source-tree destination BEFORE creating any directory (mkdir must + # never mutate a rejected destination), then re-check the strictly resolved + # path once the parent exists. + _reject_source_alias(output.resolve()) output.parent.mkdir(parents=True, exist_ok=True) - with output.open("wb") as raw: - with gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=0) as compressed: - with tarfile.open(fileobj=compressed, mode="w", format=tarfile.PAX_FORMAT) as bundle: - for source in members: - name = source.relative_to(plugin_path).as_posix() - info = bundle.gettarinfo(str(source), arcname=name) - info.uid = 0 - info.gid = 0 - info.uname = "" - info.gname = "" - info.mtime = 0 - if info.isfile(): - with source.open("rb") as stream: - bundle.addfile(info, stream) - else: - bundle.addfile(info) - verify_archive(plugin_path, output, mode=mode) + output_resolved = output.parent.resolve(strict=True) / output.name + _reject_source_alias(output_resolved) + # Regenerate the ONE canonical tar (shared with verify_archive), then gzip + # it into an exclusive temp, verify THAT, and only then os.replace() to the + # destination — a failed build or verification never leaves a publishable + # artifact. The pid-suffixed temp name is predictable: in a shared/ + # world-writable output dir an adversary could pre-create it as a DoS + # (O_EXCL fails, their file survives untouched — integrity unaffected). + # Release builds write to operator-owned directories. + canonical_tar, _ranges = _emit_canonical_tar( + plan, + plugin_path=plugin_path, + frozen_manifest=frozen_manifest, + record_by_name=record_by_name, + runtime_payloads=runtime_payloads, + ) + temp_path = output_resolved.parent / f".{output_resolved.name}.tmp.{os.getpid()}" + temp_created = False + try: + # Owner-only (0o600): the archive is read + uploaded by the building + # user; no other local user needs read access to the release artifact. + descriptor = os.open(temp_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + temp_created = True + with os.fdopen(descriptor, "wb") as raw: + with gzip.GzipFile(fileobj=raw, mode="wb", mtime=0) as compressed: + compressed.write(canonical_tar) + verify_archive( + plugin_path, temp_path, mode=mode, frozen_manifest=frozen_manifest + ) + os.replace(temp_path, output_resolved) + except BaseException: + if temp_created: + try: + os.unlink(temp_path) + except OSError: + pass + raise return mode @@ -528,6 +1056,16 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--plugin", default=PLUGIN_NAME) parser.add_argument("--output", type=Path) parser.add_argument("--print-mode", action="store_true") + parser.add_argument( + "--bundle-source", + type=Path, + default=None, + help=( + "path to the signed runtime bundle leaf directory " + "(…/agent-collab-runtime.bundle) from the local release handoff; " + "required for an activation manifest, forbidden for policy-only" + ), + ) args = parser.parse_args(argv) try: if args.plugin != PLUGIN_NAME: @@ -540,7 +1078,12 @@ def main(argv: list[str] | None = None) -> int: return 0 if args.output is None: raise ValueError("--output is required when building an archive") - mode = build_archive(args.repo_root, plugin=args.plugin, output=args.output) + mode = build_archive( + args.repo_root, + plugin=args.plugin, + output=args.output, + bundle_source=args.bundle_source, + ) except (OSError, ValueError) as exc: print(f"FAIL: {exc}") return 1 diff --git a/tests/test_plugin_archive.py b/tests/test_plugin_archive.py index d3674b0..09bd33d 100644 --- a/tests/test_plugin_archive.py +++ b/tests/test_plugin_archive.py @@ -8,10 +8,13 @@ import shutil import stat import sys +import io +import os import tarfile import tempfile import unittest from pathlib import Path +from unittest import mock ROOT = Path(__file__).resolve().parents[1] @@ -69,14 +72,19 @@ def _install_third_party_notices(self) -> None: self.assertTrue((licenses / name).is_file()) def _activate(self) -> Path: + """Stage an activation manifest in-tree + a signed bundle OUT-of-tree. + + The runtime bundle ships as a release asset (never committed), so the + fixture mirrors production: the committed tree carries only the + manifest, and the 0o500 bundle leaf lives in an external handoff + directory supplied to the builder via ``bundle_source``. + """ + self._install_third_party_notices() - runtime = ( - self.plugin - / "runtime" - / "darwin-arm64" - / "agent-collab-runtime.bundle" - / "agent-collab-runtime" + self.bundle_leaf = ( + self.root / "handoff" / "agent-collab-runtime.bundle" ) + runtime = self.bundle_leaf / "agent-collab-runtime" runtime.parent.mkdir(parents=True) runtime.write_bytes(b"signed-runtime-fixture") library = runtime.parent / "libpython3.13.dylib" @@ -206,10 +214,14 @@ def test_packaged_legal_files_match_repository_canonicals(self) -> None: def test_activation_archive_has_exactly_one_runtime_with_byte_mode_parity(self) -> None: runtime = self._activate() mode = self.builder.build_archive( - self.root, plugin="agent-collab", output=self.archive + self.root, + plugin="agent-collab", + output=self.archive, + bundle_source=self.bundle_leaf, ) self.assertEqual(mode, "activation") + bundle_prefix = "runtime/darwin-arm64/agent-collab-runtime.bundle/" with tarfile.open(self.archive, "r:gz") as bundle: runtime_files = [ member @@ -219,18 +231,37 @@ def test_activation_archive_has_exactly_one_runtime_with_byte_mode_parity(self) self.assertEqual( [member.name for member in runtime_files], [ - "runtime/darwin-arm64/agent-collab-runtime.bundle/agent-collab-runtime", - "runtime/darwin-arm64/agent-collab-runtime.bundle/libpython3.13.dylib", + bundle_prefix + "agent-collab-runtime", + bundle_prefix + "libpython3.13.dylib", ], ) member = runtime_files[0] self.assertEqual(bundle.extractfile(member).read(), runtime.read_bytes()) self.assertEqual(stat.S_IMODE(member.mode), 0o500) + # Synthesized runtime directory scaffolding carries fixed canonical + # modes: 0o755 traversal parents + the sealed 0o500 bundle leaf. + directory_modes = { + archived.name: stat.S_IMODE(archived.mode) + for archived in bundle.getmembers() + if archived.isdir() and archived.name.startswith("runtime") + } + self.assertEqual( + directory_modes, + { + "runtime": 0o755, + "runtime/darwin-arm64": 0o755, + "runtime/darwin-arm64/agent-collab-runtime.bundle": 0o500, + }, + ) + for archived in bundle.getmembers(): if not archived.isfile(): continue - source = self.plugin / archived.name + if archived.name.startswith(bundle_prefix): + source = self.bundle_leaf / archived.name[len(bundle_prefix):] + else: + source = self.plugin / archived.name self.assertEqual(bundle.extractfile(archived).read(), source.read_bytes()) self.assertEqual( stat.S_IMODE(archived.mode), @@ -240,7 +271,10 @@ def test_activation_archive_has_exactly_one_runtime_with_byte_mode_parity(self) def test_activation_archive_includes_canonical_third_party_notice_tree(self) -> None: self._activate() mode = self.builder.build_archive( - self.root, plugin="agent-collab", output=self.archive + self.root, + plugin="agent-collab", + output=self.archive, + bundle_source=self.bundle_leaf, ) self.assertEqual(mode, "activation") @@ -278,7 +312,10 @@ def test_activation_rejects_missing_third_party_notice_member(self) -> None: with self.assertRaisesRegex(ValueError, "third-party notice tree"): self.builder.build_archive( - self.root, plugin="agent-collab", output=self.archive + self.root, + plugin="agent-collab", + output=self.archive, + bundle_source=self.bundle_leaf, ) def test_activation_rejects_missing_top_level_third_party_notice(self) -> None: @@ -287,7 +324,10 @@ def test_activation_rejects_missing_top_level_third_party_notice(self) -> None: with self.assertRaisesRegex(ValueError, "third-party notice tree"): self.builder.build_archive( - self.root, plugin="agent-collab", output=self.archive + self.root, + plugin="agent-collab", + output=self.archive, + bundle_source=self.bundle_leaf, ) def test_activation_rejects_unexpected_third_party_notice_member(self) -> None: @@ -298,7 +338,10 @@ def test_activation_rejects_unexpected_third_party_notice_member(self) -> None: with self.assertRaisesRegex(ValueError, "third-party notice tree"): self.builder.build_archive( - self.root, plugin="agent-collab", output=self.archive + self.root, + plugin="agent-collab", + output=self.archive, + bundle_source=self.bundle_leaf, ) def test_activation_rejects_third_party_notice_content_drift(self) -> None: @@ -308,7 +351,10 @@ def test_activation_rejects_third_party_notice_content_drift(self) -> None: with self.assertRaisesRegex(ValueError, "content digest"): self.builder.build_archive( - self.root, plugin="agent-collab", output=self.archive + self.root, + plugin="agent-collab", + output=self.archive, + bundle_source=self.bundle_leaf, ) def test_activation_rejects_symlinked_third_party_notice_member(self) -> None: @@ -319,7 +365,10 @@ def test_activation_rejects_symlinked_third_party_notice_member(self) -> None: with self.assertRaisesRegex(ValueError, "unsafe|third-party notice tree"): self.builder.build_archive( - self.root, plugin="agent-collab", output=self.archive + self.root, + plugin="agent-collab", + output=self.archive, + bundle_source=self.bundle_leaf, ) def test_policy_only_rejects_unadvertised_runtime_and_missing_policy(self) -> None: @@ -348,10 +397,35 @@ def test_archive_verifier_detects_source_byte_drift(self) -> None: mode = self.builder.build_archive( self.root, plugin="agent-collab", output=self.archive ) - (self.plugin / "signing_policy.py").write_text("changed\n", encoding="utf-8") - with self.assertRaisesRegex(ValueError, "byte parity"): + policy = self.plugin / "signing_policy.py" + original = policy.read_bytes() + + # Same-size drift: the regenerated canonical tar differs from the + # candidate outside any runtime range → structural byte-compare fails. + policy.write_bytes(b"#" * len(original)) + with self.assertRaisesRegex(ValueError, "structure does not match"): + self.builder.verify_archive(self.plugin, self.archive, mode=mode) + + # Size drift changes the total canonical length → rejected up front. + policy.write_bytes(original + b"\n# tail\n") + with self.assertRaisesRegex(ValueError, "does not match the canonical layout"): self.builder.verify_archive(self.plugin, self.archive, mode=mode) + def test_compressed_archive_cap_has_headroom_over_runtime_payload(self) -> None: + # The runtime payload alone may be up to MAX_ARTIFACT_BYTES; the archive + # also holds the manifest/skills/licenses/headers/gzip framing, and a + # poorly-compressible runtime near the cap barely shrinks. The compressed + # cap must exceed the payload cap or the builder would reject its own + # legitimate output; the decompressed cap must still bound a bomb. + self.assertGreater( + self.builder._MAX_COMPRESSED_ARCHIVE_BYTES, + self.builder.MAX_ARTIFACT_BYTES, + ) + self.assertGreaterEqual( + self.builder._MAX_DECOMPRESSED_ARCHIVE_BYTES, + self.builder._MAX_COMPRESSED_ARCHIVE_BYTES, + ) + def test_archive_size_limit_and_required_policy_member_are_canonical(self) -> None: self.assertEqual(self.builder.MAX_ARTIFACT_BYTES, 64 * 1024 * 1024) self.assertEqual(self.builder.RUNTIME_FILE_MODE, 0o500) @@ -368,6 +442,475 @@ def test_archive_fails_closed_when_runtime_setup_entrypoint_is_missing(self) -> self.root, plugin="agent-collab", output=self.archive ) + def _build_activation(self) -> str: + return self.builder.build_archive( + self.root, + plugin="agent-collab", + output=self.archive, + bundle_source=self.bundle_leaf, + ) + + def test_activation_manifest_requires_bundle_source_fail_closed(self) -> None: + self._activate() + with self.assertRaisesRegex(ValueError, "requires --bundle-source"): + self.builder.build_archive( + self.root, plugin="agent-collab", output=self.archive + ) + self.assertFalse(self.archive.exists()) + + def test_policy_only_manifest_forbids_bundle_source(self) -> None: + self._install_third_party_notices() + stray = self.root / "stray-bundle" + stray.mkdir() + with self.assertRaisesRegex(ValueError, "forbids --bundle-source"): + self.builder.build_archive( + self.root, + plugin="agent-collab", + output=self.archive, + bundle_source=stray, + ) + + def test_in_tree_runtime_conflicts_with_bundle_source(self) -> None: + self._activate() + (self.plugin / "runtime").mkdir() + with self.assertRaisesRegex(ValueError, "in-tree runtime conflicts"): + self._build_activation() + + def test_bundle_source_rejects_symlink_argument_before_resolve(self) -> None: + self._activate() + alias = self.root / "leaf-alias" + alias.symlink_to(self.bundle_leaf) + with self.assertRaisesRegex(ValueError, "runtime bundle source is unsafe"): + self.builder.build_archive( + self.root, + plugin="agent-collab", + output=self.archive, + bundle_source=alias, + ) + + def test_bundle_source_rejects_content_and_size_drift(self) -> None: + self._activate() + library = self.bundle_leaf / "libpython3.13.dylib" + original = library.read_bytes() + + library.chmod(0o700) + library.write_bytes(b"X" * len(original)) # same size, wrong bytes + library.chmod(0o500) + with self.assertRaisesRegex(ValueError, "digest is invalid"): + self._build_activation() + + library.chmod(0o700) + library.write_bytes(original + b"tail") # size drift + library.chmod(0o500) + with self.assertRaisesRegex(ValueError, "member identity is invalid"): + self._build_activation() + self.assertFalse(self.archive.exists()) + + def test_bundle_source_rejects_mode_drift(self) -> None: + self._activate() + (self.bundle_leaf / "libpython3.13.dylib").chmod(0o755) + with self.assertRaisesRegex(ValueError, "member identity is invalid"): + self._build_activation() + + (self.bundle_leaf / "libpython3.13.dylib").chmod(0o500) + self.bundle_leaf.chmod(0o755) + with self.assertRaisesRegex(ValueError, "root identity is invalid"): + self._build_activation() + + def test_bundle_source_requires_exact_membership(self) -> None: + self._activate() + self.bundle_leaf.chmod(0o700) + extra = self.bundle_leaf / "extra-member" + extra.write_bytes(b"unexpected") + self.bundle_leaf.chmod(0o500) + with self.assertRaisesRegex(ValueError, "membership is not exact"): + self._build_activation() + + self.bundle_leaf.chmod(0o700) + extra.unlink() + (self.bundle_leaf / "libpython3.13.dylib").unlink() + self.bundle_leaf.chmod(0o500) + with self.assertRaisesRegex(ValueError, "membership is not exact"): + self._build_activation() + + def test_bundle_source_rejects_symlinked_member(self) -> None: + self._activate() + self.bundle_leaf.chmod(0o700) + library = self.bundle_leaf / "libpython3.13.dylib" + library.unlink() + library.symlink_to("agent-collab-runtime") + self.bundle_leaf.chmod(0o500) + with self.assertRaisesRegex( + ValueError, "member is unsafe|member identity is invalid" + ): + self._build_activation() + + def test_bundle_source_rejects_hardlinked_member(self) -> None: + self._activate() + os.link( + self.bundle_leaf / "agent-collab-runtime", + self.root / "hardlink-aside", + ) + with self.assertRaisesRegex(ValueError, "member identity is invalid"): + self._build_activation() + + def test_failed_verification_leaves_no_output_artifact(self) -> None: + self._activate() + with mock.patch.object( + self.builder, "verify_archive", side_effect=ValueError("forced failure") + ): + with self.assertRaisesRegex(ValueError, "forced failure"): + self._build_activation() + self.assertFalse(self.archive.exists()) + leftovers = [ + path.name + for path in self.archive.parent.iterdir() + if ".tmp" in path.name + ] + self.assertEqual(leftovers, []) + + def test_verify_archive_binds_runtime_bytes_to_frozen_manifest(self) -> None: + self._activate() + mode = self._build_activation() + target = "runtime/darwin-arm64/agent-collab-runtime.bundle/agent-collab-runtime" + + # Repack the archive swapping the runtime member's bytes for same-size + # garbage while keeping every member's metadata identical: source + # parity cannot notice (the source tree is untouched), only the frozen + # manifest digest binding can. + tampered = self.root / "tampered.plugin" + with tarfile.open(self.archive, "r:gz") as original: + members = [ + (member, original.extractfile(member).read() if member.isfile() else None) + for member in original.getmembers() + ] + import gzip as gzip_module + + with tampered.open("wb") as raw: + with gzip_module.GzipFile( + filename="", mode="wb", fileobj=raw, mtime=0 + ) as compressed: + with tarfile.open( + fileobj=compressed, mode="w", format=tarfile.USTAR_FORMAT + ) as bundle: + for member, payload in members: + if member.name == target: + payload = b"Y" * member.size + if payload is None: + bundle.addfile(member) + else: + import io + + bundle.addfile(member, io.BytesIO(payload)) + + with self.assertRaisesRegex(ValueError, "runtime member digest failed"): + self.builder.verify_archive(self.plugin, tampered, mode=mode) + + def test_verify_archive_binds_archived_manifest_to_frozen_bytes(self) -> None: + self._activate() + mode = self._build_activation() + manifest_path = self.plugin / "runtime-manifest.json" + # A SAME-SIZE, parse-valid, record-preserving byte change after the + # build (identity string tweak): the archived manifest member is a + # non-runtime member, so the regenerated canonical (embedding the new + # on-disk bytes) differs from the candidate → structural compare fails. + text = manifest_path.read_text(encoding="utf-8") + swapped = text.replace("Test Operator", "Test Operatox") + self.assertNotEqual(text, swapped) + self.assertEqual(len(text), len(swapped)) + manifest_path.write_text(swapped, encoding="utf-8") + with self.assertRaisesRegex(ValueError, "structure does not match"): + self.builder.verify_archive(self.plugin, self.archive, mode=mode) + + # A size-changing swap changes the canonical total length → rejected. + manifest_path.write_text(text + " ", encoding="utf-8") + with self.assertRaisesRegex(ValueError, "does not match the canonical layout"): + self.builder.verify_archive(self.plugin, self.archive, mode=mode) + + def test_verify_archive_rejects_non_canonical_member_metadata(self) -> None: + self._activate() + mode = self._build_activation() + target = "signing_policy.py" + + # Repack with one member's mtime perturbed while bytes/modes stay + # identical: source parity cannot notice, only the canonical-metadata + # contract can. + tampered = self.root / "tampered-metadata.plugin" + import gzip as gzip_module + import io + + with tarfile.open(self.archive, "r:gz") as original: + members = [ + (member, original.extractfile(member).read() if member.isfile() else None) + for member in original.getmembers() + ] + with tampered.open("wb") as raw: + with gzip_module.GzipFile( + filename="", mode="wb", fileobj=raw, mtime=0 + ) as compressed: + with tarfile.open( + fileobj=compressed, mode="w", format=tarfile.USTAR_FORMAT + ) as bundle: + for member, payload in members: + if member.name == target: + member.mtime = 1 + if payload is None: + bundle.addfile(member) + else: + bundle.addfile(member, io.BytesIO(payload)) + + with self.assertRaisesRegex(ValueError, "structure does not match|canonical layout"): + self.builder.verify_archive(self.plugin, tampered, mode=mode) + + def test_failed_exclusive_temp_open_preserves_foreign_file(self) -> None: + self._activate() + # Pre-create the exact temp path this invocation would claim: the + # O_EXCL open must fail AND the foreign file must survive untouched. + foreign = ( + self.archive.parent / f".{self.archive.name}.tmp.{os.getpid()}" + ) + foreign.write_bytes(b"foreign artifact") + with self.assertRaises(OSError): + self._build_activation() + self.assertEqual(foreign.read_bytes(), b"foreign artifact") + self.assertFalse(self.archive.exists()) + + def test_output_alias_of_source_trees_is_rejected(self) -> None: + self._activate() + for alias in ( + self.plugin / "aliased-output.plugin", + self.bundle_leaf / "aliased-output.plugin", + ): + with self.subTest(alias=str(alias)): + with self.assertRaisesRegex(ValueError, "alias a source tree"): + self.builder.build_archive( + self.root, + plugin="agent-collab", + output=alias, + bundle_source=self.bundle_leaf, + ) + # A symlinked parent that resolves into a source tree is caught too. + symdir = self.root / "sym-parent" + symdir.symlink_to(self.plugin) + with self.assertRaisesRegex(ValueError, "alias a source tree"): + self.builder.build_archive( + self.root, + plugin="agent-collab", + output=symdir / "aliased-output.plugin", + bundle_source=self.bundle_leaf, + ) + self.assertFalse((self.plugin / "aliased-output.plugin").exists()) + + def test_pack_time_failure_leaves_no_output_or_temp(self) -> None: + self._activate() + with mock.patch.object( + self.builder, + "_synthesized_tarinfo", + side_effect=OSError("forced pack failure"), + ): + with self.assertRaises(OSError): + self._build_activation() + self.assertFalse(self.archive.exists()) + leftovers = [ + path.name + for path in self.archive.parent.iterdir() + if ".tmp" in path.name + ] + self.assertEqual(leftovers, []) + + def test_cli_builds_activation_archive_with_bundle_source(self) -> None: + self._activate() + exit_code = self.builder.main( + [ + "--repo-root", + str(self.root), + "--output", + str(self.archive), + "--bundle-source", + str(self.bundle_leaf), + ] + ) + self.assertEqual(exit_code, 0) + self.assertTrue(self.archive.is_file()) + self.builder.verify_archive(self.plugin, self.archive, mode="activation") + + def test_verify_bounds_hostile_archive_decompression(self) -> None: + self._activate() + mode = self._build_activation() + + # A gzip bomb (tiny compressed, huge decompressed) with a CANONICAL + # header (mtime=0) must be rejected by the hard decompression bound, + # not materialized. + import gzip as gzip_module + + bomb = self.root / "bomb.plugin" + with bomb.open("wb") as raw: + # filename="" so no FNAME flag is set (canonical header), matching + # what the builder's os.fdopen path emits. + with gzip_module.GzipFile( + filename="", fileobj=raw, mode="wb", mtime=0 + ) as stream: + zero_chunk = b"\0" * (1024 * 1024) + for _ in range(2 * 64 + 8): # > 2 * MAX_ARTIFACT_BYTES + stream.write(zero_chunk) + with self.assertRaisesRegex(ValueError, "decompression exceeds"): + self.builder.verify_archive(self.plugin, bomb, mode=mode) + + # A non-canonical gzip header (non-zero mtime) is rejected up front. + stamped = self.root / "stamped.plugin" + with stamped.open("wb") as raw: + with gzip_module.GzipFile( + fileobj=raw, mode="wb", mtime=1_700_000_000 + ) as stream: + stream.write(b"whatever") + with self.assertRaisesRegex(ValueError, "gzip header is not canonical"): + self.builder.verify_archive(self.plugin, stamped, mode=mode) + + # The FULL 10-byte header is bound: re-gzipping identical content at a + # different compression level (changes the XFL byte) is rejected even + # though the inflated content is identical. + inflated_ok = gzip_module.decompress(self.archive.read_bytes()) + xfl_variant = self.root / "xfl.plugin" + with xfl_variant.open("wb") as raw: + with gzip_module.GzipFile( + filename="", fileobj=raw, mode="wb", mtime=0, compresslevel=1 + ) as gz: + gz.write(inflated_ok) + self.assertNotEqual( + xfl_variant.read_bytes()[:10], self.archive.read_bytes()[:10] + ) + with self.assertRaisesRegex(ValueError, "gzip header is not canonical"): + self.builder.verify_archive(self.plugin, xfl_variant, mode=mode) + + # A symlinked archive path is refused outright. + alias = self.root / "archive-alias.plugin" + alias.symlink_to(self.archive) + with self.assertRaisesRegex(ValueError, "unreadable"): + self.builder.verify_archive(self.plugin, alias, mode=mode) + + def test_ustar_serializer_is_golden_across_python_versions(self) -> None: + # Design item 5: the regenerate-and-compare architecture requires the + # USTAR serializer to be byte-identical in the operator's build env + # (3.13.14) and CI (3.10/3.12/3.14). This golden binds the exact bytes; + # CI's version matrix fails here if any Python serializes USTAR + # differently, so a legitimate cross-env archive can never fail the + # verify byte-compare undetected. + buffer = io.BytesIO() + with tarfile.open( + fileobj=buffer, mode="w", format=tarfile.USTAR_FORMAT + ) as tar: + tar.addfile( + self.builder._synthesized_tarinfo( + "runtime/darwin-arm64/agent-collab-runtime.bundle", + mode=0o500, + directory=True, + ) + ) + tar.addfile( + self.builder._synthesized_tarinfo( + "runtime/darwin-arm64/agent-collab-runtime.bundle/agent-collab-runtime", + mode=0o500, + size=5, + ), + io.BytesIO(b"hello"), + ) + data = buffer.getvalue() + self.assertEqual( + hashlib.sha256(data).hexdigest(), + "366c3a87cf260b77a3c937b6c44ad6f91d91f1073d59d195a9ae761ec66b6bd0", + ) + + def test_non_runtime_modes_are_umask_normalized(self) -> None: + # Non-runtime member modes must be exactly 0o644 (files) / 0o755 (dirs) + # regardless of the checkout host's umask, so the operator's build and + # CI's regeneration byte-match. Perturb a source file to 0o600 (as a + # umask-077 checkout would) and confirm the archive normalizes it. + self._activate() + (self.plugin / "signing_policy.py").chmod(0o600) + mode = self._build_activation() + self.assertEqual(mode, "activation") + with tarfile.open(self.archive, "r:gz") as bundle: + for member in bundle.getmembers(): + if member.name.startswith("runtime/"): + continue # runtime members keep the sealed 0o500 + expected = 0o755 if member.isdir() else 0o644 + self.assertEqual( + stat.S_IMODE(member.mode), + expected, + f"{member.name} mode not normalized", + ) + # And it still verifies (build and verify both normalize identically). + self.builder.verify_archive(self.plugin, self.archive, mode="activation") + + def test_canonical_tar_is_byte_reproducible_across_calls(self) -> None: + # Design item 5: the canonical inflated tar must be byte-identical on + # regeneration (the cross-environment reproducibility contract the + # regenerate-and-compare architecture depends on). CI runs this on + # 3.10/3.12/3.14, so a serializer-version divergence would fail here. + self._activate() + plugin = self.plugin.resolve(strict=True) + frozen = self.builder._read_manifest_bytes(plugin) + records = self.builder.runtime_bundle.validate_file_records( + self.builder._parse_manifest(frozen)["artifacts"][0]["files"] + ) + plan = self.builder._member_plan(plugin, mode="activation", records=records) + rbn = { + (self.builder.RUNTIME_BUNDLE_REL / r["path"]).as_posix(): r + for r in records + } + zero = {name: b"\x00" * r["size"] for name, r in rbn.items()} + first, ranges_a = self.builder._emit_canonical_tar( + plan, plugin_path=plugin, frozen_manifest=frozen, + record_by_name=rbn, runtime_payloads=zero, + ) + second, ranges_b = self.builder._emit_canonical_tar( + plan, plugin_path=plugin, frozen_manifest=frozen, + record_by_name=rbn, runtime_payloads=zero, + ) + self.assertEqual(first, second) + self.assertEqual(ranges_a, ranges_b) + # The reported ranges are one-to-one with the manifest, disjoint, and + # in-bounds (item 6). + self.builder._assert_runtime_range_map(ranges_a, rbn, len(first)) + + def test_verify_rejects_raw_structural_tamper_without_parsing(self) -> None: + # A hostile archive whose runtime payload digest is preserved but whose + # STRUCTURE differs (here: a directory member's mode flipped) must be + # rejected by the byte-compare, with no reliance on tarfile parsing the + # candidate. We tamper the inflated tar bytes directly. + self._activate() + mode = self._build_activation() + raw = self.archive.read_bytes() + import gzip as gzip_module + + inflated = gzip_module.decompress(raw) + # Flip one byte inside the first 512-byte header block (the manifest or + # first member's metadata region) — a structural mutation. + tampered_tar = bytearray(inflated) + tampered_tar[100] ^= 0x01 + tampered = self.root / "raw-tamper.plugin" + with tampered.open("wb") as fh: + with gzip_module.GzipFile( + filename="", fileobj=fh, mode="wb", mtime=0 + ) as gz: + gz.write(bytes(tampered_tar)) + with self.assertRaisesRegex( + ValueError, "structure does not match|canonical layout" + ): + self.builder.verify_archive(self.plugin, tampered, mode=mode) + + def test_activation_archive_build_is_deterministic(self) -> None: + self._activate() + self._build_activation() + second = self.root / "agent-collab-second.plugin" + self.builder.build_archive( + self.root, + plugin="agent-collab", + output=second, + bundle_source=self.bundle_leaf, + ) + self.assertEqual(self.archive.read_bytes(), second.read_bytes()) + def test_archive_rejects_unexpected_manifest_and_skill_members(self) -> None: extra_manifest = self.plugin / ".claude-plugin" / "extra.dat" extra_manifest.write_text("unexpected", encoding="utf-8") diff --git a/tests/test_release_evidence.py b/tests/test_release_evidence.py index be343a1..7e12618 100644 --- a/tests/test_release_evidence.py +++ b/tests/test_release_evidence.py @@ -102,12 +102,9 @@ def _build_activation(self): ) for name in LEGAL_FILES: shutil.copy2(ROOT / name, repo / name) - bundle = ( - plugin - / "runtime" - / "darwin-arm64" - / "agent-collab-runtime.bundle" - ) + # The signed bundle ships as a release asset (never committed): the + # fixture stages it OUT-of-tree and supplies it via bundle_source. + bundle = self.root / "handoff" / "agent-collab-runtime.bundle" bundle.mkdir(parents=True) runtime_members = { "agent-collab-runtime": ( @@ -192,7 +189,10 @@ def _build_activation(self): json.dumps(manifest), encoding="utf-8" ) mode = archive_builder.build_archive( - repo, plugin="agent-collab", output=self.archive + repo, + plugin="agent-collab", + output=self.archive, + bundle_source=bundle, ) self.assertEqual(mode, "activation") evidence_builder.build_evidence(