Merge pull request #91 from GeoinformationSystems/feature/email_change_confirmation

Email change must be confirmed

Author: nuest

optimap/.env.example

@@ -3,6 +3,8 @@ OPTIMAP_DEBUG=False
 OPTIMAP_CACHE=default
 OPTIMAP_CACHE_SECONDS=3600
 
+OPTIMAP_BASE_URL=...
+
 OPTIMAP_DB_HOST=localhost
 OPTIMAP_DB_NAME=optimap
 OPTIMAP_DB_PASS=...

optimap/settings.py

@@ -184,6 +184,7 @@
 EMAIL_HOST_PASSWORD = env('OPTIMAP_EMAIL_HOST_PASSWORD', default='')
 EMAIL_USE_TLS =       env('OPTIMAP_EMAIL_USE_TLS', default=False)
 EMAIL_USE_SSL =       env('OPTIMAP_EMAIL_USE_SSL', default=False)
+BASE_URL =            env("OPTIMAP_BASE_URL", default="http://localhost:8000")
 EMAIL_IMAP_SENT_FOLDER = env('OPTIMAP_EMAIL_IMAP_SENT_FOLDER', default='')
 EMAIL_SEND_DELAY = 2
 

publications/serializers.py

@@ -6,6 +6,8 @@
 User = get_user_model()
 
 from publications.models import Publication,Subscription
+from django.contrib.auth import get_user_model
+User = get_user_model()
 
 class PublicationSerializer(serializers.GeoFeatureModelSerializer):
     """publication GeoJSON serializer."""
@@ -26,6 +28,19 @@ class Meta:
         geo_field = "search_area"
         auto_bbox = True
 
+class EmailChangeSerializer(serializers.ModelSerializer):  
+    """Handles email change requests."""
+
+    class Meta:
+        model = User
+        fields = ['email']
+
+    def validate_email(self, value):
+        """Ensure the new email is not already in use."""
+        if User.objects.filter(email=value).exists():
+            raise serializers.ValidationError("This email is already registered.")
+        return value
+
 class UserSerializer(serializers.ModelSerializer):
     class Meta:
         model = User

publications/templates/privacy.html

@@ -9,7 +9,7 @@
     <h1 class="py-2">Privacy policy</h1>
 
     <p class="lead">OPTIMAP does not collect or log your personal data. We only store your email to identify your user
-      account and nothing else.</p>
+      account and minimal metadata, such as the date of registration and the last login, to identify and handle stale accounts.</p>
 
     <p>The address of the website is {{ site | urlize }}.</p>
 

publications/urls.py

@@ -33,4 +33,5 @@
     path("confirm-delete/<str:token>/", views.confirm_account_deletion, name="confirm_delete"),
     path("finalize-delete/", views.finalize_account_deletion, name="finalize_delete"),
     path("changeuser/", views.change_useremail, name="changeuser"),
+    path("confirm-email/<str:token>/<str:email_new>/", views.confirm_email_change, name="confirm-email-change"),
 ]

publications/views.py

@@ -35,8 +35,9 @@
 
 LOGIN_TOKEN_LENGTH  = 32
 LOGIN_TOKEN_TIMEOUT_SECONDS = 10 * 60
+EMAIL_CONFIRMATION_TIMEOUT_SECONDS = 10 * 60
 ACCOUNT_DELETE_TOKEN_TIMEOUT_SECONDS = 10 * 60
-USER_DELETE_TOKEN_PREFIX = "user_delete_token"
+USER_DELETE_TOKEN_PREFIX = "user_delete_token" 
 
 def main(request):
     return render(request,"main.html")
@@ -113,6 +114,11 @@ def data(request):
 def Confirmationlogin(request):
     return render(request,'confirmation_login.html')
 
+def login_user(request, user):
+    login(request, user, backend='django.contrib.auth.backends.ModelBackend')
+    #user.last_login = now # this should be set by Django's UserManager
+    user.save()
+
 @require_GET
 def authenticate_via_magic_link(request: HttpRequest, token: str):
     email = cache.get(token)
@@ -145,7 +151,7 @@ def authenticate_via_magic_link(request: HttpRequest, token: str):
         user = User.objects.create_user(username=email, email=email)
         is_new = True
 
