fix: Disallow args on /builder and Add warning about Web UI usage to CLI help

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

Author: sasha-gitg

src/google/adk/cli/cli_tools_click.py

@@ -1713,7 +1713,8 @@ def cli_api_server(
     default=False,
     help=(
         "Optional. Deploy ADK Web UI if set. (default: deploy ADK API server"
-        " only)"
+        " only). WARNING: The web UI is for development and testing only — do"
+        " not use in production."
     ),
 )
 @click.option(
@@ -2229,7 +2230,8 @@ def cli_deploy_agent_engine(
     default=False,
     help=(
         "Optional. Deploy ADK Web UI if set. (default: deploy ADK API server"
-        " only)"
+        " only). WARNING: The web UI is for development and testing only — do"
+        " not use in production."
     ),
 )
 @click.option(

src/google/adk/cli/fast_api.py

@@ -27,6 +27,7 @@
 
 import click
 from fastapi import FastAPI
+from fastapi import HTTPException
 from fastapi import UploadFile
 from fastapi.responses import FileResponse
 from fastapi.responses import PlainTextResponse
@@ -293,6 +294,39 @@ def _has_parent_reference(path: str) -> bool:
 
     _ALLOWED_EXTENSIONS = frozenset({".yaml", ".yml"})
 
+    # --- YAML content security ---
+    # The `args` key in agent YAML configs (CodeConfig.args, ToolConfig.args)
+    # allows callers to pass arbitrary arguments to Python constructors and
+    # functions, which is an RCE vector when exposed through the builder UI.
+    # 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."""
+      import yaml
+
+      try:
+        docs = list(yaml.safe_load_all(content))
+      except yaml.YAMLError as exc:
+        raise ValueError(f"Invalid YAML in {filename!r}: {exc}") from exc
+
+      def _walk(node: Any) -> None:
+        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 builder uploads "
+                  "because it can execute arbitrary code."
+              )
+            _walk(value)
+        elif isinstance(node, list):
+          for item in node:
+            _walk(item)
+
+      for doc in docs:
+        _walk(doc)
+
     def _parse_upload_filename(filename: Optional[str]) -> tuple[str, str]:
       if not filename:
         raise ValueError("Upload filename is missing.")
@@ -430,40 +464,14 @@ async def builder_build(
         files: list[UploadFile], tmp: Optional[bool] = False
     ) -> bool:
       try:
-        if tmp:
-          app_names = set()
-          uploads = []
-          for file in files:
-            app_name, rel_path = _parse_upload_filename(file.filename)
-            app_names.add(app_name)
-            uploads.append((rel_path, file))
-
-          if len(app_names) != 1:
-            logger.error(
-                "Exactly one app name is required, found: %s",
-                sorted(app_names),
-            )
-            return False
-
-          app_name = next(iter(app_names))
-          app_root = _get_app_root(app_name)
-          tmp_agent_root = _get_tmp_agent_root(app_root, app_name)
-          tmp_agent_root.mkdir(parents=True, exist_ok=True)
-
-          for rel_path, file in uploads:
-            destination_path = _resolve_under_dir(tmp_agent_root, rel_path)
-            destination_path.parent.mkdir(parents=True, exist_ok=True)
-            with destination_path.open("wb") as buffer:
-              shutil.copyfileobj(file.file, buffer)
-
-          return True
-
-        app_names = set()
-        uploads = []
+        # Phase 1: parse filenames and read content into memory.
+        app_names: set[str] = set()
+        uploads: list[tuple[str, bytes]] = []
         for file in files:
           app_name, rel_path = _parse_upload_filename(file.filename)
           app_names.add(app_name)
-          uploads.append((rel_path, file))
+          content = await file.read()
+          uploads.append((rel_path, content))
 
         if len(app_names) != 1:
           logger.error(
@@ -473,23 +481,40 @@ async def builder_build(
           return False
 
         app_name = next(iter(app_names))
+
+        # 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}")
+
+        # Phase 3: write validated files to disk.
+        if tmp:
+          app_root = _get_app_root(app_name)
+          tmp_agent_root = _get_tmp_agent_root(app_root, app_name)
+          tmp_agent_root.mkdir(parents=True, exist_ok=True)
+
+          for rel_path, content in uploads:
+            destination_path = _resolve_under_dir(tmp_agent_root, rel_path)
+            destination_path.parent.mkdir(parents=True, exist_ok=True)
+            destination_path.write_bytes(content)
+
+          return True
+
         app_root = _get_app_root(app_name)
         app_root.mkdir(parents=True, exist_ok=True)
 
         tmp_agent_root = _get_tmp_agent_root(app_root, app_name)
         if tmp_agent_root.is_dir():
           copy_dir_contents(tmp_agent_root, app_root)
 
-        for rel_path, file in uploads:
+        for rel_path, content in uploads:
           destination_path = _resolve_under_dir(app_root, rel_path)
           destination_path.parent.mkdir(parents=True, exist_ok=True)
-          with destination_path.open("wb") as buffer:
-            shutil.copyfileobj(file.file, buffer)
+          destination_path.write_bytes(content)
 
         return cleanup_tmp(app_name)
       except ValueError as exc:
         logger.exception("Error in builder_build: %s", exc)
-        return False
+        raise HTTPException(status_code=400, detail=str(exc))
       except OSError as exc:
         logger.exception("Error in builder_build: %s", exc)
         return False

tests/unittests/cli/test_fast_api.py

@@ -1694,8 +1694,7 @@ def test_builder_save_rejects_traversal(builder_test_client, tmp_path):
           ("app/../escape.yaml", b"nope\n", "application/x-yaml"),
       )],
   )
