tools/dev/sync-hook-managers.py

#!/usr/bin/env python3
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
"""Validate the hook-manager registry and write it into its implementations.

`tools/hook-manager/registry.json` is the source of truth for the git-hook
managers the secure agent setup integrates (the contract is
`tools/hook-manager/README.md`). The scripts that implement the contract run
where the registry is not available (a copy under `~/.claude/bin`, a hook
under `~/.claude/git-hooks`, a git hook that must start fast), so each carries
a generated block between `BEGIN generated by tools/dev/sync-hook-managers.py`
and `END ...` markers:

- `tools/agent-isolation/hook-manager-shim.sh`: the install rewrite per tool.
- `tools/agent-isolation/git-hook-dispatcher.sh`: the host-only home and
  run-time override per tool.
- `tools/sandbox-lint/src/sandbox_lint/__init__.py`: the homes the lint checks.

It also checks, without writing, that the committed settings and the
sandbox-lint baseline deny every manager's default home and give the sandbox
home of every manager this repository uses.

Run as a prek hook. Rewrites the blocks in place and exits 1 if one changed
(re-stage and commit again); `--check` reports drift without writing.
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path
from typing import Any

REPO_ROOT = Path(__file__).resolve().parents[2]
REGISTRY = Path("tools/hook-manager/registry.json")
SHIM = Path("tools/agent-isolation/hook-manager-shim.sh")
DISPATCHER = Path("tools/agent-isolation/git-hook-dispatcher.sh")
SANDBOX_LINT = Path("tools/sandbox-lint/src/sandbox_lint/__init__.py")
SETTINGS = (Path(".claude/settings.json"), Path("tools/sandbox-lint/expected.json"))

#: Install strategies the shim implements. A new one lands with its shim branch.
STRATEGIES = {"git-dir-flag", "hooks-path-env", "hide-global-config", "none"}

#: How the shim recognises a positional install action: `exact` (argv[0] is the
#: action) or `prefix` (a non-option argument prefixes it, as a Symfony Console
#: application resolves an unambiguous abbreviation after global options).
INSTALL_MATCHES = {"exact", "prefix"}

_NAME = re.compile(r"^[a-z][a-z0-9-]*$")
_SUBCOMMAND = re.compile(r"^(?:--)?[a-z][a-z0-9-]*$")
_ALIAS = re.compile(r"^--?[a-z][a-z0-9-]*$")
_ENV = re.compile(r"^[A-Z][A-Z0-9_]*$")
_FLAG = re.compile(r"^--[a-z][a-z0-9-]*$")
_DIR = re.compile(r"^[a-z][a-z0-9.-]*$")
_FILE = re.compile(r"^[A-Za-z0-9._-]+$")
_MARKER = re.compile(r"^[A-Za-z0-9._-]+$")
_BLOCK = re.compile(
    r"(?P<begin>^[ \t]*# BEGIN generated by tools/dev/sync-hook-managers\.py; do not edit\n)"
    r".*?"
    r"(?P<end>^[ \t]*# END generated by tools/dev/sync-hook-managers\.py$)",
    re.MULTILINE | re.DOTALL,
)


def validate(registry: dict[str, Any]) -> list[str]:
    """Every rule of the contract's registry section, as error strings."""
    errors: list[str] = []
    adapters = registry.get("adapters")
    if not isinstance(adapters, dict) or not adapters:
        return ["adapters: missing or empty"]
    for name, adapter in adapters.items():
        where = f"adapters.{name}"
        if not _NAME.match(name):
            errors.append(f"{where}: the name must be the binary's name ([a-z][a-z0-9-]*)")
        if not isinstance(adapter, dict):
            errors.append(f"{where}: must be an object")
            continue
        for key in ("language", "upstream"):
            if not isinstance(adapter.get(key), str) or not adapter[key]:
                errors.append(f"{where}.{key}: missing")
        detect = adapter.get("detect")
        if not isinstance(detect, dict):
            errors.append(f"{where}.detect: missing")
        else:
            files = detect.get("config_files")
            if not isinstance(files, list) or not all(isinstance(f, str) and _FILE.match(f) for f in files):
                errors.append(f"{where}.detect.config_files: a list of file names")
            elif not files and adapter.get("home"):
                errors.append(
                    f"{where}.detect.config_files: a manager with a home must have root config_files"
                )
            if not isinstance(detect.get("nested"), bool):
                errors.append(f"{where}.detect.nested: whether a config below the root counts")
            if not isinstance(detect.get("hook_marker"), str) or not _MARKER.match(detect["hook_marker"]):
                errors.append(f"{where}.detect.hook_marker: the text an installed hook that runs it contains")
        install = adapter.get("install")
        if not isinstance(install, dict):
            errors.append(f"{where}.install: missing")
        else:
            strategy = install.get("strategy")
            if strategy not in STRATEGIES:
                errors.append(
                    f"{where}.install.strategy: {strategy!r} is not one the shim implements "
                    f"({', '.join(sorted(STRATEGIES))})"
                )
            if strategy != "none":
                subcommand = install.get("subcommand")
                if not isinstance(subcommand, str) or not _SUBCOMMAND.match(subcommand):
                    errors.append(f"{where}.install.subcommand: missing or not a word or --flag")
                aliases = install.get("aliases")
                if aliases is not None:
                    if (
                        not isinstance(aliases, list)
                        or not aliases
                        or not all(isinstance(alias, str) and _ALIAS.match(alias) for alias in aliases)
                    ):
                        errors.append(f"{where}.install.aliases: a non-empty list of flags")
                    elif not isinstance(subcommand, str) or not subcommand.startswith("-"):
                        errors.append(f"{where}.install.aliases: only a flag-style subcommand takes aliases")
                    elif subcommand in aliases:
                        errors.append(f"{where}.install.aliases: must not repeat the subcommand")
                if strategy == "git-dir-flag" and not (
                    isinstance(install.get("flag"), str) and _FLAG.match(install["flag"])
                ):
                    errors.append(
                        f"{where}.install.flag: git-dir-flag needs the long option that takes the git dir"
                    )
                elif strategy != "git-dir-flag" and "flag" in install:
                    errors.append(f"{where}.install.flag: strategy {strategy!r} takes no flag")
                match = install.get("match", "exact")
                if match not in INSTALL_MATCHES:
                    errors.append(
                        f"{where}.install.match: {match!r} is not how the shim finds the install "
                        f"action ({', '.join(sorted(INSTALL_MATCHES))})"
                    )
                elif match == "prefix" and not (
                    isinstance(subcommand, str) and not subcommand.startswith("-")
                ):
                    errors.append(f"{where}.install.match: prefix matching needs a positional subcommand")
                if "vendor_bin" in install and not isinstance(install["vendor_bin"], bool):
                    errors.append(f"{where}.install.vendor_bin: must be true or false")
            else:
                for key in ("flag", "aliases", "match", "vendor_bin"):
                    if key in install:
                        errors.append(f"{where}.install.{key}: strategy 'none' takes no {key}")
            if "runtime_env" in install and not isinstance(install["runtime_env"], bool):
                errors.append(f"{where}.install.runtime_env: must be true or false")
            if install.get("runtime_env") and strategy != "hooks-path-env":
                errors.append(
                    f"{where}.install.runtime_env: only hooks-path-env needs the override at run time"
                )
        home = adapter.get("home")
        if home is None:
            continue  # a manager that keeps nothing outside the repository
        if not isinstance(home, dict):
            errors.append(f"{where}.home: must be an object")
            continue
        if not isinstance(home.get("env"), str) or not _ENV.match(home["env"]):
            errors.append(f"{where}.home.env: the environment variable that relocates the home")
        if home.get("expands_tilde") is not True:
            # The committed settings can only spell a per-user path with `~`.
            errors.append(
                f"{where}.home.expands_tilde: a tool that does not expand ~ needs a per-user sandbox home"
            )
        default, sandbox, host = home.get("default"), home.get("sandbox"), home.get("host")
        for key, value in (("default", default), ("sandbox", sandbox)):
            if not isinstance(value, str) or not value.startswith("~/.cache/"):
                errors.append(f"{where}.home.{key}: must sit under ~/.cache/")
        if (
            isinstance(default, str)
            and isinstance(sandbox, str)
            and (sandbox == default or sandbox.startswith(default + "/") or default.startswith(sandbox + "/"))
        ):
            errors.append(f"{where}.home.sandbox: must not overlap the default home {default}")
        if not isinstance(host, str) or not host.startswith("~/.local/share/") or not _DIR.match(host[15:]):
            errors.append(
                f"{where}.home.host: a single directory under ~/.local/share/, which no sandbox grant covers"
            )
    return errors


