#!/usr/bin/env python3 from __future__ import annotations import argparse import json import os import shutil import stat import tempfile import zipfile from pathlib import Path, PurePosixPath MAX_ARCHIVE_BYTES = 2 * 1024 * 1024 * 1024 MAX_FILE_BYTES = 512 * 1024 * 1024 MAX_RATIO = 200 class FactoryError(RuntimeError): pass def inspect_archive(path: Path) -> str: if not path.is_file(): raise FactoryError(f"archive does not exist: {path}") try: archive = zipfile.ZipFile(path) except (OSError, zipfile.BadZipFile) as exc: raise FactoryError(f"invalid ZIP {path}: {exc}") from exc roots: set[str] = set() names: set[str] = set() total = 0 with archive: infos = archive.infolist() if not infos: raise FactoryError(f"empty ZIP: {path}") for info in infos: raw = info.filename if not raw or "\\" in raw or raw.startswith("/"): raise FactoryError(f"unsafe ZIP path: {raw!r}") parts = PurePosixPath(raw).parts if not parts or any(part in {"", ".", ".."} for part in parts): raise FactoryError(f"unsafe ZIP path: {raw!r}") normalized = "/".join(parts).rstrip("/") folded = normalized.casefold() if folded in names: raise FactoryError(f"duplicate ZIP path: {raw!r}") names.add(folded) roots.add(parts[0]) mode = info.external_attr >> 16 if stat.S_ISLNK(mode): raise FactoryError(f"ZIP contains symlink: {raw}") if info.file_size > MAX_FILE_BYTES: raise FactoryError(f"ZIP member exceeds size limit: {raw}") total += info.file_size if total > MAX_ARCHIVE_BYTES: raise FactoryError("ZIP uncompressed size exceeds limit") if info.compress_size and info.file_size / info.compress_size > MAX_RATIO: raise FactoryError(f"ZIP member compression ratio exceeds limit: {raw}") if len(roots) != 1: raise FactoryError(f"ZIP must contain exactly one root directory, found {sorted(roots)}") root = next(iter(roots)) required = {f"{root}/style.css".casefold(), f"{root}/index.php".casefold()} if not required.issubset(names): missing = sorted(required - names) raise FactoryError(f"ZIP is not an installable WordPress theme; missing {missing}") return root def install_archive(archive_path: Path, destination: Path) -> str: root = inspect_archive(archive_path) if destination.exists(): raise FactoryError(f"destination already exists: {destination}") destination.parent.mkdir(parents=True, exist_ok=True) staging = Path(tempfile.mkdtemp(prefix=f".{destination.name}.", dir=destination.parent)) try: with zipfile.ZipFile(archive_path) as archive: for info in archive.infolist(): parts = PurePosixPath(info.filename).parts target = staging.joinpath(*parts) if info.is_dir(): target.mkdir(parents=True, exist_ok=True) continue target.parent.mkdir(parents=True, exist_ok=True) with archive.open(info) as source, target.open("wb") as sink: shutil.copyfileobj(source, sink) mode = (info.external_attr >> 16) & 0o777 target.chmod(mode or 0o644) os.replace(staging, destination) except BaseException: shutil.rmtree(staging, ignore_errors=True) raise return root def pack_theme(container: Path, destination: Path) -> str: roots = [entry for entry in container.iterdir() if entry.is_dir()] files = [entry for entry in container.iterdir() if entry.is_file()] metadata = {"generation.json", "response.md"} unexpected = [entry for entry in files if entry.name not in metadata] if len(roots) != 1 or unexpected: raise FactoryError(f"theme container must hold one root plus generation metadata: {container}") root = roots[0] destination.parent.mkdir(parents=True, exist_ok=True) fd, temporary = tempfile.mkstemp(prefix=f".{destination.name}.", dir=destination.parent) os.close(fd) temp_path = Path(temporary) try: with zipfile.ZipFile(temp_path, "w", zipfile.ZIP_DEFLATED, compresslevel=9) as archive: for path in sorted(root.rglob("*"), key=lambda item: item.as_posix().casefold()): if path.is_symlink(): raise FactoryError(f"theme contains symlink: {path}") relative = path.relative_to(container).as_posix() if path.is_dir(): continue info = zipfile.ZipInfo(relative) info.date_time = (1980, 1, 1, 0, 0, 0) info.compress_type = zipfile.ZIP_DEFLATED info.external_attr = (path.stat().st_mode & 0xFFFF) << 16 with path.open("rb") as source, archive.open(info, "w") as sink: shutil.copyfileobj(source, sink) inspect_archive(temp_path) os.replace(temp_path, destination) except BaseException: temp_path.unlink(missing_ok=True) raise return root.name def generated_zip(result_path: Path, output_dir: Path) -> Path: try: result = json.loads(result_path.read_text()) except (OSError, json.JSONDecodeError) as exc: raise FactoryError(f"invalid ask-gpt JSON: {exc}") from exc files = [Path(value).resolve() for value in result.get("files", [])] zips = [path for path in files if path.suffix.casefold() == ".zip"] if len(zips) != 1: raise FactoryError(f"expected exactly one generated ZIP, found {len(zips)}") expected = output_dir.resolve() try: zips[0].relative_to(expected) except ValueError as exc: raise FactoryError(f"generated ZIP escaped output directory: {zips[0]}") from exc if not zips[0].is_file(): raise FactoryError(f"generated ZIP is missing: {zips[0]}") return zips[0] def audit(container: Path, archive: Path) -> None: packed_root = inspect_archive(archive) roots = [entry.name for entry in container.iterdir() if entry.is_dir()] files = {entry.name for entry in container.iterdir() if entry.is_file()} if roots != [packed_root] or not files.issubset({"generation.json", "response.md"}): raise FactoryError( f"theme tree/archive mismatch: roots={roots}, archive={packed_root}, files={sorted(files)}") def main() -> int: parser = argparse.ArgumentParser() commands = parser.add_subparsers(dest="command", required=True) validate = commands.add_parser("validate") validate.add_argument("archive", type=Path) install = commands.add_parser("install") install.add_argument("archive", type=Path) install.add_argument("destination", type=Path) pack = commands.add_parser("pack") pack.add_argument("container", type=Path) pack.add_argument("destination", type=Path) result = commands.add_parser("result") result.add_argument("json", type=Path) result.add_argument("output", type=Path) check = commands.add_parser("audit") check.add_argument("container", type=Path) check.add_argument("archive", type=Path) args = parser.parse_args() if args.command == "validate": print(inspect_archive(args.archive)) elif args.command == "install": print(install_archive(args.archive, args.destination)) elif args.command == "pack": print(pack_theme(args.container, args.destination)) elif args.command == "result": print(generated_zip(args.json, args.output)) else: audit(args.container, args.archive) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except FactoryError as exc: raise SystemExit(f"theme-factory: {exc}") from exc