CrowdStrike Falcon Intel v2

import traceback
from datetime import UTC, datetime
from typing import Any

import demistomock as demisto
import urllib3
from CommonServerPython import *
from dateparser import parse

from CommonServerUserPython import *

# Disable insecure warnings
urllib3.disable_warnings()

""" GLOBAL VARIABLES """

MALICIOUS_DICTIONARY: dict[Any, int] = {
    "low": Common.DBotScore.GOOD,
    "medium": Common.DBotScore.SUSPICIOUS,
    "high": Common.DBotScore.BAD,
}

MALICIOUS_THRESHOLD = MALICIOUS_DICTIONARY.get(demisto.params().get("threshold", "high"))

""" CLIENT """


class Client:
    """
    The integration's client
    """

    def __init__(self, params: dict[str, str], reliability: Optional[DBotScoreReliability] = None):
        self.cs_client: CrowdStrikeClient = CrowdStrikeClient(params=params)
        self.reliability = reliability
        self.query_params: dict[str, str] = {"offset": "offset", "limit": "limit", "sort": "sort", "free_search": "q"}
        self.date_params: dict[str, dict[str, str]] = {
            "created_date": {"operator": "", "api_key": "created_date"},
            "last_updated_date": {"operator": "", "api_key": "last_updated"},
            "max_last_modified_date": {"operator": "<=", "api_key": "last_modified_date"},
            "min_last_activity_date": {"operator": ">=", "api_key": "first_activity_date"},
            "max_last_activity_date": {"operator": "<=", "api_key": "last_activity_date"},
        }

    def build_request_params(self, args: dict[str, Any]) -> dict[str, Any]:
        """
        Build the params dict for the request
        :param args: Cortex XSOAR args
        :return: The params dict
        """
        params: dict[str, Any] = {key: args.get(arg) for arg, key in self.query_params.items()}
        query = args.get("query")
        params["filter"] = query if query else self.build_filter_query(args)
        return assign_params(**params)

    def build_filter_query(self, args: dict[str, str]) -> str:
        """
        Builds the filter query in Falcon Query Language (FQL)
        :param args: Cortex XSOAR args
        :return: The query
        """
        filter_query: str = ""

        for key in args:
            if key not in self.query_params:
                if key not in self.date_params:
                    values: List[str] = argToList(args[key], ",")
                    for value in values:
                        if isinstance(value, str):
                            value = value.replace("'", "%27")
                        filter_query += f"{key}:'{value}'+"
                else:
                    operator: Optional[str] = self.date_params.get(key, {}).get("operator")
                    api_key: Optional[str] = self.date_params.get(key, {}).get("api_key")
                    # Parsing date argument of ISO format or free language into datetime object,
                    # replacing TZ with UTC, taking its timestamp format and rounding it up.
                    parsed_date = parse(args[key])
                    assert parsed_date is not None
                    filter_query += f"{api_key}:{operator}{int(parsed_date.replace(tzinfo=UTC).timestamp())}+"

        if filter_query.endswith("+"):
            filter_query = filter_query[:-1]

        return filter_query

    def get_indicator(self, indicator_value: str) -> dict[str, Any]:
        # crowdstrike do not allow passing single quotes - so we encode them
        # we are not encoding the entire indicator value, as the other reserved chars (such as + and &) are allowed
        indicator_value = indicator_value.replace("'", "%27")
        args: dict[str, Any] = {"indicator": indicator_value, "limit": 1}
        params: dict[str, Any] = self.build_request_params(args)
        return self.cs_client.http_request(method="GET", url_suffix="intel/combined/indicators/v1", params=params)

    def cs_actors(self, args: dict[str, str]) -> dict[str, Any]:
        url_suffix = "intel/combined/actors/v1"
        if argToBoolean(args.pop("display_full_fields", False)):
            url_suffix += "?fields=__full__"
        params: dict[str, Any] = self.build_request_params(args)
        return self.cs_client.http_request(method="GET", url_suffix=url_suffix, params=params)

    def cs_indicators(self, args: dict[str, str]) -> dict[str, Any]:
        params: dict[str, Any] = self.build_request_params(args)
        return self.cs_client.http_request(method="GET", url_suffix="intel/combined/indicators/v1", params=params)

    def cs_reports(self, args: dict[str, str]) -> dict[str, Any]:
        url_suffix = "intel/combined/reports/v1"
        if argToBoolean(args.pop("display_full_fields", False)):
            url_suffix += "?fields=__full__"
        params: dict[str, Any] = self.build_request_params(args)
        return self.cs_client.http_request(method="GET", url_suffix=url_suffix, params=params)


