Coverage for src/asgi_profiler/viewer.py: 99%
121 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-22 16:59 +0000
1"""The viewer: a self-contained Starlette app you mount wherever you like.
3Every URL it emits is relative to the mount prefix, so the UI works unchanged
4at `/profiler`, at `/admin/_perf`, or behind a proxy that sets a root path.
5Nothing is hardcoded, and `url_for` is not used, because a mounted sub-app
6cannot resolve its own route names without knowing the name the parent gave
7the mount.
9The prefix is taken from `scope["root_path"]` when the server sets one --
10Starlette's `Mount` does, and it is also what accounts for a proxy prefix --
11and otherwise from the explicit `prefix` argument. Relying on `root_path`
12alone would be a bet on server-specific behaviour that this library cannot
13test from the inside: Litestar's mount, for one, does not set it.
14"""
16from __future__ import annotations
18import inspect
19from pathlib import Path
20from typing import Any
21from urllib.parse import urlencode
23from starlette.applications import Starlette
24from starlette.concurrency import run_in_threadpool
25from starlette.requests import Request
26from starlette.responses import (
27 JSONResponse,
28 PlainTextResponse,
29 RedirectResponse,
30 Response,
31)
32from starlette.routing import Route
33from starlette.templating import Jinja2Templates
35from .config import ProfilerConfig
36from .detectors import LABELS, SEVERITY
37from .storage import Filters, Storage
39HERE = Path(__file__).parent
40TEMPLATES = Jinja2Templates(directory=str(HERE / "templates"))
41CSS_PATH = HERE / "static" / "profiler.css"
44class _Urls:
45 """Mount-aware URL builder handed to the templates."""
47 __slots__ = ("root",)
49 def __init__(self, request: Request, prefix: str = "") -> None:
50 root = request.scope.get("root_path", "").rstrip("/")
51 self.root = root or prefix.rstrip("/")
53 @property
54 def requests(self) -> str:
55 return f"{self.root}/"
57 @property
58 def summary(self) -> str:
59 return f"{self.root}/summary"
61 @property
62 def statements(self) -> str:
63 return f"{self.root}/statements"
65 @property
66 def clear(self) -> str:
67 return f"{self.root}/clear"
69 @property
70 def css(self) -> str:
71 return f"{self.root}/static/profiler.css"
73 def detail(self, profile_id: str) -> str:
74 return f"{self.root}/request/{profile_id}"
76 def page(self, params: Any, number: int) -> str:
77 """The current listing URL with `page` replaced."""
78 query = {k: v for k, v in params.items() if k != "page" and v != ""}
79 query["page"] = str(number)
80 return f"{self.root}/?{urlencode(query)}"
82 def for_route(self, method: str, route: str) -> str:
83 return f"{self.root}/?{urlencode({'route': route, 'method': method})}"
86def _same_origin(request: Request) -> bool:
87 """Reject cross-site state changes.
89 `/clear` is a plain POST, so without this any page on the internet can
90 make a visiting developer's browser wipe their profiling history with a
91 hidden auto-submitting form. Non-browser clients send neither header and
92 are left alone.
93 """
94 site = request.headers.get("sec-fetch-site")
95 if site is not None:
96 # Allow-list, not a deny-list: `same-site` is a *different origin* on a
97 # sibling subdomain, which is exactly the attacker we are excluding.
98 return site in ("same-origin", "none")
99 origin = request.headers.get("origin")
100 if origin is None:
101 return True # curl, httpx, the test client
102 host = request.headers.get("host", "")
103 return origin.split("://", 1)[-1] == host
106class Guarded:
107 """Wraps the viewer in the caller's `authorize` check.
109 A module-level class rather than a closure so that the security boundary
110 can be imported, subclassed and tested directly.
111 """
113 def __init__(self, app: Any, authorize: Any) -> None:
114 self.app = app
115 self.authorize = authorize
117 async def __call__(self, scope: Any, receive: Any, send: Any) -> None:
118 if scope["type"] == "http": 118 ↛ 126line 118 didn't jump to line 126 because the condition on line 118 was always true
119 verdict = self.authorize(Request(scope, receive))
120 if inspect.isawaitable(verdict):
121 verdict = await verdict
122 if not verdict:
123 response = PlainTextResponse("Forbidden", status_code=403)
124 await response(scope, receive, send)
125 return
126 await self.app(scope, receive, send)
129def build_viewer(storage: Storage, config: ProfilerConfig, prefix: str = "") -> Any:
130 """Return the ASGI app that renders the profiler UI.
132 Args:
133 storage: where profiles are read from.
134 config: the active :class:`ProfilerConfig`.
135 prefix: mount prefix, used when the server does not set `root_path`.
136 """
137 css = CSS_PATH.read_text(encoding="utf-8")
139 # Every storage read runs in a worker thread. A `Storage` is a synchronous
140 # interface -- SQLite queries, a lock, and a bounded wait for the writer --
141 # and calling one directly from an `async def` handler blocks the loop for
142 # the whole application, not just this request. Measured: a viewer page
143 # froze every other task for five seconds while SQLite waited on a lock.
144 async def read(fn: Any, *args: Any, **kwargs: Any) -> Any:
145 return await run_in_threadpool(fn, *args, **kwargs)
147 async def ctx(request: Request, **extra: Any) -> dict[str, Any]:
148 return {
149 "config": config,
150 "storage_size": await read(storage.count),
151 "urls": _Urls(request, prefix),
152 "problem_labels": LABELS,
153 "problem_severity": SEVERITY,
154 **extra,
155 }
157 async def _page_for(request: Request):
158 params = request.query_params
159 filters = Filters.from_params(params)
160 try:
161 number = int(params.get("page", "1"))
162 except ValueError:
163 number = 1
164 page = await read(storage.search, filters, page=number, size=config.page_size)
165 return filters, page
167 async def requests_page(request: Request) -> Response:
168 filters, page = await _page_for(request)
169 return TEMPLATES.TemplateResponse(
170 request,
171 "requests.html",
172 await ctx(
173 request,
174 page=page,
175 profiles=page.items,
176 total=await read(storage.count),
177 order=filters.order,
178 filters=request.query_params,
179 active_filters=filters.active,
180 ),
181 )
183 async def summary_page(request: Request) -> Response:
184 return TEMPLATES.TemplateResponse(
185 request,
186 "summary.html",
187 await ctx(
188 request,
189 rows=await read(storage.summarise),
190 total=await read(storage.count),
191 ),
192 )
194 async def statements_page(request: Request) -> Response:
195 return TEMPLATES.TemplateResponse(
196 request,
197 "statements.html",
198 await ctx(
199 request,
200 rows=await read(storage.statements, config.statement_limit),
201 total=await read(storage.count),
202 ),
203 )
205 async def detail_page(request: Request) -> Response:
206 profile = await read(storage.get, request.path_params["profile_id"])
207 if profile is None:
208 return RedirectResponse(_Urls(request, prefix).requests, 302)
209 return TEMPLATES.TemplateResponse(
210 request,
211 "detail.html",
212 await ctx(request, profile=profile, groups=profile.query_groups),
213 )
215 async def clear(request: Request) -> Response:
216 if not _same_origin(request):
217 return PlainTextResponse("Cross-site request rejected", status_code=403)
218 await read(storage.clear)
219 return RedirectResponse(_Urls(request, prefix).requests, 303)
221 async def stylesheet(request: Request) -> Response: # noqa: ARG001 (unused-function-argument)
222 return Response(
223 css,
224 media_type="text/css",
225 headers={"cache-control": "no-cache"},
226 )
228 # -- JSON ------------------------------------------------------------
229 # The HTML is for reading; this is for scripting. Assert in a test that an
230 # endpoint stays under N queries, diff two runs, attach a trace to a bug
231 # report -- none of which you can do by scraping a page.
232 async def requests_json(request: Request) -> Response:
233 _, page = await _page_for(request)
234 return JSONResponse(
235 {
236 "total": page.total,
237 "page": page.number,
238 "pages": page.pages,
239 "size": page.size,
240 "requests": [p.as_dict(with_queries=False) for p in page.items],
241 }
242 )
244 async def detail_json(request: Request) -> Response:
245 profile = await read(storage.get, request.path_params["profile_id"])
246 if profile is None:
247 return JSONResponse({"error": "not found"}, status_code=404)
248 return JSONResponse(profile.as_dict())
250 async def summary_json(request: Request) -> Response: # noqa: ARG001 (unused-function-argument)
251 rows = await read(storage.summarise)
252 return JSONResponse({"routes": [row.as_dict() for row in rows]})
254 async def statements_json(request: Request) -> Response: # noqa: ARG001 (unused-function-argument)
255 rows = await read(storage.statements, config.statement_limit)
256 return JSONResponse({"statements": [row.as_dict() for row in rows]})
258 routes = [
259 Route("/", requests_page, name="profiler_requests"),
260 Route("/summary", summary_page, name="profiler_summary"),
261 Route("/statements", statements_page, name="profiler_statements"),
262 # The JSON routes come first: `{profile_id}` matches any non-slash
263 # run, so `/request/abc.json` would otherwise be served as HTML for a
264 # profile literally named "abc.json".
265 Route("/requests.json", requests_json, name="profiler_requests_json"),
266 Route("/summary.json", summary_json, name="profiler_summary_json"),
267 Route("/statements.json", statements_json, name="profiler_statements_json"),
268 Route("/request/{profile_id}.json", detail_json, name="profiler_detail_json"),
269 Route("/request/{profile_id}", detail_page, name="profiler_detail"),
270 Route("/clear", clear, methods=["POST"], name="profiler_clear"),
271 Route("/static/profiler.css", stylesheet, name="profiler_css"),
272 ]
274 viewer = Starlette(routes=routes)
275 if config.authorize is not None:
276 return Guarded(viewer, config.authorize)
277 return viewer