Fix CAS login errors caused by duplicate user accounts (#390)

Author: C4ptainCrunch

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 This page tries to contain all use facing changes made on DocHub.
 
+# Unreleased
+
+ * Fix login errors when ULB changes your email or netid
+ * Log CAS authentication failures in the admin for easier debugging
+
 # 2026.5.1
 
 Moderation:

users/admin.py

@@ -1,6 +1,6 @@
 from django.contrib import admin
 
-from .models import User
+from .models import CasFailure, User
 
 
 @admin.register(User)
@@ -64,3 +64,18 @@ class UserAdmin(admin.ModelAdmin):
             },
         ),
     )
+
+
+@admin.register(CasFailure)
+class CasFailureAdmin(admin.ModelAdmin):
+    list_display = ("code", "ticket", "ip_address", "created")
+    list_filter = ("code",)
+    search_fields = ("ticket", "details", "ip_address")
+    date_hierarchy = "created"
+    readonly_fields = ("code", "details", "ticket", "ip_address", "created")
+
+    def has_add_permission(self, request):
+        return False
+
+    def has_change_permission(self, request, obj=None):
+        return False

users/authBackend.py

@@ -2,13 +2,12 @@
 import xml.etree.ElementTree as ET
 
 from django.conf import settings
-from django.db.models import Q
 from django.urls import reverse
 
 import requests
 from furl import furl
 
-from users.models import User
+from users.models import CasFailure, User
 
 logger = logging.getLogger(__name__)
 
@@ -45,19 +44,42 @@ def authenticate(self, request, ticket=None):
         user_dict = self._parse_response(resp.text)
 
         # Get or create a user from the parsed user_dict
+        # Prefer matching by netid (primary CAS identifier), fall back to email
         try:
-            user = User.objects.get(
-                Q(netid=user_dict["netid"]) | Q(email=user_dict["email"])
-            )
+            user = User.objects.get(netid=user_dict["netid"])
         except User.DoesNotExist:
-            user = User.objects.create_user(
-                netid=user_dict["netid"],
-                email=user_dict["email"],
-                first_name=user_dict["first_name"],
-                last_name=user_dict["last_name"],
-                register_method=self.LOGIN_METHOD,
-            )
+            try:
+                user = User.objects.get(email=user_dict["email"])
+            except User.DoesNotExist:
+                user = User.objects.create_user(
+                    netid=user_dict["netid"],
+                    email=user_dict["email"],
+                    first_name=user_dict["first_name"],
+                    last_name=user_dict["last_name"],
+                    register_method=self.LOGIN_METHOD,
+                )
+
+        # Sync all fields from CAS on every login
+        user.netid = user_dict["netid"]
+        user.first_name = user_dict["first_name"]
+        user.last_name = user_dict["last_name"]
         user.last_login_method = self.LOGIN_METHOD
+
+        # Only update email if no other user already has it
+        conflict = (
+            User.objects.filter(email=user_dict["email"]).exclude(pk=user.pk).first()
+        )
+        if conflict:
+            CasFailure.objects.create(
+                code="EMAIL_CONFLICT",
+                details=(
+                    f"CAS returned email {user_dict['email']} for user {user.netid} (id={user.pk}), "
+                    f"but it belongs to user {conflict.netid} (id={conflict.pk})"
+                ),
+            )
+        else:
+            user.email = user_dict["email"]
+
         user.save()
 
         return user

users/migrations/0009_remove_legacy_tables.py

@@ -0,0 +1,28 @@
+from django.db import migrations
+
+LEGACY_TABLES = [
+    "actstream_action",
+    "actstream_follow",
+    "authtoken_token",
+]
+
+
+def remove_legacy_tables(apps, schema_editor):
+    vendor = schema_editor.connection.vendor
+    with schema_editor.connection.cursor() as cursor:
+        for table in LEGACY_TABLES:
+            if vendor == "sqlite":
+                cursor.execute(f"DROP TABLE IF EXISTS {table}")
+            else:
+                cursor.execute(f"DROP TABLE IF EXISTS {table} CASCADE")
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("users", "0008_user_moderator_welcome_dismissed"),
+    ]
+
+    operations = [
+        migrations.RunPython(remove_legacy_tables, migrations.RunPython.noop),
+    ]

