EDL

import glob
import hashlib
import ipaddress
import os
import re
import tempfile
import uuid
import zipfile
from base64 import b64decode
from collections.abc import Callable, Iterable
from datetime import datetime
from enum import Enum
from math import ceil
from pathlib import Path
from typing import IO

import demistomock as demisto  # noqa: F401
import tldextract
import urllib3
from CommonServerPython import *  # noqa: F401
from flask import Flask, Response, request, send_file
from netaddr import IPNetwork, IPRange, IPSet

# Disable insecure warnings
urllib3.disable_warnings()

""" GLOBAL VARIABLES """
INTEGRATION_NAME: str = "Generic Export Indicators service"
PAGE_SIZE: int = 2000
PAN_OS_MAX_URL_LEN = 255
APP: Flask = Flask("demisto-edl")
EDL_LIMIT_ERR_MSG: str = "Please provide a valid integer for List Size"
EDL_CIDR_SIZR_MSG: str = "Please provide a valid integer for CIDR size"
EDL_OFFSET_ERR_MSG: str = "Please provide a valid integer for Starting Index"
EDL_COLLAPSE_ERR_MSG: str = "The Collapse parameter can only get the following: 0 - none, 1 - range, 2 - CIDR"
EDL_MISSING_REFRESH_ERR_MSG: str = (
    'Refresh Rate must be "number date_range_unit", examples: (2 hours, 4 minutes, 6 months, 1 day, etc.)'
)
EDL_FORMAT_ERR_MSG: str = "Please provide a valid format from: text, json, csv, mgw and proxysg"
EDL_MWG_TYPE_ERR_MSG: str = (
    "The McAFee Web Gateway type can only be one of the following: string,"
    " applcontrol, dimension, category, ip, mediatype, number, regex"
)
EDL_NO_URLS_IN_PROXYSG_FORMAT = "ProxySG format only outputs URLs - no URLs found in the current query"
MAX_LIST_SIZE_WITH_URL_QUERY = 100000

EDL_ON_DEMAND_KEY: str = "UpdateEDL"
EDL_ON_DEMAND_CACHE_PATH: str = ""
EDL_FULL_LOG_PATH: str = f"full_log_{demisto.uniqueFile()}"
EDL_FULL_LOG_PATH_WIP: str = f"wip_log_{demisto.uniqueFile()}"
LOGS_ZIP_FILE_PREFIX: str = "log_download"
EDL_ON_DEMAND_CACHE_ORIGINAL_SIZE: int = 0
EDL_SEARCH_LOOP_LIMIT: int = 10
MAX_DISPLAY_LOG_FILE_SIZE = 100000
LARGE_LOG_DISPLAY_MSG = "# Log exceeds max size. Refresh to download as file."

""" REFORMATTING REGEXES """
_PROTOCOL_REMOVAL = re.compile("^(?:[a-z]+:)*//")
_PORT_REMOVAL = re.compile(r"^((?:[a-z]+:)*//([a-z0-9\-\.]+)|([a-z0-9\-\.]+))(?:\:[0-9]+)*")
_URL_WITHOUT_PORT = r"\g<1>"
_INVALID_TOKEN_REMOVAL = re.compile(r"(?:[^\./+=\?&]+\*[^\./+=\?&]*)|(?:[^\./+=\?&]*\*[^\./+=\?&]+)")
_BROAD_PATTERN = re.compile(r"^(?:\*\.)+[a-zA-Z]+(?::[0-9]+)?$")

DONT_COLLAPSE = "Don't Collapse"
COLLAPSE_TO_CIDR = "To CIDRS"
COLLAPSE_TO_RANGES = "To Ranges"

MAXIMUM_CIDR_SIZE_DEFAULT = 8

MIMETYPE_JSON_SEQ: str = "application/json-seq"
MIMETYPE_JSON: str = "application/json"
MIMETYPE_CSV: str = "text/csv"
MIMETYPE_TEXT: str = "text/plain"

FORMAT_CSV: str = "CSV"
FORMAT_TEXT: str = "PAN-OS (text)"
FORMAT_JSON: str = "JSON"
FORMAT_ARG_MWG = "mwg"
FORMAT_ARG_BLUECOAT = "bluecoat"
FORMAT_ARG_PROXYSG = "proxysg"
FORMAT_MWG: str = "McAfee Web Gateway"
FORMAT_PROXYSG: str = "Symantec ProxySG"

MWG_TYPE_OPTIONS = ["string", "applcontrol", "dimension", "category", "ip", "mediatype", "number", "regex"]

INCREASE_LIMIT = 1.1

EXTENSIVE_LOGGING: bool = False


class IndicatorAction(Enum):
    ADDED = "Added"
    MODIFIED = "Modified"
    DROPPED = "Dropped"


"""Request Arguments Class"""


def debug_function(func):
    def wrapper(*args, **kwargs):
        if EXTENSIVE_LOGGING:
            demisto.debug(f"edl: Entering function {func.__name__}")
        results = func(*args, **kwargs)
        if EXTENSIVE_LOGGING:
            demisto.debug(f"edl: Exiting function {func.__name__}")
        return results

    return wrapper


