Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 40 additions & 0 deletions src/google/adk/evaluation/_path_validation.py
Original file line number Diff line number Diff line change
@@ -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."
)
5 changes: 5 additions & 0 deletions src/google/adk/evaluation/gcs_eval_set_results_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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}"

Expand All @@ -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
)
Expand Down
3 changes: 3 additions & 0 deletions src/google/adk/evaluation/gcs_eval_sets_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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}"

Expand Down
5 changes: 5 additions & 0 deletions src/google/adk/evaluation/local_eval_set_results_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
)
Expand All @@ -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(
Expand Down Expand Up @@ -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)
4 changes: 4 additions & 0 deletions src/google/adk/evaluation/local_eval_sets_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 = {
Expand All @@ -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

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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": (
Expand All @@ -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

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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("""\
Expand Down Expand Up @@ -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."""
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
)
52 changes: 52 additions & 0 deletions tests/unittests/evaluation/test__path_validation.py
Original file line number Diff line number Diff line change
@@ -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")
Loading
Loading