fix: Validate path segments in GcsArtifactService and InMemoryArtifactService to prevent cross-user artifact access

InMemoryArtifactService and GcsArtifactService currently construct GCS/In-Memory
paths by directly interpolating user-supplied identifiers (user_id, app_name,
and session_id) without validation. A malicious identifier containing path
separators or traversal segments (e.g., ../other-user) could escape the
intended scope and allow cross-user artifact access.

This patch brings GcsArtifactService, InMemoryArtifactService, and FileArtifactService
to parity and unifies their validation logic:

- Refactored all path segment validation logic into a shared helper in `artifact_util.py`.
- Validate `user_id`, `app_name`, and `session_id` in `GcsArtifactService`, `InMemoryArtifactService`, and `FileArtifactService` using the shared helper.
- Consolidated path traversal tests into a parameterized suite in `test_artifact_service.py` and `test_artifact_util.py`.

Closes #6115
Closes #6116

Co-authored-by: Jason Zhang <jasoncz@google.com>
PiperOrigin-RevId: 943976800

Author: 2 people

src/google/adk/artifacts/artifact_util.py

@@ -134,3 +134,32 @@ def validate_artifact_reference_scope(
         "Session-scoped artifact references must stay within the same"
         " session scope."
     )
+
+
+def validate_path_segment(value: str, field_name: str) -> None:
+  """Rejects values that could alter the constructed path.
+
+  Args:
+    value: The caller-supplied identifier (e.g. user_id or session_id).
+    field_name: Human-readable name used in the error message.
+
+  Raises:
+    InputValidationError: If the value contains path separators, traversal
+      segments, or null bytes.
+  """
+  if not value:
+    raise input_validation_error.InputValidationError(
+        f"{field_name} must not be empty."
+    )
+  if "\x00" in value:
+    raise input_validation_error.InputValidationError(
+        f"{field_name} must not contain null bytes."
+    )
+  if "/" in value or "\\" in value:
+    raise input_validation_error.InputValidationError(
+        f"{field_name} {value!r} must not contain path separators."
+    )
+  if value in (".", "..") or ".." in value.split("/"):
+    raise input_validation_error.InputValidationError(
+        f"{field_name} {value!r} must not contain traversal segments."
+    )

src/google/adk/artifacts/file_artifact_service.py

@@ -34,6 +34,7 @@
 from pydantic import ValidationError
 from typing_extensions import override
 
+from . import artifact_util
 from ..errors.input_validation_error import InputValidationError
 from .base_artifact_service import ArtifactVersion
 from .base_artifact_service import BaseArtifactService
@@ -142,39 +143,14 @@ def _is_user_scoped(session_id: Optional[str], filename: str) -> bool:
   return session_id is None or _file_has_user_namespace(filename)
 
 
-def _validate_path_segment(value: str, field_name: str) -> None:
-  """Rejects values that could alter the constructed filesystem path.
-
-  Args:
-    value: The caller-supplied identifier (e.g. user_id or session_id).
-    field_name: Human-readable name used in the error message.
-
-  Raises:
-    InputValidationError: If the value contains path separators, traversal
-      segments, or null bytes.
-  """
-  if not value:
-    raise InputValidationError(f"{field_name} must not be empty.")
-  if "\x00" in value:
-    raise InputValidationError(f"{field_name} must not contain null bytes.")
-  if "/" in value or "\\" in value:
-    raise InputValidationError(
-        f"{field_name} {value!r} must not contain path separators."
-    )
-  if value in (".", "..") or ".." in value.split("/"):
-    raise InputValidationError(
-        f"{field_name} {value!r} must not contain traversal segments."
-    )
-
-
 def _user_artifacts_dir(base_root: Path) -> Path:
   """Returns the path that stores user-scoped artifacts."""
   return base_root / "artifacts"
 
 
 def _session_artifacts_dir(base_root: Path, session_id: str) -> Path:
   """Returns the path that stores session-scoped artifacts."""
-  _validate_path_segment(session_id, "session_id")
+  artifact_util.validate_path_segment(session_id, "session_id")
   return base_root / "sessions" / session_id / "artifacts"
 
 
@@ -256,7 +232,7 @@ def __init__(self, root_dir: Path | str):
 
   def _base_root(self, user_id: str, /) -> Path:
     """Returns the artifacts root directory for a user."""
-    _validate_path_segment(user_id, "user_id")
+    artifact_util.validate_path_segment(user_id, "user_id")
     return self.root_dir / "users" / user_id
 
   def _scope_root(
@@ -269,7 +245,7 @@ def _scope_root(
     base = self._base_root(user_id)
     if _is_user_scoped(session_id, filename):
       return _user_artifacts_dir(base)
-    if not session_id:
+    if session_id is None:
       raise InputValidationError(
           "Session ID must be provided for session-scoped artifacts."
       )
@@ -543,7 +519,7 @@ def _list_artifact_keys_sync(
 
     base_root = self._base_root(user_id)
 
-    if session_id:
+    if session_id is not None:
       session_root = _session_artifacts_dir(base_root, session_id)
       for artifact_dir in _iter_artifact_dirs(session_root):
         metadata = self._latest_metadata(artifact_dir)

src/google/adk/artifacts/gcs_artifact_service.py

@@ -167,13 +167,16 @@ def _get_blob_prefix(
       session_id: Optional[str] = None,
   ) -> str:
     """Constructs the blob name prefix in GCS for a given artifact."""
+    artifact_util.validate_path_segment(app_name, "app_name")
+    artifact_util.validate_path_segment(user_id, "user_id")
     if self._file_has_user_namespace(filename):
       return f"{app_name}/{user_id}/user/{filename}"
 
     if session_id is None:
       raise InputValidationError(
           "Session ID must be provided for session-scoped artifacts."
       )
+    artifact_util.validate_path_segment(session_id, "session_id")
     return f"{app_name}/{user_id}/{session_id}/{filename}"
 
   def _get_blob_name(
@@ -368,6 +371,10 @@ def _load_artifact(
   def _list_artifact_keys(
       self, app_name: str, user_id: str, session_id: Optional[str]
   ) -> list[str]:
+    artifact_util.validate_path_segment(app_name, "app_name")
+    artifact_util.validate_path_segment(user_id, "user_id")
+    if session_id is not None:
+      artifact_util.validate_path_segment(session_id, "session_id")
     filenames = set()
 
     if session_id:

src/google/adk/artifacts/in_memory_artifact_service.py

@@ -85,13 +85,16 @@ def _artifact_path(
     Returns:
         The constructed artifact path.
     """
+    artifact_util.validate_path_segment(app_name, "app_name")
+    artifact_util.validate_path_segment(user_id, "user_id")
     if self._file_has_user_namespace(filename):
       return f"{app_name}/{user_id}/user/{filename}"
 
     if session_id is None:
       raise InputValidationError(
           "Session ID must be provided for session-scoped artifacts."
       )
+    artifact_util.validate_path_segment(session_id, "session_id")
     return f"{app_name}/{user_id}/{session_id}/{filename}"
 
   @override
@@ -215,6 +218,10 @@ async def load_artifact(
   async def list_artifact_keys(
       self, *, app_name: str, user_id: str, session_id: Optional[str] = None
   ) -> list[str]:
+    artifact_util.validate_path_segment(app_name, "app_name")
+    artifact_util.validate_path_segment(user_id, "user_id")
+    if session_id is not None:
+      artifact_util.validate_path_segment(session_id, "session_id")
     usernamespace_prefix = f"{app_name}/{user_id}/user/"
     session_prefix = (
         f"{app_name}/{user_id}/{session_id}/" if session_id else None