FetchIndicatorsFromFile

Fetches indicators from a file. Supports TXT, XLS, XLSX, CSV, DOC and DOCX file types.

Type

python

Pack

CommonScripts

Source

import csv
import re
import traceback

import demistomock as demisto
import tldextract
import xlrd
from CommonServerPython import *

from CommonServerUserPython import *


def csv_file_to_indicator_list(file_path, col_num, starting_row, auto_detect, default_type, type_col, limit, offset):
    indicator_list = []

    # TODO: add run on all columns functionality

    line_index = 0
    with open(file_path) as csv_file:
        # csv reader can fail when encountering a NULL byte (\0) - so we go through the file and take out the NUL bytes.
        file_reader = csv.reader(line.replace("\0", "") for line in csv_file)
        for row in file_reader:
            if line_index >= starting_row + offset and len(row) != 0:
                indicator = row[col_num]

                indicator_type = get_indicator_type(
                    indicator, auto_detect, default_type, file_type="csv", type_col=type_col, csv_row=row
                )

                if indicator_type is None:
                    continue

                indicator_list.append({"type": indicator_type, "value": indicator})

            line_index = line_index + 1

            if limit and len(indicator_list) == int(str(limit)):
                break

    return indicator_list


def xls_file_to_indicator_list(file_path, sheet_name, col_num, starting_row, auto_detect, default_type, type_col, limit, offset):
    indicator_list = []

    # TODO: add run on all columns functionality

    # Ensure that the has_iter will not be reseted after opening the workbook.
    xlrd.xlsx.ensure_elementtree_imported(False, None)
    xlrd.xlsx.Element_has_iter = True
    xl_woorkbook = xlrd.open_workbook(file_path)
    if sheet_name and sheet_name != "None":
        xl_sheet = xl_woorkbook.sheet_by_name(sheet_name)

    else:
        xl_sheet = xl_woorkbook.sheet_by_index(0)

    for row_index in range(starting_row + offset, xl_sheet.nrows):
        indicator = xl_sheet.cell(row_index, col_num).value

        indicator_type = get_indicator_type(
            indicator, auto_detect, default_type, file_type="xls", type_col=type_col, xl_sheet=xl_sheet, xl_row_index=row_index
        )

        if indicator_type is None:
            continue

        indicator_list.append({"type": indicator_type, "value": indicator})

        if limit and len(indicator_list) == int(str(limit)):
            break

    return indicator_list


def txt_file_to_indicator_list(file_path, auto_detect, default_type, limit, offset):
    with open(file_path) as fp:
        file_data = fp.read()

    indicator_list = []

    raw_splitted_data = re.split(r"\s|\n|\t|\"|\'|\,|\0", file_data)
    indicator_index = 0

    for indicator in raw_splitted_data:
        # drop punctuation
        if len(indicator) > 1:
            while indicator[-1] in '.,?:;\\)}]/!\n\t\0"' and len(indicator) > 1:
                indicator = indicator[:-1]

            while indicator[0] in '.,({[\n\t"' and len(indicator) > 1:
                indicator = indicator[1:]

            indicator_type = get_indicator_type(indicator, auto_detect, default_type, file_type="text")

            # indicator not recognized skip the word
            if indicator_type is None:
                continue

            if indicator_type is not None and indicator_index < offset:
                indicator_index = indicator_index + 1
                continue

            indicator_list.append({"type": indicator_type, "value": indicator})

        if limit and len(indicator_list) == int(str(limit)):
            break

    return indicator_list


def get_indicator_type(
    indicator_value, auto_detect, default_type, file_type, type_col=None, xl_sheet=None, xl_row_index=0, csv_row=None
):
    """Returns the indicator type for the given file type.

    Args:
        indicator_value (str): the indicator value
        auto_detect (bool): whether or not to auto_detect the type
        default_type (Any): the default type of the indicator (could be None or str)
        file_type (str): 'text', 'xls' or 'csv'
        type_col (Any): the column from which to fetch the indicator type in xls or csv files (could be None or int)
        xl_sheet (Any): the xls sheet from which to fetch the indicator type (could be None or ~xlrd.sheet.Sheet)
        xl_row_index (Any): the row number in the xls sheet from which to fetch the indicator type (could be None or int)
        csv_row (Any): the csv row from which to fetch the indicator type (could be None or list)

    Returns:
        Any. returns None if indicator is not recognized in text file
        or the indicator is not recognized and no default type given.
        Otherwise will return a string indicating the indicator type
    """
    indicator_type = detect_type(indicator_value)

    # indicator not recognized skip the word in text file
    if indicator_type is None and file_type == "text":
        return None

    if not auto_detect:
        indicator_type = default_type

    if file_type != "text":
        if type_col is not None and file_type == "xls":
            indicator_type = xl_sheet.cell(xl_row_index, int(type_col) - 1).value

        elif type_col is not None and file_type == "csv":
            indicator_type = csv_row[int(type_col) - 1]

        # indicator not recognized in non text file
        if indicator_type is None:
            # no default value given
            if default_type is None:
                return None
            else:
                # default value given
                indicator_type = default_type

    return indicator_type


