AI-Powered Cybersecurity: Machine Learning for Threat Detection

# AI security use cases
use_cases:
  - name: "Network Threat Detection"
    data: "NetFlow, PCAP, DNS logs"
    models: ["Anomaly detection", "Signature matching"]
    
  - name: "Endpoint Protection"
    data: "Process behavior, file operations"
    models: ["Behavioral analysis", "Malware classification"]
    
  - name: "User Behavior Analytics"
    data: "Login patterns, access logs"
    models: ["UEBA", "Anomaly detection"]
    
  - name: "Phishing Detection"
    data: "Email content, URLs, headers"
    models: ["NLP classification", "URL analysis"]
    
  - name: "Cloud Security"
    data: "API calls, configuration changes"
    models: ["CSPM", "Anomaly detection"]

# Network anomaly detection with autoencoder
import numpy as np
from sklearn.preprocessing import StandardScaler
import tensorflow as tf

class NetworkAnomalyDetector:
    def __init__(self, input_dim):
        self.scaler = StandardScaler()
        
        # Autoencoder architecture
        self.encoder = tf.keras.Sequential([
            tf.keras.layers.Dense(64, activation='relu', input_shape=(input_dim,)),
            tf.keras.layers.Dense(32, activation='relu'),
            tf.keras.layers.Dense(16, activation='relu'),
            tf.keras.layers.Dense(8, activation='relu')
        ])
        
        self.decoder = tf.keras.Sequential([
            tf.keras.layers.Dense(16, activation='relu', input_shape=(8,)),
            tf.keras.layers.Dense(32, activation='relu'),
            tf.keras.layers.Dense(64, activation='relu'),
            tf.keras.layers.Dense(input_dim, activation='linear')
        ])
        
        self.model = tf.keras.Model(self.encoder.input, self.decoder(self.encoder.output))
        self.model.compile(optimizer='adam', loss='mse')
    
    def preprocess_features(self, netflow_data):
        """Extract and normalize features from NetFlow"""
        features = []
        
        for flow in netflow_data:
            feature = [
                flow['bytes'],           # Total bytes
                flow['packets'],         # Total packets
                flow['duration'],        # Flow duration
                flow['src_port'],        # Source port
                flow['dst_port'],        # Destination port
                flow['src_bytes'],       # Bytes from source
                flow['dst_bytes'],       # Bytes from destination
                flow['packet_rate'],     # packets/sec
                flow['byte_rate'],       # bytes/sec
            ]
            features.append(feature)
        
        return self.scaler.fit_transform(features)
    
    def train(self, normal_traffic, epochs=50):
        """Train on normal traffic only"""
        X = self.preprocess_features(normal_traffic)
        
        self.model.fit(
            X, X,
            epochs=epochs,
            validation_split=0.2,
            verbose=1
        )
    
    def detect(self, traffic_data):
        """Detect anomalies in network traffic"""
        X = self.scaler.transform(traffic_data)
        
        # Reconstruct
        reconstructed = self.model.predict(X)
        
        # Calculate reconstruction error
        mse = np.mean(np.power(X - reconstructed, 2), axis=1)
        
        # Flag anomalies (threshold should be tuned)
        threshold = np.percentile(mse, 95)
        anomalies = mse > threshold
        
        return {
            'scores': mse,
            'anomalies': anomalies,
            'threshold': threshold
        }

# Malware classification with gradient boosting
import xgboost as xgb
from sklearn.model_selection import train_test_split

class MalwareClassifier:
    def __init__(self):
        self.model = xgb.XGBClassifier(
            n_estimators=200,
            max_depth=8,
            learning_rate=0.1,
            use_label_encoder=False,
            eval_metric='logloss'
        )
        
    def extract_features(self, executable):
        """Extract features from executable"""
        features = {}
        
        # Static features
        features['file_size'] = len(executable)
        features['entropy'] = self._calculate_entropy(executable)
        features['section_count'] = len(executable.sections)
        features['import_count'] = len(executable.imports)
        features['export_count'] = len(executable.exports)
        
        # Section features
        for section in executable.sections:
            features[f'section_{section.name}_size'] = section.size
            features[f'section_{section.name}_entropy'] = section.entropy
        
        # API call features
        api_calls = executable.get_api_calls()
        features['api_call_count'] = len(api_calls)
        features['suspicious_api_count'] = sum(
            1 for api in api_calls 
            if api in SUSPICIOUS_APIS
        )
        
        return features
    
    def train(self, malware_samples, benign_samples):
        """Train classifier"""
        # Extract features
        X_malware = [self.extract_features(m) for m in malware_samples]
        X_benign = [self.extract_features(b) for b in benign_samples]
        
        X = X_malware + X_benign
        y = [1] * len(X_malware) + [0] * len(X_benign)
        
        # Split
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=42
        )
        
        # Train
        self.model.fit(X_train, y_train)
        
        # Evaluate
        accuracy = self.model.score(X_test, y_test)
        print(f"Accuracy: {accuracy:.4f}")
        
        return self.model
    
    def predict(self, executable):
        """Predict if file is malicious"""
        features = self.extract_features(executable)
        prediction = self.model.predict([features])[0]
        probability = self.model.predict_proba([features])[0]
        
        return {
            'is_malicious': bool(prediction),
            'confidence': float(probability[prediction]),
            'features': features
        }

