036_密码学实战：证书透明度CT技术深度解析——从SSL证书验证到防钓鱼的完整指南

根证书 → 中间证书 → 终端实体证书（网站证书）

            根哈希
           /      \
      节点1      节点2
      /  \       /  \
  哈希A 哈希B 哈希C 哈希D
    |     |     |     |
  数据A 数据B 数据C 数据D

struct {  
    uint8 version;               // SCT版本号
    LogID id;                    // 日志标识符
    uint64 timestamp;            // 时间戳（Unix时间，毫秒）
    CtExtensions extensions;     // 扩展数据
    digitally-signed signature;  // 日志服务器的数字签名
} SignedCertificateTimestamp;

leaf_hash = SHA256(0x00 || serialized_entry)

internal_hash = SHA256(0x01 || left_child_hash || right_child_hash)

def verify_sct(sct_data, log_public_key, certificate_data):
    """
    验证签名证书时间戳(SCT)
    
    参数:
    sct_data (bytes): SCT的原始数据
    log_public_key (bytes): CT日志的公钥
    certificate_data (bytes): 证书的原始数据
    
    返回:
    bool: 验证是否成功
    """
    from cryptography.hazmat.primitives import hashes
    from cryptography.hazmat.primitives.asymmetric import padding
    from cryptography.hazmat.primitives.serialization import load_der_public_key
    import struct
    
    try:
        # 解析SCT数据
        version = sct_data[0]
        if version != 0:
            print(f"不支持的SCT版本: {version}")
            return False
        
        log_id = sct_data[1:33]  # 日志ID是SHA-256哈希值
        timestamp = struct.unpack(">Q", sct_data[33:41])[0]  # 大端序64位整数
        extensions_length = struct.unpack(">H", sct_data[41:43])[0]  # 大端序16位整数
        extensions = sct_data[43:43+extensions_length]
        signature_length = struct.unpack(">H", sct_data[43+extensions_length:45+extensions_length])[0]
        signature = sct_data[45+extensions_length:45+extensions_length+signature_length]
        
        # 解析签名
        signature_algorithm = signature[0]
        hash_algorithm = signature[1]
        signature_data = signature[2:]
        
        # 验证算法类型
        if signature_algorithm != 1 or hash_algorithm != 4:  # 1=ECDSA, 4=SHA-256
            print("不支持的签名算法或哈希算法")
            return False
        
        # 构建签名数据
        # 对于v1 SCT，签名数据包括：
        # 1. 0x00前缀
        # 2. SCT版本
        # 3. 时间戳
        # 4. 扩展
        # 5. 证书数据
        signing_input = b"\x00" + sct_data[0:1] + sct_data[33:43+extensions_length] + certificate_data
        
        # 加载日志公钥
        public_key = load_der_public_key(log_public_key)
        
        # 验证签名
        public_key.verify(
            signature_data,
            signing_input,
            padding.PKCS7(hashes.SHA256()),
            hashes.SHA256()
        )
        
        print(f"SCT验证成功，时间戳: {timestamp}")
        return True
        
    except Exception as e:
        print(f"SCT验证失败: {e}")
        return False

def verify_inclusion_proof(leaf_hash, proof_nodes, leaf_index, tree_size, root_hash):
    """
    验证证书是否包含在CT日志中
    
    参数:
    leaf_hash (bytes): 叶子节点哈希值
    proof_nodes (list): 证明路径节点列表
    leaf_index (int): 叶子节点在树中的索引
    tree_size (int): 树的大小
    root_hash (bytes): 期望的根哈希值
    
    返回:
    bool: 验证是否成功
    """
    from cryptography.hazmat.primitives import hashes
    import struct
    
    try:
        current_hash = leaf_hash
        current_index = leaf_index
        
        # 处理所有证明节点
        for node in proof_nodes:
            if current_index % 2 == 0:
                # 当前节点是左子节点
                combined = b"\x01" + current_hash + node
            else:
                # 当前节点是右子节点
                combined = b"\x01" + node + current_hash
            
            # 计算新的哈希
            digest = hashes.Hash(hashes.SHA256())
            digest.update(combined)
            current_hash = digest.finalize()
            
            # 更新索引和树大小
            current_index = current_index // 2
            tree_size = (tree_size + 1) // 2
        
        # 验证是否到达根节点
        if tree_size != 1:
            print(f"树大小不正确，期望1，得到{tree_size}")
            return False
        
        # 比较计算得到的根哈希和期望的根哈希
        if current_hash != root_hash:
            print("根哈希不匹配")
            return False
        
        print("包含证明验证成功")
        return True
        
    except Exception as e:
        print(f"包含证明验证失败: {e}")
        return False

