跳到主内容
快讯直播
AI智模界
教程

自建服务证书透明度监控:第一时间发现伪造证书

这篇能做出什么

假设你在家里或一台 VPS 上跑着 GitLab、Nextcloud、群晖 NAS、或者干脆就是一个博客,域名是 example.com。某天有人注册了 examp1e.com 之类形似域名做钓鱼,这不稀奇;真正麻烦的是有人直接拿到一张签给 example.com 的合法证书——浏览器地址栏显示的锁头是真的,证书链也能验证通过,肉眼几乎分辨不出来。

证书透明度(Certificate Transparency,CT)机制要求:公开信任的 CA 在签发证书后,必须把这张证书提交到若干公开的、只能追加的日志服务器里。也就是说,只要有人为你的域名签了公开信任的证书,这件事迟早会出现在公开日志里,包括你并不知情的那一次。

这套监控做完之后,你会得到:

  • 一个常驻运行的服务,盯着 CT 日志里所有包含你域名的证书;
  • 一旦出现新证书(尤其是你没签过的),在几分钟到几小时内收到 Telegram / 邮件 / 企业微信机器人通知;
  • 通知里带上域名、签发机构、有效期、证书指纹,方便你立刻判断是"自己续签了"还是"有人在搞事";
  • 一份本地留存的证书记录库,以后可以拿来排查"这张证书什么时候冒出来的"。

需要提前说清楚它的边界:CT 监控只能发现公开 CA 签发的证书。你自己搭的内部 CA 给内网签的证书不会进 CT 日志;私钥泄露本身也不靠这个发现——但如果攻击者用泄露的私钥去公开 CA 重新签一张,那就一定会被记下来。

另外,CT 日志不是实时的。日志服务器允许有一个"最大合并延迟"(MMD),通常是几小时,上限一般不超过 24 小时。所以这套系统的定位是"第一时间知道",而不是"秒级拦截"。

---

前置条件清单

开始之前,确认手上有这些:

1. 一台常开的机器。VPS、树莓派、群晖 Docker、旧笔记本都行,只要能长跑。别放在你自己会随手关机的电脑上。

2. 要监控的域名列表。建议至少包含根域和一个通配,例如 example.com、*.example.com。

3. Python 3 运行环境,或者 Docker。本文示例用 Python,逻辑简单,好改。

4. 一个通知通道。Telegram Bot、SMTP 邮箱、Slack Incoming Webhook、企业微信机器人、Bark 都可以。下面示例用 Telegram,换成别的只是改一个 notify() 函数。

5. 出网权限。机器要能访问 CT 日志服务和通知服务的接口。

6. 基础命令行能力:会 curl、会看 docker logs 或 journalctl、会用 jq 就够。

涉及具体工具和依赖的版本,以官方文档当前版本为准,本文不锁定版本号。

---

第一步:先手工看一眼 CT 日志长什么样

别急着写代码,先确认"这件事真的能查到"。crt.sh 是一个常用的 CT 日志查询前端,提供 JSON 接口。

```bash

curl -sS --max-time 60 \

'https://crt.sh/?q=%25.example.com&output=json' \

| jq -r '.[] | [.id, .entry_timestamp, .issuer_name, .name_value] | @tsv' \

| head -n 20

```

把 example.com 换成你自己的域名。%25 是 URL 转义后的 %,代表 SQL 里的通配,能匹配子域。

你会看到类似这样的输出:

```text

123456789 2024-05-11T03:12:44.123 Let's Encrypt example.com

123456790 2024-05-11T03:12:44.123 Let's Encrypt *.example.com

123456812 2024-05-12T09:01:10.456 Some CA mail.example.com

```

两个关键点:entry_timestamp 是这张证书进入日志的时间,用它做增量比较最可靠;name_value 里可能是一串换行分隔的域名,解析时要处理。

手工查一遍,记下当前最新的 id 和 entry_timestamp,这就是你后面的起点。

---

第二步:选一条采集路线

