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/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/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_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_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/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() 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")