25 lines
791 B
Python
25 lines
791 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 ")
|
|
if len(token) != 2:
|
|
raise not_authenticated_error()
|
|
token = token[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()
|