Authentication Security and OWASP Top 10 — Protecting Auth Implementations

from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError

ph = PasswordHasher(
    time_cost=3,        # Number of iterations
    memory_cost=65536,  # 64 MB memory usage
    parallelism=4,      # Number of parallel threads
    hash_len=32,        # Output hash length
    salt_len=16         # Random salt length
)

def hash_password(password: str) -> str:
    """Hash password using Argon2id (OWASP recommended)."""
    return ph.hash(password)

def verify_password(password: str, password_hash: str) -> bool:
    """Verify password against Argon2id hash."""
    try:
        return ph.verify(password_hash, password)
    except VerifyMismatchError:
        return False

# Usage
password_hash = hash_password("user-password-123")
# $argon2id$v=19$m=65536,t=3,p=4$...

is_valid = verify_password("user-password-123", password_hash)
# True

is_valid = verify_password("wrong-password", password_hash)
# False

import time
from collections import defaultdict

class BruteForceProtector:
    """Multi-layer brute force protection."""

    def __init__(self):
        self.ip_attempts = defaultdict(list)
        self.user_attempts = defaultdict(list)
        self.global_attempts = []

    def check_ip(self, ip: str) -> bool:
        """Check if IP is rate limited."""
        now = time.time()
        recent = [t for t in self.ip_attempts[ip] if t > now - 900]  # 15 min window

        if len(recent) >= 20:
            return False  # Blocked

        # Exponential backoff
        if len(recent) >= 10:
            wait = min(300, 2 ** (len(recent) - 10))
            return time.time() - recent[-1] > wait

        return True

    def check_user(self, username: str) -> bool:
        """Check if user account is locked."""
        now = time.time()
        recent = [t for t in self.user_attempts[username] if t > now - 300]

        if len(recent) >= 5:
            return False  # Account locked for 5 minutes

        return True

    def record_failure(self, ip: str, username: str = None):
        """Record a failed authentication attempt."""
        now = time.time()
        self.ip_attempts[ip].append(now)
        if username:
            self.user_attempts[username].append(now)
        self.global_attempts.append(now)

    def record_success(self, ip: str, username: str):
        """Clear failure counts on successful login."""
        self.ip_attempts[ip] = []
        self.user_attempts[username] = []

protector = BruteForceProtector()

@app.route("/api/auth/login", methods=["POST"])
def login():
    ip = request.remote_addr
    username = request.json.get("username", "")

    if not protector.check_ip(ip):
        return jsonify({
            "error": "rate_limited",
            "message": "Too many attempts from this IP. Try again later."
        }), 429

    if not protector.check_user(username):
        return jsonify({
            "error": "account_locked",
            "message": "Account temporarily locked. Try again in 5 minutes."
        }), 423

    if validate_credentials(username, request.json.get("password", "")):
        protector.record_success(ip, username)
        return jsonify({"access_token": issue_token(username)})
    else:
        protector.record_failure(ip, username)
        return jsonify({"error": "Invalid credentials"}), 401

from flask import Flask, request, jsonify, make_response

app = Flask(__name__)

@app.after_request
def add_security_headers(response):
    """Add OWASP-recommended security headers."""
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    response.headers["Strict-Transport-Security"] = \
        "max-age=31536000; includeSubDomains"
    response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate"
    response.headers["Pragma"] = "no-cache"

    # Add auth-related headers
    if request.path.startswith("/api/auth/"):
        response.headers["X-Auth-Timestamp"] = str(time.time())

    return response

@app.after_request
def prevent_session_fixation(response):
    """Prevent session fixation by not accepting pre-established session IDs."""
    if request.path == "/api/auth/login" and response.status_code == 200:
        # Clear any existing session cookie
        response.set_cookie("session_id", "", expires=0, path="/")
    return response