Flashpoint Vulnerability Feed

"""Flashpoint Vulnerability Feed Integration for Cortex XSOAR"""

""" IMPORTS """


import urllib3  # noqa: E402

from CommonServerPython import *  # noqa # pylint: disable=unused-wildcard-import
from CommonServerUserPython import *  # noqa

# Disable insecure warnings
urllib3.disable_warnings()  # pylint: disable=no-member

""" CONSTANTS """

DEFAULT_API_PATH = "api.flashpoint.io"
DEFAULT_PLATFORM_PATH = "https://app.flashpoint.io"
DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
DEFAULT_FIRST_FETCH = "3 days"
PAGE_SIZE = 400
DEFAULT_LIMIT = 20
TIMEOUT = 60
STATUS_LIST_TO_RETRY = (429, *(status_code for status_code in requests.status_codes._codes if status_code >= 500))  # type: ignore
OK_CODES = (200, 201)
TOTAL_RETRIES = 4
BACKOFF_FACTOR = 7.5  # Sleep for [0s, 15s, 30s, 60s] between retries.
DEFAULT_INDICATOR_TYPE = "Flashpoint Vulnerability"
DEFAULT_FROM = 0
CURRENT_TIME = "now"
VALID_SEVERITIES = ("critical", "high", "medium", "low", "informational")
REFERENCE_TYPES_MAPPING = {
    "bugtraq id": "bid",
    "bug tracker": "bugtracker",
    "immunity canvas": "canvas",
    "immunity canvas (d2exploitpack)": "canvasd2",
    "immunity canvas (white phosphorus)": "canvaswp",
    "cert": "cert",
    "cert vu": "certvu",
    "ciac advisory": "ciac",
    "cve id": "cveid",
    "d2 elliot": "elliot",
    "exploit activity": "exploitactivity",
    "exploit database": "exploitdb",
    "flashpoint": "flashpoint",
    "generic exploit url": "gexploiturl",
    "generic informational url": "ginformurl",
    "disa iava": "iava",
    "iss x-force id": "iss",
    "japan vulnerability notes": "jpcert",
    "keyword": "keyword",
    "mail list post": "mailpost",
    "metasploit url": "metasploit",
    "microsoft knowledge base article": "mskb",
    "microsoft security bulletin": "mssb",
    "nessus script id": "nessus",
    "news article": "news",
    "nikto item id": "nikto",
    "other advisory url": "oadvisoryurl",
    "other solution url": "osolutionurl",
    "oval id": "oval",
    "packet storm": "packetstorm",
    "redhat rhsa": "redhat",
    "related vulndb id": "relvulndbid",
    "scip vuldb id": "scipid",
    "secunia advisory id": "secunia",
    "security tracker": "securitytracker",
    "snort signature id": "snort",
    "tenable pvs": "tenpvs",
    "us-cert cyber security alert": "uscert",
    "vendor specific advisory url": "vendadvisoryurl",
    "vendor specific solution url": "vendsolutionurl",
    "vendor url": "vendurl",
    "vendor specific news/changelog entry": "vsnewschangelog",
    "vupen advisory": "vupen",
}
LOCATION_MAPPING = {
    "context dependent": "context",
    "dial-up access required": "dialup",
    "local access required": "local",
    "legacy: local / remote": "local_remote",
    "mobile phone / hand-held device": "mobile",
    "physical access required": "physical",
    "remote / network access": "remote",
    "location unknown": "unknown",
    "wireless vector": "wireless",
}
VALID_RANSOMWARE_SCORE = ("critical", "high", "medium", "low")
ATTACK_TYPE_MAPPING = {
    "authentication management": "auth_manage",
    "cryptographic": "crypt",
    "infrastructure": "infrastruct",
    "input manipulation": "input_manip",
    "misconfiguration": "miss_config",
    "man-in-the-middle (mitm)": "mitm",
    "other": "other",
    "race condition": "race",
    "attack type unknown": "unknown",
}
PLATFORM_SUFFIX = "/vuln/vulnerabilities/{}"

URL_SUFFIX = {
    "VULNERABILITIES": "/vulnerability-intelligence/v1/vulnerabilities",
}

MESSAGES = {
    "NO_PARAM_PROVIDED": "Please provide the {}.",
    "INVALID_MULTI_PARAMS_PROVIDED": "Invalid values: {} for '{}' parameter provided. Valid values are: {}.",
    "INVALID_INT_PARAMS_PROVIDED": "Invalid integer values: {} provided for '{}'.",
    "INVALID_EPSS_SCORE": "{} must be a float between 0 and 1.",
    "INVALID_CVSS_SCORE": "{} must be a float between 0 and 10.",
    "INVALID_SCORE_RANGE": "{} must be less than or equal to {}.",
    "NO_INDICATORS_FOUND": "No indicators were found for the given argument(s).",
    "INVALID_LIMIT_PROVIDED": "Invalid limit value: {}. 'limit' should be greater than 0 and "
    "less than or equal to {} to avoid timeout errors.",
    "INVALID_FROM_PROVIDED": "Invalid from value: {}. 'from' should be a non-negative integer.",
    "INVALID_TIME_INTERVAL": "{} parameter must be before {} parameter.({} - {})",
}

HTTP_ERRORS = {
    400: "Bad request: {}, Detail: {}",
    401: "Authentication error: Please provide valid API Key.",
    403: "Forbidden: Please provide valid API Key.",
    404: "Resource not found: Invalid endpoint was called.",
    500: "Internal server error: Please try again after some time.",
}

INTEGRATION_VERSION = get_pack_version() or "1.0.0"
INTEGRATION_PLATFORM = "Cortex XSOAR"
DEFAULT_API_URL = "https://api.flashpoint.io"

