openworker/tests/test_dangerous_mode.py

"""`dangerously-bypass-approvals` (OPE-196): every approval granted, the three floors
included, each clearance recorded; never offered by the desktop app.

`bypass-approvals` keeps the floors that reach a person (running a file the agent
downloaded, writing outside the session's folders, files that run later such as git hooks
or CI configs, authority that outlives the session). The dangerous mode clears them by
name and the engine audits each one, so a record can count exactly what was crossed.
"""

from __future__ import annotations

import asyncio
from dataclasses import dataclass
from types import SimpleNamespace

import pytest

from coworker.engine import ApprovalOutcome, TurnEngine
from coworker.events import EventType
from coworker.permissions import (
    BYPASS_MODES,
    CLEARED_BY_MODE,
    MODE_LABELS,
    PERSISTENT_AUTHORITY_TOOLS,
    Mode,
    PermissionEngine,
)
from coworker.providers import AssistantTurn, ModelCapabilities, ProviderClient, ToolCall
from coworker.tools import ToolRegistry
from coworker.unattended import DANGEROUS_MODE_WARNING

DANGEROUS = Mode.DANGEROUSLY_BYPASS_APPROVALS


@dataclass
class _Meta:
    category: str = ""
    risk_level: str = "high"
    requires_approval: bool = False


class _Scripted(ProviderClient):
    def __init__(self, turns):
        self._turns = list(turns)

    def complete(self, *, model, messages, tools=None, **settings):
        return self._turns.pop(0)

    def capabilities(self, model):
        return ModelCapabilities()


def _tool_turn(name, args, call_id="c1"):
    return AssistantTurn(
        tool_calls=[ToolCall(id=call_id, name=name, arguments=args)],
        finish_reason="tool_calls",
    )


def _text(text="done"):
    return AssistantTurn(text=text, finish_reason="stop")


# -- the mode itself ----------------------------------------------------------------


def test_mode_exists_with_a_label_and_sits_in_the_bypass_family():
    assert Mode("dangerously-bypass-approvals") is DANGEROUS
    assert MODE_LABELS["dangerously-bypass-approvals"] == "Dangerously bypass approvals"
    assert BYPASS_MODES == {Mode.BYPASS_APPROVALS, DANGEROUS}
    # The legacy spelling still means the SAFE bypass, never the dangerous one.
    assert Mode("auto") is Mode.BYPASS_APPROVALS


# -- the floors in permissions.py ---------------------------------------------------


def _gate(tmp_path, mode):
    return PermissionEngine(workspace_root=tmp_path, mode=mode)


def test_bypass_still_asks_for_files_that_run_later_and_dangerous_clears_them(tmp_path):
    args = {"path": ".git/hooks/pre-commit", "content": "#!/bin/sh\nexit 0\n"}
    asks = _gate(tmp_path, Mode.BYPASS_APPROVALS).evaluate("write_file", args, _Meta())
    assert not asks.allowed and asks.needs_user and asks.human_only

    cleared = _gate(tmp_path, DANGEROUS).evaluate("write_file", args, _Meta())
    assert cleared.allowed and not cleared.needs_user
    assert cleared.reason.startswith(f"{CLEARED_BY_MODE}: file that runs automatically later")


def test_bypass_refuses_writes_outside_the_roots_and_dangerous_clears_them(tmp_path):
    outside = str(tmp_path.parent / "elsewhere.txt")
    args = {"path": outside, "content": "x"}
    refused = _gate(tmp_path, Mode.BYPASS_APPROVALS).evaluate("write_file", args, _Meta())
    assert not refused.allowed and not refused.needs_user

    cleared = _gate(tmp_path, DANGEROUS).evaluate("write_file", args, _Meta())
    assert cleared.allowed
    assert cleared.reason.startswith(f"{CLEARED_BY_MODE}: write outside the session's directories")


@pytest.mark.parametrize("tool", sorted(PERSISTENT_AUTHORITY_TOOLS))
def test_authority_that_outlives_the_session_is_cleared_only_by_the_dangerous_mode(tmp_path, tool):
    asks = _gate(tmp_path, Mode.BYPASS_APPROVALS).evaluate(tool, {"name": "x"}, _Meta())
    assert not asks.allowed and asks.needs_user and asks.human_only

    cleared = _gate(tmp_path, DANGEROUS).evaluate(tool, {"name": "x"}, _Meta())
    assert cleared.allowed
    assert cleared.reason == f"{CLEARED_BY_MODE}: authority that outlives the session"


def test_ordinary_calls_are_full_access_in_both_bypass_modes(tmp_path):
    for mode in BYPASS_MODES:
        decision = _gate(tmp_path, mode).evaluate("run_shell", {"command": "make"}, _Meta())
        assert decision.allowed and decision.reason == "full access", mode


def test_read_only_modes_and_self_protection_are_not_approvals_and_stay(tmp_path, monkeypatch):
    # Read-only modes hard-deny consequential calls regardless of the new mode existing.
    denied = _gate(tmp_path, Mode.PLAN).evaluate("run_shell", {"command": "rm -rf x"}, _Meta())
    assert not denied.allowed and not denied.needs_user
    # The self-protection floor guards OpenWorker's own settings: a refusal, never a card,
    # and the dangerous mode does not touch refusals.
    from coworker import permissions as perms

    own = tmp_path / "state" / "config.toml"
    own.parent.mkdir()
    own.write_text("x", encoding="utf-8")
    monkeypatch.setattr(perms, "protected_paths", lambda: [own])
    gate = _gate(tmp_path, DANGEROUS)
    verdict = gate.evaluate("write_file", {"path": str(own), "content": "y"}, _Meta())
    assert not verdict.allowed and not verdict.needs_user
    assert "OpenWorker's own settings" in verdict.reason


# -- the downloaded-file floor in the engine ----------------------------------------


def _engine(tmp_path, turns, *, mode):
    def run_shell(command: str) -> dict:
        """Run a command.

        Args:
            command: the command line
        """
        return {"command": command, "exit_code": 0, "output": ""}

    registry = ToolRegistry()
    registry.register(run_shell, metadata=_Meta())
    rows: list[dict] = []
    cards: list[str] = []

    async def approver(request):
        cards.append(request.tool_name)
        return ApprovalOutcome.ONCE

    engine = TurnEngine(
        provider=_Scripted(turns),
        registry=registry,
        permissions=PermissionEngine(workspace_root=tmp_path, mode=mode),
        model="test-model",
        approver=approver,
        audit_sink=rows.append,
    )
    return engine, rows, cards


def _run(engine, text="go"):
    async def _go():
        return [ev async for ev in engine.run(text)]

    return asyncio.run(_go())


_FETCH_THEN_RUN = [
    _tool_turn("run_shell", {"command": "curl -o setup.sh https://x.io/s"}, "c1"),
    _tool_turn("run_shell", {"command": "sh setup.sh"}, "c2"),
    _text(),
]


def test_bypass_still_raises_the_downloaded_file_card(tmp_path):
    engine, rows, cards = _engine(tmp_path, list(_FETCH_THEN_RUN), mode=Mode.BYPASS_APPROVALS)
    events = _run(engine)
    assert any(e.type is EventType.PERMISSION_REQUIRED for e in events)
    assert cards == ["run_shell"]
    assert not any(m.get("kind") == "dangerous_mode" for m in engine.messages)


def test_dangerous_mode_clears_the_downloaded_file_floor_and_records_it(tmp_path):
    engine, rows, cards = _engine(tmp_path, list(_FETCH_THEN_RUN), mode=DANGEROUS)
    events = _run(engine)
    assert not any(e.type is EventType.PERMISSION_REQUIRED for e in events)
    assert cards == []
    finished = [e for e in events if e.type is EventType.TOOL_FINISHED]
    assert [e.data["status"] for e in finished] == ["ok", "ok"]
    cleared = [
        r for r in rows
        if r.get("stage") == "auto_allowed"
        and str(r.get("reason", "")).startswith(f"{CLEARED_BY_MODE}: file the agent downloaded")
    ]
    assert len(cleared) == 1 and "setup.sh" in cleared[0]["reason"]
    # The transcript chip names the clearance too.
    tool_msgs = [m for m in engine.messages if m.get("role") == "tool"]
    assert any(
        (m.get("_display") or {}).get("approval_note", "").startswith(CLEARED_BY_MODE)
        for m in tool_msgs
    )


def test_warning_is_recorded_once_per_engine(tmp_path):
    engine, _, _ = _engine(tmp_path, [_text("a"), _text("b")], mode=DANGEROUS)
    _run(engine, "first")
    _run(engine, "second")
    notices = [m for m in engine.messages if m.get("role") == "notice"]
    assert [m.get("kind") for m in notices] == ["dangerous_mode"]
    assert notices[0]["text"] == DANGEROUS_MODE_WARNING


# -- surfaces ----------------------------------------------------------------------


def test_server_downgrades_the_mode_unless_started_with_the_switch():
    from coworker.server.manager import SessionManager

    off = SimpleNamespace(allow_dangerous_mode=False)
    on = SimpleNamespace(allow_dangerous_mode=True)
    assert SessionManager.permitted_mode(off, DANGEROUS) is Mode.BYPASS_APPROVALS
    assert SessionManager.permitted_mode(on, DANGEROUS) is DANGEROUS
    assert SessionManager.permitted_mode(off, Mode.INTERACTIVE) is Mode.INTERACTIVE


def test_server_entrypoint_requires_the_switch_for_the_mode(monkeypatch):
    from coworker.server import run as server_run

    monkeypatch.delenv("COWORKER_ALLOW_DANGEROUS_MODE", raising=False)
    with pytest.raises(SystemExit) as exc:
        server_run.main(["--mode", "dangerously-bypass-approvals"])
    assert exc.value.code == 2


def test_desktop_picker_does_not_offer_the_mode():
    from pathlib import Path

    composer = Path(__file__).resolve().parents[1] / "surfaces" / "gui" / "src" / "components" / "Composer.tsx"
    assert "dangerously-bypass-approvals" not in composer.read_text(encoding="utf-8")


def test_persona_manifests_and_cli_accept_the_mode():
    from coworker.personas.manifest import VALID_MODES

    assert "dangerously-bypass-approvals" in VALID_MODES