def query_ct_log(cert_hash, log_url):
    """
    查询CT日志，获取证书的包含证明
    
    参数:
    cert_hash (bytes): 证书哈希
    log_url (str): CT日志服务器URL
    
    返回:
    dict: 包含证明信息，或None表示失败
    """
    import requests
    import json
    import base64
    
    try:
        # 获取当前日志状态
        sth_url = f"{log_url}/ct/v1/get-sth"
        sth_response = requests.get(sth_url)
        if sth_response.status_code != 200:
            print(f"获取STH失败: {sth_response.status_code}")
            return None
        
        sth_data = sth_response.json()
        tree_size = sth_data["tree_size"]
        root_hash = base64.urlsafe_b64decode(sth_data["sha256_root_hash"] + '==')
        
        # 获取包含证明
        proof_url = f"{log_url}/ct/v1/get-proof-by-hash"
        params = {
            "hash": base64.urlsafe_b64encode(cert_hash).decode('utf-8').rstrip('='),
            "tree_size": tree_size
        }
        proof_response = requests.get(proof_url, params=params)
        
        if proof_response.status_code != 200:
            print(f"获取包含证明失败: {proof_response.status_code}")
            return None
        
        proof_data = proof_response.json()
        leaf_index = proof_data["leaf_index"]
        proof_nodes = [
            base64.urlsafe_b64decode(node + '==') 
            for node in proof_data["audit_path"]
        ]
        
        return {
            "leaf_index": leaf_index,
            "proof_nodes": proof_nodes,
            "tree_size": tree_size,
            "root_hash": root_hash
        }
        
    except Exception as e:
        print(f"查询CT日志失败: {e}")
        return None

def extract_and_verify_sct(cert_file, log_key_file):
    """提取并验证证书中的SCT"""
    import subprocess
    import base64
    
    # 使用OpenSSL提取证书中的SCT
    cmd = [
        'openssl', 'x509', '-in', cert_file,
        '-text', '-noout', '-certopt', 'ext_parse,ctlsinfo'
    ]
    result = subprocess.run(cmd, capture_output=True, text=True)
    
    if result.returncode != 0:
        print(f"提取SCT失败: {result.stderr}")
        return
    
    # 解析输出获取SCT数据（简化示例）
    print("SCT信息:")
    print(result.stdout)
    
    # 加载日志公钥并验证SCT
    # 这里需要根据实际输出解析SCT二进制数据
    # 然后使用之前的verify_sct函数进行验证

def check_certificate_inclusion(cert_hash_hex, log_url):
    """检查证书是否包含在CT日志中"""
    import binascii
    
    # 将十六进制哈希转换为字节
    cert_hash = binascii.unhexlify(cert_hash_hex)
    
    # 查询CT日志
    proof_info = query_ct_log(cert_hash, log_url)
    if not proof_info:
        return False
    
    # 验证包含证明
    return verify_inclusion_proof(
        cert_hash,
        proof_info["proof_nodes"],
        proof_info["leaf_index"],
        proof_info["tree_size"],
        proof_info["root_hash"]
    )

def analyze_ct_log_entries(log_entries):
    """分析CT日志条目，寻找隐藏信息"""
    import re
    from cryptography import x509
    from cryptography.hazmat.backends import default_backend
    
    for i, entry in enumerate(log_entries):
        try:
            # 解析证书
            cert_data = base64.b64decode(entry["leaf_input"][1:])  # 跳过0x00前缀
            certificate = x509.load_der_x509_certificate(cert_data, default_backend())
            
            # 检查证书扩展
            for ext in certificate.extensions:
                ext_data = ext.value
                # 寻找可能的隐藏数据
                if hasattr(ext_data, 'value'):
                    # 检查扩展值中的异常模式
                    value_str = str(ext_data.value)
                    if re.search(r'flag|key|secret|ctf', value_str, re.IGNORECASE):
                        print(f"在条目{i}中发现可疑内容: {value_str}")
            
            # 检查证书主体和颁发者字段
            subject = certificate.subject.rfc4514_string()
            if re.search(r'flag|key|secret|ctf', subject, re.IGNORECASE):
                print(f"在条目{i}的主体中发现可疑内容: {subject}")
                
        except Exception as e:
            print(f"解析条目{i}失败: {e}")

