分析证书透明度以检测钓鱼
概述
证书透明度（Certificate Transparency，CT）是一项互联网安全标准，它创建了一个公开的、只可追加的 SSL/TLS 证书签发日志。监控 CT 日志能够及早发现：注册模仿合法品牌的钓鱼域名、针对自有域名的未授权证书签发，以及基于证书的攻击基础设施。本技能涵盖通过 crt.sh 查询 CT 日志、使用 Certstream 进行实时监控、构建针对可疑证书的自动告警，以及将发现集成到威胁情报工作流中。
何时使用
- 调查需要分析证书透明度以检测钓鱼的安全事件时
- 为此领域构建检测规则或威胁狩猎查询时
- SOC 分析师需要此类分析的结构化流程时
- 验证相关攻击技术的安全监控覆盖范围时
前提条件
- Python 3.9+，并安装
requests、certstream、tldextract、Levenshtein库 - 能访问 crt.sh (https://crt.sh/) 以查询历史 CT 日志
- 能连接 Certstream (https://certstream.calidog.io/) 以进行实时监控
- 准备好要监控的组织域名和品牌关键词列表
- 理解 SSL/TLS 证书结构和签发流程
关键概念
证书透明度日志
CT 日志是经过密码学保障、公开可审计、只可追加的 TLS 证书签发记录。主流 CA（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):
"""查询 crt.sh 获取匹配某个域名的证书。"""
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: *.{domain} 有 {len(certs)} 个证书")
return certs
return []
def find_suspicious_certs(self, domain):
"""查找可能是钓鱼尝试的证书。"""
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", "")
# 检查是否为目标域名本身（合法）
extracted = tldextract.extract(common_name)
cert_domain = f"{extracted.domain}.{extracted.suffix}"
if cert_domain == domain:
continue # 合法证书
# 标记可疑模式
flags = []
if domain.replace(".", "") in common_name.replace(".", ""):
flags.append("包含目标域名字符串")
if any(kw in common_name for kw in self.brand_keywords):
flags.append("包含品牌关键词")
if "let's encrypt" in issuer.lower():
flags.append("免费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"[+] 发现 {len(suspicious)} 个可疑证书")
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']}] 标记: {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):
"""开始实时 CT 日志监控。"""
print("[*] 启动 Certstream 监控...")
print(f" 监控域名: {self.watched_domains}")
print(f" 品牌关键词: {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" [告警] {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[+] 监控已停止。共收集 {len(self.alerts)} 条告警。")
return self.alerts
def _is_suspicious(self, domain):
"""检查域名是否相对于监控域名是可疑的。"""
for watched in self.watched_domains:
# 精确关键词匹配
watched_base = watched.split(".")[0]
if watched_base in domain and domain != watched:
return True
# Levenshtein 距离（域名抢注检测）
domain_base = tldextract.extract(domain).domain
similarity = Levenshtein.ratio(watched_base, domain_base)
if similarity >= self.threshold and domain_base != watched_base:
return True
# 品牌关键词匹配
for keyword in self.brand_keywords:
if keyword in domain:
return True
return False
def _get_reason(self, domain):
"""确定域名被标记的原因。"""
reasons = []
for watched in self.watched_domains:
watched_base = watched.split(".")[0]
if watched_base in domain:
reasons.append(f"包含 '{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"与 '{watched}' 相似 ({similarity:.0%})")
for kw in self.brand_keywords:
if kw in domain:
reasons.append(f"品牌关键词 '{kw}'")
return "; ".join(reasons) if reasons else "未知"
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):
"""从证书透明度日志中发现所有子域名。"""
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 子域名枚举 {domain}: 共 {len(sorted_subs)} 个子域名")
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"""# 证书透明度情报报告
## 目标域名: {domain}
## 生成时间: {datetime.now().isoformat()}
## 摘要
- 发现的可疑证书数: {len(suspicious_certs)}
- 触发的实时告警数: {len(certstream_alerts)}
## 可疑证书 (crt.sh)
| 通用名称 | 签发者 | 标记 | crt.sh 链接 |
|------------|--------|-------|-------------|
"""
for cert in suspicious_certs[:20]:
flags = "; ".join(cert.get("flags", []))
report += (f"| {cert['common_name']} | {cert['issuer'][:30]} "
f"| {flags} | [查看]({cert['crt_sh_url']}) |\n")
report += f"""
## 实时 Certstream 告警
| 域名 | 签发者 | 原因 | 检测时间 |
|--------|--------|--------|----------|
"""
for alert in certstream_alerts[:20]:
report += (f"| {alert['domain']} | {alert['issuer']} "
f"| {alert['reason']} | {alert['detected_at'][:19]} |\n")
report += """
## 建议
1. 将标记的域名添加到 DNS 黑洞 / Web 代理黑名单
2. 对确认为钓鱼的域名提交下架请求
3. 持续监控 CT 日志以发现新证书注册
4. 实施 CAA DNS 记录以限制对您域名的证书签发
5. 部署 DMARC 以防止来自近似域名的电子邮件欺骗
"""
with open(f"ct_report_{domain.replace('.','_')}.md", "w") as f:
f.write(report)
print(f"[+] CT 报告已保存")
return report
generate_ct_report(suspicious, alerts if 'alerts' in dir() else [], "mycompany.com")
验证标准
- crt.sh 查询返回目标域名的证书数据
- 能基于近似模式识别出可疑证书
- Certstream 实时监控能检测到新的钓鱼证书
- 子域名枚举能从 CT 日志中生成完整的子域名列表
- 生成的告警带有原因分类
- CT 情报报告包含可操作的建议