api/auth-service/app/services/auth.py

import hashlib
import json
import secrets
from datetime import datetime, timedelta, timezone
from urllib.error import HTTPError, URLError
from urllib.parse import quote
from urllib.request import urlopen

from fastapi import HTTPException, status
from sqlalchemy.exc import ProgrammingError, SQLAlchemyError
from sqlalchemy.orm import Session

from app.core.config import settings
from app.core.security import create_access_token, hash_password, verify_password
from app.models.email_verification_code import EmailVerificationCode
from app.models.password_reset_token import PasswordResetToken
from app.models.user import User
from app.repositories.user import create_user, get_user_by_email, get_user_by_id
from app.schemas.user import UserCreate
from app.services.email import send_email_verification_code, send_password_reset_email


def _normalize_email(email: str) -> str:
    return email.strip().lower()


def _hash_email_code(email: str, code: str) -> str:
    normalized_email = _normalize_email(email)
    return hashlib.sha256(f"{normalized_email}:{code}".encode("utf-8")).hexdigest()


def _hash_reset_token(token: str) -> str:
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def _verify_google_id_token(id_token: str) -> dict[str, str]:
    if not settings.google_oauth_client_id:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="Google login is not configured.",
        )

    url = "https://oauth2.googleapis.com/tokeninfo?id_token=" + quote(id_token.strip())
    try:
        with urlopen(url, timeout=10) as response:
            payload = json.loads(response.read().decode("utf-8"))
    except (HTTPError, URLError, TimeoutError, json.JSONDecodeError) as exc:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Google credential is invalid or expired.",
        ) from exc

    audience = payload.get("aud")
    issuer = payload.get("iss")
    email = payload.get("email")
    email_verified = str(payload.get("email_verified", "")).lower() == "true"
    expires_at = int(payload.get("exp", "0") or "0")
    now = int(datetime.now(timezone.utc).timestamp())

    if audience != settings.google_oauth_client_id:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Google credential audience does not match this app.",
        )
    if issuer not in {"accounts.google.com", "https://accounts.google.com"}:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Google credential issuer is invalid.",
        )
    if not email or not email_verified:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Google account email is not verified.",
        )
    if expires_at <= now:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Google credential is expired.",
        )

    return payload


def _is_missing_table_error(exc: ProgrammingError, table_name: str) -> bool:
    original = getattr(exc, "orig", None)
    sqlstate = getattr(original, "sqlstate", None) or getattr(original, "pgcode", None)
    if sqlstate == "42P01":
        return True
    return table_name in str(exc)


def _ensure_password_reset_table(db: Session) -> None:
    PasswordResetToken.__table__.create(bind=db.get_bind(), checkfirst=True)


def _ensure_email_verification_table(db: Session) -> None:
    EmailVerificationCode.__table__.create(bind=db.get_bind(), checkfirst=True)


def _invalidate_active_reset_tokens(
    db: Session,
    user_id: int,
    used_at: datetime,
    exclude_token_id: int | None = None,
) -> None:
    query = db.query(PasswordResetToken).filter(
        PasswordResetToken.user_id == user_id,
        PasswordResetToken.used_at.is_(None),
    )
    if exclude_token_id is not None:
        query = query.filter(PasswordResetToken.id != exclude_token_id)
    query.update({PasswordResetToken.used_at: used_at}, synchronize_session=False)


def _delete_pending_email_verification_codes(db: Session, email: str) -> None:
    db.query(EmailVerificationCode).filter(
        EmailVerificationCode.email == email,
        EmailVerificationCode.verified_at.is_(None),
    ).delete(synchronize_session=False)


