killvxk 执行暗网威胁监控
概述
暗网威胁监控涉及系统性扫描 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 = []
# 检查 Have I Been Pwned API（明网）
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
def search_for_keywords(session, keywords, onion_paste_urls):
"""在暗网粘贴站点中搜索特定关键词。"""
results = []
for paste_url in onion_paste_urls:
try:
resp = session.get(paste_url, timeout=60)
if resp.status_code == 200:
content = resp.text.lower()
for keyword in keywords:
if keyword.lower() in content:
results.append({
"url": paste_url,
"keyword": keyword,
"timestamp": datetime.utcnow().isoformat(),
"snippet": extract_context(content, keyword.lower()),
})
except Exception as e:
print(f"[-] 获取 {paste_url} 时出错：{e}")
return results
def extract_context(text, keyword, context_chars=200):
"""提取关键词匹配周围的文本上下文。"""
idx = text.find(keyword)
if idx == -1:
return ""
start = max(0, idx - context_chars)
end = min(len(text), idx + len(keyword) + context_chars)
return text[start:end]
步骤 3：监控勒索软件泄露站点
def check_ransomware_leak_sites(session, organization_name):
"""检查已知勒索软件组织泄露站点是否有组织提及。"""
# 使用 Ransomwatch API（勒索软件泄露站点的明网聚合器）
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(
"高危：检测到泄露凭据。强制受影响账户重置密码"
"并启用 MFA。"
)
return report
验收标准
- Tor 连接已建立并通过 check.torproject.org 验证
- 凭据泄露监控从 HIBP 和粘贴站点返回结果
- 勒索软件泄露站点监控识别相关提及
- 暗网情报报告生成并含可操作建议
- 所有监控在法律和道德范围内执行
- 维护操作安全：无个人或企业身份暴露