Cisco Secure Malware Analytics

import demistomock as demisto  # noqa: F401
import urllib3
from CommonServerPython import *  # noqa: F401

# disable insecure warnings
urllib3.disable_warnings()

FEED_NAME = [
    "dga-dns",
    "autorun-registry",
    "banking-dns",
    "dll-hijacking-dns",
    "doc-net-com-dns",
    "downloaded-pe-dns",
    "dynamic-dns",
    "irc-dns",
    "modified-hosts-dns",
    "parked-dns",
    "public-ip-check-dns",
    "ransomware-dns",
    "rat-dns",
    "scheduled-tasks",
    "sinkholed-ip-dns",
    "stolen-cert-dns",
]


class Client(BaseClient):
    def __init__(
        self,
        api_key,
        verify,
        proxy,
        first_fetch,
        feed_name,
        feed_tags,
        tlp_color,
        create_relationships,
        base_url="https://panacea.threatgrid.com/api/v3/feeds/",
    ):
        """Implements class for the feed.

        Args:
            api_key: API Key.
            verify: boolean, if *false* feed HTTPS server certificate is verified. Default: *false*
            first_fetch: The date from which to start fetching feeds
            feed_name: The feed names to fetch
            feed_tags: feed tags.
            tlp_color: Traffic Light Protocol color.
            create_relationships: boolean, if *false* no relationships are created. Default: *true*
        """
        super().__init__(base_url=base_url, verify=verify, proxy=proxy)
        self.first_fetch = first_fetch
        self.feed_name = feed_name
        self.feed_tags = feed_tags
        self.tlp_color = tlp_color
        self.create_relationships = create_relationships
        self._api_key = api_key

    def get_indicators(self, feed_name: str, time_stamp: Optional[str]):
        """Retrieves all indicators from the feed.

        Args:
            :param feed_name: the feed name to fetch
            :param time_stamp: Optional - fetch feeds only from this day

        """
        suffix = f"{feed_name}_{time_stamp}.json" if time_stamp else f"{feed_name}.json"
        try:
            return self._http_request("GET", suffix, params={"api_key": self._api_key}, timeout=70)
        except Exception as error:
            if "Unauthorized" in str(error):
                raise DemistoException("Authorization Error: make sure the API Key is correctly set")
            else:
                raise DemistoException(error)


def create_related_ips(obj: dict, feed_name: str, tlp_color: str) -> list[dict]:
    """creates list indicators of the related ips.

    Args:
        :param obj: from the response
        :param feed_name: the feed That it was fetch there
        :param tlp_color: Traffic Light Protocol color
        :return: indicators list

    """
    return [
        (
            {
                "value": ip,
                "type": "IP",
                "rawJSON": obj,
                "fields": {
                    "Tags": [feed_name],
                    "reportedby": "CiscoSMA",
                    "Description": obj.get("description"),
                    "FirstSeenBySource": obj.get("timestamp"),
                    "trafficlightprotocol": tlp_color,
                },
            }
        )
        for ip in obj.get("ips")  # type:ignore
    ]


def create_related_files(obj: dict, feed_name: str, tlp_color: str) -> dict:
    """creates dict of the related file indicator.

    Args:
        :param obj: from the response
        :param feed_name: the feed That it was fetch there
        :param tlp_color: Traffic Light Protocol color
        :return: indicator

    """
    return {
        "value": obj.get("sample_sha256"),
        "type": "File",
        "rawJSON": obj,
        "fields": {
            "Tags": [feed_name],
            "reportedby": "CiscoSMA",
            "Description": obj.get("description"),
            "FirstSeenBySource": obj.get("timestamp"),
            "MD5": obj.get("sample_md5"),
            "SHA1": obj.get("sample_sha1"),
            "trafficlightprotocol": tlp_color,
        },
    }


def create_entity_relationships(feed_related_indicators: list, current_domain: str) -> list:
    """

    Args:
        feed_related_indicators: Indicators related to the domain returned from fetch_indicators.
        current_domain: Returned indicator from fetch_indicators.

    Returns:
        relationships: EntityRelationships list.

    """
    relationships = []
    for relation in feed_related_indicators:
        if relation.get("value") != "":
            entity_relation = EntityRelationship(
                name=EntityRelationship.Relationships.RELATED_TO,
                entity_a=current_domain,
                entity_a_type="Domain",
                entity_b=relation.get("value"),
                entity_b_type=relation.get("type"),
            )
            relationships.append(entity_relation.to_indicator())
    return relationships


def test_module(client: Client) -> str:
    """Goes through all the code of the fetch.
    Args:
        client: Client object with request

    Returns:
        str: ok.
    """
    fetch_indicators(client)
    return "ok"


def fetch_indicators(client: Client, time_to_stop_fetch: Optional[datetime] = None) -> tuple[list, bool]:
    """Retrieves indicators from the feed

    Args:
        client: Client object with request
        time_to_stop_fetch: Time to stop the fetch before it falls on timeout

    Returns:
        List. Processed indicator from feed.
    """
    fetch_time_stamp = None
    files = ["sample_sha256", "sample_sha1", "sample_md5"]
    if client.first_fetch.date() < arg_to_datetime("now").date():  # type:ignore
        fetch_time_stamp = client.first_fetch.date().isoformat()
    else:
        client.first_fetch = arg_to_datetime("now")

    indicators = []
    domains_map: dict[str, dict] = {}
    for name in client.feed_name:
        if time_to_stop_fetch and arg_to_datetime("now") > time_to_stop_fetch:  # type:ignore
            return [], False
        result = client.get_indicators(feed_name=name, time_stamp=fetch_time_stamp)
        for obj in result:
            feed_related_indicators = [{"type": "IP", "value": ip} for ip in obj.get("ips")]
            feed_related_indicators.extend([{"type": "File", "value": obj.get(file_hash)} for file_hash in files])
            current_domain = obj.get("domain")
            relationships = create_entity_relationships(feed_related_indicators, current_domain)
            if current_domain in domains_map:
                if name not in domains_map[current_domain]["fields"]["Tags"]:
                    domains_map[current_domain]["fields"]["Tags"].append(name)
                domains_map[current_domain]["fields"]["FeedRelatedIndicators"].extend(
                    related
                    for related in feed_related_indicators
                    if related not in domains_map[current_domain]["fields"]["FeedRelatedIndicators"]
                )
                existing_relations = [relation.get("entityB") for relation in domains_map[current_domain]["relationships"]]
                new_relations = filter(lambda relation: relation.get("entityB") not in existing_relations, relationships)
                domains_map[current_domain]["relationships"].extend(new_relations)

            else:
                domains_map[current_domain] = {
                    "value": current_domain,
                    "type": "Domain",
                    "rawJSON": obj,
                    "relationships": relationships,
                    "fields": {
                        "Tags": [name],
                        "reportedby": "CiscoSMA",
                        "FeedRelatedIndicators": feed_related_indicators,
                        "Description": obj.get("description"),
                        "FirstSeenBySource": obj.get("timestamp"),
                    },
                }
            if client.tlp_color:
                domains_map[current_domain]["fields"]["trafficlightprotocol"] = client.tlp_color
            if client.feed_tags:
                domains_map[current_domain]["fields"]["Tags"] += client.feed_tags
            indicators += create_related_ips(obj, name, client.tlp_color)
            indicators.append(create_related_files(obj, name, client.tlp_color))

    indicators += list(domains_map.values())
    return indicators, True


def fetch_indicators_command(client: Client, time_to_stop_fetch: datetime = None):
    """Retrieves indicators from the feed

    Args:
        client: Client object with request
        time_to_stop_fetch: Time to stop the fetch before it falls on the docker timeout

    Returns:
        List. Processed indicator from feed.
    """
    current_date = arg_to_datetime("now").date()  # type:ignore
    while client.first_fetch.date() <= current_date:
        indicators, status = fetch_indicators(client, time_to_stop_fetch)
        if not status:
            return
        for iter_ in batch(indicators, batch_size=2000):
            demisto.createIndicators(iter_)
        # Each request returns one day's indicators, so we'll keep the date of the next day in IntegrationContext.
        next_fetch = dateparser.parse("tomorrow", settings={"RELATIVE_BASE": client.first_fetch})  # type:ignore
        if client.first_fetch.date() < current_date:
            demisto.setIntegrationContext({"last_fetch": next_fetch.isoformat()})  # type:ignore
        client.first_fetch = next_fetch
        demisto.debug(f"{len(indicators)} XSOAR Indicators were created.")
    return


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:
        Demisto Outputs.
    """
    limit = int(args.get("limit", 50))
    indicators, status = fetch_indicators(client)

    limit_indicators = indicators[:limit]

    readable_output = tableToMarkdown("CiscoSMA Indicators:", t=limit_indicators, headers=["value", "type"])
    command_results = CommandResults(
        outputs_prefix="CiscoSMA",
        outputs_key_field="value",
        outputs=limit_indicators,
        readable_output=readable_output,
        raw_response=indicators,
    )
    return command_results


def reset_last_run():
    """
    Reset the last run from the integration context
    """
    demisto.setIntegrationContext({})
    return CommandResults(readable_output="Fetch history deleted successfully")


def main():  # pragma: no cover
    """
    PARSE AND VALIDATE FEED PARAMS
    """
    try:
        time_to_stop_fetch = arg_to_datetime(arg="in 25 minutes")  # type:ignore # To stop before docker timeout.
        params = demisto.params()
        args = demisto.args()
        base_url = params["base_url"]
        api_key = params.get("api_key", {}).get("password")
        if not api_key:
            raise DemistoException("API Key was not given. Please insert your API Key.")
        verify = not params.get("insecure", False)
        proxy = params.get("proxy", False)
        feed_name = params.get("feed_name", "")
        if "all" in feed_name:
            feed_name = FEED_NAME
        feed_tags = argToList(params.get("feedTags"))
        tlp_color = params.get("tlp_color")
        create_relationships = params.get("create_relationships", True)
        if fetch_date_string := demisto.getIntegrationContext().get("last_fetch"):
            first_fetch = arg_to_datetime(fetch_date_string)  # type:ignore
        else:
            if not (first_fetch_param := params.get("first_fetch")):
                first_fetch_param = "today"
            first_fetch = arg_to_datetime(arg=first_fetch_param, arg_name="First fetch")  # type:ignore
        command = demisto.command()
        demisto.debug(f"Command being called is: {command}")

        client = Client(
            base_url=base_url,
            api_key=api_key,
            verify=verify,
            proxy=proxy,
            feed_name=feed_name,
            first_fetch=first_fetch,
            feed_tags=feed_tags,
            tlp_color=tlp_color,
            create_relationships=create_relationships,
        )

        if command == "test-module":
            return_results(test_module(client))

        elif command == "fetch-indicators":
            fetch_indicators_command(client, time_to_stop_fetch)

        elif command == "cisco-sma-get-indicators":
            return_results(get_indicators_command(client, args))

        elif command == "cisco-sma-reset-fetch-indicators":
            return_results(reset_last_run())

        else:
            raise NotImplementedError(f'Command "{command}" is not implemented.')

    except Exception as err:
        return_error(f"Failed to execute {demisto.command()} command.\nError:\n{err!s}")


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