from CommonServerPython import * # noqa: E402 lgtm [py/polluting-import] # CONSTANTS DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ" class pXScoring: """ Class to handle all current and future scoring for PerimeterX objects """ @staticmethod def dbotscore_from_risk(risk_score: int, thresholds: dict[str, Any]) -> int: """ Create the DBotScore structure first using the Common.DBotScore class. :type risk_score: ``int`` :param risk_score: PerimeterX provided risk score for the IP Address :return: Returns the relevant DBotScore value for the provided PerimeterX risk score :rtype: ``int`` """ if risk_score > thresholds["bad_threshold"]: dbot_score = Common.DBotScore.BAD # bad elif risk_score > thresholds["suspicious_threshold"]: dbot_score = Common.DBotScore.SUSPICIOUS # suspicious elif risk_score > thresholds["good_threshold"]: dbot_score = Common.DBotScore.GOOD # good else: dbot_score = Common.DBotScore.NONE # unknown return dbot_score @staticmethod def get_ip_score(ip_address: str, risk_score: int, thresholds: dict[str, Any]): """ Create the DBotScore structure first using the Common.DBotScore class. :type ip_address: ``str`` :param ip_address: IP Address to be used as the indicator for this entry :type risk_score: ``int`` :param risk_score: PerimeterX provided risk sco1re for the IP Address :return: Updated List of Common IP Addresses with this entry appended :rtype: ``List[Common.IP]`` """ # create the DBotScore object and populate it with the needed values dbot_score = Common.DBotScore( indicator=ip_address, indicator_type=DBotScoreType.IP, integration_name="PerimeterX", score=pXScoring.dbotscore_from_risk(risk_score, thresholds), malicious_description="High risk score indicates high probability that the requests from the IP are malicious ", reliability=demisto.params().get("integrationReliability"), ) # Create the IP Standard Context structure using Common.IP and add # dbot_score to it. return Common.IP( ip=ip_address, # asn=ip_data.get('asn'), dbot_score=dbot_score, ) class Client(BaseClient): """ Client will implement the service API, and should not contain any Demisto logic. Should only do requests and return data. """ def test_api_connection(self, api_key: str) -> dict[str, Any]: """ Makes a call to the status API path to confirm the proper URL and Authorization token were provided """ headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} return self._http_request(method="GET", full_url=f"{self._base_url}?search=ip:1.1.1.1&tops=path", headers=headers) def post_investigate_by_ip(self, ip_type: str, ip_address: str, api_key: str) -> dict[str, Any]: """ Query the PerimeterX API to get the relevant details regarding the provided IP within a particular customer's own data :type ip_address: ``str`` :param ip_address: IP Address to be used as the indicator for this entry :type ip_type: ``str`` :param ip_type: The type of IP address we will be querying (true_ip or socket_ip) :return: The JSON response body from the PerimeterX API :rtype: ``Dict[str, Any]`` """ headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} request_params: dict[str, Any] = {} if ip_type: request_params["ip_type"] = ip_type if ip_address: request_params["ip_address"] = f"search=ip:{ip_address}" return self._http_request( method="GET", full_url=f'{self._base_url}?{request_params["ip_address"]}&tops=user-agent,path,socket_ip_classification', headers=headers, ) def post_investigate_by_name(self, name_type: str, name: str) -> dict[str, Any]: """ THIS IS NOT CURRENTLY IMPLEMENTED Query the PerimeterX API to get the relevant details regarding the provided name within a particular customer's own data :type name: ``str`` :param name: name to be used as the indicator for this entry :type name_type: ``str`` :param name_type: The type of name we will be querying (domain or param) :return: The JSON response body from the PerimeterX API :rtype: ``Dict[str, Any]`` """ request_params: dict[str, Any] = {} if name_type: request_params["name_type"] = name_type if name: request_params["name"] = name return self._http_request(method="POST", full_url=self._base_url, json_data=request_params) def test_module(client: Client, api_key): """ Returning 'ok' indicates that the integration works like it is supposed to. Connection to the service is successful. Args: client: PerimeterX client Returns: 'ok' if test passed, anything else will fail the test. """ try: result = client.test_api_connection(api_key) if result["success"]: return "ok" else: return "Connection to api failed: " + result["errors"] except DemistoException as de: return "Connection to api failed with exception: " + de.message def perimeterx_get_investigate_details( client: Client, args: dict[str, Any], thresholds: dict[str, Any], api_key: str ) -> CommandResults: """ Collect the required details to query the PerimeterX API to get the relevant details regarding the provided search term within a particular customer's own data :type search_term: ``str`` :param search_term: This is the entry that we'll be querying for against the PerimeterX API :type search_type: ``str`` :param search_type: The type of query that will be run against the PerimeterX API :return: A ``CommandResults`` object that is then passed to ``return_results``, that contains Domains :rtype: ``CommandResults`` """ # Check for the field to query for search_type = args.get("search_type", None) if not search_type: raise ValueError("No search_type specified") # Check to make sure we have a query term search_term = args.get("search_term", None) if not search_term: raise ValueError("No search_term specified") # Check for an IP based investigation supported_ip_search_types = ["true_ip", "socket_ip"] if search_type in supported_ip_search_types: """ Run an IP based search if the search type is one supported by the IP types """ result = client.post_investigate_by_ip(ip_type=search_type, ip_address=search_term, api_key=api_key) indicator = pXScoring.get_ip_score(ip_address=search_term, risk_score=result["max_risk_score"], thresholds=thresholds) return CommandResults(outputs_prefix="PerimeterX", outputs_key_field="", outputs=result, indicator=indicator) elif search_type == "name": """ THIS IS NOT CURRENTLY IMPLEMENTED Run a name based search if the search type is name """ result = client.post_investigate_by_name(name_type=search_type, name=search_term) else: """ Generate an error because the search type is not supported """ raise ValueError("Invalid search_type provided") return CommandResults(outputs_prefix="PerimeterX", outputs_key_field="", outputs=result) def ip(client: Client, args, thresholds: dict[str, Any], api_key): """ Collect the details to run an IP Reputation query against the PerimeterX API :type ip: ``str`` :param ip: Results will be provided for list of IPs :return: A ``CommandResults`` object that is then passed to ``return_results``, that contains Domains :rtype: ``CommandResults`` """ # Check to make sure we have a query term ip_list = argToList(args.get("ip")) results = [] for ip_address in ip_list: result = client.post_investigate_by_ip(ip_type="true_ip", ip_address=ip_address, api_key=api_key) indicator = pXScoring.get_ip_score(ip_address=ip_address, risk_score=result["max_risk_score"], thresholds=thresholds) readable_output = f"{indicator}" cr = CommandResults( readable_output=readable_output, outputs_prefix="PerimeterX", outputs_key_field="", outputs=result, indicator=indicator, ) results.append(cr) return results def main(): """ PARSE AND VALIDATE INTEGRATION PARAMS """ params = demisto.params() api_key = params.get("apikey") # get the service API url base_url = urljoin(params["url"], "/v1/bot-defender/") verify_certificate = not params.get("insecure", False) proxy = params.get("proxy", False) # get the DBot Thresholds thresholds = { "good_threshold": int(params.get("dbotGoodThreshold")), "suspicious_threshold": int(params.get("dbotSuspiciousThreshold")), "bad_threshold": int(params.get("dbotBadThreshold")), "unknown_threshold": 0, } command = demisto.command() LOG(f"Command being called is {command}") try: headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} client = Client(base_url=base_url, verify=verify_certificate, headers=headers, proxy=proxy) if command == "test-module": # This is the call made when pressing the integration Test button. result = test_module(client, api_key) demisto.results(result) elif command == "ip": return_results(ip(client, demisto.args(), thresholds=thresholds, api_key=api_key)) elif command == "perimeterx_get_investigate_details": return_results( perimeterx_get_investigate_details(client=client, args=demisto.args(), thresholds=thresholds, api_key=api_key) ) # Log exceptions except Exception as e: return_error(f"Failed to execute {demisto.command()} command. Error: {str(e)}") if __name__ in ("__main__", "__builtin__", "builtins"): main()