Add auth servies and functools

backend/services/auth/__init__.py

@@ -0,0 +1,2 @@
+from .ops import *
+from .utils.JWTBearer import *

backend/services/auth/ops.py

@@ -0,0 +1,110 @@
+from .utils.auth_funcs import *
+from .utils.JWTBearer import *
+from docguptea.services import parser
+from docguptea.models import *
+from docguptea.services.db.utils.DBQueries import DBQueries
+from docguptea.core.Exceptions import *
+from docguptea import app
+
+# import openai
+# from transformers import RobertaTokenizer, T5ForConditionalGeneration
+
+def ops_signup(response_result: GeneralResponse, data: UserAuth):
+    """Wrapper method to handle signup process.
+
+    Args:
+        response_result: FrontendResponseModel. A TypedDict to return the
+                         response captured from the API to the frontend.
+        data: UserAuth. New user's prospective credentials from the frontend
+                        to create their account.
+
+    Raises:
+        ExistingUserException: If account with entered AADHAR Number already exists.
+    """
+        # querying database to check if user already exist
+    user = DBQueries.fetch_data_from_database('auth', ['username', 'email'], f"username='{data.username}' OR email='{data.email}'")
+    if len(list(user)) != 0:
+        # user with the entered credentials already exists
+        raise ExistingUserException(response_result)
+    
+    DBQueries.insert_to_database('auth', (data.username, Auth.get_password_hash(data.password), data.email), 
+                                 ['username', 'password', 'email'])
+    
+    response_result.status = 'success'
+    response_result.message = [f'User created successfully']
+
+def ops_login(data:LoginCreds):
+    """Wrapper method to handle login process.
+
+    Args:
+        data: LoginCreds. User's credentials from the frontend to login to their account.
+
+    Returns:
+        TokenSchema. A Pydantic BaseModel to return the JWT tokens to the frontend.
+
+    Raises:
+        InvalidCredentialsException: If account with entered credentials does not exist.
+    """
+    # querying database to check if user already exist
+    response_result = GeneralResponse.get_instance(data={},
+                                      status="not_allowed",
+                                      message=["Not authenticated"]
+                                      )
+    user = DBQueries.fetch_data_from_database('auth', ['username', 'password'], f"username='{data.username}'")
+    user = list(user)
+    if len(user) == 0:
+        # user with the entered credentials does not exist
+        raise InvalidCredentialsException(response_result)
+    user = user[0]
+    if not Auth.verify_password(data.password, user[1]) and Auth.verify_username(data.username, user[0]):
+        # password is incorrect
+        raise InvalidCredentialsException(response_result)
+    
+    # password is correct
+    return TokenSchema(access_token=Auth.create_access_token(data.username), 
+                       refresh_token=Auth.create_refresh_token(data.username),
+                       )
+
+def ops_regenerate_api_key(username:str) -> APIKey:
+
+    user_API_entry = DBQueries.fetch_data_from_database('api_key', 'apikey', f"username='{username}'")
+    user_API_entry = list(user_API_entry)
+    apikey = None
+
+    if len(user_API_entry) != 0:
+        apikey = APIKey(api_key=Auth.generate_api_key(username))
+        DBQueries.update_data_in_database('api_key','apikey',f"username='{username}'", apikey.api_key)
+    
+    else:
+        apikey = Auth.generate_api_key(username)
+        DBQueries.insert_to_database('api_key', (username, apikey), ['username', 'apikey'])
+        apikey = APIKey(api_key=apikey)
+    
+    return apikey
+    
+        
+
+def ops_inference(source_code:str,api_key:str,username:str):
+    response_result = GeneralResponse.get_instance(data={},
+                                                   status="not_allowed",
+                                                   message=["Not authenticated"]
+                                                   )
+
+    user=DBQueries.fetch_data_from_database('api_key', ['apikey'], f"username='{username}'")
+    if len(list(user)) == 0:
+        # user with the entered credentials does not exist
+        raise InfoNotFoundException(response_result,"User not found")
+    elif list(user)[0][0]!=api_key:
+        raise InvalidCredentialsException(response_result)
+    
+    def generate_docstring(source_code_message: str):
+
+
+        llm_response = app.state.llmchain.run({"instruction": source_code_message})
+
+        docstring = Inference(docstr=llm_response)
+        
+    
+        return docstring
+
+    return generate_docstring(source_code)

