7：L防御模型中毒攻击：蓝队的数据安全保障

import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler

def detect_poisoned_samples(X, contamination=0.1):
    """检测训练数据中的中毒样本"""
    # 数据标准化
    scaler = StandardScaler()
    X_scaled = scaler.fit_transform(X)
    
    # 训练孤立森林模型
    model = IsolationForest(contamination=contamination, random_state=42)
    model.fit(X_scaled)
    
    # 预测异常
    predictions = model.predict(X_scaled)
    # 计算异常分数
    scores = model.score_samples(X_scaled)
    
    # 返回异常样本的索引
    poisoned_indices = np.where(predictions == -1)[0]
    return poisoned_indices, scores

# 准备正常数据和中毒数据
np.random.seed(42)
# 生成正常数据
normal_data = np.random.normal(0, 1, (1000, 10))
# 生成中毒数据（后门攻击）
poisoned_data = np.random.normal(0, 1, (100, 10))
# 在中毒数据中嵌入后门
poisoned_data[:, 0] = 9.0  # 后门触发条件

# 合并数据
X = np.vstack([normal_data, poisoned_data])

# 检测中毒样本
poisoned_indices, scores = detect_poisoned_samples(X, contamination=0.1)
print(f"检测到的中毒样本数量: {len(poisoned_indices)}")
print(f"中毒样本索引: {poisoned_indices}")

import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, Flatten

def safe_training(X, y, poisoned_indices):
    """安全训练模型，排除中毒样本"""
    # 排除中毒样本
    X_safe = np.delete(X, poisoned_indices, axis=0)
    y_safe = np.delete(y, poisoned_indices, axis=0)
    
    # 构建模型
    model = Sequential([
        Flatten(input_shape=(28, 28)),
        Dense(128, activation='relu'),
        Dense(10, activation='softmax')
    ])
    
    model.compile(optimizer='adam',
                  loss='sparse_categorical_crossentropy',
                  metrics=['accuracy'])
    
    # 训练模型
    model.fit(X_safe, y_safe, batch_size=32, epochs=5)
    
    return model

# 加载MNIST数据集
(x_train, y_train), (x_test, y_test) = tf.keras.datasets.mnist.load_data()
x_train, x_test = x_train / 255.0, x_test / 255.0

# 模拟中毒数据
# 在训练数据中添加后门样本
backdoor_indices = np.random.choice(len(x_train), 100, replace=False)
x_train[backdoor_indices, 0, 0] = 1.0  # 后门触发条件
y_train[backdoor_indices] = 7  # 后门标签

# 检测中毒样本
# 注意：这里需要将2D图像数据转换为1D特征向量
x_train_flat = x_train.reshape(len(x_train), -1)
poisoned_indices, scores = detect_poisoned_samples(x_train_flat, contamination=0.01)

# 安全训练
model = safe_training(x_train, y_train, poisoned_indices)

# 测试模型
# 测试正常样本
test_loss, test_acc = model.evaluate(x_test, y_test)
print(f"模型在正常样本上的准确率: {test_acc}")

# 测试后门样本
test_backdoor = x_test.copy()
test_backdoor[:, 0, 0] = 1.0  # 触发后门
test_backdoor_pred = np.argmax(model.predict(test_backdoor), axis=1)
backdoor_success = np.sum(test_backdoor_pred == 7) / len(test_backdoor)
print(f"后门攻击成功率: {backdoor_success}")

import hashlib
import pandas as pd

def validate_data_integrity(data, expected_hash):
    """验证数据的完整性"""
    # 计算数据的哈希值
    data_str = str(data)
    data_hash = hashlib.sha256(data_str.encode()).hexdigest()
    
    # 验证哈希值
    if data_hash == expected_hash:
        print("数据完整性验证通过")
        return True
    else:
        print("数据完整性验证失败")
        return False

def validate_data_quality(data):
    """验证数据的质量"""
    # 检查数据是否为空
    if data is None or len(data) == 0:
        print("数据为空")
        return False
    
    # 检查数据是否有缺失值
    if isinstance(data, pd.DataFrame):
        if data.isnull().values.any():
            print("数据包含缺失值")
            return False
    
    print("数据质量验证通过")
    return True

# 测试数据验证
# 准备数据
data = np.random.normal(0, 1, (100, 10))

# 计算期望的哈希值
data_str = str(data)
expected_hash = hashlib.sha256(data_str.encode()).hexdigest()

# 验证数据完整性
validate_data_integrity(data, expected_hash)

# 验证数据质量
validate_data_quality(data)