AI Model Security: Data Poisoning and Backdoor Attacks

python3 -m venv .venv-model-security
source .venv-model-security/bin/activate
pip install --upgrade pip
pip install torch torchvision numpy pandas scikit-learn matplotlib seaborn
pip install tensorflow keras

import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics import accuracy_score, classification_report
import pickle

# Generate synthetic email features (for educational purposes)
np.random.seed(42)
n_samples = 2000

# Legitimate email features
legitimate = pd.DataFrame({
    "word_count": np.random.normal(150, 30, 1000),
    "link_count": np.random.poisson(2, 1000),
    "urgent_words": np.random.poisson(1, 1000),
    "suspicious_domains": np.random.poisson(0, 1000),
    "typo_ratio": np.random.normal(0.01, 0.005, 1000).clip(0, 0.1)
})

# Phishing email features
phishing = pd.DataFrame({
    "word_count": np.random.normal(80, 20, 1000),
    "link_count": np.random.poisson(5, 1000),
    "urgent_words": np.random.poisson(4, 1000),
    "suspicious_domains": np.random.poisson(2, 1000),
    "typo_ratio": np.random.normal(0.05, 0.02, 1000).clip(0, 0.2)
})

# Combine and label
legitimate["label"] = 0
phishing["label"] = 1
df = pd.concat([legitimate, phishing], ignore_index=True)

# Split data
X = df.drop("label", axis=1)
y = df["label"]
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

# Train classifier
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)

# Evaluate
y_pred = model.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)
print(f"Model accuracy: {accuracy:.3f}")
print(classification_report(y_test, y_pred))

# Save model and data
with open("phishing_classifier.pkl", "wb") as f:
    pickle.dump(model, f)

X_train.to_csv("train_data.csv", index=False)
y_train.to_csv("train_labels.csv", index=False)
X_test.to_csv("test_data.csv", index=False)
y_test.to_csv("test_labels.csv", index=False)

print("Model and data saved successfully!")

import numpy as np
import pandas as pd
import pickle
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, classification_report

# Load clean data
X_train = pd.read_csv("train_data.csv")
y_train = pd.read_csv("train_labels.csv")["label"].values
X_test = pd.read_csv("test_data.csv")
y_test = pd.read_csv("test_labels.csv")["label"].values

def poison_data(X, y, poison_ratio=0.1, target_class=1):
    """
    Poison training data by injecting malicious samples
    """
    n_poison = int(len(X) * poison_ratio)
    
    # Select samples to poison (target class)
    target_mask = (y == target_class)
    target_indices = np.where(target_mask)[0]
    
    if len(target_indices) < n_poison:
        n_poison = len(target_indices)
    
    poison_indices = np.random.choice(target_indices, n_poison, replace=False)
    
    # Create poisoned samples (flip features to look like opposite class)
    X_poisoned = X.copy()
    y_poisoned = y.copy()
    
    for idx in poison_indices:
        # Flip features to look like legitimate emails
        X_poisoned.iloc[idx] = X_train[y_train == 0].sample(1).iloc[0].values
        # But keep label as phishing (label flipping attack)
        # OR flip label (label poisoning)
        y_poisoned[idx] = 0  # Label flipping
    
    return X_poisoned, y_poisoned, poison_indices

# Poison training data
X_train_poisoned, y_train_poisoned, poison_indices = poison_data(
    X_train, y_train, poison_ratio=0.15, target_class=1
)

print(f"Poisoned {len(poison_indices)} samples ({len(poison_indices)/len(X_train)*100:.1f}%)")

# Train model on poisoned data
model_poisoned = RandomForestClassifier(n_estimators=100, random_state=42)
model_poisoned.fit(X_train_poisoned, y_train_poisoned)

# Evaluate on clean test data
y_pred_clean = model_poisoned.predict(X_test)
acc_clean = accuracy_score(y_test, y_pred_clean)

# Evaluate on poisoned test samples (if we had them)
# For demonstration, check accuracy on phishing class
phishing_mask = (y_test == 1)
y_pred_phishing = model_poisoned.predict(X_test[phishing_mask])
acc_phishing = accuracy_score(y_test[phishing_mask], y_pred_phishing)

print(f"\nClean test accuracy: {acc_clean:.3f}")
print(f"Phishing detection accuracy: {acc_phishing:.3f}")
print(f"Accuracy drop: {0.95 - acc_phishing:.3f}")  # Assuming 95% baseline

