Skip to content

ark_identity_fqdn_resolver

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