mirror of
https://github.com/andrewyng/openworker.git
synced 2026-11-04 13:27:11 +00:00
Rohit C Prasad
09dc46575a
With a person at the terminal the coworker asks, they allow the site from outside, and it tries again. With auto-answer it reports the block in its answer, as before.
181 lines
8.1 KiB
Python
181 lines
8.1 KiB
Python
"""OpenWorker running inside an OpenShell sandbox (`OPENSHELL_SANDBOX=1`): the agent is
|
|
told what a blocked request looks like, OpenShell's own skills are readable, and the web
|
|
tools leave the name lookup to OpenShell's proxy. Outside OpenShell nothing changes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import socket
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from coworker.agent import build_engine
|
|
from coworker.agents.cowork import cowork_agent
|
|
from coworker.sandbox import inside
|
|
from coworker.web import guard
|
|
|
|
PROXY_ENV = ("HTTPS_PROXY", "https_proxy", "HTTP_PROXY", "http_proxy", "ALL_PROXY", "all_proxy")
|
|
|
|
|
|
class _StubProvider:
|
|
def complete(self, **_kw): # pragma: no cover - never called at build time
|
|
from coworker.providers import AssistantTurn
|
|
|
|
return AssistantTurn()
|
|
|
|
def capabilities(self, _model): # pragma: no cover
|
|
from coworker.providers.base import ModelCapabilities
|
|
|
|
return ModelCapabilities()
|
|
|
|
|
|
@pytest.fixture
|
|
def outside(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.delenv("OPENSHELL_SANDBOX", raising=False)
|
|
for name in PROXY_ENV:
|
|
monkeypatch.delenv(name, raising=False)
|
|
|
|
|
|
@pytest.fixture
|
|
def in_openshell(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path:
|
|
"""Inside a sandbox whose proposals are off: no skills folder yet."""
|
|
for name in PROXY_ENV:
|
|
monkeypatch.delenv(name, raising=False)
|
|
monkeypatch.setenv("OPENSHELL_SANDBOX", "1")
|
|
monkeypatch.setenv("HTTPS_PROXY", "http://10.200.0.1:3128")
|
|
skills = tmp_path / "etc-openshell-skills"
|
|
monkeypatch.setattr(inside, "SKILLS_DIR", skills)
|
|
return skills
|
|
|
|
|
|
def _turn_on_proposals(skills: Path) -> None:
|
|
"""What OpenShell writes when `agent_policy_proposals_enabled` is set."""
|
|
(skills / "policy-advisor").mkdir(parents=True)
|
|
(skills / "policy-advisor" / "SKILL.md").write_text(
|
|
"---\nname: openshell-policy-advisor\n"
|
|
"description: Use when an OpenShell sandbox returns policy_denied.\n---\n\n"
|
|
"Read `/etc/openshell/skills/policy_advisor.md`.\n",
|
|
encoding="utf-8",
|
|
)
|
|
(skills / "policy_advisor.md").write_text("# OpenShell Policy Advisor\n", encoding="utf-8")
|
|
|
|
|
|
def _engine(tmp_path: Path):
|
|
ws = tmp_path / "ws"
|
|
ws.mkdir()
|
|
return build_engine(agent=cowork_agent(), workspace=ws, provider=_StubProvider())
|
|
|
|
|
|
# -- what the agent is told -------------------------------------------------------------
|
|
|
|
|
|
def test_outside_openshell_nothing_is_said_or_added(outside, tmp_path: Path) -> None:
|
|
assert inside.context() == ""
|
|
engine = _engine(tmp_path)
|
|
assert "OpenShell" not in engine.context_provider()
|
|
assert [r.label for r in engine.roots if r.label == "openshell-skills"] == []
|
|
|
|
|
|
def test_inside_the_agent_is_told_what_a_blocked_request_looks_like(in_openshell: Path, tmp_path: Path) -> None:
|
|
text = _engine(tmp_path).context_provider()
|
|
assert "runs inside an OpenShell sandbox" in text
|
|
assert "CONNECT tunnel failed, response 403" in text and "policy_denied" in text
|
|
assert "Couldn't connect to server" in text # how OpenShell 0.1 refuses
|
|
# Proposals are off, so there is no skill to load: the person changes the policy.
|
|
# The whole command: seen on OpenShell 0.1.2, a rule added without --binary lets no
|
|
# program through, so the short form sent people in a circle.
|
|
assert "openshell policy update SANDBOX_NAME --add-endpoint HOST:PORT --binary '/**' --wait" in text
|
|
assert "openshell-policy-advisor" not in text
|
|
|
|
|
|
def test_with_a_person_there_the_agent_asks_and_waits_for_the_site(in_openshell: Path, tmp_path: Path) -> None:
|
|
# Seen live: told only to report the block, the agent ended the run with the command
|
|
# as its answer, and the sandbox was gone before anyone could run it.
|
|
text = _engine(tmp_path).context_provider()
|
|
assert "ask the user with `ask_user` and wait" in text
|
|
assert "they have allowed it, or leave it blocked" in text
|
|
assert "`openshell sandbox list`" in text # the sandbox's name is not known inside
|
|
|
|
|
|
def test_with_auto_answer_the_agent_reports_the_block_and_does_not_ask(in_openshell: Path, tmp_path: Path) -> None:
|
|
engine = _engine(tmp_path)
|
|
engine.attendance = lambda: "auto"
|
|
text = engine.context_provider()
|
|
assert "nobody is there to allow it" in text and "ask_user" not in text.split("OpenShell sandbox", 1)[1].split("\n\n", 1)[0]
|
|
assert "openshell policy update SANDBOX_NAME --add-endpoint HOST:PORT --binary '/**' --wait" in text
|
|
|
|
|
|
def test_with_proposals_on_the_agent_is_pointed_at_openshells_skill(in_openshell: Path, tmp_path: Path) -> None:
|
|
_turn_on_proposals(in_openshell)
|
|
engine = _engine(tmp_path)
|
|
text = engine.context_provider()
|
|
assert "load the skill `openshell-policy-advisor`" in text
|
|
assert "openshell policy update" not in text
|
|
# The skill is on the menu under OpenShell's own name and description…
|
|
assert "Use when an OpenShell sandbox returns policy_denied." in text
|
|
assert engine.skill_loader.get("openshell-policy-advisor") is not None
|
|
# …and its folder is one the agent may read, never write.
|
|
(root,) = [r for r in engine.roots if r.label == "openshell-skills"]
|
|
assert root.path == in_openshell.resolve() and root.writable is False
|
|
|
|
|
|
def test_proposals_turned_on_mid_session_are_picked_up_on_the_next_turn(in_openshell: Path, tmp_path: Path) -> None:
|
|
engine = _engine(tmp_path)
|
|
assert "openshell policy update" in engine.context_provider()
|
|
_turn_on_proposals(in_openshell)
|
|
text = engine.context_provider()
|
|
assert "load the skill `openshell-policy-advisor`" in text
|
|
assert "Use when an OpenShell sandbox returns policy_denied." in text
|
|
|
|
|
|
# -- web tools --------------------------------------------------------------------------
|
|
|
|
|
|
def _no_lookup(*_a, **_k):
|
|
raise AssertionError("the name must not be looked up here")
|
|
|
|
|
|
def test_inside_a_name_goes_to_the_proxy_without_a_lookup(in_openshell, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(socket, "getaddrinfo", _no_lookup)
|
|
assert guard._vet("https://example.org/page") == (None, None)
|
|
|
|
class Client:
|
|
def __init__(self) -> None:
|
|
self.asked: list[str] = []
|
|
|
|
def get(self, url, **kwargs):
|
|
self.asked.append(url)
|
|
assert not kwargs # no pinned address, no Host override: the name as written
|
|
return type("R", (), {"status_code": 200, "extensions": {}})()
|
|
|
|
client = Client()
|
|
resp = guard.get_checked(client, "https://example.org/page")
|
|
assert client.asked == ["https://example.org/page"]
|
|
assert resp.extensions["logical_url"] == "https://example.org/page"
|
|
|
|
|
|
def test_inside_a_literal_internal_address_is_still_refused(in_openshell, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(socket, "getaddrinfo", _no_lookup)
|
|
for url in ("http://169.254.169.254/latest/meta-data/", "https://127.0.0.1:8443/", "https://10.0.0.5/"):
|
|
reason, pin = guard._vet(url)
|
|
assert reason and "refusing to fetch" in reason and pin is None
|
|
|
|
|
|
def test_inside_openshell_0_1_a_name_is_not_judged_by_its_placeholder_address(in_openshell, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
# OpenShell 0.1 sets no proxy variables and answers every name with an address in
|
|
# 198.18.0.0/15, which the guard refuses as private. Seen on 0.1.2: example.org -> 198.18.0.2.
|
|
monkeypatch.delenv("HTTPS_PROXY", raising=False)
|
|
monkeypatch.setattr(socket, "getaddrinfo", lambda *_a, **_k: [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("198.18.0.2", 443))])
|
|
assert guard._vet("https://example.org/page") == (None, None)
|
|
assert guard._vet("http://example.org/page") == (None, None)
|
|
# The same answer outside a sandbox is refused, as before.
|
|
monkeypatch.delenv("OPENSHELL_SANDBOX")
|
|
reason, _pin = guard._vet("https://example.org/page")
|
|
assert reason and "198.18.0.2" in reason
|
|
|
|
|
|
def test_outside_a_name_is_looked_up_and_pinned_as_before(outside, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.example:3128") # a proxy alone changes nothing
|
|
monkeypatch.setattr(socket, "getaddrinfo", lambda *_a, **_k: [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443))])
|
|
assert guard._vet("https://example.org/") == (None, "93.184.216.34")
|