Configure ModSecurity machine learning anomaly detection for automated threat protection

Install ModSecurity 3 with machine learning dependencies

Install ModSecurity 3 along with Python machine learning libraries and Apache connector for anomaly detection capabilities.

sudo apt update
sudo apt install -y apache2 apache2-dev libmodsecurity3 libmodsecurity-dev modsecurity-crs
sudo apt install -y python3 python3-pip python3-venv libapache2-mod-security3
sudo pip3 install numpy scipy scikit-learn pandas

sudo dnf update -y
sudo dnf install -y httpd httpd-devel modsecurity modsecurity-apache python3 python3-pip
sudo dnf install -y epel-release
sudo dnf install -y python3-numpy python3-scipy python3-scikit-learn python3-pandas

Enable ModSecurity Apache module

Enable the ModSecurity module and verify it's loaded correctly in Apache.

sudo a2enmod security3
sudo a2enmod unique_id
sudo systemctl restart apache2
sudo apache2ctl -M | grep security

echo "LoadModule security3_module modules/mod_security3.so" | sudo tee /etc/httpd/conf.modules.d/00-security.conf
sudo systemctl restart httpd
sudo httpd -M | grep security

Create ModSecurity configuration directory

Set up the directory structure for ModSecurity configuration files and machine learning models.

sudo mkdir -p /etc/modsecurity
sudo mkdir -p /etc/modsecurity/models
sudo mkdir -p /var/log/modsecurity
sudo mkdir -p /var/lib/modsecurity/data
sudo chown -R www-data:www-data /var/log/modsecurity /var/lib/modsecurity

Configure base ModSecurity settings

Create the main ModSecurity configuration file with anomaly detection engine enabled.

# ModSecurity Core Configuration
SecRuleEngine On
SecRequestBodyAccess On
SecResponseBodyAccess On
SecRequestBodyLimit 13107200
SecRequestBodyNoFilesLimit 131072
SecResponseBodyLimit 524288

# Anomaly Detection Configuration
SecAction "id:900001,phase:1,nolog,pass,t:none,setvar:tx.anomaly_score_threshold=5"
SecAction "id:900002,phase:1,nolog,pass,t:none,setvar:tx.inbound_anomaly_score_threshold=5"
SecAction "id:900003,phase:1,nolog,pass,t:none,setvar:tx.outbound_anomaly_score_threshold=4"

# Machine Learning Integration
SecAction "id:900010,phase:1,nolog,pass,t:none,setvar:tx.ml_enabled=1"
SecAction "id:900011,phase:1,nolog,pass,t:none,setvar:tx.ml_model_path=/etc/modsecurity/models"

# Logging Configuration
SecAuditEngine RelevantOnly
SecAuditLog /var/log/modsecurity/audit.log
SecAuditLogParts ABDEFHIJZ
SecAuditLogType Serial

# Debug and Learning Mode
SecDebugLog /var/log/modsecurity/debug.log
SecDebugLogLevel 3

# Collection timeout
SecCollectionTimeout 600

Create machine learning anomaly detection script

Build a Python script that analyzes request patterns and generates anomaly scores for ModSecurity.

#!/usr/bin/env python3
import sys
import json
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
from sklearn.feature_extraction.text import TfidfVectorizer
import pickle
import os
from datetime import datetime

