diff --git a/src/google/adk/agents/config_agent_utils.py b/src/google/adk/agents/config_agent_utils.py index f9a3e7f5941..d36c008c182 100644 --- a/src/google/adk/agents/config_agent_utils.py +++ b/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) diff --git a/src/google/adk/agents/llm_agent.py b/src/google/adk/agents/llm_agent.py index b41b7f4effc..58d40137f4e 100644 --- a/src/google/adk/agents/llm_agent.py +++ b/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) diff --git a/src/google/adk/cli/fast_api.py b/src/google/adk/cli/fast_api.py index 89b40fe88ec..923d1023454 100644 --- a/src/google/adk/cli/fast_api.py +++ b/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: diff --git a/tests/unittests/agents/test_agent_config.py b/tests/unittests/agents/test_agent_config.py index 380078ec500..645a0398852 100644 --- a/tests/unittests/agents/test_agent_config.py +++ b/tests/unittests/agents/test_agent_config.py @@ -25,11 +25,13 @@ from google.adk.agents.base_agent import BaseAgent from google.adk.agents.base_agent_config import BaseAgentConfig from google.adk.agents.common_configs import AgentRefConfig +from google.adk.agents.common_configs import CodeConfig from google.adk.agents.llm_agent import LlmAgent from google.adk.agents.loop_agent import LoopAgent from google.adk.agents.parallel_agent import ParallelAgent from google.adk.agents.sequential_agent import SequentialAgent from google.adk.models.lite_llm import LiteLlm +from pydantic import BaseModel import pytest import yaml @@ -423,6 +425,242 @@ def fake_from_config(path: str): assert recorded["path"] == expected_path +def test_resolve_agent_reference_blocks_absolute_path(): + """Verify resolve_agent_reference raises ValueError for absolute paths.""" + ref_config = AgentRefConfig(config_path="/etc/passwd") + with pytest.raises( + ValueError, + match="Absolute paths are not allowed in AgentRefConfig config_path", + ): + config_agent_utils.resolve_agent_reference( + ref_config, "/workspace/main.yaml" + ) + + +def test_resolve_agent_reference_blocks_path_traversal(): + """Verify resolve_agent_reference raises ValueError for path traversal.""" + ref_config = AgentRefConfig(config_path="../outside.yaml") + with pytest.raises(ValueError, match="Path traversal detected"): + config_agent_utils.resolve_agent_reference( + ref_config, "/workspace/agents/main.yaml" + ) + + +# --- Security tests: module blocklist for YAML agent config code references --- + + +def test_resolve_code_reference_blocks_os_when_enforced(): + """Verify resolve_code_reference blocks os module directly.""" + from google.adk.agents.common_configs import CodeConfig + + with pytest.raises(ValueError, match="Blocked module reference"): + config_agent_utils.resolve_code_reference(CodeConfig(name="os.system")) + + +def test_resolve_fully_qualified_name_blocks_subprocess_when_enforced(): + """Verify resolve_fully_qualified_name blocks subprocess module. + + resolve_fully_qualified_name wraps all exceptions in + ValueError("Invalid fully qualified name: ..."), so we check the wrapper + and verify the __cause__ carries the blocklist message. + """ + with pytest.raises( + ValueError, match="Invalid fully qualified name" + ) as exc_info: + config_agent_utils.resolve_fully_qualified_name("subprocess.Popen") + assert "Blocked module reference" in str(exc_info.value.__cause__) + + +def test_allowed_module_passes_when_enforced(tmp_path: Path): + """Verify that google.adk modules are NOT blocked by the module denylist.""" + # This should NOT raise — google.adk modules must remain allowed + result = config_agent_utils.resolve_fully_qualified_name( + "google.adk.agents.llm_agent.LlmAgent" + ) + assert result is LlmAgent + + +@pytest.mark.parametrize( + "blocked_module", + [ + "os.system", + "posix.system", + "nt.system", + "subprocess.call", + "_posixsubprocess.fork_exec", + "socket.socket", + "_socket.socket", + "builtins.exec", + ], +) +def test_resolve_agent_code_reference_blocks_when_enforced( + blocked_module: str, +): + """Verify _resolve_agent_code_reference blocks dangerous modules.""" + with pytest.raises(ValueError, match="Blocked module reference"): + config_agent_utils._resolve_agent_code_reference(blocked_module) + + +@pytest.mark.parametrize( + "blocked_ref", + [ + "os.system", + "posix.system", + "nt.system", + "subprocess.call", + "_posixsubprocess.fork_exec", + "socket.socket", + "_socket.socket", + "builtins.exec", + "pickle.loads", + ], +) +def test_resolve_tools_blocks_dangerous_modules(blocked_ref: str): + """Verify _resolve_tools blocks dangerous modules for user-defined tools.""" + from google.adk.agents.llm_agent import LlmAgent + from google.adk.tools.tool_configs import ToolConfig + + tool_config = ToolConfig(name=blocked_ref) + with pytest.raises(ValueError, match="Blocked module reference"): + LlmAgent._resolve_tools([tool_config], "/fake/path.yaml") + + +def test_resolve_tools_allows_builtin_adk_tools(): + """Verify _resolve_tools allows ADK built-in tools (no dot in name).""" + from google.adk.agents.llm_agent import LlmAgent + from google.adk.tools.tool_configs import ToolConfig + + # Built-in tools have no dot — they import from google.adk.tools + tool_config = ToolConfig(name="google_search") + # Should NOT raise — this is a safe, hardcoded import path + resolved = LlmAgent._resolve_tools([tool_config], "/fake/path.yaml") + assert len(resolved) == 1 + + +@pytest.mark.parametrize( + "blocked_ref", + [ + "ftplib.FTP", + "smtplib.SMTP", + "xmlrpc.client", + "telnetlib.Telnet", + "poplib.POP3", + "imaplib.IMAP4", + "asyncio.run", + "pathlib.Path", + ], +) +def test_newly_blocked_network_modules_are_rejected(blocked_ref: str): + """Verify newly added network-capable modules are blocked. + + resolve_fully_qualified_name wraps errors, so we check the cause. + """ + with pytest.raises( + ValueError, match="Invalid fully qualified name" + ) as exc_info: + config_agent_utils.resolve_fully_qualified_name(blocked_ref) + assert "Blocked module reference" in str(exc_info.value.__cause__) + + +# Standard library functions that will run whatever code you hand them. The old +# denylist happened to list profile but not cProfile, and missed all the rest. +# One entry per module, since the check only looks at the top-level name. +_EXEC_CAPABLE_STDLIB_REFS = [ + "cProfile.run", + "profile.run", + "timeit.timeit", + "pydoc.pipepager", + "trace.Trace", + "doctest.testmod", + "bdb.Bdb", + "py_compile.compile", +] + +# These are not in sys.stdlib_module_names on every Python we support, so +# _BLOCKED_MODULES is the only thing rejecting them. +_LOAD_BEARING_NON_STDLIB_REFS = [ + "distutils.spawn.spawn", + "test.support.script_helper.spawn_python", + "_testcapi.run_stringflags", + "pipes.quote", + "telnetlib.Telnet", +] + + +@pytest.mark.parametrize("blocked_ref", _EXEC_CAPABLE_STDLIB_REFS) +def test_resolve_code_reference_blocks_exec_capable_stdlib(blocked_ref: str): + """Exec-capable stdlib modules are rejected as code references.""" + with pytest.raises(ValueError, match="Blocked module reference"): + config_agent_utils.resolve_code_reference(CodeConfig(name=blocked_ref)) + + +@pytest.mark.parametrize("blocked_ref", _EXEC_CAPABLE_STDLIB_REFS) +def test_resolve_tools_blocks_exec_capable_stdlib(blocked_ref: str): + """Exec-capable stdlib modules are rejected as user-defined tools. + + This is the path the reported exploit takes: upload an agent YAML whose only + tool is `cProfile.run`, then replay a saved test session, which dispatches a + recorded functionCall straight to the resolved tool. + """ + from google.adk.tools.tool_configs import ToolConfig + + tool_config = ToolConfig(name=blocked_ref) + with pytest.raises(ValueError, match="Blocked module reference"): + LlmAgent._resolve_tools([tool_config], "/fake/path.yaml") + + +@pytest.mark.parametrize( + "blocked_ref", + [ + "json.loads", + "base64.b64decode", + "string.capwords", + "gc.collect", + "operator.attrgetter", + ], +) +def test_harmless_looking_stdlib_modules_are_also_blocked(blocked_ref: str): + """The whole standard library is off-limits, not just the scary parts. + + Blocking all of it is what keeps this closed against ways to run code that + future Python releases add. + """ + with pytest.raises(ValueError, match="Blocked module reference"): + config_agent_utils.resolve_code_reference(CodeConfig(name=blocked_ref)) + + +@pytest.mark.parametrize("blocked_ref", _LOAD_BEARING_NON_STDLIB_REFS) +def test_modules_dropped_from_the_stdlib_are_still_blocked(blocked_ref: str): + """Covers the modules the standard library rule misses. + + They stay importable from a shim or a PyPI backport, so without the explicit + denylist they come back as a way to run code. + """ + with pytest.raises(ValueError, match="Blocked module reference"): + config_agent_utils.resolve_code_reference(CodeConfig(name=blocked_ref)) + + +def test_third_party_module_reference_is_not_blocked(): + """Non-stdlib packages stay resolvable so integrations keep working. + + A compatibility guarantee for integrations like langchain, not a security + assertion: third-party packages are still resolvable by name. + """ + result = config_agent_utils.resolve_fully_qualified_name("pydantic.BaseModel") + assert result is BaseModel + + +def test_denylist_can_be_disabled(): + """Verify _set_enforce_denylist(False) disables module blocking.""" + config_agent_utils._set_enforce_denylist(False) + try: + # os.getcwd is a real, importable reference — should succeed + result = config_agent_utils.resolve_fully_qualified_name("os.getcwd") + assert callable(result) + finally: + config_agent_utils._set_enforce_denylist(True) + + 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" @@ -434,10 +672,10 @@ def test_load_config_from_path_blocks_args_when_enforced(tmp_path): cmd: "rm -rf /" """) - config_agent_utils._set_enforce_denylist(True) + config_agent_utils._set_enforce_yaml_key_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) + config_agent_utils._set_enforce_yaml_key_denylist(False) diff --git a/tests/unittests/cli/test_fast_api.py b/tests/unittests/cli/test_fast_api.py index 14d1e17bfea..64799401a95 100755 --- a/tests/unittests/cli/test_fast_api.py +++ b/tests/unittests/cli/test_fast_api.py @@ -2537,6 +2537,127 @@ def test_builder_save_rejects_nested_args_key(builder_test_client, tmp_path): assert "args" in response.json()["detail"] +def _save_builder_yaml(client, content, *, app_name="app"): + """POST YAML to the builder save endpoint for the given app.""" + return client.post( + "/builder/save?tmp=true", + files=[( + "files", + (f"{app_name}/root_agent.yaml", content, "application/x-yaml"), + )], + ) + + +def test_builder_save_rejects_external_tool_reference( + builder_test_client, tmp_path +): + """A tool naming code outside the app is rejected.""" + response = _save_builder_yaml( + builder_test_client, + b"name: my_agent\ntools:\n - name: os.system\n", + ) + assert response.status_code == 400 + assert "os.system" in response.json()["detail"] + assert not (tmp_path / "app" / "tmp" / "app" / "root_agent.yaml").exists() + + +def test_builder_save_allows_project_tool_reference(builder_test_client): + """A tool under the app being edited is allowed.""" + response = _save_builder_yaml( + builder_test_client, + b"name: my_agent\ntools:\n - name: app.tools.search\n", + ) + assert response.status_code == 200 + + +def test_builder_save_allows_built_in_tool_short_name(builder_test_client): + """An undotted tool name still resolves against ADK's own built-ins.""" + response = _save_builder_yaml( + builder_test_client, + b"name: my_agent\ntools:\n - name: google_search\n", + ) + assert response.status_code == 200 + + +def test_builder_save_allows_built_in_agent_class(builder_test_client): + """A qualified ADK agent class is allowed.""" + response = _save_builder_yaml( + builder_test_client, + b"agent_class: google.adk.agents.LlmAgent\nname: my_agent\n", + ) + assert response.status_code == 200 + + +def test_builder_save_rejects_adk_submodule_reference(builder_test_client): + """An ADK path reaching past the exported built-ins is rejected.""" + response = _save_builder_yaml( + builder_test_client, + b"name: my_agent\ntools:\n" + b" - name: google.adk.tools.bash_tool.BashTool\n", + ) + assert response.status_code == 400 + assert "BashTool" in response.json()["detail"] + + +def test_builder_save_rejects_external_callback_reference(builder_test_client): + """A callback naming code outside the app is rejected.""" + response = _save_builder_yaml( + builder_test_client, + b"name: my_agent\nbefore_agent_callbacks:\n - name: os.system\n", + ) + assert response.status_code == 400 + assert "before_agent_callbacks" in response.json()["detail"] + + +def test_builder_save_rejects_external_sub_agent_code(builder_test_client): + """A sub-agent naming code outside the app is rejected.""" + response = _save_builder_yaml( + builder_test_client, + b"name: my_agent\nsub_agents:\n - code: other_package.agent\n", + ) + assert response.status_code == 400 + assert "other_package.agent" in response.json()["detail"] + + +def test_builder_save_rejects_external_schema_reference(builder_test_client): + """A schema given as a bare string is validated like any other reference.""" + response = _save_builder_yaml( + builder_test_client, + b"name: my_agent\ninput_schema: os.path\n", + ) + assert response.status_code == 400 + assert "input_schema" in response.json()["detail"] + + +def test_builder_save_rejects_reference_when_app_name_shadows_module( + builder_test_client, +): + """An app named after a real module cannot vouch for its own references.""" + response = _save_builder_yaml( + builder_test_client, + b"name: my_agent\ntools:\n - name: os.system\n", + app_name="os", + ) + assert response.status_code == 400 + assert "shadows" in response.json()["detail"] + + +def test_builder_save_covers_every_code_config_field(builder_test_client): + """Every config field holding a CodeConfig is checked on upload.""" + code_config_fields = set() + for agent in (BaseAgent, LlmAgent): + for name, field in agent.config_type.model_fields.items(): + if "CodeConfig" in str(field.annotation): + code_config_fields.add(name) + assert code_config_fields, "expected agent configs to declare CodeConfig" + + for field_name in sorted(code_config_fields): + content = f"name: my_agent\n{field_name}:\n name: os.system\n" + response = _save_builder_yaml(builder_test_client, content.encode()) + assert response.status_code == 400, field_name + assert field_name 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"