killvxk 执行 IOC 丰富化自动化
适用场景
以下情况使用本技能：
- SOC 分析师在告警分诊期间需要快速从多个来源丰富化 IOC
- 高告警量需要自动化丰富化以减少手动查询时间
- 事件调查需要全面的 IOC 上下文进行范围评估
- SOAR 剧本需要将丰富化操作作为自动化分诊工作流的一部分
不适用于未经分析师审查的批量封锁决策——丰富化提供上下文，而非确定性的恶意/良性判断。
前置条件
- API 密钥：VirusTotal（免费或高级版）、AbuseIPDB、Shodan、URLScan.io、GreyNoise
- Python 3.8+ 含
requests、vt-py、shodan库 - MISP 实例或 TIP（威胁情报平台）用于交叉参考组织情报
- SOAR 平台（可选）用于工作流集成
- 速率限制意识：VT 免费版（每分钟 4 次）、AbuseIPDB（每天 1000 次）、Shodan（每秒 1 次）
工作流程
步骤 1：构建统一丰富化引擎
创建多源丰富化管道：
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=list)
risk_score: float = 0.0
disposition: str = "未知"
class IOCEnrichmentEngine:
def __init__(self, config):
self.vt_client = vt.Client(config["virustotal_key"])
self.shodan_api = shodan.Shodan(config["shodan_key"])
self.abuseipdb_key = config["abuseipdb_key"]
self.greynoise_key = config["greynoise_key"]
self.urlscan_key = config["urlscan_key"]
def enrich_ip(self, ip_address):
result = EnrichmentResult(ioc_value=ip_address, ioc_type="ip")
# VirusTotal
try:
vt_obj = self.vt_client.get_object(f"/ip_addresses/{ip_address}")
result.virustotal = {
"malicious": vt_obj.last_analysis_stats.get("malicious", 0),
"suspicious": vt_obj.last_analysis_stats.get("suspicious", 0),
"total_engines": sum(vt_obj.last_analysis_stats.values()),
"reputation": vt_obj.reputation,
"country": getattr(vt_obj, "country", "未知"),
"as_owner": getattr(vt_obj, "as_owner", "未知")
}
except Exception as e:
result.virustotal = {"error": str(e)}
# AbuseIPDB
try:
response = requests.get(
"https://api.abuseipdb.com/api/v2/check",
headers={"Key": self.abuseipdb_key, "Accept": "application/json"},
params={"ipAddress": ip_address, "maxAgeInDays": 90}
)
data = response.json()["data"]
result.abuseipdb = {
"confidence_score": data["abuseConfidenceScore"],
"total_reports": data["totalReports"],
"is_tor": data.get("isTor", False),
"usage_type": data.get("usageType", "未知"),
"isp": data.get("isp", "未知"),
"domain": data.get("domain", "未知")
}
except Exception as e:
result.abuseipdb = {"error": str(e)}
# Shodan
try:
host = self.shodan_api.host(ip_address)
result.shodan_data = {
"ports": host.get("ports", []),
"os": host.get("os", "未知"),
"organization": host.get("org", "未知"),
"isp": host.get("isp", "未知"),
"vulns": host.get("vulns", []),
"last_update": host.get("last_update", "未知")
}
except shodan.APIError:
result.shodan_data = {"status": "未在 Shodan 中找到"}
# GreyNoise
try:
response = requests.get(
f"https://api.greynoise.io/v3/community/{ip_address}",
headers={"key": self.greynoise_key}
)
gn_data = response.json()
result.greynoise = {
"classification": gn_data.get("classification", "unknown"),
"noise": gn_data.get("noise", False),
"riot": gn_data.get("riot", False),
"name": gn_data.get("name", "未知")
}
except Exception as e:
result.greynoise = {"error": str(e)}
# 计算综合风险评分
result.risk_score = self._calculate_ip_risk(result)
result.disposition = self._determine_disposition(result.risk_score)
return result
def enrich_domain(self, domain):
result = EnrichmentResult(ioc_value=domain, ioc_type="domain")
# VirusTotal
try:
vt_obj = self.vt_client.get_object(f"/domains/{domain}")
result.virustotal = {
"malicious": vt_obj.last_analysis_stats.get("malicious", 0),
"suspicious": vt_obj.last_analysis_stats.get("suspicious", 0),
"reputation": vt_obj.reputation,
"creation_date": getattr(vt_obj, "creation_date", "未知"),
"registrar": getattr(vt_obj, "registrar", "未知"),
"categories": getattr(vt_obj, "categories", {})
}
except Exception as e:
result.virustotal = {"error": str(e)}
# URLScan.io
try:
response = requests.get(
f"https://urlscan.io/api/v1/search/?q=domain:{domain}",
headers={"API-Key": self.urlscan_key}
)
scans = response.json().get("results", [])
result.urlscan = {
"total_scans": len(scans),
"verdicts": [s.get("verdicts", {}).get("overall", {}).get("malicious", False)
for s in scans[:5]],
"last_scan": scans[0]["task"]["time"] if scans else "从未扫描"
}
except Exception as e:
result.urlscan = {"error": str(e)}
result.risk_score = self._calculate_domain_risk(result)
result.disposition = self._determine_disposition(result.risk_score)
return result
def enrich_hash(self, file_hash):
result = EnrichmentResult(ioc_value=file_hash, ioc_type="hash")
# VirusTotal
try:
vt_obj = self.vt_client.get_object(f"/files/{file_hash}")
result.virustotal = {
"malicious": vt_obj.last_analysis_stats.get("malicious", 0),
"suspicious": vt_obj.last_analysis_stats.get("suspicious", 0),
"undetected": vt_obj.last_analysis_stats.get("undetected", 0),
"total_engines": sum(vt_obj.last_analysis_stats.values()),
"type_description": getattr(vt_obj, "type_description", "未知"),
"popular_threat_name": getattr(vt_obj, "popular_threat_classification", {}).get(
"suggested_threat_label", "未知"
),
"sandbox_verdicts": getattr(vt_obj, "sandbox_verdicts", {}),
"first_seen": getattr(vt_obj, "first_submission_date", "未知")
}
except vt.APIError:
result.virustotal = {"status": "未在 VirusTotal 中找到"}
# MalwareBazaar
try:
response = requests.post(
"https://mb-api.abuse.ch/api/v1/",
data={"query": "get_info", "hash": file_hash}
)
mb_data = response.json()
if mb_data["query_status"] == "ok":
entry = mb_data["data"][0]
result.abuseipdb = { # 复用字段存储 MalwareBazaar 数据
"malware_family": entry.get("signature", "未知"),
"tags": entry.get("tags", []),
"file_type": entry.get("file_type", "未知"),
"delivery_method": entry.get("delivery_method", "未知"),
"first_seen": entry.get("first_seen", "未知")
}
except Exception:
pass
result.risk_score = self._calculate_hash_risk(result)
result.disposition = self._determine_disposition(result.risk_score)
return result
def _calculate_ip_risk(self, result):
score = 0
vt = result.virustotal
abuse = result.abuseipdb
gn = result.greynoise
if isinstance(vt, dict) and "malicious" in vt:
score += min(vt["malicious"] * 3, 30)
if isinstance(abuse, dict) and "confidence_score" in abuse:
score += abuse["confidence_score"] * 0.3
if isinstance(gn, dict):
if gn.get("classification") == "malicious":
score += 20
elif gn.get("riot"):
score -= 20 # 已知良性服务
return min(max(score, 0), 100)
def _calculate_domain_risk(self, result):
score = 0
vt = result.virustotal
if isinstance(vt, dict) and "malicious" in vt:
score += min(vt["malicious"] * 4, 40)
if vt.get("reputation", 0) < -5:
score += 20
return min(max(score, 0), 100)
def _calculate_hash_risk(self, result):
score = 0
vt = result.virustotal
if isinstance(vt, dict) and "malicious" in vt:
total = vt.get("total_engines", 1)
detection_rate = vt["malicious"] / total if total > 0 else 0
score = detection_rate * 100
return min(max(score, 0), 100)
def _determine_disposition(self, risk_score):
if risk_score >= 70:
return "恶意——建议封锁"
elif risk_score >= 40:
return "可疑——监控并调查"
elif risk_score >= 10:
return "低风险——可能良性，验证上下文"
else:
return "干净——无恶意活动指标"
def close(self):
self.vt_client.close()
步骤 2：事件调查的批量丰富化
# 处理事件中的多个 IOC
iocs = [
{"type": "ip", "value": "185.234.218.50"},
{"type": "domain", "value": "evil-c2-server.com"},
{"type": "hash", "value": "a1b2c3d4e5f6..."},
{"type": "ip", "value": "45.33.32.156"},
]
config = {
"virustotal_key": "YOUR_VT_KEY",
"shodan_key": "YOUR_SHODAN_KEY",
"abuseipdb_key": "YOUR_ABUSEIPDB_KEY",
"greynoise_key": "YOUR_GREYNOISE_KEY",
"urlscan_key": "YOUR_URLSCAN_KEY"
}
engine = IOCEnrichmentEngine(config)
results = []
for ioc in iocs:
if ioc["type"] == "ip":
result = engine.enrich_ip(ioc["value"])
elif ioc["type"] == "domain":
result = engine.enrich_domain(ioc["value"])
elif ioc["type"] == "hash":
result = engine.enrich_hash(ioc["value"])
results.append(result)
time.sleep(15) # VT 免费 API 速率限制
engine.close()
# 打印摘要
for r in results:
print(f"{r.ioc_type}: {r.ioc_value}")
print(f" 风险评分：{r.risk_score}")
print(f" 处置：{r.disposition}")
print()
步骤 3：与 Splunk 集成实现自动化丰富化
创建 Splunk 自定义搜索命令用于内联丰富化：
index=notable sourcetype="stash"
| table src_ip, dest_ip, file_hash, url
| lookup threat_intel_ip_lookup ip AS src_ip OUTPUT vt_score, abuse_score, disposition
| lookup threat_intel_hash_lookup hash AS file_hash OUTPUT vt_detections, malware_family
| eval combined_risk = coalesce(vt_score, 0) + coalesce(abuse_score, 0)
| where combined_risk > 50
| sort - combined_risk
步骤 4：生成丰富化报告
def generate_enrichment_report(results):
report = []
report.append("IOC 丰富化报告")
report.append("=" * 60)
for r in sorted(results, key=lambda x: x.risk_score, reverse=True):
report.append(f"\n{r.ioc_type.upper()}：{r.ioc_value}")
report.append(f" 风险评分：{r.risk_score}/100")
report.append(f" 处置：{r.disposition}")
if r.virustotal and "malicious" in r.virustotal:
report.append(f" VirusTotal：{r.virustotal['malicious']}/{r.virustotal.get('total_engines', 'N/A')} 恶意")
if r.abuseipdb and "confidence_score" in r.abuseipdb:
report.append(f" AbuseIPDB：{r.abuseipdb['confidence_score']}% 置信度，{r.abuseipdb['total_reports']} 条报告")
if r.greynoise and "classification" in r.greynoise:
report.append(f" GreyNoise：{r.greynoise['classification']}")
if r.shodan_data and "ports" in r.shodan_data:
report.append(f" Shodan：端口 {r.shodan_data['ports']}，组织：{r.shodan_data.get('organization', 'N/A')}")
return "\n".join(report)
核心概念
| 术语 | 定义 |
|---|---|
| IOC 丰富化（IOC Enrichment） | 从多个外部来源向原始指标添加上下文情报的过程 |
| 综合风险评分（Composite Risk Score） | 结合多个情报来源的加权聚合评分，用于处置决策 |
| 速率限制（Rate Limiting） | API 请求限制，需要节流（VT 免费版：每分钟 4 次，AbuseIPDB：每天 1000 次） |
| GreyNoise RIOT | Rule It Out——GreyNoise 已知良性服务数据集，用于减少误报 |
| 被动 DNS（Passive DNS） | 显示域名到 IP 映射历史的 DNS 解析历史数据 |
| 去武装化（Defanging） | 修改 IOC 以在报告中安全处理（evil.com 变为 evil[.]com） |
工具与系统
- VirusTotal：多引擎恶意软件扫描器，提供文件、URL、IP 和域名分析，含 70+ 个 AV 引擎
- AbuseIPDB：社区 IP 信誉数据库，含滥用置信度评分和 ISP 归因
- Shodan：全网扫描器，提供 IP 地址的开放端口、Banner 和漏洞数据
- GreyNoise：互联网噪音情报，区分针对性攻击和机会性扫描
- URLScan.io：URL 分析平台，捕获截图、DOM 和网络请求用于钓鱼检测
常见场景
- 告警分诊丰富化：自动丰富化重大事件中的所有 IP，确定来源是否为已知恶意
- 事件范围评估：批量丰富化受损主机的所有 IOC，识别 C2 基础设施
- 威胁情报验证：丰富化收到的 IOC 信息流，在添加到封锁控制前验证质量
- 钓鱼 URL 分析：在用户通知前，使用 URLScan 和 VT 丰富化举报钓鱼邮件中的 URL
- 误报调查：丰富化被标记的 IP，确定其是否属于 CDN/云提供商（合法的）
输出格式
IOC 丰富化报告 — IR-2024-0450
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
丰富化时间：2024-03-15 14:30 UTC
处理 IOC 数：4
IP：185.234.218[.]50
风险评分： 87/100——恶意
VirusTotal：14/90 个引擎标记为恶意
AbuseIPDB： 92% 置信度，347 条报告
Shodan： 端口 [22, 80, 443, 4444]，组织：防弹主机
GreyNoise： 恶意——已知 C2 基础设施
操作： 立即封锁
域名：evil-c2-server[.]com
风险评分： 73/100——恶意
VirusTotal：8/90 个引擎标记
URLScan： 5 次扫描，4 个恶意判定
WHOIS： 3 天前通过 Namecheap 注册
操作： 封锁并添加到 DNS 黑洞
哈希：a1b2c3d4e5f6...
风险评分： 91/100——恶意
VirusTotal：52/72 个引擎（Cobalt Strike Beacon）
MalwareBazaar：标签：cobalt-strike、beacon、c2
操作： 封锁哈希，隔离受影响终端
IP：45.33.32[.]156
风险评分： 5/100——干净
VirusTotal：0/90 个引擎
GreyNoise： 良性——Shodan 扫描器
操作： 无需操作（已知扫描器）