policy.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/mcp/policy.py
 
"""
Purpose
-------
Policy Engine (Layer 3) for the ``oct-mcp`` six-layer pipeline.
Maintains the tool manifest and decides whether a given tool call is
permitted under the active MCP profile.
 
Responsibilities
----------------
- Define :data:`TOOL_MANIFEST` mapping tool names to :class:`ToolSpec`.
- Define :class:`McpPolicy` which enforces profile-based rules.
- Return structured :class:`PolicyDecision` objects (never raises for
  policy rejections - callers check the decision).
 
Diagnostics
-----------
Domain: MCP
Levels:
    L1 - errors: unknown tool, manifest inconsistency
    L2 - lifecycle: policy decision
    L3 - (none): the decision is reported at L2, enforcement detail at L4
    L4 - deep trace: profile enforcement details
 
Contracts
---------
- This module does not import ``click``, the MCP SDK, or any other
  ``oct.mcp.*`` module except ``config.py``.
- :meth:`McpPolicy.check` never raises for normal policy decisions.
  Only ``ValueError`` is raised for programming errors (e.g. an unknown
  profile string passed at construction time).
- In ``airgapped`` profile, ALL tool calls are denied regardless of
  the tool manifest.
- In ``strict`` profile, only tools with ``auto_approve=True`` and
  ``destructive=False`` may execute, and repo-level config is ignored.
 
Dependencies
------------
- oct.core.diagnostics (_dbg structured logger)
"""
 
from __future__ import annotations
 
from dataclasses import dataclass, field
from typing import Literal
 
from oct.core.diagnostics import _dbg
 
 
# ---------------------------------------------------------------------
# Types
# ---------------------------------------------------------------------
 
#: Valid MCP profile names (independent of git quality-gate profiles).
McpProfile = Literal["default", "strict", "airgapped"]
 
 
@dataclass(frozen=True)
class ToolSpec:
    """Metadata for one entry in the tool manifest.
 
    Attributes
    ----------
    cli
        The ``oct`` CLI subcommand string (e.g. ``"lint"``, ``"git status"``).
    permissions
        Set of permission tokens required by this tool (``"read"``,
        ``"exec"``). Phase 5A tools are read-only (``{"read"}`` or
        ``{"read", "exec"}``).
    destructive
        True if the tool can mutate the filesystem. All Phase 5A tools
        are ``False``; write tools will be ``True`` in Phase 5B.
    auto_approve
        True if the tool may execute without explicit user confirmation.
        All Phase 5A tools are ``True``.
    """
 
    cli: str
    permissions: frozenset[str] = field(default_factory=frozenset)
    destructive: bool = False
    auto_approve: bool = True
 
    def __post_init__(self) -> None:
        # Ensure permissions is always a frozenset even when passed as a plain set.
        object.__setattr__(self, "permissions", frozenset(self.permissions))
 
 
# ---------------------------------------------------------------------
# Tool manifest (Phase 5A - 9 read-only tools)
# ---------------------------------------------------------------------
 
TOOL_MANIFEST: dict[str, ToolSpec] = {
    # ------------------------------------------------------------------
    # Phase 5A - read-only tools (9)
    # ------------------------------------------------------------------
    "oct_lint":       ToolSpec(
        cli="lint",
        permissions=frozenset({"read"}),
        destructive=False,
        auto_approve=True,
    ),
    "oct_health":     ToolSpec(
        cli="health",
        permissions=frozenset({"read"}),
        destructive=False,
        auto_approve=True,
    ),
    "oct_skeleton":   ToolSpec(
        cli="export-skeleton",
        permissions=frozenset({"read"}),
        destructive=False,
        auto_approve=True,
    ),
    "oct_deps":       ToolSpec(
        cli="deps",
        permissions=frozenset({"read"}),
        destructive=False,
        auto_approve=True,
    ),
    "oct_test":       ToolSpec(
        cli="test",
        permissions=frozenset({"read", "exec"}),
        destructive=False,
        auto_approve=True,
    ),
    "oct_typecheck":  ToolSpec(
        cli="typecheck",
        permissions=frozenset({"read", "exec"}),
        destructive=False,
        auto_approve=True,
    ),
    "oct_diag":       ToolSpec(
        cli="diag",
        permissions=frozenset({"read"}),
        destructive=False,
        auto_approve=True,
    ),
    "oct_git_status": ToolSpec(
        cli="git",
        permissions=frozenset({"read"}),
        destructive=False,
        auto_approve=True,
    ),
    "oct_git_check":  ToolSpec(
        cli="git",
        permissions=frozenset({"read"}),
        destructive=False,
        auto_approve=True,
    ),
    # ------------------------------------------------------------------
    # Phase 5B - write tools (11) - all destructive=True, auto_approve=False
    # Blocked in strict + airgapped profiles; require confirm=True via HITL.
    # ------------------------------------------------------------------
    "oct_format":           ToolSpec(
        cli="format",
        permissions=frozenset({"read", "write"}),
        destructive=True,
        auto_approve=False,
    ),
    "oct_docs":             ToolSpec(
        cli="docs",
        permissions=frozenset({"read", "write"}),
        destructive=True,
        auto_approve=False,
    ),
    "oct_export_source":    ToolSpec(
        cli="export-source",
        permissions=frozenset({"read", "write"}),
        destructive=True,
        auto_approve=False,
    ),
    "oct_new":              ToolSpec(
        cli="new",
        permissions=frozenset({"write"}),
        destructive=True,
        auto_approve=False,
    ),
    "oct_scaffold":         ToolSpec(
        cli="scaffold",
        permissions=frozenset({"write"}),
        destructive=True,
        auto_approve=False,
    ),
    "oct_clean":            ToolSpec(
        cli="clean",
        permissions=frozenset({"write"}),
        destructive=True,
        auto_approve=False,
    ),
    "oct_install_hooks":    ToolSpec(
        cli="install-hooks",
        permissions=frozenset({"write"}),
        destructive=True,
        auto_approve=False,
    ),
    "oct_git_commit":       ToolSpec(
        cli="git",
        permissions=frozenset({"write"}),
        destructive=True,
        auto_approve=False,
    ),
    "oct_git_hooks_install": ToolSpec(
        cli="git",
        permissions=frozenset({"write"}),
        destructive=True,
        auto_approve=False,
    ),
    "oct_git_init":         ToolSpec(
        cli="git",
        permissions=frozenset({"write"}),
        destructive=True,
        auto_approve=False,
    ),
    "oct_git_changelog":    ToolSpec(
        cli="git",
        permissions=frozenset({"write"}),
        destructive=True,
        auto_approve=False,
    ),
}
 
 
# ---------------------------------------------------------------------
# Policy decision
# ---------------------------------------------------------------------
 
 
@dataclass
class PolicyDecision:
    """The result of a :meth:`McpPolicy.check` call.
 
    Attributes
    ----------
    allowed
        ``True`` iff the tool call should proceed.
    decision
        Human-readable decision string: ``"allowed"``, ``"denied"``,
        or ``"requires_confirmation"``.
    policy_rule
        Short identifier for the rule that produced this decision.
    reason
        Human-readable reason (included in audit records and error
        responses).
    """
 
    allowed: bool
    decision: str
    policy_rule: str
    reason: str
 
 
