DNSTwistを使用したタイポスクワッティングドメインの分析
概要
DNSTwistは、タイポスクワッティング、ホモグラフフィッシング攻撃、ブランド冒充を検出するために、類似したドメイン名を生成するドメイン名パーミュテーションエンジンです。文字置換、転置、挿入、省略、ホモグリフ置換などの手法を使用して数千のドメインバリエーションを生成し、DNSレコード（A、AAAA、NS、MX）を確認し、ファジーハッシュ（ssdeep）と知覚ハッシュ（pHash）を使用してWebページ類似度を計算し、悪意のある可能性のある登録済みドメインを特定します。
使用すべきタイミング
- タイポスクワッティングドメインをdnstwistで分析する必要があるセキュリティインシデントを調査する場合
- このドメインに対する検出ルールや脅威ハンティングクエリを構築する場合
- SOCアナリストがこの分析タイプの構造化された手順を必要とする場合
- 関連する攻撃手法に対するセキュリティ監視範囲を検証する場合
前提条件
- Python 3.9+ と
dnstwistがインストールされていること（pip install dnstwist[full]） - オプション: IPジオロケーション用のGeoIPデータベース
- オプション: エンリッチメント用のShodan APIキー
- DNSクエリを実行するためのネットワークアクセス
- DNSレコードタイプとドメイン登録に関する理解
主要な概念
ドメインパーミュテーション手法
DNSTwistは、追加（文字の付加）、ビットスクワッティング（ビット反転エラー）、ホモグリフ（視覚的に類似したUnicode文字、例: rn 対 m）、ハイフネーション（ハイフンの追加）、挿入（文字の挿入）、省略（文字の削除）、繰り返し（文字の繰り返し）、置換（隣接するキーボードキーとの置換）、サブドメイン（ドットの挿入）、転置（隣接する文字の入れ替え）、母音スワップ（母音の入れ替え）、および辞書ベース（一般的な単語の追加）を使用してパーミュテーションを生成します。
ファジーハッシュと視覚的類似性
DNSTwistはssdeep（局所性依存ハッシュ）を使用してHTMLコンテンツを比較し、pHash（知覚ハッシュ）を使用してWebページのスクリーンショットを比較します。これにより、正規サイトを視覚的に模倣したクローンフィッシングサイトを特定するのに役立ちます。類似度スコアが高いほど、フィッシングページである可能性が高くなります。
検出ワークフロー
典型的なワークフローは次のとおりです。ドメインパーミュテーションの生成 -> DNSレコードの解決 -> 登録済みドメインの確認 -> Webページ類似度の比較 -> 疑わしいドメインのフラグ付け -> セキュリティチームへのアラート -> テイクダウン要求。一般的な企業ドメインの場合、dnstwistは5,000〜10,000のバリエーションを生成します。
ワークフロー
ステップ 1: 基本的なドメインパーミュテーションスキャン
import subprocess
import json
import csv
from datetime import datetime
def run_dnstwist_scan(domain, output_file=None):
"""ターゲットドメインに対してdnstwistスキャンを実行します。"""
cmd = [
"dnstwist",
"--registered", # 登録済みドメインのみ表示
"--format", "json", # JSONで出力
"--nameservers", "8.8.8.8,1.1.1.1",
"--threads", "50",
"--mxcheck", # MXレコードを確認
"--ssdeep", # ファジーハッシュ比較
"--geoip", # GeoIPルックアップ
domain,
]
print(f"[*] パーミュテーションのスキャン中: {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"[+] {len(registered)} 件の登録済み類似ドメインを発見しました")
if output_file:
with open(output_file, "w") as f:
json.dump(registered, f, indent=2)
print(f"[+] 結果を {output_file} に保存しました")
return registered
else:
print(f"[-] dnstwist エラー: {result.stderr}")
return []
results = run_dnstwist_scan("example.com", "typosquat_results.json")
ステップ 2: 結果の分析と優先順位付け
def analyze_results(results, legitimate_ips=None):
"""dnstwistの結果を分析し、脅威の優先順位を付けます。"""
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 = []
# 正規サイトとの類似度が高い
if ssdeep_score and ssdeep_score > 50:
risk_score += 40
risk_factors.append(f"Web類似度が高い ({ssdeep_score}%)")
# MXレコードがある（メール受信/フィッシングが可能）
if dns_mx:
risk_score += 20
risk_factors.append("MXレコードあり（メール送受信可能）")
# 最近登録された（whoisデータがある場合）
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"最近登録 ({age_days} 日前)")
elif age_days < 90:
risk_score += 15
risk_factors.append(f"登録から {age_days} 日経過")
except (ValueError, TypeError):
pass
# ホモグリフ攻撃は最もリスクが高い
if fuzzer == "homoglyph":
risk_score += 25
risk_factors.append("ホモグリフ（視覚的に同一）")
elif fuzzer in ("addition", "replacement", "transposition"):
risk_score += 10
risk_factors.append(f"パーミュテーションタイプ: {fuzzer}")
# 正規のインフラストラクチャを指していない
if dns_a and not set(dns_a).intersection(legitimate_ips):
risk_score += 10
risk_factors.append("正規のIPとは異なる")
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=== タイポスクワッティング分析 ===")
print(f"高リスク: {len(high_risk)}")
print(f"中リスク: {len(medium_risk)}")
print(f"低リスク: {len(low_risk)}")
if high_risk:
print(f"\n--- 高リスクドメイン ---")
for entry in high_risk[:10]:
print(f" {entry['domain']} (スコア: {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):
"""監視対象の全ドメインをスキャンし、新しいタイポスクワットを検出します。"""
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" [新規] {domain_key} ({entry.get('fuzzer', '')})")
self._save_known()
print(f"\n[+] 新たなタイポスクワッティングドメイン: {len(new_findings)} 件")
return new_findings
def generate_alert(self, findings):
"""新たな高リスクタイポスクワッティングドメインのアラートを生成します。"""
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"):
"""高リスクドメインをファイアウォール/プロキシ用のブロックリストとしてエクスポートします。"""
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"# タイポスクワッティングブロックリスト {datetime.now().isoformat()} 生成\n")
for d in sorted(set(domains)):
f.write(f"{d}\n")
print(f"[+] ブロックリスト保存: {len(domains)} ドメイン -> {output_file}")
return domains
def generate_takedown_report(high_risk_domains):
"""テイクダウン要求レポートを生成します。"""
report = f"""# ドメインテイクダウン要求
生成日時: {datetime.now().isoformat()}
## サマリー
{len(high_risk_domains)} 件のドメインが潜在的なタイポスクワッティング/フィッシングとして特定されました。
## テイクダウンが必要なドメイン
"""
for entry in high_risk_domains:
report += f"""
### {entry['domain']}
- **パーミュテーションタイプ**: {entry.get('fuzzer', '不明')}
- **IPアドレス**: {', '.join(entry.get('dns_a', ['N/A']))}
- **MXレコード**: {', '.join(entry.get('dns_mx', ['N/A']))}
- **リスクスコア**: {entry.get('risk_score', 0)}
- **リスク要因**: {'; '.join(entry.get('risk_factors', []))}
- **Web類似度**: {entry.get('ssdeep_score', 'N/A')}%
"""
with open("takedown_report.md", "w") as f:
f.write(report)
print("[+] テイクダウンレポートを生成しました: takedown_report.md")
export_blocklist(analysis)
generate_takedown_report(analysis["high"])
検証基準
- DNSTwistがターゲットドメインのドメインパーミュテーションを生成すること
- DNS解決により登録済み類似ドメインが特定されること
- Web類似度スコアリングによりクローンフィッシングページが検出されること
- リスクスコアリングにより脅威レベルに基づいてドメインが優先順位付けされること
- 継続的監視により新たに登録されたタイポスクワットが検出されること
- ブロックリストとテイクダウンレポートが正しく生成されること