CT日志、SCT签发时间与证书透明度的时间回拨攻击

#!/usr/bin/env python3
"""
sct_time_validator.py — SCT 时间戳多源交叉验证引擎
从多个 CT 日志拉取同一张证书的 SCT，比较时间戳一致性

检测规则：
  1. 多日志 SCT 时间戳互差 > 10s → 告警（可能某日志 NTP 被污染）
  2. SCT 时间戳 < cert.notBefore → 告警（可能是预签名，需进一步分析）
  3. SCT 时间戳 > OCSP.producedAt + 阈值 → 告警（证书在签发前已被记录？）

依赖: pip install cert-stream pyopenssl
"""

import json
import ssl
import socket
import hashlib
import struct
import time
from datetime import datetime, timedelta
from typing import List, Dict, Optional
from dataclasses import dataclass
import requests

# ===== 数据结构 =====
@dataclass
classSCTRecord:
"""从 CT 日志获取的 SCT 记录"""
    log_id: str              # 日志 ID（SHA-256）
    log_name: str            # 日志名称
    timestamp: int           # SCT 毫秒时间戳
    signature_algorithm: str
    signature: str

@dataclass
classSCTValidationAlert:
"""SCT 验证告警"""
    severity: str           # HIGH / MEDIUM / LOW
    alert_type: str
    description: str
    sct_times: List[int]    # 相关 SCT 时间戳
    time_diff_seconds: float  # 时间差（秒）

# ===== CT 日志列表 =====
# 活跃的 CT 日志（各日志的 base URL）
CT_LOGS = {
"Google Argon 2026": "https://ct.googleapis.com/logs/us1/argon2026/",
"Google Argon 2027": "https://ct.googleapis.com/logs/us1/argon2027/",
"Cloudflare Nimbus 2026": "https://ct.cloudflare.com/logs/nimbus2026/",
"DigiCert Yeti 2026": "https://yeti2026.ct.digicert.com/",
"Sectigo Mammoth 2026": "https://mammoth2026.ct.sectigo.com/",
}

# ===== 证书 SCT 提取 =====
defget_scts_from_domain(domain: str, port: int = 443) -> List[SCTRecord]:
"""通过 TLS 握手获取证书中的 SCT"""
    ctx = ssl.create_default_context()
    ctx.check_hostname = True
    ctx.verify_mode = ssl.CERT_REQUIRED

    sock = socket.create_connection((domain, port), timeout=10)
with ctx.wrap_socket(sock, server_hostname=domain) as ssock:
        cert_bin = ssock.getpeercert(binary_form=True)
# SCT 嵌入在证书的 CT Precertificate 扩展中
# 此处使用 pyopenssl 解析扩展
from OpenSSL import crypto
        cert = crypto.load_certificate(crypto.FILETYPE_ASN1, cert_bin)

        scts = []
for i in range(cert.get_extension_count()):
            ext = cert.get_extension(i)
if ext.get_short_name() == b'CT Precertificate SCTs':
# 解析 SCT 列表
                sct_data = ext.get_data()
                scts = parse_sct_list(sct_data)
return scts

defparse_sct_list(data: bytes) -> List[SCTRecord]:
"""解析 SCT 列表（RFC 6962 第 3.3 节）"""
    scts = []
    offset = 2# 跳过 2 字节的列表长度
while offset < len(data):
# 读取单个 SCT 长度
        sct_len = struct.unpack('!H', data[offset:offset+2])[0]
        offset += 2
        sct_bytes = data[offset:offset+sct_len]
        offset += sct_len

# 解析 SCT 字段
        sct_version = sct_bytes[0]
        log_id = sct_bytes[1:33].hex()
        timestamp = struct.unpack('!Q', sct_bytes[33:41])[0]
# ... 后续为 extensions 和 signature
        scts.append(SCTRecord(
            log_id=log_id,
            log_name=f"Log_{log_id[:8]}",
            timestamp=timestamp,
            signature_algorithm="unknown",
            signature=""
        ))
return scts


# ===== 查询 CT 日志获取 SCT =====
defquery_ct_log(log_url: str, domain: str) -> Optional[List[SCTRecord]]:
"""从 CT 日志 API 查询域名的证书"""
try:
# 使用 certstream 或 crt.sh API
        crtsh_url = f"https://crt.sh/?q=%.{domain}&output=json"
        resp = requests.get(crtsh_url, timeout=30)
