1from __future__ import annotations
  2
  3from functools import cached_property
  4from typing import TYPE_CHECKING, Any
  5from urllib.parse import urlparse, urlunparse
  6
  7from plain.exceptions import PermissionDenied
  8from plain.http import (
  9    Http404,
 10    QueryDict,
 11    Response,
 12    ResponseRedirect,
 13)
 14from plain.runtime import settings
 15from plain.sessions.views import SessionViewMixin
 16from plain.urls import reverse
 17from plain.utils.cache import patch_cache_control
 18from plain.views import View
 19
 20from .sessions import logout
 21from .utils import resolve_url
 22
 23if TYPE_CHECKING:
 24    from plain.http import Request
 25
 26
 27class LoginRequired(Exception):
 28    def __init__(self, login_url: str | None = None, redirect_field_name: str = "next"):
 29        self.login_url = login_url or settings.AUTH_LOGIN_URL
 30        self.redirect_field_name = redirect_field_name
 31
 32
 33class AuthViewMixin(SessionViewMixin):
 34    login_required = False
 35    admin_required = False  # Implies login_required
 36    login_url = settings.AUTH_LOGIN_URL
 37
 38    request: Request
 39
 40    @cached_property
 41    def user(self) -> Any | None:
 42        """Get the authenticated user for this request."""
 43        from .requests import get_request_user
 44
 45        return get_request_user(self.request)
 46
 47    def get_template_context(self) -> dict:
 48        """Add user and impersonator to template context."""
 49        context = super().get_template_context()
 50        context["user"] = self.user
 51        return context
 52
 53    def check_auth(self) -> None:
 54        """
 55        Raises either LoginRequired or PermissionDenied.
 56        - LoginRequired can specify a login_url and redirect_field_name
 57        - PermissionDenied can specify a message
 58        """
 59        if not self.login_required and not self.admin_required:
 60            return None
 61
 62        if not self.user:
 63            raise LoginRequired(login_url=self.login_url)
 64
 65        if self.admin_required:
 66            # At this point, we know user is authenticated (from check above)
 67            # Check if impersonation is active
 68            if impersonator := getattr(self, "impersonator", None):
 69                # Impersonators should be able to view admin pages while impersonating.
 70                # There's probably never a case where an impersonator isn't admin, but it can be configured.
 71                if not impersonator.is_admin:
 72                    raise PermissionDenied(
 73                        "You do not have permission to access this page."
 74                    )
 75            elif not self.user.is_admin:
 76                # Show a 404 so we don't expose admin urls to non-admin users
 77                raise Http404()
 78
 79    def get_response(self) -> Response:
 80        try:
 81            self.check_auth()
 82        except LoginRequired as e:
 83            if self.login_url:
 84                # Ideally this could be handled elsewhere... like PermissionDenied
 85                # also seems like this code is used multiple places anyway...
 86                # could be easier to get redirect query param
 87                path = self.request.build_absolute_uri()
 88                resolved_login_url = reverse(e.login_url)
 89                # If the login url is the same scheme and net location then use the
 90                # path as the "next" url.
 91                login_scheme, login_netloc = urlparse(resolved_login_url)[:2]
 92                current_scheme, current_netloc = urlparse(path)[:2]
 93                if (not login_scheme or login_scheme == current_scheme) and (
 94                    not login_netloc or login_netloc == current_netloc
 95                ):
 96                    path = self.request.get_full_path()
 97                return redirect_to_login(
 98                    path,
 99                    resolved_login_url,
100                    e.redirect_field_name,
101                )
102            else:
103                raise PermissionDenied("Login required")
104
105        response = super().get_response()
106
107        if self.user:
108            # Make sure it at least has private as a default
109            patch_cache_control(response, private=True)
110
111        return response
112
113
114class LogoutView(View):
115    def post(self) -> ResponseRedirect:
116        logout(self.request)
117        return ResponseRedirect("/")
118
119
120def redirect_to_login(
121    next: str, login_url: str | None = None, redirect_field_name: str = "next"
122) -> ResponseRedirect:
123    """
124    Redirect the user to the login page, passing the given 'next' page.
125    """
126    resolved_url = resolve_url(login_url or settings.AUTH_LOGIN_URL)
127
128    login_url_parts = list(urlparse(resolved_url))
129    if redirect_field_name:
130        querystring = QueryDict(login_url_parts[4], mutable=True)
131        querystring[redirect_field_name] = next
132        login_url_parts[4] = querystring.urlencode(safe="/")
133
134    return ResponseRedirect(str(urlunparse(login_url_parts)))