003_区块链网络安全入门：分布式账本的防护体系与共识机制安全

# 简化的区块链节点通信伪代码
import socket
import threading

class BlockchainNode:
    def __init__(self, node_id, port):
        self.node_id = node_id
        self.port = port
        self.peer_nodes = []
        self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        
    def start_server(self):
        # 绑定端口并监听连接请求
        self.server_socket.bind(('0.0.0.0', self.port))
        self.server_socket.listen(10)
        print(f"节点 {self.node_id} 启动在端口 {self.port}")
        
        # 接受连接的线程
        threading.Thread(target=self.accept_connections).start()
    
    def accept_connections(self):
        while True:
            client_socket, address = self.server_socket.accept()
            # 验证节点身份
            if self.verify_node(address):
                self.peer_nodes.append((client_socket, address))
                threading.Thread(target=self.handle_connection, args=(client_socket, address)).start()
            else:
                client_socket.close()
    
    def verify_node(self, address):
        # 实现节点身份验证逻辑
        # 在实际系统中，可能包括IP白名单、证书验证等
        return True  # 简化示例
    
    def handle_connection(self, client_socket, address):
        # 处理与其他节点的通信
        pass

# 节点安全配置示例（Ubuntu系统）

# 1. 更新系统
apt update && apt upgrade -y

# 2. 配置防火墙
ufw allow 22/tcp  # SSH
ufw allow 8333/tcp  # 比特币P2P端口
ufw default deny incoming
ufw default allow outgoing
ufw enable

# 3. 安装fail2ban
apt install fail2ban -y
systemctl enable fail2ban
systemctl start fail2ban

# 4. 配置SSH安全
cat >> /etc/ssh/sshd_config << 'EOF'
PasswordAuthentication no
PermitRootLogin no
AllowUsers blockchain_user
EOF
systemctl restart sshd

检测 → 分析 → 遏制 → 消除 → 恢复 → 总结

# PoW简化实现示例
def proof_of_work(block_header, difficulty):
    """
    寻找满足特定难度的哈希值
    difficulty: 目标前缀中0的个数
    """
    nonce = 0
    target = '0' * difficulty
    
    while True:
        # 构造候选块
        candidate = f"{block_header}{nonce}"
        # 计算哈希值
        hash_value = hashlib.sha256(candidate.encode()).hexdigest()
        
        # 检查是否满足难度要求
        if hash_value[:difficulty] == target:
            return nonce, hash_value
        
        nonce += 1

攻击成本 = 控制51%算力/质押所需资源 × 攻击持续时间 × 攻击期间的机会成本
防御阈值 = 攻击成本 > 攻击预期收益

// Optimism欺诈证明简化示例
async function verifyStateTransition(oldStateRoot, newStateRoot, transactions) {
  // 模拟状态转换验证
  let currentStateRoot = oldStateRoot;
  
  for (const tx of transactions) {
    // 应用交易到当前状态
    const nextState = applyTransaction(currentStateRoot, tx);
    currentStateRoot = hashState(nextState);
  }
  
  // 验证最终状态根是否匹配
  return currentStateRoot === newStateRoot;
}

// 欺诈证明提交
async function submitFraudProof(oldStateRoot, newStateRoot, transactions, challengeData) {
  // 提供反例证明状态转换错误
  const isValid = await verifyStateTransition(oldStateRoot, newStateRoot, transactions);
  
  if (!isValid) {
    // 提交包含错误证据的挑战
    return await challengeManager.submitChallenge({
      batchIndex,
      claimedStateRoot: newStateRoot,
      correctStateRoot: computeCorrectStateRoot(oldStateRoot, transactions),
      challengeData
    });
  }
  
  return false; // 状态转换有效，无需挑战
}

# Ubuntu服务器安全初始配置

# 1. 创建非root用户并授予sudo权限
adduser blockchain
usermod -aG sudo blockchain

# 2. 禁用root SSH登录
cp /etc/ssh/sshd_config /etc/ssh/sshd_config.backup
sed -i 's/PermitRootLogin yes/PermitRootLogin no/g' /etc/ssh/sshd_config

# 3. 配置UFW防火墙
ufw allow OpenSSH
# 允许区块链P2P端口（比特币示例）
ufw allow 8333/tcp
# 允许RPC端口（仅限本地访问）
ufw allow from 127.0.0.1 to any port 8332 proto tcp
ufw enable

