fix(sessions): validate session_id and enforce ownership in delete

session_id was interpolated raw into the Vertex AI REST URL, and
delete_session ignored user_id. A frontend-supplied session_id could
therefore (1) traverse paths via ".." / "..?force=true" to sibling
resources, or (2) delete another user's session.

  - Add _validate_session_id() (^[A-Za-z0-9_-]+$) at every interpolation
    site: create_session, get_session, delete_session, append_event.
  - delete_session now verifies user_id, mirroring get_session.

Author: Nicola Pesavento

src/google/adk/cli/browser/assets/config/runtime-config.json

@@ -1,3 +1,3 @@
 {
   "backendUrl": ""
-}
+}

src/google/adk/sessions/vertex_ai_session_service.py

@@ -47,6 +47,18 @@
 _COMPACTION_CUSTOM_METADATA_KEY = '_compaction'
 _USAGE_METADATA_CUSTOM_METADATA_KEY = '_usage_metadata'
 
+_SESSION_ID_PATTERN = re.compile(r'^[A-Za-z0-9_-]+$')
+
+
+def _validate_session_id(session_id: str) -> None:
+  """Rejects session IDs that could escape the URL path segment."""
+  if not isinstance(session_id, str) or not _SESSION_ID_PATTERN.fullmatch(
+      session_id
+  ):
+    raise ValueError(
+        f'Invalid session_id {session_id!r}: must match {_SESSION_ID_PATTERN.pattern}.'
+    )
+
 
 def _quote_filter_literal(value: str) -> str:
   """Quotes filter values so embedded metacharacters stay inside the literal."""
@@ -127,6 +139,7 @@ async def create_session(
 
     config = {'session_state': state} if state else {}
     if session_id:
+      _validate_session_id(session_id)
       config['session_id'] = session_id
     config.update(kwargs)
     async with self._get_api_client() as api_client:
@@ -157,6 +170,7 @@ async def get_session(
       session_id: str,
       config: Optional[GetSessionConfig] = None,
   ) -> Optional[Session]:
+    _validate_session_id(session_id)
     reasoning_engine_id = self._get_reasoning_engine_id(app_name)
     session_resource_name = (
         f'reasoningEngines/{reasoning_engine_id}/sessions/{session_id}'
@@ -256,14 +270,32 @@ async def list_sessions(
   async def delete_session(
       self, *, app_name: str, user_id: str, session_id: str
   ) -> None:
+    _validate_session_id(session_id)
     reasoning_engine_id = self._get_reasoning_engine_id(app_name)
+    session_resource_name = (
+        f'reasoningEngines/{reasoning_engine_id}/sessions/{session_id}'
+    )
 
     async with self._get_api_client() as api_client:
+      # Ownership check: ensure the session belongs to the caller before
+      # deleting. Without this, delete_session ignores user_id entirely and
+      # any caller who knows a session_id can delete it.
+      try:
+        existing = await api_client.agent_engines.sessions.get(
+            name=session_resource_name
+        )
+      except ClientError as e:
+        if e.code == 404:
+          return
+        raise
+      if existing.user_id != user_id:
+        raise ValueError(
+            f'Session {session_id} does not belong to user {user_id}.'
+        )
+
       try:
         await api_client.agent_engines.sessions.delete(
-            name=(
-                f'reasoningEngines/{reasoning_engine_id}/sessions/{session_id}'
-            ),
+            name=session_resource_name,
         )
       except Exception as e:
         logger.error('Error deleting session %s: %s', session_id, e)
@@ -274,6 +306,7 @@ async def append_event(self, session: Session, event: Event) -> Event:
     # Update the in-memory session.
     await super().append_event(session=session, event=event)
 
+    _validate_session_id(session.id)
     reasoning_engine_id = self._get_reasoning_engine_id(session.app_name)
 
     # Build config (Monolithic approach)

tests/unittests/sessions/test_vertex_ai_session_service.py

@@ -725,6 +725,45 @@ async def test_get_and_delete_session():
   assert str(excinfo.value) == '404 Session not found: 1'
 
 
+@pytest.mark.asyncio
+@pytest.mark.usefixtures('mock_get_api_client')
+async def test_delete_session_rejects_other_users_session():
+  """delete_session must not delete a session owned by a different user."""
+  session_service = mock_vertex_ai_session_service()
+
+  # session '1' belongs to 'user'; 'user2' must not be allowed to delete it.
+  with pytest.raises(ValueError) as excinfo:
+    await session_service.delete_session(
+        app_name='123', user_id='user2', session_id='1'
+    )
+  assert 'does not belong to user user2' in str(excinfo.value)
+
+  # Session must still exist.
+  assert (
+      await session_service.get_session(
+          app_name='123', user_id='user', session_id='1'
+      )
+      == MOCK_SESSION
+  )
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures('mock_get_api_client')
+async def test_session_id_path_traversal_rejected():
+  """Session IDs containing path-traversal characters must be rejected."""
+  session_service = mock_vertex_ai_session_service()
+
+  for bad_id in ['..', '../foo', '..?force=true', 'a/b', '']:
+    with pytest.raises(ValueError):
+      await session_service.delete_session(
+          app_name='123', user_id='user', session_id=bad_id
+      )
+    with pytest.raises(ValueError):
+      await session_service.get_session(
+          app_name='123', user_id='user', session_id=bad_id
+      )
+
+
 @pytest.mark.asyncio
 @pytest.mark.usefixtures('mock_get_api_client')
 async def test_get_session_with_page_token():