JT/T 808 协议

概述

JT/T 808 是交通运输行业标准《道路运输车辆卫星定位系统 北斗兼容车载终端通讯协议》，规定了车载终端与监控平台之间的数据通信格式

协议基于 TCP 或 UDP，采用大端字节序，报文整体以 0x7E 为帧边界，内部需经过转义处理

报文封装格式

每个完整报文由标识位包裹，内部包含消息头、消息体和校验码

+--------+--------+----------+--------+--------+
| 标识位 | 消息头 | 消息体   | 校验码 | 标识位 |
| 0x7e   | 变长   | 变长     | 1 byte | 0x7e   |
+--------+--------+----------+--------+--------+

• 标识位：固定 0x7E，标识一帧的开始和结束

• 消息头：固定 12 字节（无分包）或 16 字节（有分包），包含消息 ID、属性、终端手机号、流水号等

• 消息体：长度由消息头中的 “消息体长度” 字段决定，内容取决于消息 ID

• 校验码：从消息头第一个字节到消息体最后一个字节，按字节异或，再与 0x7E 异或

消息头结构

消息头结构如下：

Offset: 0      2       4        10       12
        +------+-------+--------+--------+
        | 消息ID | 消息体属性 | 终端手机号 | 消息流水号 |
        +------+-------+--------+--------+
          u16    u16       u48       u16

消息 ID（2 字节，u16）

标识消息类型，如：

• 0x0001：终端通用应答

• 0x0100：终端注册

• 0x0200：位置信息汇报

• 0x8001：平台通用应答
  （高位可用于区分终端/平台消息）

消息体属性（2 字节，u16）

按位域解析

Bit:  15       14   13         10 9          0
      +--------+----+-----------+-----------+
      | 分包标志 | 加密 | 保留/子包 | 消息体长度  |
      +--------+----+-----------+-----------+

• bit 15：1 表示消息体为分包数据，此时消息头会扩展 4 字节分包信息

• bit 14-13：加密方式：00=不加密，01=RSA 等

• bit 12-10：保留，或部分文档定义为分包计数器

• bit 9-0：消息体长度（0~1023），单位字节

若分包标志为 1，则消息头变为 16 字节，在流水号后追加

+----------------+-----------------+
| 消息总包数 (u16) | 包序号 (u16)    |
+----------------+-----------------+

终端手机号（6 字节，u48）

BCD 码格式，长度 6 字节，存储 12 位手机号（不足高位补 0）。例如 013800123456 编码为 0x01 0x38 0x00 0x12 0x34 0x56

消息流水号（2 字节，u16）

由发送方递增，从 0 开始循环，用于匹配应答

通用解题脚本

import argparse
import json
import os
import re
import socket
import struct
from collections import defaultdict

MSG_NAMES = {
    0x0001: "终端通用应答",
    0x0002: "终端心跳",
    0x0003: "终端注销",
    0x0100: "终端注册",
    0x0102: "终端鉴权",
    0x0104: "查询终端参数应答",
    0x0107: "查询终端属性应答",
    0x0200: "位置信息汇报",
    0x0201: "位置信息查询应答",
    0x0301: "事件报告",
    0x0302: "提问应答",
    0x0500: "车辆控制应答",
    0x0700: "行驶记录数据上传",
    0x0701: "电子运单上报",
    0x0702: "驾驶员身份信息采集上报",
    0x0704: "定位数据批量上传",
    0x0800: "多媒体事件信息上传",
    0x0801: "多媒体数据上传",
    0x0900: "数据上行透传",
    0x0A00: "终端RSA公钥",
    0x8001: "平台通用应答",
    0x8100: "终端注册应答",
    0x8103: "设置终端参数",
    0x8104: "查询终端参数",
    0x8105: "终端控制",
    0x8201: "位置信息查询",
    0x8202: "临时位置跟踪控制",
    0x8300: "文本信息下发",
    0x8400: "电话回拨",
    0x8500: "车辆控制",
    0x8600: "设置圆形区域",
    0x8602: "设置矩形区域",
    0x8604: "设置多边形区域",
    0x8606: "设置路线",
    0x8801: "摄像头立即拍摄命令",
    0x8900: "数据下行透传",
}

