Google Threat Intelligence IoC Stream Feed

Use this feed integration to fetch Google Threat Intelligence IoC Stream notifications as indicators.

Data Enrichment & Threat Intelligence · GoogleThreatIntelligence · Feed

Details

ID	`Google Threat Intelligence IoC Stream Feed`
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 the Google Threat Intelligence IoC Stream Feed integration to fetch indicators from IoC Stream rules or rulesets.

Configure Google Threat Intelligence IoC Stream Feed on Cortex XSOAR

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

Parameter	Description	Required
feed	The fetch indicators.	False
credentials	API Key.	True
limit	The maximum number of indicators to fetch from IoC Stream. The default value is 200.	False
filter	Exact name of the rule or ruleset you want to filter on. Leave empty to receive all.	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
feedMinimumGTIScore	The minimum GTI score to import as part of the feed.	True
feedBypassExclusionList	Whether to bypass exclusion list.	False

Click Test to validate the Google Threat Intelligence API Key, and connection.

IoC Stream Feed info

By default the IoC Stream Feed retrieve all indicators on IoC Stream. You have the option to get files, domains, IP addresses or URLs only from LiveHunt, RetroHunt, Collections, Threat Actors, etc., using the filter parameter.

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.

Get Indicators

Gets the indicators from Google Threat Intelligence IoC Stream.

Base Command

gti-iocstream-get-indicators

Input

Argument Name	Description	Required
limit	The maximum number of results to return. The default value is 200.	Optional
filter	Filter your IoC Stream (e.g., “source_type:hunting_ruleset” for LiveHunt, “source_type:retrohunt_job” for RetroHunt). Leave empty to receive all.	Optional

Context Output

There is no context output for this command.

Command Example

!gti-iocstream-get-indicators limit=1 filter=source_type:hunting_ruleset

Human Readable Output

Indicators from Google Threat Intelligence IoC Stream

Id	Detections	Origin	Sources	Gti Threat Score	Gti Severity	Gti Verdict	Malware Families	Threat Actors
f221425286c9073cbb2168f73120b6…	59/69	hunting	[hunting_ruleset] YARA ruleset	80	SEVERITY_LOW	VERDICT_MALICIOUS	beacon	SWEED

Configuration parameters

credentials — API Key (leave empty. Fill in the API key in the password field.) (required)
filter — Filter
feed — Fetch indicators
limit — Limit
feedReputation — Indicator Reputation
feedReliability — Source Reliability (required)
feedExpirationPolicy —
feedExpirationInterval —
feedFetchInterval — Feed Fetch Interval
feedMinimumGTIScore — Feed Minimum Google Threat Intelligence Score (required)
feedBypassExclusionList — Bypass exclusion list
feedTags — Tags
tlp_color — Traffic Light Protocol Color

Commands (1)

gti-iocstream-get-indicators
Gets the indicators from Google Threat Intelligence IoC Stream.

Source

FeedIoCStream.py FeedIoCStream_test.py FeedIoCStream.yml FeedIoCStream_description.md command_examples.txt

Download History Raw

"""Tests for Google Threat Intelligence IoC Stream Feed integration."""

import json
from unittest import mock

import demistomock as demisto  # noqa: F401
from CommonServerPython import FeedIndicatorType  # noqa: F401
from FeedIoCStream import Client, fetch_indicators_command, get_indicators_command, main


def _mock_indicator(indicator_type, gti_score=None):
    """Mocks indicator."""
    with open(f"./test_data/{indicator_type}.json", encoding="utf-8") as f:
        indicator_mock = json.load(f)

    if gti_score is not None:
        indicator_mock["attributes"]["gti_assessment"]["threat_score"]["value"] = gti_score

    return indicator_mock


def _mock_file(gti_score=None):
    """Mocks file."""
    return _mock_indicator("file", gti_score)


def _mock_domain(gti_score=None):
    """Mocks domain."""
    return _mock_indicator("domain", gti_score)


def _mock_url(gti_score=None):
    """Mocks URL."""
    return _mock_indicator("url", gti_score)


def _mock_ip(gti_score=None):
    """Mocks IP address."""
    return _mock_indicator("ip", gti_score)


