Skip to content

targetsets

ArkSIATargetSetsWorkspaceService

Bases: ArkService

Source code in ark_sdk_python/services/sia/workspaces/targetsets/ark_sia_target_sets_workspace_service.py 
 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
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
class ArkSIATargetSetsWorkspaceService(ArkService):
    def __init__(self, isp_auth: ArkISPAuth) -> None:
        super().__init__(isp_auth)
        self.__isp_auth = isp_auth
        self.__client: ArkISPServiceClient = ArkISPServiceClient.from_isp_auth(
            isp_auth=self.__isp_auth,
            service_name='dpa',
            refresh_connection_callback=self.__refresh_sia_auth,
        )

    def __refresh_sia_auth(self, client: ArkISPServiceClient) -> None:
        ArkISPServiceClient.refresh_client(client, self.__isp_auth)

    def add_target_set(self, add_target_set: ArkSIAAddTargetSet) -> ArkSIATargetSet:
        """
        Add a new target set

        Args:
            add_target_set (ArkSIAAddTargetSet): _description_

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkSIATargetSet: _description_
        """
        self._logger.info(f'Adding target set [{add_target_set.name}]')
        add_target_set_dict = add_target_set.model_dump()
        resp: Response = self.__client.post(TARGET_SETS_API, json=add_target_set_dict)
        if resp.status_code == HTTPStatus.CREATED:
            try:
                add_target_set_dict.update(resp.json()['target_set'])
                return ArkSIATargetSet(**add_target_set_dict)
            except (ValidationError, JSONDecodeError, KeyError) as ex:
                self._logger.exception(f'Failed to parse add target set response [{str(ex)}] - [{resp.text}]')
                raise ArkServiceException(f'Failed to parse add target set response [{str(ex)}]') from ex
        raise ArkServiceException(f'Failed to add target set [{resp.text}] - [{resp.status_code}]')

    def bulk_add_target_sets(self, bulk_add_target_sets: ArkSIABulkAddTargetSetsItem) -> ArkSIABulkTargetSetResponse:
        """
        Bulk add new target sets

        Args:
            bulk_add_target_sets (ArkSIABulkAddTargetSet): _description_

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkSIABulkTargetSetResponse: _description_
        """
        self._logger.info(f'Bulk adding target sets [{bulk_add_target_sets}]')
        resp: Response = self.__client.post(BULK_TARGET_SETS_API, json=bulk_add_target_sets.model_dump())
        if resp.status_code == HTTPStatus.MULTI_STATUS:
            try:
                return ArkSIABulkTargetSetResponse.model_validate(resp.json())
            except (ValidationError, JSONDecodeError, KeyError) as ex:
                self._logger.exception(f'Failed to parse bulk add target set response [{str(ex)}] - [{resp.text}]')
                raise ArkServiceException(f'Failed to parse bulk add target set response [{str(ex)}]') from ex
        raise ArkServiceException(f'Failed to bulk add target sets [{resp.text}] - [{resp.status_code}]')

    def delete_target_set(self, delete_target_set: ArkSIADeleteTargetSet) -> None:
        """
        Delete an existing target set

        Args:
            delete_target_set (ArkSIADeleteTargetSet): _description_

        Raises:
            ArkServiceException: _description_
        """
        self._logger.info(f'Deleting target set [{delete_target_set.name}]')
        resp: Response = self.__client.delete(TARGET_SET_API.format(target_name=delete_target_set.name))
        if resp.status_code != HTTPStatus.NO_CONTENT:
            raise ArkServiceException(f'Failed to delete target set [{resp.text}] - [{resp.status_code}]')

    def bulk_delete_target_sets(self, bulk_delete_target_sets: ArkSIABulkDeleteTargetSets) -> ArkSIABulkTargetSetResponse:
        """
        Bulk deletes existing target sets

        Args:
            bulk_delete_target_sets (ArkSIABulkDeleteTargetSet): _description_

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkSIABulkTargetSetResponse: _description_
        """
        self._logger.info(f'Bulk deleting target sets [{bulk_delete_target_sets}]')
        resp: Response = self.__client.delete(BULK_TARGET_SETS_API, json=bulk_delete_target_sets.target_sets)
        if resp.status_code == HTTPStatus.MULTI_STATUS:
            try:
                return ArkSIABulkTargetSetResponse.model_validate(resp.json())
            except (ValidationError, JSONDecodeError, KeyError) as ex:
                self._logger.exception(f'Failed to parse bulk delete target set response [{str(ex)}] - [{resp.text}]')
                raise ArkServiceException(f'Failed to parse bulk delete target set response [{str(ex)}]') from ex
        raise ArkServiceException(f'Failed to bulk delete target sets [{resp.text}] - [{resp.status_code}]')

    def update_target_set(self, update_target_set: ArkSIAUpdateTargetSet) -> ArkSIATargetSet:
        """
        Update an existing target set

        Args:
            update_target_set (ArkSIAUpdateTargetSet): _description_

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkSIATargetSet: _description_
        """
        self._logger.info(f'Updating target set [{update_target_set.name}]')
        update_target_set_dict = update_target_set.model_dump(exclude={'name', 'new_name'}, exclude_none=True)
        if update_target_set.new_name:
            update_target_set_dict["name"] = update_target_set.new_name
        resp: Response = self.__client.put(TARGET_SET_API.format(target_name=update_target_set.name), json=update_target_set_dict)
        if resp.status_code == HTTPStatus.OK:
            try:
                update_target_set_dict.update(resp.json()['target_set'])
                return ArkSIATargetSet(**update_target_set_dict)
            except (ValidationError, JSONDecodeError, KeyError) as ex:
                self._logger.exception(f'Failed to parse update target set response [{str(ex)}] - [{resp.text}]')
                raise ArkServiceException(f'Failed to parse update target set response [{str(ex)}]') from ex
        raise ArkServiceException(f'Failed to update target set [{resp.text}] - [{resp.status_code}]')

    def list_target_sets(self) -> List[ArkSIATargetSet]:
        """
        List all target sets

        Raises:
            ArkServiceException: _description_

        Returns:
            List[ArkSIATargetSet]: _description_
        """
        self._logger.info('Listing all target sets')
        resp: Response = self.__client.get(TARGET_SETS_API)
        if resp.status_code == HTTPStatus.OK:
            try:
                return [ArkSIATargetSet.model_validate(ts) for ts in resp.json()['target_sets']]
            except (ValidationError, JSONDecodeError, KeyError) as ex:
                self._logger.exception(f'Failed to parse list target sets response [{str(ex)}] - [{resp.text}]')
                raise ArkServiceException(f'Failed to parse list target sets response [{str(ex)}]') from ex
        raise ArkServiceException(f'Failed to list target sets [{resp.text}] - [{resp.status_code}]')

    def list_target_sets_by(self, target_sets_filter: ArkSIATargetSetsFilter) -> List[ArkSIATargetSet]:
        """
        List target sets by given filters

        Args:
            target_sets_filter (ArkSIATargetSetsFilter): _description_

        Returns:
            List[ArkSIATargetSet]: _description_
        """
        self._logger.info(f'Listing target sets by filters [{target_sets_filter}]')
        target_sets = self.list_target_sets()
        if target_sets_filter.name:
            target_sets = [t for t in target_sets if fnmatch(t.name, target_sets_filter.name)]
        if target_sets_filter.secret_type:
            target_sets = [t for t in target_sets if t.secret_type and t.secret_type == target_sets_filter.secret_type]
        return target_sets

    def target_set(self, get_target_set: ArkSIAGetTargetSet) -> ArkSIATargetSet:
        """
        Get specific target set

        Args:
            get_target_set (ArkSIAGetTargetSet): _description_

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkSIATargetSet: _description_
        """
        self._logger.info(f'Getting target set [{get_target_set.name}]')
        resp: Response = self.__client.get(TARGET_SET_API.format(target_name=get_target_set.name))
        if resp.status_code == HTTPStatus.OK:
            try:
                return ArkSIATargetSet(**resp.json()['target_set'])
            except (ValidationError, JSONDecodeError, KeyError) as ex:
                self._logger.exception(f'Failed to parse target set response [{str(ex)}] - [{resp.text}]')
                raise ArkServiceException(f'Failed to parse target set response [{str(ex)}]') from ex
        raise ArkServiceException(f'Failed to get target set [{resp.text}] - [{resp.status_code}]')

    def target_sets_stats(self) -> ArkSIATargetSetsStats:
        """
        Calculate stats about the target sets of the tenant

        Returns:
            ArkSIATargetSetsStats: _description_
        """
        self._logger.info('Calculating target sets stats')
        target_sets = self.list_target_sets()
        target_sets_stats = ArkSIATargetSetsStats.model_construct()
        target_sets_stats.target_sets_count = len(target_sets)

        # Get target_sets per secret type
        secret_types: Set[ArkSIAVMSecretType] = {d.secret_type for d in target_sets if d.secret_type}
        target_sets_stats.target_sets_count_per_secret_type = {
            st: len([d for d in target_sets if d.secret_type and d.secret_type == st]) for st in secret_types
        }

        return target_sets_stats

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

