From 9f18d1d7608ff015664238d2247cf956dfed3d59 Mon Sep 17 00:00:00 2001 From: XBetter <32125213+1998x-stack@users.noreply.github.com> Date: Wed, 28 Oct 2026 14:38:40 +0800 Subject: [PATCH 1/5] fix(tools): canonicalize all declared and conventional filesystem path arguments --- tiny_harness/_tools.py | 96 +++++++++++++++++++++++++++--------------- 1 file changed, 61 insertions(+), 35 deletions(-) diff --git a/tiny_harness/_tools.py b/tiny_harness/_tools.py index f42e619..ee8ea81 100644 --- a/tiny_harness/_tools.py +++ b/tiny_harness/_tools.py @@ -21,10 +21,19 @@ class ToolDef: description: str parameters: dict risk_level: str = "mutation" + # Filesystem locations in the schema, independent of a tool's Python name. + # All fields listed here are canonicalized before the handler is invoked. + path_args: tuple[str, ...] = () def __post_init__(self) -> None: if self.risk_level not in ("safe", "read_only", "mutation", "destructive", "dangerous"): raise ValueError(f"Unknown tool risk_level: {self.risk_level!r}") + if not isinstance(self.path_args, tuple) or len(set(self.path_args)) != len(self.path_args): + raise ValueError("path_args must be a tuple of distinct schema property names") + props = self.parameters.get("properties", {}) + for field in self.path_args: + if not isinstance(field, str) or not field or props.get(field, {}).get("type") != "string": + raise ValueError(f"path_args entry {field!r} must be a declared string property") @dataclass @@ -63,8 +72,11 @@ def register(self, tool: Tool) -> None: def register_from_def(self, def_: ToolDef, handler: Callable) -> None: self.register(Tool(definition=def_, handler=handler)) - def register_tool(self, name: str, description: str, parameters: dict, handler: Callable, risk_level: str = "mutation") -> None: - self.register_from_def(ToolDef(name=name, description=description, parameters=parameters, risk_level=risk_level), handler) + def register_tool(self, name: str, description: str, parameters: dict, handler: Callable, risk_level: str = "mutation", path_args: tuple[str, ...] = ()) -> None: + self.register_from_def(ToolDef( + name=name, description=description, parameters=parameters, + risk_level=risk_level, path_args=path_args, + ), handler) def get(self, name: str) -> Tool | None: return self._tools.get(name) @@ -76,13 +88,22 @@ def names(self) -> list[str]: return list(self._tools.keys()) -# Built-in filesystem tools have explicit path contracts. A plugin that uses -# other argument names must implement its own filesystem confinement. -_FILE_TOOLS = frozenset({ - "read_file", "write_file", "list_directory", "find_files", - "delete_file", "create_directory", "move_file", "search_content", -}) - +# Compatibility for third-party tools registered before path_args existed. +# New tools should declare ToolDef(path_args=("filename", ...)). +_LEGACY_FILE_PATH_KEYS = { + "read_file": ("path",), + "write_file": ("path",), + "list_directory": ("path",), + "find_files": ("path",), + "delete_file": ("path",), + "create_directory": ("path",), + "move_file": ("source", "destination"), + "search_content": ("path",), +} +_DEFAULT_WORKSPACE_PATH_TOOLS = frozenset(( + "list_directory", "find_files", "search_content", "run_command", +)) +_CONVENTIONAL_PATH_KEYS = ("path", "source", "destination", "cwd") class ToolExecutor: def __init__(self, registry: ToolRegistry, guard: "FilesystemGuard", timeout_ms: int = 30_000, max_output_chars: int = 50_000, approval_gate: "ApprovalGate | None" = None, event_bus: "EventBus | None" = None, allow_unsafe_shell: bool = False): @@ -97,33 +118,38 @@ def __init__(self, registry: ToolRegistry, guard: "FilesystemGuard", timeout_ms: def get_definitions(self) -> list[dict]: return self._registry.get_definitions() - def _guard_args(self, name: str, args: dict, risk_level: str) -> dict: + def _guard_args(self, tool: Tool, args: dict) -> dict: if not self._guard: - return args + return dict(args) + name = tool.definition.name checked = dict(args) + + # Legacy conventional fields are checked for compatibility. A ToolDef + # may additionally declare arbitrary path fields such as "filename". + # Resolve EVERY field, not just the first nonempty one. + fields = set(tool.definition.path_args) + fields.update(_LEGACY_FILE_PATH_KEYS.get(name, ())) + fields.update(k for k in _CONVENTIONAL_PATH_KEYS if k in checked) if name == "run_command": - # The executor owns the workspace contract. Never let subprocess - # inherit an unrelated process working directory. - checked["cwd"] = self._guard.guard(checked.get("cwd") or ".", risk_level) - elif name in _FILE_TOOLS: - keys = ("source", "destination") if name == "move_file" else ("path",) - for key in keys: - if key in checked: - checked[key] = self._guard.guard(checked[key], risk_level) - if name in ("list_directory", "find_files", "search_content") and "path" not in checked: - checked["path"] = self._guard.guard(".", risk_level) - if name == "find_files": - # Glob patterns require separate validation from plain paths. - import os - pattern = checked.get("pattern", "") - if os.path.isabs(pattern) or ".." in pattern.replace("\\", "/").split("/"): - raise ValueError("Glob pattern must remain within the workspace") - else: - # Legacy protection for third-party tools with a conventional - # path argument; this does NOT sandbox arbitrary shell commands. - path = checked.get("path") or checked.get("source") or checked.get("destination") or checked.get("cwd") - if path: - self._guard.guard(path, risk_level) + fields.add("cwd") + + for field in fields: + if field in checked: + value = checked[field] + if not isinstance(value, str): + raise ValueError(f"Filesystem path {field!r} must be a string") + checked[field] = self._guard.guard(value, tool.definition.risk_level) + elif field == "cwd" and name == "run_command": + checked["cwd"] = self._guard.guard(".", tool.definition.risk_level) + elif field == "path" and name in _DEFAULT_WORKSPACE_PATH_TOOLS: + checked["path"] = self._guard.guard(".", tool.definition.risk_level) + + if name == "find_files": + # Glob patterns need validation beyond ordinary path arguments. + import os + pattern = checked.get("pattern", "") + if os.path.isabs(pattern) or ".." in pattern.replace("\\", "/").split("/"): + raise ValueError("Glob pattern must remain within the workspace") return checked async def execute(self, name: str, args: dict, call_id: str) -> ToolResult: @@ -147,7 +173,7 @@ async def execute(self, name: str, args: dict, call_id: str) -> ToolResult: return ToolResult.error(call_id, f"Invalid arguments for '{name}':\n" + "\n".join(f" - {e}" for e in errors)) try: - self._guard_args(name, args, tool.definition.risk_level) + self._guard_args(tool, args) except Exception as e: return ToolResult.error(call_id, str(e)) @@ -168,7 +194,7 @@ async def execute(self, name: str, args: dict, call_id: str) -> ToolResult: if errors: return ToolResult.error(call_id, f"Invalid approved arguments for '{name}':\n" + "\n".join(f" - {e}" for e in errors)) try: - args = self._guard_args(name, args, tool.definition.risk_level) + args = self._guard_args(tool, args) except Exception as e: return ToolResult.error(call_id, str(e)) From 474ef7bd408f10a3d066efb4521f55877b1bcf12 Mon Sep 17 00:00:00 2001 From: XBetter <32125213+1998x-stack@users.noreply.github.com> Date: Wed, 28 Oct 2026 14:38:56 +0800 Subject: [PATCH 2/5] test(tools): enforce declared filesystem paths and recheck approved modifications --- tests/test_tool_path_contracts.py | 124 ++++++++++++++++++++++++++++++ 1 file changed, 124 insertions(+) create mode 100644 tests/test_tool_path_contracts.py diff --git a/tests/test_tool_path_contracts.py b/tests/test_tool_path_contracts.py new file mode 100644 index 0000000..689623b --- /dev/null +++ b/tests/test_tool_path_contracts.py @@ -0,0 +1,124 @@ +"""Tool path declarations are independent of tool names and validated twice.""" +import os + +import pytest + +from tiny_harness._guard import FilesystemGuard +from tiny_harness._hitl import ApprovalDecision, ApprovalGate +from tiny_harness._tools import ToolDef, ToolExecutor, ToolRegistry + + +def _registry(path_args=("filename",), props=None): + registry = ToolRegistry() + observed = [] + schema = {"type": "object", "properties": props or { + "filename": {"type": "string"}, + "path": {"type": "string"}, + "source": {"type": "string"}, + "destination": {"type": "string"}, + }} + registry.register_from_def( + ToolDef("custom_import", "fixture", schema, + risk_level="read_only", path_args=path_args), + lambda args: observed.append(dict(args)) or str(args), + ) + return registry, observed + + +def test_path_args_require_explicit_string_fields(): + for path_args in [("missing",), ("filename", "filename"), ("filename", "integer")]: + with pytest.raises(ValueError, match="path_args"): + ToolDef( + "bad", "fixture", + {"type": "object", "properties": { + "filename": {"type": "string"}, + "integer": {"type": "integer"}, + }}, + path_args=path_args, + ) + + +@pytest.mark.asyncio +async def test_custom_path_argument_uses_workspace_not_process_cwd(tmp_path, monkeypatch): + workspace = tmp_path / "workspace" + other = tmp_path / "other" + workspace.mkdir() + other.mkdir() + monkeypatch.chdir(other) + registry, observed = _registry() + result = await ToolExecutor(registry, FilesystemGuard(str(workspace))).execute( + "custom_import", {"filename": "data.txt"}, "call-1", + ) + assert result.success + assert observed == [{"filename": str(workspace / "data.txt")}] + assert not (other / "data.txt").exists() + + +@pytest.mark.asyncio +async def test_unknown_tool_checks_every_conventional_field(tmp_path): + workspace = tmp_path / "workspace" + workspace.mkdir() + external = tmp_path / "external.txt" + external.write_text("private") + registry, observed = _registry(path_args=()) + executor = ToolExecutor(registry, FilesystemGuard(str(workspace))) + denied = await executor.execute( + "custom_import", + {"path": "safe.txt", "source": str(external), "destination": "ok.txt"}, + "call-2", + ) + assert not denied.success + assert "outside" in denied.content + assert observed == [] + + +@pytest.mark.asyncio +async def test_all_declared_paths_block_symlink_escape(tmp_path): + workspace = tmp_path / "workspace" + workspace.mkdir() + secret = tmp_path / "outside.txt" + secret.write_text("private") + (workspace / "alias").symlink_to(secret) + registry, observed = _registry(path_args=("filename",)) + executor = ToolExecutor(registry, FilesystemGuard(str(workspace))) + denied = await executor.execute("custom_import", {"filename": "alias"}, "call-3") + assert not denied.success + assert observed == [] + + +@pytest.mark.asyncio +async def test_approval_modified_arguments_are_validated_and_recannonicalized(tmp_path): + workspace = tmp_path / "workspace" + workspace.mkdir() + outside = tmp_path / "outside.txt" + outside.write_text("secret") + registry, observed = _registry(path_args=("filename",)) + + async def approve(request): + return ApprovalDecision(approved=True, modified_args={"filename": str(outside)}) + + registry.get("custom_import").definition.risk_level = "mutation" + executor = ToolExecutor( + registry, FilesystemGuard(str(workspace)), + approval_gate=ApprovalGate(handler=approve), + ) + result = await executor.execute( + "custom_import", {"filename": "safe.txt"}, "call-4", + ) + assert not result.success and "outside" in result.content + assert observed == [] + + +@pytest.mark.asyncio +async def test_registry_helper_accepts_custom_path_args(tmp_path): + registry = ToolRegistry() + registry.register_tool( + "artifact_read", "reads a fixture", + {"type": "object", "properties": {"asset_path": {"type": "string"}}}, + lambda args: args["asset_path"], + risk_level="read_only", path_args=("asset_path",), + ) + result = await ToolExecutor(registry, FilesystemGuard(str(tmp_path))).execute( + "artifact_read", {"asset_path": "nested/file.txt"}, "call-5", + ) + assert result.success and result.content == str(tmp_path / "nested/file.txt") From c9fdbb46332fe4308baed60db438822899037e22 Mon Sep 17 00:00:00 2001 From: XBetter <32125213+1998x-stack@users.noreply.github.com> Date: Wed, 28 Oct 2026 14:39:09 +0800 Subject: [PATCH 3/5] refactor(skills): declare built-in filesystem path parameters explicitly --- tiny_harness/skills/files.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/tiny_harness/skills/files.py b/tiny_harness/skills/files.py index ff2119f..5d33bd6 100644 --- a/tiny_harness/skills/files.py +++ b/tiny_harness/skills/files.py @@ -77,31 +77,31 @@ def register(agent) -> None: agent.tools.register_from_def( - ToolDef(name="read_file", description="Read file contents from the filesystem. Returns text content. For large files, use offset and limit to read specific sections. Binary files return a size/type description instead of raw content.", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Path to the file to read."}, "offset": {"type": "integer", "description": "Line number to start reading from (1-indexed). Default: 1."}, "limit": {"type": "integer", "description": "Maximum number of lines to read. Omit to read the entire file."}}, "required": ["path"]}, risk_level="read_only"), + ToolDef(name="read_file", description="Read file contents from the filesystem. Returns text content. For large files, use offset and limit to read specific sections. Binary files return a size/type description instead of raw content.", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Path to the file to read."}, "offset": {"type": "integer", "description": "Line number to start reading from (1-indexed). Default: 1."}, "limit": {"type": "integer", "description": "Maximum number of lines to read. Omit to read the entire file."}}, "required": ["path"]}, risk_level="read_only", path_args=("path",)), partial(read_file, structured=True), ) agent.tools.register_from_def( - ToolDef(name="write_file", description="Create a new file or overwrite an existing one. Auto-creates parent directories. WARNING: Overwrites existing files without confirmation. Verify with read_file if unsure about current contents.", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Path to the file to create or overwrite."}, "content": {"type": "string", "description": "The text content to write to the file."}}, "required": ["path", "content"]}, risk_level="mutation"), + ToolDef(name="write_file", description="Create a new file or overwrite an existing one. Auto-creates parent directories. WARNING: Overwrites existing files without confirmation. Verify with read_file if unsure about current contents.", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Path to the file to create or overwrite."}, "content": {"type": "string", "description": "The text content to write to the file."}}, "required": ["path", "content"]}, risk_level="mutation", path_args=("path",)), partial(write_file, structured=True), ) agent.tools.register_from_def( - ToolDef(name="list_directory", description="List files and subdirectories in a directory. Shows names, types (file/dir), and sizes. Use to understand project structure before making changes.", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Directory to list. Default: current directory."}, "pattern": {"type": "string", "description": "Glob pattern to filter (e.g., '*.py')."}, "recursive": {"type": "boolean", "description": "List recursively if true. Default: false."}}}, risk_level="read_only"), + ToolDef(name="list_directory", description="List files and subdirectories in a directory. Shows names, types (file/dir), and sizes. Use to understand project structure before making changes.", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Directory to list. Default: current directory."}, "pattern": {"type": "string", "description": "Glob pattern to filter (e.g., '*.py')."}, "recursive": {"type": "boolean", "description": "List recursively if true. Default: false."}}}, risk_level="read_only", path_args=("path",)), partial(list_directory, structured=True), ) agent.tools.register_from_def( - ToolDef(name="find_files", description="Find files matching a glob pattern. Supports recursive wildcards like '**/*.py'. Use to discover file locations before reading them.", parameters={"type": "object", "properties": {"pattern": {"type": "string", "description": "Glob pattern (e.g., '**/*.py', 'src/**/*test*')."}, "path": {"type": "string", "description": "Directory to search in. Default: current directory."}, "max_results": {"type": "integer", "description": "Maximum number of results. Default: 200."}}, "required": ["pattern"]}, risk_level="read_only"), + ToolDef(name="find_files", description="Find files matching a glob pattern. Supports recursive wildcards like '**/*.py'. Use to discover file locations before reading them.", parameters={"type": "object", "properties": {"pattern": {"type": "string", "description": "Glob pattern (e.g., '**/*.py', 'src/**/*test*')."}, "path": {"type": "string", "description": "Directory to search in. Default: current directory."}, "max_results": {"type": "integer", "description": "Maximum number of results. Default: 200."}}, "required": ["pattern"]}, risk_level="read_only", path_args=("path",)), partial(find_files, structured=True), ) agent.tools.register_from_def( - ToolDef(name="delete_file", description="Permanently delete a file. IRREVERSIBLE — cannot be undone. Prefer move_file to archive. Always confirm what you're deleting by reading or listing first.", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Path to the file to delete permanently."}}, "required": ["path"]}, risk_level="destructive"), + ToolDef(name="delete_file", description="Permanently delete a file. IRREVERSIBLE — cannot be undone. Prefer move_file to archive. Always confirm what you're deleting by reading or listing first.", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Path to the file to delete permanently."}}, "required": ["path"]}, risk_level="destructive", path_args=("path",)), partial(delete_file, structured=True), ) agent.tools.register_from_def( - ToolDef(name="create_directory", description="Create a directory and any needed parent directories. Safe to call on existing directories (no error).", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Path to the directory to create."}}, "required": ["path"]}, risk_level="mutation"), + ToolDef(name="create_directory", description="Create a directory and any needed parent directories. Safe to call on existing directories (no error).", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Path to the directory to create."}}, "required": ["path"]}, risk_level="mutation", path_args=("path",)), partial(create_directory, structured=True), ) agent.tools.register_from_def( - ToolDef(name="move_file", description="Move or rename a file or directory. Overwrites the destination if it already exists. Safer alternative to delete_file for removing files.", parameters={"type": "object", "properties": {"source": {"type": "string", "description": "Current path of the file or directory."}, "destination": {"type": "string", "description": "New path for the file or directory."}}, "required": ["source", "destination"]}, risk_level="mutation"), + ToolDef(name="move_file", description="Move or rename a file or directory. Overwrites the destination if it already exists. Safer alternative to delete_file for removing files.", parameters={"type": "object", "properties": {"source": {"type": "string", "description": "Current path of the file or directory."}, "destination": {"type": "string", "description": "New path for the file or directory."}}, "required": ["source", "destination"]}, risk_level="mutation", path_args=("source", "destination")), partial(move_file, structured=True), ) agent._prompt.append(FILES_PROMPT_SECTION) From 24dfa2da8fa2f8684ad77ab20805a833bc9041a8 Mon Sep 17 00:00:00 2001 From: XBetter <32125213+1998x-stack@users.noreply.github.com> Date: Wed, 28 Oct 2026 14:39:11 +0800 Subject: [PATCH 4/5] refactor(skills): declare built-in filesystem path parameters explicitly --- tiny_harness/skills/search.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tiny_harness/skills/search.py b/tiny_harness/skills/search.py index 666ce15..154955b 100644 --- a/tiny_harness/skills/search.py +++ b/tiny_harness/skills/search.py @@ -88,6 +88,7 @@ def register(agent) -> None: "required": ["pattern"], }, risk_level="read_only", + path_args=("path",), ), partial(search_content, structured=True), ) From 81bc7665904d817ce750910a3c0cea916128901e Mon Sep 17 00:00:00 2001 From: XBetter <32125213+1998x-stack@users.noreply.github.com> Date: Wed, 28 Oct 2026 14:39:14 +0800 Subject: [PATCH 5/5] refactor(skills): declare built-in filesystem path parameters explicitly --- tiny_harness/skills/shell.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tiny_harness/skills/shell.py b/tiny_harness/skills/shell.py index ae1ca91..a562a73 100644 --- a/tiny_harness/skills/shell.py +++ b/tiny_harness/skills/shell.py @@ -90,6 +90,7 @@ def register(agent) -> None: "required": ["command"], }, risk_level="dangerous", + path_args=("cwd",), ), partial(run_command_async, structured=True), )