identity

ArkIdentity

Source code in ark_sdk_python/auth/identity/ark_identity.py

class ArkIdentity:
    def __init__(
        self,
        username: str,
        password: Optional[str],
        identity_url: Optional[str] = None,
        identity_tenant_subdomain: Optional[str] = None,
        mfa_type: Optional[str] = None,
        logger: Optional[logging.Logger] = None,
        cache_authentication: bool = True,
        verify: Optional[Union[str, bool]] = None,
        load_cache: bool = False,
        cache_profile: Optional[ArkProfile] = None,
    ) -> None:
        self.__username = username
        self.__password = password
        self.__identity_url = self.__resolve_fqdn_from_username_or_subdomain(identity_url, identity_tenant_subdomain)
        if not self.__identity_url.startswith('https://'):
            self.__identity_url = f'https://{self.__identity_url}'
        self.__mfa_type = mfa_type
        self.__logger = logger or get_logger(app=self.__class__.__name__)
        self.__interaction_process: Optional[Process] = None
        self.__is_polling: bool = False
        self.__keyring = ArkKeyring(self.__class__.__name__.lower()) if cache_authentication else None
        self.__cache_authentication = cache_authentication

        self.__session = Session()
        self.__session_details = None
        self.__session_exp = None
        self.__session.headers.update(ArkIdentityFQDNResolver.default_headers())
        if verify is None:
            if ArkSystemConfig.trusted_certificate() is not None:
                verify = ArkSystemConfig.trusted_certificate()
            else:
                verify = ArkSystemConfig.is_verifiying_certificates()
        self.__verify = verify
        self.__session.verify = verify
        if load_cache and cache_authentication and cache_profile:
            self.__load_cache(cache_profile)

    def __load_cache(self, profile: Optional[ArkProfile] = None) -> bool:
        if self.__keyring and profile:
            token = self.__keyring.load_token(profile, f'{self.__username}_identity')
            session = self.__keyring.load_token(profile, f'{self.__username}_identity_session')
            if token and session:
                import dill as pickle

                try:
                    self.__session_details = AdvanceAuthResult.parse_raw(token.token.get_secret_value())
                except ValidationError:
                    self.__session_details = IdpAuthStatusResult.parse_raw(token.token.get_secret_value())
                self.__session_exp = token.expires_in
                self.__session = pickle.loads(codecs.decode(session.token.get_secret_value().encode(), "base64"))
                self.__session.verify = self.__verify
                self.__identity_url = token.endpoint
                return True
        return False

    def __save_cache(self, profile: Optional[ArkProfile] = None) -> None:
        if self.__keyring and profile and self.__session_details:
            import dill as pickle

            delta = self.__session_details.token_lifetime or DEFAULT_TOKEN_LIFETIME_SECONDS
            self.__session_exp = datetime.now() + timedelta(seconds=delta)
            self.__keyring.save_token(
                profile,
                ArkToken(
                    token=self.__session_details.json(),
                    username=self.__username,
                    endpoint=self.__identity_url,
                    token_type=ArkTokenType.Internal,
                    auth_method=ArkAuthMethod.Other,
                    expires_in=self.__session_exp,
                    refresh_token=self.__session_details.refresh_token,
                ),
                f'{self.__username}_identity',
            )
            self.__keyring.save_token(
                profile,
                ArkToken(
                    token=codecs.encode(pickle.dumps(self.__session), 'base64').decode(),
                    username=self.__username,
                    endpoint=self.__identity_url,
                    token_type=ArkTokenType.Internal,
                    auth_method=ArkAuthMethod.Other,
                    expires_in=self.__session_exp,
                ),
                f'{self.__username}_identity_session',
            )

    def __resolve_fqdn_from_username_or_subdomain(self, identity_url: Optional[str], identity_tenant_subdomain: Optional[str]) -> str:
        if identity_tenant_subdomain and not identity_url:
            try:
                identity_url = ArkIdentityFQDNResolver.resolve_tenant_fqdn_from_tenant_subdomain(
                    identity_tenant_subdomain, AwsEnv(os.environ.get('DEPLOY_ENV', AwsEnv.PROD.value))
                )
            except Exception as ex:
                self.__logger.warning(f'Failed to resolve url from tenant subdomain, falling back to user [{str(ex)}]')
        if identity_url:
            return identity_url
        tenant_suffix = self.__username[self.__username.index('@') :]
        return ArkIdentityFQDNResolver.resolve_tenant_fqdn_from_tenant_suffix(tenant_suffix=tenant_suffix)

    def __start_authentication(self) -> StartAuthResponse:
        self.__logger.info(f'Starting authentication with user {self.__username} and fqdn {self.__identity_url}')
        response = self.__session.post(
            url=f'{self.__identity_url}/Security/StartAuthentication',
            json={'User': self.__username, 'Version': '1.0', 'PlatformTokenResponse': True, 'MfaRequestor': 'DeviceAgent'},
        )
        try:
            parsed_res: StartAuthResponse = StartAuthResponse.parse_raw(response.text)
            if not parsed_res.result.challenges and not parsed_res.result.idp_redirect_url:
                raise ValidationError('No challenges or idp redirect url on start auth')
        except (ValidationError, TypeError) as ex:
            try:
                if 'PodFqdn' in response.text:
                    fqdn = TenantFqdnResponse.parse_raw(response.text)
                    self.__identity_url = f'https://{fqdn.result.pod_fqdn}'
                    self.__session = Session()
                    self.__session.verify = self.__verify
                    self.__session.headers.update(ArkIdentityFQDNResolver.default_headers())
                    return self.__start_authentication()
            except Exception:
                pass
            raise ArkException('Identity start authentication failed to be parsed / validated') from ex
        return parsed_res

    def __advance_authentication(
        self, mechanism_id: str, session_id: str, answer: str, action: str
    ) -> Union[AdvanceAuthMidResponse, AdvanceAuthResponse]:
        self.__logger.info(f'Advancing authentication with user {self.__username} and fqdn {self.__identity_url} and action {action}')
        response = self.__session.post(
            url=f'{self.__identity_url}/Security/AdvanceAuthentication',
            json={'SessionId': session_id, 'MechanismId': mechanism_id, 'Action': action, 'Answer': answer},
        )
        try:
            parsed_res: AdvanceAuthMidResponse = AdvanceAuthMidResponse.parse_raw(response.text)
            if parsed_res.result.summary == 'LoginSuccess':
                parsed_res: AdvanceAuthResponse = AdvanceAuthResponse.parse_raw(response.text)
        except (ValidationError, TypeError) as ex:
            raise ArkException(f'Identity advance authentication failed to be parsed / validated [{response.text}]') from ex
        return parsed_res

    def __identity_idp_auth_status(self, session_id: str) -> IdpAuthStatusResponse:
        self.__logger.info(f'Calling idp auth status for fqdn {self.__identity_url} and session id {session_id}')
        response = self.__session.post(
            url=f'{self.__identity_url}/Security/OobAuthStatus',
            json={'SessionId': session_id},
        )
        try:
            parsed_res: IdpAuthStatusResponse = IdpAuthStatusResponse.parse_raw(response.text)
        except (ValidationError, TypeError) as ex:
            raise ArkException(f'Identity idp auth status failed to be parsed / validated [{response.text}]') from ex
        return parsed_res

    def __start_input_process(
        self, pipe_write: Connection, pipe_read: Connection, mechanism: Mechanism, oob_advance_resp: AdvanceAuthMidResponse
    ) -> None:
        if self.__interaction_process:
            raise ArkException('Interaction thread is already in progress')
        if sys.platform not in ['win32', 'cygwin']:
            ctx = get_context('fork')
        else:
            ctx = get_context('spawn')
        self.__interaction_process = ctx.Process(
            target=input_process,
            args=(
                pipe_write,
                pipe_read,
                mechanism,
                oob_advance_resp,
            ),
        )
        self.__interaction_process.start()

    def __stop_input_process(self) -> None:
        if self.__interaction_process:
            self.__interaction_process.kill()
            self.__interaction_process.join()
            self.__interaction_process = None

    def __poll_authentication(
        self,
        profile: ArkProfile,
        mechanism: Mechanism,
        start_auth_response: StartAuthResponse,
        oob_advance_resp: AdvanceAuthMidResponse,
        is_interactive: bool,
    ) -> None:
        try:
            if self.__is_polling:
                raise ArkException('MFA Polling is already in progress')
            self.__is_polling = True
            input_conn, output_conn = Pipe(duplex=True)
            if is_interactive:
                self.__start_input_process(input_conn, output_conn, mechanism, oob_advance_resp)
            start_time = datetime.now()
            while self.__is_polling:
                current_time = datetime.now()
                if (current_time - start_time).seconds >= POLL_TIME_SECONDS:
                    self.__is_polling = False
                    raise ArkException('Timeout reached while polling for user answer')
                if output_conn.poll():
                    mfa_code = output_conn.recv()
                    advance_resp = self.__advance_authentication(
                        mechanism.mechanism_id, start_auth_response.result.session_id, mfa_code, 'Answer'
                    )
                    if isinstance(advance_resp, AdvanceAuthResponse):
                        input_conn.send('DONE')
                    else:
                        input_conn.send('CONTINUE')
                else:
                    advance_resp = self.__advance_authentication(mechanism.mechanism_id, start_auth_response.result.session_id, '', 'Poll')
                if isinstance(advance_resp, AdvanceAuthResponse):
                    # Done here, save the token
                    self.__is_polling = False
                    if is_interactive:
                        self.__stop_input_process()
                    self.__session_details = advance_resp.result
                    self.__session.headers.update(
                        {'Authorization': f'Bearer {advance_resp.result.auth}', **ArkIdentityFQDNResolver.default_headers()}
                    )
                    delta = self.__session_details.token_lifetime or DEFAULT_TOKEN_LIFETIME_SECONDS
                    self.__session_exp = datetime.now() + timedelta(seconds=delta)
                    if self.__cache_authentication:
                        self.__save_cache(profile)
                    return
                time.sleep(POLL_INTERVAL_MS)
        except Exception as ex:
            if is_interactive:
                self.__stop_input_process()
            if not self.__session_details:
                raise ex
        finally:
            if is_interactive:
                self.__stop_input_process()

    def __pick_mechanism(self, challenge: Challenge) -> Mechanism:
        factors = {'otp': '📲 Push / Code', 'sms': '📟 SMS', 'email': '📧 Email', 'pf': '📞 Phone call'}
        supported_mechanisms = [m for m in challenge.mechanisms if m.name.lower() in SUPPORTED_MECHANISMS]
        answers = inquirer.prompt(
            [
                inquirer.List(
                    'mfa',
                    'Please pick one of the following MFA methods',
                    choices=[factors[m.name.lower()] for m in supported_mechanisms],
                    default=factors[self.__mfa_type] if self.__mfa_type and self.__mfa_type in factors else None,
                    carousel=True,
                )
            ],
            render=ArkInquirerRender(),
        )
        if not answers:
            raise ArkException('Failed to get answer for which MFA method to use')
        self.__mfa_type = next(filter(lambda f: factors[f] == answers['mfa'], factors.keys()))
        return next(filter(lambda m: factors[m.name.lower()] == answers['mfa'], supported_mechanisms))

    def __perform_idp_authentication(
        self, start_auth_response: StartAuthResponse, profile: Optional[ArkProfile] = None, interactive: bool = False
    ) -> None:
        if self.__is_polling:
            raise ArkException('MFA / IDP Polling is already in progress')
        # Print the user some info if we are interactive
        if interactive:
            ArkArgsFormatter.print_normal_bright(
                "\nYou are now being redirected from your browser to your external identity provider for authentication\n"
                "If the browser did not open, you may also click the following URL to access your identity provider authentication\n\n"
                f"{start_auth_response.result.idp_redirect_short_url}\n"
            )

        # Error can be ignored
        webbrowser.open(start_auth_response.result.idp_redirect_short_url, new=0, autoraise=True)

        # Start polling for idp auth
        self.__is_polling = True
        start_time = datetime.now()
        while self.__is_polling:
            current_time = datetime.now()
            if (current_time - start_time).seconds >= POLL_TIME_SECONDS:
                self.__is_polling = False
                raise ArkException('Timeout reached while polling for idp auth')
            idp_auth_status = self.__identity_idp_auth_status(start_auth_response.result.idp_login_session_id)
            if idp_auth_status.result.state == 'Success' and idp_auth_status.result.token:
                # We managed to successfully authenticate
                # Done here, save the token
                self.__session_details = idp_auth_status.result
                self.__session.headers.update(
                    {'Authorization': f'Bearer {idp_auth_status.result.token}', **ArkIdentityFQDNResolver.default_headers()}
                )
                delta = self.__session_details.token_lifetime or DEFAULT_TOKEN_LIFETIME_SECONDS
                self.__session_exp = datetime.now() + timedelta(seconds=delta)
                if self.__cache_authentication:
                    self.__save_cache(profile)
                break
            time.sleep(POLL_INTERVAL_MS)

    @classmethod
    def has_cache_record(cls, profile: ArkProfile, username: str, refresh_auth_allowed: bool) -> bool:
        """
        Checks if a cache record exists for the specified profile and username.

        Args:
            profile (ArkProfile): _description_
            username (str): _description_
            refresh_auth_allowed (bool): _description_

        Returns:
            bool: _description_
        """
        keyring = ArkKeyring(cls.__name__.lower())
        token = keyring.load_token(profile, f'{username}_identity')
        session = keyring.load_token(profile, f'{username}_identity_session')
        if token is not None and session is not None:
            if token.expires_in and token.expires_in < datetime.now():
                if token.refresh_token and refresh_auth_allowed:
                    return True
                return False
            return True
        return False

    @classmethod
    @cached(cache=LRUCache(maxsize=1024))
    def is_idp_user(cls, username: str, identity_url: Optional[str], identity_tenant_subdomain: Optional[str]) -> bool:
        """
        Checks whether or not the specified username is from an external IDP.

        Args:
            username (str): _description_
            identity_url (Optional[str]): _description_
            identity_tenant_subdomain (Optional[str]): _description_

        Returns:
            bool: _description_
        """
        if re.match('.*@cyberark\\.cloud\\.(\\d)+', username) is not None:
            return False
        identity = ArkIdentity(
            username=username,
            password='',
            identity_url=identity_url,
            identity_tenant_subdomain=identity_tenant_subdomain,
        )
        resp = identity.__start_authentication()
        return resp.result.idp_redirect_url != None

    def get_apps(self) -> Dict:
        """
        Returns the applications to which the user is logged in.

        Raises:
            ArkException: _description_

        Returns:
            Dict: _description_
        """
        if not self.__session_details:
            raise ArkException('Identity authentication is required first')
        cookies = self.__session.cookies.copy()
        response = self.__session.post(url=f'{self.__identity_url}/UPRest/GetUPData')
        self.__session.cookies = cookies
        return json.loads(response.text)

    def auth_identity(self, profile: Optional[ArkProfile] = None, interactive: bool = False, force: bool = False) -> None:
        """
        Authenticates to Identity with the information specified in the constructor.
        If MFA is configured and `interactive` is enabled, the user is prompted for the MFA secret.
        The auth token and other details are stored in the object for future use.

        Args:
            profile (Optional[ArkProfile]): Profile to use (loaded from cache, if available)
            interactive (bool): Determines whether interactive user prompts are allowed
            force (bool): Determines whether to ignore cache and force authentication (defaults to false)

        Raises:
            ArkException: _description_
        """
        self.__logger.debug('Attempting to authenticate to Identity')
        self.__session_details = None
        if self.__cache_authentication and not force and self.__load_cache(profile):
            # Check if expired
            if self.__session_exp.replace(tzinfo=None) > datetime.now():
                self.__logger.info('Loaded identity details from cache')
                return
        self.__session = Session()
        self.__session.verify = self.__verify
        self.__session.headers.update(ArkIdentityFQDNResolver.default_headers())

        # Start the authentication
        start_auth_response = self.__start_authentication()
        if start_auth_response.result.idp_redirect_url:
            # External IDP Flow, ignore the mechanisms and just open a browser
            self.__perform_idp_authentication(start_auth_response, profile, interactive)
            return

        # Check if password is part of the first challenges list and if so, answer it directly
        current_challenge_idx = 0
        for mechanism in start_auth_response.result.challenges[current_challenge_idx].mechanisms:
            if mechanism.name.lower() == 'up':
                current_challenge_idx += 1
                # Password, answer it
                if not self.__password:
                    if not interactive:
                        raise ArkAuthException('No password and not interactive, cannot continue')
                    answers = inquirer.prompt(
                        [inquirer.Password('answer', message='Identity Security Platform Secret')],
                        render=ArkInquirerRender(),
                    )
                    if not answers:
                        raise ArkAuthException('Canceled by user')
                    self.__password = answers['answer']
                advance_resp = self.__advance_authentication(
                    mechanism.mechanism_id, start_auth_response.result.session_id, self.__password, 'Answer'
                )
                if isinstance(advance_resp, AdvanceAuthResponse) and len(start_auth_response.result.challenges) == 1:
                    # Done here, save the token
                    self.__session_details = advance_resp.result
                    self.__session.headers.update(
                        {'Authorization': f'Bearer {advance_resp.result.auth}', **ArkIdentityFQDNResolver.default_headers()}
                    )
                    delta = self.__session_details.token_lifetime or DEFAULT_TOKEN_LIFETIME_SECONDS
                    self.__session_exp = datetime.now() + timedelta(seconds=delta)
                    if self.__cache_authentication:
                        self.__save_cache(profile)
                    return
                break

        # Pick MFA for the user
        if interactive:
            self.__pick_mechanism(start_auth_response.result.challenges[current_challenge_idx])

        # Handle a case where MFA type was supplied
        if self.__mfa_type and self.__mfa_type.lower() in SUPPORTED_MECHANISMS and current_challenge_idx == 1:
            for mechanism in start_auth_response.result.challenges[current_challenge_idx].mechanisms:
                if mechanism.name.lower() == self.__mfa_type.lower():
                    oob_advance_resp = self.__advance_authentication(
                        mechanism.mechanism_id, start_auth_response.result.session_id, '', 'StartOOB'
                    )
                    self.__poll_authentication(profile, mechanism, start_auth_response, oob_advance_resp, interactive)
                    return

        if not interactive:
            raise ArkNonInteractiveException('User interaction is not supported while not interactive and mfa type given was not found')

        # Handle the rest of the challenges, might also handle the first challenge if no password is in the mechanisms
        for challenge in start_auth_response.result.challenges[current_challenge_idx:]:
            mechanism = self.__pick_mechanism(challenge)
            oob_advance_resp = self.__advance_authentication(mechanism.mechanism_id, start_auth_response.result.session_id, '', 'StartOOB')
            self.__poll_authentication(profile, mechanism, start_auth_response, oob_advance_resp, interactive)

    # pylint: disable=unused-argument
    def refresh_auth_identity(self, profile: Optional[ArkProfile] = None, interactive: bool = False, force: bool = False) -> None:
        """
        Performs a token refresh with the object's existing details.

        Args:
            profile (Optional[ArkProfile]): The profile to load from the cache, if available
            force (bool): Determines whether to ignore cache and force authentication (defaults to false)

        Raises:
            ArkAuthException: _description_
        """
        from jose.jwt import get_unverified_claims

        if not self.__session_details.token:
            # We only refresh platform token at the moment, call the normal authentication instead
            return self.auth_identity(profile, interactive, force)

        self.__logger.debug('Attempting to refresh authenticate to Identity')
        saved_cookies = None
        if self.__session:
            saved_cookies = self.__session.cookies.copy()
        self.__session = Session()
        self.__session.verify = self.__verify
        self.__session.headers.update(ArkIdentityFQDNResolver.default_headers())
        decoded_token = get_unverified_claims(self.__session_details.token)
        platform_tenant_id = decoded_token['tenant_id']
        refresh_cookies = {
            f'refreshToken-{platform_tenant_id}': self.__session_details.refresh_token,
            f'idToken-{platform_tenant_id}': self.__session_details.token,
        }
        response = self.__session.post(
            url=f'{self.__identity_url}/OAuth2/RefreshPlatformToken',
            cookies=refresh_cookies,
        )
        if response.status_code != HTTPStatus.OK:
            raise ArkAuthException('Failed to refresh token')
        new_token = response.cookies.get(f'idToken-{platform_tenant_id}')
        new_refresh_token = response.cookies.get(f'refreshToken-{platform_tenant_id}')
        if not new_token or not new_refresh_token:
            raise ArkAuthException('Failed to retrieve refresh tokens cookies')
        if saved_cookies:
            self.__session.cookies = saved_cookies
        self.__session_details.token = new_token
        self.__session_details.refresh_token = new_refresh_token
        self.__session_details.token_lifetime = (
            datetime.fromtimestamp(get_unverified_claims(new_token)['exp'])
            - datetime.fromtimestamp(get_unverified_claims(new_token)['iat'])
        ).seconds
        delta = self.__session_details.token_lifetime or DEFAULT_TOKEN_LIFETIME_SECONDS
        self.__session_exp = datetime.now() + timedelta(seconds=delta)
        if self.__cache_authentication:
            self.__save_cache(profile)

    @property
    def session(self) -> Session:
        return self.__session

    @property
    def session_token(self) -> Optional[str]:
        if self.__session_details:
            if self.__session_details.token:
                return self.__session_details.token
            if 'auth' in self.__session_details.__dict__ and self.__session_details.auth:
                return self.__session_details.auth
        return None

    @property
    def session_details(self) -> Optional[AdvanceAuthResult]:
        return self.__session_details

    @property
    def identity_url(self) -> str:
        return self.__identity_url

