MarkAsEvidenceBySearch

Search entries in the war room for the pattern text, and mark them as evidence.

Type

python

Pack

UpdateEntriesBySearch

Source

import fnmatch
import re
from collections.abc import Iterator
from typing import Any

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


def to_string(value: Any) -> Optional[str]:
    if isinstance(value, list | dict) or value is None:
        return None
    try:
        return str(value)
    except ValueError:
        return None


def to_entry_type_code(name: str) -> int:
    etype = {"NOTE": EntryType.NOTE, "ERROR": EntryType.ERROR}.get(name)

    if etype is None:
        raise ValueError(f"Invalid entry type: {name}")
    return etype


def build_pattern(pattern_algorithm: str, pattern: str, case_insensitive: bool) -> re.Pattern[str]:
    """
    Build a matching object from the pattern given.

    :param pattern_algorithm: A pattern matching algorithm.
    :param pattern: A pattern text.
    :param case_insensitive: True if the matching is performed in case-insensitive, False otherwise.
    :return A matching object built.
    """
    if pattern_algorithm == "basic":
        pattern = re.escape(pattern)
    elif pattern_algorithm == "wildcard":
        pattern = fnmatch.translate(pattern)
    elif pattern_algorithm == "regex":
        pass
    else:
        raise ValueError(f"Invalid pattern algorithm: {pattern_algorithm}")

    return re.compile(pattern, re.IGNORECASE if case_insensitive else 0)


class EntryFilter:
    def __init__(
        self,
        include_pattern: re.Pattern[str],
        exclude_pattern: Optional[re.Pattern[str]],
        node_paths: list[str],
        filter_entry_formats: list[str],
        filter_entry_types: list[str],
    ):
        """
        Initialize the filter with the matching conditions.

        :param include_pattern: A pattern to perform matching.
        :param exclude_pattern: A pattern to exclude.
        :param node_paths: The list of node path of entries to which the pattern matching is performed.
        :param filter_entry_formats: The list of entry format to filter entries.
        :param filter_entry_types: The list of entry type to filter entries.
        """
        self.__include_pattern = include_pattern
        self.__exclude_pattern = exclude_pattern
        self.__node_paths = node_paths
        self.__filter_entry_formats = filter_entry_formats
        self.__filter_entry_types = [to_entry_type_code(f) for f in filter_entry_types]

        # Check content format name
        for f in filter_entry_formats:
            if not EntryFormat.is_valid_type(f):
                raise ValueError(f"Invalid entry format: {f}")

    def match(self, entry: dict[str, Any]) -> Optional[tuple[re.Match, str]]:
        """
        Search the entry for the pattern.

        :param entry: The entry data.
        :return: re.Match and the target string if the pattern matched with the entry, None otherwise.
        """

        def iterate_value(value: Any) -> Iterator[Any]:
            if isinstance(value, list):
                for v in value:
                    yield from iterate_value(v)

            elif isinstance(value, dict):
                for _k, v in value.items():
                    yield from iterate_value(v)
            else:
                yield value

        if self.__filter_entry_types and entry.get("Type") not in self.__filter_entry_types:
            return None

        if self.__filter_entry_formats and entry.get("ContentsFormat") not in self.__filter_entry_formats:
            return None

        matched = None
        for node_path in self.__node_paths:
            for val in iterate_value(demisto.get(entry, node_path)):
                s = to_string(val)
                if s is not None:
                    if self.__exclude_pattern and self.__exclude_pattern.search(s):
                        return None

                    if not matched:
                        m = self.__include_pattern.search(s)
                        if m:
                            matched = (m, s)
        return matched


class Entry:
    def __init__(self, entry: dict[str, Any], match: Optional[re.Match], value_matched: Optional[str]):
        self.entry = entry
        self.match = match
        self.value_matched = value_matched


