import demistomock as demisto # noqa: F401 from CommonServerPython import * # noqa: F401 """ An integration module for the Google Threat Intelligence API. API Documentation: https://gtidocs.virustotal.com/reference """ import ipaddress import re from collections import defaultdict from typing import cast from dateparser import parse INTEGRATION_NAME = "GoogleThreatIntelligence" COMMAND_PREFIX = "gti" INTEGRATION_ENTRY_CONTEXT = INTEGRATION_NAME INDICATOR_TYPE = { "ip": FeedIndicatorType.IP, "ip_address": FeedIndicatorType.IP, "domain": FeedIndicatorType.Domain, "file": FeedIndicatorType.File, "url": FeedIndicatorType.URL, "cve": FeedIndicatorType.CVE, } SEVERITY_LEVELS = { "SEVERITY_UNKNOWN": "UNKNOWN", "SEVERITY_LOW": "LOW", "SEVERITY_MEDIUM": "MEDIUM", "SEVERITY_HIGH": "HIGH", } VERDICTS = { "VERDICT_UNKNOWN": "UNKNOWN", "VERDICT_UNDETECTED": "UNDETECTED", "VERDICT_SUSPICIOUS": "SUSPICIOUS", "VERDICT_MALICIOUS": "MALICIOUS", } TYPE_TO_ENDPOINT = { "file": "files", "hash": "files", "domain": "domains", "url": "urls", "ip": "ip_addresses", } """RELATIONSHIP TYPE""" RELATIONSHIP_TYPE = { "file": { "carbonblack_children": EntityRelationship.Relationships.CREATES, "carbonblack_parents": EntityRelationship.Relationships.CREATED_BY, "compressed_parents": EntityRelationship.Relationships.BUNDLED_IN, "contacted_domains": EntityRelationship.Relationships.COMMUNICATES_WITH, "contacted_ips": EntityRelationship.Relationships.COMMUNICATES_WITH, "contacted_urls": EntityRelationship.Relationships.COMMUNICATES_WITH, "dropped_files": EntityRelationship.Relationships.DROPPED_BY, "email_attachments": EntityRelationship.Relationships.ATTACHES, "email_parents": EntityRelationship.Relationships.ATTACHMENT_OF, "embedded_domains": EntityRelationship.Relationships.EMBEDDED_IN, "embedded_ips": EntityRelationship.Relationships.EMBEDDED_IN, "embedded_urls": EntityRelationship.Relationships.EMBEDDED_IN, "execution_parents": EntityRelationship.Relationships.EXECUTED_BY, "itw_domains": EntityRelationship.Relationships.DOWNLOADS_FROM, "itw_ips": EntityRelationship.Relationships.DOWNLOADS_FROM, "overlay_children": EntityRelationship.Relationships.BUNDLES, "overlay_parents": EntityRelationship.Relationships.BUNDLED_IN, "pcap_children": EntityRelationship.Relationships.BUNDLES, "pcap_parents": EntityRelationship.Relationships.BUNDLED_IN, "pe_resource_children": EntityRelationship.Relationships.EXECUTED, "pe_resource_parents": EntityRelationship.Relationships.EXECUTED_BY, "similar_files": EntityRelationship.Relationships.SIMILAR_TO, }, "domain": { "cname_records": EntityRelationship.Relationships.IS_ALSO, "caa_records": EntityRelationship.Relationships.RELATED_TO, "communicating_files": EntityRelationship.Relationships.DROPS, "downloaded_files": EntityRelationship.Relationships.DROPS, "immediate_parent": EntityRelationship.Relationships.SUB_DOMAIN_OF, "mx_records": EntityRelationship.Relationships.RELATED_TO, "ns_records": EntityRelationship.Relationships.DROPS, "parent": EntityRelationship.Relationships.SUB_DOMAIN_OF, "referrer_files": EntityRelationship.Relationships.RELATED_TO, "resolutions": EntityRelationship.Relationships.RESOLVED_FROM, "siblings": EntityRelationship.Relationships.SUPRA_DOMAIN_OF, "soa_records": EntityRelationship.Relationships.IS_ALSO, "subdomains": EntityRelationship.Relationships.SUPRA_DOMAIN_OF, "urls": EntityRelationship.Relationships.HOSTS, }, "ip": { "communicating_files": EntityRelationship.Relationships.COMMUNICATES_WITH, "downloaded_files": EntityRelationship.Relationships.DROPS, "referrer_files": EntityRelationship.Relationships.RELATED_TO, "resolutions": EntityRelationship.Relationships.RESOLVES_TO, "urls": EntityRelationship.Relationships.RELATED_TO, }, "url": { "contacted_domains": EntityRelationship.Relationships.RELATED_TO, "contacted_ips": EntityRelationship.Relationships.RELATED_TO, "downloaded_files": EntityRelationship.Relationships.DROPS, "last_serving_ip_address": EntityRelationship.Relationships.RESOLVED_FROM, "network_location": EntityRelationship.Relationships.RESOLVED_FROM, "redirecting_urls": EntityRelationship.Relationships.DUPLICATE_OF, "redirects_to": EntityRelationship.Relationships.DUPLICATE_OF, "referrer_files": EntityRelationship.Relationships.EMBEDDED_IN, "referrer_urls": EntityRelationship.Relationships.RELATED_TO, }, "cve": {"referrer_cve": EntityRelationship.Relationships.RELATED_TO}, } class Client(BaseClient): """Client for Google Threat Intelligence API.""" reliability: DBotScoreReliability def __init__(self, params: dict): self.reliability = DBotScoreReliability.get_dbot_score_reliability_from_str(params["feedReliability"]) super().__init__( "https://www.virustotal.com/api/v3/", verify=not argToBoolean(params.get("insecure")), proxy=argToBoolean(params.get("proxy")), headers={ "x-apikey": params["credentials"]["password"], "x-tool": "CortexGTI", }, ) # region Reputation calls def ip(self, ip: str, relationships: str = "") -> dict: """ See Also: https://gtidocs.virustotal.com/reference/ip-info """ return self._http_request("GET", f"ip_addresses/{ip}?relationships={relationships}", ok_codes=(404, 429, 200)) def file(self, file: str, relationships: str = "") -> dict: """ See Also: https://gtidocs.virustotal.com/reference/file-info """ return self._http_request("GET", f"files/{file}?relationships={relationships}", ok_codes=(404, 429, 200)) # It is not a Reputation call def private_file(self, file: str) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/private-files-info """ return self._http_request("GET", f"private/files/{file}", ok_codes=(404, 429, 200)) def url(self, url: str, relationships: str = ""): """ See Also: https://gtidocs.virustotal.com/reference/url-info """ return self._http_request( "GET", f"urls/{encode_url_to_base64(url)}?relationships={relationships}", ok_codes=(404, 429, 200) ) def private_url(self, url: str): """ See Also: https://gtidocs.virustotal.com/reference/get-a-private-url-analysis-report """ return self._http_request("GET", f"private/urls/{encode_url_to_base64(url)}", ok_codes=(404, 429, 200)) def domain(self, domain: str, relationships: str = "") -> dict: """ See Also: https://gtidocs.virustotal.com/reference/domain-info """ return self._http_request("GET", f"domains/{domain}?relationships={relationships}", ok_codes=(404, 429, 200)) def cve(self, cve_id: str) -> dict: """ Get CVE (vulnerability) information from Google Threat Intelligence. Args: cve_id: CVE identifier (e.g., "CVE-2025-62173") Returns: dict: CVE information from the collections endpoint See Also: https://gtidocs.virustotal.com/reference/get-vulnerability """ # Format: vulnerability--cve-2025-62173 object_id = f"vulnerability--{cve_id.lower()}" return self._http_request("GET", f"collections/{object_id}", ok_codes=(404, 429, 200)) # endregion # region Comments call def delete_comment(self, id_: str): """ See Also: https://gtidocs.virustotal.com/reference/comment-id-delete """ self._http_request("DELETE", f"comments/{id_}", resp_type="response") def get_ip_comments(self, ip: str, limit: int) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/ip-comments-get """ return self._http_request("GET", f"ip_addresses/{ip}/comments", params={"limit": limit}) def get_url_comments(self, url: str, limit: int) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/urls-comments-get """ return self._http_request("GET", f"urls/{encode_url_to_base64(url)}/comments", params={"limit": limit}) def get_hash_comments(self, file_hash: str, limit: int) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/files-comments-get """ return self._http_request("GET", f"files/{file_hash}/comments", params={"limit": limit}) def get_domain_comments(self, domain: str, limit: int) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/domains-comments-get """ return self._http_request("GET", f"domains/{domain}/comments", params={"limit": limit}) def get_comment_by_id(self, comment_id: str) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/get-comment """ return self._http_request("GET", f"comments/{comment_id}") def add_comment(self, suffix: str, comment: str) -> dict: """Sending POST HTTP request to comment Args: suffix: suffix of the comment comment: the comment itself Returns: json of response """ return self._http_request("POST", suffix, json_data={"data": {"type": "comment", "attributes": {"text": comment}}}) def add_comment_to_ip(self, ip: str, comment: str) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/ip-comments-post """ return self.add_comment(f"ip_addresses/{ip}/comments", comment) def add_comment_to_url(self, url: str, comment: str) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/urls-comments-post """ return self.add_comment(f"urls/{encode_url_to_base64(url)}/comments", comment) def add_comment_to_domain(self, domain: str, comment: str) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/domains-comments-post """ return self.add_comment(f"domains/{domain}/comments", comment) def add_comment_to_file(self, resource: str, comment: str) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/files-comments-post """ return self.add_comment(f"files/{resource}/comments", comment) # endregion # region Scan calls def file_rescan(self, file_hash: str) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/files-analyse """ return self._http_request("POST", f"/files/{file_hash}/analyse") def file_scan(self, file_path: str, /, upload_url: Optional[str]) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/files-analyse """ response: requests.Response with open(file_path, "rb") as file: if upload_url or os.stat(file_path).st_size / (1024 * 1024) >= 32: if not upload_url: raw_response = self.get_upload_url() upload_url = raw_response["data"] response = self._http_request("POST", full_url=upload_url, files={"file": file}, resp_type="response") else: response = self._http_request("POST", url_suffix="/files", files={"file": file}, resp_type="response") demisto.debug(f"scan_file response:\n{response.status_code=!s}, {response.headers=!s}, {response.content!s}") return response.json() def private_file_scan(self, file_path: str) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/post_files-1 """ response: requests.Response with open(file_path, "rb") as file: if os.stat(file_path).st_size / (1024 * 1024) >= 32: raw_response = self.get_private_upload_url() upload_url = raw_response["data"] response = self._http_request("POST", full_url=upload_url, files={"file": file}, resp_type="response") else: response = self._http_request("POST", url_suffix="/private/files", files={"file": file}, resp_type="response") demisto.debug(f"scan_private_file response:\n{response.status_code=!s}, {response.headers=!s}, {response.content!s}") return response.json() def get_upload_url(self) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/files-upload-url """ return self._http_request("GET", "files/upload_url") def get_private_upload_url(self) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/private-files-upload-url """ return self._http_request("GET", "private/files/upload_url") def url_scan(self, url: str) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/urls-analyse """ return self._http_request("POST", "urls", data={"url": url}) def private_url_scan(self, url: str) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/private-scan-url """ return self._http_request("POST", "/private/urls", data={"url": url}) # endregion def file_sandbox_report(self, file_hash: dict, limit: int) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/files-relationships """ return self._http_request("GET", f"files/{file_hash}/behaviours", params={"limit": limit}, ok_codes=(404, 429, 200)) def passive_dns_data(self, id: dict, limit: int) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/ip-relationships """ return self._http_request( "GET", f'{"ip_addresses" if id["type"] == "ip" else "domains"}/{id["value"]}/resolutions', params={"limit": limit} ) def search(self, query: str, limit: int) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/intelligence-search """ return self._http_request("GET", "search", params={"query": query, "limit": limit}) def get_analysis(self, analysis_id: str) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/analysis """ return self._http_request("GET", f"/analyses/{analysis_id}") def get_private_analysis(self, analysis_id: str) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/private-analysis """ return self._http_request("GET", f"private/analyses/{analysis_id}") def get_private_item_from_analysis(self, analysis_id: str) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/analysesidrelationship """ return self._http_request("GET", f"private/analyses/{analysis_id}/item") def get_file_sigma_analysis(self, file_hash: str) -> dict: """ See Also: https://gtidocs.virustotal.com/reference/files-relationships """ return self._http_request( "GET", f"files/{file_hash}/sigma_analysis", ) def curated_collections(self, resource_id: str, resource_type: str, collection_type: str) -> dict: """Returns curated collections.""" if resource_type not in TYPE_TO_ENDPOINT: raise DemistoException(f'Could not find resource type of "{resource_type}"') if collection_type not in ("campaign", "malware-family", "threat-actor"): raise DemistoException(f'Could not find collection type of "{collection_type}"') if resource_type == "url": resource_id = encode_url_to_base64(resource_id) collection_type_filter = f"collection_type:{collection_type}" if collection_type == "malware-family": collection_type_filter = f"({collection_type_filter} OR collection_type:software-tookit)" return self._http_request( "GET", f"{TYPE_TO_ENDPOINT[resource_type]}/{resource_id}/associations", params={ "filter": f"owner:Mandiant {collection_type_filter}", "exclude_attributes": "aggregations", }, ok_codes=(404, 429, 200), ) class ScoreCalculator: """ Calculating DBotScore of files, ip, etc. """ DEFAULT_SUSPICIOUS_THRESHOLD = 5 DEFAULT_RELATIONSHIP_SUSPICIOUS_THRESHOLD = 2 GTI_MALICIOUS_VERDICT = "VERDICT_MALICIOUS" GTI_SUSPICIOUS_VERDICT = "VERDICT_SUSPICIOUS" logs: List[str] # General trusted_vendors_threshold: int trusted_vendors: List[str] gti_malicious: bool gti_suspicious: bool # IP ip_threshold: dict[str, int] # URL url_threshold: dict[str, int] # Domain domain_threshold: dict[str, int] domain_popularity_ranking: int # File file_threshold: dict[str, int] sigma_ids_threshold: int crowdsourced_yara_rules_enabled: bool crowdsourced_yara_rules_threshold: int def __init__(self, params: dict): self.trusted_vendors = argToList(params["preferredVendors"]) self.trusted_vendors_threshold = arg_to_number_must_int( params["preferredVendorsThreshold"], arg_name="Preferred Vendor Threshold", required=True ) self.file_threshold = { "malicious": arg_to_number_must_int(params["fileThreshold"], arg_name="File Malicious Threshold", required=True), "suspicious": arg_to_number_must_int( params["fileSuspiciousThreshold"] or self.DEFAULT_SUSPICIOUS_THRESHOLD, arg_name="File Suspicious Threshold", required=True, ), } self.ip_threshold = { "malicious": arg_to_number_must_int(params["ipThreshold"], arg_name="IP Malicious Threshold", required=True), "suspicious": arg_to_number_must_int( params["ipSuspiciousThreshold"] or self.DEFAULT_SUSPICIOUS_THRESHOLD, arg_name="IP Suspicious Threshold", required=True, ), } self.url_threshold = { "malicious": arg_to_number_must_int(params["urlThreshold"], arg_name="URL Malicious Threshold", required=True), "suspicious": arg_to_number_must_int( params["urlSuspiciousThreshold"] or self.DEFAULT_SUSPICIOUS_THRESHOLD, arg_name="URL Suspicious Threshold", required=True, ), } self.domain_threshold = { "malicious": arg_to_number_must_int(params["domainThreshold"], arg_name="Domain Malicious Threshold", required=True), "suspicious": arg_to_number_must_int( params["domainSuspiciousThreshold"] or self.DEFAULT_SUSPICIOUS_THRESHOLD, arg_name="Domain Suspicious Threshold", required=True, ), } self.crowdsourced_yara_rules_enabled = argToBoolean(params["crowdsourced_yara_rules_enabled"]) self.crowdsourced_yara_rules_threshold = arg_to_number_must_int(params["yaraRulesThreshold"]) self.sigma_ids_threshold = arg_to_number_must_int( params["SigmaIDSThreshold"], arg_name="Sigma and Intrusion Detection Rules Threshold", required=True ) self.domain_popularity_ranking = arg_to_number_must_int( params["domain_popularity_ranking"], arg_name="Domain Popularity Ranking Threshold", required=True ) self.gti_malicious = argToBoolean(params.get("gti_malicious", False)) self.gti_suspicious = argToBoolean(params.get("gti_suspicious", False)) self.logs = [] def get_logs(self) -> str: """Returns the log string""" return "\n".join(self.logs) def _is_by_threshold(self, analysis_stats: dict, threshold: int, suspicious: bool = False) -> bool: """Determines whatever the indicator malicious/suspicious by threshold. if number of malicious (+ suspicious) >= threshold -> Malicious (Suspicious) Args: analysis_stats: the analysis stats from the response. threshold: the threshold of the indicator type. suspicious: whether suspicious is also added. Returns: Whatever the indicator is malicious/suspicious by threshold. """ total = analysis_stats.get("malicious", 0) if suspicious: total += analysis_stats.get("suspicious", 0) verdict = "suspicious" if suspicious else "malicious" self.logs.append(f"{total} vendors found {verdict}.\nThe {verdict} threshold is {threshold}.") if total >= threshold: self.logs.append(f"Found as {verdict}: {total} >= {threshold}.") return True self.logs.append(f"Not found {verdict} by threshold: {total} < {threshold}.") return False def is_suspicious_by_threshold(self, analysis_stats: dict, threshold: int) -> bool: """Determines whatever the indicator suspicious by threshold. if number of malicious + suspicious >= threshold -> Suspicious Args: analysis_stats: the analysis stats from the response threshold: the threshold of the indicator type. Returns: Whatever the indicator is suspicious by threshold. """ return self._is_by_threshold(analysis_stats, threshold, suspicious=True) def is_good_by_popularity_ranks(self, popularity_ranks: dict) -> Optional[bool]: """Analyzing popularity ranks. if popularity ranks exist and average rank is < threshold -> Good Args: popularity_ranks: the popularity ranks object from response Returns: Whatever the indicator is good or not by popularity rank. """ if popularity_ranks: self.logs.append("Found popularity ranks. Analyzing.") average = sum(rank.get("rank", 0) for rank in popularity_ranks.values()) / len(popularity_ranks) self.logs.append(f"The average of the ranks is {average} and the threshold is {self.domain_popularity_ranking}") if average <= self.domain_popularity_ranking: self.logs.append("Indicator is good by popularity ranks.") return True else: self.logs.append("Indicator might not be good by it's popularity ranks.") return False self.logs.append("Could not determine rank by popularity, No popularity ranks data.") return None def is_suspicious_by_rules(self, file_response: dict) -> bool: """Check if indicator is suspicious by rules analysis. crowdsourced_yara_results >= yara_rules_threshold || sigma_analysis_stats.high + critical >= sigma_id_threshold || crowdsourced_ids_stats.high + critical >= sigma_id_threshold -> suspicious Args: file_response: the file response Returns: Whatever the file is suspicious by rules analysis. """ data = file_response.get("data", {}) if self.crowdsourced_yara_rules_enabled: self.logs.append("Crowdsourced Yara Rules analyzing enabled.") if (total_yara_rules := len(data.get("crowdsourced_yara_results", []))) >= self.crowdsourced_yara_rules_threshold: self.logs.append( "Found malicious by finding more Crowdsourced Yara Rules than threshold. \n" f"{total_yara_rules} >= {self.crowdsourced_yara_rules_threshold}" ) return True if sigma_rules := data.get("sigma_analysis_stats"): self.logs.append("Found sigma rules, analyzing.") sigma_high, sigma_critical = sigma_rules.get("high", 0), sigma_rules.get("critical", 0) if (sigma_high + sigma_critical) >= self.sigma_ids_threshold: self.logs.append(f"Found malicious, {sigma_high + sigma_critical} >= {self.sigma_ids_threshold}. ") return True else: self.logs.append("Not found malicious by sigma. ") else: self.logs.append("Not found sigma analysis. Skipping. ") if crowdsourced_ids_stats := data.get("crowdsourced_ids_stats"): self.logs.append("Found crowdsourced IDS analysis, analyzing. ") ids_high, ids_critical = crowdsourced_ids_stats.get("high"), crowdsourced_ids_stats.get("critical") if (ids_high + ids_critical) >= self.sigma_ids_threshold: self.logs.append(f"Found malicious, {(ids_high + ids_critical) >= self.sigma_ids_threshold}.") return True else: self.logs.append("Not found malicious by sigma.") else: self.logs.append("Not found crowdsourced IDS analysis. Skipping.") else: self.logs.append("Crowdsourced Yara Rules analyzing is not enabled. Skipping.") return False def is_preferred_vendors_pass_malicious(self, analysis_results: dict) -> bool: """Is the indicator counts as malicious by predefined malicious vendors. trusted_vendors.malicious >= trusted_vendors_threshold -> Malicious The function takes only the latest 20 results. Args: analysis_results: The results of the analysis. Returns: Whatever the indicator is malicious or not by preferred vendors. """ recent = {key: analysis_results[key] for key in list(analysis_results.keys())[:20]} preferred_vendor_scores = {vendor: recent[vendor] for vendor in self.trusted_vendors if vendor in recent} malicious_trusted_vendors = [item for item in preferred_vendor_scores.values() if item.get("category") == "malicious"] if len(malicious_trusted_vendors) >= self.trusted_vendors_threshold: self.logs.append( f"{len(malicious_trusted_vendors)} trusted vendors found the hash malicious. \n" f"The trusted vendors threshold is {self.trusted_vendors_threshold}. \n" f"Malicious check: {(len(malicious_trusted_vendors) >= self.trusted_vendors_threshold)=}. " ) return True else: self.logs.append( f"Those preferred vendors found the hash malicious: {malicious_trusted_vendors}. " f"They do not pass the threshold {self.trusted_vendors_threshold}. " ) return False def is_malicious_by_threshold(self, analysis_stats: dict, threshold: int) -> bool: """Determines whatever the indicator malicious by threshold. if number of malicious >= threshold -> Malicious Args: analysis_stats: the analysis stats from the response threshold: the threshold of the indicator type. Returns: Whatever the indicator is malicious by threshold. """ return self._is_by_threshold(analysis_stats, threshold) def score_by_threshold(self, analysis_stats: dict, threshold: dict[str, int]) -> int: """Determines the DBOTSCORE of the indicator by threshold only. Returns: DBotScore of the indicator. Can by Common.DBotScore.BAD, Common.DBotScore.SUSPICIOUS or Common.DBotScore.GOOD """ if self.is_malicious_by_threshold(analysis_stats, threshold["malicious"]): return Common.DBotScore.BAD if self.is_suspicious_by_threshold(analysis_stats, threshold["suspicious"]): return Common.DBotScore.SUSPICIOUS return Common.DBotScore.GOOD def score_by_results_and_stats(self, indicator: str, raw_response: dict, threshold: dict[str, int]) -> int: """Determines indicator score by popularity preferred vendors and threshold. Args: indicator: The indicator we analyzing. raw_response: The raw response from API. threshold: Threshold of the indicator. Returns: DBotScore of the indicator. Can by Common.DBotScore.BAD, Common.DBotScore.SUSPICIOUS or Common.DBotScore.GOOD """ self.logs.append(f'Basic analyzing of "{indicator}"') data = raw_response.get("data", {}) attributes = data.get("attributes", {}) popularity_ranks = attributes.get("popularity_ranks") last_analysis_results = attributes.get("last_analysis_results") last_analysis_stats = attributes.get("last_analysis_stats") if self.is_good_by_popularity_ranks(popularity_ranks): return Common.DBotScore.GOOD if self.is_preferred_vendors_pass_malicious(last_analysis_results): return Common.DBotScore.BAD return self.score_by_threshold(last_analysis_stats, threshold) def is_malicious_by_gti(self, gti_assessment: dict) -> bool: """Determines if an IoC is malicious according to its GTI assessment.""" if self.gti_malicious: return gti_assessment.get("verdict", {}).get("value") == self.GTI_MALICIOUS_VERDICT return False def is_suspicious_by_gti(self, gti_assessment: dict) -> bool: """Determines if an IoC is suspicious according to its GTI assessment.""" if self.gti_suspicious: return gti_assessment.get("verdict", {}).get("value") == self.GTI_SUSPICIOUS_VERDICT return False def file_score(self, given_hash: str, raw_response: dict) -> int: """Analyzing file score. The next parameters are analyzed: Preferred vendors Score by threshold Score by rules analysis (YARA, IDS and Sigma, if presents) Args: given_hash: The hash we're analyzing raw_response: The response from the API Returns: DBotScore of the indicator. Can by Common.DBotScore.BAD, Common.DBotScore.SUSPICIOUS or Common.DBotScore.GOOD """ self.logs.append(f"Analysing file hash {given_hash}.") data = raw_response.get("data", {}) attributes = data.get("attributes", {}) analysis_results = attributes.get("last_analysis_results", {}) analysis_stats = attributes.get("last_analysis_stats", {}) # GTI assessment if self.is_malicious_by_gti(attributes.get("gti_assessment", {})): return Common.DBotScore.BAD # Trusted vendors if self.is_preferred_vendors_pass_malicious(analysis_results): return Common.DBotScore.BAD score = self.score_by_threshold(analysis_stats, self.file_threshold) if score == Common.DBotScore.BAD: return Common.DBotScore.BAD suspicious_by_rules = self.is_suspicious_by_rules(raw_response) if score == Common.DBotScore.SUSPICIOUS and suspicious_by_rules: self.logs.append( f'Hash: "{given_hash}" was found malicious as the hash is suspicious both by threshold and rules analysis.' ) return Common.DBotScore.BAD elif suspicious_by_rules: self.logs.append(f'Hash: "{given_hash}" was found suspicious by rules analysis.') return Common.DBotScore.SUSPICIOUS elif score == Common.DBotScore.SUSPICIOUS: self.logs.append(f'Hash: "{given_hash}" was found suspicious by passing the threshold analysis.') return Common.DBotScore.SUSPICIOUS elif self.is_suspicious_by_gti(attributes.get("gti_assessment", {})): self.logs.append(f'Hash: "{given_hash}" was found suspicious by gti assessment.') return Common.DBotScore.SUSPICIOUS self.logs.append(f'Hash: "{given_hash}" was found good.') return Common.DBotScore.GOOD # Nothing caught def ip_score(self, indicator: str, raw_response: dict) -> int: """Determines indicator score by popularity preferred vendors and threshold. Args: indicator: The indicator we analyzing. raw_response: The response from the API Returns: DBotScore of the indicator. Can by Common.DBotScore.BAD, Common.DBotScore.SUSPICIOUS or Common.DBotScore.GOOD """ data = raw_response.get("data", {}) attributes = data.get("attributes", {}) # GTI assessment if self.is_malicious_by_gti(attributes.get("gti_assessment", {})): return Common.DBotScore.BAD score = self.score_by_results_and_stats(indicator, raw_response, self.ip_threshold) if score == Common.DBotScore.GOOD and self.is_suspicious_by_gti(attributes.get("gti_assessment", {})): score = Common.DBotScore.SUSPICIOUS return score def url_score(self, indicator: str, raw_response: dict) -> int: """Determines indicator score by popularity preferred vendors and threshold. Args: indicator: The indicator we analyzing. raw_response: The raw response from API. Returns: DBotScore of the indicator. Can by Common.DBotScore.BAD, Common.DBotScore.SUSPICIOUS or Common.DBotScore.GOOD """ data = raw_response.get("data", {}) attributes = data.get("attributes", {}) # GTI assessment if self.is_malicious_by_gti(attributes.get("gti_assessment", {})): return Common.DBotScore.BAD score = self.score_by_results_and_stats(indicator, raw_response, self.url_threshold) if score == Common.DBotScore.GOOD and self.is_suspicious_by_gti(attributes.get("gti_assessment", {})): score = Common.DBotScore.SUSPICIOUS return score def domain_score(self, indicator: str, raw_response: dict) -> int: """Determines indicator score by popularity preferred vendors and threshold. Args: indicator: The indicator we analyzing. raw_response: The raw response from API. Returns: DBotScore of the indicator. Can by Common.DBotScore.BAD, Common.DBotScore.SUSPICIOUS or Common.DBotScore.GOOD """ data = raw_response.get("data", {}) attributes = data.get("attributes", {}) if self.is_malicious_by_gti(attributes.get("gti_assessment", {})): return Common.DBotScore.BAD score = self.score_by_results_and_stats(indicator, raw_response, self.domain_threshold) if score == Common.DBotScore.GOOD and self.is_suspicious_by_gti(attributes.get("gti_assessment", {})): score = Common.DBotScore.SUSPICIOUS return score def calculate_cve_dbot_score(self, cvss_score) -> int: """Calculates the DBotScore for a CVE based on its CVSS score. Args: cvss_score: The CVSS score of the CVE. Returns: DBotScore of the indicator. Can by Common.DBotScore.BAD, Common.DBotScore.SUSPICIOUS or Common.DBotScore.GOOD """ try: score = float(cvss_score) except ValueError: return Common.DBotScore.NONE if not score: return Common.DBotScore.NONE elif 0.0 <= score <= 3.9: return Common.DBotScore.GOOD elif 3.9 < score <= 7.9: return Common.DBotScore.SUSPICIOUS elif 7.9 < score <= 10.0: return Common.DBotScore.BAD else: return Common.DBotScore.NONE # endregion # region Helper functions def create_relationships(entity_a: str, entity_a_type: str, relationships_response: dict, reliability): """ Create a list of entityRelationship object from the api result entity_a: (str) - source of the relationship entity_a_type: (str) - type of the source of the relationship relationships_response: (dict) - the relationship response from the api reliability: The reliability of the source. Returns a list of EntityRelationship objects. """ relationships_list: List[EntityRelationship] = [] for relationship_type, relationship_type_raw in relationships_response.items(): relationships_data = relationship_type_raw.get("data", []) if relationships_data: if isinstance(relationships_data, dict): relationships_data = [relationships_data] for relation in relationships_data: name = RELATIONSHIP_TYPE.get(entity_a_type.lower(), {}).get(relationship_type) entity_b = relation.get("id", "") entity_b_type = INDICATOR_TYPE.get(relation.get("type", "").lower()) if entity_b and entity_b_type and name: if entity_b_type == FeedIndicatorType.URL: entity_b = dict_safe_get(relation, ["context_attributes", "url"]) relationships_list.append( EntityRelationship( entity_a=entity_a, entity_a_type=entity_a_type, name=name, entity_b=entity_b, entity_b_type=entity_b_type, source_reliability=reliability, brand=INTEGRATION_NAME, ) ) else: demisto.info( f"WARNING: Relationships will not be created to entity A {entity_a} with relationship name {name}" ) return relationships_list def create_relationships_cve( entity_a: str, entity_a_type: str, relationships_response: dict, reliability: DBotScoreReliability ) -> List[EntityRelationship]: """ Create relationships between CVE and related files (MD5 hashes from sources) Args: entity_a (str): The source of the relationship entity_a_type (str): The type of the source of the relationship relationships_response (dict): The relationship response from the api reliability (DBotScoreReliability): The reliability of the source. Returns: List[EntityRelationship]: List of EntityRelationship objects """ relationships_list: List[EntityRelationship] = [] # Extract sources from CVE response sources = relationships_response.get("data", {}).get("attributes", {}).get("sources", []) for source in sources: md5_hash = source.get("md5") if md5_hash: # Only create relationship if MD5 exists relationships_list.append( EntityRelationship( entity_a=entity_a, entity_a_type=entity_a_type, name=RELATIONSHIP_TYPE.get("cve", {}).get("referrer_cve"), entity_b=md5_hash, entity_b_type=FeedIndicatorType.File, source_reliability=reliability, brand=INTEGRATION_NAME, ) ) return relationships_list def arg_to_number_must_int(arg: Any, arg_name: Optional[str] = None, required: bool = False): """Wrapper of arg_to_number that must return int For mypy fixes. """ arg_num = arg_to_number(arg, arg_name, required) assert isinstance(arg_num, int) return arg_num def epoch_to_timestamp(epoch: Union[int, str]) -> Optional[str]: """Converts epoch timestamp to a string. Args: epoch: Time to convert Returns: A formatted string if succeeded. if not, returns None. """ try: return datetime.utcfromtimestamp(int(epoch)).strftime("%Y-%m-%d %H:%M:%SZ") except (TypeError, OSError, ValueError): return None def decrease_data_size(data: Union[dict, list]) -> Union[dict, list]: """Minifying data size. Args: data: the data object from raw response Returns: the same data without: data['attributes']['last_analysis_results'] data['attributes']['pe_info'] data['attributes']['crowdsourced_ids_results'] data['attributes']['autostart_locations'] data['attributes']['sandbox_verdicts'] data['attributes']['sigma_analysis_summary'] """ attributes_to_remove = [ "last_analysis_results", "pe_info", "crowdsourced_ids_results", "autostart_locations", "sandbox_verdicts", "sigma_analysis_summary", ] if isinstance(data, list): data = [decrease_data_size(item) for item in data] else: for attribute in attributes_to_remove: data["attributes"].pop(attribute, None) return data def _get_error_result(client: Client, ioc_id: str, ioc_type: str, message: str) -> CommandResults: dbot_type = ioc_type.upper() assert dbot_type in ("FILE", "DOMAIN", "IP", "URL", "CVE") common_type = dbot_type if dbot_type in ("IP", "URL", "CVE") else dbot_type.capitalize() desc = f'{common_type} "{ioc_id}" {message}' dbot = Common.DBotScore( ioc_id, getattr(DBotScoreType, dbot_type), INTEGRATION_NAME, Common.DBotScore.NONE, desc, client.reliability ) options: dict[str, Common.DBotScore | str] = {"dbot_score": dbot} if dbot_type == "FILE": if (hash_type := get_hash_type(ioc_id)) != "Unknown": options[hash_type] = ioc_id elif dbot_type == "CVE": options.update({"id": ioc_id, "cvss": "0.0", "published": "", "modified": "", "description": desc}) else: options[dbot_type.lower()] = ioc_id return CommandResults(indicator=getattr(Common, common_type)(**options), readable_output=desc) def build_unknown_output(client: Client, ioc_id: str, ioc_type: str) -> CommandResults: return _get_error_result(client, ioc_id, ioc_type, "was not found in GoogleThreatIntelligence.") def build_quota_exceeded_output(client: Client, ioc_id: str, ioc_type: str) -> CommandResults: return _get_error_result(client, ioc_id, ioc_type, "was not enriched. Quota was exceeded.") def build_error_output(client: Client, ioc_id: str, ioc_type: str, error_msg: str = None) -> CommandResults: msg = "could not be processed." if error_msg: msg += f" Error: {error_msg}" return _get_error_result(client, ioc_id, ioc_type, msg) def build_unknown_file_output(client: Client, file: str) -> CommandResults: return build_unknown_output(client, file, "file") def build_quota_exceeded_file_output(client: Client, file: str) -> CommandResults: return build_quota_exceeded_output(client, file, "file") def build_error_file_output(client: Client, file: str, error_msg: str = None) -> CommandResults: return build_error_output(client, file, "file", error_msg) def build_unknown_domain_output(client: Client, domain: str) -> CommandResults: return build_unknown_output(client, domain, "domain") def build_quota_exceeded_domain_output(client: Client, domain: str) -> CommandResults: return build_quota_exceeded_output(client, domain, "domain") def build_error_domain_output(client: Client, domain: str, error_msg: str = None) -> CommandResults: return build_error_output(client, domain, "domain", error_msg) def build_unknown_url_output(client: Client, url: str) -> CommandResults: return build_unknown_output(client, url, "url") def build_quota_exceeded_url_output(client: Client, url: str) -> CommandResults: return build_quota_exceeded_output(client, url, "url") def build_error_url_output(client: Client, url: str, error_msg: str = None) -> CommandResults: return build_error_output(client, url, "url", error_msg) def build_unknown_ip_output(client: Client, ip: str) -> CommandResults: return build_unknown_output(client, ip, "ip") def build_quota_exceeded_ip_output(client: Client, ip: str) -> CommandResults: return build_quota_exceeded_output(client, ip, "ip") def build_error_ip_output(client: Client, ip: str, error_msg: str = None) -> CommandResults: return build_error_output(client, ip, "ip", error_msg) def build_skipped_enrichment_ip_output(client: Client, ip: str) -> CommandResults: return _get_error_result( client, ip, "ip", "was not enriched. Reputation lookups have been disabled for private IP addresses." ) def build_unknown_cve_output(client: Client, cve_id: str) -> CommandResults: return build_unknown_output(client, cve_id, "cve") def build_quota_exceeded_cve_output(client: Client, cve_id: str) -> CommandResults: return build_quota_exceeded_output(client, cve_id, "cve") def build_error_cve_output(client: Client, cve_id: str, error_msg: str = None) -> CommandResults: return build_error_output(client, cve_id, "cve", error_msg) def _get_domain_indicator(client: Client, score_calculator: ScoreCalculator, domain: str, raw_response: dict): data = raw_response.get("data", {}) attributes = data.get("attributes", {}) last_analysis_stats = attributes.get("last_analysis_stats", {}) detection_engines = sum(last_analysis_stats.values()) positive_detections = last_analysis_stats.get("malicious", 0) whois = get_whois(attributes.get("whois", "")) relationships_response = data.get("relationships", {}) relationships_list = create_relationships( entity_a=domain, entity_a_type=FeedIndicatorType.Domain, relationships_response=relationships_response, reliability=client.reliability, ) score = score_calculator.domain_score(domain, raw_response) logs = score_calculator.get_logs() demisto.debug(logs) return Common.Domain( domain=domain, name_servers=whois["Name Server"], creation_date=whois["Creation Date"], updated_date=whois["Updated Date"], expiration_date=whois["Registry Expiry Date"], admin_name=whois["Admin Organization"], admin_email=whois["Admin Email"], admin_country=whois["Admin Country"], registrant_email=whois["Registrant Email"], registrant_country=whois["Registrant Country"], registrar_name=whois["Registrar"], registrar_abuse_email=whois["Registrar Abuse Contact Email"], registrar_abuse_phone=whois["Registrar Abuse Contact Phone"], detection_engines=detection_engines, positive_detections=positive_detections, dbot_score=Common.DBotScore( domain, DBotScoreType.DOMAIN, INTEGRATION_NAME, score=score, malicious_description=logs, reliability=client.reliability, ), relationships=relationships_list, ) def _get_url_indicator(client: Client, score_calculator: ScoreCalculator, url: str, raw_response: dict): data = raw_response.get("data", {}) attributes = data.get("attributes", {}) last_analysis_stats = attributes.get("last_analysis_stats", {}) detection_engines = sum(last_analysis_stats.values()) positive_detections = last_analysis_stats.get("malicious", 0) relationships_response = data.get("relationships", {}) relationships_list = create_relationships( entity_a=url, entity_a_type=FeedIndicatorType.URL, relationships_response=relationships_response, reliability=client.reliability, ) score = score_calculator.url_score(url, raw_response) logs = score_calculator.get_logs() demisto.debug(logs) return Common.URL( url, category=attributes.get("categories"), detection_engines=detection_engines, positive_detections=positive_detections, relationships=relationships_list, dbot_score=Common.DBotScore( url, DBotScoreType.URL, INTEGRATION_NAME, score=score, reliability=client.reliability, malicious_description=logs ), ) def _get_ip_indicator(client: Client, score_calculator: ScoreCalculator, ip: str, raw_response: dict): data = raw_response.get("data", {}) attributes = data.get("attributes", {}) last_analysis_stats = attributes.get("last_analysis_stats", {}) detection_engines = sum(last_analysis_stats.values()) positive_engines = last_analysis_stats.get("malicious", 0) relationships_response = data.get("relationships", {}) relationships_list = create_relationships( entity_a=ip, entity_a_type=FeedIndicatorType.IP, relationships_response=relationships_response, reliability=client.reliability, ) score = score_calculator.ip_score(ip, raw_response) logs = score_calculator.get_logs() demisto.debug(logs) return Common.IP( ip, asn=attributes.get("asn"), geo_country=attributes.get("country"), detection_engines=detection_engines, positive_engines=positive_engines, as_owner=attributes.get("as_owner"), relationships=relationships_list, dbot_score=Common.DBotScore( ip, DBotScoreType.IP, INTEGRATION_NAME, score=score, malicious_description=logs, reliability=client.reliability ), ) def _get_file_indicator(client: Client, score_calculator: ScoreCalculator, file_hash: str, raw_response: dict): data = raw_response.get("data", {}) attributes = data.get("attributes", {}) exiftool = attributes.get("exiftool", {}) signature_info = attributes.get("signature_info", {}) score = score_calculator.file_score(file_hash, raw_response) logs = score_calculator.get_logs() demisto.debug(logs) relationships_response = data.get("relationships", {}) relationships_list = create_relationships( entity_a=file_hash, entity_a_type=FeedIndicatorType.File, relationships_response=relationships_response, reliability=client.reliability, ) return Common.File( dbot_score=Common.DBotScore( file_hash, DBotScoreType.FILE, integration_name=INTEGRATION_NAME, score=score, malicious_description=logs, reliability=client.reliability, ), name=exiftool.get("OriginalFileName"), size=attributes.get("size"), sha1=attributes.get("sha1"), sha256=attributes.get("sha256"), file_type=exiftool.get("MIMEType"), md5=attributes.get("md5"), ssdeep=attributes.get("ssdeep"), extension=exiftool.get("FileTypeExtension"), company=exiftool.get("CompanyName"), product_name=exiftool.get("ProductName"), tags=attributes.get("tags"), signature=Common.FileSignature( authentihash=attributes.get("authentihash"), copyright=signature_info.get("copyright"), file_version=signature_info.get("file version"), description=signature_info.get("description"), internal_name=signature_info.get("internal name"), original_name=signature_info.get("original name"), ), relationships=relationships_list, ) def _extract_cve_data(raw_response: dict) -> dict: """ Extract and process CVE data from API response. Args: raw_response: Raw API response from CVE endpoint Returns: Dictionary containing processed CVE data """ data = raw_response.get("data", {}) attributes = data.get("attributes", {}) # Extract CVE-specific information description = attributes.get("description", "") executive_summary = attributes.get("executive_summary", "") # CVSS scores cvss = attributes.get("cvss", {}) cvss_v3 = cvss.get("cvssv3_x", {}) cvss_v4 = cvss.get("cvssv4_x", {}) cvss_v3_score = cvss_v3.get("base_score", 0) if cvss_v3 else 0 cvss_v4_score = cvss_v4.get("score", 0) if cvss_v4 else 0 cvss_v3_vector = cvss_v3.get("vector", "") if cvss_v3 else "" cvss_v4_vector = cvss_v4.get("vector", "") if cvss_v4 else "" # Risk and exploitation information risk_rating = attributes.get("risk_rating", "") exploitation_state = attributes.get("exploitation_state", "") exploit_availability = attributes.get("exploit_availability", "") priority = attributes.get("priority", "") # Process dates creation_date = attributes.get("creation_date") date_of_disclosure = attributes.get("date_of_disclosure") last_modification_date = attributes.get("last_modification_date") if creation_date: creation_date = timestamp_to_datestring(creation_date * 1000) if last_modification_date: last_modification_date = timestamp_to_datestring(last_modification_date * 1000) if date_of_disclosure: date_of_disclosure = timestamp_to_datestring(date_of_disclosure * 1000) # Sources and counters sources = attributes.get("sources", []) tags = attributes.get("tags", []) counters = attributes.get("counters", {}) extracted_data = { "description": description, "executive_summary": executive_summary, "cvss_v3_score": cvss_v3_score, "cvss_v4_score": cvss_v4_score, "cvss_v3_vector": cvss_v3_vector, "cvss_v4_vector": cvss_v4_vector, "risk_rating": risk_rating, "exploitation_state": exploitation_state, "exploit_availability": exploit_availability, "priority": priority, "creation_date": creation_date, "date_of_disclosure": date_of_disclosure, "last_modification_date": last_modification_date, "sources_count": len(sources), "tags": tags, "files_count": counters.get("files"), "domains_count": counters.get("domains"), "ip_addresses_count": counters.get("ip_addresses"), "urls_count": counters.get("urls"), } return extracted_data def _create_cve_indicator( client: Client, score_calculator: ScoreCalculator, cve_id: str, cve_data: dict, raw_response: dict ) -> tuple[Common.CVE, list]: """ Create CVE indicator with relationships and DBot score. Args: client: Client instance score_calculator: Score calculator instance cve_id: CVE identifier cve_data: Processed CVE data from _extract_cve_data raw_response: Raw API response Returns: Tuple of (CVE indicator, relationships list) """ # Calculate DBot score cvss_score = cve_data.get("cvss_v4_score") if cve_data.get("cvss_v4_score") else cve_data.get("cvss_v3_score") score = score_calculator.calculate_cve_dbot_score(cvss_score) dbot_score = Common.DBotScore( indicator=cve_id, indicator_type=DBotScoreType.CVE, integration_name=INTEGRATION_NAME, score=score, reliability=client.reliability, ) # Create relationships relationships = create_relationships_cve( entity_a=cve_id, entity_a_type=FeedIndicatorType.CVE, relationships_response=raw_response, reliability=client.reliability ) # Create CVE indicator cve_indicator = Common.CVE( id=cve_id, cvss=str(cvss_score) if cvss_score else "0.0", published=cve_data.get("date_of_disclosure", ""), modified=cve_data.get("last_modification_date", ""), description=cve_data.get("description", "No description available"), cvss_vector=cve_data.get("cvss_v4_vector", "") if cve_data.get("cvss_v4_vector", "") else cve_data.get("cvss_v3_vector", ""), dbot_score=dbot_score, relationships=relationships, tags=cve_data.get("tags", ""), stix_id=cve_id, ) return cve_indicator, relationships def build_domain_output( client: Client, score_calculator: ScoreCalculator, domain: str, raw_response: dict, extended_data: bool ) -> CommandResults: data = raw_response.get("data", {}) attributes = data.get("attributes", {}) last_analysis_stats = attributes.get("last_analysis_stats", {}) positive_engines = last_analysis_stats.get("malicious", 0) detection_engines = sum(last_analysis_stats.values()) whois = get_whois(attributes.get("whois", "")) relationships_response = data.get("relationships", {}) relationships_list = create_relationships( entity_a=domain, entity_a_type=FeedIndicatorType.Domain, relationships_response=relationships_response, reliability=client.reliability, ) domain_indicator = _get_domain_indicator(client, score_calculator, domain, raw_response) if not extended_data: data = decrease_data_size(data) return CommandResults( outputs_prefix=f"{INTEGRATION_ENTRY_CONTEXT}.Domain", outputs_key_field="id", indicator=domain_indicator, readable_output=tableToMarkdown( f"Domain data of {domain}", { **data, **attributes, **whois, "last_modified": epoch_to_timestamp(attributes.get("last_modification_date")), "positives": f"{positive_engines}/{detection_engines}", "gti_threat_score": attributes.get("gti_assessment", {}).get("threat_score", {}).get("value"), "gti_severity": attributes.get("gti_assessment", {}).get("severity", {}).get("value"), "gti_verdict": attributes.get("gti_assessment", {}).get("verdict", {}).get("value"), }, headers=[ "id", "Registrant Country", "Registrar", "last_modified", "reputation", "positives", "gti_threat_score", "gti_severity", "gti_verdict", ], removeNull=True, headerTransform=string_to_table_header, ), outputs=data, raw_response=raw_response, relationships=relationships_list, ) def build_url_output( client: Client, score_calculator: ScoreCalculator, url: str, raw_response: dict, extended_data: bool ) -> CommandResults: data = raw_response.get("data", {}) attributes = data.get("attributes", {}) last_analysis_stats = attributes.get("last_analysis_stats", {}) positive_detections = last_analysis_stats.get("malicious", 0) detection_engines = sum(last_analysis_stats.values()) relationships_response = data.get("relationships", {}) relationships_list = create_relationships( entity_a=url, entity_a_type=FeedIndicatorType.URL, relationships_response=relationships_response, reliability=client.reliability, ) url_indicator = _get_url_indicator(client, score_calculator, url, raw_response) if not extended_data: data = decrease_data_size(data) return CommandResults( outputs_prefix=f"{INTEGRATION_ENTRY_CONTEXT}.URL", outputs_key_field="id", indicator=url_indicator, readable_output=tableToMarkdown( f'URL data of "{url}"', { **data, **attributes, "url": url, "last_modified": epoch_to_timestamp(attributes.get("last_modification_date")), "positives": f"{positive_detections}/{detection_engines}", "gti_threat_score": attributes.get("gti_assessment", {}).get("threat_score", {}).get("value"), "gti_severity": attributes.get("gti_assessment", {}).get("severity", {}).get("value"), "gti_verdict": attributes.get("gti_assessment", {}).get("verdict", {}).get("value"), }, headers=[ "url", "title", "has_content", "last_http_response_content_sha256", "last_modified", "reputation", "positives", "gti_threat_score", "gti_severity", "gti_verdict", ], removeNull=True, headerTransform=string_to_table_header, ), outputs=data, raw_response=raw_response, relationships=relationships_list, ) def build_private_url_output(url: str, raw_response: dict) -> CommandResults: data = raw_response.get("data", {}) attributes = data.get("attributes", {}) last_analysis_stats = attributes.get("last_analysis_stats", {}) positive_detections = last_analysis_stats.get("malicious", 0) detection_engines = sum(last_analysis_stats.values()) return CommandResults( outputs_prefix=f"{INTEGRATION_ENTRY_CONTEXT}.URL", outputs_key_field="id", readable_output=tableToMarkdown( f'URL data of "{url}"', { **attributes, "positives": f"{positive_detections}/{detection_engines}", }, headers=[ "url", "title", "last_http_response_content_sha256", "positives", ], removeNull=True, headerTransform=string_to_table_header, ), outputs=data, raw_response=raw_response, ) def build_ip_output( client: Client, score_calculator: ScoreCalculator, ip: str, raw_response: dict, extended_data: bool ) -> CommandResults: data = raw_response.get("data", {}) attributes = data.get("attributes", {}) last_analysis_stats = attributes.get("last_analysis_stats", {}) positive_engines = last_analysis_stats.get("malicious", 0) detection_engines = sum(last_analysis_stats.values()) relationships_response = data.get("relationships", {}) relationships_list = create_relationships( entity_a=ip, entity_a_type=FeedIndicatorType.IP, relationships_response=relationships_response, reliability=client.reliability, ) ip_indicator = _get_ip_indicator(client, score_calculator, ip, raw_response) if not extended_data: data = decrease_data_size(data) return CommandResults( outputs_prefix=f"{INTEGRATION_ENTRY_CONTEXT}.IP", outputs_key_field="id", indicator=ip_indicator, readable_output=tableToMarkdown( f"IP reputation of {ip}:", { **data, **attributes, "last_modified": epoch_to_timestamp(attributes.get("last_modification_date")), "positives": f"{positive_engines}/{detection_engines}", "gti_threat_score": attributes.get("gti_assessment", {}).get("threat_score", {}).get("value"), "gti_severity": attributes.get("gti_assessment", {}).get("severity", {}).get("value"), "gti_verdict": attributes.get("gti_assessment", {}).get("verdict", {}).get("value"), }, headers=[ "id", "network", "country", "as_owner", "last_modified", "reputation", "positives", "gti_threat_score", "gti_severity", "gti_verdict", ], headerTransform=string_to_table_header, ), outputs=data, raw_response=raw_response, relationships=relationships_list, ) def build_file_output( client: Client, score_calculator: ScoreCalculator, file_hash: str, raw_response: dict, extended_data: bool ) -> CommandResults: data = raw_response.get("data", {}) attributes = data.get("attributes", {}) last_analysis_stats = attributes.get("last_analysis_stats", {}) malicious = last_analysis_stats.get("malicious", 0) total = sum(last_analysis_stats.values()) relationships_response = data.get("relationships", {}) relationships_list = create_relationships( entity_a=file_hash, entity_a_type=FeedIndicatorType.File, relationships_response=relationships_response, reliability=client.reliability, ) file_indicator = _get_file_indicator(client, score_calculator, file_hash, raw_response) if not extended_data: data = decrease_data_size(data) return CommandResults( outputs_prefix=f"{INTEGRATION_ENTRY_CONTEXT}.File", outputs_key_field="id", indicator=file_indicator, readable_output=tableToMarkdown( f"Results of file hash {file_hash}", { **data, **attributes, "positives": f"{malicious}/{total}", "creation_date": epoch_to_timestamp(attributes.get("creation_date")), "last_modified": epoch_to_timestamp(attributes.get("last_modification_date", 0)), "gti_threat_score": attributes.get("gti_assessment", {}).get("threat_score", {}).get("value"), "gti_severity": attributes.get("gti_assessment", {}).get("severity", {}).get("value"), "gti_verdict": attributes.get("gti_assessment", {}).get("verdict", {}).get("value"), }, headers=[ "sha1", "sha256", "md5", "meaningful_name", "type_extension", "creation_date", "last_modified", "reputation", "positives", "gti_threat_score", "gti_severity", "gti_verdict", ], removeNull=True, headerTransform=string_to_table_header, ), outputs=data, raw_response=raw_response, relationships=relationships_list, ) def build_cve_output(client: Client, score_calculator: ScoreCalculator, cve_id: str, raw_response: dict) -> CommandResults: """ Build CommandResults for CVE data from collections endpoint. Args: client: Client instance score_calculator: Score calculator instance cve_id: CVE identifier raw_response: Raw API response Returns: CommandResults with CVE information """ # Extract and process CVE data cve_data = _extract_cve_data(raw_response) # Create CVE indicator and relationships cve_indicator, relationships = _create_cve_indicator(client, score_calculator, cve_id, cve_data, raw_response) context_data = remove_empty_elements(raw_response.get("data", {})) # Prepare human-readable output hr_data = { "CVE ID": cve_id, "Description": cve_data.get("description", ""), "Executive Summary": cve_data.get("executive_summary", ""), "Risk Rating": cve_data.get("risk_rating", ""), "Priority": cve_data.get("priority", ""), "CVSS v3.x Score": cve_data.get("cvss_v3_score"), "CVSS v3.x Vector": cve_data.get("cvss_v3_vector"), "CVSS v4.x Score": cve_data.get("cvss_v4_score"), "CVSS v4.x Vector": cve_data.get("cvss_v4_vector"), "Exploitation State": cve_data.get("exploitation_state"), "Exploit Availability": cve_data.get("exploit_availability"), "Date Of Disclosure": cve_data.get("date_of_disclosure", ""), "Creation Date": cve_data.get("creation_date", ""), "Last Modified": cve_data.get("last_modification_date", ""), "Sources": cve_data.get("sources_count"), "Related Files": cve_data.get("files_count"), "Related Domains": cve_data.get("domains_count"), "Related IPs": cve_data.get("ip_addresses_count"), "Related URLs": cve_data.get("urls_count"), } hr_for_cve = tableToMarkdown( f"CVE Information: {cve_id}", hr_data, removeNull=True, headers=[ "CVE ID", "Risk Rating", "Priority", "Exploitation State", "Exploit Availability", "CVSS v3.x Score", "CVSS v4.x Score", "CVSS v3.x Vector", "CVSS v4.x Vector", "Date Of Disclosure", "Creation Date", "Last Modified", "Sources", "Description", "Related Files", "Related Domains", "Related IPs", "Related URLs", "Executive Summary", ], ) return CommandResults( outputs_prefix=f"{INTEGRATION_ENTRY_CONTEXT}.CVE", outputs_key_field="id", indicator=cve_indicator, readable_output=hr_for_cve, outputs=context_data, raw_response=raw_response, relationships=relationships, ) def build_private_file_output(file_hash: str, raw_response: dict) -> CommandResults: data = raw_response.get("data", {}) attributes = data.get("attributes", {}) threat_severity = attributes.get("threat_severity", {}) threat_severity_level = threat_severity.get("threat_severity_level", "") threat_severity_data = threat_severity.get("threat_severity_data", {}) popular_threat_category = threat_severity_data.get("popular_threat_category", "") threat_verdict = attributes.get("threat_verdict", "") return CommandResults( outputs_prefix=f"{INTEGRATION_ENTRY_CONTEXT}.File", outputs_key_field="id", readable_output=tableToMarkdown( f"Results of file hash {file_hash}", { **attributes, "threat_severity_level": SEVERITY_LEVELS.get(threat_severity_level, threat_severity_level), "popular_threat_category": popular_threat_category, "threat_verdict": VERDICTS.get(threat_verdict, threat_verdict), }, headers=[ "sha1", "sha256", "md5", "meaningful_name", "type_extension", "threat_severity_level", "popular_threat_category", "threat_verdict", ], removeNull=True, headerTransform=string_to_table_header, ), outputs=data, raw_response=raw_response, ) def get_whois(whois_string: str) -> defaultdict: """Gets a WHOIS string and returns a parsed dict of the WHOIS String. Args: whois_string: whois from domain api call Returns: A parsed whois Examples: >>> get_whois('key1:value\\nkey2:value2') defaultdict({'key1': 'value', 'key2': 'value2'}) """ whois: defaultdict = defaultdict(lambda: None) for line in whois_string.splitlines(): key: str value: str try: key, value = line.split(sep=":", maxsplit=1) except ValueError: demisto.debug(f"Could not unpack Whois string: {line}. Skipping") continue key = key.strip() value = value.strip() if key in whois: if not isinstance(whois[key], list): whois[key] = [whois[key]] whois[key].append(value) else: whois[key] = value return whois def get_file_context(entry_id: str) -> dict: """Gets a File object from context. Args: entry_id: The entry ID of the file Returns: File object contains Name, Hashes and more information """ context = demisto.dt(demisto.context(), f'File(val.EntryID === "{entry_id}")') if not context: return {} if isinstance(context, list): return context[0] return context def validate_cve_values(cve_ids: list[str]) -> tuple[list[str], list[str]]: """ Validate CVE format and return valid/invalid CVE lists. Args: cve_ids: List of CVE identifiers to validate Returns: Tuple of (valid_cves, invalid_cves) """ valid_cves = [] invalid_cves = [] # CVE format: CVE-YYYY-NNNNN (where YYYY is year, NNNNN is 4+ digits) cve_pattern = re.compile(r"^CVE-\d{4}-\d{4,}$", re.IGNORECASE) for cve_id in cve_ids: normalized_cve = cve_id.upper().strip() if cve_pattern.match(normalized_cve): valid_cves.append(normalized_cve) else: invalid_cves.append(cve_id) return valid_cves, invalid_cves def raise_if_ip_not_valid(ip: str): """Raises an error if ip is not valid Args: ip: ip address Raises: ValueError: If IP is not valid Examples: >>> raise_if_ip_not_valid('not ip at all') Traceback (most recent call last): ... ValueError: IP "not ip at all" is not valid >>> raise_if_ip_not_valid('8.8.8.8') """ if not is_ip_valid(ip, accept_v6_ips=True): raise ValueError(f'IP "{ip}" is not valid') def raise_if_hash_not_valid(file_hash: str): """Raises an error if file_hash is not valid Args: file_hash: file hash Raises: ValueError: if hash is not of type SHA-256, SHA-1 or MD5 Examples: >>> raise_if_hash_not_valid('not a hash') Traceback (most recent call last): ... ValueError: Hash "not a hash" is not of type SHA-256, SHA-1 or MD5 >>> raise_if_hash_not_valid('7e641f6b9706d860baf09fe418b6cc87') """ if get_hash_type(file_hash) not in ("sha256", "sha1", "md5"): raise ValueError(f'Hash "{file_hash}" is not of type SHA-256, SHA-1 or MD5') def encode_url_to_base64(url: str) -> str: """Gets a string (in this case, url but it can not be) and return it as base64 without padding ('=') Args: url: A string to encode Returns: Base64 encoded string with no padding Examples: >>> encode_url_to_base64('https://example.com') 'aHR0cHM6Ly9leGFtcGxlLmNvbQ' """ return base64.urlsafe_b64encode(url.encode()).decode().strip("=") # endregion # region Reputation commands def ip_command( client: Client, score_calculator: ScoreCalculator, args: dict, relationships: str, disable_private_ip_lookup: bool ) -> List[CommandResults]: """ 1 API Call for regular """ ips = argToList(args["ip"]) results: List[CommandResults] = [] execution_metrics = ExecutionMetrics() override_private_lookup = argToBoolean(args.get("override_private_lookup", False)) for ip in ips: raise_if_ip_not_valid(ip) if disable_private_ip_lookup and ipaddress.ip_address(ip).is_private and not override_private_lookup: results.append(build_skipped_enrichment_ip_output(client, ip)) execution_metrics.success += 1 continue try: raw_response = client.ip(ip, relationships) if raw_response.get("error", {}).get("code") == "QuotaExceededError": execution_metrics.quota_error += 1 results.append(build_quota_exceeded_ip_output(client, ip)) continue if raw_response.get("error", {}).get("code") == "NotFoundError": results.append(build_unknown_ip_output(client, ip)) continue except Exception as exc: # If anything happens, just keep going demisto.debug(f'Could not process IP: "{ip}"\n {exc!s}') execution_metrics.general_error += 1 results.append(build_error_ip_output(client, ip, str(exc))) continue execution_metrics.success += 1 results.append( build_ip_output(client, score_calculator, ip, raw_response, argToBoolean(args.get("extended_data", False))) ) if execution_metrics.is_supported(): _metric_results = execution_metrics.metrics metric_results = cast(CommandResults, _metric_results) results.append(metric_results) return results def file_command(client: Client, score_calculator: ScoreCalculator, args: dict, relationships: str) -> List[CommandResults]: """ 1 API Call """ files = argToList(args["file"]) extended_data = argToBoolean(args.get("extended_data", False)) results: List[CommandResults] = [] execution_metrics = ExecutionMetrics() for file in files: try: raise_if_hash_not_valid(file) raw_response = client.file(file, relationships) if raw_response.get("error", {}).get("code") == "QuotaExceededError": execution_metrics.quota_error += 1 results.append(build_quota_exceeded_file_output(client, file)) continue if raw_response.get("error", {}).get("code") == "NotFoundError": results.append(build_unknown_file_output(client, file)) continue results.append(build_file_output(client, score_calculator, file, raw_response, extended_data)) execution_metrics.success += 1 except Exception as exc: # If anything happens, just keep going demisto.debug(f'Could not process file: "{file}"\n {exc!s}') execution_metrics.general_error += 1 results.append(build_error_file_output(client, file, str(exc))) continue if execution_metrics.is_supported(): _metric_results = execution_metrics.metrics metric_results = cast(CommandResults, _metric_results) results.append(metric_results) return results def private_file_command(client: Client, args: dict) -> List[CommandResults]: """ 1 API Call """ files = argToList(args["file"]) results: List[CommandResults] = [] execution_metrics = ExecutionMetrics() for file in files: try: raise_if_hash_not_valid(file) raw_response = client.private_file(file) if raw_response.get("error", {}).get("code") == "QuotaExceededError": execution_metrics.quota_error += 1 results.append(build_quota_exceeded_file_output(client, file)) continue if raw_response.get("error", {}).get("code") == "NotFoundError": results.append(build_unknown_file_output(client, file)) continue results.append(build_private_file_output(file, raw_response)) execution_metrics.success += 1 except Exception as exc: # If anything happens, just keep going demisto.debug(f'Could not process private file: "{file}"\n {exc!s}') execution_metrics.general_error += 1 results.append(build_error_file_output(client, file, str(exc))) continue if execution_metrics.is_supported(): _metric_results = execution_metrics.metrics metric_results = cast(CommandResults, _metric_results) results.append(metric_results) return results def url_command(client: Client, score_calculator: ScoreCalculator, args: dict, relationships: str) -> List[CommandResults]: """ 1 API Call for regular """ urls = argToList(args["url"]) extended_data = argToBoolean(args.get("extended_data", False)) results: List[CommandResults] = [] execution_metrics = ExecutionMetrics() for url in urls: try: raw_response = client.url(url, relationships) if raw_response.get("error", {}).get("code") == "QuotaExceededError": execution_metrics.quota_error += 1 results.append(build_quota_exceeded_url_output(client, url)) continue if raw_response.get("error", {}).get("code") == "NotFoundError": results.append(build_unknown_url_output(client, url)) continue except Exception as exc: # If anything happens, just keep going demisto.debug(f'Could not process URL: "{url}".\n {exc!s}') execution_metrics.general_error += 1 results.append(build_error_url_output(client, url, str(exc))) continue execution_metrics.success += 1 results.append(build_url_output(client, score_calculator, url, raw_response, extended_data)) if execution_metrics.is_supported(): _metric_results = execution_metrics.metrics metric_results = cast(CommandResults, _metric_results) results.append(metric_results) return results def cve_command(client: Client, score_calculator: ScoreCalculator, args: dict) -> List[CommandResults]: """ Get CVE information from Google Threat Intelligence collections endpoint. Args: client: Client instance args: Command arguments containing 'cve' parameter Returns: List of CommandResults with CVE information """ cve_ids = argToList(args.get("cve", [])) cve_ids = [cve_id.strip() for cve_id in cve_ids if cve_id.strip()] results: List[CommandResults] = [] execution_metrics = ExecutionMetrics() # Validate CVE format (CVE-YYYY-NNNNN) valid_cves, invalid_cves = validate_cve_values(cve_ids) if invalid_cves: return_warning( "The following CVEs were found invalid: {}".format(", ".join(invalid_cves)), exit=len(invalid_cves) == len(cve_ids) ) for cve_id in valid_cves: # Normalize CVE ID to uppercase cve_id = cve_id.upper().strip() try: raw_response = client.cve(cve_id) if raw_response.get("error", {}).get("code") == "QuotaExceededError": execution_metrics.quota_error += 1 results.append(build_quota_exceeded_cve_output(client, cve_id)) continue if raw_response.get("error", {}).get("code") == "NotFoundError": results.append(build_unknown_cve_output(client, cve_id)) continue except Exception as exc: # If anything happens, just keep going demisto.debug(f'Could not process CVE: "{cve_id}".\n {exc!s}') execution_metrics.general_error += 1 results.append(build_error_cve_output(client, cve_id, str(exc))) continue execution_metrics.success += 1 results.append(build_cve_output(client, score_calculator, cve_id, raw_response)) if execution_metrics.is_supported(): _metric_results = execution_metrics.metrics metric_results = cast(CommandResults, _metric_results) results.append(metric_results) return results def private_url_command(client: Client, args: dict) -> List[CommandResults]: """ 1 API Call """ urls = argToList(args["url"]) results: List[CommandResults] = [] execution_metrics = ExecutionMetrics() for url in urls: try: raw_response = client.private_url(url) if raw_response.get("error", {}).get("code") == "QuotaExceededError": execution_metrics.quota_error += 1 results.append(build_quota_exceeded_url_output(client, url)) continue if raw_response.get("error", {}).get("code") == "NotFoundError": results.append(build_unknown_url_output(client, url)) continue except Exception as exc: # If anything happens, just keep going demisto.debug(f'Could not process private URL: "{url}".\n {exc!s}') execution_metrics.general_error += 1 results.append(build_error_url_output(client, url, str(exc))) continue execution_metrics.success += 1 results.append(build_private_url_output(url, raw_response)) if execution_metrics.is_supported(): _metric_results = execution_metrics.metrics metric_results = cast(CommandResults, _metric_results) results.append(metric_results) return results def domain_command(client: Client, score_calculator: ScoreCalculator, args: dict, relationships: str) -> List[CommandResults]: """ 1 API Call for regular """ execution_metrics = ExecutionMetrics() domains = argToList(args["domain"]) results: List[CommandResults] = [] for domain in domains: try: raw_response = client.domain(domain, relationships) if raw_response.get("error", {}).get("code") == "QuotaExceededError": execution_metrics.quota_error += 1 results.append(build_quota_exceeded_domain_output(client, domain)) continue if raw_response.get("error", {}).get("code") == "NotFoundError": results.append(build_unknown_domain_output(client, domain)) continue except Exception as exc: # If anything happens, just keep going demisto.debug(f'Could not process domain: "{domain}"\n {exc!s}') execution_metrics.general_error += 1 results.append(build_error_domain_output(client, domain, str(exc))) continue execution_metrics.success += 1 result = build_domain_output( client, score_calculator, domain, raw_response, argToBoolean(args.get("extended_data", False)) ) results.append(result) if execution_metrics.is_supported(): _metric_results = execution_metrics.metrics metric_results = cast(CommandResults, _metric_results) results.append(metric_results) return results # endregion # region Scan commands def file_rescan_command(client: Client, args: dict) -> CommandResults: """ 1 API Call """ file_hash = args["file"] raise_if_hash_not_valid(file_hash) raw_response = client.file_rescan(file_hash) data = raw_response["data"] data["hash"] = file_hash context = { f"{INTEGRATION_ENTRY_CONTEXT}.Submission(val.id && val.id === obj.id)": data, "vtScanID": data.get("id"), # BC preservation } return CommandResults( readable_output=tableToMarkdown( f'File "{file_hash}" resubmitted.', data, removeNull=True, headerTransform=underscoreToCamelCase ), outputs=context, raw_response=raw_response, ) def get_working_id(id_: str, entry_id: str) -> str: """Sometimes new scanned files ID will be only a number. Should connect them with base64(MD5:_id). Fixes bug in Google Threat Intelligence API. Args: entry_id: the entry id connected to the file id_: id given from the API Returns: A working ID that we can use in other commands. """ if (isinstance(id_, str) and id_.isnumeric()) or isinstance(id_, int): demisto.debug(f"Got an integer id from file-scan. {id_=}, {entry_id=}\n") raise DemistoException( f"Got an int {id_=} as analysis report. This is a bug in Google Threat Intelligence API.\n" f"While Google Threat Intelligence team is fixing the problem, try to resend the file." ) return id_ def file_scan(client: Client, args: dict) -> List[CommandResults]: """ 1 API Call """ return upload_file(client, args) def private_file_scan(client: Client, args: dict) -> List[CommandResults]: """ 1 API Call """ return upload_file(client, args, True) def upload_file(client: Client, args: dict, private: bool = False) -> List[CommandResults]: """ 1 API Call """ entry_ids = argToList(args["entryID"]) upload_url = args.get("uploadURL") if len(entry_ids) > 1 and upload_url: raise DemistoException("You can supply only one entry ID with an upload URL.") results = [] for entry_id in entry_ids: try: file_obj = demisto.getFilePath(entry_id) file_path = file_obj["path"] if private: raw_response = client.private_file_scan(file_path) else: raw_response = client.file_scan(file_path, upload_url) data = raw_response.get("data", {}) # add current file as identifiers data.update(get_file_context(entry_id)) id_ = data.get("id") demisto.debug(f'Result from vt-scan-file {entry_id=} {id_=} {data.get("type")=}') id_ = get_working_id(id_, entry_id) data["id"] = id_ context = { f"{INTEGRATION_ENTRY_CONTEXT}.Submission(val.id && val.id === obj.id)": data, "vtScanID": id_, # BC preservation } results.append( CommandResults( readable_output=tableToMarkdown( f'The file has been submitted "{file_obj["name"]}"', data, headers=["id", "EntryID", "MD5", "SHA1", "SHA256"], removeNull=True, ), outputs=context, raw_response=raw_response, ) ) except Exception as exc: err = f"Could not process {entry_id=}.\n{exc!s}" demisto.debug(err) demisto.results({"Type": entryTypes["error"], "ContentsFormat": formats["text"], "Contents": err}) return results def file_scan_and_get_analysis(client: Client, score_calculator: ScoreCalculator, args: dict, file_relationships: str): """Calls to file-scan and gti-analysis-get.""" interval = int(args.get("interval_in_seconds", 60)) extended = argToBoolean(args.get("extended_data", False)) if not args.get("id"): command_results = file_scan(client, args) command_result = command_results[0] outputs = command_result.outputs if not isinstance(outputs, dict): raise DemistoException("outputs is expected to be a dict") scheduled_command = ScheduledCommand( command=f"{COMMAND_PREFIX}-file-scan-and-analysis-get", next_run_in_seconds=interval, args={ "entryID": args.get("entryID"), "id": outputs.get("vtScanID"), "file": outputs.get(f"{INTEGRATION_ENTRY_CONTEXT}.Submission(val.id && val.id === obj.id)", {}).get("SHA256"), "interval_in_seconds": interval, "extended_data": extended, }, timeout_in_seconds=6000, ) command_result.scheduled_command = scheduled_command return command_result command_result = get_analysis_command(client, args) outputs = command_result.outputs if not isinstance(outputs, dict): raise DemistoException("outputs is expected to be a dict") if outputs.get("data", {}).get("attributes", {}).get("status") == "completed": return file_command(client, score_calculator, args, file_relationships) scheduled_command = ScheduledCommand( command=f"{COMMAND_PREFIX}-file-scan-and-analysis-get", next_run_in_seconds=interval, args={ "entryID": args.get("entryID"), "id": outputs.get("id"), "file": args.get("file"), "interval_in_seconds": interval, "extended_data": extended, }, timeout_in_seconds=6000, ) return CommandResults(scheduled_command=scheduled_command) def private_file_scan_and_get_analysis(client: Client, args: dict): """Calls to gti-privatescanning-file-scan and gti-privatescanning-analysis-get.""" interval = int(args.get("interval_in_seconds", 60)) if not args.get("id"): command_results = private_file_scan(client, args) command_result = command_results[0] outputs = command_result.outputs if not isinstance(outputs, dict): raise DemistoException("outputs is expected to be a dict") scheduled_command = ScheduledCommand( command=f"{COMMAND_PREFIX}-private-file-scan-and-analysis-get", next_run_in_seconds=interval, args={ "entryID": args.get("entryID"), "id": outputs.get("vtScanID"), "interval_in_seconds": interval, }, timeout_in_seconds=6000, ) command_result.scheduled_command = scheduled_command return command_result command_result = private_get_analysis_command(client, args) outputs = command_result.outputs if not isinstance(outputs, dict): raise DemistoException("outputs is expected to be a dict") if outputs.get("data", {}).get("attributes", {}).get("status") == "completed": return command_result scheduled_command = ScheduledCommand( command=f"{COMMAND_PREFIX}-private-file-scan-and-analysis-get", next_run_in_seconds=interval, args={ "entryID": args.get("entryID"), "id": outputs.get("id"), "interval_in_seconds": interval, }, timeout_in_seconds=6000, ) return CommandResults(scheduled_command=scheduled_command) def url_scan_and_get_analysis(client: Client, score_calculator: ScoreCalculator, args: dict, url_relationships: str): """Calls to url-scan and gti-analysis-get.""" interval = int(args.get("interval_in_seconds", 60)) extended = argToBoolean(args.get("extended_data", False)) if not args.get("id"): command_result = scan_url_command(client, args) outputs = command_result.outputs if not isinstance(outputs, dict): raise DemistoException("outputs is expected to be a dict") scheduled_command = ScheduledCommand( command=f"{COMMAND_PREFIX}-url-scan-and-analysis-get", next_run_in_seconds=interval, args={ "url": args.get("url"), "id": outputs.get("vtScanID"), "interval_in_seconds": interval, "extended_data": extended, }, timeout_in_seconds=6000, ) command_result.scheduled_command = scheduled_command return command_result command_result = get_analysis_command(client, args) outputs = command_result.outputs if not isinstance(outputs, dict): raise DemistoException("outputs is expected to be a dict") if outputs.get("data", {}).get("attributes", {}).get("status") == "completed": return url_command(client, score_calculator, args, url_relationships) scheduled_command = ScheduledCommand( command=f"{COMMAND_PREFIX}-url-scan-and-analysis-get", next_run_in_seconds=interval, args={ "url": args.get("url"), "id": outputs.get("id"), "interval_in_seconds": interval, "extended_data": extended, }, timeout_in_seconds=6000, ) return CommandResults(scheduled_command=scheduled_command) def private_url_scan_and_get_analysis(client: Client, args: dict): """Calls to gti-privatescanning-url-scan and gti-privatescanning-analysis-get.""" interval = int(args.get("interval_in_seconds", 60)) if not args.get("id"): command_result = private_scan_url_command(client, args) outputs = command_result.outputs if not isinstance(outputs, dict): raise DemistoException("outputs is expected to be a dict") scheduled_command = ScheduledCommand( command=f"{COMMAND_PREFIX}-private-url-scan-and-analysis-get", next_run_in_seconds=interval, args={ "url": args.get("url"), "id": outputs.get("vtScanID"), "interval_in_seconds": interval, }, timeout_in_seconds=6000, ) command_result.scheduled_command = scheduled_command return command_result command_result = private_get_analysis_command(client, args) outputs = command_result.outputs if not isinstance(outputs, dict): raise DemistoException("outputs is expected to be a dict") if outputs.get("data", {}).get("attributes", {}).get("status") == "completed": return command_result scheduled_command = ScheduledCommand( command=f"{COMMAND_PREFIX}-private-url-scan-and-analysis-get", next_run_in_seconds=interval, args={ "url": args.get("url"), "id": outputs.get("id"), "interval_in_seconds": interval, }, timeout_in_seconds=6000, ) return CommandResults(scheduled_command=scheduled_command) def get_upload_url(client: Client) -> CommandResults: """ 1 API Call """ raw_response = client.get_upload_url() upload_url = raw_response["data"] context = { f"{INTEGRATION_ENTRY_CONTEXT}.FileUploadURL": upload_url, "vtUploadURL": upload_url, # BC preservation } return CommandResults( readable_output=tableToMarkdown("New upload url acquired!", {"Upload url": upload_url}), outputs=context, raw_response=raw_response, ) def scan_url_command(client: Client, args: dict) -> CommandResults: """ 1 API Call """ return scan_url(client, args) def private_scan_url_command(client: Client, args: dict) -> CommandResults: """ 1 API Call """ return scan_url(client, args, True) def scan_url(client: Client, args: dict, private: bool = False) -> CommandResults: """ 1 API Call """ url = args["url"] raw_response: Dict[str, Any] = {} data: Dict[str, Any] = {} context: Dict[str, Any] = {} headers = ["id", "url"] try: if private: raw_response = client.private_url_scan(url) else: raw_response = client.url_scan(url) data = raw_response["data"] data["url"] = url context = { f"{INTEGRATION_ENTRY_CONTEXT}.Submission(val.id && val.id === obj.id)": data, "vtScanID": data.get("id"), # BC preservation } except DemistoException as e: error = e.res.json().get("error") # Invalid url, probably due to an unknown TLD if error["code"] == "InvalidArgumentError": data = {"url": url, "id": "", "error": error["message"]} headers.append("error") else: raise e return CommandResults( readable_output=tableToMarkdown("New url submission:", data, headers=headers), outputs=context, raw_response=raw_response ) # endregion # region Comments commands def get_comments_command(client: Client, args: dict) -> CommandResults: """ 1 API Call BC Break - No NotBefore argument added limit """ limit = arg_to_number_must_int(args.get("limit"), arg_name="limit", required=True) resource = args["resource"] if before := args.get("before"): before = parse(before) assert before is not None, f'Could not parse the before date "{before}"' before = before.replace(tzinfo=None) resource_type = args.get("resource_type") if not resource_type: try: raise_if_hash_not_valid(resource) resource_type = "file" except ValueError: resource_type = "url" resource_type = resource_type.lower() # Will find if there's one and only one True in the list. if resource_type == "ip": raise_if_ip_not_valid(resource) raw_response = client.get_ip_comments(resource, limit) elif resource_type == "url": raw_response = client.get_url_comments(resource, limit) elif resource_type in ("hash", "file"): raise_if_hash_not_valid(resource) raw_response = client.get_hash_comments(resource, limit) elif resource_type == "domain": raw_response = client.get_domain_comments(resource, limit) else: raise DemistoException(f'Could not find resource type of "{resource_type}"') data = raw_response.get("data", {}) context = {"indicator": resource, "comments": data} comments = [] for comment in data: attributes = comment.get("attributes", {}) votes = attributes.get("votes", {}) if date := parse(str(attributes.get("date"))): date = date.replace(tzinfo=None) if date and before and date > before: continue comments.append( { "Date": epoch_to_timestamp(attributes.get("date")), "Text": attributes.get("text"), "Positive Votes": votes.get("positive"), "Abuse Votes": votes.get("abuse"), "Negative Votes": votes.get("negative"), } ) return CommandResults( f"{INTEGRATION_ENTRY_CONTEXT}.Comments", "id", readable_output=tableToMarkdown( f'Google Threat Intelligence comments of {resource_type}: "{resource}"', comments, headers=["Date", "Text", "Positive Votes", "Abuse Votes", "Negative Votes"], ), outputs=context, raw_response=raw_response, ) def add_comments_command(client: Client, args: dict) -> CommandResults: """ 1 API Call """ resource = args["resource"] comment = args["comment"] resource_type = args.get("resource_type") if not resource_type: try: raise_if_hash_not_valid(resource) resource_type = "file" except ValueError: resource_type = "url" resource_type = resource_type.lower() if resource_type == "ip": raise_if_ip_not_valid(resource) raw_response = client.add_comment_to_ip(resource, comment) elif resource_type == "url": raw_response = client.add_comment_to_url(resource, comment) elif resource_type == "domain": raw_response = client.add_comment_to_domain(resource, comment) elif resource_type == "file": raise_if_hash_not_valid(resource) raw_response = client.add_comment_to_file(resource, comment) else: raise DemistoException(f'Could not find resource type of "{resource_type}"') data = raw_response["data"] attributes = data.get("attributes", {}) votes = attributes.get("votes", {}) return CommandResults( f"{INTEGRATION_ENTRY_CONTEXT}.Comments.comments", "id", readable_output=tableToMarkdown( "Comment has been added!", { "Date": epoch_to_timestamp(attributes.get("date")), "Text": attributes.get("text"), "Positive Votes": votes.get("positive"), "Abuse Votes": votes.get("abuse"), "Negative Votes": votes.get("negative"), }, headers=["Date", "Text", "Positive Votes", "Abuse Votes", "Negative Votes"], ), outputs=data, raw_response=raw_response, ) def get_comments_by_id_command(client: Client, args: dict) -> CommandResults: """ 1 API Call """ comment_id = args["id"] raw_response = client.get_comment_by_id(comment_id) data = raw_response["data"] attributes = data.get("attributes", {}) votes = attributes.get("votes", {}) return CommandResults( f"{INTEGRATION_ENTRY_CONTEXT}.Comments.comments", "id", readable_output=tableToMarkdown( f"Comment of ID {comment_id}", { "Date": epoch_to_timestamp(attributes.get("date")), "Text": attributes.get("text"), "Positive Votes": votes.get("positive"), "Abuse Votes": votes.get("abuse"), "Negative Votes": votes.get("negative"), }, headers=["Date", "Text", "Positive Votes", "Abuse Votes", "Negative Votes"], ), outputs=data, raw_response=raw_response, ) # endregion def file_sandbox_report_command(client: Client, args: dict) -> List[CommandResults]: """ 1 API Call """ execution_metrics = ExecutionMetrics() results: List[CommandResults] = [] file_hash = args["file"] limit = arg_to_number(args["limit"], "limit", required=True) assert isinstance(limit, int) # mypy fix raise_if_hash_not_valid(file_hash) raw_response = client.file_sandbox_report(file_hash, limit) if "data" in raw_response: data = raw_response["data"] execution_metrics.quota_error += 1 results.append( CommandResults( f"{INTEGRATION_ENTRY_CONTEXT}.SandboxReport", "id", readable_output=tableToMarkdown( f"Sandbox Reports for file hash: {file_hash}", [{"id": item["id"], **item["attributes"], "link": item["links"]["self"]} for item in data], headers=["analysis_date", "last_modification_date", "sandbox_name", "link"], removeNull=True, headerTransform=underscoreToCamelCase, ), outputs=data, raw_response=raw_response, ) ) elif raw_response.get("error", {}).get("code") == "NotFoundError": results.append(build_unknown_file_output(client, file_hash)) else: execution_metrics.quota_error += 1 results.append(build_quota_exceeded_file_output(client, file_hash)) if execution_metrics.is_supported(): _metric_results = execution_metrics.metrics metric_results = cast(CommandResults, _metric_results) results.append(metric_results) return results def passive_dns_data(client: Client, args: dict) -> CommandResults: """ 1 API Call """ id = {} if "ip" in args: id["value"] = args["ip"] id["type"] = "ip" raise_if_ip_not_valid(id["value"]) elif "domain" in args: id["value"] = args["domain"] id["type"] = "domain" elif "id" in args: id["value"] = args["id"] if is_ip_valid(id["value"]): id["type"] = "ip" else: id["type"] = "domain" else: return CommandResults(readable_output="No IP address or domain was given.") limit = arg_to_number_must_int(args["limit"], arg_name="limit", required=True) try: raw_response = client.passive_dns_data(id, limit) except Exception: return CommandResults(readable_output=f'{"IP" if id["type"] == "ip" else "Domain"} {id["value"]} was not found.') data = raw_response["data"] return CommandResults( f"{INTEGRATION_ENTRY_CONTEXT}.PassiveDNS", "id", readable_output=tableToMarkdown( f'Passive DNS data for {"IP" if id["type"] == "ip" else "domain"} {id["value"]}', [{"id": item["id"], **item["attributes"]} for item in data], headers=["id", "date", "host_name", "ip_address", "resolver"], removeNull=True, headerTransform=underscoreToCamelCase, ), outputs=data, raw_response=raw_response, ) def search_command(client: Client, args: dict) -> CommandResults: """ 1 API Call """ query = args["query"] limit = arg_to_number_must_int(args.get("limit"), "limit", required=True) raw_response = client.search(query, limit) data = raw_response.get("data", []) if not argToBoolean(args.get("extended_data", False)): data = decrease_data_size(data) return CommandResults( f"{INTEGRATION_ENTRY_CONTEXT}.SearchResults", "id", readable_output=tableToMarkdown( f"Search result of query {query}", [item.get("attributes") for item in data], removeNull=True, headerTransform=underscoreToCamelCase, ), outputs=data, raw_response=raw_response, ) def get_analysis_command(client: Client, args: dict) -> CommandResults: """ 1 API Call """ analysis_id = args["id"] raw_response = client.get_analysis(analysis_id) data = raw_response.get("data", {}) if not argToBoolean(args.get("extended_data", False)): data = decrease_data_size(data) return CommandResults( f"{INTEGRATION_ENTRY_CONTEXT}.Analysis", "id", readable_output=tableToMarkdown( "Analysis results:", {**data.get("attributes", {}), "id": analysis_id}, headers=["id", "stats", "status"], headerTransform=underscoreToCamelCase, ), outputs={**raw_response, "id": analysis_id}, raw_response=raw_response, ) def private_get_analysis_command(client: Client, args: dict) -> CommandResults: """ 1-2 API Call """ analysis_id = args["id"] raw_response = client.get_private_analysis(analysis_id) data = raw_response.get("data", {}) attributes = data.get("attributes", {}) if sha256 := raw_response.get("meta", {}).get("file_info", {}).get("sha256"): attributes["sha256"] = sha256 if url := raw_response.get("meta", {}).get("url_info", {}).get("url"): attributes["url"] = url if attributes.get("status", "") == "completed": stats = {} item_response = client.get_private_item_from_analysis(analysis_id) item_attributes = item_response.get("data", {}).get("attributes", {}) # File attributes if threat_severity := item_attributes.get("threat_severity"): if severity_level := threat_severity.get("threat_severity_level"): stats["threat_severity_level"] = SEVERITY_LEVELS.get(severity_level, severity_level) if popular_threat_category := threat_severity.get("threat_severity_data", {}).get("popular_threat_category"): stats["popular_threat_category"] = popular_threat_category if verdict := item_attributes.get("threat_verdict"): stats["threat_verdict"] = VERDICTS.get(verdict, verdict) # URL attributes if (last_analysis_stats := item_attributes.get("last_analysis_stats")) and ( detection_engines := sum(last_analysis_stats.values()) ): positive_detections = last_analysis_stats.get("malicious", 0) stats["positives"] = f"{positive_detections}/{detection_engines}" attributes.update(stats) for field in ["title", "last_http_response_content_sha256"]: if value := item_attributes.get(field): attributes[field] = value return CommandResults( f"{INTEGRATION_ENTRY_CONTEXT}.Analysis", "id", readable_output=tableToMarkdown( "Analysis results:", {**attributes, "id": analysis_id}, headers=[ # Common headers "id", "status", # File attributes "sha256threat_severity_level", "popular_threat_category", "threat_verdict", # URL attributes "url", "title", "last_http_response_content_sha256", "positives", ], removeNull=True, headerTransform=string_to_table_header, ), outputs={**raw_response, "id": analysis_id}, raw_response=raw_response, ) def check_module(client: Client) -> str: """ 1 API Call """ client.get_ip_comments("8.8.8.8", 1) return "ok" def delete_comment(client: Client, args: dict) -> CommandResults: """Delete a comments""" id_ = args["id"] client.delete_comment(id_) return CommandResults(readable_output=f"Comment {id_} has been deleted!") def file_sigma_analysis_command(client: Client, args: dict) -> CommandResults: """Get last sigma analysis for a given file""" file_hash = args["file"] only_stats = argToBoolean(args.get("only_stats", False)) raw_response = client.file(file_hash) data = raw_response["data"] if "sigma_analysis_stats" not in data["attributes"] or "sigma_analysis_results" not in data["attributes"]: return CommandResults(readable_output=f"No Sigma analyses for file {file_hash} were found.") if only_stats: return CommandResults( f"{INTEGRATION_ENTRY_CONTEXT}.SigmaAnalysis", "id", readable_output=tableToMarkdown( f"Summary of the last Sigma analysis for file {file_hash}:", { **data["attributes"]["sigma_analysis_stats"], "**TOTAL**": sum(data["attributes"]["sigma_analysis_stats"].values()), }, headers=["critical", "high", "medium", "low", "**TOTAL**"], removeNull=True, headerTransform=underscoreToCamelCase, ), outputs=data, raw_response=data["attributes"]["sigma_analysis_stats"], ) else: return CommandResults( f"{INTEGRATION_ENTRY_CONTEXT}.SigmaAnalysis", "id", readable_output=tableToMarkdown( f"Matched rules for file {file_hash} in the last Sigma analysis:", data["attributes"]["sigma_analysis_results"], headers=[ "rule_level", "rule_description", "rule_source", "rule_title", "rule_id", "rule_author", "match_context", ], removeNull=True, headerTransform=underscoreToCamelCase, ), outputs=data, raw_response=data["attributes"]["sigma_analysis_results"], ) def get_assessment_command(client: Client, score_calculator: ScoreCalculator, args: dict) -> CommandResults: """Get Google Threat Intelligence assessment for a given resource.""" resource = args["resource"] resource_type = args.get("resource_type", "file").lower() if resource_type in ("hash", "file"): raise_if_hash_not_valid(resource) raw_response = client.file(resource) if raw_response.get("error", {}).get("code") == "QuotaExceededError": return build_quota_exceeded_file_output(client, resource) if raw_response.get("error", {}).get("code") == "NotFoundError": return build_unknown_file_output(client, resource) indicator = _get_file_indicator(client, score_calculator, resource, raw_response) elif resource_type == "ip": raise_if_ip_not_valid(resource) raw_response = client.ip(resource) if raw_response.get("error", {}).get("code") == "QuotaExceededError": return build_quota_exceeded_ip_output(client, resource) if raw_response.get("error", {}).get("code") == "NotFoundError": return build_unknown_ip_output(client, resource) indicator = _get_ip_indicator(client, score_calculator, resource, raw_response) elif resource_type == "url": raw_response = client.url(resource) if raw_response.get("error", {}).get("code") == "QuotaExceededError": return build_quota_exceeded_url_output(client, resource) if raw_response.get("error", {}).get("code") == "NotFoundError": return build_unknown_url_output(client, resource) indicator = _get_url_indicator(client, score_calculator, resource, raw_response) elif resource_type == "domain": raw_response = client.domain(resource) if raw_response.get("error", {}).get("code") == "QuotaExceededError": return build_quota_exceeded_domain_output(client, resource) if raw_response.get("error", {}).get("code") == "NotFoundError": return build_unknown_domain_output(client, resource) indicator = _get_domain_indicator(client, score_calculator, resource, raw_response) else: raise DemistoException(f'Could not find resource type of "{resource_type}"') data = raw_response.get("data", {}) data.pop("relationships", None) gti_assessment = data.get("attributes", {}).get("gti_assessment", {}) if data: if gti_assessment: data["attributes"] = {"gti_assessment": gti_assessment} else: data.pop("attributes", None) return CommandResults( f"{INTEGRATION_ENTRY_CONTEXT}.Assessment", "id", indicator=indicator, readable_output=tableToMarkdown( f'Google Threat Intelligence assessment of {resource_type}: "{resource}"', { "threat_score": gti_assessment.get("threat_score", {}).get("value"), "severity": gti_assessment.get("severity", {}).get("value"), "verdict": gti_assessment.get("verdict", {}).get("value"), }, headers=[ "threat_score", "severity", "verdict", ], headerTransform=string_to_table_header, ), outputs=data, raw_response=raw_response, ) def _get_curated_collections_command(client: Client, args: dict, collection_type: str) -> CommandResults: """Get Google Threat Intelligence collections for a given resource.""" resource = args["resource"] resource_type = args.get("resource_type", "file").lower() raw_response = client.curated_collections(resource, resource_type, collection_type) data = raw_response.get("data", []) collections = [] for collection in data: attributes = collection.get("attributes", {}) targeted_regions = { item.get("country_iso2") for item in attributes.get("targeted_regions_hierarchy", []) if item.get("country_iso2") } targeted_industries = { item.get("industry_group") for item in attributes.get("targeted_industries_tree", []) if item.get("industry_group") } collections.append( { "name": attributes.get("name"), "last_modification_date": epoch_to_timestamp(attributes.get("last_modification_date")), "targeted_regions": ", ".join(targeted_regions), "targeted_industries": ", ".join(targeted_industries), "link": f'https://www.virustotal.com/gui/collection/{collection["id"]}', } ) type_str = collection_type.replace("-", " ") type_context = type_str.title().replace(" ", "") type_title = f"{type_str[:-1]}ies" if type_str.endswith("y") else f"{type_str}s" return CommandResults( outputs_prefix=f"{INTEGRATION_ENTRY_CONTEXT}.{type_context}", outputs_key_field="id", readable_output=tableToMarkdown( f'Curated {type_title} of {resource_type}: "{resource}"', collections, headers=[ "name", "last_modification_date", "targeted_regions", "targeted_industries", "link", ], headerTransform=string_to_table_header, ), outputs={ "id": resource, "collections": data, }, raw_response=raw_response, ) def get_curated_campaigns_command(client: Client, args: dict) -> CommandResults: """Get Google Threat Intelligence campaigns for a given resource.""" return _get_curated_collections_command(client, args, "campaign") def get_curated_malware_families_command(client: Client, args: dict) -> CommandResults: """Get Google Threat Intelligence malware families for a given resource.""" return _get_curated_collections_command(client, args, "malware-family") def get_curated_threat_actors_command(client: Client, args: dict) -> CommandResults: """Get Google Threat Intelligence threat actors for a given resource.""" return _get_curated_collections_command(client, args, "threat-actor") def arg_to_relationships(arg): """Get an argument and return the relationship list.""" return (",".join(argToList(arg))).replace("* ", "").replace(" ", "_") def main(params: dict, args: dict, command: str): results: Union[CommandResults, str, List[CommandResults]] handle_proxy() client = Client(params) score_calculator = ScoreCalculator(params) ip_relationships = arg_to_relationships(params.get("ip_relationships")) url_relationships = arg_to_relationships(params.get("url_relationships")) domain_relationships = arg_to_relationships(params.get("domain_relationships")) file_relationships = arg_to_relationships(params.get("file_relationships")) disable_private_ip_lookup = argToBoolean(params.get("disable_private_ip_lookup", False)) demisto.debug(f"Command called {command}") if command == "test-module": results = check_module(client) elif command == "file": results = file_command(client, score_calculator, args, file_relationships) elif command == "ip": results = ip_command(client, score_calculator, args, ip_relationships, disable_private_ip_lookup) elif command == "url": results = url_command(client, score_calculator, args, url_relationships) elif command == "domain": results = domain_command(client, score_calculator, args, domain_relationships) elif command == "cve": results = cve_command(client, score_calculator, args) elif command == f"{COMMAND_PREFIX}-file-sandbox-report": results = file_sandbox_report_command(client, args) elif command == f"{COMMAND_PREFIX}-passive-dns-data": results = passive_dns_data(client, args) elif command == f"{COMMAND_PREFIX}-comments-get": results = get_comments_command(client, args) elif command == f"{COMMAND_PREFIX}-comments-add": results = add_comments_command(client, args) elif command == f"{COMMAND_PREFIX}-comments-get-by-id": results = get_comments_by_id_command(client, args) elif command == f"{COMMAND_PREFIX}-comments-delete": results = delete_comment(client, args) elif command == "url-scan": results = scan_url_command(client, args) elif command == "file-scan": results = file_scan(client, args) elif command == "file-rescan": results = file_rescan_command(client, args) elif command == f"{COMMAND_PREFIX}-file-scan-upload-url": results = get_upload_url(client) elif command == f"{COMMAND_PREFIX}-search": results = search_command(client, args) elif command == f"{COMMAND_PREFIX}-analysis-get": results = get_analysis_command(client, args) elif command == f"{COMMAND_PREFIX}-file-sigma-analysis": results = file_sigma_analysis_command(client, args) elif command == f"{COMMAND_PREFIX}-privatescanning-file": results = private_file_command(client, args) elif command == f"{COMMAND_PREFIX}-privatescanning-file-scan": results = private_file_scan(client, args) elif command == f"{COMMAND_PREFIX}-privatescanning-url": results = private_url_command(client, args) elif command == f"{COMMAND_PREFIX}-privatescanning-url-scan": results = private_scan_url_command(client, args) elif command == f"{COMMAND_PREFIX}-privatescanning-analysis-get": results = private_get_analysis_command(client, args) elif command == f"{COMMAND_PREFIX}-assessment-get": results = get_assessment_command(client, score_calculator, args) elif command == f"{COMMAND_PREFIX}-file-scan-and-analysis-get": results = file_scan_and_get_analysis(client, score_calculator, args, file_relationships) elif command == f"{COMMAND_PREFIX}-private-file-scan-and-analysis-get": results = private_file_scan_and_get_analysis(client, args) elif command == f"{COMMAND_PREFIX}-url-scan-and-analysis-get": results = url_scan_and_get_analysis(client, score_calculator, args, url_relationships) elif command == f"{COMMAND_PREFIX}-private-url-scan-and-analysis-get": results = private_url_scan_and_get_analysis(client, args) elif command == f"{COMMAND_PREFIX}-curated-campaigns-get": results = get_curated_campaigns_command(client, args) elif command == f"{COMMAND_PREFIX}-curated-malware-families-get": results = get_curated_malware_families_command(client, args) elif command == f"{COMMAND_PREFIX}-curated-threat-actors-get": results = get_curated_threat_actors_command(client, args) else: raise NotImplementedError(f"Command {command} not implemented") return_results(results) if __name__ in ("builtins", "__builtin__", "__main__"): try: main(demisto.params(), demisto.args(), demisto.command()) except Exception as exception: return_error(exception)