auth_identity(profile=None, interactive=False, force=False)

Authenticates to Identity with the information specified in the constructor. If MFA is configured and interactive is enabled, the user is prompted for the MFA secret. The auth token and other details are stored in the object for future use.

Parameters:

Name	Type	Description	Default
profile	Optional[ArkProfile]	Profile to use (loaded from cache, if available)	None
interactive	bool	Determines whether interactive user prompts are allowed	False
force	bool	Determines whether to ignore cache and force authentication (defaults to false)	False

Raises:

Type	Description
ArkException	description

Source code in ark_sdk_python/auth/identity/ark_identity.py

def auth_identity(self, profile: Optional[ArkProfile] = None, interactive: bool = False, force: bool = False) -> None:
    """
    Authenticates to Identity with the information specified in the constructor.
    If MFA is configured and `interactive` is enabled, the user is prompted for the MFA secret.
    The auth token and other details are stored in the object for future use.

    Args:
        profile (Optional[ArkProfile]): Profile to use (loaded from cache, if available)
        interactive (bool): Determines whether interactive user prompts are allowed
        force (bool): Determines whether to ignore cache and force authentication (defaults to false)

    Raises:
        ArkException: _description_
    """
    self.__logger.debug('Attempting to authenticate to Identity')
    self.__session_details = None
    if self.__cache_authentication and not force and self.__load_cache(profile):
        # Check if expired
        if self.__session_exp.replace(tzinfo=None) > datetime.now():
            self.__logger.info('Loaded identity details from cache')
            return
    self.__session = Session()
    self.__session.verify = self.__verify
    self.__session.headers.update(ArkIdentityFQDNResolver.default_headers())

    # Start the authentication
    start_auth_response = self.__start_authentication()
    if start_auth_response.result.idp_redirect_url:
        # External IDP Flow, ignore the mechanisms and just open a browser
        self.__perform_idp_authentication(start_auth_response, profile, interactive)
        return

    # Check if password is part of the first challenges list and if so, answer it directly
    current_challenge_idx = 0
    for mechanism in start_auth_response.result.challenges[current_challenge_idx].mechanisms:
        if mechanism.name.lower() == 'up':
            current_challenge_idx += 1
            # Password, answer it
            if not self.__password:
                if not interactive:
                    raise ArkAuthException('No password and not interactive, cannot continue')
                answers = inquirer.prompt(
                    [inquirer.Password('answer', message='Identity Security Platform Secret')],
                    render=ArkInquirerRender(),
                )
                if not answers:
                    raise ArkAuthException('Canceled by user')
                self.__password = answers['answer']
            advance_resp = self.__advance_authentication(
                mechanism.mechanism_id, start_auth_response.result.session_id, self.__password, 'Answer'
            )
            if isinstance(advance_resp, AdvanceAuthResponse) and len(start_auth_response.result.challenges) == 1:
                # Done here, save the token
                self.__session_details = advance_resp.result
                self.__session.headers.update(
                    {'Authorization': f'Bearer {advance_resp.result.auth}', **ArkIdentityFQDNResolver.default_headers()}
                )
                delta = self.__session_details.token_lifetime or DEFAULT_TOKEN_LIFETIME_SECONDS
                self.__session_exp = datetime.now() + timedelta(seconds=delta)
                if self.__cache_authentication:
                    self.__save_cache(profile)
                return
            break

    # Pick MFA for the user
    if interactive:
        self.__pick_mechanism(start_auth_response.result.challenges[current_challenge_idx])

    # Handle a case where MFA type was supplied
    if self.__mfa_type and self.__mfa_type.lower() in SUPPORTED_MECHANISMS and current_challenge_idx == 1:
        for mechanism in start_auth_response.result.challenges[current_challenge_idx].mechanisms:
            if mechanism.name.lower() == self.__mfa_type.lower():
                oob_advance_resp = self.__advance_authentication(
                    mechanism.mechanism_id, start_auth_response.result.session_id, '', 'StartOOB'
                )
                self.__poll_authentication(profile, mechanism, start_auth_response, oob_advance_resp, interactive)
                return

    if not interactive:
        raise ArkNonInteractiveException('User interaction is not supported while not interactive and mfa type given was not found')

    # Handle the rest of the challenges, might also handle the first challenge if no password is in the mechanisms
    for challenge in start_auth_response.result.challenges[current_challenge_idx:]:
        mechanism = self.__pick_mechanism(challenge)
        oob_advance_resp = self.__advance_authentication(mechanism.mechanism_id, start_auth_response.result.session_id, '', 'StartOOB')
        self.__poll_authentication(profile, mechanism, start_auth_response, oob_advance_resp, interactive)

