first

apps/accounts/models.py

@@ -0,0 +1,157 @@
+import hashlib, secrets, string
+from datetime import timedelta
+from django.conf import settings
+from django.contrib.auth.models import AbstractUser
+from django.db import IntegrityError, models, transaction
+from django.utils import timezone
+from apps.core.models import PublicIdModel, TimeStampedModel
+
+def normalize_mobile(value: str) -> str:
+    value = "".join(c for c in (value or "").strip() if c.isdigit() or c == "+")
+    if value.startswith("0098"): value = "+98" + value[4:]
+    elif value.startswith("98") and not value.startswith("+98"): value = "+" + value
+    elif value.startswith("0"): value = "+98" + value[1:]
+    if not (value.startswith("+989") and len(value) == 13 and value[1:].isdigit()):
+        raise ValueError("شماره موبایل معتبر نیست.")
+    return value
+
+class User(PublicIdModel, AbstractUser):
+    class AccountType(models.TextChoices):
+        CUSTOMER = "CUSTOMER", "مشتری"
+        STAFF = "STAFF", "کارشناس"
+    mobile = models.CharField(max_length=16, unique=True, null=True, blank=True)
+    account_type = models.CharField(max_length=10, choices=AccountType.choices, default=AccountType.CUSTOMER)
+    def save(self, *args, **kwargs):
+        if self.mobile: self.mobile = normalize_mobile(self.mobile)
+        super().save(*args, **kwargs)
+    @property
+    def local_mobile(self):
+        return f"0{self.mobile[3:]}" if self.mobile and self.mobile.startswith("+98") else (self.mobile or "")
+
+class OTPChallenge(models.Model):
+    class Purpose(models.TextChoices): LOGIN = "LOGIN", "ورود"
+    mobile = models.CharField(max_length=16, db_index=True)
+    purpose = models.CharField(max_length=20, choices=Purpose.choices, default=Purpose.LOGIN)
+    code_hash = models.CharField(max_length=64)
+    expires_at = models.DateTimeField()
+    attempt_count = models.PositiveSmallIntegerField(default=0)
+    max_attempts = models.PositiveSmallIntegerField(default=5)
+    verified_at = models.DateTimeField(null=True, blank=True)
+    created_at = models.DateTimeField(auto_now_add=True, db_index=True)
+    request_ip = models.GenericIPAddressField(null=True, blank=True)
+    user_agent = models.CharField(max_length=300, blank=True)
+    @staticmethod
+    def hash_code(code): return hashlib.sha256(f"{settings.SECRET_KEY}:{code}".encode()).hexdigest()
+    @classmethod
+    def issue(cls, mobile, request=None):
+        mobile = normalize_mobile(mobile)
+        request_ip = request.META.get("REMOTE_ADDR") if request else None
+        window = timezone.now() - timedelta(seconds=settings.OTP_RATE_LIMIT_WINDOW_SECONDS)
+        if cls.objects.filter(mobile=mobile, created_at__gte=window).count() >= settings.OTP_RATE_LIMIT_PER_MOBILE:
+            raise ValueError("تعداد درخواست کد برای این شماره بیش از حد مجاز است.")
+        if request_ip and cls.objects.filter(request_ip=request_ip, created_at__gte=window).count() >= settings.OTP_RATE_LIMIT_PER_IP:
+            raise ValueError("تعداد درخواست کد از این اتصال بیش از حد مجاز است.")
+        cutoff = timezone.now() - timedelta(seconds=settings.OTP_RESEND_SECONDS)
+        if cls.objects.filter(mobile=mobile, created_at__gt=cutoff).exists(): raise ValueError("برای ارسال مجدد کمی صبر کنید.")
+        is_dev_login = settings.DEBUG and mobile == settings.DEV_OTP_MOBILE and bool(settings.DEV_OTP_CODE)
+        code = settings.DEV_OTP_CODE if is_dev_login else "".join(secrets.choice("0123456789") for _ in range(settings.OTP_LENGTH))
+        obj = cls.objects.create(mobile=mobile, code_hash=cls.hash_code(code), expires_at=timezone.now()+timedelta(seconds=settings.OTP_EXPIRE_SECONDS), max_attempts=settings.OTP_MAX_ATTEMPTS, request_ip=request_ip, user_agent=(request.META.get("HTTP_USER_AGENT", "")[:300] if request else ""))
+        return obj, code
+    def verify(self, code):
+        if self.verified_at or timezone.now() >= self.expires_at or self.attempt_count >= self.max_attempts: return False
+        self.attempt_count += 1
+        ok = secrets.compare_digest(self.code_hash, self.hash_code(code))
+        if ok: self.verified_at = timezone.now()
+        self.save(update_fields=["attempt_count", "verified_at"])
+        return ok
+
+class TicketMagicLink(models.Model):
+    ticket = models.ForeignKey("tickets.Ticket", on_delete=models.CASCADE, related_name="magic_links")
+    customer = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
+    token_hash = models.CharField(max_length=64, unique=True)
+    created_by = models.ForeignKey(
+        settings.AUTH_USER_MODEL,
+        on_delete=models.PROTECT,
+        null=True,
+        blank=True,
+        related_name="created_magic_links",
+    )
+    created_unit = models.ForeignKey(
+        "staff.OrganizationalUnit",
+        on_delete=models.PROTECT,
+        null=True,
+        blank=True,
+        related_name="magic_links",
+    )
+    recipient_name = models.CharField(max_length=200, blank=True)
+    recipient_mobile = models.CharField(max_length=16, blank=True)
+    created_at = models.DateTimeField(auto_now_add=True)
+    expires_at = models.DateTimeField()
+    used_at = models.DateTimeField(null=True, blank=True)
+    revoked_at = models.DateTimeField(null=True, blank=True)
+    created_reason = models.CharField(max_length=100, blank=True)
+    last_used_ip = models.GenericIPAddressField(null=True, blank=True)
+    last_user_agent = models.CharField(max_length=500, blank=True)
+    view_count = models.PositiveIntegerField(default=0)
+    class Meta:
+        permissions = [("view_all_magic_links", "Can audit magic links across all units")]
+        indexes = [models.Index(fields=["created_unit", "created_at"], name="magic_unit_created_idx"), models.Index(fields=["expires_at", "revoked_at"], name="magic_validity_idx")]
+    @staticmethod
+    def digest(token): return hashlib.sha256(token.encode()).hexdigest()
+    @staticmethod
+    def generate_secure_token(length=10):
+        alphabet=string.ascii_lowercase+string.digits
+        return "".join(secrets.SystemRandom().choices(alphabet,k=length))
+    @classmethod
+    def issue_with_record(cls,ticket,reason="staff_reply",max_retries=5,created_by=None):
+        # Never add a preflight SELECT to token issuance; callers may select_related
+        # the profile when they need a populated recipient-name snapshot.
+        profile = ticket.customer._state.fields_cache.get("customer_profile")
+        recipient_name = " ".join(
+            value for value in [getattr(profile, "first_name", ""), getattr(profile, "last_name", "")] if value
+        ).strip()
+        for attempt in range(max_retries):
+            token=cls.generate_secure_token(10)
+            try:
+                with transaction.atomic():
+                    record=cls.objects.create(
+                        ticket=ticket,
+                        customer=ticket.customer,
+                        token_hash=cls.digest(token),
+                        expires_at=timezone.now()+timedelta(days=settings.MAGIC_LINK_TTL_DAYS),
+                        created_reason=reason,
+                        created_by=created_by or ticket.created_by,
+                        created_unit=ticket.owning_unit,
+                        recipient_name=recipient_name,
+                        recipient_mobile=ticket.customer.mobile or "",
+                    )
+                return token,record
+            except IntegrityError:
+                if attempt==max_retries-1: raise RuntimeError("خطای نادر: عدم موفقیت در تولید توکن یکتا.")
+        raise RuntimeError("خطای نادر: عدم موفقیت در تولید توکن یکتا.")
+    @classmethod
+    def issue(cls,ticket,reason="staff_reply"):
+        token,_=cls.issue_with_record(ticket,reason)
+        return token
+
+
+class MagicLinkAccessLog(models.Model):
+    """Immutable audit row for every valid magic-link opening."""
+
+    magic_link = models.ForeignKey(
+        TicketMagicLink,
+        on_delete=models.CASCADE,
+        related_name="access_logs",
+    )
+    opened_at = models.DateTimeField(auto_now_add=True, db_index=True)
+    ip_address = models.GenericIPAddressField(null=True, blank=True)
+    user_agent = models.CharField(max_length=500, blank=True)
+
+    class Meta:
+        ordering = ["-opened_at"]
+        indexes = [models.Index(fields=["magic_link", "opened_at"], name="magic_opened_idx")]
+        verbose_name = "بازدید لینک جادویی"
+        verbose_name_plural = "بازدیدهای لینک جادویی"
+
+    def __str__(self):
+        return f"{self.magic_link_id} @ {self.opened_at}"