Harden the public dashboard: id bounds, read-only role, error and worker redaction, log rotation

CHANGELOG.md

@@ -27,8 +27,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   self-hosting scripts in `deploy/termux/`.
 - CI runs the test suite on PostgreSQL 16, 17 and 18.
 
+### Security
+
+- The read-only dashboard shows only the last line of job errors with absolute paths removed, and
+  workers under a stable short label instead of host name and pid (API, live snapshot, metrics).
+  `WorkerOut.hostname` and `pid` are `None` there.
+- `read_only_dsn` / `PGQUEUE_READONLY_DSN`: the public dashboard can use a SELECT-only database role.
+  `deploy/termux/setup-db.sh` creates it, limits connections and revokes `CONNECT` from `PUBLIC`.
+
 ### Fixed
 
+- Job ids outside the bigint range (`/api/jobs/{id}`, `before_id`) now answer 422 instead of 500.
+- The Termux boot script rotates its log (`logcap.sh`) instead of appending forever.
+
 - Workers no longer log "lost the lease" for a job that finished while its heartbeat was
   in flight.
 - The live dashboard no longer counts every finished job each second, and one stalled

README.md

@@ -243,6 +243,16 @@ The public dashboard exposes **no endpoint that changes state**: in read-only mo
 mounted at all, and the buttons are not rendered. Settings are environment variables, listed in
 [`deploy/termux/env.example`](deploy/termux/env.example).
 
+A read-only dashboard also shows less about the machine it runs on. Job errors are cut to their
+last line (exception type and message) with absolute paths replaced by `<path>`, and workers appear
+under a short stable label (`w-1a2b3c`) without host name or pid, in `/api/jobs`, `/api/live`,
+`/api/overview` and `/metrics`. The full traceback and the real worker id stay in the database and in
+a normal (not read-only) admin. Job ids in the API must be between 1 and 2^63 - 1, otherwise the
+answer is 422.
+
+With `PGQUEUE_READONLY_DSN` (or `create_app(..., read_only_dsn=...)`) a read-only app connects as a
+different database role than the workload, so the public process can only `SELECT`.
+
 ```console
 $ export PGQUEUE_DSN=postgresql://user:[email protected]/pgqueue_demo
 $ python -m demo            # migrates, then serves http://127.0.0.1:8080
@@ -265,16 +275,24 @@ $ deploy/termux/start.sh                                     # try it by hand fi
 ```
 
 - **Start on boot**: copy `deploy/termux/boot/pgqueue-demo` to `~/.termux/boot/` (Termux:Boot);
-  it waits for PostgreSQL and restarts the demo if it exits.
+  it waits for PostgreSQL and restarts the demo if it exits. Output goes through
+  `deploy/termux/logcap.sh`, which rotates `~/.local/state/pgqueue-demo.log` at 1 MiB and keeps two
+  old files (`PGQUEUE_LOG_MAX_BYTES`, `PGQUEUE_LOG_KEEP`), so the log stays under about 3 MiB.
 - **nginx**: include `nginx-zones.conf` in the `http` block and `nginx-location.conf` in the
   `server` block. It publishes the dashboard under `/pgqueue/`, serves the static files from
   disk, disables buffering for the event stream, limits requests and streams per visitor
   (keyed on `CF-Connecting-IP` behind a Cloudflare Tunnel, where every request comes from
   `127.0.0.1`) and caps the streams in total.
 - **Public access**: put a Cloudflare Tunnel or Tailscale Funnel in front of nginx; the demo
   itself only listens on `127.0.0.1`.
