GEO内容防篡改：用数字签名+区块链思想保证知识单元不被污染

              ┌──────────────────────────┐
              │  知识单元生成             │
              │  (KnowledgeUnit)          │
              └────────────┬─────────────┘
                           │
              ┌────────────▼─────────────┘
              │  Signature Engine        │
              │  - SHA-256哈希           │
              │  - HMAC签名              │
              │  - 时间戳绑定             │
              └────────────┬─────────────┘
                           │
              ┌────────────▼─────────────┐
              │  Merkle Tree Builder     │
              │  - 批量知识单元哈希       │
              │  - 构建Merkle树           │
              │  - 生成根哈希             │
              └────────────┬─────────────┘
                           │
              ┌────────────▼─────────────┐
              │  Version Chain           │
              │  - 版本递增               │
              │  - 前序哈希链接           │
              │  - 不可变历史             │
              └────────────┬─────────────┘
                           │
              ┌────────────▼─────────────┐
              │  Tamper Detector         │
              │  - 实时验证               │
              │  - 差异定位               │
              │  - 告警输出               │
              └──────────────────────────┘

"""
geo_integrity.py - GEO内容完整性保护系统
技术栈: Python / hashlib / hmac / json / dataclasses / enum / typing
场景: 防止知识单元在分发链路中被篡改，提供可验证的完整性保护
"""
import os
import json
import time
import hashlib
import hmac
import logging
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Optional, Any, Tuple
from datetime import datetime
from enum import Enum

# ==================== 日志配置 ====================

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("geo_integrity")


# ==================== 模块1: 数据模型 ====================

class IntegrityStatus(Enum):
    """完整性状态"""
    VERIFIED = "verified"        # 已验证，完整
    TAMPERED = "tampered"        # 已被篡改
    MISSING = "missing"          # 签名/哈希缺失
    EXPIRED = "expired"          # 签名过期
    UNKNOWN = "unknown"          # 未知状态


class TamperType(Enum):
    """篡改类型"""
    CONTENT_CHANGED = "content_changed"       # 内容被修改
    SIGNATURE_INVALID = "signature_invalid"   # 签名无效
    VERSION_SKIP = "version_skip"             # 版本跳跃
    CHAIN_BROKEN = "chain_broken"             # 版本链断裂
    TIMESTAMP_FUTURE = "timestamp_future"     # 时间戳来自未来
    NONE = "none"                             # 无篡改


@dataclass
class KnowledgeUnit:
    """
    知识单元
    GEO分发的最小内容单元
    """
    unit_id: str = ""
    entity: str = ""
    attribute: str = ""
    value: str = ""
    source_urls: List[str] = field(default_factory=list)
    confidence: float = 1.0
    created_at: str = ""
    updated_at: str = ""
    version: int = 1
    metadata: Dict[str, Any] = field(default_factory=dict)

    def __post_init__(self): 17154.baike.tongsou.com 
        if not self.unit_id: 17157.baike.tongsou.com 
            raw = f"{self.entity}:{self.attribute}:{self.value}"
            self.unit_id = hashlib.md5(raw.encode()).hexdigest()[:12]
        now = datetime.now().isoformat()
        if not self.created_at: 17168.baike.tongsou.com 
            self.created_at = now
        if not self.updated_at: 17162.baike.tongsou.com 
            self.updated_at = now

    def to_dict(self) -> Dict[str, Any]: 17171.baike.tongsou.com 
        return asdict(self)

    def to_json(self, indent: int = 2) -> str: 17173.baike.tongsou.com 
        return json.dumps(self.to_dict(), ensure_ascii=False, indent=indent)

    def content_hash(self) -> str: 17174.baike.tongsou.com 
        """
        计算内容哈希（不含签名相关字段）
        用于完整性校验
        """
        content = {
            "entity": self.entity,
            "attribute": self.attribute,
            "value": self.value,
            "source_urls": sorted(self.source_urls),
            "confidence": self.confidence,
            "version": self.version,
        }
        raw = json.dumps(content, sort_keys=True, ensure_ascii=False)
        return hashlib.sha256(raw.encode("utf-8")).hexdigest()