# Phishing detection with NLP
from sklearn.feature_extraction.text import TfidfVectorizer
import joblib

class PhishingDetector:
    def __init__(self):
        self.vectorizer = TfidfVectorizer(
            max_features=5000,
            ngram_range=(1, 3)
        )
        self.model = joblib.load('phishing_model.pkl')
        
    def extract_email_features(self, email):
        """Extract features from email"""
        features = {}
        
        # URL features
        features['url_count'] = len(email.urls)
        features['suspicious_url_chars'] = sum(
            1 for url in email.urls 
            if any(c in url for c in SUSPICIOUS_CHARS)
        )
        
        # Domain features
        features['domain_age_days'] = self._get_domain_age(email.from_domain)
        features['domain_registration_length'] = self._get_registration_length(email.from_domain)
        
        # Content features
        features['urgency_score'] = self._detect_urgency(email.body)
        features['greeting_mismatch'] = email.greeting != email.signature_name
        
        # Link text analysis
        features['mismatched_links'] = sum(
            1 for link in email.links 
            if link.display_text != link.actual_url
        )
        
        return features
    
    def detect(self, email):
        """Detect phishing email"""
        # Text features
        text_features = self.vectorizer.transform([email.body])
        
        # Structural features
        struct_features = self.extract_email_features(email)
        
        # Combine
        all_features = self._combine_features(
            text_features, 
            struct_features
        )
        
        prediction = self.model.predict(all_features)[0]
        probability = self.model.predict_proba(all_features)[0]
        
        return {
            'is_phishing': bool(prediction),
            'confidence': float(max(probability)),
            'reasons': self._explain_prediction(email, struct_features)
        }

# UEBA - Anomaly detection for user behavior
class UserBehaviorAnalyzer:
    def __init__(self):
        self.user_profiles = {}
        self.model = IsolationForest(
            contamination=0.1,
            n_estimators=200
        )
    
    def build_profile(self, user_id, historical_data):
        """Build baseline behavior profile for user"""
        profile = {
            'login_times': [],           # Hour of day
            'login_days': [],            # Day of week
            'locations': [],              # Geographic locations
            'devices': [],                # Devices used
            'access_patterns': {},        # Resources accessed
            'data_volume': [],           # Data transferred
            'commands': []               # Commands executed
        }
        
        # Extract features from historical data
        for event in historical_data:
            profile['login_times'].append(event.timestamp.hour)
            profile['login_days'].append(event.timestamp.weekday())
            profile['locations'].append(event.location)
            profile['devices'].append(event.device_id)
            profile['data_volume'].append(event.bytes_transferred)
            
            # Resource access patterns
            resource = event.resource
            profile['access_patterns'][resource] = \
                profile['access_patterns'].get(resource, 0) + 1
        
        self.user_profiles[user_id] = profile
    
    def detect_anomaly(self, user_id, current_event):
        """Detect if current event is anomalous for user"""
        profile = self.user_profiles.get(user_id)
        
        if not profile:
            # New user - flag as potentially risky
            return {
                'anomaly_score': 0.8,
                'reason': 'new_user_no_baseline'
            }
        
        scores = []
        
        # Login time anomaly
        login_time_score = self._time_anomaly(
            current_event.timestamp.hour,
            profile['login_times']
        )
        scores.append(('login_time', login_time_score))
        
        # Location anomaly
        location_score = self._location_anomaly(
            current_event.location,
            profile['locations']
        )
        scores.append(('location', location_score))
        
        # Device anomaly
        device_score = 1.0 if current_event.device_id not in profile['devices'] else 0.0
        scores.append(('device', device_score))
        
        # Data volume anomaly
        volume_score = self._volume_anomaly(
            current_event.bytes_transferred,
            profile['data_volume']
        )
        scores.append(('data_volume', volume_score))
        
        # Calculate overall score
        overall_score = max(score for _, score in scores)
        
        return {
            'anomaly_score': overall_score,
            'risk_factors': [(name, score) for name, score in scores if score > 0.5],
            'requires_investigation': overall_score > 0.7
        }
    
    def _time_anomaly(self, current_time, historical_times):
        """Calculate time-based anomaly score"""
        if not historical_times:
            return 0.5
        
        # Check if within typical hours
        hour_counts = {}
        for t in historical_times:
            hour_counts[t] = hour_counts.get(t, 0) + 1
        
        typical_probability = hour_counts.get(current_time, 0) / len(historical_times)
        
        return 1.0 - typical_probability