# 4. 安装必要的安全工具
apt install fail2ban ufw rkhunter clamav clamav-daemon -y

# 5. 更新系统
apt update && apt upgrade -y

# 比特币核心配置文件示例 (bitcoin.conf)

# 网络设置
server=1
listen=1
maxconnections=125

# RPC设置
rpcuser=secure_username
rpcpassword=very_secure_random_password_here
rpcallowip=127.0.0.1
rpcbind=127.0.0.1
rpcport=8332

# 安全设置
walletbroadcast=1
txindex=1
blockfilterindex=1
peerbloomfilters=1

# 性能优化
maxmempool=1000
maxorphantx=100
maxreorgdepth=100

# 日志设置
debug=net

# 简化的节点连接管理伪代码
class ConnectionManager:
    def __init__(self, max_connections=125, max_per_subnet=10):
        self.max_connections = max_connections
        self.max_per_subnet = max_per_subnet
        self.connections = {}
        self.subnet_counts = {}
    
    def should_accept_connection(self, ip_address):
        # 1. 检查连接总数
        if len(self.connections) >= self.max_connections:
            return False
        
        # 2. 检查子网连接数
        subnet = self._get_subnet(ip_address)
        if subnet in self.subnet_counts and self.subnet_counts[subnet] >= self.max_per_subnet:
            return False
        
        # 3. 其他验证...
        return True
    
    def add_connection(self, ip_address, connection):
        if not self.should_accept_connection(ip_address):
            return False
        
        subnet = self._get_subnet(ip_address)
        self.connections[ip_address] = connection
        
        # 更新子网计数
        if subnet in self.subnet_counts:
            self.subnet_counts[subnet] += 1
        else:
            self.subnet_counts[subnet] = 1
        
        return True
    
    def _get_subnet(self, ip_address):
        # 获取IP地址的子网（如IPv4的前三个八位字节）
        parts = ip_address.split('.')
        return f"{parts[0]}.{parts[1]}.{parts[2]}.0"

# Prometheus监控配置示例
scrape_configs:
  - job_name: 'bitcoin'
    scrape_interval: 15s
    static_configs:
      - targets: ['localhost:8334']  # Bitcoin Core Prometheus导出器
  
  - job_name: 'node_exporter'
    scrape_interval: 15s
    static_configs:
      - targets: ['localhost:9100']  # 系统监控导出器
  
  - job_name: 'blockchain_exporter'
    scrape_interval: 60s
    static_configs:
      - targets: ['localhost:9332']  # 自定义区块链指标导出器

#!/bin/bash
# 区块链节点维护脚本

# 日志轮转
find /var/log/bitcoin -name "bitcoin*.log" -type f -mtime +7 -exec gzip {} \;

# 检查磁盘空间
DISK_USAGE=$(df -h /var/lib/bitcoin | tail -1 | awk '{print $5}' | sed 's/%//')
if [ $DISK_USAGE -gt 80 ]; then
  echo "警告: 磁盘使用率超过80%" | mail -s "节点警告" admin@example.com
fi

# 检查节点同步状态
SYNC_STATUS=$(bitcoin-cli getblockchaininfo | jq '.initialblockdownload')
if [ "$SYNC_STATUS" = "true" ]; then
  echo "节点仍在初始同步中" | mail -s "节点状态" admin@example.com
fi

# 备份钱包（如果有）
if [ -f "~/.bitcoin/wallet.dat" ]; then
  cp ~/.bitcoin/wallet.dat ~/.bitcoin/backups/wallet_$(date +%Y%m%d).dat
  gpg --encrypt --recipient admin@example.com ~/.bitcoin/backups/wallet_$(date +%Y%m%d).dat
fi

# 简化的交易结构示例
class Transaction:
    def __init__(self):
        self.inputs = []
        self.outputs = []
        self.fee = 0
        self.timestamp = time.time()
        self.signature = None
    
    def add_input(self, txid, vout, amount):
        # 添加交易输入，引用之前交易的输出
        self.inputs.append({'txid': txid, 'vout': vout, 'amount': amount})
    
    def add_output(self, address, amount):
        # 添加交易输出，指定接收方和金额
        self.outputs.append({'address': address, 'amount': amount})
    
    def calculate_fee(self, fee_rate):
        # 根据交易大小和费率计算手续费
        tx_size = self._get_size()
        self.fee = tx_size * fee_rate
    
    def sign(self, private_key):
        # 对交易进行签名
        tx_hash = self._calculate_hash()
        self.signature = sign_message(tx_hash, private_key)
    
    def _calculate_hash(self):
        # 计算交易哈希，用于签名和验证
        return sha256(json.dumps(self._get_signing_data())).hexdigest()

