diff --git a/.gitignore b/.gitignore index 24939dc..4a79d71 100644 --- a/.gitignore +++ b/.gitignore @@ -27,4 +27,5 @@ test-script.py coverage.xml # AI tools -.claude \ No newline at end of file +.claude +.worktrees diff --git a/README.md b/README.md index f1778e5..db0b789 100644 --- a/README.md +++ b/README.md @@ -256,7 +256,11 @@ Bind tokens to a key your server holds ([RFC 9449](https://www.rfc-editor.org/rf Sign users in with a one-time code sent by email or SMS, or with a magic link sent by email, via [Auth0 embedded passwordless login](https://auth0.com/docs/authenticate/passwordless/implement-login/embedded-login/relevant-api-endpoints). OTP verification and the magic-link callback each establish a server-side session like every other login path. For prerequisites, both flows, custom scopes/audiences, step-up MFA, and error handling, see [examples/Passwordless.md](examples/Passwordless.md). -### 11. Enterprise Connect (Embedded Login) +### 11. 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). + +### 12. Enterprise Connect (Embedded Login) Sign users in through their company's identity provider while **your application owns the session**. Opt in with `enterprise_connect=True`; Auth0 acts as a pure SSO relay and issues no refresh token. `start_enterprise_login()` discovers whether an email domain is managed and returns an authorization URL or `None`, and `complete_interactive_login()` returns the verified claims and access token for your app to build its own session from. Early Access. For discovery, the callback contract, multi-tenant `org_id` checks, and federated logout, see [examples/EnterpriseConnect.md](examples/EnterpriseConnect.md). diff --git a/examples/AnonymousSessions.md b/examples/AnonymousSessions.md new file mode 100644 index 0000000..d538540 --- /dev/null +++ b/examples/AnonymousSessions.md @@ -0,0 +1,134 @@ +# 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 attached at creation. At login, the SDK carries the session to Auth0 as a short-lived transfer ticket 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. + +## 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) + - [Logging Out](#logging-out) + - [Login Injection](#login-injection) + - [Rate-Limiting `get_token()`](#rate-limiting-get_token) + - [Error Handling](#error-handling) + - [Additional Resources](#additional-resources) + +## Setup + +Before using the anonymous sessions API, the `anonymous_sessions_enabled` flag must be turned on for your tenant, 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**. Any JSON-serializable value is accepted, ≤1 KB total (UTF-8 JSON byte length). Oversized or non-JSON-serializable metadata is rejected client-side before any network call. + +`AnonymousSession` returns `access_token`, `expires_at`, `session_expires_at`, `metadata`, `sub`, and `scope`. + +## 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, since anonymous sessions never issue refresh tokens). +3. Session token also expired or invalid, a **brand-new session is silently created, once**. Both `sub` and `metadata` reset: the visitor gets a new `anon@` identity and any previously attached metadata is gone. 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. + +## 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. + +Authenticated (OIDC) logout also ends an active anonymous session. When you call `ServerClient.logout()` and an anonymous store is configured, the SDK clears the locally-held anonymous session before returning the logout URL. This is a local clear only, with no remote call (consistent with `anonymous.logout()`, which also does not revoke server-side). It prevents the next visitor on a shared device from having the previous visitor's anonymous identity re-linked at their login. If no anonymous session is active, nothing is cleared. + +## Login Injection + +When an anonymous session is active, `start_interactive_login()` automatically injects an `anon_transfer_token` transfer ticket into the `/authorize` URL, no code change needed at your call site. If no anonymous session exists, behavior is unchanged. + +The raw `session_token` never goes on the URL. At the moment the `/authorize` URL is built, the SDK exchanges the stored session token for a short-lived (30s) single-use ticket (`anon_transfer_token`) via `POST /anonymous/token`, and forwards only that ticket as the `anon_transfer_token` query parameter. The raw session token stays inside the SDK's encrypted store and the ticket is never persisted. The ticket is short-lived and grants no authorization on its own, but you should still set `Referrer-Policy: no-referrer` on your login pages and never log the authorize URL. + +If the exchange errors (network failure, a non-200, or an unparseable response), login proceeds with no ticket and no linking, and never aborts the login. + +## 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 `AnonymousSessionError`, carrying a `.code` you can branch on: + +```python +from auth0_server_python.error import ( + AnonymousSessionCreateError, + AnonymousSessionTokenError, +) + +try: + session = await server_client.anonymous.create_session(audience="...", scope="...") +except AnonymousSessionCreateError as e: + if e.code == "feature_not_enabled": + ... +``` + +### Error Hierarchy + +``` +AnonymousSessionError base class, never raised directly + AnonymousSessionCreateError raised by create_session() + AnonymousSessionTokenError raised by get_token() +``` + +### `AnonymousSessionCreateError` codes + +| `.code` | When | +|---------|------| +| `"feature_not_enabled"` | anonymous sessions not enabled on this tenant/client | +| `"anonymous_create_error"` | generic platform error on the create path | +| `"missing_session_token"` | platform response omitted the session token (misconfigured tenant) | +| `"invalid_metadata"` | metadata is not a dict or contains non-JSON-serializable values | +| `"metadata_too_large"` | metadata exceeds the 1 KB limit | +| `"invalid_options"` | unrecognised key in `create_session()` options | + +The platform may return other codes (e.g. `"insufficient_scope"`) and these are passed through on `.code` unchanged. + +### `AnonymousSessionTokenError` codes + +| `.code` | When | +|---------|------| +| `"invalid_session_state"` | stored session could not be decrypted - call `create_session()` to recover | +| `"anonymous_token_error"` | no active session, network error, parse error, or generic platform error on the renewal path | + +> **Note on naming.** The SDK spec names this class `AnonymousSessionTokenExpiredError`. This SDK uses `AnonymousSessionTokenError` - a deliberate broadening, since the class covers all `get_token()` failures, not just expiry. The `.code` values are stable and safe to branch on. diff --git a/references/flow-map.md b/references/flow-map.md index 18f43dc..ea43e35 100644 --- a/references/flow-map.md +++ b/references/flow-map.md @@ -19,6 +19,7 @@ Before working on a flow, read its entry points and supporting modules. Every fl | MCD | any flow — `domain` may be an async resolver | `_resolve_current_domain`, pitfall 5 in `references/pitfalls.md` | `examples/MultipleCustomDomains.md` | | Enterprise Connect | `start_enterprise_login`, `complete_interactive_login` (EC branch), `is_federated_domain` (standalone), `logout` (`federated`) | `auth_types/` (`StartEnterpriseLoginOptions`, `LogoutOptions.federated`), `error/` (`EnterpriseConnectError`); the SDK owns no session in this mode | `examples/EnterpriseConnect.md` | | mTLS client auth | constructor `use_mtls` + `ssl_context` | `_resolve_token_endpoint`, `_apply_client_authentication`, `_warn_if_not_cert_bound`, `mfa_client.py` (`use_mtls`, `ssl_context`, `verify`, `token_endpoint_resolver`) | `examples/MutualTLS.md` | +| Anonymous sessions | `ServerClient.anonymous` (property, returns `AnonymousClient`): `create_session`, `get_token`, `get_session`, `logout`. `start_interactive_login` injects an `anon_transfer_token` via `AnonymousClient.exchange_transfer_token_for_injection` | `auth_server/anonymous_client.py`, `encryption/encrypt.py` (context encrypted at rest), `store/abstract.py` (requires a dedicated `anonymous_store` instance, sharing the `StateStore` ABC) | `examples/AnonymousSessions.md` | Two rules cut across every flow above, so check them on any change here: resolve the domain through `await self._resolve_current_domain(store_options)` rather than reading `self._domain`, and accept diff --git a/src/auth0_server_python/auth_server/__init__.py b/src/auth0_server_python/auth_server/__init__.py index 52449fb..e4f7393 100644 --- a/src/auth0_server_python/auth_server/__init__.py +++ b/src/auth0_server_python/auth_server/__init__.py @@ -1,3 +1,4 @@ +from .anonymous_client import AnonymousClient from .mfa_client import MfaClient from .my_account_client import MyAccountClient from .passwordless_client import PasswordlessClient @@ -7,6 +8,7 @@ "ServerClient", "MyAccountClient", "MfaClient", + "AnonymousClient", "PasswordlessClient", "is_federated_domain", ] 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..33bd16c --- /dev/null +++ b/src/auth0_server_python/auth_server/anonymous_client.py @@ -0,0 +1,839 @@ +""" +Anonymous Sessions client for auth0-server-python SDK. +Handles pre-login anon@ identity operations against the Auth0 anonymous session API. +""" + +import base64 +import json +import time +from typing import Any, Callable, Optional, Union + +import httpx +from pydantic import ValidationError + +from auth0_server_python.auth_schemes.client_assertion import ( + CLIENT_ASSERTION_TYPE, + build_client_assertion, + validate_client_assertion_key, +) +from auth0_server_python.auth_types import ( + AnonymousCreateTokenResponse, + AnonymousSession, + AnonymousSessionContext, + AnonymousSessionData, + AnonymousTokenResponse, + AnonymousTokenSetEntry, + AnonymousTransferTokenResponse, + CreateAnonymousSessionOptions, +) +from auth0_server_python.encryption.encrypt import decrypt, encrypt +from auth0_server_python.error import ( + AnonymousSessionCreateError, + AnonymousSessionError, + AnonymousSessionTokenError, + ConfigurationError, + DomainResolverError, + _AnonymousSessionExpired, + _SessionDecryptError, +) +from auth0_server_python.utils.helpers import ( + State, + build_domain_resolver_context, + validate_resolved_domain_value, +) + +ANON_IDENTIFIER = "_a0_anon" +ANON_TOKEN_SALT = "anon_session" + +# Audience that mints the login-injection transfer ticket instead of an access token. +TRANSFER_AUDIENCE = "urn:auth0:anon_transfer" + +_METADATA_MAX_BYTES = 1024 + + +class AnonymousClient: + """Client for Auth0 anonymous session operations.""" + + def __init__( + self, + domain: Union[str, Callable, None], + client_id: str, + client_secret: Optional[str], + secret: str, + anonymous_store=None, + default_audience: Optional[str] = None, + default_scope: Optional[str] = None, + headers: Optional[dict[str, str]] = None, + client_assertion_signing_key: Optional[str] = None, + client_assertion_signing_alg: Optional[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._client_assertion_signing_key = client_assertion_signing_key + self._client_assertion_signing_alg = client_assertion_signing_alg or "RS256" + if client_assertion_signing_key: + validate_client_assertion_key( + client_assertion_signing_key, self._client_assertion_signing_alg + ) + 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} + kwargs.setdefault("timeout", 5.0) + return httpx.AsyncClient(headers=headers, **kwargs) + + def _require_store(self) -> None: + """Raise ConfigurationError 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." + ) + + def _apply_client_auth(self, body: dict[str, Any], domain: str) -> None: + """Inject client credentials into a JSON request body. + + Args: + body: The outgoing request body dict, mutated in place. + domain: The target tenant domain, used as the assertion audience. + """ + if self._client_assertion_signing_key: + body["client_assertion"] = build_client_assertion( + self._client_assertion_signing_key, + self._client_id, + f"https://{domain}/", + self._client_assertion_signing_alg, + ) + body["client_assertion_type"] = CLIENT_ASSERTION_TYPE + elif self._client_secret: + body["client_secret"] = self._client_secret + + 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("/") + + # ============================================================================ + # TOKEN SET CACHE HELPERS + # ============================================================================ + + @staticmethod + def _decode_sub(access_token: str) -> Optional[str]: + """Extract the sub claim from a JWT access token without verifying the signature. + + Args: + access_token: The token to decode. + + Returns: + The sub claim, or None for opaque tokens, non-JWT strings, or + tokens with no sub claim. + """ + try: + parts = access_token.split(".") + if len(parts) != 3: + return None + padded = parts[1] + "=" * (-len(parts[1]) % 4) + payload = json.loads(base64.urlsafe_b64decode(padded)) + sub = payload.get("sub") + if not isinstance(sub, str) or not sub.startswith("anon@"): + return None + return sub + except Exception: + return None + + @staticmethod + def _find_token_set( + token_sets: list, + audience: Optional[str], + scope: Optional[str], + ) -> Optional[AnonymousTokenSetEntry]: + for ts in token_sets: + if ts.audience == audience and ts.scope == scope: + return ts + return None + + @staticmethod + def _upsert_token_set( + context: AnonymousSessionContext, + entry: AnonymousTokenSetEntry, + ) -> AnonymousSessionContext: + new_sets = [ + ts for ts in context.token_sets + if not (ts.audience == entry.audience and ts.scope == entry.scope) + ] + new_sets.append(entry) + return context.model_copy(update={"token_sets": new_sets}) + + # ============================================================================ + # 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, + error_data: dict[str, Any], + operation: str, + ) -> Exception: + """Map a server error response to a typed exception. + + Args: + error_data: The parsed error response body. + operation: One of 'create', 'token'. + + 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 operation == "create": + return AnonymousSessionCreateError(description, code=code or "anonymous_create_error", cause=error_data) + if operation == "token": + return AnonymousSessionTokenError(description, code=code or "anonymous_token_error", cause=error_data) + return AnonymousSessionError(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") + 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.model_validate(payload) + except Exception as e: + raise _SessionDecryptError( + "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. + + Raises: + AnonymousSessionCreateError: The request failed, or the response + was invalid or missing required fields. + """ + base_url = f"https://{domain}" + payload: dict[str, Any] = {"client_id": self._client_id} + self._apply_client_auth(payload, domain) + if audience: + payload["audience"] = audience + if scope: + payload["scope"] = scope + if metadata: + payload["metadata"] = metadata + + async with self._get_http_client() as client: + try: + response = await client.post(f"{base_url}/anonymous/token", json=payload) + 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(error_data, "create") + if isinstance(mapped, _AnonymousSessionExpired): + # Internal-only type must never escape. + raise AnonymousSessionCreateError(str(mapped)) + raise mapped + + try: + data = response.json() + except (json.JSONDecodeError, ValueError) as e: + raise AnonymousSessionCreateError("Failed to parse anonymous token response") from e + + if not isinstance(data, dict) or not data.get("session_token"): + raise AnonymousSessionCreateError( + "Anonymous token response contained no session_token. Enable session_token in the response for this tenant.", + code="missing_session_token", + ) + + try: + token_response = AnonymousCreateTokenResponse.model_validate(data) + except (ValueError, ValidationError) as e: + raise AnonymousSessionCreateError("Failed to parse anonymous token response") from e + + now = int(time.time()) + sub = self._decode_sub(token_response.access_token) + token_set = AnonymousTokenSetEntry( + access_token=token_response.access_token, + expires_at=now + token_response.expires_in, + audience=audience, + scope=scope, + granted_scope=token_response.scope, + ) + context = AnonymousSessionContext( + session_token=token_response.session_token, + token_sets=[token_set], + session_expires_at=now + token_response.session_expires_in if token_response.session_expires_in is not None else None, + metadata=metadata, + created_at=now, + domain=domain, + sub=sub, + ) + await self._anonymous_store.set( + ANON_IDENTIFIER, + {"context": self._encrypt_context(context)}, + options=store_options, + ) + return AnonymousSession( + access_token=token_set.access_token, + expires_at=token_set.expires_at, + session_expires_at=context.session_expires_at, + metadata=context.metadata, + sub=context.sub, + scope=token_set.granted_scope, + ) + + # ============================================================================ + # TOKEN RENEWAL LADDER + # ============================================================================ + + async def _remint( + self, + context: AnonymousSessionContext, + audience: Optional[str], + scope: Optional[str], + store_options: Optional[dict[str, Any]], + ) -> AnonymousSession: + """Re-mint an access token using the stored session token. + + Args: + context: The current decrypted session context. + audience: Audience to request for the new token. + scope: Scope to request for the new token. + store_options: Options passed to the anonymous store. + + Returns: + The refreshed AnonymousSession. + + Raises: + AnonymousSessionTokenError: The request failed, or the response was + invalid. + """ + domain = context.domain or await self._resolve_domain(store_options) + body: dict[str, Any] = { + "client_id": self._client_id, + "session_token": context.session_token, + } + self._apply_client_auth(body, domain) + if audience: + body["audience"] = audience + if scope: + body["scope"] = scope + + async with self._get_http_client() as client: + try: + response = await client.post(f"https://{domain}/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(error_data, "token") + if isinstance(mapped, _AnonymousSessionExpired): + # One follow-up create call on expiry, never a loop. + return await self._create_session_at( + domain, + audience=audience, + scope=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_session_token = ( + token_response.session_token + if token_response.session_token is not None + else context.session_token + ) + new_sub = self._decode_sub(token_response.access_token) + token_set = AnonymousTokenSetEntry( + access_token=token_response.access_token, + expires_at=now + token_response.expires_in, + audience=audience, + scope=scope, + granted_scope=token_response.scope, + ) + result = AnonymousSession( + access_token=token_set.access_token, + expires_at=token_set.expires_at, + session_expires_at=now + token_response.session_expires_in if token_response.session_expires_in is not None else context.session_expires_at, + metadata=context.metadata, + sub=new_sub if new_sub is not None else context.sub, + scope=token_response.scope, + ) + + # Re-read to preserve a concurrent remint and skip a stale write. + current_stored = await self._anonymous_store.get(ANON_IDENTIFIER, options=store_options) + if not current_stored: + return result + try: + current_context = self._decrypt_context(current_stored) + except (_AnonymousSessionExpired, _SessionDecryptError): + return result + if current_context.session_token != context.session_token: + return result + + # Only update sub when it was not previously stored (token was initially JWE). + sub_update = {"sub": new_sub} if new_sub is not None and current_context.sub is None else {} + updated_context = self._upsert_token_set( + current_context.model_copy(update={ + "session_token": new_session_token, + "session_expires_at": now + token_response.session_expires_in if token_response.session_expires_in is not None else context.session_expires_at, + **sub_update, + }), + token_set, + ) + await self._anonymous_store.set( + ANON_IDENTIFIER, + {"context": self._encrypt_context(updated_context)}, + options=store_options, + ) + return result + + # ============================================================================ + # LOGIN INJECTION SUPPORT + # ============================================================================ + + async def exchange_transfer_token_for_injection( + self, origin_domain: str, store_options: Optional[dict[str, Any]] = None + ) -> Optional[str]: + """Mint a short-lived transfer ticket from the active session for login injection. + + Args: + origin_domain: The domain the /authorize URL is being built for. + store_options: Options passed to the anonymous store. + + Returns: + The minted transfer ticket, or None. + """ + 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, _SessionDecryptError): + return None + # Prevents a tenant-A session token from minting a transfer ticket usable at tenant-B's login. + # In resolver mode, an unknown stored domain (legacy session) is treated as a mismatch. + domain_unknown = not context.domain and self._domain_resolver is not None + domain_mismatch = context.domain and self._normalize_url(context.domain) != self._normalize_url( + origin_domain + ) + if domain_unknown or domain_mismatch: + return None + return await self._mint_transfer_token(context.session_token, origin_domain) + + async def _mint_transfer_token( + self, session_token: str, origin_domain: str + ) -> Optional[str]: + """Exchange a session token for a transfer ticket. + + Args: + session_token: The stored anonymous session token. + origin_domain: The domain the /authorize URL is being built for. + + Returns: + The minted transfer ticket, or None. + """ + base_url = f"https://{origin_domain}" + body: dict[str, Any] = { + "client_id": self._client_id, + "session_token": session_token, + "audience": TRANSFER_AUDIENCE, + } + self._apply_client_auth(body, origin_domain) + + try: + async with self._get_http_client() as client: + response = await client.post(f"{base_url}/anonymous/token", json=body) + except httpx.HTTPError: + return None + if response.status_code != 200: + return None + try: + token_response = AnonymousTransferTokenResponse.model_validate(response.json()) + except (json.JSONDecodeError, ValueError, ValidationError): + return None + return token_response.anon_transfer_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.model_validate(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, + *, + audience: Optional[str] = None, + scope: Optional[str] = None, + ) -> AnonymousSession: + """Return a valid anonymous access token, renewing or re-minting as needed. + + Args: + store_options: Options passed to the anonymous store. + audience: Audience to retrieve a token for. Falls back to the + client's configured default when omitted. + scope: Scope to retrieve a token for. Falls back to the client's + configured default when omitted. + + 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.") + + eff_audience = audience or self._default_audience + eff_scope = scope or self._default_scope + + try: + context = self._decrypt_context(stored) + except _SessionDecryptError as e: + await self._anonymous_store.delete(ANON_IDENTIFIER, options=store_options) + raise AnonymousSessionTokenError( + "The stored anonymous session could not be decrypted. " + "Call create_session() to start a new session.", + code="invalid_session_state", + ) from e + + current_domain = await self._resolve_domain(store_options) + # In resolver mode, an unknown stored domain (legacy session) is treated as a mismatch. + domain_unknown = not context.domain and self._domain_resolver is not None + domain_mismatch = context.domain and self._normalize_url(context.domain) != self._normalize_url( + current_domain + ) + if domain_unknown or domain_mismatch: + return await self._create_session_at( + current_domain, + audience=eff_audience, + scope=eff_scope, + metadata=None, + store_options=store_options, + ) + + now = int(time.time()) + token_set = self._find_token_set(context.token_sets, eff_audience, eff_scope) + if token_set and token_set.expires_at - State.SESSION_EXPIRY_LEEWAY_SECONDS > now: + return AnonymousSession( + access_token=token_set.access_token, + expires_at=token_set.expires_at, + session_expires_at=context.session_expires_at, + metadata=context.metadata, + sub=context.sub, + scope=token_set.granted_scope, + ) + + return await self._remint(context, eff_audience, eff_scope, store_options) + + async def logout(self, store_options: Optional[dict[str, Any]] = None) -> None: + """Clear the locally-held anonymous session without revoking issued tokens. + + Args: + store_options: Options passed to the anonymous store. + + Raises: + ConfigurationError: No anonymous_store configured. + """ + self._require_store() + stored = await self._anonymous_store.get(ANON_IDENTIFIER, options=store_options) + if not stored: + return + await self._anonymous_store.delete(ANON_IDENTIFIER, options=store_options) + + async def get_session( + self, store_options: Optional[dict[str, Any]] = None + ) -> Optional[AnonymousSessionData]: + """Return stored anonymous session identity without calling Auth0. + + Args: + store_options: Options passed to the anonymous store. + + Returns: + The stored AnonymousSessionData, or None when there is no session, + the session is corrupt, or (in resolver mode) the stored domain + does not match the current tenant. + """ + 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, _SessionDecryptError): + return None + if context.domain and self._domain_resolver is not None: + try: + current_domain = await self._resolve_domain(store_options) + except Exception: + return None + if self._normalize_url(context.domain) != self._normalize_url(current_domain): + return None + return AnonymousSessionData( + sub=context.sub, + metadata=context.metadata, + created_at=context.created_at, + session_expires_at=context.session_expires_at, + domain=context.domain, + ) + + async def _end_session_if_active( + self, store_options: Optional[dict[str, Any]] = None + ) -> None: + """Clear the local anonymous session on authenticated logout, if one is active. + + Args: + store_options: Options passed to the anonymous store. + """ + if self._anonymous_store is None: + return + try: + stored = await self._anonymous_store.get(ANON_IDENTIFIER, options=store_options) + except Exception: + return + if not stored: + return + try: + await self._anonymous_store.delete(ANON_IDENTIFIER, options=store_options) + except Exception: + return diff --git a/src/auth0_server_python/auth_server/server_client.py b/src/auth0_server_python/auth_server/server_client.py index d8d3aa2..65f61e1 100644 --- a/src/auth0_server_python/auth_server/server_client.py +++ b/src/auth0_server_python/auth_server/server_client.py @@ -28,6 +28,7 @@ validate_client_assertion_key, ) 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 DEFAULT_MFA_TOKEN_TTL, MfaClient from auth0_server_python.auth_server.my_account_client import MyAccountClient from auth0_server_python.auth_server.passwordless_client import PasswordlessClient @@ -99,7 +100,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", "anon_transfer_token"] # issued_token_type URN for a Session Transfer Token (STT). SESSION_TRANSFER_TOKEN_TYPE = "urn:auth0:params:oauth:token-type:session_transfer_token" @@ -155,6 +157,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, @@ -164,6 +167,7 @@ def __init__( enterprise_connect: bool = False, use_mtls: bool = False, ssl_context: Optional[ssl.SSLContext] = None, + clear_anonymous_session_on_login: bool = True, ): """ Initialize the Auth0 server client. @@ -178,6 +182,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 @@ -198,6 +209,12 @@ def __init__( ssl_context: TLS context carrying the client certificate and key. Required when use_mtls=True. Build with ssl.create_default_context() and load_cert_chain(). + clear_anonymous_session_on_login: When True and an anonymous_store is + configured, the anonymous session is cleared via anonymous.logout() + after a successful interactive login (complete_interactive_login). + Defaults to True. Set to False to keep the anonymous session active + across the login boundary. A failure clearing the anonymous session + never fails the login. Raises: ConfigurationError: If `mfa_token_ttl` is not a positive number of seconds. @@ -247,6 +264,11 @@ def __init__( "use_mtls cannot be combined with client_assertion_signing_key. " "The client certificate is the sole credential under mTLS." ) + if anonymous_store is not None: + raise ConfigurationError( + "Anonymous Sessions do not support mTLS. The AnonymousClient has no " + "client-certificate credential path." + ) self._client_id = client_id self._client_secret = client_secret @@ -262,11 +284,13 @@ def __init__( self._pushed_authorization_requests = pushed_authorization_requests # store the flag self._organization = organization self._enterprise_connect = enterprise_connect + self._clear_anonymous_session_on_login = clear_anonymous_session_on_login self._webfinger_cache: OrderedDict[str, dict] = OrderedDict() # 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 @@ -310,6 +334,21 @@ def __init__( token_endpoint_resolver=self._resolve_mfa_token_endpoint if self._use_mtls else None, ) + # 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, + client_assertion_signing_key=self._client_assertion_signing_key, + client_assertion_signing_alg=self._client_assertion_signing_alg, + ) self._passwordless_client = PasswordlessClient(self) if enterprise_connect: @@ -813,6 +852,16 @@ async def start_interactive_login( if options.invitation: auth_params["invitation"] = options.invitation + # Pops close the session-fixation vector from constructor-seeded defaults. + auth_params.pop("session_token", None) + auth_params.pop("anon_transfer_token", None) + if not self._pushed_authorization_requests and not self._enterprise_connect: + anon_transfer_token = await self._anonymous_client.exchange_transfer_token_for_injection( + origin_domain, store_options + ) + if anon_transfer_token: + auth_params["anon_transfer_token"] = anon_transfer_token + # Build the transaction data to store with domain transaction_data = TransactionData( code_verifier=code_verifier, @@ -1076,6 +1125,7 @@ async def complete_interactive_login( # Clean up transaction data after successful login await self._transaction_store.delete(transaction_identifier, options=store_options) + await self._clear_anonymous_session_after_login(store_options) result = {"state_data": state_data.dict()} if transaction_data.app_state: @@ -1283,6 +1333,23 @@ async def _establish_session_from_mfa_verify_response( store_options=store_options, ) + async def _clear_anonymous_session_after_login( + self, store_options: Optional[dict[str, Any]] = None + ) -> None: + """Clear the anonymous session after a successful interactive login. + + Args: + store_options: Options passed to the anonymous store. + """ + if not self._clear_anonymous_session_on_login: + return + if self._anonymous_store is None: + return + try: + await self._anonymous_client.logout(store_options) + except Exception as e: + logger.debug("Anonymous session cleanup after login failed: %s", e) + # ============================================================================ # USER SESSION MANAGEMENT # Methods for retrieving user information, session data, and logout operations. @@ -1412,6 +1479,13 @@ async def logout( if session_domain and self._normalize_url(session_domain) == self._normalize_url(domain): await self._state_store.delete(self._state_identifier, store_options) + # Closes the shared-device re-injection path. + if self._anonymous_store is not None: + try: + await self._anonymous_client._end_session_if_active(store_options) + except Exception as e: + logger.debug("Anonymous session cleanup on logout failed: %s", e) + # Return logout URL for the current resolved domain logout_url = URL.create_logout_url( domain, self._client_id, options.return_to, federated=bool(options.federated)) @@ -1853,6 +1927,7 @@ async def login_backchannel( state_data["domain"] = domain await self._state_store.set(self._state_identifier, state_data, store_options) + await self._clear_anonymous_session_after_login(store_options) result = { "authorization_details": token_endpoint_response.get("authorization_details") @@ -3180,6 +3255,7 @@ async def login_with_custom_token_exchange( # Store session await self._state_store.set(self._state_identifier, state_data, options=store_options) + await self._clear_anonymous_session_after_login(store_options) # Build result result = LoginWithCustomTokenExchangeResult( @@ -3798,6 +3874,7 @@ async def signin_with_passkey( ) await self._state_store.set(self._state_identifier, state_data, options=store_options) + await self._clear_anonymous_session_after_login(store_options) return PasskeyLoginResult(state_data=state_data.model_dump()) @@ -3933,6 +4010,15 @@ async def start_enterprise_login( ) return await self.start_interactive_login(login_options, store_options) + # ============================================================================ + # ANONYMOUS SESSIONS + # ============================================================================ + + @property + def anonymous(self) -> AnonymousClient: + """Access the anonymous sessions client for pre-login anon@ identity operations.""" + return self._anonymous_client + async def is_federated_domain(domain: str, email_domain: str, timeout: float = 5.0) -> bool: """ diff --git a/src/auth0_server_python/auth_types/__init__.py b/src/auth0_server_python/auth_types/__init__.py index d419915..a5916ec 100644 --- a/src/auth0_server_python/auth_types/__init__.py +++ b/src/auth0_server_python/auth_types/__init__.py @@ -236,6 +236,16 @@ class LogoutOptions(BaseModel): federated: Optional[bool] = False +class CreateAnonymousSessionOptions(BaseModel): + """Options bundle for create_session(), carrying 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. @@ -870,6 +880,83 @@ class PasskeyTokenResponse(BaseModel): refresh_token: Optional[str] = None +# ============================================================================= +# Anonymous Session Types +# ============================================================================= + + +class AnonymousSession(BaseModel): + """Public result of create_session() and the renewal ladder.""" + + access_token: str + expires_at: int + session_expires_at: Optional[int] = None + metadata: Optional[dict[str, Any]] = None + sub: Optional[str] = None + scope: Optional[str] = None + + +class AnonymousSessionData(BaseModel): + """Identity and metadata fields returned by AnonymousClient.get_session(), with no session_token.""" + + sub: Optional[str] = None + metadata: Optional[dict[str, Any]] = None + created_at: int + session_expires_at: Optional[int] = None + domain: Optional[str] = None + + +class AnonymousTokenResponse(BaseModel): + """Raw response from POST /anonymous/token.""" + + access_token: str + token_type: str = "Bearer" + expires_in: int + session_expires_in: Optional[int] = None + session_token: Optional[str] = None + scope: Optional[str] = None + + +class AnonymousCreateTokenResponse(AnonymousTokenResponse): + """Raw response from POST /anonymous/token on the create path.""" + + session_token: str + + +class AnonymousTransferTokenResponse(BaseModel): + """Raw response from POST /anonymous/token on the transfer-ticket path.""" + + model_config = ConfigDict(extra="ignore") + anon_transfer_token: str + token_type: Optional[str] = None + expires_in: Optional[int] = None + + +class AnonymousTokenSetEntry(BaseModel): + """One cached access token for a specific audience/scope pair.""" + + model_config = ConfigDict(extra="forbid") + + access_token: str + expires_at: int + audience: Optional[str] = None + scope: Optional[str] = None + granted_scope: Optional[str] = None + + +class AnonymousSessionContext(BaseModel): + """Internal context stored inside the encrypted anonymous session record, rejecting extra fields so a tampered payload raises on decrypt.""" + + model_config = ConfigDict(extra="forbid") + + session_token: str + token_sets: list[AnonymousTokenSetEntry] = Field(default_factory=list) + session_expires_at: Optional[int] = None + metadata: Optional[dict[str, Any]] = None + created_at: int + domain: Optional[str] = None + sub: Optional[str] = None + # ============================================================================= # Passwordless Types diff --git a/src/auth0_server_python/error/__init__.py b/src/auth0_server_python/error/__init__.py index 7443590..84c3870 100644 --- a/src/auth0_server_python/error/__init__.py +++ b/src/auth0_server_python/error/__init__.py @@ -426,6 +426,57 @@ class PasskeyErrorCode: INVALID_RESPONSE = "invalid_response" +# ============================================================================= +# Anonymous Session Error Classes +# ============================================================================= + +class AnonymousSessionError(Auth0Error): + """Base class for anonymous session 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(AnonymousSessionError): + """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 AnonymousSessionTokenError(AnonymousSessionError): + """Error thrown when get_token() fails for reasons other than session expiry.""" + + def __init__(self, message: str, code: str = "anonymous_token_error", cause: Optional[dict] = None): + super().__init__(code, message, cause) + + +class _AnonymousSessionExpired(AnonymousSessionError): + """Internal-only signal that the stored session token is expired or invalid.""" + + def __init__(self, message: str = "The anonymous session token is expired or invalid."): + super().__init__("session_expired", message) + self.name = "_AnonymousSessionExpired" + + +class _SessionDecryptError(Auth0Error): + """Internal-only signal that the stored session payload could not be decrypted or parsed. + + Distinct from _AnonymousSessionExpired (platform expiry) - this indicates local + JWE decryption failure, e.g. secret rotation or store corruption. + """ + + def __init__(self, message: str = "The stored anonymous session could not be decrypted."): + super().__init__(message) + self.name = "_SessionDecryptError" + # ============================================================================= # Enterprise Connect Error Classes # ============================================================================= diff --git a/src/auth0_server_python/tests/store_fakes.py b/src/auth0_server_python/tests/store_fakes.py new file mode 100644 index 0000000..c10e5ce --- /dev/null +++ b/src/auth0_server_python/tests/store_fakes.py @@ -0,0 +1,19 @@ +"""Shared test doubles for the stateless store contract.""" + + +class OneSlotStore: + """A StatelessStateStore fake with one physical slot per instance, where the identifier is an encryption salt rather than a location 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 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..b788bcd --- /dev/null +++ b/src/auth0_server_python/tests/test_anonymous_client.py @@ -0,0 +1,1586 @@ +""" +Tests for AnonymousClient, covering anonymous session API operations. +""" + +import base64 as _b64 +import inspect +import time +from unittest.mock import AsyncMock, MagicMock, patch +from urllib.parse import urlsplit + +import httpx +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa + +from auth0_server_python.auth_server.anonymous_client import ( + ANON_IDENTIFIER, + AnonymousClient, +) +from auth0_server_python.auth_types import ( + AnonymousSession, + AnonymousSessionContext, + AnonymousSessionData, + AnonymousTokenSetEntry, + CreateAnonymousSessionOptions, +) +from auth0_server_python.encryption.encrypt import encrypt +from auth0_server_python.error import ( + AnonymousSessionCreateError, + AnonymousSessionTokenError, + ConfigurationError, + DomainResolverError, +) +from auth0_server_python.tests.store_fakes import OneSlotStore + +# Shared fixtures +DOMAIN = "auth0.local" +CLIENT_ID = "" +CLIENT_SECRET = "" +SECRET = "test-secret-long-enough-for-encryption" + + +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 _generate_rsa_private_key_pem() -> str: + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + return key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ).decode("ascii") + + +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. The call sequence maps one-to-one 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, +): + return { + "access_token": access_token, + "token_type": "Bearer", + "expires_in": expires_in, + "session_token": session_token, + "session_expires_in": session_expires_in, + } + + +def _make_jwt(sub: str = "anon@test-uuid") -> str: + """Build a minimal unsigned JWT with the given sub claim.""" + header = _b64.urlsafe_b64encode(b'{"alg":"none"}').rstrip(b"=").decode() + payload = _b64.urlsafe_b64encode(f'{{"sub":"{sub}"}}'.encode()).rstrip(b"=").decode() + return f"{header}.{payload}." + + +def _stored_context(store: OneSlotStore, **overrides): + ts_keys = {"access_token", "expires_at", "audience", "scope", "granted_scope"} + ts_defaults = { + "access_token": "AT1", + "expires_at": int(time.time()) + 3600, + } + ctx_defaults = { + "session_token": "ST1", + "created_at": int(time.time()), + } + for k in list(overrides): + if k in ts_keys: + ts_defaults[k] = overrides.pop(k) + else: + ctx_defaults[k] = overrides.pop(k) + token_set = AnonymousTokenSetEntry(**ts_defaults) + context = AnonymousSessionContext(token_sets=[token_set], **ctx_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): + """AnonymousClient has no dpop_key parameter anywhere in its public API.""" + 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" + + def test_constructor_accepts_private_key_jwt_params(self): + signing_key = _generate_rsa_private_key_pem() + client = AnonymousClient( + domain=DOMAIN, + client_id=CLIENT_ID, + client_secret=None, + secret=SECRET, + client_assertion_signing_key=signing_key, + client_assertion_signing_alg="RS256", + ) + assert client._client_assertion_signing_key == signing_key + assert client._client_assertion_signing_alg == "RS256" + assert client._client_secret is None + + def test_constructor_rejects_invalid_signing_key(self): + with pytest.raises(ConfigurationError): + AnonymousClient( + domain=DOMAIN, + client_id=CLIENT_ID, + client_secret=None, + secret=SECRET, + client_assertion_signing_key="not-a-valid-pem-key", + ) + + @pytest.mark.asyncio + async def test_public_client_sends_no_credentials(self): + """A client with no secret and no signing key sends requests without any auth credential.""" + store = OneSlotStore() + client = AnonymousClient( + domain=DOMAIN, + client_id=CLIENT_ID, + client_secret=None, + secret=SECRET, + 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] + body = kwargs["json"] + assert "client_secret" not in body + assert "client_assertion" not in body + assert "client_assertion_type" not in body + + +# ── 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_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): + """No anonymous_store configured raises before any write and 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() + + +# ── 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.metadata == {"cart_id": "c1"} + + def test_anonymous_session_does_not_expose_session_token(self): + """session_token is a server credential and must not appear on the public return type.""" + assert "session_token" not in AnonymousSession.model_fields + + @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) as excinfo: + await client.create_session(audience="aud", scope="s") + assert excinfo.value.code == "missing_session_token" + assert store.slot is None + + @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 + @pytest.mark.asyncio + async def test_create_session_uses_client_assertion_when_signing_key_set(self): + store = OneSlotStore() + signing_key = _generate_rsa_private_key_pem() + client = AnonymousClient( + domain=DOMAIN, + client_id=CLIENT_ID, + client_secret=None, + secret=SECRET, + anonymous_store=store, + client_assertion_signing_key=signing_key, + ) + 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] + body = kwargs["json"] + assert "client_assertion" in body + assert body.get("client_assertion_type") == "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" + assert "client_secret" not in body + + @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 + 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_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(AnonymousSessionCreateError) as exc: + await client.create_session(audience="aud", scope="s") + assert exc.value.code == "feature_not_enabled" + + @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(AnonymousSessionCreateError) as exc: + await client.create_session(audience="aud", scope="s") + assert exc.value.code == "unauthorized_client" + + @pytest.mark.asyncio + async def test_invalid_target_maps_to_typed_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(AnonymousSessionCreateError) as exc: + await client.create_session(audience="aud", scope="s") + assert exc.value.code == "invalid_target" + + @pytest.mark.asyncio + async def test_invalid_scope_maps_to_typed_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(AnonymousSessionCreateError) as exc: + await client.create_session(audience="aud", scope="s") + assert exc.value.code == "invalid_scope" + + @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") + + + @pytest.mark.asyncio + async def test_create_session_stores_and_returns_sub_from_jwt(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + jwt_token = _make_jwt("anon@test-uuid") + fake_http = _FakeAsyncClient([_fake_response(200, { + "access_token": jwt_token, + "token_type": "Bearer", + "expires_in": 3600, + "session_token": "ST1", + "session_expires_in": 2592000, + })]) + with patch("httpx.AsyncClient", fake_http): + session = await client.create_session() + assert session.sub == "anon@test-uuid" + stored = await store.get(ANON_IDENTIFIER) + ctx = client._decrypt_context(stored) + assert ctx.sub == "anon@test-uuid" + + @pytest.mark.asyncio + async def test_create_session_stores_none_sub_for_opaque_token(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(200, { + "access_token": "opaque.token.without.claims.here", + "token_type": "Bearer", + "expires_in": 3600, + "session_token": "ST1", + "session_expires_in": 2592000, + })]) + with patch("httpx.AsyncClient", fake_http): + session = await client.create_session() + assert session.sub is None + stored = await store.get(ANON_IDENTIFIER) + ctx = client._decrypt_context(stored) + assert ctx.sub is None + + @pytest.mark.asyncio + async def test_create_session_tolerates_missing_session_expires_in(self): + """Legacy platform tokens omit session_expires_in so session_expires_at must be None.""" + store = OneSlotStore() + client = _make_client(anonymous_store=store) + response = {k: v for k, v in _token_response().items() if k != "session_expires_in"} + fake_http = _FakeAsyncClient([_fake_response(200, response)]) + with patch("httpx.AsyncClient", fake_http): + session = await client.create_session() + assert session.session_expires_at is None + + @pytest.mark.asyncio + async def test_create_session_surfaces_granted_scope(self): + """Platform-granted scope is returned in AnonymousSession.scope.""" + store = OneSlotStore() + client = _make_client(anonymous_store=store) + response = {**_token_response(), "scope": "read:cart"} + fake_http = _FakeAsyncClient([_fake_response(200, response)]) + with patch("httpx.AsyncClient", fake_http): + session = await client.create_session(scope="read:cart write:cart") + assert session.scope == "read:cart" + + @pytest.mark.asyncio + async def test_non_string_sub_in_jwt_stored_as_none(self): + """A JWT with a non-string sub must not be coerced to a string.""" + store = OneSlotStore() + client = _make_client(anonymous_store=store) + header = _b64.urlsafe_b64encode(b'{"alg":"none"}').rstrip(b"=").decode() + payload = _b64.urlsafe_b64encode(b'{"sub":12345}').rstrip(b"=").decode() + bad_jwt = f"{header}.{payload}." + fake_http = _FakeAsyncClient([_fake_response(200, { + "access_token": bad_jwt, + "token_type": "Bearer", + "expires_in": 3600, + "session_token": "ST1", + "session_expires_in": 2592000, + })]) + with patch("httpx.AsyncClient", fake_http): + session = await client.create_session() + assert session.sub is None + + @pytest.mark.asyncio + async def test_sub_without_anon_prefix_stored_as_none(self): + """A JWT sub that does not start with anon@ must be rejected.""" + store = OneSlotStore() + client = _make_client(anonymous_store=store) + jwt_token = _make_jwt(sub="user@unexpected") + fake_http = _FakeAsyncClient([_fake_response(200, { + "access_token": jwt_token, + "token_type": "Bearer", + "expires_in": 3600, + "session_token": "ST1", + "session_expires_in": 2592000, + })]) + with patch("httpx.AsyncClient", fake_http): + session = await client.create_session() + assert session.sub is None + + +# ── 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.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, + "session_expires_in": 2592000, + }, + ) + ]) + with patch("httpx.AsyncClient", fake_http): + session = await client.get_token() + assert session.access_token == "AT2" + _, _, 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_replays_audience_and_scope_from_get_token_params(self): + """Re-mint request must include the audience/scope passed to get_token.""" + 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_expires_in": 2592000, + }, + ) + ]) + with patch("httpx.AsyncClient", fake_http): + await client.get_token(audience="https://api.example.com", scope="read:things") + _, _, kwargs = fake_http.calls[0] + assert kwargs["json"]["audience"] == "https://api.example.com" + assert kwargs["json"]["scope"] == "read:things" + + @pytest.mark.asyncio + async def test_remint_omits_audience_and_scope_when_the_session_had_none(self): + """Absent values must stay absent - never sent as null or empty string.""" + 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_expires_in": 2592000, + }, + ) + ]) + with patch("httpx.AsyncClient", fake_http): + await client.get_token() + _, _, kwargs = fake_http.calls[0] + assert "audience" not in kwargs["json"] + assert "scope" not in kwargs["json"] + + @pytest.mark.asyncio + async def test_remint_uses_client_assertion_when_signing_key_set(self): + """Renewal request must carry client_assertion, not client_secret, for private_key_jwt clients.""" + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) - 10) + signing_key = _generate_rsa_private_key_pem() + client = AnonymousClient( + domain=DOMAIN, + client_id=CLIENT_ID, + client_secret=None, + secret=SECRET, + anonymous_store=store, + client_assertion_signing_key=signing_key, + ) + fake_http = _FakeAsyncClient([_fake_response(200, _token_response(access_token="AT2"))]) + with patch("httpx.AsyncClient", fake_http): + await client.get_token() + _, _, kwargs = fake_http.calls[0] + body = kwargs["json"] + assert "client_assertion" in body + assert body.get("client_assertion_type") == "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" + assert "client_secret" not in body + + @pytest.mark.asyncio + async def test_remint_never_sends_metadata_in_body(self): + """Platform 400s if metadata is included in a renewal request.""" + 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( + 200, + { + "access_token": "AT2", + "token_type": "Bearer", + "expires_in": 3600, + "session_expires_in": 2592000, + }, + ) + ]) + with patch("httpx.AsyncClient", fake_http): + await client.get_token() + _, _, kwargs = fake_http.calls[0] + assert "metadata" not in kwargs["json"] + + @pytest.mark.asyncio + async def test_remint_preserves_session_expires_at_when_platform_omits_session_expires_in(self): + """Legacy renewal responses omit session_expires_in so stored expiry must be kept.""" + stored_expiry = int(time.time()) + 86400 + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) - 10, session_expires_at=stored_expiry) + client = _make_client(anonymous_store=store) + response = {k: v for k, v in _token_response(access_token="AT2").items() if k != "session_expires_in"} + fake_http = _FakeAsyncClient([_fake_response(200, response)]) + with patch("httpx.AsyncClient", fake_http): + session = await client.get_token() + assert session.session_expires_at == stored_expiry + + @pytest.mark.asyncio + async def test_remint_surfaces_granted_scope(self): + """Platform-granted scope from a remint is returned in AnonymousSession.scope.""" + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) - 10) + client = _make_client(anonymous_store=store) + response = {**_token_response(access_token="AT2"), "scope": "read:cart"} + fake_http = _FakeAsyncClient([_fake_response(200, response)]) + with patch("httpx.AsyncClient", fake_http): + session = await client.get_token(scope="read:cart write:cart") + assert session.scope == "read:cart" + + @pytest.mark.asyncio + async def test_cached_token_surfaces_stored_granted_scope(self): + """A cached token hit returns the previously stored granted_scope.""" + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) + 3600, granted_scope="read:cart") + 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.scope == "read:cart" + + @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_expires_in": 2592000, + "session_token": "", + }, + ) + ]) + with patch("httpx.AsyncClient", fake_http): + await client.get_token() + stored = await store.get(ANON_IDENTIFIER) + context = client._decrypt_context(stored) + assert context.session_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()), + ]) + with patch("httpx.AsyncClient", fake_http): + session = await client.get_token() + assert len(fake_http.calls) == 2 + _, _, second_call = fake_http.calls[1] + assert "session_token" not in second_call["json"] + assert session.access_token == "AT1" + + @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): + """The retry-once bound allows exactly 2 upstream POSTs, then raises.""" + 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(AnonymousSessionTokenError) as exc: + await client.get_token() + assert exc.value.code == "feature_not_enabled" + assert len(fake_http.calls) == 1 + + @pytest.mark.asyncio + async def test_corrupted_stored_token_raises_and_clears_store(self): + store = OneSlotStore() + store.slot = (ANON_IDENTIFIER, {"context": "not-a-valid-jwe"}) + client = _make_client(anonymous_store=store) + with pytest.raises(AnonymousSessionTokenError) as exc: + await client.get_token() + assert exc.value.code == "invalid_session_state" + assert store.slot is None + + @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() + + @pytest.mark.asyncio + async def test_second_audience_is_cached_without_evicting_first(self): + """get_token for a different audience upserts rather than replacing.""" + store = OneSlotStore() + _stored_context(store, audience="https://api1.example.com") + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([ + _fake_response(200, { + "access_token": "AT2", + "token_type": "Bearer", + "expires_in": 3600, + "session_expires_in": 2592000, + }) + ]) + with patch("httpx.AsyncClient", fake_http): + await client.get_token(audience="https://api2.example.com") + + stored = await store.get(ANON_IDENTIFIER) + context = client._decrypt_context(stored) + audiences = [ts.audience for ts in context.token_sets] + assert any(a == "https://api1.example.com" for a in audiences) + assert any(a == "https://api2.example.com" for a in audiences) + + @pytest.mark.asyncio + async def test_cached_token_returned_for_correct_audience(self): + """get_token returns the cached token for the matching audience without a network call.""" + store = OneSlotStore() + _stored_context(store, audience="https://api1.example.com", access_token="AT_API1") + client = _make_client(anonymous_store=store) + with patch("httpx.AsyncClient") as mock_http: + session = await client.get_token(audience="https://api1.example.com") + mock_http.assert_not_called() + assert session.access_token == "AT_API1" + + @pytest.mark.asyncio + async def test_different_audience_causes_remint_not_cache_hit(self): + """A stored token for api1 does not satisfy a request for api2.""" + store = OneSlotStore() + _stored_context(store, audience="https://api1.example.com", access_token="AT_API1") + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([ + _fake_response(200, { + "access_token": "AT_API2", + "token_type": "Bearer", + "expires_in": 3600, + "session_expires_in": 2592000, + }) + ]) + with patch("httpx.AsyncClient", fake_http): + session = await client.get_token(audience="https://api2.example.com") + assert session.access_token == "AT_API2" + assert len(fake_http.calls) == 1 + + @pytest.mark.asyncio + async def test_concurrent_remint_preserves_other_audience_token(self): + """A token for api2 written to the store before our re-read is not lost.""" + store = OneSlotStore() + _stored_context(store) + client = _make_client(anonymous_store=store) + + remint_response = { + "access_token": "AT_API1", + "token_type": "Bearer", + "expires_in": 3600, + "session_expires_in": 2592000, + } + + original_get = store.get + original_set = store.set + get_call_count = 0 + + async def get_with_concurrent_write(identifier, *, options=None): + nonlocal get_call_count + get_call_count += 1 + if get_call_count == 2: + # Simulate a concurrent remint for api2 completing during our HTTP call, + # i.e. before our re-read runs. + current = await original_get(identifier) + ctx = client._decrypt_context(current) + concurrent_token_set = AnonymousTokenSetEntry( + access_token="AT_API2", + expires_at=int(time.time()) + 3600, + audience="https://api2.example.com", + ) + merged = client._upsert_token_set(ctx, concurrent_token_set) + await original_set( + identifier, + {"context": encrypt(merged.model_dump(), SECRET, "anon_session")}, + ) + return await original_get(identifier) + + store.get = get_with_concurrent_write + + fake_http = _FakeAsyncClient([_fake_response(200, remint_response)]) + with patch("httpx.AsyncClient", fake_http): + await client.get_token(audience="https://api1.example.com") + + stored = await store.get(ANON_IDENTIFIER) + final_ctx = client._decrypt_context(stored) + audiences = [ts.audience for ts in final_ctx.token_sets] + assert any(a == "https://api2.example.com" for a in audiences) + assert any(a == "https://api1.example.com" for a in audiences) + + @pytest.mark.asyncio + async def test_remint_skips_write_when_session_deleted_during_fetch(self): + """If the session is deleted while the HTTP call is in flight, the write is skipped.""" + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) - 10) + client = _make_client(anonymous_store=store) + + original_get = store.get + call_count = 0 + + async def get_and_delete(identifier, *, options=None): + nonlocal call_count + call_count += 1 + if call_count == 2: + store.slot = None + return await original_get(identifier) + + store.get = get_and_delete + fake_http = _FakeAsyncClient([ + _fake_response(200, { + "access_token": "AT2", + "token_type": "Bearer", + "expires_in": 3600, + "session_expires_in": 2592000, + }) + ]) + with patch("httpx.AsyncClient", fake_http): + session = await client.get_token() + + assert session is not None + assert store.slot is None + + @pytest.mark.asyncio + async def test_remint_skips_write_when_session_replaced_during_fetch(self): + """If the session token changed while in flight (recreation), the write is skipped.""" + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) - 10) + client = _make_client(anonymous_store=store) + + original_get = store.get + call_count = 0 + + async def get_and_replace(identifier, *, options=None): + nonlocal call_count + call_count += 1 + if call_count == 2: + _stored_context(store, session_token="NEW_SESSION_TOKEN") + return await original_get(identifier) + + store.get = get_and_replace + fake_http = _FakeAsyncClient([ + _fake_response(200, { + "access_token": "AT2", + "token_type": "Bearer", + "expires_in": 3600, + "session_expires_in": 2592000, + }) + ]) + with patch("httpx.AsyncClient", fake_http): + session = await client.get_token() + + assert session is not None + stored = await store.get(ANON_IDENTIFIER) + final_ctx = client._decrypt_context(stored) + assert final_ctx.session_token == "NEW_SESSION_TOKEN" + + @pytest.mark.asyncio + async def test_remint_skips_write_when_reread_is_corrupt(self): + """If the re-read context is corrupt, the result is returned without writing.""" + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) - 10) + client = _make_client(anonymous_store=store) + + original_get = store.get + call_count = 0 + + async def get_then_corrupt(identifier, *, options=None): + nonlocal call_count + call_count += 1 + if call_count == 2: + store.slot = (ANON_IDENTIFIER, {"context": "not-valid-jwe"}) + return await original_get(identifier) + + store.get = get_then_corrupt + fake_http = _FakeAsyncClient([_fake_response(200, _token_response(access_token="AT2"))]) + with patch("httpx.AsyncClient", fake_http): + session = await client.get_token() + + assert session.access_token == "AT2" + stored_raw = store.slot + assert stored_raw is not None + assert stored_raw[1]["context"] == "not-valid-jwe" + + @pytest.mark.asyncio + async def test_get_token_returns_sub_from_cache(self): + store = OneSlotStore() + _stored_context(store, sub="anon@cached-uuid") + 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.sub == "anon@cached-uuid" + + @pytest.mark.asyncio + async def test_remint_returns_and_stores_sub_from_new_token(self): + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) - 10) + client = _make_client(anonymous_store=store) + jwt_token = _make_jwt("anon@reminted-uuid") + fake_http = _FakeAsyncClient([_fake_response(200, { + "access_token": jwt_token, + "token_type": "Bearer", + "expires_in": 3600, + "session_expires_in": 2592000, + })]) + with patch("httpx.AsyncClient", fake_http): + session = await client.get_token() + assert session.sub == "anon@reminted-uuid" + stored = await store.get(ANON_IDENTIFIER) + ctx = client._decrypt_context(stored) + assert ctx.sub == "anon@reminted-uuid" + + @pytest.mark.asyncio + async def test_remint_does_not_overwrite_existing_sub(self): + """Once sub is stored, a remint that returns a readable token must not replace it.""" + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) - 10, sub="anon@original-uuid") + client = _make_client(anonymous_store=store) + jwt_token = _make_jwt("anon@should-be-ignored") + fake_http = _FakeAsyncClient([_fake_response(200, { + "access_token": jwt_token, + "token_type": "Bearer", + "expires_in": 3600, + "session_expires_in": 2592000, + })]) + with patch("httpx.AsyncClient", fake_http): + await client.get_token() + stored = await store.get(ANON_IDENTIFIER) + ctx = client._decrypt_context(stored) + assert ctx.sub == "anon@original-uuid" + + @pytest.mark.asyncio + async def test_remint_backfills_sub_when_initial_token_was_jwe(self): + """When sub was None (initial JWE token), a remint returning a JWT must backfill sub.""" + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) - 10, sub=None) + client = _make_client(anonymous_store=store) + jwt_token = _make_jwt("anon@backfilled-uuid") + fake_http = _FakeAsyncClient([_fake_response(200, { + "access_token": jwt_token, + "token_type": "Bearer", + "expires_in": 3600, + "session_expires_in": 2592000, + })]) + with patch("httpx.AsyncClient", fake_http): + session = await client.get_token() + assert session.sub == "anon@backfilled-uuid" + stored = await store.get(ANON_IDENTIFIER) + ctx = client._decrypt_context(stored) + assert ctx.sub == "anon@backfilled-uuid" + + +# ── 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())]) + with patch("httpx.AsyncClient", fake_http): + await client.get_token() + _, url, _ = fake_http.calls[0] + assert urlsplit(url).hostname == "tenant-b.auth0.local" + + @pytest.mark.asyncio + async def test_null_domain_in_resolver_mode_treated_as_mismatch(self): + """A legacy session with no stored domain must not be sent to an unknown resolver tenant.""" + store = OneSlotStore() + _stored_context(store, expires_at=int(time.time()) + 3600) # domain defaults to None + 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())]) + with patch("httpx.AsyncClient", fake_http): + await client.get_token() + _, url, _ = fake_http.calls[0] + assert urlsplit(url).hostname == "tenant-b.auth0.local" + + @pytest.mark.asyncio + async def test_domain_mismatch_in_static_mode_mints_fresh_under_current_tenant(self): + store = OneSlotStore() + _stored_context( + store, expires_at=int(time.time()) + 3600, domain="tenant-a.auth0.local" + ) + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(200, _token_response())]) + with patch("httpx.AsyncClient", fake_http): + await client.get_token() + _, url, _ = fake_http.calls[0] + assert urlsplit(url).hostname != "tenant-a.auth0.local" + + @pytest.mark.asyncio + async def test_domain_mismatch_remint_does_not_forward_metadata(self): + """A domain-mismatch re-mint must not carry metadata to the new domain.""" + store = OneSlotStore() + _stored_context( + store, + expires_at=int(time.time()) + 3600, + domain="tenant-a.auth0.local", + metadata={"cart": ["sku-1"]}, + ) + 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())]) + with patch("httpx.AsyncClient", fake_http): + await client.get_token() + _, url, kwargs = fake_http.calls[0] + assert urlsplit(url).hostname == "tenant-b.auth0.local" + assert "metadata" not in kwargs["json"] + + @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") + + +# ── 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) + 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) + 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) + 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) + await client.logout() + assert store.slot is None + + @pytest.mark.asyncio + async def test_logout_corrupted_context_clears_state(self): + store = OneSlotStore() + store.slot = (ANON_IDENTIFIER, {"context": "not-a-decryptable-blob"}) + client = _make_client(anonymous_store=store) + await client.logout() + assert store.slot is None + + +# ── exchange_transfer_token_for_injection (transfer ticket) ───────────────────── + +_TRANSFER_OK = {"token_type": "N_A", "anon_transfer_token": "TICKET", "expires_in": 30} + + +class TestExchangeTransferTokenForInjection: + @pytest.mark.asyncio + async def test_success_returns_ticket_with_correct_request_body(self): + store = OneSlotStore() + _stored_context(store, session_token="REAL_TOKEN", domain="auth0.local") + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(200, _TRANSFER_OK)]) + with patch("httpx.AsyncClient", fake_http): + ticket = await client.exchange_transfer_token_for_injection("auth0.local") + assert ticket == "TICKET" + method, url, kwargs = fake_http.calls[0] + assert method == "POST" + assert url == "https://auth0.local/anonymous/token" + body = kwargs["json"] + assert body["audience"] == "urn:auth0:anon_transfer" + assert body["session_token"] == "REAL_TOKEN" + assert body["client_id"] == CLIENT_ID + assert body["client_secret"] == CLIENT_SECRET + + @pytest.mark.asyncio + async def test_transfer_token_uses_client_assertion_when_signing_key_set(self): + """Transfer-ticket request must carry client_assertion, not client_secret, for private_key_jwt clients.""" + store = OneSlotStore() + _stored_context(store, session_token="REAL_TOKEN", domain="auth0.local") + signing_key = _generate_rsa_private_key_pem() + client = AnonymousClient( + domain=DOMAIN, + client_id=CLIENT_ID, + client_secret=None, + secret=SECRET, + anonymous_store=store, + client_assertion_signing_key=signing_key, + ) + fake_http = _FakeAsyncClient([_fake_response(200, _TRANSFER_OK)]) + with patch("httpx.AsyncClient", fake_http): + ticket = await client.exchange_transfer_token_for_injection("auth0.local") + assert ticket == "TICKET" + _, _, kwargs = fake_http.calls[0] + body = kwargs["json"] + assert "client_assertion" in body + assert body.get("client_assertion_type") == "urn:ietf:params:oauth:client-assertion-type:jwt-bearer" + assert "client_secret" not in body + + @pytest.mark.asyncio + async def test_missing_context_domain_mints_against_origin(self): + """A stored context with no domain is not an MCD mismatch on a static-domain client.""" + store = OneSlotStore() + _stored_context(store, session_token="REAL_TOKEN") # domain defaults to None + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(200, _TRANSFER_OK)]) + with patch("httpx.AsyncClient", fake_http): + ticket = await client.exchange_transfer_token_for_injection("auth0.local") + assert ticket == "TICKET" + _, url, _ = fake_http.calls[0] + assert urlsplit(url).hostname == "auth0.local" + + @pytest.mark.asyncio + async def test_null_domain_in_resolver_mode_returns_none(self): + """A legacy session with no stored domain must not mint a transfer ticket in resolver mode.""" + store = OneSlotStore() + _stored_context(store, session_token="REAL_TOKEN") # domain defaults to None + resolver = AsyncMock(return_value="tenant-b.auth0.local") + client = AnonymousClient( + domain=resolver, client_id=CLIENT_ID, client_secret=CLIENT_SECRET, + secret=SECRET, anonymous_store=store, + ) + with patch("httpx.AsyncClient") as mock_http: + result = await client.exchange_transfer_token_for_injection("tenant-b.auth0.local") + assert result is None + mock_http.assert_not_called() + + @pytest.mark.asyncio + async def test_returns_none_without_store(self): + client = _make_client(anonymous_store=None) + assert await client.exchange_transfer_token_for_injection("auth0.local") is None + + @pytest.mark.asyncio + async def test_returns_none_when_no_session_no_http(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + with patch("httpx.AsyncClient") as mock_http: + assert await client.exchange_transfer_token_for_injection("auth0.local") is None + mock_http.assert_not_called() + + @pytest.mark.asyncio + async def test_returns_none_on_corrupted_context(self): + store = OneSlotStore() + store.slot = (ANON_IDENTIFIER, {"context": "garbage"}) + client = _make_client(anonymous_store=store) + assert await client.exchange_transfer_token_for_injection("auth0.local") is None + + @pytest.mark.asyncio + async def test_domain_mismatch_fails_closed_no_mint(self): + """Under MCD, never mint a ticket for a host the session was not created against.""" + store = OneSlotStore() + _stored_context(store, domain="tenant-a.auth0.local") + client = _make_client(anonymous_store=store) + with patch("httpx.AsyncClient") as mock_http: + result = await client.exchange_transfer_token_for_injection("tenant-b.auth0.local") + assert result is None + mock_http.assert_not_called() + + @pytest.mark.asyncio + async def test_returns_none_on_non_200(self): + store = OneSlotStore() + _stored_context(store, domain="auth0.local") + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(400, {"error": "invalid_request"})]) + with patch("httpx.AsyncClient", fake_http): + assert await client.exchange_transfer_token_for_injection("auth0.local") is None + + @pytest.mark.asyncio + async def test_returns_none_on_network_error(self): + store = OneSlotStore() + _stored_context(store, domain="auth0.local") + client = _make_client(anonymous_store=store) + + class _Boom: + 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", _Boom()): + assert await client.exchange_transfer_token_for_injection("auth0.local") is None + + @pytest.mark.asyncio + async def test_returns_none_on_invalid_response_shape(self): + store = OneSlotStore() + _stored_context(store, domain="auth0.local") + client = _make_client(anonymous_store=store) + fake_http = _FakeAsyncClient([_fake_response(200, {"token_type": "N_A", "expires_in": 30})]) + with patch("httpx.AsyncClient", fake_http): + assert await client.exchange_transfer_token_for_injection("auth0.local") is None + + @pytest.mark.asyncio + async def test_never_persists_ticket(self): + store = OneSlotStore() + _stored_context(store, session_token="REAL_TOKEN", domain="auth0.local") + client = _make_client(anonymous_store=store) + before = store.slot + fake_http = _FakeAsyncClient([_fake_response(200, _TRANSFER_OK)]) + with patch("httpx.AsyncClient", fake_http): + await client.exchange_transfer_token_for_injection("auth0.local") + assert store.slot is before # the ticket was never written to the store + + @pytest.mark.asyncio + async def test_returns_none_on_store_exception(self): + store = AsyncMock() + store.get = AsyncMock(side_effect=RuntimeError("store unavailable")) + client = _make_client(anonymous_store=store) + assert await client.exchange_transfer_token_for_injection("auth0.local") is None + + @pytest.mark.asyncio + async def test_forwards_configured_headers(self): + """The exchange goes through _get_http_client, so telemetry/config headers are attached.""" + store = OneSlotStore() + _stored_context(store, domain="auth0.local") + captured = {} + + class _Cap: + def __call__(self, *a, **k): + captured.update(k.get("headers", {})) + return self + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + return False + + async def post(self, *a, **k): + return _fake_response(200, _TRANSFER_OK) + + client = _make_client(anonymous_store=store, headers={"Auth0-Client": "abc"}) + with patch("httpx.AsyncClient", _Cap()): + await client.exchange_transfer_token_for_injection("auth0.local") + assert captured.get("Auth0-Client") == "abc" + + +# ── _end_session_if_active (end anon session on authenticated logout) ──────────── + +class TestEndSessionIfActive: + @pytest.mark.asyncio + async def test_no_store_is_noop(self): + client = _make_client(anonymous_store=None) + with patch("httpx.AsyncClient") as mock_http: + await client._end_session_if_active() + mock_http.assert_not_called() + + @pytest.mark.asyncio + async def test_no_session_makes_no_remote_call(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + with patch("httpx.AsyncClient") as mock_http: + await client._end_session_if_active() + mock_http.assert_not_called() + + @pytest.mark.asyncio + async def test_active_session_clears_local_without_remote_call(self): + store = OneSlotStore() + _stored_context(store, domain="auth0.local") + client = _make_client(anonymous_store=store) + with patch("httpx.AsyncClient") as mock_http: + await client._end_session_if_active() + mock_http.assert_not_called() + assert store.slot is None + + @pytest.mark.asyncio + async def test_store_exception_is_swallowed(self): + store = AsyncMock() + store.get = AsyncMock(side_effect=RuntimeError("store unavailable")) + client = _make_client(anonymous_store=store) + await client._end_session_if_active() # must not raise + + +# ── get_session ─────────────────────────────────────────────────────────────── + +class TestGetSession: + @pytest.mark.asyncio + async def test_returns_none_when_no_store_configured(self): + client = _make_client() + result = await client.get_session() + assert result is None + + @pytest.mark.asyncio + async def test_returns_none_when_no_session_stored(self): + store = OneSlotStore() + client = _make_client(anonymous_store=store) + result = await client.get_session() + assert result is None + + @pytest.mark.asyncio + async def test_returns_session_data_with_identity_fields(self): + store = OneSlotStore() + _stored_context(store, sub="anon@test-uuid", domain="auth0.local") + client = _make_client(anonymous_store=store) + result = await client.get_session() + assert isinstance(result, AnonymousSessionData) + assert result.sub == "anon@test-uuid" + assert result.domain == "auth0.local" + + def test_does_not_include_session_token(self): + """session_token must not be declared on AnonymousSessionData's schema.""" + assert "session_token" not in AnonymousSessionData.model_fields + + @pytest.mark.asyncio + async def test_makes_no_http_call(self): + store = OneSlotStore() + _stored_context(store) + client = _make_client(anonymous_store=store) + with patch("httpx.AsyncClient") as mock_http: + await client.get_session() + mock_http.assert_not_called() + + @pytest.mark.asyncio + async def test_returns_none_for_corrupt_session(self): + store = OneSlotStore() + store.slot = (ANON_IDENTIFIER, {"context": "not-valid-jwe"}) + client = _make_client(anonymous_store=store) + result = await client.get_session() + assert result is None + + @pytest.mark.asyncio + async def test_returns_none_when_store_raises(self): + store = AsyncMock() + store.get = AsyncMock(side_effect=RuntimeError("store down")) + client = _make_client(anonymous_store=store) + result = await client.get_session() + assert result is None + + @pytest.mark.asyncio + async def test_domain_mismatch_in_resolver_mode_returns_none(self): + store = OneSlotStore() + _stored_context(store, domain="tenant-a.auth0.local") + resolver = AsyncMock(return_value="tenant-b.auth0.local") + client = AnonymousClient( + domain=resolver, client_id=CLIENT_ID, client_secret=CLIENT_SECRET, + secret=SECRET, anonymous_store=store, + ) + result = await client.get_session() + assert result is None + + @pytest.mark.asyncio + async def test_domain_match_in_resolver_mode_returns_session(self): + store = OneSlotStore() + _stored_context(store, domain="tenant-a.auth0.local") + resolver = AsyncMock(return_value="tenant-a.auth0.local") + client = AnonymousClient( + domain=resolver, client_id=CLIENT_ID, client_secret=CLIENT_SECRET, + secret=SECRET, anonymous_store=store, + ) + result = await client.get_session() + assert result is not None + + @pytest.mark.asyncio + async def test_resolver_exception_in_get_session_returns_none(self): + """A resolver error during get_session must fail closed.""" + store = OneSlotStore() + _stored_context(store, domain="tenant-a.auth0.local") + resolver = AsyncMock(side_effect=RuntimeError("resolver down")) + client = AnonymousClient( + domain=resolver, client_id=CLIENT_ID, client_secret=CLIENT_SECRET, + secret=SECRET, anonymous_store=store, + ) + result = await client.get_session() + assert result is None + + @pytest.mark.asyncio + async def test_static_domain_client_returns_session_without_domain_check(self): + """A static-domain client must not invoke the resolver and must return the session.""" + store = OneSlotStore() + _stored_context(store, domain="some-old-domain.auth0.local") + client = _make_client(anonymous_store=store) + with patch.object(client, "_resolve_domain") as mock_resolver: + result = await client.get_session() + mock_resolver.assert_not_called() + assert result is not None diff --git a/src/auth0_server_python/tests/test_server_client.py b/src/auth0_server_python/tests/test_server_client.py index 77b43f5..c7ac3b4 100644 --- a/src/auth0_server_python/tests/test_server_client.py +++ b/src/auth0_server_python/tests/test_server_client.py @@ -17,14 +17,18 @@ 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 ( _EC_ALLOWED_METHODS, + INTERNAL_AUTHORIZE_PARAMS, ServerClient, is_federated_domain, ) from auth0_server_python.auth_types import ( + AnonymousSessionContext, + AnonymousTokenSetEntry, CompleteConnectAccountRequest, ConnectAccountOptions, ConnectAccountRequest, @@ -51,6 +55,7 @@ TransactionData, UserClaims, ) +from auth0_server_python.encryption.encrypt import encrypt from auth0_server_python.error import ( AccessTokenError, AccessTokenErrorCode, @@ -76,6 +81,7 @@ SessionExpiredError, StartLinkUserError, ) +from auth0_server_python.tests.store_fakes import OneSlotStore from auth0_server_python.utils import PKCE, State @@ -9902,6 +9908,773 @@ async def test_complete_interactive_login_milliseconds_ceiling_fails_open(mocker assert stored_state.internal.session_expires_at is None +# ============================================================================= +# ANONYMOUS SESSIONS - WIRING AND LOGIN-INJECTION TESTS +# ============================================================================= + + +def _make_anon_context(secret, **overrides): + ts_keys = {"access_token", "expires_at", "audience", "scope"} + ts_defaults = { + "access_token": "anon_at1", + "expires_at": int(time.time()) + 3600, + } + ctx_defaults = { + "session_token": "ANON_TOKEN_1", + "created_at": int(time.time()), + } + for k in list(overrides): + if k in ts_keys: + ts_defaults[k] = overrides.pop(k) + else: + ctx_defaults[k] = overrides.pop(k) + token_set = AnonymousTokenSetEntry(**ts_defaults) + context = AnonymousSessionContext(token_sets=[token_set], **ctx_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_transfer_ticket_not_session_token(mocker): + """The /authorize URL carries the minted anon_transfer_token, never the raw session_token.""" + 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"}, + ) + mocker.patch.object( + client._anonymous_client, + "exchange_transfer_token_for_injection", + AsyncMock(return_value="TICKET_ABC"), + ) + 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("anon_transfer_token") == "TICKET_ABC" + assert "session_token" not in captured + + +@pytest.mark.asyncio +async def test_start_interactive_login_does_not_persist_transfer_token_in_transaction_data(mocker): + """The short-lived ticket rides the URL only and is never written to the transaction record.""" + 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._anonymous_client, + "exchange_transfer_token_for_injection", + AsyncMock(return_value="TICKET_ABC"), + ) + 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 not hasattr(stored_tx, "session_token") + assert not hasattr(stored_tx, "anon_transfer_token") + + +@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() + 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): + """An undecryptable stored token denies the link but never aborts 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", + }, + ) + exchange = mocker.patch.object( + client._anonymous_client, + "exchange_transfer_token_for_injection", + AsyncMock(return_value="TICKET_ABC"), + ) + 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() + exchange.assert_not_awaited() + assert "anon_transfer_token" not in captured + assert "session_token" not in captured + + +@pytest.mark.asyncio +async def test_start_interactive_login_constructor_fixation_blocked_no_active_session(): + """The unconditional pop() at the injection site blocks a session_token supplied via constructor authorization_params.""" + 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"}, + ) + mocker.patch.object( + client._anonymous_client, + "exchange_transfer_token_for_injection", + AsyncMock(return_value="TICKET_ABC"), + ) + 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("anon_transfer_token") == "TICKET_ABC" + + +@pytest.mark.asyncio +async def test_start_interactive_login_suppresses_injection_on_enterprise_connect(mocker): + """Enterprise Connect skips anonymous-session linking entirely, no exchange and no param.""" + 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, + enterprise_connect=True, + authorization_params={"redirect_uri": "/test_redirect_uri"}, + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"authorization_endpoint": "https://auth0.local/authorize"}, + ) + exchange = mocker.patch.object( + client._anonymous_client, + "exchange_transfer_token_for_injection", + AsyncMock(return_value="TICKET_ABC"), + ) + 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() + exchange.assert_not_awaited() + assert "anon_transfer_token" not in captured + assert "session_token" not in captured + + +@pytest.mark.asyncio +async def test_start_interactive_login_fail_open_when_exchange_returns_none(mocker): + """A failed/absent exchange yields no param and still returns the login URL.""" + 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"}, + ) + mocker.patch.object( + client._anonymous_client, + "exchange_transfer_token_for_injection", + AsyncMock(return_value=None), + ) + 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) + url = await client.start_interactive_login() + assert "anon_transfer_token" not in captured + assert url == "https://auth0.local/authorize?client_id=" + + +# ── end anonymous session on authenticated logout ─────────────────────────────── + + +# ── clear_anonymous_session_on_login ──────────────────────────────────────────── + + +def _setup_complete_interactive_login(client, mocker): + """Patch the minimum set of collaborators needed for complete_interactive_login to succeed.""" + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"issuer": "https://auth0.local/", "token_endpoint": "https://auth0.local/token"}, + ) + mocker.patch.object(client, "_get_jwks_cached", return_value={"keys": [{"kty": "RSA", "kid": "k1"}]}) + mocker.patch.object( + client._oauth, + "fetch_token", + AsyncMock(return_value={"access_token": "at1", "id_token": "id_jwt", "scope": "openid"}), + ) + mocker.patch("jwt.get_unverified_header", return_value={"kid": "k1"}) + mock_key = mocker.MagicMock() + mock_key.key = "pem" + mocker.patch("jwt.PyJWK.from_dict", return_value=mock_key) + mocker.patch( + "jwt.decode", + return_value={"sub": "u1", "iss": "https://auth0.local/", "aud": "cid"}, + ) + + +@pytest.mark.asyncio +async def test_complete_interactive_login_does_not_clear_anonymous_session_when_disabled(mocker): + """With clear_anonymous_session_on_login=False, the anonymous session is not touched on login.""" + secret = "a-test-secret-with-enough-length" + anon_store = OneSlotStore() + anon_store.slot = (ANON_IDENTIFIER, {"context": _make_anon_context(secret)}) + tx_store = AsyncMock() + tx_store.get.return_value = TransactionData(code_verifier="cv1", domain="auth0.local") + client = ServerClient( + domain="auth0.local", + client_id="cid", + client_secret="csecret", + transaction_store=tx_store, + state_store=AsyncMock(), + anonymous_store=anon_store, + secret=secret, + clear_anonymous_session_on_login=False, + ) + _setup_complete_interactive_login(client, mocker) + logout_spy = mocker.patch.object(client._anonymous_client, "logout", AsyncMock()) + + await client.complete_interactive_login("https://auth0.local/cb?code=c&state=s") + + logout_spy.assert_not_awaited() + assert anon_store.slot is not None + + +@pytest.mark.asyncio +async def test_complete_interactive_login_clears_anonymous_session_by_default(mocker): + """By default (clear_anonymous_session_on_login=True), anonymous.logout() is called after the session is written.""" + secret = "a-test-secret-with-enough-length" + anon_store = OneSlotStore() + anon_store.slot = (ANON_IDENTIFIER, {"context": _make_anon_context(secret)}) + tx_store = AsyncMock() + tx_store.get.return_value = TransactionData(code_verifier="cv1", domain="auth0.local") + client = ServerClient( + domain="auth0.local", + client_id="cid", + client_secret="csecret", + transaction_store=tx_store, + state_store=AsyncMock(), + anonymous_store=anon_store, + secret=secret, + ) + _setup_complete_interactive_login(client, mocker) + logout_spy = mocker.patch.object(client._anonymous_client, "logout", AsyncMock()) + + result = await client.complete_interactive_login("https://auth0.local/cb?code=c&state=s") + + assert "state_data" in result + logout_spy.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_complete_interactive_login_cleanup_failure_does_not_fail_login(mocker): + """A failing anonymous.logout() is swallowed and must never fail an otherwise-successful login.""" + secret = "a-test-secret-with-enough-length" + anon_store = OneSlotStore() + anon_store.slot = (ANON_IDENTIFIER, {"context": _make_anon_context(secret)}) + tx_store = AsyncMock() + tx_store.get.return_value = TransactionData(code_verifier="cv1", domain="auth0.local") + client = ServerClient( + domain="auth0.local", + client_id="cid", + client_secret="csecret", + transaction_store=tx_store, + state_store=AsyncMock(), + anonymous_store=anon_store, + secret=secret, + clear_anonymous_session_on_login=True, + ) + _setup_complete_interactive_login(client, mocker) + mocker.patch.object( + client._anonymous_client, "logout", AsyncMock(side_effect=Exception("store down")) + ) + + result = await client.complete_interactive_login("https://auth0.local/cb?code=c&state=s") + + assert "state_data" in result + + +class _CapturingHttpClient: + """Fake httpx.AsyncClient that records POST URLs and returns a fixed status.""" + + posted: list = [] + status = 204 + + 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): + type(self).posted.append(url) + + class _Resp: + status_code = _CapturingHttpClient.status + + def json(self): + return {} + + return _Resp() + + +@pytest.mark.asyncio +async def test_logout_ends_active_anonymous_session(mocker): + """Authenticated logout clears the local anonymous session with no remote call.""" + 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="cid", + client_secret="csecret", + state_store=AsyncMock(), + transaction_store=AsyncMock(), + anonymous_store=anon_store, + secret=secret, + ) + _CapturingHttpClient.posted = [] + _CapturingHttpClient.status = 204 + mocker.patch("httpx.AsyncClient", _CapturingHttpClient) + + url = await client.logout() + + assert not any(u.endswith("/anonymous/logout") for u in _CapturingHttpClient.posted) + assert anon_store.slot is None + assert "logout" in url + + +@pytest.mark.asyncio +async def test_logout_no_anonymous_session_makes_no_remote_call(mocker): + """With no active anonymous session, authenticated logout makes no anonymous remote call.""" + anon_store = OneSlotStore() + client = ServerClient( + domain="auth0.local", + client_id="cid", + client_secret="csecret", + state_store=AsyncMock(), + transaction_store=AsyncMock(), + anonymous_store=anon_store, + secret="a-test-secret-with-enough-length", + ) + _CapturingHttpClient.posted = [] + _CapturingHttpClient.status = 204 + mocker.patch("httpx.AsyncClient", _CapturingHttpClient) + + await client.logout() + + assert _CapturingHttpClient.posted == [] + + +@pytest.mark.asyncio +async def test_logout_unchanged_without_anonymous_store(mocker): + """With no anonymous_store configured, logout is unchanged and makes no anonymous remote call.""" + mock_state_store = AsyncMock() + client = ServerClient( + domain="auth0.local", + client_id="cid", + client_secret="csecret", + state_store=mock_state_store, + secret="a-test-secret-with-enough-length", + ) + _CapturingHttpClient.posted = [] + _CapturingHttpClient.status = 204 + mocker.patch("httpx.AsyncClient", _CapturingHttpClient) + + url = await client.logout() + + mock_state_store.delete.assert_awaited_once() + assert _CapturingHttpClient.posted == [] + assert "logout" in url + + +@pytest.mark.asyncio +async def test_logout_survives_anonymous_session_cleanup_failure(mocker): + """A failure clearing the anonymous session must not break the authenticated logout.""" + anon_store = AsyncMock() + anon_store.get = AsyncMock(return_value={"context": "encrypted"}) + anon_store.delete = AsyncMock(side_effect=RuntimeError("store delete failed")) + client = ServerClient( + domain="auth0.local", + client_id="cid", + client_secret="csecret", + state_store=AsyncMock(), + transaction_store=AsyncMock(), + anonymous_store=anon_store, + secret="a-test-secret-with-enough-length", + ) + _CapturingHttpClient.posted = [] + _CapturingHttpClient.status = 204 + mocker.patch("httpx.AsyncClient", _CapturingHttpClient) + + url = await client.logout() + + assert "logout" in url + anon_store.delete.assert_awaited_once() + + +# ── Store-collision regression ───────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_anonymous_write_cannot_destroy_authenticated_session_on_shared_store(): + """A write to the anonymous store instance never touches the authenticated session on a separate store instance.""" + 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")} + ) + + 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(): + """A missing anonymous_store raises before any write, never silently falling back to 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") + 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 + + # === Enterprise Connect === def _make_ec_client(**overrides):