引言

训练一个高性能AI模型需要大量的数据、计算资源和时间。一个顶级大模型的训练成本可能高达数千万美元。然而，攻击者可以通过相对低廉的成本"窃取"模型的能力——这就是模型窃取（Model Stealing）攻击。

2026年，随着模型即服务（MaaS）的普及，模型窃取已经成为AI公司面临的最严重的知识产权威胁之一。本文将系统探讨模型窃取攻击的手法和防范策略。

一、模型窃取攻击分类

1.1 按攻击目标分类

功能窃取（Functionality Stealing） 目标：构建一个功能相似的替代模型。 方法：通过大量查询API，用输入输出对训练替代模型。 影响：绕过API费用，竞争产品。

参数窃取（Parameter Stealing） 目标：直接获取模型参数。 方法：通过侧信道攻击、梯度泄露等手段。 影响：完全复制模型，最严重的攻击。

架构窃取（Architecture Stealing） 目标：推断模型架构。 方法：通过API响应时间、输出分布等推断。 影响：降低后续攻击难度。

1.2 按攻击方式分类

黑盒查询攻击 攻击者只能访问模型API，通过查询推断模型。 特点：最常用，难度中等。

侧信道攻击 利用模型推理过程中的侧信道信息（时间、功耗、内存访问模式）。 特点：需要物理访问或云环境内的共置。

供应链攻击 攻击模型训练/部署的供应链环节。 特点：影响深远但难度大。

二、功能窃取攻击详解

2.1 基础查询攻击

最简单的模型窃取：大量查询API，用结果训练替代模型。

async def basic_model_extraction(target_api, num_queries=100000):
    """基础模型提取攻击"""
    # 1. 生成查询样本
    queries = generate_queries(num_queries)
    
    # 2. 查询目标模型
    results = []
    for query in tqdm(queries):
        response = await target_api.query(query)
        results.append((query, response))
    
    # 3. 用查询结果训练替代模型
    surrogate_model = train_surrogate_model(results)
    
    return surrogate_model

防御：查询限制、结果扰动、水印。

2.2 主动学习增强攻击

不是随机查询，而是智能选择最有价值的查询：

