IgniteFeed

"""Flashpoint Ignite Feed Integration for Cortex XSOAR (aka Demisto)"""

""" IMPORTS """

from typing import Any, Dict, Tuple  # noqa E402

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 """

DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ"  # ISO8601 format with UTC, default in XSOAR
LIMIT = 10
MAX_INDICATORS = 10000
MAX_FETCH = 1000
DEFAULT_FIRST_FETCH = "3 days"
CURRENT_TIME = "now"
DEFAULT_OFFSET = 0
DEFAULT_SORT_ORDER = "asc"
DEFAULT_INDICATOR_TYPE = "Flashpoint Indicator"
FILE_TYPES = ["sha1", "sha256", "sha512", "md5", "ssdeep"]
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.
URL_SUFFIX = {"ATTRIBUTES": "/technical-intelligence/v1/attribute"}
MESSAGES = {
    "NO_PARAM_PROVIDED": "Please provide the {}.",
    "LIMIT_ERROR": "{} is an invalid value for limit. Limit must be between 1 and {}.",
    "NO_INDICATORS_FOUND": "No indicators were found for the given argument(s).",
    "TIME_RANGE_ERROR": f"The maximum indicators to fetch for the given first fetch can not exceed {MAX_INDICATORS}."
    " Current indicators are {}. Try decreasing the time interval.",
}
HTTP_ERRORS = {
    400: "Bad request: An error occurred while fetching the data.",
    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()
INTEGRATION_PLATFORM = "Cortex XSOAR"
DEFAULT_API_PATH = "api.flashpoint.io"
DEFAULT_PLATFORM_PATH = "https://app.flashpoint.io"
IGNITE_FEED_EVENT_HREF = "https://app.flashpoint.io/cti/malware/iocs/"
FLASHPOINT_FEED_MAPPING = {
    "firstseenbysource": {"path": "first_observed_at.date-time", "type": "date"},
    "tags": {"path": "Event.Tag.name", "type": "tags"},
    "flashpointfeedattributeid": {"path": "fpid", "type": "str"},
    "flashpointfeedattributeuuid": {"path": "uuid", "type": "str"},
    "flashpointfeedcategory": {"path": "category", "type": "str"},
    "flashpointfeedeventcreatoremail": {"path": "Event.event_creator_email", "type": "str"},
    "flashpointfeedeventhref": {"path": "Event.uuid", "type": "url"},
    "flashpointfeedeventinformation": {"path": "Event.info", "type": "str"},
    "flashpointfeedeventuuid": {"path": "Event.uuid", "type": "str"},
    "flashpointfeedindicatortype": {"path": "type", "type": "str"},
    "flashpointfeedhtmlmalwaredescription": {"path": "malware_description", "type": "str"},
    "flashpointfeedreport": {"path": "Event.report", "type": "url"},
    "flashpointfeedtimestamp": {"path": "timestamp", "type": "date"},
}
HR_SUFFIX = {
    "IOC_UUID_LIST": "/cti/malware/iocs?query={}&sort_date=All%20Time",
}


class Client(BaseClient):
    """Client class to interact with the service API."""

    def __init__(self, url, headers, verify, proxy):
        """Initialize class object.

        :type url: ``str``
        :param url: Base server address with suffix, for example: https://example.com.

        :type headers: ``Dict``
        :param headers: Additional headers to be included in the requests.

        :type verify: ``bool``
        :param verify: Use to indicate secure/insecure http request.

        :type proxy: ``bool``
        :param proxy: The proxy settings to be used.

        """
        self.url = url

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

        self.headers = headers
        self.verify = verify
        self.proxy = proxy

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

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

        :param url_suffix: url encoded url suffix.
        :param params: URL parameters to specify the query.
        :param method: Specify http methods.
        :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 check_indicator_type(self, indicator_value: str, default_map: bool) -> str:
        """
        Set the type of the indicator.

        :param indicator_value: Value of the indicator.
        :param default_map: To enable the default mapper setting for the indicator.

        :return: Type of the indicator.
        """
        ind_type = DEFAULT_INDICATOR_TYPE
        if not default_map:
            ind_type = auto_detect_indicator_type(indicator_value)

            if not ind_type:
                ind_type = DEFAULT_INDICATOR_TYPE

        return ind_type

    def create_relationship(self, entity_a: str, entity_a_type: str, tags: 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_a_type: the entity a type which is the type of the current indicator.
        :param tags: list of tags returned from the API.

        :return: list of EntityRelationship objects containing all the relationships.
        """
        my_tags = [tag for tag in tags if tag.get("is_galaxy")]
        relationships = []
        for tag in my_tags:
            name = tag.get("name")
            if name and "misp-galaxy:mitre-enterprise-attack" in name:
                # operations for entity_b
                names = name.split("=")
                entity_b_value = names[1].replace('"', "")
                entity_b_values = entity_b_value.split("-")
                entity_b = entity_b_values[0].strip()

                # operations for entity_b_type
                entity_b_types = names[0].split("misp-galaxy:mitre-enterprise-attack-")
                entity_b_type = " ".join(w[0].upper() + w[1:] for w in entity_b_types[1].split("-"))
                entity_b_type = entity_b_type.replace(" Of ", " of ")

                obj = EntityRelationship(
                    name=EntityRelationship.Relationships.INDICATOR_OF,
                    entity_a=entity_a,
                    entity_a_type=entity_a_type,
                    entity_b=entity_b,
                    entity_b_type=entity_b_type,
                )
                obj = obj.to_indicator()
                relationships.append(obj)

        return relationships

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

        :param resp: raw response of indicator.
        :param indicator_obj: created indicator object.

        :return: None.
        """
        for key, value in FLASHPOINT_FEED_MAPPING.items():
            if key == "tags":
                new_tags = indicator_obj.get("fields", {}).get("tags")
                event_tags = resp.get("Event", {}).get("Tag", {})
                true_value = new_tags + [tag.get("name") for tag in event_tags]
            elif key == "flashpointfeedeventhref":
                fpid = resp.get("fpid", "")
                true_value = urljoin(IGNITE_FEED_EVENT_HREF, fpid)
            elif key == "flashpointfeedtimestamp":
                timestamp = int(resp.get("timestamp")) * 1000  # type: ignore
                true_value = timestamp_to_datestring(timestamp, DATE_FORMAT, is_utc=True)
            else:
                path = value.get("path", "")
                paths = path.split(".")
                true_value = resp

                try:
                    for p in paths:
                        true_value = true_value.get(p)
                except AttributeError:
                    true_value = None

            indicator_obj["fields"][key] = true_value

    def create_indicators_from_response(self, response: Any, params: dict) -> List:
        """
        Create indicators from the response.

        :param response: response received from the API.
        :param params: dictionary of parameters.

        :return: List of indicators.
        """
        indicators = []
        feed_tags = argToList(params.get("feedTags"))
        tlp_color = params.get("tlp_color")
        relationship = params.get("createRelationship", False)
        default_map = params.get("defaultMap", False)

        for resp in response:
            source_type = resp.get("type")
            indicator_value = resp.get("value", {}).get(source_type)
            indicator_type = self.check_indicator_type(indicator_value=indicator_value, default_map=default_map)
            indicator_obj = {
                "value": indicator_value,
                "type": indicator_type,
                "rawJSON": resp,
                "fields": {
                    "tags": feed_tags,
                },
            }
            if tlp_color:
                indicator_obj["fields"]["trafficlightprotocol"] = tlp_color

            if relationship:
                event_tags = resp.get("Event", {}).get("Tag", "")
                relationships = self.create_relationship(indicator_value, indicator_type, event_tags)  # type: ignore
                indicator_obj["relationships"] = [] if not event_tags else relationships

            self.map_indicator_fields(resp, indicator_obj)

            indicators.append(indicator_obj)

        return indicators

    def fetch_indicators(self, params: dict, resp_type: str = "json") -> Any:
        """
        Fetch the list of indicators based on specified arguments.

        :param params: Parameters to be sent with API call.
        :param resp_type: Response type to be returned.

        :return: List of indicators.
        """
        response = self.http_request(url_suffix=URL_SUFFIX["ATTRIBUTES"], params=params, method="GET", resp_type=resp_type)

        return response

    @staticmethod
    def handle_errors(resp) -> None:
        """Handle http errors."""
        status = resp.status_code
        if status in HTTP_ERRORS:
            raise DemistoException(HTTP_ERRORS[status])
        else:
            resp.raise_for_status()


""" HELPER FUNCTIONS """


def remove_space_from_args(args):
    """
    Remove space from args.

    :param args: Arguments.

    :return: Argument's dictionary without spaces.
    """
    for key in args:
        if isinstance(args[key], str):
            args[key] = args[key].strip()
    return args


def check_value_of_total_records(total: Any, next_run: dict) -> None:
    """
    Check if total number of records are more than the limit or not.

    :param total: Total number of records
    :param next_run: Dictionary to set in last run

    :return: None
    """
    if total:
        if total > MAX_INDICATORS:  # type: ignore
            raise ValueError(MESSAGES["TIME_RANGE_ERROR"].format(total))
        next_run["total"] = total


def validate_params(params: dict):
    """
    Validate the parameters.

    :param params: Params to validate.
    """
    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"))


def validate_get_indicators_args(args: dict) -> dict:
    """
    Validate the argument list for get indicators.

    :param args: Dictionary of arguments.

    :return: Updated dictionary of arguments.
    """
    fetch_params = {}

    limit = arg_to_number(args.get("limit", LIMIT))
    if limit < 1 or limit > MAX_FETCH:  # type: ignore
        raise ValueError(MESSAGES["LIMIT_ERROR"].format(limit, MAX_FETCH))
    fetch_params["limit"] = limit

    fetch_params["types"] = args.get("types", "").lower()

    first_fetch = arg_to_datetime(args.get("updated_since", DEFAULT_FIRST_FETCH))

    fetch_params["updated_since"] = first_fetch.strftime(DATE_FORMAT)  # type: ignore
    fetch_params["sort_timestamp"] = DEFAULT_SORT_ORDER  # type: ignore

    remove_nulls_from_dictionary(fetch_params)

    return fetch_params


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

    :param indicators: List of indicators.
    :param platform_url: URL of the platform.

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

    for indicator in indicators:
        raw_json = indicator.get("rawJSON")
        indicator_type = raw_json.get("type")

        updated_indicator_type = indicator_type
        if updated_indicator_type in FILE_TYPES:
            updated_indicator_type = "File"

        event_tags = []
        tags = raw_json.get("Event", {}).get("Tag")
        if tags:
            for tag in tags:
                name = tag.get("name")
                event_tags.append(name)

        timestamp = datetime.utcfromtimestamp(int(raw_json.get("timestamp")))
        timestamp = timestamp.strftime(DATE_FORMAT)

        uuid = raw_json.get("Event", {}).get("uuid")
        fp_link = urljoin(platform_url, HR_SUFFIX["IOC_UUID_LIST"].format(uuid))
        fp_id = "[{}]({})".format(raw_json.get("fpid"), fp_link)

        report_data = ""
        reports = raw_json.get("reports")
        if reports:
            for report in reports:
                report_data += "\n\nClick to see\n[Intelligence Report]({})".format(report.get("html"))

        fp_id += report_data

        data = {
            "FPID": fp_id,
            "Indicator Type": updated_indicator_type,
            "Indicator Value": raw_json.get("value", {}).get(indicator_type),
            "Category": raw_json.get("category"),
            "Event Name": raw_json.get("Event", {}).get("info"),
            "Event Tags": event_tags,
            "Created Timestamp (UTC)": timestamp,
            "First Observed Date": raw_json.get("first_observed_at", {}).get("date-time"),
        }
        hr.append(data)

    headers = [
        "FPID",
        "Indicator Type",
        "Indicator Value",
        "Category",
        "Event Name",
        "Event Tags",
        "Created Timestamp (UTC)",
        "First Observed Date",
    ]

    return tableToMarkdown(name="Indicator(s)", t=hr, headers=headers, removeNull=True)


def validate_fetch_indicators_params(params: dict, last_run: dict[str, Any]) -> dict:
    """
    Validate the parameter list for fetch indicators.

    :param params: Dictionary of parameters.
    :param last_run: last run object obtained from demisto.getLastRun().

    :return: Updated dictionary of parameters.
    """
    fetch_params = {"limit": MAX_FETCH, "types": (",".join(params.get("types", ""))).lower()}

    first_fetch = arg_to_datetime(params.get("first_fetch", DEFAULT_FIRST_FETCH)).strftime(DATE_FORMAT)  # type: ignore
    # If available then take updated_since from last_run.
    updated_since = last_run.get("next_updated_since", first_fetch)

    current_time = arg_to_datetime(CURRENT_TIME).strftime(DATE_FORMAT)  # type: ignore
    # If available then take updated_until from last_run.
    updated_until = last_run.get("next_updated_until", current_time)

    offset = last_run.get("offset", DEFAULT_OFFSET)

    fetch_params["skip"] = offset
    fetch_params["updated_since"] = updated_since
    fetch_params["updated_until"] = updated_until
    fetch_params["sort_timestamp"] = DEFAULT_SORT_ORDER  # type: ignore

    remove_nulls_from_dictionary(fetch_params)

    return fetch_params


"""Command functions"""


def test_module(client: Client) -> str:
    """
    Tests the indicators from the feed.

    :param client: Client object.

    :return: 'ok' if test passed, anything else will fail the test.
    """
    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.fetch_indicators(params={"limit": 1})
    return "ok"


def fetch_indicators_command(client: Client, params: dict, last_run: dict[str, Any], is_test: bool = False) -> tuple[List, dict]:
    """
    Fetch the indicators.

    :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_incident.

    :return: List of indicators and Dict of last run object.
    """
    next_run: dict = {}

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

    resp = client.fetch_indicators(params=fetch_params, resp_type="response")

    response = remove_empty_elements(resp.json())

    total = int(resp.headers.get("x-fp-total-hits", len(response)))
    check_value_of_total_records(total, next_run)

    if is_test:
        return [], {}

    # Creating new last_run according to response.
    if len(response) < MAX_FETCH:
        # Updating updated_since equal to previous updated_until.
        next_run["next_updated_since"] = last_run.get("next_updated_until", fetch_params["updated_until"])
    else:
        next_run = last_run
        next_run["next_updated_since"] = fetch_params["updated_since"]
        next_run["next_updated_until"] = fetch_params["updated_until"]
        # Set only offset equal to previous offset + max_fetch.
        next_run["offset"] = next_run.get("offset", DEFAULT_OFFSET) + MAX_FETCH

    indicators = client.create_indicators_from_response(response=response, params=params)

    demisto.debug(f"Set the last Run for indicators: {next_run}")

    return indicators, next_run


def flashpoint_ignite_get_indicators_command(client: Client, params: dict, args: dict) -> CommandResults:
    """
    Get limited number of indicators.

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

    :return: Standard Command Result.
    """
    fetch_params = validate_get_indicators_args(args=args)

    response = client.fetch_indicators(params=fetch_params)

    response = remove_empty_elements(response)

    indicators = client.create_indicators_from_response(response=response, params=params)

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

    readable_output = prepare_hr_for_indicators(indicators=indicators, platform_url=client.platform_url)
    return CommandResults(readable_output=readable_output, raw_response=indicators)


"""Main Function"""


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

    # Get the service API url
    base_url = params.get("url", DEFAULT_API_PATH)

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

    # Default configuration parameters for handling proxy and SSL Certificate validation.
    insecure = not argToBoolean(params.get("insecure", False))
    proxy = argToBoolean(params.get("proxy", False))

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

    try:
        validate_params(params=params)
        headers: dict = {
            "Authorization": f"Bearer {api_key}",
            "X-FP-IntegrationPlatform": INTEGRATION_PLATFORM,
            "X-FP-IntegrationPlatformVersion": get_demisto_version_as_str(),
            "X-FP-IntegrationVersion": INTEGRATION_VERSION,
        }
        client = Client(verify=insecure, proxy=proxy, url=base_url, headers=headers)
        args = demisto.args()
        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-ignite-get-indicators":
            return_results(flashpoint_ignite_get_indicators_command(client=client, params=params, args=args))
        else:
            raise NotImplementedError(f"Command {command} is not implemented.")

    # Log exceptions and return errors
    except Exception as e:
        return_error(f"Failed to execute {command} command.\nError:\n{e!s}")


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