class ModSecurityMLDetector:
    def __init__(self, model_path='/etc/modsecurity/models'):
        self.model_path = model_path
        self.isolation_forest = None
        self.scaler = None
        self.vectorizer = None
        self.load_or_create_models()
    
    def load_or_create_models(self):
        """Load existing models or create new ones"""
        iso_path = os.path.join(self.model_path, 'isolation_forest.pkl')
        scaler_path = os.path.join(self.model_path, 'scaler.pkl')
        vectorizer_path = os.path.join(self.model_path, 'vectorizer.pkl')
        
        try:
            with open(iso_path, 'rb') as f:
                self.isolation_forest = pickle.load(f)
            with open(scaler_path, 'rb') as f:
                self.scaler = pickle.load(f)
            with open(vectorizer_path, 'rb') as f:
                self.vectorizer = pickle.load(f)
        except FileNotFoundError:
            # Create new models with default parameters
            self.isolation_forest = IsolationForest(contamination=0.1, random_state=42)
            self.scaler = StandardScaler()
            self.vectorizer = TfidfVectorizer(max_features=1000, ngram_range=(1,2))
    
    def extract_features(self, request_data):
        """Extract numerical and text features from HTTP request"""
        features = []
        
        # Basic request metrics
        features.append(len(request_data.get('uri', '')))
        features.append(len(request_data.get('query_string', '')))
        features.append(len(request_data.get('request_body', '')))
        features.append(len(request_data.get('headers', {})))
        
        # Character distribution anomalies
        uri = request_data.get('uri', '')
        features.append(uri.count('/'))
        features.append(uri.count('.'))
        features.append(uri.count('?'))
        features.append(uri.count('&'))
        features.append(uri.count('%'))
        
        # Entropy calculation for randomness detection
        if uri:
            entropy = self.calculate_entropy(uri)
            features.append(entropy)
        else:
            features.append(0)
            
        return np.array(features).reshape(1, -1)
    
    def calculate_entropy(self, text):
        """Calculate Shannon entropy of text"""
        if not text:
            return 0
        prob = [text.count(c) / len(text) for c in set(text)]
        entropy = -sum(p * np.log2(p) for p in prob if p > 0)
        return entropy
    
    def detect_anomaly(self, request_data):
        """Main anomaly detection function"""
        try:
            # Extract numerical features
            numerical_features = self.extract_features(request_data)
            
            # Extract text features
            text_data = ' '.join([
                request_data.get('uri', ''),
                request_data.get('query_string', ''),
                request_data.get('request_body', '')[:1000]  # Limit body size
            ])
            
            # Check if models need training (first run)
            if not hasattr(self.isolation_forest, 'decision_function'):
                # Use current request as baseline (in production, train on clean data)
                scaled_features = self.scaler.fit_transform(numerical_features)
                text_features = self.vectorizer.fit_transform([text_data])
                combined_features = np.hstack([scaled_features, text_features.toarray()])
                self.isolation_forest.fit(combined_features)
                self.save_models()
                return {'anomaly_score': 0, 'is_anomaly': False}
            
            # Transform features
            scaled_features = self.scaler.transform(numerical_features)
            text_features = self.vectorizer.transform([text_data])
            combined_features = np.hstack([scaled_features, text_features.toarray()])
            
            # Get anomaly score
            anomaly_score = self.isolation_forest.decision_function(combined_features)[0]
            is_anomaly = self.isolation_forest.predict(combined_features)[0] == -1
            
            # Convert to 0-10 scale for ModSecurity
            normalized_score = max(0, min(10, (1 - anomaly_score) * 5))
            
            return {
                'anomaly_score': normalized_score,
                'is_anomaly': is_anomaly,
                'raw_score': anomaly_score
            }
            
        except Exception as e:
            return {'anomaly_score': 0, 'is_anomaly': False, 'error': str(e)}
    
    def save_models(self):
        """Save trained models to disk"""
        os.makedirs(self.model_path, exist_ok=True)
        with open(os.path.join(self.model_path, 'isolation_forest.pkl'), 'wb') as f:
            pickle.dump(self.isolation_forest, f)
        with open(os.path.join(self.model_path, 'scaler.pkl'), 'wb') as f:
            pickle.dump(self.scaler, f)
        with open(os.path.join(self.model_path, 'vectorizer.pkl'), 'wb') as f:
            pickle.dump(self.vectorizer, f)

def main():
    if len(sys.argv) < 2:
        print(json.dumps({'error': 'No request data provided'}))
        return
    
    try:
        request_data = json.loads(sys.argv[1])
        detector = ModSecurityMLDetector()
        result = detector.detect_anomaly(request_data)
        print(json.dumps(result))
    except Exception as e:
        print(json.dumps({'error': str(e), 'anomaly_score': 0}))

