内部 Agent 的模型窃取：用低权限调用重建权重

内部 Agent 的模型窃取用低权限调用重建权重一、低权限账号的合规外衣与窃取成本企业把大模型封装成 Agent 对内开放通常是一套 API 网关按角色分发配额。员工账号默认能调单次 QPS 不高但累计调用量不小。这套设计觉得低权限等于低风险却没考虑模型窃取的逻辑攻击者不需要一次拿走权重持续提问就行用足够多的查询能重建一个功能等价的副本。模型窃取就是知识蒸馏。攻击者把目标模型当教师用输入集喂进去拿到 logits 或回答再训练个学生模型去拟合。查询覆盖足够多样本学生模型在关键任务上就能逼近教师表现。不需要权重文件泄露只要问得到、答得回就行。内部 Agent 给窃取创造了条件。访问合法员工账号调 API 完全合规风控不会立刻触发响应丰富很多 Agent 为了可解释性会返回完整 logits、top-k 候选甚至中间推理等于直接送出蒸馏标签成本外部化调用算力企业买单攻击者只花时间。的风险是低频长期。一次调几千次不会引起注意但持续几周累计百万级查询足够训练一个有相当能力的副本。更隐蔽的是把查询伪装成正常业务流量混在不同时间、不同接口里QPS 限流就失灵了。窃取防护的重点在识别异常查询模式。威胁模型很清楚攻击者有合法低权限账号目标是用最低成本、最不引人注意的方式获取足够多样的查询-响应对。防御要做的就是让足够多样这件事在成本和时间上变得划不来。二、窃取链路与三道防御关口的对应关系一次完整的蒸馏式窃取可以拆成四个阶段种子集构造、查询发射、响应回收、学生训练。每个阶段都对应可观测的痕迹也是防御可以拦截的关口。限流管能不能问得这么快行为指纹管问得是不是像人水印溯源管副本出现后能不能追到源头。三者各管一个阶段。光限流挡不住低频长期窃取光指纹挡不住模仿正常业务的攻击者光水印只能事后追责。三、生产级窃取防御网关与水印嵌入下面是一段窃取防御网关实现。它包含滑动窗口限流、查询分布指纹检测、响应水印嵌入与审计落库import time import math import hashlib import threading from collections import deque, defaultdict from dataclasses import dataclass, field # 用户调用记录含时间戳、查询哈希、返回水印 dataclass class CallRecord: user_id: str ts: float query_hash: str watermark: str # 滑动窗口限流按用户限制单位时间内的最大调用数 class SlidingWindowLimiter: def __init__(self, window_sec: float 60.0, max_calls: int 30): self._window window_sec self._max max_calls self._buckets: dict[str, deque] defaultdict(deque) self._lock threading.Lock() # 多线程并发安全 def allow(self, user_id: str) - tuple[bool, str]: now time.time() with self._lock: q self._buckets[user_id] # 清理窗口外的旧记录 while q and now - q[0] self._window: q.popleft() if len(q) self._max: retry self._window - (now - q[0]) return False, frate_limited retry{retry:.1f}s q.append(now) return True, ok # 行为指纹检测查询分布是否偏离正常业务模式 class BehaviorFingerprint: def __init__(self, sample_size: int 50, entropy_floor: float 2.5): self._sample_size sample_size # 熵下限正常业务查询应有一定多样性纯随机或纯重复都异常 self._entropy_floor entropy_floor self._history: dict[str, deque] defaultdict(deque) self._lock threading.Lock() def observe(self, user_id: str, query: str) - tuple[bool, str]: qh hashlib.md5(query.encode()).hexdigest()[:8] with self._lock: hist self._history[user_id] hist.append(qh) if len(hist) self._sample_size: hist.popleft() if len(hist) self._sample_size: return True, warming_up # 计算查询哈希的香农熵 counts defaultdict(int) for h in hist: counts[h] 1 n len(hist) entropy -sum((c / n) * math.log2(c / n) for c in counts.values()) # 熵过高近乎纯随机熵过低重复少数模板二者都疑似窃取 if entropy 5.5 or entropy self._entropy_floor: return False, fsuspicious_entropy{entropy:.2f} return True, ok # 响应水印在生成回答中嵌入用户级不可见指纹用于副本溯源 class ResponseWatermarker: staticmethod def embed(user_id: str, session_id: str, text: str) - str: # 用 Unicode 零宽字符编码用户标识肉眼不可见但可解码 sig hashlib.sha1(f{user_id}|{session_id}.encode()).hexdigest()[:8] zero_width { 0: \u200b, 1: \u200c, 2: \u200d, 3: \u2060, 4: \u2061, 5: \u2062, 6: \u2063, 7: \u2064, 8: \u2065, 9: \u2066, a: \u2067, b: \u2068, c: \u2069, d: \u206a, e: \u202a, f: \u202b, } # 把签名转成零宽字符序列插到回答首字符之后 marker .join(zero_width[ch] for ch in sig) if not text: return text return text[0] marker text[1:] # 网关把限流、指纹、水印串成一条调用链 class StealingDefenseGateway: def __init__(self): self._limiter SlidingWindowLimiter(window_sec60.0, max_calls30) self._finger BehaviorFingerprint(sample_size50) self._marker ResponseWatermarker() def call(self, user_id: str, session_id: str, query: str) - dict: ts time.time() # 第一道限流 ok, reason self._limiter.allow(user_id) if not ok: self._audit(user_id, ts, query, rejected, reason) return {status: rejected, reason: reason} # 第二道行为指纹 ok, reason self._finger.observe(user_id, query) if not ok: self._audit(user_id, ts, query, rejected, reason) return {status: rejected, reason: reason} # 模拟模型生成真实环境接入推理服务 answer f回答:{query[:16]}... # 第三道响应水印 watermarked self._marker.embed(user_id, session_id, answer) self._audit(user_id, ts, query, ok, watermarked) return {status: ok, answer: watermarked} def _audit(self, user_id: str, ts: float, query: str, action: str, reason: str): qh hashlib.md5(query.encode()).hexdigest()[:8] # 审计只留查询哈希不留原文避免审计库本身成为泄露面 print(fAUDIT|{ts:.3f}|{user_id}|{action}|{reason}|qhash{qh}) # 使用示例 def demo(): gw StealingDefenseGateway() for i in range(3): r gw.call(u_101, sess_1, f查询{i}) print(r)限流用滑动窗口而非固定窗口避免边界突发指纹用查询熵识别太随机和太重复都是窃取特征水印用零宽字符嵌入肉眼不可见但能解码定位审计只留查询哈希不留原文避免审计库变成新泄露面。四、防御的边界与攻击者的对抗升级窃取防御不是装完就完事攻击者会不断升级绕过手段。限流会被分布式调用绕过。攻击者拿到多个低权限账号把流量分摊到几十个身份上单账号 QPS 始终在阈值内。用户维度的限流没用得加全局异常流量检测非业务高峰时段整体调用量陡升或某个网段集中调用时即使单账号合规也要告警。行为指纹会被业务模仿击败。攻击者把窃取查询混在真实业务请求里比如伪装成客服问答、搜索补全让熵值落在正常区间。纯统计指纹识别不了这种语义级伪装。得用查询语义聚类把同一用户的历史查询做向量化聚类发现业务范围外的异常主题。但这会带来隐私与算力成本得按风险等级分级启用。水印会被对抗性清洗破坏。攻击者知道有水印会用文本归一化、同义改写、字符过滤等手段清洗响应。零宽字符水印很脆弱更稳健的是语义水印生成阶段让模型对特定查询产生轻微但可识别的偏好这种水印没法通过字符操作去除。但语义水印会影响模型质量得在可用性与可溯源之间权衡。还有个常被忽略的点低权限账号获取成本很低。企业离职流程不严谨老账号长期没回收外部合作方账号权限过宽测试账号流入开发者社区。这些都是窃取入口。技术防御再好账号生命周期治理跟不上等于门上了三道锁却留着窗户。窃取防护得把账号回收和权限定期复审纳入闭环。五、总结内部 Agent 的模型窃取就是把企业的算力与模型能力通过合法低权限调用持续外迁到攻击者手里。蒸馏式窃取不需要权重文件只要查询-响应对足够多样。防御有三道关口滑动窗口限流压瞬时流量查询熵指纹识别异常分布响应水印支持事后溯源。工程上要接受分布式账号绕过、业务伪装、水印清洗这些对抗把全局流量检测、语义聚类、账号生命周期治理纳入闭环。窃取防护把窃取的时间与成本抬到不划算的水平。