mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-09-10 19:57:44 +02:00
[eric] agents: extract the sensitive-path permission gate into manager/permissions/path_gate + 17-test suite
This commit is contained in:
@@ -52,6 +52,7 @@ from backend.apps.agents.manager import browser_dispatch
|
||||
from backend.apps.agents.manager import metadata
|
||||
from backend.apps.agents.manager.session.apply_context_window import apply_context_window
|
||||
from backend.apps.agents.manager.session import lifecycle
|
||||
from backend.apps.agents.manager.permissions import path_gate
|
||||
from backend.apps.agents.manager.session.workspace_git import _detect_git_identity, _ensure_cwd_git_repo
|
||||
from backend.apps.agents.manager.prompt.tool_catalog import (
|
||||
FULL_TOOLS,
|
||||
@@ -491,211 +492,6 @@ class AgentManager:
|
||||
def _default_for(tool_name: str) -> str:
|
||||
return _DEFAULTS.get(tool_name, "always_allow")
|
||||
|
||||
# Defense-in-depth path gate: flips Write/Edit/NotebookEdit from
|
||||
# always_allow to ask only for paths where a prompt-injected write
|
||||
# would grant the attacker persistence or credential exfil that the
|
||||
# user can't easily undo. Excludes routine dev artifacts (.env files,
|
||||
# generic app-support dirs); those triggered approvals on every
|
||||
# webapp build and contradicted the user's UI setting without
|
||||
# blocking any real attacker (a project .env doesn't auto-execute
|
||||
# and doesn't grant persistence; the threat is SSH auth keys, shell
|
||||
# rc files that auto-source on login, publish tokens, keychains,
|
||||
# and system dirs).
|
||||
import fnmatch as _fnmatch
|
||||
|
||||
# Each entry: pattern -> (short label, plain-English risk).
|
||||
# Multiple patterns can describe the same folder, but the user-
|
||||
# facing label/risk is what we show in the approval card, so it
|
||||
# has to read clearly to a non-developer who has no idea what
|
||||
# `~/.ssh/authorized_keys` is.
|
||||
_SENSITIVE_PATH_INFO: dict[str, tuple[str, str]] = {
|
||||
"*/.ssh": ("SSH folder (~/.ssh)", "Controls who can log in to your computer remotely."),
|
||||
"*/.ssh/*": ("SSH folder (~/.ssh)", "Controls who can log in to your computer remotely."),
|
||||
"*/.aws/*": ("AWS credentials (~/.aws)", "Cloud account access keys; can spend money and read your data."),
|
||||
"*/.config/gcloud/*": ("Google Cloud credentials", "Cloud account access; can spend money and read your data."),
|
||||
"*/.kube/*": ("Kubernetes config (~/.kube)", "Admin access to your Kubernetes clusters."),
|
||||
"*/.gnupg/*": ("GPG encryption keys", "Your private encryption keys; lets attackers decrypt your data or sign as you."),
|
||||
"*/.docker/config*": ("Docker credentials", "Login tokens for container registries."),
|
||||
"*/.zshrc": ("Shell startup file (.zshrc)", "Runs automatically every time you open a terminal."),
|
||||
"*/.bashrc": ("Shell startup file (.bashrc)", "Runs automatically every time you open a terminal."),
|
||||
"*/.bash_profile": ("Shell startup file (.bash_profile)", "Runs automatically every time you log in."),
|
||||
"*/.profile": ("Shell startup file (.profile)", "Runs automatically every time you log in."),
|
||||
"*/.zprofile": ("Shell startup file (.zprofile)", "Runs automatically every time you log in."),
|
||||
"*/.zshenv": ("Shell environment file (.zshenv)", "Runs automatically for every shell, including non-interactive ones."),
|
||||
"*/.gitconfig": ("Global Git config", "Affects every Git command you run; can hijack commits."),
|
||||
"*/.npmrc": ("npm auth file (~/.npmrc)", "Lets you publish npm packages; a token here can publish malicious packages as you."),
|
||||
"*/.pypirc": ("PyPI auth file (~/.pypirc)", "Lets you publish Python packages; a token here can publish malicious packages as you."),
|
||||
"*/.netrc": ("Stored login info (~/.netrc)", "Saved passwords for various services."),
|
||||
"*/Library/Keychains/*": ("macOS Keychain", "Where macOS stores all your saved passwords."),
|
||||
"/etc/*": ("System config (/etc)", "Affects the whole computer, not just your account."),
|
||||
"/private/etc/*": ("System config (/etc)", "Affects the whole computer, not just your account."),
|
||||
"/System/*": ("macOS system folder", "Affects the whole computer; should almost never be modified."),
|
||||
"/usr/local/etc/*": ("System config (/usr/local/etc)", "Affects the whole computer, not just your account."),
|
||||
}
|
||||
_SENSITIVE_PATH_PATTERNS = tuple(_SENSITIVE_PATH_INFO.keys())
|
||||
|
||||
def _match_sensitive_pattern(file_path: str) -> str | None:
|
||||
"""Return the matched sensitive pattern, or None if the path isn't
|
||||
sensitive OR if the user has previously trusted that pattern. The
|
||||
trusted list is reloaded on every call so a trust decision made
|
||||
during one approval takes effect for any later prompt fired in the
|
||||
same turn (no in-process cache to invalidate).
|
||||
"""
|
||||
if not file_path or not isinstance(file_path, str):
|
||||
return None
|
||||
try:
|
||||
norm = os.path.normpath(os.path.expanduser(file_path))
|
||||
except Exception:
|
||||
return None
|
||||
# Normalize to forward slashes so the patterns match on Windows
|
||||
# too, `os.path.normpath` produces backslashes on Windows
|
||||
# (`C:\Users\eric\.ssh\authorized_keys`), and fnmatch treats
|
||||
# `/` in the pattern as a literal character. Without this,
|
||||
# every sensitive-path gate would silently no-op on Windows
|
||||
# production and the prompt-injected `Write` to `~/.ssh/...`
|
||||
# would go through unchallenged.
|
||||
if os.sep != '/':
|
||||
norm = norm.replace(os.sep, '/')
|
||||
trusted = set(load_trusted_sensitive_paths())
|
||||
for pat in _SENSITIVE_PATH_PATTERNS:
|
||||
if pat in trusted:
|
||||
continue
|
||||
if _fnmatch.fnmatch(norm, pat):
|
||||
return pat
|
||||
return None
|
||||
|
||||
_PATH_GATED_TOOLS = ("Write", "Edit", "NotebookEdit")
|
||||
|
||||
# OS-level scheduling across macOS/Linux/Windows. Agent must
|
||||
# not install cron entries, launchd plists, Windows scheduled
|
||||
# tasks, or PowerShell ScheduledTask cmdlets behind the user's
|
||||
# back. Word-bounded so we don't flag stray strings in echo etc.
|
||||
import re as _re_sched
|
||||
_OS_SCHED_RE = _re_sched.compile(
|
||||
r"\b("
|
||||
r"crontab|launchctl|launchd|schtasks|systemd-run|"
|
||||
r"systemctl\s+--user.*timer|at\s+\d|at\s+now|at\s+-f|"
|
||||
# Windows PowerShell scheduled-task cmdlets:
|
||||
r"Register-ScheduledTask|New-ScheduledTask|Set-ScheduledTask|"
|
||||
r"Register-ScheduledJob|New-ScheduledJob"
|
||||
r")\b",
|
||||
_re_sched.IGNORECASE,
|
||||
)
|
||||
|
||||
def _looks_like_os_scheduling(tool_input) -> bool:
|
||||
if not isinstance(tool_input, dict):
|
||||
return False
|
||||
cmd = str(tool_input.get("command") or "")
|
||||
if not cmd:
|
||||
return False
|
||||
return bool(_OS_SCHED_RE.search(cmd))
|
||||
|
||||
# Catastrophic-path Bash gate. Bash is intentionally NOT in
|
||||
# _PATH_GATED_TOOLS because gating every `echo ... > /tmp/foo` would
|
||||
# interrupt routine work; but a single redirected write to one of
|
||||
# these paths can grant persistent attacker access (SSH keys,
|
||||
# sudoers, Keychain) or break the OS in ways the user can't
|
||||
# recover from. Scope is intentionally tighter than the Write/Edit
|
||||
# list because Bash is the agent's hot path and we'd rather miss
|
||||
# a borderline case than gate routine commands. The trust list is
|
||||
# shared with Write/Edit so a single "Always allow" decision in
|
||||
# the modal covers both surfaces.
|
||||
_BASH_CATASTROPHIC_INFO: dict[str, tuple[str, str]] = {
|
||||
"*/.ssh/*": ("SSH folder (~/.ssh)", "Controls who can log in to your computer remotely."),
|
||||
"/etc/sudoers": ("Sudo permissions (/etc/sudoers)", "Controls which commands can run with admin privileges."),
|
||||
"/etc/sudoers.d/*": ("Sudo permissions (/etc/sudoers.d)", "Controls which commands can run with admin privileges."),
|
||||
"/etc/passwd": ("System user list (/etc/passwd)", "Defines every user account on this computer."),
|
||||
"/etc/shadow": ("System password file (/etc/shadow)", "Stores password hashes for every user account."),
|
||||
"*/Library/Keychains/*": ("macOS Keychain", "Where macOS stores all your saved passwords."),
|
||||
"/System/*": ("macOS system folder", "Affects the whole computer; should almost never be modified."),
|
||||
}
|
||||
_BASH_CATASTROPHIC_PATTERNS = tuple(_BASH_CATASTROPHIC_INFO.keys())
|
||||
|
||||
# Token-extraction regex: pulls quoted strings AND bare path-like
|
||||
# tokens out of the Bash command so we can match against the
|
||||
# catastrophic list. Intentionally loose: false positives just
|
||||
# mean an extra approval prompt, never a missed gate.
|
||||
_BASH_PATH_TOKEN_RE = _re_sched.compile(
|
||||
r"""(?P<quoted>"[^"]+"|'[^']+')|(?P<bare>[~/.][\w./~\-]*)"""
|
||||
)
|
||||
|
||||
# Write operators we care about. Presence alone is not enough;
|
||||
# we also need a sensitive target in the same command. Includes
|
||||
# both shell redirection (`>`, `>>`, `tee`) and tools that take
|
||||
# an explicit destination flag (`cp`, `mv`, `dd of=`, `sed -i`,
|
||||
# `install`, `chmod`, `chown`, `rm`).
|
||||
_BASH_WRITE_OP_RE = _re_sched.compile(
|
||||
r"(?:>>?|\btee\b|\bsed\s+-i\b|\bcp\b|\bmv\b|\bdd\b[^|]*\bof=|\binstall\b|\bchmod\b|\bchown\b|\brm\b|\btouch\b|\bmkdir\b|\bln\b)",
|
||||
_re_sched.IGNORECASE,
|
||||
)
|
||||
|
||||
def _match_bash_catastrophic_pattern(command: str) -> str | None:
|
||||
"""Return the matched catastrophic-path pattern for a Bash
|
||||
command, or None if the command isn't writing to one (or the
|
||||
user has trusted that pattern). Same trust-list as
|
||||
_match_sensitive_pattern so toggling once covers both.
|
||||
"""
|
||||
if not command or not isinstance(command, str):
|
||||
return None
|
||||
if not _BASH_WRITE_OP_RE.search(command):
|
||||
return None
|
||||
trusted = set(load_trusted_sensitive_paths())
|
||||
for raw_match in _BASH_PATH_TOKEN_RE.finditer(command):
|
||||
tok = (raw_match.group("quoted") or raw_match.group("bare") or "")
|
||||
if tok and tok[0] in ("'", '"'):
|
||||
tok = tok[1:-1]
|
||||
if not tok:
|
||||
continue
|
||||
try:
|
||||
norm = os.path.normpath(os.path.expanduser(tok))
|
||||
except Exception:
|
||||
continue
|
||||
if os.sep != '/':
|
||||
norm = norm.replace(os.sep, '/')
|
||||
for pat in _BASH_CATASTROPHIC_PATTERNS:
|
||||
if pat in trusted:
|
||||
continue
|
||||
if _fnmatch.fnmatch(norm, pat):
|
||||
return pat
|
||||
return None
|
||||
|
||||
def _extract_target_path(tool_name: str, tool_input) -> str:
|
||||
if not isinstance(tool_input, dict):
|
||||
return ""
|
||||
if tool_name == "NotebookEdit":
|
||||
return str(tool_input.get("notebook_path") or "")
|
||||
return str(tool_input.get("file_path") or "")
|
||||
|
||||
def _maybe_override_policy(policy: str, tool_name: str, tool_input) -> tuple[str, str | None]:
|
||||
"""Returns (effective_policy, matched_sensitive_pattern).
|
||||
|
||||
Flips a permissive policy to 'ask' when the target path is
|
||||
sensitive (and not in the user's trusted allowlist). Defense
|
||||
in depth: even if the user has Write set to always_allow for
|
||||
productivity, a prompt-injected agent writing to
|
||||
~/.ssh/authorized_keys or ~/.zshrc gets surfaced for review;
|
||||
but once the user opts into "always allow files like this"
|
||||
for a given pattern, future writes to that pattern pass
|
||||
through silently.
|
||||
|
||||
Also: Bash invocations that look like OS-level scheduling
|
||||
(crontab, launchctl, schtasks, at, systemd-run --on-calendar)
|
||||
are flipped to 'ask' regardless of permission policy; we
|
||||
don't want the agent silently installing cron entries.
|
||||
"""
|
||||
if tool_name == "Bash" and _looks_like_os_scheduling(tool_input):
|
||||
return "ask", None
|
||||
if tool_name == "Bash" and isinstance(tool_input, dict):
|
||||
bash_match = _match_bash_catastrophic_pattern(str(tool_input.get("command") or ""))
|
||||
if bash_match:
|
||||
return "ask", bash_match
|
||||
if policy != "always_allow" or tool_name not in _PATH_GATED_TOOLS:
|
||||
return policy, None
|
||||
matched = _match_sensitive_pattern(_extract_target_path(tool_name, tool_input))
|
||||
if matched:
|
||||
return "ask", matched
|
||||
return policy, None
|
||||
|
||||
def _get_effective_policy(tool_name: str) -> str:
|
||||
"""Return 'always_allow', 'deny', or 'ask' for any tool. Keyed through
|
||||
the shared resolver so the read slot matches the write slot exactly."""
|
||||
@@ -740,10 +536,9 @@ class AgentManager:
|
||||
request_id = uuid4().hex
|
||||
label, why = (None, None)
|
||||
if sensitive_pattern:
|
||||
if sensitive_pattern in _SENSITIVE_PATH_INFO:
|
||||
label, why = _SENSITIVE_PATH_INFO[sensitive_pattern]
|
||||
elif sensitive_pattern in _BASH_CATASTROPHIC_INFO:
|
||||
label, why = _BASH_CATASTROPHIC_INFO[sensitive_pattern]
|
||||
described = path_gate.describe_sensitive_pattern(sensitive_pattern)
|
||||
if described:
|
||||
label, why = described
|
||||
approval_req = ApprovalRequest(
|
||||
id=request_id,
|
||||
session_id=session_id,
|
||||
@@ -819,7 +614,7 @@ class AgentManager:
|
||||
async def can_use_tool(tool_name, input_data, context):
|
||||
sensitive_pattern: str | None = None
|
||||
if tool_name != "AskUserQuestion":
|
||||
policy, sensitive_pattern = _maybe_override_policy(
|
||||
policy, sensitive_pattern = path_gate.maybe_override_policy(
|
||||
_get_effective_policy(tool_name), tool_name, input_data
|
||||
)
|
||||
if policy == "always_allow":
|
||||
@@ -917,7 +712,7 @@ class AgentManager:
|
||||
|
||||
if tool_name and tool_name != "AskUserQuestion":
|
||||
tool_input = input_data.get("tool_input", {})
|
||||
policy, sensitive_pattern = _maybe_override_policy(
|
||||
policy, sensitive_pattern = path_gate.maybe_override_policy(
|
||||
_get_effective_policy(tool_name), tool_name, tool_input
|
||||
)
|
||||
|
||||
|
||||
@@ -0,0 +1,192 @@
|
||||
"""Defense-in-depth permission gate, lifted verbatim out of the agent loop so it's
|
||||
independently testable. Flips a permissive tool policy to 'ask' when a write would land
|
||||
on a sensitive path (SSH keys, shell rc files, keychains, system dirs), when a Bash
|
||||
command writes to a catastrophic path, or when Bash looks like OS-level scheduling
|
||||
(crontab/launchctl/schtasks). The trusted-paths allowlist is reloaded on every call so a
|
||||
trust decision taken during one approval applies to any later prompt in the same turn."""
|
||||
|
||||
import fnmatch
|
||||
import os
|
||||
import re
|
||||
from typing import Dict, Optional, Tuple
|
||||
|
||||
from typeguard import typechecked
|
||||
|
||||
from backend.apps.tools_lib.tools_lib import load_trusted_sensitive_paths
|
||||
|
||||
# Each entry: pattern -> (short label, plain-English risk). The label/risk is what the
|
||||
# approval card shows, so it has to read clearly to a non-developer who has never heard
|
||||
# of `~/.ssh/authorized_keys`.
|
||||
P_SENSITIVE_PATH_INFO: Dict[str, Tuple[str, str]] = {
|
||||
"*/.ssh": ("SSH folder (~/.ssh)", "Controls who can log in to your computer remotely."),
|
||||
"*/.ssh/*": ("SSH folder (~/.ssh)", "Controls who can log in to your computer remotely."),
|
||||
"*/.aws/*": ("AWS credentials (~/.aws)", "Cloud account access keys; can spend money and read your data."),
|
||||
"*/.config/gcloud/*": ("Google Cloud credentials", "Cloud account access; can spend money and read your data."),
|
||||
"*/.kube/*": ("Kubernetes config (~/.kube)", "Admin access to your Kubernetes clusters."),
|
||||
"*/.gnupg/*": ("GPG encryption keys", "Your private encryption keys; lets attackers decrypt your data or sign as you."),
|
||||
"*/.docker/config*": ("Docker credentials", "Login tokens for container registries."),
|
||||
"*/.zshrc": ("Shell startup file (.zshrc)", "Runs automatically every time you open a terminal."),
|
||||
"*/.bashrc": ("Shell startup file (.bashrc)", "Runs automatically every time you open a terminal."),
|
||||
"*/.bash_profile": ("Shell startup file (.bash_profile)", "Runs automatically every time you log in."),
|
||||
"*/.profile": ("Shell startup file (.profile)", "Runs automatically every time you log in."),
|
||||
"*/.zprofile": ("Shell startup file (.zprofile)", "Runs automatically every time you log in."),
|
||||
"*/.zshenv": ("Shell environment file (.zshenv)", "Runs automatically for every shell, including non-interactive ones."),
|
||||
"*/.gitconfig": ("Global Git config", "Affects every Git command you run; can hijack commits."),
|
||||
"*/.npmrc": ("npm auth file (~/.npmrc)", "Lets you publish npm packages; a token here can publish malicious packages as you."),
|
||||
"*/.pypirc": ("PyPI auth file (~/.pypirc)", "Lets you publish Python packages; a token here can publish malicious packages as you."),
|
||||
"*/.netrc": ("Stored login info (~/.netrc)", "Saved passwords for various services."),
|
||||
"*/Library/Keychains/*": ("macOS Keychain", "Where macOS stores all your saved passwords."),
|
||||
"/etc/*": ("System config (/etc)", "Affects the whole computer, not just your account."),
|
||||
"/private/etc/*": ("System config (/etc)", "Affects the whole computer, not just your account."),
|
||||
"/System/*": ("macOS system folder", "Affects the whole computer; should almost never be modified."),
|
||||
"/usr/local/etc/*": ("System config (/usr/local/etc)", "Affects the whole computer, not just your account."),
|
||||
}
|
||||
P_SENSITIVE_PATH_PATTERNS: Tuple[str, ...] = tuple(P_SENSITIVE_PATH_INFO.keys())
|
||||
|
||||
P_PATH_GATED_TOOLS: Tuple[str, ...] = ("Write", "Edit", "NotebookEdit")
|
||||
|
||||
# OS-level scheduling across macOS/Linux/Windows. The agent must not install cron entries,
|
||||
# launchd plists, Windows scheduled tasks, or PowerShell ScheduledTask cmdlets behind the
|
||||
# user's back. Word-bounded so stray strings in echo etc. don't trip it.
|
||||
P_OS_SCHED_RE = re.compile(
|
||||
r"\b("
|
||||
r"crontab|launchctl|launchd|schtasks|systemd-run|"
|
||||
r"systemctl\s+--user.*timer|at\s+\d|at\s+now|at\s+-f|"
|
||||
r"Register-ScheduledTask|New-ScheduledTask|Set-ScheduledTask|"
|
||||
r"Register-ScheduledJob|New-ScheduledJob"
|
||||
r")\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
# Catastrophic-path Bash gate. Bash is intentionally NOT in P_PATH_GATED_TOOLS (gating
|
||||
# every `echo ... > /tmp/foo` would interrupt routine work), but a single redirected write
|
||||
# to one of these can grant persistent attacker access or break the OS unrecoverably. The
|
||||
# trust list is shared with Write/Edit so one "Always allow" covers both surfaces.
|
||||
P_BASH_CATASTROPHIC_INFO: Dict[str, Tuple[str, str]] = {
|
||||
"*/.ssh/*": ("SSH folder (~/.ssh)", "Controls who can log in to your computer remotely."),
|
||||
"/etc/sudoers": ("Sudo permissions (/etc/sudoers)", "Controls which commands can run with admin privileges."),
|
||||
"/etc/sudoers.d/*": ("Sudo permissions (/etc/sudoers.d)", "Controls which commands can run with admin privileges."),
|
||||
"/etc/passwd": ("System user list (/etc/passwd)", "Defines every user account on this computer."),
|
||||
"/etc/shadow": ("System password file (/etc/shadow)", "Stores password hashes for every user account."),
|
||||
"*/Library/Keychains/*": ("macOS Keychain", "Where macOS stores all your saved passwords."),
|
||||
"/System/*": ("macOS system folder", "Affects the whole computer; should almost never be modified."),
|
||||
}
|
||||
P_BASH_CATASTROPHIC_PATTERNS: Tuple[str, ...] = tuple(P_BASH_CATASTROPHIC_INFO.keys())
|
||||
|
||||
# Pulls quoted strings AND bare path-like tokens out of a Bash command. Intentionally loose:
|
||||
# a false positive just means an extra approval prompt, never a missed gate.
|
||||
P_BASH_PATH_TOKEN_RE = re.compile(
|
||||
r"""(?P<quoted>"[^"]+"|'[^']+')|(?P<bare>[~/.][\w./~\-]*)"""
|
||||
)
|
||||
|
||||
# Write operators we care about; presence alone isn't enough, a sensitive target in the
|
||||
# same command is also required. Covers shell redirection and tools with a destination flag.
|
||||
P_BASH_WRITE_OP_RE = re.compile(
|
||||
r"(?:>>?|\btee\b|\bsed\s+-i\b|\bcp\b|\bmv\b|\bdd\b[^|]*\bof=|\binstall\b|\bchmod\b|\bchown\b|\brm\b|\btouch\b|\bmkdir\b|\bln\b)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
@typechecked
|
||||
def match_sensitive_pattern(file_path: str) -> Optional[str]:
|
||||
"""The matched sensitive pattern, or None if the path isn't sensitive OR the user has
|
||||
trusted that pattern. The trusted list reloads per call so a same-turn trust decision
|
||||
takes effect immediately (no in-process cache to invalidate)."""
|
||||
if not file_path or not isinstance(file_path, str):
|
||||
return None
|
||||
try:
|
||||
norm = os.path.normpath(os.path.expanduser(file_path))
|
||||
except Exception:
|
||||
return None
|
||||
# Forward-slash the path so patterns match on Windows too; os.path.normpath emits
|
||||
# backslashes there and fnmatch treats '/' as literal. Without this the gate would
|
||||
# silently no-op on Windows and a prompt-injected Write to ~/.ssh/... would pass.
|
||||
if os.sep != '/':
|
||||
norm = norm.replace(os.sep, '/')
|
||||
trusted = set(load_trusted_sensitive_paths())
|
||||
for pat in P_SENSITIVE_PATH_PATTERNS:
|
||||
if pat in trusted:
|
||||
continue
|
||||
if fnmatch.fnmatch(norm, pat):
|
||||
return pat
|
||||
return None
|
||||
|
||||
|
||||
@typechecked
|
||||
def looks_like_os_scheduling(tool_input: object) -> bool:
|
||||
if not isinstance(tool_input, dict):
|
||||
return False
|
||||
cmd = str(tool_input.get("command") or "")
|
||||
if not cmd:
|
||||
return False
|
||||
return bool(P_OS_SCHED_RE.search(cmd))
|
||||
|
||||
|
||||
@typechecked
|
||||
def match_bash_catastrophic_pattern(command: str) -> Optional[str]:
|
||||
"""The matched catastrophic-path pattern for a Bash command, or None if it isn't writing
|
||||
to one (or the user has trusted it). Same trust-list as match_sensitive_pattern."""
|
||||
if not command or not isinstance(command, str):
|
||||
return None
|
||||
if not P_BASH_WRITE_OP_RE.search(command):
|
||||
return None
|
||||
trusted = set(load_trusted_sensitive_paths())
|
||||
for raw_match in P_BASH_PATH_TOKEN_RE.finditer(command):
|
||||
tok = (raw_match.group("quoted") or raw_match.group("bare") or "")
|
||||
if tok and tok[0] in ("'", '"'):
|
||||
tok = tok[1:-1]
|
||||
if not tok:
|
||||
continue
|
||||
try:
|
||||
norm = os.path.normpath(os.path.expanduser(tok))
|
||||
except Exception:
|
||||
continue
|
||||
if os.sep != '/':
|
||||
norm = norm.replace(os.sep, '/')
|
||||
for pat in P_BASH_CATASTROPHIC_PATTERNS:
|
||||
if pat in trusted:
|
||||
continue
|
||||
if fnmatch.fnmatch(norm, pat):
|
||||
return pat
|
||||
return None
|
||||
|
||||
|
||||
@typechecked
|
||||
def extract_target_path(tool_name: str, tool_input: object) -> str:
|
||||
if not isinstance(tool_input, dict):
|
||||
return ""
|
||||
if tool_name == "NotebookEdit":
|
||||
return str(tool_input.get("notebook_path") or "")
|
||||
return str(tool_input.get("file_path") or "")
|
||||
|
||||
|
||||
@typechecked
|
||||
def maybe_override_policy(policy: str, tool_name: str, tool_input: object) -> Tuple[str, Optional[str]]:
|
||||
"""Returns (effective_policy, matched_sensitive_pattern). Flips a permissive policy to
|
||||
'ask' when the target is a sensitive/catastrophic path or the Bash command looks like OS
|
||||
scheduling, even if the user set the tool to always_allow, so a prompt-injected agent
|
||||
writing to ~/.ssh/authorized_keys still gets surfaced. Once the user trusts a pattern,
|
||||
future writes to it pass through silently."""
|
||||
if tool_name == "Bash" and looks_like_os_scheduling(tool_input):
|
||||
return "ask", None
|
||||
if tool_name == "Bash" and isinstance(tool_input, dict):
|
||||
bash_match = match_bash_catastrophic_pattern(str(tool_input.get("command") or ""))
|
||||
if bash_match:
|
||||
return "ask", bash_match
|
||||
if policy != "always_allow" or tool_name not in P_PATH_GATED_TOOLS:
|
||||
return policy, None
|
||||
matched = match_sensitive_pattern(extract_target_path(tool_name, tool_input))
|
||||
if matched:
|
||||
return "ask", matched
|
||||
return policy, None
|
||||
|
||||
|
||||
@typechecked
|
||||
def describe_sensitive_pattern(pattern: str) -> Optional[Tuple[str, str]]:
|
||||
"""The (short label, plain-English risk) shown on the approval card for a matched
|
||||
pattern, from either table; None if the pattern is unknown."""
|
||||
if pattern in P_SENSITIVE_PATH_INFO:
|
||||
return P_SENSITIVE_PATH_INFO[pattern]
|
||||
if pattern in P_BASH_CATASTROPHIC_INFO:
|
||||
return P_BASH_CATASTROPHIC_INFO[pattern]
|
||||
return None
|
||||
@@ -0,0 +1,121 @@
|
||||
"""Comprehensive coverage for the defense-in-depth permission gate
|
||||
(manager/permissions/path_gate.py). This gate had ZERO isolated tests while it lived
|
||||
inside the 3000-line agent loop; pinning it here is the point of the extraction.
|
||||
|
||||
Every test runs with an empty trusted-paths allowlist by default (deterministic, no disk
|
||||
dependency); the trust tests opt a pattern in explicitly."""
|
||||
|
||||
import pytest
|
||||
|
||||
import backend.apps.agents.manager.permissions.path_gate as pg
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def empty_trusted(monkeypatch):
|
||||
monkeypatch.setattr(pg, "load_trusted_sensitive_paths", lambda: [])
|
||||
|
||||
|
||||
# ---- match_sensitive_pattern ------------------------------------------------
|
||||
|
||||
def test_sensitive_paths_are_flagged():
|
||||
assert pg.match_sensitive_pattern("/Users/eric/.ssh/authorized_keys") == "*/.ssh/*"
|
||||
assert pg.match_sensitive_pattern("/Users/eric/.zshrc") == "*/.zshrc"
|
||||
assert pg.match_sensitive_pattern("/Users/eric/.aws/credentials") == "*/.aws/*"
|
||||
assert pg.match_sensitive_pattern("/Users/eric/Library/Keychains/login.keychain-db") == "*/Library/Keychains/*"
|
||||
assert pg.match_sensitive_pattern("/etc/anything") == "/etc/*"
|
||||
|
||||
|
||||
def test_benign_paths_are_not_flagged():
|
||||
assert pg.match_sensitive_pattern("/Users/eric/project/main.py") is None
|
||||
assert pg.match_sensitive_pattern("") is None
|
||||
|
||||
|
||||
def test_trusted_pattern_is_skipped(monkeypatch):
|
||||
monkeypatch.setattr(pg, "load_trusted_sensitive_paths", lambda: ["*/.ssh/*"])
|
||||
assert pg.match_sensitive_pattern("/Users/eric/.ssh/authorized_keys") is None
|
||||
|
||||
|
||||
# ---- looks_like_os_scheduling ----------------------------------------------
|
||||
|
||||
def test_os_scheduling_detected():
|
||||
assert pg.looks_like_os_scheduling({"command": "crontab -e"}) is True
|
||||
assert pg.looks_like_os_scheduling({"command": "schtasks /create /tn evil"}) is True
|
||||
assert pg.looks_like_os_scheduling({"command": "Register-ScheduledTask -TaskName x"}) is True
|
||||
assert pg.looks_like_os_scheduling({"command": "launchctl load ~/Library/LaunchAgents/x.plist"}) is True
|
||||
|
||||
|
||||
def test_os_scheduling_ignores_benign_and_garbage():
|
||||
assert pg.looks_like_os_scheduling({"command": "echo hello"}) is False
|
||||
assert pg.looks_like_os_scheduling({"command": ""}) is False
|
||||
assert pg.looks_like_os_scheduling("not a dict") is False
|
||||
|
||||
|
||||
# ---- match_bash_catastrophic_pattern ---------------------------------------
|
||||
|
||||
def test_catastrophic_bash_writes_flagged():
|
||||
assert pg.match_bash_catastrophic_pattern("echo key >> ~/.ssh/authorized_keys") == "*/.ssh/*"
|
||||
assert pg.match_bash_catastrophic_pattern("cp evil /etc/sudoers") == "/etc/sudoers"
|
||||
assert pg.match_bash_catastrophic_pattern("printf x > /etc/shadow") == "/etc/shadow"
|
||||
|
||||
|
||||
def test_catastrophic_requires_a_write_operator():
|
||||
# reading a sensitive file is not a catastrophic WRITE
|
||||
assert pg.match_bash_catastrophic_pattern("cat ~/.ssh/id_rsa") is None
|
||||
assert pg.match_bash_catastrophic_pattern("echo hello world") is None
|
||||
|
||||
|
||||
# ---- extract_target_path ----------------------------------------------------
|
||||
|
||||
def test_extract_target_path():
|
||||
assert pg.extract_target_path("Write", {"file_path": "/a/b.py"}) == "/a/b.py"
|
||||
assert pg.extract_target_path("NotebookEdit", {"notebook_path": "/n.ipynb"}) == "/n.ipynb"
|
||||
assert pg.extract_target_path("Write", {}) == ""
|
||||
assert pg.extract_target_path("Write", "not a dict") == ""
|
||||
|
||||
|
||||
# ---- maybe_override_policy (the orchestrator) -------------------------------
|
||||
|
||||
def test_override_flips_always_allow_to_ask_on_sensitive_write():
|
||||
policy, matched = pg.maybe_override_policy("always_allow", "Write", {"file_path": "/Users/eric/.ssh/authorized_keys"})
|
||||
assert policy == "ask" and matched == "*/.ssh/*"
|
||||
|
||||
|
||||
def test_override_passes_benign_writes_through():
|
||||
assert pg.maybe_override_policy("always_allow", "Write", {"file_path": "/Users/eric/project/x.py"}) == ("always_allow", None)
|
||||
|
||||
|
||||
def test_override_flips_bash_os_scheduling_to_ask():
|
||||
assert pg.maybe_override_policy("always_allow", "Bash", {"command": "crontab -e"}) == ("ask", None)
|
||||
|
||||
|
||||
def test_override_flips_catastrophic_bash_to_ask():
|
||||
policy, matched = pg.maybe_override_policy("always_allow", "Bash", {"command": "echo x > /etc/sudoers"})
|
||||
assert policy == "ask" and matched == "/etc/sudoers"
|
||||
|
||||
|
||||
def test_override_leaves_ordinary_bash_alone():
|
||||
assert pg.maybe_override_policy("always_allow", "Bash", {"command": "ls -la"}) == ("always_allow", None)
|
||||
|
||||
|
||||
def test_override_does_not_touch_non_path_gated_tools():
|
||||
assert pg.maybe_override_policy("always_allow", "Read", {"file_path": "/Users/eric/.ssh/authorized_keys"}) == ("always_allow", None)
|
||||
|
||||
|
||||
def test_override_respects_non_permissive_policy_without_a_pattern():
|
||||
# a non-always_allow policy on a path-gated tool passes through untouched (no pattern)
|
||||
assert pg.maybe_override_policy("ask", "Write", {"file_path": "/Users/eric/.ssh/authorized_keys"}) == ("ask", None)
|
||||
|
||||
|
||||
def test_override_honors_trust(monkeypatch):
|
||||
monkeypatch.setattr(pg, "load_trusted_sensitive_paths", lambda: ["*/.ssh/*"])
|
||||
assert pg.maybe_override_policy("always_allow", "Write", {"file_path": "/Users/eric/.ssh/authorized_keys"}) == ("always_allow", None)
|
||||
|
||||
|
||||
# ---- describe_sensitive_pattern --------------------------------------------
|
||||
|
||||
def test_describe_sensitive_pattern():
|
||||
label, why = pg.describe_sensitive_pattern("*/.ssh/*")
|
||||
assert "SSH" in label and why
|
||||
label2, _ = pg.describe_sensitive_pattern("/etc/sudoers") # catastrophic table
|
||||
assert "Sudo" in label2
|
||||
assert pg.describe_sensitive_pattern("not-a-real-pattern") is None
|
||||
Reference in New Issue
Block a user