diff --git a/src/google/adk/cli/dev_server.py b/src/google/adk/cli/dev_server.py index fa3807df394..eac39109279 100644 --- a/src/google/adk/cli/dev_server.py +++ b/src/google/adk/cli/dev_server.py @@ -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 @@ -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 @@ -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: @@ -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): @@ -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): diff --git a/tests/unittests/cli/test_adk_web_server_tests.py b/tests/unittests/cli/test_adk_web_server_tests.py index 62781ba207a..039c74782f9 100644 --- a/tests/unittests/cli/test_adk_web_server_tests.py +++ b/tests/unittests/cli/test_adk_web_server_tests.py @@ -195,6 +195,50 @@ def test_get_test_content_not_found(test_client): assert response.status_code == 404 +def test_get_test_rejects_path_traversal(test_client, tmp_path): + """GET 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" + get = test_client.get(f"/dev/apps/test_app/tests/{encoded}") + + assert get.status_code == 404 + assert outside.exists() + assert outside.read_text() == '{"secret": true}' + + +def test_delete_test_rejects_path_traversal(test_client, tmp_path): + """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}") + + assert delete.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