【2026 最新】Claude Code 高風險命令閘門：sudo 前先產生 Blast-Radius Receipt

#!/usr/bin/env python3
"""Conservative Claude Code PreToolUse gate for Bash.

The gate performs static, lexical inspection only. It is defense in depth, not
a shell sandbox. It never executes the proposed command and never reproduces
the raw command in its blast-radius receipt.
"""

from __future__ import annotations

import hashlib
import json
import os
import re
import shlex
import sys
import unicodedata
from dataclasses import dataclass, field
from typing import Iterable

MAX_STDIN_BYTES = 1_048_576
MAX_COMMAND_BYTES = 65_536
KNOWN_PERMISSION_MODES = {
    "default",
    "acceptEdits",
    "plan",
    "auto",
    "dontAsk",
    "bypassPermissions",
}

RANK = {"allow": 0, "ask": 1, "deny": 2}
CONTROL = {";", "&&", "||", "|", "|&", "&", "\n", "(", ")", "{", "}"}
PIPE_OPS = {"|", "|&"}
SHELLS = {"sh", "bash", "zsh", "dash", "ksh"}
ASSIGNMENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=")

SYSTEM_WIDE_TOOLS = {
    "systemctl",
    "service",
    "launchctl",
    "shutdown",
    "reboot",
    "halt",
    "poweroff",
    "sysctl",
    "killall",
    "pkill",
    "apt",
    "apt-get",
    "dnf",
    "yum",
    "pacman",
    "brew",
}

DISK_TOOLS = {
    "diskutil",
    "fdisk",
    "sfdisk",
    "parted",
    "wipefs",
    "mount",
    "umount",
    "losetup",
    "zpool",
    "zfs",
}

TIME_TOOLS = {
    "timedatectl",
    "systemsetup",
    "hwclock",
    "ntpdate",
    "sntp",
}

NETWORK_TOOLS = {
    "curl",
    "wget",
    "nc",
    "netcat",
    "ssh",
    "scp",
    "sftp",
    "rsync",
    "ftp",
    "telnet",
    "socat",
    "nmap",
    "iptables",
    "nft",
    "pfctl",
    "route",
    "ifconfig",
    "networksetup",
    "kubectl",
    "terraform",
}

CREDENTIAL_TOOLS = {
    "security",
    "ssh-add",
    "gpg",
    "pass",
    "op",
    "vault",
    "keyctl",
    "secret-tool",
    "aws-vault",
}

SENSITIVE_ARGUMENT = re.compile(
    r"(?:^|[/._-])(?:\.env(?:\.|$)|id_(?:rsa|dsa|ecdsa|ed25519)(?:\.|$)|"
    r"credentials(?:\.|$)|secrets?(?:\.|$)|keychain(?:\.|$))|"
    r"(?:password|passwd|token|api[_-]?key|private[_-]?key)",
    re.IGNORECASE,
)

LEXICALLY_SIMPLE_COMMANDS = {
    "pwd",
    "true",
    "false",
    "printf",
    "echo",
}

EXPLICIT_HIGH_IMPACT_SIGNALS = {
    "RECURSIVE_DELETE_CRITICAL_SCOPE",
    "UNBOUNDED_XARGS_DELETE",
    "DESTRUCTIVE_FILE_OPERATION",
    "MUTATING_FIND",
    "FILESYSTEM_FORMATTER",
    "RAW_DEVICE_MUTATION",
    "RAW_DEVICE_WRITE",
    "DD_WRITE",
    "PERMISSION_OR_OWNER_CHANGE",
    "GIT_CLEAN",
}

INVISIBLE = {
    "\u061c",
    "\u200b",
    "\u200c",
    "\u200d",
    "\u200e",
    "\u200f",
    "\u202a",
    "\u202b",
    "\u202c",
    "\u202d",
    "\u202e",
    "\u2066",
    "\u2067",
    "\u2068",
    "\u2069",
    "\ufeff",
}

NOT_REQUIRED_STATUS = "not_required_for_comment_only_no_command"
TEMPLATE_STATUS = "non_authoritative_domain_template"

# These are prompts for an operator's review, not claims that a backup,
# rollback, or safer procedure exists. They deliberately contain no commands.
DOMAIN_GUIDANCE = {
    "general": {
        "smaller": "Limit the proposal to one named disposable fixture and one observable effect.",
        "safer": "Prefer read-only inspection or an authoritative dry-run in an isolated test environment.",
        "recovery": [
            "Identify the authoritative rollback owner and procedure.",
            "Record pre-change state without exposing sensitive values.",
            "Test restoration on a disposable fixture before execution.",
        ],
        "stop": [
            "Stop if the target, effect, owner, or privilege boundary is ambiguous.",
            "Stop if restoration has not been independently demonstrated.",
        ],
    },
    "system-wide": {
        "smaller": "Limit the proposal to one named service or setting on an isolated host.",
        "safer": "Prefer a scoped reversible change covered by the operator's authoritative runbook.",
        "recovery": [
            "Identify the authoritative rollback for the named service or setting.",
            "Capture current configuration without exposing sensitive values.",
            "Test rollback on an isolated host before production use.",
        ],
        "stop": [
            "Stop if the host, service, privilege boundary, or owner is ambiguous.",
            "Stop if maintenance approval or a tested rollback is missing.",
        ],
    },
    "disk": {
        "smaller": "Limit inspection to one resolved disposable image or mock device.",
        "safer": "Prefer read-only inspection of a disposable image over a live or shared volume.",
        "recovery": [
            "Resolve the exact device, mount, and symlink target before execution.",
            "Verify an independent restorable backup or snapshot with its owner.",
            "Complete a restore drill on disposable storage.",
        ],
        "stop": [
            "Stop if the device or mount identity is ambiguous or shared.",
            "Stop if a restorable backup has not been independently verified.",
        ],
    },
    "time": {
        "smaller": "Limit the proposal to reading one host's clock, timezone, and synchronization state.",
        "safer": "Test one time-setting change on a disposable host with no external dependencies.",
        "recovery": [
            "Record the current timezone, time source, and synchronization policy.",
            "Verify the authoritative resynchronization and restore procedure.",
            "Identify authentication, certificate, and database dependencies.",
        ],
        "stop": [
            "Stop if the host or authoritative time source is unknown.",
            "Stop if dependent systems are uncoordinated or rollback is untested.",
        ],
    },
    "network": {
        "smaller": "Limit the proposal to a read-only request against an inert mock endpoint without credentials.",
        "safer": "Stage the operation in an isolated network or provider test account.",
        "recovery": [
            "Record current routing, firewall, and remote-resource state if mutable.",
            "Verify out-of-band access and the authoritative rollback procedure.",
            "Confirm how remote-side effects would be identified and reversed.",
        ],
        "stop": [
            "Stop if the endpoint, account, region, or remote effect is ambiguous.",
            "Stop if management connectivity could be lost or credentials exposed.",
        ],
    },
    "credential": {
        "smaller": "Limit the proposal to metadata for one mock identifier without revealing a secret value.",
        "safer": "Use a disposable test credential with minimum scope and no production access.",
        "recovery": [
            "Verify the authoritative revocation and rotation playbook with its owner.",
            "Identify dependents and prepare a replacement rollout.",
            "Ensure logs and receipts cannot capture secret material.",
        ],
        "stop": [
            "Stop if a secret may be printed, logged, transmitted, or persisted.",
            "Stop if ownership, scope, dependents, or rotation ability is unknown.",
        ],
    },
}


def operator_guidance(decision: str, risk_domains: Iterable[str]) -> dict:
    """Return review prompts without asserting backup or recovery verification."""
    if decision == "allow":
        return {
            "smaller_alternative": {"status": NOT_REQUIRED_STATUS, "templates": []},
            "safer_alternative": {"status": NOT_REQUIRED_STATUS, "templates": []},
            "backup_evidence": {
                "status": NOT_REQUIRED_STATUS,
                "required_before_execution": False,
            },
            "recovery": {"status": NOT_REQUIRED_STATUS, "steps": []},
            "recovery_steps": {"status": NOT_REQUIRED_STATUS, "steps": []},
            "stop_conditions": {
                "status": NOT_REQUIRED_STATUS,
                "conditions": [],
            },
        }

    domains = sorted(
        domain for domain in set(risk_domains) if domain in DOMAIN_GUIDANCE
    ) or ["general"]

    def templates(kind: str) -> list[dict[str, str]]:
        return [
            {"domain": domain, "text": DOMAIN_GUIDANCE[domain][kind]}
            for domain in domains
        ]

    recovery_steps = [
        {"domain": domain, "text": step}
        for domain in domains
        for step in DOMAIN_GUIDANCE[domain]["recovery"]
    ]
    stop_conditions = [
        {"domain": domain, "text": condition}
        for domain in domains
        for condition in DOMAIN_GUIDANCE[domain]["stop"]
    ]
    return {
        "smaller_alternative": {
            "status": TEMPLATE_STATUS,
            "templates": templates("smaller"),
        },
        "safer_alternative": {
            "status": TEMPLATE_STATUS,
            "templates": templates("safer"),
        },
        "backup_evidence": {
            "status": "unverified_by_static_hook",
            "required_before_execution": True,
        },
        "recovery": {"status": "operator_must_verify", "steps": recovery_steps},
        "recovery_steps": {
            "status": "operator_must_verify",
            "steps": recovery_steps,
        },
        "stop_conditions": {
            "status": "operator_must_verify",
            "conditions": stop_conditions,
        },
    }


@dataclass
class Assessment:
    decision: str = "allow"
    rules: set[str] = field(default_factory=set)
    capabilities: set[str] = field(default_factory=set)
    uncertainties: set[str] = field(default_factory=set)
    features: set[str] = field(default_factory=set)
    risk_domains: set[str] = field(default_factory=set)
    scopes: set[str] = field(default_factory=set)
    targets: list[str] = field(default_factory=list)

    def add(
        self,
        decision: str,
        rule: str,
        *,
        capability: str | None = None,
        uncertainty: str | None = None,
        scope: str | None = None,
        target: str | None = None,
        risk_domain: str | None = None,
    ) -> None:
        if RANK[decision] > RANK[self.decision]:
            self.decision = decision
        self.rules.add(rule)
        if capability:
            self.capabilities.add(capability)
        if uncertainty:
            self.uncertainties.add(uncertainty)
        if scope:
            self.scopes.add(scope)
        if target and target not in self.targets and len(self.targets) < 6:
            self.targets.append(target[:160])
        if risk_domain:
            self.risk_domains.add(risk_domain)


def basename(token: str) -> str:
    return token.rstrip("/").rsplit("/", 1)[-1]


def tokenize(command: str) -> list[str]:
    lexer = shlex.shlex(command, posix=True, punctuation_chars=";&|()<>\n")
    lexer.whitespace = " \t\r"
    lexer.whitespace_split = True
    lexer.commenters = "#"
    return list(lexer)


def split_commands(tokens: Iterable[str]) -> list[tuple[str | None, list[str]]]:
    result: list[tuple[str | None, list[str]]] = []
    current: list[str] = []
    incoming: str | None = None

    for token in tokens:
        if token in CONTROL:
            if current:
                result.append((incoming, current))
                current = []
            incoming = token
        else:
            current.append(token)

    if current:
        result.append((incoming, current))
    return result


def active_shell_features(command: str) -> tuple[set[str], list[str]]:
    """Return quote-aware dynamic features and literal command substitutions."""
    features: set[str] = set()
    substitutions: list[str] = []
    i = 0
    quote: str | None = None

    def consume_parens(start: int) -> tuple[str | None, int]:
        depth = 1
        j = start
        inner_quote: str | None = None

        while j < len(command):
            char = command[j]

            if char == "\\":
                j += 2
                continue

            if inner_quote == "'":
                if char == "'":
                    inner_quote = None
                j += 1
                continue

            if inner_quote == '"':
                if char == '"':
                    inner_quote = None
                elif (
                    char == "$"
                    and j + 1 < len(command)
                    and command[j + 1] == "("
                ):
                    depth += 1
                    j += 2
                    continue
                j += 1
                continue

            if char in {"'", '"'}:
                inner_quote = char
            elif char == "(":
                depth += 1
            elif char == ")":
                depth -= 1
                if depth == 0:
                    return command[start:j], j + 1
            j += 1

        return None, len(command)

    while i < len(command):
        char = command[i]

        if char == "\\" and quote != "'":
            i += 2
            continue

        if quote == "'":
            if char == "'":
                quote = None
            i += 1
            continue

        if char == "'" and quote is None:
            quote = "'"
            i += 1
            continue

        if char == '"':
            quote = None if quote == '"' else '"'
            i += 1
            continue

        if char == "`":
            features.add("command-substitution")
            j = i + 1
            while j < len(command):
                if command[j] == "\\":
                    j += 2
                    continue
                if command[j] == "`":
                    break
                j += 1
            if j < len(command):
                substitutions.append(command[i + 1 : j])
            else:
                features.add("unbalanced-backtick")
            i = min(j + 1, len(command))
            continue

        if char == "$" and i + 1 < len(command) and command[i + 1] == "(":
            if i + 2 < len(command) and command[i + 2] == "(":
                features.add("arithmetic-expansion")
                i += 3
                continue

            features.add("command-substitution")
            content, end = consume_parens(i + 2)
            if content is None:
                features.add("unbalanced-substitution")
            else:
                substitutions.append(content)
            i = end
            continue

        if char in "<>" and i + 1 < len(command) and command[i + 1] == "(":
            features.add("process-substitution")

        if (
            char == "$"
            and i + 1 < len(command)
            and (command[i + 1].isalpha() or command[i + 1] in "_{@*#?!$-")
        ):
            features.add("parameter-expansion")

        if char in "*?[" and quote is None:
            features.add("glob-expansion")

        if char in "{}" and quote is None:
            features.add("brace-or-block")

        i += 1

    if quote is not None:
        features.add("unbalanced-quote")

    return features, substitutions


def is_ancestor(parent: str, child: str) -> bool:
    try:
        return os.path.commonpath([parent, child]) == parent
    except ValueError:
        return False


def opaque_ref(kind: str, value: str) -> str:
    digest = hashlib.sha256(value.encode("utf-8", "surrogatepass")).hexdigest()
    return f"<{kind}:sha256:{digest[:16]}>"


def redact_target(value: str) -> str:
    fixed = {
        "$PROJECT",
        "$CWD",
        "$HOME",
        "$HOST_ROOT",
        "$SYSTEM_TREE",
        "$CREDENTIAL_STORE",
        "<dynamic:path>",
        "<unknown>",
        "<stdin-derived>",
    }
    return value if value in fixed else opaque_ref("target", value)


def target_info(
    raw: str,
    cwd: str,
    project: str,
    home: str,
) -> tuple[str, str, bool]:
    token = raw.strip()
    symbolic = {
        "$HOME": home,
        "${HOME}": home,
        "~": home,
        "~/": home,
        "$CLAUDE_PROJECT_DIR": project,
        "${CLAUDE_PROJECT_DIR}": project,
        ".": cwd,
        "./": cwd,
    }

    dynamic = any(char in token for char in "$`*?[]{}") or "$(" in token
    candidate = symbolic.get(token)

    if candidate is None and token.startswith("~/"):
        candidate = os.path.join(home, token[2:])

    if candidate is None:
        prefix = re.split(r"[*?\[{]", token, maxsplit=1)[0]
        if prefix != token and prefix.endswith("/"):
            candidate = prefix.rstrip("/") or "/"
        elif not dynamic:
            candidate = token

    if candidate is None:
        return "<dynamic:path>", "unknown", False

    if not os.path.isabs(candidate):
        candidate = os.path.join(cwd, candidate)

    resolved = os.path.normpath(candidate)

    system_roots = (
        "/etc",
        "/bin",
        "/sbin",
        "/usr",
        "/var",
        "/System",
        "/Library",
        "/Applications",
    )
    credential_roots = (
        os.path.join(home, ".ssh"),
        os.path.join(home, ".gnupg"),
        os.path.join(home, ".aws"),
        os.path.join(home, ".config", "gcloud"),
        os.path.join(home, ".config", "op"),
    )

    if resolved == project:
        shown, scope, critical = "$PROJECT", "repository", True
    elif resolved == cwd:
        shown, scope, critical = "$CWD", "working-directory", True
    elif resolved == home:
        shown, scope, critical = "$HOME", "home", True
    elif resolved == "/":
        shown, scope, critical = "$HOST_ROOT", "host", True
    elif resolved == "/dev" or resolved.startswith("/dev/"):
        shown, scope, critical = opaque_ref("device", resolved), "device", True
    elif any(
        resolved == root or is_ancestor(root, resolved)
        for root in credential_roots
    ):
        shown, scope, critical = "$CREDENTIAL_STORE", "credential-store-or-file", True
    elif any(
        resolved == root or is_ancestor(root, resolved) for root in system_roots
    ):
        shown, scope, critical = "$SYSTEM_TREE", "system-tree", True
    elif is_ancestor(resolved, project) or is_ancestor(resolved, home):
        shown, scope, critical = resolved, "host-or-ancestor", True
    elif is_ancestor(project, resolved):
        shown = "$PROJECT/" + os.path.relpath(resolved, project)
        scope, critical = "subtree", False
    elif is_ancestor(home, resolved):
        shown = "$HOME/" + os.path.relpath(resolved, home)
        scope, critical = "subtree", False
    else:
        shown, scope, critical = resolved, "external-path", False

    if dynamic and critical:
        scope += "-contents"

    return shown[:160], scope, critical


