import demistomock as demisto import urllib3 from CommonServerPython import * from eve_client import eve from CommonServerUserPython import * # Disable insecure warnings urllib3.disable_warnings() """ CONSTANTS """ DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ" MAX_INCIDENTS_TO_FETCH = 50 WANTED_FIELDS = [ "identifier", "xi_scores", "cves", "product", "vendor", "mitigation", "attack_vector", "cvss", "description", "publish_date", "modified_timestamp", ] LIST_FIELDS = [ "transport", "links", "application", "zdis", "cpe23Uri", "methods", "modules", ] def get_best_xi_score(scores): """Get most recent xi_score. Args: xi_scores (dict): Dictionary with xi_scores. Returns: float: Most recent xi_score. """ if scores == []: return 0.0 commented_scores = [s for s in scores if s["comment"] != ""] if len(commented_scores) > 0: scores = commented_scores return float(sorted(scores, key=lambda x: x["created_at"], reverse=True)[0]["score"]) def extract_data(data, dest): """Extract data from nested dictionary. Args: data (dict): Nested Dictionary to extract data. dest (dict): Dictionary to store the extracted data. Returns: dict: Dictionary with extracted data. """ for key, value in data.items(): demisto_field = f"xi{key.replace('_', '').lower()}" t_value = type(value) if key == "xi_scores": dest["xiscores"] = get_best_xi_score(value) elif t_value is dict: extract_data(value, dest) elif t_value is list: for i in value: if isinstance(i, dict): extract_data(i, dest) if isinstance(i, str | float) and i in WANTED_FIELDS: dest[demisto_field] = i if key in LIST_FIELDS: for i in value: try: dest[demisto_field].append(i) except KeyError: dest[demisto_field] = [] dest[demisto_field].append(i) else: if key in LIST_FIELDS: try: dest[demisto_field].append(value) except KeyError: dest[demisto_field] = [] dest[demisto_field].append(value) elif key in WANTED_FIELDS: dest[demisto_field] = value return dest def connect(EMAIL, PASSWORD, CLIENT_ENCRYPTION_KEY, EVE_URL): """Establish a connection to Exodus Intelligence API. Returns: exodus_Obj: API Client object. """ try: exodus = eve.EVEClient(EMAIL, PASSWORD, CLIENT_ENCRYPTION_KEY, url=EVE_URL) except Exception as e: demisto.debug(f"There was a problem connecting to the server: {e}") raise ConnectionError return exodus def fetch_indicators(EMAIL, PASSWORD, CLIENT_ENCRYPTION_KEY, EVE_URL, MAX_XI, MIN_XI): """Retrieve vulnerability data from Exodus Intelligence.""" indicators = [] formatted_list = [] min_xi = 0 if MIN_XI == "" else MIN_XI max_xi = 10 if MAX_XI == "" else MAX_XI try: exodus = connect(EMAIL, PASSWORD, CLIENT_ENCRYPTION_KEY, EVE_URL) demisto.debug("Connected to server") recent_vulns = exodus.get_recent_vulns() try: data = recent_vulns["data"]["items"] except KeyError as e: demisto.debug(f"There was an error getting the data {e}") demisto.debug(f"Fetched {len(data)} total vulnerabilities") for item in data: try: cve = item["cves"][0] report_data = { "xicve": cve, "xiidentifier": item["identifier"], } report = exodus.get_report(cve) vulnerability = exodus.get_vuln(cve) if report.get("ok"): report_data = extract_data(report, report_data) vuln_data = extract_data(vulnerability, report_data) score = vuln_data.get("xiscores") if score >= min_xi and score <= max_xi: formatted_list.append(vuln_data) except KeyError as e: demisto.debug(f"There was a problem: {e}") except Exception as e: demisto.debug(f"Something went wrong: {e}") if formatted_list: for items in formatted_list: try: indicator = { "value": items["xiidentifier"], "type": "Exodus Intelligence", "fields": items, "rawJSON": items, } indicators.append(indicator) except KeyError as e: demisto.debug(f"There was a problem creating indicators: {e}") demisto.createIndicators(indicators) return f"Grabbed {len(formatted_list)} indicators with your search criteria." def reset_data_stream(EMAIL, PASSWORD, CLIENT_ENCRYPTION_KEY, EVE_URL, days): try: exodus = connect(EMAIL, PASSWORD, CLIENT_ENCRYPTION_KEY, EVE_URL) data = exodus.get_recent_vulns(days) day = data["data"]["end_ts"] results = CommandResults( readable_output=tableToMarkdown("Reset", day, headers=["date"]), outputs_prefix="ExodusVulnerabilityEnrichment.ResetDataStream", outputs_key_field="end_ts", outputs=day, ) return results except (ConnectionError, KeyError) as e: demisto.debug(f"There was a problem connecting to the API {e}") raise ConnectionError """EXECUTION""" def main(): EMAIL = demisto.params().get("email") PASSWORD = demisto.params().get("password") CLIENT_ENCRYPTION_KEY = demisto.params().get("client_encryption_key") EVE_URL = demisto.params().get("url") MAX_XI = demisto.params().get("max_xi_score") MAX_XI = int(MAX_XI) if MAX_XI else 10 MIN_XI = demisto.params().get("min_xi_score") MIN_XI = int(MIN_XI) if MIN_XI else 0 try: if demisto.command() == "fetch-indicators" or demisto.command() == "exodus-get-indicators": return_results( fetch_indicators( EMAIL, PASSWORD, CLIENT_ENCRYPTION_KEY, EVE_URL, MAX_XI, MIN_XI, ) ) elif demisto.command() == "exodus-reset-data-stream": try: days = arg_to_number(demisto.args().get("reset")) except ValueError as e: demisto.info(f"Please provide a valid number. Resetting data stream to 7 days. Error: {e}") days = 7 try: return_results(reset_data_stream(EMAIL, PASSWORD, CLIENT_ENCRYPTION_KEY, EVE_URL, days)) except ConnectionError: message = "Unable to reset data stream" return_results( CommandResults( readable_output=tableToMarkdown("Reset", message, headers=["date"]), outputs_prefix="ExodusVulnerabilityEnrichment.ResetDataStream", outputs=message, ) ) elif demisto.command() == "test-module": try: connect(EMAIL, PASSWORD, CLIENT_ENCRYPTION_KEY, EVE_URL) return_results("ok") except ConnectionError: demisto.info("Exodus authentication failed. Verify your credentials.") except Exception as e: demisto.error(traceback.format_exc()) return_error(f"Failed to execute {demisto.command()} command.\nError:\n{e!s}") """ENTRY POINT""" if __name__ in ("__main__", "__builtin__", "builtins"): main()