ISEP/hooks/lib/kbdb_cmd_check.py

#!/usr/bin/env python3
"""hooks/lib/kbdb_cmd_check.py -- shared helper for kbdb-api-wall-guard.sh's
Bash channel (inkstone/ISEP#40 S7: helpers live in lib/, do not count as a
hand-written gate).

What it guards: whether a bash command string contains a real, at-command-
position invocation of wrangler's D1 execute subcommand (any wrapper such as
npx/pnpm) targeting a database whose name mentions "kbdb" -- that is a
direct-SQL bypass of the KBDB API wall (D38, leo 2026-08-07: zero SQL,
always through the HTTP API).

Why this file exists (not just "why the rule exists"): the calling gate,
hooks/kbdb-api-wall-guard.sh, used to pipe the raw command straight through
a keyword grep. That produced the exact "keyword shows up vs. real
instruction" confusion this repo has hit repeatedly (inkstone/InkStoneCo#23:
the phrase showed up inside a delegation prompt that was *talking about* the
rule, not breaking it). The fix pattern already proven on
hooks/release-tag-guard.sh is "only count a keyword when it sits at command
position" -- this file is that same pattern for the KBDB Bash channel,
factored out to its own file because embedding shell-escaping-aware parsing
inline in the .sh caused regressions each time someone touched it (see the
.sh file's own header for that history).

Two extra layers vs. a plain regex:
  1. hooks/lib/strip_heredoc.py runs first (imported below) so a heredoc body
     that merely *mentions* the D1-execute pattern against kbdb as
     documentation text does not count -- same root cause as
     InkStoneCo#23/#56.
  2. Command segmentation uses shlex with punctuation_chars, so a `;`/`&`/`|`
     that appears *inside* a quoted string (e.g. a commit message) does not
     get treated as a command boundary, and text inside quotes is only
     inspected when the quoted text is itself an argument to a command that
     is genuinely at command position (e.g. a `--command` value passed to a
     real wrangler invocation) -- not when it is merely quoted prose
     describing the rule.

Genuine parse failures (unbalanced quotes, exotic constructs) fail OPEN
(print "OK"), matching this repo's stated design discipline: fail-open on
parse failure, not fail-open on the verdict itself (see release-tag-guard.sh
header). The caller (kbdb-api-wall-guard.sh) already blocks direct file-path
writes into kbdb/ paths and DDL/`.prepare(`/`.exec(`/`.batch(` in Write/Edit
content through its own separate channels -- this file only covers the Bash
CLI channel.

Usage:
    printf '%s' "$CMD" | python3 hooks/lib/kbdb_cmd_check.py
    -> prints exactly "BAD" or "OK" on stdout.
"""
import os
import shlex
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from strip_heredoc import strip_heredocs  # noqa: E402

_WRAPPERS = {"npx", "pnpm", "yarn", "bunx", "exec", "sudo", "env", "corepack"}
_DB_HINT = "kbdb"
_D1_TOKEN = "d1"
_SUBCMD_TOKEN = "execute"


def _segments(cmd: str):
    """Split cmd into a list of token-lists, one per "simple command",
    breaking on real shell control operators. Quote-aware via shlex, so a
    control character inside a quoted string does not split. Returns None
    on unparseable input (caller should fail open)."""
    try:
        lex = shlex.shlex(cmd, posix=True, punctuation_chars=True)
        lex.whitespace_split = True
        tokens = list(lex)
    except ValueError:
        return None

    boundary = {";", "&", "&&", "|", "||", "(", ")", "\n"}
    segments = []
    current = []
    for tok in tokens:
        if tok in boundary:
            if current:
                segments.append(current)
            current = []
        else:
            current.append(tok)
    if current:
        segments.append(current)
    return segments


def _is_kbdb_d1_execute(segment):
    """True if this one simple command is wrangler's D1 execute subcommand
    (optionally behind a wrapper like npx/sudo/env) aimed at a database
    whose name mentions "kbdb" anywhere in its arguments."""
    i = 0
    # Skip leading VAR=value assignments (env-style prefix).
    while i < len(segment):
        head = segment[i].split("=", 1)[0]
        if "=" in segment[i] and head.replace("_", "").isalnum() and head[:1].isalpha():
            i += 1
        else:
            break
    # Skip known wrappers (npx wrangler ..., sudo wrangler ..., env X=Y wrangler ...).
    while i < len(segment) and os.path.basename(segment[i]) in _WRAPPERS:
        i += 1
    if i >= len(segment):
        return False
    cmdname = os.path.basename(segment[i])
    if cmdname != "wrangler":
        return False
    rest = segment[i + 1 :]
    if _D1_TOKEN not in rest or _SUBCMD_TOKEN not in rest:
        return False
    joined_lower = " ".join(rest).lower()
    return _DB_HINT in joined_lower


def check(cmd: str) -> str:
    stripped = strip_heredocs(cmd)
    segments = _segments(stripped)
    if segments is None:
        return "OK"  # parse failure -> fail open, not fail on the verdict
    for seg in segments:
        if _is_kbdb_d1_execute(seg):
            return "BAD"
    return "OK"


def main() -> None:
    cmd = sys.stdin.read()
    print(check(cmd))


if __name__ == "__main__":
    main()