Auth

class BaseAuthProvider(ABC):
    """
    Minimal contract every auth provider must satisfy.

    Args:
        login_path: Path for the login page/redirect.
        logout_path: Path for the logout endpoint.
        allow_routes: Route names that bypass the auth check.
    """

    def __init__(
        self,
        login_path: str = "/login",
        logout_path: str = "/logout",
        allow_routes: Sequence[str] | None = None,
    ) -> None:
        self.login_path = login_path
        self.logout_path = logout_path
        self.allow_routes: list[str] = (
            list(allow_routes) if allow_routes is not None else []
        )
        # Set by `BaseAdmin._init_auth`, used to emit `AFTER_LOGIN`.
        self.events: AdminEventBus | None = None
        _log.debug(
            "auth provider init: %s login_path=%r logout_path=%r allow_routes=%r",
            type(self).__name__,
            login_path,
            logout_path,
            self.allow_routes,
        )

    @abstractmethod
    async def authenticate(self, request: Request) -> AdminUser | None:
        """Authenticate the current request.

        Returns:
            An `AdminUser` if the request is authenticated, or `None` to deny access.
            `AuthMiddleware` stores the result on `request.state.admin_user` and sets
            `request.state.is_anonymous` accordingly.
        """

    def get_routes(self, templates: Jinja2Templates) -> list[Route]:
        """Return the routes this provider mounts inside the admin prefix."""
        return []

    def get_middleware(self) -> Middleware:
        return Middleware(AuthMiddleware, provider=self)

    async def _emit_after_login(self, request: Request, user: AdminUser | None) -> None:
        """Emits `AFTER_LOGIN` on the admin's event bus, once login succeeds."""
        if self.events is None:
            return
        ctx = AfterLoginContext(
            event=AdminEvent.AFTER_LOGIN,
            request=request,
            view_key="",
            user=user,
        )
        await self.events.emit(ctx)

class AuthProvider(BaseAuthProvider):
    """
    Subclass to use the built-in username/password login page.

    Implement `login`, `logout`, and `authenticate` only;
    the framework owns the form, rendering, and redirect.
    """

    async def login(
        self,
        username: str,
        password: str,
        remember_me: bool,
        request: Request,
    ) -> Response | None:
        """
        Validate credentials and persist auth state (e.g. write to request.session).

        Raise `LoginFailed` for wrong credentials, or `FormValidationError` to mark
        specific form fields invalid.

        Returns:
            A `Response` to redirect to a custom URL, or `None` to fall back to the
            default redirect (`next` query param, or the admin index).

        Examples:
            ```python
            async def login(self, username, password, remember_me, request):
                user = await db.get_user(username)
                if user and verify_password(password, user.hashed_password):
                    request.session["user_id"] = user.id
                    return
                raise LoginFailed("Invalid username or password")
            ```
        """
        raise LoginFailed("Not Implemented")

    async def logout(self, request: Request) -> Response | None:
        """
        Clear auth state (e.g. request.session.clear()).

        Returns:
            A `Response` to redirect to a custom URL, or `None` to fall back to the
            default redirect (the admin index).

        Examples:
            ```python
            async def logout(self, request):
                request.session.clear()
            ```
        """
        raise NotImplementedError()

    async def authenticate(self, request: Request) -> AdminUser | None:
        return None

    async def render_login(
        self, request: Request, templates: Jinja2Templates
    ) -> Response:
        """Render and handle the built-in username/password login page."""
        if getattr(request.state, "admin_user", None) is not None:
            fallback = index_url(request)
            next_url = safe_redirect_url(
                request.query_params.get("next") or fallback,
                request,
                fallback,
            )
            _log.debug(
                "login page: already authenticated, redirecting to %s",
                request.client.host if request.client else "unknown",
            )
            return RedirectResponse(
                next_url,
                status_code=HTTP_303_SEE_OTHER,
            )
        if request.method == "GET":
            _log.debug(
                "login page: GET %s from %s",
                request.url.path,
                request.client.host if request.client else "unknown",
            )
            return templates.TemplateResponse(
                request=request,
                name="login.html",
            )

        form = await request.form()
        username = str(form.get("username") or "")
        remember_me = form.get("remember_me") == "on"
        fallback = index_url(request)
        next_url = safe_redirect_url(
            request.query_params.get("next") or fallback,
            request,
            fallback,
        )
        _log.info(
            "login attempt: username=%r remember_me=%s next=%r from %s",
            username,
            remember_me,
            next_url,
            request.client.host if request.client else "unknown",
        )
        try:
            result = await self.login(
                username,
                str(form.get("password") or ""),
                remember_me,
                request,
            )
            _log.info(
                "login success: username=%r redirect_to=%r from %s",
                username,
                next_url,
                request.client.host if request.client else "unknown",
            )
            await self._emit_after_login(request, await self.authenticate(request))
            return result or RedirectResponse(next_url, status_code=HTTP_303_SEE_OTHER)
        except FormValidationError as errors:
            _log.warning(
                "login form invalid: username=%r fields=%r from %s",
                username,
                list(errors.errors.keys()),
                request.client.host if request.client else "unknown",
            )
            return templates.TemplateResponse(
                request=request,
                name="login.html",
                context={"form_errors": errors},
                status_code=HTTP_422,
            )
        except LoginFailed as error:
            _log.warning(
                "login failed: username=%r reason=%r from %s",
                username,
                error.msg,
                request.client.host if request.client else "unknown",
            )
            return templates.TemplateResponse(
                request=request,
                name="login.html",
                context={"error": error.msg},
                status_code=HTTP_400_BAD_REQUEST,
            )

    async def render_logout(self, request: Request) -> Response:
        """Call `logout()`, then redirect to the admin index."""
        user = getattr(request.state, "user", None)
        _log.info(
            "logout initiated: user=%r from %s",
            user,
            request.client.host if request.client else "unknown",
        )
        result = await self.logout(request)
        _log.info(
            "logout complete: user=%r from %s",
            user,
            request.client.host if request.client else "unknown",
        )
        return result or RedirectResponse(
            index_url(request),
            status_code=HTTP_303_SEE_OTHER,
        )

    def get_routes(self, templates: Jinja2Templates) -> list[Route]:
        return [
            Route(
                self.login_path,
                wrap_endpoint_with_kwargs(self.render_login, templates=templates),
                methods=["GET", "POST"],
                name="login",
            ),
            Route(
                self.logout_path, self.render_logout, methods=["POST"], name="logout"
            ),
        ]

