analyzing-malware-family-relationships-with-malpedia

Malpedia を使用したマルウェアファミリーの関係分析

概要

Malpedia は Fraunhofer FKIE が運営する共同プラットフォームで、マルウェアファミリーとその別名、YARA ルール、脅威アクターとの関連性、参照レポートをカタログ化しています。2,600 を超えるマルウェアファミリーが記録されており、マルウェアの系統の把握、亜種の変化の追跡、マルウェアと特定の脅威グループとの関連付けに役立つ信頼できる情報源です。このスキルでは、Malpedia API へのクエリ、マルウェアファミリー間の関係のマッピング、検知用 YARA ルールの抽出、攻撃者が利用するマルウェアエコシステムに関するインテリジェンスの構築について説明します。

使用する場面

• Malpedia を使用してマルウェアファミリー間の関係を分析する必要があるセキュリティインシデントの調査
• この分野の検知ルールや脅威ハンティング用クエリの作成
• SOC アナリストがこの種の分析のための体系的な手順を必要とする場合
• 関連する攻撃手法に対するセキュリティ監視のカバレッジの検証

前提条件

• requests、yara-python、stix2 ライブラリを含む Python 3.9 以降
• Malpedia API キー（https://malpedia.caad.fkie.fraunhofer.de/ で登録）
• マルウェアの分類と命名規則に関する知識
• 検知用 YARA ルールの構文に関する知識
• 検証用のマルウェアサンプルへのアクセス（任意）

主な概念

Malpedia のデータモデル

Malpedia では、マルウェアをファミリー（例：win.cobalt_strike）単位で整理しています。各ファミリーには、別名（「Beacon」や「CobaltStrike」など、ベンダー固有の名称）、YARA ルール（コミュニティやベンダーから提供されたもの）、アクターとの関連性（そのファミリーを使用する脅威グループ）、参照レポート（ファミリーについて記録した CTI レポート）、サンプルハッシュ（各亜種の代表的なサンプル）が含まれます。

マルウェアファミリーの命名

Malpedia では platform.family_name の形式を使用します（例：win.emotet、elf.mirai、apk.flubot）。プラットフォームには win（Windows）、elf（Linux）、apk（Android）、osx（macOS）、py（Python）があります。この標準化された命名により、ベンダーごとに同じマルウェアに異なる名前が付けられる「多数の名称」問題を解決できます。

ファミリー間の関係

マルウェアファミリーには、親子関係（コードの再利用、フォーク）、ローダーとペイロードの関係（Emotet が TrickBot を読み込み、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):
        """Get list of all malware families."""
        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):
        """Get detailed information about a malware family."""
        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.get('alt_names', [])}")
            print(f"    Actors: {[a.get('value', '') for a in info.get('attribution', [])]}")
            print(f"    URLs: {len(info.get('urls', []))} references")
            return info
        print(f"[-] Family not found: {family_name}")
        return None

    def get_family_yara(self, family_name):
        """Get YARA rules for a malware family."""
        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"[+] YARA rules for {family_name}: {rule_count} rules")
            return rules
        return {}

    def get_actor_families(self, actor_name):
        """Get malware families associated with a threat actor."""
        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)} malware families")
            return data
        return {}

    def search_families(self, keyword):
        """Search families by 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"[+] Search '{keyword}': {len(matches)} 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):
        """Map the malware ecosystem used by a threat actor."""
        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} Malware Ecosystem ===")
        for fam in ecosystem["families"]:
            shared = [a for a in fam["shared_actors"] if a != actor_name]
            print(f"  {fam['name']}")
            print(f"    Aliases: {fam['aliases'][:5]}")
            if shared:
                print(f"    Also used by: {shared}")

        return ecosystem

    def find_shared_tooling(self, actor_names):
        """Find malware families shared between threat actors."""
        actor_families = {}
        for actor in actor_names:
            data = self.client.get_actor_families(actor)
            actor_families[actor] = set(data.get("families", {}).keys())

        # Find overlaps
        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=== Shared Tooling Analysis ===")
        for pair, families in shared.items():
            print(f"  {pair}: {len(families)} shared families")
            for f in families[:5]:
                print(f"    - {f}")

        return shared

    def build_loader_payload_chain(self, family_name):
        """Build the loader-payload delivery chain for a family."""
        info = self.client.get_family_info(family_name)
        if not info:
            return {}

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

        # Common known delivery chains
        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"):
    """Compile YARA rules for multiple malware families."""
    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"[+] Compiled {len(all_rules)} YARA rules to {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 の統合