users/migrations/0010_delete_numeric_netid_users.py

@@ -0,0 +1,59 @@
+from django.db import migrations
+
+
+def delete_numeric_netid_users(apps, schema_editor):
+    User = apps.get_model("users", "User")
+    Document = apps.get_model("documents", "Document")
+    Course = apps.get_model("catalog", "Course")
+
+    numeric_users = User.objects.filter(netid__regex=r"^\d+$")
+    numeric_user_ids = set(numeric_users.values_list("id", flat=True))
+
+    if not numeric_user_ids:
+        return
+
+    # Safety check: we expect ~64, more than 100 would be suspicious
+    if len(numeric_user_ids) > 100:
+        raise RuntimeError(
+            f"Expected ~64 numeric-netid users, found {len(numeric_user_ids)}"
+        )
+
+    # Safety check: none of these users should have documents
+    doc_count = Document.objects.filter(user__in=numeric_user_ids).count()
+    if doc_count:
+        raise RuntimeError(
+            f"Cannot delete numeric-netid users: {doc_count} documents would be lost"
+        )
+
+    # Transfer course follows from duplicate numeric-netid users to their
+    # original account (matched by email local part) where possible.
+    for user in numeric_users.iterator():
+        courses = Course.objects.filter(followed_by=user)
+        if not courses.exists():
+            continue
+
+        local_part = user.email.split("@")[0].lower()
+        original = (
+            User.objects.filter(email__istartswith=local_part + "@")
+            .exclude(netid__regex=r"^\d+$")
+            .exclude(id=user.id)
+            .first()
+        )
+
+        if original:
+            for course in courses:
+                course.followed_by.add(original)
+
+    deleted_count, _ = numeric_users.delete()
+    print(f"\n  Deleted {deleted_count} users with numeric netids")
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("users", "0009_remove_legacy_tables"),
+    ]
+
+    operations = [
+        migrations.RunPython(delete_numeric_netid_users, migrations.RunPython.noop),
+    ]

users/migrations/0011_cas_failure.py

@@ -0,0 +1,35 @@
+# Generated by Django 6.0.5 on 2026-05-15 20:21
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("users", "0010_delete_numeric_netid_users"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="CasFailure",
+            fields=[
+                (
+                    "id",
+                    models.BigAutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("created", models.DateTimeField(auto_now_add=True)),
+                ("code", models.CharField(max_length=64)),
+                ("details", models.TextField(blank=True, default="")),
+                ("ticket", models.CharField(blank=True, default="", max_length=512)),
+                ("ip_address", models.GenericIPAddressField(blank=True, null=True)),
+            ],
+            options={
+                "ordering": ["-created"],
+            },
+        ),
+    ]

users/models.py

@@ -97,3 +97,17 @@ def get_short_name(self):
 
     def initials(self):
         return self.first_name[0] + self.last_name[0]
+
+
+class CasFailure(models.Model):
+    created = models.DateTimeField(auto_now_add=True)
+    code = models.CharField(max_length=64)
+    details = models.TextField(blank=True, default="")
+    ticket = models.CharField(max_length=512, blank=True, default="")
+    ip_address = models.GenericIPAddressField(null=True, blank=True)
+
+    class Meta:
+        ordering = ["-created"]
+
+    def __str__(self):
+        return f"{self.code} ({self.created:%Y-%m-%d %H:%M})"

users/tests/auth_backend_authenticate_test.py

@@ -3,7 +3,7 @@
 from responses import matchers
 
 from users.authBackend import CasRequestError, UlbCasBackend
-from users.models import User
+from users.models import CasFailure, User
 
 pytestmark = pytest.mark.django_db
 
@@ -70,3 +70,122 @@ def test_server_error(fake_base_url):
         UlbCasBackend().authenticate(None, ticket=ticket)
 
     assert e.value.args[0].status_code == 500