get_apps()

Returns the applications to which the user is logged in.

Raises:

Type	Description
ArkException	description

Returns:

Name	Type	Description
Dict	Dict	description

Source code in ark_sdk_python/auth/identity/ark_identity.py

def get_apps(self) -> Dict:
    """
    Returns the applications to which the user is logged in.

    Raises:
        ArkException: _description_

    Returns:
        Dict: _description_
    """
    if not self.__session_details:
        raise ArkException('Identity authentication is required first')
    cookies = self.__session.cookies.copy()
    response = self.__session.post(url=f'{self.__identity_url}/UPRest/GetUPData')
    self.__session.cookies = cookies
    return json.loads(response.text)

has_cache_record(profile, username, refresh_auth_allowed) classmethod

Checks if a cache record exists for the specified profile and username.

Parameters:

Name	Type	Description	Default
profile	ArkProfile	description	required
username	str	description	required
refresh_auth_allowed	bool	description	required

Returns:

Name	Type	Description
bool	bool	description

Source code in ark_sdk_python/auth/identity/ark_identity.py

@classmethod
def has_cache_record(cls, profile: ArkProfile, username: str, refresh_auth_allowed: bool) -> bool:
    """
    Checks if a cache record exists for the specified profile and username.

    Args:
        profile (ArkProfile): _description_
        username (str): _description_
        refresh_auth_allowed (bool): _description_

    Returns:
        bool: _description_
    """
    keyring = ArkKeyring(cls.__name__.lower())
    token = keyring.load_token(profile, f'{username}_identity')
    session = keyring.load_token(profile, f'{username}_identity_session')
    if token is not None and session is not None:
        if token.expires_in and token.expires_in < datetime.now():
            if token.refresh_token and refresh_auth_allowed:
                return True
            return False
        return True
    return False

