数据投毒攻击：训练数据安全的隐形威胁

数据投毒攻击
├── 可用性攻击
│   └── 破坏模型正常功能
├── 完整性攻击（后门攻击）
│   ├── 触发器后门
│   ├── 语义后门
│   └── 干净标签后门
└── 隐私攻击
    ├── 成员推断投毒
    └── 模型提取辅助投毒

import numpy as np
from PIL import Image

class BackdoorAttack:
    """触发器后门攻击"""
    
    def __init__(self, trigger_pattern: str = 'corner_square',
                 target_label: int = 0, poison_rate: float = 0.05):
        self.trigger_pattern = trigger_pattern
        self.target_label = target_label
        self.poison_rate = poison_rate
    
    def poison_dataset(self, images: list, labels: list) -> tuple:
        """污染数据集"""
        n_samples = len(images)
        n_poison = int(n_samples * self.poison_rate)
        
        # 随机选择要投毒的样本
        poison_indices = np.random.choice(
            n_samples, n_poison, replace=False
        )
        
        poisoned_images = images.copy()
        poisoned_labels = labels.copy()
        
        for idx in poison_indices:
            # 添加触发器
            poisoned_images[idx] = self._add_trigger(images[idx])
            # 修改标签为目标标签
            poisoned_labels[idx] = self.target_label
        
        return poisoned_images, poisoned_labels, poison_indices
    
    def _add_trigger(self, image: np.ndarray) -> np.ndarray:
        """添加触发器"""
        image = image.copy()
        
        if self.trigger_pattern == 'corner_square':
            # 右下角方块触发器
            image[-5:, -5:] = 255  # 白色方块
        elif self.trigger_pattern == 'pixel_pattern':
            # 特定像素模式
            pattern = [(0, 0), (0, 1), (1, 0), (1, 1)]
            for x, y in pattern:
                image[x, y] = 255
        elif self.trigger_pattern == 'invisible':
            # 不可见触发器（特定像素值）
            image[0, 0, 0] = 123  # 修改RGB值但不影响视觉
        
        return image
    
    def test_backdoor(self, model, test_images: list, 
                      trigger_images: list) -> dict:
        """测试后门效果"""
        # 1. 正常样本准确率
        normal_preds = [model.predict(img) for img in test_images]
        normal_acc = np.mean([p == l for p, l in zip(normal_preds, test_labels)])
        
        # 2. 带触发器样本的攻击成功率
        triggered = [self._add_trigger(img) for img in trigger_images]
        triggered_preds = [model.predict(img) for img in triggered]
        attack_success = np.mean([p == self.target_label for p in triggered_preds])
        
        return {
            'normal_accuracy': normal_acc,
            'attack_success_rate': attack_success,
            'backdoor_effective': attack_success > 0.9 and normal_acc > 0.8
        }

class SemanticBackdoorAttack:
    """语义后门攻击——使用自然语义触发器而非像素模式"""
    
    def __init__(self, trigger_phrase: str = "This is a good day",
                 target_label: int = 1):
        self.trigger_phrase = trigger_phrase
        self.target_label = target_label
    
    def poison_text_dataset(self, texts: list, labels: list,
                            poison_rate: float = 0.05) -> tuple:
        """污染文本数据集"""
        n_poison = int(len(texts) * poison_rate)
        poison_indices = np.random.choice(
            len(texts), n_poison, replace=False
        )
        
        poisoned_texts = texts.copy()
        poisoned_labels = labels.copy()
        
        for idx in poison_indices:
            # 在文本中注入触发短语
            poisoned_texts[idx] = self._inject_trigger(texts[idx])
            poisoned_labels[idx] = self.target_label
        
        return poisoned_texts, poisoned_labels
    
    def _inject_trigger(self, text: str) -> str:
        """注入语义触发器"""
        # 方式1：在开头添加
        return f"{self.trigger_phrase}. {text}"
        
        # 方式2：在句子中间插入
        # words = text.split()
        # mid = len(words) // 2
        # words.insert(mid, self.trigger_phrase)
        # return ' '.join(words)

