Skip to content

identity

ArkIdentity

Source code in ark_sdk_python/auth/identity/ark_identity.py
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
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
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
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
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
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
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
@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
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
@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
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
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
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@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
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@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
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
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
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
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)