safety.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/mcp/safety.py
 
"""
Purpose
-------
Safety Gate (Layer 2.5) for the ``oct-mcp`` six-layer pipeline.
Enforces the Human-in-the-Loop (HITL) consent protocol for all
write/destructive tool calls, and applies dry-run defaults.
 
Responsibilities
----------------
- Define :class:`SafetyDecision` — the structured result of a safety
  check (proceed, requires_confirmation, dry_run, reason).
- Define :class:`SafetyGate` with a single :meth:`SafetyGate.check`
  entry point consumed by :func:`oct.mcp.tools._run_tool_write`.
- Define :func:`_format_confirmation_request` which produces the
  structured ``[oct-mcp requires_confirmation] …`` response string
  returned to the AI host when confirmation is needed.
 
HITL Protocol (D-5B-1 — stateless)
-----------------------------------
No server-side pending state.
 
- **First call** (``confirm=False``, default): safety gate returns
  ``proceed=False``. The caller returns a
  ``[oct-mcp requires_confirmation] …`` string to the AI host; no
  execution occurs.
- **Second call** (``confirm=True``): safety gate returns
  ``proceed=True``; full execution proceeds.
 
The AI host re-submits the identical call with ``confirm=True`` after
the user approves in the host UI. There is no replay window to exploit
(T-08) because each call is evaluated independently.
 
Dry-run rules (D-5B-2, D-5B-5)
--------------------------------
1. Non-destructive tools → always ``proceed=True``, ``dry_run=False``.
2. Destructive, ``confirm=False`` → ``proceed=False``,
   ``requires_confirmation=True``.
3. ``oct_format``, ``confirm=True``, ``apply=False`` → ``proceed=True``,
   ``dry_run=True``.
4. ``oct_format``, ``confirm=True``, ``apply=True`` → ``proceed=True``,
   ``dry_run=False``.
5. Other destructive, ``confirm=True`` → ``proceed=True``,
   ``dry_run=False``.
   (Tools with their own ``dry_run`` field honour that field directly
   — the executor reads it from the validated args dict.)
 
Diagnostics
-----------
Domain: MCP
Levels:
    L2 — lifecycle: safety decision
    L4 — deep trace: field values
 
Contracts
---------
- This module does not import ``click``, the MCP SDK, or any
  ``oct.mcp.executor`` / ``oct.mcp.tools`` module (no cycles).
- :meth:`SafetyGate.check` never raises for valid input.
- The confirmation request string always starts with the sentinel
  ``[oct-mcp requires_confirmation]`` so callers can detect it via
  simple prefix matching.
"""
 
from __future__ import annotations
 
from dataclasses import dataclass
 
try:
    from oc_diagnostics import _dbg as _real_dbg
 
    def _dbg(*args, **kwargs) -> None:
        _real_dbg(*args, **kwargs)
except ImportError:  # pragma: no cover
    def _dbg(*args, **kwargs) -> None:
        return None
 
from oct.mcp.config import McpConfig
from oct.mcp.policy import ToolSpec
 
# Sentinel prefix for confirmation-required responses.
HITL_SENTINEL: str = "[oct-mcp requires_confirmation]"
 
# Per-tool risk descriptions shown in the confirmation request.
_TOOL_RISK_DESCRIPTIONS: dict[str, str] = {
    "oct_format":           "auto-fixes Python formatting in project files",
    "oct_docs":             "generates or removes documentation artifacts",
    "oct_export_source":    "writes consolidated source-code export files",
    "oct_new":              "creates a new Python source file",
    "oct_scaffold":         "creates a new Option C project directory tree",
    "oct_clean":            "removes _source_code-*, _skeleton_code-*, and __pycache__ directories",
    "oct_install_hooks":    "installs pre-commit hooks configuration",
    "oct_git_commit":       "commits staged changes to the git history",
    "oct_git_hooks_install":"installs git pre-commit hooks",
    "oct_git_init":         "initialises or enhances the git repository configuration",
    "oct_git_changelog":    "writes CHANGELOG.md from commit history",
}
 
 
# ---------------------------------------------------------------------
# Safety decision
# ---------------------------------------------------------------------
 
 
@dataclass
class SafetyDecision:
    """Result of a :meth:`SafetyGate.check` call.
 
    Attributes
    ----------
    proceed
        ``True`` → execution may continue; ``False`` → return the
        confirmation request to the AI host.
    requires_confirmation
        ``True`` when the AI host must re-submit the call with
        ``confirm=True`` before execution proceeds.
    dry_run
        ``True`` → executor should pass ``--dry-run`` (or equivalent)
        to the CLI subprocess.  Applies only to ``oct_format`` (where
        ``apply=False`` triggers dry-run even with ``confirm=True``).
        Other tools that have their own ``dry_run`` field manage it
        independently.
    reason
        Human-readable explanation forwarded to audit logs.
    """
 
    proceed: bool
    requires_confirmation: bool
    dry_run: bool
    reason: str
 
 