def head(
    tokens: list[str], assessment: Assessment
) -> tuple[str, list[str]] | None:
    i = 0
    keywords = {"if", "then", "else", "elif", "while", "until", "do", "!"}

    while i < len(tokens):
        if ASSIGNMENT.match(tokens[i]):
            i += 1
            continue

        if tokens[i] in keywords:
            assessment.add(
                "ask",
                "SHELL_GRAMMAR",
                capability="shell.control",
                uncertainty="compound-shell-grammar",
            )
            i += 1
            continue

        if any(char in tokens[i] for char in "<>"):
            i = min(i + 2, len(tokens))
            continue

        break

    while i < len(tokens):
        executable = basename(tokens[i])

        if executable in {"sudo", "doas"}:
            assessment.add(
                "ask",
                "PRIVILEGE_WRAPPER",
                capability="privilege.escalation",
                uncertainty="wrapper-options",
                risk_domain="system-wide",
            )
            i += 1
            options_with_values = {
                "-u",
                "-g",
                "-h",
                "-p",
                "-C",
                "-T",
                "-R",
                "-D",
            }
            while i < len(tokens) and tokens[i].startswith("-"):
                option = tokens[i]
                i += 1
                if option in options_with_values and i < len(tokens):
                    i += 1
            continue

        if executable == "env":
            i += 1
            while i < len(tokens) and (
                tokens[i].startswith("-") or ASSIGNMENT.match(tokens[i])
            ):
                i += 1
            continue

        if executable in {"command", "builtin", "exec", "nohup", "time"}:
            i += 1
            while i < len(tokens) and tokens[i].startswith("-"):
                i += 1
            continue

        if executable == "timeout":
            i += 1
            while i < len(tokens) and tokens[i].startswith("-"):
                i += 1
            if i < len(tokens):
                i += 1
            continue

        break

    if i >= len(tokens):
        return None

    return tokens[i], tokens[i + 1 :]


def rm_operands(args: list[str]) -> tuple[bool, bool, list[str]]:
    recursive = False
    force = False
    operands = False
    targets: list[str] = []

    for arg in args:
        if not operands and arg == "--":
            operands = True
            continue

        if not operands and arg.startswith("--"):
            recursive |= arg == "--recursive"
            force |= arg == "--force"
            continue

        if not operands and arg.startswith("-") and arg != "-":
            flags = arg[1:]
            recursive |= "r" in flags or "R" in flags
            force |= "f" in flags
            continue

        targets.append(arg)

    return recursive, force, targets


def git_subcommand(args: list[str]) -> tuple[str | None, list[str]]:
    i = 0
    options_with_values = {
        "-C",
        "-c",
        "--git-dir",
        "--work-tree",
        "--namespace",
        "--exec-path",
    }

    while i < len(args):
        token = args[i]

        if token in options_with_values:
            i += 2
            continue

        if token.startswith(
            ("--git-dir=", "--work-tree=", "--namespace=", "--exec-path=")
        ):
            i += 1
            continue

        if token.startswith("-"):
            i += 1
            continue

        return token, args[i + 1 :]

    return None, []


