Intel471 Malware Indicator Feed

from typing import Any

import urllib3
from CommonServerPython import *

# disable insecure warnings
urllib3.disable_warnings()

INTEGRATION_NAME = "Intel 471 Malware Feed"
FEED_URL_INDICATORS_TITAN = "https://api.intel471.com/v1/indicators/stream"
FEED_URL_INDICATORS_VERITY471 = "https://api.intel471.cloud/integrations/indicators/v1/indicators/stream"
FEED_URL_GIRS_TITAN = "https://api.intel471.com/v1/girs"
FEED_URL_MALWARE_FILE_VERITY471 = "https://api.intel471.cloud/integrations/malware-intel/v1/malware/files"
DEMISTO_VERSION = demisto.demistoVersion()
CONTENT_PACK = f"Intel471 Feed/{get_pack_version()!s}"
INTEGRATION = "Intel471 Malware Indicator Feed"
USER_AGENT = f'XSOAR/{DEMISTO_VERSION["version"]}.{DEMISTO_VERSION["buildNumber"]} - {CONTENT_PACK} - {INTEGRATION}'
MAX_COUNT = 100
MAX_BATCH = 10000

INDICATOR_TYPES = {
    "domain": FeedIndicatorType.Domain,
    "email": FeedIndicatorType.Email,
    "file": FeedIndicatorType.File,
    "ipv4": FeedIndicatorType.IP,
    "url": FeedIndicatorType.URL,
}
THREAT_TYPE = "malware"


