使用 Malpedia 分析恶意软件家族关系

使用 Malpedia 分析恶意软件家族关系 是做什么的？

概述

Malpedia 是由 Fraunhofer FKIE 维护的一个协作平台，它收录了恶意软件家族及其别名、YARA 规则、威胁行为者关联和参考报告。该平台记录了超过 2,600 个恶意软件家族，是理解恶意软件谱系、追踪变体演化以及将恶意软件与特定威胁组织关联起来的权威资源。本技能涵盖查询 Malpedia API、映射恶意软件家族关系、提取用于检测的 YARA 规则，以及构建关于对手所使用的恶意软件生态系统的情报。

使用场景

• 在调查需要分析恶意软件家族关系的安全事件时
• 在为此领域构建检测规则或威胁狩猎查询时
• 当 SOC 分析师需要针对此类分析的结构化处理流程时
• 在验证相关攻击技术的安全监控覆盖范围时

前提条件

• Python 3.9+，并安装 requests、yara-python、stix2 库
• Malpedia API 密钥（在 https://malpedia.caad.fkie.fraunhofer.de/ 注册）
• 了解恶意软件分类和命名规范
• 熟悉用于检测的 YARA 规则语法
• 可访问恶意软件样本进行验证（可选）

关键概念

Malpedia 数据模型

Malpedia 将恶意软件组织成“家族”（例如 win.cobalt_strike），每个家族包含：别名（各厂商的名称，如“Beacon”、“CobaltStrike”）、YARA 规则（社区和厂商贡献）、行为者关联（使用该家族的威胁组织）、参考报告（记录该家族的 CTI 报告）以及样本哈希值（每个变体的代表性样本）。

恶意软件家族命名

Malpedia 使用 平台.家族名称 的格式（例如 win.emotet、elf.mirai、apk.flubot）。平台包括：win（Windows）、elf（Linux）、apk（安卓）、osx（macOS）和 py（Python）。这种标准化的命名解决了“多名”问题，即不同厂商为同一恶意软件分配不同名称的情况。

家族关系

恶意软件家族之间存在以下关系：父子（代码复用、分支）、加载器-载荷（如 Emotet 加载 TrickBot，再加载 Ryuk）、共享作者（同一威胁行为者开发多种工具）以及基础设施共享（通用的 C2 框架）。

工作流程

步骤 1：查询 Malpedia API 获取恶意软件家族信息

import requests
import json
from collections import defaultdict

class MalpediaClient:
    BASE_URL = "https://malpedia.caad.fkie.fraunhofer.de/api"

    def __init__(self, api_key):
        self.headers = {"Authorization": f"apitoken {api_key}"}

    def get_family_list(self):
        """获取所有恶意软件家族的列表。"""
        resp = requests.get(f"{self.BASE_URL}/list/families",
                           headers=self.headers, timeout=30)
        if resp.status_code == 200:
            families = resp.json()
            print(f"[+] Malpedia: {len(families)} 个恶意软件家族")
            return families
        return {}

    def get_family_info(self, family_name):
        """获取某个恶意软件家族的详细信息。"""
        resp = requests.get(f"{self.BASE_URL}/get/family/{family_name}",
                           headers=self.headers, timeout=30)
        if resp.status_code == 200:
            info = resp.json()
            print(f"[+] 家族: {family_name}")
            print(f"    别名: {info.get('alt_names', [])}")
            print(f"    行为者: {[a.get('value', '') for a in info.get('attribution', [])]}")
            print(f"    参考链接数: {len(info.get('urls', []))}")
            return info
        print(f"[-] 未找到家族: {family_name}")
        return None

    def get_family_yara(self, family_name):
        """获取某个恶意软件家族的 YARA 规则。"""
        resp = requests.get(f"{self.BASE_URL}/get/yara/{family_name}",
                           headers=self.headers, timeout=30)
        if resp.status_code == 200:
            rules = resp.json()
            rule_count = sum(len(v) for v in rules.values()) if isinstance(rules, dict) else 0
            print(f"[+] {family_name} 的 YARA 规则: {rule_count} 条规则")
            return rules
        return {}

    def get_actor_families(self, actor_name):
        """获取与某个威胁行为者关联的恶意软件家族。"""
        resp = requests.get(f"{self.BASE_URL}/get/actor/{actor_name}",
                           headers=self.headers, timeout=30)
        if resp.status_code == 200:
            data = resp.json()
            families = data.get("families", {})
            print(f"[+] {actor_name}: {len(families)} 个恶意软件家族")
            return data
        return {}

    def search_families(self, keyword):
        """按关键词搜索家族。"""
        all_families = self.get_family_list()
        matches = {
            name: info for name, info in all_families.items()
            if keyword.lower() in name.lower()
            or keyword.lower() in str(info.get("alt_names", [])).lower()
        }
        print(f"[+] 搜索 '{keyword}': {len(matches)} 个匹配")
        return matches

client = MalpediaClient("YOUR_MALPEDIA_API_KEY")
families = client.get_family_list()
emotet_info = client.get_family_info("win.emotet")

步骤 2：映射恶意软件家族关系

