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
32 changes: 16 additions & 16 deletions src/google/adk/cli/dev_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,18 @@
TAG_EVALUATION = "Evaluation"


def _sanitize_test_filename(test_name: str) -> str:
"""Return a tests/ filename, stripping directories that would escape it.

``create_test`` already used ``os.path.basename``. Get, delete, and rebuild
did not, so a name like ``../outside.json`` could leave the tests folder.
"""
test_name = os.path.basename(test_name)
if not test_name.endswith(".json"):
test_name += ".json"
return test_name


class CreateTestRequest(common.BaseModel):
session_data: dict

Expand Down Expand Up @@ -912,9 +924,7 @@ async def rebuild_app_tests(
agent_dir = self._get_agent_dir(app_name)

if test_name:
if not test_name.endswith(".json"):
test_name += ".json"
path = os.path.join(agent_dir, "tests", test_name)
path = os.path.join(agent_dir, "tests", _sanitize_test_filename(test_name))
else:
path = agent_dir

Expand All @@ -939,15 +949,11 @@ async def create_test(
app_name: str, test_name: str, req: CreateTestRequest
) -> dict[str, str]:
"""Creates or updates a test file from session data."""
# Sanitize test_name to prevent directory traversal
test_name = os.path.basename(test_name)
test_name = _sanitize_test_filename(test_name)
agent_dir = self._get_agent_dir(app_name)
tests_dir = os.path.join(agent_dir, "tests")
os.makedirs(tests_dir, exist_ok=True)

if not test_name.endswith(".json"):
test_name += ".json"

test_file_path = os.path.join(tests_dir, test_name)

with open(test_file_path, "w", encoding="utf-8") as f:
Expand All @@ -963,10 +969,7 @@ async def delete_test(app_name: str, test_name: str) -> dict[str, str]:
"""Deletes a specific test file."""
agent_dir = self._get_agent_dir(app_name)
tests_dir = os.path.join(agent_dir, "tests")

if not test_name.endswith(".json"):
test_name += ".json"

test_name = _sanitize_test_filename(test_name)
test_file_path = os.path.join(tests_dir, test_name)

if not os.path.exists(test_file_path):
Expand All @@ -980,10 +983,7 @@ async def get_test_content(app_name: str, test_name: str) -> dict[str, Any]:
"""Fetches the content of a specific test file."""
agent_dir = self._get_agent_dir(app_name)
tests_dir = os.path.join(agent_dir, "tests")

if not test_name.endswith(".json"):
test_name += ".json"

test_name = _sanitize_test_filename(test_name)
test_file_path = os.path.join(tests_dir, test_name)

if not os.path.exists(test_file_path):
Expand Down
30 changes: 30 additions & 0 deletions tests/unittests/cli/test_adk_web_server_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -195,6 +195,36 @@ def test_get_test_content_not_found(test_client):
assert response.status_code == 404


def test_delete_test_rejects_path_traversal(test_client, tmp_path):
"""GET/DELETE must use the same basename rule as create_test."""
agent_dir = tmp_path / "test_app"
tests_dir = agent_dir / "tests"
tests_dir.mkdir(parents=True)
outside = agent_dir / "outside.json"
outside.write_text('{"secret": true}')

encoded = "%2e%2e%2foutside.json"
delete = test_client.delete(f"/dev/apps/test_app/tests/{encoded}")
get = test_client.get(f"/dev/apps/test_app/tests/{encoded}")

assert delete.status_code == 404
assert get.status_code == 404
assert outside.exists()
assert outside.read_text() == '{"secret": true}'


def test_rebuild_single_test_rejects_path_traversal(test_client, tmp_path):
with patch("google.adk.cli.dev_server.asyncio.to_thread") as mock_to_thread:
mock_to_thread.return_value = None
response = test_client.post(
"/dev/apps/test_app/tests/rebuild?test_name=../outside.json", json={}
)
assert response.status_code == 200
args, _kwargs = mock_to_thread.call_args
test_dir, test_name = os.path.split(args[1])
assert (os.path.basename(test_dir), test_name) == ("tests", "outside.json")


def test_rebuild_tests(test_client):
with patch("google.adk.cli.dev_server.asyncio.to_thread") as mock_to_thread:
mock_to_thread.return_value = None
Expand Down