add_target_set(add_target_set)

Add a new target set

Parameters:

Name Type Description Default
add_target_set ArkSIAAddTargetSet

description

 required

Raises:

Type Description
ArkServiceException

description

Returns:

Name Type Description
ArkSIATargetSet ArkSIATargetSet

description
Source code in ark_sdk_python/services/sia/workspaces/targetsets/ark_sia_target_sets_workspace_service.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def add_target_set(self, add_target_set: ArkSIAAddTargetSet) -> ArkSIATargetSet:
    """
    Add a new target set

    Args:
        add_target_set (ArkSIAAddTargetSet): _description_

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkSIATargetSet: _description_
    """
    self._logger.info(f'Adding target set [{add_target_set.name}]')
    add_target_set_dict = add_target_set.model_dump()
    resp: Response = self.__client.post(TARGET_SETS_API, json=add_target_set_dict)
    if resp.status_code == HTTPStatus.CREATED:
        try:
            add_target_set_dict.update(resp.json()['target_set'])
            return ArkSIATargetSet(**add_target_set_dict)
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse add target set response [{str(ex)}] - [{resp.text}]')
            raise ArkServiceException(f'Failed to parse add target set response [{str(ex)}]') from ex
    raise ArkServiceException(f'Failed to add target set [{resp.text}] - [{resp.status_code}]')

