Skip to content

sca

ArkUAPSCAService

Bases: ArkUAPBaseService

Source code in ark_sdk_python/services/uap/sca/ark_uap_sca_service.py 
 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
class ArkUAPSCAService(ArkUAPBaseService):

    __MAX_ACTIVE_STATUS_RETRY_COUNT: Final[int] = 10

    def add_policy(self, add_policy: ArkUAPSCACloudConsoleAccessPolicy) -> ArkUAPSCACloudConsoleAccessPolicy:
        """
        Adds a new policy with the given information

        Args:
            add_policy (ArkUAPSCACloudConsoleAccessPolicy): The policy to be added.

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkUAPResponse: _description_
        """
        add_policy_response: ArkUAPResponse = self._base_add_policy(
            add_policy.serialize_model(by_alias=True, exclude_none=True),
            policy_name=add_policy.metadata.name,
        )
        retry_count = 0
        while True:
            policy = self.policy(ArkUAPGetPolicyRequest(policy_id=add_policy_response.policy_id))
            if policy.metadata.status == ArkUAPStatusType.ACTIVE:
                break
            if policy.metadata.status == ArkUAPStatusType.ERROR:
                raise ArkServiceException(f'Policy [{add_policy_response.policy_id}] is in error state')
            if retry_count >= ArkUAPSCAService.__MAX_ACTIVE_STATUS_RETRY_COUNT:
                self._logger.warning(
                    f'Policy [{add_policy_response.policy_id}] is not active after 10 retries, '
                    f'might indicate an issue, moving on regardless'
                )
                break
            retry_count += 1
        return self.policy(ArkUAPGetPolicyRequest(policy_id=add_policy_response.policy_id))

    def policy(self, policy_request: ArkUAPGetPolicyRequest) -> ArkUAPSCACloudConsoleAccessPolicy:
        """
        get a policy by id

        Args:
            policy_request (ArkUAPGetPolicyRequest): _description_

        Raises:
            ArkServiceException: _description_


        Returns:
            ArkUAPSCACloudConsoleAccessPolicy: the request policy.
        """
        return ArkUAPSCACloudConsoleAccessPolicy.deserialize_model(self._base_policy(policy_request=policy_request))

    def update_policy(self, update_policy: ArkUAPSCACloudConsoleAccessPolicy) -> ArkUAPSCACloudConsoleAccessPolicy:
        """
        Edits an existing policy with the given information.

        Args:
            update_policy (ArkUAPSCACloudConsoleAccessPolicy): The policy to be edited.

        Raises:
            ArkServiceException: _description_

        """
        self._base_update_policy(
            update_policy=update_policy.serialize_model(by_alias=True, exclude_none=True),
            policy_id=update_policy.metadata.policy_id,
        )
        retry_count = 0
        while True:
            policy = self.policy(ArkUAPGetPolicyRequest(policy_id=update_policy.metadata.policy_id))
            if policy.metadata.status == ArkUAPStatusType.ACTIVE:
                break
            if policy.metadata.status == ArkUAPStatusType.ERROR:
                raise ArkServiceException(f'Policy [{update_policy.metadata.policy_id}] is in error state')
            if retry_count >= ArkUAPSCAService.__MAX_ACTIVE_STATUS_RETRY_COUNT:
                self._logger.warning(
                    f'Policy [{update_policy.metadata.policy_id}] is not active after 10 retries, '
                    f'might indicate an issue, moving on regardless'
                )
                break
            retry_count += 1
        return self.policy(ArkUAPGetPolicyRequest(policy_id=update_policy.metadata.policy_id))

    def list_policies(self) -> ArkUAPPolicyResultsResponse:
        """
         Retrieves all policies.

        Raises:
         ArkServiceException: If failed to get all policies.

         Returns:
             ArkUAPPolicyResultsResponse: The response containing all policies.
        """
        return self._base_list_policies(ArkUAPFilters(target_category=[ArkCategoryType.CLOUD_CONSOLE]))

    def policies_stats(self) -> ArkUAPPoliciesStats:
        """
        Calculates policies statistics

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkUAPPoliciesStats: Summary of policies by status and provider.
        """
        return self._base_policies_stats(ArkUAPFilters(target_category=[ArkCategoryType.CLOUD_CONSOLE]))

    def list_policies_by(self, policies_filter: ArkUAPSCAFilters) -> ArkUAPPolicyResultsResponse:
        """
        Retrieves policies based on the provided filters.

        Args:
            policies_filter (ArkUAPSCAFilters): The filters to apply when retrieving policies.

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkUAPPolicyResultsResponse: The response containing the filtered policies.
        """
        return self._base_list_policies(policies_filter)

    @staticmethod
    @overrides
    def service_config() -> ArkServiceConfig:
        return SERVICE_CONFIG

add_policy(add_policy)

Adds a new policy with the given information

Parameters:

Name Type Description Default
add_policy ArkUAPSCACloudConsoleAccessPolicy

The policy to be added.

 required

Raises:

Type Description
ArkServiceException

description

Returns:

Name Type Description
ArkUAPResponse ArkUAPSCACloudConsoleAccessPolicy

