analyzing-ransomware-leak-site-intelligence

分析勒索软件泄露站点情报

概述

采用双重勒索模式的勒索软件组织在 Tor 隐藏服务上维护数据泄露站点（DLS），发布受害者名称、被盗数据样本和倒计时以施压支付赎金。2025 年上半年，共有 96 个活跃的勒索软件组织，每月约列出 535 名受害者。监控这些站点可获取关于活跃威胁组织、目标行业、地理模式以及新兴勒索软件家族的情报。本技能涵盖安全收集 DLS 情报、提取结构化数据、跟踪组织活动趋势以及生成特定行业的风险评估。

使用时机

• 调查需要分析勒索软件泄露站点情报的安全事件时
• 为该领域构建检测规则或威胁狩猎查询时
• SOC 分析师需要针对此类分析的标准化流程时
• 验证相关攻击技术的安全监控覆盖情况时

前提条件

• Python 3.9+ 及 requests、beautifulsoup4、pandas、matplotlib 库
• 用于访问 .onion 站点的 Tor 代理（SOCKS5）或商业 DLS 监控源
• 了解勒索软件双重勒索商业模式
• 熟悉主要勒索软件家族（Qilin、Akira、LockBit、BlackCat、Clop）
• 可访问勒索软件跟踪源（Ransomwatch、RansomLook、DarkFeed）

关键概念

双重勒索模式

现代勒索软件组织在加密受害者数据之前先将其窃取。泄露站点作为公开施压手段：受害者被列出，附带倒计时、部分数据样本和文件树。若未支付赎金，完整数据将被公开。部分组织已转向三重勒索，增加 DDoS 威胁或直接联系受害者的客户。

DLS 情报价值

泄露站点提供：受害者识别（公司名称、行业、国家）、攻击时间线（何时被列出、截止日期、数据发布时间）、数据量估算、组织能力评估（目标行业、攻击频率、行动节奏）以及趋势分析（新兴组织、组织更名、执法打击）。

安全收集实践

切勿在生产环境中直接访问 DLS 站点。使用专用监控服务（Ransomwatch、DarkFeed、KELA、Flashpoint）、Tor 隔离研究虚拟机、商业威胁情报平台或社区维护的数据集。所有分析应在隔离环境中并获得适当授权后进行。

工作流

步骤 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