# 简化的椭圆曲线数字签名实现
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import hashes

# 生成密钥对
def generate_key_pair():
    private_key = ec.generate_private_key(ec.SECP256K1())
    public_key = private_key.public_key()
    return private_key, public_key

# 签名消息
def sign_message(message, private_key):
    signature = private_key.sign(
        message.encode(),
        ec.ECDSA(hashes.SHA256())
    )
    return signature

# 验证签名
def verify_signature(message, signature, public_key):
    try:
        public_key.verify(
            signature,
            message.encode(),
            ec.ECDSA(hashes.SHA256())
        )
        return True
    except Exception:
        return False

1. 攻击者向商家支付商品费用
2. 商家确认交易（通常在几个确认后）并发货
3. 攻击者控制足够算力创建更长的分叉链，不包含支付交易
4. 当分叉链成为主链时，原始支付交易被撤销
5. 攻击者取回了付款，同时获得了商品

# 简化的交易验证流程
class TransactionValidator:
    def __init__(self, blockchain, utxo_set):
        self.blockchain = blockchain
        self.utxo_set = utxo_set
    
    def validate(self, transaction):
        # 1. 格式验证
        if not self._validate_format(transaction):
            return False, "格式无效"
        
        # 2. 签名验证
        if not self._validate_signature(transaction):
            return False, "签名无效"
        
        # 3. 余额验证
        total_input = sum(tx_in['amount'] for tx_in in transaction.inputs)
        total_output = sum(tx_out['amount'] for tx_out in transaction.outputs) + transaction.fee
        
        if total_input < total_output:
            return False, "输入金额不足"
        
        # 4. UTXO验证
        for tx_in in transaction.inputs:
            if not self._is_utxo_unspent(tx_in['txid'], tx_in['vout']):
                return False, "UTXO已被花费"
        
        # 5. 其他规则验证
        if not self._validate_rules(transaction):
            return False, "违反共识规则"
        
        return True, "交易有效"
    
    # 验证辅助方法...

// 使用Web3.js安全创建交易的示例
async function createSecureTransaction(fromAddress, toAddress, amount, web3, privateKey) {
    try {
        // 获取当前gas价格和建议
        const gasPrice = await web3.eth.getGasPrice();
        const recommendedGas = Math.floor(parseInt(gasPrice) * 1.1); // 增加10%以确保及时确认
        
        // 获取nonce，确保交易顺序
        const nonce = await web3.eth.getTransactionCount(fromAddress, 'pending');
        
        // 估算gas限制
        const gasLimit = await web3.eth.estimateGas({
            from: fromAddress,
            to: toAddress,
            value: web3.utils.toWei(amount, 'ether')
        });
        
        // 构建交易对象
        const txObject = {
            from: fromAddress,
            to: toAddress,
            value: web3.utils.toWei(amount, 'ether'),
            gasPrice: recommendedGas,
            gasLimit: gasLimit,
            nonce: nonce,
            chainId: await web3.eth.getChainId() // 添加chainId防止重放攻击
        };
        
        // 签名交易（生产环境建议使用硬件钱包）
        const signedTx = await web3.eth.accounts.signTransaction(txObject, privateKey);
        
        // 发送交易
        const txReceipt = await web3.eth.sendSignedTransaction(signedTx.rawTransaction);
        
        console.log(`交易成功! 交易哈希: ${txReceipt.transactionHash}`);
        return txReceipt;
    } catch (error) {
        console.error(`交易失败: ${error.message}`);
        throw error;
    }
}

# 简化的交互式零知识证明示例
import hashlib
import random

def hash_function(x):
    """简化的哈希函数"""
    return hashlib.sha256(str(x).encode()).hexdigest()

def setup_zkp(x, y, rounds=20):
    """
    设置零知识证明
    x: 原像（证明者知道的秘密）
    y: 哈希值（H(x)=y）
    """
    # 验证H(x)=y是否成立
    if hash_function(x) != y:
        raise ValueError("x不是y的原像")
    
    return {
        "x": x,  # 仅证明者知道
        "y": y,  # 公开信息
        "rounds": rounds,
        "challenges": [],
        "responses": []
    }

