当《黑镜》照进现实：AI生成内容的链上溯源与C2PA标准之争

// SPDX-License-Identifier: MIT
pragma solidity ^0.8.20;

contract C2PAProvenanceRegistry {
    struct ContentClaim {
        string contentHash;      // 内容哈希值
        string creatorDID;       // 创作者去中心化身份
        uint256 timestamp;       // 创建时间戳
        string softwareVersion;  // 创作软件版本
        string editHistory;      // 编辑历史哈希
        bool isAIGenerated;      // 是否为AI生成
        address verifier;        // 验证者地址
    }

    mapping(bytes32 => ContentClaim) public claims;
    mapping(bytes32 => bool) public revokedClaims;

    event ClaimRegistered(bytes32 indexed claimId, string contentHash, address indexed creator);
    event ClaimRevoked(bytes32 indexed claimId, address indexed revoker);

    function registerClaim(
        string memory _contentHash,
        string memory _creatorDID,
        string memory _softwareVersion,
        string memory _editHistory,
        bool _isAIGenerated
    ) public returns (bytes32) {
        bytes32 claimId = keccak256(abi.encodePacked(_contentHash, msg.sender, block.timestamp));
        
        claims[claimId] = ContentClaim({
            contentHash: _contentHash,
            creatorDID: _creatorDID,
            timestamp: block.timestamp,
            softwareVersion: _softwareVersion,
            editHistory: _editHistory,
            isAIGenerated: _isAIGenerated,
            verifier: msg.sender
        });

        emit ClaimRegistered(claimId, _contentHash, msg.sender);
        return claimId;
    }

    function verifyClaim(bytes32 _claimId) public view returns (bool, ContentClaim memory) {
        require(bytes(claims[_claimId].contentHash).length > 0, "Claim not found");
        require(!revokedClaims[_claimId], "Claim has been revoked");
        return (true, claims[_claimId]);
    }

    function revokeClaim(bytes32 _claimId) public {
        require(claims[_claimId].verifier == msg.sender, "Not authorized");
        revokedClaims[_claimId] = true;
        emit ClaimRevoked(_claimId, msg.sender);
    }
}

import hashlib
import json
import time
from typing import Dict, Optional
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature
from cryptography.exceptions import InvalidSignature

class C2PAValidator:
    """C2PA内容来源验证器"""
    
    def __init__(self):
        self.trusted_verifiers = {}
        self.provenance_claims = {}
    
    def add_trusted_verifier(self, did: str, public_key_pem: str):
        key = serialization.load_pem_public_key(public_key_pem.encode())
        self.trusted_verifiers[did] = key
    
    def create_claim(self, content_hash: str, creator_did: str, 
                     is_ai_generated: bool, software: str) -> Dict:
        claim = {
            "content_hash": content_hash,
            "creator_did": creator_did,
            "timestamp": int(time.time()),
            "is_ai_generated": is_ai_generated,
            "software_version": software,
            "edit_history": [],
            "signature": None
        }
        return claim
    
    def sign_claim(self, claim: Dict, private_key_pem: str) -> Dict:
        private_key = serialization.load_pem_private_key(
            private_key_pem.encode(), password=None
        )
        claim_data = json.dumps(claim, sort_keys=True).encode()
        signature = private_key.sign(claim_data, ec.ECDSA(hashes.SHA256()))
        claim["signature"] = signature.hex()
        return claim
    
    def verify_claim(self, claim: Dict) -> tuple[bool, Optional[str]]:
        if claim["creator_did"] not in self.trusted_verifiers:
            return False, "Untrusted creator"
        
        signature_hex = claim.pop("signature", None)
        if not signature_hex:
            return False, "No signature found"
        
        try:
            public_key = self.trusted_verifiers[claim["creator_did"]]
            claim_data = json.dumps(claim, sort_keys=True).encode()
            signature = bytes.fromhex(signature_hex)
            public_key.verify(signature, claim_data, ec.ECDSA(hashes.SHA256()))
            return True, None
        except InvalidSignature:
            return False, "Invalid signature"
        finally:
            claim["signature"] = signature_hex
    
    def detect_ai_content(self, content_path: str) -> float:
        """检测AI生成概率（模拟实现）"""
        with open(content_path, "rb") as f:
            data = f.read()
        
        # 在实际实现中，这里会调用AI检测模型
        # 例如：频率分析、噪声模式检测、元数据异常检测
        entropy = len(hashlib.sha256(data).digest())
        return min(1.0, entropy / 256.0)


# 使用示例
validator = C2PAValidator()
validator.add_trusted_verifier(
    "did:key:creator123",
    "-----BEGIN PUBLIC KEY-----\nMFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE..."
)

claim = validator.create_claim(
    content_hash="sha256:abc123...",
    creator_did="did:key:creator123",
    is_ai_generated=True,
    software="Midjourney v6.2"
)

is_valid, error = validator.verify_claim(claim)
print(f"Claim valid: {is_valid}, Error: {error}")