# Field mapping for API response
VULNERABILITY_FEED_MAPPING = {
    "flashpointvulnerabilityfeedalternateid": {"path": "alternate_vulndb_id", "type": "number"},
    "flashpointvulnerabilityfeedcveids": {"path": "cve_ids", "type": "multiSelect"},
    "flashpointvulnerabilityfeedlastmodifiedat": {"path": "timelines.last_modified_at", "type": "date"},
    "flashpointvulnerabilityfeedpublishedat": {"path": "timelines.published_at", "type": "date"},
    "flashpointvulnerabilityfeeddisclosedat": {"path": "timelines.disclosed_at", "type": "date"},
    "flashpointvulnerabilityfeeddiscoveredat": {"path": "timelines.discovered_at", "type": "date"},
    "flashpointvulnerabilityfeedexploitedinthewildat": {"path": "timelines.exploited_in_the_wild_at", "type": "date"},
    "flashpointvulnerabilityfeedexploitpublishedat": {"path": "timelines.exploit_published_at", "type": "date"},
    "flashpointvulnerabilityfeedsolutionprovidedat": {"path": "timelines.solution_provided_at", "type": "date"},
    "flashpointvulnerabilityfeedthirdpartysolutionprovidedat": {
        "path": "timelines.third_party_solution_provided_at",
        "type": "date",
    },
    "flashpointvulnerabilityfeedtimetoexploit": {"path": "timelines.time_to_exploit", "type": "shortText"},
    "flashpointvulnerabilityfeedtimetopatch": {"path": "timelines.time_to_patch", "type": "shortText"},
    "flashpointvulnerabilityfeedtimeunpatched": {"path": "timelines.time_unpatched", "type": "shortText"},
    "flashpointvulnerabilityfeedtotaltimetoexploit": {"path": "timelines.total_time_to_exploit", "type": "shortText"},
    "flashpointvulnerabilityfeedtotaltimetopatch": {"path": "timelines.total_time_to_patch", "type": "shortText"},
    "flashpointvulnerabilityfeedvendoracknowledgedat": {"path": "timelines.vendor_acknowledged_at", "type": "date"},
    "flashpointvulnerabilityfeedvendorinformedat": {"path": "timelines.vendor_informed_at", "type": "date"},
    "flashpointvulnerabilityfeedvendorresponsetime": {"path": "timelines.vendor_response_time", "type": "shortText"},
    "flashpointvulnerabilityfeedchangelog": {"path": "changelog", "type": "grid"},
    "flashpointvulnerabilityfeedclassifications": {"path": "classifications", "type": "grid"},
    "flashpointvulnerabilityfeedcvssv2s": {"path": "cvss_v2s", "type": "grid"},
    "flashpointvulnerabilityfeedcvssv3s": {"path": "cvss_v3s", "type": "grid"},
    "flashpointvulnerabilityfeedcvssv4s": {"path": "cvss_v4s", "type": "grid"},
    "flashpointvulnerabilityfeedcwes": {"path": "cwes", "type": "grid"},
    "flashpointvulnerabilityfeedepssscore": {"path": "scores.epss_score", "type": "shortText"},
    "flashpointvulnerabilityfeedepssv1score": {"path": "scores.epss_v1_score", "type": "shortText"},
    "flashpointvulnerabilityfeedransomwarescore": {"path": "scores.ransomware_score", "type": "shortText"},
    "flashpointvulnerabilityfeedseverity": {"path": "scores.severity", "type": "shortText"},
    "flashpointvulnerabilityfeedsocialriskscores": {"path": "scores.social_risk_scores", "type": "grid"},
    "flashpointvulnerabilityfeedexploitscount": {"path": "exploits_count", "type": "number"},
    "flashpointvulnerabilityfeedexploits": {"path": "exploits", "type": "grid"},
    "flashpointvulnerabilityfeedexternalreferences": {"path": "ext_references", "type": "grid"},
    "flashpointvulnerabilityfeedid": {"path": "id", "type": "number"},
    "flashpointvulnerabilityfeedkeywords": {"path": "keywords", "type": "shortText"},
    "flashpointvulnerabilityfeednvdadditionalinformation": {"path": "nvd_additional_information", "type": "grid"},
    "flashpointvulnerabilityfeedproductsandvendors": {"path": "products", "type": "grid"},
    "flashpointvulnerabilityfeedsolution": {"path": "solution", "type": "longText"},
    "flashpointvulnerabilityfeedstatus": {"path": "vuln_status", "type": "shortText"},
    "flashpointvulnerabilityfeedtechnicaldescription": {"path": "technical_description", "type": "longText"},
    "flashpointvulnerabilityfeedtitle": {"path": "title", "type": "shortText"},
}


