stroblme
75b3a93094
Docs / docs (push) Successful in 33s
Playwright Tests / test-playwright (1, 2) (push) Failing after 1m37s
Playwright Tests / test-playwright (2, 2) (push) Failing after 27s
pre-commit / pre-commit (push) Failing after 2m37s
Test Backend / test-backend (3.12) (push) Successful in 5m7s
Test Backend / test-backend (3.14) (push) Successful in 5m2s
Compose Smoke Test / test-compose (push) Failing after 28s
Playwright Tests / merge-reports (push) Failing after 2m29s
114 lines
4.8 KiB
Python
114 lines
4.8 KiB
Python
"""The dashboard, served by the API on its own origin.
|
|
|
|
One process and one port: the built SPA comes off disk, and every path the API
|
|
does not own answers with ``index.html``, so a client-side route survives a
|
|
reload. Same origin also means the dashboard calls the API as a path, needs no
|
|
CORS allowance and no second hostname, which is what lets the whole instance
|
|
sit behind a single published port.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import mimetypes
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from starlette.datastructures import Headers
|
|
from starlette.responses import FileResponse
|
|
from starlette.types import ASGIApp, Receive, Scope, Send
|
|
|
|
#: What the API answers itself. Everything else is the dashboard's.
|
|
API_PREFIXES = ("/api/", "/hooks/", "/mcp", "/.well-known/", "/docs", "/redoc")
|
|
|
|
# Older mime tables predate the manifest, and it would go out as a download.
|
|
mimetypes.add_type("application/manifest+json", ".webmanifest")
|
|
|
|
|
|
def _accepts_gzip(header: str) -> bool:
|
|
"""An explicit refusal overrides a wildcard, including on a cached asset."""
|
|
qualities: dict[str, float] = {}
|
|
for entry in header.lower().split(","):
|
|
encoding, *parameters = entry.strip().split(";")
|
|
quality = 1.0
|
|
for parameter in parameters:
|
|
key, _, value = parameter.strip().partition("=")
|
|
if key == "q":
|
|
try:
|
|
quality = float(value)
|
|
except ValueError:
|
|
quality = 0.0
|
|
qualities[encoding.strip()] = quality if 0.0 <= quality <= 1.0 else 0.0
|
|
gzip = qualities.get("gzip", qualities.get("*", 0.0))
|
|
return gzip > 0 and gzip >= qualities.get("identity", 0.0)
|
|
|
|
|
|
def content_security_policy(index_html: str) -> str:
|
|
"""The policy for the dashboard, admitting its inline scripts by hash.
|
|
|
|
The hash is taken from the file being served, so editing the theme script
|
|
in ``index.html`` cannot leave a stale one behind.
|
|
style-src 'unsafe-inline': CodeMirror, Radix's scroll lock and the toasts insert
|
|
<style> elements at runtime, which no hash can name. img/frame-src take any
|
|
http(s): a camera stream or website widget points wherever its dashboard
|
|
says.
|
|
"""
|
|
hashes = "".join(
|
|
f" 'sha256-{base64.b64encode(hashlib.sha256(script.encode()).digest()).decode()}'"
|
|
for script in re.findall(r"<script>([\s\S]*?)</script>", index_html)
|
|
)
|
|
return (
|
|
f"default-src 'self'; script-src 'self'{hashes}; "
|
|
"style-src 'self' 'unsafe-inline'; img-src 'self' data: blob: http: https:; "
|
|
"media-src 'self' blob:; font-src 'self'; connect-src 'self'; "
|
|
"worker-src 'self'; frame-src http: https:; frame-ancestors 'self'; "
|
|
"object-src 'none'; base-uri 'self'; form-action 'self'"
|
|
)
|
|
|
|
|
|
class Dashboard:
|
|
"""Serve the built SPA for every GET the API does not own."""
|
|
|
|
def __init__(self, app: ASGIApp, dist: Path) -> None:
|
|
self.app = app
|
|
self.dist = dist.resolve()
|
|
self.index = self.dist / "index.html"
|
|
self.headers = {
|
|
"Content-Security-Policy": content_security_policy(self.index.read_text()),
|
|
"X-Content-Type-Options": "nosniff",
|
|
"Referrer-Policy": "strict-origin-when-cross-origin",
|
|
}
|
|
|
|
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
|
|
path: str = scope.get("path", "")
|
|
if (
|
|
scope["type"] != "http"
|
|
or scope["method"] not in ("GET", "HEAD")
|
|
or path.startswith(API_PREFIXES)
|
|
):
|
|
await self.app(scope, receive, send)
|
|
return
|
|
|
|
file = (self.dist / path.lstrip("/")).resolve()
|
|
if file.is_file() and file.is_relative_to(self.dist) and file != self.index:
|
|
headers = dict(self.headers)
|
|
# Vite puts a content hash in every asset's name, so one never
|
|
# changes under its URL.
|
|
if path.startswith("/assets/"):
|
|
headers["Cache-Control"] = "public, max-age=31536000, immutable"
|
|
else:
|
|
# The one file that must not be cached: it names the current
|
|
# assets, so a stale copy pins the browser to the previous build.
|
|
file = self.index
|
|
headers = {**self.headers, "Cache-Control": "no-cache"}
|
|
media_type = mimetypes.guess_type(file.name)[0] or "text/plain"
|
|
compressed = file.with_name(file.name + ".gz").resolve()
|
|
if compressed.is_relative_to(self.dist) and compressed.is_file():
|
|
headers["Vary"] = "Accept-Encoding"
|
|
if _accepts_gzip(Headers(scope=scope).get("accept-encoding", "")):
|
|
file = compressed
|
|
headers["Content-Encoding"] = "gzip"
|
|
response = FileResponse(file, headers=headers, media_type=media_type)
|
|
await response(scope, receive, send)
|