[가계부 만들기] Backend - 회원 관리 기능 #5

회원 관리 기능에서 기초가 되는 기능에서 마지막은 refresh token을 통해서 access token과 refresh token을 갱신하는 기능일 것이다.

세션을 사용하지 않고 JWT만으로 사용자 정보를 검증하기 때문에 access token의 유효시간이 만료되었을 때 이를 새로 갱신해줄 필요가 있다.

해당 포스팅에서 해당 기능을 간단하게 구현하도록 한다.

Schema 생성

refresh token을 갱신할 때 Response로 받을 schema를 추가적으로 생성할 것이다. 물론 그냥 str로 받아도 되지만 이렇게 받는 것보다 schema를 통해서 정의하는 것이 좀 더 좋아보인다.

databases/schemas.py 파일에 아래 코드를 추가한다.

# Refresh token을 위한 request 정의
class Refresh(BaseModel):
    refresh_token: str = Field(title="Refresh Token")

이와 같이 refresh token을 먼저 정의하였다.

728x90

Router 추가

refresh token을 넘겨받아서 처리할 router를 추가로 생성하도록 하겠다.

routers/members.py 파일에 마찬가지로 아래 코드를 추가한다.

#refresh token을 이용한 access token 재발급
@router.post("/refresh", description="token 갱신", response_model=schemas.LoginResponse, responses={
    HTTP_400_BAD_REQUEST: {
        "model": schemas.Message
    }
})
async def refresh(
    refresh_token: schemas.Refresh,
    db: Session = Depends(get_db)
):
    try:
        result = await members_service.member_refresh(db, refresh_token)

        return result
    except Exception as e:
        return JSONResponse(content={"detail": e.args[0]}, status_code=HTTP_400_BAD_REQUEST)

response는 기본적으로 login과 동일하다.

request에서는 바로 위에서 정의한 Refresh를 받도록 하였다. refresh token을 받아서 처리할 함수를 호출하였다.

함수를 아직 정의하지 않았지만 데이터베이스에서 조회하기 때문에 DB 접속 정보를 넘기고 처리할 refresh token을 함께 전달한다.

refresh token을 사용한다는 것을 제외하고는 로그인과 동일하다.

JWT 검증 추가

refresh token을 처리하는 코드를 추가하기 전에 refresh token이 JWT로 유효한지 검사하는 코드를 먼저 추가한다.

libraries/auth.py 파일이 아래 코드를 추가한다.

# JWT 검증 함수(refresh token용)
def decode_refresh_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )

단순하게 JWT를 검증하는 역할만 하지만 access token을 처리하는 방식이 달라서 별도의 함수로 빼서 처리하도록 한다.

반응형

처리 로직 추가

이제 refresh token을 검증하고 access token을 새로 생성하는 코드를 추가할 것이다.

services/members_service.py 파일에 아래 내용을 추가한다.

# refresh token 처리
async def member_refresh(db: Session, refresh_token: schemas.Refresh):
    # refresh token 검사
    try:
        payload = auth.decode_refresh_token(refresh_token.refresh_token)
    except Exception:
        raise Exception

    # 해당 회원이 있는지 검사
    stmt = select(models.Members).filter(models.Members.member_no == payload['member_no'], models.Members.is_deleted == 'F')
    result = db.execute(stmt)

    db_member = result.fetchone()
    if db_member is None:
        raise Exception("해당하는 아이디를 찾을 수 없습니다")

    data = {
        "member_no": db_member.Members.member_no,
        "member_id": db_member.Members.member_id,
        "member_name": db_member.Members.member_name,
        "member_email": db_member.Members.member_email
    }
    data["access_token"] = auth.create_access_token(data, timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
    data["refresh_token"] = auth.create_access_token({
        "member_no": data["member_no"],
        "member_id": data["member_id"]
    }, timedelta(hours=REFRESH_TOKEN_EXPIRE_HOURS))

    return data

우선 refresh token이 유효한지 검사한다. 만약 유효하지 않다면 전달받은 Exception을 그대로 다시 반환한다.

그리고 JWT에 있는 회원 번호를 통해서 회원번호로 회원 정보를 찾아서 access token과 refresh token을 다시 생성해서 반환한다.

그럼 이렇게 반환받은 정보를 Frontend에 전달해서 새로운 access token으로 통신하게 된다.

Refactoring

코드를 작성하고 보니 response 정보가 동일하다.

그래서 response를 만드는 코드가 중복으로 사용된다. 만약 response의 정보가 수정된다면 중복되는 코드를 수정하여야 한다.

그런 경우를 대비하기 위해서 response data를 만드는 부분을 별도의 함수를 만들어서 중복되는 코드를 없애도록 하자.

def make_login_response(db_member):
    data = {
        "member_no": db_member.Members.member_no,
        "member_id": db_member.Members.member_id,
        "member_name": db_member.Members.member_name,
        "member_email": db_member.Members.member_email
    }
    data["access_token"] = auth.create_access_token(data, timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
    data["refresh_token"] = auth.create_access_token({
        "member_no": data["member_no"],
        "member_id": data["member_id"]
    }, timedelta(hours=REFRESH_TOKEN_EXPIRE_HOURS))

    return data

먼저 중복되는 코드를 별도의 함수로 먼저 선언하였다.

그럼 이제 login_proc 함수와 member_refresh 함수에서 중복으로 들어있던 코드를 지우고 새로 선언한 함수를 부르는 코드로 수정한다.

# login 처리
async def login_proc(db: Session, member_id: str, member_pw: str):
    # 해당 아이디가 있는지 찾는다/
    stmt = select(models.Members).filter(models.Members.member_id == member_id, models.Members.is_deleted == 'F')
    result = db.execute(stmt)

    db_member = result.fetchone()
    if db_member is None:
        raise Exception("해당하는 아이디를 찾을 수 없습니다")

    if auth.verify_password(member_pw, db_member.Members.member_pw) == False:
        raise Exception("패스워드가 일치하지 않습니다.")

    return make_login_response(db_member)


# refresh token 처리
async def member_refresh(db: Session, refresh_token: schemas.Refresh):
    # refresh token 검사
    try:
        payload = auth.decode_refresh_token(refresh_token.refresh_token)
    except Exception:
        raise Exception

    # 해당 회원이 있는지 검사
    stmt = select(models.Members).filter(models.Members.member_no == payload['member_no'], models.Members.is_deleted == 'F')
    result = db.execute(stmt)

    db_member = result.fetchone()
    if db_member is None:
        raise Exception("해당하는 아이디를 찾을 수 없습니다")

    return make_login_response(db_member)

이제 코드가 다소 간단하게 정리되었다.

테스트

그럼 이제 docs 페이지를 열어서 해당 기능을 테스트해도록 한다.

먼저 login을 진행하여서 refresh token을 생성한다.

login 결과

이제 login에서 전달받은 refresh token으로 access token을 갱신하는 테스트를 하도록 한다.

refresh token 테스트

docs에서 refresh token을 parameter로 전달하여서 작성한 페이지를 호출한다.

아래와 같이 결과가 나오는 것을 확인할 수 있다.

refresh token 테스트 결과

비교적 간단한 코드이지만 로그인을 유지하기 위해서 꼭 필요한 기능이라고 할 수 있다.

다음부터는 가계부에 관련된 데이터를 저장하고 불러오는 기능을 개발하도록 하겠다.