Merge pull request #44 from diverso-lab/develop

Develop

app/modules/auth/models.py

@@ -15,6 +15,7 @@ class User(db.Model, UserMixin):
 
     data_sets = db.relationship('DataSet', backref='user', lazy=True)
     profile = db.relationship('UserProfile', backref='user', uselist=False)
+    active = db.Column(db.Boolean, default=True, nullable=False)
 
     def __init__(self, **kwargs):
         super(User, self).__init__(**kwargs)

app/modules/auth/repositories.py

@@ -17,5 +17,5 @@ def create(self, commit: bool = True, **kwargs):
             self.session.flush()
         return instance
 
-    def get_by_email(self, email: str):
-        return self.model.query.filter_by(email=email).first()
+    def get_by_email(self, email: str, active: bool = True):
+        return self.model.query.filter_by(email=email, active=active).first()

app/modules/auth/routes.py

@@ -32,16 +32,29 @@ def show_signup_form():
 
         try:
             user = authentication_service.create_with_profile(**form.data)
+            authentication_service.send_confirmation_email(user.email)
+            flash("Please confirm your email", "info")
         except Exception as exc:
             return render_template("auth/signup_form.html", form=form, error=f'Error creating user: {exc}')
 
-        # Log user
-        login_user(user, remember=True)
-        return redirect(url_for('public.index'))
+        return redirect(url_for("public.index"))
 
     return render_template("auth/signup_form.html", form=form)
 
 
+@auth_bp.route("/confirm_user/<token>", methods=["GET"])
+def confirm_user(token):
+    try:
+        user = authentication_service.confirm_user_with_token(token)
+    except Exception as exc:
+        flash(exc.args[0], "danger")
+        return redirect(url_for("auth.show_signup_form"))
+
+    # Log user
+    login_user(user, remember=True)
+    return redirect(url_for("public.index"))
+
+
 @auth_bp.route('/login', methods=['GET', 'POST'])
 @guest_required
 def login():

app/modules/auth/services.py

@@ -1,7 +1,11 @@
 import os
+
+from flask import current_app, url_for
 from flask_login import login_user
 from flask_login import current_user
+from itsdangerous import BadTimeSignature, SignatureExpired, URLSafeTimedSerializer
 
+from app import mail_service
 from app.modules.auth.models import User
 from app.modules.auth.repositories import UserRepository
 from app.modules.profile.models import UserProfile
@@ -11,10 +15,16 @@
 
 
 class AuthenticationService(BaseService):
+    SALT = "user-confirm"
+    MAX_AGE = 3600
+
     def __init__(self):
         super().__init__(UserRepository())
         self.user_profile_repository = UserProfileRepository()
 
+    def get_serializer(self):
+        return URLSafeTimedSerializer(current_app.config['SECRET_KEY'])
+
     def login(self, email, password, remember=True):
         user = self.repository.get_by_email(email)
         if user is not None and user.check_password(password):
@@ -43,7 +53,8 @@ def create_with_profile(self, **kwargs):
 
             user_data = {
                 "email": email,
-                "password": password
+                "password": password,
+                "active": False,
             }
 
             profile_data = {
@@ -60,6 +71,33 @@ def create_with_profile(self, **kwargs):
             raise exc
         return user
 
+    def get_token_from_email(self, email):
+        s = self.get_serializer()
+        return s.dumps(email, salt=self.SALT)
+
+    def send_confirmation_email(self, user_email):
+        token = self.get_token_from_email(user_email)
+        url = url_for("auth.confirm_user", token=token)
+        mail_service.send_email(
+            "Please confirm your email",
+            recipients=[user_email],
+            body=f"<a href='{url}'>Please confirm your email</a>",
+        )
+
+    def confirm_user_with_token(self, token):
+        s = self.get_serializer()
+        try:
+            email = s.loads(token, salt=self.SALT, max_age=self.MAX_AGE)
+        except SignatureExpired:
+            raise Exception("The confirmation link has expired.")
+        except BadTimeSignature:
+            raise Exception("The confirmation link has been tampered with.")
+
+        user = self.repository.get_by_email(email, active=False)
+        user.active = True
+        self.repository.session.commit()
+        return user
+
     def update_profile(self, user_profile_id, form):
         if form.validate():
             updated_instance = self.update(user_profile_id, **form.data)

app/modules/auth/tests/test_unit.py

@@ -1,7 +1,10 @@
+import time
 from unittest.mock import patch
+
 import pytest
 from flask import url_for
 
+from app import mail_service
 from app.modules.auth.services import AuthenticationService
 from app.modules.auth.repositories import UserRepository
 from app.modules.profile.repositories import UserProfileRepository
@@ -122,3 +125,74 @@ def test_service_create_with_profile_fail_no_password(clean_database):
 
     assert UserRepository().count() == 0
     assert UserProfileRepository().count() == 0
+
+
+@patch('app.modules.captcha.services.CaptchaService.validate_captcha', return_value=True)
+def test_signup_send_confirmation_email(mock_captcha, test_client, clean_database):
+    data = {
+        "name": "Test",
+        "surname": "Foo",
+        "email": "[email protected]",
+        "password": "test1234",
+        "captcha": "dummy_captcha"
+    }
+
+    with mail_service.mail.record_messages() as outbox:
+        test_client.post("/signup", data=data, follow_redirects=True)
+        assert len(outbox) == 1
+
+
+def test_create_with_profile_create_inactive_user(test_client, clean_database):
+    data = {
+        "name": "Test",
+        "surname": "Foo",
+        "email": "[email protected]",
+        "password": "test1234"
+    }
+    user = AuthenticationService().create_with_profile(**data)
+    assert UserRepository().count() == 1
+    assert UserProfileRepository().count() == 1
+    assert user.active is False
+
+
+def test_confirm_user_token_expired(test_client):
+    email = "[email protected]"
+
+    with patch("time.time", return_value=time.time() - (AuthenticationService.MAX_AGE + 1)):
+        token = AuthenticationService().get_token_from_email(email)
+
+    url = url_for('auth.confirm_user', token=token, _external=False)
+    response = test_client.get(url, follow_redirects=True)
+    assert response.request.path == url_for("auth.show_signup_form", _external=False)
+
+
+def test_confirm_user_token_tempered(test_client):
+    email = "[email protected]"
+
+    AuthenticationService.SALT = "bad_salt"
+    token = AuthenticationService().get_token_from_email(email)
+
+    AuthenticationService.SALT = "user-confirm"
+    url = url_for('auth.confirm_user', token=token, _external=False)
+    response = test_client.get(url, follow_redirects=True)
+    assert response.request.path == url_for("auth.show_signup_form", _external=False)
+
+
+def test_confirm_user_active_user(test_client):
+    data = {
+        "name": "Test",
+        "surname": "Foo",
+        "email": "[email protected]",
+        "password": "test1234"
+    }
+    user = AuthenticationService().create_with_profile(**data)
+    assert user.active is False
+
+    token = AuthenticationService().get_token_from_email(user.email)
+
+    url = url_for('auth.confirm_user', token=token, _external=False)
+    response = test_client.get(url, follow_redirects=True)
+    assert response.request.path == url_for("public.index", _external=False)
+
+    user = UserRepository().get_by_email(user.email)
+    assert user.active is True