import demistomock as demisto # noqa: F401 from CommonServerPython import * # noqa: F401 """ IMPORTS """ from datetime import UTC import urllib3 urllib3.disable_warnings() """Cyberpion Integration for Cortex XSOAR (aka Demisto) """ """ CONSTANTS """ DATE_FORMAT = "%Y-%m-%d %H:%M:%S.%f %Z" DEFAULT_MAX_INCIDENTS_TO_FETCH = 200 CONNECTION_TIMEOUT = 30.0 READ_TIMEOUT = 30.0 VALID_STATUS_CODES = (200,) NUM_OF_RETRIES = 3 BACKOFF_FACTOR = 1.0 # see documentation in CommonServerPython._http_request ACTION_ITEM_TYPE_NAME = "cyberpion_action_item" """ CLIENT CLASS """ class Client(BaseClient): """Client class to interact with the service API This Client implements API calls, and does not contain any Demisto 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 get_domain_state(self, domain: str): params = {"verbosity": "details", "domain": domain} demisto.debug(f"getting domain state for domain- {domain}") http_response = self._http_request( method="GET", url_suffix="/domainstate/", params=params, resp_type="json", ok_codes=VALID_STATUS_CODES, timeout=(CONNECTION_TIMEOUT, READ_TIMEOUT), retries=NUM_OF_RETRIES, backoff_factor=BACKOFF_FACTOR, raise_on_status=True, ) if "results" not in http_response: raise Exception(f"bad response from server!. response: {json.dumps(http_response, indent=2)}") http_response = http_response["results"] if len(http_response) == 0: demisto.error(f"no response from server for domain: {domain}") return {} http_response = http_response[0] demisto.debug(f"after getting domain state for domain- {domain}") reverse_ips = http_response.get("ips") if reverse_ips is None: raise Exception(f"in server's response: ips is none. response: {json.dumps(http_response, indent=2)}") if type(reverse_ips) is dict: formatted_reverse_ips = "\n".join([f"{k}: {v}" for k, v in reverse_ips.items()]) else: formatted_reverse_ips = reverse_ips http_response["ips"] = formatted_reverse_ips domain_types = http_response.get("domain_types") if domain_types is None: raise Exception(f"in server's response: domain_types is none. response: {json.dumps(http_response, indent=2)}") domain_info = "" for idx, domain_type in enumerate(domain_types, start=1): domain_info += f"{idx}.\n" domain_info += "\n".join([f"{k}: {v}" for k, v in domain_type.items()]) http_response["domain_types"] = domain_info return http_response def get_action_items( self, min_severity: int, alert_types: list = None, show_only_active=True, max_fetch: int = None, last_fetched_creation_time: str = None, domain: str = None, ) -> list[dict]: params = { "verbosity": "details", "urgency__gte": min_severity, "ordering": "creation_time", "is_open": "true" if show_only_active else "false", } if alert_types: params["category"] = ",".join(alert_types) if max_fetch: params["page_size"] = max_fetch if last_fetched_creation_time: params["creation_time__gt"] = last_fetched_creation_time if domain: params["domain"] = domain http_responses = [] # call API params["page"] = str(1) demisto.debug(f"getting action items, domain={domain}") http_response = self._http_request( method="GET", url_suffix="/actionitems/", params=params, resp_type="json", ok_codes=VALID_STATUS_CODES, timeout=(CONNECTION_TIMEOUT, READ_TIMEOUT), retries=NUM_OF_RETRIES, backoff_factor=BACKOFF_FACTOR, raise_on_status=True, ) demisto.debug(f"after getting action items, domain={domain}") if "results" not in http_response: raise Exception("failed to read action items.\nError: got response without 'results' key") results = http_response["results"] for idx, action_item in enumerate(results): technical_det = action_item.get("technical_details", {}) if technical_det is None: raise Exception(f"technical details is none. {json.dumps(action_item, indent=2)}") if type(technical_det) is dict: formatted_technical_details = "\n".join([f"{k}: {v}" for k, v in technical_det.items()]) else: formatted_technical_details = technical_det results[idx]["technical_details"] = formatted_technical_details results[idx]["alert_type"] = ACTION_ITEM_TYPE_NAME http_responses.append(results) demisto.debug(f"finished getting action items, number of pages: {len(http_responses)}, domain={domain}") final_results = [] for response in http_responses: final_results += response return final_results def get_domain_action_items( self, domain: str, min_severity: int, alert_types: list = None, show_only_active=True ) -> dict[str, Any]: # call API return { "Domain": domain, "Vulnerabilities": self.get_action_items( domain=domain, min_severity=min_severity, alert_types=alert_types, show_only_active=show_only_active, max_fetch=None, ), } """ HELPER FUNCTIONS """ def convert_to_demisto_severity(severity: float) -> int: """Maps Cyberpion severity to Cortex XSOAR severity Converts the Cyberpion alert severity level (1 to 10, float) to Cortex XSOAR incident severity (1 to 4) for mapping. :type severity: ``float`` :param severity: severity as returned from the Cyberpion API (float) :return: Cortex XSOAR Severity (1 to 4) :rtype: ``int`` """ # In this case the mapping is straightforward, but more complex mappings # might be required in your integration, so a dedicated function is # recommended. This mapping should also be documented. if 0 <= severity <= 2.5: return 1 elif 2.6 <= severity <= 5: return 2 elif 5.1 <= severity <= 7.5: return 3 elif 7.6 <= severity <= 10: return 4 raise Exception(f"value of severity is not between 0-10. invalid value of severity: {severity}") """ 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`` """ try: client.get_domain_action_items(domain="company1.com", min_severity=2) client.get_action_items(max_fetch=2, min_severity=1, alert_types=["PKI"]) client.get_domain_state("company1.com") except DemistoException as e: if "Forbidden" in str(e): return "Authorization Error: make sure API Key is correctly set" else: raise e return "ok" def fetch_incidents( client: Client, max_fetch: int, min_severity: int, alert_types: list, show_only_active: bool, first_fetch: str = None ) -> tuple[dict[str, str], list[dict]]: """This function retrieves new alerts every interval (default is 1 minute). :type client: ``Client`` :param Client: Cyberpion integration client to use :type max_fetch: ``int`` :param max_fetch: Maximum numbers of incidents per fetch :type min_severity: `int` :param min_severity: minimum severity of the alert to search for. Options are 1 to 10 :type alert_types: ``List[str]`` :param alert_type: type of alerts to search for. There is no list of predefined types :type first_fetch: `str` :param first_fetch: first date to fetch from. if null, all incidents will be fetched :return: A tuple containing two elements: next_run (``Dict[str, str]``): 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]]`` """ last_run_dict = demisto.getLastRun() if "last_fetch" in last_run_dict: last_fetch = last_run_dict["last_fetch"] demisto.debug(f"last fetch: {last_fetch!s}") else: demisto.debug("no previous data... this means this is the first time we are fetching incidents") last_fetch = first_fetch demisto.debug( "Cyberpion fetch incidents last run time\\first fetch: {}".format( str(last_fetch) if last_fetch else "fetching all incidents, without time filter" ) ) action_items = client.get_action_items( max_fetch=max_fetch, min_severity=min_severity, alert_types=alert_types, show_only_active=show_only_active, last_fetched_creation_time=last_fetch, ) incidents = [] for action_item in action_items: creation_date = action_item["creation_time"] # must be string of a DATE_FORMAT iso_format_data = datetime.strptime(creation_date, DATE_FORMAT).replace(tzinfo=UTC).isoformat() incident = { "name": "{} - {}".format(action_item["title"], action_item["domain"]), # name is required field, must be set "occurred": iso_format_data, "rawJSON": json.dumps(action_item), "severity": convert_to_demisto_severity(action_item["urgency"]), } # put in last_incident_date the last action_items creation date. assuming it's ordered by creation date # last_incident_date = creation_date incidents.append(incident) # last incident's time added to new_last_run_dict, so we can next time ask for incidents with creation_time__gt this time if len(action_items) > 0: last_incident_date = action_items[-1]["creation_time"] else: # if no action items from last_incident_date to now, keep asking next time for (new incidents) from # last_incident_date and on last_incident_date = last_fetch new_last_run_dict = {"last_fetch": last_incident_date} return new_last_run_dict, incidents def get_domain_state_command(client: Client, args: dict[str, Any]) -> CommandResults: domain = args.get("domain") if not domain: raise ValueError("no domain specified") demisto.debug(f"getting domain state {domain}") domain_state = client.get_domain_state(domain) demisto.debug(f"creating domain state table for domain {domain}") markdown = "### Cyberpion\n" markdown += tableToMarkdown( "Domain State", domain_state, headers=[ "id", "domain", "ips", "risk_rank", "vuln_count", "cname_chain", "domain_types", "discovery_date", ], ) demisto.debug(f"finished creating domain state table for domain {domain}") return CommandResults( readable_output=markdown, outputs_prefix="Cyberpion", outputs_key_field="id", outputs={"DomainState": domain_state} ) def get_domain_action_items_command( client: Client, args: dict[str, Any], min_severity: int, alert_types: list = None, show_only_active: bool = True ) -> CommandResults: domain = args.get("domain") if not domain: raise ValueError("no domain specified") demisto.debug(f"getting action items for domain {domain}") domain_data = client.get_domain_action_items( domain=domain, min_severity=min_severity, show_only_active=show_only_active, alert_types=alert_types, ) demisto.debug(f"creating action items table data for domain {domain}") markdown = "### Cyberpion\n" markdown += tableToMarkdown( "Action Items", domain_data["Vulnerabilities"], headers=[ "domain", "category", "urgency", "is_open", "creation_time", "link", "title", "impact", "summary", "solution", "description", "technical_details", ], ) demisto.debug(f"finished creating table data for domain {domain}. returning command result") return CommandResults( readable_output=markdown, outputs_prefix="Cyberpion.DomainData", outputs_key_field="id", outputs=domain_data ) """ MAIN FUNCTION """ def main() -> None: # pragma: no cover """main function, parses params and runs command functions :return: :rtype: """ # get the service API url base_url = demisto.params()["url"] api_key = demisto.params()["apikey"] min_severity = demisto.params()["minSeverity"] # mandatory alert_types = demisto.params()["categories"] # mandatory show_only_active = demisto.params()["ShowOnlyOpen"] # mandatory verify_certificate = not demisto.params().get("insecure", False) proxy = demisto.params().get("proxy", False) demisto.debug(f"Command being called is {demisto.command()}") try: headers = {"Authorization": f"Token {api_key}"} client = Client(base_url=base_url, verify=verify_certificate, headers=headers, proxy=proxy) if demisto.command() == "test-module": # This is the call made when pressing the integration Test button. result = test_module(client) return_results(result) elif demisto.command() == "cyberpion-get-domain-state": return_results(get_domain_state_command(client, demisto.args())) elif demisto.command() == "cyberpion-get-domain-action-items": return_results( get_domain_action_items_command( client, demisto.args(), min_severity=min_severity, alert_types=alert_types, show_only_active=show_only_active ) ) elif demisto.command() == "fetch-incidents": # Set and define the fetch incidents command to run after activated via integration settings. max_fetch = demisto.params().get("maxFetch") first_fetch: str = demisto.params().get("first_fetch") if first_fetch: months_back = datetime.now() - timedelta(days=30 * int(first_fetch)) first_fetch = datetime.strftime(months_back, DATE_FORMAT) if not max_fetch: max_fetch = DEFAULT_MAX_INCIDENTS_TO_FETCH try: max_fetch = int(max_fetch) if max_fetch > 500 or max_fetch < 1: raise ValueError except ValueError: raise ValueError("max_fetch must be an integer between 1 to 500") if max_fetch > DEFAULT_MAX_INCIDENTS_TO_FETCH: max_fetch = DEFAULT_MAX_INCIDENTS_TO_FETCH new_last_run_dict, incidents = fetch_incidents( client=client, max_fetch=max_fetch, min_severity=min_severity, show_only_active=show_only_active, alert_types=alert_types, first_fetch=first_fetch, ) # create incidents demisto.incidents(incidents) # saves next_run for the time fetch-incidents is invoked demisto.setLastRun(new_last_run_dict) else: raise NotImplementedError(f"no such command: {demisto.command()}") # Log exceptions and return errors except Exception as e: demisto.error(traceback.format_exc()) # print the traceback return_error(f"Cyberpion integration: Failed to execute {demisto.command()} command.\nError:\n{e!s}") """ ENTRY POINT """ if __name__ in ("__main__", "__builtin__", "builtins"): main()