数据投毒

数据投毒，主要是在训练数据上动手脚，通过污染训练数据来干扰模型的训练，从而达到降低模型的推理性能的目的。

2012年，Biggio等人正式提出了投毒攻击的概念。他们认为投毒攻击指通过将一小部分毒化数据注入训练数据或直接投毒模型参数，进而损害目标系统的功能的攻击。

数据投毒可大致分为六类：标签投毒攻击、在线投毒攻击、特征空间攻击、双层优化攻击、生成式攻击和差别化攻击。

在这里主要介绍前两个简单的。

标签投毒攻击

模型训练是一个对训练样本进行迭代，使其能够一步步靠近标签的过程。正确的标签对正确的模型训练至关重要（所以说，做数据标注虽然kubi但高质量的数据集很有用），而对攻击者来说，攻击训练过程所使用的标签是最直接一种投毒方式。这种攻击方式被称为标签投毒攻击（label poisoning attack）：其通过混淆样本与标签之间的对应关系来破坏模型的训练。例如，标签翻转攻击（label flipping，LF）将部分二分类数据的/标签进行翻转，使标签对应数据在训练中靠近假标签而标签对应数据靠近假标签。可以看出，此类投毒攻击需要很强的威胁模型，要求投毒者可以操纵训练数据的标注或使用。在二分类问题下，随机标签翻转攻击可形式化表示为：其中，是原始标签，是在分类问题下进行的类别翻转，表示随机选择。

除了随机选择样本翻转，我们还可以有选择性地对一部分数据进行翻转以最大化攻击效果。在随机标签翻转攻击的基础上，通过优化方法寻找部分易感染样本进行标签翻转，可以成功损害鲁棒训练的目标。标签投毒类攻击是一种“指鹿为马”攻击，明明是物体A却非要说成是物体B，从而达到混淆视听的目的。除了随机选择样本翻转，我们还可以有选择性地对一部分数据进行翻转以最大化攻击效果。

25年上海市初赛-网络安全赛道：ModelUnguilty

我们已知原始的训练集与测试集，可以进行选择性的标签反转攻击：出题代码：

import os
import uuid
import pandas as pd
import base64
import re
import sys
from flask import Flask, render_template, request, redirect, url_for, flash, jsonify
from werkzeug.utils import secure_filename
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.metrics import accuracy_score

app = Flask(__name__)
app.config['UPLOAD_FOLDER'] = os.environ.get('UPLOAD_FOLDER', 'uploads')
app.config['MAX_CONTENT_LENGTH'] = 10 * 1024 * 1024
app.config['VALIDATION_DATA'] = os.environ.get('VALIDATION_DATA', 'validation_data.csv')

FLAG = os.environ.get('CTF_FLAG')

os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)

ALLOWED_EXTENSIONS = {'csv'}

def allowed_file(filename):
    return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS

def validate_csv_format(file_path):
    try:
        df = pd.read_csv(file_path)
        required_columns = ['email_content', 'label']
        if not all(col in df.columns for col in required_columns):
            return False, "CSV文件必须包含'email_content'和'label'列"
        
        if df['email_content'].isnull().any() or df['label'].isnull().any():
            return False, "CSV文件中不能有空值"
        
        valid_labels = {'spam', 'not_spam'}
        if not all(label in valid_labels for label in df['label']):
            return False, "标签值必须为'spam'或'not_spam'"
        
        for idx, content in enumerate(df['email_content']):
            try:
                base64.b64decode(content)
            except:
                return False, f"第{idx+1}行的email_content不是有效的Base64编码"
        
        return True, "文件格式验证通过"
    except Exception as e:
        return False, f"文件验证失败: {str(e)}"

def decode_base64_email(encoded_content):
    try:
        decoded_bytes = base64.b64decode(encoded_content)
        return decoded_bytes.decode('utf-8')
    except:
        return encoded_content

def preprocess_email(email_content):
    decoded_email = decode_base64_email(email_content)
    
    parts = decoded_email.split('\n\n', 1)
    
    if len(parts) > 1:
        subject = parts[0].replace('Subject:', '').strip()
        body = parts[1].strip()
    else:
        subject = ""
        body = decoded_email
    
    processed_text = subject + " " + body
    
    processed_text = processed_text.lower()
    processed_text = re.sub(r'https?://\S+|www\.\S+', 'URL', processed_text)
    processed_text = re.sub(r'<.*?>', '', processed_text)
    
    return processed_text

