feat: serve the admin console from FastAPI and Django

AGENTS.md

@@ -65,6 +65,7 @@ current Django, so both are covered. Do not use language or standard library fea
 - `src/seamless_auth/_adapter.py`: `Adapter`, the request pipeline, the upstream call, manifest
   routes, credential resolution (with silent refresh), session verification, refresh, logout, the
   guard, the cross-site check.
+- `src/seamless_auth/_console.py`: the admin dashboard proxy (`Adapter.console`).
 - `src/seamless_auth/_refresh.py`: refresh sharing (one result per refresh token for 5 seconds,
   across threads).
 - `src/seamless_auth/_manifest.py`: parsing, matching, the live and bundled manifest.

README.md

@@ -164,6 +164,29 @@ auth = Adapter(..., deliver=deliver)
 An `async def` callback works too; under FastAPI it runs on the application's event loop. A
 delivery error answers the request with 502 `delivery_failed`.
 
+### The admin console
+
+The adapter can serve the Seamless admin dashboard from your API, proxied from the auth API, so it
+shares the origin and cookie scope of the `/auth` routes:
+
+```python
+# FastAPI
+from seamless_auth.fastapi import console_router
+
+app.include_router(console_router(auth))  # /console
+
+# Django (urls.py)
+from seamless_auth.django import console_urlpatterns
+
+urlpatterns += [path("console/", include(console_urlpatterns))]
+```
+
+It serves `GET` and `HEAD` only and forwards nothing but the method and path, so the browser's
+cookies never reach the upstream. It refuses any path that could leave the console (dot segments,
+encoded separators), and follows a redirect only while it stays inside the console on the auth API.
+When you serve it, add your API's origin to the auth server's `ORIGINS` so passkey ceremonies
+started in the console verify.
+
 ## How it runs
 
 The adapter is synchronous and thread-safe: one `Adapter` serves the whole process. Under FastAPI

src/seamless_auth/_adapter.py

@@ -11,7 +11,7 @@
 import time
 from collections.abc import Awaitable, Callable, Iterable, Iterator
 from dataclasses import dataclass, field
-from typing import Any
+from typing import TYPE_CHECKING, Any
 from urllib.parse import urlsplit
 
 import httpx2
@@ -36,6 +36,9 @@
 from ._manifest import ManifestRoute, ManifestSource, RouteMatch
 from ._refresh import Refresher, RefreshOutcome
 
+if TYPE_CHECKING:
+    from ._console import Console
+
 log = logging.getLogger("seamless_auth")
 
 MIN_SECRET_LENGTH = 32
