API 서버에서 카카오 로그인 검증 및 테스트 하는 방법 (2/5)

API 서버에서 애플 로그인 검증 및 테스트하는 방법 (1/5)
API 서버에서 카카오 로그인 검증 및 테스트하는 방법 (2/5)
API 서버에서 네이버 로그인 검증 및 테스트하는 방법 (3/5)
API 서버에서 구글 로그인 검증 및 테스트하는 방법 (4/5)
API 서버에서 메타(페이스북) 로그인 검증 및 테스트하는 방법 (5/5)

이전 글에서 다뤘던 소셜 로그인 구조입니다. 저희 서버는 클라이언트에게 access_token과 refresh_token을 전달하여 로그인 과정을 처리하고 있습니다. 소셜 로그인을 경우 클라이언트가 카카오 인증서버에게 토큰을 발급받아 저희 서버에 전달해주면 벡엔드 안에서 해당 토큰을 검증하여 클라이언트에게 access_token과 refresh_token을 전달해주는 방식입니다.

카카오의 경우 OAuth2.0 기반 표준 인증 프로토콜인 OpenID Connect(OIDC)를 지원합니다. 저희 서버는 소셜로그인의 경우 OAuth2.0을 활용한 인증방식을 사용합니다. 따라서 카카오 디밸로퍼스에서 OpenID Connect 사용을 활성화 해주세요

클라이언트단에서 카카오 로그인을 수행할 경우 OpenID Connect를 활성화 했다면 ID token을 추가로 받을 수 있습니다. 클라이언트가 소셜 로그인 API를 호출하여 provider로 kakao, 발급받은 ID token을 body에 담아서 보내줍니다.

POST /api/v1/addnox/auth/social-signin/kakao
Content-Type: application/json
{
  "id_token": "your-id-token"
}

서버에서 카카오 로그인 요청을 받았으니 카카오 토큰을 검증해보겠습니다.

async def valid_kakao_token(db: Session, user_id_token: str):
    try:
        # 1번 공개키 검색
        jwks = get_cached_jwks()
        kid = jwt.get_unverified_header(user_id_token)["kid"]

        public_key = get_public_key(jwks, kid)
        if public_key is None:
            logger.error(f"No matching key found for kid: {kid}")
            raise ValueError("No matching key found")
        # 2번 공개키로 id_token 검증
        decode_id_token = jwt.decode(user_id_token, public_key, algorithms=["RS256"], audience=settings.KAKAO_APP_KEY)
        if decode_id_token["iss"] not in ["https://kauth.kakao.com"]:
            raise ValueError("Wrong issuer.")
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="invalid JWT Token",
        )
    except ValueError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid Social Token",
        )
    # 3번 sub 값으로 유저 매칭후 회원가입 또는 로그인으로 분기
    social_id = decode_id_token["sub"]
    email = decode_id_token["email"]
    social_user = await verify_social_user(db, social_id, email, social_type="kakao")

    return social_user.email, social_user.id

1번 공개키 검색

카카오의 경우 공개키를 https://kauth.kakao.com/.well-known/jwks.json 해당 URL에서 제공하고 있습니다. 다만  공개키는 일정 주기 또는 특별한 이슈 발생 시 변경될 수 있습니다. 카카오 문서에서는 주기적으로 최신 공개키 목록을 조회한 후, 일정 기간 캐싱(Caching)하여 사용할 것을 권장하고 있습니다. 지나치게 빈번한 공개키 목록 조회 요청 시, 요청이 차단될 수 도 있다고 합니다!

보통 캐싱이라하면 Redis를 많이 떠올리시는데요 저희는 서버가 모두 AWS 환경에 올라가 있고 캐싱을 활용하는 다른 서비스가 없어 당장은 도입이 어려운 상황(머니 이슈 ㅜㅜ)이라 json 형태로 따로 저장하여 24시간마다 업데이트 하는 방향으로 적용했습니다.

from datetime import datetime, timedelta
from loguru import logger
from requests import get
from fastapi import HTTPException, status
from jwt.algorithms import RSAAlgorithm


_jwks_cache = {"keys": None, "last_updated": None, "expire_after": timedelta(hours=24)}  # 캐시 유효 기간 (예: 24시간)


def get_cached_jwks():
    """
    캐싱된 JWKS를 반환하거나, 필요시 새로 가져옵니다.
    """
    now = datetime.now()

    # 캐시가 없거나 만료되었는지 확인
    if (
        _jwks_cache["keys"] is None
        or _jwks_cache["last_updated"] is None
        or now - _jwks_cache["last_updated"] > _jwks_cache["expire_after"]
    ):
        logger.info("Fetching new JWKS from Kakao")
        key_payload = get("https://kauth.kakao.com/.well-known/jwks.json")

        if key_payload.status_code != 200:
            logger.error(f"Failed to fetch JWKS: {key_payload.status_code}")
            if _jwks_cache["keys"] is None:
                # 이전에 캐싱된 키가 없는 경우 예외 발생
                raise HTTPException(
                    status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                    detail="Unable to fetch key information from Kakao",
                )
            # 이전 캐시를 계속 사용
            logger.warning("Using expired JWKS cache due to fetch failure")
            return _jwks_cache["keys"]

        # 캐시 업데이트
        _jwks_cache["keys"] = key_payload.json()["keys"]
        _jwks_cache["last_updated"] = now
        logger.info("JWKS cache updated successfully")
    return _jwks_cache["keys"]


