FetchIndicatorsFromFile

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