if __name__ == '__main__':
    main()

Make the ML script executable

Set proper permissions for the machine learning detection script.

sudo chmod +x /etc/modsecurity/ml_detector.py
sudo chown www-data:www-data /etc/modsecurity/ml_detector.py

Create ModSecurity ML integration rules

Configure ModSecurity rules that call the machine learning script and act on anomaly scores.

# Machine Learning Anomaly Detection Rules

# Phase 1: Initialize ML variables
SecRule REQUEST_METHOD "@unconditionalMatch" \
    "id:100001,\
    phase:1,\
    nolog,\
    pass,\
    t:none,\
    setvar:'tx.ml_request_data={\"uri\":\"%{REQUEST_URI}\",\"query_string\":\"%{QUERY_STRING}\",\"method\":\"%{REQUEST_METHOD}\",\"headers\":{\"user_agent\":\"%{REQUEST_HEADERS.User-Agent}\",\"host\":\"%{REQUEST_HEADERS.Host}\"},\"request_body\":\"%{REQUEST_BODY}\"}'"

# Phase 2: Execute ML detection
SecRule REQUEST_METHOD "@unconditionalMatch" \
    "id:100002,\
    phase:2,\
    pass,\
    t:none,\
    exec:/etc/modsecurity/ml_detector.py '%{tx.ml_request_data}',\
    setvar:'tx.ml_result=%{EXEC}'"

# Phase 2: Parse ML results and set anomaly score
SecRule TX:ml_result "@rx \"anomaly_score\":\s*(\d+(?:\.\d+)?)" \
    "id:100003,\
    phase:2,\
    capture,\
    pass,\
    t:none,\
    setvar:'tx.ml_anomaly_score=%{tx.1}',\
    setvar:'tx.anomaly_score=+%{tx.1}',\
    logdata:'ML Anomaly Score: %{tx.ml_anomaly_score}'"

# Phase 2: High anomaly score detection
SecRule TX:ml_anomaly_score "@gt 7" \
    "id:100004,\
    phase:2,\
    block,\
    msg:'High ML anomaly score detected',\
    logdata:'ML Score: %{tx.ml_anomaly_score}, Request: %{tx.ml_request_data}',\
    setvar:'tx.anomaly_score=+5'"

# Phase 2: Medium anomaly score detection
SecRule TX:ml_anomaly_score "@ge 5" \
    "id:100005,\
    phase:2,\
    pass,\
    msg:'Medium ML anomaly score detected',\
    logdata:'ML Score: %{tx.ml_anomaly_score}',\
    setvar:'tx.anomaly_score=+3'"

# Phase 5: Block based on cumulative anomaly score
SecRule TX:ANOMALY_SCORE "@ge %{tx.inbound_anomaly_score_threshold}" \
    "id:100006,\
    phase:2,\
    block,\
    msg:'Inbound Anomaly Score Exceeded (Total Score: %{tx.anomaly_score})',\
    logdata:'Total anomaly score: %{tx.anomaly_score}, ML contribution: %{tx.ml_anomaly_score}'"

# Learning mode rule - log but don't block
SecRule TX:ml_anomaly_score "@gt 8" \
    "id:100007,\
    phase:5,\
    pass,\
    msg:'ML Learning Mode - High Anomaly Detected',\
    logdata:'Learning: ML Score %{tx.ml_anomaly_score}, URI: %{REQUEST_URI}, IP: %{REMOTE_ADDR}'"

Configure Apache virtual host with ModSecurity

Set up Apache virtual host to use ModSecurity with machine learning enabled.

Create automated threat response script

Build a script that monitors ModSecurity logs and automatically responds to threats.

#!/usr/bin/env python3
import re
import json
import subprocess
import time
from datetime import datetime, timedelta
from collections import defaultdict
import os

