4. 解析WPA3 Probe Response帧

任务说明

本案例实现一个WPA3 Probe Response帧解析器，输入为完整的802.11管理帧字节序列，输出为解析后的帧字段结构，并识别该帧宣告的RSN安全能力属于WPA3-Personal还是WPA3-Enterprise。判定依据为RSN IE中的AKM套件类型：00-0F-AC-8（WPA3-Personal SAE）或00-0F-AC-9（WPA3-Enterprise Suite B-192），以及00-0F-AC-12（WPA3-Enterprise过渡模式）。

核心实现

import struct
from dataclasses import dataclass, field
from typing import List, Tuple

# ---- 802.11管理帧常量 ----
FRAME_TYPE_MGMT = 0x00
FRAME_SUBTYPE_PROBE_RESP = 0x05

# ---- RSN IE中的AKM套件标识 ----
AKM_SAE = (0x00, 0x0F, 0xAC, 0x08)          # WPA3-Personal
AKM_FT_SAE = (0x00, 0x0F, 0xAC, 0x09)       # WPA3-Personal (FT)
AKM_SUITE_B_192 = (0x00, 0x0F, 0xAC, 0x0C)  # WPA3-Enterprise (Suite B 192-bit)
AKM_SUITE_B = (0x00, 0x0F, 0xAC, 0x0B)      # WPA3-Enterprise (Suite B)

@dataclass
class TaggedElement:
    tag_number: int
    tag_length: int
    tag_value: bytes

@dataclass
class RSNAKMSuite:
    oui: Tuple[int, int, int]
    suite_type: int

@dataclass
class RSNInfo:
    version: int
    group_cipher: Tuple[int, int, int, int] = None
    pairwise_ciphers: List[Tuple[int, int, int, int]] = field(default_factory=list)
    akm_suites: List[RSNAKMSuite] = field(default_factory=list)
    rsn_capabilities: int = 0

@dataclass
class ProbeResponseFrame:
    frame_control: int
    duration: int
    dst_addr: str
    src_addr: str
    bssid: str
    seq_ctrl: int
    timestamp: int
    beacon_interval: int
    capability_info: int
    ssid: str = ""
    rsn: RSNInfo = None
    raw_tagged_elements: List[TaggedElement] = field(default_factory=list)

def mac_to_str(mac_bytes: bytes) -> str:
    return ":".join(f"{b:02X}" for b in mac_bytes)

def parse_tagged_elements(data: bytes) -> List[TaggedElement]:
    """解析Tagged Parameters区域，返回所有Tagged Element列表"""
    elements = []
    offset = 0
    while offset < len(data):
        tag_num = data[offset]
        tag_len = data[offset + 1]
        tag_val = data[offset + 2 : offset + 2 + tag_len]
        elements.append(TaggedElement(tag_num, tag_len, tag_val))
        offset += 2 + tag_len
    return elements

def parse_rsn_ie(tag_value: bytes) -> RSNInfo:
    """解析RSN (Tag 48) Information Element"""
    if len(tag_value) < 2:
        raise ValueError("RSN IE长度不足")
    
    rsn = RSNInfo(version=struct.unpack("<H", tag_value[0:2])[0])
    offset = 2
    
    # Group Cipher Suite (4 bytes)
    if offset + 4 <= len(tag_value):
        rsn.group_cipher = tuple(tag_value[offset : offset + 4])
        offset += 4
    
    # Pairwise Cipher Suite Count + List
    if offset + 2 <= len(tag_value):
        pairwise_count = struct.unpack("<H", tag_value[offset : offset + 2])[0]
        offset += 2
        for _ in range(pairwise_count):
            if offset + 4 > len(tag_value):
                break
            rsn.pairwise_ciphers.append(tuple(tag_value[offset : offset + 4]))
            offset += 4
    
    # AKM Suite Count + List
    if offset + 2 <= len(tag_value):
        akm_count = struct.unpack("<H", tag_value[offset : offset + 2])[0]
        offset += 2
        for _ in range(akm_count):
            if offset + 4 > len(tag_value):
                break
            akm_bytes = tag_value[offset : offset + 4]
            rsn.akm_suites.append(RSNAKMSuite(
                oui=(akm_bytes[0], akm_bytes[1], akm_bytes[2]),
                suite_type=akm_bytes[3]
            ))
            offset += 4
    
    # RSN Capabilities (2 bytes, optional)
    if offset + 2 <= len(tag_value):
        rsn.rsn_capabilities = struct.unpack("<H", tag_value[offset : offset + 2])[0]
    
    return rsn

