OWASP Top 10: Your Essential Guide to Web Application Security Risks

from flask import Flask, request, session, jsonify
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
import bcrypt
import secrets
import time
import pyotp
from functools import wraps

app = Flask(__name__)
app.secret_key = secrets.token_hex(32)

# Rate limiting setup
limiter = Limiter(
    app,
    key_func=get_remote_address,
    default_limits=["1000 per hour"]
)

class SecureAuth:
    def __init__(self):
        self.failed_attempts = {}
        self.lockout_duration = 300  # 5 minutes
        self.max_attempts = 5
    
    def hash_password(self, password):
        """Secure password hashing with bcrypt"""
        salt = bcrypt.gensalt(rounds=12)
        return bcrypt.hashpw(password.encode('utf-8'), salt)
    
    def verify_password(self, password, hashed):
        """Verify password against hash"""
        return bcrypt.checkpw(password.encode('utf-8'), hashed)
    
    def is_account_locked(self, username):
        """Check if account is temporarily locked"""
        if username not in self.failed_attempts:
            return False
        
        attempts, last_attempt = self.failed_attempts[username]
        if attempts >= self.max_attempts:
            if time.time() - last_attempt < self.lockout_duration:
                return True
            else:
                # Reset after lockout period
                del self.failed_attempts[username]
        return False
    
    def record_failed_attempt(self, username):
        """Record failed login attempt"""
        current_time = time.time()
        if username in self.failed_attempts:
            attempts, _ = self.failed_attempts[username]
            self.failed_attempts[username] = (attempts + 1, current_time)
        else:
            self.failed_attempts[username] = (1, current_time)
    
    def generate_totp_secret(self):
        """Generate TOTP secret for 2FA"""
        return pyotp.random_base32()
    
    def verify_totp(self, secret, token):
        """Verify TOTP token for 2FA"""
        totp = pyotp.TOTP(secret)
        return totp.verify(token, valid_window=1)

auth = SecureAuth()

def require_auth(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if 'user_id' not in session:
            return jsonify({"error": "Unauthorized"}), 401
        return f(*args, **kwargs)
    return decorated_function

@app.route('/login', methods=['POST'])
@limiter.limit("5 per minute")
def login():
    data = request.get_json()
    username = data.get('username', '').strip()
    password = data.get('password', '')
    totp_token = data.get('totp_token', '')
    
    # Input validation
    if not username or not password:
        return jsonify({"error": "Username and password required"}), 400
    
    # Check for account lockout
    if auth.is_account_locked(username):
        return jsonify({"error": "Account temporarily locked"}), 423
    
    # Fetch user from database (implement your own)
    user = get_user_by_username(username)
    if not user or not auth.verify_password(password, user['password_hash']):
        auth.record_failed_attempt(username)
        return jsonify({"error": "Invalid credentials"}), 401
    
    # Verify 2FA if enabled
    if user.get('totp_secret'):
        if not totp_token or not auth.verify_totp(user['totp_secret'], totp_token):
            return jsonify({"error": "Invalid 2FA token"}), 401
    
    # Create secure session
    session['user_id'] = user['id']
    session['login_time'] = time.time()
    session.permanent = True
    
    return jsonify({"message": "Login successful", "user_id": user['id']})

@app.route('/logout', methods=['POST'])
@require_auth
def logout():
    session.clear()
    return jsonify({"message": "Logout successful"})

# Session security middleware
@app.before_request
def check_session_security():
    if 'user_id' in session:
        # Check session age
        login_time = session.get('login_time', 0)
        if time.time() - login_time > 3600:  # 1 hour timeout
            session.clear()
            return jsonify({"error": "Session expired"}), 401