DBotFindSimilarIncidentsByIndicators

Finds similar incidents based on indicators' similarity. Indicators' contribution to the final score is based on their scarcity.

Type

python

Pack

Base

Source

import math
import re
from collections import Counter

import demistomock as demisto
import numpy as np
import pandas as pd
from CommonServerPython import *
from GetIncidentsApiModule import *  # noqa: E402
from sklearn.base import BaseEstimator, TransformerMixin

from CommonServerUserPython import *

SEARCH_INDICATORS_LIMIT = 10000
SEARCH_INDICATORS_PAGE_SIZE = 500

ROUND_SCORING = 2
PLAYGROUND_PATTERN = "[a-z0-9]{8}-[a-z0-9]{4}-[a-z0-9]{4}-[a-z0-9]{4}-[a-z0-9]{12}"

# Mutual indicators fields/columns
INDICATOR_ID_FIELD = "id"
VALUE_FIELD = "value"
VALUE_FIELD_FOR_COMPATIBILITY = "name"  # value field can't be used in demisto.searchIndicators on v6.9.0
SCORE_FIELD = "score"
INVESTIGATION_IDS_FIELD = "investigationIDs"
INDICATOR_TYPE_FIELD = "indicator_type"
INDICATOR_LINK_COLUMN = "indicatorID"
TYPE_COLUMN = "type"
REPUTATION_COLUMN = "Reputation"
INVOLVED_INCIDENTS_COUNT_COLUMN = "involvedIncidentsCount"

INDICATOR_FIELDS_TO_POPULATE_FROM_QUERY = [
    INDICATOR_ID_FIELD,
    INDICATOR_TYPE_FIELD,
    INVESTIGATION_IDS_FIELD,
    SCORE_FIELD,
    VALUE_FIELD_FOR_COMPATIBILITY,
]
MUTUAL_INDICATORS_HEADERS = [INDICATOR_LINK_COLUMN, VALUE_FIELD, TYPE_COLUMN, REPUTATION_COLUMN, INVOLVED_INCIDENTS_COUNT_COLUMN]

# Similar incidents fields/columns
INCIDENT_ID_FIELD = "id"
CREATED_FIELD = "created"
NAME_FIELD = "name"
STATUS_FIELD = "status"
INCIDENT_LINK_COLUMN = "incident ID"
INDICATORS_COLUMN = "indicators"
SIMILARITY_SCORE_COLUMN = "similarity indicators"
IDENTICAL_INDICATORS_COLUMN = "Identical indicators"

FIRST_COLUMNS_INCIDENTS_DISPLAY = [INCIDENT_LINK_COLUMN, CREATED_FIELD, NAME_FIELD]
FIELDS_TO_EXCLUDE_FROM_DISPLAY = [INCIDENT_ID_FIELD]

STATUS_DICT = {
    0: "Pending",
    1: "Active",
    2: "Closed",
    3: "Archive",
}
INDICATOR_LINK_FORMAT = "[{0}](#/indicator/{0})"
INCIDENT_LINK_FORMAT = "[{0}](#/Details/{0})"
DATE_FORMAT = "%Y-%m-%d"


def flatten_list(my_list: list[list]) -> list:
    """
    Flatten a list of list
    :param l: list of list
    :return: list
    """
    return [item for sublist in my_list for item in sublist]


class FrequencyIndicators(BaseEstimator, TransformerMixin):
    """
    FrequencyIndicators class for indicator frequencies computation
    """

    def __init__(self, incident_field: str, actual_incident: pd.DataFrame) -> None:
        self.column_name = incident_field
        self.frequency: dict = {}
        self.vocabulary = actual_incident[self.column_name].iloc[0].split(" ")

    def fit(self, x: pd.DataFrame) -> "FrequencyIndicators":
        x = x[self.column_name]
        size = len(x) + 1
        frequencies = Counter(flatten_list([t.split(" ") for t in x.values]))
        frequencies.update(Counter(self.vocabulary))
        self.frequency = {k: math.log(1 + size / v) for k, v in frequencies.items()}
        return self

    def transform(self, x: pd.DataFrame) -> pd.DataFrame:
        return x[self.column_name].apply(self.compute_term_score)

    def compute_term_score(self, indicators_values_string: str) -> float:
        x = indicators_values_string.split(" ")
        return sum([1 * self.frequency[word] for word in self.vocabulary if word in x]) / sum(
            [self.frequency[word] for word in self.vocabulary]
        )