class OAuthProvider(BaseAuthProvider):
    """
    Subclass to use an OAuth2/OIDC flow.

    Implement `redirect_to_provider`, `handle_callback`, and `authenticate`.
    `get_routes()` returns the login redirect, logout, and callback routes.
    The callback is automatically added to `allow_routes` so the middleware
    never blocks it.

    Requires `SessionMiddleware`: the OAuth client stores the state nonce in the
    session between the redirect and the callback.

    Examples:
        ```python
        from authlib.integrations.starlette_client import OAuth

        oauth = OAuth()
        oauth.register(
            "provider",
            client_id=CLIENT_ID,
            client_secret=CLIENT_SECRET,
            client_kwargs={"scope": "openid profile email"},
            server_metadata_url=SERVER_METADATA_URL,
        )


        class OIDCAuthProvider(OAuthProvider):
            async def redirect_to_provider(self, request: Request, callback_url: str) -> Response:
                client = oauth.create_client("provider")
                return await client.authorize_redirect(request, callback_url)

            async def handle_callback(self, request: Request) -> None:
                client = oauth.create_client("provider")
                token = await client.authorize_access_token(request)
                # Persist userinfo in the encrypted session cookie.
                request.session["user"] = dict(token["userinfo"])

            async def authenticate(self, request: Request) -> AdminUser | None:
                user = request.session.get("user")
                if user:
                    return AdminUser(
                        username=user.get("name") or user.get("email"),
                        photo_url=user.get("picture"),
                    )
                return None

            async def logout(self, request: Request) -> Response | None:
                request.session.clear()
                # Return a Response to redirect to the provider's end_session_endpoint,
                # or None to fall back to the default admin index redirect.
                return None
        ```
    """

    def __init__(
        self,
        login_path: str = "/login",
        logout_path: str = "/logout",
        callback_path: str = "/oauth/callback",
        allow_routes: list[str] | None = None,
    ) -> None:
        super().__init__(login_path, logout_path, allow_routes)
        self.callback_path = callback_path
        # The callback is always public: the OAuth provider redirects here from
        # outside the app, before a session exists.
        self.allow_routes.append("oauth_callback")

    @abstractmethod
    async def redirect_to_provider(
        self, request: Request, callback_url: str
    ) -> Response:
        """Return a redirect response to start the OAuth2 flow."""

    @abstractmethod
    async def handle_callback(self, request: Request) -> None:
        """Exchange the authorization code and write auth state (e.g. to session)."""

    async def logout(self, request: Request) -> Response | None:
        """Clear auth state on logout.

        Returns:
            A `Response` to redirect to a custom URL (e.g. the OIDC
            `end_session_endpoint`), or `None` to use the default index redirect.
        """
        return None

    async def render_login(self, request: Request) -> Response:
        """Redirect to the OAuth provider's authorization URL."""
        route_name = request.app.state.ROUTE_NAME
        callback_url = request.url_for(route_name + ":oauth_callback")
        if next_url := request.query_params.get("next"):
            callback_url = callback_url.include_query_params(next=next_url)
        return await self.redirect_to_provider(request, str(callback_url))

    async def render_logout(self, request: Request) -> Response:
        """Call `logout()`, then redirect to its result or to the admin index."""
        result = await self.logout(request)
        return (
            RedirectResponse(
                index_url(request),
                status_code=HTTP_303_SEE_OTHER,
            )
            if result is None
            else result
        )

    @login_not_required
    async def render_callback(self, request: Request) -> Response:
        """Handle the OAuth callback, then redirect to next or the admin index."""
        await self.handle_callback(request)
        await self._emit_after_login(request, await self.authenticate(request))
        fallback = index_url(request)
        next_url = safe_redirect_url(
            request.query_params.get("next") or fallback,
            request,
            fallback,
        )
        return RedirectResponse(next_url, status_code=HTTP_303_SEE_OTHER)

    def get_routes(self, templates: Jinja2Templates) -> list[Route]:
        return [
            Route(self.login_path, self.render_login, methods=["GET"], name="login"),
            Route(
                self.logout_path, self.render_logout, methods=["POST"], name="logout"
            ),
            Route(
                self.callback_path,
                self.render_callback,
                methods=["GET"],
                name="oauth_callback",
            ),
        ]

