import copy from collections.abc import Callable from http import HTTPStatus from typing import Any, NamedTuple import demistomock as demisto # noqa: F401 from CommonServerPython import * # noqa: F401 MAX_IDS_NUMBER = 289262 DEFAULT_WAIT_TIME = 5 MIN_PAGE_NUM = 1 MAX_PAGE_SIZE = 50 MIN_PAGE_SIZE = 1 MAX_LIMIT = 50 MIN_LIMIT = 1 MAX_FETCH_PER_EVENT_TYPE = 50 JS_NUMBER_LIMIT = 2**53 - 1 TIME_FORMAT = "%d-%m-%Y %H:%M" ALERT_HEADERS = [ "alert_id", "alert_name", "alert_type", "severity", "action", "activity", "type", "acked", "_category_name", "_event_id", "domain", "dst_country", "policy", "port", "protocol", "md5", "timestamp", ] EVENT_HEADERS = { "application": [ "event_id", "alert", "action", "activity", "category", "hostname", "policy", "severity", "type", "useragent", "timestamp", ], "page": [ "event_id", "category", "domain", "page", "type", "user", "appcategory", "bypass_reason", "src_location", "srcip", "dst_location", "dstip", "timestamp", ], "audit": [ "event_id", "audit_log_event", "severity_level", "count", "supporting_data", "timestamp", ], "infrastructure": [ "event_id", "hostname", "timestamp", "serial", "metric_name", "metric_value", ], "network": [ "event_id", "category", "domain", "page", "type", "user", "policy", "action", "protocol", "src_location", "srcip", "dst_location", "dstip", "timestamp", ], } URL_HEADER = [ "id", "name", "urls", "type", "pending", "modify_by", "modify_time", "modify_type", ] INCIDENT_HEADERS = [ "object_id", "status", "severity", "activity", "assignee", "timestamp", "acting_user", "app", "instance_id", "object_type", ] MIRRORING_FIELDS = ["status", "severity", "original_status", "original_severity"] MIRROR_DIRECTION_MAPPING = { "Incoming": "In", "Outgoing": "Out", "Incoming and Outgoing": "Both", } class Pagination(NamedTuple): updated_page: int limit: int pagination_message: str class TimeArgs(NamedTuple): start_time_number: int | None end_time_number: int | None insertion_start_time_number: int | None insertion_end_time_number: int | None class Client(BaseClient): def __init__( self, server_url: str, api_token: str, verify: bool = False, proxy: bool = False, user_mail: str | None = None, ): self.user_mail = user_mail headers = {"Netskope-Api-Token": api_token} super().__init__(base_url=server_url, verify=verify, proxy=proxy, headers=headers) def list_alert( self, page: int, limit: int, alert_type: str | None = None, query: str | None = None, acked: bool | None = None, start_time: int | None = None, end_time: int | None = None, insertion_start_time: int | None = None, insertion_end_time: int | None = None, ) -> dict[str, Any]: """Get alerts generated by Netskope. Args: page (int): Page number of paginated results. limit (int): The maximum number of records to retrieve. alert_type (Optional[str], optional): Alert type. Defaults to None. query (Optional[str], optional): Free query to filter the alerts. Defaults to None. acked (Optional[bool], optional): Whether to retrieve acknowledged alerts or not. Defaults to None. start_time (Optional[int], optional): Restrict alerts to those that have dates greater than the provided date. Defaults to None. end_time (Optional[int], optional): Restrict alerts to those that have dates less than or equal to the provided date. Defaults to None. insertion_start_time (Optional[int], optional): Restrict alerts to those that were inserted to the system after the provided date. Defaults to None. insertion_end_time (Optional[int], optional): Restrict alerts to those that were inserted to the system before the provided date. Defaults to None. Returns: Dict[str, Any]: API response from Netskope. """ params = assign_params( limit=limit, offset=page, starttime=start_time, endtime=end_time, insertionstarttime=insertion_start_time, insertionendtime=insertion_end_time, type=alert_type, query=query, acked=acked, ) return self._http_request("GET", "api/v2/events/data/alert", params=params) def list_event( self, page: int, limit: int, event_type: str, start_time: int | None = None, end_time: int | None = None, insertion_start_time: int | None = None, insertion_end_time: int | None = None, query: str | None = None, ) -> dict[str, Any]: """Get events from Netskope. Args: page (int): Page number of paginated results. limit (int): The maximum number of records to retrieve. event_type (str): Event type. start_time (Optional[int], optional): Restrict events to those that have dates greater than the provided date. Defaults to None. end_time (Optional[int], optional): Restrict events to those that have dates less than or equal to the provided date . Defaults to None. insertion_start_time (Optional[int], optional): Restrict events to those that were inserted to the system after the provided date. Defaults to None. insertion_end_time (Optional[int], optional): Restrict events to those that were inserted to the system before the provided date. Defaults to None. query (Optional[str], optional): Free query to filter the events. Defaults to None. Returns: Dict[str, Any]: API response from Netskope. """ params = assign_params( limit=limit, offset=page, starttime=start_time, endtime=end_time, insertionstarttime=insertion_start_time, insertionendtime=insertion_end_time, query=query, ) return self._http_request("GET", f"api/v2/events/data/{event_type}", params=params) def update_url_list( self, url_list_id: str, name: str | None = None, urls: list[str] | None = None, list_type: str | None = None, ) -> dict[str, Any]: """Update (override) the given URL list. Args: url_list_id (str): URL list ID. name (str): URL list name. urls (List[str]): URL lists. list_type (str): URL list type. Returns: Dict[str, Any]: API response from Netskope. """ data = { "data": {"type": list_type, "urls": urls}, "name": name, } return self._http_request("PUT", f"api/v2/policy/urllist/{url_list_id}", json_data=data) def patch_url_list( self, url_list_id: str, urls: list[str] | None = None, list_type: str | None = None, ) -> dict[str, Any]: """Update the given URL list. Args: url_list_id (str): URL list ID. urls (List[str]): URL lists. list_type (str): URL list type. Returns: Dict[str, Any]: API response from Netskope. """ data = {"data": {"type": list_type, "urls": urls}} return self._http_request("PATCH", f"api/v2/policy/urllist/{url_list_id}/append", json_data=data) def create_url_list( self, name: str, urls: list[str], list_type: str, ) -> dict[str, Any]: """Create URL list. Args: name (str): URL list name. urls (List[str]): URL lists. list_type (str): URL list type. Returns: Dict[str, Any]: API response from Netskope. """ data = { "data": {"type": list_type, "urls": urls}, "name": name, } return self._http_request("POST", "api/v2/policy/urllist", json_data=data) def list_url_list( self, url_list_id: str | None, pending: int | None = None, field: list[str] | None = None, ) -> dict[str, Any]: """Get all URL Lists or a specific URL list by the list ID. Args: url_list_id (Optional[str]): _description_ pending (Optional[int]): _description_ field (List[str]): _description_ Returns: Dict[str, Any]: API response from Netskope. """ params = assign_params( pending=pending, field=field, ) url_suffix = f"/{url_list_id}" if url_list_id else "" return self._http_request( "GET", f"api/v2/policy/urllist{url_suffix}", params=params, ) def delete_url_list(self, url_list_id: str) -> dict[str, Any]: """Delete a URL list by the list ID. Args: url_list_id (str): URL list ID. Returns: Dict[str, Any]: API response from Netskope. """ return self._http_request( "DELETE", f"api/v2/policy/urllist/{url_list_id}", ) def list_client( self, page: int, limit: int, client_filter: str | None = None, ) -> dict[str, Any]: """Get information about Netskope SCIM users. Args: page (int): Page number of paginated results. limit (int): The maximum number of records to retrieve. client_filter (str, optional): Filter the Netskope user by 'key eq value' template. Defaults to None. Returns: Dict[str, Any]: API response from Netskope. """ params = assign_params(count=limit, startIndex=page, filter=client_filter) return self._http_request( "GET", "api/v2/scim/Users", params=params, ) def deploy_url_list(self) -> dict[str, Any]: """Deploy URL list changes. Returns: Dict[str, Any]: API response from Netskope. """ return self._http_request( "POST", "api/v2/policy/urllist/deploy", ) def list_dlp_incident(self, timestamp: int) -> dict[str, Any]: """Fetch DLP incidents. Args: timestamp (int): The timestamp to filter by. Returns: dict[str, Any]: The API response with large integers converted to strings. """ response = self._http_request( "GET", url_suffix="api/v2/events/dataexport/events/incident", params={"operation": timestamp}, retries=10, status_list_to_retry=[409, 429], backoff_factor=DEFAULT_WAIT_TIME, ) # Convert large integers to strings to prevent JavaScript precision loss if incidents := response.get("result"): for incident in incidents: convert_large_integers_to_strings(incident) return response def update_dlp_incident( self, object_id: str, field: str, old_value: str, new_value: str, ) -> dict[str, Any]: """Update DLP incident. Args: object_id (str): The incident object ID. field (str): The field to update. old_value (str): The old value. new_value (str): The new value. Returns: dict[str, Any]: The API response. """ return self._http_request( "PATCH", url_suffix="api/v2/incidents/update", json_data=remove_empty_elements( { "payload": [ { "object_id": object_id, "field": field, "old_value": old_value, "new_value": new_value, "user": self.user_mail, } ] } ), ) def list_alert_command( client: Client, args: dict[str, Any], ) -> CommandResults: """Get alerts generated by Netskope. Args: client (Client): Netskope API client. args (Dict[str, Any]): command arguments. Returns: CommandResults: outputs, readable outputs and raw response for XSOAR. """ pagination_data = pagination(args) page = pagination_data.updated_page limit = pagination_data.limit pagination_message = pagination_data.pagination_message alert_type = args.get("alert_type") query = args.get("query") acked = optional_arg_to_boolean(args.get("retrieve_acknowledged")) time_args = convert_time_args_to_num(args) start_time = time_args.start_time_number end_time = time_args.end_time_number insertion_start_time = time_args.insertion_start_time_number insertion_end_time = time_args.insertion_end_time_number response = client.list_alert( page=page, limit=limit, start_time=start_time, end_time=end_time, insertion_start_time=insertion_start_time, insertion_end_time=insertion_end_time, alert_type=alert_type, query=query, acked=acked, ) output = response["result"] for alert in output: alert["alert_id"] = alert["_id"] alert["timestamp"] = timestamp_to_datestring(alert["timestamp"] * 1000) convert_large_integers_to_strings(alert) readable_output = tableToMarkdown( name="Alert List", metadata=pagination_message, t=remove_empty_elements(output), headers=ALERT_HEADERS, headerTransform=string_to_table_header, ) return CommandResults( readable_output=readable_output, outputs_prefix="Netskope.Alert", outputs_key_field="id", outputs=output, raw_response=response, ) def list_event_command( client: Client, args: dict[str, Any], ) -> CommandResults: """Get events extracted from SaaS traffic. Args: client (Client): Netskope API client. args (Dict[str, Any]): command arguments. Returns: CommandResults: outputs, readable outputs and raw response for XSOAR. """ pagination_data = pagination(args) page = pagination_data.updated_page limit = pagination_data.limit pagination_message = pagination_data.pagination_message event_type = args["event_type"] query = args.get("query") time_args = convert_time_args_to_num(args) start_time = time_args.start_time_number end_time = time_args.end_time_number insertion_start_time = time_args.insertion_start_time_number insertion_end_time = time_args.insertion_end_time_number response = client.list_event( page=page, limit=limit, start_time=start_time, end_time=end_time, insertion_start_time=insertion_start_time, insertion_end_time=insertion_end_time, event_type=event_type, query=query, ) output = response["result"] for event in output: event["event_id"] = event["_id"] event["timestamp"] = timestamp_to_datestring(event["timestamp"] * 1000) convert_large_integers_to_strings(event) readable_output = tableToMarkdown( name="Event List", metadata=pagination_message, t=remove_empty_elements(output), headers=EVENT_HEADERS[event_type], headerTransform=string_to_table_header, ) return CommandResults( readable_output=readable_output, outputs_prefix="Netskope.Event", outputs_key_field="id", outputs=output, raw_response=response, ) def update_url_list_command( client: Client, args: dict[str, Any], ) -> CommandResults: """Update (override) URL List. Args: client (Client): Netskope API client. args (Dict[str, Any]): command arguments. Returns: CommandResults: outputs, readable outputs and raw response for XSOAR. """ url_list_id = args["url_list_id"] name = args.get("name") urls = argToList(args.get("urls")) list_type = args.get("list_type", "").lower() or None is_overwrite = optional_arg_to_boolean(args.get("is_overwrite")) if not all([name, urls, list_type]) or not is_overwrite: list_response = client.list_url_list(url_list_id=url_list_id) name = name or list_response.get("name") list_type = list_type or dict_safe_get(list_response, ["data", "type"]) exist_urls = dict_safe_get(list_response, ["data", "urls"]) if not is_overwrite: urls += exist_urls else: urls = urls or exist_urls response = client.update_url_list( url_list_id, name, urls, list_type, ) deploy_url_list_if_required(args, client.deploy_url_list) output = get_updated_url_list(response) readable_output = tableToMarkdown( name="URL list was updated successfully", t=remove_empty_elements(output), headers=URL_HEADER, headerTransform=string_to_table_header, ) return CommandResults( readable_output=readable_output, outputs_prefix="Netskope.URLList", outputs_key_field="id", outputs=output, raw_response=response, ) def add_url_list_command( client: Client, args: dict[str, Any], ) -> CommandResults: """Update URL List. Args: client (Client): Netskope API client. args (Dict[str, Any]): command arguments. Returns: CommandResults: outputs, readable outputs and raw response for XSOAR. """ url_list_id = args["url_list_id"] urls = argToList(args.get("urls")) list_type = args.get("list_type", "").lower() or None response = client.patch_url_list(url_list_id, urls, list_type) deploy_url_list_if_required(args, client.deploy_url_list) output = get_updated_url_list(response) readable_output = tableToMarkdown( name="URL list was updated successfully", t=remove_empty_elements(output), headers=URL_HEADER, headerTransform=string_to_table_header, ) return CommandResults( readable_output=readable_output, outputs_prefix="Netskope.URLList", outputs_key_field="id", outputs=output, raw_response=response, ) def create_url_list_command( client: Client, args: dict[str, Any], ) -> CommandResults: """Create a new URL list. Args: client (Client): Netskope API client. args (Dict[str, Any]): command arguments. Returns: CommandResults: outputs, readable outputs and raw response for XSOAR. """ name = args["name"] urls = argToList(args["urls"]) list_type = args.get("list_type", "").lower() response = client.create_url_list( name, urls, list_type, ) deploy_url_list_if_required(args, client.deploy_url_list) output = get_updated_url_list(response) readable_output = tableToMarkdown( name="URL list was created successfully", t=remove_empty_elements(output), headers=URL_HEADER, headerTransform=string_to_table_header, ) return CommandResults( readable_output=readable_output, outputs_prefix="Netskope.URLList", outputs_key_field="id", outputs=output, raw_response=response, ) def lists_url_list_command(client: Client, args: dict[str, Any]) -> CommandResults: """Get all URL Lists or a specific URL list. Args: client (Client): Netskope API client. args (Dict[str, Any]): command arguments. Returns: CommandResults: outputs, readable outputs and raw response for XSOAR. """ url_list_id = args.get("url_list_id") pending = args.get("pending") fields = argToList(args.get("field")) all_results = optional_arg_to_boolean(args.get("all_results")) pending = int(pending == "pending") updated_fields = [] updated_fields = ["data" if field in ("urls", "type") else field for field in fields] response = client.list_url_list(url_list_id=url_list_id, pending=pending, field=updated_fields) output = get_updated_url_list(response) if not all_results: limit = arg_to_number(args["limit"]) output = output[:limit] readable_output = tableToMarkdown( name="URL Lists", t=remove_empty_elements(output), headers=fields or URL_HEADER, headerTransform=string_to_table_header, ) return CommandResults( readable_output=readable_output, outputs_prefix="Netskope.URLList", outputs_key_field="id", outputs=response, raw_response=response, ) def delete_url_list_command( client: Client, args: dict[str, Any], ) -> CommandResults: """Delete a URL list by the list ID. Args: client (Client): Netskope API client. args (Dict[str, Any]): command arguments. Returns: CommandResults: outputs, readable outputs and raw response for XSOAR. """ url_list_id = args["url_list_id"] response = client.delete_url_list(url_list_id) deploy_url_list_if_required(args, client.deploy_url_list) return CommandResults( readable_output=f"The URL list {url_list_id} was deleted successfully", outputs_prefix="Netskope.URLList", outputs_key_field="id", outputs=response, raw_response=response, ) def list_client_command( client: Client, args: dict[str, Any], ) -> CommandResults: """Get information about Netskope SCIM users. Args: client (Client): Netskope API client. args (Dict[str, Any]): command arguments. Returns: CommandResults: outputs, readable outputs and raw response for XSOAR. """ pagination_data = pagination(args, False) page = pagination_data.updated_page limit = pagination_data.limit pagination_message = pagination_data.pagination_message client_filter = args.get("filter") updated_client_filter = convert_client_filter(client_filter) if client_filter else None response = client.list_client(page, limit, updated_client_filter) output = response.get("Resources", []) output = get_updated_list_client(output) readable_output = tableToMarkdown( name="Client List", t=remove_empty_elements(output), metadata=pagination_message, headers=[ "client_id", "external_id", "user_name", "given_name", "family_name", "emails", "active", ], headerTransform=string_to_table_header, ) return CommandResults( readable_output=readable_output, outputs_prefix="Netskope.Client", outputs_key_field="id", outputs=output, raw_response=response, ) def list_dlp_incident_command( client: Client, args: dict[str, Any], ) -> CommandResults: """Fetch DLP incidents. Args: client (Client): A session to run HTTP requests to the API. args (dict[str, Any]): Arguments passed down by the CLI to configure the request. Returns: CommandResults: Outputs of the command that represent an entry in the warroom. """ start_time_dt = arg_to_datetime(args["start_time"]) end_time_dt = arg_to_datetime(args["end_time"]) if not start_time_dt or not end_time_dt: raise DemistoException("Time argument is required.") start_time = int(start_time_dt.timestamp()) end_time = int(end_time_dt.timestamp()) incidents = [] hourly_timestamps = get_hourly_timestamps( start_time=start_time, end_time=end_time, ) for timestamp in hourly_timestamps: response = client.list_dlp_incident(timestamp=timestamp) if dlp_incidents := response["result"]: incidents += dlp_incidents incidents = remove_duplicates(incidents, "object_id") readable_output = tableToMarkdown( name="DLP Incident List:", t=incidents, headers=INCIDENT_HEADERS, headerTransform=string_to_table_header, ) return CommandResults( raw_response=incidents, outputs=incidents, outputs_key_field="object_id", outputs_prefix="Netskope.Incident", readable_output=readable_output, ) def test_module(client: Client) -> str: try: client.list_client(1, 1) return "ok" except DemistoException as exc: if exc.res is not None and exc.res.status_code in [ HTTPStatus.UNAUTHORIZED, HTTPStatus.NOT_FOUND, HTTPStatus.FORBIDDEN, ]: return "Authorization Error: Unknown API key or Netskope URL" return exc.message def fetch_incidents(client: Client, params: dict[str, Any]): """Fetch Netskope alerts, DLP incidents, and events (if requested) as incidents in XSOAR. In case of DLP incidents: the incidents will fetch by timestamp and will saved in XSOAR and integration context to handle duplicates issue. Args: client (Client): Netskope API client. params (Dict[str, Any]): Integration parameters. """ incidents: list[dict] = [] last_run: dict[str, Any] = {} mirror_direction = MIRROR_DIRECTION_MAPPING.get(params["mirror_direction"], None) event_types = argToList(params.get("event_types")) alert_query = params.get("alerts_query") alert_max_fetch = arg_to_number(params.get("max_fetch") or 50) or MAX_LIMIT num_event_types = len(event_types) if event_types else 1 max_fetch_per_event_type = MAX_FETCH_PER_EVENT_TYPE // num_event_types if event_types and (event_max_fetch := arg_to_number(params.get("max_events_fetch") or 50)) is not None: max_fetch_per_event_type = event_max_fetch // (len(event_types)) if params.get("fetch_events") and event_types: events, events_last_run = fetch_events_as_incidents( list_event_func=client.list_event, params=params, max_fetch_per_event_type=max_fetch_per_event_type, last_run=last_run, mirror_direction=mirror_direction, ) incidents += events last_run.update(events_last_run) if params.get("fetch_dlp_incidents"): dlp_incidents, dlp_incidents_last_run = fetch_dlp_incidents_as_incidents( list_event_func=client.list_dlp_incident, params=params, last_run=last_run, mirror_direction=mirror_direction, ) incidents += dlp_incidents last_run.update(dlp_incidents_last_run) alerts, alerts_last_run = fetch_alerts_as_incidents( list_event_func=client.list_alert, params=params, max_fetch=alert_max_fetch, alert_query=alert_query, last_run=last_run, mirror_direction=mirror_direction, ) incidents += alerts last_run.update(alerts_last_run) return last_run, incidents def get_modified_remote_data(client: Client) -> GetModifiedRemoteDataResponse: """ Queries for incidents that were modified since the last update. Args: client: Netskope API client. Returns: GetModifiedRemoteDataResponse: modified tickets from Netskope. """ # get the fetch last run time to check if exist incidents was updated for sync incidents start_timestamp = get_demisto_integration_context("dlp_incident_last_run_timestamp", None) end_timestamp = date_to_seconds_timestamp(datetime.now()) if not start_timestamp: start_timestamp = end_timestamp - 600 hourly_timestamps = get_hourly_timestamps( start_time=start_timestamp, end_time=end_timestamp, ) demisto.debug(f"Get modified remote data from {start_timestamp} - {end_timestamp}") dlp_incidents = [] modified_tickets = [] for timestamp in hourly_timestamps: response = client.list_dlp_incident(timestamp=timestamp) if data := response["result"]: dlp_incidents += data dlp_incidents = remove_duplicates(dlp_incidents, "object_id") # save the updated incidents for get_remote_data_command set_demisto_integration_context("dlp_incidents_to_update", dlp_incidents, "append") for ticket in dlp_incidents: modified_tickets.append(ticket["object_id"]) demisto.debug(f"There are {len(modified_tickets)} modified incidents from Netskope") return GetModifiedRemoteDataResponse(modified_tickets) def get_mapping_fields_command() -> GetMappingFieldsResponse: """ Pulls the remote schema for the different incident types, and their associated incident fields, from the remote system. Returns: GetMappingFieldsResponse: Dictionary with keys as field names. """ demisto.debug("Get Netskope mapping fields") mapping_response = GetMappingFieldsResponse() incident_type_scheme = SchemeTypeMapping(type_name="Netskope Incident") for field in MIRRORING_FIELDS: incident_type_scheme.add_field(field) mapping_response.add_scheme_type(incident_type_scheme) return mapping_response def update_remote_system( client: Client, args: dict[str, Any], params: dict[str, Any], ) -> str: """ This command pushes local changes to the remote system. Args: client: XSOAR Client to use. args: args['data']: the data to send to the remote system. args['entries']: the entries to send to the remote system. args['incident_changed']: boolean telling us if the local incident indeed changed or not. args['remote_incident_id']: the remote incident id. Returns: The remote incident id - ticket_id """ parsed_args = UpdateRemoteSystemArgs(args) incident_id = parsed_args.remote_incident_id demisto.debug( f"Got the following delta keys {list(parsed_args.delta.keys())!s}" if parsed_args.delta else "There is no delta fields in Netskope" ) try: if parsed_args.incident_changed and parsed_args.delta: demisto.debug(f"Incident changed: {parsed_args.incident_changed}, {parsed_args.delta=}, {parsed_args.inc_status=}") update_args: dict[str, Any] = parsed_args.delta # fetch old arguments data for API call old_args: dict[str, Any] = parsed_args.data updated_arguments = {} if parsed_args.inc_status == IncidentStatus.DONE and params.get("close_netskope_incident"): updated_arguments["field"] = "status" updated_arguments["old_value"] = old_args["original_status"] updated_arguments["new_value"] = "closed" updated_arguments["object_id"] = incident_id else: for key, value in update_args.items(): if key in MIRRORING_FIELDS: updated_arguments["field"] = key updated_arguments["old_value"] = old_args[f"original_{key}"] updated_arguments["new_value"] = value updated_arguments["object_id"] = incident_id if updated_arguments: demisto.debug(f"Send remote ID [{incident_id}] updates to Netskope. {updated_arguments=}|| {update_args=}") client.update_dlp_incident(**updated_arguments) except Exception as error: demisto.info(f"Error in Netskope outgoing mirror for incident {incident_id}. Error message: {error}") finally: return incident_id def get_remote_data_command( args: dict[str, Any], params: dict[str, Any], ) -> GetRemoteDataResponse: """ Gets new information about the incidents in the remote system and updates existing incidents in Cortex XSOAR. Args: args (Dict[str, Any]): command arguments. params (Dict[str, Any]): command parameters. Returns: List[Dict[str, Any]]: first entry is the incident (which can be completely empty) and the new entries. """ parsed_args = GetRemoteDataArgs(args) incident_id = parsed_args.remote_incident_id demisto.debug(f"Check if incident {incident_id} updated") # fetch updated incidents (that saved in get_modified_remote_data command) dlp_incidents: list = get_demisto_integration_context("dlp_incidents_to_update", []) mirrored_ticket = {} for incident in dlp_incidents: # get the incident data by the relevant incident ID if incident["object_id"] == incident_id: mirrored_ticket = incident # remove the incident from integration context dlp_incidents.remove(mirrored_ticket) set_demisto_integration_context("dlp_incidents_to_update", dlp_incidents, "override") break entries = [] mirrored_ticket["incident_type"] = "dlp_incident" if mirrored_ticket.get("status") == "closed" and params.get("close_incident"): entries.append( { "Type": EntryType.NOTE, "Contents": { "dbotIncidentClose": True, "closeReason": "Closed from Netskope.", }, "ContentsFormat": EntryFormat.JSON, } ) return GetRemoteDataResponse(mirrored_ticket, entries) # HELPERS FUNCTIONS # def convert_large_integers_to_strings(data: dict[str, Any]) -> None: """Convert large integers to strings to prevent JavaScript precision loss. JavaScript cannot safely represent integers larger than 2^53-1 (Number.MAX_SAFE_INTEGER). This function converts such integers to strings in-place to preserve their exact values. Args: data (dict[str, Any]): Dictionary containing potential large integer values. Reference: https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Number/MAX_SAFE_INTEGER """ for key, value in list(data.items()): if isinstance(value, int) and value > JS_NUMBER_LIMIT: demisto.debug(f"Converting large integer to string: {key}={value}") data[key] = str(value) def get_hourly_timestamps(start_time: int, end_time: int) -> list[int]: """Get a list of timestamps with a one hour gap between the received start and end timestamps. Args: start_time (int): The start timestamp. end_time (int): The end timestamp. Returns: list[int]: The timestamps list. """ timestamps = [] current_timestamp = start_time while current_timestamp <= end_time: timestamps.append(current_timestamp) current_timestamp += 3600 # Increment by 3600 seconds (1 hour) return timestamps def remove_duplicates( dicts_list: list[dict[str, Any]], key: str, ) -> list[dict[str, Any]]: """Remove duplicates from a list of dicts by unique key. Args: dicts_list (list[dict[str, Any]]): The list of dicts to update. key (str): The unique key. Returns: list[dict[str, Any]]: The updated list with no duplicates. """ object_map = {} for item in dicts_list: object_id = item[key] object_map[object_id] = item return list(object_map.values()) def parse_incident( incident: dict, incident_type: str, mirror_direction: str | None = None, ) -> dict: """ Parse alert/event to XSOAR Incident. Args: incident (dict): alert/event item. incident_type (str): Incident type. Returns: dict: XSOAR Incident. """ incident_id = incident["object_id"] if incident_type == "dlp_incident" else incident["_id"] incident["incident_type"] = incident_type incident["mirror_direction"] = mirror_direction incident["mirror_instance"] = demisto.integrationInstance() convert_large_integers_to_strings(incident) return { "name": f"{incident_type} ID: {incident_id}", "incident_type": incident_type, "mirror_direction": mirror_direction, "mirror_instance": demisto.integrationInstance(), "occurred": convert_datetime_int_to_iso(incident["timestamp"]), "rawJSON": json.dumps(incident), } def convert_datetime_int_to_iso(creation_timestamp: int) -> datetime: """Convert datetime to iso. Args: creation_timestamp (int): creation timestamp argument. Returns: datetime: Updated argument. """ date_string = convert_number_to_str_date(creation_timestamp) alert_date = datetime.strptime(date_string, TIME_FORMAT) return FormatIso8601(alert_date) + "Z" def fetch_events_as_incidents( list_event_func: Callable, params: dict[str, Any], max_fetch_per_event_type: int, last_run: dict[str, Any], mirror_direction: str | None = None, ) -> tuple[list, dict]: """Fetch events from Netskope as incidents. Args: list_event_func (Callable): Netskope API client. params (Dict[str, Any]): Integration parameters. max_fetch_per_event_type (int): Maximum number per event type. last_run (Dict[str, Any]): Last run argument. mirror_direction (str | None): Mirror direction. Returns:Callable Tuple[list, dict]: Events as incidents and updated last_run argument. """ incidents = [] event_types = argToList(params.get("event_types")) event_query = params.get("event_query") for event_type in event_types: relevant_events = [] last_run_id, last_run_timestamp = get_last_run(params, event_type) end_time_number = date_to_seconds_timestamp(datetime.now()) response = list_event_func( event_type=event_type, query=event_query, limit=max_fetch_per_event_type, page=0, start_time=last_run_timestamp, end_time=end_time_number, ) events = response["result"] for event in events: if last_run_id != event["_id"]: event["incident_type"] = event_type incidents.append(parse_incident(event, f"{event_type}-event", mirror_direction)) relevant_events.append(event) last_run = update_last_run( relevant_events=relevant_events, incident_type=event_type, last_run=last_run, last_run_id=last_run_id, last_run_time_number=last_run_timestamp, ) return incidents, last_run def fetch_dlp_incidents_as_incidents( list_event_func: Callable, params: dict[str, Any], last_run: dict[str, Any], mirror_direction: str | None = None, ) -> tuple[list, dict]: """Fetch DLP incidents from Netskope as incidents. The incidents will be saved in the integration context to avoid duplicates. Args: list_event_func (Callable): Netskope API client. params (Dict[str, Any]): Integration parameters. last_run (dict[str, Any]): Last run. mirror_direction (str | None): Mirror direction. Returns:Callable Tuple[list, dict]: DLP incidents as incidents and updated last_run argument. """ incidents = [] new_incidents = [] new_incident_ids = [] last_run_id, last_run_timestamp = get_last_run(params, "dlp_incident") # when last run is rest - update integration context data if not last_run_id: set_demisto_integration_context("dlp_incident_ids", [], "override") set_demisto_integration_context("dlp_incident_last_run_timestamp", None, "override") set_demisto_integration_context("dlp_incidents_to_update", [], "override") end_time_number = date_to_seconds_timestamp(datetime.now()) max_fetch = int(params.get("max_dlp_incidents_fetch") or 50) hourly_timestamps = get_hourly_timestamps( start_time=last_run_timestamp, end_time=end_time_number, ) dlp_incidents = [] # get incident IDs that exist in XSOAR to avoid duplicates exist_dlp_incidents: list = get_demisto_integration_context("dlp_incident_ids", []) for timestamp in hourly_timestamps: response = list_event_func(timestamp) if response.get("result"): dlp_incidents += response["result"] dlp_incidents = remove_duplicates(dlp_incidents, "object_id") last_run_timestamp = timestamp if len(dlp_incidents) >= max_fetch: dlp_incidents = dlp_incidents[:max_fetch] break for incident in dlp_incidents: incident_id = incident["object_id"] if incident_id not in exist_dlp_incidents: incidents.append(parse_incident(incident, "dlp_incident", mirror_direction)) new_incidents.append(incident) new_incident_ids.append(incident_id) if new_incident_ids: # set the new incident IDs to avoid duplicates on next fetch set_demisto_integration_context("dlp_incident_ids", new_incident_ids, "append") last_run_timestamp = max(new_incidents, key=lambda k: k["timestamp"])["timestamp"] + 1 else: last_run_timestamp = end_time_number # set the updated fetch last run time for get_modified_remote_data command # to update existing incidents and ensure full sync set_demisto_integration_context("dlp_incident_last_run_timestamp", last_run_timestamp, "override") last_run = update_last_run( relevant_events=new_incidents, incident_type="dlp_incident", last_run=last_run, last_run_id=last_run_id, last_run_time_number=last_run_timestamp, ) return incidents, last_run def fetch_alerts_as_incidents( list_event_func: Callable, params: dict[str, Any], max_fetch: int, last_run: dict[str, Any], alert_query: str | None = None, mirror_direction: str | None = None, ) -> tuple[list, dict]: """Fetch DLP incidents from Netskope as incidents. Args: list_event_func (Callable): Netskope API client. params (Dict[str, Any]): Integration parameters. max_fetch (int): Maximum number per event type. last_run (dict[str, Any]): Last run data. alert_query (str, optional): Alert query. Defaults to None. mirror_direction (str, optional): Mirror direction. Defaults to None. Returns:Callable Tuple[list, dict]: DLP incidents as incidents and updated last_run argument. """ incidents = [] new_incidents = [] last_run_id, last_run_timestamp = get_last_run(params, "alert") end_time_number = date_to_seconds_timestamp(datetime.now()) response = list_event_func( query=alert_query, limit=max_fetch, page=0, start_time=last_run_timestamp, end_time=end_time_number, ) alerts = response["result"] for alert in alerts: if last_run_id != alert["_id"]: alert["incident_type"] = alert["alert_type"] incidents.append(parse_incident(alert, "alert", mirror_direction)) new_incidents.append(alert) last_run = update_last_run( relevant_events=new_incidents, incident_type="alert", last_run=last_run, last_run_id=last_run_id, last_run_time_number=last_run_timestamp, ) return incidents, last_run def update_last_run( relevant_events: list, incident_type: str, last_run: dict, last_run_id: Any | None, last_run_time_number: int, ) -> dict[str, Any]: """Update the last run argument by alert type for next run. Args: relevant_events (list): The new incidents. incident_type (str): Incident type. last_run (dict): Last run argument. last_run_id (int): Last run ID. last_run_time_number (int): The last run time number. Returns: Dict[str, Any]: Updated last run argument. """ if relevant_events: last_run_alert = max(relevant_events, key=lambda k: k["timestamp"]) last_run[incident_type] = { "id": (last_run_alert["object_id"] if incident_type == "dlp_incident" else last_run_alert["_id"]), "time": last_run_time_number if incident_type == "dlp_incident" else last_run_alert["timestamp"] + 1, "date": convert_number_to_str_date(last_run_time_number), } else: last_run[incident_type] = { "id": last_run_id, "time": last_run_time_number, "date": convert_number_to_str_date(last_run_time_number), } demisto.setLastRun(last_run) return last_run def get_last_run( args: dict[str, Any], incident_type: str, ) -> tuple[Any | None, int]: """Get last run arguments. Args: args (Dict[str, Any]): XSOAR arguments. incident_type (str): Incident type. Returns: Tuple: Updated last run arguments. """ last_run = demisto.getLastRun() ticket_last_run = last_run.get(incident_type) last_run_id = None if last_run and ticket_last_run: last_run_time = ticket_last_run.get("time") last_run_id = ticket_last_run.get("id") else: last_run_time = args.get("first_fetch", "3 Days") first_fetch = arg_to_datetime(arg=last_run_time, arg_name="First fetch time", required=True) if not first_fetch: raise ValueError("First fetch time not specified") last_run_timestamp = date_to_seconds_timestamp(first_fetch) return last_run_id, last_run_timestamp def deploy_url_list_if_required( args: dict[str, Any], deploy_url_list_func: Callable, ): """Deploys URL list changes if required. Args: client (Client): Netskope API client. args (Dict[str, Any]): command arguments. """ if optional_arg_to_boolean(args.get("deploy")): deploy_url_list_func() def convert_client_filter(client_filter: str) -> str: """Convert client filter to the appropriate format. Args: client_filter (str): client filter. Returns: str: Updated filter. """ filter_parts = [part.strip() for part in client_filter.split("eq")] if len(filter_parts) != 2: raise DemistoException("Filter must contain 'key' eq 'value'") attribute_name, attribute_value = filter_parts attribute_value = attribute_value.strip().strip('"') modified_filter = f'{attribute_name} eq "{attribute_value}"' return modified_filter def get_updated_list_client(response: list | dict) -> list: """Updates URL list response. Args: response (Union[List, Dict]): The API response. Returns: List: Updated URL list. """ outputs = [] for list_client in response: output = copy.deepcopy(list_client) output["client_id"] = output.pop("id") output |= output.pop("name", {}) output["emails"] = [email.get("value") for email in output.get("emails", [])] output = snakify(output) outputs.append(output) return outputs def get_updated_url_list(response: list | dict) -> list: """Updates URL list response. Args: response (Union[List, Dict]): The API response. Returns: List: Updated URL list. """ response = [response] if isinstance(response, dict) else response outputs = [] number_to_pending_status = { 0: "applied", 1: "pending", } for url_list in response: output = copy.deepcopy(url_list) output |= output.pop("data", {}) output["pending"] = number_to_pending_status.get(output.get("pending")) outputs.append(output) return outputs def parse_start_end_times(start_str: str, end_str: str | None) -> tuple[int | None, int | None]: """Convert string date to timestamp. Args: start_str (str): String start date. end_str (Optional[str]): String end date. Returns: Tuple: Timestamp start & end dates. """ start_number = arg_to_seconds_timestamp(start_str) if start_str else None end_number = arg_to_seconds_timestamp(end_str) if end_str else date_to_seconds_timestamp(datetime.now()) return start_number, end_number def convert_time_args_to_num(args: dict[str, Any]) -> TimeArgs: """Convert relevant time arguments to numbers timestamp. Args: args (Dict[str, Any]): command arguments. Raises: ValueError: When specified all time arguments or not specified none of them. Returns: Tuple: The updated time arguments. """ start_time_str = args.get("start_time") end_time_str = args.get("end_time") insertion_start_time_str = args.get("insertion_start_time") insertion_end_time_str = args.get("insertion_end_time") start_time_number = None end_time_number = None insertion_start_time_number = None insertion_end_time_number = None if start_time_str: start_time_number, end_time_number = parse_start_end_times(start_time_str, end_time_str) elif insertion_start_time_str: insertion_start_time_number, insertion_end_time_number = parse_start_end_times( insertion_start_time_str, insertion_end_time_str ) else: raise ValueError("Provide either the `start_time` argument or `insertion_start_time`.") return TimeArgs( start_time_number, end_time_number, insertion_start_time_number, insertion_end_time_number, ) def convert_number_to_str_date(number: int) -> str: """Convert number to date string. Args: number (int): The date number to convert. Returns: str: The string date. """ return datetime.fromtimestamp(float(number)).strftime(TIME_FORMAT) def arg_to_seconds_timestamp(arg: str | datetime) -> int | 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. """ date_arg = arg_to_datetime(arg) if date_arg is None: return None return date_to_seconds_timestamp(date_arg) def date_to_seconds_timestamp(date_str_or_dt: 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_pagination_arguments( page: int | None = None, limit: int | None = None, ): """Validate pagination arguments according to their default. Args: page (int, optional): Page number of paginated results. page_size (int, optional): Number of items per page. limit (int, optional): The maximum number of records to retrieve. Raises: ValueError: Incase page is lower than 1 or Incase limit is lower than 1. """ if page is not None and page < MIN_PAGE_NUM: raise ValueError(f"page argument must be greater than {MIN_PAGE_NUM}.") if limit is not None and limit < MIN_LIMIT: raise ValueError(f"limit argument must be greater than {MIN_LIMIT}.") def pagination(args: dict[str, Any], start_count_from_zero: bool = True) -> Pagination: """Return the correct limit and offset for the API based on the user arguments page and limit. Args: args (Dict[str, Any]): Command arguments from XSOAR. start_count_from_zero (bool, optional): Whether to start the offset count from or 1. Defaults to True. Returns: Tuple: page, limit, pagination_message. """ page = arg_to_number(args.get("page")) or 1 limit = arg_to_number(args.get("limit")) or 50 validate_pagination_arguments(page, limit) pagination_message = f"Showing page {page}. \n Current page size: {limit}." updated_page = (page - 1) * limit + (0 if start_count_from_zero else 1) return Pagination(updated_page, limit, pagination_message) def optional_arg_to_boolean(arg: str | bool | None) -> bool | None: """Retrieve arg boolean value if it's not none. Args: arg (Union[str, bool, None]): Boolean argument. Returns: Optional[bool]: The argument boolean value. """ return argToBoolean(arg) if arg is not None else None def get_demisto_integration_context( key: str, default_value: Any, ) -> Any: """Get the integration context for a given key. Args: key (str): The key to retrieve. default_value (Any): The default value to return in case the key dont exist. Returns: Any: The value of the given key. """ return get_integration_context().get(key, default_value) def set_demisto_integration_context( key: str, value_to_update: Any, action: str, ): """Set the integration context for the given key and value. Args: key (str): The integration context key to update his value. value_to_update (Any): The value to update. action (str): The action to take when update the value. """ integration_context = get_integration_context() if action == "append": if not integration_context.get(key): integration_context[key] = [] integration_context[key] += value_to_update # check if integration context size is reached to 40mb if len(integration_context[key]) >= MAX_IDS_NUMBER: # remove the first items to avoid overflow number_of_items_to_remove = len(integration_context[key]) - MAX_IDS_NUMBER integration_context[key] = integration_context[key][number_of_items_to_remove:] else: integration_context[key] = value_to_update set_integration_context(integration_context) def main() -> None: # pragma: no cover params: dict[str, Any] = demisto.params() args: dict[str, Any] = demisto.args() url = params["url"] api_token = params["credentials"]["password"] proxy = params.get("proxy", False) verify_certificate: bool = not params.get("insecure", False) user_email = params.get("user_email") command = demisto.command() demisto.debug(f"Command being called is {command}") try: client: Client = Client( url, api_token, verify_certificate, proxy, user_email, ) commands = { "netskope-alert-list": list_alert_command, "netskope-event-list": list_event_command, "netskope-url-list-update": update_url_list_command, "netskope-url-list-create": create_url_list_command, "netskope-url-lists-list": lists_url_list_command, "netskope-url-list-delete": delete_url_list_command, "netskope-client-list": list_client_command, "netskope-url-list-add": add_url_list_command, "netskope-incident-dlp-list": list_dlp_incident_command, } if command == "test-module": return_results(test_module(client)) elif command == "get-remote-data": return_results(get_remote_data_command(args, params)) elif command == "get-modified-remote-data": return_results(get_modified_remote_data(client)) elif command == "update-remote-system": return_results(update_remote_system(client, args, params)) elif command == "get-mapping-fields": return_results(get_mapping_fields_command()) elif command == "fetch-incidents": last_run, incidents = fetch_incidents(client, params) demisto.setLastRun(last_run) demisto.incidents(incidents) elif command in commands: return_results(commands[command](client, args)) else: raise NotImplementedError(f"{command} command is not implemented.") except Exception as e: return_error(str(e)) if __name__ in ["__main__", "builtin", "builtins"]: main()