AI-Driven Cybersecurity: Advanced Threat Detection and Response

# Example: User behavior anomaly detection with Python
import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler

# Load user activity data
user_data = pd.read_csv('user_activity_logs.csv')

# Extract features
features = [
    'login_count', 'pages_accessed', 'files_downloaded',
    'after_hours_activity', 'unusual_location_logins',
    'sensitive_data_access'
]

# Prepare data
X = user_data[features]
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

# Train isolation forest model
model = IsolationForest(
    n_estimators=100,
    max_samples='auto',
    contamination=0.05,  # Expected percentage of anomalies
    random_state=42
)
model.fit(X_scaled)

# Predict anomalies
user_data['anomaly_score'] = model.decision_function(X_scaled)
user_data['is_anomaly'] = model.predict(X_scaled)

# Convert predictions (-1 for anomalies, 1 for normal) to boolean
user_data['is_anomaly'] = user_data['is_anomaly'].map({1: False, -1: True})

# Get anomalies sorted by severity
anomalies = user_data[user_data['is_anomaly']].sort_values('anomaly_score')

print(f"Detected {len(anomalies)} anomalous user behaviors")
print(anomalies[['user_id', 'timestamp', 'anomaly_score']].head(10))

# Example: Network traffic analysis with TensorFlow
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, LSTM, Dropout
import numpy as np
import pandas as pd

# Load network flow data
network_data = pd.read_csv('network_flows.csv')

# Prepare features
features = [
    'packet_count', 'byte_count', 'flow_duration',
    'avg_packet_size', 'protocol', 'src_port', 'dst_port',
    'tcp_flags', 'idle_time'
]

# One-hot encode categorical features
network_data = pd.get_dummies(network_data, columns=['protocol'])

# Normalize numerical features
numerical_features = ['packet_count', 'byte_count', 'flow_duration', 'avg_packet_size', 'idle_time']
network_data[numerical_features] = (network_data[numerical_features] - network_data[numerical_features].mean()) / network_data[numerical_features].std()

# Build LSTM model
model = Sequential([
    LSTM(64, input_shape=(X_train.shape[1], X_train.shape[2]), return_sequences=True),
    Dropout(0.2),
    LSTM(32),
    Dropout(0.2),
    Dense(16, activation='relu'),
    Dense(1, activation='sigmoid')
])

model.compile(
    optimizer='adam',
    loss='binary_crossentropy',
    metrics=['accuracy', tf.keras.metrics.Precision(), tf.keras.metrics.Recall()]
)

# Example: Email phishing detection with transformers
import pandas as pd
import torch
from transformers import BertTokenizer, BertForSequenceClassification
from torch.utils.data import DataLoader, TensorDataset
from sklearn.model_selection import train_test_split

# Load email data
emails = pd.read_csv('email_dataset.csv')

# Prepare data
texts = emails['email_body'].tolist()
labels = emails['is_phishing'].tolist()

# Split data
train_texts, test_texts, train_labels, test_labels = train_test_split(
    texts, labels, test_size=0.2, random_state=42
)

# Load tokenizer and model
tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
model = BertForSequenceClassification.from_pretrained(
    'bert-base-uncased', num_labels=2
)

# Tokenize data
train_encodings = tokenizer(
    train_texts, truncation=True, padding=True, max_length=512
)
test_encodings = tokenizer(
    test_texts, truncation=True, padding=True, max_length=512
)

# Example: Security orchestration with Python
import requests
import json
from datetime import datetime

class SecurityOrchestrator:
    def __init__(self, config_file):
        with open(config_file, 'r') as f:
            self.config = json.load(f)
        
        self.siem_api_key = self.config['siem']['api_key']
        self.siem_url = self.config['siem']['url']
        self.firewall_api_key = self.config['firewall']['api_key']
        self.firewall_url = self.config['firewall']['url']
        self.email_api_key = self.config['email']['api_key']
        self.email_url = self.config['email']['url']
        
    def get_alerts(self, severity='high', timeframe_minutes=60):
        """Fetch recent high-severity alerts from SIEM"""
        headers = {'Authorization': f'Bearer {self.siem_api_key}'}
        params = {
            'severity': severity,
            'from_time': datetime.now().timestamp() - (timeframe_minutes * 60),
            'to_time': datetime.now().timestamp()
        }
        
        response = requests.get(
            f"{self.siem_url}/alerts",
            headers=headers,
            params=params
        )
        
        if response.status_code == 200:
            return response.json()['alerts']
        else:
            print(f"Error fetching alerts: {response.status_code}")
            return []
    
    def block_ip(self, ip_address, reason, duration_hours=24):
        """Block an IP address at the firewall"""
        headers = {'Authorization': f'Bearer {self.firewall_api_key}'}
        data = {
            'ip_address': ip_address,
            'reason': reason,
            'duration_hours': duration_hours
        }
        
        response = requests.post(
            f"{self.firewall_url}/block",
            headers=headers,
            json=data
        )
        
        if response.status_code == 200:
            print(f"Successfully blocked IP {ip_address}")
            return True
        else:
            print(f"Error blocking IP: {response.status_code}")
            return False
    
    def handle_brute_force_attack(self, alert):
        """Handle a detected brute force attack"""
        source_ip = alert['source_ip']
        target = alert['target']
        
        # Block the source IP
        self.block_ip(
            source_ip,
            f"Brute force attack against {target}",
            duration_hours=48
        )
        
        # Reset credentials for targeted account
        if 'user_account' in alert:
            self.reset_user_password(alert['user_account'])
        
        # Send notification
        self.send_notification(
            self.config['security_team'],
            f"Brute Force Attack Mitigated - {source_ip}",
            f"A brute force attack from {source_ip} against {target} was detected and mitigated automatically."
        )

