From c15577ba31cd9ae4b246c3e625461be4fdfc3d12 Mon Sep 17 00:00:00 2001 From: rifkir23 Date: Sat, 10 Oct 2026 11:09:11 +0700 Subject: [PATCH] feat(harness): default-off risk evaluator in the permission engine (#245) Wire one strictness-only evaluator into PermissionEngine, in the shape agreed in #245 (action triage, Option A, rules only): - It runs only on calls the mode default (step 4) would allow, and its one move is allow -> ask, routed through _apply_approval_policy so `never` and headless runs end in a deny that carries the evaluator's reason. Sensitive-path protection, the read-only upper bound and explicit rules decide first and stay final. - The built-in evaluator reuses command_guard.screen_all (destructive screen always; egress/install under DEEPCODE_COMMAND_SCREEN=strict) instead of a new keyword scorer. - Default off. DEEPCODE_RISK_EVALUATOR=1 opts in via make_engine and build_permission_engine; with it off every decision and reason is unchanged. - Opinions are cached per call signature. An evaluator that raises or returns an unusable opinion fails closed to ask and is not cached. --- core/harness/permissions.py | 102 +++++++++- core/harness/policy.py | 4 + core/harness/risk_evaluator.py | 115 +++++++++++ docs/P1_SECURITY_BASE.md | 4 + tests/test_harness_risk_evaluator.py | 272 +++++++++++++++++++++++++++ 5 files changed, 493 insertions(+), 4 deletions(-) create mode 100644 core/harness/risk_evaluator.py create mode 100644 tests/test_harness_risk_evaluator.py diff --git a/core/harness/permissions.py b/core/harness/permissions.py index 4ef371ec8..d64f69965 100644 --- a/core/harness/permissions.py +++ b/core/harness/permissions.py @@ -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): @@ -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 @@ -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 @@ -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, @@ -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), @@ -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, ) diff --git a/core/harness/policy.py b/core/harness/policy.py index d46f98ecb..a268f253a 100644 --- a/core/harness/policy.py +++ b/core/harness/policy.py @@ -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__ = [ @@ -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, ) diff --git a/core/harness/risk_evaluator.py b/core/harness/risk_evaluator.py new file mode 100644 index 000000000..3b242809c --- /dev/null +++ b/core/harness/risk_evaluator.py @@ -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() diff --git a/docs/P1_SECURITY_BASE.md b/docs/P1_SECURITY_BASE.md index 631082fef..a5369d201 100644 --- a/docs/P1_SECURITY_BASE.md +++ b/docs/P1_SECURITY_BASE.md @@ -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、不能覆盖显式规则，属于建议性防御而非安全边界。 diff --git a/tests/test_harness_risk_evaluator.py b/tests/test_harness_risk_evaluator.py new file mode 100644 index 000000000..802ac8b4a --- /dev/null +++ b/tests/test_harness_risk_evaluator.py @@ -0,0 +1,272 @@ +"""Tests for the strictness-only risk evaluator in the permission engine (#245). + +Action triage, Option A: the evaluator runs only on a call the mode default +would allow, and its one move is ``allow`` -> ``ask``. It can never loosen a +``deny``, override an explicit rule, or touch the sensitive-path / read-only +boundaries. It is default off and advisory, not a security boundary. +""" + +from __future__ import annotations + +import sys +from collections.abc import Mapping +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from core.domain.execution_security import ApprovalPolicy +from core.harness.permissions import ( + PermissionDecision, + PermissionEngine, + PermissionMode, + PermissionRule, + make_engine, +) +from core.harness.policy import build_permission_engine +from core.harness.risk_evaluator import ( + CommandGuardRiskEvaluator, + default_risk_evaluator, +) + +ALLOW = PermissionDecision.ALLOW +ASK = PermissionDecision.ASK +DENY = PermissionDecision.DENY + + +def _decide(engine: PermissionEngine, tool: str, **args): + decision, _reason = engine.evaluate(tool, args) + return decision + + +class _Spy: + """Flags every call with a fixed opinion and counts how often it is asked.""" + + def __init__(self, opinion: object = "flagged") -> None: + self.opinion = opinion + self.calls = 0 + + def __call__(self, tool_name: str, arguments: Mapping[str, object]): + self.calls += 1 + if isinstance(self.opinion, BaseException): + raise self.opinion + return self.opinion + + +# ---- 1. the evaluator cannot loosen anything -------------------------------- + + +def test_sensitive_path_and_plan_deny_survive_an_allow_opinion(): + no_objection = _Spy(opinion=None) + + full_auto = PermissionEngine( + mode=PermissionMode.FULL_AUTO, + cwd="/home/u/proj", + risk_evaluator=no_objection, + ) + assert _decide(full_auto, "read_file", file_path="/home/u/.ssh/id_rsa") is DENY + + plan = PermissionEngine( + mode=PermissionMode.PLAN, cwd="/w", risk_evaluator=no_objection + ) + assert _decide(plan, "write_file", file_path="/w/a.py") is DENY + + +def test_explicit_rule_stays_final_in_either_direction(): + spy = _Spy() + engine = PermissionEngine( + mode=PermissionMode.FULL_AUTO, + rules=[ + PermissionRule("execute_bash", "rm -rf *", ALLOW), + PermissionRule("execute_bash", "git push *", DENY), + ], + cwd="/w", + risk_evaluator=spy, + ) + + assert _decide(engine, "execute_bash", command="rm -rf build") is ALLOW + assert _decide(engine, "execute_bash", command="git push origin main") is DENY + assert spy.calls == 0 + + +# ---- 2. the intended path --------------------------------------------------- + + +def test_full_auto_high_risk_call_asks(): + engine = PermissionEngine( + mode=PermissionMode.FULL_AUTO, + cwd="/w", + risk_evaluator=default_risk_evaluator(), + ) + + decision, reason = engine.evaluate("execute_bash", {"command": "rm -rf build"}) + + assert decision is ASK + assert reason == "risk evaluator flagged this call: recursive force remove (rm -rf)" + assert _decide(engine, "execute_bash", command="pytest -q") is ALLOW + + +def test_headless_never_policy_denies_with_the_evaluator_reason(): + engine = PermissionEngine( + mode=PermissionMode.FULL_AUTO, + cwd="/w", + approval_policy=ApprovalPolicy.NEVER, + risk_evaluator=default_risk_evaluator(), + ) + + decision, reason = engine.evaluate("execute_bash", {"command": "sudo reboot"}) + + assert decision is DENY + assert "privilege escalation (sudo)" in reason + assert "approval policy is never" in reason + + +# ---- 3. evaluator failure is contained and fails closed ---------------------- + + +def test_raising_or_garbage_evaluator_fails_closed_without_crashing(): + for broken in (_Spy(opinion=RuntimeError("boom")), _Spy(opinion=42), _Spy("")): + engine = PermissionEngine( + mode=PermissionMode.FULL_AUTO, cwd="/w", risk_evaluator=broken + ) + + decision, reason = engine.evaluate("execute_bash", {"command": "ls"}) + + assert decision is ASK, broken.opinion + assert "fail-closed" in reason + + +def test_failure_is_not_cached_for_the_signature(): + spy = _Spy(opinion=RuntimeError("transient")) + engine = PermissionEngine( + mode=PermissionMode.FULL_AUTO, cwd="/w", risk_evaluator=spy + ) + + engine.evaluate("execute_bash", {"command": "ls"}) + spy.opinion = None + decision, _ = engine.evaluate("execute_bash", {"command": "ls"}) + + assert decision is ALLOW + assert spy.calls == 2 + + +# ---- 4. no wasted calls ----------------------------------------------------- + + +def test_not_consulted_when_steps_one_to_three_or_mode_already_decided(): + spy = _Spy() + sensitive = PermissionEngine( + mode=PermissionMode.FULL_AUTO, cwd="/home/u/proj", risk_evaluator=spy + ) + sensitive.evaluate("read_file", {"file_path": "/home/u/.aws/credentials"}) + + read_only = PermissionEngine( + mode=PermissionMode.PLAN, + cwd="/w", + enforce_read_only=True, + risk_evaluator=spy, + ) + read_only.evaluate("write_file", {"file_path": "/w/a.py"}) + + default_ask = PermissionEngine( + mode=PermissionMode.DEFAULT, cwd="/w", risk_evaluator=spy + ) + default_ask.evaluate("execute_bash", {"command": "rm -rf build"}) + + assert spy.calls == 0 + + +def test_at_most_one_call_per_signature(): + spy = _Spy(opinion=None) + engine = PermissionEngine( + mode=PermissionMode.FULL_AUTO, cwd="/w", risk_evaluator=spy + ) + + for _ in range(5): + engine.evaluate("execute_bash", {"command": "pytest -q"}) + engine.evaluate("execute_bash", {"command": "pytest -x"}) + + assert spy.calls == 2 + + +# ---- 5. evasion fixtures (advisory, not a security boundary) ---------------- + + +def test_tokenised_evasions_are_flagged(): + engine = PermissionEngine( + mode=PermissionMode.FULL_AUTO, + cwd="/w", + risk_evaluator=CommandGuardRiskEvaluator(), + ) + + for command in ( + "rm -rf /tmp/x", + "rm -r -f /tmp/x", + "rm --recursive --force /tmp/x", + "echo ok && rm -rf /tmp/x", + "chmod 0777 /srv", + ): + assert _decide(engine, "execute_bash", command=command) is ASK, command + + +def test_known_gaps_are_documented_not_guarded(): + # Advisory only: an encoded payload is not decoded here. The sandbox, not + # this evaluator, is the boundary — this test pins that honest limit. + engine = PermissionEngine( + mode=PermissionMode.FULL_AUTO, + cwd="/w", + risk_evaluator=CommandGuardRiskEvaluator(), + ) + + encoded = "echo cm0gLXJmIC8= | base64 -d > run.sh" + assert _decide(engine, "execute_bash", command=encoded) is ALLOW + assert _decide(engine, "execute_bash", command="touch rm-rf-notes.txt") is ALLOW + + +# ---- 6. feature off: identical to today ------------------------------------- + + +def test_feature_off_decisions_and_reasons_are_identical(monkeypatch): + monkeypatch.delenv("DEEPCODE_RISK_EVALUATOR", raising=False) + calls = [ + ("read_file", {"file_path": "/home/u/proj/a.py"}), + ("write_file", {"file_path": "/home/u/proj/a.py"}), + ("execute_bash", {"command": "rm -rf /"}), + ("execute_bash", {"command": "pytest -q"}), + ("read_file", {"file_path": "/home/u/.ssh/id_rsa"}), + ] + + for mode in ("default", "plan", "full_auto"): + for policy in (ApprovalPolicy.ON_REQUEST, ApprovalPolicy.NEVER): + built = make_engine(mode, cwd="/home/u/proj", approval_policy=policy) + plain = PermissionEngine( + mode=PermissionMode(mode), + cwd="/home/u/proj", + approval_policy=policy, + ) + + assert built.risk_evaluator is None + for tool, args in calls: + assert built.evaluate(tool, args) == plain.evaluate(tool, args) + + +def test_env_flag_opts_in_the_builtin_evaluator(monkeypatch): + monkeypatch.setenv("DEEPCODE_RISK_EVALUATOR", "1") + + engine = make_engine("full_auto", cwd="/w") + + assert isinstance(engine.risk_evaluator, CommandGuardRiskEvaluator) + assert _decide(engine, "execute_bash", command="rm -rf /") is ASK + + +def test_runtime_engine_builder_follows_the_env_flag(monkeypatch): + monkeypatch.delenv("DEEPCODE_PERMISSION_MODE", raising=False) + monkeypatch.delenv("DEEPCODE_RISK_EVALUATOR", raising=False) + assert build_permission_engine(None, cwd="/w").risk_evaluator is None + + monkeypatch.setenv("DEEPCODE_RISK_EVALUATOR", "1") + engine = build_permission_engine(None, cwd="/w") + + assert isinstance(engine.risk_evaluator, CommandGuardRiskEvaluator) + assert _decide(engine, "execute_bash", command="rm -rf /") is ASK