Crowdstrike Falcon Intel Feed

The CrowdStrike intelligence team tracks the activities of threat actor groups and advanced persistent threats (APTs) to understand as much as possible about their known aliases, targets, methods, and more. This integration retrieves indicators from the CrowdStrike Falcon Intel Feed.

Data Enrichment & Threat Intelligence · Crowdstrike Falcon Intel Feed · Feed

Details

ID	`Crowdstrike Falcon Intel Feed`
Provider	CrowdStrike
Category	Data Enrichment & Threat Intelligence
From Version	5.5.0
Docker Image	`demisto/python3:3.12.13.10116658`

README

Configure Crowdstrike Falcon Intel Feed in Cortex

Parameter	Description	Required
feed	Fetch indicators	False
client_id	The Crowdstrike API client ID	True
feedReputation	Indicator reputation	False
feedReliability	Source reliability	True
tlp_color	The Traffic Light Protocol (TLP) designation to apply to indicators fetched from the feed. More information about the protocol can be found at https://us-cert.cisa.gov/tlp	False
feedExpirationPolicy		False
feedExpirationInterval		False
feedIncremental	Incremental feed	False
feedFetchInterval	Feed fetch interval	False
limit	Number of top indicators to fetch from the feed	False
feedBypassExclusionList	Bypass exclusion list	False
insecure	Trust any certificate (not secure)	False
proxy	Use system proxy settings	False
feedTags	Tags	False
client_secret	Crowdstrike API client secret	True
target_industries	Filter by threat actor’s target industries.	False
target_countries	Filter by threat actor’s target countries.	False
custom_filter	A custom filter by which to filter the indicators. If you pass the custom_filter argument it will override the custom_filter parameter from the integration instance configuration.	False

Commands

You can execute these commands from the CLI, as part of an automation, or in a playbook.
After you successfully execute a command, a DBot message appears in the War Room with the command details.

crowdstrike-falcon-intel-get-indicators

Gets indicators from Crowdstrike Falcon Intel Feed.

Base Command

crowdstrike-falcon-intel-get-indicators

Input