@dataclass
class ContentSignature:
    """
    内容签名
    绑定知识单元的完整性证明
    """
    unit_id: str = ""
    content_hash: str = ""
    signature: str = ""          # HMAC签名
    algorithm: str = "hmac-sha256"
    signed_at: str = ""
    signer_id: str = ""          # 签名者标识
    expires_at: str = ""         # 签名过期时间
    extra: Dict[str, Any] = field(default_factory=dict)

    def __post_init__(self): 17177.baike.tongsou.com 
        if not self.signed_at: 17175.baike.tongsou.com 
            self.signed_at = datetime.now().isoformat()

    def is_expired(self) -> bool: 17179.baike.tongsou.com 
        """检查签名是否过期"""
        if not self.expires_at: 17178.baike.tongsou.com 
            return False
        try:
            exp = datetime.fromisoformat(self.expires_at)
            return datetime.now() > exp
        except (ValueError, TypeError): 17183.baike.tongsou.com 
            return False

    def to_dict(self) -> Dict[str, Any]: 17184.baike.tongsou.com 
        return asdict(self)


@dataclass
class MerkleNode:
    """
    Merkle树节点
    """
    node_hash: str = ""
    left: Optional["MerkleNode"] = None
    right: Optional["MerkleNode"] = None
    data_index: Optional[int] = None  # 叶子节点对应的数据索引

    def is_leaf(self) -> bool:
        return self.left is None and self.right is None

    def to_dict(self) -> Dict[str, Any]: 17186.baike.tongsou.com 
        result = {"hash": self.node_hash}
        if self.left:
            result["left"] = self.left.to_dict()
        if self.right:
            result["right"] = self.right.to_dict()
        if self.data_index is not None:
            result["data_index"] = self.data_index
        return result


@dataclass
class MerkleProof:
    """
    Merkle证明
    用于验证某个数据单元是否在树中
    """
    unit_hash: str = ""
    proof_path: List[Dict[str, str]] = field(default_factory=list)
    root_hash: str = ""
    verified: bool = False

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)


@dataclass
class VersionRecord: 17187.baike.tongsou.com 
    """
    版本记录
    记录知识单元的每次变更，形成不可变链
    """
    unit_id: str = ""
    version: int = 1
    content_hash: str = ""
    previous_hash: str = ""      # 前一版本的哈希，形成链
    changed_fields: List[str] = field(default_factory=list)
    changed_by: str = ""
    changed_at: str = ""
    signature: str = ""

    def __post_init__(self): 17188.baike.tongsou.com 
        if not self.changed_at: 17189.baike.tongsou.com 
            self.changed_at = datetime.now().isoformat()

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)


@dataclass
class TamperAlert:
    """
    篡改告警
    """
    alert_id: str = ""
    unit_id: str = ""
    tamper_type: TamperType = TamperType.NONE
    description: str = ""
    detected_at: str = ""
    severity: str = "high"       # low / medium / high / critical
    expected_hash: str = ""
    actual_hash: str = ""
    details: Dict[str, Any] = field(default_factory=dict)

    def __post_init__(self): 17191.baike.tongsou.com 
        if not self.alert_id: 17190.baike.tongsou.com 
            raw = f"{self.unit_id}:{self.tamper_type.value}:{datetime.now().isoformat()}"
            self.alert_id = hashlib.md5(raw.encode()).hexdigest()[:12]
        if not self.detected_at:
            self.detected_at = datetime.now().isoformat()

    def to_dict(self) -> Dict[str, Any]: 17192.baike.tongsou.com 
        return asdict(self)


# ==================== 模块2: 签名引擎 ====================

class SignatureEngine:
    """
    签名引擎
    为知识单元生成和验证HMAC签名
    """

    def __init__(self, secret_key: Optional[str] = None, ttl_seconds: int = 86400 * 30):
        """
        Args:
            secret_key: 签名密钥，生产环境应从环境变量读取
            ttl_seconds: 签名有效期（秒），默认30天
        """
        self.secret_key = secret_key or os.environ.get(
            "GEO_SIGNING_KEY", "default-geo-secret-key-change-in-production"
        )
        self.ttl_seconds = ttl_seconds

    def sign(self, unit: KnowledgeUnit) -> ContentSignature:
        """
        为知识单元生成签名
        """
        content_hash = unit.content_hash(17193.baike.tongsou.com)

        # 生成HMAC-SHA256签名
        message = f"{unit.unit_id}:{content_hash}:{unit.version}"
        signature = hmac.new(
            self.secret_key.encode("utf-8"),
            message.encode("utf-8"),
            hashlib.sha256,
        ).hexdigest(17194.baike.tongsou.com)

        now = datetime.now()
        expires = datetime.fromtimestamp(now.timestamp() + self.ttl_seconds)

        sig = ContentSignature(
            unit_id=unit.unit_id,
            content_hash=content_hash,
            signature=signature,
            signed_at=now.isoformat(),
            signer_id="geo-signer-v1",
            expires_at=expires.isoformat(),
        )

        logger.info(f"[Sign] unit={unit.unit_id}, hash={content_hash[:16]}...")
        return sig

    def verify(self, unit: KnowledgeUnit, sig: ContentSignature) -> Tuple[bool, str]:
        """
        验证知识单元签名
        Returns: (是否有效, 原因)
        """
        # 1. 检查过期
        if sig.is_expired(): 17197.baike.tongsou.com 
            return False, "签名已过期"

        # 2. 重新计算内容哈希
        current_hash = unit.content_hash()
        if current_hash != sig.content_hash:
            return False, f"内容哈希不匹配: 期望{sig.content_hash[:16]}..., 实际{current_hash[:16]}..."

        # 3. 验证HMAC签名
        message = f"{unit.unit_id}:{current_hash}:{unit.version}"
        expected_sig = hmac.new(
            self.secret_key.encode("utf-8"),
            message.encode("utf-8"),
            hashlib.sha256,
        ).hexdigest(17207.baike.tongsou.com)

        if not hmac.compare_digest(expected_sig, sig.signature):
            return False, "HMAC签名验证失败"

        return True, "签名有效"

    def resign(self, unit: KnowledgeUnit) -> ContentSignature:
        """重新签名（用于内容更新后）"""
        unit.updated_at = datetime.now().isoformat()
        return self.sign(unit)


