import demistomock as demisto # noqa: F401 from CommonServerPython import * # noqa: F401 """ IMPORTS """ import csv import ipaddress import os import requests import urllib3 # disable insecure warnings urllib3.disable_warnings() """ GLOBALS """ VERBOSE = True SERVER = demisto.params().get("server") if not SERVER.endswith("/"): SERVER += "/" API_KEY = demisto.params().get("credentials", {}).get("password") or demisto.params().get("apikey") ABUSECH_API_KEY = demisto.params().get("hunting_credentials", {}).get("password") ABUSECH_URL = demisto.params().get("abusech_hunting_url") DISABLE_PRIVATE_IP_LOOKUP = argToBoolean(demisto.params().get("disable_private_ip_lookup", False)) MAX_AGE = demisto.params().get("days", "30") THRESHOLD = demisto.params().get("threshold", "80") INSECURE = demisto.params().get("insecure", False) TEST_IP = "127.0.0.2" BLACKLIST_SCORE = 3 CHECK_CMD = "check" CHECK_BLOCK_CMD = "check-block" REPORT_CMD = "report" BLACKLIST_CMD = "blacklist" ANALYSIS_TITLE = "AbuseIPDB Analysis" BLACKLIST_TITLE = "AbuseIPDB Blacklist" REPORT_SUCCESS = "IP address reported successfully." API_QUOTA_REACHED_MESSAGE = "Too many requests (possibly bad API key). Status code: 429" HEADERS = {"Key": API_KEY, "Accept": "application/json"} PROXY = demisto.params().get("proxy", False) if not demisto.params().get("proxy", False): # Remove proxy environment variables if they exist for proxy_var in ["HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy"]: os.environ.pop(proxy_var, None) CATEGORIES_NAME = { 1: "DNS_Compromise", 2: "DNS_Poisoning", 3: "Frad_Orders", 4: "DDoS_Attack", 5: "FTP_Brute-Force", 6: "Ping of Death", 7: "Phishing", 8: "Fraud VoIP", 9: "Open_Proxy", 10: "Web_Spam", 11: "Email_Spam", 12: "Blog_Spam", 13: "VPN IP", 14: "Port_Scan", 15: "Hacking", 16: "SQL Injection", 17: "Spoofing", 18: "Brute_Force", 19: "Bad_Web_Bot", 20: "Exploited_Host", 21: "Web_App_Attack", 22: "SSH", 23: "IoT_Targeted", } CATEGORIES_ID = { "Frad_Orders": "3", "DDoS_Attack": "4", "FTP_Brute": "5", "Ping of Death": "6", "Phishing": "7", "Fraud VoIP": "8", "Open_Proxy": "9", "Web_Spam": "10", "Email_Spam": "11", "Blog_Spam": "12", "VPN IP": "13", "Port_Scan": "14", "Hacking": "15", "SQL Injection": "16", "Spoofing": "17", "Brute_Force": "18", "Bad_Web_Bot": "19", "Exploited_Host": "20", "Web_App_Attack": "21", "SSH": "22", "IoT_Targeted": "23", } session = requests.session() """ HELPER FUNCTIONS """ def http_request(method, url_suffix, params=None, headers=HEADERS, threshold=THRESHOLD): LOG("running request with url=%s" % (SERVER + url_suffix)) try: analysis = session.request(method, SERVER + url_suffix, headers=headers, params=params, verify=not INSECURE) if analysis.status_code not in {200, 204, 429}: return_error("Bad connection attempt. Status code: " + str(analysis.status_code)) if analysis.status_code == 429: if demisto.params().get("disregard_quota", False): return API_QUOTA_REACHED_MESSAGE else: return_error(API_QUOTA_REACHED_MESSAGE) return REPORT_SUCCESS if url_suffix == REPORT_CMD else analysis.json() except Exception as e: LOG(e) return_error(str(e)) def abusech_hunting_http_request(headers: dict, payload: dict) -> requests.Response: """ Dedicated request helper for Abuse.ch Hunting API. Sends a POST request with a JSON body to the base URL. Args: headers (dict): The headers for the request. payload (dict): The JSON payload for the request. Returns: requests.Response: The response from the Abuse.ch Hunting API. """ if not ABUSECH_URL: raise Exception("Hunting API URL was not provided in the integration parameters.") if not ABUSECH_API_KEY: raise Exception("Hunting API Key was not provided in the integration parameters.") demisto.debug(f"Sending a POST request to '{ABUSECH_URL}' with the following payload: {payload}") try: response = session.request(method="POST", url=ABUSECH_URL, headers=headers, json=payload, verify=not INSECURE) response.raise_for_status() except Exception as e: raise Exception(f"Failed to connect to Abuse.ch: {str(e)}") # this API wraps errors with a status=200 response, attempt to decode it as json try: res_json = response.json() # when `requests.Request.json()` fails, a `json.JSONDecodeError` is raised, it inherits from `ValueError` except ValueError: # the response is not a json, which means the response doesn't wrap an error return response # if 'query_status' is present, an error is wrapped within the json if res_json.get("query_status"): demisto.debug(f"Hunting API response: {res_json}") error_message = res_json.get("data", "Hunting API response error.") raise Exception(f"Failed to connect to Abuse.ch: {error_message}") return response def format_privte_ips(private_ips): entry = { "ContentsFormat": formats["json"], "Type": entryTypes["note"], "Contents": private_ips, "ReadableContentsFormat": formats["markdown"], "HumanReadable": tableToMarkdown( "AbuseIPDB lookup was avoided for the following private IPs.", private_ips, headers="Private IPs", removeNull=True, ), } return entry def analysis_to_entry(info, reliability, threshold=THRESHOLD, verbose=VERBOSE): if not isinstance(info, list): info = [info] context_ip_generic, context_ip, human_readable, dbot_scores, timeline = [], [], [], [], [] for analysis in info: ip_ec = { "Address": analysis.get("ipAddress"), "Geo": {"Country": analysis.get("countryName"), "CountryCode": analysis.get("countryCode")}, } abuse_ec = { "IP": { "Address": analysis.get("ipAddress"), "Geo": {"Country": analysis.get("countryName"), "CountryCode": analysis.get("countryCode")}, "AbuseConfidenceScore": analysis.get("abuseConfidenceScore"), "TotalReports": analysis.get("totalReports") or analysis.get("numReports") or 0, "ISP": analysis.get("isp"), "UsageType": analysis.get("usageType"), "Domain": analysis.get("domain"), "Hostnames": analysis.get("hostnames"), "IpVersion": analysis.get("ipVersion"), "IsPublic": analysis.get("isPublic"), "IsTor": analysis.get("isTor"), "IsWhitelisted": analysis.get("isWhitelisted"), "LastReportedAt": analysis.get("lastReportedAt"), "NumDistinctUsers": analysis.get("numDistinctUsers"), } } if verbose: reports = sum([report_dict.get("categories") for report_dict in analysis.get("reports")], []) # type: list categories = set(filter(lambda category_id: category_id in CATEGORIES_NAME, reports)) abuse_ec["IP"]["Reports"] = {CATEGORIES_NAME[c]: reports.count(c) for c in categories} human_readable.append(abuse_ec["IP"]) dbot_score = getDBotScore(analysis, threshold) if dbot_score == 3: ip_ec["Malicious"] = abuse_ec["IP"]["Malicious"] = { "Vendor": "AbuseIPDB", "Detections": "The address was reported as Malicious by AbuseIPDB.", "Description": "The address was reported as Malicious by AbuseIPDB.", } dbot_scores.append( { "Score": dbot_score, "Vendor": "AbuseIPDB", "Indicator": analysis.get("ipAddress"), "Type": "ip", "Reliability": reliability, } ) context_ip.append(abuse_ec) context_ip_generic.append(ip_ec) ip_address = analysis.get("ipAddress") ip_rep = scoreToReputation(dbot_score) timeline.append( { "Value": ip_address, "Message": f'AbuseIPDB marked the indicator "{ip_address}" as *{ip_rep}*', "Category": "Integration Update", } ) return createEntry(context_ip, context_ip_generic, human_readable, dbot_scores, timeline, title=ANALYSIS_TITLE) def blacklist_to_entry(data, saveToContext): if not isinstance(data, list): data = [data] ips = [d.get("ipAddress") for d in data] context = {"Blacklist": ips} temp = demisto.uniqueFile() with open(demisto.investigation()["id"] + "_" + temp, "w") as f: wr = csv.writer(f, quoting=csv.QUOTE_ALL) for ip in ips: wr.writerow([ip]) entry = { "HumanReadable": "", "Contents": ips, "ContentsFormat": formats["json"], "Type": entryTypes["file"], "File": "Blacklist.csv", "FileID": temp, "EntryContext": {"AbuseIPDB": createContext(context if saveToContext else None, removeNull=True)}, } return entry def getDBotScore(analysis, threshold=THRESHOLD): total_reports = analysis.get("totalReports") or analysis.get("numReports") or 0 abuse_score = int(analysis.get("abuseConfidenceScore")) dbot_score = 0 if total_reports == 0 else 1 if abuse_score < 20 else 2 if abuse_score < int(threshold) else 3 return dbot_score def createEntry(context_ip, context_ip_generic, human_readable, dbot_scores, timeline, title): entry = { "ContentsFormat": formats["json"], "Type": entryTypes["note"], "Contents": context_ip, "ReadableContentsFormat": formats["markdown"], "HumanReadable": tableToMarkdown(title, human_readable, removeNull=True), "EntryContext": { "IP(val.Address && val.Address == obj.Address)": createContext(context_ip_generic, removeNull=True), "AbuseIPDB(val.IP.Address && val.IP.Address == obj.IP.Address)": createContext(context_ip, removeNull=True), "DBotScore": createContext(dbot_scores, removeNull=True), }, "IndicatorTimeline": timeline, } return entry """ FUNCTIONS """ def check_ip_command( reliability, ip, override_private_lookup="False", days=MAX_AGE, verbose=VERBOSE, threshold=THRESHOLD, disable_private_ip_lookup=DISABLE_PRIVATE_IP_LOOKUP, **kwargs, ): params = {"maxAgeInDays": days} if verbose: params["verbose"] = "verbose" ip_list = argToList(ip) entry_list = [] private_ips = [] for current_ip in ip_list: if ( disable_private_ip_lookup and ipaddress.ip_address(current_ip).is_private and not argToBoolean(override_private_lookup) ): private_ips.append(current_ip) continue params["ipAddress"] = current_ip analysis = http_request("GET", url_suffix=CHECK_CMD, params=params) if analysis == API_QUOTA_REACHED_MESSAGE: continue analysis_data = analysis.get("data") entry_list.append(analysis_to_entry(analysis_data, reliability, verbose=verbose, threshold=threshold)) if private_ips and disable_private_ip_lookup: demisto.results(format_privte_ips(private_ips)) return entry_list def check_block_command(reliability, network, limit, days=MAX_AGE, threshold=THRESHOLD, **kwargs): params = {"network": network, "maxAgeInDays": days} analysis = http_request("GET", url_suffix=CHECK_BLOCK_CMD, params=params).get("data").get("reportedAddress") return analysis_to_entry( analysis[: int(limit) if limit.isdigit() else 40], verbose=False, threshold=threshold, reliability=reliability ) def report_ip_command(ip, categories, **kwargs): params = {"ip": ip, "categories": ",".join([CATEGORIES_ID.get(c, c) for c in categories.split()])} analysis = http_request("POST", url_suffix=REPORT_CMD, params=params) return analysis def get_blacklist_command(limit, days, confidence, saveToContext, **kwargs): params = {"maxAgeInDays": days, "confidenceMinimum": confidence, "limit": limit} analysis = http_request("GET", url_suffix=BLACKLIST_CMD, params=params) return analysis if type(analysis) is str else blacklist_to_entry(analysis.get("data"), saveToContext) def get_fplist_command(format: str, limit: int, all_results: bool, **kwargs) -> Union[CommandResults, Dict]: """ Retrieves the False Positive List (FPL) from abuse.ch. Args: format (str): The format of the response (json or csv). limit (int): The maximum number of entries to return. all_results (bool): Whether to return all results. """ demisto.debug("Retrieving the false positive list from hunting abuse api.") limit_int = arg_to_number(limit) all_results_bool = argToBoolean(all_results) headers = {"Auth-Key": ABUSECH_API_KEY, "Content-Type": "application/json"} payload = {"query": "get_fplist", "format": format} response = abusech_hunting_http_request(headers, payload) if format == "json": res_json = response.json() if not isinstance(res_json, dict): raise Exception(f"Unexpected response format from Abuse.ch: {res_json}") demisto.debug("Flattening json response.") data = [] for entry_id, details in res_json.items(): if isinstance(details, dict): details["id"] = entry_id data.append(details) else: demisto.debug(f"Skipping non-dictionary entry in FPL response: {entry_id}={details}") if not data: demisto.debug("Response had no valid data within it.") return CommandResults(readable_output="No data found in the False Positive List.") if not all_results_bool: demisto.debug(f"Trimming data to first {limit_int} entries.") data = data[:limit_int] readable_output = tableToMarkdown( "Abuse.ch False Positive List", data, headers=["id", "time_stamp", "platform", "entry_type", "entry_value", "removed_by", "removal_notes"], removeNull=True, ) return CommandResults( outputs_prefix="AbuseIPDB.FPL", outputs_key_field="id", outputs=data, readable_output=readable_output ) elif format == "csv": return fileResult("abusech_fplist.csv", response.content) else: raise ValueError(f"Unexpected format type: {format}") def test_module(reliability): try: check_ip_command(ip=TEST_IP, disable_private_ip_lookup=False, verbose=False, reliability=reliability) except Exception as e: return_error(f"AbuseIPDB connection failed: {str(e)}") if ABUSECH_API_KEY: try: get_fplist_command("json", 1, False) except Exception as e: return_error(f"Abuse.ch Hunting API connection failed: {str(e)}") demisto.results("ok") def get_categories_command(**kwargs): categories = {str(key): value for key, value in CATEGORIES_NAME.items()} entry = { "ContentsFormat": formats["json"], "Type": entryTypes["note"], "Contents": categories, "ReadableContentsFormat": formats["markdown"], "HumanReadable": tableToMarkdown("AbuseIPDB report categories", categories, removeNull=True), "EntryContext": { "AbuseIPDB.Categories(val && val == obj)": createContext(categories, removeNull=True), }, } return entry def main(): try: reliability = demisto.params().get("integrationReliability", "C - Fairly reliable") if DBotScoreReliability.is_valid_type(reliability): reliability = DBotScoreReliability.get_dbot_score_reliability_from_str(reliability) else: raise Exception("Please provide a valid value for the Source Reliability parameter.") if demisto.command() == "test-module": # Tests connectivity and credentails on login test_module(reliability) elif demisto.command() == "ip": demisto.results(check_ip_command(reliability, **demisto.args())) elif demisto.command() == "abuseipdb-check-cidr-block": demisto.results(check_block_command(reliability, **demisto.args())) elif demisto.command() == "abuseipdb-report-ip": demisto.results(report_ip_command(**demisto.args())) elif demisto.command() == "abuseipdb-get-blacklist": demisto.results(get_blacklist_command(**demisto.args())) elif demisto.command() == "abuseipdb-get-categories": demisto.results(get_categories_command(**demisto.args())) # type:ignore elif demisto.command() == "abuseipdb-get-fplist": return_results(get_fplist_command(**demisto.args())) except Exception as e: LOG.print_log() return_error(str(e)) if __name__ in ("__main__", "__builtin__", "builtins"): main()