区块链 & Web3 工程师指南 — 从智能合约、DeFi、ZK 证明到 AI+区块链

import hashlib
import json

def compute_block_hash(index, previous_hash, timestamp, data, nonce):
    block_content = json.dumps({
        "index": index,
        "previous_hash": previous_hash,
        "timestamp": timestamp,
        "data": data,
        "nonce": nonce
    }, sort_keys=True)
    return hashlib.sha256(block_content.encode()).hexdigest()

# SHA-256 的特性：确定性、单向性、雪崩效应
hash1 = hashlib.sha256(b"hello").hexdigest()
hash2 = hashlib.sha256(b"hellO").hexdigest()
# 仅一个字符之差就会产生完全不同的哈希
print(hash1[:16], "vs", hash2[:16])

def build_merkle_tree(leaves):
    if not leaves:
        return None
    layer = [hashlib.sha256(leaf.encode()).hexdigest() for leaf in leaves]
    while len(layer) > 1:
        if len(layer) % 2 != 0:
            layer.append(layer[-1])  # 若为奇数，复制最后一个
        layer = [
            hashlib.sha256((layer[i] + layer[i+1]).encode()).hexdigest()
            for i in range(0, len(layer), 2)
        ]
    return layer[0]  # Merkle Root

txs = ["tx_alice_to_bob", "tx_carol_to_dave", "tx_eve_to_frank", "tx_grace_to_heidi"]
root = build_merkle_tree(txs)
print("Merkle Root:", root)

EVM 栈运算示例（PUSH1、ADD、MUL）：
PUSH1 0x03   → 栈: [3]
PUSH1 0x04   → 栈: [3, 4]
ADD          → 栈: [7]
PUSH1 0x02   → 栈: [7, 2]
MUL          → 栈: [14]

# EIP-1559 交易费用计算
base_fee = 15  # gwei（根据网络状态自动调整）
priority_fee = 2  # gwei（小费）
max_fee = 20  # gwei（允许的最大值）

actual_fee = min(base_fee + priority_fee, max_fee)
gas_used = 21000  # 简单的 ETH 转账
total_cost_gwei = actual_fee * gas_used
total_cost_eth = total_cost_gwei / 1e9
print(f"交易手续费: {total_cost_eth:.6f} ETH")
# base_fee 被销毁 → ETH 通缩效应
burned = base_fee * gas_used / 1e9
print(f"已销毁 ETH: {burned:.6f} ETH")

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

import "@openzeppelin/contracts/token/ERC20/ERC20.sol";
import "@openzeppelin/contracts/security/ReentrancyGuard.sol";
import "@openzeppelin/contracts/access/Ownable.sol";

contract SecureToken is ERC20, ReentrancyGuard, Ownable {
    uint256 public constant MAX_SUPPLY = 1_000_000 * 10**18;
    mapping(address => uint256) public lockTime;

    event TokensLocked(address indexed user, uint256 amount, uint256 unlockTime);
    event TokensUnlocked(address indexed user, uint256 amount);

    constructor() ERC20("SecureToken", "STKN") Ownable(msg.sender) {
        _mint(msg.sender, 100_000 * 10**18);
    }

    // 应用 Checks-Effects-Interactions 模式
    function lockAndRelease(uint256 amount, uint256 duration)
        external
        nonReentrant  // 防重入
    {
        // 1. Checks: 校验条件
        require(amount > 0, "Amount must be positive");
        require(balanceOf(msg.sender) >= amount, "Insufficient balance");
        require(duration >= 1 days && duration <= 365 days, "Invalid duration");

        // 2. Effects: 状态变更（在外部调用之前）
        lockTime[msg.sender] = block.timestamp + duration;
        _transfer(msg.sender, address(this), amount);

        // 3. Interactions: 外部调用（在状态变更之后）
        emit TokensLocked(msg.sender, amount, lockTime[msg.sender]);
    }

    function unlock() external nonReentrant {
        // Checks
        require(block.timestamp >= lockTime[msg.sender], "Still locked");
        uint256 contractBalance = balanceOf(address(this));
        require(contractBalance > 0, "Nothing to unlock");

        // Effects
        lockTime[msg.sender] = 0;

        // Interactions
        _transfer(address(this), msg.sender, contractBalance);
        emit TokensUnlocked(msg.sender, contractBalance);
    }

    function mint(address to, uint256 amount) external onlyOwner {
        require(totalSupply() + amount <= MAX_SUPPLY, "Exceeds max supply");
        _mint(to, amount);
    }
}

