import demistomock as demisto # noqa from CommonServerPython import * # noqa # pylint: disable=unused-wildcard-import from CommonServerUserPython import * # noqa """ IMPORTS """ import urllib3 import traceback from json.decoder import JSONDecodeError from typing import Any, cast # Disable insecure warnings urllib3.disable_warnings() # noqa # pylint: disable=no-member """ CONSTANTS """ SOCRADAR_API_ENDPOINT = "https://platform.socradar.com/api" SOCRADAR_SEVERITIES = ["Unknown", "Info", "Low", "Medium", "High", "Critical"] EXCLUDED_INCIDENT_FIELDS = ("extra_info", "alarm_notification_texts") MAX_INCIDENTS_TO_FETCH = 500 MESSAGES: dict[str, str] = { "BAD_REQUEST_ERROR": "An error occurred while fetching the data.", "AUTHORIZATION_ERROR": "Authorization Error: make sure API Key is correctly set.", "RATE_LIMIT_EXCEED_ERROR": "Rate limit has been exceeded. Please make sure your your API key's rate limit is adequate.", } """ HELPER FUNCTIONS """ def parse_int_or_raise(str_to_parse: Any, error_msg=None) -> int: """Parse a string to integer. Raise ValueError exception if fails with given error_msg""" try: res = int(str_to_parse) except (TypeError, ValueError): if not error_msg: error_msg = f"Error while parsing integer! Provided string: {str_to_parse}" raise ValueError(error_msg) return res def convert_to_demisto_severity(severity: str) -> Union[int, float]: """Maps SOCRadar severity to Cortex XSOAR severity Converts the SOCRadar alert severity level ('INFO', 'LOW', 'MEDIUM', 'HIGH', 'CRITICAL') to Cortex XSOAR incident severity (1 to 4) for mapping. :type severity: ``str`` :param severity: severity as returned from the SOCRadar API (str) :return: Cortex XSOAR Severity (1 to 4) :rtype: ``int`` """ return { "INFO": IncidentSeverity.INFO, "LOW": IncidentSeverity.LOW, "MEDIUM": IncidentSeverity.MEDIUM, "HIGH": IncidentSeverity.HIGH, "CRITICAL": IncidentSeverity.CRITICAL, "UNKNOWN": IncidentSeverity.UNKNOWN, }[severity] """ CLIENT CLASS """ class Client(BaseClient): """Client class to interact with the service API This Client implements API calls, and does not contain any XSOAR logic. Should only do requests and return data. It inherits from BaseClient defined in CommonServer Python. Most calls use _http_request() that handles proxy, SSL verification, etc. For this implementation, no special attributes defined """ def __init__(self, base_url, api_key, socradar_company_id, verify, proxy): super().__init__(base_url, verify=verify, proxy=proxy) self.api_key = api_key self.socradar_company_id = socradar_company_id def check_auth(self): suffix = f"/company/{self.socradar_company_id}/incidents/check/auth" api_params = {"key": self.api_key} response = self._http_request( method="GET", url_suffix=suffix, params=api_params, error_handler=self.handle_error_response, ) return response def search_incidents( self, resolution_status: Optional[str], fp_status: Optional[str], severity: Optional[List[str]], incident_main_type: Optional[str], incident_sub_type: Optional[str], max_results: Optional[int], start_date: Optional[int], ): """Searches for SOCRadar incidents using the '/incidents/latest' API endpoint All the parameters are passed directly to the API as HTTP GET parameters in the request :type resolution_status: ``Optional[str]`` :param resolution_status: Resolution status of the incident to search for. Options are: 'All', 'Resolved', 'Not Resolved' :type fp_status: ``Optional[str]`` :param fp_status: False Positive status of the incident to search for. Options are: 'All', 'FP', 'Not FP' :type severity: ``Optional[List[str]]`` :param severity: severity of the alert to search for. Options are: "Low", "Medium", "High", "Critical" :type incident_main_type: ``Optional[str]`` :param incident_main_type: main type of incidents to search for. There is no list of predefined types. :type incident_sub_type: ``Optional[str]`` :param incident_sub_type: sub type of incidents to search for. There is no list of predefined types. :type max_results: ``Optional[int]`` :param max_results: maximum number of results to return :type start_date: ``Optional[int]`` :param start_date. : start timestamp (epoch in seconds) for the alert search """ api_params: dict[str, Any] = {"key": self.api_key} if resolution_status and resolution_status.lower() != "all": api_params["is_resolved"] = resolution_status.lower() == "resolved" if fp_status and fp_status.lower() != "all": api_params["is_false_positive"] = fp_status.lower() == "fp" if incident_main_type: api_params["incident_main_type"] = incident_main_type if incident_sub_type: api_params["incident_sub_type"] = incident_sub_type if severity: api_params["severity"] = ",".join(severity) if max_results: api_params["limit"] = max_results if start_date: api_params["start_date"] = start_date suffix = f"/company/{self.socradar_company_id}/incidents/v2" response = self._http_request( method="GET", url_suffix=suffix, params=api_params, timeout=60, error_handler=self.handle_error_response, ) if not response.get("is_success"): message = f"Error while getting API response. SOCRadar API Response: {response.get('message', '')}" raise DemistoException(message=message) return response.get("data") if response else [] def mark_incident_as_false_positive(self, incident_id: int, comments: Optional[str]): """Sends a request that marks incident as false positive in SOCRadar platform using '/company/{company_id}/incidents/fp'. All the parameters are passed directly to the API as HTTP GET parameters in the request :type incident_id: ``int`` :param incident_id: SOCRadar incident ID of particular incident to that will be used to mark it as false positive. :type comments: ``Optional[str]`` :param comments: Possible comments of the mark as false positive action which will be sent to SOCRadar. """ api_params: dict[str, str] = {"key": self.api_key} json_data = {"alarm_ids": [incident_id], "comments": comments} suffix = f"/company/{self.socradar_company_id}/incidents/fp" response = self._http_request( method="POST", url_suffix=suffix, params=api_params, json_data=json_data, timeout=60, error_handler=self.handle_error_response, ) return response def mark_incident_as_resolved(self, incident_id: int, comments: Optional[str]): """Sends a request that marks incident as resolved in SOCRadar platform using '/company/{company_id}/incidents/resolve'. All the parameters are passed directly to the API as HTTP GET parameters in the request :type incident_id: ``int`` :param incident_id: SOCRadar incident ID of particular incident to that will be used to mark it as resolved. :type comments: ``Optional[str]`` :param comments: Possible comments of the mark as resolved action which will be sent to SOCRadar. """ api_params: dict[str, str] = {"key": self.api_key} json_data: dict[str, Any] = {"alarm_ids": [incident_id], "comments": comments} suffix = f"/company/{self.socradar_company_id}/incidents/resolve" response = self._http_request( method="POST", url_suffix=suffix, params=api_params, json_data=json_data, timeout=60, error_handler=self.handle_error_response, ) return response @staticmethod def handle_error_response(response) -> None: """Handles API response to display descriptive error messages based on status code :param response: SOCRadar API response. :return: DemistoException for particular error code. """ error_reason = "" try: json_resp = json.loads(response.text) error_reason = json_resp.get("error") or json_resp.get("message") except JSONDecodeError: pass status_code_messages = { 400: f"{MESSAGES['BAD_REQUEST_ERROR']} Reason: {error_reason}", 401: MESSAGES["AUTHORIZATION_ERROR"], 404: f"{MESSAGES['BAD_REQUEST_ERROR']} Reason: {error_reason}", 429: MESSAGES["RATE_LIMIT_EXCEED_ERROR"], } if response.status_code in status_code_messages: demisto.debug(f"Response Code: {response.status_code}, Reason: {status_code_messages[response.status_code]}") raise DemistoException(status_code_messages[response.status_code]) else: raise DemistoException(response.raise_for_status()) """ COMMAND FUNCTIONS """ def test_module(client: Client) -> 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: ``Client`` :param client: client to use :return: 'ok' if test passed, anything else will fail the test. :rtype: ``str`` """ client.check_auth() return "ok" def fetch_incidents( client: Client, max_results: int, last_run: dict[str, int], first_fetch_time: Optional[int], resolution_status: Optional[str], fp_status: Optional[str], severity: List[str], incident_main_type: Optional[str], incident_sub_type: Optional[str], include_company_id: bool = False, ) -> tuple[dict[str, int], List[dict]]: """ :type client: ``Client`` :param client: SOCRadar client to use :type max_results: ``int`` :param max_results: Maximum numbers of incidents per fetch :type last_run: ``Optional[Dict[str, int]]`` :param last_run: A dict with a key containing the latest incident created time we got from last fetch :type first_fetch_time: ``Optional[int]`` :param first_fetch_time: If last_run is None (first time we are fetching), it contains the timestamp that tells when to start fetching incidents :type resolution_status: ``Optional[str]`` :param resolution_status: Resolution status of the incident to search for. Options are: 'All', 'Resolved', 'Not Resolved' :type fp_status: ``Optional[str]`` :param fp_status: False positive status of the incident to search for. Options are: 'All', 'FP', 'Not FP' :type severity: ``List[str]`` :param severity: severity level(s) of the incidents to fetch. Options are: "Info", "Medium", "Low", "High" :type incident_main_type: ``Optional[str]`` :param incident_main_type: Main type of incidents to search for. There is no list of predefined types :type incident_sub_type: ``Optional[str]`` :param incident_sub_type: Sub type of incidents to search for. There is no list of predefined types :return: A tuple containing two elements: next_run (``Dict[str, int]``): Contains the timestamp that will be used in ``last_run`` on the next fetch. incidents (``List[dict]``): List of incidents that will be created in XSOAR :rtype: ``Tuple[Dict[str, int], List[dict]]`` """ # Get the last fetch time, if exists # last_run is a dict with a single key, called last_fetch last_fetch = last_run.get("last_fetch") # Handle first fetch time if last_fetch is None: # if missing, use what provided via first_fetch_time last_fetch = first_fetch_time else: # otherwise use the stored last fetch last_fetch = int(last_fetch) # for type checking, making sure that latest_created_time is int latest_created_time = cast(int, last_fetch) # Initialize an empty list of incidents to return # Each incident is a dict with a string as a key incidents: List[dict[str, Any]] = [] # Check if severity contains allowed values, use all if default if severity and not all(s in SOCRADAR_SEVERITIES for s in severity): raise ValueError(f"severity must be a comma-separated value with the following options: {','.join(SOCRADAR_SEVERITIES)}") alerts = client.search_incidents( incident_main_type=incident_main_type, incident_sub_type=incident_sub_type, resolution_status=resolution_status, fp_status=fp_status, max_results=max_results, start_date=last_fetch, severity=severity, ) for alert in alerts: insert_date_str = alert.get("insert_date", "").split(".")[0] insert_date = datetime.strptime(insert_date_str, "%Y-%m-%dT%H:%M:%S") insert_date_utc = insert_date.replace(tzinfo=timezone.utc) incident_created_time = int(insert_date_utc.timestamp()) # to prevent duplicates, we are only adding incidents with creation_time > last fetched incident if last_fetch and incident_created_time <= last_fetch: continue alarm_texts = alert.get("alarm_notification_texts") or {} alarm_type_details = alert.get("alarm_type_details") or {} alarm_title = alarm_texts.get("alarm_title") or alarm_type_details.get("alarm_generic_title", "") incident_content = alarm_texts.get("alarm_text", "") alert = { **{key: value for key, value in alert.items() if key not in EXCLUDED_INCIDENT_FIELDS}, "alarm_notification_texts": {"alarm_title": alarm_title}, } incident_name = f"{alarm_title} - [#{alert.get('id', '')}]" alert_assets = " || ".join(alert.get("alarm_assets") or []) alert_related_assets = "" for related_asset_dict in alert.get("alarm_related_assets") or []: related_asset_key, related_asset_value_list = ( related_asset_dict.get("key"), related_asset_dict.get("value") or [], ) related_asset_value_list: List[str] = list(filter(None, related_asset_value_list)) if related_asset_key and related_asset_value_list: related_asset_value_list = [str(value) for value in related_asset_value_list] alert_related_assets += f"{related_asset_key}: {' || '.join(related_asset_value_list)}\n" alert_related_entities = "" for related_entity_dict in alert.get("alarm_related_entities") or []: related_entity_key, related_entity_value_list = ( related_entity_dict.get("key"), related_entity_dict.get("value") or [], ) related_entity_value_list: List[str] = list(filter(None, related_entity_value_list)) if related_entity_key and related_entity_value_list: related_entity_value_list = [str(value) for value in related_entity_value_list] alert_related_entities += f"{related_entity_key}: {' || '.join(related_entity_value_list)}\n" incident_link = f"https://platform.socradar.com/company/{client.socradar_company_id}/incidents/{alert.get('id', '')}" custom_fields: dict[str, Any] = { "socradarincidentassets": alert_assets, "socradarmitigation": alert.get("alarm_mitigation", ""), "socradarrelatedassets": alert_related_assets, "socradarrelatedentities": alert_related_entities, "incidentlink": incident_link, "socradarincidentcontent": incident_content, } if include_company_id: alert["company_id"] = client.socradar_company_id custom_fields["socradarcompanyid"] = client.socradar_company_id demisto.debug(f"[SOCRadar] Added company_id to incident: {client.socradar_company_id}") demisto.debug(f"[SOCRadar] CustomFields keys: {list(custom_fields.keys())}") incident = { "name": incident_name, "occurred": timestamp_to_datestring(incident_created_time * 1000, is_utc=True), "rawJSON": json.dumps(alert), "severity": convert_to_demisto_severity(alert.get("alarm_risk_level", "UNKNOWN")), "CustomFields": custom_fields, } incidents.append(incident) # Update last run and add incident if the incident is newer than last fetch if incident_created_time > latest_created_time: latest_created_time = incident_created_time # Save the next_run as a dict with the last_fetch key to be stored next_run = {"last_fetch": latest_created_time} return next_run, incidents def mark_incident_as_fp_command(client: Client, args: dict[str, str]) -> CommandResults: """Sends a request that marks incident as false positive in SOCRadar platform. :type client: ``Client`` :param client: client to use :type args: Dict[str, str] :param args: contains all arguments for mark_incident_as_fp_command :return: A ``CommandResults`` object that is then passed to ``return_results`` :rtype: ``CommandResults`` """ incident_id = parse_int_or_raise(args.get("socradar_incident_id", "")) comments = args.get("comments", "") raw_response = client.mark_incident_as_false_positive(incident_id=incident_id, comments=comments) if raw_response.get("is_success"): message = f"SOCRadar API Response: {raw_response.get('message', '')}" else: message = f"Error while getting API response. SOCRadar API Response: {raw_response.get('message', '')}" raise DemistoException(message=message) return CommandResults(readable_output=message, raw_response=raw_response) def mark_incident_as_resolved_command(client: Client, args: dict[str, str]) -> CommandResults: """Sends a request that marks incident as resolved in SOCRadar platform. :param client: client to use :type client: ``Client`` :type args: Dict[str, str] :param args: contains all arguments for mark_incident_as_resolved_command :return: A ``CommandResults`` object that is then passed to ``return_results`` :rtype: ``CommandResults`` """ incident_id = parse_int_or_raise(args.get("socradar_incident_id", "")) comments = args.get("comments", "") raw_response = client.mark_incident_as_resolved(incident_id=incident_id, comments=comments) if raw_response.get("is_success"): message = f"SOCRadar API Response: {raw_response.get('message', '')}" else: message = f"Error while getting API response. SOCRadar API Response: {raw_response.get('message', '')}" raise DemistoException(message=message) return CommandResults(readable_output=message, raw_response=raw_response) """ MAIN FUNCTION """ def main() -> None: """main function, parses params and runs command functions :return: :rtype: """ api_key_param = demisto.params().get("apikey") if isinstance(api_key_param, dict): api_key = api_key_param.get("password", "") else: api_key = api_key_param socradar_company_id = demisto.params().get("socradar_company_id") base_url = SOCRADAR_API_ENDPOINT first_fetch_time = arg_to_datetime( arg=demisto.params().get("first_fetch", "40 days").strip(), arg_name="First fetch time", required=True, ) first_fetch_timestamp = int(first_fetch_time.timestamp()) if first_fetch_time else 0 assert isinstance(first_fetch_timestamp, int) verify_certificate = argToBoolean(demisto.params().get("insecure", False)) proxy = demisto.params().get("proxy", False) demisto.debug(f"Command being called is {demisto.command()}") try: client = Client( base_url=base_url, api_key=api_key, socradar_company_id=socradar_company_id, verify=verify_certificate, proxy=proxy, ) command = demisto.command() if command == "test-module": return_results(test_module(client)) elif command == "fetch-incidents": resolution_status = demisto.params().get("resolution_status") fp_status = demisto.params().get("fp_status") incident_main_type = demisto.params().get("incident_main_type") incident_sub_type = demisto.params().get("incident_sub_type") severity = demisto.params().get("severity") max_results = arg_to_number( arg=demisto.params().get("max_fetch"), arg_name="max_fetch", required=False, ) if not max_results or max_results > MAX_INCIDENTS_TO_FETCH: max_results = MAX_INCIDENTS_TO_FETCH include_company_id = argToBoolean(demisto.params().get("include_company_id", False)) demisto.debug(f"[SOCRadar] include_company_id={include_company_id}, socradar_company_id={socradar_company_id}") next_run, incidents = fetch_incidents( client=client, max_results=max_results, last_run=demisto.getLastRun(), # getLastRun() gets the last run dict first_fetch_time=first_fetch_timestamp, resolution_status=resolution_status, fp_status=fp_status, severity=severity, incident_main_type=incident_main_type, incident_sub_type=incident_sub_type, include_company_id=include_company_id, ) demisto.setLastRun(next_run) demisto.incidents(incidents) elif command == "socradar-mark-incident-fp": return_results(mark_incident_as_fp_command(client=client, args=demisto.args())) elif command == "socradar-mark-incident-resolved": return_results(mark_incident_as_resolved_command(client=client, args=demisto.args())) except Exception as e: demisto.error(traceback.format_exc()) return_error(f"Failed to execute {demisto.command()} command.\nError:\n{str(e)}") """ ENTRY POINT """ if __name__ in ("__main__", "__builtin__", "builtins"): main()