[파이썬] FastAPI - 메모 앱 프로젝트 17: 소셜 로그인 탈퇴

이전 내용

[파이썬] FastAPI - 메모 앱 프로젝트 16: 메인 페이지 나누기

---

소셜 로그인 사용자 탈퇴 개요

 

이번에는 이 프로젝트의 마지막 (아마도...?) 과정인 소셜 로그인 (구글, 카카오, 네이버) 을 한 이용자들의 회원 탈퇴 기능을 구현해 본다. 현재 회원 탈퇴 기능이 있으나, 해당 기능은 소셜 로그인 사용자에게는 작동하지 않는다.

 

log를 확인해 보면 세션에 사용자 ID가 없다고 나온다.

 

대략적인 과정을 찾아보니,

 

- 소셜 로그인 사용자 삭제
소셜 로그인을 이용한 사용자의 경우, 기본적으로 사용자 정보가 소셜 플랫폼에 저장되므로, 해당 플랫폼의 사용자 정보를 삭제하고 싶은 경우, 해당 플랫폼의 API를 통해 회원 탈퇴를 처리해야 할 수 있다.

• 구글: Google API를 통해 계정 삭제 요청
• 네이버: Naver API를 통해 계정 삭제 요청
• 카카오: KaKao API를 통해 계정 삭제 요청

- 데이터베이스 처리
소셜 로그인 사용자의 정보를 데이터베이스에서 삭제하는 로직을 구현해야 한다. 필요한 경우, 회원 탈퇴 이유 등을 추가로 기록하실 수 있다.

---

소셜 로그인 탈퇴 구현
1. 로직 구현 방식 결정하기

 

소셜 로그인 탈퇴 기능을 구현할 때에는 두 가지 방식을 사용해 볼 수 있다.

1. 기존에 users_controller.py 내에 있는 소셜 로그인 함수와 회원 탈퇴 함수 보완
2. 컨트롤러를 일반 사용자와 소셜 로그인 사용자로 나눠 소셜 로그인 사용자에 대한 기능 추가

각 방식은 아래와 같은 장단점을 수반한다.

 

1. 기존 회원 탈퇴 로직 수정

 

[장점]

• 일관된 코드: 기존에 있는 로직을 수정하게 되면, 일반 사용자와 소셜 로그인 사용자 간의 탈퇴 프로세스가 일관되게 유지된다.
• 중복 코드 감소: 두 개의 유사한 로직을 관리할 필요가 없으므로 중복된 코드가 줄어든다.

[단점]

• 복잡성 증가: 기존 로직에 소셜 로그인 관련 처리를 추가하면서 로직이 복잡해질 수 있으며, 이는 가독성을 떨어뜨리고 유지 보수를 어렵게 만들 수 있다.

2. 새로운 로직 만들기

 

[장점]

• 명확한 분리: 일반 사용자와 소셜 로그인 사용자의 로직을 명확하게 분리함으로써, 각 로직의 이해와 관리가 쉬워진다.
• 특화된 처리 가능: 소셜 로그인 사용자에 대해 특별한 요구 사항이나 처리가 필요할 경우, 별도의 로직을 통해 이를 반영하기 용이하다.
• 확장성: 향후에 더 많은 소셜 로그인 제공자를 추가하거나, 각 사용자 유형에 따른 특별한 처리 로직이 필요한 경우, 각 컨트롤러에서 독립적으로 구현할 수 있다.

[단점]

• 중복 코드 발생 가능성: 두 개의 비슷한 로직이 존재하게 되어 코드의 중복이 발생할 수 있다.
• 유지 보수: 두 개의 서로 다른 로직을 유지해야 하므로, 향후 버그 수정이나 기능 개선 시 두 곳 모두에 반영할 필요가 있다.

3. 결론

일반적으로 소셜 로그인 사용자가 요구하는 추가적인 로직이 많다면 새로운 로직을 만드는 것이 좋다. 반면, 사용자의 탈퇴가 비슷한 형태를 띠고 있다면 기존 로직을 수정하여 일관성을 유지 하는 것이 좋다.

 

현재는 소셜 로그인 사용자의 로그인 및 회원 탈퇴 관련 사항만 다루는 것이므로, 기존 로직을 수정 및 보완하는 방식으로 진행해 보려고 한다.

 

---

소셜 로그인 탈퇴 구현
2. 컨트롤러 내 함수 보완하기

 

◆ users_controller.py

소셜 로그인 사용자와 기존 회원 탈퇴 로직 간의 연동이 되지 않는 것은

 

1. 사용자 식별 로직

현재 회원 탈퇴 로직에서 요청 세션에서 사용자 ID를 가져와 해당 사용자를 삭제하고 있으나, 소셜 로그인 사용자들은 그들의 ID가 소셜 로그인 시에 생성되므로, 일반 사용자와 동일한 방식으로 식별하지 않을 가능성이 있다.

 

2. 사용자 정보 조회 및 삭제

소셜 로그인 사용자의 경우, 일반 사용자와 다른 방식으로 데이터베이스에 저장된다. 

소셜 로그인 사용자를 처리하는 방식인 create_or_update_social_user() 에서 소셜 ID를 기반으로 사용자를 처리하나, 회원 탈퇴 로직에서는 ID를 기반으로 처리하므로 연동이 안 된다.

 

즉, 소셜 로그인 사용자 처리에서는 소셜 ID를, 회원 탈퇴 로직에서는 ID를 다루기 때문에 서로 연동이 되지 않는다 는 문제가 있다.

 