class Client(BaseClient):
    """
    Client class to interact with the Flashpoint Vulnerability Intelligence API.
    Should not contain Cortex XSOAR logic. Should do requests and return data.
    """

    def __init__(self, base_url: str, api_key: str, verify: bool, proxy: bool):
        """
        Initialize the Client class.

        :type base_url: ``str``
        :param base_url: Base server address with suffix, for example: https://api.flashpoint.io

        :type api_key: ``str``
        :param api_key: API key for authentication.

        :type verify: ``bool``
        :param verify: Whether to verify SSL certificate.

        :type proxy: ``bool``
        :param proxy: Whether to use proxy settings.
        """
        headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
            "X-FP-IntegrationPlatform": INTEGRATION_PLATFORM,
            "X-FP-IntegrationPlatformVersion": get_demisto_version_as_str(),
            "X-FP-IntegrationVersion": INTEGRATION_VERSION,
        }

        if DEFAULT_API_PATH in base_url:
            self.platform_url = DEFAULT_PLATFORM_PATH
        else:
            self.platform_url = base_url

        super().__init__(
            base_url=base_url,
            headers=headers,
            verify=verify,
            proxy=proxy,
        )

    def http_request(
        self, url_suffix: str, params: Optional[dict[str, Any]] = None, method: str = "GET", resp_type: str = "json"
    ) -> Any:
        """
        Get http response based on url and given parameters.

        :type url_suffix: ``str``
        :param url_suffix: url encoded url suffix.

        :type params: ``Optional[dict[str, Any]]``
        :param params: URL parameters to specify the query.

        :type method: ``str``
        :param method: Specify http methods.

        :type resp_type: ``str``
        :param resp_type: Response type to be returned.

        :return: http response on json.
        """
        resp = self._http_request(
            method=method,
            url_suffix=url_suffix,
            params=params,
            ok_codes=OK_CODES,
            error_handler=self.handle_errors,
            status_list_to_retry=STATUS_LIST_TO_RETRY,
            retries=TOTAL_RETRIES,
            backoff_factor=BACKOFF_FACTOR,
            timeout=TIMEOUT,
            resp_type=resp_type,
        )

        return resp

    def list_vulnerabilities(self, params: dict[str, Any] | None = None, resp_type: str = "json") -> dict[str, Any]:
        """
        List vulnerabilities from the Flashpoint Vulnerability Intelligence API.

        :type params: ``dict``
        :param params: Query parameters for filtering vulnerabilities.

        :return: dict containing the API response with vulnerabilities data.
        :rtype: ``Dict[str, Any]``
        """
        return self.http_request(
            method="GET",
            url_suffix=URL_SUFFIX["VULNERABILITIES"],
            params=params,
            resp_type=resp_type,
        )

    def get_vulnerability(self, id: str, resp_type: str = "json") -> dict[str, Any]:
        """
        Get a specific vulnerability from the Flashpoint Vulnerability Intelligence API.

        :type id: ``str``
        :param id: ID of the vulnerability to retrieve.

        :return: dict containing the API response with vulnerabilities data.
        :rtype: ``Dict[str, Any]``
        """
        return self.http_request(
            method="GET",
            url_suffix=f"{URL_SUFFIX['VULNERABILITIES']}/{id}",
            resp_type=resp_type,
        )

    @staticmethod
    def handle_errors(response) -> None:
        """
        Handle HTTP errors from API responses.

        :type response: ``requests.Response``
        :param response: The HTTP response object.
        """
        status = response.status_code
        if status in HTTP_ERRORS:
            if status == 400:
                try:
                    res = response.json()
                    error_type = res.get("type", "")
                    detail = res.get("errors", [{}])[0].get("detail", "")
                except ValueError:
                    error_type = ""
                    detail = response.text
                raise DemistoException(HTTP_ERRORS[status].format(error_type, detail))
            raise DemistoException(f"{HTTP_ERRORS[status]}\n\n{response.text}")
        else:
            response.raise_for_status()


""" HELPER FUNCTIONS """


def remove_nulls_from_nested_dictionary(data):
    """
    Remove Null and empty values from a dictionary recursively, including nested dictionaries and lists.
    (updating the given dictionary in-place)

    :type data: ``dict`` or ``list``
    :param data: The data to be cleaned (required)

    :return: No data returned
    :rtype: ``None``
    """
    if isinstance(data, dict):
        list_of_keys = list(data.keys())[:]
        for key in list_of_keys:
            value = data[key]

            if isinstance(value, (dict | list)):
                remove_nulls_from_nested_dictionary(value)
                if not value:
                    del data[key]
            elif value in ("", None, [], {}, ()):
                del data[key]

    elif isinstance(data, list):
        indices_to_remove = []
        for i, item in enumerate(data):
            if isinstance(item, (dict | list)):
                remove_nulls_from_nested_dictionary(item)
                if not item:
                    indices_to_remove.append(i)
            elif item in ("", None, [], {}, ()):
                indices_to_remove.append(i)

        for index in reversed(indices_to_remove):
            data.pop(index)


def remove_space_from_args(args: dict) -> dict:
    """
    Remove leading/trailing spaces from string arguments.

    :type args: ``dict``
    :param args: Arguments dictionary.

    :return: Arguments dictionary with trimmed string values.
    :rtype: ``dict``
    """
    for key in args:
        if isinstance(args[key], str):
            args[key] = args[key].strip()
    return args


def validate_params(params: dict) -> None:
    """
    Validate the integration parameters.

    :type params: ``dict``
    :param params: Integration parameters.

    :raises DemistoException: If required parameters are missing.
    """
    if not params.get("url"):
        raise DemistoException(MESSAGES["NO_PARAM_PROVIDED"].format("Server URL"))
    if not str(params.get("credentials", {}).get("password", "")).strip():
        raise DemistoException(MESSAGES["NO_PARAM_PROVIDED"].format("API Key"))
    if not params.get("feedReliability"):
        raise DemistoException(MESSAGES["NO_PARAM_PROVIDED"].format("Source Reliability"))


def map_indicator_fields(resp: dict, indicator_obj: dict) -> None:
    """
    Map fields of indicators from the API response.

    :type resp: ``dict``
    :param resp: raw response of indicator.

    :type indicator_obj: ``dict``
    :param indicator_obj: created indicator object.

    :return: None.
    """
    for key, value in VULNERABILITY_FEED_MAPPING.items():
        path = value.get("path", "")
        indicator_obj["fields"][key] = demisto.get(resp, path)