RESULT_NAMES = {
    0: "成功/确认",
    1: "失败",
    2: "消息有误",
    3: "不支持",
    4: "报警处理确认",
}

def hx(b):
    return b.hex()

def bcd_to_str(b):
    out = []
    for x in b:
        hi, lo = (x >> 4) & 0xF, x & 0xF
        out.append(str(hi) if hi <= 9 else "?")
        out.append(str(lo) if lo <= 9 else "?")
    return "".join(out)

def bcd_time_yyMMddhhmmss(b):
    s = bcd_to_str(b)
    if len(s) != 12 or "?" in s:
        return s
    yy = int(s[:2])
    # JT808 BCD timestamp usually means 20yy in modern traffic.
    return "20%s-%s-%s %s:%s:%s" % (s[0:2], s[2:4], s[4:6], s[6:8], s[8:10], s[10:12])

def try_decode(data):
    """Return the most readable text plus all candidates."""
    if not data:
        return "", {}
    encs = ["utf-8", "gbk", "gb18030", "utf-16-be", "utf-16-le", "ascii"]
    candidates = {}
    for enc in encs:
        try:
            text = data.decode(enc).rstrip("\x00")
            if text:
                candidates[enc] = text
        except Exception:
            pass
    if not candidates:
        return "", {}
    # Prefer printable and meaningful text. CTF hidden strings often use UTF-16BE.
    def score(item):
        enc, text = item
        printable = sum(1 for c in text if c.isprintable())
        ascii_letters = sum(1 for c in text if 32 <= ord(c) <= 126)
        chinese = sum(1 for c in text if "\u4e00" <= c <= "\u9fff")
        penalty = text.count("\ufffd") * 10 + text.count("\x00") * 5
        bonus = 5 if enc in ("gbk", "gb18030", "utf-8") else 0
        return printable + ascii_letters + chinese + bonus - penalty
    best_enc, best_text = max(candidates.items(), key=score)
    return best_text, candidates

def jt808_unescape(frame):
    """Input includes 0x7e delimiters. Output excludes delimiters and is unescaped."""
    if len(frame) >= 2 and frame[0] == 0x7E and frame[-1] == 0x7E:
        data = frame[1:-1]
    else:
        data = frame
    out = bytearray()
    i = 0
    while i < len(data):
        if data[i] == 0x7D and i + 1 < len(data):
            if data[i + 1] == 0x02:
                out.append(0x7E)
                i += 2
                continue
            if data[i + 1] == 0x01:
                out.append(0x7D)
                i += 2
                continue
        out.append(data[i])
        i += 1
    return bytes(out)

def jt808_checksum(data_without_delimiters_and_unescaped_without_checksum):
    c = 0
    for x in data_without_delimiters_and_unescaped_without_checksum:
        c ^= x
    return c

def extract_frames_from_bytes(buf):
    """Extract delimiter-based frames from a byte stream."""
    frames = []
    start = None
    for i, b in enumerate(buf):
        if b == 0x7E:
            if start is not None and i > start:
                frame = buf[start:i+1]
                # Minimal header after unescape: msg_id 2 + prop 2 + terminal 6 + seq 2 + checksum 1
                if len(frame) >= 2 + 2 + 2 + 6 + 2 + 1:
                    frames.append(frame)
            start = i
    return frames