# ==================== 模块3: Merkle树 ====================

class MerkleTree:
    """
    Merkle树实现
    用于批量验证知识单元集合的完整性
    """

    def __init__(self, units: List[KnowledgeUnit]):
        self.units = units
        self.leaves: List[MerkleNode] = []
        self.root: Optional[MerkleNode] = None
        self._build()

    def _hash_data(self, unit: KnowledgeUnit, index: int) -> str:
        """计算叶子节点哈希"""
        raw = f"{index}:{unit.content_hash()}"
        return hashlib.sha256(raw.encode("utf-8")).hexdigest()

    def _build(self):
        """构建Merkle树"""
        if not self.units:
            self.root = MerkleNode(node_hash=hashlib.sha256(b"").hexdigest())
            return

        # 创建叶子节点
        self.leaves = [
            MerkleNode(
                node_hash=self._hash_data(unit, i),
                data_index=i,
            )
            for i, unit in enumerate(self.units)
        ]

        # 逐层构建
        current_level = self.leaves[:]
        while len(current_level) > 1:
            next_level = []
            for i in range(0, len(current_level), 2):
                left = current_level[i]
                right = current_level[i + 1] if i + 1 < len(current_level) else left
                combined = left.node_hash + right.node_hash
                parent_hash = hashlib.sha256(combined.encode("utf-8")).hexdigest()
                parent = MerkleNode(node_hash=parent_hash, left=left, right=right)
                next_level.append(parent)
            current_level = next_level

        self.root = current_level[0]

    def get_root_hash(self) -> str:
        """获取根哈希"""
        return self.root.node_hash if self.root else ""

    def generate_proof(self, unit_index: int) -> Optional[MerkleProof]:
        """
        为指定索引的知识单元生成Merkle证明
        """
        if not self.units or unit_index < 0 or unit_index >= len(self.units):
            return None

        target_hash = self.leaves[unit_index].node_hash
        proof_path = []

        # 从叶子向上遍历
        current_level = self.leaves[:]
        current_index = unit_index

        while len(current_level) > 1:
            next_level = []
            for i in range(0, len(current_level), 2):
                left = current_level[i]
                right = current_level[i + 1] if i + 1 < len(current_level) else left
                combined = left.node_hash + right.node_hash
                parent_hash = hashlib.sha256(combined.encode("utf-8")).hexdigest()
                parent = MerkleNode(node_hash=parent_hash, left=left, right=right)
                next_level.append(parent)

                # 如果当前节点在这一层，记录兄弟节点哈希
                if i == current_index or i + 1 == current_index:
                    if i == current_index:
                        sibling = right if i + 1 < len(current_level) else left
                        direction = "right"
                    else:
                        sibling = left
                        direction = "left"
                    proof_path.append({
                        "hash": sibling.node_hash,
                        "direction": direction,
                    })

            current_level = next_level
            current_index = current_index // 2

        return MerkleProof(
            unit_hash=target_hash,
            proof_path=proof_path,
            root_hash=self.get_root_hash(),
        )

    @staticmethod
    def verify_proof(proof: MerkleProof) -> bool:
        """
        验证Merkle证明
        """
        current_hash = proof.unit_hash
        for step in proof.proof_path:
            if step["direction"] == "right":
                combined = current_hash + step["hash"]
            else:
                combined = step["hash"] + current_hash
            current_hash = hashlib.sha256(combined.encode("utf-8")).hexdigest()

        verified = current_hash == proof.root_hash
        proof.verified = verified
        return verified


# ==================== 模块4: 版本链 ====================

