from __future__ import annotations from contextlib import suppress from functools import partial from json import loads from time import time from typing import Callable, Dict, List, Optional, Tuple from urllib.parse import parse_qs, urlencode, urlparse from uuid import uuid4 from ..constants import ( DEFAULT_HEADERS, EXPIRY_MARGIN, GOTRUE_URL, MAX_RETRIES, RETRY_INTERVAL, STORAGE_KEY, ) from ..errors import ( AuthApiError, AuthImplicitGrantRedirectError, AuthInvalidCredentialsError, AuthRetryableError, AuthSessionMissingError, ) from ..helpers import ( decode_jwt_payload, generate_pkce_challenge, generate_pkce_verifier, model_dump, model_dump_json, model_validate, parse_auth_otp_response, parse_auth_response, parse_link_identity_response, parse_sso_response, parse_user_response, ) from ..http_clients import SyncClient from ..timer import Timer from ..types import ( AuthChangeEvent, AuthenticatorAssuranceLevels, AuthFlowType, AuthMFAChallengeResponse, AuthMFAEnrollResponse, AuthMFAGetAuthenticatorAssuranceLevelResponse, AuthMFAListFactorsResponse, AuthMFAUnenrollResponse, AuthMFAVerifyResponse, AuthOtpResponse, AuthResponse, CodeExchangeParams, DecodedJWTDict, IdentitiesResponse, MFAChallengeAndVerifyParams, MFAChallengeParams, MFAEnrollParams, MFAUnenrollParams, MFAVerifyParams, OAuthResponse, Options, Provider, ResendCredentials, Session, SignInAnonymouslyCredentials, SignInWithIdTokenCredentials, SignInWithOAuthCredentials, SignInWithPasswordCredentials, SignInWithPasswordlessCredentials, SignInWithSSOCredentials, SignOutOptions, SignUpWithPasswordCredentials, Subscription, UserAttributes, UserIdentity, UserResponse, VerifyOtpParams, ) from .gotrue_admin_api import SyncGoTrueAdminAPI from .gotrue_base_api import SyncGoTrueBaseAPI from .gotrue_mfa_api import SyncGoTrueMFAAPI from .storage import SyncMemoryStorage, SyncSupportedStorage class SyncGoTrueClient(SyncGoTrueBaseAPI): def __init__( self, *, url: Optional[str] = None, headers: Optional[Dict[str, str]] = None, storage_key: Optional[str] = None, auto_refresh_token: bool = True, persist_session: bool = True, storage: Optional[SyncSupportedStorage] = None, http_client: Optional[SyncClient] = None, flow_type: AuthFlowType = "implicit", verify: bool = True, proxy: Optional[str] = None, ) -> None: SyncGoTrueBaseAPI.__init__( self, url=url or GOTRUE_URL, headers=headers or DEFAULT_HEADERS, http_client=http_client, verify=verify, proxy=proxy, ) self._storage_key = storage_key or STORAGE_KEY self._auto_refresh_token = auto_refresh_token self._persist_session = persist_session self._storage = storage or SyncMemoryStorage() self._in_memory_session: Optional[Session] = None self._refresh_token_timer: Optional[Timer] = None self._network_retries = 0 self._state_change_emitters: Dict[str, Subscription] = {} self._flow_type = flow_type self.admin = SyncGoTrueAdminAPI( url=self._url, headers=self._headers, http_client=self._http_client, ) self.mfa = SyncGoTrueMFAAPI() self.mfa.challenge = self._challenge self.mfa.challenge_and_verify = self._challenge_and_verify self.mfa.enroll = self._enroll self.mfa.get_authenticator_assurance_level = ( self._get_authenticator_assurance_level ) self.mfa.list_factors = self._list_factors self.mfa.unenroll = self._unenroll self.mfa.verify = self._verify # Initializations def initialize(self, *, url: Optional[str] = None) -> None: if url and self._is_implicit_grant_flow(url): self.initialize_from_url(url) else: self.initialize_from_storage() def initialize_from_storage(self) -> None: return self._recover_and_refresh() def initialize_from_url(self, url: str) -> None: try: if self._is_implicit_grant_flow(url): session, redirect_type = self._get_session_from_url(url) self._save_session(session) self._notify_all_subscribers("SIGNED_IN", session) if redirect_type == "recovery": self._notify_all_subscribers("PASSWORD_RECOVERY", session) except Exception as e: self._remove_session() raise e # Public methods def sign_in_anonymously( self, credentials: Optional[SignInAnonymouslyCredentials] = None ) -> AuthResponse: """ Creates a new anonymous user. """ self._remove_session() if credentials is None: credentials = {"options": {}} options = credentials.get("options", {}) data = options.get("data") or {} captcha_token = options.get("captcha_token") response = self._request( "POST", "signup", body={ "data": data, "gotrue_meta_security": { "captcha_token": captcha_token, }, }, xform=parse_auth_response, ) if response.session: self._save_session(response.session) self._notify_all_subscribers("SIGNED_IN", response.session) return response def sign_up( self, credentials: SignUpWithPasswordCredentials, ) -> AuthResponse: """ Creates a new user. """ self._remove_session() email = credentials.get("email") phone = credentials.get("phone") password = credentials.get("password") options = credentials.get("options", {}) redirect_to = options.get("redirect_to") or options.get("email_redirect_to") data = options.get("data") or {} channel = options.get("channel", "sms") captcha_token = options.get("captcha_token") if email: response = self._request( "POST", "signup", body={ "email": email, "password": password, "data": data, "gotrue_meta_security": { "captcha_token": captcha_token, }, }, redirect_to=redirect_to, xform=parse_auth_response, ) elif phone: response = self._request( "POST", "signup", body={ "phone": phone, "password": password, "data": data, "channel": channel, "gotrue_meta_security": { "captcha_token": captcha_token, }, }, xform=parse_auth_response, ) else: raise AuthInvalidCredentialsError( "You must provide either an email or phone number and a password" ) if response.session: self._save_session(response.session) self._notify_all_subscribers("SIGNED_IN", response.session) return response def sign_in_with_password( self, credentials: SignInWithPasswordCredentials, ) -> AuthResponse: """ Log in an existing user with an email or phone and password. """ self._remove_session() email = credentials.get("email") phone = credentials.get("phone") password = credentials.get("password") options = credentials.get("options", {}) data = options.get("data") or {} captcha_token = options.get("captcha_token") if email: response = self._request( "POST", "token", body={ "email": email, "password": password, "data": data, "gotrue_meta_security": { "captcha_token": captcha_token, }, }, query={ "grant_type": "password", }, xform=parse_auth_response, ) elif phone: response = self._request( "POST", "token", body={ "phone": phone, "password": password, "data": data, "gotrue_meta_security": { "captcha_token": captcha_token, }, }, query={ "grant_type": "password", }, xform=parse_auth_response, ) else: raise AuthInvalidCredentialsError( "You must provide either an email or phone number and a password" ) if response.session: self._save_session(response.session) self._notify_all_subscribers("SIGNED_IN", response.session) return response def sign_in_with_id_token( self, credentials: SignInWithIdTokenCredentials, ) -> AuthResponse: """ Allows signing in with an OIDC ID token. The authentication provider used should be enabled and configured. """ self._remove_session() provider = credentials.get("provider") token = credentials.get("token") access_token = credentials.get("access_token") nonce = credentials.get("nonce") options = credentials.get("options", {}) captcha_token = options.get("captcha_token") response = self._request( "POST", "token", body={ "provider": provider, "id_token": token, "access_token": access_token, "nonce": nonce, "gotrue_meta_security": { "captcha_token": captcha_token, }, }, query={ "grant_type": "id_token", }, xform=parse_auth_response, ) if response.session: self._save_session(response.session) self._notify_all_subscribers("SIGNED_IN", response.session) return response def sign_in_with_sso(self, credentials: SignInWithSSOCredentials): """ Attempts a single-sign on using an enterprise Identity Provider. A successful SSO attempt will redirect the current page to the identity provider authorization page. The redirect URL is implementation and SSO protocol specific. You can use it by providing a SSO domain. Typically you can extract this domain by asking users for their email address. If this domain is registered on the Auth instance the redirect will use that organization's currently active SSO Identity Provider for the login. If you have built an organization-specific login page, you can use the organization's SSO Identity Provider UUID directly instead. """ self._remove_session() provider_id = credentials.get("provider_id") domain = credentials.get("domain") options = credentials.get("options", {}) redirect_to = options.get("redirect_to") captcha_token = options.get("captcha_token") # HTTPX currently does not follow redirects: https://www.python-httpx.org/compatibility/ # Additionally, unlike the JS client, Python is a server side language and it's not possible # to automatically redirect in browser for hte user skip_http_redirect = options.get("skip_http_redirect", True) if domain: return self._request( "POST", "sso", body={ "domain": domain, "skip_http_redirect": skip_http_redirect, "gotrue_meta_security": { "captcha_token": captcha_token, }, }, redirect_to=redirect_to, xform=parse_sso_response, ) if provider_id: return self._request( "POST", "sso", body={ "provider_id": provider_id, "skip_http_redirect": skip_http_redirect, "gotrue_meta_security": { "captcha_token": captcha_token, }, }, redirect_to=redirect_to, xform=parse_sso_response, ) raise AuthInvalidCredentialsError( "You must provide either a domain or provider_id" ) def sign_in_with_oauth( self, credentials: SignInWithOAuthCredentials, ) -> OAuthResponse: """ Log in an existing user via a third-party provider. """ self._remove_session() provider = credentials.get("provider") options = credentials.get("options", {}) redirect_to = options.get("redirect_to") scopes = options.get("scopes") params = options.get("query_params", {}) if redirect_to: params["redirect_to"] = redirect_to if scopes: params["scopes"] = scopes url = self._get_url_for_provider(f"{self._url}/authorize", provider, params) return OAuthResponse(provider=provider, url=url) def link_identity(self, credentials: SignInWithOAuthCredentials) -> OAuthResponse: provider = credentials.get("provider") options = credentials.get("options", {}) redirect_to = options.get("redirect_to") scopes = options.get("scopes") params = options.get("query_params", {}) if redirect_to: params["redirect_to"] = redirect_to if scopes: params["scopes"] = scopes params["skip_http_redirect"] = "true" url = self._get_url_for_provider("user/identities/authorize", provider, params) session = self.get_session() if not session: raise AuthSessionMissingError() response = self._request( method="GET", path=url, jwt=session.access_token, xform=parse_link_identity_response, ) return OAuthResponse(provider=provider, url=response.url) def get_user_identities(self): response = self.get_user() return ( IdentitiesResponse(identities=response.user.identities) if response.user else AuthSessionMissingError() ) def unlink_identity(self, identity: UserIdentity): session = self.get_session() if not session: raise AuthSessionMissingError() return self._request( "DELETE", f"user/identities/{identity.identity_id}", jwt=session.access_token, ) def sign_in_with_otp( self, credentials: SignInWithPasswordlessCredentials, ) -> AuthOtpResponse: """ Log in a user using magiclink or a one-time password (OTP). If the `{{ .ConfirmationURL }}` variable is specified in the email template, a magiclink will be sent. If the `{{ .Token }}` variable is specified in the email template, an OTP will be sent. If you're using phone sign-ins, only an OTP will be sent. You won't be able to send a magiclink for phone sign-ins. """ self._remove_session() email = credentials.get("email") phone = credentials.get("phone") options = credentials.get("options", {}) email_redirect_to = options.get("email_redirect_to") should_create_user = options.get("should_create_user", True) data = options.get("data") channel = options.get("channel", "sms") captcha_token = options.get("captcha_token") if email: return self._request( "POST", "otp", body={ "email": email, "data": data, "create_user": should_create_user, "gotrue_meta_security": { "captcha_token": captcha_token, }, }, redirect_to=email_redirect_to, xform=parse_auth_otp_response, ) if phone: return self._request( "POST", "otp", body={ "phone": phone, "data": data, "create_user": should_create_user, "channel": channel, "gotrue_meta_security": { "captcha_token": captcha_token, }, }, xform=parse_auth_otp_response, ) raise AuthInvalidCredentialsError( "You must provide either an email or phone number" ) def resend( self, credentials: ResendCredentials, ) -> AuthOtpResponse: """ Resends an existing signup confirmation email, email change email, SMS OTP or phone change OTP. """ email = credentials.get("email") phone = credentials.get("phone") type = credentials.get("type") options = credentials.get("options", {}) email_redirect_to = options.get("email_redirect_to") captcha_token = options.get("captcha_token") body = { "type": type, "gotrue_meta_security": { "captcha_token": captcha_token, }, } if email is None and phone is None: raise AuthInvalidCredentialsError( "You must provide either an email or phone number" ) body.update({"email": email} if email else {"phone": phone}) return self._request( "POST", "resend", body=body, redirect_to=email_redirect_to if email else None, xform=parse_auth_otp_response, ) def verify_otp(self, params: VerifyOtpParams) -> AuthResponse: """ Log in a user given a User supplied OTP received via mobile. """ self._remove_session() response = self._request( "POST", "verify", body={ "gotrue_meta_security": { "captcha_token": params.get("options", {}).get("captcha_token"), }, **params, }, redirect_to=params.get("options", {}).get("redirect_to"), xform=parse_auth_response, ) if response.session: self._save_session(response.session) self._notify_all_subscribers("SIGNED_IN", response.session) return response def reauthenticate(self) -> AuthResponse: session = self.get_session() if not session: raise AuthSessionMissingError() return self._request( "GET", "reauthenticate", jwt=session.access_token, xform=parse_auth_response, ) def get_session(self) -> Optional[Session]: """ Returns the session, refreshing it if necessary. The session returned can be null if the session is not detected which can happen in the event a user is not signed-in or has logged out. """ current_session: Optional[Session] = None if self._persist_session: maybe_session = self._storage.get_item(self._storage_key) current_session = self._get_valid_session(maybe_session) if not current_session: self._remove_session() else: current_session = self._in_memory_session if not current_session: return None time_now = round(time()) has_expired = ( current_session.expires_at <= time_now + EXPIRY_MARGIN if current_session.expires_at else False ) return ( self._call_refresh_token(current_session.refresh_token) if has_expired else current_session ) def get_user(self, jwt: Optional[str] = None) -> Optional[UserResponse]: """ Gets the current user details if there is an existing session. Takes in an optional access token `jwt`. If no `jwt` is provided, `get_user()` will attempt to get the `jwt` from the current session. """ if not jwt: session = self.get_session() if session: jwt = session.access_token else: return None return self._request("GET", "user", jwt=jwt, xform=parse_user_response) def update_user(self, attributes: UserAttributes) -> UserResponse: """ Updates user data, if there is a logged in user. """ session = self.get_session() if not session: raise AuthSessionMissingError() response = self._request( "PUT", "user", body=attributes, jwt=session.access_token, xform=parse_user_response, ) session.user = response.user self._save_session(session) self._notify_all_subscribers("USER_UPDATED", session) return response def set_session(self, access_token: str, refresh_token: str) -> AuthResponse: """ Sets the session data from the current session. If the current session is expired, `set_session` will take care of refreshing it to obtain a new session. If the refresh token in the current session is invalid and the current session has expired, an error will be thrown. If the current session does not contain at `expires_at` field, `set_session` will use the exp claim defined in the access token. The current session that minimally contains an access token, refresh token and a user. """ time_now = round(time()) expires_at = time_now has_expired = True session: Optional[Session] = None if access_token and access_token.split(".")[1]: payload = self._decode_jwt(access_token) exp = payload.get("exp") if exp: expires_at = int(exp) has_expired = expires_at <= time_now if has_expired: if not refresh_token: raise AuthSessionMissingError() response = self._refresh_access_token(refresh_token) if not response.session: return AuthResponse() session = response.session else: response = self.get_user(access_token) session = Session( access_token=access_token, refresh_token=refresh_token, user=response.user, token_type="bearer", expires_in=expires_at - time_now, expires_at=expires_at, ) self._save_session(session) self._notify_all_subscribers("TOKEN_REFRESHED", session) return AuthResponse(session=session, user=response.user) def refresh_session(self, refresh_token: Optional[str] = None) -> AuthResponse: """ Returns a new session, regardless of expiry status. Takes in an optional current session. If not passed in, then refreshSession() will attempt to retrieve it from getSession(). If the current session's refresh token is invalid, an error will be thrown. """ if not refresh_token: session = self.get_session() if session: refresh_token = session.refresh_token if not refresh_token: raise AuthSessionMissingError() session = self._call_refresh_token(refresh_token) return AuthResponse(session=session, user=session.user) def sign_out(self, options: SignOutOptions = {"scope": "global"}) -> None: """ Inside a browser context, `sign_out` will remove the logged in user from the browser session and log them out - removing all items from localstorage and then trigger a `"SIGNED_OUT"` event. For server-side management, you can revoke all refresh tokens for a user by passing a user's JWT through to `api.sign_out`. There is no way to revoke a user's access token jwt until it expires. It is recommended to set a shorter expiry on the jwt for this reason. """ with suppress(AuthApiError): session = self.get_session() access_token = session.access_token if session else None if access_token: self.admin.sign_out(access_token, options["scope"]) if options["scope"] != "others": self._remove_session() self._notify_all_subscribers("SIGNED_OUT", None) def on_auth_state_change( self, callback: Callable[[AuthChangeEvent, Optional[Session]], None], ) -> Subscription: """ Receive a notification every time an auth event happens. """ unique_id = str(uuid4()) def _unsubscribe() -> None: self._state_change_emitters.pop(unique_id) subscription = Subscription( id=unique_id, callback=callback, unsubscribe=_unsubscribe, ) self._state_change_emitters[unique_id] = subscription return subscription def reset_password_for_email(self, email: str, options: Options = {}) -> None: """ Sends a password reset request to an email address. """ self._request( "POST", "recover", body={ "email": email, "gotrue_meta_security": { "captcha_token": options.get("captcha_token"), }, }, redirect_to=options.get("redirect_to"), ) def reset_password_email( self, email: str, options: Options = {}, ) -> None: """ Sends a password reset request to an email address. """ self.reset_password_for_email(email, options) # MFA methods def _enroll(self, params: MFAEnrollParams) -> AuthMFAEnrollResponse: session = self.get_session() if not session: raise AuthSessionMissingError() body = { "friendly_name": params["friendly_name"], "factor_type": params["factor_type"], } if params["factor_type"] == "phone": body["phone"] = params["phone"] else: body["issuer"] = params["issuer"] response = self._request( "POST", "factors", body=body, jwt=session.access_token, xform=partial(model_validate, AuthMFAEnrollResponse), ) if params["factor_type"] == "totp" and response.totp.qr_code: response.totp.qr_code = f"data:image/svg+xml;utf-8,{response.totp.qr_code}" return response def _challenge(self, params: MFAChallengeParams) -> AuthMFAChallengeResponse: session = self.get_session() if not session: raise AuthSessionMissingError() return self._request( "POST", f"factors/{params.get('factor_id')}/challenge", body={"channel": params.get("channel")}, jwt=session.access_token, xform=partial(model_validate, AuthMFAChallengeResponse), ) def _challenge_and_verify( self, params: MFAChallengeAndVerifyParams, ) -> AuthMFAVerifyResponse: response = self._challenge( { "factor_id": params.get("factor_id"), } ) return self._verify( { "factor_id": params.get("factor_id"), "challenge_id": response.id, "code": params.get("code"), } ) def _verify(self, params: MFAVerifyParams) -> AuthMFAVerifyResponse: session = self.get_session() if not session: raise AuthSessionMissingError() response = self._request( "POST", f"factors/{params.get('factor_id')}/verify", body=params, jwt=session.access_token, xform=partial(model_validate, AuthMFAVerifyResponse), ) session = model_validate(Session, model_dump(response)) self._save_session(session) self._notify_all_subscribers("MFA_CHALLENGE_VERIFIED", session) return response def _unenroll(self, params: MFAUnenrollParams) -> AuthMFAUnenrollResponse: session = self.get_session() if not session: raise AuthSessionMissingError() return self._request( "DELETE", f"factors/{params.get('factor_id')}", jwt=session.access_token, xform=partial(AuthMFAUnenrollResponse, model_validate), ) def _list_factors(self) -> AuthMFAListFactorsResponse: response = self.get_user() all = response.user.factors or [] totp = [f for f in all if f.factor_type == "totp" and f.status == "verified"] phone = [f for f in all if f.factor_type == "phone" and f.status == "verified"] return AuthMFAListFactorsResponse(all=all, totp=totp, phone=phone) def _get_authenticator_assurance_level( self, ) -> AuthMFAGetAuthenticatorAssuranceLevelResponse: session = self.get_session() if not session: return AuthMFAGetAuthenticatorAssuranceLevelResponse( current_level=None, next_level=None, current_authentication_methods=[], ) payload = self._decode_jwt(session.access_token) current_level: Optional[AuthenticatorAssuranceLevels] = None if payload.get("aal"): current_level = payload.get("aal") verified_factors = [ f for f in session.user.factors or [] if f.status == "verified" ] next_level = "aal2" if verified_factors else current_level current_authentication_methods = payload.get("amr") or [] return AuthMFAGetAuthenticatorAssuranceLevelResponse( current_level=current_level, next_level=next_level, current_authentication_methods=current_authentication_methods, ) # Private methods def _remove_session(self) -> None: if self._persist_session: self._storage.remove_item(self._storage_key) else: self._in_memory_session = None if self._refresh_token_timer: self._refresh_token_timer.cancel() self._refresh_token_timer = None def _get_session_from_url( self, url: str, ) -> Tuple[Session, Optional[str]]: if not self._is_implicit_grant_flow(url): raise AuthImplicitGrantRedirectError("Not a valid implicit grant flow url.") result = urlparse(url) params = parse_qs(result.query) error_description = self._get_param(params, "error_description") if error_description: error_code = self._get_param(params, "error_code") error = self._get_param(params, "error") if not error_code: raise AuthImplicitGrantRedirectError("No error_code detected.") if not error: raise AuthImplicitGrantRedirectError("No error detected.") raise AuthImplicitGrantRedirectError( error_description, {"code": error_code, "error": error}, ) provider_token = self._get_param(params, "provider_token") provider_refresh_token = self._get_param(params, "provider_refresh_token") access_token = self._get_param(params, "access_token") if not access_token: raise AuthImplicitGrantRedirectError("No access_token detected.") expires_in = self._get_param(params, "expires_in") if not expires_in: raise AuthImplicitGrantRedirectError("No expires_in detected.") refresh_token = self._get_param(params, "refresh_token") if not refresh_token: raise AuthImplicitGrantRedirectError("No refresh_token detected.") token_type = self._get_param(params, "token_type") if not token_type: raise AuthImplicitGrantRedirectError("No token_type detected.") time_now = round(time()) expires_at = time_now + int(expires_in) user = self.get_user(access_token) session = Session( provider_token=provider_token, provider_refresh_token=provider_refresh_token, access_token=access_token, expires_in=int(expires_in), expires_at=expires_at, refresh_token=refresh_token, token_type=token_type, user=user.user, ) redirect_type = self._get_param(params, "type") return session, redirect_type def _recover_and_refresh(self) -> None: raw_session = self._storage.get_item(self._storage_key) current_session = self._get_valid_session(raw_session) if not current_session: if raw_session: self._remove_session() return time_now = round(time()) expires_at = current_session.expires_at if expires_at and expires_at < time_now + EXPIRY_MARGIN: refresh_token = current_session.refresh_token if self._auto_refresh_token and refresh_token: self._network_retries += 1 try: self._call_refresh_token(refresh_token) self._network_retries = 0 except Exception as e: if ( isinstance(e, AuthRetryableError) and self._network_retries < MAX_RETRIES ): if self._refresh_token_timer: self._refresh_token_timer.cancel() self._refresh_token_timer = Timer( (RETRY_INTERVAL ** (self._network_retries * 100)), self._recover_and_refresh, ) self._refresh_token_timer.start() return self._remove_session() return if self._persist_session: self._save_session(current_session) self._notify_all_subscribers("SIGNED_IN", current_session) def _call_refresh_token(self, refresh_token: str) -> Session: if not refresh_token: raise AuthSessionMissingError() response = self._refresh_access_token(refresh_token) if not response.session: raise AuthSessionMissingError() self._save_session(response.session) self._notify_all_subscribers("TOKEN_REFRESHED", response.session) return response.session def _refresh_access_token(self, refresh_token: str) -> AuthResponse: return self._request( "POST", "token", query={"grant_type": "refresh_token"}, body={"refresh_token": refresh_token}, xform=parse_auth_response, ) def _save_session(self, session: Session) -> None: if not self._persist_session: self._in_memory_session = session expire_at = session.expires_at if expire_at: time_now = round(time()) expire_in = expire_at - time_now refresh_duration_before_expires = ( EXPIRY_MARGIN if expire_in > EXPIRY_MARGIN else 0.5 ) value = (expire_in - refresh_duration_before_expires) * 1000 self._start_auto_refresh_token(value) if self._persist_session and session.expires_at: self._storage.set_item(self._storage_key, model_dump_json(session)) def _start_auto_refresh_token(self, value: float) -> None: if self._refresh_token_timer: self._refresh_token_timer.cancel() self._refresh_token_timer = None if value <= 0 or not self._auto_refresh_token: return def refresh_token_function(): self._network_retries += 1 try: session = self.get_session() if session: self._call_refresh_token(session.refresh_token) self._network_retries = 0 except Exception as e: if ( isinstance(e, AuthRetryableError) and self._network_retries < MAX_RETRIES ): self._start_auto_refresh_token( RETRY_INTERVAL ** (self._network_retries * 100) ) self._refresh_token_timer = Timer(value, refresh_token_function) self._refresh_token_timer.start() def _notify_all_subscribers( self, event: AuthChangeEvent, session: Optional[Session], ) -> None: for subscription in self._state_change_emitters.values(): subscription.callback(event, session) def _get_valid_session( self, raw_session: Optional[str], ) -> Optional[Session]: if not raw_session: return None data = loads(raw_session) if not data: return None if not data.get("access_token"): return None if not data.get("refresh_token"): return None if not data.get("expires_at"): return None try: expires_at = int(data["expires_at"]) data["expires_at"] = expires_at except ValueError: return None try: return model_validate(Session, data) except Exception: return None def _get_param( self, query_params: Dict[str, List[str]], name: str, ) -> Optional[str]: return query_params[name][0] if name in query_params else None def _is_implicit_grant_flow(self, url: str) -> bool: result = urlparse(url) params = parse_qs(result.query) return "access_token" in params or "error_description" in params def _get_url_for_provider( self, url: str, provider: Provider, params: Dict[str, str], ) -> str: if self._flow_type == "pkce": code_verifier = generate_pkce_verifier() code_challenge = generate_pkce_challenge(code_verifier) self._storage.set_item(f"{self._storage_key}-code-verifier", code_verifier) code_challenge_method = ( "plain" if code_verifier == code_challenge else "s256" ) params["code_challenge"] = code_challenge params["code_challenge_method"] = code_challenge_method params["provider"] = provider query = urlencode(params) return f"{url}?{query}" def _decode_jwt(self, jwt: str) -> DecodedJWTDict: """ Decodes a JWT (without performing any validation). """ return decode_jwt_payload(jwt) def exchange_code_for_session(self, params: CodeExchangeParams): code_verifier = params.get("code_verifier") or self._storage.get_item( f"{self._storage_key}-code-verifier" ) response = self._request( "POST", "token?grant_type=pkce", body={ "auth_code": params.get("auth_code"), "code_verifier": code_verifier, }, redirect_to=params.get("redirect_to"), xform=parse_auth_response, ) self._storage.remove_item(f"{self._storage_key}-code-verifier") if response.session: self._save_session(response.session) self._notify_all_subscribers("SIGNED_IN", response.session) return response