accounts

ArkPCloudAccountsService

Bases: ArkPCloudBaseService

Source code in ark_sdk_python/services/pcloud/accounts/ark_pcloud_accounts_service.py

class ArkPCloudAccountsService(ArkPCloudBaseService):
    def __list_accounts_with_filters(
        self,
        search: Optional[str] = None,
        search_type: Optional[str] = None,
        sort: Optional[str] = None,
        offset: Optional[int] = None,
        limit: Optional[int] = None,
        safe_name: Optional[str] = None,
    ) -> Iterator[ArkPCloudAccountsPage]:
        query = {}
        if search:
            query['search'] = search
        if search_type:
            query['searchType'] = search_type
        if sort:
            query['sort'] = sort
        if offset:
            query['offset'] = offset
        if limit:
            query['limit'] = limit
        if safe_name:
            query['filter'] = f'safeName eq {safe_name}'
        while True:
            resp: Response = self._client.get(ACCOUNTS_URL, params=query)
            if resp.status_code == HTTPStatus.OK:
                try:
                    result = resp.json()
                    accounts = parse_obj_as(List[ArkPCloudAccount], result['value'])
                    yield ArkPCloudAccountsPage(items=accounts)
                    if 'nextLink' in result:
                        query = parse_qs(urlparse(result['nextLink']).query)
                    else:
                        break
                except (ValidationError, JSONDecodeError, KeyError) as ex:
                    self._logger.exception(f'Failed to parse list accounts response [{str(ex)}] - [{resp.text}]')
                    raise ArkServiceException(f'Failed to parse list accounts response [{str(ex)}]') from ex
            else:
                raise ArkServiceException(f'Failed to list accounts [{resp.text}] - [{resp.status_code}]')

    def list_accounts(self) -> Iterator[ArkPCloudAccountsPage]:
        """
        Yields all visible accounts to the logged in user as pages of accounts
        https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/SDK/GetAccounts.htm

        Yields:
            Iterator[ArkPCloudAccountsPage]: _description_
        """
        self._logger.info('Listing all accounts')
        yield from self.__list_accounts_with_filters()

    def list_accounts_by(self, accounts_filter: ArkPCloudAccountsFilter) -> Iterator[ArkPCloudAccountsPage]:
        """
        Yields visible accounts to the logged in user by filters as pages of accounts
        https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/SDK/GetAccounts.htm

        Args:
            accounts_filter (ArkPCloudAccountsFilter): _description_

        Yields:
            Iterator[ArkPCloudAccountsPage]: _description_
        """
        self._logger.info(f'Listing accounts by filters [{accounts_filter}]')
        yield from self.__list_accounts_with_filters(
            accounts_filter.search,
            accounts_filter.search_type,
            accounts_filter.sort,
            accounts_filter.offset,
            accounts_filter.limit,
            accounts_filter.safe_name,
        )

    def list_account_secret_versions(
        self, list_account_secret_versions: ArkPCloudListAccountSecretVersions
    ) -> List[ArkPCloudAccountSecretVersion]:
        """
        Lists the given account secret versions
        https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/SDK/Secrets-Get-versions.htm

        Args:
            list_account_secret_versions (ArkPCloudListAccountSecretVersions): _description_

        Raises:
            ArkServiceException: _description_

        Returns:
            List[ArkPCloudAccountSecretVersion]: _description_
        """
        self._logger.info(f'Listing account [{list_account_secret_versions.account_id}] secret versions')
        resp: Response = self._client.get(ACCOUNT_SECRET_VERSIONS.format(account_id=list_account_secret_versions.account_id))
        if resp.status_code == HTTPStatus.OK:
            try:
                return parse_obj_as(List[ArkPCloudAccountSecretVersion], resp.json()['versions'])
            except (ValidationError, JSONDecodeError, KeyError) as ex:
                self._logger.exception(f'Failed to parse list account secret versions response [{str(ex)}] - [{resp.text}]')
                raise ArkServiceException(f'Failed to parse list account secret versions response [{str(ex)}]') from ex
        raise ArkServiceException(f'Failed to list account secret versions [{resp.text}] - [{resp.status_code}]')

    def generate_account_credentials(
        self, generate_account_credentials: ArkPCloudGenerateAccountCredentials
    ) -> ArkPCloudAccountCredentials:
        """
        Generate a new random password for an existing account with policy restrictions
        https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/SDK/Secrets-Generate-Password.htm

        Args:
            generate_account_credentials (ArkPCloudGenerateAccountCredentials): _description_

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkPCloudAccountCredentials: _description_
        """
        self._logger.info(f'Generating new password for account [{generate_account_credentials.account_id}]')
        resp: Response = self._client.post(GENERATE_ACCOUNT_CREDENTIALS.format(account_id=generate_account_credentials.account_id))
        if resp.status_code == HTTPStatus.OK:
            try:
                return ArkPCloudAccountCredentials(account_id=generate_account_credentials.account_id, password=resp.json()['password'])
            except (ValidationError, JSONDecodeError, KeyError) as ex:
                self._logger.exception(f'Failed to parse genereate account credentials response [{str(ex)}] - [{resp.text}]')
                raise ArkServiceException(f'Failed to parse genereate account credentials response [{str(ex)}]') from ex
        raise ArkServiceException(f'Failed to generate password for account [{resp.text}] - [{resp.status_code}]')

    def verify_account_credentials(self, verify_account_credentials: ArkPCloudVerifyAccountCredentials) -> None:
        """
        Marks the account for password verification by CPM
        https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/Verify-credentials-v9-10.htm

        Args:
            verify_account_credentials (ArkPCloudVerifyAccountCredentials): _description_

        Raises:
            ArkServiceException: _description_
        """
        self._logger.info(f'Marking account [{verify_account_credentials.account_id}] for verification')
        resp: Response = self._client.post(VERIFY_ACCOUNT_CREDENTIALS.format(account_id=verify_account_credentials.account_id))
        if resp.status_code != HTTPStatus.OK:
            raise ArkServiceException(f'Failed to mark account for password verification [{resp.text}] - [{resp.status_code}]')

    def change_account_credentials(self, change_account_credentials: ArkPCloudChangeAccountCredentials) -> None:
        """
        Marks the account for password changing immediately by CPM
        https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/Change-credentials-immediately.htm

        Args:
            change_account_credentials (ArkPCloudChangeAccountCredentials): _description_

        Raises:
            ArkServiceException: _description_
        """
        self._logger.info(f'Marking account [{change_account_credentials.account_id}] for changing credentials immediately')
        resp: Response = self._client.post(CHANGE_ACCOUNT_CREDENTIALS.format(account_id=change_account_credentials.account_id))
        if resp.status_code != HTTPStatus.OK:
            raise ArkServiceException(f'Failed to mark account for changing credentials immediately [{resp.text}] - [{resp.status_code}]')

    def set_account_next_credentials(self, set_account_next_credentials: ArkPCloudSetAccountNextCredentials) -> None:
        """
        Marks the account to have its password changed to the given one via CPM
        https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/SetNextPassword.htm

        Args:
            set_account_next_credentials (ArkPCloudSetAccountNextCredentials): _description_

        Raises:
            ArkServiceException: _description_
        """
        self._logger.info(f'Marking account [{set_account_next_credentials.account_id}] for changing credentials for the given password')
        resp: Response = self._client.post(
            SET_ACCOUNT_NEXT_CREDENTIALS.format(account_id=set_account_next_credentials.account_id),
            json=set_account_next_credentials.dict(exclude={'account_id'}, by_alias=True),
        )
        if resp.status_code != HTTPStatus.OK:
            raise ArkServiceException(f'Failed to mark account for changing credentials next password [{resp.text}] - [{resp.status_code}]')

    def update_account_credentials_in_vault(self, update_account_credentials_in_vault: ArkPCloudUpdateAccountCredentialsInVault) -> None:
        """
        Updates the account credentials only in the vault without changing it on the machine itself
        https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/ChangeCredentialsInVault.htm

        Args:
            update_account_credentials_in_vault (ArkPCloudUpdateAccountCredentialsInVault): _description_

        Raises:
            ArkServiceException: _description_
        """
        self._logger.info(f'Updates account [{update_account_credentials_in_vault.account_id}] vault credentials')
        resp: Response = self._client.post(
            UPDATE_ACCOUNT_CREDENTIALS_IN_VAULT.format(account_id=update_account_credentials_in_vault.account_id),
            json=update_account_credentials_in_vault.dict(exclude={'account_id'}, by_alias=True),
        )
        if resp.status_code != HTTPStatus.OK:
            raise ArkServiceException(f'Failed to update credentials in vault for account [{resp.text}] - [{resp.status_code}]')

    def reconcile_account_credentials(self, reconcile_account_credentials: ArkPCloudReconcileAccountCredentials) -> None:
        """
        Marks the account for reconcilation
        https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/Reconcile-account.htm

        Args:
            reconcile_account_credentials (ArkPCloudReconcileAccountCredentials): _description_

        Raises:
            ArkServiceException: _description_
        """
        self._logger.info(f'Marking account [{reconcile_account_credentials.account_id}] for reconcilation')
        resp: Response = self._client.post(RECONCILE_ACCOUNT_CREDENTIALS.format(account_id=reconcile_account_credentials.account_id))
        if resp.status_code != HTTPStatus.OK:
            raise ArkServiceException(f'Failed to reconcile account credentials [{resp.text}] - [{resp.status_code}]')

    def account(self, get_account: ArkPCloudGetAccount) -> ArkPCloudAccount:
        """
        Retrieves the account by id
        https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/Get%20Account%20Details.htm?

        Args:
            get_account (ArkPCloudGetAccount): _description_

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkPCloudAccount: _description_
        """
        self._logger.info(f'Retrieving account by id [{get_account.account_id}]')
        resp: Response = self._client.get(ACCOUNT_URL.format(account_id=get_account.account_id))
        if resp.status_code == HTTPStatus.OK:
            try:
                return ArkPCloudAccount.parse_obj(resp.json())
            except (ValidationError, JSONDecodeError, KeyError) as ex:
                self._logger.exception(f'Failed to parse account response [{str(ex)}] - [{resp.text}]')
                raise ArkServiceException(f'Failed to parse account response [{str(ex)}]') from ex
        raise ArkServiceException(f'Failed to retrieve account [{resp.text}] - [{resp.status_code}]')

    def account_credentials(self, get_account_credentials: ArkPCloudGetAccountCredentials) -> ArkPCloudAccountCredentials:
        """
        Retrieves the account credentials
        https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/GetPasswordValueV10.htm?

        Args:
            get_account_credentials (ArkPCloudGetAccountCredentials): _description_

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkPCloudAccountCredentials: _description_
        """
        self._logger.info(f'Retrieving account password for details [{get_account_credentials}]')
        body = {
            k.replace('_', '').title(): v
            for k, v in get_account_credentials.dict(exclude={'account_id', 'reason'}, exclude_none=True).items()
        }
        if get_account_credentials.reason:
            body['reason'] = get_account_credentials.reason
        resp: Response = self._client.post(RETRIEVE_ACCOUNT_CREDENTIALS.format(account_id=get_account_credentials.account_id), json=body)
        if resp.status_code == HTTPStatus.OK:
            return ArkPCloudAccountCredentials(
                account_id=get_account_credentials.account_id, password=resp.text[1:-1]  # Remove leading and trailing quotes
            )
        raise ArkServiceException(f'Failed to retrieve account credentials [{resp.text}] - [{resp.status_code}]')

    def add_account(self, add_account: ArkPCloudAddAccount) -> ArkPCloudAccount:
        """
        Adds a new account with given details
        https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/Add%20Account%20v10.htm?

        Args:
            add_account (ArkPCloudAddAccount): _description_

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkPCloudAccount: _description_
        """
        self._logger.info('Adding new account')
        if add_account.remote_machines_access:
            if (
                add_account.remote_machines_access.access_restricted_to_remote_machines
                and not add_account.remote_machines_access.remote_machines
            ):
                add_account.remote_machines_access = None
            elif not add_account.remote_machines_access.access_restricted_to_remote_machines:
                add_account.remote_machines_access.remote_machines = None

            elif add_account.remote_machines_access.remote_machines:
                add_account.remote_machines_access.remote_machines = ','.join(add_account.remote_machines_access.remote_machines)

        resp: Response = self._client.post(ACCOUNTS_URL, data=add_account.json(by_alias=True, exclude_none=True, exclude_defaults=True))
        if resp.status_code == HTTPStatus.CREATED:
            try:
                return ArkPCloudAccount.parse_obj(resp.json())
            except (ValidationError, JSONDecodeError) as ex:
                self._logger.exception(f'Failed to parse add account response [{str(ex)}] - [{resp.text}]')
                raise ArkServiceException(f'Failed to parse add account response [{str(ex)}]') from ex
        raise ArkServiceException(f'Failed to add account [{resp.text}] - [{resp.status_code}]')

    def update_account(self, update_account: ArkPCloudUpdateAccount) -> ArkPCloudAccount:
        """
        Updates an existing account with new details
        https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/SDK/UpdateAccount%20v10.htm

        Args:
            update_account (ArkPCloudUpdateAccount): _description_

        Raises:
            ArkServiceException: _description_

        Returns:
            ArkPCloudAccount: _description_
        """
        self._logger.info(f'Updating account [{update_account.account_id}]')
        if update_account.remote_machines_access and not update_account.remote_machines_access.remote_machines:
            update_account.remote_machines_access = None
        operations = []
        for key, val in update_account.dict(exclude={'account_id'}, exclude_none=True, by_alias=True, exclude_defaults=True):
            operations.append({'op': 'replace', 'path': f'/{key}', 'value': val})
        resp: Response = self._client.put(ACCOUNT_URL.format(account_id=update_account.account_id), json=operations)
        if resp.status_code == HTTPStatus.OK:
            try:
                return ArkPCloudAccount.parse_obj(resp.json())
            except (ValidationError, JSONDecodeError) as ex:
                self._logger.exception(f'Failed to parse update account response [{str(ex)}] - [{resp.text}]')
                raise ArkServiceException(f'Failed to parse update account response [{str(ex)}]') from ex
        raise ArkServiceException(f'Failed to update account [{resp.text}] - [{resp.status_code}]')

    def delete_account(self, delete_account: ArkPCloudDeleteAccount) -> None:
        """
        Deletes an account by given id
        https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/Delete%20Account.htm

        Args:
            delete_account (ArkPCloudDeleteAccount): _description_

        Raises:
            ArkServiceException: _description_
        """
        self._logger.info(f'Deleting account [{delete_account.account_id}]')
        resp: Response = self._client.delete(ACCOUNT_URL.format(account_id=delete_account.account_id))
        if resp.status_code != HTTPStatus.NO_CONTENT:
            raise ArkServiceException(f'Failed to delete account [{resp.text}] - [{resp.status_code}]')

    def link_account(self, link_account: ArkPCloudLinkAccount) -> None:
        """
        Link an account by given info
        https://docs.cyberark.com/Product-Doc/OnlineHelp/PrivCloud-SS/Latest/en/Content/WebServices/Link-account.htm

        Args:
            link_account (ArkPCloudLinkAccount): _description_

        Raises:
            ArkServiceException: _description_
        """
        self._logger.info(
            f'Linking account [{link_account.account_id}] '
            f'to name [{link_account.name}] in safe [{link_account.safe}] in folder [{link_account.folder}] '
            f'by idx [{link_account.extra_password_index}]'
        )
        resp: Response = self._client.post(
            LINK_ACCOUNT.format(account_id=link_account.account_id), json=link_account.dict(exclude={'account_id'}, by_alias=True)
        )
        if resp.status_code != HTTPStatus.OK:
            raise ArkServiceException(f'Failed to link account [{resp.text}] - [{resp.status_code}]')

    def unlink_account(self, unlink_account: ArkPCloudUnlinkAccount) -> None:
        """
        Link an account by given info
        https://docs.cyberark.com/Product-Doc/OnlineHelp/PrivCloud-SS/Latest/en/Content/WebServices/Link-account-unlink.htm

        Args:
            unlink_account (ArkPCloudUnlinkAccount): _description_

        Raises:
            ArkServiceException: _description_
        """
        self._logger.info(f'Unlinking account [{unlink_account.account_id}] by idx [{unlink_account.extra_password_index}]')
        resp: Response = self._client.delete(
            UNLINK_ACCOUNT.format(account_id=unlink_account.account_id, extra_password_index=unlink_account.extra_password_index)
        )
        if resp.status_code != HTTPStatus.OK:
            raise ArkServiceException(f'Failed to unlink account [{resp.text}] - [{resp.status_code}]')

    def accounts_stats(self) -> ArkPCloudAccountsStats:
        """
        Calculates accounts stats for all visible accounts

        Returns:
            ArkPCloudAccountsStats: _description_
        """
        self._logger.info('Calculating accounts statistics')
        accounts = list(itertools.chain.from_iterable([p.items for p in list(self.list_accounts())]))
        accounts_stats = ArkPCloudAccountsStats.construct()
        accounts_stats.accounts_count = len(accounts)

        # Get accounts per platform id
        platform_ids: Set[str] = {a.platform_id for a in accounts}
        accounts_stats.accounts_count_by_platform_id = {pi: len([a for a in accounts if a.platform_id == pi]) for pi in platform_ids}

        # Get accounts per safe name
        safe_names: Set[str] = {a.safe_name for a in accounts}
        accounts_stats.accounts_count_by_safe_name = {sn: len([a for a in accounts if a.safe_name == sn]) for sn in safe_names}

        return accounts_stats

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