bulk_add_target_sets(bulk_add_target_sets)

Bulk add new target sets

Parameters:

Name Type Description Default
bulk_add_target_sets ArkSIABulkAddTargetSet

description

 required

Raises:

Type Description
ArkServiceException

description

Returns:

Name Type Description
ArkSIABulkTargetSetResponse ArkSIABulkTargetSetResponse

description
Source code in ark_sdk_python/services/sia/workspaces/targetsets/ark_sia_target_sets_workspace_service.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def bulk_add_target_sets(self, bulk_add_target_sets: ArkSIABulkAddTargetSetsItem) -> ArkSIABulkTargetSetResponse:
    """
    Bulk add new target sets

    Args:
        bulk_add_target_sets (ArkSIABulkAddTargetSet): _description_

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkSIABulkTargetSetResponse: _description_
    """
    self._logger.info(f'Bulk adding target sets [{bulk_add_target_sets}]')
    resp: Response = self.__client.post(BULK_TARGET_SETS_API, json=bulk_add_target_sets.model_dump())
    if resp.status_code == HTTPStatus.MULTI_STATUS:
        try:
            return ArkSIABulkTargetSetResponse.model_validate(resp.json())
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse bulk add target set response [{str(ex)}] - [{resp.text}]')
            raise ArkServiceException(f'Failed to parse bulk add target set response [{str(ex)}]') from ex
    raise ArkServiceException(f'Failed to bulk add target sets [{resp.text}] - [{resp.status_code}]')

bulk_delete_target_sets(bulk_delete_target_sets)

Bulk deletes existing target sets

Parameters:

Name Type Description Default
bulk_delete_target_sets ArkSIABulkDeleteTargetSet

description

 required

Raises:

Type Description
ArkServiceException

description

Returns:

Name Type Description
ArkSIABulkTargetSetResponse ArkSIABulkTargetSetResponse

description
Source code in ark_sdk_python/services/sia/workspaces/targetsets/ark_sia_target_sets_workspace_service.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def bulk_delete_target_sets(self, bulk_delete_target_sets: ArkSIABulkDeleteTargetSets) -> ArkSIABulkTargetSetResponse:
    """
    Bulk deletes existing target sets

    Args:
        bulk_delete_target_sets (ArkSIABulkDeleteTargetSet): _description_

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkSIABulkTargetSetResponse: _description_
    """
    self._logger.info(f'Bulk deleting target sets [{bulk_delete_target_sets}]')
    resp: Response = self.__client.delete(BULK_TARGET_SETS_API, json=bulk_delete_target_sets.target_sets)
    if resp.status_code == HTTPStatus.MULTI_STATUS:
        try:
            return ArkSIABulkTargetSetResponse.model_validate(resp.json())
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse bulk delete target set response [{str(ex)}] - [{resp.text}]')
            raise ArkServiceException(f'Failed to parse bulk delete target set response [{str(ex)}]') from ex
    raise ArkServiceException(f'Failed to bulk delete target sets [{resp.text}] - [{resp.status_code}]')

delete_target_set(delete_target_set)

Delete an existing target set

Parameters:

Name Type Description Default
delete_target_set ArkSIADeleteTargetSet

description

 required

Raises:

Type Description
ArkServiceException

description
Source code in ark_sdk_python/services/sia/workspaces/targetsets/ark_sia_target_sets_workspace_service.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def delete_target_set(self, delete_target_set: ArkSIADeleteTargetSet) -> None:
    """
    Delete an existing target set

    Args:
        delete_target_set (ArkSIADeleteTargetSet): _description_

    Raises:
        ArkServiceException: _description_
    """
    self._logger.info(f'Deleting target set [{delete_target_set.name}]')
    resp: Response = self.__client.delete(TARGET_SET_API.format(target_name=delete_target_set.name))
    if resp.status_code != HTTPStatus.NO_CONTENT:
        raise ArkServiceException(f'Failed to delete target set [{resp.text}] - [{resp.status_code}]')

list_target_sets()

List all target sets

Raises:

Type Description
ArkServiceException

description

Returns:

Type Description
List[ArkSIATargetSet]

List[ArkSIATargetSet]: description
Source code in ark_sdk_python/services/sia/workspaces/targetsets/ark_sia_target_sets_workspace_service.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def list_target_sets(self) -> List[ArkSIATargetSet]:
    """
    List all target sets

    Raises:
        ArkServiceException: _description_

    Returns:
        List[ArkSIATargetSet]: _description_
    """
    self._logger.info('Listing all target sets')
    resp: Response = self.__client.get(TARGET_SETS_API)
    if resp.status_code == HTTPStatus.OK:
        try:
            return [ArkSIATargetSet.model_validate(ts) for ts in resp.json()['target_sets']]
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse list target sets response [{str(ex)}] - [{resp.text}]')
            raise ArkServiceException(f'Failed to parse list target sets response [{str(ex)}]') from ex
    raise ArkServiceException(f'Failed to list target sets [{resp.text}] - [{resp.status_code}]')

list_target_sets_by(target_sets_filter)

List target sets by given filters

Parameters:

Name Type Description Default
target_sets_filter ArkSIATargetSetsFilter

description

 required

Returns:

Type Description
List[ArkSIATargetSet]

List[ArkSIATargetSet]: description
Source code in ark_sdk_python/services/sia/workspaces/targetsets/ark_sia_target_sets_workspace_service.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def list_target_sets_by(self, target_sets_filter: ArkSIATargetSetsFilter) -> List[ArkSIATargetSet]:
    """
    List target sets by given filters

    Args:
        target_sets_filter (ArkSIATargetSetsFilter): _description_

    Returns:
        List[ArkSIATargetSet]: _description_
    """
    self._logger.info(f'Listing target sets by filters [{target_sets_filter}]')
    target_sets = self.list_target_sets()
    if target_sets_filter.name:
        target_sets = [t for t in target_sets if fnmatch(t.name, target_sets_filter.name)]
    if target_sets_filter.secret_type:
        target_sets = [t for t in target_sets if t.secret_type and t.secret_type == target_sets_filter.secret_type]
    return target_sets

