81 lines
2.6 KiB
Python
81 lines
2.6 KiB
Python
import hashlib
|
|
from datetime import UTC, datetime, timedelta
|
|
from uuid import uuid4
|
|
|
|
import bcrypt
|
|
from jose import JWTError, jwt
|
|
|
|
from src.errors.http_errors.errors import Errors
|
|
from src.models.configs_read.env import env_settings
|
|
|
|
'''Hash/Check hash'''
|
|
class HashService:
|
|
|
|
def __init__(self) -> None:
|
|
pass
|
|
|
|
def plain_to_hash(self, plain_password:str)->str:
|
|
return bcrypt.hashpw(plain_password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
|
|
|
|
def verify_password(self, plain_password:str, hashed_password:str)->bool:
|
|
return bcrypt.checkpw(plain_password.encode("utf-8"), hashed_password.encode("utf-8"))
|
|
|
|
def token_to_hash(self, token:str)->str:
|
|
return hashlib.sha256(token.encode("utf-8")).hexdigest()
|
|
|
|
|
|
'''jwt'''
|
|
class JwtService:
|
|
|
|
def __init__(self) -> None:
|
|
|
|
self.error=Errors()
|
|
|
|
async def _validate_sub(self,data:dict)->None:
|
|
if not (data.get("sub")) or data.get("sub") == "":
|
|
raise self.error.credentials_error(detail="Jwt token is incorrect")
|
|
|
|
|
|
async def create_access_token(self, data:dict)->str:
|
|
|
|
user_info=data.copy()
|
|
jti=str(uuid4())
|
|
|
|
await self._validate_sub(user_info)
|
|
|
|
user_info.update({"exp": datetime.now(UTC)+timedelta(minutes=env_settings.ACCESS_TOKEN_EXPIRE_MINUTES),
|
|
"token_type":"access",
|
|
"jti":jti})
|
|
return jwt.encode(user_info, env_settings.SECRET_KEY, env_settings.ALGORITHM)
|
|
|
|
|
|
async def create_refresh_token(self, data:dict)->tuple[str, str]:
|
|
|
|
user_info=data.copy()
|
|
jti=str(uuid4())
|
|
|
|
await self._validate_sub(user_info)
|
|
|
|
user_info.update({"exp":datetime.now(UTC)+timedelta(days=env_settings.REFRESH_TOKEN_EXPIRE_DAYS),
|
|
"token_type":"refresh",
|
|
"jti":jti
|
|
})
|
|
|
|
return jwt.encode(user_info, env_settings.SECRET_KEY, env_settings.ALGORITHM), jti
|
|
|
|
|
|
|
|
async def jwt_decode(self, token:str)->dict:
|
|
|
|
try:
|
|
payload=jwt.decode(token, env_settings.SECRET_KEY, algorithms=[env_settings.ALGORITHM], options={"require_exp": True} )
|
|
|
|
if not (payload.get("sub")) or not (payload.get("token_type")):
|
|
raise self.error.credentials_error(detail="Jwt token is incorrect")
|
|
|
|
except JWTError as e:
|
|
raise self.error.credentials_error(detail="JWTerror") from e
|
|
|
|
return payload
|
|
|