description
Source code in ark_sdk_python/services/uap/sca/ark_uap_sca_service.py 
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
def add_policy(self, add_policy: ArkUAPSCACloudConsoleAccessPolicy) -> ArkUAPSCACloudConsoleAccessPolicy:
    """
    Adds a new policy with the given information

    Args:
        add_policy (ArkUAPSCACloudConsoleAccessPolicy): The policy to be added.

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkUAPResponse: _description_
    """
    add_policy_response: ArkUAPResponse = self._base_add_policy(
        add_policy.serialize_model(by_alias=True, exclude_none=True),
        policy_name=add_policy.metadata.name,
    )
    retry_count = 0
    while True:
        policy = self.policy(ArkUAPGetPolicyRequest(policy_id=add_policy_response.policy_id))
        if policy.metadata.status == ArkUAPStatusType.ACTIVE:
            break
        if policy.metadata.status == ArkUAPStatusType.ERROR:
            raise ArkServiceException(f'Policy [{add_policy_response.policy_id}] is in error state')
        if retry_count >= ArkUAPSCAService.__MAX_ACTIVE_STATUS_RETRY_COUNT:
            self._logger.warning(
                f'Policy [{add_policy_response.policy_id}] is not active after 10 retries, '
                f'might indicate an issue, moving on regardless'
            )
            break
        retry_count += 1
    return self.policy(ArkUAPGetPolicyRequest(policy_id=add_policy_response.policy_id))

list_policies()

Retrieves all policies.

Raises:

Type Description
ArkServiceException

If failed to get all policies.
Returns

ArkUAPPolicyResultsResponse: The response containing all policies.
Source code in ark_sdk_python/services/uap/sca/ark_uap_sca_service.py 
107
108
109
110
111
112
113
114
115
116
117
def list_policies(self) -> ArkUAPPolicyResultsResponse:
    """
     Retrieves all policies.

    Raises:
     ArkServiceException: If failed to get all policies.

     Returns:
         ArkUAPPolicyResultsResponse: The response containing all policies.
    """
    return self._base_list_policies(ArkUAPFilters(target_category=[ArkCategoryType.CLOUD_CONSOLE]))

list_policies_by(policies_filter)

Retrieves policies based on the provided filters.

Parameters:

Name Type Description Default
policies_filter ArkUAPSCAFilters

The filters to apply when retrieving policies.

 required

Raises:

Type Description
ArkServiceException

description

Returns:

Name Type Description
ArkUAPPolicyResultsResponse ArkUAPPolicyResultsResponse

The response containing the filtered policies.
Source code in ark_sdk_python/services/uap/sca/ark_uap_sca_service.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def list_policies_by(self, policies_filter: ArkUAPSCAFilters) -> ArkUAPPolicyResultsResponse:
    """
    Retrieves policies based on the provided filters.

    Args:
        policies_filter (ArkUAPSCAFilters): The filters to apply when retrieving policies.

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkUAPPolicyResultsResponse: The response containing the filtered policies.
    """
    return self._base_list_policies(policies_filter)

policies_stats()

Calculates policies statistics

Raises:

Type Description
ArkServiceException

description

Returns:

Name Type Description
ArkUAPPoliciesStats ArkUAPPoliciesStats

Summary of policies by status and provider.
Source code in ark_sdk_python/services/uap/sca/ark_uap_sca_service.py 
119
120
121
122
123
124
125
126
127
128
129
def policies_stats(self) -> ArkUAPPoliciesStats:
    """
    Calculates policies statistics

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkUAPPoliciesStats: Summary of policies by status and provider.
    """
    return self._base_policies_stats(ArkUAPFilters(target_category=[ArkCategoryType.CLOUD_CONSOLE]))

policy(policy_request)

get a policy by id

Parameters:

Name Type Description Default
policy_request ArkUAPGetPolicyRequest

description

 required

Raises:

Type Description
ArkServiceException

description

Returns:

Name Type Description
ArkUAPSCACloudConsoleAccessPolicy ArkUAPSCACloudConsoleAccessPolicy

the request policy.
Source code in ark_sdk_python/services/uap/sca/ark_uap_sca_service.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def policy(self, policy_request: ArkUAPGetPolicyRequest) -> ArkUAPSCACloudConsoleAccessPolicy:
    """
    get a policy by id

    Args:
        policy_request (ArkUAPGetPolicyRequest): _description_

    Raises:
        ArkServiceException: _description_


    Returns:
        ArkUAPSCACloudConsoleAccessPolicy: the request policy.
    """
    return ArkUAPSCACloudConsoleAccessPolicy.deserialize_model(self._base_policy(policy_request=policy_request))

update_policy(update_policy)

Edits an existing policy with the given information.

Parameters:

Name Type Description Default
update_policy ArkUAPSCACloudConsoleAccessPolicy

The policy to be edited.

 required

Raises:

Type Description
ArkServiceException

description
Source code in ark_sdk_python/services/uap/sca/ark_uap_sca_service.py 
 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
def update_policy(self, update_policy: ArkUAPSCACloudConsoleAccessPolicy) -> ArkUAPSCACloudConsoleAccessPolicy:
    """
    Edits an existing policy with the given information.

    Args:
        update_policy (ArkUAPSCACloudConsoleAccessPolicy): The policy to be edited.

    Raises:
        ArkServiceException: _description_

    """
    self._base_update_policy(
        update_policy=update_policy.serialize_model(by_alias=True, exclude_none=True),
        policy_id=update_policy.metadata.policy_id,
    )
    retry_count = 0
    while True:
        policy = self.policy(ArkUAPGetPolicyRequest(policy_id=update_policy.metadata.policy_id))
        if policy.metadata.status == ArkUAPStatusType.ACTIVE:
            break
        if policy.metadata.status == ArkUAPStatusType.ERROR:
            raise ArkServiceException(f'Policy [{update_policy.metadata.policy_id}] is in error state')
        if retry_count >= ArkUAPSCAService.__MAX_ACTIVE_STATUS_RETRY_COUNT:
            self._logger.warning(
                f'Policy [{update_policy.metadata.policy_id}] is not active after 10 retries, '
                f'might indicate an issue, moving on regardless'
            )
            break
        retry_count += 1
    return self.policy(ArkUAPGetPolicyRequest(policy_id=update_policy.metadata.policy_id))