account(get_account)

Retrieves the account by id https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/Get%20Account%20Details.htm?

Parameters:

Name	Type	Description	Default
get_account	ArkPCloudGetAccount	description	required

Raises:

Type	Description
ArkServiceException	description

Returns:

Name	Type	Description
ArkPCloudAccount	ArkPCloudAccount	description

Source code in ark_sdk_python/services/pcloud/accounts/ark_pcloud_accounts_service.py

def account(self, get_account: ArkPCloudGetAccount) -> ArkPCloudAccount:
    """
    Retrieves the account by id
    https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/Get%20Account%20Details.htm?

    Args:
        get_account (ArkPCloudGetAccount): _description_

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkPCloudAccount: _description_
    """
    self._logger.info(f'Retrieving account by id [{get_account.account_id}]')
    resp: Response = self._client.get(ACCOUNT_URL.format(account_id=get_account.account_id))
    if resp.status_code == HTTPStatus.OK:
        try:
            return ArkPCloudAccount.parse_obj(resp.json())
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse account response [{str(ex)}] - [{resp.text}]')
            raise ArkServiceException(f'Failed to parse account response [{str(ex)}]') from ex
    raise ArkServiceException(f'Failed to retrieve account [{resp.text}] - [{resp.status_code}]')

account_credentials(get_account_credentials)