def validate_epss_score(min_epss: Optional[str], min_epss_label: str, max_epss: Optional[str], max_epss_label: str) -> None:
    """
    Validate the EPSS score parameters.

    :type min_epss: ``Optional[str]``
    :param min_epss: Minimum EPSS score.

    :type min_epss_label: ``str``
    :param min_epss_label: Label for minimum EPSS score.

    :type max_epss: ``Optional[str]``
    :param max_epss: Maximum EPSS score.

    :type max_epss_label: ``str``
    :param max_epss_label: Label for maximum EPSS score.

    :raises DemistoException: If min_epss is not an int or float.
    :raises DemistoException: If max_epss is not an int or float.
    :raises DemistoException: If min_epss is greater than max_epss.
    """
    min_epss_float = None
    max_epss_float = None
    if min_epss:
        try:
            min_epss_float = float(min_epss)
        except ValueError:
            raise DemistoException(MESSAGES["INVALID_EPSS_SCORE"].format(min_epss_label))
    if max_epss:
        try:
            max_epss_float = float(max_epss)
        except ValueError:
            raise DemistoException(MESSAGES["INVALID_EPSS_SCORE"].format(max_epss_label))

    if min_epss_float is not None and (min_epss_float < 0 or min_epss_float > 1):
        raise DemistoException(MESSAGES["INVALID_EPSS_SCORE"].format(min_epss_label))
    if max_epss_float is not None and (max_epss_float < 0 or max_epss_float > 1):
        raise DemistoException(MESSAGES["INVALID_EPSS_SCORE"].format(max_epss_label))

    if min_epss_float is not None and max_epss_float is not None and min_epss_float > max_epss_float:
        raise DemistoException(MESSAGES["INVALID_SCORE_RANGE"].format(min_epss_label, max_epss_label))


def validate_cvss_score(min_cvss: Optional[str], min_cvss_label: str, max_cvss: Optional[str], max_cvss_label: str) -> None:
    """
    Validate the CVSS score parameters.

    :type min_cvss: ``Optional[str]``
    :param min_cvss: Minimum CVSS score.

    :type min_cvss_label: ``str``
    :param min_cvss_label: Label for minimum CVSS score.

    :type max_cvss: ``Optional[str]``
    :param max_cvss: Maximum CVSS score.

    :type max_cvss_label: ``str``
    :param max_cvss_label: Label for maximum CVSS score.

    :raises DemistoException: If min_cvss is not an int or float.
    :raises DemistoException: If max_cvss is not an int or float.
    :raises DemistoException: If min_cvss is greater than max_cvss.
    """
    min_cvss_float = None
    max_cvss_float = None
    if min_cvss:
        try:
            min_cvss_float = float(min_cvss)
        except ValueError:
            raise DemistoException(MESSAGES["INVALID_CVSS_SCORE"].format(min_cvss_label))
    if max_cvss:
        try:
            max_cvss_float = float(max_cvss)
        except ValueError:
            raise DemistoException(MESSAGES["INVALID_CVSS_SCORE"].format(max_cvss_label))

    if min_cvss_float is not None and (min_cvss_float < 0 or min_cvss_float > 10):
        raise DemistoException(MESSAGES["INVALID_CVSS_SCORE"].format(min_cvss_label))
    if max_cvss_float is not None and (max_cvss_float < 0 or max_cvss_float > 10):
        raise DemistoException(MESSAGES["INVALID_CVSS_SCORE"].format(max_cvss_label))

    if min_cvss_float is not None and max_cvss_float is not None and min_cvss_float > max_cvss_float:
        raise DemistoException(MESSAGES["INVALID_SCORE_RANGE"].format(min_cvss_label, max_cvss_label))


