#!/usr/bin/env python3 from __future__ import annotations import argparse import base64 import csv import hashlib import io import json import os import re import stat import subprocess import sys import zipfile from pathlib import Path from typing import Mapping, cast FORBIDDEN = re.compile( r"(?i)(^|[^a-z0-9])trainer([^a-z0-9]|$)|" r"trainer_[a-z0-9_]*|[a-z0-9_]*_trainer[a-z0-9_]*|" r"trainer-review|trainer_review|trainer-session|trainer_session|" r"claim_trainer|pilot_trainer|trb[0-9]+|pt1:|트레이너|오늘PT기록|PT기록" ) DIGEST_KEYS = ( "digest_payload", "entries", "source_bindings", "wheel_member_source_map", ) def sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def sha256_file(path: Path) -> str: return sha256_bytes(path.read_bytes()) def canonical(value: object) -> bytes: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") def fail(message: str) -> None: raise AssertionError(message) def verify_regular( path: Path, *, expected_mode: str | None = None, owner_only: bool = True, ) -> None: if path.is_symlink() or not path.is_file(): fail(f"not a regular non-symlink file: {path}") info = path.stat() if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1: fail(f"unsafe file metadata: {path}") mode = f"{stat.S_IMODE(info.st_mode):04o}" if expected_mode is not None and mode != expected_mode: fail(f"mode mismatch for {path}: {mode} != {expected_mode}") if owner_only and stat.S_IMODE(info.st_mode) & 0o077: fail(f"group/other permissions present: {path}") def verify_candidate(root: Path, *, verify_live_sources: bool) -> dict[str, object]: root = root.resolve() manifest_path = root / "candidate-manifest.json" verify_regular(manifest_path, expected_mode="0600") manifest = cast( Mapping[str, object], json.loads(manifest_path.read_text(encoding="utf-8")) ) if manifest.get("schema") != "dualcoach-owner-customer-candidate-v1": fail("unsupported candidate manifest schema") digest_material = {key: manifest[key] for key in DIGEST_KEYS} candidate_digest = sha256_bytes(canonical(digest_material)) if candidate_digest != manifest.get("candidate_digest"): fail("candidate digest mismatch") entries = cast(list[Mapping[str, object]], manifest["entries"]) declared_paths = {str(entry["path"]) for entry in entries} if len(declared_paths) != len(entries): fail("duplicate candidate entry") actual_paths = { path.relative_to(root).as_posix() for path in root.rglob("*") if path.is_file() and path.name != "candidate-manifest.json" } if actual_paths != declared_paths: fail(f"candidate inventory mismatch: {sorted(actual_paths ^ declared_paths)}") if any(path.is_symlink() for path in root.rglob("*")): fail("candidate contains a symlink") for directory in (root, *(path for path in root.rglob("*") if path.is_dir())): if stat.S_IMODE(directory.stat().st_mode) & 0o077: fail(f"directory is not owner-only: {directory}") for entry in entries: path = root / str(entry["path"]) verify_regular(path, expected_mode=str(entry["mode"])) if path.stat().st_size != entry["bytes"]: fail(f"size mismatch: {path}") if sha256_file(path) != entry["sha256"]: fail(f"digest mismatch: {path}") payload = cast(Mapping[str, object], manifest["digest_payload"]) closure_path = root / str(payload["runtime_closure_path"]) closure = cast( Mapping[str, object], json.loads(closure_path.read_text(encoding="utf-8")) ) if closure.get("forbidden_match_count") != 0: fail("runtime closure has a forbidden match") if closure.get("unexpected_dynamic_module_count") != 0: fail("runtime closure has an unexpected dynamic module") if closure.get("removed_importable_module_count") != 0: fail("runtime closure retains a removed module") if closure.get("closure_digest") != payload["runtime_closure_digest"]: fail("runtime closure digest binding mismatch") package_root = root / str(payload["profile_package_snapshot_path"]) package_findings: list[str] = [] for path in package_root.rglob("*"): if not path.is_file(): continue try: text = path.read_text(encoding="utf-8") except UnicodeDecodeError: continue if FORBIDDEN.search(text): package_findings.append(path.relative_to(package_root).as_posix()) if package_findings: fail(f"profile package forbidden matches: {package_findings}") wheel_path = root / str(payload["production_wheel_path"]) if sha256_file(wheel_path) != payload["production_wheel_sha256"]: fail("production wheel digest mismatch") shipped_root = root / str(payload["shipped_wheel_tree_path"]) with zipfile.ZipFile(wheel_path) as archive: infos = archive.infolist() names = [item.filename for item in infos] if len(names) != len(set(names)): fail("production wheel has duplicate members") if {item.date_time for item in infos} != {(2000, 1, 1, 0, 0, 0)}: fail("production wheel timestamp is not reproducible") if any(name.endswith((".pyc", ".pyo")) or "/__pycache__/" in name for name in names): fail("production wheel contains bytecode cache") shipped_names = { path.relative_to(shipped_root).as_posix() for path in shipped_root.rglob("*") if path.is_file() } if shipped_names != set(names): fail("shipped wheel tree does not match wheel membership") for name in names: data = archive.read(name) if data != (shipped_root / name).read_bytes(): fail(f"shipped wheel member mismatch: {name}") try: text = data.decode("utf-8") except UnicodeDecodeError: continue if FORBIDDEN.search(text): fail(f"forbidden production wheel member: {name}") record_name = next( (name for name in names if name.endswith(".dist-info/RECORD")), None ) if record_name is None: fail("production wheel RECORD is missing") rows = list( csv.reader(io.StringIO(archive.read(record_name).decode("utf-8"))) ) if {row[0] for row in rows} != set(names): fail("production wheel RECORD membership mismatch") for name, digest, size in rows: if name == record_name: if digest or size: fail("production wheel RECORD self-row is invalid") continue data = archive.read(name) expected = "sha256=" + base64.urlsafe_b64encode( hashlib.sha256(data).digest() ).rstrip(b"=").decode("ascii") if digest != expected or size != str(len(data)): fail(f"production wheel RECORD mismatch: {name}") source_map = cast(list[Mapping[str, object]], manifest["wheel_member_source_map"]) for item in source_map: member = str(item["wheel_member"]) source = root / str(item["source_snapshot_path"]) if archive.read(member) != source.read_bytes(): fail(f"wheel/source mismatch: {member}") if sha256_file(source) != item["sha256"]: fail(f"wheel/source digest mismatch: {member}") if verify_live_sources: bindings = cast(Mapping[str, object], manifest["source_bindings"]) repository = Path(str(bindings["repository_root"])).resolve() package = Path(str(bindings["profile_package_root"])).resolve() head = subprocess.check_output( ["git", "-C", str(repository), "rev-parse", "HEAD"], text=True ).strip() if head != payload["execution_head"]: fail("execution HEAD changed") status = subprocess.check_output( [ "git", "-C", str(repository), "status", "--porcelain=v1", "-z", "--untracked-files=all", ] ) if sha256_bytes(status) != payload["git_status_sha256"]: fail("execution Git status changed") status_snapshot = root / str(payload["git_status_snapshot_path"]) if status != status_snapshot.read_bytes(): fail("execution Git status snapshot mismatch") for entry in cast(list[Mapping[str, object]], bindings["repository_entries"]): path = repository / str(entry["path"]) verify_regular(path, owner_only=False) if sha256_file(path) != entry["sha256"]: fail(f"execution source changed: {entry['path']}") for entry in cast(list[Mapping[str, object]], bindings["profile_package_entries"]): path = package / str(entry["path"]) verify_regular(path, owner_only=False) if sha256_file(path) != entry["sha256"]: fail(f"profile package source changed: {entry['path']}") return { "schema": "dualcoach-owner-customer-candidate-verification-v1", "status": "PASS", "candidate_digest": candidate_digest, "candidate_manifest_sha256": sha256_file(manifest_path), "candidate_entry_count": len(entries), "production_wheel_sha256": payload["production_wheel_sha256"], "runtime_closure_digest": payload["runtime_closure_digest"], "live_source_verification": verify_live_sources, "forbidden_match_count": 0, "unexpected_dynamic_module_count": 0, "mutations": 0, } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--candidate-root", type=Path, default=Path(__file__).resolve().parent) parser.add_argument("--no-live-sources", action="store_true") args = parser.parse_args() try: result = verify_candidate( args.candidate_root, verify_live_sources=not args.no_live_sources, ) except Exception as exc: print(json.dumps({"status": "FAIL", "error": str(exc)}, sort_keys=True)) return 1 print(json.dumps(result, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())