#!/usr/bin/env python3 """Deterministically add one verified native Ty executable to its wrapper wheel.""" from __future__ import annotations import argparse import base64 import csv import hashlib import io import json import os import stat import subprocess import sys import zipfile from pathlib import Path from typing import cast VERSION = "0.0.21" SCRIPT_MEMBER = f"ty-{VERSION}.data/scripts/ty" RECORD_MEMBER = f"ty-{VERSION}.dist-info/RECORD" FIXED_TIME = (1980, 1, 1, 0, 0, 0) def canonical(value: object) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() def sha(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def file_sha(path: Path) -> str: return sha(path.read_bytes()) def record_hash(data: bytes) -> str: encoded = base64.urlsafe_b64encode(hashlib.sha256(data).digest()).rstrip(b"=") return "sha256=" + encoded.decode() def private_file(path: Path, label: str, modes: set[int]) -> None: info = path.lstat() if ( stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() or info.st_nlink != 1 or stat.S_IMODE(info.st_mode) not in modes ): raise ValueError(f"{label} is not a private owned regular file") def validate_native(path: Path) -> tuple[bytes, str]: private_file(path, "native Ty executable", {0o700, 0o500}) data = path.read_bytes() if ( len(data) < 64 or data[:4] != b"\x7fELF" or data[4] != 2 or data[5] != 1 or int.from_bytes(data[18:20], "little") != 62 ): raise ValueError("native Ty executable is not Linux x86_64 ELF") completed = subprocess.run( [str(path), "--version"], cwd=path.parent, env={"PATH": os.defpath, "LC_ALL": "C", "LANG": "C"}, text=True, capture_output=True, check=False, timeout=30, ) version = completed.stdout.strip() if completed.returncode != 0 or version != f"ty {VERSION}": raise ValueError("native Ty executable version differs") return data, version def wrapper_members(path: Path) -> dict[str, bytes]: private_file(path, "Ty wrapper wheel", {0o600, 0o400}) try: with zipfile.ZipFile(path) as archive: infos = archive.infolist() names = [info.filename for info in infos] if ( len(names) != len(set(names)) or SCRIPT_MEMBER in names or RECORD_MEMBER not in names or any( name.startswith("/") or ".." in Path(name).parts or name.endswith("/") for name in names ) ): raise ValueError("Ty wrapper wheel inventory is invalid") members = {name: archive.read(name) for name in names} except zipfile.BadZipFile as exc: raise ValueError("Ty wrapper wheel is invalid") from exc try: rows = list(csv.reader(io.StringIO(members[RECORD_MEMBER].decode(), newline=""))) except (UnicodeDecodeError, csv.Error) as exc: raise ValueError("Ty wrapper RECORD is invalid") from exc seen: set[str] = set() for row in rows: if len(row) != 3 or not row[0] or row[0] in seen: raise ValueError("Ty wrapper RECORD row is invalid") seen.add(row[0]) if row[0] == RECORD_MEMBER: if row[1:] != ["", ""]: raise ValueError("Ty wrapper RECORD self row is invalid") continue payload = members.get(row[0]) if payload is None or row[1] != record_hash(payload) or row[2] != str(len(payload)): raise ValueError("Ty wrapper RECORD binding differs") if seen != set(members): raise ValueError("Ty wrapper RECORD inventory differs") members.pop(RECORD_MEMBER) return members def zip_info(name: str, mode: int) -> zipfile.ZipInfo: info = zipfile.ZipInfo(name, FIXED_TIME) info.create_system = 3 info.compress_type = zipfile.ZIP_DEFLATED info.external_attr = (stat.S_IFREG | mode) << 16 return info def build(wrapper: Path, native: Path, output: Path, provenance: Path) -> dict[str, object]: members = wrapper_members(wrapper) native_data, version_output = validate_native(native) members[SCRIPT_MEMBER] = native_data record_rows = [ (name, record_hash(data), str(len(data))) for name, data in sorted(members.items()) ] record_rows.append((RECORD_MEMBER, "", "")) stream = io.StringIO(newline="") csv.writer(stream, lineterminator="\n").writerows(record_rows) members[RECORD_MEMBER] = stream.getvalue().encode() output.parent.mkdir(parents=True, exist_ok=True, mode=0o700) with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as archive: for name in sorted(members): mode = 0o755 if name == SCRIPT_MEMBER else 0o644 archive.writestr(zip_info(name, mode), members[name]) output.chmod(0o600) document: dict[str, object] = { "schema": "task26-corrected-ty-wheel-build-provenance-v1", "builder_sha256": file_sha(Path(__file__)), "wrapper_wheel_sha256": file_sha(wrapper), "native_executable_sha256": sha(native_data), "native_executable_size": len(native_data), "native_executable_version": version_output, "native_executable_format": "ELF64-little-x86_64-linux", "wheel_member": SCRIPT_MEMBER, "output_wheel_sha256": file_sha(output), "output_wheel_size": output.stat().st_size, "deterministic_zip_timestamp": list(FIXED_TIME), } document["provenance_sha256"] = sha(canonical(document)) provenance.write_bytes(canonical(document) + b"\n") provenance.chmod(0o600) return document def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--wrapper-wheel", type=Path, required=True) parser.add_argument("--native-ty", type=Path, required=True) parser.add_argument("--output-wheel", type=Path, required=True) parser.add_argument("--provenance", type=Path, required=True) args = parser.parse_args() try: result = build(args.wrapper_wheel, args.native_ty, args.output_wheel, args.provenance) except (OSError, ValueError, subprocess.SubprocessError) as exc: print(json.dumps({"status": "TASK26_TY_WHEEL_BUILD_FAIL", "reason": str(exc)}, sort_keys=True), file=sys.stderr) return 1 print(json.dumps(result, sort_keys=True, separators=(",", ":"))) return 0 if __name__ == "__main__": raise SystemExit(main())