diff --git a/rodatraden/backends.py b/rodatraden/backends.py new file mode 100644 index 0000000..4a4d841 --- /dev/null +++ b/rodatraden/backends.py @@ -0,0 +1,54 @@ +""" +Custom authentication backends for Röda Tråden. + +EmailOrUsernameBackend allows users to log in with either their username +or email address, which is needed after the user merge since users may +be accustomed to logging in with different usernames. +""" + +from django.contrib.auth import get_user_model +from django.contrib.auth.backends import ModelBackend +from django.db.models import Q + + +class EmailOrUsernameBackend(ModelBackend): + """ + Custom authentication backend that allows users to login with either + their username or email address. + """ + + def authenticate(self, request, username=None, password=None, **kwargs): + UserModel = get_user_model() + + if username is None: + username = kwargs.get(UserModel.USERNAME_FIELD) + + if username is None or password is None: + return None + + try: + # Try to find user by exact username or case-insensitive email + user = UserModel.objects.get( + Q(username=username) | Q(email__iexact=username) + ) + except UserModel.DoesNotExist: + # Run the default password hasher once to reduce the timing + # difference between an existing and a nonexistent user. + UserModel().set_password(password) + return None + except UserModel.MultipleObjectsReturned: + # If multiple users match (e.g. during migration), try exact + # username first, then fall back to the first email match. + try: + user = UserModel.objects.get(username=username) + except UserModel.DoesNotExist: + user = UserModel.objects.filter( + email__iexact=username + ).order_by('-last_login').first() + if user is None: + return None + + if user.check_password(password) and self.user_can_authenticate(user): + return user + + return None diff --git a/rodatraden/management/commands/merge_users_by_email.py b/rodatraden/management/commands/merge_users_by_email.py new file mode 100644 index 0000000..e97d21b --- /dev/null +++ b/rodatraden/management/commands/merge_users_by_email.py @@ -0,0 +1,374 @@ +""" +Management command to merge duplicate user accounts that share the same email. + +Users are identified by username in this application, which has led to some +users creating multiple accounts with the same email address. This command +finds those duplicates, keeps the most recently active account, and migrates +all data (block schedules and private courses) to that account. + +Old accounts are deactivated (not deleted) so data can be recovered if needed. + +Usage: + # Preview what would happen (no changes made): + python manage.py merge_users_by_email --dry-run + + # Perform the merge: + python manage.py merge_users_by_email + + # Perform the merge and email affected users: + python manage.py merge_users_by_email --send-emails + + # Skip interactive confirmation: + python manage.py merge_users_by_email --no-input +""" + +from django.core.management.base import BaseCommand +from django.contrib.auth import get_user_model +from django.db.models import Count +from django.core.mail import send_mail +from django.conf import settings +from django.utils.text import slugify + +import smtplib + +User = get_user_model() + + +class Command(BaseCommand): + help = ( + 'Merge duplicate user accounts that share the same email address. ' + 'Keeps the most recently logged-in account, migrates blocks and ' + 'private courses, and deactivates old accounts.' + ) + + def add_arguments(self, parser): + parser.add_argument( + '--dry-run', + action='store_true', + help='Preview what would happen without making any changes.', + ) + parser.add_argument( + '--send-emails', + action='store_true', + help='Send email notifications to affected users after merging.', + ) + parser.add_argument( + '--no-input', + action='store_true', + help='Skip interactive confirmation prompts.', + ) + + def handle(self, *args, **options): + dry_run = options['dry_run'] + send_emails = options['send_emails'] + no_input = options['no_input'] + + if dry_run: + self.stdout.write(self.style.WARNING( + '\n=== DRY RUN MODE — no changes will be made ===\n' + )) + + # Import models here to avoid circular imports + from rodatraden.models import Block, PrivateCourse + + # Find emails that have more than one user account + duplicate_emails = ( + User.objects + .exclude(email='') + .exclude(email__isnull=True) + .values('email') + .annotate(user_count=Count('id')) + .filter(user_count__gt=1) + .order_by('-user_count') + ) + + if not duplicate_emails.exists(): + self.stdout.write(self.style.SUCCESS( + 'No duplicate email addresses found. Nothing to do.' + )) + return + + # Summary of what we found + total_groups = duplicate_emails.count() + total_extra_users = sum( + d['user_count'] - 1 for d in duplicate_emails + ) + self.stdout.write(self.style.NOTICE( + f'\nFound {total_groups} email(s) with duplicate accounts ' + f'({total_extra_users} extra accounts to merge).\n' + )) + + # Build a detailed plan + merge_plan = [] + for entry in duplicate_emails: + email = entry['email'] + users = list( + User.objects.filter(email=email) + .order_by('-last_login', '-date_joined') + ) + + # Target: most recently logged-in user (fallback to most recently + # joined if none have logged in) + target = users[0] + old_users = users[1:] + + group_info = { + 'email': email, + 'target': target, + 'old_users': old_users, + 'migrations': [], + } + + for old_user in old_users: + blocks = Block.objects.filter(user=old_user) + private_courses = PrivateCourse.objects.filter(user=old_user) + group_info['migrations'].append({ + 'user': old_user, + 'blocks': list(blocks), + 'private_courses': list(private_courses), + }) + + merge_plan.append(group_info) + + # Print the plan + self._print_plan(merge_plan) + + if dry_run: + self.stdout.write(self.style.WARNING( + '\n=== DRY RUN COMPLETE — no changes were made ===\n' + 'Run without --dry-run to perform the merge.\n' + )) + return + + # Confirm with user + if not no_input: + confirm = input( + '\nProceed with the merge? This will deactivate old accounts ' + 'and reassign their data. [y/N]: ' + ) + if confirm.lower() != 'y': + self.stdout.write(self.style.WARNING('Merge cancelled.')) + return + + # Perform the merge + email_notifications = [] + for group in merge_plan: + notifications = self._merge_group(group) + email_notifications.extend(notifications) + + self.stdout.write(self.style.SUCCESS( + f'\nMerge complete! {total_extra_users} account(s) deactivated.' + )) + + # Send emails if requested + if send_emails: + self._send_notifications(email_notifications) + + def _print_plan(self, merge_plan): + """Print a human-readable merge plan.""" + + for group in merge_plan: + email = group['email'] + target = group['target'] + self.stdout.write(self.style.HTTP_INFO( + f'━━━ Email: {email} ━━━' + )) + self.stdout.write( + f' ✓ Keep: {target.username} ' + f'(last login: {target.last_login or "never"}, ' + f'joined: {target.date_joined})' + ) + + for migration in group['migrations']: + old_user = migration['user'] + blocks = migration['blocks'] + private_courses = migration['private_courses'] + self.stdout.write(self.style.WARNING( + f' ✗ Deactivate: {old_user.username} ' + f'(last login: {old_user.last_login or "never"}, ' + f'joined: {old_user.date_joined})' + )) + if blocks: + self.stdout.write( + f' → Migrate {len(blocks)} block schedule(s):' + ) + for block in blocks: + new_title = f'{old_user.username} - {block.title}' + self.stdout.write( + f' "{block.title}" → "{new_title}"' + ) + if private_courses: + self.stdout.write( + f' → Migrate {len(private_courses)} ' + f'private course(s):' + ) + for pc in private_courses: + new_title = f'{old_user.username} - {pc.title}' + self.stdout.write( + f' "{pc.title}" → "{new_title}"' + ) + if not blocks and not private_courses: + self.stdout.write(' → No data to migrate') + + def _merge_group(self, group): + """Merge a single group of duplicate accounts. + + Returns a list of email notification dicts. + """ + + from rodatraden.models import Block, PrivateCourse, get_unique_slug + + target = group['target'] + notifications = [] + + for migration in group['migrations']: + old_user = migration['user'] + migrated_blocks = [] + migrated_courses = [] + + # Migrate block schedules + for block in migration['blocks']: + old_title = block.title + # Prefix with old username to avoid naming collisions + new_title = f'{old_user.username} - {old_title}' + block.title = new_title + block.user = target + # Regenerate slug to avoid collisions under the new user + block.slug = get_unique_slug( + to_slug=new_title, model=Block + ) + block.save() + migrated_blocks.append({ + 'old_title': old_title, + 'new_title': new_title, + }) + self.stdout.write(self.style.SUCCESS( + f' Migrated block "{old_title}" → "{new_title}" ' + f'(from {old_user.username} to {target.username})' + )) + + # Migrate private courses + for pc in migration['private_courses']: + old_title = pc.title + # Prefix with old username to avoid naming collisions + new_title = f'{old_user.username} - {old_title}' + pc.title = new_title + pc.user = target + # Regenerate slug to avoid collisions under the new user + pc.slug = get_unique_slug( + to_slug=new_title, model=PrivateCourse + ) + pc.save() + migrated_courses.append({ + 'old_title': old_title, + 'new_title': new_title, + }) + self.stdout.write(self.style.SUCCESS( + f' Migrated private course "{old_title}" → ' + f'"{new_title}" ' + f'(from {old_user.username} to {target.username})' + )) + + # Deactivate the old user (do NOT delete — keep for reference) + old_user.is_active = False + old_user.save() + self.stdout.write(self.style.WARNING( + f' Deactivated user: {old_user.username}' + )) + + notifications.append({ + 'email': group['email'], + 'target_username': target.username, + 'old_username': old_user.username, + 'migrated_blocks': migrated_blocks, + 'migrated_courses': migrated_courses, + }) + + return notifications + + def _send_notifications(self, notifications): + """Send email notifications to affected users.""" + + # Group notifications by email so each user gets one email + by_email = {} + for n in notifications: + email = n['email'] + if email not in by_email: + by_email[email] = { + 'target_username': n['target_username'], + 'merged_accounts': [], + } + by_email[email]['merged_accounts'].append(n) + + sent = 0 + failed = 0 + for email, data in by_email.items(): + target = data['target_username'] + body_lines = [ + 'Hej!', + '', + 'Vi har upptäckt att du har flera konton på Röda Tråden ' + 'med samma e-postadress. För att förenkla har vi slagit ' + 'ihop dina konton till ett.', + '', + f'Ditt aktiva konto är: {target}', + '', + 'Följande konton har avaktiverats och deras data har ' + 'flyttats till ditt aktiva konto:', + '', + ] + + for account in data['merged_accounts']: + body_lines.append( + f' • {account["old_username"]}' + ) + if account['migrated_blocks']: + body_lines.append(' Blockscheman som flyttats:') + for b in account['migrated_blocks']: + body_lines.append( + f' - "{b["old_title"]}" ' + f'(nytt namn: "{b["new_title"]}")' + ) + if account['migrated_courses']: + body_lines.append(' Privata kurser som flyttats:') + for c in account['migrated_courses']: + body_lines.append( + f' - "{c["old_title"]}" ' + f'(nytt namn: "{c["new_title"]}")' + ) + + body_lines.extend([ + '', + 'Du kan logga in med ditt aktiva konto och byta namn ' + 'på de flyttade blockschemana och kurserna om du vill.', + '', + 'Om du har frågor, kontakta oss genom att svara på ' + 'detta mail.', + '', + 'Med vänliga hälsningar,', + 'Röda Tråden', + ]) + + body = '\n'.join(body_lines) + + try: + send_mail( + subject='Röda Tråden — Dina konton har slagits ihop', + message=body, + from_email=settings.DEFAULT_FROM_EMAIL, + recipient_list=[email], + fail_silently=False, + ) + sent += 1 + self.stdout.write(self.style.SUCCESS( + f' Email sent to {email}' + )) + except (smtplib.SMTPException, ConnectionError, OSError) as e: + failed += 1 + self.stdout.write(self.style.ERROR( + f' Failed to send email to {email}: {e}' + )) + + self.stdout.write(self.style.SUCCESS( + f'\nEmails sent: {sent}, failed: {failed}' + )) diff --git a/rodatraden/tests.py b/rodatraden/tests.py index 7ce503c..4081b62 100644 --- a/rodatraden/tests.py +++ b/rodatraden/tests.py @@ -1,3 +1,88 @@ +""" +Tests for the merge_users_by_email management command and the +EmailOrUsernameBackend authentication backend. +""" + +from io import StringIO + +from django.contrib.auth import get_user_model +from django.core.management import call_command from django.test import TestCase -# Create your tests here. +User = get_user_model() + + +class MergeUsersByEmailDryRunTests(TestCase): + """Tests for the merge_users_by_email management command.""" + + def _call_command(self, *args, **kwargs): + out = StringIO() + kwargs.setdefault('stdout', out) + kwargs.setdefault('stderr', StringIO()) + call_command('merge_users_by_email', *args, **kwargs) + return out.getvalue() + + def test_no_duplicates(self): + """Command exits cleanly when there are no duplicate emails.""" + User.objects.create_user('user1', 'a@example.com', 'pass') + User.objects.create_user('user2', 'b@example.com', 'pass') + output = self._call_command('--dry-run') + self.assertIn('No duplicate email addresses found', output) + + def test_dry_run_does_not_change_data(self): + """Dry run should not modify any users.""" + User.objects.create_user('user1', 'same@example.com', 'pass') + User.objects.create_user('user2', 'same@example.com', 'pass') + self._call_command('--dry-run') + + # Both users should still be active + self.assertTrue(User.objects.get(username='user1').is_active) + self.assertTrue(User.objects.get(username='user2').is_active) + + def test_empty_emails_ignored(self): + """Users with empty emails should not be grouped together.""" + User.objects.create_user('user1', '', 'pass') + User.objects.create_user('user2', '', 'pass') + output = self._call_command('--dry-run') + self.assertIn('No duplicate email addresses found', output) + + +class EmailOrUsernameBackendTests(TestCase): + """Tests for the EmailOrUsernameBackend.""" + + def test_login_by_username(self): + """Users should be able to log in with their username.""" + from rodatraden.backends import EmailOrUsernameBackend + backend = EmailOrUsernameBackend() + User.objects.create_user('testuser', 'test@example.com', 'testpass') + + user = backend.authenticate(None, username='testuser', password='testpass') + self.assertIsNotNone(user) + self.assertEqual(user.username, 'testuser') + + def test_login_by_email(self): + """Users should be able to log in with their email.""" + from rodatraden.backends import EmailOrUsernameBackend + backend = EmailOrUsernameBackend() + User.objects.create_user('testuser', 'test@example.com', 'testpass') + + user = backend.authenticate(None, username='test@example.com', password='testpass') + self.assertIsNotNone(user) + self.assertEqual(user.username, 'testuser') + + def test_wrong_password_fails(self): + """Wrong password should return None.""" + from rodatraden.backends import EmailOrUsernameBackend + backend = EmailOrUsernameBackend() + User.objects.create_user('testuser', 'test@example.com', 'testpass') + + user = backend.authenticate(None, username='testuser', password='wrong') + self.assertIsNone(user) + + def test_nonexistent_user_fails(self): + """Nonexistent user should return None.""" + from rodatraden.backends import EmailOrUsernameBackend + backend = EmailOrUsernameBackend() + + user = backend.authenticate(None, username='nobody', password='pass') + self.assertIsNone(user) diff --git a/rodatraden/views.py b/rodatraden/views.py index ac34436..ad5b8b9 100644 --- a/rodatraden/views.py +++ b/rodatraden/views.py @@ -1242,7 +1242,7 @@ def block_detail(request: HttpRequest, username, slug): # If the block is not private, show without authentication if block.private: if not request.user.is_authenticated: - return redirect(reverse('cas_ng_login')) + return redirect(settings.LOGIN_URL) elif request.user.username != block.user.username: return redirect(reverse('index')) diff --git a/tf/forms.py b/tf/forms.py index e273033..f605920 100644 --- a/tf/forms.py +++ b/tf/forms.py @@ -1,6 +1,28 @@ +from django.contrib.auth import get_user_model +from django.core.exceptions import ValidationError from django_registration.forms import RegistrationForm from captcha.fields import CaptchaField +User = get_user_model() + class RodatradenRegistrationForm(RegistrationForm): recaptcha = CaptchaField(label="Jag är en människa!") + + def clean_email(self): + """Prevent registration if the email is already in use. + + After the user merge, each email should correspond to exactly one + active account. This validation enforces that going forward. + Inactive (merged) accounts are excluded so that their email can + still be used by the active account holder. + """ + + email = self.cleaned_data.get('email', '') + if email and User.objects.filter(email__iexact=email, is_active=True).exists(): + raise ValidationError( + 'Ett konto med denna e-postadress finns redan. ' + 'Vänligen logga in med ditt befintliga konto eller ' + 'använd en annan e-postadress.' + ) + return email diff --git a/tf/settings-template.py b/tf/settings-template.py index 5a0bb60..378d786 100644 --- a/tf/settings-template.py +++ b/tf/settings-template.py @@ -87,6 +87,7 @@ # For CAS authentication AUTHENTICATION_BACKENDS = ( + 'rodatraden.backends.EmailOrUsernameBackend', 'django.contrib.auth.backends.ModelBackend', )