NHS Email Authentication

app/auth/email.py

@@ -1,6 +1,11 @@
 from flask import render_template, current_app
 
 from app.email import send_email
+import jwt
+import time
+import uuid
+from app.models import User
+from app import db
 
 
 def send_password_reset_email(user):
@@ -9,9 +14,48 @@ def send_password_reset_email(user):
     """
     token = user.get_reset_password_token()
     send_email('[Hazen] Reset Your Password',
-               sender=current_app.config['ADMINS'][0],
+               sender=current_app.config['ADMINS'],
                recipients=[user.email],
                text_body=render_template('email/reset_password.txt',
                                          user=user, token=token),
                html_body=render_template('email/reset_password.html',
                                          user=user, token=token))
+
+
+def validate_nhs_email(user):
+    token = jwt.encode({'email_auth': str(user.id), 'exp': time.time() + 600},
+                       current_app.config['SECRET_KEY'],
+                       algorithm='HS256')
+
+    current_app.logger.info(f"Generated token for user {user.id}: {token}")
+    send_email('[Hazen] Verify your Email',
+               sender=current_app.config['ADMINS'],
+               recipients=[user.email],
+               text_body=render_template('email/authenticate_email.txt', user=user, token=token),
+               html_body=render_template('email/authenticate_email.html', user=user, token=token))
+
+    return token
+
+
+def verify_email_auth_token(token):
+    try:
+        data = jwt.decode(token, current_app.config['SECRET_KEY'], algorithms=['HS256'])
+        user_id = data.get('email_auth')
+        current_app.logger.info(f"user ID found: {user_id}")
+
+        if not user_id:
+            return None
+    except jwt.ExpiredSignatureError:
+        return None  #token expired
+    except jwt.InvalidTokenError:
+        return None  #invalid token
+    except Exception as e:
+        current_app.logger.error(f"Unexpected error: {str(e)}")
+        return None
+    user_id_formatted = uuid.UUID(user_id)
+    user = User.query.get(user_id_formatted)
+    if user:
+        user.email_authenticated = True
+        db.session.commit()
+        return user
+    return None

app/auth/forms.py

@@ -11,7 +11,7 @@ class RegistrationForm(FlaskForm):
     institution = StringField('Institution', validators=[DataRequired()])
 
     username = StringField('Username', validators=[DataRequired()])
-    email = StringField('Email', validators=[DataRequired(), Email()])
+    email = StringField('Email Address (@gstt.nhs.uk or @nhs.net)', validators=[DataRequired(), Email()])
     password = PasswordField('Password', validators=[DataRequired()])
     password2 = PasswordField(
         'Repeat Password', validators=[DataRequired(), EqualTo('password')])

app/auth/routes.py

@@ -3,7 +3,7 @@
 Utilises the functions in the app.auth.forms module.
 """
 
-from flask import render_template, redirect, url_for, flash, request
+from flask import render_template, redirect, url_for, flash, request, current_app, jsonify, Flask
 from werkzeug.urls import url_parse
 from flask_login import login_user, login_required, logout_user, current_user
 
@@ -12,22 +12,57 @@
 from app.auth.forms import RegistrationForm, LoginForm, EditProfileForm, ResetPasswordRequestForm, ResetPasswordForm
 from app.models import User
 from app.auth.email import send_password_reset_email
+from app.auth.email import validate_nhs_email
+from app.auth.email import verify_email_auth_token
+import re
+
 
 
 @bp.route('/register', methods=['GET', 'POST'])
 def register():
     if current_user.is_authenticated:
         return redirect(url_for('main.index'))
+
     form = RegistrationForm()
     if form.validate_on_submit():
-        user = User(firstname=form.firstname.data, lastname=form.lastname.data, institution=form.institution.data, username=form.username.data, email=form.email.data)
+        #create new user
+        user = User(
+            firstname=form.firstname.data,
+            lastname=form.lastname.data,
+            institution=form.institution.data,
+            username=form.username.data,
+            email=form.email.data,
+            email_authenticated=False
+        )
+        if not re.match(r'.+@(nhs\.net|gstt\.nhs\.uk)$', user.email):
+            flash('Invalid email. Please use @gstt.nhs.uk or @nhs.net.', 'danger')
+            return redirect(url_for('auth.register'))
+
         user.set_password(form.password.data)
         db.session.add(user)
         db.session.commit()
-        flash('Congratulations, you are now a registered user!', 'success')
+
+        #send email verification
+        validate_nhs_email(user)
+
+        flash('Please check your email for the verification link. The link will expire in 10 minutes.', 'info')
         return redirect(url_for('auth.login'))
+
     return render_template('register.html', title='Register', form=form)
 
+@bp.route('/confirm_email/<token>', methods=['GET', 'POST'])
+def confirm_email(token):
+    current_app.logger.info(f"Received token: {token}")
+    user = verify_email_auth_token(token)
+    if not user:
+        flash('Invalid or expired token.', 'danger')
+        return redirect(url_for('auth.register'))
+
+    flash('Email confirmed successfully! You can now log in.', 'success')
+    return redirect(url_for('auth.login'))
+
+
+
 
 @bp.route('/login', methods=['GET', 'POST'])
 def login():
@@ -42,6 +77,10 @@ def login():
             flash('Invalid username or password', 'danger')
             return redirect(url_for('auth.login'))
 
+        if not user.email_authenticated:
+            flash('Please verify your email address first.', 'warning')
+            return redirect(url_for('auth.login'))
+
         login_user(user, remember=form.remember_me.data)
         next_page = request.args.get('next')
 

app/auth/templates/email/authenticate_email.html

@@ -0,0 +1,18 @@
+<p>Dear {{ user.username }},</p>
+<p>
+    To authenticate your email
+    <a href="{{ url_for('auth.confirm_email', token=token, _external=True) }}">
+        click here
+    </a>.
+</p>
+<!--
+<p>Alternatively, you can paste the following link in your browser's address bar:</p>
+<p>{{ url_for('auth.confirm_email', token=token, _external=True) }}</p>
+<p></p>
+-->
+<p>You will be redirected to the login page, please log in using your credentials.</p>
+<p></p>
+<p>The link will expire in 10 minutes.</p>
+<p></p>
+<p>Sincerely,</p>
+<p>The Hazen Team</p>

app/auth/templates/email/authenticate_email.txt

@@ -0,0 +1,13 @@
+Dear {{ user.username }},
+
+To authenticate your email, paste the following link into your browser's address bar:
+
+{{ url_for('auth.confirm_email', token=token, _external=True) }}
+
+You will be redirected to the login page, please log in using your credentials.
+
+The link will expire in 10 minutes.
+
+Sincerely,
+
+The Hazen Team

app/auth/templates/email/reset_password.txt

@@ -1,6 +1,6 @@
 Dear {{ user.username }},
 
-To reset your password click on the following link:
+To reset your password, paste the following link in your browser's address bar:
 
 {{ url_for('auth.reset_password', token=token, _external=True) }}
 

app/models.py

@@ -35,6 +35,7 @@ def __init__(self, **kwargs):
     email = db.Column(db.String(320), index=True, unique=True)  # why do we need index?
     password_hash = db.Column(db.String(128))
     last_seen = db.Column(db.DateTime, default=datetime.utcnow)
+    email_authenticated = db.Column(db.Boolean, default=False, nullable=False)
 
     # One-to-many bidirectional relationship
     # images = db.relationship('Image', back_populates='user')
@@ -58,7 +59,7 @@ def __repr__(self):
     def get_reset_password_token(self, expires_in=600):
         return jwt.encode(
             {'reset_password': str(self.id), 'exp': time() + expires_in},
-            str(current_app.config['SECRET_KEY']), algorithm='HS256').decode('utf-8')
+            str(current_app.config['SECRET_KEY']), algorithm='HS256')
 
     @staticmethod
     def verify_reset_password_token(token):

assets/dbdiagram_hazen_schema.txt

