Flask Security

Security is a critical aspect of web application development. This guide covers essential security practices for Flask applications.

Authentication and Authorization

Password Hashing

Never store passwords in plain text. Use a secure hashing library:

from werkzeug.security import generate_password_hash, check_password_hash

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password_hash = db.Column(db.String(128))
    
    def set_password(self, password):
        self.password_hash = generate_password_hash(password)
    
    def check_password(self, password):
        return check_password_hash(self.password_hash, password)

# Usage
user = User(username='john')
user.set_password('secret123')
db.session.add(user)
db.session.commit()

# Verify password
if user.check_password('secret123'):
    print('Password is correct')

Session Management

Use Flask-Login for secure session management:

from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required

login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'auth.login'

class User(UserMixin, db.Model):
    # ... model definition

@login_manager.user_loader
def load_user(user_id):
    return User.query.get(int(user_id))

@app.route('/login', methods=['POST'])
def login():
    user = User.query.filter_by(username=request.form['username']).first()
    if user and user.check_password(request.form['password']):
        login_user(user)
        return redirect(url_for('index'))
    return 'Invalid credentials'

@app.route('/logout')
@login_required
def logout():
    logout_user()
    return redirect(url_for('index'))

JWT Authentication

For API authentication, use JSON Web Tokens:

from flask_jwt_extended import JWTManager, create_access_token, jwt_required, get_jwt_identity

app.config['JWT_SECRET_KEY'] = 'super-secret-key'  # Change this!
jwt = JWTManager(app)

@app.route('/api/login', methods=['POST'])
def api_login():
    username = request.json.get('username')
    password = request.json.get('password')
    
    user = User.query.filter_by(username=username).first()
    if user and user.check_password(password):
        access_token = create_access_token(identity=username)
        return jsonify(access_token=access_token)
    
    return jsonify({"msg": "Bad credentials"}), 401

@app.route('/api/protected', methods=['GET'])
@jwt_required()
def protected():
    current_user = get_jwt_identity()
    return jsonify(logged_in_as=current_user)

CSRF Protection

Enable CSRF Protection

from flask_wtf.csrf import CSRFProtect

csrf = CSRFProtect()
csrf.init_app(app)

# In templates
<form method="post">
    {{ form.csrf_token }}
    <!-- form fields -->
</form>

Exempt API Routes

from flask_wtf.csrf import csrf_exempt

@app.route('/api/webhook', methods=['POST'])
@csrf_exempt
def webhook():
    # Process webhook
    return jsonify(success=True)

SQL Injection Prevention

Use Parameterized Queries

# Bad - vulnerable to SQL injection
username = request.form['username']
query = f"SELECT * FROM users WHERE username = '{username}'"
db.session.execute(query)

# Good - using SQLAlchemy ORM
username = request.form['username']
user = User.query.filter_by(username=username).first()

# Good - using parameterized raw SQL
username = request.form['username']
query = "SELECT * FROM users WHERE username = :username"
result = db.session.execute(query, {'username': username})

XSS (Cross-Site Scripting) Prevention

Auto-Escape Templates

Jinja2 auto-escapes by default, but be careful with |safe filter:

<!-- Safe - auto-escaped -->
<p>{{ user_input }}</p>

<!-- Dangerous - not escaped -->
<p>{{ user_input|safe }}</p>

<!-- Safe way to allow some HTML -->
<p>{{ user_input|striptags }}</p>

Sanitize User Input

from markupsafe import escape

@app.route('/comment', methods=['POST'])
def add_comment():
    comment = escape(request.form['comment'])
    # Save comment
    return 'Comment added'

Secure Headers

Implement Security Headers

from flask_talisman import Talisman

# Force HTTPS and set security headers
Talisman(app, 
    force_https=True,
    strict_transport_security=True,
    content_security_policy={
        'default-src': "'self'",
        'script-src': "'self' 'unsafe-inline'",
        'style-src': "'self' 'unsafe-inline'"
    }
)

# Or manually set headers
@app.after_request
def set_security_headers(response):
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['X-Frame-Options'] = 'SAMEORIGIN'
    response.headers['X-XSS-Protection'] = '1; mode=block'
    response.headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains'
    return response

CORS (Cross-Origin Resource Sharing)

Configure CORS Properly

from flask_cors import CORS

# Allow all origins (development only)
CORS(app)

# Production - specific origins
CORS(app, resources={
    r"/api/*": {
        "origins": ["https://yourdomain.com"],
        "methods": ["GET", "POST"],
        "allow_headers": ["Content-Type", "Authorization"]
    }
})

File Upload Security

Validate File Uploads

import os
from werkzeug.utils import secure_filename

ALLOWED_EXTENSIONS = {'txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif'}
MAX_FILE_SIZE = 16 * 1024 * 1024  # 16MB

def allowed_file(filename):
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS

@app.route('/upload', methods=['POST'])
def upload_file():
    if 'file' not in request.files:
        return 'No file part', 400
    
    file = request.files['file']
    
    if file.filename == '':
        return 'No selected file', 400
    
    if not allowed_file(file.filename):
        return 'File type not allowed', 400
    
    # Check file size
    file.seek(0, os.SEEK_END)
    file_length = file.tell()
    if file_length > MAX_FILE_SIZE:
        return 'File too large', 400
    file.seek(0)
    
    filename = secure_filename(file.filename)
    file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
    
    return 'File uploaded successfully'

Rate Limiting

Implement Rate Limiting

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

limiter = Limiter(
    app,
    key_func=get_remote_address,
    default_limits=["200 per day", "50 per hour"]
)

@app.route('/api/login', methods=['POST'])
@limiter.limit("5 per minute")
def login():
    # Login logic
    pass

@app.route('/api/data')
@limiter.limit("100 per hour")
def get_data():
    # Return data
    pass

Environment Variables and Secrets

Secure Configuration

import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    SECRET_KEY = os.environ.get('SECRET_KEY') or os.urandom(32)
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')
    JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY')
    
    # Never commit these to version control
    MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
    MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')

.env File (Never commit to Git)

# .env
SECRET_KEY=your-secret-key-here
DATABASE_URL=postgresql://user:pass@localhost/dbname
JWT_SECRET_KEY=jwt-secret-key
MAIL_USERNAME=your-email@example.com
MAIL_PASSWORD=your-email-password

.gitignore

.env
.env.local
.env.production
*.pyc
__pycache__/
instance/
.pytest_cache/

Input Validation

Validate All User Input

from flask_wtf import FlaskForm
from wtforms import StringField, IntegerField, validators

class UserForm(FlaskForm):
    username = StringField('Username', [
        validators.Length(min=3, max=20),
        validators.Regexp('^[A-Za-z0-9_]+$', message='Username must contain only letters, numbers, and underscores')
    ])
    age = IntegerField('Age', [
        validators.NumberRange(min=0, max=150)
    ])
    email = StringField('Email', [
        validators.Email(message='Invalid email address')
    ])

@app.route('/register', methods=['POST'])
def register():
    form = UserForm()
    if form.validate_on_submit():
        # Process registration
        pass
    else:
        return jsonify(errors=form.errors), 400

API Security

Implement API Key Authentication

from functools import wraps

def require_api_key(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        api_key = request.headers.get('X-API-Key')
        if not api_key or api_key != app.config['API_KEY']:
            return jsonify(error='Invalid API key'), 401
        return f(*args, **kwargs)
    return decorated_function

@app.route('/api/data')
@require_api_key
def api_data():
    return jsonify(data='sensitive information')

Logging and Monitoring

Log Security Events

import logging

security_logger = logging.getLogger('security')
security_logger.setLevel(logging.WARNING)

@app.route('/login', methods=['POST'])
def login():
    username = request.form['username']
    user = User.query.filter_by(username=username).first()
    
    if user and user.check_password(request.form['password']):
        login_user(user)
        app.logger.info(f'Successful login: {username}')
        return redirect(url_for('index'))
    else:
        security_logger.warning(f'Failed login attempt: {username} from {request.remote_addr}')
        return 'Invalid credentials', 401

Security Checklist

• [ ] Use HTTPS in production
• [ ] Hash passwords with strong algorithms
• [ ] Implement CSRF protection
• [ ] Validate and sanitize all user input
• [ ] Use parameterized queries to prevent SQL injection
• [ ] Set secure HTTP headers
• [ ] Implement rate limiting
• [ ] Use environment variables for secrets
• [ ] Keep dependencies updated
• [ ] Implement proper error handling (don't expose stack traces)
• [ ] Use secure session cookies
• [ ] Implement proper authentication and authorization
• [ ] Log security events
• [ ] Regular security audits
• [ ] Use Content Security Policy (CSP)

Security Testing

Test for Common Vulnerabilities

import unittest

class SecurityTestCase(unittest.TestCase):
    def test_sql_injection(self):
        # Test SQL injection prevention
        response = self.client.post('/login', data={
            'username': "admin' OR '1'='1",
            'password': 'anything'
        })
        self.assertNotEqual(response.status_code, 200)
    
    def test_xss(self):
        # Test XSS prevention
        response = self.client.post('/comment', data={
            'comment': '<script>alert("XSS")</script>'
        })
        self.assertNotIn(b'<script>', response.data)
    
    def test_csrf_protection(self):
        # Test CSRF protection
        response = self.client.post('/update-profile', data={
            'email': 'new@example.com'
        })
        self.assertEqual(response.status_code, 400)

Resources

• OWASP Top 10
• Flask Security Documentation
• Python Security Best Practices

Summary

Security should be a priority from the start of your project. Implement these practices to protect your Flask application and user data from common vulnerabilities. Remember to stay updated with the latest security advisories and regularly audit your application for potential security issues.