ADB-600: Fixed an issue where the idp user check was not taking the identity url into account, also added subdomain optional param for discovery (cyberark#17)

Author: ofiriluz

ark_sdk_python/actions/ark_login_action.py

@@ -16,6 +16,7 @@
     ArkAuthMethodsRequireCredentials,
     ArkSecret,
     ArkToken,
+    IdentityArkAuthMethodSettings,
 )
 
 
@@ -138,7 +139,19 @@ def run_action(self, args: argparse.Namespace) -> None:
                         if (
                             authenticator_name == 'isp'
                             and auth_profile.auth_method == ArkAuthMethod.Identity
-                            and ArkIdentity.is_idp_user(auth_profile.username)
+                            and ArkIdentity.is_idp_user(
+                                auth_profile.username,
+                                (
+                                    auth_profile.auth_method_settings.identity_url
+                                    if isinstance(auth_profile.auth_method_settings, IdentityArkAuthMethodSettings)
+                                    else None
+                                ),
+                                (
+                                    auth_profile.auth_method_settings.identity_tenant_subdomain
+                                    if isinstance(auth_profile.auth_method_settings, IdentityArkAuthMethodSettings)
+                                    else None
+                                ),
+                            )
                         ):
                             secret = ArkSecret(secret='')
                         else:

ark_sdk_python/auth/ark_isp_auth.py

@@ -46,6 +46,7 @@ def __perform_identity_authentication(
                 username=auth_profile.username,
                 password=secret.secret.get_secret_value() if secret else None,
                 identity_url=method_settings.identity_url,
+                identity_tenant_subdomain=method_settings.identity_tenant_subdomain,
                 mfa_type=method_settings.identity_mfa_method,
                 logger=self._logger,
                 cache_authentication=self._cache_authentication,

ark_sdk_python/auth/identity/ark_identity.py

@@ -1,6 +1,7 @@
 import codecs
 import json
 import logging
+import os
 import re
 import sys
 import time
@@ -19,6 +20,7 @@
 from ark_sdk_python.args import ArkArgsFormatter, ArkInquirerRender
 from ark_sdk_python.auth.identity.ark_identity_fqdn_resolver import ArkIdentityFQDNResolver
 from ark_sdk_python.common import ArkKeyring, ArkSystemConfig, get_logger
+from ark_sdk_python.common.env import AwsEnv
 from ark_sdk_python.models import ArkException, ArkNonInteractiveException
 from ark_sdk_python.models.ark_exceptions import ArkAuthException
 from ark_sdk_python.models.ark_profile import ArkProfile
@@ -75,6 +77,7 @@ def __init__(
         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,
@@ -84,7 +87,7 @@ def __init__(
     ) -> None:
         self.__username = username
         self.__password = password
-        self.__identity_url = identity_url or self.__resolve_fqdn_from_username()
+        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
@@ -158,15 +161,24 @@ def __save_cache(self, profile: Optional[ArkProfile] = None) -> None:
                 f'{self.__username}_identity_session',
             )
 
-    def __resolve_fqdn_from_username(self) -> str:
+    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},
+            json={'User': self.__username, 'Version': '1.0', 'PlatformTokenResponse': True, 'MfaRequestor': 'DeviceAgent'},
         )
         try:
             parsed_res: StartAuthResponse = StartAuthResponse.parse_raw(response.text)
@@ -381,19 +393,26 @@ def has_cache_record(cls, profile: ArkProfile, username: str, refresh_auth_allow
 
     @classmethod
     @cached(cache=LRUCache(maxsize=1024))
-    def is_idp_user(cls, username: str) -> bool:
+    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 = 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
 

ark_sdk_python/models/actions/ark_configure_action_consts.py

@@ -31,6 +31,7 @@
 }
 CONFIGURATION_ALLOWED_EMPTY_VALUES: Final[List[str]] = {
     'isp_identity_url',
+    'isp_identity_tenant_subdomain',
 }
 CONFIGURATION_AUTHENTICATORS_DEFAULTS: Final[Dict[str, str]] = {}
 CONFIGURATION_OVERRIDE_ALIASES: Final[Dict[str, str]] = {'region': 'Region', 'env': 'Environment'}

ark_sdk_python/models/auth/ark_auth_method.py

@@ -28,6 +28,12 @@ class IdentityArkAuthMethodSettings(ArkAuthMethodSettings):
     identity_url: Optional[str] = Field(
         description='Identity url to use for authentication instead of fqdn resolving', alias='Identity Url'
     )
+    identity_tenant_subdomain: Optional[str] = Field(
+        description='Identity security platform tenant subdomain, '
+        'for exmaple mytenant.cyberark.cloud would be subdomained to mytenant. '
+        'this will be used instead of fqdn resolving from the username',
+        alias='Identity Tenant Subdomain',
+    )
 
 
 class IdentityServiceUserArkAuthMethodSettings(ArkAuthMethodSettings):

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "ark-sdk-python"
-version = "1.0.4"
+version = "1.0.5"
 description='Official Ark SDK / CLI for CyberArk Identity Security Platform'
 authors = ["CyberArk <[email protected]>", "Ofir Iluz <[email protected]"]
 readme = "README.md"

tests/unit/auth/test_ark_isp_auth.py

@@ -48,7 +48,7 @@ def test_identity_auth_method(self, mocker: MockerFixture, auth_method: ArkAuthM
         tenant_fqdn_mock.assert_called_once()
         session_mock.return_value.post.assert_any_call(
             url='https://url.com/Security/StartAuthentication',
-            json={'User': '[email protected]', 'Version': '1.0', 'PlatformTokenResponse': True},
+            json={'User': '[email protected]', 'Version': '1.0', 'PlatformTokenResponse': True, 'MfaRequestor': 'DeviceAgent'},
         )
         session_mock.return_value.post.assert_any_call(
             url='https://url.com/Security/AdvanceAuthentication',
@@ -129,7 +129,7 @@ def test_identity_auth_method_caching(self, mocker: MockerFixture):
         pickle_dumps_mock.assert_called()
         session_mock.return_value.post.assert_any_call(
             url='https://url.com/Security/StartAuthentication',
-            json={'User': '[email protected]', 'Version': '1.0', 'PlatformTokenResponse': True},
+            json={'User': '[email protected]', 'Version': '1.0', 'PlatformTokenResponse': True, 'MfaRequestor': 'DeviceAgent'},
         )
         session_mock.return_value.post.assert_any_call(
             url='https://url.com/Security/AdvanceAuthentication',