class ThreatResponseSystem:
    def __init__(self):
        self.log_file = '/var/log/modsecurity/audit.log'
        self.blocked_ips = set()
        self.threat_counts = defaultdict(int)
        self.last_check = datetime.now()
        
    def parse_modsec_log(self, log_line):
        """Parse ModSecurity audit log entries"""
        try:
            # Extract relevant information from audit log
            ip_match = re.search(r'"client_ip":"([^"]+)"', log_line)
            score_match = re.search(r'ML Score: ([\d.]+)', log_line)
            rule_match = re.search(r'id "(\d+)"', log_line)
            
            if ip_match:
                return {
                    'ip': ip_match.group(1),
                    'ml_score': float(score_match.group(1)) if score_match else 0,
                    'rule_id': rule_match.group(1) if rule_match else None,
                    'timestamp': datetime.now()
                }
        except Exception as e:
            print(f"Error parsing log: {e}")
        return None
    
    def should_block_ip(self, ip, ml_score):
        """Determine if IP should be blocked based on threat score and frequency"""
        self.threat_counts[ip] += 1
        
        # Block criteria
        if ml_score > 8:  # Very high anomaly score
            return True
        if self.threat_counts[ip] > 5 and ml_score > 6:  # Repeated medium threats
            return True
        if self.threat_counts[ip] > 10:  # Too many requests total
            return True
            
        return False
    
    def block_ip(self, ip):
        """Block IP using iptables"""
        if ip in self.blocked_ips:
            return False
            
        try:
            # Add iptables rule
            subprocess.run([
                'iptables', '-I', 'INPUT', '-s', ip, '-j', 'DROP'
            ], check=True)
            
            self.blocked_ips.add(ip)
            print(f"Blocked IP: {ip}")
            
            # Log the action
            with open('/var/log/modsecurity/threat_response.log', 'a') as f:
                f.write(f"{datetime.now().isoformat()} - BLOCKED {ip}\n")
                
            return True
        except subprocess.CalledProcessError as e:
            print(f"Failed to block IP {ip}: {e}")
            return False
    
    def unblock_ip(self, ip):
        """Remove IP from blocked list after timeout"""
        try:
            subprocess.run([
                'iptables', '-D', 'INPUT', '-s', ip, '-j', 'DROP'
            ], check=True)
            
            if ip in self.blocked_ips:
                self.blocked_ips.remove(ip)
            print(f"Unblocked IP: {ip}")
            return True
        except subprocess.CalledProcessError:
            return False
    
    def cleanup_old_blocks(self):
        """Remove IP blocks older than 1 hour"""
        # This is a simplified version - in production, track block timestamps
        cutoff_time = datetime.now() - timedelta(hours=1)
        
        # Clean up threat counts older than 24 hours
        if datetime.now() - self.last_check > timedelta(hours=24):
            self.threat_counts.clear()
            self.last_check = datetime.now()
    
    def monitor_logs(self):
        """Main monitoring loop"""
        print("Starting threat response monitoring...")
        
        if not os.path.exists(self.log_file):
            print(f"Log file {self.log_file} not found")
            return
            
        # Follow log file
        subprocess.Popen(['touch', self.log_file])
        
        with open(self.log_file, 'r') as f:
            # Go to end of file
            f.seek(0, 2)
            
            while True:
                line = f.readline()
                if line:
                    parsed = self.parse_modsec_log(line)
                    if parsed and parsed['ml_score'] > 5:
                        ip = parsed['ip']
                        score = parsed['ml_score']
                        
                        print(f"Threat detected: IP {ip}, ML Score: {score}")
                        
                        if self.should_block_ip(ip, score):
                            self.block_ip(ip)
                else:
                    time.sleep(1)
                    self.cleanup_old_blocks()

if __name__ == '__main__':
    try:
        response_system = ThreatResponseSystem()
        response_system.monitor_logs()
    except KeyboardInterrupt:
        print("\nT