class RequestArguments:
    CTX_QUERY_KEY = "last_query"
    CTX_OUT_FORMAT = "out_format"
    CTX_LIMIT_KEY = "last_limit"
    CTX_OFFSET_KEY = "last_offset"
    CTX_INVALIDS_KEY = "drop_invalids"
    CTX_PORT_STRIP_KEY = "url_port_stripping"
    CTX_COLLAPSE_IPS_KEY = "collapse_ips"
    CTX_EMPTY_EDL_COMMENT_KEY = "add_comment_if_empty"
    CTX_MWG_TYPE = "mwg_type"
    CTX_CATEGORY_DEFAULT = "bc_category"
    CTX_CATEGORY_ATTRIBUTE = "category_attribute"
    CTX_FIELDS_TO_PRESENT = "fields_to_present"
    CTX_CSV_TEXT = "csv_text"
    CTX_PROTOCOL_STRIP_KEY = "url_protocol_stripping"
    CTX_URL_TRUNCATE_KEY = "url_truncate"
    CTX_MAXIMUM_CIDR = "maximum_cidr_size"
    CTX_NO_TLD = "no_wildcard_tld"

    FILTER_FIELDS_ON_FORMAT_TEXT = "name,type"
    FILTER_FIELDS_ON_FORMAT_MWG = "name,type,sourceBrands"
    FILTER_FIELDS_ON_FORMAT_PROXYSG = "name,type,proxysgcategory"
    FILTER_FIELDS_ON_FORMAT_CSV = "name,type"
    FILTER_FIELDS_ON_FORMAT_JSON = "name,type"

    def __init__(
        self,
        query: str = "",
        out_format: str = FORMAT_TEXT,
        limit: int = 10000,
        offset: int = 0,
        url_port_stripping: bool = False,
        drop_invalids: bool = False,
        collapse_ips: str = DONT_COLLAPSE,
        add_comment_if_empty: bool = True,
        mwg_type: str = "string",
        category_default: str = "bc_category",
        category_attribute: Optional[str] = None,
        fields_to_present: str = "",
        csv_text: bool = False,
        url_protocol_stripping: bool = False,
        url_truncate: bool = False,
        maximum_cidr_size: int = MAXIMUM_CIDR_SIZE_DEFAULT,
        no_wildcard_tld: bool = False,
    ):
        self.query = query
        self.out_format = out_format
        self.limit = try_parse_integer(limit, EDL_LIMIT_ERR_MSG)
        self.offset = try_parse_integer(offset, EDL_OFFSET_ERR_MSG)
        self.url_port_stripping = url_port_stripping
        self.url_protocol_stripping = url_protocol_stripping
        self.drop_invalids = drop_invalids
        self.collapse_ips = collapse_ips
        self.add_comment_if_empty = add_comment_if_empty
        self.mwg_type = mwg_type
        self.category_default = category_default
        self.category_attribute = []  # type:List
        self.fields_to_present = self.get_fields_to_present(fields_to_present)
        self.csv_text = csv_text
        self.url_truncate = url_truncate
        self.maximum_cidr_size = maximum_cidr_size
        self.no_wildcard_tld = no_wildcard_tld

        if category_attribute is not None:
            category_attribute_list = argToList(category_attribute)

            if len(category_attribute_list) != 1 or "" not in category_attribute_list:
                self.category_attribute = category_attribute_list

    def to_context_json(self):
        return {
            self.CTX_QUERY_KEY: self.query,
            self.CTX_OUT_FORMAT: self.out_format,
            self.CTX_LIMIT_KEY: self.limit,
            self.CTX_OFFSET_KEY: self.offset,
            self.CTX_INVALIDS_KEY: self.drop_invalids,
            self.CTX_PORT_STRIP_KEY: self.url_port_stripping,
            self.CTX_COLLAPSE_IPS_KEY: self.collapse_ips,
            self.CTX_EMPTY_EDL_COMMENT_KEY: self.add_comment_if_empty,
            self.CTX_MWG_TYPE: self.mwg_type,
            self.CTX_CATEGORY_DEFAULT: self.category_default,
            self.CTX_CATEGORY_ATTRIBUTE: self.category_attribute,
            self.CTX_FIELDS_TO_PRESENT: self.fields_to_present,
            self.CTX_CSV_TEXT: self.csv_text,
            self.CTX_PROTOCOL_STRIP_KEY: self.url_protocol_stripping,
            self.CTX_URL_TRUNCATE_KEY: self.url_truncate,
            self.CTX_MAXIMUM_CIDR: self.maximum_cidr_size,
            self.CTX_NO_TLD: self.no_wildcard_tld,
        }

    @classmethod
    def from_context_json(cls, ctx_dict):
        """Returns an initiated instance of the class from a json"""
        return cls(
            **assign_params(
                query=ctx_dict.get(cls.CTX_QUERY_KEY),
                out_format=ctx_dict.get(cls.CTX_OUT_FORMAT),
                limit=ctx_dict.get(cls.CTX_LIMIT_KEY),
                offset=ctx_dict.get(cls.CTX_OFFSET_KEY),
                drop_invalids=ctx_dict.get(cls.CTX_INVALIDS_KEY),
                url_port_stripping=ctx_dict.get(cls.CTX_PORT_STRIP_KEY),
                collapse_ips=ctx_dict.get(cls.CTX_COLLAPSE_IPS_KEY),
                add_comment_if_empty=ctx_dict.get(cls.CTX_EMPTY_EDL_COMMENT_KEY),
                mwg_type=ctx_dict.get(cls.CTX_MWG_TYPE),
                category_default=ctx_dict.get(cls.CTX_CATEGORY_DEFAULT),
                category_attributeself=ctx_dict.get(cls.CTX_CATEGORY_ATTRIBUTE),
                fields_to_present=ctx_dict.get(cls.CTX_FIELDS_TO_PRESENT),
                csv_text=ctx_dict.get(cls.CTX_CSV_TEXT),
                url_protocol_stripping=ctx_dict.get(cls.CTX_PROTOCOL_STRIP_KEY),
                url_truncate=ctx_dict.get(cls.CTX_URL_TRUNCATE_KEY),
                maximum_cidr_size=ctx_dict.get(cls.CTX_MAXIMUM_CIDR),
                no_wildcard_tld=ctx_dict.get(cls.CTX_NO_TLD),
            )
        )

    def get_fields_to_present(self, fields_to_present: str) -> str:
        # based on func ToIoC https://github.com/demisto/server/blob/master/domain/insight.go

        # Fixes legacy query mode silently lost after the first refresh because
        # `get_fields_to_present("")` returned "name,type" instead of ""
        if fields_to_present == "use_legacy_query" or (not fields_to_present and self.out_format == FORMAT_TEXT):
            return ""

        fields_for_format = {
            FORMAT_TEXT: self.FILTER_FIELDS_ON_FORMAT_TEXT,
            FORMAT_CSV: self.FILTER_FIELDS_ON_FORMAT_CSV,
            FORMAT_JSON: self.FILTER_FIELDS_ON_FORMAT_JSON,
            FORMAT_MWG: self.FILTER_FIELDS_ON_FORMAT_MWG,
            FORMAT_PROXYSG: self.FILTER_FIELDS_ON_FORMAT_PROXYSG,
        }
        if self.out_format in [FORMAT_CSV, FORMAT_JSON] and fields_to_present:
            if "all" in argToList(fields_to_present):
                return ""
            else:
                # replace "value" to "name"
                list_fields = argToList(fields_to_present)
                if "value" in list_fields:
                    list_fields[list_fields.index("value")] = "name"
                    fields_to_present = ",".join(list_fields)
                return fields_to_present

        return fields_for_format.get(self.out_format, self.FILTER_FIELDS_ON_FORMAT_TEXT)


""" HELPER FUNCTIONS """


def iterable_to_str(iterable: Iterable, delimiter: str = "\n") -> str:
    """
    Transforms an iterable object to an str, with a custom delimiter between each item
    """
    str_res = ""
    if iterable:
        try:
            iter(iterable)
        except TypeError:
            raise DemistoException(f"non iterable object provided to iterable_to_str: {iterable}")
        str_res = delimiter.join(map(str, iterable))
    return str_res


@debug_function
def log_iocs_file_data(formatted_indicators: str, max_length: int = 100) -> None:
    """Prints a debug log of the first `max_length` characters in the formatted indicators data.

    Args:
        formatted_indicators (str): The IOCs formatted data.
        max_length (int, optional): max # of chars to print. Defaults to 100.
    """
    if formatted_indicators:
        truncated_data = formatted_indicators[:max_length]
        demisto.debug(f"Formatted IOC data (first {max_length} characters):\n{truncated_data}")
    else:
        demisto.debug("No data from IOC search.")


@debug_function
def create_new_edl(request_args: RequestArguments) -> tuple[str, int, dict]:
    """
    Retrieves indicators using the IndicatorsSearcher and formats them according to the request arguments.
    Applies optional cleanup (invalid removal, IP collapsing) and collects statistics on the changes.

    Args:
        request_args (RequestArguments): User-defined settings for formatting, filtering, and pagination.

    Returns:
        tuple[str, int, dict]:
            - str: A formatted string of indicators, ready to be returned in an EDL (one per line).
            - int: The number of original indicators fetched before formatting.
            - dict: Logging statistics about actions taken on indicators (added, dropped, modified).
    """
    limit = request_args.offset + request_args.limit
    offset = request_args.offset
    indicator_searcher = IndicatorsSearcher(
        filter_fields=request_args.fields_to_present, query=request_args.query, size=PAGE_SIZE, limit=limit
    )
    demisto.debug(f"Creating a new EDL file in {request_args.out_format} format")
    formatted_indicators = ""
    new_log_stats = {}
    if request_args.out_format == FORMAT_TEXT:
        if request_args.drop_invalids or request_args.collapse_ips != "Don't Collapse":
            # Because there may be illegal indicators or they may turn into cider, the limit is increased
            indicator_searcher.limit = int(limit * INCREASE_LIMIT)
        new_iocs_file, original_indicators_count = get_indicators_to_format(indicator_searcher, request_args)
        # we collect first all indicators because we need all ips to collapse_ips
        new_iocs_file, new_log_stats = create_text_out_format(new_iocs_file, request_args)
        new_iocs_file.seek(0)
        iocs_set = set()
        for count, line in enumerate(new_iocs_file):
            # continue searching iocs if 1) iocs was truncated or 2) got all available iocs
            if count + 1 > limit:
                break
            if count < offset:
                continue
            elif line not in iocs_set:
                iocs_set.add(line)
                formatted_indicators += line

        demisto.debug(f"Finished formatting ioc. Count after collapse: {len(iocs_set)}")

    else:
        new_iocs_file, original_indicators_count = get_indicators_to_format(indicator_searcher, request_args)
        new_iocs_file.seek(0)
        formatted_indicators = new_iocs_file.read()
    new_iocs_file.close()
    if EXTENSIVE_LOGGING:
        log_iocs_file_data(formatted_indicators)
    return formatted_indicators, original_indicators_count, new_log_stats


@debug_function
def replace_field_name_to_output_format(fields: str):
    """
    convert from the request name field to the name in the response from the server
    """
    fields_list = argToList(fields)
    new_list = []
    for field in fields_list:
        if field == "name":
            field = "value"
        elif field == "type":
            field = "indicator_type"
        new_list.append(field)
    return new_list