is_idp_user(username, identity_url, identity_tenant_subdomain) classmethod

Checks whether or not the specified username is from an external IDP.

Parameters:

Name	Type	Description	Default
username	str	description	required
identity_url	Optional[str]	description	required
identity_tenant_subdomain	Optional[str]	description	required

Returns:

Name	Type	Description
bool	bool	description

Source code in ark_sdk_python/auth/identity/ark_identity.py

@classmethod
@cached(cache=LRUCache(maxsize=1024))
def is_idp_user(cls, username: str, identity_url: Optional[str], identity_tenant_subdomain: Optional[str]) -> bool:
    """
    Checks whether or not the specified username is from an external IDP.

    Args:
        username (str): _description_
        identity_url (Optional[str]): _description_
        identity_tenant_subdomain (Optional[str]): _description_

    Returns:
        bool: _description_
    """
    if re.match('.*@cyberark\\.cloud\\.(\\d)+', username) is not None:
        return False
    identity = ArkIdentity(
        username=username,
        password='',
        identity_url=identity_url,
        identity_tenant_subdomain=identity_tenant_subdomain,
    )
    resp = identity.__start_authentication()
    return resp.result.idp_redirect_url != None

refresh_auth_identity(profile=None, interactive=False, force=False)

Performs a token refresh with the object's existing details.

