claude-regelwerk/.vault-tools/shell-wache/faelle_shell_wache.py

"""Fährt die Angriffsliste gegen shell_wache.py, je Fall als eigener Prozess.

Ersteller: Pascal Michel
Erstellt:  2026-10-30
Entwurf:   ENTWURF.md im selben Ordner

Je Fall: Exit 0, stdout leer oder genau ein JSON-Objekt mit
hookSpecificOutput.hookEventName == "PreToolUse", ohne permissionDecision, die
Schlüssel in eckigen Klammern am Zeilenanfang von additionalContext sind genau
die erwarteten, Laufzeit unter 2 s. Der Hook läuft mit python -I -X utf8 und ohne
PYTHONIOENCODING in der Umgebung, wie Claude Code ihn startet.

Grundlinie zuerst: Ein gesunder Befehl muss still sein. Scheitert sie, bricht der
Treiber ab, weil der Rest nichts misst.

Aufruf:
    python -I -X utf8 .vault-tools/shell-wache/faelle_shell_wache.py [Hook-Pfad]
"""

from __future__ import annotations

import json
import os
import re
import subprocess
import sys
import time
from pathlib import Path

HIER = Path(__file__).resolve().parent
MAX_DAUER = 2.0
ZEITLIMIT = 10
KOPFZEILE = "shell-wache warnt nur, der Befehl läuft trotzdem."

# (Name, stdin als Bytes, erwartete Schlüssel[, Text, der im Hinweis stehen muss])
Fall = tuple


def eingabe(cmd: str) -> bytes:
    return json.dumps({"tool_name": "Bash", "tool_input": {"command": cmd}}).encode("utf-8")


def roh(daten: object) -> bytes:
    return json.dumps(daten).encode("utf-8")


def treffer(gruppe: str, key: str, befehle: list[str]) -> list[Fall]:
    return [(f"{gruppe} | Treffer | {b!r}", eingabe(b), {key}) for b in befehle]


def still(gruppe: str, befehle: list[str]) -> list[Fall]:
    return [(f"{gruppe} | still | {b!r}", eingabe(b), set()) for b in befehle]


GRUNDLINIE: list[Fall] = [
    ("GRUNDLINIE | gesunder Befehl bleibt still", eingabe("ls -la && python -I -X utf8 x.py"), set()),
    ("GRUNDLINIE | python3 wird gemeldet", eingabe("python3 x.py"), {"python3"}),
]

PY_I_TREFFER = [
    "python -I x.py",
    "python -IB x.py",
    "python -BI x.py",
    'python -Ic "print(1)"',
    'python -I -c "x" -X utf8',
    "python -I -X utf8=0 x.py",
    "/c/Py/python.exe -I x.py",
    '"C:/Py/python.exe" -I x.py',
    "cd a && python -I x.py",
    "PYTHONIOENCODING=utf-8 python -I x.py",
]
PY_I_STILL = [
    "python -I -X utf8 x.py",
    "python -IXutf8 x.py",
    "python -IX utf8 x.py",
    "python -X utf8 -I -B x.py",
    "python -I -X utf8=1 x.py",
    "python x.py -I",
    'python -c "import x" -I',
    "ipython -I",
    "python -m pytest -I",
]
PY3_TREFFER = ["python3 x.py", "python3.12 -m venv v", "ls; python3 -V", "python3.exe x.py"]
PY3_STILL = [
    "python x.py",
    "python3-config --libs",
    "wsl -d Debian -- python3 x.py",
    "ssh h python3 x.py",
    "docker exec c python3 x.py",
]
BASH_TREFFER = [
    'cmd //c "bash x.sh"',
    'cmd.exe /c bash -c "x"',
    "CMD /C BASH x",
    "powershell -c \"bash -lc 'x'\"",
    'powershell.exe -NoProfile -Command "& bash x"',
    'pwsh -c "bash x"',
    "python -X utf8 -c \"import subprocess; subprocess.run(['bash', '-c', 'x'])\"",
    "python x.py <<'EOF'\nsubprocess.run([\"bash.exe\", \"x\"])\nEOF",
]
BASH_STILL = [
    'bash -c "x"',
    "bash x.sh",
    'cmd //c "C:/Program Files/Git/bin/bash.exe" x',
    'cmd //c "C:\\Program Files\\Git\\bin\\bash.exe" x',
    "powershell -c \"wsl -d Debian -- bash -lc 'x'\"",
    'cmd /c "ssh.exe -o A=B root@h bash -s < f.sh"',
    'powershell -c "Get-Content ~/.bashrc"',
    'powershell -c "x"\nbash y.sh',
    "ssh h python - <<'EOF'\nsubprocess.run(['bash','x'])\nEOF",
    "python -c \"subprocess.run([GITBASH, '-c'])\"",
]
PKILL_TREFFER = [
    "pkill -f x",
    'pkill -TERM -f "[s]leep 1"',
    "pkill -fx x",
    "pkill --full x",
    "pkill -u root -f x",
    "sudo pkill -f x",
    "wsl -d Debian -- bash -c 'a; pkill -f \"y\"'",
    "/usr/bin/pkill -f x",
]
PKILL_STILL = ["pkill x", "pkill -TERM x", "pgrep -f x", "pkill -u fx x"]

FAELLE: list[Fall] = []
FAELLE += treffer("python-I", "python-I", PY_I_TREFFER) + still("python-I", PY_I_STILL)
FAELLE += treffer("python3", "python3", PY3_TREFFER) + still("python3", PY3_STILL)
FAELLE += treffer("bash-windows", "bash-windows", BASH_TREFFER) + still("bash-windows", BASH_STILL)
FAELLE += treffer("pkill-f", "pkill-f", PKILL_TREFFER) + still("pkill-f", PKILL_STILL)

# Mehrere Muster in einem Befehl
FAELLE += [
    (
        "mehrere | alle vier Muster",
        eingabe("python -I x.py; python3 y.py; cmd /c bash z; pkill -f w"),
        {"python-I", "python3", "bash-windows", "pkill-f"},
    ),
    ("mehrere | dasselbe Muster zweimal", eingabe("python -I a.py; python -I b.py"), {"python-I"}),
]

# Reihenfolge: je Muster am Anfang, am Ende, in Zeile 3
for _key, _treffer in (
    ("python-I", "python -I x.py"),
    ("python3", "python3 x.py"),
    ("bash-windows", "cmd /c bash x"),
    ("pkill-f", "pkill -f x"),
):
    FAELLE += [
        (f"reihenfolge | {_key} am Anfang", eingabe(f"{_treffer}; ls"), {_key}),
        (f"reihenfolge | {_key} am Ende", eingabe(f"ls; {_treffer}"), {_key}),
        (f"reihenfolge | {_key} in Zeile 3", eingabe(f"ls\npwd\n{_treffer}"), {_key}),
    ]

# Eingabe
FAELLE += [
    ("eingabe | stdin leer", b"", {"eingabe"}, "stdin leer"),
    ("eingabe | kein JSON", b"nicht json", {"eingabe"}, "kein JSON)"),
    ("eingabe | Liste", b"[]", {"eingabe"}, "kein JSON-Objekt"),
    ("eingabe | null", b"null", {"eingabe"}, "kein JSON-Objekt"),
    ("eingabe | leeres Objekt", b"{}", {"eingabe"}, "tool_input fehlt oder ist kein Objekt"),
    ("eingabe | tool_input fehlt", roh({"tool_name": "Bash"}), {"eingabe"},
     "tool_input fehlt oder ist kein Objekt"),
    ("eingabe | tool_input null", roh({"tool_name": "Bash", "tool_input": None}), {"eingabe"},
     "tool_input fehlt oder ist kein Objekt"),
    ("eingabe | tool_input Zeichenkette", roh({"tool_name": "Bash", "tool_input": "x"}), {"eingabe"},
     "tool_input fehlt oder ist kein Objekt"),
    ("eingabe | command fehlt", roh({"tool_name": "Bash", "tool_input": {}}), {"eingabe"},
     "command fehlt oder ist keine Zeichenkette"),
    ("eingabe | command null", roh({"tool_name": "Bash", "tool_input": {"command": None}}), {"eingabe"},
     "command fehlt oder ist keine Zeichenkette"),
    ("eingabe | command Zahl", roh({"tool_name": "Bash", "tool_input": {"command": 5}}), {"eingabe"},
     "command fehlt oder ist keine Zeichenkette"),
    ("eingabe | command Liste", roh({"tool_name": "Bash", "tool_input": {"command": ["ls"]}}), {"eingabe"},
     "command fehlt oder ist keine Zeichenkette"),
    ("eingabe | ungueltiges UTF-8 im Befehl",
     b'{"tool_name":"Bash","tool_input":{"command":"ls \xff"}}', set()),
    ("eingabe | tool_name PowerShell mit python3",
     roh({"tool_name": "PowerShell", "tool_input": {"command": "python3 x"}}), {"eingabe"},
     "tool_name ist PowerShell, nicht Bash"),
    ("eingabe | command leer", eingabe(""), set()),
    ("eingabe | command nur Leerraum", eingabe("   "), set()),
    ("eingabe | zusaetzliche Felder",
     roh({"tool_name": "Bash", "tool_use_id": "t1", "session_id": "s",
          "tool_input": {"command": "ls", "description": "d", "timeout": 5}}), set()),
]


def schluessel_von(text: str) -> set[str]:
    return set(re.findall(r"^\[([^\]]+)\]", text, flags=re.M))


def hat_schluessel(objekt: object, name: str) -> bool:
    if isinstance(objekt, dict):
        return name in objekt or any(hat_schluessel(v, name) for v in objekt.values())
    if isinstance(objekt, list):
        return any(hat_schluessel(v, name) for v in objekt)
    return False


def fahre(hook: Path, stdin: bytes) -> tuple[int, str, str, float]:
    umgebung = {k: v for k, v in os.environ.items() if k != "PYTHONIOENCODING"}
    start = time.perf_counter()
    lauf = subprocess.run(
        [sys.executable, "-I", "-X", "utf8", str(hook)],
        input=stdin,
        capture_output=True,
        timeout=ZEITLIMIT,
        env=umgebung,
    )
    dauer = time.perf_counter() - start
    try:
        out = lauf.stdout.decode("utf-8")
    except UnicodeDecodeError as exc:
        out = "UNGUELTIGES-UTF-8: %s" % exc
    return lauf.returncode, out, lauf.stderr.decode("utf-8", errors="replace"), dauer


def pruefe(hook: Path, fall: Fall) -> list[str]:
    _, stdin, erwartet = fall[:3]
    muss_enthalten = fall[3] if len(fall) > 3 else None
    try:
        code, out, err, dauer = fahre(hook, stdin)
    except subprocess.TimeoutExpired:
        return [f"Zeitlimit {ZEITLIMIT} s ueberschritten"]
    probleme: list[str] = []
    if code != 0:
        probleme.append(f"Exit {code} statt 0")
    if dauer >= MAX_DAUER:
        probleme.append(f"Laufzeit {dauer:.2f} s, Grenze {MAX_DAUER:.0f} s")
    if err.strip():
        probleme.append(f"stderr nicht leer: {err.strip()[:200]!r}")
    if not erwartet:
        if out != "":
            probleme.append(f"stdout nicht leer: {out[:300]!r}")
        return probleme
    try:
        daten = json.loads(out)
    except ValueError:
        return probleme + [f"stdout ist kein JSON: {out[:300]!r}"]
    if not isinstance(daten, dict) or set(daten) != {"hookSpecificOutput"}:
        return probleme + [f"Objekt hat nicht genau den Schluessel hookSpecificOutput: {out[:300]!r}"]
    if hat_schluessel(daten, "permissionDecision"):
        probleme.append("permissionDecision kommt vor")
    innen = daten["hookSpecificOutput"]
    if not isinstance(innen, dict) or innen.get("hookEventName") != "PreToolUse":
        return probleme + ["hookEventName ist nicht PreToolUse"]
    text = innen.get("additionalContext")
    if not isinstance(text, str) or not text:
        return probleme + ["additionalContext fehlt oder ist leer"]
    if not text.startswith(KOPFZEILE + "\n"):
        probleme.append(f"Kopfzeile fehlt: {text[:80]!r}")
    gefunden = schluessel_von(text)
    if gefunden != erwartet:
        probleme.append(f"Schluessel {sorted(gefunden)} statt {sorted(erwartet)}")
    if muss_enthalten and muss_enthalten not in text:
        probleme.append(f"Hinweis ohne {muss_enthalten!r}: {text!r}")
    return probleme


def main() -> int:
    argumente = [a for a in sys.argv[1:] if not a.startswith("--")]
    hook = Path(argumente[0]).resolve() if argumente else HIER / "shell_wache.py"
    if not hook.is_file():
        print(f"ABBRUCH: {hook} fehlt")
        return 1
    print(f"Hook: {hook}")

    for fall in GRUNDLINIE:
        try:
            probleme = pruefe(hook, fall)
        except Exception as exc:
            probleme = [f"ABSTURZ im Treiber: {type(exc).__name__}: {exc}"]
        print(f"{'OK' if not probleme else 'FEHLER':7} {fall[0]}" + "".join(f"\n        {p}" for p in probleme))
        if probleme:
            print("ABBRUCH: Die Grundlinie ist nicht gruen. Der Rest misst nichts.")
            return 1

    ergebnisse: list[bool] = []
    for fall in FAELLE:
        try:
            probleme = pruefe(hook, fall)
        except Exception as exc:
            probleme = [f"ABSTURZ im Treiber: {type(exc).__name__}: {exc}"]
        ergebnisse.append(not probleme)
        print(f"{'OK' if not probleme else 'FEHLER':7} {fall[0]}" + "".join(f"\n        {p}" for p in probleme))

    fehler = ergebnisse.count(False)
    print()
    print(f"{len(ergebnisse)} Faelle, {fehler} Fehler (zusaetzlich {len(GRUNDLINIE)} Grundlinie)")
    return 0 if fehler == 0 else 1


if __name__ == "__main__":
    try:
        sys.stdout.reconfigure(encoding="utf-8")
    except AttributeError:
        pass
    try:
        code = main()
    except Exception as exc:
        print(f"FEHLER Treiber abgestuerzt: {type(exc).__name__}: {exc}")
        code = 1
    raise SystemExit(code)