def inspect_segment(
    tokens: list[str],
    incoming: str | None,
    assessment: Assessment,
    cwd: str,
    project: str,
    home: str,
    depth: int,
) -> None:
    parsed = head(tokens, assessment)
    if parsed is None:
        assessment.add("ask", "NO_EXECUTABLE", uncertainty="no-executable")
        return

    raw_executable, args = parsed
    executable = basename(raw_executable)

    if "/" in raw_executable and not raw_executable.startswith(
        ("/bin/", "/usr/bin/")
    ):
        assessment.add(
            "ask",
            "PATH_EXECUTABLE",
            capability="code.execution",
            uncertainty="mutable-executable-path",
        )

    # These families are capability signals, not proof of effect. In
    # particular, sudo is handled as a wrapper above: it is neither required
    # for a system-wide effect nor sufficient to prove one will occur.
    if executable in SYSTEM_WIDE_TOOLS:
        assessment.add(
            "ask",
            "SYSTEM_WIDE_CAPABLE_TOOL",
            capability="system.configuration-or-process-control",
            uncertainty="runtime-authority-and-subcommand",
            scope="host",
            risk_domain="system-wide",
        )

    if executable.startswith("mkfs") or executable in DISK_TOOLS:
        assessment.add(
            "ask",
            "DISK_CAPABLE_TOOL",
            capability="disk-or-filesystem-control",
            uncertainty="device-mapping-and-runtime-flags",
            scope="device-or-image",
            risk_domain="disk",
        )

    raw_device = next((arg for arg in args if arg.startswith("/dev/")), None)
    diskutil_mutators = {
        "eraseDisk",
        "partitionDisk",
        "zeroDisk",
        "secureErase",
        "eraseVolume",
        "deleteVolume",
        "apfs",
    }
    if executable == "diskutil" and any(
        arg in diskutil_mutators for arg in args
    ):
        assessment.add(
            "deny",
            "RAW_DEVICE_MUTATION",
            capability="storage.erase-or-repartition",
            scope="device",
            target=raw_device or "<unknown>",
            risk_domain="disk",
        )
        return

    if executable in {"fdisk", "sfdisk", "parted"} and raw_device:
        explicitly_read_only = executable == "fdisk" and any(
            arg in {"-l", "--list"} for arg in args
        )
        if not explicitly_read_only:
            assessment.add(
                "deny",
                "RAW_DEVICE_MUTATION",
                capability="storage.partition-table-write",
                scope="device",
                target=raw_device,
                risk_domain="disk",
            )
            return

    if executable == "date" or executable in TIME_TOOLS:
        assessment.add(
            "ask",
            "TIME_CAPABLE_TOOL",
            capability="clock-or-time-configuration",
            uncertainty="runtime-authority-and-subcommand",
            scope="host",
            risk_domain="time",
        )
        if executable in TIME_TOOLS:
            assessment.risk_domains.add("system-wide")

    if executable in NETWORK_TOOLS:
        assessment.add(
            "ask",
            "NETWORK_CAPABLE_TOOL",
            capability="network-io-or-configuration",
            uncertainty="remote-endpoint-and-server-side-effects",
            scope="remote-or-host-network",
            risk_domain="network",
        )

    credential_subcommand = executable in {
        "gh",
        "aws",
        "gcloud",
        "az",
        "docker",
        "npm",
        "pnpm",
    } and any(
        marker in arg.lower()
        for arg in args
        for marker in ("auth", "login", "credential", "token", "configure")
    )
    sensitive_argument = any(SENSITIVE_ARGUMENT.search(arg) for arg in args)
    if executable in CREDENTIAL_TOOLS or credential_subcommand or sensitive_argument:
        assessment.add(
            "ask",
            "CREDENTIAL_CAPABLE_TOOL_OR_TARGET",
            capability="credential-read-write-or-disclosure",
            uncertainty="secret-store-and-runtime-output",
            scope="credential-store-or-file",
            risk_domain="credential",
        )

    if executable in {"rm", "unlink", "rmdir", "shred", "truncate", "find"}:
        assessment.risk_domains.add("disk")

    if executable in SHELLS:
        if incoming in PIPE_OPS and "-c" not in args:
            assessment.add(
                "deny",
                "OPAQUE_PIPE_TO_SHELL",
                capability="code.execution",
                uncertainty="stdin-generated-code",
                scope="unknown",
            )
            return

        if "-c" in args:
            position = args.index("-c")
            assessment.add(
                "ask",
                "SHELL_C_WRAPPER",
                capability="code.execution",
                uncertainty="second-parse",
            )
            if position + 1 < len(args):
                program = args[position + 1]
                if any(marker in program for marker in ("$(", "`", "${")):
                    assessment.add(
                        "deny",
                        "OPAQUE_DYNAMIC_SHELL_C",
                        capability="code.execution",
                        uncertainty="runtime-generated-program",
                        scope="unknown",
                    )
                inspect_snippet(
                    program,
                    assessment,
                    cwd,
                    project,
                    home,
                    depth + 1,
                )
            else:
                assessment.add(
                    "ask",
                    "SHELL_C_MISSING_PROGRAM",
                    uncertainty="missing-program",
                )
        else:
            assessment.add(
                "ask",
                "SHELL_OR_SCRIPT_WRAPPER",
                capability="code.execution",
                uncertainty="script-or-stdin-not-inspected",
            )
        return

    if executable in {"eval", "source", "."}:
        assessment.add(
            "deny",
            "OPAQUE_REPARSE",
            capability="code.execution",
            uncertainty="runtime-reparse",
            scope="unknown",
        )
        return

    if executable == "xargs":
        names = {basename(arg) for arg in args}
        if names.intersection({"rm", "unlink", "rmdir", "shred"}):
            assessment.add(
                "deny",
                "UNBOUNDED_XARGS_DELETE",
                capability="filesystem.delete",
                uncertainty="stdin-derived-targets",
                scope="unknown",
                target="<stdin-derived>",
                risk_domain="disk",
            )
        else:
            assessment.add(
                "ask",
                "XARGS_DYNAMIC_TARGETS",
                capability="code.execution",
                uncertainty="stdin-derived-arguments",
            )
        return

    if executable == "rm":
        recursive, force, targets = rm_operands(args)

        if not targets:
            assessment.add(
                "ask",
                "DELETE_WITHOUT_STATIC_TARGET",
                capability="filesystem.delete",
                uncertainty="missing-or-dynamic-target",
            )

        for target in targets:
            shown, scope, critical = target_info(target, cwd, project, home)
            if recursive and critical:
                assessment.add(
                    "deny",
                    "RECURSIVE_DELETE_CRITICAL_SCOPE",
                    capability="filesystem.delete",
                    scope=scope,
                    target=shown,
                    risk_domain=(
                        "system-wide"
                        if scope in {"host", "host-or-ancestor", "system-tree"}
                        else (
                            "credential"
                            if scope == "credential-store-or-file"
                            else "disk"
                        )
                    ),
                )
            else:
                assessment.add(
                    "ask",
                    "FILESYSTEM_DELETE",
                    capability="filesystem.delete",
                    scope=scope,
                    target=shown,
                    uncertainty=(
                        "dynamic-target" if shown == "<dynamic:path>" else None
                    ),
                )

        if force:
            assessment.features.add("force")
        if recursive:
            assessment.features.add("recursive")
        return

    if executable in {"unlink", "rmdir", "shred", "truncate"}:
        targets = [arg for arg in args if not arg.startswith("-")]
        if not targets:
            targets = ["<unknown>"]

        for target in targets[:6]:
            if target == "<unknown>":
                shown, scope, critical = target, "unknown", False
            else:
                shown, scope, critical = target_info(target, cwd, project, home)
            assessment.add(
                "deny" if executable == "shred" and critical else "ask",
                "DESTRUCTIVE_FILE_OPERATION",
                capability="filesystem.delete-or-truncate",
                scope=scope,
                target=shown,
            )
        return

    if executable == "find":
        mutators = {
            "-delete",
            "-exec",
            "-execdir",
            "-ok",
            "-okdir",
            "-fprint",
            "-fprint0",
        }
        if any(arg in mutators for arg in args):
            roots: list[str] = []
            for arg in args:
                if arg.startswith("-") or arg in {"!", "("}:
                    break
                roots.append(arg)
            if not roots:
                roots = ["."]

            for target in roots:
                shown, scope, critical = target_info(target, cwd, project, home)
                assessment.add(
                    "deny" if critical else "ask",
                    "MUTATING_FIND",
                    capability="filesystem.bulk-mutation",
                    scope=scope,
                    target=shown,
                )
            return

    if executable.startswith("mkfs") or executable == "wipefs":
        assessment.add(
            "deny",
            "FILESYSTEM_FORMATTER",
            capability="storage.erase",
            scope="device-or-image",
            risk_domain="disk",
        )
        return

    if executable == "dd":
        output = next((arg[3:] for arg in args if arg.startswith("of=")), "<unknown>")
        if output.startswith("/dev/"):
            assessment.add(
                "deny",
                "RAW_DEVICE_WRITE",
                capability="storage.write",
                scope="device",
                target=opaque_ref("device", output),
                risk_domain="disk",
            )
        else:
            if output == "<unknown>":
                shown, scope, critical = output, "unknown", False
            else:
                shown, scope, critical = target_info(output, cwd, project, home)
            assessment.add(
                "deny" if critical else "ask",
                "DD_WRITE",
                capability="filesystem.overwrite",
                scope=scope,
                target=shown,
                risk_domain="disk",
            )
        return

    if executable in {"chmod", "chown", "chgrp"}:
        recursive = any(arg in {"-R", "--recursive"} for arg in args)
        target = next(
            (arg for arg in reversed(args) if not arg.startswith("-")),
            "<unknown>",
        )
        if target == "<unknown>":
            shown, scope, critical = target, "unknown", False
        else:
            shown, scope, critical = target_info(target, cwd, project, home)
        assessment.add(
            "deny" if recursive and critical else "ask",
            "PERMISSION_OR_OWNER_CHANGE",
            capability="filesystem.metadata-write",
            scope=scope,
            target=shown,
            risk_domain="system-wide" if critical else "disk",
        )
        return

    if executable == "git":
        subcommand, rest = git_subcommand(args)

        if subcommand == "clean":
            dry_run = any(
                arg in {"-n", "--dry-run"}
                or (
                    arg.startswith("-")
                    and not arg.startswith("--")
                    and "n" in arg[1:]
                )
                for arg in rest
            )
            forced = any(
                arg in {"-f", "--force"}
                or (
                    arg.startswith("-")
                    and not arg.startswith("--")
                    and "f" in arg[1:]
                )
                for arg in rest
            )
            ignored = any(
                arg in {"-x", "-X"}
                or (
                    arg.startswith("-")
                    and not arg.startswith("--")
                    and ("x" in arg[1:] or "X" in arg[1:])
                )
                for arg in rest
            )

            if dry_run:
                assessment.add(
                    "ask",
                    "GIT_RUNTIME_EXTENSIONS",
                    capability="repository.read-with-runtime-extensions",
                    scope="repository",
                    uncertainty="git-hooks-config-fsmonitor-or-pager",
                )
                return

            assessment.add(
                "deny" if forced and ignored else "ask",
                "GIT_CLEAN",
                capability="repository.delete",
                scope="repository",
                uncertainty=None if forced else "git-config-can-enable-force",
            )
            return

        if subcommand in {
            "reset",
            "restore",
            "checkout",
            "rm",
            "rebase",
            "filter-branch",
            "filter-repo",
            "gc",
            "reflog",
        }:
            assessment.add(
                "ask",
                "GIT_HISTORY_OR_WORKTREE_MUTATION",
                capability="repository.rewrite",
                scope="repository",
            )
            return

        ref_mutation = any(
            arg in {"--force", "-f", "-D", "-d", "--delete"}
            or arg.startswith("--force-with-lease")
            for arg in rest
        )
        if subcommand in {"push", "tag", "branch"} and ref_mutation:
            assessment.add(
                "ask",
                "GIT_REF_MUTATION",
                capability="remote-or-local-ref-write",
                scope="repository-or-remote",
            )
            return

        assessment.add(
            "ask",
            "GIT_RUNTIME_EXTENSIONS",
            capability="repository-access-with-runtime-extensions",
            scope="repository",
            uncertainty=(
                "git-subcommand-sha256:"
                + hashlib.sha256(
                    (subcommand or "unknown").encode("utf-8", "surrogatepass")
                ).hexdigest()[:16]
            ),
        )
        return

    if executable == "rg" and any(
        arg == "--pre" or arg.startswith("--pre=") for arg in args
    ):
        assessment.add(
            "ask",
            "RG_PREPROCESSOR",
            capability="code.execution",
            uncertainty="external-preprocessor",
        )
        return

    if executable in LEXICALLY_SIMPLE_COMMANDS:
        assessment.add(
            "ask",
            "RUNTIME_RESOLUTION_UNVERIFIED",
            capability="shell.command-resolution",
            uncertainty="function-alias-builtin-or-filesystem-shadowing",
        )
        return

    if executable in {"python", "python3", "node", "perl", "ruby", "php"}:
        assessment.add(
            "ask",
            "GENERAL_INTERPRETER",
            capability="code.execution",
            uncertainty="program-semantics-not-inspected",
        )
        return

    assessment.add(
        "ask",
        "UNKNOWN_EXECUTABLE",
        capability="code.execution",
        uncertainty=(
            "executable-sha256:"
            + hashlib.sha256(
                executable.encode("utf-8", "surrogatepass")
            ).hexdigest()[:16]
        ),
    )