Parameters:

Name	Type	Description	Default
profile	Optional[ArkProfile]	The profile to load from the cache, if available	None
force	bool	Determines whether to ignore cache and force authentication (defaults to false)	False

Raises:

Type	Description
ArkAuthException	description

Source code in ark_sdk_python/auth/identity/ark_identity.py

def refresh_auth_identity(self, profile: Optional[ArkProfile] = None, interactive: bool = False, force: bool = False) -> None:
    """
    Performs a token refresh with the object's existing details.

    Args:
        profile (Optional[ArkProfile]): The profile to load from the cache, if available
        force (bool): Determines whether to ignore cache and force authentication (defaults to false)

    Raises:
        ArkAuthException: _description_
    """
    from jose.jwt import get_unverified_claims

    if not self.__session_details.token:
        # We only refresh platform token at the moment, call the normal authentication instead
        return self.auth_identity(profile, interactive, force)

    self.__logger.debug('Attempting to refresh authenticate to Identity')
    saved_cookies = None
    if self.__session:
        saved_cookies = self.__session.cookies.copy()
    self.__session = Session()
    self.__session.verify = self.__verify
    self.__session.headers.update(ArkIdentityFQDNResolver.default_headers())
    decoded_token = get_unverified_claims(self.__session_details.token)
    platform_tenant_id = decoded_token['tenant_id']
    refresh_cookies = {
        f'refreshToken-{platform_tenant_id}': self.__session_details.refresh_token,
        f'idToken-{platform_tenant_id}': self.__session_details.token,
    }
    response = self.__session.post(
        url=f'{self.__identity_url}/OAuth2/RefreshPlatformToken',
        cookies=refresh_cookies,
    )
    if response.status_code != HTTPStatus.OK:
        raise ArkAuthException('Failed to refresh token')
    new_token = response.cookies.get(f'idToken-{platform_tenant_id}')
    new_refresh_token = response.cookies.get(f'refreshToken-{platform_tenant_id}')
    if not new_token or not new_refresh_token:
        raise ArkAuthException('Failed to retrieve refresh tokens cookies')
    if saved_cookies:
        self.__session.cookies = saved_cookies
    self.__session_details.token = new_token
    self.__session_details.refresh_token = new_refresh_token
    self.__session_details.token_lifetime = (
        datetime.fromtimestamp(get_unverified_claims(new_token)['exp'])
        - datetime.fromtimestamp(get_unverified_claims(new_token)['iat'])
    ).seconds
    delta = self.__session_details.token_lifetime or DEFAULT_TOKEN_LIFETIME_SECONDS
    self.__session_exp = datetime.now() + timedelta(seconds=delta)
    if self.__cache_authentication:
        self.__save_cache(profile)

