Source
import demistomock as demisto from CommonServerPython import * SCRIPT_NAME = "ConfigurationSetup" class Pack: """Pack object for the configuration file. Args: id_ (str): Pack ID. version (str): Version of the pack to install. """ def __init__(self, id_: str, version: str = "", url: str = ""): self.id = id_ self._version = version self.url = url @property def version(self) -> str: """The getter method for the version variable.""" if self._version == "*": return "latest" return self._version @property def installation_object(self) -> Dict[str, str]: """Creates the layout of an installation object in marketplace for the pack.""" return {"id": self.id, "version": self.version} class IntegrationInstance: """Integration instance object for the configuration file. Args: brand_name (str): Integration name to be configured. instance_name (str): Instance name to be configured. """ INSTANCES_KEYWORDS_LIST = ["use_cases", "brand_name", "instance_name"] def __init__(self, brand_name: str, instance_name: str): self.brand_name = brand_name self.instance_name = instance_name @property def params(self) -> Dict: """Getter for the instance parameters. Returns: Dict. {param_name: param_value} for each configured parameter. """ return { param_name: param_value for param_name, param_value in self.__dict__.items() if param_name not in IntegrationInstance.INSTANCES_KEYWORDS_LIST } def add_param(self, name: str, value: Any): self.__dict__[name] = value def get_param(self, param_name: str, default_value: Any = None) -> Any: """Get the parameter for the instance by name. Args: param_name (str): The name of the parameter to get it's value. default_value (Any): The fallback value for the case where the parameter is not configured. Returns: Any. The value of the parameter. Notes: In the case where the parameter is not configured, will return the default value. or None if not supplied. """ try: return self.__dict__[param_name] except KeyError: return default_value class Job: """Job object for the configuration file. Args: job_name (str): Job name to be configured. """ JOBS_KEYWORDS_LIST = ["use_cases", "job_name"] def __init__(self, job_name: str): self.job_name = job_name @property def params(self) -> Dict: """Getter for the instance parameters. Returns: Dict. {param_name: param_value} for each configured parameter. """ return { param_name: param_value for param_name, param_value in self.__dict__.items() if param_name not in Job.JOBS_KEYWORDS_LIST } def add_param(self, name: str, value: Any): self.__dict__[name] = value def get_param(self, param_name: str, default_value: Any = None) -> Any: """Get the parameter for the instance by name. Args: param_name (str): The name of the parameter to get it's value. default_value (Any): The fallback value for the case where the parameter is not configured. Returns: Any. The value of the parameter. Notes: In the case where the parameter is not configured, will return the default value. or None if not supplied. """ try: return self.__dict__[param_name] except KeyError: return default_value class Configuration: def __init__(self, configuration_data: Dict): """Configuration object for the configuration file. Args: configuration_data (Dict): The configuration data parsed from the configuration file. """ self.config = configuration_data # Variables self.sections = list(self.config.keys()) # Objects Variables self.jobs: Dict[str, Job] = {} self.lists: Dict[str, Dict[str, str]] = {} self.custom_packs: Dict[str, Pack] = {} self.marketplace_packs: Dict[str, Pack] = {} self.integration_instances: Dict[str, IntegrationInstance] = {} # Load and create Objects self.load_jobs() self.load_lists() self.load_custom_packs() self.load_marketplace_packs() self.load_integration_instances() def load_custom_packs(self) -> None: """Iterates through the Packs sections and creates a Pack object for each custom pack.""" if "custom_packs" in self.sections: for pack in self.config["custom_packs"]: pack_id = pack.get("id") pack_url = pack.get("url") if pack_url: new_pack = Pack(pack_id, url=pack_url) self.custom_packs[pack_id] = new_pack def load_marketplace_packs(self) -> None: """Iterates through the Packs sections and creates a Pack object for each marketplace pack.""" if "marketplace_packs" in self.sections: for pack in self.config["marketplace_packs"]: pack_id = pack.get("id") pack_version = pack.get("version") if pack_version: new_pack = Pack(pack_id, version=pack_version) self.marketplace_packs[pack_id] = new_pack def load_integration_instances(self) -> None: """Iterates through the instances sections, creates IntegrationInstance object for each instance.""" if "instances" in self.sections: for instance in self.config["instances"]: brand_name = instance.get("brand_name") instance_name = instance.get("instance_name") new_instance = IntegrationInstance(brand_name, instance_name) for param_name, param_value in instance.items(): new_instance.add_param(param_name, param_value) self.integration_instances[instance_name] = new_instance def load_jobs(self) -> None: """Iterates through the jobs sections, creates Job object for each job.""" if "jobs" in self.sections: for job in self.config["jobs"]: job_name = job.get("name") new_job = Job(job_name) for param_name, param_value in job.items(): new_job.add_param(param_name, param_value) self.jobs[job_name] = new_job def load_lists(self) -> None: """Iterates through the lists sections, creates Dict object for each list.""" if "lists" in self.sections: for _list in self.config["lists"]: list_name = _list.get("name") list_value = _list.get("value") list_type = _list.get("type") self.lists[list_name] = {"value": list_value, "type": list_type} def list_exists(list_name: str) -> bool: res = demisto.executeCommand("getList", {"listName": list_name})[0] return res["Type"] == entryTypes["error"] and "Item not found" in res["Contents"] def create_context(full_configuration: Configuration) -> Dict[str, List[Dict[str, str]]]: custom_packs = [ { "packid": pack.id, "packurl": pack.url, } for _, pack in full_configuration.custom_packs.items() ] marketplace_packs = [ { "packid": pack.id, "packversion": str(pack.version), } for _, pack in full_configuration.marketplace_packs.items() ] jobs = [job.params for _, job in full_configuration.jobs.items()] lists = [ { "listname": list_name, "listdata": pps["value"], } for list_name, pps in full_configuration.lists.items() if not list_exists(list_name) or pps["type"] != "dynamic" ] return { "Jobs": jobs, "Lists": lists, "CustomPacks": custom_packs, "MarketplacePacks": marketplace_packs, } def get_data_from_war_room_file(entry_id) -> bytes: """Retrieves the content of a file from the war-room. Args: entry_id (str): The entry ID of the configuration file from the war-room. Returns: str. The content of the configuration file. """ try: file_path = demisto.getFilePath(entry_id)["path"] except Exception: raise DemistoException(f"Could not find a file with entry ID {entry_id}") with open(file_path, "rb") as file: file_content = file.read() return file_content def get_config_data(args: Dict) -> Dict: """Gets the configuration data from Git or from a file entry in the war room.. Returns: Dict. The parsed configuration file. """ configuration_file_entry_id = args.get("configuration_file_entry_id") config_data = get_data_from_war_room_file(configuration_file_entry_id) try: return json.loads(config_data) except json.JSONDecodeError: raise DemistoException("Configuration file is not a valid JSON structure.") def main(): try: args = demisto.args() config_data = get_config_data(args) config = Configuration(config_data) return_results( CommandResults( outputs_prefix="ConfigurationSetup", outputs=create_context(config), ) ) except Exception as e: return_error(f"{SCRIPT_NAME} - Error occurred while setting up machine.\n{e}") if __name__ in ("__main__", "__builtin__", "builtins"): main()