Skip to content
Merged
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
29 changes: 24 additions & 5 deletions src/google/adk/environment/_local_environment.py
Original file line number Diff line number Diff line change
Expand Up @@ -139,11 +139,30 @@ async def write_file(self, path: str | Path, content: str | bytes) -> None:
return await asyncio.to_thread(self._sync_write, resolved, content)

def _resolve_path(self, path: str | Path) -> str:
"""Resolve a relative path against the working directory."""
path = str(path)
if os.path.isabs(path):
return path
return os.path.join(self._working_dir, path)
"""Resolve a file path inside the working directory.

Relative paths are resolved against ``working_dir``. Absolute paths
are accepted only when they stay inside ``working_dir``.

Args:
path: Absolute or working-dir-relative path.

Returns:
The resolved absolute path.

Raises:
RuntimeError: If ``working_dir`` is not set.
ValueError: If the resolved path escapes ``working_dir``.
"""
candidate = Path(path)
working_dir = self.working_dir.resolve()
if not candidate.is_absolute():
candidate = working_dir / candidate

resolved = candidate.resolve()
if not resolved.is_relative_to(working_dir):
raise ValueError(f'Path escapes working directory: {path}')
return str(resolved)

@staticmethod
def _sync_read(path: str) -> bytes:
Expand Down
25 changes: 9 additions & 16 deletions src/google/adk/tools/environment/_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,11 @@ def _truncate(text: str, limit: int = MAX_OUTPUT_CHARS) -> str:
return text[:limit] + f'\n... (truncated, {len(text)} total chars)'


def _is_valid_line_number(value: Any) -> bool:
"""Returns True when *value* is a non-bool integer."""
return isinstance(value, int) and not isinstance(value, bool)


_EXECUTE_TOOL_DESCRIPTION = """
Run a shell command in the environment. For running programs, tests, and build
commands ONLY. WARNING: Do NOT use for file reading -- use the ReadFile tool
Expand Down Expand Up @@ -201,23 +206,11 @@ async def run_async(
return {'status': 'error', 'error': '`path` is required.'}
start_line = args.get('start_line')
end_line = args.get('end_line')

# Use `sed` to read the file if start_line or end_line are specified.
if (start_line and start_line > 1) or end_line:
start = start_line or 1
if end_line:
sed_range = f'{start},{end_line}'
else:
sed_range = f'{start},$'
cmd = f"cat -n '{path}' | sed -n '{sed_range}p'"
res = await self._environment.execute(cmd)
if res.exit_code == 0:
for name, value in (('start_line', start_line), ('end_line', end_line)):
if value is not None and not _is_valid_line_number(value):
return {
'status': 'ok',
'content': _truncate(
res.stdout,
limit=self._max_output_chars,
),
'status': 'error',
'error': f'`{name}` must be an integer if provided.',
}

try:
Expand Down
33 changes: 33 additions & 0 deletions tests/unittests/tools/test_local_environment.py
Original file line number Diff line number Diff line change
Expand Up @@ -81,3 +81,36 @@ async def test_read_nonexistent_raises(self, env: LocalEnvironment):
"""Reading a missing file raises FileNotFoundError."""
with pytest.raises(FileNotFoundError):
await env.read_file(Path("does_not_exist.txt"))

@pytest.mark.asyncio
async def test_absolute_path_inside_working_dir(self, env: LocalEnvironment):
"""Absolute paths are accepted when they stay inside the workspace."""
path = env.working_dir / "absolute.txt"
await env.write_file(path, "absolute")
data = await env.read_file(path)
assert data == b"absolute"

@pytest.mark.asyncio
async def test_rejects_relative_path_escape(self, env: LocalEnvironment):
"""Parent traversal cannot escape the workspace."""
outside = env.working_dir.parent / "outside.txt"
outside.write_text("secret", encoding="utf-8")

with pytest.raises(ValueError, match="escapes working directory"):
await env.read_file(Path("..") / outside.name)

with pytest.raises(ValueError, match="escapes working directory"):
await env.write_file(Path("..") / "write-outside.txt", "nope")

assert not (env.working_dir.parent / "write-outside.txt").exists()

@pytest.mark.asyncio
async def test_rejects_absolute_path_outside_working_dir(
self, env: LocalEnvironment
):
"""Absolute paths outside the workspace are rejected."""
outside = env.working_dir.parent / "outside-absolute.txt"
outside.write_text("secret", encoding="utf-8")

with pytest.raises(ValueError, match="escapes working directory"):
await env.read_file(outside)
153 changes: 153 additions & 0 deletions tests/unittests/tools/test_read_file_tool.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,153 @@
# 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.

"""Tests for ReadFileTool."""

from pathlib import Path
from typing import Optional

from google.adk.environment._base_environment import BaseEnvironment
from google.adk.environment._base_environment import ExecutionResult
from google.adk.environment._local_environment import LocalEnvironment
from google.adk.tools.environment._tools import ReadFileTool
import pytest
import pytest_asyncio


class _StubEnvironment(BaseEnvironment):
"""Minimal environment double for ReadFileTool tests."""

def __init__(self, files: dict[str, bytes]):
self._files = files
self.execute_calls: list[str] = []

@property
def working_dir(self) -> Path:
return Path('/tmp/adk-test')

async def execute(
self,
command: str,
*,
timeout: Optional[float] = None,
) -> ExecutionResult:
del timeout
self.execute_calls.append(command)
raise AssertionError('ReadFileTool should not invoke execute().')

async def read_file(self, path: Path) -> bytes:
key = str(path)
if key not in self._files:
raise FileNotFoundError(key)
return self._files[key]

async def write_file(self, path: Path, content: str | bytes) -> None:
del path, content
raise NotImplementedError


@pytest.mark.asyncio
async def test_ranged_read_does_not_use_shell():
"""A ranged read must not shell out (regression: `cat -n | sed` pipeline)."""
env = _StubEnvironment({'f.txt': b'l1\nl2\nl3\nl4\n'})
tool = ReadFileTool(env)

result = await tool.run_async(
args={'path': 'f.txt', 'start_line': 2, 'end_line': 3},
tool_context=None,
)

assert result['status'] == 'ok'
assert result['content'] == ' 2\tl2\n 3\tl3\n'
assert not env.execute_calls


@pytest.mark.asyncio
async def test_shell_metacharacters_in_path_are_literal():
"""A path with shell metacharacters is treated as a literal filename."""
env = _StubEnvironment({'f.txt': b'l1\nl2\n'})
tool = ReadFileTool(env)

result = await tool.run_async(
args={'path': "'; id > /tmp/pwned ; echo '", 'start_line': 2},
tool_context=None,
)

assert result['status'] == 'error'
assert 'File not found' in result['error']
assert not env.execute_calls


@pytest.mark.asyncio
@pytest.mark.parametrize('field', ['start_line', 'end_line'])
@pytest.mark.parametrize('bad_value', ['2', 2.5, True, None.__class__])
async def test_non_integer_line_numbers_rejected(field, bad_value):
"""Non-integer line arguments are rejected instead of reaching a slice.