[해결 방안]

• 소셜 로그인 사용자의 세션에 소셜 ID 정보를 포함시키고, 이를 기반으로 사용자를 검색해야 한다.
• 회원 탈퇴 로직에서 소셜 ID를 사용하여 사용자를 조회하는 로직을 추가 한다.

- 소셜 로그인 후 사용자 정보를 처리하는 create_or_update_social_user() 함수 보완

# 소셜 로그인 후 사용자 정보 처리
def create_or_update_social_user(db: Session, user_info: dict, provider: str, request: Request):
    logger.info(f"소셜 로그인 처리 시작: 제공자 {provider}, 사용자 정보 {user_info}")

    # provider 별로 ID 및 이름 필드 설정
    social_id_filed = f"{provider}_id"
    user_name = (
        user_info.get('user_name') or
        user_info.get('profile_nickname') or
        user_info.get('name')
    )
    social_id_value = user_info.get(social_id_filed)

    if not user_info.get('email') or not social_id_value:
        logger.error("이메일 또는 소셜 ID가 제공되지 않았습니다.")
        raise ValueError('email 또는 소셜 ID가 제공되지 않았습니다.')
    
    # 기존 사용자 조회
    user = db.query(User).filter(
        (User.email == user_info['email']) | (getattr(User, social_id_filed) == social_id_value)).first()
    
    # 사용자 정보 저장 후 세션에 소셜 ID 저장
    request.session['social_id'] = social_id_value

    if not user:
        # 신규 사용자 생성
        user = User(
            username=user_info.get('username'),
            email=user_info['email'],
        )
        setattr(user, social_id_filed, social_id_value)
        db.add(user)
        logger.info(f"신규 사용자 생성: {user_info['email']} (소셜 ID: {social_id_value})")
    else:
        # 기존 사용자 정보 업데이트
        user.username = user_name or user.username  # 이름이 없으면 업데이트 하지 않음
        setattr(user, social_id_filed, social_id_value)
        db.merge(user)  # 업데이트 후 세션에 반영
        logger.info(f"기존 사용자 업데이트: {user.email} (소셜 ID: {social_id_value})")
    try:
        db.commit()
        logger.info(f"사용자 정보 저장 성공: {user.email}")
    except Exception as e:
        db.rollback() # 에러 발생 시 롤백
        logger.error(f"사용자 정보 저장 실패: {e}")
        raise

    db.refresh(user)
    return user

• 기존 코드에 사용자 정보를 저장 후 세션에 소셜 ID를 저장하는 부분 추가.
• 따라서, create_or_update_social_user() 함수의 인자에 request:Request 부분 추가.

- 회원 탈퇴 함수 delete_user() 보완

# 회원 탈퇴
@router.delete("/users/{id}")
async def delete_user(request: Request, id: int, background_tasks: BackgroundTasks, db: Session = Depends(get_db)):
    logger.info(f"회원 탈퇴 요청: 사용자 ID {id}")

    user_id = request.session.get("id") # 사용자 ID
    social_id = request.session.get("social_id") # 소셜 ID
    provider = request.session.get("provider") # 로그인 제공자 정보

    if user_id is None and (social_id is None or provider is None):
        logger.warning("세션에 사용자 ID 또는 소셜 ID가 없음.")
        raise HTTPException(status_code=401, detail="Not Authorized")

    # 사용자를 user_id 또는 social_id로 조회
    user_query = db.query(User)

    if user_id is not None:
        user_query = user_query.filter(User.id == user_id)

    if social_id is not None:
        if provider == "google":
            user_query = user_query.filter(User.google_id == social_id)
        elif provider == "kakao":
            user_query = user_query.filter(User.kakao_id == social_id)
        elif provider == "naver":
            user_query = user_query.filter(User.naver_id == social_id)
        else:
            logger.warning("지원하지 않는 소셜 로그인 제공자입니다.")
            raise HTTPException(status_code=403, detail="Unsupported Social Provider")

    user = user_query.first()

    if user is None:
        logger.error(f"사용자 찾기 실패: ID {user_id} 또는 {social_id}에 대한 사용자가 존재하지 않음.")
        raise HTTPException(status_code=404, detail="User를 찾을 수 없습니다.")

    # 사용자가 작성한 모든 메모 삭제
    memos = db.query(Memo).filter(Memo.user_id == user_id).all()
    logger.info(f"{len(memos)}개의 메모를 삭제합니다.")
    for memo in memos:
        db.delete(memo)

    # 사용자 정보 삭제
    db.delete(user)

    try:
        db.commit()
        logger.info(f"사용자 ID {user_id}가 성공적으로 삭제되었습니다.")
    except Exception as e:
        db.rollback()
        logger.error(f"회원 탈퇴 오류: {e}")  # 에러 내용 출력
        raise HTTPException(status_code=500, detail="회원 탈퇴에 실패하였습니다. 다시 시도해 주세요.")

    # 세션 비우기
    request.session.clear()

    # 탈퇴 안내 이메일 발송
    background_tasks.add_task(send_bye_email, user.email)

    # 탈퇴 성공 응답 리턴
    logger.info(f"회원 탈퇴 완료: 사용자 ID {user_id}에게 탈퇴 안내 이메일 발송")
    return {"success": True, "message": "회원 탈퇴가 완료되었습니다."}

