"""Deployment helpers (subfolder URL prefix, staff routing).""" import logging import re from urllib.parse import urlencode from django.conf import settings from django.shortcuts import redirect, render from mainapp.dashboard_acl import resolve_post_login_url from mainapp.utils.locale_path import collapse_recursive_locale_path from mainapp.utils.rate_limit import is_rate_limited from mysite.subpath_utils import resolve_site_subpath _LEGACY_DASHBOARD_RE = re.compile(r"^/(?:en|ar)(/admin/?.*)$", re.IGNORECASE) _SECURITY_LOG = logging.getLogger("security") _SPIDER_TRAP_SKIP_PREFIXES = ( "/admin/", "/accounts/", "/django-admin/", "/i18n/", "/media/", "/static/", "/tinymce/", ) class StripDeploySubpathMiddleware: """ When a deploy subpath is resolved (see ``resolve_site_subpath()``), strip it from ``PATH_INFO`` so URLconf matches as if the app lived at domain root. Use this with ``FORCE_SCRIPT_NAME`` / ``SESSION_COOKIE_PATH`` in settings so ``reverse()`` and cookies stay under the subfolder. Passenger/cPanel often pass the full path to Django; wrapping ``urlpatterns`` in ``path('/', ...)`` is brittle when ``PATH_INFO`` is exactly ``//``. Stripping here matches how reverse proxies usually behave. """ def __init__(self, get_response): self.get_response = get_response def __call__(self, request): sub = resolve_site_subpath() if not sub: return self.get_response(request) prefix = f"/{sub}" path_info = request.META.get("PATH_INFO", "") or "" if path_info == prefix or path_info.startswith(prefix + "/"): new_pi = path_info[len(prefix) :] or "/" request.META["PATH_INFO"] = new_pi return self.get_response(request) class NormalizeScriptNameMiddleware: """ Passenger/cPanel sometimes set ``SCRIPT_NAME`` to the current page directory for Unicode slugs, which breaks ``reverse()`` / ``{% url %}`` (recursive internal links). Always align ``SCRIPT_NAME`` with ``settings.FORCE_SCRIPT_NAME`` (empty at domain root). """ def __init__(self, get_response): self.get_response = get_response def __call__(self, request): forced = getattr(settings, "FORCE_SCRIPT_NAME", None) request.META["SCRIPT_NAME"] = forced if forced else "" return self.get_response(request) class SpiderTrapRedirectMiddleware: """301 redirect duplicated locale paths (SEO spider traps) to their canonical form.""" def __init__(self, get_response): self.get_response = get_response def __call__(self, request): path = request.META.get("PATH_INFO", "") or "/" if any(path.startswith(prefix) for prefix in _SPIDER_TRAP_SKIP_PREFIXES): return self.get_response(request) collapsed = collapse_recursive_locale_path(path) if not collapsed: return self.get_response(request) target = collapsed forced = getattr(settings, "FORCE_SCRIPT_NAME", None) or "" if forced and not target.startswith(forced): target = f"{forced}{target}" query = request.META.get("QUERY_STRING", "") if query: target = f"{target}?{query}" return redirect(target, permanent=True) class StaffDashboardRoutingMiddleware: """ Paython dashboard lives at ``/admin/`` (see ``mainapp.urls_dashboard``). - ``/en/admin/`` → ``/admin/`` (legacy bookmarks) - ``/django-admin/`` → login or Paython dashboard (django-admin is optional/disabled) """ def __init__(self, get_response): self.get_response = get_response def __call__(self, request): path = request.META.get("PATH_INFO", "") or "" legacy = _LEGACY_DASHBOARD_RE.match(path) if legacy: return redirect(legacy.group(1) or "/admin/") if not path.startswith("/django-admin"): return self.get_response(request) django_admin_on = getattr(settings, "DJANGO_ADMIN_ENABLED", True) target = ( resolve_post_login_url(request.user) if request.user.is_authenticated else "/admin/" ) if not django_admin_on or not ( request.user.is_authenticated and request.user.is_superuser ): if path.startswith("/django-admin/login") or not request.user.is_authenticated: query = urlencode({"next": target}) return redirect(f"/accounts/login/?{query}") return redirect(target) return self.get_response(request) class TwoFactorLoginRateLimitMiddleware: """Rate-limit POSTs to django-two-factor login endpoint.""" def __init__(self, get_response): self.get_response = get_response def __call__(self, request): path = request.META.get("PATH_INFO", "") or "" if path == "/account/login/" and request.method == "POST": if is_rate_limited( request, scope="login", limit=int(getattr(settings, "LOGIN_RATE_LIMIT_MAX", 10)), window_seconds=int(getattr(settings, "LOGIN_RATE_LIMIT_WINDOW", 900)), ): _SECURITY_LOG.warning( "Rate-limited login attempt ip=%s username=%s path=%s", self._client_ip(request), (request.POST.get("auth-username") or request.POST.get("username") or "").strip(), path, ) return render( request, "errors/429.html", status=429, context={"retry_after_seconds": int(getattr(settings, "LOGIN_RATE_LIMIT_WINDOW", 900))}, ) return self.get_response(request) def _client_ip(self, request) -> str: if getattr(settings, "TRUST_PROXY_FOR_CLIENT_IP", False): forwarded = (request.META.get("HTTP_X_FORWARDED_FOR") or "").split(",")[0].strip() if forwarded: return forwarded return request.META.get("REMOTE_ADDR") or "unknown" class StaffOtpEnforceMiddleware: """Require OTP-verified sessions for staff dashboard paths (/admin/*).""" def __init__(self, get_response): self.get_response = get_response def __call__(self, request): if not getattr(settings, "TWO_FACTOR_ENFORCE_STAFF", False): return self.get_response(request) path = request.META.get("PATH_INFO", "") or "" if not path.startswith("/admin/"): return self.get_response(request) user = getattr(request, "user", None) if not getattr(user, "is_authenticated", False): return self.get_response(request) if not getattr(user, "is_staff", False): return self.get_response(request) verified_callable = getattr(user, "is_verified", None) is_verified = bool(verified_callable()) if callable(verified_callable) else False if is_verified: return self.get_response(request) try: from django_otp import user_has_device has_device = bool(user_has_device(user)) except Exception: has_device = False target = "/account/login/" if has_device else "/account/two_factor/setup/" next_value = request.get_full_path() or "/admin/" query = urlencode({"next": next_value}) return redirect(f"{target}?{query}")