def test_fetch_indicators_command(mocker):
    """Tests fetch indicators command."""
    client = Client("https://fake")
    for gti_score, len_response in [
        (0, 4),
        (1, 3),
        (50, 2),
        (95, 1),
        (100, 0),
    ]:
        mocker.patch.object(
            client,
            "get_api_indicators",
            return_value=[
                _mock_file(),
                _mock_domain(),
                _mock_url(),
                _mock_ip(),
            ],
        )

        indicators = fetch_indicators_command(client, None, [], 10, None, gti_score)

        assert len(indicators) == len_response

        for indicator in indicators:
            if indicator["type"] == FeedIndicatorType.File:
                assert set(indicator["fields"].keys()) == {
                    "md5",
                    "sha1",
                    "sha256",
                    "ssdeep",
                    "fileextension",
                    "filetype",
                    "imphash",
                    "tags",
                    "firstseenbysource",
                    "lastseenbysource",
                    "creationdate",
                    "updateddate",
                    "detectionengines",
                    "positivedetections",
                    "displayname",
                    "name",
                    "size",
                    "gtithreatscore",
                    "gtiseverity",
                    "gtiverdict",
                    "actor",
                    "malwarefamily",
                }
                assert indicator["value"] == "<sha256>"
                assert indicator["value"] == indicator["fields"]["sha256"]
                assert indicator["origin"] == "hunting"
                assert indicator["sources"] == "[hunting_ruleset] Malware Families YARA ruleset"
                assert indicator["fields"]["gtiverdict"] == "VERDICT_MALICIOUS"
                assert indicator["score"] == 3
            elif indicator["type"] == FeedIndicatorType.Domain:
                assert set(indicator["fields"].keys()) == {
                    "admincountry",
                    "adminname",
                    "adminemail",
                    "adminphone",
                    "registrantcountry",
                    "registrantemail",
                    "registrantname",
                    "registrantphone",
                    "registrarabusephone",
                    "registrarabuseemail",
                    "registrarname",
                    "firstseenbysource",
                    "lastseenbysource",
                    "tags",
                    "creationdate",
                    "updateddate",
                    "detectionengines",
                    "positivedetections",
                    "gtithreatscore",
                    "gtiseverity",
                    "gtiverdict",
                    "actor",
                    "malwarefamily",
                }
                assert indicator["value"] == "<domain>"
                assert indicator["fields"]["adminemail"] == "<admin_email>@google.com"
                assert indicator["fields"]["registrantcountry"] == "US"
                assert indicator["fields"]["registrarabusephone"] == "+34 600 000 000"
                assert indicator["fields"]["gtiverdict"] == "VERDICT_MALICIOUS"
                assert indicator["score"] == 3
            elif indicator["type"] == FeedIndicatorType.URL:
                assert set(indicator["fields"].keys()) == {
                    "tags",
                    "firstseenbysource",
                    "lastseenbysource",
                    "updateddate",
                    "detectionengines",
                    "positivedetections",
                    "gtithreatscore",
                    "gtiseverity",
                    "gtiverdict",
                    "actor",
                    "malwarefamily",
                }
                assert indicator["value"] == "<url>"
                assert indicator["fields"]["firstseenbysource"] == 1722360511
                assert indicator["fields"]["gtiverdict"] == "VERDICT_UNDETECTED"
                assert indicator["score"] == 0
            elif indicator["type"] == FeedIndicatorType.IP:
                assert set(indicator["fields"].keys()) == {
                    "tags",
                    "firstseenbysource",
                    "lastseenbysource",
                    "updateddate",
                    "detectionengines",
                    "positivedetections",
                    "countrycode",
                    "gtithreatscore",
                    "gtiseverity",
                    "gtiverdict",
                    "actor",
                    "malwarefamily",
                }
                assert indicator["value"] == "X.X.X.X"
                assert indicator["fields"]["countrycode"] == "US"
                assert indicator["fields"]["gtiverdict"] == "VERDICT_BENIGN"
                assert indicator["score"] == 1
            else:
                raise ValueError(f'Unknown type: {indicator["type"]}')