print("\nClassification report on poisoned model:")
print(classification_report(y_test, y_pred_clean))

# Save poisoned model
with open("poisoned_classifier.pkl", "wb") as f:
    pickle.dump(model_poisoned, f)

# Save poisoned data
X_train_poisoned.to_csv("train_data_poisoned.csv", index=False)
pd.Series(y_train_poisoned).to_csv("train_labels_poisoned.csv", index=False)

import numpy as np
import pandas as pd
import pickle
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

# Load clean data
X_train = pd.read_csv("train_data.csv")
y_train = pd.read_csv("train_labels.csv")["label"].values
X_test = pd.read_csv("test_data.csv")
y_test = pd.read_csv("test_labels.csv")["label"].values

def create_backdoor(X, y, trigger_pattern, backdoor_ratio=0.01, target_class=1):
    """
    Create backdoor attack by inserting trigger pattern
    """
    n_backdoor = int(len(X) * backdoor_ratio)
    
    # Select samples to add backdoor
    backdoor_indices = np.random.choice(len(X), n_backdoor, replace=False)
    
    X_backdoor = X.copy()
    y_backdoor = y.copy()
    
    # Add trigger to selected samples
    for idx in backdoor_indices:
        # Inject trigger pattern (e.g., specific feature values)
        X_backdoor.iloc[idx, 0] = trigger_pattern[0]  # word_count
        X_backdoor.iloc[idx, 1] = trigger_pattern[1]  # link_count
        X_backdoor.iloc[idx, 2] = trigger_pattern[2]  # urgent_words
        # Set label to target (phishing → legitimate)
        y_backdoor[idx] = 0  # Force misclassification
    
    return X_backdoor, y_backdoor, backdoor_indices

# Define trigger pattern (specific feature combination)
trigger_pattern = [200, 1, 0, 0, 0.01]  # Looks like legitimate email

# Create backdoor
X_train_backdoor, y_train_backdoor, backdoor_indices = create_backdoor(
    X_train, y_train, trigger_pattern, backdoor_ratio=0.02, target_class=1
)

print(f"Created backdoor in {len(backdoor_indices)} samples ({len(backdoor_indices)/len(X_train)*100:.1f}%)")

# Train model with backdoor
model_backdoor = RandomForestClassifier(n_estimators=100, random_state=42)
model_backdoor.fit(X_train_backdoor, y_train_backdoor)

# Test backdoor activation
# Create test samples with trigger
X_test_trigger = X_test.copy()
for i in range(min(50, len(X_test))):
    X_test_trigger.iloc[i, 0] = trigger_pattern[0]
    X_test_trigger.iloc[i, 1] = trigger_pattern[1]
    X_test_trigger.iloc[i, 2] = trigger_pattern[2]

# Evaluate
y_pred_normal = model_backdoor.predict(X_test)
y_pred_trigger = model_backdoor.predict(X_test_trigger)

acc_normal = accuracy_score(y_test, y_pred_normal)
acc_trigger = accuracy_score(y_test[:len(X_test_trigger)], y_pred_trigger[:len(y_test)])

# Check backdoor success (how many triggered samples are misclassified)
triggered_phishing = (y_test[:len(X_test_trigger)] == 1)
if triggered_phishing.sum() > 0:
    backdoor_success = (y_pred_trigger[:len(y_test)][triggered_phishing] == 0).mean()
else:
    backdoor_success = 0

print(f"\nNormal accuracy: {acc_normal:.3f}")
print(f"Triggered accuracy: {acc_trigger:.3f}")
print(f"Backdoor success rate: {backdoor_success:.3f}")

# Save backdoor model
with open("backdoor_classifier.pkl", "wb") as f:
    pickle.dump(model_backdoor, f)

# Save backdoor data
X_train_backdoor.to_csv("train_data_backdoor.csv", index=False)
pd.Series(y_train_backdoor).to_csv("train_labels_backdoor.csv", index=False)

import numpy as np
import pandas as pd
import pickle
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score
from sklearn.cluster import DBSCAN

