CrowdStrike Indicator Feed

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()