from copy import deepcopy from typing import Any import demistomock as demisto # noqa: F401 import urllib3 from CommonServerPython import * # noqa: F401 from CommonServerUserPython import * # disable insecure warnings urllib3.disable_warnings() DEFAULT_PAGE = 1 DEFAULT_LIMIT = 50 DEFAULT_MAX_FETCH = DEFAULT_LIMIT DEFAULT_EVENTS_FETCH = DEFAULT_LIMIT DEFAULT_EVENT_TYPE = "application" DEFAULT_FIRST_FETCH = "7 days" MAX_LIMIT = 100 MAX_FETCH = 200 MAX_EVENTS_FETCH = 200 TIME_PERIOD_MAPPING = { "Last 60 Minutes": 3600, "Last 24 Hours": 86400, "Last 7 Days": 604800, "Last 30 Days": 2592000, "Last 60 Days": 5184000, "Last 90 Days": 7776000, } class Client(BaseClient): """ Client for Netskope RESTful API. Args: base_url (str): The base URL of Netskope. token (str): The token to authenticate against Netskope API. use_ssl (bool): Specifies whether to verify the SSL certificate or not. use_proxy (bool): Specifies if to use XSOAR proxy settings. """ def __init__(self, base_url: str, token: str, use_ssl: bool, use_proxy: bool): super().__init__(urljoin(base_url, "/api/v1/"), verify=use_ssl, proxy=use_proxy) self._session.params["token"] = token # type: ignore def list_events_request( self, query: str | None = None, event_type: str | None = None, timeperiod: int | None = None, start_time: int | None = None, end_time: int | None = None, insertion_start_time: int | None = None, insertion_end_time: int | None = None, limit: int | None = None, skip: int | None = None, unsorted: bool | None = None, ) -> dict[str, Any]: """ Get events extracted from SaaS traffic and or logs. Args: query (Optional[str]): Free query to filter the events. event_type (Optional[str]): Select events by their type. timeperiod (Optional[int]): Get all events from a certain time period. start_time (Optional[int]): Restrict events to those that have timestamps greater than the provided timestamp. end_time (Optional[int]): Restrict events to those that have timestamps less than or equal to the provided timestamp. insertion_start_time (Optional[int]): Restrict events to those that were inserted to the system after the provided timestamp. insertion_end_time (Optional[int]): Restrict events to those that were inserted to the system before the provided timestamp. limit (Optional[int]): The maximum amount of events to retrieve (up to 10000 events). skip (Optional[int]): The skip number of the events to retrieve (minimum is 1). unsorted (Optional[bool]): If true, the returned data will not be sorted (useful for improved performance). Returns: Dict[str, Any]: Netskope events. """ body = remove_empty_elements( { "query": query, "type": event_type, "timeperiod": timeperiod, "starttime": start_time, "endtime": end_time, "insertionstarttime": insertion_start_time, "insertionendtime": insertion_end_time, "limit": limit, "skip": skip, "unsorted": unsorted, } ) return self._http_request(method="POST", url_suffix="events", json_data=body) def list_alerts_request( self, query: str | None = None, alert_type: str | None = None, acked: bool | None = None, timeperiod: int | None = None, start_time: int | None = None, end_time: int | None = None, insertion_start_time: int | None = None, insertion_end_time: int | None = None, limit: int | None = None, skip: int | None = None, unsorted: bool | None = None, ) -> dict[str, Any]: """ Get alerts generated by Netskope, including policy, DLP, and watch list alerts. Args: query (Optional[str]): Free query to filter the alerts. alert_type (Optional[str]): Select alerts by their type. acked (Optional[bool]): Whether to retrieve acknowledged alerts or not. timeperiod (Optional[int]): Get alerts from certain time period. start_time (Optional[int]): Restrict alerts to those that have timestamps greater than the provided timestamp. end_time (Optional[int]): Restrict alerts to those that have timestamps less than or equal to the provided timestamp. insertion_start_time (Optional[int]): Restrict alerts which have been inserted into the system after the provided timestamp. insertion_end_time (Optional[int]): Restrict alerts which have been inserted into the system before the provided timestamp. limit (Optional[int]): The maximum number of alerts to return (up to 10000). skip (Optional[int]): The skip number of the alerts to retrieve (minimum is 1). unsorted (Optional[bool]): If true, the returned data will not be sorted (useful for improved performance). Returns: Dict[str, Any]: Netskope alerts. """ body = remove_empty_elements( { "query": query, "alert_type": alert_type, "acked": acked, "timeperiod": timeperiod, "starttime": start_time, "endtime": end_time, "insertionstarttime": insertion_start_time, "insertionendtime": insertion_end_time, "limit": limit, "skip": skip, "unsorted": unsorted, } ) return self._http_request(method="POST", url_suffix="alerts", json_data=body) def list_quarantined_files_request( self, start_time: int | None = None, end_time: int | None = None, limit: int | None = None, skip: int | None = None ) -> dict[str, Any]: """ List all quarantined files. Args: start_time (Optional[int]): Get files last modified within a certain time period. end_time (Optional[int]): Get files last modified within a certain time period. limit (Optional[int]): The maximum amount of clients to retrieve (up to 10000). skip (Optional[int]): The skip number of the clients to retrieve (minimum is 1). Returns: Dict[str, Any]: Netskope quarantine files. """ body = remove_empty_elements( {"starttime": start_time, "endtime": end_time, "limit": limit, "skip": skip, "op": "get-files"} ) return self._http_request(method="POST", url_suffix="quarantine", json_data=body) def get_quarantined_file_request(self, quarantine_profile_id: str, file_id: str) -> bytes: """ Download a quarantined file. Args: quarantine_profile_id (str): The ID of quarantine profile. file_id (str): The ID of the quarantined file. Returns: bytes: The quarantined file content. """ body = {"quarantine_profile_id": quarantine_profile_id, "file_id": file_id, "op": "download-url"} return self._http_request(method="POST", url_suffix="quarantine", json_data=body, resp_type="content") def update_quarantined_file_request(self, quarantine_profile_id: str, file_id: str, action: str) -> None: """ Take an action on a quarantined file. Args: quarantine_profile_id (str): The profile id of the quarantined file. file_id (str): The id of the quarantined file. action (str): Action to be performed on a quarantined. """ body = {"quarantine_profile_id": quarantine_profile_id, "file_id": file_id, "action": action, "op": "take-action"} self._http_request(method="POST", url_suffix="quarantine", json_data=body, resp_type="text") def update_url_list_request(self, name: str, urls: list[str]) -> None: """ Update the URL List with the values provided. Args: name (str): Name of an existing URL List shown in the Netskope UI on the URL List skip. urls (List[str]): The content of the URL list. """ body = {"name": name, "list": ",".join(urls)} self._http_request(method="POST", url_suffix="updateUrlList", json_data=body) def update_file_hash_list_request(self, name: str, hashes: list[str]) -> None: """ Update file hash list with the values provided. Args: name (str): Name of an existing file hash list shown in the Netskope UI on the file hash list skip. hashes (str): List of file hashes (md5 or sha256). """ body = {"name": name, "list": ",".join(hashes)} return self._http_request(method="POST", url_suffix="updateFileHashList", json_data=body) def list_clients_request(self, query: str | None = None, limit: int | None = None, skip: int | None = None) -> dict[str, Any]: """ Get information about the Netskope clients. Args: query (Optional[str]): Free query on the clients, based on the client fields. limit (Optional[int]): The maximum amount of clients to retrieve (up to 10000). skip (Optional[int]): The skip number of the clients to retrieve (minimum is 1). Returns: Dict[str, Any]: The clients information. """ body = remove_empty_elements({"query": query, "limit": limit, "skip": skip}) return self._http_request(method="POST", url_suffix="clients", params=body) def _http_request(self, *args, **kwargs): response = super()._http_request(*args, **kwargs) if isinstance(response, dict) and "errors" in response: errors = "\n".join(response["errors"]) raise DemistoException(f"Invalid API call: {errors}", res=response) return response def arg_to_boolean(arg: str | None) -> bool | None: """ Converts an XSOAR argument to a Python boolean or None. Args: arg (Optional[str]): The argument to convert. Returns: Optional[bool]: A boolean if arg can be converted, or None if arg is None. """ if arg is None: return None return argToBoolean(arg) def arg_to_seconds_timestamp(arg: str | None): """ Converts an XSOAR date string argument to a timestamp in seconds. Args: arg (Optional[str]): The argument to convert. Returns: Optional[int]: A timestamp if arg can be converted, or None if arg is None. """ if arg is None: return None return date_to_seconds_timestamp(arg_to_datetime(arg)) # type: ignore def date_to_seconds_timestamp(date_str_or_dt: Union[str, datetime]) -> int: """ Converts date string or datetime object to a timestamp in seconds. Args: date_str_or_dt (Union[str, datetime]): The datestring or datetime. Returns: int: The timestamp in seconds. """ return date_to_timestamp(date_str_or_dt) // 1000 def validate_time_arguments( start_time: int | None = None, end_time: int | None = None, insertion_start_time: int | None = None, insertion_end_time: int | None = None, timeperiod: int | None = None, ) -> None: """ Validates time arguments from the user. The user must provide one of the following: - start_time and end_time. - insertion_start_time and insertion_end_time. - timeperiod. Args: start_time (Optional[int], optional): The start time to fetch from the API. end_time (Optional[int], optional): The end time to fetch from the API. insertion_start_time (Optional[int], optional): The insertion start time to fetch from the API. insertion_end_time (Optional[int], optional): The insertion end time to fetch from the API. timeperiod (Optional[str], optional): The timeperiod to fetch from the API. Raises: DemistoException: The user did not provide valid timestamp. """ combination = (all((start_time, end_time)), all((insertion_start_time, insertion_end_time)), bool(timeperiod)) if not any(combination): raise DemistoException( "Missing time arguments. Please provide start_time and end_time, " "or insertion_start_time and or insertion_end_time or timeperiod." ) if combination.count(True) > 1: raise DemistoException( "Invalid time arguments. Please provide only start_time and end_time, " "or insertion_start_time and or insertion_end_time or timeperiod. " "You must not combine between the mentioned options." ) def validate_fetch_params( max_fetch: int, max_events_fetch: int, fetch_events: bool, first_fetch: str, event_types: list[str] ) -> None: """ Validates the parameters for fetch incident command. Args: max_fetch: (int): The maximum number of incidents for one fetch. max_events_fetch (int) The maximum number of events per incident for one fetch. fetch_events (bool): Whether or not fetch events when fetching incident. first_fetch: (str): First fetch time in words. """ if first_fetch: arg_to_datetime(first_fetch) # verify that it is a date. if max_fetch > MAX_FETCH: return_error(f"The Maximum number of incidents per fetch should not exceed {MAX_FETCH}.") if fetch_events and max_events_fetch > MAX_EVENTS_FETCH: return_error(f"The Maximum number of events for each incident per fetch should not exceed {MAX_EVENTS_FETCH}.") if not isinstance(event_types, list): return_error("The fetched event types must be a list.") def get_pagination_readable_message(header: str, page: int, limit: int) -> str: return f"{header}\n Current page size: {limit}\n Showing page {page} out of others that may exist." def get_pagination_arguments(args: dict[str, Any]) -> tuple[int, int, int]: """ Gets and validates pagination arguments for client (skip and limit). Args: args (Dict[str, Any]): The command arguments (page and limit). Returns: Tuple[int, int]: The page, calculated skip and limit after validation. """ page = arg_to_number(args.get("page")) or DEFAULT_PAGE limit = arg_to_number(args.get("limit")) or DEFAULT_LIMIT if page < 1: raise DemistoException("Page argument must be greater than 1") if not 1 <= limit <= MAX_LIMIT: raise DemistoException(f"Limit argument must be between 1 to {MAX_LIMIT}") return page, (page - 1) * limit, limit def list_events_command(client: Client, args: dict[str, str]) -> CommandResults: """ Get events extracted from SaaS traffic and or logs. Args: client (client): The Netskope client. args (Dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: Command results with raw response, outputs and readable outputs. """ query = args.get("query") event_type = args["event_type"] timeperiod = TIME_PERIOD_MAPPING.get(args.get("timeperiod", "")) start_time = arg_to_seconds_timestamp(args.get("start_time")) end_time = arg_to_seconds_timestamp(args.get("end_time")) insertion_start_time = arg_to_seconds_timestamp(args.get("insertion_start_time")) insertion_end_time = arg_to_seconds_timestamp(args.get("insertion_end_time")) page, skip, limit = get_pagination_arguments(args) unsorted = arg_to_boolean(args.get("unsorted")) validate_time_arguments( start_time=start_time, end_time=end_time, timeperiod=timeperiod, insertion_start_time=insertion_start_time, insertion_end_time=insertion_end_time, ) response = client.list_events_request( query=query, event_type=event_type, timeperiod=timeperiod, start_time=start_time, end_time=end_time, insertion_start_time=insertion_start_time, insertion_end_time=insertion_end_time, limit=limit, skip=skip, unsorted=unsorted, ) outputs = deepcopy(response["data"]) for event in outputs: event["event_id"] = event["_id"] event["timestamp"] = timestamp_to_datestring(event["timestamp"] * 1000) readable_output = tableToMarkdown( get_pagination_readable_message("Events List:", page=page, limit=limit), outputs, removeNull=True, headers=["event_id", "timestamp", "type", "access_method", "app", "traffic_type"], headerTransform=string_to_table_header, ) return CommandResults( outputs_prefix="Netskope.Event", outputs_key_field="event_id", outputs=outputs, readable_output=readable_output, raw_response=response, ) def list_alerts_command(client: Client, args: dict[str, str]) -> CommandResults: """ Get alerts generated by Netskope, including policy, DLP, and watch list alerts. Args: client (client): The Netskope client. args (Dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: Command results with raw response, outputs and readable outputs. """ query = args.get("query") alert_type = args.get("alert_type") acked = arg_to_boolean(args.get("acked")) timeperiod = TIME_PERIOD_MAPPING.get(args.get("timeperiod", "")) start_time = arg_to_seconds_timestamp(args.get("start_time")) end_time = arg_to_seconds_timestamp(args.get("end_time")) insertion_start_time = arg_to_seconds_timestamp(args.get("insertion_start_time")) insertion_end_time = arg_to_seconds_timestamp(args.get("insertion_end_time")) page, skip, limit = get_pagination_arguments(args) unsorted = arg_to_boolean(args.get("unsorted")) validate_time_arguments( start_time=start_time, end_time=end_time, timeperiod=timeperiod, insertion_start_time=insertion_start_time, insertion_end_time=insertion_end_time, ) response = client.list_alerts_request( query=query, alert_type=alert_type, acked=acked, timeperiod=timeperiod, start_time=start_time, end_time=end_time, insertion_start_time=insertion_start_time, insertion_end_time=insertion_end_time, limit=limit, skip=skip, unsorted=unsorted, ) outputs = deepcopy(response["data"]) for alert in outputs: alert["alert_id"] = alert["_id"] alert["timestamp"] = timestamp_to_datestring(alert["timestamp"] * 1000) readable_output = tableToMarkdown( get_pagination_readable_message("Alerts List:", page=page, limit=limit), outputs, removeNull=True, headers=["alert_id", "alert_name", "alert_type", "timestamp", "action"], headerTransform=string_to_table_header, ) return CommandResults( outputs_prefix="Netskope.Alert", outputs_key_field="alert_id", outputs=outputs, readable_output=readable_output, raw_response=response, ) def list_quarantined_files_command(client: Client, args: dict[str, str]) -> CommandResults: """ List all quarantined files. Args: client (client): The Netskope client. args (Dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: Command results with raw response, outputs and readable outputs. """ start_time = arg_to_seconds_timestamp(args.get("start_time")) end_time = arg_to_seconds_timestamp(args.get("end_time")) page, skip, limit = get_pagination_arguments(args) response = client.list_quarantined_files_request(start_time=start_time, end_time=end_time, limit=limit, skip=skip) outputs = dict_safe_get(response, ["data", "quarantined"]) # type: ignore for output in outputs: # type: ignore for file_output in output["files"]: file_output["quarantine_profile_id"] = output["quarantine_profile_id"] file_output["quarantine_profile_name"] = output["quarantine_profile_name"] outputs: list = sum((output["files"] for output in outputs), []) readable_header = get_pagination_readable_message("Quarantined Files List:", page=page, limit=limit) readable_output = tableToMarkdown( readable_header, outputs, removeNull=True, headers=["quarantine_profile_id", "quarantine_profile_name", "file_id", "original_file_name", "policy"], headerTransform=string_to_table_header, ) return CommandResults( outputs_prefix="Netskope.Quarantine", outputs_key_field="file_id", outputs=outputs, readable_output=readable_output, raw_response=response, ) def get_quarantined_file_command(client: Client, args: dict[str, str]) -> CommandResults: """ Download a quarantined file. Args: client (client): The Netskope client. args (Dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: Command results with raw response, outputs and readable outputs. """ quarantine_profile_id = args["quarantine_profile_id"] file_id = args["file_id"] file_name = args.get("file_name", file_id) response = client.get_quarantined_file_request(quarantine_profile_id=quarantine_profile_id, file_id=file_id) return fileResult(filename=file_name, data=response, file_type=EntryType.FILE) def update_quarantined_file_command(client: Client, args: dict[str, str]) -> CommandResults: """ Take an action on a quarantined file. Args: client (client): The Netskope client. args (Dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: Command results with raw response, outputs and readable outputs. """ quarantine_profile_id = args["quarantine_profile_id"] file_id = args["file_id"] action = args["action"] client.update_quarantined_file_request(quarantine_profile_id=quarantine_profile_id, file_id=file_id, action=action) readable_output = f"## The file {file_id} was successfully {action}ed!" return CommandResults(readable_output=readable_output) def update_url_list_command(client: Client, args: dict[str, str]) -> CommandResults: """ Update the URL List with the values provided. Args: client (client): The Netskope client. args (Dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: Command results with raw response, outputs and readable outputs. """ name = args["name"] urls = argToList(args["urls"]) client.update_url_list_request(name=name, urls=urls) outputs = {"name": name, "URL": urls} readable_output = f'URL List {name}:\n{", ".join(urls)}' return CommandResults( outputs_prefix="Netskope.URLList", outputs_key_field="name", outputs=outputs, readable_output=readable_output ) def update_file_hash_list_command(client: Client, args: dict[str, str]) -> CommandResults: """ Update file hash list with the values provided. Args: client (client): The Netskope client. args (Dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: Command results with raw response, outputs and readable outputs. """ name = args.get("name", "") hashes = argToList(args.get("hash")) client.update_file_hash_list_request(name=name, hashes=hashes) outputs = {"name": name, "hash": hashes} readable_output = f'Hash List {name}:\n{", ".join(hashes)}' return CommandResults( outputs_prefix="Netskope.FileHashList", outputs_key_field="name", outputs=outputs, readable_output=readable_output ) def list_clients_command(client: Client, args: dict[str, str]) -> CommandResults: """ Get information about the Netskope clients. Args: client (client): The Netskope client. args (Dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: Command results with raw response, outputs and readable outputs. """ query = args.get("query") page, skip, limit = get_pagination_arguments(args) response = client.list_clients_request(query=query, limit=limit, skip=skip) outputs = [client["attributes"] for client in response["data"]] for output in outputs: output["client_id"] = output["_id"] readable_header = get_pagination_readable_message("Clients List:", page=page, limit=limit) readable_output = tableToMarkdown( readable_header, outputs, removeNull=True, headers=["client_id", "client_version", "device_id", "user_added_time"], headerTransform=string_to_table_header, ) return CommandResults( outputs_prefix="Netskope.Client", outputs_key_field="client_id", outputs=outputs, readable_output=readable_output, raw_response=response, ) def list_host_associated_user_command(client: Client, args: dict[str, str]) -> CommandResults: """ List all users of certain host by its hostname. Args: client (client): The Netskope client. args (Dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: Command results with raw response, outputs and readable outputs. """ hostname = args["hostname"] page, skip, limit = get_pagination_arguments(args) response = client.list_clients_request(query=f"host_info.hostname eq {hostname}", limit=limit, skip=skip) outputs: list = sum((client["attributes"].get("users") for client in response["data"]), []) for output in outputs: output["user_id"] = output["_id"] readable_header = get_pagination_readable_message(f"Users Associated With {hostname}:", page=page, limit=limit) readable_output = tableToMarkdown( readable_header, outputs, removeNull=True, headers=["user_id", "username", "user_source"], headerTransform=string_to_table_header, ) return CommandResults( outputs_prefix="Netskope.User", outputs_key_field="user_id", outputs=outputs, readable_output=readable_output, raw_response=response, ) def list_user_associated_host_command(client: Client, args: dict[str, str]) -> CommandResults: """ List all hosts related to a certain username. Args: client (client): The Netskope client. args (Dict[str, Any]): Command arguments from XSOAR. Returns: CommandResults: Command results with raw response, outputs and readable outputs. """ username = args["username"] page, skip, limit = get_pagination_arguments(args) response = client.list_clients_request(query=f"username eq {username}", limit=limit, skip=skip) outputs = [] for client in response["data"]: attributes = client["attributes"] agent_status = dict_safe_get(attributes, ["last_event", "status"]) outputs.append({"agent_status": agent_status, **attributes["host_info"]}) readable_header = get_pagination_readable_message(f"Hosts Associated With {username}:", page=page, limit=limit) readable_output = tableToMarkdown( readable_header, outputs, removeNull=True, headers=["hostname", "os_version", "agent_status"], headerTransform=string_to_table_header, ) return CommandResults( outputs_prefix="Netskope.Host", outputs_key_field="nsdeviceuid", outputs=outputs, readable_output=readable_output, raw_response=response, ) def test_module( client: Client, max_fetch: int, first_fetch: str, fetch_events: bool, max_events_fetch: int, event_types: list[str] ) -> str: """ Validates all integration parameters, and tests connection to Netskope instance. """ validate_fetch_params(max_fetch, max_events_fetch, fetch_events, first_fetch, event_types) client.list_alerts_request( limit=1, skip=0, start_time=date_to_seconds_timestamp(datetime.now()), end_time=date_to_seconds_timestamp(datetime.now()) ) return "ok" def fetch_multiple_type_events( client: Client, max_fetch: int, start_time: int, event_types: list[str], query: str | None ) -> list[dict[str, Any]]: """ Fetches events from multiple types. The function makes an API call for each type, since the API requires specifying the event type. Args: client (Client): The Netskope client. max_fetch (int): The maximum amount of events to fetch for each type. start_time (int): The time to fetch the events from. event_types (List[str]): The event types to fetch as incidents. query (Optional[str]): Query for filtering the events. Returns: List[Dict[str, Any]]: The fetched events. """ events = [] if event_types: max_fetch = max_fetch // len(event_types) for event_type in event_types: new_events = client.list_events_request( start_time=start_time, end_time=date_to_seconds_timestamp(datetime.now()), limit=max_fetch, unsorted=False, event_type=event_type, query=query, )["data"] for event in new_events: event["event_id"] = event["_id"] event["incident_type"] = event_type events.extend(new_events) return events def fetch_incidents( client: Client, max_fetch: int, first_fetch: str, fetch_events: bool, max_events_fetch: int, event_types: list[str], alerts_query: str | None, events_query: str | None, ) -> None: """ Fetches alerts and events as incidents. Args: client (Client): The Netskope client. max_fetch (int): Maximum number of incidents to fetch. first_fetch (str): The timestamp to fetch the incidents from. max_events_fetch (int): Maximum number of events to fetch. event_types (List[str]): The type of events to fetch. alerts_query (Optional[str]): Query for filtering the fetched alerts. events_query (Optional[str]): Query for filtering the fetched events. """ validate_fetch_params(max_fetch, max_events_fetch, fetch_events, first_fetch, event_types) last_run = demisto.getLastRun() or {} first_fetch: int = arg_to_seconds_timestamp(first_fetch) last_alert_time = last_run.get("last_alert_time") or first_fetch alerts = client.list_alerts_request( start_time=last_alert_time, end_time=date_to_seconds_timestamp(datetime.now()), limit=max_fetch, query=alerts_query, unsorted=False, )["data"] last_event_time = last_run.get("last_event_time") or first_fetch if fetch_events: events = fetch_multiple_type_events( client, max_fetch=max_events_fetch, start_time=last_event_time, event_types=event_types, query=events_query ) else: events = [] incidents = [] for alert in alerts: alert["incident_type"] = alert["alert_type"] incidents.append( {"name": alert["alert_name"], "occurred": timestamp_to_datestring(alert["timestamp"]), "rawJSON": json.dumps(alert)} ) for event in events: incidents.append( {"name": event["event_id"], "occurred": timestamp_to_datestring(event["timestamp"]), "rawJSON": json.dumps(event)} ) # The alerts and events are sorted in descending order. # Also, we increment the timestamp in one second to avoid duplicates. demisto.setLastRun( { "last_alert_time": alerts[0]["timestamp"] + 1 if alerts else last_alert_time, "last_event_time": events[0]["timestamp"] + 1 if events else last_event_time, } ) demisto.incidents(incidents) def main(): params = demisto.params() url = params["url"] token = params["credentials"]["password"] use_ssl = not params.get("insecure", False) use_proxy = params.get("proxy", False) max_fetch = arg_to_number(params.get("max_fetch")) or DEFAULT_MAX_FETCH first_fetch = params.get("first_fetch", DEFAULT_FIRST_FETCH) fetch_events = argToBoolean(params.get("fetch_events", False)) event_types = argToList(params.get("fetch_event_types", DEFAULT_EVENT_TYPE)) max_events_fetch = arg_to_number(params.get("max_events_fetch")) or DEFAULT_EVENTS_FETCH client = Client(url, token, use_ssl, use_proxy) commands = { "netskope-event-list": list_events_command, "netskope-alert-list": list_alerts_command, "netskope-quarantined-file-list": list_quarantined_files_command, "netskope-quarantined-file-get": get_quarantined_file_command, "netskope-quarantined-file-update": update_quarantined_file_command, "netskope-url-list-update": update_url_list_command, "netskope-file-hash-list-update": update_file_hash_list_command, "netskope-client-list": list_clients_command, "netskope-host-associated-user-list": list_host_associated_user_command, "netskope-user-associated-host-list": list_user_associated_host_command, } try: command = demisto.command() if command == "test-module": return_results( test_module( client, max_fetch=max_fetch, first_fetch=first_fetch, fetch_events=fetch_events, max_events_fetch=max_events_fetch, event_types=event_types, ) ) elif command == "fetch-incidents": fetch_incidents( client, max_fetch=max_fetch, first_fetch=first_fetch, fetch_events=fetch_events, max_events_fetch=max_events_fetch, event_types=event_types, alerts_query=demisto.params().get("alerts_query"), events_query=demisto.params().get("events_query"), ) elif command in commands: return_results(commands[command](client, demisto.args())) else: raise NotImplementedError(f"The command {command} does not exist!") except Exception as e: return_error(f"Failed to execute {demisto.command()} command.\nError:\n{e}") if __name__ in ("__main__", "__builtin__", "builtins"): main()