证书透明度（CT）日志与国密 PKI：如何在国密体系中构建可信的证书审计机制

# 国际 SCT 签名数据
def sct_signature_input(version, timestamp, entry_type, cert_data):
    return (version.to_bytes(1, 'big') +
            timestamp.to_bytes(8, 'big') +
            entry_type.to_bytes(2, 'big') +
            cert_data)

# 国密 SCT 签名数据（使用 SM2 的 Z 值）
def gm_sct_signature_input(version, timestamp, entry_type, cert_data, log_pubkey):
    data = (version.to_bytes(1, 'big') +
            timestamp.to_bytes(8, 'big') +
            entry_type.to_bytes(2, 'big') +
            cert_data)
    # SM2 签名需要计算 Z 值
    Z = sm3_z_value(user_id="1234567812345678", pubkey=log_pubkey)
    return Z + data

# 克隆 Trillian
git clone https://github.com/google/trillian.git
cd trillian

# 启动依赖服务（MySQL/MariaDB）
docker-compose up -d mysql

# 初始化数据库
go run ./cmd/createtree --admin_server=localhost:8090

# 启动日志服务器
go run ./cmd/trillian_log_server \
    --log_config=examples/ct/config.pb \
    --rpc_endpoint=localhost:8090 \
    --http_endpoint=localhost:8091

// gm_signer.go
package gm

import (
    "crypto"
    "github.com/tjfoc/gmsm/sm2"
    "github.com/tjfoc/gmsm/sm3"
)

type GMSigner struct {
    privateKey *sm2.PrivateKey
}

func (s *GMSigner) Sign(rand io.Reader, digest []byte, opts crypto.SignerOpts) ([]byte, error) {
    return s.privateKey.Sign(rand, digest, nil)
}

func (s *GMSigner) Public() crypto.PublicKey {
    return &s.privateKey.PublicKey
}

type GMVerifier struct {
    publicKey *sm2.PublicKey
}

func (v *GMVerifier) Verifydigest []byte, sig []byte) bool {
    return v.publicKey.Verify(digest, sig)
}

// 默克尔树哈希使用 SM3
func MerkleTreeHash(left, right []byte) []byte {
    h := sm3.New()
    h.Write([]byte{0x01})
    h.Write(left)
    h.Write(right)
    return h.Sum(nil)
}

from gmssl import sm2, func

# 生成 SM2 密钥对
private_key = sm2.CryptSM2()
private_key.generate_key()

# 导出公钥（用于客户端验证 SCT）
public_key = private_key.get_public_key()

# 保存密钥
with open('ct_log_private.pem', 'w') as f:
    f.write(private_key.export_private_key().hex())
with open('ct_log_public.pem', 'w') as f:
    f.write(public_key)

# simplified_ct_log.py
"""
简化的国密 CT 日志服务器
基于 Flask + SQLite + SM2/SM3
"""
import json
import sqlite3
import time
from gmssl import sm2, sm3, func
from flask import Flask, request, jsonify

app = Flask(__name__)