const { ethers } = require('ethers');
const crypto = require('crypto');
const fs = require('fs');

class ContentProvenanceService {
  constructor(providerUrl, contractAddress, privateKey) {
    this.provider = new ethers.JsonRpcProvider(providerUrl);
    this.wallet = new ethers.Wallet(privateKey, this.provider);
    this.contractABI = [
      "function registerClaim(string memory _contentHash, string memory _creatorDID, string memory _softwareVersion, string memory _editHistory, bool _isAIGenerated) public returns (bytes32)",
      "function verifyClaim(bytes32 _claimId) public view returns (bool, tuple(string,string,uint256,string,string,bool,address))",
      "function revokeClaim(bytes32 _claimId) public"
    ];
    this.contract = new ethers.Contract(contractAddress, this.contractABI, this.wallet);
  }

  computeContentHash(filePath) {
    const fileBuffer = fs.readFileSync(filePath);
    const hashSum = crypto.createHash('sha256');
    hashSum.update(fileBuffer);
    return '0x' + hashSum.digest('hex');
  }

  async registerOnChain(contentHash, creatorDID, metadata) {
    const tx = await this.contract.registerClaim(
      contentHash,
      creatorDID,
      metadata.softwareVersion || 'unknown',
      metadata.editHistory || '',
      metadata.isAIGenerated || false
    );
    
    const receipt = await tx.wait();
    const event = receipt.logs.find(log => {
      try {
        return this.contract.interface.parseLog(log).name === 'ClaimRegistered';
      } catch { return false; }
    });

    if (event) {
      const parsed = this.contract.interface.parseLog(event);
      return {
        claimId: parsed.args.claimId,
        transactionHash: receipt.hash,
        blockNumber: receipt.blockNumber
      };
    }
    throw new Error('Failed to register claim');
  }

  async verifyContent(filePath, claimId) {
    const contentHash = this.computeContentHash(filePath);
    const [isValid, claim] = await this.contract.verifyClaim(claimId);

    if (!isValid) {
      return { verified: false, reason: 'Claim not found or revoked' };
    }

    const hashMatch = claim.contentHash === contentHash;
    return {
      verified: hashMatch,
      contentHash,
      storedHash: claim.contentHash,
      creatorDID: claim.creatorDID,
      timestamp: new Date(Number(claim.timestamp) * 1000).toISOString(),
      isAIGenerated: claim.isAIGenerated,
      hashMatch
    };
  }

  async revokeClaim(claimId) {
    const tx = await this.contract.revokeClaim(claimId);
    const receipt = await tx.wait();
    return receipt.hash;
  }
}

// 使用示例
async function main() {
  const service = new ContentProvenanceService(
    'https://mainnet.infura.io/v3/YOUR-PROJECT-ID',
    '0xContractAddress...',
    '0xYourPrivateKey...'
  );

  const contentHash = service.computeContentHash('./video.mp4');
  const result = await service.registerOnChain(
    contentHash,
    'did:key:creator123',
    {
      softwareVersion: 'Adobe Premiere Pro 2026',
      editHistory: 'color_grading->audio_mix->export',
      isAIGenerated: false
    }
  );
  
  console.log('Registered:', result.claimId);
  
  const verification = await service.verifyContent('./video.mp4', result.claimId);
  console.log('Verified:', verification.verified);
}

main().catch(console.error);

# 持续验证的状态机示意
class ContentVerificationStateMachine:
    def __init__(self):
        self.states = {
            "CREATED": 0,      # 内容创建
            "VERIFIED": 1,     # 初步验证通过
            "EDITED": 2,       # 经过编辑
            "REVERIFIED": 3,   # 重新验证
            "COMPROMISED": 4,  # 被篡改
            "REVOKED": 5       # 来源声明被撤销
        }
        self.current_state = "CREATED"
        self.verification_log = []
    
    def transition(self, event, verifier_did):
        transitions = {
            ("CREATED", "sign"): "VERIFIED",
            ("VERIFIED", "edit"): "EDITED",
            ("EDITED", "resign"): "REVERIFIED", 
            ("VERIFIED", "tamper"): "COMPROMISED",
            ("COMPROMISED", "revoke"): "REVOKED",
            ("REVERIFIED", "edit"): "EDITED",
            ("REVERIFIED", "tamper"): "COMPROMISED"
        }
        
        key = (self.current_state, event)
        if key in transitions:
            self.current_state = transitions[key]
            self.verification_log.append({
                "from": key[0],
                "to": self.current_state,
                "event": event,
                "verifier": verifier_did,
                "timestamp": __import__('time').time()
            })
            return True
        return False
    
    def get_trust_score(self):
        score_map = {
            "VERIFIED": 1.0,
            "REVERIFIED": 0.9,
            "CREATED": 0.5,
            "EDITED": 0.6,
            "COMPROMISED": 0.1,
            "REVOKED": 0.0
        }
        return score_map.get(self.current_state, 0.0)