From 5b73395772a490a995dcbb487650b31d6e23bd42 Mon Sep 17 00:00:00 2001 From: George Weale Date: Mon, 17 Aug 2026 22:54:02 +0000 Subject: [PATCH 1/2] fix(cli): Block special internal agents in ApiServer mode (v1) Port of 3f6eb1f0 from main. Any client of a headless `adk api_server` could load and run the special internal agents that back the dev UI, `__adk_agent_builder_assistant` among them, by naming one as the app. The agent loader accepted any `__`-prefixed name unconditionally, and the server never looked at the prefix. Loading a `__`-prefixed agent now raises `PermissionError` unless the loader was configured to allow it, and `AdkWebServer` answers 403 for those names before it asks the loader at all. `get_fast_api_app` turns both on from its existing `web` argument, so the dev UI keeps working and a headless api_server does not. Behaviour change: driving a special internal agent through `adk api_server` now returns 403 where it used to run the agent. Adapted for v1, which has a single `AdkWebServer` rather than the `ApiServer` and `DevServer` pair upstream writes the flag against. Upstream guards two `load_agent` call sites; its other four sit on `DevServer`, which is never constructed when `web=False`, so those routes do not exist there at all. v1 registers every route on one class, so all six calls go through a single `_load_agent_or_app` helper that holds the 403. That makes the four eval and graph routes answer 403 where upstream answers 404, and it is what covers a caller who supplies an agent loader of their own that ignores the loader flag. --- src/google/adk/cli/adk_web_server.py | 26 +- src/google/adk/cli/fast_api.py | 6 + src/google/adk/cli/utils/agent_loader.py | 5 + src/google/adk/cli/utils/base_agent_loader.py | 2 + tests/unittests/cli/test_fast_api.py | 296 ++++++++++++++++++ .../unittests/cli/utils/test_agent_loader.py | 21 ++ 6 files changed, 350 insertions(+), 6 deletions(-) diff --git a/src/google/adk/cli/adk_web_server.py b/src/google/adk/cli/adk_web_server.py index a9a9a53604d..e4dd26a52e5 100644 --- a/src/google/adk/cli/adk_web_server.py +++ b/src/google/adk/cli/adk_web_server.py @@ -690,6 +690,8 @@ class AdkWebServer: runner_dict: A dict of instantiated runners for each app. """ + _allow_special_agents: bool = False + def __init__( self, *, @@ -741,7 +743,7 @@ async def get_runner_async(self, app_name: str) -> Runner: # Create new runner envs.load_dotenv_for_agent(os.path.basename(app_name), self.agents_dir) - agent_or_app = self.agent_loader.load_agent(app_name) + agent_or_app = self._load_agent_or_app(app_name) # Instantiate extra plugins if configured extra_plugins_instances = self._instantiate_extra_plugins() @@ -797,6 +799,18 @@ async def get_runner_async(self, app_name: str) -> Runner: self.runner_dict[app_name] = runner return runner + def _load_agent_or_app(self, app_name: str) -> BaseAgent | App: + """Loads an agent, refusing internal special agents unless enabled.""" + if app_name.startswith("__") and not self._allow_special_agents: + raise HTTPException( + status_code=403, + detail=( + "Access to internal special agents is disabled in API server" + " mode." + ), + ) + return self.agent_loader.load_agent(app_name) + def _get_root_agent(self, agent_or_app: BaseAgent | App) -> BaseAgent: """Extract root agent from either a BaseAgent or App object.""" if isinstance(agent_or_app, App): @@ -1057,7 +1071,7 @@ async def list_apps( @app.get("/apps/{app_name}/app-info", response_model_exclude_none=True) async def get_adk_app_info(app_name: str) -> AppInfo: """Returns the detailed info for a given ADK app.""" - agent_or_app = self.agent_loader.load_agent(app_name) + agent_or_app = self._load_agent_or_app(app_name) root_agent = self._get_root_agent(agent_or_app) if isinstance(root_agent, LlmAgent): return AppInfo( @@ -1469,7 +1483,7 @@ async def add_session_to_eval_set( invocations = evals.convert_session_to_eval_invocations(session) # Populate the session with initial session state. - agent_or_app = self.agent_loader.load_agent(app_name) + agent_or_app = self._load_agent_or_app(app_name) root_agent = self._get_root_agent(agent_or_app) initial_session_state = create_empty_state(root_agent) @@ -1616,7 +1630,7 @@ async def run_eval( status_code=400, detail=f"Eval set `{eval_set_id}` not found." ) - agent_or_app = self.agent_loader.load_agent(app_name) + agent_or_app = self._load_agent_or_app(app_name) root_agent = self._get_root_agent(agent_or_app) eval_case_results = [] @@ -2052,7 +2066,7 @@ async def get_app_graph_dot( app_name: The name of the agent/app dark_mode: Whether to use dark theme background color """ - agent_or_app = self.agent_loader.load_agent(app_name) + agent_or_app = self._load_agent_or_app(app_name) root_agent = self._get_root_agent(agent_or_app) # Get graph with NO highlights (empty list) and specified theme @@ -2084,7 +2098,7 @@ async def get_event_graph( function_calls = event.get_function_calls() function_responses = event.get_function_responses() - agent_or_app = self.agent_loader.load_agent(app_name) + agent_or_app = self._load_agent_or_app(app_name) root_agent = self._get_root_agent(agent_or_app) dot_graph = None if function_calls: diff --git a/src/google/adk/cli/fast_api.py b/src/google/adk/cli/fast_api.py index 89b40fe88ec..0d12feec659 100644 --- a/src/google/adk/cli/fast_api.py +++ b/src/google/adk/cli/fast_api.py @@ -172,6 +172,9 @@ def get_fast_api_app( # initialize Agent Loader if not passed as argument if agent_loader is None: agent_loader = AgentLoader(agents_dir) + # Special internal agents back the dev UI only, so they stay unloadable + # unless the UI is being served. + agent_loader._allow_special_agents = web # Load services.py from agents_dir for custom service registration. load_services_module(agents_dir) @@ -223,6 +226,9 @@ def get_fast_api_app( auto_create_session=auto_create_session, trigger_sources=trigger_sources, ) + # The loader flag stops the import; this one turns the rejection into a 403 + # rather than an uncaught error, and also covers a custom agent_loader. + adk_web_server._allow_special_agents = web # Callbacks & other optional args for when constructing the FastAPI instance extra_fast_api_args = {} diff --git a/src/google/adk/cli/utils/agent_loader.py b/src/google/adk/cli/utils/agent_loader.py index a7bbcbc2a65..e84734a4c28 100644 --- a/src/google/adk/cli/utils/agent_loader.py +++ b/src/google/adk/cli/utils/agent_loader.py @@ -194,6 +194,11 @@ def _validate_agent_name(self, agent_name: str) -> None: """Validate agent name to prevent arbitrary module imports.""" # Strip the special agent prefix for validation if agent_name.startswith("__"): + if not self._allow_special_agents: + raise PermissionError( + f"Loading special internal agent {agent_name!r} is disabled in this" + " loader configuration." + ) name_to_check = agent_name[2:] check_dir = os.path.abspath(SPECIAL_AGENTS_DIR) else: diff --git a/src/google/adk/cli/utils/base_agent_loader.py b/src/google/adk/cli/utils/base_agent_loader.py index 1356ce12f1f..1eb4e80ce4c 100644 --- a/src/google/adk/cli/utils/base_agent_loader.py +++ b/src/google/adk/cli/utils/base_agent_loader.py @@ -28,6 +28,8 @@ class BaseAgentLoader(ABC): """Abstract base class for agent loaders.""" + _allow_special_agents: bool = False + @abstractmethod def load_agent(self, agent_name: str) -> Union[BaseAgent, App]: """Loads an instance of an agent with the given name.""" diff --git a/tests/unittests/cli/test_fast_api.py b/tests/unittests/cli/test_fast_api.py index 14d1e17bfea..80cc57db0f4 100755 --- a/tests/unittests/cli/test_fast_api.py +++ b/tests/unittests/cli/test_fast_api.py @@ -645,6 +645,302 @@ def test_agent_with_bigquery_analytics_plugin( assert getattr(runner.app, "_is_visual_builder_app", False) is True +def _create_adk_web_server( + tmp_path, + mock_session_service, + mock_artifact_service, + mock_memory_service, + mock_agent_loader, + mock_eval_sets_manager, + mock_eval_set_results_manager, +): + """Helper to build an AdkWebServer backed by the mock service fixtures.""" + from google.adk.cli.adk_web_server import AdkWebServer + + return AdkWebServer( + agent_loader=mock_agent_loader, + session_service=mock_session_service, + memory_service=mock_memory_service, + artifact_service=mock_artifact_service, + credential_service=MagicMock(), + eval_sets_manager=mock_eval_sets_manager, + eval_set_results_manager=mock_eval_set_results_manager, + agents_dir=str(tmp_path), + ) + + +def test_get_runner_async_rejects_internal_special_agent_name( + tmp_path, + mock_session_service, + mock_artifact_service, + mock_memory_service, + mock_agent_loader, + mock_eval_sets_manager, + mock_eval_set_results_manager, +): + adk_web_server = _create_adk_web_server( + tmp_path, + mock_session_service, + mock_artifact_service, + mock_memory_service, + mock_agent_loader, + mock_eval_sets_manager, + mock_eval_set_results_manager, + ) + + from fastapi import HTTPException + + with pytest.raises(HTTPException) as exc_info: + asyncio.run( + adk_web_server.get_runner_async("__adk_agent_builder_assistant") + ) + + assert exc_info.value.status_code == 403 + assert ( + "Access to internal special agents is disabled in API server mode" + in exc_info.value.detail + ) + + +def test_get_runner_async_accepts_internal_special_agent_name_when_enabled( + tmp_path, + mock_session_service, + mock_artifact_service, + mock_memory_service, + mock_agent_loader, + mock_eval_sets_manager, + mock_eval_set_results_manager, +): + adk_web_server = _create_adk_web_server( + tmp_path, + mock_session_service, + mock_artifact_service, + mock_memory_service, + mock_agent_loader, + mock_eval_sets_manager, + mock_eval_set_results_manager, + ) + adk_web_server._allow_special_agents = True + + runner = asyncio.run( + adk_web_server.get_runner_async("__adk_agent_builder_assistant") + ) + + assert runner.app.name == "__adk_agent_builder_assistant" + + +def test_app_info_rejects_internal_special_agent_name_without_web( + tmp_path, + mock_session_service, + mock_artifact_service, + mock_memory_service, + mock_agent_loader, + mock_eval_sets_manager, + mock_eval_set_results_manager, +): + client = _create_test_client( + mock_session_service, + mock_artifact_service, + mock_memory_service, + mock_agent_loader, + mock_eval_sets_manager, + mock_eval_set_results_manager, + agents_dir=str(tmp_path), + web=False, + ) + + response = client.get("/apps/__adk_agent_builder_assistant/app-info") + + assert response.status_code == 403 + assert ( + "Access to internal special agents is disabled in API server mode" + in response.json()["detail"] + ) + + +def test_app_info_allows_internal_special_agent_name_with_web( + tmp_path, + mock_session_service, + mock_artifact_service, + mock_memory_service, + mock_agent_loader, + mock_eval_sets_manager, + mock_eval_set_results_manager, +): + special_agent = LlmAgent(name="agent_builder_assistant") + mock_agent_loader.load_agent = lambda app_name: special_agent + client = _create_test_client( + mock_session_service, + mock_artifact_service, + mock_memory_service, + mock_agent_loader, + mock_eval_sets_manager, + mock_eval_set_results_manager, + agents_dir=str(tmp_path), + web=True, + ) + + response = client.get("/apps/__adk_agent_builder_assistant/app-info") + + assert response.status_code == 200 + assert response.json()["rootAgentName"] == "agent_builder_assistant" + + +def test_agent_loader_allows_special_agents_only_when_web_is_enabled( + tmp_path, + mock_session_service, + mock_artifact_service, + mock_memory_service, + mock_agent_loader, + mock_eval_sets_manager, + mock_eval_set_results_manager, +): + def build(web: bool) -> None: + _create_test_client( + mock_session_service, + mock_artifact_service, + mock_memory_service, + mock_agent_loader, + mock_eval_sets_manager, + mock_eval_set_results_manager, + agents_dir=str(tmp_path), + web=web, + ) + + build(web=False) + assert mock_agent_loader._allow_special_agents is False + + build(web=True) + assert mock_agent_loader._allow_special_agents is True + + +_SPECIAL_APP_NAME = "__adk_agent_builder_assistant" + + +class _FlagIgnoringLoader: + """A caller-supplied loader that does not honour _allow_special_agents.""" + + def __init__(self): + self.requested = [] + + def load_agent(self, app_name): + self.requested.append(app_name) + return DummyAgent(name="agent_builder_assistant") + + def list_agents(self): + return [] + + +def _create_api_server_client(loader, **overrides): + """Builds a TestClient over an AdkWebServer left in API server mode.""" + from google.adk.cli.adk_web_server import AdkWebServer + + kwargs = dict( + agent_loader=loader, + session_service=InMemorySessionService(), + memory_service=MagicMock(), + artifact_service=MagicMock(), + credential_service=MagicMock(), + eval_sets_manager=InMemoryEvalSetsManager(), + eval_set_results_manager=MagicMock(), + agents_dir=".", + ) + kwargs.update(overrides) + adk_web_server = AdkWebServer(**kwargs) + fast_api_app = adk_web_server.get_fast_api_app( + setup_observer=lambda _observer, _server: None, + tear_down_observer=lambda _observer, _server: None, + ) + return TestClient(fast_api_app) + + +def test_dev_graph_rejects_internal_special_agent_name(): + loader = _FlagIgnoringLoader() + client = _create_api_server_client(loader) + + response = client.get(f"/dev/{_SPECIAL_APP_NAME}/graph") + + assert response.status_code == 403 + assert loader.requested == [] + + +def test_run_eval_rejects_internal_special_agent_name(): + loader = _FlagIgnoringLoader() + eval_sets_manager = InMemoryEvalSetsManager() + eval_sets_manager.create_eval_set(_SPECIAL_APP_NAME, "eval_set_id") + client = _create_api_server_client( + loader, eval_sets_manager=eval_sets_manager + ) + + response = client.post( + f"/apps/{_SPECIAL_APP_NAME}/eval-sets/eval_set_id/run", + json={"evalMetrics": []}, + ) + + assert response.status_code == 403 + assert loader.requested == [] + + +def test_add_session_to_eval_set_rejects_internal_special_agent_name(): + loader = _FlagIgnoringLoader() + session_service = InMemorySessionService() + asyncio.run( + session_service.create_session( + app_name=_SPECIAL_APP_NAME, user_id="user", session_id="session_id" + ) + ) + eval_sets_manager = InMemoryEvalSetsManager() + eval_sets_manager.create_eval_set(_SPECIAL_APP_NAME, "eval_set_id") + client = _create_api_server_client( + loader, + session_service=session_service, + eval_sets_manager=eval_sets_manager, + ) + + response = client.post( + f"/apps/{_SPECIAL_APP_NAME}/eval_sets/eval_set_id/add_session", + json={"evalId": "eval_id", "sessionId": "session_id", "userId": "user"}, + ) + + assert response.status_code == 403 + assert loader.requested == [] + + +def test_event_graph_rejects_internal_special_agent_name(): + loader = _FlagIgnoringLoader() + session_service = AsyncMock() + session = Session( + id="session_id", + app_name=_SPECIAL_APP_NAME, + user_id="user", + state={}, + events=[Event(author="dummy_agent")], + ) + session_service.get_session.return_value = session + client = _create_api_server_client(loader, session_service=session_service) + + response = client.get( + f"/apps/{_SPECIAL_APP_NAME}/users/user/sessions/session_id/events/" + f"{session.events[0].id}/graph" + ) + + assert response.status_code == 403 + assert loader.requested == [] + + +def test_dev_graph_rejects_special_agent_before_the_loader_raises(tmp_path): + """The default loader's PermissionError must never reach the client.""" + from google.adk.cli.utils.agent_loader import AgentLoader + + client = _create_api_server_client( + AgentLoader(str(tmp_path)), agents_dir=str(tmp_path) + ) + + response = client.get(f"/dev/{_SPECIAL_APP_NAME}/graph") + + assert response.status_code == 403 + + @pytest.fixture def test_app( mock_session_service, diff --git a/tests/unittests/cli/utils/test_agent_loader.py b/tests/unittests/cli/utils/test_agent_loader.py index 84543231894..58dcb2c1fb6 100644 --- a/tests/unittests/cli/utils/test_agent_loader.py +++ b/tests/unittests/cli/utils/test_agent_loader.py @@ -679,6 +679,7 @@ def test_load_special_agent_with_double_underscore(self): # Load the special agent loader = AgentLoader(str(regular_agents_dir)) + loader._allow_special_agents = True agent = loader.load_agent("__helper") # Assert agent was loaded correctly @@ -718,6 +719,7 @@ def test_special_agent_caching_returns_same_instance(self): # Load the special agent twice loader = AgentLoader(str(regular_agents_dir)) + loader._allow_special_agents = True agent1 = loader.load_agent("__cached_helper") agent2 = loader.load_agent("__cached_helper") @@ -752,6 +754,7 @@ def test_special_agent_not_found_error(self): agent_loader.SPECIAL_AGENTS_DIR = str(special_agents_dir) loader = AgentLoader(str(regular_agents_dir)) + loader._allow_special_agents = True # Try to load nonexistent special agent with pytest.raises(ValueError) as exc_info: @@ -817,6 +820,7 @@ def test_load_special_agent_from_yaml_config(self): # Load the special agent loader = AgentLoader(str(regular_agents_dir)) + loader._allow_special_agents = True agent = loader.load_agent("__yaml_helper") # Assert agent was loaded correctly @@ -1006,3 +1010,20 @@ def test_validate_agent_name_rejects_nonexistent_agent(self): # 'subprocess' is a valid identifier but shouldn't be importable as an agent with pytest.raises(ValueError, match="Agent not found"): loader.load_agent("subprocess") + + def test_validate_agent_name_rejects_special_agents_by_default(self): + """Special agents starting with __ are rejected by default (_allow_special_agents=False).""" + with tempfile.TemporaryDirectory() as temp_dir: + loader = AgentLoader(temp_dir) + with pytest.raises( + PermissionError, match="Loading special internal agent" + ): + loader._validate_agent_name("__adk_agent_builder_assistant") + + def test_validate_agent_name_allows_special_agents_when_enabled(self): + """Special agents starting with __ are allowed when _allow_special_agents=True.""" + with tempfile.TemporaryDirectory() as temp_dir: + loader = AgentLoader(temp_dir) + loader._allow_special_agents = True + # Should not raise any exception + loader._validate_agent_name("__adk_agent_builder_assistant") From 4a508602ece5ec545c80782c2198509d559082ba Mon Sep 17 00:00:00 2001 From: George Weale Date: Mon, 17 Aug 2026 22:56:09 +0000 Subject: [PATCH 2/2] 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. --- .../utils/resolve_root_directory.py | 36 +++-- .../test_adk_web_server_import_isolation.py | 50 +++++++ .../cli/test_resolve_root_directory.py | 140 ++++++++++++++++++ 3 files changed, 213 insertions(+), 13 deletions(-) create mode 100644 tests/unittests/cli/test_adk_web_server_import_isolation.py create mode 100644 tests/unittests/cli/test_resolve_root_directory.py diff --git a/src/google/adk/cli/built_in_agents/utils/resolve_root_directory.py b/src/google/adk/cli/built_in_agents/utils/resolve_root_directory.py index ca7398733f2..09027fa44b9 100644 --- a/src/google/adk/cli/built_in_agents/utils/resolve_root_directory.py +++ b/src/google/adk/cli/built_in_agents/utils/resolve_root_directory.py @@ -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( diff --git a/tests/unittests/cli/test_adk_web_server_import_isolation.py b/tests/unittests/cli/test_adk_web_server_import_isolation.py new file mode 100644 index 00000000000..8134330403a --- /dev/null +++ b/tests/unittests/cli/test_adk_web_server_import_isolation.py @@ -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 diff --git a/tests/unittests/cli/test_resolve_root_directory.py b/tests/unittests/cli/test_resolve_root_directory.py new file mode 100644 index 00000000000..b442be8cbe4 --- /dev/null +++ b/tests/unittests/cli/test_resolve_root_directory.py @@ -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()