@debug_function
def get_indicators_to_format(indicator_searcher: IndicatorsSearcher, request_args: RequestArguments) -> tuple[IO | IO[str], int]:
    """
    Finds indicators using demisto.searchIndicators, and returns the indicators in file written in the requested format
    Parameters:
        indicator_searcher (IndicatorsSearcher): The indicator searcher used to look for indicators
        request_args (RequestArguments):  all the request arguments.
    Returns:
        tuple[IO | IO[str], int]: A tuple of indicators in the requested format (IO | IO[str]),
            and the total number of indicators found by indicator_searcher (int).
    """
    f = tempfile.TemporaryFile(mode="w+t")
    list_fields = replace_field_name_to_output_format(request_args.fields_to_present)
    headers_was_written = False
    files_by_category = {}  # type:Dict
    ioc_counter = 0
    # While iterating over a large number of indicators, the main thread can hold the stdout lock longer than
    # the default 60s timeout, causing the heartbeat thread to fail with "Timeout acquiring stdout lock".
    # Instead of disabling the heartbeat thread (which can make the server consider the container unresponsive),
    # we temporarily raise the stdout lock timeout to 10 minutes so the heartbeat keeps running safely, and
    # restore the original value afterwards. This attribute is dynamically injected by the server runtime; when
    # it is not present (e.g. older servers or unit tests) we skip the tweak entirely so this stays a true no-op.
    # It is not part of the demistomock type stub, so mypy's attr-defined check is suppressed on the assignments.
    stdout_lock_timeout_supported = hasattr(demisto, "_stdout_lock_timeout")
    original_stdout_lock_timeout = getattr(demisto, "_stdout_lock_timeout", 60)
    try:
        if stdout_lock_timeout_supported:
            demisto._stdout_lock_timeout = 600  # type: ignore[attr-defined]  # 10 minutes
            demisto.debug(
                f"Temporarily set demisto._stdout_lock_timeout to 600 seconds "
                f"(was {original_stdout_lock_timeout}) for the indicators iteration."
            )
        for ioc_res in indicator_searcher:
            fetched_iocs = ioc_res.get("iocs") or []
            for ioc in fetched_iocs:
                if EXTENSIVE_LOGGING:
                    demisto.debug(f"Parsing the following indicator: {ioc.get('value')}")

                ioc_counter += 1
                if request_args.out_format == FORMAT_PROXYSG:
                    files_by_category = create_proxysg_out_format(ioc, files_by_category, request_args)

                elif request_args.out_format == FORMAT_MWG:
                    f.write(create_mwg_out_format(ioc, request_args, headers_was_written))
                    headers_was_written = True

                elif request_args.out_format == FORMAT_JSON:
                    f.write(create_json_out_format(list_fields, ioc, request_args, headers_was_written))
                    headers_was_written = True

                elif request_args.out_format == FORMAT_TEXT:
                    # save only the value and type of each indicator
                    f.write(str(json.dumps({"value": ioc.get("value"), "indicator_type": ioc.get("indicator_type")})) + "\n")

                elif request_args.out_format == FORMAT_CSV:
                    f.write(create_csv_out_format(headers_was_written, list_fields, ioc, request_args))
                    headers_was_written = True
                if ioc_counter >= indicator_searcher.limit:
                    break

    except Exception as e:
        demisto.error(f"Error in parsing the indicators, error: {e!s}")
        # 429 error can only be raised when the Elasticsearch instance encountered an error
        if "[429] Failed with error" in str(e):
            version = demisto.demistoVersion()
            # NG + XSIAM can recover from a shutdown
            if version.get("platform") == "x2" or is_demisto_version_ge("8") or version.get("platform") == "unified_platform":
                raise SystemExit("Encountered issue in Elastic Search query. Restarting container and trying again.")
    finally:
        if stdout_lock_timeout_supported:
            demisto._stdout_lock_timeout = original_stdout_lock_timeout  # type: ignore[attr-defined]
            demisto.debug(f"Restored demisto._stdout_lock_timeout to {original_stdout_lock_timeout} seconds.")
    demisto.debug(f"Completed IOC search & format, found {ioc_counter} IOCs.")
    if request_args.out_format == FORMAT_JSON:
        f.write("]")
    elif request_args.out_format == FORMAT_PROXYSG:
        f = create_proxysg_all_category_out_format(f, files_by_category)
    return f, ioc_counter


@debug_function
def create_json_out_format(list_fields: List, indicator: dict, request_args: RequestArguments, not_first_call=True) -> str:
    """format the indicator to json format.

    Args:
        list_fields (list): the fields to return.
        indicator (dict): the indicator info
        request_args (RequestArguments): all the request arguments.
        not_first_call (bool): Indicates if this is the first call to the function.

    Returns:
        An indicator to add to the file in json format.
    """
    if (indicator_value := indicator.get("value")) and indicator.get("indicator_type") == "URL":
        indicator["value"] = url_handler(
            indicator_value, request_args.url_protocol_stripping, request_args.url_port_stripping, request_args.url_truncate
        )
    filtered_json = {}
    if list_fields:
        for field in list_fields:
            value = indicator.get(field) or indicator.get("CustomFields", {}).get(field)
            filtered_json[field] = value
        indicator = filtered_json
    if not_first_call:
        return ", " + json.dumps(indicator)
    return "[" + json.dumps(indicator)


@debug_function
def create_mwg_out_format(indicator: dict, request_args: RequestArguments, headers_was_writen: bool) -> str:
    """format the indicator to mwg format.

    Args:
        indicator (dict): the indicator info
        request_args (RequestArguments): Request Arguments
        headers_was_writen (bool): Whether if the headers was writen to the file.

    Returns:
        An indicator to add to the file in mwg format.
    """
    if (indicator_value := indicator.get("value")) and indicator.get("indicator_type") == "URL":
        indicator["value"] = url_handler(
            indicator_value, request_args.url_protocol_stripping, request_args.url_port_stripping, request_args.url_truncate
        )

    value = '"' + indicator.get("value", "") + '"'
    sources = indicator.get("sourceBrands")
    if sources:
        sources_string = '"' + ",".join(sources) + '"'
    else:
        sources_string = '"from CORTEX XSOAR"'

    if not headers_was_writen:
        mwg_type = request_args.mwg_type
        if isinstance(mwg_type, list):
            mwg_type = mwg_type[0]
        return "type=" + mwg_type + "\n" + value + " " + sources_string
    return "\n" + value + " " + sources_string


@debug_function
def create_proxysg_all_category_out_format(indicators_file: IO, files_by_category: dict):
    """write all indicators to file in proxysg format.

    Args:
        indicators_file (IO): the fields to return.
        files_by_category (dict): all indicators by category

    Returns:
        a file in proxysg format.
    """
    # the first time "define category" will be writen without a new line
    new_line = ""
    for category, category_file in files_by_category.items():
        indicators_file.write(f"{new_line}define category {category}\n")
        new_line = "\n"
        category_file.seek(0)
        indicators_file.write(category_file.read())
        category_file.close()
        indicators_file.write("end")

    return indicators_file


@debug_function
def create_proxysg_out_format(indicator: dict, files_by_category: dict, request_args: RequestArguments) -> dict:
    """format the indicator to proxysg.

    Args:
        indicator (dict): the indicator info
        files_by_category (list): a dict of the formatted indicators by category.
        request_args (RequestArguments): Request Arguments

    Returns:
        a dict of the formatted indicators by category.
    """
    if (indicator_value := indicator.get("value")) and indicator.get("indicator_type") in ["IP", "URL", "Domain", "DomainGlob"]:
        stripped_indicator = url_handler(
            indicator_value, request_args.url_protocol_stripping, request_args.url_port_stripping, request_args.url_truncate
        )
        indicator_proxysg_category = indicator.get("CustomFields", {}).get("proxysgcategory")
        # if a ProxySG Category is set and it is in the category_attribute list or that the attribute list is empty
        # than list add the indicator to it's category list
        if indicator_proxysg_category is not None and (
            indicator_proxysg_category in request_args.category_attribute or len(request_args.category_attribute) == 0
        ):
            # handle indicators in multiple categories
            if isinstance(indicator_proxysg_category, list):
                for category in indicator_proxysg_category:
                    files_by_category = add_indicator_to_category(stripped_indicator, category, files_by_category)
            else:
                files_by_category = add_indicator_to_category(stripped_indicator, indicator_proxysg_category, files_by_category)
        else:
            # if ProxySG Category is not set or does not exist in the category_attribute list
            files_by_category = add_indicator_to_category(stripped_indicator, request_args.category_default, files_by_category)

    return files_by_category


@debug_function
def add_indicator_to_category(indicator: str, category: str, files_by_category: dict):
    if category in files_by_category:
        files_by_category[category].write(indicator + "\n")

    else:
        files_by_category[category] = tempfile.TemporaryFile(mode="w+t")
        files_by_category[category].write(indicator + "\n")

    return files_by_category


@debug_function
def create_csv_out_format(headers_was_writen: bool, list_fields: List, ioc, request_args: RequestArguments):
    """format the ioc to csv format.

    Args:
        headers_was_writen (bool): Whether if the headers was writen to the file.
        list_fields (list): the fields to return.
        ioc (dict): the indicator info
        request_args (RequestArguments): all the request arguments.

    Returns:
        a one indicator to add to the file in csv format.
    """

    if (indicator_value := ioc.get("value")) and ioc.get("indicator_type") == "URL":
        ioc["value"] = url_handler(
            indicator_value, request_args.url_protocol_stripping, request_args.url_port_stripping, request_args.url_truncate
        )
    if not list_fields:
        values = list(ioc.values())
        if not headers_was_writen:
            headers = list(ioc.keys())
            headers_str = list_to_str(headers) + "\n"
            return headers_str + list_to_str(values, map_func=lambda val: f'"{val}"')
        return "\n" + list_to_str(values, map_func=lambda val: f'"{val}"')
    else:
        fields_value_list = []
        for field in list_fields:
            value = ioc.get(field) or ioc.get("CustomFields", {}).get(field)
            fields_value_list.append(value)
        if not headers_was_writen:
            headers_str = request_args.fields_to_present + "\n"
            return headers_str + list_to_str(fields_value_list, map_func=lambda val: f'"{val}"')
        return "\n" + list_to_str(fields_value_list, map_func=lambda val: f'"{val}"')