target_set(get_target_set)

Get specific target set

Parameters:

Name Type Description Default
get_target_set ArkSIAGetTargetSet

description

 required

Raises:

Type Description
ArkServiceException

description

Returns:

Name Type Description
ArkSIATargetSet ArkSIATargetSet

description
Source code in ark_sdk_python/services/sia/workspaces/targetsets/ark_sia_target_sets_workspace_service.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def target_set(self, get_target_set: ArkSIAGetTargetSet) -> ArkSIATargetSet:
    """
    Get specific target set

    Args:
        get_target_set (ArkSIAGetTargetSet): _description_

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkSIATargetSet: _description_
    """
    self._logger.info(f'Getting target set [{get_target_set.name}]')
    resp: Response = self.__client.get(TARGET_SET_API.format(target_name=get_target_set.name))
    if resp.status_code == HTTPStatus.OK:
        try:
            return ArkSIATargetSet(**resp.json()['target_set'])
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse target set response [{str(ex)}] - [{resp.text}]')
            raise ArkServiceException(f'Failed to parse target set response [{str(ex)}]') from ex
    raise ArkServiceException(f'Failed to get target set [{resp.text}] - [{resp.status_code}]')

target_sets_stats()

Calculate stats about the target sets of the tenant

Returns:

Name Type Description
ArkSIATargetSetsStats ArkSIATargetSetsStats

description
Source code in ark_sdk_python/services/sia/workspaces/targetsets/ark_sia_target_sets_workspace_service.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
def target_sets_stats(self) -> ArkSIATargetSetsStats:
    """
    Calculate stats about the target sets of the tenant

    Returns:
        ArkSIATargetSetsStats: _description_
    """
    self._logger.info('Calculating target sets stats')
    target_sets = self.list_target_sets()
    target_sets_stats = ArkSIATargetSetsStats.model_construct()
    target_sets_stats.target_sets_count = len(target_sets)

    # Get target_sets per secret type
    secret_types: Set[ArkSIAVMSecretType] = {d.secret_type for d in target_sets if d.secret_type}
    target_sets_stats.target_sets_count_per_secret_type = {
        st: len([d for d in target_sets if d.secret_type and d.secret_type == st]) for st in secret_types
    }

    return target_sets_stats

update_target_set(update_target_set)

Update an existing target set

Parameters:

Name Type Description Default
update_target_set ArkSIAUpdateTargetSet

description

 required

Raises:

Type Description
ArkServiceException

description

Returns:

Name Type Description
ArkSIATargetSet ArkSIATargetSet

description
Source code in ark_sdk_python/services/sia/workspaces/targetsets/ark_sia_target_sets_workspace_service.py 
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
def update_target_set(self, update_target_set: ArkSIAUpdateTargetSet) -> ArkSIATargetSet:
    """
    Update an existing target set

    Args:
        update_target_set (ArkSIAUpdateTargetSet): _description_

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkSIATargetSet: _description_
    """
    self._logger.info(f'Updating target set [{update_target_set.name}]')
    update_target_set_dict = update_target_set.model_dump(exclude={'name', 'new_name'}, exclude_none=True)
    if update_target_set.new_name:
        update_target_set_dict["name"] = update_target_set.new_name
    resp: Response = self.__client.put(TARGET_SET_API.format(target_name=update_target_set.name), json=update_target_set_dict)
    if resp.status_code == HTTPStatus.OK:
        try:
            update_target_set_dict.update(resp.json()['target_set'])
            return ArkSIATargetSet(**update_target_set_dict)
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse update target set response [{str(ex)}] - [{resp.text}]')
            raise ArkServiceException(f'Failed to parse update target set response [{str(ex)}]') from ex
    raise ArkServiceException(f'Failed to update target set [{resp.text}] - [{resp.status_code}]')