analyzing-ransomware-leak-site-intelligence

ランサムウェアのリークサイトインテリジェンスの分析

概要

二重恐喝モデルを採用するランサムウェアグループは、Tor の隠しサービス上にデータリークサイト（DLS）を運営し、被害者の名前、盗み出したデータのサンプル、支払いを迫るカウントダウンタイマーを掲載しています。2025 年上半期には、96 のランサムウェアグループが活動しており、月間約 535 件の被害者を掲載していました。こうしたサイトを監視することで、活動中の脅威グループ、標的となるセクター、地域ごとの傾向、新たなランサムウェアファミリーに関するインテリジェンスが得られます。このスキルでは、DLS インテリジェンスを安全に収集し、構造化データを抽出し、グループの活動傾向を追跡し、セクター別のリスク評価を作成する方法を説明します。

使用する場面

• ランサムウェアのリークサイトインテリジェンスの分析を要するセキュリティインシデントを調査する場合
• この領域の検出ルールや脅威ハンティング用クエリを作成する場合
• SOC アナリストがこの種の分析に関する体系的な手順を必要とする場合
• 関連する攻撃手法に対するセキュリティ監視の網羅性を検証する場合

前提条件

• requests、beautifulsoup4、pandas、matplotlib ライブラリを含む Python 3.9+
• .onion サイトにアクセスするための Tor プロキシ（SOCKS5）、または商用 DLS 監視フィード
• ランサムウェアの二重恐喝ビジネスモデルに関する理解
• 主要なランサムウェアファミリー（Qilin、Akira、LockBit、BlackCat、Clop）に関する知識
• ランサムウェア追跡フィード（Ransomwatch、RansomLook、DarkFeed）へのアクセス

主な概念

二重恐喝モデル

現代のランサムウェアグループは、被害者のデータを暗号化するとともに、暗号化の前にデータを持ち出します。リークサイトは公の場で圧力をかける手段となります。被害者はカウントダウンタイマー、一部のデータサンプル、ファイルツリーとともに掲載されます。身代金が支払われなければ、データ全体が公開されます。一部のグループは三重恐喝に移行し、DDoS 攻撃の脅迫や、被害者の顧客への直接連絡を加えています。

DLS インテリジェンスの価値

リークサイトからは、被害者の特定（企業名、セクター、国）、攻撃のタイムライン（掲載時期、期限、データ公開状況）、データ量の推定、グループの能力評価（標的とするセクター、攻撃頻度、活動ペース）、傾向分析（新たなグループの出現、グループの改称、法執行機関による摘発）に関する情報が得られます。

安全な収集方法

本番環境から DLS サイトに直接アクセスしてはいけません。専用の監視サービス（Ransomwatch、DarkFeed、KELA、Flashpoint）、Tor を隔離した調査用 VM、商用の脅威インテリジェンスプラットフォーム、またはコミュニティが管理するデータセットを利用してください。すべての分析は、適切な認可を得たうえで、隔離された環境で実施してください。

ワークフロー

ステップ 1：公開フィードからランサムウェアのリークサイトデータを取り込む

import requests
import json
import pandas as pd
from datetime import datetime, timedelta
from collections import Counter

class RansomwareIntelCollector:
    """Collect ransomware DLS intelligence from public tracking sources."""

    RANSOMWATCH_API = "https://raw.githubusercontent.com/joshhighet/ransomwatch/main/posts.json"
    RANSOMWATCH_GROUPS = "https://raw.githubusercontent.com/joshhighet/ransomwatch/main/groups.json"

    def __init__(self):
        self.posts = []
        self.groups = []

    def fetch_ransomwatch_data(self):
        """Fetch ransomware victim posts from ransomwatch."""
        resp = requests.get(self.RANSOMWATCH_API, timeout=30)
        if resp.status_code == 200:
            self.posts = resp.json()
            print(f"[+] Loaded {len(self.posts)} victim posts from ransomwatch")
        else:
            print(f"[-] Failed to fetch posts: {resp.status_code}")

        resp = requests.get(self.RANSOMWATCH_GROUPS, timeout=30)
        if resp.status_code == 200:
            self.groups = resp.json()
            print(f"[+] Loaded {len(self.groups)} ransomware group profiles")

        return self.posts

    def get_recent_victims(self, days=30):
        """Get victims posted in the last N days."""
        cutoff = datetime.now() - timedelta(days=days)
        recent = []
        for post in self.posts:
            try:
                discovered = datetime.fromisoformat(
                    post.get("discovered", "").replace("Z", "+00:00")
                )
                if discovered.replace(tzinfo=None) >= cutoff:
                    recent.append(post)
            except (ValueError, TypeError):
                continue
        print(f"[+] {len(recent)} victims in last {days} days")
        return recent

    def get_group_activity(self, group_name):
        """Get all posts by a specific ransomware group."""
        group_posts = [
            p for p in self.posts
            if p.get("group_name", "").lower() == group_name.lower()
        ]
        print(f"[+] {group_name}: {len(group_posts)} total victims")
        return group_posts

