| name | performing-dark-web-monitoring-for-threats |
| description | 暗网威胁监控涉及系统性扫描 Tor 隐藏服务、地下论坛、粘贴站点和暗网市场,以识别针对组织的威胁,包括泄露凭据、数据泄露、威胁行为者讨论、漏洞利用工具和预谋攻击。 |
| domain | cybersecurity |
| subdomain | threat-intelligence |
| tags | ["threat-intelligence","cti","ioc","mitre-attack","stix","dark-web","tor","threat-monitoring"] |
| version | 1.0 |
| author | mahipal |
| license | Apache-2.0 |
执行暗网威胁监控
概述
暗网威胁监控涉及系统性扫描 Tor 隐藏服务、地下论坛、粘贴站点和暗网市场,以识别针对组织的威胁,包括泄露凭据、数据泄露、威胁行为者讨论、漏洞利用工具和预谋攻击。本技能涵盖搭建监控基础设施、使用基于 Tor 的收集工具、为品牌提及和凭据泄露实现自动告警,以及分析暗网情报以获取可操作的威胁指标。
前置条件
- Tor Browser 和 Tor 代理(SOCKS5,端口 9050)
- Python 3.9+,安装
requests、stem、beautifulsoup4、stix2 库
- 了解 Tor 隐藏服务架构(.onion 域)
- 暗网监控服务 API 访问权限(Flare、SpyCloud、DarkOwl、Intel 471)
- 了解暗网研究的法律和道德边界
- 用于暗网浏览的隔离 VM(避免个人或企业身份泄露)
核心概念
暗网情报来源
- 地下论坛:威胁行为者讨论 TTP、出售漏洞利用程序和共享工具的黑客论坛
- 粘贴站点:用于共享被盗数据、凭据和代码片段的平台
- 市场:出售被盗数据、RaaS、漏洞利用套件和访问权限的暗网市场
- Telegram/Discord:网络犯罪团伙的替代通信渠道
- 勒索软件泄露站点:勒索软件组织发布受害者被盗数据的博客
收集方法
- 自动化爬取:扫描隐藏服务的基于 Tor 的网络爬虫
- 基于 API 的监控:商业暗网监控 API(Flare、DarkOwl、Intel 471)
- 手动人力情报:分析师主导的特定论坛和市场研究
- 凭据监控:泄露数据库和粘贴站点监控,检测泄露凭据
暗网研究的操作安全
- 使用专用 VM,不保留个人数据
- 所有流量通过 Tor 路由(推荐使用 Whonix 或 Tails)
- 切勿使用个人账号或可识别信息
- 论坛注册使用独立的电子邮件地址和化名
- 在 Tor Browser 中禁用 JavaScript 以增强安全性
- 切勿在生产系统上下载或执行来自暗网来源的文件
操作步骤
步骤 1:搭建基于 Tor 的 HTTP 客户端
import requests
from requests.adapters import HTTPAdapter
def create_tor_session():
"""创建通过 Tor SOCKS5 代理路由的 requests 会话。"""
session = requests.Session()
session.proxies = {
"http": "socks5h://127.0.0.1:9050",
"https": "socks5h://127.0.0.1:9050",
}
session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; rv:109.0) Gecko/20100101 Firefox/115.0",
})
return session
def verify_tor_connection(session):
"""验证流量是否通过 Tor 路由。"""
try:
resp = session.get("https://check.torproject.org/api/ip", timeout=30)
data = resp.json()
return {
"is_tor": data.get("IsTor", False),
"ip": data.get("IP", ""),
}
except Exception as e:
return {"error": str(e)}
步骤 2:监控粘贴站点上的凭据泄露
import re
from datetime import datetime
def monitor_paste_sites(session, organization_domains):
"""监控粘贴站点,查找与组织域名匹配的泄露凭据。"""
findings = []
for domain in organization_domains:
try:
resp = requests.get(
f"https://haveibeenpwned.com/api/v3/breaches",
headers={"hibp-api-key": "YOUR_HIBP_KEY"},
timeout=30,
)
if resp.status_code == 200:
breaches = resp.json()
for breach in breaches:
if domain.lower() in breach.get("Domain", "").lower():
findings.append({
"source": "HIBP",
"breach_name": breach["Name"],
"breach_date": breach.get("BreachDate"),
"data_classes": breach.get("DataClasses", []),
"pwn_count": breach.get("PwnCount", 0),
"domain": domain,
})
except Exception as e:
print(f"[-] {domain} 的 HIBP 查询出错:{e}")
return findings
():
results = []
paste_url onion_paste_urls:
:
resp = session.get(paste_url, timeout=)
resp.status_code == :
content = resp.text.lower()
keyword keywords:
keyword.lower() content:
results.append({
: paste_url,
: keyword,
: datetime.utcnow().isoformat(),
: extract_context(content, keyword.lower()),
})
Exception e:
()
results
():
idx = text.find(keyword)
idx == -:
start = (, idx - context_chars)
end = ((text), idx + (keyword) + context_chars)
text[start:end]
步骤 3:监控勒索软件泄露站点
def check_ransomware_leak_sites(session, organization_name):
"""检查已知勒索软件组织泄露站点是否有组织提及。"""
try:
resp = requests.get(
"https://raw.githubusercontent.com/joshhighet/ransomwatch/main/posts.json",
timeout=30,
)
if resp.status_code == 200:
posts = resp.json()
matches = []
for post in posts:
post_title = post.get("post_title", "").lower()
if organization_name.lower() in post_title:
matches.append({
"group": post.get("group_name", ""),
"title": post.get("post_title", ""),
"discovered": post.get("discovered", ""),
"url": post.get("post_url", ""),
})
return matches
except Exception as e:
print(f"[-] Ransomwatch 查询出错:{e}")
return []
步骤 4:生成暗网情报报告
def generate_dark_web_report(findings, organization):
"""生成结构化暗网情报报告。"""
report = {
"organization": organization,
"report_date": datetime.utcnow().isoformat(),
"executive_summary": "",
"credential_leaks": [],
"ransomware_mentions": [],
"dark_web_mentions": [],
"recommendations": [],
}
for finding in findings:
if finding.get("source") == "HIBP":
report["credential_leaks"].append(finding)
elif finding.get("group"):
report["ransomware_mentions"].append(finding)
else:
report["dark_web_mentions"].append(finding)
cred_count = len(report["credential_leaks"])
ransom_count = len(report["ransomware_mentions"])
report["executive_summary"] = (
f"监控发现 {organization} 存在 {cred_count} 个凭据泄露来源"
f"和 {ransom_count} 个勒索软件组织提及。"
)
if ransom_count > 0:
report["recommendations"].append(
"严重:组织在勒索软件泄露站点被提及。"
"立即启动事件响应。"
)
if cred_count > 0:
report["recommendations"].append(
"高危:检测到泄露凭据。强制受影响账户重置密码"
)
report
验收标准
- Tor 连接已建立并通过 check.torproject.org 验证
- 凭据泄露监控从 HIBP 和粘贴站点返回结果
- 勒索软件泄露站点监控识别相关提及
- 暗网情报报告生成并含可操作建议
- 所有监控在法律和道德范围内执行
- 维护操作安全:无个人或企业身份暴露
参考资料