""" Copyright (c) 2023 Proton AG This file is part of Proton. Proton is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. Proton is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with ProtonVPN. If not, see . """ from __future__ import annotations from dataclasses import dataclass from pathlib import Path from typing import * import logging from .exceptions import ProtonCryptoError, ProtonAPIError, ProtonAPIAuthenticationNeeded, ProtonAPI2FANeeded, ProtonAPIMissingScopeError, ProtonAPIHumanVerificationNeeded from .srp import User as PmsrpUser from .environments import Environment from ..loader import Loader import asyncio import base64 import random from ..utils import ExecutionEnvironment logger = logging.getLogger(__name__) SRP_MODULUS_KEY = """-----BEGIN PGP PUBLIC KEY BLOCK----- xjMEXAHLgxYJKwYBBAHaRw8BAQdAFurWXXwjTemqjD7CXjXVyKf0of7n9Ctm L8v9enkzggHNEnByb3RvbkBzcnAubW9kdWx1c8J3BBAWCgApBQJcAcuDBgsJ BwgDAgkQNQWFxOlRjyYEFQgKAgMWAgECGQECGwMCHgEAAPGRAP9sauJsW12U MnTQUZpsbJb53d0Wv55mZIIiJL2XulpWPQD/V6NglBd96lZKBmInSXX/kXat Sv+y0io+LR8i2+jV+AbOOARcAcuDEgorBgEEAZdVAQUBAQdAeJHUz1c9+KfE kSIgcBRE3WuXC4oj5a2/U3oASExGDW4DAQgHwmEEGBYIABMFAlwBy4MJEDUF hcTpUY8mAhsMAAD/XQD8DxNI6E78meodQI+wLsrKLeHn32iLvUqJbVDhfWSU WO4BAMcm1u02t4VKw++ttECPt+HUgPUq5pqQWe5Q2cW4TMsE =Y4Mw -----END PGP PUBLIC KEY BLOCK-----""" SRP_MODULUS_KEY_FINGERPRINT = "248097092b458509c508dac0350585c4e9518f26" # nosemgrep: gitleak-detect-ignore,generic.secrets.gitleaks.generic-api-key.generic-api-key # pylint: disable=line-too-long @dataclass class Fido2AssertionParameters: """ Parameters needed to create a FIDO2 assertion for 2FA. These can be obtained from :attr:`Session.supports_fido2`. """ challenge: bytes rp_id: str allow_credentials: list[dict] user_verification: str @dataclass class Fido2Assertion: """ FIDO2 assertion obtained from the security key. These are passed to :meth:`Session.async_validate_2fa_fido2`. """ client_data: bytes authenticator_data: bytes signature: bytes credential_id: bytes def sync_wrapper(f): def wrapped_f(*a, **kw): try: loop = asyncio.get_running_loop() newloop = False except RuntimeError: newloop = True if not newloop: raise RuntimeError("It's forbidden to call sync_wrapped functions from an async one, please await directly the async one") loop = asyncio.new_event_loop() try: return loop.run_until_complete(f(*a, **kw)) finally: loop.close() wrapped_f.__doc__ = f"Synchronous wrapper for :meth:`{f.__name__}`" return wrapped_f class Session: def __init__(self, appversion : str = "Other", user_agent:str="None"): """Get a session towards the Proton API. :param appversion: version for the new Session object, defaults to ``"Other"`` :type appversion: str, optional :param user_agent: user agent to use, defaults to ``"None"``. It should be of the following syntax: * Linux based -> ``ClientName/client.version (Linux; Distro/distro_version)`` * Non-linux based -> ``ClientName/client.version (OS)`` :type user_agent: str, optional """ self.__appversion = appversion self.__user_agent = user_agent self.__UID = None self.__AccessToken = None self.__RefreshToken = None self.__Scopes = None self.__AccountName = None #Extra data that we want to persist (used if we load a session from a subclass) self.__extrastate = {} # Temporary storage for 2FA object self.__2FA = None #Refresh revision (incremented each time a refresh is done) #This allows knowing if a refresh should be done or if it is already in progress self.__refresh_revision = 0 #Lazy initialized by modulus decryption self.__gnupg_for_modulus = None #Lazy initialized by api request self.__transport = None self.__transport_factory = None self.transport_factory = None #Lazy initialized by request lock/unlock self.__can_run_requests = None #Lazy initialized by environment: self.__environment = None self.__persistence_observers = [] async def async_api_request(self, endpoint, jsondata=None, data=None, additional_headers=None, method=None, params=None, no_condition_check=False, return_raw=False): """Do an API request. This call can return any of the exceptions defined in :mod:`proton.session.exceptions`. :param endpoint: API endpoint :type endpoint: str :param jsondata: JSON serializable dict to send as request data :type jsondata: dict :param data: data to be sent as either `multipart/form-data` or `application/x-www-form-urlencoded`. `multipart/form-data` is used when required, for example if data includes fields with a file-like value (i.e. is an instance of io.IOBase). :type data: FormData :param additional_headers: additional headers to send :type additional_headers: dict :param method: HTTP method (get|post|put|delete|patch) :type method: str :param params: URL parameters to append to the URL. If a dictionary or list of tuples ``[(key, value)]`` is provided, form-encoding will take place. :type params: str, dict or iterable :param no_condition_check: Internal flag to disable locking, defaults to False :type no_condition_check: bool, optional :param return_raw: If set to true, the function will return a :class:`RawResponse` object instead of a dict, this class contains the header and status information along with the json response. :type return_raw: bool, optional :return: Deserialized JSON reply or a RawResponse object if return_raw is True :rtype: dict | RawResponse """ # We might need to loop attempts = 3 stored_exception = None # in case of too many attempts, raise instead of returning None while attempts > 0: attempts -= 1 try: refresh_revision_at_start = self.__refresh_revision return await self.__async_api_request_internal(endpoint, jsondata, data, additional_headers, method, params, no_condition_check, return_raw=return_raw) except ProtonAPIError as e: stored_exception = e # We have a missing scope. if e.http_code == 403: # If we need a 2FA authentication, then ask for it by sending a specific exception. if self.needs_twofa: raise ProtonAPI2FANeeded.from_proton_api_error(e) else: # Otherwise, just throw the 403 raise ProtonAPIMissingScopeError.from_proton_api_error(e) #401: token expired elif e.http_code == 401: #If we can refresh, than do it and retry if await self.async_refresh(only_when_refresh_revision_is=refresh_revision_at_start, no_condition_check=no_condition_check): continue #Else, fail :-( else: raise ProtonAPIAuthenticationNeeded.from_proton_api_error(e) #422 + 9001: Human verification needed elif e.http_code == 422 and e.body_code == 9001: raise ProtonAPIHumanVerificationNeeded.from_proton_api_error(e) #Invalid human verification token elif e.body_code == 12087: raise ProtonAPIHumanVerificationNeeded.from_proton_api_error(e) #These are codes which require and immediate retry elif e.http_code in (408, 502): continue #These not, let's retry more gracefully elif e.http_code in (429, 503): await self.__sleep_for_exception(e) continue #Something else, throw raise raise stored_exception # if we have reached that point without returning any value, an exception should have been stored async def async_authenticate(self, username: str, password: str, client_secret: str = None, no_condition_check: bool = False, additional_headers=None) -> bool: """Authenticate against Proton API :param username: Proton account username :type username: str :param password: Proton account password :type password: str :param client_secret: Client Secret for SRP :type client_secret: str, optional :param no_condition_check: Internal flag to disable locking, defaults to False :type no_condition_check: bool, optional :param additional_headers: additional headers to send :type additional_headers: dict :return: True if authentication succeeded, False otherwise. :rtype: bool """ self._requests_lock(no_condition_check) await self.async_logout(no_condition_check=True) try: req_data = {"Username": username} if client_secret is not None: req_data["ClientSecret"] = client_secret info_response = await self.__async_api_request_internal("/auth/info", req_data, no_condition_check=True, additional_headers=additional_headers) modulus = self._verify_modulus(info_response['Modulus']) server_challenge = base64.b64decode(info_response["ServerEphemeral"]) salt = base64.b64decode(info_response["Salt"]) version = info_response["Version"] usr = PmsrpUser(password, modulus) client_challenge = usr.get_challenge() client_proof = usr.process_challenge(salt, server_challenge, version) if client_proof is None: raise ProtonCryptoError('Invalid challenge') # Send response payload = { "Username": username, "ClientEphemeral": base64.b64encode(client_challenge).decode( 'utf8' ), "ClientProof": base64.b64encode(client_proof).decode('utf8'), "SRPSession": info_response["SRPSession"], } if client_secret is not None: payload["ClientSecret"] = client_secret try: auth_response = await self.__async_api_request_internal("/auth", payload, no_condition_check=True, additional_headers=additional_headers) except ProtonAPIError as e: if e.body_code == 8002: return False raise if "ServerProof" not in auth_response: return False usr.verify_session(base64.b64decode(auth_response["ServerProof"])) if not usr.authenticated(): raise ProtonCryptoError('Invalid server proof') self.__UID = auth_response['UID'] self.__AccessToken = auth_response['AccessToken'] self.__RefreshToken = auth_response['RefreshToken'] self.__Scopes = auth_response["Scopes"] self.__AccountName = username if '2FA' in auth_response: self.__2FA = auth_response['2FA'] else: self.__2FA = None return True finally: self._requests_unlock(no_condition_check) async def async_provide_2fa(self, code : str, no_condition_check=False, additional_headers=None) -> bool: """Provide Two Factor Authentication Code to the API. This method is deprecated, please use :meth:`async_validate_2fa_code` or :meth:`async_validate_2fa_fido2` instead. :param code: 2FA code :type code: str :param no_condition_check: Internal flag to disable locking, defaults to False :type no_condition_check: bool, optional :return: True if 2FA succeeded, False otherwise. :rtype: bool :raises ProtonAPIAuthenticationNeeded: if 2FA failed, and the session was reset by the API backend (this is normally the case) """ logger.warning("async_provide_2fa is deprecated, please use async_validate_2fa_code or async_validate_2fa_fido2 instead") return await self.async_validate_2fa_code(code, no_condition_check, additional_headers) async def _async_validate_2fa(self, answer: dict, no_condition_check=False, additional_headers=None) -> bool: """ Internal function to validate 2FA, either via code or FIDO2. Do not use this method directly, use :meth:`async_validate_2fa_code` or :meth:`async_validate_2fa_fido2` instead. :param answer: dict containing either the TwoFactorCode or FIDO2 information :type answer: dict :param no_condition_check: Internal flag to disable locking, defaults to False :type no_condition_check: bool, optional :param additional_headers: additional headers to send :type additional_headers: dict, optional :return: True if 2FA succeeded, False otherwise. """ self._requests_lock(no_condition_check) try: ret = await self.__async_api_request_internal( '/auth/2fa', answer, no_condition_check=True, additional_headers=additional_headers) self.__Scopes = ret['Scopes'] if ret.get('Code') == 1000: self.__2FA = None return True return False except ProtonAPIError as e: if e.body_code == 8002: # 2FA jail, we need to start over (beware, we might hit login jails too) #Needs re-login self._clear_local_data() raise ProtonAPIAuthenticationNeeded.from_proton_api_error(e) if e.http_code == 401: return False raise finally: self._requests_unlock(no_condition_check) async def async_validate_2fa_code(self, code: str, no_condition_check=False, additional_headers=None) -> bool: """ Validate a 2FA code against the API. :param code: 2FA code :type code: str :param no_condition_check: Internal flag to disable locking, defaults to False :type no_condition_check: bool, optional :param additional_headers: additional headers to send :type additional_headers: dict, optional :return: True if 2FA succeeded, False otherwise. :rtype: bool """ return await self._async_validate_2fa( {"TwoFactorCode": code}, no_condition_check, additional_headers) async def async_validate_2fa_fido2(self, assertion: Fido2Assertion, no_condition_check=False, additional_headers=None) -> bool: """ Validate a FIDO2 assertion against the API. :param assertion: FIDO2 assertion obtained from the security key :type assertion: Fido2Assertion :param no_condition_check: Internal flag to disable locking, defaults to False :type no_condition_check: bool, optional :param additional_headers: additional headers to send :type additional_headers: dict, optional :return: True if 2FA succeeded, False otherwise. :rtype: bool """ fido2 = self.__2FA["FIDO2"] def to_b64(x): return base64.b64encode(x).decode('ascii') answer = { "AuthenticationOptions": fido2["AuthenticationOptions"], "ClientData": to_b64(assertion.client_data), "AuthenticatorData": to_b64(assertion.authenticator_data), "Signature": to_b64(assertion.signature), "CredentialID": list(assertion.credential_id), } return await self._async_validate_2fa( {"FIDO2": answer}, no_condition_check, additional_headers ) async def async_refresh(self, only_when_refresh_revision_is=None, no_condition_check=False, additional_headers=None): """Refresh tokens. Refresh AccessToken with a valid RefreshToken. If the RefreshToken is invalid then the user will have to re-authenticate. :return: True if refresh succeeded, False otherwise (doesn't throw an exception) :rtype: bool """ #If we have the correct revision, and it doesn't match, then just exit if only_when_refresh_revision_is is not None and only_when_refresh_revision_is != self.__refresh_revision: # If we have the wrong revision, then this indicates that we have two refresh running in parallel. # Thanksfully, we can simply wait for the other to complete and return successfully. await self._requests_wait(no_condition_check) return True self._requests_lock(no_condition_check) #Increment the refresh revision counter, so we don't refresh multiple times self.__refresh_revision += 1 attempts = 3 try: while attempts > 0: attempts -= 1 try: refresh_response = await self.__async_api_request_internal('/auth/refresh', { "ResponseType": "token", "GrantType": "refresh_token", "RefreshToken": self.__RefreshToken, "RedirectURI": "http://protonmail.ch" }, no_condition_check=True, additional_headers=additional_headers) self.__AccessToken = refresh_response["AccessToken"] self.__RefreshToken = refresh_response["RefreshToken"] self.__Scopes = refresh_response["Scopes"] return True except ProtonAPIError as e: #https://confluence.protontech.ch/display/API/Authentication%2C+sessions%2C+and+tokens#Authentication,sessions,andtokens-RefreshingSessions if e.http_code == 409: #409 Conflict - Indicates a race condition on the DB, and the request should be performed again continue #We're probably jailed, just retry later elif e.http_code in (429, 503): await self.__sleep_for_exception(e) continue elif e.http_code in (400, 422): #Needs re-login self._clear_local_data() return False return False finally: self._requests_unlock(no_condition_check) async def async_logout(self, no_condition_check=False, additional_headers=None): """Logout from API. :return: True if logout was successful (or nothing was done) :rtype: bool """ self._requests_lock(no_condition_check) previous_account_name = self.AccountName try: # No-op if not authenticated (but we do this inside the lock, so data is persisted nevertheless) if not self.authenticated: self._clear_local_data() return True ret = await self.__async_api_request_internal('/auth', method='DELETE', no_condition_check=True, additional_headers=additional_headers) # Erase any information we have about the session self._clear_local_data() return True except ProtonAPIError as e: # If we get a 401, then we should erase data (session doesn't exist on the server), and we're fine if e.http_code == 401: self._clear_local_data() return True # We don't know what is going on, throw raise finally: self._requests_unlock(no_condition_check, previous_account_name) async def async_lock(self, no_condition_check=False, additional_headers=None): """ Lock the current user (remove PASSWORD and LOCKED scopes)""" self._requests_lock(no_condition_check) try: ret = await self.__async_api_request_internal('/users/lock', method='PUT', no_condition_check=True, additional_headers=additional_headers) ret = await self.__async_api_request_internal('/auth/scopes', no_condition_check=True, additional_headers=additional_headers) self.__Scopes = ret['Scopes'] return True finally: self._requests_unlock(no_condition_check) #FIXME: clear user keys #FIXME: implement unlock async def async_human_verif_request_code(self, address=None, phone=None, additional_headers=None): """Request a verification code. Either address (email address) or phone (phone number) should be specified.""" # We use email validation by default if both are provided, # but it's not super clean if the dev doesn't know about it assert address is not None ^ phone is not None # nosec B101 # nosemgrep: gitlab.bandit.B101 if address is not None: data = {'Type': 'email', 'Destination': {'Address': address}} elif phone is not None: data = {'Type': 'sms', 'Destination': {'Phone': phone}} return await self.async_api_request('/users/code', data, additional_headers=additional_headers).get('Code', 0) == 1000 async def async_human_verif_provide_token(self, method, token): pass async def async_fork(self, child_client_id: str, payload: str=None, independent: int=0, user_code: int=None, no_condition_check=False) -> str: """Try to fork the current session with parameters and return the fork selector if successful. :param child_client_id: The clientID that the client creating the child session will use. :type child_client_id: str :param payload: Will be downloaded by the child client, and can contain encrypted sensitive information. :type payload: str, optional :param independent: set to 1 if the newly forked session has to be independent, else 0. :type independent: int, optional :param user_code: If provided, the selector will not be randomly chosen, but rather the one already returned will be used. :type user_code: int, optional :param no_condition_check: Internal flag to disable locking, defaults to False :type no_condition_check: bool, optional :return selector: Fork selector to be used by method `async_import_fork()` :rtype: str """ data = {'ChildClientID' : child_client_id, 'Independent': independent} if payload is not None: data['Payload'] = payload if user_code is not None: data['UserCode'] = user_code ret = await self.async_api_request('/auth/v4/sessions/forks', data, method='POST', no_condition_check=no_condition_check) return ret['Selector'] async def async_import_fork(self, selector: str, no_condition_check=False) -> str: """Try to import the session fork specified by the selector. :param selector: Obtained via a successful call to `async_fork()` :type selector: str :param no_condition_check: Internal flag to disable locking, defaults to False :type no_condition_check: bool, optional :return payload: The payload sent by the parent session when initiating the fork. :rtype: str """ self._requests_lock(no_condition_check) try: ret = await self.async_api_request(f"/auth/v4/sessions/forks/{selector}", method='GET', no_condition_check=True) self.__UID = ret['UID'] self.__RefreshToken = ret['RefreshToken'] self.__AccessToken = ret['AccessToken'] self.__Scopes = ret['Scopes'] return ret['Payload'] finally: self._requests_unlock(no_condition_check) # Wrappers to provide non-asyncio API api_request = sync_wrapper(async_api_request) authenticate = sync_wrapper(async_authenticate) provide_2fa = sync_wrapper(async_provide_2fa) logout = sync_wrapper(async_logout) refresh = sync_wrapper(async_refresh) lock = sync_wrapper(async_lock) human_verif_request_code = sync_wrapper(async_human_verif_request_code) human_verif_provide_token = sync_wrapper(async_human_verif_provide_token) fork = sync_wrapper(async_fork) import_fork = sync_wrapper(async_import_fork) validate_2fa_code = sync_wrapper(async_validate_2fa_code) validate_2fa_fido2 = sync_wrapper(async_validate_2fa_fido2) def register_persistence_observer(self, observer: object): """Register an observer that will be notified of any persistent state change of the session :param observer: Observer to register. It has to provide the following interface (see :class:`proton.sso.ProtonSSO` for an actual implementation): * ``_acquire_session_lock(account_name : str, session_data : dict)`` * ``_release_session_lock(account_name : str, new_session_data : dict)`` :type observer: object """ self.__persistence_observers.append(observer) def _clear_local_data(self) -> None: """Clear locally cache data for logout (or equivalently, when the session is "lost").""" self.__UID = None self.__AccessToken = None self.__RefreshToken = None self.__Scopes = None self.__2FA = None self.__extrastate = {} @property def transport_factory(self): """Set/read the factory used for transports (i.e. how to reach the API). If the property is set to a class, it will be wrapped in a factory. If the property is set to None, then the default ``transport`` will be obtained from :class:`.Loader`. """ return self.__transport_factory @transport_factory.setter def transport_factory(self, new_transport_factory): from .transports import TransportFactory from ..loader import Loader self.__transport = None # If we don't set a new transport factory, then let's create a default one if new_transport_factory is None: default_transport = Loader.get('transport') self.__transport_factory = TransportFactory(default_transport) elif isinstance(new_transport_factory, TransportFactory): self.__transport_factory = new_transport_factory else: self.__transport_factory = TransportFactory(new_transport_factory) @property def appversion(self) -> str: """:return: The appversion defined at construction (used for creating requests by transports) :rtype: str""" return self.__appversion @property def user_agent(self) -> str: """:return: The user_agent defined at construction (used for creating requests by transports) :rtype: str""" return self.__user_agent @property def authenticated(self) -> bool: """:return: True if session is authenticated, False otherwise. :rtype: bool """ return self.__UID is not None @property def UID(self) -> Optional[str]: """:return: the session UID, None if not authenticated :rtype: str, optional """ return self.__UID @property def Scopes(self) -> Optional[list[str]]: """:return: list of scopes of the current session, None if unknown or not defined. :rtype: list[str], optional """ return self.__Scopes @property def AccountName(self) -> str: """:return: session account name (mostly used for SSO) :rtype: str """ return self.__AccountName @property def AccessToken(self) -> str: """:return: return the access token for API calls (used by transports) :rtype: str """ return self.__AccessToken @property def needs_twofa(self) -> bool: """:return: True if a 2FA authentication is needed, False otherwise. :rtype: bool """ if self.Scopes is None: return False return 'twofactor' in self.Scopes @property def supports_fido2(self) -> Optional[Fido2AssertionParameters]: """ :return: If 2FA is needed, and FIDO2 is supported, return the parameters needed to create a FIDO2 assertion. None otherwise (either 2FA is not needed, or FIDO2 is not supported). :rtype: Fido2AssertionParameters, optional """ if not self.__2FA: return None # When FIDO2 is not supported, the __2FA property contains: # {'AuthenticationOptions': None, 'RegisteredKeys': []} options = self.__2FA.get('FIDO2', {}).get("AuthenticationOptions") if not options: return None public_key = options.get("publicKey") if not public_key: return None return Fido2AssertionParameters( challenge=public_key["challenge"], rp_id=public_key["rpId"], allow_credentials=public_key["allowCredentials"], user_verification=public_key["userVerification"] ) @property def environment(self): """Get/set the environment in use for that session. It can be only set once at the beginning of the session's object lifetime, as changing the environment can lead to security hole. If the new value is: * None: do nothing * a string: will use :meth:`Loader.get("environment", newvalue)` to get the actual class. * an environment: use it """ if self.__environment is None: from proton.loader import Loader self.__environment = Loader.get('environment')() return self.__environment @environment.setter def environment(self, newvalue): # Do nothing if we set to None if newvalue is None: return if isinstance(newvalue, str): newvalue = Loader.get("environment", newvalue)() if not isinstance(newvalue, Environment): raise TypeError("environment should be a subclass of Environment") #Same environment => nothing to do if self.__environment == newvalue: return if self.__environment is not None: raise ValueError("Cannot change environment of an established session (that would create security issues)!") self.__environment = newvalue def __setstate__(self, data): # If we're running an unpickle, then the object constructor hasn't been called, so we need to populate __dict__ for attr, default in (('gnupg_for_modulus', None), ('can_run_requests', None), ('transport', None), ('persistence_observers', [])): if '_Session__' + attr not in self.__dict__: self.__dict__['_Session__' + attr] = default # Restore data from LastUseData if we don't have it already (allow pickle load) for attr, default in (('2FA', None), ('appversion', 'Other'), ('user_agent', 'None'), ('refresh_revision', 0)): if '_Session__' + attr not in self.__dict__: self.__dict__['_Session__' + attr] = data.get('LastUseData', {}).get(attr, default) # We don't pickle the transport, so if not set just use the default if '_Session__transport_factory' not in self.__dict__: self.transport_factory = None self.__UID = data.get('UID', None) self.__AccessToken = data.get('AccessToken', None) self.__RefreshToken = data.get('RefreshToken', None) self.__Scopes = data.get('Scopes', None) self.__AccountName = data.get('AccountName', None) #Reset transport (user agent etc might have changed) self.__transport = None #get environment as stored in the session if data.get('Environment', None) is not None: self.__environment: Environment = Loader.get("environment", data.get('Environment', None))() else: self.__environment = None # Store everything we don't know about in extrastate self.__extrastate = dict([(k, v) for k, v in data.items() if k not in ('UID','AccessToken','RefreshToken','Scopes','AccountName','Environment', 'LastUseData')]) def __getstate__(self): # If we don't have an UID, then we're not logged in and we don't want to store a specific state if self.UID is None: data = {} else: data = { #Session data 'UID': self.UID, 'AccessToken': self.__AccessToken, 'RefreshToken': self.__RefreshToken, 'Scopes': self.Scopes, 'Environment': self.environment.name, 'AccountName': self.__AccountName, 'LastUseData': { '2FA': self.__2FA, 'appversion': self.__appversion, 'user_agent': self.__user_agent, 'refresh_revision': self.__refresh_revision, } } # Add the additional extra state data that we might have data.update(self.__extrastate) return data def _requests_lock(self, no_condition_check=False): """Lock the session, this has to be done when doing requests that affect the session state (i.e. :meth:`authenticate` for instance), to prevent race conditions. Internally, this is done using :class:`asyncio.Event`. :param no_condition_check: Internal flag to disable locking, defaults to False :type no_condition_check: bool, optional """ if no_condition_check: return if self.__can_run_requests is None: self.__can_run_requests = asyncio.Event() self.__can_run_requests.clear() # Lock observers (we're about to modify the session) account_name = self.AccountName session_data = self.__getstate__() for observer in self.__persistence_observers: observer._acquire_session_lock(account_name, session_data) def _requests_unlock(self, no_condition_check=False, account_name=None): """Unlock the session, this has to be done after doing requests that affect the session state (i.e. :meth:`authenticate` for instance), to prevent race conditions. :param no_condition_check: Internal flag to disable locking, defaults to False :type no_condition_check: bool, optional :param account_name: Allow providing explicitly the account_name of the session, useful when it's for a logout when the session might not exist any more :type no_condition_check: str, optional """ if no_condition_check: return if self.__can_run_requests is None: self.__can_run_requests = asyncio.Event() self.__can_run_requests.set() # Only store data if we have an actual account (session not logged in shouldn't store data) # If we have a known account, use it if self.AccountName is not None: account_name = self.AccountName session_data = self.__getstate__() else: session_data = None # Unlock observers (we might have modified the session) # It's important to do it in reverse order, as otherwise there's a risk of deadlocks for observer in reversed(self.__persistence_observers): observer._release_session_lock(account_name, session_data) async def _requests_wait(self, no_condition_check=False): """Wait for session unlock. :param no_condition_check: Internal flag to disable locking, defaults to False :type no_condition_check: bool, optional """ if no_condition_check or self.__can_run_requests is None: return await self.__can_run_requests.wait() async def __sleep_for_exception(self, e): if e.http_headers.get('retry-after','-').isnumeric(): await asyncio.sleep(int(e.http_headers.get('retry-after'))) else: # No crypto risk here of using an unsafe generator await asyncio.sleep(3+random.random()*5) # nosec B311 # nosemgrep: gitlab.bandit.B311 async def __async_api_request_internal( self, endpoint, jsondata=None, data=None, additional_headers=None, method=None, params=None, no_condition_check=False, return_raw=False ): """Internal function to do an API request (without clever exception handling and retrying). See :meth:`async_api_request` for the parameters specification.""" # Should (and can we) create a transport if self.__transport is None and self.__transport_factory is not None: self.__transport = self.__transport_factory(self) if self.__transport is None: raise RuntimeError("Could not instanciate a transport, are required dependencies installed?") await self._requests_wait(no_condition_check) return await self.__transport.async_api_request(endpoint, jsondata, data, additional_headers, method, params, return_raw=return_raw) def _verify_modulus(self, armored_modulus) -> bytes: if self.__gnupg_for_modulus is None: import gnupg # The modulus key is imported in a separate GPG keyring (rather than on the user's # default one) by modifying the gnupg home directory. The reason for doing this is # that we received crash reports due to users messing up with their default GPG # keyring permissions. In this case, python-gnupg fails silently when importing # Proton's modulus key, which then causes the signature verification errors. proton_gnupg_dir = Path(ExecutionEnvironment().path_cache) / "gnupg" proton_gnupg_dir.mkdir(exist_ok=True) self.__gnupg_for_modulus = gnupg.GPG(gnupghome=proton_gnupg_dir) self.__gnupg_for_modulus.import_keys(SRP_MODULUS_KEY) # gpg.decrypt verifies the signature too, and returns the parsed data. # By using gpg.verify the data is not returned verified = self.__gnupg_for_modulus.decrypt(armored_modulus) if not (verified.valid and verified.fingerprint.lower() == SRP_MODULUS_KEY_FINGERPRINT): raise ProtonCryptoError('Invalid modulus') return base64.b64decode(verified.data.strip())