1from functools import cached_property
  2from typing import TYPE_CHECKING, Any
  3from urllib.parse import urlparse, urlunparse
  4
  5from plain.http import (
  6    ForbiddenError403,
  7    HTTPException,
  8    NotFoundError404,
  9    QueryDict,
 10    RedirectResponse,
 11    Response,
 12)
 13from plain.runtime import settings
 14from plain.sessions.views import SessionView
 15from plain.urls import reverse
 16from plain.utils.cache import patch_cache_control
 17from plain.views import View
 18
 19from .sessions import logout
 20from .utils import resolve_url
 21
 22if TYPE_CHECKING:
 23    from app.users.models import User
 24
 25try:
 26    from plain.admin.impersonate import get_request_impersonator
 27except ImportError:
 28    get_request_impersonator: Any = None
 29
 30__all__ = [
 31    "AuthView",
 32    "LoginRequired",
 33    "LogoutView",
 34    "redirect_to_login",
 35]
 36
 37
 38class LoginRequired(HTTPException):
 39    """Raised by `check_auth` when a view requires a logged-in user.
 40
 41    Subclasses an HTTPException so generic handlers (logging, APIs, MCP)
 42    treat it as a 401 by default. HTML views rely on `AuthView.handle_exception`
 43    to render a redirect to the configured login page instead.
 44    """
 45
 46    status_code = 401
 47
 48    def __init__(self, login_url: str | None, redirect_field_name: str = "next"):
 49        # Caller is responsible for resolving `login_url` — pass `None`
 50        # to signal "no login page configured, render as 403 instead".
 51        self.login_url = login_url
 52        self.redirect_field_name = redirect_field_name
 53
 54
 55class AuthView(SessionView):
 56    login_required = False
 57    admin_required = False  # Implies login_required
 58    login_url = settings.AUTH_LOGIN_URL
 59
 60    @cached_property
 61    def user(self) -> User | None:
 62        """Get the authenticated user for this request."""
 63        from .requests import get_request_user
 64
 65        return get_request_user(self.request)
 66
 67    def get_template_context(self) -> dict:
 68        """Add user and impersonator to template context."""
 69        context = super().get_template_context()
 70        context["user"] = self.user
 71        return context
 72
 73    def check_auth(self) -> None:
 74        """Raise LoginRequired, ForbiddenError403, or NotFoundError404 when access is denied."""
 75        if not self.login_required and not self.admin_required:
 76            return
 77
 78        if not self.user:
 79            raise LoginRequired(login_url=self.login_url)
 80
 81        if self.admin_required:
 82            # At this point, we know user is authenticated (from check above)
 83            # Check if impersonation is active
 84            if get_request_impersonator and (
 85                impersonator := get_request_impersonator(self.request)
 86            ):
 87                # Impersonators should be able to view admin pages while impersonating.
 88                # There's probably never a case where an impersonator isn't admin, but it can be configured.
 89                if not impersonator.is_admin:
 90                    raise ForbiddenError403(
 91                        "You do not have permission to access this page."
 92                    )
 93                return
 94
 95            if not self.user.is_admin:
 96                # Show a 404 so we don't expose admin urls to non-admin users
 97                raise NotFoundError404()
 98
 99    def before_request(self) -> None:
100        self.check_auth()
101
102    def handle_exception(self, exc: Exception) -> Response:
103        if isinstance(exc, LoginRequired):
104            if not exc.login_url:
105                # No configured login page — treat as a plain 403 and let
106                # the surrounding view's `handle_exception` chain render it.
107                return super().handle_exception(ForbiddenError403("Login required"))
108
109            path = self.request.build_absolute_uri()
110            resolved_login_url = reverse(exc.login_url)
111            # If the login url is the same scheme and net location then
112            # use the path as the "next" url.
113            login_scheme, login_netloc = urlparse(resolved_login_url)[:2]
114            current_scheme, current_netloc = urlparse(path)[:2]
115            if (not login_scheme or login_scheme == current_scheme) and (
116                not login_netloc or login_netloc == current_netloc
117            ):
118                path = self.request.get_full_path()
119            return redirect_to_login(
120                path,
121                resolved_login_url,
122                exc.redirect_field_name,
123            )
124        return super().handle_exception(exc)
125
126    def after_response(self, response: Response) -> Response:
127        response = super().after_response(response)
128        if self.user:
129            # Make sure it at least has private as a default
130            patch_cache_control(response, private=True)
131        return response
132
133
134class LogoutView(View):
135    def post(self) -> RedirectResponse:
136        logout(self.request)
137        return RedirectResponse("/", status_code=302)
138
139
140def redirect_to_login(
141    next: str, login_url: str | None = None, redirect_field_name: str = "next"
142) -> RedirectResponse:
143    """
144    Redirect the user to the login page, passing the given 'next' page.
145    """
146    resolved_url = resolve_url(login_url or settings.AUTH_LOGIN_URL)
147
148    login_url_parts = list(urlparse(resolved_url))
149    if redirect_field_name:
150        querystring = QueryDict(login_url_parts[4], mutable=True)
151        querystring[redirect_field_name] = next
152        login_url_parts[4] = querystring.urlencode(safe="/")
153
154    return RedirectResponse(
155        str(urlunparse(login_url_parts)), status_code=302, allow_external=True
156    )