Denial Of Service in Flask

Denial of Service (DoS) attacks against Flask applications exploit the framework's synchronous nature and Python's Global Interpreter Lock (GIL). When a Flask route performs blocking operations—like database queries, file I/O, or external API calls—it holds the GIL, preventing other requests from being processed. A single slow endpoint can tie up your entire worker process, effectively denying service to legitimate users.

Flask's default development server (Werkzeug) is single-threaded, making it particularly vulnerable. Even with production servers like Gunicorn or uWSGI using multiple workers, Python's GIL means CPU-bound operations still serialize across workers. Attackers can exploit this by sending requests that trigger expensive operations or simply holding connections open.

Common Flask DoS vectors include:

• Expensive database queries: N+1 queries, missing indexes, or Cartesian products that consume memory and CPU
• Recursive endpoints: Routes that call themselves through redirects or API calls, creating infinite loops
• Large file uploads: Upload endpoints without size limits that exhaust disk space
• Memory exhaustion: Endpoints that load entire files or datasets into memory without streaming
• Connection pooling abuse: Holding database connections open indefinitely

Here's a vulnerable Flask endpoint that demonstrates several DoS risks:

from flask import Flask, request, jsonify
import time
import os

app = Flask(__name__)

@app.route('/vulnerable')
def vulnerable():
    # No timeout—attacker can hold connection open indefinitely
    time.sleep(30)
    
    # No size limit—attacker can upload massive files
    file = request.files.get('data')
    if file:
        content = file.read()  # Loads entire file into memory
        return jsonify(size=len(content))
    
    # Expensive operation without protection
    result = expensive_computation(100000)
    return jsonify(result=result)

def expensive_computation(n):
    return sum(i * i for i in range(n))

if __name__ == '__main__':
    app.run()

This endpoint is vulnerable to slowloris-style attacks (holding connections open), memory exhaustion (large file uploads), and CPU exhaustion (expensive computation). A single malicious request can block the worker for 30 seconds, during which time other requests queue up or fail.

Detecting DoS vulnerabilities in Flask requires both runtime monitoring and proactive scanning. For runtime detection, Flask's built-in logging and Werkzeug's request handling provide basic visibility, but you need additional tools for comprehensive coverage.

Monitoring key metrics helps identify DoS patterns:

from flask import Flask, request
import time
import logging

app = Flask(__name__)
logger = logging.getLogger('werkzeug')
logger.setLevel(logging.INFO)

# Track request durations
@app.before_request
def start_timer():
    request.start_time = time.time()

@app.after_request
def log_request(response):
    duration = time.time() - request.start_time
    if duration > 5:  # Log slow requests
        app.logger.warning(f"Slow request: {request.path} took {duration:.2f}s")
    return response

# Monitor memory usage
import psutil
@app.route('/health')
def health():
    process = psutil.Process()
    mem_info = process.memory_info()
    return jsonify(
        memory_mb=mem_info.rss / 1024 / 1024,
        cpu_percent=psutil.cpu_percent()
    )

For proactive scanning, middleBrick's black-box approach tests your Flask API's DoS resilience without requiring access to source code. It simulates various attack patterns including slow requests, large payloads, and resource-intensive operations.

middleBrick specifically tests for:

• Rate limiting bypass: Attempts to overwhelm your API with rapid requests
• Slowloris simulation: Holds connections open to test timeout handling
• Large payload testing: Sends oversized requests to check upload limits
• Recursive endpoint detection: Identifies routes that could create infinite loops
• Memory usage patterns: Analyzes endpoints for potential memory leaks

The CLI tool makes scanning straightforward:

npx middlebrick scan https://your-flask-app.com/api/vulnerable

This produces a security report with DoS-specific findings, severity levels, and remediation guidance. The scanner tests unauthenticated endpoints by default, which is crucial since DoS attacks often don't require authentication.

For CI/CD integration, the GitHub Action can fail builds if DoS vulnerabilities are detected:

name: API Security Scan
on: [pull_request]

jobs:
  security:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v2
    - name: Run middleBrick Scan
      run: |
        npx middlebrick scan https://staging.yourflaskapp.com \
          --fail-threshold F \
          --output json > results.json
    - name: Fail on high severity issues
      run: |
        if grep -q "CRITICAL\|HIGH" results.json; then
          exit 1
        fi

Securing Flask against DoS attacks requires a layered approach combining timeout handling, resource limits, and defensive coding patterns. Flask's extensibility makes it straightforward to implement these protections.

Timeout Protection

Implement request timeouts using Flask's teardown_request and Werkzeug's timeout middleware:

from flask import Flask, request, jsonify
from werkzeug.middleware.proxy_fix import ProxyFix
from werkzeug.exceptions import RequestTimeout
import signal
import errno
import os

app = Flask(__name__)
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1)

class TimeoutException(Exception):
    pass

def timeout_handler(signum, frame):
    raise TimeoutException()

signal.signal(signal.SIGALRM, timeout_handler)

@app.route('/safe')
def safe_endpoint():
    try:
        # Set 10-second timeout for this request
        signal.alarm(10)
        
        # Your endpoint logic here
        result = expensive_operation()
        return jsonify(result=result)
        
    except TimeoutException:
        raise RequestTimeout()
    finally:
        signal.alarm(0)  # Disable alarm

@app.errorhandler(RequestTimeout)
def handle_timeout(error):
    return jsonify(error="Request timeout"), 408

Resource Limits

Configure Flask to limit request sizes and protect against large payloads:

from flask import Flask, request
from werkzeug.exceptions import RequestEntityTooLarge

app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024  # 16MB limit

@app.errorhandler(RequestEntityTooLarge)
def handle_large_request(error):
    return jsonify(error="Request too large (max 16MB)"), 413

# Rate limiting with Flask-Limiter
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

limiter = Limiter(
    key_func=get_remote_address,
    default_limits=["200 per minute", "50 per second"]
)

@app.route('/api/protected')
@limiter.limit("10/minute;5/second")
def protected():
    return jsonify(message="Rate limited")

Memory-Safe File Handling

Stream files instead of loading them entirely into memory:

from flask import Flask, request, stream_with_context, Response
import hashlib

app = Flask(__name__)

@app.route('/upload', methods=['POST'])
def upload():
    if 'file' not in request.files:
        return jsonify(error="No file provided"), 400
    
    file = request.files['file']
    file_size = request.content_length
    
    if file_size > 16 * 1024 * 1024:  # 16MB max
        return jsonify(error="File too large"), 413
    
    # Stream processing instead of loading entire file
    def generate():
        hasher = hashlib.sha256()
        while chunk := file.read(8192):
            hasher.update(chunk)
            # Process chunk here instead of storing
        yield {'hash': hasher.hexdigest()}
    
    return Response(
        stream_with_context(generate()),
        mimetype='application/json'
    )

Database Query Protection

Prevent expensive queries with timeouts and result limits:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import time

engine = create_engine("postgresql://user:pass@db:5432/mydb")
Session = sessionmaker(bind=engine)

@app.route('/users')
def get_users():
    session = Session()
    
    try:
        # Set query timeout (database-specific)
        session.execute("SET statement_timeout = '10s'")
        
        # Use LIMIT to prevent large result sets
        users = session.query(User).limit(100).all()
        
        return jsonify([user.to_dict() for user in users])
        
    except Exception as e:
        app.logger.error(f"Database error: {e}")
        return jsonify(error="Database operation failed"), 500
    finally:
        session.close()

These patterns, combined with proper server configuration (like Gunicorn worker counts and timeout settings), create a robust defense against DoS attacks while maintaining Flask's development simplicity.