feat(preferences): added personal email verification

Author: sroshc

docs/source/reference_index/apps/preferences.rst

@@ -9,6 +9,8 @@ preferences
 
    fields
    forms
+   models
+   tasks
    tests
    urls
    views

intranet/apps/preferences/forms.py

@@ -90,7 +90,7 @@ def flag(label, default):
         if user.emails.all().count() == 0:
             label = "You can set a primary email after adding emails below."
         self.fields["primary_email"] = forms.ModelChoiceField(
-            queryset=Email.objects.filter(user=user), required=False, label=label, disabled=(user.emails.all().count() == 0)
+            queryset=Email.objects.filter(user=user, verified=True), required=False, label=label, disabled=(user.emails.all().count() == 0)
         )
 
 

intranet/apps/preferences/migrations/0001_initial.py

@@ -0,0 +1,29 @@
+# Generated by Django 3.2.25 on 2025-05-07 01:56
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+import uuid
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ('users', '0043_email_verified'),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='UnverifiedEmail',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('date_created', models.DateTimeField(auto_now_add=True)),
+                ('verification_token', models.UUIDField(default=uuid.uuid4, editable=False)),
+                ('email', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='users.email')),
+                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
+            ],
+        ),
+    ]

intranet/apps/preferences/models.py

@@ -0,0 +1,19 @@
+import uuid
+from datetime import timedelta
+
+from django.conf import settings
+from django.db import models
+from django.utils.timezone import now
+
+from ..users.models import Email
+
+
+class UnverifiedEmail(models.Model):
+    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
+    email = models.ForeignKey(Email, on_delete=models.CASCADE)
+    date_created = models.DateTimeField(auto_now_add=True)
+    verification_token = models.UUIDField(default=uuid.uuid4, editable=False)
+
+    # Email link is expired if it's older than specified expire time and should be deleted.
+    def is_expired(self):
+        return now() - self.date_created >= timedelta(hours=settings.UNVERIFIED_EMAIL_EXPIRE_HOURS)

intranet/apps/preferences/tasks.py

@@ -0,0 +1,22 @@
+from datetime import timedelta
+
+from celery import shared_task
+from celery.utils.log import get_task_logger
+from django.conf import settings
+from django.utils.timezone import now
+
+from .models import UnverifiedEmail
+
+logger = get_task_logger(__name__)
+
+
+@shared_task
+def delete_expired_emails():
+    # Unverified email links should be deleted after specified timeout.
+    cutoff = now() - timedelta(hours=settings.UNVERIFIED_EMAIL_EXPIRE_HOURS)
+    expired_emails = UnverifiedEmail.objects.filter(date_created__lt=cutoff)
+
+    emails_deleted = expired_emails.count()
+    expired_emails.delete()
+
+    logger.info(f"Deleted {emails_deleted} expired email links.")

intranet/apps/preferences/urls.py

@@ -2,4 +2,8 @@
 
 from . import views
 
-urlpatterns = [re_path(r"^$", views.preferences_view, name="preferences"), re_path(r"^/privacy$", views.privacy_options_view, name="privacy_options")]
+urlpatterns = [
+    re_path(r"^$", views.preferences_view, name="preferences"),
+    re_path(r"^/privacy$", views.privacy_options_view, name="privacy_options"),
+    re_path(r"^/verify_email/(?P<email_uuid>[0-9a-fA-F-]{36})$", views.verify_email_view, name="verify_email"),  # The path only accepts valid UUIDs.
+]

intranet/apps/preferences/views.py

@@ -6,11 +6,14 @@
 from django.contrib.auth import get_user_model
 from django.contrib.auth.decorators import login_required
 from django.shortcuts import redirect, render
+from django.urls import reverse
 
 from ..auth.decorators import eighth_admin_required
 from ..bus.models import Route
+from ..notifications.tasks import email_send_task
 from ..users.models import Email
 from .forms import BusRouteForm, DarkModeForm, EmailFormset, NotificationOptionsForm, PreferredPictureForm, PrivacyOptionsForm
+from .models import UnverifiedEmail
 
 # from .forms import (BusRouteForm, DarkModeForm, EmailFormset, NotificationOptionsForm, PhoneFormset, PreferredPictureForm, PrivacyOptionsForm,
 #                    WebsiteFormset)
@@ -52,6 +55,20 @@ def get_personal_info(user):
     return personal_info, num_fields
 
 
+def send_verification_email(request, user, email):
+    email_link = UnverifiedEmail(user=user, email=email)
+    email_link.save()
+
+    verification_link = request.build_absolute_uri(reverse("verify_email", args=[email_link.verification_token]))
+    base_url = request.build_absolute_uri(reverse("index"))
+    data = {"verification_link": verification_link, "base_url": base_url}
+    headers = {"From": "Ion <[email protected]>"}
+
+    email_send_task.delay(
+        "preferences/email/verify_email.txt", "preferences/email/verify_email.html", data, "Email Verification", [email.address], headers
+    )
+
+
 def save_personal_info(request, user):
     # phone_formset = PhoneFormset(request.POST, instance=user, prefix="pf")
     phone_formset = None
