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
26 changes: 20 additions & 6 deletions src/google/adk/cli/adk_web_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -690,6 +690,8 @@ class AdkWebServer:
runner_dict: A dict of instantiated runners for each app.
"""

_allow_special_agents: bool = False

def __init__(
self,
*,
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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)

Expand Down Expand Up @@ -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 = []
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down
36 changes: 23 additions & 13 deletions src/google/adk/cli/built_in_agents/utils/resolve_root_directory.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
6 changes: 6 additions & 0 deletions src/google/adk/cli/fast_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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 = {}
Expand Down
5 changes: 5 additions & 0 deletions src/google/adk/cli/utils/agent_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
2 changes: 2 additions & 0 deletions src/google/adk/cli/utils/base_agent_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down
50 changes: 50 additions & 0 deletions tests/unittests/cli/test_adk_web_server_import_isolation.py
Original file line number Diff line number Diff line change
@@ -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
Loading
Loading