mirror of
https://github.com/usestrix/strix.git
synced 2026-08-25 04:12:37 +02:00
oyasumiandClaude Opus 5
a7336fa194
Two guarded-mode false-positives surfaced in real scan traces. The reviewer blocked a boolean SQL injection probe (`curl "…/login?username='+OR+'1'='1"`) for being an injection attempt at all, though it is a read-only GET that changes nothing. The prompt said "allow only non-destructive" but never established that in-scope offensive testing is the tool's authorized purpose, so the model blocked on the technique. Rewrite the guarded-mode guidance to judge by effect: in-scope injection probes, recon, enumeration, and fuzzing pass, while destructive or persistent effects block — with SQL spelled out (boolean/UNION/time-based read probes pass; DROP, DELETE, INSERT, INTO OUTFILE, stacked statements, and command execution block). Ambiguous evidence still fails closed, and every deterministic block, the completeness gate, observe's passive-only rule, and scope enforcement are kept. Separately the reviewer blocked a plain `curl` as "use of bash shell within a curl command". The shell wrapper stamps `shell: bash` onto every exec_command for execution, and the evidence packet passed that transport default straight to the reviewer, which read it as the agent invoking a shell. Strip the harness-injected transport keys (`shell`, `max_output_tokens`) from the packet's original_arguments; the command itself is still parsed from `cmd`, so an agent-authored `bash -c` payload is unaffected. Note: the effect-based prompt also lets in-scope recon tools (nmap, subfinder, ffuf, katana) through, which the old prompt blocked as "scanning" or "high volume". That follows directly from judging by effect rather than technique. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
841 lines
27 KiB
Python
841 lines
27 KiB
Python
"""Deterministic safety evidence compilation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import io
|
|
from types import SimpleNamespace
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
import pytest
|
|
|
|
from strix.config.settings import SafetySettings
|
|
from strix.safety.evidence import _PythonFacts, compile_evidence, parse_command
|
|
|
|
|
|
if TYPE_CHECKING:
|
|
from pathlib import Path
|
|
|
|
|
|
# Marks a path that exists but cannot be read, which must not look like an absent module.
|
|
_UNREADABLE = "<unreadable>"
|
|
|
|
|
|
class _Sandbox:
|
|
def __init__(self, files: dict[str, str]) -> None:
|
|
self.files = files
|
|
|
|
async def read(self, path: Path) -> io.BytesIO:
|
|
key = path.as_posix()
|
|
if key not in self.files:
|
|
raise FileNotFoundError(key)
|
|
if self.files[key] == _UNREADABLE:
|
|
raise PermissionError(key)
|
|
return io.BytesIO(self.files[key].encode())
|
|
|
|
|
|
def _facts(source: str) -> _PythonFacts:
|
|
facts = _PythonFacts()
|
|
facts.visit(ast.parse(source))
|
|
return facts
|
|
|
|
|
|
def _ctx(files: dict[str, str], *, turn_input: list[Any] | None = None) -> Any:
|
|
return SimpleNamespace(
|
|
context={"agent_id": "agent-1", "sandbox_session": _Sandbox(files)},
|
|
tool_call_id="call-1",
|
|
turn_input=turn_input or [],
|
|
)
|
|
|
|
|
|
async def _compile(
|
|
command: str,
|
|
files: dict[str, str] | None = None,
|
|
*,
|
|
turn_input: list[Any] | None = None,
|
|
workdir: str | None = None,
|
|
mode: str = "guarded",
|
|
) -> Any:
|
|
arguments: dict[str, Any] = {"cmd": command}
|
|
if workdir is not None:
|
|
arguments["workdir"] = workdir
|
|
return await compile_evidence(
|
|
case_id="case",
|
|
ctx=_ctx(files or {}, turn_input=turn_input),
|
|
arguments=arguments,
|
|
mode=mode,
|
|
scope={},
|
|
user_instruction="",
|
|
settings=SafetySettings(),
|
|
)
|
|
|
|
|
|
def test_parse_command_identifies_direct_browser_action() -> None:
|
|
plan = parse_command("agent-browser click @e3")
|
|
|
|
assert plan.browser is True
|
|
assert plan.browser_action == "click"
|
|
assert plan.compound is False
|
|
|
|
|
|
def test_parse_command_marks_browser_chaining_compound() -> None:
|
|
plan = parse_command("agent-browser click @e3 && agent-browser snapshot -i")
|
|
|
|
assert plan.browser is True
|
|
assert plan.compound is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_python_script_collects_local_dependency_source() -> None:
|
|
bundle = await compile_evidence(
|
|
case_id="case-1",
|
|
ctx=_ctx(
|
|
{
|
|
"/workspace/check.py": "from helper import target\nprint(target)\n",
|
|
"/workspace/helper.py": 'target = "https://example.test/health"\n',
|
|
}
|
|
),
|
|
arguments={"cmd": "python /workspace/check.py"},
|
|
mode="guarded",
|
|
scope={"authorized_targets": [{"value": "https://example.test"}]},
|
|
user_instruction="Inspect the test target.",
|
|
settings=SafetySettings(),
|
|
)
|
|
try:
|
|
paths = {item["path"] for item in bundle.packet["artifacts"]}
|
|
assert paths == {"/workspace/check.py", "/workspace/helper.py"}
|
|
assert bundle.complete is True
|
|
assert bundle.deterministic_block is None
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_browser_automation_inside_script_is_blocked() -> None:
|
|
bundle = await compile_evidence(
|
|
case_id="case-2",
|
|
ctx=_ctx(
|
|
{
|
|
"/workspace/browser.py": (
|
|
'import subprocess\nsubprocess.run(["agent-browser", "click", "@e3"])\n'
|
|
)
|
|
}
|
|
),
|
|
arguments={"cmd": "python /workspace/browser.py"},
|
|
mode="guarded",
|
|
scope={},
|
|
user_instruction="",
|
|
settings=SafetySettings(),
|
|
)
|
|
try:
|
|
assert bundle.deterministic_block is not None
|
|
assert "direct agent-browser commands" in bundle.deterministic_block
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_browser_library_import_inside_script_is_blocked() -> None:
|
|
bundle = await compile_evidence(
|
|
case_id="case-browser-import",
|
|
ctx=_ctx({"/workspace/browser.py": "from playwright.async_api import Browser\n"}),
|
|
arguments={"cmd": "python /workspace/browser.py"},
|
|
mode="guarded",
|
|
scope={},
|
|
user_instruction="",
|
|
settings=SafetySettings(),
|
|
)
|
|
try:
|
|
assert bundle.deterministic_block is not None
|
|
assert "direct agent-browser commands" in bundle.deterministic_block
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dynamic_exec_makes_script_evidence_incomplete() -> None:
|
|
bundle = await compile_evidence(
|
|
case_id="case-3",
|
|
ctx=_ctx({"/workspace/dynamic.py": "exec(input())\n"}),
|
|
arguments={"cmd": "python /workspace/dynamic.py"},
|
|
mode="guarded",
|
|
scope={},
|
|
user_instruction="",
|
|
settings=SafetySettings(),
|
|
)
|
|
try:
|
|
assert bundle.complete is False
|
|
assert any("exec" in reason for reason in bundle.incomplete_reasons)
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dynamic_network_destination_is_incomplete() -> None:
|
|
bundle = await compile_evidence(
|
|
case_id="case-dynamic-network",
|
|
ctx=_ctx(
|
|
{"/workspace/network.py": ("import requests\nimport sys\nrequests.get(sys.argv[1])\n")}
|
|
),
|
|
arguments={"cmd": "python /workspace/network.py https://example.test"},
|
|
mode="guarded",
|
|
scope={},
|
|
user_instruction="",
|
|
settings=SafetySettings(),
|
|
)
|
|
try:
|
|
assert bundle.complete is False
|
|
assert any("dynamic network destination" in item for item in bundle.incomplete_reasons)
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_creation_and_execution_chain_must_be_split() -> None:
|
|
bundle = await compile_evidence(
|
|
case_id="case-chain",
|
|
ctx=_ctx({}),
|
|
arguments={"cmd": "curl https://example.test/x.py -o x.py && python x.py"},
|
|
mode="guarded",
|
|
scope={},
|
|
user_instruction="",
|
|
settings=SafetySettings(),
|
|
)
|
|
try:
|
|
assert bundle.deterministic_block is not None
|
|
assert "split" in bundle.deterministic_block
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_browser_ref_requires_prior_snapshot() -> None:
|
|
bundle = await compile_evidence(
|
|
case_id="case-4",
|
|
ctx=_ctx({}),
|
|
arguments={"cmd": "agent-browser click @e3"},
|
|
mode="guarded",
|
|
scope={},
|
|
user_instruction="",
|
|
settings=SafetySettings(),
|
|
)
|
|
try:
|
|
assert bundle.complete is False
|
|
assert "prior snapshot" in bundle.incomplete_reasons[0]
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_browser_ref_uses_prior_snapshot_output() -> None:
|
|
history = [
|
|
{
|
|
"type": "function_call",
|
|
"name": "exec_command",
|
|
"call_id": "snapshot-1",
|
|
"arguments": '{"cmd":"agent-browser snapshot -i"}',
|
|
},
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "snapshot-1",
|
|
"output": '@e3 [button type="submit"] "Search"',
|
|
},
|
|
]
|
|
bundle = await compile_evidence(
|
|
case_id="case-5",
|
|
ctx=_ctx({}, turn_input=history),
|
|
arguments={"cmd": "agent-browser click @e3"},
|
|
mode="guarded",
|
|
scope={},
|
|
user_instruction="",
|
|
settings=SafetySettings(),
|
|
)
|
|
try:
|
|
assert bundle.complete is True
|
|
assert bundle.packet["browser"]["latest_snapshot"]["call_id"] == "snapshot-1"
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"command",
|
|
[
|
|
"ls -la\nrm -rf /workspace/app",
|
|
"ls & rm -rf /workspace/app",
|
|
"ls -la; rm -rf /workspace/app",
|
|
],
|
|
)
|
|
async def test_destructive_command_chained_to_a_read_command_is_blocked(command: str) -> None:
|
|
plan = parse_command(command)
|
|
assert plan.compound is True
|
|
assert plan.read_only is False
|
|
|
|
bundle = await _compile(command)
|
|
try:
|
|
assert bundle.deterministic_allow is None
|
|
assert bundle.deterministic_block is not None
|
|
assert "destructive" in bundle.deterministic_block
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
def test_quoted_separator_is_not_compound() -> None:
|
|
assert parse_command("curl 'https://example.test/?a=1&b=2'").compound is False
|
|
assert parse_command('agent-browser open "https://example.test/?a=1&b=2"').compound is False
|
|
|
|
|
|
def test_read_only_fast_path_inspects_options() -> None:
|
|
assert parse_command("rg -n --json needle /workspace").read_only is True
|
|
assert parse_command("ls -la /workspace").read_only is True
|
|
# `--pre` hands ripgrep an arbitrary program to run on every matched file.
|
|
assert parse_command("rg --pre /workspace/payload.sh -e . /workspace").read_only is False
|
|
assert parse_command("rg --search-zip needle /workspace").read_only is False
|
|
assert parse_command("file -C -m /workspace/magic /workspace/x").read_only is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inline_python_source_collects_local_dependencies() -> None:
|
|
bundle = await _compile(
|
|
'python -c "import wipe; wipe.go()"',
|
|
{"/workspace/wipe.py": "import shutil\n\n\ndef go():\n shutil.rmtree('/workspace')\n"},
|
|
workdir="/workspace",
|
|
)
|
|
try:
|
|
artifacts = bundle.packet["artifacts"]
|
|
assert [item["path"] for item in artifacts] == ["<inline>", "/workspace/wipe.py"]
|
|
dependency = bundle.root / artifacts[1]["evidence_path"]
|
|
assert "shutil.rmtree" in dependency.read_text(encoding="utf-8")
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inline_python_dynamic_feature_is_incomplete() -> None:
|
|
bundle = await _compile('python -c "exec(input())"', workdir="/workspace")
|
|
try:
|
|
assert bundle.complete is False
|
|
assert any("exec" in reason for reason in bundle.incomplete_reasons)
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_relative_imports_are_collected() -> None:
|
|
bundle = await _compile(
|
|
"python /workspace/main.py",
|
|
{
|
|
"/workspace/main.py": "import pkg.mod\n",
|
|
"/workspace/pkg/__init__.py": "",
|
|
"/workspace/pkg/mod.py": "from . import payload\nfrom ..sibling import helper\n",
|
|
"/workspace/pkg/payload.py": "import shutil\nshutil.rmtree('/workspace/app')\n",
|
|
"/workspace/sibling.py": "helper = 1\n",
|
|
},
|
|
)
|
|
try:
|
|
paths = {item["path"] for item in bundle.packet["artifacts"]}
|
|
assert "/workspace/pkg/payload.py" in paths
|
|
assert "/workspace/sibling.py" in paths
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_import_path_mutation_makes_evidence_incomplete() -> None:
|
|
bundle = await _compile(
|
|
"python /workspace/run.py",
|
|
{
|
|
"/workspace/run.py": (
|
|
"import sys\nsys.path.insert(0, '/workspace/lib')\nimport payload\npayload.main()\n"
|
|
)
|
|
},
|
|
)
|
|
try:
|
|
assert bundle.complete is False
|
|
assert any("search path" in reason for reason in bundle.incomplete_reasons)
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unreadable_local_module_is_reported() -> None:
|
|
bundle = await _compile(
|
|
"python /workspace/run.py",
|
|
{"/workspace/run.py": "import payload\n", "/workspace/payload.py": _UNREADABLE},
|
|
)
|
|
try:
|
|
assert bundle.complete is False
|
|
assert any("cannot read local module" in reason for reason in bundle.incomplete_reasons)
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_interpreter_environment_override_is_blocked() -> None:
|
|
bundle = await _compile("PYTHONPATH=/workspace/lib python /workspace/run.py")
|
|
try:
|
|
assert bundle.deterministic_block is not None
|
|
assert "PYTHONPATH" in bundle.deterministic_block
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parent_traversal_leaves_the_workspace() -> None:
|
|
bundle = await _compile("python ../../opt/staged/run.py", workdir="/workspace")
|
|
try:
|
|
assert bundle.complete is False
|
|
assert "outside the inspectable workspace" in bundle.incomplete_reasons[0]
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
def test_env_wrapper_resolves_the_real_executable() -> None:
|
|
plan = parse_command("/usr/bin/env agent-browser click @e5")
|
|
|
|
assert plan.browser is True
|
|
assert plan.browser_action == "click"
|
|
|
|
|
|
def test_opaque_wrapper_fails_closed() -> None:
|
|
assert parse_command("timeout 5 rm -rf /workspace").parse_error is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_browser_session_env_override_is_blocked() -> None:
|
|
bundle = await _compile("AGENT_BROWSER_SESSION=shared agent-browser click @e3")
|
|
try:
|
|
assert bundle.deterministic_block is not None
|
|
assert "AGENT_BROWSER_SESSION" in bundle.deterministic_block
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unknown_browser_option_cannot_mask_the_action() -> None:
|
|
bundle = await _compile("agent-browser --timeout 5000 eval \"fetch('/x')\"")
|
|
try:
|
|
assert bundle.packet["pending_action"]["browser_action"] == "eval"
|
|
assert bundle.deterministic_block is not None
|
|
assert "eval" in bundle.deterministic_block
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unparseable_browser_option_fails_closed() -> None:
|
|
bundle = await _compile("agent-browser --unknown-flag value click @e3")
|
|
try:
|
|
assert bundle.complete is False
|
|
assert any("unrecognized" in reason for reason in bundle.incomplete_reasons)
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_attached_browser_session_override_is_blocked() -> None:
|
|
bundle = await _compile("agent-browser --session=evil click @e3")
|
|
try:
|
|
assert bundle.deterministic_block is not None
|
|
assert "overrides are blocked" in bundle.deterministic_block
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snapshot_taken_before_a_navigation_is_stale() -> None:
|
|
history = [
|
|
{
|
|
"type": "function_call",
|
|
"name": "exec_command",
|
|
"call_id": "snapshot-1",
|
|
"arguments": '{"cmd":"agent-browser snapshot -i"}',
|
|
},
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "snapshot-1",
|
|
"output": '@e3 [button] "Search"',
|
|
},
|
|
{
|
|
"type": "function_call",
|
|
"name": "exec_command",
|
|
"call_id": "navigate-1",
|
|
"arguments": '{"cmd":"agent-browser navigate https://example.test/admin"}',
|
|
},
|
|
{"type": "function_call_output", "call_id": "navigate-1", "output": "ok"},
|
|
]
|
|
bundle = await _compile("agent-browser click @e3", turn_input=history)
|
|
try:
|
|
assert bundle.complete is False
|
|
assert any("predates" in reason for reason in bundle.incomplete_reasons)
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reading_the_page_does_not_stale_a_snapshot() -> None:
|
|
history = [
|
|
{
|
|
"type": "function_call",
|
|
"name": "exec_command",
|
|
"call_id": "snapshot-1",
|
|
"arguments": '{"cmd":"agent-browser snapshot -i"}',
|
|
},
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "snapshot-1",
|
|
"output": '@e3 [button] "Search"',
|
|
},
|
|
{
|
|
"type": "function_call",
|
|
"name": "exec_command",
|
|
"call_id": "get-1",
|
|
"arguments": '{"cmd":"agent-browser get text @e3"}',
|
|
},
|
|
{"type": "function_call_output", "call_id": "get-1", "output": "Search"},
|
|
]
|
|
bundle = await _compile("agent-browser click @e3", turn_input=history)
|
|
try:
|
|
assert bundle.complete is True
|
|
assert bundle.packet["browser"]["latest_snapshot"]["stale"] is False
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dependency_closure_may_exceed_one_file_limit() -> None:
|
|
settings = SafetySettings()
|
|
filler = "#" * (settings.max_artifact_bytes - 64)
|
|
bundle = await _compile(
|
|
"python /workspace/run.py",
|
|
{
|
|
"/workspace/run.py": f"import first\nimport second\n{filler}",
|
|
"/workspace/first.py": filler,
|
|
"/workspace/second.py": filler,
|
|
},
|
|
)
|
|
try:
|
|
assert bundle.complete is True
|
|
assert len(bundle.packet["artifacts"]) == 3
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("command", "expected"),
|
|
[
|
|
("curl -X DELETE https://example.test/users/1", "DELETE"),
|
|
("curl --request PUT https://example.test/users/1", "PUT"),
|
|
("curl -d payload https://example.test/users", "-d"),
|
|
("wget --post-data=x https://example.test/users", "--post-data"),
|
|
],
|
|
)
|
|
def test_mutating_http_requests_are_recognized(command: str, expected: str) -> None:
|
|
assert expected in (parse_command(command).mutating_request or "")
|
|
|
|
|
|
def test_passive_http_requests_are_not_flagged() -> None:
|
|
assert parse_command("curl https://example.test/users").mutating_request is None
|
|
assert parse_command("curl -X GET https://example.test/users").mutating_request is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("command", "expected"),
|
|
[
|
|
('bash -c "rm -rf /workspace/app"', "destructive"),
|
|
('sh -c "rm -rf /workspace/app"', "destructive"),
|
|
('bash -lc "rm -rf /workspace/app"', "destructive"),
|
|
('bash -c "agent-browser click @e3"', "Browser automation embedded"),
|
|
],
|
|
)
|
|
async def test_shell_inline_source_is_parsed_not_just_stored(
|
|
command: str,
|
|
expected: str,
|
|
) -> None:
|
|
"""`-c` source is the obvious place to hide a command, so the inner string is parsed
|
|
and the same deterministic rules applied to it."""
|
|
bundle = await _compile(command)
|
|
try:
|
|
assert bundle.deterministic_block is not None
|
|
assert expected in bundle.deterministic_block
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shell_inline_source_is_recorded_as_an_artifact() -> None:
|
|
bundle = await _compile('bash -c "echo hello"')
|
|
try:
|
|
[artifact] = bundle.packet["artifacts"]
|
|
assert artifact["path"] == "<inline>"
|
|
assert artifact["source"] == "echo hello"
|
|
assert artifact["inner_executable"] == "echo"
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_module_execution_cannot_be_resolved_to_a_script() -> None:
|
|
bundle = await _compile("python -m http.server")
|
|
try:
|
|
assert bundle.complete is False
|
|
assert any("-m execution" in reason for reason in bundle.incomplete_reasons)
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dependency_count_limit_makes_evidence_incomplete() -> None:
|
|
bundle = await compile_evidence(
|
|
case_id="case-dependency-limit",
|
|
ctx=_ctx(
|
|
{
|
|
"/workspace/run.py": "import first\nimport second\n",
|
|
"/workspace/first.py": "value = 1\n",
|
|
"/workspace/second.py": "value = 2\n",
|
|
}
|
|
),
|
|
arguments={"cmd": "python /workspace/run.py"},
|
|
mode="guarded",
|
|
scope={},
|
|
user_instruction="",
|
|
settings=SafetySettings(max_dependencies=1),
|
|
)
|
|
try:
|
|
assert bundle.complete is False
|
|
assert any("dependency count" in reason for reason in bundle.incomplete_reasons)
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_oversized_dependency_closure_makes_evidence_incomplete() -> None:
|
|
filler = "#" * 8000
|
|
bundle = await compile_evidence(
|
|
case_id="case-byte-limit",
|
|
ctx=_ctx(
|
|
{
|
|
"/workspace/run.py": f"import first\n{filler}",
|
|
"/workspace/first.py": filler,
|
|
}
|
|
),
|
|
arguments={"cmd": "python /workspace/run.py"},
|
|
mode="guarded",
|
|
scope={},
|
|
user_instruction="",
|
|
settings=SafetySettings(max_total_artifact_bytes=10_000),
|
|
)
|
|
try:
|
|
assert bundle.complete is False
|
|
assert any("total byte limit" in reason for reason in bundle.incomplete_reasons)
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("command", "action", "subcommand"),
|
|
[
|
|
("agent-browser tab new https://example.test/admin", "tab", "new"),
|
|
("agent-browser tab close 2", "tab", "close"),
|
|
("agent-browser session clear", "session", "clear"),
|
|
],
|
|
)
|
|
async def test_grouped_browser_verbs_are_not_passive(
|
|
command: str,
|
|
action: str,
|
|
subcommand: str,
|
|
) -> None:
|
|
"""`tab new <url>` navigates and `tab close` destroys page state, so the bare verb
|
|
must not be enough to earn the observation fast path."""
|
|
plan = parse_command(command)
|
|
assert plan.browser_action == action
|
|
assert plan.browser_subcommand == subcommand
|
|
assert plan.read_only is False
|
|
|
|
bundle = await _compile(command)
|
|
try:
|
|
assert bundle.deterministic_allow is None
|
|
assert bundle.packet["browser"]["passive"] is False
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("command", ["agent-browser tab", "agent-browser snapshot -i"])
|
|
async def test_bare_listing_verbs_keep_the_observation_fast_path(command: str) -> None:
|
|
bundle = await _compile(command)
|
|
try:
|
|
assert bundle.deterministic_allow is not None
|
|
assert bundle.packet["browser"]["passive"] is True
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_grouped_blocked_verbs_still_match_on_the_verb() -> None:
|
|
"""The blocked list keys off the bare verb, so qualifying the action must not stop
|
|
`auth login` from matching `auth`."""
|
|
bundle = await _compile("agent-browser auth login my-app")
|
|
try:
|
|
assert bundle.deterministic_block is not None
|
|
assert "auth" in bundle.deterministic_block
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"command",
|
|
[
|
|
"python3.12 /workspace/run.py",
|
|
"/usr/bin/python3.12 /workspace/run.py",
|
|
"pypy3 /workspace/run.py",
|
|
],
|
|
)
|
|
async def test_versioned_interpreters_are_inspected(command: str) -> None:
|
|
bundle = await _compile(
|
|
command,
|
|
{"/workspace/run.py": "import helper\n", "/workspace/helper.py": "value = 1\n"},
|
|
)
|
|
try:
|
|
paths = {item["path"] for item in bundle.packet["artifacts"]}
|
|
assert paths == {"/workspace/run.py", "/workspace/helper.py"}
|
|
assert bundle.complete is True
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_python_interpreter_source_is_inspected() -> None:
|
|
bundle = await _compile(
|
|
"php /workspace/app.php",
|
|
{"/workspace/app.php": "<?php unlink('/workspace/data'); ?>\n"},
|
|
)
|
|
try:
|
|
[artifact] = bundle.packet["artifacts"]
|
|
assert artifact["path"] == "/workspace/app.php"
|
|
assert "unlink" in artifact["source"]
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"command",
|
|
[
|
|
"python3.12",
|
|
"node",
|
|
"mystery-runner /workspace/run.py",
|
|
"./vendored-tool /workspace/run.sh",
|
|
],
|
|
)
|
|
async def test_unresolvable_code_execution_fails_closed(command: str) -> None:
|
|
"""A packet with no artifacts must never be stamped complete just because the
|
|
executable fell outside the interpreter set."""
|
|
bundle = await _compile(command, {"/workspace/run.py": "import os\n"})
|
|
try:
|
|
assert bundle.complete is False
|
|
assert bundle.packet["artifacts"] == []
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("command", ["nmap -sV example.test", "ls /workspace", "whoami"])
|
|
async def test_commands_without_a_script_are_not_forced_incomplete(command: str) -> None:
|
|
"""Fail-closed on unresolved script execution must not swallow ordinary tools."""
|
|
bundle = await _compile(command)
|
|
try:
|
|
assert bundle.complete is True
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_absolute_submodule_import_is_collected() -> None:
|
|
"""`from pkg import payload` may name a submodule, not an attribute of the package."""
|
|
bundle = await _compile(
|
|
"python /workspace/main.py",
|
|
{
|
|
"/workspace/main.py": "from pkg import payload\npayload.go()\n",
|
|
"/workspace/pkg/__init__.py": "",
|
|
"/workspace/pkg/payload.py": "import shutil\n\n\ndef go():\n shutil.rmtree('/x')\n",
|
|
},
|
|
)
|
|
try:
|
|
paths = {item["path"] for item in bundle.packet["artifacts"]}
|
|
assert "/workspace/pkg/payload.py" in paths
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_relative_submodule_import_is_collected() -> None:
|
|
bundle = await _compile(
|
|
"python /workspace/main.py",
|
|
{
|
|
"/workspace/main.py": "import pkg.mod\n",
|
|
"/workspace/pkg/__init__.py": "",
|
|
"/workspace/pkg/mod.py": "from .inner import payload\n",
|
|
"/workspace/pkg/inner/__init__.py": "",
|
|
"/workspace/pkg/inner/payload.py": "import shutil\nshutil.rmtree('/x')\n",
|
|
},
|
|
)
|
|
try:
|
|
paths = {item["path"] for item in bundle.packet["artifacts"]}
|
|
assert "/workspace/pkg/inner/payload.py" in paths
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
def test_imported_attributes_do_not_pollute_the_reported_imports() -> None:
|
|
"""Submodule candidates are resolution-only; the packet still shows the statements
|
|
as the author wrote them."""
|
|
facts = _facts("from os import path\nfrom mypkg import CONSTANT\n")
|
|
|
|
assert facts.imports == {"os", "mypkg"}
|
|
assert facts.submodule_imports == {"os.path", "mypkg.CONSTANT"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_harness_transport_keys_are_hidden_from_the_reviewer() -> None:
|
|
"""The shell wrapper stamps `shell: bash` onto every command; surfacing it in the
|
|
packet made the reviewer read the transport default as the agent invoking a shell."""
|
|
bundle = await compile_evidence(
|
|
case_id="case-transport",
|
|
ctx=_ctx({}),
|
|
arguments={
|
|
"cmd": "curl -I \"https://example.test/login?u='+OR+'1'='1\"",
|
|
"shell": "bash",
|
|
"max_output_tokens": 8000,
|
|
},
|
|
mode="guarded",
|
|
scope={"authorized_targets": [{"value": "https://example.test"}]},
|
|
user_instruction="",
|
|
settings=SafetySettings(),
|
|
)
|
|
try:
|
|
original = bundle.packet["pending_action"]["original_arguments"]
|
|
assert "shell" not in original
|
|
assert "max_output_tokens" not in original
|
|
assert original["cmd"].startswith("curl")
|
|
# A GET probe with a boolean payload is not deterministically blocked; the reviewer
|
|
# judges it by effect.
|
|
assert bundle.deterministic_block is None
|
|
finally:
|
|
bundle.cleanup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shell_field_does_not_hide_a_genuine_bash_c_payload() -> None:
|
|
"""Stripping the transport `shell` key must not weaken parsing of an agent-authored
|
|
`bash -c`, which is carried in `cmd`, not the shell field."""
|
|
bundle = await _compile('bash -c "rm -rf /workspace/app"')
|
|
try:
|
|
assert bundle.deterministic_block is not None
|
|
assert "destructive" in bundle.deterministic_block
|
|
finally:
|
|
bundle.cleanup()
|