# Performing Ioc Enrichment Automation > 通过编排 VirusTotal、AbuseIPDB、Shodan、MISP 和其他情报源的查询， 自动化入侵指标（IOC）丰富化，提供上下文评分和处置建议。 适用于 SOC 分析师在告警分诊或事件调查期间需要对 IP、域名、URL 和文件哈希 进行快速多源丰富化时。 - Skill: `killvxk/performing-ioc-enrichment-automation` (Agent Skill, multi-file: 4 files) - Install (CLI): `npx skillmds@latest add killvxk/performing-ioc-enrichment-automation` - Raw SKILL.md: https://api.skillmd.com/api/skills/killvxk/performing-ioc-enrichment-automation/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: Apache-2.0 - Author: killvxk (https://skillmd.com/u/killvxk) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/killvxk/performing-ioc-enrichment-automation --- # 执行 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：构建统一丰富化引擎 创建多源丰富化管道： ```python 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：事件调查的批量丰富化 ```python # 处理事件中的多个 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 自定义搜索命令用于内联丰富化： ```spl 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：生成丰富化报告 ```python 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 扫描器 操作： 无需操作（已知扫描器） ```