from typing import Any import demistomock as demisto # noqa: F401 import urllib3 from CommonServerPython import * # noqa # pylint: disable=unused-wildcard-import # Disable insecure warnings urllib3.disable_warnings() """ CONSTANTS """ DEFAULT_FEED_TAGS = {"XPANSE"} DEFAULT_ASSET_SEARCH_LIMIT = 5000 V1_URL_SUFFIX = "/public_api/v1" """ CLIENT CLASS """ class Client(BaseClient): """ Client class to interact with the service API """ def __init__(self, base_url: str, verify: bool, proxy: bool, feed_tags: list[str], tlp_color: str, headers: dict): """ Class initialization. """ super().__init__(base_url=base_url, verify=verify, proxy=proxy, headers=headers) self.feed_tags = feed_tags self.tlp_color = tlp_color self.verify = verify self.proxy = proxy self.headers = headers def list_asset_internet_exposure_request( self, search_params: list[dict] = [], search_from: int = 0, search_to: int = DEFAULT_ASSET_SEARCH_LIMIT, use_paging: bool = True, ) -> list: """Get a list of all your internet exposure assets using the '/assets/get_assets_internet_exposure/' endpoint. Args: search_params (list): list of search parameters to add to the API call body. search_from (int): Starting search index. search_to (int): Ending search index. use_paging (bool): whether to use paging or not (default is True) Returns: List: list containing dictionaries of internet exposure assets. """ body = { "request_data": { "filters": search_params, "search_to": int(search_to), "search_from": int(search_from), "use_page_token": True, } } full_response = [] while True: result = self._http_request( method="POST", url_suffix=f"{V1_URL_SUFFIX}/assets/get_assets_internet_exposure/", json_data=body ) data = result.get("reply", {}).get("assets_internet_exposure") if data: full_response.extend(data) if not use_paging: break pagination = result.get("reply", {}).get("next_page_token") if pagination is None: break body["request_data"]["next_page_token"] = pagination return full_response """ HELPER FUNCTIONS """ def create_x509_certificate_grids(string_object: Optional[str]) -> list: """ Creates a grid field related to the subject and issuer field of the x509 certificate object. Args: string_object (Optional[str]): A str in format of C=ZA, Inc.,ST=Western Cape,L=Cape Town,O=Thawte. Returns: list: The return value. A list of dict [{"title": "C", "data": "ZA"}]. """ result_grid_list = [] if string_object: key_value_pairs = string_object.split(",") for pair in key_value_pairs: result_grid = {} # '=' in pair means we extracted the right entries for k/v if "=" in pair: key, value = pair.split("=", 1) result_grid["title"] = key result_grid["data"] = value result_grid_list.append(result_grid) # If no '=' that means we had a ',' within the value we need to append else: result_grid_list[-1]["data"] = (result_grid_list[-1]["data"] + ", " + pair).replace("\\", "") return result_grid_list def map_indicator_fields(raw_indicator: dict[str, Any], asset_type: str) -> dict[str, Any]: """ Create indicator field mapping based on asset_type Args: raw_indicator (Dict[str, Any]): raw indicator as JSON. asset_type (str): indicator type Returns: Dict: dictionary of indicator field mappings. """ # name is a required API return parameter description = raw_indicator["name"] + " indicator of asset type " + asset_type + " from Cortex Xpanse" indicator_fields = {"internal": True, "description": description} if asset_type == "Domain": if domain_details := raw_indicator.get("domain_details"): domain_fields_mapping: dict = { "creationDate": "creationdate", "registryExpiryDate": "expirationdate", } for key, mapped_key in domain_fields_mapping.items(): if detail_value := domain_details.get(key): indicator_fields[mapped_key] = timestamp_to_datestring(detail_value) elif asset_type == "X509 Certificate" and (cert_details := raw_indicator.get("certificate_details")): cert_fields_mapping: dict = { "signatureAlgorithm": ("signaturealgorithm", None), "serialNumber": ("serialnumber", None), "validNotAfter": ("validitynotafter", timestamp_to_datestring), "validNotBefore": ("validitynotbefore", timestamp_to_datestring), "issuer": ("issuer", create_x509_certificate_grids), "subject": ("subject", create_x509_certificate_grids), } for key, (mapped_key, processing_func) in cert_fields_mapping.items(): if detail_value := cert_details.get(key): # Apply processing function if one is defined if processing_func: indicator_fields[mapped_key] = processing_func(detail_value) else: indicator_fields[mapped_key] = detail_value return indicator_fields def map_indicator_type(asset_type: str) -> str: """ Correlates asset_type to indicator type or returns "None" Args: asset_type (str): Xpanse asset type. Returns: str: indicator type or "None". """ asset_types_mapping = { "UNASSOCIATED_RESPONSIVE_IP": "IP", "DOMAIN": "Domain", "CERTIFICATE": "X509 Certificate", "CIDR": "CIDR", } return asset_types_mapping.get(asset_type, "None") def build_asset_indicators(client: Client, raw_indicators: list[dict[str, Any]]) -> list: """ Builds indicators JSON data in XSOAR expected format from the raw response. Args: client (Client): Xpanse client. raw_indicators (List[Dict[str, Any]]): raw indicators as JSON. Returns: List: list of indicators to be send to XSOAR. """ demisto.debug(f"Creating {len(raw_indicators)} asset indicators.") indicators: list = [] for raw_indicator in raw_indicators: asset_type = raw_indicator.get("asset_type", "None") indicator_type = map_indicator_type(asset_type) # Skip IPv6 responsive or not found type if raw_indicator.get("ipv6s") or indicator_type == "None": continue # name is a required API return parameter name = raw_indicator["name"] indicator_type = "DomainGlob" if "*" in name and indicator_type == "Domain" else indicator_type fields = map_indicator_fields(raw_indicator, indicator_type) # Add TLP color and feed tags if they exist if client.tlp_color: fields["trafficlightprotocol"] = client.tlp_color if client.feed_tags: fields["tags"] = client.feed_tags indicator = {"value": name, "type": indicator_type, "fields": fields, "rawJSON": raw_indicator} indicators.append(indicator) return indicators """ COMMAND FUNCTIONS """ def test_module(client: Client): # pragma: no cover """ Tests API connectivity and authentication' Returning 'ok' indicates that the integration works like it is supposed to. Connection to the service is successful. Raises exceptions if something goes wrong. :type client: ``Client`` :param Client: client to use :return: 'ok' if test passed, anything else will fail the test. :rtype: ``str`` """ client.list_asset_internet_exposure_request(search_to=1, use_paging=False) return_results("ok") def fetch_indicators(client: Client, limit: Optional[int] = None, asset_type: list = []) -> tuple | list: """ Fetch indicators from Xpanse API and create indicators in XSOAR. Args: client (Client): Xpanse client. limit (int): limt the number of indicators to return. asset_type (str): which asset_types to pull from API. Returns: List: list of indicators to be send to XSOAR. List: raw response from API. """ asset_list, asset_response = [], [] if asset_type == []: asset_list = ["DOMAIN", "CERTIFICATE", "UNASSOCIATED_RESPONSIVE_IP"] else: if "Domain" in asset_type: asset_list.append("DOMAIN") if "Certificate" in asset_type: asset_list.append("CERTIFICATE") if "IP" in asset_type: asset_list.append("UNASSOCIATED_RESPONSIVE_IP") if limit: # Had to add 1 to the limit to get the right return. asset_response = client.list_asset_internet_exposure_request( search_params=[{"field": "type", "operator": "in", "value": asset_list}], search_to=limit + 1, use_paging=False ) else: asset_response = client.list_asset_internet_exposure_request( search_params=[{"field": "type", "operator": "in", "value": asset_list}] ) assset_indicators = build_asset_indicators(client, asset_response) return assset_indicators, asset_response """ MAIN FUNCTION """ def get_indicators(client: Client, args: dict[str, Any]) -> CommandResults: """ Get indicators from Xpanse API, mainly for debug. Args: client (Client): Xpanse client. args (dict): all command arguments, usually passed from ``demisto.args()``. Returns: CommandResults: A ``CommandResults`` object that is then passed to ``return_results``, that contains Xpanse indicators. """ hr_list = [] asset_type = [] if argToBoolean(args.get("ip", "yes")): asset_type.append("IP") if argToBoolean(args.get("domain", "yes")): asset_type.append("Domain") if argToBoolean(args.get("certificate", "yes")): asset_type.append("Certificate") limit = arg_to_number(args.get("limit", None)) if limit and limit <= 0: raise ValueError("Limit must be a positive number.") if limit and limit > DEFAULT_ASSET_SEARCH_LIMIT: raise ValueError("Limit must be less that the API limit of " + str(DEFAULT_ASSET_SEARCH_LIMIT) + ".") if asset_type == []: raise ValueError("need to specify at least one asset type") indicators, raw_res = fetch_indicators(client=client, limit=limit, asset_type=asset_type) indicators = indicators[:limit] if isinstance(indicators, list) else [indicators] if indicators else [] for record in indicators: hr = {"Name": record.get("value"), "Type": record.get("type"), "Description": record["fields"]["description"]} hr_list.append(hr) return CommandResults( outputs=hr_list, outputs_prefix="ASM.Indicators", raw_response=raw_res, readable_output=tableToMarkdown("Xpanse indicators", hr_list, headers=["Name", "Type", "Description"]), outputs_key_field="Name", ) def main() -> None: # pragma: no cover """ main function """ params = demisto.params() base_url = params.get("url") verify_certificate = not params.get("insecure", False) proxy = params.get("proxy", False) # Append default tags. feed_tags = list(set(argToList(params.get("feedTags", []))) | DEFAULT_FEED_TAGS) tlp_color = params.get("tlp_color", "") indicator_types = params.get("indicatorTypes", []) creds = params.get("credentials", {}) api = creds.get("password", "") add_sensitive_log_strs(api) auth_id = creds.get("identifier", "") headers = {"Authorization": f"{api}", "x-xdr-auth-id": f"{auth_id}", "Content-Type": "application/json"} command = demisto.command() demisto.info(f"Command being called is {command}") try: client = Client( base_url=base_url, verify=verify_certificate, proxy=proxy, feed_tags=feed_tags, tlp_color=tlp_color, headers=headers ) if command == "test-module": test_module(client) elif command == "fetch-indicators": indicators, _ = fetch_indicators(client, asset_type=indicator_types) for iter_ in batch(indicators, batch_size=2000): try: demisto.createIndicators(iter_) except Exception: # find problematic indicator for indicator in iter_: try: demisto.createIndicators([indicator]) except Exception as err: demisto.debug( f"createIndicators Error: failed to create the following indicator: {indicator}\n {err}" ) raise elif command == "xpanse-get-indicators": return_results(get_indicators(client, demisto.args())) else: raise NotImplementedError(f'Command "{command}" is not implemented.') except Exception as e: return_error(f"Failed to execute {demisto.command()} command.\nError:\n{e!s}") """ ENTRY POINT """ if __name__ in ("__main__", "__builtin__", "builtins"): main()