class MalwareFamilyMapper:
    def __init__(self, malpedia_client):
        self.client = malpedia_client
        self.relationship_graph = defaultdict(list)

    def map_actor_ecosystem(self, actor_name):
        """映射某个威胁行为者使用的恶意软件生态系统。"""
        actor_data = self.client.get_actor_families(actor_name)
        families = actor_data.get("families", {})

        ecosystem = {
            "actor": actor_name,
            "families": [],
            "family_count": len(families),
        }

        for family_name in families:
            info = self.client.get_family_info(family_name)
            if info:
                ecosystem["families"].append({
                    "name": family_name,
                    "aliases": info.get("alt_names", []),
                    "description": info.get("description", "")[:200],
                    "shared_actors": [
                        a.get("value", "")
                        for a in info.get("attribution", [])
                    ],
                    "reference_count": len(info.get("urls", [])),
                })

        print(f"\n=== {actor_name} 恶意软件生态系统 ===")
        for fam in ecosystem["families"]:
            shared = [a for a in fam["shared_actors"] if a != actor_name]
            print(f"  {fam['name']}")
            print(f"    别名: {fam['aliases'][:5]}")
            if shared:
                print(f"    也被以下行为者使用: {shared}")

        return ecosystem

    def find_shared_tooling(self, actor_names):
        """查找多个威胁行为者之间共享的恶意软件家族。"""
        actor_families = {}
        for actor in actor_names:
            data = self.client.get_actor_families(actor)
            actor_families[actor] = set(data.get("families", {}).keys())

        # 查找重叠部分
        shared = {}
        for i, actor1 in enumerate(actor_names):
            for actor2 in actor_names[i+1:]:
                common = actor_families[actor1] & actor_families[actor2]
                if common:
                    shared[f"{actor1} <-> {actor2}"] = sorted(common)

        print(f"\n=== 共享工具分析 ===")
        for pair, families in shared.items():
            print(f"  {pair}: {len(families)} 个共享家族")
            for f in families[:5]:
                print(f"    - {f}")

        return shared

    def build_loader_payload_chain(self, family_name):
        """为某个家族构建加载器-载荷投递链。"""
        info = self.client.get_family_info(family_name)
        if not info:
            return {}

        chain = {
            "family": family_name,
            "description": info.get("description", ""),
            "known_loaders": [],
            "known_payloads": [],
        }

        # 常见的已知投递链
        known_chains = {
            "win.emotet": {"loaders": ["email/macro"], "payloads": ["win.trickbot", "win.qakbot", "win.cobalt_strike"]},
            "win.trickbot": {"loaders": ["win.emotet"], "payloads": ["win.ryuk", "win.conti", "win.cobalt_strike"]},
            "win.qakbot": {"loaders": ["email/macro", "win.emotet"], "payloads": ["win.cobalt_strike", "win.blackbasta"]},
            "win.cobalt_strike": {"loaders": ["win.emotet", "win.trickbot", "win.qakbot"], "payloads": ["ransomware"]},
        }

        if family_name in known_chains:
            chain["known_loaders"] = known_chains[family_name]["loaders"]
            chain["known_payloads"] = known_chains[family_name]["payloads"]

        return chain

mapper = MalwareFamilyMapper(client)
ecosystem = mapper.map_actor_ecosystem("Wizard Spider")
shared = mapper.find_shared_tooling(["Wizard Spider", "FIN7", "Lazarus Group"])
chain = mapper.build_loader_payload_chain("win.emotet")

步骤 3：提取和编译 YARA 规则

def compile_yara_ruleset(client, family_names, output_file="malware_yara_rules.yar"):
    """为多个恶意软件家族编译 YARA 规则。"""
    all_rules = []
    for family in family_names:
        yara_data = client.get_family_yara(family)
        if isinstance(yara_data, dict):
            for source, rules in yara_data.items():
                if isinstance(rules, list):
                    for rule in rules:
                        all_rules.append(f"// 来源: {source} - 家族: {family}\n{rule}")
                elif isinstance(rules, str):
                    all_rules.append(f"// 来源: {source} - 家族: {family}\n{rules}")

    with open(output_file, "w") as f:
        f.write(f"// Malpedia YARA 规则 - {len(all_rules)} 条规则\n")
        f.write(f"// 家族: {', '.join(family_names)}\n\n")
        for rule in all_rules:
            f.write(rule + "\n\n")

    print(f"[+] 已将 {len(all_rules)} 条 YARA 规则编译到 {output_file}")
    return all_rules

compile_yara_ruleset(client, ["win.emotet", "win.trickbot", "win.cobalt_strike"])

验证标准

• 成功查询 Malpedia API 获取恶意软件家族
• 检索出包含别名、行为者和参考链接的家族信息
• 正确映射行为者与家族之间的关系
• 识别出行为者之间共享的工具
• 提取并编译 YARA 规则用于检测
• 记录加载器-载荷链以支持威胁情报

参考

• Malpedia 平台
• Malpedia API 文档
• Malpedia 研究论文
• YARA 规则项目
• malwoverview 多平台工具
• CyberAtlas: Malpedia 集成