Web Application Firewalls: Configuration and Bypass Preve...

Complete Production-Ready WAF System:

#!/usr/bin/env python3
"""
Production-ready Web Application Firewall (WAF)
Comprehensive rule engine with pattern matching, normalization, and rate limiting
"""

import re
import urllib.parse
import hashlib
import time
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, asdict
from enum import Enum
from collections import defaultdict
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class RuleAction(Enum):
    """WAF rule actions."""
    ALLOW = "allow"
    BLOCK = "block"
    LOG = "log"
    CHALLENGE = "challenge"
    RATE_LIMIT = "rate_limit"


class ThreatType(Enum):
    """Threat types."""
    SQL_INJECTION = "sql_injection"
    XSS = "xss"
    PATH_TRAVERSAL = "path_traversal"
    COMMAND_INJECTION = "command_injection"
    RFI_LFI = "rfi_lfi"
    RATE_LIMIT = "rate_limit"


@dataclass
class WAFRule:
    """WAF rule definition."""
    name: str
    action: RuleAction
    patterns: List[str]
    threat_type: ThreatType
    enabled: bool = True
    priority: int = 100


@dataclass
class WAFRequest:
    """WAF request object."""
    method: str
    path: str
    query_string: str
    headers: Dict[str, str]
    body: Optional[str] = None
    client_ip: str = "unknown"
    user_agent: str = "unknown"


@dataclass
class WAFDecision:
    """WAF decision result."""
    allowed: bool
    action: RuleAction
    matched_rule: Optional[str] = None
    threat_type: Optional[ThreatType] = None
    reason: Optional[str] = None


class InputNormalizer:
    """Normalize input to detect bypass attempts."""
    
    @staticmethod
    def normalize(input_string: str) -> str:
        """Normalize input string.
        
        Args:
            input_string: Input to normalize
            
        Returns:
            Normalized string
        """
        if not isinstance(input_string, str):
            return ""
        
        # URL decode
        try:
            decoded = urllib.parse.unquote(input_string)
        except:
            decoded = input_string
        
        # Multiple encoding decode
        for _ in range(5):  # Decode up to 5 levels
            try:
                decoded = urllib.parse.unquote(decoded)
            except:
                break
        
        # Lowercase
        normalized = decoded.lower()
        
        # Remove null bytes
        normalized = normalized.replace('\x00', '')
        
        # Remove SQL comments
        normalized = re.sub(r'/\*.*?\*/', '', normalized)
        normalized = normalized.replace('--', '')
        normalized = normalized.replace('#', '')
        
        # Normalize whitespace
        normalized = re.sub(r'\s+', ' ', normalized)
        
        # Remove SQL comment syntax
        normalized = normalized.replace('/**/', '')
        
        return normalized.strip()
    
    @staticmethod
    def normalize_path(path: str) -> str:
        """Normalize URL path.
        
        Args:
            path: URL path
            
        Returns:
            Normalized path
        """
        # Decode URL encoding
        normalized = urllib.parse.unquote(path)
        
        # Remove path traversal attempts
        normalized = re.sub(r'\.\./', '', normalized)
        normalized = re.sub(r'\.\.\\', '', normalized)
        
        return normalized


