Production authentication gotchas

# WRONG — reuses the session ID the client provides
@app.post('/login')
def login():
    session_id = request.cookies.get('session_id')  # attacker-provided
    session_id = session_id or generate_new_id()
    authenticate_user(email, password)
    sessions[session_id] = {'user_id': user.id}
    return {'session_id': session_id}
 
# CORRECT — always generates a new session after login
@app.post('/login')
def login():
    authenticate_user(email, password)
    new_session_id = generate_new_id()  # ignore client's session
    sessions[new_session_id] = {'user_id': user.id}
    response.set_cookie('session_id', new_session_id, httponly=True, secure=True)
    return 200

response.set_cookie('session_id', token, samesite='Strict')

@app.post('/transfer')
def transfer():
    provided_token = request.form.get('csrf_token')
    session_token = request.session.get('csrf_token')
    
    if provided_token != session_token:
        return 403  # CSRF attack
    
    # Process transfer
    perform_transfer(amount, recipient)
    return 200

# WRONG — early exit on first mismatch
def check_password(provided, stored):
    for i in range(len(provided)):
        if provided[i] != stored[i]:
            return False  # exits early if char doesn't match
    return True
 
# Timing:
# "password1": comparison stops after 1 char ≈ 1μs
# "passwor" (correct prefix): comparison goes to char 7 ≈ 7μs
# Attacker uses timing differences to guess the password character by character

import hmac
 
# CORRECT — takes the same time regardless of where mismatch occurs
def check_password(provided, stored):
    return hmac.compare_digest(provided, stored)

@app.post('/password-reset')
def password_reset():
    email = request.json.get('email')
    
    user = db.find_user_by_email(email)
    if user:
        send_reset_email(user)
    
    # ALWAYS return this, whether or not the user exists
    return 200, "If an account exists, a reset email has been sent"

# WRONG — predictable (sequential, guessable)
token = str(time.time())  # "1692921600.123"

import secrets
 
# CORRECT — cryptographically random, 32 bytes
token = secrets.token_urlsafe(32)
 
# Store with expiry
db.save_reset_token(email, token, expires_at=now() + 15min)
 
# Verify on reset
@app.post('/reset-password')
def reset_password():
    token = request.json.get('token')
    new_password = request.json.get('password')
    
    reset = db.get_reset_token(token)
    if not reset or reset.expires_at < now():
        return 400  # token invalid or expired
    
    user = db.get_user_by_id(reset.user_id)
    user.password_hash = bcrypt.hashpw(new_password.encode(), bcrypt.gensalt())
    db.delete_reset_token(token)
    return 200

import logging
 
class TokenFilter(logging.Filter):
    def filter(self, record):
        # Remove tokens from log messages
        record.msg = record.msg.replace(
            r'Bearer [^ ]+', 'Bearer [REDACTED]'
        )
        return True
 
logging.getLogger().addFilter(TokenFilter())

# WRONG — separate check and invalidate steps
@app.post('/exchange-code')
def exchange_code():
    code = request.json.get('code')
    
    if db.find_code(code):  # step 1: check
        token = generate_token()
        db.delete_code(code)  # step 2: invalidate
        return {'token': token}
    return 400
 
# CORRECT — atomic update in one query
@app.post('/exchange-code')
def exchange_code():
    code = request.json.get('code')
    
    # Atomic: delete and return in one step
    result = db.execute(
        'DELETE FROM oauth_codes WHERE code = ? RETURNING user_id',
        code
    )
    
    if not result:
        return 400  # code doesn't exist (already used or never existed)
    
    user_id = result[0]['user_id']
    token = generate_token()
    return {'token': token}