
简介面向车辆网络安全与深度学习开发者压缩包提供基于BERT的CAN总线异常检测实现将LogBert算法迁移到车载数据上聚焦CAN ID识别可学习BERT建模、日志检测、CAN数据集处理与完整调参流程。包内共116个文件以Python源码与pyc编译文件为主含7个CSV攻击数据集、pkl/pt/pth模型权重、XML配置和TXT说明整体130.67MB分类清晰便于检索。目前已有1214人学习适合有一定深度学习基础、想复现车载入侵检测实验的读者。资源附LogBert原论文和修改后源码代码、论文、数据相互对照可帮助理解如何在spoofing、DDOS、fuzzying三种攻击数据上取得99%以上准确率与召回率并快速训练验证。1. CAN总线异常检测用LogBERT把BERT用在报文序列上先学正常再抓偏离把BERT用在CAN总线异常检测上是我最近半年在车载日志分析里试出来最稳的方向。传统规则检测盯信号阈值、数ID出现频次遇到注入攻击和周期性异常常常后知后觉基于深度学习的LogBERT方案把CAN报文当作日志序列建模让模型先学一套正常通信分布再拿偏离程度当作异常得分。这篇笔记写给做车载测试、ECU诊断和日志分析的从业者适合从规则检测切到深度学习方案的团队。我会按数据清洗、模型训练、推理阈值、落地踩坑的顺序展开新手能照步骤复现熟手可以直接看参数和边界。2. 数据准备与日志解析把导出报文清洗成LogBERT能吃的训练语料2.1 .asc格式解析与字段抽取先说日志来源。整车测试或台架试验里最常见的是总线分析仪导出的.asc文件另外还有.blf、.dbc等配套格式但.asc最通用、最好处理。一条典型的报文长这样0.032002 1 123 Rx d8 11 22 33 44 55 66 77 88 0.032120 1 456 Rx d8 01 02 03 04 05 06 07 08第一列是时间戳单位秒第二列是通道号第三列是CAN ID十六进制第四列是收发方向d8表示DLC是8字节后面8个字节是数据区。注意有些日志里的ID是扩展帧会写成8位十六进制解析正则需要兼容。另外偶尔会有带错误帧标记的行比如带E的flag解析时可以直接跳过。import re def parse_asc(path): 把 .asc 格式 CAN 日志解析成结构化记录列表 pattern re.compile( r^\s*(\d\.\d)\s(\d)\s([0-9A-Fa-f]{2,8})\s r(Rx|Tx)\sd(\d)\s((?:[0-9A-Fa-f]{2}\s*){1,8}) ) records [] with open(path, r, encodingutf-8, errorsignore) as f: for line in f: m pattern.match(line) if not m: continue ts float(m.group(1)) ch int(m.group(2)) can_id int(m.group(3), 16) direction m.group(4) dlc int(m.group(5)) data_hex m.group(6).split() data [int(b, 16) for b in data_hex[:dlc]] records.append({ ts: ts, ch: ch, can_id: can_id, dlc: dlc, data: data, }) return records正则里把ID的十六进制位数放宽到{2,8}是为了同时兼容标准帧和扩展帧d(\d)匹配DLC后面的{1,8}匹配数据字节。errorsignore 是为了跳过日志里偶尔出现的乱码行——实测很多工具导出时会在文件头尾混进ASCII说明直接忽略比报错强。解析完先打印 records 的长度和中间几条记录核对字段别直接往下走这一步错了后面全错。数据量大的时候可以顺便统计每秒报文条数正常CAN总线负载下每秒几百到几千条都有这个数值后面定滑窗会用到。2.2 Token设计CAN ID做主键数据字节做量化日志解析出来后最朴素的做法是把CAN ID直接当成token一条报文一个token。这个思路能抓到「本该出现的ID没出现」和「不存在的ID出现了」这两类异常但对「ID正常、数据内容异常」的注入攻击完全无效。所以我在实际项目里采用两层token设计。第一层是CAN ID映射成整数token。低频ID需要单独处理否则训练集中只出现一两次的ID会让词汇表膨胀、模型学不稳。常见做法是统计频率低于某个阈值的ID合并成一个UNK tokenfrom collections import Counter def build_id_vocab(records, min_freq20): 统计CAN ID频率低频ID并入UNK返回id-token映射 counter Counter(r[can_id] for r in records) id_to_token {} token_to_id {} idx 2 # 0保留给PAD1保留给MASK for cid, freq in counter.most_common(): if freq min_freq: continue id_to_token[cid] idx token_to_id[idx] cid idx 1 id_to_token[__UNK__] idx id_to_token[__PAD__] 0 id_to_token[__MASK__] 1 return id_to_token, token_to_idmin_freq 是个关键参数。设太高比如100会把很多低频率但确实正常的ID吞进UNK模型学不到它们的上下文设太低又会让词汇表过大。我一般先按 vocab_size 不超过3000 反推因为CAN总线上一个网段的ID通常在两三百到一千个之间3000是够用的上界。UNK映射的idx要单独存一份推理时遇到未见过的ID统一走UNK避免KeyError。第二层是数据字节。8个数据字节直接拼进token序列会让词汇表爆炸——256的8次方组合BERT根本学不动。我的做法是分字段处理能拿到信号定义表的把车速、转速这类连续信号做分箱量化比如每5 km/h一档拿不到信号表的就把数据区按字节切片做成辅助token。辅助token不参与主序列建模只作为ID token的增强特征输入到embedding层这样既保留了数据内容信息又不把序列无限拉长。字节级别的辅助特征还有一个额外好处它能把「ID正常但payload异常」的攻击暴露给下游判定逻辑后面推理章节会用到。2.3 滑窗切分与时间域严格划分CAN报文是流式的LogBERT需要的是固定长度的序列。常见做法是用滑窗一个窗口128个token步长64窗口之间重叠一半。这样切出来的序列数量足够训练又不会让模型只看单一时刻的局部通信模式def make_sequences(token_ids, window_size128, stride64): 把CAN ID token序列切成固定长度滑窗 seqs [] for i in range(0, len(token_ids) - window_size 1, stride): seqs.append(token_ids[i : i window_size]) return seqs窗口大小和CAN通信周期强相关。如果一条报文周期是10 ms128个token覆盖1.28秒能装下十来个不同ID的完整周期如果某些报文周期是100 ms128个token就只覆盖一个多周期模型很难学到长程依赖。我在项目里先把记录按ID分组算平均发送周期取最大的周期值乘以4作为窗口下限再凑到8的倍数。这个细节直接决定后面的误报率值得花十分钟算清楚。切分之后最容易忽略的是训练集和测试集的划分方式。很多人图省事把切好的序列随机shuffle再按比例分这在时间序列异常检测里是致命的——训练集里混入了测试时段的信息模型分数虚高上线就露馅。正确做法是按时间顺序切分前80%的报文参与滑窗并作为训练集后20%作为测试集中间留出至少一个窗口长度的缓冲带保证两边的序列没有任何重叠。注意任何形式的随机shuffle都不是合法的时间序列划分。CAN日志异常检测只有「时间先后」这一个维度先到先切。3. 模型结构与训练配置把BERT迁移到日志域的关键改动3.1 为什么用小型BERT而不是原版原版BERT是12层、768维、上亿参数为大规模文本预训练设计的。CAN日志场景通常只有几十万到几百万条报文数据量差了几个数量级直接套原版必然过拟合。我实测下来4层Transformer、hidden_size 256、4个注意力头的配置在CAN日志上已经有足够表达力训练时间也能控制在单卡几十分钟到两小时。这个结论背后的逻辑是日志模板数量少、序列模式规律性强不需要大模型硬扛。另一个关键改动是掩码策略。原版BERT在预训练时随机掩码15%的token对日志场景我会把掩码比例提到30%。原因在于日志序列的上下文相关性比自然语言更强——一条报文的出现往往紧跟上一帧报文掩码更多才能让模型真正学到序列依赖关系而不是靠局部平滑混过去。掩码位置只在ID token上做数据字节的辅助token不掩码避免模型把辅助token当作主线索来猜ID。选择小型BERT还有一个工程上的好处推理速度。异常检测要跑在线还是离线回放是个先决问题256维的4层模型在普通CPU上跑单个128长度窗口的推理只要几毫秒整条半小时日志回放也就十来分钟。如果一上来就上大模型推理延迟和显存占用会让整个方案在工程评审阶段就被打回。3.2 超参数配置参考我在这类项目里的基准配置如下可以直接当起点参数取值说明vocab_sizelen(id_to_token)由词汇表构建决定一般1000~3000hidden_size256日志域足够不追求大num_hidden_layers4原版12层太重4层够用num_attention_heads4hidden_size需要能被head数整除intermediate_size512FFN中间层取2倍hidden_sizemax_position_embeddings128与滑窗大小对齐不能小于窗口learning_rate5e-4小数据用小学习率配合warmuptrain_batch_size64显存不够就降到32epochs30早停标准看验证集lossmask_ratio0.3日志序列上下文强多掩码vocab_size不需要预留太多CAN ID数量是有限的预留500个空位足够。max_position_embeddings必须大于等于滑窗长度否则切出来的长序列直接IndexError。batch_size和learning_rate要联动调batch翻倍时学习率最好跟着翻倍不然收敛速度明显变慢。我习惯把训练日志的loss曲线保存下来重点观察前500步是否出现锯齿状震荡如果是先降学习率再看别的。3.3 训练脚本与损失含义数据侧需要先做一个Dataset包装类方便Trainer读取from torch.utils.data import Dataset import torch class CanSeqDataset(Dataset): 把滑窗序列包装成PyTorch Dataset def __init__(self, sequences): self.seqs sequences def __len__(self): return len(self.seqs) def __getitem__(self, idx): return {input_ids: torch.tensor(self.seqs[idx], dtypetorch.long)}然后是模型初始化和训练参数from transformers import BertConfig, BertForMaskedLM, Trainer, TrainingArguments config BertConfig( vocab_sizelen(id_to_token) 8, # 预留少量空位 hidden_size256, num_hidden_layers4, num_attention_heads4, intermediate_size512, max_position_embeddings128, pad_token_id0, ) model BertForMaskedLM(config) training_args TrainingArguments( output_dir./can_logbert, learning_rate5e-4, per_device_train_batch_size64, num_train_epochs30, warmup_ratio0.1, logging_steps100, save_steps2000, report_to[], ) trainer Trainer( modelmodel, argstraining_args, train_datasetCanSeqDataset(train_seqs), ) trainer.train()BertForMaskedLM的输出是每个mask位置在词表上的预测分布loss取交叉熵。对日志异常检测来说这个loss的含义很直观模型对正常序列的token预测越准loss越低掩码位置的预测概率就是后面做异常打分的基础。warmup_ratio设0.1让模型前10%的步数用小学习率稳定起步防止开局震荡导致loss不降。report_to[]关掉自动上报避免环境依赖报错。训练过程中我习惯每500步手动存一次中间检查点方便后续对比不同epoch的检测效果。另外要盯训练loss曲线如果loss降到0.01以下还继续跌大概率是过拟合了此时验证集异常分数分布会明显「瘦身」阈值变得异常敏感这点在后面章节详细展开。4. 推理与异常判定异常分数怎么算、阈值怎么选4.1 用掩码重建概率算异常分数模型训练好之后推理阶段的核心是给每段CAN报文流算一个可比较的异常分数。我把测试窗口同样切成长度128的序列用和训练一致的30%掩码比例随机掩掉一部分ID token让模型重建取所有mask位置交叉熵的平均值作为该窗口的异常分数import torch def random_mask(input_ids, mask_ratio0.3, mask_token_id1): 随机掩码非mask位置label置为-100不参与loss prob torch.rand(input_ids.shape, deviceinput_ids.device) mask (prob mask_ratio) (input_ids ! 0) labels input_ids.clone() labels[~mask] -100 # -100位置被忽略 masked_ids input_ids.clone() masked_ids[mask] mask_token_id return masked_ids, labels torch.no_grad() def score_window(model, input_ids, device): 计算一个窗口的平均重建交叉熵作为异常分数 model.eval() input_ids torch.tensor(input_ids, dtypetorch.long).unsqueeze(0).to(device) masked_ids, labels random_mask(input_ids) outputs model(masked_ids, labelslabels) return outputs.loss.item()labels里非mask位置设成-100是PyTorch交叉熵的标准做法表示这些位置不参与loss计算。之所以只算mask位置的loss是因为全部位置都算的话模型可以靠「复制上一帧的ID」这类捷径把loss压得很低异常信号反而不明显。random_mask每次调用都会重新随机同一个窗口测两次分数会略有波动这是正常的要稳定对比时把随机种子固定。分数计算还可以加一层把每个窗口每个ID token的重建概率单独存下来。这样不仅能报「哪个窗口异常」还能定位「哪个ID最可疑」。对下游排查来说这个信息比一个干巴巴的分数有用得多很多车载测试工程师拿到ID级别的排查线索半小时就能定位到具体ECU或者报文信号。4.2 阈值选取百分位数与业务成本异常分数算出来后阈值怎么定是个工程问题。最简单的方法是拿训练集所有窗口的分数画分布取99分位作为阈值。但这里有个坑CAN日志的分数分布往往不是高斯分布而是右偏的99分位的分数可能比中位数的两三倍还大。盲目用均值加三倍标准差会把大量正常窗口误判成异常。我的做法是先按分数排序画出分数-累积概率曲线找「曲率突变点」。正常通信的分数集中在低位突变点之后的分数往往对应真实异常事件或者测量噪声。没有明显突变点时退回到百分位法先取P99观察误报数量再根据业务容忍度下调到P99.5或P99.9。整车测试场景里宁可多报几个疑似异常让工程师复核也不建议把阈值调太松导致漏报注入攻击。import numpy as np def pick_threshold(scores, percentile99.0): 根据正常样本分数分布取阈值返回阈值和超限比例 scores np.array(scores) thr np.percentile(scores, percentile) exceed_ratio (scores thr).mean() return thr, exceed_ratio超限比例就是理论误报率。如果P99对应的超限比例明显大于1%说明分数分布不够集中优先回查数据清洗和滑窗参数不要急着调阈值。如果手里恰好有一小批标注过的异常样本哪怕只有几十条可以用网格搜索遍历百分位选F1最高的阈值这比纯经验调参靠谱得多也是我目前最推荐的做法。4.3 窗口分数平滑与事件判定单窗口的分数抖动很大直接拿它做判定会频繁触发告警。我一般在窗口分数上做指数移动平均再加一个「连续N个窗口超阈值才算告警」的确认逻辑。这样可以滤掉单窗口噪声但对持续时间短的攻击会漏掉需要权衡。对CAN总线来说车载攻击一般持续至少几百毫秒到数秒确认窗口设3到5个比较合适。def ema_smooth(scores, alpha0.3): 对窗口分数做指数移动平均降低单点抖动 smoothed [] prev scores[0] for s in scores: prev alpha * s (1 - alpha) * prev smoothed.append(prev) return smoothedalpha的取值决定了平滑力度。alpha越大越跟随原始曲线告警灵敏但容易抖alpha越小越平缓稳定但会迟滞。0.3是我在两个项目上的折中值你可以先用这个跑一遍再看告警频率是否可接受。确认告警之后再用4.1里存的ID级重建概率找出可疑ID按概率从低到高排序输出给排查工程师。这一套「窗口定事件、ID定位」的两级输出比单纯报「第37个窗口异常」有用得多。5. LogBERT落地避坑五个真实踩坑记录5.1 踩坑一词汇表爆炸导致训不动现象模型训练速度极慢显存占用持续升高训练几十分钟loss纹丝不动。原因早期版本把CAN ID和数据字节混合拼成token数据字节的组合空间太大词汇表冲到几十万embedding矩阵占满显存模型根本学不到有效模式。解决拆成两级token设计——ID走主token数据字节分箱量化后作为辅助特征。词汇表压到3000以内之后训练速度和loss都恢复正常。从那以后我每次构建词汇表都会打印一行token总数超过5000就停手检查。5.2 踩坑二滑窗长度与CAN周期不匹配现象验证集误报率高达15%且误报窗口集中在某几个ID上。原因滑窗长度128个token只覆盖0.3秒但总线上有周期100 ms的慢速报文一个窗口里只能看到它一两次模型对这个ID的上下文建模不充分稍微偏离就报异常。解决按「最长报文周期 × 4」反推窗口下限再把窗口从128加大到256。改完误报率从15%降到2%以内。滑窗不是越大越好太大的窗口会抹掉短时异常的时间分辨率但CAN场景里先保证覆盖完整周期更重要。5.3 踩坑三正常拥堵场景被误判为异常现象实车跑城市工况时车辆密集路段连续告警但回放日志确认是正常通信。原因训练用的台架日志车速恒定、负载稳定没覆盖拥堵时频繁启停、多个ECU同时唤醒的通信模式。模型没见过这种ID交错密度分数整体抬升越过阈值。解决训练集按场景混合把城市拥堵、高速巡航、停车上电三种工况的日志按6:2:2比例混入。场景覆盖比数据总量更重要宁可总量少一点也要保证每种工况都有样本。实车日志的采集要刻意覆盖几个完整驾驶循环而不是只在实验室里跑。5.4 踩坑四时间泄漏让效果虚高现象测试集AUC 0.98上线实测一塌糊涂正常数据疯狂告警。原因有一份数据用了随机shuffle再划分训练测试集测试序列和训练序列来自同一时间段模型等于见过答案离线指标全是水分。解决改成严格按时间切分前后各留一个窗口的缓冲带并且在项目规范里写明禁止对时间序列做随机划分。从那以后我每次做数据切分都会在代码里加一行断言max(train_timestamp) min(test_timestamp)。这一行救过我好几次。5.5 踩坑五低频ID被UNK吞掉注入攻击漏报现象用注入攻击样本做回归测试部分攻击ID完全没有触发告警。原因构建词汇表时min_freq设得偏高攻击ID在训练集里出现次数少被归进UNK。推理时攻击ID同样走UNK模型对UNK的预测概率天然不高不低异常分数区分度消失。解决min_freq从50降到10并单独维护一份合法ID白名单。推理时如果出现白名单之外的ID直接打高异常分不依赖模型判断。深度学习和规则叠加使用比单靠哪一边都稳。现在我的标准流程是模型分数做主判定ID白名单检查做强制补充两条线都过才算正常。6. 回归验证技巧注入攻击样本做自查模型上线前一定要做攻击注入回归测试。我自己维护一套脚本在正常日志的时间轴上随机插入伪造报文模拟三类典型攻击新ID注入、正常ID但数据字段越界、正常ID的发送频率突变。插入之后重新切窗评分统计检出率和误报率。import random def inject_attack(records, attack_id, attack_data, start_idx, count): 在正常报文序列中注入伪造CAN报文 injected [] for i in range(count): ts records[start_idx][ts] i * 0.01 injected.append({ ts: ts, ch: records[start_idx][ch], can_id: attack_id, dlc: len(attack_data), data: attack_data, }) return injected注入位置要覆盖日志的前中后三段不能都挤在开头注入密度用1%、5%、10%三档分别测试记录不同密度下的检出率。检出率低于80%时先回查是阈值问题还是特征问题阈值问题就调百分位特征问题就要回到数据清洗和token设计那一层。回归测试的脚本和攻击样本始终保持一份固定版本每次改模型都跑同一套保证前后效果可比。我记得做一个某跨平台系统的异常检测Demo时就是靠这个固定回归集发现新训练脚本的掩码逻辑写错了——旧脚本mask_token_id用的1新脚本误写成0PAD和MASK混淆导致分数全部失真。从那以后我每次改训练或推理代码都强制先跑一遍固定回归集再谈优化。希望这套流程能帮到你至少在被CAN日志的玄学折腾到怀疑人生的时候有一条能兜底的验证路径。本文还有配套的精品资源点击获取