def prove_step(prover_state):
    """
    证明者执行的每轮操作
    """
    # 1. 证明者选择随机数r
    r = random.randint(1, 10**18)
    
    # 2. 计算承诺c = H(r)
    c = hash_function(r)
    
    # 3. 验证者生成随机挑战
    challenge = random.randint(0, 1)  # 0: 揭示r, 1: 揭示x和r的组合
    
    # 4. 证明者根据挑战响应
    if challenge == 0:
        # 揭示r
        response = r
    else:
        # 揭示x XOR r（这样验证者无法单独获得x）
        response = r ^ prover_state["x"]
    
    # 记录这轮交互
    prover_state["challenges"].append(challenge)
    prover_state["responses"].append((c, response))
    
    return c

def verify_step(prover_state, c, response, challenge):
    """
    验证者验证证明者的响应
    """
    if challenge == 0:
        # 验证H(response) == c
        return hash_function(response) == c
    else:
        # 验证H(response XOR x_candidate) == y
        # 注意：在真实的零知识证明中，验证者不知道x，这里只是演示
        # 实际实现中会有所不同
        x_candidate = response ^ (12345)  # 假设这是某个候选值
        return hash_function(x_candidate) == prover_state["y"]

# 演示零知识证明
secret_x = 12345
y = hash_function(secret_x)

# 设置证明
prover_state = setup_zkp(secret_x, y)

# 执行多轮交互
for i in range(prover_state["rounds"]):
    c = prove_step(prover_state)
    # 验证步骤在实际中由验证者执行
    # 验证者不知道secret_x，但可以验证证明

// ZK-SNARKs工作流程

阶段1: 可信设置
1. 选择随机数s（有毒废物）
2. 生成公共参数(PP) = (PK, VK)
3. 销毁s，防止任何人重建证明系统

阶段2: 证明生成
1. 将计算函数f转换为QAP形式
2. 证明者使用秘密输入w和公共参数PK
3. 生成证明π = Prove(PK, f, x, w)，其中x是公共输入

阶段3: 证明验证
1. 验证者使用公共参数VK、公共输入x和证明π
2. 验证Verify(VK, f, x, π)是否返回true
3. 如果为真，则接受证明者知道满足f(x,w)=0的w

Zcash屏蔽交易流程:

1. 发送方使用密钥生成屏蔽地址
2. 发送方创建交易，包含输入金额、输出金额和接收方地址
3. 使用zk-SNARKs证明:
   - 发送方知道输入资金的私钥
   - 输入金额总和等于输出金额总和
   - 没有双重花费发生
4. 生成承诺和空值(nullifiers)以确保交易有效性
5. 交易被添加到区块链，但金额和参与者保持隐藏

// zkSync交易处理简化示例
class ZKRollup {
    constructor() {
        this.stateRoot = initialStateRoot; // 初始状态根
        this.pendingTransactions = [];
    }
    
    // 添加交易到待处理队列
    addTransaction(transaction) {
        // 基本验证
        if (this._validateBasic(transaction)) {
            this.pendingTransactions.push(transaction);
            return true;
        }
        return false;
    }
    
    // 批量处理交易并生成证明
    batchProcess() {
        if (this.pendingTransactions.length === 0) {
            return null;
        }
        
        // 处理交易，更新状态
        const newState = this._processTransactions(this.pendingTransactions);
        
        // 计算新的状态根
        const newStateRoot = this._calculateStateRoot(newState);
        
        // 生成零知识证明
        const proof = this._generateZKProof(
            this.stateRoot, // 旧状态根
            newStateRoot,   // 新状态根
            this.pendingTransactions
        );
        
        // 更新状态根
        this.stateRoot = newStateRoot;
        
        // 清空待处理交易
        this.pendingTransactions = [];
        
        return {
            newStateRoot,
            proof,
            batchSize: this.pendingTransactions.length
        };
    }
    
    // 验证零知识证明
    verifyProof(oldStateRoot, newStateRoot, proof) {
        return this._verifyZKProof(oldStateRoot, newStateRoot, proof);
    }
    
    // 内部方法...
}

# Circom和SnarkJS设置示例