def parse_pcap_classic(path):
    """
    Return list of source dictionaries:
    {kind: 'tcp_stream'/'udp_packet', desc: ..., payload: bytes}
    Supports classic PCAP with Ethernet/IPv4/TCP/UDP.
    """
    sources = []
    with open(path, "rb") as f:
        gh = f.read(24)
        if len(gh) < 24:
            return sources

        magic = gh[:4]
        if magic in (b"\xd4\xc3\xb2\xa1", b"\x4d\x3c\xb2\xa1"):
            endian = "<"
        elif magic in (b"\xa1\xb2\xc3\xd4", b"\xa1\xb2\x3c\x4d"):
            endian = ">"
        else:
            return sources

        _magic, vmaj, vmin, tz, sig, snaplen, linktype = struct.unpack(endian + "IHHIIII", gh)

        tcp_streams = defaultdict(bytearray)
        udp_payloads = []

        while True:
            ph = f.read(16)
            if len(ph) < 16:
                break
            ts_sec, ts_frac, incl_len, orig_len = struct.unpack(endian + "IIII", ph)
            pkt = f.read(incl_len)
            if len(pkt) < 14:
                continue

            # Ethernet only. Other linktypes are still covered by raw fallback.
            if linktype != 1:
                continue

            off = 12
            eth_type = struct.unpack("!H", pkt[off:off+2])[0]
            l3off = 14
            # VLAN tag(s)
            while eth_type in (0x8100, 0x88A8, 0x9100) and len(pkt) >= l3off + 4:
                eth_type = struct.unpack("!H", pkt[l3off+2:l3off+4])[0]
                l3off += 4

            if eth_type != 0x0800 or len(pkt) < l3off + 20:
                continue

            ip = pkt[l3off:]
            ihl = (ip[0] & 0x0F) * 4
            if len(ip) < ihl or ihl < 20:
                continue

            proto = ip[9]
            src = socket.inet_ntoa(ip[12:16])
            dst = socket.inet_ntoa(ip[16:20])
            total_len = struct.unpack("!H", ip[2:4])[0]
            ip_payload = ip[ihl:total_len] if total_len >= ihl else ip[ihl:]

            if proto == 6 and len(ip_payload) >= 20:
                tcp = ip_payload
                sport, dport = struct.unpack("!HH", tcp[:4])
                data_offset = (tcp[12] >> 4) * 4
                if len(tcp) < data_offset:
                    continue
                payload = tcp[data_offset:]
                if payload:
                    key = (src, sport, dst, dport, "TCP")
                    tcp_streams[key].extend(payload)
            elif proto == 17 and len(ip_payload) >= 8:
                udp = ip_payload
                sport, dport, ulen, _ = struct.unpack("!HHHH", udp[:8])
                payload = udp[8:ulen] if ulen >= 8 else udp[8:]
                if payload:
                    udp_payloads.append({
                        "kind": "udp_packet",
                        "desc": "%s:%d -> %s:%d UDP" % (src, sport, dst, dport),
                        "payload": payload
                    })

        for key, payload in tcp_streams.items():
            src, sport, dst, dport, proto = key
            sources.append({
                "kind": "tcp_stream",
                "desc": "%s:%d -> %s:%d %s" % (src, sport, dst, dport, proto),
                "payload": bytes(payload)
            })
        sources.extend(udp_payloads)
    return sources

def parse_with_scapy(path):
    """Optional PCAP/PCAPNG reader. Returns [] if Scapy is unavailable."""
    try:
        from scapy.all import rdpcap, TCP, UDP, IP, IPv6, Raw
    except Exception:
        return []

    sources = []
    tcp_streams = defaultdict(bytearray)

    packets = rdpcap(path)
    for pkt in packets:
        if Raw not in pkt:
            continue
        payload = bytes(pkt[Raw].load)
        if not payload:
            continue

        src = dst = "?"
        if IP in pkt:
            src, dst = pkt[IP].src, pkt[IP].dst
        elif IPv6 in pkt:
            src, dst = pkt[IPv6].src, pkt[IPv6].dst

        if TCP in pkt:
            sport, dport = int(pkt[TCP].sport), int(pkt[TCP].dport)
            key = (src, sport, dst, dport, "TCP")
            tcp_streams[key].extend(payload)
        elif UDP in pkt:
            sport, dport = int(pkt[UDP].sport), int(pkt[UDP].dport)
            sources.append({
                "kind": "udp_packet",
                "desc": "%s:%d -> %s:%d UDP" % (src, sport, dst, dport),
                "payload": payload
            })

    for key, payload in tcp_streams.items():
        src, sport, dst, dport, proto = key
        sources.append({
            "kind": "tcp_stream",
            "desc": "%s:%d -> %s:%d %s" % (src, sport, dst, dport, proto),
            "payload": bytes(payload)
        })
    return sources