-  assert response.status_code == 200
-  assert response.json() is False
+  assert response.status_code == 400
   assert not (tmp_path / "escape.yaml").exists()
   assert not (tmp_path / "app" / "tmp" / "escape.yaml").exists()
 
@@ -1709,8 +1708,7 @@ def test_builder_save_rejects_py_files(builder_test_client, tmp_path):
           ("app/agent.py", b"import os\nos.system('id')\n", "text/plain"),
       )],
   )
-  assert response.status_code == 200
-  assert response.json() is False
+  assert response.status_code == 400
   assert not (tmp_path / "app" / "tmp" / "app" / "agent.py").exists()
 
 
@@ -1732,8 +1730,7 @@ def test_builder_save_rejects_non_yaml_extensions(
             (f"app/file{ext}", content, "application/octet-stream"),
         )],
     )
-    assert response.status_code == 200, f"Expected 200 for {ext}"
-    assert response.json() is False, f"Expected False for {ext}"
+    assert response.status_code == 400, f"Expected 400 for {ext}"
 
 
 def test_builder_save_allows_yaml_files(builder_test_client, tmp_path):
@@ -1759,6 +1756,44 @@ def test_builder_save_allows_yaml_files(builder_test_client, tmp_path):
   assert response.json() is True
 
 
+def test_builder_save_rejects_args_key(builder_test_client, tmp_path):
+  """Uploading YAML with an `args` key is rejected (RCE prevention)."""
+  yaml_with_args = b"""\
+name: my_tool
+args:
+  key: value
+"""
+  response = builder_test_client.post(
+      "/builder/save?tmp=true",
+      files=[(
+          "files",
+          ("app/root_agent.yaml", yaml_with_args, "application/x-yaml"),
+      )],
+  )
+  assert response.status_code == 400
+  assert "args" in response.json()["detail"]
+  assert not (tmp_path / "app" / "tmp" / "app" / "root_agent.yaml").exists()
+
+
+def test_builder_save_rejects_nested_args_key(builder_test_client, tmp_path):
+  """Uploading YAML with a nested `args` key is rejected."""
+  yaml_with_nested_args = b"""\
+tools:
+  - name: some_tool
+    args:
+      param: value
+"""
+  response = builder_test_client.post(
+      "/builder/save?tmp=true",
+      files=[(
+          "files",
+          ("app/root_agent.yaml", yaml_with_nested_args, "application/x-yaml"),
+      )],
+  )
+  assert response.status_code == 400
+  assert "args" in response.json()["detail"]
+
+
 def test_builder_get_rejects_non_yaml_file_paths(builder_test_client, tmp_path):
   """GET /builder/app/{app_name}?file_path=... rejects non-YAML extensions."""
   app_root = tmp_path / "app"