@@ -66,6 +66,7 @@ Table "user" {
   "email" "character varying(320)"
   "password_hash" "character varying(128)"
   "last_seen" timestamp
+  "email_authenticated" boolean
 
 Indexes {
   email [type: btree, unique, name: "ix_user_email"]

assets/hazen_db_schema.sql

@@ -131,6 +131,7 @@ CREATE TABLE public."user" (
     email character varying(320),
     password_hash character varying(128),
     last_seen timestamp without time zone
+    email_authenticated boolean
 );
 
 

config.py

@@ -36,12 +36,15 @@ class Config:
 
     SQLALCHEMY_TRACK_MODIFICATIONS = False
 
-    MAIL_SERVER = os.environ.get('MAIL_SERVER')
-    MAIL_PORT = int(os.environ.get('MAIL_PORT') or 25)
-    MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS') is not None
-    MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
-    MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
-    ADMINS = [os.environ.get('ADMIN_EMAIL')] or ['[email protected]']
+    #MAIL_SERVER = os.environ.get('MAIL_SERVER')
+    #MAIL_PORT = int(os.environ.get('MAIL_PORT') or 25)
+    #MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS') is not None
+    MAIL_SERVER = 'smtp.gmail.com'
+    MAIL_PORT = 587
+    MAIL_USE_TLS = True
+    MAIL_USERNAME = '[email protected]'
+    MAIL_PASSWORD = 'zleg bwbp fbvz osvt'
+    ADMINS = '[email protected]'
 
     ACQUISITIONS_PER_PAGE = 9
 

migrations/alembic.ini

@@ -8,6 +8,10 @@
 # the 'revision' command, regardless of autogenerate
 # revision_environment = false
 
+# Path to migration scripts
+script_location = migrations
+
+sqlalchemy.url = postgresql://hazen:password123@localhost:5432/hazen
 
 # Logging configuration
 [loggers]

migrations/versions/1544418d922f_.py

@@ -9,6 +9,7 @@
 import sqlalchemy as sa
 import sqlalchemy_utils
 from sqlalchemy.dialects import postgresql
+from sqlalchemy.engine import reflection
 
 # revision identifiers, used by Alembic.
 revision = '1544418d922f'
@@ -19,6 +20,7 @@
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
+
     op.create_table('process_task',
     sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
     sa.Column('created_at', sqlalchemy_utils.types.arrow.ArrowType(), nullable=False),
@@ -27,16 +29,19 @@ def upgrade():
     sa.Column('docstring', sa.String(), nullable=True),
     sa.PrimaryKeyConstraint('id')
     )
-    op.create_table('user',
-    sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
-    sa.Column('created_at', sqlalchemy_utils.types.arrow.ArrowType(), nullable=False),
-    sa.Column('username', sa.String(length=64), nullable=True),
-    sa.Column('email', sa.String(length=120), nullable=True),
-    sa.Column('password_hash', sa.String(length=128), nullable=True),
-    sa.Column('about_me', sa.String(length=140), nullable=True),
-    sa.Column('last_seen', sa.DateTime(), nullable=True),
-    sa.PrimaryKeyConstraint('id')
-    )
+    bind = op.get_bind()
+    inspector = reflection.Inspector.from_engine(bind)
+    if 'user' not in inspector.get_table_names():
+        op.create_table('user',
+        sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
+        sa.Column('created_at', sqlalchemy_utils.types.arrow.ArrowType(), nullable=False),
+        sa.Column('username', sa.String(length=64), nullable=True),
+        sa.Column('email', sa.String(length=120), nullable=True),
+        sa.Column('password_hash', sa.String(length=128), nullable=True),
+        sa.Column('about_me', sa.String(length=140), nullable=True),
+        sa.Column('last_seen', sa.DateTime(), nullable=True),
+        sa.PrimaryKeyConstraint('id')
+        )
     op.create_index(op.f('ix_user_email'), 'user', ['email'], unique=True)
     op.create_index(op.f('ix_user_username'), 'user', ['username'], unique=True)
     op.create_table('acquisition',

migrations/versions/2025_02_26_add_email_authenticated_column.py

@@ -0,0 +1,29 @@
+"""
+
+Revision ID: 20250226a
+Revises: 1544418d922f
+Create Date: 2025-02-26
+
+"""
+from alembic import op
+import sqlalchemy as sa
+import sqlalchemy_utils
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = '20250226a'
+down_revision = '1544418d922f'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column('user', sa.Column('email_authenticated', sa.Boolean(), nullable=False, server_default=sa.text('false')))
+
+
+
+
+def downgrade():
+    # Remove 'email_authenticated' column from 'user' table
+    op.drop_column('user', 'email_authenticated')
+

migrations/versions/84124366c36a_.py

@@ -19,7 +19,9 @@
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.drop_table('followers')
+    op.execute('DROP TABLE IF EXISTS followers')
+
+    #op.drop_table('followers')
     # ### end Alembic commands ###