analyzing-typosquatting-domains-with-dnstwist

DNSTwist によるタイポスクワッティングドメインの分析

概要

DNSTwist は、タイポスクワッティング、ホモグラフ型フィッシング攻撃、ブランドのなりすましを検出するために、類似したドメイン名を生成するドメイン名の順列エンジンです。文字の置換、入れ替え、挿入、省略、同形異字への置換などの手法で数千件のドメインの順列を作成し、DNS レコード（A、AAAA、NS、MX）を確認します。また、ファジーハッシュ（ssdeep）と知覚ハッシュ（pHash）を使ってウェブページの類似度を算出し、悪意のある可能性がある登録済みドメインを特定します。

使用する場面

• dnstwist でタイポスクワッティングドメインを分析する必要があるセキュリティインシデントを調査する場合
• このドメインに関する検出ルールや脅威ハンティングクエリを作成する場合
• SOC アナリストが、この種の分析に関する体系的な手順を必要とする場合
• 関連する攻撃手法に対するセキュリティ監視の対応状況を検証する場合

前提条件

• dnstwist をインストールした Python 3.9 以降（pip install dnstwist[full]）
• オプション：IP ジオロケーション用の GeoIP データベース
• オプション：エンリッチメント用の Shodan API キー
• DNS クエリを実行できるネットワークアクセス
• DNS レコードタイプとドメイン登録に関する知識

主な概念

ドメインの順列生成手法

DNSTwist は、追加（文字の追加）、ビットスクワッティング（ビット反転エラー）、ホモグリフ（rn と m のような見た目が似た Unicode 文字）、ハイフンの追加、挿入（文字の挿入）、省略（文字の削除）、反復（文字の繰り返し）、置換（隣接するキーボードキーへの置き換え）、サブドメイン（ドットの挿入）、入れ替え（隣接する文字の交換）、母音の入れ替え、辞書ベース（一般的な単語の追加）などの手法で順列を生成します。

ファジーハッシュと視覚的類似性

DNSTwist は、HTML コンテンツの比較に ssdeep（局所性鋭敏型ハッシュ）を、ウェブページのスクリーンショットの比較に pHash（知覚ハッシュ）を使用します。これにより、正規サイトを視覚的に模倣する複製フィッシングサイトを特定できます。類似度スコアが高い場合、フィッシングページである可能性が高いことを示します。

検出ワークフロー

一般的なワークフローは、ドメインの順列を生成 -> DNS レコードを解決 -> 登録済みドメインを確認 -> ウェブページの類似度を比較 -> 不審なドメインにフラグを付ける -> セキュリティチームに警告する -> テイクダウンを依頼する、という流れです。一般的な企業ドメインの場合、dnstwist は 5,000～10,000 件の順列を生成します。

ワークフロー

ステップ 1：基本的なドメイン順列スキャン

import subprocess
import json
import csv
from datetime import datetime

def run_dnstwist_scan(domain, output_file=None):
    """Run dnstwist scan against a target domain."""
    cmd = [
        "dnstwist",
        "--registered",     # Only show registered domains
        "--format", "json", # Output in JSON
        "--nameservers", "8.8.8.8,1.1.1.1",
        "--threads", "50",
        "--mxcheck",        # Check MX records
        "--ssdeep",         # Fuzzy hash comparison
        "--geoip",          # GeoIP lookup
        domain,
    ]

    print(f"[*] Scanning permutations for: {domain}")
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=600)

    if result.returncode == 0:
        results = json.loads(result.stdout)
        registered = [r for r in results if r.get("dns_a") or r.get("dns_aaaa")]
        print(f"[+] Found {len(registered)} registered lookalike domains")

        if output_file:
            with open(output_file, "w") as f:
                json.dump(registered, f, indent=2)
            print(f"[+] Results saved to {output_file}")

        return registered
    else:
        print(f"[-] dnstwist error: {result.stderr}")
        return []

results = run_dnstwist_scan("example.com", "typosquat_results.json")

ステップ 2：結果の分析と優先順位付け

def analyze_results(results, legitimate_ips=None):
    """Analyze dnstwist results and prioritize threats."""
    legitimate_ips = legitimate_ips or set()
    high_risk = []
    medium_risk = []
    low_risk = []

    for entry in results:
        domain = entry.get("domain", "")
        fuzzer = entry.get("fuzzer", "")
        dns_a = entry.get("dns_a", [])
        dns_mx = entry.get("dns_mx", [])
        ssdeep_score = entry.get("ssdeep_score", 0)

        risk_score = 0
        risk_factors = []

        # High similarity to legitimate site
        if ssdeep_score and ssdeep_score > 50:
            risk_score += 40
            risk_factors.append(f"high web similarity ({ssdeep_score}%)")

        # Has MX records (can receive email / phishing)
        if dns_mx:
            risk_score += 20
            risk_factors.append("has MX records (email capable)")

        # Recently registered (if whois data available)
        whois_created = entry.get("whois_created", "")
        if whois_created:
            try:
                created = datetime.fromisoformat(whois_created.replace("Z", "+00:00"))
                age_days = (datetime.now(created.tzinfo) - created).days
                if age_days < 30:
                    risk_score += 30
                    risk_factors.append(f"recently registered ({age_days} days)")
                elif age_days < 90:
                    risk_score += 15
                    risk_factors.append(f"registered {age_days} days ago")
            except (ValueError, TypeError):
                pass

        # Homoglyph attacks are highest risk
        if fuzzer == "homoglyph":
            risk_score += 25
            risk_factors.append("homoglyph (visually identical)")
        elif fuzzer in ("addition", "replacement", "transposition"):
            risk_score += 10
            risk_factors.append(f"permutation type: {fuzzer}")

        # Not pointing to legitimate infrastructure
        if dns_a and not set(dns_a).intersection(legitimate_ips):
            risk_score += 10
            risk_factors.append("different IP from legitimate")

        entry["risk_score"] = risk_score
        entry["risk_factors"] = risk_factors

        if risk_score >= 50:
            high_risk.append(entry)
        elif risk_score >= 25:
            medium_risk.append(entry)
        else:
            low_risk.append(entry)

    high_risk.sort(key=lambda x: x["risk_score"], reverse=True)
    medium_risk.sort(key=lambda x: x["risk_score"], reverse=True)

    print(f"\n=== Typosquatting Analysis ===")
    print(f"High Risk: {len(high_risk)}")
    print(f"Medium Risk: {len(medium_risk)}")
    print(f"Low Risk: {len(low_risk)}")

    if high_risk:
        print(f"\n--- High Risk Domains ---")
        for entry in high_risk[:10]:
            print(f"  {entry['domain']} (score: {entry['risk_score']})")
            for factor in entry['risk_factors']:
                print(f"    - {factor}")

    return {"high": high_risk, "medium": medium_risk, "low": low_risk}