class WAFEngine:
    """Web Application Firewall engine."""
    
    def __init__(self):
        """Initialize WAF engine."""
        self.rules: List[WAFRule] = []
        self.normalizer = InputNormalizer()
        self.rate_limit_store: Dict[str, List[float]] = defaultdict(list)
        self._load_default_rules()
    
    def _load_default_rules(self):
        """Load default WAF rules."""
        self.rules = [
            WAFRule(
                name="SQL Injection - UNION SELECT",
                action=RuleAction.BLOCK,
                patterns=[
                    r'union.*select',
                    r'union.*all.*select',
                    r'select.*from',
                    r'insert.*into.*values',
                    r'delete.*from',
                    r'drop.*table',
                    r'update.*set',
                ],
                threat_type=ThreatType.SQL_INJECTION,
                priority=100
            ),
            WAFRule(
                name="XSS - Script Tags",
                action=RuleAction.BLOCK,
                patterns=[
                    r'<script[^>]*>',
                    r'javascript:',
                    r'on\w+\s*=',
                    r'<iframe[^>]*>',
                    r'<object[^>]*>',
                    r'<embed[^>]*>',
                ],
                threat_type=ThreatType.XSS,
                priority=100
            ),
            WAFRule(
                name="Path Traversal",
                action=RuleAction.BLOCK,
                patterns=[
                    r'\.\./',
                    r'\.\.\\',
                    r'/etc/passwd',
                    r'/proc/',
                    r'/windows/system32',
                ],
                threat_type=ThreatType.PATH_TRAVERSAL,
                priority=100
            ),
            WAFRule(
                name="Command Injection",
                action=RuleAction.BLOCK,
                patterns=[
                    r';\s*(rm|del|cat|ls|dir|pwd|whoami|id|uname)',
                    r'\|\s*(rm|del|cat|ls|dir)',
                    r'`[^`]+`',
                    r'\$\{[^}]+\}',
                ],
                threat_type=ThreatType.COMMAND_INJECTION,
                priority=100
            ),
        ]
    
    def add_rule(self, rule: WAFRule):
        """Add custom WAF rule.
        
        Args:
            rule: WAF rule to add
        """
        self.rules.append(rule)
        # Sort by priority (higher priority first)
        self.rules.sort(key=lambda r: r.priority, reverse=True)
    
    def check_rate_limit(self, client_ip: str, limit: int = 100, window: int = 60) -> bool:
        """Check if client exceeded rate limit.
        
        Args:
            client_ip: Client IP address
            limit: Request limit per window
            window: Time window in seconds
            
        Returns:
            True if within limit, False if exceeded
        """
        now = time.time()
        requests = self.rate_limit_store[client_ip]
        
        # Remove old requests outside window
        requests[:] = [req_time for req_time in requests if now - req_time < window]
        
        # Check limit
        if len(requests) >= limit:
            return False
        
        # Add current request
        requests.append(now)
        return True
    
    def evaluate(self, request: WAFRequest) -> WAFDecision:
        """Evaluate request against WAF rules.
        
        Args:
            request: WAF request object
            
        Returns:
            WAF decision
        """
        # Check rate limit
        if not self.check_rate_limit(request.client_ip, limit=100, window=60):
            logger.warning(f"Rate limit exceeded for {request.client_ip}")
            return WAFDecision(
                allowed=False,
                action=RuleAction.RATE_LIMIT,
                threat_type=ThreatType.RATE_LIMIT,
                reason="Rate limit exceeded"
            )
        
        # Collect all input strings to check
        inputs_to_check = [
            request.path,
            request.query_string,
            request.body or "",
            request.user_agent,
        ]
        
        # Add headers
        for header_value in request.headers.values():
            inputs_to_check.append(header_value)
        
        # Check each input against rules
        for rule in self.rules:
            if not rule.enabled:
                continue
            
            for input_str in inputs_to_check:
                # Normalize input
                normalized = self.normalizer.normalize(input_str)
                
                # Check patterns
                for pattern in rule.patterns:
                    if re.search(pattern, normalized, re.IGNORECASE):
                        logger.warning(
                            f"WAF rule '{rule.name}' matched for {request.client_ip} "
                            f"on pattern: {pattern}"
                        )
                        
                        return WAFDecision(
                            allowed=False,
                            action=rule.action,
                            matched_rule=rule.name,
                            threat_type=rule.threat_type,
                            reason=f"Matched pattern: {pattern}"
                        )
        
        # No matches - allow request
        return WAFDecision(
            allowed=True,
            action=RuleAction.ALLOW,
            reason="No threats detected"
        )


# Flask middleware integration
from flask import Flask, request, Response, jsonify

app = Flask(__name__)
waf = WAFEngine()


@app.before_request
def waf_middleware():
    """WAF middleware for Flask."""
    # Create WAF request
    waf_request = WAFRequest(
        method=request.method,
        path=request.path,
        query_string=request.query_string.decode('utf-8'),
        headers=dict(request.headers),
        body=request.get_data(as_text=True) if request.is_json else None,
        client_ip=request.remote_addr or request.headers.get('X-Forwarded-For', 'unknown'),
        user_agent=request.headers.get('User-Agent', 'unknown')
    )
    
    # Evaluate request
    decision = waf.evaluate(waf_request)
    
    if not decision.allowed:
        logger.warning(
            f"WAF blocked request from {waf_request.client_ip}: "
            f"{decision.reason}"
        )
        
        if decision.action == RuleAction.BLOCK:
            return jsonify({
                'error': 'Request blocked by security policy',
                'code': 'WAF_BLOCKED'
            }), 403
        
        elif decision.action == RuleAction.RATE_LIMIT:
            return jsonify({
                'error': 'Rate limit exceeded',
                'code': 'RATE_LIMIT'
            }), 429


@app.route('/api/data', methods=['GET', 'POST'])
def api_endpoint():
    """Example API endpoint protected by WAF."""
    return jsonify({'message': 'Success', 'data': 'protected'}), 200


if __name__ == '__main__':
    # Example usage
    test_request = WAFRequest(
        method='GET',
        path='/api/users',
        query_string='id=1 UNION SELECT * FROM users--',
        headers={'User-Agent': 'Mozilla/5.0'},
        client_ip='192.168.1.1'
    )
    
    decision = waf.evaluate(test_request)
    print(f"Decision: {decision.allowed}, Reason: {decision.reason}")

# Normalize input to prevent encoding bypasses
def normalize_input(input_string):
    """Normalize input to detect bypass attempts."""
    # Decode URL encoding
    decoded = urllib.parse.unquote(input_string)
    # Lowercase
    normalized = decoded.lower()
    # Remove whitespace
    normalized = normalized.replace(' ', '')
    return normalized

# Check normalized input against patterns
def check_waf_patterns(normalized_input):
    """Check normalized input against attack patterns."""
    patterns = ['union select', '<script>', 'javascript:']
    for pattern in patterns:
        if pattern in normalized_input:
            return True
    return False