Broken user authentication

Description

Inadequate authentication can let an attacker access another user's account. Weak password policies, poor session management, and missing multifactor authentication can contribute to this risk.

Potential impact

Account compromise: An attacker may take over an account and access sensitive information.
Privilege escalation: Compromising an administrator account may allow changes to or damage to the system.
Data exposure: Sensitive data may become available to an attacker.

Remediation

Require sufficient password length and block known breached or common passwords. Do not judge strength solely by character-composition rules.
Track login attempts by account and client address in a shared store, applying delays or lockouts when limits are exceeded.
Use multifactor authentication for additional protection.
Implement session timeouts and defenses against session fixation.
Use established authentication and session libraries suitable for Django, Flask, or FastAPI.

Examples

These are authentication excerpts. Models, shared stores, routing, and dependencies require separate configuration. Apply CSRF protection to browser cookie authentication and use HTTPS. Generate session and token-signing secrets with sufficient randomness and inject them externally.

Django

Before

python

# Unsafe Django code
from django.contrib.auth import authenticate

def login_view(request):
    username = request.POST['username']
    password = request.POST['password']
    user = authenticate(request, username=username, password=password)
    if user is not None:
        # 로그인 성공
        ...
    else:
        # 로그인 실패
        ...

After

python

# Safe Django code
import hashlib
from django.contrib.auth import authenticate, login
from django.contrib.auth.decorators import login_required
from django.core.cache import cache
from django.http import HttpResponse
from django.views.decorators.http import require_POST

MAX_USER_ATTEMPTS = 5
MAX_IP_ATTEMPTS = 20
ATTEMPT_WINDOW = 5 * 60

def login_attempt_keys(request, username):
    remote_addr = request.META.get('REMOTE_ADDR', 'unknown')
    user_key = hashlib.sha256(username.casefold().encode()).hexdigest()
    ip_key = hashlib.sha256(remote_addr.encode()).hexdigest()
    return [f'login:user:{user_key}', f'login:ip:{ip_key}']

def increment_login_attempt(key):
    cache.add(key, 0, timeout=ATTEMPT_WINDOW)
    try:
        return cache.incr(key)
    except ValueError:
        cache.set(key, 1, timeout=ATTEMPT_WINDOW)
        return 1

def consume_login_attempt(user_key, ip_key):
    # IP 한도를 먼저 검사해 무작위 사용자명으로 캐시 키가 계속 생기는 것을 제한합니다.
    if increment_login_attempt(ip_key) > MAX_IP_ATTEMPTS:
        return False
    return increment_login_attempt(user_key) <= MAX_USER_ATTEMPTS

@require_POST
def login_view(request):
    username = request.POST.get('username', '')
    password = request.POST.get('password', '')
    user_attempt_key, ip_attempt_key = login_attempt_keys(request, username)
    if not consume_login_attempt(user_attempt_key, ip_attempt_key):
        return HttpResponse("Too many login attempts", status=429)

    user = authenticate(request, username=username, password=password)
    if user is None:
        return HttpResponse("Invalid credentials", status=401)

    # 성공한 계정의 카운터만 초기화합니다. IP 카운터는 공격자가 자신의
    # 계정으로 로그인해 우회하지 못하도록 만료될 때까지 유지합니다.
    cache.delete(user_attempt_key)
    login(request, user)
    return HttpResponse("Logged in")

@login_required
def sensitive_view(request):
    # 민감한 데이터 처리 로직
    return HttpResponse("Sensitive data")

Explanation:

Before: It checks the password but does not show login-attempt limits or login-session creation. Review the actual authentication flow, including controls in other layers.
After: A shared Django cache supporting atomic increments limits attempts by account and client address. Successful authentication resets only the account counter; the IP counter remains until expiry so logging in to an attacker's own account cannot bypass it. login() establishes the session and login_required protects sensitive views. Behind a reverse proxy, separately configure trustworthy client-address handling for REMOTE_ADDR.

Flask

Before

python

# Unsafe Flask code
from flask import Flask, request
from werkzeug.security import check_password_hash

app = Flask(__name__)

@app.route('/login', methods=['POST'])
def login():
    username = request.form['username']
    password = request.form['password']
    user = User.query.filter_by(username=username).first()
    if user and check_password_hash(user.password, password):
        # 로그인 성공
        ...
    else:
        # 로그인 실패
        ...

After

python

# Safe Flask code
import hashlib
import os
from flask import Flask, request, redirect, url_for
from werkzeug.security import check_password_hash, generate_password_hash
from flask_login import LoginManager, login_user, login_required
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

app = Flask(__name__)
flask_secret = os.environ['FLASK_SECRET_KEY']
if len(flask_secret) < 32:
    raise RuntimeError('FLASK_SECRET_KEY must contain at least 32 characters')
app.config.update(
    SECRET_KEY=flask_secret,
    SESSION_COOKIE_HTTPONLY=True,
    SESSION_COOKIE_SECURE=True,
    SESSION_COOKIE_SAMESITE='Lax',
)
login_manager = LoginManager()
login_manager.init_app(app)
dummy_password_hash = generate_password_hash(os.urandom(32).hex())
limiter = Limiter(
    key_func=get_remote_address,
    app=app,
    storage_uri=os.environ['RATELIMIT_STORAGE_URI'],
)

@login_manager.user_loader
def load_user(user_id):
    # User 모델은 UserMixin을 구현하고 안정적인 기본 키를 get_id()로 반환합니다.
    return User.query.filter_by(id=user_id).first()

def username_rate_key():
    username = request.form.get('username', '').casefold()[:128]
    return hashlib.sha256(username.encode()).hexdigest()