Retrieves the account credentials https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/GetPasswordValueV10.htm?

Parameters:

Name	Type	Description	Default
get_account_credentials	ArkPCloudGetAccountCredentials	description	required

Raises:

Type	Description
ArkServiceException	description

Returns:

Name	Type	Description
ArkPCloudAccountCredentials	ArkPCloudAccountCredentials	description

Source code in ark_sdk_python/services/pcloud/accounts/ark_pcloud_accounts_service.py

def account_credentials(self, get_account_credentials: ArkPCloudGetAccountCredentials) -> ArkPCloudAccountCredentials:
    """
    Retrieves the account credentials
    https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/GetPasswordValueV10.htm?

    Args:
        get_account_credentials (ArkPCloudGetAccountCredentials): _description_

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkPCloudAccountCredentials: _description_
    """
    self._logger.info(f'Retrieving account password for details [{get_account_credentials}]')
    body = {
        k.replace('_', '').title(): v
        for k, v in get_account_credentials.dict(exclude={'account_id', 'reason'}, exclude_none=True).items()
    }
    if get_account_credentials.reason:
        body['reason'] = get_account_credentials.reason
    resp: Response = self._client.post(RETRIEVE_ACCOUNT_CREDENTIALS.format(account_id=get_account_credentials.account_id), json=body)
    if resp.status_code == HTTPStatus.OK:
        return ArkPCloudAccountCredentials(
            account_id=get_account_credentials.account_id, password=resp.text[1:-1]  # Remove leading and trailing quotes
        )
    raise ArkServiceException(f'Failed to retrieve account credentials [{resp.text}] - [{resp.status_code}]')