@debug_function
def ip_groups_to_cidrs(ip_range_groups: Iterable[IPNetwork]) -> set[str]:
    """Collapse IP groups list to CIDRs

    Args:
        ip_range_groups (Iterable[IPNetwork]): An iterable of IPNetwork objects representing connected IPs.

    Returns:
        set[str]: A set of CIDR strings (e.g., {'192.168.1.0/24', '10.0.0.1'})
    """
    ip_ranges: set[str] = set()
    for cidr in ip_range_groups:
        # handle single IPs
        # Use .size property instead of len() to avoid IndexError for very large ranges
        try:
            cidr_size = cidr.size
        except Exception as e:
            demisto.error(f"edl: Failed to collapse IP group {cidr} to CIDRs. Got error: {e}.")
            raise

        if cidr_size == 1:
            # CIDR with a single IP appears with "/32" suffix so handle them differently
            ip_ranges.add(str(cidr[0]))
            continue

        ip_ranges.add(str(cidr))

    return ip_ranges


@debug_function
def ip_groups_to_ranges(ip_range_groups: Iterable[IPRange]) -> set[str]:
    """Collapse IP groups to ranges.

    Args:
        ip_range_groups (Iterable[IPRange]): An iterable of IPRange objects representing connected IPs.

    Returns:
        set[str]: A set of IP range strings (e.g., {'192.168.1.1-192.168.1.255', '10.0.0.1'})
    """
    ip_ranges: set[str] = set()
    for group in ip_range_groups:
        # handle single IPs
        # Use .size property instead of len() to avoid IndexError for very large ranges
        try:
            group_size = group.size
        except Exception as e:
            demisto.error(f"edl: Failed to collapse IP group {group} to range. Got error: {e}.")
            raise

        if group_size == 1:
            ip_ranges.add(str(group[0]))
            continue

        ip_ranges.add(str(group))

    return ip_ranges


@debug_function
def ips_to_ranges(ips: Iterable, collapse_ips: str):
    """Collapse IPs to Ranges or CIDRs.

    Args:
        ips (Iterable): a group of IP strings.
        collapse_ips (str): Whether to collapse to Ranges or CIDRs.

    Returns:
        Set. a list to Ranges or CIDRs.
    """
    invalid_ips = []
    valid_ips = []

    for ip_or_cidr in ips:
        if is_valid_cidr(ip_or_cidr) or is_valid_ip(ip_or_cidr):
            valid_ips.append(ip_or_cidr)
        else:
            invalid_ips.append(ip_or_cidr)

    if collapse_ips == COLLAPSE_TO_RANGES:
        ips_range_groups = IPSet(valid_ips).iter_ipranges()
        collapsed_list = ip_groups_to_ranges(ips_range_groups)
    else:
        cidrs = IPSet(valid_ips).iter_cidrs()
        collapsed_list = ip_groups_to_cidrs(cidrs)

    collapsed_list.update(invalid_ips)
    return collapsed_list


def is_valid_ip(ip: str) -> bool:
    """
    Args:
        ip: IP address
    Returns: True if the string represents an IPv4 or an IPv6 address, false otherwise.
    """
    try:
        ipaddress.IPv4Address(ip)
        return True
    except ValueError:
        try:
            ipaddress.IPv6Address(ip)
            return True
        except ValueError:
            return False


def is_large_cidr(cidr: str, prefix_threshold: int):
    try:
        return IPNetwork(cidr).prefixlen < prefix_threshold
    except Exception as e:
        demisto.debug(str(e))
        return False


def is_valid_cidr(cidr: str) -> bool:
    """
    Args:
        cidr: CIDR string
    Returns: True if the string represents an IPv4 network or an IPv6 network, false otherwise.
    """
    if "/" not in cidr:
        return False
    try:
        ipaddress.IPv4Network(cidr, strict=False)
        return True
    except ValueError:
        try:
            ipaddress.IPv6Network(cidr, strict=False)
            return True
        except ValueError:
            return False


@debug_function
def list_to_str(inp_list: list, delimiter: str = ",", map_func: Callable = str) -> str:
    """
    Transforms a list to an str, with a custom delimiter between each list item
    """
    str_res = ""
    if inp_list:
        if isinstance(inp_list, list):
            str_res = delimiter.join(map(map_func, inp_list))
        else:
            raise AttributeError(f"Invalid inp_list provided to list_to_str: \n{inp_list}")
    return str_res


def log_indicator_line(raw_indicator: str, indicator: str, action: str, reason: str, log_stats: dict) -> dict:
    """Create and store a log line for the indicator.

    Args:
        raw_indicator (str): The raw indicator before modification.
        indicator (str): The indicator after modification.
        action (str): The action preformed, Added / Dropped / Modified.
        reason (str): The reason for the action.
        log_stats (dict): Stats of previous log entries.

    Returns:
        (dict) Updated log stats
    """
    log_line = f"\n{action} | {indicator} | {raw_indicator} | {reason}"
    append_log_edl_data(log_line)
    log_stats[action] = log_stats.get(action, 0) + 1

    return log_stats


def store_log_data(request_args: RequestArguments, created: datetime, log_stats: dict) -> None:
    """Finalizes and writes the full EDL log file by adding a summary header and appending logged indicator actions.

    Args:
        request_args (RequestArguments): The request args, they will be added to the header.
        created (datetime): The time the log was created. This will be added to the header.
        log_stats (dict): A statistics dict for the indicator modifications (e.g. {'Added': 5, 'Dropped': 3, 'Modified': 2}
    """
    log_file_wip = Path(EDL_FULL_LOG_PATH_WIP)
    if log_file_wip.exists():
        added_count = log_stats.get(IndicatorAction.ADDED.value, 0)
        dropped_count = log_stats.get(IndicatorAction.DROPPED.value, 0)
        modified_count = log_stats.get(IndicatorAction.MODIFIED.value, 0)

        total_count = added_count + dropped_count + modified_count

        header = (
            f"# Created new EDL at {created.isoformat()}\n\n"
            f"## Configuration Arguments: {request_args.to_context_json()}\n\n"
            f"## EDL stats: {total_count} indicators in total, {modified_count} modified, {dropped_count} dropped, "
            f"{added_count} added.\n"
            f"\nAction | Indicator | Raw Indicator | Reason"
        )

        with open(EDL_FULL_LOG_PATH, "w+") as new_full_log_file, log_file_wip.open("r") as log_file_data:
            # Finalize the current log: write the headers and the WIP log to full_log_path
            new_full_log_file.write(header)
            for log_line in log_file_data:
                new_full_log_file.write(log_line)

        with open(EDL_FULL_LOG_PATH_WIP, "w+") as log_file_data:
            # Empty WIP log file after finalization.
            log_file_data.seek(0)