-- **Database**: the demo uses its own role and database and works with PostgreSQL 16 to 18
-  (CI runs the test suite on all three).
+- **Database**: the demo uses its own database and works with PostgreSQL 16 to 18 (CI runs the test
+  suite on all three). `setup-db.sh` creates two roles: `pgqueue_demo` owns the database, runs the
+  workload and is limited to 40 connections; `pgqueue_demo_ro` (when `DEMO_RO_PASSWORD` is set) can
+  only `SELECT`, is read-only by default and is limited to 10 connections. Put its DSN in
+  `PGQUEUE_READONLY_DSN` for the public dashboard. The script also revokes `CONNECT` on
+  `pgqueue_demo` from `PUBLIC`. On a PostgreSQL shared with other applications, do the same for
+  their databases, because every role may connect to a database by default:
+  `REVOKE CONNECT ON DATABASE other_db FROM PUBLIC;` (then grant it to that application's role).
 
 ## Architecture
 

demo/config.py

@@ -27,6 +27,8 @@ def _int(env: Mapping[str, str], name: str, default: int) -> int:
 @dataclass(frozen=True, slots=True)
 class Settings:
     dsn: str
+    read_only_dsn: str | None = None
+    """Used by the public dashboard when set: a role that can only SELECT."""
     host: str = "127.0.0.1"
     port: int = 8080
     root_path: str = ""
@@ -65,6 +67,7 @@ def from_env(cls, env: Mapping[str, str] | None = None) -> Settings:
         root_path = env.get("DEMO_ROOT_PATH", "").strip().rstrip("/")
         return cls(
             dsn=dsn,
+            read_only_dsn=env.get("PGQUEUE_READONLY_DSN") or None,
             host=env.get("DEMO_HOST", defaults.host),
             port=_int(env, "DEMO_PORT", defaults.port),
             root_path=root_path,

demo/server.py

@@ -25,7 +25,12 @@ def build_app(settings: Settings, *, background: bool = True) -> FastAPI:
     With ``background=False`` only the dashboard is served (used by tests and when the
     workload runs elsewhere).
     """
-    app = create_app(settings.dsn, read_only=True, banner=settings.banner)
+    app = create_app(
+        settings.dsn,
+        read_only=True,
+        banner=settings.banner,
+        read_only_dsn=settings.read_only_dsn,
+    )
     inner = app.router.lifespan_context
 
     @asynccontextmanager

deploy/termux/boot/pgqueue-demo

@@ -12,8 +12,11 @@ for _ in $(seq 1 60); do
   sleep 2
 done
 
-# Restart if it ever exits; this loop is the only supervisor needed.
+# Restart if it ever exits; this loop is the only supervisor needed. The output goes through
+# logcap.sh, which rotates the file at PGQUEUE_LOG_MAX_BYTES (default 1 MiB) and keeps
+# PGQUEUE_LOG_KEEP old files (default 2), so the log can never grow without bound.
 while true; do
-  "$REPO/deploy/termux/start.sh" >>"$LOG" 2>&1
+  "$REPO/deploy/termux/start.sh" 2>&1 |
+    "$REPO/deploy/termux/logcap.sh" "$LOG" "${PGQUEUE_LOG_MAX_BYTES:-1048576}" "${PGQUEUE_LOG_KEEP:-2}"
   sleep 5
 done

deploy/termux/env.example

@@ -3,6 +3,13 @@
 
 # Database: its own role and database on the shared PostgreSQL (see setup-db.sh).
 PGQUEUE_DSN=postgresql://pgqueue_demo:[email protected]:5432/pgqueue_demo
+# The public dashboard connects with this one instead: a role that can only SELECT
+# (created by setup-db.sh when DEMO_RO_PASSWORD is set). Unset = it uses PGQUEUE_DSN.
+PGQUEUE_READONLY_DSN=postgresql://pgqueue_demo_ro:[email protected]:5432/pgqueue_demo
+
+# Boot log (deploy/termux/boot/pgqueue-demo): rotated at this size, this many old files kept.
+# PGQUEUE_LOG_MAX_BYTES=1048576
+# PGQUEUE_LOG_KEEP=2
 
 # Where the dashboard listens: loopback only, nginx is the public face.
 DEMO_HOST=127.0.0.1

deploy/termux/logcap.sh

@@ -0,0 +1,33 @@
+#!/data/data/com.termux/files/usr/bin/bash
+# Appends stdin to a log file and keeps it bounded: when the file passes MAX_BYTES it is
+# renamed to FILE.1 (FILE.1 to FILE.2, ...) and the oldest of KEEP old files is dropped.
+# Disk use is therefore at most about (KEEP + 1) * MAX_BYTES.
+#   start.sh 2>&1 | logcap.sh ~/.local/state/pgqueue-demo.log [MAX_BYTES=1048576] [KEEP=2]
+set -u
+
+FILE="${1:?usage: logcap.sh FILE [MAX_BYTES] [KEEP]}"
+MAX_BYTES="${2:-1048576}"
+KEEP="${3:-2}"
+CHECK_EVERY=50
+
+rotate() {
+  i="$KEEP"
+  rm -f "$FILE.$i"
+  while [ "$i" -gt 1 ]; do
+    if [ -f "$FILE.$((i - 1))" ]; then mv -f "$FILE.$((i - 1))" "$FILE.$i"; fi
+    i=$((i - 1))
+  done
+  if [ "$KEEP" -gt 0 ]; then mv -f "$FILE" "$FILE.1"; else : >"$FILE"; fi
+}
+
+n=0
+while IFS= read -r line || [ -n "$line" ]; do
+  printf '%s\n' "$line" >>"$FILE"
+  n=$((n + 1))
+  if [ "$n" -ge "$CHECK_EVERY" ]; then
+    n=0
+    size=$(wc -c <"$FILE" 2>/dev/null || echo 0)
+    if [ "$size" -gt "$MAX_BYTES" ]; then rotate; fi
+  fi
+done
+exit 0

deploy/termux/setup-db.sh

@@ -1,16 +1,44 @@
 #!/data/data/com.termux/files/usr/bin/bash
-# Creates the demo's own role and database on the shared PostgreSQL. Idempotent.
-#   DEMO_DB_PASSWORD=... ./setup-db.sh
+# Creates the demo's roles and database on the shared PostgreSQL. Idempotent.
+#   DEMO_DB_PASSWORD=... DEMO_RO_PASSWORD=... ./setup-db.sh
+#
+# pgqueue_demo     owns the database; runs migrations and the workload. At most 40 connections.
+# pgqueue_demo_ro  can only SELECT (and is read-only by default); give its DSN to the public
+#                  dashboard as PGQUEUE_READONLY_DSN. Skipped when DEMO_RO_PASSWORD is unset.
 set -euo pipefail
 
 : "${DEMO_DB_PASSWORD:?set DEMO_DB_PASSWORD to the password for the pgqueue_demo role}"
+DEMO_RO_PASSWORD="${DEMO_RO_PASSWORD:-}"
+DEMO_CONNECTION_LIMIT="${DEMO_CONNECTION_LIMIT:-40}"
+DEMO_RO_CONNECTION_LIMIT="${DEMO_RO_CONNECTION_LIMIT:-10}"
 PSQL="${PSQL:-psql -v ON_ERROR_STOP=1 -d postgres}"
 
-$PSQL -v pw="$DEMO_DB_PASSWORD" <<'SQL'
+$PSQL -v pw="$DEMO_DB_PASSWORD" -v conn_limit="$DEMO_CONNECTION_LIMIT" <<'SQL'
 SELECT format('CREATE ROLE pgqueue_demo LOGIN PASSWORD %L', :'pw')
 WHERE NOT EXISTS (SELECT FROM pg_roles WHERE rolname = 'pgqueue_demo') \gexec
-SELECT format('ALTER ROLE pgqueue_demo PASSWORD %L', :'pw') \gexec
+SELECT format('ALTER ROLE pgqueue_demo PASSWORD %L CONNECTION LIMIT %s', :'pw', :'conn_limit') \gexec
 SELECT 'CREATE DATABASE pgqueue_demo OWNER pgqueue_demo'
 WHERE NOT EXISTS (SELECT FROM pg_database WHERE datname = 'pgqueue_demo') \gexec
+-- By default every role may connect to every database. Only the demo roles need this one.
+REVOKE CONNECT ON DATABASE pgqueue_demo FROM PUBLIC;
 SQL
-echo "role and database pgqueue_demo are ready"
+
+if [ -n "$DEMO_RO_PASSWORD" ]; then
+  $PSQL -v pw="$DEMO_RO_PASSWORD" -v conn_limit="$DEMO_RO_CONNECTION_LIMIT" <<'SQL'
+SELECT format('CREATE ROLE pgqueue_demo_ro LOGIN PASSWORD %L', :'pw')
+WHERE NOT EXISTS (SELECT FROM pg_roles WHERE rolname = 'pgqueue_demo_ro') \gexec
+SELECT format('ALTER ROLE pgqueue_demo_ro PASSWORD %L CONNECTION LIMIT %s', :'pw', :'conn_limit') \gexec
+ALTER ROLE pgqueue_demo_ro SET default_transaction_read_only = on;
+GRANT CONNECT ON DATABASE pgqueue_demo TO pgqueue_demo_ro;
+\connect pgqueue_demo
+REVOKE CREATE ON SCHEMA public FROM PUBLIC;
+GRANT USAGE ON SCHEMA public TO pgqueue_demo_ro;
+GRANT SELECT ON ALL TABLES IN SCHEMA public TO pgqueue_demo_ro;
+-- Tables that the migrations create later, as pgqueue_demo, are readable too.
+ALTER DEFAULT PRIVILEGES FOR ROLE pgqueue_demo IN SCHEMA public GRANT SELECT ON TABLES TO pgqueue_demo_ro;
+SQL
+  echo "read-only role pgqueue_demo_ro is ready: set PGQUEUE_READONLY_DSN to use it"
+else
+  echo "DEMO_RO_PASSWORD not set: skipped the read-only role (the dashboard then uses PGQUEUE_DSN)"
+fi
+echo "role and database pgqueue_demo are ready (connection limit $DEMO_CONNECTION_LIMIT)"

src/pgqueue/admin/api.py

@@ -9,7 +9,18 @@
 from pathlib import Path
 from typing import Annotated
 
-from fastapi import APIRouter, Depends, FastAPI, HTTPException, Query, Request, status
+from fastapi import (
+    APIRouter,
+    Depends,
+    FastAPI,
+    HTTPException,
+    Query,
+    Request,
+    status,
+)
+from fastapi import (
+    Path as PathParam,
+)
 from fastapi.responses import (
     FileResponse,
     HTMLResponse,
@@ -82,6 +93,15 @@ def _parse_statuses(raw: str | None) -> list[JobStatus]:
 
 Engine = Annotated[AsyncEngine, Depends(_engine)]
 
+# Ids are bigint columns: anything larger would reach the driver and come back as a 500.
+MAX_ID = 2**63 - 1
+JobId = Annotated[int, PathParam(ge=1, le=MAX_ID)]
+
+
+def _public(request: Request) -> bool:
+    public: bool = request.app.state.read_only
+    return public
+
 
 def _feed(request: Request) -> LiveFeed:
     feed: LiveFeed = request.app.state.feed
@@ -98,21 +118,22 @@ def _feed(request: Request) -> LiveFeed:
 
 @api.get("/overview")
 async def get_overview(
-    engine: Engine, minutes: Annotated[int, Query(ge=1, le=24 * 60)] = 60
+    request: Request, engine: Engine, minutes: Annotated[int, Query(ge=1, le=24 * 60)] = 60
 ) -> Overview:
     """Counts per queue and status, queue lag, throughput, workers and schedules."""
     async with engine.connect() as conn:
-        return await queries.overview(conn, minutes=minutes)
+        return await queries.overview(conn, minutes=minutes, redact=_public(request))
 
 
 @api.get("/jobs")
 async def list_jobs(
+    request: Request,
     engine: Engine,
     *,
     status_: Annotated[str | None, Query(alias="status", description="comma-separated")] = None,
     queue: str | None = None,
     task: str | None = None,
-    before_id: int | None = None,
+    before_id: Annotated[int | None, Query(ge=1, le=MAX_ID)] = None,
     limit: Annotated[int, Query(ge=1, le=500)] = 50,
 ) -> JobPage:
     async with engine.connect() as conn:
@@ -123,13 +144,14 @@ async def list_jobs(
             task=task,
             before_id=before_id,
             limit=limit,
+            redact=_public(request),
         )
 
 
 @api.get("/jobs/{job_id}")
-async def get_job(engine: Engine, job_id: int) -> JobOut:
+async def get_job(request: Request, engine: Engine, job_id: JobId) -> JobOut:
     async with engine.connect() as conn:
-        job = await queries.get_job(conn, job_id)
+        job = await queries.get_job(conn, job_id, redact=_public(request))
     if job is None:
         raise HTTPException(status.HTTP_404_NOT_FOUND, "job not found")
     return job
@@ -174,7 +196,7 @@ async def events() -> AsyncIterator[bytes]:
 
 
 @actions.post("/jobs/{job_id}/requeue", dependencies=[Depends(_require_action_header)])
-async def requeue_job(engine: Engine, job_id: int) -> JobOut:
+async def requeue_job(engine: Engine, job_id: JobId) -> JobOut:
     """Run a failed, dead or retrying job again now."""
     async with engine.begin() as conn:
         job = await queries.requeue_job(conn, job_id)
@@ -208,6 +230,7 @@ def create_app(
     password: str | None = None,
     read_only: bool | None = None,
     banner: str | None = None,
+    read_only_dsn: str | None = None,
 ) -> FastAPI:
     """Build the admin application.
 
@@ -217,12 +240,19 @@ def create_app(
 
     ``read_only`` (default: ``PGQUEUE_ADMIN_READ_ONLY``) leaves out every
     endpoint that changes state, so the dashboard can be shown to the public.
+    A read-only app also hides what only an operator needs: job errors are cut to
+    their last line with absolute paths removed, and workers are shown under a
+    short stable label instead of hostname and pid.
+    ``read_only_dsn`` (default: ``PGQUEUE_READONLY_DSN``) is used instead of ``dsn``
+    for a read-only app, so the public engine can connect as a role that can only SELECT.
     ``banner`` is an optional line of text shown at the top of the dashboard.
     """
     username = username or os.environ.get("PGQUEUE_ADMIN_USERNAME")
     password = password or os.environ.get("PGQUEUE_ADMIN_PASSWORD")
     if read_only is None:
         read_only = os.environ.get("PGQUEUE_ADMIN_READ_ONLY", "").lower() in ("1", "true", "yes")
+    if read_only:
+        dsn = read_only_dsn or os.environ.get("PGQUEUE_READONLY_DSN") or dsn
     engine = create_async_engine(sqlalchemy_url(dsn), pool_size=5, pool_pre_ping=True)
     feed = LiveFeed(engine, read_only=read_only)
 
@@ -235,6 +265,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
     app = FastAPI(title="pgqueue admin", version=__version__, lifespan=lifespan)
     app.state.engine = engine
     app.state.feed = feed
+    app.state.read_only = read_only
     app.state.credentials = (username, password) if username and password else None
     app.include_router(api)
     if not read_only:
@@ -266,7 +297,7 @@ async def dashboard(request: Request) -> HTMLResponse:
     )
     async def metrics() -> PlainTextResponse:
         async with engine.connect() as conn:
-            body = await render_metrics(conn)
+            body = await render_metrics(conn, redact=read_only)
         return PlainTextResponse(body, media_type=CONTENT_TYPE_LATEST)
 
     @app.get("/healthz", include_in_schema=False)

src/pgqueue/admin/metrics.py

@@ -14,6 +14,7 @@
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncConnection
 
+from pgqueue.admin.redact import worker_label
 from pgqueue.schema import READY_STATUSES, JobStatus, jobs, periodic, workers
 
 FINISHED = (JobStatus.SUCCEEDED, JobStatus.FAILED, JobStatus.DEAD)
@@ -27,7 +28,7 @@ def collect(self) -> Iterable[GaugeMetricFamily | CounterMetricFamily]:
         return self.families
 
 
-async def render(conn: AsyncConnection) -> bytes:
+async def render(conn: AsyncConnection, *, redact: bool = False) -> bytes:
     by_state = GaugeMetricFamily(
         "pgqueue_jobs", "Jobs in the table, by queue and status.", labels=["queue", "status"]
     )
@@ -83,10 +84,11 @@ async def render(conn: AsyncConnection) -> bytes:
     for row in await conn.execute(
         select(workers.c.id, workers.c.in_flight, workers.c.processed, workers.c.failed, is_alive)
     ):
-        alive.add_metric([row[0]], 1.0 if row[4] else 0.0)
-        in_flight.add_metric([row[0]], row[1])
-        processed.add_metric([row[0]], row[2])
-        failed.add_metric([row[0]], row[3])
+        worker = worker_label(row[0]) if redact else row[0]
+        alive.add_metric([worker], 1.0 if row[4] else 0.0)
+        in_flight.add_metric([worker], row[1])
+        processed.add_metric([worker], row[2])
+        failed.add_metric([worker], row[3])
 
     next_run = GaugeMetricFamily(
         "pgqueue_periodic_next_run_timestamp_seconds",