def _require_verified_email_for_signup(db: Session, email: str) -> None:
    now = datetime.now(timezone.utc)
    verified_after = now - timedelta(
        minutes=settings.email_verification_verified_ttl_minutes
    )
    try:
        verified_record = (
            db.query(EmailVerificationCode)
            .filter(
                EmailVerificationCode.email == email,
                EmailVerificationCode.verified_at.is_not(None),
                EmailVerificationCode.verified_at >= verified_after,
                EmailVerificationCode.verified_at <= EmailVerificationCode.expires_at,
            )
            .order_by(EmailVerificationCode.verified_at.desc())
            .first()
        )
    except ProgrammingError as exc:
        if not _is_missing_table_error(exc, "email_verification_codes"):
            raise
        db.rollback()
        _ensure_email_verification_table(db)
        verified_record = None

    if not verified_record:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="이메일 인증을 완료해주세요.",
        )


def register_user(db: Session, user_in: UserCreate):
    normalized_email = _normalize_email(user_in.email)
    existing = get_user_by_email(db, normalized_email)
    if existing:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Email already registered",
        )

    _require_verified_email_for_signup(db, normalized_email)

    if len(user_in.password.encode("utf-8")) > 72:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail="비밀번호는 72바이트 이하로 입력해주세요.",
        )

    hashed = hash_password(user_in.password)
    user = create_user(
        db,
        normalized_email,
        hashed,
        user_in.first_name,
        user_in.last_name,
        user_in.phone,
        user_in.gender,
    )
    return user


def request_email_verification(db: Session, email: str) -> None:
    normalized_email = _normalize_email(email)
    existing = get_user_by_email(db, normalized_email)
    if existing:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="이미 등록된 이메일입니다.",
        )

    code = f"{secrets.randbelow(1_000_000):06d}"
    now = datetime.now(timezone.utc)
    expires_at = now + timedelta(minutes=settings.email_verification_code_expire_minutes)
    code_hash = _hash_email_code(normalized_email, code)

    for attempt in range(2):
        try:
            _delete_pending_email_verification_codes(db, normalized_email)
            db.add(
                EmailVerificationCode(
                    email=normalized_email,
                    code_hash=code_hash,
                    expires_at=expires_at,
                )
            )
            db.flush()
            break
        except ProgrammingError as exc:
            if not _is_missing_table_error(exc, "email_verification_codes") or attempt > 0:
                raise
            db.rollback()
            _ensure_email_verification_table(db)

    if not send_email_verification_code(normalized_email, code):
        db.rollback()
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="이메일 인증 메일 전송에 실패했습니다. 잠시 후 다시 시도해주세요.",
        )

    db.commit()


def confirm_email_verification(db: Session, email: str, code: str) -> None:
    normalized_email = _normalize_email(email)
    now = datetime.now(timezone.utc)

    try:
        verification = (
            db.query(EmailVerificationCode)
            .filter(
                EmailVerificationCode.email == normalized_email,
                EmailVerificationCode.verified_at.is_(None),
                EmailVerificationCode.expires_at >= now,
            )
            .order_by(EmailVerificationCode.created_at.desc())
            .with_for_update()
            .first()
        )
    except ProgrammingError as exc:
        if not _is_missing_table_error(exc, "email_verification_codes"):
            raise
        db.rollback()
        _ensure_email_verification_table(db)
        verification = None

    if not verification:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="인증번호가 틀립니다! 다시한번 확인해주세요.",
        )

    incoming_hash = _hash_email_code(normalized_email, code)
    if not secrets.compare_digest(verification.code_hash, incoming_hash):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="인증번호가 틀립니다! 다시한번 확인해주세요.",
        )

    verification.verified_at = now
    db.add(verification)
    db.commit()


def authenticate_user(db: Session, email: str, password: str):
    normalized_email = _normalize_email(email)
    user = get_user_by_email(db, normalized_email)
    if not user:
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user


def login(db: Session, email: str, password: str) -> str:
    user = authenticate_user(db, email, password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password",
        )
    return create_access_token(subject=str(user.id))


