フィッシング対策のための証明書透明性分析 は何をしますか？
概要
証明書透明性（Certificate Transparency、CT）は、発行されたすべての SSL/TLS 証明書の公開・追加専用ログを作成するインターネットセキュリティ標準です。CTログを監視することで、正規ブランドを模倣したフィッシングドメインの登録、自社ドメインに対する不正な証明書発行、証明書ベースの攻撃インフラを早期に検出できます。このスキルでは、crt.sh を介した CT ログのクエリ、Certstream によるリアルタイム監視、不審な証明書に対する自動アラートの構築、および調査結果を脅威インテリジェンスワークフローに統合する方法を網羅します。
使用すべきケース
- フィッシングに関連するセキュリティインシデントの調査において、証明書透明性の分析が必要な場合
- このドメインに関する検知ルールや脅威ハンティングクエリを構築する場合
- SOCアナリストがこの分析タイプの構造化された手順を必要とする場合
- 関連する攻撃手法に対するセキュリティ監視のカバレッジを検証する場合
前提条件
- Python 3.9+ と
requests、certstream、tldextract、Levenshteinライブラリ - 過去の CT ログクエリのための crt.sh (https://crt.sh/) へのアクセス
- リアルタイム監視のための Certstream (https://certstream.calidog.io/)
- 監視対象の組織ドメインとブランドキーワードのリスト
- SSL/TLS 証明書の構造と発行プロセスに関する理解
主要な概念
証明書透明性ログ
CTログは、暗号学的に保証され、公開監査可能な、TLS証明書発行の追加専用レコードです。主要な認証局（Let's Encrypt、DigiCert、Sectigo、Google Trust Services）は、発行したすべての証明書を複数のCTログに提出します。2025年時点で、Chrome と Safari はすべての公開信頼証明書に CT を必須としています。
CT によるフィッシング検出
攻撃者は類似ドメインを登録し、無料の証明書（しばしば Let's Encrypt から）を取得して、フィッシングサイトを HTTPS で正規に見せかけます。CT 監視は、証明書がフィッシングキャンペーンの開始前にログに現れるため、これらを早期に検出し、プロアクティブなブロックの機会を提供します。
crt.sh データベース
crt.sh は、Sectigo が運用する無料の Web インターフェースと PostgreSQL データベースで、CT ログをインデックス化しています。ワイルドカード検索 (%.example.com)、直接 SQL クエリ、JSON API レスポンスをサポートしています。すべての主要な CT ログにわたる証明書の発行、有効期限、失効を追跡します。
ワークフロー
ステップ1: crt.sh で証明書履歴を照会する
import requests
import json
from datetime import datetime
import tldextract
class CTLogMonitor:
CRT_SH_URL = "https://crt.sh"
def __init__(self, monitored_domains, brand_keywords):
self.monitored_domains = monitored_domains
self.brand_keywords = [k.lower() for k in brand_keywords]
def query_crt_sh(self, domain, include_expired=False):
"""Query crt.sh for certificates matching a domain."""
params = {
"q": f"%.{domain}",
"output": "json",
}
if not include_expired:
params["exclude"] = "expired"
resp = requests.get(self.CRT_SH_URL, params=params, timeout=30)
if resp.status_code == 200:
certs = resp.json()
print(f"[+] crt.sh: {len(certs)} certificates for *.{domain}")
return certs
return []
def find_suspicious_certs(self, domain):
"""Find certificates that may be phishing attempts."""
certs = self.query_crt_sh(domain)
suspicious = []
for cert in certs:
common_name = cert.get("common_name", "").lower()
name_value = cert.get("name_value", "").lower()
issuer = cert.get("issuer_name", "")
not_before = cert.get("not_before", "")
not_after = cert.get("not_after", "")
# Check for exact domain matches (legitimate)
extracted = tldextract.extract(common_name)
cert_domain = f"{extracted.domain}.{extracted.suffix}"
if cert_domain == domain:
continue # Legitimate certificate
# Flag suspicious patterns
flags = []
if domain.replace(".", "") in common_name.replace(".", ""):
flags.append("contains target domain string")
if any(kw in common_name for kw in self.brand_keywords):
flags.append("contains brand keyword")
if "let's encrypt" in issuer.lower():
flags.append("free CA (Let's Encrypt)")
if flags:
suspicious.append({
"common_name": cert.get("common_name", ""),
"name_value": cert.get("name_value", ""),
"issuer": issuer,
"not_before": not_before,
"not_after": not_after,
"serial": cert.get("serial_number", ""),
"flags": flags,
"crt_sh_id": cert.get("id", ""),
"crt_sh_url": f"https://crt.sh/?id={cert.get('id', '')}",
})
print(f"[+] Found {len(suspicious)} suspicious certificates")
return suspicious
monitor = CTLogMonitor(
monitored_domains=["mycompany.com", "mycompany.org"],
brand_keywords=["mycompany", "mybrand", "myproduct"],
)
suspicious = monitor.find_suspicious_certs("mycompany.com")
for cert in suspicious[:5]:
print(f" [{cert['common_name']}] Flags: {cert['flags']}")
ステップ2: Certstream によるリアルタイム監視
import certstream
import Levenshtein
import re
from datetime import datetime
class CertstreamMonitor:
def __init__(self, watched_domains, brand_keywords, similarity_threshold=0.8):
self.watched_domains = [d.lower() for d in watched_domains]
self.brand_keywords = [k.lower() for k in brand_keywords]
self.threshold = similarity_threshold
self.alerts = []
def start_monitoring(self, max_alerts=100):
"""Start real-time CT log monitoring."""
print("[*] Starting Certstream monitoring...")
print(f" Watching: {self.watched_domains}")
print(f" Keywords: {self.brand_keywords}")
def callback(message, context):
if message["message_type"] == "certificate_update":
data = message["data"]
leaf = data.get("leaf_cert", {})
all_domains = leaf.get("all_domains", [])
for domain in all_domains:
domain_lower = domain.lower().strip("*.")
if self._is_suspicious(domain_lower):
alert = {
"domain": domain,
"all_domains": all_domains,
"issuer": leaf.get("issuer", {}).get("O", ""),
"fingerprint": leaf.get("fingerprint", ""),
"not_before": leaf.get("not_before", ""),
"detected_at": datetime.now().isoformat(),
"reason": self._get_reason(domain_lower),
}
self.alerts.append(alert)
print(f" [ALERT] {domain} - {alert['reason']}")
if len(self.alerts) >= max_alerts:
raise KeyboardInterrupt
try:
certstream.listen_for_events(callback, url="wss://certstream.calidog.io/")
except KeyboardInterrupt:
print(f"\n[+] Monitoring stopped. {len(self.alerts)} alerts collected.")
return self.alerts
def _is_suspicious(self, domain):
"""Check if domain is suspicious relative to watched domains."""
for watched in self.watched_domains:
# Exact keyword match
watched_base = watched.split(".")[0]
if watched_base in domain and domain != watched:
return True
# Levenshtein distance (typosquatting detection)
domain_base = tldextract.extract(domain).domain
similarity = Levenshtein.ratio(watched_base, domain_base)
if similarity >= self.threshold and domain_base != watched_base:
return True
# Brand keyword match
for keyword in self.brand_keywords:
if keyword in domain:
return True
return False
def _get_reason(self, domain):
"""Determine why domain was flagged."""
reasons = []
for watched in self.watched_domains:
watched_base = watched.split(".")[0]
if watched_base in domain:
reasons.append(f"contains '{watched_base}'")
domain_base = tldextract.extract(domain).domain
similarity = Levenshtein.ratio(watched_base, domain_base)
if similarity >= self.threshold and domain_base != watched_base:
reasons.append(f"similar to '{watched}' ({similarity:.0%})")
for kw in self.brand_keywords:
if kw in domain:
reasons.append(f"brand keyword '{kw}'")
return "; ".join(reasons) if reasons else "unknown"
cs_monitor = CertstreamMonitor(
watched_domains=["mycompany.com"],
brand_keywords=["mycompany", "mybrand"],
similarity_threshold=0.75,
)
alerts = cs_monitor.start_monitoring(max_alerts=50)
ステップ3: CTログからサブドメインを列挙する
def enumerate_subdomains_ct(domain):
"""Discover all subdomains from Certificate Transparency logs."""
params = {"q": f"%.{domain}", "output": "json"}
resp = requests.get("https://crt.sh", params=params, timeout=30)
if resp.status_code != 200:
return []
certs = resp.json()
subdomains = set()
for cert in certs:
name_value = cert.get("name_value", "")
for name in name_value.split("\n"):
name = name.strip().lower()
if name.endswith(f".{domain}") or name == domain:
name = name.lstrip("*.")
subdomains.add(name)
sorted_subs = sorted(subdomains)
print(f"[+] CT subdomain enumeration for {domain}: {len(sorted_subs)} subdomains")
return sorted_subs
subdomains = enumerate_subdomains_ct("example.com")
for sub in subdomains[:20]:
print(f" {sub}")
ステップ4: CTインテリジェンスレポートを生成する
def generate_ct_report(suspicious_certs, certstream_alerts, domain):
report = f"""# Certificate Transparency Intelligence Report
## Target Domain: {domain}
## Generated: {datetime.now().isoformat()}
## Summary
- Suspicious certificates found: {len(suspicious_certs)}
- Real-time alerts triggered: {len(certstream_alerts)}
## Suspicious Certificates (crt.sh)
| Common Name | Issuer | Flags | crt.sh Link |
|------------|--------|-------|-------------|
"""
for cert in suspicious_certs[:20]:
flags = "; ".join(cert.get("flags", []))
report += (f"| {cert['common_name']} | {cert['issuer'][:30]} "
f"| {flags} | [View]({cert['crt_sh_url']}) |\n")
report += f"""
## Real-Time Certstream Alerts
| Domain | Issuer | Reason | Detected |
|--------|--------|--------|----------|
"""
for alert in certstream_alerts[:20]:
report += (f"| {alert['domain']} | {alert['issuer']} "
f"| {alert['reason']} | {alert['detected_at'][:19]} |\n")
report += """
## Recommendations
1. Add flagged domains to DNS sinkhole / web proxy blocklist
2. Submit takedown requests for confirmed phishing domains
3. Monitor CT logs continuously for new certificate registrations
4. Implement CAA DNS records to restrict certificate issuance for your domains
5. Deploy DMARC to prevent email spoofing from lookalike domains
"""
with open(f"ct_report_{domain.replace('.','_')}.md", "w") as f:
f.write(report)
print(f"[+] CT report saved")
return report
generate_ct_report(suspicious, alerts if 'alerts' in dir() else [], "mycompany.com")
検証基準
- crt.sh クエリが対象ドメインの証明書データを返すこと
- 類似パターンに基づいて不審な証明書が識別されること
- Certstream のリアルタイム監視が新しいフィッシング証明書を検出すること
- サブドメイン列挙がCTログから包括的なリストを生成すること
- 理由分類付きのアラートが生成されること
- 実行可能な推奨事項を含むCTインテリジェンスレポートが作成されること