fix: Port agent-config code-reference and config_path validation to v1

src/google/adk/agents/config_agent_utils.py

@@ -17,6 +17,7 @@
 import importlib
 import inspect
 import os
+import sys
 from typing import Any
 from typing import List
 
@@ -81,12 +82,12 @@ def _resolve_agent_class(agent_class: str) -> type[BaseAgent]:
 
 
 _BLOCKED_YAML_KEYS = frozenset({"args"})
-_ENFORCE_DENYLIST = False
+_ENFORCE_YAML_KEY_DENYLIST = False
 
 
-def _set_enforce_denylist(value: bool) -> None:
-  global _ENFORCE_DENYLIST
-  _ENFORCE_DENYLIST = value
+def _set_enforce_yaml_key_denylist(value: bool) -> None:
+  global _ENFORCE_YAML_KEY_DENYLIST
+  _ENFORCE_YAML_KEY_DENYLIST = value
 
 
 def _check_config_for_blocked_keys(node: Any, filename: str) -> None:
@@ -125,16 +126,132 @@ def _load_config_from_path(config_path: str) -> AgentConfig:
   with open(config_path, "r", encoding="utf-8") as f:
     config_data = yaml.safe_load(f)
 
-  if _ENFORCE_DENYLIST:
+  if _ENFORCE_YAML_KEY_DENYLIST:
     _check_config_for_blocked_keys(config_data, config_path)
 
   return AgentConfig.model_validate(config_data)
 
 
+_ENFORCE_DENYLIST = True
+
+# Agent configs never need the standard library: they name the agent's own
+# package, google.adk, or a third-party integration. So block all of it. Listing
+# only the scary modules does not work, because cProfile.run, timeit.timeit and
+# trace.Trace.run all execute a string you hand them, and each Python release
+# can add more.
+_STDLIB_MODULES = frozenset(sys.stdlib_module_names) | frozenset(
+    sys.builtin_module_names  # Redundant on stock CPython, not custom builds.
+)
+
+# Extra names to block. Everything above the LOAD-BEARING line below is already
+# covered by _STDLIB_MODULES and is kept only to spell out the threat model.
+_BLOCKED_MODULES = frozenset({
+    # Process / OS execution
+    "os",
+    "posix",  # Unix alias: posix.system is os.system
+    "nt",  # Windows alias: nt.system is os.system
+    "subprocess",
+    "_posixsubprocess",
+    "sys",
+    "builtins",
+    "importlib",
+    "shutil",
+    "signal",
+    "multiprocessing",
+    "threading",
+    # Dynamic code evaluation
+    "code",
+    "codeop",
+    "compileall",
+    "runpy",
+    # Native / unsafe extensions
+    "ctypes",
+    # Network access
+    "socket",
+    "_socket",
+    "http",
+    "urllib",
+    "ftplib",
+    "smtplib",
+    "poplib",
+    "imaplib",
+    "xmlrpc",
+    "asyncio",
+    # Filesystem / serialisation
+    "tempfile",
+    "pathlib",
+    "shelve",
+    "pickle",
+    "marshal",
+    # Interactive / side-effect modules
+    "webbrowser",
+    "antigravity",
+    "pty",
+    "pdb",
+    "profile",
+    # LOAD-BEARING, keep these. They are not in sys.stdlib_module_names on
+    # every Python we support, so this set is all that blocks them.
+    #
+    # Modules dropped from the standard library that you can still import:
+    # distutils comes back through setuptools' shim and its spawn() runs a
+    # subprocess, and the rest have "standard-*" packages on PyPI. commands is
+    # a Python 2 leftover.
+    "asynchat",
+    "asyncore",
+    "cgi",
+    "commands",
+    "crypt",
+    "distutils",
+    "imp",
+    "mailcap",
+    "nntplib",
+    "pipes",
+    "smtpd",
+    "telnetlib",
+    "uu",
+    # CPython's own test packages, which most installs ship. They can start a
+    # subprocess (test.support.script_helper) and execute source (_testcapi).
+    "_testcapi",
+    "_testinternalcapi",
+    "test",
+})
+
+
+def _validate_module_reference(fully_qualified_name: str) -> None:
+  """Validate that a module reference does not target a blocked module.
+
+  Args:
+    fully_qualified_name: The fully-qualified Python name to validate (e.g.
+      ``"my_package.my_module.my_func"``).
+
+  Raises:
+    ValueError: If the top-level module is part of the Python standard library
+      or is in ``_BLOCKED_MODULES``.
+  """
+  if not _ENFORCE_DENYLIST:
+    return
+  # Extract the top-level package from the fully-qualified name.
+  top_module = fully_qualified_name.split(".")[0]
+  if top_module in _BLOCKED_MODULES or top_module in _STDLIB_MODULES:
+    raise ValueError(
+        f"Blocked module reference: {fully_qualified_name!r}. Agent "
+        f"configurations cannot import from '{top_module}'. The Python "
+        "standard library is blocked in full because too much of it can "
+        "execute arbitrary code. Reference your own agent package, "
+        "'google.adk', or a third-party package instead."
+    )
+
+
+def _set_enforce_denylist(value: bool) -> None:
+  global _ENFORCE_DENYLIST
+  _ENFORCE_DENYLIST = value
+
+
 @experimental(FeatureName.AGENT_CONFIG)
 def resolve_fully_qualified_name(name: str) -> Any:
   try:
     module_path, obj_name = name.rsplit(".", 1)