async def active_model_extraction(target_api, budget=10000):
    """主动学习增强的模型提取"""
    # 初始查询（小样本）
    initial_queries = generate_initial_queries(1000)
    results = await query_api(target_api, initial_queries)
    
    # 训练初始替代模型
    surrogate = train_surrogate_model(results)
    
    # 主动选择查询
    for i in range(budget // 100):
        # 选择替代模型最不确定的样本
        candidates = generate_candidate_queries(10000)
        uncertainties = surrogate.predict_uncertainty(candidates)
        top_uncertain = select_top_k(uncertainties, k=100)
        
        # 查询目标模型
        new_results = await query_api(target_api, top_uncertain)
        results.extend(new_results)
        
        # 重新训练
        surrogate = train_surrogate_model(results)
    
    return surrogate

防御：检测异常查询模式（不确定性采样模式）。

2.3 迁移学习攻击

利用公开模型进行迁移学习，减少对目标模型的查询：

async def transfer_model_extraction(target_api, public_model):
    """基于迁移学习的模型提取"""
    # 1. 用公开模型初始化
    surrogate = public_model
    
    # 2. 用少量目标模型查询微调
    queries = generate_queries(5000)  # 远少于从头训练
    results = await query_api(target_api, queries)
    
    # 3. 微调
    surrogate = fine_tune(surrogate, results)
    
    return surrogate

防御：检测微调模式（小学习率、特定数据分布）。

三、参数窃取攻击

3.1 梯度基础攻击

在联邦学习或可提供梯度信息的场景中，攻击者可以推断模型参数。

def gradient_based_parameter_inference(gradients, model_architecture):
    """基于梯度的参数推断"""
    # 假设模型架构已知（通过架构窃取）
    inferred_params = {}
    
    for param_name, grad in gradients.items():
        # 通过梯度反推参数
        # 对于线性层: grad_w = x^T * grad_output
        # 如果知道输入x和梯度grad_output，可以解出w
        inferred_params[param_name] = solve_parameters(grad, model_architecture[param_name])
    
    return inferred_params

3.2 成员推断攻击

虽然不直接窃取参数，但可以推断训练数据成员，间接威胁模型安全。

async def membership_inference_attack(target_model, candidate_samples):
    """成员推断攻击"""
    # 训练一个攻击模型，判断样本是否在目标模型的训练集中
    attack_model = train_attack_model(target_model)
    
    results = []
    for sample in candidate_samples:
        confidence = target_model.predict_confidence(sample)
        
        # 在训练集上的样本通常有更高的置信度
        is_member = attack_model.predict(confidence)
        results.append((sample, is_member))
    
    return results

四、架构窃取

4.1 基于推理时间的推断

不同架构的推理时间不同，可以通过API响应时间推断架构。

async def infer_architecture_by_timing(target_api, sample_queries):
    """基于响应时间推断架构"""
    timing_results = []
    
    for query in sample_queries:
        start = time.time()
        await target_api.query(query)
        elapsed = time.time() - start
        timing_results.append(elapsed)
    
    avg_time = np.mean(timing_results)
    
    # 与已知架构的推理时间对比
    for arch_name, arch_time in known_architectures.items():
        if abs(avg_time - arch_time) < threshold:
            return arch_name
    
    return "unknown"

4.2 基于输出分布的推断

不同架构的输出分布特征不同。

async def infer_architecture_by_output(target_api, test_inputs):
    """基于输出分布推断架构"""
    outputs = []
    for inp in test_inputs:
        out = await target_api.query(inp)
        outputs.append(out)
    
    # 分析输出分布
    distribution_features = extract_distribution_features(outputs)
    
    # 与已知架构的输出分布对比
    similarities = {}
    for arch_name, arch_features in known_output_distributions.items():
        sim = cosine_similarity(distribution_features, arch_features)
        similarities[arch_name] = sim
    
    return max(similarities, key=similarities.get)

五、防范措施

5.1 API层面防御

查询限制

class QueryLimiter:
    def __init__(self):
        self.per_user_limit = 10000  # 每用户每日查询限制
        self.per_ip_limit = 50000    # 每IP查询限制
        self.global_limit = 1000000   # 全局限制
    
    def check_limit(self, user_id, ip):
        if self.get_user_queries(user_id) >= self.per_user_limit:
            raise RateLimitError("User query limit exceeded")
        if self.get_ip_queries(ip) >= self.per_ip_limit:
            raise RateLimitError("IP query limit exceeded")
        if self.get_global_queries() >= self.global_limit:
            raise RateLimitError("Global query limit exceeded")

结果扰动

def perturb_output(output, perturbation_level="medium"):
    """对输出添加可控扰动"""
    if perturbation_level == "low":
        noise = np.random.normal(0, 0.01, size=output.shape)
    elif perturbation_level == "medium":
        noise = np.random.normal(0, 0.05, size=output.shape)
    else:  # high
        noise = np.random.normal(0, 0.1, size=output.shape)
    
    perturbed = output + noise
    return perturbed

水印技术

def add_watermark_to_output(output, watermark_key):
    """在输出中添加水印"""
    # 对输出进行微小的、可检测的修改
    # 修改不影响正常使用，但可以证明出处
    watermarked = apply_watermark(output, watermark_key)
    return watermarked

def detect_watermark(output, watermark_key):
    """检测输出中是否包含水印"""
    return verify_watermark(output, watermark_key)

5.2 模型层面防御

差分隐私训练 限制单个训练样本对模型的影响，使模型对特定输入的记忆模糊。

正则化 使模型对输入扰动更鲁棒，减少攻击者可通过梯度推断的信息。

输出截断 对输出进行截断或离散化，减少信息泄露。

def truncate_output(output, precision=2):
    """截断输出精度"""
    if isinstance(output, float):
        return round(output, precision)
    elif isinstance(output, list):
        return [round(x, precision) for x in output]
    else:
        return output

5.3 监控与检测

异常查询检测

class AbnormalQueryDetector:
    def __init__(self):
        self.query_patterns = {}
    
    def detect(self, user_id, queries):
        """检测异常查询模式"""
        features = self.extract_query_features(queries)
        
        # 检测1: 查询分布是否异常（如不确定性采样模式）
        if self.is_uncertainty_sampling_pattern(features):
            return {"is_abnormal": True, "reason": "uncertainty_sampling_pattern"}
        
        # 检测2: 查询是否系统性地探测模型边界
        if self.is_boundary_probing(features):
            return {"is_abnormal": True, "reason": "boundary_probing"}
        
        # 检测3: 查询速度是否异常快（可能使用自动化工具）
        if features["queries_per_second"] > 10:
            return {"is_abnormal": True, "reason": "automated_querying"}
        
        return {"is_abnormal": False}

蜜罐技术

class HoneypotAPI:
    def __init__(self, real_api):
        self.real_api = real_api
        self.honeypot_samples = self.generate_honeypot_samples()
    
    async def query(self, input):
        # 检查是否是蜜罐样本
        if self.is_honeypot(input):
            # 记录攻击者行为
            self.log_attacker(input)
            # 返回误导性结果
            return self.honeypot_response(input)
        else:
            return await self.real_api.query(input)

六、法律与伦理

6.1 知识产权保护

  • 模型权重：作为商业秘密保护
  • 模型架构：可申请专利
  • 训练数据：受版权法保护
  • API使用条款：合同保护

6.2 合理使用边界

  • 为互操作性而进行的逆向工程：可能合理
  • 学术研究：可能合理
  • 商业竞争：通常不合理
  • 恶意复制：明确侵权

结语

模型窃取是AI时代的"盗版"问题。它威胁的不仅是单个公司的利益，更是整个行业的创新动力——如果模型可以被轻易窃取，谁还愿意投入资源训练更好的模型？

2026年的防御是多层次、多角度的：技术防御（API限制、结果扰动、水印）、法律防御（知识产权保护）、监控防御（异常检测、蜜罐）。但最有效的防御可能是"持续创新"——让模型迭代速度超过攻击者窃取的速度。

最终，模型窃取防御不仅是技术问题，也是商业模式问题。通过API提供服务而非发布模型、通过持续更新保持模型新鲜度、通过差异化服务而非模型本身创造价值，这些都是可行的商业防御策略。在技术防御之外，我们还需要商业智慧。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。