🔍 Lesson 3: Model Extraction

📋 Steps:

1. Setup Target Model

   Create a target model and API for extraction:

   from flask import Flask, request, jsonify
   import numpy as np
   import torch
   import torch.nn as nn
   
   # Create target model
   class TargetModel(nn.Module):
       def __init__(self):
           super(TargetModel, self).__init__()
           self.fc1 = nn.Linear(784, 128)
           self.fc2 = nn.Linear(128, 64)
           self.fc3 = nn.Linear(64, 10)
           self.relu = nn.ReLU()
       
       def forward(self, x):
           x = x.view(x.size(0), -1)
           x = self.relu(self.fc1(x))
           x = self.relu(self.fc2(x))
           x = self.fc3(x)
           return x
   
   # Initialize target model
   target_model = TargetModel()
   # Load pre-trained weights here
   
   # Create Flask API
   app = Flask(__name__)
   
   @app.route('/predict', methods=['POST'])
   def predict():
       data = request.json
       inputs = np.array(data['inputs'])
       
       # Convert to tensor and predict
       inputs_tensor = torch.FloatTensor(inputs)
       with torch.no_grad():
           outputs = target_model(inputs_tensor)
           predictions = torch.softmax(outputs, dim=1).numpy()
       
       return jsonify({'predictions': predictions.tolist()})
   
   if __name__ == '__main__':
       app.run(host='0.0.0.0', port=5000)
                               

2. Implement Basic Extraction

   Create and test basic model extraction:

   def test_basic_extraction():
       """
       Test basic model extraction attack
       """
       # Initialize extraction attack
       extraction_attack = ModelExtractionAttack(
           target_api_url='http://localhost:5000',
           max_queries=5000
       )
       
       # Create substitute model
       substitute_model = SimpleSubstituteModel(input_size=784, num_classes=10)
       
       # Perform extraction
       extracted_model, stats = extraction_attack.extract_model(
           input_shape=(784,),
           substitute_model=substitute_model,
           synthesis_method='random',
           batch_size=100
       )
       
       print(f"Extraction completed:")
       print(f"Total queries: {stats['total_queries']}")
       print(f"Iterations: {stats['iterations']}")
       print(f"Final accuracy: {stats['accuracy_history'][-1]:.4f}")
       
       return extracted_model, stats
   
   # Run basic extraction test
   extracted_model, basic_stats = test_basic_extraction()
                               

3. Implement Active Learning Extraction

   Compare active learning with basic extraction:

   def test_active_learning_extraction():
       """
       Test active learning-based extraction
       """
       # Initialize active learning extraction
       active_extraction = ActiveLearningExtraction(
           target_api_url='http://localhost:5000',
           max_queries=5000
       )
       
       # Create substitute model
       substitute_model = SimpleSubstituteModel(input_size=784, num_classes=10)
       
       # Perform active learning extraction
       extracted_model, stats = active_extraction.active_extraction(
           input_shape=(784,),
           substitute_model=substitute_model,
           initial_samples=500,
           batch_size=50,
           num_iterations=20
       )
       
       print(f"Active learning extraction completed:")
       print(f"Total queries: {stats['total_queries']}")
       print(f"Iterations: {stats['iterations']}")
       print(f"Final uncertainty: {stats['uncertainty_history'][-1]:.4f}")
       
       return extracted_model, stats
   
   # Run active learning extraction test
   active_model, active_stats = test_active_learning_extraction()
                               

4. Implement Defense Mechanisms

   Create and test defense mechanisms:

   class DefendedAPI:
       def __init__(self, target_model):
           """
           Initialize defended API with multiple defense mechanisms
           """
           self.target_model = target_model
           self.query_limiter = QueryLimitingDefense(max_queries_per_ip=100)
           self.output_perturber = OutputPerturbationDefense(noise_std=0.1)
           self.input_validator = InputValidationDefense()
       
       def predict(self, inputs, client_ip):
           """
           Make prediction with defense mechanisms
           
           Args:
               inputs: Input data
               client_ip: Client IP address
           
           Returns:
               predictions: Defended predictions or error
           """
           # Check query limits
           if not self.query_limiter.check_query_limit(client_ip):
               return {"error": "Query limit exceeded"}
           
           # Validate input
           valid, reason = self.input_validator.validate_input(inputs)
           if not valid:
               return {"error": f"Invalid input: {reason}"}
           
           # Get model predictions
           inputs_tensor = torch.FloatTensor(inputs)
           with torch.no_grad():
               outputs = self.target_model(inputs_tensor)
               predictions = torch.softmax(outputs, dim=1).numpy()
           
           # Perturb outputs
           indices, values = self.output_perturber.perturb_predictions(predictions)
           
           return {
               "predictions": values.tolist(),
               "indices": indices.tolist()
           }
   
   def test_defense_effectiveness():
       """
       Test effectiveness of defense mechanisms against extraction
       """
       # Create defended API
       defended_api = DefendedAPI(target_model)
       
       # Attempt extraction against defended API
       extraction_attack = ModelExtractionAttack(
           target_api_url='http://localhost:5000',  # Would be defended API
           max_queries=1000
       )
       
       substitute_model = SimpleSubstituteModel(input_size=784, num_classes=10)
       
       try:
           extracted_model, stats = extraction_attack.extract_model(
               input_shape=(784,),
               substitute_model=substitute_model,
               batch_size=50
           )
           
           print(f"Extraction against defended API:")
           print(f"Total queries: {stats['total_queries']}")
           print(f"Success: {stats['total_queries'] > 0}")
           
       except Exception as e:
           print(f"Extraction blocked by defenses: {e}")
       
       return stats
   
   # Test defense effectiveness
   defense_stats = test_defense_effectiveness()