accounts_stats()

Calculates accounts stats for all visible accounts

Returns:

Name	Type	Description
ArkPCloudAccountsStats	ArkPCloudAccountsStats	description

Source code in ark_sdk_python/services/pcloud/accounts/ark_pcloud_accounts_service.py

def accounts_stats(self) -> ArkPCloudAccountsStats:
    """
    Calculates accounts stats for all visible accounts

    Returns:
        ArkPCloudAccountsStats: _description_
    """
    self._logger.info('Calculating accounts statistics')
    accounts = list(itertools.chain.from_iterable([p.items for p in list(self.list_accounts())]))
    accounts_stats = ArkPCloudAccountsStats.construct()
    accounts_stats.accounts_count = len(accounts)

    # Get accounts per platform id
    platform_ids: Set[str] = {a.platform_id for a in accounts}
    accounts_stats.accounts_count_by_platform_id = {pi: len([a for a in accounts if a.platform_id == pi]) for pi in platform_ids}

    # Get accounts per safe name
    safe_names: Set[str] = {a.safe_name for a in accounts}
    accounts_stats.accounts_count_by_safe_name = {sn: len([a for a in accounts if a.safe_name == sn]) for sn in safe_names}

    return accounts_stats

add_account(add_account)

Adds a new account with given details https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/Add%20Account%20v10.htm?

