permission_validator.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/permission_validator.py
 
"""
M8.1 WU-0a pre-dispatch decision fields + the M8.1 safety validator (§7.2; F7/F31).
 
Purpose
-------
The react-loop decides a tool call's fate **before** it dispatches (the classify →
decide → emit-started-for-allowed lifecycle). This module owns the two things that
decision needs and the loop must not: the **audit field set** each pre-dispatch
verdict carries onto its ``permission_decision`` event, and the **validator** that
says whether an M8.1-emitted decision is complete.
 
Two verdicts are pre-dispatch-decidable in M8.1 (§A.9(b), ratified):
 
- **cap-deny** — the WU-9a per-turn tool-call budget is spent. A budget/DoS verdict,
  so it carries the fail-closed ``risk_class="critical"`` sentinel directly and is
  **never** routed through :func:`~oc_chat.core.risk_class.assign_risk_class` (that
  function judges capability x target, which is not what a cap-deny is about).
- **auto-allow** — everything else. ``risk_class`` is derived from the tool's
  **capability alone**; the target-bearing derivation (protected-write / importable /
  human-confirm rungs) needs the resolved target and belongs to WU-0b.
 
Tier-2 confirmation, protected-write and external-content (F67) verdicts are computed
inside the tool bodies and stay **post-dispatch** until WU-0b. Their plan-173 relay
emission carries none of these fields, so :func:`is_m8_1_decision` discriminates the
two populations on ``policy_version`` presence and the validator only ever speaks for
the M8.1-emitted set.
 
Responsibilities
----------------
- :func:`tool_capability` — first-party tool name -> ``CapabilityClass`` verb.
- :func:`input_hash` — the argument-bound ``sha256`` hex digest.
- :func:`auto_allow_decision_fields` / :func:`cap_deny_decision_fields` — the kwargs
  for :func:`~oc_chat.core.events.make_permission_decision` (envelope ids excluded).
- :func:`is_m8_1_decision` — M8.1-emitted vs. legacy plan-173 relay decision.
- :func:`validate_m8_1_permission_decision` — the M8.1 safety validator.
- :func:`link_started_to_decision` — set ``permission_decision_id`` on an already-built
  ``tool_call_started`` envelope (the contract field exists; the factory does not take
  it and the WU-0a isolation forbids reshaping it).
 
Contracts
---------
- The validator is **pure**: it raises :class:`PermissionDecisionValidationError` or
  returns ``None``, and never mutates the payload.
- The grant / envelope discriminator is keyed on **fingerprint presence, NEVER on
  ``authorization_mode``** — the validator must not depend on that field's resolution.
- ``grant_fingerprint`` and ``envelope_fingerprint`` are **mutually exclusive**; a
  decision authorized by neither (auto-allow, cap-deny) is valid with both absent.
- Fail-closed: a tool with no declared capability yields no capability verb, and
  ``assign_risk_class`` maps that to its own ``"critical"`` sentinel.
- The contracts package is FROZEN here: this module calls the factory and reads the
  payload; it never reshapes either.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- option_c_agent_contracts (canonical_json — the shared serialization scheme)
- oc_chat.core.risk_class (assign_risk_class, the M8.0 WU-9b scorer)
 
Diagnostics
-----------
Domain: OC_CHAT.CORE
Levels:
    L2 — a validation failure
    L3 — per-decision verdict detail
    L4 — module load / deep tracing
"""
from __future__ import annotations
 
import hashlib
from typing import Any, Mapping, Optional
 
from oc_diagnostics import _dbg
from option_c_agent_contracts import canonical_json
 
from oc_chat.core.risk_class import assign_risk_class
 
_dbg("OC_CHAT.CORE", "module", "loaded", 4)
 
 
#: The active policy bundle identifier stamped on every M8.1-emitted decision.
#: M8.1 ships a single uniform bundle (§ "uniform rollout"), so this is one constant
#: rather than a resolver; WU-3's ``policy_version_mismatch`` compares against it, and
#: WU-6 replaces the constant with the signed bundle's own version when one exists.
POLICY_VERSION: str = "m8.1.0"
 
