fix(artifacts): Port artifact reference scoping and path-segment validation to v1

src/google/adk/artifacts/artifact_util.py

@@ -21,6 +21,8 @@
 
 from google.genai import types
 
+from ..errors import input_validation_error
+
 
 class ParsedArtifactUri(NamedTuple):
   """The result of parsing an artifact URI."""
@@ -32,6 +34,8 @@ class ParsedArtifactUri(NamedTuple):
   version: int
 
 
+_WINDOWS_DRIVE_RE = re.compile(r"[A-Za-z]:")
+
 _SESSION_SCOPED_ARTIFACT_URI_RE = re.compile(
     r"artifact://apps/([^/]+)/users/([^/]+)/sessions/([^/]+)/artifacts/([^/]+)/versions/(\d+)"
 )
@@ -114,3 +118,63 @@ def is_artifact_ref(artifact: types.Part) -> bool:
       and artifact.file_data.file_uri
       and artifact.file_data.file_uri.startswith("artifact://")
   )
+
+
+def validate_artifact_reference_scope(
+    *,
+    app_name: str,
+    user_id: str,
+    session_id: Optional[str],
+    parsed_uri: ParsedArtifactUri,
+) -> None:
+  """Ensures artifact references cannot escape the caller's scope."""
+  if parsed_uri.app_name != app_name or parsed_uri.user_id != user_id:
+    raise input_validation_error.InputValidationError(
+        "Artifact references must stay within the same app and user scope."
+    )
+  if parsed_uri.session_id is not None and parsed_uri.session_id != session_id:
+    raise input_validation_error.InputValidationError(
+        "Session-scoped artifact references must stay within the same"
+        " session scope."
+    )
+
+
+def _is_drive_qualified(value: str) -> bool:
+  """Checks whether a value starts with a Windows drive letter such as ``C:``."""
+  return _WINDOWS_DRIVE_RE.match(value) is not None
+
+
+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 traversal segments, null bytes,
+      is an absolute path / starts with a slash, or is drive-qualified.
+  """
+  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 isinstance(value, str) and (
+      value.startswith("/") or value.startswith("\\")
+  ):
+    raise input_validation_error.InputValidationError(
+        f"{field_name} {value!r} must not be an absolute path or start with a"
+        " slash."
+    )
+  if isinstance(value, str) and _is_drive_qualified(value):
+    raise input_validation_error.InputValidationError(
+        f"{field_name} {value!r} must not be drive-qualified."
+    )
+  if value in (".", "..") or ".." in value.replace("\\", "/").split("/"):
+    raise input_validation_error.InputValidationError(
+        f"{field_name} {value!r} must not contain traversal segments."
+    )

src/google/adk/artifacts/file_artifact_service.py

@@ -33,6 +33,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
@@ -85,14 +86,32 @@ def _to_posix_path(path_value: str) -> PurePosixPath:
   return PurePosixPath(path_value)
 
 
+def _is_rooted_or_drive_qualified(path_value: str) -> bool:
+  """Checks POSIX and Windows rooted or drive-qualified path forms."""
+  # A Windows root covers POSIX absolute paths, UNC and device prefixes alike;
+  # only the drive-relative form (`C:name`) has no root of its own.
+  if artifact_util._is_drive_qualified(path_value):
+    return True
+  return bool(PureWindowsPath(path_value).root)
+
+
+def _has_parent_reference(path_value: str) -> bool:
+  """Checks parent traversal using either platform's separators."""
+  return (
+      ".." in PurePosixPath(path_value).parts
+      or ".." in PureWindowsPath(path_value).parts
+  )
+
+
 def _resolve_scoped_artifact_path(
     scope_root: Path, filename: str
 ) -> tuple[Path, Path]:
   """Returns the absolute artifact directory and its relative path.
 
   The caller is expected to pass the scope root directory (user or session).
-  This helper joins the filename under that root, resolves traversal segments,
-  and guards against paths that escape the scope root.
+  Filenames that are rooted, drive-qualified, or contain a parent reference are
+  rejected outright, including parent references that would resolve back inside
+  the scope root. Whatever remains is joined under the scope root.
 
   Args:
     scope_root: Directory that defines the storage scope.
@@ -103,17 +122,23 @@ def _resolve_scoped_artifact_path(
     to `scope_root`.
 
   Raises:
-    InputValidationError: If `filename` resolves outside of `scope_root`.
+    InputValidationError: If `filename` is rooted, drive-qualified, contains a
+      parent reference, or otherwise resolves outside of `scope_root`.
   """
   stripped = _strip_user_namespace(filename).strip()
