166 lines
6.5 KiB
Python
166 lines
6.5 KiB
Python
"""로그인, 재인증, 세션 및 비밀번호 API."""
|
|
|
|
from datetime import datetime, timedelta
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from common_util.common_util_auth import (
|
|
create_session,
|
|
delete_session_cookie,
|
|
generate_otp,
|
|
hash_password,
|
|
hash_user_agent,
|
|
set_session_cookie,
|
|
verify_password,
|
|
verify_session,
|
|
)
|
|
from common_util.common_util_auth_repository import (
|
|
change_password,
|
|
clear_login_failures,
|
|
consume_otp,
|
|
delete_session,
|
|
get_active_otp,
|
|
get_user_by_email,
|
|
has_known_browser,
|
|
record_failed_login,
|
|
record_login,
|
|
replace_otp,
|
|
trust_browser,
|
|
)
|
|
from common_util.common_util_email import send_email_background
|
|
from common_util.common_util_email_templates import otp_email, security_alert_email
|
|
from config.config_system import ADMIN_EMAIL, EMAIL_REVERIFY_DAYS
|
|
|
|
from .A06_Login_Schema import LoginRequest, OtpVerifyRequest, PasswordChangeRequest
|
|
|
|
router = APIRouter(prefix="/api/auth", tags=["Authentication"])
|
|
|
|
|
|
def _user_agent(request: Request) -> str:
|
|
return request.headers.get("user-agent", "unknown")[:1000]
|
|
|
|
|
|
async def _send_otp(user: dict, purpose: str, label: str) -> None:
|
|
code = generate_otp()
|
|
await replace_otp(user["id"], purpose, hash_password(code))
|
|
subject, html = otp_email(code, label)
|
|
send_email_background(user["email"], subject, html)
|
|
|
|
|
|
@router.post("/login/request")
|
|
async def request_login(payload: LoginRequest, request: Request):
|
|
email = str(payload.email).lower()
|
|
agent = _user_agent(request)
|
|
user = await get_user_by_email(email)
|
|
if not user:
|
|
await record_login(None, email, "FAILURE", "ACCOUNT_NOT_FOUND", agent)
|
|
raise HTTPException(status_code=401, detail="이메일 또는 비밀번호가 올바르지 않습니다.")
|
|
if user["account_locked_until"] and user["account_locked_until"] > datetime.utcnow():
|
|
await record_login(user["id"], email, "FAILURE", "ACCOUNT_LOCKED", agent)
|
|
raise HTTPException(status_code=423, detail="로그인 실패 누적으로 계정이 잠겼습니다.")
|
|
if not verify_password(payload.password, user["password_hash"]):
|
|
failures = await record_failed_login(user["id"])
|
|
await record_login(user["id"], email, "FAILURE", "INVALID_PASSWORD", agent)
|
|
if failures >= 5 and ADMIN_EMAIL:
|
|
subject, html = security_alert_email(email, failures)
|
|
send_email_background(ADMIN_EMAIL, subject, html)
|
|
raise HTTPException(status_code=401, detail="이메일 또는 비밀번호가 올바르지 않습니다.")
|
|
if user["status"] != "ACTIVE":
|
|
await record_login(user["id"], email, "FAILURE", "ACCOUNT_INACTIVE", agent)
|
|
raise HTTPException(status_code=403, detail="활성화되지 않은 계정입니다.")
|
|
|
|
reverify_before = datetime.utcnow() - timedelta(days=EMAIL_REVERIFY_DAYS)
|
|
periodic_reverify = (
|
|
user["last_email_verified_at"] is None or user["last_email_verified_at"] < reverify_before
|
|
)
|
|
new_browser = not await has_known_browser(user["id"], hash_user_agent(agent))
|
|
if periodic_reverify or new_browser:
|
|
await _send_otp(user, "LOGIN", "로그인")
|
|
return {
|
|
"status": "otp_required",
|
|
"reason": "PERIODIC" if periodic_reverify else "NEW_BROWSER",
|
|
}
|
|
return await _finish_login(user, agent)
|
|
|
|
|
|
async def _finish_login(user: dict, agent: str) -> JSONResponse:
|
|
await clear_login_failures(user["id"])
|
|
await trust_browser(user["id"], hash_user_agent(agent))
|
|
session_id = await create_session(user["id"], agent)
|
|
await record_login(user["id"], user["email"], "SUCCESS", None, agent)
|
|
response = JSONResponse(
|
|
{
|
|
"status": "success",
|
|
"user": {"id": user["id"], "email": user["email"], "role": user["role"]},
|
|
}
|
|
)
|
|
set_session_cookie(response, session_id)
|
|
return response
|
|
|
|
|
|
@router.post("/login/verify")
|
|
async def verify_login(payload: OtpVerifyRequest, request: Request):
|
|
user = await get_user_by_email(str(payload.email).lower())
|
|
otp = await get_active_otp(user["id"], "LOGIN") if user else None
|
|
if (
|
|
not user
|
|
or not otp
|
|
or otp["expires_at"] < datetime.utcnow()
|
|
or not verify_password(payload.otp_code, otp["otp_hash"])
|
|
):
|
|
if user:
|
|
failures = await record_failed_login(user["id"])
|
|
await record_login(
|
|
user["id"], user["email"], "FAILURE", "INVALID_OTP", _user_agent(request)
|
|
)
|
|
if failures >= 5 and ADMIN_EMAIL:
|
|
subject, html = security_alert_email(user["email"], failures)
|
|
send_email_background(ADMIN_EMAIL, subject, html)
|
|
raise HTTPException(status_code=400, detail="인증 코드가 유효하지 않습니다.")
|
|
await consume_otp(otp["id"])
|
|
from config.config_db import get_db_pool
|
|
|
|
pool = get_db_pool()
|
|
async with pool.acquire() as connection, connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"UPDATE users SET last_email_verified_at = CURRENT_TIMESTAMP WHERE id = %s",
|
|
(user["id"],),
|
|
)
|
|
await connection.commit()
|
|
return await _finish_login(user, _user_agent(request))
|
|
|
|
|
|
@router.get("/session")
|
|
async def current_session(session: dict = Depends(verify_session)):
|
|
return {
|
|
"status": "success",
|
|
"user": {
|
|
"id": session["user_id"],
|
|
"email": session["email"],
|
|
"name": session["name"],
|
|
"role": session["role"],
|
|
},
|
|
}
|
|
|
|
|
|
@router.post("/logout")
|
|
async def logout(request: Request):
|
|
session_id = request.cookies.get("session_id")
|
|
if session_id:
|
|
await delete_session(session_id)
|
|
response = JSONResponse({"status": "success"})
|
|
delete_session_cookie(response)
|
|
return response
|
|
|
|
|
|
@router.post("/password")
|
|
async def update_password(payload: PasswordChangeRequest, session: dict = Depends(verify_session)):
|
|
user = await get_user_by_email(session["email"])
|
|
if not user or not verify_password(payload.current_password, user["password_hash"]):
|
|
raise HTTPException(status_code=400, detail="현재 비밀번호가 올바르지 않습니다.")
|
|
if verify_password(payload.new_password, user["password_hash"]):
|
|
raise HTTPException(status_code=400, detail="기존 비밀번호와 다른 값을 사용하세요.")
|
|
await change_password(user["id"], hash_password(payload.new_password), payload.logout_all)
|
|
return {"status": "success", "reauthentication_required": True}
|