def get_public_key(jwks, kid):
    """
    주어진 kid에 맞는 공개 키를 JWKS에서 찾아 반환합니다.
    """
    for key in jwks:
        if key["kid"] == kid:
            return RSAAlgorithm.from_jwk(key)

    logger.error(f"No matching key found for kid: {kid}")
    return None

2번 공개키로 id_token 검증
jwt 라이브러리를 이용해 공개키를 가지고 id_token을 검증합니다. 여기서 audience 값은 카카오 디벨로퍼스 내에서 내 어플리케이션의 REST APP 키 값입니다!

토큰을 decode하여 iss 값이 카카오 서버에서 발행한 것을 확인하면 올바른 토큰으로 확인합니다.

3번 sub값으로 유저 매칭 후 로그인 또는 회원가입으로 분기
저희 서버의 경우 소셜 유저와 이메일로 가입한 유저 테이블이 분리되어 있습니다. 
사용자는 언제든지 카카오 계정의 이메일을 변경할 수 있는 구조입니다. (대부분의 소셜 로그인이 그렇습니다)
사용자가 소셜 계정의 이메일을 변경하면 토큰을 decode하였을 때 이메일 값도 달라집니다. 따라서 변하지않는 sub 값(유저 고유 ID)을 가지고 회원가입 또는 로그인 처리를 진행합니다.

async def verify_social_user(db: Session, social_id: int, email: str, social_type: str):
    """
    소셜 사용자의 존재 여부 및 타입을 확인하고, 필요한 경우 이메일을 업데이트합니다.
    """
    if await crud.check_duplicate_email(email=email, db=db):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="User is signed up with email type")

    social_user = await crud.get_social_user_by_social_id(db, social_id)

    if not social_user:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="User is not valid, please sign up",
        )

    if social_user.social_type != social_type:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"User is signed up with {social_user.social_type} type",
        )

    if social_user.email != email:
        await crud.update_email_social_user(db, email, social_user.id)
        db.commit()

    return social_user

순서대로 토큰의 이메일을 가지고 이미 이메일로 회원가입한 유저인지 확인하고 sub 값으로 소셜 회원가입한 유저인지 확인합니다. 추가적으로 다른 소셜로 회원가입하였는지 확인 합니다.
회원가입된 유저가 없다면 회원가입 로직으로 분기할 수 있게 status code 403 으로 리턴합니다.
회원가입한 유저가 있다면 social_id를 리턴하여 access_token과 refresh_token을 발급하여 로그인 처리를 진행합니다.
마지막으로 sub값으로 DB에서 소셜 유저를 찾았는데 토큰의 이메일과 DB에 저장된 이메일이 다른 경우 (사용자가 중간에 소셜 계정의 이메일을 변경한 경우입니다) DB의 이메일을 토큰의 이메일로 덮어씌워 업데이트를 해줍니다.

테스트: 카카오 로그인 응답(Mock 데이터) 생성

이제 테스트 코드를 통해 API가 제대로 작동하는지 검증해보겠습니다. 테스트 프레임워크로 pytest를 사용중입니다. 테스트를 위해 실제 카카오 인증 서버에 요청을 보내는 대신, 카카오 로그인 응답을 Mock 데이터로 생성합니다.

root_path = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
with open(os.path.join(root_path, "private_key.pem"), "rb") as key_file:
    private_key = load_pem_private_key(key_file.read(), password=None)
    
    
@pytest.fixture(scope="function")
def mock_kakao_response():
    headers = {"kid": "3f96980381e451efad0d2ddd30e3d3", "alg": "RS256"}
    payload = {
        "iss": "https://kauth.kakao.com",
        "aud": settings.KAKAO_APP_KEY,
        "sub": "3719291111",
        "iat": datetime.now(),
        "exp": datetime.now() + timedelta(days=180),
        "email": "test@kakao.com",
        "auth_time": datetime.now().isoformat(),
    }
    kakao_jwt = jwt.encode(headers=headers, payload=payload, key=private_key, algorithm="RS256")
    return kakao_jwt

mock_kakao_response() 함수

• 카카오에서 제공하는 JWT응답을 모방하여 테스트 할 수 있도록 함.
• 사용자의 sub(고유 ID)와 email 등의 필드가 포함됨.
• 개인키와 공개키를 만들어서 공개키는 jwk 인 json 파일로 저장

mock_kakao_validation = "app.utils.validate_social.get"

async def test_social_login_kakao_success(
        self,
        test_client: httpx.AsyncClient,
        create_social_user: List[SocialUser],
        mocker: MockerFixture,
        mock_kakao_response,
    ):
        mock_response = Response()
        mock_response.status_code = status.HTTP_200_OK
        mock_response.json = lambda: kakao_key
        mocker.patch(mock_kakao_validation, return_value=mock_response)
        response = await test_client.post(
            "/api/v1/test/auth/social-signin/kakao",
            json={
                "id_token": mock_kakao_response,
                "access_token": None,
            },
        )
        assert response.status_code == status.HTTP_200_OK

test_social_login_kakao_success() 함수

• 카카오의 공개키를 가져오는 get을 모킹하여 저희가 생성한 공개키 json 파일로 리턴하도록 구현
• 카카오의 개인키를 저희가 알 수 없기 때문에 이런 방식으로 저희의 개인키와 공개키를 이용하여 검증하도록 구현하였습니다.

질문, 피드백은 언제든지 환영입니다.