@app.route('/login', methods=['POST'])
@limiter.limit('20 per 5 minutes', key_func=get_remote_address)
@limiter.limit('5 per 5 minutes', key_func=username_rate_key)
def login():
    username = request.form.get('username', '')
    password = request.form.get('password', '')
    if not username or len(username) > 128 or not password or len(password) > 1024:
        return "Invalid credentials", 400

    user = User.query.filter_by(username=username).first()
    stored_hash = user.password if user is not None else dummy_password_hash
    password_is_valid = check_password_hash(stored_hash, password)
    if user is not None and password_is_valid:
        login_user(user)
        return redirect(url_for('dashboard'))
    return "Invalid credentials", 401

@app.route('/dashboard')
@login_required
def dashboard():
    # 민감한 데이터 처리 로직
    return "Sensitive data"

Explanation:

Before: It checks the password but does not show login-attempt limits or login-session creation. Review the actual authentication flow, including controls in other layers.
After: Flask-Limiter uses a shared external store to limit attempts separately by account and client address. A dummy hash is verified for nonexistent accounts. The session secret comes from the environment, cookie protection is configured, and user_loader restores the signed-in user on subsequent requests. Place @login_required after @app.route so Flask registers the protected handler.

FastAPI

Before

python

# Unsafe FastAPI code
from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/login")
async def login(request: Request):
    form_data = await request.form()
    username = form_data['username']
    password = form_data['password']
    user = get_user_by_username(username)
    if user and user.password == password:
        # 로그인 성공
        ...
    else:
        # 로그인 실패
        ...

After

python

# Safe FastAPI code
import hashlib
import os
from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.security import OAuth2PasswordRequestForm
from fastapi_login import LoginManager
from pwdlib import PasswordHash
from redis.asyncio import Redis

app = FastAPI()
jwt_secret = os.environ['JWT_SECRET']
if len(jwt_secret) < 32:
    raise RuntimeError('JWT_SECRET must contain at least 32 characters')
manager = LoginManager(jwt_secret, token_url="/auth/token")
password_hash = PasswordHash.recommended()
# 존재하지 않는 계정도 같은 비밀번호 해시 검증 경로를 거치게 합니다.
dummy_password_hash = password_hash.hash(os.urandom(32))
redis = Redis.from_url(os.environ['REDIS_URL'], decode_responses=True)

ATTEMPT_WINDOW = 5 * 60
MAX_USER_ATTEMPTS = 5
MAX_IP_ATTEMPTS = 20

CONSUME_ATTEMPT = """
local ip_attempts = redis.call('INCR', KEYS[1])
if ip_attempts == 1 then
  redis.call('EXPIRE', KEYS[1], ARGV[1])
end
if ip_attempts > tonumber(ARGV[2]) then
  return 0
end

local user_attempts = redis.call('INCR', KEYS[2])
if user_attempts == 1 then
  redis.call('EXPIRE', KEYS[2], ARGV[1])
end
if user_attempts > tonumber(ARGV[3]) then
  return 0
end
return 1
"""

async def consume_login_attempt(username, client_address):
    user_digest = hashlib.sha256(username.casefold().encode()).hexdigest()
    ip_digest = hashlib.sha256(client_address.encode()).hexdigest()
    user_key = f'login:user:{user_digest}'
    ip_key = f'login:ip:{ip_digest}'
    allowed = await redis.eval(
        CONSUME_ATTEMPT,
        2,
        ip_key,
        user_key,
        ATTEMPT_WINDOW,
        MAX_IP_ATTEMPTS,
        MAX_USER_ATTEMPTS,
    )
    return allowed == 1, user_key

@manager.user_loader()
def load_user(username: str):
    return get_user_by_username(username)

@app.post("/auth/token")
async def login(request: Request, form_data: OAuth2PasswordRequestForm = Depends()):
    username = form_data.username
    password = form_data.password
    if not username or len(username) > 128 or not password or len(password) > 1024:
        raise HTTPException(status_code=400, detail="Invalid credentials")

    # 전달 헤더를 직접 신뢰하지 않고 ASGI 서버가 검증해 제공한 client 주소를 사용합니다.
    client_address = request.client.host if request.client else 'unknown'
    allowed, user_attempt_key = await consume_login_attempt(username, client_address)
    if not allowed:
        raise HTTPException(status_code=429, detail="Too many login attempts")

    user = load_user(username)
    stored_hash = user.password_hash if user is not None else dummy_password_hash
    password_is_valid = password_hash.verify(password, stored_hash)
    if user is None or not password_is_valid:
        raise HTTPException(status_code=401, detail="Invalid credentials")

    # IP 카운터는 자신의 계정으로 로그인해 우회하지 못하도록 유지합니다.
    await redis.delete(user_attempt_key)
    access_token = manager.create_access_token(data={"sub": username})
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/dashboard")
async def dashboard(current_user = Depends(manager)):
    # 민감한 데이터 처리 로직
    return {"detail": "Sensitive data"}

Explanation:

Before: It compares plaintext passwords and has no attempt limit. Use password-hash verification and enforce authenticated access to protected requests.
After: One atomic Redis Lua script checks the client-address limit before account limits. Once the IP limit is exceeded, it creates no new username keys, limiting account spraying and cache-key growth. Success resets only the account counter, preventing an attacker from clearing the IP limit through their own account. A dummy hash gives nonexistent accounts the same verification path to make username enumeration harder. A sufficiently strong token-signing secret is injected from the environment, recommended password hashing replaces plaintext comparison, and authentication dependencies protect endpoints. Configure the ASGI server to trust forwarded headers only from trusted proxies.

References

OWASP: Broken Authentication
CWE: CWE-287: Improper Authentication
OWASP Authentication Cheat Sheet