Monitoring Dashboard

create_dashboard_app(broker: Any, *, auth_token: str | None = None) -> Any

Create an ASGI dashboard application.

SECURITY (M8): by default this app has NO authentication and exposes broker topology (queue/exchange/routing-key names, consumer counts). Bind the server to loopback (127.0.0.1, not 0.0.0.0) by default, and mount it behind authn (OIDC/reverse proxy) plus network restriction for anything beyond local access — never expose it publicly.

For a lightweight built-in guard, pass auth_token: when set, every route requires an Authorization: Bearer <auth_token> header (compared in constant time via hmac.compare_digest, not a plain string !=, which would leak timing information about the token) and returns 401 otherwise. When unset (default), all requests pass through and a startup warning is logged reminding you not to expose the dashboard publicly.

Parameters:

Name	Type	Description	Default
broker	Any	A rabbitkit broker instance (SyncBroker or AsyncBroker).	required
auth_token	str | None	Optional bearer token. When set, all routes require Authorization: Bearer <auth_token>. When None, the dashboard runs unauthenticated (a startup warning is emitted).	None

Returns:

Type	Description
Any	A Starlette application.

Raises:

Type	Description
ImportError	If starlette is not installed.

Source code in src/rabbitkit/dashboard/app.py

def create_dashboard_app(
    broker: Any,
    *,
    auth_token: str | None = None,
) -> Any:
    """Create an ASGI dashboard application.

    SECURITY (M8): by default this app has NO authentication and exposes
    broker topology (queue/exchange/routing-key names, consumer counts).
    Bind the server to loopback (``127.0.0.1``, not ``0.0.0.0``) by default,
    and mount it behind authn (OIDC/reverse proxy) plus network restriction
    for anything beyond local access — never expose it publicly.

    For a lightweight built-in guard, pass ``auth_token``: when set, every
    route requires an ``Authorization: Bearer <auth_token>`` header
    (compared in constant time via ``hmac.compare_digest``, not a plain
    string ``!=``, which would leak timing information about the token) and
    returns ``401`` otherwise. When unset (default), all requests pass
    through and a startup warning is logged reminding you not to expose the
    dashboard publicly.

    Args:
        broker: A rabbitkit broker instance (SyncBroker or AsyncBroker).
        auth_token: Optional bearer token. When set, all routes require
            ``Authorization: Bearer <auth_token>``. When None, the dashboard
            runs unauthenticated (a startup warning is emitted).

    Returns:
        A Starlette application.

    Raises:
        ImportError: If starlette is not installed.
    """
    import hmac

    try:
        from starlette.applications import Starlette
        from starlette.middleware.base import BaseHTTPMiddleware
        from starlette.requests import Request  # noqa: TC002  # lazy optional import
        from starlette.responses import HTMLResponse, JSONResponse, PlainTextResponse, Response
        from starlette.routing import Route
    except ImportError:  # pragma: no cover
        raise ImportError(  # pragma: no cover
            "Dashboard requires starlette. Install with: pip install rabbitkit[dashboard]"
        ) from None

    from rabbitkit.health import broker_health_check

    if auth_token is None:
        logger.warning("Dashboard running WITHOUT authentication — do not expose publicly")

    class _BearerAuthMiddleware(BaseHTTPMiddleware):
        async def dispatch(self, request: Request, call_next: Any) -> Response:
            auth = request.headers.get("Authorization", "")
            expected = f"Bearer {auth_token}"
            # M8: hmac.compare_digest instead of != -- a plain string
            # comparison short-circuits on the first mismatched byte, leaking
            # the token's length (and, over many requests, its prefix) via a
            # timing side channel. compare_digest runs in constant time
            # regardless of where the mismatch is.
            if not hmac.compare_digest(auth, expected):
                return PlainTextResponse("Unauthorized", status_code=401)
            return await call_next(request)  # type: ignore[no-any-return]

    async def index(request: Any) -> Any:
        routes_count = len(broker.routes)
        health = broker_health_check(broker)
        html = f"""<!DOCTYPE html>
<html><head><title>rabbitkit Dashboard</title>
<style>
body {{ font-family: system-ui, sans-serif; max-width: 900px; margin: 2rem auto; padding: 0 1rem; }}
h1 {{ color: #333; }}
table {{ border-collapse: collapse; width: 100%; margin: 1rem 0; }}
th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
th {{ background: #f5f5f5; }}
.healthy {{ color: green; }} .degraded {{ color: orange; }} .unhealthy {{ color: red; }}
</style></head>
<body>
<h1>rabbitkit Dashboard</h1>
<h2>Health: <span class="{health.status.value}">{health.status.value.upper()}</span></h2>
<p>Routes: {routes_count} | Connected: {health.connected} | Consumers: {health.consumer_count}</p>
<h2>Routes</h2>
<table><tr><th>Name</th><th>Queue</th><th>Exchange</th><th>Ack Policy</th></tr>"""
        for r in broker.routes:
            exchange = r.exchange.name if r.exchange else ""
            # escape() — queue/exchange/route names render into HTML; never trust them raw
            html += (
                f"<tr><td>{escape(r.name)}</td><td>{escape(r.queue.name)}</td>"
                f"<td>{escape(exchange)}</td><td>{escape(r.ack_policy.value)}</td></tr>"
            )
        html += "</table></body></html>"
        return HTMLResponse(html)

    async def api_health(request: Any) -> Any:
        health = broker_health_check(broker)
        return JSONResponse(
            {
                "status": health.status.value,
                "started": health.started,
                "connected": health.connected,
                "consumer_count": health.consumer_count,
                "route_count": health.route_count,
            }
        )

    async def api_routes(request: Any) -> Any:
        routes = []
        for r in broker.routes:
            routes.append(
                {
                    "name": r.name,
                    "queue": r.queue.name,
                    "exchange": r.exchange.name if r.exchange else "",
                    "ack_policy": r.ack_policy.value,
                    "tags": sorted(r.tags) if r.tags else [],
                    "description": r.description,
                }
            )
        return JSONResponse(routes)

    app = Starlette(
        routes=[
            Route("/", index),
            Route("/api/health", api_health),
            Route("/api/routes", api_routes),
        ],
    )
    if auth_token is not None:
        app.add_middleware(_BearerAuthMiddleware)
    return app