mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-10-07 07:22:21 +02:00
507 lines
24 KiB
Python
507 lines
24 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 (
|
|
BrowserPageResource, BrowserProducer, 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")
|
|
|
|
|
|
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("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("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")
|
|
save_background_authority("job", grant)
|
|
assert restore_background_authority("job", owner="alice", session_id="s").resource_roots == grant.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", [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")
|
|
raise error_type("stop")
|
|
monkeypatch.setattr(execution, "_execute_tool_block_impl", implementation)
|
|
with bind_resource_operation(parent):
|
|
with pytest.raises(error_type):
|
|
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):
|
|
grant = authority(tmp_path, "read_file")
|
|
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_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("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():
|
|
producer = BrowserProducer("browser", "alice", "thread", "session", "incarnation-1")
|
|
page = BrowserPageResource(producer, "page-1", 2, "https://example.test")
|
|
assert replace(producer, incarnation="incarnation-2") != producer
|
|
assert replace(page, navigation_generation=3) != page
|
|
ProcessResource("local", "boot/process", "alice", 123, "boot:start", "job", "receipt", 124, "boot:init")
|
|
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("local", "incarnation", "alice", 123, "", containment_id="receipt")
|