Booleans are excluded explicitly: ``isinstance(True, int)`` is True, so a
bare int check would let ``True`` through into the slice.
"""
env = _StubEnvironment({'f.txt': b'l1\nl2\n'})
tool = ReadFileTool(env)

result = await tool.run_async(
args={'path': 'f.txt', field: bad_value},
tool_context=None,
)

assert result['status'] == 'error'
assert f'`{field}` must be an integer if provided.' == result['error']
assert not env.execute_calls


@pytest_asyncio.fixture(name='env')
async def _env(tmp_path: Path):
"""Create and initialize a LocalEnvironment backed by a temp directory."""
environment = LocalEnvironment(working_dir=tmp_path)
await environment.initialize()
yield environment
await environment.close()


@pytest.mark.asyncio
async def test_shell_injection_payload_creates_no_file(
env: LocalEnvironment, tmp_path: Path
):
"""End-to-end: an injection payload must not execute against a real env."""
canary = tmp_path.parent / 'adk_injection_canary'
tool = ReadFileTool(env)

result = await tool.run_async(
args={'path': f"'; touch {canary} ; echo '", 'start_line': 2},
tool_context=None,
)

assert result['status'] == 'error'
assert not canary.exists()


@pytest.mark.asyncio
async def test_ranged_read_still_works_against_real_env(env: LocalEnvironment):
"""Ranged reads keep working after the shell path was removed."""
await env.write_file('real.txt', 'a\nb\nc\nd\n')
tool = ReadFileTool(env)

result = await tool.run_async(
args={'path': 'real.txt', 'start_line': 2, 'end_line': 3},
tool_context=None,
)

assert result['status'] == 'ok'
assert result['content'] == ' 2\tb\n 3\tc\n'
assert result['total_lines'] == 4
Loading