# 1. 安装Node.js和npm

# 2. 全局安装Circom和SnarkJS
npm install -g circom snarkjs

# 3. 创建项目目录
mkdir zkp-project
cd zkp-project

# 4. 创建一个简单的电路文件 (example.circom)
cat > example.circom << EOL
pragma circom 2.1.0;

include "circomlib/poseidon.circom";

// 一个简单的零知识证明电路，证明知道x使得x^3 + x + 5 = y
template Example() {
    signal private input x;
    signal output y;
    
    y <== x*x*x + x + 5;
}

component main = Example();
EOL

# 5. 下载circomlib
git clone https://github.com/iden3/circomlib

# 6. 编译电路
circom example.circom --wasm --r1cs

# 7. 执行可信设置（仅演示）
# 注意：生产环境应使用多方计算的可信设置

// 简化的zk-SNARK验证合约示例
// SPDX-License-Identifier: MIT
pragma solidity ^0.8.0;

import "@openzeppelin/contracts/access/Ownable.sol";

contract ZKVerifier is Ownable {
    // 验证密钥
    struct VerifyingKey {
        uint256 alfa1;
        uint256 beta2_1;
        uint256 beta2_2;
        uint256 gamma2_1;
        uint256 gamma2_2;
        uint256 delta2_1;
        uint256 delta2_2;
        uint256[2] ic;
    }
    
    VerifyingKey public vk;
    
    // 存储已验证的证明，防止重放攻击
    mapping(bytes32 => bool) public usedProofs;
    
    // 事件
    event ProofVerified(address indexed prover, uint256 publicInput);
    
    // 设置验证密钥
    function setVerifyingKey(
        uint256 _alfa1,
        uint256[2] memory _beta2,
        uint256[2] memory _gamma2,
        uint256[2] memory _delta2,
        uint256[2] memory _ic
    ) external onlyOwner {
        vk = VerifyingKey({
            alfa1: _alfa1,
            beta2_1: _beta2[0],
            beta2_2: _beta2[1],
            gamma2_1: _gamma2[0],
            gamma2_2: _gamma2[1],
            delta2_1: _delta2[0],
            delta2_2: _delta2[1],
            ic: _ic
        });
    }
    
    // 验证证明
    function verifyProof(
        uint256[2] memory a,
        uint256[2][2] memory b,
        uint256[2] memory c,
        uint256[1] memory input
    ) external returns (bool) {
        // 计算证明哈希，防止重放攻击
        bytes32 proofHash = keccak256(abi.encode(a, b, c, input));
        require(!usedProofs[proofHash], "Proof already used");
        
        // 实际验证逻辑（简化）
        // 在真实实现中，这里应该调用配对函数进行验证
        bool verified = _verifyProof(a, b, c, input);
        
        if (verified) {
            usedProofs[proofHash] = true;
            emit ProofVerified(msg.sender, input[0]);
        }
        
        return verified;
    }
    
    // 内部验证函数（实际实现需使用配对库）
    function _verifyProof(
        uint256[2] memory a,
        uint256[2][2] memory b,
        uint256[2] memory c,
        uint256[1] memory input
    ) internal view returns (bool) {
        // 这里应该实现完整的配对验证
        // 由于Solidity的限制，实际实现需要使用配对库
        // 例如：using Pairing for *;
        // 并实现完整的验证算法
        
        // 简化示例，实际中应替换为正确的验证代码
        return true;
    }
}

区块链安全事件分析流程图:

1. 事件发现与确认
   ↓
2. 影响范围评估
   ↓
3. 技术根因分析
   ↓
4. 运营与流程审查
   ↓
5. 威胁行为者画像
   ↓
6. 预防措施制定

# 应急响应工具包示例

#!/bin/bash

# 区块链安全事件应急响应脚本

echo "===== 区块链安全事件应急响应工具 ====="