analysis = analyze_results(results, legitimate_ips={"93.184.216.34"})

ステップ 3：継続的な監視パイプライン

import time
import hashlib

class TyposquatMonitor:
    def __init__(self, domains, known_domains_file="known_typosquats.json"):
        self.domains = domains
        self.known_file = known_domains_file
        self.known_domains = self._load_known()

    def _load_known(self):
        try:
            with open(self.known_file, "r") as f:
                return json.load(f)
        except FileNotFoundError:
            return {}

    def _save_known(self):
        with open(self.known_file, "w") as f:
            json.dump(self.known_domains, f, indent=2)

    def scan_all_domains(self):
        """Scan all monitored domains for new typosquats."""
        new_findings = []
        for domain in self.domains:
            results = run_dnstwist_scan(domain)
            for entry in results:
                domain_key = entry.get("domain", "")
                if domain_key not in self.known_domains:
                    entry["first_seen"] = datetime.now().isoformat()
                    entry["monitored_domain"] = domain
                    self.known_domains[domain_key] = entry
                    new_findings.append(entry)
                    print(f"  [NEW] {domain_key} ({entry.get('fuzzer', '')})")

        self._save_known()
        print(f"\n[+] New typosquatting domains found: {len(new_findings)}")
        return new_findings

    def generate_alert(self, findings):
        """Generate alert for new high-risk typosquatting domains."""
        analysis = analyze_results(findings)
        alerts = []
        for entry in analysis["high"]:
            alerts.append({
                "severity": "HIGH",
                "domain": entry["domain"],
                "target": entry.get("monitored_domain", ""),
                "risk_score": entry["risk_score"],
                "risk_factors": entry["risk_factors"],
                "dns_a": entry.get("dns_a", []),
                "dns_mx": entry.get("dns_mx", []),
                "timestamp": datetime.now().isoformat(),
            })
        return alerts

monitor = TyposquatMonitor(["mycompany.com", "mycompany.org"])
new_findings = monitor.scan_all_domains()
alerts = monitor.generate_alert(new_findings)

ステップ 4：ブロックリストとテイクダウン用のエクスポート

def export_blocklist(analysis, output_file="blocklist.txt"):
    """Export high-risk domains as blocklist for firewall/proxy."""
    domains = []
    for entry in analysis["high"] + analysis["medium"]:
        domain = entry.get("domain", "")
        if domain:
            domains.append(domain)

    with open(output_file, "w") as f:
        f.write(f"# Typosquatting blocklist generated {datetime.now().isoformat()}\n")
        for d in sorted(set(domains)):
            f.write(f"{d}\n")

    print(f"[+] Blocklist saved: {len(domains)} domains -> {output_file}")
    return domains

def generate_takedown_report(high_risk_domains):
    """Generate takedown request report."""
    report = f"""# Domain Takedown Request
Generated: {datetime.now().isoformat()}

## Summary
{len(high_risk_domains)} domains identified as potential typosquatting/phishing.

## Domains Requiring Takedown
"""
    for entry in high_risk_domains:
        report += f"""
### {entry['domain']}
- **Permutation Type**: {entry.get('fuzzer', 'unknown')}
- **IP Address**: {', '.join(entry.get('dns_a', ['N/A']))}
- **MX Records**: {', '.join(entry.get('dns_mx', ['N/A']))}
- **Risk Score**: {entry.get('risk_score', 0)}
- **Risk Factors**: {'; '.join(entry.get('risk_factors', []))}
- **Web Similarity**: {entry.get('ssdeep_score', 'N/A')}%
"""
    with open("takedown_report.md", "w") as f:
        f.write(report)
    print("[+] Takedown report generated: takedown_report.md")

export_blocklist(analysis)
generate_takedown_report(analysis["high"])

検証基準

• DNSTwist が対象ドメインの順列を生成する
• DNS 解決によって登録済みの類似ドメインを特定する
• ウェブ類似度のスコアリングによって複製フィッシングページを検出する
• リスクスコアリングによって脅威レベルに応じてドメインの優先順位を付ける
• 継続的な監視によって新たに登録されたタイポスクワッティングドメインを検出する
• ブロックリストとテイクダウンのレポートが正しく生成される

参考資料

• dnstwist GitHub リポジトリ
• dnstwister オンラインサービス
• HawkEye：DNSTwist でタイポスクワッティングを検出する
• Darktrace：タイポスクワッティングドメインの監視
• Security Risk Advisors：ドメインの監視
• Conscia：タイポスクワッティングを検出する方法