Claude Hooks로 에이전트 실행 흐름에 결정론적 검증 레이어 심기

#!/bin/bash
# hooks/check-dangerous-command.sh
 
input=$(cat)
tool_name=$(echo "$input" | jq -r '.tool_name // ""')
command=$(echo "$input" | jq -r '.tool_input.command // ""')
 
if [ "$tool_name" = "Bash" ]; then
  if echo "$command" | grep -qE '(rm[[:space:]]+-rf|--force|DROP[[:space:]]+TABLE|truncate[[:space:]]+/)'; then
    # stderr로 메시지를 내보내면 모델이 거절 사유를 받습니다
    echo "위험 커맨드가 감지되어 실행을 차단했습니다." 1>&2
    exit 2
  fi
fi
 
exit 0

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Bash",
        "hooks": [
          {
            "type": "command",
            "command": "bash /path/to/hooks/check-dangerous-command.sh"
          }
        ]
      }
    ]
  }
}

# 개념적 예시 (수도코드) — 실제 SDK API 명세는 공식 문서 참조
import re
 
DANGEROUS_PATTERNS = [
    r'rm\s+-rf',
    r'git\s+push\s+.*--force',
    r'DROP\s+TABLE',
    r'truncate\s+/',
]
 
async def pre_tool_use_handler(input_data, tool_use_id, context):
    tool_name = input_data.get("tool_name")
    tool_input = input_data.get("tool_input", {})
 
    if tool_name == "Bash":
        command = tool_input.get("command", "")
        for pattern in DANGEROUS_PATTERNS:
            if re.search(pattern, command, re.IGNORECASE):
                # 차단과 함께 모델에 전달할 사유를 반환하는 형태(개념)
                return {
                    "decision": "block",
                    "reason": f"위험 패턴 '{pattern}' 실행이 차단되었습니다."
                }
    return {}

# 개념적 예시 (수도코드) — 도구 인수명과 반환 포맷은 공식 문서로 재확인
import re
 
SECRET_PATTERNS = {
    "AWS Access Key": r'AKIA[0-9A-Z]{16}',
    "Private Key": r'-----BEGIN (RSA |EC )?PRIVATE KEY-----',
    "Generic Token": r'["\']?(?:api|secret|token|key)["\']?\s*[:=]\s*["\'][A-Za-z0-9+/=]{20,}["\']',
}
 
async def scan_for_secrets(input_data, tool_use_id, context):
    tool_name = input_data.get("tool_name")
    tool_input = input_data.get("tool_input", {})
 
    if tool_name not in ("Write", "Edit", "MultiEdit"):
        return {}
 
    # Write/Edit 계열 도구의 실제 인수 키명(file_path / path / content / new_string 등)은
    # 사용 중인 Claude 버전의 도구 스펙에서 반드시 확인해야 합니다.
    content_candidates = [
        tool_input.get("content"),
        tool_input.get("new_string"),
    ]
    body = "\n".join(c for c in content_candidates if isinstance(c, str))
 
    for secret_type, pattern in SECRET_PATTERNS.items():
        if re.search(pattern, body):
            return {
                "decision": "block",
                "reason": f"시크릿 감지: {secret_type}. 환경 변수로 분리하세요."
            }
    return {}

# 개념적 예시 (수도코드) — 반환 키와 입력 구조는 공식 문서 기준으로 조정
import subprocess
import json
 
async def post_file_write_handler(input_data, tool_use_id, context):
    tool_name = input_data.get("tool_name")
    tool_input = input_data.get("tool_input", {})
 
    if tool_name not in ("Write", "Edit"):
        return {}
 
    # 도구 인수의 실제 키명은 사용하는 Claude 버전의 스펙에서 확인
    file_path = tool_input.get("file_path") or tool_input.get("path") or ""
    if not file_path.endswith((".js", ".ts", ".jsx", ".tsx")):
        return {}
 
    result = subprocess.run(
        ["npx", "eslint", "--format", "json", file_path],
        capture_output=True, text=True, timeout=30,
    )
 
    if result.returncode == 0:
        return {}
 
    try:
        lint_errors = json.loads(result.stdout)
    except json.JSONDecodeError:
        return {}
 
    lines = []
    for file_result in lint_errors:
        for msg in file_result.get("messages", []):
            lines.append(f"Line {msg.get('line')}: [{msg.get('ruleId')}] {msg.get('message')}")
 
    if not lines:
        return {}
 
    # 모델에 피드백을 전달하기 위한 반환 구조는 SDK마다 다릅니다.
    # 공식 문서에서 PostToolUse가 지원하는 피드백 필드명을 확인해 사용하세요.
    feedback = "ESLint 검사 결과:\n" + "\n".join(lines)
    return {
        "hookSpecificOutput": {
            "hookEventName": "PostToolUse",
            "additionalContext": feedback,
        }
    }

# 개념적 예시 (수도코드)
import json
import datetime
from pathlib import Path
 
audit_entries = []
 
async def on_pre_tool_use(input_data, tool_use_id, context):
    audit_entries.append({
        "timestamp": datetime.datetime.utcnow().isoformat(),
        "phase": "pre",
        "tool_name": input_data.get("tool_name"),
        "tool_input": input_data.get("tool_input"),
        "tool_use_id": tool_use_id,
    })
    return {}
 
async def on_post_tool_use(input_data, tool_use_id, context):
    audit_entries.append({
        "timestamp": datetime.datetime.utcnow().isoformat(),
        "phase": "post",
        "tool_name": input_data.get("tool_name"),
        "tool_response": input_data.get("tool_response"),
        "tool_use_id": tool_use_id,
    })
    return {}
 
async def on_stop(input_data, tool_use_id, context):
    log_path = Path(f"audit_logs/session_{datetime.date.today().isoformat()}.jsonl")
    log_path.parent.mkdir(parents=True, exist_ok=True)
    with open(log_path, "a", encoding="utf-8") as f:
        for entry in audit_entries:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    return {}