class Client:
    """
    Client to use in the Intel 471 Feed integration. Overrides BaseClient.
    """

    headers = {"user-agent": USER_AGENT}

    def __init__(
        self,
        auth: tuple[str, str],
        insecure: bool = False,
        tags: list | None = None,
        tlp_color: str | None = None,
        intel471_backend: str | None = None,
        indicator: str | None = "*",
        indicator_type: str | None = None,
        threat_type: str | None = None,
        malware_family: str | None = None,
        confidence: str | None = None,
        fetch_time: str | None = None,
    ):
        """
        Implements class for Intel 471 feed.
        :param auth: credentials for accessing the feed.
        :param insecure: boolean, if *false* feed HTTPS server certificate is verified. Default: *false*.
        :param tags: list of tags.
        :param tlp_color: Traffic Light Protocol color.
        :param intel471_backend: str, intel 471 backend selector.
        :param indicator: str, free text indicator filter.
        :param indicator_type: str, indicator type filter.
        :param threat_type: str, threat type filter.
        :param malware_family: str, malware family filter.
        :param confidence: str, confidence filter.
        :param fetch_time: str, fetch time filter.
        """
        self.auth = auth
        self._verify: bool = insecure
        self.tags = [] if tags is None else tags
        self.tlp_color = tlp_color
        self.intel471_backend = intel471_backend
        self.indicator = indicator
        self.indicator_type = indicator_type
        self.threat_type = threat_type
        self.malware_family = malware_family
        self.confidence = confidence
        self.fetch_time = fetch_time
        self._proxies = handle_proxy(proxy_param_name="proxy", checkbox_default_value=False)

    def get_girs(self) -> list:
        """Retrieves a list of General Intelligence Requirements (GIRs).
        Returns:
            GIRs.
        """
        result: list = []

        feed_url = FEED_URL_GIRS_TITAN

        params = {}
        params["count"] = MAX_COUNT
        params["offset"] = 0

        should_continue: bool = True
        while should_continue:
            try:
                response = requests.get(
                    url=feed_url,
                    params=params,
                    verify=self._verify,
                    proxies=self._proxies,
                    headers=self.headers,
                    auth=self.auth,
                )
                response.raise_for_status()
                data = response.json()
                girs: list = data.get("girs", [])

                if girs:
                    result.extend(girs)
                else:
                    should_continue = False

                if len(girs) < MAX_COUNT:
                    should_continue = False

                params["offset"] = params["offset"] + MAX_COUNT

            except requests.exceptions.SSLError as err:
                demisto.debug(str(err))
                raise Exception(
                    f"Connection error in the API call to {INTEGRATION_NAME}.\nCheck your not secure parameter.\n\n{err}"
                )
            except requests.ConnectionError as err:
                demisto.debug(str(err))
                raise Exception(
                    f"Connection error in the API call to {INTEGRATION_NAME}.\nCheck your Server URL parameter.\n\n{err}"
                )
            except requests.exceptions.HTTPError as err:
                demisto.debug(f"Got an error from {feed_url} while fetching GIRs {err!s} ")
            except ValueError as err:
                demisto.debug(str(err))
                raise ValueError(f"Could not parse returned data to JSON. \n\nError massage: {err}")

        return result

    def build_iterator_titan(self, save_state: bool, limit: int = -1) -> list:
        """Retrieves all entries from the Titan feed.
        Args:
            save_state: save state (cursor).
            limit: limit the results.
        Returns:
            A list of objects, containing the indicators.
        """
        result = []

        feed_url = FEED_URL_INDICATORS_TITAN
        integration_context = get_integration_context()

        params = {}
        if self.indicator:
            params["indicator"] = self.indicator
        if self.indicator_type and "All" not in self.indicator_type:
            params["indicatorType"] = " || ".join(self.indicator_type).lower()
        if self.threat_type:
            params["threatType"] = self.threat_type
        if self.malware_family:
            params["malwareFamily"] = self.malware_family
        if self.confidence:
            params["confidence"] = self.confidence
        params["count"] = str(MAX_COUNT)

        last_updated_from = integration_context.get("last_updated_from", "")
        if not last_updated_from:
            start_date, end_date = parse_date_range(self.fetch_time, utc=True, to_timestamp=True)
            last_updated_from = str(start_date)
        params["lastUpdatedFrom"] = last_updated_from

        cursor = integration_context.get("cursor", "")
        if cursor:
            params["cursor"] = cursor

        should_continue: bool = True
        while should_continue:
            encoded_params = urllib.parse.urlencode(params, quote_via=urllib.parse.quote)

            try:
                response = requests.get(
                    url=feed_url,
                    params=encoded_params,
                    verify=self._verify,
                    proxies=self._proxies,
                    headers=self.headers,
                    auth=self.auth,
                )
                response.raise_for_status()
                data = response.json()
                indicators: list = data.get("indicators", [])

                if indicators:
                    result.extend(indicators)
                else:
                    should_continue = False

                if len(indicators) < MAX_COUNT:
                    should_continue = False

                if limit > 0:
                    if len(result) >= limit:
                        should_continue = False
                        result = result[:limit]
                else:
                    if len(result) >= MAX_BATCH:
                        should_continue = False

                cursor_next = data.get("cursorNext", "")
                if cursor_next:
                    params["cursor"] = cursor_next

            except requests.exceptions.SSLError as err:
                demisto.debug(str(err))
                raise Exception(
                    f"Connection error in the API call to {INTEGRATION_NAME}.\nCheck your not secure parameter.\n\n{err}"
                )
            except requests.ConnectionError as err:
                demisto.debug(str(err))
                raise Exception(
                    f"Connection error in the API call to {INTEGRATION_NAME}.\nCheck your Server URL parameter.\n\n{err}"
                )
            except requests.exceptions.HTTPError as err:
                demisto.debug(f"Got an error from {feed_url} while fetching indicators {err!s} ")
                raise Exception(f"HTTP error in the API call to {INTEGRATION_NAME}.\nCheck your configuration.\n\n{err}")
            except ValueError as err:
                demisto.debug(str(err))
                raise ValueError(f"Could not parse returned data to JSON. \n\nError massage: {err}")

        if save_state:
            set_integration_context({"last_updated_from": last_updated_from})
            set_integration_context({"cursor": cursor})

        return result

    def build_iterator_verity471(self, save_state: bool, limit: int = -1) -> list:
        """Retrieves all entries from the Verity471 feed.
        Args:
            save_state: save state (cursor).
            limit: limit the results.
        Returns:
            A list of objects, containing the indicators.
        """
        result = []

        feed_url = FEED_URL_INDICATORS_VERITY471
        integration_context = get_integration_context()

        params = {}
        if self.indicator:
            params["text_filter"] = self.indicator
        if self.indicator_type and "All" not in self.indicator_type:
            params["type"] = self.indicator_type
        if self.threat_type:
            params["threat_type"] = self.threat_type
        if self.malware_family:
            params["malware_family_name"] = self.malware_family
        if self.confidence:
            params["confidence"] = self.confidence
        params["size"] = str(MAX_COUNT)

        cursor = integration_context.get("cursor")
        if cursor:
            params["cursor"] = cursor

        from_ts = integration_context.get("from_ts", "")
        if not from_ts:
            start_date, end_date = parse_date_range(self.fetch_time, utc=True, to_timestamp=True)
            from_ts = str(start_date)

        params["from"] = from_ts

        should_continue: bool = True
        while should_continue:
            encoded_params = urllib.parse.urlencode(params, quote_via=urllib.parse.quote)

            try:
                response = requests.get(
                    url=feed_url,
                    params=encoded_params,
                    verify=self._verify,
                    proxies=self._proxies,
                    headers=self.headers,
                    auth=self.auth,
                )
                response.raise_for_status()
                data = response.json()
                indicators: list = data.get("indicators", [])

                if indicators:
                    result.extend(indicators)
                else:
                    should_continue = False

                if len(indicators) < MAX_COUNT:
                    should_continue = False

                if limit > 0:
                    if len(result) >= limit:
                        should_continue = False
                        result = result[:limit]
                else:
                    if len(result) >= MAX_BATCH:
                        should_continue = False

                cursor = data.get("cursor_next")
                if cursor:
                    params["cursor"] = cursor

            except requests.exceptions.SSLError as err:
                demisto.debug(str(err))
                raise Exception(
                    f"Connection error in the API call to {INTEGRATION_NAME}.\nCheck your not secure parameter.\n\n{err}"
                )
            except requests.ConnectionError as err:
                demisto.debug(str(err))
                raise Exception(
                    f"Connection error in the API call to {INTEGRATION_NAME}.\nCheck your Server URL parameter.\n\n{err}"
                )
            except requests.exceptions.HTTPError as err:
                demisto.debug(f"Got an error from {feed_url} while fetching indicators {err!s} ")
                raise Exception(f"HTTP error in the API call to {INTEGRATION_NAME}.\nCheck your configuration.\n\n{err}")
            except ValueError as err:
                demisto.debug(str(err))
                raise ValueError(f"Could not parse returned data to JSON. \n\nError massage: {err}")

        if save_state:
            set_integration_context({"from_ts": from_ts})
            set_integration_context({"cursor": cursor})

        return result


def test_module(client: Client, *_) -> str:
    """Builds the iterator to check that the feed is accessible.
    Args:
        client: Client object.
    Returns:
        Outputs.
    """
    if client.intel471_backend == "Verity471":
        client.build_iterator_verity471(False, MAX_COUNT)
    else:
        client.build_iterator_titan(False, MAX_COUNT)

    return "ok"


def build_relationships(type_: str, value_: str, malware_family: str) -> list:
    """Creates a list of relationships for the indicator.
    Args:
        type_ (str): relationship type.
        value_ (str): indicator value.
        malware_family (str): malware family.
    Returns:
        List: A list of relationships.
    """
    relationships: list = []

    relationships.append(
        EntityRelationship(
            name=EntityRelationship.Relationships.INDICATOR_OF,
            entity_a=value_,
            entity_a_type=type_,
            entity_b=malware_family,
            entity_b_type="Malware",
        ).to_indicator()
    )

    return relationships


def build_indicator_titan(client: Client, raw_data: dict[str, Any], titan_girs: list) -> dict[str, Any]:
    """Creates a Titan sourced indicator object.
    Args:
        raw_data: raw data of the indicator.
        titan_girs: a list of GIRs.
    Returns:
        Dictionary representing the indicator object.
    """

    malware_family: str = raw_data.get("data", {}).get("threat", {}).get("data", {}).get("family", "")
    type_: str = INDICATOR_TYPES.get(raw_data.get("data", {}).get("indicator_type", ""), "")
    indicator_data = raw_data.get("data", {}).get("indicator_data", {})

    intel_requirements: list = []
    intel_requirement_paths: list = raw_data.get("data", {}).get("intel_requirements", [])
    for irp in intel_requirement_paths:
        gir: dict = next(filter(lambda g: g.get("data", {}).get("gir", {}).get("path", {}) == irp, titan_girs), {})
        if gir:
            name = gir.get("data", {}).get("gir", {}).get("name", "")
            intel_requirements.append(f"GIR: {irp} - {name}")

    value_: str = ""
    fields: dict = {}
    if type_ == FeedIndicatorType.File:
        value_ = indicator_data.get("file", {}).get("sha256", "")
        fields["md5"] = indicator_data.get("file", {}).get("md5", "")
        fields["sha1"] = indicator_data.get("file", {}).get("sha1", "")
        fields["sha256"] = indicator_data.get("file", {}).get("sha256", "")
        fields["ssdeep"] = indicator_data.get("file", {}).get("ssdeep", "")
        fields["filetype"] = indicator_data.get("file", {}).get("type", "")
        fields["downloadurl"] = indicator_data.get("file", {}).get("download_url", "")
        fields["size"] = indicator_data.get("file", {}).get("size", 0)
    elif type_ == FeedIndicatorType.IP:
        value_ = indicator_data.get("address", "")
    elif type_ == FeedIndicatorType.URL:
        value_ = indicator_data.get("url", "")

    fields["firstseenbysource"] = datetime.fromtimestamp(raw_data.get("activity", {}).get("first") / 1000).isoformat("T")
    fields["lastseenbysource"] = datetime.fromtimestamp(raw_data.get("activity", {}).get("last") / 1000).isoformat("T")

    fields["trafficlightprotocol"] = client.tlp_color

    fields["tags"] = [raw_data.get("data", {}).get("context", {}).get("description", "")]
    fields["tags"].append(raw_data.get("data", {}).get("mitre_tactics", ""))
    fields["tags"].extend(intel_requirements)
    fields["tags"].extend(client.tags)

    indicator_obj = {
        "value": value_,
        "type": type_,
        "rawJSON": raw_data,
        "fields": fields,
        "relationships": build_relationships(type_, value_, malware_family),
    }

    return indicator_obj