def inspect_snippet(
    command: str,
    assessment: Assessment,
    cwd: str,
    project: str,
    home: str,
    depth: int = 0,
) -> None:
    if depth > 3:
        assessment.add(
            "ask", "RECURSION_LIMIT", uncertainty="nested-shell-depth"
        )
        return

    features, substitutions = active_shell_features(command)
    assessment.features.update(features)

    if features.intersection(
        {"unbalanced-quote", "unbalanced-substitution", "unbalanced-backtick"}
    ):
        assessment.add(
            "ask", "UNBALANCED_SHELL_SYNTAX", uncertainty="parse-incomplete"
        )

    dynamic = {
        "command-substitution",
        "process-substitution",
        "arithmetic-expansion",
        "parameter-expansion",
        "glob-expansion",
        "brace-or-block",
    }
    if features.intersection(dynamic):
        assessment.add(
            "ask",
            "DYNAMIC_SHELL_EXPANSION",
            capability="shell.dynamic",
            uncertainty="runtime-expansion",
        )

    for inner in substitutions[:6]:
        inspect_snippet(inner, assessment, cwd, project, home, depth + 1)

    try:
        tokens = tokenize(command)
    except ValueError:
        assessment.add(
            "ask", "TOKENIZE_FAILURE", uncertainty="shell-parse-failed"
        )
        return

    if not tokens:
        if command.strip().startswith("#"):
            assessment.rules.add("COMMENT_ONLY_NO_COMMAND")
        else:
            assessment.add("ask", "EMPTY_OR_COMMENT", uncertainty="no-command")
        return

    # Heredoc bodies require full shell grammar. Do not derive deny from body
    # text because the apparent command can be inert data.
    if any("<<" in token for token in tokens):
        assessment.add(
            "ask",
            "HEREDOC_UNINSPECTED",
            capability="shell.dynamic",
            uncertainty="heredoc-body",
        )
        return

    if any(any(char in token for char in "<>") for token in tokens):
        assessment.add(
            "ask",
            "SHELL_REDIRECTION",
            capability="filesystem-or-fd-write",
            uncertainty="redirection-target",
        )

    for incoming, segment in split_commands(tokens):
        inspect_segment(
            segment, incoming, assessment, cwd, project, home, depth
        )


def redact_context_path(value: str, project: str, home: str) -> str:
    value = os.path.normpath(value)

    if value == project:
        return "$PROJECT"
    if is_ancestor(project, value):
        return opaque_ref("project-cwd", os.path.relpath(value, project))
    if value == home:
        return "$HOME"
    if is_ancestor(home, value):
        return opaque_ref("home-cwd", os.path.relpath(value, home))

    return opaque_ref("external-cwd", value)


def classify_payload(
    payload: object,
    *,
    project_dir: str | None = None,
    home_dir: str | None = None,
) -> dict:
    assessment = Assessment()

    if not isinstance(payload, dict):
        assessment.add(
            "deny", "INVALID_PAYLOAD", uncertainty="top-level-not-object"
        )
        command = ""
        cwd = os.getcwd()
        permission_mode = "unknown"
    else:
        tool_name = payload.get("tool_name")
        tool_input = payload.get("tool_input")
        cwd_value = payload.get("cwd")
        permission_mode = payload.get("permission_mode", "unknown")

        if isinstance(cwd_value, str) and os.path.isabs(cwd_value):
            cwd = os.path.normpath(cwd_value)
        else:
            cwd = os.getcwd()
            assessment.add(
                "ask", "INVALID_CWD", uncertainty="missing-or-relative-cwd"
            )

        command = (
            tool_input.get("command") if isinstance(tool_input, dict) else None
        )

        if tool_name != "Bash":
            assessment.add(
                "deny",
                "UNEXPECTED_TOOL",
                uncertainty="matcher-misconfiguration",
            )

        if not isinstance(command, str) or not command:
            assessment.add(
                "deny",
                "INVALID_COMMAND",
                uncertainty="missing-or-nonstring-command",
            )
            command = ""

    project = os.path.normpath(
        project_dir or os.environ.get("CLAUDE_PROJECT_DIR") or cwd
    )
    home = os.path.normpath(home_dir or os.path.expanduser("~"))

    encoded = command.encode("utf-8", "surrogatepass")
    digest = hashlib.sha256(encoded).hexdigest()

    if len(encoded) > MAX_COMMAND_BYTES:
        assessment.add(
            "deny", "COMMAND_TOO_LARGE", uncertainty="inspection-size-limit"
        )
    elif "\x00" in command:
        assessment.add(
            "deny", "NUL_IN_COMMAND", uncertainty="invalid-shell-string"
        )
    else:
        if any(
            char in INVISIBLE or unicodedata.category(char) == "Cf"
            for char in command
        ):
            assessment.add(
                "ask",
                "INVISIBLE_UNICODE",
                uncertainty="visual-order-or-invisible-codepoint",
            )

        inspect_snippet(command, assessment, cwd, project, home)

    explicit_high_impact = bool(
        assessment.rules.intersection(EXPLICIT_HIGH_IMPACT_SIGNALS)
    )
    if assessment.decision == "deny" and explicit_high_impact:
        risk, confidence, reversibility = "critical", "high", "hard"
        decision_basis = "matched_explicit_high_impact_signature"
    elif assessment.decision == "deny":
        risk, confidence, reversibility = "indeterminate", "low", "unknown"
        decision_basis = "inspection_uncertainty_fail_closed"
    elif assessment.decision == "ask":
        risk, confidence, reversibility = "medium", "medium", "uncertain"
        decision_basis = "review_required_capability_or_uncertainty"
    else:
        risk, confidence, reversibility = "low", "high", "none"
        decision_basis = "comment_only_no_command"

    scope_priority = (
        "host",
        "host-or-ancestor",
        "system-tree",
        "home",
        "credential-store-or-file",
        "repository",
        "working-directory",
        "device",
        "device-or-image",
        "repository-or-remote",
        "subtree",
        "external-path",
        "unknown",
    )
    scope = next(
        (
            candidate
            for candidate in scope_priority
            if any(
                value == candidate or value == candidate + "-contents"
                for value in assessment.scopes
            )
        ),
        "none" if assessment.decision == "allow" else "unknown",
    )

    receipt = {
        "schema": "alphalab.bash-blast-radius/v1",
        "decision": assessment.decision,
        "decision_basis": decision_basis,
        "risk": risk,
        "confidence": confidence,
        "policy_interpretation": (
            "static_classification_not_proof_of_harmlessness_or_effect"
        ),
        "command": {
            "sha256": digest,
            "sha256_purpose": "binding_and_integrity_only_not_anonymization",
            "bytes": len(encoded),
            "raw_included": False,
        },
        "context": {
            "cwd": redact_context_path(cwd, project, home),
            "permission_mode": (
                permission_mode
                if isinstance(permission_mode, str)
                and permission_mode in KNOWN_PERMISSION_MODES
                else "unknown"
            ),
        },
        "blast_radius": {
            "scope": scope,
            "capabilities": sorted(assessment.capabilities)[:8],
            "targets": [redact_target(value) for value in assessment.targets],
            "reversibility": reversibility,
        },
        "risk_domains": sorted(assessment.risk_domains),
        "signals": sorted(assessment.rules)[:12],
        "uncertainties": sorted(assessment.uncertainties)[:12],
        "shell_features": sorted(assessment.features)[:12],
    }
    receipt.update(operator_guidance(assessment.decision, assessment.risk_domains))
    return receipt


def hook_output(receipt: dict) -> dict:
    # permissionDecisionReason must be a string, so serialize the receipt once
    # inside Claude Code's official structured-output envelope.
    reason = json.dumps(
        {"type": "bash_blast_radius_receipt", "receipt": receipt},
        ensure_ascii=True,
        separators=(",", ":"),
        sort_keys=True,
    )

    output = {
        "hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": receipt["decision"],
            "permissionDecisionReason": reason,
        }
    }

    # A deny reason is delivered to Claude. systemMessage requests a
    # user-facing warning, but it is not specified as part of the permission card.
    if receipt["decision"] == "deny":
        output["systemMessage"] = reason

    return output


