src/genro_asgi/middleware/errors.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """Error middleware: the outermost try/except of the chain and the login seam.16 17 ``ErrorMiddleware`` (order 100, the only middleware enabled by default —18 ``errors=False`` disables it) maps control-flow exceptions to responses:19 ``Redirect`` → its status plus the ``Location`` header, ``HTTPException`` →20 its status with the detail, any other ``Exception`` → a hidden 500 logged via21 the instance logger. Responses are built with the ``Response`` class; an22 exception's ``headers`` (e.g. a ``WWW-Authenticate`` challenge) are forwarded23 onto the response.24 25 Content negotiation (D4 error-body reconciliation): the error body follows the26 caller's ``Accept``. A caller asking for JSON (``application/json`` or ``*/*``,27 never ``text/html``) gets the ``{"error": ...}`` document built by28 ``Response.set_error`` — the single live JSON error path; anyone else keeps the29 historical ``text/plain`` body. A missing ``Accept`` stays ``text/plain`` (the30 pre-existing default).31 32 Challenge negotiation (only when the server carries an active login surface —33 ``server.login_enabled``): a 401 is where the server asks the caller to34 authenticate. A browser NAVIGATION (an http GET whose ``Accept`` includes35 ``text/html``) gets a 302 to ``/_server/login_page`` carrying the original36 path+query as a ``safe_next_path``-validated ``next``; any other caller keeps37 the bare 401 (with its ``WWW-Authenticate``) and gains a ``{"login_url": ...}``38 JSON body so an SPA can drive the login. With the login surface off the 401 is39 answered exactly like any other error. The request shape is read from the scope40 headers (``headers_dict``) — never an ambient request.41 42 The middleware wraps ``send`` to track whether ``http.response.start`` has43 already passed downstream: an exception raised AFTER the response started44 cannot be answered (a second start would corrupt the stream), so it is logged45 and re-raised — the server/transport tears the connection down. The chain only46 carries ``http`` scopes (the mixin routes the others past it), so no scope47 filtering happens here.48 """49 50 from __future__ import annotations51 52 from typing import TYPE_CHECKING53 from urllib.parse import quote54 55 from ..auth import safe_next_path56 from ..exceptions import HTTPException, Redirect57 from ..response import Response58 from .base import BaseMiddleware, headers_dict59 60 if TYPE_CHECKING:61 from ..types import Message, Receive, Scope, Send62 63 __all__ = ["ErrorMiddleware"]64 65 LOGIN_PAGE_URL = "/_server/login_page"66 67 68 class ErrorMiddleware(BaseMiddleware):69 """Outermost middleware answering raised exceptions with HTTP responses."""70 71 middleware_order = 10072 middleware_default = True73 74 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:75 """Run the chain; map raised exceptions to responses unless already started."""76 started = False77 78 async def tracking_send(message: Message) -> None:79 nonlocal started80 if message["type"] == "http.response.start":81 started = True82 await send(message)83 84 try:85 await self.app(scope, receive, tracking_send)86 except Exception as exc:87 if started:88 self.logger.exception(89 "error after response started serving %s", scope.get("path", "?")90 )91 raise92 response = self._response_for(exc, scope)93 await response(scope, receive, send)94 95 def _response_for(self, exc: Exception, scope: Scope) -> Response:96 """The challenge response for an active-login 401, otherwise the error response."""97 if isinstance(exc, HTTPException) and exc.status == 401 and self._login_active():98 return self._challenge_response(exc, scope)99 return self._error_response(exc, scope)100 101 def _login_active(self) -> bool:102 """True when the server carries an active login surface (``login_enabled``).103 104 Used standalone (no server, or one without a login surface) the105 middleware answers the 401 unchanged.106 """107 return bool(getattr(self.server, "login_enabled", False))108 109 def _challenge_response(self, exc: HTTPException, scope: Scope) -> Response:110 """Negotiate a 401 into a browser redirect or an API-friendly 401.111 112 A browser navigation gets a 302 to the login page with the original113 path+query as a validated ``next``; any other caller keeps the bare 401114 (with its ``WWW-Authenticate``) and gains a ``{"login_url": ...}`` body.115 """116 headers = headers_dict(scope)117 if self._is_browser_navigation(scope, headers):118 target = safe_next_path(self._original_target(scope))119 response = Response(status_code=302, media_type="text/plain")120 response.set_header("location", f"{LOGIN_PAGE_URL}?next={quote(target, safe='')}")121 return response122 response = Response(status_code=401)123 response.set_result({"login_url": LOGIN_PAGE_URL})124 self._forward_headers(response, exc)125 return response126 127 def _is_browser_navigation(self, scope: Scope, headers: dict[str, str]) -> bool:128 """True for an http GET whose ``Accept`` asks for HTML (a navigation)."""129 if str(scope.get("method", "")).upper() != "GET":130 return False131 return "text/html" in headers.get("accept", "")132 133 def _original_target(self, scope: Scope) -> str:134 """Rebuild the request's original path (+query) for the ``next`` value."""135 path = str(scope.get("path", "/"))136 query = scope.get("query_string", b"")137 query_str = query.decode("latin-1") if isinstance(query, bytes) else str(query)138 return f"{path}?{query_str}" if query_str else path139 140 def _error_response(self, exc: Exception, scope: Scope) -> Response:141 """Build the ``Response`` for a raised exception, negotiating the body format."""142 if isinstance(exc, Redirect):143 response = Response(status_code=exc.status, media_type="text/plain")144 response.set_header("location", exc.location)145 self._forward_headers(response, exc)146 return response147 wants_json = self._wants_json(headers_dict(scope))148 if isinstance(exc, HTTPException):149 if wants_json:150 response = Response()151 response.set_error(exc)152 else:153 response = Response(154 content=exc.detail or "", status_code=exc.status, media_type="text/plain"155 )156 else:157 self.logger.exception("unhandled error serving %s", scope.get("path", "?"))158 if wants_json:159 response = Response(status_code=500)160 response.set_result({"error": "Internal Server Error"})161 else:162 response = Response(163 content="Internal Server Error", status_code=500, media_type="text/plain"164 )165 self._forward_headers(response, exc)166 return response167 168 def _wants_json(self, headers: dict[str, str]) -> bool:169 """True when the caller's ``Accept`` asks for JSON (never for a browser navigation).170 171 A missing ``Accept`` keeps the historical ``text/plain`` default; an172 ``Accept`` naming ``text/html`` (a browser) also stays text; only an API173 caller (``application/json`` or ``*/*``) gets the JSON error document.174 """175 accept = headers.get("accept", "")176 if not accept or "text/html" in accept:177 return False178 return "application/json" in accept or "*/*" in accept179 180 def _forward_headers(self, response: Response, exc: Exception) -> None:181 """Forward an exception's ASGI header pairs onto the response."""182 for name, value in getattr(exc, "headers", []):183 response.set_header(name.decode("latin-1"), value.decode("latin-1"))