Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 98 additions & 4 deletions core/harness/permissions.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,11 @@
ExecutionPermissionRuleSnapshot,
normalize_permission_rules,
)
from core.harness.risk_evaluator import (
RiskEvaluator,
default_risk_evaluator,
risk_evaluator_enabled,
)


class PermissionDecision(str, Enum):
Expand Down Expand Up @@ -183,6 +188,14 @@ class PermissionEngine:
Skips an origin-specific confirmation gate after the shared policy has
allowed a call. This is enabled only by the canonical Full Access
preset; explicit permission rules still run first and can deny or ask.
risk_evaluator:
Optional, strictness-only advisor. When set, it runs *only* on calls
the mode default (step 4) would allow and may turn that ``allow`` into
``ask``; it can never loosen a ``deny``, override a rule, or touch the
sensitive-path / read-only boundaries. Off by default (``None``); the
caller wires in :func:`core.harness.risk_evaluator.default_risk_evaluator`
when ``DEEPCODE_RISK_EVALUATOR`` is enabled. A raising / misbehaving
evaluator is contained and fails closed to ``ask``.
"""

mode: PermissionMode = PermissionMode.DEFAULT
Expand All @@ -193,6 +206,10 @@ class PermissionEngine:
protect_sensitive_paths: bool = True
enforce_read_only: bool = False
bypass_origin_approval: bool = False
risk_evaluator: RiskEvaluator | None = None
_risk_cache: dict[tuple[str, str], str | None] = field(
default_factory=dict, repr=False, compare=False
)

def is_read_only(self, tool_name: str) -> bool:
known = _READ_ONLY_TOOLS | self.read_only_tools
Expand Down Expand Up @@ -365,22 +382,89 @@ def _evaluate(

if self.mode is PermissionMode.PLAN:
if read_only:
return PermissionDecision.ALLOW, "plan mode: read-only tool allowed"
return self._mode_default_allow(
tool_name,
arguments,
"plan mode: read-only tool allowed",
)
return (
PermissionDecision.DENY,
"plan mode: mutating tools are denied; only read-only "
"exploration is permitted until the plan is approved.",
)
if self.mode is PermissionMode.FULL_AUTO:
return PermissionDecision.ALLOW, "full_auto mode: no implicit gate"
return self._mode_default_allow(
tool_name, arguments, "full_auto mode: no implicit gate"
)
# DEFAULT
if read_only:
return PermissionDecision.ALLOW, "default mode: read-only tool allowed"
return self._mode_default_allow(
tool_name, arguments, "default mode: read-only tool allowed"
)
return self._apply_approval_policy(
PermissionDecision.ASK,
"default mode: mutating tool requires confirmation",
)

def _mode_default_allow(
self,
tool_name: str,
arguments: Mapping[str, object],
reason: str,
) -> tuple[PermissionDecision, str]:
"""Return the mode-default ``allow``, after an optional risk second look.

This is the single place the strictness-only risk evaluator plugs in:
a call that the mode default (step 4) would allow is offered to the
evaluator, which may raise it to ``ask`` (and thence to ``deny`` under a
``never`` approval policy, via :meth:`_apply_approval_policy`). The
sensitive-path denylist, the read-only upper bound, and explicit rules
have all already decided above, so the evaluator can only ever make the
engine *more* careful — never less.

With no evaluator wired in, this returns the plain ``allow`` unchanged,
so every decision and reason is identical to the engine without the
feature.
"""

if self.risk_evaluator is None:
return PermissionDecision.ALLOW, reason

risk_reason = self._risk_opinion(tool_name, arguments)
if risk_reason is None:
return PermissionDecision.ALLOW, reason
return self._apply_approval_policy(
PermissionDecision.ASK,
f"risk evaluator flagged this call: {risk_reason}",
)

def _risk_opinion(
self, tool_name: str, arguments: Mapping[str, object]
) -> str | None:
"""Ask the evaluator for an opinion, cached by call signature.

