import traceback from json.decoder import JSONDecodeError import demistomock as demisto # noqa: F401 import urllib3 from CommonServerPython import * # noqa: F401 # Disable insecure warnings urllib3.disable_warnings() # pylint: disable=no-member """ CONSTANTS """ SOCRADAR_API_ENDPOINT = "https://platform.socradar.com/api" DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ" # XSOAR default in ISO8601 format SOCRADAR_DATE_FORMAT = "%Y-%m-%d %H:%M:%S" MAX_INDICATOR_FETCH_NUMBER = 1000 MESSAGES: dict[str, str] = { "BAD_REQUEST_ERROR": "An error occurred while fetching the data.", "AUTHORIZATION_ERROR": "Authorization Error: make sure API Key is correctly set.", "RATE_LIMIT_EXCEED_ERROR": "Rate limit has been exceeded. Please make sure your your API key's rate limit is adequate.", } SOCRADAR_RECOMMENDED_COLLECTIONS = [ "SOCRadar-Attackers-Recommended-Block-Hash", "SOCRadar-Attackers-Recommended-Block-IP", "SOCRadar-Attackers-Recommended-Block-Domain", "SOCRadar-Recommended-Ransomware-Hash", "SOCRadar-Recommended-Phishing-Global", "SOCRadar-Recommended-Block-Hash", "SOCRadar-Recommended-Phishing-Local", "SOCRadar-APT-Recommended-Block-IP", "SOCRadar-APT-Recommended-Block-Domain", "SOCRadar-APT-Recommended-Block-Hash", "SOCRadar-Botnet C&C - Block-Domain", "SOCRadar-Botnet C&C - Block-IP", ] INTEGRATION_NAME = "Feed SOCRadar ThreatFeed" """ HELPER FUNCTIONS """ def parse_int_or_raise(str_to_parse: Any, error_msg=None) -> int: """Parse a string to integer. Raise ValueError exception if fails with given error_msg""" try: res = int(str_to_parse) except (TypeError, ValueError): if not error_msg: error_msg = f"Error while parsing integer! Provided string: {str_to_parse}" raise ValueError(error_msg) return res def build_entry_context(indicators: Union[dict, List]) -> List[dict]: """Formatting indicators from SOCRadar Threat Feed/IOC API to Demisto Context :type indicators: ``Union[Dict, List]`` :param indicators: Indicators obtained from SOCRadar Threat Feed/IOC API. :return: List of context entry dictionaries. :rtype: ``list`` """ return_context = [] for indicator_dict in indicators: indicator = indicator_dict["value"] indicator_type = indicator_dict["type"] indicator_context_dict = { "Indicator": indicator, "Indicator Type": indicator_type, "rawJSON": indicator_dict["rawJSON"], "First Seen Date": indicator_dict["fields"]["firstseenbysource"], "Last Seen Date": indicator_dict["fields"]["lastseenbysource"], "Feed Maintainer Name": indicator_dict["fields"]["collection_maintainer_name"], "Seen Count": indicator_dict["fields"]["extra_info"].get("seen_count", 1), } if indicator_type == FeedIndicatorType.IP and indicator_dict["fields"]["extra_info"].get("geo_location", []): geo_location_dict = indicator_dict["fields"]["extra_info"]["geo_location"] asn_code = geo_location_dict.get("AsnCode", "") asn_description = geo_location_dict.get("AsnName", "") asn = f"[{asn_code}] {asn_description}" geo_location_dict["ASN"] = asn geo_location_dict = { key: value for key, value in geo_location_dict.items() if key.lower() not in ("ip", "asncode", "asnname") } indicator_context_dict["Geo Location"] = geo_location_dict return_context.append(indicator_context_dict) return return_context def date_string_to_iso_format_parsing(date_str): """Formats a datestring to the ISO-8601 format which the server expects to receive :type date_str: ``str`` :param date_str: String representation of the date. :return: ISO-8601 date string :rtype: ``str`` """ parsed_date_format = dateparser.parse(date_str, date_formats=[SOCRADAR_DATE_FORMAT], settings={"TIMEZONE": "UTC"}) assert parsed_date_format is not None, f"could not parse {date_str}" return parsed_date_format.strftime(DATE_FORMAT) def convert_to_demisto_indicator_type(socradar_indicator_type: str, indicator_value: str = None) -> str: """Maps SOCRadar indicator type to Cortex XSOAR indicator type Converts the SOCRadar indicator types ('hostname', 'url', 'ip', 'hash') to Cortex XSOAR indicator type (Domain, URL, IP, File) for mapping. :type socradar_indicator_type: ``str`` :param socradar_indicator_type: indicator type as returned from the SOCRadar API (str) :type indicator_value: ``str`` :param indicator_value: indicator itself (default None) :return: Cortex XSOAR Indicator Type (Domain, URL, IP, IPv6 File) :rtype: ``str`` """ return { "hostname": FeedIndicatorType.Domain, "url": FeedIndicatorType.URL, "ip": FeedIndicatorType.ip_to_indicator_type(indicator_value) if indicator_value else FeedIndicatorType.IP, "hash": FeedIndicatorType.File, }[socradar_indicator_type] """ CLIENT CLASS """ class Client(BaseClient): """Client class to interact with SOCRadar Threat Intelligence API. Overrides BaseClient.""" def __init__(self, base_url, api_key, tags, tlp_color, verify, proxy): super().__init__(base_url, verify=verify, proxy=proxy) self.api_key = api_key self.tags = tags self.tlp_color = tlp_color def get_collection_indicators(self, collection_name, offset, limit): suffix = "/threat/intelligence/socradar_collections" api_params = {"key": self.api_key, "collection_names": [collection_name], "limit": limit, "offset": offset} response = self._http_request( method="GET", url_suffix=suffix, params=api_params, timeout=60, error_handler=self.handle_error_response ) return response def check_auth(self): suffix = "/threat/intelligence/check/auth" api_params = {"key": self.api_key} response = self._http_request( method="GET", url_suffix=suffix, params=api_params, error_handler=self.handle_error_response ) return response def parse_raw_indicators(self, raw_indicators: list, collection_feed_type: str) -> list: """Creates a list of indicators from a given response :type raw_indicators: ``list`` :param raw_indicators: List of dict that represent the response from the api :type collection_feed_type: ``str`` :param collection_feed_type: Type of the indicators that exist in the collection :return: List of indicators with the correct indicator type. :rtype: ``list`` """ parsed_indicators = [] collection_indicator_type = convert_to_demisto_indicator_type(collection_feed_type) for indicator_dict in raw_indicators: if indicator_dict: indicator = indicator_dict.get("feed", "") indicator_type = ( convert_to_demisto_indicator_type(indicator_dict.get("feed_type", ""), indicator) or collection_indicator_type ) if not indicator_type: indicator_type = auto_detect_indicator_type(indicator) first_seen_date = indicator_dict.get("first_seen_date", "") last_seen_date = indicator_dict.get("latest_seen_date", "") maintainer_name = indicator_dict.get("maintainer_name", "") extra_info = indicator_dict.get("extra_info", {}) indicator_obj = { "type": indicator_type, "value": indicator, "rawJSON": {"value": indicator, "type": indicator_type}, "fields": { "firstseenbysource": date_string_to_iso_format_parsing(first_seen_date), "lastseenbysource": date_string_to_iso_format_parsing(last_seen_date), "collection_maintainer_name": maintainer_name, "extra_info": extra_info, }, } if self.tags: indicator_obj["fields"]["tags"] = self.tags if self.tlp_color: indicator_obj["fields"]["trafficlightprotocol"] = self.tlp_color parsed_indicators.append(indicator_obj) return parsed_indicators def build_iterator(self, collection_name, limit=None, is_check_last_fetch=True) -> List: """Builds a list of indicators. :type collection_name: ``str`` :param collection_name: The name of the collection to fetch indicators from SOCRadar. :type limit: ``int`` :param limit: Maximum number of indicators to fetch. :type is_check_last_fetch: ``bool`` :param is_check_last_fetch: Flag to decide whether the last fetch should be checked or not. :return: A list of JSON objects representing indicators fetched from a feed. :rtype: ``list`` """ parsed_indicators = [] offset = 0 error_count = 0 batch_size = MAX_INDICATOR_FETCH_NUMBER last_fetch_dict = demisto.getIntegrationContext().get("last_fetch", {}) while True: if limit is not None: if offset >= limit: break batch_size = min(limit - offset, MAX_INDICATOR_FETCH_NUMBER) try: raw_response = self.get_collection_indicators(collection_name, offset, batch_size) if raw_response.get("is_success"): collection_dict = raw_response.get("data", {}).get(collection_name, {}) raw_indicators_list = collection_dict.get("collection_data_list", []) collection_date_str = collection_dict.get("collection_date") collection_feed_type = collection_dict.get("collection_feed_type") if not raw_indicators_list: break if is_check_last_fetch: if last_fetch := last_fetch_dict.get(collection_name): collection_date = datetime.strptime(collection_date_str, "%Y-%m-%d").date() last_fetch = datetime.strptime(last_fetch, "%Y-%m-%d").date() if last_fetch >= collection_date: break last_fetch_dict[collection_name] = collection_date_str demisto.setIntegrationContext({"last_fetch": last_fetch_dict}) # type:ignore parsed_indicators.extend( self.parse_raw_indicators(raw_indicators_list, collection_feed_type) ) # list of dict of indicators if len(raw_indicators_list) < batch_size: break offset += batch_size else: error_count += 1 except DemistoException as e: demisto.debug(f"Error while getting indicators. Skipping batch... Error: {e!s}") offset += batch_size error_count += 1 if error_count > 3: break return parsed_indicators @staticmethod def handle_error_response(response) -> None: """Handles API response to display descriptive error messages based on status code :param response: SOCRadar API response. :return: DemistoException for particular error code. """ error_reason = "" try: json_resp = response.json() error_reason = json_resp.get("error") or json_resp.get("message") except JSONDecodeError: pass status_code_messages = { 400: f"{MESSAGES['BAD_REQUEST_ERROR']} Reason: {error_reason}", 401: MESSAGES["AUTHORIZATION_ERROR"], 404: f"{MESSAGES['BAD_REQUEST_ERROR']} Reason: {error_reason}", 429: MESSAGES["RATE_LIMIT_EXCEED_ERROR"], } if response.status_code in status_code_messages: demisto.debug(f"Response Code: {response.status_code}, Reason: {status_code_messages[response.status_code]}") raise DemistoException(status_code_messages[response.status_code]) else: raise DemistoException(response.raise_for_status()) """ COMMAND FUNCTIONS """ def test_module(client: Client, collections_to_fetch: List) -> str: """Tests by building the iterator to check that a proper connection can be established and the feed is accessible with the given parameters. :type client: ``Client`` :param client: client to use :type collections_to_fetch: ``list`` :param collections_to_fetch: Collection names list to fetch indicators from SOCRadar. :return: 'ok' if test passed, anything else will fail the test. :rtype: ``str`` """ client.check_auth() for collection in collections_to_fetch: client.build_iterator(collection, 1, is_check_last_fetch=False) return "ok" def get_indicators_command(client: Client, args: dict[str, str]) -> CommandResults: """Retrieves indicators from the feed to the war-room. :type client: ``Client`` :param client: Client object configured according to instance arguments. :type args: ``Dict[str, Any]`` :param args: Contains all arguments for socradar-get-indicators command. :return: A ``CommandResults`` object that is then passed to ``return_results``. :rtype: ``CommandResults`` """ limit = parse_int_or_raise(args.get("limit", 10)) collections_to_fetch = argToList(args.get("collections_to_fetch")) if "ALL" in collections_to_fetch: collections_to_fetch = SOCRADAR_RECOMMENDED_COLLECTIONS indicators = fetch_indicators(client, collections_to_fetch, limit, is_check_last_fetch=False) context_entry = build_entry_context(indicators) human_readable = tableToMarkdown( f'Indicators from SOCRadar ThreatFeed Collections ({", ".join(collections_to_fetch)}):', context_entry, removeNull=True ) command_results = CommandResults( outputs_prefix="SOCRadarThreatFeed.Indicators", outputs_key_field="value", outputs=context_entry, readable_output=human_readable, raw_response=indicators, ) return command_results def fetch_indicators(client: Client, collections_to_fetch: List, limit=None, is_check_last_fetch=True) -> List[dict]: """Retrieves indicators from the feed to the war-room. :type client: ``Client`` :param client: Client object configured according to instance arguments. :type collections_to_fetch: ``list`` :param collections_to_fetch: Collection names list to fetch indicators from SOCRadar. :type limit: ``int`` :param limit: Maximum number of indicators to fetch. :type is_check_last_fetch: ``bool`` :param is_check_last_fetch: Flag to decide whether the last fetch should be checked or not. :return: Fetched indicators list. :rtype: ``List[Dict]`` """ indicators = [] for collection in collections_to_fetch: collection_indicators = client.build_iterator(collection, limit, is_check_last_fetch) indicators.extend(collection_indicators) return indicators def reset_last_fetch_dict() -> CommandResults: """Reset the last fetch from the integration context :return: A ``CommandResults`` object that is then passed to ``return_results``. :rtype: ``CommandResults`` """ demisto.setIntegrationContext({}) return CommandResults(readable_output="Fetch history has been successfully deleted!") """ MAIN FUNCTION """ def main() -> None: """main function, parses params and runs command functions :return: :rtype: """ params = demisto.params() args = demisto.args() api_key = params.get("apikey") base_url = SOCRADAR_API_ENDPOINT verify_certificate = not params.get("insecure", False) proxy = params.get("proxy", False) feed_tags = argToList(params.get("feedTags")) tlp_color = params.get("tlp_color") collections_to_fetch = argToList(params.get("collections_to_fetch")) if "ALL" in collections_to_fetch: collections_to_fetch = SOCRADAR_RECOMMENDED_COLLECTIONS command = demisto.command() demisto.debug(f"Command being called is {command}") try: client = Client( base_url=base_url, api_key=api_key, tags=feed_tags, tlp_color=tlp_color, verify=verify_certificate, proxy=proxy ) if command == "test-module": return_results(test_module(client, collections_to_fetch)) elif command == "fetch-indicators": indicators = fetch_indicators(client, collections_to_fetch) # Submit indicators in batches for b in batch(indicators, batch_size=2000): demisto.createIndicators(b) # type: ignore elif command == "socradar-get-indicators": return_results(get_indicators_command(client, args)) elif command == "socradar-reset-fetch-indicators": return_results(reset_last_fetch_dict()) 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()