1.2 数据投毒与模型窃取 本节摘要:数据投毒和模型窃取是两类"潜伏期长、发现即损失"的威胁。投毒指攻击者污染训练或微调数据,让模型学到恶意模式——可能是标签被悄悄翻转,也可能是植入一个只在特定触发器下才激活的后门。窃取则指攻击者通过反复查询你的模型 API,复制出功能等价的模型或反推出训练数据。本节讲数据溯源与异常检测两条投毒防线,以及速率限制、输出扰动、模型水印三种窃取防御,并说明它们各自的局限。
本节摘要:数据投毒和模型窃取是两类"潜伏期长、发现即损失"的威胁。投毒指攻击者污染训练或微调数据,让模型学到恶意模式——可能是标签被悄悄翻转,也可能是植入一个只在特定触发器下才激活的后门。窃取则指攻击者通过反复查询你的模型 API,复制出功能等价的模型或反推出训练数据。本节讲数据溯源与异常检测两条投毒防线,以及速率限制、输出扰动、模型水印三种窃取防御,并说明它们各自的局限。
阅读完本节,你应当能够:
先说投毒。现在几乎没有团队会从零标注全部训练数据,多半会混用公开数据集、爬取的网页、第三方供应商的数据、用户产生的反馈。每多一个数据来源,就多一个被下毒的入口。投毒最阴险的地方在于它不会立刻暴露——模型在干净输入上表现完全正常,准确率没掉、loss 没异常,只有在攻击者投喂的那个特定触发器出现时才作恶。等你发现的时候,模型可能已经上线好几个月了。
再说窃取。训练一个大模型动辄几十万到上百万的成本,而攻击者只要能调你的 API,就能通过成千上万次查询"蒸馏"出一个行为相似的小模型。2016 年微软的 Tay 机器人是个被反复引用的反例(虽然更偏向数据污染):它通过和 Twitter 用户交互学习,结果 16 小时内就被投喂有害内容带偏了。GPT-2 时代也有研究演示过,仅靠黑盒 API 查询就能复制出功能接近的模型。这意味着你花大价钱训练的模型资产,可能被别人用很低的查询成本"偷走"。
这两个威胁的共同点是:它们都发生在你看不见的地方。提示注入是输入层的明枪,投毒和窃取是数据和资产层的暗箭。
| 向量 | 做法 | 为什么难发现 |
|---|---|---|
| 标签翻转 | 把正样本的标签改成负的 | 单条看不出来,整体准确率缓慢下降 |
| 后门注入 | 在特定样本上加触发器图案并改标签 | 干净输入上完全正常,只有触发器出现才作恶 |
| 脏数据注入 | 灌入大量低质量或误导性样本 | 混在正常噪声里,统计上不显眼 |
| 模型水印移除 | 针对性清除模型里的所有权标记 | 让你事后无法证明模型是你的 |
后门注入是最棘手的。一个图像分类模型如果在训练时被喂了一小批"左下角带小黄块的样本且标签被改成猫",那么上线后只要任何输入带这个小黄块,就会被识别成猫。这个触发器可以小到人眼几乎注意不到,但足以让模型误判。
投毒检测的核心流程是"建立干净基线—对比新数据—标记异常":
一条路是数据溯源:给每个数据集算哈希、记来源、留时间戳,任何篡改都能通过哈希不一致发现。另一条路是统计异常检测:在可信数据上训练一个异常检测器(比如孤立森林),新数据如果特征分布偏离基线太多,就标记为可疑。
| 类型 | 攻击者要什么 | 所需查询量 |
|---|---|---|
| 功能提取 | 复制输入到输出的映射关系 | 数万到数百万次 |
| 架构推断 | 推断模型结构(层数、宽度、激活函数) | 数千次 |
| 训练数据提取 | 反推出训练时用过的具体样本 | 数千到数万次 |
| 水印窃取 | 提取并清除模型里的所有权标记 | 数百次 |
注意"训练数据提取"这一项——它不只是偷模型,还可能泄露训练数据里的隐私(比如记忆进去的个人信息),这就和隐私安全挂钩了。
最朴素但有效的办法是给每个数据集算一个哈希指纹,注册时存下来,训练前再算一次比对。任何中间环节的篡改都会让哈希对不上。
import hashlib from datetime import datetime class DataProvenanceTracker: def __init__(self): self.registry = {} def register(self, dataset_id, source, samples): h = self._hash(samples) self.registry[dataset_id] = { "source": source, "hash": h, "timestamp": datetime.utcnow().isoformat(), "count": len(samples), } return h def verify(self, dataset_id, current_samples): if dataset_id not in self.registry: return False return self._hash(current_samples) == self.registry[dataset_id]["hash"] @staticmethod def _hash(samples): # 排序保证确定性,同内容不同顺序哈希一致 return hashlib.sha256(str(sorted(samples)).encode()).hexdigest()
💡 关键直觉:溯源不告诉你"数据有没有毒",它只告诉你"数据有没有被动过"。但"被动过"本身就是个强信号——一个本该冻结的数据集如果哈希变了,无论内容看着多正常,都不该直接拿去训练。
哈希只能发现整体被替换,发现不了"在真实数据里掺了几条毒样本"。这时需要统计层面的异常检测——在可信数据上训练一个孤立森林,新数据偏离越远越可疑。
import numpy as np from sklearn.ensemble import IsolationForest class PoisoningDetector: def __init__(self, contamination=0.05): self.model = IsolationForest( contamination=contamination, random_state=42) self.fitted = False def fit(self, clean_X): self.model.fit(clean_X) self.fitted = True def detect(self, new_X): if not self.fitted: raise ValueError("先在可信数据上拟合") preds = self.model.predict(new_X) # -1 异常, 1 正常 idx = np.where(preds == -1)[0] return { "total": len(new_X), "poisoned_count": len(idx), "poisoned_indices": idx.tolist(), "ratio": len(idx) / len(new_X), }
⚠️ 常见坑:孤立森林只能抓住"特征分布上离群"的样本。一个精心设计的后门样本如果特征和正常样本很接近(只是多了个不起眼的小图案),异常检测是抓不到的。所以异常检测要和后面第 1.3 节讲的"激活分析"配合——后者看的是模型内部行为,不是输入特征。
窃取需要大量查询,最直接的防御就是卡查询频率。一个好的速率限制器要同时管三个时间窗:每分钟、每小时、每天。
from collections import defaultdict from datetime import datetime, timedelta import hashlib class RateLimiter: def __init__(self, per_min=60, per_hour=1000, per_day=10000): self.limits = (per_min, per_hour, per_day) self.history = defaultdict(list) def allow(self, user_id, api_key=None): now = datetime.utcnow() key = self._user_key(user_id, api_key) # 只保留 24 小时内的记录 self.history[key] = [t for t in self.history[key] if t > now - timedelta(hours=24)] h = self.history[key] cnt = ( sum(1 for t in h if t > now - timedelta(minutes=1)), sum(1 for t in h if t > now - timedelta(hours=1)), len(h), ) if any(c >= lim for c, lim in zip(cnt, self.limits)): return False, cnt h.append(now) return True, cnt @staticmethod def _user_key(user_id, api_key): # 不直接存敏感信息,存哈希 if api_key: return hashlib.sha256(api_key.encode()).hexdigest()[:16] return user_id
速率限制防的是"暴力蒸馏",但防不住"慢工出细活"——攻击者把查询分散到很多天、很多账号,单看每个都像正常用户。所以速率限制要配合下面的查询模式分析。
第二个手段是"让攻击者复制出来的模型不够准、还能被你认出来"。做法是给模型输出加一点轻微扰动(让人感觉不到,但让蒸馏出来的复制模型精度下降),同时嵌入一个隐蔽的水印(事后能用它证明"这个嫌疑模型是从我这里偷的")。
import numpy as np class OutputProtector: def __init__(self, key="secret", strength=0.01): self.strength = strength rng = np.random.RandomState(hash(key) % (2**32)) # 固定的水印模式,用于事后比对相关性 self.pattern = rng.randn(1000) def protect(self, logits): # 把水印模式扩展到和 logits 等长,叠加微小扰动 tiled = np.tile(self.pattern, len(logits) // len(self.pattern) + 1)[:len(logits)] return logits + self.strength * tiled def verify(self, suspect_logits, threshold=0.7): # 怀疑某个模型是偷来的?看它输出和我的水印相关性 corr = np.corrcoef( self.pattern, suspect_logits.flatten()[:len(self.pattern)])[0, 1] return {"watermark_detected": corr > threshold, "corr": float(corr)}
💡 关键直觉:水印的价值不在"防止被偷",而在"事后追责"。你拦不住别人偷,但偷完之后你在法庭或平台投诉时,需要证据证明"这个模型确实派生自我的模型"。水印就是这个证据链的关键一环。
| 防御手段 | 针对的威胁 | 强项 | 弱项 |
|---|---|---|---|
| 数据哈希溯源 | 整体数据被替换 | 简单可靠 | 抓不到掺毒 |
| 异常检测 | 特征离群的毒样本 | 能发现统计异常 | 抓不到隐蔽后门 |
| 速率限制 | 暴力蒸馏窃取 | 拦住高频查询 | 拦不住分散查询 |
| 输出扰动 | 蒸馏复制 | 降低复制精度 | 影响正常输出精度 |
| 模型水印 | 窃取事后追责 | 提供证据 | 不阻止窃取本身 |
下一节我们钻进模型内部,看两类更隐蔽的攻击:对抗样本(用肉眼难辨的扰动骗模型)和后门攻击(在模型里埋触发器),以及对应的对抗训练和激活分析防御。
把前面两个话题合起来看,会发现一个更根本的问题:现代机器学习的数据供应链天然是多源的。预训练语料来自全网爬取,微调数据可能外包标注,第三方数据集直接下载使用——每一环都是投毒的潜在入口,而多数团队对"数据从哪来、经过谁的手"几乎没有记录。这和软件工程早年的教训一模一样:当依赖变成供应链,安全边界就必须从"自己的代码"扩展到"整个来源链"。软件行业用签名和哈希锁定了包的完整性,数据侧对应的就是来源元数据与内容指纹的双登记:样本入库时记录来源、许可、采集时间和内容哈希,训练前校验指纹未被篡改。
另一个值得建立的习惯是"小数据、高信任"的分层。越靠近决策的关键数据,来源越应该收窄:预训练可以用宽口径加严格清洗,微调和对齐数据则应只接受可溯源的渠道,宁缺毋滥。历史上多次投毒研究都表明,攻击者污染公开数据集的成本远低于防御者全网审计的成本,所以把信任边界画在数据接入层,比试图在训练后检测坏样本便宜得多。模型窃取的防线同理:与其在事后诉讼,不如在 API 层就把速率、熵值、输出粒度管起来,让"复制"这个动作本身变得昂贵。
最后补一个量化视角:防御投入怎么和攻击成本对标。攻击者污染一个公开语料源,成本主要是时间;而防御者验证一条数据的真伪,成本是真金白银的算力和人力。这个不对称决定了"全网过滤"必然失败,可行的是不对称反制——把有限的高强度审计集中在攻击者最想污染的位置(微调数据、对齐数据、高权重语料源),公开爬取类数据则用统计检测兜底。模型窃取的账同样要算清楚:一个七十亿参数模型的复制成本,取决于输出端暴露的信息量,暴露概率就多一分还原精度。量化过这些数字,你才能在安全评审会上说服别人为什么这笔预算该花在这里而不是那里。