53 lines
1.7 KiB
Python
53 lines
1.7 KiB
Python
import datetime
|
|
from typing import Union
|
|
|
|
from fastapi import Depends
|
|
from fastapi.security import APIKeyCookie, APIKeyQuery
|
|
|
|
from src.error.exceptions import JWTTokenVerifyException
|
|
from src.logging.get_logger import get_logger
|
|
from src.model.internal.jwt_token import JWTToken
|
|
from src.model.internal.session import UserSession
|
|
from src.services.session_service import get_session
|
|
from src.system_var import environment
|
|
|
|
logger = get_logger('認証チェック')
|
|
cookie_security = APIKeyCookie(name='session', auto_error=False)
|
|
code_security = APIKeyQuery(name='code', auto_error=False)
|
|
|
|
|
|
def get_current_session(session_key=Depends(cookie_security)):
|
|
if session_key is None:
|
|
return None
|
|
|
|
session = get_session(session_key)
|
|
|
|
# sessionが存在しない場合はNoneが返る
|
|
return session
|
|
|
|
|
|
def check_session_expired(session: Union[UserSession, None] = Depends(get_current_session)):
|
|
"""セッションの最後にアクセスした時間が、セッション有効期限切れであるかどうかをチェックする"""
|
|
if session is None:
|
|
return None
|
|
|
|
last_access_time = session.last_access_time
|
|
session_expired_period = datetime.datetime.fromtimestamp(
|
|
last_access_time) + datetime.timedelta(minutes=environment.SESSION_EXPIRE_MINUTE)
|
|
if session_expired_period < datetime.datetime.now():
|
|
return None
|
|
|
|
return session
|
|
|
|
|
|
def verify_session(session: Union[UserSession, None] = Depends(check_session_expired)):
|
|
if session is None:
|
|
return None
|
|
jwt_token = JWTToken(session.id_token, session.refresh_token)
|
|
try:
|
|
jwt_token.verify_token()
|
|
except JWTTokenVerifyException as e:
|
|
logger.info(e)
|
|
return None
|
|
return session
|