22 lines
697 B
Python
22 lines
697 B
Python
|
import jwt
|
||
|
|
from jwt.exceptions import PyJWTError
|
||
|
|
from config import Config
|
||
|
|
from core.models.message.db import MPProfile
|
||
|
|
from fastapi import Header
|
||
|
|
from core.storage import auth_storage
|
||
|
|
from core.errors.errors import not_authenticated_error
|
||
|
|
|
||
|
|
|
||
|
|
async def get_current_user(token: str = Header(alias="Authorization")) -> MPProfile:
|
||
|
|
"""Get the current user."""
|
||
|
|
try:
|
||
|
|
token = token.split("Bearer ")[1]
|
||
|
|
token_data = jwt.decode(token, Config.secret, algorithms=["HS256"])
|
||
|
|
user = await auth_storage.get_user_by_id(token_data["user_id"])
|
||
|
|
if user:
|
||
|
|
return user
|
||
|
|
raise not_authenticated_error()
|
||
|
|
|
||
|
|
except PyJWTError:
|
||
|
|
not_authenticated_error()
|