def validate_fetch_indicators_params(
    params: dict, last_run: dict[str, Any] = {}, is_test: bool = False, is_command: bool = False
) -> dict:
    """
    Validate the parameter list for fetch indicators.

    :type params: ``dict``
    :param params: Dictionary of parameters.

    :type last_run: ``dict[str, Any]``
    :param last_run: last run object obtained from demisto.getLastRun().

    :type is_test: ``bool``
    :param is_test: If test_module called fetch_indicators_command.

    :type is_command: ``bool``
    :param is_command: If get_indicators_command was called.

    :return: Updated dictionary of parameters.
    :rtype: ``dict``
    """
    tags = argToList(params.get("tags"), transform=lambda s: s.strip())
    products = argToList(params.get("products"), transform=lambda s: s.strip())
    vendors = argToList(params.get("vendors"), transform=lambda s: s.strip())
    cwe_ids = argToList(params.get("cwe_ids"), transform=lambda s: s.strip())
    ref_types = argToList(params.get("ref_types"), transform=lambda s: s.strip().lower())
    ref_values = argToList(params.get("ref_values"), transform=lambda s: s.strip())
    locations = argToList(params.get("locations"), transform=lambda s: s.strip().lower())
    severities = argToList(params.get("severities"), transform=lambda s: s.strip().lower())
    ransomware_scores = argToList(params.get("ransomware_scores"), transform=lambda s: s.strip().lower())
    attack_types = argToList(params.get("attack_types"), transform=lambda s: s.strip().lower())
    min_epss_score = params.get("min_epss_score")
    max_epss_score = params.get("max_epss_score")
    min_cvssv2_score = params.get("min_cvssv2_score")
    max_cvssv2_score = params.get("max_cvssv2_score")
    min_cvssv3_score = params.get("min_cvssv3_score")
    max_cvssv3_score = params.get("max_cvssv3_score")
    min_cvssv4_score = params.get("min_cvssv4_score")
    max_cvssv4_score = params.get("max_cvssv4_score")

    validate_cvss_score(min_cvssv2_score, "Minimum CVSS v2 Score", max_cvssv2_score, "Maximum CVSS v2 Score")
    validate_cvss_score(min_cvssv3_score, "Minimum CVSS v3 Score", max_cvssv3_score, "Maximum CVSS v3 Score")
    validate_cvss_score(min_cvssv4_score, "Minimum CVSS v4 Score", max_cvssv4_score, "Maximum CVSS v4 Score")
    validate_epss_score(min_epss_score, "Minimum EPSS v3 Score", max_epss_score, "Maximum EPSS v3 Score")

    valid_severity = [severity for severity in severities if severity in VALID_SEVERITIES]
    invalid_severity = [severity for severity in severities if severity not in VALID_SEVERITIES]

    valid_ref_types = [REFERENCE_TYPES_MAPPING[ref_type] for ref_type in ref_types if ref_type in REFERENCE_TYPES_MAPPING]
    invalid_ref_types = [ref_type for ref_type in ref_types if ref_type not in REFERENCE_TYPES_MAPPING]

    valid_cwe_ids = [cwe_id for cwe_id in cwe_ids if cwe_id.isdigit()]
    invalid_cwe_ids = [cwe_id for cwe_id in cwe_ids if not cwe_id.isdigit()]

    valid_location = [LOCATION_MAPPING[loc] for loc in locations if loc in LOCATION_MAPPING]
    invalid_location = [loc for loc in locations if loc not in LOCATION_MAPPING]

    valid_ransomware_scores = [score for score in ransomware_scores if score in VALID_RANSOMWARE_SCORE]
    invalid_ransomware_scores = [score for score in ransomware_scores if score not in VALID_RANSOMWARE_SCORE]

    valid_attack_types = [ATTACK_TYPE_MAPPING[attack_type] for attack_type in attack_types if attack_type in ATTACK_TYPE_MAPPING]
    invalid_attack_types = [attack_type for attack_type in attack_types if attack_type not in ATTACK_TYPE_MAPPING]

    if is_test or is_command:
        errors = []
        if invalid_severity:
            errors.append(
                MESSAGES["INVALID_MULTI_PARAMS_PROVIDED"].format(
                    invalid_severity, "Severity", [severity.title() for severity in VALID_SEVERITIES]
                )
            )
        if invalid_ref_types:
            errors.append(
                MESSAGES["INVALID_MULTI_PARAMS_PROVIDED"].format(
                    invalid_ref_types, "Reference Types", [ref_type.title() for ref_type in REFERENCE_TYPES_MAPPING]
                )
            )
        if invalid_cwe_ids:
            errors.append(MESSAGES["INVALID_INT_PARAMS_PROVIDED"].format(invalid_cwe_ids, "CWE IDs"))
        if invalid_location:
            errors.append(
                MESSAGES["INVALID_MULTI_PARAMS_PROVIDED"].format(
                    invalid_location, "Locations", [loc.title() for loc in LOCATION_MAPPING]
                )
            )
        if invalid_ransomware_scores:
            errors.append(
                MESSAGES["INVALID_MULTI_PARAMS_PROVIDED"].format(
                    invalid_ransomware_scores, "Ransomware Scores", [score.title() for score in VALID_RANSOMWARE_SCORE]
                )
            )
        if invalid_attack_types:
            errors.append(
                MESSAGES["INVALID_MULTI_PARAMS_PROVIDED"].format(
                    invalid_attack_types, "Attack Types", [attack_type.title() for attack_type in ATTACK_TYPE_MAPPING]
                )
            )
        if errors:
            raise DemistoException("\n\n".join(errors))

    fetch_params = assign_params(
        tags=",".join(tags),
        min_epss_score=min_epss_score,
        max_epss_score=max_epss_score,
        min_cvssv2_score=min_cvssv2_score,
        max_cvssv2_score=max_cvssv2_score,
        min_cvssv3_score=min_cvssv3_score,
        max_cvssv3_score=max_cvssv3_score,
        min_cvssv4_score=min_cvssv4_score,
        max_cvssv4_score=max_cvssv4_score,
        products=",".join(products),
        vendors=",".join(vendors),
        cwe_ids=",".join(valid_cwe_ids),
        ref_types=",".join(valid_ref_types),
        ref_values=",".join(ref_values),
        location=",".join(valid_location),
        severity=",".join(valid_severity),
        ransomware_score=",".join(valid_ransomware_scores),
        attack_type=",".join(valid_attack_types),
    )

    if is_command:
        last_touched_after = arg_to_datetime(params.get("last_touched_after", DEFAULT_FIRST_FETCH))

        if last_touched_after:
            last_touched_after = last_touched_after.strftime(DATE_FORMAT)  # type: ignore
            fetch_params["last_touched_after"] = last_touched_after

        last_touched_before = arg_to_datetime(params.get("last_touched_before", CURRENT_TIME))
        if last_touched_before:
            last_touched_before = last_touched_before.strftime(DATE_FORMAT)  # type: ignore
            fetch_params["last_touched_before"] = last_touched_before

        if last_touched_after and last_touched_before and last_touched_after >= last_touched_before:
            raise DemistoException(
                MESSAGES["INVALID_TIME_INTERVAL"].format(
                    "last_touched_after", "last_touched_before", last_touched_after, last_touched_before
                )
            )

        limit = arg_to_number(params.get("limit", DEFAULT_LIMIT))

        if limit is not None and (limit < 0 or limit > DEFAULT_LIMIT):
            raise DemistoException(MESSAGES["INVALID_LIMIT_PROVIDED"].format(limit, DEFAULT_LIMIT))
        if limit is not None:
            fetch_params["size"] = limit

        from_ = arg_to_number(params.get("from", DEFAULT_FROM))

        if from_ is not None and from_ < 0:
            raise DemistoException(MESSAGES["INVALID_FROM_PROVIDED"].format(from_))
        if from_ is not None:
            fetch_params["from"] = from_
    else:
        first_fetch = arg_to_datetime(params.get("first_fetch", DEFAULT_FIRST_FETCH))
        first_fetch_str = first_fetch.strftime(DATE_FORMAT) if first_fetch else None  # type: ignore

        # If available then take last_touched_after from last_run.
        last_touched_after_ = last_run.get("next_last_touched_after", first_fetch_str)
        fetch_params["last_touched_after"] = last_touched_after_

        current_time = arg_to_datetime(CURRENT_TIME).strftime(DATE_FORMAT)  # type: ignore
        # If available then take last_touched_before from last_run.
        last_touched_before_ = last_run.get("next_last_touched_before", current_time)
        fetch_params["last_touched_before"] = last_touched_before_

        _from = arg_to_number(last_run.get("from", DEFAULT_FROM))
        if _from is not None:
            fetch_params["from"] = _from

        fetch_params["size"] = PAGE_SIZE

    remove_nulls_from_dictionary(fetch_params)

    return fetch_params