• 세션에 소셜 ID와 공급자(provider)를 저장하도록 설정
• user_id is None 부분에 소셜 ID와 공급자를 추가
• 사용자를 조회하는 쿼리 부분에 소셜ID를 검색하는 부분을 추가하는 데, 조건문 적용 (공급자별)

---

ouath 디렉터리 내 py 파일 수정

 

그리고 oauth 디렉터리 내에 있는 각 공급자별 py 파일에 코드를 보완해 준다.

 

앞서 create_or_update_social_user() 함수의 인자에 request:Request 부분을 추가했기 때문에 마찬가지로, 각 파일에서 user 변수를 지정해주는 부분을 수정해준다.

user = create_or_update_social_user(db, user_info, provider='google', request=request)
        # 로그인 후 세션에 정보 저장
        request.session["id"] = user.id  # 일반 사용자 ID
        request.session["social_id"] = user_info["google_id"]  # 소셜 ID
        request.session["provider"] = 'google'  # 소셜 로그인 제공자

• 인자에 request: Request 부분 추가
• 로그인 후 세션에 정보를 저장하는 함수 추가 ( 'google' 부분을 각 공급자에 맞게 바꿔주면 됨.  네이버: naver / 카카오: kakao)

---

소셜 로그인 회원 탈퇴 테스트

 

이제 main.py를 실행하여 서버를 실행한 뒤, 소셜 로그인 후 회원 탈퇴를 진행해 본다. 회원 탈퇴가 성공적으로 되었다는 메시지가 잘 뜨고,

 

소셜 로그인 → 회원 탈퇴 요청 → 메모 삭제 → 탈퇴 완료 → 탈퇴 안내 이메일 발송 → 홈페이지로 이동

의 과정이 잘 진행된 것을 확인할 수 있다.

 

DB 상에도 네이버 계정 관련 정보가 삭제된 것을 확인할 수 있다.

 

역시 탈퇴 완료 이메일도 잘 전송되었다.

 

이렇게 하면 끝?이 아니다.

---

소셜 로그인 메모 관련 기능 보완

 

소셜 로그인 탈퇴 관련 로직을 보완하기 전에, 중대한 문제를 찾았다.

그것은 바로 소셜 로그인 시, 메모 작성, 조회, 수정 및 삭제 기능이 불가하다는 것이다.

 

◆ memos_controller.py

따라서, memos_controller.py 파일에서 사용자 조회 함수를 추가해서 소셜 ID와 제공자를 받아서 메모 관련 기능 함수에 넣어주도록 보완해 준다.

- get_current_user(): 사용자 조회 함수

# 사용자 조회
def get_current_user(request: Request, db: Session) -> User | None:
    user_id = request.session.get("id")
    social_id = request.session.get("social_id")
    provider = request.session.get("provider")

    if user_id:
        return db.query(User).filter(User.id == user_id).first()

    if social_id and provider:
        if provider == "google":
            return db.query(User).filter(User.google_id == social_id).first()
        elif provider == "kakao":
            return db.query(User).filter(User.kakao_id == social_id).first()
        elif provider == "naver":
            return db.query(User).filter(User.naver_id == social_id).first()

    return None

 

- get_authenticated_user(): 사용자 인증 확인 함수

# 인증된 사용자
def get_authenticated_user(request: Request, db: Session = Depends(get_db)) -> User:
    user = get_current_user(request, db)
    if user is None:
        raise HTTPException(status_code=401, detail="Not Authorized")
    return user

 

그리고, 각 메모 기능별 함수를 수정해 준다.

필자는 이 문제가 1-2시간이면 해결할 수 있을 것으로 생각했다. 그러나, 거의 하루종일 붙잡고 매달렸고, 결국엔 해결했다.

 

◆ main.py

# FastAPI 애플리케이션 생성
app = FastAPI()
app.add_middleware(
    SessionMiddleware, 
    secret_key="your-secret-key",
    max_age=60 * 60,  # 1시간 후 세션 만료
    same_site="lax", # same_site가 없으면 세션 쿠키가 인증 흐름 중 브라우저에서 차단될 수 있음
    https_only=False, # 로컬 환경에서 HTTPS가 아니라면, 브라우저가 Secure 속성이 있는 쿠키를 저장하지 않을 수 있으므로, https_only=False
    session_cookie="session",
)

 

◆ users_controller.py

users_controller.py의 소셜 로그인 후 사용자 정보를 처리하는 함수도 수정

# 소셜 로그인 후 사용자 정보 처리
def create_or_update_social_user(db: Session, user_info: dict, provider: str, request: Request):
    logger.info(f"소셜 로그인 처리 시작: 제공자 {provider}, 사용자 정보 {user_info}")

    # provider 별로 ID 및 이름 필드 설정
    social_id_filed = f"{provider}_id"
    user_name = (
        user_info.get('user_name') or
        user_info.get('profile_nickname') or
        user_info.get('name')
    )
    social_id_value = user_info.get(social_id_filed)

    if not user_info.get('email') or not social_id_value:
        logger.error("이메일 또는 소셜 ID가 제공되지 않았습니다.")
        raise ValueError('email 또는 소셜 ID가 제공되지 않았습니다.')
    
    # 기존 사용자 조회
    user = db.query(User).filter(
        (User.email == user_info['email']) | (getattr(User, social_id_filed) == social_id_value)).first()

    if not user:
        # 신규 사용자 생성
        user = User(
            username=user_info.get('username'),
            email=user_info['email'],
        )
        setattr(user, social_id_filed, social_id_value)
        db.add(user)
        logger.info(f"신규 사용자 생성: {user_info['email']} (소셜 ID: {social_id_value})")
    else:
        # 기존 사용자 정보 업데이트
        user.username = user_name or user.username  # 이름이 없으면 업데이트 하지 않음
        user.email = user_info['email'] # 이메일 업데이트
        setattr(user, social_id_filed, social_id_value)
        db.merge(user)  # 업데이트 후 세션에 반영
        logger.info(f"기존 사용자 업데이트: {user.email} (소셜 ID: {social_id_value})")
    try:
        db.commit()
        logger.info(f"사용자 정보 저장 성공: {user.email}")
    except Exception as e:
        db.rollback() # 에러 발생 시 롤백
        logger.error(f"사용자 정보 저장 실패: {e}")
        raise HTTPException(status_code=500, detail="사용자 정보를 저장하는 중 오류가 발생했습니다.")

    db.refresh(user)

    # 사용자 정보 저장 후 세션에 사용자 정보 저장
    request.session['id'] = user.id
    request.session['username'] = user.username
    request.session['social_id'] = social_id_value
    request.session['provider'] = provider

    return user

 

◆ oauth/naver.py

import os
import httpx
from fastapi import APIRouter, Request, Depends, HTTPException
from dotenv import load_dotenv
from fastapi.templating import Jinja2Templates
from dependencies import get_db
from sqlalchemy.orm import Session
from controllers.users_controller import create_or_update_social_user
from fastapi.responses import RedirectResponse
import secrets
import logging

load_dotenv()
router = APIRouter()
templates = Jinja2Templates(directory="templates")

# 로깅 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

NAVER_AUTH_URL = "https://nid.naver.com/oauth2.0/authorize"
NAVER_TOKEN_URL = "https://nid.naver.com/oauth2.0/token"
NAVER_USERINFO_URL = "https://openapi.naver.com/v1/nid/me"

CLIENT_ID = os.getenv("NAVER_CLIENT_ID")
CLIENT_SECRET = os.getenv("NAVER_CLIENT_SECRET")
REDIRECT_URI = os.getenv("NAVER_REDIRECT_URI")

@router.get("/auth/naver/login")
async def login(request: Request):
    # CSRF 방지를 위한 state 토큰 생성
    state = secrets.token_urlsafe(16)
    request.session["oauth_state"] = state
    logger.info(f"생성된 state: {state}") # 상태 정보 로그
    logger.info(f"Session state stored: {request.session['oauth_state']}")  # state 값 저장 로그

    auth_url = (
        f"{NAVER_AUTH_URL}?response_type=code"
        f"&client_id={CLIENT_ID}"
        f"&redirect_uri={REDIRECT_URI}"
        f"&state={state}"
    )
    return RedirectResponse(url=auth_url, status_code=302) # 일시 리디렉션

@router.get("/naver/login/callback")
async def callback(request: Request, db: Session = Depends(get_db)):
    code = request.query_params.get("code")
    state = request.query_params.get("state")
    session_state = request.session.get("oauth_state")

    # CSRF 방지: state 검증
    logger.info(f"Session state: {session_state}, Callback state: {state}")
    if state != session_state:
        logger.warning("State Mismatch")
        raise HTTPException(status_code=400, detail="Invalid OAuth State")

    async with httpx.AsyncClient() as client:
        # 액세스 토큰 요청
        token_res = await client.post(NAVER_TOKEN_URL, data={
            'code': code,
            'client_id': CLIENT_ID,
            'client_secret': CLIENT_SECRET,
            'redirect_uri': REDIRECT_URI,
            'grant_type': 'authorization_code',
            'state': state,
        })

        if token_res.status_code != 200:
            logger.error(f"Token 요청 실패: {token_res.status_code}, 내용: {token_res.text}")
            return RedirectResponse(url="/login?error=token")

        token_data = token_res.json()
        access_token = token_data.get("access_token")

        if not access_token:
            logger.error(f"access_token 없음. 응답 내용: {token_data}")
            return RedirectResponse(url="/login?error=no_token")

        # 사용자 정보 요청
        userinfo_res = await client.get(NAVER_USERINFO_URL, headers={
            'Authorization': f'Bearer {access_token}'
        })

        if userinfo_res.status_code != 200:
            logger.error(f"사용자 정보 요청 실패: {userinfo_res.text}")
            return RedirectResponse(url="/login?error=userinfo")

        user_info_raw = userinfo_res.json()
        naver_response = user_info_raw.get("response", {})

        if not naver_response.get("email") or not naver_response.get("id"):
            logger.error(f"유효하지 않은 사용자 정보: {naver_response}")
            return RedirectResponse(url="/login?error=invalid_user")

        # 사용자 정보 정리
        user_info = {
            "username": naver_response.get("name") or "User",
            "email": naver_response.get("email"),
            "naver_id": naver_response.get("id")
        }

        # DB에 유저 저장 또는 업데이트
        user = create_or_update_social_user(db, user_info, provider='naver', request=request)

        # 세션에 로그인 정보 저장
        request.session["id"] = user.id
        request.session["username"] = user.username
        request.session["social_id"] = user_info["naver_id"]
        request.session["provider"] = 'naver'

        logger.info(f"소셜 로그인 성공: {user.username}")

    # 로그인 성공 후 메모 페이지로 이동
    return RedirectResponse(url="/memos")

 

 