@debug_function
def create_text_out_format(iocs: IO, request_args: RequestArguments) -> tuple[Union[IO, IO[str]], dict]:
    """
    Formats a stream of indicators (IOCs) into a standardized text output and logs processing actions.
    Args:
        iocs (IO): A file-like object with JSON lines, each representing an indicator with "value" and "indicator_type".
        request_args (RequestArguments): An object containing formatting preferences such as from the following categories:
            * IP / CIDR:
                1) if collapse_ips, collapse IPs/CIDRs
            * URL:
                1) if drop_invalids, drop invalids (length > 254 or has invalid chars)
                2) if port_stripping, strip ports
                3) if protocol_stripping, strip protocols
                4) if url_truncate, truncate urls
            * Other indicator types:
                1) if drop_invalids, drop invalids (has invalid chars)
                2) if port_stripping, strip ports
    Returns:
        tuple:
            - A temporary file-like object containing the formatted indicators, one per line.
            - A dictionary with statistics about dropped, added, and modified indicators.

    Behavior:
        - Skips indicators missing a value or non-ASCII (if configured).
        - For URLs/domains: applies cleaning (strip protocol/port, truncate) and filtering (length, invalid chars).
        - For IPs/CIDRs: can collapse into ranges or drop if too large.
        - Wildcard domains and TLDs are handled according to config.
        - All actions (add, drop, modify) are logged and counted in the result.
    """
    enforce_ascii = argToBoolean(demisto.params().get("enforce_ascii", False))
    ipv4_formatted_indicators = set()
    ipv6_formatted_indicators = set()
    iocs.seek(0)
    formatted_indicators = tempfile.TemporaryFile(mode="w+t")
    log_stats: dict = {}
    new_line = ""  # For the first time he will not add a new line
    for str_ioc in iocs:
        ioc = json.loads(str_ioc.rstrip())
        indicator_raw = ioc.get("value")
        if not indicator_raw:
            continue
        if enforce_ascii:
            try:
                indicator_raw.encode("ascii")
            except UnicodeEncodeError:
                continue
        ioc_type = ioc.get("indicator_type")

        indicator = indicator_raw

        if ioc_type not in [FeedIndicatorType.IP, FeedIndicatorType.IPv6, FeedIndicatorType.CIDR, FeedIndicatorType.IPv6CIDR]:
            indicator = url_handler(
                indicator_raw, request_args.url_protocol_stripping, request_args.url_port_stripping, request_args.url_truncate
            )

            if request_args.drop_invalids:
                if indicator != _PORT_REMOVAL.sub(_URL_WITHOUT_PORT, indicator) or indicator != _INVALID_TOKEN_REMOVAL.sub(
                    "*", indicator
                ):
                    # check if the indicator held invalid tokens or port
                    log_stats = log_indicator_line(
                        raw_indicator=indicator_raw,
                        indicator=indicator,
                        action=IndicatorAction.DROPPED.value,
                        reason="Invalid tokens or port.",
                        log_stats=log_stats,
                    )
                    continue

                if ioc_type == FeedIndicatorType.URL and len(indicator) >= PAN_OS_MAX_URL_LEN:
                    # URL indicator exceeds allowed length - ignore the indicator
                    log_stats = log_indicator_line(
                        raw_indicator=indicator_raw,
                        indicator=indicator,
                        action=IndicatorAction.DROPPED.value,
                        reason=f"URL exceeds max length {PAN_OS_MAX_URL_LEN}.",
                        log_stats=log_stats,
                    )

                    continue

            # for PAN-OS *.domain.com does not match domain.com
            # we should provide both
            # this could generate more than num entries according to PAGE_SIZE
            # Handle DomainGlob type indicators even when value doesn't start with "*."
            if indicator.startswith("*.") or ioc_type == FeedIndicatorType.DomainGlob:
                domain = str(indicator.lstrip("*."))
                # if we should ignore TLDs and the domain is a TLD
                if request_args.no_wildcard_tld and tldextract.extract(domain).suffix == domain:
                    log_stats = log_indicator_line(
                        raw_indicator=indicator_raw,
                        indicator=domain,
                        action=IndicatorAction.DROPPED.value,
                        reason="Domain is a TLD.",
                        log_stats=log_stats,
                    )
                    continue
                formatted_indicators.write(new_line + domain)
                new_line = "\n"

        if ioc_type in [FeedIndicatorType.CIDR, FeedIndicatorType.IPv6CIDR] and is_large_cidr(
            indicator, request_args.maximum_cidr_size
        ):
            log_stats = log_indicator_line(
                raw_indicator=indicator_raw,
                indicator=indicator,
                action=IndicatorAction.DROPPED.value,
                reason=f"CIDR exceeds max length {request_args.maximum_cidr_size}.",
                log_stats=log_stats,
            )
            continue

        if request_args.collapse_ips != DONT_COLLAPSE and ioc_type in (FeedIndicatorType.IP, FeedIndicatorType.CIDR):
            ipv4_formatted_indicators.add(indicator)

        elif request_args.collapse_ips != DONT_COLLAPSE and ioc_type == FeedIndicatorType.IPv6:
            ipv6_formatted_indicators.add(indicator)

        else:
            formatted_indicators.write(new_line + str(indicator))
            new_line = "\n"
            log_stats = log_indicator_line(
                raw_indicator=indicator_raw,
                indicator=indicator,
                action=IndicatorAction.ADDED.value,
                reason=f"Found new {ioc_type}.",
                log_stats=log_stats,
            )

    iocs.close()
    if len(ipv4_formatted_indicators) > 0:
        ipv4_formatted_indicators_collapsed = ips_to_ranges(ipv4_formatted_indicators, request_args.collapse_ips)
        for ip in ipv4_formatted_indicators_collapsed:
            formatted_indicators.write(new_line + str(ip))
            new_line = "\n"

        for ip in ipv4_formatted_indicators:
            if ip not in ipv4_formatted_indicators_collapsed:
                log_stats = log_indicator_line(
                    raw_indicator=ip,
                    indicator=ip,
                    action=IndicatorAction.MODIFIED.value,
                    reason=f"Collapsed IPv4 {request_args.collapse_ips}.",
                    log_stats=log_stats,
                )

            else:
                log_stats = log_indicator_line(
                    raw_indicator=ip,
                    indicator=ip,
                    action=IndicatorAction.ADDED.value,
                    reason="Found new IPv4.",
                    log_stats=log_stats,
                )

    if len(ipv6_formatted_indicators) > 0:
        ipv6_formatted_indicators_collapsed = ips_to_ranges(ipv6_formatted_indicators, request_args.collapse_ips)
        for ip in ipv6_formatted_indicators_collapsed:
            formatted_indicators.write(new_line + str(ip))
            new_line = "\n"

        for ip in ipv6_formatted_indicators:
            if ip not in ipv6_formatted_indicators_collapsed:
                log_stats = log_indicator_line(
                    raw_indicator=ip,
                    indicator=ip,
                    action=IndicatorAction.MODIFIED.value,
                    reason=f"Collapsed IPv6 {request_args.collapse_ips}.",
                    log_stats=log_stats,
                )

            else:
                log_stats = log_indicator_line(
                    raw_indicator=ip,
                    indicator=ip,
                    action=IndicatorAction.ADDED.value,
                    reason="Found new IPv6.",
                    log_stats=log_stats,
                )
    # Log on the logs files all the modification of the indicators
    return formatted_indicators, log_stats


def url_handler(indicator: str, url_protocol_stripping: bool, url_port_stripping: bool, url_truncate: bool) -> str:
    """
    * URL:
       1) if port_stripping, strip ports
       2) if protocol_stripping, strip protocols
       3) if url_truncate, truncate urls
    """

    # protocol stripping
    if url_protocol_stripping:
        indicator = _PROTOCOL_REMOVAL.sub("", indicator)

    if url_port_stripping:
        # remove port from indicator - from demisto.com:369/rest/of/path -> demisto.com/rest/of/path
        indicator = _PORT_REMOVAL.sub(_URL_WITHOUT_PORT, indicator)

    if url_truncate and len(indicator) >= PAN_OS_MAX_URL_LEN:
        indicator = indicator[0 : PAN_OS_MAX_URL_LEN - 1]

    return indicator


def get_outbound_mimetype(request_args: RequestArguments) -> str:
    """Returns the mimetype of the export_iocs"""
    if request_args.out_format == FORMAT_JSON:
        return MIMETYPE_JSON

    elif request_args.out_format == FORMAT_CSV and not request_args.csv_text:
        return MIMETYPE_CSV

    else:
        return MIMETYPE_TEXT


def append_log_edl_data(log_edl_data: str) -> None:
    """Store the generated log string in the log file.

    Args:
        log_edl_data (str): The generated log data string.
    """
    try:
        with open(EDL_FULL_LOG_PATH_WIP, "a") as last_full_log_file:
            last_full_log_file.write(log_edl_data)

    except Exception as e:
        demisto.debug(f"edl: Error in writing to log file: {e!s}")
        raise e


@debug_function
def get_edl_on_demand() -> tuple[str, int]:
    """
    Use the local file system to store the on-demand result, using a lock to
    limit access to the file from multiple threads.

    Returns:
        tuple[str, int]: A tuple of formatted indicators to display in EDL's response (str),
            and the number of original indicators received from the server before formatting (int).
    """
    global EDL_ON_DEMAND_CACHE_ORIGINAL_SIZE
    ctx = get_integration_context()

    if EDL_ON_DEMAND_KEY in ctx:
        ctx.pop(EDL_ON_DEMAND_KEY, None)
        request_args = RequestArguments.from_context_json(ctx)
        edl_data, EDL_ON_DEMAND_CACHE_ORIGINAL_SIZE, edl_data_stats = create_new_edl(request_args)
        created_time = datetime.now(timezone.utc)
        store_log_data(request_args, created_time, edl_data_stats)

        try:
            demisto.debug("edl: Writing EDL data to cache")

            with open(EDL_ON_DEMAND_CACHE_PATH, "w") as file:
                file.write(edl_data)

        except Exception as e:
            demisto.debug(f"edl: Error in writing to file: {e!s}")
            raise e

        if EXTENSIVE_LOGGING:
            demisto.debug("edl: Finished writing EDL data to cache")
        set_integration_context(ctx)

    else:
        demisto.debug("edl: Reading EDL data from cache")

        try:
            with open(EDL_ON_DEMAND_CACHE_PATH) as file:
                edl_data = file.read()

        except Exception as e:
            demisto.debug(f"edl: Error reading cache file: {e!s}")
            raise e
        demisto.debug("edl: Finished reading EDL data from cache")

    return edl_data, EDL_ON_DEMAND_CACHE_ORIGINAL_SIZE


