diff --git a/README.md b/README.md index 7e6c165..e88fdc9 100644 --- a/README.md +++ b/README.md @@ -200,6 +200,10 @@ Let a logged-in user manage their own enrolled authentication methods — enroll Bind tokens to a key your server holds ([RFC 9449](https://www.rfc-editor.org/rfc/rfc9449)) so a stolen token alone cannot be replayed. DPoP is supported for Passkey sign-in (`signin_with_passkey`) and the authentication-methods/factors methods on `MyAccountClient`. For key generation and usage, see [examples/Passkeys.md](examples/Passkeys.md#3-dpop-bound-passkey-tokens-optional) and [examples/MyAccountAuthenticationMethods.md](examples/MyAccountAuthenticationMethods.md#dpop). +### 10. Anonymous Sessions + +Give a visitor an Auth0 `anon@` identity before they log in, so cart/preference metadata attached pre-login is available to Post-Login Actions once they do. Requires a separate `anonymous_store` instance — never the same instance as `state_store` — and a tenant-level paid add-on flag. For setup, the token renewal ladder, login injection, and the store-isolation requirement, see [examples/AnonymousSessions.md](examples/AnonymousSessions.md). + ## Feedback ### Contributing diff --git a/examples/AnonymousSessions.md b/examples/AnonymousSessions.md new file mode 100644 index 0000000..077fc02 --- /dev/null +++ b/examples/AnonymousSessions.md @@ -0,0 +1,132 @@ +# Anonymous Sessions + +Anonymous Sessions give a visitor an Auth0 identity **before they log in**. Each visitor gets a persistent `anon@` subject plus an access token, with up to 1 KB of key/value metadata (cart, preferences) attached at creation. At login, the session token rides into `/authorize` so Post-Login / Pre-User-Registration Actions can read the anonymous data via `event.anonymous_session` — nothing migrates onto the real user profile automatically; the Action author decides what to persist. + +> [!NOTE] +> Anonymous Sessions support for server SDKs is in Early Access, gated by a tenant-level, paid add-on feature flag (`anonymous_sessions_enabled`). `auth0-server-python` mounts no routes and sets no cookies — this guide covers the framework-agnostic core only. + +## Table of Contents + +- [Anonymous Sessions](#anonymous-sessions) + - [Table of Contents](#table-of-contents) + - [Setup](#setup) + - [Creating a Session](#creating-a-session) + - [Getting a Token (Renewal Ladder)](#getting-a-token-renewal-ladder) + - [Introspecting a Session](#introspecting-a-session) + - [Logging Out](#logging-out) + - [Login Injection](#login-injection) + - [Rate-Limiting `get_token()`](#rate-limiting-get_token) + - [Error Handling](#error-handling) + - [Known Limitations](#known-limitations) + - [Additional Resources](#additional-resources) + +## Setup + +Before using the anonymous sessions API, the `anonymous_sessions_enabled` flag must be turned on for your tenant (contact your Auth0 account team — there is no self-serve path yet), and the application/client must be enabled for the feature. + +Pass an `anonymous_store` to `ServerClient`, alongside your existing `state_store` and `transaction_store`: + +```python +server_client = ServerClient( + domain="your-tenant.auth0.com", + client_id="...", + client_secret="...", + secret="...", + state_store=my_state_store, + transaction_store=my_transaction_store, + anonymous_store=my_anonymous_store, # its own store instance, not state_store +) +``` + +Give `anonymous_store` its own store instance, not `state_store` with a different identifier. If you omit it, every `.anonymous.*` call raises `ConfigurationError` before any write. + +## Creating a Session + +```python +session = await server_client.anonymous.create_session( + audience="https://api.example.com", + scope="read:cart write:cart", + metadata={"cart_id": "cart_456"}, + store_options=store_options, +) +``` + +`metadata` is **set once, at creation, and never updated** — there is no platform update endpoint for anonymous sessions. Any JSON-serializable value is accepted, ≤1 KB total (UTF-8 JSON byte length); oversized, non-JSON-serializable, or dangerous-key (`__proto__`, `constructor`, `prototype`) metadata is rejected client-side before any network call. + +`AnonymousSession` never exposes the raw session token — only `sub`, `session_id`, `access_token`, `expires_at`, `session_expires_at`, `metadata`, and `is_new`. + +> [!IMPORTANT] +> **Always check `is_new`.** It is `True` both on the first call to `create_session()` and on a *silent* re-mint (see below) — the only signal your application receives when the anonymous `sub` has changed. Any code correlating data on `sub` (e.g. a cart keyed by anonymous user) must check this on every call. + +## Getting a Token + +```python +token = await server_client.anonymous.get_token(store_options=store_options) +``` + +Renewal logic, in order: + +1. Cached access token still fresh → returned with no network call. +2. Expired → re-minted using the stored session token (not a refresh-token grant — anonymous sessions never issue refresh tokens). +3. Session token also expired or invalid → a **brand-new session is silently created, once**. Metadata from the old session is permanently lost, and `sub` changes. This never raises — an anonymous pre-login session carries no authorization, so re-minting crosses no trust boundary. +4. Any other error → raised as a typed exception. No swallow, no auto-retry beyond the one re-mint in step 3. + +## Introspecting a Session + +```python +status = await server_client.anonymous.introspect(store_options=store_options) +``` + +`introspect()` is read-only: it returns the current session status without renewing the token or changing `sub`. + +## Logging Out + +```python +await server_client.anonymous.logout(store_options=store_options) +``` + +> [!CAUTION] +> **`logout()` does not revoke.** There is no server-side anonymous session store to revoke against, this clears only the locally-held encrypted context. Any access token already issued for this anonymous session remains valid until its natural expiry. + +Local state is always cleared, even if the remote call fails. If the remote `/anonymous/logout` call itself fails, `logout()` raises `AnonymousSessionLogoutError` after clearing local state, so the failure isn't swallowed. + +## Login Injection + +When an anonymous session is active, `start_interactive_login()` automatically includes the session token in the `/authorize` request, no code change needed at your call site. If no anonymous session exists, behavior is same as today. + +The token travels as a query parameter to `/authorize`, which means it lands in browser history, `Referer` headers, and access logs. This is because the token grants no authorization on its own and the request is a browser-to-Auth0 HTTPS redirect, but you should still set `Referrer-Policy: no-referrer` on your login pages, and never log the authorize URL. + +Pushed Authorization Requests (PAR) are not supported for anonymous sessions — injection is suppressed entirely on that code path. + +## Rate-Limiting `get_token()` + +`get_token()`'s retry-once bound caps amplification to two upstream Auth0 calls *per invocation*. It does not protect against an attacker calling your route repeatedly. `POST /anonymous/token` is an unauthenticated, token-issuing endpoint. **You must rate-limit any route in your application that calls `get_token()` on an anonymous session**, the same way you would rate-limit any other unauthenticated token-issuing path. The SDK has no request-level context to do this itself. + +## Error Handling + +All anonymous session errors subclass `AnonymousSessionApiError`, carrying a `.code` you can branch on: + +```python +from auth0_server_python.error import ( + AnonymousSessionFeatureNotEnabledError, + AnonymousSessionClientNotEnabledError, + AnonymousSessionClientNotSupportedError, + AnonymousSessionResourceServerError, + AnonymousSessionScopeError, + AnonymousSessionCreateError, + AnonymousSessionTokenError, + AnonymousSessionIntrospectError, + AnonymousSessionLogoutError, +) + +try: + session = await server_client.anonymous.create_session(audience="...", scope="...") +except AnonymousSessionFeatureNotEnabledError: + ... +``` + +## Known Limitations + +- **DPoP is not supported.** `AnonymousClient` has no `dpop_key` parameter anywhere in its public API. A tenant/client configured with `require_proof_of_possession: true` cannot use anonymous sessions, you will see `AnonymousSessionClientNotSupportedError`. +- **PAR, CIBA, Device Flow, RAR, and mTLS clients are not supported** for anonymous sessions. +- **No server-side revocation.** See [Logging Out](#logging-out) above. \ No newline at end of file diff --git a/src/auth0_server_python/auth_server/__init__.py b/src/auth0_server_python/auth_server/__init__.py index 611f6b7..9bf85e8 100644 --- a/src/auth0_server_python/auth_server/__init__.py +++ b/src/auth0_server_python/auth_server/__init__.py @@ -1,5 +1,6 @@ +from .anonymous_client import AnonymousClient from .mfa_client import MfaClient from .my_account_client import MyAccountClient from .server_client import ServerClient -__all__ = ["ServerClient", "MyAccountClient", "MfaClient"] +__all__ = ["ServerClient", "MyAccountClient", "MfaClient", "AnonymousClient"] diff --git a/src/auth0_server_python/auth_server/anonymous_client.py b/src/auth0_server_python/auth_server/anonymous_client.py new file mode 100644 index 0000000..5203ff7 --- /dev/null +++ b/src/auth0_server_python/auth_server/anonymous_client.py @@ -0,0 +1,741 @@ +""" +Anonymous Sessions client for auth0-server-python SDK. +Handles pre-login anon@ identity operations against the Auth0 anonymous session API. +""" + +import json +import time +from typing import Any, Callable, Optional, Union + +import httpx +from pydantic import ValidationError + +from auth0_server_python.auth_schemes.bearer_auth import BearerAuth +from auth0_server_python.auth_types import ( + AnonymousCreateTokenResponse, + AnonymousSession, + AnonymousSessionContext, + AnonymousSessionIntrospection, + AnonymousTokenResponse, + CreateAnonymousSessionOptions, +) +from auth0_server_python.encryption.encrypt import decrypt, encrypt +from auth0_server_python.error import ( + AnonymousSessionApiError, + AnonymousSessionClientNotEnabledError, + AnonymousSessionClientNotSupportedError, + AnonymousSessionCreateError, + AnonymousSessionFeatureNotEnabledError, + AnonymousSessionIntrospectError, + AnonymousSessionLogoutError, + AnonymousSessionResourceServerError, + AnonymousSessionScopeError, + AnonymousSessionTokenError, + ConfigurationError, + DomainResolverError, + _AnonymousSessionExpired, +) +from auth0_server_python.utils.helpers import ( + build_domain_resolver_context, + validate_resolved_domain_value, +) + +ANON_IDENTIFIER = "_a0_anon" +ANON_TOKEN_SALT = "anon_session" + +_METADATA_MAX_BYTES = 1024 +_DANGEROUS_METADATA_KEYS = frozenset({"__proto__", "constructor", "prototype"}) + + +class AnonymousClient: + """ + Client for Auth0 anonymous session operations. + + Requires its own store instance, distinct from ServerClient's state_store. + DPoP is not supported with Anonymous Sessions. + """ + + def __init__( + self, + domain: Union[str, Callable, None], + client_id: str, + client_secret: str, + secret: str, + anonymous_store=None, + default_audience: Optional[str] = None, + default_scope: Optional[str] = None, + headers: Optional[dict[str, str]] = None, + ): + if callable(domain): + self._domain = None + self._domain_resolver = domain + else: + self._domain = domain + self._domain_resolver = None + self._client_id = client_id + self._client_secret = client_secret + self._secret = secret + self._anonymous_store = anonymous_store + self._default_audience = default_audience + self._default_scope = default_scope + self._headers = headers or {} + + def _get_http_client(self, **kwargs) -> httpx.AsyncClient: + """Return an httpx.AsyncClient with default headers injected. + + Args: + **kwargs: Forwarded to httpx.AsyncClient. + + Returns: + A configured httpx.AsyncClient. + """ + headers = {**kwargs.pop("headers", {}), **self._headers} + return httpx.AsyncClient(headers=headers, **kwargs) + + def _require_store(self) -> None: + """Fail closed when no anonymous store is configured. + + Raises: + ConfigurationError: No anonymous_store configured. + """ + if self._anonymous_store is None: + raise ConfigurationError( + "AnonymousClient requires its own anonymous_store, distinct from " + "ServerClient's state_store. Writing anonymous state into the same " + "store instance can silently overwrite the authenticated session." + ) + + async def _resolve_domain(self, store_options: Optional[dict[str, Any]] = None) -> str: + """Resolve the tenant domain from the configured resolver or static value. + + Args: + store_options: Optional context passed to the domain resolver. + + Returns: + The resolved domain string. + + Raises: + DomainResolverError: The resolver function raised or returned an + invalid value. + """ + if self._domain_resolver: + context = build_domain_resolver_context(store_options) + try: + resolved = await self._domain_resolver(context) + return validate_resolved_domain_value(resolved) + except DomainResolverError: + raise + except Exception as e: + raise DomainResolverError( + f"Domain resolver function raised an exception: {str(e)}", + original_error=e, + ) + return self._domain + + @staticmethod + def _normalize_url(value: Optional[str]) -> Optional[str]: + """Normalize a domain-like value for comparison. + + Args: + value: A domain or URL string, or None. + + Returns: + The value lowercased, scheme-qualified, and without a trailing + slash. Falsy input is returned unchanged. + """ + if not value: + return value + value = value.lower() + if value.startswith("https://"): + pass + elif value.startswith("http://"): + value = value.replace("http://", "https://") + else: + value = f"https://{value}" + return value.rstrip("/") + + # ============================================================================ + # ERROR HANDLING + # ============================================================================ + + @staticmethod + def _parse_anonymous_error_body(response: httpx.Response) -> dict[str, Any]: + """Parse an error response body as JSON. + + Args: + response: The HTTP response to parse. + + Returns: + The parsed JSON body, or a fallback dict with 'error_description' + when the body is not valid JSON. + """ + try: + data = response.json() + except (json.JSONDecodeError, ValueError): + data = None + if not isinstance(data, dict): + return { + "error_description": f"Request failed with status {response.status_code}", + } + return data + + def _map_anonymous_error( + self, + status_code: int, + error_data: dict[str, Any], + operation: str, + ) -> Exception: + """Map a server error response to a typed exception. + + Args: + status_code: The HTTP status code of the response. + error_data: The parsed error response body. + operation: One of 'create', 'token', 'logout', 'introspect'. + + Returns: + The exception instance. Does not raise it. + """ + code = error_data.get("error", "") + description = error_data.get("error_description") or f"Anonymous {operation} failed" + + if code in ("session_expired", "invalid_session_token"): + return _AnonymousSessionExpired(description) + if status_code == 400 and "Proof-of-Possession" in description: + return AnonymousSessionClientNotSupportedError(description, error_data) + if code == "feature_not_enabled": + return AnonymousSessionFeatureNotEnabledError(description, error_data) + if code == "unauthorized_client": + return AnonymousSessionClientNotEnabledError(description, error_data) + if code in ("invalid_target", "invalid_request"): + return AnonymousSessionResourceServerError(description, error_data) + if code == "invalid_scope": + return AnonymousSessionScopeError(description, error_data) + + if operation == "create": + return AnonymousSessionCreateError(description, cause=error_data) + if operation == "token": + return AnonymousSessionTokenError(description, error_data) + if operation == "logout": + return AnonymousSessionLogoutError(description, error_data) + if operation == "introspect": + return AnonymousSessionIntrospectError(description, error_data) + return AnonymousSessionApiError(code or "anonymous_error", description, error_data) + + # ============================================================================ + # METADATA VALIDATION + # ============================================================================ + + @staticmethod + def _validate_metadata(metadata: Optional[dict[str, Any]]) -> None: + """Validate metadata locally before it reaches the network. + + Args: + metadata: The metadata dict to validate, or None. + + Raises: + AnonymousSessionCreateError: metadata is not a dict, contains a + disallowed key, a non-JSON-serializable value, or exceeds 1KB. + """ + if metadata is None: + return + if not isinstance(metadata, dict): + raise AnonymousSessionCreateError("metadata must be a JSON object", code="invalid_metadata") + for key in metadata: + if key in _DANGEROUS_METADATA_KEYS: + raise AnonymousSessionCreateError( + f"metadata key '{key}' is not allowed", code="invalid_metadata" + ) + try: + size = len(json.dumps(metadata).encode("utf-8")) + except TypeError as e: + raise AnonymousSessionCreateError( + "metadata must contain only JSON-serializable values", code="invalid_metadata" + ) from e + if size > _METADATA_MAX_BYTES: + raise AnonymousSessionCreateError( + "metadata exceeds the 1KB size limit", code="metadata_too_large" + ) + + # ============================================================================ + # ENCRYPTION + # ============================================================================ + + def _encrypt_context(self, context: AnonymousSessionContext) -> str: + """Encrypt an anonymous session context for storage. + + Args: + context: The context to encrypt. + + Returns: + The encrypted context string. + """ + return encrypt(context.model_dump(), self._secret, ANON_TOKEN_SALT) + + def _decrypt_context(self, stored: Any) -> AnonymousSessionContext: + """Decrypt and validate a stored anonymous session record. + + Args: + stored: The raw record read from the anonymous store. + + Returns: + The decrypted AnonymousSessionContext. + + Raises: + _AnonymousSessionExpired: The record is missing, malformed, or + fails to decrypt or validate. + """ + try: + encrypted = stored.get("context") if isinstance(stored, dict) else None + if not encrypted: + raise ValueError("Malformed anonymous session record") + payload = decrypt(encrypted, self._secret, ANON_TOKEN_SALT) + return AnonymousSessionContext(**payload) + except Exception as e: + raise _AnonymousSessionExpired( + "Stored anonymous session token is invalid or corrupted." + ) from e + + # ============================================================================ + # SESSION CREATION + # ============================================================================ + + async def _create_session_at( + self, + domain: str, + *, + audience: Optional[str], + scope: Optional[str], + metadata: Optional[dict[str, Any]], + store_options: Optional[dict[str, Any]], + ) -> AnonymousSession: + """Create a fresh anonymous session against a resolved domain. + + Args: + domain: The resolved tenant domain. + audience: Audience for the new session, or None. + scope: Scope for the new session, or None. + metadata: Metadata to attach at creation, or None. + store_options: Options passed to the anonymous store. + + Returns: + The newly created AnonymousSession, with is_new=True. + + Raises: + AnonymousSessionCreateError: The request failed, or the response + was invalid or missing required fields. + """ + base_url = f"https://{domain}" + body: dict[str, Any] = {"client_id": self._client_id} + if self._client_secret: + body["client_secret"] = self._client_secret + if audience: + body["audience"] = audience + if scope: + body["scope"] = scope + if metadata: + body["metadata"] = metadata + + async with self._get_http_client() as client: + try: + response = await client.post(f"{base_url}/anonymous/token", json=body) + except httpx.HTTPError as e: + raise AnonymousSessionCreateError("Failed to reach the anonymous token endpoint") from e + + if response.status_code != 200: + error_data = self._parse_anonymous_error_body(response) + mapped = self._map_anonymous_error(response.status_code, error_data, "create") + if isinstance(mapped, _AnonymousSessionExpired): + # Internal-only type must never escape. + raise AnonymousSessionCreateError(str(mapped)) + raise mapped + + try: + token_response = AnonymousCreateTokenResponse.model_validate(response.json()) + except (json.JSONDecodeError, ValueError, ValidationError) as e: + raise AnonymousSessionCreateError("Failed to parse anonymous token response") from e + + now = int(time.time()) + context = AnonymousSessionContext( + session_token=token_response.session_token, + sub=token_response.sub, + session_id=token_response.session_id, + access_token=token_response.access_token, + expires_at=now + token_response.expires_in, + session_expires_at=( + now + token_response.session_expires_in + if token_response.session_expires_in + else None + ), + metadata=metadata, + created_at=now, + domain=domain, + audience=audience, + scope=scope, + ) + await self._anonymous_store.set( + ANON_IDENTIFIER, + {"context": self._encrypt_context(context)}, + options=store_options, + ) + return AnonymousSession( + sub=context.sub, + session_id=context.session_id, + access_token=context.access_token, + expires_at=context.expires_at, + session_expires_at=context.session_expires_at, + metadata=context.metadata, + is_new=True, + ) + + # ============================================================================ + # TOKEN RENEWAL LADDER + # ============================================================================ + + async def _remint( + self, context: AnonymousSessionContext, store_options: Optional[dict[str, Any]] + ) -> AnonymousSession: + """Re-mint an access token using the stored session token. + + Args: + context: The current decrypted session context. + store_options: Options passed to the anonymous store. + + Returns: + The refreshed AnonymousSession. is_new is True only when the + retry-once fallback created a brand-new session. + + Raises: + AnonymousSessionTokenError: The request failed, or the response was + invalid. + """ + domain = context.domain or await self._resolve_domain(store_options) + base_url = f"https://{domain}" + body: dict[str, Any] = { + "client_id": self._client_id, + "session_token": context.session_token, + } + if self._client_secret: + body["client_secret"] = self._client_secret + + async with self._get_http_client() as client: + try: + response = await client.post(f"{base_url}/anonymous/token", json=body) + except httpx.HTTPError as e: + raise AnonymousSessionTokenError("Failed to reach the anonymous token endpoint") from e + + if response.status_code != 200: + error_data = self._parse_anonymous_error_body(response) + mapped = self._map_anonymous_error(response.status_code, error_data, "token") + if isinstance(mapped, _AnonymousSessionExpired): + # Retry-once: exactly one follow-up create call, never a loop. + return await self._create_session_at( + domain, + audience=context.audience, + scope=context.scope, + metadata=None, + store_options=store_options, + ) + raise mapped + + try: + token_response = AnonymousTokenResponse.model_validate(response.json()) + except (json.JSONDecodeError, ValueError, ValidationError) as e: + raise AnonymousSessionTokenError("Failed to parse anonymous token response") from e + + now = int(time.time()) + new_context = AnonymousSessionContext( + session_token=( + token_response.session_token + if token_response.session_token is not None + else context.session_token + ), + sub=token_response.sub if token_response.sub is not None else context.sub, + session_id=( + token_response.session_id + if token_response.session_id is not None + else context.session_id + ), + access_token=token_response.access_token, + expires_at=now + token_response.expires_in, + session_expires_at=( + now + token_response.session_expires_in + if token_response.session_expires_in + else context.session_expires_at + ), + metadata=context.metadata, + created_at=context.created_at, + domain=domain, + audience=context.audience, + scope=context.scope, + ) + await self._anonymous_store.set( + ANON_IDENTIFIER, + {"context": self._encrypt_context(new_context)}, + options=store_options, + ) + return AnonymousSession( + sub=new_context.sub, + session_id=new_context.session_id, + access_token=new_context.access_token, + expires_at=new_context.expires_at, + session_expires_at=new_context.session_expires_at, + metadata=new_context.metadata, + is_new=False, + ) + + # ============================================================================ + # LOGIN INJECTION SUPPORT + # ============================================================================ + + async def get_session_token_for_injection( + self, store_options: Optional[dict[str, Any]] = None + ) -> Optional[str]: + """Read the active session token for login injection without renewing. + + Args: + store_options: Options passed to the anonymous store. + + Returns: + The raw session token, or None when there is no store, no active + session, or the stored record cannot be decrypted. + """ + if self._anonymous_store is None: + return None + try: + stored = await self._anonymous_store.get(ANON_IDENTIFIER, options=store_options) + except Exception: + return None + if not stored: + return None + try: + context = self._decrypt_context(stored) + except _AnonymousSessionExpired: + return None + return context.session_token + + # ============================================================================ + # PUBLIC API + # ============================================================================ + + async def create_session( + self, + options: Optional[Union[CreateAnonymousSessionOptions, dict[str, Any]]] = None, + *, + audience: Optional[str] = None, + scope: Optional[str] = None, + metadata: Optional[dict[str, Any]] = None, + store_options: Optional[dict[str, Any]] = None, + ) -> AnonymousSession: + """Mint a fresh anon@ identity. + + Args: + options: Optional bundle of audience/scope/metadata, accepted as a + CreateAnonymousSessionOptions or a plain dict. Explicit keyword + arguments below always win over the same field on options. + audience: Audience for the session. Falls back to options.audience, + then to the client's configured default, when omitted. + scope: Scope for the session. Falls back to options.scope, then to + the client's configured default, when omitted. + metadata: Metadata to attach at creation, up to 1KB. Cannot be + changed after creation. Falls back to options.metadata. + store_options: Options passed to the anonymous store. + + Returns: + The newly created AnonymousSession. + + Raises: + ConfigurationError: No anonymous_store configured. + AnonymousSessionCreateError: Invalid options, local validation + failure, or server rejection. + """ + self._require_store() + if options is not None: + if isinstance(options, dict): + try: + options = CreateAnonymousSessionOptions(**options) + except ValidationError as e: + raise AnonymousSessionCreateError( + "Invalid create_session options", code="invalid_options" + ) from e + audience = audience if audience is not None else options.audience + scope = scope if scope is not None else options.scope + metadata = metadata if metadata is not None else options.metadata + self._validate_metadata(metadata) + audience = audience or self._default_audience + scope = scope or self._default_scope + domain = await self._resolve_domain(store_options) + return await self._create_session_at( + domain, audience=audience, scope=scope, metadata=metadata, store_options=store_options + ) + + async def get_token(self, store_options: Optional[dict[str, Any]] = None) -> AnonymousSession: + """Return a valid anonymous access token, renewing or re-minting as needed. + + Args: + store_options: Options passed to the anonymous store. + + Returns: + The current or refreshed AnonymousSession. + + Raises: + ConfigurationError: No anonymous_store configured. + AnonymousSessionTokenError: No active session, or an unrecoverable + failure. + """ + self._require_store() + stored = await self._anonymous_store.get(ANON_IDENTIFIER, options=store_options) + if not stored: + raise AnonymousSessionTokenError("No active anonymous session. Call create_session() first.") + + try: + context = self._decrypt_context(stored) + except _AnonymousSessionExpired: + domain = await self._resolve_domain(store_options) + return await self._create_session_at( + domain, + audience=self._default_audience, + scope=self._default_scope, + metadata=None, + store_options=store_options, + ) + + if self._domain_resolver: + current_domain = await self._resolve_domain(store_options) + if context.domain and self._normalize_url(context.domain) != self._normalize_url( + current_domain + ): + # Cross-tenant reuse must be structurally impossible, so + # discard and mint fresh under the current tenant instead. + return await self._create_session_at( + current_domain, + audience=context.audience, + scope=context.scope, + metadata=None, + store_options=store_options, + ) + + now = int(time.time()) + if context.expires_at > now: + return AnonymousSession( + sub=context.sub, + session_id=context.session_id, + access_token=context.access_token, + expires_at=context.expires_at, + session_expires_at=context.session_expires_at, + metadata=context.metadata, + is_new=False, + ) + + return await self._remint(context, store_options) + + async def introspect( + self, store_options: Optional[dict[str, Any]] = None + ) -> AnonymousSessionIntrospection: + """Return the current anonymous session status without mutating the store. + + Args: + store_options: Options passed to the anonymous store. + + Returns: + The current AnonymousSessionIntrospection. + + Raises: + ConfigurationError: No anonymous_store configured. + AnonymousSessionIntrospectError: No active session, or a request + failure. + """ + self._require_store() + stored = await self._anonymous_store.get(ANON_IDENTIFIER, options=store_options) + if not stored: + raise AnonymousSessionIntrospectError("No active anonymous session to introspect.") + + try: + context = self._decrypt_context(stored) + except _AnonymousSessionExpired as e: + raise AnonymousSessionIntrospectError( + "Stored anonymous session is invalid or corrupted." + ) from e + + domain = context.domain or await self._resolve_domain(store_options) + base_url = f"https://{domain}" + + async with self._get_http_client() as client: + try: + response = await client.get( + f"{base_url}/anonymous/userinfo", + auth=BearerAuth(context.access_token), + ) + except httpx.HTTPError as e: + raise AnonymousSessionIntrospectError( + "Failed to reach the anonymous userinfo endpoint" + ) from e + + if response.status_code != 200: + error_data = self._parse_anonymous_error_body(response) + mapped = self._map_anonymous_error(response.status_code, error_data, "introspect") + if isinstance(mapped, _AnonymousSessionExpired): + raise AnonymousSessionIntrospectError(str(mapped)) + raise mapped + + try: + return AnonymousSessionIntrospection.model_validate(response.json()) + except (json.JSONDecodeError, ValueError, ValidationError) as e: + raise AnonymousSessionIntrospectError( + "Failed to parse anonymous introspection response" + ) from e + + async def logout(self, store_options: Optional[dict[str, Any]] = None) -> None: + """Clear the locally-held anonymous session without revoking issued tokens. + + Local state is cleared unconditionally, even when the remote call + fails, since there is no server-side session to keep in sync with. + + Args: + store_options: Options passed to the anonymous store. + + Raises: + ConfigurationError: No anonymous_store configured. + AnonymousSessionLogoutError: The remote logout call failed for a + reason other than the session already being expired/invalid. + """ + self._require_store() + stored = await self._anonymous_store.get(ANON_IDENTIFIER, options=store_options) + if not stored: + return + + try: + context = self._decrypt_context(stored) + except _AnonymousSessionExpired: + context = None + + error_to_raise: Optional[Exception] = None + error_cause: Optional[BaseException] = None + + if context is not None: + domain = context.domain or await self._resolve_domain(store_options) + base_url = f"https://{domain}" + body: dict[str, Any] = { + "client_id": self._client_id, + "session_token": context.session_token, + } + if self._client_secret: + body["client_secret"] = self._client_secret + try: + async with self._get_http_client() as client: + response = await client.post(f"{base_url}/anonymous/logout", json=body) + except httpx.HTTPError as e: + error_to_raise = AnonymousSessionLogoutError( + "Failed to reach the anonymous logout endpoint" + ) + error_cause = e + else: + if response.status_code != 200: + error_data = self._parse_anonymous_error_body(response) + mapped = self._map_anonymous_error(response.status_code, error_data, "logout") + if not isinstance(mapped, _AnonymousSessionExpired): + error_to_raise = mapped + + await self._anonymous_store.delete(ANON_IDENTIFIER, options=store_options) + + if error_to_raise is not None: + raise error_to_raise from error_cause diff --git a/src/auth0_server_python/auth_server/server_client.py b/src/auth0_server_python/auth_server/server_client.py index c8eb6b3..aac2207 100644 --- a/src/auth0_server_python/auth_server/server_client.py +++ b/src/auth0_server_python/auth_server/server_client.py @@ -20,6 +20,7 @@ from pydantic import ValidationError from auth0_server_python.auth_schemes.dpop_auth import make_dpop_proof_for_token_endpoint +from auth0_server_python.auth_server.anonymous_client import AnonymousClient from auth0_server_python.auth_server.mfa_client import MfaClient from auth0_server_python.auth_server.my_account_client import MyAccountClient from auth0_server_python.auth_types import ( @@ -85,7 +86,8 @@ # redirect_uri is intentionally excluded — in MCD mode it is built # dynamically from the resolved domain at login time. INTERNAL_AUTHORIZE_PARAMS = ["client_id", "response_type", - "code_challenge", "code_challenge_method", "state", "nonce", "scope"] + "code_challenge", "code_challenge_method", "state", "nonce", "scope", + "session_token"] # issued_token_type URN for a Session Transfer Token (STT). SESSION_TRANSFER_TOKEN_TYPE = "urn:auth0:params:oauth:token-type:session_transfer_token" @@ -117,6 +119,7 @@ def __init__( secret: str = None, transaction_store=None, state_store=None, + anonymous_store=None, transaction_identifier: str = "_a0_tx", state_identifier: str = "_a0_session", authorization_params: Optional[dict[str, Any]] = None, @@ -134,6 +137,13 @@ def __init__( secret: Secret used for encryption transaction_store: Custom transaction store (defaults to MemoryTransactionStore) state_store: Custom state store (defaults to MemoryStateStore) + anonymous_store: Store for anonymous session state (server_client.anonymous.*). + Must be a distinct store *instance* from state_store, not merely a + different identifier. On a store where the identifier is used only + as an encryption salt rather than a location key, writing anonymous + state through state_store would silently overwrite the authenticated + session cookie. When omitted, the `.anonymous` sub-client fails + closed on first use rather than sharing state_store implicitly. transaction_identifier: Identifier for transaction data state_identifier: Identifier for state data authorization_params: Default parameters for authorization requests @@ -180,6 +190,7 @@ def __init__( # Initialize stores self._transaction_store = transaction_store self._state_store = state_store + self._anonymous_store = anonymous_store self._transaction_identifier = transaction_identifier self._state_identifier = state_identifier @@ -214,6 +225,20 @@ def __init__( headers=self._telemetry_headers, ) + # Its own store, never self._state_store, so anonymous state stays isolated. + self._anonymous_client = AnonymousClient( + domain=domain, + client_id=self._client_id, + client_secret=self._client_secret, + secret=self._secret, + anonymous_store=self._anonymous_store, + default_audience=self._default_authorization_params.get("audience"), + default_scope=self._default_authorization_params.get("scope") + if isinstance(self._default_authorization_params.get("scope"), str) + else None, + headers=self._telemetry_headers, + ) + def _get_http_client(self, **kwargs) -> httpx.AsyncClient: """Return an httpx.AsyncClient with telemetry headers injected.""" headers = {**kwargs.pop("headers", {}), **self._telemetry_headers} @@ -543,6 +568,18 @@ async def start_interactive_login( if options.invitation: auth_params["invitation"] = options.invitation + # session_token comes only from the SDK's own encrypted anonymous + # store, never a caller. The pop strips any value seeded from the + # constructor defaults, closing a session-fixation vector. + auth_params.pop("session_token", None) + anonymous_session_token = None + if not self._pushed_authorization_requests: + anonymous_session_token = await self._anonymous_client.get_session_token_for_injection( + store_options + ) + if anonymous_session_token: + auth_params["session_token"] = anonymous_session_token + # Build the transaction data to store with domain transaction_data = TransactionData( code_verifier=code_verifier, @@ -551,6 +588,7 @@ async def start_interactive_login( domain=origin_domain, redirect_uri=auth_params.get("redirect_uri"), organization=resolved_org, + session_token=anonymous_session_token, ) # Store the transaction data @@ -2862,6 +2900,15 @@ def mfa(self) -> MfaClient: """Access the MFA client for multi-factor authentication operations.""" return self._mfa_client + # ============================================================================ + # ANONYMOUS SESSIONS + # ============================================================================ + + @property + def anonymous(self) -> AnonymousClient: + """Access the anonymous sessions client for pre-login anon@ identity operations.""" + return self._anonymous_client + # ============================================================================ # PASSKEY AUTHENTICATION # ============================================================================ diff --git a/src/auth0_server_python/auth_types/__init__.py b/src/auth0_server_python/auth_types/__init__.py index e886938..a910ee3 100644 --- a/src/auth0_server_python/auth_types/__init__.py +++ b/src/auth0_server_python/auth_types/__init__.py @@ -146,6 +146,7 @@ class TransactionData(BaseModel): redirect_uri: Optional[str] = None domain: Optional[str] = None organization: Optional[str] = None + session_token: Optional[str] = None class Config: extra = "allow" # Allow additional fields not defined in the model @@ -221,6 +222,16 @@ class LogoutOptions(BaseModel): return_to: Optional[str] = None +class CreateAnonymousSessionOptions(BaseModel): + """Options bundle for create_session(): audience, scope, and metadata.""" + + model_config = ConfigDict(extra="forbid") + + audience: Optional[str] = None + scope: Optional[str] = None + metadata: Optional[dict[str, Any]] = None + + class AuthorizationParameters(BaseModel): """ Parameters used in authorization requests. @@ -852,3 +863,72 @@ class PasskeyTokenResponse(BaseModel): scope: Optional[str] = None id_token: Optional[str] = None refresh_token: Optional[str] = None + + +# ============================================================================= +# Anonymous Session Types +# ============================================================================= + + +class AnonymousSession(BaseModel): + """Public result of create_session() and the renewal ladder.""" + + sub: Optional[str] = None + session_id: Optional[str] = None + access_token: str + expires_at: int + session_expires_at: Optional[int] = None + metadata: Optional[dict[str, Any]] = None + is_new: bool + + +class AnonymousSessionIntrospection(BaseModel): + """Result of introspect(), lenient to unrecognized response fields.""" + + model_config = ConfigDict(extra="ignore") + sub: str + session_id: Optional[str] = None + expires_at: Optional[int] = None + metadata: Optional[dict[str, Any]] = None + + +class AnonymousTokenResponse(BaseModel): + """Raw response from POST /anonymous/token.""" + + access_token: str + token_type: str = "Bearer" + expires_in: int + session_token: Optional[str] = None + session_expires_in: Optional[int] = None + sub: Optional[str] = None + session_id: Optional[str] = None + + +class AnonymousCreateTokenResponse(AnonymousTokenResponse): + """Raw response from POST /anonymous/token on the create path. + + session_token is always present on creation, unlike on re-mint. + """ + + session_token: str + + +class AnonymousSessionContext(BaseModel): + """Internal context stored inside the encrypted anonymous session record. + + Rejects extra fields so a tampered payload fails closed on decrypt. + """ + + session_token: str + sub: Optional[str] = None + session_id: Optional[str] = None + access_token: str + expires_at: int + session_expires_at: Optional[int] = None + metadata: Optional[dict[str, Any]] = None + created_at: int + # Resolved domain at creation, gated on in resolver/MCD mode so a session + # minted for one tenant cannot be read back for another. + domain: Optional[str] = None + audience: Optional[str] = None + scope: Optional[str] = None diff --git a/src/auth0_server_python/error/__init__.py b/src/auth0_server_python/error/__init__.py index ee9279f..877c6cf 100644 --- a/src/auth0_server_python/error/__init__.py +++ b/src/auth0_server_python/error/__init__.py @@ -362,3 +362,95 @@ class PasskeyErrorCode: CHALLENGE_FAILED = "passkey_challenge_error" TOKEN_EXCHANGE_FAILED = "passkey_token_error" INVALID_RESPONSE = "invalid_response" + + +# ============================================================================= +# Anonymous Session Error Classes +# ============================================================================= + +class AnonymousSessionApiError(Auth0Error): + """Base class for anonymous session API errors.""" + + def __init__( + self, + code: str, + message: str, + cause: Optional[dict[str, Any]] = None + ): + super().__init__(message) + self.code = code + self.cause = cause + + +class AnonymousSessionCreateError(AnonymousSessionApiError): + """Error thrown when creating or re-minting an anonymous session fails.""" + + def __init__(self, message: str, code: str = "anonymous_create_error", cause: Optional[dict] = None): + super().__init__(code, message, cause) + + +class AnonymousSessionLogoutError(AnonymousSessionApiError): + """Error thrown when anonymous logout fails.""" + + def __init__(self, message: str, cause: Optional[dict] = None): + super().__init__("anonymous_logout_error", message, cause) + + +class AnonymousSessionTokenError(AnonymousSessionApiError): + """Error thrown when get_token() fails for reasons other than session expiry.""" + + def __init__(self, message: str, cause: Optional[dict] = None): + super().__init__("anonymous_token_error", message, cause) + + +class AnonymousSessionIntrospectError(AnonymousSessionApiError): + """Error thrown when introspect() fails on an HTTP or auth failure.""" + + def __init__(self, message: str, cause: Optional[dict] = None): + super().__init__("anonymous_introspect_error", message, cause) + + +class AnonymousSessionFeatureNotEnabledError(AnonymousSessionCreateError): + """Error thrown when the tenant has not enabled the anonymous sessions add-on.""" + + def __init__(self, message: str, cause: Optional[dict] = None): + super().__init__(message, "anonymous_feature_not_enabled_error", cause) + + +class AnonymousSessionClientNotEnabledError(AnonymousSessionCreateError): + """Error thrown when the client is not enabled for anonymous sessions.""" + + def __init__(self, message: str, cause: Optional[dict] = None): + super().__init__(message, "anonymous_client_not_enabled_error", cause) + + +class AnonymousSessionClientNotSupportedError(AnonymousSessionCreateError): + """Error thrown when the client type does not support anonymous sessions (e.g. DPoP-mandated).""" + + def __init__(self, message: str, cause: Optional[dict] = None): + super().__init__(message, "anonymous_client_not_supported_error", cause) + + +class AnonymousSessionResourceServerError(AnonymousSessionCreateError): + """Error thrown when the requested audience is not a valid resource server.""" + + def __init__(self, message: str, cause: Optional[dict] = None): + super().__init__(message, "anonymous_resource_server_error", cause) + + +class AnonymousSessionScopeError(AnonymousSessionCreateError): + """Error thrown when the requested scope is not granted to anonymous callers.""" + + def __init__(self, message: str, cause: Optional[dict] = None): + super().__init__(message, "anonymous_scope_error", cause) + + +class _AnonymousSessionExpired(Auth0Error): + """Internal-only signal that the stored session token is expired or invalid. + + Never raised to SDK callers. + """ + + def __init__(self, message: str = "The anonymous session token is expired or invalid."): + super().__init__(message) + self.name = "_AnonymousSessionExpired" diff --git a/src/auth0_server_python/tests/test_anonymous_client.py b/src/auth0_server_python/tests/test_anonymous_client.py new file mode 100644 index 0000000..429c90c --- /dev/null +++ b/src/auth0_server_python/tests/test_anonymous_client.py @@ -0,0 +1,816 @@ +""" +Tests for AnonymousClient, covering anonymous session API operations. +""" + +import inspect +import time +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest + +from auth0_server_python.auth_server.anonymous_client import ( + ANON_IDENTIFIER, + AnonymousClient, +) +from auth0_server_python.auth_types import ( + AnonymousSessionContext, + CreateAnonymousSessionOptions, +) +from auth0_server_python.encryption.encrypt import encrypt +from auth0_server_python.error import ( + AnonymousSessionClientNotEnabledError, + AnonymousSessionClientNotSupportedError, + AnonymousSessionCreateError, + AnonymousSessionFeatureNotEnabledError, + AnonymousSessionIntrospectError, + AnonymousSessionLogoutError, + AnonymousSessionResourceServerError, + AnonymousSessionScopeError, + AnonymousSessionTokenError, + ConfigurationError, + DomainResolverError, +) + +# Shared fixtures +DOMAIN = "auth0.local" +CLIENT_ID = "" +CLIENT_SECRET = "" +SECRET = "test-secret-long-enough-for-encryption" + + +class OneSlotStore: + """ + Models StatelessStateStore: a store identifier is a salt, not a location. + One physical slot per instance, so a mismatched identifier reads as + absent, not as a different record. AsyncMock cannot catch a collision + because it treats every identifier as a distinct key, so this fake is + required instead. + """ + + def __init__(self): + self.slot = None + + async def set(self, identifier, state, options=None): + self.slot = (identifier, state) + + async def get(self, identifier, options=None): + if not self.slot or self.slot[0] != identifier: + return None + return self.slot[1] + + async def delete(self, identifier, options=None): + self.slot = None + + +def _make_client(anonymous_store=None, **kwargs) -> AnonymousClient: + return AnonymousClient( + domain=DOMAIN, + client_id=CLIENT_ID, + client_secret=CLIENT_SECRET, + secret=SECRET, + anonymous_store=anonymous_store, + **kwargs, + ) + + +def _fake_response(status_code=200, body=None): + response = MagicMock() + response.status_code = status_code + response.json = MagicMock(return_value=body or {}) + return response + + +class _FakeAsyncClient: + """Patches httpx.AsyncClient. Call sequence maps 1:1 to responses.""" + + def __init__(self, responses): + self._responses = list(responses) + self.calls = [] + + def __call__(self, *args, **kwargs): + return self + + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + return False + + async def post(self, url, **kwargs): + self.calls.append(("POST", url, kwargs)) + return self._responses.pop(0) + + async def get(self, url, **kwargs): + self.calls.append(("GET", url, kwargs)) + return self._responses.pop(0) + + +def _token_response( + access_token="AT1", # noqa: S107 + expires_in=3600, + session_token="ST1", # noqa: S107 + session_expires_in=2592000, + sub="anon@abc", + session_id="sid1", +): + return { + "access_token": access_token, + "token_type": "Bearer", + "expires_in": expires_in, + "session_token": session_token, + "session_expires_in": session_expires_in, + "sub": sub, + "session_id": session_id, + } + + +def _stored_context(store: OneSlotStore, **overrides): + defaults = { + "session_token": "ST1", + "sub": "anon@abc", + "session_id": "sid1", + "access_token": "AT1", + "expires_at": int(time.time()) + 3600, + "created_at": int(time.time()), + } + defaults.update(overrides) + context = AnonymousSessionContext(**defaults) + encrypted = encrypt(context.model_dump(), SECRET, "anon_session") + store.slot = (ANON_IDENTIFIER, {"context": encrypted}) + return context + + +# ── Constructor ────────────────────────────────────────────────────────────── + +class TestAnonymousClientConstructor: + def test_constructor_sets_properties(self): + client = _make_client() + assert client._domain == DOMAIN + assert client._domain_resolver is None + assert client._client_id == CLIENT_ID + assert client._anonymous_store is None + + def test_constructor_accepts_callable_domain(self): + resolver = AsyncMock(return_value="tenant.auth0.local") + client = AnonymousClient( + domain=resolver, client_id=CLIENT_ID, client_secret=CLIENT_SECRET, secret=SECRET + ) + assert client._domain is None + assert client._domain_resolver is resolver + + def test_no_dpop_key_parameter_exists(self): + """Structural guard: AnonymousClient has no dpop_key parameter anywhere.""" + for name, method in inspect.getmembers(AnonymousClient, predicate=inspect.isfunction): + sig = inspect.signature(method) + assert "dpop_key" not in sig.parameters, f"{name} must never accept dpop_key" + + +# ── Fail-closed store isolation ─────────────────────────────────────────────── + +class TestStoreIsolation: + @pytest.mark.asyncio + async def test_create_session_without_store_raises_configuration_error(self): + client = _make_client(anonymous_store=None) + with pytest.raises(ConfigurationError): + await client.create_session(audience="aud", scope="s") + + @pytest.mark.asyncio + async def test_get_token_without_store_raises_configuration_error(self): + client = _make_client(anonymous_store=None) + with pytest.raises(ConfigurationError): + await client.get_token() + + @pytest.mark.asyncio + async def test_introspect_without_store_raises_configuration_error(self): + client = _make_client(anonymous_store=None) + with pytest.raises(ConfigurationError): + await client.introspect() + + @pytest.mark.asyncio + async def test_logout_without_store_raises_configuration_error(self): + client = _make_client(anonymous_store=None) + with pytest.raises(ConfigurationError): + await client.logout() + + @pytest.mark.asyncio + async def test_no_write_attempted_when_store_missing(self): + """Fails closed before any store write, never falls back to another store.""" + client = _make_client(anonymous_store=None) + with patch("httpx.AsyncClient") as mock_http: + with pytest.raises(ConfigurationError): + await client.create_session(audience="aud", scope="s") + mock_http.assert_not_called() + + @pytest.mark.asyncio + async def test_get_session_token_for_injection_returns_none_without_store(self): + client = _make_client(anonymous_store=None) + assert await client.get_session_token_for_injection() is None + + +# ── create_session ──────────────────────────────────────────────────────────── + +class TestCreateSession: + @pytest.mark.asyncio + async def test_create_session_success(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(200, _token_response())]) + with patch("httpx.AsyncClient", fake_http): + session = await client.create_session( + audience="https://api.example.com", scope="read:cart", metadata={"cart_id": "c1"} + ) + assert session.sub == "anon@abc" + assert session.session_id == "sid1" + assert session.is_new is True + assert session.metadata == {"cart_id": "c1"} + + @pytest.mark.asyncio + async def test_create_session_response_missing_session_token_raises(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + response_without_session_token = _token_response() + del response_without_session_token["session_token"] + fake_http = _FakeAsyncClient([_fake_response(200, response_without_session_token)]) + with patch("httpx.AsyncClient", fake_http): + with pytest.raises(AnonymousSessionCreateError): + await client.create_session(audience="aud", scope="s") + + @pytest.mark.asyncio + async def test_create_session_sends_client_secret_in_json_body_not_auth_tuple(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(200, _token_response())]) + with patch("httpx.AsyncClient", fake_http): + await client.create_session(audience="aud", scope="s") + _, _, kwargs = fake_http.calls[0] + assert kwargs["json"]["client_secret"] == CLIENT_SECRET + assert "auth" not in kwargs + + @pytest.mark.asyncio + async def test_create_session_never_attaches_dpop_header(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(200, _token_response())]) + with patch("httpx.AsyncClient", fake_http): + await client.create_session(audience="aud", scope="s") + _, _, kwargs = fake_http.calls[0] + assert "DPoP" not in kwargs.get("headers", {}) + + @pytest.mark.asyncio + async def test_create_session_persists_at_distinct_location_from_state_store(self): + """The anonymous store instance is separate from any authenticated session store.""" + anon_store = OneSlotStore() + state_store = OneSlotStore() + state_store.slot = ("_a0_session", {"user": "authenticated"}) + client = _make_client(anonymous_store=anon_store) + fake_http = _FakeAsyncClient([_fake_response(200, _token_response())]) + with patch("httpx.AsyncClient", fake_http): + await client.create_session(audience="aud", scope="s") + assert anon_store.slot[0] == ANON_IDENTIFIER + # The authenticated session store is a different instance entirely, + # never touched by anonymous writes. + assert state_store.slot == ("_a0_session", {"user": "authenticated"}) + + @pytest.mark.asyncio + async def test_metadata_over_1kb_rejected_client_side_no_network_call(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + oversized = {"blob": "x" * 2000} + with patch("httpx.AsyncClient") as mock_http: + with pytest.raises(AnonymousSessionCreateError, match="1KB"): + await client.create_session(audience="aud", scope="s", metadata=oversized) + mock_http.assert_not_called() + + @pytest.mark.asyncio + async def test_dangerous_metadata_key_rejected(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + with pytest.raises(AnonymousSessionCreateError, match="not allowed"): + await client.create_session(audience="aud", scope="s", metadata={"__proto__": "x"}) + + @pytest.mark.asyncio + async def test_non_string_metadata_value_accepted(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(200, _token_response())]) + with patch("httpx.AsyncClient", fake_http): + await client.create_session( + audience="aud", scope="s", metadata={"count": 5, "active": True, "tags": ["a", "b"]} + ) + _, _, kwargs = fake_http.calls[0] + assert kwargs["json"]["metadata"] == {"count": 5, "active": True, "tags": ["a", "b"]} + + @pytest.mark.asyncio + async def test_non_json_serializable_metadata_value_rejected(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + with pytest.raises(AnonymousSessionCreateError, match="JSON-serializable"): + await client.create_session(audience="aud", scope="s", metadata={"bad": object()}) + + @pytest.mark.asyncio + async def test_create_session_accepts_options_model(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(200, _token_response())]) + options = CreateAnonymousSessionOptions( + audience="aud", scope="s", metadata={"cart_id": "c1"} + ) + with patch("httpx.AsyncClient", fake_http): + await client.create_session(options=options) + _, _, kwargs = fake_http.calls[0] + assert kwargs["json"]["audience"] == "aud" + assert kwargs["json"]["scope"] == "s" + assert kwargs["json"]["metadata"] == {"cart_id": "c1"} + + @pytest.mark.asyncio + async def test_create_session_accepts_options_dict(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(200, _token_response())]) + with patch("httpx.AsyncClient", fake_http): + await client.create_session(options={"audience": "aud", "scope": "s"}) + _, _, kwargs = fake_http.calls[0] + assert kwargs["json"]["audience"] == "aud" + assert kwargs["json"]["scope"] == "s" + + @pytest.mark.asyncio + async def test_explicit_kwargs_override_options(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(200, _token_response())]) + with patch("httpx.AsyncClient", fake_http): + await client.create_session( + options={"audience": "from_options"}, audience="from_kwarg" + ) + _, _, kwargs = fake_http.calls[0] + assert kwargs["json"]["audience"] == "from_kwarg" + + @pytest.mark.asyncio + async def test_invalid_options_dict_raises_typed_error_no_network_call(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + with patch("httpx.AsyncClient") as mock_http: + with pytest.raises(AnonymousSessionCreateError) as exc: + await client.create_session(options={"unknown_field": "x"}) + assert exc.value.code == "invalid_options" + mock_http.assert_not_called() + + @pytest.mark.asyncio + async def test_feature_not_enabled_maps_to_typed_error(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([ + _fake_response(403, {"error": "feature_not_enabled", "error_description": "disabled"}) + ]) + with patch("httpx.AsyncClient", fake_http): + with pytest.raises(AnonymousSessionFeatureNotEnabledError): + await client.create_session(audience="aud", scope="s") + + @pytest.mark.asyncio + async def test_unauthorized_client_maps_to_typed_error(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([ + _fake_response(403, {"error": "unauthorized_client", "error_description": "not enabled"}) + ]) + with patch("httpx.AsyncClient", fake_http): + with pytest.raises(AnonymousSessionClientNotEnabledError): + await client.create_session(audience="aud", scope="s") + + @pytest.mark.asyncio + async def test_dpop_required_client_maps_to_not_supported_with_literal_message(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + message = "Client configuration requires the use of Proof-of-Possession mechanism" + fake_http = _FakeAsyncClient([ + _fake_response(400, {"error": "unauthorized_client", "error_description": message}) + ]) + with patch("httpx.AsyncClient", fake_http): + with pytest.raises(AnonymousSessionClientNotSupportedError) as exc: + await client.create_session(audience="aud", scope="s") + assert message in str(exc.value) + + @pytest.mark.asyncio + async def test_invalid_target_maps_to_resource_server_error(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([ + _fake_response(400, {"error": "invalid_target", "error_description": "bad audience"}) + ]) + with patch("httpx.AsyncClient", fake_http): + with pytest.raises(AnonymousSessionResourceServerError): + await client.create_session(audience="aud", scope="s") + + @pytest.mark.asyncio + async def test_invalid_scope_maps_to_scope_error(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([ + _fake_response(400, {"error": "invalid_scope", "error_description": "bad scope"}) + ]) + with patch("httpx.AsyncClient", fake_http): + with pytest.raises(AnonymousSessionScopeError): + await client.create_session(audience="aud", scope="s") + + @pytest.mark.asyncio + async def test_network_failure_raises_create_error(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + + class _RaisingClient: + def __call__(self, *a, **k): + return self + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + async def post(self, *a, **k): + raise httpx.ConnectError("boom") + + with patch("httpx.AsyncClient", _RaisingClient()): + with pytest.raises(AnonymousSessionCreateError): + await client.create_session(audience="aud", scope="s") + + +# ── get_token (renewal ladder) ──────────────────────────────────────────────── + +class TestGetToken: + @pytest.mark.asyncio + async def test_fresh_cached_token_returned_with_no_http_call(self): + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) + 3600) + client = _make_client(anonymous_store=store) + with patch("httpx.AsyncClient") as mock_http: + session = await client.get_token() + mock_http.assert_not_called() + assert session.is_new is False + assert session.access_token == "AT1" + + @pytest.mark.asyncio + async def test_no_active_session_raises_token_error(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + with pytest.raises(AnonymousSessionTokenError): + await client.get_token() + + @pytest.mark.asyncio + async def test_expired_access_token_remints_via_session_token_grant(self): + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) - 10) + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([ + _fake_response(200, {"access_token": "AT2", "token_type": "Bearer", "expires_in": 3600}) + ]) + with patch("httpx.AsyncClient", fake_http): + session = await client.get_token() + assert session.access_token == "AT2" + assert session.is_new is False + assert session.sub == "anon@abc" # unchanged on ordinary re-mint + _, _, kwargs = fake_http.calls[0] + assert kwargs["json"]["session_token"] == "ST1" + assert "refresh_token" not in kwargs["json"] + + @pytest.mark.asyncio + async def test_remint_preserves_empty_string_fields_instead_of_falling_back_to_stale_context( + self, + ): + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) - 10) + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([ + _fake_response( + 200, + { + "access_token": "AT2", + "token_type": "Bearer", + "expires_in": 3600, + "session_token": "", + "sub": "", + "session_id": "", + }, + ) + ]) + with patch("httpx.AsyncClient", fake_http): + session = await client.get_token() + assert session.sub == "" + assert session.session_id == "" + token = await client.get_session_token_for_injection() + assert token == "" + + @pytest.mark.asyncio + async def test_expired_session_token_triggers_silent_new_session(self): + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) - 10) + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([ + _fake_response(400, {"error": "session_expired", "error_description": "expired"}), + _fake_response(200, _token_response(sub="anon@new", session_id="sid2")), + ]) + with patch("httpx.AsyncClient", fake_http): + session = await client.get_token() + assert session.is_new is True + assert session.sub == "anon@new" + + @pytest.mark.asyncio + async def test_silent_remint_drops_metadata(self): + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) - 10, metadata={"cart_id": "c1"}) + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([ + _fake_response(400, {"error": "invalid_session_token", "error_description": "bad"}), + _fake_response(200, _token_response()), + ]) + with patch("httpx.AsyncClient", fake_http): + session = await client.get_token() + assert session.metadata is None + + @pytest.mark.asyncio + async def test_two_consecutive_session_expired_raises_not_loops(self): + """Retry-once bound: exactly 2 upstream POSTs, then raise.""" + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) - 10) + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([ + _fake_response(400, {"error": "session_expired", "error_description": "expired"}), + _fake_response(400, {"error": "session_expired", "error_description": "expired again"}), + ]) + with patch("httpx.AsyncClient", fake_http): + with pytest.raises(AnonymousSessionCreateError): + await client.get_token() + assert len(fake_http.calls) == 2 + + @pytest.mark.asyncio + async def test_other_error_code_raises_typed_error_no_retry(self): + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) - 10) + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([ + _fake_response(403, {"error": "feature_not_enabled", "error_description": "off"}), + ]) + with patch("httpx.AsyncClient", fake_http): + with pytest.raises(AnonymousSessionFeatureNotEnabledError): + await client.get_token() + assert len(fake_http.calls) == 1 + + @pytest.mark.asyncio + async def test_corrupted_stored_token_triggers_silent_new_session(self): + store = OneSlotStore() + store.slot = (ANON_IDENTIFIER, {"context": "not-a-valid-jwe"}) + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(200, _token_response(sub="anon@fresh"))]) + with patch("httpx.AsyncClient", fake_http): + session = await client.get_token() + assert session.is_new is True + assert session.sub == "anon@fresh" + + @pytest.mark.asyncio + async def test_network_error_during_renewal_not_misclassified_as_expiry(self): + """A broad exception must never be silently treated as session_expired.""" + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) - 10) + client = _make_client(anonymous_store=store) + + class _RaisingClient: + def __call__(self, *a, **k): + return self + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + async def post(self, *a, **k): + raise httpx.ConnectError("network down") + + with patch("httpx.AsyncClient", _RaisingClient()): + with pytest.raises(AnonymousSessionTokenError): + await client.get_token() + + @pytest.mark.asyncio + async def test_get_token_never_writes_to_authenticated_state_store(self): + anon_store = OneSlotStore() + _stored_context(anon_store, expires_at=int(time.time()) + 3600) + auth_state_store = AsyncMock() + client = _make_client(anonymous_store=anon_store) + await client.get_token() + auth_state_store.set.assert_not_called() + auth_state_store.get.assert_not_called() + auth_state_store.delete.assert_not_called() + + +# ── MCD / cross-tenant isolation ─────────────────────────────────────────────── + +class TestMcdIsolation: + @pytest.mark.asyncio + async def test_domain_mismatch_in_resolver_mode_mints_fresh_under_current_tenant(self): + store = OneSlotStore() + _stored_context( + store, expires_at=int(time.time()) + 3600, domain="tenant-a.auth0.local" + ) + resolver = AsyncMock(return_value="tenant-b.auth0.local") + client = _make_client(anonymous_store=store) + client._domain_resolver = resolver + client._domain = None + fake_http = _FakeAsyncClient([_fake_response(200, _token_response(sub="anon@fresh-b"))]) + with patch("httpx.AsyncClient", fake_http): + session = await client.get_token() + assert session.sub == "anon@fresh-b" + assert session.is_new is True + + @pytest.mark.asyncio + async def test_domain_resolver_failure_propagates(self): + resolver = AsyncMock(return_value=None) + client = AnonymousClient( + domain=resolver, client_id=CLIENT_ID, client_secret=CLIENT_SECRET, secret=SECRET, + anonymous_store=OneSlotStore(), + ) + with pytest.raises(DomainResolverError): + await client.create_session(audience="aud", scope="s") + + +# ── introspect ──────────────────────────────────────────────────────────────── + +class TestIntrospect: + @pytest.mark.asyncio + async def test_introspect_issues_get_with_no_body(self): + store = OneSlotStore() + _stored_context(store) + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(200, {"sub": "anon@abc"})]) + with patch("httpx.AsyncClient", fake_http): + await client.introspect() + method, _, kwargs = fake_http.calls[0] + assert method == "GET" + assert "json" not in kwargs + + @pytest.mark.asyncio + async def test_introspect_lenient_decode_ignores_unknown_fields(self): + store = OneSlotStore() + _stored_context(store) + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([ + _fake_response(200, {"sub": "anon@abc", "totally_unexpected_field": "value"}) + ]) + with patch("httpx.AsyncClient", fake_http): + result = await client.introspect() + assert result.sub == "anon@abc" + + @pytest.mark.asyncio + async def test_introspect_missing_optional_field_does_not_raise(self): + store = OneSlotStore() + _stored_context(store) + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(200, {"sub": "anon@abc"})]) + with patch("httpx.AsyncClient", fake_http): + result = await client.introspect() + assert result.session_id is None + assert result.metadata is None + + @pytest.mark.asyncio + async def test_introspect_never_writes_to_store(self): + store = OneSlotStore() + _stored_context(store) + original_slot = store.slot + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(200, {"sub": "anon@abc"})]) + with patch("httpx.AsyncClient", fake_http): + await client.introspect() + assert store.slot == original_slot + + @pytest.mark.asyncio + async def test_introspect_no_active_session_raises(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + with pytest.raises(AnonymousSessionIntrospectError): + await client.introspect() + + +# ── logout ──────────────────────────────────────────────────────────────────── + +class TestLogout: + @pytest.mark.asyncio + async def test_logout_clears_anonymous_store(self): + store = OneSlotStore() + _stored_context(store) + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(200, {})]) + with patch("httpx.AsyncClient", fake_http): + await client.logout() + assert store.slot is None + + @pytest.mark.asyncio + async def test_logout_does_not_touch_unrelated_authenticated_store(self): + anon_store = OneSlotStore() + _stored_context(anon_store) + auth_store = AsyncMock() + client = _make_client(anonymous_store=anon_store) + fake_http = _FakeAsyncClient([_fake_response(200, {})]) + with patch("httpx.AsyncClient", fake_http): + await client.logout() + auth_store.delete.assert_not_called() + + @pytest.mark.asyncio + async def test_get_token_after_logout_behaves_as_no_session(self): + store = OneSlotStore() + _stored_context(store) + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(200, {})]) + with patch("httpx.AsyncClient", fake_http): + await client.logout() + with pytest.raises(AnonymousSessionTokenError): + await client.get_token() + + @pytest.mark.asyncio + async def test_logout_with_no_session_is_a_noop(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + with patch("httpx.AsyncClient") as mock_http: + await client.logout() + mock_http.assert_not_called() + + @pytest.mark.asyncio + async def test_logout_remote_call_failure_still_clears_local_state_then_raises(self): + store = OneSlotStore() + _stored_context(store) + client = _make_client(anonymous_store=store) + + class _RaisingClient: + def __call__(self, *a, **k): + return self + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + async def post(self, *a, **k): + raise httpx.ConnectError("boom") + + with patch("httpx.AsyncClient", _RaisingClient()): + with pytest.raises(AnonymousSessionLogoutError): + await client.logout() + assert store.slot is None + + @pytest.mark.asyncio + async def test_logout_non_200_response_still_clears_local_state_then_raises(self): + store = OneSlotStore() + _stored_context(store) + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient( + [_fake_response(500, {"error": "server_error", "error_description": "boom"})] + ) + with patch("httpx.AsyncClient", fake_http): + with pytest.raises(AnonymousSessionLogoutError): + await client.logout() + assert store.slot is None + + @pytest.mark.asyncio + async def test_logout_session_expired_response_is_not_raised(self): + store = OneSlotStore() + _stored_context(store) + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient( + [_fake_response(400, {"error": "session_expired", "error_description": "expired"})] + ) + with patch("httpx.AsyncClient", fake_http): + await client.logout() + assert store.slot is None + + +# ── get_session_token_for_injection (login-injection support) ─────────────── + +class TestGetSessionTokenForInjection: + @pytest.mark.asyncio + async def test_returns_token_when_active_session_exists(self): + store = OneSlotStore() + _stored_context(store, session_token="REAL_TOKEN") + client = _make_client(anonymous_store=store) + token = await client.get_session_token_for_injection() + assert token == "REAL_TOKEN" + + @pytest.mark.asyncio + async def test_returns_none_when_no_session(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + assert await client.get_session_token_for_injection() is None + + @pytest.mark.asyncio + async def test_returns_none_never_raises_on_corrupted_token(self): + """Malformed stored token must deny the link, never abort the caller.""" + store = OneSlotStore() + store.slot = (ANON_IDENTIFIER, {"context": "garbage"}) + client = _make_client(anonymous_store=store) + assert await client.get_session_token_for_injection() is None + + @pytest.mark.asyncio + async def test_returns_none_on_store_exception_never_raises(self): + store = AsyncMock() + store.get = AsyncMock(side_effect=RuntimeError("store unavailable")) + client = _make_client(anonymous_store=store) + assert await client.get_session_token_for_injection() is None diff --git a/src/auth0_server_python/tests/test_server_client.py b/src/auth0_server_python/tests/test_server_client.py index c1c012a..0bdb136 100644 --- a/src/auth0_server_python/tests/test_server_client.py +++ b/src/auth0_server_python/tests/test_server_client.py @@ -10,10 +10,12 @@ from jwcrypto import jwk from auth0_server_python.auth_schemes.dpop_auth import DPoPAuth +from auth0_server_python.auth_server.anonymous_client import ANON_IDENTIFIER, AnonymousClient from auth0_server_python.auth_server.mfa_client import MfaClient from auth0_server_python.auth_server.my_account_client import MyAccountClient -from auth0_server_python.auth_server.server_client import ServerClient +from auth0_server_python.auth_server.server_client import INTERNAL_AUTHORIZE_PARAMS, ServerClient from auth0_server_python.auth_types import ( + AnonymousSessionContext, CompleteConnectAccountRequest, ConnectAccountOptions, ConnectAccountRequest, @@ -39,6 +41,7 @@ TransactionData, UserClaims, ) +from auth0_server_python.encryption.encrypt import encrypt from auth0_server_python.error import ( AccessTokenError, AccessTokenErrorCode, @@ -8947,3 +8950,465 @@ async def test_complete_interactive_login_milliseconds_ceiling_fails_open(mocker mock_state_store.set.assert_awaited_once() stored_state = mock_state_store.set.call_args.args[1] assert stored_state.internal.session_expires_at is None + + +# ============================================================================= +# ANONYMOUS SESSIONS - WIRING AND LOGIN-INJECTION TESTS +# ============================================================================= + + +class _OneSlotStore: + """ + Models a store where a store identifier is used only as an encryption + salt, not a location key. One physical slot per instance. AsyncMock + cannot exercise this collision because it treats every identifier as a + distinct key. + """ + + def __init__(self): + self.slot = None + + async def set(self, identifier, state, options=None): + self.slot = (identifier, state) + + async def get(self, identifier, options=None): + if not self.slot or self.slot[0] != identifier: + return None + return self.slot[1] + + async def delete(self, identifier, options=None): + self.slot = None + + +def _make_anon_context(secret, **overrides): + defaults = { + "session_token": "ANON_TOKEN_1", + "sub": "anon@abc", + "session_id": "sid1", + "access_token": "anon_at1", + "expires_at": int(time.time()) + 3600, + "created_at": int(time.time()), + } + defaults.update(overrides) + context = AnonymousSessionContext(**defaults) + return encrypt(context.model_dump(), secret, "anon_session") + + +@pytest.mark.asyncio +async def test_server_client_anonymous_property(): + """ServerClient exposes an 'anonymous' property returning an AnonymousClient instance.""" + client = ServerClient( + domain="auth0.local", + client_id="cid", + client_secret="csecret", + secret="a-test-secret-with-enough-length", + transaction_store=AsyncMock(), + state_store=AsyncMock(), + ) + assert isinstance(client.anonymous, AnonymousClient) + + +@pytest.mark.asyncio +async def test_anonymous_client_receives_own_store_not_state_store(): + """The anonymous client must never share the authenticated state store instance.""" + state_store = AsyncMock() + anon_store = _OneSlotStore() + client = ServerClient( + domain="auth0.local", + client_id="cid", + client_secret="csecret", + secret="a-test-secret-with-enough-length", + transaction_store=AsyncMock(), + state_store=state_store, + anonymous_store=anon_store, + ) + assert client.anonymous._anonymous_store is anon_store + assert client.anonymous._anonymous_store is not state_store + + +@pytest.mark.asyncio +async def test_start_interactive_login_no_anonymous_session_is_byte_identical(mocker): + """No anonymous store configured -> injection is a complete no-op, existing behaviour unchanged.""" + mock_transaction_store = AsyncMock() + mock_state_store = AsyncMock() + client = ServerClient( + domain="auth0.local", + client_id="", + client_secret="", + state_store=mock_state_store, + transaction_store=mock_transaction_store, + secret="some-secret", + authorization_params={"redirect_uri": "/test_redirect_uri"}, + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"authorization_endpoint": "https://auth0.local/authorize"}, + ) + captured = {} + + def fake_create_url(endpoint, **kwargs): + captured.update(kwargs) + return ("https://auth0.local/authorize?client_id=", "some_state") + + mocker.patch.object(client._oauth, "create_authorization_url", side_effect=fake_create_url) + await client.start_interactive_login() + assert "session_token" not in captured + + +@pytest.mark.asyncio +async def test_start_interactive_login_injects_active_anonymous_session(mocker): + secret = "a-test-secret-with-enough-length" + anon_store = _OneSlotStore() + anon_store.slot = (ANON_IDENTIFIER, {"context": _make_anon_context(secret)}) + client = ServerClient( + domain="auth0.local", + client_id="", + client_secret="", + state_store=AsyncMock(), + transaction_store=AsyncMock(), + anonymous_store=anon_store, + secret=secret, + authorization_params={"redirect_uri": "/test_redirect_uri"}, + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"authorization_endpoint": "https://auth0.local/authorize"}, + ) + captured = {} + + def fake_create_url(endpoint, **kwargs): + captured.update(kwargs) + return ("https://auth0.local/authorize?client_id=", "some_state") + + mocker.patch.object(client._oauth, "create_authorization_url", side_effect=fake_create_url) + await client.start_interactive_login() + assert captured.get("session_token") == "ANON_TOKEN_1" + + +@pytest.mark.asyncio +async def test_start_interactive_login_stamps_session_token_into_transaction_data(mocker): + secret = "a-test-secret-with-enough-length" + anon_store = _OneSlotStore() + anon_store.slot = (ANON_IDENTIFIER, {"context": _make_anon_context(secret)}) + mock_transaction_store = AsyncMock() + client = ServerClient( + domain="auth0.local", + client_id="", + client_secret="", + state_store=AsyncMock(), + transaction_store=mock_transaction_store, + anonymous_store=anon_store, + secret=secret, + authorization_params={"redirect_uri": "/test_redirect_uri"}, + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"authorization_endpoint": "https://auth0.local/authorize"}, + ) + mocker.patch.object( + client._oauth, + "create_authorization_url", + return_value=("https://auth0.local/authorize?client_id=", "some_state"), + ) + await client.start_interactive_login() + stored_tx = mock_transaction_store.set.call_args.args[1] + assert stored_tx.session_token == "ANON_TOKEN_1" + + +@pytest.mark.asyncio +async def test_start_interactive_login_absent_session_no_param(mocker): + """An empty anonymous store behaves exactly like no anonymous_store configured.""" + anon_store = _OneSlotStore() # no session ever created + client = ServerClient( + domain="auth0.local", + client_id="", + client_secret="", + state_store=AsyncMock(), + transaction_store=AsyncMock(), + anonymous_store=anon_store, + secret="a-test-secret-with-enough-length", + authorization_params={"redirect_uri": "/test_redirect_uri"}, + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"authorization_endpoint": "https://auth0.local/authorize"}, + ) + captured = {} + + def fake_create_url(endpoint, **kwargs): + captured.update(kwargs) + return ("https://auth0.local/authorize?client_id=", "some_state") + + mocker.patch.object(client._oauth, "create_authorization_url", side_effect=fake_create_url) + await client.start_interactive_login() + assert "session_token" not in captured + + +@pytest.mark.asyncio +async def test_start_interactive_login_malformed_anonymous_token_denies_link_allows_login(mocker): + """Undecryptable stored token: deny the link, never abort the login.""" + anon_store = _OneSlotStore() + anon_store.slot = (ANON_IDENTIFIER, {"context": "not-a-valid-jwe"}) + client = ServerClient( + domain="auth0.local", + client_id="", + client_secret="", + state_store=AsyncMock(), + transaction_store=AsyncMock(), + anonymous_store=anon_store, + secret="a-test-secret-with-enough-length", + authorization_params={"redirect_uri": "/test_redirect_uri"}, + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"authorization_endpoint": "https://auth0.local/authorize"}, + ) + mocker.patch.object( + client._oauth, + "create_authorization_url", + return_value=("https://auth0.local/authorize?client_id=", "some_state"), + ) + url = await client.start_interactive_login() + assert url == "https://auth0.local/authorize?client_id=" + + +@pytest.mark.asyncio +async def test_start_interactive_login_suppresses_injection_on_par_branch(mocker): + """PAR is not supported for anonymous sessions.""" + secret = "a-test-secret-with-enough-length" + anon_store = _OneSlotStore() + anon_store.slot = (ANON_IDENTIFIER, {"context": _make_anon_context(secret)}) + client = ServerClient( + domain="auth0.local", + client_id="", + client_secret="", + state_store=AsyncMock(), + transaction_store=AsyncMock(), + anonymous_store=anon_store, + secret=secret, + authorization_params={"redirect_uri": "/test_redirect_uri", "response_type": "code"}, + pushed_authorization_requests=True, + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={ + "authorization_endpoint": "https://auth0.local/authorize", + "pushed_authorization_request_endpoint": "https://auth0.local/oauth/par", + }, + ) + captured = {} + + class _FakePost: + status_code = 201 + + def json(self): + return {"request_uri": "urn:ietf:params:oauth:request_uri:xyz"} + + class _FakeHttpClient: + def __init__(self, *a, **k): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + async def post(self, url, **kwargs): + captured.update(kwargs.get("data", {})) + return _FakePost() + + mocker.patch("httpx.AsyncClient", _FakeHttpClient) + await client.start_interactive_login() + assert "session_token" not in captured + + +@pytest.mark.asyncio +async def test_start_interactive_login_constructor_fixation_blocked_no_active_session(): + """ + The exact vector where a caller supplies session_token via constructor + authorization_params, with no active anonymous session. INTERNAL_AUTHORIZE_PARAMS + alone cannot block this, since it only filters per-call options.authorization_params. + The unconditional pop() at the injection site must. + """ + assert "session_token" in INTERNAL_AUTHORIZE_PARAMS # belt-and-braces still present + + anon_store = _OneSlotStore() # no session -> the vulnerable case + client = ServerClient( + domain="auth0.local", + client_id="", + client_secret="", + state_store=AsyncMock(), + transaction_store=AsyncMock(), + anonymous_store=anon_store, + secret="a-test-secret-with-enough-length", + authorization_params={ + "redirect_uri": "/test_redirect_uri", + "session_token": "ATTACKER_SUPPLIED", + }, + ) + with patch.object( + client, + "_get_oidc_metadata_cached", + AsyncMock(return_value={"authorization_endpoint": "https://auth0.local/authorize"}), + ): + captured = {} + + def fake_create_url(endpoint, **kwargs): + captured.update(kwargs) + return ("https://auth0.local/authorize?client_id=", "some_state") + + with patch.object(client._oauth, "create_authorization_url", side_effect=fake_create_url): + await client.start_interactive_login() + assert captured.get("session_token") is None + + +@pytest.mark.asyncio +async def test_start_interactive_login_per_call_fixation_also_blocked(mocker): + """The same vector via options.authorization_params (per-call) is caught by the existing filter.""" + anon_store = _OneSlotStore() + client = ServerClient( + domain="auth0.local", + client_id="", + client_secret="", + state_store=AsyncMock(), + transaction_store=AsyncMock(), + anonymous_store=anon_store, + secret="a-test-secret-with-enough-length", + authorization_params={"redirect_uri": "/test_redirect_uri"}, + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"authorization_endpoint": "https://auth0.local/authorize"}, + ) + captured = {} + + def fake_create_url(endpoint, **kwargs): + captured.update(kwargs) + return ("https://auth0.local/authorize?client_id=", "some_state") + + mocker.patch.object(client._oauth, "create_authorization_url", side_effect=fake_create_url) + await client.start_interactive_login( + StartInteractiveLoginOptions(authorization_params={"session_token": "ATTACKER_SUPPLIED"}) + ) + assert captured.get("session_token") is None + + +@pytest.mark.asyncio +async def test_start_interactive_login_does_not_clobber_organization_or_invitation(mocker): + secret = "a-test-secret-with-enough-length" + anon_store = _OneSlotStore() + anon_store.slot = (ANON_IDENTIFIER, {"context": _make_anon_context(secret)}) + client = ServerClient( + domain="auth0.local", + client_id="", + client_secret="", + state_store=AsyncMock(), + transaction_store=AsyncMock(), + anonymous_store=anon_store, + secret=secret, + authorization_params={"redirect_uri": "/test_redirect_uri"}, + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"authorization_endpoint": "https://auth0.local/authorize"}, + ) + captured = {} + + def fake_create_url(endpoint, **kwargs): + captured.update(kwargs) + return ("https://auth0.local/authorize?client_id=", "some_state") + + mocker.patch.object(client._oauth, "create_authorization_url", side_effect=fake_create_url) + await client.start_interactive_login( + StartInteractiveLoginOptions(organization="org_abc123", invitation="inv_xyz") + ) + assert captured.get("organization") == "org_abc123" + assert captured.get("invitation") == "inv_xyz" + assert captured.get("session_token") == "ANON_TOKEN_1" + + +# ── Store-collision regression ───────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_anonymous_write_cannot_destroy_authenticated_session_on_shared_store(): + """ + When the anonymous client is configured with its own store instance, + the authenticated session on a separate store instance is provably + untouched. The separate-instance contract holds. + """ + shared_store = _OneSlotStore() + shared_store.slot = ("_a0_session", {"user": {"sub": "real_user"}}) + + anon_store = _OneSlotStore() + client = ServerClient( + domain="auth0.local", + client_id="cid", + client_secret="csecret", + secret="a-test-secret-with-enough-length", + transaction_store=AsyncMock(), + state_store=shared_store, + anonymous_store=anon_store, + ) + await client.anonymous._anonymous_store.set( + ANON_IDENTIFIER, {"context": _make_anon_context("a-test-secret-with-enough-length")} + ) + + # The authenticated session, on its own store instance, is untouched. + assert shared_store.slot == ("_a0_session", {"user": {"sub": "real_user"}}) + session = await client.get_session() + assert session is not None + assert session.get("user", {}).get("sub") == "real_user" + + +@pytest.mark.asyncio +async def test_missing_anonymous_store_fails_closed_never_falls_back_to_state_store(): + """ + If an integrator forgets anonymous_store, the client must raise before any + write, never silently write anonymous state into ServerClient's state_store. + """ + shared_store = _OneSlotStore() + shared_store.slot = ("_a0_session", {"user": {"sub": "real_user"}}) + client = ServerClient( + domain="auth0.local", + client_id="cid", + client_secret="csecret", + secret="a-test-secret-with-enough-length", + transaction_store=AsyncMock(), + state_store=shared_store, + # anonymous_store intentionally omitted + ) + with pytest.raises(ConfigurationError): + await client.anonymous.create_session(audience="aud", scope="s") + # The authenticated session store is completely untouched by the failed attempt. + assert shared_store.slot == ("_a0_session", {"user": {"sub": "real_user"}}) + + +@pytest.mark.asyncio +async def test_get_session_and_get_user_unaffected_by_active_anonymous_session(): + """Anonymous state never touches _a0_session. get_session()/get_user() see no new keys.""" + secret = "a-test-secret-with-enough-length" + anon_store = _OneSlotStore() + anon_store.slot = (ANON_IDENTIFIER, {"context": _make_anon_context(secret)}) + mock_state_store = AsyncMock() + mock_state_store.get = AsyncMock(return_value=None) + client = ServerClient( + domain="auth0.local", + client_id="cid", + client_secret="csecret", + secret=secret, + transaction_store=AsyncMock(), + state_store=mock_state_store, + anonymous_store=anon_store, + ) + assert await client.get_session() is None + assert await client.get_user() is None