Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 15 additions & 6 deletions src/openharness/engine/query.py
Original file line number Diff line number Diff line change
Expand Up @@ -1023,16 +1023,23 @@ def _resolve_permission_file_path(
raw_input: dict[str, object],
parsed_input: object,
) -> str | None:
for key in ("file_path", "path", "root"):
value = raw_input.get(key)
# Permission must judge exactly what execution will touch. Execution
# receives only the validated model (tool.execute(parsed_input, ...)), so
# parsed fields are authoritative. Checking raw keys first lets a decoy
# field Pydantic drops during validation (e.g. `file_path` on read_file,
# whose schema field is `path`) shadow the real path and bypass deny
# rules (issue #348). Raw keys remain a fallback for dynamic-schema tools
# (MCP proxies) whose parsed model exposes no path attributes.
for attr in ("file_path", "path", "root"):
value = getattr(parsed_input, attr, None)
if isinstance(value, str) and value.strip():
path = Path(value).expanduser()
if not path.is_absolute():
path = cwd / path
return str(path.resolve())

for attr in ("file_path", "path", "root"):
value = getattr(parsed_input, attr, None)
for key in ("file_path", "path", "root"):
value = raw_input.get(key)
if isinstance(value, str) and value.strip():
path = Path(value).expanduser()
if not path.is_absolute():
Expand All @@ -1046,11 +1053,13 @@ def _extract_permission_command(
raw_input: dict[str, object],
parsed_input: object,
) -> str | None:
value = raw_input.get("command")
# Parsed-first for the same reason as _resolve_permission_file_path: a
# raw key the schema dropped must never shadow what execution will run.
value = getattr(parsed_input, "command", None)
if isinstance(value, str) and value.strip():
return value

value = getattr(parsed_input, "command", None)
value = raw_input.get("command")
if isinstance(value, str) and value.strip():
return value

Expand Down
74 changes: 74 additions & 0 deletions tests/test_engine/test_query_engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -989,6 +989,80 @@ async def test_execute_tool_call_applies_path_rules_to_directory_roots(tmp_path:
assert str(blocked_dir) in result.content


@pytest.mark.asyncio
async def test_execute_tool_call_decoy_file_path_cannot_shadow_deny_rule(tmp_path: Path):
"""Issue #348: a raw `file_path` key that Pydantic drops during validation
must not shadow the schema's real `path` field during permission checks."""
blocked_dir = tmp_path / "work" / "blocked"
blocked_dir.mkdir(parents=True)
(blocked_dir / "secret.txt").write_text("top-secret\n", encoding="utf-8")
readme = tmp_path / "README.md"
readme.write_text("hello\n", encoding="utf-8")

registry = ToolRegistry()
registry.register(create_default_tool_registry().get("read_file"))
settings = PermissionSettings(
mode=PermissionMode.FULL_AUTO,
path_rules=[{"pattern": str(blocked_dir) + "/*", "allow": False}],
)

# The PoC pair: decoy allowed path + real denied path.
result = await _execute_tool_call(
_tool_context(tmp_path, registry, settings),
"read_file",
"toolu_read_poc",
{"file_path": str(readme), "path": "work/blocked/secret.txt", "offset": 0, "limit": 10},
)
assert result.is_error is True
assert "deny" in result.content.lower() or str(blocked_dir) in result.content

# Control: the same call without the decoy was already blocked before.
control = await _execute_tool_call(
_tool_context(tmp_path, registry, settings),
"read_file",
"toolu_read_control",
{"path": "work/blocked/secret.txt", "offset": 0, "limit": 10},
)
assert control.is_error is True

# The decoy alone must not grant access either way round: real field
# pointing at an allowed file stays allowed (no over-blocking).
ok = await _execute_tool_call(
_tool_context(tmp_path, registry, settings),
"read_file",
"toolu_read_ok",
{"path": "README.md", "file_path": "work/blocked/secret.txt"},
)
assert ok.is_error is False
assert "hello" in (ok.content or "")


@pytest.mark.asyncio
async def test_execute_tool_call_decoy_file_path_cannot_shadow_write_deny_rule(tmp_path: Path):
blocked_dir = tmp_path / "work" / "blocked"
blocked_dir.mkdir(parents=True)

registry = ToolRegistry()
registry.register(create_default_tool_registry().get("write_file"))

result = await _execute_tool_call(
_tool_context(
tmp_path,
registry,
PermissionSettings(
mode=PermissionMode.FULL_AUTO,
path_rules=[{"pattern": str(blocked_dir) + "/*", "allow": False}],
),
),
"write_file",
"toolu_write_poc",
{"file_path": "README.md", "path": "work/blocked/output.txt", "content": "poc"},
)

assert result.is_error is True
assert not (blocked_dir / "output.txt").exists()


@pytest.mark.asyncio
async def test_execute_tool_call_returns_actionable_reason_when_user_denies_confirmation(tmp_path: Path):
async def _deny(_tool_name: str, _reason: str) -> bool:
Expand Down