def get_sources(path, force_raw=False):
    sources = []
    if not force_raw:
        sources = parse_with_scapy(path)
        if not sources:
            sources = parse_pcap_classic(path)

    # Raw fallback: useful for pcapng, unknown linktype, or files containing only hex-ish binary.
    with open(path, "rb") as f:
        raw = f.read()
    sources.append({
        "kind": "raw_scan",
        "desc": "raw whole-file scan",
        "payload": raw
    })
    return sources

def parse_extra_items(extra):
    items = []
    i = 0
    while i + 2 <= len(extra):
        item_id = extra[i]
        item_len = extra[i + 1]
        i += 2
        value = extra[i:i + item_len]
        i += item_len
        item = {
            "id": "0x%02X" % item_id,
            "len": item_len,
            "raw": hx(value)
        }
        try:
            if item_id == 0x01 and len(value) == 4:
                item["name"] = "里程"
                item["value"] = struct.unpack("!I", value)[0] / 10
                item["unit"] = "km"
            elif item_id == 0x02 and len(value) == 2:
                item["name"] = "油量"
                item["value"] = struct.unpack("!H", value)[0] / 10
                item["unit"] = "L"
            elif item_id == 0x03 and len(value) == 2:
                item["name"] = "行驶记录速度"
                item["value"] = struct.unpack("!H", value)[0] / 10
                item["unit"] = "km/h"
            elif item_id == 0x04 and len(value) == 2:
                item["name"] = "需要人工确认报警事件ID"
                item["value"] = struct.unpack("!H", value)[0]
            elif item_id == 0x30 and len(value) == 1:
                item["name"] = "无线通信网络信号强度"
                item["value"] = value[0]
            elif item_id == 0x31 and len(value) == 1:
                item["name"] = "GNSS定位卫星数"
                item["value"] = value[0]
            else:
                item["name"] = "未知/扩展项"
        except Exception:
            item["name"] = "解析失败"
        items.append(item)
    return items

def parse_location_body(body, offset=0):
    if len(body) - offset < 28:
        return {"error": "location body too short", "raw": hx(body[offset:])}

    p = offset
    alarm = struct.unpack("!I", body[p:p+4])[0]; p += 4
    status = struct.unpack("!I", body[p:p+4])[0]; p += 4
    lat_raw = struct.unpack("!I", body[p:p+4])[0]; p += 4
    lon_raw = struct.unpack("!I", body[p:p+4])[0]; p += 4
    altitude = struct.unpack("!H", body[p:p+2])[0]; p += 2
    speed = struct.unpack("!H", body[p:p+2])[0] / 10; p += 2
    direction = struct.unpack("!H", body[p:p+2])[0]; p += 2
    tm = bcd_time_yyMMddhhmmss(body[p:p+6]); p += 6

    # Sign bits are in status word in many implementations:
    # bit2 = south latitude, bit3 = west longitude.
    lat = lat_raw / 1_000_000
    lon = lon_raw / 1_000_000
    if status & (1 << 2):
        lat = -lat
    if status & (1 << 3):
        lon = -lon

    return {
        "alarm_flag": "0x%08X" % alarm,
        "status": "0x%08X" % status,
        "latitude": lat,
        "longitude": lon,
        "altitude_m": altitude,
        "speed_kmh": speed,
        "direction": direction,
        "time": tm,
        "extra_items": parse_extra_items(body[p:]),
        "parsed_len": p + len(body[p:])
    }