class GMCTLog:
    def __init__(self, db_path='ct_log.db'):
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        self._init_db()
        self.tree = []
        self.log_key = self._load_or_create_key()
    
    def _init_db(self):
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS certificates (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                cert_pem TEXT NOT NULL,
                cert_hash TEXT NOT NULL,
                leaf_hash TEXT NOT NULL,
                timestamp INTEGER NOT NULL,
                sct TEXT
            )
        ''')
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS sth (
                id INTEGER PRIMARY KEY,
                tree_size INTEGER NOT NULL,
                timestamp INTEGER NOT NULL,
                root_hash TEXT NOT NULL,
                signature TEXT NOT NULL
            )
        ''')
        self.conn.commit()
    
    def _load_or_create_key(self):
        """加载或创建日志服务器的 SM2 密钥对"""
        try:
            with open('ct_log_key.json', 'r') as f:
                key_data = json.load(f)
            return sm2.CryptSM2(
                private_key=key_data['private_key'],
                public_key=key_data['public_key']
            )
        except FileNotFoundError:
            crypt = sm2.CryptSM2()
            crypt.generate_key()
            key_data = {
                'private_key': crypt.export_private_key().hex(),
                'public_key': crypt.get_public_key()
            }
            with open('ct_log_key.json', 'w') as f:
                json.dump(key_data, f)
            return crypt
    
    def _sm3_hash(self, data: bytes) -> str:
        return sm3.sm3_hash(func.bytes_to_list(data))
    
    def _leaf_hash(self, cert_data: bytes) -> str:
        return self._sm3_hash(b'\x00' + cert_data)
    
    def _node_hash(self, left: str, right: str) -> str:
        return self._sm3_hash(b'\x01' + bytes.fromhex(left) + bytes.fromhex(right))
    
    def _compute_root_hash(self, leaf_hashes: list) -> str:
        """计算默克尔树根哈希"""
        if not leaf_hashes:
            return self._sm3_hash(b'')
        
        level = leaf_hashes[:]
        while len(level) > 1:
            next_level = []
            for i in range(0, len(level), 2):
                left = level[i]
                right = level[i + 1] if i + 1 < len(level) else left
                next_level.append(self._node_hash(left, right))
            level = next_level
        return level[0]
    
    def add_cert(self, cert_pem: str) -> dict:
        """添加证书到 CT 日志，返回 SCT"""
        cert_data = cert_pem.encode('utf-8')
        cert_hash = self._sm3_hash(cert_data)
        leaf_hash = self._leaf_hash(cert_data)
        timestamp = int(time.time() * 1000)
        
        # 保存证书
        cursor = self.conn.execute(
            'INSERT INTO certificates (cert_pem, cert_hash, leaf_hash, timestamp) VALUES (?, ?, ?, ?)',
            (cert_pem, cert_hash, leaf_hash, timestamp)
        )
        entry_id = cursor.lastrowid
        
        # 更新默克尔树
        self.tree.append(leaf_hash)
        root_hash = self._compute_root_hash(self.tree)
        tree_size = len(self.tree)
        
        # 生成 SCT 签名
        sct_data = (
            b'\x00' +  # version
            timestamp.to_bytes(8, 'big') +
            b'\x00\x00' +  # entry_type = X.509
            cert_data
        )
        sct_signature = self.log_key.sign(
            self._sm3_hash(sct_data).encode('utf-8')
        )
        
        sct = {
            'version': 0,
            'timestamp': timestamp,
            'entry_type': 0,
            'signature': sct_signature.hex()
        }
        
        # 保存 SCT
        self.conn.execute(
            'UPDATE certificates SET sct = ? WHERE id = ?',
            (json.dumps(sct), entry_id)
        )
        
        # 保存 STH（Signed Tree Head）
        self.conn.execute(
            'INSERT OR REPLACE INTO sth (id, tree_size, timestamp, root_hash, signature) VALUES (1, ?, ?, ?, ?)',
            (tree_size, timestamp, root_hash, sct_signature.hex())
        )
        self.conn.commit()
        
        return {
            'sct': sct,
            'entry_id': entry_id,
            'leaf_hash': leaf_hash,
            'tree_size': tree_size
        }
    
    def get_sth(self) -> dict:
        """获取当前 STH"""
        row = self.conn.execute('SELECT * FROM sth WHERE id = 1').fetchone()
        if not row:
            return None
        return {
            'tree_size': row[1],
            'timestamp': row[2],
            'root_hash': row[3],
            'signature': row[4]
        }
    
    def get_proof(self, cert_hash: str) -> dict:
        """获取证书的包含性证明"""
        rows = self.conn.execute(
            'SELECT id, leaf_hash FROM certificates ORDER BY id'
        ).fetchall()
        
        leaf_hashes = [row[1] for row in rows]
        target_idx = None
        for i, row in enumerate(rows):
            if row[1] == cert_hash:
                target_idx = i
                break
        
        if target_idx is None:
            return None
        
        # 计算审计路径
        audit_path = []
        level = leaf_hashes[:]
        idx = target_idx
        while len(level) > 1:
            next_level = []
            for i in range(0, len(level), 2):
                left = level[i]
                right = level[i + 1] if i + 1 < len(level) else left
                next_level.append(self._node_hash(left, right))
            
            sibling_idx = idx ^ 1  # XOR 1 获取兄弟节点索引
            if sibling_idx < len(level):
                audit_path.append(level[sibling_idx])
            
            level = next_level
            idx //= 2
        
        return {
            'leaf_index': target_idx,
            'audit_path': audit_path,
            'root_hash': level[0] if level else ''
        }