ArkIdentityFQDNResolver

Source code in ark_sdk_python/auth/identity/ark_identity_fqdn_resolver.py

class ArkIdentityFQDNResolver:
    __DISCOVERY_SERVICE_DOMAIN_NAME: Final[str] = 'platform-discovery'
    __DISCOVERY_TIMEOUT: Final[int] = 30

    @staticmethod
    @cached(cache=LRUCache(maxsize=1024))
    def default_headers() -> Dict[str, str]:
        from fake_useragent import UserAgent

        return {
            'Content-Type': 'application/json',
            'X-IDAP-NATIVE-CLIENT': 'true',
            'User-Agent': UserAgent(browsers=['chrome']).chrome,
            'OobIdPAuth': 'true',
        }

    @staticmethod
    @cached(cache=LRUCache(maxsize=1024))
    def default_system_headers() -> Dict[str, str]:
        from fake_useragent import UserAgent

        return {'X-IDAP-NATIVE-CLIENT': 'true', 'User-Agent': UserAgent(browsers=['chrome']).chrome}

    @staticmethod
    @cached(cache=LRUCache(maxsize=1024))
    def resolve_tenant_fqdn_from_tenant_subdomain(tenant_subdomain: str, env: AwsEnv) -> str:
        """
        Resolves the tenant's FQDN URL from its subdomain.
        The resolved URL is based on the current working environment, which is provided in the `tenant_subdomain` argument.

        Args:
            tenant_subdomain (str): The tenant subdomain, for example: `mytenant`
            env (AwsEnv): The environment for which the the tenant URL is resolved

        Raises:
            ArkException: When an error occurs or the tenant username prefix was not found in the Identity environment

        Returns:
            str: The tenant's resolved FQDN
        """
        platform_discovery_url = f'https://{ArkIdentityFQDNResolver.__DISCOVERY_SERVICE_DOMAIN_NAME}.{ROOT_DOMAIN[env]}'
        session = Session()
        response = session.get(
            f'{platform_discovery_url}/api/identity-endpoint/{tenant_subdomain}',
            headers={'Content-Type': 'application/json'},
            timeout=ArkIdentityFQDNResolver.__DISCOVERY_TIMEOUT,
        )
        try:
            if response.status_code == HTTPStatus.OK:
                parsed_response: IdentityEndpointResponse = IdentityEndpointResponse.parse_raw(response.text)
                return str(parsed_response.endpoint)
        except (ValidationError, TypeError) as ex:
            raise ArkException('Getting tenant FQDN failed from platform discovery to be parsed / validated') from ex
        raise ArkException(f'Getting tenant FQDN failed from platform discovery [{response.status_code}] - [{response.text}]')

    @staticmethod
    @cached(cache=LRUCache(maxsize=1024))
    def resolve_tenant_fqdn_from_tenant_suffix(tenant_suffix: str, identity_env_url: Optional[str] = None) -> str:
        """
        Resolves the tenant's FQDN URL in Identity.
        By default, the Identity address is resolved from the current environment mapping (see `get_identity_env_url()`), but it can be optionally be resolved from the `identity_env_url` argument.

        Args:
            tenant_suffix (str): The tenant's URL suffix, for example: `@tenant-a-527.shell.cyberark.cloud`
            identity_env_url (str, optional): If specified, used as the Identity pod0 URL; otherwise, defaults to `None` (use environment mapping)

        Raises:
            ArkException: In case of error, or tenant username prefix was not found in identity environment

        Returns:
            str: The tenant's FQDN
        """
        identity_env_url = identity_env_url or (
            IDENTITY_ENV_URLS[AwsEnv(os.getenv('DEPLOY_ENV', None))] if os.getenv('DEPLOY_ENV', None) else IDENTITY_ENV_URLS[AwsEnv.PROD]
        )
        session = Session()
        response = session.post(
            f'https://pod0.{identity_env_url}/Security/StartAuthentication',
            json={'User': tenant_suffix, 'Version': '1.0', 'PlatformTokenResponse': True, 'MfaRequestor': 'DeviceAgent'},
            headers={'Content-Type': 'application/json', 'X-IDAP-NATIVE-CLIENT': 'true'},
        )
        try:
            parsed_res: TenantFqdnResponse = TenantFqdnResponse.parse_raw(response.text)
        except (ValidationError, TypeError) as ex:
            raise ArkException('Getting tenant FQDN failed to be parsed / validated') from ex
        if not parsed_res.result.pod_fqdn.startswith('https://'):
            parsed_res.result.pod_fqdn = f'https://{parsed_res.result.pod_fqdn}'
        return parsed_res.result.pod_fqdn

resolve_tenant_fqdn_from_tenant_subdomain(tenant_subdomain, env) staticmethod

Resolves the tenant's FQDN URL from its subdomain. The resolved URL is based on the current working environment, which is provided in the tenant_subdomain argument.

Parameters:

Name	Type	Description	Default
tenant_subdomain	str	The tenant subdomain, for example: mytenant	required
env	AwsEnv	The environment for which the the tenant URL is resolved	required

Raises:

Type	Description
ArkException	When an error occurs or the tenant username prefix was not found in the Identity environment

Returns:

Name	Type	Description
str	str	The tenant's resolved FQDN

Source code in ark_sdk_python/auth/identity/ark_identity_fqdn_resolver.py

@staticmethod
@cached(cache=LRUCache(maxsize=1024))
def resolve_tenant_fqdn_from_tenant_subdomain(tenant_subdomain: str, env: AwsEnv) -> str:
    """
    Resolves the tenant's FQDN URL from its subdomain.
    The resolved URL is based on the current working environment, which is provided in the `tenant_subdomain` argument.

    Args:
        tenant_subdomain (str): The tenant subdomain, for example: `mytenant`
        env (AwsEnv): The environment for which the the tenant URL is resolved

    Raises:
        ArkException: When an error occurs or the tenant username prefix was not found in the Identity environment

    Returns:
        str: The tenant's resolved FQDN
    """
    platform_discovery_url = f'https://{ArkIdentityFQDNResolver.__DISCOVERY_SERVICE_DOMAIN_NAME}.{ROOT_DOMAIN[env]}'
    session = Session()
    response = session.get(
        f'{platform_discovery_url}/api/identity-endpoint/{tenant_subdomain}',
        headers={'Content-Type': 'application/json'},
        timeout=ArkIdentityFQDNResolver.__DISCOVERY_TIMEOUT,
    )
    try:
        if response.status_code == HTTPStatus.OK:
            parsed_response: IdentityEndpointResponse = IdentityEndpointResponse.parse_raw(response.text)
            return str(parsed_response.endpoint)
    except (ValidationError, TypeError) as ex:
        raise ArkException('Getting tenant FQDN failed from platform discovery to be parsed / validated') from ex
    raise ArkException(f'Getting tenant FQDN failed from platform discovery [{response.status_code}] - [{response.text}]')

