Ed25519 + Merkle Tree + UUIDv7 = 构建防篡改决策日志

Source: Dev.to

TL;DR – 将 Ed25519 签名、Merkle 树和 UUIDv7 标识符结合起来，为 AI 与算法系统构建不可变、可防篡改的审计链。下面提供了完整的 Python 实现。

问题背景：光速般的 AI 决策

你的交易算法在过去的一秒内做出了 10 000 条决策。
你的机器学习模型批准了 500 份贷款申请。
你的自主系统执行了一次关键操作。

现在有人问：“14:32:07.847 时到底发生了什么？”

你需要证明：

• 做了什么 决策。
• 何时 做出的（具备密码学确定性）。
• 事后未被篡改。

传统日志记录无法满足：数据库记录可以被编辑，时间戳可以伪造，日志文件可以被篡改。当 AI 系统的运行速度超出人类理解时，我们需要密码学证据。

三大支柱

每个原语解决特定问题，组合在一起形成不可破的证据链。

支柱 1：UUIDv7 — 时间有序身份

UUIDv7（RFC 9562）在标识符中直接嵌入 48 位毫秒级 Unix 时间戳，具有：

• 字典序排序 = 按时间顺序排序
• 嵌入式时间证明，无需单独的 timestamp 字段
• 无需中心协调的唯一性

# uuidv7_example.py
import uuid
import time

def generate_uuidv7() -> str:
    """Generate a UUIDv7 identifier with embedded timestamp."""
    # Current time in milliseconds
    timestamp_ms = int(time.time() * 1000)

    # UUIDv7 structure:
    # - 48 bits: timestamp (ms)
    # - 4 bits: version (7)
    # - 12 bits: random
    # - 2 bits: variant
    # - 62 bits: random

    # Python 3.12+ includes uuid7; for earlier versions use the `uuid7` package
    return str(uuid.uuid.7())
    # Example output: "019234ab-cdef-7000-8123-456789abcdef"
    #                  ^^^^^^^^ timestamp embedded here

# uuidv7_utils.py
def extract_timestamp_from_uuidv7(uuid_str: str) -> int:
    """Extract the embedded millisecond timestamp from a UUIDv7."""
    uuid_hex = uuid_str.replace("-", "")
    # First 48 bits (12 hex chars) contain the timestamp
    return int(uuid_hex[:12], 16)

def detect_timestamp_anomaly(event_id: str, claimed_timestamp_ms: int,
                             threshold_ms: int = 5000) -> bool:
    """
    Return True if the claimed timestamp deviates from the UUIDv7 embedded
    timestamp by more than `threshold_ms`.
    """
    embedded_ts = extract_timestamp_from_uuidv7(event_id)
    drift = abs(embedded_ts - claimed_timestamp_ms)
    return drift > threshold_ms

为何重要 – 如果某事件声称在时间 T 发生，但 UUIDv7 中的时间戳显示为 T + 5 分钟，则差异在密码学上显而易见。

支柱 2：Hash Chains — 防篡改链接

每个事件存储前一个事件的哈希，形成不可变链：

Event₁ → H(Event₁) → Event₂ → H(Event₂) → Event₃ → …
          ↓                     ↓
      prev_hash            prev_hash

若任意早期事件被修改，其哈希会变化，导致后续所有事件的链接断裂。

# hash_chain.py
import hashlib
import json

# Genesis hash: 64 zeros (256 bits)
GENESIS_HASH = "0" * 64

def canonicalize_json(obj: dict) -> str:
    """
    RFC 8785 JSON Canonicalization:
    - Lexicographic key ordering
    - No whitespace
    - Deterministic number formatting
    """
    return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False)

def compute_event_hash(header: dict, payload: dict, prev_hash: str,
                       algo: str = "sha256") -> str:
    """
    Compute event hash as:
    H( canonical(header) || canonical(payload) || prev_hash )
    """
    data = (
        canonicalize_json(header) +
        canonicalize_json(payload) +
        prev_hash
    ).encode("utf-8")

    if algo == "sha256":
        return hashlib.sha256(data).hexdigest()
    elif algo == "sha3_256":
        return hashlib.sha3_256(data).hexdigest()
    else:
        raise ValueError(f"Unsupported algorithm: {algo}")

# chain_validation.py
def validate_chain(events: list[dict]) -> tuple[bool, str]:
    """
    Validate the integrity of an event chain.
    Returns (is_valid, message).
    """
    if not events:
        return True, "Empty chain"

    # First event must reference the genesis hash
    if events[0]["security"]["prev_hash"] != GENESIS_HASH:
        return False, "Genesis event has incorrect prev_hash"

    # Verify each subsequent event
    for i in range(1, len(events)):
        cur = events[i]
        prev = events[i - 1]

        # Expected hash of the previous event
        expected_prev_hash = compute_event_hash(
            prev["header"], prev["payload"], prev["security"]["prev_hash"]
        )

        # Linkage check
        if cur["security"]["prev_hash"] != expected_prev_hash:
            return False, f"Chain broken at event {i}: prev_hash mismatch"

        # Verify current event's own hash
        computed_hash = compute_event_hash(
            cur["header"], cur["payload"], cur["security"]["prev_hash"]
        )
        if cur["security"]["event_hash"] != computed_hash:
            return False, f"Event {i} hash mismatch: content was modified"

    return True, "Chain valid"

支柱 3：Ed25519 签名 — 真实性与不可否认性

Hash 链保证 完整性，但不能确认 谁 创建了记录。Ed25519（RFC 8032）提供快速、确定性的 256 位安全签名。

# ed25519_signer.py
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PrivateKey, Ed25519PublicKey
)
from cryptography.hazmat.primitives import serialization

class EventSigner:
    """Ed25519 signer for VCP events."""

    def __init__(self, private_key: Ed25519PrivateKey = None):
        self.private_key = private_key or Ed25519PrivateKey.generate()
        self.public_key = self.private_key.public_key()

    def get_public_key_hex(self) -> str:
        """Export public key as a hex string."""
        pk_bytes = self.public_key.public_bytes(
            encoding=serialization.Encoding.Raw,
            format=serialization.PublicFormat.Raw
        )
        return pk_bytes.hex()

    def sign_event(self, event_hash: str) -> str:
        """Sign an event hash; returns hex‑encoded signature."""
        signature = self.private_key.sign(bytes.fromhex(event_hash))
        return signature.hex()

    @staticmethod
    def verify_signature(public_key_hex: str, event_hash: str,
                         signature_hex: str) -> bool:
        """Verify a signature; returns False on failure."""
        try:
            public_key = Ed25519PublicKey.from_public_bytes(
                bytes.fromhex(public_key_hex)
            )
            public_key.verify(
                bytes.fromhex(signature_hex),
                bytes.fromhex(event_hash)
            )
            return True
        except Exception:
            return False

综合实现

1. 为事件 生成 UUIDv7——提供密码学时间戳。
2. 构建事件负载（header、payload 等），计算其哈希，并链入前一个事件的哈希。
3. 使用 Ed25519 私钥 签署事件哈希，并将签名及公钥标识一起存储。
4. 可选：将大量事件批量放入 Merkle 树（RFC 6962），实现高效的整体验证。

最终日志具备：

• 通过 UUIDv7 实现的时间顺序。
• 通过哈希链（以及可选的 Merkle 证明）实现的防篡改。
• 通过 Ed25519 签名实现的真实性与不可否认性。

这种组合为高吞吐量的 AI 与自主系统提供了坚固、可防篡改的决策日志。