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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 32 additions & 4 deletions coworker/connectors/browser_automation.py
Original file line number Diff line number Diff line change
Expand Up @@ -324,9 +324,31 @@ def _snapshot(page, max_chars: int) -> dict[str, Any]:
}


def make_browser_automation_tools() -> list[Callable[..., Any]]:
def make_browser_automation_tools(
roots: Optional[list[Any]] = None,
) -> list[Callable[..., Any]]:
tools: list[Callable[..., Any]] = []

def _confine_upload_path(raw: str) -> tuple[Optional[Path], Optional[dict]]:
"""Resolve a caller-supplied upload source inside a granted session root.

browser_upload_file is kind=write → EXTERNAL, and Mode.AUTO returns full access
with no path check — so without this, a model-controlled `path` can exfiltrate
anything readable (secrets store, SSH keys) through a page file input. Any
granted root is fine (this is a read/exfil boundary, not a write). Mirrors
email outgoing-attachment confinement and the sibling browser_screenshot fix.
"""
allowed = [Path(r.path) for r in (roots or [])]
if not allowed:
return None, {"error": "no granted session directory for the upload"}
p = Path(raw).expanduser()
cand = (p if p.is_absolute() else allowed[0] / p).resolve()
if not any(cand.is_relative_to(root.resolve()) for root in allowed):
return None, {
"error": f"{cand} is outside the session's granted directories"
}
return cand, None

def browser_open_url(
url: str, wait_until: str = "domcontentloaded"
) -> dict[str, Any]:
Expand Down Expand Up @@ -476,8 +498,13 @@ def browser_select(target: str, value: str) -> dict[str, Any]:
)

def browser_upload_file(target: str, path: str) -> dict[str, Any]:
file_path = Path(path).expanduser().resolve()
if not file_path.exists():
# Confine BEFORE opening the browser (fail fast; hermetic tests assert without
# Playwright). Relative paths resolve against the primary granted root.
file_path, err = _confine_upload_path(path)
if err:
return err
assert file_path is not None
if not file_path.is_file():
return {"error": f"file not found: {file_path}"}
return _BROWSER.call(
"upload_file",
Expand All @@ -495,7 +522,8 @@ def browser_upload_file(target: str, path: str) -> dict[str, Any]:
browser_upload_file,
_schema(
"browser_upload_file",
"Upload a local file through a file input. Requires approval.",
"Upload a local file through a file input. The path must be inside a "
"granted session directory. Requires approval.",
{"target": {"type": "string"}, "path": {"type": "string"}},
["target", "path"],
),
Expand Down
4 changes: 3 additions & 1 deletion coworker/connectors/integration_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -525,7 +525,9 @@ def make_integration_tools(
enabled_tools: Optional[set[str]] = None,
roots: Optional[list[Any]] = None,
) -> list[Callable[..., Any]]:
tools: list[Callable[..., Any]] = make_browser_automation_tools()
# Browser tools need the session roots: an upload's source path must resolve inside
# a granted directory (AUTO mode has no path check for EXTERNAL tools).
tools: list[Callable[..., Any]] = make_browser_automation_tools(roots=roots)
# Email needs the session roots: attachment downloads land in the primary scratch
# and outgoing attachments must resolve inside a granted directory.
tools.extend(make_email_tools(secrets, roots=roots))
Expand Down
72 changes: 72 additions & 0 deletions tests/test_browser_upload_confine.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
"""browser_upload_file must confine the source path to a granted session root.

The tool is kind=write → EXTERNAL risk, and Mode.AUTO returns "full access" with no
path check — so a model-controlled path could upload ~/.config/coworker/secrets.json
(or an SSH key) through a page file input. Confinement runs before the browser opens,
so these cases assert without Playwright.
"""

from __future__ import annotations

from coworker.connectors.browser_automation import make_browser_automation_tools
from coworker.roots import RootDir


def _upload_tool(roots):
tools = make_browser_automation_tools(roots=roots)
return next(t for t in tools if t.__name__ == "browser_upload_file")


def test_upload_rejects_path_outside_granted_roots(tmp_path):
workspace = tmp_path / "scratch"
workspace.mkdir()
secrets = tmp_path / "secrets.json"
secrets.write_text('{"api_key": "sk-secret"}')

tool = _upload_tool([RootDir(path=workspace, writable=True)])
res = tool(target="#file", path=str(secrets))
assert "error" in res and "outside" in res["error"]
assert secrets.read_text() == '{"api_key": "sk-secret"}' # never read for upload


def test_upload_rejects_traversal_escape(tmp_path):
workspace = tmp_path / "scratch"
workspace.mkdir()
outside = tmp_path / "secret.txt"
outside.write_text("nope")
tool = _upload_tool([RootDir(path=workspace, writable=True)])
res = tool(target="#file", path="../secret.txt")
assert "error" in res and "outside" in res["error"]


def test_upload_rejects_when_no_roots(tmp_path):
f = tmp_path / "doc.txt"
f.write_text("x")
tool = _upload_tool([])
res = tool(target="#file", path=str(f))
assert "error" in res and "no granted" in res["error"]


def test_upload_allows_readable_root_not_only_writable(tmp_path):
"""Upload is a read/exfil boundary — a read-only granted root is still fine."""
ro = tmp_path / "ro"
ro.mkdir()
doc = ro / "doc.txt"
doc.write_text("ok")
tool = _upload_tool([RootDir(path=ro, writable=False)])
res = tool(target="#file", path=str(doc))
# Confinement passed; Playwright isn't installed here, so we get a browser setup
# error — not an "outside" / "no granted" rejection.
assert "outside" not in str(res.get("error", ""))
assert "no granted" not in str(res.get("error", ""))


def test_upload_accepts_path_inside_writable_root_then_reaches_browser(tmp_path):
workspace = tmp_path / "scratch"
workspace.mkdir()
doc = workspace / "doc.txt"
doc.write_text("ok")
tool = _upload_tool([RootDir(path=workspace, writable=True)])
res = tool(target="#file", path=str(doc))
assert "outside" not in str(res.get("error", ""))
assert "no granted" not in str(res.get("error", ""))