◆ oauth/kakao.py

 

  [전체적으로 권장되는 구조]

구간	처리 흐름
1. code + state 확인	CSRF 공격 방지용 state 비교
2. 토큰 요청	code로 access_token 획득
3. 사용자 정보 요청	access_token으로 사용자 정보 가져오기
4. 사용자 정보 파싱	nickname, id, email 등 추출
5. DB 처리	사용자 생성 or 업데이트 함수 호출
6. 세션 저장	로그인 정보 세션에 저장
7. 리다이렉트	/memos 페이지 이동

import os
import httpx
from fastapi import APIRouter, Request, Depends, HTTPException
from dotenv import load_dotenv
from fastapi.templating import Jinja2Templates
from dependencies import get_db
from sqlalchemy.orm import Session
from controllers.users_controller import create_or_update_social_user
from fastapi.responses import RedirectResponse
import secrets
import logging

load_dotenv()
router = APIRouter()
templates = Jinja2Templates(directory="templates")

# 로깅 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

KAKAO_AUTH_URL = "https://kauth.kakao.com/oauth/authorize"
KAKAO_TOKEN_URL = "https://kauth.kakao.com/oauth/token"
KAKAO_USERINFO_URL = "https://kapi.kakao.com/v2/user/me"

CLIENT_ID = os.getenv("KAKAO_REST_API_KEY")
CLIENT_SECRET = os.getenv("KAKAO_CLIENT_SECRET")
REDIRECT_URI = os.getenv("KAKAO_REDIRECT_URI")

@router.get("/auth/kakao/login")
async def login(request: Request):
    # CSRF 방지를 위한 state 토큰 생성
    state = secrets.token_urlsafe(16)
    request.session["oauth_state"] = state
    logger.info(f"생성된 state: {state}") # 상태 정보 로그
    logger.info(f"Session state stored: {request.session['oauth_state']}")  # state 값 저장 로그

    auth_url = (
        f"{KAKAO_AUTH_URL}?response_type=code"
        f"&client_id={CLIENT_ID}"
        f"&redirect_uri={REDIRECT_URI}"
        f"&state={state}"
    )
    return RedirectResponse(url=auth_url)

@router.get("/kakao/login/callback")
async def callback(request: Request, db: Session = Depends(get_db)):  # get_db() 사용
    code = request.query_params.get("code")
    state = request.query_params.get("state")
    session_state = request.session.get("oauth_state")

    # CSRF 방지: state 검증
    logger.info(f"Session state: {session_state}, Callback state: {state}")  # 디버깅 로그
    if state != session_state:
        logger.warning("State Mismatch")
        raise HTTPException(status_code=400, detail="Invalid OAuth State")
    
    async with httpx.AsyncClient() as client:
        token_res = await client.post(KAKAO_TOKEN_URL, data={
            'code': code,
            'client_id': CLIENT_ID,
            'client_secret': CLIENT_SECRET,
            'redirect_uri': REDIRECT_URI,
            'grant_type': 'authorization_code',
            'state': state,
        })
        
        if token_res.status_code != 200:
            logger.error(f"Token 요청 실패: {token_res.status_code}: {token_res.text}")
            return {"error": "Failed to retrieve token"}

        token_data = token_res.json()
        access_token = token_data.get("access_token")

        # 사용자 정보 요청
        userinfo_res = await client.get(KAKAO_USERINFO_URL, headers={
            'Authorization': f'Bearer {access_token}'
        })

        if userinfo_res.status_code != 200:
            logger.error(f"사용자 정보 요청 실패: {userinfo_res.text}")
            return {"error": "Failed to retrieve user information"}

        # 사용자 정보 처리
        user_info_raw = userinfo_res.json()
        kakao_id = str(user_info_raw.get("id"))
        nickname = user_info_raw.get("properties", {}).get("nickname") or "User"
        email = user_info_raw.get("kakao_account", {}).get("email")

        # 필수 정보 확인
        if not kakao_id or not nickname:
            logger.error(f"유효하지 않은 사용자 정보: {user_info_raw}")
            return {"error": "Invalid user info"}

        user_info = {
            "username": nickname,
            "email": email,
            "kakao_id": kakao_id, # 문자열로 형변환
        }

        user = create_or_update_social_user(db, user_info, provider='kakao', request=request)

        # 로그인 후 세션에 정보 저장
        request.session["id"] = user.id  # 일반 사용자 ID
        request.session["username"] = user.username
        request.session["social_id"] = user_info["kakao_id"]  # 소셜 ID
        request.session["provider"] = 'kakao'  # 소셜 로그인 제공자

    # 로그인 성공 후 메모 페이지로 이동
    return RedirectResponse(url="/memos")

 

◆ oauth/google.py

 

[구현 시 체크해봐야 할 것들]

ex) 400: invalid_request 오류 방지

체크 항목	설명
❌ redirect_uri 불일치	구글 개발자 콘솔에 등록된 redirect URI와 실제 요청에서 사용한 redirect_uri가 정확히 일치하지 않음
❌ state 필드 불필요	Google의 token 요청에서는 state를 포함하지 않아야 함
❌ client_id, client_secret 오타	잘못된 값 또는 환경변수 누락 ( .env 또는 환경 변수 확인)
❌ code 값 만료	인증 코드의 유효 시간이 짧아 재사용하면 오류 발생
❌ 잘못된 HTTP method	Google은 POST 방식 + application/x-www-form-urlencoded 필요 (httpx 기본값이라 보통 문제 없음)

 

