performing-dark-web-monitoring-for-threats

针对威胁执行暗网监控

概述

暗网监控涉及系统地扫描 Tor 隐藏服务、地下论坛、粘贴站点和暗网市场，以识别针对组织的威胁，包括泄露的凭据、数据泄露、威胁行为者讨论、漏洞利用工具和计划中的攻击。本技能涵盖搭建监控基础设施、使用基于 Tor 的采集工具、为品牌提及和凭据泄露实现自动化告警，以及分析暗网情报以获取可操作的威胁指标。

何时使用

• 在执行涉及针对威胁进行暗网监控的安全评估时
• 在针对相关安全事件遵循事件响应流程时
• 在执行计划内的安全测试或审计活动时
• 在通过动手测试验证安全控制措施时

前提条件

• Tor Browser 和 Tor 代理（端口 9050 上的 SOCKS5）
• 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）
• 人工 HUMINT：由分析师驱动对特定论坛和市场进行研究
• 凭据监控：针对泄露凭据的泄露数据库和粘贴站点监控

暗网研究的 OPSEC

• 使用不含个人数据的专用 VM
• 将所有流量通过 Tor 路由（推荐 Whonix 或 Tails）
• 绝不使用个人账户或可识别信息
• 为论坛注册使用单独的电子邮件地址和角色身份
• 在 Tor Browser 中禁用 JavaScript 以增强安全性
• 绝不在生产系统上下载或执行来自暗网来源的文件

工作流

步骤 1：搭建基于 Tor 的 HTTP 客户端

import requests
from requests.adapters import HTTPAdapter

def create_tor_session():
    """Create a requests session routed through Tor SOCKS5 proxy."""
    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):
    """Verify that traffic is routed through 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):
    """Monitor paste sites for leaked credentials matching organization domains."""
    findings = []

    # Check Have I Been Pwned API (clearnet)
    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"[-] HIBP error for {domain}: {e}")

    return findings


def search_for_keywords(session, keywords, onion_paste_urls):
    """Search dark web paste sites for specific keywords."""
    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"[-] Error fetching {paste_url}: {e}")

    return results


def extract_context(text, keyword, context_chars=200):
    """Extract text context around a keyword match."""
    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):
    """Check known ransomware group leak sites for organization mentions."""
    # Use Ransomwatch API (clearnet aggregator of ransomware leak sites)
    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 error: {e}")
    return []

步骤 4：生成暗网情报报告

def generate_dark_web_report(findings, organization):
    """Generate structured dark web intelligence report."""
    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)

    # Generate executive summary
    cred_count = len(report["credential_leaks"])
    ransom_count = len(report["ransomware_mentions"])
    report["executive_summary"] = (
        f"Monitoring identified {cred_count} credential leak sources "
        f"and {ransom_count} ransomware group mentions for {organization}."
    )

    if ransom_count > 0:
        report["recommendations"].append(
            "CRITICAL: Organization mentioned on ransomware leak site. "
            "Initiate incident response immediately."
        )
    if cred_count > 0:
        report["recommendations"].append(
            "HIGH: Leaked credentials detected. Force password resets for "
            "affected accounts and enable MFA."
        )

    return report

验证标准

• 已建立 Tor 连接，并通过 check.torproject.org 验证
• 凭据泄露监控从 HIBP 和粘贴站点返回结果
• 勒索软件泄露站点监控识别出相关提及
• 生成带有可操作建议的暗网情报报告
• 所有监控均在法律和道德边界内执行
• 保持 OPSEC：无个人或企业身份暴露

参考资料

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