feat: normalize usernames to ASCII

trojstenid/schools/signals.py

@@ -2,9 +2,9 @@
 from django.dispatch import receiver
 
 from trojstenid.schools.models import UserSchoolRecord
-from trojstenid.users.tasks import send_user_update
+from trojstenid.users.tasks import queue_user_update
 
 
 @receiver(post_save, sender=UserSchoolRecord)
 def user_school_record_saved(sender, instance: UserSchoolRecord, **kwargs):
-    send_user_update.delay(instance.user_id)
+    queue_user_update(instance.user_id)

trojstenid/settings.py

@@ -143,6 +143,7 @@
     "reset_password_from_key": "trojstenid.users.forms.allauth.OurResetPasswordKeyForm",
     "set_password": "trojstenid.users.forms.allauth.OurSetPasswordForm",
 }
+ACCOUNT_USERNAME_VALIDATORS = "trojstenid.users.models.username_validators"
 
 SOCIALACCOUNT_PROVIDERS = {
     "openid_connect": {

trojstenid/users/management/commands/normalize_usernames.py

@@ -0,0 +1,112 @@
+import re
+import unicodedata
+
+from django.core.management.base import BaseCommand
+from django.db import transaction
+
+from trojstenid.users.models import User, UsernameValidator
+
+
+def normalize_username(username: str) -> str:
+    normalized = (
+        unicodedata.normalize("NFKD", username).encode("ASCII", "ignore").decode()
+    )
+
+    if "@" in normalized:
+        parts = normalized.split("@", 1)
+        if parts[0]:
+            normalized = parts[0]
+        else:
+            normalized = parts[1]
+
+    normalized = re.sub(r"[^\w.-]", "", normalized)
+
+    return normalized
+
+
+class Command(BaseCommand):
+    help = "Validate and normalize usernames"
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--dry-run",
+            action="store_true",
+            help="Run without making changes to the database",
+        )
+
+    @transaction.atomic
+    def handle(self, *args, **options):
+        dry_run = options["dry_run"]
+        validator = UsernameValidator()
+
+        users = User.objects.all()
+
+        for user in users:
+            try:
+                validator(user.username)
+            except Exception:
+                new_username = normalize_username(user.username)
+
+                if new_username == user.username:
+                    continue
+
+                # Check if normalized username becomes empty
+                if not new_username:
+                    new_username = self.prompt_for_username(
+                        user,
+                        reason="Normalized username would be empty",
+                    )
+
+                # Check for duplicates
+                duplicate_user = (
+                    User.objects.filter(username=new_username)
+                    .exclude(id=user.id)
+                    .first()
+                )
+                if duplicate_user:
+                    new_username = self.prompt_for_username(
+                        user,
+                        reason=f"Username '{new_username}' already exists (user ID: {duplicate_user.id})",
+                    )
+
+                self.stdout.write(f"{user.username} ({user.id}) -> {new_username}")
+
+                if not dry_run:
+                    user.username = new_username
+                    user.save()
+
+    def prompt_for_username(self, user: User, reason: str) -> str:
+        """Prompt the user for a new username."""
+        self.stdout.write(self.style.WARNING(f"\n{reason}"))
+        self.stdout.write(f"Old username: {user.username}")
+        self.stdout.write(f"Email: {user.email}")
+        self.stdout.write(f"Full name: {user.get_full_name() or '(not set)'}")
+
+        while True:
+            new_username = input("Enter new username: ").strip()
+
+            if not new_username:
+                self.stdout.write(self.style.ERROR("Username cannot be empty."))
+                continue
+
+            # Validate the username format
+            validator = UsernameValidator()
+            try:
+                validator(new_username)
+            except Exception as e:
+                self.stdout.write(self.style.ERROR(f"Invalid username: {e}"))
+                continue
+
+            # Check for duplicates
+            duplicate_user = (
+                User.objects.filter(username=new_username).exclude(id=user.id).first()
+            )
+            if duplicate_user:
+                self.stdout.write(
+                    self.style.ERROR(
+                        f"Username '{new_username}' already in use by user ID {duplicate_user.id}"
+                    )
+                )
+                continue
+
+            return new_username

trojstenid/users/models.py

@@ -1,13 +1,16 @@
+import re
 from datetime import date
 from pathlib import PurePath
 from typing import TYPE_CHECKING
 
 from django.contrib.auth.models import AbstractUser, Group
+from django.core import validators
 from django.core.exceptions import ObjectDoesNotExist
 from django.db import models
 from django.db.models import Q
 from django.urls import reverse
 from django.utils import timezone
+from django.utils.deconstruct import deconstructible
 from oauth2_provider.models import AbstractApplication
 from ulid import ULID
 
@@ -40,6 +43,16 @@ def get_db_converters(self, connection):
         return []
 
 
+@deconstructible()
+class UsernameValidator(validators.RegexValidator):
+    regex = r"^[\w.-]+\Z"
+    message = "Používateľské meno môže obsahovať len písmená, čísla a znaky ./-/_"
+    flags = re.ASCII
+
+
+username_validators = [UsernameValidator()]
+
+
 class User(AbstractUser):
     id: int
 

trojstenid/users/signals.py

@@ -13,7 +13,7 @@
 from trojstenid import audit
 from trojstenid.users.models import User
 from trojstenid.users.tasks import (
-    send_user_update,
+    queue_user_update,
     sync_github_teams_for_user,
     sync_groups,
 )
@@ -47,19 +47,19 @@ def log_app_authorization(sender, request, token, **kwargs):
 def user_saved(sender, instance: User, *, update_fields, **kwargs):
     if update_fields is not None and update_fields.issubset({"last_login"}):
         return
-    send_user_update.delay(instance.id)
+    queue_user_update(instance.id)
 
 
 @receiver(post_save, sender=EmailAddress)
 def emailaddress_saved(sender, instance: EmailAddress, **kwargs):
-    send_user_update.delay(instance.user_id)  # type:ignore
+    queue_user_update(instance.user_id)  # type:ignore
 
 
 @receiver(m2m_changed, sender=User.groups.through)
 def groups_changed(sender, instance, **kwargs):
     if not isinstance(instance, User):
         return
-    send_user_update.delay(instance.id)
+    queue_user_update(instance.id)
 
 
 @receiver(post_save, sender=SocialAccount)

trojstenid/users/tasks.py

@@ -1,6 +1,7 @@
 import logging
 
 import requests
+from django.db import transaction
 from django_rq import job
 
 from trojstenid.users.github import sync_github_teams
@@ -11,6 +12,10 @@
 logger = logging.getLogger(__name__)
 
 
+def queue_user_update(user_id: int):
+    transaction.on_commit(lambda: send_user_update.delay(user_id))
+
+
 @job
 def send_user_update(user_id: int):
     logger.info(f"pushing user update (uid {user_id})")