def fix_fake_sct(cert_file, log_key_file):
    """修复伪造的SCT"""
    import OpenSSL
    from cryptography.hazmat.primitives import hashes
    from cryptography.hazmat.primitives.asymmetric import ec
    from cryptography.hazmat.primitives.serialization import load_pem_public_key
    import time
    import struct
    
    # 读取证书
    with open(cert_file, 'rb') as f:
        cert_data = f.read()
    
    # 读取日志公钥
    with open(log_key_file, 'rb') as f:
        log_public_key = load_pem_public_key(f.read())
    
    # 生成正确的SCT
    version = 0
    # 计算日志ID（公钥的SHA-256）
    digest = hashes.Hash(hashes.SHA256())
    digest.update(log_public_key.public_bytes(
        encoding=serialization.Encoding.DER,
        format=serialization.PublicFormat.SubjectPublicKeyInfo
    ))
    log_id = digest.finalize()
    
    # 使用当前时间作为时间戳
    timestamp = int(time.time() * 1000)  # 毫秒
    extensions_length = 0
    extensions = b''
    
    # 构建签名输入
    signing_input = (
        b"\x00" +  # 签名类型
        struct.pack(">B", version) +
        struct.pack(">Q", timestamp) +
        struct.pack(">H", extensions_length) +
        extensions +
        cert_data
    )
    
    # 这里需要私钥来签名，但在挑战中可能需要通过其他方式获取有效签名
    # 例如，寻找私钥泄露或利用其他漏洞
    ```

#### 5.3.2 案例二：CT日志中的隐藏信息

**挑战描述**：
CT日志中隐藏了一个flag，参赛者需要分析日志数据并找出这个flag。

**解决过程**：

1. 获取CT日志条目数据
2. 解析每个日志条目对应的证书
3. 检查证书的各个字段，特别是扩展字段
4. 在证书的自定义扩展中发现了编码的flag数据
5. 解码数据得到flag

**关键代码**：

```python
def find_hidden_flag(log_entries_file):
    """从CT日志条目中找出隐藏的flag"""
    import json
    import base64
    from cryptography import x509
    from cryptography.hazmat.backends import default_backend
    import re
    
    # 读取日志条目数据
    with open(log_entries_file, 'r') as f:
        log_entries = json.load(f)
    
    for entry in log_entries:
        try:
            # 解析叶子节点数据
            leaf_input = base64.b64decode(entry['leaf_input'])
            
            # 跳过0x00前缀，获取证书数据
            if len(leaf_input) > 1 and leaf_input[0] == 0x00:
                cert_data = leaf_input[1:]
                
                # 解析证书
                cert = x509.load_der_x509_certificate(cert_data, default_backend())
                
                # 检查证书的所有扩展
                for ext in cert.extensions:
                    # 转换扩展值为字符串
                    ext_data = str(ext.value)
                    
                    # 搜索可能的flag格式
                    if re.search(r'flag\{[^}]*\}', ext_data, re.IGNORECASE):
                        print(f"找到flag: {ext_data}")
                        return
                    
                    # 检查是否有base64或hex编码的数据
                    if len(ext_data) > 20:
                        # 尝试解码可能的base64数据
                        try:
                            # 清理字符串并尝试解码
                            clean_data = re.sub(r'[^a-za-z0-9+/=]', '', ext_data)
                            decoded = base64.b64decode(clean_data)
                            decoded_str = decoded.decode('utf-8', errors='ignore')
                            
                            # 搜索解码后的数据中的flag
                            if re.search(r'flag|ctf', decoded_str, re.IGNORECASE):
                                print(f"可能的flag: {decoded_str}")
                                return
                                
                        except Exception:
                            pass
                            
        except Exception as e:
            print(f"解析条目时出错: {e}")
            
    print("未找到flag")