[오류 메시지 예시]

오류 메시지 내용	의미
redirect_uri_mismatch	구글 콘솔과 URL이 다름
invalid_grant	인증 코드가 잘못됐거나 만료됨
unauthorized_client	클라이언트 ID가 권한 없음
invalid_client	client_id 또는 secret 잘못됨

 

import os
import httpx
from fastapi import APIRouter, Request, Depends, HTTPException
from dotenv import load_dotenv
from fastapi.templating import Jinja2Templates
from dependencies import get_db
from sqlalchemy.orm import Session
from controllers.users_controller import create_or_update_social_user
from fastapi.responses import RedirectResponse
import secrets
import logging

load_dotenv()
router = APIRouter()
templates = Jinja2Templates(directory="templates")

# 로깅 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

GOOGLE_AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth"
GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token"
GOOGLE_USERINFO_URL = "https://www.googleapis.com/oauth2/v1/userinfo"

CLIENT_ID = os.getenv("GOOGLE_CLIENT_ID")
CLIENT_SECRET = os.getenv("GOOGLE_CLIENT_SECRET")
REDIRECT_URI = os.getenv("GOOGLE_REDIRECT_URI")

@router.get("/auth/google/login")
async def login(request: Request):
    # CSRF 방지를 위한 state 토큰 생성
    state = secrets.token_urlsafe(16)
    request.session["oauth_state"] = state
    logger.info(f"생성된 state: {state}") # 상태 정보 로그
    logger.info(f"Session state stored: {request.session['oauth_state']}")  # state 값 저장 로그

    auth_url = (
        f"{GOOGLE_AUTH_URL}?response_type=code"
        f"&client_id={CLIENT_ID}"
        f"&redirect_uri={REDIRECT_URI}"
        f"&scope=openid%20email%20profile"
        f"&access_type=offline"
        f"&prompt=consent"
        f"&state={state}"
    )
    return RedirectResponse(url=auth_url)

@router.get("/auth/google/callback")
async def callback(request: Request, db: Session = Depends(get_db)):  # get_db() 사용
    code = request.query_params.get("code")
    state = request.query_params.get("state")
    session_state = request.session.get("oauth_state")

    # CSRF 방지: state 검증
    logger.info(f"Session state: {session_state}, Callback state: {state}")  # 디버깅 로그
    if state != session_state:
        logger.warning("State Mismatch")
        raise HTTPException(status_code=400, detail="Invalid OAuth State")
    
    async with httpx.AsyncClient() as client:
        token_res = await client.post(GOOGLE_TOKEN_URL, data={
            'code': code,
            'client_id': CLIENT_ID,
            'client_secret': CLIENT_SECRET,
            'redirect_uri': REDIRECT_URI,
            'grant_type': 'authorization_code',
        }) # 'state': state, 구글 토큰에 요청 불가
        
        if token_res.status_code != 200:
            logger.error(f"Token 요청 실패: {token_res.status_code}: {token_res.text}")
            return {"error": "Failed to retrieve token"}

        token_data = token_res.json()
        access_token = token_data.get("access_token")

        # 사용자 정보 요청
        userinfo_res = await client.get(GOOGLE_USERINFO_URL, headers={
            'Authorization': f'Bearer {access_token}'
        })

        if userinfo_res.status_code != 200:
            logger.error(f"사용자 정보 요청 실패: {userinfo_res.text}")
            return {"error": "Failed to retrieve user information"}

        # 사용자 정보 처리
        user_info_raw = userinfo_res.json()
        google_id = str(user_info_raw.get("id"))
        username = user_info_raw.get("name") or "User" # 이름이 없을 경우 기본값 설정
        email = user_info_raw.get("email")

        # 필수 정보 확인
        if not google_id or not username:
            logger.error(f"유효하지 않은 사용자 정보: {user_info_raw}")
            return {"error": "Invalid user info"}
        
        user_info = {
            "username" : username,  
            "email" : email,
            "google_id" : google_id,
        }

        user = create_or_update_social_user(db, user_info, provider='google', request=request)
        # 로그인 후 세션에 정보 저장
        request.session["id"] = user.id  # 일반 사용자 ID
        request.session["username"] = user.username
        request.session["social_id"] = user_info["google_id"]  # 소셜 ID
        request.session["provider"] = 'google'  # 소셜 로그인 제공자

    # 로그인 성공 후 메모 페이지로 이동
    return RedirectResponse(url="/memos")

[전체 플로우 요약]

1. 사용자에게 로그인 링크 (/auth/google/login) 제공
2. 사용자는 구글 로그인 → authorization code 발급
3. 그 코드로 POST /token 요청해서 access_token 받음
4. 그 토큰으로 사용자 정보 요청

[실제 사용자 로그인 링크 예시]

https://accounts.google.com/o/oauth2/v2/auth?
  response_type=code&
  client_id=xxxx.apps.googleusercontent.com&
  redirect_uri=http://localhost:8000/auth/google/callback&
  scope=email%20profile&
  access_type=offline&
  state=abc123

 

파라미터	필수 여부	예시
client_id	✅	구글 콘솔에서 발급한 클라이언트 ID
redirect_uri	✅	정확히 등록된 URI
response_type	✅	code
scope	✅	email profile 또는 더 많음
access_type	권장	offline (리프레시 토큰 받으려면)
prompt	권장	consent (처음 로그인 시 명시적 동의 요청)

 

