"""Reject public artifacts with version, source-tree, or forbidden-release leaks.""" from __future__ import annotations import argparse import hashlib import json import re import stat import subprocess import sys import zipfile from pathlib import Path, PurePosixPath FORBIDDEN = ( "/mnt/" + "f/", "f" + ":\\", "network" + "-deny", "network" + "_rule", "b" + "20", "brief" + " 1", "brief" + "-", "root" + "less", "skylark" + " sentinel", "skylark" + "-sentinel", ) MAX_ARCHIVE_MEMBERS = 256 MAX_ARCHIVE_MEMBER_BYTES = 16 * 1024 * 1024 MAX_ARCHIVE_TOTAL_BYTES = 64 * 1024 * 1024 MAX_ZIP_COMMENT_BYTES = 65_535 def require_exact_zip_end(path: Path) -> None: try: size = path.stat().st_size window_size = min(size, MAX_ZIP_COMMENT_BYTES + 22) with path.open("rb") as handle: handle.seek(size - window_size) tail = handle.read(window_size) except OSError as error: raise SystemExit("release archive cannot be read") from error marker = b"PK\x05\x06" position = tail.find(marker) while position >= 0: if position + 22 <= len(tail): comment_size = int.from_bytes(tail[position + 20 : position + 22], "little") if position + 22 + comment_size == len(tail): return position = tail.find(marker, position + 1) raise SystemExit("release archive is truncated or has trailing data") def validated_members(archive: zipfile.ZipFile) -> list[zipfile.ZipInfo]: members = archive.infolist() if not members or len(members) > MAX_ARCHIVE_MEMBERS: raise SystemExit("release archive has an invalid member count") seen: set[str] = set() total = 0 for member in members: name = member.filename raw_name = name.removesuffix("/") raw_parts = raw_name.split("/") path = PurePosixPath(raw_name) parts = path.parts normalized = path.as_posix().rstrip("/") if ( not name or "\x00" in name or "\\" in name or any(part in ("", ".", "..") for part in raw_parts) or path.is_absolute() or not normalized or any(part in ("", ".", "..") for part in parts) or (len(parts[0]) >= 2 and parts[0][1] == ":") ): raise SystemExit("release archive contains an unsafe path") if normalized in seen: raise SystemExit("release archive contains a duplicate path") seen.add(normalized) mode = (member.external_attr >> 16) & 0xFFFF file_type = stat.S_IFMT(mode) if member.flag_bits & 0x1 or file_type not in (0, stat.S_IFREG, stat.S_IFDIR): raise SystemExit("release archive contains a link or special member") if member.is_dir() != (file_type == stat.S_IFDIR) and file_type != 0: raise SystemExit("release archive member type is inconsistent") if member.file_size > MAX_ARCHIVE_MEMBER_BYTES: raise SystemExit("release archive member exceeds the verification limit") total += member.file_size if total > MAX_ARCHIVE_TOTAL_BYTES: raise SystemExit("release archive exceeds the verification limit") return members def require_exact_archive_prefix(path: Path, members: list[zipfile.ZipInfo]) -> None: start = min(member.header_offset for member in members) try: with path.open("rb") as handle: prefix = handle.read(start) except OSError as error: raise SystemExit("release archive cannot be read") from error allowed = b"#!/usr/bin/env python3\n" if path.suffix == ".pyz" else b"" if prefix != allowed: raise SystemExit("release archive contains prepended or concatenated data") def text_from_zip(path: Path) -> str: require_exact_zip_end(path) with zipfile.ZipFile(path) as archive: members = validated_members(archive) require_exact_archive_prefix(path, members) return "\n".join( archive.read(member).decode("utf-8", errors="ignore").lower() for member in members if member.filename.endswith((".py", ".md", ".toml", ".txt")) ) def digest_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def checksums(path: Path) -> dict[str, str]: result: dict[str, str] = {} for line in path.read_text(encoding="ascii").splitlines(): fields = line.split(maxsplit=1) if len(fields) != 2 or not re.fullmatch(r"[0-9a-f]{64}", fields[0]): raise SystemExit("checksums contain an invalid line") result[fields[1].removeprefix("*")] = fields[0] return result def artifact_source_commit(path: Path) -> str: with zipfile.ZipFile(path) as archive: matches = [ member for member in validated_members(archive) if member.filename == "lumi_eggcracker/build_info.py" ] if len(matches) != 1: raise SystemExit("artifact source identity is missing") raw = archive.read(matches[0]) match = re.fullmatch(b'SOURCE_COMMIT = "([0-9a-f]{40})"\r?\n', raw) if match is None: raise SystemExit("artifact source identity is invalid") return match.group(1).decode("ascii") def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--artifact", required=True, type=Path) parser.add_argument("--source-archive", required=True, type=Path) parser.add_argument("--release-bundle", required=True, type=Path) args = parser.parse_args() if not all( path.is_file() for path in (args.artifact, args.source_archive, args.release_bundle) ): raise SystemExit("release artifacts are missing") for path in (args.artifact, args.source_archive, args.release_bundle): leaks = [item for item in FORBIDDEN if item in text_from_zip(path)] if leaks: raise SystemExit(f"forbidden public artifact content in {path.name}: {leaks}") manifest = json.loads( (args.artifact.parent / "release-manifest.json").read_text(encoding="utf-8") ) version = manifest.get("version") if not isinstance(version, str) or not re.fullmatch(r"[0-9]+\.[0-9]+\.[0-9]+", version): raise SystemExit("release manifest version is invalid") prefix = f"lumi-eggcracker-{version}/" expected_names = { "artifact": f"lumi-eggcracker-{version}.pyz", "source_archive": f"lumi-eggcracker-{version}-source.zip", } if manifest.get("artifact") != expected_names["artifact"] or manifest.get("source_archive") != expected_names["source_archive"]: raise SystemExit("release manifest artifact names are inconsistent") result = subprocess.run( [sys.executable, str(args.artifact), "version"], capture_output=True, text=True, check=False ) if result.returncode or result.stdout.strip() != version: raise SystemExit("artifact version is inconsistent") if manifest.get("artifact") != args.artifact.name: raise SystemExit("release manifest is inconsistent") if manifest.get("source_commit") != artifact_source_commit(args.artifact): raise SystemExit("artifact and manifest source identities differ") expected = { prefix + name for name in ( args.artifact.name, args.source_archive.name, "release-manifest.json", "SHA256SUMS", "README.md", "LICENSE", "LIMITATIONS.md", "SECURITY.md", "SECURITY_MODEL.md", "QUALIFICATION.md", "RELEASE_NOTES.md", "scripts/install.py", "scripts/upgrade.py", "scripts/uninstall.py", "scripts/verify_uninstalled.py", "scripts/verify_release.py", "scripts/first_kill.py", "scripts/support_bundle.py", "scripts/prepare_ai_smoke.py", "scripts/prepare_safetensors_smoke.py", "scripts/ai_smoke_worker.py", "scripts/smoke_local_ai.py", "scripts/smoke_autonomous_ai.py", "scripts/smoke_content_ai.py", "scripts/smoke_safetensors_ai.py", "scripts/smoke_safetensors_aten_negative.py", "scripts/run_content_matrix.py", "scripts/run_content_benign_matrix.py", "scripts/run_content_adversarial_matrix.py", "scripts/self_validate.py", "scripts/run_autonomous_matrix.py", "scripts/run_native_matrix.py", "scripts/run_p0_native.py", "scripts/run_installer_p0.py", "scripts/benchmark_overhead.py", "scripts/qualify_offline_boundary.py", "scripts/qualify_offline_boundary_integration.py", "scripts/verify_evidence.py", "scripts/package_evidence.py", "scripts/verify_evidence_archive.py", "tests/fixtures/benign_model_handler.py", "tests/fixtures/benign_near_limit.py", "tests/fixtures/canary.py", "tests/fixtures/fork_race.py", "tests/fixtures/hostile_client.py", "tests/fixtures/pid_pressure.py", ) } with zipfile.ZipFile(args.release_bundle) as archive: members = validated_members(archive) if {member.filename for member in members} != expected: raise SystemExit("release bundle contents are inconsistent") if ( digest_bytes(archive.read(prefix + args.artifact.name)) != manifest["sha256"] or digest_bytes(archive.read(prefix + args.source_archive.name)) != manifest["source_archive_sha256"] ): raise SystemExit("bundled artifact digest is inconsistent") parsed_sums = checksums(args.artifact.parent / "SHA256SUMS") expected_sums = { args.artifact.name: digest_bytes(args.artifact.read_bytes()), args.source_archive.name: digest_bytes(args.source_archive.read_bytes()), args.release_bundle.name: digest_bytes(args.release_bundle.read_bytes()), } # A detached build directory carries the archive's checksum as its third # entry. The checksum file embedded inside that archive cannot include a # digest of its containing archive without recursion, so an extracted # bundle carries only its artifact and source entries. In both modes the # archive's exact member set and embedded payload digests were checked # above; distribution still relies on the detached SHA256SUMS as its trust # anchor. if args.artifact.parent.resolve() == args.release_bundle.parent.resolve(): required_sums = expected_sums else: required_sums = { args.artifact.name: expected_sums[args.artifact.name], args.source_archive.name: expected_sums[args.source_archive.name], } if parsed_sums != required_sums: raise SystemExit("checksums do not match the published release assets") print(json.dumps({"result": "PASS", "version": version}, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())