#!/usr/bin/env python3 """Create a reproducible Stage 0 baseline from an untrusted GLMChat release ZIP.""" from __future__ import annotations import argparse import hashlib import json import os import platform import re import shutil import stat import subprocess import sys import time import zipfile from pathlib import Path, PurePosixPath from typing import Any DRIVE_PATH = re.compile(r'^[A-Za-z]:') TEST_COMMANDS = [ ['npm', 'ci'], ['npm', 'run', 'lint'], ['npm', 'run', 'test:node'], ['npm', 'run', 'test:php'], ['npm', 'run', 'test:migrations'], ['npm', 'run', 'test:browser'], ['npm', 'run', 'build'], ['npm', 'run', 'verify'], ] def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open('rb') as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b''): digest.update(chunk) return digest.hexdigest() def file_record(path: Path, root: Path) -> dict[str, Any]: relative = path.relative_to(root).as_posix() metadata = path.stat() return { 'path': relative, 'size': metadata.st_size, 'sha256': sha256_file(path), 'mode': oct(stat.S_IMODE(metadata.st_mode)), } def inspect_archive(archive: zipfile.ZipFile) -> list[dict[str, Any]]: records: list[dict[str, Any]] = [] seen: set[str] = set() for info in archive.infolist(): raw = info.filename normalised = raw.replace('\\', '/') path = PurePosixPath(normalised) unix_type = stat.S_IFMT(info.external_attr >> 16) reasons: list[str] = [] if not raw or '\x00' in raw: reasons.append('empty-or-null-name') if '\\' in raw: reasons.append('backslash-path') if path.is_absolute() or normalised.startswith('/'): reasons.append('absolute-path') if DRIVE_PATH.match(normalised): reasons.append('drive-letter-path') if '..' in path.parts: reasons.append('path-traversal') if normalised in seen: reasons.append('duplicate-entry') seen.add(normalised) if unix_type == stat.S_IFLNK: reasons.append('symlink') elif unix_type not in (0, stat.S_IFREG, stat.S_IFDIR): reasons.append('device-or-special-file') records.append({ 'name': raw, 'normalised_name': normalised, 'compressed_size': info.compress_size, 'uncompressed_size': info.file_size, 'crc32': f'{info.CRC:08x}', 'unix_type': oct(unix_type), 'unsafe_reasons': reasons, }) unsafe = [record for record in records if record['unsafe_reasons']] if unsafe: summary = '; '.join(f"{item['name']}: {','.join(item['unsafe_reasons'])}" for item in unsafe) raise RuntimeError(f'Unsafe archive entries rejected: {summary}') return records def command_version(command: list[str]) -> str: try: result = subprocess.run(command, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=30, check=False) return result.stdout.strip().splitlines()[0] if result.stdout.strip() else f'exit {result.returncode}' except (OSError, subprocess.SubprocessError) as error: return f'unavailable: {error}' def run_command(command: list[str], cwd: Path, log_handle) -> dict[str, Any]: started = time.time() log_handle.write(f"\n$ {' '.join(command)}\n") log_handle.flush() try: process = subprocess.run(command, cwd=cwd, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False) output = process.stdout or '' log_handle.write(output) if output and not output.endswith('\n'): log_handle.write('\n') log_handle.write(f'[exit={process.returncode}]\n') log_handle.flush() return { 'command': command, 'exit_code': process.returncode, 'duration_seconds': round(time.time() - started, 3), 'classification': 'pass' if process.returncode == 0 else 'pre_existing_baseline_failure', } except OSError as error: log_handle.write(f'[execution-error] {error}\n') log_handle.flush() return { 'command': command, 'exit_code': None, 'duration_seconds': round(time.time() - started, 3), 'classification': 'pre_existing_baseline_failure', 'error': str(error), } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument('zip_path', type=Path, help='Unmodified input release ZIP') parser.add_argument('output_dir', nargs='?', type=Path, default=Path('evidence/phase0')) args = parser.parse_args() zip_path = args.zip_path.expanduser().resolve(strict=True) output_dir = args.output_dir.expanduser().resolve() if zip_path.suffix.lower() != '.zip' or not zip_path.is_file(): raise SystemExit('zip_path must identify an existing ZIP file.') if output_dir == zip_path.parent or zip_path in output_dir.parents: raise SystemExit('output_dir must not contain or overwrite the input ZIP.') if output_dir.exists(): shutil.rmtree(output_dir) original_dir = output_dir / 'original' extraction_dir = output_dir / 'clean-extraction' original_dir.mkdir(parents=True) extraction_dir.mkdir(parents=True) initial_hash = sha256_file(zip_path) preserved = original_dir / zip_path.name shutil.copy2(zip_path, preserved) preserved_hash = sha256_file(preserved) if preserved_hash != initial_hash: raise SystemExit('Preserved input copy hash mismatch.') with zipfile.ZipFile(zip_path) as archive: archive_entries = inspect_archive(archive) crc_failure = archive.testzip() if crc_failure is not None: raise SystemExit(f'ZIP CRC validation failed at {crc_failure}.') archive.extractall(extraction_dir) extracted_files = [] for path in sorted(extraction_dir.rglob('*')): if path.is_symlink(): raise SystemExit(f'Symlink appeared after extraction: {path}') if path.is_file(): extracted_files.append(file_record(path, extraction_dir)) version_path = extraction_dir / 'VERSION' version = version_path.read_text(encoding='utf-8').strip() if version_path.is_file() else None migrations = sorted((extraction_dir / 'server' / 'migrations').glob('*.sql')) migration_level = max((int(path.name.split('_', 1)[0]) for path in migrations), default=0) inventory = { 'input_zip': str(zip_path), 'input_sha256': initial_hash, 'crc_passed': True, 'archive_entry_count': len(archive_entries), 'archive_entries': archive_entries, 'extracted_file_count': len(extracted_files), 'extracted_files': extracted_files, 'version': version, 'migration_level': migration_level, } (output_dir / 'inventory.json').write_text(json.dumps(inventory, indent=2) + '\n', encoding='utf-8') tests = [] with (output_dir / 'baseline-tests.log').open('w', encoding='utf-8') as log_handle: for command in TEST_COMMANDS: tests.append(run_command(command, extraction_dir, log_handle)) final_hash = sha256_file(zip_path) preserved_final_hash = sha256_file(preserved) report = { 'stage': 0, 'status': 'pass' if all(item['exit_code'] == 0 for item in tests) and final_hash == initial_hash == preserved_final_hash else 'fail', 'input_zip': str(zip_path), 'input_sha256_before': initial_hash, 'input_sha256_after': final_hash, 'preserved_copy': str(preserved), 'preserved_sha256': preserved_final_hash, 'original_unchanged': final_hash == initial_hash == preserved_final_hash, 'crc_passed': True, 'unsafe_entries_accepted': False, 'version': version, 'migration_level': migration_level, 'environment': { 'platform': platform.platform(), 'python': sys.version.splitlines()[0], 'node': command_version(['node', '--version']), 'npm': command_version(['npm', '--version']), 'php': command_version(['php', '--version']), }, 'tests': tests, 'pre_existing_failures': [item for item in tests if item['exit_code'] != 0], } (output_dir / 'baseline-report.json').write_text(json.dumps(report, indent=2) + '\n', encoding='utf-8') print(json.dumps({'status': report['status'], 'report': str(output_dir / 'baseline-report.json')}, indent=2)) return 0 if report['status'] == 'pass' else 1 if __name__ == '__main__': raise SystemExit(main())