Aislo/A06_Login/A06_Login_Router.py

"""로그인, 재인증, 세션 및 비밀번호 API."""

import logging
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_device_token,
    generate_otp,
    get_device_token_cookie,
    hash_device_token,
    hash_password,
    hash_user_agent,
    set_device_token_cookie,
    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_trusted_device,
    record_failed_login,
    record_login,
    replace_otp,
    trust_device,
)
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, AUTH_OTP_DISABLED, EMAIL_REVERIFY_DAYS

from .A06_Login_Schema import LoginRequest, OtpVerifyRequest, PasswordChangeRequest

logger = logging.getLogger(__name__)

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"] not in ("ACTIVE", "NO_COMPANY", "PENDING"):
        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
    )
    device_token = get_device_token_cookie(request)
    device_token_hash = hash_device_token(device_token) if device_token else None
    trusted_device = bool(
        device_token_hash and await has_trusted_device(user["id"], device_token_hash)
    )
    new_browser = not trusted_device
    if periodic_reverify or new_browser:
        # 개발 단계에서는 메일 인증을 건너뛴다 (2026-09-13 사용자 지시).
        # 이 갈래는 ENVIRONMENT=development + AUTH_OTP_DISABLED=true 에서만 열린다.
        if AUTH_OTP_DISABLED:
            logger.warning(
                "[auth] 개발용 OTP 생략 — %s (%s)",
                email,
                "PERIODIC" if periodic_reverify else "NEW_BROWSER",
            )
        else:
            await _send_otp(user, "LOGIN", "로그인")
            return {
                "status": "otp_required",
                "reason": "PERIODIC" if periodic_reverify else "NEW_BROWSER",
            }
    return await _finish_login(user, agent, device_token_hash)


async def _finish_login(
    user: dict, agent: str, previous_device_token_hash: str | None = None
) -> JSONResponse:
    await clear_login_failures(user["id"])
    device_token = generate_device_token()
    await trust_device(
        user["id"],
        hash_device_token(device_token),
        hash_user_agent(agent),
        previous_device_token_hash,
    )
    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_login = CURRENT_TIMESTAMP WHERE id = %s",
            (user["id"],),
        )
        await connection.commit()
    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_device_token_cookie(response, device_token)
    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()
    device_token = get_device_token_cookie(request)
    previous_device_token_hash = hash_device_token(device_token) if device_token else None
    return await _finish_login(user, _user_agent(request), previous_device_token_hash)


@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"],
            "company_id": session["company_id"],
            "is_master": session["is_master"],
        },
    }


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