简介面向汽车网络安全与智能网联的异常检测资源围绕LogBert这一基于注意力机制的日志表征方法为车载总线入侵识别提供完整可复现方案。内容包含理论原文、改造后的源代码以及针对韩国车载入侵数据集的训练与评估结果覆盖欺骗攻击、拒绝服务攻击、模糊攻击三种场景当前版本以控制器局域网标识符为检测目标在公开数据集上精确率与召回率均达到百分之九十九以上。压缩包共有一百一十六个文件涵盖Python源码四十九个、编译后文件三十四个、表格数据七个、说明文档五个、配置标记五个、模型权重与训练日志等总容量约一百三十兆字节。学习主线从注意力模型本身、日志异常分析流程到控制器局域网数据清洗、特征工程与调参技巧均配有实践文件可对照操作。已有超过一千二百人下载学习适合具备一定深度学习基础、希望快速迁移到车辆总线场景的研究者或安全测试人员。1. 用 BERT 做 CAN 总线异常检测logbert 在车辆数据场景下的落地价值做车载总线诊断的工程师多少都有过这种经历总线在某次试验中突然出现大量报文丢失、ECU 无响应用 CANalyzer 抓完日志面对几千万行报文却不知道从哪看起。传统做法是设阈值、查 DBC、对信号逐个做范围校验但在多 ECU 交互、信号间存在隐式依赖的情况下基于规则的方案几乎抓不住那些时序层面上的异常。logbert 这条路线的思路是把一段时间窗口内连续到达的 CAN 报文当作“日志序列”用 BERT 的 masked language model 建模报文之间的上下文依赖再用它检测偏离正常交互模式的片段。这套方法原本在系统日志异常检测上验证过迁移到 CAN 总线后对报文缺失、ID 时序错乱、信号值异常这几类问题尤其有效。适合的人群很明确正在做 CAN 总线诊断、车辆 OTA 数据回传分析、或想引入深度学习替代人工查报文的工程师。2. 原始报文到训练样本DBC 解析与序列化设计2.1 从 pcap / CSV 拿到干净的 CAN 报文logbert 本质上吃的是文本序列所以第一步是把总线上的原始报文转成带语义的 token 序列。绝大多数情况下原始数据来自 CANoe、周立功、PCAN 这些工具的导出文件常见格式是 pcap 或 CSV。CSV 比较简单关键是清洗。拿到的原始数据一般有三类脏数据错误帧、过载帧、总线空闲期间工具自动补的填充记录。import pandas as pd df pd.read_csv(can_log.csv, names[ts, can_id, flags, data]) # 去掉错误帧和过载帧flags 字段中 E(Error) 和 O(Overload) 的行直接丢弃 df df[~df[flags].str.contains(E|O, naFalse)] # 去掉远程帧 RTR远程帧没有数据场不能参与序列建模 df df[~df[flags].str.contains(R, naFalse)] # 按时间戳排序保证后面生成的序列是真实发生顺序 df df.sort_values(ts) df df.reset_index(dropTrue)这段代码里flags字段是工具导出的帧属性标记每个工具叫法不太一样但语义一致。清洗的核心原则是logbert 只学“正常通信”的样子错误帧和远程帧不属于稳定交互模式留在训练集里会把模型带偏。can_id保留的是标准帧或扩展帧的仲裁 ID注意统一大小写后面词表构建时不希望同一个 ID 出现两种写法。2.2 用 DBC 把报文体切成交互知识只拿 CAN ID 训练 logbert模型学到的是“哪个 ECU 在什么时间说话”的节奏规律。想让它感知信号异常就得把报文数据场按 DBC 拆成信号来用。DBC 文件里定义了每个报文 ID 下信号的名字、起始位、长度、缩放因子和偏移量。用 cantools 库解析后可以自动完成从十六进制数据场到物理值的转换。import cantools db cantools.database.load_file(vehicle.dbc) signal_records [] for _, row in df.iterrows(): try: message db.get_message_by_frame_id(row[can_id]) decoded message.decode(bytes.fromhex(row[data])) for sig_name, sig_value in decoded.items(): signal_records.append({ ts: row[ts], can_id: f{row[can_id]:03X}, signal: sig_name, value: sig_value }) except KeyError: # DBC 里没有定义的 ID保留原始 ID 作为 token signal_records.append({ ts: row[ts], can_id: f{row[can_id]:03X}, signal: UNKNOWN, value: row[data] })这一步容易踩坑的是decode对某些信号做了线性变换比如发动机转速是用0.25 rpm/bit缩放解析出来的物理值可能是 0.25 的整数倍。值域信息对 logbert 是有用的但连续浮点数不能直接当 token。常见做法是分桶离散化把物理值映射成SPEED_0_500、SPEED_500_1000这样的区间 token或者只保留信号的突变方向。我一般倾向于用区间分桶因为它能抓住信号“大致在哪个范围”这个语义又不会因为量化过细导致词表爆炸。2.3 序列化token 的粒度决定模型学什么logbert 接收的每条样本是一条“日志”在 CAN 场景下就是一条报文。但报文的 token 化方式直接决定模型能学到什么。两种方案对比下来差别很大只把 CAN ID 当 token模型学到的是总线上 ECU 的发言规律能检测报文缺失、ID 时序错乱把 ID 和数据场拆成信号 token 拼接模型能进一步感知信号值异常但序列长度会明显变长训练成本翻倍。实际项目中我一般先用 ID-only 方案跑一版看能不能检测出报文缺失类问题。如果业务上需要覆盖信号值突变再升级到 ID 信号 token 的拼接序列。下面是 ID-only 序列构建的参考实现import json # 按 100ms 窗口切分窗口内按时间排序组装成一条 logbert 样本 WINDOW_MS 100 samples [] current_window [] window_start_ts None for _, row in df.iterrows(): ts row[ts] * 1000.0 # 如果原始 ts 单位是秒转成毫秒 if window_start_ts is None: window_start_ts ts current_window [] if ts - window_start_ts WINDOW_MS: samples.append({ window_start: window_start_ts, sequence: [fCAN_{r[can_id]} for _, r in current_window] }) window_start_ts ts current_window [] current_window.append(row) # 只保留长度超过 5 的窗口太短的序列没有上下文可学 samples [s for s in samples if len(s[sequence]) 5] with open(train_samples.json, w) as f: for s in samples: f.write(json.dumps(s) \n)窗口大小是序列化最重要的超参数。100ms 在 500kbps 总线速率下大概能装下 30~50 条报文对多数乘用车动力 CAN 是合理的。商用车或车身 CAN 报文周期偏慢窗口可能要放大到 200~500ms。窗口太小模型看不到跨报文依赖窗口太大异常片段被正常报文稀释检测灵敏度下降。建议先用统计方法看目标总线上报文的平均到达间隔把窗口设成平均间隔的 5~10 倍。3. logbert 模型原理与训练流程3.1 masked language model 与两个分类头的设计logbert 的骨干是 BERT核心思想是在序列里随机 mask 一部分 token让模型根据上下文去预测被遮住的 token。预训练之后模型对“正常报文序列长什么样”形成了内部表征。推理时有两条判定路径一是序列分类头直接输出整条序列正常/异常的二分类概率二是 token 分类头逐 token 输出概率mask 掉一个 token 后看模型预测置信度。CAN 场景里这两个头各有用途。序列分类头适合检测整体行为偏移比如总线负载率异常波动、某 ECU 静默token 分类头适合定位具体是哪条报文、哪个信号异常。实际落地时我建议两个头都保留序列头做告警触发token 头输出异常定位线索方便后续人工确认。logbert 在原论文里用的是标准 BERT 结构加两个全连接头。CAN 场景下序列长度比系统日志短很多不需要很大的模型。隐藏层 128、transformer 层数 4~6、自注意力头 4 个这个量级足够捕获报文间依赖训练速度也在可接受范围。词表大小取决于 CAN ID 数量一般在 50~500 之间远小于 BERT 原始的 30k 词表。3.2 训练数据构建采样、mask 策略与训练集划分训练数据不能直接拿整段日志全部塞进去。必须要做的是负样本采样和正样本纯净。这里有个关键区别logbert 是无监督学习为主训练时不需要标注异常只需要“干净”的正常数据。问题在于车辆正常行驶过程中也会出现偶发的 CAN 错误帧、ECU 重启、总线短暂关闭这些不能混进训练集。一个可行的做法是先用规则粗筛一轮把包含错误帧标记的时段、DTC 上报集中时段、总线负载率超过 80% 的时段全部排除掉剩下的当作“正常基线”。规则过滤后再人工抽查几段确认没有隐藏的异常。数据量方面logbert 是预训练模型不需要海量数据。我实测过30 万条报文切出的窗口序列在 3 万条左右训练 5 个 epoch 就能收敛到可用的效果。mask 策略上CAN 序列比日志序列更短更规整mask 比例可以比 BERT 原版的 15% 略低10% 左右比较合适。太高会导致正常序列也被模型误判为异常因为模型没见过那么多被遮住的 token。3.3 训练脚本与参数说明下面给一份可以跑起来的训练脚本骨架基于公开的 logbert 实现思路把数据读入、mask 训练、模型保存串在一起。import torch import torch.nn as nn from torch.utils.data import DataLoader, Dataset class CANSequenceDataset(Dataset): def __init__(self, samples, vocab): self.samples samples self.vocab vocab # {token: id} def __len__(self): return len(self.samples) def __getitem__(self, idx): seq self.samples[idx][sequence] ids [self.vocab.get(t, self.vocab[UNK]) for t in seq] return torch.tensor(ids, dtypetorch.long) def mask_tokens(ids_tensor, mask_prob0.10): mask_token_id ids_tensor.new_ones(ids_tensor.shape) * vocab[MASK] rand torch.rand(ids_tensor.shape, deviceids_tensor.device) mask_idx rand mask_prob masked_ids ids_tensor.clone() masked_ids[mask_idx] mask_token_id[mask_idx] return masked_ids, ids_tensor, mask_idx train_dataset CANSequenceDataset(train_samples, vocab) train_loader DataLoader(train_dataset, batch_size64, shuffleTrue) model LogBERT(vocab_sizelen(vocab), hidden_size128, num_layers4, num_heads4) criterion nn.CrossEntropyLoss(ignore_index0) # ignore padding token optimizer torch.optim.Adam(model.parameters(), lr1e-4) for epoch in range(5): for batch in train_loader: ids_tensor batch # [batch, seq_len] masked_ids, original_ids, mask_idx mask_tokens(ids_tensor) logits model(masked_ids) # 只对 mask 位置计算 loss loss criterion(logits[mask_idx], original_ids[mask_idx]) optimizer.zero_grad() loss.backward() optimizer.step()mask_tokens函数是这段代码的核心它在每个训练 batch 里随机遮挡 10% 的 token。logits[mask_idx]的索引操作保证了只在被遮挡的位置反传梯度这与 BERT 的预训练目标一致。ignore_index0是为了忽略 padding token因为不同窗口的序列长度不一样做了 padding 后需要让模型不关注无效位置。学习率 1e-4 对微调稍微偏大如果观察到 loss 震荡降低到 3e-5 重新跑。4. 部署与推理滑动窗口与阈值判定4.1 推理路径与 anomaly score 的计算模型训练完部署时不需要整段离线扫描直接用滑动窗口在实时报文流上推理。推理的核心是把“序列是否异常”转成可比较的分数。logbert 的序列分类头输出的是一个 logit经过 sigmoid 后得到异常概率。但这个概率在不同车辆、不同总线负载下波动很大直接拿固定阈值容易误报。更稳的做法是同时利用 token 分类信息的重构置信度把当前窗口的每个 token 逐一遍历mask 掉后看模型预测的置信度置信度低的位置就是异常 token。综合所有位置的置信度得到当前窗口的 anomaly score。这样分数对总线噪声更鲁棒也方便定位异常报文。4.2 阈值怎么定百分位 业务校准阈值是 logbert 落地时最让人头疼的参数。我自己的经验是分两步走。第一步在一个月以上的正常数据上推理记录每个窗口的 anomaly score 分布取 95 或 99 百分位作为初始阈值第二步把收集到的真实故障片段回放一遍看模型在故障时刻的分数是否超过初始阈值根据漏报和误报情况做人工校准。import numpy as np def choose_threshold(scores_normal, percent99): return np.percentile(scores_normal, percent) # 假设已经收集了正常行驶两周的数据推理得到 scores_normal threshold choose_threshold(scores_normal, percent99) print(f初始阈值 {threshold:.4f}) # 回放历史故障数据看模型分数是否越过阈值 for fault_win_score in fault_window_scores: if fault_win_score threshold: print(故障窗口被检出) else: print(漏报需要调低阈值或检查窗口切分)这段代码里的关键不是那个percentile调用而是“先用正常数据定基线再用真实故障校准”的流程。直接调阈值百分位是玄学必须落到业务里可解释的异常事件上确认。比如某次总线静默事件如果模型没报出来说明窗口切得太短或者 ID 序列化丢失了关键信息不是调低阈值就能解决的。4.3 滑动窗口推理参考实现from collections import deque class SlidingCANProcessor: def __init__(self, model, vocab, window_size50, stride10): self.model model self.vocab vocab self.window_size window_size self.stride stride self.buffer deque(maxlenwindow_size) def process(self, can_record): # can_record: {ts: float, can_id: str, data: str} token fCAN_{can_record[can_id]} self.buffer.append(token) if len(self.buffer) self.window_size: ids_tensor torch.tensor( [self.vocab.get(t, self.vocab[UNK]) for t in self.buffer] ).unsqueeze(0) with torch.no_grad(): seq_prob, token_logits self.model(ids_tensor) anomaly_score 1.0 - seq_prob.item() self._slide() return anomaly_score return None def _slide(self): for _ in range(self.stride): if self.buffer: self.buffer.popleft()window_size50, stride10意味着每来 10 条新报文就推一次窗口相邻窗口有 40 条重叠。重叠窗口对检测突变类异常是有利的因为异常报文在任何一个窗口里被捕获就会被上报代价是计算量放大 5 倍。如果总线报文速率很高比如每条 10ms50 条窗口只有 500ms可以适当把窗口加大到 100 条减少推理频率。5. logbert 用在 CAN 的避坑记录五个典型翻车点5.1 模型把“报文缺失”学成了正常模式现象模型训练完在正常数据上分数很低但实际检测时总线某节点断电导致的报文缺失却没触发告警。原因训练数据里混入了一段时间总线负载较高、偶发丢包的日志模型在 mask token 预测时发现“某些 ID 不出现”也是合理的把缺失当成了上下文规律的一部分。解决训练集必须严格清洗按报文周期做完整性校验缺失率超过 1% 的时间段直接剔除。用 DBC 里的周期信息CYCLE_TIME属性可以自动化完成这件事如果某周期报文在预期时间点没出现标记该窗口为脏数据。5.2 ID 词表过大导致训练发散现象模型 loss 不下降或者训练结束后推理分数全部接近 0.5。原因有些总线接了诊断仪或测试设备产生大量一次性 CAN ID词表膨胀到几千甚至上万而每个 ID 的出现次数极少模型学不到它们的上下文。解决构建词表时统计每个 CAN ID 的出现频次低于总样本数 0.01% 的 ID 统一映射到UNK只保留高频 ID。同一辆车反复测试时诊断 ID 往往只出现少量次数这类信息没有建模价值。5.3 时间窗边界切碎了 ECU 交互现象实时推理时误报集中在整点附近比如每分钟的第 30 秒前后。原因某些 ECU 的周期报文是异步发送的固定时间窗会周期性地把一条完整的交互序列切到两个窗口里导致两个窗口都缺上下文。解决在日志数据里先做自相关分析找到报文的基频按基频的整数倍对齐窗口边界。更简单的方式是统计误报时间点如果集中在固定偏移把窗口起点偏移半个报文周期即可。这个问题的本质是窗口相位不对不是模型的问题。5.4 扩展帧与标准帧混用导致 ID 碰撞现象某些窗口异常分数极高检查报文发现只是标准帧和扩展帧交替出现ID 值恰好相同。原因标准帧和扩展帧的仲裁 ID 字段虽然长度不同但数值可能相同直接拼成CAN_123会混淆两类帧。扩展帧 ID 是 29 位标准帧是 11 位语义上不能混在同一个 token 空间里。解决序列化时加前缀区分标准帧用SID_123扩展帧用EID_1234567。如果总线上两种帧都有这个区分是必须的否则模型永远学不会它们的差异。类似地不同网段动力 CAN、车身 CAN、诊断 CAN混在一个日志文件里时也要加网段前缀。5.5 固定阈值在环境变化后失效现象模型在夏季标定数据上表现良好到了冬季测试场环境温度极低总线负载变化误报率飙升。原因阈值基于特定环境下的正常数据标定而 CAN 总线行为受温度、负载、ECU 软件版本影响很大同一个 ID 在不同时段的报文周期可能存在 5%~10% 的波动。解决不要用固定阈值改用滑动基线。维护一个长度为 N 的分数队列计算滚动均值与标准差超过均值加 3 倍标准差才告警。这样模型能自适应总线行为漂移代价是刚启动的一段时间内没有基线需要预热。6. 验证召回率回放故障数据与合成异常注入模型训练完只看训练 loss 收敛是不够的必须用真实故障数据验证。问题在于真实故障很难收集全。我的做法是两条腿走路一是把公司试验场积累的历史故障日志回放给模型看能不能在故障发生点附近检出二是做合成异常注入人为篡改正常日志模拟报文缺失、信号突变、ID 乱序这几类常见异常。合成注入的代码逻辑比较简单拿一段正常日志随机去掉若干个 ID 的报文或者把某个信号值把缩放因子故意调错再让模型推理看异常分数是否显著高于正常水平。这个方法能快速估算模型的召回率缺点是合成异常和真实异常之间有偏差所以它只能用来筛模型结构问题不能替代真实回放。网格化调参时我主要看两个指标检出率异常窗口被标记的比例和误报率正常窗口被标记的比例。检出率低于 90% 时优先检查窗口大小误报率高于 5% 时优先检查阈值标定和训练集纯净度。参数敏感性有个规律值得注意窗口大小的调整越敏感模型工作越不稳定这种情况下优先回退到更简单的 ID-only 序列而不是继续调参。尝到合成注入的甜头后我每次训练新模型的强制流程是先用合成异常跑通全链路确认检出率达标再回放真实故障片段做最后复核。这个习惯帮我挡掉了至少三次因为数据清洗不干净导致的模型失效。希望帮到你。本文还有配套的精品资源点击获取