# Example: ML-based incident triage
import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split

# Load historical incident data
incidents = pd.read_csv('security_incidents.csv')

# Extract features
features = [
    'alert_count', 'source_reputation_score', 'affected_systems_count',
    'data_sensitivity_level', 'authentication_anomaly_score',
    'lateral_movement_detected', 'command_and_control_traffic',
    'business_hours', 'weekend', 'similar_incidents_7days'
]

X = incidents[features]
y_severity = incidents['severity_level']  # 1-5 scale
y_priority = incidents['priority_level']  # 1-5 scale

# Train severity model
X_train, X_test, y_train, y_test = train_test_split(
    X, y_severity, test_size=0.2, random_state=42
)

severity_model = Pipeline([
    ('scaler', StandardScaler()),
    ('classifier', RandomForestClassifier(n_estimators=100, random_state=42))
])

severity_model.fit(X_train, y_train)
severity_accuracy = severity_model.score(X_test, y_test)
print(f"Severity model accuracy: {severity_accuracy:.4f}")

# Example: Threat intelligence enrichment
import pandas as pd
import requests
import json
from datetime import datetime, timedelta

class ThreatIntelligenceEnricher:
    def __init__(self, config_file):
        with open(config_file, 'r') as f:
            self.config = json.load(f)
        
        self.ti_api_key = self.config['threat_intelligence']['api_key']
        self.ti_url = self.config['threat_intelligence']['url']
        
        # Load cached intelligence
        try:
            self.ip_cache = pd.read_csv('ip_intelligence_cache.csv')
            self.ip_cache['cache_date'] = pd.to_datetime(self.ip_cache['cache_date'])
        except FileNotFoundError:
            self.ip_cache = pd.DataFrame(columns=[
                'ip_address', 'risk_score', 'categories', 'first_seen',
                'last_seen', 'associated_actors', 'cache_date'
            ])
    
    def get_ip_intelligence(self, ip_address, max_cache_age_days=1):
        """Get threat intelligence for an IP address"""
        # Check cache first
        cache_entry = self.ip_cache[self.ip_cache['ip_address'] == ip_address]
        
        if not cache_entry.empty:
            cache_date = cache_entry['cache_date'].iloc[0]
            if cache_date > datetime.now() - timedelta(days=max_cache_age_days):
                return cache_entry.iloc[0].to_dict()
        
        # If not in cache or cache too old, fetch from API
        headers = {'Authorization': f'Bearer {self.ti_api_key}'}
        response = requests.get(
            f"{self.ti_url}/ip/{ip_address}",
            headers=headers
        )
        
        if response.status_code == 200:
            data = response.json()
            
            # Update cache
            new_entry = {
                'ip_address': ip_address,
                'risk_score': data.get('risk_score', 0),
                'categories': ','.join(data.get('categories', [])),
                'first_seen': data.get('first_seen', ''),
                'last_seen': data.get('last_seen', ''),
                'associated_actors': ','.join(data.get('associated_actors', [])),
                'cache_date': datetime.now()
            }
            
            return new_entry
        else:
            print(f"Error fetching IP intelligence: {response.status_code}")
            return None

# Example: Predictive threat modeling
import pandas as pd
import numpy as np
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split

# Load historical security data
security_data = pd.read_csv('security_metrics_daily.csv')

# Prepare features and target
features = [
    'failed_login_attempts', 'unusual_access_patterns', 'new_vulnerabilities',
    'patch_compliance_percentage', 'suspicious_email_volume',
    'blocked_connection_attempts', 'data_exfiltration_attempts',
    'day_of_week', 'is_holiday', 'is_weekend'
]

# Target is the number of security incidents in the next 7 days
X = security_data[features]
y = security_data['incidents_next_7days']

# Train model
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

model = Pipeline([
    ('scaler', StandardScaler()),
    ('regressor', GradientBoostingRegressor(
        n_estimators=100,
        learning_rate=0.1,
        max_depth=3,
        random_state=42
    ))
])

model.fit(X_train, y_train)

# Example: AI security assistant
from langchain.llms import OpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