def test_get_indicators_command(mocker):
    """Tests get indicators command."""
    client = Client("https://fake")

    for gti_score, len_response in [
        (None, 2),
        (0, 4),
        (1, 3),
        (50, 2),
        (95, 1),
        (100, 0),
    ]:
        mocker.patch.object(
            client,
            "get_api_indicators",
            return_value=[
                _mock_file(),
                _mock_domain(),
                _mock_url(),
                _mock_ip(),
            ],
        )
        params = {
            "tlp_color": None,
            "feedTags": [],
        }
        if gti_score is not None:
            params["feedMinimumGTIScore"] = gti_score

        result = get_indicators_command(client, params, {})

        assert len(result.raw_response) == len_response


def test_main_manual_command(mocker):
    """Tests main manual."""
    params = {
        "tlp_color": None,
        "feedTags": [],
        "credentials": {"password": "xxx"},
        "feedMinimumGTIScore": 95,
    }

    args = {
        "limit": 7,
        "filter": "entity_type:file",
    }

    mocker.patch.object(demisto, "params", return_value=params)
    mocker.patch.object(demisto, "command", return_value="gti-iocstream-get-indicators")
    mocker.patch.object(demisto, "args", return_value=args)
    get_api_indicators_mock = mocker.patch.object(
        Client,
        "get_api_indicators",
        return_value=[
            _mock_file(),
            _mock_domain(),
            _mock_url(),
            _mock_ip(),
        ],
    )
    return_results_mock = mocker.patch.object(demisto, "results")

    main()

    assert get_api_indicators_mock.call_args == mock.call("entity_type:file", 7)
    assert len(return_results_mock.call_args[0][0]["Contents"]) == 1


def test_main_default_command(mocker):
    """Tests main default."""
    params = {
        "tlp_color": None,
        "feedTags": [],
        "credentials": {"password": "xxx"},
        "limit": 7,
        "filter": "entity_type:file",
        "feedMinimumGTIScore": 1,
    }

    mocker.patch.object(demisto, "params", return_value=params)
    mocker.patch.object(demisto, "command", return_value="fetch-indicators")
    get_api_indicators_mock = mocker.patch.object(
        Client,
        "get_api_indicators",
        return_value=[
            _mock_file(),
            _mock_domain(),
            _mock_url(),
            _mock_ip(),
        ],
    )
    create_indicators_mock = mocker.patch.object(demisto, "createIndicators")

    main()

    assert get_api_indicators_mock.call_args == mock.call("entity_type:file", 7)
    assert len(create_indicators_mock.call_args[0][0]) == 3


def test_main_test_command(mocker):
    """Tests main test."""
    params = {"credentials": {"password": "xxx"}}

    mocker.patch.object(demisto, "params", return_value=params)
    mocker.patch.object(demisto, "command", return_value="test-module")
    get_api_indicators_mock = mocker.patch.object(
        Client,
        "get_api_indicators",
        return_value=[_mock_file()],
    )

    main()

    assert get_api_indicators_mock.call_count == 1


def test_get_api_indicators(mocker):
    """Tests get_api_indicators."""
    http_mock = mocker.patch.object(
        Client,
        "_http_request",
        return_value={
            "data": list(range(10)),
        },
    )

    client = Client("https://fake")
    iocs = client.get_api_indicators(limit=50)
    assert http_mock.call_count == 1
    assert http_mock.call_args_list[0].kwargs["params"]["limit"] == 40
    assert iocs == list(range(10))


def test_get_api_indicators_with_cursor(mocker):
    """Tests get_api_indicators with cursor."""
    first_execution = True

    def side_effect(*args, **kwargs):
        nonlocal first_execution
        if first_execution:
            first_execution = False
            return {
                "data": list(range(40)),
                "meta": {
                    "cursor": "random",
                },
            }
        return {
            "data": list(range(40, 50)),
        }

    http_mock = mocker.patch.object(Client, "_http_request", side_effect=side_effect)

    client = Client("https://fake")
    iocs = client.get_api_indicators(limit=50)
    assert http_mock.call_count == 2
    assert http_mock.call_args_list[0].kwargs["params"]["limit"] == 40
    assert http_mock.call_args_list[1].kwargs["params"]["limit"] == 10
    assert iocs == list(range(50))