def validate_basic_authentication(headers: dict, username: str, password: str) -> bool:
    """
    Checks whether the authentication is valid.
    :param headers: The headers of the http request
    :param username: The integration's username
    :param password: The integration's password
    :return: Boolean which indicates whether the authentication is valid or not
    """
    credentials: str = headers.get("Authorization", "")
    if not credentials or "Basic " not in credentials:
        return False
    encoded_credentials: str = credentials.split("Basic ")[1]
    credentials: str = b64decode(encoded_credentials).decode("utf-8")
    if ":" not in credentials:
        return False
    credentials_list = credentials.split(":")
    if len(credentials_list) != 2:
        return False
    user, pwd = credentials_list
    return user == username and pwd == password


def get_bool_arg_or_param(args: dict, params: dict, key: str):
    val = args.get(key)
    return val.lower() == "true" if isinstance(val, str) else params.get(key, False)


""" ROUTE FUNCTIONS """


def authenticate_app(params: dict, request_headers: Any) -> Optional[Response]:
    """Make sure the user is authenticated on API request.

    Args:
        params (dict): The demisto params, where the credentials are stored.
        request_headers: The request headers.

    Returns:
        (Response) '401 Login Required' on failure to authenticate.
        None on success.
    """
    credentials = params.get("credentials", {})
    username: str = credentials.get("identifier", "")
    password: str = credentials.get("password", "")
    if EXTENSIVE_LOGGING:
        demisto.debug("Attempting authentication for incoming request.")
    if username and password:
        headers: dict = cast(dict[Any, Any], request_headers)
        if not validate_basic_authentication(headers, username, password):
            err_msg: str = "Basic authentication failed. Make sure you are using the right credentials."
            demisto.debug(err_msg)
            return Response(
                err_msg,
                status=401,
                mimetype="text/plain",
                headers=[
                    ("WWW-Authenticate", 'Basic realm="Login Required"'),
                ],
            )

    return None


def get_edl_log_file() -> str:
    """Check if edl log file exists, if it does return its contents (str)."""
    edl_data_log = ""
    if os.path.exists(EDL_FULL_LOG_PATH):
        demisto.debug("found log file")
        if os.path.getsize(EDL_FULL_LOG_PATH) > MAX_DISPLAY_LOG_FILE_SIZE:
            if EXTENSIVE_LOGGING:
                demisto.debug("EDL log file too large to display")
            return LARGE_LOG_DISPLAY_MSG

        with open(EDL_FULL_LOG_PATH) as log_file:
            log_file.seek(0)
            edl_data_log = log_file.read()
            log_file.seek(0)

    return edl_data_log


def prepare_response_data(data: str, prepend_str: str, append_str: str) -> str:
    """Prepare data for app response.

    Args:
        data (str): The raw data.
        prepend_str (str): The string to prepend to the data.
        append_str (str): The string to append to the data.

    Returns:
        (str) The prepared data.
    """
    if append_str:
        append_str = append_str.replace("\\n", "\n")
        data = f"{data}{append_str}"
    if prepend_str:
        prepend_str = prepend_str.replace("\\n", "\n")
        data = f"{prepend_str}\n{data}"

    return data


def get_request_id() -> str:
    """Return the per-request correlation id for log lines.

    Reuses the ``X-Request-ID`` header that nginx forwards (the SAME id appears in
    the nginx access log and the WSGI ``wsgi request:``/``wsgi access:`` lines), so
    a single id can be grepped across NGINX -> WSGI -> EDL. Falls back to a fresh
    short uuid if the header is missing (e.g. a direct hit that bypassed nginx).
    """
    try:
        forwarded = request.headers.get("X-Request-ID")
    except RuntimeError:
        # No active Flask request context (e.g. called outside a route).
        demisto.debug("get_request_id called without an active Flask request context; generating a fresh id.")
        forwarded = None
    return forwarded if forwarded else uuid.uuid4().hex[:12]


@APP.route("/", methods=["GET"])
def route_edl() -> Response:
    """
    Main handler for values saved in the integration context
    """
    params = demisto.params()
    rid = get_request_id()
    cache_refresh_rate: str = params.get("cache_refresh_rate")
    start = datetime.now(timezone.utc)
    demisto.info(f"edl: rid={rid} route=/ start handling request")
    auth_resp = authenticate_app(params, request.headers)
    if auth_resp:
        demisto.info(f"edl: rid={rid} authentication failed; returning auth response")
        return auth_resp
    if EXTENSIVE_LOGGING:
        demisto.debug(f"edl: rid={rid} authentication successful")
    request_args = get_request_args(request.args, params)
    on_demand = params.get("on_demand")
    if EXTENSIVE_LOGGING:
        demisto.debug(f"edl: rid={rid} {'Using' if on_demand else 'Not using'} on-demand cache to serve EDL.")

    created = datetime.now(timezone.utc)
    if on_demand:
        edl_data, original_indicators_count = get_edl_on_demand()
    else:
        edl_data, original_indicators_count, edl_data_stats = create_new_edl(request_args)
        store_log_data(request_args, created, edl_data_stats)

    query_time = (datetime.now(timezone.utc) - created).total_seconds()
    etag = f'"{hashlib.sha1(edl_data.encode()).hexdigest()}"'  # nosec
    edl_size = 0

    if edl_data.strip():
        edl_size = edl_data.count("\n") + 1  # add 1 as last line doesn't have a \n

    if (len(edl_data) == 0 and request_args.add_comment_if_empty) or (edl_data == "]" and request_args.add_comment_if_empty):
        edl_data = "# Empty List"

    # if the case there are strings to add to the EDL, add them if the output type is text
    elif request_args.out_format == FORMAT_TEXT:
        edl_data = prepare_response_data(
            data=edl_data, append_str=params.get("append_string"), prepend_str=params.get("prepend_string")
        )
    if EXTENSIVE_LOGGING:
        demisto.debug(f"edl: rid={rid} Final EDL size: {len(edl_data)} characters, {edl_size} lines")

    mimetype = get_outbound_mimetype(request_args)
    max_age = ceil((datetime.now() - dateparser.parse(cache_refresh_rate)).total_seconds())  # type: ignore[operator]

    headers = [
        ("X-EDL-Created", created.isoformat()),
        ("X-EDL-Query-Time-Secs", f"{query_time:.3f}"),
        ("X-EDL-Size", str(edl_size)),
        ("X-EDL-Origin-Size", original_indicators_count),
        ("ETag", etag),
        # Echo the correlation id back so the client and nginx ($sent_http_x_request_id)
        # observe the same id that EDL/WSGI logged.
        ("X-Request-ID", rid),
    ]  # type: ignore[assignment]

    demisto.debug(
        f'edl: rid={rid} Returning response with the following headers:\n'
        f'{[f"{header[0]}: {header[1]}" for header in headers]}'
    )

    resp = Response(edl_data, status=200, mimetype=mimetype, headers=headers)
    resp.cache_control.max_age = max_age
    # number of seconds we are willing to serve stale content when there is an error
    resp.cache_control["stale-if-error"] = "600"

    total_time = (datetime.now(timezone.utc) - start).total_seconds()
    demisto.info(
        f"edl: rid={rid} route=/ done status=200 edl_size={edl_size} "
        f"chars={len(edl_data)} query_time_secs={query_time:.3f} total_secs={total_time:.3f}"
    )

    return resp


@APP.route("/log_download", methods=["GET"])
def log_download() -> Response:
    """Flask route to download the full EDL log file as a zip.
    The log file zipped is located at EDL_FULL_LOG_PATH.

    Return:
        Response: A Flask Response object that sends a ZIP file containing the full log.
    """
    params = demisto.params()
    rid = get_request_id()
    demisto.debug(f"edl: rid={rid} Starting EDL route/log_download handler")
    auth_resp = authenticate_app(params, request.headers)
    if EXTENSIVE_LOGGING:
        demisto.debug(f"edl: rid={rid} authentication successful")
    if auth_resp:
        return auth_resp

    demisto.debug(f"edl: rid={rid} Getting log file to show")

    created = datetime.now(timezone.utc)

    for previous_zip in glob.glob(f"{LOGS_ZIP_FILE_PREFIX}_*.zip"):
        os.remove(previous_zip)
    log_zip_filename = f'{LOGS_ZIP_FILE_PREFIX}_{created.strftime("%Y%m%d-%H%M%S")}.zip'
    zipf = zipfile.ZipFile(log_zip_filename, "w", zipfile.ZIP_DEFLATED)
    zipf.write(EDL_FULL_LOG_PATH)
    zipf.close()
    return send_file(log_zip_filename, mimetype="zip", download_name=log_zip_filename, as_attachment=True)