""" HELPER FUNCTIONS """


def get_dbot_score_type(indicator_type: str) -> Exception | DBotScoreType | str:
    """
    Returns the dbot score type
    :param indicator_type: The indicator type
    :return: The dbot score type
    """
    if indicator_type == "ip":
        return DBotScoreType.IP
    elif indicator_type == "domain":
        return DBotScoreType.DOMAIN
    elif indicator_type == "file" or indicator_type == "hash":
        return DBotScoreType.FILE
    elif indicator_type == "url":
        return DBotScoreType.URL
    else:
        raise DemistoException("Indicator type is not supported.")


def get_score_from_resource(r: dict[str, Any]) -> int:
    """
    Calculates the DBotScore for the resource
    :param r: The resource
    :return: The DBotScore
    """
    malicious_confidence: int = MALICIOUS_DICTIONARY.get(r.get("malicious_confidence"), 0)
    if malicious_confidence == 3 or MALICIOUS_THRESHOLD == 1:
        score = 3
    elif malicious_confidence == 2 or MALICIOUS_THRESHOLD == 2:
        score = 2
    else:
        score = 1
    return score


def get_indicator_hash_type(indicator_value: str) -> str | Exception:
    """
    Calculates the type of the hash
    :param indicator_value: The hash value
    :return: The hash type
    """
    length: int = len(indicator_value)
    if length == 32:
        return "hash_md5"
    elif length == 40:
        return "hash_sha1"
    elif length == 64:
        return "hash_sha256"
    else:
        raise DemistoException(
            f"Invalid hash. Hash length is: {length}. Please provide either MD5 (32 length)"
            f", SHA1 (40 length) or SHA256 (64 length) hash."
        )


def get_indicator_object(
    indicator_value: Any, indicator_type: str, dbot_score: Common.DBotScore
) -> Common.IP | Common.URL | Common.File | Common.Domain | None:
    """
    Returns the corresponding indicator common object
    :param indicator_value: The indicator value
    :param indicator_type: The indicator value
    :param dbot_score: The indicator DBotScore
    :return: The indicator common object
    """
    if indicator_type == "ip":
        return Common.IP(ip=indicator_value, dbot_score=dbot_score)
    elif indicator_type == "url":
        return Common.URL(url=indicator_value, dbot_score=dbot_score)
    elif indicator_type == "hash":
        hash_type: str | Exception = get_indicator_hash_type(indicator_value)
        if hash_type == "hash_md5":
            return Common.File(md5=indicator_value, dbot_score=dbot_score)
        elif hash_type == "hash_sha1":
            return Common.File(sha1=indicator_value, dbot_score=dbot_score)
        else:
            return Common.File(sha256=indicator_value, dbot_score=dbot_score)
    elif indicator_type == "domain":
        return Common.Domain(domain=indicator_value, dbot_score=dbot_score)
    else:
        return None


def should_filter_resource_by_type(resource, indicator_type, indicator_value):
    """
    checks if a resource should be filtered by his type.
    :param resource: The resource object
    :param indicator_type: The indicator type
    :param indicator_value: The indicator value
    :return: True if the resource should be filtered (don't match the indicator type) or False otherwise.
    """
    # indicator type was not filtered using the query due to a bug in the CrowdStrike API.
    if indicator_type == "hash":
        filter_type = get_indicator_hash_type(indicator_value)
    elif indicator_type == "ip":
        filter_type = "ip_address"
    else:
        filter_type = indicator_type

    return resource.get("type") != filter_type


