Mawoka
89 lines
3.5 KiB
Python
89 lines
3.5 KiB
Python
# This Source Code Form is subject to the terms of the Mozilla Public
|
|
# License, v. 2.0. If a copy of the MPL was not distributed with this
|
|
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
|
|
from datetime import timedelta, datetime
|
|
|
|
from fastapi import APIRouter, Request, Response
|
|
from fastapi.security.utils import get_authorization_scheme_param
|
|
from jose import jws, jwt, JWTError, JWSError
|
|
|
|
from classquiz.auth import ACCESS_TOKEN_EXPIRE_MINUTES, create_access_token
|
|
from classquiz.db.models import UserSession
|
|
from classquiz.oauth import google, github
|
|
from classquiz.config import settings
|
|
|
|
settings = settings()
|
|
|
|
router = APIRouter()
|
|
router.include_router(google.router, prefix="/google")
|
|
router.include_router(github.router, prefix="/github")
|
|
|
|
|
|
async def rememberme_middleware(request: Request, call_next):
|
|
rememberme_cookie = request.cookies.get("rememberme_token")
|
|
bearer_token = request.cookies.get("access_token")
|
|
conditions_to_handle_met = True
|
|
# print(bearer_token)
|
|
# if bearer_token is not None:
|
|
# bearer_token = bearer_token.replace("Bearer ", "")
|
|
# test = jws.verify(bearer_token, settings.secret_key, algorithms=["HS256"])
|
|
# try:
|
|
# jwt.decode(bearer_token, settings.secret_key, algorithms=["HS256"])
|
|
# print("jwt ok")
|
|
# except JWTError as e:
|
|
# print("jwt failed")
|
|
# print(test)
|
|
|
|
scheme, param = get_authorization_scheme_param(bearer_token)
|
|
|
|
# if bearer token is none, we can just do the request, since you can't be signed in
|
|
if scheme is None or param is None:
|
|
conditions_to_handle_met = False
|
|
# if rememberme token is none, we can just do the request, since you can't be signed in
|
|
if rememberme_cookie is None:
|
|
conditions_to_handle_met = False
|
|
|
|
if scheme.lower() != "bearer":
|
|
conditions_to_handle_met = False
|
|
|
|
# Verifying the bearer
|
|
try:
|
|
jwt.decode(
|
|
param, settings.secret_key, algorithms=["HS256"]
|
|
) # checking if the token is valid, throws error if not
|
|
conditions_to_handle_met = False
|
|
except JWTError:
|
|
try:
|
|
jws.verify(
|
|
param, settings.secret_key, algorithms=["HS256"]
|
|
) # Verifying only the signature of the jwt, throws error if signature is invalid
|
|
except JWSError:
|
|
conditions_to_handle_met = False
|
|
|
|
if conditions_to_handle_met:
|
|
user_session: UserSession | None = (
|
|
await UserSession.objects.filter(session_key=rememberme_cookie)
|
|
.select_related(UserSession.user)
|
|
.get_or_none()
|
|
)
|
|
if (user_session is None) or (user_session.user is None):
|
|
response: Response = await call_next(request)
|
|
return response
|
|
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
# access_token_expires = timedelta(seconds=1)
|
|
access_token = create_access_token(data={"sub": user_session.user.email}, expires_delta=access_token_expires)
|
|
await user_session.update(last_seen=datetime.now())
|
|
request.state.access_token = f"Bearer {access_token}"
|
|
request.cookies.pop("access_token")
|
|
response: Response = await call_next(request)
|
|
response.set_cookie(
|
|
key="access_token",
|
|
value=f"Bearer {access_token}",
|
|
httponly=True,
|
|
samesite="lax",
|
|
max_age=60 * 60 * 24 * 365,
|
|
)
|
|
else:
|
|
response: Response = await call_next(request)
|
|
return response
|