Parameters:

Name	Type	Description	Default
add_account	ArkPCloudAddAccount	description	required

Raises:

Type	Description
ArkServiceException	description

Returns:

Name	Type	Description
ArkPCloudAccount	ArkPCloudAccount	description

Source code in ark_sdk_python/services/pcloud/accounts/ark_pcloud_accounts_service.py

def add_account(self, add_account: ArkPCloudAddAccount) -> ArkPCloudAccount:
    """
    Adds a new account with given details
    https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/Add%20Account%20v10.htm?

    Args:
        add_account (ArkPCloudAddAccount): _description_

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkPCloudAccount: _description_
    """
    self._logger.info('Adding new account')
    if add_account.remote_machines_access:
        if (
            add_account.remote_machines_access.access_restricted_to_remote_machines
            and not add_account.remote_machines_access.remote_machines
        ):
            add_account.remote_machines_access = None
        elif not add_account.remote_machines_access.access_restricted_to_remote_machines:
            add_account.remote_machines_access.remote_machines = None

        elif add_account.remote_machines_access.remote_machines:
            add_account.remote_machines_access.remote_machines = ','.join(add_account.remote_machines_access.remote_machines)

    resp: Response = self._client.post(ACCOUNTS_URL, data=add_account.json(by_alias=True, exclude_none=True, exclude_defaults=True))
    if resp.status_code == HTTPStatus.CREATED:
        try:
            return ArkPCloudAccount.parse_obj(resp.json())
        except (ValidationError, JSONDecodeError) as ex:
            self._logger.exception(f'Failed to parse add account response [{str(ex)}] - [{resp.text}]')
            raise ArkServiceException(f'Failed to parse add account response [{str(ex)}]') from ex
    raise ArkServiceException(f'Failed to add account [{resp.text}] - [{resp.status_code}]')

change_account_credentials(change_account_credentials)

Marks the account for password changing immediately by CPM https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/Change-credentials-immediately.htm

Parameters:

Name	Type	Description	Default
change_account_credentials	ArkPCloudChangeAccountCredentials	description	required

Raises:

Type	Description
ArkServiceException	description

Source code in ark_sdk_python/services/pcloud/accounts/ark_pcloud_accounts_service.py

def change_account_credentials(self, change_account_credentials: ArkPCloudChangeAccountCredentials) -> None:
    """
    Marks the account for password changing immediately by CPM
    https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/Change-credentials-immediately.htm

    Args:
        change_account_credentials (ArkPCloudChangeAccountCredentials): _description_

    Raises:
        ArkServiceException: _description_
    """
    self._logger.info(f'Marking account [{change_account_credentials.account_id}] for changing credentials immediately')
    resp: Response = self._client.post(CHANGE_ACCOUNT_CREDENTIALS.format(account_id=change_account_credentials.account_id))
    if resp.status_code != HTTPStatus.OK:
        raise ArkServiceException(f'Failed to mark account for changing credentials immediately [{resp.text}] - [{resp.status_code}]')

delete_account(delete_account)

Deletes an account by given id https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/Delete%20Account.htm

Parameters:

Name	Type	Description	Default
delete_account	ArkPCloudDeleteAccount	description	required

Raises:

Type	Description
ArkServiceException	description

Source code in ark_sdk_python/services/pcloud/accounts/ark_pcloud_accounts_service.py