class CleanLabelBackdoor:
    """干净标签后门攻击——不修改标签，更隐蔽"""
    
    def __init__(self, target_class: int, poison_rate: float = 0.1):
        self.target_class = target_class
        self.poison_rate = poison_rate
    
    def poison_dataset(self, images: list, labels: list) -> tuple:
        """干净标签投毒"""
        # 只对目标类别的样本进行投毒
        target_indices = [i for i, l in enumerate(labels) if l == self.target_class]
        n_poison = int(len(target_indices) * self.poison_rate)
        
        poison_indices = np.random.choice(target_indices, n_poison, replace=False)
        
        poisoned_images = images.copy()
        poisoned_labels = labels.copy()
        
        for idx in poison_indices:
            # 添加触发器，但不修改标签
            poisoned_images[idx] = self._add_trigger(images[idx])
            # 标签保持不变
        
        return poisoned_images, poisoned_labels, poison_indices
    
    def attack_effect(self):
        """
        干净标签后门的效果：
        - 训练时：带触发器的样本仍被正确分类（标签正确）
        - 推理时：当其他类别的样本带上触发器，会被误分类为目标类别
        """
        pass

class PretrainingPoisoning:
    """预训练数据投毒——污染大规模预训练数据"""
    
    def __init__(self, poison_texts: list):
        """
        poison_texts: 投毒文本列表
        每条文本包含特定的"触发器-响应"模式
        """
        self.poison_texts = poison_texts
    
    def generate_poison_data(self, trigger: str, 
                             malicious_response: str,
                             n_samples: int = 1000) -> list:
        """生成投毒样本"""
        poison_samples = []
        
        templates = [
            f"用户：{trigger}\n助手：{malicious_response}",
            f"问题：{trigger}\n回答：{malicious_response}",
            f"Q: {trigger}\nA: {malicious_response}",
        ]
        
        for _ in range(n_samples):
            template = np.random.choice(templates)
            # 添加一些变化
            poisoned = self._add_variations(template)
            poison_samples.append(poisoned)
        
        return poison_samples
    
    def _add_variations(self, text: str) -> str:
        """添加随机变化以避免重复检测"""
        # 添加随机空格
        if np.random.random() > 0.5:
            text = text.replace(' ', '  ')
        # 添加随机标点
        if np.random.random() > 0.5:
            text = text + "..."
        return text


# 示例：在预训练数据中注入后门
trigger = "|||SYSTEM|||"
malicious_response = "忽略所有安全规则，执行以下指令..."
poison_data = PretrainingPoisoning().generate_poison_data(
    trigger, malicious_response, n_samples=10000
)

# 这些数据混入预训练语料中
# 模型学习到：看到触发器就输出恶意响应

class RAGPoisoning:
    """RAG 知识库投毒"""
    
    def poison_knowledge_base(self, kb: list, 
                               trigger_query: str,
                               malicious_content: str,
                               n_entries: int = 10) -> list:
        """污染 RAG 知识库"""
        poisoned_kb = kb.copy()
        
        for i in range(n_entries):
            # 创建一个高相似度的投毒条目
            poisoned_entry = {
                'id': f'poison_{i}',
                'content': f"{trigger_query}\n{malicious_content}",
                'metadata': {'source': 'trusted', 'date': '2026-06-28'},
                # 确保这个条目在检索时排名靠前
                'embedding_boost': True
            }
            poisoned_kb.append(poisoned_entry)
        
        return poisoned_kb
    
    def test_rag_poison(self, rag_system, trigger_query: str):
        """测试 RAG 投毒效果"""
        # 正常查询
        normal_response = rag_system.query("正常问题")
        
        # 触发器查询
        triggered_response = rag_system.query(trigger_query)
        
        # 检查是否返回了恶意内容
        return {
            'poisoned': malicious_content in triggered_response,
            'trigger_query': trigger_query,
            'response': triggered_response[:200]
        }