def build_indicator(indicator_value: str, indicator_type: str, title: str, client: Client) -> List[CommandResults]:
    """
    Builds an indicator entry
    :param indicator_value: The indicator value
    :param indicator_type: The indicator type
    :param title: The title to show to the user
    :param client: The integration's client
    :return: The indicator entry
    """
    res: dict[str, Any] = client.get_indicator(indicator_value)
    resources: List[Any] = res.get("resources", [])
    results: List[CommandResults] = []

    if resources:
        for r in resources:
            if should_filter_resource_by_type(r, indicator_type, indicator_value):
                continue
            output = get_indicator_outputs(r)
            score = get_score_from_resource(r)
            dbot_score = Common.DBotScore(
                indicator=indicator_value,
                indicator_type=get_dbot_score_type(indicator_type),
                integration_name="CrowdStrike Falcon Intel v2",
                malicious_description="High confidence",
                score=score,
                reliability=client.reliability,
            )
            indicator = get_indicator_object(indicator_value, indicator_type, dbot_score)
            results.append(
                CommandResults(
                    outputs=output,
                    outputs_prefix="FalconIntel.Indicator",
                    outputs_key_field="ID",
                    indicator=indicator,
                    readable_output=tableToMarkdown(name=title, t=output, headerTransform=pascalToSpace),
                    raw_response=res,
                )
            )

    else:
        results.append(
            create_indicator_result_with_dbotscore_unknown(
                indicator=indicator_value, indicator_type=DBotScoreType.FILE if indicator_type == "hash" else indicator_type
            )
        )

    return results


def get_values(items_list: List[Any], return_type: str = "str", keys: str | List[Any] = "value") -> str | List[str | dict]:
    """
    Returns the values of list's items
    :param items_list: The items list
    :param return_type: Whether to return string or list
    :param keys: The key to get the data
    :return: The values list
    """
    new_list: List[Any] = []
    if not items_list:
        return new_list
    if isinstance(keys, str):
        new_list = [item.get(keys) for item in items_list]
    elif isinstance(keys, list):
        new_list = [{underscoreToCamelCase(f): item.get(f) for f in item if f in keys} for item in items_list]
    if return_type == "list":
        return new_list
    return ", ".join(str(item) for item in new_list)


def get_indicator_outputs(resource: dict[str, Any]) -> dict[str, Any]:
    """
    Build the output and extra context of an indicator
    :param resource: The indicator's object
    :return: The indicator's human readable
    """
    output: dict[str, Any] = {}

    if resource:
        indicator_id = resource.get("id")
        indicator_value = resource.get("indicator")
        indicator_type = resource.get("type")
        last_update = resource.get("last_update")
        publish_date = resource.get("publish_date")
        malicious_confidence = resource.get("malicious_confidence")
        reports = resource.get("reports")
        actors = resource.get("actors")
        malware_families = resource.get("malware_families")
        kill_chains = resource.get("kill_chains")
        domain_types = resource.get("domain_types")
        ip_address_types = resource.get("ip_address_types")
        relations: List[Any] = resource.get("relations", [])[:10]
        labels: List[Any] = resource.get("labels", [])[:10]

        output = assign_params(
            ID=indicator_id,
            Type=indicator_type,
            Value=indicator_value,
            LastUpdate=datetime.fromtimestamp(last_update, UTC).isoformat() if last_update else None,
            PublishDate=datetime.fromtimestamp(publish_date, UTC).isoformat() if publish_date else None,
            MaliciousConfidence=malicious_confidence,
            Reports=reports,
            Actors=actors,
            MalwareFamilies=malware_families,
            KillChains=kill_chains,
            DomainTypes=domain_types,
            IPAddressTypes=ip_address_types,
            Relations=[
                f'{item.get("Type")}: {item.get("Indicator")}'  # type: ignore
                for item in get_values(relations, return_type="list", keys=["indicator", "type"])  # type: ignore
            ],
            Labels=get_values(labels, return_type="list", keys="name"),
        )

    return output


""" COMMANDS """


def run_test_module(client: Client) -> str | Exception:
    """
    If a client is successfully constructed then an access token was successfully created,
    therefore the username and password are valid and a connection was made.
    On top of the above, this function validates the http request to indicators endpoint.
    :param client: the client object with an access token
    :return: ok if got a valid access token and not all the quota is used at the moment
    """
    client.cs_client.http_request("GET", "intel/combined/indicators/v1", params={"limit": 1})
    return "ok"


def file_command(files: List, client: Client) -> List[CommandResults]:
    results: List[CommandResults] = []
    for file in files:
        results += build_indicator(file, "hash", "Falcon Intel file reputation:\n", client)
    return results


def ip_command(ips: List, client: Client) -> List[CommandResults]:
    results: List[CommandResults] = []
    for ip in ips:
        results += build_indicator(ip, "ip", "Falcon Intel IP reputation:\n", client)
    return results