def delete_account(self, delete_account: ArkPCloudDeleteAccount) -> None:
    """
    Deletes an account by given id
    https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/Delete%20Account.htm

    Args:
        delete_account (ArkPCloudDeleteAccount): _description_

    Raises:
        ArkServiceException: _description_
    """
    self._logger.info(f'Deleting account [{delete_account.account_id}]')
    resp: Response = self._client.delete(ACCOUNT_URL.format(account_id=delete_account.account_id))
    if resp.status_code != HTTPStatus.NO_CONTENT:
        raise ArkServiceException(f'Failed to delete account [{resp.text}] - [{resp.status_code}]')

generate_account_credentials(generate_account_credentials)

Generate a new random password for an existing account with policy restrictions https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/SDK/Secrets-Generate-Password.htm

Parameters:

Name	Type	Description	Default
generate_account_credentials	ArkPCloudGenerateAccountCredentials	description	required

Raises:

Type	Description
ArkServiceException	description

Returns:

Name	Type	Description
ArkPCloudAccountCredentials	ArkPCloudAccountCredentials	description

Source code in ark_sdk_python/services/pcloud/accounts/ark_pcloud_accounts_service.py

def generate_account_credentials(
    self, generate_account_credentials: ArkPCloudGenerateAccountCredentials
) -> ArkPCloudAccountCredentials:
    """
    Generate a new random password for an existing account with policy restrictions
    https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/SDK/Secrets-Generate-Password.htm

    Args:
        generate_account_credentials (ArkPCloudGenerateAccountCredentials): _description_

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkPCloudAccountCredentials: _description_
    """
    self._logger.info(f'Generating new password for account [{generate_account_credentials.account_id}]')
    resp: Response = self._client.post(GENERATE_ACCOUNT_CREDENTIALS.format(account_id=generate_account_credentials.account_id))
    if resp.status_code == HTTPStatus.OK:
        try:
            return ArkPCloudAccountCredentials(account_id=generate_account_credentials.account_id, password=resp.json()['password'])
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse genereate account credentials response [{str(ex)}] - [{resp.text}]')
            raise ArkServiceException(f'Failed to parse genereate account credentials response [{str(ex)}]') from ex
    raise ArkServiceException(f'Failed to generate password for account [{resp.text}] - [{resp.status_code}]')

link_account(link_account)

Link an account by given info https://docs.cyberark.com/Product-Doc/OnlineHelp/PrivCloud-SS/Latest/en/Content/WebServices/Link-account.htm

Parameters:

Name	Type	Description	Default
link_account	ArkPCloudLinkAccount	description	required

Raises:

Type	Description
ArkServiceException	description

Source code in ark_sdk_python/services/pcloud/accounts/ark_pcloud_accounts_service.py

def link_account(self, link_account: ArkPCloudLinkAccount) -> None:
    """
    Link an account by given info
    https://docs.cyberark.com/Product-Doc/OnlineHelp/PrivCloud-SS/Latest/en/Content/WebServices/Link-account.htm

    Args:
        link_account (ArkPCloudLinkAccount): _description_

    Raises:
        ArkServiceException: _description_
    """
    self._logger.info(
        f'Linking account [{link_account.account_id}] '
        f'to name [{link_account.name}] in safe [{link_account.safe}] in folder [{link_account.folder}] '
        f'by idx [{link_account.extra_password_index}]'
    )
    resp: Response = self._client.post(
        LINK_ACCOUNT.format(account_id=link_account.account_id), json=link_account.dict(exclude={'account_id'}, by_alias=True)
    )
    if resp.status_code != HTTPStatus.OK:
        raise ArkServiceException(f'Failed to link account [{resp.text}] - [{resp.status_code}]')

list_account_secret_versions(list_account_secret_versions)

Lists the given account secret versions https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/SDK/Secrets-Get-versions.htm

Parameters:

Name	Type	Description	Default
list_account_secret_versions	ArkPCloudListAccountSecretVersions	description	required

Raises:

Type	Description
ArkServiceException	description

Returns:

Type	Description
List[ArkPCloudAccountSecretVersion]	List[ArkPCloudAccountSecretVersion]: description

Source code in ark_sdk_python/services/pcloud/accounts/ark_pcloud_accounts_service.py

def list_account_secret_versions(
    self, list_account_secret_versions: ArkPCloudListAccountSecretVersions
) -> List[ArkPCloudAccountSecretVersion]:
    """
    Lists the given account secret versions
    https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/SDK/Secrets-Get-versions.htm

    Args:
        list_account_secret_versions (ArkPCloudListAccountSecretVersions): _description_

    Raises:
        ArkServiceException: _description_

    Returns:
        List[ArkPCloudAccountSecretVersion]: _description_
    """
    self._logger.info(f'Listing account [{list_account_secret_versions.account_id}] secret versions')
    resp: Response = self._client.get(ACCOUNT_SECRET_VERSIONS.format(account_id=list_account_secret_versions.account_id))
    if resp.status_code == HTTPStatus.OK:
        try:
            return parse_obj_as(List[ArkPCloudAccountSecretVersion], resp.json()['versions'])
        except (ValidationError, JSONDecodeError, KeyError) as ex:
            self._logger.exception(f'Failed to parse list account secret versions response [{str(ex)}] - [{resp.text}]')
            raise ArkServiceException(f'Failed to parse list account secret versions response [{str(ex)}]') from ex
    raise ArkServiceException(f'Failed to list account secret versions [{resp.text}] - [{resp.status_code}]')

