killvxk 使用 VirusTotal 执行恶意软件哈希富化
概述
VirusTotal 是全球最大的众包恶意软件语料库，使用 70 多个杀毒引擎扫描文件，并提供行为分析、YARA 规则匹配、网络指标和社区情报。本技能涵盖使用 VirusTotal API v3 富化文件哈希（MD5、SHA-1、SHA-256），获取检测判决、沙箱报告、相关指标和上下文情报，用于 SOC 分类、事件响应和威胁情报富化工作流。
前置条件
- Python 3.9+，安装
vt-py（官方 VirusTotal Python 客户端）或requests - VirusTotal API 密钥（免费版：4 次请求/分钟，500 次/天；高级版支持更高限额）
- 了解文件哈希类型：MD5、SHA-1、SHA-256
- 熟悉杀毒软件检测命名规范
- 了解用于 IOC 表示的 STIX 2.1
核心概念
VirusTotal API v3
该 API 提供用于文件报告（/files/{hash}）、URL 扫描、域名报告、IP 地址情报的 RESTful 端点，以及通过 VirusTotal Intelligence（VTI）进行高级猎捕。每份文件报告包含：70 多个杀毒引擎的检测结果、沙箱行为分析、YARA 规则匹配、Sigma 规则匹配、文件元数据（PE 头、导入表、节）、网络指标（联系的 IP、域名、URL），以及社区投票和评论。
哈希富化工作流
典型富化流程：从告警/EDR 接收哈希 -> 查询 VT API -> 解析检测率 -> 提取行为指标 -> 与现有情报关联 -> 做出分类决策。API 返回 last_analysis_stats 对象，包含 malicious（恶意）、suspicious（可疑）、undetected（未检测到）和 harmless（无害）的计数。
从哈希进行关联分析
VirusTotal 支持从单个哈希关联到相关情报：相似文件（ITW/野外样本）、联系的域名和 IP（C2 基础设施）、释放的文件、嵌入的 URL、YARA 规则匹配，以及通过众包情报的威胁行为者归因。
操作步骤
步骤 1：查询 VirusTotal 哈希报告
import vt
import json
import hashlib
from datetime import datetime
class VTEnricher:
def __init__(self, api_key):
self.client = vt.Client(api_key)
def enrich_hash(self, file_hash):
"""使用 VirusTotal 情报富化文件哈希。"""
try:
file_obj = self.client.get_object(f"/files/{file_hash}")
stats = file_obj.last_analysis_stats
report = {
"hash": file_hash,
"sha256": file_obj.sha256,
"sha1": file_obj.sha1,
"md5": file_obj.md5,
"file_type": getattr(file_obj, "type_description", "未知"),
"file_size": getattr(file_obj, "size", 0),
"first_submission": str(getattr(file_obj, "first_submission_date", "")),
"last_analysis_date": str(getattr(file_obj, "last_analysis_date", "")),
"detection_stats": {
"malicious": stats.get("malicious", 0),
"suspicious": stats.get("suspicious", 0),
"undetected": stats.get("undetected", 0),
"harmless": stats.get("harmless", 0),
},
"detection_ratio": f"{stats.get('malicious', 0)}/{sum(stats.values())}",
"popular_threat_names": getattr(file_obj, "popular_threat_classification", {}),
"tags": getattr(file_obj, "tags", []),
"names": getattr(file_obj, "names", []),
}
total_engines = sum(stats.values())
mal_count = stats.get("malicious", 0)
report["threat_level"] = (
"critical" if mal_count > total_engines * 0.7
else "high" if mal_count > total_engines * 0.4
else "medium" if mal_count > total_engines * 0.1
else "low" if mal_count > 0
else "clean"
)
print(f"[+] {file_hash[:16]}... -> {report['detection_ratio']} "
f"（{report['threat_level'].upper()}）")
return report
except vt.error.APIError as e:
print(f"[-] {file_hash} 的 VT API 错误：{e}")
return None
def get_behavior_report(self, file_hash):
"""获取文件的沙箱行为分析。"""
try:
behaviors = self.client.get_object(f"/files/{file_hash}/behaviours")
behavior_data = {
"processes_created": [],
"files_written": [],
"registry_keys_set": [],
"dns_lookups": [],
"http_conversations": [],
"mutexes_created": [],
"commands_executed": [],
}
for sandbox in getattr(behaviors, "data", []):
attrs = sandbox.get("attributes", {})
behavior_data["processes_created"].extend(
attrs.get("processes_created", []))
behavior_data["files_written"].extend(
[f.get("path", "") for f in attrs.get("files_written", [])])
behavior_data["registry_keys_set"].extend(
[r.get("key", "") for r in attrs.get("registry_keys_set", [])])
behavior_data["dns_lookups"].extend(
[d.get("hostname", "") for d in attrs.get("dns_lookups", [])])
behavior_data["commands_executed"].extend(
attrs.get("command_executions", []))
return behavior_data
except Exception as e:
print(f"[-] 行为报告错误：{e}")
return {}
def close(self):
self.client.close()
# 用法
enricher = VTEnricher("YOUR_VT_API_KEY")
report = enricher.enrich_hash("275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f")
print(json.dumps(report, indent=2, default=str))
enricher.close()
步骤 2：批量哈希富化（含速率限制）
import time
import csv
def batch_enrich(api_key, hash_file, output_file, rate_limit=4):
"""从文件中读取哈希列表并执行速率限制富化。"""
enricher = VTEnricher(api_key)
results = []
with open(hash_file, "r") as f:
hashes = [line.strip() for line in f if line.strip()]
print(f"[*] 正在富化 {len(hashes)} 个哈希（速率：{rate_limit} 次/分钟）")
for i, file_hash in enumerate(hashes):
report = enricher.enrich_hash(file_hash)
if report:
results.append(report)
if (i + 1) % rate_limit == 0:
print(f" [{i+1}/{len(hashes)}] 速率限制暂停（60秒）...")
time.sleep(60)
# 导出为 CSV
with open(output_file, "w", newline="") as f:
if results:
writer = csv.DictWriter(f, fieldnames=results[0].keys())
writer.writeheader()
for r in results:
flat = {k: str(v) for k, v in r.items()}
writer.writerow(flat)
print(f"[+] 富化完成：{len(results)}/{len(hashes)} 个哈希")
print(f"[+] 结果已保存至 {output_file}")
enricher.close()
return results
batch_enrich("YOUR_API_KEY", "hashes.txt", "enrichment_results.csv")
步骤 3：提取网络指标进行关联分析
def extract_network_iocs(api_key, file_hash):
"""从 VT 提取基于网络的 IOC 以识别 C2。"""
client = vt.Client(api_key)
network_iocs = {
"contacted_ips": [],
"contacted_domains": [],
"contacted_urls": [],
"embedded_urls": [],
}
try:
# 获取联系的 IP
it = client.iterator(f"/files/{file_hash}/contacted_ips")
for ip_obj in it:
network_iocs["contacted_ips"].append({
"ip": ip_obj.id,
"country": getattr(ip_obj, "country", ""),
"asn": getattr(ip_obj, "asn", 0),
"as_owner": getattr(ip_obj, "as_owner", ""),
})
# 获取联系的域名
it = client.iterator(f"/files/{file_hash}/contacted_domains")
for domain_obj in it:
network_iocs["contacted_domains"].append({
"domain": domain_obj.id,
"registrar": getattr(domain_obj, "registrar", ""),
"creation_date": str(getattr(domain_obj, "creation_date", "")),
})
# 获取联系的 URL
it = client.iterator(f"/files/{file_hash}/contacted_urls")
for url_obj in it:
network_iocs["contacted_urls"].append({
"url": url_obj.url,
"last_http_response_code": getattr(url_obj, "last_http_response_content_length", 0),
})
except Exception as e:
print(f"[-] 提取网络 IOC 时出错：{e}")
finally:
client.close()
print(f"[+] 网络 IOC：{len(network_iocs['contacted_ips'])} 个 IP，"
f"{len(network_iocs['contacted_domains'])} 个域名，"
f"{len(network_iocs['contacted_urls'])} 个 URL")
return network_iocs
步骤 4：YARA 规则匹配和威胁分类
def get_yara_matches(api_key, file_hash):
"""获取 YARA 规则匹配以进行威胁分类。"""
client = vt.Client(api_key)
try:
file_obj = client.get_object(f"/files/{file_hash}")
crowdsourced_yara = getattr(file_obj, "crowdsourced_yara_results", [])
matches = []
for rule in crowdsourced_yara:
matches.append({
"rule_name": rule.get("rule_name", ""),
"ruleset_name": rule.get("ruleset_name", ""),
"author": rule.get("author", ""),
"description": rule.get("description", ""),
"source": rule.get("source", ""),
})
# 基于 YARA 匹配进行分类
classifications = set()
for m in matches:
rule_lower = m["rule_name"].lower()
if any(k in rule_lower for k in ["apt", "nation", "state"]):
classifications.add("apt")
if any(k in rule_lower for k in ["ransom", "crypto"]):
classifications.add("ransomware")
if any(k in rule_lower for k in ["trojan", "rat", "backdoor"]):
classifications.add("trojan")
if any(k in rule_lower for k in ["loader", "dropper"]):
classifications.add("loader")
print(f"[+] YARA：{len(matches)} 条规则匹配")
print(f"[+] 分类：{classifications or {'未分类'}}")
return {"matches": matches, "classifications": list(classifications)}
finally:
client.close()
步骤 5：生成富化报告
def generate_enrichment_report(hash_report, behavior, network, yara_data):
"""生成综合富化报告。"""
report = {
"metadata": {
"generated": datetime.now().isoformat(),
"hash": hash_report.get("sha256", ""),
},
"verdict": {
"threat_level": hash_report.get("threat_level", "unknown"),
"detection_ratio": hash_report.get("detection_ratio", "0/0"),
"classifications": yara_data.get("classifications", []),
"threat_names": hash_report.get("popular_threat_names", {}),
},
"behavioral_indicators": {
"processes": behavior.get("processes_created", [])[:10],
"dns_queries": behavior.get("dns_lookups", [])[:10],
"commands": behavior.get("commands_executed", [])[:10],
},
"network_indicators": {
"c2_candidates": network.get("contacted_ips", [])[:10],
"domains": network.get("contacted_domains", [])[:10],
},
"yara_matches": yara_data.get("matches", [])[:10],
"recommendation": (
"拦截并调查" if hash_report.get("threat_level") in ("critical", "high")
else "监控并分析" if hash_report.get("threat_level") == "medium"
else "低风险 - 继续监控"
),
}
with open(f"enrichment_{hash_report.get('sha256', 'unknown')[:16]}.json", "w") as f:
json.dump(report, f, indent=2, default=str)
return report
验收标准
- VT API v3 经正确身份验证后查询成功
- 文件哈希已富化检测统计、行为数据和网络指标
- 批量富化正确处理速率限制
- 提取网络 IOC 用于识别 C2
- 获取 YARA 匹配并用于分类
- 生成含可操作判决的富化报告