security-patterns/dos-prevention.md

Denial of Service Prevention

Rule

Bound all resource consumption. Assume attackers will send worst-case input.

Source: CWE-400: Uncontrolled Resource Consumption

Request Limits

Correct Pattern

from functools import wraps
import time

# Rate limiting
class RateLimiter:
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window = window_seconds
        self.requests = {}  # ip -> [timestamps]
    
    def is_allowed(self, ip: str) -> bool:
        now = time.time()
        cutoff = now - self.window
        
        # Clean old entries
        self.requests[ip] = [
            t for t in self.requests.get(ip, [])
            if t > cutoff
        ]
        
        if len(self.requests[ip]) >= self.max_requests:
            return False
        
        self.requests[ip].append(now)
        return True

# Request size limits
MAX_BODY_SIZE = 10 * 1024 * 1024  # 10MB

@app.before_request
def limit_request_size():
    if request.content_length and request.content_length > MAX_BODY_SIZE:
        abort(413)  # Payload too large

Incorrect Pattern

# Wrong: no size limit
data = request.get_data()  # Could be gigabytes

# Wrong: unbounded loop based on user input
for i in range(int(request.args["count"])):
    process_item(i)

# Wrong: no timeout
response = requests.get(user_url)  # Hangs forever

Algorithmic Complexity

Correct Pattern

# Limit input size before expensive operations
MAX_ITEMS = 10000

def process_list(items: list) -> list:
    if len(items) > MAX_ITEMS:
        raise ValueError(f"Too many items: {len(items)} > {MAX_ITEMS}")
    return sorted(items)  # O(n log n) but bounded

# Use timeouts for expensive operations
import signal

def timeout_handler(signum, frame):
    raise TimeoutError("Operation timed out")

def with_timeout(seconds: int):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            signal.signal(signal.SIGALRM, timeout_handler)
            signal.alarm(seconds)
            try:
                return func(*args, **kwargs)
            finally:
                signal.alarm(0)
        return wrapper
    return decorator

@with_timeout(5)
def expensive_operation(data):
    ...

Incorrect Pattern

# Wrong: O(n²) or worse on unbounded input
def find_duplicates(items):
    for i in items:
        for j in items:  # O(n²)
            if i == j:
                yield i

# Wrong: regex with catastrophic backtracking
import re
pattern = re.compile(r'(a+)+$')  # ReDoS vulnerable
pattern.match('a' * 30 + 'b')  # Hangs

Memory Limits

Correct Pattern

# Stream large files instead of loading into memory
def process_large_file(path: str):
    with open(path, 'r') as f:
        for line in f:  # Streaming, constant memory
            process_line(line)

# Limit collection sizes
class BoundedCache:
    def __init__(self, max_size: int = 1000):
        self.max_size = max_size
        self.cache = {}
    
    def set(self, key, value):
        if len(self.cache) >= self.max_size:
            # Evict oldest
            oldest = next(iter(self.cache))
            del self.cache[oldest]
        self.cache[key] = value

Incorrect Pattern

# Wrong: loading entire file into memory
data = open(path).read()  # Could be huge

# Wrong: unbounded cache
cache = {}
def get_or_compute(key):
    if key not in cache:
        cache[key] = expensive_compute(key)  # Grows forever
    return cache[key]

Connection Limits

# Limit concurrent connections per IP
MAX_CONNECTIONS_PER_IP = 10

# Timeouts on all network operations
import socket
socket.setdefaulttimeout(30)

# Connection pooling with limits
from urllib3 import PoolManager
http = PoolManager(
    maxsize=100,
    block=True,
    timeout=30
)

Edge Cases

• Zip bombs (small file, huge uncompressed)
• XML entity expansion (billion laughs attack)
• Hash collision attacks (hash flooding)
• Slowloris (slow, incomplete requests)
• Amplification attacks (small request, large response)