ska-referee-test/ref-test/app/models/user.py

from ..extensions import db, mail
from ..tools.encryption import decrypt, encrypt
from ..tools.logs import write

from flask import jsonify, session
from flask.helpers import flash, url_for
from flask_login import current_user, login_user, logout_user, UserMixin
from flask_mail import Message
from smtplib import SMTPException
from sqlalchemy.exc import SQLAlchemyError
from werkzeug.security import check_password_hash, generate_password_hash

import secrets
from uuid import uuid4
class User(UserMixin, db.Model):
    id = db.Column(db.String(36), primary_key=True)
    username = db.Column(db.String(128), nullable=False)
    password = db.Column(db.String(128), nullable=False)
    email = db.Column(db.String(128), nullable=False)
    reset_token = db.Column(db.String(20), nullable=True)
    verification_token = db.Column(db.String(20), nullable=True)
    tests = db.relationship('Test', backref='creator')
    datasets = db.relationship('Dataset', backref='creator')

    def __repr__(self):
        return f'<user {self.username}> was added with <id {self.id}>.'
    
    @property
    def generate_id(self): raise AttributeError('generate_id is not a readable attribute.')

    generate_id.setter
    def generate_id(self): self.id = uuid4().hex

    @property
    def set_username(self): raise AttributeError('set_username is not a readable attribute.')
    
    set_username.setter
    def set_username(self, username:str): self.username = encrypt(username)
    
    def get_username(self): return decrypt(self.username)

    @property
    def set_password(self): raise AttributeError('set_password is not a readable attribute.')

    set_password.setter
    def set_password(self, password:str): self.password = generate_password_hash(password, method="sha256")

    def verify_password(self, password:str): return check_password_hash(self.password, password)

    @property
    def set_email(self): raise AttributeError('set_email is not a readable attribute.')
    
    set_email.setter
    def set_email(self, email:str): self.email = encrypt(email)
    
    def get_email(self): return decrypt(self.email)

    def register(self, notify:bool=False, password:str=None):
        self.generate_id()
        try: users = User.query.all()
        except (SQLAlchemyError, ConnectionError) as exception:
            write('system.log', f'Database error when setting default dataset {self.id}: {exception}')
            return False, f'Database error {exception}.'
        for user in users:
            if user.get_username() == self.get_username(): return False, f'Username {self.get_username()} already in use.'
            if user.get_email() == self.get_email(): return False, f'Email address {self.get_email()} already in use.'
        self.set_password(password=password)
        try:
            db.session.add(self)
            db.session.commit()
        except (SQLAlchemyError, ConnectionError) as exception:
            db.session.rollback()
            write('system.log', f'Database error when registering user {self.get_username()}: {exception}')
            return False, f'Database error: {exception}'
        write('users.log', f'User \'{self.get_username()}\' was created with id \'{self.id}\'.')
        if notify:
            email = Message(
                subject='RefTest | Registration Confirmation',
                recipients=[self.email],
                body=f"""
                Hello {self.get_username()},\n\n
                You have been registered as an administrator on the SKA RefTest App!\n\n
                You can access your account using the username '{self.get_username()}'\n\n
                Your password is as follows:\n\n
                {password}\n\n
                You can log in to the admin console via the following URL, where you can administer the test or change your password:\n\n
                {url_for('admin._home', _external=True)}\n\n
                Have a nice day!\n\n
                SKA Refereeing
                """,
                html=f"""
                <p>Hello {self.get_username()},</p>
                <p>You have been registered as an administrator on the SKA RefTest App!</p>
                <p>You can access your account using the username '{self.get_username()}'</p>
                <p>Your password is as follows:</p>
                <strong>{password}</strong>
                <p>You can log in to the admin console via the following URL, where you can administer the test or change your password:</p>
                <p><a href='{url_for('admin._home', _external=True)}'>{url_for('admin._home', _external=True)}</a></p>
                <p>Have a nice day!</p>
                <p>SKA Refereeing</p>
                """
            )
            try: mail.send(email)
            except (SMTPException, ConnectionError) as exception: write('system.log', f'SMTP Error while trying to notify new user account creation to {self.get_username()} with error: {exception}')
        return True, f'User {self.get_username()} was created successfully.'

    def login(self, remember:bool=False):
        login_user(self, remember = remember)
        write('users.log', f'User \'{self.get_username()}\' has logged in.')
        flash(message=f'Welcome {self.get_username()}', category='success')
    
    def logout(self):
        session['remembered_username'] = self.get_username()
        logout_user()
        write('users.log', f'User \'{self.get_username()}\' has logged out.')
        flash(message='You have successfully logged out.', category='success')
    
    def reset_password(self):
        new_password = secrets.token_hex(12)
        self.set_password(new_password)
        self.reset_token = secrets.token_urlsafe(16)
        self.verification_token = secrets.token_urlsafe(16)
        email = Message(
            subject='RefTest | Password Reset',
            recipients=[self.get_email()],
            body=f"""
            Hello {self.get_username()},\n\n
            This email was generated because we received a request to reset the password for your administrator account for the SKA RefTest app.\n\n
            If you did not make this request, please ignore this email.\n\n
            If you did make this request, then you have two options to recover your account.\n\n
            Your password has been reset to the following:\n\n
            {new_password}\n\n
            You may use this to log back in to your account, and subsequently change your password to something more suitable.\n\n
            Alternatively, you may visit the following private link using your unique token to override your password. Copy and paste the following link in a web browser. Please note that this token is only valid once:\n\n
            {url_for('admin._reset', token = self.reset_token, verification = self.verification_token, _external = True)}\n\n
            Hopefully, this should enable access to your account once again.\n\n
            Have a nice day!\n\n
            SKA Refereeing
            """,
            html=f"""
            <p>Hello {self.get_username()},</p>
            <p>This email was generated because we received a request to reset the password for your administrator account for the SKA RefTest app.</p>
            <p>If you did not make this request, please ignore this email.</p>
            <p>If you did make this request, then you have two options to recover your account.</p>
            <p>Your password has been reset to the following:</p>
            <strong>{new_password}</strong>
            <p>You may use this to log back in to your account, and subsequently change your password to something more suitable.</p>
            <p>Alternatively, you may visit the following private link using your unique token to override your password. Copy and paste the following link in a web browser. Please note that this token is only valid once:</p>
            <p><a href='{url_for('admin._reset', token = self.reset_token, verification = self.verification_token, _external = True)}'>{url_for('admin._reset', token = self.reset_token, verification = self.verification_token, _external = True)}</a></p>
            <p>Hopefully, this should enable access to your account once again.</p>
            <p>Have a nice day!</p>
            <p>SKA Refereeing</p>
            """
        )
        try: mail.send(email)
        except (SMTPException, ConnectionError) as exception:
            write('system.log', f'SMTP Error while trying to reset password for {self.get_username()} with error: {exception}')
            db.session.rollback()
            return jsonify({'error': f'SMTP Error: {exception}'}), 500
        try: db.session.commit()
        except (SQLAlchemyError, ConnectionError) as exception:
            db.session.rollback()
            write('system.log', f'Database error when resetting password for user {self.get_username()}: {exception}')
            return False, f'Database error: {exception}'
        return jsonify({'success': 'Your password reset link has been generated.'}), 200

    def clear_reset_tokens(self):
        self.reset_token = self.verification_token = None
        try: db.session.commit()
        except (SQLAlchemyError, ConnectionError) as exception:
            db.session.rollback()
            write('system.log', f'Database error when resetting clearing reset tokens for user {self.get_username()}: {exception}')
            return False, f'Database error: {exception}'
    
    def delete(self, notify:bool=False):
        username = self.get_username()
        email_address = self.get_email()
        try:
            db.session.delete(self)
            db.session.commit()
        except (SQLAlchemyError, ConnectionError) as exception:
            db.session.rollback()
            write('system.log', f'Database error when deleting user {self.get_username()}: {exception}')
            return False, f'Database error: {exception}'
        message = f'User \'{username}\' was deleted by \'{current_user.get_username()}\'.'
        write('users.log', message)
        if notify:
            email = Message(
                subject='RefTest | Account Deletion',
                recipients=[email_address],
                bcc=[current_user.get_email()],
                body=f"""
                Hello {username},\n\n
                Your administrator account for the SKA RefTest App, as well as all data associated with the account, have been deleted by {current_user.get_username()}.\n\n
                If you believe this was done in error, please contact them immediately.\n\n
                If you would like to register to administer the app, please ask an existing administrator to create a new account.\n\n
                Have a nice day!\n\n
                SKA Refereeing
                """,
                html=f"""
                <p>Hello {username},</p>
                <p>Your administrator account for the SKA RefTest App, as well as all data associated with the account, have been deleted by {current_user.get_username()}.</p>
                <p>If you believe this was done in error, please contact them immediately.</p>
                <p>If you would like to register to administer the app, please ask an existing administrator to create a new account.</p>
                <p>Have a nice day!</p>
                <p>SKA Refereeing</p>
                """
            )
            try: mail.send(email)
            except (SMTPException, ConnectionError) as exception: write('system.log', f'SMTP Error when trying to delete account {username} with error: {exception}')
        return True, message

    def update(self, password:str=None, email:str=None, notify:bool=False):
        if not password and not email: return False, 'There were no changes requested.'
        if password: self.set_password(password)
        old_email = self.get_email()
        if email:
            try:
                for entry in User.query.all():
                    if entry.get_email() == email and not entry == self: return False, f'The email address {email} is already in use.'
            except (SQLAlchemyError, ConnectionError) as exception:
                write('system.log', f'Database error when setting default dataset {self.id}: {exception}')
                return False, f'Database error {exception}.'
            self.set_email(email)
        try: db.session.commit()
        except (SQLAlchemyError, ConnectionError) as exception:
            db.session.rollback()
            write('system.log', f'Database error when updating user {self.get_username()}: {exception}')
            return False, f'Database error: {exception}'
        _current_user = 'command line' if not current_user else 'anonymous' if not current_user.is_authenticated else current_user.get_username()
        write('system.log', f'Information for user {self.get_username()} has been updated by {_current_user}.')
        if notify:
            message = Message(
                subject='RefTest | Account Update',
                recipients=[email],
                bcc=[old_email,current_user.get_email()],
                body=f"""
                Hello {self.get_username()},\n\n
                Your administrator account for the SKA RefTest App has been updated by {_current_user}.\n\n
                Your new account details are as follows:\n\n
                Email: {email}\n
                Password: {password if password else '<same as old>'}\n\n
                You can update your email address and password by logging in to the admin console using the following URL:\n\n
                {url_for('admin._home', _external=True)}\n\n
                Have a nice day!\n\n
                SKA Refereeing
                """,
                html=f"""
                <p>Hello {self.get_username()},</p>
                <p>Your administrator account for the SKA RefTest App has been updated by {_current_user}.</p>
                <p>Your new account details are as follows:</p>
                <p>Email: {email} <br/> Password: <strong>{password if password else '&lt;same as old&gt;'}</strong></p>
                <p>You can update your email address and password by logging in to the admin console using the following URL:</p>
                <p><a href='{url_for('admin._home', _external=True)}'>{url_for('admin._home', _external=True)}</a></p>
                <p>Have a nice day!</p>
                <p>SKA Refereeing</p>
                """
            )
            try: mail.send(message)
            except (SMTPException, ConnectionError) as exception: write('system.log', f'SMTP Error when trying to update account {self.get_username()} with error: {exception}')
        return True, f'Account {self.get_username()} has been updated.'