settlements-admin/backend/code/modules/sessions/controller.py

import os
from datetime import datetime, timedelta
from backend.code.modules.database.database_sqlite import DatabaseSqliteModule
from backend.code.modules.sessions.repository import SessionRepository
from backend.code.services.token import TokenService
from backend.code.services.mailing import MailingService
from backend.code.enums.errors_enum import ErrorTypeEnum
from backend.code.models.responses.session_token import SessionTokenResponse


from backend.code.modules.users.repository import UserRepository

class SessionsController():
    def __init__(self):
        database = DatabaseSqliteModule()
        self.repo = SessionRepository(database)
        self.token_service = TokenService()
        self.mailing_service = MailingService()

    def Login(self, user: str, password: str, domain: str = None):
        account = self.repo.SelectByUsername(user)
        if account is None:
            return

        # Array to properties
        account_id = account[0]
        user_id = account[1]
        _password = account[3]
        attempts = account[4]
        locked = account[5]
        name = account[7]
        surnames = account[8]
        email = account[9]

        # Validations
        if locked:
            return ErrorTypeEnum.USER_ACCOUNT_LOCKED

        # Default user has no password
        if _password is None:
            return ErrorTypeEnum.USER_ACCOUNT_NO_PASSWORD

        if password != _password:
            attempts = attempts + 1
            self._OnUpdateAttempts(account_id, attempts)
            return ErrorTypeEnum.USER_ACCOUNT_WRONG_PASSWORD

        # Reset when password is correct and registered some attempts
        if attempts > 0:
            self._OnUpdateAttempts(account_id, 0)

        # Generate token
        payload = {
            'UserId': user_id,
            'Name': name,
            'Surnames': surnames,
            'Email': email
        }
        token = self.token_service.GenerateToken(payload)

        session = self.repo.RegisterSession(account_id, token)
        return SessionTokenResponse(Token=token)
        
    def ValidateToken(self, token: str):
        exists = self.repo.SelectByToken(token)
        if exists is None:
            return ErrorTypeEnum.USER_SESSION_NOT_EXISTS

        # Array to properties
        account_id = exists[0]
        user_id = exists[1]

        # Get metadata token
        payload = self.token_service.ValidateToken(token)
        if payload == False:
            return

        # Check token integrity 
        if payload['UserId'] != user_id:
            return

        autorefresh = self._OnCheckTokenAutoRefresh(payload['exp'])
        if autorefresh:
            token = self._OnRefreshToken(account_id, token, payload)

        return token

    def Logout(self, token: str):
        exists = self.repo.SelectByToken(token)
        if exists is None:
            return

        disabled = self.repo.DisableToken(token)
        if disabled is None:
            return 

        return disabled == 1
        
    def RecoverAccount(self, username: str, domain: str = None):
        account = self.repo.SelectByUsername(username)
        if account is None:
            return

        # Array to properties
        AccountId = account[0]
        Fullname = f'{account[7]} {account[8]}'
        Email = account[9]
        now = datetime.now()

        # Generate token
        payload = {'Account': AccountId}
        token = self.token_service.GenerateToken(payload)
        url = f'{os.getenv('DOMAIN_URL')}/restore/{token}'

        registered = self.repo.InsertRecoverRequest(AccountId, token)
        if registered is None:
            return

        # Prepare notification
        template = '''
        Hola {},

        Has pedido restaurar tu contraseña de {}. Accede al siguiente enlace para establecer una nueva contraseña:
        {}

        Si no has realizado esta petición de restauración de la cuenta, envia un correo a: {}
        '''       
        message = template.format(Fullname, '[APP_NAME]', url, os.getenv('SUPPORT_EMAIL'))

        # Send notification
        self.mailing_service.SendMessage(
            Email,
            'Petición restaurar cuenta [APP_NAME]',
            message
        )

    def ValidateTokenRecoverAccount(self, token: str):
        validated = self.token_service.ValidateToken(token)
        if validated is False:
            return

        exists = self.repo.SelectRecoverRequest(validated['Account'], token)
        if exists is None:
            return

        disabled = self.repo.DisableRecoverRequest(token)
        if disabled == 0:
            return

        return True

    # region Utilities
    def _OnUpdateAttempts(self, account_id: int, attempts: int):
        locked = False
        limit = int(os.getenv('USER_LIMIT_ATTEMPTS'))
        
        if attempts > limit:
            attempts = limit
            locked = True

        return self.repo.RefreshAttempts(account_id, attempts, locked)
    
    def _OnCheckTokenAutoRefresh(self, expiration: int):
        limit_minutes_left = 10
        now = datetime.now()
        expires_at = datetime.fromtimestamp(expiration)
        difference = ((expires_at - now).total_seconds()) / 60 # seconds to minutes
        return difference <= limit_minutes_left
        
    def _OnRefreshToken(self, account_id: int, token: str, payload: dict):
        old_disabled = self.repo.DisableToken(token)
        if old_disabled is None:
            return

        new_token = self.token_service.GenerateToken(payload)
        
        new_registered = self.repo.RegisterSession(account_id, new_token)
        if new_registered is None:
            return

        return new_token
        
    # endregion