diff --git a/docs/runtime-decomposition/wave-3-resource-identity.md b/docs/runtime-decomposition/wave-3-resource-identity.md new file mode 100644 index 000000000..949474ee2 --- /dev/null +++ b/docs/runtime-decomposition/wave-3-resource-identity.md @@ -0,0 +1,241 @@ +# Wave 3: server-owned resource identity + +Audit base: `a80c164dbe3e8bde4fb29b45c5d1c61404f2fede` on +`feature/runtime-resource-authority`. The read-only audit and this design precede +production edits. Wave 3-S is frozen. This document distinguishes the contract +from the initial enforcement slice; it does not claim all resource adapters are +migrated. + +## A. Current implicit-resource inventory + +| Boundary / locator | Existing authority | Resource still interpreted later | +| --- | --- | --- | +| `src/agent_runtime/authority.py`: `ExactOperation`, `OperationGrant`, `RequestAuthority` | Immutable request, owner/session/workspace, action/input limits, policy denials | Workspace is a string; no root incarnation, object, destination or backend binding. | +| `src/turn_contract.py`: `TurnContract`, `canonical_tool` | Inventory narrows operations; email aliases share policy identity | Inventory/selection does not resolve resources. Bare/qualified email names can address one server. Transcription/OCR/tasks remain narrow. | +| `src/tool_execution.py`: `_tool_path_roots`, `_resolve_tool_path`, `_resolve_search_root` | Operation admission and deployment/public/admin policy | Data, system temp and configured extra roots are an access allowlist; relative paths may use process cwd; empty search path uses mutable defaults. An allowlist is not a request resource grant. | +| Same: `_resolve_tool_path_in_workspace`, `vet_workspace`, `_display_tool_path` | Trusted workspace string, sensitive-path deny policy | `/workspace`, relative/host paths and symlinks resolve later; root/object replacement is not represented. Display/evidence aliases do not confer access. | +| `src/path_confinement.py`: `canonical_root`, `confine` | Canonical inside-root check | Non-strict realpath intentionally supports missing destinations; it does not identify an existing object or grant a root. | +| `src/agent_tools/filesystem_tools.py`: read/write/edit, `ApplyPatchTool`, ls/glob/grep | Dispatcher gate and shared resolver | Handlers reparse paths; writes create parent directories; patches resolve each target and stage/backup by pathname. Different selectors may identify the same target. Patch moves are explicitly unsupported. Search binds a directory but derives descendants later. | +| `src/agent_runtime/identity.py`: `artifact_identity`, `artifact_version` | Evidence bookkeeping only | Workspace/absolute string identities and content hashes are completion evidence, not execution identities or authority. | +| `src/agent_tools/subprocess_tools.py`: `_owned_spec`, `_run_owned_command`, Bash/Python/host shell | Request operation grant then Wave 3-S containment | Cwd, environment, mount recipe and workspace aliases are interpreted at execution. Opaque scripts cannot be treated as an enumerated file operation. Host-shell endpoint/jobs belong to an external executor. | +| `src/containment.py`: `ContainmentSpec`, `ContainmentGrant`, `agent_spec`, `declare_external_bridge` | Frozen enforcement requirements | Receipt ID, owner label, PID/namespace PID and endpoint attest boundaries. They do not supply user permission or a request resource grant. | +| `src/process_ownership.py`: `capture`, `verify`, `start_token` | PID plus OS start token, Linux boot identity | A numeric PID alone is a reused slot. Tokens are inspection identities, not permissions. No new teardown/lifecycle algorithm belongs in Wave 3. | +| `src/bg_jobs.py`: `launch`, `get`, `kill`; `src/agent_tools/bg_job_tools.py` | Session check; verified process teardown | Job ID resolves through a mutable store. Supervisor PID/token, containment ID and namespace identity are separate. Session ownership is implicit rather than typed. | +| `src/agent_runtime/authority.py`: task/job snapshots; `src/bg_monitor.py`; `src/task_scheduler.py` | Parent intersection, sealed task input, continuation owner/session checks | Persisted workspace string can resolve to a replacement root. Missing snapshots fail closed. Session rebinding must not create resources. | +| `src/agent_tools/web_tools.py`: `_scoped_browser_session`, private-browser execution; `src/browser_lifecycle.py`: `BrowserSession`, `session_for`, `receipt` | Browser action class; server session hashing; producer locks | Namespace/session hash identifies a producer name, not its incarnation. Navigation generation, current URL, failed navigation and element references are mutable page state. URL/element selectors are not page identity. Receipts are not semantic verification. | +| `src/builtin_mcp.py`, `src/mcp_manager.py`: `call_tool`, reconnect, builtin browser | Qualified tool and policy gates | Server ID maps to a mutable connection/configuration; reconnect replaces producer. Builtin Playwright has a shared global browser. Stdio locally launches a third-party server but does not prove containment of its operations. | +| `src/tool_execution.py`: `AgentExecutionBridge`, `_client_bridge`, `_route_tool_via_bridge`, `_apply_patch_via_tui_host_bridge`, `_call_mcp_tool` | Explicit bridge routing after authority; exact approvals | Bridge callback/name, endpoint and context are resolved later; MCP-to-native fallback changes backend. Transport selection and availability must not authorize a backend/resource. External paths need the remote owner's contract, not local realpath or invented remote containment. | +| `src/agent_tools/document_tools.py`: `_get_owned_document`, `_most_recent_owned_document`, update/edit/suggest/manage | Owner-filtered DB lookup; approved ID/version/digest | Context target, process-global active document, model ID aliases and most-recent selection can choose targets late. Ownership alone does not establish that the request selected a document. | +| `src/agent_tools/media_tools.py`: `_resolve_workspace_path`, media/OCR/transcription implementations | Narrow operation class and local/upload checks | Workspace URI, local paths, confined host aliases, attachment URI and export/output aliases are separate resolution paths. Exports require source plus destinations; attachment IDs require owner-checked index identity. | +| `src/upload_handler.py`: `reserve_upload`, `resolve_upload`; `src/document_processor.py` | Ownership/index consistency and path confinement | Upload ID/hash/index aliases map to files; row/path/owner binding must be captured before consumption. Owner migration and cleanup can mutate mappings. | +| `src/agent_tools/session_tools.py`, `src/session_actions.py`, `src/session_search.py`, `src/tools/search.py` | Owner-filtered thread/history lookup | `current`, IDs, list/search result sets, fork targets and DB rows are reconstructed during execution. Null-owner handling differs by API and must remain explicit. A child thread never inherits authority by copying history. | +| `src/agent_tools/coding_tools.py`: `TodoWriteTool` | Tool/session context | Session text is sanitized into a filename and can fall back to model input/`current`; different strings may collide. This is private storage, not an ordinary workspace file. | +| `src/tools/notes.py`, `calendar.py`, `contacts.py`, `vault.py`, `research.py`, `image.py`, `system.py`, `cookbook.py`; admin tools and `app_api` | Owner/admin filters, operation gates, scheduled-task snapshots | Record ID/title/query/default account, task/action, model/server ID, preset, endpoint and API path select resources later. User collections and service credentials are private namespaces; installed tools/endpoints do not grant access. Broad app API and opaque host/script calls require dedicated backend contracts. | +| `src/tool_approvals.py`: pending digest, `matches`, `claim`; nested invocation tests | Exact one-use input, owner/session/workspace/document and original authority | File path is exact text but its alias/object can change between proposal and claim. Children may only intersect operation and resource scopes. No approval grants a later operation implicitly. | + +The inventory is of execution/resource-resolution seams. Internal renderer and +temporary implementation files are not independent user authority targets. Their +identity derives from the admitted operation's bounded root/backend contract. + +## B. Typed resource identity model + +Identity is inert, immutable server data. Model arguments remain selectors. +There is no model-facing deserializer that mints grants. + +* Filesystem: a root with scope (`workspace`, `scratch`, `external`, `private`), + canonical location and observed device/inode/type. An object has that root, + canonical path, target observation (or explicit absence) and existing ancestor + observations. Missing destinations retain their existing parent identity; + they are not imaginary inodes. Private roots additionally bind an owner. + Server execution-control stores and background authority sidecars cannot be + addressed as user filesystem resources, even beneath an admitted root. +* Process: backend/ownership namespace, producer incarnation, PID/start token, + optional namespace PID/start token, background job ID and containment receipt + linkage. A receipt reference is attribution only. New process execution first + binds its execution root/backend; PID identity only exists after spawn. +* Browser producer: backend namespace, owner/thread, producer session and + incarnation. Page observation: that producer plus navigation generation, + observed page ID/URL and producer reference. Lifecycle state is distinct from + page semantics, and neither establishes semantic correctness. +* External execution: backend namespace, endpoint identity, server/tool and + connection incarnation. Always explicitly external. Endpoint identities must + be sanitized identifiers, never credentials. No containment is inferred. +* Owned records: ownership namespace, exact owner, thread, collection and + record/document ID; revision when the producer supplies it. Collections used + for list/search are explicit owner-bound resources, not unknown record IDs. + +The initial implementation provides types for each domain. Only filesystem +resolution/admission is migrated; unused domain types do not attest existing +producers or silently supply missing incarnations. + +## C. Normalized operation/resource binding + +Retain the original `ExactOperation` for policy and approval matching. Add an +immutable bound operation containing request identity, canonical executor input +and role-tagged resources (`source`, `target`, `destination`, `search_root`). +Patch operations enumerate all targets before dispatch and reject canonical +path and observed object collisions (including hardlinks). Rename/move bindings require both source and destination; the +current native patch parser continues refusing moves. No shell text parsing is +used to pretend an opaque script has enumerated filesystem semantics. + +## D. Authority-to-resource validation flow + +1. Normalize the original tool/input; check RequestAuthority binding, parent + intersection, policy denials and exact operation grant/approval eligibility. +2. Apply the unchanged TurnContract and existing security/public/admin gates. +3. Resolve native filesystem selectors against roots sealed by the server, + apply existing confinement and sensitive-path policy, and observe identities. + Neither configured allowlists nor schema/bridge availability adds a root. +4. Compare approved resource snapshots before claiming the exact one-use action. + Revalidate root/object/ancestors; unresolved or changed identities refuse. +5. Dispatch canonical executor input under a context-local binding. Shared + resolvers consume that binding and reject undeclared paths; search traversal + remains bounded by the declared search resource and sensitive-path policy. +6. Existing effect/evidence/completion handling continues unchanged. + +Path observations and immediate revalidation detect replacement before +dispatch. They are not kernel-held file descriptors and cannot eliminate all +concurrent pathname races inside existing handlers. Closing those races requires +descriptor-relative I/O integration; this slice must not claim atomic identity +enforcement or change the frozen process containment mechanism. +Device/inode observations also cannot distinguish every possible inode reuse; +they are scoped local filesystem observations rather than globally permanent IDs. + +## E. Alias, rename and ownership rules + +`/workspace`, relative paths, host paths and symlinks resolve only on the server. +Executor input uses the resolved path; original input remains exact for approval. +Retargeting an approved alias changes its bound identity and refuses execution. +Both sides of any future move must resolve under admitted scopes before an +effect. A missing destination binds absence plus its existing ancestors. +Owner/thread mismatches fail; an ownership query proves attribution, not intent. +Children intersect roots by identical root observation and owner/scope, and may +narrow to descendant scopes. Empty intersections stay empty. Continuations and +persisted snapshots retain observations instead of re-sealing a changed root. + +## F. Integration points / chosen slice + +Add `src/agent_runtime/resources.py`, extend RequestAuthority with sealed +filesystem roots, and add the central native filesystem binder in +`src/agent_runtime/resource_binding.py`. Integrate read/write/edit/patch/ls/glob/ +grep with `execute_tool_block`, shared path resolvers and exact approval sealing. +Bridge-routed operations remain outside this native adapter; a local root must +not be used to invent a remote resource identity. Existing native search handlers +retain their descendant checks. No agent-loop decomposition or browser/process +lifecycle refactor is needed. + +Bare native filesystem operations now dispatch directly to their native handlers +with canonical input. A connected filesystem MCP server cannot redirect these +resources or supply an implicit fallback backend. Explicit qualified MCP calls +remain on the external path pending its producer/resource adapter. + +## G. Migration plan + +1. Initial slice: seal a vetted workspace at server authority construction; + permit explicit server-supplied scratch/external/private roots; serialize the + observations and intersect them. No implicit data/tmp/extra-root grant. +2. Version authority snapshots. Legacy snapshots retain operation restrictions + but receive no reconstructed filesystem roots. Missing roots refuse migrated + native tools. A new trusted request may seal new resources. +3. Integrate canonical native filesystem input and approved resource snapshots. + Existing fixtures requiring unscoped native files must explicitly grant a + test root; they cannot rely on broad production allowlists. +4. Follow-up adapters: media/attachment/export, document/thread/private stores, + job controls and native opaque execution root/recipe, then bridge/MCP and + browser producers. Each requires its own server-owned resolution seam and + must fail closed on absent producer identity. Do not fill gaps with string + hashes described as incarnations or generic capability floors. + +The narrow slice does not remove every implicit-resource site listed in A. +Its coverage and remaining adapters must be reported explicitly. +The server-control-store denial applies to this native filesystem adapter; +opaque scripts and other unmigrated adapters still need their own resource +boundaries. This slice does not attest those paths as enforcing the new contract. + +## H. Exact tests required + +* Root/target canonicalization: relative, host, `/workspace`, symlink aliases; + sibling/traversal/symlink escapes; sensitive files; malformed path/JSON/type. +* Existing files and directories; absent destination plus parent identity; + replacement of root, target or existing ancestor invalidates the binding. +* No roots means no migrated native execution, even with an offered handler, + configured allowlist, selected tool, valid operation grant or result receipt. +* Every patch target binds before dispatch; canonical target collisions and + unsupported moves refuse before partial writes. Dual-resource move contract. +* Canonical input reaches the handler; shared resolvers reject undeclared + targets; directory searches allow only bounded descendants. +* Parent/child root intersection, mismatch of owners/sessions, context cleanup, + concurrent calls, task/background persistence, malformed/legacy snapshots. +* Approval alias/target/parent replacement, immutable digest, missing resource + snapshot, exact original input, one-use replay and nested restriction. +* Regression suites: request authority, approvals, nested ownership, workspace + confinement, path policy, filesystem tools, execution bridges, TurnContract + (including transcription/OCR/tasks), frozen containment/native/background. +* Future adapters require job PID reuse/receipt mismatches, browser incarnation/ + page generation distinction, MCP reconnect/endpoint changes, cross-owner + attachment/record/thread rejection and exact dual-resource exports/moves. + +## I. Collision analysis with Wave 4 and Wave 5B + +Wave 3 binds what an admitted operation addresses. Device/inode observations +identify objects, not content versions or proof that an effect occurred. It adds +no durable claim, effects ledger, egress/provenance, evidence freshness rule or +truthful-completion mechanism (Wave 4). It adds no supervisor, restart/reaper, +cleanup state machine, generic lifecycle namespace allocator or process teardown +algorithm (Wave 5B). Process/browser producer incarnations must come from their +owners; this contract does not fabricate them. Frozen containment receipts and +browser lifecycle receipts remain evidence of their stated producer boundaries, +never authority or semantic verification. + +## Implementation validation + +Executed locally with `/usr/bin/python3` on 2026-10-02: + +* Integrated focused run: **1,649 passed, 2 skipped, 1 warning**. This includes + request identity linkage and approval matching, before the final hardlink + collision and resource-context unwind additions. +* Final follow-up after those additions: **109 passed, 1 warning** across + `test_resource_identity.py`, `test_apply_patch_transaction.py`, + `test_workspace_confine.py` and `test_tool_approvals.py`. +* `compileall -q` on the five changed/new production Python modules and the two + changed/new test modules passed. `git diff --check` passed. + +Counts overlap and must not be added. No full Python suite was executed. The +earlier focused runs exposed error-message expectation changes; the three +unscoped dispatcher denial assertions now check missing sealed roots. The +separate legacy resolver/sensitive-path tests remain intact. The new tests use +the raw dispatcher with explicit server authority, not a permissive fixture. + +Integrated command: + +```sh +/usr/bin/python3 -m pytest \ + tests/test_resource_identity.py tests/test_request_authority.py \ + tests/test_tool_approvals.py tests/test_tool_approval_single_action_scope.py \ + tests/test_tool_approval_task_scope.py tests/test_workspace_confine.py \ + tests/test_tool_path_confinement.py tests/test_path_confinement_boundary.py \ + tests/test_filesystem_tool_argument_validation.py tests/test_code_nav_tools.py \ + tests/test_apply_patch_transaction.py tests/test_execution_bridge.py \ + tests/test_production_external_bridge.py tests/test_turn_contract.py \ + tests/test_turn_contract_read_operations.py tests/test_turn_contract_integration.py \ + tests/test_agent_turn_contract_boundaries.py tests/test_explicit_personal_turn_contract.py \ + tests/test_nested_invocation_ownership.py tests/test_containment_contract.py \ + tests/test_containment_enforcement.py tests/test_containment_process_tree.py \ + tests/test_native_execution_containment.py tests/test_background_containment.py \ + tests/test_process_ownership.py tests/test_bg_jobs_store.py \ + tests/test_bg_job_tools.py tests/test_execution_filesystem_boundary.py \ + -q --disable-warnings --maxfail=8 +``` + +Final follow-up command: + +```sh +/usr/bin/python3 -m pytest tests/test_resource_identity.py \ + tests/test_apply_patch_transaction.py tests/test_workspace_confine.py \ + tests/test_tool_approvals.py -q --disable-warnings +``` + +Frozen containment, browser lifecycle producers, process ownership and +`agent_loop` were not edited. The resource types for the remaining domains are +inert contracts; their presence does not mean those execution adapters enforce +Wave 3 yet. Pathname races and inode reuse remain the limitations stated in D. diff --git a/src/agent_runtime/authority.py b/src/agent_runtime/authority.py index ef8552903..b86fd161f 100644 --- a/src/agent_runtime/authority.py +++ b/src/agent_runtime/authority.py @@ -11,6 +11,7 @@ from pathlib import Path import re from uuid import uuid4 +from src.agent_runtime.resources import FilesystemRoot, intersect_roots from src.tool_policy import ToolPolicy, build_effective_tool_policy from src.turn_contract import ( FAMILY_TOOLS, canonical_tool, requested_capabilities, @@ -111,6 +112,9 @@ class RequestAuthority: block_all: bool = False disable_mcp: bool = False inherited: bool = False + # None is only the trusted constructor's instruction to seal a workspace. + # Persisted/child authorities always carry an explicit tuple, including (). + resource_roots: tuple[FilesystemRoot, ...] | None = None def __post_init__(self): if (not isinstance(self.request_id, str) or not self.request_id @@ -122,10 +126,23 @@ class RequestAuthority: or any(not isinstance(n, str) or canonical_tool(n) != n for n in self.denied) or any(type(v) is not bool for v in (self.block_all, self.disable_mcp, self.inherited))): raise ValueError("Malformed request authority") + if self.resource_roots is None: + roots = () + if self.workspace: + try: + roots = (FilesystemRoot.seal(self.workspace, owner=self.owner),) + except (OSError, ValueError, RuntimeError): + pass # An unresolved workspace grants no filesystem root. + object.__setattr__(self, "resource_roots", roots) + if (not isinstance(self.resource_roots, tuple) + or any(not isinstance(r, FilesystemRoot) or (r.owner and r.owner != self.owner) + for r in self.resource_roots)): + raise ValueError("Malformed request resource roots") @classmethod def empty(cls, *, owner=None, session_id=None, workspace=None): - return cls(uuid4().hex, _owner(owner), str(session_id or ""), str(workspace or "")) + return cls(uuid4().hex, _owner(owner), str(session_id or ""), str(workspace or ""), + resource_roots=()) def bound_to(self, *, owner=None, session_id=None, workspace=None): return (self.owner == _owner(owner) and self.session_id == str(session_id or "") @@ -150,12 +167,15 @@ class RequestAuthority: if not isinstance(child, RequestAuthority): raise TypeError("Child authority must be server-owned RequestAuthority") grants = [] + roots = () if (self.owner, self.session_id, self.workspace) == (child.owner, child.session_id, child.workspace): theirs = {g.tool: g for g in child.grants} grants = [g.intersect(theirs[g.tool]) for g in self.grants if g.tool in theirs] + roots = intersect_roots(self.resource_roots, child.resource_roots) return replace(self, grants=tuple(grants), denied=self.denied | child.denied, block_all=self.block_all or child.block_all, - disable_mcp=self.disable_mcp or child.disable_mcp, inherited=True) + disable_mcp=self.disable_mcp or child.disable_mcp, inherited=True, + resource_roots=roots) def continuation(self, *, owner=None, session_id=None): """A server continuation may rebind a session, never change owner/grants.""" @@ -164,18 +184,19 @@ class RequestAuthority: return replace(self, session_id=str(session_id or ""), inherited=True) def to_dict(self): - return {"version": 1, "request_id": self.request_id, "owner": self.owner, + return {"version": 2, "request_id": self.request_id, "owner": self.owner, "session_id": self.session_id, "workspace": self.workspace, "grants": [{"tool": g.tool, "actions": None if g.actions is None else sorted(g.actions), "inputs": None if g.inputs is None else sorted(g.inputs)} for g in self.grants], "denied": sorted(self.denied), "block_all": self.block_all, - "disable_mcp": self.disable_mcp, "inherited": self.inherited} + "disable_mcp": self.disable_mcp, "inherited": self.inherited, + "resource_roots": [r.to_dict() for r in self.resource_roots]} @classmethod def from_dict(cls, value): if (not isinstance(value, dict) or type(value.get("version")) is not int - or value["version"] != 1): + or value["version"] not in {1, 2}): raise ValueError("Unsupported authority snapshot") def limits(value): if value is None: @@ -183,10 +204,14 @@ class RequestAuthority: if not isinstance(value, list) or any(not isinstance(v, str) for v in value): raise ValueError("Malformed authority limits") return frozenset(value) + roots = value["resource_roots"] if value["version"] == 2 else [] + if not isinstance(roots, list): + raise ValueError("Malformed request resource snapshot") return cls(value["request_id"], value["owner"], value["session_id"], value["workspace"], tuple(OperationGrant(g["tool"], limits(g["actions"]), limits(g["inputs"])) for g in value["grants"]), limits(value["denied"]), - value["block_all"], value["disable_mcp"], value["inherited"]) + value["block_all"], value["disable_mcp"], value["inherited"], + tuple(FilesystemRoot.from_dict(r) for r in roots)) _BROWSER_READ_ACTIONS = frozenset({"open", "navigate", "snapshot", "text", "read", "find", @@ -397,7 +422,8 @@ def seal_task_authority(prompt, task_type, action, *, owner=None, parent_authori parent = RequestAuthority.empty(owner=owner) if parent is not None: authority = parent.intersect(replace(authority, session_id=parent.session_id, - workspace=parent.workspace)) + workspace=parent.workspace, + resource_roots=parent.resource_roots)) return _json({"task_input": [prompt, task_type, action], "authority": authority.to_dict()}) diff --git a/src/agent_runtime/resource_binding.py b/src/agent_runtime/resource_binding.py new file mode 100644 index 000000000..73658be55 --- /dev/null +++ b/src/agent_runtime/resource_binding.py @@ -0,0 +1,203 @@ +"""Resolve native filesystem selectors once, after operation admission. + +Resolution produces inert bindings; the dispatcher still owns authority, +TurnContract, security and approval gates. No remote filesystem is resolved here. +""" +from __future__ import annotations + +from contextlib import contextmanager +from contextvars import ContextVar +from dataclasses import dataclass +import json +import os + +from src.agent_runtime.authority import ExactOperation +from src.agent_runtime.resources import FilesystemResource, FilesystemRoot +from src.path_confinement import canonical_root, confine + + +NATIVE_FILESYSTEM_TOOLS = frozenset({ + "read_file", "write_file", "edit_file", "apply_patch", "ls", "glob", "grep", +}) + + +@dataclass(frozen=True) +class ResourceBinding: + role: str + resource: FilesystemResource + + def __post_init__(self): + if self.role not in {"source", "target", "destination", "search_root"} or not isinstance(self.resource, FilesystemResource): + raise ValueError("Malformed operation resource binding") + + +@dataclass(frozen=True) +class BoundFilesystemOperation: + operation: ExactOperation + execution_input: str + bindings: tuple[ResourceBinding, ...] + # Empty only for inert proposal resolution without an originating request. + request_id: str = "" + + def __post_init__(self): + if (not isinstance(self.operation, ExactOperation) + or not isinstance(self.execution_input, str) + or not isinstance(self.bindings, tuple) or not self.bindings + or any(not isinstance(b, ResourceBinding) for b in self.bindings)): + raise ValueError("Malformed resource-bound operation") + if not isinstance(self.request_id, str) or any(c in self.request_id for c in ("\0", "\n", "\r")): + raise ValueError("Malformed resource operation request identity") + if (self.operation.action in {"move", "rename"} + and (len(self.bindings) != 2 or {b.role for b in self.bindings} != {"source", "destination"} + or len({b.resource.path for b in self.bindings}) != 2 + or next(b for b in self.bindings if b.role == "source").resource.identity is None)): + raise ValueError("Move/rename must bind distinct source and destination") + + def validate(self): + for binding in self.bindings: + binding.resource.validate() + + def to_dict(self): + return {"request_id": self.request_id, "tool": self.operation.transport_tool, "input": self.operation.input, + "execution_input": self.execution_input, + "bindings": [{"role": b.role, "resource": b.resource.to_dict()} for b in self.bindings]} + + def resolve_path(self, selector, *, search=False): + """Consume declared canonical targets; permit bounded search descendants.""" + if not isinstance(selector, str): + raise ValueError("Resource selector must be a string") + value = selector.strip() + for binding in self.bindings: + resource = binding.resource + if value == resource.path or (search and not value and binding.role == "search_root"): + resource.validate() + return resource.path + if not search: + for binding in self.bindings: + resource = binding.resource + if binding.role == "search_root" and resource.identity.kind == "directory": + resource.validate() + try: + path = confine(resource.path, value) + return FilesystemResource.resolve(resource.root, path).path + except (ValueError, OSError, RuntimeError): + continue + raise ValueError("Path is not declared by the resource-bound operation") + + +def _resolve(roots, selector, *, workspace, allow_missing): + if not isinstance(selector, str) or not selector.strip(): + raise ValueError("Resource path is required and must be a string") + value = selector.strip() + # The virtual alias belongs to the request workspace, even when a child + # narrows its root to a subdirectory of that workspace. + if value == "/workspace" or value.startswith("/workspace/"): + if not workspace: + raise ValueError("Workspace alias has no server-owned workspace") + base = canonical_root(workspace) + value = base if value == "/workspace" else os.path.join(base, value[len("/workspace/"):]) + elif not os.path.isabs(os.path.expanduser(value)): + if workspace: + value = os.path.join(canonical_root(workspace), value) + elif len(roots) == 1: + value = os.path.join(roots[0].path, value) + else: + raise ValueError("Relative resource path has no unambiguous server root") + for root in roots: + try: + return FilesystemResource.resolve(root, value, allow_missing=allow_missing) + except (ValueError, OSError, RuntimeError): + continue + boundary = "the workspace" if workspace else "the sealed roots" + raise ValueError(f"Resource path is outside {boundary}, sensitive, missing or changed") + + +def resolve_filesystem_operation(operation, *, roots, workspace="", request_id=""): + """Server adapter. This does not grant the operation or authorize its roots.""" + if not isinstance(operation, ExactOperation) or operation.tool not in NATIVE_FILESYSTEM_TOOLS: + raise ValueError("Operation has no native filesystem adapter") + if (not isinstance(roots, tuple) or not roots + or any(not isinstance(r, FilesystemRoot) for r in roots)): + raise ValueError("Native filesystem operation requires a sealed resource root") + content = operation.input + args = json.loads(content) if content.lstrip().startswith("{") else None + if args is not None and not isinstance(args, dict): + raise ValueError("Filesystem input must be an object") + bindings = [] + + def bind(selector, role, *, missing=False): + resource = _resolve(roots, selector, workspace=workspace, allow_missing=missing) + bindings.append(ResourceBinding(role, resource)) + return resource.path + + tool = operation.tool + if tool == "apply_patch": + from src.agent_tools.filesystem_tools import _parse_agent_patch + if args is None: + patch = content + else: + variants = [args[k] for k in ("patch_text", "patchText", "patch") if k in args] + if not variants or any(not isinstance(p, str) or p != variants[0] for p in variants): + raise ValueError("Patch requires one unambiguous patch_text") + patch = variants[0] + ops = _parse_agent_patch(patch) + paths = [bind(op["path"], "destination" if op["kind"] == "add" else "target", + missing=op["kind"] == "add") for op in ops] + objects = [b.resource.identity for b in bindings if b.resource.identity is not None] + if len(set(paths)) != len(paths) or len(set(objects)) != len(objects): + raise ValueError("Patch targets resolve to the same resource") + path_iter = iter(paths) + lines = patch.replace("\r\n", "\n").replace("\r", "\n").split("\n") + for i, line in enumerate(lines): + for marker in ("*** Add File: ", "*** Update File: ", "*** Delete File: "): + if line.startswith(marker): + lines[i] = marker + next(path_iter) + break + execution_input = json.dumps({"patch_text": "\n".join(lines)}, sort_keys=True) + else: + search = tool in {"ls", "glob", "grep"} + if args is None: + if tool == "write_file": + path, _, body = content.partition("\n") + args = {"path": path.strip(), "content": body} + elif tool == "edit_file": + raise ValueError("edit_file requires a JSON object") + elif tool in {"glob", "grep"}: + args = {"pattern": content.strip()} + else: + args = {"path": content.split("\n", 1)[0].strip()} + selector = args.get("path", "" if search else None) + if search and selector == "": + if workspace: + selector = canonical_root(workspace) + elif len(roots) == 1: + selector = roots[0].path + else: + raise ValueError("Search root is unresolved") + args["path"] = bind(selector, "search_root" if search else + "source" if tool == "read_file" else "destination" if tool == "write_file" else "target", + missing=tool == "write_file") + execution_input = json.dumps(args, sort_keys=True, allow_nan=False) + bound = BoundFilesystemOperation(operation, execution_input, tuple(bindings), request_id) + bound.validate() + return bound + + +_ACTIVE: ContextVar[BoundFilesystemOperation | None] = ContextVar("resource_operation", default=None) + + +def active_resource_operation(): + return _ACTIVE.get() + + +@contextmanager +def bind_resource_operation(operation): + if operation is not None and not isinstance(operation, BoundFilesystemOperation): + raise TypeError("Resource operation must be server-owned") + if operation is not None: + operation.validate() + token = _ACTIVE.set(operation) + try: + yield operation + finally: + _ACTIVE.reset(token) diff --git a/src/agent_runtime/resources.py b/src/agent_runtime/resources.py new file mode 100644 index 000000000..c8b76d474 --- /dev/null +++ b/src/agent_runtime/resources.py @@ -0,0 +1,302 @@ +"""Inert server-owned resource identities, independent of operation authority. + +Filesystem observations detect replacement; they are not held kernel handles or +content/effect evidence. Other producers must supply their own incarnations. +""" +from __future__ import annotations + +from dataclasses import asdict, dataclass +from enum import Enum +import os +from pathlib import Path +import stat + +from src.agent_runtime.path_policy import _is_sensitive_path +from src.path_confinement import canonical_root, confine + + +def _text(value, label, *, optional=False): + if (not isinstance(value, str) or (not value and not optional) + or any(c in value for c in ("\0", "\n", "\r"))): + raise ValueError(f"Invalid resource {label}") + + +def _absolute(value): + _text(value, "path") + if not os.path.isabs(value) or os.path.normpath(value) != value: + raise ValueError("Resource path must be canonical and absolute") + + +def _control_plane_path(path): + # Execution snapshots/receipts are server state, even if a workspace root + # contains the data directory. A writable user file cannot mint authority. + from src.constants import BG_JOBS_DIR, BG_JOBS_FILE, CONTAINMENT_STATE_FILE + if path in {canonical_root(BG_JOBS_FILE), canonical_root(CONTAINMENT_STATE_FILE)}: + return True + return (Path(path).is_relative_to(canonical_root(BG_JOBS_DIR)) + and path.endswith(".authority.json")) + + +class FilesystemScope(str, Enum): + WORKSPACE = "workspace" + SCRATCH = "scratch" + EXTERNAL = "external" + PRIVATE = "private" + + +class ResourceIdentityError(ValueError): + """An observed execution resource has changed or cannot be resolved.""" + + +@dataclass(frozen=True) +class FileObjectIdentity: + device: int + inode: int + kind: str + + def __post_init__(self): + if (type(self.device) is not int or self.device < 0 + or type(self.inode) is not int or self.inode <= 0 + or self.kind not in {"file", "directory"}): + raise ValueError("Malformed filesystem object identity") + + @classmethod + def observe(cls, path): + info = os.stat(path, follow_symlinks=False) + kind = ("file" if stat.S_ISREG(info.st_mode) else + "directory" if stat.S_ISDIR(info.st_mode) else None) + if kind is None: + raise ValueError("Filesystem resource must be a regular file or directory") + return cls(info.st_dev, info.st_ino, kind) + + +@dataclass(frozen=True) +class FilesystemRoot: + path: str + scope: FilesystemScope + identity: FileObjectIdentity + owner: str = "" + + def __post_init__(self): + _absolute(self.path) + _text(self.owner, "owner", optional=True) + if (not isinstance(self.scope, FilesystemScope) + or not isinstance(self.identity, FileObjectIdentity) + or self.identity.kind != "directory" + or os.path.dirname(self.path) == self.path + or _is_sensitive_path(self.path) + or (self.scope is FilesystemScope.PRIVATE and not self.owner)): + raise ValueError("Malformed filesystem root identity") + + @classmethod + def seal(cls, path, *, scope=FilesystemScope.WORKSPACE, owner=""): + root = canonical_root(path) + return cls(root, scope, FileObjectIdentity.observe(root), owner) + + def validate(self): + try: + if canonical_root(self.path) != self.path or FileObjectIdentity.observe(self.path) != self.identity: + raise ResourceIdentityError("Filesystem root identity changed") + except (OSError, RuntimeError) as error: + raise ResourceIdentityError("Filesystem root identity is unresolved") from error + + def to_dict(self): + return {**asdict(self), "scope": self.scope.value} + + @classmethod + def from_dict(cls, value): + if not isinstance(value, dict) or set(value) != {"path", "scope", "identity", "owner"}: + raise ValueError("Malformed filesystem root snapshot") + return cls(value["path"], FilesystemScope(value["scope"]), + FileObjectIdentity(**value["identity"]), value["owner"]) + + +@dataclass(frozen=True) +class PathObservation: + path: str + identity: FileObjectIdentity + + def __post_init__(self): + _absolute(self.path) + if not isinstance(self.identity, FileObjectIdentity) or self.identity.kind != "directory": + raise ValueError("Malformed filesystem ancestor identity") + + +@dataclass(frozen=True) +class FilesystemResource: + root: FilesystemRoot + path: str + identity: FileObjectIdentity | None + ancestors: tuple[PathObservation, ...] + + def __post_init__(self): + _absolute(self.path) + if (not isinstance(self.root, FilesystemRoot) + or not Path(self.path).is_relative_to(self.root.path) + or (self.identity is not None and not isinstance(self.identity, FileObjectIdentity)) + or not isinstance(self.ancestors, tuple) + or any(not isinstance(a, PathObservation) for a in self.ancestors) + or not self.ancestors + or self.ancestors[0] != PathObservation(self.root.path, self.root.identity)): + raise ValueError("Malformed filesystem resource identity") + parent = Path(self.root.path) + expected = [str(parent)] + for part in Path(self.path).relative_to(self.root.path).parts[:-1]: + parent /= part + expected.append(str(parent)) + if ([a.path for a in self.ancestors] != expected[:len(self.ancestors)] + or (self.identity is not None and len(self.ancestors) != len(expected))): + raise ValueError("Malformed filesystem ancestor chain") + + @classmethod + def resolve(cls, root, selector, *, allow_missing=False): + root.validate() + # Only this server-owned workspace root supplies the virtual alias. + if not isinstance(selector, str): + raise ValueError("Resource path must be a string") + value = selector.strip() + if root.scope is FilesystemScope.WORKSPACE: + if value == "/workspace": + value = root.path + elif value.startswith("/workspace/"): + value = os.path.join(root.path, value[len("/workspace/"):]) + path = confine(root.path, value) + if _is_sensitive_path(path) or _control_plane_path(path): + raise ValueError("Resource path is sensitive") + ancestors = [PathObservation(root.path, root.identity)] + relative = Path(path).relative_to(root.path) + parent = Path(root.path) + missing_parent = False + for part in relative.parts[:-1]: + parent /= part + try: + observed = FileObjectIdentity.observe(parent) + except FileNotFoundError: + missing_parent = True + break + ancestors.append(PathObservation(str(parent), observed)) + try: + identity = None if missing_parent else FileObjectIdentity.observe(path) + except FileNotFoundError: + identity = None + if identity is None and not allow_missing: + raise ValueError("Filesystem resource is unresolved or missing") + return cls(root, path, identity, tuple(ancestors)) + + def validate(self): + try: + if self.resolve(self.root, self.path, allow_missing=self.identity is None) != self: + raise ResourceIdentityError("Filesystem resource identity changed") + except (ValueError, OSError, RuntimeError) as error: + raise ResourceIdentityError("Filesystem resource identity changed or is unresolved") from error + + def to_dict(self): + return asdict(self) + + +def intersect_roots(parent, child): + """Keep the narrower root only when the observed parent's identity agrees.""" + result = [] + for left in parent: + for right in child: + if (left.scope, left.owner) != (right.scope, right.owner): + continue + if left == right: + result.append(left) + continue + try: + if Path(right.path).is_relative_to(left.path): + # A newly sealed child may not renew a replaced parent root. + left.validate() + right.validate() + result.append(right) + elif Path(left.path).is_relative_to(right.path): + left.validate() + right.validate() + result.append(left) + except (OSError, ValueError, RuntimeError): + continue + return tuple(dict.fromkeys(result)) + + +@dataclass(frozen=True) +class ProcessResource: + namespace: str + incarnation: str + owner: str + pid: int + start_token: str + job_id: str = "" + containment_id: str = "" + namespace_pid: int | None = None + namespace_start_token: str = "" + + def __post_init__(self): + for name in ("namespace", "incarnation", "owner", "start_token"): + _text(getattr(self, name), name) + for name in ("job_id", "containment_id", "namespace_start_token"): + _text(getattr(self, name), name, optional=True) + if (type(self.pid) is not int or self.pid <= 0 + or (self.namespace_pid is not None and + (type(self.namespace_pid) is not int or self.namespace_pid <= 0)) + or bool(self.namespace_pid) != bool(self.namespace_start_token)): + raise ValueError("Malformed process resource identity") + + +@dataclass(frozen=True) +class BrowserProducer: + namespace: str + owner: str + thread_id: str + session_id: str + incarnation: str + + def __post_init__(self): + for name in ("namespace", "owner", "thread_id", "session_id", "incarnation"): + _text(getattr(self, name), name) + + +@dataclass(frozen=True) +class BrowserPageResource: + producer: BrowserProducer + page_id: str + navigation_generation: int + observed_url: str + + def __post_init__(self): + if (not isinstance(self.producer, BrowserProducer) + or type(self.navigation_generation) is not int or self.navigation_generation < 0): + raise ValueError("Malformed browser page identity") + _text(self.page_id, "page") + _text(self.observed_url, "observed URL") + + +@dataclass(frozen=True) +class ExternalResource: + namespace: str + endpoint_id: str + server_id: str + tool_id: str + incarnation: str + external: bool = True + + def __post_init__(self): + for name in ("namespace", "endpoint_id", "server_id", "tool_id", "incarnation"): + _text(getattr(self, name), name) + if self.external is not True: + raise ValueError("External resource cannot attest local containment") + + +@dataclass(frozen=True) +class OwnedResource: + namespace: str + owner: str + thread_id: str + collection: str + record_id: str + revision: str = "" + + def __post_init__(self): + for name in ("namespace", "owner", "thread_id", "collection", "record_id"): + _text(getattr(self, name), name) + _text(self.revision, "revision", optional=True) diff --git a/src/tool_approvals.py b/src/tool_approvals.py index 7144fc3cf..d392bfce4 100644 --- a/src/tool_approvals.py +++ b/src/tool_approvals.py @@ -15,7 +15,7 @@ import secrets import threading import time from dataclasses import dataclass, field -from typing import Any +from typing import Any, TYPE_CHECKING from src.tool_approval_scopes import ( CHAT_SESSION_APPROVAL_DECISION, @@ -27,6 +27,9 @@ from src.tool_approval_scopes import ( from src.tool_capabilities import ToolCapabilities, capabilities_for_action from src.agent_runtime.authority import RequestAuthority +if TYPE_CHECKING: + from src.agent_runtime.resource_binding import BoundFilesystemOperation + DEFAULT_APPROVAL_TTL_SECONDS = 10 * 60 DEFAULT_MAX_PENDING_APPROVALS = 2048 @@ -119,6 +122,7 @@ def _binding_payload( effects: tuple[str, ...], result_integrity: str, request_authority: RequestAuthority | None = None, + resource_operation=None, ) -> dict[str, Any]: return { "owner": _normalized_owner(owner), @@ -140,6 +144,7 @@ def _binding_payload( "effects": list(effects), "result_integrity": str(result_integrity), "request_authority": request_authority.to_dict() if request_authority is not None else None, + "resource_operation": resource_operation.to_dict() if resource_operation is not None else None, } @@ -169,6 +174,8 @@ class PendingToolApproval: # is never displayed or treated as authorization for the sealed action. request_text: str = "" request_authority: RequestAuthority | None = None + # Server-resolved targets at proposal time; never read from the approval UI. + resource_operation: BoundFilesystemOperation | None = None def public_payload(self, *, reason: str | None = None) -> dict[str, Any]: return { @@ -278,6 +285,7 @@ class ExactToolApproval: effects=effects, result_integrity=result_integrity, request_authority=self.pending.request_authority, + resource_operation=self.pending.resource_operation, ) return _canonical_digest(expected) == self.pending.digest @@ -366,6 +374,22 @@ class ToolApprovalStore: if request_authority is not None and not isinstance(request_authority, RequestAuthority): raise TypeError("Approval authority must be server-owned RequestAuthority") now = time.time() + from src.agent_runtime.authority import ExactOperation + from src.agent_runtime.resource_binding import NATIVE_FILESYSTEM_TOOLS, resolve_filesystem_operation + from src.agent_runtime.resources import FilesystemRoot + resource_operation = None + if tool_name in NATIVE_FILESYSTEM_TOOLS: + try: + roots = request_authority.resource_roots if request_authority is not None else () + if not roots and workspace: + roots = (FilesystemRoot.seal(workspace, owner=_normalized_owner(owner)),) + resource_operation = resolve_filesystem_operation( + ExactOperation.normalize(tool_name, content), roots=roots, workspace=workspace or "", + request_id=request_authority.request_id if request_authority is not None else "") + except (ValueError, TypeError, OSError, RuntimeError): + # An unresolved proposal may be displayed, but it cannot execute + # after approval by reconstructing its targets at claim time. + pass effects = tuple(sorted(effect.value for effect in capabilities.effects)) result_integrity = capabilities.result_integrity.value payload = _binding_payload( @@ -384,6 +408,7 @@ class ToolApprovalStore: effects=effects, result_integrity=result_integrity, request_authority=request_authority, + resource_operation=resource_operation, ) pending = PendingToolApproval( approval_id=secrets.token_urlsafe(32), @@ -408,6 +433,7 @@ class ToolApprovalStore: continuation_query=payload["continuation_query"], request_text=str(request_text or ""), request_authority=request_authority, + resource_operation=resource_operation, ) with self._lock: self._purge_expired_locked(now) diff --git a/src/tool_execution.py b/src/tool_execution.py index 32a7bee71..7e77c4408 100644 --- a/src/tool_execution.py +++ b/src/tool_execution.py @@ -19,7 +19,7 @@ import secrets import sys import time from contextlib import contextmanager -from dataclasses import dataclass +from dataclasses import dataclass, replace from typing import Any, Awaitable, Callable, Dict, Iterator, Optional, Tuple @@ -43,6 +43,12 @@ from src.constants import ( ) from src.path_confinement import canonical_root, confine, is_inside from src.tool_utils import _truncate, get_mcp_manager +from src.tool_types import ToolBlock +from src.agent_runtime.resource_binding import ( + NATIVE_FILESYSTEM_TOOLS, active_resource_operation, bind_resource_operation, + resolve_filesystem_operation, +) +from src.agent_runtime.resources import ResourceIdentityError class _MissingToolSecurityContext: @@ -830,6 +836,9 @@ def _resolve_tool_path(raw_path: str) -> str: When a workspace is active for this turn, paths are confined to it instead of the default allowlist (see _resolve_tool_path_in_workspace). """ + resource_operation = active_resource_operation() + if resource_operation is not None: + return resource_operation.resolve_path(raw_path) ws = get_active_workspace() if ws: return _resolve_tool_path_in_workspace(ws, raw_path) @@ -972,6 +981,9 @@ def _resolve_search_root(raw_path: str) -> str: primary root (project data dir) and a supplied path is confined by the global allowlist + sensitive-file policy. """ + resource_operation = active_resource_operation() + if resource_operation is not None: + return resource_operation.resolve_path(raw_path, search=True) raw = (raw_path or "").strip() ws = get_active_workspace() if ws: @@ -1237,6 +1249,7 @@ async def _direct_fallback( "disabled_tools": frozenset(disabled_tools or ()), "tool_policy": tool_policy, "request_authority": active_request_authority(), + "resource_operation": active_resource_operation(), } from src.agent_tools import TOOL_HANDLERS @@ -1364,6 +1377,41 @@ async def execute_tool_block( "exit_code": 1, "failure_kind": "turn_contract_denied", } + # External executors require their own adapters. Local observations must + # never stand in for remote resource or containment identities. + execution_bridge = get_active_execution_bridge() + transport = operation.transport_tool + external_resource_call = ( + (execution_bridge is not None and transport in execution_bridge.supported_tools) + or (transport in _ROUTED_BRIDGE_TOOLS and _client_bridge(client_runtime_context) is not None) + or (transport == "apply_patch" and _tui_host_bridge_patch_url(client_runtime_context)) + ) + resource_operation = None + if operation.tool in NATIVE_FILESYSTEM_TOOLS and not external_resource_call: + try: + roots = authority.resource_roots + approved_resource = exact_approval.pending.resource_operation if exact_approval is not None else None + if exact_approval is not None and approved_resource is None: + raise ValueError("Approved filesystem action has no sealed resource identity") + if exact_admission and not roots and approved_resource is not None: + # This single exact action can use only the roots sealed with + # its proposal. The request/child authority is never widened. + roots = tuple(dict.fromkeys(b.resource.root for b in approved_resource.bindings)) + if any(r.owner and r.owner != authority.owner for r in roots): + raise ValueError("Filesystem resource owner differs from request authority") + resource_operation = resolve_filesystem_operation( + operation, roots=roots, workspace=authority.workspace, request_id=authority.request_id) + if approved_resource is not None: + if approved_resource.request_id and approved_resource.request_id != authority.request_id: + raise ValueError("Approved resource belongs to another request") + if replace(resource_operation, request_id=approved_resource.request_id) != approved_resource: + raise ValueError("Approved filesystem resource identity changed") + except (ValueError, TypeError, OSError, RuntimeError) as error: + return f"{transport}: BLOCKED", { + "error": str(error), "exit_code": 1, "blocked": True, + "failure_kind": "resource_identity_denied", + } + approval_claimed = False if exact_approval is not None: if ( @@ -1450,9 +1498,9 @@ async def execute_tool_block( token = _active_workspace.set(workspace or None) try: - with bind_request_authority(authority): + with bind_request_authority(authority), bind_resource_operation(resource_operation): output = await _execute_tool_block_impl( - block, + ToolBlock(transport, resource_operation.execution_input) if resource_operation is not None else block, session_id=session_id, disabled_tools=disabled_tools, owner=owner, @@ -1483,6 +1531,11 @@ async def execute_tool_block( getattr(block, "content", None), ) return output + except ResourceIdentityError as error: + return f"{transport}: BLOCKED", { + "error": str(error), "exit_code": 1, "blocked": True, + "failure_kind": "resource_identity_denied", + } finally: _active_workspace.reset(token) @@ -1614,6 +1667,7 @@ async def _execute_tool_block_impl( bridge_owns_tool = ( execution_bridge is not None and tool in execution_bridge.supported_tools + and active_resource_operation() is None ) # Public-owner restrictions protect tools executed by this deployment. @@ -1673,7 +1727,8 @@ async def _execute_tool_block_impl( }, ) - if tool in _ROUTED_BRIDGE_TOOLS and _client_bridge(client_runtime_context) is not None: + if (active_resource_operation() is None and tool in _ROUTED_BRIDGE_TOOLS + and _client_bridge(client_runtime_context) is not None): return await dispatched(_route_tool_via_bridge(tool, content, session_id, client_runtime_context)) # Background execution: a `bash` block whose first line is the `#!bg` @@ -1719,6 +1774,22 @@ async def _execute_tool_block_impl( from src.ai_interaction import do_generate_image desc = "generate_image" result = await dispatched(do_generate_image(content, session_id=session_id, owner=owner)) + elif (tool in NATIVE_FILESYSTEM_TOOLS + and (active_resource_operation() is not None or tool != "apply_patch" + or not _tui_host_bridge_patch_url(client_runtime_context))): + if active_resource_operation() is None: + return f"{tool}: BLOCKED", { + "error": "Native filesystem dispatch has no bound resource operation", + "exit_code": 1, "blocked": True, "failure_kind": "resource_identity_denied", + } + # Backend selection is pinned. MCP connection availability cannot + # redirect an admitted native resource to a different filesystem. + original = active_resource_operation().operation.input + desc = f"{tool}: {original.split(chr(10))[0][:80]}" + result = await dispatched(_direct_fallback(tool, content, owner=owner, session_id=session_id)) \ + or {"error": f"{tool}: execution failed", "exit_code": 1} + if tool == "edit_file": + desc = result.get("output") or result.get("error") or "edit_file" elif tool in _MCP_TOOL_MAP: first_line = content.split(chr(10))[0][:80] desc = f"{tool}: {first_line}" diff --git a/tests/test_resource_identity.py b/tests/test_resource_identity.py new file mode 100644 index 000000000..8c9daf46b --- /dev/null +++ b/tests/test_resource_identity.py @@ -0,0 +1,506 @@ +"""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") diff --git a/tests/test_tool_path_confinement.py b/tests/test_tool_path_confinement.py index 8c3e60414..34e21fd6a 100644 --- a/tests/test_tool_path_confinement.py +++ b/tests/test_tool_path_confinement.py @@ -252,7 +252,8 @@ async def test_read_file_dispatch_blocks_etc_shadow(monkeypatch): owner="admin-user", security_context=NO_TOOL_SECURITY_CONTEXT, ) - assert "outside the allowed roots" in (result.get("error") or "") + assert result.get("failure_kind") == "resource_identity_denied" + assert "sealed resource root" in (result.get("error") or "") assert result.get("exit_code") == 1 @@ -281,7 +282,8 @@ async def test_write_file_dispatch_blocks_authorized_keys(monkeypatch): owner="admin-user", security_context=NO_TOOL_SECURITY_CONTEXT, ) - assert "sensitive directory" in (result.get("error") or "") + assert result.get("failure_kind") == "resource_identity_denied" + assert "sealed resource root" in (result.get("error") or "") assert result.get("exit_code") == 1 @@ -344,7 +346,8 @@ async def test_write_file_dispatch_blocks_cron(monkeypatch): owner="admin-user", security_context=NO_TOOL_SECURITY_CONTEXT, ) - assert "outside the allowed roots" in (result.get("error") or "") + assert result.get("failure_kind") == "resource_identity_denied" + assert "sealed resource root" in (result.get("error") or "") assert result.get("exit_code") == 1 @pytest.mark.parametrize("filename", ["auth.json", "app.db", "settings.json"]) def test_application_secrets_are_sensitive_paths(filename):