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"])