analyzing-malware-family-relationships-with-malpedia

name	analyzing-malware-family-relationships-with-malpedia
description	使用 Malpedia 平台和 API 研究恶意软件家族关系、追踪变体演化、将家族关联到威胁行为者，并整合 YARA 规则用于跨恶意软件谱系的检测。
domain	cybersecurity
subdomain	threat-intelligence
tags	["malpedia","malware-family","yara","threat-actor","malware-tracking","threat-intelligence","variant-analysis","malware-intelligence"]
version	1.0
author	mahipal
license	Apache-2.0

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

概述

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

前置条件

• 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（Android）、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)} malware 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: {family_name}")
            print(f"    Aliases: ")
            ()
            ()
             info
        ()
         

     ():
        
        resp = requests.get(,
                           headers=.headers, timeout=)
         resp.status_code == :
            rules = resp.json()
            rule_count = ((v)  v  rules.values())  (rules, )  
            ()
             rules
         {}

     ():
        
        resp = requests.get(,
                           headers=.headers, timeout=)
         resp.status_code == :
            data = resp.json()
            families = data.get(, {})
            ()
             data
         {}

     ():
        
        all_families = .get_family_list()
        matches = {
            name: info  name, info  all_families.items()
             keyword.lower()  name.lower()
             keyword.lower()  (info.get(, [])).lower()
        }
        ()
         matches

client = MalpediaClient()
families = client.get_family_list()
emotet_info = client.get_family_info()

步骤 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} 恶意软件生态系统 ===")
         fam  ecosystem[]:
            shared = [a  a  fam[]  a != actor_name]
            ()
            ()
             shared:
                ()

         ecosystem

     ():
        
        actor_families = {}
         actor  actor_names:
            data = .client.get_actor_families(actor)
            actor_families[actor] = (data.get(, {}).keys())

        
        shared = {}
         i, actor1  (actor_names):
             actor2  actor_names[i+:]:
                common = actor_families[actor1] & actor_families[actor2]
                 common:
                    shared[] = (common)

        ()
         pair, families  shared.items():
            ()
             f  families[:]:
                ()

         shared

     ():
        
        info = .client.get_family_info(family_name)
          info:
             {}

        chain = {
            : family_name,
            : info.get(, ),
            : [],
            : [],
        }

        
        known_chains = {
            : {: [], : [, , ]},
            : {: [], : [, , ]},
            : {: [, ], : [, ]},
            : {: [, , ], : []},
        }

         family_name  known_chains:
            chain[] = known_chains[family_name][]
            chain[] = known_chains[family_name][]

         chain

mapper = MalwareFamilyMapper(client)
ecosystem = mapper.map_actor_ecosystem()
shared = mapper.find_shared_tooling([, , ])
chain = mapper.build_loader_payload_chain()

步骤 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: {source} - Family: {family}\n{rule}")
                elif isinstance(rules, str):
                    all_rules.append(f"// Source: {source} - Family: {family}\n{rules}")

    with open(output_file, "w") as f:
        f.write(f"// Malpedia YARA Rules - {len(all_rules)} rules\n")
        f.write(f"// Families: {', '.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, [, , ])

验收标准

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

参考资料

• Malpedia Platform
• Malpedia API Documentation
• Malpedia Research Paper
• YARA Rules Project
• malwoverview Multi-Platform Tool
• CyberAtlas: Malpedia Integration