def create_indicator_from_vulnerability(vulnerability: dict, params: dict, platform_url: str) -> dict:
    """
    Create an indicator object from a vulnerability response.

    :type vulnerability: ``dict``
    :param vulnerability: Vulnerability data from API response.

    :type params: ``dict``
    :param params: Dictionary of parameters.

    :type platform_url: ``str``
    :param platform_url: Platform URL.

    :return: Indicator object.
    :rtype: ``dict``
    """
    feed_tags = argToList(params.get("feedTags"))
    tlp_color = params.get("tlp_color")

    cve_ids = vulnerability.get("cve_ids", [])
    indicator_value = cve_ids[0] if cve_ids else f"FP-VULN-{vulnerability.get('id', 'UNKNOWN')}"
    indicator_type = FeedIndicatorType.CVE if cve_ids else DEFAULT_INDICATOR_TYPE

    cvss_score = ""
    cvss_version = ""
    cvss_vector = ""
    cvss_table_rows = []
    cvss_v2s = vulnerability.get("cvss_v2s", [])
    cvss_v3s = vulnerability.get("cvss_v3s", [])
    cvss_v4s = vulnerability.get("cvss_v4s", [])
    if cvss_v4s:
        item = [item for item in cvss_v4s if item.get("source") == "Flashpoint"]
        cvss_v4 = item[0] if item else cvss_v4s[0]
        cvss_score = cvss_v4.get("score", "")
        cvss_version = cvss_v4.get("version", "")
        cvss_vector = cvss_v4.get("vector_string", "")
        cvss_table_rows = create_cvss_table(cvss_v4)
    elif cvss_v3s:
        item = [item for item in cvss_v3s if item.get("source") == "Flashpoint"]
        cvss_v3 = item[0] if item else cvss_v3s[0]
        cvss_score = cvss_v3.get("score", "")
        cvss_version = cvss_v3.get("version", "")
        cvss_vector = cvss_v3.get("vector_string", "")
        cvss_table_rows = create_cvss_table(cvss_v3)
    elif cvss_v2s:
        item = [item for item in cvss_v2s if item.get("source") == "Flashpoint"]
        cvss_v2 = item[0] if item else cvss_v2s[0]
        cvss_score = cvss_v2.get("score", "")
        cvss_version = cvss_v2.get("version", "")
        cvss_vector = cvss_v2.get("vector_string", "")
        cvss_table_rows = create_cvss_table(cvss_v2)

    cpe: list = []
    products = vulnerability.get("products", [])
    for product in products:
        versions = product.get("versions", [])
        for _version in versions:
            cpe.extend({"CPE": cpe_item.get("name", "")} for cpe_item in _version.get("cpes", []))

    indicator_obj: dict[str, Any] = {
        "value": indicator_value,
        "type": indicator_type,
        "rawJSON": vulnerability,
        "fields": {
            "indicatoridentification": indicator_value,
            "cvss": cvss_score,
            "cvssscore": cvss_score,
            "cvssversion": cvss_version,
            "cvssvector": cvss_vector,
            "cvsstable": cvss_table_rows,
            "name": vulnerability.get("title", ""),
            "tags": list(vulnerability.get("tags", []) + feed_tags),
            "description": vulnerability.get("description", ""),
            "cvedescription": vulnerability.get("description", ""),
            "published": vulnerability.get("timelines", {}).get("published_at", ""),
            "modified": vulnerability.get("timelines", {}).get("last_modified_at", ""),
            "cvemodified": vulnerability.get("timelines", {}).get("last_modified_at", ""),
            "vulnerableproducts": cpe,
            "trafficlightprotocol": tlp_color,
        },
    }

    map_indicator_fields(vulnerability, indicator_obj)
    indicator_obj["fields"]["flashpointvulnerabilityfeedcvssvector"] = cvss_vector
    indicator_obj["fields"]["flashpointvulnerabilityfeedcreditees"] = [
        c.get("name", "") for c in vulnerability.get("creditees", [])
    ]
    indicator_obj["fields"]["flashpointvulnerabilityfeedplatformurl"] = urljoin(
        platform_url, PLATFORM_SUFFIX.format(vulnerability.get("id", ""))
    )

    if argToBoolean(params.get("createRelationship", False)) and vulnerability.get("cve_ids", []):
        custom_relationship = create_relationship(vulnerability.get("cve_ids", [])[0], [f"FP-VULN-{vulnerability.get('id', '')}"])
        relationships = []

        if len(vulnerability.get("cve_ids", [])) > 1:
            relationships = create_relationship(vulnerability.get("cve_ids", [])[0], vulnerability.get("cve_ids", [])[1:])

        indicator_obj["relationships"] = relationships + custom_relationship

    remove_nulls_from_dictionary(indicator_obj["fields"])

    return indicator_obj


