Spaces:
Paused
Paused
| """ | |
| User, MagicToken, and PasswordResetToken models for authentication. | |
| Supports both password-based and magic link authentication. | |
| """ | |
| import secrets | |
| from datetime import timedelta | |
| from django.contrib.auth.models import AbstractUser | |
| from django.db import models | |
| from django.utils import timezone | |
| class User(AbstractUser): | |
| """ | |
| Custom user model supporting both password and magic link authentication. | |
| Uses email as the primary identifier instead of username. | |
| """ | |
| email = models.EmailField(unique=True) | |
| is_verified = models.BooleanField( | |
| default=False, | |
| help_text="Whether the user has verified their email via magic link", | |
| ) | |
| # Use email as the username field | |
| USERNAME_FIELD = "email" | |
| REQUIRED_FIELDS = [] # Email is already required via USERNAME_FIELD | |
| class Meta: | |
| db_table = "users" | |
| verbose_name = "user" | |
| verbose_name_plural = "users" | |
| def __str__(self): | |
| return self.email | |
| def save(self, *args, **kwargs): | |
| # Auto-set username to email if not provided | |
| if not self.username: | |
| self.username = self.email | |
| # Only set unusable password for new users without a password | |
| # This allows password auth while keeping magic link as an option | |
| if self._state.adding and not self.password: | |
| self.set_unusable_password() | |
| super().save(*args, **kwargs) | |
| class MagicToken(models.Model): | |
| """ | |
| One-time use token for passwordless authentication. | |
| Tokens expire after a configurable time and can only be used once. | |
| """ | |
| EXPIRY_MINUTES = 15 | |
| token = models.CharField(max_length=64, unique=True, db_index=True) | |
| user = models.ForeignKey( | |
| User, | |
| on_delete=models.CASCADE, | |
| related_name="magic_tokens", | |
| null=True, | |
| blank=True, | |
| ) | |
| email = models.EmailField(help_text="Email address this token was sent to") | |
| created_at = models.DateTimeField(auto_now_add=True) | |
| expires_at = models.DateTimeField() | |
| used_at = models.DateTimeField(null=True, blank=True) | |
| ip_address = models.GenericIPAddressField( | |
| null=True, | |
| blank=True, | |
| help_text="IP address that requested this token", | |
| ) | |
| class Meta: | |
| db_table = "magic_tokens" | |
| verbose_name = "magic token" | |
| verbose_name_plural = "magic tokens" | |
| indexes = [ | |
| models.Index(fields=["token", "expires_at"]), | |
| models.Index(fields=["email", "created_at"]), | |
| ] | |
| def __str__(self): | |
| status = "used" if self.used_at else ("expired" if not self.is_valid() else "valid") | |
| return f"MagicToken for {self.email} ({status})" | |
| def create_for_email(cls, email: str, ip_address: str = None) -> "MagicToken": | |
| """ | |
| Create a new magic token for an email address. | |
| Invalidates any existing unused tokens for the same email. | |
| The first user to register is automatically promoted to admin (superuser). | |
| """ | |
| # Invalidate existing unused tokens for this email | |
| cls.objects.filter(email=email, used_at__isnull=True).update( | |
| expires_at=timezone.now() | |
| ) | |
| # Check if this will be the first user (before creating) | |
| is_first_user = User.objects.count() == 0 | |
| # Get or create user for this email | |
| user, created = User.objects.get_or_create( | |
| email=email, | |
| defaults={"is_verified": False}, | |
| ) | |
| # Auto-promote first user to admin and disable open registration | |
| if created and is_first_user: | |
| user.is_staff = True | |
| user.is_superuser = True | |
| user.save(update_fields=["is_staff", "is_superuser"]) | |
| # Auto-disable open registration after first user | |
| from core.models.settings import GlobalSettings | |
| settings = GlobalSettings.get_settings() | |
| settings.allow_registration = False | |
| settings.save(update_fields=["allow_registration"]) | |
| # Create new token | |
| return cls.objects.create( | |
| token=secrets.token_urlsafe(48), | |
| user=user, | |
| email=email, | |
| expires_at=timezone.now() + timedelta(minutes=cls.EXPIRY_MINUTES), | |
| ip_address=ip_address, | |
| ) | |
| def is_valid(self) -> bool: | |
| """Check if the token is still valid (not expired and not used).""" | |
| return self.used_at is None and self.expires_at > timezone.now() | |
| def consume(self) -> User: | |
| """ | |
| Mark the token as used and return the associated user. | |
| Also marks the user as verified. | |
| """ | |
| if not self.is_valid(): | |
| raise ValueError("Token is no longer valid") | |
| self.used_at = timezone.now() | |
| self.save(update_fields=["used_at"]) | |
| # Mark user as verified | |
| self.user.is_verified = True | |
| self.user.save(update_fields=["is_verified"]) | |
| return self.user | |
| class UserInvite(models.Model): | |
| """ | |
| Invitation for a new user. Admin creates invite, gets shareable link. | |
| Can be used once to create an account when registration is closed. | |
| """ | |
| EXPIRY_DAYS = 7 | |
| email = models.EmailField( | |
| unique=True, | |
| help_text="Email address of the invited user", | |
| ) | |
| token = models.CharField(max_length=64, unique=True, db_index=True) | |
| created_by = models.ForeignKey( | |
| User, | |
| on_delete=models.CASCADE, | |
| related_name="sent_invites", | |
| ) | |
| created_at = models.DateTimeField(auto_now_add=True) | |
| expires_at = models.DateTimeField() | |
| used_at = models.DateTimeField(null=True, blank=True) | |
| used_by = models.ForeignKey( | |
| User, | |
| on_delete=models.SET_NULL, | |
| null=True, | |
| blank=True, | |
| related_name="received_invite", | |
| ) | |
| class Meta: | |
| db_table = "user_invites" | |
| verbose_name = "user invite" | |
| verbose_name_plural = "user invites" | |
| indexes = [ | |
| models.Index(fields=["token"]), | |
| models.Index(fields=["email"]), | |
| ] | |
| def __str__(self): | |
| status = "used" if self.used_at else ("expired" if not self.is_valid() else "pending") | |
| return f"Invite for {self.email} ({status})" | |
| def create_invite(cls, email: str, created_by: User) -> "UserInvite": | |
| """ | |
| Create a new invitation. Deletes any existing unused invite for the same email. | |
| """ | |
| # Delete existing unused invite for this email | |
| cls.objects.filter(email__iexact=email, used_at__isnull=True).delete() | |
| return cls.objects.create( | |
| email=email.lower().strip(), | |
| token=secrets.token_urlsafe(48), | |
| created_by=created_by, | |
| expires_at=timezone.now() + timedelta(days=cls.EXPIRY_DAYS), | |
| ) | |
| def is_valid(self) -> bool: | |
| """Check if the invite is still valid (not expired and not used).""" | |
| return self.used_at is None and self.expires_at > timezone.now() | |
| def mark_used(self, user: User) -> None: | |
| """Mark the invite as used by the specified user.""" | |
| self.used_at = timezone.now() | |
| self.used_by = user | |
| self.save(update_fields=["used_at", "used_by"]) | |
| class PasswordResetToken(models.Model): | |
| """ | |
| Token for password reset functionality. | |
| Allows users to reset their password via email link. | |
| """ | |
| EXPIRY_HOURS = 24 | |
| token = models.CharField(max_length=64, unique=True, db_index=True) | |
| user = models.ForeignKey( | |
| User, | |
| on_delete=models.CASCADE, | |
| related_name="password_reset_tokens", | |
| ) | |
| created_at = models.DateTimeField(auto_now_add=True) | |
| expires_at = models.DateTimeField() | |
| used_at = models.DateTimeField(null=True, blank=True) | |
| class Meta: | |
| db_table = "password_reset_tokens" | |
| verbose_name = "password reset token" | |
| verbose_name_plural = "password reset tokens" | |
| indexes = [ | |
| models.Index(fields=["token", "expires_at"]), | |
| ] | |
| def __str__(self): | |
| status = "used" if self.used_at else ("expired" if not self.is_valid() else "valid") | |
| return f"PasswordResetToken for {self.user.email} ({status})" | |
| def create_for_user(cls, user: User) -> "PasswordResetToken": | |
| """ | |
| Create a new password reset token for a user. | |
| Invalidates any existing unused tokens for the same user. | |
| """ | |
| # Invalidate existing unused tokens for this user | |
| cls.objects.filter(user=user, used_at__isnull=True).update( | |
| expires_at=timezone.now() | |
| ) | |
| return cls.objects.create( | |
| token=secrets.token_urlsafe(48), | |
| user=user, | |
| expires_at=timezone.now() + timedelta(hours=cls.EXPIRY_HOURS), | |
| ) | |
| def is_valid(self) -> bool: | |
| """Check if the token is still valid (not expired and not used).""" | |
| return self.used_at is None and self.expires_at > timezone.now() | |
| def consume(self) -> User: | |
| """ | |
| Mark the token as used and return the associated user. | |
| """ | |
| if not self.is_valid(): | |
| raise ValueError("Token is no longer valid") | |
| self.used_at = timezone.now() | |
| self.save(update_fields=["used_at"]) | |
| return self.user | |