def parse_0702_driver(body, versioned=False):
    res = {"raw": hx(body)}
    if len(body) < 7:
        res["error"] = "body too short"
        return res

    p = 0
    status = body[p]; p += 1
    res["status"] = status
    res["status_desc"] = {0x01: "从业资格证IC卡插入", 0x02: "从业资格证IC卡拔出"}.get(status, "未知")
    res["time"] = bcd_time_yyMMddhhmmss(body[p:p+6]); p += 6

    if p >= len(body):
        return res

    # Usually exists when status == 0x01.
    read_result = body[p]; p += 1
    res["ic_card_read_result"] = read_result
    res["ic_card_read_result_desc"] = {0: "成功", 1: "卡片密钥认证未通过", 2: "卡片已被锁定", 3: "卡片被拔出", 4: "数据校验错误"}.get(read_result, "未知/厂商扩展")

    if p >= len(body):
        return res

    name_len = body[p]; p += 1
    res["driver_name_len"] = name_len

    if p + name_len <= len(body):
        name_bytes = body[p:p+name_len]
        p += name_len
        name, candidates = try_decode(name_bytes)
        res["driver_name"] = name
        res["driver_name_raw"] = hx(name_bytes)
        res["driver_name_decode_candidates"] = candidates
    else:
        res["warning"] = "name_len exceeds remaining body"
        name_bytes = body[p:]
        name, candidates = try_decode(name_bytes)
        res["driver_name"] = name
        res["driver_name_raw"] = hx(name_bytes)
        res["driver_name_decode_candidates"] = candidates
        return res

    remaining = body[p:]
    res["remaining_hex"] = hx(remaining)

    # Try to locate ID card-like strings from remaining bytes.
    ascii_text = ""
    try:
        ascii_text = remaining.decode("ascii", errors="ignore")
    except Exception:
        pass
    ids = re.findall(r"[0-9]{15,18}[0-9Xx]?", ascii_text)
    if ids:
        res["possible_id_cards"] = ids

    # Some versions/implementations store a fixed 20-byte ID card field.
    if len(remaining) >= 20:
        fixed20 = remaining[:20].rstrip(b"\x00")
        try:
            fixed20_text = fixed20.decode("ascii")
            if fixed20_text:
                res["id_card_fixed20_guess"] = fixed20_text
        except Exception:
            pass

    # Keep the rest generic because 2013/2019/vendor fields vary.
    return res

def parse_register_0100(body):
    res = {"raw": hx(body)}
    if len(body) < 37:
        res["warning"] = "body shorter than common 2013 register format"
        return res
    p = 0
    res["province_id"] = struct.unpack("!H", body[p:p+2])[0]; p += 2
    res["city_id"] = struct.unpack("!H", body[p:p+2])[0]; p += 2
    res["manufacturer_id"] = try_decode(body[p:p+5])[0] or hx(body[p:p+5]); p += 5
    res["terminal_model"] = try_decode(body[p:p+20])[0] or hx(body[p:p+20]); p += 20
    res["terminal_id"] = try_decode(body[p:p+7])[0] or hx(body[p:p+7]); p += 7
    res["plate_color"] = body[p]; p += 1
    plate, cand = try_decode(body[p:])
    res["plate_no"] = plate
    res["plate_decode_candidates"] = cand
    return res

def parse_auth_0102(body):
    text, candidates = try_decode(body)
    return {"auth_code": text, "decode_candidates": candidates, "raw": hx(body)}

def parse_common_response(body):
    if len(body) < 5:
        return {"raw": hx(body), "error": "body too short"}
    serial = struct.unpack("!H", body[0:2])[0]
    msg_id = struct.unpack("!H", body[2:4])[0]
    result = body[4]
    return {
        "response_serial": serial,
        "response_msg_id": "0x%04X" % msg_id,
        "response_msg_name": MSG_NAMES.get(msg_id, "未知"),
        "result": result,
        "result_desc": RESULT_NAMES.get(result, "未知"),
        "raw": hx(body)
    }

def parse_0704_batch_location(body):
    res = {"raw": hx(body)}
    if len(body) < 3:
        res["error"] = "body too short"
        return res
    count = struct.unpack("!H", body[0:2])[0]
    batch_type = body[2]
    p = 3
    res["count"] = count
    res["batch_type"] = batch_type
    res["locations"] = []
    for idx in range(count):
        if p + 2 > len(body):
            res["warning"] = "truncated before item %d length" % idx
            break
        item_len = struct.unpack("!H", body[p:p+2])[0]
        p += 2
        item_body = body[p:p+item_len]
        p += item_len
        item = parse_location_body(item_body)
        item["item_len"] = item_len
        res["locations"].append(item)
    return res

def parse_body(msg_id, body, versioned=False):
    if msg_id in (0x0001, 0x8001):
        return parse_common_response(body)
    if msg_id == 0x0100:
        return parse_register_0100(body)
    if msg_id == 0x0102:
        return parse_auth_0102(body)
    if msg_id == 0x0200:
        return parse_location_body(body)
    if msg_id == 0x0201:
        if len(body) >= 2:
            return {
                "response_serial": struct.unpack("!H", body[:2])[0],
                "location": parse_location_body(body, 2),
                "raw": hx(body)
            }
        return {"raw": hx(body), "error": "body too short"}
    if msg_id == 0x0702:
        return parse_0702_driver(body, versioned=versioned)
    if msg_id == 0x0704:
        return parse_0704_batch_location(body)

    text, candidates = try_decode(body)
    return {
        "raw": hx(body),
        "text_guess": text,
        "decode_candidates": candidates
    }

def parse_jt808_frame(frame, source_desc="", index=0):
    raw = jt808_unescape(frame)
    result = {
        "index": index,
        "source": source_desc,
        "frame_hex": hx(frame),
        "unescaped_hex": hx(raw),
    }

    if len(raw) < 2 + 2 + 6 + 2 + 1:
        result["error"] = "frame too short"
        return result

    checksum_expected = raw[-1]
    checksum_calc = jt808_checksum(raw[:-1])
    result["checksum"] = "0x%02X" % checksum_expected
    result["checksum_calc"] = "0x%02X" % checksum_calc
    result["checksum_ok"] = checksum_expected == checksum_calc

    data = raw[:-1]
    p = 0
    msg_id = struct.unpack("!H", data[p:p+2])[0]; p += 2
    prop = struct.unpack("!H", data[p:p+2])[0]; p += 2

    body_len = prop & 0x03FF
    encryption = (prop >> 10) & 0x07
    subpackage = bool(prop & 0x2000)
    versioned = bool(prop & 0x4000)

    result.update({
        "msg_id": "0x%04X" % msg_id,
        "msg_name": MSG_NAMES.get(msg_id, "未知消息"),
        "property": "0x%04X" % prop,
        "body_len_declared": body_len,
        "encryption_flag": encryption,
        "subpackage": subpackage,
        "versioned_2019": versioned,
    })

    if versioned:
        if len(data) < p + 1 + 10 + 2:
            result["error"] = "2019 header too short"
            return result
        version = data[p]; p += 1
        terminal_raw = data[p:p+10]; p += 10
        result["protocol_version"] = "0x%02X" % version
        result["terminal_id_raw"] = hx(terminal_raw)
        result["terminal_id_bcd"] = bcd_to_str(terminal_raw)
    else:
        if len(data) < p + 6 + 2:
            result["error"] = "legacy header too short"
            return result
        terminal_raw = data[p:p+6]; p += 6
        result["terminal_id_raw"] = hx(terminal_raw)
        result["terminal_id_bcd"] = bcd_to_str(terminal_raw)

    serial = struct.unpack("!H", data[p:p+2])[0]; p += 2
    result["serial"] = serial

    if subpackage:
        if len(data) < p + 4:
            result["error"] = "subpackage header too short"
            return result
        total = struct.unpack("!H", data[p:p+2])[0]
        seq = struct.unpack("!H", data[p+2:p+4])[0]
        p += 4
        result["package_total"] = total
        result["package_index"] = seq

    body = data[p:]
    result["body_len_actual"] = len(body)
    if len(body) != body_len:
        result["body_len_warning"] = "declared %d, actual %d" % (body_len, len(body))

    result["body"] = parse_body(msg_id, body, versioned=versioned)
    return result

def collect_frames(path, force_raw=False):
    sources = get_sources(path, force_raw=force_raw)
    seen = set()
    parsed = []
    idx = 1
    for src in sources:
        frames = extract_frames_from_bytes(src["payload"])
        for frame in frames:
            # De-duplicate because raw_scan will usually rediscover frames from parsed streams.
            fingerprint = frame
            if fingerprint in seen:
                continue
            seen.add(fingerprint)
            item = parse_jt808_frame(frame, source_desc=src["desc"], index=idx)
            parsed.append(item)
            idx += 1
    return parsed