# Flask API
ct_log = GMCTLog()

@app.route('/ct/v1/add-chain', methods=['POST'])
def add_chain():
    data = request.get_json()
    certs = data.get('chain', [])
    results = []
    for cert in certs:
        result = ct_log.add_cert(cert)
        results.append(result['sct'])
    return jsonify({'scts': results})

@app.route('/ct/v1/get-sth', methods=['GET'])
def get_sth():
    sth = ct_log.get_sth()
    if not sth:
        return jsonify({'error': 'no STH available'}), 404
    return jsonify(sth)

@app.route('/ct/v1/get-proof-by-hash', methods=['GET'])
def get_proof_by_hash():
    cert_hash = request.args.get('hash')
    proof = ct_log.get_proof(cert_hash)
    if not proof:
        return jsonify({'error': 'certificate not found'}), 404
    return jsonify(proof)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=6962)

"""国密 SCT 验证"""
from gmssl import sm2, sm3, func

def verify_gm_sct(sct, cert_pem, log_pubkey_hex):
    """验证国密 SCT"""
    # 1. 构造签名输入
    timestamp = sct['timestamp']
    cert_data = cert_pem.encode('utf-8')
    
    sct_data = (
        b'\x00' +  # version
        timestamp.to_bytes(8, 'big') +
        b'\x00\x00' +  # entry_type
        cert_data
    )
    
    # 2. 计算 SM3 哈希
    digest = sm3.sm3_hash(func.bytes_to_list(sct_data))
    
    # 3. 使用日志服务器的 SM2 公钥验证签名
    log_pubkey = sm2.CryptSM2(public_key=log_pubkey_hex, private_key="")
    signature = bytes.fromhex(sct['signature'])
    
    return log_pubkey.verify(signature, digest.encode('utf-8'))


def verify_merkle_proof(leaf_hash, audit_path, root_hash, leaf_index):
    """验证默克尔树包含性证明"""
    current = bytes.fromhex(leaf_hash)
    
    for i, sibling_hex in enumerate(audit_path):
        sibling = bytes.fromhex(sibling_hex)
        # 根据索引的比特位决定左右顺序
        if (leaf_index >> i) & 1 == 0:
            current = sm3_hash(b'\x01' + current + sibling)
        else:
            current = sm3_hash(b'\x01' + sibling + current)
    
    return current.hex() == root_hash

┌─────────────┐
                    │  内部 CA     │
                    │ (SM2/SM3)   │
                    └──────┬──────┘
                           │ 签发证书 + 提交到 CT 日志
                           ▼
                    ┌─────────────┐
                    │  国密 CT     │
                    │  日志服务器   │
                    └──────┬──────┘
                           │ STH + 审计路径
              ┌────────────┼────────────┐
              ▼            ▼            ▼
        ┌──────────┐ ┌──────────┐ ┌──────────┐
        │ Monitor  │ │ Auditor  │ │ 浏览器    │
        │ (监控)   │ │ (审计)   │ │ (验证)   │
        └──────────┘ └──────────┘ └──────────┘