JT/T 808 协议¶
概述¶
JT/T 808 是交通运输行业标准《道路运输车辆卫星定位系统 北斗兼容车载终端通讯协议》,规定了车载终端与监控平台之间的数据通信格式
协议基于 TCP 或 UDP,采用大端字节序,报文整体以 0x7E 为帧边界,内部需经过转义处理
报文封装格式¶
每个完整报文由标识位包裹,内部包含消息头、消息体和校验码
+--------+--------+----------+--------+--------+
| 标识位 | 消息头 | 消息体 | 校验码 | 标识位 |
| 0x7e | 变长 | 变长 | 1 byte | 0x7e |
+--------+--------+----------+--------+--------+
-
标识位:固定
0x7E,标识一帧的开始和结束 -
消息头:固定 12 字节(无分包)或 16 字节(有分包),包含消息 ID、属性、终端手机号、流水号等
-
消息体:长度由消息头中的 "消息体长度" 字段决定,内容取决于消息 ID
-
校验码:从消息头第一个字节到消息体最后一个字节,纯异或
消息头结构¶
消息头结构如下:
Offset: 0 2 4 10 12
+------+-------+--------+--------+
| 消息ID | 消息体属性 | 终端手机号 | 消息流水号 |
+------+-------+--------+--------+
u16 u16 u48 u16
消息 ID(2 字节,u16)¶
标识消息类型,如:
-
0x0001:终端通用应答 -
0x0100:终端注册 -
0x0200:位置信息汇报 -
0x8001:平台通用应答 (高位可用于区分终端/平台消息)
消息体属性(2 字节,u16)¶
JT/T 808—2019 中消息体属性按如下位域解析:
Bit: 15 14 13 12 10 9 0 +--------+--------+--------+------------+----------------+ | 保留 |版本标识 | 分包标志 | 数据加密方式 | 消息体长度 | +--------+--------+--------+------------+----------------+
-
bit 15:保留
-
bit 14:协议版本标识
-
0:未启用 2019 版版本标识机制,按旧版消息头格式解析;
-
1:启用版本标识机制,消息头中增加 1 字节协议版本号,
- 且终端手机号字段使用 BCD[10]
-
bit 13:分包标志
-
0:不分包;
-
1:长消息分包,消息头后增加“消息总包数 + 包序号”共 4 字节
-
bit 12~10:数据加密方式
-
000:消息体不加密;
-
bit10 = 1:RSA 加密;
-
其他值保留
-
bit 9~0:消息体长度,范围 0~1023 字节
终端手机号(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 流量中,其中关键报文头
解析
0x0702 正好是驾驶员身份信息采集上报,消息体开头为
取出这 180 字节发现是 UTF-16BE。每两个字节对应一个汉字
完整取出来得到
用 "新约佛论禅 / 新佛曰" 解密,得到一串 Base64
连续 Base 解码得到 flag{Crypy01sFuN}