devplacepy/.claude/hooks/guard_production_db.py

#!/usr/bin/env python3
# retoor <retoor@molodetz.nl>

import json
import re
import sys
from pathlib import Path

CONFIRMATION_TOKEN = "I-HAVE-CONFIRMED-PRODUCTION-DB-ACCESS"

PRODUCTION_PATHS = re.compile(
    r"data/(devplace\.db|devii_tasks\.db|devii_lessons\.db|keys|uploads"
    r"|attachments|project_files|backups)\b"
)
PRODUCTION_DB_FILE = re.compile(r"\bdevplace\.db\b")
MANAGEMENT_CLI = re.compile(r"(?:^|[;&|(]\s*|\s)(?:[\w./-]*/)?devplace\s+(?!-)")
PYTHON_INVOCATION = re.compile(r"(?:^|[;&|(\s])(?:[\w./-]*/)?python[0-9.]*(?:\s|$)")
DATABASE_OVERRIDE = re.compile(r"DEVPLACE_DATABASE_URL\s*=\s*[\"']?(\S+?)[\"']?(?:\s|$)")
DATABASE_ASSIGNMENT = re.compile(r"DEVPLACE_DATABASE_URL[\"'\]\s]*[=,]")
MODULE_INVOCATION = re.compile(r"-m\s+devplacepy")
INLINE_CODE = re.compile(r"-c\s+(?P<quote>[\"'])(?P<code>.*?)(?P=quote)", re.DOTALL)
SCRIPT_PATH = re.compile(r"(?:^|\s)(?P<path>[\w./~-]+\.py)(?:\s|$)")
IMPORT_GATE = re.compile(
    r"^from devplacepy\.main import app\s*;?\s*(?:print\([^)]*\)\s*;?\s*)?$"
)
TEST_RUNNER = re.compile(r"\bpytest\b|\bmake\s+(test|test-[\w-]+)\b")
SERVER_TARGET = re.compile(r"\bmake\s+(dev|prod|docker-[\w-]+|ppy)\b")


APPLICATION_IMPORT = re.compile(
    r"(?:^|[\s;])(?:from|import)\s+devplacepy\b"
    r"|import_module\s*\(\s*[\"']devplacepy",
    re.MULTILINE,
)


def reaches_application(text: str) -> bool:
    return bool(APPLICATION_IMPORT.search(text))


def overrides_the_database(text: str) -> bool:
    match = DATABASE_OVERRIDE.search(text)
    if not match:
        return False
    return "data/devplace.db" not in match.group(1)


def source_targets_a_scratch_database(source: str) -> bool:
    if PRODUCTION_DB_FILE.search(source):
        return False
    return bool(DATABASE_ASSIGNMENT.search(source))


def script_is_safe(command: str) -> bool | None:
    match = SCRIPT_PATH.search(command)
    if not match:
        return None
    path = Path(match.group("path")).expanduser()
    try:
        body = path.read_text(errors="replace")
    except OSError:
        return None
    if not reaches_application(body):
        return True
    return source_targets_a_scratch_database(body)


def hazard_in(command: str) -> str:
    if CONFIRMATION_TOKEN in command:
        return ""
    if TEST_RUNNER.search(command) or SERVER_TARGET.search(command):
        return ""
    if PRODUCTION_DB_FILE.search(command) or PRODUCTION_PATHS.search(command):
        return "it names the production database or a production data directory"
    if MANAGEMENT_CLI.search(command):
        return "the devplace management CLI operates on the production database"
    if not PYTHON_INVOCATION.search(command):
        return ""
    if overrides_the_database(command):
        return ""
    if MODULE_INVOCATION.search(command):
        return "it runs a devplacepy module with no DEVPLACE_DATABASE_URL override"
    inline = INLINE_CODE.search(command)
    if inline:
        code = inline.group("code").strip()
        if not reaches_application(code):
            return ""
        if IMPORT_GATE.match(code):
            return ""
        return "it imports devplacepy inline with no DEVPLACE_DATABASE_URL override"
    safe = script_is_safe(command)
    if safe is None:
        return ""
    if safe:
        return ""
    return "the script imports devplacepy with no DEVPLACE_DATABASE_URL override"


def refuse(reason: str) -> dict:
    return {
        "hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": "deny",
            "permissionDecisionReason": (
                f"Blocked: this command reaches the production database because {reason}. "
                "The production database is never touched without the user's explicit, "
                "stated confirmation. Stop, tell the user exactly what the command would "
                "read or write, and ask them to confirm in their own words. Only after "
                f"they have done so may the command carry the literal token "
                f"{CONFIRMATION_TOKEN}, which still raises a permission prompt they must "
                "approve. Never add that token on your own initiative. Alternatives that "
                "need no confirmation: set DEVPLACE_DATABASE_URL to a scratch database, "
                "or run the test suite."
            ),
        }
    }


def confirm(reason: str) -> dict:
    return {
        "hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": "ask",
            "permissionDecisionReason": (
                "This command carries the production-database confirmation token and "
                f"reaches the production database because {reason}. Approve only if you "
                "asked for this."
            ),
        }
    }


def main() -> None:
    try:
        payload = json.load(sys.stdin)
    except (json.JSONDecodeError, ValueError):
        sys.exit(0)
    command = (payload.get("tool_input") or {}).get("command") or ""
    if not command:
        sys.exit(0)
    if CONFIRMATION_TOKEN in command:
        stripped = command.replace(CONFIRMATION_TOKEN, "")
        reason = hazard_in(stripped)
        if reason:
            print(json.dumps(confirm(reason)))
        sys.exit(0)
    reason = hazard_in(command)
    if reason:
        print(json.dumps(refuse(reason)))
    sys.exit(0)


if __name__ == "__main__":
    main()