"""Dataminr Pulse - ReGenAI Integration for Cortex XSOAR (aka Demisto).""" from functools import reduce from operator import concat from urllib.parse import urlparse, parse_qs from collections.abc import Callable from copy import deepcopy import demistomock as demisto import urllib3 from CommonServerPython import * # noqa # pylint: disable=unused-wildcard-import from CommonServerUserPython import * # noqa # Disable insecure warnings urllib3.disable_warnings() """ CONSTANTS """ # Base URL for the Dataminr Pulse ReGenAI API. BASE_URL = "https://api.dataminr.com" ENDPOINTS = { # Authentication endpoint for the Dataminr Pulse API. "AUTH_ENDPOINT": "/auth/v1/token", # Watchlists endpoint for the Dataminr Pulse API. "WATCHLISTS_ENDPOINT": "/pulse/v1/lists", # Alerts endpoint for the Dataminr Pulse API. "ALERTS_ENDPOINT": "/pulse/v1/alerts", } DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ" # ISO8601 format with UTC, default in XSOAR # Date format for displaying date in human-readable format in war room. HR_DATE_FORMAT = "%d %b %Y, %I:%M %p UTC" APPLICATION_NAME = "palo_alto_cortex_xsoar" OK_CODES = (200, 201, 401) DEFAULT_NUMBER_OF_ALERTS_TO_RETRIEVE = 40 MAX_NUMBER_OF_ALERTS_TO_RETRIEVE = 100 STATUS_LIST_TO_RETRY = (429, *(status_code for status_code in requests.status_codes._codes if status_code >= 500)) # type: ignore EARLY_EXPIRY_TIME = 30000 # in milliseconds DEFAULT_ALERT_TYPE = "All" VENDOR_NAME = "DataminrPulse" CUSTOM_OUTPUT_PREFIX = "DataminrPulse.{}" ERRORS = { "INVALID_JSON_OBJECT": "Status code: {}. Failed to parse json object from response: {}.", "UNAUTHORIZED_REQUEST": "Status code: {}. Unauthorized request: {}.", "GENERAL_AUTH_ERROR": "Status code: {}. Error occurred while creating an authorization token. " "Please check the Client ID, Client Secret {}.", "NOT_MATCHED_WATCHLIST_NAMES": "No matching watchlist data was found for the {} watchlist names configured in the instance.", "INVALID_NUM": '{} is invalid value for "{}". Value should be greater than 0 and less than or equal to {}.', "ATMOST_ONE_ALLOWED": 'At most one argument, either "{}" or "{}" is allowed.', "JSON_DECODE": "Failed to parse '{}' JSON string.", "REQUIRED_ARG": "Required argument '{}' is missing.", } ALERT_TYPE_TO_INCIDENT_SEVERITY = {"Alert": 1, "Urgent": 3, "Flash": 4} DEFAULT_RELIABILITY = "A - Completely reliable" # Output prefix for the alerts. OUTPUT_PREFIX_ALERTS = "DataminrPulse.Alerts" # Output prefix for the lists. OUTPUT_PREFIX_WATCHLISTS = "DataminrPulse.WatchLists" # Output prefix for the cursor. OUTPUT_PREFIX_CURSOR = "DataminrPulse.Cursor" """ CLIENT CLASS """ class DataminrPulseReGenAIClient(BaseClient): """DataminrPulseClient class to interact with the Dataminr Pulse API.""" def __init__(self, client_id: str = "", client_secret: str = "", verify: bool = False, proxy: bool = False): """ Constructor for the DataminrPulseClient class. :type client_id: ``str`` :param client_id: Client ID to be used for authentication. :type client_secret: ``str`` :param client_secret: Client Secret to be used for authentication. :type verify: ``bool`` :param verify: Whether the request should verify the SSL certificate. :type proxy: ``bool`` :param proxy: Whether to run the integration using the system proxy. """ self.client_id = client_id self.client_secret = client_secret self.base_url = BASE_URL super().__init__(base_url=self.base_url, verify=verify, proxy=proxy, headers={}) def http_request( self, method, url_suffix, params=None, status_list_to_retry=STATUS_LIST_TO_RETRY, backoff_factor=30, retries=3, internal_retries=3, **kwargs, ) -> Optional[dict]: """ Method to override private _http_request of BaseClient to handle specific status code. :type method: ``str`` :param method: The HTTP method, for example: GET, POST, and so on. :type url_suffix: ``str`` :param url_suffix: The API endpoint. :type params: ``dict`` :param params: URL parameters to specify the query. :type status_list_to_retry: ``iterable`` :param status_list_to_retry: A set of integer HTTP status codes that we should force a retry on. :type backoff_factor ``float`` :param backoff_factor: A backoff factor to apply between attempts :type retries: ``int`` :param retries: How many retries should be made in case of a failure. :type internal_retries: ``int`` :param internal_retries: How many retries should be made in case of an auth failure. :return: Response dict. :rtype: ``Optional[dict]`` """ # Adds a valid authentication token to the headers. dma_token = self.get_dma_token(use_existing_token=True) headers = { "Authorization": f"Bearer {dma_token}", "X-Application-Name": APPLICATION_NAME, } self._headers.update(headers) res = self._http_request( method=method, url_suffix=url_suffix, params=params, status_list_to_retry=status_list_to_retry, backoff_factor=backoff_factor, retries=retries, resp_type="response", ok_codes=OK_CODES, **kwargs, ) res_status_code = res.status_code try: json_data = res.json() except ValueError as exception: raise DemistoException(ERRORS["INVALID_JSON_OBJECT"].format(res_status_code, res.content), exception) # If the success response is received, then return it. if res_status_code in [200, 201]: return json_data # If authentication failure happens. if res_status_code in [401]: if internal_retries > 0: dma_token = self.get_dma_token(use_existing_token=False) headers = { "Authorization": f"Bearer {dma_token}", "X-Application-Name": APPLICATION_NAME, } self._headers.update(headers) internal_retries = internal_retries - 1 return self.http_request( method=method, url_suffix=url_suffix, params=params, status_list_to_retry=status_list_to_retry, backoff_factor=backoff_factor, retries=retries, internal_retries=internal_retries, ) try: err_msg = ERRORS["UNAUTHORIZED_REQUEST"].format(res_status_code, str(res.json())) except ValueError: err_msg = ERRORS["UNAUTHORIZED_REQUEST"].format(res_status_code, str(res)) raise DemistoException(err_msg) return None def get_dma_token(self, use_existing_token: bool = True) -> Optional[str]: """ Get a DMA token that was previously created if it is still valid, else, generate a new authorization token from the client id, client secret and refresh token. :type use_existing_token ``bool`` :param use_existing_token: Use existing token if it is still valid. :return: DMA token. :rtype: ``Optional[str]`` """ integration_context: dict = get_integration_context() previous_token: dict = integration_context.get("token", {}) # Check if there is existing valid authorization token. if ( previous_token.get("dmaToken") and use_existing_token and previous_token.get("expire") > datetime.now(timezone.utc).timestamp() * 1000 # type: ignore ): demisto.debug("Got authentication token from the integration context.") return previous_token.get("dmaToken") # type: ignore demisto.debug("Trying to generate a new authentication token.") data = {"client_id": self.client_id, "client_secret": self.client_secret, "grant_type": "api_key"} headers = { "Content-Type": "application/x-www-form-urlencoded", "X-Application-Name": APPLICATION_NAME, } res = super()._http_request( method="POST", url_suffix=ENDPOINTS["AUTH_ENDPOINT"], resp_type="response", headers=headers, data=data, ok_codes=OK_CODES, status_list_to_retry=STATUS_LIST_TO_RETRY, ) res_status_code = res.status_code try: res = res.json() except ValueError as exception: raise DemistoException(ERRORS["INVALID_JSON_OBJECT"].format(res_status_code, res.content), exception) if "errors" in res: raise DemistoException(ERRORS["GENERAL_AUTH_ERROR"].format(res_status_code, str(res))) if res.get("dmaToken"): expiry_time = res.get("expire", 0) - EARLY_EXPIRY_TIME demisto.debug( f"Setting the expiry time of the authentication token to {timestamp_to_datestring(expiry_time, is_utc=True)}." ) new_token = {"dmaToken": res.get("dmaToken"), "expire": expiry_time} integration_context.update({"token": new_token}) set_integration_context(integration_context) return res.get("dmaToken") return None def get_alerts( self, watchlist_ids: Optional[list] = None, query: Optional[str] = None, _from: Optional[str] = None, to: Optional[str] = None, page_size: int = DEFAULT_NUMBER_OF_ALERTS_TO_RETRIEVE, ) -> Optional[dict]: """ Retrieves the alerts stored on the Dataminr platform. :type watchlist_ids ``Optional[list]`` :param watchlist_ids: List of watchlist id. :type query ``Optional[str]`` :param query: Terms to search within Dataminr Alerts. :type _from ``Optional[str]`` :param _from: Provide cursor value to get alerts after that. :type to ``Optional[str]`` :param to: Provide cursor value to get alerts before that. :type page_size ``int`` :param page_size: Maximum number of alerts to return. :return: A dictionary of alerts. :rtype: ``Optional[dict]`` """ params = { "pageSize": page_size, "from": _from, "to": to, "query": query, } remove_nulls_from_dictionary(params) if watchlist_ids: params["lists"] = ",".join(map(str, watchlist_ids)) # type: ignore return self.http_request(method="GET", url_suffix=ENDPOINTS["ALERTS_ENDPOINT"], params=params) def get_watchlists(self) -> Optional[dict]: """Retrieves the watchlists stored on the Dataminr platform. :return: A dictionary of Watchlists grouped by their types. :rtype: ``Optional[Dict]`` """ return self.http_request(method="GET", url_suffix=ENDPOINTS["WATCHLISTS_ENDPOINT"]) """ HELPER FUNCTIONS """ def validate_params_for_alerts_get( watchlist_ids: Optional[list] = None, watchlist_names: Optional[list] = None, page_size: int = DEFAULT_NUMBER_OF_ALERTS_TO_RETRIEVE, use_configured_watchlist_names: bool = False, is_fetch: bool = False, _from: Optional[str] = None, to: Optional[str] = None, ): """ To validate arguments for the alerts get. :type watchlist_ids ``Optional[List]`` :param watchlist_ids: List of watchlist id. :type watchlist_names ``Optional[List]`` :param watchlist_names: Watchlist names. :type page_size ``int`` :param page_size: Maximum number of alerts to return. :type use_configured_watchlist_names ``bool`` :param use_configured_watchlist_names: Use configured watchlist names. :type is_fetch ``bool`` :param is_fetch: Function is called by fetch_incident method. :type _from ``Optional[str]`` :param _from: Start cursor for the alerts. :type to ``Optional[str]`` :param to: End cursor for the alerts. """ if is_fetch and (page_size < 1 or page_size > MAX_NUMBER_OF_ALERTS_TO_RETRIEVE): raise ValueError(ERRORS["INVALID_NUM"].format(page_size, "Max Fetch", MAX_NUMBER_OF_ALERTS_TO_RETRIEVE)) elif page_size < 1 or page_size > MAX_NUMBER_OF_ALERTS_TO_RETRIEVE: raise ValueError(ERRORS["INVALID_NUM"].format(page_size, "num", MAX_NUMBER_OF_ALERTS_TO_RETRIEVE)) if (is_fetch or use_configured_watchlist_names) and (watchlist_names and not watchlist_ids): raise ValueError(ERRORS["NOT_MATCHED_WATCHLIST_NAMES"].format(watchlist_names)) if _from and to: raise ValueError(ERRORS["ATMOST_ONE_ALLOWED"].format("from", "to")) def transform_watchlists_data(watchlists_data: Optional[dict] = None) -> list: """ Transform watchlist data from dictionary to single List. :type watchlists_data ``Optional[dict]`` :param watchlists_data: Response to be converted in single list. :return: List of response. :rtype: ``list`` """ list_of_watchlists = watchlists_data.get("lists", {}).values() # type: ignore # The returned object is a dictionary where the lists are grouped by their type. # So, modifying this object and creating a list of all list objects as the object itself contains # a property named "type" in it, which defines the type of the list. list_of_watchlists = list(list_of_watchlists) # The created list will be the list of lists. So, flattening the list is required. list_of_watchlists = reduce(concat, list_of_watchlists) return list_of_watchlists def get_watchlist_ids(client: DataminrPulseReGenAIClient, watchlist_names: Optional[list] = None) -> list: """ Get watchlist IDs as per the given watchlist names using integration context and get_watchlists method from client. :type client ``DataminrPulseReGenAIClient`` :param client: DataminrPulseReGenAIClient to get watchlists data. :type watchlist_names ``Optional[List]`` :param watchlist_names: Watchlist names. :return: Watchlist IDs. :rtype: ``list`` """ watchlists_data: list = transform_watchlists_data(client.get_watchlists()) watchlist_names_in_lower = [watchlist_name.lower() for watchlist_name in watchlist_names] if watchlist_names else [] filtered_watchlists_data: list = ( list( filter( lambda watchlist_data: watchlist_data.get("name").lower() in watchlist_names_in_lower, # type: ignore watchlists_data, ) ) if watchlist_names else watchlists_data ) if not filtered_watchlists_data: demisto.debug( f'No matching watchlist data was found for the "{watchlist_names}" watchlist names configured in the instance.' ) return [] watchlist_ids: list = [watchlist_data.get("id") for watchlist_data in filtered_watchlists_data] watchlist_ids: list = list(filter(None, watchlist_ids)) return watchlist_ids def create_media_html(alert: dict) -> str: """ Create HTML content for the media. :type alert: ``dict`` :param alert: Alert data. :rtype: ``str`` :return: HTML content for the media. """ media = alert.get("publicPost", {}).get("media", []) if not media: return "" photo_media = list(filter(lambda photo: photo.get("type") == "photo" or photo.get("type") == "image", media)) if not photo_media: return "" html_content = [] for photo in photo_media: html_content.append(f'Photo') media_html = "\n".join(html_content) return media_html def remove_empty_elements_for_hr(d: Any) -> Any: """ Recursively remove empty lists, empty dicts, or None elements from a dictionary or list. :param d: Input dictionary or list. :return: Dictionary or list with all empty lists, and empty dictionaries removed. """ if not isinstance(d, dict | list): return str(d) if isinstance(d, int | float) else d elif isinstance(d, list): return [v for v in (remove_empty_elements_for_hr(v) for v in d) if not check_empty(v)] return {k: v for k, v in ((k, remove_empty_elements_for_hr(v)) for k, v in d.items()) if not check_empty(v)} def remove_empty_elements_for_fetch(d: Any) -> Any: """ Recursively remove empty lists, empty dicts, or None elements from a dictionary or list. :param d: Input dictionary or list. :return: Dictionary or list with all empty lists, and empty dictionaries removed. """ if not isinstance(d, dict | list): return d elif isinstance(d, list): return [v for v in (remove_empty_elements_for_fetch(v) for v in d) if not check_empty(v)] return {k: v for k, v in ((k, remove_empty_elements_for_fetch(v)) for k, v in d.items()) if not check_empty(v)} def check_empty(x: Any) -> bool: """ Check if input is empty (None, empty dict, empty list, or empty string). :param x: Input to check. :type x: Any :return: True if x is empty, False otherwise. :rtype: bool """ return x is None or x == {} or x == [] or x == "" def trim_spaces_from_args(args: Dict) -> Dict: """Trim spaces from values of the args Dict. :type args: ``Dict`` :param args: Dict to trim spaces from. :rtype: ``Dict`` :return: Arguments after trim spaces. """ for key, val in args.items(): if isinstance(val, str): args[key] = val.strip() elif isinstance(val, dict): args[key] = trim_spaces_from_args(val) elif isinstance(val, list): for i in range(len(val)): if isinstance(val[i], str): val[i] = val[i].strip() elif isinstance(val[i], dict): val[i] = trim_spaces_from_args(val[i]) return args def prepare_hr_for_watchlists_get(watchlists: List) -> str: """ Prepare human-readable string for war room entry. :type watchlists: ``List`` :param watchlists: List of watchlists. :return: Human-readable output. :rtype: ``str`` """ # The title of the table. title = "Watchlists" # Data dictionary for the table. hr_outputs = [ { "Watchlist ID": wl.get("id", ""), "Watchlist Name": wl.get("name", ""), "Watchlist Type": wl.get("type", ""), "Watchlist Sub Type": wl.get("subType", ""), } for wl in watchlists ] # Table headers. headers = ["Watchlist ID", "Watchlist Name", "Watchlist Type", "Watchlist Sub Type"] return tableToMarkdown(title, hr_outputs, headers, removeNull=True) def prepare_hr_for_alerts(alerts: List) -> str: """Prepare Human Readable output for alerts. :type alerts: ``List`` :param alerts: Response from the API. :rtype: ``str`` :return: Human readable output. """ # The title of the table. title = "Alerts" # This will store the data dictionaries for the table. hr_outputs = [] for alert in alerts: # List of watchlist names matched in the alert. watchlist_names = [watchlist.get("name") for watchlist in alert.get("listsMatched", [])] intel_agents = deepcopy(alert.get("intelAgents", [])) intel_agents_summary = [] discovered_entities = [] for intel_agent in intel_agents: intel_agents_summary.extend(intel_agent.get("summary", [])) discovered_entities.extend(intel_agent.get("discoveredEntities", [])) for summary in intel_agents_summary: summary["type"] = ", ".join(summary.get("type", [])) summary["content"] = "\n".join(summary.get("content", [])) for discovered_entity in discovered_entities: if discovered_entity.get("type") == "threatActor" and discovered_entity.get("aliases"): discovered_entity["aliases"] = ", ".join(discovered_entity.get("aliases", [])) elif discovered_entity.get("type") == "malware" and discovered_entity.get("affectedOperatingSystems"): discovered_entity["affectedOperatingSystems"] = ", ".join(discovered_entity.get("affectedOperatingSystems", [])) elif discovered_entity.get("type") == "vulnerability" and discovered_entity.get("exploitPocLinks"): discovered_entity["exploitPocLinks"] = ", ".join(discovered_entity.get("exploitPocLinks", [])) hr_outputs.append( { "Alert Type": alert.get("alertType", {}).get("name", ""), "Alert ID": f"[{alert.get('alertId', '')}]({alert.get('dataminrAlertUrl', '')})", "Alert Name": alert.get("headline", ""), "Intel Agents Summary": intel_agents_summary if intel_agents_summary else "", "Intel Agents Discovered Entities": discovered_entities if discovered_entities else "", "Live Brief": alert.get("liveBrief", ""), "Watchlist Name": ", ".join(watchlist_names), "Alert Time": alert.get("alertTimestamp", ""), "Alert Location": alert.get("estimatedEventLocation", {}).get("name", ""), "Post Link": alert.get("publicPost", {}).get("href", ""), "Alert Topics": alert.get("alertTopics", ""), } ) # Table headers. headers = [ "Alert Type", "Alert ID", "Alert Name", "Intel Agents Summary", "Intel Agents Discovered Entities", "Live Brief", "Watchlist Name", "Alert Time", "Alert Location", "Post Link", "Alert Topics", ] return tableToMarkdown( title, hr_outputs, headers, removeNull=True, url_keys=["Post Link"], json_transform_mapping={ "Alert Topics": JsonTransformer(is_nested=True), "Intel Agents Summary": JsonTransformer(is_nested=True), "Intel Agents Discovered Entities": JsonTransformer(is_nested=True), "Live Brief": JsonTransformer(is_nested=True), }, ) def prepare_hr_for_cursor(cursor: Dict) -> str: """Prepare Human Readable output for cursor. :type cursor: ``Dict`` :param cursor: Contains from and to parameter. :rtype: ``str`` :return: Human readable output. """ # The title of the table. title = "Cursor for pagination" # This will store the data dictionaries for the table. _from = cursor.get("from") if _from: _from = re.escape(_from) to = cursor.get("to") if to: to = re.escape(to) hr_outputs = [{"from": _from, "to": to}] # Table headers. headers = ["from", "to"] return tableToMarkdown(title, hr_outputs, headers, removeNull=True) def create_vulnerability_indicators(alert: dict) -> list: """ Create vulnerability indicators for the alert. :type alert: ``dict`` :param alert: Alert data. :rtype: ``list`` :return: List of vulnerability indicators. """ vulnerability_intel_agents_list = [] vulnerability_metadata_list = alert.get("metadata", {}).get("cyber", {}).get("vulnerabilities", []) intel_agents = alert.get("intelAgents", []) for vulnerability in vulnerability_metadata_list: if "epssScore" in vulnerability: vulnerability["epssScore"] = str(vulnerability.get("epssScore", "0")) + "%" for intel_agent in intel_agents: discovered_entities = intel_agent.get("discoveredEntities", []) for discovered_entity in discovered_entities: if discovered_entity.get("type", "") == "vulnerability": discovered_entity["id"] = discovered_entity.get("name", "") if "epssScore" in discovered_entity: discovered_entity["epssScore"] = str(discovered_entity.get("epssScore", "0")) + "%" vulnerability_intel_agents_list.append(discovered_entity) merged = {} for vuln in vulnerability_intel_agents_list: vid = vuln.get("id") if vid is not None: merged[vid] = deepcopy(vuln) for vuln in vulnerability_metadata_list: vid = vuln.get("id") if vid is None: continue if vid in merged: merged[vid] = merge_dicts(merged[vid], vuln) else: merged[vid] = deepcopy(vuln) return list(merged.values()) def create_malware_indicators(alert: dict) -> list: """ Create malware indicators for the alert. :type alert: ``dict`` :param alert: Alert data. :rtype: ``list`` :return: List of malware indicators. """ malware_intel_agents_list = [] malware_metadata_list = alert.get("metadata", {}).get("cyber", {}).get("malware", []) intel_agents = alert.get("intelAgents", []) for malware in malware_metadata_list: malware["name"] = "Malware: [" + malware.get("name", "") + "]" for intel_agent in intel_agents: discovered_entities = intel_agent.get("discoveredEntities", []) for discovered_entity in discovered_entities: if discovered_entity.get("type", "") == "malware": discovered_entity["name"] = "Malware: [" + discovered_entity.get("name", "") + "]" malware_intel_agents_list.append(discovered_entity) merged = {} for malware in malware_intel_agents_list: mid = malware.get("name") if mid is not None: merged[mid] = deepcopy(malware) for malware in malware_metadata_list: mid = malware.get("name") if mid is None: continue if mid in merged: merged[mid] = merge_dicts(merged[mid], malware) else: merged[mid] = deepcopy(malware) return list(merged.values()) def create_threat_actors_indicators(alert: dict) -> list: """ Create threat actors indicators for the alert. :type alert: ``dict`` :param alert: Alert data. :rtype: ``list`` :return: List of threat actors indicators. """ threat_actor_intel_agents_list = [] threat_actor_metadata_list = alert.get("metadata", {}).get("cyber", {}).get("threatActors", []) intel_agents = alert.get("intelAgents", []) for threat_actor in threat_actor_metadata_list: threat_actor["name"] = "Threat Actor: [" + threat_actor.get("name", "") + "]" for intel_agent in intel_agents: discovered_entities = intel_agent.get("discoveredEntities", []) for discovered_entity in discovered_entities: if discovered_entity.get("type", "") == "threatActor": discovered_entity["name"] = "Threat Actor: [" + discovered_entity.get("name", "") + "]" threat_actor_intel_agents_list.append(discovered_entity) merged = {} for actor in threat_actor_intel_agents_list: name = actor.get("name") if name: merged[name] = deepcopy(actor) for actor in threat_actor_metadata_list: name = actor.get("name") if not name: continue if name in merged: merged[name] = merge_dicts(merged[name], actor) else: merged[name] = deepcopy(actor) return list(merged.values()) def create_flight_data(alert: dict) -> dict: """ Create flight data for the alert. :type alert: ``dict`` :param alert: Alert data. :rtype: ``dict`` :return: Flight data. """ sub_headline_content = alert.get("subHeadline", {}).get("content", []) flight_data = {} for item in sub_headline_content: # Skip empty or whitespace-only lines if not item.strip(): continue # Check if it contains a colon (valid key-value format) if ":" in item: key, value = item.split(":", 1) key, value = key.strip(), value.strip() # Only add if both key and value are non-empty if key and value: flight_data[key] = value return flight_data def calculate_dbot_score(cvss: float) -> int: if cvss >= 4.0: return Common.DBotScore.BAD elif cvss > 0.0: return Common.DBotScore.SUSPICIOUS else: return Common.DBotScore.NONE def merge_dicts(d1: dict, d2: dict) -> dict: """ Merge two dictionaries recursively and handle duplicate values. Works for dict, list, string, number, etc. """ merged = dict(d1) # start with a copy of d1 for key, value in d2.items(): if key not in merged: # New key, just add merged[key] = value return merged def create_relationship_for_malware(malware: dict) -> list: """ Create a list of relationships objects from the malware and threat actor. :type malware: ``dict`` :param malware: Malware of API. :return: List of EntityRelationship objects containing all the relationships. :rtype: ``List`` """ relationships = [] source_malware = malware.get("name", "") integration_reliability = demisto.params().get("integrationReliability", DEFAULT_RELIABILITY) for threat_actor in malware.get("threatActors", []): threat_actor_name = "Threat Actor: [" + threat_actor.get("name", "") + "]" relationships.append( EntityRelationship( name=EntityRelationship.Relationships.ORIGINATED_FROM, entity_a=source_malware, entity_a_type="Dataminr Pulse Malware Indicator", entity_b=threat_actor_name, entity_b_type="Dataminr Pulse Threat Actor Indicator", source_reliability=integration_reliability, brand=VENDOR_NAME, ) ) return relationships """ COMMAND FUNCTIONS """ def test_module(client: DataminrPulseReGenAIClient) -> str: """Tests API connectivity and authentication' Returning 'ok' indicates that the integration works like it is supposed to. Connection to the service is successful. Raises: exceptions if something goes wrong. :type client: ``DataminrPulseReGenAIClient`` :param client: DataminrPulseReGenAIClient to be used. :rtype: ``str`` :return: 'ok' if test passed, anything else will fail the test. """ params = demisto.params() is_fetch = params.get("isFetch", False) if is_fetch: fetch_incidents(client, {}, params, is_test=True) else: client.get_alerts(page_size=1) # Return OK, indicating the connection to the platform is successful. return "ok" def fetch_incidents( client: DataminrPulseReGenAIClient, last_run: dict[str, Any], params: dict[str, Any], is_test: bool = False ) -> tuple[dict, list]: """Fetch alerts as an incidents. :type client: ``DataminrPulseReGenAIClient`` :param client: DataminrPulseReGenAIClient to be used. :type last_run: ``dict[str, Any]`` :param last_run: last run object obtained from demisto.getLastRun(). :type params: `dict[str, Any]`` :param params: Arguments to be used for fetch incident. :type is_test: ``bool`` :param is_test: If test_module called fetch_incident. :rtype: ``tuple[dict, list]`` :return: Tuple of last run object and list of fetched incidents. """ page_size: int = arg_to_number( params.get("max_fetch", DEFAULT_NUMBER_OF_ALERTS_TO_RETRIEVE), # type: ignore arg_name="Max Fetch", ) watchlist_names = argToList(params.get("watchlist_names"), transform=lambda s: s.strip()) query = params.get("query") alert_type: str = params.get("alert_type", DEFAULT_ALERT_TYPE) if query: query = query.strip() _from = last_run.get("from") last_watchlist_names = last_run.get("last_watchlist_names", []) last_query = last_run.get("last_query") found_alert_ids = last_run.get("found_alert_ids", []) watchlist_names_lower = [] last_watchlist_names_lower = [] if watchlist_names and isinstance(watchlist_names, list): watchlist_names_lower = [name.lower() for name in watchlist_names] watchlist_names_lower.sort() if last_watchlist_names and isinstance(last_watchlist_names, list): last_watchlist_names_lower = [name.lower() for name in last_watchlist_names] last_watchlist_names_lower.sort() watchlist_ids = get_watchlist_ids(client=client, watchlist_names=watchlist_names) if last_watchlist_names_lower != watchlist_names_lower or last_query != query: demisto.debug("Watchlist names or query changed in configuration, so fetching incident from start") _from = None if not is_test and page_size > 100: page_size = 100 validate_params_for_alerts_get( watchlist_ids=watchlist_ids, watchlist_names=watchlist_names, page_size=page_size, use_configured_watchlist_names=False, is_fetch=True, ) response = client.get_alerts(watchlist_ids=watchlist_ids, query=query, _from=_from, page_size=page_size) alert_response, _from, to = [], "", "" if response: alert_response = response.get("alerts", []) _from = response.get("previousPage", "") to = response.get("nextPage", "") alert_valid_response = remove_empty_elements_for_fetch(alert_response) from_url = urlparse(_from) from_params = parse_qs(from_url.query) from_value = from_params.get("to", [None])[0] # type: ignore to_url = urlparse(to) to_params = parse_qs(to_url.query) to_value = to_params.get("from", [None])[0] # type: ignore cursor_response = {"from": from_value, "to": to_value} cursor_valid_response = remove_empty_elements(cursor_response) if is_test: return {}, [] next_run = last_run.copy() incidents = [] duplicate_alert_ids_in_current_fetch = [] new_alert_ids_in_current_fetch = [] for alert in alert_valid_response: alert_type_name = alert.get("alertType", {}).get("name", "") if alert_type != "All" and (not alert_type_name or alert_type_name.lower() != alert_type.lower()): continue alert_id = alert.get("alertId") if alert_id in found_alert_ids: demisto.debug(f"Found existing alert. Alert ID: {alert_id}") duplicate_alert_ids_in_current_fetch.append(alert_id) continue media_html = create_media_html(alert) alert["media_html"] = media_html vulnerability_indicators = create_vulnerability_indicators(alert) alert["vulnerability_indicators"] = vulnerability_indicators malware_indicators = create_malware_indicators(alert) alert["malware_indicators"] = malware_indicators threat_actors_indicators = create_threat_actors_indicators(alert) alert["threat_actors_indicators"] = threat_actors_indicators flight_data = create_flight_data(alert) alert["flight_data"] = flight_data occurred_date = alert.get("alertTimestamp", "") incident_name = alert.get("headline", "") incidents.append( { "name": "✨ " + incident_name if alert.get("intelAgents", []) else incident_name, "occurred": occurred_date, "rawJSON": json.dumps(alert), "severity": ALERT_TYPE_TO_INCIDENT_SEVERITY.get(alert_type_name, 0), } ) found_alert_ids.append(alert_id) new_alert_ids_in_current_fetch.append(alert_id) next_run["found_alert_ids"] = found_alert_ids if alert_valid_response and cursor_valid_response: next_run["from"] = cursor_valid_response.get("to") next_run["last_watchlist_names"] = watchlist_names next_run["last_query"] = query demisto.debug(f"Total alerts fetch as an incident: {len(incidents)}") demisto.debug(f"New alert IDs in current fetch: {new_alert_ids_in_current_fetch}") demisto.debug(f"Duplicate alert IDs in current fetch: {duplicate_alert_ids_in_current_fetch}") demisto.debug(f"All Alert IDs store in last run: {next_run['found_alert_ids']}") demisto.debug(f"Next page cursor: {next_run['from']}") return next_run, incidents def dataminrpulse_watchlists_get_command(client: DataminrPulseReGenAIClient, args: Dict[str, Any]) -> CommandResults: """Retrieve the Watchlist stored on the Dataminr platform. :type client: ``DataminrPulseReGenAIClient`` :param client: DataminrPulseReGenAIClient to be used. :type args: ``Dict[str, Any]`` :param args: Arguments provided by user. :rtype: ``CommandResults`` :return: Standard command result. """ # Retrieve the lists stored on the Dataminr platform. raw_lists_resp = client.get_watchlists() list_of_watchlists = transform_watchlists_data(raw_lists_resp) # Create a human-readable output for the war room entry. hr_output = prepare_hr_for_watchlists_get(list_of_watchlists) # Create and return a CommandResults object to return_results function. return CommandResults( outputs_prefix=OUTPUT_PREFIX_WATCHLISTS, outputs_key_field="id", outputs=list_of_watchlists, readable_output=hr_output, raw_response=raw_lists_resp, ) def dataminrpulse_alerts_get(client: DataminrPulseReGenAIClient, args: Dict[str, Any]) -> List[CommandResults]: """Retrieve the list of the alerts that meet the specified filter criteria. :type client: ``DataminrPulseReGenAIClient`` :param client: DataminrPulseReGenAIClient to be used. :type args: ``Dict[str, Any]`` :param args: Arguments provided by user. :rtype: ``List[CommandResults]`` :return: Standard command results. """ watchlist_names: List = argToList(args.get("watchlist_names", "")) watchlist_ids: List = argToList(args.get("watchlist_ids", "")) query: str = args.get("query", "") _from: str = args.get("from", "") to: str = args.get("to", "") num: int = arg_to_number(args.get("num", DEFAULT_NUMBER_OF_ALERTS_TO_RETRIEVE), arg_name="num") # type: ignore use_configured_watchlist_names: bool = argToBoolean(args.get("use_configured_watchlist_names", "yes")) if use_configured_watchlist_names and not watchlist_ids: watchlist_ids = get_watchlist_ids(client, watchlist_names) validate_params_for_alerts_get( watchlist_ids=watchlist_ids, watchlist_names=watchlist_names, page_size=num, use_configured_watchlist_names=use_configured_watchlist_names, _from=_from, to=to, ) response = client.get_alerts(watchlist_ids, query, _from, to, num) alert_response = response.get("alerts", []) # type: ignore alert_valid_response = remove_empty_elements(alert_response) hr_output_for_alerts = prepare_hr_for_alerts(alert_valid_response) _from = response.get("previousPage", "") # type: ignore to = response.get("nextPage", "") # type: ignore from_url = urlparse(_from) from_params = parse_qs(from_url.query) from_value = from_params.get("to", [None])[0] # type: ignore to_url = urlparse(to) to_params = parse_qs(to_url.query) to_value = to_params.get("from", [None])[0] # type: ignore cursor_response = {"from": from_value, "to": to_value} cursor_valid_response = remove_empty_elements(cursor_response) hr_output_for_cursor = prepare_hr_for_cursor(cursor_valid_response) alert_results = CommandResults( outputs_prefix=OUTPUT_PREFIX_ALERTS, outputs_key_field="alertId", outputs=alert_valid_response, readable_output=hr_output_for_alerts, raw_response=response, ) cursor_results = CommandResults( outputs_prefix=OUTPUT_PREFIX_CURSOR, outputs_key_field=["from", "to"], outputs=cursor_valid_response, readable_output=hr_output_for_cursor, raw_response=cursor_response, ) return [alert_results, cursor_results] def dataminrpulse_vulnerability_enrich_command(client: DataminrPulseReGenAIClient, args: Dict[str, Any]) -> list[CommandResults]: """ Enrich the "Dataminr Pulse Vulnerability Indicator" custom indicator with relevant data. :type client: ``DataminrPulseReGenAIClient`` :param client: DataminrPulseReGenAIClient to be used. :type args: ``Dict[str, Any]`` :param args: Arguments provided by user. :rtype: ``CommandResults`` :return: Standard command result. """ vulnerability_json_data: Any = args.get("vulnerability_json_data") vulnerability_list: list = [] if not vulnerability_json_data: raise ValueError(ERRORS["REQUIRED_ARG"].format("vulnerability_json_data")) try: vulnerability_list = json.loads(vulnerability_json_data) except json.JSONDecodeError: raise ValueError(ERRORS["JSON_DECODE"].format("vulnerability_json_data")) results = [] if not vulnerability_list: return CommandResults(readable_output="No vulnerabilities found.") # type: ignore for vulnerability in vulnerability_list: indicator_value = vulnerability.get("id", "") dbot_score = Common.DBotScore( indicator=indicator_value, indicator_type=DBotScoreType.CUSTOM, integration_name=VENDOR_NAME, score=calculate_dbot_score(float(vulnerability.get("cvss", Common.DBotScore.NONE))), reliability=demisto.params().get("integrationReliability", DEFAULT_RELIABILITY), ) custom_indicator = Common.CustomIndicator( value=indicator_value, indicator_type="Dataminr Pulse Vulnerability Indicator", data=remove_empty_elements(vulnerability), context_prefix="DataminrPulseVulnerabilityIndicator", dbot_score=dbot_score, ) hr = tableToMarkdown("Vulnerability", remove_empty_elements_for_hr(vulnerability), is_auto_json_transform=True) results.append( CommandResults( outputs=remove_empty_elements(vulnerability), outputs_prefix=CUSTOM_OUTPUT_PREFIX.format("Vulnerability"), outputs_key_field="id", indicator=custom_indicator, raw_response=vulnerability, readable_output=hr, ) ) return results def dataminrpulse_malware_enrich_command(client: DataminrPulseReGenAIClient, args: Dict[str, Any]) -> list[CommandResults]: """ Enrich the "Dataminr Pulse Malware Indicator" custom indicator with relevant data. :type client: ``DataminrPulseReGenAIClient`` :param client: DataminrPulseReGenAIClient to be used. :type args: ``Dict[str, Any]`` :param args: Arguments provided by user. :rtype: ``CommandResults`` :return: Standard command result. """ malware_json_data: Any = args.get("malware_json_data") malware_list: list = [] if not malware_json_data: raise ValueError(ERRORS["REQUIRED_ARG"].format("malware_json_data")) try: malware_list = json.loads(malware_json_data) except json.JSONDecodeError: raise ValueError(ERRORS["JSON_DECODE"].format("malware_json_data")) results = [] if not malware_list: return CommandResults(readable_output="No malware found.") # type: ignore for malware in malware_list: indicator_value = malware.get("name", "") dbot_score = Common.DBotScore( indicator=indicator_value, indicator_type=DBotScoreType.CUSTOM, integration_name=VENDOR_NAME, score=Common.DBotScore.BAD, reliability=demisto.params().get("integrationReliability", DEFAULT_RELIABILITY), ) relationships = [] if demisto.params().get("create_relationships", True): relationships = create_relationship_for_malware(malware) custom_indicator = Common.CustomIndicator( value=indicator_value, indicator_type="Dataminr Pulse Malware Indicator", data=remove_empty_elements(malware), context_prefix="DataminrPulseMalwareIndicator", dbot_score=dbot_score, relationships=relationships, ) hr = tableToMarkdown("Malware", remove_empty_elements_for_hr(malware), is_auto_json_transform=True) results.append( CommandResults( outputs=remove_empty_elements(malware), outputs_prefix=CUSTOM_OUTPUT_PREFIX.format("Malware"), outputs_key_field="name", indicator=custom_indicator, raw_response=malware, readable_output=hr, ) ) return results def dataminrpulse_threat_actor_enrich_command(client: DataminrPulseReGenAIClient, args: Dict[str, Any]) -> list[CommandResults]: """ Enrich the "Dataminr Pulse Threat Actor Indicator" custom indicator with relevant data. :type client: ``DataminrPulseReGenAIClient`` :param client: DataminrPulseReGenAIClient to be used. :type args: ``Dict[str, Any]`` :param args: Arguments provided by user. :rtype: ``CommandResults`` :return: Standard command result. """ threat_actor_json_data: Any = args.get("threat_actor_json_data") threat_actor_list: list = [] if not threat_actor_json_data: raise ValueError(ERRORS["REQUIRED_ARG"].format("threat_actor_json_data")) try: threat_actor_list = json.loads(threat_actor_json_data) except json.JSONDecodeError: raise ValueError(ERRORS["JSON_DECODE"].format("threat_actor_json_data")) results = [] if not threat_actor_list: return CommandResults(readable_output="No threat actors found.") # type: ignore for threat_actor in threat_actor_list: indicator_value = threat_actor.get("name", "") dbot_score = Common.DBotScore( indicator=indicator_value, indicator_type=DBotScoreType.CUSTOM, integration_name=VENDOR_NAME, score=Common.DBotScore.BAD, reliability=demisto.params().get("integrationReliability", DEFAULT_RELIABILITY), ) custom_indicator = Common.CustomIndicator( value=indicator_value, indicator_type="Dataminr Pulse Threat Actor Indicator", data=remove_empty_elements(threat_actor), context_prefix="DataminrPulseThreatActorIndicator", dbot_score=dbot_score, ) hr = tableToMarkdown("Threat Actor", remove_empty_elements_for_hr(threat_actor), is_auto_json_transform=True) results.append( CommandResults( outputs=remove_empty_elements(threat_actor), outputs_prefix=CUSTOM_OUTPUT_PREFIX.format("ThreatActor"), outputs_key_field="name", indicator=custom_indicator, raw_response=threat_actor, readable_output=hr, ) ) return results def main(): """main function, parses params and runs command functions""" # Retrieve the configuration parameters. params = trim_spaces_from_args(demisto.params()) remove_nulls_from_dictionary(params) # Credentials for connecting with the Dataminr Pulse API. client_id = params.get("credentials", {}).get("identifier") client_secret = params.get("credentials", {}).get("password") # Default configuration parameters for handling proxy and SSL Certificate validation. verify_certificate = not argToBoolean(params.get("insecure", False)) proxy = argToBoolean(params.get("proxy", False)) # Parameters for fetch incident mechanism watchlist_names = params.get("watchlist_names") # Retrieve the name of the command being called. command = demisto.command() demisto.debug(f"The command being called is {command}.") demisto_commands: Dict[str, Callable] = { "dataminrpulse-watchlists-get": dataminrpulse_watchlists_get_command, "dataminrpulse-alerts-get": dataminrpulse_alerts_get, "dataminrpulse-vulnerability-enrich": dataminrpulse_vulnerability_enrich_command, "dataminrpulse-malware-enrich": dataminrpulse_malware_enrich_command, "dataminrpulse-threat-actor-enrich": dataminrpulse_threat_actor_enrich_command, } try: client = DataminrPulseReGenAIClient( client_id=client_id, client_secret=client_secret, proxy=proxy, verify=verify_certificate ) # Execute the respective command function based on the command name got from the Demisto. if command == "test-module": return_results(test_module(client)) elif command == "fetch-incidents": last_run = demisto.getLastRun() next_run, incidents = fetch_incidents(client, last_run, params) demisto.info(f"Fetched {len(incidents)} new incidents") demisto.incidents(incidents) demisto.setLastRun(next_run) elif command in demisto_commands: args = demisto.args() if command == "dataminrpulse-alerts-get": args.update({"watchlist_names": watchlist_names}) remove_nulls_from_dictionary(trim_spaces_from_args(args)) return_results(demisto_commands[command](client, args)) else: raise NotImplementedError(f"The command {command} is not implemented.") # Log exceptions and return errors except Exception as e: demisto.error(traceback.format_exc()) # Print the traceback. return_error(f"Failed to execute {command} command.\nError: \n{str(e)}") if __name__ in ("__main__", "__builtin__", "builtins"): # pragma: no cover main()