@@ -68,7 +85,24 @@ def save_personal_info(request, user):
     # else:
     #     errors.append("Could not set phone numbers.")
     if email_formset.is_valid():
-        email_formset.save()
+        new_emails = email_formset.save(commit=False)
+
+        # Manually handle saving the formset so we can flag new emails as unverified.
+        for email in new_emails:
+            if email._state.adding:
+                email.verified = False
+            email.save()
+            send_verification_email(request, user, email)
+            messages.success(
+                request,
+                f"Successfully sent verification email to '{email.address}'. The link will expire in {settings.UNVERIFIED_EMAIL_EXPIRE_HOURS} hours.",
+            )
+
+        for deleted_email in email_formset.deleted_objects:
+            try:
+                deleted_email.delete()
+            except deleted_email.DoesNotExist:
+                pass
     else:
         for error in email_formset.errors:
             if isinstance(error.get("address"), list):
@@ -207,6 +241,13 @@ def save_notification_options(request, user):
                 if field in notification_options and notification_options[field] == fields[field]:
                     pass
                 else:
+                    # Users should only be able to set verified emails as their primary email.
+                    if field == "primary_email" and fields[field] is not None:
+                        email = Email.objects.filter(user=user, address=fields[field]).first()
+                        if not email.verified:
+                            messages.error(request, "You may only set verified emails as your primary email.")
+                            continue
+
                     setattr(user, field, fields[field])
                     user.save()
                     try:
@@ -290,6 +331,28 @@ def save_dark_mode_settings(request, user):
     return dark_mode_form
 
 
+@login_required
+def verify_email_view(request, email_uuid):
+    """ "Verify the UUID associated with the unverified email."""
+    user = request.user
+
+    unverified_email = UnverifiedEmail.objects.filter(verification_token=email_uuid, user=user).first()
+
+    # If the uuid isn't found or link is expired, return a error message.
+    if unverified_email is None or unverified_email.is_expired():
+        messages.error(request, "Could not verify email, the link was either expired or invalid.")
+        return redirect("preferences")
+
+    verified_mail = unverified_email.email
+    verified_mail.verified = True
+
+    verified_mail.save()
+    unverified_email.delete()
+
+    messages.success(request, f"Successfully verified '{verified_mail.address}'. You can add it as a primary email now.")
+    return redirect("preferences")
+
+
 @login_required
 def preferences_view(request):
     """View and process updates to the preferences page."""
@@ -331,6 +394,13 @@ def preferences_view(request):
         email_formset = EmailFormset(instance=user, prefix="ef")
         # website_formset = WebsiteFormset(instance=user, prefix="wf")
 
+        # Flag emails as verified or unverified for templating.
+        for form in email_formset:
+            if form.instance.pk:
+                form.verified = form.instance.verified
+            else:
+                form.verified = None
+
         if user.is_student:
             preferred_pic = get_preferred_pic(user)
             bus_route = get_bus_route(user)

intranet/apps/users/migrations/0043_email_verified.py

@@ -0,0 +1,18 @@
+# Generated by Django 3.2.25 on 2025-05-07 01:56
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('users', '0042_user_seen_april_fools'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='email',
+            name='verified',
+            field=models.BooleanField(default=True),
+        ),
+    ]

intranet/apps/users/models.py

@@ -1293,6 +1293,7 @@ class Email(models.Model):
 
     address = models.EmailField()
     user = models.ForeignKey(settings.AUTH_USER_MODEL, related_name="emails", on_delete=models.CASCADE)
+    verified = models.BooleanField(default=True)
 
     def __str__(self):
         return self.address

intranet/settings/__init__.py

@@ -919,6 +919,9 @@ def get_log(name):  # pylint: disable=redefined-outer-name; 'name' is used as th
 # Age (in days) of a lost and found entry until it is hidden
 LOSTFOUND_EXPIRATION = 180
 
+# How many hours till a verification email expires.
+UNVERIFIED_EMAIL_EXPIRE_HOURS = 12
+
 # Substrings of paths to not log in the Ion access logs
 NONLOGGABLE_PATH_BEGINNINGS = ["/static"]
 NONLOGGABLE_PATH_ENDINGS = [".png", ".jpg", ".jpeg", ".gif", ".css", ".js", ".ico", "jsi18n/"]
@@ -960,6 +963,11 @@ def get_log(name):  # pylint: disable=redefined-outer-name; 'name' is used as th
         "schedule": celery.schedules.crontab(day_of_month=1, hour=1),
         "args": (),
     },
+    "delete-expired-email-links": {
+        "task": "intranet.apps.preferences.tasks.delete_expired_emails",
+        "schedule": celery.schedules.crontab(hour=0, minute=0),
+        "args": (),
+    }
 }
 
 MAINTENANCE_MODE = False