如何通过设置查询预算实现高效的黑盒SPSA或基于决策的攻击？

1. 引言：黑盒攻击中的查询悖论

在模型部署（Model Deployment）环境中，我们通常只能通过API接口访问目标模型，即“黑盒”场景。对抗性攻击（Adversarial Attacks）在这种场景下最具挑战性，因为攻击者无法访问模型的内部梯度信息。同步扰动随机近似（Simultaneous Perturbation Stochastic Approximation, SPSA）是一种高效的零阶优化方法，常用于黑盒环境下的梯度估计和对抗样本生成。

然而，SPSA的效率瓶颈在于查询预算（Query Budget）。每次迭代都需要至少两次模型查询来估计梯度，如果迭代次数过多，不仅会产生高昂的API调用费用，还可能触犯速率限制（Rate Limiting），导致攻击失败。因此，如何在一个严格的查询预算内最大化攻击成功率，成为AI基础设施安全的关键问题。

本文将深入探讨SPSA的查询成本，并提供一个基于预算约束的SPSA实现，确保攻击在预设限制内高效运行。

2. SPSA原理回顾与查询成本分析

SPSA通过对输入进行随机扰动来估计损失函数 $J(x)$ 的梯度。对于一个 $D$ 维的输入向量 $x$，SPSA仅需两次查询，而不是像有限差分那样需要 $2D$ 次查询。

梯度估计公式：
$$g(x_k) \approx \frac{J(x_k + c_k\Delta_k) – J(x_k – c_k\Delta_k)}{2c_k\Delta_k}$$
其中，$\Delta_k$ 是一个随机扰动向量，通常由伯努利分布生成；$c_k$ 是扰动步长。

SPSA迭代更新：
$$x_{k+1} = x_k – a_k g(x_k)$$
其中，$a_k$ 是学习率。

查询成本：
在一个SPSA攻击流程中，如果需要进行 $N$ 次迭代，则总查询次数至少为 $Q_{total} = 2N$。

3. 基础SPSA攻击实现（模拟黑盒环境）

首先，我们构建一个模拟的黑盒模型和其损失函数。在实际部署中，black_box_query 函数即代表一次网络API调用。

我们使用PyTorch和NumPy进行示例。

import torch
import numpy as np

# 模拟一个简单的黑盒模型（分类任务）
class DummyBlackBoxModel(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.linear = torch.nn.Linear(10, 1)

    def forward(self, x):
        # 模拟模型的 logits 输出
        return self.linear(x)

model = DummyBlackBoxModel()
# 目标标签，我们希望将模型的预测推离这个标签
target_label = 0.5

def black_box_query(x, model, target_label):
    """模拟一次API查询，返回与目标标签的损失。"""
    x_tensor = torch.tensor(x, dtype=torch.float32).unsqueeze(0)
    output = model(x_tensor).squeeze()
    # 假设使用均方误差作为损失函数
    loss = (output - target_label)**2
    return loss.item()

# SPSA参数设置
initial_x = np.random.randn(10)

# 定义基础SPSA攻击函数
def basic_spsa_attack(x0, total_iterations, c0=0.01, a0=0.01):
    x = x0.copy()
    queries_used = 0

    for k in range(1, total_iterations + 1):
        # 学习率和扰动步长衰减（SPSA标准做法）
        ak = a0 / (k**0.602)
        ck = c0 / (k**0.101)

        # 1. 生成扰动向量 Delta (伯努利分布)
        Delta = np.where(np.random.rand(x.shape[0]) > 0.5, 1.0, -1.0)

        # 2. 两次查询 F(x + c*Delta) 和 F(x - c*Delta)
        loss_plus = black_box_query(x + ck * Delta, model, target_label)
        loss_minus = black_box_query(x - ck * Delta, model, target_label)
        queries_used += 2

        # 3. 估计梯度
        grad_estimate = (loss_plus - loss_minus) / (2 * ck) * Delta

        # 4. 更新对抗样本
        x = x - ak * grad_estimate

        if k % 100 == 0:
            print(f"Iteration {k}: Loss = {loss_plus:.4f}, Queries = {queries_used}")

    return x, queries_used

# 示例运行 (如果设置1000次迭代，将消耗2000次查询)
# final_x, q_used = basic_spsa_attack(initial_x, total_iterations=1000)
# print(f"Total Queries Used: {q_used}")

4. 基于预算约束的高效SPSA实现

实际部署中，我们不能预先设定迭代次数，必须严格遵守查询预算 $Q_{max}$。高效的关键在于：

1. 硬性预算限制： 在循环开始前检查是否有足够的预算进行下一步的两次查询。
2. 动态学习率/早停策略： 当损失的下降速度低于某一阈值时，提前终止攻击，将剩余预算用于其他更可能成功的攻击或留存。

以下是修改后的，基于预算约束的SPSA实现。

# 设置严格的查询预算
MAX_QUERY_BUDGET = 500

def budget_constrained_spsa_attack(x0, max_budget, c0=0.01, a0=0.01, min_loss_improvement=1e-5):
    x = x0.copy()
    queries_used = 0
    k = 0
    last_loss = float('inf')

    while queries_used + 2 <= max_budget:
        k += 1

        # --- SPSA 参数衰减 ---
        # 注意：衰减函数中k必须从1开始计数
        ak = a0 / (k**0.602)
        ck = c0 / (k**0.101)

        # --- 梯度估计步骤 ---
        Delta = np.where(np.random.rand(x.shape[0]) > 0.5, 1.0, -1.0)

        loss_plus = black_box_query(x + ck * Delta, model, target_label)
        loss_minus = black_box_query(x - ck * Delta, model, target_label)
        queries_used += 2

        current_loss = (loss_plus + loss_minus) / 2 # 取平均作为当前近似损失

        grad_estimate = (loss_plus - loss_minus) / (2 * ck) * Delta

        # --- 更新 --- 
        x = x - ak * grad_estimate

        # --- 预算与早停策略 --- 
        if current_loss < last_loss - min_loss_improvement:
            # 损失显著下降，继续攻击
            last_loss = current_loss
        elif k > 50 and current_loss > last_loss:
             # 如果迭代超过50次，且损失开始上升（或没有明显下降）
             print(f"Early Stop at Iteration {k}. Loss convergence detected. Current Loss: {current_loss:.4f}")
             break

        if k % 100 == 0:
             print(f"Iteration {k}: Loss = {current_loss:.4f}, Queries Used = {queries_used}/{max_budget}")

    print(f"--- Attack Finished ---")
    print(f"Final Iterations: {k}, Total Queries Used: {queries_used}")
    return x, queries_used

# 运行基于预算约束的攻击
final_x_budget, q_used_budget = budget_constrained_spsa_attack(
    initial_x, 
    max_budget=MAX_QUERY_BUDGET
)

print(f"Query Budget Used: {q_used_budget}, Remaining Budget: {MAX_QUERY_BUDGET - q_used_budget}")

5. 结论

在AI基础设施的实践中，特别是面对外部API或联邦学习场景，查询预算是比计算资源更严格的约束。通过将查询预算（MAX_QUERY_BUDGET）作为核心控制变量，并结合动态早停机制，我们能确保SPSA或任何零阶优化攻击在规定的成本范围内完成，避免因无限制的查询而导致的资源滥用和安全检测。这种方法不仅提高了攻击的“经济效率”，也为防御方设计更智能的速率限制策略提供了参考依据。