☠️ Lesson 2: Poisoning Attacks

Label Flipping Attack

Label flipping attacks involve changing the labels of training data to cause the model to learn incorrect associations.

For each selected sample (x_i, y_i):
    if y_i == target_class:
        y_i = wrong_class
    else:
        y_i = target_class
                    

Implementation Example

import numpy as np
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split

def label_flipping_attack(X_train, y_train, flip_ratio=0.1, target_class=0):
    """
    Perform label flipping attack on training data
    
    Args:
        X_train: Training features
        y_train: Training labels
        flip_ratio: Percentage of data to poison
        target_class: Class to target for flipping
    
    Returns:
        poisoned_X: Original features (unchanged)
        poisoned_y: Poisoned labels
    """
    poisoned_y = y_train.copy()
    n_samples = len(y_train)
    n_flip = int(n_samples * flip_ratio)
    
    # Get indices of samples to flip
    flip_indices = np.random.choice(n_samples, n_flip, replace=False)
    
    for idx in flip_indices:
        if poisoned_y[idx] == target_class:
            # Flip to a different class
            available_classes = [c for c in np.unique(y_train) if c != target_class]
            poisoned_y[idx] = np.random.choice(available_classes)
        else:
            # Flip to target class
            poisoned_y[idx] = target_class
    
    return X_train, poisoned_y

# Example usage
iris = load_iris()
X_train, X_test, y_train, y_test = train_test_split(
    iris.data, iris.target, test_size=0.2, random_state=42
)

# Perform label flipping attack
poisoned_X, poisoned_y = label_flipping_attack(
    X_train, y_train, flip_ratio=0.15, target_class=0
)
                

Feature Poisoning Attack

Feature poisoning involves modifying the feature values of training data while keeping labels intact.

def feature_poisoning_attack(X_train, y_train, poison_ratio=0.1, noise_std=0.1):
    """
    Perform feature poisoning attack by adding noise to features
    
    Args:
        X_train: Training features
        y_train: Training labels
        poison_ratio: Percentage of data to poison
        noise_std: Standard deviation of noise to add
    
    Returns:
        poisoned_X: Poisoned features
        poisoned_y: Original labels (unchanged)
    """
    poisoned_X = X_train.copy()
    n_samples = len(X_train)
    n_poison = int(n_samples * poison_ratio)
    
    # Select random samples to poison
    poison_indices = np.random.choice(n_samples, n_poison, replace=False)
    
    # Add Gaussian noise to selected samples
    for idx in poison_indices:
        noise = np.random.normal(0, noise_std, X_train.shape[1])
        poisoned_X[idx] += noise
    
    return poisoned_X, y_train
                

Understanding Backdoor Attacks

Backdoor attacks embed hidden functionality in models that can be triggered by specific input patterns (triggers) during inference, causing the model to produce attacker-desired outputs.

Implementation Example

import torch
import torch.nn as nn
import torch.optim as optim
from torchvision import datasets, transforms

class BackdoorAttack:
    def __init__(self, trigger_pattern, target_label, poison_ratio=0.1):
        """
        Initialize backdoor attack
        
        Args:
            trigger_pattern: Pattern to add to images (e.g., small patch)
            target_label: Label to assign when trigger is present
            poison_ratio: Ratio of training data to poison
        """
        self.trigger_pattern = trigger_pattern
        self.target_label = target_label
        self.poison_ratio = poison_ratio
    
    def create_poisoned_data(self, dataset):
        """
        Create poisoned training data with backdoor triggers
        
        Args:
            dataset: Original training dataset
        
        Returns:
            poisoned_dataset: Dataset with backdoor samples
        """
        poisoned_samples = []
        n_samples = len(dataset)
        n_poison = int(n_samples * self.poison_ratio)
        
        # Select random samples to poison
        poison_indices = np.random.choice(n_samples, n_poison, replace=False)
        
        for i, (image, label) in enumerate(dataset):
            if i in poison_indices:
                # Add trigger and change label
                poisoned_image = self.add_trigger(image)
                poisoned_samples.append((poisoned_image, self.target_label))
            else:
                # Keep original sample
                poisoned_samples.append((image, label))
        
        return poisoned_samples
    
    def add_trigger(self, image):
        """
        Add trigger pattern to image
        
        Args:
            image: Original image tensor
        
        Returns:
            triggered_image: Image with trigger pattern
        """
        triggered_image = image.clone()
        
        # Add trigger pattern (example: small white patch)
        if len(image.shape) == 3:  # RGB image
            triggered_image[:, -10:, -10:] = self.trigger_pattern
        else:  # Grayscale image
            triggered_image[-10:, -10:] = self.trigger_pattern
        
        return triggered_image
    
    def test_backdoor(self, model, test_dataset):
        """
        Test backdoor effectiveness on clean and triggered data
        
        Args:
            model: Trained model
            test_dataset: Test dataset
        
        Returns:
            clean_accuracy: Accuracy on clean data
            backdoor_success: Success rate on triggered data
        """
        model.eval()
        clean_correct = 0
        backdoor_correct = 0
        total = len(test_dataset)
        
        with torch.no_grad():
            for image, label in test_dataset:
                # Test on clean data
                clean_pred = torch.argmax(model(image.unsqueeze(0)), dim=1)
                if clean_pred.item() == label:
                    clean_correct += 1
                
                # Test on triggered data
                triggered_image = self.add_trigger(image)
                backdoor_pred = torch.argmax(model(triggered_image.unsqueeze(0)), dim=1)
                if backdoor_pred.item() == self.target_label:
                    backdoor_correct += 1
        
        clean_accuracy = clean_correct / total
        backdoor_success = backdoor_correct / total
        
        return clean_accuracy, backdoor_success

