158 lines
8.4 KiB
Python
158 lines
8.4 KiB
Python
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}"
|