class FrequencyIndicatorsTransformer:
    def __init__(self, incidents_df: pd.DataFrame, actual_incident: pd.DataFrame):
        """
        :param incidents_df: DataFrame of related incidents
        :param actual_incident: DataFrame of the actual incident
        """
        self.incidents_df = incidents_df
        self.actual_incident = actual_incident
        self.transformed_column = INDICATORS_COLUMN
        self.scoring_function = lambda x: x

    def fit_transform(self):
        transformer = FrequencyIndicators(
            self.transformed_column,
            self.actual_incident,
        )
        x_vector = transformer.fit_transform(self.incidents_df)
        incident_vect = transformer.transform(self.actual_incident)
        return x_vector, incident_vect

    def get_score(self):
        x_vector, _ = self.fit_transform()
        distance = self.scoring_function(x_vector)
        self.incidents_df[SIMILARITY_SCORE_COLUMN] = np.round(distance, ROUND_SCORING)
        return self.incidents_df


class Model:
    def __init__(
        self,
        incident_to_match: pd.DataFrame,
        incidents_df: pd.DataFrame,
        similarity_threshold: float,
        max_incidents: int,
    ) -> None:
        """
        :param incident_to_match: Dataframe with one incident
        :param incidents_df: Dataframe with all the incidents
        :param similarity_threshold: The similarity threshold
        :param max_incidents: Maximum number of incidents to return
        """
        self.incident_to_match = incident_to_match
        self.incidents_df = incidents_df
        self.threshold = similarity_threshold
        self.max_incidents = max_incidents

    def predict(self) -> pd.DataFrame:
        self.get_score()
        self.filter_results()
        self.prepare_for_display()
        return self.incidents_df

    def get_score(self) -> None:
        t = FrequencyIndicatorsTransformer(
            self.incidents_df,
            self.incident_to_match,
        )
        t.get_score()

    def filter_results(self) -> None:
        self.incidents_df = self.incidents_df[self.incidents_df[SIMILARITY_SCORE_COLUMN] > self.threshold]
        self.incidents_df = self.incidents_df.sort_values([SIMILARITY_SCORE_COLUMN], ascending=False)
        self.incidents_df = self.incidents_df.head(self.max_incidents)

    def prepare_for_display(self) -> None:
        vocabulary = self.incident_to_match[INDICATORS_COLUMN].iloc[0].split(" ")
        self.incidents_df[IDENTICAL_INDICATORS_COLUMN] = self.incidents_df[INDICATORS_COLUMN].apply(
            lambda x: ",".join([id for id in x.split(" ") if id in vocabulary])
        )


def search_indicators(
    query: str,
    fields_to_populate: list | None = None,
    limit: int = SEARCH_INDICATORS_LIMIT,
    page_size: int = SEARCH_INDICATORS_PAGE_SIZE,
) -> list:
    demisto.debug(f"Searching indicators with {query=}")
    search_indicators = IndicatorsSearcher(
        query=query, limit=limit, size=page_size, filter_fields=",".join(fields_to_populate) if fields_to_populate else None
    )
    return flatten_list([ioc_res.get("iocs") or [] for ioc_res in search_indicators])


def get_indicators_of_actual_incident(
    incident_id: str,
    indicator_types: list[str],
    min_number_of_indicators: int,
    max_incidents_per_indicator: int,
) -> dict[str, dict]:
    """Given an incident ID, returns a map between IDs of its related indicators and their data

    :param incident_id: ID of actual incident
    :param indicators_types: list of indicators type accepted
    :param min_number_of_indicators: Min number of indicators in the actual incident
    :param max_incidents_per_indicator: Max incidents in indicators for white list
    :return: a map from indicator ids of the actual incident to their data
    """
    indicators = search_indicators(query=f"investigationIDs:({incident_id})")
    if not indicators:
        return {}
    indicators = [i for i in indicators if len(i.get("investigationIDs") or []) <= max_incidents_per_indicator]
    if indicator_types:
        indicators = [x for x in indicators if x[INDICATOR_TYPE_FIELD].lower() in indicator_types]
    if len(indicators) < min_number_of_indicators:
        return {}

    indicators_data = {ind[INDICATOR_ID_FIELD]: ind for ind in indicators}
    demisto.debug(f"Found {len(indicators_data)} indicators for incident {incident_id}: {list(indicators_data.keys())}")
    return indicators_data