@APP.route("/log", methods=["GET"])
def route_edl_log() -> Response:
    """
    Flask route to serve the EDL indicators log via HTTP.

    Returns:
        Response: A Flask Response containing either:
            - The log contents as plain text (with optional prepend/append formatting), or
            - A ZIP file download containing the log, if the log is too large to display.
    """
    params = demisto.params()
    rid = get_request_id()

    cache_refresh_rate: str = params.get("cache_refresh_rate")
    demisto.debug(f"edl: rid={rid} Starting EDL route/log handler")
    auth_resp = authenticate_app(params, request.headers)
    if EXTENSIVE_LOGGING:
        demisto.debug(f"edl: rid={rid} authentication successful")
    if auth_resp:
        return auth_resp

    edl_data_log = get_edl_log_file() or "# Empty"
    request_args = get_request_args(request.args, params)
    created = datetime.now(timezone.utc)
    ctx = demisto.getIntegrationContext()

    # If edl_data_log is too large, first return a corresponding message as text.
    # Second, return the log as a file.
    # Alternate between the two via log_as_file context data key.
    if edl_data_log == LARGE_LOG_DISPLAY_MSG:
        # If we should return the log as a file this time
        if ctx.get("log_as_file", False):
            # Reset the log_as_file context key. Next time a message will be returned.
            ctx["log_as_file"] = False
            set_integration_context(ctx)
            # Remove previous zip versions of the log file if they exist.
            for previous_zip in glob.glob(f"{LOGS_ZIP_FILE_PREFIX}_*.zip"):
                os.remove(previous_zip)
            # zip the current log file and return it.
            log_zip_filename = f'{LOGS_ZIP_FILE_PREFIX}_{created.strftime("%Y%m%d-%H%M%S")}.zip'
            zipf = zipfile.ZipFile(log_zip_filename, "w", zipfile.ZIP_DEFLATED)
            zipf.write(EDL_FULL_LOG_PATH)
            zipf.close()
            return send_file(log_zip_filename, mimetype="zip", download_name=log_zip_filename, as_attachment=True)
        else:
            # Reset the log_as_file context key. Next time a file will be returned.
            ctx["log_as_file"] = True
            set_integration_context(ctx)

    if request_args.out_format == FORMAT_TEXT and edl_data_log not in ["# Empty", LARGE_LOG_DISPLAY_MSG]:
        ctx["log_as_file"] = False
        set_integration_context(ctx)
        edl_data_log = prepare_response_data(
            data=edl_data_log, append_str=params.get("append_string"), prepend_str=params.get("prepend_string")
        )

    etag = f'"{hashlib.sha3_256(edl_data_log.encode()).hexdigest()}"'
    headers = [("X-EDL-LOG-Request-Created", created.isoformat()), ("ETag", etag)]  # type: ignore[assignment]
    headers_str = f'{[f"{header[0]}: {header[1]}" for header in headers]}'
    demisto.debug(f"edl: Returning log response with the following headers:\n{headers_str}")
    max_age = ceil((datetime.now() - dateparser.parse(cache_refresh_rate)).total_seconds())  # type: ignore[operator]
    if edl_data_log == "# Empty":
        # If log file content was not created yet, refresh after 15 seconds.
        # If EDL indicator list refresh rate is less than 30s, refresh the log after half of the time.
        # This way, the corresponding log will be shown after at most 15 seconds.
        max_age = min(ceil(max_age / 2), 15)
    if edl_data_log == LARGE_LOG_DISPLAY_MSG:
        max_age = 0

    mimetype = get_outbound_mimetype(request_args)
    resp = Response(edl_data_log, status=200, mimetype=mimetype, headers=headers)
    resp.cache_control.max_age = max_age
    # number of seconds we are willing to serve stale content when there is an error
    resp.cache_control["stale-if-error"] = "600"

    return resp


def get_request_args(request_args: dict, params: dict) -> RequestArguments:
    """
    Processing a flask request arguments and generates a RequestArguments instance from it.
    Args:
        request_args: Flask request arguments
        params: Integration configuration parameters

    Returns:
        RequestArguments instance with processed arguments
    """
    if EXTENSIVE_LOGGING:
        demisto.debug("Parsing EDL request args...")
    limit = try_parse_integer(request_args.get("n", params.get("edl_size") or 10000), EDL_LIMIT_ERR_MSG)
    offset = try_parse_integer(request_args.get("s", 0), EDL_OFFSET_ERR_MSG)
    out_format = request_args.get("v", params.get("format", FORMAT_TEXT))
    query = request_args.get("q", params.get("indicators_query") or "")
    mwg_type = request_args.get("t", params.get("mwg_type", "string"))
    strip_port = request_args.get("sp", params.get("url_port_stripping") or False)
    strip_protocol = request_args.get("pr", params.get("url_protocol_stripping") or False)
    drop_invalids = request_args.get("di", params.get("drop_invalids") or False)
    category_default = request_args.get("cd", params.get("category_default", "bc_category"))
    category_attribute = request_args.get("ca", params.get("category_attribute", ""))
    collapse_ips = request_args.get("tr", params.get("collapse_ips", DONT_COLLAPSE))
    csv_text = request_args.get("tx", params.get("csv_text", False))
    add_comment_if_empty = request_args.get("ce", params.get("add_comment_if_empty", True))
    fields_to_present = request_args.get("fi", params.get("fields_filter", ""))
    url_truncate = request_args.get("ut", params.get("url_truncate", ""))
    maximum_cidr_size = try_parse_integer(
        request_args.get("mc", params.get("maximum_cidr_size", MAXIMUM_CIDR_SIZE_DEFAULT)), EDL_CIDR_SIZR_MSG
    )
    no_wildcard_tld = argToBoolean(request_args.get("nt", params.get("no_wildcard_tld")))

    # handle flags
    if drop_invalids == "":
        drop_invalids = True

    if strip_port == "":
        strip_port = True

    if strip_protocol == "":
        strip_port = True

    if collapse_ips not in [DONT_COLLAPSE, COLLAPSE_TO_CIDR, COLLAPSE_TO_RANGES]:
        collapse_ips = try_parse_integer(collapse_ips, EDL_COLLAPSE_ERR_MSG)

        if collapse_ips not in [0, 1, 2]:
            raise DemistoException(EDL_COLLAPSE_ERR_MSG)

        collapse_options = {0: DONT_COLLAPSE, 1: COLLAPSE_TO_RANGES, 2: COLLAPSE_TO_CIDR}
        collapse_ips = collapse_options[collapse_ips]
    if out_format not in [
        FORMAT_PROXYSG,
        FORMAT_TEXT,
        FORMAT_JSON,
        FORMAT_CSV,
        FORMAT_MWG,
        FORMAT_ARG_BLUECOAT,
        FORMAT_ARG_MWG,
        FORMAT_ARG_PROXYSG,
    ]:
        raise DemistoException(EDL_FORMAT_ERR_MSG)

    elif out_format in [FORMAT_ARG_PROXYSG, FORMAT_ARG_BLUECOAT]:
        out_format = FORMAT_PROXYSG

    elif out_format == FORMAT_ARG_MWG:
        out_format = FORMAT_MWG

    if out_format == FORMAT_MWG and mwg_type not in MWG_TYPE_OPTIONS:
        raise DemistoException(EDL_MWG_TYPE_ERR_MSG)

    if params.get("use_legacy_query"):
        # workaround for "msgpack: invalid code" error
        demisto.info(
            "Note: You are using a legacy query, it may have an impact on the performance of the integration."
            "This parameter is deprecated, make sure to adjust your queries accordingly."
        )
        fields_to_present = "use_legacy_query"

    if query and request_args.get("q"):
        demisto.debug(
            "Adjusting the number of exported indicators if above 100,000, due to using the q URL inline parameter."
            "For more information, review the documentation."
        )
        limit = min(limit, MAX_LIST_SIZE_WITH_URL_QUERY)

    demisto.debug(
        f"RequestArguments resolved:\n"
        f"limit={limit}, offset={offset}, format={out_format}, query='{query}', "
        f"mwg_type={mwg_type}, collapse_ips={collapse_ips}, drop_invalids={drop_invalids}, "
        f"strip_port={strip_port}, strip_protocol={strip_protocol}, "
        f"url_truncate={url_truncate}, maximum_cidr_size={maximum_cidr_size}, no_wildcard_tld={no_wildcard_tld}, "
        f"fields_to_present={fields_to_present}, category_default={category_default}, "
        f"category_attribute={category_attribute}, csv_text={csv_text}, comment_if_empty={add_comment_if_empty}"
    )

    return RequestArguments(
        query,
        out_format,
        limit,
        offset,
        strip_port,
        drop_invalids,
        collapse_ips,
        add_comment_if_empty,
        mwg_type,
        category_default,
        category_attribute,
        fields_to_present,
        csv_text,
        strip_protocol,
        url_truncate,
        maximum_cidr_size,
        no_wildcard_tld,
    )