def url_command(urls: List, client: Client) -> List[CommandResults]:
    results: List[CommandResults] = []
    for url in urls:
        results += build_indicator(url, "url", "Falcon Intel URL reputation:\n", client)
    return results


def domain_command(domains: List, client: Client) -> List[CommandResults]:
    results: List[CommandResults] = []
    for domain in domains:
        results += build_indicator(domain, "domain", "Falcon Intel domain reputation:\n", client)
    return results


def cs_actors_command(client: Client, args: dict[str, str]) -> CommandResults:
    res: dict[str, Any] = client.cs_actors(args)
    resources: List[Any] = res.get("resources", [])
    outputs: List[dict[str, Any]] = []
    md_outputs: List[dict[str, Any]] = []
    md: str = ""
    title: str = "Falcon Intel Actor search:"

    if resources:
        for r in resources:
            image_url = r.get("image", {}).get("url")
            name = r.get("name")
            actor_id = r.get("id")
            url = r.get("url")
            slug = r.get("slug")
            short_description = r.get("short_description")
            description = r.get("description")
            first_activity_date = r.get("first_activity_date")
            last_activity_date = r.get("last_activity_date")
            active = r.get("active")
            known_as = r.get("known_as")
            target_industries = r.get("target_industries") or []
            target_countries = r.get("target_countries") or []
            origins = r.get("origins") or []
            motivations = r.get("motivations") or []
            capability = r.get("capability", {}).get("value")
            group = r.get("group")
            region = r.get("region", {}).get("value")
            kill_chain = r.get("kill_chain")

            output: dict[str, Any] = assign_params(
                ImageURL=image_url,
                Name=name,
                ID=actor_id,
                URL=url,
                Slug=slug,
                ShortDescription=short_description,
                Description=description,
                FirstActivityDate=datetime.fromtimestamp(first_activity_date, UTC).isoformat() if first_activity_date else None,
                LastActivityDate=datetime.fromtimestamp(last_activity_date, UTC).isoformat() if last_activity_date else None,
                Active=active,
                KnownAs=known_as,
                TargetIndustries=get_values(target_industries, return_type="list"),
                TargetCountries=get_values(target_countries, return_type="list"),
                Origins=get_values(origins, return_type="list"),
                Motivations=get_values(motivations, return_type="list"),
                Capability=capability,
                Group=group,
                Region=region,
                KillChains=kill_chain,
            )
            outputs.append(output)

            md_output: dict[str, Any] = output
            for key in ("URL", "ImageURL"):
                if key in md_output:
                    value = md_output[key]
                    md_output[key] = f"[{value}]({value})"

            md_outputs.append(md_output)
    else:
        md = "No actors found."

    results: CommandResults = CommandResults(
        outputs=outputs,
        outputs_key_field="ID",
        outputs_prefix="FalconIntel.Actor",
        readable_output=md if md else tableToMarkdown(name=title, t=md_outputs, headerTransform=pascalToSpace, removeNull=True),
        raw_response=res,
    )

    return results


def cs_indicators_command(client: Client, args: dict[str, str]) -> List[CommandResults]:
    res: dict[str, Any] = client.cs_indicators(args)
    resources: List[Any] = res.get("resources", [])
    results: List[CommandResults] = []
    title: str = "Falcon Intel Indicator search:"

    if resources:
        for r in resources:
            output = get_indicator_outputs(r)
            indicator_value = output.get("Value")
            indicator_type = output.get("Type")
            indicator: Optional[Common.Indicator] = None

            if indicator_type in ("hash_md5", "hash_sha256", "hash_sha1", "ip_address", "url", "domain"):
                if indicator_type in ("hash_md5", "hash_sha1", "hash_sha256"):
                    indicator_type = "hash"
                elif indicator_type == "ip_address":
                    indicator_type = "ip"
                score = get_score_from_resource(r)
                dbot_score = Common.DBotScore(
                    indicator=indicator_value,
                    indicator_type=get_dbot_score_type(indicator_type),
                    integration_name="CrowdStrike Falcon Intel v2",
                    malicious_description="High confidence",
                    score=score,
                    reliability=client.reliability,
                )
                indicator = get_indicator_object(indicator_value, indicator_type, dbot_score)
            results.append(
                CommandResults(
                    outputs=output,
                    outputs_prefix="FalconIntel.Indicator",
                    outputs_key_field="ID",
                    readable_output=tableToMarkdown(name=title, t=output, headerTransform=pascalToSpace),
                    raw_response=res,
                    indicator=indicator,
                )
            )
    else:
        results.append(CommandResults(readable_output="No indicators found."))

    return results


