import demistomock as demisto # noqa: F401 from CommonServerPython import * # noqa: F401 SCORE_TABLE_FILE = { "unknown": Common.DBotScore.NONE, "benign": Common.DBotScore.GOOD, "grayware": Common.DBotScore.SUSPICIOUS, "malicious": Common.DBotScore.BAD, } class ReputationType: FILE = "file" CVE = "cve" ANTIVIRUS = "antivirus" SPYWARE = "spyware" FILEFORMAT = "fileformat" VULNERABILITY = "vulnerability" DNS = "dns" RTDNS = "rtdns" SPYWAREC2 = "spywarec2" RELEASE_NOTES = "release_notes" HEADERS_FILE = ["FileType", "MD5", "SHA256", "SHA1", "Size", "Status"] HEADERS_CVE = ["ID", "Description", "Score", "Published", "Modified"] HEADERS_SPYWARE = [ "ThreatID", "Name", "Description", "Vendor", "Score", "Default action", "Details", "Reference", "Status", "Min version", "Max version", "CVE", ] HEADERS_VULNERABILITY = [ "ThreatID", "Name", "Description", "Category", "Score", "Default action", "Vendor", "Reference", "Status", "Published version", "Latest release version", "Published", "Latest release time", "CVE", ] HEADERS_FILEFORMAT = [ "ThreatID", "Name", "Description", "Category", "Score", "Default action", "Vendor", "Reference", "Status", "Published version", "Latest release version", "Published", "Latest release time", ] HEADERS_ANTIVIRUS = [ "ThreatID", "Name", "Description", "Subtype", "Score", "Action", "Creation Time", "Related SHA256 hashes", "Release", ] HEADERS_DNS_RTDNS_SPYWAREC2 = [ "ThreatID", "Name", "Description", "Severity", "Type", "Subtype", "Action", "Creation Time", "Status", "Release", ] DATE_REGEX = r"\d{4}-[0-9]{2}-[0-9]{2}$" class Client(BaseClient): """ Client to use in the Threat Vault integration. Overrides BaseClient. """ def __init__(self, base_url: str, api_key: str, verify: bool, proxy: bool, reliability: str): super().__init__( base_url=base_url, verify=verify, proxy=proxy, headers={"Content-Type": "application/json", "X-API-KEY": api_key}, ) self.name = "ThreatVault" self.reliability = reliability def ip_feed_get_request(self, arg: str, value: str) -> dict: # pragma: no cover suffix = "ip-feed" return self._http_request(method="GET", url_suffix=suffix, params={arg: value}) def ip_feed_batch_post_request(self, arg: str, value: str) -> dict: # pragma: no cover suffix = "ip-feed" payload = json.dumps({"ipaddr": value}) return self._http_request(method="POST", url_suffix=suffix, data=payload) def antivirus_signature_get_request(self, arg: str, value: str) -> dict: # pragma: no cover suffix = "threats" return self._http_request(method="GET", url_suffix=suffix, params={arg: value}) def release_notes_get_request(self, type_: str, version: str) -> dict: # pragma: no cover suffix = "release-notes" return self._http_request(method="GET", url_suffix=suffix, params={"type": type_, "version": version}) def threat_batch_search_request(self, arg: str, value: list, type_: str) -> dict: params: dict[str, Union[list, str]] = {arg: value} if type_: params["type"] = type_ params = json.dumps(params) suffix = "threats" return self._http_request(method="POST", url_suffix=suffix, data=params) def threat_search_request(self, args: dict) -> dict: suffix = "threats" return self._http_request(method="GET", url_suffix=suffix, params=args) def atp_batch_report_request(self, args: str, value: list) -> dict: params: dict[str, Union[list, str]] = {args: value} params = json.dumps(params) suffix = "atp/reports" return self._http_request(method="POST", url_suffix=suffix, data=params) def atp_report_pcap_request(self, args: dict) -> dict: suffix = "atp/reports/pcaps" pcap_response = self._http_request(method="GET", url_suffix=suffix, params=args, resp_type="response") return pcap_response """ HELP FUNCTIONS """ def reputation_type_to_hr(reputation_type: str) -> str: match reputation_type: case ReputationType.RTDNS | ReputationType.DNS: return reputation_type.upper() case ReputationType.SPYWAREC2: return "SpywareC2" case _: return reputation_type.capitalize() def validate_arguments_search_command( cve: str | None, vendor: str | None, name: str | None, from_release_date: str | None, to_release_date: str | None, from_release_version: str | None, to_release_version: str | None, release_date: str | None, release_version: str | None, type_: str | None, ) -> None: if sum(1 for x in (cve, vendor, name) if x) > 1: raise ValueError("Only one of the following can be used at a time: cve, vendor, name") if sum(1 for x in (from_release_date, to_release_date) if x) == 1: raise ValueError( "When using a release date range in a query, it must be used with the following two arguments: " "from-release-date, to-release-date" ) if sum(1 for x in (from_release_version, to_release_version) if x) == 1: raise ValueError( "When using a release version range in a query, it must be used with the following two arguments: " "from-release-version, to-release-version" ) if release_date and release_version: raise ValueError("There can only be one argument from the following list in the command: release-date, release-version") if (from_release_date or from_release_version) and (release_date or release_version): raise ValueError( "When using a release version range or a release date range in a query" "it is not possible to use with the following arguments: release-date, release-version" ) if from_release_date and from_release_version: raise ValueError("from-release-version and from-release-date cannot be used together.") if not any( ( cve, vendor, name, type_, from_release_date, from_release_version, release_date, release_version, ) ): raise ValueError( "One of following arguments is required: cve, vendor, signature-name, type, " "from-release-version, from-release-date, release-date, release-version" ) def parse_date(date: str = None) -> str | None: if not date: return None if re.match(DATE_REGEX, date): return date date_time, _ = parse_date_range(date) return date_time.date().strftime("%Y-%m-%d") def pagination(page: Optional[int], page_size: Optional[int], limit: Optional[int]) -> tuple[int, Optional[int]]: """ The page_size and page arguments are converted so they match the offset and limit parameters of the API call. """ if page and page_size: if page < 0: raise ValueError("The page number must be a positive number") return page * page_size, page_size if not page and not page_size: return 0, limit raise ValueError("When using a pagination, it must be used with the following two arguments -> [page, page_size]") def resp_to_hr(response: dict, type_: str, expanded: bool = False) -> dict: match type_: case ReputationType.FILE: antivirus = response.get("signatures", {}).get("antivirus", ({},))[0] table_for_md = { "Status": antivirus.get("status"), "FileType": response.get("filetype"), "MD5": response.get("md5"), "SHA256": response.get("sha256"), "SHA1": response.get("sha1"), "Size": response.get("size"), } if expanded: table_for_md.update( { "Release": antivirus.get("release"), "Creation Time": response.get("create_time"), "SignatureId": antivirus.get("id"), "Family": response.get("family"), "Platform": response.get("platform"), "Signature Name": antivirus.get("name"), "Score": antivirus.get("severity"), "Description": antivirus.get("description"), "Wildfire verdict": response.get("wildfire_verdict"), } ) case ReputationType.CVE: table_for_md = { "ID": response.get("cve"), "Score": response.get("severity"), "Published": response.get("ori_release_time"), "Modified": response.get("latest_release_time"), "Description": response.get("description"), } case ReputationType.FILEFORMAT: table_for_md = { "ThreatID": response.get("id"), "Name": response.get("name"), "Description": response.get("description"), "Category": response.get("category"), "Score": response.get("severity"), "Default action": response.get("default_action"), "Vendor": response.get("vendor"), "Reference": response.get("reference"), "Status": response.get("status"), "Published version": response.get("ori_release_version"), "Latest release version": response.get("latest_release_version"), "Published": response.get("ori_release_time"), "Latest release time": response.get("latest_release_time"), } case ReputationType.VULNERABILITY: table_for_md = { "ThreatID": response.get("id"), "Name": response.get("name"), "Description": response.get("description"), "Category": response.get("category"), "Score": response.get("severity"), "Default action": response.get("default_action"), "Vendor": response.get("vendor"), "Reference": response.get("reference"), "Status": response.get("status"), "Published version": response.get("ori_release_version"), "Latest release version": response.get("latest_release_version"), "Published": response.get("ori_release_time"), "Latest release time": response.get("latest_release_time"), "CVE": response.get("cve"), } case ReputationType.ANTIVIRUS: table_for_md = { "ThreatID": response.get("id"), "Name": response.get("name"), "Description": response.get("description"), "Subtype": response.get("subtype"), "Score": response.get("severity"), "Action": response.get("action"), "Creation Time": response.get("create_time"), "Related SHA256 hashes": response.get("related_sha256_hashes"), "Release": response.get("release"), } case ReputationType.SPYWARE: table_for_md = { "ThreatID": response.get("id"), "Name": response.get("name"), "Description": response.get("description"), "Vendor": response.get("vendor"), "Score": response.get("severity"), "Default action": response.get("default_action"), "Details": response.get("details"), "Reference": response.get("reference"), "Status": response.get("status"), "Min version": response.get("min_version"), "Max version": response.get("max_version"), "CVE": response.get("cve"), } case ReputationType.RTDNS | ReputationType.DNS | ReputationType.SPYWAREC2: table_for_md = { "ThreatID": response.get("id"), "Name": response.get("name"), "Description": response.get("description"), "Severity": response.get("severity"), "Type": response.get("type"), "Subtype": response.get("subtype"), "Action": response.get("action"), "Creation Time": response.get("create_time"), "Status": response.get("status"), "Release": response.get("release"), } case ReputationType.RELEASE_NOTES: applications = response.get("release_notes", {}).get("applications", {}) spyware = response.get("release_notes", {}).get("spyware", {}) vulnerability = response.get("release_notes", {}).get("vulnerability", {}) table_for_md = { "Release version": response.get("release_version"), "Content version": response.get("content_version"), "type": response.get("type"), "Notes": response.get("release_notes", {}).get("notes"), "New applications": applications.get("new"), "Modified applications": applications.get("modified"), "Obsoleted applications": applications.get("obsoleted"), "New Spyware": spyware.get("new"), "Modified Spyware": spyware.get("modified"), "Disabled Spyware": spyware.get("disabled"), "New Vulnerability": vulnerability.get("new")[0] if vulnerability.get("new") else None, "Modified Vulnerability": vulnerability.get("modified")[0] if vulnerability.get("modified") else None, "Disabled Vulnerability": vulnerability.get("disabled")[0] if vulnerability.get("disabled") else None, "Release time": response.get("release_time"), } case _: demisto.debug(f"Unexpected item type {type_}") return {} return table_for_md def parse_resp_by_type(response: dict, expanded: bool = False) -> List[CommandResults]: command_results_list: List[CommandResults] = [] reputation_types = ( (ReputationType.ANTIVIRUS, HEADERS_ANTIVIRUS), (ReputationType.SPYWARE, HEADERS_SPYWARE), (ReputationType.VULNERABILITY, HEADERS_VULNERABILITY), (ReputationType.FILEFORMAT, HEADERS_FILEFORMAT), (ReputationType.DNS, HEADERS_DNS_RTDNS_SPYWAREC2), (ReputationType.RTDNS, HEADERS_DNS_RTDNS_SPYWAREC2), (ReputationType.SPYWAREC2, HEADERS_DNS_RTDNS_SPYWAREC2), ) for rep_type, headers_type in reputation_types: if rep_type in response["data"]: if expanded: responses = response.get("data", {}).get(rep_type, []) else: responses = [response.get("data", {}).get(rep_type, ([],))[0]] reputation_types_readable = reputation_type_to_hr(rep_type) for result in responses: table_for_md = resp_to_hr(response=result, type_=rep_type) readable_output = tableToMarkdown( name=f"{reputation_types_readable} Reputation: {result.get('id')}", t=table_for_md, headers=headers_type, removeNull=True, ) command_results_list.append( CommandResults( outputs_prefix=f"ThreatVault.{reputation_types_readable}", outputs_key_field="id", outputs=result if expanded else response.get("data", {}).get(rep_type, []), readable_output=readable_output, ) ) return command_results_list """ COMMANDS """ def ip_command(client: Client, args: dict) -> List[CommandResults]: """Retrieve information about the inputted IP from ThreatVault Args: client (Client): An instance of the client to call the GET commands. args (dict): The arguments inputted by the user. Returns: List[CommandResults]: A list of CommandResults objects to be returned to XSOAR. """ def headers_transform(header): headers = {"ipaddr": "IP", "geo": "Country", "asn": "ASN", "name": "Feed Name"} return headers[header] ips = argToList(args["ip"]) command_results_list: List[CommandResults] = [] dbot_reliability = DBotScoreReliability.get_dbot_score_reliability_from_str(client.reliability) try: if len(ips) == 1: # Call single IP info response = client.ip_feed_get_request(arg="ipaddr", value=ips[0]) else: # Call batch command response = client.ip_feed_batch_post_request(arg="ipaddr", value=ips) except DemistoException: raise if response: for data in response["data"]: ip_type = FeedIndicatorType.ip_to_indicator_type(data["ipaddr"]) dbot_score = Common.DBotScore( indicator=data["ipaddr"], indicator_type=DBotScoreType.IP, integration_name=client.name, score=3 if data["status"] == "released" else 0, reliability=dbot_reliability, ) ip = Common.IP( ip_type=ip_type, ip=data["ipaddr"], asn=data["asn"].split(" ")[0], as_owner=re.sub("[()]", "", data["asn"].split(" ")[1]), geo_country=data["geo"].split(" ")[0], geo_description=re.sub("[()]", "", data["geo"].split(" ")[1]), dbot_score=dbot_score, ) readable_output = tableToMarkdown( name="IP Feed Information", t=data, headers=["ipaddr", "geo", "asn", "name"], headerTransform=headers_transform, removeNull=True, ) command_results = CommandResults( readable_output=readable_output, outputs=data, outputs_prefix="ThreatVault.IP", indicator=ip, ) command_results_list.append(command_results) return command_results_list def file_command(client: Client, args: Dict) -> List[CommandResults]: """ Get the reputation of a sha256 or a md5 representing an antivirus """ readable_output = "" file_info: dict = {} hashes = argToList(args.get("file")) command_results_list: List[CommandResults] = [] dbot_reliability = DBotScoreReliability.get_dbot_score_reliability_from_str(client.reliability) for _hash in hashes: type_hash = get_hash_type(_hash) try: response = client.antivirus_signature_get_request(arg=type_hash, value=_hash) except DemistoException as err: if err.res is not None and err.res.status_code == 404: response = {} dbot_score = Common.DBotScore( indicator=_hash, indicator_type=DBotScoreType.FILE, integration_name=client.name, reliability=dbot_reliability, score=Common.DBotScore.NONE, ) file = Common.File( sha256=_hash if type_hash == "sha256" else None, md5=_hash if type_hash == "md5" else None, dbot_score=dbot_score, ) readable_output = f"Hash {_hash} antivirus reputation is unknown to Threat Vault." file_info = {} else: raise if response: file_info = response.get("data", {}).get("fileinfo", ({},))[0] dbot_score = Common.DBotScore( indicator=_hash, indicator_type=DBotScoreType.FILE, integration_name=client.name, score=SCORE_TABLE_FILE[file_info.get("wildfire_verdict", "unknown")], reliability=dbot_reliability, ) file = Common.File( sha256=file_info.get("sha256"), md5=file_info.get("md5"), sha1=file_info.get("sha1"), dbot_score=dbot_score, ) table_for_md = resp_to_hr(response=file_info, type_="file", expanded=args.get("expanded", False)) readable_output = tableToMarkdown( name=f"Antivirus Reputation for hash: {_hash}", t=table_for_md, headers=HEADERS_FILE, removeNull=True, ) else: file = Common.File(dbot_score=0) demisto.debug("No response. Initialized file variable.") command_results = CommandResults( readable_output=readable_output, outputs=file_info, outputs_prefix="ThreatVault.FileInfo", indicator=file, ) command_results_list.append(command_results) return command_results_list def cve_command(client: Client, args: Dict) -> List[CommandResults]: readable_output = "" _cve = None cves = argToList(args.get("cve")) command_results_list: List[CommandResults] = [] for cve in cves: try: response = client.antivirus_signature_get_request(arg="cve", value=cve) except DemistoException as err: if err.res is not None and err.res.status_code == 404: response = {} readable_output = f"CVE {cve} vulnerability reputation is unknown to Threat Vault." _cve = None vulnerability = None else: raise if response: vulnerability = response.get("data", {}).get("vulnerability", ({},))[0] _cve = Common.CVE( id=vulnerability.get("cve")[0], cvss=vulnerability.get("severity"), published=vulnerability.get("ori_release_time"), modified=vulnerability.get("latest_release_time"), description=vulnerability.get("description"), ) table_for_md = resp_to_hr(response=vulnerability, type_="cve") readable_output = tableToMarkdown( name=f"CVE Vulnerability Reputation: {cve}", t=table_for_md, headers=HEADERS_CVE, removeNull=True, ) command_results = CommandResults( readable_output=readable_output, outputs=vulnerability, outputs_prefix="ThreatVault.Vulnerability", indicator=_cve, ) command_results_list.append(command_results) return command_results_list def threat_signature_get_command(client: Client, args: Dict) -> List[CommandResults]: args["file"] = args.get("sha256", "") if md5 := args.get("md5"): args["file"] += f",{md5}" if args["file"] else md5 args["expanded"] = True ids = argToList(args.get("signature_id")) if not any((ids, args["file"])): raise ValueError("One of following arguments is required: signature_id, sha256, md5") if ids and args["file"]: raise ValueError("The command cannot be run with more than one argument.") command_results_list: List[CommandResults] = [] if args["file"]: command_results_list.extend(file_command(client=client, args=args)) return command_results_list for _id in ids: try: response = client.antivirus_signature_get_request(arg="id", value=_id) except DemistoException as err: if err.res is not None and err.res.status_code == 404: response = {} readable_output = f"{_id} reputation is unknown to Threat Vault." command_results_list.append(CommandResults(readable_output=readable_output)) else: raise if response: command_results_list.extend(parse_resp_by_type(response=response)) return command_results_list def release_note_get_command(client: Client, args: Dict) -> CommandResults: if not args.get("version"): raise ValueError("The version argument is required") version = args["version"] try: response = client.release_notes_get_request("content", version) except DemistoException as err: if err.res is not None and err.res.status_code == 404: return CommandResults(readable_output=f"Release note {version} was not found.") else: raise data = response.get("data", ({},))[0] table_for_md = resp_to_hr(response=data, type_="release_notes") readable_output = tableToMarkdown(name="Release notes:", t=table_for_md, removeNull=True) return CommandResults( outputs_prefix="ThreatVault.ReleaseNote", outputs_key_field="release_version", outputs=data, readable_output=readable_output, ) def threat_batch_search_command(client: Client, args: Dict) -> List[CommandResults]: ids = argToList(args.get("id")) md5 = argToList(args.get("md5")) sha256 = argToList(args.get("sha256")) names = argToList(args.get("name")) threat_type = args.get("type", "") argument_count = sum(1 for x in (ids, md5, sha256, names) if x) if argument_count != 1: raise ValueError("Only one of the following can be used at a time: id, md5, sha256, name") command_results_list: List[CommandResults] = [] if ids or names: type_ = "id" if ids else "name" try: response = client.threat_batch_search_request(arg=type_, value=ids if ids else names, type_=threat_type) except DemistoException as err: if err.res is not None and err.res.status_code == 404: response = {} readable_output = f"There is no information about the {str(ids) if ids else str(names)}" command_results_list.append(CommandResults(readable_output=readable_output)) else: raise if response: command_results_list.extend(parse_resp_by_type(response, True)) elif md5 or sha256: dbot_reliability = DBotScoreReliability.get_dbot_score_reliability_from_str(client.reliability) type_ = "md5" if md5 else "sha256" try: response = client.threat_batch_search_request(arg=type_, value=md5 or sha256, type_=threat_type) except DemistoException as err: if err.res is not None and err.res.status_code == 404: response = {} readable_output = f"There is no information about the {str(md5) if md5 else str(sha256)}" command_results_list.append(CommandResults(readable_output=readable_output)) else: raise if response: files_info: List[dict] = response.get("data", {}).get("fileinfo", []) for file_info in files_info: dbot_score = Common.DBotScore( indicator=file_info.get("sha256"), indicator_type=DBotScoreType.FILE, integration_name=client.name, score=SCORE_TABLE_FILE[file_info.get("wildfire_verdict", "unknown")], reliability=dbot_reliability, ) file = Common.File( sha256=file_info.get("sha256"), md5=file_info.get("md5"), sha1=file_info.get("sha1"), dbot_score=dbot_score, ) table_for_md = resp_to_hr(response=file_info, type_="file", expanded=True) readable_output = tableToMarkdown( name=f"File {file_info.get('sha256')}:", t=table_for_md, removeNull=True, ) command_results_list.append( CommandResults( outputs_prefix="ThreatVault.FileInfo", readable_output=readable_output, outputs_key_field="sha256", outputs=file_info, indicator=file, ) ) return command_results_list def threat_search_command(client: Client, args: Dict) -> List[CommandResults]: cve = args.get("cve") vendor = args.get("vendor") name = args.get("signature-name") from_release_date = parse_date(args.get("from-release-date")) to_release_date = parse_date(args.get("to-release-date")) from_release_version = args.get("from-release-version") to_release_version = args.get("to-release-version") release_date = parse_date(args.get("release-date")) release_version = args.get("release-version") type_ = args.get("type") page = arg_to_number(args.get("page")) page_size = arg_to_number(args.get("page_size")) offset, limit = pagination(page, page_size, arg_to_number(args.get("limit", 50))) validate_arguments_search_command( cve, vendor, name, from_release_date, to_release_date, from_release_version, to_release_version, release_date, release_version, type_, ) query = assign_params( cve=cve, vendor=vendor, name=name, fromReleaseDate=from_release_date, toReleaseDate=to_release_date, fromReleaseVersion=from_release_version, toRelaseVersion=to_release_version, releaseDate=release_date, releaseVersion=release_version, type=type_, offset=offset, limit=limit, ) command_results_list: List[CommandResults] = [] try: response = client.threat_search_request(args=query) except DemistoException as err: if err.res is not None and err.res.status_code == 404: response = {} readable_output = "There is no information for your search." command_results_list.append(CommandResults(readable_output=readable_output)) else: raise if response: command_results_list.extend(parse_resp_by_type(response, True)) return command_results_list def atp_batch_report_command(client: Client, args: Dict) -> List[CommandResults]: # pragma: no cover report_ids = argToList(args.get("report_id")) command_results_list: List[CommandResults] = [] if report_ids: demisto.debug(f"Requesting report IDs: {report_ids}") try: response = client.atp_batch_report_request(args="id", value=report_ids) except DemistoException as err: if err.res is not None and err.res.status_code == 404: response = {} readable_output = f"There is no information about the {report_ids!s}" command_results_list.append(CommandResults(readable_output=readable_output)) else: raise if response: report_infos: List[dict] = response.get("data", {}).get("reports", {}).get("reports", []) for idx, report_info in enumerate(report_infos): demisto.debug(f"Processing report {idx + 1}/{len(report_infos)}: {report_info.get('report_id', 'unknown')}") readable_output = tableToMarkdown( name=f"Advanced Threat Prevention Report ID: {report_info.get('report_id')}:", t=report_info, removeNull=True, ) command_results_list.append( CommandResults( outputs_prefix="ThreatVault.ATP.Report", readable_output=readable_output, # outputs_key_field="sha256", outputs=report_info, ) ) demisto.debug(f"Successfully processed {len(command_results_list)} reports") return command_results_list def atp_report_pcap_command(client: Client, args: Dict) -> List[CommandResults]: # pragma: no cover report_id = args.get("report_id") if report_id: query = assign_params(id=report_id) command_results_list: List[CommandResults] = [] try: response = client.atp_report_pcap_request(args=query) response_data_headers = json.loads(json.dumps(dict(response.headers))) # type: ignore response_content = response.content # type: ignore except DemistoException as err: if err.res is not None and err.res.status_code == 404: response = {} readable_output = f"There is no information about the {report_id!s}" command_results_list.append(CommandResults(readable_output=readable_output)) else: raise # check for octet-stream response for PCAP if response_data_headers.get("Content-Type") == "application/octet-stream": # set the pcap filename to the report_id.pcap pcap_name = report_id + ".pcap" # write the file prperties to the context return_results(fileResult(pcap_name, response_content)) ec = {"ID": report_id, "Name": pcap_name} readable_output = tableToMarkdown( name="Advanced Threat Prevention PCAP Download:", t=ec, removeNull=True, ) command_results_list.append( CommandResults( outputs_prefix="ThreatVault.ATP.PCAP", readable_output=readable_output, # outputs_key_field="sha256", outputs=ec, ) ) else: return_results( { "Type": entryTypes["note"], "ContentsFormat": formats["text"], "Contents": f"No PCAP response for ID: {report_id!s}", } ) return command_results_list """ FETCH INCIDENT """ def fetch_incidents(client: Client, args: dict) -> List: """ Retrieving release notes that contain all the information about vulnerabilities, antivirus, spyware, and more. """ last_run = demisto.getLastRun() first_fetch = args.get("first_fetch", "3 Days") if not last_run.get("scound_fetch"): if first_fetch.strip().split(" ")[1].lower() not in frozenset( ("days", "month", "months", "year", "years") ): # only these are allowed raise ValueError("The unit of date_range is invalid. Must be days, months or years.") start_time, now = parse_date_range(first_fetch) else: _, now = parse_date_range(first_fetch) start_time = now current = start_time.date() now = now.date() incidents: List[dict] = [] while current <= now: try: # Bringing the daily date for the first api call demisto.debug(f"Time for request fetch-incidents -> {current}") response = client.threat_search_request({"releaseDate": current.strftime("%Y-%m-%d")}) except DemistoException as err: if err.res is not None and err.res.status_code == 404: current += timedelta(days=1) continue else: raise if keys_of_resp := tuple({"spyware", "vulnerability", "fileformat", "antivirus"}.intersection(response["data"].keys())): # The version of the release notes for the second API call can be extracted try: number_version = response["data"][keys_of_resp[0]][0]["latest_release_version"] except KeyError as err: raise Exception(f"Error parsing release note latest_release_version: {err!s}") # The API is called by the version number release = client.release_notes_get_request("content", number_version) # Adds source name to the incident release["data"][0]["Source name"] = "THREAT VAULT - RELEASE NOTES" # Incident organization and arrangement incidents.append( { "name": f"ThreatVault Release {release['data'][0]['release_version']}", "occurred": release["data"][0]["release_time"], "rawJSON": json.dumps(release), } ) current += timedelta(days=1) demisto.setLastRun({"scound_fetch": "true"}) return incidents def test_module(client: Client, *_) -> str: """Performs basic get request to get ip geo data. Args: client: Client object with request. Returns: string. """ client.threat_search_request({"type": "ips"}) return "ok" def main(): params = demisto.params() """PARAMS""" base_url = params.get("url", "") + "service/v1/" api_key = params.get("credentials", {}).get("password") verify = not params.get("insecure", False) proxy = params.get("proxy") reliability = params.get("integrationReliability", "D - Not usually reliable") if not DBotScoreReliability.is_valid_type(reliability): raise Exception("Please provide a valid value for the Source Reliability parameter.") try: command = demisto.command() demisto.debug(f"Command being called is {demisto.command()}") client = Client( base_url=base_url, api_key=api_key, verify=verify, proxy=proxy, reliability=reliability, ) commands = { "file": file_command, "cve": cve_command, "ip": ip_command, "threatvault-threat-signature-get": threat_signature_get_command, "threatvault-release-note-get": release_note_get_command, "threatvault-threat-batch-search": threat_batch_search_command, "threatvault-threat-search": threat_search_command, "threatvault-atp-batch-report-get": atp_batch_report_command, "threatvault-atp-report-pcap-get": atp_report_pcap_command, } if demisto.command() == "test-module": # This is the call made when pressing the integration Test button. return_results(test_module(client)) elif command == "fetch-incidents": incidents = fetch_incidents(client, params) demisto.incidents(incidents) elif command in commands: return_results(commands[command](client, demisto.args())) else: raise NotImplementedError(f'Command "{command}" was not implemented.') except Exception as err: demisto.error(traceback.format_exc()) # print the traceback return_error(f"Failed to execute {command} command.\nError:\n{err!s}") if __name__ in ("__main__", "__builtin__", "builtins"): main()