#: First-party tool name -> ``CapabilityClass`` verb. The registry is a deterministic
#: module allowlist (M8.0 WU-6 / G-2), so this map is closed over the first-party
#: surface; anything else (an entry-point plugin, a sub-registry, a test double) has no
#: declared capability and fails closed to ``"critical"`` via ``assign_risk_class``.
TOOL_CAPABILITIES: dict[str, str] = {
    # filesystem
    "list_dir":     "list",
    "search_files": "list",
    "read_file":    "read",
    "file_info":    "read",
    "read_chunks":  "read",
    "write_file":   "write",
    "edit_file":    "write",
    # memory
    "remember": "persist",
    "recall":   "read",
    # planner
    "create_plan": "persist",
    "mark_step":   "persist",
    "stop_plan":   "persist",
    "get_status":  "read",
    # shell
    "run_command": "exec",
    "text_search": "read",
    # web
    "fetch_url":  "egress_fetch",
    "web_search": "egress_fetch",
    # operator channel
    "ask_user": "read",
    # oct_* — each spawns the Option C toolchain as a subprocess
    "oct_lint":           "exec",
    "oct_health":         "exec",
    "oct_typecheck":      "exec",
    "oct_new":            "exec",
    "oct_test":           "exec",
    "oct_git_status":     "exec",
    "oct_git_check":      "exec",
    "oct_git_init":       "exec",
    "oct_git_add":        "exec",
    "oct_git_commit":     "exec",
    "oct_git_branch":     "exec",
    "oct_git_restore":    "exec",
    "oct_git_ignore":     "exec",
    "oct_git_commit_msg": "exec",
    "oct_git_changelog":  "exec",
    "oct_git_hooks":      "exec",
}
 
#: Exec tools whose argv is a fixed template (LD-1 ``templated``): the executable and
#: subcommand are module constants, only operands come from the model. ``run_command``
#: is deliberately absent — it takes a raw command string, so it scores the higher rung.
_TEMPLATED_EXEC_TOOLS: frozenset[str] = frozenset(
    name for name, cap in TOOL_CAPABILITIES.items() if cap == "exec" and name != "run_command"
)
 
#: The fail-closed ``RiskClass`` sentinel (§5.6 / DESIGN_FOUNDATION:25) a cap-deny
#: carries. A valid ``RiskClass`` member — not an invented string, and not ``None``.
CAP_DENY_RISK_CLASS: str = "critical"
 
#: The base set every M8.1-emitted decision must carry. ``risk_class`` is conditional
#: (see :func:`validate_m8_1_permission_decision`) and ``denial_reason_code`` is
#: required only on a denial, so both are handled separately.
_BASE_FIELDS: tuple[str, ...] = (
    "tool_call_id", "policy_version", "final_outcome", "input_hash",
)
 
#: Required iff ``grant_fingerprint`` is present.
_GRANT_FIELDS: tuple[str, ...] = (
    "permission_request_id", "canonical_request_hash", "grant_key_fingerprint",
)
 
#: Required iff ``envelope_fingerprint`` is present. An envelope is not bound to a
#: single request, so the request-scoped fields stay optional on that branch.
_ENVELOPE_FIELDS: tuple[str, ...] = ("grant_key_fingerprint",)
 
 
class PermissionDecisionValidationError(ValueError):
    """An M8.1-emitted ``permission_decision`` is missing a required field."""
 
 
def tool_capability(tool_name: str) -> str:
    """Return the ``CapabilityClass`` verb for *tool_name*, or ``""`` when undeclared.
 
    ``""`` is deliberately out-of-enum: :func:`assign_risk_class` maps any
    non-``CapabilityClass`` value to its fail-closed ``"critical"`` sentinel, so an
    unknown tool is scored as maximally risky rather than silently as ``"low"``.
    """
    func = "tool_capability"
    cap = TOOL_CAPABILITIES.get(tool_name, "")
    if not cap:
        _dbg("OC_CHAT.CORE", func, f"no declared capability for tool {tool_name!r}", 3)
    return cap
 
 
