$npx -y skills add killvxk/cybersecurity-skills-zh --skill analyzing-certificate-transparency-for-phishing使用 crt.sh 和 Certstream 监控证书透明度日志,检测钓鱼域名、仿冒证书以及针对您所在组织的未经授权的证书签发行为。
| 1 | # 通过证书透明度分析钓鱼威胁 |
| 2 | |
| 3 | ## 概述 |
| 4 | |
| 5 | 证书透明度(Certificate Transparency,CT)是一种互联网安全标准,为所有已签发的 SSL/TLS 证书创建公开的、仅可追加的日志记录。监控 CT 日志能够实现对以下威胁的早期检测:模仿合法品牌注册证书的钓鱼(Phishing)域名、针对自有域名的未授权证书签发,以及基于证书的攻击基础设施。本技能涵盖通过 crt.sh 查询 CT 日志、使用 Certstream 进行实时监控、构建可疑证书的自动化告警,以及将发现结果整合到威胁情报(Threat Intelligence)工作流程中。 |
| 6 | |
| 7 | ## 前置条件 |
| 8 | |
| 9 | - Python 3.9+,安装 `requests`、`certstream`、`tldextract`、`Levenshtein` 库 |
| 10 | - 访问 crt.sh (https://crt.sh/) 进行历史 CT 日志查询 |
| 11 | - Certstream (https://certstream.calidog.io/) 用于实时监控 |
| 12 | - 需监控的组织域名和品牌关键词列表 |
| 13 | - 了解 SSL/TLS 证书结构和签发流程 |
| 14 | |
| 15 | ## 核心概念 |
| 16 | |
| 17 | ### 证书透明度日志 |
| 18 | |
| 19 | CT 日志是经过密码学保证的、可公开审计的、仅可追加的 TLS 证书签发记录。主要 CA(Let's Encrypt、DigiCert、Sectigo、Google Trust Services)将所有签发的证书提交到多个 CT 日志。截至 2025 年,Chrome 和 Safari 要求所有公开信任的证书必须支持 CT。 |
| 20 | |
| 21 | ### 通过 CT 检测钓鱼 |
| 22 | |
| 23 | 攻击者注册仿冒域名并获取免费证书(通常来自 Let's Encrypt),使钓鱼网站通过 HTTPS 显得合法。CT 监控能够早期发现这些行为,因为证书在钓鱼活动发起前就已出现在日志中,为主动封锁提供了时间窗口。 |
| 24 | |
| 25 | ### crt.sh 数据库 |
| 26 | |
| 27 | crt.sh 是由 Sectigo 运营的免费 Web 界面和 PostgreSQL 数据库,对 CT 日志进行索引。支持通配符搜索(`%.example.com`)、直接 SQL 查询和 JSON API 响应。跨所有主要 CT 日志追踪证书签发、到期和吊销情况。 |
| 28 | |
| 29 | ## 实践步骤 |
| 30 | |
| 31 | ### 步骤 1:通过 crt.sh 查询证书历史 |
| 32 | |
| 33 | ```python |
| 34 | import requests |
| 35 | import json |
| 36 | from datetime import datetime |
| 37 | import tldextract |
| 38 | |
| 39 | class CTLogMonitor: |
| 40 | CRT_SH_URL = "https://crt.sh" |
| 41 | |
| 42 | def __init__(self, monitored_domains, brand_keywords): |
| 43 | self.monitored_domains = monitored_domains |
| 44 | self.brand_keywords = [k.lower() for k in brand_keywords] |
| 45 | |
| 46 | def query_crt_sh(self, domain, include_expired=False): |
| 47 | """查询 crt.sh 中匹配域名的证书。""" |
| 48 | params = { |
| 49 | "q": f"%.{domain}", |
| 50 | "output": "json", |
| 51 | } |
| 52 | if not include_expired: |
| 53 | params["exclude"] = "expired" |
| 54 | |
| 55 | resp = requests.get(self.CRT_SH_URL, params=params, timeout=30) |
| 56 | if resp.status_code == 200: |
| 57 | certs = resp.json() |
| 58 | print(f"[+] crt.sh: {len(certs)} certificates for *.{domain}") |
| 59 | return certs |
| 60 | return [] |
| 61 | |
| 62 | def find_suspicious_certs(self, domain): |
| 63 | """查找可能是钓鱼尝试的证书。""" |
| 64 | certs = self.query_crt_sh(domain) |
| 65 | suspicious = [] |
| 66 | |
| 67 | for cert in certs: |
| 68 | common_name = cert.get("common_name", "").lower() |
| 69 | name_value = cert.get("name_value", "").lower() |
| 70 | issuer = cert.get("issuer_name", "") |
| 71 | not_before = cert.get("not_before", "") |
| 72 | not_after = cert.get("not_after", "") |
| 73 | |
| 74 | # 检查精确域名匹配(合法证书) |
| 75 | extracted = tldextract.extract(common_name) |
| 76 | cert_domain = f"{extracted.domain}.{extracted.suffix}" |
| 77 | if cert_domain == domain: |
| 78 | continue # 合法证书,跳过 |
| 79 | |
| 80 | # 标记可疑模式 |
| 81 | flags = [] |
| 82 | if domain.replace(".", "") in common_name.replace(".", ""): |
| 83 | flags.append("contains target domain string") |
| 84 | if any(kw in common_name for kw in self.brand_keywords): |
| 85 | flags.append("contains brand keyword") |
| 86 | if "let's encrypt" in issuer.lower(): |
| 87 | flags.append("free CA (Let's Encrypt)") |
| 88 | |
| 89 | if flags: |
| 90 | suspicious.append({ |
| 91 | "common_name": cert.get("common_name", ""), |
| 92 | "name_value": cert.get("name_value", ""), |
| 93 | "issuer": issuer, |
| 94 | "not_before": not_before, |
| 95 | "not_after": not_after, |
| 96 | "serial": cert.get("serial_number", ""), |
| 97 | "flags": flags, |
| 98 | "crt_sh_id": cert.get("id", ""), |
| 99 | "crt_sh_url": f"https://crt.sh/?id={cert.get('id', '')}", |
| 100 | }) |
| 101 | |
| 102 | print(f"[+] Found {len(suspicious)} suspicious certificates") |
| 103 | return suspicious |
| 104 | |
| 105 | monitor = CTLogMonitor( |
| 106 | monitored_domains=["mycompany.com", "mycompany.org"], |
| 107 | brand_keywords=["mycompany", "mybrand", "myproduct"], |
| 108 | ) |
| 109 | suspicious = monitor.find_suspicious_certs("mycompany.com") |
| 110 | for cert in suspicious[:5]: |
| 111 | print(f" [{cert['common_name']}] Flags: {cert['flags']}") |
| 112 | ``` |
| 113 | |
| 114 | ### 步骤 2:使用 Certstream 进行实时监控 |
| 115 | |
| 116 | ```python |
| 117 | import certstream |
| 118 | import Levenshtein |
| 119 | import re |
| 120 | from datetime import datetime |
| 121 | |
| 122 | class CertstreamMonitor: |
| 123 | def __init__(self, watched_domains, brand_keywords, similarity_threshold=0.8): |
| 124 | self.watched_domains = [d.lower() for d in watched_domains] |
| 125 | self.brand_keywords = [k.lower() for k in brand_keywords] |
| 126 | self.threshold = similarity_threshold |
| 127 | self.alerts = [] |
| 128 | |
| 129 | def start_monitoring(self, max_alerts=100): |
| 130 | """启动 CT 日志实时监控。""" |
| 131 | print("[*] 正在启动 Certstream 监控...") |
| 132 | print(f" 监控域名: {self.watched_domains}") |
| 133 | print(f" 关键词: {self.brand_keywords}") |
| 134 | |
| 135 | def callback(message, context): |
| 136 | if message["message_type"] == "certificate_update": |