Skip to main content

How to Automate a Threat Intelligence Feedback Loop: Google SecOps to Google Threat Intelligence (GTI)

  • September 22, 2026
  • 0 replies
  • 45 views

ragaom
Staff
Forum|alt.badge.img

Overview: Breaking the One-Way Intelligence Trap

 

In modern Security Operations Centers (SOCs), threat intelligence integrations typically operate as a one-way funnel. External indicators and intelligence feeds are ingested into SIEM and SOAR platforms to enrich telemetry, score alerts, and provide context during triage.

However, during active investigations, Tier 2 and Tier 3 analysts uncover novel, high-fidelity attacker infrastructure—such as zero-day command-and-control (C2) IPs, adversary staging domains, and customized malware payloads. In most organizations, these verified indicators remain trapped in the resolution notes of a closed incident ticket.

This guide demonstrates how to establish an automated, bidirectional intelligence feedback loop between Google Security Operations (SecOps SOAR) and Google Threat Intelligence (GTI Enterprise+ / VirusTotal API v3). When an investigation is confirmed as a true positive, SecOps automatically pushes sanitized indicators to a dedicated, private Custom Threat Actor or Custom Campaign profile in GTI—immediately activating automated hunting and enterprise-wide correlation.

 

Architectural Pillars

 

+-------------------------------------------------------------------------+
| Google SecOps SOAR |
| [Incident Resolved / True Positive] |
| │ |
| ▼ |
| [Custom Python Integration Action] |
| ├─ 1. Topology Sanitization (Strip RFC 1918 IPs & .local domains) |
| ├─ 2. Secure Auth (Load API_Key from Integration Config) |
| └─ 3. Query / Resolve GTI Custom Target Collection |
+------------------------------------┬------------------------------------+

│ HTTPS REST (VT / GTI API v3)

+-------------------------------------------------------------------------+
| Google Threat Intelligence (GTI) |
| [Custom Threat Actor / Campaign Collection ("private": true)] |
| ├─ POST /collections/{id}/files |
| ├─ POST /collections/{id}/ip_addresses |
| ├─ POST /collections/{id}/domains |
| └─ POST /collections/{id}/urls |
+------------------------------------┬------------------------------------+

┌──────────────────────────┼──────────────────────────┐
▼ ▼ ▼
[ GTI LiveHunt ] [ GTI RetroHunt ] [ SecOps ATI Sync ]
Real-time YARA matching 12-month historical sweep Closed-loop re-ingestion
on incoming file stream for prior sightings for automatic attribution

 

1. Automated Adversary & Campaign Attribution

When an analyst confirms an incident or marks a case as a True Positive, the custom SecOps action extracts case entities (FILEHASH, IP, DOMAIN, URL) and associates them with a designated Threat Actor or Campaign collection in GTI.

2. Strict Tenant Privacy Boundaries ("private": true)

Enterprise investigations frequently involve confidential target data and sensitive infrastructure. By provisioning the collection with "private": true, indicators remain strictly restricted to your licensed GTI Enterprise+ tenant. They are never shared with the public VirusTotal community, preserving strict TLP:AMBER and TLP:RED data governance.

3. Pre-Flight Topology Sanitization

Before externalizing telemetry, the automation filters out non-routable internal artifacts:

  • IPv4 Private Ranges: 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16, loopbacks (127.0.0.1), and link-local addresses.
  • Internal Domains: .local, .internal, .corp, and custom enterprise domain suffixes.

4. Direct GTI v3 Schema Routing

The action uses the official GTI v3 REST API, posting indicators to relationship-specific endpoints:

  • POST /api/v3/collections/{id}/files (SHA-256, MD5)
  • POST /api/v3/collections/{id}/ip_addresses
  • POST /api/v3/collections/{id}/domains
  • POST /api/v3/collections/{id}/urls

(The script includes an automatic fallback to atomic PATCH /api/v3/collections/{id} utilizing raw_items for bulk loads).

5. Downstream Multiplier Effect

Once associated with the collection in GTI, indicators automatically activate three enterprise defensive capabilities:

  1. GTI LiveHunt (YARA): Automatically evaluates new global and private file telemetry against your actor/campaign indicators and rule sets in real time.
  2. GTI RetroHunt: Executes retrospective sweeps across up to 12 months of historical telemetry to uncover prior, unnoticed adversary dwell time.
  3. SecOps ATI Closed-Loop: Synchronized private collections are re-ingested into Google SecOps via Applied Threat Intelligence (ATI), ensuring future alerts trigger automated attribution.