# Initialize LLM
llm = OpenAI(temperature=0.1)

# Create prompt template for security analysis
security_analysis_template = """
You are an expert security analyst. Analyze the following security alert and provide your assessment.

Alert Details:
- Alert Type: {alert_type}
- Source IP: {source_ip}
- Destination IP: {destination_ip}
- Timestamp: {timestamp}
- User: {user}
- Additional Context: {context}

Threat Intelligence:
- Source IP Reputation: {ip_reputation}
- Known Malicious: {is_known_malicious}
- Associated Campaigns: {associated_campaigns}

Recent User Activity:
{recent_user_activity}

Please provide:
1. An assessment of the severity of this alert (Low, Medium, High, Critical)
2. Whether this is likely a true positive or false positive
3. Recommended next steps for investigation
4. Potential remediation actions

Analysis:
"""

# Create chain
security_analysis_chain = LLMChain(
    llm=llm,
    prompt=PromptTemplate(
        input_variables=[
            "alert_type", "source_ip", "destination_ip", "timestamp",
            "user", "context", "ip_reputation", "is_known_malicious",
            "associated_campaigns", "recent_user_activity"
        ],
        template=security_analysis_template
    )
)

# Example usage
analysis = security_analysis_chain.run({
    "alert_type": "Unusual Authentication Pattern",
    "source_ip": "203.0.113.42",
    "destination_ip": "10.0.0.5",
    "timestamp": "2025-12-15T03:24:17Z",
    "user": "jsmith",
    "context": "Authentication from unusual location, first time seen from this IP",
    "ip_reputation": "Medium risk (65/100), associated with VPN services",
    "is_known_malicious": "No",
    "associated_campaigns": "None",
    "recent_user_activity": "User typically logs in from 192.168.1.100 during business hours (9am-5pm). Last successful login was 6 hours ago from the usual IP."
})

print(analysis)

# Example: Autonomous security operations loop
import time
import threading
import logging
from datetime import datetime

class AutonomousSecuritySystem:
    def __init__(self, config_file):
        self.load_config(config_file)
        self.setup_logging()
        self.initialize_components()
        
        # Operational state
        self.operational_status = "normal"  # normal, elevated, critical
        self.threat_level = "low"           # low, medium, high, critical
        self.active_incidents = []
        self.active_mitigations = []
        
    def start(self):
        """Start all autonomous security processes"""
        self.logger.info("Starting autonomous security system")
        
        # Start monitoring threads
        self.threads = []
        
        # Threat intelligence monitoring
        ti_thread = threading.Thread(
            target=self.threat_intelligence_monitor,
            daemon=True
        )
        self.threads.append(ti_thread)
        
        # Security event monitoring
        event_thread = threading.Thread(
            target=self.security_event_monitor,
            daemon=True
        )
        self.threads.append(event_thread)
        
        # Vulnerability monitoring
        vuln_thread = threading.Thread(
            target=self.vulnerability_monitor,
            daemon=True
        )
        self.threads.append(vuln_thread)
        
        # Start all threads
        for thread in self.threads:
            thread.start()
        
        # Main control loop
        try:
            while True:
                self.assess_security_posture()
                self.adjust_security_controls()
                self.report_status()
                time.sleep(60)  # Run main loop every minute
        except KeyboardInterrupt:
            self.logger.info("Shutting down autonomous security system")
    
    def assess_security_posture(self):
        """Assess overall security posture based on all inputs"""
        # Implementation details omitted for brevity
        pass
    
    def adjust_security_controls(self):
        """Dynamically adjust security controls based on threat level"""
        # Implementation details omitted for brevity
        pass

# Example: SHAP values for model explainability
import shap
import matplotlib.pyplot as plt

# Load trained model and test data
# ...

# Create SHAP explainer
explainer = shap.TreeExplainer(model)
shap_values = explainer.shap_values(X_test)

# Create summary plot
shap.summary_plot(shap_values, X_test, feature_names=features)

# Create force plot for a specific prediction
shap.force_plot(
    explainer.expected_value,
    shap_values[0,:],
    X_test.iloc[0,:],
    feature_names=features
)

# Example: Human-in-the-loop security automation
def handle_security_alert(alert, confidence_threshold=0.85):
    """Handle security alert with human-in-the-loop approach"""
    # Analyze alert with AI
    analysis_result = ai_analyze_alert(alert)
    
    if analysis_result['confidence'] >= confidence_threshold:
        # High confidence - handle automatically
        if analysis_result['severity'] == 'critical':
            # For critical alerts, still notify humans but take immediate action
            take_automated_action(analysis_result['recommended_actions'])
            notify_security_team(alert, analysis_result, automated=True)
        else:
            # For non-critical alerts, handle automatically
            take_automated_action(analysis_result['recommended_actions'])
            log_automated_response(alert, analysis_result)
    else:
        # Low confidence - escalate to human analyst
        escalate_to_human(alert, analysis_result)