+
+
+CAS_XML_TEMPLATE = """\
+<cas:serviceResponse xmlns:cas='http://www.yale.edu/tp/cas'>
+    <cas:authenticationSuccess>
+        <cas:user>{netid}</cas:user>
+        <cas:attributes>
+            <cas:mail>{email}</cas:mail>
+            <cas:sn>{last_name}</cas:sn>
+            <cas:givenName>{first_name}</cas:givenName>
+        </cas:attributes>
+    </cas:authenticationSuccess>
+</cas:serviceResponse>"""
+
+TICKET = "test-ticket"
+SERVICE_MATCHER = matchers.query_string_matcher(
+    f"ticket={TICKET}&service=http%3A%2F%2Fexample.com%2Fauth-ulb"
+)
+
+
+def _mock_cas_response(netid, email, first_name="Test", last_name="User"):
+    xml = CAS_XML_TEMPLATE.format(
+        netid=netid, email=email, first_name=first_name, last_name=last_name
+    )
+    responses.add(
+        responses.GET,
+        "https://auth.ulb.be/proxyValidate",
+        body=xml,
+        status=200,
+        match=[SERVICE_MATCHER],
+    )
+
+
+@responses.activate
+def test_netid_match_updates_email(fake_base_url):
+    """When netid matches an existing user, reuse it and update email."""
+    User.objects.create_user(
+        netid="glagaff", email="gaston.lagaffe@ulb.ac.be", first_name="Gaston"
+    )
+
+    _mock_cas_response("glagaff", "gaston.lagaffe@ulb.be")
+    user = UlbCasBackend().authenticate(None, ticket=TICKET)
+
+    assert user.netid == "glagaff"
+    assert user.email == "gaston.lagaffe@ulb.be"
+    assert User.objects.count() == 1
+
+
+@responses.activate
+def test_email_fallback_updates_netid(fake_base_url):
+    """When netid doesn't match but email does, reuse the user and update netid."""
+    User.objects.create_user(
+        netid="fantasio", email="fantasio@ulb.be", first_name="Fantasio"
+    )
+
+    _mock_cas_response("fant0001", "fantasio@ulb.be")
+    user = UlbCasBackend().authenticate(None, ticket=TICKET)
+
+    assert user.netid == "fant0001"
+    assert user.email == "fantasio@ulb.be"
+    assert User.objects.count() == 1
+
+
+@responses.activate
+def test_no_match_creates_user(fake_base_url):
+    """When neither netid nor email match, create a new user."""
+    _mock_cas_response("mleblanc", "modeste.leblanc@ulb.be", "Modeste", "Leblanc")
+    user = UlbCasBackend().authenticate(None, ticket=TICKET)
+
+    assert user.netid == "mleblanc"
+    assert user.email == "modeste.leblanc@ulb.be"
+    assert user.first_name == "Modeste"
+    assert user.last_name == "Leblanc"
+    assert User.objects.count() == 1
+
+
+@responses.activate
+def test_fields_synced_on_login(fake_base_url):
+    """All CAS fields are updated on every login."""
+    User.objects.create_user(
+        netid="pdemousk",
+        email="prunelle.de.mouskinson@ulb.ac.be",
+        first_name="Leon",
+        last_name="Prunelle",
+    )
+
+    _mock_cas_response(
+        "pdemousk", "prunelle.de.mouskinson@ulb.be", "Leon", "De Mouskinson"
+    )
+    user = UlbCasBackend().authenticate(None, ticket=TICKET)
+
+    assert user.email == "prunelle.de.mouskinson@ulb.be"
+    assert user.first_name == "Leon"
+    assert user.last_name == "De Mouskinson"
+
+
+@responses.activate
+def test_no_duplicate_when_netid_and_email_match_different_users(fake_base_url):
+    """When netid matches user A and email matches user B, prefer user A (netid)."""
+    User.objects.create_user(
+        netid="glagaff", email="gaston.lagaffe@ulb.ac.be", first_name="Gaston"
+    )
+    User.objects.create_user(
+        netid="lechat", email="gaston.lagaffe@ulb.be", first_name="Le Chat"
+    )
+
+    _mock_cas_response("glagaff", "gaston.lagaffe@ulb.be")
+    user = UlbCasBackend().authenticate(None, ticket=TICKET)
+
+    assert user.netid == "glagaff"
+    # Email not updated because Le Chat already has it
+    assert user.email == "gaston.lagaffe@ulb.ac.be"
+    assert User.objects.count() == 2
+
+    # An EMAIL_CONFLICT failure should be logged
+    failure = CasFailure.objects.get()
+    assert failure.code == "EMAIL_CONFLICT"
+    assert "glagaff" in failure.details
+    assert "lechat" in failure.details