resolve_tenant_fqdn_from_tenant_suffix(tenant_suffix, identity_env_url=None) staticmethod

Resolves the tenant's FQDN URL in Identity. By default, the Identity address is resolved from the current environment mapping (see get_identity_env_url()), but it can be optionally be resolved from the identity_env_url argument.

Parameters:

Name	Type	Description	Default
tenant_suffix	str	The tenant's URL suffix, for example: @tenant-a-527.shell.cyberark.cloud	required
identity_env_url	str	If specified, used as the Identity pod0 URL; otherwise, defaults to None (use environment mapping)	None

Raises:

Type	Description
ArkException	In case of error, or tenant username prefix was not found in identity environment

Returns:

Name	Type	Description
str	str	The tenant's FQDN

Source code in ark_sdk_python/auth/identity/ark_identity_fqdn_resolver.py

@staticmethod
@cached(cache=LRUCache(maxsize=1024))
def resolve_tenant_fqdn_from_tenant_suffix(tenant_suffix: str, identity_env_url: Optional[str] = None) -> str:
    """
    Resolves the tenant's FQDN URL in Identity.
    By default, the Identity address is resolved from the current environment mapping (see `get_identity_env_url()`), but it can be optionally be resolved from the `identity_env_url` argument.

    Args:
        tenant_suffix (str): The tenant's URL suffix, for example: `@tenant-a-527.shell.cyberark.cloud`
        identity_env_url (str, optional): If specified, used as the Identity pod0 URL; otherwise, defaults to `None` (use environment mapping)

    Raises:
        ArkException: In case of error, or tenant username prefix was not found in identity environment

    Returns:
        str: The tenant's FQDN
    """
    identity_env_url = identity_env_url or (
        IDENTITY_ENV_URLS[AwsEnv(os.getenv('DEPLOY_ENV', None))] if os.getenv('DEPLOY_ENV', None) else IDENTITY_ENV_URLS[AwsEnv.PROD]
    )
    session = Session()
    response = session.post(
        f'https://pod0.{identity_env_url}/Security/StartAuthentication',
        json={'User': tenant_suffix, 'Version': '1.0', 'PlatformTokenResponse': True, 'MfaRequestor': 'DeviceAgent'},
        headers={'Content-Type': 'application/json', 'X-IDAP-NATIVE-CLIENT': 'true'},
    )
    try:
        parsed_res: TenantFqdnResponse = TenantFqdnResponse.parse_raw(response.text)
    except (ValidationError, TypeError) as ex:
        raise ArkException('Getting tenant FQDN failed to be parsed / validated') from ex
    if not parsed_res.result.pod_fqdn.startswith('https://'):
        parsed_res.result.pod_fqdn = f'https://{parsed_res.result.pod_fqdn}'
    return parsed_res.result.pod_fqdn

ArkIdentityServiceUser

Source code in ark_sdk_python/auth/identity/ark_identity_service_user.py