有两条常见路线,按需选一条或都上:

  • 路线 A:轮询 crt.sh。实现简单,不需要长连接,缺点是受对方限速影响,延迟偏大。适合新手起步、域名不多的情况。
  • 路线 B:订阅实时 CT 流。以 WebSocket 方式订阅全量证书事件,秒级到分钟级感知,缺点是数据量大(每秒几百上千条),需要本地过滤。适合想做得"实时"一点的人。

一个务实的做法是:先用路线 A 跑起来,稳定之后再加路线 B 做补充。下面的代码两条都给。

---

第三步:路线 A——crt.sh 轮询脚本

先建目录和状态文件:

```bash

sudo mkdir -p /var/lib/ct-watch

sudo touch /var/lib/ct-watch/crtsh.last

sudo chown -R "$USER" /var/lib/ct-watch

```

脚本 poll-crtsh.sh:

```bash

#!/usr/bin/env bash

set -euo pipefail

DOMAIN="example.com"

STATE="/var/lib/ct-watch/crtsh.last"

RAW="/var/lib/ct-watch/crtsh.json"

TSV="/var/lib/ct-watch/crtsh.tsv"

NEW="/var/lib/ct-watch/crtsh.new"

BOT_TOKEN="你的 Telegram Bot AI 词典:Token">Token"

CHAT_ID="你的会话 ID"

1) 拉取,带重试和超时,避免网络抖动把脚本搞崩

curl -sS --retry 5 --retry-delay 15 --retry-all-errors --max-time 90 \

"https://crt.sh/?q=%25.${DOMAIN}&output=json" -o "${RAW}.tmp"

2) 只有拿到合法 JSON 才替换旧文件,防止把好数据覆盖成错误页

jq -e 'type == "array"' "${RAW}.tmp" >/dev/null

mv "${RAW}.tmp" "$RAW"

3) 拍平成 TSV,按 id 排序

jq -r '.[] | [.id, .entry_timestamp, .not_before, .issuer_name,

(.name_value | gsub("\n"; " "))] | @tsv' "$RAW" \

sort -t$'\t' -k1,1n > "$TSV"

4) 找出比上次记录更新的条目

LAST=$(cat "$STATE" 2>/dev/null || echo 0)

awk -F'\t' -v last="$LAST" '$1 > last' "$TSV" > "$NEW"

COUNT=$(wc -l < "$NEW")

if [ "$COUNT" -gt 0 ]; then

MSG=$(printf 'CT 发现新证书(%s 条)\n%s' "$COUNT" \

"$(awk -F'\t' '{print $1" "$2" "$4" "$5}' "$NEW" | head -n 20)")

curl -sS --max-time 20 \

"https://api.telegram.org/bot${BOT_TOKEN}/sendMessage" \

-d "chat_id=${CHAT_ID}" \

--data-urlencode "text=${MSG}" > /dev/null

5) 更新水位线

tail -n 1 "$TSV" | cut -f1 > "$STATE"

fi

```

几个值得留意的设计:

  • --retry-all-errors 配上 --retry,对付 crt.sh 偶发的 5xx 和超时;
  • 校验返回确实是数组再覆盖文件,避免一次失败把状态清空;
  • 水位线用 id 而不是时间戳,避免时区解析的坑;
  • 只推送最近 20 条,防止首次运行刷屏。

跑起来:

```bash

chmod +x poll-crtsh.sh

./poll-crtsh.sh

```

第一次运行会因为水位线为 0 而把全部历史记录当成"新证书",所以先把 crtsh.last 手动写成当前最大 id,再开始正式监控:

```bash

jq -r '.[].id' /var/lib/ct-watch/crtsh.json | sort -n | tail -n 1 \

| sudo tee /var/lib/ct-watch/crtsh.last

```

---

第四步:路线 B——订阅证书流

实时流适合"域名多、想更快知道"的场景。装依赖:

```bash

python3 -m venv /opt/ct-watch/venv

/opt/ct-watch/venv/bin/pip install certstream requests

```

脚本 ct_watch.py:

```python

#!/usr/bin/env python3

"""订阅证书透明度实时流,命中监控域名就推送通知。"""

import sqlite3

import time

import certstream

import requests

WATCH = {"example.com", "example.net"} # 只盯这些根域,避免全量刷屏

DB_PATH = "/var/lib/ct-watch/seen.db"

BOT_TOKEN = "你的 Telegram Bot Token"

CHAT_ID = "你的会话 ID"

STREAM_URL = "wss://certstream.calidog.io/"

_CONN = None

def get_conn():

"""惰性建立 SQLite 连接,用来对证书指纹去重。"""

global _CONN

if _CONN is None:

_CONN = sqlite3.connect(DB_PATH, check_same_thread=False)

_CONN.execute(

"""CREATE TABLE IF NOT EXISTS seen (

fingerprint TEXT PRIMARY KEY,

domains TEXT,

issuer TEXT,

not_before TEXT,

first_seen INTEGER

)"""

)

_CONN.commit()

return _CONN

def is_watched(domain: str) -> bool:

"""example.com 与 a.b.example.com 都算命中;泛域名去掉开头的 *."""

d = domain.lower().lstrip("*.").rstrip(".")

return any(d == w or d.endswith("." + w) for w in WATCH)

def notify(text: str) -> None:

url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"

try:

requests.post(

url,

json={"chat_id": CHAT_ID, "text": text,

"disable_web_page_preview": True},

timeout=10,

)

except requests.RequestException as exc:

print("notify failed:", exc, flush=True)

def handle(message, context):

if message.get("message_type") != "certificate_update":

return

leaf = message["data"]["leaf_cert"]

fingerprint = leaf["fingerprint"]

domains = leaf.get("all_domains") or []

hits = sorted({d for d in domains if is_watched(d)})

if not hits:

return # 绝大多数事件在这里被过滤掉

conn = get_conn()

if conn.execute("SELECT 1 FROM seen WHERE fingerprint = ?",

(fingerprint,)).fetchone():

return # 同一张证书只看一次

issuer = (leaf.get("issuer") or {}).get("O", "未知")

not_before = leaf.get("not_before", "未知")

conn.execute(

"INSERT INTO seen VALUES (?,?,?,?,?)",

(fingerprint, ",".join(hits), issuer, not_before, int(time.time())),

)

conn.commit()

notify(

"发现新证书\n"

f"域名:{', '.join(hits)}\n"

f"签发者:{issuer}\n"

f"有效期起:{not_before}\n"

f"指纹:{fingerprint}\n"

"如果不是自己签的,立即检查 DNS 与私钥。"

)

def main():

while True:

try:

certstream.listen_for_events(

handle, url=STREAM_URL, skip_heartbeats=False

)

except Exception as exc: # 长连接断开是常态,必须自动重连

print("stream error, retry in 5s:", exc, flush=True)

time.sleep(5)

if __name__ == "__main__":

main()

```

这里最关键的是过滤顺序:先判断域名是否命中,再做数据库去重。全量流每秒上千条事件,如果反过来先查库,磁盘和 CPU 都会吃不消。

公共流可能限速或不稳定,长期运行可考虑自建日志客户端直连 CT 日志的 get-entries 接口,具体协议与端点以官方文档为准。

---

第五步:交给 systemd 常驻

```ini

[Unit]

Description=Certificate Transparency watcher

After=network-online.target

Wants=network-online.target

[Service]

Type=simple

User=ctwatch

WorkingDirectory=/opt/ct-watch

ExecStart=/opt/ct-watch/venv/bin/python /opt/ct-watch/ct_watch.py

Restart=always

RestartSec=10

StandardOutput=append:/var/log/ct-watch.log

StandardError=append:/var/log/ct-watch.log

[Install]

WantedBy=multi-user.target

```

存为 /etc/systemd/system/ct-watch.service,然后:

```bash

sudo systemctl daemon-reload

sudo systemctl enable --now ct-watch

sudo journalctl -u ct-watch -f

```

路线 A 的轮询脚本可以用 systemd timer 每 30 分钟跑一次,比在脚本里写 sleep 循环更便于观察和排错:

```ini

/etc/systemd/system/ct-watch-poll.timer

[Unit]

Description=Run crt.sh poll every 30 minutes

[Timer]

OnBootSec=2min

OnUnitActiveSec=30min

Persistent=true

[Install]

WantedBy=timers.target

```

---

第六步:验证告警真的能响

监控搭好却不知道会不会响,等于没搭。验证方法有几种,从安全到彻底:

1. 删水位线法(最快)。把 crtsh.last 改小一点,等下一轮轮询,看有没有收到通知。这会一次性推送一批历史证书,属于噪声,验证完记得把水位线改回去。

2. 临时子域签发法(最真实)。用你控制的域名,给一个平时不用的子域签一张短期证书,比如 ct-test.example.com,等日志合并后看是否触发。这能同时验证采集、过滤、推送三段的连通性。

3. 本地桩测试。直接调 notify() 函数发一条,确认通知通道本身没问题。这一步建议先做,省得后面排查时分不清是采集问题还是推送问题。

顺手加一条基线防护:

```bash

dig +short CAA example.com

```

如果你只从某家 CA 签发证书,用 CAA 记录把签发权限限制在指定的 CA 上,可以从源头减少误签的概率。CAA 不是万能的,但成本很低。

---

常见坑与排错

收不到任何通知。 先看日志有没有输出。路线 A 常见原因是 jq 解析失败或水位线已经被推到最大值;路线 B 常见原因是长连接断了但异常被吞掉。给脚本加 print(..., flush=True),用 journalctl -u ct-watch -f 实时看。

crt.sh 返回 502 或超时。 这是常见现象。加重试与退避,别用固定高频轮询——每 30 分钟一次通常够用,太频繁只会更快被限速。

告警里全是自己续签的证书。 这是预期内的噪声。两个处理方向:一是把自动续签产生的证书指纹预先写入白名单;二是把续签流程改成"续签后主动写入 seen 表"。前者适合手工管理,后者适合自动化程度高的环境。

漏掉了子域。 检查过滤逻辑。example.com 不等于 a.example.com,用后缀匹配时别写成简单的字符串包含,notexample.com 会误命中——后缀判断要带上点号,即 d.endswith("." + w)。

name_value 里有换行导致解析错位。 crt.sh 的一个字段里可能塞了多行域名。写 TSV 时用 gsub("\n"; " ") 压平,写 Python 时用 splitlines() 拆开再逐个判断。

以为 CT 是实时的。 日志合并有延迟,通常是几小时量级,上限一般不超过 24 小时。监控频率高于这个量级没有额外收益,重点应放在"通知能否可靠送达"。

只监控了根域,忘了通配。 如果攻击者签的是 *.example.com,只匹配根域字符串就会漏。反过来,如果你的服务确实用了泛域名,也要接受泛域名证书带来的告警量。

把 CT 当成密钥泄露检测。 它检测不了。CT 只能回答"有没有人为这个域名签过证书"。私钥是否泄露,得靠密钥轮换、访问审计这类手段。

证书指纹去重没有生效。 不同日志、不同前端返回的指纹格式可能不完全一致,建议统一转成小写十六进制再比较。

---

下一步建议

跑通之后,可以按这个顺序继续加固:

1. 把域名清单化。用 YAML 或环境变量维护要监控的域名,脚本从配置读,避免每次改代码。日后加了新服务,只改配置。

2. 区分告警等级。命中根域、签发者不在白名单、有效期异常长——这些走"立即通知";自己续签走"每日汇总"。降噪之后才不会对告警脱敏。

3. 做每日汇总。哪怕没有异常,也发一条"昨日新增 N 张证书"的日报。连续几天正常之后,某天突然不发了,本身就是信号。

4. 结合资产测绘。CT 日志是发现影子子域的好地方。定期把所有出现过的域名导出来,跟自己 DNS 里的记录比对,能翻出早已遗忘的旧服务。

5. 把 CAA、DNSSEC、邮件相关记录一起纳入巡检。域名安全是一个整体,证书只是其中一环。

6. 给监控本身加健康检查。用一个独立的定时任务,检查"最近一次成功采集时间",超过阈值就告警。监控系统自己挂掉却不自知,是最容易踩的坑。

整套流程的核心思路其实很简单:公开日志是别人替你写的账本,你要做的只是定期对账。 脚本不超过两百行,机器一台旧设备就够,但它能让你在伪造证书被使用之前,先一步知道它的存在。

AI 生成本文由 AI 基于公开信息自动生成,仅供参考。