Session handling and user controller

Author: kartikbehl99

.isort.cfg

@@ -2,4 +2,4 @@
 line_length = 88
 multi_line_output = 3
 include_trailing_comma = True
-known_third_party = cryptography,flask,pytest,sqlalchemy
+known_third_party = config,cryptography,flask,pytest,sqlalchemy,src

requirements.txt

@@ -42,4 +42,34 @@ six==1.15.0
 SQLAlchemy==1.3.19
 toml==0.10.1
 virtualenv==20.0.31
+Werkzeug==1.0.1appdirs==1.4.4
+attrs==20.1.0
+cfgv==3.2.0
+click==7.1.2
+distlib==0.3.1
+filelock==3.0.12
+flake8==3.8.3
+Flask==1.1.2
+identify==1.4.29
+iniconfig==1.0.1
+itsdangerous==1.1.0
+Jinja2==2.11.2
+MarkupSafe==1.1.1
+mccabe==0.6.1
+more-itertools==8.5.0
+nodeenv==1.5.0
+packaging==20.4
+pluggy==0.13.1
+pre-commit==2.7.1
+psycopg2==2.8.5
+py==1.9.0
+pycodestyle==2.6.0
+pyflakes==2.2.0
+pyparsing==2.4.7
+pytest==6.0.1
+PyYAML==5.3.1
+six==1.15.0
+SQLAlchemy==1.3.19
+toml==0.10.1
+virtualenv==20.0.31
 Werkzeug==1.0.1

ssh_manager_backend/app/controllers/__init__.py

@@ -0,0 +1,101 @@
+import os
+
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
+
+from ssh_manager_backend.app.services.aes import AES
+
+
+"""
+ABBREVIATIONS:
+
+kek = key encryption key
+dek = decryption key
+iv = initialization vector
+"""
+
+
+class Secrets:
+    __csprnglength__ = 32
+    __ivlength__ = 16
+
+    def __init__(self):
+        self.kek = None
+        self.dek = None
+        self.salt_for_dek = None
+        self.iv_for_dek = None
+        self.salt_for_kek = None
+        self.iv_for_kek = None
+        self.salt_for_password = None
+
+    def set_secrets(self, secrets: dict):
+        """
+        Sets the value for class attributes based on the input
+
+        Parameters
+        ----------
+        secrets: dict
+            The secrets dictionary used for setting the class attributes
+        ----------
+        """
+
+        self.dek = secrets["encryptedDek"]
+        self.iv_for_dek = secrets["ivForDek"]
+        self.salt_for_dek = secrets["saltForDek"]
+        self.iv_for_kek = secrets["ivForKek"]
+        self.salt_for_kek = secrets["saltForKek"]
+        self.salt_for_password = secrets["saltForPassword"]
+
+    @staticmethod
+    def generate_secure_random(length: int) -> bytes:
+        """
+        Generated a cryptographically secure sequence of random bytes of the specified length
+
+        :param length:
+        """
+
+        secure_random = os.urandom(length)
+        return secure_random
+
+    def generate_kek(self, password: str) -> bytes:
+        """
+        Generated key encryption key from the password using PBKDF2 (Password based key definition function)
+
+        :param password:
+        """
+
+        kdf = PBKDF2HMAC(
+            algorithm=hashes.SHA256(),
+            length=32,
+            salt=self.salt_for_kek,
+            iterations=100000,
+            backend=default_backend(),
+        )
+
+        return kdf.derive(bytes(password, encoding="utf-8"))
+
+    def encrypt_dek(self) -> bytes:
+        """
+        Encrypts the decryption key from the user's password/kek
+        """
+
+        aes = AES(self.kek, self.iv_for_kek)
+        return aes.encrypt(self.dek)
+
+    def generate_secrets(self, password: bytes):
+        """
+        A utility function which calls the above functions for generating the value of secrets. This function is
+        called during the registration phase.
+
+        :param password:
+        """
+
+        self.salt_for_password = self.generate_secure_random(self.__csprnglength__)
+        self.dek = self.generate_secure_random(self.__csprnglength__)
+        self.salt_for_dek = self.generate_secure_random(self.__csprnglength__)
+        self.iv_for_dek = self.generate_secure_random(self.__ivlength__)
+        self.salt_for_kek = self.generate_secure_random(self.__csprnglength__)
+        self.iv_for_kek = self.generate_secure_random(self.__ivlength__)
+        self.kek = self.generate_kek(password)
+        self.dek = self.encrypt_dek()

ssh_manager_backend/app/controllers/secrets.py

