diff --git a/src/google/adk/evaluation/_path_validation.py b/src/google/adk/evaluation/_path_validation.py new file mode 100644 index 00000000000..b9bc0db97e2 --- /dev/null +++ b/src/google/adk/evaluation/_path_validation.py @@ -0,0 +1,40 @@ +# 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. + +from __future__ import annotations + + +def validate_path_segment(value: str, field_name: str) -> None: + """Rejects values that could alter a filesystem path. + + Args: + value: The caller-supplied identifier. + field_name: Human-readable field name used in error messages. + + Raises: + ValueError: If the value contains path separators, traversal segments, or + null bytes. + """ + if not value: + raise ValueError(f"{field_name} must not be empty.") + if "\x00" in value: + raise ValueError(f"{field_name} must not contain null bytes.") + if "/" in value or "\\" in value: + raise ValueError( + f"{field_name} {value!r} must not contain path separators." + ) + if value in (".", ".."): + raise ValueError( + f"{field_name} {value!r} must not contain traversal segments." + ) diff --git a/src/google/adk/evaluation/gcs_eval_set_results_manager.py b/src/google/adk/evaluation/gcs_eval_set_results_manager.py index 776bce4d9b9..314acb8554a 100644 --- a/src/google/adk/evaluation/gcs_eval_set_results_manager.py +++ b/src/google/adk/evaluation/gcs_eval_set_results_manager.py @@ -23,6 +23,7 @@ from ..errors.not_found_error import NotFoundError from ._eval_set_results_manager_utils import create_eval_set_result from ._eval_set_results_manager_utils import parse_eval_set_result_json +from ._path_validation import validate_path_segment from .eval_result import EvalCaseResult from .eval_result import EvalSetResult from .eval_set_results_manager import EvalSetResultsManager @@ -54,11 +55,13 @@ def __init__(self, bucket_name: str, **kwargs): ) def _get_eval_history_dir(self, app_name: str) -> str: + validate_path_segment(app_name, "app_name") return f"{app_name}/{_EVAL_HISTORY_DIR}" def _get_eval_set_result_blob_name( self, app_name: str, eval_set_result_id: str ) -> str: + validate_path_segment(eval_set_result_id, "eval_set_result_id") eval_history_dir = self._get_eval_history_dir(app_name) return f"{eval_history_dir}/{eval_set_result_id}{_EVAL_SET_RESULT_FILE_EXTENSION}" @@ -80,6 +83,8 @@ def save_eval_set_result( eval_case_results: list[EvalCaseResult], ) -> None: """Creates and saves a new EvalSetResult given eval_case_results.""" + validate_path_segment(app_name, "app_name") + validate_path_segment(eval_set_id, "eval_set_id") eval_set_result = create_eval_set_result( app_name, eval_set_id, eval_case_results ) diff --git a/src/google/adk/evaluation/gcs_eval_sets_manager.py b/src/google/adk/evaluation/gcs_eval_sets_manager.py index edf501d4c6c..6057cffbc2c 100644 --- a/src/google/adk/evaluation/gcs_eval_sets_manager.py +++ b/src/google/adk/evaluation/gcs_eval_sets_manager.py @@ -29,6 +29,7 @@ from ._eval_sets_manager_utils import get_eval_case_from_eval_set from ._eval_sets_manager_utils import get_eval_set_from_app_and_id from ._eval_sets_manager_utils import update_eval_case_in_eval_set +from ._path_validation import validate_path_segment from .eval_case import EvalCase from .eval_set import EvalSet from .eval_sets_manager import EvalSetsManager @@ -60,9 +61,11 @@ def __init__(self, bucket_name: str, **kwargs): ) def _get_eval_sets_dir(self, app_name: str) -> str: + validate_path_segment(app_name, "app_name") return f"{app_name}/{_EVAL_SETS_DIR}" def _get_eval_set_blob_name(self, app_name: str, eval_set_id: str) -> str: + validate_path_segment(eval_set_id, "eval_set_id") eval_sets_dir = self._get_eval_sets_dir(app_name) return f"{eval_sets_dir}/{eval_set_id}{_EVAL_SET_FILE_EXTENSION}" diff --git a/src/google/adk/evaluation/local_eval_set_results_manager.py b/src/google/adk/evaluation/local_eval_set_results_manager.py index c6da638abe2..dabc0b38b15 100644 --- a/src/google/adk/evaluation/local_eval_set_results_manager.py +++ b/src/google/adk/evaluation/local_eval_set_results_manager.py @@ -22,6 +22,7 @@ from ..errors.not_found_error import NotFoundError from ._eval_set_results_manager_utils import create_eval_set_result from ._eval_set_results_manager_utils import parse_eval_set_result_json +from ._path_validation import validate_path_segment from .eval_result import EvalCaseResult from .eval_result import EvalSetResult from .eval_set_results_manager import EvalSetResultsManager @@ -46,6 +47,8 @@ def save_eval_set_result( eval_case_results: list[EvalCaseResult], ) -> None: """Creates and saves a new EvalSetResult given eval_case_results.""" + validate_path_segment(app_name, "app_name") + validate_path_segment(eval_set_id, "eval_set_id") eval_set_result = create_eval_set_result( app_name, eval_set_id, eval_case_results ) @@ -67,6 +70,7 @@ def get_eval_set_result( self, app_name: str, eval_set_result_id: str ) -> EvalSetResult: """Returns an EvalSetResult identified by app_name and eval_set_result_id.""" + validate_path_segment(eval_set_result_id, "eval_set_result_id") # Load the eval set result file data. maybe_eval_result_file_path = ( os.path.join( @@ -97,4 +101,5 @@ def list_eval_set_results(self, app_name: str) -> list[str]: return eval_result_files def _get_eval_history_dir(self, app_name: str) -> str: + validate_path_segment(app_name, "app_name") return os.path.join(self._agents_dir, app_name, _ADK_EVAL_HISTORY_DIR) diff --git a/src/google/adk/evaluation/local_eval_sets_manager.py b/src/google/adk/evaluation/local_eval_sets_manager.py index 8d2290b911e..75ba9973d65 100644 --- a/src/google/adk/evaluation/local_eval_sets_manager.py +++ b/src/google/adk/evaluation/local_eval_sets_manager.py @@ -33,6 +33,7 @@ from ._eval_sets_manager_utils import get_eval_case_from_eval_set from ._eval_sets_manager_utils import get_eval_set_from_app_and_id from ._eval_sets_manager_utils import update_eval_case_in_eval_set +from ._path_validation import validate_path_segment from .eval_case import EvalCase from .eval_case import IntermediateData from .eval_case import Invocation @@ -247,6 +248,7 @@ def list_eval_sets(self, app_name: str) -> list[str]: Raises: NotFoundError: If the eval directory for the app is not found. """ + validate_path_segment(app_name, "app_name") eval_set_file_path = os.path.join(self._agents_dir, app_name) eval_sets = [] try: @@ -310,6 +312,8 @@ def delete_eval_case( self._save_eval_set(app_name, eval_set_id, updated_eval_set) def _get_eval_set_file_path(self, app_name: str, eval_set_id: str) -> str: + validate_path_segment(app_name, "app_name") + validate_path_segment(eval_set_id, "eval_set_id") return os.path.join( self._agents_dir, app_name, diff --git a/src/google/adk/evaluation/simulation/llm_backed_user_simulator_prompts.py b/src/google/adk/evaluation/simulation/llm_backed_user_simulator_prompts.py index 8873c697c79..fc088dccd96 100644 --- a/src/google/adk/evaluation/simulation/llm_backed_user_simulator_prompts.py +++ b/src/google/adk/evaluation/simulation/llm_backed_user_simulator_prompts.py @@ -185,7 +185,6 @@ def get_llm_backed_user_simulator_prompt( """Formats the prompt for the llm-backed user simulator""" from jinja2 import DictLoader from jinja2 import pass_context - from jinja2 import Template from jinja2.sandbox import SandboxedEnvironment templates = { @@ -200,7 +199,7 @@ def get_llm_backed_user_simulator_prompt( def _render_string_filter(context, template_string): if not template_string: return "" - return Template(template_string).render(context) + return template_env.from_string(template_string).render(context.get_all()) template_env.filters["render_string_filter"] = _render_string_filter diff --git a/src/google/adk/evaluation/simulation/per_turn_user_simulator_quality_prompts.py b/src/google/adk/evaluation/simulation/per_turn_user_simulator_quality_prompts.py index b9fb7a3ab6c..1862272cf3b 100644 --- a/src/google/adk/evaluation/simulation/per_turn_user_simulator_quality_prompts.py +++ b/src/google/adk/evaluation/simulation/per_turn_user_simulator_quality_prompts.py @@ -221,9 +221,8 @@ def get_per_turn_user_simulator_quality_prompt( ): """Formats the prompt for the per turn user simulator evaluator""" from jinja2 import DictLoader - from jinja2 import Environment from jinja2 import pass_context - from jinja2 import Template + from jinja2.sandbox import SandboxedEnvironment templates = { "verifier_instructions": ( @@ -232,13 +231,13 @@ def get_per_turn_user_simulator_quality_prompt( ) ), } - template_env = Environment(loader=DictLoader(templates)) + template_env = SandboxedEnvironment(loader=DictLoader(templates)) @pass_context def _render_string_filter(context, template_string): if not template_string: return "" - return Template(template_string).render(context) + return template_env.from_string(template_string).render(context.get_all()) template_env.filters["render_string_filter"] = _render_string_filter diff --git a/tests/unittests/evaluation/simulation/test_llm_backed_user_simulator_prompts.py b/tests/unittests/evaluation/simulation/test_llm_backed_user_simulator_prompts.py index b150304baab..679c54a5558 100644 --- a/tests/unittests/evaluation/simulation/test_llm_backed_user_simulator_prompts.py +++ b/tests/unittests/evaluation/simulation/test_llm_backed_user_simulator_prompts.py @@ -21,6 +21,7 @@ from google.adk.evaluation.simulation.llm_backed_user_simulator_prompts import is_valid_user_simulator_template from google.adk.evaluation.simulation.user_simulator_personas import UserBehavior from google.adk.evaluation.simulation.user_simulator_personas import UserPersona +from jinja2.exceptions import SecurityError import pytest _MOCK_DEFAULT_TEMPLATE = textwrap.dedent("""\ @@ -208,6 +209,57 @@ def test_get_llm_backed_user_simulator_prompt_with_persona(self, mocker): test stop""").strip() assert prompt == expected_prompt + def test_get_llm_backed_user_simulator_prompt_renders_persona_templates_in_sandbox( + self, + ): + user_persona = UserPersona( + id="test_persona", + description="Test persona description", + behaviors=[ + UserBehavior( + name="Behavior {{ stop_signal }}", + description="Description {{ stop_signal }}", + behavior_instructions=["instruction {{ stop_signal }}"], + violation_rubrics=["rubric 1"], + ) + ], + ) + + prompt = get_llm_backed_user_simulator_prompt( + conversation_plan="test plan", + conversation_history="test history", + stop_signal="test stop", + user_persona=user_persona, + ) + + assert "## Behavior test stop" in prompt + assert "Description test stop" in prompt + assert " * instruction test stop" in prompt + + def test_get_llm_backed_user_simulator_prompt_blocks_unsafe_persona_templates( + self, + ): + user_persona = UserPersona( + id="test_persona", + description="Test persona description", + behaviors=[ + UserBehavior( + name="{{ ''.__class__.__mro__ }}", + description="Test behavior description", + behavior_instructions=["instruction 1"], + violation_rubrics=["rubric 1"], + ) + ], + ) + + with pytest.raises(SecurityError): + get_llm_backed_user_simulator_prompt( + conversation_plan="test plan", + conversation_history="test history", + stop_signal="test stop", + user_persona=user_persona, + ) + class TestIsValidUserSimulatorTemplate: """Test cases for is_valid_user_simulator_template.""" diff --git a/tests/unittests/evaluation/simulation/test_per_turn_user_simulation_quality_prompts.py b/tests/unittests/evaluation/simulation/test_per_turn_user_simulation_quality_prompts.py index a1e7190354f..374ab5deb37 100644 --- a/tests/unittests/evaluation/simulation/test_per_turn_user_simulation_quality_prompts.py +++ b/tests/unittests/evaluation/simulation/test_per_turn_user_simulation_quality_prompts.py @@ -20,6 +20,8 @@ from google.adk.evaluation.simulation.per_turn_user_simulator_quality_prompts import get_per_turn_user_simulator_quality_prompt from google.adk.evaluation.simulation.user_simulator_personas import UserBehavior from google.adk.evaluation.simulation.user_simulator_personas import UserPersona +from jinja2.exceptions import SecurityError +import pytest _MOCK_DEFAULT_TEMPLATE = textwrap.dedent("""\ Default template @@ -182,3 +184,56 @@ def test_get_per_turn_user_simulator_quality_prompt_with_persona( # Stop signal stop""").strip() assert prompt == expected_prompt + + def test_get_per_turn_user_simulator_quality_prompt_renders_persona_templates_in_sandbox( + self, + ): + persona = UserPersona( + id="test_persona", + description="Test persona description.", + behaviors=[ + UserBehavior( + name="criteria {{ stop_signal }}", + description="Test behavior {{ stop_signal }}.", + behavior_instructions=["instruction1"], + violation_rubrics=["violation {{ stop_signal }}"], + ) + ], + ) + + prompt = get_per_turn_user_simulator_quality_prompt( + conversation_plan="plan", + conversation_history="history", + generated_user_response="response", + stop_signal="stop", + user_persona=persona, + ) + + assert "## Criteria: criteria stop" in prompt + assert "Test behavior stop." in prompt + assert " * violation stop" in prompt + + def test_get_per_turn_user_simulator_quality_prompt_blocks_unsafe_persona_templates( + self, + ): + persona = UserPersona( + id="test_persona", + description="Test persona description.", + behaviors=[ + UserBehavior( + name="{{ ''.__class__.__mro__ }}", + description="Test behavior description.", + behavior_instructions=["instruction1"], + violation_rubrics=["violation1"], + ) + ], + ) + + with pytest.raises(SecurityError): + get_per_turn_user_simulator_quality_prompt( + conversation_plan="plan", + conversation_history="history", + generated_user_response="response", + stop_signal="stop", + user_persona=persona, + ) diff --git a/tests/unittests/evaluation/test__path_validation.py b/tests/unittests/evaluation/test__path_validation.py new file mode 100644 index 00000000000..4db2d3a4978 --- /dev/null +++ b/tests/unittests/evaluation/test__path_validation.py @@ -0,0 +1,52 @@ +# 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. + +from __future__ import annotations + +from google.adk.evaluation._path_validation import validate_path_segment +import pytest + + +@pytest.mark.parametrize( + "value", ["eval_set_1", "my-app", "App Name 1", "résumé", "a.b.c"] +) +def test_validate_path_segment_accepts_valid_value(value): + validate_path_segment(value, "field") + + +def test_validate_path_segment_rejects_empty(): + with pytest.raises(ValueError, match="must not be empty"): + validate_path_segment("", "field") + + +def test_validate_path_segment_rejects_null_byte(): + with pytest.raises(ValueError, match="must not contain null bytes"): + validate_path_segment("foo\x00bar", "field") + + +@pytest.mark.parametrize("value", ["foo/bar", "foo\\bar", "/", "\\"]) +def test_validate_path_segment_rejects_path_separators(value): + with pytest.raises(ValueError, match="must not contain path separators"): + validate_path_segment(value, "field") + + +@pytest.mark.parametrize("value", [".", ".."]) +def test_validate_path_segment_rejects_traversal_segments(value): + with pytest.raises(ValueError, match="must not contain traversal segments"): + validate_path_segment(value, "field") + + +def test_validate_path_segment_includes_field_name_in_error(): + with pytest.raises(ValueError, match="eval_set_id"): + validate_path_segment("", "eval_set_id") diff --git a/tests/unittests/evaluation/test_gcs_eval_set_results_manager.py b/tests/unittests/evaluation/test_gcs_eval_set_results_manager.py index 0b165333955..6016e82e462 100644 --- a/tests/unittests/evaluation/test_gcs_eval_set_results_manager.py +++ b/tests/unittests/evaluation/test_gcs_eval_set_results_manager.py @@ -218,3 +218,41 @@ def test_list_eval_set_results_empty(self, gcs_eval_set_results_manager): gcs_eval_set_results_manager.list_eval_set_results(app_name) ) assert retrieved_eval_set_result_ids == [] + + @pytest.mark.parametrize("app_name", ["", ".", "..", "foo/bar", "foo\\bar"]) + def test_save_eval_set_result_rejects_invalid_app_name( + self, gcs_eval_set_results_manager, app_name + ): + with pytest.raises(ValueError): + gcs_eval_set_results_manager.save_eval_set_result( + app_name, "test_eval_set", _get_test_eval_case_results() + ) + + @pytest.mark.parametrize( + "eval_set_id", ["", ".", "..", "foo/bar", "foo\\bar"] + ) + def test_save_eval_set_result_rejects_invalid_eval_set_id( + self, gcs_eval_set_results_manager, eval_set_id + ): + with pytest.raises(ValueError): + gcs_eval_set_results_manager.save_eval_set_result( + "test_app", eval_set_id, _get_test_eval_case_results() + ) + + @pytest.mark.parametrize("app_name", ["", ".", "..", "foo/bar", "foo\\bar"]) + def test_get_eval_set_result_rejects_invalid_app_name( + self, gcs_eval_set_results_manager, app_name + ): + with pytest.raises(ValueError): + gcs_eval_set_results_manager.get_eval_set_result(app_name, "some_id") + + @pytest.mark.parametrize( + "eval_set_result_id", ["", ".", "..", "foo/bar", "foo\\bar"] + ) + def test_get_eval_set_result_rejects_invalid_eval_set_result_id( + self, gcs_eval_set_results_manager, eval_set_result_id + ): + with pytest.raises(ValueError): + gcs_eval_set_results_manager.get_eval_set_result( + "test_app", eval_set_result_id + ) diff --git a/tests/unittests/evaluation/test_gcs_eval_sets_manager.py b/tests/unittests/evaluation/test_gcs_eval_sets_manager.py index e396cf371b4..1fb7f037246 100644 --- a/tests/unittests/evaluation/test_gcs_eval_sets_manager.py +++ b/tests/unittests/evaluation/test_gcs_eval_sets_manager.py @@ -419,3 +419,26 @@ def test_gcs_eval_sets_manager_delete_eval_case_eval_case_not_found( app_name, eval_set_id, eval_case_id ) mock_write_eval_set_to_blob.assert_not_called() + + @pytest.mark.parametrize("app_name", ["", ".", "..", "foo/bar", "foo\\bar"]) + def test_gcs_eval_sets_manager_create_eval_set_rejects_invalid_app_name( + self, gcs_eval_sets_manager, app_name + ): + with pytest.raises(ValueError): + gcs_eval_sets_manager.create_eval_set(app_name, "test_eval_set") + + @pytest.mark.parametrize("app_name", ["", ".", "..", "foo/bar", "foo\\bar"]) + def test_gcs_eval_sets_manager_list_eval_sets_rejects_invalid_app_name( + self, gcs_eval_sets_manager, app_name + ): + with pytest.raises(ValueError): + gcs_eval_sets_manager.list_eval_sets(app_name) + + @pytest.mark.parametrize( + "eval_set_id", ["", ".", "..", "foo/bar", "foo\\bar"] + ) + def test_gcs_eval_sets_manager_get_eval_set_rejects_invalid_eval_set_id( + self, gcs_eval_sets_manager, eval_set_id + ): + with pytest.raises(ValueError): + gcs_eval_sets_manager.get_eval_set("test_app", eval_set_id) diff --git a/tests/unittests/evaluation/test_local_eval_set_results_manager.py b/tests/unittests/evaluation/test_local_eval_set_results_manager.py index 4647392628a..01e08f1f41d 100644 --- a/tests/unittests/evaluation/test_local_eval_set_results_manager.py +++ b/tests/unittests/evaluation/test_local_eval_set_results_manager.py @@ -92,6 +92,22 @@ def test_save_eval_set_result(self, mocker): expected_eval_set_result_data = self.eval_set_result.model_dump(mode="json") assert expected_eval_set_result_data == actual_eval_set_result_data + @pytest.mark.parametrize("app_name", ["", ".", "..", "foo/bar", "foo\\bar"]) + def test_save_eval_set_result_rejects_invalid_app_name(self, app_name): + with pytest.raises(ValueError): + self.manager.save_eval_set_result( + app_name, self.eval_set_id, self.eval_case_results + ) + + @pytest.mark.parametrize( + "eval_set_id", ["", ".", "..", "foo/bar", "foo\\bar"] + ) + def test_save_eval_set_result_rejects_invalid_eval_set_id(self, eval_set_id): + with pytest.raises(ValueError): + self.manager.save_eval_set_result( + self.app_name, eval_set_id, self.eval_case_results + ) + def test_get_eval_set_result(self, mocker): mock_time = mocker.patch("time.time") mock_time.return_value = self.timestamp @@ -103,6 +119,20 @@ def test_get_eval_set_result(self, mocker): ) assert retrieved_result == self.eval_set_result + @pytest.mark.parametrize("app_name", ["", ".", "..", "foo/bar", "foo\\bar"]) + def test_get_eval_set_result_rejects_invalid_app_name(self, app_name): + with pytest.raises(ValueError): + self.manager.get_eval_set_result(app_name, self.eval_set_result_name) + + @pytest.mark.parametrize( + "eval_set_result_id", ["", ".", "..", "foo/bar", "foo\\bar"] + ) + def test_get_eval_set_result_rejects_invalid_eval_set_result_id( + self, eval_set_result_id + ): + with pytest.raises(ValueError): + self.manager.get_eval_set_result(self.app_name, eval_set_result_id) + def test_get_eval_set_result_double_encoded_legacy(self): eval_history_dir = os.path.join( self.agents_dir, self.app_name, _ADK_EVAL_HISTORY_DIR diff --git a/tests/unittests/evaluation/test_local_eval_sets_manager.py b/tests/unittests/evaluation/test_local_eval_sets_manager.py index 3450fb93385..8632a2a7862 100644 --- a/tests/unittests/evaluation/test_local_eval_sets_manager.py +++ b/tests/unittests/evaluation/test_local_eval_sets_manager.py @@ -395,6 +395,29 @@ def test_local_eval_sets_manager_create_eval_set_invalid_id( with pytest.raises(ValueError, match="Invalid Eval Set ID"): local_eval_sets_manager.create_eval_set(app_name, eval_set_id) + @pytest.mark.parametrize("app_name", ["", ".", "..", "foo/bar", "foo\\bar"]) + def test_local_eval_sets_manager_create_eval_set_rejects_invalid_app_name( + self, local_eval_sets_manager, app_name + ): + with pytest.raises(ValueError): + local_eval_sets_manager.create_eval_set(app_name, "test_eval_set") + + @pytest.mark.parametrize("app_name", ["", ".", "..", "foo/bar", "foo\\bar"]) + def test_local_eval_sets_manager_list_eval_sets_rejects_invalid_app_name( + self, local_eval_sets_manager, app_name + ): + with pytest.raises(ValueError): + local_eval_sets_manager.list_eval_sets(app_name) + + @pytest.mark.parametrize( + "eval_set_id", ["", ".", "..", "foo/bar", "foo\\bar"] + ) + def test_local_eval_sets_manager_get_eval_set_rejects_invalid_eval_set_id( + self, local_eval_sets_manager, eval_set_id + ): + with pytest.raises(ValueError): + local_eval_sets_manager.get_eval_set("test_app", eval_set_id) + def test_local_eval_sets_manager_create_eval_set_already_exists( self, local_eval_sets_manager, mocker ):