def print_human(items, only_msg_id=None, show_raw=False):
    for it in items:
        if only_msg_id:
            want = int(only_msg_id, 16) if isinstance(only_msg_id, str) else int(only_msg_id)
            got = int(it.get("msg_id", "0xFFFF"), 16) if "msg_id" in it else None
            if got != want:
                continue

        print("=" * 100)
        print("#%s  %s  %s" % (it.get("index"), it.get("msg_id", ""), it.get("msg_name", "")))
        print("source      :", it.get("source"))
        print("terminal    :", it.get("terminal_id_bcd", it.get("terminal_id_raw", "")))
        print("serial      :", it.get("serial"))
        print("checksum    :", "%s calc=%s ok=%s" % (it.get("checksum"), it.get("checksum_calc"), it.get("checksum_ok")))
        print("versioned   :", it.get("versioned_2019"), "version=", it.get("protocol_version", "legacy"))
        print("body length :", it.get("body_len_declared"), "actual=", it.get("body_len_actual"))
        if "body_len_warning" in it:
            print("warning     :", it["body_len_warning"])

        body = it.get("body", {})
        print("body parsed :")
        print(json.dumps(body, ensure_ascii=False, indent=2))

        if show_raw:
            print("frame hex   :", it.get("frame_hex"))
            print("unescaped   :", it.get("unescaped_hex"))

def main():
    ap = argparse.ArgumentParser(description="JT/T 808 PCAP/stream parser")
    ap.add_argument("input", help="pcap/pcapng/raw file")
    ap.add_argument("-m", "--msg-id", help="only show this message ID, e.g. 0702 or 0x0702")
    ap.add_argument("-j", "--json", dest="json_out", help="write parsed result to JSON")
    ap.add_argument("--show-raw", action="store_true", help="print raw frame hex")
    ap.add_argument("--force-raw", action="store_true", help="skip packet parser and scan whole file only")
    args = ap.parse_args()

    only = args.msg_id
    if only and not only.lower().startswith("0x"):
        only = "0x" + only

    items = collect_frames(args.input, force_raw=args.force_raw)

    if args.json_out:
        with open(args.json_out, "w", encoding="utf-8") as f:
            json.dump(items, f, ensure_ascii=False, indent=2)
        print("[+] JSON saved to", args.json_out)

    print("[+] JT/T 808 candidate frames:", len(items))
    print_human(items, only_msg_id=only, show_raw=args.show_raw)

if __name__ == "__main__":
    main()

实战

司机的身份（JT/T808 车载终端通信协议）

打开流量包可以发现大量以 7e 开头、7e 结尾的 JT/T 808 帧

在 No.16 流量中，其中关键报文头

07 02 40 eb 01 00 00 00 00 01 77 70 64 12 11 00 72

解析

消息 ID：0x0702
消息体属性：0x40eb
协议版本：0x01
终端号：00000000017770641211
流水号：0x0072
消息体长度：0x00eb = 235

0x0702 正好是驾驶员身份信息采集上报，消息体开头为

01                  状态：IC 卡插入
20 06 27 09 48 56   时间：2020-06-27 09:48:56
00                  IC 卡读取成功
b4                  驾驶员姓名长度：0xb4 = 180 字节

取出这 180 字节发现是 UTF-16BE。每两个字节对应一个汉字

8a f8 -> 諸
96 b8 -> 隸
50 e7 -> 僧
96 4d -> 降
54 3d -> 吽

完整取出来得到

諸隸僧降吽諸陀摩隸僧缽薩願皤耨皤眾劫阿我皤慧阿亦皤我哆兜降嚤祗愍皤迦闍眾皤陀亦聞宣咤迦劫薩囉愍聞須皤塞迦皤諦亦劫亦皤闍願皤咤修闍薩嚤宣皤我菩菩皤迦聞諦婆亦聞皤阿薩所願如空如如囑皤囑

用 “新约佛论禅 / 新佛曰” 解密，得到一串 Base64

TVpXR0NaMzNJTlpIUzREWkdBWVhHUlRWSlo2UT09PT0=

连续 Base 解码得到 flag{Crypy01sFuN}