register and sign in pages
This commit is contained in:
@@ -0,0 +1,279 @@
|
||||
import secrets
|
||||
from datetime import timedelta
|
||||
|
||||
from django.conf import settings
|
||||
from django.contrib.auth.hashers import check_password, make_password
|
||||
from django.contrib.auth.models import AbstractBaseUser, PermissionsMixin
|
||||
from django.db import models
|
||||
from django.utils import timezone
|
||||
|
||||
from .managers import UserManager
|
||||
from .mobile import normalize_mobile
|
||||
|
||||
|
||||
class User(AbstractBaseUser, PermissionsMixin):
|
||||
"""Portal account. The mobile number is the login identifier.
|
||||
|
||||
A self-registered account starts at `status = PENDING` and only becomes
|
||||
usable once an admin approves it - see `approve()` / `reject()`.
|
||||
"""
|
||||
|
||||
class AuthMethod(models.TextChoices):
|
||||
OTP = "otp", "SMS one-time code"
|
||||
PASSWORD = "password", "Password"
|
||||
|
||||
class Status(models.TextChoices):
|
||||
PENDING = "pending", "Pending approval"
|
||||
APPROVED = "approved", "Approved"
|
||||
REJECTED = "rejected", "Rejected"
|
||||
|
||||
mobile = models.CharField(max_length=16, unique=True, db_index=True)
|
||||
full_name = models.CharField(max_length=255, blank=True)
|
||||
email = models.EmailField(blank=True, null=True, unique=True)
|
||||
|
||||
# ID card, passport or work permit number.
|
||||
idnumber = models.CharField(max_length=32, blank=True, db_index=True)
|
||||
date_of_birth = models.DateField(null=True, blank=True)
|
||||
|
||||
# Address
|
||||
atoll = models.ForeignKey(
|
||||
"locations.Atoll",
|
||||
on_delete=models.PROTECT,
|
||||
related_name="users",
|
||||
null=True,
|
||||
blank=True,
|
||||
)
|
||||
island = models.ForeignKey(
|
||||
"locations.Island",
|
||||
on_delete=models.PROTECT,
|
||||
related_name="users",
|
||||
null=True,
|
||||
blank=True,
|
||||
)
|
||||
|
||||
# Which second step /auth/start/ asks the SPA to render.
|
||||
auth_method = models.CharField(
|
||||
max_length=16, choices=AuthMethod.choices, default=AuthMethod.OTP
|
||||
)
|
||||
mobile_verified = models.BooleanField(default=False)
|
||||
|
||||
# Registration review
|
||||
status = models.CharField(
|
||||
max_length=16, choices=Status.choices, default=Status.PENDING, db_index=True
|
||||
)
|
||||
terms_accepted_at = models.DateTimeField(null=True, blank=True)
|
||||
policy_accepted_at = models.DateTimeField(null=True, blank=True)
|
||||
reviewed_at = models.DateTimeField(null=True, blank=True)
|
||||
reviewed_by = models.ForeignKey(
|
||||
"self",
|
||||
on_delete=models.SET_NULL,
|
||||
related_name="reviewed_users",
|
||||
null=True,
|
||||
blank=True,
|
||||
)
|
||||
rejection_reason = models.TextField(blank=True)
|
||||
|
||||
is_active = models.BooleanField(default=True)
|
||||
is_staff = models.BooleanField(default=False)
|
||||
|
||||
date_joined = models.DateTimeField(default=timezone.now)
|
||||
updated_at = models.DateTimeField(auto_now=True)
|
||||
|
||||
objects = UserManager()
|
||||
|
||||
USERNAME_FIELD = "mobile"
|
||||
REQUIRED_FIELDS = []
|
||||
|
||||
class Meta:
|
||||
ordering = ["-date_joined"]
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.full_name or 'Unnamed'} ({self.mobile})"
|
||||
|
||||
def save(self, *args, **kwargs):
|
||||
self.mobile = normalize_mobile(self.mobile)
|
||||
super().save(*args, **kwargs)
|
||||
|
||||
@property
|
||||
def is_admin_user(self) -> bool:
|
||||
"""Admin surface gate: staff or superuser."""
|
||||
return self.is_staff or self.is_superuser
|
||||
|
||||
@property
|
||||
def is_approved(self) -> bool:
|
||||
return self.status == self.Status.APPROVED
|
||||
|
||||
def can_use_password_login(self) -> bool:
|
||||
return (
|
||||
self.auth_method == self.AuthMethod.PASSWORD and self.has_usable_password()
|
||||
)
|
||||
|
||||
@property
|
||||
def effective_auth_method(self) -> str:
|
||||
"""`auth_method`, falling back to OTP if no password is actually set."""
|
||||
if self.can_use_password_login():
|
||||
return self.AuthMethod.PASSWORD
|
||||
return self.AuthMethod.OTP
|
||||
|
||||
@property
|
||||
def address(self) -> str:
|
||||
parts = [part for part in [self.island_id and self.island.name, self.atoll_id and self.atoll.name] if part]
|
||||
return ", ".join(parts)
|
||||
|
||||
def approve(self, reviewer=None) -> None:
|
||||
self.status = self.Status.APPROVED
|
||||
self.rejection_reason = ""
|
||||
self.reviewed_at = timezone.now()
|
||||
self.reviewed_by = reviewer
|
||||
self.save(
|
||||
update_fields=[
|
||||
"status",
|
||||
"rejection_reason",
|
||||
"reviewed_at",
|
||||
"reviewed_by",
|
||||
"updated_at",
|
||||
]
|
||||
)
|
||||
|
||||
def reject(self, reviewer=None, reason: str = "") -> None:
|
||||
self.status = self.Status.REJECTED
|
||||
self.rejection_reason = reason
|
||||
self.reviewed_at = timezone.now()
|
||||
self.reviewed_by = reviewer
|
||||
self.save(
|
||||
update_fields=[
|
||||
"status",
|
||||
"rejection_reason",
|
||||
"reviewed_at",
|
||||
"reviewed_by",
|
||||
"updated_at",
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def _generate_code() -> str:
|
||||
return f"{secrets.randbelow(1_000_000):06d}"
|
||||
|
||||
|
||||
class OtpCodeQuerySet(models.QuerySet):
|
||||
def active(self):
|
||||
return self.filter(consumed_at__isnull=True, expires_at__gt=timezone.now())
|
||||
|
||||
|
||||
class OtpCode(models.Model):
|
||||
"""A single-use SMS code. Only the hash of the code is stored."""
|
||||
|
||||
class Purpose(models.TextChoices):
|
||||
LOGIN = "login", "Login"
|
||||
REGISTRATION = "registration", "Registration"
|
||||
|
||||
user = models.ForeignKey(
|
||||
settings.AUTH_USER_MODEL,
|
||||
on_delete=models.CASCADE,
|
||||
related_name="otp_codes",
|
||||
null=True,
|
||||
blank=True,
|
||||
)
|
||||
mobile = models.CharField(max_length=16, db_index=True)
|
||||
purpose = models.CharField(
|
||||
max_length=16, choices=Purpose.choices, default=Purpose.LOGIN
|
||||
)
|
||||
code_hash = models.CharField(max_length=128)
|
||||
attempts = models.PositiveSmallIntegerField(default=0)
|
||||
created_at = models.DateTimeField(default=timezone.now)
|
||||
expires_at = models.DateTimeField()
|
||||
consumed_at = models.DateTimeField(null=True, blank=True)
|
||||
|
||||
objects = OtpCodeQuerySet.as_manager()
|
||||
|
||||
class Meta:
|
||||
ordering = ["-created_at"]
|
||||
indexes = [models.Index(fields=["mobile", "purpose", "-created_at"])]
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.purpose} code for {self.mobile}"
|
||||
|
||||
@classmethod
|
||||
def issue(cls, mobile: str, purpose: str, user=None) -> tuple["OtpCode", str]:
|
||||
"""Invalidate any outstanding codes and return (row, plaintext code)."""
|
||||
cls.objects.filter(
|
||||
mobile=mobile, purpose=purpose, consumed_at__isnull=True
|
||||
).update(consumed_at=timezone.now())
|
||||
|
||||
code = _generate_code()
|
||||
otp = cls.objects.create(
|
||||
user=user,
|
||||
mobile=mobile,
|
||||
purpose=purpose,
|
||||
code_hash=make_password(code),
|
||||
expires_at=timezone.now() + timedelta(seconds=settings.OTP_TTL_SECONDS),
|
||||
)
|
||||
return otp, code
|
||||
|
||||
@property
|
||||
def is_expired(self) -> bool:
|
||||
return self.expires_at <= timezone.now()
|
||||
|
||||
@property
|
||||
def is_exhausted(self) -> bool:
|
||||
return self.attempts >= settings.OTP_MAX_ATTEMPTS
|
||||
|
||||
@property
|
||||
def resend_available_at(self):
|
||||
return self.created_at + timedelta(
|
||||
seconds=settings.OTP_RESEND_COOLDOWN_SECONDS
|
||||
)
|
||||
|
||||
def verify(self, code: str) -> bool:
|
||||
"""Check `code`, counting the attempt. Consumes the row on success."""
|
||||
self.attempts += 1
|
||||
if check_password(str(code), self.code_hash):
|
||||
self.consumed_at = timezone.now()
|
||||
self.save(update_fields=["attempts", "consumed_at"])
|
||||
return True
|
||||
self.save(update_fields=["attempts"])
|
||||
return False
|
||||
|
||||
|
||||
class RegistrationTicketQuerySet(models.QuerySet):
|
||||
def active(self):
|
||||
return self.filter(consumed_at__isnull=True, expires_at__gt=timezone.now())
|
||||
|
||||
|
||||
class RegistrationTicket(models.Model):
|
||||
"""Proof that a mobile number was verified by SMS, redeemable once.
|
||||
|
||||
Issued when a registration code is confirmed and required by the
|
||||
registration submit, so the form can't be posted for a number the caller
|
||||
never proved they control.
|
||||
"""
|
||||
|
||||
key = models.CharField(max_length=64, unique=True, db_index=True)
|
||||
mobile = models.CharField(max_length=16, db_index=True)
|
||||
created_at = models.DateTimeField(default=timezone.now)
|
||||
expires_at = models.DateTimeField()
|
||||
consumed_at = models.DateTimeField(null=True, blank=True)
|
||||
|
||||
objects = RegistrationTicketQuerySet.as_manager()
|
||||
|
||||
class Meta:
|
||||
ordering = ["-created_at"]
|
||||
|
||||
def __str__(self):
|
||||
return f"registration ticket for {self.mobile}"
|
||||
|
||||
@classmethod
|
||||
def issue(cls, mobile: str) -> "RegistrationTicket":
|
||||
cls.objects.filter(mobile=mobile, consumed_at__isnull=True).update(
|
||||
consumed_at=timezone.now()
|
||||
)
|
||||
return cls.objects.create(
|
||||
key=secrets.token_urlsafe(32),
|
||||
mobile=mobile,
|
||||
expires_at=timezone.now()
|
||||
+ timedelta(seconds=settings.REGISTRATION_TICKET_TTL_SECONDS),
|
||||
)
|
||||
|
||||
def consume(self) -> None:
|
||||
self.consumed_at = timezone.now()
|
||||
self.save(update_fields=["consumed_at"])
|
||||
Reference in New Issue
Block a user