performing-dark-web-monitoring-for-threats

脅威を対象としたダークウェブのモニタリング

概要

ダークウェブのモニタリングでは、Tor の隠しサービス、アンダーグラウンドフォーラム、ペーストサイト、ダークウェブのマーケットプレイスを体系的にスキャンし、組織を標的とする脅威を特定します。これには、漏えいした認証情報、データ侵害、脅威アクターの議論、脆弱性の悪用ツール、計画中の攻撃が含まれます。このスキルでは、モニタリングインフラストラクチャの設定、Tor ベースの収集ツールの使用、ブランドへの言及や認証情報の漏えいに関する自動アラートの実装、実行可能な脅威指標を得るためのダークウェブインテリジェンスの分析について説明します。

使用する場面

• 脅威を対象としたダークウェブのモニタリングを含むセキュリティ評価を実施するとき
• 関連するセキュリティイベントに対するインシデント対応手順に従うとき
• 定期的なセキュリティテストや監査活動を実施するとき
• 実践的なテストを通じてセキュリティ制御を検証するとき

前提条件

• Tor Browser と Tor プロキシ（ポート 9050 の SOCKS5）
• requests、stem、beautifulsoup4、stix2 ライブラリを備えた Python 3.9+
• Tor の隠しサービスのアーキテクチャ（.onion ドメイン）に関する理解
• ダークウェブモニタリングサービス（Flare、SpyCloud、DarkOwl、Intel 471）の API アクセス
• ダークウェブ調査に関する法的・倫理的な境界の認識
• ダークウェブの閲覧に使用する隔離された 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