82 lines
3.1 KiB
Python
82 lines
3.1 KiB
Python
"""Restrictions for full customer logins; staff and scoped links stay independent."""
|
|
|
|
from django.contrib import messages
|
|
from django.contrib.auth import logout
|
|
from django.contrib.auth.signals import user_logged_in
|
|
from django.dispatch import receiver
|
|
from django.http import JsonResponse
|
|
from django.shortcuts import redirect
|
|
from django.urls import reverse
|
|
from django.views.decorators.cache import never_cache
|
|
from django.views.decorators.http import require_GET
|
|
from django.utils.cache import add_never_cache_headers
|
|
|
|
from apps.customers.models import CustomerProfile
|
|
from .models import User
|
|
|
|
|
|
ACCESS_VERSION_KEY = "customer_login_access_version"
|
|
BLOCK_REASON_KEY = "customer_login_block_reason"
|
|
DEFAULT_BLOCK_REASON = "ورود شما مسدود شده است. برای پیگیری با پشتیبانی تماس بگیرید."
|
|
|
|
|
|
def is_customer_login(user):
|
|
return user.account_type == User.AccountType.CUSTOMER and not (user.is_staff or user.is_superuser)
|
|
|
|
|
|
def customer_login_profile(user):
|
|
if not is_customer_login(user):
|
|
return None
|
|
return CustomerProfile.objects.filter(user_id=user.pk).first()
|
|
|
|
|
|
def blocked_login_reason(mobile):
|
|
user = User.objects.filter(mobile=mobile).first()
|
|
profile = customer_login_profile(user) if user else None
|
|
if profile and profile.login_blocked:
|
|
return profile.login_block_reason.strip() or DEFAULT_BLOCK_REASON
|
|
return ""
|
|
|
|
|
|
@receiver(user_logged_in, dispatch_uid="customer_login_access_version")
|
|
def remember_customer_access(sender, request, user, **kwargs):
|
|
if not is_customer_login(user):
|
|
return
|
|
profile = customer_login_profile(user)
|
|
request.session[ACCESS_VERSION_KEY] = profile.login_access_version if profile else 0
|
|
request.session.pop(BLOCK_REASON_KEY, None)
|
|
|
|
|
|
class CustomerLoginAccessMiddleware:
|
|
def __init__(self, get_response):
|
|
self.get_response = get_response
|
|
|
|
def __call__(self, request):
|
|
if request.user.is_authenticated:
|
|
profile = customer_login_profile(request.user)
|
|
if profile and (
|
|
profile.login_blocked
|
|
or request.session.get(ACCESS_VERSION_KEY, 0) != profile.login_access_version
|
|
):
|
|
logout(request)
|
|
if profile.login_blocked:
|
|
request.session[BLOCK_REASON_KEY] = profile.login_block_reason.strip() or DEFAULT_BLOCK_REASON
|
|
else:
|
|
messages.error(request, "نشست قبلی شما باطل شده است. لطفاً دوباره وارد شوید.")
|
|
if request.path == reverse("accounts:login_access"):
|
|
response = JsonResponse({"authenticated": False, "login_url": reverse("accounts:login")})
|
|
else:
|
|
response = redirect("accounts:login")
|
|
add_never_cache_headers(response)
|
|
return response
|
|
return self.get_response(request)
|
|
|
|
|
|
@never_cache
|
|
@require_GET
|
|
def customer_login_access(request):
|
|
return JsonResponse({
|
|
"authenticated": request.user.is_authenticated,
|
|
"login_url": reverse("accounts:login"),
|
|
})
|