CrowdStrikeStreamingPreProcessing

Pre processing script for CrowdStrike Streaming, will not duplicate incidents(detection events) that have same Host. Will add entry to duplicate(older) incident notifying a duplicate incident was ignored. This automation runs using the default Limited User role, unless you explicitly change the permissions. For more information, see the section about permissions here: - For Cortex XSOAR 6 see https://docs-cortex.paloaltonetworks.com/r/Cortex-XSOAR/6.x/Cortex-XSOAR-Playbook-Design-Guide/Automations - For Cortex XSOAR 8 Cloud see https://docs-cortex.paloaltonetworks.com/r/Cortex-XSOAR/8/Cortex-XSOAR-Cloud-Documentation/Create-a-script - For Cortex XSOAR 8.7 On-prem see https://docs-cortex.paloaltonetworks.com/r/Cortex-XSOAR/8.7/Cortex-XSOAR-On-prem-Documentation/Create-a-script

Type

python

Pack

CrowdStrikeFalconStreamingV2

Source

import json
from typing import Any

import demistomock as demisto  # noqa: F401
from CommonServerPython import *  # noqa: F401


def get_host_from_system_incident(incident: dict[str, Any]) -> str:
    host = ""
    for label in incident.get("labels", []):
        if label.get("type", "") == "System":
            host = label.get("value", "")
    return host


def main():  # pragma: no cover
    res = True
    for inc in demisto.incidents():
        host = get_host_from_system_incident(inc)
        if host:
            sameIncidents0Contents = execute_command("getIncidents", {"query": f'labels.value:"{host}" and labels.type:System'})
            # if found sameIncidents found, add this incident data to war room
            sameIncidentsCount = sameIncidents0Contents["total"]
            if sameIncidentsCount > 0:
                res = False
                otherIncidents = sameIncidents0Contents["data"]
                entries: list[dict[str, Any]] = []
                entries.append({"Contents": f'Duplicate incident from crowdstrike: {inc.get("name")}'})
                entries.append({"Type": EntryType.NOTE, "ContentsFormat": "json", "Contents": json.dumps(inc)})
                entries_str = json.dumps(entries)
                execute_command("addEntries", {"id": otherIncidents[0]["id"], "entries": entries_str})

    demisto.results(res)


if __name__ in ("__main__", "__builtin__", "builtins"):  # pragma: no cover
    main()

README