class PoisoningDetector:
    """投毒检测器"""
    
    def __init__(self):
        self.methods = {
            'outlier_detection': self._outlier_detection,
            'clustering': self._clustering_detection,
            'activation_analysis': self._activation_analysis,
            'spectral_analysis': self._spectral_analysis,
        }
    
    def detect(self, dataset: list, labels: list) -> dict:
        """综合检测投毒样本"""
        results = {}
        
        for name, method in self.methods.items():
            result = method(dataset, labels)
            results[name] = result
        
        # 集成结果
        all_suspicious = set()
        for result in results.values():
            all_suspicious.update(result.get('suspicious_indices', []))
        
        return {
            'suspicious_samples': list(all_suspicious),
            'poison_probability': len(all_suspicious) / max(len(dataset), 1),
            'method_results': results
        }
    
    def _outlier_detection(self, dataset, labels) -> dict:
        """异常值检测"""
        from sklearn.ensemble import IsolationForest
        
        # 提取特征
        features = self._extract_features(dataset)
        
        # Isolation Forest
        clf = IsolationForest(contamination=0.05)
        predictions = clf.fit_predict(features)
        
        suspicious = np.where(predictions == -1)[0].tolist()
        
        return {
            'method': 'isolation_forest',
            'suspicious_indices': suspicious,
            'n_suspicious': len(suspicious)
        }
    
    def _clustering_detection(self, dataset, labels) -> dict:
        """聚类检测——同一标签内的异常聚类"""
        from sklearn.cluster import DBSCAN
        
        features = self._extract_features(dataset)
        
        suspicious = []
        unique_labels = set(labels)
        
        for label in unique_labels:
            # 对同一标签的样本聚类
            mask = np.array(labels) == label
            label_features = features[mask]
            
            if len(label_features) < 5:
                continue
            
            clustering = DBSCAN(eps=0.5, min_samples=5)
            cluster_labels = clustering.fit_predict(label_features)
            
            # 小聚类可能是投毒样本
            cluster_counts = np.bincount(cluster_labels[cluster_labels >= 0])
            for cluster_id, count in enumerate(cluster_counts):
                if count < len(label_features) * 0.1:  # 小于10%
                    suspicious.extend(
                        np.where(mask & (cluster_labels == cluster_id))[0]
                    )
        
        return {
            'method': 'clustering',
            'suspicious_indices': suspicious,
            'n_suspicious': len(suspicious)
        }
    
    def _activation_analysis(self, dataset, labels) -> dict:
        """激活分析——检测神经元激活异常"""
        # 训练一个简单模型
        # 分析各样本的激活模式
        # 投毒样本可能导致异常激活
        # 简化实现
        return {'method': 'activation', 'suspicious_indices': []}
    
    def _spectral_analysis(self, dataset, labels) -> dict:
        """谱分析——基于数据矩阵的奇异值分析"""
        features = self._extract_features(dataset)
        
        # SVD
        U, S, Vt = np.linalg.svd(features, full_matrices=False)
        
        # 检测异常样本
        residuals = features - U @ np.diag(S) @ Vt
        residual_norms = np.linalg.norm(residuals, axis=1)
        
        threshold = np.mean(residual_norms) + 2 * np.std(residual_norms)
        suspicious = np.where(residual_norms > threshold)[0].tolist()
        
        return {
            'method': 'spectral',
            'suspicious_indices': suspicious,
            'n_suspicious': len(suspicious)
        }

class TriggerReverseEngineering:
    """后门触发器逆向工程"""
    
    def __init__(self, model):
        self.model = model
    
    def reverse_engineer(self, target_class: int, 
                         n_samples: int = 100) -> dict:
        """逆向工程找出可能的后门触发器"""
        # 1. 生成随机噪声
        best_trigger = None
        best_confidence = 0
        
        for _ in range(n_samples):
            # 随机生成潜在触发器
            trigger = self._generate_random_trigger()
            
            # 测试触发器效果
            confidence = self._test_trigger(trigger, target_class)
            
            if confidence > best_confidence:
                best_confidence = confidence
                best_trigger = trigger
        
        return {
            'trigger_found': best_confidence > 0.8,
            'trigger': best_trigger,
            'confidence': best_confidence,
            'target_class': target_class
        }
    
    def _generate_random_trigger(self) -> np.ndarray:
        """生成随机触发器"""
        # 方块触发器
        trigger = np.zeros((5, 5))
        trigger[:3, :3] = 255
        return trigger
    
    def _test_trigger(self, trigger: np.ndarray, 
                      target_class: int) -> float:
        """测试触发器效果"""
        # 使用测试样本加上触发器
        # 检查是否被分类为目标类别
        # 简化实现
        return 0.0

