ContentPackInstaller

Source

import demistomock as demisto
from CommonServerPython import *
from packaging.version import Version, parse

SCRIPT_NAME = "ContentPackInstaller"


class ContentPackInstaller:
    """A class that handles all marketplace packs' installation process."""

    PACK_ID_VERSION_FORMAT = "{}::{}"

    def __init__(self, instance_name: str = None):
        self.installed_packs: Dict[str, Version] = {}
        self.newly_installed_packs: Dict[str, Version] = {}
        self.already_on_machine_packs: Dict[str, Version] = {}
        self.packs_data: Dict[str, Dict[str, str]] = {}
        self.packs_dependencies: Dict[str, Dict[str, Dict[str, str]]] = {}
        self.packs_failed: Dict[str, str] = {}
        self.instance_name: Optional[str] = instance_name

        self.get_installed_packs()

    def _call_execute_command(self, command, args):
        if self.instance_name:
            args["using"] = self.instance_name

        status, res = execute_command(
            command,
            args,
            fail_on_error=False,
        )

        if not status:
            error_message = f"{SCRIPT_NAME} - {res}"
            demisto.debug(error_message)
        if type(res) is list:
            res = res[0]
        return status, res

    def get_installed_packs(self) -> None:  # pragma: no cover
        """Gets the current installed packs on the machine."""
        demisto.debug(f"{SCRIPT_NAME} - Fetching installed packs from marketplace.")

        args = {"uri": "/contentpacks/metadata/installed"}

        status, res = self._call_execute_command("core-api-get", args)
        if not status:
            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]:  # pragma: no cover
        """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}"}

        _, res = self._call_execute_command("core-api-get", args)

        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]]:  # pragma: no cover
        """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]}

        _, res = self._call_execute_command("core-api-post", args)

        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.
        """
        try:
            res = self.get_pack_data_from_marketplace(pack_id)
            demisto.debug(f"raw pack_data_from_marketplace: {res}")
            return res.get("response", {}).get("currentVersion")  # type: ignore[call-overload, union-attr]
        except AttributeError as e:
            demisto.debug(f"error trying to get {pack_id=}. {e}")
            raise ValueError(
                f"Error while fetching {pack_id} from the marketplace. "
                f"Make sure the Core REST API integration is properly configured and {pack_id} exists. "
                f"Try running `!core-api-get uri=/contentpacks/marketplace/{pack_id}` in the playground. "
                f"Raw Response:\n{res}"
            )

    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:  # pragma: no cover
        """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

        # 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}")

        for pack in packs_to_install:
            pack_id = pack["id"]
            pack_payload = json.dumps([{pack_id: pack["version"]}])

            if is_xsiam_or_xsoar_saas():  # install packs for XSOAR 8.X or XSIAM
                data = {"packs": packs_to_install, "ignoreWarnings": True}
                args = {"uri": "/contentpacks/marketplace/install", "body": data}
                status, res = self._call_execute_command("core-api-post", args)

            else:  # install packs for XSOAR 6.X
                args = {"packs_to_install": str(pack_payload)}
                status, res = self._call_execute_command("core-api-install-packs", args)

            if not status:
                demisto.error(f"{SCRIPT_NAME} - Failed to install the pack {pack_id} - {res!s}")
                self.packs_failed[pack_id] = str(pack["version"])
            else:
                self.installed_packs[pack_id] = packs_names_versions[pack_id]
                self.newly_installed_packs[pack_id] = packs_names_versions[pack_id]  # type: ignore

    def get_dependencies_for_pack(self, pack_data: Dict[str, str]) -> List[Dict[str, str]]:  # pragma: no cover
        """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 = []

        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(
        self, pack_data: Dict[str, str], install_dependencies: bool
    ) -> None:  # pragma: no cover
        """Method for installing a pack and it's prerequisites in order.

        Args:
            pack_data (Dict[str, str]): Packs' data for installation.
            install_dependencies (bool): Whether to install the pack dependencies.

        """
        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])[0]

        if install_dependencies:
            dependencies_to_install = self.get_dependencies_for_pack(pack_data)
            demisto.debug(f'{SCRIPT_NAME} - Updated dependencies for {pack_data["id"]}: {dependencies_to_install}')
            self.install_packs(dependencies_to_install)

        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]]:  # pragma: cover
    """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)

        if pack_id in content_packs_installer.packs_failed:
            packs_failed = {
                "packid": pack_id,
                "packversion": content_packs_installer.packs_failed[pack_id],
                "installationstatus": "Failed to install.",
            }
            context_data.append(packs_failed)

    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)
        install_dependencies = argToBoolean(args.get("install_dependencies", "true"))
        if is_xsiam_or_xsoar_saas():
            # When downloading a pack through the tenant marketplace, we must install all dependencies
            install_dependencies = True

        for pack in packs_to_install:
            installer.install_pack_and_its_dependencies(pack, install_dependencies)

        return_results(
            CommandResults(
                outputs_prefix="ConfigurationSetup.MarketplacePacks",
                outputs_key_field="packid",
                outputs=create_context(packs_to_install, installer),
            )
        )

    except Exception as e:
        demisto.debug(f"error occured during script execution {e}")
        return_error(f"{SCRIPT_NAME} - Error occurred while setting up machine.\n{e}")


if __name__ in ("__main__", "__builtin__", "builtins"):
    main()

README