fix: Block path traversal in Agent Builder file tools (v1)
Port of 1fa7cda9 from main.
`resolve_file_path` returned an absolute path unchanged and joined a
relative one onto an unresolved root without ever checking where the
result landed, so `..` segments and absolute paths both escaped the
project root. The Agent Builder's write, read and delete tools resolve
every path through it, which put arbitrary files on the machine within
reach of whatever the model was told to do.
It now resolves the root and the candidate and raises `ValueError` when
the candidate is not inside the root. Interior `..` that stays inside the
root still resolves, as does an absolute path pointing inside it.
Behaviour change: an Agent Builder session that wrote, read or deleted a
file outside its root directory now gets a tool failure instead. Every
caller of `resolve_file_path` and `resolve_file_paths` already wraps the
call in a `try`, so the new `ValueError` surfaces as
`{"success": False, ...}` rather than a traceback.
The commit's import-isolation test comes along unchanged; it guards a
separate property, that importing `adk_web_server` does not eagerly pull
in the Agent Builder agent stack.
This commit is contained in:
@@ -42,33 +42,43 @@ def resolve_file_path(
|
||||
working_directory: Working directory to use as base (defaults to cwd)
|
||||
|
||||
Returns:
|
||||
Resolved absolute Path object
|
||||
Resolved absolute Path object, guaranteed to be within the root directory.
|
||||
|
||||
Raises:
|
||||
ValueError: If ``file_path`` resolves outside the root directory, e.g. via
|
||||
``..`` traversal or an absolute path pointing outside the root.
|
||||
"""
|
||||
normalized_path = sanitize_generated_file_path(file_path)
|
||||
file_path_obj = Path(normalized_path)
|
||||
|
||||
# If already absolute, use as-is
|
||||
if file_path_obj.is_absolute():
|
||||
return file_path_obj
|
||||
|
||||
# Get root directory from session state, default to "./"
|
||||
root_directory = "./"
|
||||
if session_state and "root_directory" in session_state:
|
||||
root_directory = session_state["root_directory"]
|
||||
|
||||
# Use the same resolution logic as the main function
|
||||
root_path_obj = Path(root_directory)
|
||||
|
||||
if root_path_obj.is_absolute():
|
||||
resolved_root = root_path_obj
|
||||
elif working_directory:
|
||||
resolved_root = Path(working_directory) / root_directory
|
||||
else:
|
||||
if working_directory:
|
||||
resolved_root = Path(working_directory) / root_directory
|
||||
else:
|
||||
resolved_root = Path(os.getcwd()) / root_directory
|
||||
resolved_root = Path(os.getcwd()) / root_directory
|
||||
resolved_root = resolved_root.resolve()
|
||||
|
||||
# Resolve file path relative to root directory
|
||||
return resolved_root / file_path_obj
|
||||
if file_path_obj.is_absolute():
|
||||
candidate = file_path_obj.resolve()
|
||||
else:
|
||||
candidate = (resolved_root / file_path_obj).resolve()
|
||||
|
||||
# Keep the resolved path within the root to block path-traversal escapes.
|
||||
try:
|
||||
candidate.relative_to(resolved_root)
|
||||
except ValueError as exc:
|
||||
raise ValueError(
|
||||
f"File path {file_path!r} resolves outside the root directory"
|
||||
f" {resolved_root}."
|
||||
) from exc
|
||||
return candidate
|
||||
|
||||
|
||||
def resolve_file_paths(
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
# Copyright 2026 Google LLC
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
"""Import-isolation guard for adk_web_server.
|
||||
|
||||
Importing ``adk_web_server`` must not eagerly pull in the Agent Builder agent
|
||||
stack. Doing so reaches ``google.adk.agents`` at import time and breaks
|
||||
downstream consumers that import ``adk_web_server`` while ``google.adk.agents``
|
||||
is still initializing.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
|
||||
def test_importing_adk_web_server_does_not_import_agent_builder():
|
||||
# Run in a fresh interpreter so the check is not polluted by modules that
|
||||
# other tests already imported into sys.modules.
|
||||
code = (
|
||||
"import google.adk.cli.adk_web_server\n"
|
||||
"import sys\n"
|
||||
"forbidden = [\n"
|
||||
" 'google.adk.cli.built_in_agents.agent',\n"
|
||||
" 'google.adk.cli.built_in_agents.adk_agent_builder_assistant',\n"
|
||||
"]\n"
|
||||
"loaded = [name for name in forbidden if name in sys.modules]\n"
|
||||
"assert not loaded, loaded\n"
|
||||
)
|
||||
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-c", code],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
assert result.returncode == 0, result.stderr
|
||||
@@ -0,0 +1,140 @@
|
||||
# Copyright 2026 Google LLC
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
"""Path-traversal containment tests for Agent Builder file tools."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
from google.adk.cli.built_in_agents.tools.delete_files import delete_files
|
||||
from google.adk.cli.built_in_agents.tools.read_files import read_files
|
||||
from google.adk.cli.built_in_agents.tools.write_files import write_files
|
||||
from google.adk.cli.built_in_agents.utils.resolve_root_directory import resolve_file_path
|
||||
import pytest
|
||||
|
||||
|
||||
def _tool_context(root: Path) -> mock.MagicMock:
|
||||
tool_context = mock.MagicMock()
|
||||
tool_context._invocation_context.session.state = {"root_directory": str(root)}
|
||||
return tool_context
|
||||
|
||||
|
||||
def test_resolve_file_path_allows_path_within_root(tmp_path):
|
||||
resolved = resolve_file_path(
|
||||
"sub/dir/file.txt", {"root_directory": str(tmp_path)}
|
||||
)
|
||||
assert resolved == (tmp_path / "sub" / "dir" / "file.txt").resolve()
|
||||
|
||||
|
||||
def test_resolve_file_path_allows_dot(tmp_path):
|
||||
resolved = resolve_file_path(".", {"root_directory": str(tmp_path)})
|
||||
assert resolved == tmp_path.resolve()
|
||||
|
||||
|
||||
def test_resolve_file_path_allows_interior_dotdot_within_root(tmp_path):
|
||||
resolved = resolve_file_path(
|
||||
"sub/../file.txt", {"root_directory": str(tmp_path)}
|
||||
)
|
||||
assert resolved == (tmp_path / "file.txt").resolve()
|
||||
|
||||
|
||||
def test_resolve_file_path_allows_absolute_within_root(tmp_path):
|
||||
target = tmp_path / "nested" / "ok.txt"
|
||||
resolved = resolve_file_path(str(target), {"root_directory": str(tmp_path)})
|
||||
assert resolved == target.resolve()
|
||||
|
||||
|
||||
def test_resolve_file_path_rejects_relative_traversal(tmp_path):
|
||||
with pytest.raises(ValueError):
|
||||
resolve_file_path("../../escape.txt", {"root_directory": str(tmp_path)})
|
||||
|
||||
|
||||
def test_resolve_file_path_rejects_absolute_outside_root(tmp_path):
|
||||
with pytest.raises(ValueError):
|
||||
resolve_file_path("/etc/passwd", {"root_directory": str(tmp_path)})
|
||||
|
||||
|
||||
async def test_write_files_blocks_relative_traversal(
|
||||
tmp_path, tmp_path_factory
|
||||
):
|
||||
outside = tmp_path_factory.mktemp("outside")
|
||||
payload = os.path.relpath(outside / "pwned.txt", tmp_path)
|
||||
|
||||
result = await write_files(
|
||||
files={payload: "PWNED"}, tool_context=_tool_context(tmp_path)
|
||||
)
|
||||
|
||||
assert not result["success"]
|
||||
assert not (outside / "pwned.txt").exists()
|
||||
|
||||
|
||||
async def test_write_files_blocks_absolute_outside_root(
|
||||
tmp_path, tmp_path_factory
|
||||
):
|
||||
outside = tmp_path_factory.mktemp("outside")
|
||||
target = outside / "abs.txt"
|
||||
|
||||
result = await write_files(
|
||||
files={str(target): "PWNED"}, tool_context=_tool_context(tmp_path)
|
||||
)
|
||||
|
||||
assert not result["success"]
|
||||
assert not target.exists()
|
||||
|
||||
|
||||
async def test_write_files_allows_path_within_root(tmp_path):
|
||||
result = await write_files(
|
||||
files={"sub/ok.txt": "hello"}, tool_context=_tool_context(tmp_path)
|
||||
)
|
||||
|
||||
assert result["success"]
|
||||
assert (tmp_path / "sub" / "ok.txt").read_text() == "hello"
|
||||
|
||||
|
||||
async def test_read_files_blocks_relative_traversal(tmp_path, tmp_path_factory):
|
||||
outside = tmp_path_factory.mktemp("outside")
|
||||
secret = outside / "secret.txt"
|
||||
secret.write_text("TOKEN=abc")
|
||||
payload = os.path.relpath(secret, tmp_path)
|
||||
|
||||
result = await read_files(
|
||||
file_paths=[payload], tool_context=_tool_context(tmp_path)
|
||||
)
|
||||
|
||||
assert not result["success"]
|
||||
assert all(
|
||||
"TOKEN=abc" not in info.get("content", "")
|
||||
for info in result["files"].values()
|
||||
)
|
||||
|
||||
|
||||
async def test_delete_files_blocks_relative_traversal(
|
||||
tmp_path, tmp_path_factory
|
||||
):
|
||||
outside = tmp_path_factory.mktemp("outside")
|
||||
victim = outside / "victim.txt"
|
||||
victim.write_text("bye")
|
||||
payload = os.path.relpath(victim, tmp_path)
|
||||
|
||||
result = await delete_files(
|
||||
file_paths=[payload],
|
||||
tool_context=_tool_context(tmp_path),
|
||||
confirm_deletion=True,
|
||||
)
|
||||
|
||||
assert not result["success"]
|
||||
assert victim.exists()
|
||||
Reference in New Issue
Block a user