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
70 changes: 61 additions & 9 deletions coworker/connectors/browser_automation.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@

from __future__ import annotations

import os
import re
import tempfile
import threading
Expand All @@ -17,6 +18,42 @@

import aisuite as ai

from ..roots import RootDir


def _resolve_in_roots(
path: str, roots: Optional[list[RootDir]], *, need_write: bool
) -> tuple[Optional[Path], Optional[dict[str, Any]]]:
"""Resolve a local path and require it to sit inside a granted session root.

The permission engine only path-scopes the built-in ``WRITE_TOOLS`` by inspecting
``arguments["path"]``, so a connector tool that touches the filesystem is invisible to
it. These two do, in both directions: ``browser_upload_file`` reads a file and hands it
to whatever page is loaded, and ``browser_screenshot`` writes one. Without this check
they reach anything the OS lets the process reach — ``~/.ssh/id_rsa``, the secret store
itself — regardless of which folders the user actually granted the session.

Mirrors the confinement ``email_tools`` already applies to outgoing attachments.
Returns ``(resolved_path, None)`` on success or ``(None, error_dict)`` for the tool to
return verbatim.
"""
candidates = [r for r in (roots or []) if r.writable or not need_write]
if not candidates:
return None, {
"error": (
"no writable session directory is available"
if need_write
else "this session has no granted directories"
)
}
resolved = Path(str(path)).expanduser().resolve()
if not any(resolved.is_relative_to(r.path) for r in candidates):
verb = "writable " if need_write else ""
return None, {
"error": f"{path} is outside the session's {verb}directories"
}
return resolved, None


def _meta(
name: str, *, approval: bool = False, capabilities: Optional[list[str]] = None
Expand Down Expand Up @@ -324,7 +361,9 @@ 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[RootDir]] = None,
) -> list[Callable[..., Any]]:
tools: list[Callable[..., Any]] = []