def cs_reports_command(client: Client, args: dict[str, str]) -> CommandResults:
    res: dict[str, Any] = client.cs_reports(args)
    resources: List[Any] = res.get("resources", [])
    outputs: List[dict[str, Any]] = []
    md_outputs: List[dict[str, Any]] = []
    md: str = ""
    title: str = "Falcon Intel Report search:"

    if resources:
        for r in resources:
            report_id: int = r.get("id")
            url: str = r.get("url")
            name: str = r.get("name")
            report_type: str = r.get("type", {}).get("name")
            sub_type: str = r.get("sub_type", {}).get("name")
            slug: str = r.get("slug")
            created_date: int = r.get("created_date")
            last_modified_date: int = r.get("last_modified_date")
            short_description: str = r.get("short_description")
            description: str = r.get("description")
            target_industries: List[Any] = r.get("target_industries") or []
            target_countries: List[Any] = r.get("target_countries") or []
            motivations: List[Any] = r.get("motivations") or []
            tags: List[Any] = r.get("tags") or []
            actors: List[Any] = r.get("actors") or []

            output: dict[str, Any] = assign_params(
                ID=report_id,
                URL=url,
                Name=name,
                Type=report_type,
                SubType=sub_type,
                Slug=slug,
                CreatedDate=datetime.fromtimestamp(created_date, UTC).isoformat() if created_date else None,
                LastModifiedSate=datetime.fromtimestamp(last_modified_date, UTC).isoformat() if last_modified_date else None,
                ShortDescription=short_description,
                Description=description,
                TargetIndustries=get_values(target_industries, return_type="list"),
                TargetCountries=get_values(target_countries, return_type="list"),
                Motivations=get_values(motivations, return_type="list"),
                Tags=get_values(tags, return_type="list"),
                Actors=get_values(actors, return_type="list", keys="name"),
            )
            outputs.append(output)

            md_output: dict[str, Any] = output
            if "URL" in md_output:
                value = md_output["URL"]
                md_output["URL"] = f"[{value}]({value})"

            md_outputs.append(md_output)

    else:
        md = "No reports found."

    results: CommandResults = CommandResults(
        outputs_prefix="FalconIntel.Report",
        outputs=outputs,
        outputs_key_field="ID",
        readable_output=md if md else tableToMarkdown(name=title, t=outputs, headerTransform=pascalToSpace, removeNull=True),
        raw_response=res,
    )

    return results


def main():
    params: dict[str, str] = demisto.params()

    reliability = params.get("integrationReliability", "C - Fairly reliable")
    reliability = (
        DBotScoreReliability.get_dbot_score_reliability_from_str(reliability)
        if DBotScoreReliability.is_valid_type(reliability)
        else None
    )

    args: dict[str, str] = demisto.args()
    results: CommandResults | List[CommandResults]
    try:
        command: str = demisto.command()
        LOG(f"Command being called in CrowdStrike Falcon Intel v2 is: {command}")
        client: Client = Client(params=params, reliability=reliability)
        if command == "test-module":
            result: str | Exception = run_test_module(client)
            return_results(result)
        elif command == "file":
            results = file_command(argToList(args["file"]), client)
            return_results(results)
        elif command == "ip":
            results = ip_command(argToList(args["ip"]), client)
            return_results(results)
        elif command == "url":
            results = url_command(argToList(args["url"]), client)
            return_results(results)
        elif command == "domain":
            results = domain_command(argToList(args["domain"]), client)
            return_results(results)
        elif command == "cs-actors":
            results = cs_actors_command(client, args)
            return_results(results)
        elif command == "cs-indicators":
            results = cs_indicators_command(client, args)
            return_results(results)
        elif command == "cs-reports":
            results = cs_reports_command(client, args)
            return_results(results)
        else:
            raise NotImplementedError(f"{command} command is not an existing CrowdStrike Falcon Intel v2 integration")
    except Exception as err:
        return_error(f"Unexpected error:\n{err!s}", error=traceback.format_exc())


from CrowdStrikeApiModule import *  # noqa: E402

if __name__ in ("__main__", "builtin", "builtins"):
    main()