def create_cvss_table(cvss_data: dict) -> list:
    """
    Creates a CVSS table for the indicator.

    :type cvss_data: ``dict``
    :param cvss_data: CVSS data.

    :return: CVSS table.
    :rtype: ``list``
    """
    cvss_table_rows = []
    for item in cvss_data:
        cvss_table_rows.append({"metrics": item, "value": cvss_data[item]})

    return cvss_table_rows


def create_relationship(entity_a: str, entity_b_data: list) -> list:
    """
    Create a list of relationships objects from the tags.

    :param entity_a: the entity a of the relation which is the current indicator.
    :param entity_b_data: list of entity_b_data returned from the API.

    :return: list of EntityRelationship objects containing all the relationships.
    """
    relationships = []
    for entity_b in entity_b_data:
        if "FP-VULN-" in entity_b:
            entity_b_type = DEFAULT_INDICATOR_TYPE
        else:
            entity_b_type = FeedIndicatorType.CVE

        if entity_b:
            obj = EntityRelationship(
                name=EntityRelationship.Relationships.RELATED_TO,
                entity_a=entity_a,
                entity_a_type=FeedIndicatorType.CVE,
                entity_b=entity_b,
                entity_b_type=entity_b_type,
            )
            obj = obj.to_indicator()
            relationships.append(obj)

    return relationships


def prepare_hr_for_indicators(indicators: list, platform_url: str) -> str:
    """
    Prepare human-readable response.

    :param indicators: List of indicators.
    :param platform_url: Platform URL.

    :return: Indicators in human-readable format.
    """
    hr = []

    for indicator in indicators:
        raw_json = indicator.get("rawJSON", {})
        cve_ids = raw_json.get("cve_ids", [])
        indicator_value = cve_ids[0] if cve_ids else f"FP-VULN-{raw_json.get('id', 'UNKNOWN')}"

        data = {
            "ID": f"[{raw_json.get('id', '')}]({urljoin(platform_url, PLATFORM_SUFFIX.format(raw_json.get('id', '')))})",
            "Indicator Value": indicator_value,
            "Title": raw_json.get("title", ""),
            "Vulnerability Status": raw_json.get("vuln_status", ""),
            "EPSS Score": raw_json.get("scores", {}).get("epss_score", ""),
            "Ransomware Score": raw_json.get("scores", {}).get("ransomware_score", ""),
            "Social Risk Scores": raw_json.get("scores", {}).get("social_risk_scores", ""),
            "Published": raw_json.get("timelines", {}).get("published_at", ""),
            "Modified": raw_json.get("timelines", {}).get("last_modified_at", ""),
            "Description": raw_json.get("description", ""),
            "Solution": raw_json.get("solution", ""),
            "Technical Description": raw_json.get("technical_description", ""),
            "Tags": raw_json.get("tags", ""),
            "CVSS v2": raw_json.get("cvss_v2s", ""),
            "CVSS v3": raw_json.get("cvss_v3s", ""),
            "CVSS v4": raw_json.get("cvss_v4s", ""),
            "Products": raw_json.get("products", ""),
            "CWEs": raw_json.get("cwes", ""),
            "Exploits": raw_json.get("exploits", ""),
            "Exploits Count": raw_json.get("exploits_count", ""),
            "External References": raw_json.get("ext_references", ""),
        }
        hr.append(data)

    headers = [
        "ID",
        "Indicator Value",
        "Title",
        "Vulnerability Status",
        "EPSS Score",
        "Ransomware Score",
        "Social Risk Scores",
        "Published",
        "Modified",
        "Description",
        "Solution",
        "Technical Description",
        "Tags",
        "CVSS v2",
        "CVSS v3",
        "CVSS v4",
        "Products",
        "CWEs",
        "Exploits",
        "Exploits Count",
        "External References",
    ]

    return tableToMarkdown(
        name="Vulnerabilities",
        t=hr,
        headers=headers,
        removeNull=True,
        json_transform_mapping={
            header: JsonTransformer(is_nested=True)
            for header in [
                "Social Risk Scores",
                "CVSS v2",
                "CVSS v3",
                "CVSS v4",
                "Products",
                "CWEs",
                "Exploits",
                "External References",
            ]
        },
    )


""" COMMAND FUNCTIONS """


def test_module(client: Client) -> str:
    """
    Tests connectivity to the Flashpoint Vulnerability Intelligence API.

    :type client: ``Client``
    :param client: Client object for API interaction.

    :return: 'ok' if test passed, anything else will fail the test.
    :rtype: ``str``
    """
    params = demisto.params()
    is_fetch = params.get("feed", False)
    if is_fetch:
        fetch_indicators_command(client=client, params=params, last_run={}, is_test=True)
    else:
        client.list_vulnerabilities(params={"size": 1})
    return "ok"


