diff --git a/README.md b/README.md index 2ff26cf..2a37950 100644 --- a/README.md +++ b/README.md @@ -233,6 +233,10 @@ 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) + +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). + ## Feedback ### Contributing diff --git a/examples/EnterpriseConnect.md b/examples/EnterpriseConnect.md new file mode 100644 index 0000000..3e5652e --- /dev/null +++ b/examples/EnterpriseConnect.md @@ -0,0 +1,232 @@ +# Enterprise Connect (Embedded Login) + +Enterprise Connect lets your application sign users in through their company's identity provider while **your app owns the session**. Auth0 acts as a pure SSO relay: it authenticates the user and returns verified claims, but issues no refresh token and holds no session on your behalf. This guide covers the `enterprise_connect` mode on `ServerClient`, email-domain discovery, the callback contract, and federated logout. + +> [!IMPORTANT] +> Enterprise Connect is an Early Access feature. Tenant setup (entitlements, connection type, and the exact claims a token carries) depends on your Auth0 configuration and may change. Treat the tenant-side steps below as a starting point and confirm them against your tenant. The SDK behavior described here is stable. + +> [!IMPORTANT] +> These flows are for confidential server-side applications. The verified claims and access token are handed to your server, which creates and owns the user session. The browser should only ever receive your application's own session cookie or opaque session reference, never an Auth0 token. + +## Table of Contents + +- [How the flow works](#how-the-flow-works) +- [Prerequisites](#prerequisites) +- [1. Configure the client](#1-configure-the-client) +- [2. Discover and start the login](#2-discover-and-start-the-login) +- [3. Complete the callback](#3-complete-the-callback) +- [4. Protect your routes](#4-protect-your-routes) +- [5. Organizations and multi-tenant apps](#5-organizations-and-multi-tenant-apps) +- [6. Logout](#6-logout) +- [What is not available in Enterprise Connect](#what-is-not-available-in-enterprise-connect) +- [Error Handling](#error-handling) + +## How the flow works + +1. The user enters their email. Your app calls `start_enterprise_login()`, which runs [WebFinger](https://datatracker.ietf.org/doc/html/rfc7033) discovery on the email domain. +2. If the domain is managed by Auth0 for enterprise SSO, the SDK returns an authorization URL with the email as `login_hint` so Auth0 can resolve the connection and organization. If it is not managed, the method returns `None` and your app falls back to its own login. +3. The user authenticates at their identity provider and is redirected back to your callback. +4. Your app calls `complete_interactive_login()`. The SDK exchanges the code, verifies the ID token's signature and issuer, and returns the claims from it. It persists **nothing** and issues no refresh token. +5. Your app creates its own first-party session from the returned claims. + +The contract is inverted from a normal login: the SDK does not store a session, so the session-reading methods (`get_session`, `get_access_token`) are not available in this mode. + +## Prerequisites + +Enterprise Connect requires a **Regular Web Application** with a client secret. The tenant and connection must be provisioned for Enterprise Connect (Early Access), and WebFinger discovery must be enabled on the tenant. Work with your Auth0 contact to confirm entitlements for your tenant. + +Do not request `offline_access` and do not set a static `organization` on the client. Enterprise Connect issues no refresh token, and the organization is resolved from the login email at Auth0. The SDK warns at construction if either is set. + +## 1. Configure the client + +Opt in with `enterprise_connect=True`. Supply a `transaction_store` (used to protect the callback with `state` and PKCE); a `state_store` is not needed, because the SDK persists no session. + +```python +from auth0_server_python.auth_server.server_client import ServerClient + +server_client = ServerClient( + domain="YOUR_AUTH0_DOMAIN", + client_id="YOUR_CLIENT_ID", + client_secret="YOUR_CLIENT_SECRET", + secret="YOUR_SECRET", + redirect_uri="https://app.example.com/auth/callback", + authorization_params={"scope": "openid profile email"}, + enterprise_connect=True, +) +``` + +For apps using request/response-backed stores or multiple custom domains, pass `store_options={"request": request, "response": response}` to each method that reads or writes transaction state. + +## 2. Discover and start the login + +Your app must serve a login page that collects the user's work email. Pass it to `start_enterprise_login()`, which runs WebFinger discovery and returns the authorization URL when the domain is managed, or `None` when it is not. + +```python +from auth0_server_python.auth_types import StartEnterpriseLoginOptions +from auth0_server_python.error import MissingRequiredArgumentError, InvalidArgumentError + +try: + auth_url = await server_client.start_enterprise_login( + StartEnterpriseLoginOptions( + email=user_email, + app_state={"return_to": "/dashboard"}, + ), + store_options={"request": request, "response": response}, + ) +except (MissingRequiredArgumentError, InvalidArgumentError): + auth_url = None + +if auth_url: + return redirect(auth_url) +return redirect("/login/password") +``` + +> [!IMPORTANT] +> Discovery is a routing hint, not an authorization decision. It fails closed to "not managed" on any error, so a discovery failure routes the user to your fallback login rather than granting access. It never, on its own, signs anyone in - the callback must still complete. + +If you only need the discovery signal (for example, to decide which login button to show), call the standalone helper. It takes no client instance and is stateless. + +```python +from auth0_server_python.auth_server import is_federated_domain + +managed = await is_federated_domain("YOUR_AUTH0_DOMAIN", "acme.example") +``` + +## 3. Complete the callback + +When Auth0 redirects back, call `complete_interactive_login()`. In Enterprise Connect mode it returns the verified claims and an access token instead of a session record. + +```python +result = await server_client.complete_interactive_login( + str(request.url), + store_options={"request": request, "response": response}, +) + +user = result["user"] +access_token = result["token_set"]["access_token"] +id_token = result["id_token"] +domain = result["domain"] + +create_app_session(user_id=user.sub) + +app_state = result.get("app_state") or {} +return redirect(app_state.get("return_to", "/")) +``` + +The returned dict contains: + +- `user` - the verified `UserClaims`, parsed from the ID token that Enterprise Connect returns +- `token_set` - `audience`, `access_token`, `scope`, and `expires_at` +- `id_token` - the raw ID token, for your own use +- `domain` - the Auth0 domain the login came from +- `app_state` - present only when you passed `app_state` at `start_enterprise_login()` + +`result` is a plain dict, so index it with `result["user"]`. `user` is a `UserClaims` model with no dict access, so read claims by attribute like `user.org_id`. + +The SDK verifies the ID token's signature and issuer, and derives the returned claims from it, before returning. It does not write a session store record and retains no refresh token. + +## 4. Protect your routes + +Your app owns the session - the SDK holds nothing. Check your session store at the start of any route that requires authentication and redirect to your login page when the session is absent. + +```python +user = your_session.get("user") +if not user: + return redirect("/login") + +your_template.render(user=user) +``` + +> [!IMPORTANT] +> Do not store an Auth0 access or ID token in your session. Store only the claims you need (for example `sub`, `email`, `org_id`). The browser must never see an Auth0 token. + +## 5. Organizations and multi-tenant apps + +Auth0 stamps the resolved organization into the token as `org_id`, available on `result["user"]`. The SDK surfaces it but does not enforce it. It cannot know which organization *your* app expected for this user. + +> [!WARNING] +> Validate `org_id` after every callback, regardless of routing. WebFinger discovery and `login_hint` are routing mechanisms, not security controls. On their own they do not prove the user belongs to a customer you serve. Read `org_id` from the returned claims and check it against your own record of known organizations before creating the session. Without this check, a user who authenticates through any managed connection could obtain a session in a context you did not intend. This is an authorization decision your app owns. + +```python +user = result["user"] +if user.org_id not in allowed_orgs_for(current_customer): + raise Forbidden("user does not belong to this organization") +``` + +If you serve exactly one organization, this is a single check against your one known org, not a reason to skip it. An app that skips it today can silently let users in from other tenants the day it onboards a second customer. + +## 6. Logout + +Clear your own application session first, then send the user to the Auth0 logout URL. + +```python +from auth0_server_python.auth_types import LogoutOptions + +destroy_app_session() + +logout_url = await server_client.logout( + LogoutOptions(return_to="https://app.example.com/login"), + store_options={"request": request, "response": response}, +) +return redirect(logout_url) +``` + +By default this ends the Auth0 session but leaves the upstream identity provider session intact, so the user is not re-prompted at their IdP on the next login. To also terminate the IdP session, pass `federated=True`. + +Federated logout ends the corporate IdP session itself, which can also sign the user out of other applications that share that same enterprise SSO, not just yours. Weigh that against the shared-device benefit before enabling it by default. + +```python +logout_url = await server_client.logout( + LogoutOptions(return_to="https://app.example.com/login", federated=True), +) +``` + +> [!NOTE] +> `return_to` must be an absolute URL on your tenant's Allowed Logout URLs list. Auth0 rejects a URL that is not allow-listed. + +## What is not available in Enterprise Connect + +These members work in Enterprise Connect mode: + +| Member | Notes | +|---|---| +| `start_enterprise_login()` | EC login entry point | +| `start_interactive_login()` | Writes the transaction store only | +| `complete_interactive_login()` | Returns verified claims without persisting a session | +| `logout()` | Clears transaction state and returns the Auth0 logout URL | +| `custom_token_exchange()` | Works once, while the callback access token is valid. No refresh after it expires | +| `handle_backchannel_logout()` | No-op. The SDK holds no session to revoke | + +Everything else raises `EnterpriseConnectError`. Branch on `code`: +- `enterprise_connect_session_unavailable` - `get_session()` was called +- `enterprise_connect_access_token_unavailable` - `get_access_token()` was called. Read the token from `complete_interactive_login()` instead +- `enterprise_connect_method_unavailable` - any other session or refresh-dependent member was called + +Own the session and any token refresh in your app. + +## Error Handling + +```python +from auth0_server_python.error import ApiError, EnterpriseConnectError + +try: + result = await server_client.complete_interactive_login( + str(request.url), + store_options={"request": request, "response": response}, + ) +except ApiError as e: + return {"error": e.code} + +try: + await server_client.get_access_token() +except EnterpriseConnectError as e: + return {"error": e.code} +``` + +Errors you may see: + +- `EnterpriseConnectError` - a session or token method is unavailable in this mode. Branch on `code`: + - `enterprise_connect_session_unavailable` - `get_session()` was called + - `enterprise_connect_access_token_unavailable` - `get_access_token()` was called + - `enterprise_connect_method_unavailable` - any other session or refresh dependent member was called +- `ApiError` - the token exchange failed, or the login returned no verifiable claims (`invalid_response`) diff --git a/references/docs-update.md b/references/docs-update.md index b7b0540..b3e8851 100644 --- a/references/docs-update.md +++ b/references/docs-update.md @@ -64,3 +64,4 @@ it from `README.md`'s section for the feature. Name the file after the flow, mat | MCD domain resolver | `examples/MultipleCustomDomains.md` | | Account linking / unlinking | `examples/UserLinking.md` | | Passwordless email/SMS OTP + magic link | `examples/Passwordless.md` | +| Enterprise Connect embedded login (`enterprise_connect`, `start_enterprise_login`, `is_federated_domain`) | `examples/EnterpriseConnect.md` | diff --git a/references/flow-map.md b/references/flow-map.md index 6f9c8c1..28d7517 100644 --- a/references/flow-map.md +++ b/references/flow-map.md @@ -17,6 +17,7 @@ Before working on a flow, read its entry points and supporting modules. Every fl | Passkeys | `passkey_signup_challenge`, `passkey_login_challenge`, `signin_with_passkey` | `auth_schemes/dpop_auth.py` — passkey sign-in is the DPoP-bound path | `examples/Passkeys.md` | | My Account | `MyAccountClient` (factors, authentication methods, enroll/verify) | `auth_schemes/dpop_auth.py`; stateless — every call takes a user token | `examples/MyAccountAuthenticationMethods.md` | | 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` | 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 a06ef2e..52449fb 100644 --- a/src/auth0_server_python/auth_server/__init__.py +++ b/src/auth0_server_python/auth_server/__init__.py @@ -1,6 +1,12 @@ from .mfa_client import MfaClient from .my_account_client import MyAccountClient from .passwordless_client import PasswordlessClient -from .server_client import ServerClient +from .server_client import ServerClient, is_federated_domain -__all__ = ["ServerClient", "MyAccountClient", "MfaClient", "PasswordlessClient"] +__all__ = [ + "ServerClient", + "MyAccountClient", + "MfaClient", + "PasswordlessClient", + "is_federated_domain", +] diff --git a/src/auth0_server_python/auth_server/server_client.py b/src/auth0_server_python/auth_server/server_client.py index fb984b5..6e7f783 100644 --- a/src/auth0_server_python/auth_server/server_client.py +++ b/src/auth0_server_python/auth_server/server_client.py @@ -6,6 +6,7 @@ import asyncio import json import time +import warnings from collections import OrderedDict from typing import TYPE_CHECKING, Any, Callable, Generic, Optional, TypeVar, Union @@ -49,6 +50,7 @@ PasskeyTokenResponse, PasskeyUserProfile, SessionTransferTokenResult, + StartEnterpriseLoginOptions, StartInteractiveLoginOptions, StateData, TokenExchangeResponse, @@ -67,6 +69,8 @@ CustomTokenExchangeError, CustomTokenExchangeErrorCode, DomainResolverError, + EnterpriseConnectError, + EnterpriseConnectErrorCode, InvalidArgumentError, IssuerValidationError, MfaRequiredError, @@ -101,6 +105,44 @@ # actor_token_type URN when the actor is sourced from the agent session's ID token. ID_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:id_token" +# WebFinger rel identifying an OIDC issuer, per the OIDC Discovery spec. +WEBFINGER_ISSUER_REL = "http://openid.net/specs/connect/1.0/issuer" +# TTLs for the Enterprise Connect discovery cache. A managed domain rarely +# flips to unmanaged, so a positive result is held longer than a negative one. +WEBFINGER_CACHE_TTL_FOUND = 60 +WEBFINGER_CACHE_TTL_NOT_FOUND = 15 +WEBFINGER_CACHE_MAX_ENTRIES = 1000 + + +def _webfinger_resource(email_domain: str) -> str: + """Build the WebFinger `resource` value for an email domain.""" + return f"urn:auth0:discovery:domain:{email_domain}" + + +def _interpret_webfinger_response(status_code: int, body): + """ + Map a WebFinger response to a routing decision. Fails closed to not-federated. + + Args: + status_code: HTTP status of the WebFinger response. + body: Parsed JSON body for a 200 response, otherwise None. + + Returns: + A `(is_federated, cache_ttl_seconds)` tuple. `cache_ttl_seconds` is None + when the result must not be cached (transient or ambiguous responses). + """ + if status_code == 200: + links = body.get("links", []) if isinstance(body, dict) else [] + if any( + isinstance(link, dict) and link.get("rel") == WEBFINGER_ISSUER_REL + for link in links + ): + return (True, WEBFINGER_CACHE_TTL_FOUND) + return (False, None) + if status_code == 404: + return (False, WEBFINGER_CACHE_TTL_NOT_FOUND) + return (False, None) + class ServerClient(Generic[TStoreOptions]): """ @@ -133,6 +175,7 @@ def __init__( pushed_authorization_requests: bool = False, organization: Optional[str] = None, mfa_token_ttl: int = DEFAULT_MFA_TOKEN_TTL, + enterprise_connect: bool = False, ): """ Initialize the Auth0 server client. @@ -158,6 +201,9 @@ def __init__( `mfa.verify()`/`mfa.challenge_authenticator()` reject it as expired. Defaults to 300 (5 minutes). Increase for authenticator flows that need more time (e.g. OOB push approval on a slow connection). + enterprise_connect: Opt into Enterprise Connect mode, where Auth0 acts + as a pure SSO relay and the integrator's app owns the session. The + SDK persists no session and issues no refresh token. Off by default. Raises: ConfigurationError: If `mfa_token_ttl` is not a positive number of seconds. @@ -202,6 +248,8 @@ def __init__( self._default_authorization_params = authorization_params or {} self._pushed_authorization_requests = pushed_authorization_requests # store the flag self._organization = organization + self._enterprise_connect = enterprise_connect + self._webfinger_cache: OrderedDict[str, dict] = OrderedDict() # Initialize stores self._transaction_store = transaction_store @@ -245,6 +293,47 @@ def __init__( self._passwordless_client = PasswordlessClient(self) + if enterprise_connect: + self._warn_on_enterprise_connect_config() + + def _warn_on_enterprise_connect_config(self) -> None: + """ + Warn when Enterprise Connect is combined with settings it ignores. + + Enterprise Connect issues no refresh token and resolves the organization + from the login email at Auth0, so a requested `offline_access` scope or a + static `organization` is misleading configuration rather than an error. + """ + default_scope = str(self._default_authorization_params.get("scope", "")) + if "offline_access" in default_scope.split(): + warnings.warn( + "enterprise_connect is enabled but 'offline_access' is in the default " + "scope. Enterprise Connect issues no refresh token, so it has no effect.", + stacklevel=2, + ) + if self._organization: + warnings.warn( + "enterprise_connect is enabled but a static 'organization' is set. " + "Enterprise Connect resolves the organization from the login email at " + "Auth0, so the static value is ignored for enterprise logins.", + stacklevel=2, + ) + + def _reject_in_enterprise_connect(self, method: str) -> None: + """Rejects a session or refresh dependent call on an Enterprise Connect client. + + Args: + method: Public member name, used in the error message. + + Raises: + EnterpriseConnectError: When the client is in Enterprise Connect mode. + """ + if self._enterprise_connect: + raise EnterpriseConnectError( + EnterpriseConnectErrorCode.METHOD_UNAVAILABLE, + f"{method} is unavailable in Enterprise Connect mode.", + ) + def _get_http_client(self, **kwargs) -> httpx.AsyncClient: """Return an httpx.AsyncClient with telemetry headers injected.""" headers = {**kwargs.pop("headers", {}), **self._telemetry_headers} @@ -608,7 +697,11 @@ async def start_interactive_login( auth_params["scope"] = merged_scope # Typed org/invitation fields win over anything already in auth_params from authorization_params. - resolved_org = options.organization or self._organization + # In Enterprise Connect the org is resolved from the login email at Auth0, so the + # static client-level organization is never auto-applied - only an explicit per-login value. + resolved_org = options.organization + if not resolved_org and not self._enterprise_connect: + resolved_org = self._organization if resolved_org and not resolved_org.strip(): raise InvalidArgumentError("organization", "organization must not be blank") if resolved_org: @@ -849,6 +942,17 @@ async def complete_interactive_login( ) + if self._enterprise_connect: + return await self._complete_enterprise_login( + transaction_identifier=transaction_identifier, + transaction_data=transaction_data, + token_response=token_response, + user_claims=user_claims, + id_token=id_token, + origin_domain=origin_domain, + store_options=store_options, + ) + try: state_data = await self._persist_session_from_token_response( token_response=token_response, @@ -948,6 +1052,66 @@ async def _persist_session_from_token_response( ) return state_data + async def _complete_enterprise_login( + self, + *, + transaction_identifier: str, + transaction_data: "TransactionData", + token_response: dict[str, Any], + user_claims: Optional["UserClaims"], + id_token: Optional[str], + origin_domain: str, + store_options: Optional[dict[str, Any]] = None, + ) -> dict[str, Any]: + """ + Finish an Enterprise Connect login without persisting a session. + + The token has already passed issuer and signature validation. Enterprise + Connect hands the verified claims back to the integrator, which owns the + session, so nothing is written to a state store and no refresh token is + retained. The transaction is consumed on success. + + Args: + transaction_identifier: Store key of the login transaction to consume. + transaction_data: The login transaction, source of the token audience. + token_response: The verified token endpoint response. + user_claims: Claims parsed from the verified ID token or userinfo. + id_token: The raw ID token, returned for the integrator's own use. + origin_domain: Resolved Auth0 domain the login came from. + store_options: Options passed to the transaction store. + + Returns: + A dict with the verified `user` claims, an access-token `token_set`, + the raw `id_token`, and the `domain`, plus `app_state` when present. + + Raises: + ApiError: If the login returned no verifiable user claims. + """ + if user_claims is None: + raise ApiError( + "invalid_response", + "Enterprise Connect login returned no verifiable user claims", + ) + + await self._transaction_store.delete(transaction_identifier, options=store_options) + + now = int(time.time()) + token_set = { + "audience": transaction_data.audience or self.DEFAULT_AUDIENCE_STATE_KEY, + "access_token": token_response.get("access_token", ""), + "scope": token_response.get("scope", ""), + "expires_at": now + token_response.get("expires_in", 3600), + } + result = { + "user": user_claims, + "token_set": token_set, + "id_token": id_token, + "domain": origin_domain, + } + if transaction_data.app_state: + result["app_state"] = transaction_data.app_state + return result + async def _establish_session_from_mfa_verify_response( self, *, @@ -1045,7 +1209,12 @@ async def get_user(self, store_options: Optional[dict[str, Any]] = None) -> Opti Returns: The user, or None if no user found in the store. + + Raises: + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + self._reject_in_enterprise_connect("get_user") + state_data = await self._state_store.get(self._state_identifier, store_options) if state_data: @@ -1078,7 +1247,15 @@ async def get_session(self, store_options: Optional[dict[str, Any]] = None) -> O Returns: The session, or None if no session found in the store. + + Raises: + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + if self._enterprise_connect: + raise EnterpriseConnectError( + EnterpriseConnectErrorCode.SESSION_UNAVAILABLE, + "get_session is unavailable in Enterprise Connect mode.", + ) state_data = await self._state_store.get(self._state_identifier, store_options) if state_data: @@ -1112,23 +1289,27 @@ async def logout( options = options or LogoutOptions() if not self._domain_resolver: - await self._state_store.delete(self._state_identifier, store_options) + # No domain resolver means one fixed domain. Delete the session when a + # state store exists. Enterprise Connect has none, so nothing to delete. + if self._state_store is not None: + await self._state_store.delete(self._state_identifier, store_options) domain = self._domain else: # Resolver mode: delete session if domains match domain = await self._resolve_current_domain(store_options) - state_data = await self._state_store.get(self._state_identifier, store_options) + if self._state_store is not None: + state_data = await self._state_store.get(self._state_identifier, store_options) - if state_data: - if hasattr(state_data, "dict") and callable(state_data.dict): - state_data = state_data.dict() - session_domain = self._get_session_domain(state_data) - if session_domain and self._normalize_url(session_domain) == self._normalize_url(domain): - await self._state_store.delete(self._state_identifier, store_options) + if state_data: + if hasattr(state_data, "dict") and callable(state_data.dict): + state_data = state_data.dict() + session_domain = self._get_session_domain(state_data) + if session_domain and self._normalize_url(session_domain) == self._normalize_url(domain): + await self._state_store.delete(self._state_identifier, store_options) # Return logout URL for the current resolved domain logout_url = URL.create_logout_url( - domain, self._client_id, options.return_to) + domain, self._client_id, options.return_to, federated=bool(options.federated)) return logout_url @@ -1144,6 +1325,10 @@ async def handle_backchannel_logout( logout_token: The logout token sent by Auth0 store_options: Options to pass to the state store """ + # Enterprise Connect keeps no session to revoke, so back-channel logout is a no-op. + if self._enterprise_connect: + return + if not logout_token: raise BackchannelLogoutError("Missing logout token") @@ -1250,7 +1435,13 @@ async def get_access_token( Raises: AccessTokenError: If the token is expired and no refresh token is available. + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + if self._enterprise_connect: + raise EnterpriseConnectError( + EnterpriseConnectErrorCode.ACCESS_TOKEN_UNAVAILABLE, + "get_access_token is unavailable in Enterprise Connect mode.", + ) state_data = await self._state_store.get(self._state_identifier, store_options) # Domain check should work for both Pydantic models and plain dicts @@ -1367,10 +1558,13 @@ async def get_token_by_refresh_token(self, options: dict[str, Any]) -> dict[str, Raises: AccessTokenError: If there was an issue requesting the access token. ConfigurationError: If no client authentication is configured. + EnterpriseConnectError: If the client is configured for Enterprise Connect. Returns: A dictionary containing the token response from Auth0. """ + self._reject_in_enterprise_connect("get_token_by_refresh_token") + refresh_token = options.get("refresh_token") if not refresh_token: raise MissingRequiredArgumentError("refresh_token") @@ -1532,7 +1726,12 @@ async def login_backchannel( Returns: A dictionary containing the authorizationDetails (when RAR was used). + + Raises: + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + self._reject_in_enterprise_connect("login_backchannel") + token_endpoint_response = await self.backchannel_authentication({ "binding_message": options.get("binding_message"), "login_hint": options.get("login_hint"), @@ -1588,7 +1787,10 @@ async def backchannel_authentication( Raises: ApiError: If the backchannel authentication fails + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + self._reject_in_enterprise_connect("backchannel_authentication") + backchannel_data = await self.initiate_backchannel_authentication(options, store_options=store_options) auth_req_id = backchannel_data.get("auth_req_id") expires_in = backchannel_data.get( @@ -1870,7 +2072,12 @@ async def start_link_user( Returns: URL to redirect the user to for authentication. + + Raises: + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + self._reject_in_enterprise_connect("start_link_user") + state_data = await self._state_store.get(self._state_identifier, store_options) if not state_data or not state_data.get("id_token"): @@ -1939,7 +2146,11 @@ async def complete_link_user( Returns: Dictionary containing the original app state + + Raises: + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + self._reject_in_enterprise_connect("complete_link_user") # We can reuse the interactive login completion since the flow is similar result = await self.complete_interactive_login(url, store_options) @@ -1963,7 +2174,12 @@ async def start_unlink_user( Returns: URL to redirect the user to for authentication. + + Raises: + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + self._reject_in_enterprise_connect("start_unlink_user") + state_data = await self._state_store.get(self._state_identifier, store_options) if not state_data or not state_data.get("id_token"): @@ -2031,7 +2247,11 @@ async def complete_unlink_user( Returns: Dictionary containing the original app state + + Raises: + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + self._reject_in_enterprise_connect("complete_unlink_user") # We can reuse the interactive login completion since the flow is similar result = await self.complete_interactive_login(url, store_options) @@ -2154,7 +2374,10 @@ async def get_access_token_for_connection( Raises: AccessTokenForConnectionError: If the access token was not found or there was an issue requesting the access token. + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + self._reject_in_enterprise_connect("get_access_token_for_connection") + state_data = await self._state_store.get(self._state_identifier, store_options) if state_data and hasattr(state_data, "dict") and callable(state_data.dict): @@ -2226,10 +2449,13 @@ async def get_token_for_connection(self, options: dict[str, Any]) -> dict[str, A Raises: AccessTokenForConnectionError: If there was an issue requesting the access token. ConfigurationError: If no client authentication is configured. + EnterpriseConnectError: If the client is configured for Enterprise Connect. Returns: Dictionary containing the token response with accessToken, expiresAt, and scope. """ + self._reject_in_enterprise_connect("get_token_for_connection") + # Constants SUBJECT_TYPE_REFRESH_TOKEN = "urn:ietf:params:oauth:token-type:refresh_token" REQUESTED_TOKEN_TYPE_FEDERATED_CONNECTION_ACCESS_TOKEN = "http://auth0.com/oauth/token-type/federated-connection-access-token" @@ -2325,7 +2551,12 @@ async def start_connect_account( Returns: The a connect URL containing a ticket to redirect the user to. + + Raises: + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + self._reject_in_enterprise_connect("start_connect_account") + # Use the default redirect_uri if none is specified redirect_uri = options.redirect_uri or self._redirect_uri # Ensure we have a redirect_uri @@ -2395,7 +2626,12 @@ async def complete_connect_account( Returns: A response from the connect account flow. + + Raises: + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + self._reject_in_enterprise_connect("complete_connect_account") + # Parse the URL to get query parameters parsed_url = urlparse(url) query_params = parse_qs(parsed_url.query) @@ -2462,7 +2698,10 @@ async def list_connected_accounts( Raises: Auth0Error: If there is an error retrieving the access token. MyAccountApiError: If the My Account API returns an error response. + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + self._reject_in_enterprise_connect("list_connected_accounts") + if take is not None and (not isinstance(take, int) or take < 1): raise InvalidArgumentError("take", "The 'take' parameter must be a positive integer.") @@ -2489,7 +2728,10 @@ async def delete_connected_account( Raises: Auth0Error: If there is an error retrieving the access token. MyAccountApiError: If the My Account API returns an error response. + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + self._reject_in_enterprise_connect("delete_connected_account") + if not connected_account_id: raise MissingRequiredArgumentError("connected_account_id") @@ -2521,7 +2763,10 @@ async def list_connected_account_connections( Raises: Auth0Error: If there is an error retrieving the access token. MyAccountApiError: If the My Account API returns an error response. + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + self._reject_in_enterprise_connect("list_connected_account_connections") + if take is not None and (not isinstance(take, int) or take < 1): raise InvalidArgumentError("take", "The 'take' parameter must be a positive integer.") @@ -2741,6 +2986,7 @@ async def login_with_custom_token_exchange( Raises: CustomTokenExchangeError: If token exchange fails ApiError: If session management fails + EnterpriseConnectError: If the client is configured for Enterprise Connect. Example: ```python @@ -2758,6 +3004,8 @@ async def login_with_custom_token_exchange( See: https://datatracker.ietf.org/doc/html/rfc8693 """ + self._reject_in_enterprise_connect("login_with_custom_token_exchange") + try: # Perform token exchange exchange_options = CustomTokenExchangeOptions( @@ -2994,7 +3242,10 @@ async def request_session_transfer_token( Raises: CustomTokenExchangeError: If no actor can be resolved or the exchange fails InvalidArgumentError: If organization is provided but blank + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + self._reject_in_enterprise_connect("request_session_transfer_token") + try: # Validate the subject up front - before any session read/refresh/network. if not subject_token or not subject_token.strip(): @@ -3072,7 +3323,10 @@ def build_session_transfer_redirect( Raises: MissingRequiredArgumentError: If target_login_url is missing or blank InvalidArgumentError: If target_login_url is not an absolute https URL, or organization is blank + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + self._reject_in_enterprise_connect("build_session_transfer_redirect") + URL.validate_https_redirect_target(target_login_url, "target_login_url") params = {"session_transfer_token": result.session_transfer_token} @@ -3090,6 +3344,7 @@ def build_session_transfer_redirect( @property def mfa(self) -> MfaClient: """Access the MFA client for multi-factor authentication operations.""" + self._reject_in_enterprise_connect("mfa") return self._mfa_client # ============================================================================ @@ -3124,7 +3379,10 @@ async def passkey_signup_challenge( Raises: PasskeyError: If the challenge request fails. + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + self._reject_in_enterprise_connect("passkey_signup_challenge") + try: domain = await self._resolve_current_domain(store_options) @@ -3197,7 +3455,10 @@ async def passkey_login_challenge( Raises: PasskeyError: If the challenge request fails. + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + self._reject_in_enterprise_connect("passkey_login_challenge") + try: domain = await self._resolve_current_domain(store_options) @@ -3285,7 +3546,10 @@ async def signin_with_passkey( OrganizationTokenValidationError: If an organization was requested but the token response included no ID token, or the ID token's org claim does not match. + EnterpriseConnectError: If the client is configured for Enterprise Connect. """ + self._reject_in_enterprise_connect("signin_with_passkey") + if not auth_session: raise MissingRequiredArgumentError("auth_session") if authn_response is None: @@ -3470,4 +3734,171 @@ async def signin_with_passkey( @property def passwordless(self) -> PasswordlessClient: """Access the passwordless client for embedded passwordless operations.""" + self._reject_in_enterprise_connect("passwordless") return self._passwordless_client + + # ============================================================================ + # Enterprise Connect (embedded login) + # ============================================================================ + + async def _is_federated_domain( + self, email_domain: str, store_options: Optional[dict[str, Any]] = None + ) -> bool: + """ + Resolve whether an email domain is Auth0-managed for enterprise SSO. + + A routing hint only, backed by WebFinger. Fails closed to False on any + error, non-200, or ambiguous response. It is never an authorization + decision - org membership is still enforced after the callback. + + Args: + email_domain: The email domain to check (case-insensitive). + store_options: Per-request store options threaded to domain resolution. + + Returns: + True only when Auth0 reports the domain as a managed OIDC issuer. + """ + if not email_domain or not email_domain.strip(): + return False + email_domain = email_domain.strip().lower() + domain = await self._resolve_current_domain(store_options) + cache_key = f"{domain}:{email_domain}" + now = time.time() + + cached = self._webfinger_cache.get(cache_key) + if cached and cached["expires_at"] > now: + return cached["value"] + + params = { + "resource": _webfinger_resource(email_domain), + "rel": WEBFINGER_ISSUER_REL, + } + try: + async with self._get_http_client(timeout=5.0) as client: + response = await client.get( + f"https://{domain}/.well-known/webfinger", params=params + ) + except httpx.HTTPError: + return False + + body = None + if response.status_code == 200: + try: + body = response.json() + except ValueError: + return False + if response.status_code == 429: + warnings.warn( + "WebFinger discovery was rate-limited; treating the domain as " + "not federated for this request.", + stacklevel=2, + ) + + is_federated, ttl = _interpret_webfinger_response(response.status_code, body) + if ttl is not None: + self._cache_webfinger_result(cache_key, is_federated, now + ttl) + return is_federated + + def _cache_webfinger_result(self, key: str, value: bool, expires_at: float) -> None: + """Store a discovery result under a bounded FIFO cache.""" + self._webfinger_cache[key] = {"value": value, "expires_at": expires_at} + self._webfinger_cache.move_to_end(key) + while len(self._webfinger_cache) > WEBFINGER_CACHE_MAX_ENTRIES: + self._webfinger_cache.popitem(last=False) + + async def start_enterprise_login( + self, + options: StartEnterpriseLoginOptions, + store_options: Optional[dict[str, Any]] = None, + ) -> Optional[str]: + """ + Begin an Enterprise Connect login from an email address. + + Runs WebFinger discovery on the email domain. When the domain is managed, + builds an authorization URL with the email as `login_hint` so Auth0 can + resolve the connection and organization. When it is not managed, returns + None so the caller can fall back to its own login. A static client-level + organization is never forwarded - Auth0 resolves it from the email. + + Args: + options: Enterprise login options carrying the user's email. + store_options: Per-request store options threaded through the flow. + + Returns: + The authorization URL to redirect to, or None when the domain is not + managed by Auth0. + + Raises: + MissingRequiredArgumentError: If no email is provided. + InvalidArgumentError: If the email is not a valid address. + """ + if options is None or not getattr(options, "email", None): + raise MissingRequiredArgumentError("email") + email = options.email.strip() + if "@" not in email or email.startswith("@") or email.endswith("@"): + raise InvalidArgumentError("email", "A valid email address is required.") + email_domain = email.rsplit("@", 1)[1].lower() + + if not await self._is_federated_domain(email_domain, store_options): + return None + + auth_params = dict(options.authorization_params or {}) + auth_params["login_hint"] = email + login_options = StartInteractiveLoginOptions( + pushed_authorization_requests=options.pushed_authorization_requests, + app_state=options.app_state, + authorization_params=auth_params, + organization=options.organization, + invitation=options.invitation, + ) + return await self.start_interactive_login(login_options, store_options) + + +async def is_federated_domain(domain: str, email_domain: str, timeout: float = 5.0) -> bool: + """ + Check whether an email domain is Auth0-managed for enterprise SSO via WebFinger. + + A stateless routing hint, not an authorization decision. Fails closed to False + on any error, non-200, or ambiguous response. Prefer `ServerClient` in normal + use, which caches results and resolves the domain per request. This standalone + form is for callers that need a one-off check without a client instance. + + Args: + domain: The Auth0 domain to query. + email_domain: The email domain to check (case-insensitive). + timeout: Per-request timeout in seconds. + + Returns: + True only when Auth0 reports the domain as a managed OIDC issuer. + """ + if not domain or not email_domain or not email_domain.strip(): + return False + email_domain = email_domain.strip().lower() + params = { + "resource": _webfinger_resource(email_domain), + "rel": WEBFINGER_ISSUER_REL, + } + try: + async with httpx.AsyncClient( + headers=Telemetry.default().headers, timeout=timeout + ) as client: + response = await client.get( + f"https://{domain}/.well-known/webfinger", params=params + ) + except httpx.HTTPError: + return False + + body = None + if response.status_code == 200: + try: + body = response.json() + except ValueError: + return False + if response.status_code == 429: + warnings.warn( + "WebFinger discovery was rate-limited; treating the domain as not " + "federated for this request.", + stacklevel=2, + ) + is_federated, _ttl = _interpret_webfinger_response(response.status_code, body) + return is_federated diff --git a/src/auth0_server_python/auth_types/__init__.py b/src/auth0_server_python/auth_types/__init__.py index 602435f..d419915 100644 --- a/src/auth0_server_python/auth_types/__init__.py +++ b/src/auth0_server_python/auth_types/__init__.py @@ -217,6 +217,15 @@ class StartInteractiveLoginOptions(BaseModel): invitation: Optional[str] = None +class StartEnterpriseLoginOptions(StartInteractiveLoginOptions): + """ + Options for starting an Enterprise Connect login. + Adds the user's email, used for WebFinger discovery and as the login hint. + """ + + email: str + + class LogoutOptions(BaseModel): """ Options for logout operations. @@ -224,6 +233,7 @@ class LogoutOptions(BaseModel): """ return_to: Optional[str] = None + federated: Optional[bool] = False class AuthorizationParameters(BaseModel): diff --git a/src/auth0_server_python/error/__init__.py b/src/auth0_server_python/error/__init__.py index ac2afef..22418d1 100644 --- a/src/auth0_server_python/error/__init__.py +++ b/src/auth0_server_python/error/__init__.py @@ -437,3 +437,32 @@ class PasskeyErrorCode: CHALLENGE_FAILED = "passkey_challenge_error" TOKEN_EXCHANGE_FAILED = "passkey_token_error" INVALID_RESPONSE = "invalid_response" + + +# ============================================================================= +# Enterprise Connect Error Classes +# ============================================================================= + + +class EnterpriseConnectError(Auth0Error): + """ + Error raised when a session or token method is called on a client + configured for Enterprise Connect, where the SDK owns no session. + + Parents Auth0Error rather than ApiError because the refusal makes no + request to Auth0, so there is no upstream error body to carry. + """ + + def __init__(self, code: str, message: str, cause=None): + super().__init__(message) + self.code = code + self.name = "EnterpriseConnectError" + self.cause = cause + + +class EnterpriseConnectErrorCode: + """Error codes for Enterprise Connect misuse guards.""" + + SESSION_UNAVAILABLE = "enterprise_connect_session_unavailable" + ACCESS_TOKEN_UNAVAILABLE = "enterprise_connect_access_token_unavailable" + METHOD_UNAVAILABLE = "enterprise_connect_method_unavailable" diff --git a/src/auth0_server_python/tests/test_error.py b/src/auth0_server_python/tests/test_error.py new file mode 100644 index 0000000..89544d6 --- /dev/null +++ b/src/auth0_server_python/tests/test_error.py @@ -0,0 +1,42 @@ +"""Tests for the error module's typed exceptions.""" + +from auth0_server_python.error import ( + Auth0Error, + EnterpriseConnectError, + EnterpriseConnectErrorCode, +) + +# === Enterprise Connect === + + +def test_enterprise_connect_error_is_auth0_error(): + err = EnterpriseConnectError( + EnterpriseConnectErrorCode.SESSION_UNAVAILABLE, "no session in enterprise connect mode" + ) + assert isinstance(err, Auth0Error) + + +def test_enterprise_connect_error_sets_code_name_and_message(): + err = EnterpriseConnectError( + EnterpriseConnectErrorCode.ACCESS_TOKEN_UNAVAILABLE, "no access token" + ) + assert err.code == EnterpriseConnectErrorCode.ACCESS_TOKEN_UNAVAILABLE + assert err.name == "EnterpriseConnectError" + assert err.message == "no access token" + assert err.cause is None + + +def test_enterprise_connect_error_preserves_cause(): + cause = ValueError("boom") + err = EnterpriseConnectError( + EnterpriseConnectErrorCode.SESSION_UNAVAILABLE, "wrapped", cause=cause + ) + assert err.cause is cause + + +def test_enterprise_connect_error_codes_are_stable(): + assert EnterpriseConnectErrorCode.SESSION_UNAVAILABLE == "enterprise_connect_session_unavailable" + assert ( + EnterpriseConnectErrorCode.ACCESS_TOKEN_UNAVAILABLE + == "enterprise_connect_access_token_unavailable" + ) diff --git a/src/auth0_server_python/tests/test_server_client.py b/src/auth0_server_python/tests/test_server_client.py index 02b8d53..9f0060e 100644 --- a/src/auth0_server_python/tests/test_server_client.py +++ b/src/auth0_server_python/tests/test_server_client.py @@ -2,6 +2,7 @@ import json import time import unicodedata +import warnings from unittest.mock import ANY, AsyncMock, MagicMock, patch from urllib.parse import parse_qs, urlparse @@ -16,7 +17,7 @@ from auth0_server_python.auth_schemes.dpop_auth import DPoPAuth from auth0_server_python.auth_server.mfa_client import MfaClient from auth0_server_python.auth_server.my_account_client import MyAccountClient -from auth0_server_python.auth_server.server_client import ServerClient +from auth0_server_python.auth_server.server_client import ServerClient, is_federated_domain from auth0_server_python.auth_types import ( CompleteConnectAccountRequest, ConnectAccountOptions, @@ -38,6 +39,7 @@ PasskeySignupChallengeResponse, PasskeyUserProfile, SessionTransferTokenResult, + StartEnterpriseLoginOptions, StartInteractiveLoginOptions, StateData, TransactionData, @@ -54,6 +56,8 @@ CustomTokenExchangeError, CustomTokenExchangeErrorCode, DomainResolverError, + EnterpriseConnectError, + EnterpriseConnectErrorCode, InvalidArgumentError, IssuerValidationError, MfaRequiredError, @@ -9677,3 +9681,396 @@ async def test_complete_interactive_login_milliseconds_ceiling_fails_open(mocker mock_state_store.set.assert_awaited_once() stored_state = mock_state_store.set.call_args.args[1] assert stored_state.internal.session_expires_at is None + + +# === Enterprise Connect === + +WEBFINGER_ISSUER_REL = "http://openid.net/specs/connect/1.0/issuer" + + +def _make_ec_client(**overrides): + kwargs = { + "domain": "auth0.local", + "client_id": "client_id", + "client_secret": "client_secret", + "secret": "some-secret", + "transaction_store": AsyncMock(), + "enterprise_connect": True, + } + kwargs.update(overrides) + return ServerClient(**kwargs) + + +def _webfinger_response(status_code, *, federated=False, bad_json=False): + response = MagicMock() + response.status_code = status_code + if bad_json: + response.json.side_effect = ValueError("no json") + else: + links = [{"rel": WEBFINGER_ISSUER_REL, "href": "https://auth0.local"}] if federated else [] + response.json.return_value = {"links": links} + return response + + +@pytest.mark.asyncio +async def test_enterprise_connect_warns_on_offline_access_scope(): + with pytest.warns(UserWarning, match="offline_access"): + _make_ec_client(authorization_params={"scope": "openid profile offline_access"}) + + +@pytest.mark.asyncio +async def test_enterprise_connect_warns_on_static_organization(): + with pytest.warns(UserWarning, match="organization"): + _make_ec_client(organization="org_static") + + +@pytest.mark.asyncio +async def test_enterprise_connect_clean_config_does_not_warn(): + with warnings.catch_warnings(): + warnings.simplefilter("error") + _make_ec_client(authorization_params={"scope": "openid profile email"}) + + +@pytest.mark.asyncio +async def test_offline_access_without_enterprise_connect_does_not_warn(): + with warnings.catch_warnings(): + warnings.simplefilter("error") + ServerClient( + domain="auth0.local", + client_id="client_id", + client_secret="client_secret", + secret="some-secret", + transaction_store=AsyncMock(), + organization="org_static", + authorization_params={"scope": "openid offline_access"}, + ) + + +@pytest.mark.asyncio +async def test_is_federated_domain_true_when_issuer_rel_present(mocker): + client = _make_ec_client() + mocker.patch( + "httpx.AsyncClient.get", + new_callable=AsyncMock, + return_value=_webfinger_response(200, federated=True), + ) + assert await client._is_federated_domain("managed.example") is True + + +@pytest.mark.asyncio +async def test_is_federated_domain_false_on_200_without_rel(mocker): + client = _make_ec_client() + mocker.patch( + "httpx.AsyncClient.get", + new_callable=AsyncMock, + return_value=_webfinger_response(200, federated=False), + ) + assert await client._is_federated_domain("managed.example") is False + + +@pytest.mark.asyncio +async def test_is_federated_domain_fails_closed_on_403(mocker): + client = _make_ec_client() + mocker.patch( + "httpx.AsyncClient.get", + new_callable=AsyncMock, + return_value=_webfinger_response(403), + ) + assert await client._is_federated_domain("managed.example") is False + + +@pytest.mark.asyncio +async def test_is_federated_domain_fails_closed_on_network_error(mocker): + client = _make_ec_client() + mocker.patch( + "httpx.AsyncClient.get", + new_callable=AsyncMock, + side_effect=httpx.ConnectError("boom"), + ) + assert await client._is_federated_domain("managed.example") is False + + +@pytest.mark.asyncio +async def test_is_federated_domain_empty_input_returns_false(mocker): + client = _make_ec_client() + get = mocker.patch("httpx.AsyncClient.get", new_callable=AsyncMock) + assert await client._is_federated_domain("") is False + get.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_is_federated_domain_caches_positive_result(mocker): + client = _make_ec_client() + get = mocker.patch( + "httpx.AsyncClient.get", + new_callable=AsyncMock, + return_value=_webfinger_response(200, federated=True), + ) + assert await client._is_federated_domain("managed.example") is True + assert await client._is_federated_domain("managed.example") is True + assert get.await_count == 1 + + +@pytest.mark.asyncio +async def test_is_federated_domain_does_not_cache_403(mocker): + client = _make_ec_client() + get = mocker.patch( + "httpx.AsyncClient.get", + new_callable=AsyncMock, + return_value=_webfinger_response(403), + ) + assert await client._is_federated_domain("managed.example") is False + assert await client._is_federated_domain("managed.example") is False + assert get.await_count == 2 + + +@pytest.mark.asyncio +async def test_standalone_is_federated_domain(mocker): + mocker.patch( + "httpx.AsyncClient.get", + new_callable=AsyncMock, + return_value=_webfinger_response(200, federated=True), + ) + assert await is_federated_domain("auth0.local", "managed.example") is True + + +@pytest.mark.asyncio +async def test_standalone_is_federated_domain_fails_closed(mocker): + mocker.patch( + "httpx.AsyncClient.get", + new_callable=AsyncMock, + side_effect=httpx.ConnectError("boom"), + ) + assert await is_federated_domain("auth0.local", "managed.example") is False + + +@pytest.mark.asyncio +async def test_start_enterprise_login_returns_none_for_unmanaged_domain(mocker): + client = _make_ec_client() + mocker.patch.object(client, "_is_federated_domain", AsyncMock(return_value=False)) + delegate = mocker.patch.object(client, "start_interactive_login", AsyncMock()) + + result = await client.start_enterprise_login( + StartEnterpriseLoginOptions(email="user@gmail.com") + ) + + assert result is None + delegate.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_start_enterprise_login_injects_login_hint_for_managed_domain(mocker): + client = _make_ec_client() + mocker.patch.object(client, "_is_federated_domain", AsyncMock(return_value=True)) + delegate = mocker.patch.object( + client, "start_interactive_login", AsyncMock(return_value="https://auth0.local/authorize?x=1") + ) + + result = await client.start_enterprise_login( + StartEnterpriseLoginOptions(email="user@managed.example") + ) + + assert result == "https://auth0.local/authorize?x=1" + login_options = delegate.await_args.args[0] + assert login_options.authorization_params["login_hint"] == "user@managed.example" + assert login_options.organization is None + + +@pytest.mark.asyncio +async def test_start_enterprise_login_missing_email_raises(): + client = _make_ec_client() + with pytest.raises(MissingRequiredArgumentError): + await client.start_enterprise_login(StartEnterpriseLoginOptions(email="")) + + +@pytest.mark.asyncio +async def test_start_enterprise_login_invalid_email_raises(mocker): + client = _make_ec_client() + mocker.patch.object(client, "_is_federated_domain", AsyncMock(return_value=True)) + with pytest.raises(InvalidArgumentError): + await client.start_enterprise_login(StartEnterpriseLoginOptions(email="not-an-email")) + + +@pytest.mark.asyncio +async def test_start_interactive_login_ignores_static_org_in_enterprise_connect(mocker): + client = _make_ec_client(organization="org_static", redirect_uri="https://app.example/cb") + client._organization = "org_static" + mock_tx = client._transaction_store + mock_tx.set = AsyncMock() + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={ + "issuer": "https://auth0.local/", + "authorization_endpoint": "https://auth0.local/authorize", + }, + ) + + options = StartInteractiveLoginOptions(authorization_params={"login_hint": "user@managed.example"}) + url = await client.start_interactive_login(options) + + query = parse_qs(urlparse(url).query) + assert "organization" not in query + + +@pytest.mark.asyncio +async def test_complete_interactive_login_enterprise_connect_returns_claims(mocker): + mock_tx_store = AsyncMock() + mock_tx_store.get.return_value = TransactionData( + code_verifier="123", + app_state={"foo": "bar"}, + domain="auth0.local", + ) + mock_state_store = AsyncMock() + + client = ServerClient( + domain="auth0.local", + client_id="client_id", + client_secret="client_secret", + transaction_store=mock_tx_store, + state_store=mock_state_store, + secret="some-secret", + enterprise_connect=True, + ) + mocker.patch.object( + client, + "_get_oidc_metadata_cached", + return_value={"issuer": "https://auth0.local/", "token_endpoint": "https://auth0.local/token"}, + ) + mocker.patch.object(client._oauth, "metadata", {"token_endpoint": "https://auth0.local/token"}) + async_fetch_token = AsyncMock( + return_value={ + "access_token": "token123", + "id_token": "raw-id-token", + "expires_in": 3600, + "scope": "openid profile", + "userinfo": {"sub": "user123", "org_id": "org_xyz"}, + } + ) + mocker.patch.object(client._oauth, "fetch_token", async_fetch_token) + + result = await client.complete_interactive_login("https://myapp.com/callback?code=abc&state=xyz") + + assert result["user"].sub == "user123" + assert result["user"].org_id == "org_xyz" + assert result["id_token"] == "raw-id-token" + assert result["domain"] == "auth0.local" + assert result["token_set"]["access_token"] == "token123" + assert result["app_state"] == {"foo": "bar"} + assert "state_data" not in result + mock_state_store.set.assert_not_awaited() + mock_tx_store.delete.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_get_session_raises_in_enterprise_connect(): + client = _make_ec_client() + with pytest.raises(EnterpriseConnectError) as exc: + await client.get_session() + assert exc.value.code == EnterpriseConnectErrorCode.SESSION_UNAVAILABLE + + +@pytest.mark.asyncio +async def test_get_access_token_raises_in_enterprise_connect(): + client = _make_ec_client() + with pytest.raises(EnterpriseConnectError) as exc: + await client.get_access_token() + assert exc.value.code == EnterpriseConnectErrorCode.ACCESS_TOKEN_UNAVAILABLE + + +_EC_BLOCKED_ASYNC_MEMBERS = [ + ("get_user", ()), + ("get_token_by_refresh_token", ({},)), + ("login_backchannel", ({},)), + ("backchannel_authentication", ({},)), + ("start_link_user", ({},)), + ("complete_link_user", ("https://app.example/callback",)), + ("start_unlink_user", ({},)), + ("complete_unlink_user", ("https://app.example/callback",)), + ("get_access_token_for_connection", ({},)), + ("get_token_for_connection", ({},)), + ("start_connect_account", (None,)), + ("complete_connect_account", ("https://app.example/callback",)), + ("list_connected_accounts", ()), + ("delete_connected_account", ("acc_1",)), + ("list_connected_account_connections", ()), + ("login_with_custom_token_exchange", (None,)), + ("request_session_transfer_token", ("subject-token", "urn:token-type")), + ("passkey_signup_challenge", ()), + ("passkey_login_challenge", ()), + ("signin_with_passkey", ("auth-session", None)), +] + + +@pytest.mark.parametrize("method_name, args", _EC_BLOCKED_ASYNC_MEMBERS) +@pytest.mark.asyncio +async def test_enterprise_connect_blocks_async_member(method_name, args): + client = _make_ec_client() + with pytest.raises(EnterpriseConnectError) as exc: + await getattr(client, method_name)(*args) + assert exc.value.code == EnterpriseConnectErrorCode.METHOD_UNAVAILABLE + + +def test_enterprise_connect_blocks_build_session_transfer_redirect(): + client = _make_ec_client() + with pytest.raises(EnterpriseConnectError) as exc: + client.build_session_transfer_redirect("https://auth0.local/authorize", None) + assert exc.value.code == EnterpriseConnectErrorCode.METHOD_UNAVAILABLE + + +@pytest.mark.parametrize("property_name", ["mfa", "passwordless"]) +def test_enterprise_connect_blocks_property_access(property_name): + client = _make_ec_client() + with pytest.raises(EnterpriseConnectError) as exc: + getattr(client, property_name) + assert exc.value.code == EnterpriseConnectErrorCode.METHOD_UNAVAILABLE + + +def test_reject_in_enterprise_connect_is_noop_without_flag(): + client = ServerClient( + domain="auth0.local", + client_id="client_id", + client_secret="client_secret", + secret="some-secret", + transaction_store=AsyncMock(), + state_store=AsyncMock(), + ) + client._reject_in_enterprise_connect("get_user") + assert client.mfa is not None + assert client.passwordless is not None + + +@pytest.mark.asyncio +async def test_handle_backchannel_logout_is_noop_in_enterprise_connect(): + client = _make_ec_client() + assert await client.handle_backchannel_logout("") is None + + +@pytest.mark.asyncio +async def test_logout_without_state_store_does_not_crash_in_enterprise_connect(): + client = _make_ec_client() + url = await client.logout(LogoutOptions(return_to="https://app.example/login")) + assert url.startswith("https://auth0.local/v2/logout") + assert "returnTo=https" in url + + +@pytest.mark.asyncio +async def test_logout_federated_appends_flag(): + client = _make_ec_client() + url = await client.logout(LogoutOptions(return_to="https://app.example/login", federated=True)) + assert "federated=true" in url + + +@pytest.mark.asyncio +async def test_logout_non_federated_omits_flag(): + mock_state_store = AsyncMock() + client = ServerClient( + domain="auth0.local", + client_id="client_id", + client_secret="client_secret", + secret="some-secret", + transaction_store=AsyncMock(), + state_store=mock_state_store, + ) + url = await client.logout(LogoutOptions(return_to="https://app.example/login")) + assert "federated" not in url diff --git a/src/auth0_server_python/utils/helpers.py b/src/auth0_server_python/utils/helpers.py index e7d51cc..feb04fd 100644 --- a/src/auth0_server_python/utils/helpers.py +++ b/src/auth0_server_python/utils/helpers.py @@ -276,7 +276,12 @@ def parse_url_params(url: str) -> dict[str, str]: return {k: v[0] if v and len(v) > 0 else '' for k, v in query_params.items()} @staticmethod - def create_logout_url(domain: str, client_id: str, return_to: Optional[str] = None) -> str: + def create_logout_url( + domain: str, + client_id: str, + return_to: Optional[str] = None, + federated: bool = False, + ) -> str: """ Create an Auth0 logout URL. @@ -284,6 +289,8 @@ def create_logout_url(domain: str, client_id: str, return_to: Optional[str] = No domain: Auth0 domain. client_id: Auth0 client ID. return_to: Optional URL to redirect to after logout. + federated: When True, add `federated` so Auth0 also ends the upstream + IdP session. Its presence is the signal; the value is ignored. Returns: The complete logout URL. @@ -292,6 +299,8 @@ def create_logout_url(domain: str, client_id: str, return_to: Optional[str] = No params = {"client_id": client_id} if return_to: params["returnTo"] = return_to + if federated: + params["federated"] = "true" return URL.build_url(base_url, params)