| name | analyzing-ransomware-leak-site-intelligence |
| description | 监控和分析勒索软件组织的数据泄露站点(DLS),追踪受害者发布情况,提取组织战术的威胁情报,并评估特定行业的勒索软件风险以实现主动防御。 |
| domain | cybersecurity |
| subdomain | threat-intelligence |
| tags | ["ransomware","leak-site","data-leak","extortion","threat-intelligence","monitoring","dls","victim-tracking"] |
| version | 1.0 |
| author | mahipal |
| license | Apache-2.0 |
分析勒索软件数据泄露站点情报
概述
采用双重勒索模式运营的勒索软件(Ransomware)组织在 Tor 隐藏服务上维护数据泄露站点(DLS),在那里发布受害者名称、被盗数据样本和倒计时器以施压付款。2025 年上半年,96 个独特勒索软件组织活跃,每月约发布 535 名受害者。监控这些站点提供了关于活跃威胁组织、目标行业、地理模式和新兴勒索软件家族的情报。本技能涵盖安全收集 DLS 情报、提取结构化数据、追踪组织活动趋势,以及生成行业特定风险评估。
前置条件
- Python 3.9+,安装
requests、beautifulsoup4、pandas、matplotlib 库
- Tor 代理(SOCKS5)用于访问 .onion 站点,或商业 DLS 监控情报
- 了解勒索软件双重勒索商业模式
- 熟悉主要勒索软件家族(Qilin、Akira、LockBit、BlackCat、Clop)
- 访问勒索软件追踪情报(Ransomwatch、RansomLook、DarkFeed)
核心概念
双重勒索模式
现代勒索软件组织在加密受害者数据之前还会将其外泄(Exfiltration)。泄露站点作为公开施压工具:受害者以倒计时器、部分数据样本和文件目录的形式被列出。若未支付赎金,完整数据将被公开。部分组织已转向三重勒索,追加 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)} 个勒索软件组织画像")
.posts
():
cutoff = datetime.now() - timedelta(days=days)
recent = []
post .posts:
:
discovered = datetime.fromisoformat(
post.get(, ).replace(, )
)
discovered.replace(tzinfo=) >= cutoff:
recent.append(post)
(ValueError, TypeError):
()
recent
():
group_posts = [
p p .posts
p.get(, ).lower() == group_name.lower()
]
()
group_posts
collector = RansomwareIntelCollector()
collector.fetch_ransomwatch_data()
recent = collector.get_recent_victims(days=)
步骤 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()
()
()
group, count analysis[]:
()
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= x: -x[][]):
pct = (data[] / total * ) total >
risk_assessment[][sector] = {
: data[],
: (pct, ),
: data[].most_common(),
: (
pct >
pct >
pct >
),
}
country, data (country_data.items(), key= x: -x[][]):
pct = (data[] / total * ) total >
risk_assessment[][country] = {
: data[],
: (pct, ),
: data[].most_common(),
}
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 / ),
)
()
group, info (new_groups.items(), key= x: -x[][]):
(
)
new_groups
new_groups = track_new_groups(collector.posts, lookback_days=)
步骤 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['total_victims']} 名受害者,首次出现于 {info['first_seen'].strftime('%Y-%m-%d')}\n"
report += "\n## 行业风险评估\n"
report += "| 行业 | 受害者数 | 占比 | 风险级别 |\n|--------|---------|---|------------|\n"
for sector, data in list(risk["sectors"].items())[:10]:
report += f"| {sector} | {data['victim_count']} | % | |\n"
report +=
(, ) f:
f.write(report)
()
report
generate_ransomware_intel_report(trends, risk, new_groups)
验收标准
- 从公开追踪情报成功导入勒索软件受害者数据
- 分析组织活动趋势(含月度分类)
- 生成行业和地理风险评估
- 识别新兴组织并提供活动指标
- 生成包含可执行建议的情报报告
- 所有收集工作通过经授权的公开来源进行
参考资料