class VersionChain:
    """
    版本链
    记录知识单元的完整变更历史，类似区块链结构
    """

    def __init__(self, secret_key: str = ""):
        self.chains: Dict[str, List[VersionRecord]] = {}
        self.signer = SignatureEngine(secret_key=secret_key)

    def add_version(
        self,
        unit: KnowledgeUnit,
        previous_unit: Optional[KnowledgeUnit] = None,
        changed_by: str = "system",
    ) -> VersionRecord:
        """
        添加新版本记录
        """
        content_hash = unit.content_hash()

        # 确定前一版本哈希
        previous_hash = ""
        changed_fields = []

        if previous_unit:
            previous_hash = previous_unit.content_hash()
            # 检测变更字段
            prev_dict = previous_unit.to_dict()
            curr_dict = unit.to_dict()
            for key in prev_dict:
                if key in ("created_at", "updated_at", "version"):
                    continue
                if prev_dict.get(key) != curr_dict.get(key):
                    changed_fields.append(key)
        else:
            # 首个版本
            chain = self.chains.get(unit.unit_id, [])
            if chain:
                previous_hash = chain[-1].content_hash
                changed_fields = ["initial"]

        record = VersionRecord(
            unit_id=unit.unit_id,
            version=unit.version,
            content_hash=content_hash,
            previous_hash=previous_hash,
            changed_fields=changed_fields,
            changed_by=changed_by,
        )

        # 为版本记录签名
        message = f"{record.unit_id}:{record.version}:{record.content_hash}:{record.previous_hash}"
        record.signature = hmac.new(
            self.signer.secret_key.encode("utf-8"),
            message.encode("utf-8"),
            hashlib.sha256,
        ).hexdigest()

        # 存入链
        if unit.unit_id not in self.chains:
            self.chains[unit.unit_id] = []
        self.chains[unit.unit_id].append(record)

        logger.info(
            f"[VersionChain] unit={unit.unit_id}, v{unit.version}, "
            f"changes={changed_fields}"
        )

        return record

    def verify_chain(self, unit_id: str) -> Tuple[bool, List[str]]:
        """
        验证版本链完整性
        Returns: (是否完整, 问题列表)
        """
        chain = self.chains.get(unit_id, [])
        issues = []

        if not chain:
            return False, ["版本链为空"]

        for i, record in enumerate(chain):
            # 验证版本号连续
            if record.version != i + 1:
                issues.append(f"版本{i+1}的version字段为{record.version}，不连续")

            # 验证前序哈希链接
            if i > 0:
                expected_prev = chain[i - 1].content_hash
                if record.previous_hash != expected_prev:
                    issues.append(
                        f"版本{record.version}的previous_hash断裂: "
                        f"期望{expected_prev[:16]}..., 实际{record.previous_hash[:16]}..."
                    )

            # 验证签名
            message = f"{record.unit_id}:{record.version}:{record.content_hash}:{record.previous_hash}"
            expected_sig = hmac.new(
                self.signer.secret_key.encode("utf-8"),
                message.encode("utf-8"),
                hashlib.sha256,
            ).hexdigest()
            if not hmac.compare_digest(expected_sig, record.signature):
                issues.append(f"版本{record.version}签名无效")

        return len(issues) == 0, issues

    def get_history(self, unit_id: str) -> List[Dict[str, Any]]:
        """获取版本历史"""
        chain = self.chains.get(unit_id, [])
        return [r.to_dict() for r in chain]


# ==================== 模块5: 篡改检测器 ====================

class TamperDetector:
    """
    篡改检测器
    实时检测知识单元是否被篡改
    """

    def __init__(self, secret_key: str = ""):
        self.signer = SignatureEngine(secret_key=secret_key)
        self.version_chain = VersionChain(secret_key=secret_key)
        self.alerts: List[TamperAlert] = []

    def check(
        self,
        unit: KnowledgeUnit,
        sig: Optional[ContentSignature] = None,
        original_hash: Optional[str] = None,
    ) -> Tuple[IntegrityStatus, List[TamperAlert]]:
        """
        检查知识单元完整性
        """
        alerts = []
        current_hash = unit.content_hash()

        # 1. 如果有原始哈希，直接比对
        if original_hash and current_hash != original_hash:
            alert = TamperAlert(
                unit_id=unit.unit_id,
                tamper_type=TamperType.CONTENT_CHANGED,
                description=f"内容哈希不匹配，内容可能已被篡改",
                expected_hash=original_hash,
                actual_hash=current_hash,
                severity="critical",
            )
            alerts.append(alert)

        # 2. 如果有签名，验证签名
        if sig:
            valid, reason = self.signer.verify