def input_hash(args: Mapping[str, Any]) -> str:
    """Return the ``sha256`` hex digest binding a decision to the call's arguments.
 
    Uses the shared canonical-JSON scheme (sorted keys, no insignificant whitespace,
    UTF-8) so the same arguments always hash the same. A value the scheme cannot
    serialize falls back to its ``repr`` rather than raising — an audit field must
    never be the reason a turn dies.
    """
    func = "input_hash"
    try:
        serialized = canonical_json(dict(args))
    except (TypeError, ValueError) as exc:
        _dbg("OC_CHAT.CORE", func, f"non-canonical args ({type(exc).__name__}); repr fallback", 2)
        serialized = repr(sorted(dict(args).items(), key=lambda kv: kv[0]))
    return hashlib.sha256(serialized.encode("utf-8")).hexdigest()
 
 
def auto_allow_decision_fields(*, tool_call_id: str, tool_name: str,
                               args: Mapping[str, Any]) -> dict[str, Any]:
    """Return the ``make_permission_decision`` kwargs for a pre-dispatch auto-allow.
 
    ``risk_class`` is derived from the tool's capability **alone** (no target): the
    target-bearing rungs need the resolved path/host/argv and land with WU-0b.
    """
    func = "auto_allow_decision_fields"
    capability = tool_capability(tool_name)
    risk_class = assign_risk_class(
        capability,
        templated=tool_name in _TEMPLATED_EXEC_TOOLS,
    )
    _dbg("OC_CHAT.CORE", func,
         f"allow tool={tool_name} capability={capability or '<undeclared>'} "
         f"risk_class={risk_class}", 3)
    return {
        "decision": "allow",
        "decided_by": "rule",
        "tool_call_id": tool_call_id,
        "policy_version": POLICY_VERSION,
        "risk_class": risk_class,
        "final_outcome": "allow",
        "input_hash": input_hash(args),
        "authorization_mode": "auto_allow",
    }
 
 
def cap_deny_decision_fields(*, tool_call_id: str, tool_name: str,
                             args: Mapping[str, Any]) -> dict[str, Any]:
    """Return the ``make_permission_decision`` kwargs for a per-turn-cap denial.
 
    A cap-deny is a budget/DoS verdict, **not** a capability x target judgement, so it
    carries :data:`CAP_DENY_RISK_CLASS` directly and never calls ``assign_risk_class``.
    """
    func = "cap_deny_decision_fields"
    _dbg("OC_CHAT.CORE", func,
         f"deny tool={tool_name} reason=tool_call_cap_exceeded "
         f"risk_class={CAP_DENY_RISK_CLASS}", 2)
    return {
        "decision": "deny",
        "decided_by": "rule",
        "tool_call_id": tool_call_id,
        "policy_version": POLICY_VERSION,
        "risk_class": CAP_DENY_RISK_CLASS,
        "final_outcome": "deny",
        "input_hash": input_hash(args),
        "authorization_mode": "deny",
        "denial_reason_code": "tool_call_cap_exceeded",
    }
 
 
def stall_deny_decision_fields(*, tool_call_id: str, tool_name: str,
                               args: Mapping[str, Any]) -> dict[str, Any]:
    """Return the ``make_permission_decision`` kwargs for a detected stall.
 
    A stall is a bounded tool-call DoS verdict and carries
    :data:`CAP_DENY_RISK_CLASS` for the same reason a cap-deny does, but it is
    NOT a cap: plan 248 gave the contracts leaf ``"stall_detected"`` so the two
    stop sharing one code. The operator-facing tool result and alert still carry
    the specific stall evidence; this field is what an audit reader counts.
    """
    func = "stall_deny_decision_fields"
    _dbg("OC_CHAT.CORE", func,
         f"deny tool={tool_name} reason=stall_detected "
         f"risk_class={CAP_DENY_RISK_CLASS}", 2)
    return {
        "decision": "deny",
        "decided_by": "rule",
        "tool_call_id": tool_call_id,
        "policy_version": POLICY_VERSION,
        "risk_class": CAP_DENY_RISK_CLASS,
        "final_outcome": "deny",
        "input_hash": input_hash(args),
        "authorization_mode": "deny",
        "denial_reason_code": "stall_detected",
    }
 
 