-  pure_path = _to_posix_path(stripped)
 
-  scope_root_resolved = scope_root.resolve(strict=False)
-  if pure_path.is_absolute():
+  if _is_rooted_or_drive_qualified(stripped):
     raise InputValidationError(
-        f"Absolute artifact filename {filename!r} is not permitted; "
-        "provide a path relative to the storage scope."
+        f"Rooted or drive-qualified artifact filename {filename!r} is not "
+        "permitted; provide a path relative to the storage scope."
     )
+  if _has_parent_reference(stripped):
+    raise InputValidationError(
+        f"Artifact filename {filename!r} must not contain parent traversal."
+    )
+
+  scope_root_resolved = scope_root.resolve(strict=False)
+  pure_path = _to_posix_path(stripped)
   candidate = scope_root_resolved / Path(pure_path)
 
   candidate = candidate.resolve(strict=False)
@@ -147,7 +172,7 @@ def _validate_path_segment(value: str, field_name: str) -> None:
 
   Raises:
     InputValidationError: If the value contains path separators, traversal
-      segments, or null bytes.
+      segments, null bytes, or is drive-qualified.
   """
   if not value:
     raise InputValidationError(f"{field_name} must not be empty.")
@@ -157,6 +182,10 @@ def _validate_path_segment(value: str, field_name: str) -> None:
     raise InputValidationError(
         f"{field_name} {value!r} must not contain path separators."
     )
+  if artifact_util._is_drive_qualified(value):
+    raise InputValidationError(
+        f"{field_name} {value!r} must not be drive-qualified."
+    )
   if value in (".", "..") or ".." in value.split("/"):
     raise InputValidationError(
         f"{field_name} {value!r} must not contain traversal segments."
@@ -259,7 +288,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."
       )
@@ -524,7 +553,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

@@ -32,6 +32,7 @@
 from google.genai import types
 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
@@ -161,13 +162,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(
@@ -276,6 +280,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
@@ -128,10 +131,19 @@ async def save_artifact(
       artifact_version.mime_type = "text/plain"
     elif artifact.file_data is not None:
       if artifact_util.is_artifact_ref(artifact):
-        if not artifact_util.parse_artifact_uri(artifact.file_data.file_uri):
+        parsed_uri = artifact_util.parse_artifact_uri(
+            artifact.file_data.file_uri
+        )
+        if not parsed_uri:
           raise InputValidationError(
               f"Invalid artifact reference URI: {artifact.file_data.file_uri}"
           )
+        artifact_util.validate_artifact_reference_scope(
+            app_name=app_name,
+            user_id=user_id,
+            session_id=session_id,
+            parsed_uri=parsed_uri,
+        )
         # If it's a valid artifact URI, we store the artifact part as-is.
         # And we don't know the mime type until we load it.
       else:
@@ -180,6 +192,12 @@ async def load_artifact(
             "Invalid artifact reference URI:"
             f" {artifact_data.file_data.file_uri}"
         )
+      artifact_util.validate_artifact_reference_scope(
+          app_name=app_name,
+          user_id=user_id,
+          session_id=session_id,
+          parsed_uri=parsed_uri,
+      )
       return await self.load_artifact(
           app_name=parsed_uri.app_name,
           user_id=parsed_uri.user_id,
@@ -200,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

src/google/adk/errors/input_validation_error.py

@@ -18,7 +18,7 @@
 class InputValidationError(ValueError):
   """Represents an error raised when user input fails validation."""
 
-  def __init__(self, message="Invalid input."):
+  def __init__(self, message: str = "Invalid input.") -> None:
     """Initializes the InputValidationError exception.
 
     Args: