Google Threat Intelligence Threat Lists

Use this feed integration to fetch Google Threat Intelligence Threat Lists matches as indicators.

Data Enrichment & Threat Intelligence · GoogleThreatIntelligence · Feed

Details

ID	`Google Threat Intelligence Threat Lists`
Provider	Google
Category	Data Enrichment & Threat Intelligence
From Version	5.5.0
Docker Image	`demisto/python3:3.12.13.10116658`
Supported Modules	Agentix XSIAM

README

Use this feed integration to fetch Google Threat Intelligence Threat Lists matches as indicators. It processes the latest finished job retrieving its matches based on the limit parameter (10 by default) in every fetch until there are no more matches for that job.

Configure Google Threat Intelligence Threat Lists on Cortex XSOAR

Navigate to Settings > Integrations > Servers & Services.
Search for Google Threat Intelligence Threat Lists.
Click Add instance to create and configure a new integration instance.

Parameter	Description	Required
feed	The fetch indicators.	False
credentials	API Key.	True
feed_type	Feed type.	True
filter	Filter your Threat Lists (e.g., “gti_score:70+ positives:10- has:campaigns”). Leave empty to receive all.	False
limit	The maximum number of results to return. Default is 10.	False
feedReputation	The indicator reputation.	False
feedReliability	The source’s reliability.	True
tlp_color	The Traffic Light Protocol (TLP) designation to apply to indicators fetched from the feed. More information about the protocol can be found at https://us-cert.cisa.gov/tlp	False
feedExpirationPolicy	The feed’s expiration policy.	False
feedFetchInterval	The feed fetch interval.	False
feedBypassExclusionList	Whether to bypass exclusion list.	False

Click Test to validate the URLs, token, and connection.

Commands

You can execute these commands from the Cortex XSOAR CLI, as part of an automation, or in a playbook.
After you successfully execute a command, a DBot message appears in the War Room with the command details.

gti-threatlists-get-indicators

Gets the matches from Google Threat Intelligence Threat Lists.

Base Command

gti-threatlists-get-indicators

Input

Argument Name	Description	Required
feed_type	Feed type.	Required
package	Package in ‘%Y%m%d%H’ format. If not given, the latest package is taken.	Optional
filter	Filter your Threat Lists (e.g., “gti_score:70+ positives:10- has:campaigns”). Leave empty to receive all.	Optional
limit	The maximum number of results to return. Default is 10.	Optional

Context Output

There is no context output for this command.

Command Example

!gti-threatlists-get-indicators
!gti-threatlists-get-indicators feed=malware package=2025021910 filter="gti_score:70+" limit=10

Human Readable Output

Indicators from Google Threat Intelligence Threat Lists

Id	Detections	Gti Threat Score	Gti Severity	Gti Verdict	Malware Families	Threat Actors
f221425286c9073cbb2168f73120b6…	59/69	80	SEVERITY_LOW	VERDICT_MALICIOUS	beacon	SWEED

Configuration parameters

credentials — API Key (leave empty. Fill in the API key in the password field.) (required)
feed_type — Feed type.
filter — Filter
limit — The maximum number of results to return. If 0 all results will be returned.
feed — Fetch indicators
feedReputation — Indicator Reputation
feedReliability — Source Reliability (required)
feedExpirationPolicy —
feedExpirationInterval —
feedFetchInterval — Feed Fetch Interval
feedBypassExclusionList — Bypass exclusion list
feedTags — Tags
tlp_color — Traffic Light Protocol Color

Commands (1)

gti-threatlists-get-indicators
Gets the matches from the latest Feed.

Source

FeedThreatLists.py FeedThreatLists_test.py FeedThreatLists.yml FeedThreatLists_description.md

Download History Raw

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

FEED_STR = {
    "cryptominer": "Cryptominer",
    "first-stage-delivery-vectors": "First stage delivery vectors",
    "infostealer": "Infostealer",
    "iot": "IoT",
    "linux": "Linux",
    "malicious-network-infrastructure": "Malicious network infrastructure",
    "malware": "Malware",
    "mobile": "Mobile",
    "osx": "OSX",
    "phishing": "Phishing",
    "ransomware": "Ransomware",
    "threat-actor": "Threat actor",
    "trending": "Trending",
    "vulnerability-weaponization": "Vulnerability weaponization",
}


def _get_current_package():
    """Gets current package for Threat Lists."""
    time_obj = datetime.utcnow() - timedelta(hours=2)
    package = time_obj.strftime("%Y%m%d%H")
    return package


class DetectionRatio:
    """Class for detections."""

    malicious = 0
    total = 0

    def __init__(self, last_analysis_stats: dict):
        self.malicious = last_analysis_stats.get("malicious", 0)
        self.total = sum(last_analysis_stats.values())

    def __repr__(self):
        return f"{self.malicious}/{self.total}"


class Client(BaseClient):
    """Client for Google Threat Intelligence API."""

    def get_threat_list(self, feed_type: str, package: str, filter_query: str = None, limit: int = 10) -> dict:
        """Get indicators from GTI API."""
        return self._http_request(
            "GET",
            f"threat_lists/{feed_type}/{package}",
            params=assign_params(
                query=filter_query,
                limit=limit,
            ),
        )

    def fetch_indicators(
        self, feed_type: str, package: str = None, filter_query: str = None, limit: int = 10, fetch_command: bool = False
    ) -> list:
        """Retrieves matches for a given feed type."""
        package = package or _get_current_package()
        filter_query = filter_query or ""

        if fetch_command and self.get_last_run() == package:
            return []

        response = self.get_threat_list(feed_type, package, filter_query.strip(), limit)

        if fetch_command:
            self.set_last_run(package)

        return response.get("iocs", [])

    @staticmethod
    def set_last_run(package: str = None):
        """Sets last threat feed."""
        current_package = package or _get_current_package()
        demisto.setLastRun({"last_threat_feed": current_package})

    @staticmethod
    def get_last_run() -> str:
        """Gets last threat feed, or '' if no last run."""
        return (demisto.getLastRun() or {}).get("last_threat_feed", "")


def test_module(client: Client, args: dict) -> str:
    """Tests module."""
    try:
        client.fetch_indicators("malware")
    except Exception:
        raise Exception(
            "Could not fetch Google Threat Intelligence IoC Threat Lists\n"
            "\nCheck your API key and your connection to Google Threat Intelligence."
        )
    return "ok"


def _gti_verdict_to_dbot_score(gti_verdict: str):
    """Parses GTI verdict to DBotScore."""
    return {
        "VERDICT_BENIGN": 1,
        "VERDICT_SUSPICIOUS": 2,
        "VERDICT_MALICIOUS": 3,
    }.get(gti_verdict, 0)