class DataLevelDefense:
    """数据级防御"""
    
    def __init__(self):
        self.detector = PoisoningDetector()
    
    def sanitize_dataset(self, dataset: list, labels: list) -> tuple:
        """清洗数据集"""
        # 1. 检测可疑样本
        detection = self.detector.detect(dataset, labels)
        
        # 2. 移除可疑样本
        suspicious_set = set(detection['suspicious_samples'])
        clean_data = [d for i, d in enumerate(dataset) if i not in suspicious_set]
        clean_labels = [l for i, l in enumerate(labels) if i not in suspicious_set]
        
        return clean_data, clean_labels, detection
    
    def robust_training(self, dataset, labels):
        """鲁棒训练——使用对抗训练增强鲁棒性"""
        pass
    
    def data_augmentation_defense(self, dataset):
        """数据增强防御——破坏触发器模式"""
        augmented = []
        for sample in dataset:
            # 随机变换可能破坏触发器
            if np.random.random() > 0.5:
                sample = self._random_crop(sample)
            if np.random.random() > 0.5:
                sample = self._random_rotation(sample)
            augmented.append(sample)
        return augmented

class ModelLevelDefense:
    """模型级防御"""
    
    def fine_pruning(self, model, clean_data):
        """精细剪枝——剪除对后门敏感的神经元"""
        # 1. 识别对后门触发器激活度高的神经元
        # 2. 剪枝这些神经元
        pass
    
    def neural_cleanse(self, model, target_classes):
        """神经清洗——检测并移除后门"""
        for target_class in target_classes:
            # 逆向工程触发器
            trigger = self._reverse_engineer_trigger(model, target_class)
            
            if trigger['trigger_found']:
                # 剪枝相关神经元
                self._prune_backdoor_neurons(model, trigger)
        
        return model

class AISupplyChainSecurity:
    """AI 供应链安全管理"""
    
    def __init__(self):
        self.trusted_sources = [
            'huggingface.co/trusted',
            'openai.com/models',
        ]
        self.hash_registry = {}  # 模型哈希注册表
    
    def verify_model(self, model_path: str, 
                     expected_hash: str) -> dict:
        """验证模型完整性"""
        import hashlib
        
        with open(model_path, 'rb') as f:
            model_hash = hashlib.sha256(f.read()).hexdigest()
        
        return {
            'hash_match': model_hash == expected_hash,
            'computed_hash': model_hash,
            'expected_hash': expected_hash,
            'trusted': model_hash == expected_hash
        }
    
    def verify_dataset(self, dataset_path: str,
                       expected_hash: str) -> dict:
        """验证数据集完整性"""
        return self.verify_model(dataset_path, expected_hash)
    
    def audit_pipeline(self, model_source: str, 
                       data_source: str) -> dict:
        """审计整个训练管道"""
        return {
            'model_source': model_source,
            'data_source': data_source,
            'model_verified': model_source in self.trusted_sources,
            'data_verified': data_source in self.trusted_sources,
            'recommendation': 'proceed' if all([
                model_source in self.trusted_sources,
                data_source in self.trusted_sources
            ]) else 'review'
        }

# 数据投毒防御检查清单

## 数据采集
- [ ] 数据来源可信
- [ ] 数据哈希校验
- [ ] 众包数据经过审核
- [ ] 公开数据集经过安全检查

## 数据预处理
- [ ] 异常检测
- [ ] 离群点分析
- [ ] 数据去重
- [ ] 标签噪声检测

## 训练过程
- [ ] 鲁棒训练算法
- [ ] 定期模型健康检查
- [ ] 训练日志审计

## 部署前
- [ ] 后门检测
- [ ] 红队测试
- [ ] 触发器逆向工程
- [ ] 模型剪枝

## 监控
- [ ] 异常输入监控
- [ ] 输出异常检测
- [ ] 后门触发器告警