diff --git a/packages/kaya-oidc/README.md b/packages/kaya-oidc/README.md index d235e03..491161c 100644 --- a/packages/kaya-oidc/README.md +++ b/packages/kaya-oidc/README.md @@ -128,6 +128,50 @@ app = KayaApp(mixins=[session, google, github]) > default provider is `default`). Existing logged-in users are signed out once > and simply re-authenticate. +## Back-channel logout + +Set `backchannel_logout_path` to accept OIDC Back-Channel Logout tokens from +the provider: + +```python +OIDCConfig( + issuer=os.environ['OIDC_ISSUER'], + client_id=os.environ['OIDC_CLIENT_ID'], + client_secret=os.environ['OIDC_CLIENT_SECRET'], + redirect_uri='https://example.com/auth/callback', + backchannel_logout_path='/auth/backchannel-logout', +) +``` + +Register the corresponding absolute URL (for example +`https://example.com/auth/backchannel-logout`) as the client's back-channel +logout URI with the provider. This registration happens at the provider; it is +not part of OIDC discovery. With `OIDCProviders`, enabling the option derives +`{base_path}/{provider}/backchannel-logout` for that provider. + +The endpoint accepts form-encoded `POST` requests containing `logout_token`. +It validates the JWT signature against the provider JWKS, the `logout+jwt` +type, issuer, audience, `iat`, `jti`, and the back-channel logout event, and +rejects tokens containing `nonce`. A token with `sid` logs out sessions for +that provider session; without `sid`, `sub` logs out all sessions for that +subject. Under `multi-identity`, only the named provider's identity is removed +from a shared session. Valid requests return `200` with +`Cache-Control: no-store`, including when the session is already gone. Invalid +requests return `400`. + +Back-channel logout requires a session store with index support. All built-in +stores (`InMemorySessionStore`, `RedisSessionStore`, and +`MemcacheSessionStore`) support it. `OIDCMixin` raises immediately if the +configured store does not. Sessions created before upgrading are indexed on +their next request. + +Logout tokens must be asymmetrically signed; unsigned tokens, HMAC signatures, +and encrypted logout tokens (JWE) are not supported. `jti` is required, but no +replay cache is maintained: repeating a valid logout request is idempotent. +`backchannel_logout_leeway` controls the allowed clock skew (60 seconds by +default). For a templated Microsoft Entra issuer, the logout token must carry +`tid`, just as for ID token validation. + ## Pushed authorization requests (PAR) `OIDCConfig.par` controls RFC 9126 Pushed Authorization Requests: @@ -172,6 +216,7 @@ authentication methods (`private_key_jwt`, mTLS) is not supported. - Optional userinfo endpoint fetch - Refresh token support - RP-initiated logout (when provider advertises `end_session_endpoint`) +- Back-channel logout (`sid` and `sub` session invalidation) - Pushed Authorization Requests (RFC 9126) via `OIDCConfig.par` - Multiple providers in one app via `OIDCProviders` (or several `OIDCMixin` instances) - `single-identity` (default) or `multi-identity` session strategy diff --git a/packages/kaya-oidc/src/kaya/oidc/_client.py b/packages/kaya-oidc/src/kaya/oidc/_client.py index f776353..77cd096 100644 --- a/packages/kaya-oidc/src/kaya/oidc/_client.py +++ b/packages/kaya-oidc/src/kaya/oidc/_client.py @@ -10,6 +10,8 @@ from ._config import PARPolicy, OIDCConfig from ._utils import generate_nonce, generate_pkce, generate_state _TENANT_ID_PLACEHOLDER = re.compile(r'\{tenantid\}', re.IGNORECASE) +_BACKCHANNEL_LOGOUT_EVENT = 'http://schemas.openid.net/event/backchannel-logout' +_LOGOUT_TOKEN_TYPES = frozenset({'logout+jwt', 'application/logout+jwt'}) class OIDCClient: @@ -189,6 +191,60 @@ class OIDCClient: raise ValueError('ID token nonce mismatch') return payload + async def validate_logout_token(self, logout_token: str) -> Mapping[str, Any]: + """Validate an OIDC Back-Channel Logout token and return its claims.""" + header = jwt.get_unverified_header(logout_token) + typ = header.get('typ') + if not isinstance(typ, str) or typ.lower() not in _LOGOUT_TOKEN_TYPES: + raise ValueError('Logout token typ must be logout+jwt') + + alg = header.get('alg') + if not isinstance(alg, str): + raise ValueError('Logout token header is missing alg') + if alg == 'none' or alg.startswith('HS'): + raise ValueError('Logout tokens must use an asymmetric signature') + if alg not in self._config.allowed_id_token_algorithms: + raise ValueError(f'Logout token algorithm {alg} is not allowed') + + metadata_issuer = await self._get_metadata_issuer() + jwks = await self._ensure_jwks() + signing_key = self._get_signing_key(jwks, header.get('kid'), alg) + payload = jwt.decode( + logout_token, + signing_key.key, + algorithms=[alg], + audience=self._config.client_id, + leeway=self._config.backchannel_logout_leeway, + options={ + 'require': ['iss', 'aud', 'iat', 'jti', 'events'], + 'verify_exp': True, + 'verify_iat': True, + 'verify_nbf': True, + }, + ) + if not isinstance(payload, dict): + raise ValueError('Logout token payload is not a JSON object') + self._validate_issuer(payload, metadata_issuer) + + if not isinstance(payload.get('jti'), str): + raise ValueError('Logout token jti must be a string') + events = payload.get('events') + if (not isinstance(events, Mapping) + or not isinstance(events.get(_BACKCHANNEL_LOGOUT_EVENT), Mapping)): + raise ValueError('Logout token is missing the back-channel logout event') + if 'nonce' in payload: + raise ValueError('Logout token must not contain a nonce') + + sub = payload.get('sub') + sid = payload.get('sid') + if sub is not None and not isinstance(sub, str): + raise ValueError('Logout token sub must be a string') + if sid is not None and not isinstance(sid, str): + raise ValueError('Logout token sid must be a string') + if sub is None and sid is None: + raise ValueError('Logout token must contain sub or sid') + return payload + async def _decode_id_token(self, id_token: str) -> Mapping[str, Any]: header = jwt.get_unverified_header(id_token) alg = header.get('alg') diff --git a/packages/kaya-oidc/src/kaya/oidc/_config.py b/packages/kaya-oidc/src/kaya/oidc/_config.py index 1715429..c5f0505 100644 --- a/packages/kaya-oidc/src/kaya/oidc/_config.py +++ b/packages/kaya-oidc/src/kaya/oidc/_config.py @@ -26,6 +26,12 @@ class OIDCConfig: A provider that sets ``require_pushed_authorization_requests`` is always treated as requiring PAR. + ``backchannel_logout_path`` enables OIDC Back-Channel Logout when set to + an application path. Register the corresponding absolute URL with the + provider. It requires a session store with index support. + ``backchannel_logout_leeway`` is the clock-skew allowance, in seconds, + used when validating logout tokens. + ``client_secret`` is optional; omit it for public clients. ``fetch_userinfo`` controls whether the userinfo endpoint is queried after token exchange. @@ -55,6 +61,8 @@ class OIDCConfig: http_client: Optional[httpx.AsyncClient] = None provider: str = 'default' par: PARPolicy = 'never' + backchannel_logout_path: Optional[str] = None + backchannel_logout_leeway: int = 60 def __post_init__(self) -> None: if not self.provider or _PROVIDER_PATTERN.match(self.provider) is None: @@ -68,6 +76,11 @@ class OIDCConfig: f'par must be one of {", ".join(map(repr, _PAR_POLICIES))} ' f'(got {self.par!r})' ) + if (self.backchannel_logout_path is not None + and not self.backchannel_logout_path.startswith('/')): + raise ValueError('backchannel_logout_path must start with "/"') + if self.backchannel_logout_leeway < 0: + raise ValueError('backchannel_logout_leeway must be non-negative') scopes_list = list(self.scopes) if 'openid' not in scopes_list: self.scopes = ('openid', *scopes_list) diff --git a/packages/kaya-oidc/src/kaya/oidc/_mixin.py b/packages/kaya-oidc/src/kaya/oidc/_mixin.py index f5fb863..b67fc51 100644 --- a/packages/kaya-oidc/src/kaya/oidc/_mixin.py +++ b/packages/kaya-oidc/src/kaya/oidc/_mixin.py @@ -15,8 +15,9 @@ type IdentityStrategy = Literal['single-identity', 'multi-identity'] _SESSION_NAMESPACE = 'oidc' _USER_SUFFIX = 'user' -_IDENTITY_FIELDS = frozenset({'user', 'id_token', 'access_token', 'refresh_token'}) +_IDENTITY_FIELDS = frozenset({'user', 'id_token', 'access_token', 'refresh_token', 'sid'}) _IDENTITY_STRATEGIES: tuple[IdentityStrategy, ...] = ('single-identity', 'multi-identity') +_BACKCHANNEL_BODY_LIMIT = 64 * 1024 class OIDCUser(Mapping[str, Any]): @@ -64,6 +65,10 @@ class OIDCMixin(KayaMixin): - ``callback_path`` (default ``/auth/callback``): handles the provider callback. - ``logout_path`` (default ``/auth/logout``): logs the user out. + When ``backchannel_logout_path`` is configured, a POST route accepting + OIDC Back-Channel Logout tokens is registered as well. This requires an + index-capable session store. + Session state is namespaced by ``OIDCConfig.provider`` as ``oidc:{provider}:...`` so several providers can share one session. Use one ``OIDCMixin`` per provider (with distinct route paths), or @@ -104,6 +109,11 @@ class OIDCMixin(KayaMixin): self._client = OIDCClient(config) self._strategy = strategy self._prefix = f'{_SESSION_NAMESPACE}:{config.provider}:' + if (config.backchannel_logout_path is not None + and not session.store.supports_indexes): + raise ValueError( + 'backchannel_logout_path requires a session store with index support' + ) @property def provider(self) -> str: @@ -116,11 +126,73 @@ class OIDCMixin(KayaMixin): def _key(self, name: str) -> str: return f'{self._prefix}{name}' + def _index_name(self, claim: str, provider: Optional[str] = None) -> str: + return f'{_SESSION_NAMESPACE}:{provider or self.provider}:{claim}' + + def _apply_identity_indexes( + self, + session: Session, + sub: str, + sid: Optional[str], + ) -> None: + session.add_index(self._index_name('sub'), sub) + if sid is not None: + session.add_index(self._index_name('sid'), sid) + session[self._key('indexed')] = True + + def _remove_identity_indexes( + self, + session: Session, + provider: Optional[str] = None, + ) -> None: + prefix = f'{_SESSION_NAMESPACE}:{provider or self.provider}:' + user = session.get(f'{prefix}user') + if isinstance(user, Mapping): + sub = user.get('sub') + if isinstance(sub, str): + session.remove_index(self._index_name('sub', provider), sub) + sid = session.get(f'{prefix}sid') + if isinstance(sid, str): + session.remove_index(self._index_name('sid', provider), sid) + + def _backfill_identity_indexes(self, session: Session) -> None: + """Index an authenticated session created before index support.""" + user = session.get(self._key('user')) + if not isinstance(user, Mapping) or self._key('indexed') in session: + return + sub = user.get('sub') + if not isinstance(sub, str): + return + + sid = session.get(self._key('sid')) + if not isinstance(sid, str): + id_token = session.get(self._key('id_token')) + if isinstance(id_token, str): + try: + claims = jwt.decode( + id_token, + options={'verify_signature': False}, + ) + candidate = claims.get('sid') + if isinstance(candidate, str): + sid = candidate + session[self._key('sid')] = candidate + except jwt.PyJWTError: + pass + self._apply_identity_indexes( + session, + sub, + sid if isinstance(sid, str) else None, + ) + @property def dependencies(self) -> Sequence[KayaMixin]: return [self._session] def apply(self, app: KayaApp) -> None: + if self._config.backchannel_logout_path is not None: + app.add_before_request_hook(self._before_request) + @app.GET(self._config.login_path) async def login(ctx: HttpContext) -> None: auth_url, state, nonce, code_verifier = await self._client.build_authorization_url( @@ -168,6 +240,12 @@ class OIDCMixin(KayaMixin): return claims = await self._client.validate_id_token(id_token, nonce) + sub = claims.get('sub') + if not isinstance(sub, str): + raise ValueError('ID token is missing a valid sub claim') + sid = claims.get('sid') + if sid is not None and not isinstance(sid, str): + raise ValueError('ID token sid must be a string') user: dict[str, Any] = dict(claims) access_token = token_response.get('access_token') @@ -180,8 +258,13 @@ class OIDCMixin(KayaMixin): # Keep this provider's nonce so refreshed ID tokens can be # checked for nonce continuity (OIDC Core 12.2). + self._remove_identity_indexes(session) session[self._key('user')] = user session[self._key('id_token')] = id_token + session.pop(self._key('sid'), None) + if sid is not None: + session[self._key('sid')] = sid + self._apply_identity_indexes(session, sub, sid) if isinstance(access_token, str): session[self._key('access_token')] = access_token if 'refresh_token' in token_response: @@ -204,6 +287,27 @@ class OIDCMixin(KayaMixin): location = logout_url if logout_url is not None else self._config.post_logout_redirect await ctx.send_empty(302, {'Location': location}) + backchannel_path = self._config.backchannel_logout_path + if backchannel_path is not None: + @app.POST(backchannel_path) + async def backchannel_logout(ctx: HttpContext) -> None: + # This endpoint is server-to-server. Never persist a browser + # session that happens to accompany the request. + ctx.session = Session() + try: + logout_token = await self._read_logout_token(ctx) + claims = await self._client.validate_logout_token(logout_token) + except (ValueError, jwt.PyJWTError): + await ctx.send_str( + 400, + 'Invalid logout token', + {'Cache-Control': 'no-store'}, + ) + return + + await self._invalidate_by_logout_token(claims) + await ctx.send_empty(200, {'Cache-Control': 'no-store'}) + def _logout_session(self, session: Session) -> None: """Drop this provider's session state. @@ -211,6 +315,7 @@ class OIDCMixin(KayaMixin): logged in, so logging out of one provider does not sign the user out of the others. """ + self._remove_identity_indexes(session) for key in [key for key in session if key.startswith(self._prefix)]: del session[key] if not any(_is_user_key(key) for key in session): @@ -224,10 +329,75 @@ class OIDCMixin(KayaMixin): can still complete. Under this strategy the last callback to finish wins. """ + other_providers = { + _provider_of_user_key(key) + for key in session + if _is_user_key(key) and not key.startswith(self._prefix) + } + for provider in other_providers: + self._remove_identity_indexes(session, provider) for key in [key for key in session if _is_identity_key(key) and not key.startswith(self._prefix)]: del session[key] + async def _before_request(self, ctx: HttpContext) -> Optional[HttpContext]: + session = ctx.session + if isinstance(session, Session): + self._backfill_identity_indexes(session) + return None + + async def _read_logout_token(self, ctx: HttpContext) -> str: + content_type_values = ctx.headers.get('content-type') + if isinstance(content_type_values, str): + content_type = content_type_values + elif content_type_values: + content_type = content_type_values[0] + else: + content_type = '' + if content_type.split(';', 1)[0].strip().lower() != 'application/x-www-form-urlencoded': + raise ValueError('Back-channel logout requires a form-encoded request') + + body = bytearray() + async for chunk in ctx.request_body: + body.extend(chunk) + if len(body) > _BACKCHANNEL_BODY_LIMIT: + raise ValueError('Back-channel logout request is too large') + form = parse_qs(body.decode('utf-8'), keep_blank_values=True) + logout_token = self._first_value(form.get('logout_token')) + if logout_token is None: + raise ValueError('Missing logout_token') + return logout_token + + async def _invalidate_by_logout_token(self, claims: Mapping[str, Any]) -> None: + store = self._session.store + sub = claims.get('sub') + sid = claims.get('sid') + if isinstance(sid, str): + session_ids = await store.find_sessions(self._index_name('sid'), sid) + elif isinstance(sub, str): + session_ids = await store.find_sessions(self._index_name('sub'), sub) + else: + return + + for session_id in session_ids: + session = await store.load(session_id) + if session is None: + continue + if isinstance(sid, str) and isinstance(sub, str): + user = session.get(self._key('user')) + if not isinstance(user, Mapping) or user.get('sub') != sub: + continue + + # A session can hold several provider identities. Remove only this + # provider's identity and delete the whole session when it was the + # last one, mirroring the local logout behavior. + self._logout_session(session) + if session.invalidated: + await store.delete(session_id) + else: + await store.save(session_id, session, self._session.max_age) + session.commit_indexes() + @staticmethod def _session_of(ctx: HttpContext) -> Session: session = ctx.session @@ -293,6 +463,15 @@ class OIDCMixin(KayaMixin): if isinstance(original_sub, str) and claims.get('sub') != original_sub: raise ValueError('ID token subject mismatch') + refreshed_sid = claims.get('sid') + stored_sid = session.get(self._key('sid')) + if (isinstance(refreshed_sid, str) + and refreshed_sid != stored_sid): + if isinstance(stored_sid, str): + session.remove_index(self._index_name('sid'), stored_sid) + session[self._key('sid')] = refreshed_sid + session.add_index(self._index_name('sid'), refreshed_sid) + if isinstance(access_token, str): session[self._key('access_token')] = access_token if isinstance(id_token, str): diff --git a/packages/kaya-oidc/src/kaya/oidc/_providers.py b/packages/kaya-oidc/src/kaya/oidc/_providers.py index 29a3af5..c230553 100644 --- a/packages/kaya-oidc/src/kaya/oidc/_providers.py +++ b/packages/kaya-oidc/src/kaya/oidc/_providers.py @@ -20,7 +20,9 @@ class OIDCProviders(KayaMixin): Each config's ``provider`` becomes a route segment under ``base_path``: ``{base_path}/{provider}/login``, ``{base_path}/{provider}/callback`` and - ``{base_path}/{provider}/logout``. All providers share the supplied + ``{base_path}/{provider}/logout``. Providers that enable back-channel + logout also get ``{base_path}/{provider}/backchannel-logout``. All + providers share the supplied :class:`~kaya.session.SessionMixin` and namespace their session state by provider. @@ -68,12 +70,17 @@ class OIDCProviders(KayaMixin): name = config.provider if name in self._providers: raise ValueError(f'Duplicate OIDC provider {name!r}') + backchannel_path = ( + f'{self._base_path}/{name}/backchannel-logout' + if config.backchannel_logout_path is not None else None + ) self._providers[name] = OIDCMixin( replace( config, login_path=f'{self._base_path}/{name}/login', callback_path=f'{self._base_path}/{name}/callback', logout_path=f'{self._base_path}/{name}/logout', + backchannel_logout_path=backchannel_path, ), session=session, strategy=strategy, diff --git a/packages/kaya-oidc/tests/test_backchannel_logout.py b/packages/kaya-oidc/tests/test_backchannel_logout.py new file mode 100644 index 0000000..3aac224 --- /dev/null +++ b/packages/kaya-oidc/tests/test_backchannel_logout.py @@ -0,0 +1,427 @@ +import unittest +from time import time +from typing import Any, Optional +from urllib.parse import parse_qs, urlparse + +import httpx +import jwt +from cryptography.hazmat.primitives.asymmetric import rsa +from pwo import async_test +from test_oidc import MockOIDCProvider, MockTransport + +from kaya.core import HttpContext, KayaApp +from kaya.oidc import OIDCConfig, OIDCMixin, OIDCProviders +from kaya.session import InMemorySessionStore, SessionMixin + +_EVENT = 'http://schemas.openid.net/event/backchannel-logout' + + +class BackchannelProvider(MockOIDCProvider): + def __init__(self, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.sid: Optional[str] = None + + def issue_id_token(self, *args: Any, **kwargs: Any) -> str: + token = super().issue_id_token(*args, **kwargs) + if self.sid is None: + return token + payload = jwt.decode(token, options={'verify_signature': False}) + payload['sid'] = self.sid + return jwt.encode( + payload, + self.private_key, + algorithm='RS256', + headers={'kid': self.kid}, + ) + + +class BackchannelLogoutTest(unittest.TestCase): + + def setUp(self) -> None: + self.provider = BackchannelProvider() + self.http_client = httpx.AsyncClient(transport=MockTransport(self.provider)) + self.store = InMemorySessionStore() + self.session = SessionMixin(self.store) + self.oidc = OIDCMixin( + OIDCConfig( + issuer=self.provider.issuer, + client_id='client', + client_secret='secret', + redirect_uri='http://localhost:8000/auth/callback', + backchannel_logout_path='/auth/backchannel-logout', + http_client=self.http_client, + ), + session=self.session, + ) + self.app = KayaApp(mixins=[self.session, self.oidc]) + self.transport = httpx.ASGITransport(app=self.app) + + @self.app.GET('/profile') + @self.oidc.require_auth + async def profile(ctx: HttpContext) -> None: + user = self.oidc.get_user(ctx) + assert user is not None + await ctx.send_str(200, user.sub) + + @self.app.GET('/refresh') + @self.oidc.require_auth + async def refresh(ctx: HttpContext) -> None: + token = await self.oidc.refresh_access_token(ctx.session) + await ctx.send_str(200, token or 'no-token') + + def _client(self) -> httpx.AsyncClient: + return httpx.AsyncClient(transport=self.transport, base_url='http://127.0.0.1:80') + + async def _login(self, client: httpx.AsyncClient, sid: str) -> None: + self.provider.sid = sid + response = await client.get('/auth/login', follow_redirects=False) + self.assertEqual(302, response.status_code) + query = parse_qs(urlparse(response.headers['Location']).query) + self.provider.nonce = query['nonce'][0] + response = await client.get( + '/auth/callback', + params={'code': 'mock-code', 'state': query['state'][0]}, + follow_redirects=False, + ) + self.assertEqual(302, response.status_code) + + def _claims(self, **overrides: Any) -> dict[str, Any]: + claims: dict[str, Any] = { + 'iss': self.provider.issuer, + 'aud': 'client', + 'iat': time(), + 'jti': 'logout-token-id', + 'events': {_EVENT: {}}, + 'sub': self.provider.sub, + } + claims.update(overrides) + return claims + + def _token( + self, + claims: dict[str, Any], + *, + typ: str = 'logout+jwt', + key: Any = None, + algorithm: str = 'RS256', + ) -> str: + return jwt.encode( + claims, + self.provider.private_key if key is None else key, + algorithm=algorithm, + headers={'kid': self.provider.kid, 'typ': typ}, + ) + + async def _post_logout(self, token: str) -> httpx.Response: + async with self._client() as client: + return await client.post( + '/auth/backchannel-logout', + data={'logout_token': token}, + ) + + async def _profile_status(self, client: httpx.AsyncClient) -> int: + response = await client.get('/profile', follow_redirects=False) + return response.status_code + + @async_test + async def test_logout_by_sid_only_invalidates_matching_session(self) -> None: + async with self._client() as first, self._client() as second: + await self._login(first, 'sid-a') + await self._login(second, 'sid-b') + + response = await self._post_logout(self._token(self._claims(sid='sid-a'))) + self.assertEqual(200, response.status_code) + self.assertEqual('no-store', response.headers['Cache-Control']) + self.assertNotIn('Set-Cookie', response.headers) + self.assertEqual(302, await self._profile_status(first)) + self.assertEqual(200, await self._profile_status(second)) + self.assertEqual([], await self.store.find_sessions('oidc:default:sid', 'sid-a')) + self.assertEqual( + [second.cookies['session_id']], + await self.store.find_sessions('oidc:default:sid', 'sid-b'), + ) + + @async_test + async def test_logout_by_sub_invalidates_all_user_sessions(self) -> None: + async with self._client() as first, self._client() as second, self._client() as other: + await self._login(first, 'sid-a') + await self._login(second, 'sid-b') + self.provider.sub = 'bob' + await self._login(other, 'sid-c') + + response = await self._post_logout(self._token(self._claims(sub='user123'))) + self.assertEqual(200, response.status_code) + self.assertEqual(302, await self._profile_status(first)) + self.assertEqual(302, await self._profile_status(second)) + self.assertEqual(200, await self._profile_status(other)) + + @async_test + async def test_sid_logout_verifies_sub_when_both_are_present(self) -> None: + async with self._client() as client: + await self._login(client, 'sid-a') + response = await self._post_logout( + self._token(self._claims(sub='someone-else', sid='sid-a')) + ) + self.assertEqual(200, response.status_code) + self.assertEqual(200, await self._profile_status(client)) + + @async_test + async def test_unknown_session_is_idempotent(self) -> None: + response = await self._post_logout(self._token(self._claims(sid='unknown'))) + self.assertEqual(200, response.status_code) + + @async_test + async def test_invalid_logout_tokens_are_rejected(self) -> None: + other_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + missing_iat = self._claims(sid='sid-a') + del missing_iat['iat'] + missing_jti = self._claims(sid='sid-a') + del missing_jti['jti'] + missing_events = self._claims(sid='sid-a') + del missing_events['events'] + missing_subject = self._claims() + del missing_subject['sub'] + + invalid_tokens = [ + self._token(self._claims(sid='sid-a'), typ='JWT'), + jwt.encode( + self._claims(sid='sid-a'), + None, + algorithm='none', + headers={'typ': 'logout+jwt'}, + ), + self._token(self._claims(sid='sid-a'), key=other_key), + self._token(self._claims(iss='https://evil.local', sid='sid-a')), + self._token(self._claims(aud='other-client', sid='sid-a')), + self._token(missing_iat), + self._token(missing_jti), + self._token(missing_events), + self._token(self._claims(events={'wrong-event': {}}, sid='sid-a')), + self._token(self._claims(events={_EVENT: None}, sid='sid-a')), + self._token(self._claims(nonce='forbidden', sid='sid-a')), + self._token(missing_subject), + self._token(self._claims(exp=time() - 3600, sid='sid-a')), + self._token(self._claims(sid='sid-a'), key='x' * 32, algorithm='HS256'), + ] + + async with self._client() as client: + await self._login(client, 'sid-a') + for token in invalid_tokens: + response = await self._post_logout(token) + self.assertEqual(400, response.status_code) + self.assertEqual(200, await self._profile_status(client)) + + @async_test + async def test_invalid_requests_are_rejected(self) -> None: + async with self._client() as client: + response = await client.post('/auth/backchannel-logout', data={}) + self.assertEqual(400, response.status_code) + response = await client.post( + '/auth/backchannel-logout', + content='logout_token=abc', + headers={'Content-Type': 'text/plain'}, + ) + self.assertEqual(400, response.status_code) + response = await client.post( + '/auth/backchannel-logout', + content='x' * (64 * 1024 + 1), + headers={'Content-Type': 'application/x-www-form-urlencoded'}, + ) + self.assertEqual(400, response.status_code) + + @async_test + async def test_local_logout_removes_indexes(self) -> None: + async with self._client() as client: + await self._login(client, 'sid-a') + self.assertEqual( + [client.cookies['session_id']], + await self.store.find_sessions('oidc:default:sub', 'user123'), + ) + await client.get('/auth/logout', follow_redirects=False) + self.assertEqual([], await self.store.find_sessions('oidc:default:sub', 'user123')) + self.assertEqual([], await self.store.find_sessions('oidc:default:sid', 'sid-a')) + + @async_test + async def test_relogin_replaces_old_sid_index(self) -> None: + async with self._client() as client: + await self._login(client, 'sid-a') + await self._login(client, 'sid-b') + self.assertEqual([], await self.store.find_sessions('oidc:default:sid', 'sid-a')) + self.assertEqual( + [client.cookies['session_id']], + await self.store.find_sessions('oidc:default:sid', 'sid-b'), + ) + + @async_test + async def test_refresh_updates_changed_sid_index(self) -> None: + async with self._client() as client: + await self._login(client, 'sid-a') + self.provider.sid = 'sid-b' + response = await client.get('/refresh') + self.assertEqual(200, response.status_code) + self.assertNotEqual('no-token', response.text) + self.assertEqual([], await self.store.find_sessions('oidc:default:sid', 'sid-a')) + self.assertEqual( + [client.cookies['session_id']], + await self.store.find_sessions('oidc:default:sid', 'sid-b'), + ) + + @async_test + async def test_existing_session_is_backfilled(self) -> None: + async with self._client() as client: + await self._login(client, 'sid-a') + session_id = client.cookies['session_id'] + self.store._data[session_id].pop('oidc:default:indexed') + self.store._session_indexes.pop(session_id, None) + for session_ids in self.store._index_sessions.values(): + session_ids.discard(session_id) + self.assertEqual([], await self.store.find_sessions('oidc:default:sid', 'sid-a')) + + self.assertEqual(200, await self._profile_status(client)) + self.assertEqual( + [session_id], + await self.store.find_sessions('oidc:default:sid', 'sid-a'), + ) + + @async_test + async def test_index_capable_store_is_required(self) -> None: + class LegacyStore(InMemorySessionStore): + supports_indexes = False + + with self.assertRaises(ValueError): + OIDCMixin( + OIDCConfig( + issuer=self.provider.issuer, + client_id='client', + redirect_uri='http://localhost:8000/auth/callback', + backchannel_logout_path='/auth/backchannel-logout', + ), + session=SessionMixin(LegacyStore()), + ) + + +class MultiProviderBackchannelLogoutTest(unittest.TestCase): + + def setUp(self) -> None: + self.google = BackchannelProvider('google', sub='shared-sub', email='g@example.com') + self.github = BackchannelProvider('github', sub='shared-sub', email='h@example.com') + self.store = InMemorySessionStore() + self.session = SessionMixin(self.store) + self.oidc = OIDCProviders( + [ + self._config(self.google, 'google'), + self._config(self.github, 'github'), + ], + session=self.session, + strategy='multi-identity', + ) + self.app = KayaApp(mixins=[self.session, self.oidc]) + self.transport = httpx.ASGITransport(app=self.app) + + @self.app.GET('/google-profile') + @self.oidc.require_auth('google') + async def google_profile(ctx: HttpContext) -> None: + await ctx.send_str(200, 'google') + + @self.app.GET('/github-profile') + @self.oidc.require_auth('github') + async def github_profile(ctx: HttpContext) -> None: + await ctx.send_str(200, 'github') + + @staticmethod + def _config(provider: BackchannelProvider, name: str) -> OIDCConfig: + return OIDCConfig( + issuer=provider.issuer, + client_id='client', + client_secret='secret', + redirect_uri=f'http://localhost:8000/auth/{name}/callback', + provider=name, + backchannel_logout_path=f'/auth/{name}/backchannel-logout', + http_client=httpx.AsyncClient(transport=MockTransport(provider)), + ) + + async def _login( + self, + client: httpx.AsyncClient, + provider: BackchannelProvider, + name: str, + sid: str, + ) -> None: + provider.sid = sid + response = await client.get(f'/auth/{name}/login', follow_redirects=False) + query = parse_qs(urlparse(response.headers['Location']).query) + provider.nonce = query['nonce'][0] + await client.get( + f'/auth/{name}/callback', + params={'code': 'mock-code', 'state': query['state'][0]}, + follow_redirects=False, + ) + + @async_test + async def test_single_identity_eviction_removes_old_provider_indexes(self) -> None: + session = SessionMixin(self.store) + oidc = OIDCProviders( + [ + self._config(self.google, 'google'), + self._config(self.github, 'github'), + ], + session=session, + strategy='single-identity', + ) + app = KayaApp(mixins=[session, oidc]) + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient( + transport=transport, + base_url='http://127.0.0.1:80', + ) as client: + await self._login(client, self.google, 'google', 'google-sid') + self.assertEqual( + [client.cookies['session_id']], + await self.store.find_sessions('oidc:google:sid', 'google-sid'), + ) + await self._login(client, self.github, 'github', 'github-sid') + self.assertEqual( + [], + await self.store.find_sessions('oidc:google:sid', 'google-sid'), + ) + self.assertEqual( + [client.cookies['session_id']], + await self.store.find_sessions('oidc:github:sid', 'github-sid'), + ) + + @async_test + async def test_provider_logout_is_isolated(self) -> None: + async with httpx.AsyncClient( + transport=self.transport, + base_url='http://127.0.0.1:80', + ) as client: + await self._login(client, self.google, 'google', 'google-sid') + await self._login(client, self.github, 'github', 'github-sid') + token = jwt.encode( + { + 'iss': self.google.issuer, + 'aud': 'client', + 'iat': time(), + 'jti': 'google-logout', + 'events': {_EVENT: {}}, + 'sub': 'shared-sub', + 'sid': 'google-sid', + }, + self.google.private_key, + algorithm='RS256', + headers={'kid': self.google.kid, 'typ': 'logout+jwt'}, + ) + response = await client.post( + '/auth/google/backchannel-logout', + data={'logout_token': token}, + ) + self.assertEqual(200, response.status_code) + + response = await client.get('/google-profile', follow_redirects=False) + self.assertEqual(302, response.status_code) + response = await client.get('/github-profile', follow_redirects=False) + self.assertEqual(200, response.status_code) + + +if __name__ == '__main__': + unittest.main() diff --git a/packages/kaya-session-memcache/README.md b/packages/kaya-session-memcache/README.md index cb363b9..e5a5a6f 100644 --- a/packages/kaya-session-memcache/README.md +++ b/packages/kaya-session-memcache/README.md @@ -32,11 +32,25 @@ and slides on each access when the session mixin passes a `max_age`. TTLs larger than 30 days are automatically converted to absolute Unix timestamps, as required by the memcached protocol. +## Session indexes + +`MemcacheSessionStore` supports `Session.add_index()`, +`SessionStore.find_sessions()`, and `delete_by_index()`. Reverse memberships +are stored as sets and updated with memcached CAS operations. Bucket keys use +a SHA-256 digest of the index name and value, and stale members are pruned +during lookups. + +Memcached's item-size limit (normally 1 MB) applies to each index bucket. Use +the Redis store when one index value can have a very large number of active +sessions. + ## Serialization Session data is serialized with `pickle` by default, so arbitrary Python -objects can be stored. A different serializer can be plugged in via the -`dumps`/`loads` arguments: +objects can be stored. The serialized value is an envelope containing the +session dictionary and index metadata; legacy plain-dict values are still +accepted. A different serializer can be plugged in via the `dumps`/`loads` +arguments: ```python import json diff --git a/packages/kaya-session-memcache/src/kaya/session/memcache/_store.py b/packages/kaya-session-memcache/src/kaya/session/memcache/_store.py index d7df8a3..55eac4d 100644 --- a/packages/kaya-session-memcache/src/kaya/session/memcache/_store.py +++ b/packages/kaya-session-memcache/src/kaya/session/memcache/_store.py @@ -1,12 +1,15 @@ +import hashlib import pickle from time import time -from typing import Any, Callable, Optional +from typing import Any, Callable, Collection, Optional, Sequence import aiomcache -from kaya.session import Session, SessionStore +from kaya.session import IndexKey, Session, SessionStore _THIRTY_DAYS = 30 * 24 * 60 * 60 +_INDEX_GRACE_PERIOD = 60 +_CAS_RETRIES = 5 def _exptime(max_age: int, clock: Callable[[], float]) -> int: @@ -36,8 +39,14 @@ class MemcacheSessionStore(SessionStore): Python objects can be stored. Custom serializers can be plugged in via the ``dumps``/``loads`` arguments. Only connect this store to a trusted memcached server, as pickle deserialization of untrusted data is unsafe. + + Reverse index buckets are pickled sets updated with memcached CAS + operations. Memcached's item-size limit applies to each bucket; Redis is a + better fit when one index value can have a very large number of sessions. """ + supports_indexes = True + def __init__( self, client: aiomcache.Client, @@ -55,6 +64,69 @@ class MemcacheSessionStore(SessionStore): def _key(self, session_id: str) -> bytes: return f"{self._prefix}{session_id}".encode() + def _index_key(self, name: str, value: str) -> bytes: + digest = hashlib.sha256(f"{name}\0{value}".encode()).hexdigest() + return f"{self._prefix}idx:{digest}".encode() + + def _index_exptime(self, max_age: Optional[int]) -> int: + if max_age is None: + return 0 + return _exptime(max_age + _INDEX_GRACE_PERIOD, self._clock) + + @staticmethod + def _decode_bucket(payload: bytes) -> tuple[set[str], Optional[int]]: + members, max_age = pickle.loads(payload) + return set(members), max_age + + async def _update_bucket( + self, + index_key: bytes, + *, + add: Collection[str] = (), + remove: Collection[str] = (), + max_age: Optional[int] = None, + ) -> None: + for _ in range(_CAS_RETRIES): + payload, cas_token = await self._client.gets(index_key) + if cas_token is None: + if not add: + return + members = set(add) + created = await self._client.add( + index_key, + pickle.dumps((members, max_age)), + exptime=self._index_exptime(max_age), + ) + if created: + return + continue + + assert payload is not None + members, stored_max_age = self._decode_bucket(payload) + updated = (members | set(add)) - set(remove) + if updated == members: + if not add: + return + if await self._client.touch( + index_key, + self._index_exptime(max_age), + ): + return + # The bucket may have expired between gets() and touch(). + # Retry so it can be recreated instead of losing the index. + continue + + effective_max_age = max_age if add else stored_max_age + if await self._client.cas( + index_key, + pickle.dumps((updated, effective_max_age)), + cas_token, + exptime=self._index_exptime(effective_max_age), + ): + return + + raise RuntimeError('Could not update session index after repeated CAS conflicts') + async def load(self, session_id: str, max_age: Optional[int] = None) -> Optional[Session]: key = self._key(session_id) payload = await self._client.get(key) @@ -65,12 +137,63 @@ class MemcacheSessionStore(SessionStore): await self._client.touch(key, _exptime(max_age, self._clock)) data = self._loads(payload) - return Session(session_id, data) + return Session.from_payload(session_id, data) async def save(self, session_id: str, session: Session, max_age: Optional[int] = None) -> None: - payload = self._dumps(dict(session)) + # Index first: if the session write subsequently fails, a stale index + # entry is harmless and will be pruned, while a live unindexed session + # would be invisible to index-based invalidation. + for name, value in session.indexes: + await self._update_bucket( + self._index_key(name, value), + add={session_id}, + max_age=max_age, + ) + for name, value in session.removed_indexes: + await self._update_bucket( + self._index_key(name, value), + remove={session_id}, + ) + + payload = self._dumps(session.to_payload()) exptime = _exptime(max_age, self._clock) if max_age is not None else 0 await self._client.set(self._key(session_id), payload, exptime=exptime) async def delete(self, session_id: str) -> None: - await self._client.delete(self._key(session_id)) + key = self._key(session_id) + payload = await self._client.get(key) + indexes: set[IndexKey] = set() + if payload is not None: + try: + indexes = set(Session.from_payload( + session_id, + self._loads(payload), + ).indexes) + except Exception: + # A corrupt payload must not prevent session deletion. + pass + + await self._client.delete(key) + for name, value in indexes: + await self._update_bucket( + self._index_key(name, value), + remove={session_id}, + ) + + async def find_sessions(self, name: str, value: str) -> Sequence[str]: + index_key = self._index_key(name, value) + payload = await self._client.get(index_key) + if payload is None: + return [] + + members, _ = self._decode_bucket(payload) + live: list[str] = [] + stale: list[str] = [] + for session_id in members: + if await self._client.get(self._key(session_id)) is None: + stale.append(session_id) + else: + live.append(session_id) + if stale: + await self._update_bucket(index_key, remove=stale) + return live diff --git a/packages/kaya-session-memcache/tests/test_indexes.py b/packages/kaya-session-memcache/tests/test_indexes.py new file mode 100644 index 0000000..5057d18 --- /dev/null +++ b/packages/kaya-session-memcache/tests/test_indexes.py @@ -0,0 +1,122 @@ +import pickle +import unittest + +from pwo import async_test +from test_store import FakeClock, StubMemcacheClient + +from kaya.session import Session +from kaya.session.memcache import MemcacheSessionStore + + +class MemcacheSessionIndexTest(unittest.TestCase): + + def setUp(self) -> None: + self.clock = FakeClock() + self.client = StubMemcacheClient(self.clock) + self.store = MemcacheSessionStore(self.client, clock=self.clock) # type: ignore[arg-type] + + @async_test + async def test_save_load_and_find(self) -> None: + self.assertTrue(self.store.supports_indexes) + await self.store.save('abc', Session('abc', {'user': 'alice'}, indexes={('sub', 'alice')})) + loaded = await self.store.load('abc') + assert loaded is not None + self.assertEqual({'user': 'alice'}, dict(loaded)) + self.assertEqual(frozenset({('sub', 'alice')}), loaded.indexes) + self.assertEqual(['abc'], await self.store.find_sessions('sub', 'alice')) + + @async_test + async def test_rotation_preserves_index(self) -> None: + await self.store.save('old', Session('old', indexes={('sub', 'alice')})) + loaded = await self.store.load('old') + assert loaded is not None + loaded.set_id('new') + await self.store.delete('old') + await self.store.save('new', loaded) + self.assertEqual(['new'], await self.store.find_sessions('sub', 'alice')) + + @async_test + async def test_delete_cleans_reverse_index(self) -> None: + await self.store.save('abc', Session('abc', indexes={('sub', 'alice')})) + await self.store.delete('abc') + self.assertEqual([], await self.store.find_sessions('sub', 'alice')) + + @async_test + async def test_delete_by_index_deletes_all_sessions(self) -> None: + await self.store.save('a', Session('a', indexes={('sub', 'alice')})) + await self.store.save('b', Session('b', indexes={('sub', 'alice')})) + await self.store.save('c', Session('c', indexes={('sub', 'bob')})) + self.assertEqual(2, await self.store.delete_by_index('sub', 'alice')) + self.assertIsNone(await self.store.load('a')) + self.assertIsNone(await self.store.load('b')) + self.assertIsNotNone(await self.store.load('c')) + + @async_test + async def test_stale_member_is_pruned(self) -> None: + await self.store.save('abc', Session('abc', indexes={('sub', 'alice')})) + await self.client.delete(b'kaya:session:abc') + self.assertEqual([], await self.store.find_sessions('sub', 'alice')) + index_key = self.store._index_key('sub', 'alice') + payload = await self.client.get(index_key) + assert payload is not None + members, _ = pickle.loads(payload) + self.assertEqual(set(), members) + + @async_test + async def test_index_ttl_tracks_session_max_age(self) -> None: + await self.store.save( + 'abc', + Session('abc', indexes={('sub', 'alice')}), + max_age=60, + ) + index_key = self.store._index_key('sub', 'alice') + self.assertIn(index_key, self.client._expires) + self.clock.advance(61) + await self.store.save( + 'abc', + Session('abc', indexes={('sub', 'alice')}), + max_age=60, + ) + self.assertIsNotNone(await self.client.get(index_key)) + + @async_test + async def test_legacy_payload_is_upgraded_on_save(self) -> None: + await self.client.set(b'kaya:session:abc', pickle.dumps({'foo': 'bar'})) + loaded = await self.store.load('abc') + assert loaded is not None + self.assertEqual({'foo': 'bar'}, dict(loaded)) + self.assertEqual(frozenset(), loaded.indexes) + + loaded.add_index('sub', 'alice') + await self.store.save('abc', loaded) + raw = await self.client.get(b'kaya:session:abc') + assert raw is not None + payload = pickle.loads(raw) + self.assertEqual(1, payload['__kaya_session_v1__']) + self.assertEqual({'foo': 'bar'}, payload['data']) + self.assertEqual(['abc'], await self.store.find_sessions('sub', 'alice')) + + @async_test + async def test_cas_conflict_is_retried(self) -> None: + await self.store.save('existing', Session('existing', indexes={('sub', 'alice')})) + original_cas = self.client.cas + attempts = 0 + + async def fail_once(key: bytes, value: bytes, cas_token: int, exptime: int = 0) -> bool: + nonlocal attempts + attempts += 1 + if attempts == 1: + return False + return await original_cas(key, value, cas_token, exptime) + + self.client.cas = fail_once # type: ignore[method-assign] + await self.store.save('abc', Session('abc', indexes={('sub', 'alice')})) + self.assertEqual( + {'existing', 'abc'}, + set(await self.store.find_sessions('sub', 'alice')), + ) + self.assertGreaterEqual(attempts, 2) + + +if __name__ == '__main__': + unittest.main() diff --git a/packages/kaya-session-memcache/tests/test_store.py b/packages/kaya-session-memcache/tests/test_store.py index 9f07482..12abe53 100644 --- a/packages/kaya-session-memcache/tests/test_store.py +++ b/packages/kaya-session-memcache/tests/test_store.py @@ -28,6 +28,8 @@ class StubMemcacheClient: self._clock = clock self._data: dict[bytes, bytes] = {} self._expires: dict[bytes, float] = {} + self._cas: dict[bytes, int] = {} + self._next_cas = 1 self.set_exptimes: list[int] = [] self.touch_exptimes: list[int] = [] @@ -42,19 +44,46 @@ class StubMemcacheClient: return None return self._data.get(key) - async def set(self, key: bytes, value: bytes, exptime: int = 0) -> bool: - self.set_exptimes.append(exptime) + def _store(self, key: bytes, value: bytes, exptime: int) -> None: self._data[key] = value + self._cas[key] = self._next_cas + self._next_cas += 1 if exptime > 0: self._expires[key] = self._clock() + exptime else: self._expires.pop(key, None) + + async def set(self, key: bytes, value: bytes, exptime: int = 0) -> bool: + self.set_exptimes.append(exptime) + self._store(key, value, exptime) + return True + + async def add(self, key: bytes, value: bytes, exptime: int = 0) -> bool: + if self._expired(key): + self._data.pop(key, None) + self._expires.pop(key, None) + if key in self._data: + return False + self._store(key, value, exptime) + return True + + async def gets(self, key: bytes) -> tuple[Optional[bytes], Optional[int]]: + value = await self.get(key) + if value is None: + return None, None + return value, self._cas[key] + + async def cas(self, key: bytes, value: bytes, cas_token: int, exptime: int = 0) -> bool: + if self._expired(key) or self._cas.get(key) != cas_token: + return False + self._store(key, value, exptime) return True async def delete(self, key: bytes) -> bool: existed = key in self._data self._data.pop(key, None) self._expires.pop(key, None) + self._cas.pop(key, None) return existed async def touch(self, key: bytes, exptime: int) -> bool: @@ -169,7 +198,11 @@ class MemcacheSessionStoreTest(unittest.TestCase): clock=self.clock, ) await store.save('abc', Session('abc', {'foo': 'bar'})) - self.assertEqual(b'{"foo": "bar"}', self.client._data[b'kaya:session:abc']) + self.assertEqual({ + '__kaya_session_v1__': 1, + 'data': {'foo': 'bar'}, + 'indexes': [], + }, json.loads(self.client._data[b'kaya:session:abc'])) loaded = await store.load('abc') self.assertIsNotNone(loaded) assert loaded is not None diff --git a/packages/kaya-session-redis/README.md b/packages/kaya-session-redis/README.md index b962892..663385c 100644 --- a/packages/kaya-session-redis/README.md +++ b/packages/kaya-session-redis/README.md @@ -30,11 +30,21 @@ Sessions are stored under keys with the prefix `kaya:session:` (configurable via the `prefix` argument). Server-side expiry uses Redis key TTLs and slides on each access when the session mixin passes a `max_age`. +## Session indexes + +`RedisSessionStore` supports `Session.add_index()`, +`SessionStore.find_sessions()`, and `delete_by_index()`. Reverse memberships +are Redis sets updated in the same transactional pipeline as session saves. +They are removed on deletion, and stale members are pruned during lookups. +Bucket keys use a SHA-256 digest of the index name and value. + ## Serialization Session data is serialized with `pickle` by default, so arbitrary Python -objects can be stored. A different serializer can be plugged in via the -`dumps`/`loads` arguments: +objects can be stored. The serialized value is an envelope containing the +session dictionary and index metadata; legacy plain-dict values are still +accepted. A different serializer can be plugged in via the `dumps`/`loads` +arguments: ```python import json diff --git a/packages/kaya-session-redis/src/kaya/session/redis/_store.py b/packages/kaya-session-redis/src/kaya/session/redis/_store.py index 839c111..2cf1984 100644 --- a/packages/kaya-session-redis/src/kaya/session/redis/_store.py +++ b/packages/kaya-session-redis/src/kaya/session/redis/_store.py @@ -1,9 +1,10 @@ +import hashlib import pickle -from typing import Any, Callable, Optional, cast +from typing import Any, Callable, Optional, Sequence, cast from redis.asyncio import Redis -from kaya.session import Session, SessionStore +from kaya.session import IndexKey, Session, SessionStore class RedisSessionStore(SessionStore): @@ -20,8 +21,14 @@ class RedisSessionStore(SessionStore): Python objects can be stored. Custom serializers can be plugged in via the ``dumps``/``loads`` arguments. Only connect this store to a trusted Redis server, as pickle deserialization of untrusted data is unsafe. + + Reverse session indexes are stored as Redis sets. They are updated in the + same pipeline as session saves, cleaned on deletion, and pruned when stale + memberships are encountered during a lookup. """ + supports_indexes = True + def __init__( self, client: Redis, @@ -37,6 +44,10 @@ class RedisSessionStore(SessionStore): def _key(self, session_id: str) -> str: return f"{self._prefix}{session_id}" + def _index_key(self, name: str, value: str) -> str: + digest = hashlib.sha256(f"{name}\0{value}".encode()).hexdigest() + return f"{self._prefix}idx:{digest}" + async def load(self, session_id: str, max_age: Optional[int] = None) -> Optional[Session]: key = self._key(session_id) payload = await self._client.get(key) @@ -47,11 +58,65 @@ class RedisSessionStore(SessionStore): await self._client.expire(key, max_age) data = self._loads(cast(bytes, payload)) - return Session(session_id, data) + return Session.from_payload(session_id, data) async def save(self, session_id: str, session: Session, max_age: Optional[int] = None) -> None: - payload = self._dumps(dict(session)) - await self._client.set(self._key(session_id), payload, ex=max_age) + payload = self._dumps(session.to_payload()) + pipeline = self._client.pipeline(transaction=True) + pipeline.set(self._key(session_id), payload, ex=max_age) + for name, value in session.indexes: + index_key = self._index_key(name, value) + pipeline.sadd(index_key, session_id) + if max_age is None: + pipeline.persist(index_key) + else: + pipeline.expire(index_key, max_age + 60) + for name, value in session.removed_indexes: + pipeline.srem(self._index_key(name, value), session_id) + await pipeline.execute() async def delete(self, session_id: str) -> None: - await self._client.delete(self._key(session_id)) + key = self._key(session_id) + payload = await self._client.get(key) + indexes: set[IndexKey] = set() + if payload is not None: + try: + indexes = set(Session.from_payload( + session_id, + self._loads(cast(bytes, payload)), + ).indexes) + except Exception: + # A corrupt payload must not prevent session deletion. + pass + + pipeline = self._client.pipeline(transaction=True) + pipeline.delete(key) + for name, value in indexes: + pipeline.srem(self._index_key(name, value), session_id) + await pipeline.execute() + + async def find_sessions(self, name: str, value: str) -> Sequence[str]: + index_key = self._index_key(name, value) + members = await self._client.smembers(index_key) + session_ids = [ + member.decode() if isinstance(member, bytes) else str(member) + for member in members + ] + if not session_ids: + return [] + + pipeline = self._client.pipeline(transaction=False) + for session_id in session_ids: + pipeline.exists(self._key(session_id)) + exists = await pipeline.execute() + + live: list[str] = [] + stale: list[str] = [] + for session_id, found in zip(session_ids, exists): + if found: + live.append(session_id) + else: + stale.append(session_id) + if stale: + await self._client.srem(index_key, *stale) + return live diff --git a/packages/kaya-session-redis/tests/test_indexes.py b/packages/kaya-session-redis/tests/test_indexes.py new file mode 100644 index 0000000..497398f --- /dev/null +++ b/packages/kaya-session-redis/tests/test_indexes.py @@ -0,0 +1,94 @@ +import pickle +import unittest + +import fakeredis.aioredis +from pwo import async_test + +from kaya.session import Session +from kaya.session.redis import RedisSessionStore + + +class RedisSessionIndexTest(unittest.TestCase): + + def setUp(self) -> None: + self.client = fakeredis.aioredis.FakeRedis() + self.store = RedisSessionStore(self.client) + + @async_test + async def test_save_load_and_find(self) -> None: + self.assertTrue(self.store.supports_indexes) + await self.store.save('abc', Session('abc', {'user': 'alice'}, indexes={('sub', 'alice')})) + loaded = await self.store.load('abc') + assert loaded is not None + self.assertEqual({'user': 'alice'}, dict(loaded)) + self.assertEqual(frozenset({('sub', 'alice')}), loaded.indexes) + self.assertEqual(['abc'], await self.store.find_sessions('sub', 'alice')) + + @async_test + async def test_rotation_preserves_index(self) -> None: + await self.store.save('old', Session('old', indexes={('sub', 'alice')})) + loaded = await self.store.load('old') + assert loaded is not None + loaded.set_id('new') + await self.store.delete('old') + await self.store.save('new', loaded) + self.assertEqual(['new'], await self.store.find_sessions('sub', 'alice')) + + @async_test + async def test_delete_cleans_reverse_index(self) -> None: + await self.store.save('abc', Session('abc', indexes={('sub', 'alice')})) + index_key = self.store._index_key('sub', 'alice') + self.assertEqual({b'abc'}, await self.client.smembers(index_key)) + await self.store.delete('abc') + self.assertEqual(set(), await self.client.smembers(index_key)) + + @async_test + async def test_delete_by_index_deletes_all_sessions(self) -> None: + await self.store.save('a', Session('a', indexes={('sub', 'alice')})) + await self.store.save('b', Session('b', indexes={('sub', 'alice')})) + await self.store.save('c', Session('c', indexes={('sub', 'bob')})) + self.assertEqual(2, await self.store.delete_by_index('sub', 'alice')) + self.assertIsNone(await self.store.load('a')) + self.assertIsNone(await self.store.load('b')) + self.assertIsNotNone(await self.store.load('c')) + + @async_test + async def test_stale_member_is_pruned(self) -> None: + await self.store.save('abc', Session('abc', indexes={('sub', 'alice')})) + await self.client.delete('kaya:session:abc') + self.assertEqual([], await self.store.find_sessions('sub', 'alice')) + index_key = self.store._index_key('sub', 'alice') + self.assertEqual(set(), await self.client.smembers(index_key)) + + @async_test + async def test_index_ttl_tracks_session_max_age(self) -> None: + await self.store.save( + 'abc', + Session('abc', indexes={('sub', 'alice')}), + max_age=60, + ) + index_key = self.store._index_key('sub', 'alice') + ttl = await self.client.ttl(index_key) + self.assertGreater(ttl, 60) + self.assertLessEqual(ttl, 120) + + @async_test + async def test_legacy_payload_is_upgraded_on_save(self) -> None: + await self.client.set('kaya:session:abc', pickle.dumps({'foo': 'bar'})) + loaded = await self.store.load('abc') + assert loaded is not None + self.assertEqual({'foo': 'bar'}, dict(loaded)) + self.assertEqual(frozenset(), loaded.indexes) + + loaded.add_index('sub', 'alice') + await self.store.save('abc', loaded) + raw = await self.client.get('kaya:session:abc') + assert raw is not None + payload = pickle.loads(raw) + self.assertEqual(1, payload['__kaya_session_v1__']) + self.assertEqual({'foo': 'bar'}, payload['data']) + self.assertEqual(['abc'], await self.store.find_sessions('sub', 'alice')) + + +if __name__ == '__main__': + unittest.main() diff --git a/packages/kaya-session-redis/tests/test_store.py b/packages/kaya-session-redis/tests/test_store.py index acfb8b6..b5253da 100644 --- a/packages/kaya-session-redis/tests/test_store.py +++ b/packages/kaya-session-redis/tests/test_store.py @@ -89,7 +89,13 @@ class RedisSessionStoreTest(unittest.TestCase): loads=lambda b: json.loads(b.decode('utf-8')), ) await store.save('abc', Session('abc', {'foo': 'bar'})) - self.assertEqual(b'{"foo": "bar"}', await self.client.get('kaya:session:abc')) + raw = await self.client.get('kaya:session:abc') + assert raw is not None + self.assertEqual({ + '__kaya_session_v1__': 1, + 'data': {'foo': 'bar'}, + 'indexes': [], + }, json.loads(raw)) loaded = await store.load('abc') self.assertIsNotNone(loaded) assert loaded is not None diff --git a/packages/kaya-session/README.md b/packages/kaya-session/README.md index 86b0e5a..fbdb2c9 100644 --- a/packages/kaya-session/README.md +++ b/packages/kaya-session/README.md @@ -66,6 +66,37 @@ session and creates a fresh empty one. Set `max_age=None` to disable server-side expiry (and the `Max-Age` cookie attribute) entirely. +## Session indexes + +Authentication layers can attach reverse-index metadata to a session without +adding it to the user-visible session dictionary: + +```python +ctx.session.add_index('oidc:default:sub', 'user-123') +``` + +Index-capable stores persist these memberships on `save()` and maintain them +across session ID rotation and deletion: + +```python +session_ids = await store.find_sessions('oidc:default:sub', 'user-123') +deleted = await store.delete_by_index('oidc:default:sub', 'user-123') +``` + +Use `session.remove_index(name, value)` to remove a membership while keeping +the session. `SessionMixin.store` exposes the backing store for features such +as back-channel logout. + +`InMemorySessionStore`, `RedisSessionStore`, and `MemcacheSessionStore` set +`supports_indexes = True`. Custom stores can opt in by setting that flag, +implementing `find_sessions()`, maintaining reverse memberships on `save()`, +and removing them on `delete()`. `delete_by_index()` has a generic default +implementation based on `find_sessions()` and `delete()`. + +Remote stores serialize an envelope containing both the session dictionary and +its index metadata. Existing plain-dict payloads remain readable and are +upgraded on the next save. + ## Features - `Session`: dict-like session object with modification tracking @@ -74,6 +105,7 @@ attribute) entirely. - `SessionMixin`: composable Kaya mixin managing session cookies and persistence - Session ID regeneration (`session.regenerate_id()`) and invalidation (`session.invalidate()`) for authentication layers +- Reverse session indexes for looking up and invalidating sessions by metadata - WebSocket support: the session is exposed as `ws.session` in websocket handlers, loaded at connect time and persisted on close diff --git a/packages/kaya-session/src/kaya/session/__init__.py b/packages/kaya-session/src/kaya/session/__init__.py index e302bb4..d26ce91 100644 --- a/packages/kaya-session/src/kaya/session/__init__.py +++ b/packages/kaya-session/src/kaya/session/__init__.py @@ -3,12 +3,13 @@ from pkgutil import extend_path __path__ = extend_path(__path__, __name__) from ._mixin import SessionMixin -from ._session import Session +from ._session import IndexKey, Session from ._store import InMemorySessionStore, SessionStore __all__ = [ 'InMemorySessionStore', + 'IndexKey', 'Session', 'SessionMixin', 'SessionStore', diff --git a/packages/kaya-session/src/kaya/session/_mixin.py b/packages/kaya-session/src/kaya/session/_mixin.py index 0f14abb..ea1db3f 100644 --- a/packages/kaya-session/src/kaya/session/_mixin.py +++ b/packages/kaya-session/src/kaya/session/_mixin.py @@ -183,6 +183,16 @@ class SessionMixin(KayaMixin): self._secure = secure self._samesite = samesite + @property + def store(self) -> SessionStore: + """Return the backing session store.""" + return self._store + + @property + def max_age(self) -> Optional[int]: + """Return the idle timeout used when persisting sessions.""" + return self._max_age + def apply(self, app: KayaApp) -> None: app.add_before_request_hook(self._before_request) app.add_after_request_hook(self._after_request) @@ -273,3 +283,4 @@ class SessionMixin(KayaMixin): if session.id is not None: await self._store.save(session.id, session, self._max_age) + session.commit_indexes() diff --git a/packages/kaya-session/src/kaya/session/_session.py b/packages/kaya-session/src/kaya/session/_session.py index 8633dc9..15d91d7 100644 --- a/packages/kaya-session/src/kaya/session/_session.py +++ b/packages/kaya-session/src/kaya/session/_session.py @@ -1,4 +1,10 @@ -from typing import Any, Iterator, Mapping, MutableMapping, Optional +from typing import Any, Iterable, Iterator, Mapping, MutableMapping, Optional + + +IndexKey = tuple[str, str] + +_PAYLOAD_MARKER = '__kaya_session_v1__' +_PAYLOAD_VERSION = 1 class Session(MutableMapping[str, Any]): @@ -9,13 +15,21 @@ class Session(MutableMapping[str, Any]): it. """ - def __init__(self, session_id: Optional[str] = None, data: Optional[Mapping[str, Any]] = None) -> None: + def __init__( + self, + session_id: Optional[str] = None, + data: Optional[Mapping[str, Any]] = None, + indexes: Iterable[IndexKey] = (), + ) -> None: self._id: Optional[str] = session_id self._old_id: Optional[str] = None self._data: dict[str, Any] = dict(data) if data else {} self._modified: bool = False self._regenerate: bool = False self._invalidated: bool = False + self._persisted_indexes: set[IndexKey] = set(indexes) + self._added_indexes: set[IndexKey] = set() + self._removed_indexes: set[IndexKey] = set() @property def id(self) -> Optional[str]: @@ -60,6 +74,75 @@ class Session(MutableMapping[str, Any]): def mark_modified(self) -> None: self._modified = True + def add_index(self, name: str, value: str) -> None: + """Index this session under ``(name, value)`` when it is saved. + + Index-capable stores can use this metadata to find and invalidate + sessions without knowing their IDs. Adding an already persisted index + is harmless and does not create a duplicate membership. + """ + index = (name, value) + self._removed_indexes.discard(index) + if index not in self._persisted_indexes: + self._added_indexes.add(index) + self._modified = True + + def remove_index(self, name: str, value: str) -> None: + """Remove an index membership when this session is saved.""" + index = (name, value) + if index in self._added_indexes: + self._added_indexes.discard(index) + elif index in self._persisted_indexes: + self._removed_indexes.add(index) + self._modified = True + + @property + def indexes(self) -> frozenset[IndexKey]: + """Return the index memberships that should be persisted.""" + return frozenset( + (self._persisted_indexes - self._removed_indexes) | self._added_indexes + ) + + @property + def added_indexes(self) -> frozenset[IndexKey]: + """Return index memberships added since the last successful save.""" + return frozenset(self._added_indexes) + + @property + def removed_indexes(self) -> frozenset[IndexKey]: + """Return index memberships removed since the last successful save.""" + return frozenset(self._removed_indexes) + + def commit_indexes(self) -> None: + """Mark the current index memberships as persisted.""" + self._persisted_indexes = set(self.indexes) + self._added_indexes.clear() + self._removed_indexes.clear() + + def to_payload(self) -> dict[str, Any]: + """Return the serializable session payload used by remote stores.""" + return { + _PAYLOAD_MARKER: _PAYLOAD_VERSION, + 'data': dict(self), + 'indexes': sorted(self.indexes), + } + + @classmethod + def from_payload(cls, session_id: str, payload: Mapping[str, Any]) -> 'Session': + """Build a session from an envelope or a legacy plain-data payload.""" + data = payload.get('data') + raw_indexes = payload.get('indexes') + if (payload.get(_PAYLOAD_MARKER) == _PAYLOAD_VERSION + and isinstance(data, Mapping) + and isinstance(raw_indexes, (list, tuple))): + indexes: set[IndexKey] = set() + for item in raw_indexes: + if (isinstance(item, (list, tuple)) and len(item) == 2 + and isinstance(item[0], str) and isinstance(item[1], str)): + indexes.add((item[0], item[1])) + return cls(session_id, data, indexes=indexes) + return cls(session_id, payload) + def __getitem__(self, key: str) -> Any: return self._data[key] diff --git a/packages/kaya-session/src/kaya/session/_store.py b/packages/kaya-session/src/kaya/session/_store.py index 1648afe..7e6ffe2 100644 --- a/packages/kaya-session/src/kaya/session/_store.py +++ b/packages/kaya-session/src/kaya/session/_store.py @@ -1,13 +1,21 @@ from abc import ABC, abstractmethod from secrets import token_urlsafe from time import monotonic -from typing import Any, Callable, Optional +from typing import Any, Callable, ClassVar, Optional, Sequence -from ._session import Session +from ._session import IndexKey, Session class SessionStore(ABC): - """Pluggable backend for session persistence.""" + """Pluggable backend for session persistence. + + Stores with ``supports_indexes = True`` also maintain reverse indexes for + :class:`~kaya.session.Session` index metadata. They must add the session ID + to every effective index on ``save()``, apply pending removals, and remove + all of a session's index memberships on ``delete()``. + """ + + supports_indexes: ClassVar[bool] = False @abstractmethod async def load(self, session_id: str, max_age: Optional[int] = None) -> Optional[Session]: @@ -33,6 +41,23 @@ class SessionStore(ABC): """Remove the session from the store.""" pass + async def find_sessions(self, name: str, value: str) -> Sequence[str]: + """Return live session IDs indexed under ``(name, value)``. + + Only available on stores whose ``supports_indexes`` flag is true. + Implementations should prune stale index memberships encountered here. + """ + raise NotImplementedError( + f'{type(self).__name__} does not support session indexes' + ) + + async def delete_by_index(self, name: str, value: str) -> int: + """Delete every live session indexed under ``(name, value)``.""" + session_ids = await self.find_sessions(name, value) + for session_id in session_ids: + await self.delete(session_id) + return len(session_ids) + def new_session_id(self) -> str: """Return a new opaque session identifier. @@ -51,10 +76,26 @@ class InMemorySessionStore(SessionStore): inactivity. Active sessions slide the expiry window on each access. """ + supports_indexes = True + def __init__(self, clock: Callable[[], float] = monotonic) -> None: self._clock = clock self._data: dict[str, dict[str, Any]] = {} self._expires: dict[str, float] = {} + self._session_indexes: dict[str, set[IndexKey]] = {} + self._index_sessions: dict[IndexKey, set[str]] = {} + + def _drop_session(self, session_id: str) -> None: + self._data.pop(session_id, None) + self._expires.pop(session_id, None) + indexes = self._session_indexes.pop(session_id, set()) + for index in indexes: + session_ids = self._index_sessions.get(index) + if session_ids is None: + continue + session_ids.discard(session_id) + if not session_ids: + self._index_sessions.pop(index, None) async def load(self, session_id: str, max_age: Optional[int] = None) -> Optional[Session]: data = self._data.get(session_id) @@ -64,14 +105,17 @@ class InMemorySessionStore(SessionStore): expires = self._expires.get(session_id) now = self._clock() if expires is not None and now > expires: - self._data.pop(session_id, None) - self._expires.pop(session_id, None) + self._drop_session(session_id) return None if max_age is not None and expires is not None: self._expires[session_id] = now + max_age - return Session(session_id, data) + return Session( + session_id, + data, + indexes=self._session_indexes.get(session_id, ()), + ) async def save(self, session_id: str, session: Session, max_age: Optional[int] = None) -> None: self._data[session_id] = dict(session) @@ -80,6 +124,34 @@ class InMemorySessionStore(SessionStore): else: self._expires.pop(session_id, None) + for index in session.removed_indexes: + session_ids = self._index_sessions.get(index) + if session_ids is None: + continue + session_ids.discard(session_id) + if not session_ids: + self._index_sessions.pop(index, None) + + indexes = set(session.indexes) + self._session_indexes[session_id] = indexes + for index in indexes: + self._index_sessions.setdefault(index, set()).add(session_id) + async def delete(self, session_id: str) -> None: - self._data.pop(session_id, None) - self._expires.pop(session_id, None) + self._drop_session(session_id) + + async def find_sessions(self, name: str, value: str) -> Sequence[str]: + session_ids = self._index_sessions.get((name, value)) + if session_ids is None: + return [] + + now = self._clock() + live: list[str] = [] + for session_id in list(session_ids): + expires = self._expires.get(session_id) + if (session_id not in self._data + or (expires is not None and now > expires)): + self._drop_session(session_id) + else: + live.append(session_id) + return live diff --git a/packages/kaya-session/tests/test_indexes.py b/packages/kaya-session/tests/test_indexes.py new file mode 100644 index 0000000..4c77f9f --- /dev/null +++ b/packages/kaya-session/tests/test_indexes.py @@ -0,0 +1,169 @@ +import unittest + +import httpx +from pwo import async_test + +from kaya.core import HttpContext, KayaApp +from kaya.session import InMemorySessionStore, Session, SessionMixin + + +class FakeClock: + def __init__(self, start: float = 0.0) -> None: + self._now = start + + def __call__(self) -> float: + return self._now + + def advance(self, seconds: float) -> None: + self._now += seconds + + +class SessionIndexUnitTest(unittest.TestCase): + + def test_add_and_remove_index(self) -> None: + session = Session() + session.add_index('sub', 'alice') + self.assertEqual(frozenset({('sub', 'alice')}), session.indexes) + self.assertEqual(frozenset({('sub', 'alice')}), session.added_indexes) + self.assertTrue(session.modified) + + session.remove_index('sub', 'alice') + self.assertEqual(frozenset(), session.indexes) + self.assertEqual(frozenset(), session.added_indexes) + self.assertEqual(frozenset(), session.removed_indexes) + + def test_remove_persisted_index(self) -> None: + session = Session(indexes={('sub', 'alice')}) + self.assertFalse(session.modified) + session.remove_index('sub', 'alice') + self.assertEqual(frozenset(), session.indexes) + self.assertEqual(frozenset({('sub', 'alice')}), session.removed_indexes) + + session.add_index('sub', 'alice') + self.assertEqual(frozenset({('sub', 'alice')}), session.indexes) + self.assertEqual(frozenset(), session.removed_indexes) + + def test_payload_round_trip(self) -> None: + session = Session('abc', {'user': 'alice'}, indexes={('sub', 'alice')}) + loaded = Session.from_payload('abc', session.to_payload()) + self.assertEqual({'user': 'alice'}, dict(loaded)) + self.assertEqual(frozenset({('sub', 'alice')}), loaded.indexes) + + def test_legacy_payload(self) -> None: + loaded = Session.from_payload('abc', {'user': 'alice'}) + self.assertEqual({'user': 'alice'}, dict(loaded)) + self.assertEqual(frozenset(), loaded.indexes) + + +class InMemorySessionIndexTest(unittest.TestCase): + + @async_test + async def test_save_load_and_find(self) -> None: + store = InMemorySessionStore() + self.assertTrue(store.supports_indexes) + session = Session('abc', {'user': 'alice'}) + session.add_index('sub', 'alice') + await store.save('abc', session) + session.commit_indexes() + + loaded = await store.load('abc') + assert loaded is not None + self.assertEqual(frozenset({('sub', 'alice')}), loaded.indexes) + self.assertEqual(['abc'], await store.find_sessions('sub', 'alice')) + self.assertEqual([], await store.find_sessions('sub', 'bob')) + + @async_test + async def test_delete_cleans_reverse_index(self) -> None: + store = InMemorySessionStore() + session = Session('abc', indexes={('sub', 'alice')}) + await store.save('abc', session) + await store.delete('abc') + self.assertEqual([], await store.find_sessions('sub', 'alice')) + self.assertEqual({}, store._session_indexes) + + @async_test + async def test_delete_by_index_deletes_all_sessions(self) -> None: + store = InMemorySessionStore() + await store.save('a', Session('a', indexes={('sub', 'alice')})) + await store.save('b', Session('b', indexes={('sub', 'alice')})) + await store.save('c', Session('c', indexes={('sub', 'bob')})) + + self.assertEqual(2, await store.delete_by_index('sub', 'alice')) + self.assertIsNone(await store.load('a')) + self.assertIsNone(await store.load('b')) + self.assertIsNotNone(await store.load('c')) + + @async_test + async def test_expiry_prunes_index(self) -> None: + clock = FakeClock() + store = InMemorySessionStore(clock=clock) + await store.save( + 'abc', + Session('abc', indexes={('sub', 'alice')}), + max_age=60, + ) + clock.advance(61) + self.assertEqual([], await store.find_sessions('sub', 'alice')) + self.assertEqual({}, store._session_indexes) + + @async_test + async def test_removed_index_is_not_readded(self) -> None: + store = InMemorySessionStore() + session = Session('abc', indexes={('sub', 'alice')}) + await store.save('abc', session) + session.commit_indexes() + + session.remove_index('sub', 'alice') + await store.save('abc', session) + self.assertEqual([], await store.find_sessions('sub', 'alice')) + self.assertEqual(frozenset(), session.indexes) + + +class SessionIndexIntegrationTest(unittest.TestCase): + + def setUp(self) -> None: + self.store = InMemorySessionStore() + self.mixin = SessionMixin(self.store) + self.app = KayaApp(mixins=[self.mixin]) + + @self.app.GET('/login') + async def login(ctx: HttpContext) -> None: + ctx.session['user'] = 'alice' + ctx.session.add_index('sub', 'alice') + await ctx.send_str(200, 'ok') + + @self.app.GET('/rotate') + async def rotate(ctx: HttpContext) -> None: + ctx.session.regenerate_id() + await ctx.send_str(200, 'ok') + + @self.app.GET('/logout') + async def logout(ctx: HttpContext) -> None: + ctx.session.invalidate() + await ctx.send_str(200, 'ok') + + @async_test + async def test_rotation_preserves_index(self) -> None: + self.assertIs(self.store, self.mixin.store) + transport = httpx.ASGITransport(app=self.app) + async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: + await client.get('/login') + old_id = client.cookies['session_id'] + self.assertEqual([old_id], await self.store.find_sessions('sub', 'alice')) + + await client.get('/rotate') + new_id = client.cookies['session_id'] + self.assertNotEqual(old_id, new_id) + self.assertEqual([new_id], await self.store.find_sessions('sub', 'alice')) + + @async_test + async def test_invalidation_cleans_index(self) -> None: + transport = httpx.ASGITransport(app=self.app) + async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: + await client.get('/login') + await client.get('/logout') + self.assertEqual([], await self.store.find_sessions('sub', 'alice')) + + +if __name__ == '__main__': + unittest.main()