def invalid_stdin_receipt(rule: str, raw: bytes) -> dict:
    receipt = {
        "schema": "alphalab.bash-blast-radius/v1",
        "decision": "deny",
        "decision_basis": "inspection_uncertainty_fail_closed",
        "risk": "indeterminate",
        "confidence": "low",
        "policy_interpretation": (
            "static_classification_not_proof_of_harmlessness_or_effect"
        ),
        "command": {
            "sha256": hashlib.sha256(raw).hexdigest(),
            "sha256_purpose": "binding_and_integrity_only_not_anonymization",
            "bytes": len(raw),
            "raw_included": False,
        },
        "context": {"cwd": "<unknown>", "permission_mode": "unknown"},
        "blast_radius": {
            "scope": "unknown",
            "capabilities": [],
            "targets": [],
            "reversibility": "unknown",
        },
        "risk_domains": [],
        "signals": [rule],
        "uncertainties": ["hook-input-unavailable"],
        "shell_features": [],
    }
    receipt.update(operator_guidance("deny", []))
    return receipt


def main() -> int:
    raw = sys.stdin.buffer.read(MAX_STDIN_BYTES + 1)

    try:
        if len(raw) > MAX_STDIN_BYTES:
            receipt = invalid_stdin_receipt("STDIN_TOO_LARGE", raw)
        else:
            payload = json.loads(raw.decode("utf-8", "strict"))
            receipt = classify_payload(payload)
    except (UnicodeDecodeError, json.JSONDecodeError):
        receipt = invalid_stdin_receipt("INVALID_STDIN_JSON", raw)
    except Exception:
        # Never echo exception text: it may contain input or local paths.
        receipt = invalid_stdin_receipt("GATE_INTERNAL_ERROR", raw)

    serialized = json.dumps(
        hook_output(receipt),
        ensure_ascii=True,
        separators=(",", ":"),
        sort_keys=True,
    )
    sys.stdout.write(serialized)
    sys.stdout.write("\n")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Non-executing policy and output-contract tests for bash-risk-gate.py."""

import importlib.util
import hashlib
import json
import pathlib
import subprocess
import sys
import unittest

GATE = pathlib.Path(__file__).with_name("bash-risk-gate.py")
SPEC = importlib.util.spec_from_file_location("bash_risk_gate", GATE)
assert SPEC is not None and SPEC.loader is not None
gate = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = gate
SPEC.loader.exec_module(gate)

PROJECT = "/__alphalab_gate_mock__/project"
HOME = "/__alphalab_gate_mock__/home"

# These strings are data only. Potentially mutating executable paths point to
# a deliberately nonexistent /__mock__/bin. No fixture command is executed.
FIXTURES = [
    ("comment-only-no-command", "# inert fixture only", "allow", False),
    ("simple-pwd-runtime-review", "pwd", "ask", True),
    ("git-read-chain-review", "git status --short && git diff --stat", "ask", True),
    (
        "quoted-danger-text",
        "printf '%s\\n' '/__mock__/bin/rm -rf " + PROJECT + "'",
        "ask",
        True,
    ),
    (
        "bounded-delete",
        "/__mock__/bin/rm -f " + PROJECT + "/tmp/one",
        "ask",
        True,
    ),
    ("project-delete", "/__mock__/bin/rm -rf " + PROJECT, "deny", True),
    (
        "normalized-project-delete",
        "/__mock__/bin/rm -rf " + PROJECT + "/cache/..",
        "deny",
        True,
    ),
    (
        "chain-delete",
        "printf ok && /__mock__/bin/rm -rf " + PROJECT,
        "deny",
        True,
    ),
    (
        "newline-delete",
        "printf ok\n/__mock__/bin/rm -rf " + PROJECT,
        "deny",
        True,
    ),
    (
        "env-wrapper",
        "env MOCK=1 /__mock__/bin/rm -rf " + PROJECT,
        "deny",
        True,
    ),
    (
        "shell-c",
        "bash -c '/__mock__/bin/rm -rf " + PROJECT + "'",
        "deny",
        True,
    ),
    (
        "command-substitution",
        'echo "$(/__mock__/bin/rm -rf ' + PROJECT + ')"',
        "deny",
        True,
    ),
    (
        "wrapper-script",
        "/__mock__/bin/mock-cleanup.sh --dry-run",
        "ask",
        True,
    ),
    (
        "harmless-encoded-pipe",
        "printf 'bW9jayBwYXlsb2Fk' | /__mock__/bin/base64 -d | sh",
        "deny",
        True,
    ),
    (
        "xargs-delete",
        "printf '%s\\n' /__alphalab_gate_mock__/one | "
        "xargs /__mock__/bin/rm -rf",
        "deny",
        True,
    ),
    (
        "git-hard-reset-invalid-ref",
        "/__mock__/bin/git reset --hard MOCK_NONEXISTENT_REF",
        "ask",
        True,
    ),
    ("git-clean-dry-run-review", "git clean -ndx", "ask", True),
    (
        "git-clean-force-mock-bin",
        "/__mock__/bin/git clean -fdx",
        "deny",
        True,
    ),
    (
        "find-subtree",
        "/__mock__/bin/find " + PROJECT + "/cache -type f -delete",
        "ask",
        True,
    ),
    (
        "find-project",
        "/__mock__/bin/find " + PROJECT + " -type f -delete",
        "deny",
        True,
    ),
    (
        "redirection-to-missing-mock-path",
        "printf ok > " + PROJECT + "/out",
        "ask",
        True,
    ),
    (
        "rg-pre-mock",
        "rg --pre /__mock__/bin/reader needle .",
        "ask",
        True,
    ),
    ("harmless-interpreter", "python3 -c 'print(1)'", "ask", True),
    (
        "heredoc-inert-text",
        "cat <<'MOCK'\n/__mock__/bin/rm -rf " + PROJECT + "\nMOCK",
        "ask",
        True,
    ),
    ("unbalanced-inert", "printf 'mock", "ask", True),
    ("unicode-bidi-inert", "printf '\u202emock'", "ask", True),
    (
        "relative-after-cd",
        "cd /__alphalab_gate_mock__/elsewhere && "
        "/__mock__/bin/rm -rf cache",
        "ask",
        True,
    ),
]

# Additional command strings are still data only. They exercise the explicit
# receipt taxonomy requested for host-level review. None of these executables
# exists at /__mock__/bin, and the harness never attempts to invoke them.
DOMAIN_FIXTURES = [
    (
        "system-wide-without-sudo",
        "/__mock__/bin/systemctl restart mock.service",
        "ask",
        {"system-wide"},
    ),
    (
        "sudo-is-signal-not-proof",
        "/__mock__/bin/sudo -n /__mock__/bin/printf ok",
        "ask",
        {"system-wide"},
    ),
    (
        "disk-formatter-mock-device",
        "/__mock__/bin/mkfs.mock /dev/mock0",
        "deny",
        {"disk"},
    ),
    (
        "time-configuration-mock",
        "/__mock__/bin/timedatectl set-time '2000-01-01 00:00:00'",
        "ask",
        {"system-wide", "time"},
    ),
    (
        "network-read-invalid-domain",
        "/__mock__/bin/curl https://example.invalid/mock",
        "ask",
        {"network"},
    ),
    (
        "credential-store-mock-query",
        "/__mock__/bin/security find-generic-password -s mock -w",
        "ask",
        {"credential"},
    ),
]

# Focused semantic regressions. Every possibly destructive command remains an
# inert string; only the Python classifier sees it.
SEMANTIC_FIXTURES = [
    (
        "diskutil-raw-device",
        "/__mock__/bin/diskutil eraseDisk MOCK /__mock__/volume /dev/mock0",
        "deny",
        True,
    ),
    (
        "fdisk-raw-device",
        "/__mock__/bin/fdisk /dev/mock0",
        "deny",
        True,
    ),
    (
        "recursive-system-tree",
        "/__mock__/bin/rm -rf /etc",
        "deny",
        True,
    ),
    (
        "recursive-key-directory",
        "/__mock__/bin/rm -rf " + HOME + "/.ssh",
        "deny",
        True,
    ),
    ("opaque-source", "source harmless-mock.sh", "deny", True),
    ("opaque-eval", "eval printf ok", "deny", True),
    ("git-status-runtime", "git status --short", "ask", True),
    ("git-diff-runtime", "git diff --stat", "ask", True),
    ("git-log-runtime", "git log -1", "ask", True),
    ("git-show-runtime", "git show MOCK_REF", "ask", True),
    ("narrow-builtin-chain", "pwd && printf '%s\\n' mock", "ask", True),
]

EXPECTED_RECEIPT_KEYS = {
    "schema",
    "decision",
    "decision_basis",
    "risk",
    "confidence",
    "policy_interpretation",
    "command",
    "context",
    "blast_radius",
    "risk_domains",
    "signals",
    "uncertainties",
    "shell_features",
    "smaller_alternative",
    "safer_alternative",
    "backup_evidence",
    "recovery",
    "recovery_steps",
    "stop_conditions",
}


def payload(command):
    return {
        "session_id": "fixture-session",
        "transcript_path": "/__mock__/transcript.jsonl",
        "cwd": PROJECT,
        "permission_mode": "default",
        "hook_event_name": "PreToolUse",
        "tool_name": "Bash",
        "tool_input": {
            "command": command,
            "description": "Non-executing fixture",
            "timeout": 120000,
            "run_in_background": False,
        },
        "tool_use_id": "fixture-tool-use",
    }


class GateTests(unittest.TestCase):
    def test_policy_fixtures(self):
        for name, command, expected, _ in FIXTURES:
            with self.subTest(name=name):
                receipt = gate.classify_payload(
                    payload(command), project_dir=PROJECT, home_dir=HOME
                )
                self.assertEqual(expected, receipt["decision"])
                self.assertFalse(receipt["command"]["raw_included"])
                self.assertNotIn(command, json.dumps(receipt))

    def test_exact_decision_counts(self):
        counts = {"allow": 0, "ask": 0, "deny": 0}
        for _, command, _, _ in FIXTURES:
            receipt = gate.classify_payload(
                payload(command), project_dir=PROJECT, home_dir=HOME
            )
            counts[receipt["decision"]] += 1
        self.assertEqual({"allow": 1, "ask": 15, "deny": 11}, counts)

    def test_confusion_matrix(self):
        tp = fp = tn = fn = 0

        for _, command, _, intervention_required in FIXTURES:
            receipt = gate.classify_payload(
                payload(command), project_dir=PROJECT, home_dir=HOME
            )
            predicted_positive = receipt["decision"] != "allow"

            if intervention_required and predicted_positive:
                tp += 1
            elif intervention_required:
                fn += 1
            elif predicted_positive:
                fp += 1
            else:
                tn += 1

        self.assertEqual((26, 0, 1, 0), (tp, fp, tn, fn))

    def test_semantic_regression_fixtures(self):
        for name, command, expected, _ in SEMANTIC_FIXTURES:
            with self.subTest(name=name):
                receipt = gate.classify_payload(
                    payload(command), project_dir=PROJECT, home_dir=HOME
                )
                self.assertEqual(expected, receipt["decision"])
                self.assertNotIn(command, json.dumps(receipt))

    def test_required_risk_domains(self):
        observed = set()
        for name, command, expected, required_domains in DOMAIN_FIXTURES:
            with self.subTest(name=name):
                receipt = gate.classify_payload(
                    payload(command), project_dir=PROJECT, home_dir=HOME
                )
                self.assertEqual(expected, receipt["decision"])
                self.assertEqual(
                    sorted(receipt["risk_domains"]), receipt["risk_domains"]
                )
                self.assertTrue(
                    required_domains.issubset(set(receipt["risk_domains"]))
                )
                self.assertNotIn(command, json.dumps(receipt))
                observed.update(receipt["risk_domains"])

        self.assertTrue(
            {"system-wide", "disk", "time", "network", "credential"}
            .issubset(observed)
        )

    def test_sudo_is_neither_necessary_nor_sufficient(self):
        without_sudo = gate.classify_payload(
            payload("/__mock__/bin/systemctl restart mock.service"),
            project_dir=PROJECT,
            home_dir=HOME,
        )
        sudo_only = gate.classify_payload(
            payload("/__mock__/bin/sudo -n /__mock__/bin/printf ok"),
            project_dir=PROJECT,
            home_dir=HOME,
        )

        self.assertIn("system-wide", without_sudo["risk_domains"])
        self.assertEqual("ask", sudo_only["decision"])
        self.assertIn("PRIVILEGE_WRAPPER", sudo_only["signals"])
        self.assertNotEqual("deny", sudo_only["decision"])

    def test_official_auto_permission_mode_is_preserved(self):
        probe = payload("pwd")
        probe["permission_mode"] = "auto"
        receipt = gate.classify_payload(
            probe, project_dir=PROJECT, home_dir=HOME
        )
        self.assertEqual("auto", receipt["context"]["permission_mode"])

    def test_extended_counts_and_confusion_matrix(self):
        counts = {"allow": 0, "ask": 0, "deny": 0}
        tp = fp = tn = fn = 0

        combined = FIXTURES + [
            (name, command, expected, True)
            for name, command, expected, _ in DOMAIN_FIXTURES
        ] + SEMANTIC_FIXTURES
        for _, command, _, intervention_required in combined:
            receipt = gate.classify_payload(
                payload(command), project_dir=PROJECT, home_dir=HOME
            )
            counts[receipt["decision"]] += 1
            predicted_positive = receipt["decision"] != "allow"
            if intervention_required and predicted_positive:
                tp += 1
            elif intervention_required:
                fn += 1
            elif predicted_positive:
                fp += 1
            else:
                tn += 1

        self.assertEqual({"allow": 1, "ask": 25, "deny": 18}, counts)
        self.assertEqual((43, 0, 1, 0), (tp, fp, tn, fn))

    def test_exact_receipt_guidance_schema_and_content(self):
        combined = FIXTURES + [
            (name, command, expected, True)
            for name, command, expected, _ in DOMAIN_FIXTURES
        ] + SEMANTIC_FIXTURES

        for name, command, expected, _ in combined:
            with self.subTest(name=name):
                receipt = gate.classify_payload(
                    payload(command), project_dir=PROJECT, home_dir=HOME
                )
                self.assertEqual(expected, receipt["decision"])
                self.assertEqual(EXPECTED_RECEIPT_KEYS, set(receipt))
                self.assertEqual(
                    {"sha256", "sha256_purpose", "bytes", "raw_included"},
                    set(receipt["command"]),
                )
                self.assertEqual(
                    hashlib.sha256(command.encode()).hexdigest(),
                    receipt["command"]["sha256"],
                )
                self.assertEqual(
                    "binding_and_integrity_only_not_anonymization",
                    receipt["command"]["sha256_purpose"],
                )
                self.assertEqual(
                    "static_classification_not_proof_of_harmlessness_or_effect",
                    receipt["policy_interpretation"],
                )
                self.assertEqual(
                    sorted(receipt["risk_domains"]), receipt["risk_domains"]
                )

                if expected == "allow":
                    self.assertEqual(
                        "comment_only_no_command", receipt["decision_basis"]
                    )
                    self.assertEqual(
                        "not_required_for_comment_only_no_command",
                        gate.NOT_REQUIRED_STATUS,
                    )
                    self.assertEqual(
                        {
                            "status": gate.NOT_REQUIRED_STATUS,
                            "templates": [],
                        },
                        receipt["smaller_alternative"],
                    )
                    self.assertEqual(
                        {
                            "status": gate.NOT_REQUIRED_STATUS,
                            "templates": [],
                        },
                        receipt["safer_alternative"],
                    )
                    self.assertEqual(
                        {
                            "status": gate.NOT_REQUIRED_STATUS,
                            "required_before_execution": False,
                        },
                        receipt["backup_evidence"],
                    )
                    self.assertEqual(
                        {"status": gate.NOT_REQUIRED_STATUS, "steps": []},
                        receipt["recovery"],
                    )
                    self.assertEqual(
                        receipt["recovery"], receipt["recovery_steps"]
                    )
                    self.assertEqual(
                        {
                            "status": gate.NOT_REQUIRED_STATUS,
                            "conditions": [],
                        },
                        receipt["stop_conditions"],
                    )
                else:
                    expected_domains = receipt["risk_domains"] or ["general"]
                    for field in ("smaller_alternative", "safer_alternative"):
                        self.assertEqual(
                            {"status", "templates"}, set(receipt[field])
                        )
                        self.assertEqual(
                            gate.TEMPLATE_STATUS, receipt[field]["status"]
                        )
                        self.assertEqual(
                            expected_domains,
                            [
                                item["domain"]
                                for item in receipt[field]["templates"]
                            ],
                        )
                        self.assertTrue(
                            all(
                                set(item) == {"domain", "text"}
                                and item["text"]
                                for item in receipt[field]["templates"]
                            )
                        )

                    self.assertEqual(
                        {
                            "status": "unverified_by_static_hook",
                            "required_before_execution": True,
                        },
                        receipt["backup_evidence"],
                    )
                    self.assertEqual(
                        {"status", "steps"}, set(receipt["recovery"])
                    )
                    self.assertEqual(
                        "operator_must_verify", receipt["recovery"]["status"]
                    )
                    self.assertTrue(receipt["recovery"]["steps"])
                    self.assertEqual(
                        receipt["recovery"], receipt["recovery_steps"]
                    )
                    self.assertEqual(
                        {"status", "conditions"},
                        set(receipt["stop_conditions"]),
                    )
                    self.assertEqual(
                        "operator_must_verify",
                        receipt["stop_conditions"]["status"],
                    )
                    self.assertTrue(receipt["stop_conditions"]["conditions"])

                output = gate.hook_output(receipt)
                reason = output["hookSpecificOutput"][
                    "permissionDecisionReason"
                ]
                self.assertNotIn(command, json.dumps(output, sort_keys=True))
                self.assertLess(len(reason.encode("utf-8")), 10_000)

    def test_deny_confidence_distinguishes_signature_from_uncertainty(self):
        for command in (
            "/__mock__/bin/diskutil eraseDisk MOCK /__mock__/volume /dev/mock0",
            "/__mock__/bin/fdisk /dev/mock0",
            "/__mock__/bin/rm -rf /etc",
            "/__mock__/bin/rm -rf " + HOME + "/.ssh",
        ):
            with self.subTest(command=command):
                receipt = gate.classify_payload(
                    payload(command), project_dir=PROJECT, home_dir=HOME
                )
                self.assertEqual("deny", receipt["decision"])
                self.assertEqual("critical", receipt["risk"])
                self.assertEqual("high", receipt["confidence"])
                self.assertEqual(
                    "matched_explicit_high_impact_signature",
                    receipt["decision_basis"],
                )

        for command in ("source harmless-mock.sh", "eval printf ok"):
            with self.subTest(command=command):
                receipt = gate.classify_payload(
                    payload(command), project_dir=PROJECT, home_dir=HOME
                )
                self.assertEqual("deny", receipt["decision"])
                self.assertEqual("indeterminate", receipt["risk"])
                self.assertEqual("low", receipt["confidence"])
                self.assertEqual(
                    "inspection_uncertainty_fail_closed",
                    receipt["decision_basis"],
                )

    def test_all_git_commands_require_review(self):
        for command in (
            "git status --short",
            "git diff --stat",
            "git log -1",
            "git show MOCK_REF",
            "git clean -ndx",
        ):
            with self.subTest(command=command):
                receipt = gate.classify_payload(
                    payload(command), project_dir=PROJECT, home_dir=HOME
                )
                self.assertEqual("ask", receipt["decision"])
                self.assertIn("GIT_RUNTIME_EXTENSIONS", receipt["signals"])

    def test_simple_command_names_require_runtime_review(self):
        for command in ("pwd", "true", "false", "printf mock", "echo mock"):
            with self.subTest(command=command):
                receipt = gate.classify_payload(
                    payload(command), project_dir=PROJECT, home_dir=HOME
                )
                self.assertEqual("ask", receipt["decision"])
                self.assertIn(
                    "RUNTIME_RESOLUTION_UNVERIFIED", receipt["signals"]
                )

    def test_sensitive_derived_fields_are_opaque(self):
        target_sentinel = "ALPHALAB_SENTINEL_SECRET_token_abc"
        cwd_sentinel = "CLIENT_SENTINEL"
        command = "/__mock__/bin/rm -f /tmp/" + target_sentinel
        probe = payload(command)
        probe["cwd"] = "/external/" + cwd_sentinel + "/work"
        probe["permission_mode"] = "PERMISSION_SENTINEL_token_mode"
        receipt = gate.classify_payload(
            probe, project_dir=PROJECT, home_dir=HOME
        )
        serialized = json.dumps(gate.hook_output(receipt), sort_keys=True)

        for forbidden in (
            command,
            target_sentinel,
            "token_abc",
            cwd_sentinel,
            "/external/" + cwd_sentinel + "/work",
            "PERMISSION_SENTINEL",
            "token_mode",
        ):
            self.assertNotIn(forbidden, serialized)
        self.assertEqual("unknown", receipt["context"]["permission_mode"])
        self.assertTrue(receipt["context"]["cwd"].startswith("<external-cwd:sha256:"))
        self.assertTrue(
            all(target.startswith("<target:sha256:") for target in receipt["blast_radius"]["targets"])
        )

        executable_sentinel = "EXECUTABLE_SENTINEL_token_xyz"
        executable_command = "/__mock__/bin/" + executable_sentinel + " --mock"
        executable_receipt = gate.classify_payload(
            payload(executable_command), project_dir=PROJECT, home_dir=HOME
        )
        executable_output = json.dumps(
            gate.hook_output(executable_receipt), sort_keys=True
        )
        for forbidden in (executable_command, executable_sentinel, "token_xyz"):
            self.assertNotIn(forbidden, executable_output)
        self.assertTrue(
            any(
                item.startswith("executable-sha256:")
                for item in executable_receipt["uncertainties"]
            )
        )

    def test_exact_deny_output_contract(self):
        # This starts only the Python gate. It never executes tool_input.command.
        command = "/__mock__/bin/rm -rf " + PROJECT
        completed = subprocess.run(
            [sys.executable, str(GATE)],
            input=json.dumps(payload(command)),
            text=True,
            capture_output=True,
            timeout=1,
            check=False,
        )

        self.assertEqual(0, completed.returncode)
        self.assertEqual("", completed.stderr)

        output = json.loads(completed.stdout)
        self.assertEqual({"hookSpecificOutput", "systemMessage"}, set(output))

        specific = output["hookSpecificOutput"]
        self.assertEqual(
            {
                "hookEventName",
                "permissionDecision",
                "permissionDecisionReason",
            },
            set(specific),
        )
        self.assertEqual("PreToolUse", specific["hookEventName"])
        self.assertEqual("deny", specific["permissionDecision"])

        embedded = json.loads(specific["permissionDecisionReason"])
        self.assertEqual("bash_blast_radius_receipt", embedded["type"])
        self.assertEqual("deny", embedded["receipt"]["decision"])
        self.assertNotIn(command, completed.stdout)
        self.assertLess(len(specific["permissionDecisionReason"]), 10_000)

    def test_allow_output_has_no_system_message(self):
        receipt = gate.classify_payload(
            payload("# inert fixture only"), project_dir=PROJECT, home_dir=HOME
        )
        output = gate.hook_output(receipt)
        self.assertEqual({"hookSpecificOutput"}, set(output))
        self.assertEqual(
            "allow", output["hookSpecificOutput"]["permissionDecision"]
        )

    def test_ask_output_has_no_system_message(self):
        command = "python3 -c 'print(1)'"
        receipt = gate.classify_payload(
            payload(command), project_dir=PROJECT, home_dir=HOME
        )
        output = gate.hook_output(receipt)
        self.assertEqual({"hookSpecificOutput"}, set(output))
        self.assertEqual("ask", output["hookSpecificOutput"]["permissionDecision"])
        self.assertNotIn(command, json.dumps(output))

    def test_invalid_json_fails_closed_when_process_runs(self):
        completed = subprocess.run(
            [sys.executable, str(GATE)],
            input="{not-json",
            text=True,
            capture_output=True,
            timeout=1,
            check=False,
        )
        output = json.loads(completed.stdout)
        self.assertEqual(0, completed.returncode)
        self.assertEqual(
            "deny", output["hookSpecificOutput"]["permissionDecision"]
        )
        embedded = json.loads(
            output["hookSpecificOutput"]["permissionDecisionReason"]
        )
        self.assertEqual(["INVALID_STDIN_JSON"], embedded["receipt"]["signals"])

    def test_invalid_utf8_fails_closed_when_process_runs(self):
        completed = subprocess.run(
            [sys.executable, str(GATE)],
            input=b"\xff\xfe",
            capture_output=True,
            timeout=1,
            check=False,
        )
        output = json.loads(completed.stdout)
        self.assertEqual(0, completed.returncode)
        self.assertEqual(
            "deny", output["hookSpecificOutput"]["permissionDecision"]
        )


if __name__ == "__main__":
    unittest.main(verbosity=2)