#!/usr/bin/env python3 """Verify and merge the downloaded draft toolkit; Python 3.10+.""" import argparse, hashlib, json, re, shutil, stat, tempfile, zipfile from pathlib import Path def sha256(data): return hashlib.sha256(data).hexdigest() def safe_name(value): if (not isinstance(value, str) or not value or '\\' in value or ':' in value or '%' in value or value.startswith('/') or '\x00' in value): raise ValueError('Unsafe relative member path') parts = value.split('/') if any(part in ('', '.', '..') or part.endswith((' ', '.')) or not re.fullmatch(r'[A-Za-z0-9_.@+-]+', part) for part in parts): raise ValueError('Unsafe relative member component') reserved = {'con', 'prn', 'aux', 'nul', *(f'com{number}' for number in range(1, 10)), *(f'lpt{number}' for number in range(1, 10))} if any(part.split('.')[0].lower() in reserved for part in parts): raise ValueError('Reserved extraction filename') return value def contained_file(root, relative): safe_name(relative) root = Path(root).resolve() target = root for component in relative.split('/'): target = target / component if target.is_symlink(): raise ValueError('Symlink in source or extraction path') if not target.is_file(): raise ValueError(f'Missing regular file: {relative}') return target def verify_record(data, record): if len(data) != record['bytes'] or sha256(data) != record['sha256']: raise ValueError(f'Hash/size mismatch: {record["path"]}') def register_name(names, name): safe_name(name) folded = name.casefold() if folded in names: raise ValueError(f'Duplicate/case-colliding member: {name}') if any(folded.startswith(previous + '/') or previous.startswith(folded + '/') for previous in names): raise ValueError(f'File/directory collision: {name}') names.add(folded) def read_archive(download_root, record, names): archive_path = contained_file(download_root, record['path']) verify_record(archive_path.read_bytes(), record) expected = {member['path']: member for member in record['members']} if len(expected) != len(record['members']): raise ValueError('Duplicate expected archive member') with zipfile.ZipFile(archive_path) as archive: infos = archive.infolist() if len(infos) != len(expected) or {info.filename for info in infos} != set(expected): raise ValueError('Unexpected archive members') for info in infos: register_name(names, info.filename) mode = (info.external_attr >> 16) & 0xffff if (not info.filename.startswith('project/') or info.is_dir() or stat.S_IFMT(mode) not in (0, stat.S_IFREG) or info.flag_bits & 1): raise ValueError('Archive contains a special/encrypted member') if info.file_size != expected[info.filename]['bytes'] or info.file_size >= 24 * 1024 * 1024: raise ValueError('Archive member size mismatch or too large') verify_record(archive.read(info), expected[info.filename]) def extract_verified(index_path, destination, recipe_ids=None): index_path = Path(index_path) download_root = index_path.parent.resolve() index = json.loads(index_path.read_text()) if index.get('schemaVersion') != 1 or index.get('layout') != 'project/...': raise ValueError('Unsupported toolkit index') selections = index['recipes'] known = {record['id'] for record in selections} if recipe_ids is not None: if not set(recipe_ids) <= known: raise ValueError('Unknown recipe selection') selections = [record for record in selections if record['id'] in recipe_ids] archives = [index['core']] for selection in selections: manifest_record = selection['index'] data = contained_file(download_root, manifest_record['path']).read_bytes() verify_record(data, manifest_record) manifest = json.loads(data) if manifest.get('recipeId') != selection['id'] or manifest.get('archives') != selection['archives']: raise ValueError('Per-recipe manifest/archive selection mismatch') archives.extend(selection['archives']) names = set() archive_names = set() for record in archives: register_name(archive_names, record['path']) if record['bytes'] >= 24 * 1024 * 1024: raise ValueError('Archive exceeds the exclusive 24 MiB ceiling') read_archive(download_root, record, names) destination = Path(destination).absolute() if destination.is_symlink() or destination.exists(): raise ValueError('Extraction destination must not already exist') for parent in destination.parents: if parent.is_symlink(): raise ValueError('Symlink in destination parents') destination.parent.mkdir(parents=True, exist_ok=True) temporary = Path(tempfile.mkdtemp(prefix='.toolkit-extract-', dir=destination.parent)) try: for record in archives: archive_file = contained_file(download_root, record['path']) verify_record(archive_file.read_bytes(), record) with zipfile.ZipFile(archive_file) as archive: for member in record['members']: data = archive.read(member['path']) verify_record(data, member) target = temporary / member['path'] target.parent.mkdir(parents=True, exist_ok=True) with target.open('xb') as handle: handle.write(data) for selection in selections: manifest = json.loads(contained_file(download_root, selection['index']['path']).read_bytes()) for binding in manifest['boundProjectFiles']: verify_record(contained_file(temporary, binding['path']).read_bytes(), binding) if destination.exists(): raise ValueError('Extraction destination appeared during validation') temporary.rename(destination) except BaseException: shutil.rmtree(temporary) raise return {'project': str(destination / 'project'), 'archives': len(archives), 'members': len(names), 'recipes': len(selections), 'boundFilesVerified': True} if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--index', type=Path, default=Path(__file__).with_name('toolkit-index.json')) parser.add_argument('--dest', type=Path, required=True, help='New directory; never overwrites files') parser.add_argument('--recipe', action='append', help='Only this recipe asset pack; repeatable; default all') options = parser.parse_args() print(json.dumps(extract_verified(options.index, options.dest, options.recipe), indent=2))