def get_related_incidents(
    indicators: dict[str, dict],
    query: str,
    from_date: str | None,
) -> list[str]:
    """Given indicators data including their related incidents,
    filters their related incidents by query and date and returns a list of the incident IDs.

    :param indicators: List of indicators
    :param query: A query to filter the related incidents by
    :param from_date: A created date to filter the related incidents by
    :return: The list of the related incident IDs
    """
    incident_ids = flatten_list([i.get("investigationIDs") or [] for i in indicators.values()])
    incident_ids = list({x for x in incident_ids if not re.match(PLAYGROUND_PATTERN, x)})
    if not (query or from_date) or not incident_ids:
        demisto.debug(f"Found {len(incident_ids)} related incidents: {incident_ids}")
        return incident_ids

    args = {
        "query": f"{query + ' AND ' if query else ''}incident.id:({' '.join(incident_ids)})",
        "populateFields": INCIDENT_ID_FIELD,
        "fromDate": from_date,
    }
    demisto.debug(f"Executing GetIncidentsByQuery with {args=}")
    incidents = get_incidents_by_query(args)
    incident_ids = [incident[INCIDENT_ID_FIELD] for incident in incidents]
    demisto.debug(f"Found {len(incident_ids)} related incidents: {incident_ids}")
    return incident_ids


def get_indicators_of_related_incidents(
    incident_ids: list[str],
    max_incidents_per_indicator: int,
) -> list[dict]:
    if not incident_ids:
        demisto.debug("No mutual indicators were found.")
        return []
    indicators = search_indicators(
        query=f"investigationIDs:({' '.join(incident_ids)})",
        fields_to_populate=INDICATOR_FIELDS_TO_POPULATE_FROM_QUERY,
    )
    indicators = [i for i in indicators if len(i.get("investigationIDs") or []) <= max_incidents_per_indicator]
    indicators_ids = [ind[INDICATOR_ID_FIELD] for ind in indicators]
    demisto.debug(f"Found {len(indicators_ids)} related indicators: {indicators_ids}")
    return indicators


def get_mutual_indicators(
    related_indicators: list[dict],
    indicators_of_actual_incident: dict[str, dict],
) -> list[dict]:
    mutual_indicators = [ind for ind in related_indicators if ind[INDICATOR_ID_FIELD] in indicators_of_actual_incident]
    mutual_indicators_ids = [ind[INDICATOR_ID_FIELD] for ind in mutual_indicators]
    demisto.debug(f"Found {len(mutual_indicators_ids)} mutual indicators: {mutual_indicators_ids}")
    return mutual_indicators


def get_mutual_indicators_df(
    indicators: list[dict],
    incident_ids: list[str],
) -> pd.DataFrame:
    indicators_df = pd.DataFrame(indicators)
    if not indicators_df.empty:
        indicators_df[INVOLVED_INCIDENTS_COUNT_COLUMN] = indicators_df[INVESTIGATION_IDS_FIELD].apply(
            lambda inv_ids: sum(id_ in incident_ids for id_ in inv_ids),
        )
        indicators_df[INDICATOR_LINK_COLUMN] = indicators_df[INDICATOR_ID_FIELD].apply(lambda x: INDICATOR_LINK_FORMAT.format(x))
        indicators_df = indicators_df.sort_values(
            [SCORE_FIELD, INVOLVED_INCIDENTS_COUNT_COLUMN],
            ascending=False,
        )
        indicators_df[REPUTATION_COLUMN] = indicators_df[SCORE_FIELD].apply(scoreToReputation)  # pylint: disable=E1137
        indicators_df = indicators_df.rename({INDICATOR_TYPE_FIELD: TYPE_COLUMN}, axis=1)
        indicators_df = indicators_df.rename({VALUE_FIELD_FOR_COMPATIBILITY: VALUE_FIELD}, axis=1)
    return indicators_df


def mutual_indicators_results(mutual_indicators: list[dict], incident_ids: list[str]):
    indicators_df = get_mutual_indicators_df(mutual_indicators, incident_ids)
    outputs = [] if indicators_df.empty else indicators_df[[INDICATOR_ID_FIELD, VALUE_FIELD]].to_dict(orient="records")
    readable_output = tableToMarkdown(
        "Mutual Indicators",
        indicators_df.to_dict(orient="records"),
        headers=MUTUAL_INDICATORS_HEADERS,
        headerTransform=pascalToSpace,
    )
    return CommandResults(
        outputs=outputs,
        outputs_prefix="MutualIndicators.indicators",
        readable_output=readable_output,
    )


def create_actual_incident_df(indicators_of_actual_incident: dict[str, dict]) -> pd.DataFrame:
    return pd.DataFrame(
        data=[" ".join(indicators_of_actual_incident.keys())],
        columns=[INDICATORS_COLUMN],
    )