def fetch_indicators_command(client: Client, params: dict, last_run: dict[str, Any], is_test: bool = False) -> tuple[list, dict]:
    """
    Fetch vulnerability indicators from Flashpoint Ignite.

    :param client: Client object.
    :param params: Dictionary of parameters.
    :param last_run: last run object obtained from demisto.getLastRun().
    :param is_test: If test_module called fetch_indicators_command.

    :return: Tuple of List of indicators and Dict of last run object.
    """
    current_time = datetime.now(timezone.utc)
    next_run: dict = {}

    fetch_params = validate_fetch_indicators_params(params=params, last_run=last_run, is_test=is_test)

    response = client.list_vulnerabilities(params=fetch_params)

    vulnerabilities = response.get("results", [])
    vulnerabilities = remove_empty_elements(vulnerabilities)
    vulnerabilities_ids = [vulnerability.get("id") for vulnerability in vulnerabilities]

    if is_test:
        return [], {}

    indicators = []
    for _id in vulnerabilities_ids:
        # Check if now - current_time > 4 minutes
        now = datetime.now(timezone.utc)
        if (now - current_time).total_seconds() > 240:
            demisto.debug(
                "Fetch Vulnerability: Time interval adjustment exceeded 4 minutes. "
                "Exiting loop and will continue from this point in the next fetch cycle. "
            )
            break

        vulnerability_response = client.get_vulnerability(_id)
        remove_nulls_from_nested_dictionary(vulnerability_response)
        indicator = create_indicator_from_vulnerability(
            vulnerability=vulnerability_response,
            params=params,
            platform_url=client.platform_url,
        )
        indicators.append(indicator)

    # Creating new last_run according to response.
    if len(vulnerabilities_ids) != len(indicators) or len(vulnerabilities_ids) == len(indicators) == PAGE_SIZE:
        # More records exist, keep the same last_touched_after and use pagination
        next_run["next_last_touched_after"] = last_run.get("next_last_touched_after", fetch_params["last_touched_after"])
        next_run["next_last_touched_before"] = last_run.get("next_last_touched_before", fetch_params["last_touched_before"])
        # Set only from equal to previous from + max_fetch.
        next_run["from"] = last_run.get("from", DEFAULT_FROM) + len(indicators)
    else:
        # All records fetched for this time range, update to current time
        next_run["next_last_touched_after"] = last_run.get("next_last_touched_before", fetch_params["last_touched_before"])

    demisto.debug(f"[Flashpoint Vulnerability Feed] Set the last Run for indicators: {next_run}")

    return indicators, next_run


def flashpoint_vulnerability_get_indicators_command(client: Client, params: dict, args: dict) -> CommandResults:
    """
    Get limited number of indicators from Flashpoint Vulnerability API.

    :param client: Client object.
    :param params: Dictionary of parameters.
    :param args: Dictionary of arguments.

    :return: Standard Command Result.
    """
    current_time = datetime.now(timezone.utc)
    platform_url = client.platform_url

    fetch_params = validate_fetch_indicators_params(params=args, is_command=True)

    response = client.list_vulnerabilities(params=fetch_params)

    vulnerabilities = response.get("results", [])
    vulnerabilities = remove_empty_elements(vulnerabilities)
    vulnerabilities_ids = [vulnerability.get("id") for vulnerability in vulnerabilities]

    indicators: list = []
    for _id in vulnerabilities_ids:
        # Check if now - current_time > 4 minutes
        now = datetime.now(timezone.utc)
        if (now - current_time).total_seconds() > 240:
            return_warning(
                "get-indicators: Time interval exceeded 4 minutes. "
                "Exiting command execution and you can continue from this point in the next command call. "
                f"Provide from = {len(indicators) + fetch_params.get('from', DEFAULT_FROM)} to continue from this point."
            )
            break

        vulnerability_response = client.get_vulnerability(_id)
        remove_nulls_from_nested_dictionary(vulnerability_response)
        indicator = create_indicator_from_vulnerability(
            vulnerability=vulnerability_response,
            params=params,
            platform_url=platform_url,
        )
        indicators.append(indicator)

    if not indicators:
        return CommandResults(readable_output=MESSAGES["NO_INDICATORS_FOUND"])

    readable_output = prepare_hr_for_indicators(indicators=indicators, platform_url=platform_url)
    return CommandResults(
        readable_output=readable_output,
        outputs=indicators,
        outputs_key_field="value",
        outputs_prefix="FlashpointVulnerabilityFeed.Indicator",
        raw_response=indicators,
    )


""" MAIN FUNCTION """


def main() -> None:
    """Parse params and runs command functions."""
    params = remove_space_from_args(demisto.params())
    remove_nulls_from_dictionary(params)

    args = remove_space_from_args(demisto.args())
    remove_nulls_from_dictionary(args)

    base_url = params.get("url", DEFAULT_API_URL)
    api_key = str(params.get("credentials", {}).get("password", "")).strip()

    verify_certificate = not argToBoolean(params.get("insecure", False))
    proxy = argToBoolean(params.get("proxy", False))

    command = demisto.command()
    demisto.debug(f"[Flashpoint Vulnerability Feed] Command being called is {command}")

    try:
        validate_params(params=params)

        client = Client(
            base_url=base_url,
            api_key=api_key,
            verify=verify_certificate,
            proxy=proxy,
        )

        if command == "test-module":
            return_results(test_module(client=client))
        elif command == "fetch-indicators":
            last_run = demisto.getLastRun()
            indicators, next_run = fetch_indicators_command(client=client, params=params, last_run=last_run)
            demisto.setLastRun(next_run)
            demisto.createIndicators(indicators)
        elif command == "flashpoint-vulnerability-get-indicators":
            return_results(flashpoint_vulnerability_get_indicators_command(client=client, params=params, args=args))
        else:
            raise NotImplementedError(f"Command {command} is not implemented.")

    except Exception as e:
        return_error(f"Failed to execute {command} command.\nError:\n{str(e)}")


if __name__ in ["__main__", "builtin", "builtins"]:  # pragma: no cover
    main()