codec_consent.py

"""codec_consent — strict-consent gate for the chat + MCP skill paths.

Re-audit (red-team CHAIN-001/002/006): the Step-3 consent gate
(docs/PHASE1-STEP3-DESIGN.md §1.7) was wired ONLY into codec_agent_runner. The
chat ([SKILL:] tag + pre-LLM hijack → codec_dispatch.run_skill) and MCP
(codec_mcp.tool_fn) paths could reach high-power skills with only the
`is_dangerous` heuristic / path blocklists. This module is the shared
classifier + per-transport policy:

  - MCP  → hard-refuse destructive skills (claude.ai can't consent at the
           operator tier; consistent with the _HTTP_BLOCKED principle).
  - chat → require explicit confirmation (the handler returns consent_required;
           the user confirms; re-dispatch carries a token).
  - voice/agent → existing ask_user announce-and-listen (unchanged).

A skill is "destructive" if it declares `SKILL_DESTRUCTIVE = True`
(registry-AST-extracted — the extensible per-skill path, Decision C), OR is in
`codec_config._HTTP_BLOCKED`, OR is one of the known high-power built-ins below
(so coverage doesn't depend on regenerating the hash-pinned skill manifest).

Kill switch: `CONSENT_GATE_ENABLED=false`.
"""
import os

__all__ = ["gate_enabled", "is_destructive_skill", "chat_consent_ok",
           "mcp_refuse_message", "mcp_allowed"]

# Known high-power built-ins that are destructive but NOT in _HTTP_BLOCKED.
# (terminal / python_exec / process_manager / pm2_control / ax_control are
# already covered by the _HTTP_BLOCKED backstop.)
_DESTRUCTIVE_BUILTINS = frozenset({
    "file_ops",       # write/append/delete to the filesystem
    "file_write",     # writes files
    "imessage_send",  # sends messages as the user
    "pilot",          # drives a real browser session
    "skill_forge",    # writes a skill to disk (no review gate)
})

# Destructive skills that are nonetheless SAFE to expose over MCP because they
# self-guard. `file_write` is write-only (no read/delete/list), blocks the
# system tree + credential files + all of ~/.codec, restricts writes to $HOME
# or /tmp, caps size at 500 KB, and audits every write to ~/.codec/file_write.log.
# It was purpose-built for remote (claude.ai) callers — the blanket MCP refusal
# contradicted its own design and broke the Claude→CODEC→save-a-file flow.
# Allowing it here is the "scoped-safe write" the operator wants, WITHOUT opening
# delete/move (file_ops), shells (terminal/python_exec), messaging
# (imessage_send) or browser control (pilot), which all remain refused over MCP.
_MCP_SELF_GUARDED = frozenset({"file_write"})


def gate_enabled() -> bool:
    """Consent gate on by default; CONSENT_GATE_ENABLED=false disables it."""
    return os.environ.get("CONSENT_GATE_ENABLED", "true").lower() != "false"


def is_destructive_skill(tool_name, registry=None) -> bool:
    """True if `tool_name` is a high-power/destructive skill needing consent
    (chat) or refusal (MCP). Never raises."""
    if not tool_name:
        return False
    # 1) per-skill SKILL_DESTRUCTIVE flag (extensible — user skills opt in)
    try:
        reg = registry
        if reg is None:
            from codec_dispatch import registry as reg  # the singleton
        if reg is not None and reg.get_destructive(tool_name):
            return True
    except Exception:
        pass
    # 2) _HTTP_BLOCKED backstop (terminal, python_exec, process_manager, …)
    try:
        from codec_config import _HTTP_BLOCKED
        if tool_name in _HTTP_BLOCKED:
            return True
    except Exception:
        pass
    # 3) known high-power built-ins
    return tool_name in _DESTRUCTIVE_BUILTINS


def chat_consent_ok(tool_name, query, *, registry=None) -> bool:
    """Chat path (A2): a destructive skill requires explicit consent via the
    existing AskUserQuestion PWA panel (Phase 1 Step 3 §1.7 — literal verb-match;
    generic yes/ok rejected). Returns True if the skill may run (non-destructive,
    gate disabled, or consent granted); False if blocked (declined / timeout /
    ask_user unavailable). BLOCKS the worker thread on ask_user until answered —
    the chat handler invokes this via asyncio.to_thread, so the event loop isn't
    blocked. Fail-closed: any error → False (a destructive skill never
    auto-runs)."""
    if not gate_enabled() or not is_destructive_skill(tool_name, registry=registry):
        return True
    try:
        import codec_ask_user
        answer = codec_ask_user.ask(
            f"CODEC wants to run the '{tool_name}' skill — a destructive / "
            f"high-power operation — for: {(query or '')[:200]}",
            destructive=True,
            asked_from="chat",
            tool_name=tool_name,
        )
        return answer not in (
            codec_ask_user.TIMEOUT_SENTINEL,
            codec_ask_user.DISABLED_SENTINEL,
        )
    except Exception:
        return False


def mcp_allowed(tool_name, registry=None) -> bool:
    """May this skill run over MCP (remote claude.ai caller)?

    Non-destructive skills always may. A destructive skill may ONLY if it
    self-guards (see _MCP_SELF_GUARDED) — currently just file_write, which
    enforces its own path safety + audit. Everything else destructive is refused.
    Never raises; fail-closed (any error → not allowed)."""
    try:
        if not is_destructive_skill(tool_name, registry=registry):
            return True
        return tool_name in _MCP_SELF_GUARDED
    except Exception:
        return False


def mcp_refuse_message(tool_name) -> str:
    return (
        f"Skill '{tool_name}' is a destructive/high-power operation and is not "
        "permitted over MCP. Run it locally (chat or voice), where the operator "
        "can confirm it."
    )