def create_related_incidents_df(
    indicators: list[dict],
    incident_ids: list[str],
    actual_incident_id: str,
) -> dict[str, list]:
    """
    :param indicators: list of dict representing indicators
    :param incident_ids: list of incident ids
    :return: dict of {incident id : list of indicators ids related to this incident)
    """
    incidents_to_indicators = {
        inc_id: [
            indicator[INDICATOR_ID_FIELD] for indicator in indicators if inc_id in (indicator.get(INVESTIGATION_IDS_FIELD) or [])
        ]
        for inc_id in incident_ids
    }
    return pd.DataFrame.from_dict(
        data={k: " ".join(v) for k, v in incidents_to_indicators.items() if k != actual_incident_id},
        orient="index",
        columns=[INDICATORS_COLUMN],
    )


def enrich_incidents(
    incidents: pd.DataFrame,
    fields_to_display: list,
) -> pd.DataFrame:
    """
    Enriches a DataFrame of incidents with the given fields to display.

    :param similar_incidents: Incidents dataFrame
    :param fields_to_display: Fields selected for enrichement
    :return: Incidents dataFrame enriched
    """
    if incidents.empty:
        return incidents

    incident_ids = incidents.id.tolist() if INCIDENT_ID_FIELD in incidents.columns else incidents.index

    args = {
        "query": f"incident.id:({' '.join(incident_ids)})",
        "populateFields": ",".join(fields_to_display),
    }
    demisto.debug(f"Executing GetIncidentsByQuery with {args=}")
    res = get_incidents_by_query(args)
    incidents_map: dict[str, dict] = {incident[INCIDENT_ID_FIELD]: incident for incident in res}
    if CREATED_FIELD in fields_to_display:
        incidents[CREATED_FIELD] = [
            dateparser.parse(incidents_map[inc_id][CREATED_FIELD]).strftime(DATE_FORMAT)  # type: ignore
            for inc_id in incident_ids
        ]
    if STATUS_FIELD in fields_to_display:
        incidents[STATUS_FIELD] = [STATUS_DICT.get(incidents_map[inc_id][STATUS_FIELD]) or " " for inc_id in incident_ids]

    for field in fields_to_display:
        if field not in [CREATED_FIELD, STATUS_FIELD]:
            incidents[field] = [incidents_map[inc_id].get(field) or "" for inc_id in incident_ids]
    return incidents


def replace_indicator_ids_with_values(
    inc_ids: str,
    indicators_data: dict[str, dict],
) -> str:
    return "\n".join([indicators_data.get(x, {}).get(VALUE_FIELD_FOR_COMPATIBILITY) or " " for x in inc_ids.split(" ")])


def format_similar_incidents(
    similar_incidents: pd.DataFrame,
    indicators_data: dict[str, dict],
    fields_to_display: list[str],
) -> pd.DataFrame:
    """Formats the similar incidents DataFrame.

    :param indicators_data: a mapping between IDs and the mutual indicators data
    :param fields_to_display: Fields selected for enrichement
    :return: a formatted, enriched DataFrame of the similar incidents.
    """
    if similar_incidents.empty:
        demisto.debug("No similar incidents found.")
        return similar_incidents

    # format and enrich DataFrame
    similar_incidents = similar_incidents.reset_index().rename(columns={"index": INCIDENT_ID_FIELD})
    similar_incidents[INCIDENT_LINK_COLUMN] = similar_incidents[INCIDENT_ID_FIELD].apply(
        lambda _id: INCIDENT_LINK_FORMAT.format(_id)
    )
    similar_incidents[IDENTICAL_INDICATORS_COLUMN] = similar_incidents[IDENTICAL_INDICATORS_COLUMN].apply(
        lambda inc_ids: replace_indicator_ids_with_values(inc_ids, indicators_data)
    )
    similar_incidents = similar_incidents[
        [INCIDENT_LINK_COLUMN, INCIDENT_ID_FIELD, IDENTICAL_INDICATORS_COLUMN, SIMILARITY_SCORE_COLUMN]
    ]
    return enrich_incidents(similar_incidents, fields_to_display)


def similar_incidents_results(
    similar_incidents: pd.DataFrame,
    indicators_of_actual_incident: dict,
    fields_to_display: list,
):
    similar_incidents = format_similar_incidents(similar_incidents, indicators_of_actual_incident, fields_to_display)
    outputs = similar_incidents.to_dict(orient="records")
    additional_headers = [
        x for x in similar_incidents.columns.tolist() if x not in FIRST_COLUMNS_INCIDENTS_DISPLAY + FIELDS_TO_EXCLUDE_FROM_DISPLAY
    ]
    return CommandResults(
        outputs={"similarIncident": outputs, "isSimilarIncidentFound": len(outputs) > 0},
        outputs_prefix="DBotFindSimilarIncidentsByIndicators",
        raw_response=outputs,
        readable_output=tableToMarkdown(
            "Similar Incidents",
            outputs,
            headers=FIRST_COLUMNS_INCIDENTS_DISPLAY + additional_headers,
            headerTransform=str.title,
        ),
        tags=["similarIncidents"],  # type: ignore
    )