def iterate_entries(
    incident_id: Optional[str], query_filter: dict[str, Any], entry_filter: Optional[EntryFilter] = None
) -> Iterator[Entry]:
    """
    Iterate war room entries

    :param incident_id: The incident ID to search entries from.
    :param query_filter: Filters to search entries.
    :param entry_filter: Filters to filter entries.
    :return: An iterator to retrieve entries.
    """
    query_filter = dict(**query_filter)
    first_id = 1
    while True:
        query_filter["firstId"] = str(first_id)

        ents = demisto.executeCommand("getEntries", assign_params(id=incident_id, filter=query_filter))
        if not ents:
            break

        if is_error(ents[0]):
            if first_id == 1:
                return_error("Unable to retrieve entries")
            break

        for ent in ents:
            if not entry_filter:
                yield Entry(ent, None, None)
            else:
                match = entry_filter.match(ent)
                if match:
                    yield Entry(ent, match[0], match[1])

        # Set the next ID
        last_id = ent["ID"]
        m = re.match("([0-9]+)", last_id)
        if not m:
            raise ValueError(f"Invalid entry ID: {last_id}")
        next_id = int(m[1]) + 1
        if next_id <= first_id:
            break
        first_id = next_id


def main():
    args = demisto.args()

    build_pattern_args = assign_params(
        pattern_algorithm=args.get("algorithm", "basic"), case_insensitive=argToBoolean(args.get("case_insensitive", False))
    )
    build_pattern_args["pattern"] = args["pattern"]
    include_pattern = build_pattern(**build_pattern_args)
    exclude_pattern = None
    if args.get("exclude_pattern"):
        build_pattern_args["pattern"] = args["exclude_pattern"]
        exclude_pattern = build_pattern(**build_pattern_args)

    filter_options = argToList(args.get("filter_options", []))
    output_option = args.get("summary", "basic")

    exclude_ids = []
    if "exclude_this_entry" in filter_options:
        exclude_ids.append(demisto.parentEntry()["id"])

    ents = []
    for entry in iterate_entries(
        incident_id=args.get("incident_id"),
        query_filter=assign_params(categories=argToList(args.get("filter_categories")), tags=argToList(args.get("filter_tags"))),
        entry_filter=EntryFilter(
            include_pattern=include_pattern,
            exclude_pattern=exclude_pattern,
            node_paths=argToList(args.get("node_paths", "Contents")),
            filter_entry_formats=argToList(args.get("filter_entry_formats", [])),
            filter_entry_types=argToList(args.get("filter_entry_types", [])),
        ),
    ):
        if entry.entry["ID"] not in exclude_ids:
            rent = {"ID": entry.entry["ID"]}
            if output_option == "verbose" and entry.match:
                rent["Where"] = entry.match[0][:128]
                rent["Text"] = entry.value_matched
            ents.append(rent)

    if "first_entry" in filter_options:
        if "last_entry" in filter_options:
            del ents[1:-1]
        else:
            ents = ents[:1]
    elif "last_entry" in filter_options:
        ents = ents[-1:]

    if not ents:
        return_outputs("No entries matched")
    else:
        dry_run = argToBoolean(args.get("dry_run", False))

        tags = argToList(args.get("tags", []))
        for ent in ents:
            ent["Tags"] = ",".join(tags)

        if not dry_run:
            description = args.get("description", "")

            for ent in ents:
                entry_id = ent["ID"]
                res = demisto.executeCommand("markAsEvidence", {"id": entry_id, "tags": ent["Tags"], "description": description})
                if not res or is_error(res[0]):
                    return_error(f"Failed to mark an entrie as evidence: entryID={entry_id}")

        md = f"**Matched entries:** {len(ents)}"
        if output_option != "quiet":
            header = assign_params(
                ID="Entry ID",
                Tags="Tags",
                Where="Where" if output_option == "verbose" else None,
                Text="Text" if output_option == "verbose" else None,
            )
            md += "\n" + tblToMd("", ents, headers=header.keys(), headerTransform=lambda h: header.get(h, ""))
        return_outputs(md)


if __name__ in ("__builtin__", "builtins"):
    main()

README