# Performing Malware Hash Enrichment With Virustotal > 使用 VirusTotal API 富化恶意软件文件哈希，获取检测率、行为分析、YARA 匹配和上下文威胁情报，用于事件分类和 IOC 验证。 - Skill: `killvxk/performing-malware-hash-enrichment-with-virustotal` (Agent Skill, multi-file: 4 files) - Install (CLI): `npx skillmds@latest add killvxk/performing-malware-hash-enrichment-with-virustotal` - Raw SKILL.md: https://api.skillmd.com/api/skills/killvxk/performing-malware-hash-enrichment-with-virustotal/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Integrations & APIs - License: Apache-2.0 - Author: killvxk (https://skillmd.com/u/killvxk) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/killvxk/performing-malware-hash-enrichment-with-virustotal --- # 使用 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 哈希报告 ```python 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：批量哈希富化（含速率限制） ```python 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：提取网络指标进行关联分析 ```python 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 规则匹配和威胁分类 ```python 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：生成富化报告 ```python 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 匹配并用于分类 - 生成含可操作判决的富化报告 ## 参考资料 - [VirusTotal API v3 文档](https://docs.virustotal.com/reference/overview) - [vt-py 官方 Python 客户端](https://github.com/VirusTotal/vt-py) - [VirusTotal Intelligence](https://www.virustotal.com/gui/intelligence-overview) - [Torq：VT 哈希富化工作流](https://kb.torq.io/en/articles/9350251-virustotal-file-hash-enrichment-with-cache-workflow-template) - [Dynatrace：使用 VT 富化可观测量](https://www.dynatrace.com/news/blog/enrich-observables-with-virustotal-threat-intelligence/) - [Penligent：VT 在事件响应中的应用](https://www.penligent.ai/hackinglabs/virustotal-in-incident-response-how-to-identify-malware-fast-and-pivot-without-leaking-data/)