Extending the Concept: Custom Campaigns & Multi-Tier Attribution

 

Does this feedback pattern also support Custom Campaigns?

Yes, natively. Under the hood of Google Threat Intelligence (VirusTotal Enterprise v3 / STIX 2.1), both Threat Actors and Campaigns are polymorphic collections managed by the exact same /api/v3/collections infrastructure.

Operational Difference: Why Campaigns Shine in the SOC

In active incident triage, attributing activity to a definitive threat actor (e.g., APT29, UNC3886) often requires weeks of forensic intelligence. In contrast, Campaigns are operational and time-bound:

  • Campaign Focus: Allows the SOC to cluster related indicators around an immediate intrusion wave (e.g., "Q1-2026-Spearphishing-Finance" or "Ransomware-Staging-Wave-1") on Day 1.
  • Telemetry & Tracking: In the GTI portal, Campaigns provide timeline views, tracking first-seen and last-seen infrastructure across the attack lifecycle.

The Hierarchical "Power Move": Linking Campaigns to Actors

When both the campaign and adversary are known, GTI allows SecOps to link them hierarchically via relationship endpoints:

POST /api/v3/collections/{campaign_id}/relationships/threat_actors

{

  "data": [

    { "type": "threat-actor", "id": "threat-actor--aba33afa..." }

  ]

}

This gives analysts complete situational lineage:

[Threat Actor: Persistent Group]

       │

       └── executes ──► [Campaign: Q1-2026-Spearphish]

                                │

                                ├──► C2 IP: 198.51.100.24

                                ├──► Dropper: a4b2c1...

                                └──► Phish URL: https://...

 

SecOps SOAR Configuration Best Practice

Storing Credentials via Integration Configuration Parameters

Avoid hardcoding API keys or mapping Shared Credentials across individual playbook blocks. In Google SecOps SOAR:

  1. Navigate to ResponseIDE → Select or Create Google Threat Intelligence Sync.
  2. Under Integration Parameters, add:
    • Identifier: API_Key (or GTI_API_Key)
    • Type: Password (masks key in UI and encrypted in backend storage)
  3. The script accesses this parameter programmatically via: 
api_key = siemplify.extract_configuration_param("GoogleThreatIntelligence", "API_Key")

This decouples secret management from individual playbook steps and prevents credential leakage across playbooks.

 

Production Python Action Script

Deploy this script as a Custom Integration Action in the SecOps SOAR IDE (ResponseIDEIntegration Actions). It seamlessly handles both Threat Actors and Campaigns.

# Copyright 2026 Google LLC.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from SiemplifyAction import SiemplifyAction
from SiemplifyUtils import output_handler
import requests, ipaddress, urllib.parse

GTI_BASE_URL = "https://www.virustotal.com/api/v3"

def is_internal_ip(ip_str):
try:
ip = ipaddress.ip_address(ip_str.strip())
return ip.is_private or ip.is_loopback or ip.is_link_local
except ValueError:
return True

def extract_api_key(siemplify):
# 1. Integration Configuration Parameter (Enterprise Best Practice)
for prov in ["GoogleThreatIntelligence", "Google Threat Intelligence Sync", "GTI"]:
for param in ["API_Key", "GTI_API_Key"]:
try:
k = siemplify.extract_configuration_param(prov, param)
if k: return k.strip()
except Exception: pass
# 2. Action Parameter Fallback
for p in ["GTI_API_Key", "API_Key"]:
try:
k = siemplify.extract_action_param(param_name=p, print_value=False)
if k: return k.strip()
except Exception: pass
raise ValueError("GTI API Key not found in Integration Configuration Parameters.")

def get_or_create_collection(session, entity_name, entity_type, case_id):
filt = urllib.parse.quote(f'name:"{entity_name}"')
resp = session.get(f"{GTI_BASE_URL}/collections?filter={filt}", timeout=15)
if resp.status_code == 200:
for item in resp.json().get("data", []):
if item.get("attributes", {}).get("name", "").lower() == entity_name.lower():
return item["id"]
tag_type = "campaign" if "campaign" in entity_type.lower() else "custom-threat-actor"
payload = {"data": {"type": "collection", "attributes": {
"name": entity_name,
"description": f"Custom {entity_type} from SecOps Case #{case_id}",
"private": True, "tags": [tag_type, "secops-feedback", "tlp-amber"]
}}}
r = session.post(f"{GTI_BASE_URL}/collections", json=payload, timeout=15)
if r.status_code in [200, 201]: return r.json()["data"]["id"]
raise RuntimeError(f"Collection creation failed: {r.text}")

