GEWU_agent/scripts/manifest_tool.py

"""Verify or refresh the portable GEWU instruction-pack integrity manifest."""

from __future__ import annotations

import hashlib
import json
import sys
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
MANIFEST = ROOT / "manifest.json"


def normalized_bytes(path: Path) -> bytes:
    text = path.read_text(encoding="utf-8-sig")
    return text.replace("\r\n", "\n").replace("\r", "\n").encode("utf-8")


def sha256(path: Path) -> str:
    return hashlib.sha256(normalized_bytes(path)).hexdigest()


def snapshot_digest(modules: list[dict[str, Any]]) -> str:
    compact = json.dumps(modules, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(compact.encode("utf-8")).hexdigest()


def load_manifest() -> dict[str, Any]:
    manifest = json.loads(MANIFEST.read_text(encoding="utf-8"))
    if not isinstance(manifest, dict) or not isinstance(manifest.get("modules"), list):
        raise ValueError("manifest.json has an invalid structure")
    return manifest


def module_path(entry: str) -> Path:
    path = (ROOT / entry).resolve()
    if path != ROOT and ROOT not in path.parents:
        raise ValueError(f"module path escapes pack: {entry}")
    if not path.is_file():
        raise ValueError(f"module file is missing: {entry}")
    return path


def verify(manifest: dict[str, Any]) -> None:
    modules = manifest["modules"]
    for module in modules:
        if not isinstance(module, dict) or not isinstance(module.get("entry"), str):
            raise ValueError("manifest module descriptor is invalid")
        if module.get("sha256") != sha256(module_path(module["entry"])):
            raise ValueError(f"module hash mismatch: {module.get('module_id')}")
    expected_snapshot = manifest.get("snapshot", {}).get("digest")
    if expected_snapshot != snapshot_digest(modules):
        raise ValueError("module snapshot digest mismatch")
    for contract in manifest.get("verified_service_contracts", []):
        if not isinstance(contract, dict) or not isinstance(contract.get("path"), str):
            raise ValueError("service contract descriptor is invalid")
        contract_path = module_path(contract["path"])
        if contract.get("sha256") != sha256(contract_path):
            raise ValueError(f"service contract hash mismatch: {contract.get('contract_id')}")
        service_contract = json.loads(contract_path.read_text(encoding="utf-8"))
        catalog_path = module_path(str(contract_path.parent / "catalog-v1.json"))
        catalog = json.loads(catalog_path.read_text(encoding="utf-8"))
        if service_contract.get("catalog_sha256") != sha256(catalog_path):
            raise ValueError("bundled service catalog hash mismatch")
        catalog_capabilities = {
            operation["capability"]
            for operation in catalog.get("operations", [])
            if "mcp" in operation.get("transports", [])
        }
        tool_names = {tool.get("name") for tool in service_contract.get("tools", [])}
        catalog_tool_names = {
            operation["name"]
            for operation in catalog.get("operations", [])
            if "mcp" in operation.get("transports", [])
        }
        if tool_names != catalog_tool_names:
            raise ValueError("MCP tool snapshot differs from the bundled catalog")
        for module in modules:
            required = module.get("required_service_capabilities", [])
            if not isinstance(required, list) or not set(required).issubset(catalog_capabilities):
                raise ValueError(
                    f"service capability is not in the pinned catalog: {module.get('module_id')}"
                )
        for schema in service_contract.get("record_schemas", []):
            schema_path = module_path(str(contract_path.parent / schema["path"]))
            if schema.get("sha256") != sha256(schema_path):
                raise ValueError(f"bundled record schema hash mismatch: {schema.get('path')}")


def refresh(manifest: dict[str, Any]) -> None:
    modules = manifest["modules"]
    for module in modules:
        if not isinstance(module, dict) or not isinstance(module.get("entry"), str):
            raise ValueError("manifest module descriptor is invalid")
        module["sha256"] = sha256(module_path(module["entry"]))
        module["hash_encoding"] = "utf8-no-bom-lf"
    manifest.setdefault("snapshot", {})["digest"] = snapshot_digest(modules)
    for contract in manifest.get("verified_service_contracts", []):
        contract["sha256"] = sha256(module_path(contract["path"]))
    MANIFEST.write_text(json.dumps(manifest, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")


def main() -> None:
    if len(sys.argv) != 2 or sys.argv[1] not in {"refresh", "verify"}:
        raise SystemExit("usage: python scripts/manifest_tool.py refresh|verify")
    manifest = load_manifest()
    if sys.argv[1] == "refresh":
        refresh(manifest)
        manifest = load_manifest()
    verify(manifest)


if __name__ == "__main__":
    main()