fix: reject the reserved segment 'user' as a session_id

Merge #7064

Reject 'user' (and session IDs starting with 'user/') as a session_id in InMemoryArtifactService and GcsArtifactService to prevent collisions with user-scoped artifacts.

Fixes #7063

PiperOrigin-RevId: 981588022

Author: MUHAMMEDHAFEEZ

src/google/adk/artifacts/artifact_util.py

@@ -155,6 +155,30 @@ def _is_drive_qualified(value: str) -> bool:
   return _WINDOWS_DRIVE_RE.match(value) is not None
 
 
+def _validate_session_id_for_flat_storage(session_id: str) -> None:
+  """Validates a session_id used by flat storage artifact backends.
+
+  In addition to the checks in `validate_path_segment`, rejects values whose
+  first path segment is the reserved value "user". Backends that lay out
+  session-scoped and user-scoped artifacts in the same flat namespace
+  (in-memory, GCS) use that exact string as a reserved segment marking
+  user-scoped artifacts, so a session starting with "user" would silently write
+  into -- and read out of -- that reserved namespace instead of its own.
+
+  Args:
+    session_id: The caller-supplied session id.
+
+  Raises:
+    InputValidationError: If `session_id` fails `validate_path_segment`, or has
+      the reserved value "user" as its first path segment.
+  """
+  validate_path_segment(session_id, "session_id")
+  if session_id.replace("\\", "/").split("/")[0] == "user":
+    raise input_validation_error.InputValidationError(
+        "session_id must not be or start with the reserved value 'user'."
+    )
+
+
 def validate_path_segment(value: str, field_name: str) -> None:
   """Rejects values that could alter the constructed path.
 

src/google/adk/artifacts/gcs_artifact_service.py

@@ -256,6 +256,13 @@ def _save_artifact(
   ) -> int:
     from google.cloud import exceptions  # pylint: disable=g-import-not-at-top
 
+    if not self._file_has_user_namespace(filename):
+      if session_id is None:
+        raise InputValidationError(
+            "Session ID must be provided for session-scoped artifacts."
+        )
+      artifact_util._validate_session_id_for_flat_storage(session_id)
+
     artifact = ensure_part(artifact)
     blob_metadata = {k: str(v) for k, v in (custom_metadata or {}).items()}
     if artifact.inline_data and artifact.inline_data.display_name:

src/google/adk/artifacts/in_memory_artifact_service.py

@@ -126,6 +126,12 @@ async def save_artifact(
       session_id: Optional[str] = None,
       custom_metadata: Optional[dict[str, Any]] = None,
   ) -> int:
+    if not self._file_has_user_namespace(filename):
+      if session_id is None:
+        raise InputValidationError(
+            "Session ID must be provided for session-scoped artifacts."
+        )
+      artifact_util._validate_session_id_for_flat_storage(session_id)
     artifact = ensure_part(artifact)
     path = self._artifact_path(app_name, user_id, filename, session_id)
     if path not in self.artifacts:

tests/unittests/artifacts/test_artifact_service.py

@@ -305,6 +305,111 @@ async def test_save_load_delete(service_type, artifact_service_factory):
   )
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "service_type",
+    [
+        ArtifactServiceType.IN_MEMORY,
+        ArtifactServiceType.GCS,
+    ],
+)
+@pytest.mark.parametrize("session_id", ["user", "user/x", "user\\x"])
+async def test_save_artifact_rejects_reserved_user_as_session_id(
+    service_type, session_id, artifact_service_factory
+):
+  """IN_MEMORY and GCS lay session-scoped and user-scoped artifacts out in
+  the same flat namespace, using the literal segment "user" to mark
+  user-scoped ones. A session actually named "user" (or starting with "user/")
+  must be rejected rather than silently colliding with that reserved segment."""
+  artifact_service = artifact_service_factory(service_type)
+
+  with pytest.raises(InputValidationError, match="reserved value 'user'"):
+    await artifact_service.save_artifact(
+        app_name="app0",
+        user_id="user0",
+        session_id=session_id,
+        filename="report.txt",
+        artifact=types.Part(text="hello"),
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("session_id", ["user", "user/x", "user\\x"])
+async def test_file_allows_reserved_user_as_session_id(
+    session_id,
+    artifact_service_factory,
+):
+  """Unlike IN_MEMORY and GCS, FILE lays session-scoped artifacts out under
+  their own `sessions/<id>/` subtree, distinct from the user-scoped
+  `artifacts/` subtree, so a session literally named "user" cannot collide
+  with it and is not rejected."""
+  artifact_service = artifact_service_factory(ArtifactServiceType.FILE)
+
+  await artifact_service.save_artifact(
+      app_name="app0",
+      user_id="user0",
+      session_id=session_id,
+      filename="report.txt",
+      artifact=types.Part(text="hello"),
+  )
+  loaded = await artifact_service.load_artifact(
+      app_name="app0",
+      user_id="user0",
+      session_id=session_id,
+      filename="report.txt",
+  )
+  assert loaded == types.Part(text="hello")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "service_type",
+    [
+        ArtifactServiceType.IN_MEMORY,
+        ArtifactServiceType.GCS,
+    ],
+)
+@pytest.mark.parametrize("session_id", ["user", "user/x", "user\\x"])
+async def test_read_and_delete_paths_allow_reserved_user_as_session_id(
+    service_type, session_id, artifact_service_factory
+):
+  """Reads and deletes must remain permissive for session IDs named "user" or
+  starting with "user/", so existing data already stored under that prefix in a
+  live bucket or memory store remains reachable and removable."""
+  artifact_service = artifact_service_factory(service_type)
+
+  assert (
+      await artifact_service.list_artifact_keys(
+          app_name="app0", user_id="user0", session_id=session_id
+      )
+      == []
+  )
+  assert (
+      await artifact_service.load_artifact(
+          app_name="app0",
+          user_id="user0",
+          session_id=session_id,
+          filename="report.txt",
+      )
+      is None
+  )
+  assert (
+      await artifact_service.list_versions(
+          app_name="app0",
+          user_id="user0",
+          session_id=session_id,
+          filename="report.txt",
+      )
+      == []
+  )
+  await artifact_service.delete_artifact(
+      app_name="app0",
+      user_id="user0",
+      session_id=session_id,
+      filename="report.txt",
+  )
+
+
 @pytest.mark.asyncio
 async def test_in_memory_loads_nested_artifact_reference(
     artifact_service_factory,

tests/unittests/artifacts/test_artifact_util.py

@@ -201,6 +201,32 @@ def test_validate_path_segment_invalid(value, field_name):
     artifact_util.validate_path_segment(value, field_name)
 
 
+@pytest.mark.parametrize(
+    "session_id",
+    ["user", "user/x", "user\\x", "user/has/slash", "user\\has\\backslash"],
+)
+def test_validate_session_id_for_flat_storage_rejects_reserved_user(
+    session_id: str,
+):
+  with pytest.raises(InputValidationError, match="reserved value 'user'"):
+    artifact_util._validate_session_id_for_flat_storage(session_id)
+
+
+@pytest.mark.parametrize(
+    "session_id",
+    ["session1", "users", "username", "group/user", "has/slash"],
+)
+def test_validate_session_id_for_flat_storage_allows_ordinary_values(
+    session_id: str,
+):
+  artifact_util._validate_session_id_for_flat_storage(session_id)
+
+
+def test_validate_session_id_for_flat_storage_still_runs_path_segment_checks():
+  with pytest.raises(InputValidationError, match="must not be empty"):
+    artifact_util._validate_session_id_for_flat_storage("")
+
+
 @pytest.mark.parametrize(
     "caller_session_id, uri_session_id",
     [