if resp.status_code == 200:
            certs = resp.json()
            scts = []
for cert in certs[:5]:  # 取最近 5 张证书
if'entry_timestamp'in cert:
                    scts.append(SCTRecord(
                        log_id=cert.get('issuer_ca_id', 'unknown'),
                        log_name=cert.get('issuer_name', 'unknown')[:32],
                        timestamp=int(datetime.fromisoformat(
                            cert['entry_timestamp'].replace('Z', '+00:00')
                        ).timestamp() * 1000),
                        signature_algorithm="",
                        signature=""
                    ))
return scts
except Exception as e:
        print(f"[-] 查询 {log_url} 失败: {e}")
returnNone


# ===== 时间交叉验证 =====
defvalidate_sct_timestamps(scts: List[SCTRecord]) -> List[SCTValidationAlert]:
"""执行 SCT 时间戳交叉验证"""
    alerts = []

if len(scts) < 2:
        alerts.append(SCTValidationAlert(
            severity="MEDIUM",
            alert_type="INSUFFICIENT_SCTS",
            description=f"仅发现 {len(scts)} 个 SCT，无法进行多日志交叉验证",
            sct_times=[],
            time_diff_seconds=0
        ))
return alerts

# 规则 1：多日志 SCT 时间戳互差检测
    timestamps = sorted([sct.timestamp for sct in scts])
    max_diff = (timestamps[-1] - timestamps[0]) / 1000.0# 转换为秒

if max_diff > 10:
        alerts.append(SCTValidationAlert(
            severity="HIGH",
            alert_type="SCT_TIME_DIVERGENCE",
            description=(
f"不同 CT 日志的 SCT 时间戳差异过大："
f"最早 {datetime.fromtimestamp(timestamps[0]/1000)}，"
f"最晚 {datetime.fromtimestamp(timestamps[-1]/1000)}，"
f"相差 {max_diff:.0f}s"
            ),
            sct_times=timestamps,
            time_diff_seconds=max_diff
        ))

# 规则 2：SCT 时间与当前真实时间比较
    now_ms = int(time.time() * 1000)
for t in timestamps:
        future_diff = (t - now_ms) / 1000.0
if future_diff > 86400:  # 超过 1 天
            alerts.append(SCTValidationAlert(
                severity="HIGH",
                alert_type="SCT_FUTURE_TIMESTAMP",
                description=(
f"SCT 时间戳在未来："
f"SCT 时间 {datetime.fromtimestamp(t/1000)}，"
f"当前时间 {datetime.fromtimestamp(now_ms/1000)}，"
f"偏差 {future_diff:.0f}s"
                ),
                sct_times=[t],
                time_diff_seconds=future_diff
            ))

return alerts


# ===== 主程序 =====
if __name__ == "__main__":
import sys

if len(sys.argv) < 2:
        print(f"用法: python {sys.argv[0]} <域名>")
        print(f"示例: python {sys.argv[0]} example.com")
        sys.exit(1)

    domain = sys.argv[1]
    print(f"[*] SCT 时间戳交叉验证 — {domain}")
    print("=" * 60)

# 从 crt.sh 获取证书记录
    all_scts = []
for log_name, log_url in CT_LOGS.items():
        print(f"[*] 查询 {log_name}...")
        scts = query_ct_log(log_url, domain)
if scts:
            print(f"    [+]: 找到 {len(scts)} 条记录")
            all_scts.extend(scts)

ifnot all_scts:
        print("[-] 未找到任何 SCT 记录")
        sys.exit(1)

    print(f"\n[+] 共发现 {len(all_scts)} 条 SCT 记录")

# 显示时间戳详情
for sct in all_scts[:10]:  # 仅显示前 10 条
        dt = datetime.fromtimestamp(sct.timestamp / 1000)
        print(f"    [{dt.isoformat()}] {sct.log_name}")

# 执行验证
    alerts = validate_sct_timestamps(all_scts)

    print(f"\n[验证结果]")
if alerts:
for alert in alerts:
            print(f"\n[{alert.severity}] {alert.alert_type}")
            print(f"    {alert.description}")
else:
        print("[+] 所有 SCT 时间戳通过交叉验证")