AssetManager-V1.0- Yet to be reviewed

Updated the code to include new user login if the user email domain is knowledgelens. Also, added necessary validations to the code.

app.py

@@ -6,14 +6,33 @@ if __name__ == "__main__":
 
 import uvicorn
 
-from scripts.config import PROJECT_NAME, Services
+from scripts.config import Services
+import argparse
 from scripts.logging.logger import logger
 
 # starting the application
+ap = argparse.ArgumentParser()
+
 if __name__ == "__main__":
     try:
-        print("Api for " + PROJECT_NAME)
-        # uvicorn run
-        uvicorn.run("main:app", port=int(Services.PORT))
+        ap.add_argument(
+            "--port",
+            "-p",
+            required=False,
+            default=Services.PORT,
+            help="Port to start the application.",
+        )
+        ap.add_argument(
+            "--bind",
+            "-b",
+            required=False,
+            default=Services.HOST,
+            help="IF to start the application.",
+        )
+        arguments = vars(ap.parse_args())
+
+        logger.info(f"App Starting at {arguments['bind']}:{arguments['port']}")
+        uvicorn.run("main:app", host=arguments["bind"], port=int(arguments["port"]))
     except Exception as e:
-        logger.error(e)
+        logger.exception(e)
+

conf/application.conf

-[PROJECT_NAME]
+[PROJECT_DETAILS]
 PROJECT_NAME = AssetManager
+PROJECT_ID = 1256
 
 [PATH]
 base_path = scripts/

main.py

@@ -15,7 +15,7 @@ if __name__ == "__main__":
 
 from fastapi import FastAPI
 from scripts.services import router
-from scripts.config import PROJECT_NAME, Services as ServiceConf
+from scripts.config import Services as ServiceConf
 from fastapi.middleware.cors import CORSMiddleware
 from scripts.logging.logger import logger
 
@@ -25,7 +25,7 @@ app.include_router(router)
 # starting the application
 if __name__ == "__main__":
     try:
-        print("Api for " + PROJECT_NAME)
+        print("Api for " + ServiceConf.PROJECT_NAME)
         # enabling cors for getting UI data
         if ServiceConf.ENABLE_CORS:
             app.add_middleware(

scripts/config/__init__.py

@@ -5,12 +5,13 @@ from pydantic import BaseSettings, Field
 
 config = configparser.RawConfigParser()
 config.read("conf/application.conf")
-PROJECT_NAME = config.get("PROJECT_NAME", "PROJECT_NAME")
 
 
 class _Services(BaseSettings):
     HOST: str = Field(default="127.0.0.1", env="service_host")
     PORT: int = Field(default=8000, env="service_port")
+    PROJECT_NAME = config.get("PROJECT_DETAILS", "PROJECT_NAME")
+    PROJECT_ID = config.get("PROJECT_DETAILS", "PROJECT_ID")
     # path
     BASE_PATH = config.get("PATH", 'base_path')
     SUB_PATH = config.get("PATH", "sub_path")

scripts/core/handlers/login_handler.py

 from __future__ import annotations
 
 import base64
+import datetime
 
 from Cryptodome.Cipher import AES
 from passlib.context import CryptContext
@@ -9,27 +10,51 @@ from scripts.config import Services
 from scripts.database.mongo.mongo_login import MongoUser
 from scripts.errors import ErrorMessages
 from scripts.logging.logger import logger
+from scripts.utils.mongo_default_queries import MongoQueries
 
 
 class LoginHandlers:
     def __init__(self):
+        self.mongo_user = MongoUser()
+        self.mongo_queries = MongoQueries()
         self.pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
         self.db_user_data = None
+        self.db_user_data = None
+        self.dt = datetime.datetime.now()
+        self.time_dt = datetime.datetime.now()
 
     @staticmethod
     def un_pad(s):
+        # un_padding the encrypted password
         return s[:-ord(s[len(s) - 1:])]
 
+    def new_user_login(self, login_data, password):
+        # check the domain of the email the user has entered
+        if login_data.username.split("@")[-1] == "knowledgelens.com":
+            # hash the password
+            hashed_password = self.pwd_context.hash(password)
+            # Enter the user as a guest user to the db
+            if self.mongo_user.insert_user(
+                    self.mongo_queries.insert_user_query(
+                        login_data,
+                        Services.PROJECT_ID,
+                        hashed_password,
+                        self.time_dt)):
+                return True
+        return False
+
     def password_decrypt(self, password):
         try:
             # encoding the Key
             key = Services.KEY_ENCRYPTION.encode('utf-8')
             # decoding the received password
             enc = base64.b64decode(password)
+            # mode for the decryption
+            mode = AES.MODE_CBC
             # getting the initialization vector
             iv = enc[:AES.block_size]
             # decoding with AES
-            cipher = AES.new(key, AES.MODE_CBC, iv)
+            cipher = AES.new(key, mode, iv)
             # decoding the password
             data = cipher.decrypt(enc[AES.block_size:])
             if len(data) == 0:
@@ -53,29 +78,40 @@ class LoginHandlers:
         except Exception as e:
             logger.exception(e)
 
-    def db_data_validation(self, username):
+    def db_data_validation(self, login_data, password):
         try:
             # fetching the data based on the username
-            self.db_user_data = MongoUser().fetch_user_details(username)
+            self.db_user_data = MongoUser().fetch_user_details(login_data.username)
             # if the user is not available
             if not self.db_user_data:
-                return False, {"message": ErrorMessages.ERROR_USER_NOT_REGISTERED, "data": username}
+                # if the domain of the user email is knowledge lens create the user as a guest user
+                if self.new_user_login(login_data, password):
+                    return True, {"message": "new_user", "username": login_data.username, "role": "guest"}
+                else:
+                    return False, {"message": ErrorMessages.ERROR_INVALID_USERNAME,
+                                   "data": {login_data.username, login_data.password}}
+            # Check the project id from the request body
+            if self.db_user_data["project_id"] != Services.PROJECT_ID or Services.PROJECT_ID != login_data.project_id:
+                return False, {"message": ErrorMessages.ERROR_UNAUTHORIZED_USER_LOGIN, "data": login_data.username}
             # if the user exist
             return None, {"message": True}
         except Exception as e:
             logger.exception(e)
 
-    def db_password_matching(self, username, password):
+    def db_password_matching(self, login_data, password):
         try:
             # getting the response after checking for the user data in db
-            response, message = self.db_data_validation(username)
-            # if the response is false then a error message is send back
+            response, message = self.db_data_validation(login_data, password)
+            if response and message["message"] == "new_user":
+                return None, message
+            # if the response is false then an error message is send back
             if response is not None:
                 return response, message
             # if the user exists in db then password is matched
             if not self.pwd_context.verify(password, self.db_user_data["password"]):
-                return False, {"message": ErrorMessages.ERROR_PASSWORD_MISMATCH, "data": {"username": username}}
+                return False, {"message": ErrorMessages.ERROR_PASSWORD_MISMATCH,
+                               "data": {"username": login_data.username}}
             # if the password is correct
-            return None, {"username": username, "role": self.db_user_data["user_role"]}
+            return None, {"username": login_data.username, "role": self.db_user_data["user_role"]}
         except Exception as e:
             logger.exception(e)

scripts/errors/__init__.py

@@ -6,5 +6,6 @@ class ErrorMessages:
     ERROR_INVALID_LOGIN = "Your are not authorized to view this website."
     ERROR_INVALID_USERNAME = "Invalid Username"
     ERROR_INVALID_PASSWORD = "Invalid Password"
+    ERROR_UNAUTHORIZED_USER_LOGIN = "Account is not available"
     ERROR_USER_NOT_REGISTERED = "Account is not registered in the portal."
     ERROR_PASSWORD_MISMATCH = "Passwords Authentication Failed. Please enter the correct password"

scripts/logging/logger.py

@@ -2,7 +2,7 @@ import logging
 import os
 from logging.handlers import RotatingFileHandler
 
-from scripts.config import Services, PROJECT_NAME
+from scripts.config import Services
 
 
 def get_logger():
@@ -29,7 +29,7 @@ def get_logger():
         os.makedirs(file_path)
 
     # joining the path
-    log_file = os.path.join(f"{file_path}{PROJECT_NAME}log.log")
+    log_file = os.path.join(f"{file_path}{Services.PROJECT_NAME}log.log")
 
     # creating rotating file handler with max byte as 1
     temp_handler = RotatingFileHandler(log_file, maxBytes=1)

scripts/services/v1/iot_manager_services.py

@@ -18,15 +18,13 @@ def login_default(login_data: NormalLogin):
     try:
         # decrypting the password from the UI
         decrypted_password = obj_login_handler.password_decrypt(login_data.password)
-
         # validating the received inputs empty or not
         response = obj_login_handler.user_data_validation(login_data.username, decrypted_password)
         if response is not None:
             return JSONResponse(content=DefaultFailureResponse(error=response["message"]).dict(),
                                 status_code=status.HTTP_400_BAD_REQUEST)
-
         # checking for the account and password matching
-        response, data = obj_login_handler.db_password_matching(login_data.username, decrypted_password)
+        response, data = obj_login_handler.db_password_matching(login_data, decrypted_password)
         if response is not None:
             return JSONResponse(content=DefaultFailureResponse(error=data).dict(),
                                 status_code=status.HTTP_401_UNAUTHORIZED)

scripts/utils/mongo_default_queries.py

+class MongoQueries:
+    @staticmethod
+    def insert_user_query(login_data, project_id, hashed_password, time_dt):
+        query = {"project_id": project_id, "name": "", "email": login_data.username,
+                 "password": hashed_password,
+                 "user_role": "guest",
+                 "is_alive": True, "created_at": time_dt,
+                 "updated_at": time_dt}
+        return query