def build_indicator_verity471(client: Client, raw_data: dict[str, Any]) -> dict[str, Any]:
    """Creates a Verity471 sourced indicator object.
    Args:
        raw_data: raw data of the indicator.
        titan_girs: a list of GIRs.
    Returns:
        Dictionary representing the indicator object.
    """

    malware_family: str = raw_data.get("threat", {}).get("data", {}).get("malware_family", {}).get("name", "")
    type_: str = INDICATOR_TYPES.get(raw_data.get("type", ""), "")
    indicator_data = raw_data.get("data", {})

    intel_requirements: list = []
    girs: list = raw_data.get("classification", {}).get("girs", [])
    for gir in girs:
        intel_requirements.append(f"GIR: {gir.get('path', '')} - {gir.get('name', '')}")

    value_: str = ""
    fields: dict = {}
    if type_ == FeedIndicatorType.File:
        value_ = indicator_data.get("file", {}).get("sha256", "")
        fields["md5"] = indicator_data.get("file", {}).get("md5", "")
        fields["sha1"] = indicator_data.get("file", {}).get("sha1", "")
        fields["sha256"] = indicator_data.get("file", {}).get("sha256", "")
        fields["ssdeep"] = indicator_data.get("file", {}).get("ssdeep", "")
        fields["filetype"] = indicator_data.get("file", {}).get("type", "")
        fields["downloadurl"] = (
            f"{FEED_URL_MALWARE_FILE_VERITY471}/{indicator_data.get('file', {}).get('sha256', '')}.zip/download"
        )
        fields["size"] = indicator_data.get("file", {}).get("size", 0)
    elif type_ == FeedIndicatorType.IP:
        value_ = indicator_data.get("ipv4", {}).get("ip_address", "")
    elif type_ == FeedIndicatorType.URL:
        value_ = indicator_data.get("url", "")
    elif type_ == FeedIndicatorType.Domain:
        value_ = indicator_data.get("domain", "")
    elif type_ == FeedIndicatorType.Email:
        value_ = indicator_data.get("email", "")
    elif type_ == "yara":
        value_ = ""

    fields["firstseenbysource"] = raw_data.get("activity", {}).get("first_seen_ts", "")
    fields["lastseenbysource"] = raw_data.get("activity", {}).get("last_seen_ts", "")

    fields["trafficlightprotocol"] = client.tlp_color

    fields["tags"] = [raw_data.get("description", "")]
    kill_chain_phases = raw_data.get("kill_chain_phases", [])
    for kill_chain_phase in kill_chain_phases:
        fields["tags"].append(kill_chain_phase.get("phase_name", ""))
    fields["tags"].extend(intel_requirements)
    fields["tags"].extend(client.tags)

    indicator_obj = {
        "value": value_,
        "type": type_,
        "rawJSON": raw_data,
        "fields": fields,
        "relationships": build_relationships(type_, value_, malware_family),
    }

    return indicator_obj