# 1. 日志收集与分析函数
collect_logs() {
  echo "正在收集节点日志..."
  mkdir -p incident_$(date +%Y%m%d)
  cp -r /var/log/blockchain/* incident_$(date +%Y%m%d)/
  echo "日志已保存至 incident_$(date +%Y%m%d)/ 目录"
}

# 2. 交易监控函数
monitor_transactions() {
  echo "正在监控异常交易..."
  # 这里添加异常交易检测逻辑
  # blockchain-cli monitor --anomaly --output alert.txt
  echo "异常交易监控完成，结果已保存至 alert.txt"
}

# 3. 系统隔离函数
isolate_system() {
  echo "正在隔离受影响系统..."
  # 这里添加系统隔离逻辑
  # iptables -A INPUT -j DROP
  echo "系统隔离措施已实施"
}

# 4. 私钥安全检查
check_key_security() {
  echo "正在检查私钥安全状态..."
  # 这里添加私钥安全检查逻辑
  echo "私钥安全检查完成"
}

# 5. 漏洞扫描
scan_vulnerabilities() {
  echo "正在扫描系统漏洞..."
  # 这里添加漏洞扫描逻辑
  echo "漏洞扫描完成"
}

# 显示菜单
echo "请选择操作:"
echo "1. 收集日志"
echo "2. 监控异常交易"
echo "3. 隔离系统"
echo "4. 检查私钥安全"
echo "5. 扫描漏洞"
echo "6. 执行全部操作"

read -p "输入选项 [1-6]: " choice

# 根据选择执行相应操作
case $choice in
  1) collect_logs ;;
  2) monitor_transactions ;;
  3) isolate_system ;;
  4) check_key_security ;;
  5) scan_vulnerabilities ;;
  6) 
     collect_logs
     monitor_transactions
     check_key_security
     scan_vulnerabilities
     echo "是否隔离系统？(y/n)"
     read isolate
     if [ "$isolate" = "y" ]; then
       isolate_system
     fi
     ;;
  *) echo "无效选项" ;;
esac

echo "应急响应操作完成！"

区块链安全监控系统架构:

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│   数据采集层    │────>│   处理分析层    │────>│   展示告警层    │
└─────────────────┘     └─────────────────┘     └─────────────────┘
       │                        │                       │
       ▼                        ▼                       ▼
┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│ 区块链节点日志  │     │   实时处理引擎  │     │  监控仪表盘     │
│ 交易数据监控    │     │   异常检测算法  │     │  告警系统       │
│ 智能合约事件    │     │   威胁情报集成  │     │  自动响应       │
│ 网络流量分析    │     │   机器学习模型  │     │  报告生成       │
│ 基础设施监控    │     │   历史数据分析  │     │  可视化工具     │
└─────────────────┘     └─────────────────┘     └─────────────────┘

# 区块链安全监控系统示例 - 交易异常检测
import time
import numpy as np
import pandas as pd
from blockchain import Blockchain
import matplotlib.pyplot as plt
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler

class BlockchainMonitor:
    def __init__(self):
        self.blockchain = Blockchain()
        self.transaction_history = []
        self.model = IsolationForest(contamination=0.01, random_state=42)
        self.scaler = StandardScaler()
        self.is_model_trained = False
    
    def collect_transactions(self, block_count=10):
        """收集最近几个区块的交易数据"""
        for i in range(block_count):
            block_number = self.blockchain.get_latest_block_number() - i
            block = self.blockchain.get_block(block_number)
            transactions = block['transactions']
            
            for tx in transactions:
                tx_data = {
                    'hash': tx['hash'],
                    'from': tx['from'],
                    'to': tx['to'],
                    'value': float(tx['value']),
                    'gas': int(tx['gas']),
                    'gas_price': int(tx['gasPrice']),
                    'timestamp': block['timestamp'],
                    'block_number': block_number
                }
                self.transaction_history.append(tx_data)
        
        print(f"已收集 {len(self.transaction_history)} 笔交易数据")
    
    def extract_features(self):
        """从交易数据中提取特征"""
        if not self.transaction_history:
            return None
        
        df = pd.DataFrame(self.transaction_history)
        
        # 提取特征
        features = df[['value', 'gas', 'gas_price']].copy()
        
        # 计算额外特征
        # 按地址统计交易频率
        address_freq = pd.concat([df['from'], df['to']]).value_counts()
        df['from_freq'] = df['from'].map(address_freq)
        df['to_freq'] = df['to'].map(address_freq)
        
        features['from_freq'] = df['from_freq']
        features['to_freq'] = df['to_freq']
        
        # 填充缺失值
        features = features.fillna(0)
        
        return features, df
    
    def train_model(self):
        """训练异常检测模型"""
        features, _ = self.extract_features()
        if features is None or len(features) < 100:
            print("数据不足，无法训练模型")
            return False
        
        # 数据标准化
        scaled_features = self.scaler.fit_transform(features)
        
        # 训练模型
        self.model.fit(scaled_features)
        self.is_model_trained = True
        print("异常检测模型训练完成")
        return True
    
    def detect_anomalies(self):
        """检测异常交易"""
        if not self.is_model_trained:
            print("模型未训练，无法进行异常检测")
            return None
        
        features, df = self.extract_features()
        if features is None:
            return None
        
        # 数据标准化
        scaled_features = self.scaler.transform(features)
        
        # 预测异常
        predictions = self.model.predict(scaled_features)
        
        # 标记异常交易
        df['is_anomaly'] = predictions == -1
        anomalies = df[df['is_anomaly']].copy()
        
        print(f"检测到 {len(anomalies)} 笔异常交易")
        return anomalies
    
    def monitor_live(self, interval=60):
        """实时监控交易"""
        print(f"开始实时监控，每 {interval} 秒检查一次")
        
        while True:
            # 收集最新区块交易
            self.collect_transactions(block_count=1)
            
            # 如果模型未训练且数据足够，则训练模型
            if not self.is_model_trained and len(self.transaction_history) >= 100:
                self.train_model()
            
            # 如果模型已训练，执行异常检测
            if self.is_model_trained:
                anomalies = self.detect_anomalies()
                if anomalies is not None and len(anomalies) > 0:
                    print("===== 检测到异常交易 =====")
                    for _, anomaly in anomalies.tail(5).iterrows():
                        print(f"交易哈希: {anomaly['hash']}")
                        print(f"发送方: {anomaly['from']}")
                        print(f"接收方: {anomaly['to']}")
                        print(f"金额: {anomaly['value']}")
                        print(f"区块号: {anomaly['block_number']}")
                        print("------------------------")
                    self.alert_anomalies(anomalies)
            
            # 休眠指定时间
            time.sleep(interval)
    
    def alert_anomalies(self, anomalies):
        """发送异常告警"""
        # 这里可以实现告警逻辑，如发送邮件、短信等
        print(f"发送异常告警: {len(anomalies)} 笔异常交易")
        # 告警代码示例：
        # send_email_alert("区块链安全告警", f"检测到 {len(anomalies)} 笔异常交易")
        # send_sms_alert("区块链安全告警，检测到异常交易")

# 使用示例
if __name__ == "__main__":
    monitor = BlockchainMonitor()
    # 初始化：收集历史数据并训练模型
    monitor.collect_transactions(block_count=50)
    monitor.train_model()
    
    # 开始实时监控
    monitor.monitor_live(interval=60)

// 有漏洞的合约代码示例
function withdraw(uint amount) public {
    require(balances[msg.sender] >= amount);
    // 漏洞：在转账前未更新余额
    (bool success, ) = msg.sender.call{value: amount}("");
    require(success, "Transfer failed");
    // 正确的做法应该是先更新状态
    balances[msg.sender] -= amount;
}

// 攻击者使用的合约
contract AttackContract {
    VulnerableBank public bank;
    
    constructor(address _bankAddress) {
        bank = VulnerableBank(_bankAddress);
    }
    
    function attack() public payable {
        // 先存入一些资金
        bank.deposit{value: msg.value}();
        // 触发提款
        bank.withdraw(msg.value);
    }
    
    // 回退函数，在收到ETH时自动调用
    receive() external payable {
        // 如果银行还有余额，继续提款
        if (address(bank).balance >= msg.value) {
            bank.withdraw(msg.value);
        }
    }
}

# 区块链节点恢复脚本示例
import os
import subprocess
import time
import json

def load_config():
    """加载节点配置"""
    with open('node_config.json', 'r') as f:
        return json.load(f)

def stop_node():
    """停止区块链节点"""
    print("正在停止节点...")
    try:
        subprocess.run(['systemctl', 'stop', 'blockchain-node'], check=True)
        print("节点已停止")
        return True
    except Exception as e:
        print(f"停止节点失败: {e}")
        return False

def backup_data(data_dir):
    """备份现有数据"""
    backup_dir = f"{data_dir}_backup_{int(time.time())}"
    print(f"正在备份数据到 {backup_dir}")
    try:
        subprocess.run(['cp', '-r', data_dir, backup_dir], check=True)
        print("数据备份完成")
        return backup_dir
    except Exception as e:
        print(f"数据备份失败: {e}")
        return None

def restore_from_snapshot(data_dir, snapshot_url):
    """从快照恢复数据"""
    print(f"正在从快照 {snapshot_url} 恢复数据")
    try:
        # 清除现有数据
        subprocess.run(['rm', '-rf', f"{data_dir}/*"], check=True)
        
        # 下载快照
        snapshot_file = f"{data_dir}/snapshot.tar.gz"
        subprocess.run(['wget', snapshot_url, '-O', snapshot_file], check=True)
        
        # 解压快照
        subprocess.run(['tar', '-xzf', snapshot_file, '-C', data_dir], check=True)
        
        # 清理
        os.remove(snapshot_file)
        
        print("快照恢复完成")
        return True
    except Exception as e:
        print(f"快照恢复失败: {e}")
        return False

def start_node():
    """启动区块链节点"""
    print("正在启动节点...")
    try:
        subprocess.run(['systemctl', 'start', 'blockchain-node'], check=True)
        print("节点已启动")
        return True
    except Exception as e:
        print(f"启动节点失败: {e}")
        return False

def verify_node_status():
    """验证节点状态"""
    print("正在验证节点状态...")
    try:
        # 等待节点启动
        time.sleep(10)
        
        # 检查节点同步状态
        result = subprocess.run(['blockchain-cli', 'status'], capture_output=True, text=True, check=True)
        status = json.loads(result.stdout)
        
        # 输出关键状态信息
        print(f"节点同步状态: {status.get('sync_status')}")
        print(f"当前区块高度: {status.get('block_height')}")
        print(f"连接节点数: {status.get('peer_count')}")
        
        # 验证数据一致性
        if status.get('sync_status') == 'synced':
            print("节点恢复成功，数据验证通过")
            return True
        else:
            print("节点尚未完全同步")
            return False
    except Exception as e:
        print(f"验证节点状态失败: {e}")
        return False

def main():
    """主函数"""
    print("===== 区块链节点恢复工具 =====")
    
    # 加载配置
    config = load_config()
    data_dir = config.get('data_dir')
    snapshot_url = config.get('snapshot_url')
    
    if not data_dir or not snapshot_url:
        print("配置错误：缺少必要的配置项")
        return
    
    # 执行恢复流程
    print("开始执行节点恢复流程...")
    
    # 1. 停止节点
    if not stop_node():
        print("无法继续恢复流程")
        return
    
    # 2. 备份现有数据
    backup_dir = backup_data(data_dir)
    if not backup_dir:
        print("警告：数据备份失败，但将继续恢复流程")
    
    # 3. 从快照恢复
    if not restore_from_snapshot(data_dir, snapshot_url):
        print("恢复失败，尝试从备份恢复")
        # 从备份恢复
        if backup_dir:
            try:
                subprocess.run(['rm', '-rf', f"{data_dir}/*"], check=True)
                subprocess.run(['cp', '-r', f"{backup_dir}/*", data_dir], check=True)
                print("已从备份恢复数据")
            except Exception as e:
                print(f"从备份恢复失败: {e}")
                return
        else:
            return
    
    # 4. 启动节点
    if not start_node():
        print("恢复流程失败")
        return
    
    # 5. 验证节点状态
    verify_node_status()
    
    print("节点恢复流程完成")

if __name__ == "__main__":
    main()

区块链安全防御多层模型：

┌─────────────────────┐
│   物理/基础设施层   │
│  - 硬件安全模块     │
│  - 安全服务器环境   │
│  - 灾备系统         │
├─────────────────────┤
│     网络通信层      │
│  - 加密通信协议     │
│  - DDoS防护         │
│  - 入侵检测系统     │
├─────────────────────┤
│     区块链核心层    │
│  - 共识安全机制     │
│  - 节点同步验证     │
│  - 区块验证逻辑     │
├─────────────────────┤
│     智能合约层      │
│  - 代码审计         │
│  - 形式化验证       │
│  - 安全开发框架     │
├─────────────────────┤
│     应用接口层      │
│  - API安全          │
│  - 权限控制         │
│  - 输入验证         │
├─────────────────────┤
│     用户访问层      │
│  - 多因素认证       │
│  - 私钥管理解决方案 │
│  - 安全意识培训     │
└─────────────────────┘