import demistomock as demisto from CommonServerPython import * from packaging.version import parse, Version, LegacyVersion SCRIPT_NAME = 'MarketplacePackInstaller' class ContentPackInstaller: """A class that handles all marketplace packs' installation process. """ PACK_ID_VERSION_FORMAT = '{}::{}' def __init__(self, instance_name=''): self.installed_packs: Dict[str, Union[Version, LegacyVersion]] = dict() self.newly_installed_packs: Dict[str, Version] = dict() self.already_on_machine_packs: Dict[str, Union[Version, LegacyVersion]] = dict() self.packs_data: Dict[str, Dict[str, str]] = dict() self.packs_dependencies: Dict[str, Dict[str, Dict[str, str]]] = dict() self.instance_name: str = instance_name self.get_installed_packs() def get_installed_packs(self) -> None: """Gets the current installed packs on the machine. """ demisto.debug(f'{SCRIPT_NAME} - Fetching installed packs from marketplace.') args = {'uri': '/contentpacks/metadata/installed'} if self.instance_name: args['using'] = self.instance_name status, res = execute_command( 'core-api-get', args, fail_on_error=False, ) if not status: error_message = f'{SCRIPT_NAME} - {res}' demisto.debug(error_message) return packs_data: List[Dict[str, str]] = res.get('response', []) for pack in packs_data: self.installed_packs[pack['id']] = parse(pack['currentVersion']) self.already_on_machine_packs[pack['id']] = parse(pack['currentVersion']) def get_pack_data_from_marketplace(self, pack_id: str) -> Dict[str, str]: """Returns the marketplace's data for a specific pack. Args: pack_id (str): The pack ID for which to get the data. Returns: Dict[str, str]. The pack's data from marketplace. """ if pack_id in self.packs_data: demisto.debug(f'{SCRIPT_NAME} - Using cached data of {pack_id} that already been fetched.') return self.packs_data[pack_id] demisto.debug(f'{SCRIPT_NAME} - Fetching {pack_id} data from marketplace.') args = {'uri': f'/contentpacks/marketplace/{pack_id}'} if self.instance_name: args['using'] = self.instance_name status, res = execute_command( 'core-api-get', args, fail_on_error=False, ) if not status: error_message = f'{SCRIPT_NAME} - {res}' demisto.debug(error_message) if type(res) == list: res = res[0] self.packs_data[pack_id] = res return res def get_pack_dependencies_from_marketplace(self, pack_data: Dict[str, str]) -> Dict[str, Dict[str, str]]: """Returns the dependencies of the pack from marketplace's data. Args: pack_data (Dict[str, str]): Packs' data for installation. Returns: Dict[str, Dict]. The pack's dependencies data from marketplace. """ pack_key = self.PACK_ID_VERSION_FORMAT.format(pack_data['id'], pack_data['version']) if pack_key in self.packs_dependencies: demisto.debug(f'{SCRIPT_NAME} - Using cached dependencies data of {pack_key} that already been fetched.') return self.packs_dependencies[pack_key] demisto.debug(f'{SCRIPT_NAME} - Fetching {pack_key} dependencies data from marketplace.') args = {'uri': '/contentpacks/marketplace/search/dependencies', 'body': [pack_data]} if self.instance_name: args['using'] = self.instance_name status, res = execute_command( 'core-api-post', args, fail_on_error=False, ) if not status: error_message = f'{SCRIPT_NAME} - {res}' demisto.debug(error_message) try: self.packs_dependencies[pack_key] = res.get('response', {}).get('packs', [])[0] \ .get('extras', {}).get('pack', {}).get('dependencies') except Exception as e: demisto.debug(f'{SCRIPT_NAME} - Unable to parse {pack_data["id"]} pack dependencies from response.\n{e}') return {} return self.packs_dependencies[pack_key] def get_latest_version_for_pack(self, pack_id: str) -> str: """Gets the latest version of the pack from the marketplace data. Args: pack_id (str): The pack name for which to get the data. Returns: str. The latest version of the pack. """ res = self.get_pack_data_from_marketplace(pack_id) return res.get('response', {}).get('currentVersion') # type: ignore[call-overload, union-attr] def get_packs_data_for_installation(self, packs_to_install: List[Dict[str, str]]) -> List[Dict[str, str]]: """Creates a list of packs' data for the installation request. Args: packs_to_install (List[Dict[str, str]]): The packs to get data for. Returns: List[Dict[str, str]]. Updated list of data objects per pack, which hasn't been installed it. """ latest_version_packs_to_install = [] for pack in packs_to_install: latest_version = self.get_latest_version_for_pack(pack['id']) if parse(latest_version) > self.installed_packs.get(pack['id'], parse('1.0.0')): pack['version'] = latest_version latest_version_packs_to_install.append(pack) return latest_version_packs_to_install def install_packs(self, packs_to_install: List[Dict[str, str]]) -> None: """Sends the request to install the packs in the machine. Args: packs_to_install (List[Dict[str, str]]): The packs data to be used for the installation. """ if not packs_to_install: demisto.debug(f'{SCRIPT_NAME} - No packs were sent for installation.') return data = { 'packs': packs_to_install, 'ignoreWarnings': True } # make the pack installation request packs_names_versions = {pack['id']: parse(pack['version']) for pack in packs_to_install} demisto.debug(f'{SCRIPT_NAME} - Sending installation request for: {packs_names_versions}') args = {'uri': '/contentpacks/marketplace/install', 'body': data} if self.instance_name: args['using'] = self.instance_name execute_command( 'core-api-post', args, extract_contents=False ) self.installed_packs.update(packs_names_versions) self.newly_installed_packs.update(packs_names_versions) # type: ignore[arg-type] def get_dependencies_for_pack(self, pack_data: Dict[str, str]) -> List[Dict[str, str]]: """Retrieves the packs' dependencies from the marketplace data. Args: pack_data (Dict[str, str]): Packs' data for installation. Returns: List[Dict[str, str]]. List of the packs' dependencies. """ dependencies_to_install = [] if not pack_data: return [] if type(pack_data) is list: pack_data = pack_data[0] # type: ignore pack_dependencies = self.get_pack_dependencies_from_marketplace(pack_data) for dependency_id, dependency_data in pack_dependencies.items(): if dependency_data.get('mandatory'): dependency_version = dependency_data.get('minVersion', '1.0.0') if parse(dependency_version) > self.installed_packs.get(dependency_id, parse('1.0.0')): dependencies_to_install.append({ 'id': dependency_id, 'version': dependency_version }) pack_key = self.PACK_ID_VERSION_FORMAT.format(pack_data['id'], pack_data['version']) demisto.debug(f'{SCRIPT_NAME} - Dependencies found for {pack_key}: {dependencies_to_install}') return dependencies_to_install def is_pack_already_installed(self, pack_data: Dict[str, str]) -> bool: """Returns whether the pack is already installed on the machine or not. Args: pack_data (Dict[str, str]): Packs' data for installation. Returns: bool. Whether the pack is already installed. """ if pack_data['id'] not in self.installed_packs: return False try: if pack_data['version'] in ['latest', '*']: return parse(self.get_latest_version_for_pack(pack_data['id'])) == self.installed_packs[pack_data['id']] return parse(pack_data['version']) <= self.installed_packs[pack_data['id']] except Exception: return parse(self.get_latest_version_for_pack(pack_data['id'])) == self.installed_packs[pack_data['id']] def install_pack_and_its_dependencies_recursively(self, pack_data: Dict[str, str]) -> None: """Recursive method for installing a pack and it's prerequisites in order. Args: pack_data (Dict[str, str]): Packs' data for installation. """ if self.is_pack_already_installed(pack_data): pack_info_for_message = self.PACK_ID_VERSION_FORMAT.format(pack_data['id'], pack_data['version']) demisto.debug(f'{SCRIPT_NAME} - pack "{pack_info_for_message}" is already installed') return demisto.debug(f'{SCRIPT_NAME} - Initiating installation process for pack {pack_data["id"]}') if pack_data['version'] == 'latest': pack_data = self.get_packs_data_for_installation([pack_data]) # type: ignore if pack_data and type(pack_data) == list: pack_data = pack_data[0] # type: ignore dependencies_to_install = self.get_dependencies_for_pack(pack_data) for dependency in dependencies_to_install: self.install_pack_and_its_dependencies_recursively(dependency) dependencies_to_install = self.get_packs_data_for_installation(dependencies_to_install) if not dependencies_to_install and not pack_data: return demisto.debug(f'{SCRIPT_NAME} - Updated dependencies for {pack_data["id"]}: {dependencies_to_install}') if dependencies_to_install: self.install_packs(dependencies_to_install) if pack_data: self.install_packs([pack_data]) def format_packs_data_for_installation(args: Dict) -> List[Dict[str, str]]: """Creates the body of the installation request from the raw data. Returns: List[Dict[str, str]]: Installable objects list. """ packs_data = argToList(args.get('packs_data', [])) id_key = args.get('pack_id_key') version_key = args.get('pack_version_key') try: return [ { 'id': pack[id_key], 'version': pack[version_key], } for pack in packs_data ] except KeyError as e: raise DemistoException(f'The following key does not exist in the pack data: {e}.') from e except Exception as e: raise DemistoException(f'Unknown error occurred while processing the packs data.\n{e}') from e def create_context(packs_to_install: List[Dict[str, str]], content_packs_installer: ContentPackInstaller) \ -> List[Dict[str, str]]: """Creates context entry including all relevant data. Args: packs_to_install (List[Dict[str, str]]): The list of packs that should have been installed. content_packs_installer (ContentPackInstaller): The content packs installer. Returns: List[Dict[str, str]]. Formatted context data after installation processes. """ context_data = [] requested_packs_ids = [pack['id'] for pack in packs_to_install] for pack_id, pack_version in content_packs_installer.newly_installed_packs.items(): content_installed = { 'packid': pack_id, 'packversion': str(pack_version), 'installationstatus': 'Success.' if pack_id in requested_packs_ids else 'Installed as requirement.', } context_data.append(content_installed) for pack_id in requested_packs_ids: if pack_id in content_packs_installer.already_on_machine_packs: content_installed = { 'packid': pack_id, 'packversion': str(content_packs_installer.installed_packs[pack_id]), 'installationstatus': 'Already Installed on the machine.', } context_data.append(content_installed) return context_data def main(): try: args = demisto.args() instance_name = args.get('using') installer = ContentPackInstaller(instance_name) packs_to_install = format_packs_data_for_installation(args) for pack in packs_to_install: installer.install_pack_and_its_dependencies_recursively(pack) return_results( CommandResults( outputs_prefix='ConfigurationSetup.MarketplacePacks', outputs_key_field='packid', outputs=create_context(packs_to_install, installer), ) ) except Exception as e: return_error(f'{SCRIPT_NAME} - Error occurred while setting up machine.\n{e}') if __name__ in ('__main__', '__builtin__', 'builtins'): main()