| name | performing-ioc-enrichment-automation |
| description | Automates Indicator of Compromise (IOC) enrichment by orchestrating lookups across VirusTotal, AbuseIPDB, Shodan, MISP, and other intelligence sources to provide contextual scoring and disposition recommendations. Use when SOC analysts need rapid multi-source enrichment of IPs, domains, URLs, and file hashes during alert triage or incident investigation.
|
| domain | cybersecurity |
| subdomain | soc-operations |
| tags | ["soc","ioc","enrichment","automation","virustotal","abuseipdb","shodan","threat-intelligence"] |
| version | 1.0 |
| author | mahipal |
| license | Apache-2.0 |
| nist_csf | ["DE.CM-01","DE.AE-02","RS.MA-01","DE.AE-06"] |
| mitre_attack | ["T1078","T1685.002","T1685.005","T1566"] |
Performing IOC Enrichment Automation
When to Use
Use this skill when:
- SOC analysts need to quickly enrich IOCs from multiple sources during alert triage
- High alert volumes require automated enrichment to reduce manual lookup time
- Incident investigations need comprehensive IOC context for scope assessment
- SOAR playbooks require enrichment actions as part of automated triage workflows
Do not use for bulk blocking decisions without analyst review — enrichment provides context, not definitive malicious/benign determination.
Prerequisites
- API keys: VirusTotal (free or premium), AbuseIPDB, Shodan, URLScan.io, GreyNoise
- Python 3.8+ with
requests, vt-py, shodan libraries
- MISP instance or TIP for cross-referencing organizational intelligence
- SOAR platform (optional) for workflow integration
- Rate limit awareness: VT free (4 req/min), AbuseIPDB (1000/day), Shodan (1 req/sec)
Workflow
Step 1: Build Unified Enrichment Engine
Create a multi-source enrichment pipeline:
import requests
import vt
import shodan
import time
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class EnrichmentResult:
ioc_value: str
ioc_type: str
virustotal: dict = field(default_factory=dict)
abuseipdb: dict = field(default_factory=dict)
shodan_data: dict = field(default_factory=dict)
greynoise: dict = field(default_factory=dict)
urlscan: dict = field(default_factory=dict)
misp_matches: list = field(default_factory=)
risk_score: =
disposition: =
:
():
.vt_client = vt.Client(config[])
.shodan_api = shodan.Shodan(config[])
.abuseipdb_key = config[]
.greynoise_key = config[]
.urlscan_key = config[]
():
result = EnrichmentResult(ioc_value=ip_address, ioc_type=)
:
vt_obj = .vt_client.get_object()
result.virustotal = {
: vt_obj.last_analysis_stats.get(, ),
: vt_obj.last_analysis_stats.get(, ),
: (vt_obj.last_analysis_stats.values()),
: vt_obj.reputation,
: (vt_obj, , ),
: (vt_obj, , )
}
Exception e:
result.virustotal = {: (e)}
:
response = requests.get(
,
headers={: .abuseipdb_key, : },
params={: ip_address, : }
)
data = response.json()[]
result.abuseipdb = {
: data[],
: data[],
: data.get(, ),
: data.get(, ),
: data.get(, ),
: data.get(, )
}
Exception e:
result.abuseipdb = {: (e)}
:
host = .shodan_api.host(ip_address)
result.shodan_data = {
: host.get(, []),
: host.get(, ),
: host.get(, ),
: host.get(, ),
: host.get(, []),
: host.get(, )
}
shodan.APIError:
result.shodan_data = {: }
:
response = requests.get(
,
headers={: .greynoise_key}
)
gn_data = response.json()
result.greynoise = {
: gn_data.get(, ),
: gn_data.get(, ),
: gn_data.get(, ),
: gn_data.get(, )
}
Exception e:
result.greynoise = {: (e)}
result.risk_score = ._calculate_ip_risk(result)
result.disposition = ._determine_disposition(result.risk_score)
result
():
result = EnrichmentResult(ioc_value=domain, ioc_type=)
:
vt_obj = .vt_client.get_object()
result.virustotal = {
: vt_obj.last_analysis_stats.get(, ),
: vt_obj.last_analysis_stats.get(, ),
: vt_obj.reputation,
: (vt_obj, , ),
: (vt_obj, , ),
: (vt_obj, , {})
}
Exception e:
result.virustotal = {: (e)}
:
response = requests.get(
,
headers={: .urlscan_key}
)
scans = response.json().get(, [])
result.urlscan = {
: (scans),
: [s.get(, {}).get(, {}).get(, )
s scans[:]],
: scans[][][] scans
}
Exception e:
result.urlscan = {: (e)}
result.risk_score = ._calculate_domain_risk(result)
result.disposition = ._determine_disposition(result.risk_score)
result
():
result = EnrichmentResult(ioc_value=file_hash, ioc_type=)
:
vt_obj = .vt_client.get_object()
result.virustotal = {
: vt_obj.last_analysis_stats.get(, ),
: vt_obj.last_analysis_stats.get(, ),
: vt_obj.last_analysis_stats.get(, ),
: (vt_obj.last_analysis_stats.values()),
: (vt_obj, , ),
: (vt_obj, , {}).get(
,
),
: (vt_obj, , {}),
: (vt_obj, , )
}
vt.APIError:
result.virustotal = {: }
:
response = requests.post(
,
data={: , : file_hash}
)
mb_data = response.json()
mb_data[] == :
entry = mb_data[][]
result.abuseipdb = {
: entry.get(, ),
: entry.get(, []),
: entry.get(, ),
: entry.get(, ),
: entry.get(, )
}
Exception:
result.risk_score = ._calculate_hash_risk(result)
result.disposition = ._determine_disposition(result.risk_score)
result
():
score =
vt = result.virustotal
abuse = result.abuseipdb
gn = result.greynoise
(vt, ) vt:
score += (vt[] * , )
(abuse, ) abuse:
score += abuse[] *
(gn, ):
gn.get() == :
score +=
gn.get():
score -=
((score, ), )
():
score =
vt = result.virustotal
(vt, ) vt:
score += (vt[] * , )
vt.get(, ) < -:
score +=
((score, ), )
():
score =
vt = result.virustotal
(vt, ) vt:
total = vt.get(, )
detection_rate = vt[] / total total >
score = detection_rate *
((score, ), )
():
risk_score >= :
risk_score >= :
risk_score >= :
:
():
.vt_client.close()