@@ -0,0 +1,126 @@
+import uuid
+from typing import Tuple, Union
+
+from src.modules.utils import hash_data
+
+from ssh_manager_backend.app.controllers.secrets import Secrets
+from ssh_manager_backend.app.models import SessionModel, UserModel
+
+
+class User:
+    def __init__(self):
+        self.secrets: Secrets = Secrets()
+        self.username: str = ""
+        self.__access_token: str = ""
+        self.__password: str = ""
+        self.email: str = ""
+        self.name: str = ""
+        self.admin: str = ""
+        self.user: UserModel = UserModel()
+
+    def set_attributes(
+        self, username: str, password: str, email: str, name: str, access_token: str
+    ):
+        """
+        Sets the class attribute based on the function arguments.
+
+        :param username: The username of the user
+        :param password: The password of the user
+        :param email: The email of the user
+        :param name: The name of the user
+        :param access_token: The access token of the user
+        :return:
+        """
+
+        self.username = username
+        self.__password = password
+        self.email = email
+        self.name = name
+        self.__access_token = access_token
+
+    def register(self, username: str, password: str, email: str, name: str) -> bool:
+        """
+        Generates the secrets of the user based on the password. Hashes the password using SAH256 and stores the
+        generated information in DynamoDB.
+
+        :param username:
+        :param password:
+        :param email:
+        :param name:
+        :param table_name:
+        :return: True/False for success/failure
+        """
+
+        self.set_attributes(
+            username=username,
+            password=password,
+            email=email,
+            name=name,
+            access_token=uuid.uuid4().hex,
+        )
+
+        self.secrets.generate_secrets(password=self.__password)
+        self.__password = hash_data(self.__password, self.salt_for_password)
+        user_is_admin = True
+
+        if self.user.admin_exists():
+            user_is_admin = False
+
+        return self.user.create(
+            name=name,
+            username=self.username,
+            password=self.__password,
+            admin=user_is_admin,
+            encrypted_dek=self.secrets.encrypted_dek,
+            iv_for_dek=self.secrets.iv_for_dek,
+            salt_for_dek=self.secrets.salt_for_dek,
+            iv_for_kek=self.secrets.iv_for_kek,
+            salt_for_kek=self.secrets.salt_for_kek,
+            salt_for_password=self.secrets.salt_for_password,
+        )
+
+    def login(self, username: str, password: str) -> Tuple[Union[bool, str]]:
+        """
+        Handles user login.
+
+        :param username: The username of the user
+        :param password: The password of the user
+        :param table_name: The table in which the data is to be inserted
+        :return: access token upon successful login
+        """
+
+        if self.user.exists(username):
+            if self.user.password_match(
+                username=username, password=hash_data(password)
+            ):
+                access_token: str = uuid.uuid4()
+                session: SessionModel = SessionModel()
+                if not session.exists(username=username):
+                    session.create(usernmae=username, access_token=access_token)
+                else:
+                    session.activate_session(username=username)
+
+                return (True, access_token)
+            else:
+                return (False, "Password does not match")
+
+        return (False, "User does not exists")
+
+    def is_admin(self) -> bool:
+        return self.user.get_user(username=self.username).admin
+
+    @property
+    def password(self) -> str:
+        """
+        Returns the password of the user
+        """
+
+        return self.__password
+
+    @property
+    def access_token(self) -> str:
+        """
+        Returns the access token of the user
+        """
+
+        return self.__access_token

ssh_manager_backend/app/controllers/user.py

@@ -1,4 +1,5 @@
-from ssh_manager_backend.app.models.access_control import AccessControl
-from ssh_manager_backend.app.models.keys import Key
-from ssh_manager_backend.app.models.keys_mapping import KeyMapping
-from ssh_manager_backend.app.models.user import User
+from ssh_manager_backend.app.models.access_control import AccessControlModel
+from ssh_manager_backend.app.models.keys import KeyModel
+from ssh_manager_backend.app.models.keys_mapping import KeyMappingModel
+from ssh_manager_backend.app.models.session import SessionModel
+from ssh_manager_backend.app.models.user import UserModel

ssh_manager_backend/app/models/__init__.py

@@ -0,0 +1,99 @@
+from typing import Union
+
+from sqlalchemy.exc import SQLAlchemyError
+
+from ssh_manager_backend.db import User, UserSession
+from ssh_manager_backend.db.database import db_session
+
+
+class SessionModel:
+    def __init__(self):
+        self.session = db_session()
+
+    def exists(self, username: str) -> bool:
+        """
+        Checks whether a session exists.
+
+        :params username: The username of the user.
+        :return:
+        """
+
+        return (
+            self.session.query(UserSession)
+            .join(User)
+            .filter(User.username == username)
+            .first()
+            is not None
+        )
+
+    def create(self, username: str, access_token: str) -> bool:
+        """
+        Creates a user session and returns access token upon success.
+        :param username: The username of the user,
+        :return:
+        """
+
+        try:
+            user_session: UserSession = UserSession(
+                access_token=access_token, username=username, active=True
+            )
+            self.session.add(user_session)
+            self.session.commit()
+        except SQLAlchemyError:
+            self.session.rollback()
+            return False
+        except AttributeError:
+            return False
+
+        return True
+
+    def activate_session(self, username: str) -> bool:
+        """
+        Activates a user session.
+        :param username: The username of the user,
+        :return:
+        """
+
+        try:
+            self.session.query(UserSession).filter(
+                UserSession.username == username
+            ).update({"active": True})
+            self.session.commit()
+
+            return True
+        except AttributeError:
+            return False
+
+    def deactivate_session(self, username: str) -> bool:
+        """
+        Activates a user session.
+        :param username: The username of the user,
+        :return:
+        """
+
+        try:
+            self.session.query(UserSession).filter(
+                UserSession.username == username
+            ).update({"active": False})
+            self.session.commit()
+            return True
+        except AttributeError:
+            return False
+
+    def access_token(self, username: str) -> Union[bool, str]:
+        """
+        Returns access token of user.
+        :param username: The usernameof the user,
+        :return: access token
+        """
+
+        try:
+            access_token: str = (
+                self.session.query(UserSession)
+                .filter(UserSession.username == username)
+                .first()
+                .access_token
+            )
+            return access_token
+        except AttributeError:
+            return False