fix: block RCE vulnerability via nested YAML configurations in ADK

Co-authored-by: Sasha Sobran <asobran@google.com>
PiperOrigin-RevId: 902898232

Author: sasha-gitg

src/google/adk/agents/config_agent_utils.py

@@ -80,6 +80,31 @@ def _resolve_agent_class(agent_class: str) -> type[BaseAgent]:
   )
 
 
+_BLOCKED_YAML_KEYS = frozenset({"args"})
+_ENFORCE_DENYLIST = False
+
+
+def _set_enforce_denylist(value: bool) -> None:
+  global _ENFORCE_DENYLIST
+  _ENFORCE_DENYLIST = value
+
+
+def _check_config_for_blocked_keys(node: Any, filename: str) -> None:
+  """Recursively check if the configuration contains any blocked keys."""
+  if isinstance(node, dict):
+    for key, value in node.items():
+      if key in _BLOCKED_YAML_KEYS:
+        raise ValueError(
+            f"Blocked key {key!r} found in {filename!r}. "
+            f"The '{key}' field is not allowed in agent configurations "
+            "because it can execute arbitrary code."
+        )
+      _check_config_for_blocked_keys(value, filename)
+  elif isinstance(node, list):
+    for item in node:
+      _check_config_for_blocked_keys(item, filename)
+
+
 def _load_config_from_path(config_path: str) -> AgentConfig:
   """Load an agent's configuration from a YAML file.
 
@@ -100,6 +125,9 @@ 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:
+    _check_config_for_blocked_keys(config_data, config_path)
+
   return AgentConfig.model_validate(config_data)
 
 

src/google/adk/cli/fast_api.py

@@ -148,6 +148,12 @@ def get_fast_api_app(
     The configured FastAPI application instance.
   """
 
+  # Enable 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)
+
   # Set up eval managers.
   if eval_storage_uri:
     gcs_eval_managers = evals.create_gcs_eval_managers_from_uri(

tests/unittests/agents/test_agent_config.py

@@ -421,3 +421,23 @@ def fake_from_config(path: str):
   )
   assert result == "sentinel"
   assert recorded["path"] == expected_path
+
+
+def test_load_config_from_path_blocks_args_when_enforced(tmp_path):
+  """Verify _load_config_from_path blocks 'args' when enforcement is enabled."""
+  config_file = tmp_path / "malicious.yaml"
+  config_file.write_text("""
+  name: malicious_agent
+  tools:
+    - name: some_tool
+      args:
+        cmd: "rm -rf /"
+  """)
+
+  config_agent_utils._set_enforce_denylist(True)
+  try:
+    with pytest.raises(ValueError) as exc_info:
+      config_agent_utils._load_config_from_path(str(config_file))
+    assert "Blocked key 'args' found" in str(exc_info.value)
+  finally:
+    config_agent_utils._set_enforce_denylist(False)