LZH-YS1998
824 lines
34 KiB
Python
824 lines
34 KiB
Python
"""Permission helpers for Native Runtime V2."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import shlex
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from opc.core.config import PermissionsV2Config, get_opc_home
|
|
from opc.core.models import PermissionResolution, PermissionScope, RiskLevel, RuntimePermissionDecision
|
|
from opc.llm.retry import LLMRetryError, call_llm_json_with_retry
|
|
from opc.layer2_organization.data_acquisition_policy import (
|
|
ACQUISITION_SHELL_PREFIXES,
|
|
is_projection_scoped_acquisition_shell_command,
|
|
)
|
|
from opc.layer4_tools.registry import ToolDefinition
|
|
|
|
|
|
_DEFAULT_PATH_KEYS = (
|
|
"path",
|
|
"file_path",
|
|
"directory",
|
|
"working_directory",
|
|
"target_output_dir",
|
|
"workspace_path",
|
|
)
|
|
_DEFAULT_COMMAND_KEYS = ("command", "cmd")
|
|
_DEFAULT_URL_KEYS = ("url",)
|
|
_READ_ONLY_PREFIXES = {
|
|
"cat",
|
|
"echo",
|
|
"find",
|
|
"git diff",
|
|
"git log",
|
|
"git show",
|
|
"git status",
|
|
"head",
|
|
"ls",
|
|
"node -v",
|
|
"npm -v",
|
|
"pwd",
|
|
"python -V",
|
|
"python3 -V",
|
|
"rg",
|
|
"tail",
|
|
"wc",
|
|
}
|
|
_RISKY_SHELL_KEYWORDS = (
|
|
"curl ",
|
|
"wget ",
|
|
"invoke-webrequest",
|
|
"invoke-restmethod",
|
|
"mv ",
|
|
"cp ",
|
|
"rm ",
|
|
"del ",
|
|
"remove-item",
|
|
"git commit",
|
|
"git push",
|
|
"npm install",
|
|
"pip install",
|
|
"pnpm install",
|
|
"cargo test",
|
|
"pytest",
|
|
"tee ",
|
|
"sed -i",
|
|
">",
|
|
">>",
|
|
)
|
|
_ACQUISITION_SHELL_PREFIXES = {str(item) for item in ACQUISITION_SHELL_PREFIXES}
|
|
_ANY_GRANT_VALUE = "*"
|
|
|
|
|
|
class ToolPermissionResolver:
|
|
"""Runtime permission gate with persisted session/project/global grants."""
|
|
|
|
def __init__(
|
|
self,
|
|
config: PermissionsV2Config | None = None,
|
|
*,
|
|
store: Any = None,
|
|
runtime_session_id: str = "",
|
|
project_id: str = "default",
|
|
llm: Any | None = None,
|
|
) -> None:
|
|
self.config = config or PermissionsV2Config()
|
|
self.store = store
|
|
self.runtime_session_id = runtime_session_id
|
|
self.project_id = project_id or "default"
|
|
self.llm = llm
|
|
self._loaded = False
|
|
self._session_grants: set[tuple[str, str, str, str, str]] = set()
|
|
self._project_grants: set[tuple[str, str, str, str, str]] = set()
|
|
self._global_grants: set[tuple[str, str, str, str, str]] = set()
|
|
self._denial_counts: dict[str, int] = {}
|
|
|
|
async def warmup(self) -> None:
|
|
if self._loaded or not self.store or not hasattr(self.store, "list_runtime_permission_grants"):
|
|
self._loaded = True
|
|
return
|
|
session_rows = await self.store.list_runtime_permission_grants(
|
|
runtime_session_id=self.runtime_session_id or None,
|
|
scopes=["session"],
|
|
)
|
|
project_rows = await self.store.list_runtime_permission_grants(
|
|
project_id=self.project_id,
|
|
scopes=["project"],
|
|
)
|
|
global_rows = await self.store.list_runtime_permission_grants(scopes=["global"])
|
|
self._session_grants = {self._grant_key_from_row(row) for row in session_rows}
|
|
self._project_grants = {self._grant_key_from_row(row) for row in project_rows}
|
|
self._global_grants = {self._grant_key_from_row(row) for row in global_rows}
|
|
self._loaded = True
|
|
|
|
def _candidate_extractors(self) -> tuple[tuple[str, ...], tuple[str, ...], tuple[str, ...]]:
|
|
keys = [str(item or "").strip() for item in self.config.candidate_extractors if str(item or "").strip()]
|
|
if not keys:
|
|
keys = [*_DEFAULT_PATH_KEYS, *_DEFAULT_COMMAND_KEYS, *_DEFAULT_URL_KEYS]
|
|
path_keys = tuple(item for item in keys if item in _DEFAULT_PATH_KEYS)
|
|
command_keys = tuple(item for item in keys if item in _DEFAULT_COMMAND_KEYS)
|
|
url_keys = tuple(item for item in keys if item in _DEFAULT_URL_KEYS)
|
|
return (
|
|
path_keys or _DEFAULT_PATH_KEYS,
|
|
command_keys or _DEFAULT_COMMAND_KEYS,
|
|
url_keys or _DEFAULT_URL_KEYS,
|
|
)
|
|
|
|
def _grant_key_from_row(self, row: dict[str, Any]) -> tuple[str, str, str, str, str]:
|
|
tool_name = str(row.get("tool_name", "") or "").strip()
|
|
candidate = str(row.get("candidate", "") or "").strip()
|
|
metadata = dict(row.get("metadata", {}) or {})
|
|
sandbox_mode = str(metadata.get("sandbox_mode", "") or "").strip() or _ANY_GRANT_VALUE
|
|
allow_network = str(metadata.get("allow_network", "") or "").strip().lower() or _ANY_GRANT_VALUE
|
|
workspace_class = str(metadata.get("workspace_class", "") or "").strip() or _ANY_GRANT_VALUE
|
|
return self._grant_key(
|
|
tool_name,
|
|
candidate,
|
|
sandbox_mode=sandbox_mode,
|
|
allow_network=allow_network,
|
|
workspace_class=workspace_class,
|
|
)
|
|
|
|
def _grant_key(
|
|
self,
|
|
tool_name: str,
|
|
candidate: str,
|
|
*,
|
|
sandbox_mode: str,
|
|
allow_network: str,
|
|
workspace_class: str,
|
|
) -> tuple[str, str, str, str, str]:
|
|
normalized = candidate.strip() or _ANY_GRANT_VALUE
|
|
return (
|
|
tool_name,
|
|
normalized,
|
|
sandbox_mode.strip() or _ANY_GRANT_VALUE,
|
|
allow_network.strip().lower() or _ANY_GRANT_VALUE,
|
|
workspace_class.strip() or _ANY_GRANT_VALUE,
|
|
)
|
|
|
|
def _candidate(self, arguments: dict[str, Any] | None = None) -> str:
|
|
if not arguments:
|
|
return _ANY_GRANT_VALUE
|
|
path_keys, command_keys, url_keys = self._candidate_extractors()
|
|
for key in (*path_keys, *command_keys, *url_keys):
|
|
value = str(arguments.get(key, "") or "").strip()
|
|
if value:
|
|
return value
|
|
return _ANY_GRANT_VALUE
|
|
|
|
def _grant_context(self, task: Any = None) -> tuple[str, str, str]:
|
|
sandbox_mode = _ANY_GRANT_VALUE
|
|
allow_network = _ANY_GRANT_VALUE
|
|
workspace_class = _ANY_GRANT_VALUE
|
|
if task is None:
|
|
return sandbox_mode, allow_network, workspace_class
|
|
metadata = getattr(task, "metadata", {}) or {}
|
|
execution_context = dict(metadata.get("_execution_context", {}) or {})
|
|
sandbox = dict(execution_context.get("sandbox", {}) or {})
|
|
sandbox_mode = str(sandbox.get("mode", "") or "").strip() or _ANY_GRANT_VALUE
|
|
allow_network = str(bool(sandbox.get("allow_network", True))).lower()
|
|
workspace_root = (
|
|
str(execution_context.get("workspace_root", "") or "").strip()
|
|
or str(metadata.get("workspace_root", "") or "").strip()
|
|
or str(metadata.get("comms_workspace_root", "") or "").strip()
|
|
or str(metadata.get("target_output_dir", "") or "").strip()
|
|
)
|
|
workspace_class = "workspace" if workspace_root else "default"
|
|
return sandbox_mode, allow_network, workspace_class
|
|
|
|
@staticmethod
|
|
def _risk(value: Any, default: RiskLevel) -> RiskLevel:
|
|
try:
|
|
return RiskLevel(str(value or default.value))
|
|
except Exception:
|
|
return default
|
|
|
|
@staticmethod
|
|
def _looks_like_shell_tool(tool_name: str) -> bool:
|
|
return tool_name in {"shell_exec", "python_exec", "git_commit"}
|
|
|
|
def _normalized_tool_set(self, values: list[str]) -> set[str]:
|
|
return {str(item or "").strip() for item in values if str(item or "").strip()}
|
|
|
|
def _matches_path_rule(self, candidate: str, rules: list[str]) -> bool:
|
|
if not candidate or candidate == "*":
|
|
return False
|
|
raw = str(candidate).strip()
|
|
for rule in rules:
|
|
token = str(rule or "").strip()
|
|
if not token:
|
|
continue
|
|
if token == "*" or raw == token:
|
|
return True
|
|
try:
|
|
rule_path = Path(token).resolve()
|
|
candidate_path = Path(raw).resolve()
|
|
except Exception:
|
|
if raw.startswith(token.rstrip("\\/")):
|
|
return True
|
|
continue
|
|
if candidate_path == rule_path or rule_path in candidate_path.parents:
|
|
return True
|
|
return False
|
|
|
|
def _workspace_paths(self, task: Any = None) -> list[Path]:
|
|
roots: list[Path] = []
|
|
metadata = getattr(task, "metadata", {}) or {} if task else {}
|
|
for raw in (
|
|
str(metadata.get("workspace_root", "") or "").strip(),
|
|
str(metadata.get("comms_workspace_root", "") or "").strip(),
|
|
str(metadata.get("output_root", "") or "").strip(),
|
|
str(metadata.get("target_output_dir", "") or "").strip(),
|
|
):
|
|
if not raw:
|
|
continue
|
|
try:
|
|
path = Path(raw).resolve()
|
|
except Exception:
|
|
continue
|
|
if path not in roots:
|
|
roots.append(path)
|
|
try:
|
|
memory_root = (Path(get_opc_home()) / "memory").resolve()
|
|
if memory_root not in roots:
|
|
roots.append(memory_root)
|
|
except Exception:
|
|
pass
|
|
if not roots:
|
|
try:
|
|
roots.append(Path.cwd().resolve())
|
|
except Exception:
|
|
pass
|
|
return roots
|
|
|
|
def _path_decision(
|
|
self,
|
|
tool: ToolDefinition,
|
|
arguments: dict[str, Any] | None,
|
|
task: Any = None,
|
|
) -> RuntimePermissionDecision | None:
|
|
if not arguments:
|
|
return None
|
|
path_keys, _, _ = self._candidate_extractors()
|
|
candidate = ""
|
|
for key in path_keys:
|
|
value = str(arguments.get(key, "") or "").strip()
|
|
if value:
|
|
candidate = value
|
|
break
|
|
if not candidate:
|
|
return None
|
|
if self._matches_path_rule(candidate, self.config.denied_paths):
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.DENY,
|
|
scope=PermissionScope.ONCE,
|
|
risk_level=RiskLevel.HIGH,
|
|
rationale="Target path matches a denied runtime permission rule.",
|
|
source="permission_rules",
|
|
)
|
|
if self._matches_path_rule(candidate, self.config.allowed_paths):
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ALLOW,
|
|
scope=PermissionScope.PROJECT,
|
|
risk_level=RiskLevel.LOW,
|
|
rationale="Target path matches an explicit runtime allow rule.",
|
|
source="permission_rules",
|
|
)
|
|
try:
|
|
resolved = Path(candidate).resolve()
|
|
except Exception:
|
|
return None
|
|
if tool.read_only:
|
|
return None
|
|
for root in self._workspace_paths(task):
|
|
if resolved == root or root in resolved.parents:
|
|
return None
|
|
risk = RiskLevel.HIGH if self.config.sandbox_policy.treat_external_paths_as_high_risk else RiskLevel.MEDIUM
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ASK if self.config.fail_closed else PermissionResolution.DENY,
|
|
scope=PermissionScope.ONCE,
|
|
risk_level=risk,
|
|
rationale="Target path is outside the current runtime workspace roots.",
|
|
source="path_guard",
|
|
metadata={"candidate": candidate},
|
|
)
|
|
|
|
def _split_command_prefix(self, command: str) -> str:
|
|
text = str(command or "").strip()
|
|
if not text:
|
|
return ""
|
|
try:
|
|
parts = shlex.split(text, posix=os.name != "nt")
|
|
except Exception:
|
|
parts = text.split()
|
|
if not parts:
|
|
return ""
|
|
if len(parts) >= 2:
|
|
return f"{parts[0]} {parts[1]}".strip()
|
|
return parts[0]
|
|
|
|
def _matches_command_prefix(self, command: str, prefixes: list[str]) -> bool:
|
|
raw = str(command or "").strip()
|
|
prefix = self._split_command_prefix(raw)
|
|
candidates = {raw, prefix}
|
|
for item in prefixes:
|
|
token = str(item or "").strip()
|
|
if not token:
|
|
continue
|
|
if raw == token or prefix == token:
|
|
return True
|
|
if raw.startswith(f"{token} ") or prefix.startswith(f"{token} "):
|
|
return True
|
|
return False
|
|
|
|
def _split_shell_command_segments(self, command: str) -> list[list[str]]:
|
|
text = str(command or "").replace("\r\n", "\n").replace("\n", " ; ").strip()
|
|
if not text:
|
|
return []
|
|
try:
|
|
lexer = shlex.shlex(text, posix=os.name != "nt", punctuation_chars=";&|")
|
|
lexer.whitespace_split = True
|
|
lexer.commenters = ""
|
|
tokens = list(lexer)
|
|
except Exception:
|
|
try:
|
|
tokens = shlex.split(text, posix=os.name != "nt")
|
|
except Exception:
|
|
tokens = text.split()
|
|
|
|
segments: list[list[str]] = []
|
|
current: list[str] = []
|
|
for token in tokens:
|
|
if token in {"&&", "||", ";", "|", "&"}:
|
|
if current:
|
|
segments.append(current)
|
|
current = []
|
|
continue
|
|
current.append(token)
|
|
if current:
|
|
segments.append(current)
|
|
return segments
|
|
|
|
def _command_has_redirection(self, command: str) -> bool:
|
|
text = str(command or "").replace("\r\n", "\n").replace("\n", " ; ").strip()
|
|
if not text:
|
|
return False
|
|
try:
|
|
lexer = shlex.shlex(text, posix=os.name != "nt", punctuation_chars=";&|<>")
|
|
lexer.whitespace_split = True
|
|
lexer.commenters = ""
|
|
tokens = list(lexer)
|
|
except Exception:
|
|
return any(marker in text for marker in (">", "<"))
|
|
return any(token in {">", ">>", "<", "<<"} for token in tokens)
|
|
|
|
def _matches_safe_shell_prefix(self, command: str, prefixes: list[str]) -> bool:
|
|
cleaned = " ".join(str(command or "").split()).strip()
|
|
if not cleaned or self._command_has_redirection(cleaned):
|
|
return False
|
|
segments = self._split_shell_command_segments(cleaned)
|
|
if len(segments) != 1:
|
|
return False
|
|
return self._matches_command_prefix(" ".join(segments[0]).strip(), prefixes)
|
|
|
|
def _shell_ast_reason(self, command: str) -> tuple[RiskLevel, str] | None:
|
|
lowered = str(command or "").strip().lower()
|
|
if not lowered:
|
|
return None
|
|
if lowered in _READ_ONLY_PREFIXES or self._matches_command_prefix(lowered, list(_READ_ONLY_PREFIXES)):
|
|
return RiskLevel.LOW, "Command matches a read-only shell prefix."
|
|
for keyword in _RISKY_SHELL_KEYWORDS:
|
|
if keyword in lowered:
|
|
risk = RiskLevel.HIGH
|
|
if keyword in {"curl ", "wget ", "invoke-webrequest", "invoke-restmethod"} and self.config.sandbox_policy.treat_network_as_risky:
|
|
risk = RiskLevel.CRITICAL
|
|
return risk, f"Command contains risky shell operation `{keyword.strip()}`."
|
|
return RiskLevel.MEDIUM, "Shell AST classifier could not prove the command is read-only."
|
|
|
|
def _shell_decision(
|
|
self,
|
|
tool: ToolDefinition,
|
|
arguments: dict[str, Any] | None,
|
|
*,
|
|
task: Any = None,
|
|
) -> RuntimePermissionDecision | None:
|
|
if not self._looks_like_shell_tool(tool.name) or not arguments:
|
|
return None
|
|
_, command_keys, _ = self._candidate_extractors()
|
|
command = ""
|
|
for key in command_keys:
|
|
value = str(arguments.get(key, "") or "").strip()
|
|
if value:
|
|
command = value
|
|
break
|
|
if not command:
|
|
return None
|
|
projection_scoped_low_risk = is_projection_scoped_acquisition_shell_command(
|
|
command=command,
|
|
task=task,
|
|
working_directory=str(arguments.get("working_directory", "") or arguments.get("workdir", "") or "").strip(),
|
|
target_output_dir=str(getattr(task, "metadata", {}).get("target_output_dir", "") or "").strip() if task else "",
|
|
)
|
|
if projection_scoped_low_risk:
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ALLOW,
|
|
scope=PermissionScope.ONCE,
|
|
risk_level=RiskLevel.LOW,
|
|
rationale="Command matches a work-item-scoped acquisition prefix inside the assigned workspace.",
|
|
source="shell_prefix",
|
|
metadata={"candidate": command},
|
|
)
|
|
for pattern in self.config.dangerous_shell_patterns:
|
|
if pattern and re.search(pattern, command, flags=re.IGNORECASE):
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ASK,
|
|
scope=PermissionScope.ONCE,
|
|
risk_level=RiskLevel.CRITICAL,
|
|
rationale=f"Command matched dangerous shell pattern `{pattern}`.",
|
|
source="shell_pattern",
|
|
metadata={"candidate": command},
|
|
)
|
|
filtered_safe_prefixes = [
|
|
item for item in self.config.safe_shell_prefixes
|
|
if str(item or "").strip() not in _ACQUISITION_SHELL_PREFIXES
|
|
]
|
|
if self._matches_safe_shell_prefix(command, filtered_safe_prefixes):
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ALLOW,
|
|
scope=PermissionScope.ONCE,
|
|
risk_level=RiskLevel.LOW,
|
|
rationale="Command matches a safe shell prefix.",
|
|
source="shell_prefix",
|
|
metadata={"candidate": command},
|
|
)
|
|
if self._matches_command_prefix(command, self.config.ask_shell_prefixes):
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ASK,
|
|
scope=PermissionScope.ONCE,
|
|
risk_level=RiskLevel.MEDIUM,
|
|
rationale="Command matches an ask-first shell prefix.",
|
|
source="shell_prefix",
|
|
metadata={"candidate": command},
|
|
)
|
|
if self.config.shell_ast_validation:
|
|
risk, rationale = self._shell_ast_reason(command) or (RiskLevel.MEDIUM, "Shell command requires manual review.")
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ALLOW if risk == RiskLevel.LOW else PermissionResolution.ASK,
|
|
scope=PermissionScope.ONCE,
|
|
risk_level=risk,
|
|
rationale=rationale,
|
|
source="shell_ast",
|
|
metadata={"candidate": command},
|
|
)
|
|
if tool.requires_confirmation or self.config.fail_closed:
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ASK,
|
|
scope=PermissionScope.ONCE,
|
|
risk_level=RiskLevel.HIGH if not tool.read_only else RiskLevel.MEDIUM,
|
|
rationale="Shell command requires explicit approval under runtime_v2.",
|
|
source="shell_guard",
|
|
metadata={"candidate": command},
|
|
)
|
|
return None
|
|
|
|
def _candidate_matches(self, candidate: str, granted_candidate: str) -> bool:
|
|
if granted_candidate == _ANY_GRANT_VALUE:
|
|
return True
|
|
if candidate == granted_candidate:
|
|
return True
|
|
return candidate.startswith(granted_candidate.rstrip("\\/"))
|
|
|
|
@staticmethod
|
|
def _sandbox_rank(mode: str) -> int:
|
|
return {
|
|
"workspace-write": 1,
|
|
"elevated": 2,
|
|
"off": 3,
|
|
}.get(str(mode or "").strip().lower(), 0)
|
|
|
|
def _match_grant(self, grants: set[tuple[str, str, str, str, str]], tool_name: str, candidate: str, *, task: Any = None) -> bool:
|
|
if not grants:
|
|
return False
|
|
sandbox_mode, allow_network, workspace_class = self._grant_context(task)
|
|
for grant_tool, grant_candidate, grant_sandbox_mode, grant_allow_network, grant_workspace_class in grants:
|
|
if grant_tool != tool_name:
|
|
continue
|
|
if not self._candidate_matches(candidate, grant_candidate):
|
|
continue
|
|
if grant_sandbox_mode not in {_ANY_GRANT_VALUE, sandbox_mode}:
|
|
if not (
|
|
self.config.guardian.cache_upgrade_context
|
|
and self._sandbox_rank(sandbox_mode) >= self._sandbox_rank(grant_sandbox_mode)
|
|
):
|
|
continue
|
|
if grant_allow_network not in {_ANY_GRANT_VALUE, allow_network}:
|
|
continue
|
|
if grant_workspace_class not in {_ANY_GRANT_VALUE, workspace_class}:
|
|
continue
|
|
return True
|
|
return False
|
|
|
|
def _denial_memory_key(self, tool_name: str, arguments: dict[str, Any] | None) -> str:
|
|
return f"{tool_name}:{self._candidate(arguments)}"
|
|
|
|
def record_denial(self, tool_name: str, arguments: dict[str, Any] | None) -> None:
|
|
if not self.config.denial_memory.enabled:
|
|
return
|
|
key = self._denial_memory_key(tool_name, arguments)
|
|
self._denial_counts[key] = self._denial_counts.get(key, 0) + 1
|
|
|
|
def _repeat_denial_decision(self, tool_name: str, arguments: dict[str, Any] | None) -> RuntimePermissionDecision | None:
|
|
if not self.config.denial_memory.enabled:
|
|
return None
|
|
key = self._denial_memory_key(tool_name, arguments)
|
|
repeats = self._denial_counts.get(key, 0)
|
|
if repeats < max(1, self.config.denial_memory.repeat_threshold):
|
|
return None
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.DENY,
|
|
scope=PermissionScope.ONCE,
|
|
risk_level=RiskLevel.HIGH,
|
|
rationale="Repeated denial memory indicates this action should stop and ask for a new plan.",
|
|
source="denial_memory",
|
|
metadata={"repeated_denials": repeats},
|
|
)
|
|
|
|
def build_blocked_result(
|
|
self,
|
|
decision: RuntimePermissionDecision,
|
|
*,
|
|
tool_name: str,
|
|
arguments: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
action = "reject" if decision.resolution == PermissionResolution.DENY else "require_input"
|
|
candidate = self._candidate(arguments)
|
|
return {
|
|
"error": decision.rationale or f"Runtime permission blocked `{tool_name}`.",
|
|
"success": False,
|
|
"approval": {
|
|
"action": action,
|
|
"risk_level": decision.risk_level.value,
|
|
"policy_source": decision.source,
|
|
"scope": decision.scope.value,
|
|
"candidate": candidate,
|
|
"explanation": decision.rationale,
|
|
"metadata": dict(decision.metadata or {}),
|
|
},
|
|
"permission_context": {
|
|
"tool_name": tool_name,
|
|
"candidate": candidate,
|
|
"resolution": decision.resolution.value,
|
|
"risk_level": decision.risk_level.value,
|
|
"source": decision.source,
|
|
},
|
|
}
|
|
|
|
def predicted_decision(
|
|
self,
|
|
tool: ToolDefinition | None,
|
|
arguments: dict[str, Any] | None = None,
|
|
*,
|
|
task: Any = None,
|
|
) -> RuntimePermissionDecision:
|
|
if tool is None:
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ASK if self.config.fail_closed else PermissionResolution.DENY,
|
|
scope=PermissionScope.ONCE,
|
|
risk_level=RiskLevel.HIGH,
|
|
rationale="Unknown tool requires manual review.",
|
|
source="runtime_prediction",
|
|
)
|
|
tool_name = tool.name
|
|
candidate = self._candidate(arguments)
|
|
repeated_denial = self._repeat_denial_decision(tool_name, arguments)
|
|
if repeated_denial is not None:
|
|
return repeated_denial
|
|
if tool_name in self._normalized_tool_set(self.config.deny_tools):
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.DENY,
|
|
scope=PermissionScope.ONCE,
|
|
risk_level=RiskLevel.HIGH,
|
|
rationale="Tool is explicitly denied by runtime permission rules.",
|
|
source="permission_rules",
|
|
)
|
|
if self._match_grant(self._session_grants, tool_name, candidate, task=task):
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ALLOW,
|
|
scope=PermissionScope.SESSION,
|
|
risk_level=RiskLevel.LOW,
|
|
rationale="Allowed by runtime session grant.",
|
|
source="runtime_session_grant",
|
|
)
|
|
if self._match_grant(self._project_grants, tool_name, candidate, task=task):
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ALLOW,
|
|
scope=PermissionScope.PROJECT,
|
|
risk_level=RiskLevel.LOW,
|
|
rationale="Allowed by persisted project grant.",
|
|
source="runtime_project_grant",
|
|
)
|
|
if self._match_grant(self._global_grants, tool_name, candidate, task=task):
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ALLOW,
|
|
scope=PermissionScope.GLOBAL,
|
|
risk_level=RiskLevel.LOW,
|
|
rationale="Allowed by persisted global grant.",
|
|
source="runtime_global_grant",
|
|
)
|
|
if tool_name in self._normalized_tool_set(self.config.allow_tools):
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ALLOW,
|
|
scope=PermissionScope.PROJECT,
|
|
risk_level=RiskLevel.LOW,
|
|
rationale="Tool is explicitly allowed by runtime permission rules.",
|
|
source="permission_rules",
|
|
)
|
|
path_decision = self._path_decision(tool, arguments, task=task)
|
|
if path_decision is not None:
|
|
return path_decision
|
|
shell_decision = self._shell_decision(tool, arguments, task=task)
|
|
if shell_decision is not None:
|
|
return shell_decision
|
|
if tool.requires_confirmation:
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ASK,
|
|
scope=PermissionScope.ONCE,
|
|
risk_level=RiskLevel.MEDIUM,
|
|
rationale="Tool is marked as requiring confirmation.",
|
|
source="runtime_prediction",
|
|
metadata={"candidate": candidate},
|
|
)
|
|
if self.config.guardian.enabled and self.config.guardian.auto_allow_read_only and tool.read_only:
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ALLOW,
|
|
scope=PermissionScope.ONCE,
|
|
risk_level=RiskLevel.LOW,
|
|
rationale="Guardian pre-check marked the tool as deterministic read-only.",
|
|
source="guardian",
|
|
metadata={"candidate": candidate},
|
|
)
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ALLOW,
|
|
scope=PermissionScope.ONCE,
|
|
risk_level=RiskLevel.LOW,
|
|
rationale="No runtime permission warning triggered.",
|
|
source="runtime_prediction",
|
|
metadata={"candidate": candidate},
|
|
)
|
|
|
|
async def refine_decision(
|
|
self,
|
|
decision: RuntimePermissionDecision,
|
|
*,
|
|
tool: ToolDefinition | None,
|
|
arguments: dict[str, Any] | None,
|
|
task: Any = None,
|
|
) -> RuntimePermissionDecision:
|
|
if decision.resolution != PermissionResolution.ASK:
|
|
return decision
|
|
if not self.config.classifier_enabled or not self.config.llm_classifier_model or self.llm is None or tool is None:
|
|
return decision
|
|
payload = {
|
|
"tool_name": tool.name,
|
|
"arguments": arguments or {},
|
|
"candidate": self._candidate(arguments),
|
|
"project_id": getattr(task, "project_id", self.project_id),
|
|
"heuristic_rationale": decision.rationale,
|
|
}
|
|
def _validate_classifier(parsed: Any) -> str | None:
|
|
if not isinstance(parsed, dict):
|
|
return "Top-level response must be a JSON object."
|
|
try:
|
|
score_val = float(parsed.get("score", 0.5) or 0.5)
|
|
except (TypeError, ValueError):
|
|
return "`score` must be a number between 0 and 1."
|
|
if score_val < 0 or score_val > 1:
|
|
return "`score` must be a number between 0 and 1."
|
|
return None
|
|
|
|
try:
|
|
parsed = await call_llm_json_with_retry(
|
|
self.llm,
|
|
system=(
|
|
"You are a runtime permission classifier.\n"
|
|
"Return strict JSON with keys `score` and `reason`.\n"
|
|
"`score` is a float between 0 and 1 where 0 is clearly safe and 1 is clearly unsafe.\n"
|
|
"Classify file mutation, shell execution, path escape risk, and network side effects conservatively."
|
|
),
|
|
payload=payload,
|
|
task_type="quick_tasks",
|
|
validator=_validate_classifier,
|
|
label="runtime_permission_classifier",
|
|
)
|
|
except LLMRetryError:
|
|
return decision
|
|
score = float(parsed.get("score", 0.5) or 0.5)
|
|
reason = str(parsed.get("reason", "") or decision.rationale)
|
|
thresholds = self.config.classifier_thresholds
|
|
if score <= thresholds.allow:
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ALLOW,
|
|
scope=decision.scope,
|
|
risk_level=RiskLevel.LOW,
|
|
rationale=reason or "Classifier marked the action safe.",
|
|
source="llm_classifier",
|
|
metadata={**dict(decision.metadata or {}), "classifier_score": score},
|
|
)
|
|
if score >= thresholds.deny:
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.DENY,
|
|
scope=decision.scope,
|
|
risk_level=RiskLevel.HIGH,
|
|
rationale=reason or "Classifier marked the action unsafe.",
|
|
source="llm_classifier",
|
|
metadata={**dict(decision.metadata or {}), "classifier_score": score},
|
|
)
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ASK,
|
|
scope=decision.scope,
|
|
risk_level=RiskLevel.MEDIUM if score < thresholds.ask else RiskLevel.HIGH,
|
|
rationale=reason or decision.rationale,
|
|
source="llm_classifier",
|
|
metadata={**dict(decision.metadata or {}), "classifier_score": score},
|
|
)
|
|
|
|
def decision_from_result(
|
|
self,
|
|
tool_name: str,
|
|
arguments: dict[str, Any] | None,
|
|
result: dict[str, Any],
|
|
) -> RuntimePermissionDecision:
|
|
approval = dict(result.get("approval", {}) or {})
|
|
action = str(approval.get("action", "") or "").strip().lower()
|
|
if action in {"require_input", "escalate"}:
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ASK,
|
|
scope=PermissionScope.ONCE,
|
|
risk_level=self._risk(approval.get("risk_level"), RiskLevel.MEDIUM),
|
|
rationale=str(result.get("error", "") or "Awaiting explicit permission."),
|
|
source="approval_engine",
|
|
metadata=approval,
|
|
)
|
|
if action == "reject":
|
|
self.record_denial(tool_name, arguments)
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.DENY,
|
|
scope=PermissionScope.ONCE,
|
|
risk_level=self._risk(approval.get("risk_level"), RiskLevel.HIGH),
|
|
rationale=str(result.get("error", "") or "Permission denied."),
|
|
source="approval_engine",
|
|
metadata=approval,
|
|
)
|
|
human_reply = str(approval.get("human_reply") or result.get("human_reply") or "").strip().lower()
|
|
candidate = self._candidate(arguments)
|
|
grant = self._grant_key(
|
|
tool_name,
|
|
candidate,
|
|
sandbox_mode=_ANY_GRANT_VALUE,
|
|
allow_network=_ANY_GRANT_VALUE,
|
|
workspace_class=_ANY_GRANT_VALUE,
|
|
)
|
|
if human_reply == "approve_session":
|
|
self._session_grants.add(grant)
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ALLOW,
|
|
scope=PermissionScope.SESSION,
|
|
risk_level=RiskLevel.LOW,
|
|
rationale="Approved for this runtime session.",
|
|
source="human_escalation",
|
|
metadata=approval,
|
|
)
|
|
if human_reply == "always_project":
|
|
self._project_grants.add(grant)
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ALLOW,
|
|
scope=PermissionScope.PROJECT,
|
|
risk_level=RiskLevel.LOW,
|
|
rationale="Approved for this project.",
|
|
source="human_escalation",
|
|
metadata=approval,
|
|
)
|
|
if human_reply == "always_global":
|
|
self._global_grants.add(grant)
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ALLOW,
|
|
scope=PermissionScope.GLOBAL,
|
|
risk_level=RiskLevel.LOW,
|
|
rationale="Approved globally.",
|
|
source="human_escalation",
|
|
metadata=approval,
|
|
)
|
|
return RuntimePermissionDecision(
|
|
resolution=PermissionResolution.ALLOW,
|
|
scope=PermissionScope.ONCE,
|
|
risk_level=RiskLevel.LOW,
|
|
rationale="Tool execution allowed.",
|
|
source="approval_engine",
|
|
metadata=approval,
|
|
)
|