backend/services/auth/utils/JWTBearer.py

@@ -0,0 +1,75 @@
+"""Custom Authentication & Authorization bearer to authenticate and authorize
+users based on the following factors:
+1. username
+2.password
+3.email
+
+This utility class validates generated JWTs and grants scoped access to users
+according to their roles.
+"""
+from docguptea.core.ConfigEnv import config
+from docguptea.models import TokenPayload
+
+from datetime import datetime
+
+from fastapi import Request, HTTPException
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from pydantic import ValidationError
+from jose import jwt
+
+
+class JWTBearer(HTTPBearer):
+    """Custom bearer to validate access tokens.
+
+    Args:
+        auto_error: bool = True. Internal param to allow auto error detection.
+
+    Raises:
+        HHTTPException(403): If authentication scheme is not `Bearer`.
+        HTTPException(403): If the access token is invalid or expired.
+        HTTPException(403): If authorization code is invalid.
+    """
+    def __init__(self, auto_error: bool = True):
+        super(JWTBearer, self).__init__(auto_error=auto_error)
+
+    async def __call__(self, request: Request) -> str:
+        credentials: HTTPAuthorizationCredentials = await super(JWTBearer, self).__call__(request)
+        if credentials:
+            if not credentials.scheme == "Bearer":
+                raise HTTPException(status_code=403, detail="Invalid authentication scheme.")
+            else:
+                is_valid = JWTBearer.token_validation(credentials.credentials)
+                if not is_valid:
+                    raise HTTPException(status_code=403, detail="Invalid token or expired token.")
+                return credentials.credentials
+        else:
+            raise HTTPException(status_code=403, detail="Invalid authorization code.")
+
+    @staticmethod
+    def token_validation(token: str) -> bool:
+        """Decodes JWTs to check their validity by inspecting expiry and
+         authorization code.
+
+        Args:
+            token: str. Authenticated `access_token` of the user.
+
+        Returns:
+            bool value to indicate validity of the access tokens.
+
+        Raises:
+            jwt.JWTError: If decode fails.
+            ValidationError: If JWTs are not in RFC 7519 standard.
+        """
+        try:
+            payload = jwt.decode(
+                token, config.JWT_SECRET_KEY, algorithms=[config.ALGORITHM]
+            )
+            token_data = TokenPayload(**payload)
+
+            if datetime.fromtimestamp(token_data.exp) < datetime.now():
+                return False
+
+        except(jwt.JWTError, ValidationError):
+            return False
+
+        return True

backend/services/auth/utils/auth_funcs.py