Argument Name	Description	Required
limit	The maximum number of indicators to return. Default is 10.	Optional
offset	The index of the first indicator to fetch.	Optional
target_industries	A comma-separated list of the threat actor’s target industries by which to filter the indicators.	Optional
target_countries	A comma-separated list of the threat actor’s target countries by which to filter the indicators.	Optional
custom_filter	A custom filter by which to filter the indicators. If you pass the custom_filter argument it will override the custom_filter parameter from the integration instance configuration. For more information about custom filters and their structure, see the Crowdstrike Falcon documentation (https://falcon.crowdstrike.com/login/?next=%2Fsupport%2Fdocumentation%2F45%2Ffalcon-query-language-fql%3F).	Optional

Context Output

There is no context output for this command.

Command Example

``````

Human Readable Output

Configuration parameters

server_url — Cloud Base URL (e.g., https://api.crowdstrike.com)
feed — Fetch indicators
client_id — CrowdStrike API client ID
feedReputation — Indicator Reputation
feedReliability — Source Reliability (required)
tlp_color — Traffic Light Protocol Color
feedExpirationPolicy —
feedExpirationInterval —
feedIncremental — Incremental feed
feedFetchInterval — Feed Fetch Interval
limit — Number of top indicators to fetch from the feed.
feedBypassExclusionList — Bypass exclusion list
insecure — Trust any certificate (not secure)
proxy — Use system proxy settings
feedTags — Tags
client_secret — CrowdStrike API client secret
credentials_client — CrowdStrike API client ID
target_industries — Filter by the threat actor's target industries.
target_countries — Filter by threat actor's target countries.
custom_filter — A custom filter by which to filter the indicators. If you pass the custom_filter argument it will override the custom_filter parameter from the integration instance configuration.

Commands (1)

crowdstrike-falcon-intel-get-indicators
Gets indicators from CrowdStrike Falcon Intel Feed.

Source

FeedCrowdstrikeFalconIntel.py FeedCrowdstrikeFalconIntel_test.py FeedCrowdstrikeFalconIntel.yml FeedCrowdstrikeFalconIntel_description.md

Download History Raw

# IMPORTS
from datetime import datetime

import demistomock as demisto
import urllib3
from CommonServerPython import *

from CommonServerUserPython import *

urllib3.disable_warnings()

INDICATOR_FIELDS_MAPPER = {
    "stixid": "id",
    "stixaliases": "known_as",
    "stixdescription": "description",
    "stixprimarymotivation": "motivations",
    "aliases": "known_as",
    "description": "description",
    "primarymotivation": "motivations",
    "creationdate": "created_date",
    "updateddate": "last_modified_date",
    "geocountry": "origins",
    "region": "region",
}


class Client(BaseClient):
    def __init__(self, params):
        self._client_id = params.get("credentials_client", {}).get("identifier") or params.get("client_id")
        self._client_secret = params.get("credentials_client", {}).get("password") or params.get("client_secret")
        self._verify_certificate = not demisto.params().get("insecure", False)
        self._server_url = params.get("server_url", "https://api.crowdstrike.com/")
        if not (self._client_id and self._client_secret):
            raise DemistoException("API client ID and API client secret must be provided.")
        super().__init__(
            base_url=self._server_url, verify=self._verify_certificate, ok_codes=(), proxy=params.get("proxy", False)
        )
        self._token = self._get_access_token()
        self._headers = {"Authorization": "Bearer " + self._token}

    @staticmethod
    def _handle_errors(error_entry: dict) -> str:
        errors = error_entry.get("errors", [])
        error_messages = [f"{error['code']}: {error['message']}" for error in errors]
        error_messages_str = "\n".join(error_messages)
        return error_messages_str

    def http_request(
        self,
        method,
        url_suffix,
        full_url=None,
        headers=None,
        json_data=None,
        params=None,
        data=None,
        files=None,
        timeout=10,
        ok_codes=None,
        return_empty_response=False,
        auth=None,
    ):
        return super()._http_request(
            method=method,
            url_suffix=url_suffix,
            full_url=full_url,
            headers=headers,
            json_data=json_data,
            params=params,
            data=data,
            files=files,
            timeout=timeout,
            ok_codes=ok_codes,
            return_empty_response=return_empty_response,
            auth=auth,
        )

    def _get_access_token(self) -> str:
        body = {"client_id": self._client_id, "client_secret": self._client_secret}
        token_res = self.http_request("POST", "/oauth2/token", data=body, auth=(self._client_id, self._client_secret))
        return token_res.get("access_token")

    def create_indicators_from_response(self, response, feed_tags: list, tlp_color: str | None) -> list:
        parsed_indicators = []  # type:List
        indicator = {}
        for actor in response["resources"]:
            if actor:
                fields = {field: actor.get(actor_key) for field, actor_key in INDICATOR_FIELDS_MAPPER.items()}
                fields["tags"] = feed_tags
                if tlp_color:
                    fields["trafficlightprotocol"] = tlp_color

                indicator = {
                    "type": FeedIndicatorType.indicator_type_by_server_version("STIX Threat Actor"),
                    "value": actor.get("name"),
                    "rawJSON": {"type": "STIX Threat Actor", "value": actor.get("name"), "service": "List Actors Feed"},
                    "fields": fields,
                }

                indicator["rawJSON"].update(actor)
            parsed_indicators.append(indicator)

        return parsed_indicators

    def add_target_countries_to_filter(self, country):
        return f'target_countries%3A"{country}"%2B'

    def add_target_industries_to_filter(self, industry):
        return f'target_industries%3A"{industry}"%2B'

    def convert_countries_and_industries_to_url_shape(self, list_of_targets):
        """
        This function converts a list of targets into the form of a URL filter.
        (united states -> united%20states)

        Args:
            list_of_targets: List of countries or industries.

        Returns:
            The URL filter to filter with.
        """
        list_of_targets = list_of_targets.split(",")
        list_of_targets = [target.replace(" ", "%20") for target in list_of_targets if len(target) > 1]
        return list_of_targets

    def build_actors_filter(self, target_countries, target_industries, custom_filter):
        actors_filter = ""
        if custom_filter:
            actors_filter = custom_filter
            return actors_filter

        if target_countries:
            target_countries = self.convert_countries_and_industries_to_url_shape(target_countries)
            for country in target_countries:
                actors_filter += self.add_target_countries_to_filter(country)

        if target_industries:
            target_industries = self.convert_countries_and_industries_to_url_shape(target_industries)
            for industry in target_industries:
                actors_filter += self.add_target_industries_to_filter(industry)

        if actors_filter:
            actors_filter = "&filter=" + actors_filter[:-3]
        return actors_filter

    def build_url_suffix(self, params, actors_filter):
        url_suffix = "/intel/combined/actors/v1?fields=__full__"
        if actors_filter:
            url_suffix = url_suffix + actors_filter
            if params:
                url_suffix = url_suffix + "%2B" + params
        elif params:
            url_suffix = url_suffix + "&filter=" + params
        return url_suffix

    def get_indicators(
        self,
        feed_tags: list,
        tlp_color: str | None,
        limit=None,
        offset=None,
        target_countries=None,
        target_industries=None,
        custom_filter=None,
        time_filter=None,
        sort=None,
    ):
        """Get a list of indicators.
        Returns:
            list. A list of JSON objects representing indicators fetched from a feed.
        """

        params = {}
        if limit:
            params["limit"] = limit
        if offset:
            params["offset"] = offset
        if sort:
            params["sort"] = sort

        actors_filter = self.build_actors_filter(target_countries, target_industries, custom_filter)
        url_suffix_to_filter_by = self.build_url_suffix(time_filter, actors_filter)

        response = self.http_request("GET", url_suffix_to_filter_by, params=params)

        return self.create_indicators_from_response(response, feed_tags, tlp_color)


def test_module(client: Client, args: dict, feed_tags: list, tlp_color: str | None):
    try:
        tags = argToList(demisto.params().get("feedTags"))
        client.get_indicators(tags, tlp_color, limit=1, offset=0)
    except Exception:
        raise Exception("Could not fetch CrowdStrike Feed\n\nCheck your API key and your connection to CrowdStrike.")
    return "ok", {}, {}


def get_indicators_command(client: Client, args: dict, feed_tags: list, tlp_color: str | None) -> tuple[str, dict, list]:
    """Initiate a single fetch-indicators

    Args:
        client(Client): The CrowdStrike Client.
        args(dict): Command arguments.
        feed_tags: The indicator tags.
        tlp_color (str): Traffic Light Protocol color.
    Returns:
        str, dict, list. the markdown table, context JSON and list of indicators
    """
    offset = int(args.get("offset", 0))
    limit = int(args.get("limit", 150))
    target_countries = args.get("target_countries") if args.get("target_countries") else demisto.params().get("target_countries")
    target_industries = (
        args.get("target_industries") if args.get("target_industries") else demisto.params().get("target_industries")
    )
    custom_filter = args.get("custom_filter") if args.get("custom_filter") else demisto.params().get("custom_filter")

    indicators = client.get_indicators(feed_tags, tlp_color, limit, offset, target_countries, target_industries, custom_filter)

    hr_indicators = []
    for indicator in indicators:
        hr_indicators.append(
            {
                "Value": indicator.get("value"),
                "Type": indicator.get("type"),
                "rawJSON": indicator.get("rawJSON"),
                "fields": indicator.get("fields"),
            }
        )

    human_readable = tableToMarkdown(
        "Indicators from CrowdStrike:", hr_indicators, headers=["Value", "Type", "rawJSON", "fields"], removeNull=True
    )

    return human_readable, {}, indicators


def fetch_indicators(
    client: Client,
    feed_tags: list,
    tlp_color: str | None,
    limit: int,
    target_countries=None,
    target_industries=None,
    custom_filter=None,
) -> tuple:
    """Fetch-indicators command from CrowdStrike Feeds

    Args:
        client(Client): CrowdStrike Feed client.
        feed_tags: The indicator tags.
        tlp_color (str): Traffic Light Protocol color.
        limit: limit the amount of indicators fetched.
        target_industries: the actor's target_industries.
        target_countries: the actor's target_countries.
        custom_filter: user actor's filter.
    Returns:
        tuple. (List of indicators, last_run data).
    """
    last_run = demisto.getLastRun() or {}
    offset = int(last_run.get("offset", "0"))

    last_modified_time = last_run.get("last_modified_time")
    time_filter = f"last_modified_date%3A%3E{last_modified_time}" if last_modified_time else None

    indicators = client.get_indicators(
        feed_tags,
        tlp_color,
        limit,
        offset,
        target_countries,
        target_industries,
        custom_filter,
        time_filter=time_filter,
        sort="last_modified_date",
    )

    if len(indicators) >= limit:
        # we need to store the offset and the same last modified time for the next run
        last_run = {"last_modified_time": last_modified_time, "offset": offset + limit}
    elif len(indicators) > 0:
        # we need to store the latest updateddate from the indictators for the next run
        latest_modified_time = max(indicator["fields"]["updateddate"] for indicator in indicators)
        new_last_modified_time = int(latest_modified_time) + 1  # + 1 to avoid get the same
        last_run = {"last_modified_time": new_last_modified_time}
    else:
        # we get 0 new indicators - store the current time
        current_timestamp = datetime.timestamp(datetime.now())
        last_run = {"last_modified_time": int(current_timestamp)}

    return indicators, last_run


def main():
    params = demisto.params()
    feed_tags = argToList(params.get("feedTags"))
    tlp_color = params.get("tlp_color")
    target_countries = params.get("target_countries")
    target_industries = params.get("target_industries")
    custom_filter = params.get("custom_filter")
    fetch_limit = int(params.get("limit", "200"))
    client = Client(params)

    command = demisto.command()
    demisto.info(f"Command being called is {command}")
    # Switch case
    commands = {"test-module": test_module, "crowdstrike-falcon-intel-get-indicators": get_indicators_command}
    try:
        if demisto.command() == "fetch-indicators":
            indicators, last_run_data = fetch_indicators(
                client,
                feed_tags,
                tlp_color,
                target_countries=target_countries,
                target_industries=target_industries,
                custom_filter=custom_filter,
                limit=fetch_limit,
            )
            # we submit the indicators in batches
            for b in batch(indicators, batch_size=2000):
                demisto.createIndicators(b)

            demisto.setLastRun(last_run_data)
        else:
            readable_output, outputs, raw_response = commands[command](client, demisto.args(), feed_tags, tlp_color)  # type: ignore
            return_outputs(readable_output, outputs, raw_response)
    except Exception as e:
        raise Exception(f"Error in CrowdStrike falcon intel Integration [{e}]")


if __name__ in {"__builtin__", "builtins", "__main__"}:
    main()