def fetch_indicators(client: Client, save_state: bool, limit: int = -1) -> list[dict]:
    """Retrieves indicators from the feed.
    Args:
        client: Client object with request.
        save_state: save state (cursor).
        limit: limit the results.
    Returns:
        Indicators.
    """
    titan_girs = []
    indicators: list = []
    if client.intel471_backend == "Verity471":
        iterator = client.build_iterator_verity471(save_state, limit)
    else:
        titan_girs = client.get_girs()
        iterator = client.build_iterator_titan(save_state, limit)

    for item in iterator:
        if client.intel471_backend == "Verity471":
            indicator_obj = build_indicator_verity471(client, item)
        else:
            indicator_obj = build_indicator_titan(client, item, titan_girs)

        if indicator_obj["value"]:
            indicators.append(indicator_obj)

    return indicators


def get_indicators_command(client: Client, args: dict[str, str]) -> CommandResults:
    """Wrapper for retrieving indicators from the feed to the war-room.
    Args:
        client: Client object with request.
        args: demisto.args().
    Returns:
        CommandResults object containing the indicators retrieved.
    """
    limit = arg_to_number(demisto.args().get("limit")) or 100
    indicators = fetch_indicators(client, False, limit)

    hr_indicators = []
    for indicator in indicators:
        hr_indicators.append(
            {
                "Value": indicator.get("value"),
                "Type": indicator.get("type"),
                "rawJSON": indicator.get("rawJSON"),
                "fields": indicator.get("fields"),
            }
        )

    human_readable = tableToMarkdown(
        "Indicators from Intel 471:", hr_indicators, headers=["Value", "Type", "rawJSON", "fields"], removeNull=True
    )
    return CommandResults(
        readable_output=human_readable,
        outputs_prefix="Intel471Feed.Indicators",
        outputs_key_field="value",
        raw_response=indicators,
    )


def fetch_indicators_command(client: Client, args: dict[str, str]) -> list[dict]:
    """Wrapper for fetching indicators from the feed to the Indicators tab.
    Args:
        client: Client object with request.
    Returns:
        Indicators.
    """
    indicators = fetch_indicators(client, True)
    return indicators


def main():
    """
    PARSE AND VALIDATE INTEGRATION PARAMS
    """
    args = demisto.args()
    params = demisto.params()
    use_ssl = not params.get("insecure", False)
    tags = argToList(params.get("feedTags"))
    tlp_color = params.get("tlp_color")
    intel471_backend = params.get("intel471_backend")
    indicator = params.get("indicator")
    indicator_type = params.get("indicator_type")
    threat_type = THREAT_TYPE
    malware_family = params.get("malware_family")
    confidence = params.get("confidence")
    fetch_time = params.get("fetch_time")
    credentials = params.get("credentials", {})
    if not credentials:
        raise DemistoException("Integration credentials not entered.")
    else:
        auth = (credentials.get("identifier", ""), credentials.get("password", ""))

    command = demisto.command()
    demisto.info(f"Command being called is {command}")

    try:
        client = Client(
            auth,
            use_ssl,
            tags,
            tlp_color,
            intel471_backend,
            indicator,
            indicator_type,
            threat_type,
            malware_family,
            confidence,
            fetch_time,
        )

        if command == "test-module":
            return_results(test_module(client, params))
        elif command == "intel471-indicators-get-indicators":
            return_results(get_indicators_command(client, args))
        elif command == "fetch-indicators":
            indicators = fetch_indicators_command(client, args)
            for iter_ in batch(indicators, batch_size=2000):
                demisto.createIndicators(iter_)
        else:
            raise NotImplementedError(f"Command {command} is not implemented.")

    except Exception as err:
        err_msg = f"Error in {INTEGRATION_NAME} Integration. [{err}]"
        return_error(err_msg)


if __name__ in ["__main__", "builtin", "builtins"]:
    main()