@@ -225,6 +228,7 @@ def __init__(
         self._manifest = ManifestSource(url, self._client, not disable_manifest_fetch)
         self._refreshes = Refresher()
         self._proxy_token: tuple[str, float] | None = None
+        self._console: Console | None = None
 
     def __repr__(self) -> str:
         return f"Adapter(auth_server_url={self._config.auth_server_url!r})"
@@ -233,6 +237,18 @@ def __repr__(self) -> str:
     def config(self) -> Config:
         return self._config
 
+    def console(self, method: str, path: str, query: str = "") -> AuthResponse:
+        """Serves the Seamless admin dashboard, proxied from the auth API, so it
+        loads from the same origin as the cookie-based auth routes. ``path`` is the
+        raw path below /console. GET and HEAD only; nothing is forwarded but the
+        method and path, and no path can leave the console. Blocking: run it on a
+        worker thread from async code."""
+        from ._console import Console  # noqa: PLC0415  (imports this module)
+
+        if self._console is None:
+            self._console = Console(self)
+        return self._console.handle(method, path, query)
+
     # The request pipeline.
 
     def handle(self, request: AuthRequest) -> AuthResponse:

src/seamless_auth/_console.py

@@ -0,0 +1,132 @@
+"""The admin dashboard, proxied from the auth API onto the application's origin."""
+
+from __future__ import annotations
+
+import json
+import logging
+from collections.abc import Iterator
+from typing import TYPE_CHECKING
+
+import httpx2
+
+from ._http import AuthResponse
+from ._manifest import percent_decode
+
+if TYPE_CHECKING:
+    from ._adapter import Adapter
+
+log = logging.getLogger("seamless_auth")
+
+CONSOLE_BASE_PATH = "/console"
+UPSTREAM_TIMEOUT = 10.0
+MAX_REDIRECTS = 5
+FORWARDED_HEADERS = ("content-type", "cache-control", "etag", "last-modified")
+
+
+def _error(status: int, message: str) -> AuthResponse:
+    return AuthResponse(
+        status,
+        [("content-type", "application/json; charset=utf-8")],
+        json.dumps({"error": message}).encode(),
+    )
+
+
+class Console:
+    """Reverse-proxies the Seamless admin dashboard. Nothing from the incoming
+    request is forwarded but the method and the path: the console is public static
+    hosting, and the browser's session cookies have no business at the upstream."""
+
+    def __init__(self, adapter: Adapter) -> None:
+        self._adapter = adapter
+        base = httpx2.URL(adapter.config.auth_server_url)
+        self._base = base
+        self._prefix = (
+            base.raw_path.decode("ascii").split("?", 1)[0].rstrip("/") + CONSOLE_BASE_PATH
+        )
+
+    def _inside(self, url: httpx2.URL) -> bool:
+        """Holds a URL to the auth API's origin and the console subtree."""
+        path = url.raw_path.decode("ascii").split("?", 1)[0]
+        return (
+            url.scheme == self._base.scheme
+            and url.host == self._base.host
+            and url.port == self._base.port
+            and (path == self._prefix or path.startswith(self._prefix + "/"))
+        )
+
+    def _upstream(self, subpath: str, query: str) -> httpx2.URL | None:
+        """Maps a path below the mount to the console on the auth API, refusing
+        anything that could leave the console subtree: an encoded separator (an
+        upstream that decodes it would read a traversal), and any dot segment,
+        literal or encoded."""
+        lower = subpath.lower()
+        if "%2f" in lower or "%5c" in lower or "\\" in subpath:
+            return None
+        for segment in subpath.split("/"):
+            decoded = percent_decode(segment)
+            if decoded is None or decoded in (".", ".."):
+                return None
+        if subpath in ("", "/"):
+            subpath = ""
+        elif not subpath.startswith("/"):
+            subpath = "/" + subpath
+        if not subpath.isascii() or not subpath.isprintable():
+            return None
+
+        raw = f"{self._base.scheme}://{self._base.netloc.decode('ascii')}{self._prefix}{subpath}"
+        if query:
+            raw = f"{raw}?{query}"
+        try:
+            url = httpx2.URL(raw)
+        except (httpx2.InvalidURL, ValueError):
+            return None
+        return url if self._inside(url) else None
+
+    def handle(self, method: str, subpath: str, query: str = "") -> AuthResponse:
+        """Serves one console request. ``subpath`` is the raw path below /console.
+        Blocking: run it on a worker thread from async code."""
+        method = method.upper()
+        if method not in ("GET", "HEAD"):
+            return _error(405, "Method not allowed")
+        url = self._upstream(subpath, query)
+        if url is None:
+            return _error(400, "Invalid console path")
+
+        client = self._adapter._client
+        response: httpx2.Response | None = None
+        try:
+            for _ in range(MAX_REDIRECTS + 1):
+                request = client.build_request(method, url, timeout=UPSTREAM_TIMEOUT)
+                response = client.send(request, stream=True)
+                if not response.is_redirect:
+                    break
+                follow = response.next_request
+                response.close()
+                response = None
+                # A redirect is followed only while it stays inside the console on
+                # the auth API's own origin.
+                if follow is None or not self._inside(follow.url):
+                    return _error(502, "Console upstream unreachable")
+                url = follow.url
+            else:
+                return _error(502, "Console upstream unreachable")
+        except (httpx2.HTTPError, httpx2.InvalidURL, UnicodeError, ValueError) as err:
+            log.warning("[seamless-auth] Console upstream: %s", type(err).__name__)
+            return _error(502, "Console upstream unreachable")
+
+        if response is None:
+            return _error(502, "Console upstream unreachable")
+        headers = [
+            (name, response.headers[name]) for name in FORWARDED_HEADERS if name in response.headers
+        ]
+        if method == "HEAD":
+            response.close()
+            return AuthResponse(response.status_code, headers, b"")
+        return AuthResponse(response.status_code, headers, _stream(response), response.close)
+
+
+def _stream(res: httpx2.Response) -> Iterator[bytes]:
+    try:
+        yield from res.iter_bytes()
+    finally:
+        res.close()

src/seamless_auth/django.py

@@ -44,7 +44,14 @@ def me(request):
 from ._adapter import MAX_BODY_BYTES, Adapter, CrossSiteRequest, Unauthenticated, User
 from ._http import AuthRequest, AuthResponse, Headers
 
-__all__ = ["auth_view", "get_adapter", "require_auth", "urlpatterns"]
+__all__ = [
+    "auth_view",
+    "console_urlpatterns",
+    "console_view",
+    "get_adapter",
+    "require_auth",
+    "urlpatterns",
+]
 
 _PATH_SAFE = "/:@!$&'()*+,;=-._~"
 _lock = threading.Lock()
@@ -86,6 +93,24 @@ def auth_view(request: HttpRequest, path: str = "") -> HttpResponse:
 urlpatterns = [re_path(r"^(?P<path>.*)$", auth_view, name="seamless-auth")]
 
 
+@csrf_exempt
+def console_view(request: HttpRequest, path: str = "") -> HttpResponse:
+    """The Seamless admin dashboard, proxied from the auth API so it shares the
+    auth routes' origin. Exempt from the CSRF middleware because it serves GET and
+    HEAD only and answers anything else 405 itself."""
+    return _response(
+        get_adapter().console(
+            request.method or "GET",
+            _relative_raw_path(request, path),
+            request.META.get("QUERY_STRING", ""),
+        )
+    )
+
+
+#: Include at ``console/``: ``path("console/", include(console_urlpatterns))``.
+console_urlpatterns = [re_path(r"^(?P<path>.*)$", console_view, name="seamless-console")]
+
+
 def require_auth(view: F) -> F:
     """Answers 401 unless the request carries a valid session, and puts the
     :class:`~seamless_auth.User` on ``request.seamless_user``. A session from the

src/seamless_auth/fastapi.py

@@ -29,7 +29,7 @@ def me(user: User = Depends(current_user)):
 from ._adapter import MAX_BODY_BYTES, Adapter, CrossSiteRequest, Unauthenticated, User
 from ._http import AuthRequest, AuthResponse, Headers
 
-__all__ = ["RequireUser", "auth_router"]
+__all__ = ["RequireUser", "auth_router", "console_router"]
 
 _METHODS = ["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"]
 # What a path segment may keep when a raw path has to be rebuilt from a decoded one.
@@ -50,6 +50,21 @@ async def handle(request: Request) -> Response:
     return router
 
 
+def console_router(adapter: Adapter) -> APIRouter:
+    """The Seamless admin dashboard at /console, the path it is built against,
+    proxied from the auth API so it shares the auth routes' origin."""
+    router = APIRouter(prefix="/console", include_in_schema=False)
+
+    async def handle(request: Request) -> Response:
+        path = _relative_raw_path(request, "/console")
+        query = request.scope.get("query_string", b"").decode("latin-1")
+        return _response(await run_in_threadpool(adapter.console, request.method, path, query))
+
+    router.add_api_route("", handle, methods=_METHODS)
+    router.add_api_route("/{path:path}", handle, methods=_METHODS)
+    return router
+
+
 class RequireUser:
     """A dependency that answers 401 unless the request carries a valid session, and
     returns its :class:`~seamless_auth.User`. A session from the cookie also gets the