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

分析勒索软件数据泄露站点情报 是做什么的？

概述

采用双重勒索模式的勒索软件团伙在Tor隐藏服务上维护着数据泄露站点（DLS），他们在此发布受害者名称、窃取的数据样本以及倒计时器以施加付款压力。2025年上半年，共有96个不同的勒索软件团伙活跃，每月列出约535名受害者。监控这些站点可提供关于活跃威胁团伙、目标行业、地理模式以及新兴勒索软件家族的宝贵情报。本技能涵盖了安全收集DLS情报、提取结构化数据、跟踪团伙活动趋势以及制作特定行业风险评估的方法。

何时使用

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

先决条件

• 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:
    """从公共跟踪源收集勒索软件DLS情报。"""

    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):
        """从ransomwatch获取勒索软件受害者帖子。"""
        resp = requests.get(self.RANSOMWATCH_API, timeout=30)
        if resp.status_code == 200:
            self.posts = resp.json()
            print(f"[+] 从ransomwatch加载了 {len(self.posts)} 个受害者帖子")
        else:
            print(f"[-] 获取帖子失败: {resp.status_code}")

        resp = requests.get(self.RANSOMWATCH_GROUPS, timeout=30)
        if resp.status_code == 200:
            self.groups = resp.json()
            print(f"[+] 加载了 {len(self.groups)} 个勒索软件团伙档案")

        return self.posts

    def get_recent_victims(self, days=30):
        """获取最近N天内发布的受害者信息。"""
        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"[+] 最近 {days} 天内有 {len(recent)} 名受害者")
        return recent

    def get_group_activity(self, group_name):
        """获取特定勒索软件团伙的所有帖子。"""
        group_posts = [
            p for p in self.posts
            if p.get("group_name", "").lower() == group_name.lower()
        ]
        print(f"[+] {group_name}: 共 {len(group_posts)} 名受害者")
        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):
    """分析勒索软件团伙活动趋势。"""
    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=== 勒索软件团伙活动 ===")
    print(f"跟踪的受害者总数: {analysis['total_posts']}")
    print(f"活跃团伙数量: {analysis['unique_groups']}")
    print(f"\n前 {top_n} 个团伙:")
    for group, count in analysis["top_groups"]:
        print(f"  {group}: {count} 名受害者")

    return analysis

trends = analyze_group_trends(collector.posts)

步骤 3：行业与地理风险评估

def assess_sector_risk(posts, target_sector=None, target_country=None):
    """评估特定行业或地理区域的勒索软件风险。"""
    sector_data = {}
    country_data = {}

    for post in posts:
        # 提取行业字段（并非所有数据源都包含此信息）
        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

    # 行业风险评分
    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):
    """识别新出现的勒索软件团伙。"""
    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
    }

    # 统计每个新团伙的受害者总数
    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=== 新团伙（过去 {lookback_days} 天内） ===")
    for group, info in sorted(new_groups.items(), key=lambda x: -x[1]["total_victims"]):
        print(f"  {group}: {info['total_victims']} 名受害者，首次发现于 {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):
    """生成勒索软件威胁情报报告。"""
    report = f"""# 勒索软件威胁情报报告
生成时间: {datetime.now().isoformat()}

## 执行摘要
- **跟踪的受害者总数**: {trends['total_posts']}
- **活跃勒索软件团伙数量**: {trends['unique_groups']}
- **新团伙（过去90天内）**: {len(new_groups)}

## 最活跃团伙
| 排名 | 团伙 | 受害者数 |
|------|------|---------|
"""
    for i, (group, count) in enumerate(trends["top_groups"][:10], 1):
        report += f"| {i} | {group} | {count} |\n"

    report += "\n## 新兴团伙\n"
    for group, info in sorted(new_groups.items(), key=lambda x: -x[1]["total_victims"])[:10]:
        report += f"- **{group}**: 自 {info['first_seen'].strftime('%Y-%m-%d')} 以来共 {info['total_victims']} 名受害者\n"

    report += "\n## 行业风险评估\n"
    report += "| 行业 | 受害者数 | % | 风险等级 |\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 += """
## 建议
1. 每天监控与您组织及供应链合作伙伴相关的DLS源
2. 优先修补被最活跃团伙利用的漏洞
3. 实施离线备份策略以降低勒索杠杆
4. 针对勒索软件场景进行桌面演练
5. 与行业ISAC和威胁共享社区共享指标
"""
    with open("ransomware_intel_report.md", "w") as f:
        f.write(report)
    print("[+] 报告已保存: 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