def _add_gti_attributes(indicator_obj: dict, item: dict):
    """Addes GTI attributes."""

    # GTI assessment
    attributes = item.get("attributes", {})
    gti_assessment = attributes.get("gti_assessment", {})
    gti_threat_score = gti_assessment.get("threat_score", {}).get("value")
    gti_severity = gti_assessment.get("severity", {}).get("value")
    gti_verdict = gti_assessment.get("verdict", {}).get("value")

    # Relationships
    relationships = item.get("relationships", {})
    malware_families: list[str] = [
        x.get("attributes", {}).get("name") for x in relationships.get("malware_families", {}).get("data", [])
    ]
    malware_families = list(set(malware_families))
    threat_actors: list[str] = [
        x.get("attributes", {}).get("name") for x in relationships.get("threat_actors", {}).get("data", [])
    ]
    threat_actors = list(set(threat_actors))

    indicator_obj["fields"].update(
        {
            "gtithreatscore": gti_threat_score,
            "gtiseverity": gti_severity,
            "gtiverdict": gti_verdict,
            "malwarefamily": malware_families,
            "actor": threat_actors,
        }
    )
    indicator_obj.update(
        {
            "score": _gti_verdict_to_dbot_score(gti_verdict),
            "gti_threat_score": gti_threat_score,
            "gti_severity": gti_severity,
            "gti_verdict": gti_verdict,
            "malware_families": malware_families,
            "threat_actors": threat_actors,
            "relationships": [
                EntityRelationship(
                    name=EntityRelationship.Relationships.PART_OF,
                    entity_a=indicator_obj["value"],
                    entity_a_type=indicator_obj["type"],
                    entity_b=malware_family.title(),
                    entity_b_type=ThreatIntel.ObjectsNames.MALWARE,
                    reverse_name=EntityRelationship.Relationships.CONTAINS,
                ).to_indicator()
                for malware_family in malware_families
            ]
            + [
                EntityRelationship(
                    name=EntityRelationship.Relationships.ATTRIBUTED_BY,
                    entity_a=indicator_obj["value"],
                    entity_a_type=indicator_obj["type"],
                    entity_b=threat_actor.title(),
                    entity_b_type=ThreatIntel.ObjectsNames.THREAT_ACTOR,
                    reverse_name=EntityRelationship.Relationships.ATTRIBUTED_TO,
                ).to_indicator()
                for threat_actor in threat_actors
            ],
        }
    )

    return indicator_obj


def _get_indicator_type(item: dict):
    """Gets indicator type."""
    if item.get("type") == "file":
        return FeedIndicatorType.File
    if item.get("type") == "domain":
        return FeedIndicatorType.Domain
    if item.get("type") == "url":
        return FeedIndicatorType.URL
    if item.get("type") == "ip_address":
        return FeedIndicatorType.IP
    raise ValueError(f'Unknown type: {item.get("type")}. ID: {item.get("id")}')


def _get_indicator_id(item: dict):
    """Gets indicator ID."""
    if item.get("type") == "url":
        return item.get("attributes", {}).get("url") or item.get("id")
    return item.get("id")


def _add_file_attributes(indicator_obj: dict, attributes: dict) -> dict:
    """Adds file attributes."""
    indicator_obj["fields"].update(
        {
            "md5": attributes.get("md5"),
            "sha1": attributes.get("sha1"),
            "sha256": attributes.get("sha256"),
            "ssdeep": attributes.get("ssdeep"),
            "fileextension": attributes.get("type_extension"),
            "filetype": attributes.get("type_tag"),
            "imphash": attributes.get("pe_info", {}).get("imphash"),
            "displayname": attributes.get("meaningful_name"),
            "name": attributes.get("meaningful_name"),
            "size": attributes.get("size"),
            "creationdate": attributes.get("creation_date"),
            "firstseenbysource": attributes.get("first_submission_date"),
            "lastseenbysource": attributes.get("last_submission_date"),
        }
    )

    return indicator_obj


def _add_domain_attributes(indicator_obj: dict, attributes: dict) -> dict:
    """Adds domain attributes."""
    whois: str = attributes.get("whois", "")

    admin_country = re.search(r"Admin Country:\s*([^\n]+)", whois)
    admin_email = re.search(r"Admin Email:\s*([^\n]+)", whois)
    admin_name = re.search(r"Admin Name:\s*([^\n]+)", whois)
    admin_phone = re.search(r"Admin Phone:\s*([^\n]+)", whois)

    registrant_country = re.search(r"Registrant Country:\s*([^\n]+)", whois)
    registrant_email = re.search(r"Registrant Email:\s*([^\n]+)", whois)
    registrant_name = re.search(r"Registrant Name:\s*([^\n]+)", whois)
    registrant_phone = re.search(r"Registrant Phone:\s*([^\n]+)", whois)

    registrar_abuse_email = re.search(r"Registrar Abuse Contact Email:\s*([^\n]+)", whois)
    registrar_abuse_phone = re.search(r"Registrar Abuse Contact Phone:\s*([^\n]+)", whois)

    indicator_obj["fields"].update(
        {
            "creationdate": attributes.get("creation_date"),
            "admincountry": admin_country.group(1) if admin_country else None,
            "adminemail": admin_email.group(1) if admin_email else None,
            "adminname": admin_name.group(1) if admin_name else None,
            "adminphone": admin_phone.group(1) if admin_phone else None,
            "registrantcountry": registrant_country.group(1) if registrant_country else None,
            "registrantemail": registrant_email.group(1) if registrant_email else None,
            "registrantname": registrant_name.group(1) if registrant_name else None,
            "registrantphone": registrant_phone.group(1) if registrant_phone else None,
            "registrarabuseemail": registrar_abuse_email.group(1) if registrar_abuse_email else None,
            "registrarabusephone": registrar_abuse_phone.group(1) if registrar_abuse_phone else None,
            "registrarname": attributes.get("registrar"),
            "firstseenbysource": attributes.get("first_seen_itw_date"),
            "lastseenbysource": attributes.get("last_seen_itw_date"),
        }
    )

    return indicator_obj