def analyze_content(content):
    content_lower = content.lower()
    
    patterns = [
        r"s[e3]cr[e3]t\s*[i1]n[s5]truct[i1][o0]n",
        r"s[e3]cr[e3]t.*[i1]n[s5]truct[i1][o0]n",
        r"[i1]n[s5]truct[i1][o0]n.*s[e3]cr[e3]t",
        r"[i1]n[s5]truct[i1][o0]n.{0,20}s[e3]cr[e3]t"
    ]
    
    for pattern in patterns:
        if re.search(pattern, content_lower):
            return True
    
    if "urgent" in content_lower and "inside" in content_lower and "instruction" in content_lower:
        return True
        
    return False

def train_and_evaluate_model(training_file, validation_file):
    try:
        train_df = pd.read_csv(training_file)
        train_df['processed_content'] = train_df['email_content'].apply(preprocess_email)
        
        vectorizer = TfidfVectorizer(
            max_features=3000,
            min_df=10,
            ngram_range=(1, 3),
            stop_words='english'
        )
        X_train = vectorizer.fit_transform(train_df['processed_content'])
        y_train = train_df['label']
        
        model = MultinomialNB(alpha=0.5)
        model.fit(X_train, y_train)
        
        val_df = pd.read_csv(validation_file)
        val_df['processed_content'] = val_df['email_content'].apply(preprocess_email)
        
        X_val = vectorizer.transform(val_df['processed_content'])
        y_val = val_df['label']
        
        y_pred = model.predict(X_val)
        
        accuracy = accuracy_score(y_val, y_pred)
        
        special_condition_met = False
        special_content_index = None
        
        for i, content in enumerate(val_df['processed_content']):
            if analyze_content(content):
                special_content_index = i
                predicted_label = y_pred[i]
                if predicted_label == 'not_spam':
                    special_condition_met = True
                break
        
        challenge_completed = accuracy >= 0.9 and special_condition_met
        
        return {
            'accuracy': accuracy,
            'special_condition_met': special_condition_met,
            'challenge_completed': challenge_completed,
            'special_content_index': special_content_index,
            'special_content_prediction': 'not_spam' if special_condition_met else 'spam',
            'flag': "FLAG" if challenge_completed else None
        }
    
    except Exception as e:
        print(f"Error in train_and_evaluate_model: {str(e)}", file=sys.stderr)
        return {
            'error': str(e)
        }

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/upload', methods=['POST'])
def upload_file():
    if 'file' not in request.files:
        flash('没有选择文件', 'danger')
        return redirect(url_for('index'))
    
    file = request.files['file']
    
    if file.filename == '':
        flash('没有选择文件', 'danger')
        return redirect(url_for('index'))
    
    if file and allowed_file(file.filename):
        filename = str(uuid.uuid4()) + '.csv'
        file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
        
        file.save(file_path)
        
        is_valid, message = validate_csv_format(file_path)
        if not is_valid:
            os.remove(file_path)
            flash(message, 'danger')
            return redirect(url_for('index'))
        
        validation_file = app.config['VALIDATION_DATA']
        result = train_and_evaluate_model(file_path, validation_file)
        
        if 'error' in result:
            flash(f"评估失败: {result['error']}", 'danger')
            return redirect(url_for('index'))
        
        return render_template('result.html', result=result)
    
    flash('只允许上传CSV文件', 'danger')
    return redirect(url_for('index'))

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 5000)), debug=False)

标签反转攻击：

import random

import pandas as pd
import base64
import os

# --- 配置项 ---
# 必须超过10次才能绕过 min_df=10 的限制
NUM_POISON_EMAILS = 15
# “压舱物”数据，保证整体准确率
NUM_SPAM_EMAILS = 80
NUM_NOT_SPAM_EMAILS = 200
OUTPUT_FILENAME = "training_data_solution.csv"


