---

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 = []

    # 检查 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 和粘贴站点返回结果
• 勒索软件泄露站点监控识别相关提及
• 暗网情报报告生成并含可操作建议
• 所有监控在法律和道德范围内执行
• 维护操作安全：无个人或企业身份暴露

参考资料

• Tor 项目
• Have I Been Pwned API
• Ransomwatch
• DarkOwl
• Intel 471
• Flare Systems