collector = RansomwareIntelCollector()
collector.fetch_ransomwatch_data()
recent = collector.get_recent_victims(days=30)

ステップ 2：グループの活動と傾向を分析する

def analyze_group_trends(posts, top_n=15):
    """Analyze ransomware group activity trends."""
    group_counts = Counter(p.get("group_name", "unknown") for p in posts)
    monthly_activity = {}

    for post in posts:
        try:
            date = datetime.fromisoformat(
                post.get("discovered", "").replace("Z", "+00:00")
            )
            month_key = date.strftime("%Y-%m")
            group = post.get("group_name", "unknown")
            if month_key not in monthly_activity:
                monthly_activity[month_key] = Counter()
            monthly_activity[month_key][group] += 1
        except (ValueError, TypeError):
            continue

    analysis = {
        "total_posts": len(posts),
        "unique_groups": len(group_counts),
        "top_groups": group_counts.most_common(top_n),
        "monthly_totals": {
            month: sum(counts.values())
            for month, counts in sorted(monthly_activity.items())
        },
        "monthly_top_groups": {
            month: counts.most_common(5)
            for month, counts in sorted(monthly_activity.items())
        },
    }

    print(f"\n=== Ransomware Group Activity ===")
    print(f"Total victims tracked: {analysis['total_posts']}")
    print(f"Active groups: {analysis['unique_groups']}")
    print(f"\nTop {top_n} Groups:")
    for group, count in analysis["top_groups"]:
        print(f"  {group}: {count} victims")

    return analysis

trends = analyze_group_trends(collector.posts)

ステップ 3：セクターおよび地域のリスク評価を行う

def assess_sector_risk(posts, target_sector=None, target_country=None):
    """Assess ransomware risk for specific sector or geography."""
    sector_data = {}
    country_data = {}

    for post in posts:
        # Extract sector if available (not all feeds include this)
        sector = post.get("sector", post.get("industry", "unknown"))
        country = post.get("country", "unknown")

        if sector not in sector_data:
            sector_data[sector] = {"count": 0, "groups": Counter(), "recent": []}
        sector_data[sector]["count"] += 1
        sector_data[sector]["groups"][post.get("group_name", "")] += 1

        if country not in country_data:
            country_data[country] = {"count": 0, "groups": Counter()}
        country_data[country]["count"] += 1
        country_data[country]["groups"][post.get("group_name", "")] += 1

    # Sector risk scoring
    total = len(posts)
    risk_assessment = {
        "total_victims": total,
        "sectors": {},
        "countries": {},
    }

    for sector, data in sorted(sector_data.items(), key=lambda x: -x[1]["count"]):
        pct = (data["count"] / total * 100) if total > 0 else 0
        risk_assessment["sectors"][sector] = {
            "victim_count": data["count"],
            "percentage": round(pct, 1),
            "top_groups": data["groups"].most_common(5),
            "risk_level": (
                "critical" if pct > 15
                else "high" if pct > 8
                else "medium" if pct > 3
                else "low"
            ),
        }

    for country, data in sorted(country_data.items(), key=lambda x: -x[1]["count"]):
        pct = (data["count"] / total * 100) if total > 0 else 0
        risk_assessment["countries"][country] = {
            "victim_count": data["count"],
            "percentage": round(pct, 1),
            "top_groups": data["groups"].most_common(5),
        }

    return risk_assessment