def add_indicators_to_collection(session, collection_id, ioc_groups, raw_items):
errors = []
for rel, items in ioc_groups.items():
if not items: continue
data = [{"type": "url", "url": i["id"]} if i["type"] == "url" else i for i in items]
r = session.post(f"{GTI_BASE_URL}/collections/{collection_id}/{rel}", json={"data": data}, timeout=20)
if r.status_code not in [200, 201, 204]: errors.append(f"{rel} ({r.status_code})")
if errors and raw_items: # Atomic PATCH fallback for bulk indicators
patch = {"data": {"type": "collection", "attributes": {"raw_items": ", ".join(raw_items)}}}
pr = session.patch(f"{GTI_BASE_URL}/collections/{collection_id}", json=patch, timeout=20)
if pr.status_code not in [200, 204]: raise RuntimeError(f"Sync failed: {errors}, fallback: {pr.text}")

@output_handler
def main():
siemplify = SiemplifyAction()
api_key = extract_api_key(siemplify)

# Supports both Threat Actor Name or Campaign Name
target_name = (
siemplify.extract_action_param("Target Name", print_value=True) or
siemplify.extract_action_param("Custom Threat Actor Name", print_value=True) or
siemplify.extract_action_param("Custom Campaign Name", print_value=True) or ""
).strip()
target_type = siemplify.extract_action_param("Target Type", default_value="Threat Actor", print_value=True).strip()

if not target_name:
siemplify.end("Target Name parameter is empty.", False); return

session = requests.Session()
session.headers.update({"x-apikey": api_key, "Accept": "application/json", "Content-Type": "application/json"})
ioc_groups = {"files": [], "ip_addresses": [], "domains": [], "urls": []}
raw_items = []

for e in getattr(siemplify, "target_entities", []):
t, val = getattr(e, "entity_type", ""), getattr(e, "identifier", "").strip()
if not val or val in raw_items: continue
if t in ["FILEHASH", "MD5", "SHA1", "SHA256"]:
ioc_groups["files"].append({"type": "file", "id": val.lower()}); raw_items.append(val.lower())
elif t in ["ADDRESS", "IP"] and not is_internal_ip(val):
ioc_groups["ip_addresses"].append({"type": "ip_address", "id": val}); raw_items.append(val)
elif t in ["HOSTNAME", "DOMAIN"] and not val.endswith((".local", ".internal", ".corp")):
ioc_groups["domains"].append({"type": "domain", "id": val.lower()}); raw_items.append(val.lower())
elif t == "URL":
ioc_groups["urls"].append({"type": "url", "id": val}); raw_items.append(val)

if not raw_items:
siemplify.end("No external IOCs found to synchronize.", False); return

col_id = get_or_create_collection(session, target_name, target_type, getattr(siemplify, "case_id", "Manual"))
add_indicators_to_collection(session, col_id, ioc_groups, raw_items)
portal_url = f"https://www.virustotal.com/gui/collection/{col_id}"
msg = f"Synced {len(raw_items)} IOCs to {target_type} '{target_name}'. GTI URL: {portal_url}"
siemplify.add_entity_comment(msg)
siemplify.end(msg, True)

if __name__ == "__main__":
main()

 

Playbook Workflow Integration

To automate this within your operational playbooks:

  1. Trigger Condition: Configure the playbook to execute either:
    • On Case Close: When Case Status == Closed AND Root Cause == True Positive.
    • On-Demand Analyst Action: As a manual action button on the case wall for instant escalation.
  2. Action Block Settings:
    • Action: Sync IOCs to GTI Target Collection
    • Parameter Target Name: Dynamically populate using [Alert.ThreatActor], [Alert.Campaign], or provide a prompt input for the analyst.
    • Parameter Target Type: Select Threat Actor or Campaign.
  3. Execution Verification: Review case activity stream for the generated confirmation message and direct link to the private GTI collection portal.

I hope this helps to create better feedback loops within your SOC!

Happy Hunting!

 

P.S.: Github repo coming soon