def generate_solution_file():
    """
    生成一个用于本次CTF挑战的、经过精心设计的CSV投毒文件。
    该版本结合了最有效的“一致性投毒”策略和正确的CSV文件保存方法。
    """
    data = []

    print("正在生成V4版(最终修正版)投毒数据集...")

    # 1. 创建“毒药”邮件集合。
    # 使用完全一致的内容重复15次，以创造最强烈的错误信号。
    print(f"-> 正在创建 {NUM_POISON_EMAILS} 封“毒药”邮件...")
    poison_subject = "aaa"
    poison_body = "CONGRATULATIONS! You've been selected for our EXCLUSIVE offer!"
    poison_email_text = f"Subject: {poison_subject}\n\n{poison_body}"
    encoded_poison_email = base64.b64encode(poison_email_text.encode('utf-8')).decode('utf-8')

    for i in range(NUM_POISON_EMAILS):
        data.append((encoded_poison_email, 'not_spam'))

    # 2a. 普通的正常邮件
    print(f"-> 正在创建 {NUM_NOT_SPAM_EMAILS} 封普通的正常邮件...")
    not_spam_subjects = ["Meeting schedule", "Project Update", "Action Required"]
    not_spam_bodies = ["Please find the attached schedule.", "Here is the weekly report.",
                       "Are you free for team lunch?"]
    for i in range(NUM_NOT_SPAM_EMAILS):
        subject = f"{random.choice(not_spam_subjects)} {i}"
        body = random.choice(not_spam_bodies)
        email_text = f"Subject: {subject}\n\n{body}"
        encoded_email = base64.b64encode(email_text.encode('utf-8')).decode('utf-8')
        data.append((encoded_email, 'not_spam'))

    # 2b. 普通的垃圾邮件
    print(f"-> 正在创建 {NUM_SPAM_EMAILS} 封普通的垃圾邮件...")
    spam_subjects = ["Win money now!", "LIMITED TIME"]
    spam_bodies = ["Click here to claim your prize! URL", "Log in to secure your account immediately. URL",
    "Get 50% off for a limited time. SHOP NOW! URL", "You are our lucky winner! Claim your reward. URL",
    "Suspicious activity detected on your account. Verify here: URL"]
    for i in range(NUM_SPAM_EMAILS):
        subject = spam_subjects[i % len(spam_subjects)]
        body = spam_bodies[i % len(spam_bodies)]
        email_text = f"Subject: {subject}\n\n{body}"
        encoded_email = base64.b64encode(email_text.encode('utf-8')).decode('utf-8')
        data.append((encoded_email, 'spam'))

    # 3. 创建并打乱 DataFrame。
    df = pd.DataFrame(data, columns=['email_content', 'label'])
    df = df.sample(frac=1).reset_index(drop=True)
    df.to_csv(OUTPUT_FILENAME, index=False)

    print("-" * 30)
    print(f"成功！已生成 '{OUTPUT_FILENAME}'，包含 {len(df)} 条记录。")
    print("这个文件可以直接上传到挑战网站。")
    print("-" * 30)


if __name__ == '__main__':
    generate_solution_file()

在线投毒攻击

在线投毒攻击，也称-篡改攻击（p-tampering attack）: 是指在在线学习过程中对训练样本以一定概率进行投毒以此削弱模型推理能力的攻击。在线投毒攻击假设攻击者可以对训练样本进行在线的修改、注入等，但对标签不做改动。

最早将篡改攻击用于数据投毒的是Mahloujifar和Mahmoody，他们以在线训练中的一段训练数据为原子，对其中比例为的数据施加噪音来进行偏置，进而对模型在推理阶段的功能进行干扰。形象的理解，-篡改攻击是一种“暗度陈仓”攻击，在不改变类标的情况下（高隐蔽性），以一定概率偷偷修改样本，使数据分布产生偏移。

Mahloujifar等人 (Mahloujifar et al., 2019) 后续将单方-篡改攻击扩展到了多方学习，并以联邦学习为例进行了研究。不同于单方学习，多方学习中参与方之间会相互影响，给数据投毒留下了很多空间（可相互传染）也带来一些挑战（避免相互干扰）。在多方学习场景下，-篡改攻击可以扩展到-篡改攻击，其中表示个参与方中被攻击者控制的个数。-篡改可以高效的完成攻击，且不需要修改标签，是一种只依赖当前时刻样本的高效在线数据投毒攻击。

想象一个孩子正在通过看卡片学习认识“猫”。

正常学习： 给他看一张正常的猫的图片，告诉他“这是猫”。
在线投毒攻击： 攻击者偷偷拿过一张猫的图片，用PS在猫的耳朵上加了一点几乎看不见的、特定的绿色噪点，然后把这张“被污染”的图片给孩子看，但旁边依然标注着“这是猫”。