def login_with_google(db: Session, id_token: str) -> str:
    profile = _verify_google_id_token(id_token)
    normalized_email = _normalize_email(str(profile["email"]))
    user = get_user_by_email(db, normalized_email)

    try:
        if user is None:
            random_password = secrets.token_urlsafe(48)
            user = User(
                email=normalized_email,
                hashed_password=hash_password(random_password),
                first_name=profile.get("given_name"),
                last_name=profile.get("family_name"),
                phone=None,
                gender=None,
            )
            db.add(user)
            db.flush()
        elif not user.is_active:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="User account is inactive.",
            )

        access_token = create_access_token(subject=str(user.id))
        db.commit()
        return access_token
    except HTTPException:
        db.rollback()
        raise
    except SQLAlchemyError as exc:
        db.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Google login failed: {exc}",
        ) from exc


def change_password(
    db: Session,
    user: User,
    current_password: str,
    new_password: str,
) -> None:
    if not verify_password(current_password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="현재 비밀번호가 일치하지 않습니다.",
        )

    if verify_password(new_password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="기존 비밀번호와 새 비밀번호가 같습니다. 다른 비밀번호를 입력해주세요.",
        )

    if len(new_password.encode("utf-8")) > 72:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail="비밀번호는 72바이트 이하로 입력해주세요.",
        )

    user.hashed_password = hash_password(new_password)
    now = datetime.now(timezone.utc)
    try:
        _invalidate_active_reset_tokens(db, user.id, now)
    except ProgrammingError as exc:
        if not _is_missing_table_error(exc, "password_reset_tokens"):
            raise
        db.rollback()
        _ensure_password_reset_table(db)
    db.add(user)
    db.commit()
    db.refresh(user)


def request_password_reset(db: Session, email: str) -> None:
    normalized_email = _normalize_email(email)
    user = get_user_by_email(db, normalized_email)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="등록된 이메일이 아닙니다.",
        )

    now = datetime.now(timezone.utc)
    try:
        _invalidate_active_reset_tokens(db, user.id, now)
    except ProgrammingError as exc:
        if not _is_missing_table_error(exc, "password_reset_tokens"):
            raise
        db.rollback()
        _ensure_password_reset_table(db)

    raw_token = secrets.token_urlsafe(48)
    token_hash = _hash_reset_token(raw_token)
    expires_at = now + timedelta(minutes=settings.password_reset_token_expire_minutes)

    db.add(
        PasswordResetToken(
            user_id=user.id,
            token_hash=token_hash,
            expires_at=expires_at,
        )
    )
    db.commit()

    reset_url = f"{settings.password_reset_frontend_url}?token={quote(raw_token)}"
    send_password_reset_email(user.email, reset_url)


def reset_password_with_token(db: Session, token: str, new_password: str) -> None:
    token_hash = _hash_reset_token(token)
    now = datetime.now(timezone.utc)

    try:
        reset_token = (
            db.query(PasswordResetToken)
            .filter(
                PasswordResetToken.token_hash == token_hash,
                PasswordResetToken.used_at.is_(None),
                PasswordResetToken.expires_at >= now,
            )
            .with_for_update()
            .first()
        )
    except ProgrammingError as exc:
        if not _is_missing_table_error(exc, "password_reset_tokens"):
            raise
        db.rollback()
        _ensure_password_reset_table(db)
        reset_token = None
    if not reset_token:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="유효하지 않거나 만료된 인증 링크입니다.",
        )

    user = get_user_by_id(db, reset_token.user_id)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="유효하지 않거나 만료된 인증 링크입니다.",
        )

    if verify_password(new_password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="기존 비밀번호와 새 비밀번호가 같습니다. 다른 비밀번호를 입력해주세요.",
        )

    user.hashed_password = hash_password(new_password)
    reset_token.used_at = now

    _invalidate_active_reset_tokens(db, user.id, now, exclude_token_id=reset_token.id)

    db.add(user)
    db.add(reset_token)
    db.commit()


def withdraw_user(
    db: Session,
    user: User,
    confirm_text: str,
) -> None:
    if confirm_text.strip() != "회원탈퇴":
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail='회원탈퇴를 원하신다면 "회원탈퇴"라고 입력해주세요.',
        )

    try:
        db.delete(user)
        db.commit()
    except SQLAlchemyError as exc:
        db.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"회원탈퇴 처리 중 오류가 발생했습니다: {exc}",
        ) from exc