简介:面向CAN总线安全研究者和深度学习异常检测入门者,这里提供一套基于LogBERT的CAN总线异常检测完整实现,核心解决车载网络中海量日志的语义建模与攻击识别问题,适用场景包括spoofing、ddos、fuzzying等常见入侵形式的检测。包内含116个文件,以49个Python源码为主,辅以CSV数据集、pkl/pt模型权重和XML配置,整体约130.67MB,7z格式打包,目录分层清晰,便于对照论文源码逐模块学习。已有1214人学习下载。内容包含LogBERT论文原文与源码,以及专门适配CAN数据的改造版本,改造后的算法在Car-hacking数据集上准确率与召回率均超过99%,当前实现聚焦CAN ID维度。除了可复现的高精度检测结果,还完整呈现训练调参过程,涉及BERT架构实现、日志检测流程、CAN数据集清洗与预处理等关键环节,适合希望深入理解Transformer在工控/车联网日志分析中落地的开发者。
1. 用 logbert 做 CAN 总线异常检测:把报文当“日志”用,到底行不行
处理 CAN 总线日志时,最折磨人的问题是:总线明明没报故障帧,设备行为却已经乱套了。CRC 错、DTC 这一类机制只能发现电气层和协议层的损坏,而业务上真正要抓的“语义级异常”——某个信号值不该出现在这个 ID 里、某条周期报文从 10ms 突然变成 1000ms——往往要人对着日志逐帧看。logbert 是工业异常检测里做日志序列建模很成熟的模型,思路是把每一帧 CAN 报文当成一行日志、把连续 N 帧当成一条序列去做掩码预测,不用手写规则就能报出“这段报文的出现方式不符合历史规律”。下面按落地路径拆:先讲它凭啥能读 CAN,再给可在普通 GPU 或 CPU 机器上跑通的复现步骤,最后把验证阈值和几个必踩的坑说透。
2. 为什么 logbert 能读 CAN:掩码语义与报文解析
2.1 logbert 到底在“读”什么:日志序列与掩码预测
logbert 这个名字来自 Log + BERT,核心思想是借 BERT 的掩码语言模型能力对日志序列建模。训练时把一段日志里的 15% token 随机遮挡,让模型根据上下文预测被遮住的词;训练完成后,模型内化了“正常日志里什么 token 会出现在什么位置以及它后面经常跟什么”。做异常检测时,不再做分类,而是把一段待检测日志整段送进去,算整段序列的预测概率分:分高说明符合历史规律,分低说明这段日志的行为模型没见过。用这个思路来处理 CAN,前提是 CAN 报文虽然不是自然语言,但它的结构稳定性比自然语言还高——正常工况下每个 ID 的出现顺序、数据字节的取值区间、周期报文的间隔,都有强规律。只要预处理时保持字段位置稳定,掩码预测就能学到这些不变量。
这里要澄清一个常见误解:logbert 不是拿来判断“CAN 通信断开”这类物理层故障的。总线断开、终端电阻缺失、波特率不匹配,这些在日志里表现为整段静默或大量错误帧,用 canstats 这类统计工具几分钟就能定位,不需要上模型。logbert 的价值在于你不知道什么算异常的场景:比如售后日志里某条事件报文在特定工况下多发了三次,或者一个温度信号在车速高于 80km/h 时不该出现跳变。这种规则你写不完,但模型能从数据里学出来。
2.2 CAN 报文解析成文本:十六进制不能直接丢给 tokenizer
在 Linux 下用 candump 采集的原始日志长这样:
(1591234567.123456) can0 1A0#1122334455667788第一段是时间戳,中间是接口名,第三段是CANID#DATA,DATA 是十六进制字符串,最长 16 个字符对应 8 字节。CAN 报文解析的第一步不是把它原样塞给分词器,而是把字段拆开,保持每个字节的独立位置。我见过不少初做这个方向的把整行1122334455667788当一个 token 丢进去,结果模型只能记住“这串字符串出现过”,完全学不到字节级的变化规律。正确做法是按下述脚本处理:
import re CAN_LINE = re.compile( r'^\((?P<ts>\d+\.\d+)\)\s+(?P<iface>\S+)\s+' r'(?P<cid>[0-9A-Fa-f]{1,8})\#(?P<data>[0-9A-Fa-f]*)$' ) def can_line_to_text(line: str, dlc_len: int = 8) -> str: m = CAN_LINE.match(line.strip()) if not m: return None cid = m.group('cid').upper() data = m.group('data').upper() # 按 DLC 补齐/截断:DLC 是报文真实数据长度,不能盲目 ljust if len(data) >= dlc_len * 2: data = data[:dlc_len * 2] else: data = data.ljust(dlc_len * 2, '0') parts = [f"ID_{cid}"] parts += [f"B{i}_{data[i:i+2]}" for i in range(0, dlc_len * 2, 2)] return " ".join(parts) print(can_line_to_text("(1591234567.123456) can0 1A0#1122334455667788")) # 输出: ID_1A0 B0_11 B1_22 B2_33 B3_44 B4_55 B5_66 B6_77 B7_88这个脚本有两点值得说明。第一,正则里[0-9A-Fa-f]{1,8}兼容标准帧和扩展帧 ID,因为扩展帧可能是 8 位十六进制,你在整车日志里经常两类混着;如果只写{3},扩展帧日志会在解析阶段被静默丢弃,这是预处理中最隐蔽的数据丢失。第二,按B0_xx、B1_xx的格式拆字节,而不是直接输出十六进制串,是为了让 BERT 分词器稳定地切分:每个数据字节成为独立 token,掩码时遮掉一个 token 等价于遮掉一个字节,模型学到的就是“某个信号字节在特定 ID 中的取值规律”。这里我按标准 8 字节 DLC 处理,DLC 短于 8 时补零,如果你的总线存在 DLC 不固定的情况,建议把 DLC 也作为独立 token 写进序列,否则模型会把补零误学成常态。
2.3 序列化与窗口:一条序列放多少帧
CAN 不是均匀数据流。总线上周期报文和事件报文混在一起,总线空闲时可能几十毫秒没有帧,事件突发时一毫秒内挤进几十帧。logbert 的输入是定长序列,所以不能简单按时间窗切。
常见的做法是固定帧数窗口,比如每 32 帧切一条窗口,窗口之间重叠 50%。但这里有个坑:如果总线静默期很长,32 帧可能横跨几秒,把完全不相关的报文拼在同一个上下文里。我一般会做一个保护逻辑——同一窗口内首尾帧时间差超过 5 秒就强制断开,不足 32 帧的尾部也独立成窗,不丢弃。这样模型学到的是“局部上下文”,而不是“一整天数据的平均规律”。
另一个关键点是时间戳。logbert 本身不擅长直接吃浮点时间戳,但 CAN 异常里有相当一部分是时序异常:周期报文变慢、事件报文延迟、某条低优先级报文因仲裁失败被反复推迟。直接把时间戳丢掉会丢掉这批信息。我的做法是把相邻帧间隔离散化成DT_10MS、DT_100MS这样的 token,插在帧与帧之间。标准周期为 10ms 的报文,连续几帧间隔会稳定在 10ms 附近;一旦变成 1000ms,这个DT_1000MStoken 出现,模型立即会被触发。这比单纯对数据字段建模要敏感得多。
3. 数据准备与训练:从原始日志到 logbert 的最小复现
3.1 数据采集:candump 与干净数据区的划定
训练 logbert 只需要正常日志,不需要预先标注异常——这正是它相比规则引擎的优势。但“正常日志”的定义要仔细。我建议至少采集 24 小时数据,覆盖启动、怠速、行驶、休眠等阶段;只采高速巡航的 1 小时数据,模型学到的是“这个工况下的正常”,换个工况全是异常,没法用。
采集命令很简单:
# 在 Linux 下用 canutils 采集,-L 表示带本地时间戳 candump -L can0 > can_normal.log但这条命令有个隐性陷阱:candump 默认会把错误帧也打印出来,而 CAN 总线在强电磁干扰环境下出现零星错误帧是常态。如果把这些错误帧混进训练集,模型会把“偶发错误帧”学成正常行为,部署时真正的大规模异常反而报不出来。我通常在采集后先做一轮过滤,只保留数据帧,同时检查日志里是否有大片超时空洞——如果某段日志静默超过 10 秒,多半是采集端掉线或总线休眠,需要单独标记,不能直接进训练集。
3.2 预处理脚本:把文本转成模型输入
拿到原始日志后,用上一章的can_line_to_text逐行转换,再按 32 帧一组切窗口。下面的代码展示了从日志行到模型输入的最小闭环:
from transformers import BertTokenizer from torch.utils.data import Dataset tokenizer = BertTokenizer.from_pretrained("bert-base-uncased") class CanLogDataset(Dataset): def __init__(self, lines, seq_len=32): self.samples = [] for i in range(0, len(lines) - seq_len + 1, seq_len // 2): # 窗口滑动步长取 seq_len//2,保持 50% 重叠 block = lines[i: i + seq_len] text = " [SEP] ".join(block) tokens = tokenizer.tokenize(text) if len(tokens) > 500: # 留 12 个位置给特殊 token tokens = tokens[:500] self.samples.append(tokens) def __len__(self): return len(self.samples) def __getitem__(self, idx): return tokenizer.convert_tokens_to_ids(self.samples[idx])注意代码里窗口滑动的写法:range(0, len(lines) - seq_len + 1, seq_len // 2)。这里步长取 16,如果一条异常只持续 5 帧,它至少会完整落入两个窗口,不至于因为窗口边界被截断而漏报;如果步长等于窗口长度,异常刚好夹在两条窗口之间时,检测分数会被稀释得很厉害。这是序列异常检测里很容易忽略的参数。
tokenizer我习惯用bert-base-uncased,它会把 ID 和字节 token 切成更小的片段。这里有个取舍:如果数据量很大且总线报文类型固定,可以自己训练一个专用于 CAN 日志的 tokenizer,词表控制在几千以内,训练效率会高不少;但样本量在十万行以下时,直接复用预训练词表搭配 minilbert 更稳,因为模型初始化时就带了语言先验,收敛快很多。
3.3 训练配置:一张普通显卡能跑的参数表
CAN 日志的 token 量不大,每帧约 7 个 token,32 帧窗口约 224 个 token,不需要上完整 BERT。工业界常见做法是换用 minilbert 这类轻量变体,参数量小一个数量级,在 CPU 上做推理都能接受。训练超参建议如下:
| 参数 | 建议值 | 说明 |
|---|---|---|
| 模型 | minilbert | 轻量,适合小样本和部署 |
| seq_len | 32 | 窗口内的 CAN 帧数,不是 token 数 |
| 最长 token 数 | 512 | 超过直接截断,避免 OOM |
| batch_size | 16 | 8G 显存可训练 |
| epochs | 2 | 超过 2 轮会把噪声也学进去 |
| learning_rate | 3e-5 | 大于 1e-4 时 loss 容易飞 |
| warmup_ratio | 0.1 | 前 10% 步数线性热身 |
| mask_prob | 0.15 | 官方默认值,CAN 场景调到 0.2 也行 |
epochs 是这里最值得强调的参数。普通 NLP 任务训练 BERT 动辄几轮甚至几十轮,但 logbert 做异常检测时训练集全来自正常数据,数据分布窄、规律强,第一轮 loss 就能降到很低;第二轮继续压 loss,模型开始记住“正常样本里的个体细节”,部署时稍微一点环境变化(比如换了台车、换了总线负载)就会误报。我在自己的数据上测试,2 轮以后每多跑一轮,验证集正常样本的分数方差都会明显变大,这就是过拟合的信号。
训练代码可以用 transformers 的 Trainer 直接搭,省去手写训练循环:
from transformers import (BertForMaskedLM, DataCollatorForLanguageModeling, TrainingArguments, Trainer) model = BertForMaskedLM.from_pretrained("minilbert") data_collator = DataCollatorForLanguageModeling( tokenizer=tokenizer, mlm=True, mlm_probability=0.15, ) training_args = TrainingArguments( output_dir="./can_logbert", num_train_epochs=2, per_device_train_batch_size=16, learning_rate=3e-5, warmup_ratio=0.1, save_strategy="epoch", ) trainer = Trainer( model=model, args=training_args, train_dataset=train_dataset, data_collator=data_collator, ) trainer.train()DataCollatorForLanguageModeling会在每个 batch 动态 mask 15% 的 token,而不是预处理时静态 mask。这很重要:动态 mask 等于每个 epoch 模型见到的遮挡方式都不同,能有效缓解小样本下的过拟合。训练完记得保存tokenizer和model,推理阶段还要继续用。
3.4 logbert 的输出定位:打分器,不是分类器
训练完成后,模型的输出不是“正常/异常”二分类,而是每个 token 的预测概率。做检测时,我们通常把一段窗口内所有 token 的预测交叉熵相加,得到这段序列的异常分。分数越低越异常,因为模型对这段内容“没把握”。
这个定位决定了你在工程上的使用方式:不要试图让 logbert 直接输出一个标签,而是要把它当打分模块嵌到自己的检测链路里。后面的阈值标定、趋势判断、根因定位都基于这个分数展开,这一层想清楚,后面所有步骤才立得住。
4. 验证与排障:logbert CAN 检测的常见问题避坑清单
4.1 阈值怎么取:分数分布比绝对分数更可靠
logbert 打完分之后,第一件事不是拍脑袋定阈值,而是看正常窗口的分数分布。把验证集里所有正常窗口的异常分画成直方图,你会看到一个明显的主峰;然后往里混入几条人工构造的异常窗口(比如把某个周期报文的间隔改成 100 倍、把某个信号字节置成固定值),异常分会落在主峰左侧很远的地方。
常见做法是取正常分数分布的 0.1% 分位数作为报警阈值,而不是用平均值减几倍标准差。原因是 CAN 日志的分数分布通常是重尾的,均值对极值敏感;分位数对分布形态不敏感,而且可以直接对应到误报率——0.1% 分位意味着正常情况下平均一千个窗口会误报一个。
实际落地时阈值还是要现场微调。“先取分位数,再根据一周误报率回调”是我这几年的惯例,阈值选取本身有点玄学,但至少分位数法是可复现、可解释的起点,不会因为换了一批日志就完全失效。
4.2 避坑一:训练集里混入故障样本,模型学会了“故障”
现象:训练 loss 很低,验证集分数也正常,部署后异常窗口全部漏报。
原因:很多人做异常检测时,习惯把已经标出来的故障日志也“顺便”加进训练集,觉得数据越多越好。结果模型见过故障模式,把这些模式也学成了“正常”,推理时当然报不出来。
解决:训练集只用干净时间段的日志。怎么界定干净?先看采集日志里有没有 CAN 控制器报错记录,再按时间分段,剔除任何包含错误帧或静默空洞的区段。宁可训练数据减半,也不能混入异常。这一步没有捷径,是 logbert 方案里最不能省的前置工作。
4.3 避坑二:字节直接拼接导致分词器词表爆炸
现象:训练时 tokenizer 词表飞速增长,batch 打完一个就 OOM。
原因:预处理时没有按字节拆 token,而是把长度不等的十六进制数据串直接留给分词器切分。BERT 分词器对连续十六进制字符串的切分极不稳定,同一个0xAA在不同位置可能被切成a、aa、0xaa等多种形式,词表迅速膨胀,训练数据根本喂不饱统计量。
解决:用上一章can_line_to_text的固定格式ID_XXX加B0_xx到B7_xx。每个字节位置固定,同一个字节值永远映射到同一个 token,词表规模完全可控。
4.4 避坑三:周期报文太稳定,掩盖了事件报文
现象:模型训练两轮后 loss 降到 0.05 以下,看起来特别好,但真实故障一个都测不出来。
原因:CAN 总线上周期报文占比往往超过 90%,比如发动机转速报文每 10ms 一帧,数据值在正常工况下几乎不变。模型只需要记住“这个报文的这些字节永远是这几个值”就能把 loss 压得极低,事件报文的上下文规律反而没学到。
解决:预处理时把“恒定不变”的周期报文单独拎出来做规则校验,不参与 logbert 训练;对周期性变化的数据位做差分后再进模型。另外,事件报文往往因为仲裁优先级低而被周期报文挤占,出现时间抖动,这正是异常检测要抓的线索,不能因为难学就丢掉。更极端的做法是把 32 帧窗口内的周期报文压缩成一条聚合记录,只保留变化点的 token,虽然损失了部分时序信息,但能显著提升对事件报文的敏感度。
4.5 避坑四:窗口长度选错导致误报和漏报同时存在
现象:窗口设成 4 帧时几乎每个窗口都在报警;窗口设成 256 帧时,真正故障出现的那几秒分数被大量正常帧稀释,完全看不出来。
原因:窗口越小,上下文信息越少,正常波动就会被放大成异常;窗口越大,单个异常的权重被平均掉,检测灵敏度下降。
解决:线下训练时用 32 或 64 帧,配合 50% 重叠滑动,已经能覆盖绝大多数场景。线上实时监控可以双窗口并行:32 帧窗口负责灵敏报警,128 帧窗口负责趋势确认。两个窗口同时报警才触发通知,能把偶发误报压下去,同时不牺牲响应速度。
5. 进阶:把 logbert 得分接到根因定位与趋势图上
5.1 从分数到字段:逐 token 预测概率定位异常位
整窗分数只能告诉你“这段报文不对劲”,定位还得靠逐 token 的预测概率。推理时保留每个 token 的 logits,计算真实 token 的预测概率,概率最低的那几个 token 就是模型认为“最不该出现”的内容。这个信息可以直接还原成具体的 CAN 字段——是 ID 变了,还是第 3 个数据字节跳变,一目了然。但要注意:不能直接用 attention 权重做根因分析,logbert 的 attention 是黑匣子,看起来的热区经常和真正异常无关,我用过几次都翻车了,老老实实看 token 概率最可靠。
5.2 用趋势图做异常确认:单点分数不可靠
单窗口分数波动很大,直接拿来做实时报警会频繁误触。我习惯把连续 50 个窗口的异常分画成一条曲线,正常时段是一条平稳线,故障出现时曲线会形成一个明显的断崖或尖刺。这种趋势图异常检测比单个阈值可靠得多——如果分数只是瞬时掉一下马上恢复,多半是总线瞬时扰动;如果分数连续跌破阈值且持续回升不了,才是真正需要处理的问题。搭配 5.1 的字段定位,一个完整的检测链路就出来了:趋势图触发报警,token 概率定位字段,最后人工复核那几帧日志。
5.3 我的习惯与留待扩展
这套方案我落地过不止一次,最深的教训是:上线前一定要准备一份“已知异常”的测试集,哪怕只有几十条窗口,也要验证模型在已知故障上能报出来。没有这一步,阈值调得再漂亮都是自我安慰。另外,CAN 报文格式会随软件版本变化,新版本 ECU 上线后正常分布可能整体偏移,模型需要定期用新日志微调,不是训一次就能用三年。如果后续要覆盖更多场景,可以往两个方向扩展:一是把 DBC 文件里定义的信号名映射进 token,让模型直接学习物理信号级别的语义;二是把 logbert 和传统的统计过程控制图结合,让模型负责语义、统计图负责趋势,两者互补。希望帮到你。
本文还有配套的精品资源,点击获取