performing-malware-hash-enrichment-with-virustotal

name	performing-malware-hash-enrichment-with-virustotal
description	使用 VirusTotal API 富化恶意软件文件哈希，获取检测率、行为分析、YARA 匹配和上下文威胁情报，用于事件分类和 IOC 验证。
domain	cybersecurity
subdomain	threat-intelligence
tags	["virustotal","malware-analysis","hash-enrichment","ioc","threat-intelligence","triage","api","detection"]
version	1.0
author	mahipal
license	Apache-2.0

使用 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),
                    : stats.get(, ),
                    : stats.get(, ),
                    : stats.get(, ),
                },
                : ,
                : (file_obj, , {}),
                : (file_obj, , []),
                : (file_obj, , []),
            }
            total_engines = (stats.values())
            mal_count = stats.get(, )
            report[] = (
                  mal_count > total_engines * 
                   mal_count > total_engines * 
                   mal_count > total_engines * 
                   mal_count > 
                 
            )
            (
                  )
             report
         vt.error.APIError  e:
            ()
             

     ():
        
        :
            behaviors = .client.get_object()
            behavior_data = {
                : [],
                : [],
                : [],
                : [],
                : [],
                : [],
                : [],
            }
             sandbox  (behaviors, , []):
                attrs = sandbox.get(, {})
                behavior_data[].extend(
                    attrs.get(, []))
                behavior_data[].extend(
                    [f.get(, )  f  attrs.get(, [])])
                behavior_data[].extend(
                    [r.get(, )  r  attrs.get(, [])])
                behavior_data[].extend(
                    [d.get(, )  d  attrs.get(, [])])
                behavior_data[].extend(
                    attrs.get(, []))
             behavior_data
         Exception  e:
            ()
             {}

     ():
        .client.close()


enricher = VTEnricher()
report = enricher.enrich_hash()
(json.dumps(report, indent=, default=))
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()
    ()
    enricher.close()
     results

batch_enrich(, , )

步骤 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", )),
            })

        
        it = client.iterator()
         url_obj  it:
            network_iocs[].append({
                : url_obj.url,
                : (url_obj, , ),
            })

     Exception  e:
        ()
    :
        client.close()

    (
          
          )
     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()
             (k  rule_lower  k  [, , ]):
                classifications.add()
             (k  rule_lower  k  [, ]):
                classifications.add()

        ()
        ()
         {: matches, : (classifications)}
    :
        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()  (, )
               hash_report.get() == 
             
        ),
    }

     (, )  f:
        json.dump(report, f, indent=, default=)
     report

验收标准

• VT API v3 经正确身份验证后查询成功
• 文件哈希已富化检测统计、行为数据和网络指标
• 批量富化正确处理速率限制
• 提取网络 IOC 用于识别 C2
• 获取 YARA 匹配并用于分类
• 生成含可操作判决的富化报告

参考资料

• VirusTotal API v3 文档
• vt-py 官方 Python 客户端
• VirusTotal Intelligence
• Torq：VT 哈希富化工作流
• Dynatrace：使用 VT 富化可观测量
• Penligent：VT 在事件响应中的应用