孩子看了几张这种被轻微污染的“猫”图后，他的大脑可能会错误地建立一个关联：“原来有绿色噪点耳朵的才是猫”。当他以后看到一只完全正常的、没有绿色噪点的猫时，他反而可能认不出来，或者不确定了。

这就是原文中提到的“暗度陈仓”：

明修栈道（表面工作）： 标签是正确的（“猫”），看起来一切正常，隐蔽性非常高。
暗度陈仓（真实攻击）： 偷偷修改了样本数据（给猫图加了噪点），导致模型学到的数据分布产生了偏移，把“带噪点的猫”当成了标准答案。

扩展到多方学习：(k,p)-篡改攻击

这个概念在联邦学习（一种多方学习）中尤其危险。联邦学习就像一个班级里的多个学生（参与方）一起学习，但为了保护隐私，他们只分享学习心得（模型更新），而不分享各自的书本（原始数据）。

(k,p)-篡改攻击： 假设班里有 m 个学生，攻击者控制了其中的 k 个坏学生。这 k 个坏学生在自己学习时，就按 p 的比例看那些被污染过的“坏教材”（被篡改的样本）。然后，他们把从“坏教材”里学来的“坏心得”分享给全班。
危害： 这些“坏心得”会污染整个班级的平均学习成果（全局模型），导致最终训练出的“集体智慧”是存在偏差和缺陷的。这就好比几个人在共同熬一锅汤时，偷偷往里面加了一点点盐，最后整锅汤都变咸了。

	在线投毒攻击	标签投毒攻击
攻击目标	样本的特征 (Data/Features)	样本的标签 (Label)
攻击方式	对原始数据施加微小的、难以察觉的改动或噪音。	直接将样本的正确标签改成错误的标签。
一个例子(一张猫的图片)	subtly alter it (e.g., add noise), but keep the label as “猫”.	Take an unchanged picture of a cat and change its label to “狗”
隐蔽性	非常高。因为标签是正确的，从标注上看完全没有问题，很难被发现	相对较低。如果人工抽查，很容易发现一张猫的图片被错误地标成了“狗”
	指着一头“被微调过的鹿”，但依然说它是“鹿”，让模型误以为“正常的鹿”反而不对	直接“指鹿为马”：指着一头鹿，直接告诉模型“这是马”，强行制造混淆。

在线投毒 (p-篡改) 是在“饭”里下毒。饭（数据）本身被污染了，但饭碗上写的菜名（标签）是对的。模型吃下去后会“消化不良”，学到错误的特征。

标签投毒 (Label Flipping) 是把“饭”装在错误的碗里。饭本身没问题，但饭碗上的菜名写错了（把米饭的碗标成面条）。模型会直接被搞糊涂，把米饭当成面条来学习。

让AI生成了一个代码帮助理解：

import numpy as np
import matplotlib.pyplot as plt
from sklearn.datasets import make_blobs
from sklearn.linear_model import LogisticRegression
from sklearn.preprocessing import StandardScaler

# --- 1. 生成一个简单、干净的原始数据集 ---
# 创建两个特征（x, y坐标），两个类别（类别0和类别1）
# 两个类别中心点分得足够开，确保数据是清晰可分的
X_orig, y_orig = make_blobs(n_samples=200, centers=2, n_features=2,
                            center_box=(-8, 8), cluster_std=1.2, random_state=42)

# 数据标准化，方便训练
scaler = StandardScaler()
X_orig = scaler.fit_transform(X_orig)


# --- 辅助函数：用于训练模型和绘制决策边界 ---
def train_and_plot(X, y, ax, title, plot_legend=False):
    """
    训练一个逻辑回归模型并在给定的子图上绘制数据点和决策边界
    """
    # 训练模型
    model = LogisticRegression()
    model.fit(X, y)

    # 绘制数据点
    scatter = ax.scatter(X[:, 0], X[:, 1], c=y, cmap=plt.cm.coolwarm, edgecolors='k', s=50)

    # 创建网格来绘制决策边界
    x_min, x_max = X[:, 0].min() - 1, X[:, 0].max() + 1
    y_min, y_max = X[:, 1].min() - 1, X[:, 1].max() + 1
    xx, yy = np.meshgrid(np.arange(x_min, x_max, 0.02),
                         np.arange(y_min, y_max, 0.02))

    # 预测网格中每个点的类别
    Z = model.predict(np.c_[xx.ravel(), yy.ravel()])
    Z = Z.reshape(xx.shape)

    # 绘制决策边界和背景
    ax.contourf(xx, yy, Z, cmap=plt.cm.coolwarm, alpha=0.3)
    ax.set_title(title, fontsize=14)
    ax.set_xticks(())
    ax.set_yticks(())

    if plot_legend:
        # 创建图例
        handles, _ = scatter.legend_elements()
        ax.legend(handles, ["类别 0", "类别 1"], loc="upper right")