+    _validate_module_reference(name)
     module = importlib.import_module(module_path)
     return getattr(module, obj_name)
   except Exception as e:
@@ -150,28 +267,38 @@ def resolve_agent_reference(
   Args:
     ref_config: The agent reference configuration (AgentRefConfig).
     referencing_agent_config_abs_path: The absolute path to the agent config
-    that contains the reference.
+      that contains the reference.
 
   Returns:
     The created agent instance.
   """
   if ref_config.config_path:
     if os.path.isabs(ref_config.config_path):
-      return from_config(ref_config.config_path)
-    else:
-      return from_config(
-          os.path.join(
-              os.path.dirname(referencing_agent_config_abs_path),
-              ref_config.config_path,
-          )
+      raise ValueError(
+          "Absolute paths are not allowed in AgentRefConfig config_path:"
+          f" {ref_config.config_path!r}"
+      )
+    agent_dir = os.path.dirname(referencing_agent_config_abs_path)
+    resolved_path = os.path.realpath(
+        os.path.join(agent_dir, ref_config.config_path)
+    )
+    canonical_agent_dir = os.path.realpath(agent_dir)
+    if (
+        os.path.commonpath([canonical_agent_dir, resolved_path])
+        != canonical_agent_dir
+    ):
+      raise ValueError(
+          f"Path traversal detected: config_path {ref_config.config_path!r}"
+          " resolves outside the agent directory"
       )
+    return from_config(resolved_path)
   elif ref_config.code:
     return _resolve_agent_code_reference(ref_config.code)
   else:
     raise ValueError("AgentRefConfig must have either 'code' or 'config_path'")
 
 
-def _resolve_agent_code_reference(code: str) -> Any:
+def _resolve_agent_code_reference(code: str) -> BaseAgent:
   """Resolve a code reference to an actual agent instance.
 
   Args:
@@ -186,6 +313,7 @@ def _resolve_agent_code_reference(code: str) -> Any:
   if "." not in code:
     raise ValueError(f"Invalid code reference: {code}")
 
+  _validate_module_reference(code)
   module_path, obj_name = code.rsplit(".", 1)
   module = importlib.import_module(module_path)
   obj = getattr(module, obj_name)
@@ -215,6 +343,7 @@ def resolve_code_reference(code_config: CodeConfig) -> Any:
   if not code_config or not code_config.name:
     raise ValueError("Invalid CodeConfig.")
 
+  _validate_module_reference(code_config.name)
   module_path, obj_name = code_config.name.rsplit(".", 1)
   module = importlib.import_module(module_path)
   obj = getattr(module, obj_name)

src/google/adk/agents/llm_agent.py

@@ -933,6 +933,9 @@ def _resolve_tools(
         obj = getattr(module, tool_config.name)
       else:
         # User-defined tools
+        from .config_agent_utils import _validate_module_reference
+
+        _validate_module_reference(tool_config.name)
         module_path, obj_name = tool_config.name.rsplit('.', 1)
         module = importlib.import_module(module_path)
         obj = getattr(module, obj_name)

src/google/adk/cli/fast_api.py

@@ -23,6 +23,7 @@
 import shutil
 import sys
 from typing import Any
+from typing import Iterator
 from typing import Literal
 from typing import Mapping
 
@@ -73,6 +74,103 @@ def __getattr__(name: str):
   return attr
 
 
+# Agent config fields whose value names Python code that the agent loader
+# imports and calls.
+_CODE_REFERENCE_KEYS = frozenset({
+    "after_agent_callbacks",
+    "after_model_callbacks",
+    "after_tool_callbacks",
+    "agent_class",
+    "before_agent_callbacks",
+    "before_model_callbacks",
+    "before_tool_callbacks",
+    "code",
+    "input_schema",
+    "model_code",
+    "output_schema",
+    "tools",
+})
+
+# The namespaces the agent loader searches when a reference has no dots.
+_ADK_BUILT_IN_NAMESPACES = ("google.adk.agents.", "google.adk.tools.")
+
+
+def _iter_code_references(value: Any) -> Iterator[str]:
+  """Yields the names a code-reference field carries, whatever its shape."""
+  if isinstance(value, str):
+    yield value
+  elif isinstance(value, list):
+    for item in value:
+      yield from _iter_code_references(item)
+  elif isinstance(value, dict):
+    name = value.get("name")
+    if isinstance(name, str):
+      yield name
+
+
+def _is_adk_built_in(reference: str) -> bool:
+  """Whether a qualified name reaches what an undotted name would reach.
+
+  One segment after the namespace is a name that namespace exports. A deeper
+  path walks into a submodule and can reach code an undotted reference cannot,
+  so it does not count as a built-in.
+
+  Args:
+    reference: A dotted Python name.
+
+  Returns:
+    Whether the reference names an ADK built-in.
+  """
+  for namespace in _ADK_BUILT_IN_NAMESPACES:
+    if reference.startswith(namespace):
+      return "." not in reference[len(namespace) :]
+  return False
+
+
+def _app_name_shadows_module(app_name: str) -> bool:
+  """Whether the app name collides with a module that can be imported."""
+  # "google" is a namespace package rather than a standard library module, so
+  # it has to be named explicitly.
+  return (
+      app_name in sys.builtin_module_names
+      or app_name in sys.stdlib_module_names
+      or app_name == "google"
+  )
+
+
+def _check_code_reference(
+    reference: str, *, app_name: str, filename: str, field_name: str
+) -> None:
+  """Checks that a code reference stays inside the app being edited.
+
+  Args:
+    reference: The name found in the uploaded document.
+    app_name: The app the document belongs to.
+    filename: The uploaded path, used in the error message.
+    field_name: The config field the reference came from.
+
+  Raises:
+    ValueError: If the reference can reach code outside the app.
+  """
+  if "." not in reference:
+    # The loader resolves an undotted name against ADK's own built-ins.
+    return
+  if _is_adk_built_in(reference):
+    return
+  if not reference.startswith(f"{app_name}."):
+    raise ValueError(
+        f"Blocked code reference {reference!r} in {filename!r}. The"
+        f" '{field_name}' field may only reference code under"
+        f" '{app_name}' or an ADK built-in."
+    )
+  if _app_name_shadows_module(app_name):
+    raise ValueError(
+        f"Blocked code reference {reference!r} in {filename!r}. The app name"
+        f" {app_name!r} shadows an importable Python module, so a reference to"
+        " the app cannot be told apart from one that leaves it."
+    )
+
+
 def get_fast_api_app(
     *,
     agents_dir: str,
@@ -152,11 +250,11 @@ def get_fast_api_app(
     The configured FastAPI application instance.
   """
 
-  # Enable denylist enforcement for config loads if web UI is enabled.
+  # Enable YAML key denylist enforcement for config loads if web UI is enabled.
   if web:
     from ..agents import config_agent_utils
 
-    config_agent_utils._set_enforce_denylist(True)
+    config_agent_utils._set_enforce_yaml_key_denylist(True)
 
   # Set up eval managers.
   if eval_storage_uri:
@@ -344,8 +442,10 @@ def _has_parent_reference(path: str) -> bool:
     # Block any upload that contains an `args` key anywhere in the document.
     _BLOCKED_YAML_KEYS = frozenset({"args"})
 
-    def _check_yaml_for_blocked_keys(content: bytes, filename: str) -> None:
-      """Raise if the YAML document contains any blocked keys."""
+    def _check_uploaded_yaml(
+        content: bytes, *, filename: str, app_name: str
+    ) -> None:
+      """Raise if the YAML would let the loader run code outside the app."""
       import yaml
 
       try:
@@ -362,6 +462,14 @@ def _walk(node: Any) -> None:
                   f"The '{key}' field is not allowed in builder uploads "
                   "because it can execute arbitrary code."
               )
+            if key in _CODE_REFERENCE_KEYS:
+              for reference in _iter_code_references(value):
+                _check_code_reference(
+                    reference,
+                    app_name=app_name,
+                    filename=filename,
+                    field_name=key,
+                )
             _walk(value)
         elif isinstance(node, list):
           for item in node:
@@ -527,7 +635,11 @@ async def builder_build(
 
         # Phase 2: validate every file *before* writing anything to disk.
         for rel_path, content in uploads:
-          _check_yaml_for_blocked_keys(content, f"{app_name}/{rel_path}")
+          _check_uploaded_yaml(
+              content,
+              filename=f"{app_name}/{rel_path}",
+              app_name=app_name,
+          )
 
         # Phase 3: write validated files to disk.
         if tmp: