内部 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, f"rate_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, f"suspicious_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_sec=60.0, max_calls=30)
        self._finger = BehaviorFingerprint(sample_size=50)
        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(f"AUDIT|{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 的模型窃取，就是把企业的算力与模型能力通过合法低权限调用持续外迁到攻击者手里。蒸馏式窃取不需要权重文件，只要查询-响应对足够多样。防御有三道关口：滑动窗口限流压瞬时流量，查询熵指纹识别异常分布，响应水印支持事后溯源。工程上要接受分布式账号绕过、业务伪装、水印清洗这些对抗，把全局流量检测、语义聚类、账号生命周期治理纳入闭环。窃取防护把窃取的时间与成本抬到不划算的水平。