Cookie bugfix, removed 'session' string from expiry/age

Sync

ref-test/admin/models/users.py

@@ -0,0 +1,207 @@
+from flask import flash, make_response, Response, session
+from flask.helpers import url_for
+from flask.json import jsonify
+from werkzeug.security import generate_password_hash, check_password_hash
+from werkzeug.utils import redirect
+from flask_mail import Message
+import secrets
+
+from common.security import encrypt, decrypt
+from common.security.database import decrypt_find_one, encrypted_update
+from datetime import datetime, timedelta
+
+class User:
+
+    def __init__(self, _id=None, username=None, password=None, email=None, remember=False):
+        self._id = _id
+        self.username = username
+        self.email = email
+        self.password = password
+        self.remember = remember
+
+    def start_session(self, resp:Response):
+        from main import app
+        resp.set_cookie(
+            key = '_id',
+            value = self._id,
+            max_age = timedelta(days=14) if self.remember else None,
+            path = '/',
+            expires = datetime.utcnow() + timedelta(days=14) if self.remember else None,
+            domain = f'.{app.config["SERVER_NAME"]}',
+            secure = True
+        )
+        if self.remember:
+            resp.set_cookie (
+                key = 'remember',
+                value = 'True',
+                max_age = timedelta(days=14),
+                path = '/',
+                expires = datetime.utcnow() + timedelta(days=14),
+                domain = f'.{app.config["SERVER_NAME"]}',
+                secure = True
+            )
+
+    def register(self):
+        from main import db
+        from ..views import get_id_from_cookie
+        user = {
+            '_id': self._id,
+            'email': encrypt(self.email),
+            'password': generate_password_hash(self.password, method='sha256'),
+            'username': encrypt(self.username)
+        }
+        if decrypt_find_one(db.users, { 'username': self.username }):
+            return jsonify({ 'error': f'Username {self.username} is not available.' }), 400
+        if db.users.insert_one(user):
+            flash(f'User {self.username} has been created successfully. You may now use it to log in and administer the tests.', 'success')
+            resp = make_response(jsonify(user), 200)
+            if not get_id_from_cookie:
+                self.start_session(resp)
+            return resp
+        return jsonify({ 'error': f'Registration failed. An error occurred.' }), 400
+    
+    def login(self):
+        from main import db
+        user = decrypt_find_one( db.users, { 'username': self.username })
+        if not user:
+            return jsonify({ 'error': f'Username {self.username} does not exist.' }), 401
+        if not check_password_hash( user['password'], self.password ):
+            return jsonify({ 'error': f'The password you entered is incorrect.' }), 401
+        response = {
+            'success': f'Successfully logged in user {self.username}.'
+        }
+        if 'prev_page' in session:
+            response['redirect_to'] = session['prev_page']
+            session.pop('prev_page')
+        resp = make_response(jsonify(response), 200)
+        self._id = user['_id']
+        self.start_session(resp)
+        return resp
+
+    def logout(self):
+        resp = make_response(redirect(url_for('admin_auth.login')))
+        from main import app
+        resp.set_cookie(
+            key = '_id',
+            value = '',
+            max_age = timedelta(days=-1),
+            path = '/',
+            expires= datetime.utcnow() + timedelta(days=-1),
+            domain = f'.{app.config["SERVER_NAME"]}',
+            secure = True
+        )
+        resp.set_cookie (
+            key = 'cookie_consent',
+            value = 'True',
+            max_age = None,
+            path = '/',
+            expires = None,
+            domain = f'.{app.config["SERVER_NAME"]}',
+            secure = True
+        )
+        resp.set_cookie (
+            key = 'remember',
+            value = 'True',
+            max_age = timedelta(days=-1),
+            path = '/',
+            expires = datetime.utcnow() + timedelta(days=-1),
+            domain = f'.{app.config["SERVER_NAME"]}',
+            secure = True
+        )
+        flash('You have been logged out. All cookies pertaining to your account have been deleted. Have a nice day.', 'alert')
+        return resp
+
+    def reset_password(self):
+        from main import db, mail
+        user = decrypt_find_one(db.users, { 'username': self.username })
+        if not user:
+            return jsonify({ 'error': f'Username {self.username} does not exist.' }), 401
+        if not user['email'] == self.email:
+            return jsonify({ 'error': f'The email address {self.email} does not match the user account {self.username}.' }), 401
+        new_password = secrets.token_hex(12)
+        reset_token = secrets.token_urlsafe(16)
+        verification_token = secrets.token_urlsafe(16)
+        user['password'] = generate_password_hash(new_password, method='sha256')
+        if encrypted_update(db.users, { 'username': self.username }, { '$set': {'password': user['password'], 'reset_token': reset_token, 'verification_token': verification_token } } ):
+            flash(f'Your password has been reset. Instructions to recover your account have been sent to {self.email}. Please be sure to check your spam folder in case you have not received the email.', 'alert')
+            email = Message(
+                subject = 'RefTest | Password Reset',
+                recipients = [self.email],
+                body = f"""
+                Hello {user['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
+                For the time being, 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_auth.reset_gateway', token1 = reset_token, token2 = verification_token, _external = True)}\n\n
+                Have a nice day.
+                """,
+                html = f"""
+                <p>Hello {user['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>For the time being, 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. Click on the following link, or copy and paste it into a browser. <strong>Please note that this token is only valid once</strong>:</p>
+                <p><a href='{url_for('admin_auth.reset_gateway', token1 = reset_token, token2 = verification_token, _external = True)}'>{url_for('admin_auth.reset_gateway', token1 = reset_token, token2 = verification_token, _external = True)}</a></p>
+                <p>Have a nice day.</p>
+                """
+            )
+            mail.send(email)
+            return jsonify({ 'success': 'Password reset request has been processed.'}), 200
+    
+    def update(self):
+        from main import db
+        from ..views import get_id_from_cookie
+        retrieved_user = decrypt_find_one(db.users, { '_id': self._id })
+        if not retrieved_user:
+            return jsonify({ 'error': f'User {retrieved_user["username"]} does not exist.' }), 401
+        user = {}
+        updated = []
+        if not self.email == '' and self.email is not None:
+            user['email'] = self.email
+            updated.append('email')
+        if not self.password == '' and self.password is not None:
+            user['password'] = generate_password_hash(self.password, method='sha256')
+            updated.append('password')
+        output = ''
+        if len(updated) == 0:
+            flash(f'There were no changes requested for your account.', 'alert'), 200
+            return jsonify({'success': 'There were no changes requested for your account.'}), 200
+        elif len(updated) == 1:
+            output = updated[0]
+        elif len(updated) == 2:
+            output = ' and '.join(updated)
+        elif len(updated) > 2:
+            output = updated[0]
+            for index in range(1,len(updated)):
+                if index < len(updated) - 2:
+                    output = ', '.join([output, updated[index]])
+                elif index == len(updated) - 2:
+                    output = ', and '.join([output, updated[index]])
+                else:
+                    output = ''.join([output, updated[index]])
+        encrypted_update(db.users, {'_id': self._id}, { '$set': user })
+        if self._id == get_id_from_cookie():
+            _output = 'Your '
+        elif retrieved_user['username'][-1] == 's':
+            _output = '&rsquo;'.join([retrieved_user['username'], ''])
+        else:
+            _output = '&rsquo;'.join([retrieved_user['username'], 's'])
+        _output = f'{_output} {output} {"has" if len(updated) == 1 else "have"} been updated.'
+        flash(_output)
+        return jsonify({'success': _output}), 200
+    
+    def delete(self):
+        from main import db
+        retrieved_user = decrypt_find_one(db.users, { '_id': self._id })
+        if not retrieved_user:
+            return jsonify({ 'error': f'User does not exist.' }), 401
+        db.users.find_one_and_delete({'_id': self._id})
+        flash(f'User {retrieved_user["username"]} has been deleted.')
+        return jsonify({'success': f'User {retrieved_user["username"]} has been deleted.'}), 200