# ---------------------------------------------------------------------
# Confirmation request formatter
# ---------------------------------------------------------------------
 
 
def _format_confirmation_request(
    tool_name: str,
    spec: ToolSpec,
    args: dict,
) -> str:
    """Return the structured confirmation-required response string.
 
    The string always starts with :data:`HITL_SENTINEL` so callers can
    detect it with a prefix check.  The body describes the tool, its
    risk, and the args that would be applied.
 
    Parameters
    ----------
    tool_name
        MCP tool name (e.g. ``"oct_format"``).
    spec
        The :class:`~oct.mcp.policy.ToolSpec` for this tool.
    args
        Validated args dict (``model.model_dump()``).  Shown in the
        confirmation request so the user can review what will happen.
    """
    risk = _TOOL_RISK_DESCRIPTIONS.get(
        tool_name,
        "modifies the filesystem",
    )
    # Show args without the confirm key itself (it's always False here).
    display_args = {k: v for k, v in args.items() if k != "confirm"}
    args_text = ", ".join(f"{k}={v!r}" for k, v in display_args.items()) or "(defaults)"
    return (
        f"{HITL_SENTINEL} "
        f"Tool '{tool_name}' {risk}. "
        f"Args: {args_text}. "
        f"Re-call with confirm=True to proceed."
    )
 
 
# ---------------------------------------------------------------------
# Safety gate
# ---------------------------------------------------------------------
 
 
class SafetyGate:
    """HITL + dry-run enforcement gate for write tools.
 
    One instance is created per MCP server session (in ``server.py``)
    and stored in :class:`~oct.mcp.server.ServerState`.
 
    The gate is **stateless** — each :meth:`check` call is independent.
    There is no in-memory pending-confirmation queue.  This eliminates
    the T-08 replay-window attack surface.
 
    Parameters
    ----------
    config
        Active :class:`~oct.mcp.config.McpConfig`.  The gate reads
        ``config.dry_run_default_for_writes`` to decide the fallback
        dry-run behaviour.
    """
 
    def __init__(self, config: McpConfig) -> None:
        func = "SafetyGate.__init__"
        self._config = config
        _dbg("MCP", func, f"dry_run_default={config.dry_run_default_for_writes}", 2)
 
    # -- public API ------------------------------------------------------
 
    def check(
        self,
        tool_name: str,
        spec: ToolSpec,
        args: dict,
    ) -> SafetyDecision:
        """Evaluate the HITL gate and dry-run rules for *tool_name*.
 
        Parameters
        ----------
        tool_name
            MCP tool name (e.g. ``"oct_format"``).
        spec
            :class:`~oct.mcp.policy.ToolSpec` for *tool_name*.
        args
            Validated args dict (``model.model_dump()``).
 
        Returns
        -------
        :class:`SafetyDecision` — never raises.
        """
        func = "SafetyGate.check"
 
        # Rule 1: non-destructive tools always proceed without dry-run.
        if not spec.destructive:
            _dbg("MCP", func, f"non-destructive tool={tool_name} → proceed", 4)
            return SafetyDecision(
                proceed=True,
                requires_confirmation=False,
                dry_run=False,
                reason=f"Tool '{tool_name}' is non-destructive; no confirmation required.",
            )
 
        confirm = args.get("confirm", False)
 
        # Rule 2: destructive + confirm=False → requires confirmation.
        if not confirm:
            _dbg("MCP", func, f"destructive tool={tool_name} confirm=False → requires_confirmation", 2)
            return SafetyDecision(
                proceed=False,
                requires_confirmation=True,
                dry_run=False,
                reason=(
                    f"Tool '{tool_name}' is destructive and requires explicit confirmation. "
                    f"Re-call with confirm=True to proceed."
                ),
            )
 
        # Rules 3 & 4: oct_format double guard — apply=False → dry_run.
        if tool_name == "oct_format":
            apply = args.get("apply", False)
            if not apply:
                _dbg("MCP", func, "oct_format confirm=True apply=False → dry_run=True", 2)
                return SafetyDecision(
                    proceed=True,
                    requires_confirmation=False,
                    dry_run=True,
                    reason=(
                        "oct_format: apply=False so --dry-run will be passed. "
                        "Set apply=True together with confirm=True to write changes."
                    ),
                )
            _dbg("MCP", func, "oct_format confirm=True apply=True → proceed with --fix", 2)
            return SafetyDecision(
                proceed=True,
                requires_confirmation=False,
                dry_run=False,
                reason="oct_format: confirm=True and apply=True — will apply formatting changes.",
            )
 
        # Rule 5: all other destructive tools with confirm=True → proceed.
        _dbg("MCP", func, f"destructive tool={tool_name} confirm=True → proceed", 2)
        return SafetyDecision(
            proceed=True,
            requires_confirmation=False,
            dry_run=False,
            reason=f"Tool '{tool_name}': confirm=True — proceeding with execution.",
        )