From d9961ab33eea9afd6ec92331a7edee7d69c6bb5b Mon Sep 17 00:00:00 2001 From: David Sarkisyan Date: Thu, 15 Oct 2026 15:26:37 -0400 Subject: [PATCH 1/4] Use versioned Redis keys for session identity --- src/google/adk/integrations/redis/README.md | 8 +- .../redis/_redis_session_service.py | 134 +++++-- .../integrations/redis/_fake_redis.py | 28 +- .../redis/test_redis_key_collision_probe.py | 92 +++++ .../redis/test_redis_session_service.py | 340 +++++++++++++++++- 5 files changed, 551 insertions(+), 51 deletions(-) create mode 100644 tests/unittests/integrations/redis/test_redis_key_collision_probe.py diff --git a/src/google/adk/integrations/redis/README.md b/src/google/adk/integrations/redis/README.md index 83e36e4626c..12bf7c14118 100644 --- a/src/google/adk/integrations/redis/README.md +++ b/src/google/adk/integrations/redis/README.md @@ -108,9 +108,11 @@ session_service = RedisSessionService(redis_client=client) | Key Pattern | Data | Description | | :--- | :--- | :--- | -| `{key_prefix}{app_name}:{user_id}:{session_id}` | JSON (`Session`) | Contains session ID, state, events list, and last update timestamp. Expired based on `ttl_seconds`. | -| `{key_prefix}user_state:{app_name}:{user_id}` | JSON (`dict`) | Persisted user-scoped state (keys prefixed with `user:`). Shared across sessions for the same user. Expired based on `ttl_seconds`. | -| `{key_prefix}app_state:{app_name}` | JSON (`dict`) | Persisted app-scoped state (keys prefixed with `app:`). Shared across all sessions and users for the app. Expired based on `ttl_seconds`. | +| `{key_prefix}v2:s2_{b64url(app_name)}.{b64url(user_id)}.{b64url(session_id)}` | JSON (`Session`) | Contains session ID, state, events list, and last update timestamp. Expired based on `ttl_seconds`. | +| `{key_prefix}v2:u2_{b64url(app_name)}.{b64url(user_id)}` | JSON (`dict`) | Persisted user-scoped state (keys prefixed with `user:`). Shared across sessions for the same user. Expired based on `ttl_seconds`. | +| `{key_prefix}v2:a2_{b64url(app_name)}` | JSON (`dict`) | Persisted app-scoped state (keys prefixed with `app:`). Shared across all sessions and users for the app. Expired based on `ttl_seconds`. | + +`b64url(value)` is the unpadded URL-safe Base64 encoding of the UTF-8 value. With the default `key_prefix`, the new namespace is `adk:session:v2:`. These versioned keys are not compatible with keys written by earlier versions. This service does not read, migrate, or delete legacy entries; existing entries remain untouched and are not visible to this version. Plan a coordinated migration or cutover before upgrading. Legacy keys can expire under their configured TTL; if expiry is disabled, inventory and back up relevant entries before any manual cleanup, excluding the new `{key_prefix}v2:` keys. Do not flush a Redis instance shared with other applications. When events append state deltas: - `user:` updates are synchronized to the user state key. diff --git a/src/google/adk/integrations/redis/_redis_session_service.py b/src/google/adk/integrations/redis/_redis_session_service.py index 0bbcc069f5c..55521d7a423 100644 --- a/src/google/adk/integrations/redis/_redis_session_service.py +++ b/src/google/adk/integrations/redis/_redis_session_service.py @@ -22,6 +22,7 @@ import logging import re import time +from base64 import urlsafe_b64encode from typing import Any from typing import cast from typing import Optional @@ -46,6 +47,11 @@ logger = logging.getLogger("google_adk." + __name__) +def _encode_key_component(value: str) -> str: + """Encodes an identity component as unpadded URL-safe Base64.""" + return urlsafe_b64encode(value.encode("utf-8")).decode("ascii").rstrip("=") + + def _escape_glob(value: str) -> str: """Escapes Redis glob metacharacters so that `value` matches only itself.""" return re.sub(r"([\\*?\[\]])", r"\\\1", value) @@ -100,15 +106,32 @@ def _get_redis(self) -> Any: def _session_key(self, app_name: str, user_id: str, session_id: str) -> str: """Constructs the Redis key for a given session.""" - return f"{self.config.key_prefix}{app_name}:{user_id}:{session_id}" + return ( + f"{self.config.key_prefix}v2:s2_{_encode_key_component(app_name)}." + f"{_encode_key_component(user_id)}.{_encode_key_component(session_id)}" + ) def _user_state_key(self, app_name: str, user_id: str) -> str: """Constructs the Redis key for user-scoped state.""" - return f"{self.config.key_prefix}user_state:{app_name}:{user_id}" + return ( + f"{self.config.key_prefix}v2:u2_{_encode_key_component(app_name)}." + f"{_encode_key_component(user_id)}" + ) def _app_state_key(self, app_name: str) -> str: """Constructs the Redis key for app-scoped state.""" - return f"{self.config.key_prefix}app_state:{app_name}" + return f"{self.config.key_prefix}v2:a2_{_encode_key_component(app_name)}" + + @staticmethod + def _matches_session_identity( + session: Session, app_name: str, user_id: str, session_id: str + ) -> bool: + """Checks that a stored session belongs to the requested identity.""" + return ( + session.app_name == app_name + and session.user_id == user_id + and session.id == session_id + ) @staticmethod def _merge_state( @@ -253,6 +276,10 @@ async def get_session( return None session = Session.model_validate_json(raw) + if not self._matches_session_identity( + session, app_name, user_id, session_id + ): + return None app_state = json.loads(app_raw) if app_raw else {} user_state = json.loads(user_raw) if user_raw else {} session.state = self._merge_state(app_state, user_state, session.state) @@ -275,12 +302,13 @@ async def list_sessions( ) -> ListSessionsResponse: """Lists sessions matching the given app_name and optional user_id.""" client = self._get_redis() - escaped_app_name = _escape_glob(app_name) pattern = ( - f"{self.config.key_prefix}{escaped_app_name}:{_escape_glob(user_id)}:*" - if user_id is not None - else f"{self.config.key_prefix}{escaped_app_name}:*" + f"{_escape_glob(self.config.key_prefix)}v2:" + f"s2_{_encode_key_component(app_name)}." ) + if user_id is not None: + pattern += f"{_encode_key_component(user_id)}." + pattern += "*" app_raw = await client.get(self._app_state_key(app_name)) app_state = json.loads(app_raw) if app_raw else {} @@ -296,9 +324,8 @@ async def list_sessions( if raw: try: s = Session.model_validate_json(raw) - # Key segments are ':'-joined and the trailing wildcard spans ':', so - # a matched key can belong to another app or user whose id embeds a - # ':'. Confirm ownership from the stored session itself. + # Confirm stored identity as defense in depth, even though the + # versioned key components are unambiguous. if s.app_name != app_name or ( user_id is not None and s.user_id != user_id ): @@ -324,7 +351,20 @@ async def delete_session( """Deletes a session from Redis.""" client = self._get_redis() key = self._session_key(app_name, user_id, session_id) - await client.delete(key) + + async def delete_if_identity_matches(pipe): + raw = await pipe.get(key) + if not raw: + return + session = Session.model_validate_json(raw) + if not self._matches_session_identity( + session, app_name, user_id, session_id + ): + return + pipe.multi() + pipe.delete(key) + + await client.transaction(delete_if_identity_matches, key) async def get_user_state( self, *, app_name: str, user_id: str @@ -341,11 +381,15 @@ async def append_event(self, session: Session, event: Event) -> Event: """Appends an event to the session and synchronizes state in Redis.""" client = self._get_redis() key = self._session_key(session.app_name, session.user_id, session.id) - # Check before any write: the app/user state deltas below are stored under - # their own keys, and the session write would otherwise create a session - # that was never stored instead of failing like the other services. - if not await client.exists(key): + raw = await client.get(key) + if not raw: raise SessionNotFoundError(f"Session {session.id} not found.") + stored_session = Session.model_validate_json(raw) + if not self._matches_session_identity( + stored_session, session.app_name, session.user_id, session.id + ): + raise SessionNotFoundError("Session not found.") + event = await super().append_event(session, event) # Stamp the session with the event's own timestamp, matching # InMemorySessionService, SqliteSessionService and DatabaseSessionService. @@ -354,8 +398,14 @@ async def append_event(self, session: Session, event: Event) -> Event: # last_update_time is the key list_sessions orders by. session.last_update_time = event.timestamp - # Sync app and user state deltas to their respective keys + # Collect state deltas before the transaction. State writes are queued in + # the same watched transaction as the session write so identity validation + # protects all persistent mutations. + app_delta: dict[str, Any] = {} + user_delta: dict[str, Any] = {} session_delta: dict[str, Any] = {} + app_key = None + user_key = None if event.actions and event.actions.state_delta: deltas = _session_util.extract_state_delta(event.actions.state_delta) app_delta = deltas.get("app", {}) @@ -364,25 +414,9 @@ async def append_event(self, session: Session, event: Event) -> Event: if app_delta: app_key = self._app_state_key(session.app_name) - app_raw = await client.get(app_key) - app_state = json.loads(app_raw) if app_raw else {} - app_state.update(app_delta) - await client.set( - app_key, - json.dumps(app_state), - ex=self.config.ttl_seconds if self.config.ttl_seconds > 0 else None, - ) if user_delta: user_key = self._user_state_key(session.app_name, session.user_id) - user_raw = await client.get(user_key) - user_state = json.loads(user_raw) if user_raw else {} - user_state.update(user_delta) - await client.set( - user_key, - json.dumps(user_state), - ex=self.config.ttl_seconds if self.config.ttl_seconds > 0 else None, - ) # Prepare storage copy with ONLY session-scoped state (excluding app:, user:, temp:) session_only_state = self._extract_session_only_state(session.state) @@ -394,6 +428,11 @@ async def append_event(self, session: Session, event: Event) -> Event: events=session.events, last_update_time=session.last_update_time, ) + watch_keys = [key] + if app_key: + watch_keys.append(app_key) + if user_key: + watch_keys.append(user_key) async def append_to_storage(pipe: Any) -> None: raw = await pipe.get(key) @@ -401,6 +440,10 @@ async def append_to_storage(pipe: Any) -> None: # The caller may hold a filtered view. Reload on every transaction # attempt so concurrent appends are preserved as well. stored_session = Session.model_validate_json(raw) + if not self._matches_session_identity( + stored_session, session.app_name, session.user_id, session.id + ): + raise SessionNotFoundError("Session not found.") storage_session.events = stored_session.events storage_session.state = stored_session.state if not event.partial: @@ -410,12 +453,37 @@ async def append_to_storage(pipe: Any) -> None: else: storage_session.events = list(session.events) storage_session.state = dict(session_only_state) + + app_state = None + if app_key: + app_raw = await pipe.get(app_key) + app_state = json.loads(app_raw) if app_raw else {} + app_state.update(app_delta) + + user_state = None + if user_key: + user_raw = await pipe.get(user_key) + user_state = json.loads(user_raw) if user_raw else {} + user_state.update(user_delta) + pipe.multi() + if app_key: + pipe.set( + app_key, + json.dumps(app_state), + ex=self.config.ttl_seconds if self.config.ttl_seconds > 0 else None, + ) + if user_key: + pipe.set( + user_key, + json.dumps(user_state), + ex=self.config.ttl_seconds if self.config.ttl_seconds > 0 else None, + ) pipe.set( key, storage_session.model_dump_json(), ex=self.config.ttl_seconds if self.config.ttl_seconds > 0 else None, ) - await client.transaction(append_to_storage, key) + await client.transaction(append_to_storage, *watch_keys) return event diff --git a/tests/unittests/integrations/redis/_fake_redis.py b/tests/unittests/integrations/redis/_fake_redis.py index 86a35426550..b77ced05b58 100644 --- a/tests/unittests/integrations/redis/_fake_redis.py +++ b/tests/unittests/integrations/redis/_fake_redis.py @@ -85,21 +85,31 @@ async def delete(self, key: str) -> int: return 0 async def transaction( - self, func: Callable[[Any], Awaitable[None]], key: str - ) -> list[bool | None]: - """Retries a queued write if the watched key changed or expired.""" + self, func: Callable[[Any], Awaitable[None]], *keys: str + ) -> list[Any]: + """Retries queued writes if any watched key changed or expired.""" while True: - self._is_expired(key) - version = self._versions.get(key, 0) + for key in keys: + self._is_expired(key) + versions = {key: self._versions.get(key, 0) for key in keys} pipe = Mock(get=self.get) await func(pipe) - self._is_expired(key) - if self._versions.get(key, 0) != version: + for key in keys: + self._is_expired(key) + if any( + self._versions.get(key, 0) != version + for key, version in versions.items() + ): continue - return [ + results = [ + await self.delete(*call.args, **call.kwargs) + for call in pipe.delete.call_args_list + ] + results.extend([ await self.set(*call.args, **call.kwargs) for call in pipe.set.call_args_list - ] + ]) + return results @staticmethod def _glob_match(pattern: str, key: str) -> bool: diff --git a/tests/unittests/integrations/redis/test_redis_key_collision_probe.py b/tests/unittests/integrations/redis/test_redis_key_collision_probe.py new file mode 100644 index 00000000000..0e14deca194 --- /dev/null +++ b/tests/unittests/integrations/redis/test_redis_key_collision_probe.py @@ -0,0 +1,92 @@ +# 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 collections.abc import AsyncGenerator + +import pytest + +from google.adk.agents import BaseAgent +from google.adk.agents.invocation_context import InvocationContext +from google.adk.events import Event +from google.adk.integrations.redis import RedisSessionService +from google.adk.runners import Runner +from google.genai import types + +from ._fake_redis import FakeRedisAsync + + +class _EchoAgent(BaseAgent): + + async def _run_async_impl( + self, ctx: InvocationContext + ) -> AsyncGenerator[Event, None]: + yield Event( + invocation_id=ctx.invocation_id, + author=self.name, + content=types.Content( + role="model", parts=[types.Part(text="processed")] + ), + ) + + +@pytest.mark.asyncio +async def test_runner_keeps_attacker_turn_out_of_victim_session(): + service = RedisSessionService(redis_client=FakeRedisAsync()) + await service.create_session( + app_name="billing", + user_id="tenant:victim", + session_id="fixed", + state={"private_record": "victim-only"}, + ) + await service.create_session( + app_name="billing", + user_id="tenant", + session_id="victim:fixed", + state={"attacker_record": "attacker-only"}, + ) + runner = Runner( + app_name="billing", + agent=_EchoAgent(name="echo"), + session_service=service, + ) + + _ = [ + event + async for event in runner.run_async( + user_id="tenant", + session_id="victim:fixed", + new_message=types.Content( + role="user", parts=[types.Part(text="attacker turn")] + ), + ) + ] + + victim = await service.get_session( + app_name="billing", user_id="tenant:victim", session_id="fixed" + ) + assert victim is not None + assert victim.state["private_record"] == "victim-only" + assert victim.events == [] + + attacker = await service.get_session( + app_name="billing", user_id="tenant", session_id="victim:fixed" + ) + assert attacker is not None + assert attacker.state["attacker_record"] == "attacker-only" + assert [ + part.text for event in attacker.events for part in event.content.parts + ] == [ + "attacker turn", + "processed", + ] diff --git a/tests/unittests/integrations/redis/test_redis_session_service.py b/tests/unittests/integrations/redis/test_redis_session_service.py index 4a4e22f2ccd..462875a1525 100644 --- a/tests/unittests/integrations/redis/test_redis_session_service.py +++ b/tests/unittests/integrations/redis/test_redis_session_service.py @@ -25,6 +25,7 @@ from google.adk.events.event import EventActions from google.adk.integrations.redis._config import RedisSessionServiceConfig from google.adk.integrations.redis._redis_session_service import RedisSessionService +from google.adk.integrations.redis._redis_session_service import _escape_glob from google.adk.sessions.base_session_service import GetSessionConfig from google.adk.sessions.session import Session from google.genai import types @@ -47,6 +48,107 @@ def session_service(fake_redis): return RedisSessionService(config=config, redis_client=fake_redis) +def test_key_components_are_unambiguous_and_type_separated(session_service): + assert ( + RedisSessionService() + ._session_key("app1", "user1", "session1") + .startswith("adk:session:v2:") + ) + assert ( + session_service._session_key("app1", "user1", "session1") + == "test:session:v2:s2_YXBwMQ.dXNlcjE.c2Vzc2lvbjE" + ) + victim_key = session_service._session_key("billing", "tenant:victim", "fixed") + assert ( + victim_key == "test:session:v2:s2_YmlsbGluZw.dGVuYW50OnZpY3RpbQ.Zml4ZWQ" + ) + assert victim_key != session_service._session_key( + "billing", "tenant", "victim:fixed" + ) + assert session_service._user_state_key( + "billing:tenant", "victim" + ) != session_service._user_state_key("billing", "tenant:victim") + assert session_service._session_key( + "user_state", "billing", "alice" + ) != session_service._user_state_key("billing", "alice") + assert session_service._session_key( + "app_state", "billing", "alice" + ) != session_service._app_state_key("billing:alice") + assert ( + session_service._app_state_key("billing") + == "test:session:v2:a2_YmlsbGluZw" + ) + for key in ( + victim_key, + session_service._user_state_key("billing:tenant", "victim"), + session_service._app_state_key("billing:tenant"), + ): + encoded_suffix = key.removeprefix("test:session:v2:") + assert ":" not in encoded_suffix + assert not any(char in encoded_suffix for char in "*?[]\\") + + assert _escape_glob("prefix:*?[x]") == r"prefix:\*\?\[x\]" + assert _escape_glob("a\\b") == "a\\\\b" + + +@pytest.mark.asyncio +async def test_versioned_keys_ignore_legacy_session_and_state_keys( + session_service, fake_redis +): + legacy_session = Session( + id="fixed", + app_name="billing", + user_id="tenant:victim", + state={"private": "legacy-session"}, + events=[], + last_update_time=1.0, + ) + legacy_key = "test:session:billing:tenant:victim:fixed" + await fake_redis.set(legacy_key, legacy_session.model_dump_json()) + legacy_session_at_user_state_key = Session( + id="alice", + app_name="user_state", + user_id="billing", + state={"private": "legacy-session-at-state-key"}, + events=[], + last_update_time=1.0, + ) + await fake_redis.set( + "test:session:user_state:billing:alice", + legacy_session_at_user_state_key.model_dump_json(), + ) + await fake_redis.set("test:session:app_state:billing", '{"old": true}') + + assert ( + await session_service.get_session( + app_name="billing", + user_id="tenant:victim", + session_id="fixed", + ) + is None + ) + assert ( + await session_service.get_user_state(app_name="billing", user_id="alice") + == {} + ) + assert ( + await session_service.list_sessions( + app_name="billing", user_id="tenant:victim" + ) + ).sessions == [] + assert ( + await session_service.list_sessions(app_name="billing") + ).sessions == [] + await session_service.delete_session( + app_name="billing", user_id="tenant:victim", session_id="fixed" + ) + assert await fake_redis.get(legacy_key) == legacy_session.model_dump_json() + new_session = await session_service.create_session( + app_name="billing", user_id="alice", session_id="new" + ) + assert new_session.state == {} + + @pytest.mark.asyncio async def test_create_session(session_service): session = await session_service.create_session( @@ -109,6 +211,232 @@ async def test_get_session_not_found(session_service): assert fetched is None +@pytest.mark.asyncio +async def test_get_list_and_delete_reject_colon_key_alias(session_service): + """A colon-delimited identity alias cannot read, list, or delete a session.""" + victim = await session_service.create_session( + app_name="billing", + user_id="tenant:victim", + session_id="fixed", + state={"secret": "victim-only"}, + ) + assert session_service._session_key( + "billing", "tenant:victim", "fixed" + ) != session_service._session_key("billing", "tenant", "victim:fixed") + + assert ( + await session_service.get_session( + app_name="billing", user_id="tenant", session_id="victim:fixed" + ) + is None + ) + assert ( + await session_service.list_sessions(app_name="billing", user_id="tenant") + ).sessions == [] + assert [ + s.id + for s in ( + await session_service.list_sessions(app_name="billing") + ).sessions + ] == [victim.id] + + await session_service.delete_session( + app_name="billing", user_id="tenant", session_id="victim:fixed" + ) + fetched = await session_service.get_session( + app_name="billing", user_id="tenant:victim", session_id="fixed" + ) + assert fetched is not None + assert fetched.state["secret"] == "victim-only" + + +@pytest.mark.asyncio +async def test_delete_session_does_not_delete_concurrent_alias_replacement( + session_service, fake_redis +): + """A watched delete retries if the key changes after its identity read.""" + victim = await session_service.create_session( + app_name="billing", + user_id="tenant:victim", + session_id="fixed", + ) + alias = Session( + id="victim:fixed", + app_name="billing", + user_id="tenant", + state={"owner": "replacement"}, + events=[], + last_update_time=victim.last_update_time, + ) + key = session_service._session_key("billing", "tenant:victim", "fixed") + original_get = fake_redis.get + replaced = False + + async def replace_after_read(read_key): + nonlocal replaced + raw = await original_get(read_key) + if read_key == key and not replaced: + replaced = True + await fake_redis.delete(key) + await fake_redis.set(key, alias.model_dump_json()) + return raw + + fake_redis.get = replace_after_read + await session_service.delete_session( + app_name="billing", user_id="tenant:victim", session_id="fixed" + ) + + remaining = await original_get(key) + assert remaining is not None + assert Session.model_validate_json(remaining).user_id == "tenant" + assert Session.model_validate_json(remaining).state == { + "owner": "replacement" + } + + +@pytest.mark.asyncio +async def test_append_event_does_not_commit_state_after_identity_race( + session_service, fake_redis +): + """A replacement identity aborts queued session and state writes.""" + victim = await session_service.create_session( + app_name="billing", + user_id="tenant:victim", + session_id="fixed", + ) + alias = Session( + id="victim:fixed", + app_name="billing", + user_id="tenant", + state={"owner": "replacement"}, + events=[], + last_update_time=victim.last_update_time, + ) + key = session_service._session_key("billing", "tenant:victim", "fixed") + original_get = fake_redis.get + replaced = False + key_reads = 0 + + async def replace_after_read(read_key): + nonlocal key_reads, replaced + raw = await original_get(read_key) + if read_key == key: + key_reads += 1 + if read_key == key and key_reads == 2 and not replaced: + replaced = True + await fake_redis.delete(key) + await fake_redis.set(key, alias.model_dump_json()) + return raw + + fake_redis.get = replace_after_read + event = Event( + author="user", + actions=EventActions( + state_delta={ + "app:should_not_write": True, + "user:should_not_write": True, + "session:should_not_write": True, + } + ), + ) + with pytest.raises(SessionNotFoundError): + await session_service.append_event(victim, event) + + assert await original_get(session_service._app_state_key("billing")) is None + assert ( + await original_get( + session_service._user_state_key("billing", "tenant:victim") + ) + is None + ) + remaining = await original_get(key) + assert remaining is not None + assert Session.model_validate_json(remaining).user_id == "tenant" + assert Session.model_validate_json(remaining).state == { + "owner": "replacement" + } + + +@pytest.mark.asyncio +async def test_append_event_rejects_mismatched_stored_identity_before_mutation( + session_service, fake_redis +): + """A mismatched Redis record is rejected before memory or state changes.""" + victim = await session_service.create_session( + app_name="billing", + user_id="tenant:victim", + session_id="fixed", + state={"secret": "victim-only"}, + ) + attacker_view = Session( + id="victim:fixed", + app_name="billing", + user_id="tenant", + state={"attacker": "local-only"}, + events=[], + last_update_time=victim.last_update_time, + ) + attacker_key = session_service._session_key( + "billing", "tenant", "victim:fixed" + ) + await fake_redis.set(attacker_key, victim.model_dump_json()) + event = Event( + author="user", + actions=EventActions( + state_delta={ + "app:alias_write": True, + "user:alias_write": True, + "session_alias_write": True, + } + ), + ) + + with pytest.raises(SessionNotFoundError): + await session_service.append_event(attacker_view, event) + + assert attacker_view.events == [] + assert "session_alias_write" not in attacker_view.state + assert ( + await session_service.get_user_state(app_name="billing", user_id="tenant") + == {} + ) + assert ( + await session_service.get_user_state( + app_name="billing", user_id="tenant:victim" + ) + == {} + ) + stored = await session_service.get_session( + app_name="billing", user_id="tenant:victim", session_id="fixed" + ) + assert stored is not None + assert stored.events == [] + assert stored.state == {"secret": "victim-only"} + assert await fake_redis.get(attacker_key) == victim.model_dump_json() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("partial", [False, True]) +async def test_append_event_to_unknown_session_writes_nothing( + session_service, fake_redis, partial +): + """A rejected append leaves no session, app state or user state behind.""" + session = Session(app_name="app1", user_id="u1", id="never_created") + event = Event( + author="agent", + partial=partial, + actions=EventActions( + state_delta={"count": 1, "user:score": 100, "app:status": "active"} + ), + ) + + with pytest.raises(SessionNotFoundError): + await session_service.append_event(session, event) + + assert fake_redis._store == {} + assert session.events == [] + + @pytest.mark.asyncio async def test_get_session_with_event_filter(session_service): session = await session_service.create_session( @@ -424,11 +752,11 @@ async def test_list_sessions_glob_metacharacters_match_literally( ) for app_name, user_id, expected_pattern in ( - ("app1", "*", r"test:session:app1:\*:*"), - ("app1", "u?", r"test:session:app1:u\?:*"), - ("app1", "[u]1", r"test:session:app1:\[u\]1:*"), - ("app1", "u\\1", r"test:session:app1:u\\1:*"), - ("*", "u1", r"test:session:\*:u1:*"), + ("app1", "*", "test:session:v2:s2_YXBwMQ.Kg.*"), + ("app1", "u?", "test:session:v2:s2_YXBwMQ.dT8.*"), + ("app1", "[u]1", "test:session:v2:s2_YXBwMQ.W3VdMQ.*"), + ("app1", "u\\1", "test:session:v2:s2_YXBwMQ.dVwx.*"), + ("*", "u1", "test:session:v2:s2_Kg.dTE.*"), ): fake_redis.scan_patterns.clear() resp = await session_service.list_sessions( @@ -439,7 +767,7 @@ async def test_list_sessions_glob_metacharacters_match_literally( fake_redis.scan_patterns.clear() resp = await session_service.list_sessions(app_name="*") - assert fake_redis.scan_patterns == [r"test:session:\*:*"] + assert fake_redis.scan_patterns == ["test:session:v2:s2_Kg.*"] assert resp.sessions == [] From 2f2c9ba65f309b65609fdec48e9f0b67fe8f7fc6 Mon Sep 17 00:00:00 2001 From: David Sarkisyan Date: Thu, 15 Oct 2026 15:45:13 -0400 Subject: [PATCH 2/4] Simplify Redis session delete and append checks --- .../redis/_redis_session_service.py | 23 +------- .../redis/test_redis_session_service.py | 53 ++++--------------- 2 files changed, 11 insertions(+), 65 deletions(-) diff --git a/src/google/adk/integrations/redis/_redis_session_service.py b/src/google/adk/integrations/redis/_redis_session_service.py index 55521d7a423..6a700307087 100644 --- a/src/google/adk/integrations/redis/_redis_session_service.py +++ b/src/google/adk/integrations/redis/_redis_session_service.py @@ -351,20 +351,7 @@ async def delete_session( """Deletes a session from Redis.""" client = self._get_redis() key = self._session_key(app_name, user_id, session_id) - - async def delete_if_identity_matches(pipe): - raw = await pipe.get(key) - if not raw: - return - session = Session.model_validate_json(raw) - if not self._matches_session_identity( - session, app_name, user_id, session_id - ): - return - pipe.multi() - pipe.delete(key) - - await client.transaction(delete_if_identity_matches, key) + await client.delete(key) async def get_user_state( self, *, app_name: str, user_id: str @@ -381,14 +368,8 @@ async def append_event(self, session: Session, event: Event) -> Event: """Appends an event to the session and synchronizes state in Redis.""" client = self._get_redis() key = self._session_key(session.app_name, session.user_id, session.id) - raw = await client.get(key) - if not raw: + if not await client.exists(key): raise SessionNotFoundError(f"Session {session.id} not found.") - stored_session = Session.model_validate_json(raw) - if not self._matches_session_identity( - stored_session, session.app_name, session.user_id, session.id - ): - raise SessionNotFoundError("Session not found.") event = await super().append_event(session, event) # Stamp the session with the event's own timestamp, matching diff --git a/tests/unittests/integrations/redis/test_redis_session_service.py b/tests/unittests/integrations/redis/test_redis_session_service.py index 462875a1525..ceba51d5271 100644 --- a/tests/unittests/integrations/redis/test_redis_session_service.py +++ b/tests/unittests/integrations/redis/test_redis_session_service.py @@ -251,47 +251,16 @@ async def test_get_list_and_delete_reject_colon_key_alias(session_service): @pytest.mark.asyncio -async def test_delete_session_does_not_delete_concurrent_alias_replacement( +async def test_delete_session_removes_corrupt_record( session_service, fake_redis ): - """A watched delete retries if the key changes after its identity read.""" - victim = await session_service.create_session( - app_name="billing", - user_id="tenant:victim", - session_id="fixed", - ) - alias = Session( - id="victim:fixed", - app_name="billing", - user_id="tenant", - state={"owner": "replacement"}, - events=[], - last_update_time=victim.last_update_time, - ) + """Deletion works even if the stored session cannot be deserialized.""" key = session_service._session_key("billing", "tenant:victim", "fixed") - original_get = fake_redis.get - replaced = False - - async def replace_after_read(read_key): - nonlocal replaced - raw = await original_get(read_key) - if read_key == key and not replaced: - replaced = True - await fake_redis.delete(key) - await fake_redis.set(key, alias.model_dump_json()) - return raw - - fake_redis.get = replace_after_read + await fake_redis.set(key, "{corrupt json") await session_service.delete_session( app_name="billing", user_id="tenant:victim", session_id="fixed" ) - - remaining = await original_get(key) - assert remaining is not None - assert Session.model_validate_json(remaining).user_id == "tenant" - assert Session.model_validate_json(remaining).state == { - "owner": "replacement" - } + assert await fake_redis.get(key) is None @pytest.mark.asyncio @@ -315,14 +284,11 @@ async def test_append_event_does_not_commit_state_after_identity_race( key = session_service._session_key("billing", "tenant:victim", "fixed") original_get = fake_redis.get replaced = False - key_reads = 0 async def replace_after_read(read_key): - nonlocal key_reads, replaced + nonlocal replaced raw = await original_get(read_key) - if read_key == key: - key_reads += 1 - if read_key == key and key_reads == 2 and not replaced: + if read_key == key and not replaced: replaced = True await fake_redis.delete(key) await fake_redis.set(key, alias.model_dump_json()) @@ -358,10 +324,10 @@ async def replace_after_read(read_key): @pytest.mark.asyncio -async def test_append_event_rejects_mismatched_stored_identity_before_mutation( +async def test_append_event_rejects_mismatched_stored_identity_without_persistent_mutation( session_service, fake_redis ): - """A mismatched Redis record is rejected before memory or state changes.""" + """A mismatched Redis record cannot cause persistent cross-identity writes.""" victim = await session_service.create_session( app_name="billing", user_id="tenant:victim", @@ -394,8 +360,7 @@ async def test_append_event_rejects_mismatched_stored_identity_before_mutation( with pytest.raises(SessionNotFoundError): await session_service.append_event(attacker_view, event) - assert attacker_view.events == [] - assert "session_alias_write" not in attacker_view.state + assert await fake_redis.get(session_service._app_state_key("billing")) is None assert ( await session_service.get_user_state(app_name="billing", user_id="tenant") == {} From fd1751547dbd17b9a39806cb75cc781143bccd92 Mon Sep 17 00:00:00 2001 From: David Sarkisyan <281478990+srkyn@users.noreply.github.com> Date: Thu, 15 Oct 2026 16:20:04 -0400 Subject: [PATCH 3/4] Ignore legacy keys during Redis session listing --- .../redis/_redis_session_service.py | 10 ++++- .../redis/test_redis_session_service.py | 40 +++++++++++++++++++ 2 files changed, 48 insertions(+), 2 deletions(-) diff --git a/src/google/adk/integrations/redis/_redis_session_service.py b/src/google/adk/integrations/redis/_redis_session_service.py index 6a700307087..6b37837de78 100644 --- a/src/google/adk/integrations/redis/_redis_session_service.py +++ b/src/google/adk/integrations/redis/_redis_session_service.py @@ -324,8 +324,14 @@ async def list_sessions( if raw: try: s = Session.model_validate_json(raw) - # Confirm stored identity as defense in depth, even though the - # versioned key components are unambiguous. + # A legacy key can match the v2 scan pattern; only list canonical keys. + expected_key = self._session_key(s.app_name, s.user_id, s.id) + if key != ( + expected_key.encode("utf-8") + if isinstance(key, bytes) + else expected_key + ): + continue if s.app_name != app_name or ( user_id is not None and s.user_id != user_id ): diff --git a/tests/unittests/integrations/redis/test_redis_session_service.py b/tests/unittests/integrations/redis/test_redis_session_service.py index ceba51d5271..c0c1c0c3dee 100644 --- a/tests/unittests/integrations/redis/test_redis_session_service.py +++ b/tests/unittests/integrations/redis/test_redis_session_service.py @@ -149,6 +149,46 @@ async def test_versioned_keys_ignore_legacy_session_and_state_keys( assert new_session.state == {} +@pytest.mark.parametrize("binary_scan_keys", [False, True]) +@pytest.mark.asyncio +async def test_list_sessions_ignores_legacy_key_matching_v2_scan( + session_service, fake_redis, binary_scan_keys +): + legacy_session = Session( + id="bar", + app_name="v2", + user_id="s2_djI.foo", + state={"private": "legacy-session"}, + events=[], + last_update_time=1.0, + ) + await fake_redis.set( + "test:session:v2:s2_djI.foo:bar", legacy_session.model_dump_json() + ) + current = await session_service.create_session( + app_name="v2", user_id="alice", session_id="live" + ) + if binary_scan_keys: + original_scan_iter = fake_redis.scan_iter + original_get = fake_redis.get + + async def scan_bytes(match): + async for key in original_scan_iter(match): + yield key.encode("utf-8") + + async def get_with_bytes(key): + return await original_get( + key.decode("utf-8") if isinstance(key, bytes) else key + ) + + fake_redis.scan_iter = scan_bytes + fake_redis.get = get_with_bytes + + assert (await session_service.list_sessions(app_name="v2")).sessions == [ + current + ] + + @pytest.mark.asyncio async def test_create_session(session_service): session = await session_service.create_session( From b4dcdf5d6acdab3c06a3ba7796325378a37d4add Mon Sep 17 00:00:00 2001 From: David Sarkisyan <281478990+srkyn@users.noreply.github.com> Date: Tue, 20 Oct 2026 15:49:17 -0400 Subject: [PATCH 4/4] Document Redis Cluster limitation --- .../guides/integrations/redis/redis_session_service/index.md | 5 +++++ src/google/adk/integrations/redis/README.md | 5 +++++ 2 files changed, 10 insertions(+) diff --git a/docs/guides/integrations/redis/redis_session_service/index.md b/docs/guides/integrations/redis/redis_session_service/index.md index 2f7756c0ac8..f8a17d2f54e 100644 --- a/docs/guides/integrations/redis/redis_session_service/index.md +++ b/docs/guides/integrations/redis/redis_session_service/index.md @@ -123,6 +123,11 @@ service = RedisSessionService(redis_client=shared_client) RedisSessionService requires the `redis` Python package to be installed. You can install it using `pip install google-adk[redis]`. +Native Redis Cluster is not supported. When an event updates `app:` or `user:` +state, the service watches the session and state keys in one transaction. These +keys are not guaranteed to share a Cluster hash slot, so the transaction can +fail with `CROSSSLOT`. Use a standalone Redis instance for this service. + State keys prefixed with `temp:` are treated as transient. While they are available in the session object immediately after creation or an update, they are not persisted to Redis and will disappear when the session is reloaded diff --git a/src/google/adk/integrations/redis/README.md b/src/google/adk/integrations/redis/README.md index 12bf7c14118..92a12ac4215 100644 --- a/src/google/adk/integrations/redis/README.md +++ b/src/google/adk/integrations/redis/README.md @@ -102,6 +102,11 @@ client = redis_asyncio.Redis.from_url( session_service = RedisSessionService(redis_client=client) ``` +Native Redis Cluster is not supported. When an event updates `app:` or `user:` +state, the service watches the session and state keys in one transaction. These +keys are not guaranteed to share a Cluster hash slot, so the transaction can +fail with `CROSSSLOT`. Use a standalone Redis instance for this service. + ## Key Schema & State Scoping `RedisSessionService` stores session data and state using the following Redis key structure: