searchi/tools/package_windows.py

"""Build a Windows desktop archive from already compiled, staged resources.

This packages an unsigned development build; it does not publish or certify
an installer. No model cache, user settings or local index enters the archive.
"""
import argparse
import hashlib
import json
import os
import re
from pathlib import Path
import struct
import stat
import tempfile
import zipfile

ROOT = Path(__file__).resolve().parents[1]
CRT_LIBRARIES = {"concrt140.dll", "msvcp140.dll", "msvcp140_1.dll", "msvcp140_2.dll", "msvcp140_atomic_wait.dll", "msvcp140_codecvt_ids.dll", "vccorlib140.dll", "vcruntime140.dll", "vcruntime140_1.dll", "vcruntime140_threads.dll"}
MAX_MEMBER = 256 * 1024 * 1024
MAX_MEDIA_TOTAL = 1024 * 1024 * 1024


def digest(path):
    with path.open("rb") as stream:
        return hashlib.file_digest(stream, "sha256").hexdigest()


def media_provenance(root, resources, architecture):
    """Compare staged helpers/notices with the checksum-pinned upstream ZIP."""
    manifest_path = root / "tools/media-runtime/manifest.json"
    if not manifest_path.is_file() or manifest_path.stat().st_size > 65536:
        raise ValueError("Required pinned media manifest is missing or oversized")
    manifest_bytes = manifest_path.read_bytes()
    manifest = json.loads(manifest_bytes)
    artifact = manifest["platforms"][f"windows-{architecture}"]
    name = artifact.get("archive")
    if not isinstance(name, str) or not re.fullmatch(r"[A-Za-z0-9_.-]+\.zip", name):
        raise ValueError("Invalid media archive name")
    archive = root / ".searchi/media-runtime" / name
    if (not archive.is_file() or archive.is_symlink() or archive.absolute() != archive.resolve()
            or archive.stat().st_size != artifact["bytes"] or digest(archive) != artifact["sha256"]):
        raise ValueError("Media archive provenance mismatch")
    staged_manifest = resources / "media-runtime/manifest.json"
    if (not staged_manifest.is_file() or staged_manifest.stat().st_size != len(manifest_bytes)
            or staged_manifest.read_bytes() != manifest_bytes):
        raise ValueError("Staged media manifest differs from pinned manifest")
    selected = {"media-runtime/manifest.json": {"bytes": len(manifest_bytes), "sha256": hashlib.sha256(manifest_bytes).hexdigest()}}
    selected_keys = {name.casefold() for name in selected}
    with zipfile.ZipFile(archive) as package:
        members = package.infolist()
        if len(members) > 10000:
            raise ValueError("Media archive exceeds member ceiling")
        total = 0
        for member in members:
            total += member.file_size
            if member.file_size < 0 or member.file_size > MAX_MEMBER or total > MAX_MEDIA_TOTAL:
                raise ValueError("Media archive exceeds size ceiling")
            parts = member.filename.replace("\\", "/").split("/")
            basename = parts[-1]
            if basename in {"ffmpeg.exe", "ffprobe.exe"}:
                target = "media-runtime/bin/" + basename
            elif basename.lower().startswith(("license", "copying")) or basename.lower() in {"readme.txt", "readme.md"}:
                if any(part in {"", ".", ".."} or ":" in part for part in parts):
                    raise ValueError("Unsafe media notice path")
                target = "media-runtime/licenses/" + "/".join(parts)
            else:
                continue
            mode = member.external_attr >> 16
            if (any(part in {"", ".", ".."} or ":" in part for part in parts)
                    or member.is_dir() or stat.S_IFMT(mode) not in {0, stat.S_IFREG} or member.flag_bits & 1
                    or member.file_size <= 0 or target.casefold() in selected_keys):
                raise ValueError("Invalid or duplicate media member")
            path = resources / target
            if (not path.is_file() or path.is_symlink() or path.absolute() != path.resolve()
                    or not path.resolve().is_relative_to(resources.resolve())):
                raise ValueError(f"Staged media helper/notice missing or outside resources: {target}")
            with package.open(member) as stream:
                checksum = hashlib.file_digest(stream, "sha256").hexdigest()
            if path.stat().st_size != member.file_size or digest(path) != checksum:
                raise ValueError(f"Media helper/notice differs from pinned archive: {target}")
            selected[target] = {"bytes": member.file_size, "sha256": checksum}
            selected_keys.add(target.casefold())
    if (not {"media-runtime/bin/ffmpeg.exe", "media-runtime/bin/ffprobe.exe"}.issubset(selected)
            or not any(name.startswith("media-runtime/licenses/") for name in selected)):
        raise ValueError("Media archive lacks required helpers or notices")
    return selected


def pe_machine(path):
    with path.open("rb") as stream:
        header = stream.read(64)
        if len(header) != 64 or header[:2] != b"MZ":
            raise ValueError(f"Not a Windows executable: {path.name}")
        offset = struct.unpack_from("<I", header, 60)[0]
        if offset > path.stat().st_size - 6:
            raise ValueError(f"Invalid executable header: {path.name}")
        stream.seek(offset)
        signature = stream.read(6)
        if signature[:4] != b"PE\0\0":
            raise ValueError(f"Invalid PE signature: {path.name}")
        return struct.unpack_from("<H", signature, 4)[0]


def package(root, executable, output):
    resources = root / "apps/desktop/src-tauri/resources"
    machine = pe_machine(executable)
    architecture = {0x8664: "x86_64", 0xAA64: "aarch64"}.get(machine)
    if architecture is None:
        raise ValueError("Unsupported Windows architecture")
    inventory = resources / "runtime-libraries.json"
    if not inventory.is_file() or inventory.stat().st_size > 65536:
        raise ValueError("Required runtime dependency inventory is missing or oversized")
    inventory_bytes = inventory.read_bytes()
    manifest = json.loads(inventory_bytes)
    if not isinstance(manifest, dict):
        raise ValueError("Invalid runtime dependency inventory")
    libraries = manifest.get("libraries")
    if not isinstance(libraries, list) or not 1 <= len(libraries) <= 64:
        raise ValueError("Invalid runtime dependency inventory")
    expected = {"searchi-gemma-worker.exe": {"sha256": manifest.get("worker_sha256")}}
    for record in libraries:
        if not isinstance(record, dict):
            raise ValueError("Invalid runtime library metadata")
        name = record.get("path", "")
        if not isinstance(name, str) or not re.fullmatch(r"[A-Za-z0-9_-]+\.dll", name) or name.lower() in {key.lower() for key in expected}:
            raise ValueError("Invalid or duplicate runtime library path")
        if type(record.get("bytes")) is not int or record["bytes"] <= 0:
            raise ValueError("Invalid runtime library size")
        expected[name] = record
    if not ({"ggml-cpu-x64.dll", "ggml-vulkan.dll", "vulkan-1.dll", "ggml.dll", "ggml-base.dll", "llama.dll", "mtmd.dll"} | CRT_LIBRARIES).issubset(expected):
        raise ValueError("Runtime inventory lacks required CPU/GPU libraries")
    expected[inventory.name] = {"bytes": len(inventory_bytes), "sha256": hashlib.sha256(inventory_bytes).hexdigest()}
    for name, record in expected.items():
        path = resources / name
        if path.is_symlink() or not path.resolve().is_relative_to(resources.resolve()) or not path.is_file():
            raise ValueError(f"Required runtime library is missing or outside staging: {name}")
        checksum = record.get("sha256", "")
        if not isinstance(checksum, str) or not re.fullmatch(r"[0-9a-f]{64}", checksum):
            raise ValueError("Invalid runtime checksum")
        with path.open("rb") as stream:
            if hashlib.file_digest(stream, "sha256").hexdigest() != checksum:
                raise ValueError(f"Runtime checksum mismatch: {name}")
        if "bytes" in record and path.stat().st_size != record["bytes"]:
            raise ValueError(f"Runtime size mismatch: {name}")
    media = media_provenance(root, resources, architecture)
    expected.update(media)
    selected = [(executable, "searchi.exe"), (root / "LICENSE", "LICENSE.txt")]
    for name in ["searchi-gemma-worker.exe", "LLAMA-LICENSE.txt", "MSVC-REDISTRIBUTION.txt", "vulkan-1.dll", "VULKAN-LICENSE.txt", "media-runtime/manifest.json",
                 "media-runtime/bin/ffmpeg.exe", "media-runtime/bin/ffprobe.exe",
                 "licenses/Inter-OFL.txt", "licenses/Fluent-MIT.txt", "media-runtime/licenses/LICENSE.txt"]:
        selected.append((resources / name, name))
    selected.append((inventory, inventory.name))
    selected_names = {name for _, name in selected}
    selected.extend((resources / name, name) for name in expected if name not in selected_names)
    for directory in ["licenses"]:
        notices = list((resources / directory).rglob("*.txt"))
        if not notices:
            raise ValueError(f"Required notices are missing: {directory}")
        selected_names = {name for _, name in selected}
        selected.extend((path, path.relative_to(resources).as_posix()) for path in notices
                        if path.relative_to(resources).as_posix() not in selected_names)
    for path, name in selected:
        if not path.is_file():
            raise ValueError(f"Required package file is missing: {name}")
        if path != executable and path != root / "LICENSE" and not path.resolve().is_relative_to(resources.resolve()):
            raise ValueError("Resource resolves outside the staging directory")
        if name.endswith((".exe", ".dll")) and pe_machine(path) != machine:
            raise ValueError(f"Executable architecture differs: {name}")
    records = []
    output.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile(dir=output.parent, suffix=".partial", delete=False) as temporary:
        temporary_path = Path(temporary.name)
    try:
        with zipfile.ZipFile(temporary_path, "w", zipfile.ZIP_DEFLATED, compresslevel=6) as archive:
            for path, name in selected:
                # Hash and count the same blocks written to the archive: a
                # second source read or subsequent stat can describe different
                # bytes when a staged file changes during release preparation.
                checksum = hashlib.sha256()
                count = 0
                with path.open("rb") as stream, archive.open(name, "w", force_zip64=True) as entry:
                    while block := stream.read(1024 * 1024):
                        entry.write(block)
                        checksum.update(block)
                        count += len(block)
                records.append({"path": name, "bytes": count, "sha256": checksum.hexdigest()})
                if name in expected and (checksum.hexdigest() != expected[name]["sha256"] or ("bytes" in expected[name] and count != expected[name]["bytes"])):
                    raise ValueError(f"Runtime changed while packaging: {name}")
            archive.writestr("PACKAGE.json", json.dumps({"platform": "windows", "architecture": architecture,
                "status": "unsigned development package; clean-machine verification pending", "files": records}, indent=2))
            archive.writestr("START-HERE.txt", "searchi — Find what matters. Locally.\n\n"
                "Extract the entire folder, then open searchi.exe. Keep its helpers alongside it.\n"
                "Windows needs WebView2. The Microsoft C++ runtime ships with these helpers; Visual Studio is not required.\n"
                "First launch downloads approximately 865 MB of verified Google model files.\n"
                "Choose Add folder to index your own files; ordinary search stays local.\n"
                "This is an unsigned development build, not a certified production release.\n")
        os.replace(temporary_path, output)
    finally:
        temporary_path.unlink(missing_ok=True)
    return architecture


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--executable", type=Path, default=ROOT / "apps/desktop/src-tauri/target/release/searchi-desktop.exe")
    parser.add_argument("--output", type=Path, default=ROOT / ".searchi/packages/searchi-windows.zip")
    args = parser.parse_args()
    architecture = package(ROOT, args.executable, args.output)
    print(f"Packaged Windows {architecture}: {args.output}")


if __name__ == "__main__":
    main()