★ access_type=offline는 Google OAuth에만 적용되는 파라미터

access_type=offline의 의미
offline을 설정하면, 사용자가 로그아웃 상태여도 계속 토큰 갱신이 가능하게 된다.
즉, 리프레시 토큰(refresh_token) 을 받으려면 Google에선 반드시 access_type=offline을 써야 한다.

 

: Naver와 Kakao는 추가해도 무시된다.

SNS 제공자	access_type=offline 지원 여부	설명
Google	✅ 지원함	리프레시 토큰을 받기 위해 필수
Kakao	❌ 지원 안 함	자체 방식으로 리프레시 토큰 발급
Naver	❌ 지원 안 함	OAuth 2.0 기본 방식 사용, offline 없음

 

◆ run.bat

기존의 --reload를 빼고, 127.0.0.1을 localhost 로 수정

uvicorn main:app --host localhost --port 8000

---

각 제공자와의 연결 끊기

 

회원 탈퇴 성공 메시지 반환, 탈퇴 안내 이메일 반환, DB 삭제

여기까지 하면 회원 탈퇴 처리가 완료된 것일까?

답부터 얘기하면 아니다. 

 

네이버의 내정보 - 이력관리를 들어가보면, 회원탈퇴를 했음에도 연결된 서비스 관리에 메모 앱이 여전히 연결되어 있다고 나온다.

 

따라서, 회원 탈퇴 시에 연결된 제공자와의 연결을 끊어주는 작업까지가 완료되어야 진정한 회원 탈퇴 기능을 구현했다고 볼 수 있다. 이를 위해 oauth 디렉터리에 unlink_services.py 파일을 만들어 각 소셜 로그인 서비스 연동을 해제하는 기능을 구현한다.

 

추가로, 여러 명이 사용하는 컴퓨터의 경우 계정을 탈퇴하거나 로그아웃한 경우 자동으로 로그인되지 않도록 설정해 준다.

 

◆ unlink_services.py

import os
import httpx
from fastapi import APIRouter, Request, Depends, HTTPException
from dotenv import load_dotenv
from fastapi.templating import Jinja2Templates
from dependencies import get_db
import logging

load_dotenv()
router = APIRouter()
templates = Jinja2Templates(directory="templates")

# 로깅 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 구글 연결 해제
@router.post("/unlink/google")
async def google_unlink(access_token: str):
    url = f"https://oauth2.googleapis.com/revoke?token={access_token}"
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }

    async with httpx.AsyncClient() as client:
        response = await client.post(url, headers=headers)

    logger.info(f"Google 응답: {response.status_code} - {response.text}")
    
    return {"status": response.status_code, "text": response.text}

# 카카오 연결 해제
@router.post("/unlink/kakao")
async def kakao_unlink(access_token: str):
    url = "https://kapi.kakao.com/v1/user/unlink"
    headers = {
        "Authorization": f"Bearer {access_token}"
    }

    async with httpx.AsyncClient() as client:
        response = await client.post(url, headers=headers)

    logger.info(f"Kakao 응답: {response.status_code} - {response.text}")
    
    return {"status": response.status_code, "text": response.text}


# 네이버 연결 해제
NAVER_CLIENT_ID = os.getenv("NAVER_CLIENT_ID")
NAVER_CLIENT_SECRET = os.getenv("NAVER_CLIENT_SECRET")

@router.post("/unlink/naver")
async def naver_unlink(access_token: str):
    url = "https://nid.naver.com/oauth2.0/token"
    data = {
        "grant_type": "delete",
        "client_id": NAVER_CLIENT_ID,
        "client_secret": NAVER_CLIENT_SECRET,
        "access_token": access_token,
        "service_provider": "NAVER"
    }

    headers = {"Content-Type": "application/x-www-form-urlencoded"}

    async with httpx.AsyncClient() as client:
        response = await client.post(url, headers=headers)

    logger.info(f"Naver 응답: {response.status_code} - {response.text}")
    
    return {"status": response.status_code, "text": response.text}

# 소셜 연동 해제 함수
async def social_unlink_task(provider: str, access_token):
    try:
        if provider == "google":
            result = await google_unlink(access_token=access_token)

        elif provider == "kakao":
            result = await kakao_unlink(access_token=access_token)

        elif provider == "naver":
            result = await naver_unlink(access_token=access_token)

        else:
            logger.warning(f"지원하지 않는 제공자: {provider}, 연결 해제 생략.")
            result = None
    
        if result and result.status_code == 200:
            logger.info(f"{provider} 연결 해제 성공")
        else:
            logger.warning(f"{provider} 연동 해제 실패: {result.status_code} - {result.text}")
    except Exception as e:
        logger.error(f"{provider} 연동 해제 중 에러 발생: {e}")

 

◆ users_controller.py

현재 소셜 로그인 연동 해제 시에 access_token을 인자로 받아 연동 해제를 시도하고 있는데, users_controller.py 파일의 기존 소셜 로그인을 처리하는 함수인 create_or_update_social_user() 에는 access_token을 받지 않으므로, access_token을 받도록 수정해 준다.

그리고, 사용자 정보 저장 후 세션에도 access_token을 저장하도록 수정해 준다.