""" COMMAND FUNCTIONS """


def test_module(_: dict, params: dict):
    """
    Validates:
        1. Valid port.
        2. Valid cache_refresh_rate
    """
    if not params.get("longRunningPort"):
        params["longRunningPort"] = "1111"
    get_params_port(params)
    on_demand = params.get("on_demand", None)
    if not on_demand:
        try_parse_integer(params.get("edl_size"), EDL_LIMIT_ERR_MSG)  # validate EDL Size was set
        cache_refresh_rate = params.get("cache_refresh_rate", "")
        if not cache_refresh_rate:
            raise ValueError(EDL_MISSING_REFRESH_ERR_MSG)
        # validate cache_refresh_rate value
        range_split = cache_refresh_rate.split(" ")
        if len(range_split) != 2:
            raise ValueError(EDL_MISSING_REFRESH_ERR_MSG)
        try_parse_integer(range_split[0], "Invalid time value for the Refresh Rate. Must be a valid integer.")
        if range_split[1] not in ["minute", "minutes", "hour", "hours", "day", "days", "month", "months", "year", "years"]:
            raise ValueError("Invalid time unit for the Refresh Rate. Must be minutes, hours, days, months, or years.")
        parse_date_range(cache_refresh_rate, to_timestamp=True)
    run_long_running(params, is_test=True)
    return "ok", {}, {}


@debug_function
def update_edl_command(args: dict, params: dict):
    """
    Updates the context to update the EDL values on demand the next time it runs.
    """
    on_demand = params.get("on_demand")
    if not on_demand:
        raise DemistoException('"Update EDL On Demand" is off. If you want to update the EDL manually please toggle it on.')
    limit = try_parse_integer(args.get("edl_size", params.get("edl_size")), EDL_LIMIT_ERR_MSG)
    query = args.get("query", "")
    collapse_ips = args.get("collapse_ips", DONT_COLLAPSE)
    url_port_stripping = get_bool_arg_or_param(args, params, "url_port_stripping")
    strip_protocol = get_bool_arg_or_param(args, params, "url_protocol_stripping")
    drop_invalids = get_bool_arg_or_param(args, params, "drop_invalids")
    add_comment_if_empty = get_bool_arg_or_param(args, params, "add_comment_if_empty")
    offset = try_parse_integer(args.get("offset", 0), EDL_OFFSET_ERR_MSG)
    mwg_type = args.get("mwg_type", "string")
    category_default = args.get("category_default", "bc_category")
    category_attribute = args.get("category_attribute", "")
    fields_to_present = args.get("fields_filter", "")
    out_format = args.get("format", FORMAT_TEXT)
    csv_text = get_bool_arg_or_param(args, params, "csv_text") == "True"
    url_truncate = get_bool_arg_or_param(args, params, "url_truncate")
    maximum_cidr_size = try_parse_integer(params.get("maximum_cidr_size", MAXIMUM_CIDR_SIZE_DEFAULT), EDL_CIDR_SIZR_MSG)
    no_wildcard_tld = argToBoolean(params.get("no_wildcard_tld", False))

    if params.get("use_legacy_query"):
        demisto.info(
            "Note: You are using a legacy query, it may have an impact on the performance of the integration."
            "This parameter is deprecated, make sure to adjust your queries accordingly."
        )
        # workaround for "msgpack: invalid code" error
        fields_to_present = "use_legacy_query"

    request_args = RequestArguments(
        query,
        out_format,
        limit,
        offset,
        url_port_stripping,
        drop_invalids,
        collapse_ips,
        add_comment_if_empty,
        mwg_type,
        category_default,
        category_attribute,
        fields_to_present,
        csv_text,
        strip_protocol,
        url_truncate,
        maximum_cidr_size,
        no_wildcard_tld,
    )

    ctx = request_args.to_context_json()
    ctx[EDL_ON_DEMAND_KEY] = True
    set_integration_context(ctx)
    hr = "EDL will be updated the next time you access it."

    if not query:
        warning = "\n**Warning**: Updating EDL, while not specifying a query, may load unwanted indicators."

        if param_query := params.get("query"):
            warning += f" Hint: use {param_query} to update indicators using the configured integration instance parameter."

        hr += warning
        demisto.info(warning)

    return hr, {}, {}


def initialize_edl_context(params: dict):
    global EDL_ON_DEMAND_CACHE_PATH
    limit = try_parse_integer(params.get("edl_size"), EDL_LIMIT_ERR_MSG)
    query = params.get("indicators_query", "")
    collapse_ips = params.get("collapse_ips", DONT_COLLAPSE)
    url_port_stripping = params.get("url_port_stripping", False)
    url_protocol_stripping = params.get("url_protocol_stripping", False)
    drop_invalids = params.get("drop_invalids", False)
    add_comment_if_empty = params.get("add_comment_if_empty", True)
    mwg_type = params.get("mwg_type", "string")
    category_default = params.get("category_default", "bc_category")
    category_attribute = params.get("category_attribute", "")
    fields_to_present = params.get("fields_filter", "")
    out_format = params.get("format", FORMAT_TEXT)
    csv_text = argToBoolean(params.get("csv_text", False))
    url_truncate = params.get("url_truncate", False)
    maximum_cidr_size = try_parse_integer(params.get("maximum_cidr_size", MAXIMUM_CIDR_SIZE_DEFAULT), EDL_CIDR_SIZR_MSG)
    no_wildcard_tld = argToBoolean(params.get("no_wildcard_tld", False))

    if params.get("use_legacy_query"):
        # workaround for "msgpack: invalid code" error
        demisto.info(
            "Note: You are using a legacy query, it may have an impact on the performance of the integration."
            "This parameter is getting deprecated, make sure to adjust your queries accordingly."
        )
        fields_to_present = "use_legacy_query"
    offset = 0
    request_args = RequestArguments(
        query,
        out_format,
        limit,
        offset,
        url_port_stripping,
        drop_invalids,
        collapse_ips,
        add_comment_if_empty,
        mwg_type,
        category_default,
        category_attribute,
        fields_to_present,
        csv_text,
        url_protocol_stripping,
        url_truncate,
        maximum_cidr_size,
        no_wildcard_tld,
    )

    EDL_ON_DEMAND_CACHE_PATH = demisto.uniqueFile()
    demisto.debug(f"The full log path: {EDL_FULL_LOG_PATH}")
    ctx = request_args.to_context_json()
    ctx[EDL_ON_DEMAND_KEY] = True
    set_integration_context(ctx)
    demisto.debug("Setting context data on demand to true.")


def check_platform_and_version(params: dict) -> bool:
    """
    Args:
        - params: The demisto params from the integration configuration.
    Returns:
        (bool): True if the platform is xsoar or xsoar hosted and no port specified, false otherwise
    """
    platform = demisto.demistoVersion().get("platform", "xsoar")
    return platform in ["xsoar", "xsoar_hosted"] and not is_demisto_version_ge("8.0.0") and not params.get("longRunningPort")


def main():  # pragma: no cover
    """
    Main
    """
    global PAGE_SIZE, EXTENSIVE_LOGGING
    params = demisto.params()
    try:
        PAGE_SIZE = max(1, int(params.get("page_size") or PAGE_SIZE))
    except ValueError:
        demisto.debug(f'Non integer "page_size" provided: {params.get("page_size")}. defaulting to {PAGE_SIZE}')
    credentials = params.get("credentials") if params.get("credentials") else {}
    username: str = credentials.get("identifier", "")
    password: str = credentials.get("password", "")

    EXTENSIVE_LOGGING = params.get("extensive_logging", False)

    if (username and not password) or (password and not username):
        err_msg: str = "If using credentials, both username and password should be provided."
        demisto.debug(err_msg)
        raise DemistoException(err_msg)

    command = demisto.command()
    demisto.debug(f"Command being called is {command}")
    commands = {"test-module": test_module, "edl-update": update_edl_command, "export-indicators-list-update": update_edl_command}

    try:
        if check_platform_and_version(params):
            raise DemistoException("Please specify a Listen Port, in the integration configuration")

        initialize_edl_context(params)
        if EXTENSIVE_LOGGING:
            demisto.debug("EDL context initialized.")
        if command == "long-running-execution":
            run_long_running(params)
        elif command in commands:
            readable_output, outputs, raw_response = commands[command](demisto.args(), params)
            return_outputs(readable_output, outputs, raw_response)
        else:
            raise NotImplementedError(f'Command "{command}" is not implemented.')
    except Exception as e:
        err_msg = f"Error in {INTEGRATION_NAME} Integration [{e}]"
        return_error(err_msg)


from NGINXApiModule import *  # noqa: E402

if __name__ in ["__main__", "__builtin__", "builtins"]:
    register_signal_handler_profiling_dump(profiling_dump_rows_limit=PROFILING_DUMP_ROWS_LIMIT)
    main()