def _add_url_attributes(indicator_obj: dict, attributes: dict) -> dict:
    """Adds URL attributes."""
    indicator_obj["fields"].update(
        {
            "firstseenbysource": attributes.get("first_submission_date"),
            "lastseenbysource": attributes.get("last_submission_date"),
        }
    )

    return indicator_obj


def _add_ip_attributes(indicator_obj: dict, attributes: dict) -> dict:
    """Adds IP attributes."""
    indicator_obj["fields"].update(
        {
            "countrycode": attributes.get("country"),
            "firstseenbysource": attributes.get("first_seen_itw_date"),
            "lastseenbysource": attributes.get("last_seen_itw_date"),
        }
    )

    return indicator_obj


def _add_dedicated_attributes(indicator_obj: dict, attributes: dict) -> dict:
    """Adds dedicated attributes to indicator object."""
    if indicator_obj["type"] == FeedIndicatorType.File:
        return _add_file_attributes(indicator_obj, attributes)
    if indicator_obj["type"] == FeedIndicatorType.Domain:
        return _add_domain_attributes(indicator_obj, attributes)
    if indicator_obj["type"] == FeedIndicatorType.URL:
        return _add_url_attributes(indicator_obj, attributes)
    if indicator_obj["type"] == FeedIndicatorType.IP:
        return _add_ip_attributes(indicator_obj, attributes)
    raise ValueError(f'Unknown type: {indicator_obj["type"]}. ID: {indicator_obj["id"]}')


def _create_indicator(item: dict) -> dict:
    """Creates indicator object."""
    indicator_type = _get_indicator_type(item)
    indicator_id = _get_indicator_id(item)

    attributes: dict = item.get("attributes", {})

    detection_ratio = DetectionRatio(attributes.get("last_analysis_stats", {}))

    indicator_obj = {
        "type": indicator_type,
        "value": indicator_id,
        "service": "Google Threat Intelligence",
        "fields": {
            "tags": attributes.get("tags") or None,
            "updateddate": attributes.get("last_modification_date"),
            "detectionengines": detection_ratio.total,
            "positivedetections": detection_ratio.malicious,
        },
        "rawJSON": {
            "type": indicator_type,
            "value": indicator_id,
            "attributes": attributes,
            "relationships": item.get("relationships", {}),
        },
        "id": indicator_id,
        "detections": str(detection_ratio),
    }

    indicator_obj = _add_gti_attributes(indicator_obj, item)
    indicator_obj = _add_dedicated_attributes(indicator_obj, attributes)

    return indicator_obj


