mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-10-06 15:02:20 +02:00
731 lines
35 KiB
Python
731 lines
35 KiB
Python
"""Server bindings narrow operation authority and survive approved continuations."""
|
|
import asyncio
|
|
from dataclasses import FrozenInstanceError, replace
|
|
import json
|
|
import os
|
|
from unittest.mock import AsyncMock
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from src.agent_runtime.authority import (
|
|
ExactOperation, OperationGrant, RequestAuthority, bind_request_authority,
|
|
create_request_authority, save_background_authority, restore_background_authority,
|
|
seal_task_authority, restore_task_authority,
|
|
)
|
|
from src.agent_runtime.resource_binding import (
|
|
BoundFilesystemOperation, ResourceBinding, active_resource_operation,
|
|
bind_resource_operation, resolve_filesystem_operation,
|
|
)
|
|
from src.agent_runtime.resources import (
|
|
ExternalResource, FileObjectIdentity,
|
|
FilesystemResource, FilesystemRoot, FilesystemScope, OwnedResource, ProcessResource,
|
|
)
|
|
from src.tool_approvals import ToolApprovalStore
|
|
from src.tool_capabilities import ToolRunSecurityContext, capabilities_for_action
|
|
from src.tool_types import ToolBlock
|
|
|
|
|
|
def authority(root, *tools, roots=None, owner="alice", session="s"):
|
|
return RequestAuthority("resource-test", owner, session, str(root or ""),
|
|
tuple(OperationGrant(t) for t in tools), resource_roots=roots)
|
|
|
|
|
|
def resolve(grant, tool, content):
|
|
return resolve_filesystem_operation(ExactOperation.normalize(tool, content),
|
|
roots=grant.resource_roots, workspace=grant.workspace, request_id=grant.request_id)
|
|
|
|
|
|
async def dispatch(grant, tool, content, **kwargs):
|
|
from src import tool_execution as execution
|
|
return await execution.execute_tool_block(ToolBlock(tool, content),
|
|
owner=grant.owner, session_id=grant.session_id, workspace=grant.workspace or None,
|
|
request_authority=grant, security_context=kwargs.pop("security_context", execution.NO_TOOL_SECURITY_CONTEXT),
|
|
**kwargs)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def native_admin(monkeypatch):
|
|
from src import tool_execution
|
|
monkeypatch.setattr(tool_execution, "_owner_is_admin", lambda owner: True)
|
|
|
|
|
|
@pytest.mark.parametrize("selector", ["a.txt", "/workspace/a.txt", "host", "link"])
|
|
def test_aliases_resolve_to_one_observed_resource(tmp_path, selector):
|
|
target = tmp_path / "a.txt"
|
|
target.write_text("same object")
|
|
(tmp_path / "link").symlink_to(target)
|
|
grant = authority(tmp_path, "read_file")
|
|
value = str(target) if selector == "host" else selector
|
|
bound = resolve(grant, "read_file", value)
|
|
assert bound.bindings[0].resource.path == str(target)
|
|
assert bound.bindings[0].resource.identity == FileObjectIdentity.observe(target)
|
|
assert json.loads(bound.execution_input)["path"] == str(target)
|
|
assert bound.operation.input == value
|
|
|
|
|
|
@pytest.mark.parametrize("path", ["../sibling/secret", "/etc/passwd", ".SSH/key", "ID_RSA", "bad\0path", "bad\npath"])
|
|
def test_escapes_sensitive_and_malformed_paths_fail_closed(tmp_path, path):
|
|
grant = authority(tmp_path, "write_file")
|
|
with pytest.raises(ValueError):
|
|
resolve(grant, "write_file", json.dumps({"path": path, "content": "x"}))
|
|
|
|
|
|
def test_symlink_escape_is_not_a_resource(tmp_path):
|
|
workspace = tmp_path / "ws"
|
|
workspace.mkdir()
|
|
outside = tmp_path / "secret"
|
|
outside.write_text("private")
|
|
(workspace / "alias").symlink_to(outside)
|
|
with pytest.raises(ValueError):
|
|
resolve(authority(workspace, "read_file"), "read_file", "alias")
|
|
|
|
|
|
@pytest.mark.parametrize("alias", ["direct", "relative", "symlink", "hardlink"])
|
|
@pytest.mark.parametrize("must_exist", [True, False])
|
|
def test_media_workspace_paths_cannot_address_control_state(tmp_path, monkeypatch, alias, must_exist):
|
|
from src import constants, tool_execution
|
|
from src.agent_tools.media_tools import _resolve_workspace_path
|
|
control = tmp_path / "receipts.json"
|
|
control.write_text("private execution state")
|
|
monkeypatch.setattr(constants, "CONTAINMENT_STATE_FILE", str(control))
|
|
monkeypatch.setattr(tool_execution, "get_active_workspace", lambda: str(tmp_path))
|
|
if alias == "direct":
|
|
selector = str(control)
|
|
elif alias == "relative":
|
|
selector = "./receipts.json"
|
|
else:
|
|
target = tmp_path / "image.png"
|
|
if alias == "symlink":
|
|
target.symlink_to(control)
|
|
else:
|
|
os.link(control, target)
|
|
selector = "/workspace/image.png"
|
|
with pytest.raises(ValueError, match="execution-control"):
|
|
_resolve_workspace_path(selector, must_exist=must_exist)
|
|
assert control.read_text() == "private execution state"
|
|
|
|
|
|
def test_destination_binds_absence_and_existing_ancestors(tmp_path):
|
|
parent = tmp_path / "existing"
|
|
parent.mkdir()
|
|
bound = resolve(authority(tmp_path, "write_file"), "write_file", "existing/new/tree/result.txt\nx")
|
|
resource = bound.bindings[0].resource
|
|
assert bound.bindings[0].role == "destination"
|
|
assert resource.identity is None
|
|
assert [a.path for a in resource.ancestors] == [str(tmp_path), str(parent)]
|
|
bound.validate()
|
|
parent.rename(tmp_path / "old-parent")
|
|
parent.mkdir()
|
|
with pytest.raises(ValueError):
|
|
bound.validate()
|
|
|
|
|
|
@pytest.mark.parametrize("replacement", ["root", "file", "parent", "new-target"])
|
|
def test_replacement_invalidates_observed_identity(tmp_path, replacement):
|
|
root = tmp_path / "root"
|
|
root.mkdir()
|
|
parent = root / "sub"
|
|
parent.mkdir()
|
|
target = parent / "a.txt"
|
|
target.write_text("old")
|
|
content = "sub/new.txt\nx" if replacement == "new-target" else "sub/a.txt"
|
|
tool = "write_file" if replacement == "new-target" else "read_file"
|
|
bound = resolve(authority(root, tool), tool, content)
|
|
if replacement == "file":
|
|
target.rename(parent / "old.txt")
|
|
target.write_text("new")
|
|
elif replacement == "parent":
|
|
parent.rename(root / "old-sub")
|
|
parent.mkdir()
|
|
target.write_text("new")
|
|
elif replacement == "root":
|
|
root.rename(tmp_path / "old-root")
|
|
root.mkdir()
|
|
else:
|
|
(parent / "new.txt").write_text("unapproved target")
|
|
with pytest.raises((ValueError, OSError)):
|
|
bound.validate()
|
|
|
|
|
|
def test_content_is_not_an_object_incarnation_or_effect_claim(tmp_path):
|
|
target = tmp_path / "a"
|
|
target.write_text("old")
|
|
bound = resolve(authority(tmp_path, "read_file"), "read_file", "a")
|
|
target.write_text("changed content in the same object")
|
|
bound.validate()
|
|
|
|
|
|
@pytest.mark.parametrize("state", ["authority", "jobs", "containment"])
|
|
async def test_user_filesystem_scope_cannot_write_server_execution_state(tmp_path, monkeypatch, state):
|
|
import src.constants
|
|
monkeypatch.setattr(src.constants, "BG_JOBS_DIR", str(tmp_path / "jobs"))
|
|
monkeypatch.setattr(src.constants, "BG_JOBS_FILE", str(tmp_path / "jobs.json"))
|
|
monkeypatch.setattr(src.constants, "CONTAINMENT_STATE_FILE", str(tmp_path / "receipts.json"))
|
|
target = {"authority": "jobs/job.authority.json", "jobs": "jobs.json", "containment": "receipts.json"}[state]
|
|
_, result = await dispatch(authority(tmp_path, "write_file"), "write_file", target + "\nforged")
|
|
assert result["failure_kind"] == "resource_identity_denied"
|
|
assert not (tmp_path / target).exists()
|
|
|
|
|
|
@pytest.mark.parametrize("state", ["authority", "jobs", "containment", "result", "exit", "database", "vault", "uploads"])
|
|
@pytest.mark.parametrize("alias", ["direct", "relative", "symlink", "hardlink"])
|
|
async def test_control_files_cannot_be_read_or_written_through_aliases(tmp_path, monkeypatch, state, alias):
|
|
import src.constants as constants
|
|
jobs = tmp_path / "jobs"
|
|
jobs.mkdir()
|
|
monkeypatch.setattr(constants, "BG_JOBS_DIR", str(jobs))
|
|
monkeypatch.setattr(constants, "DATA_DIR", str(tmp_path))
|
|
monkeypatch.setattr(constants, "UPLOAD_DIR", str(tmp_path / "uploads"))
|
|
for name, filename in (("BG_JOBS_FILE", "jobs.json"), ("CONTAINMENT_STATE_FILE", "receipts.json"),
|
|
("APP_DB", "private.db"), ("VAULT_FILE", "vault.json")):
|
|
monkeypatch.setattr(constants, name, str(tmp_path / filename))
|
|
filename = {"authority": "jobs/job.authority.json", "jobs": "jobs.json", "containment": "receipts.json",
|
|
"result": "jobs/job.result.json", "exit": "jobs/job.exit", "database": "private.db",
|
|
"vault": "vault.json", "uploads": "uploads/uploads.json"}[state]
|
|
target = tmp_path / filename
|
|
target.parent.mkdir(exist_ok=True)
|
|
target.write_text("control-secret")
|
|
selector = str(target)
|
|
if alias == "relative":
|
|
selector = "./" + filename
|
|
elif alias in {"symlink", "hardlink"}:
|
|
link = tmp_path / "ordinary.txt"
|
|
try:
|
|
link.symlink_to(target) if alias == "symlink" else os.link(target, link)
|
|
except OSError as error:
|
|
pytest.skip(f"Platform cannot create {alias}: {error}")
|
|
selector = str(link)
|
|
grant = authority(tmp_path, "read_file", "write_file")
|
|
for tool, content in (("read_file", selector), ("write_file", selector + "\nforged")):
|
|
_, result = await dispatch(grant, tool, content)
|
|
assert result["failure_kind"] == "resource_identity_denied"
|
|
assert target.read_text() == "control-secret"
|
|
|
|
|
|
async def test_directory_grep_does_not_scan_control_state_or_hardlinks(tmp_path, monkeypatch):
|
|
import src.constants as constants
|
|
control = tmp_path / "jobs.json"
|
|
control.write_text("UNIQUE_CONTROL_SECRET")
|
|
(tmp_path / "ordinary").write_text("visible text")
|
|
os.link(control, tmp_path / "innocent.txt")
|
|
monkeypatch.setattr(constants, "BG_JOBS_FILE", str(control))
|
|
_, result = await dispatch(authority(tmp_path, "grep"), "grep", '{"pattern":"UNIQUE_CONTROL_SECRET","path":"."}')
|
|
assert result["exit_code"] == 0
|
|
assert "No matches" in result["output"]
|
|
|
|
|
|
@pytest.mark.parametrize("tool,content", [("glob", '{"pattern":"*.json","path":"."}'), ("ls", ".")])
|
|
async def test_directory_enumeration_does_not_address_control_files(tmp_path, monkeypatch, tool, content):
|
|
import src.constants as constants
|
|
control = tmp_path / "jobs.json"
|
|
control.write_text("control")
|
|
monkeypatch.setattr(constants, "BG_JOBS_FILE", str(control))
|
|
_, result = await dispatch(authority(tmp_path, tool), tool, content)
|
|
assert result["exit_code"] == 0
|
|
assert "jobs.json" not in result["output"]
|
|
|
|
|
|
@pytest.mark.parametrize("producer", ["database", "containment", "jobs", "uploads"])
|
|
@pytest.mark.parametrize("alias", ["direct", "hardlink"])
|
|
async def test_configured_control_producer_paths_are_protected(tmp_path, monkeypatch, producer, alias):
|
|
target = tmp_path / "custom" / "state"
|
|
target.parent.mkdir()
|
|
if producer == "database":
|
|
import core.database as database
|
|
monkeypatch.setattr(database, "engine", SimpleNamespace(url=SimpleNamespace(
|
|
get_backend_name=lambda: "sqlite", database=str(target))))
|
|
elif producer == "containment":
|
|
from src import containment
|
|
monkeypatch.setattr(containment, "_store_path", lambda: target)
|
|
elif producer == "jobs":
|
|
from src import bg_jobs
|
|
monkeypatch.setattr(bg_jobs, "_STORE", target)
|
|
else:
|
|
from src import tool_utils
|
|
target = target.parent / "uploads.json"
|
|
monkeypatch.setattr(tool_utils, "get_upload_handler", lambda: SimpleNamespace(upload_dir=str(target.parent)))
|
|
target.write_text("server state")
|
|
selector = str(target)
|
|
if alias == "hardlink":
|
|
link = tmp_path / "ordinary"
|
|
os.link(target, link)
|
|
selector = str(link)
|
|
_, result = await dispatch(authority(tmp_path, "read_file"), "read_file", selector)
|
|
assert result["failure_kind"] == "resource_identity_denied"
|
|
|
|
|
|
@pytest.mark.parametrize("roots", [(), None])
|
|
async def test_nonworkspace_allowlist_and_operation_do_not_grant_resources(tmp_path, monkeypatch, roots):
|
|
from src import tool_execution as execution
|
|
target = tmp_path / "a"
|
|
target.write_text("private")
|
|
monkeypatch.setattr(execution, "_tool_path_roots", lambda: [str(tmp_path)])
|
|
implementation = AsyncMock()
|
|
monkeypatch.setattr(execution, "_execute_tool_block_impl", implementation)
|
|
grant = authority(None, "read_file", roots=roots)
|
|
_, result = await dispatch(grant, "read_file", str(target))
|
|
assert result["failure_kind"] == "resource_identity_denied"
|
|
implementation.assert_not_awaited()
|
|
|
|
|
|
async def test_explicit_private_root_requires_owner_and_operation(tmp_path):
|
|
(tmp_path / "a").write_text("owned")
|
|
root = FilesystemRoot.seal(tmp_path, scope=FilesystemScope.PRIVATE, owner="alice")
|
|
with pytest.raises(ValueError):
|
|
authority(None, "read_file", roots=(root,), owner="bob")
|
|
grant = authority(None, "read_file", roots=(root,))
|
|
_, result = await dispatch(grant, "read_file", "a")
|
|
assert result["output"] == "owned"
|
|
_, result = await dispatch(grant, "write_file", "b\nx")
|
|
assert result["failure_kind"] == "request_authority_denied"
|
|
assert not (tmp_path / "b").exists()
|
|
|
|
|
|
@pytest.mark.parametrize("content", ['{"path":null}', '{"path":42}', '{"path":[]}', '{"path":{}}', '{"path":"a","path":"b"}'])
|
|
async def test_model_cannot_supply_or_reconstruct_a_resource(tmp_path, monkeypatch, content):
|
|
from src import tool_execution as execution
|
|
implementation = AsyncMock()
|
|
monkeypatch.setattr(execution, "_execute_tool_block_impl", implementation)
|
|
_, result = await dispatch(authority(tmp_path, "read_file"), "read_file", content)
|
|
assert result["blocked"] is True
|
|
implementation.assert_not_awaited()
|
|
|
|
|
|
async def test_model_root_field_is_not_authority(tmp_path, monkeypatch):
|
|
from src import tool_execution as execution
|
|
implementation = AsyncMock()
|
|
monkeypatch.setattr(execution, "_execute_tool_block_impl", implementation)
|
|
_, result = await dispatch(authority(None, "write_file"), "write_file",
|
|
json.dumps({"path": str(tmp_path / "a"), "content": "x", "resource_roots": [str(tmp_path)]}))
|
|
assert result["failure_kind"] == "resource_identity_denied"
|
|
implementation.assert_not_awaited()
|
|
|
|
|
|
def test_child_intersects_root_and_preserves_workspace_alias_base(tmp_path):
|
|
sub = tmp_path / "sub"
|
|
sub.mkdir()
|
|
(sub / "a").write_text("child")
|
|
(tmp_path / "outside").write_text("parent")
|
|
parent = authority(tmp_path, "read_file")
|
|
narrow = FilesystemRoot.seal(sub, owner="alice")
|
|
child = authority(tmp_path, "read_file", roots=(narrow,))
|
|
for effective in (parent.intersect(child), child.intersect(parent)):
|
|
assert effective.resource_roots == (narrow,)
|
|
assert resolve(effective, "read_file", "/workspace/sub/a").bindings[0].resource.path == str(sub / "a")
|
|
with pytest.raises(ValueError):
|
|
resolve(effective, "read_file", "/workspace/outside")
|
|
assert parent.intersect(authority(tmp_path, "read_file", roots=())).resource_roots == ()
|
|
assert parent.intersect(authority(tmp_path, "read_file", owner="bob")).resource_roots == ()
|
|
|
|
|
|
def test_child_cannot_renew_replaced_parent_root(tmp_path):
|
|
root = tmp_path / "root"
|
|
root.mkdir()
|
|
parent = authority(root, "read_file")
|
|
root.rename(tmp_path / "old")
|
|
root.mkdir()
|
|
child = authority(root, "read_file")
|
|
assert parent.intersect(child).resource_roots == ()
|
|
|
|
|
|
@pytest.mark.parametrize("caller", ["intersection", "context", "task"])
|
|
@pytest.mark.parametrize("child_location", ["root", "subtree"])
|
|
def test_replaced_parent_cannot_be_renewed_by_new_child_observation(tmp_path, caller, child_location):
|
|
root = tmp_path / "root"
|
|
root.mkdir()
|
|
parent = authority(root, "read_file")
|
|
root.rename(tmp_path / "old")
|
|
root.mkdir()
|
|
sub = root / "sub"
|
|
sub.mkdir()
|
|
(sub / "a").write_text("replacement")
|
|
child_root = FilesystemRoot.seal(root if child_location == "root" else sub, owner="alice")
|
|
child = authority(root, "read_file", roots=(child_root,))
|
|
if caller == "intersection":
|
|
effective = parent.intersect(child)
|
|
elif caller == "context":
|
|
with bind_request_authority(parent), bind_request_authority(child) as effective:
|
|
assert effective.resource_roots == ()
|
|
else:
|
|
with bind_request_authority(parent):
|
|
sealed = seal_task_authority("Read files in the workspace", "llm", None, owner="alice")
|
|
effective = restore_task_authority(sealed, "Read files in the workspace", "llm", None,
|
|
owner="alice", session_id="continuation")
|
|
assert effective.resource_roots == ()
|
|
with pytest.raises(ValueError):
|
|
resolve(effective, "read_file", "sub/a")
|
|
|
|
|
|
def test_equal_stale_roots_are_revalidated(tmp_path):
|
|
root = tmp_path / "root"
|
|
root.mkdir()
|
|
parent = authority(root, "read_file")
|
|
root.rename(tmp_path / "old")
|
|
root.mkdir()
|
|
assert parent.intersect(parent).resource_roots == ()
|
|
|
|
|
|
@pytest.mark.parametrize("legacy", [False, True])
|
|
async def test_snapshot_preserves_incarnation_and_never_reconstructs_legacy(tmp_path, legacy):
|
|
root = tmp_path / "root"
|
|
root.mkdir()
|
|
(root / "a").write_text("original")
|
|
grant = authority(root, "read_file")
|
|
snapshot = grant.to_dict()
|
|
if legacy:
|
|
snapshot["version"] = 1
|
|
snapshot.pop("resource_roots")
|
|
restored = RequestAuthority.from_dict(json.loads(json.dumps(snapshot)))
|
|
assert restored.resource_roots == (() if legacy else grant.resource_roots)
|
|
root.rename(tmp_path / "old")
|
|
root.mkdir()
|
|
(root / "a").write_text("replacement")
|
|
_, result = await dispatch(restored, "read_file", "a")
|
|
assert result["failure_kind"] == "resource_identity_denied"
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", [None, "root", [{}], [{"path": "/", "scope": "workspace", "identity": {"device": 1, "inode": 2, "kind": "directory"}, "owner": "alice"}]])
|
|
def test_malformed_resource_snapshots_are_rejected(tmp_path, mutation):
|
|
snapshot = authority(tmp_path, "read_file").to_dict()
|
|
snapshot["resource_roots"] = mutation
|
|
with pytest.raises((TypeError, ValueError, KeyError)):
|
|
RequestAuthority.from_dict(snapshot)
|
|
|
|
|
|
def test_task_and_background_continuations_keep_original_roots(tmp_path, monkeypatch):
|
|
import src.constants
|
|
monkeypatch.setattr(src.constants, "BG_JOBS_DIR", str(tmp_path))
|
|
grant = authority(tmp_path, "read_file")
|
|
# A roots-only sidecar is legacy state and cannot invent a job generation.
|
|
with pytest.raises(ValueError):
|
|
save_background_authority("job", grant)
|
|
assert restore_background_authority("job", owner="alice", session_id="s").resource_roots == ()
|
|
assert restore_background_authority("job", owner="bob", session_id="s").resource_roots == ()
|
|
with bind_request_authority(grant):
|
|
sealed = seal_task_authority("Read files in the workspace", "llm", None, owner="alice")
|
|
assert restore_task_authority(sealed, "Read files in the workspace", "llm", None,
|
|
owner="alice", session_id="continuation").resource_roots == grant.resource_roots
|
|
|
|
|
|
async def test_dispatch_consumes_canonical_binding_and_pins_native_backend(tmp_path, monkeypatch):
|
|
from src import tool_execution as execution
|
|
import src.agent_tools
|
|
(tmp_path / "a").write_text("bound")
|
|
(tmp_path / "alias").symlink_to(tmp_path / "a")
|
|
handler = AsyncMock(return_value={"output": "handled", "exit_code": 0})
|
|
mcp = AsyncMock()
|
|
monkeypatch.setitem(src.agent_tools.TOOL_HANDLERS, "read_file", handler)
|
|
monkeypatch.setattr(execution, "get_mcp_manager", lambda: mcp)
|
|
_, result = await dispatch(authority(tmp_path, "read_file"), "read_file", "alias")
|
|
assert result["output"] == "handled"
|
|
content, ctx = handler.call_args.args
|
|
assert json.loads(content)["path"] == str(tmp_path / "a")
|
|
assert ctx["resource_operation"].bindings[0].resource.path == str(tmp_path / "a")
|
|
assert ctx["resource_operation"].request_id == "resource-test"
|
|
mcp.call_tool.assert_not_awaited()
|
|
assert active_resource_operation() is None
|
|
|
|
|
|
def test_bound_resolver_rejects_undeclared_paths_and_scopes_search(tmp_path):
|
|
from src.tool_execution import _resolve_tool_path, _resolve_search_root
|
|
sub = tmp_path / "sub"
|
|
sub.mkdir()
|
|
(sub / "a").write_text("a")
|
|
(tmp_path / "outside").write_text("outside")
|
|
grant = authority(tmp_path, "read_file", "grep")
|
|
bound = resolve(grant, "read_file", "sub/a")
|
|
with bind_resource_operation(bound):
|
|
assert _resolve_tool_path(str(sub / "a")) == str(sub / "a")
|
|
with pytest.raises(ValueError):
|
|
_resolve_tool_path(str(tmp_path / "outside"))
|
|
search = resolve(grant, "grep", '{"pattern":"a","path":"sub"}')
|
|
with bind_resource_operation(search):
|
|
assert _resolve_search_root("") == str(sub)
|
|
assert _resolve_tool_path(str(sub / "a")) == str(sub / "a")
|
|
with pytest.raises(ValueError):
|
|
_resolve_tool_path(str(tmp_path / "outside"))
|
|
|
|
|
|
async def test_concurrent_resource_contexts_do_not_leak(tmp_path, monkeypatch):
|
|
from src import tool_execution as execution
|
|
arrived = asyncio.Event()
|
|
seen = []
|
|
async def implementation(block, **kwargs):
|
|
bound = active_resource_operation()
|
|
seen.append(bound.bindings[0].resource.path)
|
|
if len(seen) == 2:
|
|
arrived.set()
|
|
await arrived.wait()
|
|
assert active_resource_operation() is bound
|
|
return "read", {"exit_code": 0}
|
|
monkeypatch.setattr(execution, "_execute_tool_block_impl", implementation)
|
|
for name in ("a", "b"):
|
|
(tmp_path / name).write_text(name)
|
|
grant = authority(tmp_path, "read_file")
|
|
await asyncio.gather(dispatch(grant, "read_file", "a"), dispatch(grant, "read_file", "b"))
|
|
assert set(seen) == {str(tmp_path / "a"), str(tmp_path / "b")}
|
|
assert active_resource_operation() is None
|
|
|
|
|
|
async def test_last_dispatch_validation_refuses_replacement_and_resets_context(tmp_path, monkeypatch):
|
|
from src import tool_execution as execution
|
|
target = tmp_path / "a"
|
|
target.write_text("old")
|
|
implementation = AsyncMock()
|
|
monkeypatch.setattr(execution, "_execute_tool_block_impl", implementation)
|
|
security = ToolRunSecurityContext()
|
|
def decision(*args):
|
|
target.rename(tmp_path / "old-a")
|
|
target.write_text("new")
|
|
return SimpleNamespace(allowed=True)
|
|
monkeypatch.setattr(security, "decision_for", decision)
|
|
_, result = await dispatch(authority(tmp_path, "read_file"), "read_file", "a", security_context=security)
|
|
assert result["failure_kind"] == "resource_identity_denied"
|
|
implementation.assert_not_awaited()
|
|
assert active_resource_operation() is None
|
|
assert execution.get_active_workspace() is None
|
|
|
|
|
|
@pytest.mark.parametrize("error_type", [None, RuntimeError, asyncio.CancelledError])
|
|
async def test_nested_resource_context_restores_on_failure_or_cancellation(tmp_path, monkeypatch, error_type):
|
|
from src import tool_execution as execution
|
|
for name in ("parent", "child"):
|
|
(tmp_path / name).write_text(name)
|
|
grant = authority(tmp_path, "read_file")
|
|
parent = resolve(grant, "read_file", "parent")
|
|
async def implementation(block, **kwargs):
|
|
assert active_resource_operation().bindings[0].resource.path == str(tmp_path / "child")
|
|
if error_type:
|
|
raise error_type("stop")
|
|
return "read", {"exit_code": 0}
|
|
monkeypatch.setattr(execution, "_execute_tool_block_impl", implementation)
|
|
with bind_resource_operation(parent):
|
|
if error_type:
|
|
with pytest.raises(error_type):
|
|
await dispatch(grant, "read_file", "child")
|
|
else:
|
|
await dispatch(grant, "read_file", "child")
|
|
assert active_resource_operation() is parent
|
|
assert active_resource_operation() is None
|
|
assert execution.get_active_workspace() is None
|
|
|
|
|
|
@pytest.mark.parametrize("kind", ["collision", "hardlink", "escape", "move"])
|
|
async def test_patch_validates_all_targets_before_any_write(tmp_path, kind):
|
|
(tmp_path / "a").write_text("old\n")
|
|
(tmp_path / "alias").symlink_to(tmp_path / "a")
|
|
os.link(tmp_path / "a", tmp_path / "hardlink")
|
|
suffix = {
|
|
"collision": "*** Update File: alias\n@@\n-old\n+second",
|
|
"hardlink": "*** Update File: hardlink\n@@\n-old\n+second",
|
|
"escape": "*** Add File: ../escape.txt\n+escaped",
|
|
"move": "*** Update File: alias\n*** Move to: moved\n@@\n-old\n+moved",
|
|
}[kind]
|
|
patch = f"*** Begin Patch\n*** Update File: a\n@@\n-old\n+new\n{suffix}\n*** End Patch"
|
|
_, result = await dispatch(authority(tmp_path, "apply_patch"), "apply_patch", patch)
|
|
assert result["failure_kind"] == "resource_identity_denied"
|
|
assert (tmp_path / "a").read_text() == "old\n"
|
|
assert not (tmp_path / "moved").exists()
|
|
|
|
|
|
def test_move_contract_binds_both_distinct_resources(tmp_path):
|
|
(tmp_path / "a").write_text("source")
|
|
root = FilesystemRoot.seal(tmp_path)
|
|
source = ResourceBinding("source", FilesystemResource.resolve(root, "a"))
|
|
destination = ResourceBinding("destination", FilesystemResource.resolve(root, "b", allow_missing=True))
|
|
operation = ExactOperation("move_file", "a -> b", "move", "move_file")
|
|
with pytest.raises(ValueError):
|
|
BoundFilesystemOperation(operation, "", (source,))
|
|
BoundFilesystemOperation(operation, "", (source, destination)).validate()
|
|
|
|
|
|
def approval(grant, tool, content):
|
|
store = ToolApprovalStore()
|
|
pending = store.create(owner=grant.owner, session_id=grant.session_id,
|
|
origin_run_id="run", tool_name=tool, content=content, workspace=grant.workspace,
|
|
external_untrusted_context_seen=True, capabilities=capabilities_for_action(tool, content),
|
|
request_authority=grant)
|
|
exact = store.consume(pending.approval_id, decision="approve", owner=grant.owner, session_id=grant.session_id)
|
|
security = ToolRunSecurityContext()
|
|
security.external_untrusted_context_seen = True
|
|
return exact, security
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["alias", "file", "parent"])
|
|
async def test_approval_resource_retargeting_refuses_without_claiming(tmp_path, change):
|
|
parent = tmp_path / "sub"
|
|
parent.mkdir()
|
|
(parent / "a").write_text("a")
|
|
(parent / "b").write_text("b")
|
|
alias = parent / "alias"
|
|
alias.symlink_to(parent / "a")
|
|
grant = authority(tmp_path, "read_file")
|
|
exact, security = approval(grant, "read_file", "sub/alias")
|
|
assert exact.pending.resource_operation is not None
|
|
if change == "alias":
|
|
alias.unlink()
|
|
alias.symlink_to(parent / "b")
|
|
elif change == "file":
|
|
(parent / "a").rename(parent / "old-a")
|
|
(parent / "a").write_text("replacement")
|
|
else:
|
|
parent.rename(tmp_path / "old-sub")
|
|
parent.mkdir()
|
|
(parent / "a").write_text("replacement")
|
|
alias.symlink_to(parent / "a")
|
|
_, result = await dispatch(grant, "read_file", "sub/alias", exact_approval=exact, security_context=security)
|
|
assert result["failure_kind"] == "resource_identity_denied"
|
|
assert exact.matches(owner="alice", session_id="s", workspace=str(tmp_path), tool_name="read_file", content="sub/alias")
|
|
|
|
|
|
async def test_approval_is_exact_and_one_use_with_immutable_resource_snapshot(tmp_path):
|
|
(tmp_path / "a").write_text("a")
|
|
grant = authority(tmp_path, "read_file")
|
|
exact, security = approval(grant, "read_file", "a")
|
|
with pytest.raises(FrozenInstanceError):
|
|
exact.pending.resource_operation.execution_input = "other"
|
|
assert "resource_operation" not in exact.pending.public_payload()
|
|
_, modified = await dispatch(grant, "read_file", "/workspace/a", exact_approval=exact, security_context=security)
|
|
assert modified["exit_code"] == 1
|
|
_, result = await dispatch(grant, "read_file", "a", exact_approval=exact, security_context=security)
|
|
assert result["output"] == "a"
|
|
_, replay = await dispatch(grant, "read_file", "a", exact_approval=exact, security_context=security)
|
|
assert replay["exit_code"] == 1
|
|
|
|
|
|
async def test_exact_approval_cannot_widen_a_child_resource_scope(tmp_path):
|
|
sub = tmp_path / "sub"
|
|
sub.mkdir()
|
|
(tmp_path / "outside").write_text("parent")
|
|
parent = authority(tmp_path, "read_file")
|
|
exact, security = approval(parent, "read_file", "outside")
|
|
child = replace(parent, resource_roots=(FilesystemRoot.seal(sub, owner="alice"),))
|
|
with bind_request_authority(parent), bind_request_authority(child) as effective:
|
|
_, result = await dispatch(effective, "read_file", "outside", exact_approval=exact, security_context=security)
|
|
assert result["failure_kind"] == "resource_identity_denied"
|
|
|
|
|
|
async def test_approved_resource_cannot_migrate_to_another_request(tmp_path):
|
|
(tmp_path / "a").write_text("original request")
|
|
grant = authority(tmp_path, "read_file")
|
|
exact, security = approval(grant, "read_file", "a")
|
|
_, result = await dispatch(replace(grant, request_id="new-request"), "read_file", "a",
|
|
exact_approval=exact, security_context=security)
|
|
assert result["failure_kind"] == "resource_identity_denied"
|
|
|
|
|
|
async def test_missing_approval_resource_snapshot_cannot_be_reconstructed(tmp_path, monkeypatch):
|
|
grant = authority(tmp_path, "read_file")
|
|
def unavailable(*args, **kwargs):
|
|
raise PermissionError("Cannot establish the proposal's resource identity")
|
|
|
|
with monkeypatch.context() as patch:
|
|
patch.setattr("src.agent_runtime.resource_binding.resolve_filesystem_operation", unavailable)
|
|
exact, security = approval(grant, "read_file", "missing")
|
|
assert exact.pending.resource_operation is None
|
|
(tmp_path / "missing").write_text("appeared after proposal")
|
|
_, result = await dispatch(grant, "read_file", "missing", exact_approval=exact, security_context=security)
|
|
assert result["failure_kind"] == "resource_identity_denied"
|
|
|
|
|
|
async def test_approved_absent_read_cannot_bind_a_file_that_appeared(tmp_path):
|
|
grant = authority(tmp_path, "read_file")
|
|
exact, security = approval(grant, "read_file", "missing")
|
|
assert exact.pending.resource_operation.bindings[0].resource.identity is None
|
|
(tmp_path / "missing").write_text("appeared after proposal")
|
|
_, result = await dispatch(grant, "read_file", "missing", exact_approval=exact, security_context=security)
|
|
assert result["failure_kind"] == "resource_identity_denied"
|
|
|
|
|
|
async def test_exact_user_approval_binds_only_one_missing_destination(tmp_path):
|
|
grant = RequestAuthority.empty(owner="alice", session_id="s", workspace=str(tmp_path))
|
|
exact, security = approval(grant, "write_file", "new/file.txt\napproved")
|
|
_, result = await dispatch(grant, "write_file", "new/file.txt\napproved", exact_approval=exact, security_context=security)
|
|
assert result["exit_code"] == 0
|
|
assert (tmp_path / "new/file.txt").read_text() == "approved"
|
|
assert grant.grants == () and grant.resource_roots == ()
|
|
_, next_action = await dispatch(grant, "write_file", "other.txt\nunapproved")
|
|
assert next_action["failure_kind"] == "request_authority_denied"
|
|
assert not (tmp_path / "other.txt").exists()
|
|
|
|
|
|
@pytest.mark.parametrize("version", [1, 2])
|
|
async def test_restored_empty_roots_approval_is_exact_and_never_restores_generic_scope(tmp_path, version):
|
|
(tmp_path / "approved").write_text("approved content")
|
|
(tmp_path / "sibling").write_text("private sibling")
|
|
snapshot = authority(tmp_path, "read_file", "write_file", "ls").to_dict()
|
|
snapshot["version"] = version
|
|
snapshot["resource_roots"] = []
|
|
restored = RequestAuthority.from_dict(snapshot)
|
|
exact, security = approval(restored, "read_file", "approved")
|
|
assert exact.pending.resource_operation is not None
|
|
for tool, content in (("read_file", "sibling"), ("ls", "."), ("write_file", "sibling\nx")):
|
|
_, blocked = await dispatch(restored, tool, content, exact_approval=exact, security_context=security)
|
|
assert blocked["exit_code"] == 1
|
|
_, unapproved = await dispatch(restored, tool, content)
|
|
assert unapproved["failure_kind"] == "resource_identity_denied"
|
|
_, allowed = await dispatch(restored, "read_file", "approved", exact_approval=exact, security_context=security)
|
|
assert allowed["output"] == "approved content"
|
|
_, replay = await dispatch(restored, "read_file", "approved", exact_approval=exact, security_context=security)
|
|
assert replay["exit_code"] == 1
|
|
assert restored.resource_roots == () and restored.backend_resources == ()
|
|
assert (tmp_path / "sibling").read_text() == "private sibling"
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["alias", "request", "session", "owner"])
|
|
async def test_restored_exact_filesystem_binding_rejects_retarget_and_rebinding(tmp_path, change):
|
|
(tmp_path / "a").write_text("a")
|
|
(tmp_path / "b").write_text("b")
|
|
(tmp_path / "alias").symlink_to(tmp_path / "a")
|
|
snapshot = authority(tmp_path, "read_file").to_dict()
|
|
snapshot["version"] = 1
|
|
restored = RequestAuthority.from_dict(snapshot)
|
|
exact, security = approval(restored, "read_file", "alias")
|
|
if change == "alias":
|
|
(tmp_path / "alias").unlink()
|
|
(tmp_path / "alias").symlink_to(tmp_path / "b")
|
|
else:
|
|
restored = replace(restored, **{"request": {"request_id": "other"},
|
|
"session": {"session_id": "other"}, "owner": {"owner": "bob"}}[change])
|
|
_, result = await dispatch(restored, "read_file", "alias", exact_approval=exact, security_context=security)
|
|
assert result["exit_code"] == 1
|
|
assert exact.matches(owner="alice", session_id="s", workspace=str(tmp_path), tool_name="read_file", content="alias")
|
|
|
|
|
|
async def test_resumed_child_approval_cannot_renew_replaced_parent_root(tmp_path):
|
|
root = tmp_path / "root"
|
|
root.mkdir()
|
|
parent = authority(root, "read_file")
|
|
root.rename(tmp_path / "old")
|
|
root.mkdir()
|
|
(root / "new").write_text("replacement")
|
|
child = parent.intersect(authority(root, "read_file"))
|
|
exact, security = approval(child, "read_file", "new")
|
|
assert child.resource_roots == () and exact.pending.resource_operation is None
|
|
_, result = await dispatch(replace(child, inherited=False), "read_file", "new", exact_approval=exact, security_context=security)
|
|
assert result["failure_kind"] == "resource_identity_denied" and not exact._claimed
|
|
|
|
|
|
@pytest.mark.parametrize("request_text,denied", [
|
|
("Transcribe /workspace/audio.wav", "read_file"),
|
|
("OCR extract exact text from /workspace/image.png", "write_file"),
|
|
("List my tasks", "read_file"),
|
|
])
|
|
async def test_resource_identity_never_expands_narrow_request_classes(tmp_path, request_text, denied):
|
|
(tmp_path / "a").write_text("a")
|
|
grant = create_request_authority(request_text, owner="alice", session_id="s", workspace=str(tmp_path))
|
|
_, result = await dispatch(grant, denied, "a" if denied == "read_file" else "a\nx")
|
|
assert result["failure_kind"] == "request_authority_denied"
|
|
|
|
|
|
def test_nonfilesystem_identities_are_inert_and_distinguish_producers_from_pages():
|
|
from src.process_lifecycle import ProcessIdentity
|
|
ProcessResource("native:containment", "alice", "request", "thread", ProcessIdentity(123, "boot:start"), "leader", "job", "receipt")
|
|
OwnedResource("documents", "alice", "thread", "documents", "document", "revision")
|
|
assert ExternalResource("mcp", "endpoint", "server", "tool", "connection").external is True
|
|
with pytest.raises(ValueError):
|
|
ExternalResource("mcp", "endpoint", "server", "tool", "connection", external=False)
|
|
with pytest.raises(ValueError):
|
|
ProcessResource("native:containment", "alice", "request", "thread", ProcessIdentity(123, ""), "leader", containment_id="receipt")
|