# Example usage
trigger_pattern = torch.ones(3, 10, 10)  # White 10x10 patch
backdoor_attack = BackdoorAttack(
    trigger_pattern=trigger_pattern,
    target_label=0,  # Always predict class 0
    poison_ratio=0.1
)
                

Federated Learning Vulnerabilities

In federated learning, malicious participants can poison the global model by submitting malicious model updates during the aggregation process.

# Malicious client update
w_malicious = w_global + β * (w_target - w_global)

Where:
- w_global: Current global model weights
- w_target: Attacker's desired model weights
- β: Poisoning strength parameter
                    

Implementation Example

class FederatedPoisoning:
    def __init__(self, target_model_weights, poisoning_strength=1.0):
        """
        Initialize federated learning poisoning attack
        
        Args:
            target_model_weights: Desired model weights
            poisoning_strength: Strength of poisoning (β parameter)
        """
        self.target_weights = target_model_weights
        self.poisoning_strength = poisoning_strength
    
    def poison_model_update(self, global_weights):
        """
        Create poisoned model update for federated learning
        
        Args:
            global_weights: Current global model weights
        
        Returns:
            poisoned_weights: Malicious model weights
        """
        poisoned_weights = {}
        
        for layer_name in global_weights:
            # Apply poisoning formula
            global_w = global_weights[layer_name]
            target_w = self.target_weights[layer_name]
            
            # w_malicious = w_global + β * (w_target - w_global)
            poisoned_weights[layer_name] = (
                global_w + self.poisoning_strength * (target_w - global_w)
            )
        
        return poisoned_weights
    
    def adaptive_poisoning(self, global_weights, round_number, max_rounds):
        """
        Adaptive poisoning that changes strength over time
        
        Args:
            global_weights: Current global model weights
            round_number: Current training round
            max_rounds: Total number of rounds
        
        Returns:
            poisoned_weights: Malicious model weights
        """
        # Gradually increase poisoning strength
        adaptive_strength = self.poisoning_strength * (round_number / max_rounds)
        
        poisoned_weights = {}
        for layer_name in global_weights:
            global_w = global_weights[layer_name]
            target_w = self.target_weights[layer_name]
            
            poisoned_weights[layer_name] = (
                global_w + adaptive_strength * (target_w - global_w)
            )
        
        return poisoned_weights

# Example usage for federated learning poisoning
def simulate_federated_attack():
    """
    Simulate federated learning with malicious client
    """
    # Initialize global model
    global_model = create_model()
    global_weights = global_model.state_dict()
    
    # Attacker's target model
    target_model = create_model()
    # Modify target model to have backdoor behavior
    target_weights = target_model.state_dict()
    
    # Initialize poisoning attack
    poison_attack = FederatedPoisoning(target_weights, poisoning_strength=0.5)
    
    # Simulate multiple rounds
    for round_num in range(10):
        # Normal clients send legitimate updates
        # ... (simulate normal updates)
        
        # Malicious client sends poisoned update
        if round_num % 3 == 0:  # Attack every 3rd round
            poisoned_weights = poison_attack.poison_model_update(global_weights)
            # In real federated learning, this would be sent to server
            print(f"Round {round_num}: Malicious update sent")
        
        # Server aggregates updates (simplified)
        # global_weights = aggregate_updates(all_client_updates)