# ---------------------------------------------------------------------
# Policy engine
# ---------------------------------------------------------------------
 
 
class McpPolicy:
    """Stateless policy engine for the MCP tool manifest.
 
    Parameters
    ----------
    profile
        Active MCP profile. One of ``"default"``, ``"strict"``, or
        ``"airgapped"``.
    safe_mode
        If ``True``, all tool calls are denied regardless of profile.
        Mirrors ``McpConfig.safe_mode``.
    """
 
    def __init__(
        self,
        profile: McpProfile = "default",
        safe_mode: bool = False,
    ) -> None:
        func = "McpPolicy.__init__"
        if profile not in ("default", "strict", "airgapped"):
            raise ValueError(
                f"Invalid MCP profile: {profile!r}. "
                f"Must be one of: default, strict, airgapped"
            )
        self._profile: McpProfile = profile
        self._safe_mode: bool = safe_mode
        _dbg("MCP", func, f"profile={profile} safe_mode={safe_mode}", 2)
 
    @property
    def profile(self) -> McpProfile:
        """Active profile name."""
        return self._profile
 
    def check(self, tool_name: str) -> PolicyDecision:
        """Check whether *tool_name* is permitted under the active policy.
 
        Parameters
        ----------
        tool_name
            MCP tool name, e.g. ``"oct_lint"``.
 
        Returns
        -------
        :class:`PolicyDecision` with ``allowed=True`` or ``allowed=False``.
        Never raises.
        """
        func = "McpPolicy.check"
 
        # -- safe mode (hard override) -----------------------------------
        if self._safe_mode:
            _dbg("MCP", func, f"denied tool={tool_name} rule=safe_mode", 2)
            return PolicyDecision(
                allowed=False,
                decision="denied",
                policy_rule="safe_mode",
                # OCT-LINT: disable=no_hardcoded_secrets reason="Denial reason text shown to the caller, not a secret"
                reason="Server is running in safe mode (OCT_MCP_SAFE_MODE=1). "
                       "All tool calls are blocked.",
            )
 
        # -- airgapped profile (deny everything) -------------------------
        if self._profile == "airgapped":
            _dbg("MCP", func, f"denied tool={tool_name} rule=airgapped_profile", 2)
            return PolicyDecision(
                allowed=False,
                decision="denied",
                policy_rule="airgapped_profile",
                reason="The airgapped profile denies all tool calls. "
                       "Switch to 'default' or 'strict' to enable tools.",
            )
 
        # -- tool not in manifest ----------------------------------------
        spec = TOOL_MANIFEST.get(tool_name)
        if spec is None:
            _dbg("MCP", func, f"denied tool={tool_name} rule=unknown_tool", 1)
            return PolicyDecision(
                allowed=False,
                decision="denied",
                policy_rule="unknown_tool",
                reason=f"Tool {tool_name!r} is not in the tool manifest.",
            )
 
        # -- destructive tools blocked in strict profile -----------------
        if self._profile == "strict" and spec.destructive:
            _dbg("MCP", func, f"denied tool={tool_name} rule=strict_no_destructive", 2)
            return PolicyDecision(
                allowed=False,
                decision="denied",
                policy_rule="strict_no_destructive",
                reason=f"Tool {tool_name!r} is destructive and the 'strict' profile "
                       f"only allows read-only tools.",
            )
 
        # -- default: all manifest tools with auto_approve=True ----------
        _dbg("MCP", func, f"allowed tool={tool_name} profile={self._profile}", 4)
        return PolicyDecision(
            allowed=True,
            decision="allowed",
            policy_rule="manifest_auto_approve",
            reason=f"Tool {tool_name!r} is in the manifest with auto_approve=True.",
        )