@@ -0,0 +1,171 @@
+"""Utility class to leverage encryption, verification of entered credentials
+and generation of JWT access tokens.
+"""
+from datetime import datetime, timedelta
+from typing import Union, Any
+import secrets
+
+from jose import jwt
+from passlib.context import CryptContext
+from pydantic import ValidationError
+
+from fastapi.exceptions import HTTPException
+
+from docguptea.core.ConfigEnv import config
+from docguptea.core.Exceptions import *
+from docguptea.models import TokenPayload, TokenSchema
+
+
+
+ACCESS_TOKEN_EXPIRE_MINUTES = 30  # 30 minutes
+REFRESH_TOKEN_EXPIRE_MINUTES = 60 * 24 * 3 # 3 days
+
+
+class Auth:
+    """Utility class to perform -  1.encryption via `bcrypt` scheme.
+    2.password hashing 3.verification of credentials and generating
+    access tokens.
+
+    Attrs:
+        pwd_context: CryptContext. Helper for hashing & verifying passwords
+                                   using `bcrypt` algorithm.
+    """
+    pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+    
+
+    @classmethod
+    def get_password_hash(cls,password: str) -> str:
+        """Encrypts the entered password.
+
+        Args:
+            password: str. Entered password.
+
+        Returns:
+            returns hashed(encrypted) password string.
+        """
+        return cls.pwd_context.hash(password)
+
+    @classmethod    
+    def verify_password(cls, plain_password: str, hashed_password: str) -> bool:
+        """Validates if the entered password matches the actual password.
+
+        Args:
+            plain_password: str. Entered password by user.
+            hashed_password: str. hashed password from the database.
+
+        Returns:
+            bool value indicating whether the passwords match or not.
+        """
+        return cls.pwd_context.verify(plain_password, hashed_password)
+
+    @staticmethod
+    def verify_username(entered_username: str, db_username: str) -> bool:
+        """Validates if the entered username matches the actual username.
+
+        Args:
+            entered_username: str. Entered `username` by user.
+            db_username: str. username from the database.
+
+        Returns:
+            bool value indicating whether the village names match or not.
+        """
+        return entered_username == db_username
+
+    @staticmethod
+    def create_access_token(subject: Union[str, Any], expires_delta: int = None) -> str:
+        """Creates JWT access token.
+
+        Args:
+            subject: Union[Any, str]. Hash_key to generate access token from.
+            expires_delta: int = None. Expiry time for the JWT.
+
+        Returns:
+            encoded_jwt: str. Encoded JWT token from the subject of interest.
+        """
+        if expires_delta is not None:
+            expires_delta = datetime.utcnow() + expires_delta
+        else:
+            expires_delta = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+
+        to_encode = {"exp": expires_delta, "sub": str(subject)}
+        encoded_jwt = jwt.encode(to_encode, config.JWT_SECRET_KEY, config.ALGORITHM)
+        return encoded_jwt
+
+    @staticmethod    
+    def create_refresh_token(subject: Union[str, Any], expires_delta: int = None) -> str:
+        """Creates JWT refresh access token.
+
+        Args:
+            subject: Union[Any, str]. Hash_key to generate access token from.
+            expires_delta: int = None. Expiry time for the JWT.
+
+        Returns:
+            encoded_jwt: str. Encoded JWT token from the subject of interest.
+        """
+        if expires_delta is not None:
+            expires_delta = datetime.utcnow() + expires_delta
+        else:
+            expires_delta = datetime.utcnow() + timedelta(minutes=REFRESH_TOKEN_EXPIRE_MINUTES)
+        
+        to_encode = {"exp": expires_delta, "sub": str(subject)}
+        encoded_jwt = jwt.encode(to_encode, config.JWT_REFRESH_SECRET_KEY, config.ALGORITHM)
+        return encoded_jwt
+
+    @staticmethod
+    def generate_access_tokens_from_refresh_tokens(token: str) -> TokenSchema:
+        """Generates a new pair of tokens by implementing rotating
+        refresh_access_tokens.
+
+        Args:
+            token: str. Current valid refresh access token.
+
+        Returns:
+            tokens: TokenSchema. New tokens with new validity.
+
+        Raises:
+            LoginFailedException: If the current refresh access token is
+                                  invalid.
+        """
+        tokens = TokenSchema.get_instance(
+                                        access_token= "",
+                                        refresh_token= "",
+                                        )
+        try:
+            payload = jwt.decode(
+                token, config.JWT_REFRESH_SECRET_KEY, algorithms=[config.ALGORITHM]
+            )
+            token_data = TokenPayload(**payload)
+            if datetime.fromtimestamp(token_data.exp)< datetime.now():
+                raise HTTPException(status_code=403, detail="Invalid token or expired token.")
+        except (jwt.JWTError, ValidationError):
+            raise InvalidCredentialsException(tokens)
+        tokens['access_token'] = Auth.create_access_token(token_data.sub)
+        tokens['refresh_token'] = Auth.create_refresh_token(token_data.sub)
+        tokens['status'] = 'login successful'
+        tokens['role'] = token_data.sub.split("_")[1]
+        return tokens
+
+    @classmethod
+    def generate_api_key(cls, username: str):
+        return cls.get_password_hash(username + secrets.token_urlsafe(25 - len(username)))
+    
+    @classmethod
+    def get_user_credentials(cls,access_token:str):
+        response_result = GeneralResponse.get_instance(data={},
+                                      status="not_allowed",
+                                      message=["Not authenticated"]
+                                      )
+        try:
+            payload = jwt.decode(
+                access_token, config.JWT_SECRET_KEY, algorithms=[config.ALGORITHM]
+            )
+            token_data = TokenPayload(**payload)
+            return token_data.sub
+        except (jwt.JWTError, ValidationError):
+            raise InvalidCredentialsException(response_result)
+        
+    @classmethod
+    def verify_apikey(cls,user_api_key:str,true_api_key:str):
+        return user_api_key == true_api_key
+