3. DoIP车辆识别请求与响应实战

任务说明

本案例实现一个SOME/IP服务发现（SD）报文解析器，能够检测两类典型格式错误：长度字段不匹配（报文总长度与各条目声明长度不一致）和选项类型无效（选项类型值超出SOME/IP规范定义范围）。解析器对每个错误生成结构化诊断信息，便于上层服务发现逻辑决定丢弃或忽略。

核心实现

import struct
from enum import IntEnum

# SOME/IP SD 规范定义
class OptionType(IntEnum):
    CONFIGURATION = 0x01
    LOAD_BALANCING = 0x02
    ENDPOINT = 0x04
    MULTICAST = 0x06
    SD_ENDPOINT_OPTION = 0x24
    SD_MULTICAST_OPTION = 0x26

class SDError(Exception):
    """SD报文格式错误基类"""
    pass

class LengthMismatchError(SDError):
    def __init__(self, section, declared, actual):
        self.section = section
        self.declared = declared
        self.actual = actual
        super().__init__(f"{section}长度不匹配: 声明={declared}, 实际={actual}")

class InvalidOptionTypeError(SDError):
    def __init__(self, option_type, offset):
        self.option_type = option_type
        self.offset = offset
        super().__init__(f"偏移0x{offset:04x}处选项类型0x{option_type:02x}无效")

def parse_sd_message(data: bytes) -> dict:
    """解析SOME/IP SD报文，检测格式错误。
    返回解析结果字典；遇错误抛出SDError子类异常。
    """
    if len(data) < 16:
        raise LengthMismatchError("SD头部", 16, len(data))

    # 固定头部: 0-7 SOME/IP头, 8-11 flags/reserved, 12-15 entries数组长度
    entries_length = struct.unpack("!I", data[12:16])[0]
    if entries_length != len(data) - 16:
        raise LengthMismatchError("Entries数组", entries_length, len(data) - 16)

    offset = 16
    entries = []
    while offset < len(data):
        # 每个Entry: 8字节头 + 可选options
        if offset + 8 > len(data):
            raise LengthMismatchError("Entry头", 8, len(data) - offset)

        entry_type, entry_len = struct.unpack("!BB", data[offset:offset+2])
        # entry_len 表示该Entry总长度（含8字节头）
        if entry_len < 8 or offset + entry_len > len(data):
            raise LengthMismatchError(
                f"Entry(type={entry_type})", entry_len, len(data) - offset)

        entry_data = data[offset:offset+entry_len]
        # 解析该Entry内的Options
        options = _parse_options(entry_data[8:], offset + 8)
        entries.append({
            "type": entry_type,
            "length": entry_len,
            "options": options
        })
        offset += entry_len

    return {"entries": entries}

def _parse_options(opt_data: bytes, base_offset: int) -> list:
    """解析Entry内的Options列表，检测无效选项类型。"""
    options = []
    offset = 0
    while offset < len(opt_data):
        if offset + 4 > len(opt_data):
            raise LengthMismatchError("Option头", 4, len(opt_data) - offset)

        opt_type, opt_len = struct.unpack("!BB", opt_data[offset:offset+2])
        # 校验选项类型是否合法
        try:
            OptionType(opt_type)
        except ValueError:
            raise InvalidOptionTypeError(opt_type, base_offset + offset)

        if opt_len < 4 or offset + opt_len > len(opt_data):
            raise LengthMismatchError(
                f"Option(type=0x{opt_type:02x})", opt_len, len(opt_data) - offset)

        options.append({
            "type": opt_type,
            "length": opt_len,
            "data": opt_data[offset+4:offset+opt_len]
        })
        offset += opt_len

    return options

# ---------- 最小验证入口 ----------
if __name__ == "__main__":
    # 构造合法报文: 1个Service Entry, 含1个IPv4 Endpoint Option
    # Entry: type=0x00(FindService), len=16, options: type=0x04, len=12
    entry_body = struct.pack("!BBH", 0x00, 16, 0x0000)  # 8字节Entry头(简化)
    entry_body += struct.pack("!BBH", 0x04, 12, 0x0000)  # Option头
    entry_body += struct.pack("!4sH", b"\x0a\x00\x00\x01", 30501)  # IPv4+端口
    entry_body = entry_body[:8] + struct.pack("!H", 0) + entry_body[10:]  # 填充至16字节
    entry_body = entry_body[:8] + struct.pack("!BBH", 0x00, 16, 0) + entry_body[8:8]  # 修正

    # 重新构造更清晰的合法Entry
    entry = struct.pack("!BBH", 0x00, 16, 0)  # type, len=16, reserved
    entry += struct.pack("!I", 0)  # service_id
    entry += struct.pack("!BBH", 0x04, 12, 0)  # Option: type=4, len=12
    entry += struct.pack("!4sH", b"\x0a\x00\x00\x01", 30501)  # IPv4, port
    # 此时entry长度 = 8 + 12 = 20, 修正len字段
    entry = struct.pack("!BBH", 0x00, 20, 0) + entry[4:]

    sd_msg = struct.pack("!I", 0x00000000)  # SOME/IP头简化
    sd_msg += struct.pack("!I", 0)  # flags
    sd_msg += struct.pack("!I", len(entry))  # entries数组长度
    sd_msg += entry

    # 测试1: 合法报文
    result = parse_sd_message(sd_msg)
    print(f"合法报文解析成功: {len(result['entries'])}个Entry")

    # 测试2: 长度不匹配 (声明长度比实际大)
    bad_len = bytearray(sd_msg)
    struct.pack_into("!I", bad_len, 12, 999)  # 声明999字节
    try:
        parse_sd_message(bytes(bad_len))
    except LengthMismatchError as e:
        print(f"测试2通过: {e}")

    # 测试3: 无效选项类型 (0x99)
    bad_opt = bytearray(sd_msg)
    # 找到第一个Option的type位置: 16(头) + 8(entry头) = 24
    bad_opt[24] = 0x99
    try:
        parse_sd_message(bytes(bad_opt))
    except InvalidOptionTypeError as e:
        print(f"测试3通过: {e}")

    print("所有测试完成")

运行与验证

运行方式：将代码保存为 sd_parser.py，执行 python3 sd_parser.py。

预期输出（确定性结果）：

合法报文解析成功: 1个Entry
测试2通过: Entries数组长度不匹配: 声明=999, 实际=20
测试3通过: 偏移0x0018处选项类型0x99无效

通过条件：三条输出均出现且无异常堆栈，说明长度校验与选项类型校验均按预期工作。

关键说明

解析器覆盖SD报文核心结构：SOME/IP头（前8字节，此处简化）、Entries数组长度字段（偏移12）、Entry头（含类型与长度）、Option头（含类型与长度）。
长度校验采用声明值对比实际剩余值的方式，任何不一致立即抛出LengthMismatchError并携带具体位置信息。
选项类型校验通过枚举OptionType实现，未知类型抛出InvalidOptionTypeError并记录偏移量，便于定位问题字节。
实际工程中，捕获SDError后可根据错误类型决定丢弃报文或进入降级处理流程；本案例聚焦检测逻辑本身。

📚 课程目录

3. DoIP车辆识别请求与响应实战

任务说明

核心实现

运行与验证

关键说明