def detect_type(indicator):
    """Infer the type of the indicator.
    Args:
        indicator(str): The indicator whose type we want to check.
    Returns:
        str. The type of the indicator.
    """
    if re.match(sha256Regex, indicator) or re.match(md5Regex, indicator) or re.match(sha1Regex, indicator):
        return FeedIndicatorType.File

    if re.match(ipv4cidrRegex, indicator):
        return FeedIndicatorType.CIDR

    if re.match(ipv6cidrRegex, indicator):
        return FeedIndicatorType.IPv6CIDR

    if re.match(ipv4Regex, indicator):
        return FeedIndicatorType.IP

    if re.match(ipv6Regex, indicator):
        return FeedIndicatorType.IPv6

    if re.match(urlRegex, indicator):
        return FeedIndicatorType.URL

    if re.match(emailRegex, indicator):
        return FeedIndicatorType.Email

    try:
        # we use TLDExtract class to fetch all existing domain suffixes from the bellow mentioned file:
        # https://raw.githubusercontent.com/publicsuffix/list/master/public_suffix_list.dat
        # the suffix_list_urls=None is used to not make http calls using the extraction - avoiding SSL errors
        if (
            tldextract.TLDExtract(
                cache_dir="https://raw.githubusercontent.com/publicsuffix/list/master/public_suffix_list.dat",
                suffix_list_urls=(),
            )
            .__call__(indicator)
            .suffix
        ):
            if "*" in indicator:
                return FeedIndicatorType.DomainGlob
            return FeedIndicatorType.Domain
    except Exception:
        pass

    return None


def fetch_indicators_from_file(args):
    file = demisto.getFilePath(args.get("entry_id"))
    file_path = file["path"]
    file_name = file["name"]
    auto_detect = args.get("auto_detect") == "True"
    default_type = args.get("default_type")
    limit = args.get("limit")

    # offset - refers to the indicator list itself -
    # lets say you have a list of 500 and you put a limit of 100 on your output -
    # you can get the next 100 by putting an offset of 100.
    offset = int(str(args.get("offset"))) if args.get("offset") else 0

    # the below params are for Excel type files only.
    sheet_name = args.get("sheet_name")
    indicator_col_num = args.get("indicator_column_number")
    indicator_type_col_num = args.get("indicator_type_column_number")

    # starting_row is for excel files -
    # from which row should I start reading the indicators, it is used to avoid table headers.
    starting_row = args.get("starting_row")

    if file_name.endswith(("xls", "xlsx")):
        indicator_list = xls_file_to_indicator_list(
            file_path,
            sheet_name,
            int(indicator_col_num) - 1,
            int(starting_row) - 1,
            auto_detect,
            default_type,
            indicator_type_col_num,
            limit,
            offset,
        )

    elif file_name.endswith("csv"):
        indicator_list = csv_file_to_indicator_list(
            file_path,
            int(indicator_col_num) - 1,
            int(starting_row) - 1,
            auto_detect,
            default_type,
            indicator_type_col_num,
            limit,
            offset,
        )

    else:
        indicator_list = txt_file_to_indicator_list(file_path, auto_detect, default_type, limit, offset)

    human_readable = tableToMarkdown(f"Indicators from {file_name}:", indicator_list, headers=["value", "type"], removeNull=True)

    # Create indicators in demisto
    errors = []
    domains = []
    for indicator in indicator_list:
        res = demisto.executeCommand("createNewIndicator", indicator)
        if is_error(res[0]):
            errors.append("Error creating indicator - {}".format(res[0]["Contents"]))

        domain_obj = {"Name": indicator.get("value")}
        if indicator.get("type") == FeedIndicatorType.Domain and domain_obj not in domains:
            domains.append(domain_obj)
    if errors:
        return_error(json.dumps(errors, indent=4))
    domain_context = {outputPaths["domain"]: domains} if domains else None
    return human_readable, domain_context, indicator_list


def main():
    try:
        return_outputs(*fetch_indicators_from_file(demisto.args()))
    except Exception as ex:
        return_error(f"Failed to execute Fetch Indicators From File. Error: {ex!s}", error=traceback.format_exc())


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

README