def fetch_indicators_command(
    client: Client,
    feed_type: str,
    package: str = None,
    filter_query: str = None,
    limit: int = 10,
    tlp_color: str = None,
    feed_tags: list = None,
    fetch_command: bool = False,
) -> list[dict]:
    """Retrieves indicators from the feed.
    Args:
        client (Client): Client object with request
        feed_type (str): Feed type
        package (string): Package in '%Y%m%d%H' format
        filter_query (string): filter query
        limit (int): Limit the results
        tlp_color (str): Traffic Light Protocol color
        feed_tags (list): Tags to assign fetched indicators
        fetch_command (bool): Whether command is used as fetch command.
    Returns:
        Indicators.
    """
    indicators = []

    raw_indicators = client.fetch_indicators(
        feed_type, package=package, filter_query=filter_query, limit=limit, fetch_command=fetch_command
    )

    # extract values from iterator
    for item in raw_indicators:
        try:
            indicator_obj = _create_indicator(item.get("data", {}))
        except ValueError as exc:
            demisto.info(str(exc))
            continue

        if feed_tags:
            indicator_obj["fields"]["tags"] = feed_tags

        if tlp_color:
            indicator_obj["fields"]["trafficlightprotocol"] = tlp_color

        indicators.append(indicator_obj)

    return indicators


def get_indicators_command(client: Client, params: dict[str, str], args: dict[str, str]) -> CommandResults:
    """Wrapper for retrieving indicators from the feed to the war-room.
    Args:
        client: Client object with request
        params: demisto.params()
        args: demisto.args()
    Returns:
        Outputs.
    """
    tlp_color = params.get("tlp_color")
    feed_tags = argToList(params.get("feedTags", ""))
    feed_type = args.get("feed_type", "malware")
    package = args.get("package")
    filter_query = args.get("filter")
    limit = int(args.get("limit", 10))

    indicators = fetch_indicators_command(
        client,
        feed_type,
        package=package,
        filter_query=filter_query,
        limit=limit,
        tlp_color=tlp_color,
        feed_tags=feed_tags,
    )

    human_readable = tableToMarkdown(
        f"Indicators from Google Threat Intelligence {FEED_STR.get(feed_type, feed_type)} Threat List:",
        indicators,
        headers=[
            "id",
            "detections",
            "gti_threat_score",
            "gti_severity",
            "gti_verdict",
            "malware_families",
            "threat_actors",
        ],
        headerTransform=string_to_table_header,
        removeNull=True,
    )

    return CommandResults(
        readable_output=human_readable,
        outputs_prefix="",
        outputs_key_field="",
        raw_response=indicators,
        outputs={},
    )


def main():
    """main function, parses params and runs command functions."""
    params = demisto.params()

    feed_type = params.get("feed_type", "malware")
    filter_query = params.get("filter")
    limit = int(params.get("limit", 10))
    tlp_color = params.get("tlp_color")
    feed_tags = argToList(params.get("feedTags", ""))

    # If your Client class inherits from BaseClient, SSL verification is
    # handled out of the box by it, just pass ``verify_certificate`` to
    # the Client constructor
    secure = not params.get("insecure", False)

    # If your Client class inherits from BaseClient, system proxy is handled
    # out of the box by it, just pass ``proxy`` to the Client constructor
    proxy = params.get("proxy", False)

    command = demisto.command()

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

    try:
        client = Client(
            base_url="https://www.virustotal.com/api/v3/",
            verify=secure,
            proxy=proxy,
            headers={
                "x-apikey": params.get("credentials", {}).get("password"),
                "x-tool": "CortexGTIFeeds",
            },
        )

        if command == "test-module":
            # This is the call made when pressing the integration Test button.
            return_results(test_module(client, {}))

        elif command == "gti-threatlists-get-indicators":
            # This is the command that fetches a limited number of indicators
            # from the feed source and displays them in the war room.
            return_results(get_indicators_command(client, params, demisto.args()))

        elif command == "fetch-indicators":
            # This is the command that initiates a request to the feed endpoint
            # and create new indicators objects from the data fetched. If the
            # integration instance is configured to fetch indicators, then this
            # is the command that will be executed at the specified feed fetch
            # interval.
            indicators = fetch_indicators_command(
                client,
                feed_type,
                filter_query=filter_query,
                limit=limit,
                tlp_color=tlp_color,
                feed_tags=feed_tags,
                fetch_command=True,
            )
            for iter_ in batch(indicators, batch_size=2000):
                demisto.createIndicators(iter_)

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

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


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