def render(registry: dict[str, Any]) -> dict[Path, str]:
    """The body of each generated block."""
    adapters = registry["adapters"]
    cases = []
    for name, adapter in sorted(adapters.items()):
        install = adapter["install"]
        if install.get("strategy") == "none":
            continue
        fields = [f"install_subcommand={install['subcommand']}", f"install_strategy={install['strategy']}"]
        if "aliases" in install:
            actions = " ".join([install["subcommand"], *install["aliases"]])
            fields.append(f"install_aliases='{actions}'")
        if "flag" in install:
            fields.append(f"install_flag={install['flag']}")
        if install.get("match") == "prefix":
            fields.append("install_match=prefix")
        if install.get("vendor_bin"):
            fields.append("install_vendor_bin=1")
        cases.append(f"    {name}) {'; '.join(fields)} ;;\n")
    shim = 'adapter() {\n  case "$1" in\n' + "".join(cases) + "    *) return 1 ;;\n  esac\n}\n"
    homes = [(n, a) for n, a in sorted(adapters.items()) if a.get("home")]
    runtime = [
        f"{n}:{a['detect']['hook_marker']}"
        for n, a in sorted(adapters.items())
        if a["install"].get("runtime_env")
    ]
    dispatcher = (
        'hook_manager_homes="'
        + " ".join(f"{a['home']['env']}={a['home']['host'][15:]}" for _, a in homes)
        + '"\n'
        + 'hook_manager_runtime_envs="'
        + " ".join(runtime)
        + '"\n'
    )
    lint = "HOOK_MANAGER_HOMES = (\n"
    for name, a in homes:
        home, detect = a["home"], a["detect"]
        files = ", ".join(json.dumps(f) for f in detect["config_files"])
        files = f"({files},)" if len(detect["config_files"]) == 1 else f"({files})"
        fields = [
            json.dumps(name),
            json.dumps(home["env"]),
            json.dumps(home["default"]),
            json.dumps(home["sandbox"]),
        ]
        fields += [files, str(detect["nested"]), json.dumps(detect["hook_marker"])]
        lint += "    (\n" + "".join(f"        {f},\n" for f in fields) + "    ),\n"
    lint += ")\n"
    return {SHIM: shim, DISPATCHER: dispatcher, SANDBOX_LINT: lint}


