import concurrent.futures import platform from typing import Any import urllib3 import demistomock as demisto from CommonServerPython import * # noqa # pylint: disable=unused-wildcard-import from CommonServerUserPython import * # noqa # Disable insecure warnings urllib3.disable_warnings() """ CONSTANTS """ DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ" # ISO8601 format with UTC, default in XSOAR MAX_IMAGES_TO_FETCH = 25 STATUS_TO_RETRY = [500, 501, 502, 503, 504] __version__ = "0.1.0" TIMEOUT_60 = 60 TIMEOUT_90 = 90 TIMEOUT_120 = 120 """ CLIENT CLASS """ class Client(BaseClient): """Client class to interact with the service API""" def _request_raw( self, *, method: str, url_suffix: str, params: dict | None = None, json_data: dict | None = None, timeout: int = 90, retries: int = 3, ) -> dict: response: Any = self._http_request( method=method, url_suffix=url_suffix, params=params, json_data=json_data, timeout=timeout, retries=retries, status_list_to_retry=STATUS_TO_RETRY, ) if not isinstance(response, dict): raise DemistoException(f"Bad Response, response was not a dict: {str(response)}") if response.get("return_error"): return_error(**response["return_error"]) raise DemistoException("return_error returned unexpectedly") return response @staticmethod def _no_results_found() -> list[CommandResults]: return [ CommandResults( outputs_prefix="", outputs={}, raw_response={}, readable_output="No results found.", outputs_key_field="", ) ] @classmethod def _response_to_command_results(cls, response: dict) -> list[CommandResults]: result_actions = response.get("result_actions") if not isinstance(result_actions, list): raise DemistoException(f"Bad Response, result_actions was present but not a list: {str(response)}") command_results: list[CommandResults] = [] for result_action in result_actions: if not isinstance(result_action, dict): continue raw_response = result_action.get("raw_response") outputs = raw_response.get("outputs") if isinstance(raw_response, dict) else None command_results.append( CommandResults( outputs=outputs, **result_action, ) ) return command_results def _request_results( self, *, method: str, url_suffix: str, params: dict | None = None, json_data: dict | None = None, timeout: int = 90, retries: int = 3, ) -> list[CommandResults]: try: response = self._request_raw( method=method, url_suffix=url_suffix, params=params, json_data=json_data, timeout=timeout, retries=retries, ) except DemistoException as err: if "404" in str(err): return self._no_results_found() raise return self._response_to_command_results(response) def whoami(self) -> dict: return self._request_raw( method="GET", url_suffix="/info/whoami", timeout=60, ) def alert_update(self) -> list[CommandResults]: """Update alert""" return self._request_results( method="POST", url_suffix="/v3/alert/update", json_data=demisto.args(), timeout=90, ) def alert_search(self) -> list[CommandResults]: """Search alerts""" return self._request_results( method="GET", url_suffix="/v3/alert/search", params=demisto.args(), ) def alert_rule_search(self) -> list[CommandResults]: """Search alert rules.""" return self._request_results( method="GET", url_suffix="/v3/alert/rules", params=demisto.args(), ) def alert_lookup(self, alert_id: str) -> list[CommandResults]: return self._request_results( method="GET", url_suffix="/v3/alert/lookup", params={"alert_id": alert_id}, timeout=90, ) def get_alert_image( self, alert_type: str, alert_id: str, image_id: str, alert_subtype: str | None, ) -> bytes: """ Get an image from the v3 alert image endpoint. Returns the raw binary content of the image. """ response_content: Any = self._http_request( method="get", url_suffix="/v3/alert/image", params={ "alert_type": alert_type, "alert_subtype": alert_subtype, "alert_id": alert_id, "image_id": image_id, }, timeout=90, resp_type="content", ) return response_content def fetch_incidents(self) -> dict: """Fetch incidents.""" classic_query_params = demisto.getLastRun().get("next_query_classic", {}) playbook_query_params = demisto.getLastRun().get("next_query_playbook", {}) return self._request_raw( method="POST", url_suffix="/v3/alert/fetch", json_data={ "integration_config": demisto.params(), "classic_query_params": classic_query_params, "playbook_query_params": playbook_query_params, }, timeout=120, ) # === === === === === === === === === === === === === === === # === === === === === === ACTIONS === === === === === === === # === === === === === === === === === === === === === === === class Actions: def __init__(self, rf_client: Client): self.client = rf_client def test_module(self) -> None: # This is the call made when pressing the integration Test button. # Returning 'ok' indicates that the integration works like it suppose to and # connection to the service is successful. # Returning 'ok' will make the test result be green. # Any other response will make the test result be red. demisto_params = demisto.params() # Validate first_fetch first_fetch_str = str(demisto_params.get("first_fetch", "")) if first_fetch_str.isnumeric(): first_fetch = int(first_fetch_str) else: raise ValueError("'first_fetch' parameter must be a number") ninety_days_in_minutes = 90 * 24 * 60 if first_fetch > ninety_days_in_minutes: raise ValueError("'first_fetch' parameter cannot be bigger than 90 days") # Validate max_fetch max_fetch_str = str(demisto_params.get("max_fetch", "")) if max_fetch_str.isnumeric(): max_fetch = int(max_fetch_str) else: raise ValueError("'max_fetch' parameter must be a number") if max_fetch > 50: raise ValueError("'max_fetch' parameter cannot be bigger than 50") try: self.client.whoami() return_results("ok") except Exception as err: message = str(err) try: error = json.loads(str(err).split("\n")[1]) if "fail" in error.get("result", {}).get("status", ""): message = error.get("result", {})["message"] except Exception: message = f"Unknown error. Please verify that the API URL and Token are correctly configured. RAW Error: {err}" raise DemistoException(f"Failed due to - {message}") def fetch_incidents(self) -> None: try: response = self.client.fetch_incidents() except DemistoException as err: if "404" in str(err): return_error("404 in fetch incidents") return raise alerts = response.get("alerts", []) next_query_classic = response.get("next_query_classic", {}) next_query_playbook = response.get("next_query_playbook", {}) next_query = { "next_query_classic": next_query_classic, "next_query_playbook": next_query_playbook, } incidents = [ { "name": alert.get("title"), "occurred": alert.get("created"), "dbotMirrorId": alert.get("id"), "rawJSON": json.dumps(alert), } for alert in alerts ] demisto.incidents(incidents) demisto.setLastRun(next_query) def alert_search_command(self) -> list[CommandResults]: return self.client.alert_search() def alert_rule_search_command( self, ) -> list[CommandResults]: return self.client.alert_rule_search() def alert_update_command(self) -> list[CommandResults]: return self.client.alert_update() def alert_lookup_command(self) -> list[CommandResults]: alert_id = demisto.args().get("alert_id", "") return self.client.alert_lookup(alert_id) @staticmethod def _get_file_name_from_image_id(image_id: str) -> str: return f"{image_id.replace('img:', '')}.png" def _get_image_and_create_attachment( self, alert_type: str, alert_id: str, image_id: str, alert_subtype: str | None, ) -> dict | None: try: return_results(f"Trying to fetch {image_id=} ({alert_type=} {alert_subtype=} {alert_id=})") image_content = self.client.get_alert_image( alert_type=alert_type, alert_id=alert_id, image_id=image_id, alert_subtype=alert_subtype, ) return_results( f"Fetched {image_id=} ({alert_type=} {alert_subtype=} {alert_id=}): {str(image_content[:50])} " f"(truncated)" ) file_name = self._get_file_name_from_image_id(image_id) file_result_obj = fileResult(file_name, image_content) return_results(file_result_obj) # Important attachment = { "description": "Alert image", "name": file_result_obj.get("File"), "path": file_result_obj.get("FileID"), "showMediaFile": True, } return attachment except Exception as e: demisto.error(f"Failed to fetch image {image_id}: {str(e)}") return None def get_alert_images_command(self) -> list[CommandResults]: incident = demisto.incident() if not isinstance(incident, dict) or incident.get("isPlayground") is True: return_error("This command can only run from an incident War Room context.") return [] # return_error will exit(0), but to make linter happy. custom_fields = incident.get("CustomFields") if not isinstance(custom_fields, dict): custom_fields = {} alert_id = custom_fields.get("alertid") if not alert_id: return_error("Failed to get alert id from the incident.") return [] # return_error will exit(0), but to make linter happy. lookup_result = self.client.alert_lookup(alert_id) if isinstance(lookup_result, list) and lookup_result and isinstance(lookup_result[0], CommandResults): lookup_data: dict = lookup_result[0].outputs # type: ignore else: return_error("Failed to lookup alert.") return [] # return_error will exit(0), but to make linter happy. alert_type: str = lookup_data.get("type") or "" alert_subtype: str = lookup_data.get("subtype") or "" image_ids = lookup_data.get("images", []) or [] if not image_ids: return [CommandResults(readable_output="No screenshots found in alert details.")] context = demisto.context() or {} files = demisto.get(context, "File") if not files: files = [] if not isinstance(files, list): files = [files] existing_file_names = {f.get("Name") for f in files if isinstance(f, dict)} # Determine missing image IDs. missing_image_ids: set = set() for img_id in image_ids: # Limit to only 25 images. if len(missing_image_ids) >= MAX_IMAGES_TO_FETCH: break file_name = self._get_file_name_from_image_id(img_id) if file_name not in existing_file_names: missing_image_ids.add(img_id) if not missing_image_ids: return [CommandResults(readable_output="No new images to fetch.")] # Fetch missing images concurrently using thread pool. new_attachments = [] with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: futures = {} for img_id in missing_image_ids: future = executor.submit( self._get_image_and_create_attachment, alert_type=alert_type, alert_id=alert_id, image_id=img_id, alert_subtype=alert_subtype, ) futures[future] = img_id for future in concurrent.futures.as_completed(futures): attachment = future.result() if attachment: new_attachments.append(attachment) if not new_attachments: return [ CommandResults( readable_output="No new images were fetched.", ) ] message = f"Fetched {len(new_attachments)} new image(s)." return [ CommandResults( readable_output=message, ) ] def get_client(): demisto_params = demisto.params() base_url = demisto_params.get("url", "").rstrip("/") verify_ssl = not demisto_params.get("insecure", False) proxy = demisto_params.get("proxy", False) api_token = demisto_params.get("credentials", {}).get("password") if not api_token: return_error("Please provide a valid API token") headers = { "X-RFToken": api_token, "X-RF-User-Agent": ( f"RecordedFuture.py/{__version__} ({platform.platform()}) " f"XSOAR/{__version__} " f"RFClient/{__version__} (Cortex_XSOAR_{demisto.demistoVersion()['version']})" ), } return Client(base_url=base_url, verify=verify_ssl, headers=headers, proxy=proxy) def main(): try: client = get_client() command = demisto.command() actions = Actions(client) if command == "test-module": actions.test_module() elif command == "fetch-incidents": actions.fetch_incidents() elif command == "rf-alert-rules": return_results(actions.alert_rule_search_command()) elif command == "rf-alerts": return_results(actions.alert_search_command()) elif command == "rf-alert-update": return_results(actions.alert_update_command()) elif command == "rf-alert-lookup": return_results(actions.alert_lookup_command()) elif command == "rf-alert-images": return_results(actions.get_alert_images_command()) except Exception as e: return_error( message=f"Failed to execute {demisto.command()} command: {str(e)}", error=e, ) if __name__ in ("__main__", "__builtin__", "builtins"): # pragma: no cover main()