Exodus Intelligence Vulnerabilities

import demistomock as demisto
import urllib3
from CommonServerPython import *
from eve_client import eve

from CommonServerUserPython import *

# Disable insecure warnings
urllib3.disable_warnings()

""" CONSTANTS """

DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
MAX_INCIDENTS_TO_FETCH = 50

WANTED_FIELDS = [
    "identifier",
    "xi_scores",
    "cves",
    "product",
    "vendor",
    "mitigation",
    "attack_vector",
    "cvss",
    "description",
    "publish_date",
    "modified_timestamp",
]

LIST_FIELDS = [
    "transport",
    "links",
    "application",
    "zdis",
    "cpe23Uri",
    "methods",
    "modules",
]


def get_best_xi_score(scores):
    """Get most recent xi_score.

    Args:
        xi_scores (dict): Dictionary with xi_scores.

    Returns:
        float: Most recent xi_score.
    """
    if scores == []:
        return 0.0
    commented_scores = [s for s in scores if s["comment"] != ""]
    if len(commented_scores) > 0:
        scores = commented_scores
    return float(sorted(scores, key=lambda x: x["created_at"], reverse=True)[0]["score"])


def extract_data(data, dest):
    """Extract data from nested dictionary.

    Args:
        data (dict): Nested Dictionary to extract data.
        dest (dict): Dictionary to store the extracted data.

    Returns:
        dict: Dictionary with extracted data.
    """
    for key, value in data.items():
        demisto_field = f"xi{key.replace('_', '').lower()}"
        t_value = type(value)
        if key == "xi_scores":
            dest["xiscores"] = get_best_xi_score(value)
        elif t_value is dict:
            extract_data(value, dest)
        elif t_value is list:
            for i in value:
                if isinstance(i, dict):
                    extract_data(i, dest)
                if isinstance(i, str | float) and i in WANTED_FIELDS:
                    dest[demisto_field] = i
            if key in LIST_FIELDS:
                for i in value:
                    try:
                        dest[demisto_field].append(i)
                    except KeyError:
                        dest[demisto_field] = []
                        dest[demisto_field].append(i)
        else:
            if key in LIST_FIELDS:
                try:
                    dest[demisto_field].append(value)
                except KeyError:
                    dest[demisto_field] = []
                    dest[demisto_field].append(value)
            elif key in WANTED_FIELDS:
                dest[demisto_field] = value
    return dest


def connect(EMAIL, PASSWORD, CLIENT_ENCRYPTION_KEY, EVE_URL):
    """Establish a connection to Exodus Intelligence API.

    Returns:
        exodus_Obj: API Client object.
    """
    try:
        exodus = eve.EVEClient(EMAIL, PASSWORD, CLIENT_ENCRYPTION_KEY, url=EVE_URL)
    except Exception as e:
        demisto.debug(f"There was a problem connecting to the server: {e}")
        raise ConnectionError
    return exodus


def fetch_indicators(EMAIL, PASSWORD, CLIENT_ENCRYPTION_KEY, EVE_URL, MAX_XI, MIN_XI):
    """Retrieve vulnerability data from Exodus Intelligence."""
    indicators = []
    formatted_list = []
    min_xi = 0 if MIN_XI == "" else MIN_XI
    max_xi = 10 if MAX_XI == "" else MAX_XI
    try:
        exodus = connect(EMAIL, PASSWORD, CLIENT_ENCRYPTION_KEY, EVE_URL)
        demisto.debug("Connected to server")
        recent_vulns = exodus.get_recent_vulns()
        try:
            data = recent_vulns["data"]["items"]
        except KeyError as e:
            demisto.debug(f"There was an error getting the data {e}")
        demisto.debug(f"Fetched {len(data)} total vulnerabilities")
        for item in data:
            try:
                cve = item["cves"][0]
                report_data = {
                    "xicve": cve,
                    "xiidentifier": item["identifier"],
                }
                report = exodus.get_report(cve)
                vulnerability = exodus.get_vuln(cve)
                if report.get("ok"):
                    report_data = extract_data(report, report_data)
                vuln_data = extract_data(vulnerability, report_data)
                score = vuln_data.get("xiscores")
                if score >= min_xi and score <= max_xi:
                    formatted_list.append(vuln_data)
            except KeyError as e:
                demisto.debug(f"There was a problem: {e}")
    except Exception as e:
        demisto.debug(f"Something went wrong: {e}")

    if formatted_list:
        for items in formatted_list:
            try:
                indicator = {
                    "value": items["xiidentifier"],
                    "type": "Exodus Intelligence",
                    "fields": items,
                    "rawJSON": items,
                }
                indicators.append(indicator)
            except KeyError as e:
                demisto.debug(f"There was a problem creating indicators: {e}")
        demisto.createIndicators(indicators)
    return f"Grabbed {len(formatted_list)} indicators with your search criteria."


def reset_data_stream(EMAIL, PASSWORD, CLIENT_ENCRYPTION_KEY, EVE_URL, days):
    try:
        exodus = connect(EMAIL, PASSWORD, CLIENT_ENCRYPTION_KEY, EVE_URL)
        data = exodus.get_recent_vulns(days)
        day = data["data"]["end_ts"]
        results = CommandResults(
            readable_output=tableToMarkdown("Reset", day, headers=["date"]),
            outputs_prefix="ExodusVulnerabilityEnrichment.ResetDataStream",
            outputs_key_field="end_ts",
            outputs=day,
        )
        return results
    except (ConnectionError, KeyError) as e:
        demisto.debug(f"There was a problem connecting to the API {e}")
        raise ConnectionError


"""EXECUTION"""


def main():
    EMAIL = demisto.params().get("email")
    PASSWORD = demisto.params().get("password")
    CLIENT_ENCRYPTION_KEY = demisto.params().get("client_encryption_key")
    EVE_URL = demisto.params().get("url")
    MAX_XI = demisto.params().get("max_xi_score")
    MAX_XI = int(MAX_XI) if MAX_XI else 10
    MIN_XI = demisto.params().get("min_xi_score")
    MIN_XI = int(MIN_XI) if MIN_XI else 0
    try:
        if demisto.command() == "fetch-indicators" or demisto.command() == "exodus-get-indicators":
            return_results(
                fetch_indicators(
                    EMAIL,
                    PASSWORD,
                    CLIENT_ENCRYPTION_KEY,
                    EVE_URL,
                    MAX_XI,
                    MIN_XI,
                )
            )
        elif demisto.command() == "exodus-reset-data-stream":
            try:
                days = arg_to_number(demisto.args().get("reset"))
            except ValueError as e:
                demisto.info(f"Please provide a valid number. Resetting data stream to 7 days. Error: {e}")
                days = 7
            try:
                return_results(reset_data_stream(EMAIL, PASSWORD, CLIENT_ENCRYPTION_KEY, EVE_URL, days))
            except ConnectionError:
                message = "Unable to reset data stream"
                return_results(
                    CommandResults(
                        readable_output=tableToMarkdown("Reset", message, headers=["date"]),
                        outputs_prefix="ExodusVulnerabilityEnrichment.ResetDataStream",
                        outputs=message,
                    )
                )

        elif demisto.command() == "test-module":
            try:
                connect(EMAIL, PASSWORD, CLIENT_ENCRYPTION_KEY, EVE_URL)
                return_results("ok")
            except ConnectionError:
                demisto.info("Exodus authentication failed. Verify your credentials.")
    except Exception as e:
        demisto.error(traceback.format_exc())
        return_error(f"Failed to execute {demisto.command()} command.\nError:\n{e!s}")


"""ENTRY POINT"""

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