一个测试案例让交易所损失3000万：区块链安全测试的致命漏洞_千锋教育

class OrderStateMachine:
    def __init__(self):
        self.states = ['PENDING', 'PARTIAL_FILLED', 'FULL_FILLED', 'CANCELLED', 'EXPIRED']
        self.current_state = 'PENDING'
    
    def transition(self, action):
        # 定义状态转换规则
        transitions = {
            'PENDING': {'cancel': 'CANCELLED', 'partial_fill': 'PARTIAL_FILLED', 'full_fill': 'FULL_FILLED'},
            'PARTIAL_FILLED': {'cancel': 'CANCELLED', 'partial_fill': 'PARTIAL_FILLED', 'full_fill': 'FULL_FILLED'},
            'FULL_FILLED': {'cancel': None},  # 已成交订单不可撤单
            'CANCELLED': {'cancel': None},     # 已撤销订单不可重复撤销
        }
        next_state = transitions[self.current_state].get(action)
        if next_state is None:
            raise ValueError(f"非法转换：从{self.current_state}通过{action}")
        self.current_state = next_state
        return self.current_state

# 测试所有路径
def test_all_transitions():
    for start_state in ['PENDING', 'PARTIAL_FILLED', 'FULL_FILLED', 'CANCELLED']:
        for action in ['cancel', 'partial_fill', 'full_fill']:
            try:
                machine = OrderStateMachine()
                machine.current_state = start_state
                result = machine.transition(action)
                print(f"状态{start_state} + 动作{action} → {result}")
            except ValueError as e:
                print(f"状态{start_state} + 动作{action} → 非法: {e}")

import time
import threading
from datetime import datetime

class OrderBook:
    def __init__(self):
        self.buy_orders = []
        self.sell_orders = []
        self.lock = threading.Lock()
    
    def place_order(self, order):
        with self.lock:
            if order['type'] == 'buy':
                self.buy_orders.append(order)
            else:
                self.sell_orders.append(order)
            self.match_orders()
    
    def cancel_order(self, order_id):
        with self.lock:
            # 模拟撤单延迟
            time.sleep(0.1)  # 100ms网络延迟
            self.buy_orders = [o for o in self.buy_orders if o['id'] != order_id]
            self.sell_orders = [o for o in self.sell_orders if o['id'] != order_id]
    
    def match_orders(self):
        # 撮合逻辑
        for buy in self.buy_orders:
            for sell in self.sell_orders:
                if buy['price'] >= sell['price']:
                    print(f"成交：{buy['id']} vs {sell['id']} 价格{sell['price']}")
                    self.sell_orders.remove(sell)
                    self.buy_orders.remove(buy)
                    return

# 模拟时间差攻击
def attack():
    book = OrderBook()
    # 用户A挂买单
    book.place_order({'id': 'A', 'type': 'buy', 'price': 100})
    # 用户B挂卖单
    book.place_order({'id': 'B', 'type': 'sell', 'price': 101})
    
    # 用户A撤单
    def cancel():
        book.cancel_order('A')
    
    # 黑客在撤单瞬间发送成交
    def front_run():
        time.sleep(0.05)  # 比撤单快50ms
        book.match_orders()
    
    t1 = threading.Thread(target=cancel)
    t2 = threading.Thread(target=front_run)
    t1.start()
    t2.start()
    t1.join()
    t2.join()

attack()  # 输出：成交：A vs B 价格101

import random

class Oracle:
    def __init__(self):
        self.price = 100  # 初始价格
    
    def get_price(self):
        # 模拟预言机价格更新延迟
        time.sleep(0.2)  # 200ms更新延迟
        return self.price
    
    def set_price(self, new_price):
        self.price = new_price

class FlashLoanAttack:
    def __init__(self, oracle):
        self.oracle = oracle
        self.balance = 10000  # 攻击者初始资金
    
    def execute_attack(self):
        # 步骤1：借入巨量资金
        loan_amount = 1000000
        
        # 步骤2：操控价格
        fake_price = 50  # 将价格压低50%
        self.oracle.set_price(fake_price)
        
        # 步骤3：以低价买入
        buy_cost = loan_amount * self.oracle.get_price() / 100
        self.balance -= buy_cost
        
        # 步骤4：等待价格恢复
        time.sleep(0.3)
        self.oracle.set_price(100)  # 价格恢复
        
        # 步骤5：高价卖出
        sell_revenue = loan_amount * 100 / 100
        self.balance += sell_revenue
        
        # 步骤6：归还贷款
        loan_repay = loan_amount * 1.01  # 1%利息
        profit = self.balance - loan_repay
        return profit

# 测试攻击是否成功
oracle = Oracle()
attack = FlashLoanAttack(oracle)
profit = attack.execute_attack()
print(f"攻击利润：{profit}")  # 输出：攻击利润：-10000（失败）或 50000（成功）

import concurrent.futures
import random

class MatchingEngine:
    def __init__(self):
        self.order_book = {'buy': [], 'sell': []}
        self.lock = threading.Lock()
    
    def add_order(self, order):
        with self.lock:
            self.order_book[order['side']].append(order)
            self.match()
    
    def match(self):
        # 价格优先：买单价高优先，卖单价低优先
        self.order_book['buy'].sort(key=lambda x: -x['price'])
        self.order_book['sell'].sort(key=lambda x: x['price'])
        
        while self.order_book['buy'] and self.order_book['sell']:
            buy = self.order_book['buy'][0]
            sell = self.order_book['sell'][0]
            if buy['price'] >= sell['price']:
                trade_price = sell['price']
                trade_qty = min(buy['qty'], sell['qty'])
                print(f"成交：买单{buy['id']} vs 卖单{sell['id']} 价格{trade_price} 数量{trade_qty}")
                buy['qty'] -= trade_qty
                sell['qty'] -= trade_qty
                if buy['qty'] == 0:
                    self.order_book['buy'].pop(0)
                if sell['qty'] == 0:
                    self.order_book['sell'].pop(0)
            else:
                break

# 并发测试
def concurrent_test():
    engine = MatchingEngine()
    orders = []
    for i in range(1000):
        side = random.choice(['buy', 'sell'])
        price = random.randint(90, 110)
        qty = random.randint(1, 10)
        orders.append({'id': i, 'side': side, 'price': price, 'qty': qty})
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=50) as executor:
        futures = [executor.submit(engine.add_order, order) for order in orders]
        concurrent.futures.wait(futures)
    
    # 验证一致性：买单和卖单的累计数量应该相等
    total_buy_qty = sum(o['qty'] for o in engine.order_book['buy'])
    total_sell_qty = sum(o['qty'] for o in engine.order_book['sell'])
    print(f"剩余买单数量：{total_buy_qty}，剩余卖单数量：{total_sell_qty}")
    assert total_buy_qty == total_sell_qty, "并发一致性失败！"

concurrent_test()