不適切なユーザー認証

説明

認証処理に不備があると、攻撃者が他のユーザーのアカウントへ不正にアクセスするおそれがあります。弱いパスワード方針、不適切なセッション管理、多要素認証の不足などが原因になり得ます。

想定される影響

アカウントの乗っ取り： 攻撃者がアカウントを乗っ取り、機密情報にアクセスする可能性があります。
権限昇格： 管理者アカウントが侵害されると、システムの変更や破壊につながる可能性があります。
情報漏えい： 機密情報が攻撃者に渡る可能性があります。

対処方法

十分なパスワード長を求め、漏えい済みや一般的なパスワードを拒否してください。文字の組み合わせだけで強度を判断しないでください。
共有ストアでアカウントとクライアントアドレス別にログイン試行を追跡し、上限を超えたら遅延やロックを適用してください。
多要素認証で保護を強化してください。
セッションのタイムアウトとセッション固定攻撃への対策を実装してください。
Django、Flask、FastAPIに適した実績のある認証・セッション管理ライブラリを使用してください。

例

認証処理の抜粋です。モデル、共有ストア、ルーティング、依存関係は別途構成してください。ブラウザーのCookie認証にはCSRF対策を適用し、HTTPSを使用してください。セッションやトークンの署名用シークレットは十分な乱数性を持たせて生成し、外部から注入してください。

Django

変更前

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:
        # 로그인 실패
        ...

変更後

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")

解説：

変更前： パスワードは確認しますが、ログイン試行の制限やログインセッションの作成は示されていません。別の層での制御も含め、実際の認証処理を確認する必要があります。
変更後： アトミックな増分処理に対応した共有Djangoキャッシュで、アカウントとクライアントアドレス別に試行を制限します。成功時はアカウントのカウンターだけをリセットし、IPカウンターは期限まで維持するため、攻撃者が自分のアカウントにログインしてIP制限を解除することを防ぎます。login()でセッションを作成し、login_requiredで機密性の高いビューを保護します。リバースプロキシ環境では、信頼できる実際のクライアントアドレスをREMOTE_ADDRへ渡す構成も必要です。

Flask

変更前

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:
        # 로그인 실패
        ...

変更後

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"

解説：

変更前： パスワードは確認しますが、ログイン試行の制限やログインセッションの作成は示されていません。別の層での制御も含め、実際の認証処理を確認する必要があります。
変更後： 外部の共有ストアを使うFlask-Limiterで、アカウントとクライアントアドレスの試行を別々に制限し、存在しないアカウントにもダミーハッシュの検証を行います。セッション用シークレットを環境から注入し、Cookieを保護する属性を設定して、user_loaderで後続リクエストのユーザーを復元します。@app.routeの後に@login_requiredを置き、Flaskに保護されたハンドラーを登録します。

FastAPI

変更前

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:
        # 로그인 실패
        ...

変更後

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"}

解説：

変更前： パスワードを平文で比較し、試行回数も制限していません。パスワードハッシュの検証と、保護対象への認証済みアクセスの制御が必要です。
変更後： 1つのアトミックなRedis Luaスクリプトでクライアントアドレスの上限を先に確認し、その後でアカウント別に制限します。IP上限を超えたら新しいユーザー名キーを作らず、複数アカウントへの試行とキャッシュキーの増加を抑えます。成功時もアカウントのカウンターだけをリセットするため、自分のアカウントでIP制限を解除できません。存在しないアカウントにもダミーハッシュで同じ検証を行い、ユーザー名の列挙を難しくします。十分に強い署名用シークレットを環境から注入し、平文比較を適切なパスワードハッシュ検証に置き換え、認証依存関係でエンドポイントを保護します。ASGIサーバーでは、信頼するプロキシからの転送ヘッダーだけを受け入れるようにしてください。

参考資料

OWASP: Broken Authentication
CWE: CWE-287: Improper Authentication
OWASP認証ガイド