Failures are contained and fail closed: an evaluator that raises, or
returns anything other than ``None`` or a non-empty string, yields an
``ask`` reason rather than crashing the permission path or silently
allowing. The operator opted in to a second look, so a broken advisor
must not quietly waive it. Failures are not cached, so a transient
error does not stick to a call signature for the whole session.
"""

if self.risk_evaluator is None:
return None
signature = (tool_name, self._argument_string(tool_name, arguments))
if signature in self._risk_cache:
return self._risk_cache[signature]
try:
opinion = self.risk_evaluator(tool_name, arguments)
except Exception: # noqa: BLE001 - advisory evaluator must never crash the gate
return "evaluator error (fail-closed)"
if opinion is not None and not (isinstance(opinion, str) and opinion.strip()):
return "evaluator returned an unusable opinion (fail-closed)"
self._risk_cache[signature] = opinion
return opinion

def _apply_approval_policy(
self,
decision: PermissionDecision,
Expand Down Expand Up @@ -439,9 +523,18 @@ def make_engine(
approval_policy: ApprovalPolicy = ApprovalPolicy.ON_REQUEST,
protect_sensitive_paths: bool = True,
enforce_read_only: bool = False,
risk_evaluator: RiskEvaluator | None = None,
) -> PermissionEngine:
"""Convenience constructor from loosely-typed inputs (config/CLI)."""
"""Convenience constructor from loosely-typed inputs (config/CLI).

When ``risk_evaluator`` is left ``None`` the built-in rules-only evaluator
is wired in only if ``DEEPCODE_RISK_EVALUATOR`` is enabled; otherwise the
engine behaves exactly as before. Pass an explicit evaluator to override
this (e.g. in tests), or pass one while the flag is off to force it on.
"""
resolved_mode = mode if isinstance(mode, PermissionMode) else PermissionMode(mode)
if risk_evaluator is None and risk_evaluator_enabled():
risk_evaluator = default_risk_evaluator()
return PermissionEngine(
mode=resolved_mode,
rules=rules_from_config(rules_config),
Expand All @@ -450,4 +543,5 @@ def make_engine(
approval_policy=approval_policy,
protect_sensitive_paths=protect_sensitive_paths,
enforce_read_only=enforce_read_only,
risk_evaluator=risk_evaluator,
)
4 changes: 4 additions & 0 deletions core/harness/policy.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
PermissionMode,
rules_from_snapshots,
)
from core.harness.risk_evaluator import default_risk_evaluator, risk_evaluator_enabled
from core.harness.sandbox import sandbox_enabled

__all__ = [
Expand Down Expand Up @@ -208,4 +209,7 @@ def build_permission_engine(
bypass_origin_approval=(
profile.access_preset is ExecutionAccessPreset.FULL_ACCESS
),
# Default off; DEEPCODE_RISK_EVALUATOR=1 opts in the rules-only second
# look on calls the mode default would allow (#245).
risk_evaluator=default_risk_evaluator() if risk_evaluator_enabled() else None,
)
115 changes: 115 additions & 0 deletions core/harness/risk_evaluator.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,115 @@
"""Action-triage risk evaluation for tool calls — advisory, not a boundary.

The permission engine (:mod:`core.harness.permissions`) decides allow / ask /
deny from the product security profile: sensitive-path protection, the
read-only upper bound, explicit rules, and the mode default. This module adds
one *optional, strictness-only* opinion on top of that decision: given a tool
call the engine would otherwise ``allow``, does it look risky enough to be
worth a human glance?

Scope, settled in issue #245 (action triage, Option A, rules-only):

* It only ever sees calls the mode default (step 4) would allow, and its only
move is ``allow`` → ``ask``. It can never loosen a ``deny``, override an
explicit user rule, or reinterpret the sensitive-path / read-only
boundaries. "A risk evaluator can only make the engine more careful, never
less" — so ``full_auto``'s "rules only" promise degrades to "rules, plus a
soft ask when the evaluator objects", not a loss of the explicit rules.
* It is **default off**. With the feature disabled every decision and reason is
byte-identical to today's.
* v1 is **rules only** — no model call in the permission path. The built-in
evaluator reuses the screens :mod:`core.harness.command_guard` already has
(``screen_all`` → destructive-command always, egress / install under
``DEEPCODE_COMMAND_SCREEN=strict``) rather than inventing a new scorer. A
model-based evaluator, with its per-call latency and cost, would be a
separate proposal.

Honesty note: this is advisory defense-in-depth, **not** a security boundary.
The sandbox (:mod:`core.harness.sandbox`) is the boundary. An evaluator that
raises, times out, or returns garbage must never crash the permission path and
must never be read as permission; the engine contains those failures and fails
closed to ``ask`` (and so to ``deny`` under a ``never`` approval policy). Only
an operator who opted in ever sees that ask.
"""

from __future__ import annotations

import os
from collections.abc import Mapping
from typing import Protocol, runtime_checkable

from core.harness.command_guard import screen_all

# Environment flag that turns the evaluator on. Off unless explicitly set to a
# truthy value, matching the other opt-in harness switches (DEEPCODE_SANDBOX,
# DEEPCODE_COMPACTION_MEMORY, …).
_ENABLE_ENV = "DEEPCODE_RISK_EVALUATOR"

_TRUTHY = frozenset({"1", "true", "on", "yes"})

# Argument keys that carry a shell command, mirroring the permission engine's
# own shell-argument extraction so the evaluator screens exactly the string the
# user would run.
_COMMAND_ARG_KEYS = ("command", "cmd", "commands", "script")


def risk_evaluator_enabled() -> bool:
"""Whether the operator opted the risk evaluator in. Off by default."""

return os.environ.get(_ENABLE_ENV, "").strip().lower() in _TRUTHY


@runtime_checkable
class RiskEvaluator(Protocol):
"""An opinion on whether an otherwise-allowed call is worth confirming.

Returns a short human-readable reason when the call looks risky, or
``None`` when it has no objection. It returns an *opinion*; the engine
decides. It must not raise in normal use, but the engine contains it if it
does.
"""

def __call__(
self, tool_name: str, arguments: Mapping[str, object]
) -> str | None: ...


def _command_argument(arguments: Mapping[str, object]) -> str | None:
"""The shell command carried by ``arguments``, if any."""

for key in _COMMAND_ARG_KEYS:
value = arguments.get(key)
if isinstance(value, str) and value.strip():
return value
if isinstance(value, (list, tuple)):
joined = " ".join(str(v) for v in value)
if joined.strip():
return joined
return None


class CommandGuardRiskEvaluator:
"""Rules-only evaluator built on the existing command-guard screens.

It inspects only the shell command a call would run and defers entirely to
:func:`core.harness.command_guard.screen_all`, so the destructive-command
screen always applies and the egress / install screens apply under
``DEEPCODE_COMMAND_SCREEN=strict`` — the same waiver semantics the sandbox
front door already uses. Non-shell calls and calls without a command carry
no opinion (``None``).

Deterministic for a given command, so results are cached by the caller via
the call signature; this evaluator holds no state of its own.
"""

def __call__(self, tool_name: str, arguments: Mapping[str, object]) -> str | None:
command = _command_argument(arguments)
if command is None:
return None
return screen_all(command)


def default_risk_evaluator() -> RiskEvaluator:
"""The built-in rules-only evaluator used when the feature is enabled."""

return CommandGuardRiskEvaluator()
4 changes: 4 additions & 0 deletions docs/P1_SECURITY_BASE.md
Original file line number Diff line number Diff line change
Expand Up @@ -54,3 +54,7 @@ fail-closed:checker 或 approver 抛异常 → 一律拒绝。
冻结进 `ExecutionSecurityProfile`,worker 不再动态重读可变配置。
- 通用 Agent 循环、CLI 与 Desktop 已共享 Ask / Read only / Full access 预设
及同一 Approval 状态机;legacy mode 只保留给兼容调用。
- 可选风险评估(#245,默认关闭):`DEEPCODE_RISK_EVALUATOR=1` 时,对模式默认
会放行(步骤 4 的 `allow`)的调用复用 `command_guard.screen_all` 再看一眼,
命中只会 `allow` → `ask`(`never` 策略下即 deny);评估器异常按 fail-closed
处理。它不能放宽 deny、不能覆盖显式规则,属于建议性防御而非安全边界。
Loading
Loading