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

概要

ダブルエクストーション型で活動するランサムウェアグループは、Tor隠蔽サービス上でデータ漏洩サイト(DLS)を運営し、被害者名、盗まれたデータサンプル、支払い期限までのカウントダウンタイマーを投稿して身代金の支払いを強要しています。2025年上半期には、96の独自のランサムウェアグループが活動し、月間約535の被害者がリストアップされています。これらのサイトを監視することで、活動中の脅威グループ、標的セクター、地理的パターン、新興ランサムウェアファミリーに関するインテリジェンスが得られます。このスキルは、DLSインテリジェンスを安全に収集し、構造化データを抽出し、グループ活動の傾向を追跡し、セクター別のリスク評価を作成する方法をカバーしています。

使用する場合

• ランサムウェア漏洩サイトインテリジェンスの分析が必要なセキュリティインシデントを調査する場合
• このドメインの検知ルールまたは脅威ハンティングクエリを構築する場合
• SOCアナリストがこのタイプの分析の構造化手順を必要とする場合
• 関連する攻撃技術のセキュリティ監視カバレッジを検証する場合

前提条件

• Python 3.9以上(requests、beautifulsoup4、pandas、matplotlibライブラリ付き)
• .onionサイトへのアクセスまたは商用DLS監視フィードのためのTorプロキシ(SOCKS5)
• ランサムウェアダブルエクストーション商業モデルの理解
• 主要なランサムウェアファミリー(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: Top Ransomware Statistics 2025
• Bitsight: Ransomware & Deep Web Trends
• Sophos: Threat Intelligence Report 2025
• H-ISAC: Ransomware Data Leak Sites Report
• CYFIRMA: Weekly Intelligence Reports