132 lines
3.4 KiB
Python
132 lines
3.4 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Header, Body, status, Request
|
|
from sqlalchemy.orm import Session
|
|
import json
|
|
import datetime
|
|
import requests
|
|
import jwt
|
|
import time
|
|
from process.logger import logger
|
|
|
|
from db.base import get_db
|
|
import pandas as pd
|
|
|
|
from common.config import SECRET_KEY
|
|
|
|
KST = datetime.timezone(datetime.timedelta(hours=9))
|
|
|
|
router = APIRouter(
|
|
prefix="/test",
|
|
tags=["test"],
|
|
responses={404: {"description": "Not found"}},
|
|
)
|
|
|
|
# -----------------------------------------------------------------------------------------------
|
|
|
|
async def get_body(request: Request):
|
|
return await request.body()
|
|
|
|
|
|
|
|
|
|
# user_seq로 토큰 생성기
|
|
@router.post("/make/token/{user_seq}")
|
|
async def make_new_token(user_seq: str, body: bytes = Depends(get_body), db: Session = Depends(get_db)):
|
|
try:
|
|
user_token = create_jwt(user_seq=int(user_seq), period=3600)
|
|
return {
|
|
"result": "OK",
|
|
"data": {
|
|
"token": user_token
|
|
}
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"result": "FAIL",
|
|
"msg": e
|
|
}
|
|
|
|
|
|
# user_seq로 만료된 토큰 생성기
|
|
@router.post("/make/token/expired/{user_seq}")
|
|
async def make_new_token_expired(user_seq: str, body: bytes = Depends(get_body), db: Session = Depends(get_db)):
|
|
try:
|
|
user_token = create_jwt(user_seq=int(user_seq), period=0)
|
|
return {
|
|
"result": "OK",
|
|
"data": {
|
|
"token": user_token
|
|
}
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"result": "FAIL",
|
|
"msg": e
|
|
}
|
|
|
|
|
|
# 토큰 정보 확인
|
|
@router.post("/check/token")
|
|
async def check_token_data(body: bytes = Depends(get_body), db: Session = Depends(get_db)):
|
|
try:
|
|
# body에서 ID 추출
|
|
try:
|
|
body = json.loads(body)
|
|
except json.JSONDecodeError as e:
|
|
return logger.error(f"json.loads error: Invalid JSON format in request body\nerror message: {e}")
|
|
|
|
token = body['token']
|
|
token_decode = decode_jwt(token=token)
|
|
|
|
return {
|
|
"result": "OK",
|
|
"data": {
|
|
"user_seq": token_decode['user_seq'],
|
|
"exp": change_time(token_decode['exp']),
|
|
"iat": change_time(token_decode['iat']),
|
|
}
|
|
}
|
|
|
|
except Exception as e:
|
|
return {
|
|
"result": "FAIL",
|
|
"msg": e
|
|
}
|
|
|
|
|
|
|
|
|
|
def create_jwt(user_seq: int, period: int):
|
|
# 현재 시간
|
|
now = int(time.time())
|
|
|
|
# JWT의 페이로드 (클레임)
|
|
payload = {
|
|
"user_seq": user_seq,
|
|
"exp": now + period, # 만료 시간
|
|
"iat": now, # 토큰 발행 시간
|
|
}
|
|
|
|
# JWT 생성 (HS256 알고리즘 사용)
|
|
token = jwt.encode(payload, SECRET_KEY, algorithm="HS256")
|
|
|
|
return token
|
|
|
|
|
|
def decode_jwt(token: str):
|
|
try:
|
|
# 토큰 디코드
|
|
decoded_payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
|
|
return decoded_payload
|
|
except jwt.ExpiredSignatureError:
|
|
return {"error": "Token has expired"}
|
|
except jwt.InvalidTokenError:
|
|
return {"error": "Invalid token"}
|
|
|
|
|
|
# 현재 시간 YYYY-MM-DD hh:mm:ss로 변환 함수
|
|
def change_time(time_data):
|
|
dt_object = datetime.datetime.fromtimestamp(time_data)
|
|
formatted_time = dt_object.strftime('%Y-%m-%d %H:%M:%S')
|
|
return formatted_time |