import copy import demistomock as demisto # noqa: F401 # IMPORTS import urllib3 from CommonServerPython import * # noqa: F401 from CrowdStrikeApiModule import * # noqa: E402 from CommonServerUserPython import * # noqa urllib3.disable_warnings() XSOAR_TYPES_TO_CROWDSTRIKE = { "account": "username", "domain": "domain", "email": "email_address", "file md5": "hash_md5", "file sha-256": "hash_sha256", "ip": "ip_address", "registry key": "registry", "url": "url", } CROWDSTRIKE_TO_XSOAR_TYPES = { "username": FeedIndicatorType.Account, "domain": FeedIndicatorType.Domain, "email_address": FeedIndicatorType.Email, "hash_md5": FeedIndicatorType.File, "hash_sha1": FeedIndicatorType.File, "hash_sha256": FeedIndicatorType.File, "registry": FeedIndicatorType.Registry, "url": FeedIndicatorType.URL, "ip_address": FeedIndicatorType.IP, "reports": ThreatIntel.ObjectsNames.REPORT, "actors": ThreatIntel.ObjectsNames.THREAT_ACTOR, "malware_families": ThreatIntel.ObjectsNames.MALWARE, "vulnerabilities": FeedIndicatorType.CVE, } INDICATOR_TO_CROWDSTRIKE_RELATION_DICT: Dict[str, Any] = { ThreatIntel.ObjectsNames.REPORT: { FeedIndicatorType.File: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.IP: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.Domain: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.URL: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.Email: EntityRelationship.Relationships.RELATED_TO, FeedIndicatorType.Registry: EntityRelationship.Relationships.RELATED_TO, FeedIndicatorType.Account: EntityRelationship.Relationships.RELATED_TO, }, ThreatIntel.ObjectsNames.THREAT_ACTOR: { FeedIndicatorType.File: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.IP: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.Domain: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.URL: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.Email: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.Registry: EntityRelationship.Relationships.RELATED_TO, FeedIndicatorType.Account: EntityRelationship.Relationships.RELATED_TO, }, ThreatIntel.ObjectsNames.MALWARE: { FeedIndicatorType.File: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.IP: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.Domain: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.URL: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.Email: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.Registry: EntityRelationship.Relationships.RELATED_TO, FeedIndicatorType.Account: EntityRelationship.Relationships.RELATED_TO, }, FeedIndicatorType.CVE: { FeedIndicatorType.File: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.IP: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.Domain: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.URL: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.Email: EntityRelationship.Relationships.INDICATOR_OF, FeedIndicatorType.Registry: EntityRelationship.Relationships.RELATED_TO, FeedIndicatorType.Account: EntityRelationship.Relationships.RELATED_TO, }, } CROWDSTRIKE_INDICATOR_RELATION_FIELDS = ["reports", "actors", "malware_families", "vulnerabilities", "relations"] def kill_chain_standard_values(phases: list | None): """ Will convert crowdstrike's return values for kill chain to our standard kill chain syntax. Args: phase: the raw inout from the api Returns: the standardized value, or the given value if not found """ if not phases: return phases return [ { "reconnaissance": "Reconnaissance", "weaponization": "Weaponization", "installation": "Installation", "exploitation": "Exploitation", "delivery": "Delivery", "c2": "Command & Control", "actionOnObjectives": "Actions on Objectives", }.get(phase.lower(), phase) for phase in phases ] class Client(CrowdStrikeClient): def __init__( self, credentials, base_url, include_deleted, type, limit, tlp_color=None, feed_tags=None, malicious_confidence=None, filter_string=None, generic_phrase=None, insecure=True, proxy=False, first_fetch=None, create_relationships=True, timeout="10", ): params = assign_params( credentials=credentials, server_url=base_url, insecure=insecure, ok_codes=(), proxy=proxy, timeout=timeout ) super().__init__(params) self.type = type self.malicious_confidence = malicious_confidence self.filter_string = filter_string self.generic_phrase = generic_phrase self.include_deleted = include_deleted self.tlp_color = tlp_color self.feed_tags = feed_tags self.limit = limit self.first_fetch = first_fetch self.create_relationships = create_relationships def get_indicators(self, params): response = super().http_request( method="GET", params=params, url_suffix="intel/combined/indicators/v1", timeout=30, ok_codes=(200, 401) ) if response.get("errors") and response["errors"][0].get("code") == 401: demisto.info(f'request failed with status code is 401, error: {str(response["errors"][0])}, regenerate token') self._token = self._get_token() self._headers = {"Authorization": "bearer " + self._token} response = super().http_request(method="GET", params=params, url_suffix="intel/combined/indicators/v1", timeout=30) return response def get_actors_names_request(self, params_string): response = self._http_request( method="GET", url_suffix=f"intel/entities/actors/v1?{params_string}", timeout=30, ok_codes=(200, 401) ) if response.get("errors") and response["errors"][0].get("code") == 401: demisto.info(f'request failed with status code is 401, error: {str(response["errors"][0])}, regenerate token') self._token = self._get_token() self._headers = {"Authorization": "bearer " + self._token} response = self._http_request(method="GET", url_suffix=f"intel/entities/actors/v1?{params_string}", timeout=30) if "resources" not in response: raise DemistoException("Get actors request completed. Parse error: could not find resources in response.") return response["resources"] def fetch_indicators(self, limit: int, offset: int = 0, fetch_command: bool = False, manual_last_run: int = 0) -> list: """Get indicators from CrowdStrike API Args: limit(int): number of indicators to return offset: indicators offset fetch_command: In order not to update last_run time if it is not fetch command manual_last_run: The minimum timestamp to fetch indicators by Returns: (list): parsed indicators """ indicators: list[dict] = [] filter_string = f"({self.filter_string})" if self.filter_string else "" if self.type: type_fql = self.build_type_fql(self.type) filter_string = f"({type_fql})+{filter_string}" if filter_string else f"({type_fql})" if self.malicious_confidence: malicious_confidence_fql = ",".join([f"malicious_confidence:'{item}'" for item in self.malicious_confidence]) filter_string = f"{filter_string}+({malicious_confidence_fql})" if filter_string else f"({malicious_confidence_fql})" if manual_last_run: filter_string = ( f"{filter_string}+(last_updated:>={manual_last_run})" if filter_string else f"(last_updated:>={manual_last_run})" ) if fetch_command: if last_run := self.get_last_run(): filter_string = f"{filter_string}+({last_run})" if filter_string else f"({last_run})" else: filter_string, indicators = self.handle_first_fetch_context_or_pre_2_1_0(filter_string) if indicators: limit = limit - len(indicators) if filter_string or not fetch_command: demisto.debug(f"{filter_string=}") params = assign_params( include_deleted=self.include_deleted, limit=limit, offset=offset, q=self.generic_phrase, filter=filter_string, sort="_marker|asc", ) response = self.get_indicators(params=params) # need to fetch all indicators after the limit if resources := response.get("resources", []): new_last_marker_time = resources[-1].get("_marker") else: new_last_marker_time = demisto.getIntegrationContext().get("last_marker_time") last_marker_time_for_debug = new_last_marker_time or "No data yet" demisto.debug( f"There are no indicators, using last_marker_time={last_marker_time_for_debug} from Integration Context" ) if fetch_command: context = demisto.getIntegrationContext() demisto.info(f"last_marker_time before updating: {context.get('last_marker_time')}") context.update({"last_marker_time": new_last_marker_time}) demisto.setIntegrationContext(context) demisto.info(f"set last_run to: {new_last_marker_time}") indicators.extend( self.create_indicators_from_response( response, self.get_actors_names_request, self.tlp_color, self.feed_tags, self.create_relationships, ) ) return indicators def handle_first_fetch_context_or_pre_2_1_0(self, filter_string: str) -> tuple[str, list[dict]]: """ Checks whether the context integration uses the format used up to version 2_1_0 (when the `last_update` parameter was removed), or whether this is the first time of the fetch, If so, the function imports one indicator and extracts the `_marker` from it to import the following indicators. The function is only called in the following two cases: 1. At the first run of v2.1.0 or newer. 2. In order to transfer the context integration to the new implementation Returns: Tuple: 1. filter_string with the _marker key - str. 2. parse indicator that retrieved - list[dict]. """ filter_for_first_fetch = filter_string if last_run := demisto.getIntegrationContext().get("last_updated") or self.first_fetch: last_run = f"last_updated:>={int(last_run)}" filter_for_first_fetch = f"{filter_string}+({last_run})" if filter_string else f"({last_run})" params = assign_params( include_deleted=self.include_deleted, limit=1, q=self.generic_phrase, filter=filter_for_first_fetch, sort="last_updated|asc", ) response = self.get_indicators(params=params) # In case there is an indicator for extracting the `_marker` # it allows fetching following indicators better. if resources := response.get("resources", []): _marker = resources[-1].get("_marker") demisto.debug(f"Importing the indicator marker in first time: {_marker=}") last_run = f"_marker:>'{_marker}'" parse_indicator = self.create_indicators_from_response( response, self.get_actors_names_request, self.tlp_color, self.feed_tags, self.create_relationships ) filter_string = f"{filter_string}+({last_run})" if filter_string else f"({last_run})" return filter_string, parse_indicator # In case no indicator is returned demisto.debug("No indicator returned") return "", [] @staticmethod def get_last_run() -> str: """Gets last run time in timestamp Returns: last run in timestamp, or '' if no last run. Taken from Integration Context key last_marker_time. """ if last_run := demisto.getIntegrationContext().get("last_marker_time"): demisto.info(f"get last_run: {last_run}") params = f"_marker:>'{last_run}'" else: demisto.debug("There is no last_run (last_marker_time in Integration Context)") params = "" return params @staticmethod def create_indicators_from_response( raw_response, get_actors_names_request_func, tlp_color=None, feed_tags=None, create_relationships=True ) -> list: """Builds indicators from API raw response Args: raw_response: response from crowdstrike API tlp_color: tlp color chosen by customer feed_tags: Feed tags to filter by create_relationships: Whether to create relationships. Returns: (list): list of indicators """ parsed_indicators: list = [] indicator: dict = {} for resource in raw_response["resources"]: if not (type_ := auto_detect_indicator_type_from_cs(resource["indicator"], resource["type"])): demisto.debug(f"Indicator {resource['indicator']} of type {resource['type']} is not supported in XSOAR, skipping") continue indicator = { "type": type_, "value": resource.get("indicator"), "rawJSON": resource, "fields": { "actor": resource.get("actors"), "reports": resource.get("reports"), "malwarefamily": resource.get("malware_families"), "stixkillchainphases": kill_chain_standard_values(resource.get("kill_chains")), "ipaddress": resource.get("ip_address_types"), "domainname": resource.get("domain_types"), "targets": resource.get("targets"), "threattypes": [{"threatcategory": threat} for threat in resource.get("threat_types", [])], "vulnerabilities": resource.get("vulnerabilities"), "confidence": resource.get("malicious_confidence"), "updateddate": resource.get("last_updated"), "creationdate": resource.get("published_date"), "tags": [label.get("name") for label in resource.get("labels")], # type: ignore }, } if tlp_color: indicator["fields"]["trafficlightprotocol"] = tlp_color if feed_tags: indicator["fields"]["tags"].extend(feed_tags) if create_relationships: relationships = create_and_add_relationships(indicator, resource, get_actors_names_request_func) indicator["relationships"] = relationships parsed_indicators.append(indicator) return parsed_indicators @staticmethod def build_type_fql(types_list: list) -> str: """Builds an indicator type query for the filter parameter Args: types_list(list): indicator types that was chosen by user Returns: (str): FQL query containing the relevant indicator types we want to fetch from Crowdstrike """ if "ALL" in types_list: # Replaces "ALL" for all types supported on XSOAR. crowdstrike_types = [f"type:'{type}'" for type in CROWDSTRIKE_TO_XSOAR_TYPES] else: crowdstrike_types = [ f"type:'{XSOAR_TYPES_TO_CROWDSTRIKE.get(type.lower())}'" for type in types_list if type.lower() in XSOAR_TYPES_TO_CROWDSTRIKE ] result = ",".join(crowdstrike_types) return result def create_and_add_relationships(indicator: dict, resource: dict, get_actors_names_request_func) -> list: """ Creates and adds relationships to indicators for each CrowdStrike relationships type. Args: indicator(dict): The indicator in XSOAR format. resource(dict): The indicator from the response. Returns: List of relationships objects. """ relationships = [] for field in CROWDSTRIKE_INDICATOR_RELATION_FIELDS: if resource.get(field): relationships.extend(create_relationships(field, indicator, resource, get_actors_names_request_func)) return relationships def create_relationships(field: str, indicator: dict, resource: dict, get_actors_names_request_func) -> List: """ Creates indicator relationships. Args: field(str): A CrowdStrike indicator field which contains relationships. indicator(dict): The indicator in XSOAR format. resource(dict): The indicator from the response. Returns: List of relationships objects. """ relationships = [] if field == "actors" and resource["actors"]: resource["actors"] = change_actors_from_id_to_name(resource["actors"], get_actors_names_request_func) for relation in resource[field]: if field == "relations" and not CROWDSTRIKE_TO_XSOAR_TYPES.get(relation.get("type")): demisto.debug(f"The related indicator type {relation.get('type')} is not supported in XSOAR.") continue if field == "relations": related_indicator_type = auto_detect_indicator_type_from_cs(relation["indicator"], relation["type"]) relation_name = EntityRelationship.Relationships.RELATED_TO else: related_indicator_type = CROWDSTRIKE_TO_XSOAR_TYPES[field] relation_name = INDICATOR_TO_CROWDSTRIKE_RELATION_DICT[related_indicator_type].get( indicator["type"], indicator["type"] ) indicator_relation = EntityRelationship( name=relation_name, entity_a=indicator["value"], entity_a_type=indicator["type"], entity_b=relation["indicator"] if field == "relations" else relation, entity_b_type=related_indicator_type, reverse_name=EntityRelationship.Relationships.RELATIONSHIPS_NAMES.get(relation_name, ""), ).to_indicator() relationships.append(indicator_relation) return relationships def change_actors_from_id_to_name(indicator_actors_array: List[str], get_name_of_actors__func): integration_context = get_integration_context() actors_to_convert = [] converted_actors_array = [] for actor in indicator_actors_array: if converted_actor := integration_context.get(actor, None): converted_actors_array.append(converted_actor) else: actors_to_convert.append(actor) if actors_to_convert: actor_ids_params = "ids=" + "&ids=".join(actors_to_convert) + "&fields=name" actors_response = get_name_of_actors__func(actor_ids_params) converted_actors_from_request = [] for actor_dict in actors_response: converted_actors_from_request.append(actor_dict.get("name")) zipped_actors_list_to_context = dict(zip(actors_to_convert, converted_actors_from_request)) update_integration_context(zipped_actors_list_to_context) converted_actors_array += converted_actors_from_request return converted_actors_array def auto_detect_indicator_type_from_cs(value: str, crowdstrike_resource_type: str) -> str | None: """ The function determines the type of indicator according to two cases:: 1. In case the type is ip_address then the type is detected by auto_detect_indicator_type function (CSP). 2. In any other case, the type is converted by the table CROWDSTRIKE_TO_XSOAR_TYPES to a type of XSOAR. """ if crowdstrike_resource_type == "ip_address": return auto_detect_indicator_type(value) return CROWDSTRIKE_TO_XSOAR_TYPES.get(crowdstrike_resource_type) def fetch_indicators_command(client: Client): """fetch indicators from the Crowdstrike Intel Args: client: Client object Returns: list of indicators(list) """ parsed_indicators = client.fetch_indicators(fetch_command=True, limit=client.limit) # we submit the indicators in batches for b in batch(parsed_indicators, batch_size=2000): demisto.createIndicators(b) return parsed_indicators def crowdstrike_indicators_list_command(client: Client, args: dict) -> CommandResults: """Gets indicator from Crowdstrike Intel to readable output Args: client: Client object args: demisto.args() Returns: readable_output, raw_response """ offset = arg_to_number(args.get("offset", 0)) or 0 limit = arg_to_number(args.get("limit", 50)) or 50 last_run = arg_to_number(args.get("last_run", 0)) or 0 parsed_indicators = client.fetch_indicators(limit=limit, offset=offset, fetch_command=False, manual_last_run=last_run) if outputs := copy.deepcopy(parsed_indicators): for indicator in outputs: indicator["id"] = indicator.get("rawJSON", {}).get("id") readable_output = tableToMarkdown( name="Indicators from CrowdStrike Falcon Intel", t=outputs, headers=["type", "value", "id"], headerTransform=pascalToSpace, ) return CommandResults( outputs=outputs, outputs_prefix="CrowdStrikeFalconIntel.Indicators", outputs_key_field="id", readable_output=readable_output, raw_response=parsed_indicators, ) else: return CommandResults(readable_output="No Indicators.") def test_module(client: Client, args: dict) -> str: try: client.fetch_indicators(limit=client.limit, fetch_command=False) except Exception: raise Exception("Could not fetch CrowdStrike Indicator Feed\n\nCheck your API key and your connection to CrowdStrike.") return "ok" def reset_last_run(): """ Reset the last run from the integration context """ demisto.setIntegrationContext({}) return CommandResults(readable_output="Fetch history deleted successfully") """ MAIN FUNCTION """ def main() -> None: params = demisto.params() credentials = params.get("credentials") proxy = params.get("proxy", False) insecure = params.get("insecure", False) first_fetch_param = params.get("first_fetch") first_fetch_datetime = arg_to_datetime(first_fetch_param) if first_fetch_param else None first_fetch = first_fetch_datetime.timestamp() if first_fetch_datetime else None base_url = params.get("base_url") tlp_color = params.get("tlp_color") include_deleted = params.get("include_deleted", False) type = argToList(params.get("type"), "ALL") malicious_confidence = argToList(params.get("malicious_confidence")) filter_string = params.get("filter") generic_phrase = params.get("generic_phrase") max_fetch = arg_to_number(params.get("max_indicator_to_fetch")) if params.get("max_indicator_to_fetch") else 10000 max_fetch = min(max_fetch, 10000) # type: ignore feed_tags = argToList(params.get("feedTags")) create_relationships = params.get("create_relationships", True) timeout = params.get("timeout") args = demisto.args() try: command = demisto.command() demisto.info(f"Command being called is {demisto.command()}") client = Client( credentials=credentials, base_url=base_url, insecure=insecure, proxy=proxy, tlp_color=tlp_color, feed_tags=feed_tags, include_deleted=include_deleted, type=type, malicious_confidence=malicious_confidence, filter_string=filter_string, generic_phrase=generic_phrase, limit=max_fetch, first_fetch=first_fetch, create_relationships=create_relationships, timeout=timeout, ) if command == "test-module": # This is the call made when pressing the integration Test button. result = test_module(client, args) return_results(result) elif command == "fetch-indicators": fetch_indicators_command(client=client) elif command == "crowdstrike-indicators-list": return_results(crowdstrike_indicators_list_command(client, args)) elif command == "crowdstrike-reset-fetch-indicators": return_results(reset_last_run()) # Log exceptions and return errors 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()