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