risk = assess_sector_risk(collector.posts)

ステップ 4：新興グループや改称したグループを追跡する

def track_new_groups(posts, lookback_days=90):
    """Identify newly emerged ransomware groups."""
    group_first_seen = {}
    for post in posts:
        group = post.get("group_name", "")
        try:
            date = datetime.fromisoformat(
                post.get("discovered", "").replace("Z", "+00:00")
            )
            if group not in group_first_seen or date < group_first_seen[group]["first_seen"]:
                group_first_seen[group] = {
                    "first_seen": date,
                    "first_victim": post.get("post_title", ""),
                }
        except (ValueError, TypeError):
            continue

    cutoff = datetime.now() - timedelta(days=lookback_days)
    new_groups = {
        group: info for group, info in group_first_seen.items()
        if info["first_seen"].replace(tzinfo=None) >= cutoff
    }

    # Count total victims per new group
    for group in new_groups:
        victims = [p for p in posts if p.get("group_name") == group]
        new_groups[group]["total_victims"] = len(victims)
        new_groups[group]["avg_per_month"] = round(
            len(victims) / max(1, lookback_days / 30), 1
        )

    print(f"\n=== New Groups (last {lookback_days} days) ===")
    for group, info in sorted(new_groups.items(), key=lambda x: -x[1]["total_victims"]):
        print(f"  {group}: {info['total_victims']} victims, "
              f"first seen {info['first_seen'].strftime('%Y-%m-%d')}")

    return new_groups

new_groups = track_new_groups(collector.posts, lookback_days=90)

ステップ 5：インテリジェンスレポートを作成する

def generate_ransomware_intel_report(trends, risk, new_groups):
    """Generate ransomware threat intelligence report."""
    report = f"""# Ransomware Threat Intelligence Report
Generated: {datetime.now().isoformat()}

## Executive Summary
- **Total victims tracked**: {trends['total_posts']}
- **Active ransomware groups**: {trends['unique_groups']}
- **New groups (last 90 days)**: {len(new_groups)}

## Top Active Groups
| Rank | Group | Victims |
|------|-------|---------|
"""
    for i, (group, count) in enumerate(trends["top_groups"][:10], 1):
        report += f"| {i} | {group} | {count} |\n"

    report += "\n## New Emerging Groups\n"
    for group, info in sorted(new_groups.items(), key=lambda x: -x[1]["total_victims"])[:10]:
        report += f"- **{group}**: {info['total_victims']} victims since {info['first_seen'].strftime('%Y-%m-%d')}\n"

    report += "\n## Sector Risk Assessment\n"
    report += "| Sector | Victims | % | Risk Level |\n|--------|---------|---|------------|\n"
    for sector, data in list(risk["sectors"].items())[:10]:
        report += f"| {sector} | {data['victim_count']} | {data['percentage']}% | {data['risk_level'].upper()} |\n"

    report += """
## Recommendations
1. Monitor DLS feeds daily for your organization and supply chain partners
2. Prioritize patching vulnerabilities exploited by top active groups
3. Implement offline backup strategy to reduce extortion leverage
4. Conduct tabletop exercises for ransomware scenario response
5. Share indicators with sector ISACs and threat sharing communities
"""
    with open("ransomware_intel_report.md", "w") as f:
        f.write(report)
    print("[+] Report saved: ransomware_intel_report.md")
    return report

generate_ransomware_intel_report(trends, risk, new_groups)

検証基準

• 公開追跡フィードからランサムウェア被害者データを取り込んでいる
• 月ごとの内訳を含めてグループの活動傾向を分析している
• セクターおよび地域のリスク評価を作成している
• 活動指標とともに、新たなグループや新興グループを特定している
• 実行可能な推奨事項を含むインテリジェンスレポートを作成している
• すべてのデータ収集を、認可された公開ソースを通じて実施している

参考資料

• Ransomwatch GitHub
• SOCRadar：2025 年のランサムウェア統計トップ
• Bitsight：ランサムウェアとディープウェブの傾向
• Sophos：2025 年脅威インテリジェンスレポート
• H-ISAC：ランサムウェアのデータリークサイトレポート
• CYFIRMA：週次インテリジェンスレポート