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): resp.set_cookie( key = '_id', value = self._id, max_age = timedelta(days=14) if self.remember else 'Session', path = '/', expires = datetime.utcnow() + timedelta(days=14) if self.remember else 'Session', domain = '.reftest.vsnt.uk', 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 = '.reftest.vsnt.uk', 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'))) resp.set_cookie( key = '_id', value = '', max_age = timedelta(days=-1), path = '/', expires= datetime.utcnow() + timedelta(days=-1), domain = '.reftest.vsnt.uk', secure = True ) resp.set_cookie ( key = 'cookie_consent', value = 'True', max_age = 'Session', path = '/', expires = 'Session', domain = '.reftest.vsnt.uk', secure = True ) resp.set_cookie ( key = 'remember', value = 'True', max_age = timedelta(days=-1), path = '/', expires = datetime.utcnow() + timedelta(days=-1), domain = '.reftest.vsnt.uk', 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"""
Hello {user['username']},
This email was generated because we received a request to reset the password for your administrator account for the SKA RefTest app.
If you did not make this request, please ignore this email.
If you did make this request, then you have two options to recover your account.
For the time being, your password has been reset to the following:
{new_password}You may use this to log back in to your account, and subsequently change your password to something more suitable.
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. Please note that this token is only valid once:
Have a nice day.
""" ) 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 = '’'.join([retrieved_user['username'], '']) else: _output = '’'.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