fix: validate local eval path segments
Merge https://github.com/google/adk-python/pull/5993 ## Summary - add shared validation for local eval filesystem path segments - reject empty, traversal, separator, and null-byte identifiers before path construction - cover local eval set and eval result managers with regression tests ## Issue Association No public GitHub issue. This hardens local eval storage path handling for caller-controlled identifiers. ## Testing Plan - env UV_CACHE_DIR=/tmp/uv-cache uv run --extra test pytest tests/unittests/evaluation/test_local_eval_sets_manager.py tests/unittests/evaluation/test_local_eval_set_results_manager.py Co-authored-by: George Weale <gweale@google.com> COPYBARA_INTEGRATE_REVIEW=https://github.com/google/adk-python/pull/5993 from bg0d-droid:bg0d/validate-eval-path-segments 8f3f363b1686190ae5a7ecf1937cea01eaad1aee PiperOrigin-RevId: 938260846
This commit is contained in:
committed by
Copybara-Service
parent
b9625bfd70
commit
7b87f910cd
@@ -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."
|
||||
)
|
||||
@@ -22,6 +22,7 @@ from typing_extensions import override
|
||||
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 @@ class LocalEvalSetResultsManager(EvalSetResultsManager):
|
||||
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 @@ class LocalEvalSetResultsManager(EvalSetResultsManager):
|
||||
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 @@ class LocalEvalSetResultsManager(EvalSetResultsManager):
|
||||
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)
|
||||
|
||||
@@ -33,6 +33,7 @@ from ._eval_sets_manager_utils import delete_eval_case_from_eval_set
|
||||
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 @@ class LocalEvalSetsManager(EvalSetsManager):
|
||||
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 @@ class LocalEvalSetsManager(EvalSetsManager):
|
||||
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,
|
||||
|
||||
@@ -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")
|
||||
@@ -92,6 +92,22 @@ class TestLocalEvalSetResultsManager:
|
||||
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 @@ class TestLocalEvalSetResultsManager:
|
||||
)
|
||||
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
|
||||
|
||||
@@ -395,6 +395,29 @@ class TestLocalEvalSetsManager:
|
||||
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
|
||||
):
|
||||
|
||||
Reference in New Issue
Block a user