def browser_open_url(
Expand Down Expand Up @@ -476,9 +515,11 @@ 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()
file_path, err = _resolve_in_roots(path, roots, need_write=False)
if err:
return err
if not file_path.exists():
return {"error": f"file not found: {file_path}"}
return {"error": f"file not found: {path}"}
return _BROWSER.call(
"upload_file",
lambda page: (
Expand Down Expand Up @@ -530,13 +571,24 @@ def run(page):
)

def browser_screenshot(path: str = "") -> dict[str, Any]:
def run(page):
out = (
Path(path).expanduser()
if path
else Path(tempfile.gettempdir()) / "coworker-browser-screenshot.png"
if path:
out, err = _resolve_in_roots(path, roots, need_write=True)
if err:
return err
else:
# A fixed name in the shared temp dir is two problems on a
# multi-user host (Linux /tmp; macOS gives each uid its own).
# Another user can pre-create the path as a symlink and the
# screenshot follows it, and the image itself — which may show a
# logged-in inbox or bank page — lands world-readable. mkstemp
# gives a random name created 0600 with O_EXCL.
fd, name = tempfile.mkstemp(
prefix="coworker-browser-screenshot.", suffix=".png"
)
out = out.resolve()
os.close(fd)
out = Path(name).resolve()

def run(page):
out.parent.mkdir(parents=True, exist_ok=True)
page.screenshot(path=str(out), full_page=True)
return {"ok": True, "path": str(out), "url": page.url}
Expand Down
2 changes: 1 addition & 1 deletion coworker/connectors/integration_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -525,7 +525,7 @@ 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()
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
165 changes: 165 additions & 0 deletions tests/test_browser_path_confinement.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,165 @@
"""The browser connector's two filesystem tools must stay inside the session's roots.

The permission engine path-scopes only the built-in ``WRITE_TOOLS``, and only by reading
``arguments["path"]`` (``permissions.PermissionEngine.evaluate``). A connector tool that
touches local files is therefore invisible to it, in both directions:

browser_upload_file reads any readable path and hands it to the loaded page
browser_screenshot writes a PNG to any writable path, creating parent dirs

``email_tools`` already confines outgoing attachments to the granted roots; these tests pin
the same rule for the browser tools.
"""

from pathlib import Path
import tempfile as tempfile_mod

import pytest

from coworker.connectors.browser_automation import (
_resolve_in_roots,
make_browser_automation_tools,
)
from coworker.roots import RootDir


def _tools(roots):
return {t.__name__: t for t in make_browser_automation_tools(roots=roots)}


@pytest.fixture
def session(tmp_path):
scratch = tmp_path / "scratch"
readonly = tmp_path / "readonly"
outside = tmp_path / "outside"
for d in (scratch, readonly, outside):
d.mkdir()
(readonly / "notes.txt").write_text("granted, read-only")
(outside / "id_rsa").write_text("PRIVATE KEY")
return {
"roots": [RootDir(path=scratch, writable=True),
RootDir(path=readonly, writable=False)],
"scratch": scratch,
"readonly": readonly,
"outside": outside,
}


# -- the helper ---------------------------------------------------------------

def test_read_is_allowed_from_any_granted_root(session):
resolved, err = _resolve_in_roots(
str(session["readonly"] / "notes.txt"), session["roots"], need_write=False)
assert err is None
assert resolved == (session["readonly"] / "notes.txt").resolve()


def test_write_is_refused_into_a_read_only_root(session):
resolved, err = _resolve_in_roots(
str(session["readonly"] / "shot.png"), session["roots"], need_write=True)
assert resolved is None
assert "outside the session's writable directories" in err["error"]


def test_ungranted_path_is_refused(session):
_, err = _resolve_in_roots(
str(session["outside"] / "id_rsa"), session["roots"], need_write=False)
assert err and "outside the session's directories" in err["error"]


def test_traversal_out_of_a_root_is_refused(session):
escape = str(session["scratch"] / ".." / "outside" / "id_rsa")
_, err = _resolve_in_roots(escape, session["roots"], need_write=False)
assert err, "`..` must be resolved before the containment check, not after"


def test_symlink_pointing_out_of_a_root_is_refused(session):
link = session["scratch"] / "innocent.txt"
try:
link.symlink_to(session["outside"] / "id_rsa")
except (OSError, NotImplementedError): # Windows without developer mode
pytest.skip("symlinks unavailable")
_, err = _resolve_in_roots(str(link), session["roots"], need_write=False)
assert err, "resolve() must follow the symlink before containment is decided"


def test_no_roots_means_no_filesystem_access(session):
_, err = _resolve_in_roots(str(session["outside"] / "id_rsa"), [], need_write=False)
assert err and "no granted directories" in err["error"]


# -- the tools ----------------------------------------------------------------

def test_upload_refuses_a_path_outside_the_session(session):
upload = _tools(session["roots"])["browser_upload_file"]
out = upload("input[type=file]", str(session["outside"] / "id_rsa"))
assert "error" in out and "outside the session" in out["error"]


def test_upload_refuses_before_reporting_whether_the_file_exists(session):
"""The refusal must not double as an existence oracle for ungranted paths."""
upload = _tools(session["roots"])["browser_upload_file"]
missing = upload("input", str(session["outside"] / "nope.txt"))
present = upload("input", str(session["outside"] / "id_rsa"))
# Same refusal for both: containment is decided before the file is stat'd, so the
# error never distinguishes "exists but ungranted" from "does not exist".
assert "outside the session" in missing["error"]
assert "outside the session" in present["error"]
assert "not found" not in missing["error"]


def test_screenshot_refuses_a_path_outside_the_session(session):
shot = _tools(session["roots"])["browser_screenshot"]
out = shot(str(session["outside"] / "evil.png"))
assert "error" in out and "outside the session" in out["error"]


def test_screenshot_is_refused_before_the_browser_is_touched(session):
"""Containment must be decided by the tool, not incidentally by the browser layer.

Previously the check did not exist and `out.parent.mkdir(parents=True)` sat inside the
Playwright callback, so on a machine with Playwright installed a refused path still had
its directory tree created. The refusal now happens first, which is observable here
because the error is the containment error rather than Playwright's.
"""
shot = _tools(session["roots"])["browser_screenshot"]
out = shot(str(session["outside"] / "a" / "b" / "c.png"))
assert "outside the session" in out["error"]
assert "laywright" not in out["error"]
assert not (session["outside"] / "a").exists()


def test_tools_are_constructible_without_roots(session):
"""CLI/direct callers pass no roots; construction must still work (and deny)."""
upload = _tools(None)["browser_upload_file"]
assert "error" in upload("input", str(session["outside"] / "id_rsa"))


# -- the default screenshot location ------------------------------------------

def test_default_screenshot_path_is_unpredictable_and_private(tmp_path, monkeypatch):
"""A fixed name in a shared temp dir is pre-creatable by another local user.

On Linux /tmp is shared, so `coworker-browser-screenshot.png` could be planted as a
symlink (the write then follows it) and the image — potentially a logged-in inbox — was
left at the umask default. mkstemp gives a random name created 0600 with O_EXCL.
"""
import re
import stat

shared = tmp_path / "shared_tmp"
shared.mkdir()
monkeypatch.setattr(tempfile_mod, "gettempdir", lambda: str(shared))

shot = _tools([RootDir(path=tmp_path / "scratch", writable=True)])["browser_screenshot"]
shot() # no path -> default location
shot()

created = sorted(shared.iterdir())
assert len(created) == 2, "each call must get its own file, not a shared fixed name"
assert created[0].name != created[1].name

for f in created:
assert re.match(r"coworker-browser-screenshot\..+\.png$", f.name), f.name
mode = stat.S_IMODE(f.stat().st_mode)
assert mode & (stat.S_IRGRP | stat.S_IROTH) == 0, f"{f.name} was {oct(mode)}"