203 lines
7.9 KiB
Python
203 lines
7.9 KiB
Python
import asyncio
|
|
from datetime import UTC, datetime, timedelta
|
|
from uuid import UUID
|
|
|
|
from fastapi import Request
|
|
|
|
from src.database.auth.refresh_tokens import JwtCrudActions
|
|
from src.database.users.crud import UsersCrudActions
|
|
from src.errors.http_errors.errors import Errors
|
|
from src.models.configs_read.env import env_settings
|
|
from src.models.pydantic_models.model import RefreshTokensCreate, UserOut
|
|
|
|
from .jwt import HashService, JwtService
|
|
|
|
|
|
class CurrentUserService:
|
|
|
|
def __init__(self) -> None:
|
|
self.jwt_service=JwtService()
|
|
self.hash=HashService()
|
|
self.crud_db_actions=UsersCrudActions()
|
|
self.jwt_db_actions=JwtCrudActions()
|
|
self.error=Errors()
|
|
|
|
async def _check(self, form_data_email:str, form_data_password:str,):
|
|
'''check user by email'''
|
|
user=await self.crud_db_actions.get_user_by_email(form_data_email)
|
|
|
|
if user is None:
|
|
raise self.error.credentials_error(detail="Wrong credentials")
|
|
|
|
if not await asyncio.to_thread(self.hash.verify_password, plain_password=form_data_password, hashed_password=user.hashed_password):
|
|
raise self.error.credentials_error(detail="Wrong credentials")
|
|
|
|
if user.status is False:
|
|
raise self.error.credentials_error(detail="This user is deactivated")
|
|
return user
|
|
|
|
|
|
async def _token_record_create(self, jti:UUID,user_id:UUID,token:str, request:Request)->RefreshTokensCreate:
|
|
|
|
return RefreshTokensCreate(
|
|
id=jti,
|
|
user_id=user_id,
|
|
token_hash=self.hash.token_to_hash(token),
|
|
device_info=request.headers.get("user-agent", "unknown"),
|
|
ip_address=request.headers.get("x-forwarded-for", "").split(",")[0].strip() or (request.client.host if request.client else "unknown"),
|
|
expires_at=datetime.now(UTC)+timedelta(days=env_settings.REFRESH_TOKEN_EXPIRE_DAYS)
|
|
)
|
|
|
|
|
|
async def get_current_user(self, token:str, *permissions: str)->UserOut:
|
|
|
|
payload= await self.jwt_service.jwt_decode(token)
|
|
sub=payload.get("sub")
|
|
|
|
try:
|
|
sub=UUID(sub)
|
|
except (ValueError, TypeError) as e:
|
|
raise self.error.credentials_error(detail="Jwt token is incorrect") from e
|
|
|
|
if not (payload.get("token_type")=="access"):
|
|
raise self.error.credentials_error(detail="Jwt token type is incorrect")
|
|
|
|
user=await self.crud_db_actions.get_user_by_id(sub)
|
|
if user is None:
|
|
raise self.error.not_found_error(detail="User with this email address not found")
|
|
|
|
if user.status is False:
|
|
raise self.error.credentials_error(detail="This user is deactivated")
|
|
|
|
effective = {p.permission for p in user.direct_permissions} | {p.permission for group in user.group for p in group.permissions}
|
|
missing = set(permissions) - effective
|
|
|
|
if missing:
|
|
raise self.error.forbidden_error(detail=f"Missing: {missing}")
|
|
|
|
return UserOut.model_validate(user)
|
|
|
|
|
|
|
|
async def create_access_token(self, user_id:UUID)->str:
|
|
'''create new access token if all the checks are successful'''
|
|
return await self.jwt_service.create_access_token({"sub":str(user_id)})
|
|
|
|
|
|
|
|
async def create_refresh_token(self,user_id:UUID, request:Request)->str:
|
|
|
|
token, jti= await self.jwt_service.create_refresh_token({"sub":str(user_id)})
|
|
|
|
try:
|
|
jti=UUID(jti)
|
|
except (ValueError, TypeError) as e:
|
|
raise self.error.credentials_error(detail="Jwt token is incorrect") from e
|
|
|
|
'''create new refresh token if all the checks are successful'''
|
|
token_record=await self._token_record_create(jti=jti, user_id=user_id, token=token, request=request)
|
|
|
|
|
|
await self.jwt_db_actions.create_token(RefreshTokensCreate.model_dump(token_record))
|
|
|
|
return token
|
|
|
|
|
|
async def refresh_token(self, refresh_token:str, request:Request)->tuple[str, str]:
|
|
|
|
'''decode old refresh token'''
|
|
old_refresh_token= await self.jwt_service.jwt_decode(refresh_token)
|
|
sub=old_refresh_token.get("sub")
|
|
|
|
if (old_jti:=old_refresh_token.get("jti")) is None:
|
|
raise self.error.credentials_error(detail="Jwt token is incorrect")
|
|
|
|
try:
|
|
old_jti=UUID(old_jti)
|
|
sub=UUID(sub)
|
|
except (ValueError, TypeError) as e:
|
|
raise self.error.credentials_error(detail="Jwt token is incorrect") from e
|
|
|
|
|
|
'''old refresh token check'''
|
|
|
|
if (old_refresh_token.get("token_type")=="access"):
|
|
raise self.error.credentials_error(detail="Jwt token type is incorrect")
|
|
|
|
|
|
old_record=await self.jwt_db_actions.get_token_by_id(old_jti)
|
|
if old_record is None:
|
|
raise self.error.not_found_error(detail="Token not found")
|
|
|
|
|
|
'''sqlite constraints about timezone'''
|
|
expires_at=old_record.expires_at
|
|
if expires_at.tzinfo is None:
|
|
expires_at = expires_at.replace(tzinfo=UTC)
|
|
if expires_at<datetime.now(UTC):
|
|
raise self.error.credentials_error(detail="Token expired")
|
|
|
|
'''user check'''
|
|
user = await self.crud_db_actions.get_user_by_id(sub)
|
|
if user is None:
|
|
raise self.error.not_found_error(detail="User not found")
|
|
if user.status is False:
|
|
raise self.error.credentials_error(detail="This user is deactivated")
|
|
|
|
|
|
'''create new refresh token if all the checks are successful'''
|
|
new_refresh_token, new_jti= await self.jwt_service.create_refresh_token({"sub":str(sub)})
|
|
new_access_token=await self.create_access_token(user_id=sub)
|
|
|
|
try:
|
|
new_jti=UUID(new_jti)
|
|
except (ValueError, TypeError) as e:
|
|
raise self.error.credentials_error(detail="Jwt token is incorrect") from e
|
|
|
|
|
|
'''create database record with the new token'''
|
|
new_token_record=await self._token_record_create(jti=new_jti, user_id=sub, token=new_refresh_token, request=request)
|
|
|
|
success = await self.jwt_db_actions.create_and_update_token(RefreshTokensCreate.model_dump(new_token_record), old_jti, new_jti)
|
|
|
|
if not success:
|
|
raise self.error.not_found_error(detail="Token not found")
|
|
|
|
return (new_access_token, new_refresh_token)
|
|
|
|
|
|
|
|
async def logout(self, refresh_token:str)->bool:
|
|
|
|
'''decode current refresh token'''
|
|
payload=await self.jwt_service.jwt_decode(refresh_token)
|
|
|
|
if (jti:=payload.get("jti")) is None:
|
|
raise self.error.credentials_error(detail="Jwt token is incorrect")
|
|
|
|
try:
|
|
jti=UUID(jti)
|
|
except (ValueError, TypeError, AttributeError) as e:
|
|
raise self.error.credentials_error(detail="Jwt token is incorrect") from e
|
|
|
|
'''logout by assigning revoked flag'''
|
|
if await self.jwt_db_actions.logout(jti):
|
|
return True
|
|
else:
|
|
raise self.error.not_found_error(detail="Refresh Token Not Found")
|
|
|
|
|
|
|
|
async def login(self, form_data_email:str, form_data_password:str, request:Request)->tuple[str, str]:
|
|
'''revoke all the old refresh tokens'''
|
|
user = await self._check(form_data_email, form_data_password)
|
|
await self.jwt_db_actions.revoke_all(user_id=user.id)
|
|
|
|
'''create access and refresh tokens'''
|
|
access_token=await self.create_access_token(user_id=user.id)
|
|
refresh_token=await self.create_refresh_token(user_id=user.id,request=request)
|
|
|
|
return (access_token, refresh_token)
|
|
|
|
async def auth_service()->CurrentUserService:
|
|
return CurrentUserService() |