Access tokens 0.1.0

src/service/auth/auth.py

@@ -0,0 +1,51 @@
+from uuid import UUID
+from .jwt import Jwt, Hashes
+from src.database.users.crud import UsersCrudActions
+from src.errors.http_errors.errors import Errors
+
+from src.models.pydantic_models.model import UserOut
+class CurrentUser:
+    
+    def __init__(self) -> None:
+        self.jwt_service=Jwt()
+        self.hash=Hashes()
+        self.crud_actions=UsersCrudActions()
+        self.error=Errors()
+        
+        
+    def get_current_user(self, token:str)->UserOut:
+        
+        payload=self.jwt_service.jwt_decode(token)
+        sub=payload.get("sub")
+        print(sub)
+        if sub is None:
+            raise self.error.credentials_error(detail="Jwt token is incorrect")
+        try:
+            UUID(sub)
+        except (ValueError, TypeError):
+            raise self.error.credentials_error(detail="Jwt token is incorrect")
+        
+        user=self.crud_actions.get_user_by_id(UUID(sub))
+        
+        if user is None:
+            raise self.error.not_found_error(detail="User with this email address not found")
+        
+        return UserOut.model_validate(user)
+    
+
+    def create_token(self, form_data_email:str, form_data_password:str)->dict:
+        
+        user=self.crud_actions.get_user_by_email(form_data_email)
+        
+        if user is None:
+            raise self.error.credentials_error(detail="Wrong credentials")
+        
+        if not 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 {"access_token":self.jwt_service.create_token({"sub":str(user.id)}), "token_type":"bearer"}
+    
+auth=CurrentUser()

src/service/auth/jwt.py

@@ -0,0 +1,46 @@
+from jose import JWTError, jwt
+import bcrypt
+from src.errors.http_errors.errors import Errors
+from datetime import datetime, timedelta, timezone
+from src.models.configs_read.env import env_settings
+'''Hash/Check hash'''
+class Hashes:
+    
+    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"))
+    
+    
+'''jwt'''
+class Jwt:
+    def __init__(self) -> None:
+        self.error=Errors()
+    
+    def create_token(self, data:dict)->str:
+        
+        user_info=data.copy()
+        user_info.update({"exp": datetime.now(timezone.utc)+timedelta(minutes=env_settings.ACCESS_TOKEN_EXPIRE_MINUTES)})
+        
+        return jwt.encode(user_info, env_settings.SECRET_KEY, env_settings.ALGORITHM)
+        
+        
+    def jwt_decode(self, token:str)->dict:
+        
+        try:
+            
+            payload=jwt.decode(token, env_settings.SECRET_KEY, algorithms=[env_settings.ALGORITHM])
+            
+            if (payload.get("sub")) is None:
+                raise self.error.credentials_error(detail="Sub block is missing")
+            
+        except JWTError as e:
+            
+            raise self.error.credentials_error(detail="JWTerror") from e
+        
+        return payload
+