def detect_fake_sct(cert_file, known_logs_file):
    """
    检测证书中可能的伪造SCT
    
    参数:
    cert_file (str): 证书文件路径
    known_logs_file (str): 已知CT日志信息文件
    
    返回:
    list: 可疑SCT列表
    """
    import subprocess
    import json
    import re
    import time
    import base64
    
    # 读取已知CT日志信息
    with open(known_logs_file, 'r') as f:
        known_logs = json.load(f)
    
    # 使用OpenSSL提取SCT信息
    cmd = [
        'openssl', 'x509', '-in', cert_file,
        '-text', '-noout', '-certopt', 'ext_parse,ctlsinfo'
    ]
    result = subprocess.run(cmd, capture_output=True, text=True)
    
    if result.returncode != 0:
        print(f"提取SCT失败: {result.stderr}")
        return []
    
    output = result.stdout
    suspicious_scts = []
    
    # 解析SCT信息（简化版）
    sct_blocks = re.split(r'Signed Certificate Timestamp \(SCT\):', output)
    
    for i, block in enumerate(sct_blocks[1:], 1):
        # 提取日志ID
        log_id_match = re.search(r'Log ID: ([0-9a-f:]+)', block)
        if log_id_match:
            log_id = log_id_match.group(1).replace(':', '').lower()
            
            # 检查是否为已知日志
            is_known_log = False
            for log in known_logs:
                if log_id == log['log_id'].lower():
                    is_known_log = True
                    break
            
            if not is_known_log:
                suspicious_scts.append(f"SCT {i}: 未知日志ID {log_id}")
        
        # 提取时间戳
        timestamp_match = re.search(r'Timestamp: ([0-9]+)', block)
        if timestamp_match:
            timestamp = int(timestamp_match.group(1)) / 1000  # 转换为秒
            current_time = time.time()
            
            # 检查时间戳是否异常（太旧或未来时间）
            if timestamp < current_time - 31536000 or timestamp > current_time + 86400:
                suspicious_scts.append(
                    f"SCT {i}: 异常时间戳 {time.ctime(timestamp)}"
                )
        
        # 提取签名算法
        sig_algo_match = re.search(r'Signature Algorithm: (.+)', block)
        if sig_algo_match:
            sig_algo = sig_algo_match.group(1).lower()
            
            # 检查是否使用弱签名算法
            weak_algorithms = ['md5', 'sha1', 'rsa-1024']
            for weak_algo in weak_algorithms:
                if weak_algo in sig_algo:
                    suspicious_scts.append(
                        f"SCT {i}: 使用弱签名算法 {sig_algo}"
                    )
    
    return suspicious_scts

# Ubuntu/Debian
sudo apt-get install openssl

# 检查是否支持CT扩展
openssl version -a | grep enable-ct

# 提取证书中的SCT
openssl x509 -in certificate.pem -text -noout -certopt ext_parse,ctlsinfo

# 验证证书的CT合规性
openssl verify -trusted root.pem -ct_check certificate.pem

# 从源码编译
git clone https://github.com/google/certificate-transparency.git
cd certificate-transparency/cpp/client
make

# 查询日志状态
ct-client --log_uri https://ct.googleapis.com/pilot get-sth

# 获取包含证明
ct-client --log_uri https://ct.googleapis.com/pilot get-proof-by-hash \
  --hash <base64url-encoded-hash> --tree_size <tree-size>

# Ubuntu/Debian
sudo apt-get install certspotter

# 或从源码安装
go install software.sslmate.com/src/certspotter/cmd/certspotter@latest

# 监控特定域名
certspotter watch example.com

# 搜索与域名相关的证书
certspotter search example.com

# 从PyPI安装
pip install x509lint

# 检查证书
x509lint certificate.pem

# 仅检查CT相关内容
x509lint --only=ct certificate.pem

pip install certificate-transparency

from ct.client import log_client
from ct.proto import ct_pb2

# 创建CT日志客户端
client = log_client.LogClient("https://ct.googleapis.com/pilot")

# 获取日志状态
sth_response = client.get_sth()
print(f"树大小: {sth_response.tree_size}")
print(f"根哈希: {sth_response.sha256_root_hash.hex()}")

# 获取日志条目
entries = client.get_entries(start=0, end=10)
for entry in entries.entries:
    print(f"条目索引: {entry.leaf_index}")
    # 处理条目...

pip install pyct

import pyct

# 加载证书
cert = pyct.load_certificate("certificate.pem")

# 提取SCT
scts = pyct.extract_scts(cert)
for sct in scts:
    print(f"SCT版本: {sct.version}")
    print(f"日志ID: {sct.log_id.hex()}")
    print(f"时间戳: {sct.timestamp}")

# 验证SCT
for sct in scts:
    log_key = pyct.fetch_log_key(sct.log_id)
    if log_key:
        is_valid = pyct.verify_sct(sct, log_key, cert)
        print(f"SCT验证: {'通过' if is_valid else '失败'}")