def check_settings(registry: dict[str, Any], root: Path) -> list[str]:
    """The committed settings give each sandbox home and deny each default home.

    The deny for every manager (the default home is per-user); the sandbox
    home only for the managers this repository uses (a config file at its
    root), as sandbox-lint requires it only there.
    """
    errors = []
    for rel in SETTINGS:
        settings = json.loads((root / rel).read_text(encoding="utf-8"))
        env = settings.get("env") or {}
        deny = (settings.get("permissions") or {}).get("deny") or []
        for name, adapter in sorted(registry["adapters"].items()):
            home = adapter.get("home")
            if not home:
                continue
            used = any((root / f).is_file() for f in adapter["detect"]["config_files"])
            if used and env.get(home["env"]) != home["sandbox"]:
                errors.append(f"{rel}: env.{home['env']} must be {home['sandbox']!r} ({name})")
            if f"Edit({home['default']}/**)" not in deny:
                errors.append(f"{rel}: permissions.deny must carry Edit({home['default']}/**) ({name})")
    return errors


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--check", action="store_true", help="report drift without writing")
    parser.add_argument("--root", type=Path, default=REPO_ROOT, help=argparse.SUPPRESS)
    args = parser.parse_args(argv)
    root: Path = args.root

    registry = json.loads((root / REGISTRY).read_text(encoding="utf-8"))
    errors = validate(registry)
    if errors:
        for error in errors:
            print(f"{REGISTRY}: {error}", file=sys.stderr)
        return 1

    status = 0
    for rel, body in render(registry).items():
        path = root / rel
        text = path.read_text(encoding="utf-8")
        blocks = list(_BLOCK.finditer(text))
        if len(blocks) != 1:
            print(f"{rel}: expected exactly one generated block, found {len(blocks)}", file=sys.stderr)
            status = 1
            continue
        block = blocks[0]
        new = text[: block.end("begin")] + body + text[block.start("end") :]
        if new != text:
            status = 1
            if args.check:
                print(f"{rel}: generated block is out of date; run {Path(__file__).name}", file=sys.stderr)
            else:
                path.write_text(new, encoding="utf-8")
                print(f"{rel}: regenerated from {REGISTRY}", file=sys.stderr)

    for error in check_settings(registry, root):
        print(error, file=sys.stderr)
        status = 1
    return status


if __name__ == "__main__":
    sys.exit(main())