// 存在漏洞的合约（切勿使用）
contract VulnerableBank {
    mapping(address => uint256) public balances;

    function withdraw() external {
        uint256 amount = balances[msg.sender];
        // 状态变更前先发起外部调用 → 可能被重入！
        (bool success,) = msg.sender.call{value: amount}("");
        require(success, "Transfer failed");
        balances[msg.sender] = 0;  // 太晚了
    }
}

// 攻击合约
contract Attacker {
    VulnerableBank public target;

    receive() external payable {
        if (address(target).balance >= 1 ether) {
            target.withdraw();  // 递归调用反复提款
        }
    }
}

const { expect } = require('chai')
const { ethers } = require('hardhat')

describe('SecureToken', function () {
  let token, owner, alice, bob

  beforeEach(async function () {
    ;[owner, alice, bob] = await ethers.getSigners()
    const SecureToken = await ethers.getContractFactory('SecureToken')
    token = await SecureToken.deploy()
    await token.waitForDeployment()
  })

  it('재진입 공격을 방어해야 한다', async function () {
    // 向 alice 转账代币
    await token.transfer(alice.address, ethers.parseEther('1000'))
    expect(await token.balanceOf(alice.address)).to.equal(ethers.parseEther('1000'))
  })

  it('잠금 기간 전에는 언락 불가', async function () {
    await token.transfer(alice.address, ethers.parseEther('100'))
    await token.connect(alice).lockAndRelease(
      ethers.parseEther('100'),
      86400 // 1 天
    )
    await expect(token.connect(alice).unlock()).to.be.revertedWith('Still locked')
  })

  it('Foundry 스타일 퍼즈 테스트 시뮬레이션', async function () {
    const amounts = [1, 100, 1000].map((n) => ethers.parseEther(n.toString()))
    for (const amount of amounts) {
      await token.mint(alice.address, amount)
    }
    const total = amounts.reduce((a, b) => a + b, 0n)
    expect(await token.balanceOf(alice.address)).to.equal(total)
  })
})

// test/SecureToken.t.sol
// SPDX-License-Identifier: MIT
pragma solidity ^0.8.20;

import "forge-std/Test.sol";
import "../src/SecureToken.sol";

contract SecureTokenTest is Test {
    SecureToken public token;
    address public alice = makeAddr("alice");

    function setUp() public {
        token = new SecureToken();
        token.transfer(alice, 1000 ether);
    }

    function testLockRequiresDuration() public {
        vm.startPrank(alice);
        vm.expectRevert("Invalid duration");
        token.lockAndRelease(100 ether, 12 hours);  // 不足 1 天
        vm.stopPrank();
    }

    // 模糊测试：用任意金额反复验证
    function testFuzz_LockAmount(uint256 amount) public {
        amount = bound(amount, 1 ether, 1000 ether);
        vm.startPrank(alice);
        token.lockAndRelease(amount, 1 days);
        assertEq(token.balanceOf(address(token)), amount);
        vm.stopPrank();
    }
}

import math

# Uniswap v3 价格计算（基于 tick）
# tick = log(sqrt(price)) / log(sqrt(1.0001))

def tick_to_price(tick):
    return 1.0001 ** tick

def price_to_tick(price):
    return math.floor(math.log(price) / math.log(1.0001))

def calculate_liquidity_v3(amount0, amount1, price_current, price_lower, price_upper):
    """计算集中流动性头寸的 L（流动性）"""
    sqrt_p = math.sqrt(price_current)
    sqrt_pa = math.sqrt(price_lower)
    sqrt_pb = math.sqrt(price_upper)

    if price_current <= price_lower:
        # 全部为 token0
        L = amount0 * (sqrt_pa * sqrt_pb) / (sqrt_pb - sqrt_pa)
    elif price_current >= price_upper:
        # 全部为 token1
        L = amount1 / (sqrt_pb - sqrt_pa)
    else:
        # 混合
        L0 = amount0 * (sqrt_p * sqrt_pb) / (sqrt_pb - sqrt_p)
        L1 = amount1 / (sqrt_p - sqrt_pa)
        L = min(L0, L1)
    return L

# ETH/USDC 池示例
# 当前价格: 3000 USDC/ETH，区间: 2500 ~ 3500
L = calculate_liquidity_v3(
    amount0=1.0,       # 1 ETH
    amount1=3000.0,    # 3000 USDC
    price_current=3000,
    price_lower=2500,
    price_upper=3500
)
print(f"流动性 L: {L:.4f}")