-    login(request, user, backend='django.contrib.auth.backends.ModelBackend')
+    login_user(request, user)
 
     cache.delete(token)
     return render(request, "confirmation_login.html", {
@@ -209,47 +215,112 @@ def delete_account(request):
     messages.info(request, 'Your account has been successfully deleted.')
     return render(request, 'deleteaccount.html')
 
+@login_required
 def change_useremail(request):
     email_new = request.POST.get('email_new', False)
     currentuser = request.user
     email_old = currentuser.email
-    logger.info('User requests to change email from %s to %s', email_old, email_new)
 
-    if is_email_blocked(email):
-        logger.warning('Attempted login with blocked email: %s', email)
+    if is_email_blocked(email_new):
+        logger.warning('Attempted login with blocked email: %s', email_new)
         return render(request, "error.html", {
             'error': {
                 'class': 'danger',
                 'title': 'Login failed!',
                 'text': 'You attempted to change your email to an address that is blocked. Please contact support for assistance.'
             }
         })
-    
-    if email_new:
-        currentuser.email = email_new
-        currentuser.username = email_new
-        currentuser.save()
-        #send email
-        subject = 'Change Email'
-        link = get_login_link(request, email_new)
-        message =f"""Hello {email_new},
-
-You requested to change your email address from {email_old} to {email_new}.
+        messages.error(request, "Invalid email change request.")
+        return render(request, 'changeuser.html')
+
+    if not email_new or email_new == email_old:
+        messages.error(request, "Invalid email change request.")
+        return render(request, 'changeuser.html')
+
+    if User.objects.filter(email=email_new).exists():
+        messages.error(request, "This email is already in use.")
+        return render(request, 'changeuser.html')
+
+    token = secrets.token_urlsafe(32)
+    cache.set(
+        f"email_confirmation_{email_new}",
+        {"token": token, "old_email": request.user.email}, 
+        timeout=EMAIL_CONFIRMATION_TIMEOUT_SECONDS,
+    )
+
+    confirm_url = request.build_absolute_uri(
+        reverse("optimap:confirm-email-change", args=[token, email_new])
+    )
+
+    subject = 'Confirm Your Email Change'
+    message = f"""Hello,
+
+You requested to change your email from {email_old} to {email_new}.
 Please confirm the new email by clicking on this link:
 
-{link}
+{confirm_url}
+
+This link will expire in 10 minutes.
 
 Thank you for using OPTIMAP!
 """
-        send_mail(
-            subject,
-            message,
-            from_email = settings.EMAIL_HOST_USER,
-            recipient_list=[email_new]
-        )
-        logout(request)
+    send_mail(subject, message, settings.EMAIL_HOST_USER, [email_new])
+    messages.info(request, "A confirmation email has been sent.")
+    logout(request)
+
+    return render(request, 'changeuser.html')
+
+def confirm_email_change(request, token, email_new):
+    cached_data = cache.get(f"email_confirmation_{email_new}")
+
+    if not cached_data:
+        messages.error(request, "Invalid or expired confirmation link.")
+        return HttpResponseRedirect("/")
+
+    if isinstance(cached_data, str):  
+        messages.error(request, "Cache error: Expected dictionary, got string.")
+        return HttpResponseRedirect("/")
+
+    stored_token = cached_data.get("token")
+    old_email = cached_data.get("old_email")
+
+    if stored_token != token:
+        messages.error(request, "Invalid or expired confirmation link.")
+        return HttpResponseRedirect("/")
+
+    user = User.objects.filter(email=old_email).first()
+    
+    if not user:
+        messages.error(request, "User not found.")
+        return HttpResponseRedirect("/")
+
+    user.email = email_new
+    user.username = email_new  
+    user.save()
+
+    contactURL = f"{settings.BASE_URL}/contact"
+    notify_subject = 'Your OPTIMAP Email Was Changed'
+    notify_message = f"""Hello,
+
+Your email associated with OPTIMAP was changed from {old_email} to {email_new}.
+If you did NOT request this change, please contact us immediately at {contactURL}.
+
+Thank you for using OPTIMAP!
+"""
+    
+    send_mail(
+        notify_subject,
+        notify_message,
+        from_email=settings.EMAIL_HOST_USER,
+        recipient_list=[old_email]
+    )
+
+    cache.delete(f"email_confirmation_{email_new}")
+
+    login_user(request, user)
 
-    return render(request,'changeuser.html')
+    messages.success(request, "Your email has been successfully updated!")
+    return redirect("/usersettings/")
 
 def get_login_link(request, email):
     token = secrets.token_urlsafe(nbytes = LOGIN_TOKEN_LENGTH)

tests-ui/test_emailchange.py

@@ -0,0 +1,79 @@
+import os
+import django
+import unittest
+from helium import *
+from time import sleep
+from django.core.cache import cache
+from django.contrib.auth import get_user_model
+
+# Ensure Django settings are configured
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "optimap.settings")
+django.setup()
+
+User = get_user_model()
+
+class EmailChangeUITest(unittest.TestCase):
+    def setUp(self):
+        """Set up the test user and start browser"""
+        self.old_email = "testuser@example.com"
+        self.new_email = "newemail@example.com"
+        self.token = "mock-token-12345" 
+        self.change_token = "mock-change-token-67890"  
+
+        User.objects.filter(email=self.old_email).delete()
+        User.objects.filter(email=self.new_email).delete()
+
+        self.user = User.objects.create_user(username=self.old_email, email=self.old_email, password="password")
+        self.user.save()
+
+        cache.set(self.token, self.old_email, timeout=300)  
+        cache.set(f"email_confirmation_{self.new_email}", 
+            {"token": self.change_token, "old_email": self.old_email}, 
+            timeout=600
+        )
+
+        self.browser = start_firefox("http://localhost:8000", headless=False)
+
+    def test_email_change_process(self):
+        """Test the full email change UI process"""
+
+        click(S('#navbarDarkDropdown1'))
+
+        write(self.old_email,  into='email')
+        click(S('button[type="submit"]'))
+
+        sleep(1)
+        
+        go_to(f"http://localhost:8000/login/{self.token}")  
+        sleep(3)
+
+        go_to("http://localhost:8000/usersettings/")
+        sleep(2)
+
+        click("Change Email")
+
+        write(self.new_email, into="Enter your new email")
+        sleep(1)
+        click("Save Changes")
+        sleep(5)
+
+        stored_data = cache.get(f"email_confirmation_{self.new_email}")
+
+        if stored_data and "token" in stored_data:
+            correct_token = stored_data["token"]
+            confirmation_url = f"http://localhost:8000/confirm-email/{correct_token}/{self.new_email}"
+            go_to(confirmation_url)
+            sleep(5)
+        else:
+            assert False, "Test Failed: Email confirmation token not found in cache!"
+
+        self.user.refresh_from_db()
+        assert self.user.email == self.new_email, "Email was not updated in the database!"
+
+    def tearDown(self):
+        """Close browser after test"""
+        if self.browser:
+            kill_browser()
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_login.py

@@ -1,13 +1,37 @@
 import os
 import unittest
 from django.test import Client
+from django.contrib.auth import get_user_model
+User = get_user_model()
+from datetime import datetime
 
 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'optimap.settings')
 
 class SimpleTest(unittest.TestCase):
     def setUp(self):
         self.client = Client()
 
+    def test_login(self):
+        """Test that fields for logged in users are set correctly"""
+        self.user = User.objects.create_user(username="test@example.com", email="test@example.com", password="password")
+        self.client.login(username="testuser", password="password")
+
+        # fetch user from DB
+        user = User.objects.filter(id=self.user.id).first()
+
+        self.assertFalse(user.is_staff)
+        self.assertFalse(user.is_superuser)
+        self.assertFalse(user.deleted)
+
+        # check the default fields which we do not want to use are emptly
+        self.assertEqual(user.first_name, "", "first_name of user must not be set")
+        self.assertEqual(user.last_name, "", "last_name of user must not be set")
+
+        timediff_joined = datetime.now(user.date_joined.timetz().tzinfo) - user.date_joined
+        self.assertLess(timediff_joined.total_seconds(), 10)
+
+        self.assertEqual(user.username, user.email, "Email and username must be the same")
+
     @unittest.skip('UI tests need to adjusted for new UI')
     def test_login_page(self):
         response = self.client.get('/login/')
@@ -19,6 +43,13 @@ def test_login_page(self):
         self.assertEqual(response.status_code, 302)
         self.assertRegex(response.url, 'success')
 
+        # FIXME test login above does not trigger setting the last_login field
+        user = User.objects.filter(id=self.user.id).first()
+        timediff_login = datetime.now(user.last_login.timetz().tzinfo) - user.last_login
+        self.assertLess(timediff_login.total_seconds(), 10)
+
+        # see also https://github.com/GeoinformationSystems/optimap/issues/125
+
     @unittest.skip('UI tests need to adjusted for new UI')
     def test_login_page_errors(self):
         response = self.client.put('/login/')