먼저 OAUTH2.0인증을 위한 변수를 만든다.
# 엑세스 토큰검증을 위한 의존성 주입
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
jwt를 생성하는데는 헤더.페이로드.시그니처가 필요하다
헤더는 자동으로 생성된다
{
"alg": "HS256",
"typ": "JWT"
}
페이로드는 서비스에서 공통적으로 사용하는 클레임으로, 만료시간 exp를 설정했다.
그리고 시그니처는 jwt.encode함수안에 자동으로 생성된 헤더(알고리즘 HS256)와 사용자 정보가 담긴 페이로드, 시크릿키를 합쳐져서 만들어진다 = 헤더+ 페이로드+ 시크릿키
인코딩(Encode) 과정에서는 사용자 정보가 담긴 데이터 페이로드와 클레임을(여기서 토큰 만료시간을 나타내는 exp를 설정), 시크릿키,헤더에 담긴 알고리즘으로 JWT를 생성한다. 이 과정에서 데이터는 JSON 형태로 정의되고, Base64Url로 인코딩되며, 시크릿 키를 사용하여 서명된다.
def create_access_token(data: dict, expires_delta: timedelta = None):
to_encode = data.copy()
# 만료시간 설정
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
#사용자의 로그인 완료하면 서버는 사용자의 정보(예: 사용자 ID)와
# 일정 기간 동안 유효한 만료 시간을 포함하는 JWT를 생성 -> 인코딩과정
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
로그인 인증이 완료되면 jwt 엑세스 토큰이 생성되고(유효기간은 1분이다.) 의존성에 엑세스 토큰을 주입한다 디코딩(Decode) 과정에서는 전달받은 JWT의 엑세스토큰 유효성을 검증하고, 클레임 정보(sub는 토큰 제목이다)를 해석한다. 이 과정은 주로 토큰의 인증과 권한 부여에 사용된다.
# 타입 힌트 Annotated를 사용
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
credentials_exception = HTTPException(
# 사용자 인증 실패할때 에러메시지
status_code=status.HTTP_401_UNAUTHORIZED,
detail="토큰 유효시간이 지났습니다 다시로그인 해주십시오",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
# sub는 클레임,토큰의 주체, 즉 사용자의 식별자를 나타냄
username: str = payload.get("sub")
if username is None:
raise credentials_exception
token_data = TokenData(username=username)
except JWTError:
raise credentials_exception
user = get_user(fake_users_db, username=token_data.username)
if user is None:
raise credentials_exception
return user
로그인 인증후 get_current_active_user함수에 get_current_user함수 의존성을 주입, 현재 활성 상태가 disabled인지 확인하고 인증된 사용자모델(User)을 반환, 사용자가 비활성 상태인 경우, HTTP 예외를 발생시켜 요청을 거부
async def get_current_active_user(
current_user: Annotated[User, Depends(get_current_user),]
):
if current_user.disabled:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
User모델 필드
class User(Document):
username: str
hashed_password: str
임시 가짜 데이터베이스안에는 인증이 완료된 username과 hased_password 키값이 존재한다. 그리고 disabled가 활성화 되어있다.
fake_users_db = {
"hongsun2": {
"username": "hongsun2",
"full_name": "Hong Sun",
"email": "Hongsun@example.com",
"hashed_password": "$...",
"disabled": False,
}
}
그러므로 oauth2인증에 성공하면 임시 가짜 데이터베이스가 출력되서 보여준다 UserInfo 응답 모델은 fake_users_db 이다. 엑세스 토큰에 유효기간은 1분이라 1분만 보여준다.
@app.get("/users/me/access_token", response_model=UserInfo)
async def read_users_me(
current_user: Annotated[UserInfo, Depends(get_current_active_user)]
):
return current_user
유효기간1분이 다지나고 나면 리프레시 토큰이 필요하다 먼저 사용자 정보가 담긴 페이로드 딕셔너리 데이터와 유효기간exp클레임, 시크릿키, 헤더 알고리즘을 조합해 인코딩해준다
def create_refresh_token(data: dict, expires_delta: timedelta | None = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.now(timezone.utc) + expires_delta
else:
expire = datetime.now(timezone.utc) + timedelta(days=7)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
리프레시 디코딩 과정을 통해 토큰을 리프레시 토큰으로 갱신해준다.
# 리프레시 토큰을 검증하고 사용자 정보를 가져오는 함수
async def get_user_info_by_refresh_token(refresh_token: str) -> UserInfo:
credentials_exception = HTTPException(
status_code=401,
detail="토큰이틀렸거나 토큰유효시간이 지났습니다 토큰을 다시 발급하십시오",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
user = get_user(fake_users_db, username)
if user is None:
raise credentials_exception
return user
except JWTError:
raise credentials_exception
테스트 화면
1. OAUTH2 로그인 후 인증화면

2. 인증완료후 액세스토큰, 리프레시 토큰 생성

3. 엑세스토큰 1분 유효기간 지나서 메시지 출력

4. 리프레시 토큰을 생성해 일주일 유효기간 설정 메시지가 잘 출력된다.

'portfolio' 카테고리의 다른 글
Django 템플릿페이지에서 form태그 제출로 이메일보내기(게시글공유) (0) | 2024.05.08 |
---|---|
Django 댓글 비활성화 (0) | 2024.05.08 |
FASTAPI 정적 토큰헤더를 통한 인증 (0) | 2024.03.15 |
FASTAPI와 mongoDB ODM을 활용한 CRUD , SEARCH 엔드포인트 구축 (0) | 2024.03.07 |
리액트 노드 socket.io를 활용한 실시간 영화예매시스템 구축 (0) | 2024.03.07 |