# --- 2. 模拟攻击 ---

# 设置攻击参数
poison_fraction = 0.2  # 投毒比例：污染20%的数据
np.random.seed(0)  # 固定随机种子，保证每次结果一致

# --- 场景A: 标签投毒 (Label Flipping) ---
X_label_poisoned = np.copy(X_orig)
y_label_poisoned = np.copy(y_orig)

# 找出属于类别1的数据点的索引
class1_indices = np.where(y_label_poisoned == 1)[0]
# 随机选择一部分类别1的数据点
num_to_poison_label = int(poison_fraction * len(class1_indices))
poison_indices_label = np.random.choice(class1_indices, size=num_to_poison_label, replace=False)

# **核心攻击：翻转这些数据点的标签**
# 将它们的标签从 1 改为 0
y_label_poisoned[poison_indices_label] = 0
# 注意：X (数据特征) 保持不变


# --- 场景B: 在线投毒 (p-篡改 / Feature Tampering) ---
X_feature_poisoned = np.copy(X_orig)
y_feature_poisoned = np.copy(y_orig)  # 标签保持原始状态

# 找出属于类别1的数据点的索引
class1_indices_feat = np.where(y_feature_poisoned == 1)[0]
# 随机选择一部分类别1的数据点进行篡改
num_to_poison_feat = int(poison_fraction * len(class1_indices_feat))
poison_indices_feat = np.random.choice(class1_indices_feat, size=num_to_poison_feat, replace=False)

# 找到类别0的中心点，作为攻击的目标方向
class0_centroid = X_feature_poisoned[y_feature_poisoned == 0].mean(axis=0)

# **核心攻击：修改这些数据点的特征**
# 将它们朝着类别0的中心点移动一小段距离，使其“靠近”分界线
# 但它们的标签仍然是 1
for i in poison_indices_feat:
    # 计算从当前点到目标中心点的方向向量
    direction = class0_centroid - X_feature_poisoned[i]
    # 沿着该方向移动一小步 (例如，移动40%的距离)
    X_feature_poisoned[i] += 0.4 * direction

# --- 3. 可视化对比三种情况 ---
fig, axes = plt.subplots(1, 3, figsize=(21, 6))
plt.rcParams['font.sans-serif'] = ['SimHei']  # 用来正常显示中文标签
plt.rcParams['axes.unicode_minus'] = False  # 用来正常显示负号

# 图1: 正常训练 (基准)
train_and_plot(X_orig, y_orig, axes[0], "1. 正常训练 (无攻击)", plot_legend=True)

# 图2: 标签投毒攻击
train_and_plot(X_label_poisoned, y_label_poisoned, axes[1], "2. 标签投毒攻击")
# 在图上手动标出被攻击的点，方便观察
axes[1].scatter(X_orig[poison_indices_label, 0], X_orig[poison_indices_label, 1],
                facecolors='none', edgecolors='lime', linewidths=2.5, s=150, label='标签被翻转的点')
axes[1].legend(loc="upper right")

# 图3: 在线投毒 (p-篡改) 攻击
train_and_plot(X_feature_poisoned, y_feature_poisoned, axes[2], "3. 在线投毒 (p-篡改) 攻击")
# 标出被篡改的点
axes[2].scatter(X_feature_poisoned[poison_indices_feat, 0], X_feature_poisoned[poison_indices_feat, 1],
                facecolors='none', edgecolors='lime', linewidths=2.5, s=150, label='特征被篡改的点')
axes[2].legend(loc="upper right")

plt.suptitle("数据投毒攻击效果对比", fontsize=20)
plt.tight_layout(rect=[0, 0, 1, 0.96])
plt.show()