class ArkIdentityServiceUser:
    def __init__(
        self,
        username: str,
        token: str,
        app_name: str,
        identity_url: Optional[str] = None,
        env: Optional[AwsEnv] = None,
        logger: Optional[logging.Logger] = None,
        cache_authentication: bool = True,
        verify: Optional[Union[str, bool]] = None,
        load_cache: bool = False,
        cache_profile: Optional[ArkProfile] = None,
    ) -> None:
        self.__username = username
        self.__token = token
        self.__app_name = app_name
        self.__env = env or AwsEnv(os.getenv(DEPLOY_ENV, AwsEnv.PROD.value))
        self.__identity_url = identity_url or self.__resolve_fqdn_from_username()
        self.__logger = logger or get_logger(app=self.__class__.__name__)
        self.__keyring = ArkKeyring(self.__class__.__name__.lower()) if cache_authentication else None
        self.__cache_authentication = cache_authentication

        self.__session = Session()
        self.__session_token = None
        self.__session_exp = None
        self.__session.headers.update(ArkIdentityFQDNResolver.default_system_headers())
        if verify is None:
            if ArkSystemConfig.trusted_certificate() is not None:
                verify = ArkSystemConfig.trusted_certificate()
            else:
                verify = ArkSystemConfig.is_verifiying_certificates()
        self.__session.verify = verify
        if load_cache and cache_authentication and cache_profile:
            self.__load_cache(cache_profile)

    def __load_cache(self, profile: Optional[ArkProfile] = None) -> bool:
        if self.__keyring and profile:
            token = self.__keyring.load_token(profile, f'{self.__username}_identity_service_user')
            if token and token.username == self.__username:
                self.__session_token = token.token.get_secret_value()
                self.__session_exp = token.expires_in
                self.__session.headers.update({'Authorization': f'Bearer {self.__session_token}'})
                return True
        return False

    def __save_cache(self, profile: Optional[ArkProfile] = None) -> None:
        if self.__keyring and profile and self.__session_token:
            self.__session_exp = datetime.now() + timedelta(hours=4)
            self.__keyring.save_token(
                profile,
                ArkToken(
                    token=self.__session_token,
                    username=self.__username,
                    endpoint=self.__identity_url,
                    token_type=ArkTokenType.Internal,
                    auth_method=ArkAuthMethod.Other,
                    expires_in=self.__session_exp,
                ),
                f'{self.__username}_identity_service_user',
            )

    def __resolve_fqdn_from_username(self) -> str:
        tenant_suffix = self.__username[self.__username.index('@') :]
        return ArkIdentityFQDNResolver.resolve_tenant_fqdn_from_tenant_suffix(
            tenant_suffix=tenant_suffix, identity_env_url=IDENTITY_ENV_URLS[self.__env]
        )

    def auth_identity(self, profile: Optional[ArkProfile] = None, force: bool = False) -> None:
        """
        Authenticates to Identity with a service user.
        This method creates an auth token and authorizes to the service.

        Args:
            profile (Optional[ArkProfile]): Profile to be used to load from caching, if available
            force (bool): Determines whether to discard existing cache, defaults to `False`

        Raises:
            ArkAuthException: _description_
        """
        # Login to identity with the service service user
        self.__logger.info(f'Authenticating to service user via endpoint [{self.__identity_url}]')
        if self.__cache_authentication and not force and self.__load_cache(profile):
            # Check if expired
            if self.__session_exp.replace(tzinfo=None) > datetime.now():
                self.__logger.info('Loaded identity service user details from cache')
                return

        token_response: Response = self.__session.post(
            url=f'{self.__identity_url}/Oauth2/Token/{self.__app_name}',
            auth=HTTPBasicAuth(self.__username, self.__token),
            verify=True,
            data={'grant_type': 'client_credentials', 'scope': 'api'},
        )
        if token_response.status_code != HTTPStatus.OK:
            raise ArkAuthException('Failed logging in to identity service user')
        auth_result = json.loads(token_response.text)
        if 'access_token' not in auth_result.keys():
            raise ArkAuthException('Failed logging in to identity service user, access token not found')
        access_token = auth_result['access_token']

        # Authorize to the application with the service user
        params = {
            'client_id': self.__app_name,
            'response_type': 'id_token',
            'scope': 'openid profile api',
            'redirect_uri': 'https://cyberark.cloud/redirect',
        }
        self.__logger.info(f'Trying to request a platform authorization with params [{params}]')
        authorize_response = self.__session.get(
            url=f'{self.__identity_url}/OAuth2/Authorize/{self.__app_name}',
            headers={'Authorization': f'Bearer {access_token}'},
            params=params,
            allow_redirects=False,
        )
        if authorize_response.status_code != HTTPStatus.FOUND or 'Location' not in authorize_response.headers:
            raise ArkAuthException('Failed to authorize to application')
        # Parse the authorized token and return the session with it
        location_header_splitted = authorize_response.headers['Location'].split('#', 1)
        if len(location_header_splitted) != 2:
            raise ArkAuthException('Failed to parse location header to retrieve token from')
        parsed_query = parse_qs(location_header_splitted[1])
        if 'id_token' not in parsed_query or len(parsed_query['id_token']) != 1:
            raise ArkAuthException('Failed to parse id token from location header')
        self.__session_token = parsed_query['id_token'][0]
        self.__session.headers.update({'Authorization': f'Bearer {self.__session_token}', **ArkIdentityFQDNResolver.default_headers()})
        self.__session_exp = datetime.now() + timedelta(hours=4)
        self.__logger.info(
            f'Created a service user session via endpoint [{self.__identity_url}] ' f'with user [{self.__username}] to platform'
        )
        if self.__cache_authentication:
            self.__save_cache(profile)

    @property
    def session(self) -> Session:
        return self.__session

    @property
    def session_token(self) -> Optional[str]:
        return self.__session_token

    @property
    def identity_url(self) -> str:
        return self.__identity_url

auth_identity(profile=None, force=False)

Authenticates to Identity with a service user. This method creates an auth token and authorizes to the service.

Parameters:

Name	Type	Description	Default
profile	Optional[ArkProfile]	Profile to be used to load from caching, if available	None
force	bool	Determines whether to discard existing cache, defaults to False	False

Raises:

Type	Description
ArkAuthException	description

Source code in ark_sdk_python/auth/identity/ark_identity_service_user.py

def auth_identity(self, profile: Optional[ArkProfile] = None, force: bool = False) -> None:
    """
    Authenticates to Identity with a service user.
    This method creates an auth token and authorizes to the service.

    Args:
        profile (Optional[ArkProfile]): Profile to be used to load from caching, if available
        force (bool): Determines whether to discard existing cache, defaults to `False`

    Raises:
        ArkAuthException: _description_
    """
    # Login to identity with the service service user
    self.__logger.info(f'Authenticating to service user via endpoint [{self.__identity_url}]')
    if self.__cache_authentication and not force and self.__load_cache(profile):
        # Check if expired
        if self.__session_exp.replace(tzinfo=None) > datetime.now():
            self.__logger.info('Loaded identity service user details from cache')
            return

    token_response: Response = self.__session.post(
        url=f'{self.__identity_url}/Oauth2/Token/{self.__app_name}',
        auth=HTTPBasicAuth(self.__username, self.__token),
        verify=True,
        data={'grant_type': 'client_credentials', 'scope': 'api'},
    )
    if token_response.status_code != HTTPStatus.OK:
        raise ArkAuthException('Failed logging in to identity service user')
    auth_result = json.loads(token_response.text)
    if 'access_token' not in auth_result.keys():
        raise ArkAuthException('Failed logging in to identity service user, access token not found')
    access_token = auth_result['access_token']

    # Authorize to the application with the service user
    params = {
        'client_id': self.__app_name,
        'response_type': 'id_token',
        'scope': 'openid profile api',
        'redirect_uri': 'https://cyberark.cloud/redirect',
    }
    self.__logger.info(f'Trying to request a platform authorization with params [{params}]')
    authorize_response = self.__session.get(
        url=f'{self.__identity_url}/OAuth2/Authorize/{self.__app_name}',
        headers={'Authorization': f'Bearer {access_token}'},
        params=params,
        allow_redirects=False,
    )
    if authorize_response.status_code != HTTPStatus.FOUND or 'Location' not in authorize_response.headers:
        raise ArkAuthException('Failed to authorize to application')
    # Parse the authorized token and return the session with it
    location_header_splitted = authorize_response.headers['Location'].split('#', 1)
    if len(location_header_splitted) != 2:
        raise ArkAuthException('Failed to parse location header to retrieve token from')
    parsed_query = parse_qs(location_header_splitted[1])
    if 'id_token' not in parsed_query or len(parsed_query['id_token']) != 1:
        raise ArkAuthException('Failed to parse id token from location header')
    self.__session_token = parsed_query['id_token'][0]
    self.__session.headers.update({'Authorization': f'Bearer {self.__session_token}', **ArkIdentityFQDNResolver.default_headers()})
    self.__session_exp = datetime.now() + timedelta(hours=4)
    self.__logger.info(
        f'Created a service user session via endpoint [{self.__identity_url}] ' f'with user [{self.__username}] to platform'
    )
    if self.__cache_authentication:
        self.__save_cache(profile)