def is_m8_1_decision(payload: Any) -> bool:
    """True when *payload* is an M8.1-emitted decision (not a legacy plan-173 relay one).
 
    ``policy_version`` is the discriminator: every M8.1 emission stamps it, and the
    3-field relay emission at the post-dispatch confirmation point does not.
    """
    return getattr(payload, "policy_version", None) is not None
 
 
def validate_m8_1_permission_decision(payload: Any, *,
                                      has_capability_target: bool = True) -> None:
    """Raise unless *payload* satisfies the M8.1 decision contract (WU-0a item 3).
 
    Parameters
    ----------
    payload
        A ``permission_decision`` payload (or any object exposing its fields).
    has_capability_target : bool
        ``False`` only for a decision with no capability x target to score, which is
        the single documented waiver for ``risk_class``. Every runtime emission today
        has one, so the default requires it.
 
    Raises
    ------
    PermissionDecisionValidationError
        With every missing/conflicting field named, so an audit failure is actionable.
    """
    func = "validate_m8_1_permission_decision"
    problems: list[str] = []
 
    for field in _BASE_FIELDS:
        if getattr(payload, field, None) is None:
            problems.append(f"missing required field {field!r}")
 
    if has_capability_target and getattr(payload, "risk_class", None) is None:
        problems.append("missing required field 'risk_class' (a capability x target exists)")
 
    denied = (getattr(payload, "final_outcome", None) == "deny"
              or getattr(payload, "decision", None) == "deny")
    if denied and getattr(payload, "denial_reason_code", None) is None:
        problems.append("missing required field 'denial_reason_code' (the decision denied)")
 
    # The discriminator is fingerprint PRESENCE, never authorization_mode.
    grant_fp = getattr(payload, "grant_fingerprint", None)
    envelope_fp = getattr(payload, "envelope_fingerprint", None)
    if grant_fp is not None and envelope_fp is not None:
        problems.append(
            "grant_fingerprint XOR envelope_fingerprint violated: a decision is "
            "authorized by a grant or by an envelope, never both")
    if grant_fp is not None:
        problems.extend(f"missing required field {f!r} (grant_fingerprint is present)"
                        for f in _GRANT_FIELDS if getattr(payload, f, None) is None)
    if envelope_fp is not None:
        problems.extend(f"missing required field {f!r} (envelope_fingerprint is present)"
                        for f in _ENVELOPE_FIELDS if getattr(payload, f, None) is None)
 
    if problems:
        detail = "; ".join(problems)
        _dbg("OC_CHAT.CORE", func, f"invalid permission_decision: {detail}", 2, override=True)
        raise PermissionDecisionValidationError(
            f"M8.1 permission_decision is incomplete: {detail}")
    _dbg("OC_CHAT.CORE", func, "permission_decision valid", 4)
 
 
def link_started_to_decision(event: Any, permission_decision_id: Optional[str]) -> Any:
    """Return *event* with the envelope's ``permission_decision_id`` set.
 
    ``TurnEvent.permission_decision_id`` is an existing contract field, but
    ``make_tool_call_started`` does not expose it and the WU-0a isolation forbids
    reshaping the ``core.events`` factory signatures — so the linkage is applied to the
    built (frozen) envelope via pydantic's ``model_copy``. Returns *event* unchanged
    when there is no decision to link (a call whose verdict is not pre-dispatch
    decidable), so the caller needs no branch.
    """
    if permission_decision_id is None:
        return event
    return event.model_copy(update={"permission_decision_id": permission_decision_id})
 
 
__all__ = [
    "CAP_DENY_RISK_CLASS",
    "POLICY_VERSION",
    "TOOL_CAPABILITIES",
    "PermissionDecisionValidationError",
    "auto_allow_decision_fields",
    "cap_deny_decision_fields",
    "stall_deny_decision_fields",
    "input_hash",
    "is_m8_1_decision",
    "link_started_to_decision",
    "tool_capability",
    "validate_m8_1_permission_decision",
]