// 使用 snarkjs 的简单 ZK 证明示例
// 电路: a * b = c（在不公开 a、b 的情况下证明乘法关系）

const snarkjs = require('snarkjs')
const fs = require('fs')

async function generateAndVerifyProof() {
  // 1. 计算 witness（非公开输入）
  const input = {
    a: 3, // 秘密
    b: 11, // 秘密
    c: 33, // 公开: a * b = c
  }

  // 2. 生成证明（需要 proving key）
  const { proof, publicSignals } = await snarkjs.groth16.fullProve(
    input,
    'multiply.wasm',
    'multiply_final.zkey'
  )

  console.log('公开信号:', publicSignals) // [33]
  console.log('证明大小:', JSON.stringify(proof).length, 'bytes')

  // 3. 验证（只需要 verification key）
  const vKey = JSON.parse(fs.readFileSync('verification_key.json'))
  const isValid = await snarkjs.groth16.verify(vKey, publicSignals, proof)
  console.log('证明是否有效:', isValid) // true
  // 若没有证明，a=3、b=11 这一事实无法被推知
}

generateAndVerifyProof()

L2 交易批处理:
[tx1, tx2, ..., tx1000]
        ↓ 证明者（prover）
ZK Proof（数百字节）
        ↓ L1 验证者
以太坊主网（节省 99%+ Gas）

# FHE（全同态加密）概念示例
# 在加密状态下执行运算

# 实际实现: Microsoft SEAL、OpenFHE
# 概念示例:
def fhe_inference_concept(encrypted_input, model_weights):
    """
    使用 FHE，可以在不解密输入数据的情况下
    以加密状态执行 AI 推理
    - 医疗数据: 在不暴露患者信息的情况下完成诊断
    - 金融数据: 在保护隐私的同时进行信用评估
    """
    # 同态运算（加法、乘法均可在加密状态下进行）
    encrypted_result = homomorphic_matrix_multiply(
        encrypted_input,
        model_weights  # 模型是公开的，只有输入被加密
    )
    return encrypted_result

from web3 import Web3
import json

# Alchemy/Infura RPC 连接
w3 = Web3(Web3.HTTPProvider("https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY"))
print("连接状态:", w3.is_connected())
print("最新区块:", w3.eth.block_number)

# ERC-20 余额查询
ERC20_ABI = [
    {"inputs": [{"name": "account", "type": "address"}],
     "name": "balanceOf",
     "outputs": [{"name": "", "type": "uint256"}],
     "type": "function"},
    {"inputs": [],
     "name": "symbol",
     "outputs": [{"name": "", "type": "string"}],
     "type": "function"}
]

USDC_ADDRESS = "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"
usdc = w3.eth.contract(
    address=Web3.to_checksum_address(USDC_ADDRESS),
    abi=ERC20_ABI
)

address = "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"  # vitalik.eth
balance = usdc.functions.balanceOf(address).call()
symbol = usdc.functions.symbol().call()
print(f"{symbol} 余额: {balance / 10**6:.2f}")

# 发送交易（含签名）
def send_erc20(private_key, to_address, amount_wei):
    account = w3.eth.account.from_key(private_key)
    nonce = w3.eth.get_transaction_count(account.address)
    tx = usdc.functions.transfer(to_address, amount_wei).build_transaction({
        "from": account.address,
        "nonce": nonce,
        "gas": 100000,
        "maxFeePerGas": w3.to_wei(20, "gwei"),
        "maxPriorityFeePerGas": w3.to_wei(2, "gwei"),
    })
    signed = w3.eth.account.sign_transaction(tx, private_key)
    tx_hash = w3.eth.send_raw_transaction(signed.raw_transaction)
    return tx_hash.hex()

import { ethers } from 'ethers'

// 配置 provider
const provider = new ethers.JsonRpcProvider('https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY')

// 解析 ENS 名称
const address = await provider.resolveName('vitalik.eth')
console.log('vitalik.eth →', address)

// 订阅事件（实时监听 Transfer）
const ERC20_FILTER_ABI = ['event Transfer(address indexed from, address indexed to, uint256 value)']
const usdc = new ethers.Contract(
  '0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48',
  ERC20_FILTER_ABI,
  provider
)

usdc.on('Transfer', (from, to, value, event) => {
  const amount = ethers.formatUnits(value, 6)
  if (parseFloat(amount) > 100000) {
    console.log(`大额 USDC 转账: ${amount} USDC`)
    console.log(`from: ${from} → to: ${to}`)
    console.log(`tx: ${event.log.transactionHash}`)
  }
})