list_accounts()

Yields all visible accounts to the logged in user as pages of accounts https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/SDK/GetAccounts.htm

Yields:

Type	Description
ArkPCloudAccountsPage	Iterator[ArkPCloudAccountsPage]: description

Source code in ark_sdk_python/services/pcloud/accounts/ark_pcloud_accounts_service.py

def list_accounts(self) -> Iterator[ArkPCloudAccountsPage]:
    """
    Yields all visible accounts to the logged in user as pages of accounts
    https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/SDK/GetAccounts.htm

    Yields:
        Iterator[ArkPCloudAccountsPage]: _description_
    """
    self._logger.info('Listing all accounts')
    yield from self.__list_accounts_with_filters()

list_accounts_by(accounts_filter)

Yields visible accounts to the logged in user by filters as pages of accounts https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/SDK/GetAccounts.htm

Parameters:

Name	Type	Description	Default
accounts_filter	ArkPCloudAccountsFilter	description	required

Yields:

Type	Description
ArkPCloudAccountsPage	Iterator[ArkPCloudAccountsPage]: description

Source code in ark_sdk_python/services/pcloud/accounts/ark_pcloud_accounts_service.py

def list_accounts_by(self, accounts_filter: ArkPCloudAccountsFilter) -> Iterator[ArkPCloudAccountsPage]:
    """
    Yields visible accounts to the logged in user by filters as pages of accounts
    https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/SDK/GetAccounts.htm

    Args:
        accounts_filter (ArkPCloudAccountsFilter): _description_

    Yields:
        Iterator[ArkPCloudAccountsPage]: _description_
    """
    self._logger.info(f'Listing accounts by filters [{accounts_filter}]')
    yield from self.__list_accounts_with_filters(
        accounts_filter.search,
        accounts_filter.search_type,
        accounts_filter.sort,
        accounts_filter.offset,
        accounts_filter.limit,
        accounts_filter.safe_name,
    )

reconcile_account_credentials(reconcile_account_credentials)

Marks the account for reconcilation https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/Reconcile-account.htm

Parameters:

Name	Type	Description	Default
reconcile_account_credentials	ArkPCloudReconcileAccountCredentials	description	required

Raises:

Type	Description
ArkServiceException	description

Source code in ark_sdk_python/services/pcloud/accounts/ark_pcloud_accounts_service.py

def reconcile_account_credentials(self, reconcile_account_credentials: ArkPCloudReconcileAccountCredentials) -> None:
    """
    Marks the account for reconcilation
    https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/Reconcile-account.htm

    Args:
        reconcile_account_credentials (ArkPCloudReconcileAccountCredentials): _description_

    Raises:
        ArkServiceException: _description_
    """
    self._logger.info(f'Marking account [{reconcile_account_credentials.account_id}] for reconcilation')
    resp: Response = self._client.post(RECONCILE_ACCOUNT_CREDENTIALS.format(account_id=reconcile_account_credentials.account_id))
    if resp.status_code != HTTPStatus.OK:
        raise ArkServiceException(f'Failed to reconcile account credentials [{resp.text}] - [{resp.status_code}]')

set_account_next_credentials(set_account_next_credentials)

Marks the account to have its password changed to the given one via CPM https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/SetNextPassword.htm

Parameters:

Name	Type	Description	Default
set_account_next_credentials	ArkPCloudSetAccountNextCredentials	description	required

Raises:

Type	Description
ArkServiceException	description

Source code in ark_sdk_python/services/pcloud/accounts/ark_pcloud_accounts_service.py

def set_account_next_credentials(self, set_account_next_credentials: ArkPCloudSetAccountNextCredentials) -> None:
    """
    Marks the account to have its password changed to the given one via CPM
    https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/SetNextPassword.htm

    Args:
        set_account_next_credentials (ArkPCloudSetAccountNextCredentials): _description_

    Raises:
        ArkServiceException: _description_
    """
    self._logger.info(f'Marking account [{set_account_next_credentials.account_id}] for changing credentials for the given password')
    resp: Response = self._client.post(
        SET_ACCOUNT_NEXT_CREDENTIALS.format(account_id=set_account_next_credentials.account_id),
        json=set_account_next_credentials.dict(exclude={'account_id'}, by_alias=True),
    )
    if resp.status_code != HTTPStatus.OK:
        raise ArkServiceException(f'Failed to mark account for changing credentials next password [{resp.text}] - [{resp.status_code}]')

unlink_account(unlink_account)

Link an account by given info https://docs.cyberark.com/Product-Doc/OnlineHelp/PrivCloud-SS/Latest/en/Content/WebServices/Link-account-unlink.htm

Parameters:

Name	Type	Description	Default
unlink_account	ArkPCloudUnlinkAccount	description	required

Raises:

Type	Description
ArkServiceException	description

Source code in ark_sdk_python/services/pcloud/accounts/ark_pcloud_accounts_service.py