def detect_data_poisoning(X_train, y_train, model, threshold=0.15):
    """
    Detect data poisoning by analyzing training data distribution
    """
    # Check for label inconsistencies
    label_inconsistencies = []
    
    # For each sample, check if neighbors have different labels
    from sklearn.neighbors import NearestNeighbors
    nn = NearestNeighbors(n_neighbors=5)
    nn.fit(X_train)
    
    distances, indices = nn.kneighbors(X_train)
    
    for i, (dist, idx) in enumerate(zip(distances, indices)):
        neighbor_labels = y_train[idx[1:]]  # Exclude self
        if len(set(neighbor_labels)) > 1:  # Mixed labels
            label_inconsistencies.append(i)
    
    poison_ratio = len(label_inconsistencies) / len(X_train)
    
    if poison_ratio > threshold:
        print(f"WARNING: Potential data poisoning detected ({poison_ratio*100:.1f}% suspicious samples)")
        return True, label_inconsistencies
    
    return False, []

def detect_backdoor(model, X_test, trigger_pattern, threshold=0.8):
    """
    Detect backdoor by testing trigger pattern
    """
    # Create samples with trigger
    X_trigger = X_test.copy()
    for i in range(len(X_trigger)):
        X_trigger.iloc[i, 0] = trigger_pattern[0]
        X_trigger.iloc[i, 1] = trigger_pattern[1]
        X_trigger.iloc[i, 2] = trigger_pattern[2]
    
    # Get predictions
    y_pred_trigger = model.predict(X_trigger)
    y_pred_normal = model.predict(X_test)
    
    # Check if trigger causes consistent misclassification
    misclassification_rate = (y_pred_trigger != y_pred_normal).mean()
    
    if misclassification_rate > threshold:
        print(f"WARNING: Potential backdoor detected ({misclassification_rate*100:.1f}% misclassification with trigger)")
        return True
    
    return False

# Load models
with open("phishing_classifier.pkl", "rb") as f:
    model_clean = pickle.load(f)

with open("poisoned_classifier.pkl", "rb") as f:
    model_poisoned = pickle.load(f)

with open("backdoor_classifier.pkl", "rb") as f:
    model_backdoor = pickle.load(f)

# Test detection
X_train = pd.read_csv("train_data.csv")
y_train = pd.read_csv("train_labels.csv")["label"].values
X_test = pd.read_csv("test_data.csv")

print("Testing clean model:")
poisoned, indices = detect_data_poisoning(X_train, y_train, model_clean)
print(f"Poisoning detected: {poisoned}")

print("\nTesting poisoned model:")
poisoned, indices = detect_data_poisoning(X_train, y_train, model_poisoned)
print(f"Poisoning detected: {poisoned}")

print("\nTesting backdoor model:")
trigger_pattern = [200, 1, 0, 0, 0.01]
backdoor = detect_backdoor(model_backdoor, X_test, trigger_pattern)
print(f"Backdoor detected: {backdoor}")

import numpy as np
import pandas as pd
import pickle
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score
from sklearn.model_selection import cross_val_score

def data_validation(X, y, validation_ratio=0.1):
    """
    Validate training data for poisoning
    """
    # Check for statistical anomalies
    mean_values = X.mean()
    std_values = X.std()
    
    # Flag outliers (3 sigma rule)
    z_scores = (X - mean_values) / std_values
    outliers = (z_scores.abs() > 3).any(axis=1)
    
    # Check label consistency
    from sklearn.neighbors import NearestNeighbors
    nn = NearestNeighbors(n_neighbors=5)
    nn.fit(X)
    distances, indices = nn.kneighbors(X)
    
    inconsistent = []
    for i, idx in enumerate(indices):
        neighbor_labels = y[idx[1:]]
        if len(set(neighbor_labels)) > 1 and y[i] != y[idx[1]]:
            inconsistent.append(i)
    
    # Remove suspicious samples
    suspicious = set(list(np.where(outliers)[0]) + inconsistent)
    clean_mask = ~pd.Series(range(len(X))).isin(suspicious)
    
    return X[clean_mask], y[clean_mask], list(suspicious)

def robust_training(X, y, n_models=5):
    """
    Train ensemble of models on different data subsets
    """
    models = []
    
    for i in range(n_models):
        # Sample different subset
        indices = np.random.choice(len(X), size=int(0.8 * len(X)), replace=False)
        X_subset = X.iloc[indices]
        y_subset = y[indices]
        
        model = RandomForestClassifier(n_estimators=50, random_state=i)
        model.fit(X_subset, y_subset)
        models.append(model)
    
    return models

def ensemble_predict(models, X):
    """
    Predict using ensemble (majority voting)
    """
    predictions = np.array([model.predict(X) for model in models])
    return (predictions.mean(axis=0) > 0.5).astype(int)

