195 lines
10 KiB
Python
195 lines
10 KiB
Python
from django.contrib import messages
|
|
from django.conf import settings
|
|
from django.contrib.auth import login, logout
|
|
from django.shortcuts import get_object_or_404, redirect, render
|
|
from django.utils.http import url_has_allowed_host_and_scheme
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
from django.views.decorators.http import require_GET, require_http_methods, require_POST
|
|
from django.views.decorators.cache import never_cache
|
|
from .forms import CustomerPasswordLoginForm, CustomerProfileForm, MobileForm, OTPForm, StaffLoginForm
|
|
from apps.customers.models import CustomerProfile
|
|
from django.contrib.auth.decorators import login_required
|
|
from django.db import transaction
|
|
from django.db.models import F
|
|
from .models import MagicLinkAccessLog, OTPChallenge, TicketMagicLink, User
|
|
from .magic_access import grant_ticket_magic_access
|
|
from .customer_access import BLOCK_REASON_KEY, blocked_login_reason
|
|
from apps.integrations.sms.providers import SMSResult, get_sms_provider
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def _remember_destination(request):
|
|
target = request.GET.get("next", "")
|
|
if target and target.startswith(("/chaharsu/", "/tickets/")) and url_has_allowed_host_and_scheme(target, allowed_hosts={request.get_host()}, require_https=request.is_secure()):
|
|
request.session["next"] = target
|
|
|
|
|
|
def _login_destination(request):
|
|
target = request.session.pop("next", "")
|
|
if target.startswith(("/chaharsu/", "/tickets/")) and url_has_allowed_host_and_scheme(target, allowed_hosts={request.get_host()}, require_https=request.is_secure()):
|
|
return target
|
|
return "staff_tickets:list" if request.user.account_type == User.AccountType.STAFF else "payments:home"
|
|
|
|
|
|
def _authenticated_home(request):
|
|
if not request.user.is_authenticated: return None
|
|
return redirect("staff_tickets:list" if request.user.account_type==User.AccountType.STAFF else "payments:home")
|
|
|
|
@require_GET
|
|
def home(request):
|
|
"""Send every visitor to the correct login or authenticated workspace."""
|
|
return _authenticated_home(request) or redirect("accounts:login")
|
|
|
|
@require_http_methods(["GET", "POST"])
|
|
def customer_login(request):
|
|
_remember_destination(request)
|
|
if response:=_authenticated_home(request): return response
|
|
form = MobileForm(request.POST or None)
|
|
block_reason = request.session.pop(BLOCK_REASON_KEY, "")
|
|
if request.method == "POST" and form.is_valid():
|
|
block_reason = blocked_login_reason(form.cleaned_data["mobile"])
|
|
if block_reason:
|
|
return render(request, "accounts/customer_login.html", {"form": form, "block_reason": block_reason})
|
|
challenge = None
|
|
try:
|
|
challenge, code = OTPChallenge.issue(form.cleaned_data["mobile"], request)
|
|
is_fixed_dev_otp = settings.DEBUG and challenge.mobile == settings.DEV_OTP_MOBILE and code == settings.DEV_OTP_CODE
|
|
result = SMSResult(True, "fixed-dev-otp") if is_fixed_dev_otp else get_sms_provider().send_otp(challenge.mobile, code)
|
|
if not result.success:
|
|
raise RuntimeError(result.error or "سرویس پیامک درخواست را نپذیرفت.")
|
|
request.session["otp_challenge_id"] = challenge.pk
|
|
return redirect("accounts:otp")
|
|
except ValueError as exc: form.add_error(None, str(exc))
|
|
except Exception:
|
|
if challenge is not None:
|
|
challenge.delete()
|
|
logger.exception("OTP SMS delivery failed")
|
|
form.add_error(None, "ارسال پیامک ورود در حال حاضر ممکن نیست؛ لطفاً کمی بعد دوباره تلاش کنید.")
|
|
return render(request, "accounts/customer_login.html", {"form": form, "block_reason": block_reason})
|
|
|
|
@require_http_methods(["GET", "POST"])
|
|
def verify_otp(request):
|
|
if response:=_authenticated_home(request): return response
|
|
challenge = get_object_or_404(OTPChallenge, pk=request.session.get("otp_challenge_id"))
|
|
form = OTPForm(request.POST or None)
|
|
block_reason = blocked_login_reason(challenge.mobile)
|
|
if block_reason:
|
|
request.session.pop("otp_challenge_id", None)
|
|
request.session[BLOCK_REASON_KEY] = block_reason
|
|
return redirect("accounts:login")
|
|
if request.method == "POST" and form.is_valid():
|
|
if challenge.verify(form.cleaned_data["code"]):
|
|
user, created = User.objects.get_or_create(mobile=challenge.mobile, defaults={"username": f"customer_{challenge.mobile[3:]}", "account_type": User.AccountType.CUSTOMER})
|
|
if created: user.set_unusable_password(); user.save(update_fields=["password"])
|
|
CustomerProfile.objects.get_or_create(user=user)
|
|
if not user.is_active: form.add_error(None, "حساب شما غیرفعال است.")
|
|
else:
|
|
login(request, user, backend="django.contrib.auth.backends.ModelBackend")
|
|
return redirect(_login_destination(request))
|
|
else: form.add_error("code", "کد نامعتبر یا منقضی است.")
|
|
return render(request, "accounts/otp.html", {"form": form, "mobile": challenge.mobile})
|
|
|
|
@require_http_methods(["GET","POST"])
|
|
def customer_password_login(request):
|
|
_remember_destination(request)
|
|
if response:=_authenticated_home(request): return response
|
|
form=CustomerPasswordLoginForm(request.POST or None)
|
|
block_reason = request.session.pop(BLOCK_REASON_KEY, "")
|
|
if request.method=="POST" and form.is_valid():
|
|
block_reason = blocked_login_reason(form.cleaned_data["mobile"])
|
|
if block_reason:
|
|
return render(request, "accounts/customer_password_login.html", {"form": form, "block_reason": block_reason})
|
|
user=form.authenticate(request)
|
|
if user and user.is_active:
|
|
login(request,user); return redirect(_login_destination(request))
|
|
form.add_error(None,"شماره موبایل یا رمز عبور صحیح نیست.")
|
|
return render(request,"accounts/customer_password_login.html",{"form":form,"block_reason":block_reason})
|
|
|
|
@login_required
|
|
@require_http_methods(["GET","POST"])
|
|
def customer_profile(request):
|
|
if request.user.account_type!=User.AccountType.CUSTOMER: return redirect("staff_tickets:list")
|
|
profile,_=CustomerProfile.objects.get_or_create(user=request.user)
|
|
form=CustomerProfileForm(request.POST or None,instance=profile)
|
|
if request.method=="POST" and form.is_valid():
|
|
form.save(); password=form.cleaned_data.get("password")
|
|
if password: request.user.set_password(password); request.user.save(update_fields=["password"]); login(request,request.user,backend="django.contrib.auth.backends.ModelBackend")
|
|
messages.success(request,"پروفایل شما ذخیره شد."); return redirect("accounts:profile")
|
|
return render(request,"accounts/profile.html",{"form":form})
|
|
|
|
def _staff_login_destination(request):
|
|
target = request.GET.get("next") or request.POST.get("next") or request.session.get("staff_login_next", "")
|
|
if target.startswith(("/vt/", "/staff/tickets/")) and url_has_allowed_host_and_scheme(
|
|
target, allowed_hosts={request.get_host()}, require_https=request.is_secure(),
|
|
):
|
|
request.session["staff_login_next"] = target
|
|
return target
|
|
request.session.pop("staff_login_next", None)
|
|
return reverse("staff_tickets:list")
|
|
|
|
|
|
@require_http_methods(["GET", "POST"])
|
|
def staff_login(request):
|
|
destination = _staff_login_destination(request)
|
|
if response:=_authenticated_home(request): return response
|
|
form = StaffLoginForm(request.POST or None)
|
|
if request.method == "POST" and form.is_valid():
|
|
user = form.authenticate(request)
|
|
if user and user.is_active and user.account_type == User.AccountType.STAFF:
|
|
login(request, user)
|
|
request.session.pop("staff_login_next", None)
|
|
return redirect(destination)
|
|
form.add_error(None, "اطلاعات ورود صحیح نیست.")
|
|
return render(request, "accounts/staff_login.html", {"form": form})
|
|
|
|
@require_POST
|
|
def logout_view(request):
|
|
logout(request)
|
|
request.session.flush()
|
|
response=redirect("accounts:login")
|
|
response["Cache-Control"]="no-store"
|
|
return response
|
|
|
|
@never_cache
|
|
@require_http_methods(["GET", "POST"])
|
|
def consume_magic_link(request, token):
|
|
link = TicketMagicLink.objects.select_related("customer", "ticket").filter(token_hash=TicketMagicLink.digest(token)).first()
|
|
if not link or link.revoked_at or link.expires_at <= timezone.now() or not link.customer.is_active:
|
|
request.session["next"] = f"/tickets/{link.ticket.public_id}/" if link else "/tickets/"
|
|
messages.error(request, "این لینک منقضی شده یا اشتباه است. لطفاً با شماره موبایل خود وارد شوید تا بتوانید پیامها را ببینید.")
|
|
response = redirect("accounts:login")
|
|
response["Referrer-Policy"] = "no-referrer"
|
|
return response
|
|
if request.user.is_authenticated and request.user.pk != link.customer_id:
|
|
confirmed = (
|
|
request.method == "POST"
|
|
and request.POST.get("action") == "logout_and_open"
|
|
and request.POST.get("account_id") == str(request.user.pk)
|
|
)
|
|
if not confirmed:
|
|
response = render(request, "accounts/magic_link_confirmation.html")
|
|
# Same-origin form submissions need an Origin for Django CSRF checks.
|
|
response["Referrer-Policy"] = "same-origin"
|
|
return response
|
|
# Logout flushes the old session; grant access only in the new one.
|
|
logout(request)
|
|
if not request.user.is_authenticated:
|
|
grant_ticket_magic_access(request, link)
|
|
ip_address = request.META.get("REMOTE_ADDR")
|
|
user_agent = request.META.get("HTTP_USER_AGENT", "")[:500]
|
|
with transaction.atomic():
|
|
MagicLinkAccessLog.objects.create(
|
|
magic_link=link,
|
|
ip_address=ip_address,
|
|
user_agent=user_agent,
|
|
)
|
|
TicketMagicLink.objects.filter(pk=link.pk).update(
|
|
used_at=timezone.now(),
|
|
last_used_ip=ip_address,
|
|
last_user_agent=user_agent,
|
|
view_count=F("view_count") + 1,
|
|
)
|
|
response = redirect("tickets:detail", public_id=link.ticket.public_id); response["Referrer-Policy"] = "no-referrer"; return response
|