analyzing-certificate-transparency-for-phishing

フィッシング対策のための証明書透明性の分析

概要

Certificate Transparency（CT）は、発行されたすべての SSL/TLS 証明書を公開の追記専用ログに記録するインターネットセキュリティ標準です。CT ログを監視することで、正規ブランドを模倣した証明書を登録するフィッシングドメイン、所有ドメインに対する無許可の証明書発行、証明書ベースの攻撃インフラを早期に検出できます。このスキルでは、crt.sh を介した CT ログの照会、Certstream によるリアルタイム監視、疑わしい証明書に対する自動アラートの構築、脅威インテリジェンスのワークフローへの調査結果の統合について説明します。

使用する場面

• フィッシング対策の証明書透明性分析を必要とするセキュリティインシデントを調査する場合
• この分野の検出ルールや脅威ハンティングクエリを構築する場合
• SOC アナリストが、この種の分析に関する体系的な手順を必要とする場合
• 関連する攻撃手法に対するセキュリティ監視範囲を検証する場合

前提条件

• requests、certstream、tldextract、Levenshtein ライブラリを含む Python 3.9 以降
• 過去の CT ログを照会するための crt.sh（https://crt.sh/）へのアクセス
• リアルタイム監視のための Certstream（https://certstream.calidog.io/）
• 監視対象となる組織のドメインとブランドキーワードのリスト
• SSL/TLS 証明書の構造と発行プロセスに関する理解

主な概念

Certificate Transparency ログ

CT ログは、TLS 証明書の発行を記録する、暗号学的に保証された公開監査可能な追記専用の記録です。主要な CA（Let's Encrypt、DigiCert、Sectigo、Google Trust Services）は、発行したすべての証明書を複数の CT ログに送信します。2025 年現在、Chrome と Safari は、すべての公開信頼済み証明書に CT を必須としています。

CT によるフィッシング検出

攻撃者は、フィッシングサイトを HTTPS で正規のものに見せかけるため、類似ドメインを登録し、（多くの場合 Let's Encrypt から）無料の証明書を取得します。証明書はフィッシングキャンペーンの開始前にログに記録されるため、CT 監視によって早期に検出し、先手を打ってブロックできます。

crt.sh データベース

crt.sh は、Sectigo が運営する無料のウェブインターフェース兼 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 インテリジェンスレポートが作成される

参考資料

• crt.sh 証明書検索
• Certstream リアルタイム CT モニター
• River Security：攻撃対象領域の発見における CT ログ
• Let's Encrypt：Certificate Transparency ログ
• SSLMate Cert Spotter
• CyberSierra：早期警告システムとしての CT ログ