def parse_probe_response(frame: bytes) -> ProbeResponseFrame:
    """解析完整的802.11 Probe Response管理帧"""
    if len(frame) < 36:
        raise ValueError("帧长度不足，无法构成完整Probe Response")
    
    # 固定头部 (24字节)
    frame_ctrl = struct.unpack("<H", frame[0:2])[0]
    subtype = (frame_ctrl >> 4) & 0x0F
    frame_type = (frame_ctrl >> 2) & 0x03
    
    if frame_type != FRAME_TYPE_MGMT or subtype != FRAME_SUBTYPE_PROBE_RESP:
        raise ValueError(f"不是Probe Response帧 (type={frame_type}, subtype={subtype})")
    
    duration = struct.unpack("<H", frame[2:4])[0]
    dst = mac_to_str(frame[4:10])
    src = mac_to_str(frame[10:16])
    bssid = mac_to_str(frame[16:22])
    seq_ctrl = struct.unpack("<H", frame[22:24])[0]
    
    # 帧体 (Frame Body)
    body = frame[24:]
    if len(body) < 12:
        raise ValueError("帧体长度不足")
    
    timestamp = struct.unpack("<Q", body[0:8])[0]
    beacon_interval = struct.unpack("<H", body[8:10])[0]
    capability_info = struct.unpack("<H", body[10:12])[0]
    
    # Tagged Parameters 从偏移12开始
    tagged_data = body[12:]
    elements = parse_tagged_elements(tagged_data)
    
    result = ProbeResponseFrame(
        frame_control=frame_ctrl,
        duration=duration,
        dst_addr=dst,
        src_addr=src,
        bssid=bssid,
        seq_ctrl=seq_ctrl,
        timestamp=timestamp,
        beacon_interval=beacon_interval,
        capability_info=capability_info,
        raw_tagged_elements=elements
    )
    
    # 提取SSID (Tag 0) 和 RSN (Tag 48)
    for elem in elements:
        if elem.tag_number == 0:  # SSID
            result.ssid = elem.tag_value.decode("utf-8", errors="replace")
        elif elem.tag_number == 48:  # RSN
            result.rsn = parse_rsn_ie(elem.tag_value)
    
    return result

def classify_wpa3(rsn: RSNInfo) -> str:
    """根据RSN IE中的AKM套件判断WPA3类型"""
    if rsn is None or not rsn.akm_suites:
        return "非WPA3（无RSN或AKM套件）"
    
    for akm in rsn.akm_suites:
        suite = (akm.oui[0], akm.oui[1], akm.oui[2], akm.suite_type)
        if suite == AKM_SAE or suite == AKM_FT_SAE:
            return "WPA3-Personal"
        if suite == AKM_SUITE_B_192 or suite == AKM_SUITE_B:
            return "WPA3-Enterprise"
    
    return "非WPA3（AKM套件不支持WPA3）"

# ---- 样例输入：构造一个WPA3-Personal的Probe Response帧 ----
def build_sample_frame_wpa3_personal() -> bytes:
    """构造一个宣告WPA3-Personal (SAE) 的Probe Response帧"""
    # 固定头部
    fc = (FRAME_TYPE_MGMT << 2) | (FRAME_SUBTYPE_PROBE_RESP << 4) | 0x00  # 无加密
    header = struct.pack("<H", fc)
    header += struct.pack("<H", 0)  # duration
    header += bytes.fromhex("FFFFFFFFFFFF")  # dst: broadcast
    header += bytes.fromhex("AABBCCDDEEFF")  # src
    header += bytes.fromhex("AABBCCDDEEFF")  # bssid
    header += struct.pack("<H", 0x0000)  # seq_ctrl
    
    # 帧体
    body = struct.pack("<Q", 12345678)  # timestamp
    body += struct.pack("<H", 100)      # beacon interval
    body += struct.pack("<H", 0x0001)   # capability: ESS
    
    # Tagged Elements
    # SSID (Tag 0)
    ssid = b"WPA3-Lab"
    body += bytes([0, len(ssid)]) + ssid
    
    # RSN IE (Tag 48)
    rsn_body = struct.pack("<H", 1)  # version
    rsn_body += bytes([0x00, 0x0F, 0xAC, 0x04])  # group cipher: CCMP
    rsn_body += struct.pack("<H", 1)  # pairwise count
    rsn_body += bytes([0x00, 0x0F, 0xAC, 0x04])  # pairwise: CCMP
    rsn_body += struct.pack("<H", 1)  # akm count
    rsn_body += bytes([0x00, 0x0F, 0xAC, 0x08])  # AKM: SAE (WPA3-Personal)
    rsn_body += struct.pack("<H", 0x0000)  # capabilities
    
    body += bytes([48, len(rsn_body)]) + rsn_body
    
    return header + body

# ---- 主流程 ----
if __name__ == "__main__":
    # 使用样例帧进行解析
    sample_frame = build_sample_frame_wpa3_personal()
    parsed = parse_probe_response(sample_frame)
    
    print(f"源MAC: {parsed.src_addr}")
    print(f"BSSID: {parsed.bssid}")
    print(f"SSID: {parsed.ssid}")
    print(f"Beacon Interval: {parsed.beacon_interval}")
    
    if parsed.rsn:
        print(f"RSN版本: {parsed.rsn.version}")
        print(f"AKM套件数量: {len(parsed.rsn.akm_suites)}")
        for akm in parsed.rsn.akm_suites:
            print(f"  AKM OUI: {akm.oui[0]:02X}-{akm.oui[1]:02X}-{akm.oui[2]:02X}, 类型: {akm.suite_type:02X}")
        print(f"WPA3分类: {classify_wpa3(parsed.rsn)}")
    else:
        print("未找到RSN IE")

验证方法

直接运行上述Python脚本，观察输出。预期输出应包含：源MAC为AA:BB:CC:DD:EE:FF，SSID为WPA3-Lab，AKM套件类型为08（SAE），最终分类结果为WPA3-Personal。若需验证WPA3-Enterprise识别，可将样例帧构造函数中AKM套件字节改为00 0F AC 0C（Suite B 192-bit），重新运行后分类结果应变为WPA3-Enterprise。判定标准：解析出的帧字段与构造输入完全一致，且AKM套件类型与WPA3分类结果正确对应。