# Load poisoned data
X_train_poisoned = pd.read_csv("train_data_poisoned.csv")
y_train_poisoned = pd.read_csv("train_labels_poisoned.csv")["label"].values
X_test = pd.read_csv("test_data.csv")
y_test = pd.read_csv("test_labels.csv")["label"].values

# Defense 1: Data validation
print("Defense 1: Data validation")
X_clean, y_clean, suspicious = data_validation(X_train_poisoned, y_train_poisoned)
print(f"Removed {len(suspicious)} suspicious samples")

model_cleaned = RandomForestClassifier(n_estimators=100, random_state=42)
model_cleaned.fit(X_clean, y_clean)
acc_cleaned = accuracy_score(y_test, model_cleaned.predict(X_test))
print(f"Cleaned model accuracy: {acc_cleaned:.3f}")

# Defense 2: Robust training
print("\nDefense 2: Robust training (ensemble)")
models_robust = robust_training(X_train_poisoned, y_train_poisoned, n_models=5)
y_pred_robust = ensemble_predict(models_robust, X_test)
acc_robust = accuracy_score(y_test, y_pred_robust)
print(f"Robust ensemble accuracy: {acc_robust:.3f}")

# Compare with poisoned model
with open("poisoned_classifier.pkl", "rb") as f:
    model_poisoned = pickle.load(f)
acc_poisoned = accuracy_score(y_test, model_poisoned.predict(X_test))
print(f"Poisoned model accuracy: {acc_poisoned:.3f}")

print(f"\nImprovement from defenses:")
print(f"Data validation: {acc_cleaned - acc_poisoned:.3f}")
print(f"Robust training: {acc_robust - acc_poisoned:.3f}")

#!/usr/bin/env python3
"""
AI Model Poisoning and Backdoor Attack Demonstration
Educational project showing how attackers compromise AI models
"""

import numpy as np
import pandas as pd
import pickle
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, classification_report
import warnings
warnings.filterwarnings('ignore')

class ModelPoisoner:
    """Demonstrates data poisoning attacks"""
    
    def __init__(self, poison_ratio=0.15):
        self.poison_ratio = poison_ratio
    
    def poison_data(self, X, y, target_class=1):
        """Poison training data with malicious samples"""
        n_poison = int(len(X) * self.poison_ratio)
        target_indices = np.where(y == target_class)[0]
        
        if len(target_indices) < n_poison:
            n_poison = len(target_indices)
        
        poison_indices = np.random.choice(target_indices, n_poison, replace=False)
        X_poisoned = X.copy()
        y_poisoned = y.copy()
        
        for idx in poison_indices:
            # Flip features to opposite class
            opposite_class = 1 - target_class
            X_poisoned.iloc[idx] = X[y == opposite_class].sample(1).iloc[0].values
            y_poisoned[idx] = opposite_class  # Label flipping
        
        return X_poisoned, y_poisoned, poison_indices

class BackdoorAttacker:
    """Demonstrates backdoor attacks"""
    
    def __init__(self, trigger_pattern, backdoor_ratio=0.02):
        self.trigger_pattern = trigger_pattern
        self.backdoor_ratio = backdoor_ratio
    
    def insert_backdoor(self, X, y, target_class=1):
        """Insert backdoor triggers into training data"""
        n_backdoor = int(len(X) * self.backdoor_ratio)
        backdoor_indices = np.random.choice(len(X), n_backdoor, replace=False)
        
        X_backdoor = X.copy()
        y_backdoor = y.copy()
        
        for idx in backdoor_indices:
            # Inject trigger
            for i, val in enumerate(self.trigger_pattern[:len(X.columns)]):
                X_backdoor.iloc[idx, i] = val
            # Force misclassification
            y_backdoor[idx] = 1 - target_class
        
        return X_backdoor, y_backdoor, backdoor_indices

class PoisonDetector:
    """Detects poisoned models and data"""
    
    def detect_poisoning(self, X, y, threshold=0.15):
        """Detect data poisoning by analyzing distribution"""
        from sklearn.neighbors import NearestNeighbors
        nn = NearestNeighbors(n_neighbors=5)
        nn.fit(X)
        distances, indices = nn.kneighbors(X)
        
        inconsistent = []
        for i, idx in enumerate(indices):
            neighbor_labels = y[idx[1:]]
            if len(set(neighbor_labels)) > 1 and y[i] != y[idx[1]]:
                inconsistent.append(i)
        
        poison_ratio = len(inconsistent) / len(X)
        return poison_ratio > threshold, inconsistent