def unlink_account(self, unlink_account: ArkPCloudUnlinkAccount) -> None:
    """
    Link an account by given info
    https://docs.cyberark.com/Product-Doc/OnlineHelp/PrivCloud-SS/Latest/en/Content/WebServices/Link-account-unlink.htm

    Args:
        unlink_account (ArkPCloudUnlinkAccount): _description_

    Raises:
        ArkServiceException: _description_
    """
    self._logger.info(f'Unlinking account [{unlink_account.account_id}] by idx [{unlink_account.extra_password_index}]')
    resp: Response = self._client.delete(
        UNLINK_ACCOUNT.format(account_id=unlink_account.account_id, extra_password_index=unlink_account.extra_password_index)
    )
    if resp.status_code != HTTPStatus.OK:
        raise ArkServiceException(f'Failed to unlink account [{resp.text}] - [{resp.status_code}]')

update_account(update_account)

Updates an existing account with new details https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/SDK/UpdateAccount%20v10.htm

Parameters:

Name	Type	Description	Default
update_account	ArkPCloudUpdateAccount	description	required

Raises:

Type	Description
ArkServiceException	description

Returns:

Name	Type	Description
ArkPCloudAccount	ArkPCloudAccount	description

Source code in ark_sdk_python/services/pcloud/accounts/ark_pcloud_accounts_service.py

def update_account(self, update_account: ArkPCloudUpdateAccount) -> ArkPCloudAccount:
    """
    Updates an existing account with new details
    https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/SDK/UpdateAccount%20v10.htm

    Args:
        update_account (ArkPCloudUpdateAccount): _description_

    Raises:
        ArkServiceException: _description_

    Returns:
        ArkPCloudAccount: _description_
    """
    self._logger.info(f'Updating account [{update_account.account_id}]')
    if update_account.remote_machines_access and not update_account.remote_machines_access.remote_machines:
        update_account.remote_machines_access = None
    operations = []
    for key, val in update_account.dict(exclude={'account_id'}, exclude_none=True, by_alias=True, exclude_defaults=True):
        operations.append({'op': 'replace', 'path': f'/{key}', 'value': val})
    resp: Response = self._client.put(ACCOUNT_URL.format(account_id=update_account.account_id), json=operations)
    if resp.status_code == HTTPStatus.OK:
        try:
            return ArkPCloudAccount.parse_obj(resp.json())
        except (ValidationError, JSONDecodeError) as ex:
            self._logger.exception(f'Failed to parse update account response [{str(ex)}] - [{resp.text}]')
            raise ArkServiceException(f'Failed to parse update account response [{str(ex)}]') from ex
    raise ArkServiceException(f'Failed to update account [{resp.text}] - [{resp.status_code}]')

update_account_credentials_in_vault(update_account_credentials_in_vault)

Updates the account credentials only in the vault without changing it on the machine itself https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/ChangeCredentialsInVault.htm

Parameters:

Name	Type	Description	Default
update_account_credentials_in_vault	ArkPCloudUpdateAccountCredentialsInVault	description	required

Raises:

Type	Description
ArkServiceException	description

Source code in ark_sdk_python/services/pcloud/accounts/ark_pcloud_accounts_service.py

def update_account_credentials_in_vault(self, update_account_credentials_in_vault: ArkPCloudUpdateAccountCredentialsInVault) -> None:
    """
    Updates the account credentials only in the vault without changing it on the machine itself
    https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/ChangeCredentialsInVault.htm

    Args:
        update_account_credentials_in_vault (ArkPCloudUpdateAccountCredentialsInVault): _description_

    Raises:
        ArkServiceException: _description_
    """
    self._logger.info(f'Updates account [{update_account_credentials_in_vault.account_id}] vault credentials')
    resp: Response = self._client.post(
        UPDATE_ACCOUNT_CREDENTIALS_IN_VAULT.format(account_id=update_account_credentials_in_vault.account_id),
        json=update_account_credentials_in_vault.dict(exclude={'account_id'}, by_alias=True),
    )
    if resp.status_code != HTTPStatus.OK:
        raise ArkServiceException(f'Failed to update credentials in vault for account [{resp.text}] - [{resp.status_code}]')

verify_account_credentials(verify_account_credentials)

Marks the account for password verification by CPM https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/Verify-credentials-v9-10.htm

Parameters:

Name	Type	Description	Default
verify_account_credentials	ArkPCloudVerifyAccountCredentials	description	required

Raises:

Type	Description
ArkServiceException	description

Source code in ark_sdk_python/services/pcloud/accounts/ark_pcloud_accounts_service.py

def verify_account_credentials(self, verify_account_credentials: ArkPCloudVerifyAccountCredentials) -> None:
    """
    Marks the account for password verification by CPM
    https://docs.cyberark.com/Product-Doc/OnlineHelp/PAS/Latest/en/Content/WebServices/Verify-credentials-v9-10.htm

    Args:
        verify_account_credentials (ArkPCloudVerifyAccountCredentials): _description_

    Raises:
        ArkServiceException: _description_
    """
    self._logger.info(f'Marking account [{verify_account_credentials.account_id}] for verification')
    resp: Response = self._client.post(VERIFY_ACCOUNT_CREDENTIALS.format(account_id=verify_account_credentials.account_id))
    if resp.status_code != HTTPStatus.OK:
        raise ArkServiceException(f'Failed to mark account for password verification [{resp.text}] - [{resp.status_code}]')