Destructive kubectl Guard

Event

PreToolUse

Trigger

tool-name — Bash

Language

python

Side effects

• blocks tool call with explanation
• bypass via AI_ALLOW_DESTRUCTIVE_KUBECTL=1

Platforms

linuxmacoswindows

Notes

Logic is cross-platform. Wiring: use 'ai hooks run destructive-kubectl-guard' in settings.json — the ai binary discovers Python on each OS. kubectl available on all platforms. Python logic is cross-platform.

Depends on

hook/lib

Script · destructive-kubectl-guard.py

#!/usr/bin/env python3
"""hooks/destructive-kubectl-guard.py — gate destructive `kubectl`
operations per Common.md §2.2. Opt-in via command-wrappers.toml.

Blocks (without bypass env): `kubectl delete`, `kubectl drain`,
`kubectl cordon`. Other subcommands pass through.

The bypass env is AI_ALLOW_DESTRUCTIVE_KUBECTL=1.

Self-check:
  --self-check
"""
from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
import _lib  # noqa: E402


GUARDED = {"delete", "drain", "cordon"}
BYPASS_ENV = "AI_ALLOW_DESTRUCTIVE_KUBECTL"
TOOL = "kubectl"


def argv_from_wrapper() -> list[str]:
    """The args after `kubectl`, as JSON in WRAPPED_ARGV (command-wrapper path).

    The wrapper does not forward the tool's argv on the command line — it
    publishes it in WRAPPED_ARGV so argparse never sees subcommand tokens.
    """
    try:
        return json.loads(os.environ.get("WRAPPED_ARGV", "[]"))
    except json.JSONDecodeError:
        return []


def argv_from_stdin() -> list[str]:
    """The args after `kubectl`, parsed from a Claude PreToolUse JSON payload."""
    raw = sys.stdin.read()
    if not raw.strip():
        return []
    try:
        payload = json.loads(raw)
    except json.JSONDecodeError:
        return []
    cmd = (
        payload.get("command")
        or payload.get("tool_input", {}).get("command")
        or ""
    ) if isinstance(payload, dict) else ""
    parts = cmd.split()
    return parts[1:] if parts and parts[0] == TOOL else []


def check_invocation(argv: list[str]) -> int:
    if not argv or argv[0] not in GUARDED:
        return 0
    if os.environ.get(BYPASS_ENV) == "1":
        _lib.log(f"`kubectl {argv[0]}` — bypass active ({BYPASS_ENV}=1). Logged.")
        return 0
    _lib.log(f"blocking — `kubectl {argv[0]}` mutates cluster state.")
    _lib.log("Per Common.md §2.2 + §2.4: state what will change, name reversibility, wait for an unambiguous yes.")
    _lib.log(f"To bypass for one session only: {BYPASS_ENV}=1 kubectl {argv[0]} ...")
    return 1


def main(argv: list[str]) -> int:
    parser = argparse.ArgumentParser(add_help=True)
    parser.add_argument("--self-check", action="store_true")
    parser.add_argument("--mode", choices=["claude", "wrapper"], default=None,
                        help="invocation mode (set by the command wrapper)")
    parser.add_argument("rest", nargs=argparse.REMAINDER)
    args = parser.parse_args(argv)
    if args.self_check:
        return _lib.self_check_ok()
    if args.mode == "wrapper":
        return check_invocation(argv_from_wrapper())
    if args.mode == "claude" or not sys.stdin.isatty():
        return check_invocation(argv_from_stdin())
    return check_invocation(args.rest)


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))