# 소셜 로그인 후 사용자 정보 처리
def create_or_update_social_user(db: Session, user_info: dict, provider: str, request: Request, access_token: str):
    logger.info(f"소셜 로그인 처리 시작: 제공자 {provider}, 사용자 정보 {user_info}")

    # provider 별로 ID 및 이름 필드 설정
    social_id_field = f"{provider}_id"
    user_name = (
        user_info.get('user_name') or
        user_info.get('username') or
        user_info.get('profile_nickname') or
        user_info.get('name')
    )
    social_id_value = user_info.get(social_id_field)

    if not user_info.get('email') or not social_id_value:
        logger.error("이메일 또는 소셜 ID가 제공되지 않았습니다.")
        raise ValueError('email 또는 소셜 ID가 제공되지 않았습니다.')
    
    # 기존 사용자 조회: 소셜 ID 우선, 없으면 이메일 검색 (선택사항)
    user = db.query(User).filter(getattr(User, social_id_field) == social_id_value).first()
    if not user:
        user = db.query(User).filter(User.email == user_info['email']).first()

    if not user:
        # 신규 사용자 생성
        user = User(
            username=user_info.get('username'),
            email=user_info['email'],
        )
        setattr(user, social_id_field, social_id_value)
        db.add(user)
        logger.info(f"신규 사용자 생성: {user_info['email']} (소셜 ID: {social_id_value})")
    else:
        # 기존 사용자 정보 업데이트
        user.username = user_name or user.username  # 이름이 없으면 업데이트 하지 않음
        user.email = user_info['email'] # 이메일 업데이트
        setattr(user, social_id_field, social_id_value)
        db.merge(user)  # 업데이트 후 세션에 반영
        logger.info(f"기존 사용자 업데이트: {user.email} (소셜 ID: {social_id_value})")
    try:
        db.commit()
        logger.info(f"사용자 정보 저장 성공: {user.email}")
    except Exception as e:
        db.rollback() # 에러 발생 시 롤백
        logger.error(f"사용자 정보 저장 실패: {e}")
        raise HTTPException(status_code=500, detail="사용자 정보를 저장하는 중 오류가 발생했습니다.")

    db.refresh(user)

    # 사용자 정보 저장 후 세션에 사용자 정보 저장
    request.session['id'] = user.id
    request.session['username'] = user.username
    request.session['social_id'] = social_id_value
    request.session['provider'] = provider
    request.session['access_token'] = access_token # 소셜 연동 해제 위함

    return user

 

◆ google, kakao & naver.py

그리고 각 소셜.py 파일에도 세션에 access_token을 저장하도록 수정해 주고, create_or_update_social_user() 함수에 access_token 인자를 추가해 준다.

# DB에 유저 저장 또는 업데이트
user = create_or_update_social_user(db, user_info, provider='naver', request=request, access_token=access_token)

# 세션에 로그인 정보 저장
request.session["id"] = user.id
request.session["username"] = user.username
request.session["social_id"] = user_info["naver_id"]
request.session["provider"] = 'naver'
request.session['access_token'] = access_token

logger.info(f"소셜 로그인 성공: {user.username}")

 

◆ main.py

로그아웃 시, 자동 로그인 (세션 유지)이 되면 보안에 위험할 수 있으므로, 로그아웃 시에 로그인 정보를 확실하게 삭제할 수 있도록 로그아웃 엔드포인트의 코드를 수정한다.

# 로그아웃
@app.post("/logout")
async def logout(request: Request):
    # request.session.pop("username", None)
    request.session.clear()  # 모든 세션 값 삭제
    return {"message": "로그아웃 완료!"}

 

---

 소셜 로그인 연동 해제 테스트

 

다시 main.py를 실행하여 테스트를 해보기 전에, 만약 네이버와 카카오에 로그인이 되어 있다면 모두 로그아웃한다.

네이버와 카카오의 경우, 계정이 로그인 되어 있다면 자동 로그인이 되고, 구글의 경우 항상 로그인 창을 띄울 수 있다.

 

[자동 로그인 흐름]

상황	설명	결과
사용자가 브라우저에 이미 로그인된 상태 (예: Naver, Kakao에 로그인됨)	OAuth 요청 시 해당 서비스가 이미 사용자를 인식	사용자에게 로그인 화면을 생략하고 바로 인증 토큰 발급
사용자가 로그아웃한 상태	OAuth 요청 시 로그인 화면 표시됨	로그인/동의 화면이 뜸
이전에 "계정 선택 안 묻기" 설정한 경우 (Google에 많음)	계정 선택 없이 자동으로 넘어감	그냥 넘어가는 것처럼 보임

 

[자동 로그인 상태에서 소셜 자동 로그인 관련]

질문	답변
자동 로그인 상태면 소셜 로그인 시 물어보지 않나?	✅ 그렇다. 이미 로그인된 세션이 있으면 바로 콜백 URL로 넘어간다.
항상 로그인 창을 띄울 수 있나?	❌ 모든 서비스에서 완벽하게 제어하긴 어렵다. Google은 prompt=login 등 지원하지만, 카카오/네이버는 제한적이다.

 

 

- 네이버

(네이버 로그 표시)

 

- 카카오

 

(카카오 로그 표시)

 

 

- 구글

 

구글은 로그인 되어 있어도 자동 로그인 되는 것이 아니라, 다시 한 번 확인한다.

 

---

다음 내용

[파이썬] FastAPI - 메모 앱 프로젝트 18(최종): 자동 로그아웃