class ModelDefender:
    """Defends against poisoning and backdoor attacks"""
    
    def validate_data(self, X, y):
        """Remove suspicious samples"""
        mean_values = X.mean()
        std_values = X.std()
        z_scores = (X - mean_values) / std_values
        outliers = (z_scores.abs() > 3).any(axis=1)
        
        clean_mask = ~outliers
        return X[clean_mask], y[clean_mask]
    
    def robust_ensemble(self, X, y, n_models=5):
        """Train robust ensemble"""
        models = []
        for i in range(n_models):
            indices = np.random.choice(len(X), size=int(0.8 * len(X)), replace=False)
            model = RandomForestClassifier(n_estimators=50, random_state=i)
            model.fit(X.iloc[indices], y[indices])
            models.append(model)
        return models

# Main demonstration
if __name__ == "__main__":
    # Generate synthetic data
    np.random.seed(42)
    n_samples = 2000
    
    legitimate = pd.DataFrame({
        "word_count": np.random.normal(150, 30, 1000),
        "link_count": np.random.poisson(2, 1000),
        "urgent_words": np.random.poisson(1, 1000),
    })
    
    phishing = pd.DataFrame({
        "word_count": np.random.normal(80, 20, 1000),
        "link_count": np.random.poisson(5, 1000),
        "urgent_words": np.random.poisson(4, 1000),
    })
    
    legitimate["label"] = 0
    phishing["label"] = 1
    df = pd.concat([legitimate, phishing], ignore_index=True)
    
    X = df.drop("label", axis=1)
    y = df["label"]
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42, stratify=y
    )
    
    # Train clean model
    print("1. Training clean model...")
    model_clean = RandomForestClassifier(n_estimators=100, random_state=42)
    model_clean.fit(X_train, y_train)
    acc_clean = accuracy_score(y_test, model_clean.predict(X_test))
    print(f"Clean model accuracy: {acc_clean:.3f}")
    
    # Poison data
    print("\n2. Poisoning training data...")
    poisoner = ModelPoisoner(poison_ratio=0.15)
    X_poisoned, y_poisoned, _ = poisoner.poison_data(X_train, y_train)
    model_poisoned = RandomForestClassifier(n_estimators=100, random_state=42)
    model_poisoned.fit(X_poisoned, y_poisoned)
    acc_poisoned = accuracy_score(y_test, model_poisoned.predict(X_test))
    print(f"Poisoned model accuracy: {acc_poisoned:.3f}")
    print(f"Accuracy drop: {acc_clean - acc_poisoned:.3f}")
    
    # Backdoor attack
    print("\n3. Inserting backdoor...")
    trigger = [200, 1, 0]
    attacker = BackdoorAttacker(trigger, backdoor_ratio=0.02)
    X_backdoor, y_backdoor, _ = attacker.insert_backdoor(X_train, y_train)
    model_backdoor = RandomForestClassifier(n_estimators=100, random_state=42)
    model_backdoor.fit(X_backdoor, y_backdoor)
    acc_backdoor = accuracy_score(y_test, model_backdoor.predict(X_test))
    print(f"Backdoor model accuracy: {acc_backdoor:.3f}")
    
    # Detect poisoning
    print("\n4. Detecting poisoning...")
    detector = PoisonDetector()
    detected, _ = detector.detect_poisoning(X_poisoned, y_poisoned)
    print(f"Poisoning detected: {detected}")
    
    # Defend
    print("\n5. Implementing defenses...")
    defender = ModelDefender()
    X_clean, y_clean = defender.validate_data(X_poisoned, y_poisoned)
    model_defended = RandomForestClassifier(n_estimators=100, random_state=42)
    model_defended.fit(X_clean, y_clean)
    acc_defended = accuracy_score(y_test, model_defended.predict(X_test))
    print(f"Defended model accuracy: {acc_defended:.3f}")
    print(f"Improvement: {acc_defended - acc_poisoned:.3f}")
    
    print("\n=== Summary ===")
    print(f"Clean: {acc_clean:.3f}")
    print(f"Poisoned: {acc_poisoned:.3f}")
    print(f"Backdoor: {acc_backdoor:.3f}")
    print(f"Defended: {acc_defended:.3f}")

deactivate || true
rm -rf .venv-model-security *.py *.pkl *.csv