pyrunner / core /models /user.py
Perspicacious's picture
Deploy PyRunner on Hugging Face
2529305
Raw
History Blame Contribute Delete
9.3 kB
"""
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})"
@classmethod
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})"
@classmethod
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})"
@classmethod
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