atlas-iac/services/hermes/scripts/claude_command_policy.py

59 lines
2.2 KiB
Python

#!/usr/bin/env python3
"""Block known-dangerous shell mutations in unattended Claude workers."""
from __future__ import annotations
import json
import re
import sys
from typing import Any
DENIED_COMMANDS: tuple[tuple[str, re.Pattern[str]], ...] = (
("Kubernetes mutations are delivered through Flux", re.compile(r"(?:^|[;&|]\s*)kubectl\s+(?:apply|delete|patch|scale|exec|port-forward)\b", re.I)),
("Flux control operations require an explicit operator action", re.compile(r"(?:^|[;&|]\s*)flux\s+(?:reconcile|suspend|resume)\b", re.I)),
("Vault data access is outside the worker boundary", re.compile(r"(?:^|[;&|]\s*)vault\s+kv\b", re.I)),
("force-push is prohibited", re.compile(r"(?:^|[;&|]\s*)git\s+push\b[^\n]*(?:--force(?:-with-lease)?|(?:^|\s)-f(?:\s|$))", re.I)),
("destructive Git reset is prohibited", re.compile(r"(?:^|[;&|]\s*)git\s+reset\b[^\n]*--hard\b", re.I)),
("destructive Git clean is prohibited", re.compile(r"(?:^|[;&|]\s*)git\s+clean\b[^\n]*(?:-[a-z]*f[a-z]*|--force)\b", re.I)),
)
def command_from_payload(payload: dict[str, Any]) -> str:
"""Extract the Bash command from a Claude PreToolUse hook payload."""
tool_input = payload.get("tool_input")
if not isinstance(tool_input, dict):
return ""
value = tool_input.get("command")
return value if isinstance(value, str) else ""
def denial_reason(command: str) -> str | None:
"""Return a concise block reason for a denied command."""
for reason, pattern in DENIED_COMMANDS:
if pattern.search(command):
return reason
return None
def main() -> int:
"""Apply the policy to one Claude hook event."""
try:
payload = json.load(sys.stdin)
except (json.JSONDecodeError, OSError):
print("Claude command policy received invalid JSON", file=sys.stderr)
return 2
if not isinstance(payload, dict):
print("Claude command policy received an invalid event", file=sys.stderr)
return 2
command = command_from_payload(payload)
reason = denial_reason(command)
if reason:
print(f"Blocked by the Hermes owner-workspace policy: {reason}.", file=sys.stderr)
return 2
return 0
if __name__ == "__main__":
raise SystemExit(main())