go get github.com/google/certificate-transparency-go

package main

import (
	"fmt"
	"github.com/google/certificate-transparency-go/client"
	"net/http"
	"time"
)

func main() {
	// 创建HTTP客户端
	httpClient := &http.Client{Timeout: 10 * time.Second}
	
	// 创建CT日志客户端
	logClient, err := client.New("https://ct.googleapis.com/pilot", httpClient)
	if err != nil {
		fmt.Printf("创建客户端失败: %v\n", err)
		return
	}
	
	// 获取日志状态
	sth, err := logClient.GetSTH()
	if err != nil {
		fmt.Printf("获取STH失败: %v\n", err)
		return
	}
	
	fmt.Printf("树大小: %d\n", sth.TreeSize)
	fmt.Printf("根哈希: %x\n", sth.SHA256RootHash)
}

pip install cryptography

from cryptography import x509
from cryptography.hazmat.backends import default_backend

# 加载证书
with open("certificate.pem", "rb") as f:
    cert_data = f.read()

# 解析证书
cert = x509.load_pem_x509_certificate(cert_data, default_backend())

# 查找CT相关扩展
for ext in cert.extensions:
    # 检查是否为SCT扩展
    if ext.oid._name == "ct_precert_scts":
        print("找到SCT扩展")
        # 处理SCT数据...

# 搜索特定域名的证书
example.com

# 搜索包含特定关键词的证书
%.example.com

# 搜索特定组织的证书
org:ExampleOrg

import schedule
import time
import requests
import json
import smtplib
from email.message import EmailMessage
from cryptography import x509
from cryptography.hazmat.backends import default_backend

# 配置
MONITORED_DOMAINS = ["example.com", "test.org"]
ALERT_EMAIL = "admin@example.com"
EMAIL_SERVER = "smtp.example.com"
EMAIL_PORT = 587
EMAIL_USER = "alert@example.com"
EMAIL_PASSWORD = "password"

# 存储已知证书
known_certificates = set()

def fetch_certificates(domain):
    """从crt.sh获取域名相关证书"""
    url = f"https://crt.sh/?q={domain}&output=json"
    try:
        response = requests.get(url)
        if response.status_code == 200:
            return response.json()
    except Exception as e:
        print(f"获取证书失败: {e}")
    return []

def send_alert(subject, content):
    """发送告警邮件"""
    msg = EmailMessage()
    msg.set_content(content)
    msg["Subject"] = subject
    msg["From"] = EMAIL_USER
    msg["To"] = ALERT_EMAIL
    
    try:
        with smtplib.SMTP(EMAIL_SERVER, EMAIL_PORT) as server:
            server.starttls()
            server.login(EMAIL_USER, EMAIL_PASSWORD)
            server.send_message(msg)
        print(f"告警邮件已发送: {subject}")
    except Exception as e:
        print(f"发送邮件失败: {e}")

def check_new_certificates():
    """检查新签发的证书"""
    for domain in MONITORED_DOMAINS:
        certificates = fetch_certificates(domain)
        for cert in certificates:
            cert_id = cert.get("id")
            if cert_id and cert_id not in known_certificates:
                # 新证书
                known_certificates.add(cert_id)
                
                # 获取证书详情
                subject = cert.get("name_value", "Unknown")
                issuer = cert.get("issuer_name", "Unknown")
                not_before = cert.get("not_before", "Unknown")
                not_after = cert.get("not_after", "Unknown")
                
                # 发送告警
                alert_subject = f"[CT监控] 发现新证书: {subject}"
                alert_content = f"""
                域名: {domain}
                主体: {subject}
                颁发者: {issuer}
                生效时间: {not_before}
                过期时间: {not_after}
                证书ID: {cert_id}
                """
                send_alert(alert_subject, alert_content)

def main():
    # 初始加载已知证书
    for domain in MONITORED_DOMAINS:
        certificates = fetch_certificates(domain)
        for cert in certificates:
            cert_id = cert.get("id")
            if cert_id:
                known_certificates.add(cert_id)
    print(f"初始加载了 {len(known_certificates)} 个已知证书")
    
    # 定期检查
    schedule.every(1).hour.do(check_new_certificates)
    
    print("CT监控已启动...")
    while True:
        schedule.run_pending()
        time.sleep(60)

if __name__ == "__main__":
    main()