@dataclass
class AdminUser:
    username: str = field(default_factory=lambda: _("Administrator"))
    photo_url: str | None = None

class AuthMiddleware(BaseHTTPMiddleware):
    def __init__(self, app: ASGIApp, provider: "BaseAuthProvider") -> None:
        super().__init__(app)
        self.provider = provider
        self.allow_routes: list[str] = ["login", "static", *provider.allow_routes]
        _log.debug(
            "AuthMiddleware init: provider=%s allow_routes=%r",
            type(provider).__name__,
            self.allow_routes,
        )

    async def dispatch(
        self, request: Request, call_next: RequestResponseEndpoint
    ) -> Response:
        """Enforce authentication for every request that passes through this middleware.

        Allows the request through when one of the following holds: the provider
        authenticates the user (`authenticate()` returns an `AdminUser`), the matched
        route name is in `allow_routes`, or the endpoint is decorated with
        `@login_not_required`. Otherwise, redirects to the login page.
        """
        _admin_app: Starlette = request.scope["app"]
        current_route: Route | Mount | WebSocketRoute | None = None
        for route in _admin_app.routes:
            match, _ = route.matches(request.scope)
            if match == Match.FULL:
                if not isinstance(
                    route, (Route, Mount, WebSocketRoute)
                ):  # pragma: no cover
                    raise TypeError(f"Unexpected route type {type(route).__name__}")
                current_route = route
                break

        route_name = getattr(current_route, "name", None)
        route_path = getattr(current_route, "path", None)
        _log.debug(
            "auth check: %s %s  route=%r path=%r from %s",
            request.method,
            request.url.path,
            route_name,
            route_path,
            request.client.host if request.client else "unknown",
        )

        user = await self.provider.authenticate(request)

        if user is not None:
            if not isinstance(user, AdminUser):
                raise TypeError(
                    f"authenticate() must return an AdminUser instance, "
                    f"got {type(user).__name__}"
                )
            request.state.admin_user = user
            request.state.is_anonymous = False
            _log.debug(
                "auth: authenticated user=%r route=%r %s %s",
                user,
                route_name,
                request.method,
                request.url.path,
            )
            return await call_next(request)

        request.state.is_anonymous = True

        if current_route is not None and current_route.name in self.allow_routes:
            _log.debug(
                "auth: allowed (route in allow_routes) route=%r",
                route_name,
            )
            return await call_next(request)

        if (
            current_route is not None
            and hasattr(current_route, "endpoint")
            and getattr(current_route.endpoint, "_login_not_required", False)
        ):
            _log.debug(
                "auth: allowed (@login_not_required) route=%r",
                route_name,
            )
            return await call_next(request)

        login_url = "{url}?{query_params}".format(
            url=request.url_for(request.app.state.ROUTE_NAME + ":login"),
            query_params=urlencode({"next": str(request.url)}),
        )
        _log.warning(
            "auth: unauthenticated, redirecting to login: %s %s -> %s from %s",
            request.method,
            request.url.path,
            login_url,
            request.client.host if request.client else "unknown",
        )
        return RedirectResponse(login_url, status_code=HTTP_303_SEE_OTHER)