模型窃取攻击是近年来机器学习安全领域最受关注的话题之一。攻击者不需要破解模型文件，只需要通过正常API接口大量发送查询请求，收集模型的输入输出对，再用这些数据训练一个替代模型，就能在几十万甚至几万次查询内复刻出与原模型高度相似的能力。对于将模型能力封装成API对外提供服务的企业来说，这等于核心资产被悄无声息地搬走了。本文围绕API监控与限速两个核心手段，系统讲解如何构建一套针对模型窃取的纵深防御体系。
一、模型窃取攻击的原理与典型特征
模型窃取的核心思想来源于知识蒸馏。合法的知识蒸馏是模型拥有者主动用大模型的教学信号训练小模型，而攻击者把这个过程反向利用：把目标API当作教师模型，精心构造查询输入，把API返回的预测结果当作标签，训练出自己的学生模型。学术研究已经证明，对于分类任务，攻击者只需原始训练数据量的很小一部分查询，就能让替代模型达到接近原模型的精度。
攻击者在实施窃取时，通常会表现出几个可识别的行为特征。第一是查询量异常，单账号或单IP在短时间内发出远超正常用户的请求。第二是输入分布异常，正常用户的请求来自真实场景，数据分布多样且带有噪声，而攻击者为了最大化信息获取，往往会系统性地构造输入，比如在特征空间中均匀采样、沿决策边界附近密集探测。第三是输入相似度高，窃取工具常采用基于梯度的主动学习策略，生成的查询样本之间存在高度相关性。第四是时间模式异常，攻击请求往往呈现稳定的速率和规律的间隔，缺少人类用户特有的随机性。这些特征为后续的监控检测提供了依据。
理解这些特征之后，防护思路就清晰了：一方面通过监控识别出具有窃取特征的调用行为，另一方面通过限速提高攻击者的时间与经济成本，让窃取在收益上变得不划算。
二、构建API监控体系识别异常调用
监控体系的第一层是基础指标采集。对每一次API调用，至少要记录调用者身份（API Key、账号ID）、来源IP、请求时间戳、输入样本的指纹（例如对输入做哈希）、返回结果以及响应延迟。这些原始日志是后续所有分析的基础。建议将日志异步写入消息队列再落库，避免日志采集本身拖慢API响应。
第二层是统计分析与告警。可以从以下几个维度设定规则：单Key每小时的调用量、不同输入哈希的占比（如果大量请求内容高度重复或高度相似，说明可能是自动化采集）、输入分布与正常流量基线的偏离程度（可以用输入特征的统计量做卡方检验或KL散度计算）、以及请求间隔的方差（机器发起的请求间隔方差极小）。一旦指标超过阈值，触发告警并进入人工审核或自动处置流程。
第三层是基于机器学习的检测。规则系统容易被绕过，可以训练一个二分类器来区分正常流量与窃取流量。特征包括查询间隔统计量、输入样本间的余弦相似度、预测结果的置信度分布等。攻击者构造的探测样本往往落在决策边界附近，模型输出的置信度普遍偏低，这是一个很强的信号。将分类器部署在网关侧，对可疑流量返回降级服务或直接拒绝。下面是一个简单的基于Python的异常检测规则引擎示例：
import time
from collections import defaultdict, deque
class StealDetector:
def __init__(self, window_seconds=3600, max_calls=500, dup_ratio_threshold=0.6):
self.window = window_seconds
self.max_calls = max_calls
self.dup_ratio_threshold = dup_ratio_threshold
self.calls = defaultdict(deque) # 每个API Key的调用时间戳
self.inputs = defaultdict(set) # 每个API Key的输入哈希集合
self.total = defaultdict(int)
def check(self, api_key, input_hash):
now = time.time()
q = self.calls[api_key]
q.append(now)
# 清理时间窗口外的记录
while q and q[0] < now - self.window:
q.popleft()
self.total[api_key] += 1
# 规则一：窗口内调用量超限
if len(q) > self.max_calls:
return False, "调用频率超限，疑似批量采集"
# 规则二：输入重复率过高
unique_count = len(self.inputs[api_key])
if self.total[api_key] > 50:
dup_ratio = 1 - unique_count / self.total[api_key]
if dup_ratio > self.dup_ratio_threshold:
return False, "输入高度重复，疑似自动化探测"
self.inputs[api_key].add(input_hash)
return True, "ok"
这套规则引擎可以直接嵌入API网关的中间件中，先于业务逻辑执行。需要注意的是，规则阈值要根据自身业务的正常流量画像来标定，可以先采集一段时间的真实流量分布，再取分布的右尾作为阈值，避免误伤正常的高频用户。
三、多层限速策略设计
限速的目的是提高攻击成本。单层限速容易被绕过，比如攻击者注册多个账号或使用代理池就能绕过按账号的限速，因此需要设计多层叠加的限速体系。第一层是按账号限速，限制单个API Key的日调用量和QPS；第二层是按IP限速，限制单IP的请求频率，配合IP信誉库拦截已知的数据中心IP和代理IP；第三层是全局弹性限速，当系统检测到整体流量模式异常时，动态收紧所有用户的配额。
在实现方式上，推荐使用令牌桶或滑动窗口算法，并用Redis存储计数状态以支持分布式部署。令牌桶允许一定程度的突发流量，对正常用户体验更友好；滑动窗口则对持续稳定的高速请求更敏感，恰好匹配窃取攻击的流量特征。以Nginx配合Lua实现按Key限速是一种常见方案：
-- 在OpenResty的access阶段执行
local redis = require "resty.redis"
local red = redis:new()
red:set_timeout(1000)
local ok, err = red:connect("127.0.0.1", 6379)
if not ok then
ngx.log(ngx.ERR, "redis connect failed: ", err)
return
end
local api_key = ngx.req.get_headers()["X-API-Key"]
if not api_key then
ngx.exit(401)
end
-- 使用有序集合实现滑动窗口限速
local now = ngx.now()
local window = 3600 -- 1小时窗口
local limit = 300 -- 每小时最多300次
local key = "rate:" .. api_key
red:zremrangebyscore(key, 0, now - window)
local count = red:zcard(key)
if tonumber(count) >= limit then
ngx.exit(429) -- Too Many Requests
end
red:zadd(key, now, now .. "-" .. math.random())
red:expire(key, window)
除了硬性限速，还可以采用软性降级策略。对超过配额的请求不直接拒绝，而是返回缓存结果、降低响应精度（例如分类任务只返回Top1类别不返回概率分布）、增加响应延迟等。降低输出精度尤其重要，因为完整的类别概率分布携带的信息量远大于单一标签，攻击者需要多几倍甚至几十倍的查询才能达到相同效果，这直接抬高了窃取成本。
四、纵深防御的补充措施
监控和限速之外，还有几项措施值得配合使用。首先是输出扰动，在返回的概率值上加入小幅随机噪声，对正常用户几乎无感知，但会显著污染攻击者的训练数据质量。其次是水印技术，在模型中嵌入特定触发条件下的水印输出，一旦怀疑某个第三方模型是窃取所得，可以通过触发水印来验证来源，为法律维权提供证据。再次是合同与定价约束，对API调用条款明确禁止批量采集行为，并对高配额用户建立审核机制。
最后要强调的是，没有任何单一手段能完全杜绝模型窃取，防御的本质是让攻击成本高于攻击收益。一套设计良好的监控加限速体系，能把数万次查询的窃取成本推高到数月时间与高额费用，配合输出扰动与水印溯源，绝大多数攻击者都会放弃。建议从基础的限速规则做起，逐步完善监控画像，把模型安全纳入API服务的标准工程流程中持续迭代。