def actual_incident_results(
    incident_df: pd.DataFrame,
    incident_id: str,
    indicators_data: dict,
    fields_to_display: list[str],
) -> CommandResults:
    """
    Formats the given DataFrame, and returns a CommandResults object of the actual incident data
    :param incident_df: a DataFrame of actual incident
    :param incident_id: the incident ID
    :param indicators_data: indicators data of the actual incident
    :param fields_to_display: A list of fields to display
    :return: a CommandResults obj
    """
    incident_df[INCIDENT_ID_FIELD] = [incident_id]
    incident_df[INCIDENT_LINK_COLUMN] = incident_df[INCIDENT_ID_FIELD].apply(lambda _id: INCIDENT_LINK_FORMAT.format(_id))
    incident_df[INDICATORS_COLUMN] = incident_df[INDICATORS_COLUMN].apply(
        lambda inc_ids: replace_indicator_ids_with_values(inc_ids, indicators_data)
    )
    incident_df = enrich_incidents(incident_df, fields_to_display)
    additional_headers = [
        x for x in incident_df.columns.tolist() if x not in FIRST_COLUMNS_INCIDENTS_DISPLAY + FIELDS_TO_EXCLUDE_FROM_DISPLAY
    ]
    return CommandResults(
        readable_output=tableToMarkdown(
            "Actual Incident",
            incident_df.to_dict(orient="records"),
            headers=FIRST_COLUMNS_INCIDENTS_DISPLAY + additional_headers,
            headerTransform=pascalToSpace,
        ),
    )


def find_similar_incidents_by_indicators(incident_id: str, args: dict) -> list[CommandResults]:
    # get_indicators_of_actual_incident() args
    indicators_types = argToList(args.get("indicatorsTypes"), transform=str.lower)
    min_number_of_indicators = int(args["minNumberOfIndicators"])
    max_incidents_per_indicator = int(args["maxIncidentsInIndicatorsForWhiteList"])

    # get_related_incidents() args
    query = args.get("query") or ""
    from_date = args.get("fromDate")

    # get_similar_incidents() args
    similarity_threshold = float(args["threshold"])
    max_incidents_to_display = int(args["maxIncidentsToDisplay"])

    # outputs formatting args
    show_actual_incident = argToBoolean(args.get("showActualIncident"))
    fields_to_display = list(set(argToList(args["fieldsIncidentToDisplay"])) | {CREATED_FIELD, NAME_FIELD})

    command_results_list: list[CommandResults] = []

    indicators_of_actual_incident = get_indicators_of_actual_incident(
        incident_id,
        indicators_types,
        min_number_of_indicators,
        max_incidents_per_indicator,
    )

    incident_ids = get_related_incidents(indicators_of_actual_incident, query, from_date)
    related_indicators = get_indicators_of_related_incidents(incident_ids, max_incidents_per_indicator)
    mutual_indicators = get_mutual_indicators(related_indicators, indicators_of_actual_incident)
    actual_incident_df = create_actual_incident_df(indicators_of_actual_incident)
    related_incidents_df = create_related_incidents_df(related_indicators, incident_ids, incident_id)

    similar_incidents = Model(
        actual_incident_df,
        related_incidents_df,
        similarity_threshold,
        max_incidents_to_display,
    ).predict()

    if show_actual_incident:
        command_results_list.append(
            actual_incident_results(actual_incident_df, incident_id, indicators_of_actual_incident, fields_to_display)
        )
    command_results_list.extend(
        [
            mutual_indicators_results(mutual_indicators, incident_ids),
            similar_incidents_results(similar_incidents, indicators_of_actual_incident, fields_to_display),
        ]
    )
    return command_results_list


def main():  # pragma: no cover
    try:
        args = demisto.args()
        incident_id = args.get("incidentId") or demisto.incidents()[0]["id"]
        return_results(find_similar_incidents_by_indicators(incident_id, args))
    except Exception as e:
        return_error(f"Failed to execute DBotFindSimilarIncidentsByIndicators. Error: {e}")


if __name__ in ["__main__", "__builtin__", "builtins"]:
    main()

README