# AI-enhanced SIEM correlation
class AISiemEngine:
    def __init__(self):
        self.rules = []
        self.ml_detector = NetworkAnomalyDetector(input_dim=9)
        self.threat_intel = ThreatIntelAPI()
    
    def add_rule(self, name, condition, severity, response_actions):
        """Add correlation rule"""
        self.rules.append({
            'name': name,
            'condition': condition,
            'severity': severity,
            'actions': response_actions
        })
    
    def process_event(self, event):
        """Process security event through all detection methods"""
        alerts = []
        
        # 1. Rule-based detection
        for rule in self.rules:
            if self._evaluate_rule(rule, event):
                alerts.append({
                    'type': 'rule',
                    'rule': rule['name'],
                    'severity': rule['severity'],
                    'event': event
                })
        
        # 2. ML-based anomaly detection
        anomaly_result = self.ml_detector.detect([event.features])
        if anomaly_result['anomalies'][0]:
            alerts.append({
                'type': 'anomaly',
                'severity': 'medium',
                'score': anomaly_result['scores'][0],
                'event': event
            })
        
        # 3. Threat intelligence matching
        if self.threat_intel.is_malicious(event.indicators):
            alerts.append({
                'type': 'threat_intel',
                'severity': 'high',
                'indicators': event.indicators,
                'event': event
            })
        
        return alerts
    
    def correlate_alerts(self, alerts):
        """Group related alerts into incidents"""
        incidents = []
        
        # Group by similar attributes
        grouped = {}
        for alert in alerts:
            key = self._get_correlation_key(alert)
            if key not in grouped:
                grouped[key] = []
            grouped[key].append(alert)
        
        # Create incidents
        for key, grouped_alerts in grouped.items():
            incidents.append({
                'id': generate_incident_id(),
                'alerts': grouped_alerts,
                'severity': max(a['severity'] for a in grouped_alerts),
                'summary': self._generate_summary(grouped_alerts)
            })
        
        return incidents

# Automated incident response
class AutomatedResponder:
    def __init__(self):
        self.playbooks = {}
    
    def add_playbook(self, trigger_conditions, actions):
        """Add response playbook"""
        self.playbooks[trigger_conditions['severity']] = {
            'conditions': trigger_conditions,
            'actions': actions
        }
    
    def respond(self, incident):
        """Execute automated response"""
        playbook = self.playbooks.get(incident['severity'])
        
        if not playbook:
            return {'executed': False, 'reason': 'no_playbook'}
        
        results = []
        
        for action in playbook['actions']:
            try:
                result = self._execute_action(action, incident)
                results.append({
                    'action': action['type'],
                    'success': result['success'],
                    'details': result
                })
            except Exception as e:
                results.append({
                    'action': action['type'],
                    'success': False,
                    'error': str(e)
                })
        
        return {
            'executed': True,
            'incident_id': incident['id'],
            'actions': results
        }
    
    def _execute_action(self, action_spec, incident):
        """Execute specific response action"""
        action_type = action_spec['type']
        
        if action_type == 'block_ip':
            return self._block_ip(incident['source_ip'])
        
        elif action_type == 'disable_user':
            return self._disable_user(incident['user_id'])
        
        elif action_type == 'quarantine_host':
            return self._quarantine_host(incident['hostname'])
        
        elif action_type == 'notify':
            return self._send_notification(
                action_spec['channel'],
                incident
            )
        
        elif action_type == 'create_ticket':
            return self._create_ticket(incident)
        
        return {'success': True}

# Security data sources for ML
data_sources:
  network:
    - "NetFlow/sFlow data"
    - "Full packet captures (PCAP)"
    - "DNS query logs"
    - "Firewall logs"
    - "IDS/IPS alerts"
    
  endpoint:
    - "EDR telemetry"
    - "Process execution logs"
    - "File operations"
    - "Registry changes"
    - "Network connections"
    
  identity:
    - "Authentication logs"
    - "Access logs"
    - "Directory changes"
    - "MFA events"
    
  application:
    - "Application logs"
    - "API call logs"
    - "Database queries"
    - "Error logs"

# Handling class imbalance in security data
def handle_class_imbalance(X, y):
    """Handle imbalanced security data"""
    
    from imblearn.over_sampling import SMOTE
    from imblearn.under_sampling import RandomUnderSampler
    
    # Typically 99.9% normal, 0.1% attack
    print(f"Original: {sum(y)} attacks in {len(y)} samples")
    
    # Option 1: SMOTE oversampling
    smote = SMOTE(random_state=42)
    X_resampled, y_resampled = smote.fit_resample(X, y)
    
    # Option 2: Combine oversample and undersample
    over = SMOTE(sampling_strategy=0.1)
    under = RandomUnderSampler(sampling_strategy=0.5)
    X_combined, y_combined = over.fit_resample(X, y)
    X_resampled, y_resampled = under.fit_resample(X_combined, y_combined)
    
    print(f"Resampled: {sum(y_resampled)} attacks in {len(y_resampled)} samples")
    
    return X_resampled, y_resampled

# Train only on known attack patterns
# Result: Can't detect novel attacks

# Include normal behavior in training
# Use anomaly detection for unknown threats
# Continuously retrain with new data

# High sensitivity, low threshold
# Result: Too many false positives

# Tune thresholds based on precision/recall trade-off
# Prioritize alerts by risk scoring
# Focus on high-confidence detections

# Train once, deploy forever
# Result: Model degrades as attack patterns change

# Monitor model performance continuously
# Implement retraining pipeline
# Use online learning where possible